@nodii/replica-consumer (TS)

Pattern 3 replication consumer for the Nodii microservice stack: defineReplica

• startReplicaConsumer + dedupe + LWW + apply-tx wrapping + DLQ + snapshot-RPC bootstrap + drift sweeper. Polyglot per D159 — TS, Python (nodii-replica-consumer) and Go (go/replica-consumer) ship in parity.

Spec: planning hub serviceId=nodii-libs, docKey=replica-consumer.

This file previously held the unmodified ts/_template README, which claimed the package was "private": true scaffolding. It is not — it is published, and README.md is in "files", so that text shipped to npm.

Worker liveness

A worker handle (startStreamsWorker's result, and each workers[i].handle of startReplicaConsumer's) reports whether its consume loop is actually consuming. leaseAcquired cannot: on a live handle it stays true for the life of the worker.

| | TS | Python | Go | | --- | --- | --- | --- | | loop running | loopAlive | loop_alive | LoopAlive() | | holds the single-active lease now | leaseHeld | lease_held | LeaseHeld() | | last XREADGROUP returned at | lastReadAt (Date \| null) | last_read_at (UTC datetime \| None) | LastReadAt() (zero until the first read) |

• A lease blip does not end the loop. While the lease is lost the loop pauses (leaseHeld false), and it resumes when the lease self-heals. Both transitions are logged at WARN with the stream (lease lost, lease reacquired).
• The loop ends only on stop(), a poison-flood trip or a fatal escalation. Any other ending is a crash: the worker logs a FATAL line at ERROR, closes its blocking connection and releases its lease so a peer can take over. That teardown is bounded: the release gets 5 s and is then abandoned (the key expires with its TTL), so a wedged Redis cannot hang it. A release that fails or times out writes one stderr line, [replica-consumer] LEASE RELEASE FAILED at dead-loop teardown (bounded at <bound>; the key expires with its TTL): <error> lease=<key> service=<service> stream=<stream>.
• A stalled loop is loopAlive && leaseHeld with lastReadAt older than blockMs plus some slack (a hung apply, a stuck connection). Restarting the worker is the right response.
• A boot-race loser reads false / null until it converges.
• In TS the three members are optional on the WorkerHandle interface, so an existing hand-built test double or wrapper typed as WorkerHandle still compiles. The library's own handles always populate them. An absent member (undefined) means the handle does not report it: treat it as unknown, not as dead. Compare with === false / === null rather than by falsiness.
• The logger and metric-sink calls made from the lease callbacks (the lease.lost_total / lease.reacquired_total counters and their WARN lines), lease.acquired_total at live start, the boot-race start-failure WARN and the dead-loop teardown's ERROR line and crash counter are isolated in every port: a sink that throws, returns a rejected promise (an async sink: the rejection is caught, not awaited, never left unhandled) or panics there cannot crash the process or stop the lease from self-healing. A failed call is not silent: it writes one stderr line, [replica-consumer] SINK FAILED at <site> (isolated; the caller carried on): <error> service=<service> stream=<stream>. On the consume loop, the calls on its Redis-error paths are isolated the same way: the read-error counter and WARN, the group re-ensure WARN and the XCLAIM reaper WARN. The loop's other sink calls (the apply path, and the catch-all consume loop error WARN for a non-read failure, whose throw ends the loop through the dead-loop teardown) are not.
• A failing XREADGROUP is retried with a bounded backoff (200 ms doubling to 5 s, ±20 %, reset by the next successful read), counted in replica_consumer.read.error_total (counter, {service, stream}) on every failure and logged at WARN (XREADGROUP failed — retrying with backoff) on the first failure of a streak and every 10th after; stop() ends the wait at once. Same in all three ports.

Terminal topics (no base row is a no-op)

Declare a topic in terminalTopics when its event ends the aggregate, for example tenant.tenant.deleted.v1:

defineReplica({
  // ...
  topics: [...FULL, ...PARTIAL],
  partialTopics: PARTIAL,
  terminalTopics: ["tenant.tenant.deleted.v1"],
});

A live event on such a topic that finds no base row on the last attempt of retryMaxAttempts is consumed and ACKed (earlier attempts stay retryable, so a create that is still in the PEL lands first and the delete then applies): consumed_events and the cursor are written in one transaction, the outcome is terminal_no_base_row, an info line is logged and <service>_replica_terminal_no_base_row_total is counted. No row is written, no DLQ row is written and it does not count toward poisonThreshold. "No base row" means the partial-merge probe found none, or the spec's toRow / route threw NoBaseRowForPartial (the shape a multi-table router uses). Any other error on the topic keeps its retry, DLQ and poison handling, and an event whose row exists applies normally. Limits: a create that lands after the absorbed delete (for example re-driven from the DLQ out of order) re-creates the row, so re-drive in stream order; for a multi-table router the decision is per event. Without the option, the old behaviour stands: retry, then DLQ as partial_no_base_row. Python: terminal_topics; Go: TerminalTopics.

Migrations

The three operational tables (consumed_events, replication_cursor, replication_dlq) plus subscription_backsync are library-owned. The replica TABLE itself is service-owned — its column shape is service-specific.

Two supported paths, and they carry the SAME DDL:

• In-process: await applyReplicaConsumerMigrations(pool) — idempotent, sequences all five steps in order.
• External runner (Alembic / Flyway / golang-migrate / sqlx-cli): the raw files ship in the tarball under src/migrations/*.sql. Run them in numeric order. getReplicaConsumerMigrationSQL() returns the same SQL as one string if you would rather feed a runner programmatically.

"files" uses a GLOB for these, never an enumeration — an enumerated list went stale and published 1 of 4 migrations (issue #764). bun run check:publish-migrations at the repo root re-derives the set from disk against npm pack --dry-run and fails if any migration is unreachable from the tarball.

Binding a jsonb replica column

Declare the column and return a natural value from toRow:

defineReplica({
  // …
  jsonbColumns: ["metadata"],
  toRow: (e) => ({
    subscription_id: e.payload.subscription_id as string,
    metadata: e.payload.metadata as Record<string, unknown>, // raw object — correct
    source_entity_version: e.entity_version,
  }),
});

Do not JSON.stringify in toRow. The library serializes declared jsonb columns itself (normalizeJsonbValue) and emits $N::text::jsonb. A pre-serialized string still works — it passes through as JSON text — but the raw object is the intended shape.

On a multi-table router, each route carries its own declaration (routes[].jsonbColumns), and the primary table uses the top-level jsonbColumns. Both are honoured.

Why ::text::jsonb and not ::jsonb

drizzle-orm/postgres-js/driver.js mutates the postgres.js client handed to drizzle() in place, replacing the serializers for jsonb (3802), json (114) and eight date/time OIDs with an identity function. A service that gives ONE client to both drizzle() and this library's raw executor therefore runs with jsonb serialization disabled. An existing ::jsonb cast does not dodge it — postgres.js resolves the OID from the cast or the target column and runs serializers[3802] either way. ::text pins the parameter to OID 25, which drizzle does not patch, and Postgres does the text → jsonb parse server-side.

The full measured matrix (both regimes, real Postgres) is at the top of src/sql-builders.ts. $N::text::jsonb + JSON text is the only combination correct under both.

Binding a timestamptz / timestamp / date replica column

Return an ISO-8601 string from toRow, not a Date.

The same drizzle mutation disables the serializers for the date/time OIDs, and unlike jsonb there is no SQL-side fix: a Date throws under $1, $1::timestamptz and $1::text::timestamptz alike, because the identity serializer fails before any cast is reached. The library therefore emits no date/time cast and applies no value transform (a blanket Date → ISO normalization would break time-typed columns, which accept a Date but reject an ISO datetime). Bind d.toISOString(); for a time column bind 'HH:MM:SS'.

Testing trap — a test that builds its own pool cannot see any of this

The drizzle mutation only happens when drizzle and the raw executor share a client. A test that constructs its own postgres() pool is always in the un-patched regime, so it passes while production breaks. That is exactly how a money-path replica stayed dead for 26 days with a green suite.

The two regimes also fail on different cells of the matrix, so neither file is sufficient alone:

| you break… | jsonb-param.integration.test.ts (own pool) | jsonb-drizzle-shared-client.integration.test.ts (drizzle(sql)) | | --- | --- | --- | | the ::text OID-25 pin | red | green | | the value-side serialization | red | red | | per-route jsonbColumns carry-through | green | red |

If you touch jsonb binding, run both.