feat(conf): notebook read-replica aco.ro.duckdb (closes #510)
All checks were successful
CI / lint (push) Successful in 42s
CI / notebooks-smoke (push) Successful in 1m27s
Deploy / notebooks (push) Has been skipped
Deploy / zotero (push) Has been skipped
Deploy / docs (push) Has been skipped
Deploy / api (push) Has been skipped
Deploy / mc (push) Has been skipped
Infra CI / notebooks (push) Successful in 50s
Infra CI / zotero (push) Successful in 13s
Infra CI / docs (push) Successful in 1m12s
Infra CI / api (push) Successful in 1m0s
Infra CI / mc (push) Successful in 17s
Deploy / report (push) Successful in 12s
CI / test (push) Successful in 17m21s

conf.connect.publish_replica(): holds the primary's write lock,
CHECKPOINTs the WAL, copies to <name>.ro.duckdb, swaps atomically —
so the snapshot is always a consistent database and readers holding
the old file keep a valid handle. Both ingest scripts republish as
their final step, bounding staleness to ingest cadence.

conf.connect.duckdb(): read-only opens resolve to the replica when it
exists — notebook kernels never hold the primary's single-writer lock,
so ingests stop failing under open notebooks (the M2/option-C fix from
the concurrency spec). Opt-outs: replica=False param or
STACK_DUCKDB_REPLICA=0. Write opens always use the primary.

Validated live: replica published (3.1 GB), and the notebooks
container resolves conf.connect.duckdb() to /home/kert/data/
aco.ro.duckdb with the rulemaking-comments data visible. 82 conf
tests green (snapshot consistency, divergence, opt-outs, atomic
republish).
This commit is contained in:
kert
2026-07-10 22:23:57 -04:00
parent c9b6a28f07
commit 579d17beb4
5 changed files with 151 additions and 5 deletions

View File

@@ -330,7 +330,7 @@ def main() -> None:
print(f"\nLoading into DuckDB at {DUCKDB_PATH} ...") print(f"\nLoading into DuckDB at {DUCKDB_PATH} ...")
# Lock preflight + retry instead of a raw IOException when a notebook # Lock preflight + retry instead of a raw IOException when a notebook
# kernel holds the single-writer file (#508). # kernel holds the single-writer file (#508).
from conf.connect import duckdb_batch from conf.connect import duckdb_batch, publish_replica
with duckdb_batch("aco") as con: with duckdb_batch("aco") as con:
con.execute("CREATE SCHEMA IF NOT EXISTS skin_subs") con.execute("CREATE SCHEMA IF NOT EXISTS skin_subs")
@@ -354,6 +354,9 @@ def main() -> None:
) )
print(" They may be in separate NOC or tissue coding files") print(" They may be in separate NOC or tissue coding files")
# Refresh the notebook read replica (#510).
print(f"replica → {publish_replica('aco')}")
if __name__ == "__main__": if __name__ == "__main__":
main() main()

View File

@@ -354,7 +354,7 @@ def main() -> None:
# duckdb_batch preflights the single-writer lock (retry + name the # duckdb_batch preflights the single-writer lock (retry + name the
# holding PID) instead of dying on a raw IOException when a notebook # holding PID) instead of dying on a raw IOException when a notebook
# kernel holds a connection (#508). # kernel holds a connection (#508).
from conf.connect import duckdb_batch from conf.connect import duckdb_batch, publish_replica
with duckdb_batch("aco") as con: with duckdb_batch("aco") as con:
ingest_all(con, year_filter=args.year or "") ingest_all(con, year_filter=args.year or "")
@@ -368,6 +368,11 @@ def main() -> None:
cnt = con.execute(f"SELECT count(*) FROM opps.{r[0]}").fetchone()[0] # noqa: S608 cnt = con.execute(f"SELECT count(*) FROM opps.{r[0]}").fetchone()[0] # noqa: S608
print(f" opps.{r[0]:30s}: {cnt:>6} rows") print(f" opps.{r[0]:30s}: {cnt:>6} rows")
# Refresh the notebook read replica so long-running readers see the
# new data without ever locking this primary (#510).
replica = publish_replica("aco")
print(f"replica → {replica}")
print("\nDone.") print("\nDone.")

View File

@@ -45,6 +45,22 @@ with duckdb_batch("aco") as con: # retries, then names the lock holder
tab (or stop the PID) and re-run — the notebook loses nothing; it reconnects tab (or stop the PID) and re-run — the notebook loses nothing; it reconnects
on the next cell run. on the next cell run.
## Notebooks read a replica, not the primary
Read-only opens through `conf.connect.duckdb()` resolve to the snapshot
`data/aco.ro.duckdb` when it exists — so notebook kernels never hold the
primary's lock at all, and ingests stop caring how many notebooks are open.
- **Refresh cadence**: every ingest republishes the snapshot as its last step
(`conf.connect.publish_replica("aco")` — holds the write lock, `CHECKPOINT`s,
copies, then swaps atomically). Staleness is therefore bounded by ingest
frequency; to refresh manually:
`uv run python -c "from conf.connect import publish_replica; publish_replica()"`
- **Reading the live primary instead**: pass `replica=False` to
`conf.connect.duckdb()`, or set `STACK_DUCKDB_REPLICA=0`.
- Kernels holding the *old* snapshot keep a valid file handle after a swap;
re-running the connect cell picks up the fresh one.
## Per-year ingests merge, not wipe ## Per-year ingests merge, not wipe
`ingest_opps.py --year YYYY` used to `DROP TABLE` + `CREATE ... AS` from only `ingest_opps.py --year YYYY` used to `DROP TABLE` + `CREATE ... AS` from only

View File

@@ -29,7 +29,14 @@ from typing import Any
from conf import ROOT, cfg, path from conf import ROOT, cfg, path
def duckdb(name: str = "aco", *, read_only: bool = True) -> Any: def _replica_path(name: str) -> Any:
"""data/<name>.ro.duckdb next to the primary."""
return path(f"db.{name}").with_suffix(".ro.duckdb")
def duckdb(
name: str = "aco", *, read_only: bool = True, replica: bool | None = None
) -> Any:
"""Return a DuckDB connection. """Return a DuckDB connection.
Parameters Parameters
@@ -38,11 +45,25 @@ def duckdb(name: str = "aco", *, read_only: bool = True) -> Any:
Key under ``[db]`` in stack.toml (default ``"aco"``). Key under ``[db]`` in stack.toml (default ``"aco"``).
read_only read_only
Open in read-only mode (default ``True``). Open in read-only mode (default ``True``).
replica
Read-only opens resolve to the ``<name>.ro.duckdb`` snapshot
when it exists — long-running readers (notebook kernels) then
never hold the primary's single-writer lock, so ingests stop
failing (#510). Staleness is bounded by the refresh cadence:
every ingest republishes via :func:`publish_replica`. Pass
``replica=False`` (or set ``STACK_DUCKDB_REPLICA=0``) to read
the live primary.
""" """
import duckdb as _duckdb import duckdb as _duckdb
db_path = str(path(f"db.{name}")) db_path = path(f"db.{name}")
return _duckdb.connect(db_path, read_only=read_only) if read_only:
if replica is None:
replica = os.environ.get("STACK_DUCKDB_REPLICA", "1") != "0"
ro = _replica_path(name)
if replica and ro.exists():
db_path = ro
return _duckdb.connect(str(db_path), read_only=read_only)
def _lock_holders(db_path: str) -> str: def _lock_holders(db_path: str) -> str:
@@ -114,6 +135,29 @@ def duckdb_batch(
con.close() con.close()
def publish_replica(name: str = "aco") -> Any:
"""Publish/refresh the read-only snapshot ``<name>.ro.duckdb``.
Called after each ingest so notebook readers (see :func:`duckdb`)
work from a lock-free snapshot instead of contending with writers
on the primary. Holds the primary's write lock for the duration of
the copy: no writer can be mid-transaction, and ``CHECKPOINT``
flushes the WAL first, so the copied file is a consistent database.
The swap is atomic (``os.replace``); readers already holding the
old snapshot keep a valid file handle until they reconnect.
"""
import shutil
src = path(f"db.{name}")
dst = _replica_path(name)
tmp = dst.with_suffix(".tmp")
with duckdb_batch(name) as con:
con.execute("CHECKPOINT")
shutil.copy2(src, tmp)
os.replace(tmp, dst)
return dst
def trino( def trino(
*, *,
catalog: str = "", catalog: str = "",

View File

@@ -265,3 +265,81 @@ class TestDuckdbBatch:
# lock released — a fresh writer can open immediately # lock released — a fresh writer can open immediately
with connect.duckdb_batch("custom", retries=1) as con: with connect.duckdb_batch("custom", retries=1) as con:
con.execute("SELECT 1") con.execute("SELECT 1")
class TestReplica:
def _patch_path(self, monkeypatch, db_file):
monkeypatch.setattr(
"conf.connect.path",
lambda key: db_file if key == "db.custom" else path(key),
)
def _make_primary(self, tmp_path):
import duckdb as _duckdb
db_file = tmp_path / "custom.duckdb"
con = _duckdb.connect(str(db_file))
con.execute("CREATE TABLE t AS SELECT 1 AS x")
con.close()
return db_file
def test_publish_creates_consistent_snapshot(self, tmp_path, monkeypatch):
db_file = self._make_primary(tmp_path)
self._patch_path(monkeypatch, db_file)
dst = connect.publish_replica("custom")
assert dst.name == "custom.ro.duckdb"
import duckdb as _duckdb
con = _duckdb.connect(str(dst), read_only=True)
assert con.execute("SELECT x FROM t").fetchone()[0] == 1
con.close()
def test_read_only_resolves_to_replica(self, tmp_path, monkeypatch):
db_file = self._make_primary(tmp_path)
self._patch_path(monkeypatch, db_file)
connect.publish_replica("custom")
# diverge the primary so we can tell which file we read
with connect.duckdb_batch("custom") as con:
con.execute("UPDATE t SET x = 2")
assert connect.duckdb("custom").execute("SELECT x FROM t").fetchone()[0] == 1
# explicit opt-out reads the live primary
assert (
connect.duckdb("custom", replica=False)
.execute("SELECT x FROM t")
.fetchone()[0]
== 2
)
# env kill-switch
monkeypatch.setenv("STACK_DUCKDB_REPLICA", "0")
assert connect.duckdb("custom").execute("SELECT x FROM t").fetchone()[0] == 2
def test_no_replica_falls_back_to_primary(self, tmp_path, monkeypatch):
db_file = self._make_primary(tmp_path)
self._patch_path(monkeypatch, db_file)
assert connect.duckdb("custom").execute("SELECT x FROM t").fetchone()[0] == 1
def test_write_opens_never_use_replica(self, tmp_path, monkeypatch):
db_file = self._make_primary(tmp_path)
self._patch_path(monkeypatch, db_file)
connect.publish_replica("custom")
con = connect.duckdb("custom", read_only=False)
con.execute("UPDATE t SET x = 3")
con.close()
assert (
connect.duckdb("custom", replica=False)
.execute("SELECT x FROM t")
.fetchone()[0]
== 3
)
def test_republish_refreshes_atomically(self, tmp_path, monkeypatch):
db_file = self._make_primary(tmp_path)
self._patch_path(monkeypatch, db_file)
connect.publish_replica("custom")
with connect.duckdb_batch("custom") as con:
con.execute("UPDATE t SET x = 9")
connect.publish_replica("custom")
assert connect.duckdb("custom").execute("SELECT x FROM t").fetchone()[0] == 9