Fluens.Web.Messaging 0.7.10

Fluens.Web.Messaging

P2P HTTP/Protobuf mesh adapter for cross-application Fluens messaging. It layers a decentralized peer-to-peer HTTP/1.1 network hop on top of the existing Fluens.Messaging outbox/inbox durability, reusing at-least-once delivery, MessageId deduplication, retry, dead-letter, and context propagation. There is no central broker; in-process-only applications are unaffected. Per ADR 20260529000002 the wire protocol is HTTP/1.1 request/response with a Protobuf body — the mesh does not require HTTP/2.

Cluster authentication

Every mesh call is authenticated with a short-lived HMAC HS256 token signed by a shared symmetric cluster key. Tokens carry an issuer (the application name), an audience (the cluster id), iat, exp (short-lived), and jti. A ClusterTokenEndpointFilter is attached directly to the POST /mesh/deliver route (not a path-prefix middleware), so the guard travels with the route under any group prefix or hosting shape — a group prefix can never route around it. It rejects un-tokened or invalid requests with 401 Unauthorized. An expired token is rejected even with a valid signature; a reasonable clock skew is tolerated.

• ClusterTokenProvider mints and validates tokens using Microsoft.IdentityModel.Tokens primitives directly (HMAC HS256), without depending on Fluens.Web.Auth.
• ClusterTokenClientHandler is a DelegatingHandler that attaches the token to outgoing requests.
• ClusterTokenEndpointFilter validates the token on the mesh route (attached via AddEndpointFilter) and rejects invalid ones.

The cluster key is a signing secret: it is only used to sign and verify tokens, and is never placed in a request header, logs, responses, or any wire payload. Authentication correctness does not depend on TLS.

The HTTP mesh

The adapter owns the mesh wire. Protos/mesh.proto defines the message types only (no gRPC service); the single HTTP endpoint carries them with Content-Type: application/x-protobuf:

• POST /mesh/deliver — a batch-unary call with per-message acks. The receiver writes the whole batch atomically into the single ingress staging table (one DbContext, one transaction) before acking, deduplicating on MessageId (the unique index is the authoritative guard), then wakes the local ingress distribution worker. A received message is terminal — it is fanned out to local inboxes only, never relayed.

There is no /mesh/describe route: subscriptions are no longer exchanged over the wire (ADR 20260529000004). They live in a durable Consul-KV subscription store, decoupling who is owed (durable) from where to deliver now (the live catalog).

Ingress retention

Fluens.Web.Messaging is the only project that registers IngressDbContext, so it also owns pruning the staging table it creates: IngressHousekeepingWorker is registered alongside IngressDistributionProcessor by AddMessagingMesh and batch-deletes processed shared.ingress_outbox rows past MessagingOptions.IngressRetentionDays (default 7 days). An unprocessed row is never eligible for deletion regardless of age, matching the core HousekeepingWorker's never-lose-an-in-flight-row guarantee — the core worker itself cannot reach this table at all, since it is scoped to a module's own DbContext, not the shared adapter-owned schema.

A directed message (ADR 20260530000005 — a cross-application Fault<T> routed to one origin module) reuses this same POST /mesh/deliver endpoint: the origin-module target rides inside the serialized envelope's fluens-directed-module header, so no new route and no new proto field are added. The receiver stages it identically and the core ingress distribution worker reads the header to route it to exactly one origin module (still local-only — the no-relay rule holds).

The receiver reads each delivered message's PartitionKey and Priority from its envelope and carries them onto the staged IngressOutbox row's own PartitionKey/Priority columns, matching the outbox/inbox denormalization convention (ADR 20260530000002). An empty or malformed envelope blob stages the row with PartitionKey = null and Priority = MessagePriorities.Low rather than throwing — a corrupt envelope on one message must not abort the whole batch.

HttpRemoteMessageTransport implements the core IRemoteMessageTransport port: it maps each RemoteMessage to the MeshMessage proto (passing the already-serialized payload and envelope across the wire in the bytes envelope = 4 field, never re-serializing), POSTs the Protobuf body over a cluster-token-authenticated HttpClient, and maps the acked ids back. The envelope (ADR 20260530000001) carries the nested execution context plus messaging metadata in one blob, so no per-field wire columns are needed — partition_key / available_at / priority / routing all live inside it.

Durable subscription store (ADR 20260529000004)

Each application owns one Consul-KV key fluens/<ClusterId>/subscriptions/<app>, whose JSON value lists the message types it statically subscribes to. The key segment and the record's app field are the application's canonical catalog service name (AppInfo.CachePrefix — the same identity Consul self-registration uses), never the raw IAppInfo.Name, so a peer can resolve the owed application's live endpoint from the catalog by the matching service name.

• SubscriptionStore reads/writes the KV namespace: UpsertAsync is a compare-and-swap on ModifyIndex (idempotent owner rewrite; a lost race is retried next cycle), ListAsync reads every record under the cluster prefix, and WatchAsync is a KV blocking-query watch reusing the WaitIndex/LastIndex long-poll pattern.
• SubscriptionPublisher (hosted) authoritatively rewrites this application's own record at startup and on the configured interval (drift repair), keyed by the canonical service name.
• SubscriptionCatalogWorker (hosted) blocking-watches the cluster prefix and rebuilds the owed-map into RemoteSubscriptionRegistry on every change, flipping HasConverged on the first (even zero-record) load.

RemoteSubscriptionRegistry holds the messageType → {owed apps} map and resolves each owed app's current endpoint live from the central IServiceCatalog on every lookup. When an owed app currently has no healthy instance the endpoint resolves to null — a hold, not a drop: the per-destination delivery row waits for a live instance (and TTL-expires in place if the peer stays down past the message TTL) rather than being lost.

RemoteSubscriptionRegistry also exposes a HasConverged cold-start gate (flipped to true on its first owed-map load, even a zero-record one). This disambiguates an empty owed set during startup: the core transport processor fans an outbox row into per-destination delivery rows only after the registry has converged. Before the first load an empty set means "peers may exist but are not yet learned", so the row is held for retry rather than silently marked delivered — no message is lost to a cold start. (In-process-only apps use the core no-op registry, which is always converged, so publishing is never blocked.) Peer endpoints are resolved, not configured: the registry resolves scheme://host:port from the central IServiceCatalog. There is no Microsoft.Extensions.ServiceDiscovery dependency and no configured peer list. The mesh HttpClient is registered with RemoveAllLoggers(), so transient connection errors to a not-yet-healthy peer never spam operator logs; genuine delivery problems surface via outbox errors / dead-letters.

Usage

builder.AddFluens()
    .AddWeb()
    .AddMessaging()
    .AddMessagingMesh(ingress => ingress.UseNpgsql(connectionString));

var app = builder.Build();
app.MapMessagingMesh();

AddMessagingMesh's Action<MessagingMeshOptions>? callback is honoured and validated even with no matching Fluens:Web:Messaging configuration section present.

AddMessagingMesh requires service discovery: it calls TryAddDiscovery (wiring the central catalog once) and fails fast when Fluens.Web.Discovery is unregistered or DiscoveryOptions.ClusterId is blank — the cluster id is the single source of the cluster identity (the HMAC token audience and the Consul-KV subscription-store namespace). It registers a MeshRegistrationContributor that advertises the fluens-mesh tag on this application's Consul registration (the fluens-cluster meta is published by Discovery itself from DiscoveryOptions.ClusterId), overrides the core no-op remote ports with the HTTP implementations, registers a typed HttpClient (cluster-token handler, with default per-request logging removed), the ingress staging DbContext and its distribution worker, the client transport, the durable SubscriptionStore, the hosted SubscriptionPublisher and SubscriptionCatalogWorker, and the RemoteSubscriptionRegistry (live endpoint resolution from the catalog). The adapter stays provider-agnostic: the consuming application supplies the ingress database provider (e.g. Npgsql) pointed at the shared schema that owns the ingress_outbox table. When MessagingMeshOptions.Enabled is false, AddMessagingMesh is a no-op and the in-process-only messaging path is unchanged. MapMessagingMesh maps the POST /mesh/deliver minimal-API endpoint, guarded by a ClusterTokenEndpointFilter attached directly to the route (there is no /mesh/describe route). MapMessagingMesh itself is also a no-op — it maps nothing — when AddMessagingMesh was never called or MessagingMeshOptions.Enabled is false, mirroring MapMessagingDashboard's "maps nothing when absent" convention: a host that keeps app.MapMessagingMesh() in its pipeline while flipping Enabled to false starts successfully and exposes no /mesh/deliver route.

Configuration

MessagingMeshOptions is bound from the "Fluens:Web:Messaging" configuration section:

The cluster id is single-source in Fluens:Web:Discovery (DiscoveryOptions.ClusterId) — it is the HMAC token audience and the Consul-KV subscription-store namespace, and is not a Fluens:Web:Messaging option. ClusterKey (the signing secret) stays in Fluens:Web:Messaging. Subscriptions live in the durable Consul-KV store and peer endpoints are resolved from the central catalog, so there is no Peers option.