Fluens.Web.Messaging 0.7.2

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-0005 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 single server-side middleware guards all /mesh/* endpoints and 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.
• ClusterTokenMiddleware validates the incoming token on every /mesh/* request 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-0007). They live in a durable Consul-KV subscription store, decoupling who is owed (durable) from where to deliver now (the live catalog).

A directed message (ADR-0012 — 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).

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-0008) 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-0007)

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 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 registers the cluster-token guard middleware and maps the POST /mesh/deliver minimal-API endpoint (there is no /mesh/describe 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.