Repository navigation
Close the cross-process change feed: start the PG listener, route it, and re-read on an entity-less notification - #1929
Merged
Conversation
… and re-read on an entity-less notification Two halves of one defect, which is why they land together. A. The LISTEN session is registered AND started, and its events are ROUTED. PostgreSqlExtensions.cs:367-368 were commented out, and the comment above them named the real blocker rather than the wiring: in the partitioned setup the listener has to pump notifications across many per-partition adapters. That router is PartitionChangeRouter. The mesh_node_changes channel is per DATABASE — one session receives every schema's events — while the change feed under PostgreSqlPathRoutingAdapter is per SCHEMA, so an event has to be resolved to a partition before it can be published. It resolves through exactly the resolution a WRITE uses (first path segment to schema, with the satellite-node-type guard, the _-prefix globals via the registered-partition map, and the invalid-segment refusal), so a notification can never land on a feed the write itself would not have published to. B. The consumer re-reads instead of discarding. MeshDataSource gave up on any notification whose entity was absent — which is EVERY cross-process notification there is. Turning A on without B changes nothing: the notifications arrive and are dropped. Entity = null is CORRECT and stays correct. Populating it from the NOTIFY payload would reintroduce the row-level-security bypass #1250 removed, and would race row visibility besides. So the consumer reads the row back, and the result is durable by construction — which keeps the provenance AdoptPersisted depends on. Both branches funnel into one AdoptDurable local function, so adoption, echo-suppression, tombstone clearing and the forward-only guard are shared and unchanged; nothing mints (#1432). The trigger's only effects are a READ and an AdoptPersisted, neither of which writes, so the reconcile cannot feed itself (#223). Bounded: Throttle(50 ms) per hub, and each of these subscriptions only ever sees one path — its own — so that is a per-path coalesce. 200 notifications for one path cost 1 re-read, measured. Throttle deliberately, not Sample: Sample arms a periodic timer per subscription and there is one per live node hub, so an idle mesh would pay a tick per hub per window for nothing. The listener holds its own single connection (MinPoolSize/MaxPoolSize 1, KeepAlive 30, named in pg_stat_activity) built BY THE PROVIDER through the same configureDataSource hook as every other source it builds — on Azure Postgres the credential is an AAD token that arrives only through that hook, so a listener built from the connection string alone would retry for ever on exactly the deployment this feed exists for. Tests, each failing on origin/main: two independent meshes over one store converge without a recycle (on main the mirror ends at a version the store never held); the burst measurement (0 re-reads on main, 1 with the fix); the DI shape, because "registered but never started" is invisible from behaviour; and a real Postgres container carrying its own no-listener control. Closes #1440. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Pull request overview
This PR completes the cross-process Postgres change-feed chain for partitioned persistence by (1) starting a dedicated PG LISTEN mesh_node_changes session at host startup, (2) routing those notifications to the correct per-partition change feed, and (3) teaching the consumer-side reconcile to re-read when the notification has no entity (the expected cross-process shape).
Changes:
- Register and start
PostgreSqlChangeListenerfor partitioned PG persistence, backed by a dedicated single-connection data source, and route notifications viaPartitionChangeRouter. - Route external notifications onto either the per-schema adapter feed (when materialized) or the merged feed without materializing adapters.
- Update
MeshDataSourceown-node reconcile to handle entity-less notifications by triggering a throttled, serialized re-read and thenAdoptPersisted. - Add targeted unit/integration tests and a WhatsNew entry covering the end-to-end behavior and wiring invariants.
Reviewed changes
Copilot reviewed 12 out of 12 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| test/MeshWeaver.Hosting.PostgreSql.Test/PartitionChangeRouterTests.cs | Pins routing outcomes (per-partition, merged, discarded) and “no throw reaches listener” behavior. |
| test/MeshWeaver.Hosting.PostgreSql.Test/CrossProcessNotifyPgTests.cs | End-to-end LISTEN/NOTIFY + router test against real Postgres, including a no-listener control. |
| test/MeshWeaver.Hosting.PostgreSql.Test/ChangeListenerWiringTests.cs | Asserts DI wiring includes both listener singleton and hosted service starter. |
| test/MeshWeaver.Hosting.Monolith.Test/CrossProcessFeedAdapter.cs | Test-only adapter decorator modeling entity-less cross-process notifications and counting reads/writes. |
| test/MeshWeaver.Hosting.Monolith.Test/CrossProcessChangeFeedTest.cs | Two-mesh integration test proving live mirror convergence on entity-less notifications and bounded re-reads. |
| src/MeshWeaver.Hosting.PostgreSql/PostgreSqlPathRoutingAdapter.cs | Adds external-notification routing to the correct feed without materializing untouched adapters. |
| src/MeshWeaver.Hosting.PostgreSql/PostgreSqlPartitionStorageProvider.cs | Exposes routing hook and builds a dedicated listener data source via provider configuration. |
| src/MeshWeaver.Hosting.PostgreSql/PostgreSqlExtensions.cs | Enables partitioned change listener wiring (listener + hosted service + router) in both overloads. |
| src/MeshWeaver.Hosting.PostgreSql/PostgreSqlChangeListener.cs | Adds “owning data source” construction and disposal semantics for dedicated listener data source. |
| src/MeshWeaver.Hosting.PostgreSql/PartitionChangeRouter.cs | New router that delivers LISTEN notifications to the owning partition feed with discard counting and isolation. |
| src/MeshWeaver.Graph/MeshDataSource.cs | Adds re-read reconciliation path for entity-less durable notifications with throttling and serialized reads. |
| src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-20-cross-process-change-feed.md | Documents the user-visible fix and ties it to the underlying cross-process feed behavior. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| // ObjectDisposedException thrown back into the adapter's change-feed fan-out — the exact | ||
| // ordering trap the synced-query pipelines document. An OnNext into a subject whose only | ||
| // subscription is already disposed is a harmless no-op. | ||
| var reReadTrigger = new System.Reactive.Subjects.Subject<Unit>(); |
Contributor
Contributor
Contributor
Contributor
Contributor
Test Results (shard 2)3 234 tests - 100 3 042 ✅ - 100 7m 44s ⏱️ +27s Results for commit a169428. ± Comparison against base commit ddd984f. This pull request removes 355 and adds 255 tests. Note that renamed tests count towards both. |
Contributor
Test Results (shard 1)1 584 tests +103 1 580 ✅ +103 8m 38s ⏱️ +28s Results for commit a169428. ± Comparison against base commit ddd984f. This pull request removes 255 and adds 358 tests. Note that renamed tests count towards both. |
Contributor
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes #1440.
What this is
The cross-process change feed, end to end — the last leg of #1814's notify chain. Two halves that
only work together, which is why they land together:
A — the LISTEN session is registered AND started, and its events are ROUTED to a partition.
PostgreSqlExtensions.cs:367-368were commented out, and the comment above them named the realblocker rather than the wiring: "in the partitioned setup the listener [must] pump notifications
across many per-partition adapters … A follow-up will route LISTEN events to the right partition
adapter's
ChangeObservervia the partition registry." That router isPartitionChangeRouter.The
mesh_node_changeschannel is per DATABASE — one session receives every schema's events —while the change feed under
PostgreSqlPathRoutingAdapteris per SCHEMA, so an event has to beresolved to a partition before it can be published. It resolves through exactly the resolution a
WRITE uses (
ResolveSchema: first path segment → schema, with the satellite-node-type guard, the_-prefix globals via the registered-partition map, and the invalid-segment refusal), so anotification can never land on a feed the write itself would not have published to, and there is no
second resolution to drift out of step with the first.
B — the consumer re-reads instead of discarding.
MeshDataSource.cs:1479gave up on anynotification whose entity was absent (
if (notification.Entity is not MeshNode newNode) return;) —which is EVERY cross-process notification there is. Turning A on without B changes nothing: the
notifications arrive and are dropped.
Entity = nullis correct and stays correctThe fix is a re-read, never a supplemented entity. Populating
Entityfrom the NOTIFY payloadwould reintroduce the row-level-security bypass
StorageAdapterMeshQueryProviderremoved (#1250) —"whenever the supplement DID change the outcome, the re-query was the one that was right: the row it
re-added was one the read had excluded by RLS, by satellite-table routing (#1193), by a path filter,
or past the load cap" — and it races row visibility besides. So the consumer reads the row back, and
the result is durable by construction, which keeps the provenance
AdoptPersisteddepends on.The invariants at the site, preserved
AdoptPersisted, notUpdate. Both branches — the entity-carrying in-process publish and there-read — funnel into ONE
AdoptDurablelocal function that adopts. Nothing mints (Adopting an already-durable external write MINTS a phantom revision and persists it (deterministic; root cause under #1384) #1432).version the write landed at, and
newNode.Version == lastSelfWrite.Valuestill skips.cache.IsDeleted = false,recentlyDeleted?.Clear(ownPath)) and theforward-only guard are unchanged and shared by both branches.
re-read result IS the durable row.
AdoptPersisted,and
AdoptPersistedneither mints nor writes (it records the durable version so the persistencesampler and the dispose-time flush both skip it). No write on the path ⇒ nothing of our own to
re-trigger it. Pinned by a test that asserts the store never moves after an echo.
The bound
Throttle(50 ms)per hub — and each of these subscriptions only ever sees ONE path, its own, sothat is a per-path coalesce. 200 notifications for one path cost a handful of reads, not 200
(measured below).
Concatserialises them so at most one read per hub is ever in flight.Throttle, deliberately notSample:Samplearms a PERIODIC timer per subscription and thereis one subscription per live node hub, so an idle mesh would pay a timer tick per hub per window for
nothing.
Throttlearms a one-shot timer only while a burst is in flight, and it always emits theLAST trigger of a burst — the read that CONVERGES the mirror can never be the one that gets dropped.
Cheap discard: a path this hub does not own costs one ordinal string compare — no read, no
allocation. At the router, an unroutable path costs one synchronous segment resolution and does
not materialise a per-schema adapter (on a 140-partition mesh, routing-by-materialising would
leave every process holding an adapter for every partition purely because someone, somewhere, wrote).
Two things the integration test caught that review would not have
ConnectionIdleLifetime = 0; Npgsql refuses thatunder the default pruning interval, and the throw lands in the DI factory at host startup. Fixed
to
MinPoolSize = 1(pruning only ever touches connections idle IN THE POOL, and this one ischecked out for the life of the process).
configureDataSourcehook as every other source it builds. On Azure Postgres the portal has nopassword in its connection string at all — the credential is an AAD token supplied by
UsePeriodicPasswordProviderthrough that hook — so a listener built from the connection stringalone would sit in its 5-second reconnect loop for ever on exactly the deployment this feed exists
for, indistinguishable from never having been wired.
It also holds its own single connection (
MaxPoolSize = 1,KeepAlive = 30, namedmeshweaver-change-listenerinpg_stat_activity) rather than taking one from the pool the portalhas already exhausted once.
KeepAliveis not decoration: a LISTEN connection is idle by design, andan idle TCP session behind a NAT/firewall is dropped silently — after which
WaitAsyncwaits forever on a dead socket and the process is back to "never told", with nothing to notice.
Tests — each fails on
origin/maintest/MeshWeaver.Hosting.Monolith.Test/CrossProcessChangeFeedTest.cs— two independent meshes overone durable store (the
CrossClusterBuildClaimTestshape), wired with the entity-less feed:origin/mainAWriteInOneProcess_ReachesTheOtherProcessesLiveMirror_WithoutARecycle: mirror ends at v3 while the store holds v2 — a revision that exists nowhere (#1432), reached by accident through a read-seeded mintAnEntitylessBurst_…:200 notifications → own-path reads 400, query-layer baseline 400, re-reads attributable to the reconcile **0**own-path reads 401, query-layer baseline 400, re-reads attributable to the reconcile **1**AnEntitylessEchoOfTheOwnWrite_IsSuppressed_AndWritesNothingBack— passes on main too (the echo path is unchanged); it is the regression pin for #1432/#223test/MeshWeaver.Hosting.PostgreSql.Test/ChangeListenerWiringTests.cs— asserts on the serviceDESCRIPTORS, because "registered but never started" is invisible from behaviour. On main both fail:
"the partitioned wiring must register the LISTEN/NOTIFY listener … but found False."
test/MeshWeaver.Hosting.PostgreSql.Test/CrossProcessNotifyPgTests.cs— a real Postgres container:adapter A writes, adapter B's partition feed receives it through trigger →
pg_notify→ LISTEN →router. It carries its own control: a third provider configured identically but with no
listener sees nothing, evaluated at the moment B's notification lands. That control is the state
mainis in, and it is why the stale mirror was a missing edge rather than a lost race — no retrycan win a race that is not being run.
test/MeshWeaver.Hosting.PostgreSql.Test/PartitionChangeRouterTests.cs— routing decisions, nocontainer and no connection: owned partition → its feed; untouched partition → the merged feed with
no adapter materialised; unroutable path (a NodeType first segment, a URL-shaped segment, an
unregistered
_Access) → discarded;_Access→system_accessonce registered; a throwingsubscriber never reaches the listener (a throw there would tear the LISTEN loop into its reconnect
branch, so one bad subscriber would cost every path its feed).
What remains of #1814
Both named defects are already fixed (#1847 bake identity, #1870 the latching fault card), and the
notify chain's three legs are now all closed: the per-database trigger (V54/#1816), the listener
(this PR), the consumer (this PR).
What this PR does not settle, stated precisely rather than assumed:
Conflictretry's base still comes fromMeshNodeStreamCache, and that cache does notsubscribe to the change feed. The subscribers are the own-node reconcile (fixed here), the
synced-query providers, and
BuildNodeType. So if the stale snapshot behind the deterministicconflict is a mirror's rather than an owning hub's, the notification now ARRIVES in that
process but nothing hands it to the mirror. The cache already exposes
Invalidate(path); wiringit to the routed feed is the obvious next step — and deliberately not done here, because an
invalidate-per-notification is a resubscribe storm shaped exactly like the 2026-06-08 outage and
needs its own measurements.
coalesced single-node read per live own-node hub, and one extra re-query per matching synced
query. Correct but wasteful. The principled fix is a sender id in the NOTIFY payload (a per-process
application_namethe trigger emits, and the listener drops its own) — a trigger change, i.e. amigration PR next to Install mesh_node_notify per TABLE, not once per database (V54) #1816, not a heuristic here.
Verification
dotnet build -c Release -p:CIRun=true -warnaserrorclean on every touched project.| suite | result |
|---|---|
|
MeshWeaver.Graph.Test| 1381 passed, 0 failed ||
MeshWeaver.Hosting.Monolith.Test| 731 passed, 0 failed, 3 skipped (9 m 19 s) ||
MeshWeaver.Hosting.PostgreSql.Test| 734 passed, 0 failed, 1 skipped (Testcontainers, ran locally) ||
MeshWeaver.Hosting.Orleans.Test| 170 passed, 0 failed ||
MeshWeaver.AI.Test| 1146 passed, 0 failed, 3 skipped |The headline cross-process test was also run 20× in a row: 20 pass, 0 fail — it is deterministic,
not "passed once".