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 =============================================