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()