Skip to content

feat(sources): per-page stage-timing telemetry for historical backfill #3206

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Sinity wants to merge 1 commit into from
Closed

feat(sources): per-page stage-timing telemetry for historical backfill #3206

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions polylogue/sources/revision_backfill.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,11 +3,13 @@
from __future__ import annotations

import json
import logging
import os
import pickle
import sqlite3
import tempfile
import threading
import time
from collections.abc import Callable, Iterator, Sequence
from contextlib import contextmanager
from dataclasses import dataclass
Expand Down Expand Up @@ -38,6 +40,8 @@
from polylogue.sources.sqlite_snapshot import looks_like_sqlite_bytes
from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore

_LOGGER = logging.getLogger(__name__)

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 20, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Use the configured logger for telemetry

In normal polylogued run and maintenance CLI execution, this raw stdlib logger does not emit INFO records: polylogue.logging.configure_logging() configures structlog's PrintLoggerFactory but neither configures the stdlib root logger nor lowers its default WARNING threshold. Consequently _LOGGER.info(...) is discarded, so the new stage-timing line is absent in the live rebuild environment it targets. Route this through polylogue.logging.get_logger (and use its supported event formatting) or explicitly bridge stdlib logging.

Useful? React with 👍 / 👎.



def _browser_snapshot_fidelity(ingest_flags: Sequence[str]) -> Literal["dom", "native"] | None:
"""Derive membership-classification browser fidelity from parser ingest flags.
Expand Down Expand Up @@ -663,6 +667,7 @@ def backfill_historical_revision_evidence(
"""
adoption_deferred = 0
quarantined = 0
stage_timings: dict[str, float] = {}
logical_keys: set[str] = set()
replay_batch_size = commit_batch_size if commit_batch_size is not None and commit_batch_size > 0 else None
replay_batched = replay_batch_size is not None
Expand All @@ -680,6 +685,7 @@ def backfill_historical_revision_evidence(
archive_context as archive,
_ParsedSessionSpill(archive_root, max_cached_payload_bytes=spill_cache_bytes) as spill,
):
census_started = time.perf_counter()
census = _census_historical_revision_evidence(
archive,
spill,
Expand All @@ -689,6 +695,7 @@ def backfill_historical_revision_evidence(
commit_batch_size=commit_batch_size,
prefetch_cache=prefetch_cache,
)
stage_timings["census"] = time.perf_counter() - census_started

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 20, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Include census receipt work in the page timings

For pages containing many raws, the census timer stops before archive.commit() and _record_raw_authority_parser_census(). The latter loops over every expanded raw and performs multiple reads plus an upsert, so a potentially material portion of the 10–17 minute page remains absent from every reported stage. Because the log has no total or unaccounted duration, this can incorrectly make the measured index stages appear responsible for page time; time this receipt/commit phase separately or keep it inside the census interval.

Useful? React with 👍 / 👎.

censused_raw_ids, _censused_keys = archive.expand_raw_membership_selection(selected_raw_ids)
# The direct backfill entry point must publish the same current-parser
# receipt as the census-only entry point before it assigns or applies
Expand Down Expand Up @@ -740,7 +747,11 @@ def commit_replay_unit() -> None:
parsed_by_raw_id: dict[str, ParsedSession] = {}
retained_bytes = 0
for raw_id in plan.accepted_raw_ids:
spill_started = time.perf_counter()
sessions, payload_bytes = spill.for_raw(archive, raw_id)
stage_timings["spill_load"] = stage_timings.get("spill_load", 0.0) + (
time.perf_counter() - spill_started
)
Comment on lines +750 to +754

@coderabbitai coderabbitai Bot Jul 20, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win

Track spill_load timing for full-revision conversions.

The spill_load timing is accurately captured here and at line 813. However, an earlier load at line 733 (sessions, _payload_bytes = spill.for_raw(archive, raw_id)) was missed, which will cause the telemetry to under-report the total time spent loading from the spill cache during replay.

📈 Proposed fix to include the missed telemetry

Apply the same timing wrapper at line 733:

                for raw_id in archive.convertible_full_revision_raw_ids(logical_key):
                    spill_started = time.perf_counter()
                    sessions, _payload_bytes = spill.for_raw(archive, raw_id)
                    stage_timings["spill_load"] = stage_timings.get("spill_load", 0.0) + (
                        time.perf_counter() - spill_started
                    )
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@polylogue/sources/revision_backfill.py` around lines 750 - 754, Add the same
spill_load timing wrapper to the earlier spill.for_raw call inside the
convertible_full_revision_raw_ids loop, recording the elapsed time in
stage_timings["spill_load"] while preserving the existing sessions and
_payload_bytes assignment.

if len(sessions) != 1:
raise RuntimeError(f"classified raw revision {raw_id} no longer parses to one session")
parsed_by_raw_id[raw_id] = sessions[0]
Expand All @@ -761,6 +772,7 @@ def commit_replay_unit() -> None:
plan,
parsed_by_raw_id,
acquired_at_ms=0,
stage_timings_s=stage_timings,
manage_transaction=not replay_batched,
bulk_fts=bulk_fts,
bulk_build=bulk_build,
Expand Down Expand Up @@ -797,7 +809,11 @@ def commit_replay_unit() -> None:
if head_raw_id is not None and archive._raw_revision_authority(head_raw_id) == "quarantined":
candidate_raw_ids.add(head_raw_id)
for raw_id in sorted(candidate_raw_ids):
spill_started = time.perf_counter()
sessions, payload_bytes = spill.for_raw(archive, raw_id)
stage_timings["spill_load"] = stage_timings.get("spill_load", 0.0) + (
time.perf_counter() - spill_started
)
for session in sessions:
session_logical_key = f"{session.source_name.value}:{session.provider_session_id}"
if session_logical_key != logical_key:
Expand Down Expand Up @@ -839,6 +855,7 @@ def commit_replay_unit() -> None:
member_sessions,
projections,
acquired_at_ms=0,
stage_timings_s=stage_timings,
manage_transaction=not replay_batched,
bulk_fts=bulk_fts,
bulk_build=bulk_build,
Expand All @@ -854,6 +871,11 @@ def commit_replay_unit() -> None:
commit_replay_unit()
if replay_batched:
archive.commit()
if stage_timings:
_LOGGER.info(
"backfill stage timings: %s",
" ".join(f"{key}={value:.1f}s" for key, value in sorted(stage_timings.items(), key=lambda kv: -kv[1])),
)
return RevisionBackfillResult(
census.scanned,
census.classified,
Expand Down