From a12396d1e8fad7a97a60a68a0c1a83f1fa15f973 Mon Sep 17 00:00:00 2001 From: Jakob Homan Date: Thu, 3 Sep 2026 12:45:14 -0700 Subject: [PATCH] refactor: rename the single-destination replicator (was: duckling) The component collided with two other 'duckling' terms: the duckgres Duckling K8s CR and PostHog's tenant data plane (posthog-duckling-* buckets). The component is now 'single-destination viaduck' everywhere: class SingleDestinationViaduck, SingleDestinationConfig, FatalSingleDestinationError, metrics prefix viaduck_single_destination_*, logger viaduck.single_destination, instance_id default, pipeline label prefix, and the temp view name. The ducklings k8s namespace strings stay (cluster-side fact; renaming that is an infra change). No behavior change; the feature is not deployed anywhere with dashboards, so the metric-series rename is free. --- .../test_single_destination_e2e.py | 56 +++--- tests/unit/test_single_destination.py | 178 +++++++++--------- viaduck/feed.py | 2 +- viaduck/single_destination.py | 86 ++++----- 4 files changed, 162 insertions(+), 160 deletions(-) diff --git a/tests/integration/test_single_destination_e2e.py b/tests/integration/test_single_destination_e2e.py index 5fddf72..c050019 100644 --- a/tests/integration/test_single_destination_e2e.py +++ b/tests/integration/test_single_destination_e2e.py @@ -1,4 +1,4 @@ -"""End-to-end: the per-destination duckling against real catalogs. +"""End-to-end: the single-destination viaduck against real catalogs. PG testcontainer holds both ducklake catalogs (source + destination) and the viaduck_state cursor table on the SOURCE database (the colocated @@ -17,11 +17,11 @@ from testcontainers.postgres import PostgresContainer from viaduck import metrics -from viaduck.single_destination import Duckling, DucklingConfig, FatalDucklingError +from viaduck.single_destination import FatalSingleDestinationError, SingleDestinationConfig, SingleDestinationViaduck def setup_module(): - metrics.init("duckling_integration_test") + metrics.init("single_destination_integration_test") @pytest.fixture(scope="module") @@ -41,7 +41,7 @@ def _mkdb(pg_dsn: str, name: str) -> str: class Env: - """Test handle: both catalogs plus a DucklingConfig factory.""" + """Test handle: both catalogs plus a SingleDestinationConfig factory.""" def __init__(self, src, dst, src_dsn, dst_dsn, tmp_path): self.src = src @@ -51,7 +51,7 @@ def __init__(self, src, dst, src_dsn, dst_dsn, tmp_path): self.tmp_path = tmp_path self._destination_id = f"org-test-team-2-{uuid.uuid4().hex[:6]}" - def cfg(self, **kw) -> DucklingConfig: + def cfg(self, **kw) -> SingleDestinationConfig: base = dict( # ATTACH format on purpose: exercises the F2 conninfo translation source_pg_uri=f"postgres:{self.src_dsn}", @@ -67,12 +67,12 @@ def cfg(self, **kw) -> DucklingConfig: cursor_pg_uri="", destination_id=self._destination_id, # stable per test: a restart # resumes the SAME cursor row - instance_id="duckling-test-0", + instance_id="single-destination-test-0", poll_interval_s=0.1, start_snapshot_id=0, # tests want full catch-up, not start-at-head ) base.update(kw) - return DucklingConfig(**base) + return SingleDestinationConfig(**base) @pytest.fixture() @@ -114,7 +114,7 @@ def test_catchup_filters_and_advances(self, env): _insert(src, [(2, "c")]) _insert(src, [(9, "d"), (2, "e")]) - d = Duckling(env.cfg()) + d = SingleDestinationViaduck(env.cfg()) d.boot() try: d.poll_once() @@ -136,7 +136,7 @@ def test_crash_between_append_and_cursor_redelivers_once(self, env): src, dst = env.src, env.dst _insert(src, [(2, "a"), (2, "b")]) - d = Duckling(env.cfg()) + d = SingleDestinationViaduck(env.cfg()) d.boot() d._cursor_advance = lambda *a, **k: (_ for _ in ()).throw(RuntimeError("simulated crash")) try: @@ -146,7 +146,7 @@ def test_crash_between_append_and_cursor_redelivers_once(self, env): _close(d) assert _dest_rows(dst) == [(2, "a"), (2, "b")] # committed; cursor not - d2 = Duckling(env.cfg()) # "restart": same cursor row + d2 = SingleDestinationViaduck(env.cfg()) # "restart": same cursor row d2.boot() try: d2.poll_once() @@ -160,7 +160,7 @@ def test_add_column_restart_picks_up(self, env): src, dst = env.src, env.dst _insert(src, [(2, "a")]) - d = Duckling(env.cfg()) + d = SingleDestinationViaduck(env.cfg()) d.boot() try: # NB: on the deployed build the ALTER resurrects the session's @@ -173,7 +173,7 @@ def test_add_column_restart_picks_up(self, env): _close(d) assert _dest_rows(dst) == [(2, "a"), (2, "b")] - d2 = Duckling(env.cfg()) + d2 = SingleDestinationViaduck(env.cfg()) d2.boot() try: assert "extra" in d2.columns @@ -193,7 +193,7 @@ def test_delete_appearance_crashes(self, env): _insert(src, [(2, "a")]) src.connection.execute("DELETE FROM lake.main.events WHERE team_id = 2") - d = Duckling(env.cfg()) + d = SingleDestinationViaduck(env.cfg()) try: # the baseline assertions at boot already see the delete with pytest.raises(RuntimeError, match="contract violated|delete"): @@ -209,7 +209,7 @@ def test_restart_resumes_from_cursor(self, env): src, dst = env.src, env.dst _insert(src, [(2, "a")]) - d = Duckling(env.cfg()) + d = SingleDestinationViaduck(env.cfg()) d.boot() try: d.poll_once() @@ -218,7 +218,7 @@ def test_restart_resumes_from_cursor(self, env): _close(d) assert _dest_rows(dst) == [(2, "a")] - d2 = Duckling(env.cfg()) # same destination_id → same cursor row + d2 = SingleDestinationViaduck(env.cfg()) # same destination_id → same cursor row d2.boot() try: assert d2._cursor == cursor_after # resumed, not restarted @@ -246,7 +246,7 @@ def test_retention_clamp_then_poll_composes(self, env): floor = snaps[-1] pg.execute(f'DELETE FROM "{meta}".ducklake_snapshot WHERE snapshot_id < {floor}') - d = Duckling(env.cfg()) + d = SingleDestinationViaduck(env.cfg()) d.boot() # writes the cursor row at start_snapshot_id=0 try: assert d._cursor == 0 @@ -318,9 +318,9 @@ def test_inlined_delete_of_flushed_rows_crashes(self, env): pg.execute(f'DELETE FROM "{meta}".ducklake_snapshot_changes') pg.close() - d = Duckling(env.cfg(source_table="main.inlined_del2", dest_table="main.inlined_del2")) + d = SingleDestinationViaduck(env.cfg(source_table="main.inlined_del2", dest_table="main.inlined_del2")) try: - with pytest.raises(FatalDucklingError, match="inlined deletes"): + with pytest.raises(FatalSingleDestinationError, match="inlined deletes"): d.boot() # baseline assertions: store probe fires finally: _close(d) @@ -339,7 +339,7 @@ def test_accepted_delete_below_cursor_boots_clean(self, env): head = int(pg.execute(f'SELECT MAX(snapshot_id) FROM "{meta}".ducklake_snapshot').fetchone()[0]) pg.close() - d = Duckling(env.cfg(start_snapshot_id=head)) # the documented accept: cursor past the delete + d = SingleDestinationViaduck(env.cfg(start_snapshot_id=head)) # the documented accept: cursor past the delete d.boot() try: _insert(src, [(2, "b")]) @@ -358,7 +358,7 @@ def test_dest_reordered_columns_map_by_name(self, env): "CREATE TABLE dst.main.events (event VARCHAR, _inserted_at TIMESTAMPTZ DEFAULT now(), team_id BIGINT)" ) - d = Duckling(env.cfg()) + d = SingleDestinationViaduck(env.cfg()) d.boot() try: d.poll_once() @@ -377,7 +377,7 @@ def test_dest_managed_column_default_filled(self, env): "CREATE TABLE dst.main.events (team_id BIGINT, event VARCHAR, _inserted_at TIMESTAMPTZ DEFAULT now())" ) - d = Duckling(env.cfg()) + d = SingleDestinationViaduck(env.cfg()) d.boot() try: d.poll_once() @@ -398,9 +398,9 @@ def test_inlined_delete_pages_via_witness(self, env): src.connection.execute("INSERT INTO lake.main.inlined_del VALUES (2, 'a'), (2, 'b')") src.connection.execute("DELETE FROM lake.main.inlined_del WHERE event = 'a'") - d = Duckling(env.cfg(source_table="main.inlined_del", dest_table="main.inlined_del")) + d = SingleDestinationViaduck(env.cfg(source_table="main.inlined_del", dest_table="main.inlined_del")) try: - with pytest.raises(FatalDucklingError, match="delete/drop activity"): + with pytest.raises(FatalSingleDestinationError, match="delete/drop activity"): d.boot() finally: d.feed.close() @@ -410,7 +410,7 @@ def test_inlined_delete_pages_via_witness(self, env): class TestVariantSourceColumn: """The production shape: events_nrt carries millpond's VARIANT dual-write - companions. The duckling must boot, read, and deliver as if they don't + companions. The single-destination viaduck must boot, read, and deliver as if they don't exist (load_table's EXCLUDABLE_SOURCE_TYPES exclusion).""" def test_boot_and_deliver_with_variant_column(self, env): @@ -420,7 +420,7 @@ def test_boot_and_deliver_with_variant_column(self, env): ) src.connection.execute("INSERT INTO lake.main.events_v VALUES (2, 'a', {\"k\": 1}), (3, 'b', NULL)") - d = Duckling(env.cfg(source_table="main.events_v", dest_table="main.events_v")) + d = SingleDestinationViaduck(env.cfg(source_table="main.events_v", dest_table="main.events_v")) d.boot() try: assert "properties_variant" not in d.columns @@ -439,7 +439,7 @@ def test_variant_added_mid_stream(self, env): src, dst = env.src, env.dst _insert(src, [(2, "a")]) - d = Duckling(env.cfg()) + d = SingleDestinationViaduck(env.cfg()) d.boot() try: src.connection.execute("ALTER TABLE lake.main.events ADD COLUMN properties_variant VARIANT") @@ -451,7 +451,7 @@ def test_variant_added_mid_stream(self, env): _close(d) assert _dest_rows(dst) == [(2, "a"), (2, "b")] - d2 = Duckling(env.cfg()) + d2 = SingleDestinationViaduck(env.cfg()) d2.boot() # restart: load_table excludes the VARIANT again try: assert "properties_variant" not in d2.columns @@ -474,7 +474,7 @@ def test_inline_rows_served_with_page(self, env): src.connection.execute("INSERT INTO lake.main.inlined VALUES (2, 'i1'), (3, 'i2')") src.connection.execute("RESET ducklake_default_data_inlining_row_limit") - d = Duckling(env.cfg(source_table="main.inlined", dest_table="main.inlined")) + d = SingleDestinationViaduck(env.cfg(source_table="main.inlined", dest_table="main.inlined")) d.boot() try: d.poll_once() diff --git a/tests/unit/test_single_destination.py b/tests/unit/test_single_destination.py index 765e79a..e4e0f50 100644 --- a/tests/unit/test_single_destination.py +++ b/tests/unit/test_single_destination.py @@ -1,4 +1,4 @@ -"""Unit tests for viaduck.single_destination — the per-destination duckling. +"""Unit tests for viaduck.single_destination — the single-destination viaduck. The wrapper is the only layer that can silently lose data (the read core is pinned by the feed parity suite), so the cursor/ordering/crash semantics @@ -15,16 +15,16 @@ import pytest from viaduck import single_destination as sd -from viaduck.single_destination import Duckling, DucklingConfig, FatalDucklingError +from viaduck.single_destination import FatalSingleDestinationError, SingleDestinationConfig, SingleDestinationViaduck -def _cfg(**kw) -> DucklingConfig: +def _cfg(**kw) -> SingleDestinationConfig: base = dict( source_pg_uri="postgres:host=src port=5432 dbname=megaduck user=m password=pw", source_catalog="lake", source_data_path="s3://b/src", source_table="main.events_nrt", - dest_pg_uri="postgres:host=dst port=5432 dbname=duckling user=d password=pw", + dest_pg_uri="postgres:host=dst port=5432 dbname=dest user=d password=pw", dest_catalog="dest", dest_data_path="s3://b/dst", dest_table="posthog.events", @@ -33,13 +33,13 @@ def _cfg(**kw) -> DucklingConfig: destination_id="org-abc-team-2", ) base.update(kw) - return DucklingConfig(**base) + return SingleDestinationConfig(**base) -def _duckling(cfg=None, cursor=100) -> Duckling: - """A Duckling without boot(): all collaborators mocked. Catalog-side +def _single_destination(cfg=None, cursor=100) -> SingleDestinationViaduck: + """A SingleDestinationViaduck without boot(): all collaborators mocked. Catalog-side queries route through feed._pg(); cursor writes through _cursor_conn.""" - d = Duckling.__new__(Duckling) + d = SingleDestinationViaduck.__new__(SingleDestinationViaduck) d.cfg = cfg or _cfg() d._stop = threading.Event() d._last_poll_ok = time.monotonic() @@ -64,8 +64,8 @@ def _duckling(cfg=None, cursor=100) -> Duckling: return d -def _poll_ready(d: Duckling, rows: pa.Table, head=500, hi=500): - """Wire a duckling for poll_once: assertions no-op, plan/read mocked.""" +def _poll_ready(d: SingleDestinationViaduck, rows: pa.Table, head=500, hi=500): + """Wire a single-destination viaduck for poll_once: assertions no-op, plan/read mocked.""" d._assert_no_deletes = MagicMock() d._check_inline_stores = MagicMock() d._clamp_to_retention = MagicMock() @@ -85,7 +85,7 @@ def _no_sleep(self, monkeypatch): monkeypatch.setattr(sd.time, "sleep", lambda *_: None) def test_cursor_strictly_after_commit(self): - d = _duckling() + d = _single_destination() rows = pa.table({"team_id": [2, 3, 2], "event": ["a", "b", "c"]}) _poll_ready(d, rows) parent = MagicMock() @@ -101,17 +101,17 @@ def test_append_sql_is_by_name(self): """The silent-corruption guard's shape pin: the append MUST be INSERT BY NAME (positional silently swaps same-typed columns on reordered dest tables).""" - d = _duckling() + d = _single_destination() d._append(pa.table({"team_id": [2]})) appends = [c for c in d.dst_catalog.connection.execute.call_args_list if "INSERT INTO" in str(c)] assert len(appends) == 1 and "BY NAME" in appends[0].args[0] def test_append_failure_never_advances_cursor_and_is_fatal(self): - d = _duckling() + d = _single_destination() rows = pa.table({"team_id": [2], "event": ["a"]}) _poll_ready(d, rows) d.dst_catalog.connection.execute.side_effect = RuntimeError("catalog down") - with pytest.raises(FatalDucklingError, match="append failed after 3 attempts"): + with pytest.raises(FatalSingleDestinationError, match="append failed after 3 attempts"): d.poll_once() assert d._cursor == 100 assert not any("ON CONFLICT" in str(c) for c in d._cursor_conn.execute.call_args_list) @@ -122,7 +122,7 @@ def test_append_failure_never_advances_cursor_and_is_fatal(self): def test_empty_range_still_advances(self): """Foreign-commit polls: hi is a valid coverage boundary; idling forever would burn the retention window.""" - d = _duckling() + d = _single_destination() _poll_ready(d, pa.table({"team_id": pa.array([], type=pa.int64()), "event": pa.array([], type=pa.string())})) d.poll_once() assert d._cursor == 500 @@ -131,15 +131,15 @@ def test_empty_range_still_advances(self): def test_empty_range_rechecks_table_identity(self): """A silent drop+recreate is invisible to the feed's cached table_id (its plans come back empty forever) — the empty path re-resolves.""" - d = _duckling() + d = _single_destination() _poll_ready(d, pa.table({"team_id": pa.array([], type=pa.int64()), "event": pa.array([], type=pa.string())})) d._resolve_table_id = MagicMock(return_value=17) - with pytest.raises(FatalDucklingError, match="table_id changed"): + with pytest.raises(FatalSingleDestinationError, match="table_id changed"): d.poll_once() assert d._cursor == 100 # frozen def test_idle_when_at_head(self): - d = _duckling() + d = _single_destination() _poll_ready(d, pa.table({"team_id": [2]}), head=100) d.poll_once() d.feed.plan_unit.assert_not_called() @@ -148,24 +148,24 @@ def test_idle_when_at_head(self): class TestDropCreate: def test_table_id_change_on_read_error_freezes_and_pages(self): - d = _duckling() + d = _single_destination() _poll_ready(d, pa.table({"team_id": [2]})) d.feed.read.side_effect = RuntimeError("catalog read exploded") d._resolve_table_id = MagicMock(return_value=17) - with pytest.raises(FatalDucklingError, match="table_id changed 16 → 17"): + with pytest.raises(FatalSingleDestinationError, match="table_id changed 16 → 17"): d.poll_once() assert d._cursor == 100 def test_table_dropped_pages(self): - d = _duckling() + d = _single_destination() _poll_ready(d, pa.table({"team_id": [2]})) d.feed.read.side_effect = RuntimeError("boom") d._resolve_table_id = MagicMock(side_effect=sd.ConfigError("not found")) - with pytest.raises(FatalDucklingError, match="source table dropped"): + with pytest.raises(FatalSingleDestinationError, match="source table dropped"): d.poll_once() def test_transient_read_error_reraises(self): - d = _duckling() + d = _single_destination() _poll_ready(d, pa.table({"team_id": [2]})) d.feed.read.side_effect = RuntimeError("s3 flaked") d._resolve_table_id = MagicMock(return_value=16) # unchanged → transient @@ -175,17 +175,17 @@ def test_transient_read_error_reraises(self): def test_cursor_provenance_mismatch_at_boot(self): """Drop+recreate WHILE DOWN: the stored table_id is the only evidence.""" - d = _duckling() + d = _single_destination() d.table_id = 99 # resolved fresh at boot d._cursor_pg().execute.return_value.fetchone.return_value = (100, 16) # stored against 16 - with pytest.raises(FatalDucklingError, match="table_id=16"): + with pytest.raises(FatalSingleDestinationError, match="table_id=16"): d._cursor_load() def test_cursor_provenance_backfilled_when_null(self): """The fleet-transplant path: a pre-existing row with NULL provenance gets stamped at boot — otherwise a drop+recreate-while-down is undetectable (the witness is keyed to the NEW table_id).""" - d = _duckling() + d = _single_destination() d._cursor_pg().execute.return_value.fetchone.return_value = (100, None) d._cursor_load() assert d._cursor == 100 @@ -195,10 +195,10 @@ def test_cursor_provenance_backfilled_when_null(self): class TestClampAssertionOrdering: def test_clamp_runs_before_assertions(self): - """A reorder regression crash-loops the duckling inside the retention + """A reorder regression crash-loops the single-destination viaduck inside the retention window (a delete whose evidence outlives its snapshots must not veto the clamp's loud advance).""" - d = _duckling(cursor=100) + d = _single_destination(cursor=100) calls = [] d._clamp_to_retention = MagicMock(side_effect=lambda: calls.append("clamp")) d._assert_no_deletes = MagicMock(side_effect=lambda: calls.append("assert")) @@ -212,7 +212,7 @@ class TestFeedErrorClassification: def test_floor_feederror_is_transient(self): """The floor guard's FeedError re-raises as-is: next poll's clamp advances past it. Making it fatal would crash-loop in the window.""" - d = _duckling() + d = _single_destination() _poll_ready(d, pa.table({"team_id": [2]})) d.feed.read.side_effect = sd.feed.FeedError("cursor 5 is below the retained snapshot floor 9") with pytest.raises(sd.feed.FeedError, match="retained snapshot floor"): @@ -221,15 +221,15 @@ def test_floor_feederror_is_transient(self): def test_other_feederror_is_fatal(self): """A real refusal (non-additive schema, encryption) must crash — transient retry is a silent stall with a growing lag gauge.""" - d = _duckling() + d = _single_destination() _poll_ready(d, pa.table({"team_id": [2]})) d.feed.read.side_effect = sd.feed.FeedError("non-additive schema change (rename/drop) is unsupported") - with pytest.raises(FatalDucklingError, match="non-additive"): + with pytest.raises(FatalSingleDestinationError, match="non-additive"): d.poll_once() def test_missing_team_column_is_fatal(self): - d = _duckling() - with pytest.raises(FatalDucklingError, match="schema contract"): + d = _single_destination() + with pytest.raises(FatalSingleDestinationError, match="schema contract"): d._arrow_filter(pa.table({"other": [1]})) @@ -237,9 +237,9 @@ class TestHeadRegression: def test_head_below_cursor_is_fatal(self): """A source restore/rebuild regresses head under the cursor: the ONE alarm (lag) must not read 0 while wedged — crash loudly.""" - d = _duckling(cursor=100) + d = _single_destination(cursor=100) _poll_ready(d, pa.table({"team_id": [2]}), head=50) - with pytest.raises(FatalDucklingError, match="regressed below cursor"): + with pytest.raises(FatalSingleDestinationError, match="regressed below cursor"): d.poll_once() assert d._cursor == 100 @@ -250,7 +250,7 @@ def test_head_below_cursor_is_fatal(self): class TestRunSupervision: - def _runnable(self, d: Duckling): + def _runnable(self, d: SingleDestinationViaduck): d.boot = MagicMock() d._maybe_recycle = MagicMock() with patch.object(sd, "_start_health_server", return_value=MagicMock()): @@ -259,15 +259,15 @@ def _runnable(self, d: Duckling): yield d def test_fatal_crashes(self): - d = _duckling() + d = _single_destination() for d in self._runnable(d): - d.poll_once = MagicMock(side_effect=FatalDucklingError("assertion fired")) - with pytest.raises(FatalDucklingError): + d.poll_once = MagicMock(side_effect=FatalSingleDestinationError("assertion fired")) + with pytest.raises(FatalSingleDestinationError): d.run() def test_transient_retries_next_poll(self, monkeypatch): monkeypatch.setattr(sd.time, "sleep", lambda *_: None) # backoff budget - d = _duckling(_cfg(poll_interval_s=0.01)) + d = _single_destination(_cfg(poll_interval_s=0.01)) for d in self._runnable(d): calls = [0] @@ -286,7 +286,7 @@ def test_poll_wait_is_jittered(self, monkeypatch): interval synchronizes the assertion burst after any fleet-wide restart). Pin the wait's shape deterministically.""" monkeypatch.setattr(sd.random, "random", lambda: 0.5) - d = _duckling(_cfg(poll_interval_s=10.0)) + d = _single_destination(_cfg(poll_interval_s=10.0)) for d in self._runnable(d): seen = [] @@ -306,7 +306,7 @@ def stop_after_one(): class TestRetentionClamp: def test_below_floor_advances_loudly_with_loss_note(self): - d = _duckling(cursor=100) + d = _single_destination(cursor=100) d.feed._pg.return_value.execute.return_value.fetchone.return_value = (600,) # MIN(snapshot_id) d._clamp_to_retention() assert d._cursor == 599 @@ -315,14 +315,14 @@ def test_below_floor_advances_loudly_with_loss_note(self): assert "100" in str(update) and "599" in str(update) def test_at_floor_is_quiet(self): - d = _duckling(cursor=599) + d = _single_destination(cursor=599) d.feed._pg.return_value.execute.return_value.fetchone.return_value = (600,) d._clamp_to_retention() assert d._cursor == 599 assert not any("last_error" in str(c) for c in d._cursor_conn.execute.call_args_list) def test_empty_snapshot_table_is_noop(self): - d = _duckling(cursor=0) + d = _single_destination(cursor=0) d.feed._pg.return_value.execute.return_value.fetchone.return_value = (None,) d._clamp_to_retention() assert d._cursor == 0 @@ -339,7 +339,7 @@ def _no_sleep(self, monkeypatch): monkeypatch.setattr(sd.time, "sleep", lambda *_: None) def test_flush_failure_halves_budget(self): - d = _duckling() + d = _single_destination() d.dst_catalog.connection.execute.side_effect = [RuntimeError("occ contention"), None] d._append(pa.table({"team_id": [2]})) # halve on the failed attempt (50000 → 25000), then +10% recovery on @@ -347,13 +347,13 @@ def test_flush_failure_halves_budget(self): assert d._budget_rows == 27_500 def test_slow_flush_halves(self): - d = _duckling() + d = _single_destination() d.cfg = _cfg(slow_flush_seconds=0.0) # every flush is "slow" d._append(pa.table({"team_id": [2]})) assert d._budget_rows == 25_000 def test_floor_and_recovery(self): - d = _duckling() + d = _single_destination() d._budget_rows = d.cfg.aimd_floor_rows d._aimd_halve("test") assert d._budget_rows == d.cfg.aimd_floor_rows @@ -375,13 +375,13 @@ def _no_sleep(self, monkeypatch): def test_cursor_advance_has_monotonic_guard(self): """A maxSurge pair sharing the row must never regress it (fleet semantics; the guard is what makes a racing advance a no-op).""" - d = _duckling() + d = _single_destination() d._cursor_advance(500, 3) sql = d._cursor_conn.execute.call_args.args[0] assert "WHERE viaduck.viaduck_state.last_snapshot_id <= EXCLUDED.last_snapshot_id" in sql def test_retry_then_success(self): - d = _duckling() + d = _single_destination() import psycopg d._cursor_conn.execute.side_effect = [psycopg.OperationalError("pg blip"), None] @@ -390,11 +390,11 @@ def test_retry_then_success(self): assert d._cursor_conn.execute.call_count == 2 def test_exhaustion_is_fatal(self): - d = _duckling() + d = _single_destination() import psycopg d._cursor_conn.execute.side_effect = psycopg.OperationalError("pg down") - with pytest.raises(FatalDucklingError, match="cursor update failed"): + with pytest.raises(FatalSingleDestinationError, match="cursor update failed"): d._cursor_advance(500, 10) assert d._cursor == 100 # in-memory cursor did not move @@ -406,7 +406,7 @@ def test_exhaustion_is_fatal(self): class TestArrowFilter: def test_filters_to_team(self): - d = _duckling() + d = _single_destination() rows = pa.table({"team_id": [2, 3, 2], "event": ["a", "b", "c"]}) out = d._arrow_filter(rows) assert out.num_rows == 2 @@ -416,18 +416,18 @@ def test_unfiltered_read_contract(self): """poll_once must NOT pass a filter to the feed: parquet zone-maps can lie on add_files-registered files, and SQL pushdown under- delivery is invisible. The Arrow layer is the whole filter.""" - d = _duckling() + d = _single_destination() _poll_ready(d, pa.table({"team_id": [2]})) d.poll_once() assert "filter_expr" not in d.feed.read.call_args.kwargs or d.feed.read.call_args.kwargs["filter_expr"] is None def test_missing_team_column_crashes_loudly(self): - d = _duckling() + d = _single_destination() with pytest.raises(Exception): d._arrow_filter(pa.table({"other": [1]})) def test_string_team_value(self): - d = _duckling(_cfg(team_field="team", team_value="blue")) + d = _single_destination(_cfg(team_field="team", team_value="blue")) d._team_array = pa.array(["blue"], type=pa.string()) rows = pa.table({"team": ["blue", "red"]}) assert d._arrow_filter(rows).num_rows == 1 @@ -447,7 +447,7 @@ def test_assertions_scoped_to_cursor(self): """The accept path's linchpin: every assertion query is scoped to UN-CROSSED history (snapshot > cursor). Dropping a scope predicate reintroduces the no-accept-path trap (round-3 C1).""" - d = _duckling(cursor=100) + d = _single_destination(cursor=100) pg = self._pg_with_counts({}, regclass=None) d.feed._pg.return_value = pg d._assert_no_deletes() @@ -460,7 +460,7 @@ def test_assertions_scoped_to_cursor(self): def test_delete_below_cursor_does_not_fire(self): """An adjudicated (pre-cursor) delete: all checks stay quiet.""" - d = _duckling(cursor=100) + d = _single_destination(cursor=100) pg = MagicMock() @@ -485,7 +485,7 @@ def execute(sql, params=None): def test_inlined_delete_store_scoped_to_cursor(self): """The store-probe leg carries the same cursor scope — dropping that predicate reintroduces the no-accept-path trap for inlined deletes.""" - d = _duckling(cursor=100) + d = _single_destination(cursor=100) pg = self._pg_with_counts({"ducklake_inlined_delete_16": 0}, regclass="lake_meta.ducklake_inlined_delete_16") d.feed._pg.return_value = pg d._assert_no_deletes() @@ -516,24 +516,24 @@ def execute(sql, params=None): return pg def test_delete_file_appearance_crashes(self): - d = _duckling() + d = _single_destination() d.feed._pg.return_value = self._pg_with_counts({"ducklake_delete_file": 1}) - with pytest.raises(FatalDucklingError, match="append-only contract violated"): + with pytest.raises(FatalSingleDestinationError, match="append-only contract violated"): d._assert_no_deletes() def test_end_snapshot_appearance_crashes(self): - d = _duckling() + d = _single_destination() d.feed._pg.return_value = self._pg_with_counts({"ducklake_data_file": 3}) - with pytest.raises(FatalDucklingError, match="end_snapshot"): + with pytest.raises(FatalSingleDestinationError, match="end_snapshot"): d._assert_no_deletes() def test_witness_regex_scoped_to_table_and_vocabulary(self): """The fork's exact delete vocabulary: deleted_from_table / inlined_delete / rewrite_delete / dropped_table (verified against ducklake_transaction.cpp AddChangeInfo).""" - d = _duckling() + d = _single_destination() d.feed._pg.return_value = self._pg_with_counts({"ducklake_snapshot_changes": 1}) - with pytest.raises(FatalDucklingError, match="delete/drop activity"): + with pytest.raises(FatalSingleDestinationError, match="delete/drop activity"): d._assert_no_deletes() witness = next(c for c in d.feed._pg.return_value.execute.call_args_list if "snapshot_changes" in str(c)) pattern = witness.args[1][1] # params: (cursor, pattern) @@ -542,26 +542,26 @@ def test_witness_regex_scoped_to_table_and_vocabulary(self): assert ":16" in pattern def test_inlined_delete_table_nonempty_crashes(self): - d = _duckling() + d = _single_destination() d.feed._pg.return_value = self._pg_with_counts( {"ducklake_inlined_delete_16": 2}, regclass="lake_meta.ducklake_inlined_delete_16" ) - with pytest.raises(FatalDucklingError, match="inlined deletes"): + with pytest.raises(FatalSingleDestinationError, match="inlined deletes"): d._assert_no_deletes() def test_inlined_delete_table_absent_passes(self): - d = _duckling() + d = _single_destination() d.feed._pg.return_value = self._pg_with_counts({}, regclass=None) d._assert_no_deletes() def test_inline_rows_page_but_continue(self): from prometheus_client import REGISTRY - d = _duckling() + d = _single_destination() d.feed._pg.return_value = self._pg_with_counts( {"ducklake_inlined_data_16_1": 2}, stores=["ducklake_inlined_data_16_1"] ) - metric = "viaduck_duckling_assertion_failures_total" + metric = "viaduck_single_destination_assertion_failures_total" before = REGISTRY.get_sample_value(metric, {"check": "inline_rows_present"}) or 0 d._check_inline_stores() # no raise — the feed serves inline correctly after = REGISTRY.get_sample_value(metric, {"check": "inline_rows_present"}) @@ -570,7 +570,7 @@ def test_inline_rows_page_but_continue(self): def test_inline_registry_alone_is_quiet(self): """Registry membership is normal (stores register at CREATE TABLE even with row_limit=0); only ROWS in stores are drift.""" - d = _duckling() + d = _single_destination() d.feed._pg.return_value = self._pg_with_counts({}, stores=["ducklake_inlined_data_16_1"]) d._check_inline_stores() @@ -596,15 +596,15 @@ def test_from_env_minimal(self, monkeypatch): "DESTINATION_ID": "org-abc-team-2", }.items(): monkeypatch.setenv(k, v) - cfg = DucklingConfig.from_env() + cfg = SingleDestinationConfig.from_env() assert cfg.cursor_pg_uri == cfg.source_pg_uri # colocated default - assert cfg.instance_id == "duckling" # STABLE — never a pod name + assert cfg.instance_id == "single-destination" # STABLE — never a pod name assert cfg.unit_max_rows == 50_000 def test_unsafe_identifier_refused(self, monkeypatch): monkeypatch.setenv("SOURCE_TABLE", "events; DROP TABLE x") with pytest.raises(sd.ConfigError): - DucklingConfig.from_env() + SingleDestinationConfig.from_env() def test_dest_managed_columns_parsed(self, monkeypatch): monkeypatch.setenv("DEST_MANAGED_COLUMNS", "_inserted_at, _raw") @@ -623,7 +623,7 @@ def test_dest_managed_columns_parsed(self, monkeypatch): "DESTINATION_ID": "d", }.items(): monkeypatch.setenv(k, v) - assert DucklingConfig.from_env().dest_managed_columns == frozenset({"_inserted_at", "_raw"}) + assert SingleDestinationConfig.from_env().dest_managed_columns == frozenset({"_inserted_at", "_raw"}) # --------------------------------------------------------------------------- @@ -631,7 +631,7 @@ def test_dest_managed_columns_parsed(self, monkeypatch): # --------------------------------------------------------------------------- -def _boot_mocks(d: Duckling, cursor_row=(100, 16), head=900, winner_row=None): +def _boot_mocks(d: SingleDestinationViaduck, cursor_row=(100, 16), head=900, winner_row=None): """Patch catalog/feed/psycopg collaborators for boot(); returns mocks. winner_row: when cursor_row is None (first boot), the re-SELECT after @@ -688,7 +688,7 @@ def test_boot_wiring_and_conninfo_translation(self): """The F2 pin at this layer: ATTACH-format secret in, libpq conninfo to psycopg — or first boot crashes.""" cfg = _cfg() - d = Duckling(cfg) + d = SingleDestinationViaduck(cfg) src_table, cursor_pg, catalog_pg, catalog = _boot_mocks(d) with ( @@ -709,7 +709,7 @@ def test_boot_wiring_and_conninfo_translation(self): def test_cursor_initialized_at_head_when_absent(self): cfg = _cfg() - d = Duckling(cfg) + d = SingleDestinationViaduck(cfg) src_table, cursor_pg, catalog_pg, catalog = _boot_mocks(d, cursor_row=None) with ( @@ -732,7 +732,7 @@ def test_first_boot_race_adopts_winner_row(self): cursor — adopting it would silently skip the between range. The re-SELECT pins adoption.""" cfg = _cfg() - d = Duckling(cfg) + d = SingleDestinationViaduck(cfg) src_table, cursor_pg, catalog_pg, catalog = _boot_mocks(d, cursor_row=None, winner_row=(850,)) with ( @@ -749,7 +749,7 @@ def test_first_boot_race_adopts_winner_row(self): def test_dest_column_reconciliation_adds_missing(self): cfg = _cfg() - d = Duckling(cfg) + d = SingleDestinationViaduck(cfg) src_table, cursor_pg, catalog_pg, catalog = _boot_mocks(d) catalog.create_table_if_not_exists.return_value.schema.column_names.return_value = ("team_id",) @@ -768,7 +768,7 @@ def test_dest_column_reconciliation_adds_missing(self): def test_dest_extra_column_wedges_but_managed_columns_pass(self): cfg = _cfg() - d = Duckling(cfg) + d = SingleDestinationViaduck(cfg) src_table, cursor_pg, catalog_pg, catalog = _boot_mocks(d) catalog.create_table_if_not_exists.return_value.schema.column_names.return_value = ( "team_id", @@ -785,12 +785,12 @@ def test_dest_extra_column_wedges_but_managed_columns_pass(self): ): fr_cls.return_value._meta_schema = "lake_meta" fr_cls.return_value._pg.return_value = catalog_pg - with pytest.raises(FatalDucklingError, match="mystery"): + with pytest.raises(FatalSingleDestinationError, match="mystery"): d.boot() def test_integer_team_value_validated_at_boot(self): cfg = _cfg(team_value="not-an-int") - d = Duckling(cfg) + d = SingleDestinationViaduck(cfg) src_table, cursor_pg, catalog_pg, catalog = _boot_mocks(d) with ( @@ -801,7 +801,7 @@ def test_integer_team_value_validated_at_boot(self): ): fr_cls.return_value._meta_schema = "lake_meta" fr_cls.return_value._pg.return_value = catalog_pg - with pytest.raises(FatalDucklingError, match="not an integer"): + with pytest.raises(FatalSingleDestinationError, match="not an integer"): d.boot() @@ -812,11 +812,11 @@ def test_integer_team_value_validated_at_boot(self): class TestHealth: def test_healthy_after_recent_poll(self): - d = _duckling() + d = _single_destination() assert d.is_healthy() def test_stale_after_threshold(self): - d = _duckling() + d = _single_destination() d._last_poll_ok = time.monotonic() - 400 # 300s floor assert not d.is_healthy() @@ -825,10 +825,10 @@ class TestMetricsMove: def test_lag_gauge_and_delivered_counter(self): from prometheus_client import REGISTRY - d = _duckling() + d = _single_destination() rows = pa.table({"team_id": [2, 3], "event": ["a", "b"]}) _poll_ready(d, rows) - before = REGISTRY.get_sample_value("viaduck_duckling_rows_delivered_total") or 0 + before = REGISTRY.get_sample_value("viaduck_single_destination_rows_delivered_total") or 0 d.poll_once() - assert (REGISTRY.get_sample_value("viaduck_duckling_rows_delivered_total") or 0) == before + 1 - assert REGISTRY.get_sample_value("viaduck_duckling_lag_snapshots") == 400 # head 500 - cursor 100 + assert (REGISTRY.get_sample_value("viaduck_single_destination_rows_delivered_total") or 0) == before + 1 + assert REGISTRY.get_sample_value("viaduck_single_destination_lag_snapshots") == 400 # head 500 - cursor 100 diff --git a/viaduck/feed.py b/viaduck/feed.py index 44f0607..464c5de 100644 --- a/viaduck/feed.py +++ b/viaduck/feed.py @@ -167,7 +167,7 @@ def _pg(self) -> psycopg.Connection: # prepare_threshold=None: psycopg3 otherwise auto-promotes # to server-side prepared statements after 5 executions — # fatal under pgbouncer transaction pooling (per-destination - # duckling fleet runs behind it). + # single-destination fleet runs behind it). prepare_threshold=None, # idle_in_transaction_session_timeout: a wedged REPEATABLE # READ snapshot pins the catalog's vacuum horizon; at N diff --git a/viaduck/single_destination.py b/viaduck/single_destination.py index e7a12c0..8c0b6f5 100644 --- a/viaduck/single_destination.py +++ b/viaduck/single_destination.py @@ -43,14 +43,14 @@ from viaduck.logging_config import setup as setup_logging from viaduck.scrub import scrub_credentials -log = logging.getLogger("viaduck.duckling") +log = logging.getLogger("viaduck.single_destination") class ConfigError(Exception): pass -class FatalDucklingError(RuntimeError): +class FatalSingleDestinationError(RuntimeError): """Crash-class failure (append exhausted, cursor exhausted, assertion violation, source rebuild): the process must die, not retry — K8s backoff is the supervisor. run() re-raises these; everything else is a @@ -61,18 +61,18 @@ class FatalDucklingError(RuntimeError): # Metrics (one process = one destination: no pipeline label needed) # --------------------------------------------------------------------------- -rows_read_total = Counter("viaduck_duckling_rows_read_total", "Rows read from the source (pre-filter)") -rows_delivered_total = Counter("viaduck_duckling_rows_delivered_total", "Rows appended to the destination") -flush_seconds = Histogram("viaduck_duckling_flush_seconds", "Destination append latency") -unit_budget_rows = Gauge("viaduck_duckling_unit_budget_rows", "Current AIMD read-unit row budget") -lag_snapshots = Gauge("viaduck_duckling_lag_snapshots", "Source head minus committed cursor") +rows_read_total = Counter("viaduck_single_destination_rows_read_total", "Rows read from the source (pre-filter)") +rows_delivered_total = Counter("viaduck_single_destination_rows_delivered_total", "Rows appended to the destination") +flush_seconds = Histogram("viaduck_single_destination_flush_seconds", "Destination append latency") +unit_budget_rows = Gauge("viaduck_single_destination_unit_budget_rows", "Current AIMD read-unit row budget") +lag_snapshots = Gauge("viaduck_single_destination_lag_snapshots", "Source head minus committed cursor") cursor_below_floor = Gauge( - "viaduck_duckling_cursor_below_floor", "1 while the cursor is under the retained snapshot floor" + "viaduck_single_destination_cursor_below_floor", "1 while the cursor is under the retained snapshot floor" ) assertion_failures_total = Counter( - "viaduck_duckling_assertion_failures_total", "Per-poll assertion failures", ["check"] + "viaduck_single_destination_assertion_failures_total", "Per-poll assertion failures", ["check"] ) -polls_total = Counter("viaduck_duckling_polls_total", "Loop iterations", ["result"]) +polls_total = Counter("viaduck_single_destination_polls_total", "Loop iterations", ["result"]) # --------------------------------------------------------------------------- # Config @@ -90,7 +90,7 @@ def _env(name: str, default: str | None = None, required: bool = True) -> str: @dataclass -class DucklingConfig: +class SingleDestinationConfig: source_pg_uri: str source_catalog: str source_data_path: str @@ -103,7 +103,7 @@ class DucklingConfig: team_value: str # validated against the pinned column's type at boot destination_id: str cursor_pg_uri: str = "" # defaults to source_pg_uri - instance_id: str = "duckling" # STABLE per destination — never a pod name + instance_id: str = "single-destination" # STABLE per destination — never a pod name dest_managed_columns: frozenset[str] = frozenset({"_inserted_at"}) s3_properties: dict[str, str] = field(default_factory=dict) poll_interval_s: float = 5.0 @@ -123,7 +123,7 @@ def __post_init__(self) -> None: self.cursor_pg_uri = self.source_pg_uri @classmethod - def from_env(cls) -> DucklingConfig: + def from_env(cls) -> SingleDestinationConfig: def ident(name: str, pattern=_IDENT) -> str: v = _env(name) if not pattern.match(v): @@ -147,7 +147,7 @@ def ident(name: str, pattern=_IDENT) -> str: team_value=_env("TEAM_VALUE"), destination_id=_env("DESTINATION_ID"), cursor_pg_uri=_env("CURSOR_PG_URI", "", required=False), - instance_id=_env("INSTANCE_ID", "duckling", required=False), + instance_id=_env("INSTANCE_ID", "single-destination", required=False), dest_managed_columns=managed, s3_properties=s3, poll_interval_s=float(_env("POLL_INTERVAL_S", "5", required=False)), @@ -161,12 +161,12 @@ def ident(name: str, pattern=_IDENT) -> str: # --------------------------------------------------------------------------- -# The duckling +# The single-destination viaduck # --------------------------------------------------------------------------- -class Duckling: - def __init__(self, cfg: DucklingConfig): +class SingleDestinationViaduck: + def __init__(self, cfg: SingleDestinationConfig): self.cfg = cfg self._stop = threading.Event() self._last_poll_ok = time.monotonic() @@ -202,12 +202,12 @@ def _cursor_pg(self) -> psycopg.Connection: def boot(self) -> None: try: self._boot_inner() - except FatalDucklingError: + except FatalSingleDestinationError: raise except Exception as e: # ATTACH/connect errors embed the full conninfo — scrub before # the message can reach pod logs. - raise FatalDucklingError(f"boot failed: {scrub_credentials(str(e))}") from None + raise FatalSingleDestinationError(f"boot failed: {scrub_credentials(str(e))}") from None def _boot_inner(self) -> None: cfg = self.cfg @@ -252,7 +252,7 @@ def _boot_inner(self) -> None: raise ConfigError(f"destination FQN {fqn!r} contains SQL metacharacters") self._dest_fqn = fqn - # Cursor row in viaduck_state (existing table; the duckling adds one + # Cursor row in viaduck_state (existing table; the single-destination viaduck adds one # additive column for table_id provenance — a drop+recreate WHILE # DOWN is otherwise undetectable: boot would resolve the new id and # the witness is keyed to it). Baseline assertions run BEFORE the @@ -264,7 +264,7 @@ def _boot_inner(self) -> None: self._check_inline_stores() self._cursor_persist() log.info( - "duckling up: %s → %s (team %s=%s), cursor=%d, columns=%d", + "single-destination viaduck up: %s → %s (team %s=%s), cursor=%d, columns=%d", cfg.source_table, cfg.dest_table, cfg.team_field, @@ -367,7 +367,7 @@ def _ensure_state_table(self) -> None: ) except (psycopg.errors.DuplicateSchema, psycopg.errors.DuplicateTable, psycopg.errors.UniqueViolation): pass # concurrent first boot (state.py's race guard) - # Duckling-managed additive column (the fleet never reads it): the + # SingleDestinationViaduck-managed additive column (the fleet never reads it): the # table_id the cursor position was earned against. pg.execute("ALTER TABLE viaduck.viaduck_state ADD COLUMN IF NOT EXISTS source_table_id bigint") @@ -388,10 +388,10 @@ def _cursor_load(self) -> None: self._cursor_row_exists = True stored_tid = row[1] if stored_tid is not None and int(stored_tid) != self.table_id: - raise FatalDucklingError( + raise FatalSingleDestinationError( f"cursor was earned against table_id={int(stored_tid)} but the source table is now " - f"table_id={self.table_id}: the source was dropped+recreated while this duckling was " - "down — mandatory re-seed (delete the cursor row to restart at head)" + f"table_id={self.table_id}: the source was dropped+recreated while this " + "single-destination viaduck was down — mandatory re-seed (delete the cursor row to restart at head)" ) if stored_tid is None: # Backfill provenance (fleet-transplanted cursors arrive @@ -449,7 +449,7 @@ def _cursor_write(self, sql: str, params, what: str) -> None: "%s failed (attempt %d/%d): %s", what, attempt + 1, self.cfg.attempts, scrub_credentials(str(e)) ) time.sleep(0.5 * (attempt + 1)) - raise FatalDucklingError( + raise FatalSingleDestinationError( f"{what} failed after {self.cfg.attempts} attempts: {scrub_credentials(str(last_err))}" ) @@ -525,7 +525,9 @@ def _assert_no_deletes(self) -> None: n = self._count(sql, params) if n: assertion_failures_total.labels(check=name).inc() - raise FatalDucklingError(f"append-only contract violated: {n} rows in {name} for table_id={tid}") + raise FatalSingleDestinationError( + f"append-only contract violated: {n} rows in {name} for table_id={tid}" + ) # Retention-lived witness: survives the delete+merge-between-polls # race that erases the two checks above. changes_made is a # comma-joined `type:table_id` list (ducklake_transaction.cpp @@ -538,7 +540,7 @@ def _assert_no_deletes(self) -> None: ) if witness: assertion_failures_total.labels(check="snapshot_changes_witness").inc() - raise FatalDucklingError(f"delete/drop activity for table_id={tid} in snapshot_changes") + raise FatalSingleDestinationError(f"delete/drop activity for table_id={tid} in snapshot_changes") # Inlined deletes: their own per-table PG table, invisible to all of # the above (round-2 C2). Existence probe first, then count. store = ( @@ -556,7 +558,7 @@ def _assert_no_deletes(self) -> None: ) if n: assertion_failures_total.labels(check="inlined_delete").inc() - raise FatalDucklingError(f"{n} inlined deletes for table_id={tid}") + raise FatalSingleDestinationError(f"{n} inlined deletes for table_id={tid}") def _check_inline_stores(self) -> None: """Inlining is an attach-scoped writer option — unverifiable in the @@ -623,7 +625,7 @@ def _append_once(self, batch: pa.Table) -> None: for this class). BY NAME maps by column name and default-fills dest-managed columns.""" conn = self.dst_catalog.connection - view = "_duckling_append" + view = "_single_destination_append" conn.register(view, batch) try: conn.execute(f"INSERT INTO {self._dest_fqn} BY NAME SELECT * FROM {view}") @@ -650,7 +652,7 @@ def _append(self, batch: pa.Table) -> None: ) self._aimd_halve(f"flush failure: {type(e).__name__}") time.sleep(1.0 * (attempt + 1)) - raise FatalDucklingError( + raise FatalSingleDestinationError( f"append failed after {self.cfg.attempts} attempts: {scrub_credentials(str(last_err))}" ) @@ -681,7 +683,7 @@ def poll_once(self) -> None: lag_snapshots.set(head - self._cursor) # signed: a NEGATIVE lag is # a source regression (PITR restore / rebuild) — never clamp it to 0 if head < self._cursor: - raise FatalDucklingError( + raise FatalSingleDestinationError( f"source head {head} regressed below cursor {self._cursor}: the source was " "restored or rebuilt — the cursor's basis is gone; operator adjudication (re-seed)" ) @@ -708,7 +710,7 @@ def poll_once(self) -> None: except feed.FeedError as read_err: if "retained snapshot floor" in str(read_err): raise # transient in composition: next poll's clamp heals it - raise FatalDucklingError(scrub_credentials(str(read_err))) from None + raise FatalSingleDestinationError(scrub_credentials(str(read_err))) from None except Exception as read_err: # DROP+CREATE detection on the error path: a changed (or # vanished) table_id means the source was rebuilt — freeze and @@ -741,9 +743,9 @@ def _check_table_identity(self, cause: Exception | None) -> None: try: new_tid = self._resolve_table_id() except ConfigError: - raise FatalDucklingError("source table dropped — operator adjudication required") from cause + raise FatalSingleDestinationError("source table dropped — operator adjudication required") from cause if new_tid != self.table_id: - raise FatalDucklingError( + raise FatalSingleDestinationError( f"source table_id changed {self.table_id} → {new_tid}: DROP+CREATE requires re-seed" ) from cause @@ -751,7 +753,7 @@ def _arrow_filter(self, rows: pa.Table) -> pa.Table: try: col = rows.column(self.cfg.team_field) except Exception: - raise FatalDucklingError( + raise FatalSingleDestinationError( f"team field {self.cfg.team_field!r} missing from the read batch — schema contract broken" ) from None return rows.filter(pc.is_in(col, value_set=self._team_array)) @@ -765,12 +767,12 @@ def run(self) -> None: try: self.poll_once() self._last_poll_ok = time.monotonic() - except FatalDucklingError: + except FatalSingleDestinationError: raise # crash-class: die, let K8s restart except Exception: # Transient read/plan errors: the cursor never advanced, # the range is retried next poll. Append/cursor/ - # assertion failures are FatalDucklingError (above). + # assertion failures are FatalSingleDestinationError (above). polls_total.labels(result="error").inc() log.exception("poll failed (transient)") self._maybe_recycle() @@ -805,7 +807,7 @@ def is_healthy(self) -> bool: return (time.monotonic() - self._last_poll_ok) < max(3 * self.cfg.poll_interval_s, 300) -def _start_health_server(duck: Duckling) -> ThreadingHTTPServer: +def _start_health_server(duck: SingleDestinationViaduck) -> ThreadingHTTPServer: class Handler(BaseHTTPRequestHandler): def do_GET(self): if self.path == "/healthz": @@ -833,9 +835,9 @@ def log_message(self, *a): def main() -> None: setup_logging() source.sweep_spill_dirs() # crash-loops otherwise accumulate leftovers in the pod emptyDir - cfg = DucklingConfig.from_env() - metrics.init(f"duckling-{cfg.destination_id}") - duck = Duckling(cfg) + cfg = SingleDestinationConfig.from_env() + metrics.init(f"single-destination-{cfg.destination_id}") + duck = SingleDestinationViaduck(cfg) signal.signal(signal.SIGTERM, lambda *_: duck._stop.set()) duck.run()