diff --git a/docs/README.md b/docs/README.md index 8efdbed7..7d53df51 100644 --- a/docs/README.md +++ b/docs/README.md @@ -9,7 +9,7 @@ contracts (kernel purity, headless annotator) are described there and enforced i | [workspaces.md](workspaces.md) | The workspace on disk: layout, `init`/`open`, project-name uniqueness, and how services are composed | | [projects.md](projects.md) | The project lifecycle: the 1:1 dataset, renaming, and what deletion does and does not destroy | | [sources.md](sources.md) | Where raw data comes from: the two registration methods, what a video source records from the probe, why decomposition parameters live on the source, and the idempotency rule and its named uniqueness gap | -| [ingest.md](ingest.md) | Turning a source into rows: content identity versus recorded origin, the two source paths, why the decode happens outside a transaction, and the per-file report | +| [ingest.md](ingest.md) | Turning a source into rows: content identity versus recorded origin, the two source paths, why the decode happens outside a transaction, the run's lifecycle and pollable progress, and the per-file report | | [schemas.md](schemas.md) | The annotation schema: immutable monotonic versions, additive vs destructive change, and the two gates on narrowing | | [batches.md](batches.md) | The unit of annotation work: the state machine, membership frozen at approval, the schema pin, and the exact partition into jobs | | [jobs.md](jobs.md) | Annotation jobs: the job and per-asset progress machines, what counts as settled, ordered `next_pending`, and derived progress | diff --git a/docs/ingest.md b/docs/ingest.md index 5a44773b..af6c5e65 100644 --- a/docs/ingest.md +++ b/docs/ingest.md @@ -64,6 +64,77 @@ offer is a policy the kernel would be inventing. tests. Decoding each one again to re-confirm it would also route our own encoder's output into an operator's per-file report — a failure nobody could act on. +## The run has a lifecycle, and it is a table + +`INGEST_TRANSITIONS` in `domain/ingest.py` is the whole of what is legal. `IngestService` +consults it through `require_move`; nothing restates it. + +``` +pending ──▶ running ──▶ completed + │ │ + └────────▶ failed ──▶ running (resume) +``` + +A job is created `pending` and moved to `running` by whoever picks the work up. Today that is +the same call, and the state is over in microseconds — it is spelled out anyway because it is +the vocabulary a queue needs, and adding it later would mean changing what a stored row means. + +**`failed → running` is the only backward edge in this kernel**, and the argument against +reopening a [batch](batches.md) does not carry over. A batch pins a schema version at approval +and its jobs are already cut against that pin, so un-freezing one would invalidate work already +done. Nothing is pinned against an ingest run. It is a record of work, not an artifact with +dependents — so resuming is the same unit of work continuing, on the same row. A second row per +attempt would fork `batch_id` and turn `IngestService.list` into a list of retries. + +**`running → running` is deliberately missing**, so a run stuck at `running` cannot be resumed. +That state is a process that died without reporting anything, not a failure anybody can read. +The remedy already exists — ingest the source again, which creates nothing — and it leaves the +stuck row as the only evidence the crash left. + +## Progress a caller can poll + +`processed` and `total` are written to the row **as the run goes**, so +`IngestService.get(job_id)` answers "where is it now" rather than "where did it end". That is +the contract the HTTP API and the UI will reuse; nothing about it is specific to being in the +same process. + +| | what it means | +| --- | --- | +| `processed` | items dealt with — decoded and stored, or reported as unreadable | +| `total` | items the source offered, or **NULL** when that is not knowable up front | + +A directory can be listed, so it states its total before the first file; an empty one records +`0 of 0` rather than nothing. A clip cannot: `VideoMetadata` carries no frame count by design — +it would be a guess for a variable-rate clip, and the number an ingest wants is what extraction +actually produced — so `total` stays NULL and `processed` climbs alone. + +The counter is written **once per item**, not on a cadence. An interval that suits five files +and one that suits fifty thousand are different numbers, and this service cannot know which it +is looking at; the cost of not choosing is one small commit beside a decode and a hash that +cost an order of magnitude more. + +## Resuming a failed run + +`IngestService.resume(job_id)` re-runs a failed job on its own row, into the batch the first +attempt was headed for. What qualifies is whatever the table says can reach `running` — `failed`, +and also `pending`, which a synchronous run never leaves behind but a queued one would. A +`completed` or `running` job is refused with an ordinary `InvalidTransition` rather than an error +of its own. + +It is a **redo, not a skip**. There is no per-file record of what the previous attempt managed, +and there does not need to be: blobs are content-addressed and assets are deduplicated by +content, so re-reading the whole source creates nothing it created before. The cost is +re-hashing what is already stored; what it buys is that resume has no second code path to get +wrong. + +The counters, the per-file report and the fatal `error` are reset when the attempt starts, so +they describe the run somebody is watching rather than the one that failed. A run that *failed* +keeps them exactly where they stopped, which is the first thing anyone reading a failure wants. + +`batch_name` is a column for this reason alone: a run that died during the decode reached no +batch, so without it a resumed run would fall back to naming the batch after the source folder +and quietly lose the name the caller asked for. + ## Four transactions, and the middle of the run is in none of them 1. Resolve the source, decide the target batch, and insert the `IngestJob` as `running`. @@ -81,9 +152,14 @@ row exists: `BlobStore.put` is not transactional and a rollback cannot unwrite i nothing points at is harmless (content-addressed, shared, never deleted), while a row naming bytes that were never stored is not. +The progress writes that happen *between* items are not a contradiction. Each is one `UPDATE` +that opens and commits while nothing is being decoded; what the single-writer warning is about +is a transaction held **across** the decode, not the existence of writes during that phase. + The honest consequence, stated rather than hidden: a process killed between transactions can leave -assets in the project with no batch and a job stuck at `running`. That is recoverable, and finding -it is what the job record is for. +assets in the project with no batch and a job stuck at `running`. Finding it is what the job +record is for, and ingesting the source again is what fixes it — see the lifecycle above for why +that is the remedy rather than a resume. Within a run, each file is **probed before it is stored**, so a file that is going to be refused never leaves a blob behind. @@ -97,9 +173,14 @@ name and the reason are kept apart so a report renders as a table rather than as sentences, and `IngestFailureKind` exists so it can be **grouped** — real data loss must not be buried under ordinary operator noise. +The report is on the row as well as in the return value, written as the run goes rather than at +the end: a report that only appeared once the run finished would be invisible for exactly as +long as it is interesting. + A missing ffmpeg is not a file's fault at all. `MediaToolUnavailable` is recorded as the job's -`error`, the job is marked `failed`, and it is re-raised — which is precisely why it sits outside -the `MediaError` family. One broken machine is not five thousand broken files. +`error` — a separate column from the per-file `failures`, which stays empty — the job is marked +`failed`, and it is re-raised, which is precisely why it sits outside the `MediaError` family. +One broken machine is not five thousand broken files. ## The target batch @@ -113,11 +194,10 @@ order for a directory and frame order for a clip. ## What is deliberately not here yet -- **No state machine, no persisted progress, no persisted report.** The job is written straight to - `completed` or `failed`, and `IngestResult` lives only in memory. The transition table, the - processed/total counters a caller can poll mid-run, and the report as columns are #19's, which - adds them *around* this working path rather than rewriting it. - **No thumbnails.** Generating one per asset at ingest and recording `asset.thumbnail_hash` is #21, for the M5 gallery. - **No background execution.** A run is synchronous and in-process. The service API is shaped so - that moving it behind a queue changes the caller's waiting, not its vocabulary. + that moving it behind a queue changes the caller's waiting, not its vocabulary — which is why + a job is created `pending` and why progress is read off the row rather than off a callback. +- **No cross-attempt history.** The report and the counters describe the current attempt. A + resumed run overwrites them, and a log of every attempt would be its own table. diff --git a/docs/persistence.md b/docs/persistence.md index 4567a143..aecf20f0 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -115,8 +115,9 @@ MIGRATIONS: list[Migration] = [ Migration(version=6, name="release_manifest_pointer", upgrade=...), Migration(version=7, name="source_provenance", upgrade=...), Migration(version=8, name="ingest_pipeline", upgrade=...), + Migration(version=9, name="ingest_job_progress", upgrade=...), ] -FORMAT_VERSION: int = MIGRATIONS[-1].version # 8 +FORMAT_VERSION: int = MIGRATIONS[-1].version # 9 ``` `initialize()` reads the version stamped in `_visionset_meta` and runs whatever is @@ -167,6 +168,15 @@ at 7 and will never re-run it. Declared anywhere else, the `create_all` path and the `ALTER` path emit different `CREATE TABLE` text and the fresh-versus-migrated test fails — which is exactly what it is for. +Migration 009 is where that expiry actually bit. `ingest_job` was rebuilt by 008, so its column +order was free *then*; by 009 the table holds real rows and its four new columns — +`batch_name`, `processed`, `total`, `failures` — arrive by `ALTER` and sit last, in the order +the migration adds them. That path has a test of its own +(`test_migration_nine_alters_a_table_migration_eight_rebuilt`), because the fresh-versus-migrated +test walks back to generation 1, from where 008 re-creates the table whole and 009 finds its +columns already present. A migration whose only exercise is through an earlier rebuild is not +exercised at all. + **A column that carries a foreign key cannot arrive by `ALTER` at all.** SQLite spells an added key inline on the column; `create_all` spells one as a table constraint. The two texts differ, so the fresh-versus-migrated test fails — and dropping the key instead is not free either. @@ -203,10 +213,20 @@ predates the ingest pipeline", where a `server_default` would invent a format no it does refuse is data its two new unique indexes cannot accept, counted before either index is created so an `IntegrityError` never escapes `initialize()`. +Migration 009 is the plainest in the file, and the plainness is the point after 008: four +columns, none carrying a foreign key, so `ALTER` can express all of them — and every one has an +honest value for a row written before it. A pre-#19 run counted nothing and reported nothing, +which is exactly what `0` and `[]` say; NULL is what a run that named no batch meant. So unlike +006, 007 and 008 it refuses nothing and drops nothing. `failures` is a JSON column rather than a +child table on the criteria above: a per-file report is an immutable value read whole, and +nothing queries a single failed file in SQL. + The fresh-versus-migrated test is only as strong as how far back `_downgrade_to_version_one` walks, so every migration added there needs its undo added too. Migrations 006 and 007 are the two places that undo cannot borrow its DDL from `_tables`, -because `_tables` no longer describes the shape it is restoring. +because `_tables` no longer describes the shape it is restoring. Migration 009 is the one +place that needs no undo of its own: its columns live on `ingest_job`, which 008's undo rebuilds +from scratch, so restoring the generation-1 shape removes them along with everything else. `format_version` here is the *database* generation. Validating the on-disk workspace layout around it — directories, the blob-store root, what makes a directory a workspace at all — diff --git a/src/visionset/kernel/adapters/_mappers.py b/src/visionset/kernel/adapters/_mappers.py index 5fe0fa8c..20b70bce 100644 --- a/src/visionset/kernel/adapters/_mappers.py +++ b/src/visionset/kernel/adapters/_mappers.py @@ -7,10 +7,10 @@ fourteen times against fourteen tables. Most entities are flat — every field is a column — and share -``_flat_mapping``. The seven that are not say so explicitly: +``_flat_mapping``. The eight that are not say so explicitly: -- ``AnnotationSchema`` and ``Annotation`` hold immutable nested values, encoded - as JSON. +- ``AnnotationSchema``, ``Annotation`` and ``IngestJob`` hold immutable nested + values, encoded as JSON. - ``Batch`` and ``AnnotationJob`` own child tables, so their mappings carry a ``sync_children`` hook and rebuild their collections on read. - ``DatasetChange``, ``Release`` and ``Source`` encode a timezone-aware @@ -42,7 +42,9 @@ DatasetChange, DatasetMember, Geometry, + IngestFailure, IngestJob, + IngestState, LabelClass, Project, Release, @@ -125,6 +127,37 @@ def _schema_to_domain(_: Session, row: Any) -> AnnotationSchema: ) +def _ingest_job_to_row(entity: IngestJob) -> t.Base: + """Spelled out rather than left to ``_flat_mapping``, which dumps in python + mode and would hand a tuple of ``IngestFailure`` models to a ``JSON`` column. + """ + return t.IngestJobRow( + id=entity.id, + source_id=entity.source_id, + state=entity.state, + error=entity.error, + batch_id=entity.batch_id, + batch_name=entity.batch_name, + processed=entity.processed, + total=entity.total, + failures=[failure.model_dump(mode="json") for failure in entity.failures], + ) + + +def _ingest_job_to_domain(_: Session, row: Any) -> IngestJob: + return IngestJob( + id=row.id, + source_id=row.source_id, + state=IngestState(row.state), + error=row.error, + batch_id=row.batch_id, + batch_name=row.batch_name, + processed=row.processed, + total=row.total, + failures=tuple(IngestFailure.model_validate(f) for f in row.failures), + ) + + def _annotation_to_row(entity: Annotation) -> t.Base: return t.AnnotationRow( id=entity.id, @@ -309,7 +342,6 @@ def _job_sync_children(session: Session, entity: AnnotationJob) -> None: WORKSPACES = _flat_mapping(Workspace, t.WorkspaceRow, None) PROJECTS = _flat_mapping(Project, t.ProjectRow, "workspace_id") -INGEST_JOBS = _flat_mapping(IngestJob, t.IngestJobRow, "source_id") ASSETS = _flat_mapping(Asset, t.AssetRow, "project_id") TASK_GROUPS = _flat_mapping(TaskGroup, t.TaskGroupRow, "batch_id") DATASETS = _flat_mapping(Dataset, t.DatasetRow, "project_id") @@ -327,6 +359,12 @@ def _job_sync_children(session: Session, entity: AnnotationJob) -> None: to_row=_annotation_to_row, to_domain=_annotation_to_domain, ) +INGEST_JOBS: EntityMapping[IngestJob] = EntityMapping( + row=t.IngestJobRow, + parent_column="source_id", + to_row=_ingest_job_to_row, + to_domain=_ingest_job_to_domain, +) SOURCES: EntityMapping[Source] = EntityMapping( row=t.SourceRow, parent_column="project_id", diff --git a/src/visionset/kernel/adapters/_tables.py b/src/visionset/kernel/adapters/_tables.py index 7fce96cb..685f8114 100644 --- a/src/visionset/kernel/adapters/_tables.py +++ b/src/visionset/kernel/adapters/_tables.py @@ -11,8 +11,9 @@ from the asset side, which a JSON blob cannot serve. - Collections that are *immutable value objects* get JSON columns — ``annotation_schema.classes``, ``annotation.geometry``, ``annotation.attributes``, - ``release.split``. A schema version must rehydrate byte-identical, and nothing - ever queries a single ``LabelClass`` by name in SQL. + ``release.split``, ``source.video``, ``ingest_job.failures``. A schema version + must rehydrate byte-identical, and nothing ever queries a single + ``LabelClass`` by name — or a single failed file — in SQL. - A value object too large to belong in a row at all goes in the blob store, and the row keeps its hash — that is ``release.manifest_hash``. The line between the two is size and verifiability, not shape. @@ -157,17 +158,23 @@ class SourceRow(Base): class IngestJobRow(Base): - """Ingestion runs. Rebuilt by migration 8, so column order is free here. - - Rebuilt rather than altered because ``batch_id`` carries a foreign key, and - an ``ALTER TABLE ... ADD COLUMN`` cannot express one the way ``create_all`` - does: SQLite spells an added key inline on the column, while a created table - spells it as a table constraint. The two texts differ, so the fresh-versus- - migrated test would fail — and dropping the key instead would leave a run - pointing at a batch somebody deleted. This table has no children and is - provably empty at that point (nothing wrote an ingest job before this - build), which is what makes the rebuild affordable; migration 8 counts - rather than assuming it. + """Ingestion runs. Rebuilt by migration 8; migration 9 then altered it. + + Migration 8 rebuilt rather than altered because ``batch_id`` carries a + foreign key, and an ``ALTER TABLE ... ADD COLUMN`` cannot express one the way + ``create_all`` does: SQLite spells an added key inline on the column, while a + created table spells it as a table constraint. The two texts differ, so the + fresh-versus-migrated test would fail — and dropping the key instead would + leave a run pointing at a batch somebody deleted. This table had no children + and was provably empty at that point (nothing wrote an ingest job before that + build), which is what made the rebuild affordable; migration 8 counts rather + than assuming it. + + **That exemption expired the moment this build started writing rows.** A + database stamped at 8 will never re-run migration 8, so anything added after + it reaches that database by ``ALTER TABLE`` after all — which is why + migration 9's four columns are declared **last, in the order it adds them**. + A rebuild is no longer available here either: these rows are legitimate. """ __tablename__ = "ingest_job" @@ -177,6 +184,7 @@ class IngestJobRow(Base): SaUuid, ForeignKey("source.id", ondelete="CASCADE"), index=True, nullable=False ) state: Mapped[str] = mapped_column(String, nullable=False) + #: The fatal cause that stopped the run, as opposed to ``failures`` below. error: Mapped[str | None] = mapped_column(String, nullable=True) #: The batch this run materialized into. NULL until it reaches one — a run #: that dies during the decode never does, which is why this is nullable @@ -185,6 +193,23 @@ class IngestJobRow(Base): batch_id: Mapped[UUID | None] = mapped_column( SaUuid, ForeignKey("batch.id", ondelete="SET NULL"), nullable=True ) + #: The name a batch this run creates will take, so a resumed run lands where + #: the first attempt meant it to. Nullable, and honestly so: a row written + #: before migration 9 never recorded one. + batch_name: Mapped[str | None] = mapped_column(String, nullable=True) + #: Items read so far. Carries a ``server_default`` for the reason + #: ``AnnotationJobAssetRow.position`` does — SQLite refuses ``ADD COLUMN`` + #: ``NOT NULL`` without a value for the rows already there — and ``0`` is + #: what a finished pre-#19 run in fact recorded: nothing. + processed: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0")) + #: Items the source offered, or NULL when that is not knowable up front — + #: a directory can be listed, a clip cannot. See ``IngestJob.total``. + total: Mapped[int | None] = mapped_column(Integer, nullable=True) + #: The per-file report: a list of ``IngestFailure``. JSON rather than a child + #: table because it is read whole and never queried by field, the rule + #: ``source.video`` follows. ``server_default`` on + #: ``AnnotationRow.attributes``' terms. + failures: Mapped[list[Any]] = mapped_column(JSON, nullable=False, server_default=text("'[]'")) class AssetRow(Base): diff --git a/src/visionset/kernel/adapters/migrations.py b/src/visionset/kernel/adapters/migrations.py index d2066bf4..d6623ef3 100644 --- a/src/visionset/kernel/adapters/migrations.py +++ b/src/visionset/kernel/adapters/migrations.py @@ -341,6 +341,42 @@ def _add_ingest_origin_and_uniqueness(connection: Connection) -> None: connection.execute(CreateIndex(SOURCE_ORIGIN_UNIQUE, if_not_exists=True)) +def _add_ingest_progress_and_report(connection: Connection) -> None: + """Give a run somewhere to record how far it got, and what it could not read. + + Four plain columns, and the plainness is the point after migration 8: none + of them carries a foreign key, so ``ALTER TABLE`` can express all four and + the rebuild that migration 8 needed is neither available nor wanted here. + Not available, because this table is no longer empty — the build that + shipped migration 8 writes rows to it, and those rows are legitimate. Not + wanted, because nothing about them requires it. + + Idempotent the way 3 to 5 are: the inspector check inside ``_add_column`` + *is* the ``checkfirst``, because migration 1 is ``create_all`` of current + metadata. No data pre-check either, and that is a claim rather than an + oversight: every added column has an honest value for a row written before + it. ``batch_name`` is NULL — that run recorded none, and a resume falls back + to naming the batch after the source, exactly as the first attempt did. + ``processed`` is ``0`` and ``failures`` is ``[]`` — that run counted nothing + and reported nothing, which is true. ``total`` is NULL, which is the same + "not knowable" a video run writes today. + + The two ``NOT NULL`` columns carry their ``server_default`` in ``_tables`` + rather than here, because SQLite refuses ``ADD COLUMN NOT NULL`` without a + value for the rows already there and the DDL is compiled from the column + object either way. + """ + # ``.c`` is typed as the generic column collection; each entry is a real + # ``Column``, which is what ``CreateColumn`` needs. + for column in ( + IngestJobRow.__table__.c.batch_name, + IngestJobRow.__table__.c.processed, + IngestJobRow.__table__.c.total, + IngestJobRow.__table__.c.failures, + ): + _add_column(connection, cast(Column[object], column)) + + MIGRATIONS: list[Migration] = [ Migration(version=1, name="initial_schema", upgrade=_create_initial_schema), Migration( @@ -378,6 +414,11 @@ def _add_ingest_origin_and_uniqueness(connection: Connection) -> None: name="ingest_pipeline", upgrade=_add_ingest_origin_and_uniqueness, ), + Migration( + version=9, + name="ingest_job_progress", + upgrade=_add_ingest_progress_and_report, + ), ] FORMAT_VERSION: int = MIGRATIONS[-1].version diff --git a/src/visionset/kernel/domain/__init__.py b/src/visionset/kernel/domain/__init__.py index 10d52966..43a37b35 100644 --- a/src/visionset/kernel/domain/__init__.py +++ b/src/visionset/kernel/domain/__init__.py @@ -33,6 +33,7 @@ PolygonGeometry, ) from visionset.kernel.domain.ingest import ( + INGEST_TRANSITIONS, IngestFailure, IngestFailureKind, IngestJob, @@ -97,12 +98,14 @@ TaskGroup, progress_after_annotating, ) +from visionset.kernel.domain.transitions import require_move from visionset.kernel.domain.workspace import Workspace __all__ = [ "ASSET_PROGRESS_TRANSITIONS", "BATCH_TRANSITIONS", "IMPLEMENTED_GEOMETRIES", + "INGEST_TRANSITIONS", "JOB_TRANSITIONS", "MANIFEST_VERSION", "PROMOTABLE_PROGRESS", @@ -171,5 +174,6 @@ "normalize_name", "partition_assets", "progress_after_annotating", + "require_move", "sha256_hex", ] diff --git a/src/visionset/kernel/domain/ingest.py b/src/visionset/kernel/domain/ingest.py index 1ba0c05c..fb158822 100644 --- a/src/visionset/kernel/domain/ingest.py +++ b/src/visionset/kernel/domain/ingest.py @@ -2,22 +2,24 @@ """One ingestion run: the record of it, and the report it hands back. Two shapes live here and they are not the same kind of thing. ``IngestJob`` is a -**row** — it outlives the call, and #19 is what gives it a transition table, -progress counters and a persisted error report. ``IngestResult`` is a **return -value** — it exists for the caller of ``IngestService.ingest`` and is never -stored. #20 keeps them apart deliberately: the pipeline works today and reports -in memory, and #19 turns that report into columns without changing what the -pipeline computes. - -Counts are derived properties rather than stored fields, the way batch -completion and per-asset progress are derived elsewhere in this domain. #19's -counters are a different thing: a running total written to a row *while* a job -is in flight, which a summary of a finished object cannot serve. +**row** — it outlives the call, carries the run's state machine, its progress +counters and its per-file report. ``IngestResult`` is a **return value** — it +exists for the caller of ``IngestService.ingest`` and is never stored. The two +overlap on purpose: the row is what a *poller* reads while the run is in flight +or long after it, and the result is what the caller who waited already has in +hand. + +Summary counts on ``IngestResult`` stay derived properties, the way batch +completion and per-asset progress are derived elsewhere in this domain. The +job's counters are a different thing: a running total written to a row *while* +the work is happening, which a summary of a finished object cannot serve. """ from __future__ import annotations +from collections.abc import Mapping from enum import StrEnum +from typing import Final from uuid import UUID, uuid4 from pydantic import BaseModel, ConfigDict, Field @@ -26,22 +28,45 @@ class IngestState(StrEnum): + """Lifecycle: pending -> running -> (completed | failed) -> running. + + ``IngestService`` owns the moves; ``INGEST_TRANSITIONS`` below is the whole + of what is legal. + """ + PENDING = "pending" RUNNING = "running" COMPLETED = "completed" FAILED = "failed" -class IngestJob(BaseModel): - """Tracks one ingestion run of a Source into a Project's asset pool.""" - - id: UUID = Field(default_factory=uuid4) - source_id: UUID - state: IngestState = IngestState.PENDING - error: str | None = None - #: The batch this run materialized into, NULL until it has reached one. - #: Declared last: it arrives by ``ALTER TABLE`` in migration 8. - batch_id: UUID | None = None +INGEST_TRANSITIONS: Final[Mapping[IngestState, frozenset[IngestState]]] = { + IngestState.PENDING: frozenset({IngestState.RUNNING, IngestState.FAILED}), + IngestState.RUNNING: frozenset({IngestState.COMPLETED, IngestState.FAILED}), + IngestState.COMPLETED: frozenset(), + IngestState.FAILED: frozenset({IngestState.RUNNING}), +} +"""Every move a run may make. Anything absent raises ``InvalidTransition``. + +A table rather than guards inside the service, the shape ``BATCH_TRANSITIONS`` +established: "which moves are legal" is one readable fact, and the test for it +sweeps the whole ``IngestState`` column against this dict instead of restating +it. + +``failed -> running`` is **the first backward edge in this kernel**, and the +argument against reopening a batch does not transfer. A batch pins a schema +version at approval and its jobs are already partitioned against that pin, so +un-freezing one would invalidate work already done. Nothing is pinned against an +ingest run: it is a record of work, not an artifact with dependents. Resuming is +therefore the same unit of work continuing, and a second row per attempt would +fork ``batch_id`` and fill ``IngestService.list`` with retries. + +``running -> running`` is deliberately **absent**, which means a run stuck at +``running`` cannot be resumed. That state is a process that died without +reporting anything, not a failure somebody can read — and the remedy already +exists: ingest the source again, which content addressing makes create nothing. +Letting a resume overwrite the row would erase the only evidence the crash left. +""" class IngestFailureKind(StrEnum): @@ -78,13 +103,55 @@ class IngestFailure(BaseModel): reason: str +class IngestJob(BaseModel): + """Tracks one ingestion run of a Source into a Project's asset pool. + + Declared after ``IngestFailure`` because it holds a tuple of them, and + pydantic resolves that annotation when the class is created rather than when + it is first used. + + The last four fields arrive by ``ALTER TABLE`` and are therefore declared + **last, in migration order** — the rule ``AssetRow`` and ``AnnotationRow`` + already follow, and the reason is that SQLite appends an added column, so a + different order here would make the ``create_all`` path and the migration + path emit different DDL. + """ + + id: UUID = Field(default_factory=uuid4) + source_id: UUID + state: IngestState = IngestState.PENDING + #: The fatal cause that stopped the run, as opposed to the per-file report + #: below. One broken machine is not five thousand broken files. + error: str | None = None + #: The batch this run materialized into, NULL until it has reached one. + #: Declared last as of migration 8. + batch_id: UUID | None = None + #: The name a batch this run creates will take, decided before the decode so + #: that resuming a run which never reached a batch still lands where the + #: first attempt meant it to. NULL only on a row written before migration 9. + batch_name: str | None = None + #: Items read so far — decoded, hashed and stored, or reported as unreadable. + #: Written while the run is in flight, which is what makes it pollable. + processed: int = Field(default=0, ge=0) + #: Items the source offered, or NULL when that is not knowable in advance. + #: A directory can be listed; a clip cannot, because ``VideoMetadata`` + #: deliberately carries no frame count — the number an ingest wants is what + #: extraction produced, and anything else would be a guess with a VFR clip. + total: int | None = Field(default=None, ge=0) + #: The per-file report of the **current** attempt. A resumed run starts a + #: fresh one rather than accumulating across attempts. + failures: tuple[IngestFailure, ...] = () + + class IngestResult(BaseModel): """What one call to ``IngestService.ingest`` did. - In memory only: nothing reads this back, and #19 is what persists a report. - ``assets`` carries whole models rather than ids because there is no door - that reads an ``Asset`` back — a caller that has just ingested should not - have to reach into a repository to learn what it got. + The caller's copy of what the run's own row records, handed back so that + waiting for a synchronous run does not then require reading it. ``assets`` + carries whole models rather than ids because there is no door that reads an + ``Asset`` back — a caller that has just ingested should not have to reach + into a repository to learn what it got, and that is the one part of this + that the row cannot hold. """ model_config = ConfigDict(frozen=True, extra="forbid") diff --git a/src/visionset/kernel/domain/transitions.py b/src/visionset/kernel/domain/transitions.py new file mode 100644 index 00000000..369af3e5 --- /dev/null +++ b/src/visionset/kernel/domain/transitions.py @@ -0,0 +1,41 @@ +# usage: from visionset.kernel.domain import require_move +"""The one way to ask a transition table whether a move is allowed. + +Three state machines live in this domain — ``BATCH_TRANSITIONS``, +``JOB_TRANSITIONS`` / ``ASSET_PROGRESS_TRANSITIONS``, and ``INGEST_TRANSITIONS`` +— and "is this move in the table" is the same question for all of them. It is +asked here rather than once per service, so a refusal reads the same way +whichever machine produced it and no service can quietly grow a chain of guards +that disagrees with its own table. + +Domain rather than services, on ``normalize_name``'s terms: a rule every service +consults, expressed against domain values, raising a domain error. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from enum import StrEnum + +from visionset.kernel.errors import InvalidTransition + + +def require_move[S: StrEnum]( + transitions: Mapping[S, frozenset[S]], current: S, to: S, subject: str +) -> None: + """Consult a transition table, and refuse in its own vocabulary. + + Generic over every machine rather than written once per service: ``subject`` + is what lets the refusal name a batch by its name and a run by its id while + the sentence itself stays one sentence. + + Raises: + InvalidTransition: ``to`` is not in the table's entry for ``current``. + """ + if to in transitions[current]: + return + legal = ", ".join(sorted(state.value for state in transitions[current])) or "nothing" + raise InvalidTransition( + f"{subject} is {current.value!r} and cannot become {to.value!r}; " + f"from here it can only become {legal}" + ) diff --git a/src/visionset/kernel/services/batch_service.py b/src/visionset/kernel/services/batch_service.py index 0b49f219..7228b168 100644 --- a/src/visionset/kernel/services/batch_service.py +++ b/src/visionset/kernel/services/batch_service.py @@ -49,6 +49,7 @@ TaskGroup, normalize_name, partition_assets, + require_move, ) from visionset.kernel.errors import ( AssetNotFound, @@ -57,7 +58,6 @@ BatchNotFound, ConfirmationRequired, EmptyBatch, - InvalidTransition, ProjectNotFound, ) from visionset.kernel.ports import UnitOfWork @@ -193,7 +193,7 @@ def approve(self, batch_id: UUID, partition: Partition | None = None) -> Batch: """ with self._workspace.unit_of_work() as uow: batch = self.require_batch(uow, batch_id) - self._require_move(batch, BatchState.APPROVED) + require_move(BATCH_TRANSITIONS, batch.state, BatchState.APPROVED, _subject(batch)) if not batch.asset_ids: raise EmptyBatch( f"batch {batch.name!r} has no assets; an approved empty batch has no jobs " @@ -257,7 +257,7 @@ def complete(self, batch_id: UUID) -> Batch: """ with self._workspace.unit_of_work() as uow: batch = self.require_batch(uow, batch_id) - self._require_move(batch, BatchState.COMPLETED) + require_move(BATCH_TRANSITIONS, batch.state, BatchState.COMPLETED, _subject(batch)) jobs = jobs_of(uow, batch) outstanding = [j for j in jobs if j.state is not AnnotationJobState.COMPLETED] if outstanding: @@ -298,21 +298,15 @@ def delete(self, batch_id: UUID, *, confirm: bool = False) -> None: uow.batches.delete(batch.id) # --- the transition table, consulted rather than restated --------------- + # ``require_move`` lives in ``domain/transitions.py``; every machine in this + # kernel asks it the same question, so the refusal reads the same way. def _move(self, batch_id: UUID, to: BatchState) -> Batch: with self._workspace.unit_of_work() as uow: batch = self.require_batch(uow, batch_id) - self._require_move(batch, to) + require_move(BATCH_TRANSITIONS, batch.state, to, _subject(batch)) return uow.batches.update(batch.model_copy(update={"state": to})) - def _require_move(self, batch: Batch, to: BatchState) -> None: - if to not in BATCH_TRANSITIONS[batch.state]: - legal = ", ".join(sorted(s.value for s in BATCH_TRANSITIONS[batch.state])) or "nothing" - raise InvalidTransition( - f"batch {batch.name!r} is {batch.state.value!r} and cannot become {to.value!r}; " - f"from here it can only become {legal}" - ) - # --- lookups shared by the operations above ---------------------------- def _require_project(self, uow: UnitOfWork, project_id: UUID) -> Project: @@ -368,6 +362,11 @@ def require_draft(self, uow: UnitOfWork, batch_id: UUID) -> Batch: return batch +def _subject(batch: Batch) -> str: + """How a refused move names the batch. One spelling, so refusals read alike.""" + return f"batch {batch.name!r}" + + def jobs_of(uow: UnitOfWork, batch: Batch) -> list[AnnotationJob]: """Every job under the batch, task group by task group, in segment order. diff --git a/src/visionset/kernel/services/ingest_service.py b/src/visionset/kernel/services/ingest_service.py index 9553bbf4..c4ada4b4 100644 --- a/src/visionset/kernel/services/ingest_service.py +++ b/src/visionset/kernel/services/ingest_service.py @@ -20,30 +20,44 @@ follows. Re-running an ingest therefore creates nothing and is not an error; it is how a source that grew by three files is caught up. -**Four transactions, not one, and the middle of the run is in none of them.** -Decoding is a Pillow pass over thousands of files or an out-of-process ffmpeg, -and holding a write transaction open across either is how a single-writer SQLite -store starts reporting "database is locked" (#80). So the run resolves what it -needs, closes the transaction, does the work, and opens another to record it. -Blob writes happen out there too, before any row exists: ``BlobStore.put`` is not +**The long middle of the run is in no transaction.** Decoding is a Pillow pass +over thousands of files or an out-of-process ffmpeg, and holding a write +transaction open across either is how a single-writer SQLite store starts +reporting "database is locked" (#80). So the run resolves what it needs, closes +the transaction, does the work, and opens another to record it. Blob writes +happen out there too, before any row exists: ``BlobStore.put`` is not transactional and a rollback cannot unwrite it — but a blob nothing points at is harmless (content-addressed, shared, never deleted), while a row naming bytes -that were never stored is not. The honest consequence, stated rather than -hidden: a process killed between transactions can leave assets in the project -with no batch, and a job stuck at ``running``. That is recoverable, and finding -it is what #19's job record is for. +that were never stored is not. + +The progress writes that go on *between* items are not a contradiction: each is +one ``UPDATE`` that opens and commits while nothing is being decoded. What the +warning is about is a transaction held **across** the decode, not the existence +of writes during the phase. + +**The job is a state machine, and it is a table.** ``INGEST_TRANSITIONS`` in +``domain/ingest.py`` is the whole of what is legal; this service consults it +through ``require_move`` and never restates it. It has the kernel's only +backward edge, ``failed -> running``, which is :meth:`IngestService.resume`. + +**Nothing carries a job across the decode.** ``Repository.update`` replaces the +whole row, so a model read before the work and written after it would silently +undo every counter the run recorded in between. Only ``job_id`` travels; every +write re-reads the row inside its own transaction. That is what +:meth:`require_job` is public for. **Failure splits by remedy, exactly as the media errors do.** A file that is not -an image, or one whose bytes will not decode, is *reported* — one entry in -``IngestResult.failures``, and the run carries on, because an operator with five +an image, or one whose bytes will not decode, is *reported* — one entry in the +job's ``failures``, and the run carries on, because an operator with five thousand files needs the other four thousand nine hundred. A missing ffmpeg is not a file's fault at all; it fails the job outright and is re-raised, which is precisely why ``MediaToolUnavailable`` sits outside the ``MediaError`` family. -**What #20 deliberately leaves for later.** No transition table, no persisted -progress counters and no persisted error report — #19 owns the job's lifecycle -and turns ``IngestResult`` into columns. No ``thumbnail_hash``: #21. This service -writes terminal job states directly and reports in memory. +**What is still deliberately not here.** No ``thumbnail_hash``: #21. No +background execution: a run is synchronous and in-process, and the API is shaped +so that putting it behind a queue changes the caller's waiting rather than its +vocabulary — which is why a job is created ``pending`` and moved to ``running`` +by whoever picks it up, even though today that is the same call. """ from __future__ import annotations @@ -53,6 +67,7 @@ from uuid import UUID from visionset.kernel.domain import ( + INGEST_TRANSITIONS, Asset, Batch, IngestCompleted, @@ -65,6 +80,7 @@ Source, SourceKind, normalize_name, + require_move, ) from visionset.kernel.errors import ( CorruptMedia, @@ -89,7 +105,12 @@ def __init__(self, workspace: WorkspaceService) -> None: # --- reading ----------------------------------------------------------- def get(self, job_id: UUID) -> IngestJob: - """The ingest job with that id. + """The ingest job with that id, including how far it has got. + + This is the polling contract. ``processed`` / ``total`` and the per-file + ``failures`` are written while the run is in flight, so calling this + from another thread, another process or the future HTTP surface reports + the run's real position rather than its last finished state. Raises: IngestJobNotFound: no such ingest job in this workspace. @@ -123,7 +144,9 @@ def ingest( ``UnsupportedMedia`` and ``CorruptMedia`` are collected rather than raised — see ``IngestResult.failures``, where each keeps the item's name - and the remedy apart. + and the remedy apart. The same report is written to the job's row as the + run goes, next to ``processed`` and ``total``, so a caller who did not + wait for this to return can still read both. See :meth:`get`. Raises: SourceNotFound: no such source in this workspace. @@ -141,34 +164,94 @@ def ingest( source = self._sources.require_source(uow, source_id) self._require_project(uow, source.project_id) name = self._target_name(uow, source, batch_id, batch_name) - job = uow.ingest_jobs.add(IngestJob(source_id=source.id, state=IngestState.RUNNING)) + # ``pending``, not ``running``: the row exists before anybody picks + # the work up, which is the vocabulary a queue will need and costs + # nothing today. Every refusal above happens before the insert, so a + # run that fails fast leaves no job row at all. + job = uow.ingest_jobs.add(IngestJob(source_id=source.id, batch_name=name)) + + return self._run(job.id, source, name, batch_id) + + def resume(self, job_id: UUID) -> IngestResult: + """Run a failed job again, on the same row and into the same batch. + + A **redo, not a skip**. There is no per-file record of what the previous + attempt managed, and there does not need to be: blobs are + content-addressed and assets are deduplicated by content, so re-reading + the whole source creates nothing it created before. The cost is + re-hashing what is already stored; what it buys is that resume has no + second code path to get it wrong. + + The counters and the per-file report are reset — they describe *this* + attempt, and a completed run still carrying the last attempt's report + would be a lie. The fatal ``error`` is cleared for the same reason. + + What may be resumed is whatever ``INGEST_TRANSITIONS`` says can reach + ``running``: a ``failed`` job, and a ``pending`` one, which a synchronous + run never leaves behind but a queued one would. A ``completed`` job + cannot, and neither can one stuck at ``running`` — that is a process that + died without reporting anything, so ingest the source again instead, + which creates nothing and leaves the crashed row as the record it is. + + Raises: + IngestJobNotFound: no such ingest job in this workspace. + InvalidTransition: the job is ``completed``, or stuck at + ``running`` — see ``INGEST_TRANSITIONS``. + SourceNotFound: the source has since been deleted. + BatchNotEditable: the batch the first attempt reached is past + ``draft``. + plus everything :meth:`ingest` raises. + """ + with self._workspace.unit_of_work() as uow: + job = self.require_job(uow, job_id) + source = self._sources.require_source(uow, job.source_id) + self._require_project(uow, source.project_id) + # The friendly pre-check, so a completed job is refused before the + # target batch is resolved. The real one is inside ``_run``, in the + # transaction that actually moves the row. + require_move(INGEST_TRANSITIONS, job.state, IngestState.RUNNING, _subject(job.id)) + name = self._target_name(uow, source, job.batch_id, job.batch_name) + + return self._run(job.id, source, name, job.batch_id) + # --- the run, phase by phase ------------------------------------------- + + def _run(self, job_id: UUID, source: Source, name: str, batch_id: UUID | None) -> IngestResult: + """The work itself, shared by a first attempt and by a resumed one. + + Takes ``job_id`` rather than an ``IngestJob`` on purpose: the row is + rewritten many times between here and the end, and a model captured now + would overwrite all of it — see the module docstring. + """ + self._begin(job_id) try: - candidates, failures = self._read(source) + candidates, failures = self._read(source, job_id) assets, created = self._store(source.project_id, candidates) batch = self._materialize(source.project_id, name, batch_id, assets) with self._workspace.unit_of_work() as uow: - job = uow.ingest_jobs.update( + job = self.require_job(uow, job_id) + require_move(INGEST_TRANSITIONS, job.state, IngestState.COMPLETED, _subject(job_id)) + uow.ingest_jobs.update( job.model_copy( update={"state": IngestState.COMPLETED, "batch_id": batch.id}, ) ) except Exception as exc: - self._fail(job.id, str(exc) or exc.__class__.__name__) + self._fail(job_id, str(exc) or exc.__class__.__name__) raise # After the block, never inside it: a subscriber must not be able to put # its own exception on a transaction's way out. self._workspace.event_bus.publish( IngestCompleted( - ingest_job_id=job.id, + ingest_job_id=job_id, project_id=source.project_id, source_id=source.id, asset_count=len(assets), ) ) return IngestResult( - job_id=job.id, + job_id=job_id, project_id=source.project_id, source_id=source.id, batch_id=batch.id, @@ -177,19 +260,40 @@ def ingest( failures=tuple(failures), ) - # --- the run, phase by phase ------------------------------------------- + def _begin(self, job_id: UUID) -> None: + """Take the job from ``pending`` or ``failed`` to ``running``, empty-handed. + + The reset is what makes a resumed run's counters and report describe the + attempt a caller is watching rather than the one that failed. + """ + with self._workspace.unit_of_work() as uow: + job = self.require_job(uow, job_id) + require_move(INGEST_TRANSITIONS, job.state, IngestState.RUNNING, _subject(job_id)) + uow.ingest_jobs.update( + job.model_copy( + update={ + "state": IngestState.RUNNING, + "error": None, + "processed": 0, + "total": None, + "failures": (), + } + ) + ) - def _read(self, source: Source) -> tuple[list[Asset], list[IngestFailure]]: + def _read(self, source: Source, job_id: UUID) -> tuple[list[Asset], list[IngestFailure]]: """Decode and store every item, outside any transaction. Returns candidate assets in the order the source offered them, plus one entry per item that could not be read at all. """ if source.kind is SourceKind.VIDEO: - return self._read_video(source) - return self._read_directory(source) + return self._read_video(source, job_id) + return self._read_directory(source, job_id) - def _read_directory(self, source: Source) -> tuple[list[Asset], list[IngestFailure]]: + def _read_directory( + self, source: Source, job_id: UUID + ) -> tuple[list[Asset], list[IngestFailure]]: """Every file at the top of the directory, in filename order. Top level only. Recursion is not a per-run option but a question about @@ -201,11 +305,19 @@ def _read_directory(self, source: Source) -> tuple[list[Asset], list[IngestFailu No suffix filter either: a ``notes.txt`` is reported as unsupported rather than skipped, because guessing which files an operator meant to offer is a policy the kernel would be inventing. + + This is the one path that can state a ``total`` up front, because + listing a directory is cheap and exact. The write before the loop is + what publishes it — and what makes an empty directory record ``0 of 0`` + rather than nothing at all. """ candidates: list[Asset] = [] failures: list[IngestFailure] = [] directory = Path(source.path) - for path in sorted(item for item in directory.iterdir() if item.is_file()): + paths = sorted(item for item in directory.iterdir() if item.is_file()) + total = len(paths) + self._record_progress(job_id, processed=0, total=total, failures=failures) + for path in paths: try: with path.open("rb") as handle: # Probe first: a file that is going to be refused should @@ -218,21 +330,30 @@ def _read_directory(self, source: Source) -> tuple[list[Asset], list[IngestFailu content_hash = self._workspace.blob_store.put(handle) except MediaError as exc: failures.append(_failure(str(path), exc)) - continue - candidates.append( - Asset( - project_id=source.project_id, - content_hash=content_hash, - uri=str(path), - width=metadata.width, - height=metadata.height, - format=metadata.format, - source_id=source.id, + else: + candidates.append( + Asset( + project_id=source.project_id, + content_hash=content_hash, + uri=str(path), + width=metadata.width, + height=metadata.height, + format=metadata.format, + source_id=source.id, + ) ) + # After every item, read or refused alike: ``processed`` counts what + # the run has dealt with, and a report that only appeared at the end + # would be invisible for exactly as long as it is interesting. + self._record_progress( + job_id, + processed=len(candidates) + len(failures), + total=total, + failures=failures, ) return candidates, failures - def _read_video(self, source: Source) -> tuple[list[Asset], list[IngestFailure]]: + def _read_video(self, source: Source, job_id: UUID) -> tuple[list[Asset], list[IngestFailure]]: """One asset per extracted frame, at the rate the source records. The frames are **not** re-probed. ``VideoProcessor`` guarantees each one @@ -247,6 +368,12 @@ def _read_video(self, source: Source) -> tuple[list[Asset], list[IngestFailure]] loop and what was extracted is kept. The loop is left by falling out of it, which is one of the two ways the port allows an iterator to be released. + + ``total`` stays NULL for the whole run, and honestly so. ``VideoMetadata`` + carries no frame count by design — it would be a guess for a + variable-rate clip and the number an ingest wants is what extraction + actually produced — so a total here would be arithmetic presented as + fact. ``processed`` still climbs, which is what a poller needs. """ provenance = source.require_video() candidates: list[Asset] = [] @@ -255,6 +382,7 @@ def _read_video(self, source: Source) -> tuple[list[Asset], list[IngestFailure]] frames = self._workspace.video_processor.frames( clip, fps=provenance.extraction_fps, name=clip.name ) + self._record_progress(job_id, processed=0, total=None, failures=failures) try: for frame in frames: content_hash = self._workspace.blob_store.put(BytesIO(frame.content)) @@ -271,8 +399,12 @@ def _read_video(self, source: Source) -> tuple[list[Asset], list[IngestFailure]] frame_timestamp=frame.timestamp, ) ) + self._record_progress( + job_id, processed=len(candidates), total=None, failures=failures + ) except MediaError as exc: failures.append(_failure(source.path, exc)) + self._record_progress(job_id, processed=len(candidates), total=None, failures=failures) return candidates, failures def _store(self, project_id: UUID, candidates: list[Asset]) -> tuple[list[Asset], list[UUID]]: @@ -317,14 +449,56 @@ def _materialize( return self._batches.create(project_id, name, asset_ids) return self._batches.add_assets(batch_id, asset_ids) - def _fail(self, job_id: UUID, cause: str) -> None: - """Record why a run stopped, on its own row and in its own transaction.""" + def _record_progress( + self, job_id: UUID, *, processed: int, total: int | None, failures: list[IngestFailure] + ) -> None: + """Publish how far the run has got, in one ``UPDATE`` of its own. + + Called after **every** item rather than on a cadence. The number a + caller polls is then never stale by an amount nobody can predict, and + there is no interval constant to pick — one that fits a directory of + five files and one that fits a directory of fifty thousand are not the + same number, and this service cannot know which it is looking at. The + cost is one small commit beside a decode-and-hash that costs an order of + magnitude more. + + Re-reads the row rather than updating a copy held by the caller: + ``Repository.update`` replaces the whole row, so a stale model would + undo everything written since it was read. + + A run whose job row has been deleted underneath it keeps going and + reports to nobody, the way ``_fail`` already tolerates the same thing — + losing the work over a missing receipt would be the worse answer. + """ with self._workspace.unit_of_work() as uow: job = uow.ingest_jobs.get(job_id) - if job is not None: - uow.ingest_jobs.update( - job.model_copy(update={"state": IngestState.FAILED, "error": cause}) + if job is None: + return + uow.ingest_jobs.update( + job.model_copy( + update={ + "processed": processed, + "total": total, + "failures": tuple(failures), + } ) + ) + + def _fail(self, job_id: UUID, cause: str) -> None: + """Record why a run stopped, on its own row and in its own transaction. + + The counters and the report the run got as far as writing are left + exactly where they are: they say how far it had come when it stopped, + which is the first thing anyone looking at a failure wants. + """ + with self._workspace.unit_of_work() as uow: + job = uow.ingest_jobs.get(job_id) + if job is None: + return + require_move(INGEST_TRANSITIONS, job.state, IngestState.FAILED, _subject(job_id)) + uow.ingest_jobs.update( + job.model_copy(update={"state": IngestState.FAILED, "error": cause}) + ) # --- lookups shared by the operations above ---------------------------- @@ -392,6 +566,11 @@ def list(self, source_id: UUID) -> list[IngestJob]: return uow.ingest_jobs.list(source.id) +def _subject(job_id: UUID) -> str: + """How a refused move names the run. One spelling, so refusals read alike.""" + return f"ingest job {job_id}" + + def _failure(name: str, exc: MediaError) -> IngestFailure: """One report line, with the item's name kept apart from the remedy. diff --git a/src/visionset/kernel/services/job_service.py b/src/visionset/kernel/services/job_service.py index 0751bc36..a579d2eb 100644 --- a/src/visionset/kernel/services/job_service.py +++ b/src/visionset/kernel/services/job_service.py @@ -11,7 +11,8 @@ - **Two state machines, both tables.** ``JOB_TRANSITIONS`` and ``ASSET_PROGRESS_TRANSITIONS`` live in ``domain/task.py``; this service - consults them and never restates them. Adding a state is one edit there. + consults them through ``domain.require_move`` and never restates them. + Adding a state is one edit there. - **Work only happens inside an open batch.** Every write here requires the job's batch to be ``in_annotation``. ``AnnotationService`` needs the same gate, and rather than restate it, it calls :meth:`JobService.require_job` and @@ -31,8 +32,6 @@ from __future__ import annotations -from collections.abc import Mapping -from enum import StrEnum from uuid import UUID from visionset.kernel.domain import ( @@ -45,11 +44,11 @@ AssetProgress, Batch, BatchState, + require_move, ) from visionset.kernel.errors import ( AssetNotInJob, BatchNotInAnnotation, - InvalidTransition, JobNotComplete, JobNotFound, ProjectNotFound, @@ -180,7 +179,7 @@ def complete(self, job_id: UUID) -> AnnotationJob: with self._workspace.unit_of_work() as uow: job = self.require_job(uow, job_id) self.require_open_batch(uow, job) - _require_move(JOB_TRANSITIONS, job.state, AnnotationJobState.COMPLETED, f"job {job.id}") + require_move(JOB_TRANSITIONS, job.state, AnnotationJobState.COMPLETED, f"job {job.id}") unsettled = _unsettled(job) if unsettled: @@ -232,7 +231,7 @@ def mark(self, job_id: UUID, asset_id: UUID, progress: AssetProgress) -> Annotat if current is progress: return job - _require_move( + require_move( ASSET_PROGRESS_TRANSITIONS, current, progress, f"asset {asset_id} in job {job.id}" ) # Rewriting an existing key keeps its place in the dict, which keeps @@ -247,7 +246,7 @@ def _move(self, job_id: UUID, to: AnnotationJobState) -> AnnotationJob: with self._workspace.unit_of_work() as uow: job = self.require_job(uow, job_id) self.require_open_batch(uow, job) - _require_move(JOB_TRANSITIONS, job.state, to, f"job {job.id}") + require_move(JOB_TRANSITIONS, job.state, to, f"job {job.id}") return uow.annotation_jobs.update(job.model_copy(update={"state": to})) def _require_project(self, uow: UnitOfWork, project_id: UUID) -> None: @@ -312,24 +311,6 @@ def require_open_batch(self, uow: UnitOfWork, job: AnnotationJob) -> Batch: return batch -def _require_move[S: StrEnum]( - transitions: Mapping[S, frozenset[S]], current: S, to: S, subject: str -) -> None: - """Consult a transition table, and refuse in its own vocabulary. - - Generic over both machines rather than written twice: "is this move in the - table" is the same question for a job and for an asset, and the answer should - read the same way in both refusals. - """ - if to in transitions[current]: - return - legal = ", ".join(sorted(state.value for state in transitions[current])) or "nothing" - raise InvalidTransition( - f"{subject} is {current.value!r} and cannot become {to.value!r}; " - f"from here it can only become {legal}" - ) - - def _require_asset(uow: UnitOfWork, job: AnnotationJob, asset_id: UUID) -> Asset: """The asset, or report that the job is tracking one that is not there. diff --git a/tests/kernel/test_ingest_service.py b/tests/kernel/test_ingest_service.py index 2f00c1fd..03e6c911 100644 --- a/tests/kernel/test_ingest_service.py +++ b/tests/kernel/test_ingest_service.py @@ -13,6 +13,7 @@ """ from pathlib import Path +from typing import BinaryIO from uuid import UUID, uuid4 import pytest @@ -34,24 +35,36 @@ BatchNotFound, IngestJobNotFound, InvalidName, + InvalidTransition, MediaToolUnavailable, SourceNotFound, ) +from visionset.kernel.adapters import PillowImageProcessor from visionset.kernel.domain import ( + INGEST_TRANSITIONS, Asset, BatchState, + GeometryType, ImageFormat, + ImageMetadata, IngestCompleted, IngestFailureKind, + IngestJob, IngestState, + LabelClass, + Project, + Source, + SourceKind, VideoFrame, VideoMetadata, + VideoProvenance, ) from visionset.kernel.ports import FRAME_FORMAT from visionset.kernel.services import ( BatchService, IngestService, ProjectService, + SchemaService, SourceService, WorkspaceService, ) @@ -73,6 +86,87 @@ def frames( raise MediaToolUnavailable("ffmpeg is not installed; install it and try again") +class _WatchingProcessor: + """The real decoder, plus a look at the run's own row before each file. + + Injected through the composition point, like `_NoFfmpeg`. The look is taken + through a **second** `WorkspaceService` opened on the same directory, + because what is being tested is that the counters are committed while the + run is still going — a read on the service doing the work would prove less. + """ + + def __init__(self, root: Path, source_id: list[UUID], seen: list[IngestJob]) -> None: + self._root = root + self._source_id = source_id # filled by the test once the source exists + self._seen = seen + self._real = PillowImageProcessor() + + def _observe(self) -> None: + watcher = WorkspaceService.open(self._root) + try: + self._seen.append(IngestService(watcher).list(self._source_id[0])[-1]) + finally: + watcher.close() + + def probe(self, content: BinaryIO, *, name: str | None = None) -> ImageMetadata: + self._observe() + return self._real.probe(content, name=name) + + def thumbnail( + self, content: BinaryIO, *, max_edge: int = 256, name: str | None = None + ) -> bytes: + return self._real.thumbnail(content, max_edge=max_edge, name=name) + + +class _FailsOnNthFile: + """A decoder that stops being a decoder partway through, and not politely. + + `OSError` rather than a `MediaError`: the point is a cause the run cannot + report per file and has to fail on, which is what leaves the counters + holding the position it reached. + """ + + def __init__(self, nth: int) -> None: + self._nth = nth + self._calls = 0 + self._real = PillowImageProcessor() + + def probe(self, content: BinaryIO, *, name: str | None = None) -> ImageMetadata: + self._calls += 1 + if self._calls == self._nth: + raise OSError("the disk went away") + return self._real.probe(content, name=name) + + def thumbnail( + self, content: BinaryIO, *, max_edge: int = 256, name: str | None = None + ) -> bytes: + return self._real.thumbnail(content, max_edge=max_edge, name=name) + + +def _planted_video_source(workspace: WorkspaceService, project: Project, tmp_path: Path) -> Source: + """A video source written straight to the store, because registering probes. + + The subject of the tests that use this is the extraction, not the + registration, and registration would fail first on a machine with no ffmpeg. + """ + clip = tmp_path / "clip.mp4" + clip.write_bytes(b"not really a clip") + with workspace.unit_of_work() as uow: + return uow.sources.add( + Source( + project_id=project.id, + kind=SourceKind.VIDEO, + path=str(clip), + video=VideoProvenance( + metadata=VideoMetadata( + width=64, height=48, fps=10.0, duration_seconds=2.0, codec="h264" + ), + extraction_fps=1.0, + ), + ) + ) + + class Fixture: """A workspace with one project, a directory to fill, and every service.""" @@ -99,6 +193,43 @@ def assets(self) -> list[Asset]: with self.workspace.unit_of_work() as uow: return uow.assets.list(self.project.id) + def freeze(self, batch_id: UUID) -> None: + """Approve the batch, creating the schema version approval has to pin.""" + SchemaService(self.workspace).create_version( + self.project.id, [LabelClass(name="thing", geometry=GeometryType.BBOX)] + ) + self.batches.approve(batch_id) + + def job_in(self, state: IngestState) -> IngestJob: + """A job in `state`, over a source of two images that is readable now. + + `completed` and `failed` are walked to through real operations — a run + that works, and a run whose directory was taken away and put back. The + other two are written directly, and this is the only place in this file + that plants a state rather than reaching it: a synchronous run passes + through `pending` and `running` inside a single call and never leaves + one behind, so there is nothing to walk to. Leaving them out instead + would leave half the table unswept. + """ + write_images(self.stills, count=2) + source = self.sources.register_images(self.project.id, self.stills) + if state is IngestState.COMPLETED: + return self.ingest.get(self.ingest.ingest(source.id).job_id) + if state is IngestState.FAILED: + files = sorted(self.stills.iterdir()) + for path in files: + path.unlink() + self.stills.rmdir() + with pytest.raises(FileNotFoundError): + self.ingest.ingest(source.id) + self.stills.mkdir() + write_images(self.stills, count=2) + return self.ingest.list(source.id)[0] + with self.workspace.unit_of_work() as uow: + return uow.ingest_jobs.add( + IngestJob(source_id=source.id, state=state, batch_name="planted") + ) + def close(self) -> None: self.workspace.close() @@ -502,13 +633,7 @@ def test_ingesting_into_a_frozen_batch_is_refused_before_anything_is_decoded( write_images(fixture.stills, count=2) seed = fixture.sources.register_images(fixture.project.id, fixture.stills) opened = fixture.ingest.ingest(seed.id) - from visionset.kernel.domain import GeometryType, LabelClass - from visionset.kernel.services import SchemaService - - SchemaService(fixture.workspace).create_version( - fixture.project.id, [LabelClass(name="thing", geometry=GeometryType.BBOX)] - ) - fixture.batches.approve(opened.batch_id) + fixture.freeze(opened.batch_id) more = tmp_path / "more" write_images(more, count=2, first_seed=90) second = fixture.sources.register_images(fixture.project.id, more) @@ -646,27 +771,7 @@ def test_a_missing_decoder_fails_the_job_and_is_re_raised(tmp_path: Path) -> Non workspace = WorkspaceService.init(tmp_path / "ws", video_processor_factory=_NoFfmpeg) projects = ProjectService(workspace) ingest = IngestService(workspace) - project = projects.create("p") - # Registration probes too, so the source is planted by hand rather than - # registered: the subject here is the extraction, not the registration. - clip = tmp_path / "clip.mp4" - clip.write_bytes(b"not really a clip") - from visionset.kernel.domain import Source, SourceKind, VideoProvenance - - with workspace.unit_of_work() as uow: - source = uow.sources.add( - Source( - project_id=project.id, - kind=SourceKind.VIDEO, - path=str(clip), - video=VideoProvenance( - metadata=VideoMetadata( - width=64, height=48, fps=10.0, duration_seconds=2.0, codec="h264" - ), - extraction_fps=1.0, - ), - ) - ) + source = _planted_video_source(workspace, projects.create("p"), tmp_path) with pytest.raises(MediaToolUnavailable): ingest.ingest(source.id) @@ -703,7 +808,7 @@ def test_every_run_of_one_source_is_listed_in_order(tmp_path: Path) -> None: def test_require_job_resolves_inside_a_callers_transaction(tmp_path: Path) -> None: - """The shape #19 needs: a gate it can run in its own unit of work.""" + """The gate the service runs in its own unit of work before every write.""" fixture = Fixture(tmp_path) source = fixture.sources.register_images(fixture.project.id, fixture.stills) result = fixture.ingest.ingest(source.id) @@ -713,6 +818,327 @@ def test_require_job_resolves_inside_a_callers_transaction(tmp_path: Path) -> No fixture.close() +# --- the transition table, swept in full ---------------------------------- + + +@pytest.mark.parametrize("origin", list(IngestState), ids=lambda s: f"from-{s.value}") +def test_the_transition_table_is_the_whole_of_what_can_be_resumed( + tmp_path: Path, origin: IngestState +) -> None: + """Every state, checked against the table itself rather than against a list. + + `resume` is the one operation that names a target — `running` — so the + column of the square it can reach is the whole of what there is to sweep. + """ + fixture = Fixture(tmp_path) + job = fixture.job_in(origin) + + if IngestState.RUNNING in INGEST_TRANSITIONS[origin]: + assert fixture.ingest.resume(job.id).job_id == job.id + assert fixture.ingest.get(job.id).state is IngestState.COMPLETED + else: + with pytest.raises(InvalidTransition, match="cannot become"): + fixture.ingest.resume(job.id) + assert fixture.ingest.get(job.id).state is origin + fixture.close() + + +def test_a_completed_run_can_go_nowhere() -> None: + assert INGEST_TRANSITIONS[IngestState.COMPLETED] == frozenset() + + +def test_a_run_stuck_at_running_cannot_be_resumed() -> None: + """A crashed process is not a reported failure, and must not be overwritten. + + Ingesting the source again is the remedy — content addressing makes that + create nothing — and it leaves the stuck row as the record of the crash. + """ + assert IngestState.RUNNING not in INGEST_TRANSITIONS[IngestState.RUNNING] + + +def test_the_refusal_says_where_the_run_can_actually_go(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + job = fixture.job_in(IngestState.COMPLETED) + + with pytest.raises(InvalidTransition, match="can only become nothing"): + fixture.ingest.resume(job.id) + fixture.close() + + +# --- progress, while the run is still going ------------------------------- + + +def test_progress_is_visible_to_somebody_who_is_not_running_the_ingest( + tmp_path: Path, +) -> None: + """The polling contract, read the way the API and the UI will read it. + + The observer is a *second* `WorkspaceService` opened on the same directory, + not another call on the one doing the work: what is being claimed is that + the counters are committed as the run goes, and only a separate connection + can show that. + """ + root = tmp_path / "ws" + source_id: list[UUID] = [] + seen: list[IngestJob] = [] + workspace = WorkspaceService.init( + root, image_processor_factory=lambda: _WatchingProcessor(root, source_id, seen) + ) + projects = ProjectService(workspace) + sources = SourceService(workspace) + ingest = IngestService(workspace) + project = projects.create("p") + stills = tmp_path / "stills" + stills.mkdir() + write_images(stills, count=3) + source = sources.register_images(project.id, stills) + source_id.append(source.id) + + result = ingest.ingest(source.id) + + # One observation per file, each taken before that file was counted. + assert [job.processed for job in seen] == [0, 1, 2] + assert {job.total for job in seen} == {3} + assert {job.state for job in seen} == {IngestState.RUNNING} + assert ingest.get(result.job_id).processed == 3 + workspace.close() + + +def test_a_completed_run_records_how_many_items_it_processed(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=4) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id) + + job = fixture.ingest.get(result.job_id) + assert (job.processed, job.total) == (4, 4) + fixture.close() + + +def test_an_unreadable_file_still_counts_as_processed(tmp_path: Path) -> None: + """`processed` is items dealt with, not items that became assets.""" + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=2) + write_unsupported_file(fixture.stills / "notes.txt") + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id) + + job = fixture.ingest.get(result.job_id) + assert (job.processed, job.total) == (3, 3) + fixture.close() + + +def test_an_empty_directory_records_a_total_of_zero(tmp_path: Path) -> None: + """Written before the loop, which is the only reason an empty run says anything.""" + fixture = Fixture(tmp_path) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id) + + job = fixture.ingest.get(result.job_id) + assert (job.processed, job.total) == (0, 0) + fixture.close() + + +def test_a_clip_records_no_total_because_extraction_decides_it(tmp_path: Path) -> None: + """`VideoMetadata` carries no frame count, so a total here would be a guess.""" + fixture = Fixture(tmp_path) + clip = fixture.clip() + source = fixture.sources.register_video(fixture.project.id, clip.path, extraction_fps=1.0) + + result = fixture.ingest.ingest(source.id) + + job = fixture.ingest.get(result.job_id) + assert job.total is None + assert job.processed == len(result.assets) + fixture.close() + + +# --- the report, on the row ----------------------------------------------- + + +def test_a_run_records_which_files_failed_and_why(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=1) + write_corrupt_image(fixture.stills / "broken.png") + write_unsupported_file(fixture.stills / "notes.txt") + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id) + + job = fixture.ingest.get(result.job_id) + assert job.state is IngestState.COMPLETED # per-file failures do not fail a run + assert job.failures == result.failures + assert {failure.kind for failure in job.failures} == { + IngestFailureKind.CORRUPT, + IngestFailureKind.UNSUPPORTED, + } + for failure in job.failures: + assert failure.name not in failure.reason + fixture.close() + + +def test_a_fatal_cause_is_recorded_apart_from_the_per_file_report(tmp_path: Path) -> None: + """One broken machine is not five thousand broken files, on the row too.""" + workspace = WorkspaceService.init(tmp_path / "ws", video_processor_factory=_NoFfmpeg) + ingest = IngestService(workspace) + source = _planted_video_source(workspace, ProjectService(workspace).create("p"), tmp_path) + + with pytest.raises(MediaToolUnavailable): + ingest.ingest(source.id) + + job = ingest.list(source.id)[0] + assert job.state is IngestState.FAILED + assert "ffmpeg" in (job.error or "") + assert job.failures == () + workspace.close() + + +def test_a_failed_run_keeps_the_progress_it_had_made(tmp_path: Path) -> None: + """How far it got is the first thing anyone reading a failure wants. + + The decoder gives up on the third of four files with something that is not a + `MediaError` at all — a disk that went away, say — so the run stops rather + than reporting it, and the two files it had already counted stay counted. + """ + root = tmp_path / "ws" + workspace = WorkspaceService.init(root, image_processor_factory=lambda: _FailsOnNthFile(3)) + sources = SourceService(workspace) + ingest = IngestService(workspace) + project = ProjectService(workspace).create("p") + stills = tmp_path / "stills" + stills.mkdir() + write_images(stills, count=4) + source = sources.register_images(project.id, stills) + + with pytest.raises(OSError, match="the disk went away"): + ingest.ingest(source.id) + + job = ingest.list(source.id)[0] + assert job.state is IngestState.FAILED + assert (job.processed, job.total) == (2, 4) + workspace.close() + + +# --- resuming a failed run ------------------------------------------------ + + +def test_resuming_a_failed_run_completes_it_on_the_same_row(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + job = fixture.job_in(IngestState.FAILED) + + result = fixture.ingest.resume(job.id) + + assert result.job_id == job.id + assert fixture.ingest.get(job.id).state is IngestState.COMPLETED + assert fixture.batches.get(result.batch_id).asset_ids == list(result.asset_ids) + fixture.close() + + +def test_resuming_creates_no_second_job(tmp_path: Path) -> None: + """A run is one unit of work; a row per attempt would fork its batch.""" + fixture = Fixture(tmp_path) + job = fixture.job_in(IngestState.FAILED) + + fixture.ingest.resume(job.id) + + assert [stored.id for stored in fixture.ingest.list(job.source_id)] == [job.id] + fixture.close() + + +def test_resuming_clears_the_previous_attempts_report(tmp_path: Path) -> None: + """The counters and the report describe *this* attempt, not the last one.""" + fixture = Fixture(tmp_path) + job = fixture.job_in(IngestState.FAILED) + assert fixture.ingest.get(job.id).error is not None + + fixture.ingest.resume(job.id) + + resumed = fixture.ingest.get(job.id) + assert resumed.error is None + assert resumed.failures == () + assert (resumed.processed, resumed.total) == (2, 2) + fixture.close() + + +def test_resuming_keeps_the_batch_name_the_first_attempt_was_given(tmp_path: Path) -> None: + """Which is what `batch_name` is a column for: a failed run reached no batch.""" + fixture = Fixture(tmp_path) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + fixture.stills.rmdir() + with pytest.raises(FileNotFoundError): + fixture.ingest.ingest(source.id, batch_name="monday") + fixture.stills.mkdir() + write_images(fixture.stills, count=1) + + job = fixture.ingest.list(source.id)[0] + result = fixture.ingest.resume(job.id) + + assert fixture.batches.get(result.batch_id).name == "monday" + fixture.close() + + +def test_resuming_creates_no_new_blobs_for_what_was_already_stored(tmp_path: Path) -> None: + """Resume is a redo, and a redo of content-addressed work costs no storage.""" + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=3) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + result = fixture.ingest.ingest(source.id) + before = fixture.blob_count() + # Put the completed run back into a state resume accepts, which nothing + # public does: the point here is the second read of the same bytes. + with fixture.workspace.unit_of_work() as uow: + stored = uow.ingest_jobs.get(result.job_id) + assert stored is not None + uow.ingest_jobs.update(stored.model_copy(update={"state": IngestState.FAILED})) + + again = fixture.ingest.resume(result.job_id) + + assert again.created == 0 + assert again.asset_ids == result.asset_ids + assert fixture.blob_count() == before + fixture.close() + + +def test_a_resumed_run_announces_itself(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + job = fixture.job_in(IngestState.FAILED) + announced: list[IngestCompleted] = [] + fixture.workspace.event_bus.subscribe(IngestCompleted, announced.append) + + fixture.ingest.resume(job.id) + + assert [event.ingest_job_id for event in announced] == [job.id] + fixture.close() + + +def test_resuming_into_a_batch_that_was_frozen_meanwhile_is_refused(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=1) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + batch = fixture.batches.create(fixture.project.id, "target", []) + result = fixture.ingest.ingest(source.id, batch_id=batch.id) + with fixture.workspace.unit_of_work() as uow: + stored = uow.ingest_jobs.get(result.job_id) + assert stored is not None + uow.ingest_jobs.update(stored.model_copy(update={"state": IngestState.FAILED})) + fixture.freeze(batch.id) + + with pytest.raises(BatchNotEditable): + fixture.ingest.resume(result.job_id) + fixture.close() + + +def test_resuming_an_unknown_job_is_refused(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + + with pytest.raises(IngestJobNotFound): + fixture.ingest.resume(uuid4()) + fixture.close() + + # --- scope ---------------------------------------------------------------- diff --git a/tests/kernel/test_metadata_store.py b/tests/kernel/test_metadata_store.py index 3297578a..1c076850 100644 --- a/tests/kernel/test_metadata_store.py +++ b/tests/kernel/test_metadata_store.py @@ -38,7 +38,10 @@ DatasetChange, DatasetMember, GeometryType, + IngestFailure, + IngestFailureKind, IngestJob, + IngestState, LabelClass, PolygonGeometry, Project, @@ -83,7 +86,25 @@ def _seed(uow: UnitOfWork) -> list[tuple[str, UUID]]: ), ) ) - ingest = uow.ingest_jobs.add(IngestJob(source_id=source.id)) + # Counters set and a *populated* report, for the same reason the source + # above is a video one: leaving `failures` at its default would leave the + # JSON column empty in every store test and the nested round trip untested. + ingest = uow.ingest_jobs.add( + IngestJob( + source_id=source.id, + state=IngestState.COMPLETED, + batch_name="monday", + processed=3, + total=4, + failures=( + IngestFailure( + name="/data/notes.txt", + kind=IngestFailureKind.UNSUPPORTED, + reason="not an image", + ), + ), + ) + ) first = uow.assets.add( Asset(project_id=project.id, content_hash="a" * 64, uri="file:///1.png", width=8, height=6) ) diff --git a/tests/kernel/test_migrations.py b/tests/kernel/test_migrations.py index a4ced587..d6f81b16 100644 --- a/tests/kernel/test_migrations.py +++ b/tests/kernel/test_migrations.py @@ -87,6 +87,13 @@ def _downgrade_to_version_one(store: SqliteMetadataStore) -> None: # refuses to drop such a column at all — the constraint would be left # naming something that is gone. Migration 8 rebuilds this table for the # same underlying reason, so its undo is a rebuild too. It is empty here. + # + # This rebuild is also migration 9's undo: its four columns live on this + # same table, so restoring the generation-1 shape removes them with + # everything else. That is why nothing below mentions them — and why + # ``test_migration_nine_alters_a_table_migration_eight_rebuilt`` exists, + # because from here migration 8 re-creates the table whole and 9 never + # runs as the ``ALTER`` that a real generation-8 database gets. connection.execute(text("drop table ingest_job")) connection.execute( text( @@ -406,6 +413,104 @@ def test_migration_eight_refuses_a_workspace_that_still_holds_a_pre_ingest_job( store.close() +#: What migration 9 adds, and therefore what generation 8 did not have. +_MIGRATION_NINE_COLUMNS = ("batch_name", "processed", "total", "failures") + + +def _downgrade_to_generation_eight(store: SqliteMetadataStore) -> None: + """Take ``ingest_job`` back to the shape migration 8's rebuild left it in. + + Four ``DROP COLUMN``s rather than a hand-written ``CREATE TABLE``, and that + is not only convenience. These tests compare ``sqlite_master`` *text*, and + SQLite rewrites the stored statement by deleting the dropped column's + definition and leaving every other character alone — so what is left is + exactly what ``table.create()`` wrote, where a retyped baseline would differ + in whitespace and fail for a reason about this file rather than the schema. + + That these drops are even possible is migration 9's own argument restated: + none of the four carries a foreign key, which is why it could be an ``ALTER`` + at all where migration 8 needed a rebuild. + """ + with store.engine.begin() as connection: + for column in _MIGRATION_NINE_COLUMNS: + connection.execute(text(f"alter table ingest_job drop column {column}")) + connection.execute(text("update _visionset_meta set format_version = 8")) + + +def test_migration_nine_gives_a_run_its_progress_and_its_report(tmp_path: Path) -> None: + store = SqliteMetadataStore(tmp_path / "visionset.db") + store.initialize() + with store.engine.connect() as connection: + columns = {c["name"] for c in inspect(connection).get_columns("ingest_job")} + assert {"batch_name", "processed", "total", "failures"} <= columns + store.close() + + +def test_migration_nine_alters_a_table_migration_eight_rebuilt(tmp_path: Path) -> None: + """The ``ALTER`` path, which the fresh-versus-migrated test cannot reach. + + That test walks back to generation 1, from where migration 8 re-creates + ``ingest_job`` whole — including migration 9's columns, since it builds from + ``_tables`` — so migration 9 finds them present and does nothing. A database + this build actually wrote is stamped at 8, and migration 9 reaches it as four + ``ALTER TABLE ... ADD COLUMN`` statements instead. + + That is the path the declared-last rule exists for: SQLite appends an added + column, so the two spellings of ``CREATE TABLE ingest_job`` agree only while + those four stay at the end of ``IngestJobRow``. + """ + fresh = SqliteMetadataStore(tmp_path / "fresh.db") + fresh.initialize() + expected = _schema(fresh) + fresh.close() + + legacy = SqliteMetadataStore(tmp_path / "legacy.db") + legacy.initialize() + _downgrade_to_generation_eight(legacy) + assert _schema(legacy) != expected # the four columns really are gone + + legacy.initialize() # migration 9 alone, as an ALTER + assert _schema(legacy) == expected + assert legacy.format_version == FORMAT_VERSION + legacy.close() + + +def test_migration_nine_keeps_the_runs_a_workspace_already_recorded(tmp_path: Path) -> None: + """Four columns with an honest value for a row written before them. + + Nothing is refused and nothing is dropped here, unlike migrations 6 to 8: + a pre-#19 run counted nothing and reported nothing, which is exactly what + ``0`` and ``[]`` say, and NULL is what a run that named no batch meant. + """ + store = SqliteMetadataStore(tmp_path / "visionset.db") + store.initialize() + with store.engine.begin() as connection: + connection.execute(text("insert into workspace (id, name) values ('w', 'ws')")) + connection.execute( + text("insert into project (id, workspace_id, name) values ('p', 'w', 'proj')") + ) + connection.execute( + text( + "insert into source (id, project_id, kind, path, registered_at, capture_params) " + "values ('s', 'p', 'image_directory', '/in', '2026-07-27T00:00:00+00:00', '{}')" + ) + ) + _downgrade_to_generation_eight(store) + with store.engine.begin() as connection: + connection.execute( + text("insert into ingest_job (id, source_id, state) values ('j', 's', 'completed')") + ) + + store.initialize() + + with store.engine.connect() as connection: + row = connection.execute( + text("select batch_name, processed, total, failures from ingest_job where id = 'j'") + ).one() + assert row == (None, 0, None, "[]") + store.close() + + def test_a_fresh_database_and_a_migrated_one_have_the_same_schema(tmp_path: Path) -> None: """Migration 1 is ``create_all`` of *current* metadata, so the two paths differ.