diff --git a/docker-compose.yml b/docker-compose.yml index de02f84..2ba394c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -30,6 +30,7 @@ services: build: . image: ${IMAGE:-quickbeam:local} restart: unless-stopped + mem_limit: 2500m depends_on: [qdrant] environment: # Read-only work, but the fangorn CLI refuses to start without a key. @@ -83,6 +84,7 @@ services: build: . image: ${IMAGE:-quickbeam:local} restart: unless-stopped + mem_limit: 1600m depends_on: [qdrant] ports: - "${SEARCH_PORT:-8080}:8080" diff --git a/quickbeam/watcher.py b/quickbeam/watcher.py index 1c5608c..c549a41 100644 --- a/quickbeam/watcher.py +++ b/quickbeam/watcher.py @@ -364,7 +364,7 @@ async def _seed_pair(args, qdrant, embed_engine, role_map_ref, dim, truncate, On failure we leave the prior snapshot ALONE and ingest nothing: projecting an empty namespace would diff every already-embedded vertex as removed and tombstone the whole source (the exact outcome the null-head guard in _seed_read_async exists to prevent).""" - key = f"{owner}:{namespace}" + key = f"{_app_slug(args.app)}:{owner}:{namespace}" try: contents = await _seed_read_async( args.fangorn_bin, owner, namespace, args.seed_timeout, args.app) @@ -411,7 +411,7 @@ async def _stream_source_once(args, qdrant, embed_engine, role_map_ref, dim, tru once per pair, then reconnects reuse the snapshot and rely on the subscribe cursor replaying any commits missed while down. Re-reading the whole namespace on every reconnect is what made a slow read freeze the watcher in a loop.""" - key = f"{owner or '*'}:{namespace or '*'}" + key = f"{_app_slug(args.app)}:{owner or '*'}:{namespace or '*'}" app_mode = owner is None or namespace is None # Start the subscription FIRST so any commit that lands while we seed buffers in @@ -461,6 +461,11 @@ async def _stream_source_once(args, qdrant, embed_engine, role_map_ref, dim, tru ch_owner = change.get("owner") or owner ch_ns = change.get("namespace") or namespace ch_key = f"{ch_owner}:{ch_ns}" + # ch_key is the PERSISTED checkpoint key (the setdefault below). Its format + # must not change: rewriting it orphans every /data/db/checkpoint.json entry, + # so every source reads as unseeded and re-embeds the whole corpus from chain. + # ch_show is the log label only, and is where the app belongs. + ch_show = f"{_app_slug(args.app)}:{ch_key}" # First commit seen for a namespace we've never read: seed it now, so the # projection sees the whole graph and not just what streamed past since. @@ -494,7 +499,7 @@ async def _stream_source_once(args, qdrant, embed_engine, role_map_ref, dim, tru src_ck["head"] = change.get("newRoot") src_ck["block"] = change.get("blockNumber") - print(f"[Watcher] {ch_key}: change @ block {change.get('blockNumber')} " + print(f"[Watcher] {ch_show}: change @ block {change.get('blockNumber')} " f"(+{len(change.get('addedVertices', []))} / " f"-{len(change.get('removedVertexCids', []))} vertices) " f"→ {change.get('commitCid')}") @@ -507,7 +512,7 @@ async def _stream_source_once(args, qdrant, embed_engine, role_map_ref, dim, tru edges_sink=change_edges, tombstones_sink=change_tombstones, ) status = f"{n} new record(s) embedded" if n else "no new records for the active profiles" - print(f"[Watcher] {ch_key}: change applied — {status}") + print(f"[Watcher] {ch_show}: change applied — {status}") _deliver_cdn(_pair_cdn_args(args, qdrant, ch_owner, ch_ns), qdrant, n, change_edges, change_tombstones, owner=ch_owner, namespace=ch_ns, app=args.app) @@ -529,7 +534,7 @@ async def _stream_source(args, qdrant, embed_engine, role_map_ref, dim, truncate """Supervise one source forever: (re)subscribe, and if the stream drops, back off --poll-interval seconds and reconnect. `fangorn subscribe` persists its own resume cursor, so a reconnect replays commits missed while we were down.""" - key = f"{owner or '*'}:{namespace or '*'}" + key = f"{_app_slug(args.app)}:{owner or '*'}:{namespace or '*'}" # Persist the in-memory namespace snapshots across reconnects so the expensive full # seed read runs only until it succeeds once per pair; later reconnects reuse them and # lean on the subscribe cursor to replay anything missed while down. diff --git a/tests/test_ingest_checkpoint.py b/tests/test_ingest_checkpoint.py index 25a1b2f..3e337be 100644 --- a/tests/test_ingest_checkpoint.py +++ b/tests/test_ingest_checkpoint.py @@ -31,6 +31,9 @@ setattr(sys.modules[mod], attr, sys.modules["qdrant_client.models"] if attr == "models" else object) +import inspect # noqa: E402 + +from quickbeam import watcher # noqa: E402 from quickbeam.ingest.checkpoint import _load_checkpoint # noqa: E402 from quickbeam.watcher import _ingest_contents # noqa: E402 @@ -152,9 +155,28 @@ async def ok(*a, **k): assert sorted(_stored_cids(args)) == first + +def test_change_checkpoint_key_stays_app_free(): + """The per-change checkpoint key must NOT gain the app slug. + + `ch_key` is persisted into checkpoint.json's "sources" map. Reformatting it orphans + every existing entry, so each source reads as unseeded and re-embeds its whole corpus + from chain — hours of CPU that presents as data loss, not as a logging change. The app + belongs on `ch_show`, the log label, which may be reformatted freely. + + KEY above pins the other (seed-path) checkpoint key behaviourally; this pins the + per-change one, which no test otherwise covers. + """ + src = inspect.getsource(watcher._stream_source_once) + assert 'ch_key = f"{ch_owner}:{ch_ns}"' in src, "persisted per-change key format changed" + assert "setdefault(ch_key, {})" in src, "checkpoint must key off ch_key, not the log label" + assert 'ch_show = f"{_app_slug(args.app)}:{ch_key}"' in src, "app belongs on the log label" + + if __name__ == "__main__": for fn in (test_failed_upload_does_not_advance_the_checkpoint, test_records_are_reoffered_and_only_then_recorded, - test_no_new_records_still_persists): + test_no_new_records_still_persists, + test_change_checkpoint_key_stays_app_free): fn() print("ok", fn.__name__)