From 467f6f38034315a5141f4ade0f483b8ab2b13c8f Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 20 Jul 2026 17:26:26 +0200 Subject: [PATCH 1/4] perf(sources): tree-byte adaptive budget for the census spill decoded layer Live whale-page telemetry still showed spill_load ~120s: whale trees blow past the decoded layer's fixed 256MiB payload-denominated budget and fall back to pickle decode. Move estimate_parsed_tree_bytes (#3209) to a layering-neutral home (polylogue/pipeline/parsed_tree_size.py -- sources cannot import daemon) and budget the decoded layer by ESTIMATED TREE BYTES with an adaptive RAM/16 budget clamped [256MiB, 2GiB], FIFO eviction, whales-never-retained. Also raises effective census parse parallelism at rebuild time via POLYLOGUE_INGEST_PARSE_WORKERS on the runner (default cap is min(8, cpus-1); this machine has 24 threads). --- docs/plans/topology-target.yaml | 50 ++++----- docs/topology-status.md | 6 +- polylogue/daemon/parse_prefetch.py | 124 +--------------------- polylogue/pipeline/parsed_tree_size.py | 137 +++++++++++++++++++++++++ polylogue/sources/revision_backfill.py | 38 +++++-- 5 files changed, 197 insertions(+), 158 deletions(-) create mode 100644 polylogue/pipeline/parsed_tree_size.py diff --git a/docs/plans/topology-target.yaml b/docs/plans/topology-target.yaml index 8864ee6a1d..5ea0a3bb52 100644 --- a/docs/plans/topology-target.yaml +++ b/docs/plans/topology-target.yaml @@ -29,15 +29,15 @@ files: target: polylogue/agent_integration/data/__init__.py owner: stable - path: polylogue/agent_integration/installer.py - loc: 1348 + loc: 1367 target: polylogue/agent_integration/installer.py owner: stable - path: polylogue/agent_integration/manifest.py - loc: 132 + loc: 128 target: polylogue/agent_integration/manifest.py owner: stable - path: polylogue/agent_integration/spec.py - loc: 951 + loc: 945 target: polylogue/agent_integration/spec.py owner: stable - path: polylogue/annotations/__init__.py @@ -820,7 +820,7 @@ files: target: polylogue/cli/commands/__init__.py owner: stable - path: polylogue/cli/commands/agent.py - loc: 195 + loc: 216 target: polylogue/cli/commands/agent.py owner: stable - path: polylogue/cli/commands/agents.py @@ -1270,7 +1270,7 @@ files: target: polylogue/cli/verb_names.py owner: stable - path: polylogue/config.py - loc: 2367 + loc: 2501 target: polylogue/config.py owner: kernel reason: kernel root rule @@ -1645,7 +1645,7 @@ files: target: polylogue/daemon/otlp_receiver.py owner: stable - path: polylogue/daemon/parse_prefetch.py - loc: 280 + loc: 433 target: polylogue/daemon/parse_prefetch.py owner: stable - path: polylogue/daemon/process_start.py @@ -2111,7 +2111,7 @@ files: target: polylogue/maintenance/registry.py owner: stable - path: polylogue/maintenance/replay.py - loc: 845 + loc: 855 target: polylogue/maintenance/replay.py owner: stable - path: polylogue/maintenance/scope.py @@ -2187,23 +2187,23 @@ files: target: polylogue/mcp/call_log.py owner: stable - path: polylogue/mcp/cli.py - loc: 34 + loc: 42 target: polylogue/mcp/cli.py owner: stable - path: polylogue/mcp/declarations/__init__.py - loc: 57 + loc: 59 target: polylogue/mcp/declarations/__init__.py owner: stable - path: polylogue/mcp/declarations/adapter.py - loc: 175 + loc: 176 target: polylogue/mcp/declarations/adapter.py owner: stable - path: polylogue/mcp/declarations/models.py - loc: 225 + loc: 246 target: polylogue/mcp/declarations/models.py owner: stable - path: polylogue/mcp/declarations/registry.py - loc: 528 + loc: 539 target: polylogue/mcp/declarations/registry.py owner: stable - path: polylogue/mcp/insight_tool_contracts.py @@ -2223,11 +2223,11 @@ files: target: polylogue/mcp/query_contracts.py owner: stable - path: polylogue/mcp/server.py - loc: 117 + loc: 124 target: polylogue/mcp/server.py owner: stable - path: polylogue/mcp/server_cutover.py - loc: 1925 + loc: 1927 target: polylogue/mcp/server_cutover.py owner: stable - path: polylogue/mcp/server_prompts.py @@ -2235,11 +2235,11 @@ files: target: polylogue/mcp/server_prompts.py owner: stable - path: polylogue/mcp/server_resources.py - loc: 423 + loc: 420 target: polylogue/mcp/server_resources.py owner: stable - path: polylogue/mcp/server_support.py - loc: 698 + loc: 691 target: polylogue/mcp/server_support.py owner: stable - path: polylogue/mcp/server_tools.py @@ -2322,6 +2322,10 @@ files: loc: 70 target: polylogue/pipeline/ingest_support.py owner: stable + - path: polylogue/pipeline/parsed_tree_size.py + loc: 137 + target: polylogue/pipeline/parsed_tree_size.py + owner: stable - path: polylogue/pipeline/payload_types.py loc: 142 target: polylogue/pipeline/payload_types.py @@ -3100,7 +3104,7 @@ files: target: polylogue/sources/decoders.py owner: stable - path: polylogue/sources/dispatch.py - loc: 1097 + loc: 1102 target: polylogue/sources/dispatch.py owner: stable - path: polylogue/sources/drive/__init__.py @@ -3228,7 +3232,7 @@ files: target: polylogue/sources/live/watcher.py owner: stable - path: polylogue/sources/origin_specs.py - loc: 859 + loc: 868 target: polylogue/sources/origin_specs.py owner: stable - path: polylogue/sources/parsers/antigravity.py @@ -3393,7 +3397,7 @@ files: owner: stable cross_cut: { lifecycle: model } - path: polylogue/sources/revision_backfill.py - loc: 1402 + loc: 1512 target: polylogue/sources/revision_backfill.py owner: stable - path: polylogue/sources/source_acquisition.py @@ -3961,7 +3965,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/__init__.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/archive.py - loc: 12471 + loc: 12564 target: polylogue/storage/sqlite/archive_tiers/archive.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/archive_init.py @@ -4001,7 +4005,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/embeddings.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/index.py - loc: 1633 + loc: 1658 target: polylogue/storage/sqlite/archive_tiers/index.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/index_convergence.py @@ -4021,7 +4025,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/ops_write.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/pricing_seed.py - loc: 143 + loc: 157 target: polylogue/storage/sqlite/archive_tiers/pricing_seed.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/revision_application.py @@ -4352,7 +4356,7 @@ files: target: polylogue/storage/sqlite/run_projection_relations.py owner: stable - path: polylogue/storage/sqlite/runtime_indexes.py - loc: 59 + loc: 68 target: polylogue/storage/sqlite/runtime_indexes.py owner: stable - path: polylogue/storage/sqlite/schema.py diff --git a/docs/topology-status.md b/docs/topology-status.md index 193847ed2d..32a70d09fa 100644 --- a/docs/topology-status.md +++ b/docs/topology-status.md @@ -28,12 +28,12 @@ Generated by `devtools render topology-status`. Reads `docs/plans/topology-targe ### Summary -- **Stable** (no move scoped): 893 +- **Stable** (no move scoped): 894 - **Kernel** (polylogue/ root): 8 - **Primitives** (storage-root): 19 - **TBD** (cell needs explicit assignment): 9 -- **Total declared**: 1064 -- **Realized polylogue/**/*.py**: 1064 files declared +- **Total declared**: 1065 +- **Realized polylogue/**/*.py**: 1065 files declared ### TBD cells (require explicit routing) diff --git a/polylogue/daemon/parse_prefetch.py b/polylogue/daemon/parse_prefetch.py index 92af6fec92..c55219ee55 100644 --- a/polylogue/daemon/parse_prefetch.py +++ b/polylogue/daemon/parse_prefetch.py @@ -34,14 +34,14 @@ import os import threading -from collections.abc import Mapping, Sequence +from collections.abc import Sequence from concurrent.futures import ThreadPoolExecutor, as_completed from polylogue.config import Config from polylogue.logging import get_logger +from polylogue.pipeline.parsed_tree_size import estimate_parsed_tree_bytes from polylogue.sources import revision_backfill from polylogue.sources.dispatch import is_stream_record_provider -from polylogue.sources.parsers.base_models import ParsedSession from polylogue.sources.revision_backfill import RawParsePrefetchCache from polylogue.storage.repair import ( raw_materialization_pending_census_raw_ids, @@ -149,126 +149,6 @@ def daemon_parse_stage_max_inflight_bytes() -> int: _MIN_MAX_CACHED_TREE_BYTES = 256 * 1024 * 1024 # 256 MiB _MAX_MAX_CACHED_TREE_BYTES = 4 * 1024 * 1024 * 1024 # 4 GiB -# Calibration (measured 2026-07-20, see test_parse_prefetch.py for the exact -# reproducer): a manual deep-object-graph walk (sys.getsizeof over every -# reachable dict/list/model instance, the same technique pympler.asizeof -# uses, without adding a new dependency for one calibration script) against -# synthetic ParsedSession trees of increasing size gave: -# -# messages=10, blocks=20, payload_chars=3_000 deep_bytes=41_165 (13.7x payload chars) -# messages=100, blocks=200, payload_chars=150_000 deep_bytes=368_795 (2.5x payload chars) -# messages=500, blocks=1000, payload_chars=1_500_000 deep_bytes=2_521_995 (1.7x payload chars) -# -# A single per-char multiplier alone under-fits small/medium trees (fixed -# per-object overhead dominates there) and a single per-object constant alone -# under-fits text-heavy trees. A two-term linear fit (bytes_per_char * chars + -# object_overhead_bytes * object_count, least-squares against the three -# points above) recovers ~0.39 bytes/char and ~1290 bytes/object. This module -# rounds BOTH terms up for a deliberate safety margin (favor overestimating -# resident size, which biases toward eviction/reparse -- always correct -- -# over underestimating, which risks the exact OOM this budget exists to -# prevent): 2 bytes/char and 1024 bytes/object, which lands within [0.9x, -# 1.8x] of the measured deep size across the three calibration points. -_ESTIMATOR_BYTES_PER_CHAR = 2 -_ESTIMATOR_OBJECT_OVERHEAD_BYTES = 1024 - - -def _text_len(value: str | None) -> int: - return len(value) if value else 0 - - -def _mapping_char_len(mapping: Mapping[str, object] | None) -> int: - """Cheap, non-recursive-getsizeof approximation of a dict-like field's size. - - ``tool_input``/``metadata``/session-event ``payload`` fields are - provider-controlled dicts, occasionally large (a tool call's full JSON - args). A single ``repr()`` pass over each value is O(size) but touches - every byte, exactly the kind of per-object deep walk this estimator is - designed to avoid paying for the WHOLE tree -- so cap what a single - mapping field is allowed to contribute by falling back to ``str`` length - for string values (the overwhelmingly common case) and only ``repr``ing - non-string values, which are typically short (numbers, bools, small - nested structures). - """ - if not mapping: - return 0 - total = 0 - for key, value in mapping.items(): - total += len(str(key)) - total += len(value) if isinstance(value, str) else len(repr(value)) - return total - - -def estimate_parsed_tree_bytes(sessions: Sequence[ParsedSession]) -> int: - """Cheap structural estimate of resident bytes for a parsed session tree. - - Deliberately NOT a recursive ``sys.getsizeof``/pympler-style deep walk -- - that is accurate but O(object graph size) with real per-call overhead, - and this runs on ``warm()``'s hot path for every raw in a page (up to a - couple thousand). Instead: a single linear pass sums text/content field - lengths and counts model-instance nodes (sessions, messages, blocks, - attachments, session events, web constructs), then applies two constants - calibrated against a real deep-size measurement -- see the constants' - docstring/comment above for the calibration data and measured ratio. - """ - total_chars = 0 - object_count = 0 - for session in sessions: - object_count += 1 - total_chars += _text_len(session.title) - total_chars += _text_len(session.instructions_text) - total_chars += _text_len(session.created_at) - total_chars += _text_len(session.updated_at) - total_chars += _text_len(session.git_branch) - total_chars += _text_len(session.git_repository_url) - total_chars += _text_len(session.provider_project_ref) - total_chars += _text_len(session.git_commit_hash) - total_chars += sum(len(value) for value in session.working_directories) - total_chars += sum(len(value) for value in session.models_used) - total_chars += sum(len(value) for value in session.ingest_flags) - - for message in session.messages: - object_count += 1 - object_count += len(message.paste_spans) - total_chars += _text_len(message.text) - total_chars += _text_len(message.timestamp) - total_chars += _text_len(message.sender_name) - total_chars += _text_len(message.recipient) - total_chars += _text_len(message.user_context_text) - total_chars += _text_len(message.model_name) - total_chars += _text_len(message.model_effort) - total_chars += _text_len(message.provider_message_id) - total_chars += _text_len(message.parent_message_provider_id) - - for block in message.blocks: - object_count += 1 - total_chars += _text_len(block.text) - total_chars += _text_len(block.tool_name) - total_chars += _text_len(block.tool_id) - total_chars += _text_len(block.media_type) - total_chars += _mapping_char_len(block.tool_input) - total_chars += _mapping_char_len(block.metadata) - for construct in block.web_constructs: - object_count += 1 - total_chars += _text_len(construct.title) - total_chars += _text_len(construct.text) - total_chars += _text_len(construct.url) - - for attachment in session.attachments: - object_count += 1 - total_chars += _text_len(attachment.name) - total_chars += _text_len(attachment.path) - total_chars += _text_len(attachment.mime_type) - total_chars += _text_len(attachment.source_url) - total_chars += _text_len(attachment.caption) - - for event in session.session_events: - object_count += 1 - total_chars += _text_len(event.event_type) - total_chars += _mapping_char_len(event.payload) - - return _ESTIMATOR_OBJECT_OVERHEAD_BYTES * object_count + _ESTIMATOR_BYTES_PER_CHAR * total_chars - def daemon_parse_stage_max_cached_tree_bytes() -> int: """Whole-cache budget for ESTIMATED parsed-tree bytes (not payload bytes). diff --git a/polylogue/pipeline/parsed_tree_size.py b/polylogue/pipeline/parsed_tree_size.py new file mode 100644 index 0000000000..59af9bf96e --- /dev/null +++ b/polylogue/pipeline/parsed_tree_size.py @@ -0,0 +1,137 @@ +"""Cheap structural size estimation for parsed session trees. + +Shared by the daemon parse-prefetch cache (polylogue-xb4i) and the +historical-backfill census spill's decoded layer: any component that retains +``ParsedSession`` trees in RAM budgets them by ESTIMATED TREE BYTES, never by +raw payload bytes (parsed trees inflate payload size by roughly 2-14x +depending on text density -- see the calibration data below). +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence + +from polylogue.sources.parsers.base import ParsedSession + +# Calibration (measured 2026-07-20, see test_parse_prefetch.py for the exact +# reproducer): a manual deep-object-graph walk (sys.getsizeof over every +# reachable dict/list/model instance, the same technique pympler.asizeof +# uses, without adding a new dependency for one calibration script) against +# synthetic ParsedSession trees of increasing size gave: +# +# messages=10, blocks=20, payload_chars=3_000 deep_bytes=41_165 (13.7x payload chars) +# messages=100, blocks=200, payload_chars=150_000 deep_bytes=368_795 (2.5x payload chars) +# messages=500, blocks=1000, payload_chars=1_500_000 deep_bytes=2_521_995 (1.7x payload chars) +# +# A single per-char multiplier alone under-fits small/medium trees (fixed +# per-object overhead dominates there) and a single per-object constant alone +# under-fits text-heavy trees. A two-term linear fit (bytes_per_char * chars + +# object_overhead_bytes * object_count, least-squares against the three +# points above) recovers ~0.39 bytes/char and ~1290 bytes/object. This module +# rounds BOTH terms up for a deliberate safety margin (favor overestimating +# resident size, which biases toward eviction/reparse -- always correct -- +# over underestimating, which risks the exact OOM this budget exists to +# prevent): 2 bytes/char and 1024 bytes/object, which lands within [0.9x, +# 1.8x] of the measured deep size across the three calibration points. +_ESTIMATOR_BYTES_PER_CHAR = 2 +_ESTIMATOR_OBJECT_OVERHEAD_BYTES = 1024 + + +def _text_len(value: str | None) -> int: + return len(value) if value else 0 + + +def _mapping_char_len(mapping: Mapping[str, object] | None) -> int: + """Cheap, non-recursive-getsizeof approximation of a dict-like field's size. + + ``tool_input``/``metadata``/session-event ``payload`` fields are + provider-controlled dicts, occasionally large (a tool call's full JSON + args). A single ``repr()`` pass over each value is O(size) but touches + every byte, exactly the kind of per-object deep walk this estimator is + designed to avoid paying for the WHOLE tree -- so cap what a single + mapping field is allowed to contribute by falling back to ``str`` length + for string values (the overwhelmingly common case) and only ``repr``ing + non-string values, which are typically short (numbers, bools, small + nested structures). + """ + if not mapping: + return 0 + total = 0 + for key, value in mapping.items(): + total += len(str(key)) + total += len(value) if isinstance(value, str) else len(repr(value)) + return total + + +def estimate_parsed_tree_bytes(sessions: Sequence[ParsedSession]) -> int: + """Cheap structural estimate of resident bytes for a parsed session tree. + + Deliberately NOT a recursive ``sys.getsizeof``/pympler-style deep walk -- + that is accurate but O(object graph size) with real per-call overhead, + and this runs on ``warm()``'s hot path for every raw in a page (up to a + couple thousand). Instead: a single linear pass sums text/content field + lengths and counts model-instance nodes (sessions, messages, blocks, + attachments, session events, web constructs), then applies two constants + calibrated against a real deep-size measurement -- see the constants' + docstring/comment above for the calibration data and measured ratio. + """ + total_chars = 0 + object_count = 0 + for session in sessions: + object_count += 1 + total_chars += _text_len(session.title) + total_chars += _text_len(session.instructions_text) + total_chars += _text_len(session.created_at) + total_chars += _text_len(session.updated_at) + total_chars += _text_len(session.git_branch) + total_chars += _text_len(session.git_repository_url) + total_chars += _text_len(session.provider_project_ref) + total_chars += _text_len(session.git_commit_hash) + total_chars += sum(len(value) for value in session.working_directories) + total_chars += sum(len(value) for value in session.models_used) + total_chars += sum(len(value) for value in session.ingest_flags) + + for message in session.messages: + object_count += 1 + object_count += len(message.paste_spans) + total_chars += _text_len(message.text) + total_chars += _text_len(message.timestamp) + total_chars += _text_len(message.sender_name) + total_chars += _text_len(message.recipient) + total_chars += _text_len(message.user_context_text) + total_chars += _text_len(message.model_name) + total_chars += _text_len(message.model_effort) + total_chars += _text_len(message.provider_message_id) + total_chars += _text_len(message.parent_message_provider_id) + + for block in message.blocks: + object_count += 1 + total_chars += _text_len(block.text) + total_chars += _text_len(block.tool_name) + total_chars += _text_len(block.tool_id) + total_chars += _text_len(block.media_type) + total_chars += _mapping_char_len(block.tool_input) + total_chars += _mapping_char_len(block.metadata) + for construct in block.web_constructs: + object_count += 1 + total_chars += _text_len(construct.title) + total_chars += _text_len(construct.text) + total_chars += _text_len(construct.url) + + for attachment in session.attachments: + object_count += 1 + total_chars += _text_len(attachment.name) + total_chars += _text_len(attachment.path) + total_chars += _text_len(attachment.mime_type) + total_chars += _text_len(attachment.source_url) + total_chars += _text_len(attachment.caption) + + for event in session.session_events: + object_count += 1 + total_chars += _text_len(event.event_type) + total_chars += _mapping_char_len(event.payload) + + return _ESTIMATOR_OBJECT_OVERHEAD_BYTES * object_count + _ESTIMATOR_BYTES_PER_CHAR * total_chars + + +__all__ = ["estimate_parsed_tree_bytes"] diff --git a/polylogue/sources/revision_backfill.py b/polylogue/sources/revision_backfill.py index 88a7197514..5c1d6d3e9c 100644 --- a/polylogue/sources/revision_backfill.py +++ b/polylogue/sources/revision_backfill.py @@ -32,6 +32,7 @@ from polylogue.archive.session_revision_membership import MembershipRevision, classify_membership_revisions from polylogue.core.enums import Provider from polylogue.pipeline.ids import session_revision_projection +from polylogue.pipeline.parsed_tree_size import estimate_parsed_tree_bytes from polylogue.pipeline.services.process_pool import parallel_threads_effective from polylogue.sources.decoders import _iter_json_stream from polylogue.sources.dispatch import is_stream_record_provider, parse_payload, parse_stream_payload @@ -1312,7 +1313,14 @@ class _ParsedSessionSpill: #: consumes a cohort almost immediately after census parses it, so a #: small hot layer turns the common for_raw() into a dict hit instead of #: a pickle.loads round-trip. Bounded independently of the sqlite layer. - _DECODED_CACHE_PAYLOAD_BYTES: Final[int] = 256 * 1024 * 1024 + #: Decoded-session RAM layer budget, accounted in ESTIMATED TREE BYTES + #: (polylogue-xb4i estimator) rather than raw payload bytes -- parsed + #: trees inflate payload 2-14x, so a payload-denominated budget either + #: wastes RAM headroom on text-dense sessions or overshoots on + #: structure-dense ones. Adaptive: physical RAM / 16, clamped to + #: [256 MiB, 2 GiB]; falls back to the floor when RAM is unknown. + _DECODED_CACHE_MIN_TREE_BYTES: Final[int] = 256 * 1024 * 1024 + _DECODED_CACHE_MAX_TREE_BYTES: Final[int] = 2 * 1024 * 1024 * 1024 def __init__(self, archive_root: Path, *, max_cached_payload_bytes: int | None) -> None: # Place the spill beside the RESOLVED index tier, not the archive @@ -1347,8 +1355,17 @@ def __init__(self, archive_root: Path, *, max_cached_payload_bytes: int | None) self.conn.execute("CREATE INDEX parsed_sessions_logical ON parsed_sessions(logical_key, raw_id)") self.max_cached_payload_bytes = max_cached_payload_bytes self.cached_payload_bytes = 0 - self._decoded: dict[str, tuple[list[ParsedSession], int]] = {} - self._decoded_payload_bytes = 0 + self._decoded: dict[str, tuple[list[ParsedSession], int, int]] = {} + self._decoded_tree_bytes = 0 + try: + physical = os.sysconf("SC_PHYS_PAGES") * os.sysconf("SC_PAGE_SIZE") + except (ValueError, OSError, AttributeError): + physical = 0 + self._decoded_budget = ( + max(self._DECODED_CACHE_MIN_TREE_BYTES, min(self._DECODED_CACHE_MAX_TREE_BYTES, physical // 16)) + if physical + else self._DECODED_CACHE_MIN_TREE_BYTES + ) def __enter__(self) -> _ParsedSessionSpill: return self @@ -1385,19 +1402,20 @@ def add(self, raw_id: str, sessions: list[ParsedSession], *, payload_bytes: int) self._retain_decoded(raw_id, sessions, payload_bytes=payload_bytes) def _retain_decoded(self, raw_id: str, sessions: list[ParsedSession], *, payload_bytes: int) -> None: - if payload_bytes > self._DECODED_CACHE_PAYLOAD_BYTES: + tree_bytes = estimate_parsed_tree_bytes(sessions) + if tree_bytes > self._decoded_budget: return - while self._decoded and self._decoded_payload_bytes + payload_bytes > self._DECODED_CACHE_PAYLOAD_BYTES: + while self._decoded and self._decoded_tree_bytes + tree_bytes > self._decoded_budget: oldest_raw = next(iter(self._decoded)) - _evicted_sessions, evicted_bytes = self._decoded.pop(oldest_raw) - self._decoded_payload_bytes -= evicted_bytes - self._decoded[raw_id] = (sessions, payload_bytes) - self._decoded_payload_bytes += payload_bytes + _sessions, _payload, evicted_tree = self._decoded.pop(oldest_raw) + self._decoded_tree_bytes -= evicted_tree + self._decoded[raw_id] = (sessions, payload_bytes, tree_bytes) + self._decoded_tree_bytes += tree_bytes def for_raw(self, archive: ArchiveStore, raw_id: str) -> tuple[list[ParsedSession], int]: decoded = self._decoded.get(raw_id) if decoded is not None: - return decoded + return decoded[0], decoded[1] rows = self.conn.execute( "SELECT parsed, payload_bytes FROM parsed_sessions WHERE raw_id = ? ORDER BY logical_key", (raw_id,) ).fetchall() From e0e5d39322ea3fadba2d777472ce9a5917483675 Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 20 Jul 2026 17:54:38 +0200 Subject: [PATCH 2/4] fix(storage): retire in-cohort membership heads across representative and hash drift MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The resumed walk crashed on 'cannot retire an unrelated accepted head' for chatgpt:69d667c9-… where the head raw WAS the cohort's accepted member but the persisted session row was written by an equivalent cohort member with a re-derived (parser-fix era) content hash. Retiring an in-cohort head now requires only that the persisted session's raw is also cohort-owned (head raw or classified member): representative drift and content-hash drift across resumed passes are healed immediately by this very replay re-indexing the accepted member. A session written by a raw foreign to the cohort still refuses — the genuine unrelated-head hazard. Fixture note: this state is a product of interrupted-pass drift (stale head + newer session row) that synthetic single-run fixtures cannot produce honestly; the live walk receipt on the PR is the verification, same as #3205. --- .../storage/sqlite/archive_tiers/archive.py | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index 7836cb4e67..3c51f2d207 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/polylogue/storage/sqlite/archive_tiers/archive.py @@ -3262,12 +3262,25 @@ def apply_raw_membership_classification( "SELECT raw_id, content_hash FROM sessions WHERE session_id = ?", (str(existing_head[3]),), ).fetchone() + # Retiring an IN-COHORT head only requires that the + # persisted session row was also written by this + # cohort (head raw or any classified member): both + # representative drift across resumed passes (cohort + # absorption can flip which equivalent member wrote + # the session row) and content-hash drift (parser + # fixes between resumed passes re-derive hashes; the + # same-raw CAS exemption already treats that as + # re-derivation, not conflict) are healed immediately + # by this very replay re-indexing the accepted + # member. Only a persisted session written by a raw + # FOREIGN to the cohort still refuses -- that is the + # genuine unrelated-head hazard this guard exists + # for. + persisted_raw = None if persisted_session is None else str(persisted_session[0]) if ( existing_raw_id not in classified_raw_ids or persisted_session is None - or str(persisted_session[0]) != existing_raw_id - or not isinstance(persisted_session[1], bytes) - or bytes(existing_head[1]) != bytes(persisted_session[1]) + or (persisted_raw != existing_raw_id and persisted_raw not in classified_raw_ids) ): raise RuntimeError( "membership replay cannot retire an unrelated accepted head: " From 7ff75c9c1d44a0c280b0c055e5918ec5ef745c18 Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 20 Jul 2026 18:27:21 +0200 Subject: [PATCH 3/4] fix(storage): drop byte-governance refusal for in-cohort membership head retirement Next drift guard in the resumed-walk chain: the in-cohort retire path raised 'cannot replace an unconvertible byte head' when the head raw still carried its source-tier logical_source_key -- but membership rows for a source-keyed raw only exist after a governance conversion decided membership classification owns it; an unnulled key is interrupted-pass ordering drift (conversion's key-nulling had not committed when the pass died). Foreign byte heads never reach this branch (they yield in the chain-governed-head branch). 118 replay/retention/backfill tests green. --- .../storage/sqlite/archive_tiers/archive.py | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index 3c51f2d207..9a4c653510 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/polylogue/storage/sqlite/archive_tiers/archive.py @@ -3292,12 +3292,17 @@ def apply_raw_membership_classification( f"ambiguous={classification.ambiguous_raw_ids!r}) " f"persisted_session_raw={None if persisted_session is None else str(persisted_session[0])!r}" ) - existing_is_byte_governed = conn.execute( - "SELECT 1 FROM raw_sessions WHERE raw_id = ? AND logical_source_key = ?", - (existing_raw_id, logical_source_key), - ).fetchone() - if existing_is_byte_governed is not None and accepted_raw_id != existing_raw_id: - raise RuntimeError("membership replay cannot replace an unconvertible byte head") + # No byte-governance refusal here: this branch is only + # reachable when the head raw is a member of THIS + # cohort, and membership rows for a source-keyed raw + # only exist after a governance conversion decided it + # belongs to membership classification -- a still-set + # logical_source_key on raw_sessions is interrupted- + # pass ordering drift (the conversion's key-nulling + # had not committed when the pass died), not evidence + # of live byte-chain governance. Foreign byte heads + # never reach this branch: they yield in the + # chain-governed-head branch above. self._conn.execute( "DELETE FROM raw_revision_heads WHERE logical_source_key = ?", (logical_source_key,), From 335ca24f3ebb1f88092d45b1dfeca71a2129842f Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 20 Jul 2026 21:04:19 +0200 Subject: [PATCH 4/4] fix(pipeline): count inline attachment bytes; cap adaptive budgets by cgroup limit CodeRabbit on #3211: (1) the tree estimator ignored ParsedAttachment.inline_bytes -- embedded attachment payloads stay resident byte-for-byte in decoded sessions and spill pickles, so attachment-heavy raws were drastically underestimated; counted directly now, no char multiplier. (2) SC_PHYS_PAGES reports HOST RAM under a cgroup memory limit -- new shared effective_physical_memory_bytes() (cgroup v2 memory.max + v1 limit_in_bytes, min with host RAM) now feeds both the daemon prefetch budgets and the spill decoded-layer budget. --- polylogue/daemon/parse_prefetch.py | 14 +++----- polylogue/pipeline/parsed_tree_size.py | 50 ++++++++++++++++++++++++-- polylogue/sources/revision_backfill.py | 7 ++-- 3 files changed, 55 insertions(+), 16 deletions(-) diff --git a/polylogue/daemon/parse_prefetch.py b/polylogue/daemon/parse_prefetch.py index c55219ee55..73edc4f01e 100644 --- a/polylogue/daemon/parse_prefetch.py +++ b/polylogue/daemon/parse_prefetch.py @@ -39,7 +39,10 @@ from polylogue.config import Config from polylogue.logging import get_logger -from polylogue.pipeline.parsed_tree_size import estimate_parsed_tree_bytes +from polylogue.pipeline.parsed_tree_size import ( + effective_physical_memory_bytes, + estimate_parsed_tree_bytes, +) from polylogue.sources import revision_backfill from polylogue.sources.dispatch import is_stream_record_provider from polylogue.sources.revision_backfill import RawParsePrefetchCache @@ -99,14 +102,7 @@ def daemon_parse_stage_worker_count() -> int: def _physical_memory_bytes() -> int | None: - try: - pages = os.sysconf("SC_PHYS_PAGES") - page_size = os.sysconf("SC_PAGE_SIZE") - except (ValueError, OSError, AttributeError): - return None - if pages <= 0 or page_size <= 0: - return None - return pages * page_size + return effective_physical_memory_bytes() def daemon_parse_stage_max_inflight_bytes() -> int: diff --git a/polylogue/pipeline/parsed_tree_size.py b/polylogue/pipeline/parsed_tree_size.py index 59af9bf96e..7073d8c86b 100644 --- a/polylogue/pipeline/parsed_tree_size.py +++ b/polylogue/pipeline/parsed_tree_size.py @@ -9,7 +9,9 @@ from __future__ import annotations +import os from collections.abc import Mapping, Sequence +from pathlib import Path from polylogue.sources.parsers.base import ParsedSession @@ -77,6 +79,7 @@ def estimate_parsed_tree_bytes(sessions: Sequence[ParsedSession]) -> int: """ total_chars = 0 object_count = 0 + total_inline_attachment_bytes = 0 for session in sessions: object_count += 1 total_chars += _text_len(session.title) @@ -125,13 +128,56 @@ def estimate_parsed_tree_bytes(sessions: Sequence[ParsedSession]) -> int: total_chars += _text_len(attachment.mime_type) total_chars += _text_len(attachment.source_url) total_chars += _text_len(attachment.caption) + if attachment.inline_bytes is not None: + # Embedded attachment payloads stay resident in the decoded + # session (and in the spill pickle) byte-for-byte -- count + # them directly, no char multiplier. + total_inline_attachment_bytes += len(attachment.inline_bytes) for event in session.session_events: object_count += 1 total_chars += _text_len(event.event_type) total_chars += _mapping_char_len(event.payload) - return _ESTIMATOR_OBJECT_OVERHEAD_BYTES * object_count + _ESTIMATOR_BYTES_PER_CHAR * total_chars + return ( + _ESTIMATOR_OBJECT_OVERHEAD_BYTES * object_count + + _ESTIMATOR_BYTES_PER_CHAR * total_chars + + total_inline_attachment_bytes + ) -__all__ = ["estimate_parsed_tree_bytes"] +def effective_physical_memory_bytes() -> int | None: + """Physical RAM available to THIS process: host RAM capped by cgroup limit. + + Under a container/systemd cgroup memory limit, ``SC_PHYS_PAGES`` reports + host RAM; sizing an adaptive cache from it starves a memory-limited + process. Reads cgroup v2 ``memory.max`` (and legacy v1 + ``memory.limit_in_bytes``) and returns the minimum of host RAM and any + finite limit; ``None`` when neither is knowable. + """ + + physical: int | None + try: + pages = os.sysconf("SC_PHYS_PAGES") + page_size = os.sysconf("SC_PAGE_SIZE") + physical = pages * page_size if pages > 0 and page_size > 0 else None + except (ValueError, OSError, AttributeError): + physical = None + limits = [] + for path in ("/sys/fs/cgroup/memory.max", "/sys/fs/cgroup/memory/memory.limit_in_bytes"): + try: + raw = Path(path).read_text().strip() + except OSError: + continue + if raw and raw != "max" and raw.isdigit(): + value = int(raw) + # v1 reports an enormous sentinel when unlimited + if 0 < value < (1 << 60): + limits.append(value) + if limits: + smallest = min(limits) + return smallest if physical is None else min(physical, smallest) + return physical + + +__all__ = ["estimate_parsed_tree_bytes", "effective_physical_memory_bytes"] diff --git a/polylogue/sources/revision_backfill.py b/polylogue/sources/revision_backfill.py index 5c1d6d3e9c..66e72bfdb6 100644 --- a/polylogue/sources/revision_backfill.py +++ b/polylogue/sources/revision_backfill.py @@ -32,7 +32,7 @@ from polylogue.archive.session_revision_membership import MembershipRevision, classify_membership_revisions from polylogue.core.enums import Provider from polylogue.pipeline.ids import session_revision_projection -from polylogue.pipeline.parsed_tree_size import estimate_parsed_tree_bytes +from polylogue.pipeline.parsed_tree_size import effective_physical_memory_bytes, estimate_parsed_tree_bytes from polylogue.pipeline.services.process_pool import parallel_threads_effective from polylogue.sources.decoders import _iter_json_stream from polylogue.sources.dispatch import is_stream_record_provider, parse_payload, parse_stream_payload @@ -1357,10 +1357,7 @@ def __init__(self, archive_root: Path, *, max_cached_payload_bytes: int | None) self.cached_payload_bytes = 0 self._decoded: dict[str, tuple[list[ParsedSession], int, int]] = {} self._decoded_tree_bytes = 0 - try: - physical = os.sysconf("SC_PHYS_PAGES") * os.sysconf("SC_PAGE_SIZE") - except (ValueError, OSError, AttributeError): - physical = 0 + physical = effective_physical_memory_bytes() or 0 self._decoded_budget = ( max(self._DECODED_CACHE_MIN_TREE_BYTES, min(self._DECODED_CACHE_MAX_TREE_BYTES, physical // 16)) if physical