Buffered, concurrently-flushed delivery: TLA+-verified, vectorized, Postgres-backed state - #21
Merged
Merged
Conversation
Extends the spec ahead of implementation (design-first verification): BufferRead/FlushStart/FlushCommit split so TLC explores CDC reads racing in-flight flushes (the buffer-swap design), dual position tracking (persisted cursor vs in-memory bufferedThrough), and a worker-pool model via interleaved per-destination flushes with the in-flight guard keeping per-destination flushes serial. Crash model is now finer-grained: ProcessCrash (lose all buffers, keep persisted state) and FlushFail (destination txn rollback + drop-buffer recovery) are checked UNCONDITIONALLY — TLC proves buffer loss can never violate safety. Only CrashDuringFlush (destination commit lands, process dies before cursor persist) keeps the everCrashed conditioning, preserving the original spec's precisely stated at-least-once limitation. Two new invariants: BufferPositionBound (cursor <= bufferedThrough <= srcSnap, always) and FlushStateConsistency. All 7 invariants hold: 26,753,473 distinct states (251.8M generated, depth 20, ~3 min), up from 730K in the unbuffered model. TLC also pinned down a non-obvious rule the implementation must follow: a failed flush discards the LIVE buffer too and resets bufferedThrough to the persisted cursor — keeping it would leave a coverage gap over (cursor, inflightThrough].
Replaces the to_pylist() + Python row loops with Arrow compute kernels (which release the GIL — prerequisite for the threaded flush milestone): - _resolve_preimages: preimage↔postimage pairing is a hash join on rowid; classification (orphan / cross-tenant / same-tenant) and the change_type rewrite are kernels. Null-safe routing comparison; a typed __matched marker disambiguates join misses from genuinely-null routing values; last-postimage-wins preserved for out-of-contract duplicate rowids. The routing-mutation ERROR log is aggregated (one line per batch with count + sample rowids, not one per row). - _resolve_conflicts: per-rowid presence flags via group_by, joined back; drop rules and the metric increment-per-conflicting-rowid match the predecessor exactly. - Router.split_and_count: one index_in pass + stable sort + contiguous takes, replacing the filter-per-destination loop. - _apply_changes: deletes chunked at 1,000 keys per delete() call (same transaction) — kills the O(rows) Or(And(...)) expression tree on composite keys. Equivalence with the predecessors is locked by tests/unit/test_phase_equivalence.py: frozen copies of the old row loops serve as oracles against randomized CDC batches (conflicts, orphans, cross-tenant mutations, null routing values, shuffled order), asserting identical output including row order. 103 tests. Perf (tests/perf, before → after): router 100K rows / 1000 dests: 0.156s → 0.010s (15.6x) router 1M rows / 10K dests: 15.575s → 0.153s (102x) _resolve_preimages 50K rows: 0.054s → 0.005s (11x) _resolve_conflicts 50K rows: 0.044s → 0.006s (7x)
…act guards
From the milestone's QE + architect reviews:
- REAL BUG (found by writing the missing chunked-delete tests): the
composite-key delete filter's Or-chain was right-deep and to_sql()
recurses once per node — a single full 1000-row chunk blew Python's
recursion limit. Now a balanced tree-reduce (depth log2(n)). Locked
by tests/integration/test_chunked_delete_integration.py: boundary
sizes around _DELETE_CHUNK_ROWS, composite keys spanning chunks,
NULL keys across chunks, delete+upsert single-transaction.
- viaduck/arrowutil.py: row_indices/full_bool composed from C kernels
(pa.nulls + fill_null + cumulative_sum) replacing pa.array(range(n))
/ [True]*n, which iterated n Python objects under the GIL — exactly
what the threaded-flush milestone can't afford.
- Phase 1 preserves change_type's exact dtype (string vs large_string)
so buffered delivery can concat_tables Phase-1 outputs across reads.
- Null rowids now fail fast in both phases (deliberate divergence:
Arrow joins don't match null keys, so contract-violating input would
be silently misclassified; the predecessor's dicts keyed None).
- Router: _typed_value_set casts string values directly to the column
type (no datetime round-trip — preserves sub-microsecond timestamps,
decimals); routing values that collide after conversion ("1"/"01" on
an int column) are rejected (deliberate divergence: the predecessor
silently double-delivered).
- Equivalence suite grown to 112 tests: duplicate-postimage last-wins,
duplicate preimages, multi update pairs, null-mutation directions,
int/bool typed router columns, metric counter-delta parity locks.
- Perf: matched-pairs preimage benchmark (join-probe-hit path).
Deferred follow-ups (reviewer-acknowledged): last-wins dedup vs
assert, key_columns validation placement, Acero CPU-pool tuning for
the worker-pool milestone.
Replaces the DuckLake-backed _viaduck_state table with plain Postgres rows (by default in the same database that hosts the source catalog's metadata; state.postgres_uri_env overrides). The DuckLake store put an analytical format on an OLTP write pattern: every cursor advance was a catalog commit → snapshot → "new" CDC range → empty read → advance — a treadmill of empty reads and tiny parquet files at poll cadence even with zero source traffic. StateManager keeps its exact public API; an advance is now a single INSERT ... ON CONFLICT DO UPDATE (atomic natively, clears recorded errors, preserves rows_replicated when not supplied). One connection guarded by a lock with a single reconnect retry — cursor writes are per-flush, not per-cycle, so no pool. Table name is validated as a safe identifier (interpolated into DDL). Atomicity vs the destination apply is unchanged: they were always separate transactions; the at-least-once + idempotent-apply story (tla/Viaduck.tla) is untouched. Tests: mock-catalog unit tests replaced by 13 PG-backed integration tests (session-scoped postgres:16-alpine testcontainer): round-trips, instance/destination filtering, error semantics, batch advance, an 8-thread concurrency hammer (worker-pool prep), forced connection-kill reconnect, and plain-SQL inspectability. Deps: +psycopg[binary], +testcontainers (dev).
…uard, boot-race + timeout hardening
From the milestone's QE + architect reviews:
- The zero-config default was dead on arrival: source postgres_uri_env
carries DuckDB's ATTACH format ("postgres:host=H port=P dbname=DB"),
which libpq parses as an invalid keyword. _to_libpq_conninfo() strips
the prefix into valid conninfo; real URIs and bare keyword/value
strings pass through. Tested against the dev-compose format that
exposed it.
- Cursor monotonicity enforced in the store: both advance paths now
carry WHERE last_snapshot_id <= EXCLUDED.last_snapshot_id, dropping
stale acks wholesale (cursor, rows_replicated, and error state all
protected). Matches CursorMonotonicity in tla/Viaduck.tla;
defense-in-depth under the upcoming flush worker pool.
- First-boot CREATE TABLE race: IF NOT EXISTS can still raise a unique
violation when instances bootstrap concurrently — caught and logged;
locked by a 4-way concurrent-bootstrap test.
- Timeouts under the serializing lock: connect_timeout=10s +
statement_timeout=30s so a catalog-PG failover trips the loop's
fatal path instead of hanging every worker silently.
- Test gaps closed: server-side pg_terminate_backend reconnect,
advance-without-initialize (INSERT arm), stale single/batch advance.
- CI dependency review: Python-2.0/0BSD license atoms allowed; purl
exemptions for typing-extensions (PSF in substance; the GPL atom is
CNRI license-chain noise in its PEP 639 expression) and
psycopg/psycopg-binary (LGPL, linked not modified — millpond
precedent), with explanatory comments.
Implements the TLC-verified design (PR 0 / tla/Viaduck.tla): CDC reads at poll cadence, destination writes at flush cadence on a worker pool. viaduck/delivery.py — the core: - Per-destination buffers with dual position tracking: `flushed` (persisted PG cursor) and `position` (in-memory bufferedThrough); the poll thread groups reads by position so ranges stay disjoint. - Flush triggers: interval (default 120s; also lazily persists idle destinations' position-only advances), rows, bytes, memory (global watermark, largest-first), shutdown (drain on SIGTERM). - FlushStart swaps the buffer; an in-flight guard keeps flushes per-destination serial while reads keep landing in the fresh buffer. - FlushFail implements the TLC-pinned rule exactly: discard in-flight AND live buffer, reset position to the persisted cursor — keeping the live buffer would leave a coverage gap over (flushed, through]. - should_pause_reads(): reads pause only when over the watermark with every buffering destination already in flight. viaduck/apply.py — Phase 2 conflict resolution + Phase 3 atomic delete/upsert + retry, extracted from main.py to run on workers (all Arrow kernels + pyducklake — GIL released). main.py keeps Phase 1 + routing and becomes a thin read/buffer/trigger loop. DestinationPool — lock + lease pinning: get() pins until release(), LRU eviction skips pinned entries, force-evict of a pinned entry defers the close to the final release, all-pinned overshoots instead of deadlocking. Config: DeliveryConfig (workers=8, flush_interval_seconds=120, flush_max_rows/bytes, buffer_total_max_bytes), validated; workers=1 + interval=0 reproduces the pre-buffering behavior. Observability: six viaduck_delivery_* metrics, dest_write_seconds continuity, DestStatus buffer_rows/buffer_age_s + "flushing" state. Tests: 17 DeliveryManager unit tests (trigger matrix, swap during in-flight flush, failure semantics, per-dest serialization); 5 end-to-end integration tests with nothing mocked below the delivery API (real catalogs + real PG + real workers), incl. cross-read conflict cancellation and a broken-destination failure path; poll cycle tests rewritten — the full-CDC set runs through a real DeliveryManager in flush-every-cycle mode; pool tests adapted to the lease contract with three new pinning tests.
…light accounting From the milestone's QE + architect reviews. The headline (found independently by both): BufferRead was not atomic against FlushFail. The poll thread snapshots positions, spends seconds in the CDC read, then stamps the position via buffer() — if a flush failure reset the position in that window, the stale write jumped it past the dropped range, which would then never be re-read. Permanent data loss, and exactly the atomicity the TLA spec's BufferRead action assumes. - Per-destination read epochs: read_plan() atomically snapshots (position, epoch); buffer()/advance_position() discard deliveries carrying a stale epoch; _on_flush_failure bumps the epoch with its reset. A discarded read is equivalent to the read never happening — the range is re-read from the rewound position. Locked by a test replaying the exact interleaving. - Hardened failure path: the invariant-restoring reset now runs FIRST; record_error and pool.evict are individually guarded (correlated PG/ pool outages must not leave position ahead of dropped data); a done-callback logs anything escaping _flush at CRITICAL instead of vanishing into an unobserved Future. - Drain loops trigger evaluation until quiet or deadline — rows buffered during an in-flight flush at shutdown get a second pass; abandoned buffers are logged (re-read on restart) and the executor shutdown no longer blocks past the deadline. - In-flight (swapped-out) bytes now count toward the global watermark and the total-bytes gauge — real memory was watermark + workers x flush_max_bytes before. - Pool: connection creation moved outside the pool lock via slot reservation (workers create for different destinations in parallel); pool size configurable as delivery.pool_max_open (default 100). - Bounded advance_cursor retry after a successful destination commit — a transient PG blip no longer triggers the drop-buffer + evict path for a healthy catalog. - Idle position-only persists no longer fire the readiness success signal or write-latency metrics (data flushes only). Deferred (documented for the audit milestone): Arrow/DuckDB CPU-pool sizing under cgroups.
The kill-sequence soak found two delivery bugs. (1) Multiple updates to one key inside a buffered flush window all reached pyducklake's upsert as duplicate join keys, growing duplicate destination rows — upsert candidates now dedupe per key, last write wins (snapshot_id, rowid), selected via take() so NULL keys survive (Acero joins drop them). (2) CDC reads passed the cursor as an inclusive start bound to ducklake's table_changes/table_insertions, re-reading the cursor snapshot every cycle. Normally masked by idempotent upserts, but when the cursor lands exactly on a row's insert snapshot (deterministic after seeding) and the row is deleted in the next window, conflict resolution cancels the re-read insert against the genuine delete — the delete is lost and the seeded copy is a permanent phantom. Reads are now (after_snapshot, end], matching Viaduck.tla's CDCReadFrom; the parameter is renamed after_snapshot so the contract is in the signature. Also: certifi MPL-2.0 exemption (dev-only transitive of testcontainers) and a delivery config section in the dev compose config for the soak.
Closing milestone of the delivery rework. Fresh spec-impl audit (committed); spec gains FlushCommitNoCursor (the commit/cursor gap without process death — cursor-persist failure after a destination commit) and two crash-unconditioned invariants, making the "no data loss path" claim machine-checked (TLC: 9 invariants, 31.4M states). New end-to-end fanout benchmark: flat ~42 destinations/s through 1000 destinations. Seeding now verifies key_columns uniqueness per partition (DuckLake can't enforce it; violations would silently over-delete) at zero extra I/O, failing before the cursor advances. Cursor state moves to a dedicated `viaduck` Postgres schema so bookkeeping never pollutes the ducklake catalog's namespace and a future scoped-down user gets a clean GRANT boundary. Operational status now distinguishes buffering from lagging: between flushes the cursor always trails the source — that's the design working, not lag. Web UI gains a Buffered column, per-value and per-column tooltips, mode/flush-interval display, and a `just demo` recipe stands up the stack with live traffic. Grafana dashboard and Prometheus datasource are now actually provisioned in the compose stack (they never were). README/AGENT.md rewritten for buffered delivery, including the full error-states table and the honest permanent-failure caveat. Review fixes: pool eviction closes catalogs outside the lock, dead web.port and main.py constants removed, clean-exit-after-drain semantics documented.
jghoman
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Rework of the delivery path, spec-first: CDC reads stay at poll cadence, destination writes move to a flush cadence (1–5 min) served by a concurrent worker pool. Scales fanout from tens of destinations to 1000+ (measured flat ~42 dests/s end-to-end) without changing the correctness model: per-destination cursors, idempotent delete+upsert, at-least-once.
Nine commits, one milestone each, every milestone gated by Lead QE + architect review:
Viaduck.tlaextended withBufferRead/FlushStart/FlushCommit/FlushFail/CrashDuringFlush/ProcessCrashbefore any implementation. Buffer-loss crashes checked unconditionallyviaduckschema; monotonicity-guarded upsertsflushed≤position), interval/rows/bytes/memory/shutdown triggers, buffer-swap flushes on a worker pool, in-flight guard, read epochs (restore the spec's atomicBufferReadagainst concurrent failure resets), drain-on-SIGTERM, pool lease pinning(after_snapshot, end], matching the specFlushCommitNoCursormodeled (commit/cursor gap without process death), crash-unconditionedNoDataLoss/PartitionCorrectnessinvariants, end-to-end fanout bench, seed-time key-uniqueness verification, README/AGENT.md rewrite, web UI status semantics (buffering ≠ lagging) + tooltips,just demo, grafana provisioning fixVerification
NoDataLossEvenAfterCrashandPartitionCorrectnessEvenAfterCrashhold even through commit/cursor-gap windows — the "no data loss path" claim is machine-checked, not arguedspec-impl-audit.md): every spec action and invariant mapped to code; documented refinements (read epochs, Winner tiebreak); the inclusive-bounds bug recorded as a conformance-gap case study (TLC verifies the spec — the bug lived between spec and code and fell only to the soak)Known limitations (documented, precisely stated)
haltedstate + circuit breaker logged as follow-upRollback
Revert the branch; nothing is deployed anywhere yet.
workers: 1, flush_interval_seconds: 0reproduces pre-buffering behavior if needed at runtime.