From af9c088b848ff57d31fca0604906b04db2a12bbb Mon Sep 17 00:00:00 2001 From: kert Date: Sun, 25 Oct 2026 16:21:12 -0400 Subject: [PATCH] fix(bcda): legacy cut-off fallback is the job's start, not its completion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A job recorded before manifests existed has no transactionTime. The first version fell back to the instant the `complete` row was written and called it "safe, slightly overlapping" — I had that backwards: a _since later than the server's true cut-off is a gap, not an overlap. The real store showed it: the sandbox stamps transactionTime as midnight UTC of the run day, so a baseline that finished at 16:53 was being credited with a 16:53 cut-off and every same-day delta (00:00) was read as predating it. The fallback is now the job's first ledger row (the request instant), which can only overlap; Store.job_started_at() reads it. `stack bcda resume ` still replaces the fallback with the server's value — done for the four pre-manifest jobs in the real store, whose watermarks now read 2026-10-25T00:00:00Z. refs #1139 --- ...6-10-25-bcda-since-reconciliation-design.md | 8 ++++++-- src/bcda/store.py | 12 ++++++++++++ src/bcda/sync.py | 15 ++++++++++----- tests/bcda/test_sync.py | 18 ++++++++++++++++-- 4 files changed, 44 insertions(+), 9 deletions(-) diff --git a/docs/superpowers/specs/2026-10-25-bcda-since-reconciliation-design.md b/docs/superpowers/specs/2026-10-25-bcda-since-reconciliation-design.md index 4c4837b..b5ad30e 100644 --- a/docs/superpowers/specs/2026-10-25-bcda-since-reconciliation-design.md +++ b/docs/superpowers/specs/2026-10-25-bcda-since-reconciliation-design.md @@ -192,8 +192,12 @@ Details settled while building: - The store now writes `exports//manifest.json` (the completed-job body) and puts `transaction_time` and `request` on the job row; jobs recorded before that fall back to the manifest, then to the instant the - `complete` row was written (later than the true cut-off, so a safe, - overlapping `_since`; the ledger records which source was used). An + job was *started* (its first ledger row): the server's snapshot cannot + be newer than when it began processing, so that `_since` overlaps and + never gaps, whereas the completion instant would be later than the true + cut-off and could skip updates made while the job ran. The ledger + records which source was used, and `stack bcda resume ` replaces + the fallback with the server's value while the job is still listed. An adopted job's `_since` is read from its request line. - Current tables carry `_job_key`, `_transaction_time`, `_since`, `_kind`, `_applied_seq`; the connector's `_ensure()` drops `_`-prefixed columns diff --git a/src/bcda/store.py b/src/bcda/store.py index dda059f..5f6c8bc 100644 --- a/src/bcda/store.py +++ b/src/bcda/store.py @@ -240,6 +240,18 @@ class Store: return row raise KeyError(f"Job not found: {key}") + def job_started_at(self, key: str) -> datetime | None: + """Instant of the first transition recorded for *key* (the request), or ``None``. + + ``bcda.sync`` uses it as the conservative cut-off for a job whose + manifest and ``transaction_time`` are both missing: the server's + data cannot be newer than the moment it started processing the + request, so a ``_since`` at the start only overlaps, never gaps. + """ + table = self._read_parquet(self._jobs_path, JOBS_SCHEMA) + rows = [r for r in table.to_pylist() if r["key"] == key and r.get("ts")] + return min(r["ts"] for r in rows) if rows else None + def list_jobs(self, *, status: str | None = None) -> list[dict]: """List all jobs, optionally filtered by status.""" table = self._read_parquet(self._jobs_path, JOBS_SCHEMA) diff --git a/src/bcda/sync.py b/src/bcda/sync.py index a956d78..bf7dc78 100644 --- a/src/bcda/sync.py +++ b/src/bcda/sync.py @@ -162,7 +162,7 @@ class JobInfo: type_filter: str since: str transaction_time: datetime - transaction_source: str # manifest | job_row | completed_at + transaction_source: str # job_row | manifest | started_at ts: datetime @property @@ -188,8 +188,13 @@ def job_info(store: Store, job: dict) -> JobInfo: ``transactionTime`` comes from the job row (new ledgers), else the saved manifest, else — for jobs recorded before either existed — the - instant the ``complete`` row was written (later than the true - cut-off, so a safe, slightly overlapping ``_since``). + instant the job was *started* (its first ledger row). The server's + snapshot cannot be newer than the moment it began processing the + request, so a ``_since`` at the start overlaps and never gaps; the + completion instant would be *later* than the true cut-off and could + skip updates made while the job ran. ``Store.resume`` replaces the + fallback with the server's value whenever the job is still on the + server. """ types, filt = split_types(job.get("types") or "") tt = _maybe_instant(job.get("transaction_time")) @@ -198,8 +203,8 @@ def job_info(store: Store, job: dict) -> JobInfo: tt = _maybe_instant(store.manifest(job["key"]).get("transactionTime")) source = "manifest" if tt is None: - tt = _maybe_instant(job.get("ts")) - source = "completed_at" + tt = store.job_started_at(job["key"]) or _maybe_instant(job.get("ts")) + source = "started_at" return JobInfo( key=job["key"], endpoint=job.get("endpoint") or "Group/all", diff --git a/tests/bcda/test_sync.py b/tests/bcda/test_sync.py index 165bade..91e9677 100644 --- a/tests/bcda/test_sync.py +++ b/tests/bcda/test_sync.py @@ -323,11 +323,25 @@ class TestChecks: def test_job_info_falls_back_for_legacy_rows(self, tmp_path: Path) -> None: store = Store(path=str(tmp_path)) + t_start = datetime(2026, 10, 1, 9, 0, tzinfo=timezone.utc) store._append_job( - key="OLD", endpoint="Group/all", types="Patient", status="complete" + key="OLD", + endpoint="Group/all", + types="Patient", + status="polling", + ts=t_start, + ) + store._append_job( + key="OLD", + endpoint="Group/all", + types="Patient", + status="complete", + ts=t_start + timedelta(hours=1), ) info = sync.job_info(store, store.get_job("OLD")) - assert info.transaction_source == "completed_at" and info.kind == "baseline" + # the conservative fallback is the *start* of the job, not its completion + assert info.transaction_source == "started_at" and info.kind == "baseline" + assert info.transaction_time == t_start store._write_manifest("OLD", {"transactionTime": "2026-08-01T00:00:00.000Z"}) info = sync.job_info(store, store.get_job("OLD")) assert info.transaction_source == "manifest"