From 9c07d1048d3ea44b3bb5be66404ffaf213113b9c Mon Sep 17 00:00:00 2001 From: Hasnain Ibrar Date: Fri, 18 Sep 2026 02:20:09 +0300 Subject: [PATCH] fix(index): recover an index an incompatible schema wrote An index whose generations were written by an earlier index schema could not be used or discarded. `clear`, reindexing, and `status` all read the active snapshot first, and that read validates every generation manifest, so each one failed with the same error: IndexSchemaError: Manifest for generation ... is invalid. index_schema_version Input should be 7 [input_value=6] Recovery meant editing repository files by hand or importing the media into another repository. Add a supported recovery path: - `read_active_document` reads the active snapshot without validating the generations it points at. The snapshot document is still checked for integrity and identity; only the manifests are left unread, because a reset is about to discard them. - `describe_active_generations` reports what a reset would give up, so a caller can tell the user before anything is destroyed. - `discard_generations` publishes an empty snapshot whatever the generations hold, leaving retained generation directories on disk as `clear` does. - `IndexStorage.drop_all_collections` deletes every collection in the store. Publishing an empty snapshot is not enough on its own: a collection keeps the embedding dimensions it was created with, and an incompatible store can hold collections this configuration no longer names. - `vidxp index reset` exposes this, and the incompatible-schema error now names that command instead of leaving the user without a next step. Collections are dropped before the metadata is replaced. A failure part way through then leaves the repository exactly as it was, so the command can be retried and still report what it removed. Imported media is never touched, so indexing it again restores search. `status` was blocked as well, which the report did not mention, so an affected repository could not even describe its own state. Co-Authored-By: Claude Opus 5 (1M context) --- INSTALLATION_GUIDE.md | 17 ++ src/vidxp/application.py | 21 ++ src/vidxp/application_boundary.py | 5 +- src/vidxp/application_models.py | 29 ++ src/vidxp/cli_commands/index.py | 64 ++++ src/vidxp/core/storage.py | 18 ++ src/vidxp/infrastructure/local_index.py | 38 +++ src/vidxp/infrastructure/local_snapshots.py | 76 ++++- src/vidxp/ports.py | 2 + tests/test_index_reset.py | 313 ++++++++++++++++++++ 10 files changed, 581 insertions(+), 2 deletions(-) create mode 100644 tests/test_index_reset.py diff --git a/INSTALLATION_GUIDE.md b/INSTALLATION_GUIDE.md index 27b4f59b..0b3682c2 100644 --- a/INSTALLATION_GUIDE.md +++ b/INSTALLATION_GUIDE.md @@ -270,6 +270,23 @@ use the same indexing settings. Bulk indexing does not migrate an existing multi-video index to a different profile or remove old results to make room for one. Such jobs report the existing profile-compatibility error. +### Rebuild an index VidXP can no longer use + +If an upgrade changes how indexes are stored, VidXP reports that the existing +index is incompatible. Indexing, clearing, and even reading the index status +fail while that data is in place. Discard the generated index data and start +over: + +```bash +vidxp index reset --yes +vidxp index bulk --all --modality scene +``` + +Resetting deletes every generated index in the repository, including its +vector collections, and reports which videos it gave up. Imported media is +preserved, so the second command rebuilds search from the videos already in +the repository. Nothing has to be imported again. + ### Start an installed interface | Interface | Command | diff --git a/src/vidxp/application.py b/src/vidxp/application.py index 71b40d2d..c390385b 100644 --- a/src/vidxp/application.py +++ b/src/vidxp/application.py @@ -34,6 +34,7 @@ PrepareModelsCommand, PrepareModelsResult, RemoveIndexCommand, + IndexResetResult, ResourceNotFoundError, RuntimeReadiness, QueryAnswer, @@ -938,6 +939,26 @@ def clear_index(self) -> bool: self.registry.install_hint(self.registry.index_names()), ) from exc + @application_boundary + def reset_index(self) -> IndexResetResult: + """Discard generated index data so the repository can be rebuilt. + + Use this when an index cannot be cleared or rebuilt because its + generations were written by an incompatible index schema. Imported + media is preserved, so indexing it again restores search. + """ + + base_config = self._base_config() + try: + return IndexResetResult.model_validate( + self.index_backend.reset(base_config) + ) + except ModuleNotFoundError as exc: + raise DependencyUnavailableError( + self.registry.index_names(), + self.registry.install_hint(self.registry.index_names()), + ) from exc + @application_boundary def remove_from_index(self, command: RemoveIndexCommand) -> bool: return self.index_backend.remove( diff --git a/src/vidxp/application_boundary.py b/src/vidxp/application_boundary.py index 3ce724c9..5a64d30f 100644 --- a/src/vidxp/application_boundary.py +++ b/src/vidxp/application_boundary.py @@ -97,7 +97,10 @@ def wrapped(*args: Any, **kwargs: Any) -> Any: raise ApplicationError( "index_schema_incompatible", ErrorCategory.conflict, - "The index schema is incompatible with this version.", + "The index schema is incompatible with this version. " + "Discard the generated index data with " + "`vidxp index reset --yes`, then index the media again. " + "Imported media is preserved.", ) from exc except IndexStorageUnavailableError as exc: raise ApplicationError( diff --git a/src/vidxp/application_models.py b/src/vidxp/application_models.py index 257edcf8..cd260430 100644 --- a/src/vidxp/application_models.py +++ b/src/vidxp/application_models.py @@ -725,6 +725,35 @@ def skipped(self) -> tuple[BulkIndexTarget, ...]: ) +class IndexResetResult(ApplicationModel): + """What discarding generated index data gave up. + + Imported media is never part of a reset, so it is not reported here. + """ + + discarded: bool = Field( + description=( + "Whether an active snapshot with generated data was replaced." + ) + ) + media_ids: tuple[MediaId, ...] = Field( + default=(), + description="Media that must be indexed again to become searchable.", + ) + generation_ids: tuple[IndexGenerationId, ...] = Field( + default=(), + description="Generations the discarded snapshot referenced.", + ) + modalities: tuple[str, ...] = Field( + default=(), + description="Capabilities the discarded snapshot had indexed.", + ) + collections: tuple[str, ...] = Field( + default=(), + description="Vector collections deleted so they can be rebuilt.", + ) + + class RemoveIndexCommand(ApplicationModel): media_id: MediaId diff --git a/src/vidxp/cli_commands/index.py b/src/vidxp/cli_commands/index.py index d51a1593..96415bee 100644 --- a/src/vidxp/cli_commands/index.py +++ b/src/vidxp/cli_commands/index.py @@ -466,6 +466,70 @@ def index_list( ) +@app.command("reset") +def index_reset( + ctx: typer.Context, + yes: Annotated[ + bool, + typer.Option("--yes", "-y", help="Skip the confirmation prompt."), + ] = False, + json_output: Annotated[ + bool, + typer.Option("--json", help="Emit machine-readable JSON."), + ] = False, +) -> None: + """Discard generated index data so an unusable index can be rebuilt. + + Use this when indexing or clearing fails because the existing index was + written by an incompatible index schema. Imported media is preserved, so + indexing it again restores search. + """ + + state = state_from_context(ctx) + output_format = effective_output_format(state, json_output) + if not yes: + typer.secho( + "Resetting discards every generated index in this repository, " + "including its vector collections.", + fg=typer.colors.YELLOW, + ) + typer.echo( + "Imported media is preserved. Indexed media has to be indexed " + "again before it can be searched." + ) + typer.confirm( + f"Discard the generated index data at " + f"{state.service.index_directory}?", + abort=True, + ) + result = state.service.reset_index() + payload = result.model_dump(mode="json") + if output_format == OutputFormat.json: + emit_json(payload) + return + if not result.discarded and not result.collections: + typer.echo("No generated index data was found.") + return + if result.media_ids: + table = Table(title="Discarded index data") + table.add_column("Media ID") + table.add_column("Generation") + for media_id, generation_id in zip( + result.media_ids, + result.generation_ids, + ): + table.add_row(media_id, generation_id) + Console().print(table) + typer.secho("Generated index data discarded.", fg=typer.colors.GREEN) + typer.echo( + f"Media needing indexing again: {len(result.media_ids)}; " + f"capabilities affected: {', '.join(result.modalities) or 'none'}; " + f"vector collections deleted: " + f"{', '.join(result.collections) or 'none'}." + ) + typer.echo("Rebuild with: vidxp index bulk") + + @app.command("clear") def index_clear( ctx: typer.Context, diff --git a/src/vidxp/core/storage.py b/src/vidxp/core/storage.py index 6dfd9d5f..917c5a35 100644 --- a/src/vidxp/core/storage.py +++ b/src/vidxp/core/storage.py @@ -231,6 +231,24 @@ def clear(self, modalities: Iterable[str] | None = None) -> None: self._call(self.client.delete_collection, name) self._collections.pop(modality, None) + def drop_all_collections(self) -> tuple[str, ...]: + """Delete every collection in this store and report what was dropped. + + Recovery has to remove collections this configuration does not name. + An incompatible store can hold collections written by another index + profile, and a collection keeps the embedding dimensions it was + created with, so reusing one after a schema change fails. + """ + + dropped = sorted( + getattr(collection, "name", collection) + for collection in self._call(self.client.list_collections) + ) + for name in dropped: + self._call(self.client.delete_collection, name) + self._collections.clear() + return tuple(dropped) + def delete_video(self, modality: str, video_id: str) -> None: self._call( self.collection(modality).delete, diff --git a/src/vidxp/infrastructure/local_index.py b/src/vidxp/infrastructure/local_index.py index 982e4199..0d127f33 100644 --- a/src/vidxp/infrastructure/local_index.py +++ b/src/vidxp/infrastructure/local_index.py @@ -583,3 +583,41 @@ def clear(self, config: IndexConfig) -> bool: repository = self.repository with repository.lease(): return repository.clear() + + def reset(self, config: IndexConfig) -> dict[str, object]: + """Discard generated index data so the repository can be rebuilt. + + Clearing validates the active generations first, so it cannot recover + a repository whose generations were written by an incompatible index + schema. Reset reads the snapshot document without those generations + and drops the vector collections as well, because a collection keeps + the embedding dimensions it was created with. + + Imported media is untouched. Retained generation directories are left + on disk, as `clear` leaves them. + """ + + self._require_index_directory(config.index_directory) + repository = self.repository + with repository.lease(): + summary = repository.describe_active_generations() + dropped: tuple[str, ...] = () + store_config = replace( + config, + storage_directory=repository.store, + ) + # Drop the vector collections before replacing the metadata. A + # failure here then leaves the repository exactly as it was, so + # the operation can be retried and still report what it removed. + if store_config.index_directory.is_dir(): + # Opened for writing rather than through the committed-store + # helper: that helper rejects a store holding no collections, + # and a repository with nothing left to drop still has to be + # recoverable. + with IndexStorage( + store_config, + client_factory=self.chroma_clients, + ) as storage: + dropped = storage.drop_all_collections() + repository.discard_generations() + return {**summary, "collections": dropped} diff --git a/src/vidxp/infrastructure/local_snapshots.py b/src/vidxp/infrastructure/local_snapshots.py index e364f27f..4bc1cd71 100644 --- a/src/vidxp/infrastructure/local_snapshots.py +++ b/src/vidxp/infrastructure/local_snapshots.py @@ -120,10 +120,24 @@ def read_active(self, *, required: bool = False) -> IndexSnapshot | None: resolved = self._read_active(required=required) return None if resolved is None else resolved[1] + def read_active_document( + self, + ) -> tuple[ActiveSnapshotPointer, IndexSnapshot] | None: + """Read the active snapshot without validating the generations. + + The snapshot document is still checked for integrity and identity. + Only the generation manifests it points at are left unread, so a + repository whose generations were written by an incompatible index + schema can still be inspected and discarded. + """ + + return self._read_active(validate_generations=False) + def _read_active( self, *, required: bool = False, + validate_generations: bool = True, ) -> tuple[ActiveSnapshotPointer, IndexSnapshot] | None: if not self.active_pointer.is_file(): if required: @@ -138,6 +152,7 @@ def _read_active( snapshot = self.read_snapshot( pointer.snapshot_id, expected_sha256=pointer.snapshot_sha256, + validate_generations=validate_generations, ) return pointer, snapshot except IndexSchemaError: @@ -152,6 +167,7 @@ def read_snapshot( snapshot_id: str, *, expected_sha256: str | None = None, + validate_generations: bool = True, ) -> IndexSnapshot: snapshot_path = self._snapshot_path(snapshot_id) if not snapshot_path.is_file(): @@ -177,7 +193,8 @@ def read_snapshot( raise IndexSchemaError( "The snapshot filename and document identifier differ." ) - self._validate_generations(snapshot) + if validate_generations: + self._validate_generations(snapshot) return snapshot def _validate_generations(self, snapshot: IndexSnapshot) -> None: @@ -348,6 +365,63 @@ def clear(self) -> bool: ) return True + def describe_active_generations(self) -> dict[str, Any]: + """Report what the active snapshot holds without validating it. + + Read-only, so a caller can tell the user exactly which generated data + a reset would give up before anything is destroyed. + """ + + resolved = self.read_active_document() + if resolved is None: + return { + "discarded": False, + "media_ids": (), + "generation_ids": (), + "modalities": (), + } + _, snapshot = resolved + media_ids = tuple(sorted(snapshot.generations)) + return { + "discarded": bool(media_ids), + "media_ids": media_ids, + "generation_ids": tuple( + snapshot.generations[media_id].generation_id + for media_id in media_ids + ), + "modalities": tuple( + sorted( + { + modality + for reference in snapshot.generations.values() + for modality in reference.modalities + } + ) + ), + } + + def discard_generations(self) -> bool: + """Publish an empty active snapshot, whatever the generations hold. + + Unlike `clear`, this does not validate the generations it replaces, so + it can recover a repository whose generations were written by an + incompatible index schema. Retained generation directories are left on + disk, exactly as `clear` leaves them. + """ + + resolved = self.read_active_document() + if resolved is None: + return False + _, snapshot = resolved + if not snapshot.generations: + return False + self._publish( + generations={}, + config_fingerprint=snapshot.config_fingerprint, + configuration=dict(snapshot.configuration), + ) + return True + def _publish( self, *, diff --git a/src/vidxp/ports.py b/src/vidxp/ports.py index 492636ae..2fb344bb 100644 --- a/src/vidxp/ports.py +++ b/src/vidxp/ports.py @@ -430,6 +430,8 @@ def remove(self, config: IndexConfig, media_id: str) -> bool: ... def clear(self, config: IndexConfig) -> bool: ... + def reset(self, config: IndexConfig) -> dict[str, Any]: ... + class JobBackend(Protocol): """Durable lifecycle operations owned by the workflow engine.""" diff --git a/tests/test_index_reset.py b/tests/test_index_reset.py new file mode 100644 index 00000000..d4c89412 --- /dev/null +++ b/tests/test_index_reset.py @@ -0,0 +1,313 @@ +import unittest +from dataclasses import replace +from datetime import datetime, timezone +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import Mock + +from vidxp.core.contracts import ( + INDEX_SCHEMA_VERSION, + MANIFEST_SCHEMA_VERSION, + IndexConfig, + IndexSchemaError, +) +from vidxp.core.manifest import MANIFEST_FILE, sha256_file, write_json_atomic +from vidxp.core.snapshots import GenerationReference +from vidxp.core.storage import IndexStorage +from vidxp.infrastructure.local_snapshots import LocalSnapshotRepository + + +MEDIA_ID = "123456781234423481234567890abcde" +OTHER_MEDIA_ID = "223456781234423481234567890abcde" +INPUT_SHA = "b" * 64 +INCOMPATIBLE_SCHEMA = INDEX_SCHEMA_VERSION - 1 + + +class StaleRepositoryTests(unittest.TestCase): + """Recovery after an intentional index schema change (#167).""" + + def setUp(self): + self.temporary = TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.repository = LocalSnapshotRepository( + Path(self.temporary.name) / "indexes" + ) + self.repository.ensure_directories() + self.config = IndexConfig.local( + enabled_modalities=("scene",), + collection_names={"scene": "scene"}, + storage_directory=self.repository.store, + ) + + def write_generation( + self, + media_id: str, + *, + index_schema_version: int, + ) -> GenerationReference: + generation_id = self.repository.new_generation_id() + config = replace( + self.config, + video_id=media_id, + generation_id=generation_id, + generation_directory=self.repository.generation_directory( + generation_id + ), + ) + now = datetime.now(timezone.utc).isoformat() + manifest = { + "manifest_schema_version": MANIFEST_SCHEMA_VERSION, + "index_schema_version": index_schema_version, + "dataset": config.dataset, + "split": config.split, + "run_id": config.run_id, + "generation_id": generation_id, + "state": "complete", + "created_at": now, + "updated_at": now, + "completed_at": now, + "config_fingerprint": config.fingerprint(), + "execution_fingerprint": "e" * 64, + "configuration": config.to_dict(), + "models": {}, + "git": {}, + "environment": {}, + "inputs": { + media_id: { + "sha256": INPUT_SHA, + "checksums": {"video": INPUT_SHA}, + "size": 1, + "source_name": f"{media_id}.mp4", + "path": None, + "metadata": {}, + } + }, + "videos": { + media_id: { + "state": "complete", + "started_at": now, + "stages": {}, + "completed_at": now, + "summary": {}, + } + }, + "completed_videos": [media_id], + "failed_videos": [], + "interrupted_videos": [], + "processed_frames": 0, + "record_counts": {"scene": 1}, + "store_size_bytes_at_commit": 123, + } + manifest_path = config.run_directory / MANIFEST_FILE + write_json_atomic(manifest_path, manifest) + return GenerationReference( + generation_id=generation_id, + media_id=media_id, + manifest_sha256=sha256_file(manifest_path), + input_sha256=INPUT_SHA, + config_fingerprint=config.fingerprint(), + modalities=("scene",), + record_counts={"scene": 1}, + store_size_bytes_at_commit=123, + ) + + def publish_stale_snapshot(self, *media_ids: str) -> None: + """Publish a snapshot whose generations an older schema wrote.""" + + generations = { + media_id: self.write_generation( + media_id, + index_schema_version=INCOMPATIBLE_SCHEMA, + ) + for media_id in media_ids + } + self.repository._publish( + generations=generations, + config_fingerprint=self.config.fingerprint(), + configuration=self.repository.snapshot_configuration(self.config), + ) + + def test_an_incompatible_index_cannot_be_read_cleared_or_inspected(self): + self.publish_stale_snapshot(MEDIA_ID) + + for operation in ( + self.repository.read_active, + self.repository.clear, + self.repository.status, + ): + with self.subTest(operation=operation.__name__): + with self.assertRaises(IndexSchemaError): + operation() + + def test_the_snapshot_document_is_still_readable(self): + self.publish_stale_snapshot(MEDIA_ID) + + resolved = self.repository.read_active_document() + + self.assertIsNotNone(resolved) + assert resolved is not None + _pointer, snapshot = resolved + self.assertEqual(tuple(snapshot.generations), (MEDIA_ID,)) + + def test_describing_the_active_generations_changes_nothing(self): + self.publish_stale_snapshot(MEDIA_ID, OTHER_MEDIA_ID) + + summary = self.repository.describe_active_generations() + again = self.repository.describe_active_generations() + + self.assertTrue(summary["discarded"]) + self.assertEqual( + summary["media_ids"], + tuple(sorted((MEDIA_ID, OTHER_MEDIA_ID))), + ) + self.assertEqual(summary["modalities"], ("scene",)) + self.assertEqual(len(summary["generation_ids"]), 2) + self.assertEqual(summary, again) + + def test_discarding_recovers_a_repository_an_older_schema_wrote(self): + self.publish_stale_snapshot(MEDIA_ID) + + discarded = self.repository.discard_generations() + + self.assertTrue(discarded) + active = self.repository.read_active() + self.assertIsNotNone(active) + assert active is not None + self.assertEqual(active.generations, {}) + self.assertEqual(self.repository.status()["state"], "empty") + + def test_discarding_an_empty_repository_reports_no_work(self): + self.assertFalse(self.repository.discard_generations()) + self.assertEqual( + self.repository.describe_active_generations()["discarded"], + False, + ) + + def test_a_compatible_index_is_unaffected_by_the_new_read_path(self): + reference = self.write_generation( + MEDIA_ID, + index_schema_version=INDEX_SCHEMA_VERSION, + ) + self.repository._publish( + generations={MEDIA_ID: reference}, + config_fingerprint=self.config.fingerprint(), + configuration=self.repository.snapshot_configuration(self.config), + ) + + validated = self.repository.read_active() + document = self.repository.read_active_document() + + self.assertIsNotNone(validated) + self.assertIsNotNone(document) + assert validated is not None and document is not None + self.assertEqual(validated.snapshot_id, document[1].snapshot_id) + + +class DropAllCollectionsTests(unittest.TestCase): + def storage(self, names: tuple[str, ...]) -> IndexStorage: + with TemporaryDirectory() as directory: + config = IndexConfig.local( + enabled_modalities=("scene",), + collection_names={"scene": "scene"}, + storage_directory=Path(directory), + ) + storage = IndexStorage.__new__(IndexStorage) + storage.config = config + storage._names = dict(config.collection_names) + storage._collections = {"scene": object()} + storage.client = Mock() + storage.client.list_collections.return_value = [ + Mock(name=name) for name in names + ] + for collection, name in zip( + storage.client.list_collections.return_value, + names, + ): + collection.name = name + return storage + + def test_every_collection_is_dropped_even_when_unnamed_by_the_config(self): + storage = self.storage(("scene", "retired_profile")) + + dropped = storage.drop_all_collections() + + self.assertEqual(dropped, ("retired_profile", "scene")) + self.assertEqual( + sorted( + call.args[0] + for call in storage.client.delete_collection.call_args_list + ), + ["retired_profile", "scene"], + ) + self.assertEqual(storage._collections, {}) + + def test_an_empty_store_drops_nothing(self): + storage = self.storage(()) + + self.assertEqual(storage.drop_all_collections(), ()) + storage.client.delete_collection.assert_not_called() + + +class ResetIndexOperationTests(unittest.TestCase): + def application(self, root: str | Path, *, backend: Mock): + from vidxp.application import VidXPApplication + from vidxp.capabilities.registry import create_capability_registry + from vidxp.repository_layout import RepositoryLayout + from vidxp.runtime import ModelRuntime + from vidxp.settings import VidXPSettings + + settings = VidXPSettings( + repository_root=Path(root), + runtime_backend="cpu", + ) + return VidXPApplication( + settings=settings, + layout=RepositoryLayout(root=Path(root)), + registry=create_capability_registry(), + runtime=ModelRuntime(settings), + index_backend=backend, + media=Mock(), + artifacts=Mock(), + index_status=lambda: None, + ) + + def test_the_backend_summary_becomes_a_typed_result(self): + backend = Mock() + backend.reset.return_value = { + "discarded": True, + "media_ids": (MEDIA_ID,), + "generation_ids": ("323456781234423481234567890abcde",), + "modalities": ("scene",), + "collections": ("scene",), + } + with TemporaryDirectory() as root: + application = self.application(root, backend=backend) + + result = application.reset_index() + + self.assertTrue(result.discarded) + self.assertEqual(result.media_ids, (MEDIA_ID,)) + self.assertEqual(result.modalities, ("scene",)) + self.assertEqual(result.collections, ("scene",)) + + def test_an_untouched_repository_reports_nothing_discarded(self): + backend = Mock() + backend.reset.return_value = { + "discarded": False, + "media_ids": (), + "generation_ids": (), + "modalities": (), + "collections": (), + } + with TemporaryDirectory() as root: + application = self.application(root, backend=backend) + + result = application.reset_index() + + self.assertFalse(result.discarded) + self.assertEqual(result.media_ids, ()) + self.assertEqual(result.collections, ()) + + +if __name__ == "__main__": + unittest.main()