docs(localdb): README + scadaproj index row + design-doc deltas; pack-verified 3 nupkgs @ 0.1.0
Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts
This commit is contained in:
Joseph Doherty
+207
-4
@@ -1,8 +1,211 @@
|
||||
# ZB.MOM.WW.LocalDb
|
||||
|
||||
An embedded SQLite local cache with optional bidirectional async 2-node replication
|
||||
over gRPC, for the ZB.MOM.WW SCADA / OT family. It gives an application a fast,
|
||||
crash-safe local store that can optionally keep a second node in sync asynchronously,
|
||||
without requiring a central database on the hot path.
|
||||
An embedded **SQLite local cache** with **optional bidirectional async 2-node
|
||||
replication over gRPC**, for the ZB.MOM.WW SCADA / OT family. It gives an application a
|
||||
fast, crash-safe local store that can optionally keep a second node in sync
|
||||
asynchronously — local writes **never** block on the peer, and a node that never enables
|
||||
replication is simply a fast local DB.
|
||||
|
||||
Change capture uses SQLite triggers writing a single HLC-stamped oplog inside the
|
||||
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.0`)
|
||||
|
||||
| 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.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: **built, not yet published to the Gitea feed; no app has adopted it yet.**
|
||||
|
||||
## Quickstart
|
||||
|
||||
### Local-only (no replication)
|
||||
|
||||
```csharp
|
||||
builder.Services.AddZbLocalDb(builder.Configuration, db =>
|
||||
{
|
||||
// onReady: create your own tables, then opt them into (future) replication.
|
||||
db.ExecuteAsync(
|
||||
"CREATE TABLE IF NOT EXISTS widget (id INTEGER PRIMARY KEY, name TEXT, qty INTEGER)")
|
||||
.GetAwaiter().GetResult();
|
||||
db.RegisterReplicated("widget"); // PK required; installs capture triggers
|
||||
});
|
||||
|
||||
// Consume ILocalDb anywhere:
|
||||
await db.ExecuteAsync("INSERT INTO widget (id, name, qty) VALUES (@id, @n, @q)",
|
||||
new { id = 1, n = "bolt", q = 42 });
|
||||
var rows = await db.QueryAsync("SELECT name, qty FROM widget",
|
||||
r => (Name: r.GetString(0), Qty: r.GetInt32(1)));
|
||||
```
|
||||
|
||||
`AddZbLocalDb` is first-registration-wins: a second call is a no-op. `RegisterReplicated`
|
||||
installs capture triggers even without the replication package present — you can register
|
||||
tables now and turn on sync later.
|
||||
|
||||
### Replicated (2 nodes)
|
||||
|
||||
Every node registers the passive side (so either can accept a stream) and the initiator
|
||||
(idle unless a `PeerAddress` is set). Configure exactly **one** side with the peer's
|
||||
address; the other stays passive.
|
||||
|
||||
```csharp
|
||||
// Both nodes:
|
||||
builder.Services.AddZbLocalDb(builder.Configuration, db => { /* create + RegisterReplicated */ });
|
||||
builder.Services.AddGrpc();
|
||||
builder.Services.AddZbLocalDbReplication(builder.Configuration); // passive host + initiator + maintenance
|
||||
|
||||
var app = builder.Build();
|
||||
app.MapZbLocalDbSync(); // passive gRPC endpoint — host must have AddGrpc()
|
||||
app.Run();
|
||||
```
|
||||
|
||||
```jsonc
|
||||
// Node A — the INITIATOR: dials node B
|
||||
"LocalDb": {
|
||||
"Path": "nodeA.db",
|
||||
"Replication": { "PeerAddress": "https://nodeB:5001", "ApiKey": "…" }
|
||||
}
|
||||
// Node B — PASSIVE: no PeerAddress, its SyncBackgroundService idles and only serves the endpoint
|
||||
"LocalDb": { "Path": "nodeB.db", "Replication": { } }
|
||||
```
|
||||
|
||||
Both directions flow on the single bidirectional stream the initiator opens; the passive
|
||||
node still pushes its own deltas back. Only one side needs `PeerAddress`.
|
||||
|
||||
## Configuration
|
||||
|
||||
### `LocalDb` section (`LocalDbOptions`)
|
||||
|
||||
| Key | Default | Meaning |
|
||||
|---|---|---|
|
||||
| `Path` | `""` (required) | Filesystem path to the SQLite database file. |
|
||||
| `BusyTimeoutMs` | `5000` | SQLite `busy_timeout` pragma, in milliseconds. |
|
||||
| `Synchronous` | `"NORMAL"` | SQLite `synchronous` pragma (`OFF` / `NORMAL` / `FULL`). |
|
||||
|
||||
### `LocalDb:Replication` section (`ReplicationOptions`)
|
||||
|
||||
| Key | Default | Meaning |
|
||||
|---|---|---|
|
||||
| `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). |
|
||||
| `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. |
|
||||
| `ReconnectBackoffMax` | `60 s` | Upper bound on exponential reconnect backoff (validator caps ≤ 1 day). |
|
||||
| `MaintenanceInterval` | `60 s` | How often maintenance enforces caps + prunes expired tombstones. |
|
||||
| `MaxHlcDriftAhead` | `5 min` | Max tolerated HLC drift where the peer's clock runs ahead of ours. |
|
||||
| `FailClosedOnDrift` | `false` | When true, an over-drift entry is dead-lettered instead of applied. |
|
||||
|
||||
All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
|
||||
|
||||
## How it works
|
||||
|
||||
- **Capture** — three AFTER triggers per registered table write each change into one
|
||||
shared, HLC-stamped `__localdb_oplog` plus a `__localdb_row_version` ledger, in the
|
||||
consumer's own transaction. A DB-scoped `__localdb_applying` guard makes replicated
|
||||
applies skip re-capture.
|
||||
- **Convergence** — last-writer-wins: compare incoming HLC, then `node_id` ordinal as the
|
||||
tie-break; the strict-greater winner is applied, the loser discarded (but still acked).
|
||||
- **Streaming** — one long-lived bidirectional gRPC stream. The initiator dials; both
|
||||
sides push oplog batches above the peer's watermark and ack the applied watermark back,
|
||||
which drives pruning. Outbound uses a single writer fed by two lanes (an unbounded
|
||||
control lane for handshake/acks that always preempts a bounded data lane for
|
||||
delta/snapshot batches), so receiving never blocks on the ability to ack.
|
||||
- **Resync** — when a peer's watermark falls below the pruned oplog horizon, it is sent a
|
||||
full snapshot of the registered tables (streamed, keyset-paged) and deltas resume above
|
||||
the snapshot's as-of seq. **Each node decides from its OWN state** whether it owes the
|
||||
peer a snapshot.
|
||||
- **Maintenance** — a background service on **both** roles enforces the oplog row/age caps
|
||||
and prunes expired tombstones every `MaintenanceInterval`.
|
||||
|
||||
## Semantics worth knowing
|
||||
|
||||
- **PK-column updates** are captured as delete-of-old-PK + insert-of-new — a tombstone on
|
||||
the old key plus a fresh insert (full delete+insert support).
|
||||
- **Upserts** (`INSERT … ON CONFLICT DO UPDATE`) are fully supported (the row-version
|
||||
trigger uses an explicit `ON CONFLICT` clause so it is not rewritten by the outer
|
||||
statement's conflict algorithm).
|
||||
- **Clock-drift guard** — warn-once per session when a peer entry's HLC physical time is
|
||||
more than `MaxHlcDriftAhead` ahead. With `FailClosedOnDrift`, each over-drift entry is
|
||||
dead-lettered; otherwise it is applied (the HLC merge absorbs the skew). The **session
|
||||
never dies from drift** — no livelock on reconnect.
|
||||
- **Writes never block on the peer**; local commit is independent of replication.
|
||||
- **All timestamps are UTC** (HLC physical component is UTC Unix milliseconds).
|
||||
|
||||
## Limitations (honest list)
|
||||
|
||||
- **Exactly two nodes.** No mesh / N-node topologies.
|
||||
- **LWW only.** No per-table pluggable conflict resolvers.
|
||||
- **BLOB columns are rejected at registration** (a column declared `BLOB` fails
|
||||
`RegisterReplicated`, because `json_object` cannot capture it). A *typeless* column can
|
||||
still hold a BLOB value at runtime, which then surfaces as a write-time error, not a
|
||||
registration error.
|
||||
- **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.
|
||||
- **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).
|
||||
|
||||
## Observability
|
||||
|
||||
Metrics publish under the single meter **`ZB.MOM.WW.LocalDb.Replication`** (all instruments
|
||||
null-guarded — replication runs identically whether or not metrics are collected):
|
||||
|
||||
| Instrument | Kind | Notes |
|
||||
|---|---|---|
|
||||
| `localdb.oplog.depth` | ObservableGauge | Unacked oplog backlog (polled cheaply at scrape time). |
|
||||
| `localdb.sync.lag.seconds` | ObservableGauge | Seconds since the last successful sync (no measurement when never synced). |
|
||||
| `localdb.sync.applied` | Counter | Rows applied from inbound **delta** batches. |
|
||||
| `localdb.sync.conflicts_discarded` | Counter | Inbound rows that lost LWW. |
|
||||
| `localdb.sync.dead_lettered` | Counter | Poison / over-drift rows routed to the dead-letter table. |
|
||||
| `localdb.sync.reconnects` | Counter | Reconnect attempts (initiator). |
|
||||
| `localdb.sync.snapshots` | Counter | Snapshots, tagged `direction=sent`/`received` (one increment per snapshot). |
|
||||
|
||||
**Counter split to note:** rows merged from a **snapshot** are *not* counted under
|
||||
`localdb.sync.applied`; a completed snapshot instead increments `localdb.sync.snapshots`
|
||||
once (with its `direction` tag). Only delta-batch rows increment `localdb.sync.applied`.
|
||||
|
||||
`ISyncStatus` (registered singleton) surfaces live state for dashboards / health:
|
||||
`Connected` (true while **any** session is active — a node can run both roles at once),
|
||||
`PeerNodeId`, `LastSyncUtc`, `OplogBacklog` (**nullable** — null means unknown because no
|
||||
provider is wired or the poll threw, so a DB failure never reads as "0 backlog"), and
|
||||
`ConnectionAttempts`.
|
||||
|
||||
## Ops notes
|
||||
|
||||
- When the oplog row/age caps are exceeded, maintenance prunes to the ceiling and flags
|
||||
`needs_snapshot` (disk safety over delta continuity); the peer then snapshot-resyncs.
|
||||
- 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.
|
||||
- **Peer auth**: `ApiKey` is sent by the initiator as a static `Authorization: Bearer …`
|
||||
header on its channel. The library's passive service does **not** itself verify the key
|
||||
— the passive host enforces whatever gRPC auth the host application configures (e.g.
|
||||
`ZB.MOM.WW.Auth`). Configure host-side authorization to actually gate inbound streams.
|
||||
|
||||
## 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`).
|
||||
|
||||
Build / test from this directory:
|
||||
|
||||
```bash
|
||||
dotnet build ZB.MOM.WW.LocalDb.slnx
|
||||
dotnet test ZB.MOM.WW.LocalDb.slnx
|
||||
```
|
||||
|
||||
See the design doc: [`docs/plans/2026-07-17-localdb-design.md`](../docs/plans/2026-07-17-localdb-design.md).
|
||||
|
||||
+11
-2
@@ -18,6 +18,7 @@ public static class LocalDbServiceCollectionExtensions
|
||||
/// caller. This is where consumers create their tables (via the passed <see cref="ILocalDb"/>) and call
|
||||
/// <see cref="ILocalDb.RegisterReplicated"/> to opt them into replication.
|
||||
/// </param>
|
||||
/// <remarks>Subsequent calls are no-ops if an <see cref="ILocalDb"/> registration already exists (first registration wins).</remarks>
|
||||
public static IServiceCollection AddZbLocalDb(
|
||||
this IServiceCollection services,
|
||||
IConfiguration configuration,
|
||||
@@ -44,8 +45,16 @@ public static class LocalDbServiceCollectionExtensions
|
||||
catch
|
||||
{
|
||||
// A throwing callback must not leak the open master connection: MS.DI does not
|
||||
// dispose instances a failed factory never returned.
|
||||
db.Dispose();
|
||||
// dispose instances a failed factory never returned. Guard the cleanup so a Dispose
|
||||
// failure can never replace the consumer's original onReady exception.
|
||||
try
|
||||
{
|
||||
db.Dispose();
|
||||
}
|
||||
catch
|
||||
{
|
||||
// Swallow: the onReady exception below is the one the caller must see.
|
||||
}
|
||||
throw;
|
||||
}
|
||||
return db;
|
||||
|
||||
@@ -13,6 +13,7 @@ public sealed class HybridLogicalClock
|
||||
_last = initialValue;
|
||||
}
|
||||
|
||||
/// <summary>Returns the next monotonically increasing HLC stamp: the current UTC physical time when it advances the clock, otherwise the last value with its logical counter incremented by one.</summary>
|
||||
public long Next()
|
||||
{
|
||||
lock (_lock)
|
||||
@@ -26,8 +27,12 @@ public sealed class HybridLogicalClock
|
||||
/// <summary>Merge a peer's HLC so our next stamp orders after everything we've seen.</summary>
|
||||
public void Observe(long remote) { lock (_lock) { if (remote > _last) _last = remote; } }
|
||||
|
||||
/// <summary>The most recently issued/observed HLC value, without advancing the clock.</summary>
|
||||
public long Current { get { lock (_lock) return _last; } }
|
||||
|
||||
/// <summary>Extracts the physical component (UTC Unix milliseconds) from a packed HLC value.</summary>
|
||||
public static long PhysicalMs(long hlc) => hlc >> 16;
|
||||
|
||||
/// <summary>Extracts the 16-bit logical counter from a packed HLC value.</summary>
|
||||
public static int Counter(long hlc) => (int)(hlc & 0xFFFF);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user