feat(ingestion): derive the watermark from Postgres

automated_ingestion/shared/watermark.py

@@ -0,0 +1,108 @@
+# ===============================================================================
+# Copyright 2026 ross
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ===============================================================================
+"""
+Where a series got to, asked of the data rather than of a sidecar.
+
+Aqueduct keeps watermarks in a GCS object beside the raw zone, because its
+destination is FROST and cannot be queried cheaply for a maximum. Ocotillo's
+destination is Postgres, so the watermark is simply
+``MAX(observation_datetime)`` for the series.
+
+**This is a deliberate divergence, not an oversight.** A stored watermark is a
+second source of truth about what was loaded, and the two drift: a load that
+half-succeeds, or a sidecar write that fails after the rows commit, leaves the
+watermark claiming more or less than the data holds. Deriving it means the
+answer cannot disagree with reality — and it makes a backfill safe by
+construction, since re-loading an old window cannot move a maximum forward.
+
+**Keyed by thing, not by deployment.** Observations carry ``deployment_id``, but
+a series outlives its hardware: replacing a diver creates a new deployment for
+the same well, and a watermark keyed to the deployment would report nothing for
+the new one and re-fetch the entire history. The query joins through
+``deployment`` to ask the question the pipeline actually has -- how far along is
+this well's depth-to-water record.
+"""
+
+from datetime import datetime
+from typing import Any, Protocol
+
+
+class WatermarkStore(Protocol):
+    """Where a series has been loaded up to."""
+
+    def get(self, thing_id: int, parameter_id: int) -> datetime | None:
+        """Latest observation for the series, or ``None`` if never loaded."""
+        ...
+
+
+class PostgresWatermarkStore:
+    """Reads the watermark from the observations themselves.
+
+    Takes the session the loader is using, so the watermark reflects that
+    session's committed state rather than a separate connection's snapshot.
+    """
+
+    def __init__(self, session: Any) -> None:
+        self._session = session
+
+    def get(self, thing_id: int, parameter_id: int) -> datetime | None:
+        from sqlalchemy import func, select
+
+        from db.deployment import Deployment
+        from db.transducer import TransducerObservation
+
+        return self._session.scalar(
+            select(func.max(TransducerObservation.observation_datetime))
+            .join(
+                Deployment,
+                Deployment.id == TransducerObservation.deployment_id,
+            )
+            .where(Deployment.thing_id == thing_id)
+            .where(TransducerObservation.parameter_id == parameter_id)
+        )
+
+
+class InMemoryWatermarkStore:
+    """For tests, and for reasoning about a run without a database."""
+
+    def __init__(self, watermarks: dict[tuple[int, int], datetime] | None = None):
+        self._watermarks = dict(watermarks or {})
+
+    def get(self, thing_id: int, parameter_id: int) -> datetime | None:
+        return self._watermarks.get((thing_id, parameter_id))
+
+    def set(self, thing_id: int, parameter_id: int, value: datetime) -> None:
+        self._watermarks[(thing_id, parameter_id)] = value
+
+
+def resolve_start(
+    store: WatermarkStore,
+    thing_id: int,
+    parameter_id: int,
+    floor: datetime,
+) -> datetime:
+    """Where the next fetch should begin.
+
+    ``floor`` applies only to a series that has never been loaded. It is not a
+    backfill lever: lowering it will not re-fetch history for a series whose
+    watermark has already advanced past it, because the watermark wins whenever
+    one exists. Re-fetching history is what the backfill jobs are for.
+    """
+    watermark = store.get(thing_id, parameter_id)
+    return watermark if watermark is not None else floor
+
+
+# ============= EOF =============================================

docs/automated-ingestion-pipeline-plan.md

@@ -292,10 +292,18 @@ Built. Migration `a1b2c3d4e5f6`, loader in `automated_ingestion/ocotillo/loader.
 
 ### 3.5 — Watermark from Postgres
 
-- Keep Aqueduct's `WatermarkStore` interface; Postgres implementation returns `MAX(observation_datetime)` for a `(thing_id, parameter_id)`, read in the same session as the write. No GCS sidecar for normal runs.
-- Backfill never advances the normal watermark implicitly — inherent with upsert, but asserted in a test.
-- In-memory implementation kept for tests. First-ever run for a series falls back to the `initial_start_date` floor.
-- Divergence from Aqueduct recorded in the module docstring, so it reads as a decision not an oversight.
+Built. `automated_ingestion/shared/watermark.py`, seven tests.
+
+- ✅ `PostgresWatermarkStore` returns `MAX(observation_datetime)` for the series, read through the loader's own session so it reflects that session's committed state rather than another connection's snapshot.
+- ✅ `InMemoryWatermarkStore` for tests and for reasoning about a run without a database.
+- ✅ `resolve_start` falls back to the `initial_start_date` floor only for a series that has never been loaded. A test asserts a floor *ahead* of the watermark does not win either — the floor is not a backfill lever in any direction.
+- ✅ The divergence from Aqueduct is in the module docstring, so it reads as a decision rather than an oversight.
+
+**Keyed by thing, not deployment.** This section said `(thing_id, parameter_id)` and that turns out to be right for a reason worth stating: observations carry `deployment_id`, but a series outlives its hardware. Replacing a diver creates a new deployment for the same well, and a watermark keyed to the deployment would report nothing for the new one and re-fetch the entire history. The query joins through `deployment` to ask the question the pipeline actually has.
+
+**Why derive rather than store.** A stored watermark is a second source of truth about what was loaded, and the two drift — a half-succeeded load, or a sidecar write that fails after the rows commit, leaves it claiming more or less than the data holds. Aqueduct stores one because FROST cannot be queried cheaply for a maximum; Postgres can.
+
+The payoff is that "backfill never advances the normal watermark" stops being a rule to enforce and becomes a property that cannot be violated: re-loading a window behind the maximum cannot move a maximum forward. Asserted anyway, in two directions — older data, and the same window twice.
 
 ---
 

tests/test_watermark.py

@@ -0,0 +1,131 @@
+# ===============================================================================
+# Copyright 2026 ross
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ===============================================================================
+"""
+Watermark behaviour, including the property that makes backfill safe.
+"""
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import delete, select
+
+from automated_ingestion.ocotillo.loader import load_observations
+from automated_ingestion.ocotillo.structs import ObservationRecord
+from automated_ingestion.shared.watermark import (
+    InMemoryWatermarkStore,
+    PostgresWatermarkStore,
+    resolve_start,
+)
+from db.engine import session_ctx
+from db.parameter import Parameter
+from db.transducer import TransducerObservation
+
+START = datetime(2026, 1, 1, tzinfo=timezone.utc)
+FLOOR = datetime(2015, 1, 1, tzinfo=timezone.utc)
+
+
+@pytest.fixture()
+def series(sensor_to_water_well_thing_deployment):
+    """A deployment, its thing, and a parameter, cleaned up afterwards."""
+    deployment = sensor_to_water_well_thing_deployment
+    with session_ctx() as session:
+        parameter_id = session.scalar(select(Parameter.id).limit(1))
+        yield deployment.id, deployment.thing_id, parameter_id
+        session.execute(
+            delete(TransducerObservation).where(
+                TransducerObservation.deployment_id == deployment.id
+            )
+        )
+        session.commit()
+
+
+def _records(count, start=START, value=10.0):
+    return [
+        ObservationRecord(
+            external_point_id="sanacaciareach-40",
+            observation_datetime=start + timedelta(minutes=5 * i),
+            value=value,
+            units="ft",
+        )
+        for i in range(count)
+    ]
+
+
+class TestPostgresWatermark:
+    def test_unloaded_series_has_no_watermark(self, series):
+        deployment_id, thing_id, parameter_id = series
+        with session_ctx() as session:
+            store = PostgresWatermarkStore(session)
+            assert store.get(thing_id, parameter_id) is None
+
+    def test_watermark_is_the_latest_observation(self, series):
+        deployment_id, thing_id, parameter_id = series
+        with session_ctx() as session:
+            load_observations(
+                session, _records(10), deployment_id, parameter_id, "draft"
+            )
+            store = PostgresWatermarkStore(session)
+            assert store.get(thing_id, parameter_id) == START + timedelta(minutes=45)
+
+    def test_backfilling_older_data_does_not_advance_it(self, series):
+        # The property that makes backfill safe: a watermark derived from the
+        # data cannot be moved forward by re-loading a window behind it. A
+        # stored watermark has to be defended against this; a derived one
+        # cannot have the problem.
+        deployment_id, thing_id, parameter_id = series
+        with session_ctx() as session:
+            load_observations(
+                session, _records(10), deployment_id, parameter_id, "draft"
+            )
+            store = PostgresWatermarkStore(session)
+            before = store.get(thing_id, parameter_id)
+
+            older = _records(10, start=START - timedelta(days=365))
+            load_observations(session, older, deployment_id, parameter_id, "draft")
+
+            assert store.get(thing_id, parameter_id) == before
+
+    def test_reloading_the_same_window_does_not_move_it(self, series):
+        deployment_id, thing_id, parameter_id = series
+        with session_ctx() as session:
+            load_observations(
+                session, _records(5), deployment_id, parameter_id, "draft"
+            )
+            store = PostgresWatermarkStore(session)
+            before = store.get(thing_id, parameter_id)
+            load_observations(
+                session, _records(5), deployment_id, parameter_id, "draft"
+            )
+            assert store.get(thing_id, parameter_id) == before
+
+
+class TestResolveStart:
+    def test_floor_applies_only_to_a_new_series(self):
+        assert resolve_start(InMemoryWatermarkStore(), 1, 2, FLOOR) == FLOOR
+
+    def test_watermark_wins_over_the_floor(self):
+        # The floor is not a backfill lever: lowering it must not re-fetch
+        # history for a series that has already advanced past it.
+        store = InMemoryWatermarkStore({(1, 2): START})
+        assert resolve_start(store, 1, 2, FLOOR) == START
+
+    def test_a_floor_ahead_of_the_watermark_does_not_win_either(self):
+        store = InMemoryWatermarkStore({(1, 2): START})
+        ahead = START + timedelta(days=365)
+        assert resolve_start(store, 1, 2, ahead) == START
+
+
+# ============= EOF =============================================