lake/client: EINTR-safe mkdir on the external volume; cap per-symbol anomaly rows in backfills
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
3 changed files +22 −5
modified
connectors/market/src/findata_market_firstrate/client.py
+6 −1
@@ -111,7 +111,12 @@ class FRDClient: | |||||
| 111 | # ----------------------------------------------------------------- archives | 111 | # ----------------------------------------------------------------- archives | ||
| 112 | def download(self, endpoint: str, params: dict[str, Any], dest: Path) -> Downloaded: | 112 | def download(self, endpoint: str, params: dict[str, Any], dest: Path) -> Downloaded: | ||
| 113 | """Stream an archive to `dest` (atomic: `.part` then rename), computing SHA-256 on the fly.""" | 113 | """Stream an archive to `dest` (atomic: `.part` then rename), computing SHA-256 on the fly.""" | ||
| 114 | − dest.parent.mkdir(parents=True, exist_ok=True) | 114 | + for i in range(20): | ||
| 115 | + try: | ||||
| 116 | + dest.parent.mkdir(parents=True, exist_ok=True) | ||||
| 117 | + break | ||||
| 118 | + except InterruptedError: | ||||
| 119 | + time.sleep(0.2 * (i + 1)) | ||||
| 115 | part = dest.with_suffix(dest.suffix + ".part") | 120 | part = dest.with_suffix(dest.suffix + ".part") | ||
| 116 | last: Exception | None = None | 121 | last: Exception | None = None | ||
| 117 | for attempt in range(1, MAX_RETRIES + 1): | 122 | for attempt in range(1, MAX_RETRIES + 1): | ||
modified
services/ingest/src/findata_ingest/lake.py
+15 −3
@@ -47,6 +47,18 @@ class WrittenFile: | |||||
| 47 | sha256: str | 47 | sha256: str | ||
| 48 | 48 | ||||
| 49 | 49 | ||||
| 50 | +def mkdir_retry(path: Path, attempts: int = 20) -> None: | ||||
| 51 | + """mkdir -p that survives EINTR (seen on a Thunderbolt APFS volume while Spotlight was indexing it).""" | ||||
| 52 | + import time | ||||
| 53 | + for i in range(attempts): | ||||
| 54 | + try: | ||||
| 55 | + path.mkdir(parents=True, exist_ok=True) | ||||
| 56 | + return | ||||
| 57 | + except InterruptedError: | ||||
| 58 | + time.sleep(0.2 * (i + 1)) | ||||
| 59 | + path.mkdir(parents=True, exist_ok=True) | ||||
| 60 | + | ||||
| 61 | + | ||||
| 50 | class WriterLock: | 62 | class WriterLock: | ||
| 51 | def __init__(self, lake: Path): | 63 | def __init__(self, lake: Path): | ||
| 52 | self.path = lake / ".writer.lock" | 64 | self.path = lake / ".writer.lock" | ||
@@ -107,7 +119,7 @@ class Lake: | |||||
| 107 | 119 | ||||
| 108 | def raw_dir(self, source_id: str, dataset_id: str) -> Path: | 120 | def raw_dir(self, source_id: str, dataset_id: str) -> Path: | ||
| 109 | p = (self.cold or self.root) / "raw" / source_id / dataset_id / str(datetime.now(UTC).year) | 121 | p = (self.cold or self.root) / "raw" / source_id / dataset_id / str(datetime.now(UTC).year) | ||
| 110 | − p.mkdir(parents=True, exist_ok=True) | 122 | + mkdir_retry(p) | ||
| 111 | return p | 123 | return p | ||
| 112 | 124 | ||||
| 113 | def abs(self, rel: str) -> Path: | 125 | def abs(self, rel: str) -> Path: | ||
@@ -175,7 +187,7 @@ class Lake: | |||||
| 175 | with open(tmp, "rb") as f: | 187 | with open(tmp, "rb") as f: | ||
| 176 | os.fsync(f.fileno()) | 188 | os.fsync(f.fileno()) | ||
| 177 | digest = _sha256(tmp) | 189 | digest = _sha256(tmp) | ||
| 178 | − final.parent.mkdir(parents=True, exist_ok=True) | 190 | + mkdir_retry(final.parent) | ||
| 179 | os.replace(tmp, final) | 191 | os.replace(tmp, final) | ||
| 180 | import pyarrow.compute as pc | 192 | import pyarrow.compute as pc | ||
| 181 | syms = table.column("symbol_as_reported") if is_bars else table.column("underlying") | 193 | syms = table.column("symbol_as_reported") if is_bars else table.column("underlying") | ||
@@ -223,7 +235,7 @@ class Lake: | |||||
| 223 | if src.exists(): | 235 | if src.exists(): | ||
| 224 | r = self._root_of_rel(p) | 236 | r = self._root_of_rel(p) | ||
| 225 | dst = r / "_retired" / (p[len("curated/"):] if p.startswith("curated/") else p) | 237 | dst = r / "_retired" / (p[len("curated/"):] if p.startswith("curated/") else p) | ||
| 226 | − dst.parent.mkdir(parents=True, exist_ok=True) | 238 | + mkdir_retry(dst.parent) | ||
| 227 | shutil.move(str(src), str(dst)) | 239 | shutil.move(str(src), str(dst)) | ||
| 228 | return {"version": version, "snapshot_id": snapshot_id, "file_count": file_count, "row_count": row_count, "bytes": nbytes} | 240 | return {"version": version, "snapshot_id": snapshot_id, "file_count": file_count, "row_count": row_count, "bytes": nbytes} | ||
| 229 | 241 | ||||
modified
services/ingest/src/findata_ingest/tasks/backfill.py
+1 −1
@@ -103,7 +103,7 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | |||||
| 103 | lake.publish(dataset_id, written, [o["path"] for o in old], run_id, note=f"backfill {asset} {letter} {timeframe} {adjustment} {batch.symbol}") | 103 | lake.publish(dataset_id, written, [o["path"] for o in old], run_id, note=f"backfill {asset} {letter} {timeframe} {adjustment} {batch.symbol}") | ||
| 104 | with transaction() as conn: | 104 | with transaction() as conn: | ||
| 105 | upsert_coverage(conn, dataset_id, t, interval) | 105 | upsert_coverage(conn, dataset_id, t, interval) | ||
| 106 | − record_anomalies(conn, dataset_id, rep.anomalies, run_id, limit=500) | 106 | + record_anomalies(conn, dataset_id, rep.anomalies, run_id, limit=40) | ||
| 107 | stats["symbols"] += 1 | 107 | stats["symbols"] += 1 | ||
| 108 | stats["rows"] += t.num_rows | 108 | stats["rows"] += t.num_rows | ||
| 109 | stats["files"] += len(written) | 109 | stats["files"] += len(written) | ||
| 110 | 110 | ||||