Merge pull request #56 from PostHog/jakob/destination-lifecycle

feat: destination lifecycle state machine (active|paused|draining|retired)

Author: jghoman

AGENT.md

@@ -169,7 +169,7 @@ two-step flushes (buffer swap → commit/fail), concurrent per-destination
 flush workers, seeding, and commit/cursor-gap scenarios both with and
 without process death — ALL checked unconditionally; the tombstone rule
 retired the everCrashed phantom conditioning entirely — checking 7
-invariants across 19,886,377 distinct states. Modify the spec when changing
+invariants across 85,012,333 distinct states. Modify the spec when changing
 the CDC algorithm or adding new failure modes — and when designing semantic
 changes, extend the spec FIRST and let TLC pass judgment before implementing.
 Always run `just tlc` after spec changes.
@@ -188,6 +188,7 @@ Always run `just tlc` after spec changes.
 - **Schema projection** (`schema_projection.py`): per-destination drops/reorder/safe-casts onto the destination's existing schema; build-time guards refuse key/routing-column mutation and NOT-NULL-violating casts; per-value cast fallback nulls unparseable values and alarms via `projection_cast_null_fallback_total`
 - **Scan-based seeding with REPLACE semantics**: new destinations bulk-load from a filtered source scan; a cursor-0 destination with leftover rows (crashed prior seed) is truncated first (`routing.seed_truncate`, default true). Configurable via `seed_mode` (default: `scan`)
 - **Worker threads are a concurrency knob, not a CPU multiplier**: Arrow's compute pool and DuckDB's threads are process-global underneath every flush worker — see README "Worker-thread sizing"
+- **Destination lifecycle state machine** (`lifecycle.py` + `<state_table>_lifecycle` table — name derives from the cursor table so pipelines sharing a PG never share intent): per-DESTINATION operator intent — `active | paused | draining | retired`; absent row = active (no backfill), unknown value = paused + one ERROR per transition (not per cycle). Paused/retired = controlled crash: buffer discarded via the FlushFail machinery (`delivery.discard_buffer` rewinds position to the durable cursor + bumps the epoch; counted on `lifecycle_discarded_rows_total`), connection evicted ONLY once `delivery.is_clean` (an in-flight flush's retry loop re-creates the pool entry, so a transition-time evict latch leaks the connection for the stint), cursor is the resume point. A flush that COMPLETES after a discard restores `position >= flushed` in its success path (epoch-bumped) — without it, resume re-reads a committed range (deterministic duplicates in append_only). Draining = no new reads, flush out, evict when clean; the drain-complete log distinguishes "flushed out" from "ended via a flush-failure rewind" (the rewound range was NOT delivered; draining excludes re-reads, so retiring on the latter abandons it). RETIRED IS NEVER WRITTEN BY CODE (`StateManager.set_lifecycle_state` refuses) and viaduck SEVERS the cursor rows (all instances, idempotent per cycle + startup backstop against the in-flight-upsert resurrect race) — re-add = new tenant = fresh seed per `seed_mode`, deterministic regardless of partition drift. Seeding is lifecycle-gated: only ACTIVE destinations seed; a skipped cursor-0 destination stays read-gated after resume until a restart seeds it (ERROR per cycle). States re-read every poll cycle (live pause without restart); a state-store blip keeps last-known states (the lifecycle table shares the cursor store's PG — fail-to-paused would turn any PG blip into a fleet-wide self-inflicted mass discard). `/lifecycle` (read-only: state + reason/updated_by/updated_at + staleness age) + one-hot `viaduck_destination_lifecycle_state{destination,state}`; `/status` destination status short-circuits to the lifecycle state — join lag alerts on `state!="active"`. TLA: the spec HAS a `PauseDest` action (buffer discard + position rewind, in-flight flush PRESERVED — unlike ProcessCrash/FlushFail) and FlushCommit carries the implementation's success-path position restore; removing the restore lets TLC produce a 6-step BufferPositionBound counterexample (SrcInsert → BufferRead → FlushStart → PauseDest → FlushCommit), which is the formal witness for the pause-races-in-flight-flush duplicate-delivery bug. Paused DURATION needs no modeling (an action not firing is a pause; resume is BufferRead from the rewound position). The lifecycle table's DDL CHECK freezes the state vocabulary: adding a state later requires a migration on existing tables, not just a VALID_STATES change.
 
 ## Module Layout
 

README.md

@@ -317,6 +317,34 @@ Cursor advances are single `INSERT ... ON CONFLICT DO UPDATE` upserts with a mon
 
 State is keyed by `(destination_id, instance_id)`, enabling multiple viaduck instances to independently track their assigned destinations without conflicts.
 
+## Destination Lifecycle (operator runbook)
+
+Each destination has an operator-intent state in `viaduck.<state_table>_lifecycle` (per-destination — pausing pauses it on every instance; the table name derives from `state.table` so pipelines sharing a Postgres never share intent). Absent row = `active`. States re-read every poll cycle, so changes apply live, no restart. Observability: `viaduck_destination_lifecycle_state{destination,state}` (one-hot), the read-only `/lifecycle` endpoint (state + reason/updated_by/updated_at + staleness age), `viaduck_lifecycle_discarded_rows_total`, and `/status` destination status strings (`paused`/`draining`/`retired` short-circuit `lagging` — join lag alerts against `viaduck_destination_lifecycle_state{state!="active"}` so intentional pauses don't page).
+
+| state | effect |
+|---|---|
+| `active` | normal delivery |
+| `paused` | no reads, no flushes; buffer discarded (durable in source), connection released, cursor = resume point. Resume is gap-free (crash-recovery re-read) |
+| `draining` | no new reads; buffered data flushes out, then the connection is released. Reversible. **Check the drain-complete log line**: "drain complete (flushed out)" is a clean drain; "drain ended via a flush-failure rewind" means the read-but-unflushed range was NOT delivered — resume to re-read it before retiring |
+| `retired` | terminal. Excluded at startup; cursor rows are severed (all instances), so **re-add = new tenant = fresh seed** per `seed_mode` |
+
+```sql
+-- Pause (live, applies within one poll cycle):
+INSERT INTO viaduck.viaduck_state_lifecycle (destination_id, state, reason, updated_by, updated_at)
+VALUES ('team-2', 'paused', 'RDS maintenance', 'jakob', now())
+ON CONFLICT (destination_id) DO UPDATE
+  SET state = EXCLUDED.state, reason = EXCLUDED.reason,
+      updated_by = EXCLUDED.updated_by, updated_at = EXCLUDED.updated_at;
+
+-- Resume: same statement with state = 'active'.
+-- Drain (pre-retirement): state = 'draining'; wait for the drain-complete log.
+-- Retire (the explicit human ack — viaduck code REFUSES to write this value):
+--   state = 'retired'. Cursor rows are deleted by viaduck when it observes
+--   the state; re-adding the destination later re-seeds from scratch.
+```
+
+A destination paused while it has never been seeded (cursor 0) skips seeding at startup and stays read-gated after resume until a restart seeds it (loud ERROR each cycle).
+
 ## New Destination Seeding
 
 When a new destination is added to the config, it needs the current source data. Three modes are available via `routing.seed_mode`:

tests/integration/test_state_integration.py

@@ -229,3 +229,83 @@ def _boot(mgr, n):
     for m in managers:
         m.close()
     assert not errors
+
+
+# ---------------------------------------------------------------------------
+# Destination lifecycle (viaduck/lifecycle.py semantics live in the tracker;
+# these pin the StateManager's storage contract against real Postgres)
+# ---------------------------------------------------------------------------
+
+
+def test_lifecycle_round_trip_and_isolation(sm):
+    sm.initialize_destinations(["d1"])
+    # Absent rows: empty result (tracker normalizes absent -> active).
+    assert sm.load_lifecycle_states(["d1", "d2"]) == {}
+
+    sm.set_lifecycle_state("d1", "paused", reason="broken catalog", updated_by="test")
+    assert sm.load_lifecycle_states(["d1"]) == {"d1": "paused"}
+    rows = sm.load_lifecycle_rows(["d1"])
+    assert rows["d1"]["state"] == "paused"
+    assert rows["d1"]["reason"] == "broken catalog"
+    assert rows["d1"]["updated_by"] == "test"
+    assert rows["d1"]["updated_at"] is not None
+
+    # Update in place (upsert path).
+    sm.set_lifecycle_state("d1", "active", reason="fixed", updated_by="test2")
+    assert sm.load_lifecycle_states(["d1"]) == {"d1": "active"}
+
+
+def test_lifecycle_table_is_per_pipeline(pg_uri, state_table_name):
+    # The lifecycle table derives from the cursor table name — two
+    # pipelines with colliding destination ids must not share operator
+    # intent (review finding: shared hard-coded table let one pipeline's
+    # pause hit the other).
+    a = StateManager(pg_uri, "i1", StateConfig(table=state_table_name))
+    b = StateManager(pg_uri, "i1", StateConfig(table=state_table_name + "_other"))
+    try:
+        a.initialize_destinations(["d1"])
+        b.initialize_destinations(["d1"])
+        a.set_lifecycle_state("d1", "paused", reason="pipeline A only", updated_by="test")
+        assert a.load_lifecycle_states(["d1"]) == {"d1": "paused"}
+        assert b.load_lifecycle_states(["d1"]) == {}
+    finally:
+        a.close()
+        b.close()
+
+
+def test_lifecycle_check_constraint_rejects_unknown_state(sm, pg_uri, state_table_name):
+    sm.initialize_destinations(["d1"])
+    # Direct SQL (bypassing the code-level guard): the DB CHECK holds the
+    # vocabulary, which is what makes the tracker's unknown-state paused
+    # fallback forward-compat-only rather than a live path.
+    with psycopg.connect(pg_uri, autocommit=True) as conn:
+        with pytest.raises(psycopg.errors.CheckViolation):
+            conn.execute(
+                f"INSERT INTO viaduck.{state_table_name}_lifecycle "
+                "(destination_id, state, updated_at) VALUES ('d1', 'frobnicated', now())"
+            )
+
+
+def test_retirement_severs_cursor_rows_for_all_instances(pg_uri, state_table_name):
+    # Two instances own rows for the same destination (partitioning drift
+    # over time); retirement is per-destination and must sever both, so a
+    # re-add seeds fresh regardless of which instance picks it up.
+    i1 = StateManager(pg_uri, "i1", StateConfig(table=state_table_name))
+    i2 = StateManager(pg_uri, "i2", StateConfig(table=state_table_name))
+    try:
+        i1.initialize_destinations(["d1"])
+        i2.initialize_destinations(["d1"])
+        i1.advance_cursor("d1", snapshot_id=9, cumulative_rows=10)
+
+        deleted = i1.delete_destination_state("d1")
+        assert deleted == 2
+        assert i1.load_cursors(["d1"]) == {}
+        assert i2.load_cursors(["d1"]) == {}
+        # Idempotent (the per-cycle resurrect-race sweep).
+        assert i1.delete_destination_state("d1") == 0
+        # Re-add: initialize creates a FRESH cursor-0 row -> re-seed path.
+        i1.initialize_destinations(["d1"], initial_snapshot_id=0)
+        assert i1.load_cursors(["d1"])["d1"].last_snapshot_id == 0
+    finally:
+        i1.close()
+        i2.close()

tests/unit/test_delivery.py

@@ -551,3 +551,104 @@ def test_on_flush_success_fires_for_data_not_for_idle_persists():
         mgr.maybe_flush()
         assert mgr.wait_idle()
     assert hits == [1]
+
+
+# ---------------------------------------------------------------------------
+# Destination lifecycle hooks (viaduck/lifecycle.py)
+# ---------------------------------------------------------------------------
+
+
+def test_discard_buffer_rewinds_position_and_bumps_epoch():
+    mgr, _, _ = _manager(cursors={"d1": 5})
+    plan = mgr.read_plan()
+    pos, epoch = plan["d1"]
+    assert pos == 5
+    mgr.buffer("d1", _table(4), through_snapshot=9, epoch=epoch)
+    assert mgr.positions() == {"d1": 9}
+
+    dropped = mgr.discard_buffer("d1")
+    assert dropped == 4
+    # Position rewound to the durable cursor — the discarded range will be
+    # re-read on resume (controlled-crash semantics, same as FlushFail).
+    assert mgr.positions() == {"d1": 5}
+    # A read that overlapped the discard is rejected by the epoch guard.
+    mgr.buffer("d1", _table(2), through_snapshot=9, epoch=epoch)
+    assert mgr.status_snapshot()["d1"].buffer_rows == 0
+    assert mgr.positions() == {"d1": 5}
+
+
+def test_discard_buffer_noop_when_clean():
+    mgr, _, _ = _manager(cursors={"d1": 5})
+    assert mgr.discard_buffer("d1") == 0
+    # Epoch untouched on the no-op path: an in-flight read may still land.
+    _, epoch = mgr.read_plan()["d1"]
+    mgr.buffer("d1", _table(1), through_snapshot=6, epoch=epoch)
+    assert mgr.status_snapshot()["d1"].buffer_rows == 1
+
+
+def test_suspended_destination_never_flushes():
+    mgr, _, _ = _manager()
+    fake, calls = _recording_flush(mgr)
+    with patch.object(mgr, "_flush", fake):
+        mgr.buffer("d1", _table(3), through_snapshot=7)
+        mgr.set_suspended({"d1"})
+        assert mgr.maybe_flush(shutdown=True) == 0
+        assert calls == []
+        # Unsuspend: the same trigger now fires.
+        mgr.set_suspended(set())
+        assert mgr.maybe_flush(shutdown=True) == 1
+        assert calls[0][0] == "d1"
+
+
+def test_is_clean_tracks_buffer_and_position():
+    mgr, _, _ = _manager(cursors={"d1": 5})
+    assert mgr.is_clean("d1")
+    _, epoch = mgr.read_plan()["d1"]
+    mgr.buffer("d1", _table(2), through_snapshot=8, epoch=epoch)
+    assert not mgr.is_clean("d1")
+    mgr.discard_buffer("d1")
+    assert mgr.is_clean("d1")
+
+
+def test_position_only_advance_is_not_clean():
+    # An advanced position with no data still means the durable cursor is
+    # behind (a lazy persist is pending) — draining must wait for it.
+    mgr, _, _ = _manager(cursors={"d1": 5})
+    _, epoch = mgr.read_plan()["d1"]
+    mgr.advance_position("d1", 9, epoch=epoch)
+    assert not mgr.is_clean("d1")
+
+
+def test_flush_success_after_discard_restores_position():
+    # Review finding: pause racing an in-flight flush left position <
+    # flushed after the flush committed — on resume the already-applied
+    # range was re-read and re-applied (deterministic duplicates in
+    # append_only). The success path must restore position >= through.
+    import threading
+
+    mgr, sm, _ = _manager(cursors={"d1": 5})
+    _, epoch = mgr.read_plan()["d1"]
+    mgr.advance_position("d1", 9, epoch=epoch)  # position-only: no data write path
+
+    gate = threading.Event()
+    entered = threading.Event()
+
+    def _blocking_advance(dest_id, through, cumulative, attempts=3):
+        entered.set()
+        gate.wait(timeout=10)
+        sm.advance_cursor(dest_id, through, cumulative)
+
+    with patch.object(mgr, "_advance_cursor_with_retry", _blocking_advance):
+        assert mgr.maybe_flush(shutdown=True) == 1  # real _flush, empty tables
+        assert entered.wait(timeout=10)
+        # Lifecycle pause lands mid-flush: rewinds position to flushed=5.
+        mgr.discard_buffer("d1")
+        assert mgr.positions()["d1"] == 5
+        gate.set()
+        assert mgr.wait_idle(timeout_s=10)
+
+    # Flush succeeded through 9: position restored, no re-read of (5, 9].
+    assert mgr.positions()["d1"] == 9
+    snap = mgr.status_snapshot()["d1"]
+    assert snap.flushed_snapshot == 9
+    assert mgr.is_clean("d1")