fix(storage): harden durable change-train admission #3875
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -111,6 +111,7 @@ def read_raw_failure_lifecycle(source_db: Path, *, sample_limit: int = 10) -> Ra | |
| logger.warning("could not open source.db read-only", exc_info=exc) | ||
| return RawFailureLifecycleSnapshot(False, reason=f"could not open source.db read-only: {exc}") | ||
| try: | ||
| conn.execute("BEGIN") | ||
| raw_table = conn.execute( | ||
| "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'raw_sessions'" | ||
| ).fetchone() | ||
|
|
@@ -129,43 +130,95 @@ def read_raw_failure_lifecycle(source_db: Path, *, sample_limit: int = 10) -> Ra | |
| conn.execute("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'raw_artifacts'").fetchone() | ||
| is not None | ||
| ) | ||
| sample_limit = max(0, sample_limit) | ||
| failed_cte = """ | ||
| WITH failed AS ( | ||
| SELECT r.raw_id, r.origin, r.source_path, r.source_index, | ||
| r.validation_status, r.acquired_at_ms | ||
| FROM raw_sessions AS r | ||
| WHERE (r.parse_error IS NOT NULL AND TRIM(r.parse_error) != '') | ||
| OR r.validation_status = 'failed' | ||
| ) | ||
| """ | ||
| latest_artifact_join = """ | ||
| LEFT JOIN raw_artifacts AS a | ||
| ON a.raw_id = f.raw_id | ||
| AND a.origin = f.origin | ||
| AND a.source_path = f.source_path | ||
| AND a.source_index = f.source_index | ||
| AND NOT EXISTS ( | ||
| SELECT 1 | ||
| FROM raw_artifacts AS newer | ||
| WHERE newer.raw_id = a.raw_id | ||
| AND newer.origin = a.origin | ||
| AND newer.source_path = a.source_path | ||
| AND newer.source_index = a.source_index | ||
| AND (newer.last_observed_at_ms > a.last_observed_at_ms | ||
| OR (newer.last_observed_at_ms = a.last_observed_at_ms | ||
| AND newer.artifact_id > a.artifact_id)) | ||
| ) | ||
| """ | ||
|
Comment on lines
+143
to
+160
There was a problem hiding this comment. 🚀 Performance & Scalability | 🟡 Minor | ⚡ Quick win 🧩 Analysis chain🏁 Script executed: #!/bin/bash
# Description: Inspect raw_artifacts DDL and indexes in the archive schema.
set -euo pipefail
fd -e sql -e py . polylogue/storage --exec rg -n -C4 'raw_artifacts' {} \
| rg -n -C4 'CREATE (TABLE|UNIQUE )?INDEX|CREATE TABLE|PRIMARY KEY|UNIQUE'Repository: Sinity/polylogue Length of output: 8626 🏁 Script executed: #!/bin/bash
set -euo pipefail
printf '--- target file outline/section ---\n'
wc -l polylogue/storage/raw_failure_lifecycle.py
sed -n '1,220p' polylogue/storage/raw_failure_lifecycle.py
printf '\n--- raw_artifacts DDL/index occurrences ---\n'
rg -n -C6 'CREATE TABLE raw_artifacts|CREATE INDEX idx_raw_artifacts_raw_id|CREATE UNIQUE INDEX idx_raw_artifacts_source_identity|idx_raw_artifacts' polylogue/storageRepository: Sinity/polylogue Length of output: 245 🏁 Script executed: #!/bin/bash
set -euo pipefail
printf '%s\n' '--- target file section ---'
wc -l polylogue/storage/raw_failure_lifecycle.py
sed -n '1,220p' polylogue/storage/raw_failure_lifecycle.py
printf '%s\n' ''
printf '%s\n' '--- raw_artifacts DDL/index occurrences ---'
rg -n -C6 'CREATE TABLE raw_artifacts|CREATE INDEX idx_raw_artifacts_raw_id|CREATE UNIQUE INDEX idx_raw_artifacts_source_identity|idx_raw_artifacts' polylogue/storageRepository: Sinity/polylogue Length of output: 18383 🌐 Web query:
💡 Result: In SQLite, optimizing a query involving a LEFT JOIN or a NOT EXISTS correlated subquery relies heavily on indexing the columns used in the join or correlation condition [1][2][3][4]. Performance considerations: 1. Indexing: For a LEFT JOIN, the column on the right-hand table must be indexed to allow the query planner to efficiently search for matches [3]. Similarly, for a correlated NOT EXISTS subquery, the columns in the subquery's WHERE clause that link to the outer query must be indexed to allow the subquery to perform a search rather than a full table scan [2][4]. 2. LEFT JOIN vs. NOT EXISTS: While often functionally equivalent (especially when checking for non-existence via IS NULL), they are handled differently by the query planner [1][2]. A LEFT JOIN forces the database to construct a join result, which may be less efficient than a NOT EXISTS clause if the query can be short-circuited [1]. A NOT EXISTS clause, when properly indexed, allows the database to stop searching the subquery as soon as a single match is found [1]. 3. Correlated Subquery Overhead: In older or specific configurations, correlated subqueries were sometimes routed to a slower in-memory interpreter, leading to significant performance degradation [5]. However, in modern, standard SQLite, a correlated subquery that references an indexed column in the inner table should be executed efficiently using SEARCH operations [2][4]. Ensure your column affinities match, as mismatched types can sometimes prevent the optimizer from using an index [4]. 4. Optimizer Improvements: Recent versions of SQLite (as of 2026) have introduced enhanced optimizations, including EXISTS-to-JOIN transformations, which attempt to automatically convert EXISTS clauses into JOINs when the query planner determines it is more efficient [6]. Recommendations for your schema: - Ensure that the foreign key or join columns (such as artifact_id or raw_id) are explicitly indexed in the target tables [4]. - If you are frequently filtering by last_observed_at_ms, consider a composite index that includes the join/correlation column and the observation timestamp to further speed up the lookups [7]. - Use EXPLAIN QUERY PLAN to verify that your query is using SEARCH TABLE rather than SCAN TABLE for your subqueries or join operations [2][3][4]. If you see SCAN TABLE, the database is likely performing a full table scan, which indicates a missing or ineffective index [2][3]. Citations:
Add a composite index for the latest-artifact probe.
🤖 Prompt for AI Agents |
||
| if has_artifacts: | ||
| summary_sql = ( | ||
| failed_cte | ||
| + """ | ||
| SELECT f.origin, f.validation_status, a.artifact_kind, a.support_status, | ||
| COUNT(*) AS failure_count | ||
| FROM failed AS f | ||
| """ | ||
| + latest_artifact_join | ||
| + """ | ||
| GROUP BY f.origin, f.validation_status, a.artifact_kind, a.support_status | ||
| ORDER BY f.origin, f.validation_status, a.artifact_kind, a.support_status | ||
| """ | ||
| ) | ||
| sample_sql = ( | ||
| failed_cte | ||
| + """ | ||
| , sampled AS ( | ||
| SELECT f.raw_id, f.origin, f.validation_status, f.acquired_at_ms, | ||
| a.artifact_kind, a.support_status | ||
| FROM failed AS f | ||
| """ | ||
| + latest_artifact_join | ||
| + """ | ||
| ORDER BY CASE | ||
| WHEN f.validation_status = 'failed' THEN 0 | ||
| WHEN (a.artifact_kind, a.support_status) IN ( | ||
| ('deferred_hot_jsonl_capture', 'partial_decode'), | ||
| ('terminal_corrupt_input', 'decode_failed'), | ||
| ('terminal_unsupported_shape', 'unsupported_parseable') | ||
| ) THEN 1 | ||
| ELSE 2 | ||
| END, | ||
| f.acquired_at_ms DESC, f.raw_id DESC | ||
| LIMIT ? | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
| ) | ||
| SELECT raw_id, origin, validation_status, artifact_kind, support_status | ||
| FROM sampled | ||
| """ | ||
| ) | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
| else: | ||
| summary_sql = ( | ||
| failed_cte | ||
| + """ | ||
| SELECT f.origin, f.validation_status, NULL, NULL, COUNT(*) AS failure_count | ||
| FROM failed AS f | ||
| GROUP BY f.origin, f.validation_status | ||
| ORDER BY f.origin, f.validation_status | ||
| """ | ||
| ) | ||
| sample_sql = ( | ||
| failed_cte | ||
| + """ | ||
| SELECT raw_id, origin, validation_status, NULL, NULL | ||
| FROM failed | ||
| ORDER BY acquired_at_ms DESC, raw_id DESC | ||
| LIMIT ? | ||
| """ | ||
| ) | ||
| summary_rows = conn.execute(summary_sql).fetchall() | ||
| sample_rows = conn.execute(sample_sql, (sample_limit,)).fetchall() | ||
| except sqlite3.Error as exc: | ||
| logger.warning("could not read raw failure lifecycle", exc_info=exc) | ||
| return RawFailureLifecycleSnapshot(False, reason=f"could not read raw failure lifecycle: {exc}") | ||
|
|
@@ -176,25 +229,33 @@ def read_raw_failure_lifecycle(source_db: Path, *, sample_limit: int = 10) -> Ra | |
| by_artifact_kind: Counter[str] = Counter() | ||
| counts: Counter[str] = Counter() | ||
| samples: list[dict[str, str | None]] = [] | ||
| for row in summary_rows: | ||
| origin = str(row[0] or "unknown") | ||
| artifact_kind = str(row[2]) if row[2] is not None else None | ||
| support_status = str(row[3]) if row[3] is not None else None | ||
| validation_failed = str(row[1] or "") == "failed" | ||
| lifecycle = _lifecycle(artifact_kind, support_status, validation_failed=validation_failed) | ||
| count = int(row[4]) | ||
| counts[lifecycle] += count | ||
| by_origin[origin] += count | ||
| by_artifact_kind[artifact_kind or "<none>"] += count | ||
| for row in sample_rows: | ||
| origin = str(row[1] or "unknown") | ||
| artifact_kind = str(row[3]) if row[3] is not None else None | ||
| support_status = str(row[4]) if row[4] is not None else None | ||
| samples.append( | ||
| { | ||
| "raw_id": str(row[0]), | ||
| "origin": origin, | ||
| "artifact_kind": artifact_kind, | ||
| "support_status": support_status, | ||
| "lifecycle": _lifecycle( | ||
| artifact_kind, | ||
| support_status, | ||
| validation_failed=str(row[2] or "") == "failed", | ||
| ), | ||
| } | ||
| ) | ||
| return RawFailureLifecycleSnapshot( | ||
| available=True, | ||
| parse_failures=parse_failures, | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -312,15 +312,69 @@ def initialize_archive_database( | |
| def initialize_active_archive_root(root: Path) -> None: | ||
| """Create or initialize every tier database in an archive root.""" | ||
| from polylogue.storage.archive_identity import ArchiveLocation, OwnedArchiveLocation | ||
| from polylogue.storage.sqlite.durable_change_train import ( | ||
| _record_fresh_durable_bootstrap, | ||
| _record_fresh_durable_bootstrap_intent, | ||
| _validate_fresh_durable_bootstrap_intent, | ||
| ) | ||
|
|
||
| with OwnedArchiveLocation.acquire( | ||
| ArchiveLocation.resolve(root), | ||
| owner_id=f"bootstrap:{os.getpid()}", | ||
| allow_reentrant=True, | ||
| ): | ||
| # Classify the archive after acquiring ownership. Another process may | ||
| # publish a marker or durable train while the probe is in flight. | ||
| durable_tier_exists = any( | ||
| (root / archive_tier_spec(tier).filename).exists() for tier in DURABLE_MIGRATION_TIERS | ||
| ) | ||
| manifest_root = root / ".maintenance-state" / "durable-change-trains" | ||
| has_durable_train_state = any(manifest_root.glob("*.json")) | ||
| has_bootstrap_marker = (manifest_root / ".bootstrap").is_file() | ||
| pending_bootstrap_path = manifest_root / ".bootstrap.pending" | ||
| has_pending_bootstrap = pending_bootstrap_path.is_file() | ||
| if has_pending_bootstrap: | ||
| _validate_fresh_durable_bootstrap_intent(root) | ||
|
Comment on lines
+336
to
+337
There was a problem hiding this comment.
If the process or host stops after the completed Useful? React with 👍 / 👎. |
||
| if has_durable_train_state: | ||
| raise RuntimeError( | ||
| "fresh durable bootstrap intent conflicts with durable train state; " | ||
| "refusing to guess which authority is current" | ||
| ) | ||
| fresh_durable_bootstrap = ( | ||
| not durable_tier_exists | ||
| and not has_durable_train_state | ||
| and not has_bootstrap_marker | ||
| and not has_pending_bootstrap | ||
| ) | ||
| recovering_fresh_durable_bootstrap = fresh_durable_bootstrap or ( | ||
| has_pending_bootstrap and not has_bootstrap_marker | ||
| ) | ||
| pre_marker_adoption = ( | ||
| (root / archive_tier_spec(ArchiveTier.SOURCE).filename).is_file() | ||
| and all((root / archive_tier_spec(tier).filename).is_file() for tier in DURABLE_MIGRATION_TIERS) | ||
| and manifest_root.is_dir() | ||
| and not has_durable_train_state | ||
| and not has_bootstrap_marker | ||
|
Comment on lines
+352
to
+357
There was a problem hiding this comment.
When a pre-marker archive still has AGENTS.md reference: AGENTS.md:L118-L121 Useful? React with 👍 / 👎. |
||
| and not has_pending_bootstrap | ||
| ) | ||
|
Comment on lines
+353
to
+359
There was a problem hiding this comment.
When a post-adoption archive loses its durable-change-train directory, these absence checks classify it as a legacy pre-marker archive and skip reconciliation; AGENTS.md reference: AGENTS.md:L189-L192 Useful? React with 👍 / 👎. |
||
| if fresh_durable_bootstrap: | ||
| _record_fresh_durable_bootstrap_intent(root) | ||
| if not recovering_fresh_durable_bootstrap and not pre_marker_adoption: | ||
| reconcile_durable_change_trains_on_startup(root) | ||
| for spec in ARCHIVE_TIER_SPECS.values(): | ||
| initialize_archive_database(root / spec.filename, spec.tier) | ||
| if recovering_fresh_durable_bootstrap: | ||
| _record_fresh_durable_bootstrap(root) | ||
|
Comment on lines
365
to
+367
There was a problem hiding this comment.
On a brand-new archive, if a later Useful? React with 👍 / 👎. |
||
| elif pre_marker_adoption: | ||
| from polylogue.storage.sqlite.durable_change_train import _adopt_pre_marker_durable_bootstrap | ||
|
|
||
| _adopt_pre_marker_durable_bootstrap(root) | ||
| reconcile_durable_change_trains_on_startup(root) | ||
| elif has_pending_bootstrap: | ||
| # A crash after publishing the completed marker but before | ||
| # removing the intent is harmless. Keep the intent until the | ||
| # completed marker has passed normal startup reconciliation. | ||
| pending_bootstrap_path.unlink(missing_ok=True) | ||
|
|
||
|
|
||
| def reconcile_durable_change_trains_on_startup(root: Path) -> tuple[Path, ...]: | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.