publish: chained O(delta) snapshot id; backfill batches 20 symbols per version; heartbeat-based stale requeue; symbol index
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
5 changed files +59 −18
added
packages/core/src/findata_core/migrations/0003_lake_files_symbol_idx.sql
+1 −0
@@ -0,0 +1 @@ | ||
| 1 | +CREATE INDEX IF NOT EXISTS lake_files_symbol_idx ON lake_files(dataset_id, symbol_min) WHERE version_removed IS NULL; | |
modified
services/ingest/src/findata_ingest/lake.py
+18 −7
@@ -206,19 +206,30 @@ class Lake: | ||
| 206 | 206 | if row is None: |
| 207 | 207 | raise RuntimeError(f"unknown dataset {dataset_id}") |
| 208 | 208 | version = int(row["current_version"]) + 1 |
| 209 | + prev = conn.execute("SELECT snapshot_id, file_count, row_count, bytes FROM dataset_versions WHERE dataset_id=%s AND version=%s", | |
| 210 | + (dataset_id, version - 1)).fetchone() or {"snapshot_id": "", "file_count": 0, "row_count": 0, "bytes": 0} | |
| 211 | + retired_rows = [] | |
| 209 | 212 | for p in retire_paths: |
| 210 | − conn.execute("UPDATE lake_files SET version_removed=%s WHERE path=%s AND version_removed IS NULL", (version, p)) | |
| 213 | + r = conn.execute("UPDATE lake_files SET version_removed=%s WHERE path=%s AND version_removed IS NULL RETURNING path, row_count, bytes", | |
| 214 | + (version, p)).fetchone() | |
| 215 | + if r: | |
| 216 | + retired_rows.append(r) | |
| 211 | 217 | for f in added: |
| 212 | 218 | conn.execute( |
| 213 | 219 | "INSERT INTO lake_files(dataset_id, path, layout, symbol_min, symbol_max, date_min, date_max, row_count, bytes, sha256, version_added, ingest_run_id) " |
| 214 | 220 | "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", |
| 215 | 221 | (dataset_id, f.rel_path, f.layout, f.symbol_min, f.symbol_max, f.date_min, f.date_max, f.row_count, f.bytes, f.sha256, version, ingest_run_id)) |
| 216 | − live = conn.execute("SELECT path, sha256, row_count, bytes FROM lake_files WHERE dataset_id=%s AND version_removed IS NULL ORDER BY path", | |
| 217 | − (dataset_id,)).fetchall() | |
| 218 | − snapshot_id = hashlib.sha256("\n".join(f"{r['path']} {r['sha256']}" for r in live).encode()).hexdigest() | |
| 219 | − file_count = len(live) | |
| 220 | − row_count = sum(int(r["row_count"]) for r in live) | |
| 221 | − nbytes = sum(int(r["bytes"]) for r in live) | |
| 222 | + # Chained snapshot id (O(delta), reproducible): H(prev_snapshot ∥ sorted added (path, sha256) ∥ sorted retired paths). | |
| 223 | + # The full file list of any version is reconstructed from lake_files (snapshot_files()). | |
| 224 | + h = hashlib.sha256(prev["snapshot_id"].encode()) | |
| 225 | + for f in sorted(added, key=lambda x: x.rel_path): | |
| 226 | + h.update(f"+{f.rel_path} {f.sha256}\n".encode()) | |
| 227 | + for r in sorted(retired_rows, key=lambda x: x["path"]): | |
| 228 | + h.update(f"-{r['path']}\n".encode()) | |
| 229 | + snapshot_id = h.hexdigest() | |
| 230 | + file_count = int(prev["file_count"]) + len(added) - len(retired_rows) | |
| 231 | + row_count = int(prev["row_count"]) + sum(f.row_count for f in added) - sum(int(r["row_count"]) for r in retired_rows) | |
| 232 | + nbytes = int(prev["bytes"]) + sum(f.bytes for f in added) - sum(int(r["bytes"]) for r in retired_rows) | |
| 222 | 233 | conn.execute("INSERT INTO dataset_versions(dataset_id, version, snapshot_id, file_count, row_count, bytes, ingest_run_id, note) VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", |
| 223 | 234 | (dataset_id, version, snapshot_id, file_count, row_count, nbytes, ingest_run_id, note)) |
| 224 | 235 | conn.execute("UPDATE datasets SET current_version=%s, updated_at=now() WHERE id=%s", (version, dataset_id)) |
modified
services/ingest/src/findata_ingest/queue.py
+4 −2
@@ -75,8 +75,9 @@ def finish(job: dict[str, Any], ok: bool, result: dict[str, Any] | None = None, | ||
| 75 | 75 | |
| 76 | 76 | |
| 77 | 77 | def progress(job_id: str, data: dict[str, Any]) -> None: |
| 78 | + """Store progress and refresh the heartbeat (locked_at) used by requeue_stale().""" | |
| 78 | 79 | with transaction() as conn: |
| 79 | − conn.execute("UPDATE ingest_jobs SET progress=%s WHERE id=%s", (json.dumps(data, default=str), job_id)) | |
| 80 | + conn.execute("UPDATE ingest_jobs SET progress=%s, locked_at=now() WHERE id=%s", (json.dumps(data, default=str), job_id)) | |
| 80 | 81 | |
| 81 | 82 | |
| 82 | 83 | def set_checkpoint(scope: str, key: str, value: dict[str, Any]) -> None: |
@@ -92,7 +93,8 @@ def get_checkpoint(scope: str, key: str) -> dict[str, Any] | None: | ||
| 92 | 93 | |
| 93 | 94 | |
| 94 | 95 | def requeue_stale(max_age_minutes: int = 90) -> int: |
| 95 | − """Jobs left 'running' by a crashed worker are re-queued (idempotent tasks + checkpoints make this safe).""" | |
| 96 | + """Jobs 'running' without a heartbeat (progress) for `max_age_minutes` are re-queued — the worker is gone | |
| 97 | + (idempotent tasks + checkpoints make this safe). Long jobs stay alive by reporting progress.""" | |
| 96 | 98 | with transaction() as conn: |
| 97 | 99 | r = conn.execute("UPDATE ingest_jobs SET status='queued', locked_by=NULL, next_run_at=now() WHERE status='running' " |
| 98 | 100 | "AND locked_at < now() - make_interval(mins => %s) RETURNING id", (max_age_minutes,)).fetchall() |
modified
services/ingest/src/findata_ingest/tasks/backfill.py
+33 −9
@@ -26,6 +26,8 @@ from findata_ingest.tasks.common import ( | ||
| 26 | 26 | upsert_coverage, |
| 27 | 27 | ) |
| 28 | 28 | |
| 29 | +PUBLISH_BATCH = 20 | |
| 30 | + | |
| 29 | 31 | |
| 30 | 32 | def _release_memory() -> None: |
| 31 | 33 | """Return freed Arrow/Polars buffers to the OS between symbols (RSS otherwise grows to several GB per worker).""" |
@@ -83,6 +85,29 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | ||
| 83 | 85 | with connection() as conn: |
| 84 | 86 | delta_dates = live_delta_dates(conn, dataset_id) |
| 85 | 87 | stats = {"symbols": 0, "rows": 0, "files": 0, "skipped_rows_in_delta": 0, "anomalies": 0} |
| 88 | + pending_written: list = [] | |
| 89 | + pending_retire: list[str] = [] | |
| 90 | + pending_symbols: list[str] = [] | |
| 91 | + pending_tables: list = [] | |
| 92 | + | |
| 93 | + def _flush() -> None: | |
| 94 | + """Publish the accumulated symbols in one version (one transaction), then checkpoint.""" | |
| 95 | + if pending_written: | |
| 96 | + lake.publish(dataset_id, list(pending_written), sorted(set(pending_retire)), run_id, | |
| 97 | + note=f"backfill {asset} {letter} {timeframe} {adjustment} {len(pending_symbols)} symbols") | |
| 98 | + with transaction() as conn: | |
| 99 | + for t_, anomalies_ in pending_tables: | |
| 100 | + upsert_coverage(conn, dataset_id, t_, interval) | |
| 101 | + record_anomalies(conn, dataset_id, anomalies_, run_id, limit=40) | |
| 102 | + for sym in pending_symbols: | |
| 103 | + done.add(sym) | |
| 104 | + if pending_symbols: | |
| 105 | + queue.set_checkpoint(f"job:{job['id']}", "state", {"raw_path": raw_rel, "sha256": res.sha256, "bytes": res.bytes, | |
| 106 | + "fetched_at": res.fetched_at.isoformat(), "done_symbols": sorted(done)}) | |
| 107 | + queue.progress(job["id"], {"stage": "publish", **stats, "last_symbol": pending_symbols[-1]}) | |
| 108 | + for lst in (pending_written, pending_retire, pending_symbols, pending_tables): | |
| 109 | + lst.clear() | |
| 110 | + | |
| 86 | 111 | for batch in conn_fr.normalize(res, symbols=list(only) if only else None): |
| 87 | 112 | if batch.symbol in done: |
| 88 | 113 | continue |
@@ -105,25 +130,24 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | ||
| 105 | 130 | for y in sorted(set(years.to_pylist())): |
| 106 | 131 | ty = t.filter(pc.equal(years, y)) |
| 107 | 132 | written.append(lake.write_table(dataset_id, ty, "base", {"symbol": batch.symbol, "year": str(y)})) |
| 108 | − # retire previous base files of this symbol/year (re-backfill replaces) | |
| 133 | + # retire previous base files of this symbol (re-backfill / full_update replaces the whole symbol history) | |
| 109 | 134 | with connection() as conn: |
| 110 | 135 | old = conn.execute("SELECT path FROM lake_files WHERE dataset_id=%s AND layout='base' AND version_removed IS NULL AND symbol_min=%s AND symbol_max=%s", |
| 111 | 136 | (dataset_id, batch.symbol, batch.symbol)).fetchall() |
| 112 | − lake.publish(dataset_id, written, [o["path"] for o in old], run_id, note=f"backfill {asset} {letter} {timeframe} {adjustment} {batch.symbol}") | |
| 113 | − with transaction() as conn: | |
| 114 | − upsert_coverage(conn, dataset_id, t, interval) | |
| 115 | − record_anomalies(conn, dataset_id, rep.anomalies, run_id, limit=40) | |
| 137 | + pending_written.extend(written) | |
| 138 | + pending_retire.extend(o["path"] for o in old) | |
| 139 | + pending_symbols.append(batch.symbol) | |
| 140 | + pending_tables.append((t, rep.anomalies)) | |
| 116 | 141 | stats["symbols"] += 1 |
| 117 | 142 | stats["rows"] += t.num_rows |
| 118 | 143 | stats["files"] += len(written) |
| 119 | 144 | stats["anomalies"] += len(rep.anomalies) |
| 120 | − done.add(batch.symbol) | |
| 121 | 145 | del t, rep, written |
| 146 | + if len(pending_symbols) >= PUBLISH_BATCH: | |
| 147 | + _flush() | |
| 122 | 148 | if stats["symbols"] % 10 == 0: |
| 123 | 149 | _release_memory() |
| 124 | − queue.set_checkpoint(f"job:{job['id']}", "state", {"raw_path": raw_rel, "sha256": res.sha256, "bytes": res.bytes, | |
| 125 | − "fetched_at": res.fetched_at.isoformat(), "done_symbols": sorted(done)}) | |
| 126 | − queue.progress(job["id"], {"stage": "publish", **stats, "last_symbol": batch.symbol}) | |
| 150 | + _flush() | |
| 127 | 151 | with transaction() as conn: |
| 128 | 152 | conn.execute("UPDATE raw_archives SET status='published' WHERE sha256=%s", (res.sha256,)) |
| 129 | 153 | if stats["rows"]: |
modified
services/ingest/src/findata_ingest/worker.py
+3 −0
@@ -30,6 +30,9 @@ def schedule_tick(log) -> None: | ||
| 30 | 30 | """Enqueue recurring jobs (dedupe keys make this idempotent). Times in America/New_York.""" |
| 31 | 31 | now = datetime.now(ET) |
| 32 | 32 | day = now.date().isoformat() |
| 33 | + stale = queue.requeue_stale() | |
| 34 | + if stale: | |
| 35 | + log.warning("scheduler.requeued_stale", n=stale) | |
| 33 | 36 | # vendor daily update lands ~23:30–03:00 ET → poll from 02:00 every 30 min (incremental skips when vendor unchanged) |
| 34 | 37 | if now.weekday() < 6 and (now.hour >= 2 or now.hour < 1): |
| 35 | 38 | queue.enqueue("incremental", "market_bars.stock.1m.raw", {"type": "stock", "period": "day"}, priority=20, |
| 36 | 39 | |