From 1405ce9c05c75df7c0e4c6cf288249d02c3b58a8 Mon Sep 17 00:00:00 2001 From: jakeross Date: Wed, 19 Aug 2026 13:12:57 -0700 Subject: [PATCH] fix(ingestion): stop duplicate instants reaching one INSERT The first live materialization failed: ON CONFLICT DO UPDATE command cannot affect row a second time HINT: Ensure that no rows proposed for insertion within the same command have duplicate constrained values. Two causes, both here. Diver-HUB ranges are inclusive at both ends -- "up to and including end time" -- while iter_windows made adjacent windows share a boundary, so a reading logged exactly on it came back in both. And the loader never deduplicated, so that pair reached Postgres in one statement. Windows now leave exactly one second between them. Timestamps are second-resolution, so nothing falls in the gap. The loader also deduplicates within a batch, keeping the last occurrence -- which matches the upsert's own rule that a later value wins. That guard holds whatever the source does, including a vendor logging one instant twice. The existing idempotency test could not have caught this: it loads the same window in two separate statements, which Postgres allows. The new test puts the duplicates in one batch, which is what actually happened. Co-Authored-By: Claude Opus 5 --- automated_ingestion/ocotillo/loader.py | 13 +++++++-- automated_ingestion/shared/windows.py | 11 ++++++-- automated_ingestion/tests/test_windows.py | 18 ++++++++++--- tests/test_transducer_loader.py | 32 +++++++++++++++++++++++ 4 files changed, 67 insertions(+), 7 deletions(-) diff --git a/automated_ingestion/ocotillo/loader.py b/automated_ingestion/ocotillo/loader.py index 7b831bf82..cabafccf5 100644 --- a/automated_ingestion/ocotillo/loader.py +++ b/automated_ingestion/ocotillo/loader.py @@ -119,6 +119,14 @@ def load_observations( table = TransducerObservation.__table__ for batch in _batched(records, batch_size): + result.rows_seen += len(batch) + + # One row per instant within a statement. Postgres refuses an + # ON CONFLICT DO UPDATE that would touch the same row twice in one + # command, and a source can repeat a reading -- overlapping fetch + # windows, or a vendor logging the same instant twice. Keeping the last + # occurrence matches the upsert's own rule: a later value wins. + deduplicated = {record.observation_datetime: record for record in batch} rows = [ { "deployment_id": deployment_id, @@ -128,9 +136,10 @@ def load_observations( "release_status": release_status, "data_maturity": data_maturity, } - for record in batch + for record in deduplicated.values() ] - result.rows_seen += len(rows) + if not rows: + continue statement = insert(table).values(rows) # DO UPDATE rather than DO NOTHING: a vendor may correct a reading, and diff --git a/automated_ingestion/shared/windows.py b/automated_ingestion/shared/windows.py index 9fc8765d6..039eff232 100644 --- a/automated_ingestion/shared/windows.py +++ b/automated_ingestion/shared/windows.py @@ -74,10 +74,17 @@ def iter_windows(start: int, end: int, span: int = DEFAULT_SPAN) -> Iterator[Win raise ValueError(f"Window span must be positive, got {span}.") if end < start: raise ValueError(f"End {end} precedes start {start}.") + # Windows must not share a boundary. Diver-HUB's ranges are inclusive at + # both ends -- "from start time up to and including end time" -- so + # [0, span] and [span, 2*span] both return the reading logged exactly at + # `span`. That duplicate reaches the loader in one batch and Postgres + # rejects the statement: "ON CONFLICT DO UPDATE command cannot affect row a + # second time". cursor = start while cursor < end: - yield Window(cursor, min(cursor + span, end)) - cursor += span + chunk_end = min(cursor + span, end) + yield Window(cursor, chunk_end) + cursor = chunk_end + 1 # ============= EOF ============================================= diff --git a/automated_ingestion/tests/test_windows.py b/automated_ingestion/tests/test_windows.py index 34d68602d..d4d9cf2ab 100644 --- a/automated_ingestion/tests/test_windows.py +++ b/automated_ingestion/tests/test_windows.py @@ -25,12 +25,24 @@ ) -def test_windows_cover_the_range_without_gaps_or_overlap(): +def test_windows_do_not_share_a_boundary(): + # Diver-HUB ranges are inclusive at both ends, so touching windows both + # return the reading logged exactly on the boundary. That duplicate reaches + # the loader in one batch and Postgres rejects the statement: "ON CONFLICT + # DO UPDATE command cannot affect row a second time". windows = list(iter_windows(0, 10 * DAY, span=3 * DAY)) assert windows[0].start == 0 assert windows[-1].end == 10 * DAY for earlier, later in zip(windows, windows[1:]): - assert earlier.end == later.start + assert later.start == earlier.end + 1 + + +def test_windows_leave_no_second_uncovered(): + # The gap is exactly one second and timestamps are second-resolution, so + # nothing can fall between two windows. + windows = list(iter_windows(0, 10 * DAY, span=3 * DAY)) + for earlier, later in zip(windows, windows[1:]): + assert later.start - earlier.end == 1 def test_final_window_is_truncated_not_overshot(): @@ -38,7 +50,7 @@ def test_final_window_is_truncated_not_overshot(): # and at worst a 400. windows = list(iter_windows(0, 10 * DAY, span=3 * DAY)) assert windows[-1].end == 10 * DAY - assert windows[-1].span == DAY + assert all(w.end <= 10 * DAY for w in windows) def test_range_shorter_than_span_is_a_single_window(): diff --git a/tests/test_transducer_loader.py b/tests/test_transducer_loader.py index 8913713da..d9a42d42a 100644 --- a/tests/test_transducer_loader.py +++ b/tests/test_transducer_loader.py @@ -268,4 +268,36 @@ def test_rows_with_no_recorded_maturity_still_update(loader_target): assert value == 99.0 +def test_duplicate_instants_in_one_batch_do_not_break_the_statement(loader_target): + """Reproduces the first live run's failure. + + Postgres rejects an ON CONFLICT DO UPDATE that would touch the same row + twice in one command: + + ON CONFLICT DO UPDATE command cannot affect row a second time + + The existing idempotency test loads the same window in two separate + statements, which Postgres allows, so it could not catch this. A source can + repeat an instant -- overlapping fetch windows did, and a vendor may log one + twice. + """ + deployment_id, parameter_id = loader_target + duplicated = _records(3) + _records(3, value=99.0) + + with session_ctx() as session: + result = load_observations( + session, duplicated, deployment_id, parameter_id, "draft" + ) + assert _count(session, deployment_id) == 3 + assert result.rows_seen == 6 + + # The later value wins, matching the upsert's own rule. + values = session.scalars( + select(TransducerObservation.value) + .where(TransducerObservation.deployment_id == deployment_id) + .order_by(TransducerObservation.observation_datetime) + ).all() + assert values[0] == 99.0 + + # ============= EOF =============================================