dohertj2/scadaproj
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'localdb-perf-remediation'

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-15 01:46:38 -04:00
parent 92246c9734 14cb283172
commit f627cee9c0
21 changed files with 2152 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CLAUDE.md
+3 -3
View File
File diff suppressed because one or more lines are too long
ZB.MOM.WW.LocalDb/Directory.Build.props
+1 -1
View File
@@ -5,7 +5,7 @@
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<LangVersion>latest</LangVersion>
<Version>0.1.3</Version>
<Version>0.2.1</Version>
<ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<!-- NU1510 (package-pruning advice): the Replication and Tests projects carry a
ZB.MOM.WW.LocalDb/README.md
+90 -15
View File
@@ -10,16 +10,59 @@ 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.1.3`)
## Packages (all `0.2.1`)
| Package | Contents |
|---|---|
| `ZB.MOM.WW.LocalDb` | Embedded DB host: `ILocalDb` SQL surface (`ExecuteAsync` / `QueryAsync` / transactions), `RegisterReplicated(table)` (PK validation + trigger install + oplog schema), `HybridLogicalClock`, `AddZbLocalDb()` DI. |
| `ZB.MOM.WW.LocalDb` | Embedded DB host: `ILocalDb` SQL surface (`ExecuteAsync` / `QueryAsync` / transactions), `RegisterReplicated(table[, baselineExistingRows])` (PK validation + trigger install + oplog schema), `DeregisterReplicated(table)` (trigger removal + oplog/ledger prune), `HybridLogicalClock`, `AddZbLocalDb()` DI. |
| `ZB.MOM.WW.LocalDb.Contracts` | `localdb_sync.v1` proto + generated gRPC types (packable; a future peer could be non-.NET). |
| `ZB.MOM.WW.LocalDb.Replication` | gRPC sync engine: passive `LocalDbSyncService`, initiating `SyncBackgroundService`, `MaintenanceBackgroundService`, LWW apply, watermark acks + pruning, snapshot resync, `ISyncStatus`, `LocalDbMetrics`, `AddZbLocalDbReplication()` / `MapZbLocalDbSync()`. |
> Status: **published to the Gitea feed; adopted by ScadaBridge (Phases 1+2) and OtOpcUa (Phase 1).**
>
> - `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
> open is `MAX(hlc_high_water, oplog HLCs, ledger HLCs, last_seen_hlc)`, and `hlc_high_water` was
> written only by graceful disposal — while deregistration deletes the table's oplog AND ledger
> rows. A crash after a deregister boot could therefore recover the clock below stamps the peer
> already holds, and LWW discards those as older. The current clock is now flushed inside the
> deregistration transaction, before the prune (and the flush is MAX-guarded, so it can never
> regress the anchor).
> - **The sync inbox is bounded (64 messages) instead of unbounded**, so inbound memory is capped
> at roughly `64 x MaxBatchBytes` rather than "whatever the peer sends" — the bad case being a
> mutual full snapshot of a large dataset, buffered whole on both nodes at once. A full inbox now
> stops draining the gRPC stream, which closes the peer's flow-control window.
> - **A snapshot this node owes is streamed alongside a running receive loop**, not from inside the
> handshake. That is what makes bounding safe: previously neither side drained its inbox while
> sending, so a mutual snapshot converged only because the inbox was unbounded. Interleaving was
> already safe (snapshot rows and deltas both merge through LWW, the watermark write is
> MAX-guarded) — only the ordering changed.
> - **Documented the cost of late baselining**: the snapshot flag is per-database, so baselining
> one table re-streams every registered table, in both directions on a late-enabled pair. Trim
> high-volume tables (event logs) before turning replication on late.
> - `0.2.0` (2026-08-14) — performance + lifecycle work from the ScadaBridge architectural review.
> All of it is wire-compatible with `0.1.x` peers (no proto change, no `lib_schema_version` bump);
> the on-disk bookkeeping schema goes to **v2** and is upgraded in place on open.
> - **`DeregisterReplicated(table)`** — drops a table's capture triggers and prunes its oplog and
> row-version rows. A node that does not replicate should not be paying for capture on every
> write, and until now there was no way to stop once a file had been registered. Idempotent, and
> usable on triggers left behind by an earlier build.
> - **`RegisterReplicated(table, baselineExistingRows: true)`** — closes the late-opt-in hole: rows
> that predate registration were invisible to the peer forever, because the snapshot streamer
> pages from the row-version ledger and CDC never wrote them there. Opting in seeds the ledger at
> an LWW-safe floor (HLC `0`, this node's id) and flags a snapshot resync so they actually
> travel. Default is still off — the documented CDC contract is unchanged unless you ask.
> - **O(1) backlog depth** — `MAX(seq) - MAX(last_acked_seq, MIN(seq) - 1)` replaces a
> `COUNT(*)` over the whole unacked backlog that ran once a minute AND on every metrics scrape
> (measured 58 VM steps vs 100,020 on a 50k backlog).
> - **Byte-budget batching** — `MaxBatchBytes` (default 2 MB) bounds delta and snapshot messages by
> summed serialized size, with `MaxBatchSize` demoted to a secondary row cap. Consumers no longer
> have to tune the row count against their widest row to stay under the gRPC message limit.
> Inbound delta batches and acks already queued are coalesced into one transaction / one ack.
> - **`LwwApplier` prepares once per statement** instead of three SQL parses per applied row.
> - **Dropped `__localdb_oplog_hlc`** — an index no query ever used, costing a b-tree insert inside
> every captured consumer write. Existing files lose it on the next open.
> - `0.1.3` (2026-07-21) — fixes the other half of the rebuilt-peer story: a rebuilt peer's OWN
> writes were silently dropped. `last_applied_remote_seq` is a watermark in the *peer's* seq
> space, and a rebuilt peer numbers from 1 again — so the healthy node advertised the old peer's
@@ -107,7 +150,8 @@ node still pushes its own deltas back. Only one side needs `PeerAddress`.
| `PeerAddress` | `""` | gRPC address to dial. Empty ⇒ passive (initiator idles). |
| `ApiKey` | `null` | Bearer token the initiator presents on its channel; `null` ⇒ no key header. |
| `FlushInterval` | `250 ms` | How often the change pump flushes pending oplog deltas. |
| `MaxBatchSize` | `500` | Max oplog entries per outbound batch (also the snapshot page size). |
| `MaxBatchBytes` | `2 MB` | **Primary** batch cap: summed serialized bytes per outbound delta/snapshot message. Sized under the 4 MB gRPC default; a single row above the budget is still sent, alone. |
| `MaxBatchSize` | `500` | **Secondary** row-count cap per outbound batch (also the snapshot DB page size). |
| `MaxOplogRows` | `1_000_000` | Backlog ceiling; exceeding it flags a snapshot resync and prunes oldest rows. |
| `MaxOplogAge` | `7 days` | Age ceiling; an unpruned entry older than this flags a snapshot resync. |
| `TombstoneRetention` | `7 days` | How long deleted-row tombstones are kept before pruning. |
@@ -134,7 +178,9 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
- **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.
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.
- **Maintenance** — a background service on **both** roles enforces the oplog row/age caps
and prunes expired tombstones every `MaintenanceInterval`.
@@ -161,13 +207,27 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
still hold a BLOB value at runtime, which then surfaces as a write-time error, not a
registration error.
- **CDC semantics** — rows that existed **before** `RegisterReplicated` are not captured
and are not snapshotted; only changes after registration are replicated. Re-baselining
pre-existing rows requires re-writing them.
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
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`.
- **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).
- A **mutual simultaneous snapshot** buffers the peer's inbound snapshot in memory while
this node is still sending its own (acceptable for the small datasets this lib targets).
- **Inbound buffering is bounded, not zero.** The duplex inbox holds up to 64 messages, so
peak inbound memory is roughly `64 x MaxBatchBytes` (~128 MB at the 2 MB default, far
less in practice — messages are only that large when the sender has that much backlog).
Lower `MaxBatchBytes` to lower the ceiling. Beyond the bound the stream stops being
drained and the peer is slowed by flow control.
## Observability
@@ -198,6 +258,19 @@ provider is wired or the poll threw, so a DB failure never reads as "0 backlog")
- 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.
- **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
replicates stops syncing with a schema-mismatch error. Deregister on both nodes, or on
neither. Re-enabling later should pass `baselineExistingRows: true`, or the rows that
accumulated while capture was off stay invisible to the peer — but trim high-volume
tables first (see Limitations: baselining flags a full, all-tables snapshot). The HLC's
durable anchor is flushed in the same transaction, so a crash after a deregister boot
cannot bring the clock back below stamps peers already hold.
- **On-disk schema upgrades** are applied in place when the file is opened, in one
transaction, and every step is safe to re-run. `schema_version` in `__localdb_meta` is
local-file state only — it is NOT the wire `lib_schema_version`, so a pair mid-upgrade
keeps syncing.
- The **`__localdb_dead_letter`** table holds poison rows (bad JSON, null-row non-tombstone
entries) and over-drift rejects (when `FailClosedOnDrift`). Inspect it to diagnose
rejects.
@@ -208,13 +281,15 @@ provider is wired or the poll threw, so a DB failure never reads as "0 backlog")
## Testing
144 tests, all offline / no live dependencies — unit (HLC ordering + restart
persistence, trigger SQL generation, LWW decision table, watermark/pruning, digest
computation), 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), plus seeded randomized property tests (random interleaved op
sequences → both DBs identical after quiesce). Family conventions apply
(`TreatWarningsAsErrors`).
188 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).
Family conventions apply (`TreatWarningsAsErrors`).
Build / test from this directory:
ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/GrpcSyncDuplex.cs
+50 -8
View File
@@ -12,19 +12,58 @@ namespace ZB.MOM.WW.LocalDb.Replication.Internal;
internal static class GrpcSyncDuplex
{
/// <summary>
/// Wires <paramref name="reader"/> into an unbounded inbox channel drained by a background loop,
/// and exposes <paramref name="writer"/> as the duplex send. Returns the duplex plus the reader
/// loop's task so the caller can observe it. On clean stream end the inbox channel is completed;
/// on a reader fault it is completed WITH the exception — the session's handshake and receive
/// phases both await the inbox, so terminating it is what unblocks them when the call dies (they
/// would otherwise hang on a dead stream).
/// Inbound messages buffered between the transport and the session's receive loop.
/// </summary>
/// <remarks>
/// <para>
/// <b>Why bounded at all.</b> An unbounded inbox makes this process's memory a function of what
/// the PEER decides to send. The worst case is not hypothetical: a full snapshot streams every
/// row of every registered table, and both sides of a pair that enabled replication late
/// snapshot each other simultaneously — so an unbounded inbox holds an entire peer dataset (an
/// event log at its storage cap, say) in memory on both nodes at once. Bounding it converts that
/// into backpressure: the pump stops draining the gRPC stream, the HTTP/2 flow-control window
/// closes, and the SENDER slows down.
/// </para>
/// <para>
/// <b>Why 64.</b> Memory here is bounded by roughly capacity x
/// <see cref="ReplicationOptions.MaxBatchBytes"/> (the sender's own per-message budget), so the
/// ceiling is a configured number rather than a peer-controlled one. 64 is four times the
/// sender's outbound data-lane depth, so a healthy stream never reaches the bound and pays
/// nothing; it engages only when this node is genuinely slower than the peer for a sustained
/// burst, which is exactly when backpressure is wanted. Lower <c>MaxBatchBytes</c> to lower the
/// ceiling.
/// </para>
/// <para>
/// <b>What makes bounding safe.</b> Backpressure on the inbox deadlocks any design where both
/// peers can stop draining their inbox at the same time — mutual snapshots were exactly that
/// case while the send ran inside the handshake. <see cref="SyncSession"/> therefore starts its
/// receive loop BEFORE streaming a snapshot it owes, so the inbox is drained continuously for
/// the whole life of the session; see the remarks on its snapshot phase.
/// </para>
/// </remarks>
internal const int InboxCapacity = 64;
/// <summary>
/// Wires <paramref name="reader"/> into a bounded inbox channel (see
/// <see cref="InboxCapacity"/>) drained by a background loop, and exposes
/// <paramref name="writer"/> as the duplex send. Returns the duplex plus the reader loop's task
/// so the caller can observe it. On clean stream end the inbox channel is completed; on a reader
/// fault it is completed WITH the exception — the session's handshake and receive phases both
/// await the inbox, so terminating it is what unblocks them when the call dies (they would
/// otherwise hang on a dead stream).
/// </summary>
public static (SyncDuplex Duplex, Task ReaderTask) Create(
IAsyncStreamReader<SyncMessage> reader,
IAsyncStreamWriter<SyncMessage> writer,
CancellationToken ct)
{
var inbox = Channel.CreateUnbounded<SyncMessage>(new UnboundedChannelOptions { SingleWriter = true });
var inbox = Channel.CreateBounded<SyncMessage>(new BoundedChannelOptions(InboxCapacity)
{
// Wait, never Drop: every inbound message is protocol state (a delta the peer will not
// resend in-session, a snapshot frame, an ack). Dropping one is silent divergence.
FullMode = BoundedChannelFullMode.Wait,
SingleWriter = true,
});
var readerTask = PumpInboundAsync(reader, inbox.Writer, ct);
var duplex = new SyncDuplex
{
@@ -44,7 +83,10 @@ internal static class GrpcSyncDuplex
try
{
while (await reader.MoveNext(ct))
inbox.TryWrite(reader.Current);
// WriteAsync, not TryWrite: on a FULL bounded channel TryWrite returns false and the
// message is silently lost. Awaiting here is the backpressure — it stops draining the
// gRPC stream, which closes the peer's flow-control window.
await inbox.WriteAsync(reader.Current, ct);
}
catch (Exception ex)
{
ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/LwwApplier.cs
+187 -71
View File
@@ -16,10 +16,24 @@ internal sealed record ApplyResult(long AppliedThruSeq, int Applied, int Discard
/// crash-replay of the batch is a no-op (LWW's strict-greater compare loses on equal versions).
/// This never writes <c>__localdb_oplog</c>: the 2-node topology does not re-forward applied deltas.
/// </summary>
/// <remarks>
/// Every statement is issued through a <see cref="PreparedStatements"/> set held for the life of the
/// batch, so each distinct statement is prepared ONCE and then rebound per row. The previous shape
/// went through the generic <c>ILocalDbTransaction</c> façade, which builds a fresh
/// <see cref="SqliteCommand"/> per call — three SQL parses per applied row (version lookup, the
/// table's delete/upsert, the row-version write), on the path that drains a reconnect backlog.
/// </remarks>
internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow = null)
{
private readonly Func<DateTimeOffset> _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow);
/// <summary>
/// Distinct statements prepared by the most recent <see cref="ApplyBatchAsync"/>. Test-only
/// observation: it is the direct evidence that statement count tracks the number of distinct
/// STATEMENTS rather than the number of rows.
/// </summary>
internal int LastBatchStatementCount { get; private set; }
public async Task<ApplyResult> ApplyBatchAsync(IReadOnlyList<OplogEntryRecord> entries, CancellationToken ct)
{
if (entries.Count == 0)
@@ -42,13 +56,14 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
if (entry.Hlc > maxHlc) maxHlc = entry.Hlc;
}
var sqlByTable = new Dictionary<string, (string Delete, string Upsert)>(StringComparer.Ordinal);
var applied = 0;
var discarded = 0;
var deadLettered = 0;
await using var tx = await db.BeginTransactionAsync(ct);
await tx.ExecuteAsync("UPDATE __localdb_applying SET applying = 1 WHERE id = 1", null, ct);
await using var tx = await db.BeginPreparedTransactionAsync(ct);
using var sql = new PreparedStatements(tx.Connection, tx.Sqlite);
await sql.SetApplyingAsync(1, ct);
foreach (var entry in entries)
{
@@ -56,19 +71,16 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
// KEY table that silently inserts a phantom auto-rowid row — so it is poison, not a win.
if (!entry.IsTombstone && entry.RowJson is null)
{
await DeadLetterAsync(tx, entry, "non-tombstone entry with null row_json", utcNow, ct);
await sql.DeadLetterAsync(entry, "non-tombstone entry with null row_json", utcNow, ct);
deadLettered++;
continue;
}
var local = await tx.QueryAsync(
"SELECT hlc, node_id FROM __localdb_row_version WHERE table_name = @t AND pk_json = @pk",
static r => (Hlc: r.GetInt64(0), NodeId: r.GetString(1)),
new { t = entry.TableName, pk = entry.PkJson }, ct);
var local = await sql.ReadLocalVersionAsync(entry.TableName, entry.PkJson, ct);
var remoteWins = local.Count == 0
|| entry.Hlc > local[0].Hlc
|| (entry.Hlc == local[0].Hlc && string.CompareOrdinal(entry.NodeId, local[0].NodeId) > 0);
var remoteWins = local is null
|| entry.Hlc > local.Value.Hlc
|| (entry.Hlc == local.Value.Hlc && string.CompareOrdinal(entry.NodeId, local.Value.NodeId) > 0);
if (!remoteWins)
{
@@ -76,32 +88,15 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
continue;
}
if (!sqlByTable.TryGetValue(entry.TableName, out var sql))
{
sql = BuildTableSql(db.ReplicatedTables[entry.TableName]);
sqlByTable[entry.TableName] = sql;
}
try
{
var table = db.ReplicatedTables[entry.TableName];
if (entry.IsTombstone)
await tx.ExecuteAsync(sql.Delete, new { pk = entry.PkJson }, ct);
await sql.DeleteAsync(table, entry.PkJson, ct);
else
await tx.ExecuteAsync(sql.Upsert, new { row = entry.RowJson }, ct);
await sql.UpsertAsync(table, entry.RowJson!, ct);
await tx.ExecuteAsync(
"INSERT OR REPLACE INTO __localdb_row_version " +
"(table_name, pk_json, hlc, node_id, is_tombstone, tombstone_utc) " +
"VALUES (@t, @pk, @hlc, @node, @tomb, @tut)",
new
{
t = entry.TableName,
pk = entry.PkJson,
hlc = entry.Hlc,
node = entry.NodeId,
tomb = entry.IsTombstone ? 1 : 0,
tut = entry.IsTombstone ? utcNow : null,
}, ct);
await sql.WriteRowVersionAsync(entry, utcNow, ct);
applied++;
}
catch (SqliteException ex)
@@ -110,59 +105,180 @@ 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).
await DeadLetterAsync(tx, entry, ex.Message, utcNow, ct);
await sql.DeadLetterAsync(entry, ex.Message, utcNow, ct);
deadLettered++;
}
}
await tx.ExecuteAsync(
"UPDATE __localdb_peer_state SET " +
"last_applied_remote_seq = MAX(last_applied_remote_seq, @maxSeq), " +
"last_seen_hlc = MAX(last_seen_hlc, @maxHlc), last_sync_utc = @utc WHERE id = 1",
new { maxSeq, maxHlc, utc = utcNow }, ct);
await sql.AdvancePeerStateAsync(maxSeq, maxHlc, utcNow, ct);
await sql.SetApplyingAsync(0, ct);
LastBatchStatementCount = sql.StatementCount;
await tx.ExecuteAsync("UPDATE __localdb_applying SET applying = 0 WHERE id = 1", null, ct);
await tx.CommitAsync(ct);
db.Clock.Observe(maxHlc);
return new ApplyResult(maxSeq, applied, discarded, deadLettered);
}
private static Task DeadLetterAsync(
ILocalDbTransaction tx, OplogEntryRecord entry, string error, string utcNow, CancellationToken ct) =>
tx.ExecuteAsync(
"INSERT INTO __localdb_dead_letter " +
"(received_utc, table_name, pk_json, row_json, hlc, node_id, error) " +
"VALUES (@utc, @t, @pk, @row, @hlc, @node, @err)",
new
{
utc = utcNow,
t = entry.TableName,
pk = entry.PkJson,
row = entry.RowJson,
hlc = entry.Hlc,
node = entry.NodeId,
err = error,
}, ct);
private static (string Delete, string Upsert) BuildTableSql(ReplicatedTable table)
/// <summary>
/// The statement set one apply-batch runs on: each distinct statement is built and prepared on
/// first use, then rebound and re-executed per row. Statements whose SQL depends on the table
/// (delete / upsert) are cached per table; the rest are shared across all of them.
/// </summary>
/// <remarks>
/// Reusing the same <see cref="SqliteCommand"/> with unchanged <c>CommandText</c> is what makes
/// this prepare-once: Microsoft.Data.Sqlite holds the compiled <c>sqlite3_stmt</c> on the command
/// and only recompiles when the text changes. <see cref="SqliteCommand.Prepare"/> is called
/// explicitly anyway, so the intent is stated rather than inferred from an implementation detail.
/// </remarks>
private sealed class PreparedStatements(SqliteConnection connection, SqliteTransaction transaction) : IDisposable
{
var name = Ident(table.Name);
private readonly List<SqliteCommand> _owned = [];
private readonly Dictionary<string, SqliteCommand> _deletes = new(StringComparer.Ordinal);
private readonly Dictionary<string, SqliteCommand> _upserts = new(StringComparer.Ordinal);
var pkMatch = string.Join(" AND ", table.PkColumns.Select(p => $"{Ident(p)} = json_extract(@pk, {JsonPath(p)})"));
var delete = $"DELETE FROM {name} WHERE {pkMatch}";
private SqliteCommand? _applying;
private SqliteCommand? _lookup;
private SqliteCommand? _rowVersion;
private SqliteCommand? _deadLetter;
private SqliteCommand? _peerState;
var columns = string.Join(", ", table.Columns.Select(c => Ident(c.Name)));
var values = string.Join(", ", table.Columns.Select(c => $"json_extract(@row, {JsonPath(c.Name)})"));
var upsert = $"INSERT OR REPLACE INTO {name} ({columns}) VALUES ({values})";
/// <summary>How many distinct statements this set has prepared.</summary>
public int StatementCount => _owned.Count;
return (delete, upsert);
public Task SetApplyingAsync(int value, CancellationToken ct)
{
var cmd = _applying ??= Create(
"UPDATE __localdb_applying SET applying = @applying WHERE id = 1", "@applying");
cmd.Parameters["@applying"].Value = value;
return cmd.ExecuteNonQueryAsync(ct);
}
public async Task<(long Hlc, string NodeId)?> ReadLocalVersionAsync(
string tableName, string pkJson, CancellationToken ct)
{
var cmd = _lookup ??= Create(
"SELECT hlc, node_id FROM __localdb_row_version WHERE table_name = @t AND pk_json = @pk",
"@t", "@pk");
cmd.Parameters["@t"].Value = tableName;
cmd.Parameters["@pk"].Value = pkJson;
await using var reader = await cmd.ExecuteReaderAsync(ct);
if (!await reader.ReadAsync(ct))
return null;
return (reader.GetInt64(0), reader.GetString(1));
}
public Task DeleteAsync(ReplicatedTable table, string pkJson, CancellationToken ct)
{
if (!_deletes.TryGetValue(table.Name, out var cmd))
_deletes[table.Name] = cmd = Create(BuildDelete(table), "@pk");
cmd.Parameters["@pk"].Value = pkJson;
return cmd.ExecuteNonQueryAsync(ct);
}
public Task UpsertAsync(ReplicatedTable table, string rowJson, CancellationToken ct)
{
if (!_upserts.TryGetValue(table.Name, out var cmd))
_upserts[table.Name] = cmd = Create(BuildUpsert(table), "@row");
cmd.Parameters["@row"].Value = rowJson;
return cmd.ExecuteNonQueryAsync(ct);
}
public Task WriteRowVersionAsync(OplogEntryRecord entry, string utcNow, CancellationToken ct)
{
var cmd = _rowVersion ??= Create(
"INSERT OR REPLACE INTO __localdb_row_version " +
"(table_name, pk_json, hlc, node_id, is_tombstone, tombstone_utc) " +
"VALUES (@t, @pk, @hlc, @node, @tomb, @tut)",
"@t", "@pk", "@hlc", "@node", "@tomb", "@tut");
cmd.Parameters["@t"].Value = entry.TableName;
cmd.Parameters["@pk"].Value = entry.PkJson;
cmd.Parameters["@hlc"].Value = entry.Hlc;
cmd.Parameters["@node"].Value = entry.NodeId;
cmd.Parameters["@tomb"].Value = entry.IsTombstone ? 1 : 0;
cmd.Parameters["@tut"].Value = entry.IsTombstone ? utcNow : DBNull.Value;
return cmd.ExecuteNonQueryAsync(ct);
}
public Task DeadLetterAsync(OplogEntryRecord entry, string error, string utcNow, CancellationToken ct)
{
var cmd = _deadLetter ??= Create(
"INSERT INTO __localdb_dead_letter " +
"(received_utc, table_name, pk_json, row_json, hlc, node_id, error) " +
"VALUES (@utc, @t, @pk, @row, @hlc, @node, @err)",
"@utc", "@t", "@pk", "@row", "@hlc", "@node", "@err");
cmd.Parameters["@utc"].Value = utcNow;
cmd.Parameters["@t"].Value = entry.TableName;
cmd.Parameters["@pk"].Value = entry.PkJson;
cmd.Parameters["@row"].Value = (object?)entry.RowJson ?? DBNull.Value;
cmd.Parameters["@hlc"].Value = entry.Hlc;
cmd.Parameters["@node"].Value = entry.NodeId;
cmd.Parameters["@err"].Value = error;
return cmd.ExecuteNonQueryAsync(ct);
}
public Task AdvancePeerStateAsync(long maxSeq, long maxHlc, string utcNow, CancellationToken ct)
{
var cmd = _peerState ??= Create(
"UPDATE __localdb_peer_state SET " +
"last_applied_remote_seq = MAX(last_applied_remote_seq, @maxSeq), " +
"last_seen_hlc = MAX(last_seen_hlc, @maxHlc), last_sync_utc = @utc WHERE id = 1",
"@maxSeq", "@maxHlc", "@utc");
cmd.Parameters["@maxSeq"].Value = maxSeq;
cmd.Parameters["@maxHlc"].Value = maxHlc;
cmd.Parameters["@utc"].Value = utcNow;
return cmd.ExecuteNonQueryAsync(ct);
}
private SqliteCommand Create(string sql, params string[] parameterNames)
{
var cmd = connection.CreateCommand();
try
{
cmd.Transaction = transaction;
cmd.CommandText = sql;
foreach (var name in parameterNames)
cmd.Parameters.Add(new SqliteParameter(name, DBNull.Value));
cmd.Prepare();
_owned.Add(cmd);
return cmd;
}
catch
{
cmd.Dispose();
throw;
}
}
public void Dispose()
{
foreach (var cmd in _owned)
cmd.Dispose();
_owned.Clear();
}
private static string BuildDelete(ReplicatedTable table)
{
var pkMatch = string.Join(
" AND ", table.PkColumns.Select(p => $"{Ident(p)} = json_extract(@pk, {JsonPath(p)})"));
return $"DELETE FROM {Ident(table.Name)} WHERE {pkMatch}";
}
private static string BuildUpsert(ReplicatedTable table)
{
var columns = string.Join(", ", table.Columns.Select(c => Ident(c.Name)));
var values = string.Join(", ", table.Columns.Select(c => $"json_extract(@row, {JsonPath(c.Name)})"));
return $"INSERT OR REPLACE INTO {Ident(table.Name)} ({columns}) VALUES ({values})";
}
// Local copy of TriggerSqlGenerator's convention (that helper lives in the core assembly).
private static string Ident(string name) => "\"" + name.Replace("\"", "\"\"") + "\"";
// A single-quoted JSON path with a quoted key member, so a column name with reserved
// characters still resolves: $."col". The key's double quotes are backslash-escaped per
// JSON path grammar.
private static string JsonPath(string column) => "'$.\"" + column.Replace("\"", "\\\"") + "\"'";
}
// Local copy of TriggerSqlGenerator's convention (that helper lives in the core assembly).
private static string Ident(string name) => "\"" + name.Replace("\"", "\"\"") + "\"";
// A single-quoted JSON path with a quoted key member, so a column name with reserved characters
// still resolves: $."col". The key's double quotes are backslash-escaped per JSON path grammar.
private static string JsonPath(string column) => "'$.\"" + column.Replace("\"", "\\\"") + "\"'";
}
ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs
+47 -10
View File
@@ -70,12 +70,42 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
"DELETE FROM __localdb_row_version WHERE is_tombstone = 1 AND tombstone_utc < @cutoff",
new { cutoff = FormatUtc(_utcNow() - options.TombstoneRetention) }, ct);
/// <summary>Counts the unacked backlog: oplog entries above the peer's last-acked watermark.</summary>
/// <summary>
/// The unacked backlog — oplog entries above the peer's last-acked watermark — in O(1).
/// </summary>
/// <remarks>
/// <para>
/// <b>Why not <c>COUNT(*)</c>.</b> The obvious query walks every unacked row, and this value is
/// read once a minute by maintenance AND on every metrics scrape. A node whose peer has been
/// down for a day therefore pays a full scan of a million-row backlog, repeatedly, precisely
/// when it is already under stress.
/// </para>
/// <para>
/// <b>The formula.</b> <c>MAX(seq) - MAX(last_acked_seq, MIN(seq) - 1)</c>, floored at 0. Both
/// MIN and MAX over <c>seq</c> are single b-tree seeks (it is the rowid, and SQLite's min/max
/// optimization turns each into one seek), and the peer-state read is a one-row lookup — so the
/// whole thing is three seeks regardless of backlog size.
/// </para>
/// <para>
/// <b>Why it is exact.</b> <c>seq</c> is AUTOINCREMENT, so retained rows are a contiguous run
/// wherever nothing has punched a hole in the middle — and no ordinary path punches holes: ack
/// pruning deletes <c>seq &lt;= last_acked_seq</c> (a prefix), the row cap deletes below
/// <c>MAX(seq) - keep + 1</c> (a prefix), and the age cap deletes below an HLC cutoff, which is
/// a prefix because HLCs are stamped in seq order. Clamping the floor to <c>MIN(seq) - 1</c> is
/// what keeps the answer exact after a prune has moved the horizon above the ack watermark;
/// without it a cap-pruned oplog would report its whole seq range forever and re-flag the cap on
/// every tick.
/// </para>
/// <para>
/// <b>The one inexact case</b> is <c>ILocalDb.DeregisterReplicated</c>, which deletes one
/// table's rows from wherever they sit and so can leave holes. The estimate then over-reports by
/// at most the number of holes, until the watermark advances past them. Over-reporting is the
/// safe direction: it can only trigger a snapshot resync earlier, never suppress one.
/// </para>
/// </remarks>
public async Task<long> GetOplogDepthAsync(CancellationToken ct = default)
{
var rows = await db.QueryAsync(
"SELECT COUNT(*) FROM __localdb_oplog WHERE seq > (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)",
static r => r.GetInt64(0), null, ct);
var rows = await db.QueryAsync(DepthSql, static r => r.GetInt64(0), null, ct);
return rows[0];
}
@@ -84,7 +114,7 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
/// metric-observation callbacks (gauge scrape / status poll); mirrors
/// <see cref="GetOplogDepthAsync"/>'s query. Deliberate tradeoff: synchronous I/O on the
/// metrics-collection thread is acceptable here because WAL-mode readers never block (or are
/// blocked by) writers, the COUNT is indexed-cheap, and the worst case is bounded by
/// blocked by) writers, the query is three b-tree seeks, and the worst case is bounded by
/// <c>BusyTimeoutMs</c> — all at scrape frequency, not on any data path. May throw if the db
/// is disposing — callers on the observation path swallow and drop the sample.
/// </summary>
@@ -92,11 +122,18 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
{
using var conn = db.CreateConnection();
using var cmd = conn.CreateCommand();
cmd.CommandText =
"SELECT COUNT(*) FROM __localdb_oplog WHERE seq > (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)";
cmd.CommandText = DepthSql;
return (long)cmd.ExecuteScalar()!;
}
// Both depth accessors run exactly this; keeping it in one place is what makes "mirrors the
// async query" a fact rather than a comment that can rot.
internal const string DepthSql =
"SELECT MAX(0, " +
"COALESCE((SELECT MAX(seq) FROM __localdb_oplog), 0) - " +
"MAX((SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1), " +
"COALESCE((SELECT MIN(seq) FROM __localdb_oplog), 1) - 1))";
/// <summary>
/// The highest seq this node has ever produced: the oplog's max, or — once acked rows have been
/// pruned away — the ack watermark that pruned them. Nothing we emit can exceed it, so a peer
@@ -133,9 +170,9 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
// Condition reads and prunes share the transaction so the decision cannot go stale (TOCTOU).
await using var tx = await db.BeginTransactionAsync(ct);
var depth = (await tx.QueryAsync(
"SELECT COUNT(*) FROM __localdb_oplog WHERE seq > (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)",
static r => r.GetInt64(0), null, ct))[0];
// Same O(1) expression as GetOplogDepthAsync — the caps check is the per-minute reader that
// made the old COUNT(*) expensive in the first place.
var depth = (await tx.QueryAsync(DepthSql, static r => r.GetInt64(0), null, ct))[0];
var overRows = depth > options.MaxOplogRows;
var oldest = await tx.QueryAsync(
ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SnapshotStreamer.cs
+36 -4
View File
@@ -57,7 +57,13 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
break;
lastPk = rows[^1].PkJson;
// One DB page may span several messages: the page is bounded by ROW COUNT, but a
// message is bounded by BYTES (ReplicationOptions.MaxBatchBytes) — the same budget
// the delta pump honours, and for the same reason. Wide rows are exactly why a
// snapshot is the easiest way to exceed the transport's message limit: it ships
// whole rows, all of them, back to back.
var batch = new SnapshotBatch { TableName = table.Name };
var bytes = 0L;
foreach (var row in rows)
{
if (!row.IsTombstone && row.RowJson is null)
@@ -79,11 +85,25 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
};
if (row.RowJson is not null)
snapshotRow.RowJson = row.RowJson;
var size = RowSize(snapshotRow);
// Flush before adding, never after: a message keeps at least one row however
// large, so an oversized row goes out alone rather than stalling the snapshot.
if (batch.Rows.Count > 0 && bytes + size > options.MaxBatchBytes)
{
await send(new SyncMessage { SnapshotBatch = batch }, ct);
totalRows += batch.Rows.Count;
batch = new SnapshotBatch { TableName = table.Name };
bytes = 0;
}
batch.Rows.Add(snapshotRow);
bytes += size;
}
if (batch.Rows.Count > 0)
{
await send(new SyncMessage { SnapshotBatch = batch }, ct);
totalRows += batch.Rows.Count;
totalRows += batch.Rows.Count;
}
if (rows.Count < options.MaxBatchSize)
break;
@@ -104,11 +124,23 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
return asOfSeq;
}
/// <summary>
/// Serialized bytes one row contributes to its batch message: the row's own encoding plus the
/// repeated field's tag and length prefix, as protobuf itself computes them.
/// </summary>
private static long RowSize(SnapshotRow row) =>
1 + Google.Protobuf.CodedOutputStream.ComputeMessageSize(row);
// Pages one table's rows by joining the row-version ledger to the live table on the PK columns
// extracted from pk_json. Rows present in the live table but absent from __localdb_row_version
// (data written BEFORE the table was registered for replication) are intentionally NOT snapshotted:
// they were never captured, consistent with CDC semantics — a full re-registration is required to
// baseline them.
// (data written BEFORE the table was registered for replication) are NOT snapshotted: they were
// never captured, consistent with CDC semantics.
//
// That is a real hole for a node that starts replicating LATE — its whole pre-existing dataset is
// invisible to the peer, silently and permanently — and the supported way to close it is to
// register with `ILocalDb.RegisterReplicated(table, baselineExistingRows: true)`, which seeds the
// ledger at an LWW-safe floor and flags the resync that carries the rows here. Nothing changes on
// this side: baselined rows are ordinary ledger rows and page like any other.
private static string BuildPageSql(ReplicatedTable table)
{
var name = Ident(table.Name);
ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSession.cs
+191 -33
View File
@@ -10,11 +10,13 @@ namespace ZB.MOM.WW.LocalDb.Replication.Internal;
/// <summary>
/// The symmetric 2-node sync protocol as one state machine: both the gRPC service and the gRPC
/// client run this identical logic over a <see cref="SyncDuplex"/>. It performs the fail-closed
/// handshake, then runs a bidirectional steady state — an outbound change pump and an inbound
/// apply/ack loop — until the token is cancelled or the stream ends. All outbound traffic flows
/// through a single writer task fed by two lanes: an unbounded CONTROL lane (handshake messages,
/// acks) and a bounded DATA lane (delta/snapshot batches). Receiving therefore never blocks on the
/// ability to write — acks always fit — while the bounded data lane is the pump's backpressure.
/// handshake, then runs a bidirectional steady state until the token is cancelled or the stream
/// ends: an outbound phase (the full snapshot this node owes, if any, then the change pump) and an
/// inbound apply/ack loop, STARTED TOGETHER so the inbox keeps draining even while a snapshot is
/// being sent. All outbound traffic flows through a single writer task fed by two lanes: an
/// unbounded CONTROL lane (handshake messages, acks) and a bounded DATA lane (delta/snapshot
/// batches). Receiving therefore never blocks on the ability to write — acks always fit — while the
/// bounded data lane is the pump's backpressure.
/// Local writes never block on the peer; the first fault tears down all loops and propagates.
/// </summary>
internal sealed class SyncSession
@@ -97,10 +99,19 @@ internal sealed class SyncSession
var writer = WriterLoopAsync(duplex, control.Reader, data.Reader, linked.Token);
try
{
await HandshakeAsync(duplex, control.Writer, GatedDataSend, linked.Token);
var snapshotRequired = await HandshakeAsync(duplex, control.Writer, linked.Token);
var pump = PumpLoopAsync(data.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
// inbox undrained for its whole duration, so a MUTUAL snapshot (both sides flagged —
// e.g. both baselined an existing dataset) required an unbounded inbox to avoid
// deadlock, and therefore buffered a whole peer dataset in memory. With the receive loop
// running throughout, the peer's snapshot is applied as it arrives, the inbox can be
// bounded (see GrpcSyncDuplex.InboxCapacity), and mutual snapshots exchange under
// 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 loops = new[] { pump, receive, writer };
// The first COMPLETED task is the primary outcome. Cancel the siblings, observe every
@@ -123,9 +134,14 @@ internal sealed class SyncSession
}
}
private async Task HandshakeAsync(
SyncDuplex duplex, ChannelWriter<SyncMessage> controlLane,
Func<SyncMessage, CancellationToken, Task> dataSend, CancellationToken ct)
/// <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"/>.
/// </summary>
private async Task<bool> HandshakeAsync(
SyncDuplex duplex, ChannelWriter<SyncMessage> controlLane, CancellationToken ct)
{
var peerState = await _store.GetPeerStateAsync(ct);
var localHandshake = BuildHandshake(peerState);
@@ -184,13 +200,25 @@ internal sealed class SyncSession
throw new InvalidOperationException(
$"Replication protocol error: expected a HandshakeAck, got {peerAckMsg.MsgCase}.");
// We stream the snapshot when WE owe the peer one (computed from our pruned horizon /
// needs_snapshot), then it lands on the peer's receive loop. The peer's own flag drives ITS
// send direction symmetrically; both are safe to run because LWW merges either way.
// Memory profile of a MUTUAL snapshot: while we send here (receive loop not yet running),
// the peer's inbound frames — potentially its entire snapshot — buffer in the unbounded
// duplex inbox until our send completes. Worst case one full peer snapshot held in memory;
// acceptable for the small LocalDb datasets this library targets.
return snapshotRequired;
}
/// <summary>
/// The outbound steady state: stream the full snapshot this node owes the peer (if any), then
/// run the incremental delta pump. One task, because the ordering between them is a protocol
/// requirement — the pump must not put deltas on the data lane ahead of the snapshot, and it
/// starts from the as-of seq the snapshot covered.
/// </summary>
/// <remarks>
/// Split out of the handshake so the receive loop is already draining the inbox while this
/// streams (see <see cref="RunAsync"/>). Nothing else about the snapshot changes: it goes
/// through the same gated data lane, so it is still bounded by the writer's flow control rather
/// than racing the transport.
/// </remarks>
private async Task SnapshotThenPumpLoopAsync(
bool snapshotRequired, Func<SyncMessage, CancellationToken, Task> dataSend,
ChannelWriter<SyncMessage> dataLane, CancellationToken ct)
{
if (snapshotRequired)
{
if (SnapshotSender is null)
@@ -203,6 +231,8 @@ internal sealed class SyncSession
if (snapshotAsOfSeq > _sentThruSeq)
_sentThruSeq = snapshotAsOfSeq;
}
await PumpLoopAsync(dataLane, ct);
}
// The ONLY caller of duplex.Send. Control (handshake + acks) always preempts data so the peer's
@@ -246,6 +276,17 @@ internal sealed class SyncSession
}
}
/// <summary>
/// Reads oplog deltas above the sent watermark and ships them, splitting each read into as many
/// messages as <see cref="ReplicationOptions.MaxBatchBytes"/> requires.
/// </summary>
/// <remarks>
/// Row count alone cannot bound a message: rows are arbitrarily wide, so any row-count cap is
/// really a guess about the widest row, and guessing wrong produces a message the transport
/// rejects — replication then stops converging with a size error, not a data error. Budgeting by
/// summed serialized bytes bounds the thing that is actually limited, and lets the row cap go
/// back to meaning what it says.
/// </remarks>
private async Task PumpLoopAsync(ChannelWriter<SyncMessage> dataLane, CancellationToken ct)
{
while (true)
@@ -254,12 +295,30 @@ internal sealed class SyncSession
var batch = await _store.ReadBatchAboveAsync(_sentThruSeq, _options.MaxBatchSize, ct);
if (batch.Count > 0)
{
var delta = new DeltaBatch();
foreach (var entry in batch)
delta.Entries.Add(ToProto(entry));
// Blocking here on a full data lane IS the pump's backpressure.
await dataLane.WriteAsync(new SyncMessage { DeltaBatch = delta }, ct);
_sentThruSeq = batch[^1].Seq;
var index = 0;
while (index < batch.Count)
{
var delta = new DeltaBatch();
var bytes = 0L;
while (index < batch.Count)
{
var proto = ToProto(batch[index]);
var size = EntrySize(proto);
// The first entry of a message is taken however large it is: refusing an
// oversized row would stall the stream on it permanently, whereas sending it
// alone either succeeds or fails loudly at the transport with its real size.
if (delta.Entries.Count > 0 && bytes + size > _options.MaxBatchBytes)
break;
delta.Entries.Add(proto);
bytes += size;
index++;
}
// Blocking here on a full data lane IS the pump's backpressure.
await dataLane.WriteAsync(new SyncMessage { DeltaBatch = delta }, ct);
// Advanced per MESSAGE, not per read: a fault mid-split must not re-send what
// already went out, and must not skip what did not.
_sentThruSeq = batch[index - 1].Seq;
}
continue;
}
@@ -267,23 +326,43 @@ internal sealed class SyncSession
}
}
/// <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.
/// </summary>
private static long EntrySize(OplogEntry entry) =>
1 + Google.Protobuf.CodedOutputStream.ComputeMessageSize(entry);
private async Task ReceiveLoopAsync(
SyncDuplex duplex, ChannelWriter<SyncMessage> controlLane, CancellationToken ct)
{
while (await duplex.Inbox.WaitToReadAsync(ct))
// One-message lookahead. Coalescing means reading ahead until a message turns out NOT to be
// of the kind being coalesced, and a Channel cannot un-read — so that message is parked here
// and processed first on the next pass. Its presence also keeps the outer loop from waiting
// on the inbox when there is already work in hand.
SyncMessage? lookahead = null;
while (lookahead is not null || await duplex.Inbox.WaitToReadAsync(ct))
{
while (duplex.Inbox.TryRead(out var msg))
while (true)
{
var msg = lookahead;
lookahead = null;
if (msg is null)
{
if (!duplex.Inbox.TryRead(out var read))
break;
msg = read;
}
switch (msg.MsgCase)
{
case SyncMessage.MsgOneofCase.DeltaBatch:
// Deltas apply normally even mid-snapshot: LWW makes the interleave safe.
await HandleDeltaBatchAsync(controlLane, msg.DeltaBatch, ct);
lookahead = await HandleCoalescedDeltasAsync(duplex, controlLane, msg.DeltaBatch, ct);
break;
case SyncMessage.MsgOneofCase.DeltaAck:
var acked = msg.DeltaAck.AppliedThruSeq;
InterlockedMax(ref _peerAckedSeq, acked);
await _store.RecordPeerAckAsync(acked, ct);
lookahead = await HandleCoalescedAcksAsync(duplex, msg.DeltaAck, ct);
break;
case SyncMessage.MsgOneofCase.SnapshotBegin:
if (SnapshotApplier is null)
@@ -314,19 +393,98 @@ internal sealed class SyncSession
}
}
private async Task HandleDeltaBatchAsync(ChannelWriter<SyncMessage> controlLane, DeltaBatch batch, CancellationToken ct)
/// <summary>
/// Applies the given delta batch together with any delta batches ALREADY waiting in the inbox,
/// as one transaction answered by one ack; returns the non-delta message that ended the run (to
/// be processed next), or null if the inbox merely ran dry.
/// </summary>
/// <remarks>
/// <para>
/// Byte-budgeting the sender produces more, smaller messages, and one apply transaction plus one
/// ack round-trip per message is the cost that would otherwise scale with that. Coalescing pays
/// it once per burst instead — and a burst is exactly what a reconnect after an outage delivers.
/// </para>
/// <para>
/// Only messages already queued are absorbed (<c>TryRead</c>, never a wait), so this adds no
/// latency: a trickle of one batch at a time coalesces nothing and behaves exactly as before.
/// The run stops at the sender's own caps, so the extra memory held is bounded by roughly one
/// sender batch beyond the budget — the overshoot of finishing whichever batch crosses it.
/// </para>
/// </remarks>
private async Task<SyncMessage?> HandleCoalescedDeltasAsync(
SyncDuplex duplex, ChannelWriter<SyncMessage> controlLane, DeltaBatch first, CancellationToken ct)
{
if (batch.Entries.Count == 0)
var entries = new List<OplogEntry>(first.Entries);
var bytes = 0L;
foreach (var entry in first.Entries)
bytes += EntrySize(entry);
SyncMessage? lookahead = null;
while (entries.Count < _options.MaxBatchSize && bytes < _options.MaxBatchBytes
&& duplex.Inbox.TryRead(out var next))
{
if (next.MsgCase != SyncMessage.MsgOneofCase.DeltaBatch)
{
lookahead = next;
break;
}
foreach (var entry in next.DeltaBatch.Entries)
{
entries.Add(entry);
bytes += EntrySize(entry);
}
}
await ApplyDeltaEntriesAsync(controlLane, entries, ct);
return lookahead;
}
/// <summary>
/// Records the given ack together with any acks ALREADY waiting in the inbox, as a single
/// watermark write; returns the non-ack message that ended the run, or null if the inbox ran dry.
/// </summary>
/// <remarks>
/// Exactly equivalent to recording each in turn, and strictly cheaper: <c>applied_thru_seq</c> is
/// monotonic and <c>RecordPeerAckAsync</c> is MAX-guarded, so every superseded ack would have
/// written a watermark that the final one overwrites and run a DELETE below its own horizon.
/// Unbounded coalescing is safe here — acks collapse into one number, accumulating nothing.
/// </remarks>
private async Task<SyncMessage?> HandleCoalescedAcksAsync(
SyncDuplex duplex, DeltaAck first, CancellationToken ct)
{
var acked = first.AppliedThruSeq;
SyncMessage? lookahead = null;
while (duplex.Inbox.TryRead(out var next))
{
if (next.MsgCase != SyncMessage.MsgOneofCase.DeltaAck)
{
lookahead = next;
break;
}
if (next.DeltaAck.AppliedThruSeq > acked)
acked = next.DeltaAck.AppliedThruSeq;
}
InterlockedMax(ref _peerAckedSeq, acked);
await _store.RecordPeerAckAsync(acked, ct);
return lookahead;
}
private async Task ApplyDeltaEntriesAsync(
ChannelWriter<SyncMessage> controlLane, IReadOnlyList<OplogEntry> batchEntries, CancellationToken ct)
{
if (batchEntries.Count == 0)
return;
var now = _utcNow();
var nowMs = now.ToUnixTimeMilliseconds();
var driftLimitMs = nowMs + (long)_options.MaxHlcDriftAhead.TotalMilliseconds;
var toApply = new List<OplogEntryRecord>(batch.Entries.Count);
var toApply = new List<OplogEntryRecord>(batchEntries.Count);
var maxSeq = 0L;
var driftDeadLettered = 0;
foreach (var proto in batch.Entries)
foreach (var proto in batchEntries)
{
if (proto.Seq > maxSeq) maxSeq = proto.Seq;
var entry = FromProto(proto);
ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptions.cs
+36 -1
View File
@@ -14,9 +14,44 @@ public sealed class ReplicationOptions
/// <summary>How often the change pump flushes pending oplog deltas to the peer.</summary>
public TimeSpan FlushInterval { get; set; } = TimeSpan.FromMilliseconds(250);
/// <summary>Maximum oplog entries sent per outbound batch.</summary>
/// <summary>
/// Secondary, row-count cap on an outbound batch (and the snapshot streamer's DB page size).
/// <see cref="MaxBatchBytes"/> is the primary cap.
/// </summary>
/// <remarks>
/// Historically this was the ONLY cap, which made it a proxy for a byte budget that consumers
/// had to tune by hand against their widest row — ScadaBridge pinned it to 16 because a 70 KB
/// <c>config_json</c> times the 500 default is ~35 MB against a 4 MB gRPC limit. With
/// <see cref="MaxBatchBytes"/> doing that job directly, this is back to being what its name
/// says: a bound on how many rows one message may carry.
/// </remarks>
public int MaxBatchSize { get; set; } = 500;
/// <summary>
/// Primary cap on an outbound batch: the summed serialized size of the entries in one
/// <c>DeltaBatch</c> / <c>SnapshotBatch</c> message, in bytes. Default 2 MB.
/// </summary>
/// <remarks>
/// <para>
/// Sized at half the default 4 MB gRPC receive limit, so the message framing, the oneof wrapper
/// and any per-entry overhead the estimate does not model all fit inside the margin.
/// </para>
/// <para>
/// <b>A single row larger than the budget is still sent, alone.</b> Skipping it would stall the
/// stream permanently on one oversized row; sending it gives the transport the chance to accept
/// it (gRPC limits are configurable) and, if it refuses, a loud transport error naming the size
/// rather than silent non-convergence.
/// </para>
/// <para>
/// <b>Wire compatibility:</b> none required. Batch size is a sender-side choice the protocol
/// already leaves free — <c>DeltaBatch</c>/<c>SnapshotBatch</c> carry a repeated field with no
/// declared count — so a node on this version interoperates with a peer on any earlier one in
/// both directions, with no proto change and no <c>lib_schema_version</c> bump. (ScadaBridge
/// site pairs stop and start together regardless; this simply does not depend on it.)
/// </para>
/// </remarks>
public int MaxBatchBytes { get; set; } = 2 * 1024 * 1024;
/// <summary>Backlog ceiling: exceeding it flags a snapshot resync and prunes the oldest oplog rows.</summary>
public long MaxOplogRows { get; set; } = 1_000_000;
ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptionsValidator.cs
+2
View File
@@ -28,6 +28,8 @@ public sealed class ReplicationOptionsValidator : IValidateOptions<ReplicationOp
failures.Add($"LocalDb:Replication:FlushInterval must be greater than zero; got {options.FlushInterval}.");
if (options.MaxBatchSize <= 0)
failures.Add($"LocalDb:Replication:MaxBatchSize must be greater than 0; got {options.MaxBatchSize}.");
if (options.MaxBatchBytes <= 0)
failures.Add($"LocalDb:Replication:MaxBatchBytes must be greater than 0; got {options.MaxBatchBytes}.");
if (options.MaxOplogRows <= 0)
failures.Add($"LocalDb:Replication:MaxOplogRows must be greater than 0; got {options.MaxOplogRows}.");
if (options.MaxOplogAge <= TimeSpan.Zero)
ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs
+99
View File
@@ -42,6 +42,105 @@ public interface ILocalDb
/// </summary>
ReplicatedTable RegisterReplicated(string tableName);
/// <summary>
/// As <see cref="RegisterReplicated(string)"/>, additionally baselining rows that already exist
/// in the table when <paramref name="baselineExistingRows"/> is true.
/// </summary>
/// <param name="tableName">The table to opt into replication.</param>
/// <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.
/// </param>
/// <remarks>
/// <para>
/// <b>The problem this solves.</b> Capture is change-data-capture: without baselining, a row
/// written before registration exists in neither the oplog nor the row-version ledger, and the
/// snapshot streamer pages from the LEDGER — so a node that starts replicating LATE never sends
/// its pre-existing rows, silently and forever. Opting in seeds the ledger so a snapshot can
/// carry them.
/// </para>
/// <para>
/// <b>Why the floor is HLC 0, and why that is LWW-safe.</b> A seeded version must never beat a
/// genuine write. Every genuine HLC is a UTC millisecond shifted left 16 bits, so every genuine
/// HLC is greater than 0: a baselined row therefore LOSES every comparison against a real remote
/// write of the same key, and wins only where the peer has no version of that key at all
/// (which is precisely the case baselining exists to fix). It cannot perturb the local clock
/// either — clock recovery takes a MAX over existing HLCs, and 0 raises nothing.
/// </para>
/// <para>
/// <b>The one genuinely ambiguous case</b> is both nodes baselining the SAME key (e.g. both
/// restored the same legacy file). Both sides hold HLC 0, so the LWW tie-break falls to the
/// node-id ordinal: deterministic and convergent, but arbitrary as to content. Nothing can do
/// better — two rows of unknown, unordered antiquity carry no information about which is newer.
/// </para>
/// <para>
/// <b>Idempotent.</b> Seeding uses <c>ON CONFLICT DO NOTHING</c>, so a row that already has a
/// ledger entry — captured normally, or seeded on an earlier boot — keeps its genuine version
/// 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.
/// </para>
/// </remarks>
ReplicatedTable RegisterReplicated(string tableName, bool baselineExistingRows);
/// <summary>
/// Removes a table from replication: drops its three capture triggers and prunes its rows from
/// <c>__localdb_oplog</c> and <c>__localdb_row_version</c>, in one transaction. Idempotent, and
/// deliberately usable on a table this process never registered — cleaning up triggers left
/// behind by an EARLIER build is the primary use.
/// </summary>
/// <param name="tableName">The table to stop capturing. The table's own data is never touched.</param>
/// <returns>True if anything was removed (a trigger dropped or a row pruned); false if there was nothing to clean up.</returns>
/// <remarks>
/// <para>
/// <b>Intended caller:</b> an application that has decided this node no longer replicates (or no
/// longer replicates this table). Capture is not free — the three triggers run inside every
/// consumer write transaction and append to an oplog nothing will ever drain — so a node with no
/// peer should not be paying for it.
/// </para>
/// <para>
/// <b>Pruning the row-version ledger is safe with respect to re-registration</b>, because
/// <see cref="RegisterReplicated(string, bool)"/> can re-baseline. Without baselining, a
/// deregister-then-register cycle would leave the previously-known rows unstreamable — the same
/// late-opt-in hole — so the two APIs are designed as a pair.
/// </para>
/// <para>
/// <b>Deregistration must be symmetric across a replicating pair, and the library enforces it
/// loudly.</b> The sync handshake compares the two nodes' registered-table digests fail-closed,
/// so a node that drops a table its peer still replicates simply stops syncing — with a
/// schema-mismatch error, not silent divergence. Deregister on both nodes, or on neither.
/// </para>
/// <para>
/// <b>The HLC's durable crash anchor is flushed as part of the same transaction</b>, before the
/// prune. The clock recovers on open as the MAX of <c>hlc_high_water</c>, the oplog's HLCs, the
/// row-version ledger's HLCs and the last-seen peer HLC — and <c>hlc_high_water</c> is otherwise
/// written only on graceful disposal. Since this API deletes the deregistered table's oplog and
/// ledger rows, a file with no other registered table would, after an UNGRACEFUL exit (crash,
/// SIGKILL, power loss), have nothing left to recover from: the clock would restart from
/// wall-clock time and could re-issue stamps below HLCs peers already hold, which LWW then
/// discards as older. Flushing here removes that window.
/// </para>
/// <para>
/// <b>Backlog-depth residual.</b> Pruning one table's rows can leave holes in the oplog's seq
/// range, and the O(1) depth estimate assumes the retained range is contiguous, so it can
/// over-report until the peer acks past the holes. Over-reporting is conservative — it can only
/// trigger an earlier snapshot resync, never suppress one — and on the node this API is for
/// (no peer at all) depth is moot.
/// </para>
/// </remarks>
bool DeregisterReplicated(string tableName);
/// <summary>The tables currently opted into replication, keyed by table name.</summary>
IReadOnlyDictionary<string, ReplicatedTable> ReplicatedTables { get; }
}
ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/LocalDbSchema.cs
+86 -4
View File
@@ -1,3 +1,4 @@
using System.Globalization;
using Microsoft.Data.Sqlite;
namespace ZB.MOM.WW.LocalDb.Internal;
@@ -7,6 +8,29 @@ namespace ZB.MOM.WW.LocalDb.Internal;
/// </summary>
internal static class LocalDbSchema
{
/// <summary>
/// The on-disk bookkeeping-schema version this build creates and upgrades to. Bumping it means
/// adding a case to <see cref="Upgrade"/>; existing files are upgraded in place on open.
/// </summary>
/// <remarks>
/// Deliberately NOT the same number as <c>SyncSession.LibSchemaVersion</c>, which is the WIRE
/// protocol version and is compared fail-closed against the peer's. This one is local-file-only:
/// two nodes may briefly sit on different file-schema versions during an upgrade and still sync,
/// because nothing about the file layout crosses the wire.
/// <list type="bullet">
/// <item><description>
/// <b>1</b> — the original layout, which also created <c>__localdb_oplog_hlc</c>.
/// </description></item>
/// <item><description>
/// <b>2</b> — drops <c>__localdb_oplog_hlc</c>. That index (<c>table_name, pk_json, hlc</c>) was
/// never read by any query: every oplog access path is by <c>seq</c> (the rowid) or a full
/// <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>
/// </list>
/// </remarks>
internal const long CurrentSchemaVersion = 2;
private const string Ddl = """
CREATE TABLE IF NOT EXISTS __localdb_meta (
id INTEGER PRIMARY KEY CHECK (id = 1),
@@ -25,7 +49,6 @@ internal static class LocalDbSchema
node_id TEXT NOT NULL,
is_tombstone INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS __localdb_oplog_hlc ON __localdb_oplog(table_name, pk_json, hlc);
CREATE TABLE IF NOT EXISTS __localdb_peer_state (
id INTEGER PRIMARY KEY CHECK (id = 1),
peer_node_id TEXT NULL,
@@ -80,14 +103,61 @@ internal static class LocalDbSchema
meta.Transaction = transaction;
meta.CommandText =
"INSERT INTO __localdb_meta (id, node_id, schema_version) " +
"SELECT 1, $nodeId, 1 WHERE NOT EXISTS (SELECT 1 FROM __localdb_meta)";
"SELECT 1, $nodeId, $version WHERE NOT EXISTS (SELECT 1 FROM __localdb_meta)";
meta.Parameters.AddWithValue("$nodeId", Guid.NewGuid().ToString("D"));
meta.Parameters.AddWithValue("$version", CurrentSchemaVersion);
meta.ExecuteNonQuery();
}
Upgrade(connection, transaction);
transaction.Commit();
}
/// <summary>
/// Brings a file created by an OLDER build up to <see cref="CurrentSchemaVersion"/>, in the
/// caller's transaction. A file already at the current version is untouched.
/// </summary>
/// <remarks>
/// Each step is written to be safe to re-run (IF EXISTS / IF NOT EXISTS), so a crash between the
/// DDL and the version stamp costs nothing but a repeat on the next open. Steps run in ascending
/// order and the stamp is written once at the end, inside the same transaction as the DDL — a
/// file is therefore never observed at the new version with an unapplied step.
/// </remarks>
private static void Upgrade(SqliteConnection connection, SqliteTransaction transaction)
{
long current;
using (var read = connection.CreateCommand())
{
read.Transaction = transaction;
read.CommandText = "SELECT schema_version FROM __localdb_meta WHERE id = 1";
current = Convert.ToInt64(read.ExecuteScalar(), CultureInfo.InvariantCulture);
}
if (current >= CurrentSchemaVersion)
return;
var steps = new List<string>();
// 1 -> 2: drop the write-amplifying, never-read oplog index (see CurrentSchemaVersion).
if (current < 2)
steps.Add("DROP INDEX IF EXISTS __localdb_oplog_hlc;");
if (steps.Count > 0)
{
using var ddl = connection.CreateCommand();
ddl.Transaction = transaction;
ddl.CommandText = string.Concat(steps);
ddl.ExecuteNonQuery();
}
using var stamp = connection.CreateCommand();
stamp.Transaction = transaction;
stamp.CommandText = "UPDATE __localdb_meta SET schema_version = $version WHERE id = 1";
stamp.Parameters.AddWithValue("$version", CurrentSchemaVersion);
stamp.ExecuteNonQuery();
}
internal static (string NodeId, long HlcHighWater) ReadMeta(SqliteConnection connection)
{
using var cmd = connection.CreateCommand();
@@ -99,10 +169,22 @@ internal static class LocalDbSchema
return (reader.GetString(0), reader.GetInt64(1));
}
internal static void FlushHlcHighWater(SqliteConnection connection, long value)
/// <summary>
/// Writes the clock's durable crash anchor. <paramref name="transaction"/> must be supplied when
/// the connection has one open (SQLite commands are rejected otherwise), which is what lets a
/// caller flush the anchor atomically with a change that would otherwise erase the clock's other
/// recovery evidence — see <see cref="ILocalDb.DeregisterReplicated"/>.
/// </summary>
/// <remarks>
/// MAX-guarded rather than an assignment: every writer of this column is flushing a monotonic
/// clock, so a lower value can only be a stale one, and the anchor exists precisely to never
/// regress.
/// </remarks>
internal static void FlushHlcHighWater(SqliteConnection connection, long value, SqliteTransaction? transaction = null)
{
using var cmd = connection.CreateCommand();
cmd.CommandText = "UPDATE __localdb_meta SET hlc_high_water = $value WHERE id = 1";
cmd.Transaction = transaction;
cmd.CommandText = "UPDATE __localdb_meta SET hlc_high_water = MAX(hlc_high_water, $value) WHERE id = 1";
cmd.Parameters.AddWithValue("$value", value);
cmd.ExecuteNonQuery();
}
ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs
+143 -3
View File
@@ -138,7 +138,15 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
return await ReadAllAsync(cmd, map, ct);
}
public async Task<ILocalDbTransaction> BeginTransactionAsync(CancellationToken ct = default)
public async Task<ILocalDbTransaction> BeginTransactionAsync(CancellationToken ct = default) =>
await BeginPreparedTransactionAsync(ct);
/// <summary>
/// As <see cref="BeginTransactionAsync"/>, but typed so internal callers reach the underlying
/// connection and transaction — which is what lets a hot loop hold <see cref="SqliteCommand"/>s
/// across many rows and rebind them, instead of re-preparing statements per statement per row.
/// </summary>
internal async Task<Transaction> BeginPreparedTransactionAsync(CancellationToken ct = default)
{
var conn = await CreateConnectionAsync(ct);
try
@@ -155,8 +163,11 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
public IReadOnlyDictionary<string, ReplicatedTable> ReplicatedTables => _replicatedTables;
public ReplicatedTable RegisterReplicated(string tableName) =>
RegisterReplicated(tableName, baselineExistingRows: false);
// Thread-safe: all master-connection work runs under _masterLock.
public ReplicatedTable RegisterReplicated(string tableName)
public ReplicatedTable RegisterReplicated(string tableName, bool baselineExistingRows)
{
if (string.IsNullOrWhiteSpace(tableName) || !TableNameRegex.IsMatch(tableName))
throw new LocalDbRegistrationException(
@@ -201,6 +212,10 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
cmd.CommandText = script;
cmd.ExecuteNonQuery();
}
if (baselineExistingRows)
BaselineExistingRows(tx, table);
tx.Commit();
}
@@ -209,6 +224,125 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
}
}
// Thread-safe: all master-connection work runs under _masterLock.
public bool DeregisterReplicated(string tableName)
{
if (string.IsNullOrWhiteSpace(tableName) || !TableNameRegex.IsMatch(tableName))
throw new LocalDbRegistrationException(
$"Cannot deregister table '{tableName}': name must be a plain identifier matching {TableNameRegex}.");
lock (_masterLock)
{
using var tx = _master.BeginTransaction();
// Counted BEFORE the drop: DROP TRIGGER reports no row count, and "did anything change?"
// is the caller's cue to log a one-time cleanup rather than a no-op every boot.
long triggersPresent;
using (var probe = _master.CreateCommand())
{
probe.Transaction = tx;
probe.CommandText =
"SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger' AND name IN ($ai, $au, $ad)";
foreach (var suffix in TriggerSqlGenerator.Suffixes)
probe.Parameters.AddWithValue("$" + suffix, TriggerSqlGenerator.TriggerName(tableName, suffix));
triggersPresent = Convert.ToInt64(probe.ExecuteScalar(), CultureInfo.InvariantCulture);
}
using (var drop = _master.CreateCommand())
{
drop.Transaction = tx;
drop.CommandText = TriggerSqlGenerator.BuildUninstallScript(tableName);
drop.ExecuteNonQuery();
}
// Flushed BEFORE the prune below, in the SAME transaction, because that prune destroys
// the clock's crash evidence. Recovery on open takes a MAX over hlc_high_water, the
// oplog's HLCs, the row-version ledger's HLCs and last_seen_hlc — and hlc_high_water is
// otherwise written only by Dispose. Deregistering a table deletes that table's oplog
// AND ledger rows, so on a file whose registered tables are all deregistered, an
// UNGRACEFUL exit (crash, SIGKILL, power loss — no Dispose) afterwards leaves nothing
// above zero to recover from: the reopened clock would restart at wall-clock time and
// could re-issue HLCs below stamps peers have already observed, which LWW reads as
// "older" and discards. One row-level UPDATE on an administrative one-shot buys the
// anchor back; it is a separate command from the prune so it cannot inflate prunedRows.
LocalDbSchema.FlushHlcHighWater(_master, _clock.Current, tx);
// Full scan of the oplog by table_name — the index that would have served it is exactly
// the one schema v2 drops, and paying a write on every capture to speed up a one-shot
// administrative prune is the wrong trade.
int prunedRows;
using (var prune = _master.CreateCommand())
{
prune.Transaction = tx;
prune.CommandText =
"DELETE FROM __localdb_oplog WHERE table_name = $t;" +
"DELETE FROM __localdb_row_version WHERE table_name = $t;";
prune.Parameters.AddWithValue("$t", tableName);
prunedRows = prune.ExecuteNonQuery();
}
tx.Commit();
_replicatedTables.TryRemove(tableName, out _);
return triggersPresent > 0 || prunedRows > 0;
}
}
/// <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
/// anything. Runs in the registration transaction so triggers and baseline commit together.
/// </summary>
/// <remarks>
/// <para>
/// The floor is HLC <c>0</c> — see <see cref="ILocalDb.RegisterReplicated(string, bool)"/> for
/// why that is the only LWW-safe choice.
/// </para>
/// <para>
/// <b>The <c>needs_snapshot</c> flag is what makes this actually deliver.</b> Seeding the ledger
/// alone changes nothing on the wire: a node whose oplog starts at seq 1 presents no gap at the
/// handshake, so the peer is told it needs no snapshot and only ever receives post-registration
/// deltas. Flagging the resync is what turns the seeded ledger into rows the peer receives.
/// Deliberately NOT done by writing oplog entries instead: that would dump the entire table into
/// the oplog on the first boot after opting in, likely blowing <c>MaxOplogRows</c> and forcing
/// 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.
/// </para>
/// </remarks>
private void BaselineExistingRows(SqliteTransaction tx, ReplicatedTable table)
{
int seeded;
using (var cmd = _master.CreateCommand())
{
cmd.Transaction = tx;
// The trailing WHERE 1 disambiguates the upsert parse on a SELECT source — the same
// reason TriggerSqlGenerator's row-version statements carry one.
cmd.CommandText =
"INSERT INTO __localdb_row_version (table_name, pk_json, hlc, node_id, is_tombstone, tombstone_utc) " +
$"SELECT $t, {TriggerSqlGenerator.JsonObject("t", table.PkColumns)}, 0, $node, 0, NULL " +
$"FROM {TriggerSqlGenerator.Ident(table.Name)} t WHERE 1 " +
"ON CONFLICT(table_name, pk_json) DO NOTHING";
cmd.Parameters.AddWithValue("$t", table.Name);
cmd.Parameters.AddWithValue("$node", _nodeId);
seeded = cmd.ExecuteNonQuery();
}
if (seeded == 0)
return;
using var flag = _master.CreateCommand();
flag.Transaction = tx;
flag.CommandText = "UPDATE __localdb_peer_state SET needs_snapshot = 1 WHERE id = 1";
flag.ExecuteNonQuery();
}
public void Dispose()
{
if (_disposed)
@@ -329,8 +463,14 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
return Convert.ToInt64(cmd.ExecuteScalar(), CultureInfo.InvariantCulture);
}
private sealed class Transaction(SqliteConnection connection, SqliteTransaction transaction) : ILocalDbTransaction
internal sealed class Transaction(SqliteConnection connection, SqliteTransaction transaction) : ILocalDbTransaction
{
/// <summary>The pinned connection, for internal callers holding their own commands.</summary>
internal SqliteConnection Connection => connection;
/// <summary>The underlying transaction, to assign to internally-held commands.</summary>
internal SqliteTransaction Sqlite => transaction;
public async Task<int> ExecuteAsync(string sql, object? parameters = null, CancellationToken ct = default)
{
await using var cmd = connection.CreateCommand();
ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Registration/TriggerSqlGenerator.cs
+23 -4
View File
@@ -27,14 +27,16 @@ internal static class TriggerSqlGenerator
"ON CONFLICT(table_name, pk_json) DO UPDATE SET hlc = excluded.hlc, node_id = excluded.node_id, " +
"is_tombstone = excluded.is_tombstone, tombstone_utc = excluded.tombstone_utc";
/// <summary>The three trigger suffixes, in the order the install/uninstall scripts emit them.</summary>
internal static readonly string[] Suffixes = ["ai", "au", "ad"];
public static string TriggerName(string table, string suffix) => $"__localdb_{table}_{suffix}";
/// <summary>DROP IF EXISTS + CREATE for all three triggers, ready to run in one transaction.</summary>
public static string BuildInstallScript(ReplicatedTable table)
{
var sb = new StringBuilder();
foreach (var suffix in new[] { "ai", "au", "ad" })
sb.Append("DROP TRIGGER IF EXISTS ").Append(Ident(TriggerName(table.Name, suffix))).Append(";\n");
sb.Append(BuildUninstallScript(table.Name));
sb.Append(BuildInsert(table)).Append('\n');
sb.Append(BuildUpdate(table)).Append('\n');
@@ -42,6 +44,20 @@ internal static class TriggerSqlGenerator
return sb.ToString();
}
/// <summary>
/// DROP IF EXISTS for all three triggers of <paramref name="tableName"/>. Takes the bare name
/// rather than a <see cref="ReplicatedTable"/> because uninstall must work for a table this
/// process never registered — the whole point of the cleanup API is stale triggers left by an
/// EARLIER build, whose column list (and therefore digest) is unavailable and irrelevant.
/// </summary>
public static string BuildUninstallScript(string tableName)
{
var sb = new StringBuilder();
foreach (var suffix in Suffixes)
sb.Append("DROP TRIGGER IF EXISTS ").Append(Ident(TriggerName(tableName, suffix))).Append(";\n");
return sb.ToString();
}
private static string BuildInsert(ReplicatedTable t) =>
Wrap(t, "ai", "INSERT", NewRowCapture(t));
@@ -112,13 +128,16 @@ internal static class TriggerSqlGenerator
}
// json_object('col', ALIAS."col", ...) — natively handles NULLs and SQLite scalar types.
private static string JsonObject(string alias, IEnumerable<string> columns)
// Internal rather than private: baseline seeding must build a pk_json identical to what the
// capture triggers emit, character for character, or a seeded ledger key would never match the
// key a later capture writes for the same row.
internal static string JsonObject(string alias, IEnumerable<string> columns)
{
var args = columns.Select(c => $"{Literal(c)}, {alias}.{Ident(c)}");
return $"json_object({string.Join(", ", args)})";
}
private static string Ident(string name) => "\"" + name.Replace("\"", "\"\"") + "\"";
internal static string Ident(string name) => "\"" + name.Replace("\"", "\"\"") + "\"";
private static string Literal(string value) => "'" + value.Replace("'", "''") + "'";
}
ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/BatchBudgetTests.cs
+430
View File
@@ -0,0 +1,430 @@
using System.Threading.Channels;
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.LocalDb.Contracts;
using ZB.MOM.WW.LocalDb.Internal;
using ZB.MOM.WW.LocalDb.Replication;
using ZB.MOM.WW.LocalDb.Replication.Internal;
namespace ZB.MOM.WW.LocalDb.Tests;
/// <summary>
/// Outbound batching by summed serialized BYTES (with the row count kept as a secondary cap), and
/// the inbound coalescing that keeps one transaction + one ack per burst rather than per message.
/// </summary>
public sealed class BatchBudgetTests : IDisposable
{
private static readonly TimeSpan RunTimeout = TimeSpan.FromSeconds(15);
private readonly List<string> _paths = [];
private readonly List<IDisposable> _disposables = [];
public void Dispose()
{
foreach (var d in _disposables)
d.Dispose();
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 async Task<SqliteLocalDb> NewOrdersDbAsync()
{
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("CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)");
db.RegisterReplicated("orders");
return db;
}
private static (SyncDuplex Duplex, ChannelWriter<SyncMessage> ToSession, ChannelReader<SyncMessage> FromSession) ScriptedPeer()
{
var toSession = Channel.CreateUnbounded<SyncMessage>();
var fromSession = Channel.CreateUnbounded<SyncMessage>();
var duplex = new SyncDuplex
{
Send = (m, ct) => fromSession.Writer.WriteAsync(m, ct).AsTask(),
Inbox = toSession.Reader,
};
return (duplex, toSession.Writer, fromSession.Reader);
}
private static Handshake MatchingHandshake(SqliteLocalDb sessionDb, string peerNodeId)
{
var h = new Handshake { NodeId = peerNodeId, LibSchemaVersion = 1, LastAppliedRemoteSeq = 0 };
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;
}
private static async Task<SyncMessage> NextAsync(ChannelReader<SyncMessage> reader, CancellationToken ct)
{
await reader.WaitToReadAsync(ct);
reader.TryRead(out var msg);
return msg!;
}
private static async Task SwallowAsync(Task task)
{
try { await task; }
catch (OperationCanceledException) { }
}
// Queues a complete peer handshake AHEAD of the session's own, so anything queued after it is
// guaranteed to be sitting in the inbox by the time the receive loop starts. That is what makes
// the coalescing assertions deterministic rather than a race with the reader.
private static async Task PreloadHandshakeAsync(
SqliteLocalDb db, ChannelWriter<SyncMessage> toSession, CancellationToken ct)
{
await toSession.WriteAsync(new SyncMessage { Handshake = MatchingHandshake(db, "peer") }, ct);
await toSession.WriteAsync(
new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = false } }, ct);
}
private static long BatchBytes(DeltaBatch batch) =>
batch.Entries.Sum(e => 1L + Google.Protobuf.CodedOutputStream.ComputeMessageSize(e));
private static long BatchBytes(SnapshotBatch batch) =>
batch.Rows.Sum(r => 1L + Google.Protobuf.CodedOutputStream.ComputeMessageSize(r));
private static SyncSession NewSession(SqliteLocalDb db, ReplicationOptions options)
{
var store = new OplogStore(db, options);
var applier = new LwwApplier(db);
return new SyncSession(db, store, applier, options, NullLogger.Instance);
}
private static OplogEntry ProtoEntry(long seq, long id, long hlc) =>
new()
{
Seq = seq,
TableName = "orders",
PkJson = $"{{\"id\":{id}}}",
RowJson = $"{{\"id\":{id},\"sku\":\"S\",\"qty\":1}}",
Hlc = hlc,
NodeId = "peer",
IsTombstone = false,
};
// ---- Outbound: delta pump --------------------------------------------------------------
[Fact]
public async Task Pump_SplitsOneReadIntoSeveralMessages_EachWithinTheByteBudget()
{
const int budget = 4_000;
var db = await NewOrdersDbAsync();
var wide = new string('x', 900);
for (var i = 1; i <= 20; i++)
await db.ExecuteAsync(
"INSERT INTO orders (id, sku, qty) VALUES (@id, @sku, @id)", new { id = i, sku = wide });
// Row count alone would put all 20 wide rows (~18 KB) in ONE message; the byte budget is
// what actually bounds the message the transport has to carry.
var options = new ReplicationOptions
{
FlushInterval = TimeSpan.FromMilliseconds(20), MaxBatchSize = 500, MaxBatchBytes = budget,
};
var session = NewSession(db, options);
var (duplex, toSession, fromSession) = ScriptedPeer();
using var cts = new CancellationTokenSource(RunTimeout);
var run = session.RunAsync(duplex, cts.Token);
await PreloadHandshakeAsync(db, toSession, cts.Token);
var seen = new List<long>();
while (seen.Count < 20)
{
var msg = await NextAsync(fromSession, cts.Token);
if (msg.MsgCase != SyncMessage.MsgOneofCase.DeltaBatch)
continue;
Assert.True(BatchBytes(msg.DeltaBatch) <= budget,
$"a {BatchBytes(msg.DeltaBatch)}-byte message exceeded the {budget}-byte budget");
Assert.True(msg.DeltaBatch.Entries.Count > 1, "the budget should still fit several of these rows");
seen.AddRange(msg.DeltaBatch.Entries.Select(e => e.Seq));
}
// Split, not dropped or duplicated: every captured seq travelled exactly once, in order.
Assert.Equal(20, seen.Count);
Assert.Equal(seen.OrderBy(x => x).ToList(), seen);
Assert.Equal(20, seen.Distinct().Count());
cts.Cancel();
await SwallowAsync(run);
}
[Fact]
public async Task Pump_RowLargerThanTheBudget_IsSentAloneRatherThanStallingTheStream()
{
var db = await NewOrdersDbAsync();
await db.ExecuteAsync(
"INSERT INTO orders (id, sku, qty) VALUES (1, @sku, 1)", new { sku = new string('x', 5_000) });
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'small', 2)");
var options = new ReplicationOptions
{
FlushInterval = TimeSpan.FromMilliseconds(20), MaxBatchSize = 500, MaxBatchBytes = 100,
};
var session = NewSession(db, options);
var (duplex, toSession, fromSession) = ScriptedPeer();
using var cts = new CancellationTokenSource(RunTimeout);
var run = session.RunAsync(duplex, cts.Token);
await PreloadHandshakeAsync(db, toSession, cts.Token);
var batches = new List<DeltaBatch>();
while (batches.Sum(b => b.Entries.Count) < 2)
{
var msg = await NextAsync(fromSession, cts.Token);
if (msg.MsgCase == SyncMessage.MsgOneofCase.DeltaBatch)
batches.Add(msg.DeltaBatch);
}
// The oversized row goes out on its own — refusing it would wedge the stream on one row
// forever, and everything behind it with it.
Assert.Equal(2, batches.Count);
Assert.Single(batches[0].Entries);
Assert.Equal(1, batches[0].Entries[0].Seq);
Assert.True(BatchBytes(batches[0]) > options.MaxBatchBytes);
Assert.Single(batches[1].Entries);
Assert.Equal(2, batches[1].Entries[0].Seq);
cts.Cancel();
await SwallowAsync(run);
}
[Fact]
public async Task Pump_RowCountRemainsASecondaryCap()
{
var db = await NewOrdersDbAsync();
for (var i = 1; i <= 12; i++)
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'S', @id)", new { id = i });
// Tiny rows, enormous byte budget: only the row cap can be doing the splitting here.
var options = new ReplicationOptions
{
FlushInterval = TimeSpan.FromMilliseconds(20), MaxBatchSize = 5, MaxBatchBytes = 8 * 1024 * 1024,
};
var session = NewSession(db, options);
var (duplex, toSession, fromSession) = ScriptedPeer();
using var cts = new CancellationTokenSource(RunTimeout);
var run = session.RunAsync(duplex, cts.Token);
await PreloadHandshakeAsync(db, toSession, cts.Token);
var seen = 0;
while (seen < 12)
{
var msg = await NextAsync(fromSession, cts.Token);
if (msg.MsgCase != SyncMessage.MsgOneofCase.DeltaBatch)
continue;
Assert.True(msg.DeltaBatch.Entries.Count <= 5);
seen += msg.DeltaBatch.Entries.Count;
}
cts.Cancel();
await SwallowAsync(run);
}
// ---- Outbound: snapshot --------------------------------------------------------------
[Fact]
public async Task Snapshot_SplitsPagesIntoMessagesWithinTheByteBudget()
{
const int budget = 4_000;
var db = await NewOrdersDbAsync();
var wide = new string('x', 900);
for (var i = 1; i <= 20; i++)
await db.ExecuteAsync(
"INSERT INTO orders (id, sku, qty) VALUES (@id, @sku, @id)", new { id = i, sku = wide });
// A snapshot is the easiest way to blow a message limit: it ships whole rows, all of them,
// back to back — so its page size has to be bounded in bytes too, not just in rows.
var options = new ReplicationOptions { MaxBatchSize = 500, MaxBatchBytes = budget };
var store = new OplogStore(db, options);
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);
var batches = sent.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
.Select(m => m.SnapshotBatch).ToList();
Assert.True(batches.Count > 1, "20 wide rows must not have travelled as one message");
Assert.All(batches, b => Assert.True(BatchBytes(b) <= budget,
$"a {BatchBytes(b)}-byte snapshot message exceeded the {budget}-byte budget"));
// Bracketed and complete: every row, exactly once.
Assert.Equal(SyncMessage.MsgOneofCase.SnapshotBegin, sent[0].MsgCase);
Assert.Equal(SyncMessage.MsgOneofCase.SnapshotComplete, sent[^1].MsgCase);
var pks = batches.SelectMany(b => b.Rows).Select(r => r.PkJson).ToList();
Assert.Equal(20, pks.Count);
Assert.Equal(20, pks.Distinct().Count());
}
[Fact]
public async Task Snapshot_RowLargerThanTheBudget_IsSentAlone()
{
var db = await NewOrdersDbAsync();
await db.ExecuteAsync(
"INSERT INTO orders (id, sku, qty) VALUES (1, @sku, 1)", new { sku = new string('x', 5_000) });
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'small', 2)");
var options = new ReplicationOptions { MaxBatchSize = 500, MaxBatchBytes = 100 };
var store = new OplogStore(db, options);
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);
var batches = sent.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
.Select(m => m.SnapshotBatch).ToList();
Assert.Equal(2, batches.Count);
Assert.All(batches, b => Assert.Single(b.Rows));
Assert.Equal("{\"id\":1}", batches[0].Rows[0].PkJson);
Assert.Equal("{\"id\":2}", batches[1].Rows[0].PkJson);
}
// ---- Inbound: coalescing -------------------------------------------------------------
[Fact]
public async Task Receive_CoalescesQueuedDeltaBatches_IntoOneAck()
{
var db = await NewOrdersDbAsync();
var options = new ReplicationOptions { FlushInterval = TimeSpan.FromSeconds(30) };
var session = NewSession(db, options);
var (duplex, toSession, fromSession) = ScriptedPeer();
using var cts = new CancellationTokenSource(RunTimeout);
var run = session.RunAsync(duplex, cts.Token);
// Queued ahead of the receive loop, so all three are demonstrably in the inbox together.
await PreloadHandshakeAsync(db, toSession, cts.Token);
for (var i = 1; i <= 3; i++)
{
var batch = new DeltaBatch();
batch.Entries.Add(ProtoEntry(i, i, 1_000_000 + i));
await toSession.WriteAsync(new SyncMessage { DeltaBatch = batch }, cts.Token);
}
SyncMessage ack;
do { ack = await NextAsync(fromSession, cts.Token); }
while (ack.MsgCase != SyncMessage.MsgOneofCase.DeltaAck);
// The discriminator: uncoalesced, the FIRST ack could only cover seq 1.
Assert.Equal(3, ack.DeltaAck.AppliedThruSeq);
var rows = await db.QueryAsync("SELECT COUNT(*) FROM orders", r => r.GetInt64(0));
Assert.Equal(3, rows[0]);
cts.Cancel();
await SwallowAsync(run);
}
[Fact]
public async Task Receive_DeltaCoalescing_DoesNotSwallowTheMessageThatEndsTheRun()
{
// The risk the one-message lookahead exists to handle: coalescing reads ahead, and whatever
// it reads that is NOT a delta must still be processed, not dropped on the floor.
var db = await NewOrdersDbAsync();
var options = new ReplicationOptions { FlushInterval = TimeSpan.FromSeconds(30) };
var session = NewSession(db, options);
var captured = new CapturingSnapshotApplier();
session.SnapshotApplier = captured;
var (duplex, toSession, fromSession) = ScriptedPeer();
using var cts = new CancellationTokenSource(RunTimeout);
var run = session.RunAsync(duplex, cts.Token);
await PreloadHandshakeAsync(db, toSession, cts.Token);
var batch = new DeltaBatch();
batch.Entries.Add(ProtoEntry(1, 1, 1_000_001));
await toSession.WriteAsync(new SyncMessage { DeltaBatch = batch }, cts.Token);
await toSession.WriteAsync(
new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 77 } }, cts.Token);
SyncMessage ack;
do { ack = await NextAsync(fromSession, cts.Token); }
while (ack.MsgCase != SyncMessage.MsgOneofCase.DeltaAck);
Assert.Equal(1, ack.DeltaAck.AppliedThruSeq);
// The SnapshotBegin that terminated the coalescing run still reached its handler.
await WaitForAsync(() => captured.BeginAsOfSeq == 77, RunTimeout);
cts.Cancel();
await SwallowAsync(run);
}
[Fact]
public async Task Receive_CoalescesQueuedAcks_ToTheHighestWatermark_WithoutLosingTheNextMessage()
{
var db = await NewOrdersDbAsync();
for (var i = 1; i <= 9; i++)
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'S', @id)", new { id = i });
var options = new ReplicationOptions { FlushInterval = TimeSpan.FromSeconds(30) };
var session = NewSession(db, options);
var captured = new CapturingSnapshotApplier();
session.SnapshotApplier = captured;
var (duplex, toSession, fromSession) = ScriptedPeer();
using var cts = new CancellationTokenSource(RunTimeout);
var run = session.RunAsync(duplex, cts.Token);
await PreloadHandshakeAsync(db, toSession, cts.Token);
// Out of order on purpose: collapsing acks must take the MAX, never the last.
foreach (var seq in new long[] { 4, 9, 2 })
await toSession.WriteAsync(
new SyncMessage { DeltaAck = new DeltaAck { AppliedThruSeq = seq } }, cts.Token);
await toSession.WriteAsync(
new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 55 } }, cts.Token);
await WaitForAsync(async () =>
{
var acked = await db.QueryAsync(
"SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1", r => r.GetInt64(0));
return acked[0] == 9;
}, RunTimeout);
Assert.Equal(9, session.PeerAckedSeq);
await WaitForAsync(() => captured.BeginAsOfSeq == 55, RunTimeout);
cts.Cancel();
await SwallowAsync(run);
}
private sealed class CapturingSnapshotApplier : ISnapshotApplier
{
private long _beginAsOfSeq = -1;
public long BeginAsOfSeq => Interlocked.Read(ref _beginAsOfSeq);
public Task OnBeginAsync(SnapshotBegin begin, CancellationToken ct)
{
Interlocked.Exchange(ref _beginAsOfSeq, begin.AsOfSeq);
return Task.CompletedTask;
}
public Task OnBatchAsync(SnapshotBatch batch, CancellationToken ct) => Task.CompletedTask;
public Task OnCompleteAsync(SnapshotComplete complete, CancellationToken ct) => Task.CompletedTask;
}
private static Task WaitForAsync(Func<bool> predicate, TimeSpan timeout) =>
WaitForAsync(() => Task.FromResult(predicate()), timeout);
private static async Task WaitForAsync(Func<Task<bool>> predicate, TimeSpan timeout)
{
var deadline = DateTime.UtcNow + timeout;
while (DateTime.UtcNow < deadline)
{
if (await predicate()) return;
await Task.Delay(20);
}
throw new TimeoutException("Condition not reached within " + timeout);
}
}
ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/GrpcAdapterTests.cs
+65
View File
@@ -173,6 +173,71 @@ public sealed class GrpcAdapterTests : IAsyncLifetime
Assert.True(sw.Elapsed < TimeSpan.FromSeconds(3), $"teardown took {sw.Elapsed}");
}
// ---- inbox backpressure ------------------------------------------------------------------
[Fact]
public async Task Inbox_SlowReader_StopsDrainingTheStream_AndLosesNothing()
{
// The pump must STOP pulling from the gRPC stream once the inbox is full — that stall is
// what closes the peer's flow-control window and bounds this process's memory at
// InboxCapacity x MaxBatchBytes instead of "whatever the peer decides to send" (a full
// snapshot of a large dataset, on both nodes at once, when a pair baselines late).
const int capacity = Replication.Internal.GrpcSyncDuplex.InboxCapacity;
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));
var reader = new CountingReader();
var (duplex, readerTask) = Replication.Internal.GrpcSyncDuplex.Create(
reader, new HangingWriter(), cts.Token);
// Nothing reads the inbox: the pump fills it and blocks.
await WaitForAsync(() => Task.FromResult(reader.Produced >= capacity), TimeSpan.FromSeconds(5));
await Task.Delay(200, cts.Token);
// capacity in the channel + at most one in the pump's hand, blocked on WriteAsync.
Assert.True(reader.Produced <= capacity + 1,
$"the reader pump kept draining an already-full inbox: {reader.Produced} messages read for a {capacity}-message bound");
// Draining releases it, and every message arrives exactly once in stream order — TryWrite
// on a full bounded channel would have dropped them silently instead.
for (long expected = 1; expected <= capacity + 1; expected++)
{
var msg = await ReadInboxAsync(duplex, cts.Token);
Assert.Equal(expected, msg.DeltaAck.AppliedThruSeq);
}
await WaitForAsync(() => Task.FromResult(reader.Produced > capacity + 1), TimeSpan.FromSeconds(5));
cts.Cancel();
try { await readerTask.WaitAsync(TimeSpan.FromSeconds(5)); }
catch (OperationCanceledException) { }
}
private static async Task<SyncMessage> ReadInboxAsync(
Replication.Internal.SyncDuplex duplex, CancellationToken ct)
{
await duplex.Inbox.WaitToReadAsync(ct);
duplex.Inbox.TryRead(out var message);
return message!;
}
// Produces messages as fast as the inbox will take them, numbering each one so the consumer can
// prove nothing was dropped or reordered.
private sealed class CountingReader : IAsyncStreamReader<SyncMessage>
{
private long _produced;
public long Produced => Interlocked.Read(ref _produced);
public SyncMessage Current { get; private set; } = new();
public Task<bool> MoveNext(CancellationToken cancellationToken)
{
cancellationToken.ThrowIfCancellationRequested();
Current = new SyncMessage
{
DeltaAck = new DeltaAck { AppliedThruSeq = Interlocked.Increment(ref _produced) },
};
return Task.FromResult(true);
}
}
// ---- validator -------------------------------------------------------------------------
[Fact]
ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/LocalDbSchemaTests.cs
+67
View File
@@ -55,8 +55,75 @@ public sealed class LocalDbSchemaTests : IDisposable
})
Assert.Contains(expected, tables);
// __localdb_oplog_hlc is deliberately NOT created: schema v2 removed it. It indexed
// (table_name, pk_json, hlc), which no query has ever used — every oplog access path is by
// seq or a full hlc-predicate delete — while costing a b-tree insert inside every captured
// consumer write.
Assert.Equal(0, ScalarLong(conn,
"SELECT count(*) FROM sqlite_master WHERE type='index' AND name='__localdb_oplog_hlc'"));
Assert.Equal(LocalDbSchema.CurrentSchemaVersion,
ScalarLong(conn, "SELECT schema_version FROM __localdb_meta WHERE id=1"));
}
[Fact]
public void EnsureCreated_UpgradesLegacyFile_DropsOplogHlcIndex()
{
using var conn = Open();
// A file exactly as schema v1 left it: the tables, the index, and a v1 stamp.
LocalDbSchema.EnsureCreated(conn);
Exec(conn, "CREATE INDEX __localdb_oplog_hlc ON __localdb_oplog(table_name, pk_json, hlc);" +
"UPDATE __localdb_meta SET schema_version = 1 WHERE id = 1;");
Assert.Equal(1, ScalarLong(conn,
"SELECT count(*) FROM sqlite_master WHERE type='index' AND name='__localdb_oplog_hlc'"));
LocalDbSchema.EnsureCreated(conn);
Assert.Equal(0, ScalarLong(conn,
"SELECT count(*) FROM sqlite_master WHERE type='index' AND name='__localdb_oplog_hlc'"));
Assert.Equal(LocalDbSchema.CurrentSchemaVersion,
ScalarLong(conn, "SELECT schema_version FROM __localdb_meta WHERE id=1"));
}
[Fact]
public void EnsureCreated_Upgrade_PreservesNodeIdentityAndOplogRows()
{
using var conn = Open();
LocalDbSchema.EnsureCreated(conn);
var nodeId = ScalarString(conn, "SELECT node_id FROM __localdb_meta WHERE id=1");
Exec(conn,
"CREATE INDEX __localdb_oplog_hlc ON __localdb_oplog(table_name, pk_json, hlc);" +
"UPDATE __localdb_meta SET schema_version = 1 WHERE id = 1;" +
"INSERT INTO __localdb_oplog (table_name, pk_json, row_json, hlc, node_id, is_tombstone) " +
"VALUES ('orders', '{\"id\":1}', '{\"id\":1}', 42, 'n', 0);" +
"UPDATE __localdb_peer_state SET last_acked_seq = 7 WHERE id = 1;");
LocalDbSchema.EnsureCreated(conn);
// The upgrade is DDL only — identity, replication watermarks and captured data survive it.
Assert.Equal(nodeId, ScalarString(conn, "SELECT node_id FROM __localdb_meta WHERE id=1"));
Assert.Equal(1, ScalarLong(conn, "SELECT COUNT(*) FROM __localdb_oplog"));
Assert.Equal(42, ScalarLong(conn, "SELECT hlc FROM __localdb_oplog"));
Assert.Equal(7, ScalarLong(conn, "SELECT last_acked_seq FROM __localdb_peer_state WHERE id=1"));
}
[Fact]
public void EnsureCreated_AlreadyCurrent_LeavesVersionAlone()
{
using var conn = Open();
LocalDbSchema.EnsureCreated(conn);
LocalDbSchema.EnsureCreated(conn);
LocalDbSchema.EnsureCreated(conn);
Assert.Equal(LocalDbSchema.CurrentSchemaVersion,
ScalarLong(conn, "SELECT schema_version FROM __localdb_meta WHERE id=1"));
}
private static void Exec(SqliteConnection conn, string sql)
{
using var cmd = conn.CreateCommand();
cmd.CommandText = sql;
cmd.ExecuteNonQuery();
}
[Fact]
ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/LwwApplierTests.cs
+58
View File
@@ -404,6 +404,64 @@ public sealed class LwwApplierTests : IDisposable
Assert.Equal(("NEW", (long?)7), await ReadOrder(db, 2));
}
[Fact]
public async Task Apply_PreparesEachStatementOnce_NotOncePerRow()
{
// The regression this guards: the old shape went through the generic transaction façade,
// which builds a fresh SqliteCommand — and therefore re-parses the SQL — per call. Three
// statements per applied row means a 200-row batch parsed 600 statements to run 6 distinct
// ones. Statement count must track distinct SQL, not rows.
using var db = await NewOrdersDb();
await db.ExecuteAsync("CREATE TABLE widgets (id INTEGER PRIMARY KEY, name TEXT)");
db.RegisterReplicated("widgets");
var entries = new List<OplogEntryRecord>();
for (var i = 1; i <= 200; i++)
entries.Add(Entry(i, $"{{\"id\":{i}}}", $"{{\"id\":{i},\"sku\":\"S\",\"qty\":{i}}}", 1_000_000 + i, "remote"));
// A second table and a tombstone, so every statement kind in the set is exercised.
entries.Add(new OplogEntryRecord(
201, "widgets", "{\"id\":1}", "{\"id\":1,\"name\":\"W\"}", 1_000_300, "remote", false));
entries.Add(Entry(202, "{\"id\":1}", null, 1_000_400, "remote", tombstone: true));
var applier = new LwwApplier(db);
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);
// ...and the batch is still correct, statement reuse notwithstanding.
Assert.Null(await ReadOrder(db, 1));
Assert.Equal(("S", (long?)200), await ReadOrder(db, 200));
var widget = await db.QueryAsync("SELECT name FROM widgets WHERE id = 1", x => x.GetString(0));
Assert.Equal("W", widget[0]);
}
[Fact]
public async Task Apply_ReusedStatements_SurviveAPoisonRowMidBatch()
{
// A statement that faults is reset and reused for the following rows; if it were not, one
// poison row would silently corrupt every row after it in the batch.
using var db = await NewOrdersDb();
var applier = new LwwApplier(db);
var result = await applier.ApplyBatchAsync(
[
Entry(1, "{\"id\":1}", "{\"id\":1,\"sku\":\"BEFORE\",\"qty\":1}", 1_000_000, "remote"),
Entry(2, "{\"id\":2}", "NOT VALID JSON", 1_000_001, "remote"),
Entry(3, "{\"id\":3}", "{\"id\":3,\"sku\":\"AFTER\",\"qty\":3}", 1_000_002, "remote"),
Entry(4, "{\"id\":4}", "ALSO NOT JSON", 1_000_003, "remote"),
Entry(5, "{\"id\":5}", "{\"id\":5,\"sku\":\"LAST\",\"qty\":5}", 1_000_004, "remote"),
], default);
Assert.Equal(3, result.Applied);
Assert.Equal(2, result.DeadLettered);
Assert.Equal(("BEFORE", (long?)1), await ReadOrder(db, 1));
Assert.Equal(("AFTER", (long?)3), await ReadOrder(db, 3));
Assert.Equal(("LAST", (long?)5), await ReadOrder(db, 5));
}
[Fact]
public async Task Apply_ReplaceExistingRow_DoesNotFireCaptureTriggers()
{
ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/OplogStoreTests.cs
+131
View File
@@ -212,6 +212,137 @@ public sealed class OplogStoreTests : IDisposable
Assert.Equal("{\"id\":2}", remaining[0].PkJson);
}
// ---- O(1) backlog depth --------------------------------------------------------------
// The definition the O(1) formula has to reproduce. Every scenario below is asserted against
// this, not against a hand-computed number, so the two can never drift apart.
private static async Task<long> CountingDepth(SqliteLocalDb db) =>
(await db.QueryAsync(
"SELECT COUNT(*) FROM __localdb_oplog WHERE seq > (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)",
r => r.GetInt64(0)))[0];
private static async Task AssertDepthMatchesCount(SqliteLocalDb db, OplogStore store)
{
var expected = await CountingDepth(db);
Assert.Equal(expected, await store.GetOplogDepthAsync(default));
Assert.Equal(expected, store.GetOplogDepthSync());
}
[Fact]
public async Task OplogDepth_Empty_IsZero()
{
using var db = await NewOrdersDb();
var store = new OplogStore(db, new ReplicationOptions());
await AssertDepthMatchesCount(db, store);
Assert.Equal(0, await store.GetOplogDepthAsync(default));
}
[Fact]
public async Task OplogDepth_AllAcked_EmptiesToZero_NotToTheSeqRange()
{
using var db = await NewOrdersDb();
await WriteRows(db, 5);
var store = new OplogStore(db, new ReplicationOptions());
await store.RecordPeerAckAsync(await store.GetMaxSeqAsync(default), default);
await AssertDepthMatchesCount(db, store);
Assert.Equal(0, await store.GetOplogDepthAsync(default));
}
[Fact]
public async Task OplogDepth_AfterCapPrune_ReportsRetainedRows_NotTheWholeSeqRange()
{
// The reason the formula clamps its floor to MIN(seq) - 1. A cap prune leaves the retained
// rows far above an ack watermark that never moved, so a naive MAX(seq) - last_acked would
// report the entire seq range forever and re-flag the cap on every maintenance tick.
using var db = await NewOrdersDb();
await WriteRows(db, 40);
var store = new OplogStore(db, new ReplicationOptions { MaxOplogRows = 5 });
Assert.True(await store.EnforceCapsAsync(default));
await AssertDepthMatchesCount(db, store);
Assert.Equal(5, await store.GetOplogDepthAsync(default));
// And the next tick is clean: depth is not over cap any more.
Assert.False(await store.EnforceCapsAsync(default));
}
[Fact]
public async Task OplogDepth_AfterAgePrune_MatchesCount()
{
using var db = await NewOrdersDb();
await WriteRows(db, 6);
// Backdate the oldest three past the age cap; HLCs rise with seq, so this is a prefix.
var expired = DateTimeOffset.UtcNow.AddDays(-30).ToUnixTimeMilliseconds() << 16;
await db.ExecuteAsync(
"UPDATE __localdb_oplog SET hlc = @h WHERE seq IN (SELECT seq FROM __localdb_oplog ORDER BY seq LIMIT 3)",
new { h = expired });
var store = new OplogStore(db, new ReplicationOptions());
Assert.True(await store.EnforceCapsAsync(default));
await AssertDepthMatchesCount(db, store);
Assert.Equal(3, await store.GetOplogDepthAsync(default));
}
[Fact]
public async Task OplogDepth_PartialAck_MatchesCount()
{
using var db = await NewOrdersDb();
await WriteRows(db, 9);
var store = new OplogStore(db, new ReplicationOptions());
var batch = await store.ReadBatchAboveAsync(0, 100, default);
await store.RecordPeerAckAsync(batch[3].Seq, default);
await AssertDepthMatchesCount(db, store);
Assert.Equal(5, await store.GetOplogDepthAsync(default));
}
[Fact]
public async Task OplogDepth_AckPruneThenMoreWrites_MatchesCount()
{
using var db = await NewOrdersDb();
await WriteRows(db, 4);
var store = new OplogStore(db, new ReplicationOptions());
await store.RecordPeerAckAsync(await store.GetMaxSeqAsync(default), default);
// The steady state of a converged pair: oplog pruned empty, watermark high, new writes above it.
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (91,'S',1),(92,'S',2)");
await AssertDepthMatchesCount(db, store);
Assert.Equal(2, await store.GetOplogDepthAsync(default));
}
[Fact]
public async Task OplogDepth_LargeBacklog_IsConstantTime()
{
// 200k rows: a COUNT(*) over the backlog runs ~200k VM steps per call (measured: 100,020 at
// 50k rows), the seek-based formula runs 58 regardless of size. The loop below costs tens of
// milliseconds at seek shape and seconds at scan shape, so the budget separates them by a
// wide margin in both directions.
using var db = await NewOrdersDb();
await db.ExecuteAsync(
"INSERT INTO __localdb_oplog (table_name, pk_json, row_json, hlc, node_id, is_tombstone) " +
"WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM c WHERE x < 200000) " +
"SELECT 'orders', '{\"id\":' || x || '}', '{\"id\":' || x || '}', x << 16, 'n', 0 FROM c");
var store = new OplogStore(db, new ReplicationOptions());
Assert.Equal(200_000, await store.GetOplogDepthAsync(default));
using var conn = db.CreateConnection();
using var cmd = conn.CreateCommand();
cmd.CommandText = OplogStore.DepthSql;
var sw = System.Diagnostics.Stopwatch.StartNew();
for (var i = 0; i < 2_000; i++)
Assert.Equal(200_000L, (long)cmd.ExecuteScalar()!);
sw.Stop();
Assert.True(sw.ElapsedMilliseconds < 1_500,
$"2000 depth reads over a 200k backlog took {sw.ElapsedMilliseconds} ms — that is scan-shaped, not seek-shaped.");
}
[Fact]
public async Task PeerState_RoundTrips()
{
ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/RegistrationTests.cs
+355
View File
@@ -128,4 +128,359 @@ public sealed class RegistrationTests : IDisposable
Assert.Equal(new[] { "id" }, table.PkColumns);
Assert.Equal(new[] { "id", "sku", "qty" }, table.Columns.Select(c => c.Name).ToArray());
}
// ---- Deregistration ------------------------------------------------------------------
[Fact]
public async Task Deregister_DropsTriggers_PrunesOplogAndLedger_LeavesDataAlone()
{
using var db = Create();
await CreateOrders(db);
db.RegisterReplicated("orders");
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1),(2,'B',2)");
var removed = db.DeregisterReplicated("orders");
Assert.True(removed);
Assert.Empty(await Triggers(db));
Assert.Equal(0, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name='orders'"));
Assert.Equal(0, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_row_version WHERE table_name='orders'"));
Assert.False(db.ReplicatedTables.ContainsKey("orders"));
// The consumer's own data is never touched — this removes CAPTURE, not rows.
Assert.Equal(2, await CountAsync(db, "SELECT COUNT(*) FROM orders"));
}
[Fact]
public async Task Deregister_StopsCapture()
{
using var db = Create();
await CreateOrders(db);
db.RegisterReplicated("orders");
db.DeregisterReplicated("orders");
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
await db.ExecuteAsync("UPDATE orders SET qty = 9 WHERE id = 1");
await db.ExecuteAsync("DELETE FROM orders WHERE id = 1");
Assert.Equal(0, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_oplog"));
Assert.Equal(0, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_row_version"));
}
[Fact]
public async Task Deregister_IsIdempotent_SecondCallReportsNothingToDo()
{
using var db = Create();
await CreateOrders(db);
db.RegisterReplicated("orders");
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
Assert.True(db.DeregisterReplicated("orders"));
Assert.False(db.DeregisterReplicated("orders"));
Assert.False(db.DeregisterReplicated("orders"));
Assert.Empty(await Triggers(db));
}
[Fact]
public async Task Deregister_NeverRegisteredTable_IsANoOp()
{
using var db = Create();
await CreateOrders(db);
Assert.False(db.DeregisterReplicated("orders"));
Assert.False(db.DeregisterReplicated("ghost"));
}
[Fact]
public async Task Deregister_ClearsTriggersLeftByAnEarlierProcess()
{
// The primary use: a database file registered by an OLDER build, opened by one that has
// decided this node does not replicate. Nothing in memory says the table was ever
// registered — the evidence is entirely on disk.
using (var first = Create())
{
await CreateOrders(first);
first.RegisterReplicated("orders");
await first.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
}
SqliteConnection.ClearAllPools();
using var second = Create();
Assert.Equal(3, (await Triggers(second)).Count);
Assert.False(second.ReplicatedTables.ContainsKey("orders"));
Assert.True(second.DeregisterReplicated("orders"));
Assert.Empty(await Triggers(second));
Assert.Equal(0, await CountAsync(second, "SELECT COUNT(*) FROM __localdb_oplog"));
}
[Fact]
public async Task Deregister_LeavesOtherTablesCapturing()
{
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync("CREATE TABLE widgets (id INTEGER PRIMARY KEY, name TEXT)");
db.RegisterReplicated("orders");
db.RegisterReplicated("widgets");
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.DeregisterReplicated("orders");
await db.ExecuteAsync("INSERT INTO widgets (id, name) VALUES (2,'W2')");
Assert.Equal(0, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name='orders'"));
Assert.Equal(2, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name='widgets'"));
Assert.True(db.ReplicatedTables.ContainsKey("widgets"));
}
[Fact]
public async Task Deregister_ThenRegisterWithBaseline_RestoresTheLedger()
{
// Pruning the ledger is only safe because re-registration can re-baseline; this is that
// round trip.
using var db = Create();
await CreateOrders(db);
db.RegisterReplicated("orders");
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1),(2,'B',2)");
db.DeregisterReplicated("orders");
Assert.Equal(0, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_row_version"));
db.RegisterReplicated("orders", baselineExistingRows: true);
Assert.Equal(2, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_row_version WHERE table_name='orders'"));
}
// ---- Baselining ----------------------------------------------------------------------
[Fact]
public async Task Register_WithoutBaseline_LeavesPreexistingRowsUnledgered()
{
// The documented CDC contract, unchanged: opting in is what changes behaviour, not upgrading.
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
db.RegisterReplicated("orders");
Assert.Equal(0, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_row_version"));
Assert.False(await NeedsSnapshot(db));
}
[Fact]
public async Task Register_WithBaseline_SeedsAtHlcZero_AndFlagsSnapshot()
{
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1),(2,'B',2)");
db.RegisterReplicated("orders", baselineExistingRows: true);
var seeded = await db.QueryAsync(
"SELECT pk_json, hlc, node_id, is_tombstone FROM __localdb_row_version ORDER BY pk_json",
r => (Pk: r.GetString(0), Hlc: r.GetInt64(1), Node: r.GetString(2), Tomb: r.GetInt64(3)));
Assert.Equal(2, seeded.Count);
Assert.Equal(["{\"id\":1}", "{\"id\":2}"], seeded.Select(s => s.Pk));
// HLC 0 is the floor: below every genuine stamp (physical ms << 16), so it can never beat one.
Assert.All(seeded, s => Assert.Equal(0, s.Hlc));
Assert.All(seeded, s => Assert.Equal(db.NodeId, s.Node));
Assert.All(seeded, s => Assert.Equal(0, s.Tomb));
// 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));
}
[Fact]
public async Task Register_WithBaseline_IsIdempotent_AndDoesNotReflagSnapshot()
{
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);
await db.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 0 WHERE id = 1");
db.RegisterReplicated("orders", baselineExistingRows: true);
Assert.Equal(1, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_row_version"));
// Nothing new was seeded, so no resync is owed: passing true every boot is free after the first.
Assert.False(await NeedsSnapshot(db));
}
[Fact]
public async Task Register_WithBaseline_NeverOverwritesACapturedVersion()
{
// The LWW-safety core: a row with a real, captured version must keep it. A floor that
// clobbered genuine versions would make every baselined node look newer than its peer.
using var db = Create();
await CreateOrders(db);
db.RegisterReplicated("orders");
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
var captured = (await db.QueryAsync(
"SELECT hlc FROM __localdb_row_version WHERE pk_json='{\"id\":1}'", r => r.GetInt64(0)))[0];
Assert.True(captured > 0);
// A second, pre-existing row that capture never saw (inserted with triggers removed).
db.DeregisterReplicated("orders");
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2,'B',2)");
await db.ExecuteAsync(
"INSERT INTO __localdb_row_version (table_name, pk_json, hlc, node_id, is_tombstone) " +
"VALUES ('orders', '{\"id\":1}', @h, 'genuine', 0)", new { h = captured });
db.RegisterReplicated("orders", baselineExistingRows: true);
var versions = await db.QueryAsync(
"SELECT pk_json, hlc, node_id FROM __localdb_row_version ORDER BY pk_json",
r => (Pk: r.GetString(0), Hlc: r.GetInt64(1), Node: r.GetString(2)));
Assert.Equal(2, versions.Count);
Assert.Equal(captured, versions[0].Hlc);
Assert.Equal("genuine", versions[0].Node);
Assert.Equal(0, versions[1].Hlc);
}
[Fact]
public async Task Register_WithBaseline_SeedsPkJsonIdenticalToWhatCaptureWrites()
{
// A seeded key that differs by so much as a space from the capture triggers' json_object
// output would be a DIFFERENT ledger row, and the baseline would silently never merge.
using var db = Create();
await db.ExecuteAsync("CREATE TABLE kv (a INTEGER, b TEXT, v TEXT, PRIMARY KEY (a, b))");
await db.ExecuteAsync("INSERT INTO kv (a, b, v) VALUES (1, 'x', 'seeded')");
db.RegisterReplicated("kv", baselineExistingRows: true);
await db.ExecuteAsync("UPDATE kv SET v = 'captured' WHERE a = 1 AND b = 'x'");
// One ledger row, not two: the capture upsert landed on the seeded key.
var rows = await db.QueryAsync(
"SELECT pk_json, hlc FROM __localdb_row_version", r => (Pk: r.GetString(0), Hlc: r.GetInt64(1)));
Assert.Single(rows);
Assert.Equal("{\"a\":1,\"b\":\"x\"}", rows[0].Pk);
Assert.True(rows[0].Hlc > 0, "the genuine capture must have replaced the floor");
}
[Fact]
public async Task Register_WithBaseline_FloorLosesToEveryGenuineRemoteWrite()
{
// Proved through the real LWW path rather than by reasoning about the comparison: an
// absurdly small remote HLC of 1 still beats the baseline floor of 0.
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'BASELINE',1)");
db.RegisterReplicated("orders", baselineExistingRows: true);
var result = await new Replication.Internal.LwwApplier(db).ApplyBatchAsync(
[new Replication.Internal.OplogEntryRecord(
1, "orders", "{\"id\":1}", "{\"id\":1,\"sku\":\"REMOTE\",\"qty\":9}", 1, "peer", false)],
default);
Assert.Equal(1, result.Applied);
var sku = await db.QueryAsync("SELECT sku FROM orders WHERE id = 1", r => r.GetString(0));
Assert.Equal("REMOTE", sku[0]);
}
[Fact]
public async Task Register_WithBaseline_TieBetweenTwoBaselinedNodes_ResolvesByNodeIdOrdinal()
{
// Two nodes that baselined the same key both hold HLC 0 — genuinely unordered history. The
// node-id tie-break is what keeps that convergent instead of oscillating.
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'MINE',1)");
db.RegisterReplicated("orders", baselineExistingRows: true);
var applier = new Replication.Internal.LwwApplier(db);
var lower = await applier.ApplyBatchAsync(
[new Replication.Internal.OplogEntryRecord(
1, "orders", "{\"id\":1}", "{\"id\":1,\"sku\":\"LOWER\",\"qty\":9}", 0, "!" + db.NodeId, false)],
default);
Assert.Equal(1, lower.Discarded);
var higher = await applier.ApplyBatchAsync(
[new Replication.Internal.OplogEntryRecord(
2, "orders", "{\"id\":1}", "{\"id\":1,\"sku\":\"HIGHER\",\"qty\":9}", 0, db.NodeId + "z", false)],
default);
Assert.Equal(1, higher.Applied);
var sku = await db.QueryAsync("SELECT sku FROM orders WHERE id = 1", r => r.GetString(0));
Assert.Equal("HIGHER", sku[0]);
}
[Fact]
public async Task Register_WithBaseline_EmptyTable_SeedsNothing_FlagsNothing()
{
using var db = Create();
await CreateOrders(db);
db.RegisterReplicated("orders", baselineExistingRows: true);
Assert.Equal(0, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_row_version"));
Assert.False(await NeedsSnapshot(db));
}
// ---- Deregistration and the HLC crash anchor ------------------------------------------
[Fact]
public async Task Deregister_FlushesHlcAnchor_SoAnUngracefulReopenCannotRegress()
{
// The clock recovers as MAX(hlc_high_water, oplog HLCs, ledger HLCs, last_seen_hlc), and
// deregistering the only registered table empties both ledgers. Without the in-transaction
// flush, an ungraceful exit right afterwards leaves hlc_high_water at 0 and the reopened
// clock restarts from wall-clock time — below stamps the peer already holds.
var crashed = Create();
try
{
await CreateOrders(crashed);
crashed.RegisterReplicated("orders");
await crashed.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
// A peer HLC an hour ahead, merged as the applier would merge it: this is what makes the
// clock outrun wall-clock time, and therefore what a wall-clock restart would regress.
var peerHlc = DateTimeOffset.UtcNow.AddHours(1).ToUnixTimeMilliseconds() << 16;
crashed.Clock.Observe(peerHlc);
var preDereg = crashed.Clock.Current;
Assert.True(preDereg >= peerHlc);
crashed.DeregisterReplicated("orders");
Assert.Equal(preDereg, await CountAsync(crashed, "SELECT hlc_high_water FROM __localdb_meta WHERE id = 1"));
// The evidence the clock would otherwise have recovered from is gone.
Assert.Equal(0, await CountAsync(crashed, "SELECT COUNT(*) FROM __localdb_oplog"));
Assert.Equal(0, await CountAsync(crashed, "SELECT COUNT(*) FROM __localdb_row_version"));
Assert.Equal(0, await CountAsync(crashed, "SELECT last_seen_hlc FROM __localdb_peer_state WHERE id = 1"));
// Ungraceful reopen: a second instance over the same file while the first was NEVER
// disposed, so nothing but the deregistration itself can have written the anchor.
using var reopened = Create();
Assert.True(reopened.Clock.Current >= preDereg,
$"recovered clock {reopened.Clock.Current} regressed below the pre-deregistration stamp {preDereg}");
Assert.True(reopened.Clock.Next() > preDereg);
}
finally
{
crashed.Dispose();
}
}
[Fact]
public async Task Deregister_HlcFlush_DoesNotCountAsWorkDone()
{
// The flush is an UPDATE on every call, including the ones that clean up nothing; it must
// not turn the "was there anything to do?" answer into a permanent true.
using var db = Create();
await CreateOrders(db);
Assert.False(db.DeregisterReplicated("orders"));
Assert.False(db.DeregisterReplicated("ghost"));
}
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) =>
(await db.QueryAsync(
"SELECT needs_snapshot FROM __localdb_peer_state WHERE id = 1", r => r.GetInt64(0)))[0] != 0;
}
ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SnapshotResyncTests.cs
+52 -3
View File
@@ -67,10 +67,15 @@ public sealed class SnapshotResyncTests : IDisposable
return new Side(db, store, applier, session);
}
private static (SyncDuplex A, SyncDuplex B) DuplexPair()
// capacity models a transport that applies backpressure (which the gRPC duplex now is, via its
// bounded inbox); null is the historical unbounded pipe.
private static (SyncDuplex A, SyncDuplex B) DuplexPair(int? capacity = null)
{
var aToB = Channel.CreateUnbounded<SyncMessage>();
var bToA = Channel.CreateUnbounded<SyncMessage>();
Channel<SyncMessage> New() => capacity is int c
? Channel.CreateBounded<SyncMessage>(c)
: Channel.CreateUnbounded<SyncMessage>();
var aToB = New();
var bToA = New();
var a = new SyncDuplex { Send = (m, ct) => aToB.Writer.WriteAsync(m, ct).AsTask(), Inbox = bToA.Reader };
var b = new SyncDuplex { Send = (m, ct) => bToA.Writer.WriteAsync(m, ct).AsTask(), Inbox = aToB.Reader };
return (a, b);
@@ -170,6 +175,50 @@ public sealed class SnapshotResyncTests : IDisposable
IsTombstone = false,
};
[Fact]
public async Task MutualSnapshot_OverABoundedTransport_ConvergesWithoutDeadlock()
{
// BOTH sides owe a snapshot — the ordinary outcome of enabling replication late on a pair,
// since baselining flags needs_snapshot on whichever node baselined and both nodes do.
//
// The snapshot used to be streamed from INSIDE the handshake, before the receive loop
// existed, so neither side drained its inbox while sending. That exchange completed only
// because the inbox was unbounded and could hold the peer's ENTIRE snapshot in memory —
// which is the memory profile this design gave up on. Over a transport that pushes back
// (which the gRPC duplex now is, with its bounded inbox) the old shape deadlocks outright.
// Streaming the snapshot alongside an already-running receive loop is what makes both the
// bound and this test possible.
var options = new ReplicationOptions { MaxBatchSize = 2, FlushInterval = TimeSpan.FromMilliseconds(20) };
var a = await NewSideAsync(options);
var b = await NewSideAsync(options);
const int rowsPerSide = 60;
for (var i = 1; i <= rowsPerSide; i++)
{
await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i });
await b.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'B', @id)", new { id = 100 + i });
}
await a.Store.SetNeedsSnapshotAsync(true, default);
await b.Store.SetNeedsSnapshotAsync(true, default);
// Capacity 1 each way: the tightest possible bound, so a side that stops draining its inbox
// to send stalls at once instead of getting away with it on buffer slack.
var (da, db) = DuplexPair(capacity: 1);
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(a.Db)).Count == 2 * rowsPerSide
&& (await ReadOrders(b.Db)).Count == 2 * rowsPerSide,
RunTimeout);
Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db));
cts.Cancel();
await SwallowAsync(runA);
await SwallowAsync(runB);
}
[Fact]
public async Task WipedPeer_AfterEverythingWasAckedAndPruned_StillGetsSnapshot()
{