fix(state): do not stamp empty FTS after interrupted optimize-storage demote
Demote wrote the empty v23 schema via executescript inside BEGIN IMMEDIATE, which commits early and can leave trash + empty indexes without rebuild markers. Re-run then tore down trash and stamped fts_storage_version with docsize=0, permanently losing historical session search. Stage markers with the demote, create schema only after they are durable, heal empty-index bookkeeping on resume, and refuse settle until the base index is populated. Settle refusal returns ok=False instead of raising, and resume fails fast if the base v23 table cannot be re-created. Orphan-marker repair only resets a missing fts_rebuild_progress to 0 once the index is known empty: the chunk worker replays its whole selected id range without an anti-join, so a partially indexed DB that lost only its progress key is first reset to a known-empty surface, then rebuilt. Ported onto the SessionDB mixin split (hermes_state_search.py / hermes_state_schema.py).
This commit is contained in:
Adolanium
kshitij
parent
4e0a775580
commit
b2d5995fc6
|
|
@ -730,7 +730,8 @@ class SessionSchemaMixin:
|
|||
# FTS_STORAGE_VERSION; a legacy DB is left at whatever it had
|
||||
# (absent/0) until `optimize-storage` runs. An INTERRUPTED
|
||||
# optimize (legacy vtables already demoted, but rebuild markers
|
||||
# or demoted trash tables still present) is NOT stamped either —
|
||||
# or demoted trash tables still present, or an empty external
|
||||
# index against non-empty messages) is NOT stamped either —
|
||||
# the marker is the source of truth for "fully optimized", and
|
||||
# `fts_optimize_available()` keeps offering the resume until the
|
||||
# transition actually completes.
|
||||
|
|
@ -742,6 +743,7 @@ class SessionSchemaMixin:
|
|||
"WHERE key = 'fts_rebuild_high_water' LIMIT 1"
|
||||
).fetchone() is None
|
||||
and not self._has_fts_trash(cursor)
|
||||
and not self._fts_external_index_empty_with_messages(cursor)
|
||||
):
|
||||
self.set_meta(
|
||||
"fts_storage_version", str(FTS_STORAGE_VERSION), cursor=cursor
|
||||
|
|
|
|||
|
|
@ -356,13 +356,169 @@ class SessionSearchMixin:
|
|||
self._ensure_fts_cjk_schema(self._conn)
|
||||
self._conn.commit()
|
||||
|
||||
def _fts_external_index_empty_with_messages(self, conn) -> bool:
|
||||
"""True when the base FTS table exists but indexes nothing while
|
||||
``messages`` has rows. Caller must hold ``self._lock``.
|
||||
|
||||
This is the post-demote empty-index shape: external-content FTS with
|
||||
zero ``messages_fts_docsize`` rows against a non-empty messages table.
|
||||
Healthy installs (and mid-backfill installs that still hold markers)
|
||||
never match.
|
||||
"""
|
||||
try:
|
||||
n_msg = conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0]
|
||||
if int(n_msg) <= 0:
|
||||
return False
|
||||
# docsize is the authoritative "is this rowid indexed" surface for
|
||||
# external-content FTS5; COUNT(*) on the virtual table itself is
|
||||
# not reliable across SQLite builds.
|
||||
n_fts = conn.execute(
|
||||
"SELECT COUNT(*) FROM messages_fts_docsize"
|
||||
).fetchone()[0]
|
||||
return int(n_fts) == 0
|
||||
except sqlite3.OperationalError:
|
||||
# Table absent / FTS disabled mid-init — not this failure class.
|
||||
return False
|
||||
|
||||
def _fts_index_known_empty(self, conn) -> bool:
|
||||
"""True when the base external-content index holds no rows.
|
||||
|
||||
A missing table counts as empty: the schema ensure that follows
|
||||
creates it fresh.
|
||||
"""
|
||||
try:
|
||||
n = conn.execute(
|
||||
"SELECT COUNT(*) FROM messages_fts_docsize"
|
||||
).fetchone()[0]
|
||||
return int(n) == 0
|
||||
except sqlite3.OperationalError:
|
||||
return True
|
||||
|
||||
def _reset_fts_index_to_empty(self, conn) -> None:
|
||||
"""Delete every indexed row from the v23 external-content tables.
|
||||
|
||||
FTS5 supports a no-WHERE DELETE on external-content tables as an
|
||||
efficient drop-all. The backfill chunk worker replays its whole
|
||||
selected id range with no anti-join, so a replay from zero is only
|
||||
safe once the index is known empty — this is how a partially indexed
|
||||
DB gets there.
|
||||
"""
|
||||
for tbl in ("messages_fts", "messages_fts_trigram"):
|
||||
try:
|
||||
conn.execute(f"DELETE FROM {tbl}")
|
||||
except sqlite3.OperationalError:
|
||||
pass # table absent — already an empty surface
|
||||
|
||||
def _seed_fts_rebuild_markers(self, conn, *, force: bool = False) -> int:
|
||||
"""Write ``fts_rebuild_high_water`` / ``fts_rebuild_progress`` for a
|
||||
full backfill. Returns the high-water id.
|
||||
|
||||
When ``force`` is False and high_water is already set, only repairs a
|
||||
missing progress key (stuck no-op when high_water exists alone), and
|
||||
only after the index is known empty: the chunk worker replays its
|
||||
whole selected id range without an anti-join, so a partially indexed
|
||||
DB is first reset to a known-empty surface rather than rebuilt from
|
||||
zero on top of surviving rows. Caller must hold the write
|
||||
transaction / lock as appropriate.
|
||||
"""
|
||||
existing_hw = conn.execute(
|
||||
"SELECT value FROM state_meta WHERE key = 'fts_rebuild_high_water'"
|
||||
).fetchone()
|
||||
if existing_hw is not None and not force:
|
||||
hw = int(existing_hw[0])
|
||||
progress = conn.execute(
|
||||
"SELECT value FROM state_meta WHERE key = 'fts_rebuild_progress'"
|
||||
).fetchone()
|
||||
if progress is None:
|
||||
# high_water without progress: fts_rebuild_step treats missing
|
||||
# progress as "done by another process" and optimize would
|
||||
# no-op then stamp. Re-seed progress so the chunk loop runs.
|
||||
if not self._fts_index_known_empty(conn):
|
||||
self._reset_fts_index_to_empty(conn)
|
||||
conn.execute(
|
||||
"INSERT INTO state_meta (key, value) VALUES "
|
||||
"('fts_rebuild_progress', '0') "
|
||||
"ON CONFLICT(key) DO UPDATE SET value = excluded.value"
|
||||
)
|
||||
return hw
|
||||
|
||||
hw = conn.execute(
|
||||
"SELECT COALESCE(MAX(id), 0) FROM messages"
|
||||
).fetchone()[0]
|
||||
for k, v in (
|
||||
("fts_rebuild_high_water", str(hw)),
|
||||
("fts_rebuild_progress", "0"),
|
||||
):
|
||||
conn.execute(
|
||||
"INSERT INTO state_meta (key, value) VALUES (?, ?) "
|
||||
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
|
||||
(k, v),
|
||||
)
|
||||
return int(hw)
|
||||
|
||||
def _repair_optimize_bookkeeping(self) -> None:
|
||||
"""Heal interrupted demote/backfill bookkeeping before optimize runs.
|
||||
|
||||
Covers two post-#65798 failure classes:
|
||||
|
||||
1. Empty external-content index with messages present and no rebuild
|
||||
markers (demote crash window after empty v23 tables landed but
|
||||
before markers, or settle that stamped without backfill). Seed a
|
||||
full backfill.
|
||||
2. ``fts_rebuild_high_water`` present without ``fts_rebuild_progress``
|
||||
(partial meta) — seed progress so the chunk loop is not a no-op,
|
||||
resetting a partially populated index to a known-empty surface
|
||||
first so the anti-join-free chunk replay cannot duplicate rows.
|
||||
|
||||
Must not invent markers on a still-legacy inline DB: that would make
|
||||
``optimize_fts_storage`` skip demote (``legacy and not pending``) and
|
||||
attempt v23-shaped INSERTs against the inline table forever.
|
||||
"""
|
||||
def _do(conn):
|
||||
existing_hw = conn.execute(
|
||||
"SELECT value FROM state_meta "
|
||||
"WHERE key = 'fts_rebuild_high_water'"
|
||||
).fetchone()
|
||||
|
||||
if existing_hw is not None:
|
||||
# Repair orphan high_water-without-progress only. Never
|
||||
# invent a fresh claim on a healthy complete index.
|
||||
progress = conn.execute(
|
||||
"SELECT 1 FROM state_meta "
|
||||
"WHERE key = 'fts_rebuild_progress'"
|
||||
).fetchone()
|
||||
if progress is None:
|
||||
if not self._fts_index_known_empty(conn):
|
||||
self._reset_fts_index_to_empty(conn)
|
||||
conn.execute(
|
||||
"INSERT INTO state_meta (key, value) VALUES "
|
||||
"('fts_rebuild_progress', '0') "
|
||||
"ON CONFLICT(key) DO UPDATE SET value = '0'"
|
||||
)
|
||||
return
|
||||
|
||||
# No markers. On a still-legacy DB demote owns marker creation.
|
||||
if self._db_has_legacy_inline_fts(conn):
|
||||
return
|
||||
|
||||
# Non-legacy empty external index (demote crash window / premature
|
||||
# stamp): seed a full backfill claim.
|
||||
if self._fts_external_index_empty_with_messages(conn):
|
||||
conn.execute(
|
||||
"DELETE FROM state_meta WHERE key = 'fts_storage_version'"
|
||||
)
|
||||
self._seed_fts_rebuild_markers(conn, force=True)
|
||||
self._execute_write(_do)
|
||||
|
||||
def fts_optimize_available(self) -> bool:
|
||||
"""True when `optimize_fts_storage()` has work to do: either this DB
|
||||
is a legacy inline-FTS install that can be optimized to the v23
|
||||
external-content schema, or a previous optimize run was interrupted
|
||||
(legacy vtables already demoted, but backfill markers and/or trash
|
||||
tables remain) and re-running would resume it, or the CJK-bigram
|
||||
index needs a backfill/rebuild on this tokenizer-capable host.
|
||||
index needs a backfill/rebuild on this tokenizer-capable host, or
|
||||
a prior demote left an empty external-content index without markers
|
||||
(healable on re-run).
|
||||
False for fresh and fully-optimized installs (and when FTS5 is
|
||||
unavailable)."""
|
||||
if not self._fts_enabled or self.read_only:
|
||||
|
|
@ -388,14 +544,27 @@ class SessionSearchMixin:
|
|||
f"('fts_cjk_rebuild_high_water', '{FTS_CJK_STALE_KEY}') LIMIT 1"
|
||||
).fetchone():
|
||||
return True
|
||||
return self._has_fts_trash(self._conn)
|
||||
if self._has_fts_trash(self._conn):
|
||||
return True
|
||||
# Pre-fix crash window: empty external-content index with
|
||||
# messages still present, no markers, no trash (teardown already
|
||||
# finished or never needed). Re-run seeds markers and backfills.
|
||||
return self._fts_external_index_empty_with_messages(self._conn)
|
||||
|
||||
def _demote_legacy_fts_to_trash(self) -> int:
|
||||
"""Demote the legacy inline FTS vtables and stage their shadow tables
|
||||
for chunked teardown. Returns MAX(messages.id) as the rebuild high
|
||||
water. O(1) schema surgery — the heavy delete is deferred to the
|
||||
chunked teardown, exactly as the validated auto path did."""
|
||||
def _do(conn):
|
||||
chunked teardown, exactly as the validated auto path did.
|
||||
|
||||
Markers are written in the same BEGIN IMMEDIATE as the demote, *before*
|
||||
the empty v23 schema is created. Schema creation uses
|
||||
``executescript`` and therefore cannot run inside that transaction
|
||||
(it issues an implicit COMMIT — see the CJK recreate path). Creating
|
||||
the empty schema only after markers are durable closes the crash
|
||||
window where trash + empty v23 tables exist with no backfill claim.
|
||||
"""
|
||||
def _stage(conn):
|
||||
self._drop_fts_triggers(conn)
|
||||
conn.execute("DROP VIEW IF EXISTS messages_fts_trigram_src")
|
||||
had = bool(conn.execute(
|
||||
|
|
@ -420,22 +589,35 @@ class SessionSearchMixin:
|
|||
]
|
||||
for sh in shadows:
|
||||
conn.execute(f"ALTER TABLE {sh} RENAME TO fts_v22_trash_{sh}")
|
||||
# Create the new v23 empty schema + set the backfill markers.
|
||||
self._ensure_fts_schema(conn, "messages_fts", FTS_SQL)
|
||||
self._ensure_fts_schema(conn, "messages_fts_trigram", FTS_TRIGRAM_SQL)
|
||||
hw = conn.execute("SELECT COALESCE(MAX(id), 0) FROM messages").fetchone()[0]
|
||||
for k, v in (
|
||||
("fts_rebuild_high_water", str(hw)),
|
||||
("fts_rebuild_progress", "0"),
|
||||
):
|
||||
conn.execute(
|
||||
"INSERT INTO state_meta (key, value) VALUES (?, ?) "
|
||||
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
|
||||
(k, v),
|
||||
)
|
||||
conn.execute("DELETE FROM state_meta WHERE key = 'fts_optimize_available'")
|
||||
# Claim the backfill *before* empty v23 tables exist. A crash
|
||||
# between this commit and schema ensure still leaves markers, so
|
||||
# optimize-storage resumes instead of tearing down trash and
|
||||
# stamping an empty index as complete.
|
||||
hw = self._seed_fts_rebuild_markers(conn, force=True)
|
||||
conn.execute(
|
||||
"DELETE FROM state_meta WHERE key = 'fts_optimize_available'"
|
||||
)
|
||||
return hw
|
||||
return int(self._execute_write(_do))
|
||||
|
||||
hw = int(self._execute_write(_stage))
|
||||
|
||||
# Create the empty v23 schema outside the write transaction —
|
||||
# ``_ensure_fts_schema`` uses executescript(), which implicitly
|
||||
# commits any pending transaction and must not run inside
|
||||
# ``_execute_write``'s BEGIN IMMEDIATE (same rule as the CJK recreate
|
||||
# path above). Markers are already durable.
|
||||
with self._lock:
|
||||
base_ok = self._ensure_fts_schema(self._conn, "messages_fts", FTS_SQL)
|
||||
trigram_ok = self._ensure_fts_schema(
|
||||
self._conn, "messages_fts_trigram", FTS_TRIGRAM_SQL
|
||||
)
|
||||
self._trigram_available = bool(trigram_ok)
|
||||
if not base_ok:
|
||||
raise sqlite3.OperationalError(
|
||||
"failed to create v23 messages_fts during optimize-storage demote"
|
||||
)
|
||||
self._conn.commit()
|
||||
return hw
|
||||
|
||||
def optimize_fts_storage(
|
||||
self,
|
||||
|
|
@ -458,6 +640,12 @@ class SessionSearchMixin:
|
|||
if self.read_only:
|
||||
return {"ok": False, "reason": "read_only"}
|
||||
|
||||
# Heal empty-index / orphan-marker bookkeeping from an interrupted
|
||||
# demote *before* deciding whether to demote again. This re-seeds
|
||||
# markers when trash was already staged (or torn down) without a
|
||||
# backfill claim so the phases below actually run.
|
||||
self._repair_optimize_bookkeeping()
|
||||
|
||||
# Only demote if we're actually still on the legacy shape. If a prior
|
||||
# run already demoted (markers/trash present), skip straight to
|
||||
# finishing the backfill + teardown — this is what makes re-running
|
||||
|
|
@ -467,6 +655,26 @@ class SessionSearchMixin:
|
|||
pending = self.get_meta("fts_rebuild_high_water") is not None
|
||||
if legacy and not pending:
|
||||
self._demote_legacy_fts_to_trash()
|
||||
elif pending and not legacy:
|
||||
# Resume mid-demote: markers exist, empty v23 tables may still be
|
||||
# missing if the process died between the staged demote commit and
|
||||
# schema ensure. Re-ensure is IF NOT EXISTS and cheap.
|
||||
with self._lock:
|
||||
base_ok = self._ensure_fts_schema(
|
||||
self._conn, "messages_fts", FTS_SQL
|
||||
)
|
||||
trigram_ok = self._ensure_fts_schema(
|
||||
self._conn, "messages_fts_trigram", FTS_TRIGRAM_SQL
|
||||
)
|
||||
self._trigram_available = bool(trigram_ok)
|
||||
if not base_ok:
|
||||
# Fail fast: without the base table the backfill loop
|
||||
# below would retry "no such table" errors forever.
|
||||
raise sqlite3.OperationalError(
|
||||
"failed to re-create v23 messages_fts "
|
||||
"on optimize-storage resume"
|
||||
)
|
||||
self._conn.commit()
|
||||
|
||||
# A stale CJK index (triggers dropped by a tokenizer-less process)
|
||||
# can only be recovered from scratch — reset it now so the cjk
|
||||
|
|
@ -536,6 +744,29 @@ class SessionSearchMixin:
|
|||
_emit("teardown")
|
||||
_pause(time.monotonic() - _t0)
|
||||
|
||||
# Refuse to stamp "optimized" while work remains or the base index is
|
||||
# still empty against a non-empty messages table. Pre-fix code could
|
||||
# tear down trash and settle after a no-op backfill when markers were
|
||||
# missing — permanent search-index loss for historical rows.
|
||||
with self._lock:
|
||||
still_pending = self._conn.execute(
|
||||
"SELECT 1 FROM state_meta "
|
||||
"WHERE key = 'fts_rebuild_high_water' LIMIT 1"
|
||||
).fetchone() is not None
|
||||
still_trash = self._has_fts_trash(self._conn)
|
||||
empty_index = self._fts_external_index_empty_with_messages(self._conn)
|
||||
if still_pending or still_trash or empty_index:
|
||||
reason = (
|
||||
"backfill_incomplete" if still_pending or empty_index
|
||||
else "teardown_incomplete"
|
||||
)
|
||||
logger.warning(
|
||||
"FTS storage optimization did not settle (%s): "
|
||||
"pending=%s trash=%s empty_index=%s",
|
||||
reason, still_pending, still_trash, empty_index,
|
||||
)
|
||||
return {"ok": False, "reason": reason, "vacuumed": None}
|
||||
|
||||
# Phase 3: reclaim freed pages to the OS.
|
||||
vacuum_ok = None
|
||||
if vacuum:
|
||||
|
|
@ -572,6 +803,18 @@ class SessionSearchMixin:
|
|||
# DB opened only by pre-decoupling code still settles). The FTS-layout
|
||||
# marker is the source of truth for "is this DB optimized".
|
||||
def _settle(conn):
|
||||
# Re-check inside the write transaction so a concurrent writer
|
||||
# cannot race a stamp past incomplete work. Returns a refusal
|
||||
# reason (stamping nothing) or None once the stamp is written.
|
||||
if conn.execute(
|
||||
"SELECT 1 FROM state_meta "
|
||||
"WHERE key = 'fts_rebuild_high_water' LIMIT 1"
|
||||
).fetchone() is not None:
|
||||
return "backfill_incomplete"
|
||||
if self._has_fts_trash(conn):
|
||||
return "teardown_incomplete"
|
||||
if self._fts_external_index_empty_with_messages(conn):
|
||||
return "backfill_incomplete"
|
||||
conn.execute(
|
||||
"INSERT INTO state_meta (key, value) VALUES ('fts_storage_version', ?) "
|
||||
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
|
||||
|
|
@ -582,7 +825,17 @@ class SessionSearchMixin:
|
|||
"UPDATE schema_version SET version = ? WHERE version < ?",
|
||||
(SCHEMA_VERSION, SCHEMA_VERSION),
|
||||
)
|
||||
self._execute_write(_settle)
|
||||
return None
|
||||
refusal = self._execute_write(_settle)
|
||||
if refusal is not None:
|
||||
# A concurrent process re-seeded markers, left trash, or emptied
|
||||
# the index between the pre-vacuum check above and this write
|
||||
# transaction. Nothing was stamped. Report the failure instead of
|
||||
# crashing the CLI with a traceback; a re-run can still settle.
|
||||
logger.warning(
|
||||
"FTS storage optimization settle refused (%s)", refusal
|
||||
)
|
||||
return {"ok": False, "reason": refusal, "vacuumed": vacuum_ok}
|
||||
_emit("done")
|
||||
logger.info(
|
||||
"FTS storage optimization complete (layout v%d).", FTS_STORAGE_VERSION
|
||||
|
|
|
|||
|
|
@ -2319,6 +2319,319 @@ class TestFTSExternalContentMigration:
|
|||
|
||||
|
||||
|
||||
def _simulate_pre_fix_demote_crash_window(self, db):
|
||||
"""Replay the pre-fix demote crash window: trash + empty v23 schema,
|
||||
no rebuild markers (executescript committed mid-demote before markers).
|
||||
|
||||
Mirrors what happened when ``_ensure_fts_schema`` ran inside
|
||||
``_execute_write`` and the process died before the marker writes.
|
||||
"""
|
||||
from hermes_state import FTS_SQL, FTS_TRIGRAM_SQL
|
||||
|
||||
conn = db._conn
|
||||
db._drop_fts_triggers(conn)
|
||||
conn.execute("DROP VIEW IF EXISTS messages_fts_trigram_src")
|
||||
had = bool(conn.execute(
|
||||
"SELECT 1 FROM sqlite_master WHERE type = 'table' "
|
||||
"AND name IN ('messages_fts', 'messages_fts_trigram') "
|
||||
"AND sql LIKE 'CREATE VIRTUAL TABLE%' LIMIT 1"
|
||||
).fetchone())
|
||||
assert had, "sanity: expected legacy/virtual FTS tables to demote"
|
||||
conn.execute("PRAGMA writable_schema=ON")
|
||||
conn.execute(
|
||||
"DELETE FROM sqlite_master WHERE type = 'table' "
|
||||
"AND name IN ('messages_fts', 'messages_fts_trigram') "
|
||||
"AND sql LIKE 'CREATE VIRTUAL TABLE%'"
|
||||
)
|
||||
conn.execute("PRAGMA writable_schema=RESET")
|
||||
shadows = [
|
||||
r[0] for r in conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type = 'table' "
|
||||
"AND (name LIKE 'messages_fts_%' ESCAPE '\\' "
|
||||
"OR name LIKE 'messages_fts_trigram_%' ESCAPE '\\')"
|
||||
).fetchall()
|
||||
]
|
||||
for sh in shadows:
|
||||
conn.execute(f"ALTER TABLE {sh} RENAME TO fts_v22_trash_{sh}")
|
||||
# executescript commits — empty v23 tables without markers.
|
||||
conn.executescript(FTS_SQL)
|
||||
try:
|
||||
conn.executescript(FTS_TRIGRAM_SQL)
|
||||
except sqlite3.OperationalError:
|
||||
pass
|
||||
# Intentionally leave fts_rebuild_* unset (the crash window).
|
||||
|
||||
def test_optimize_resume_after_demote_crash_window_restores_search(
|
||||
self, tmp_path
|
||||
):
|
||||
"""Pre-fix: demote crash left trash + empty v23 index, no markers.
|
||||
Re-run tore down trash and stamped optimized with docsize=0 — permanent
|
||||
search loss for historical rows. Re-run must backfill and restore."""
|
||||
db_path = tmp_path / "v22.db"
|
||||
self._build_v22_db(db_path)
|
||||
|
||||
db = SessionDB(db_path=db_path)
|
||||
try:
|
||||
assert len(db.search_messages("deployment")) == 1
|
||||
self._simulate_pre_fix_demote_crash_window(db)
|
||||
# Crash window shape: no markers, trash present, empty index.
|
||||
assert db.get_meta("fts_rebuild_high_water") is None
|
||||
assert db.get_meta("fts_rebuild_progress") is None
|
||||
assert db._has_fts_trash(db._conn) is True
|
||||
assert db._conn.execute(
|
||||
"SELECT COUNT(*) FROM messages_fts_docsize"
|
||||
).fetchone()[0] == 0
|
||||
assert len(db.search_messages("deployment")) == 0
|
||||
|
||||
# Still offered (trash and/or empty-index heal).
|
||||
assert db.fts_optimize_available() is True
|
||||
|
||||
result = db.optimize_fts_storage(vacuum=False)
|
||||
assert result["ok"] is True
|
||||
assert db.fts_rebuild_status() is None
|
||||
assert db.fts_optimize_available() is False
|
||||
assert db.get_meta("fts_storage_version") == str(
|
||||
hermes_state.FTS_STORAGE_VERSION
|
||||
)
|
||||
assert db._conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE name LIKE '%_v22_trash%'"
|
||||
).fetchall() == []
|
||||
# Historical rows searchable again; index fully populated.
|
||||
assert len(db.search_messages("deployment")) == 1
|
||||
assert len(db.search_messages("TOOLBLOB")) == 1
|
||||
n_msg = db._conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0]
|
||||
n_fts = db._conn.execute(
|
||||
"SELECT COUNT(*) FROM messages_fts_docsize"
|
||||
).fetchone()[0]
|
||||
assert n_fts == n_msg
|
||||
db._conn.execute(
|
||||
"INSERT INTO messages_fts(messages_fts, rank) VALUES('integrity-check', 1)"
|
||||
)
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
def test_optimize_heals_premature_stamp_with_empty_index(self, tmp_path):
|
||||
"""Pre-fix settle could stamp fts_storage_version after tearing down
|
||||
trash with an empty index and no markers. Re-run must clear the stamp,
|
||||
backfill, and re-earn the layout version."""
|
||||
db_path = tmp_path / "v22.db"
|
||||
self._build_v22_db(db_path)
|
||||
|
||||
db = SessionDB(db_path=db_path)
|
||||
try:
|
||||
self._simulate_pre_fix_demote_crash_window(db)
|
||||
# Simulate the bad resume: trash already gone, empty index stamped.
|
||||
trash = [
|
||||
r[0] for r in db._conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type = 'table' "
|
||||
"AND name LIKE 'fts\\_v22\\_trash\\_%' ESCAPE '\\'"
|
||||
).fetchall()
|
||||
]
|
||||
for tbl in trash:
|
||||
db._conn.execute(f"DROP TABLE IF EXISTS {tbl}")
|
||||
db._conn.execute(
|
||||
"INSERT INTO state_meta (key, value) VALUES "
|
||||
"('fts_storage_version', ?) "
|
||||
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
|
||||
(str(hermes_state.FTS_STORAGE_VERSION),),
|
||||
)
|
||||
db._conn.commit()
|
||||
|
||||
assert db.get_meta("fts_rebuild_high_water") is None
|
||||
assert db._has_fts_trash(db._conn) is False
|
||||
assert db._fts_external_index_empty_with_messages(db._conn) is True
|
||||
# Must still be offered despite the premature stamp.
|
||||
assert db.fts_optimize_available() is True
|
||||
assert len(db.search_messages("deployment")) == 0
|
||||
|
||||
result = db.optimize_fts_storage(vacuum=False)
|
||||
assert result["ok"] is True
|
||||
assert len(db.search_messages("deployment")) == 1
|
||||
assert db.get_meta("fts_storage_version") == str(
|
||||
hermes_state.FTS_STORAGE_VERSION
|
||||
)
|
||||
assert db.fts_optimize_available() is False
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
def test_optimize_heals_high_water_without_progress(self, tmp_path):
|
||||
"""high_water without progress used to make fts_rebuild_step return
|
||||
False immediately (treated as finished by another process), then
|
||||
settle stamped success while the marker remained. Re-seed progress
|
||||
and complete the empty-index backfill."""
|
||||
db_path = tmp_path / "v22.db"
|
||||
self._build_v22_db(db_path)
|
||||
db = SessionDB(db_path=db_path)
|
||||
try:
|
||||
self._simulate_pre_fix_demote_crash_window(db)
|
||||
hw = db._conn.execute(
|
||||
"SELECT COALESCE(MAX(id), 0) FROM messages"
|
||||
).fetchone()[0]
|
||||
# Orphan shape: high_water alone on an empty external index.
|
||||
db.set_meta("fts_rebuild_high_water", str(hw))
|
||||
db._conn.execute(
|
||||
"DELETE FROM state_meta WHERE key = ?", ("fts_rebuild_progress",)
|
||||
)
|
||||
db._conn.commit()
|
||||
assert db.get_meta("fts_rebuild_progress") is None
|
||||
assert db.fts_optimize_available() is True
|
||||
# Empty index: base FTS MATCH finds nothing (gap LIKE may still
|
||||
# supplement when high_water is set — that is intentional).
|
||||
assert db._conn.execute(
|
||||
"SELECT COUNT(*) FROM messages_fts_docsize"
|
||||
).fetchone()[0] == 0
|
||||
|
||||
result = db.optimize_fts_storage(vacuum=False)
|
||||
assert result["ok"] is True
|
||||
assert db.get_meta("fts_rebuild_high_water") is None
|
||||
assert db.get_meta("fts_rebuild_progress") is None
|
||||
n_msg = db._conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0]
|
||||
n_fts = db._conn.execute(
|
||||
"SELECT COUNT(*) FROM messages_fts_docsize"
|
||||
).fetchone()[0]
|
||||
assert n_fts == n_msg
|
||||
assert len(db.search_messages("deployment")) == 1
|
||||
assert db.fts_optimize_available() is False
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
def test_repair_rebuilds_partial_index_without_duplicates(self, tmp_path):
|
||||
"""high_water without progress on a PARTIALLY indexed DB must not
|
||||
replay the backfill from zero on top of surviving rows: the chunk
|
||||
worker inserts its whole id range with no anti-join, so replay
|
||||
duplicates every already-indexed row. Recovery must reset the index
|
||||
to a known-empty surface first, then rebuild."""
|
||||
db_path = tmp_path / "v22.db"
|
||||
self._build_v22_db(db_path)
|
||||
db = SessionDB(db_path=db_path)
|
||||
try:
|
||||
self._simulate_pre_fix_demote_crash_window(db)
|
||||
hw = db._conn.execute(
|
||||
"SELECT COALESCE(MAX(id), 0) FROM messages"
|
||||
).fetchone()[0]
|
||||
db.set_meta("fts_rebuild_high_water", str(hw))
|
||||
db._conn.execute(
|
||||
"DELETE FROM state_meta WHERE key = ?", ("fts_rebuild_progress",)
|
||||
)
|
||||
# Partial index: one row survived from an interrupted backfill.
|
||||
db._conn.execute(
|
||||
"INSERT INTO messages_fts(rowid, content, tool_name, tool_calls) "
|
||||
"SELECT id, content, tool_name, tool_calls FROM messages "
|
||||
"WHERE id = 1"
|
||||
)
|
||||
db._conn.commit()
|
||||
assert db._conn.execute(
|
||||
"SELECT COUNT(*) FROM messages_fts_docsize"
|
||||
).fetchone()[0] == 1
|
||||
|
||||
result = db.optimize_fts_storage(vacuum=False)
|
||||
assert result["ok"] is True
|
||||
n_msg = db._conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0]
|
||||
n_fts = db._conn.execute(
|
||||
"SELECT COUNT(*) FROM messages_fts_docsize"
|
||||
).fetchone()[0]
|
||||
# Exactly one index entry per message: no replay duplicates.
|
||||
assert n_fts == n_msg
|
||||
assert len(db.search_messages("deployment")) == 1
|
||||
db._conn.execute(
|
||||
"INSERT INTO messages_fts(messages_fts, rank) VALUES('integrity-check', 1)"
|
||||
)
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
def test_repair_bookkeeping_reseeds_missing_progress(self, tmp_path):
|
||||
"""Unit: high_water without progress gets progress='0' without
|
||||
forcing a full marker reset when a real backfill is already claimed."""
|
||||
db = SessionDB(db_path=tmp_path / "fresh.db")
|
||||
try:
|
||||
db.create_session(session_id="s1", source="cli")
|
||||
db.append_message("s1", role="user", content="bookkeeping needle")
|
||||
db.set_meta("fts_rebuild_high_water", "42")
|
||||
db._conn.execute(
|
||||
"DELETE FROM state_meta WHERE key = ?", ("fts_rebuild_progress",)
|
||||
)
|
||||
db._conn.commit()
|
||||
db._repair_optimize_bookkeeping()
|
||||
assert db.get_meta("fts_rebuild_high_water") == "42"
|
||||
assert db.get_meta("fts_rebuild_progress") == "0"
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
def test_demote_writes_markers_before_empty_schema(self, tmp_path):
|
||||
"""Demote must commit rebuild markers before createscript builds the
|
||||
empty v23 tables — so a crash between stage and ensure still leaves
|
||||
a resumable claim rather than an unmarked empty index."""
|
||||
db_path = tmp_path / "v22.db"
|
||||
self._build_v22_db(db_path)
|
||||
db = SessionDB(db_path=db_path)
|
||||
try:
|
||||
# Patch ensure to fail *after* the staged write commits, simulating
|
||||
# death mid schema-create. Markers must already be durable.
|
||||
orig_ensure = db._ensure_fts_schema
|
||||
calls = {"n": 0}
|
||||
|
||||
def boom(cursor, table_name, ddl):
|
||||
calls["n"] += 1
|
||||
if table_name == "messages_fts":
|
||||
# Markers must already be on disk from the staged write.
|
||||
row = db._conn.execute(
|
||||
"SELECT value FROM state_meta "
|
||||
"WHERE key = 'fts_rebuild_high_water'"
|
||||
).fetchone()
|
||||
assert row is not None, (
|
||||
"markers must be committed before empty v23 schema create"
|
||||
)
|
||||
progress = db._conn.execute(
|
||||
"SELECT value FROM state_meta "
|
||||
"WHERE key = 'fts_rebuild_progress'"
|
||||
).fetchone()
|
||||
assert progress is not None and progress[0] == "0"
|
||||
raise sqlite3.OperationalError("simulated crash mid-ensure")
|
||||
return orig_ensure(cursor, table_name, ddl)
|
||||
|
||||
db._ensure_fts_schema = boom # type: ignore[method-assign]
|
||||
try:
|
||||
db._demote_legacy_fts_to_trash()
|
||||
raise AssertionError("demote should have raised")
|
||||
except sqlite3.OperationalError as exc:
|
||||
assert "simulated crash" in str(exc)
|
||||
|
||||
# Staged demote survived: markers + trash, no successful stamp.
|
||||
assert db.get_meta("fts_rebuild_high_water") is not None
|
||||
assert db.get_meta("fts_rebuild_progress") == "0"
|
||||
assert db._has_fts_trash(db._conn) is True
|
||||
assert db.get_meta("fts_storage_version") is None
|
||||
|
||||
# Restore ensure and resume — full optimize completes.
|
||||
db._ensure_fts_schema = orig_ensure # type: ignore[method-assign]
|
||||
result = db.optimize_fts_storage(vacuum=False)
|
||||
assert result["ok"] is True
|
||||
assert len(db.search_messages("deployment")) == 1
|
||||
assert db.fts_optimize_available() is False
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
def test_optimize_settle_refuses_pending_backfill(self, tmp_path):
|
||||
"""Settle must not stamp while high_water markers remain."""
|
||||
db = SessionDB(db_path=tmp_path / "fresh.db")
|
||||
try:
|
||||
db.create_session(session_id="s1", source="cli")
|
||||
db.append_message("s1", role="user", content="settle guard needle")
|
||||
# Plant markers without going through demote.
|
||||
db.set_meta("fts_rebuild_high_water", "1")
|
||||
db.set_meta("fts_rebuild_progress", "0")
|
||||
# The public contract: optimize returns ok=False when still
|
||||
# pending. Simulate an unfinishable backfill by stubbing the
|
||||
# chunk step to a no-op while markers stay.
|
||||
db.fts_rebuild_step = lambda: False # type: ignore[method-assign]
|
||||
result = db.optimize_fts_storage(vacuum=False)
|
||||
assert result["ok"] is False
|
||||
assert result.get("reason") == "backfill_incomplete"
|
||||
assert db.get_meta("fts_storage_version") is None
|
||||
assert db.get_meta("fts_rebuild_high_water") is not None
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
def test_v23_fresh_db_born_optimized(self, tmp_path):
|
||||
"""A brand-new DB is born on v23 — no legacy layout, no opt-in flag,
|
||||
no pending rebuild."""
|
||||
|
|
|
|||
Loading…
Reference in New Issue