8eb0cb1
commit 8eb0cb1
5 files changed
+17 -5
Showing 5 changed files+17-5
▼.github/workflows/lake.yml
+4-3
| @@ -72,13 +72,14 @@ jobs: | ||
72 | 72 | fi |
73 | 73 | |
74 | 74 | - name: Pull existing Silver from Hugging Face |
75 | - # Incremental Silver appends to the previous layer, so it must be present. | |
75 | + # Incremental Silver appends to the previous layer, so it must be | |
76 | + # present: a failed pull followed by a push would truncate the lake. | |
76 | 77 | run: uv run python -m scripts.lake_sync pull-silver |
77 | - continue-on-error: true | |
78 | 78 | |
79 | 79 | - name: Sink Kafka → Bronze Parquet |
80 | + # Exits non-zero when the broker is unreachable or nothing is drained, | |
81 | + # which is a real failure, not a quiet window. | |
80 | 82 | run: uv run python -m services.sink --window-seconds 420 |
81 | - continue-on-error: true | |
82 | 83 | |
83 | 84 | - name: Silver transform (incremental) |
84 | 85 | run: uv run python -m pipelines.silver |
85 | 86 | |
▼scripts/publish_motherduck.py
+3-0
| @@ -12,6 +12,7 @@ job still produces the static bundle. | ||
12 | 12 | |
13 | 13 | from __future__ import annotations |
14 | 14 | |
15 | +import os | |
15 | 16 | import sys |
16 | 17 | from pathlib import Path |
17 | 18 | |
| @@ -39,6 +40,8 @@ def publish(local_path: Path | None = None) -> int: | ||
39 | 40 | """Copy every local schema/table to MotherDuck. Returns tables published.""" |
40 | 41 | s = settings |
41 | 42 | if not s.motherduck_enabled: |
43 | + if os.environ.get("MOTHERDUCK_TOKEN", "unset") == "": | |
44 | + raise RuntimeError("[motherduck] MOTHERDUCK_TOKEN is set but empty — fix the secret") | |
42 | 45 | logger.warning("[motherduck] MOTHERDUCK_TOKEN not set — skipping publish") |
43 | 46 | return 0 |
44 | 47 | |
45 | 48 | |
▼scripts/publish_turso.py
+4-0
| @@ -435,6 +435,10 @@ class TursoPublisher: | ||
435 | 435 | from scripts.publish_site_tables import build_site_payloads, write_site_tables_local |
436 | 436 | |
437 | 437 | if not self.url: |
438 | + import os | |
439 | + | |
440 | + if os.environ.get("TURSO_DATABASE_URL", "unset") == "": | |
441 | + raise RuntimeError("[turso] TURSO_DATABASE_URL is set but empty — fix the secret") | |
438 | 442 | logger.warning("[turso] TURSO_DATABASE_URL not set — skipping") |
439 | 443 | return {} |
440 | 444 | if not self.db_path.exists(): |
441 | 445 | |
▼services/hf_lake.py
+3-1
| @@ -115,6 +115,8 @@ def download_prefix( | ||
115 | 115 | token=cfg.huggingface_token, |
116 | 116 | ) |
117 | 117 | except Exception as exc: # noqa: BLE001 |
118 | + # Never silently continue: the caller merges with local state and | |
119 | + # pushes the result back, so a failed pull can truncate the lake. | |
118 | 120 | logger.error("[hf] download %s failed: %s", repo_prefix, exc) |
119 | - return 0 | |
121 | + raise RuntimeError(f"download {repo_prefix} failed: {exc}") from exc | |
120 | 122 | return len([p for p in root.rglob("*") if p.is_file()]) |
121 | 123 | |
▼services/sources/airlabs.py
+3-1
| @@ -109,7 +109,9 @@ class AirlabsSource(Source): | ||
109 | 109 | def _save_state(self, state: dict[str, Any]) -> None: |
110 | 110 | try: |
111 | 111 | self._state_path.parent.mkdir(parents=True, exist_ok=True) |
112 | - self._state_path.write_text(json.dumps(state), encoding="utf-8") | |
112 | + temp = self._state_path.with_suffix(".tmp") | |
113 | + temp.write_text(json.dumps(state), encoding="utf-8") | |
114 | + temp.replace(self._state_path) # atomic: a crash cannot zero the counter | |
113 | 115 | except OSError as exc: |
114 | 116 | logger.warning("[airlabs] could not persist budget state: %s", exc) |
115 | 117 | |
116 | 118 | |