Compare commits

...

11 Commits

Author SHA1 Message Date
Joseph Doherty 41bdc178c2 fix(localdb): refuse cascade-delete FK parents; make one entry's row+ledger writes atomic (0.3.3, ScadaBridge arch-review P4-7)
Two halves of the same INSERT OR REPLACE hazard in the LWW apply path. Wire-compatible
with every earlier peer: no proto change, no lib_schema_version bump, no on-disk schema
change.

1. RegisterReplicated now REFUSES a table that another table references with a
   row-mutating ON DELETE action (CASCADE / SET NULL / SET DEFAULT). Applying a remote row
   uses INSERT OR REPLACE, whose implicit DELETE fires that referential action — while the
   apply path holds the capture guard raised, precisely so its own writes are not
   re-captured. The child rows it deletes or nulls therefore produce no oplog entry: the
   peer keeps them and the pair diverges with nothing reporting a fault. Registration is
   the only point at which the problem is still visible, so it is refused there with an
   actionable LocalDbRegistrationException naming the referencing table and its action.
   NO ACTION / RESTRICT stay allowed: they mutate no child row, so the implicit delete just
   fails and the applier dead-letters the entry — loud, not silent. Registration-time only;
   a cascading child table created LATER is not caught (SQLite offers no DDL hook).

   No current consumer has a foreign key on a replicated table (ScadaBridge's ten tables
   have none), so this is a guard against a future foot-gun, not a live fix.

2. One entry's row write and its ledger write are now atomic. They were two statements
   under the poison catch, so a failure landing between them committed a deleted/replaced
   row carrying a STALE LIVE row-version entry — exactly the shape SnapshotStreamer warns
   about and skips, so that key never travels again and the pair is permanently divergent
   for it. A per-entry SAVEPOINT makes the pair all-or-nothing while keeping the catch's
   per-ENTRY granularity: entries applied earlier in the same batch survive the rollback,
   the poison entry is still dead-lettered, and the batch still continues. The RELEASE runs
   outside the catch on purpose, so a failure there rolls the whole batch back rather than
   being swallowed as poison. Measured cost on a 100k-row backlog drain: ~7%
   (178.6k rows/s -> 166.1k rows/s).

Tests: both hazards proven against the unfixed code first (FK parent registered silently;
an injected ledger-write abort left the row deleted with a live ledger entry). Adds FK
guard coverage (parent refused, child allowed, SET NULL / SET DEFAULT refused, NO ACTION /
RESTRICT allowed, unrelated cascades ignored) and two applier atomicity tests. The
statement-count lock-in moves 7 -> 9 for the two new savepoint statements.

Version 0.3.2 -> 0.3.3; README limitations + release notes document the constraint.
2026-08-19 16:22:44 -04:00
Joseph Doherty e9dc13689a docs(index): LocalDb 0.3.2 published — cap-prune session fault, sync keepalives, newest-wins stream guard (ScadaBridge arch-review P4-1 + P4-3); ScadaBridge pins 0.3.2 2026-08-19 15:06:47 -04:00
Joseph Doherty f62b017cc7 fix(localdb): fault live sync session on cap-prune; keepalive + newest-wins sync guard (0.3.2, ScadaBridge arch-review P4-1 + P4-3) 2026-08-19 15:02:04 -04:00
Joseph Doherty dbcbc5bd41 docs(index): LocalDb 0.3.1 published — handshake-fault control-lane flush (register row 41) 2026-08-18 07:30:59 -04:00
Joseph Doherty 1145df682e docs(index): OtOpcUa now registers LocalDb tables conditionally — baseline on enable, deregister self-heal on disable (row 39 closed) 2026-08-18 07:25:18 -04:00
Joseph Doherty 3e5a5fde81 fix(localdb): flush the control lane before a faulting session cancels its writer (0.3.1)
A session that faulted during the handshake cancelled its writer loop immediately,
which could kill its OWN Handshake while it still sat in the control lane: the
enqueue -> receive -> validate -> fault path has no yields, while the writer's
wakeup waits in the thread-pool queue, so with the peer's handshake already
buffered the cancel won the race almost every time (39/40 in the new pinning
test). Mismatch detection is symmetric — the peer computes its fail-closed
verdict FROM our Handshake — so the healthy peer never learned anything: in
tests it waited out the full 15 s receive cap (the register-row-41 flake on
SyncSessionTests.Handshake_DigestMismatch_FailsClosed), and in production it
would misclassify a permanent schema mismatch as a generic disconnect and
reconnect-loop at the fast backoff instead of taking the classified
FailedPrecondition -> LogError + max-backoff path.

RunAsync's fault teardown now completes both lanes so the writer drains what is
already enqueued and exits on its own, with a 2 s deferred cancel as the backstop
for a wedged transport. Caller-initiated cancellation is unchanged (the linked
token is already cancelled, so shutdown is not delayed), and nothing on the wire
changes — a 0.3.0 peer simply receives the Handshake it was always owed.

New SyncSessionTests.Handshake_FaultTeardown_StillTransmitsTheLocalHandshake pins
the race (40 concurrent preloaded-mismatch sessions; 39/40 red pre-fix, green
post-fix).

BatchBudgetTests hardened separately: the register's other flaky test
(Pump_RowLargerThanTheBudget_IsSentAloneRatherThanStallingTheStream) awaits the
scripted-peer channel without ever observing the session task, so ANY session
fault surfaced as an opaque 15 s timeout. A run-observing NextAsync overload now
races the session task at every read site in the class, so the next flake reports
the session's real exception instead. (The handshake-race root cause does not
structurally apply to this test — no handshake fault occurs there — so its
intermittent timeout is attributed to the fault-hiding wait, now removed.)

Suite 200/200 green. Closes ScadaBridge deferred-work register row 41 (cross-repo).
2026-08-18 07:11:14 -04:00
Joseph Doherty b62981ff80 fix(build): anchor pack/push scripts to their own repo root (0038d46 pattern, all seven)
Secrets/push.sh, Theme/pack.sh+push.sh, Audit/pack.sh+push.sh and Auth/pack.sh+push.sh
all carried the CWD-pack bug LocalDb's push.sh was cured of in 0038d46: a bare
'dotnet pack -c Release -o ./artifacts' packs whatever solution sits in the CALLER's
working directory, and the push scripts then shove the caller's ./artifacts/*.nupkg
onto the live Gitea feed (the incident that briefly published 25 unrelated packages).

Each script now cds to its own repo root, names its .slnx explicitly, writes to an
absolute artifacts path, and (push.sh) pushes an anchored glob. Verified by bash -n on
all seven plus a positive control: Audit's pack.sh run from a foreign CWD packed only
ZB.MOM.WW.Audit into its own artifacts dir and left the caller's directory untouched.
Push scripts deliberately NOT executed (they publish); pack half verified only.

Closes ScadaBridge deferred-work register row 42 (cross-repo).
2026-08-18 07:00:42 -04:00
Joseph Doherty 31a7c8a88a docs(index): LocalDb consumer pins — OtOpcUa 0.1.3 -> 0.2.1 (R4 merged), skew note updated for 0.3.0 2026-08-15 04:15:28 -04:00
Joseph Doherty 7757b9c6fc Merge branch 'localdb-per-table-snapshot' — LocalDb 0.3.0: per-table needs_snapshot, capability-negotiated partial snapshots (ScadaBridge R1) 2026-08-15 04:15:02 -04:00
Joseph Doherty 0038d46c9a fix(build): push.sh must pack ITSELF, not the caller's working directory
dotnet pack with no project argument packs whatever solution sits in the caller's
CWD, so invoking this script by absolute path from another repo's checkout packed
and published 25 unrelated packages to the live Gitea feed under its name. Now it
cds to its own repo root and names the solution and an absolute output path.
2026-08-15 03:36:48 -04:00
Joseph Doherty cea3930b94 feat(localdb): per-table needs_snapshot (0.3.0, on-disk schema v3)
Baselining one table used to set a single per-DATABASE needs_snapshot bit, and a
snapshot streams every registered table in full — so opting one table into
replication late re-sent all of them, in both directions at once on a pair
enabling replication late. On a database holding a high-volume table (an event
log at its retention cap) that is the largest transfer this library performs.
0.2.1 documented the cost and deferred the fix as needing an on-disk change.

Schema v3 adds __localdb_snapshot_state (table_name PK, needs_snapshot), created
by the shared DDL with IF NOT EXISTS so an older file gains it on open; the v2->v3
transition is a version stamp only. RegisterReplicated(..., baselineExistingRows:
true) now flags that table alone. The database-wide flag survives unchanged for
the paths that genuinely reach every table — a cap prune or an impossible peer ack
loses delta continuity for the oplog as a whole. DeregisterReplicated drops a
table's pending flag with its ledger rows, since nothing would ever stream (and so
clear) it afterwards.

Wire compatibility is additive and negotiated by CAPABILITY, not version:
lib_schema_version stays 1 because ValidateHandshake compares it fail-closed for
equality, so bumping it would stop 0.1.x/0.2.x pairs syncing outright. Handshake
gains supports_partial_snapshot (field 5) and SnapshotBegin gains a tables scope
(field 2) where EMPTY means all tables — precisely what a pre-0.3.0 sender emits
and what a pre-0.3.0 receiver assumes. A peer that does not advertise the
capability is served a full snapshot: a scoped one would make it advance
last_applied_remote_seq to the snapshot's as-of and silently discard every delta
below it for the tables left out.

For the same reason a partial snapshot moves no watermark on either side. The
receiver keeps last_applied_remote_seq (carrying only the HLC across, which is
MAX-guarded), and the sender keeps _sentThruSeq, so the deltas the snapshot
overlaps are still pumped and merged by LWW.

Crash safety: snapshot debt now clears on the receiver's new SnapshotAck rather
than on enqueue, so an ungraceful exit part-way through re-streams the same scope
on reconnect. The ack rides the unbounded CONTROL lane like DeltaAck, so it
composes with 0.2.1's bounded 64-message inbox and its out-of-handshake snapshot.
Against a pre-0.3.0 peer the 0.2.x clear-on-enqueue behaviour is kept — there is
no ack to wait for — with the handshake's horizon gap check as the net.

Tests 199/199 (was 188): single-table baseline streams only that table, mixed-
version interop (peer without the capability gets a full snapshot), debt standing
until SnapshotAck, partial snapshot not advancing the receiver's watermark, no
SnapshotAck emitted to an old peer, end-to-end over two real databases, plus v2->v3
upgrade and per-table flag isolation.

README, ILocalDb remarks (which named this as the follow-up) and the umbrella
CLAUDE.md index updated.
2026-08-15 03:34:54 -04:00
34 changed files with 1874 additions and 137 deletions
+3 -3
View File
File diff suppressed because one or more lines are too long
+7 -1
View File
@@ -1,4 +1,10 @@
#!/usr/bin/env bash
# pack.sh — produce the ZB.MOM.WW.Audit NuGet package into ./artifacts.
set -euo pipefail
dotnet pack -c Release -o ./artifacts
# cd to the repo root and name the solution + an absolute output path: `dotnet pack` with
# neither a project argument nor a fixed working directory packs whatever solution happens
# to sit in the CALLER's directory (the bug behind the LocalDb feed incident, 0038d46).
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$REPO_ROOT"
dotnet pack ZB.MOM.WW.Audit.slnx -c Release -o "$REPO_ROOT/artifacts"
+13 -2
View File
@@ -16,9 +16,20 @@ set -euo pipefail
: "${GITEA_NUGET_SOURCE:?set GITEA_NUGET_SOURCE to your Gitea NuGet feed URL}"
: "${GITEA_NUGET_KEY:?set GITEA_NUGET_KEY to your Gitea access token}"
dotnet pack -c Release -o ./artifacts
# Named solution, named output, and a cd to the repo root, all for one reason: `dotnet pack`
# with neither a project argument nor a fixed working directory packs whatever solution
# happens to sit in the CALLER's directory — and then pushes it to a live feed under this
# script's name. Invoked by absolute path from another repo's checkout, the relative form
# packed and published 25 unrelated packages (LocalDb's copy of this script, fixed in
# 0038d46; recovered with `dotnet nuget delete`, but only because the token happened to
# carry delete rights). A publish script must be unambiguous about WHAT it publishes no
# matter where it is run.
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$REPO_ROOT"
dotnet nuget push "./artifacts/*.nupkg" \
dotnet pack ZB.MOM.WW.Audit.slnx -c Release -o "$REPO_ROOT/artifacts"
dotnet nuget push "$REPO_ROOT/artifacts/*.nupkg" \
--source "$GITEA_NUGET_SOURCE" \
--api-key "$GITEA_NUGET_KEY" \
--skip-duplicate
+7 -1
View File
@@ -6,4 +6,10 @@
set -euo pipefail
dotnet pack -c Release -o ./artifacts
# cd to the repo root and name the solution + an absolute output path: `dotnet pack` with
# neither a project argument nor a fixed working directory packs whatever solution happens
# to sit in the CALLER's directory (the bug behind the LocalDb feed incident, 0038d46).
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$REPO_ROOT"
dotnet pack ZB.MOM.WW.Auth.slnx -c Release -o "$REPO_ROOT/artifacts"
+13 -2
View File
@@ -16,9 +16,20 @@ set -euo pipefail
: "${GITEA_NUGET_SOURCE:?set GITEA_NUGET_SOURCE to your Gitea NuGet feed URL}"
: "${GITEA_NUGET_KEY:?set GITEA_NUGET_KEY to your Gitea access token}"
dotnet pack -c Release -o ./artifacts
# Named solution, named output, and a cd to the repo root, all for one reason: `dotnet pack`
# with neither a project argument nor a fixed working directory packs whatever solution
# happens to sit in the CALLER's directory — and then pushes it to a live feed under this
# script's name. Invoked by absolute path from another repo's checkout, the relative form
# packed and published 25 unrelated packages (LocalDb's copy of this script, fixed in
# 0038d46; recovered with `dotnet nuget delete`, but only because the token happened to
# carry delete rights). A publish script must be unambiguous about WHAT it publishes no
# matter where it is run.
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$REPO_ROOT"
dotnet nuget push "./artifacts/*.nupkg" \
dotnet pack ZB.MOM.WW.Auth.slnx -c Release -o "$REPO_ROOT/artifacts"
dotnet nuget push "$REPO_ROOT/artifacts/*.nupkg" \
--source "$GITEA_NUGET_SOURCE" \
--api-key "$GITEA_NUGET_KEY" \
--skip-duplicate
+1 -1
View File
@@ -5,7 +5,7 @@
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<LangVersion>latest</LangVersion>
<Version>0.2.1</Version>
<Version>0.3.3</Version>
<ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<!-- NU1510 (package-pruning advice): the Replication and Tests projects carry a
+112 -17
View File
@@ -10,7 +10,7 @@ Change capture uses SQLite triggers writing a single HLC-stamped oplog inside th
consumer's own write transaction (crash-consistent, non-bypassable). Convergence is
last-writer-wins over a hybrid logical clock. All timestamps are UTC.
## Packages (all `0.2.1`)
## Packages (all `0.3.3`)
| Package | Contents |
|---|---|
@@ -20,6 +20,74 @@ last-writer-wins over a hybrid logical clock. All timestamps are UTC.
> Status: **published to the Gitea feed; adopted by ScadaBridge (Phases 1+2) and OtOpcUa (Phase 1).**
>
> - `0.3.3` (2026-08-19) — one Low finding from the ScadaBridge adversarial architecture review
> (P4-7), in two halves. Wire-compatible with every earlier peer: no proto change, no
> `lib_schema_version` bump, no on-disk schema change.
> - **`RegisterReplicated` now refuses a table that another table cascades a DELETE onto.**
> Applying a remote row uses `INSERT OR REPLACE`, whose implicit DELETE fires `ON DELETE
> CASCADE` / `SET NULL` / `SET DEFAULT` on every child — while the apply path holds the capture
> guard raised, so the child rows it touches produce no oplog entry, the peer keeps them, and the
> pair diverges with nothing reporting a fault. Registration is the only place that is visible,
> so it is refused there with an actionable `LocalDbRegistrationException` naming the referencing
> table and its action. `NO ACTION` / `RESTRICT` are still allowed: they mutate no child row (the
> implicit delete simply fails, which the applier dead-letters — loud, not silent). No current
> consumer has a foreign key on a replicated table, so this is a guard against a future foot-gun
> rather than a live fix.
> - **Applying one entry is atomic across its row write and its ledger write.** They were two
> statements under the poison catch, so a failure landing between them committed a
> deleted/replaced row carrying a STALE LIVE ledger version — the shape `SnapshotStreamer` warns
> about and skips, meaning that key never travels again. A per-entry `SAVEPOINT` makes the pair
> all-or-nothing while keeping the catch's per-ENTRY granularity: entries applied earlier in the
> same batch survive, and the poison entry is still dead-lettered and the batch still continues.
> - `0.3.2` (2026-08-19) — two Medium findings from the ScadaBridge adversarial architecture review
> (P4-1, P4-3). Wire-compatible with every earlier peer: no proto change, no `lib_schema_version`
> bump, no on-disk schema change — the keepalive is transport-level and the rest is local policy.
> - **A live session now ends itself when a cap prune eats deltas it has not sent.** The pump reads
> `seq > sentThruSeq`, which resumes SILENTLY above a pruned hole; the peer acks the later seqs
> and the pair is durably divergent for the whole life of that session. `needs_snapshot` (set in
> the prune's own transaction) only heals a RECONNECT, and nothing forced one — so on a stable
> LAN with a slow-but-alive peer the divergence was unbounded. Every pump iteration now compares
> the oplog horizon (`MIN(seq)`, one b-tree seek) against its sent watermark and throws
> `LocalDbOplogGapException` when deltas it still owes are gone. The horizon, not the flag,
> because the flag is also raised for reasons that do not invalidate the running session.
> - **The sync stream has HTTP/2 keepalive, and the server's single-stream guard is newest-wins.**
> Neither end ever timed out on silence (a converged pair is legitimately quiet), so a hard crash
> or partition of the dialling node left the passive node's handler parked in `MoveNext` on a
> half-open stream — holding the guard, reporting itself connected, and rejecting the rebooted
> peer's reconnects with `AlreadyExists`, indefinitely on an idle pair. The initiator now dials
> with `KeepAlivePingDelay` / `KeepAlivePingTimeout` (new options, 60s/20s, `Always` policy so
> idle connections ping too), and a fresh authenticated `Sync` cancels the incumbent and waits
> for it to unwind before starting — the guard serializes sessions, it was never a fence, and an
> authenticated same-key dialler is by definition the peer. An incumbent that will not yield
> within the takeover timeout still gets `AlreadyExists`, because two live sessions over one
> database is worse than a refused dial.
> - `0.3.1` (2026-08-18) — a faulting session flushes its control lane before cancelling its writer,
> so a locally-detected handshake mismatch actually reaches the peer instead of leaving it to
> reconnect-loop against a generic disconnect.
> - `0.3.0` (2026-08-15) — **the snapshot-resync flag is per-table.** On-disk bookkeeping schema
> goes to **v3** (new `__localdb_snapshot_state`, upgraded in place on open); wire-compatible with
> `0.1.x`/`0.2.x` peers, and `lib_schema_version` deliberately stays `1` because it is compared
> fail-closed for equality.
> - **`RegisterReplicated(table, baselineExistingRows: true)` now flags a resync for THAT TABLE.**
> Through `0.2.x` the flag was one bit for the whole database, so baselining one table
> re-streamed every registered table in full — and on a pair enabling replication late, both
> nodes did that to each other at once. A snapshot now streams only the flagged tables.
> - **The database-wide flag survives, unchanged, for the cases that genuinely need it**: an oplog
> cap prune or an impossible peer ack loses deltas for the oplog as a whole, so nothing short of
> every table is sound. What moved is only where a *baseline* records its debt.
> - **Additive wire extension, negotiated by capability, not by version.** `Handshake` gains
> `supports_partial_snapshot`; `SnapshotBegin` gains a `tables` scope where **empty means "all
> tables"** — which is precisely what a pre-`0.3.0` sender emits and what a pre-`0.3.0` receiver
> assumes. A peer that does not advertise the capability is served a **full** snapshot, because a
> scoped one would make it advance `last_applied_remote_seq` past deltas it never received.
> - **A partial snapshot moves no watermark on either side.** The receiver keeps
> `last_applied_remote_seq` (it is one number for the whole peer oplog, and the untouched tables'
> deltas still sit below the snapshot's as-of); the sender keeps pumping from where it was. The
> overlap re-delivers rows the snapshot already carried, which LWW absorbs.
> - **Snapshot debt now clears on acknowledgement, not on enqueue** (new `SnapshotAck`, exchanged
> only between `0.3.0` peers). A crash part-way through a snapshot re-streams the same scope on
> reconnect instead of losing it. Against a pre-`0.3.0` peer the `0.2.x` clear-on-enqueue
> behaviour is kept — there is no ack to wait for.
> - `0.2.1` (2026-08-14) — two adversarial-review findings against `0.2.0`. Wire-compatible with
> `0.1.x`/`0.2.0` peers; no proto change, no `lib_schema_version` bump, no on-disk schema change.
> - **`DeregisterReplicated` no longer leaves the HLC without a crash anchor.** Clock recovery on
@@ -177,10 +245,15 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
delta/snapshot batches), so receiving never blocks on the ability to ack.
- **Resync** — when a peer's watermark falls below the pruned oplog horizon, it is sent a
full snapshot of the registered tables (streamed, keyset-paged) and deltas resume above
the snapshot's as-of seq. **Each node decides from its OWN state** whether it owes the
peer a snapshot. The snapshot is streamed *after* the handshake, alongside a running
receive loop, so a node sending one still applies what the peer sends it — a mutual
snapshot exchanges under flow control instead of being buffered whole in memory.
the snapshot's as-of seq. When instead a single table was *baselined*, only that table is
streamed and **no watermark moves on either side** — a partial snapshot speaks for nothing
outside its own tables, so the deltas it overlaps are still pumped and merged by LWW.
**Each node decides from its OWN state** whether it owes the peer a snapshot, and a peer
that does not advertise `supports_partial_snapshot` is sent a full one. The snapshot is
streamed *after* the handshake, alongside a running receive loop, so a node sending one
still applies what the peer sends it — a mutual snapshot exchanges under flow control
instead of being buffered whole in memory. Between `0.3.0` peers the sender's flags are
cleared by the receiver's `SnapshotAck`, so a crash mid-snapshot re-streams it.
- **Maintenance** — a background service on **both** roles enforces the oplog row/age caps
and prunes expired tombstones every `MaintenanceInterval`.
@@ -206,20 +279,33 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
`RegisterReplicated`, because `json_object` cannot capture it). A *typeless* column can
still hold a BLOB value at runtime, which then surfaces as a write-time error, not a
registration error.
- **Replicated tables must not be cascade-delete foreign-key parents.** If another table
references the one you are registering with `ON DELETE CASCADE` (or `SET NULL` /
`SET DEFAULT`), `RegisterReplicated` throws. Applying a remote row uses `INSERT OR
REPLACE`, whose implicit DELETE fires that referential action while change capture is
suppressed — so the child rows it deletes or nulls are never captured, the peer keeps
them, and the pair diverges with nothing reporting a fault. Model the parent/child
relationship with `ON DELETE NO ACTION` (or `RESTRICT`) and delete children explicitly, or
leave the parent unreplicated. `NO ACTION` / `RESTRICT` are allowed because they mutate no
child row: the implicit delete just fails, and the applier dead-letters the entry — loud,
not silent. The check runs at **registration time**, so a cascading child table created
*later* is not caught (SQLite offers no DDL hook); register after schema setup, which is
where startup does it anyway.
- **CDC semantics** — rows that existed **before** `RegisterReplicated` are not captured
and are not snapshotted; only changes after registration are replicated. **Opt out with
`RegisterReplicated(table, baselineExistingRows: true)`**, which seeds the row-version
ledger for existing rows at the baseline floor (HLC `0`) and flags a snapshot resync so
they reach the peer. The floor is below every genuine stamp, so a baselined row loses
ledger for existing rows at the baseline floor (HLC `0`) and flags a snapshot resync
**for that table** so they reach the peer. The floor is below every genuine stamp, so a baselined row loses
every comparison against a real remote write and wins only where the peer has no version
of that key at all. Two nodes that baseline the same key both hold HLC `0` and resolve by
the node-id tie-break — convergent, but arbitrary as to which content survives.
- **The snapshot flag is per-database, not per-table.** Baselining one table flags a resync
that streams **every** registered table in full, and on a pair enabling replication late
both nodes do that to each other simultaneously — the largest transfer this library
performs. Bounded (batch budget out, bounded inbox in), but **purge or trim high-volume
tables such as event logs before enabling replication late**, and expect one slow first
sync. A per-table flag needs an on-disk schema change and is not in `0.2.x`.
- **A snapshot still ships whole tables.** The flag is per-table as of `0.3.0`, so baselining
one table streams that table and no other — but *that* table travels in full, and two
cases still stream **everything**: an oplog cap prune / impossible peer ack (delta
continuity is lost for the oplog as a whole), and any peer older than `0.3.0` (which
cannot be told a snapshot is partial). So **purge or trim high-volume tables such as event
logs before baselining them**, and expect one slow first sync. Bounded either way (batch
budget out, bounded inbox in).
- **REAL ±Infinity** is carried in SQLite's JSON dialect on the wire (fine for the
SQLite↔SQLite peers this library targets).
- `AddZbLocalDb` / `AddZbLocalDbReplication` are **first-registration-wins** (idempotent).
@@ -256,8 +342,15 @@ provider is wired or the poll threw, so a DB failure never reads as "0 backlog")
## Ops notes
- When the oplog row/age caps are exceeded, maintenance prunes to the ceiling and flags
`needs_snapshot` (disk safety over delta continuity); the peer then snapshot-resyncs.
- When the oplog row/age caps are exceeded, maintenance prunes to the ceiling and flags the
**database-wide** `needs_snapshot` (disk safety over delta continuity); the peer then gets a
full snapshot. Per-table debt lives separately in `__localdb_snapshot_state` and is what
baselining writes.
- **Upgrading a pair to `0.3.0`** needs no coordination: the two nodes negotiate scoped snapshots
by capability, so a `0.3.0` node paired with a `0.2.x` one simply keeps sending full snapshots
until both are upgraded. Rolling the file forward is one-way — a `0.2.x` build reopening a v3
file leaves `__localdb_snapshot_state` in place and ignores it, so any per-table debt recorded
by `0.3.0` is invisible to it (roll forward again, or re-baseline).
- **Turning a table off** — `DeregisterReplicated(table)` drops its capture triggers and prunes
its oplog + ledger rows. It must be applied **symmetrically**: the handshake compares
registered-table digests fail-closed, so a node that drops a table its peer still
@@ -281,14 +374,16 @@ provider is wired or the poll threw, so a DB failure never reads as "0 backlog")
## Testing
188 tests, all offline / no live dependencies — unit (HLC ordering + restart
199 tests, all offline / no live dependencies — unit (HLC ordering + restart
persistence, deregistration's HLC anchor surviving an ungraceful reopen, trigger SQL
generation, LWW decision table, watermark/pruning, digest computation, schema-version
upgrade, deregistration idempotency, baseline-floor LWW safety, O(1) depth vs a seeded
backlog, byte-budget chunking, bounded-inbox backpressure under a slow reader), integration
over two real SQLite DBs on an in-memory (and real) gRPC pipe (bidirectional convergence,
delete/update races, offline-accumulate → reconnect, prune → snapshot resync, mutual
snapshot over a capacity-1 transport), plus seeded randomized property tests (random interleaved op sequences → both DBs identical after quiesce).
snapshot over a capacity-1 transport, single-table baseline streaming only that table,
mixed-version interop where a peer without `supports_partial_snapshot` gets a full one,
snapshot debt surviving until the peer's `SnapshotAck`), plus seeded randomized property tests (random interleaved op sequences → both DBs identical after quiesce).
Family conventions apply (`TreatWarningsAsErrors`).
Build / test from this directory:
+12 -2
View File
@@ -16,9 +16,19 @@ set -euo pipefail
: "${GITEA_NUGET_SOURCE:?set GITEA_NUGET_SOURCE to your Gitea NuGet feed URL}"
: "${GITEA_NUGET_KEY:?set GITEA_NUGET_KEY to your Gitea access token}"
dotnet pack -c Release -o ./artifacts
# Named solution, named output, and a cd to the repo root, all for one reason: `dotnet pack`
# with neither a project argument nor a fixed working directory packs whatever solution
# happens to sit in the CALLER's directory — and then pushes it to a live feed under this
# script's name. Invoked as `bash /path/to/ZB.MOM.WW.LocalDb/build/push.sh` from another
# repo's checkout, the relative form packed and published 25 unrelated packages (recovered
# with `dotnet nuget delete`, but only because the token happened to carry delete rights).
# A publish script must be unambiguous about WHAT it publishes no matter where it is run.
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$REPO_ROOT"
dotnet nuget push "./artifacts/*.nupkg" \
dotnet pack ZB.MOM.WW.LocalDb.slnx -c Release -o "$REPO_ROOT/artifacts"
dotnet nuget push "$REPO_ROOT/artifacts/*.nupkg" \
--source "$GITEA_NUGET_SOURCE" \
--api-key "$GITEA_NUGET_KEY" \
--skip-duplicate
@@ -16,6 +16,7 @@ message SyncMessage {
SnapshotBegin snapshot_begin = 5; // sender switches to snapshot mode
SnapshotBatch snapshot_batch = 6;
SnapshotComplete snapshot_complete = 7;
SnapshotAck snapshot_ack = 8; // 0.3.0+; only sent when the peer advertised support
}
}
@@ -24,6 +25,13 @@ message Handshake {
uint32 lib_schema_version = 2;
repeated TableDigest tables = 3; // fail-closed on any mismatch
int64 last_applied_remote_seq = 4; // highest of YOUR oplog seqs I have applied
// 0.3.0+ capability, additive: I understand a snapshot scoped to a SUBSET of the registered
// tables (SnapshotBegin.tables) and I will answer a completed snapshot with a SnapshotAck.
// A pre-0.3.0 peer leaves this false — proto3 default — and a 0.3.0 sender then promotes any
// scoped snapshot to a full one and clears its flags on enqueue, exactly as 0.2.x did. This is
// why lib_schema_version stays 1: it is compared fail-closed for EQUALITY, so bumping it would
// stop 0.1.x/0.2.x pairs syncing outright.
bool supports_partial_snapshot = 5;
}
message TableDigest { string table_name = 1; string digest = 2; }
message HandshakeAck { string node_id = 1; bool snapshot_required = 2; }
@@ -40,7 +48,12 @@ message OplogEntry {
message DeltaBatch { repeated OplogEntry entries = 1; }
message DeltaAck { int64 applied_thru_seq = 1; }
message SnapshotBegin { int64 as_of_seq = 1; }
// tables: the SCOPE of this snapshot (0.3.0+). EMPTY means every registered table — a full
// snapshot, which is the only thing a pre-0.3.0 sender ever emits and the only thing a
// pre-0.3.0 receiver can safely assume. A non-empty list means the snapshot is PARTIAL, so the
// receiver must NOT advance last_applied_remote_seq to as_of_seq (deltas below it for the tables
// NOT listed have not been delivered); it merges the rows through LWW and leaves the watermark.
message SnapshotBegin { int64 as_of_seq = 1; repeated string tables = 2; }
message SnapshotBatch { string table_name = 1; repeated SnapshotRow rows = 2; }
message SnapshotRow {
string pk_json = 1;
@@ -50,3 +63,8 @@ message SnapshotRow {
bool is_tombstone = 5;
}
message SnapshotComplete { int64 as_of_seq = 1; }
// Sent by the RECEIVER once a snapshot has been durably applied, so the sender clears the
// per-table snapshot flags only after the peer actually holds the rows (a crash mid-snapshot then
// re-streams the same scope on reconnect). Only exchanged when both peers advertised
// supports_partial_snapshot; a pre-0.3.0 pair keeps the 0.2.x clear-on-enqueue behaviour.
message SnapshotAck { int64 as_of_seq = 1; }
@@ -14,6 +14,8 @@ internal sealed record ApplyResult(long AppliedThruSeq, int Applied, int Discard
/// row-version (hlc, then node_id ordinal tie-break), poison rows are dead-lettered while the
/// batch continues, and the remote watermark advances in the SAME transaction as the data so a
/// crash-replay of the batch is a no-op (LWW's strict-greater compare loses on equal versions).
/// Each entry's row write and its ledger write are additionally scoped by a per-entry SAVEPOINT, so
/// the pair is all-or-nothing even when the poison catch keeps the batch alive around it.
/// This never writes <c>__localdb_oplog</c>: the 2-node topology does not re-forward applied deltas.
/// </summary>
/// <remarks>
@@ -88,6 +90,18 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
continue;
}
// The row mutation and its ledger entry are TWO statements and must land together or
// not at all: a failure between them commits a deleted/replaced row carrying a stale
// ledger version, which SnapshotStreamer then warns about and skips — so that key never
// travels again and the pair is silently, permanently divergent for it. A per-entry
// savepoint makes the pair atomic without giving up the poison catch's per-ENTRY
// granularity (the surrounding transaction, and every entry already applied in it,
// survives a rollback to the savepoint). Cost measured on the path that matters — a
// 100 000-row backlog drain — is ~7% (178.6k rows/s → 166.1k rows/s): the two extra
// statements are prepared once like the rest and neither one touches the disk, since
// an inner savepoint's release just merges into the enclosing transaction.
var pairApplied = false;
await sql.SavepointAsync(ct);
try
{
var table = db.ReplicatedTables[entry.TableName];
@@ -97,7 +111,7 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
await sql.UpsertAsync(table, entry.RowJson!, ct);
await sql.WriteRowVersionAsync(entry, utcNow, ct);
applied++;
pairApplied = true;
}
catch (SqliteException ex)
{
@@ -105,9 +119,21 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
// malformed row_json) are treated as per-entry poison — SQLite aborts just the
// statement, not the transaction, so the batch continues after dead-lettering.
// Any other exception propagates and rolls back the whole batch (fail-closed).
// The rollback comes FIRST so the dead-letter row it records is not discarded with
// the half-applied pair.
await sql.RollbackToSavepointAsync(ct);
await sql.DeadLetterAsync(entry, ex.Message, utcNow, ct);
deadLettered++;
}
// Released outside the catch on purpose: a RELEASE that itself fails means the
// transaction is no longer sound, and there is no savepoint state left to reason about,
// so it propagates and rolls the whole batch back rather than being swallowed as poison.
if (pairApplied)
{
await sql.ReleaseSavepointAsync(ct);
applied++;
}
}
await sql.AdvancePeerStateAsync(maxSeq, maxHlc, utcNow, ct);
@@ -143,10 +169,42 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
private SqliteCommand? _rowVersion;
private SqliteCommand? _deadLetter;
private SqliteCommand? _peerState;
private SqliteCommand? _savepoint;
private SqliteCommand? _releaseSavepoint;
private SqliteCommand? _rollbackToSavepoint;
/// <summary>The per-entry savepoint's name. Reused for every entry — one is open at a time.</summary>
private const string SavepointName = "__localdb_apply_entry";
/// <summary>How many distinct statements this set has prepared.</summary>
public int StatementCount => _owned.Count;
/// <summary>Opens the per-entry savepoint that scopes one entry's row + ledger writes.</summary>
public Task SavepointAsync(CancellationToken ct)
{
var cmd = _savepoint ??= Create("SAVEPOINT " + SavepointName);
return cmd.ExecuteNonQueryAsync(ct);
}
/// <summary>Commits the entry's savepoint into the surrounding batch transaction.</summary>
public Task ReleaseSavepointAsync(CancellationToken ct)
{
var cmd = _releaseSavepoint ??= Create("RELEASE " + SavepointName);
return cmd.ExecuteNonQueryAsync(ct);
}
/// <summary>
/// Undoes everything the entry did and closes its savepoint. ROLLBACK TO leaves the savepoint
/// OPEN, so the RELEASE is required — without it the names would nest one deeper per poison
/// entry and the next rollback would unwind more than one entry's work.
/// </summary>
public async Task RollbackToSavepointAsync(CancellationToken ct)
{
var cmd = _rollbackToSavepoint ??= Create("ROLLBACK TO " + SavepointName);
await cmd.ExecuteNonQueryAsync(ct);
await ReleaseSavepointAsync(ct);
}
public Task SetApplyingAsync(int value, CancellationToken ct)
{
var cmd = _applying ??= Create(
@@ -24,8 +24,9 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
/// <summary>
/// Reads up to <paramref name="maxBatch"/> oplog entries with <c>seq &gt; afterSeq</c> in ascending
/// seq order. Caller contract: check <c>needs_snapshot</c> before trusting incremental reads —
/// after cap-pruning, reads silently resume past a gap.
/// seq order. Caller contract: this read resumes SILENTLY past a cap-pruned gap, so a caller
/// streaming incrementally must first confirm its position is still below the horizon — see
/// <see cref="GetMinSeqAsync"/>, which <c>SyncSession</c>'s pump checks on every iteration.
/// </summary>
public Task<IReadOnlyList<OplogEntryRecord>> ReadBatchAboveAsync(long afterSeq, int maxBatch, CancellationToken ct = default) =>
db.QueryAsync(
@@ -148,6 +149,24 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
return rows[0];
}
/// <summary>
/// The oldest oplog seq still present — the delta horizon — or 0 when the oplog is empty.
/// </summary>
/// <remarks>
/// One b-tree seek (<c>seq</c> is the rowid and SQLite optimizes MIN into a seek), which is what
/// makes it affordable for the pump to check on every flush interval. Compared against a
/// session's sent watermark it answers the only question the pump has: is everything I have not
/// yet sent still here? Ack-pruning deletes <c>seq &lt;= last_acked_seq</c> and the peer can only
/// ack what it received, so in a healthy session the horizon never climbs past
/// <c>sentThruSeq + 1</c>; a higher one means a cap prune ate unsent deltas.
/// </remarks>
public async Task<long> GetMinSeqAsync(CancellationToken ct = default)
{
var rows = await db.QueryAsync(
"SELECT COALESCE(MIN(seq), 0) FROM __localdb_oplog", static r => r.GetInt64(0), null, ct);
return rows[0];
}
/// <summary>The highest oplog seq present, or 0 when the oplog is empty.</summary>
public async Task<long> GetMaxSeqAsync(CancellationToken ct = default)
{
@@ -233,10 +252,61 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
db.ExecuteAsync(
"UPDATE __localdb_peer_state SET last_applied_remote_seq = 0, last_seen_hlc = 0 WHERE id = 1", null, ct);
/// <summary>Sets or clears the snapshot-resync-required flag.</summary>
/// <summary>
/// Sets or clears the DATABASE-WIDE snapshot-resync flag — the one that means "every registered
/// table". Set by the oplog-horizon paths (cap prune, impossible peer ack), which invalidate
/// delta continuity for all tables at once. Per-table flags live in
/// <c>__localdb_snapshot_state</c>; see <see cref="GetTablesNeedingSnapshotAsync"/>.
/// </summary>
public Task SetNeedsSnapshotAsync(bool value, CancellationToken ct = default) =>
db.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = @v WHERE id = 1", new { v = value ? 1 : 0 }, ct);
/// <summary>
/// The tables individually flagged as owing the peer a snapshot (schema v3), in ordinal name
/// order. Independent of the database-wide flag: baselining one table flags only that table.
/// </summary>
public Task<IReadOnlyList<string>> GetTablesNeedingSnapshotAsync(CancellationToken ct = default) =>
db.QueryAsync(
"SELECT table_name FROM __localdb_snapshot_state WHERE needs_snapshot <> 0 ORDER BY table_name",
static r => r.GetString(0), null, ct);
/// <summary>
/// Clears the snapshot debt a completed snapshot discharged: the named tables when
/// <paramref name="tables"/> is non-empty, otherwise <b>everything</b> — the database-wide flag
/// and every per-table row, because a full snapshot subsumes each of them.
/// </summary>
/// <remarks>
/// <para>
/// Rows are deleted rather than zeroed: absence is the resting state, and keeping a row per
/// table that has ever been baselined would accumulate bookkeeping that means nothing. Runs in
/// one transaction so a full snapshot's two clears cannot half-apply and leave the pair
/// re-snapshotting on every reconnect.
/// </para>
/// <para>
/// A scoped clear deliberately does NOT touch the database-wide flag. If the cap enforcer
/// flagged a whole-database resync while this snapshot was in flight, that debt is real and
/// unrelated — the snapshot only carried some of the tables — so it must survive. The full
/// clear may take it down because a full snapshot carries every row state as of its as-of seq,
/// which is exactly what the pruned deltas would have delivered.
/// </para>
/// </remarks>
public async Task ClearSnapshotDebtAsync(IReadOnlyList<string> tables, CancellationToken ct = default)
{
await using var tx = await db.BeginTransactionAsync(ct);
if (tables.Count == 0)
{
await tx.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 0 WHERE id = 1", null, ct);
await tx.ExecuteAsync("DELETE FROM __localdb_snapshot_state", null, ct);
}
else
{
foreach (var table in tables)
await tx.ExecuteAsync(
"DELETE FROM __localdb_snapshot_state WHERE table_name = @t", new { t = table }, ct);
}
await tx.CommitAsync(ct);
}
/// <summary>
/// Records the highest remote seq applied locally plus the peer HLC observed with it; also stamps
/// <c>last_sync_utc</c>. Both watermarks are MAX-guarded (monotonic): a snapshot completing with a
@@ -19,21 +19,57 @@ internal interface ISnapshotApplier
}
/// <summary>
/// Streams a full snapshot of every registered table to a peer that has fallen behind the pruned
/// oplog horizon. Reads the authoritative per-row version from <c>__localdb_row_version</c> joined
/// Which tables a snapshot covers. An EMPTY <see cref="Tables"/> is the wire's own encoding of
/// "all registered tables" (see <c>SnapshotBegin.tables</c>) and is what every pre-0.3.0 peer both
/// sends and expects, so it is deliberately the same value here rather than a separate flag.
/// </summary>
/// <remarks>
/// The distinction is not cosmetic: a FULL snapshot carries every row state as of its
/// <c>as_of_seq</c>, so the receiver may jump its watermark there and the sender may skip the
/// deltas it covered. A PARTIAL one carries only the listed tables, so neither side may move a
/// watermark that also speaks for the tables it left out.
/// </remarks>
internal sealed record SnapshotScope(IReadOnlyList<string> Tables)
{
/// <summary>Every registered table — the only scope a pre-0.3.0 peer can be sent.</summary>
public static readonly SnapshotScope Full = new([]);
public bool IsPartial => Tables.Count > 0;
public bool Includes(string tableName) =>
Tables.Count == 0 || Tables.Contains(tableName, StringComparer.Ordinal);
}
/// <summary>
/// Streams a snapshot of the tables in the requested <see cref="SnapshotScope"/> — every registered
/// table for a peer that has fallen behind the pruned oplog horizon, or just the baselined ones for
/// a scoped resync. Reads the authoritative per-row version from <c>__localdb_row_version</c> joined
/// to each live table, pages it into <see cref="SnapshotBatch"/> messages through the gated data
/// lane, and brackets them with SnapshotBegin/SnapshotComplete carrying the max oplog seq as-of.
/// </summary>
/// <remarks>
/// Clearing the snapshot debt is deliberately NOT done here: whether the flags may be cleared at all
/// depends on the peer's capability (a 0.3.0 peer acknowledges the snapshot, a pre-0.3.0 one cannot),
/// which is session state. <see cref="SyncSession"/> owns that decision; this type only streams.
/// </remarks>
internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, ReplicationOptions options, ILogger logger)
{
/// <summary>Streams the full snapshot; returns the as-of oplog seq it covers, so the session can skip already-covered tail deltas.</summary>
public async Task<long> SendAsync(Func<SyncMessage, CancellationToken, Task> send, CancellationToken ct)
/// <summary>
/// Streams the snapshot for <paramref name="scope"/>; returns the as-of oplog seq it covers, so
/// the session can skip already-covered tail deltas (a FULL snapshot only — a partial one speaks
/// for nothing outside its tables).
/// </summary>
public async Task<long> SendAsync(SnapshotScope scope, Func<SyncMessage, CancellationToken, Task> send, CancellationToken ct)
{
var asOfSeq = await store.GetMaxSeqAsync(ct);
await send(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = asOfSeq } }, ct);
var begin = new SnapshotBegin { AsOfSeq = asOfSeq };
begin.Tables.AddRange(scope.Tables);
await send(new SyncMessage { SnapshotBegin = begin }, ct);
var totalRows = 0L;
foreach (var table in db.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal))
foreach (var table in db.ReplicatedTables.Values
.Where(t => scope.Includes(t.Name))
.OrderBy(t => t.Name, StringComparer.Ordinal))
{
var sql = BuildPageSql(table);
// Keyset cursor (pk_json > last seen), not LIMIT/OFFSET: concurrent inserts/prunes
@@ -111,16 +147,9 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
}
await send(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = asOfSeq } }, ct);
// The sender owed this snapshot (its own needs_snapshot); clearing it here — after the whole
// snapshot is on the wire — stops it re-streaming on the next reconnect. Narrow window: the
// messages are enqueued on the data lane, not yet transport-delivered, so a crash/drop after
// this clear could lose the snapshot with the flag already down. The reconnect handshake's
// gap check (peer watermark below the pruned horizon) re-detects that in every case where the
// oplog is non-empty; the sole undetectable corner is an EMPTY oplog (nothing to gap-check),
// where the peer stays behind until new local writes recreate a horizon. Accepted: closing it
// would need transport-level delivery acks for snapshot frames.
await store.SetNeedsSnapshotAsync(false, ct);
logger.LogInformation("Snapshot sent (as_of_seq {AsOfSeq}, {Rows} rows).", asOfSeq, totalRows);
logger.LogInformation(
"Snapshot sent (as_of_seq {AsOfSeq}, {Rows} rows, scope {Scope}).",
asOfSeq, totalRows, scope.IsPartial ? string.Join(",", scope.Tables) : "all");
return asOfSeq;
}
@@ -181,6 +210,7 @@ internal sealed class SnapshotApplier : ISnapshotApplier
private long _asOfSeq;
private long _rows;
private long _maxHlc;
private bool _partial;
// db is part of the collaborator's canonical shape (mirrors the streamer) even though the merge
// reaches the database through the applier and store; kept for construction symmetry in the factory.
@@ -197,7 +227,13 @@ internal sealed class SnapshotApplier : ISnapshotApplier
_asOfSeq = begin.AsOfSeq;
_rows = 0;
_maxHlc = 0;
_logger.LogInformation("Snapshot receive begun (as_of_seq {AsOfSeq}).", _asOfSeq);
// A non-empty table list scopes the snapshot; empty (which is all a pre-0.3.0 sender can
// emit) means every registered table. Read here rather than on complete because
// SnapshotComplete does not repeat it — the receive loop guarantees a Begin precedes both.
_partial = begin.Tables.Count > 0;
_logger.LogInformation(
"Snapshot receive begun (as_of_seq {AsOfSeq}, scope {Scope}).",
_asOfSeq, _partial ? string.Join(",", begin.Tables) : "all");
return Task.CompletedTask;
}
@@ -219,9 +255,20 @@ internal sealed class SnapshotApplier : ISnapshotApplier
public async Task OnCompleteAsync(SnapshotComplete complete, CancellationToken ct)
{
// A PARTIAL snapshot may not move the inbound seq watermark. That watermark is one number
// for the whole peer oplog, and jumping it to as_of_seq would declare every table applied
// through that point — including the ones this snapshot deliberately left out, whose deltas
// below as_of_seq the sender would then never resend. Only the HLC is carried across (seq 0
// is a no-op against the MAX guard), so the clock still cannot regress below what the peer
// has stamped. The sender makes the mirror-image concession: it does not advance its own
// sent-thru watermark past a partial snapshot, so those deltas are still pumped normally
// and merely re-deliver rows the snapshot already carried, which LWW absorbs.
var seq = _partial ? 0 : complete.AsOfSeq;
// SetLastAppliedRemoteSeqAsync is MAX-guarded, so a delta applied mid-snapshot with a seq
// above the snapshot's as-of (or an HLC above what the snapshot carried) is never regressed.
await _store.SetLastAppliedRemoteSeqAsync(complete.AsOfSeq, _maxHlc, ct);
_logger.LogInformation("Snapshot receive complete (as_of_seq {AsOfSeq}, {Rows} rows).", complete.AsOfSeq, _rows);
await _store.SetLastAppliedRemoteSeqAsync(seq, _maxHlc, ct);
_logger.LogInformation(
"Snapshot receive complete (as_of_seq {AsOfSeq}, {Rows} rows, partial {Partial}).",
complete.AsOfSeq, _rows, _partial);
}
}
@@ -27,6 +27,10 @@ internal sealed class SyncSession
// inbound batch rate (the peer's flow control), so its lane can safely be unbounded.
private const int DataLaneCapacity = 16;
// How long a faulting session's teardown lets the writer keep flushing already-enqueued
// messages before the backstop cancel; only ever reached when the transport itself stalls.
private static readonly TimeSpan FaultFlushGrace = TimeSpan.FromSeconds(2);
private readonly SqliteLocalDb _db;
private readonly OplogStore _store;
private readonly LwwApplier _applier;
@@ -47,6 +51,17 @@ internal sealed class SyncSession
// single receive-loop task, so no synchronization is needed.
private bool _snapshotReceiving;
// Whether the peer speaks the 0.3.0 snapshot extensions: scoped snapshots
// (SnapshotBegin.tables) and the SnapshotAck that answers a completed one. Written once during
// the handshake — which completes before either loop starts — then read-only.
private bool _peerSupportsPartialSnapshot;
// The scope of the snapshot this node is streaming, held until the peer's SnapshotAck discharges
// it; null on every session that owes none. Assigned on the pump task strictly BEFORE the
// snapshot's first message is enqueued and read on the receive loop when the ack lands, so the
// ordering is causal — volatile only to make the cross-thread publication explicit.
private SnapshotScope? _sentSnapshotScope;
public SyncSession(
SqliteLocalDb db, OplogStore store, LwwApplier applier, ReplicationOptions options, ILogger logger,
Func<DateTimeOffset>? utcNow = null, LocalDbMetrics? metrics = null, SyncStatus? status = null)
@@ -63,11 +78,12 @@ internal sealed class SyncSession
/// <summary>
/// Invoked when THIS node owes the peer a snapshot (its own snapshot-required computation, not
/// the peer's) before deltas resume. Receives a gated send delegate that enqueues on the bounded
/// data lane (never the raw transport); returns the as-of oplog seq the snapshot covers so the
/// pump can skip already-covered tail deltas. Wired to <see cref="SnapshotStreamer.SendAsync"/>.
/// the peer's) before deltas resume. Receives the negotiated <see cref="SnapshotScope"/> and a
/// gated send delegate that enqueues on the bounded data lane (never the raw transport); returns
/// the as-of oplog seq the snapshot covers so the pump can skip already-covered tail deltas.
/// Wired to <see cref="SnapshotStreamer.SendAsync"/>.
/// </summary>
public Func<Func<SyncMessage, CancellationToken, Task>, CancellationToken, Task<long>>? SnapshotSender { get; set; }
public Func<SnapshotScope, Func<SyncMessage, CancellationToken, Task>, CancellationToken, Task<long>>? SnapshotSender { get; set; }
/// <summary>
/// Consumes an inbound snapshot inline on the receive loop: SnapshotBegin opens the receive,
@@ -99,7 +115,7 @@ internal sealed class SyncSession
var writer = WriterLoopAsync(duplex, control.Reader, data.Reader, linked.Token);
try
{
var snapshotRequired = await HandshakeAsync(duplex, control.Writer, linked.Token);
var snapshotScope = await HandshakeAsync(duplex, control.Writer, linked.Token);
// The receive loop starts BEFORE any snapshot this node owes is streamed. That ordering
// is load-bearing, not incidental: a snapshot sent from inside the handshake left the
@@ -111,7 +127,7 @@ internal sealed class SyncSession
// ordinary flow control. Interleaving is already safe by design: snapshot rows and
// deltas both merge through LWW, and the snapshot's watermark write is MAX-guarded.
var receive = ReceiveLoopAsync(duplex, control.Writer, linked.Token);
var pump = SnapshotThenPumpLoopAsync(snapshotRequired, GatedDataSend, data.Writer, linked.Token);
var pump = SnapshotThenPumpLoopAsync(snapshotScope, GatedDataSend, data.Writer, linked.Token);
var loops = new[] { pump, receive, writer };
// The first COMPLETED task is the primary outcome. Cancel the siblings, observe every
@@ -128,19 +144,32 @@ internal sealed class SyncSession
{
// Handshake-phase faults (and the primary rethrow above) land here: the writer may
// still be running, so stop and observe it before the primary exception surfaces.
linked.Cancel();
// It must be allowed to DRAIN, not be cancelled outright: the Handshake enqueued at
// the top of HandshakeAsync can still be sitting in the control lane (the enqueue →
// receive → validate → fault path has no yields, while the writer's wakeup waits in
// the thread-pool queue), and that message is the only thing that lets the peer
// compute its own fail-closed verdict — mismatch detection is symmetric, so this
// node's locally-thrown verdict reaches the peer only as the Handshake it transmits.
// Cancelling first killed it almost every time the peer's handshake was already
// buffered, leaving the healthy peer to misread a schema mismatch as a generic
// disconnect and reconnect-loop at the fast backoff. Completing the lanes lets the
// writer send what is already enqueued and exit on its own; the deferred cancel is
// the backstop that keeps a wedged transport from holding teardown hostage.
control.Writer.TryComplete();
data.Writer.TryComplete();
linked.CancelAfter(FaultFlushGrace);
await DrainSecondaryAsync(writer);
throw;
}
}
/// <summary>
/// Runs the two-phase fail-closed handshake and returns whether THIS node owes the peer a full
/// snapshot (its own computation from its own state — the peer's answer drives the peer's send
/// direction, symmetrically). Sending that snapshot is deliberately NOT part of the handshake;
/// see <see cref="RunAsync"/>.
/// Runs the two-phase fail-closed handshake and returns the snapshot THIS node owes the peer —
/// null for none, otherwise the <see cref="SnapshotScope"/> to stream (its own computation from
/// its own state; the peer's answer drives the peer's send direction, symmetrically). Sending
/// that snapshot is deliberately NOT part of the handshake; see <see cref="RunAsync"/>.
/// </summary>
private async Task<bool> HandshakeAsync(
private async Task<SnapshotScope?> HandshakeAsync(
SyncDuplex duplex, ChannelWriter<SyncMessage> controlLane, CancellationToken ct)
{
var peerState = await _store.GetPeerStateAsync(ct);
@@ -158,6 +187,11 @@ internal sealed class SyncSession
ValidateHandshake(localHandshake, peerHandshake);
// Read before anything can act on it. A pre-0.3.0 peer leaves the field at its proto3
// default, so "false" covers both "explicitly unsupported" and "older build" without a
// version check — which is the point of advertising a capability rather than a version.
_peerSupportsPartialSnapshot = peerHandshake.SupportsPartialSnapshot;
// A different node id on the far end means the peer was replaced, not merely restarted — a
// rebuilt host, a restored-from-nothing volume. Its oplog numbers from 1 again, so the
// watermark we hold (in the OLD peer's seq space) would silently swallow its first N
@@ -191,16 +225,16 @@ internal sealed class SyncSession
_sentThruSeq = peerAppliedThruSeq;
Interlocked.Exchange(ref _peerAckedSeq, peerAppliedThruSeq);
var snapshotRequired = await ComputeSnapshotRequiredAsync(peerHandshake, peerState, ct);
var snapshotScope = await ComputeSnapshotScopeAsync(peerHandshake, peerState, ct);
await controlLane.WriteAsync(
new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = _db.NodeId, SnapshotRequired = snapshotRequired } }, ct);
new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = _db.NodeId, SnapshotRequired = snapshotScope is not null } }, ct);
var peerAckMsg = await ReceiveOneAsync(duplex, ct);
if (peerAckMsg.MsgCase != SyncMessage.MsgOneofCase.HandshakeAck)
throw new InvalidOperationException(
$"Replication protocol error: expected a HandshakeAck, got {peerAckMsg.MsgCase}.");
return snapshotRequired;
return snapshotScope;
}
/// <summary>
@@ -216,25 +250,55 @@ internal sealed class SyncSession
/// than racing the transport.
/// </remarks>
private async Task SnapshotThenPumpLoopAsync(
bool snapshotRequired, Func<SyncMessage, CancellationToken, Task> dataSend,
SnapshotScope? snapshotScope, Func<SyncMessage, CancellationToken, Task> dataSend,
ChannelWriter<SyncMessage> dataLane, CancellationToken ct)
{
if (snapshotRequired)
if (snapshotScope is not null)
{
if (SnapshotSender is null)
throw new NotSupportedException("snapshot required but no snapshot sender configured");
var snapshotAsOfSeq = await SnapshotSender(dataSend, ct);
// Published BEFORE the first frame is enqueued, so the peer's SnapshotAck can never
// arrive on the receive loop while this is still null.
Volatile.Write(ref _sentSnapshotScope, snapshotScope);
var snapshotAsOfSeq = await SnapshotSender(snapshotScope, dataSend, ct);
_metrics?.RecordSnapshotSent();
_status?.SetLastSyncUtc(_utcNow());
// The snapshot already carries every row state up to as-of; skipping the pump past it
// avoids re-sending tail deltas the peer would only discard via LWW.
if (snapshotAsOfSeq > _sentThruSeq)
// A FULL snapshot carries every row state up to as-of, so skipping the pump past it
// avoids re-sending tail deltas the peer would only discard via LWW. A PARTIAL one
// speaks only for its own tables, so the pump must still deliver every delta below
// as-of — the peer has not moved its watermark either (see SnapshotApplier.OnComplete).
if (!snapshotScope.IsPartial && snapshotAsOfSeq > _sentThruSeq)
_sentThruSeq = snapshotAsOfSeq;
// A peer that cannot acknowledge the snapshot leaves only the 0.2.x option: clear once
// the frames are on the data lane. The window that opens (enqueued, not yet delivered —
// a crash there loses the snapshot with the debt already discharged) is why 0.3.0 peers
// wait for the ack instead. The reconnect handshake's gap check re-detects it whenever
// the oplog is non-empty; an EMPTY oplog remains the undetectable corner.
if (!_peerSupportsPartialSnapshot)
await ClearSentSnapshotDebtAsync(ct);
}
await PumpLoopAsync(dataLane, ct);
}
/// <summary>
/// Discharges the snapshot debt this session's outbound snapshot covered, exactly once.
/// </summary>
/// <remarks>
/// Taking the scope out of the field before clearing makes this idempotent, which matters
/// because two paths reach it — the pre-0.3.0 clear-on-enqueue above and the peer's SnapshotAck
/// — and a peer that acknowledges anyway (a newer build than its handshake claimed) must not
/// clear a second time on top of flags a later baseline may already have re-raised.
/// </remarks>
private async Task ClearSentSnapshotDebtAsync(CancellationToken ct)
{
var scope = Interlocked.Exchange(ref _sentSnapshotScope, null);
if (scope is null)
return;
await _store.ClearSnapshotDebtAsync(scope.Tables, ct);
}
// The ONLY caller of duplex.Send. Control (handshake + acks) always preempts data so the peer's
// inbound processing is never starved of its acks behind a backlog of DeltaBatches.
private static async Task WriterLoopAsync(
@@ -292,9 +356,14 @@ internal sealed class SyncSession
while (true)
{
ct.ThrowIfCancellationRequested();
// BEFORE the read, because the read itself cannot tell a pruned gap from an idle oplog.
ThrowIfPruneGap(await _store.GetMinSeqAsync(ct));
var batch = await _store.ReadBatchAboveAsync(_sentThruSeq, _options.MaxBatchSize, ct);
if (batch.Count > 0)
{
// The exact guard on the rows about to go out, for a prune that landed between the
// probe and the read. Free (the seq is already in hand) and strictly tighter.
ThrowIfPruneGap(batch[0].Seq);
var index = 0;
while (index < batch.Count)
{
@@ -326,6 +395,40 @@ internal sealed class SyncSession
}
}
/// <summary>
/// Faults the session when the oplog horizon has climbed above this session's read position —
/// i.e. a cap prune deleted deltas between <see cref="_sentThruSeq"/> and
/// <paramref name="oldestAvailableSeq"/>. An empty oplog (0) is not a gap.
/// </summary>
/// <remarks>
/// <para>
/// <b>Why a live session has to die for this.</b> <c>EnforceCapsAsync</c> sets
/// <c>needs_snapshot</c> in the same transaction as the prune, so a RECONNECT heals — the
/// handshake serves a full snapshot instead of deltas. Nothing, however, made the live session
/// notice: the pump reads <c>seq &gt; _sentThruSeq</c>, which silently resumes above the hole, and
/// the peer acks those later seqs as if the pair were converged. On a stable LAN with a slow but
/// alive peer that session can run for days, and the pair is durably divergent for all of it —
/// the flag that would fix it is only ever read at handshake time. Tearing down is what turns a
/// persisted "needs snapshot" into an actual snapshot.
/// </para>
/// <para>
/// <b>Why the horizon, not the flag.</b> <c>needs_snapshot</c> is also set for reasons that do
/// not invalidate THIS session's position (and a full snapshot this session just streamed clears
/// it only when the peer's <c>SnapshotAck</c> lands, which is after the pump has started).
/// Comparing the horizon against the sent watermark asks the precise question instead, so a
/// session is faulted when — and only when — rows it still owes the peer are gone.
/// </para>
/// </remarks>
private void ThrowIfPruneGap(long oldestAvailableSeq)
{
if (oldestAvailableSeq == 0 || oldestAvailableSeq <= _sentThruSeq + 1)
return;
throw new LocalDbOplogGapException(
$"Oplog deltas above the sent watermark {_sentThruSeq} were pruned (oldest surviving seq " +
$"{oldestAvailableSeq}): this session can no longer stream incrementally without skipping " +
"them. Ending the session so the reconnect handshake serves a snapshot resync.");
}
/// <summary>
/// Serialized bytes one entry contributes to its batch message: the entry's own encoding plus
/// the repeated field's tag and length prefix, as protobuf itself computes them.
@@ -384,6 +487,20 @@ internal sealed class SyncSession
_snapshotReceiving = false;
_metrics?.RecordSnapshotReceived();
_status?.SetLastSyncUtc(_utcNow());
// Acknowledge only to a peer that asked for it. Sent AFTER the applier's
// durable watermark write, so the ack means "I hold these rows", which is
// what lets the sender drop its per-table flags. On the unbounded control
// lane, like DeltaAck: an ack must never queue behind a data backlog.
if (_peerSupportsPartialSnapshot)
await controlLane.WriteAsync(
new SyncMessage { SnapshotAck = new SnapshotAck { AsOfSeq = msg.SnapshotComplete.AsOfSeq } }, ct);
break;
case SyncMessage.MsgOneofCase.SnapshotAck:
// The peer has durably applied the snapshot THIS node streamed, so the
// tables it covered no longer owe one. Nothing to reconcile against the
// ack's as-of seq: only one snapshot is ever sent per session, so the scope
// held in _sentSnapshotScope is unambiguously the one being acknowledged.
await ClearSentSnapshotDebtAsync(ct);
break;
default:
throw new InvalidOperationException(
@@ -565,6 +682,9 @@ internal sealed class SyncSession
NodeId = _db.NodeId,
LibSchemaVersion = LibSchemaVersion,
LastAppliedRemoteSeq = peerState.LastAppliedRemoteSeq,
// Unconditional: this build always understands a scoped snapshot and always answers a
// completed one. It is a statement about the code, not about configuration or state.
SupportsPartialSnapshot = true,
};
foreach (var table in _db.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal))
handshake.Tables.Add(new TableDigest { TableName = table.Name, Digest = table.Digest });
@@ -595,11 +715,63 @@ internal sealed class SyncSession
"Replication handshake rejected (fail-closed): " + string.Join("; ", parts) + ".");
}
private async Task<bool> ComputeSnapshotRequiredAsync(Handshake peerHandshake, PeerState peerState, CancellationToken ct)
/// <summary>
/// Decides what snapshot this node owes the peer: null for none, <see cref="SnapshotScope.Full"/>
/// for every registered table, or a scoped snapshot naming just the tables that need one.
/// </summary>
/// <remarks>
/// <para>
/// The two sources of snapshot debt are genuinely different in reach, which is why they are
/// stored separately. An <b>oplog-horizon</b> gap — the peer's watermark below the oldest delta
/// still streamable, which is what a cap prune or an impossible peer ack leaves behind — means
/// deltas have been lost for the oplog as a whole, so nothing short of every table is sound. A
/// <b>baseline</b> means one table's pre-existing rows were never captured, and says nothing at
/// all about the others.
/// </para>
/// <para>
/// A scope that happens to name every registered table is returned as Full rather than as a
/// list: it covers exactly what a full snapshot covers, so it should also get a full snapshot's
/// watermark advance on both sides, and it needs no promotion for a pre-0.3.0 peer.
/// </para>
/// <para>
/// Flags naming tables that are not registered are ignored (they can only stream nothing).
/// <c>DeregisterReplicated</c> drops them with the table's ledger rows, so the surviving way to
/// hold one is to drop the SQL table without deregistering — which the handshake's fail-closed
/// digest comparison rejects for its own reasons.
/// </para>
/// </remarks>
private async Task<SnapshotScope?> ComputeSnapshotScopeAsync(Handshake peerHandshake, PeerState peerState, CancellationToken ct)
{
if (peerState.NeedsSnapshot)
return true;
if (peerState.NeedsSnapshot || await HasHorizonGapAsync(peerHandshake, peerState, ct))
return SnapshotScope.Full;
var flagged = await _store.GetTablesNeedingSnapshotAsync(ct);
if (flagged.Count == 0)
return null;
var scoped = flagged.Where(_db.ReplicatedTables.ContainsKey).ToList();
if (scoped.Count == 0)
return null;
if (scoped.Count >= _db.ReplicatedTables.Count)
return SnapshotScope.Full;
// A peer that would read SnapshotBegin.tables as "all tables" and then advance its watermark
// past deltas this snapshot did not carry gets the full snapshot instead. Wider than asked
// for, but it is the only shape the older protocol can express correctly — and it is what
// 0.2.x did for every baseline.
if (!_peerSupportsPartialSnapshot)
{
_logger.LogInformation(
"Peer {PeerNodeId} does not support scoped snapshots; promoting the snapshot of {Tables} to a full one.",
PeerNodeId, string.Join(",", scoped));
return SnapshotScope.Full;
}
return new SnapshotScope(scoped);
}
private async Task<bool> HasHorizonGapAsync(Handshake peerHandshake, PeerState peerState, CancellationToken ct)
{
var minRows = await _db.QueryAsync(
"SELECT COALESCE(MIN(seq), 0) FROM __localdb_oplog", static r => r.GetInt64(0), null, ct);
var minSeq = minRows[0];
@@ -6,7 +6,7 @@ namespace ZB.MOM.WW.LocalDb.Replication.Internal;
/// <summary>
/// Shared construction of a <see cref="SyncSession"/> and its dependencies (OplogStore + LwwApplier)
/// from a <see cref="SqliteLocalDb"/> and the bound options. Both the service and client adapters use
/// this so they build an identical session; snapshot hooks are left null (Task 12 fills them in).
/// this so they build an identical session.
/// </summary>
internal static class SyncSessionFactory
{
@@ -0,0 +1,23 @@
namespace ZB.MOM.WW.LocalDb.Replication;
/// <summary>
/// Thrown by a live sync session when the oplog horizon has moved ABOVE the position that session
/// is streaming from: a cap prune deleted deltas it had not yet sent, so every later delta it could
/// ship would land on the far side of a hole the peer will never receive.
/// </summary>
/// <remarks>
/// This is not a transport failure and not a data error — it is the session discovering that its own
/// incremental position has become unsound. Nothing can repair it in-session (the deleted rows are
/// gone), so the session tears itself down; the same prune set <c>needs_snapshot</c>, which makes the
/// reconnect handshake serve a snapshot instead of deltas. Catching this and continuing would
/// reintroduce exactly the silent divergence it exists to stop.
/// </remarks>
public sealed class LocalDbOplogGapException : Exception
{
/// <summary>Creates the exception with a message describing the gap.</summary>
public LocalDbOplogGapException(string message) : base(message) { }
/// <summary>Creates the exception with a message and the underlying cause.</summary>
public LocalDbOplogGapException(string message, Exception innerException)
: base(message, innerException) { }
}
@@ -21,8 +21,26 @@ public sealed class LocalDbSyncService : LocalDbSync.LocalDbSyncBase
private readonly LocalDbMetrics? _metrics;
private readonly SyncStatus? _status;
// 0 = idle, 1 = a stream is active. Interlocked-guarded across concurrent calls (singleton service).
private int _active;
// The stream currently holding the guard, or null when idle. Guarded by _gate (singleton service).
private readonly object _gate = new();
private Registration? _active;
/// <summary>
/// How long a displacing call waits for the incumbent it evicted to actually finish before
/// giving up and refusing itself. Test seam; the default is generous because eviction only has
/// to outrun the operator, not the network.
/// </summary>
internal TimeSpan TakeoverTimeout { get; set; } = TimeSpan.FromSeconds(10);
/// <summary>One in-flight <c>Sync</c> handler: its cancellation handle plus a completion signal
/// a displacing handler can await, so two sessions never run against the database at once.</summary>
private sealed class Registration(CancellationTokenSource cts)
{
public CancellationTokenSource Cts { get; } = cts;
public TaskCompletionSource Finished { get; } =
new(TaskCreationOptions.RunContinuationsAsynchronously);
}
/// <summary>
/// Creates the passive sync service. <paramref name="status"/> is taken as the public
@@ -41,21 +59,47 @@ public sealed class LocalDbSyncService : LocalDbSync.LocalDbSyncBase
_status = status as SyncStatus;
}
/// <summary>
/// Accepts one inbound sync stream, evicting any stream already holding the guard.
/// </summary>
/// <remarks>
/// <para>
/// <b>Newest wins.</b> The guard exists to SERIALIZE sessions — a node syncs with exactly one
/// peer, and two sessions against one database would fight over the same watermarks — not to
/// fence out an impostor; authentication is the interceptor's job, and an authenticated caller
/// on this key IS the peer by definition. Refusing the newcomer therefore protected nothing and
/// cost everything in the case that actually happens: the dialling node crashes or is
/// partitioned, this handler stays parked forever inside <c>MoveNext</c> on a half-open stream
/// that will never produce another byte, and every reconnect from the rebooted peer is rejected
/// <c>AlreadyExists</c> — potentially for as long as the pair stays idle, while this side keeps
/// reporting itself connected. The newest dialler is the one with a live link, so it takes over.
/// </para>
/// <para>
/// Eviction is cancel-then-await, not cancel-and-go: the incumbent's cancellation unwinds its
/// session and reader pump, and the newcomer waits for that to complete before starting its own,
/// so the "one session at a time" property the guard was built for still holds. An incumbent
/// that cannot be evicted within <see cref="TakeoverTimeout"/> is the one case where the old
/// answer is still right — the newcomer refuses itself rather than double-running.
/// </para>
/// </remarks>
public override async Task Sync(
IAsyncStreamReader<SyncMessage> requestStream,
IServerStreamWriter<SyncMessage> responseStream,
ServerCallContext context)
{
if (Interlocked.CompareExchange(ref _active, 1, 0) != 0)
throw new RpcException(new Status(StatusCode.AlreadyExists, "a sync stream is already active"));
// The linked source exists so the reader pump is deterministically torn down on EVERY exit
// path — including a session fault for a local (non-transport) reason, where the call token
// never fires and the pump would otherwise keep reading a stream whose handler has exited.
// It is also the handle a later call cancels to displace this one.
using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken);
var mine = new Registration(linkedCts);
try
{
// The linked source exists so the reader pump is deterministically torn down on EVERY
// exit path — including a session fault for a local (non-transport) reason, where the
// call token never fires and the pump would otherwise keep reading a stream whose
// handler has exited.
using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken);
// Inside the try so the finally below always releases the guard — including when the
// takeover itself is abandoned because THIS caller gave up mid-handover.
await TakeOverAsync(mine, context.CancellationToken);
var (duplex, readerTask) = GrpcSyncDuplex.Create(requestStream, responseStream, linkedCts.Token);
var session = SyncSessionFactory.Create(_db, _options, _loggerFactory, _metrics, _status);
_status?.SessionStarted();
@@ -81,13 +125,66 @@ public sealed class LocalDbSyncService : LocalDbSync.LocalDbSyncBase
{
// Client disconnect / server shutdown: the normal end of a long-lived streaming RPC.
}
catch (OperationCanceledException)
{
// Displaced by a newer stream: an orderly handover, not a failure of this node.
_logger.LogInformation("Sync stream ended: displaced by a newer stream from the peer.");
}
catch (LocalDbSchemaMismatchException ex)
{
throw new RpcException(new Status(StatusCode.FailedPrecondition, ex.Message));
}
finally
{
Interlocked.Exchange(ref _active, 0);
lock (_gate)
{
if (ReferenceEquals(_active, mine))
_active = null;
}
mine.Finished.TrySetResult();
}
}
/// <summary>
/// Claims the single-stream guard for <paramref name="mine"/>, cancelling and awaiting whatever
/// held it. Throws <c>AlreadyExists</c> only when the incumbent refuses to end.
/// </summary>
private async Task TakeOverAsync(Registration mine, CancellationToken ct)
{
Registration? displaced;
lock (_gate)
{
displaced = _active;
_active = mine;
}
if (displaced is null)
return;
_logger.LogWarning(
"A new sync stream arrived while one was already active; displacing the incumbent (newest wins).");
try
{
// The incumbent may already be disposing its own source between its finally and here.
displaced.Cts.Cancel();
}
catch (ObjectDisposedException) { }
try
{
await displaced.Finished.Task.WaitAsync(TakeoverTimeout, ct);
}
catch (TimeoutException)
{
// Could not evict it: two live sessions over one database is worse than a refused dial,
// so fall back to the historical answer — and hand the guard back to the incumbent.
lock (_gate)
{
if (ReferenceEquals(_active, mine))
_active = displaced;
}
throw new RpcException(new Status(
StatusCode.AlreadyExists,
"a sync stream is already active and did not yield within the takeover timeout"));
}
}
}
@@ -54,7 +54,7 @@ public sealed class MaintenanceBackgroundService : BackgroundService
{
_snapshotFlagWarned = true;
_logger.LogWarning(
"Oplog backlog/age caps exceeded: pruned to the ceiling and flagged needs_snapshot — the peer must snapshot-resync. Not re-logged until a clean tick.");
"Oplog backlog/age caps exceeded: pruned to the ceiling and flagged needs_snapshot — the peer must snapshot-resync. A session streaming from below the new horizon ends itself on its next pump tick so the reconnect serves that snapshot. Not re-logged until a clean tick.");
}
else if (!snapshotFlagged)
{
@@ -64,6 +64,34 @@ public sealed class ReplicationOptions
/// <summary>Upper bound on the exponential reconnect backoff.</summary>
public TimeSpan ReconnectBackoffMax { get; set; } = TimeSpan.FromSeconds(60);
/// <summary>
/// How long the initiator's HTTP/2 connection may sit without a keepalive PING before one is
/// sent. Default 60s.
/// </summary>
/// <remarks>
/// <para>
/// A sync stream is long-lived and, on a converged pair, almost entirely silent — which is
/// exactly the shape a half-open TCP connection hides behind. Without pings, a hard crash or a
/// partition of one node leaves the other holding a stream that will never produce another byte
/// and never fail either: the initiator sits in a session it believes is healthy, and the
/// passive node's handler sits inside its single-stream guard, refusing the reconnect that would
/// heal the pair. Pings turn that into a detected failure on both ends.
/// </para>
/// <para>
/// Sent with <c>KeepAlivePingPolicy.Always</c>, so they flow on an idle connection too — an
/// active-streams-only policy would not fire in precisely the silent-pair case above. Safe over
/// h2c: a PING is an HTTP/2 frame, nothing to do with TLS.
/// </para>
/// </remarks>
public TimeSpan KeepAlivePingDelay { get; set; } = TimeSpan.FromSeconds(60);
/// <summary>
/// How long the initiator waits for a keepalive PING ack before tearing the connection down.
/// Default 20s. Detection therefore lands within roughly
/// <see cref="KeepAlivePingDelay"/> + this.
/// </summary>
public TimeSpan KeepAlivePingTimeout { get; set; } = TimeSpan.FromSeconds(20);
/// <summary>How often the maintenance service enforces the oplog caps and prunes expired tombstones.</summary>
public TimeSpan MaintenanceInterval { get; set; } = TimeSpan.FromSeconds(60);
@@ -40,6 +40,10 @@ public sealed class ReplicationOptionsValidator : IValidateOptions<ReplicationOp
failures.Add($"LocalDb:Replication:ReconnectBackoffMax must be greater than zero; got {options.ReconnectBackoffMax}.");
else if (options.ReconnectBackoffMax > ReconnectBackoffCeiling)
failures.Add($"LocalDb:Replication:ReconnectBackoffMax must be at most {ReconnectBackoffCeiling}; got {options.ReconnectBackoffMax}.");
if (options.KeepAlivePingDelay <= TimeSpan.Zero)
failures.Add($"LocalDb:Replication:KeepAlivePingDelay must be greater than zero; got {options.KeepAlivePingDelay}.");
if (options.KeepAlivePingTimeout <= TimeSpan.Zero)
failures.Add($"LocalDb:Replication:KeepAlivePingTimeout must be greater than zero; got {options.KeepAlivePingTimeout}.");
if (options.MaintenanceInterval <= TimeSpan.Zero)
failures.Add($"LocalDb:Replication:MaintenanceInterval must be greater than zero; got {options.MaintenanceInterval}.");
if (options.MaxHlcDriftAhead <= TimeSpan.Zero)
@@ -1,3 +1,4 @@
using System.Net.Http;
using Grpc.Core;
using Grpc.Net.Client;
using Microsoft.Extensions.Hosting;
@@ -76,7 +77,7 @@ public sealed class SyncBackgroundService : BackgroundService
{
var channel = ChannelFactory is { } factory
? factory()
: ownedChannel = GrpcChannel.ForAddress(_options.PeerAddress);
: ownedChannel = DialPeer();
var client = new LocalDbSync.LocalDbSyncClient(channel);
var headers = new Metadata();
@@ -152,6 +153,39 @@ public sealed class SyncBackgroundService : BackgroundService
}
}
/// <summary>
/// Dials the peer over a channel that sends HTTP/2 keepalive PINGs, so a half-open link is
/// detected instead of being waited on forever.
/// </summary>
/// <remarks>
/// <para>
/// The bare <c>GrpcChannel.ForAddress(address)</c> this replaces had NO keepalive of any kind,
/// and neither end of a sync stream ever times out on silence by design (a converged pair is
/// legitimately quiet). A hard crash or partition of the peer therefore left this node inside a
/// session that could never make progress and could never fail — and its own half of that stream
/// pinned on the peer, where the single-stream guard turns it into a refusal of the very
/// reconnect that heals the pair.
/// </para>
/// <para>
/// <c>DisposeHttpClient</c> hands the handler's lifetime to the channel, which this loop already
/// disposes on every exit path.
/// </para>
/// </remarks>
private GrpcChannel DialPeer() =>
GrpcChannel.ForAddress(_options.PeerAddress, new GrpcChannelOptions
{
HttpHandler = new SocketsHttpHandler
{
KeepAlivePingDelay = _options.KeepAlivePingDelay,
KeepAlivePingTimeout = _options.KeepAlivePingTimeout,
// Always, not WithActiveRequests: the case being closed is a SILENT stream on a dead
// link, and an active-requests policy stops pinging once the peer stops responding.
KeepAlivePingPolicy = HttpKeepAlivePingPolicy.Always,
EnableMultipleHttp2Connections = true,
},
DisposeHttpClient = true,
});
private static bool IsSchemaMismatch(Exception? ex) =>
ex is LocalDbSchemaMismatchException
|| (ex is RpcException rpc && rpc.StatusCode == StatusCode.FailedPrecondition);
@@ -38,8 +38,18 @@ public interface ILocalDb
/// schema digest, and installs the three AFTER triggers that capture every change into the
/// oplog + row-version tables within the same transaction as the consumer's write. Idempotent
/// (re-registering regenerates the triggers). Throws <see cref="LocalDbRegistrationException"/>
/// if the name is unsafe, the table is missing, or it has no primary key.
/// if the name is unsafe, the table is missing, it has no primary key, it has a BLOB column, or
/// another table references it with a row-mutating <c>ON DELETE</c> action.
/// </summary>
/// <remarks>
/// <b>Replicated tables must not be cascade-delete foreign-key parents.</b> Applying a remote row
/// uses <c>INSERT OR REPLACE</c>, whose implicit DELETE fires <c>ON DELETE CASCADE</c> /
/// <c>SET NULL</c> / <c>SET DEFAULT</c> on referencing tables while change capture is suppressed
/// — the child rows it touches are never captured, so the peer keeps them and the pair diverges
/// silently. Such a table is refused here, because registration is the only point at which the
/// problem is still visible. <c>NO ACTION</c> / <c>RESTRICT</c> are allowed: they mutate no child
/// row, so the implicit delete simply fails and the applier dead-letters the entry.
/// </remarks>
ReplicatedTable RegisterReplicated(string tableName);
/// <summary>
@@ -50,7 +60,8 @@ public interface ILocalDb
/// <param name="baselineExistingRows">
/// When true, every row already in the table that has no <c>__localdb_row_version</c> entry is
/// given one at the <b>baseline floor</b> — HLC <c>0</c>, stamped with this node's id — and, if
/// anything was seeded, a snapshot resync is flagged so the peer actually receives them.
/// anything was seeded, a snapshot resync is flagged <b>for this table</b> so the peer actually
/// receives them.
/// </param>
/// <remarks>
/// <para>
@@ -80,16 +91,29 @@ public interface ILocalDb
/// and no snapshot is flagged. Passing true on every boot is therefore free after the first.
/// </para>
/// <para>
/// <b>Cost, and why it is worth planning for.</b> The snapshot flag is a single per-DATABASE
/// flag, and a snapshot streams EVERY registered table in full — so baselining one table
/// re-sends all of them, and a pair that enables replication late has both nodes doing that to
/// each other at once. On a database holding a high-volume table (an event log at its retention
/// cap, say) that is the largest transfer this library ever performs, in both directions
/// simultaneously. It is bounded work, not runaway: outbound messages respect
/// <c>MaxBatchBytes</c>, and the receive side is bounded by the transport's inbox. But the
/// operational advice is to <b>purge or trim high-volume tables before enabling replication
/// late</b>, and to expect one slow first sync rather than a slow steady state. A per-table
/// snapshot flag would narrow this; it needs an on-disk schema change and is not in 0.2.x.
/// <b>Cost, and what bounds it (0.3.0).</b> The snapshot flag is <b>per-table</b>
/// (<c>__localdb_snapshot_state</c>, on-disk schema v3), so baselining one table asks the peer
/// for that table alone — the tables around it are not re-streamed. Through 0.2.x the flag was a
/// single per-database bit and baselining anything re-sent everything, in both directions at
/// once on a pair enabling replication late.
/// </para>
/// <para>
/// Two cases still stream every registered table, by design. A <b>pre-0.3.0 peer</b> cannot be
/// told a snapshot is partial (it would advance its watermark past deltas it never received),
/// so a scoped snapshot is promoted to a full one for it. And the <b>oplog-horizon</b> paths —
/// a cap prune, an impossible peer ack — invalidate delta continuity for every table at once
/// and still set the database-wide flag. So the operational advice survives, narrowed: baseline
/// a high-volume table (an event log at its retention cap, say) deliberately, and <b>purge or
/// trim it first</b>, because that ONE table's rows still travel in full. The work is bounded
/// either way — outbound messages respect <c>MaxBatchBytes</c>, the receive side is bounded by
/// the transport's inbox.
/// </para>
/// <para>
/// <b>Crash safety.</b> Between 0.3.0 peers a table's flag is cleared only when the receiver
/// acknowledges the completed snapshot, so an ungraceful exit part-way through re-streams the
/// same scope on reconnect rather than losing it. Against a pre-0.3.0 peer (which sends no such
/// acknowledgement) the 0.2.x behaviour is kept: the flag clears once the snapshot is on the
/// wire, and the handshake's horizon gap check is the net.
/// </para>
/// </remarks>
ReplicatedTable RegisterReplicated(string tableName, bool baselineExistingRows);
@@ -27,9 +27,20 @@ internal static class LocalDbSchema
/// <c>hlc</c>-predicate delete. It cost a b-tree insert on the CAPTURE path — i.e. inside every
/// consumer write transaction — to serve nothing.
/// </description></item>
/// <item><description>
/// <b>3</b> — adds <c>__localdb_snapshot_state</c>, the PER-TABLE snapshot-resync flag. Until
/// v3 the only flag was <c>__localdb_peer_state.needs_snapshot</c>, one bit for the whole
/// database, so baselining a single table re-streamed EVERY registered table in full — in both
/// directions at once on a pair enabling replication late. The database-wide bit survives with
/// its original meaning (the oplog-horizon and cap-prune paths genuinely invalidate every
/// table); the new table narrows the baseline case to the tables that actually need it. The
/// transition itself is a version stamp only — <see cref="Ddl"/> creates the table with
/// <c>IF NOT EXISTS</c> on every open, so an older file gains it whatever its stamp says, and
/// an empty flag table means exactly what it meant before: nothing per-table is owed.
/// </description></item>
/// </list>
/// </remarks>
internal const long CurrentSchemaVersion = 2;
internal const long CurrentSchemaVersion = 3;
private const string Ddl = """
CREATE TABLE IF NOT EXISTS __localdb_meta (
@@ -73,6 +84,14 @@ internal static class LocalDbSchema
id INTEGER PRIMARY KEY CHECK (id = 1),
applying INTEGER NOT NULL DEFAULT 0
);
-- Per-table snapshot-resync flags (schema v3). A row exists only while that table owes the
-- peer a snapshot; the whole-database __localdb_peer_state.needs_snapshot bit still means
-- "every registered table" and is what the cap/horizon paths set.
-- WITHOUT ROWID: table_name IS the only access path (read as a set, cleared by name).
CREATE TABLE IF NOT EXISTS __localdb_snapshot_state (
table_name TEXT NOT NULL PRIMARY KEY,
needs_snapshot INTEGER NOT NULL DEFAULT 0
) WITHOUT ROWID;
CREATE TABLE IF NOT EXISTS __localdb_dead_letter (
id INTEGER PRIMARY KEY AUTOINCREMENT,
received_utc TEXT NOT NULL,
@@ -143,6 +162,12 @@ internal static class LocalDbSchema
if (current < 2)
steps.Add("DROP INDEX IF EXISTS __localdb_oplog_hlc;");
// 2 -> 3: __localdb_snapshot_state. No step: the shared Ddl above already ran
// CREATE TABLE IF NOT EXISTS for it in this same transaction, and an empty flag table is
// the correct v3 state for a file that has never baselined a table. Repeating the DDL here
// would be a second source of truth for the same layout. The stamp still moves, because
// schema_version records what the file's layout IS, not which statements produced it.
if (steps.Count > 0)
{
using var ddl = connection.CreateCommand();
@@ -199,6 +199,8 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
throw new LocalDbRegistrationException(
$"Cannot register table '{tableName}': replication requires an explicit primary key.");
EnsureNoRowMutatingDeleteAction(tableName);
var allColumns = columns.Select(c => (c.Name, c.Type)).ToList();
var digest = ReplicatedTable.ComputeDigest(tableName, pkColumns, allColumns);
var table = new ReplicatedTable(tableName, pkColumns, allColumns, digest);
@@ -276,7 +278,12 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
prune.Transaction = tx;
prune.CommandText =
"DELETE FROM __localdb_oplog WHERE table_name = $t;" +
"DELETE FROM __localdb_row_version WHERE table_name = $t;";
"DELETE FROM __localdb_row_version WHERE table_name = $t;" +
// A pending per-table snapshot flag for a table nothing streams any more would
// never clear — the streamer skips unregistered tables, so the flag would sit
// in the file forever, and re-registering later would find a stale "already
// owed" it did not ask for. Dropped with the ledger it refers to.
"DELETE FROM __localdb_snapshot_state WHERE table_name = $t;";
prune.Parameters.AddWithValue("$t", tableName);
prunedRows = prune.ExecuteNonQuery();
}
@@ -288,6 +295,87 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
}
}
/// <summary>
/// Refuses to register <paramref name="tableName"/> when another table references it with an
/// <c>ON DELETE</c> action that MUTATES the referencing rows (<c>CASCADE</c>, <c>SET NULL</c>,
/// <c>SET DEFAULT</c>).
/// </summary>
/// <remarks>
/// <para>
/// <b>Why this is a replication correctness rule and not a schema opinion.</b> Applying a peer's
/// row runs <c>INSERT OR REPLACE</c>, and REPLACE resolves a primary-key conflict by DELETING the
/// existing row first. Connections run with <c>PRAGMA foreign_keys=ON</c>, so that implicit delete
/// fires the referential action on every child table — while the apply path holds the capture
/// guard raised, precisely so its own writes are not re-captured. The child rows are therefore
/// deleted (or nulled) locally and produce NO oplog entry: the peer never hears about them, keeps
/// them, and the pair is durably divergent with nothing anywhere reporting a fault.
/// </para>
/// <para>
/// Refusing at registration is the only place the problem is visible. At apply time the damage is
/// already done and indistinguishable from a normal write, and no runtime signal exists to detect
/// it. <b>NO ACTION / RESTRICT are deliberately allowed</b>: they mutate no child row — the
/// implicit delete simply fails, which the applier surfaces as a dead-lettered entry. Loud, not
/// silent, so there is nothing to guard.
/// </para>
/// <para>
/// The check is registration-time, so a cascading child table CREATED LATER is not caught (SQLite
/// has no DDL hook to catch it on). Registration normally runs at startup after schema setup,
/// which is where the constraint belongs.
/// </para>
/// </remarks>
private void EnsureNoRowMutatingDeleteAction(string tableName)
{
var candidates = new List<string>();
using (var cmd = _master.CreateCommand())
{
// Every user table is a candidate CHILD; the internal sqlite_* tables carry no
// foreign keys. The registered table itself is included: a self-referencing cascade
// is the same hazard.
cmd.CommandText = "SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'";
using var reader = cmd.ExecuteReader();
while (reader.Read())
candidates.Add(reader.GetString(0));
}
foreach (var child in candidates)
{
using var cmd = _master.CreateCommand();
cmd.CommandText = $"PRAGMA foreign_key_list({TriggerSqlGenerator.Ident(child)})";
using var reader = cmd.ExecuteReader();
while (reader.Read())
{
// foreign_key_list cols: 0=id, 1=seq, 2=table (the PARENT), 3=from, 4=to,
// 5=on_update, 6=on_delete, 7=match.
var parent = reader.IsDBNull(2) ? "" : reader.GetString(2);
// SQLite identifiers are case-insensitive, and the REFERENCES clause need not
// match the CREATE TABLE spelling.
if (!string.Equals(parent, tableName, StringComparison.OrdinalIgnoreCase))
continue;
var onDelete = reader.IsDBNull(6) ? "" : reader.GetString(6);
if (!RowMutatingDeleteActions.Contains(onDelete))
continue;
throw new LocalDbRegistrationException(
$"Cannot register table '{tableName}': table '{child}' has a foreign key referencing it " +
$"with ON DELETE {onDelete}. Replicated tables must not be cascade-delete foreign-key " +
"parents: applying a remote row uses INSERT OR REPLACE, whose implicit DELETE fires the " +
"referential action while change capture is suppressed — the child rows it touches are " +
"never captured, so the peer never learns of them and the two nodes diverge silently. " +
$"Drop the referential action on '{child}' (ON DELETE NO ACTION, removing children " +
"explicitly instead), or do not replicate this table.");
}
}
}
/// <summary>
/// The <c>ON DELETE</c> actions that mutate referencing rows, as SQLite spells them in
/// <c>PRAGMA foreign_key_list</c>. NO ACTION / RESTRICT are absent on purpose — see
/// <see cref="EnsureNoRowMutatingDeleteAction"/>.
/// </summary>
private static readonly HashSet<string> RowMutatingDeleteActions =
new(["CASCADE", "SET NULL", "SET DEFAULT"], StringComparer.OrdinalIgnoreCase);
/// <summary>
/// Seeds <c>__localdb_row_version</c> at the baseline floor for every row of
/// <paramref name="table"/> that has no ledger entry, and flags a snapshot resync if it seeded
@@ -308,12 +396,12 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
/// the same snapshot anyway, via a cap-prune.
/// </para>
/// <para>
/// <b>The flag is per-DATABASE, not per-table</b>, and a snapshot streams every registered table
/// in full — so baselining one table re-sends all of them, and both nodes of a late-enabled pair
/// do it to each other at once. Bounded (batch budget outbound, bounded inbox inbound) but
/// potentially the largest transfer this library performs; see
/// <see cref="ILocalDb.RegisterReplicated(string, bool)"/> for the operational advice. Narrowing
/// it to a per-table flag needs an on-disk schema change and is deferred.
/// <b>The flag is PER-TABLE</b> (<c>__localdb_snapshot_state</c>, schema v3): baselining one
/// table asks for a snapshot of that table alone, not of every registered table. The
/// database-wide <c>__localdb_peer_state.needs_snapshot</c> bit is untouched here and keeps its
/// original meaning — the oplog-horizon and cap-prune paths invalidate every table's delta
/// continuity at once, and only those paths set it. A peer too old to understand a scoped
/// snapshot is served a full one instead; the negotiation lives in the replication package.
/// </para>
/// </remarks>
private void BaselineExistingRows(SqliteTransaction tx, ReplicatedTable table)
@@ -339,7 +427,10 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
using var flag = _master.CreateCommand();
flag.Transaction = tx;
flag.CommandText = "UPDATE __localdb_peer_state SET needs_snapshot = 1 WHERE id = 1";
flag.CommandText =
"INSERT INTO __localdb_snapshot_state (table_name, needs_snapshot) VALUES ($t, 1) " +
"ON CONFLICT(table_name) DO UPDATE SET needs_snapshot = 1";
flag.Parameters.AddWithValue("$t", table.Name);
flag.ExecuteNonQuery();
}
@@ -70,6 +70,20 @@ public sealed class BatchBudgetTests : IDisposable
return msg!;
}
// Reads the next scripted-peer message while also observing the session task: a session that
// faults would otherwise leave this wait to run out the whole test timeout, reporting an
// opaque cancellation instead of the session's real exception.
private static async Task<SyncMessage> NextAsync(Task run, ChannelReader<SyncMessage> reader, CancellationToken ct)
{
var next = NextAsync(reader, ct);
if (ReferenceEquals(await Task.WhenAny(run, next), run))
{
await run;
throw new InvalidOperationException("The sync session ended before the expected message arrived.");
}
return await next;
}
private static async Task SwallowAsync(Task task)
{
try { await task; }
@@ -140,7 +154,7 @@ public sealed class BatchBudgetTests : IDisposable
var seen = new List<long>();
while (seen.Count < 20)
{
var msg = await NextAsync(fromSession, cts.Token);
var msg = await NextAsync(run, fromSession, cts.Token);
if (msg.MsgCase != SyncMessage.MsgOneofCase.DeltaBatch)
continue;
Assert.True(BatchBytes(msg.DeltaBatch) <= budget,
@@ -180,7 +194,7 @@ public sealed class BatchBudgetTests : IDisposable
var batches = new List<DeltaBatch>();
while (batches.Sum(b => b.Entries.Count) < 2)
{
var msg = await NextAsync(fromSession, cts.Token);
var msg = await NextAsync(run, fromSession, cts.Token);
if (msg.MsgCase == SyncMessage.MsgOneofCase.DeltaBatch)
batches.Add(msg.DeltaBatch);
}
@@ -220,7 +234,7 @@ public sealed class BatchBudgetTests : IDisposable
var seen = 0;
while (seen < 12)
{
var msg = await NextAsync(fromSession, cts.Token);
var msg = await NextAsync(run, fromSession, cts.Token);
if (msg.MsgCase != SyncMessage.MsgOneofCase.DeltaBatch)
continue;
Assert.True(msg.DeltaBatch.Entries.Count <= 5);
@@ -250,7 +264,7 @@ public sealed class BatchBudgetTests : IDisposable
var streamer = new SnapshotStreamer(db, store, options, NullLogger.Instance);
var sent = new List<SyncMessage>();
await streamer.SendAsync((m, _) => { sent.Add(m); return Task.CompletedTask; }, default);
await streamer.SendAsync(SnapshotScope.Full, (m, _) => { sent.Add(m); return Task.CompletedTask; }, default);
var batches = sent.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
.Select(m => m.SnapshotBatch).ToList();
@@ -279,7 +293,7 @@ public sealed class BatchBudgetTests : IDisposable
var streamer = new SnapshotStreamer(db, store, options, NullLogger.Instance);
var sent = new List<SyncMessage>();
await streamer.SendAsync((m, _) => { sent.Add(m); return Task.CompletedTask; }, default);
await streamer.SendAsync(SnapshotScope.Full, (m, _) => { sent.Add(m); return Task.CompletedTask; }, default);
var batches = sent.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
.Select(m => m.SnapshotBatch).ToList();
@@ -312,7 +326,7 @@ public sealed class BatchBudgetTests : IDisposable
}
SyncMessage ack;
do { ack = await NextAsync(fromSession, cts.Token); }
do { ack = await NextAsync(run, fromSession, cts.Token); }
while (ack.MsgCase != SyncMessage.MsgOneofCase.DeltaAck);
// The discriminator: uncoalesced, the FIRST ack could only cover seq 1.
@@ -348,7 +362,7 @@ public sealed class BatchBudgetTests : IDisposable
new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 77 } }, cts.Token);
SyncMessage ack;
do { ack = await NextAsync(fromSession, cts.Token); }
do { ack = await NextAsync(run, fromSession, cts.Token); }
while (ack.MsgCase != SyncMessage.MsgOneofCase.DeltaAck);
Assert.Equal(1, ack.DeltaAck.AppliedThruSeq);
@@ -70,8 +70,12 @@ public sealed class GrpcAdapterTests : IAsyncLifetime
}
[Fact]
public async Task SecondConcurrentStream_Rejected()
public async Task SecondConcurrentStream_DisplacesTheIncumbent()
{
// Newest wins (0.3.2). The guard serializes sessions; it was never a fence — the auth
// interceptor is, and an authenticated same-key dialler IS the peer. Refusing the newcomer
// bought nothing and cost the pair its only route back from a half-open link, where the
// incumbent handler can hold the guard indefinitely on a dead stream.
var serverHost = await BuildServerAsync(NewDbPath(), apiKey: null, sink: null);
var serverDb = serverHost.Services.GetRequiredService<ILocalDb>();
var handler = serverHost.GetTestServer().CreateHandler();
@@ -85,11 +89,15 @@ public sealed class GrpcAdapterTests : IAsyncLifetime
// Convergence proves the first stream's server handler is active and holding the single-stream guard.
await WaitForAsync(async () => (await ReadOrders(serverDb)).Count == 1, RunTimeout);
var attemptsBefore = bg.ConnectionAttempts;
var client = new LocalDbSync.LocalDbSyncClient(ChannelOver(handler));
using var call2 = client.Sync(cancellationToken: cts.Token);
var ex = await Assert.ThrowsAsync<RpcException>(() => call2.ResponseStream.MoveNext(cts.Token));
Assert.Equal(StatusCode.AlreadyExists, ex.StatusCode);
Assert.True(await call2.ResponseStream.MoveNext(cts.Token));
Assert.Equal(SyncMessage.MsgOneofCase.Handshake, call2.ResponseStream.Current.MsgCase);
// The incumbent really ended: the displaced client notices and dials again.
await WaitForAsync(() => Task.FromResult(bg.ConnectionAttempts > attemptsBefore), RunTimeout);
await bg.StopAsync(CancellationToken.None);
}
@@ -262,9 +270,13 @@ public sealed class GrpcAdapterTests : IAsyncLifetime
TombstoneRetention = TimeSpan.Zero,
ReconnectBackoffMax = TimeSpan.Zero,
MaxHlcDriftAhead = TimeSpan.Zero,
KeepAlivePingDelay = TimeSpan.Zero,
KeepAlivePingTimeout = TimeSpan.Zero,
});
Assert.True(result.Failed);
Assert.Contains(result.Failures, f => f.Contains("KeepAlivePingDelay", StringComparison.Ordinal));
Assert.Contains(result.Failures, f => f.Contains("KeepAlivePingTimeout", StringComparison.Ordinal));
Assert.Contains(result.Failures, f => f.Contains("FlushInterval", StringComparison.Ordinal));
Assert.Contains(result.Failures, f => f.Contains("MaxBatchSize", StringComparison.Ordinal));
Assert.Contains(result.Failures, f => f.Contains("MaxOplogRows", StringComparison.Ordinal));
@@ -51,7 +51,8 @@ public sealed class LocalDbSchemaTests : IDisposable
foreach (var expected in new[]
{
"__localdb_meta", "__localdb_oplog", "__localdb_peer_state",
"__localdb_row_version", "__localdb_applying", "__localdb_dead_letter"
"__localdb_row_version", "__localdb_applying", "__localdb_dead_letter",
"__localdb_snapshot_state",
})
Assert.Contains(expected, tables);
@@ -85,6 +86,44 @@ public sealed class LocalDbSchemaTests : IDisposable
ScalarLong(conn, "SELECT schema_version FROM __localdb_meta WHERE id=1"));
}
[Fact]
public void EnsureCreated_UpgradesV2File_AddsPerTableSnapshotState()
{
using var conn = Open();
// A file exactly as schema v2 left it: everything except __localdb_snapshot_state, stamped 2.
LocalDbSchema.EnsureCreated(conn);
Exec(conn, "DROP TABLE __localdb_snapshot_state;" +
"UPDATE __localdb_meta SET schema_version = 2 WHERE id = 1;");
LocalDbSchema.EnsureCreated(conn);
Assert.Equal(1, ScalarLong(conn,
"SELECT count(*) FROM sqlite_master WHERE type='table' AND name='__localdb_snapshot_state'"));
// Empty is the right v3 state for a file that never baselined anything: no table owes a
// snapshot, and the database-wide flag it was carrying keeps whatever value it had.
Assert.Equal(0, ScalarLong(conn, "SELECT COUNT(*) FROM __localdb_snapshot_state"));
Assert.Equal(LocalDbSchema.CurrentSchemaVersion,
ScalarLong(conn, "SELECT schema_version FROM __localdb_meta WHERE id=1"));
}
[Fact]
public void EnsureCreated_Upgrade_PreservesTheDatabaseWideSnapshotFlag()
{
// A v2 file mid-resync (cap prune flagged a full snapshot, the peer was down) must still owe
// that snapshot after the upgrade. v3 narrows where NEW debt is recorded; it does not
// reinterpret debt already on disk.
using var conn = Open();
LocalDbSchema.EnsureCreated(conn);
Exec(conn, "DROP TABLE __localdb_snapshot_state;" +
"UPDATE __localdb_meta SET schema_version = 2 WHERE id = 1;" +
"UPDATE __localdb_peer_state SET needs_snapshot = 1 WHERE id = 1;");
LocalDbSchema.EnsureCreated(conn);
Assert.Equal(1, ScalarLong(conn, "SELECT needs_snapshot FROM __localdb_peer_state WHERE id=1"));
}
[Fact]
public void EnsureCreated_Upgrade_PreservesNodeIdentityAndOplogRows()
{
@@ -427,9 +427,12 @@ public sealed class LwwApplierTests : IDisposable
var result = await applier.ApplyBatchAsync(entries, default);
Assert.Equal(202, result.Applied);
// Four table-independent statements (applying guard, version lookup, row-version write,
// peer state), plus an upsert per table touched and a delete for the one table tombstoned.
Assert.Equal(7, applier.LastBatchStatementCount);
// Six table-independent statements (applying guard, version lookup, row-version write,
// peer state, plus the per-entry savepoint's SAVEPOINT and RELEASE), an upsert per table
// touched, and a delete for the one table tombstoned. The savepoint's ROLLBACK TO and the
// dead-letter insert are absent because this batch has no poison entry — statements are
// built on first use, so the count tracks the statement KINDS the batch actually ran.
Assert.Equal(9, applier.LastBatchStatementCount);
// ...and the batch is still correct, statement reuse notwithstanding.
Assert.Null(await ReadOrder(db, 1));
@@ -462,6 +465,72 @@ public sealed class LwwApplierTests : IDisposable
Assert.Equal(("LAST", (long?)5), await ReadOrder(db, 5));
}
[Fact]
public async Task Apply_LedgerWriteFails_LeavesNeitherTheDeleteNorTheLedgerEntry()
{
// The hazard (P4-7): applying a tombstone is TWO statements — delete the row, then write the
// ledger's tombstone version — and the poison catch commits whatever the first one already
// did. A failure landing between them therefore committed a DELETED ROW carrying a STALE LIVE
// ledger entry, which is exactly the shape SnapshotStreamer warns about and skips: the key
// then never travels again and the pair is divergent for that row, permanently and silently.
//
// The failure is injected with a trigger that aborts the ledger write (in production the
// realistic causes are disk-full / IO error / corruption — none of them reproducible here),
// and RAISE(ABORT) aborts only the statement, so the applier's narrow SqliteException catch
// treats it as per-entry poison exactly as it would a real one.
using var db = await NewOrdersDb();
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)");
var localHlc = await RowVersionHlc(db, Pk1);
await db.ExecuteAsync(
"CREATE TRIGGER poison_ledger BEFORE INSERT ON __localdb_row_version " +
"WHEN NEW.is_tombstone = 1 BEGIN SELECT RAISE(ABORT, 'injected ledger failure'); END");
var result = await new LwwApplier(db).ApplyBatchAsync(
[Entry(90, Pk1, null, localHlc + 1000, "remote-node", tombstone: true)], default);
// The entry is poison, so it is dead-lettered and the batch continues — unchanged semantics.
Assert.Equal(1, result.DeadLettered);
Assert.Equal(0, result.Applied);
// BOTH or NEITHER. Neither, here: the row must still be present, because its ledger entry
// still says "live at the local version".
Assert.Equal(("LOCAL", (long?)5), await ReadOrder(db, 1));
var rv = await db.QueryAsync(
"SELECT hlc, is_tombstone FROM __localdb_row_version WHERE table_name = 'orders' AND pk_json = @pk",
x => (Hlc: x.GetInt64(0), Tomb: x.GetInt64(1)), new { pk = Pk1 });
Assert.Single(rv);
Assert.Equal(localHlc, rv[0].Hlc);
Assert.Equal(0, rv[0].Tomb);
}
[Fact]
public async Task Apply_LedgerWriteFails_MidBatch_LeavesTheOtherEntriesApplied()
{
// The atomicity must be per-ENTRY, not per-batch: rolling back to the savepoint may not
// discard the rows applied before the poison one, nor block the ones after it.
using var db = await NewOrdersDb();
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'DOOMED', 2)");
var doomedHlc = await RowVersionHlc(db, "{\"id\":2}");
await db.ExecuteAsync(
"CREATE TRIGGER poison_ledger BEFORE INSERT ON __localdb_row_version " +
"WHEN NEW.is_tombstone = 1 BEGIN SELECT RAISE(ABORT, 'injected ledger failure'); END");
var result = await new LwwApplier(db).ApplyBatchAsync(
[
Entry(91, Pk1, "{\"id\":1,\"sku\":\"BEFORE\",\"qty\":1}", doomedHlc + 1, "remote-node"),
Entry(92, "{\"id\":2}", null, doomedHlc + 1000, "remote-node", tombstone: true),
Entry(93, "{\"id\":3}", "{\"id\":3,\"sku\":\"AFTER\",\"qty\":3}", doomedHlc + 3, "remote-node"),
], default);
Assert.Equal(2, result.Applied);
Assert.Equal(1, result.DeadLettered);
Assert.Equal(("BEFORE", (long?)1), await ReadOrder(db, 1));
Assert.Equal(("DOOMED", (long?)2), await ReadOrder(db, 2));
Assert.Equal(("AFTER", (long?)3), await ReadOrder(db, 3));
}
[Fact]
public async Task Apply_ReplaceExistingRow_DoesNotFireCaptureTriggers()
{
@@ -70,6 +70,97 @@ public sealed class RegistrationTests : IDisposable
Assert.Contains("BLOB", ex.Message);
}
// ---- Cascading-delete foreign-key parents (P4-7) --------------------------------------
[Fact]
public async Task Register_CascadeDeleteFkParent_Throws()
{
// The hazard: applying a remote row uses INSERT OR REPLACE, whose implicit DELETE fires
// ON DELETE CASCADE while the capture guard is RAISED — so the cascaded child rows are
// deleted locally and never captured, and the peer keeps them forever.
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync(
"CREATE TABLE order_lines (id INTEGER PRIMARY KEY, order_id INTEGER " +
"REFERENCES orders(id) ON DELETE CASCADE)");
var ex = Assert.Throws<LocalDbRegistrationException>(() => db.RegisterReplicated("orders"));
Assert.Contains("orders", ex.Message);
Assert.Contains("order_lines", ex.Message);
Assert.Contains("ON DELETE CASCADE", ex.Message);
// Refused means refused: no triggers installed, nothing in the registry.
Assert.Empty(await Triggers(db));
Assert.False(db.ReplicatedTables.ContainsKey("orders"));
}
[Fact]
public async Task Register_CascadeDeleteFkChild_IsAllowed()
{
// Only the PARENT is refused. The child side is safe: replacing a child row cascades
// nothing, and the child's own deletes are captured normally.
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync(
"CREATE TABLE order_lines (id INTEGER PRIMARY KEY, order_id INTEGER " +
"REFERENCES orders(id) ON DELETE CASCADE)");
db.RegisterReplicated("order_lines");
Assert.True(db.ReplicatedTables.ContainsKey("order_lines"));
}
[Theory]
[InlineData("ON DELETE SET NULL", "ON DELETE SET NULL")]
[InlineData("ON DELETE SET DEFAULT", "ON DELETE SET DEFAULT")]
public async Task Register_OtherRowMutatingDeleteActions_AlsoThrow(string clause, string expected)
{
// SET NULL / SET DEFAULT mutate child rows on the same implicit DELETE and are captured
// just as little — same divergence, so the same refusal.
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync(
"CREATE TABLE order_lines (id INTEGER PRIMARY KEY, order_id INTEGER DEFAULT 0 " +
$"REFERENCES orders(id) {clause})");
var ex = Assert.Throws<LocalDbRegistrationException>(() => db.RegisterReplicated("orders"));
Assert.Contains(expected, ex.Message);
}
[Fact]
public async Task Register_NonCascadingFkParent_IsAllowed()
{
// The default (NO ACTION / RESTRICT) mutates no child row: the implicit DELETE simply
// fails the statement, which the applier already surfaces as a dead-lettered entry. Noisy,
// not silent — so there is nothing to refuse.
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync(
"CREATE TABLE order_lines (id INTEGER PRIMARY KEY, order_id INTEGER REFERENCES orders(id))");
await db.ExecuteAsync("CREATE TABLE restricted (id INTEGER PRIMARY KEY, " +
"order_id INTEGER REFERENCES orders(id) ON DELETE RESTRICT)");
db.RegisterReplicated("orders");
Assert.Equal(3, (await Triggers(db)).Count);
}
[Fact]
public async Task Register_UnrelatedTablesWithCascades_DoNotBlockRegistration()
{
// The scan must key on the table being registered, not on "the database contains a cascade".
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync("CREATE TABLE widgets (id INTEGER PRIMARY KEY, name TEXT)");
await db.ExecuteAsync(
"CREATE TABLE widget_parts (id INTEGER PRIMARY KEY, widget_id INTEGER " +
"REFERENCES widgets(id) ON DELETE CASCADE)");
db.RegisterReplicated("orders");
Assert.True(db.ReplicatedTables.ContainsKey("orders"));
}
[Fact]
public async Task Register_IsIdempotent_RegeneratesTriggers()
{
@@ -291,6 +382,45 @@ public sealed class RegistrationTests : IDisposable
// Without the flag the seeding would be inert: no gap exists at the handshake, so the peer
// would be told it needs no snapshot and the baselined rows would never travel.
Assert.True(await NeedsSnapshot(db));
// ...and it is flagged for THIS TABLE ONLY. Through 0.2.x this set the one database-wide
// bit, so baselining `orders` re-streamed every other registered table with it.
Assert.False(await NeedsFullSnapshot(db));
}
[Fact]
public async Task Register_WithBaseline_FlagsOnlyTheBaselinedTable()
{
// The per-table flag's reason to exist: opting one table into replication late must not
// schedule a full re-stream of its unrelated neighbours.
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync("CREATE TABLE widgets (id INTEGER PRIMARY KEY, name TEXT)");
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
await db.ExecuteAsync("INSERT INTO widgets (id, name) VALUES (1,'W')");
db.RegisterReplicated("widgets");
db.RegisterReplicated("orders", baselineExistingRows: true);
Assert.True(await NeedsSnapshot(db, "orders"));
Assert.False(await NeedsSnapshot(db, "widgets"));
Assert.False(await NeedsFullSnapshot(db));
}
[Fact]
public async Task Deregister_DropsThePendingSnapshotFlag()
{
// A flag naming a table nothing streams any more can never clear itself — the streamer skips
// unregistered tables — so it would sit in the file and re-appear as unasked-for debt if the
// table were registered again.
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
db.RegisterReplicated("orders", baselineExistingRows: true);
Assert.True(await NeedsSnapshot(db));
db.DeregisterReplicated("orders");
Assert.False(await NeedsSnapshot(db));
}
[Fact]
@@ -301,7 +431,8 @@ public sealed class RegistrationTests : IDisposable
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
db.RegisterReplicated("orders", baselineExistingRows: true);
await db.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 0 WHERE id = 1");
// Stand in for the snapshot having been delivered and acknowledged.
await db.ExecuteAsync("DELETE FROM __localdb_snapshot_state");
db.RegisterReplicated("orders", baselineExistingRows: true);
@@ -480,7 +611,17 @@ public sealed class RegistrationTests : IDisposable
private static async Task<long> CountAsync(SqliteLocalDb db, string sql) =>
(await db.QueryAsync(sql, r => r.GetInt64(0)))[0];
private static async Task<bool> NeedsSnapshot(SqliteLocalDb db) =>
/// <summary>Whether <paramref name="table"/> individually owes the peer a snapshot (schema v3).</summary>
private static async Task<bool> NeedsSnapshot(SqliteLocalDb db, string table = "orders") =>
(await db.QueryAsync(
"SELECT COUNT(*) FROM __localdb_snapshot_state WHERE table_name = @t AND needs_snapshot <> 0",
r => r.GetInt64(0), new { t = table }))[0] != 0;
/// <summary>
/// The DATABASE-WIDE flag — "every registered table". Baselining must never set it: that is the
/// whole point of the per-table flags, and setting it would restore the 0.2.x behaviour silently.
/// </summary>
private static async Task<bool> NeedsFullSnapshot(SqliteLocalDb db) =>
(await db.QueryAsync(
"SELECT needs_snapshot FROM __localdb_peer_state WHERE id = 1", r => r.GetInt64(0)))[0] != 0;
}
@@ -89,9 +89,19 @@ public sealed class SnapshotResyncTests : IDisposable
return (duplex, toSession.Writer, fromSession.Reader);
}
private static Handshake MatchingHandshake(SqliteLocalDb sessionDb, string peerNodeId, long lastAppliedRemoteSeq = 0)
// supportsPartialSnapshot defaults to FALSE, which is exactly what a 0.1.x/0.2.x peer puts on
// the wire (the field does not exist in its proto, so proto3 decodes the default). Scripted
// peers are therefore old peers unless a test says otherwise.
private static Handshake MatchingHandshake(
SqliteLocalDb sessionDb, string peerNodeId, long lastAppliedRemoteSeq = 0, bool supportsPartialSnapshot = false)
{
var h = new Handshake { NodeId = peerNodeId, LibSchemaVersion = 1, LastAppliedRemoteSeq = lastAppliedRemoteSeq };
var h = new Handshake
{
NodeId = peerNodeId,
LibSchemaVersion = 1,
LastAppliedRemoteSeq = lastAppliedRemoteSeq,
SupportsPartialSnapshot = supportsPartialSnapshot,
};
foreach (var t in sessionDb.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal))
h.Tables.Add(new TableDigest { TableName = t.Name, Digest = t.Digest });
return h;
@@ -114,14 +124,34 @@ public sealed class SnapshotResyncTests : IDisposable
// Completes the two-phase handshake against a real session: the peer never requests a snapshot.
private static async Task DriveHandshakeAsync(
SqliteLocalDb sessionDb, ChannelWriter<SyncMessage> toSession, ChannelReader<SyncMessage> fromSession, CancellationToken ct)
SqliteLocalDb sessionDb, ChannelWriter<SyncMessage> toSession, ChannelReader<SyncMessage> fromSession,
CancellationToken ct, bool supportsPartialSnapshot = false, long lastAppliedRemoteSeq = 0)
{
await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.Handshake, ct);
await toSession.WriteAsync(new SyncMessage { Handshake = MatchingHandshake(sessionDb, "peer") }, ct);
await toSession.WriteAsync(
new SyncMessage
{
Handshake = MatchingHandshake(
sessionDb, "peer", lastAppliedRemoteSeq, supportsPartialSnapshot),
}, ct);
await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.HandshakeAck, ct);
await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = false } }, ct);
}
// Reads outbound messages until (and including) the first of the given kind.
private static async Task<List<SyncMessage>> DrainThroughAsync(
ChannelReader<SyncMessage> reader, SyncMessage.MsgOneofCase last, CancellationToken ct)
{
var all = new List<SyncMessage>();
while (true)
{
var msg = await NextAsync(reader, ct);
all.Add(msg);
if (msg.MsgCase == last)
return all;
}
}
private static async Task SwallowAsync(Task task)
{
try { await task; }
@@ -159,6 +189,12 @@ public sealed class SnapshotResyncTests : IDisposable
return r[0];
}
/// <summary>The tables individually flagged as owing the peer a snapshot (schema v3).</summary>
private static async Task<List<string>> PendingSnapshotTables(SqliteLocalDb db) =>
(await db.QueryAsync(
"SELECT table_name FROM __localdb_snapshot_state WHERE needs_snapshot <> 0 ORDER BY table_name",
x => x.GetString(0))).ToList();
private static async Task<long> OplogCount(SqliteLocalDb db)
{
var r = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", x => x.GetInt64(0));
@@ -341,6 +377,93 @@ public sealed class SnapshotResyncTests : IDisposable
await SwallowAsync(runB);
}
[Fact]
public async Task CapPrune_IntoALiveSessionsUnsentRange_FaultsIt_AndTheReconnectSnapshots()
{
// P4-1. The pump reads `seq > sentThruSeq`, which resumes SILENTLY above a hole a cap prune
// punched — so it keeps shipping later deltas, the peer acks them, and the pair is durably
// divergent. `needs_snapshot` (set in the prune's own transaction) heals only a RECONNECT,
// and nothing forced one: on a stable LAN with a slow-but-alive peer the session, and the
// divergence, could last for days. Faulting is what turns the persisted flag into a snapshot.
var a = await NewSideAsync();
for (var i = 1; i <= 6; i++)
await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i });
var (duplex, toSession, fromSession) = ScriptedPeer();
using var cts = new CancellationTokenSource(RunTimeout);
var run = a.Session.RunAsync(duplex, cts.Token);
// The peer holds everything so far: no snapshot is owed, and the pump idles at seq 6.
await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token, lastAppliedRemoteSeq: 6);
// Write three more rows and prune all but the newest, in ONE transaction — which is what a
// cap prune is, and what makes this deterministic: the pump's reader can only ever see the
// before state (nothing above seq 6) or the after state (seqs 7 and 8 already gone). It can
// never catch rows 7/8 in between and legitimately advance past them.
await using (var tx = await a.Db.BeginTransactionAsync(cts.Token))
{
await tx.ExecuteAsync(
"INSERT INTO orders (id, sku, qty) VALUES (7,'PRUNED',7),(8,'PRUNED',8),(9,'KEPT',9)", null, cts.Token);
await tx.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 1 WHERE id = 1", null, cts.Token);
await tx.ExecuteAsync(
"DELETE FROM __localdb_oplog WHERE seq < (SELECT MAX(seq) FROM __localdb_oplog)", null, cts.Token);
await tx.CommitAsync(cts.Token);
}
var fault = await Assert.ThrowsAsync<LocalDbOplogGapException>(() => run);
Assert.Contains("6", fault.Message, StringComparison.Ordinal);
// And it faulted BEFORE shipping anything from above the hole: a delta for seq 9 would have
// moved the peer's watermark past rows 7 and 8, which is the divergence itself.
while (fromSession.TryRead(out var stray))
Assert.NotEqual(SyncMessage.MsgOneofCase.DeltaBatch, stray.MsgCase);
// The reconnect is what heals: the prune's needs_snapshot makes the handshake owe a full
// snapshot, which carries the pruned rows' current state to a real peer.
var a2 = NewSessionOver(a.Db);
var b = await NewSideAsync();
var (da, db) = DuplexPair();
var run2A = a2.Session.RunAsync(da, cts.Token);
var run2B = b.Session.RunAsync(db, cts.Token);
await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 9, RunTimeout);
Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db));
cts.Cancel();
await SwallowAsync(run2A);
await SwallowAsync(run2B);
}
[Fact]
public async Task AckPruningToEmpty_DoesNotFaultALiveSession()
{
// The other half of the horizon check: ordinary ack-pruning deletes seq <= last_acked_seq,
// which by construction sits at or below the sent watermark, so a converged pair that has
// pruned its whole oplog must keep streaming. A gap test written against `needs_snapshot`,
// or one that treated any risen horizon as a fault, would kill healthy sessions here.
var a = await NewSideAsync();
var b = await NewSideAsync();
for (var i = 1; i <= 4; i++)
await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i });
var (da, db) = DuplexPair();
using var cts = new CancellationTokenSource(RunTimeout);
var runA = a.Session.RunAsync(da, cts.Token);
var runB = b.Session.RunAsync(db, cts.Token);
await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 4, RunTimeout);
await WaitForAsync(async () => await OplogCount(a.Db) == 0, RunTimeout);
// Still live, still incremental: a post-prune write reaches B as an ordinary delta.
await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (5, 'POST', 5)");
await WaitForAsync(async () => (await ReadOrders(b.Db)).Any(r => r.Id == 5), RunTimeout);
Assert.False(runA.IsFaulted, runA.Exception?.ToString());
cts.Cancel();
await SwallowAsync(runA);
await SwallowAsync(runB);
}
[Fact]
public async Task Snapshot_MergesNotTruncates()
{
@@ -585,6 +708,261 @@ public sealed class SnapshotResyncTests : IDisposable
await SwallowAsync(run);
}
// ---- Per-table snapshot scoping (0.3.0) ----------------------------------------------
private const string WidgetsSql = "CREATE TABLE widgets (id INTEGER PRIMARY KEY, name TEXT)";
/// <summary>
/// A node with two registered tables where exactly one — <c>orders</c> — was baselined.
/// <c>widgets</c> is registered BEFORE its rows are written, so ordinary capture ledgers them;
/// <c>orders</c> is registered after, so only baselining can. That asymmetry is what makes the
/// scope observable: a full snapshot carries both tables' ledgers, a scoped one carries orders
/// alone — and both tables hold rows either way, so an empty widgets stream cannot be mistaken
/// for "there was nothing to send".
/// </summary>
private async Task<Side> NewBaselinedSideAsync(ReplicationOptions? options = null)
{
var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
_paths.Add(path);
var db = new SqliteLocalDb(new LocalDbOptions { Path = path });
_disposables.Add(db);
await db.ExecuteAsync(OrdersSql);
await db.ExecuteAsync(WidgetsSql);
// Predates registration: invisible to CDC, so only baselining puts it in the ledger.
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'PRE',1),(2,'PRE',2)");
db.RegisterReplicated("widgets");
await db.ExecuteAsync("INSERT INTO widgets (id, name) VALUES (1,'W1'),(2,'W2')");
db.RegisterReplicated("orders", baselineExistingRows: true);
return NewSessionOver(db, options);
}
/// <summary>The peer of <see cref="NewBaselinedSideAsync"/>: the same two tables, empty, owing nothing.</summary>
private async Task<Side> NewTwoTableSideAsync(ReplicationOptions? options = null)
{
var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
_paths.Add(path);
var db = new SqliteLocalDb(new LocalDbOptions { Path = path });
_disposables.Add(db);
await db.ExecuteAsync(OrdersSql);
await db.ExecuteAsync(WidgetsSql);
db.RegisterReplicated("orders");
db.RegisterReplicated("widgets");
return NewSessionOver(db, options);
}
private static List<string> SnapshotTablesIn(IEnumerable<SyncMessage> messages) =>
messages.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
.Select(m => m.SnapshotBatch.TableName)
.Distinct(StringComparer.Ordinal)
.Order(StringComparer.Ordinal)
.ToList();
[Fact]
public async Task Baselining_OneTable_StreamsOnlyThatTable()
{
// The whole point of the per-table flag. Through 0.2.x the flag was one bit for the whole
// database, so baselining `orders` re-streamed `widgets` (and everything else) in full — in
// both directions at once on a pair enabling replication late, which on a database holding
// an event log at its retention cap is the largest transfer this library performs.
var a = await NewBaselinedSideAsync();
Assert.Equal(["orders"], await PendingSnapshotTables(a.Db));
Assert.Equal(0, await NeedsSnapshot(a.Db)); // NOT the database-wide flag
var (duplex, toSession, fromSession) = ScriptedPeer();
using var cts = new CancellationTokenSource(RunTimeout);
var run = a.Session.RunAsync(duplex, cts.Token);
await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token, supportsPartialSnapshot: true);
var frames = await DrainThroughAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);
var begin = frames.Single(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBegin).SnapshotBegin;
Assert.Equal(["orders"], begin.Tables);
Assert.Equal(["orders"], SnapshotTablesIn(frames));
// The baselined rows really are in there — the scope narrowed the snapshot, it did not empty it.
var pks = frames.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
.SelectMany(m => m.SnapshotBatch.Rows).Select(r => r.PkJson).Order(StringComparer.Ordinal);
Assert.Equal(["{\"id\":1}", "{\"id\":2}"], pks);
cts.Cancel();
await SwallowAsync(run);
}
[Fact]
public async Task Baselining_AgainstAPeerWithoutPartialSupport_StreamsEveryTable()
{
// Mixed-version interop, the hard constraint. A pre-0.3.0 peer ignores SnapshotBegin.tables
// and would advance last_applied_remote_seq to as_of_seq on complete — which for a SCOPED
// snapshot silently discards every delta below as_of for the tables it left out. There is no
// way to express "partial" to that peer, so it is served the 0.2.x answer: everything.
var a = await NewBaselinedSideAsync();
var (duplex, toSession, fromSession) = ScriptedPeer();
using var cts = new CancellationTokenSource(RunTimeout);
var run = a.Session.RunAsync(duplex, cts.Token);
// supportsPartialSnapshot omitted == what a 0.1.x/0.2.x handshake decodes to.
await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
var frames = await DrainThroughAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);
var begin = frames.Single(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBegin).SnapshotBegin;
Assert.Empty(begin.Tables); // empty == full, the only scope the old wire can express
Assert.Equal(["orders", "widgets"], SnapshotTablesIn(frames));
cts.Cancel();
await SwallowAsync(run);
}
[Fact]
public async Task PartialSnapshot_DebtClearsOnlyWhenThePeerAcknowledges()
{
// Crash safety: an ungraceful exit part-way through a snapshot must re-stream it, so the
// flag cannot come down when the frames are merely enqueued. The peer's SnapshotAck is
// written after its own durable watermark write, so it means "I hold these rows".
var a = await NewBaselinedSideAsync();
var (duplex, toSession, fromSession) = ScriptedPeer();
using var cts = new CancellationTokenSource(RunTimeout);
var run = a.Session.RunAsync(duplex, cts.Token);
await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token, supportsPartialSnapshot: true);
var frames = await DrainThroughAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);
var complete = frames[^1].SnapshotComplete;
// The entire snapshot is on the wire and unacknowledged: the debt must still stand.
Assert.Equal(["orders"], await PendingSnapshotTables(a.Db));
await toSession.WriteAsync(new SyncMessage { SnapshotAck = new SnapshotAck { AsOfSeq = complete.AsOfSeq } }, cts.Token);
await WaitForAsync(async () => (await PendingSnapshotTables(a.Db)).Count == 0, RunTimeout);
cts.Cancel();
await SwallowAsync(run);
}
[Fact]
public async Task Snapshot_ToAPeerWithoutPartialSupport_ClearsOnEnqueue()
{
// The 0.2.x behaviour, kept deliberately: a peer that sends no SnapshotAck would otherwise
// leave the debt standing forever and re-stream the snapshot on every reconnect.
var a = await NewBaselinedSideAsync();
var (duplex, toSession, fromSession) = ScriptedPeer();
using var cts = new CancellationTokenSource(RunTimeout);
var run = a.Session.RunAsync(duplex, cts.Token);
await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
await DrainThroughAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);
await WaitForAsync(async () => (await PendingSnapshotTables(a.Db)).Count == 0, RunTimeout);
cts.Cancel();
await SwallowAsync(run);
}
[Fact]
public async Task PartialSnapshot_DoesNotAdvanceTheReceiversWatermark()
{
// The receiver's half of the same invariant. last_applied_remote_seq is ONE number for the
// whole peer oplog; moving it to a partial snapshot's as_of would declare the untouched
// tables applied through that point, and their deltas below it would never be resent.
var a = await NewSideAsync();
await a.Store.SetLastAppliedRemoteSeqAsync(10, 0);
var (duplex, toSession, fromSession) = ScriptedPeer();
using var cts = new CancellationTokenSource(RunTimeout);
var run = a.Session.RunAsync(duplex, cts.Token);
await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token, supportsPartialSnapshot: true);
var begin = new SnapshotBegin { AsOfSeq = 900 };
begin.Tables.Add("orders");
await toSession.WriteAsync(new SyncMessage { SnapshotBegin = begin }, cts.Token);
await toSession.WriteAsync(new SyncMessage
{
SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(1, "SNAP", 6_000_000) } },
}, cts.Token);
await toSession.WriteAsync(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 900 } }, cts.Token);
// The ack proves the complete was processed — and that a 0.3.0 receiver answers one at all.
var ack = await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotAck, cts.Token);
Assert.Equal(900, ack.SnapshotAck.AsOfSeq);
Assert.Equal(10, await LastAppliedRemoteSeq(a.Db)); // NOT 900
Assert.Single(await ReadOrders(a.Db)); // the rows still merged
cts.Cancel();
await SwallowAsync(run);
}
[Fact]
public async Task FullSnapshot_ToAnOldPeer_DrawsNoSnapshotAck()
{
// Symmetry check on the negotiation: a 0.3.0 receiver must not put a message a pre-0.3.0
// sender has never heard of on the wire back to it.
var a = await NewSideAsync();
var (duplex, toSession, fromSession) = ScriptedPeer();
using var cts = new CancellationTokenSource(RunTimeout);
var run = a.Session.RunAsync(duplex, cts.Token);
await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
await toSession.WriteAsync(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 300 } }, cts.Token);
await toSession.WriteAsync(new SyncMessage
{
SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(1, "SNAP", 6_000_000) } },
}, cts.Token);
await toSession.WriteAsync(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 300 } }, cts.Token);
// A full snapshot from an old peer still advances the watermark — that is the 0.2.x contract.
await WaitForAsync(async () => await LastAppliedRemoteSeq(a.Db) == 300, RunTimeout);
// Barrier: an ack for this delta proves everything before it was processed, so an absent
// SnapshotAck is absent rather than merely late.
await toSession.WriteAsync(new SyncMessage
{
DeltaBatch = new DeltaBatch { Entries = { new OplogEntry
{
Seq = 301, TableName = "orders", PkJson = "{\"id\":2}",
RowJson = "{\"id\":2,\"sku\":\"BARRIER\",\"qty\":1}", Hlc = 6_000_001, NodeId = "peer", IsTombstone = false,
} } },
}, cts.Token);
await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token);
Assert.False(fromSession.TryPeek(out var stray) && stray.MsgCase == SyncMessage.MsgOneofCase.SnapshotAck);
cts.Cancel();
await SwallowAsync(run);
}
[Fact]
public async Task BaselinedTable_ReachesARealPeer_WithoutRestreamingTheOthers()
{
// End to end over two real databases: the baselined table's pre-existing rows arrive via the
// scoped snapshot, the untouched table's rows arrive as ordinary deltas, and the debt clears.
var options = new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) };
var a = await NewBaselinedSideAsync(options);
var b = await NewTwoTableSideAsync(options);
var (da, db) = DuplexPair();
using var cts = new CancellationTokenSource(RunTimeout);
var runA = a.Session.RunAsync(da, cts.Token);
var runB = b.Session.RunAsync(db, cts.Token);
async Task<bool> Converged() =>
(await ReadOrders(b.Db)).Count == 2
&& (await b.Db.QueryAsync("SELECT COUNT(*) FROM widgets", x => x.GetInt64(0)))[0] == 2;
await WaitForAsync(Converged, RunTimeout);
Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db));
await WaitForAsync(async () => (await PendingSnapshotTables(a.Db)).Count == 0, RunTimeout);
// A partial snapshot speaks for nothing outside its tables, so neither side may skip the
// deltas it covered: B's watermark tracks the oplog, not the snapshot's as-of.
Assert.Equal(0, await NeedsSnapshot(a.Db));
cts.Cancel();
await SwallowAsync(runA);
await SwallowAsync(runB);
}
[Fact]
public async Task Sender_ClearsNeedsSnapshotFlag()
{
@@ -208,6 +208,50 @@ public sealed class SyncSessionTests : IDisposable
Assert.Empty(await ReadOrders(b.Db));
}
[Fact]
public async Task Handshake_FaultTeardown_StillTransmitsTheLocalHandshake()
{
// Schema-mismatch detection is symmetric: each side computes its verdict FROM the other's
// Handshake. So a session that faults on the peer's handshake must still flush its own —
// otherwise the healthy peer learns nothing and waits out its receive cap (tests) or
// misclassifies the mismatch as a generic disconnect and reconnect-loops (production).
//
// The race being pinned: with the peer's handshake already buffered, the path from
// enqueueing our Handshake on the control lane to the validation throw has no yields,
// while the writer's wakeup sits in the thread-pool queue — a teardown that cancels the
// writer can therefore kill the Handshake before it ever transmits. Preloading makes the
// sprint as tight as possible, and running the scenarios concurrently starves the pool
// the way a busy test suite does.
const int Iterations = 40;
var sides = new List<Side>(Iterations);
for (var i = 0; i < Iterations; i++)
sides.Add(await NewSideAsync());
using var cts = new CancellationTokenSource(RunTimeout);
var runs = new List<(Task Run, ChannelReader<SyncMessage> FromSession)>(Iterations);
foreach (var side in sides)
{
var (duplex, toSession, fromSession) = ScriptedPeer();
var mismatched = MatchingHandshake(side.Db, "peer");
mismatched.Tables[0].Digest = "not-the-local-digest";
await toSession.WriteAsync(new SyncMessage { Handshake = mismatched }, cts.Token);
runs.Add((side.Session.RunAsync(duplex, cts.Token), fromSession));
}
var lost = 0;
foreach (var (run, fromSession) in runs)
{
await Assert.ThrowsAsync<LocalDbSchemaMismatchException>(() => run);
var sent = false;
while (fromSession.TryRead(out var msg))
sent |= msg.MsgCase == SyncMessage.MsgOneofCase.Handshake;
if (!sent)
lost++;
}
Assert.True(lost == 0, $"{lost}/{Iterations} sessions tore down without transmitting their own Handshake");
}
[Fact]
public async Task Handshake_SchemaVersionMismatch_FailsClosed()
{
@@ -553,8 +597,9 @@ public sealed class SyncSessionTests : IDisposable
await a.Store.SetNeedsSnapshotAsync(true, default);
var (duplex, toSession, fromSession) = ScriptedPeer();
var sent = new TaskCompletionSource();
a.Session.SnapshotSender = async (send, token) =>
a.Session.SnapshotSender = async (scope, send, token) =>
{
Assert.False(scope.IsPartial); // the database-wide flag always means every table
sent.TrySetResult();
await send(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 7 } }, token);
return 7;
@@ -0,0 +1,181 @@
using System.Threading.Channels;
using Grpc.Core;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.LocalDb.Contracts;
using ZB.MOM.WW.LocalDb.Replication;
namespace ZB.MOM.WW.LocalDb.Tests;
/// <summary>
/// The passive service's single-stream guard under a HALF-OPEN link: the dialling node is gone, but
/// nothing on this side knows it, so the handler sits inside <c>MoveNext</c> forever holding the
/// guard. Through 0.3.1 that made the pair unrecoverable without a restart — every reconnect from
/// the rebooted peer was refused <c>AlreadyExists</c>, indefinitely on an idle pair, while this node
/// went on reporting itself connected. The guard is there to serialize sessions, not to fence out a
/// caller the auth interceptor already vouched for, so the newest dialler wins.
/// </summary>
public sealed class SyncStreamGuardTests : IDisposable
{
private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(10);
private readonly List<string> _paths = [];
private readonly List<IDisposable> _disposables = [];
public void Dispose()
{
foreach (var d in _disposables)
d.Dispose();
Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
foreach (var p in _paths)
{
if (File.Exists(p)) File.Delete(p);
if (File.Exists(p + "-wal")) File.Delete(p + "-wal");
if (File.Exists(p + "-shm")) File.Delete(p + "-shm");
}
}
private LocalDbSyncService NewService()
{
var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
_paths.Add(path);
var db = new Internal.SqliteLocalDb(new LocalDbOptions { Path = path });
_disposables.Add(db);
db.ExecuteAsync("CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT)").GetAwaiter().GetResult();
db.RegisterReplicated("orders");
return new LocalDbSyncService(
db,
Options.Create(new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) }),
NullLoggerFactory.Instance);
}
private static ServerCallContext ContextFor(CancellationToken ct) => new FakeCallContext(ct);
[Fact]
public async Task WedgedIncumbent_IsDisplacedByANewStream()
{
var service = NewService();
// Eviction is normally generous; the test only needs it to not be the thing under test.
service.TakeoverTimeout = Timeout;
// The incumbent's peer crashed: its stream will never deliver another message, and neither
// end times out on silence — a converged pair is legitimately quiet. The handler parks in
// the handshake's first read, holding the guard.
using var incumbentCts = new CancellationTokenSource();
var incumbentOut = new RecordingWriter();
var incumbent = service.Sync(new SilentReader(), incumbentOut, ContextFor(incumbentCts.Token));
// It really is running and really did claim the guard (its handshake is on the wire).
await incumbentOut.NextAsync(Timeout);
Assert.False(incumbent.IsCompleted);
// The rebooted peer dials again. Pre-fix this threw AlreadyExists and the pair stayed wedged.
using var newcomerCts = new CancellationTokenSource();
var newcomerOut = new RecordingWriter();
var newcomer = service.Sync(new SilentReader(), newcomerOut, ContextFor(newcomerCts.Token));
// The incumbent is torn down — gracefully, not as a fault of this node.
await incumbent.WaitAsync(Timeout);
// ...and the new stream is genuinely served: its own handshake goes out.
var handshake = await newcomerOut.NextAsync(Timeout);
Assert.Equal(SyncMessage.MsgOneofCase.Handshake, handshake.MsgCase);
Assert.False(newcomer.IsCompleted);
newcomerCts.Cancel();
await SwallowAsync(newcomer);
}
[Fact]
public async Task DisplacedIncumbent_ReleasesTheGuard_SoAThirdStreamIsAlsoServed()
{
// Takeover must leave the guard in a state that can be taken over AGAIN — a handover that
// leaked the slot would simply move the wedge one reconnect down the line.
var service = NewService();
service.TakeoverTimeout = Timeout;
using var firstCts = new CancellationTokenSource();
var firstOut = new RecordingWriter();
var first = service.Sync(new SilentReader(), firstOut, ContextFor(firstCts.Token));
await firstOut.NextAsync(Timeout);
using var secondCts = new CancellationTokenSource();
var secondOut = new RecordingWriter();
var second = service.Sync(new SilentReader(), secondOut, ContextFor(secondCts.Token));
await first.WaitAsync(Timeout);
await secondOut.NextAsync(Timeout);
using var thirdCts = new CancellationTokenSource();
var thirdOut = new RecordingWriter();
var third = service.Sync(new SilentReader(), thirdOut, ContextFor(thirdCts.Token));
await second.WaitAsync(Timeout);
Assert.Equal(SyncMessage.MsgOneofCase.Handshake, (await thirdOut.NextAsync(Timeout)).MsgCase);
thirdCts.Cancel();
await SwallowAsync(third);
}
private static async Task SwallowAsync(Task task)
{
try { await task.WaitAsync(Timeout); }
catch (OperationCanceledException) { }
catch (RpcException) { }
}
/// <summary>
/// The minimum <see cref="ServerCallContext"/> the handler actually reads: its cancellation
/// token. (Grpc.Core.Api ships no test double, and the ASP.NET Core one is not reachable outside
/// a live call.)
/// </summary>
private sealed class FakeCallContext(CancellationToken ct) : ServerCallContext
{
protected override string MethodCore => "/localdb_sync.v1.LocalDbSync/Sync";
protected override string HostCore => "test";
protected override string PeerCore => "test";
protected override DateTime DeadlineCore => DateTime.MaxValue;
protected override Metadata RequestHeadersCore => [];
protected override CancellationToken CancellationTokenCore => ct;
protected override Metadata ResponseTrailersCore { get; } = [];
protected override Status StatusCore { get; set; }
protected override WriteOptions? WriteOptionsCore { get; set; }
protected override AuthContext AuthContextCore => new(null, new Dictionary<string, List<AuthProperty>>());
protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions? options) =>
throw new NotSupportedException();
protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => Task.CompletedTask;
}
/// <summary>A stream that never delivers anything and never ends — a half-open link.</summary>
private sealed class SilentReader : IAsyncStreamReader<SyncMessage>
{
public SyncMessage Current => throw new InvalidOperationException("no message");
public async Task<bool> MoveNext(CancellationToken cancellationToken)
{
await Task.Delay(System.Threading.Timeout.Infinite, cancellationToken);
return false;
}
}
private sealed class RecordingWriter : IServerStreamWriter<SyncMessage>
{
private readonly Channel<SyncMessage> _written = Channel.CreateUnbounded<SyncMessage>();
public WriteOptions? WriteOptions { get; set; }
public Task WriteAsync(SyncMessage message) => WriteAsync(message, CancellationToken.None);
public Task WriteAsync(SyncMessage message, CancellationToken cancellationToken)
{
_written.Writer.TryWrite(message);
return Task.CompletedTask;
}
public async Task<SyncMessage> NextAsync(TimeSpan timeout)
{
using var cts = new CancellationTokenSource(timeout);
return await _written.Reader.ReadAsync(cts.Token);
}
}
}
+13 -2
View File
@@ -16,9 +16,20 @@ set -euo pipefail
: "${GITEA_NUGET_SOURCE:?set GITEA_NUGET_SOURCE to your Gitea NuGet feed URL}"
: "${GITEA_NUGET_KEY:?set GITEA_NUGET_KEY to your Gitea access token}"
dotnet pack -c Release -o ./artifacts
# Named solution, named output, and a cd to the repo root, all for one reason: `dotnet pack`
# with neither a project argument nor a fixed working directory packs whatever solution
# happens to sit in the CALLER's directory — and then pushes it to a live feed under this
# script's name. Invoked by absolute path from another repo's checkout, the relative form
# packed and published 25 unrelated packages (LocalDb's copy of this script, fixed in
# 0038d46; recovered with `dotnet nuget delete`, but only because the token happened to
# carry delete rights). A publish script must be unambiguous about WHAT it publishes no
# matter where it is run.
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$REPO_ROOT"
dotnet nuget push "./artifacts/*.nupkg" \
dotnet pack ZB.MOM.WW.Secrets.slnx -c Release -o "$REPO_ROOT/artifacts"
dotnet nuget push "$REPO_ROOT/artifacts/*.nupkg" \
--source "$GITEA_NUGET_SOURCE" \
--api-key "$GITEA_NUGET_KEY" \
--skip-duplicate
+7 -1
View File
@@ -6,4 +6,10 @@
set -euo pipefail
dotnet pack -c Release -o ./artifacts
# cd to the repo root and name the solution + an absolute output path: `dotnet pack` with
# neither a project argument nor a fixed working directory packs whatever solution happens
# to sit in the CALLER's directory (the bug behind the LocalDb feed incident, 0038d46).
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$REPO_ROOT"
dotnet pack ZB.MOM.WW.Theme.slnx -c Release -o "$REPO_ROOT/artifacts"
+13 -2
View File
@@ -16,9 +16,20 @@ set -euo pipefail
: "${GITEA_NUGET_SOURCE:?set GITEA_NUGET_SOURCE to your Gitea NuGet feed URL}"
: "${GITEA_NUGET_KEY:?set GITEA_NUGET_KEY to your Gitea access token}"
dotnet pack -c Release -o ./artifacts
# Named solution, named output, and a cd to the repo root, all for one reason: `dotnet pack`
# with neither a project argument nor a fixed working directory packs whatever solution
# happens to sit in the CALLER's directory — and then pushes it to a live feed under this
# script's name. Invoked by absolute path from another repo's checkout, the relative form
# packed and published 25 unrelated packages (LocalDb's copy of this script, fixed in
# 0038d46; recovered with `dotnet nuget delete`, but only because the token happened to
# carry delete rights). A publish script must be unambiguous about WHAT it publishes no
# matter where it is run.
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$REPO_ROOT"
dotnet nuget push "./artifacts/*.nupkg" \
dotnet pack ZB.MOM.WW.Theme.slnx -c Release -o "$REPO_ROOT/artifacts"
dotnet nuget push "$REPO_ROOT/artifacts/*.nupkg" \
--source "$GITEA_NUGET_SOURCE" \
--api-key "$GITEA_NUGET_KEY" \
--skip-duplicate