From 0c96079b5334cf15ded463b1f455ca224c429b1a Mon Sep 17 00:00:00 2001 From: Armando Anaya Date: Mon, 27 Jul 2026 04:46:43 -0700 Subject: [PATCH] =?UTF-8?q?feat(kernel):=20ingest=20pipeline=20=E2=80=94?= =?UTF-8?q?=20hashing,=20blob=20dedup,=20asset=20origin,=20batch=20materia?= =?UTF-8?q?lization=20(#20)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `IngestService` is the one door that turns a registered source into rows: it hashes every item, stores the bytes once, records what the decoder made of them, and puts the result in a draft batch. It closes the last write in the kernel that had no service behind it — `_add_assets` is gone from the SDK example, which now registers a directory and ingests it like a real caller. - One `ingest(source_id, *, batch_id=None, batch_name=None)`, branching on `SourceKind`: registration needed two methods because its arguments differed, this does not — the source already carries the kind, the path and the rate. - Identity is content, origin is provenance. The same bytes in two sources are one blob and one asset, and the asset keeps the origin of the first sighting. - Four transactions with the decode outside all of them, and blobs written before any row: an out-of-process decoder inside a write transaction is how a single-writer SQLite store starts reporting "database is locked". - Failure splits by remedy: unsupported and corrupt items are reported per file and the run carries on; a missing ffmpeg fails the job and is re-raised. - First emitter of `IngestCompleted`, flipping the tripwire held since #13. Migration 8 (`FORMAT_VERSION` 8) gives `asset` its format and origin, links a run to its batch, and adds the two unique indexes both rules had been running without — `uq_asset_project_content_hash`, and the expression index `uq_source_project_kind_path_fps` that #18 named as owed. `asset` is altered rather than rebuilt (four cascading children, and pre-existing rows that were legitimate); `ingest_job` is rebuilt so its new key can match `create_all`. --- docs/README.md | 3 +- docs/events.md | 11 +- docs/examples.md | 34 +- docs/ingest.md | 123 +++ docs/media.md | 18 +- docs/persistence.md | 51 +- docs/sources.md | 41 +- examples/sdk_end_to_end.py | 96 +- src/visionset/kernel/__init__.py | 2 + src/visionset/kernel/adapters/_tables.py | 102 +++ src/visionset/kernel/adapters/migrations.py | 128 ++- src/visionset/kernel/domain/__init__.py | 11 +- src/visionset/kernel/domain/asset.py | 41 +- src/visionset/kernel/domain/ingest.py | 99 ++- src/visionset/kernel/errors.py | 9 + src/visionset/kernel/services/__init__.py | 2 + .../kernel/services/batch_service.py | 17 +- .../kernel/services/ingest_service.py | 411 +++++++++ .../kernel/services/source_service.py | 20 +- tests/examples/test_sdk_end_to_end.py | 16 +- tests/kernel/test_events.py | 11 +- tests/kernel/test_ingest_service.py | 832 ++++++++++++++++++ tests/kernel/test_migrations.py | 173 ++++ 23 files changed, 2133 insertions(+), 118 deletions(-) create mode 100644 docs/ingest.md create mode 100644 src/visionset/kernel/services/ingest_service.py create mode 100644 tests/kernel/test_ingest_service.py diff --git a/docs/README.md b/docs/README.md index 673e2b1f..8efdbed7 100644 --- a/docs/README.md +++ b/docs/README.md @@ -9,6 +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 | | [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 | @@ -18,4 +19,4 @@ contracts (kernel purity, headless annotator) are described there and enforced i | [releases.md](releases.md) | The immutable artifact: what a manifest is and is not, why two publishes agree byte for byte, hash verification, and the seeded split recipe | | [events.md](events.md) | Domain events: subscribing by type, why emission follows the commit, at-most-once delivery, and what an isolated subscriber failure does | | [persistence.md](persistence.md) | The metadata store: repositories, unit of work, table layout, migrations and `format_version` | -| [examples.md](examples.md) | The runnable end-to-end example: the whole cycle in one pass, what it is built to demonstrate, and the one step that has no service yet | +| [examples.md](examples.md) | The runnable end-to-end example: the whole cycle in one pass, and what it is built to demonstrate | diff --git a/docs/events.md b/docs/events.md index 6c0efac7..7720a657 100644 --- a/docs/events.md +++ b/docs/events.md @@ -94,7 +94,7 @@ custom encoder — which is what makes a webhook a subscriber rather than a rewr | `BatchCompleted` | `BatchService.complete` | `batch_id`, `project_id`, `asset_count` | | `AnnotationsWritten` | `AnnotationService.add` / `update` / `delete` | `job_id`, `batch_id`, `operation`, `asset_ids`, `annotation_ids` | | `ReleasePublished` | `ReleaseService.publish` | `release_id`, `dataset_id`, `project_id`, `tag`, `manifest_hash`, `schema_version`, `asset_count`, `annotation_count` | -| `IngestCompleted` | nobody yet — M2 | `ingest_job_id`, `project_id`, `source_id`, `asset_count` | +| `IngestCompleted` | `IngestService.ingest` | `ingest_job_id`, `project_id`, `source_id`, `asset_count` | `AnnotationsWritten` is one per **call**, not one per box: the three writes are all-or-nothing over a whole payload, so one call is one thing that happened. Its `asset_ids` are deduplicated — @@ -105,10 +105,11 @@ out of the blob store without being handed it. It is also why publishing writes [dataset change-log](datasets.md) entry: the log records mutations of the trunk, and publishing mutates nothing in it. "A release happened" is an event. -`IngestCompleted` is declared and emitted by nothing. Ingest is M2's; the vocabulary was settled -in one pass so that a subscriber written today already compiles against the shape it will be -handed, and a test asserts nothing in M1 emits it — so it cannot quietly acquire a caller before -M2 wires one deliberately. +`IngestCompleted` was declared in M1 and emitted by nothing until [ingest](ingest.md) wired it — +the vocabulary was settled in one pass so a subscriber written then already compiled against the +shape it would be handed, and a test held the line until the emitter arrived deliberately. Its +`asset_count` is what the run **put in the batch**: assets it created plus assets whose content +the project already held. A run that fails announces nothing at all. ### `name`, and why it is two types diff --git a/docs/examples.md b/docs/examples.md index be746e7c..2e5082f9 100644 --- a/docs/examples.md +++ b/docs/examples.md @@ -21,8 +21,9 @@ as a plain script, which is the only way to prove it still works from a clean ch | Subscribe | `event_bus.subscribe(DomainEvent, ...)` — the catch-all, matched by type | [events.md](events.md) | | Project | `ProjectService.create` — the 1:1 dataset is created in the same transaction | [projects.md](projects.md) | | Schema | `SchemaService.create_version` — version 1 of the labeling contract | [schemas.md](schemas.md) | -| Assets | six generated PNGs into the blob store, then the rows that name them | *nothing yet — see below* | -| Batch | `BatchService.create`, then `approve(BySize(size=3))` → 2 jobs, schema pinned | [batches.md](batches.md) | +| Source | six generated PNGs written to `incoming/`, registered as an image directory | [sources.md](sources.md) | +| Assets | `IngestService.ingest` — hashed, stored once, and put in a draft batch | [ingest.md](ingest.md) | +| Batch | `approve(BySize(size=3))` → 2 jobs, schema pinned | [batches.md](batches.md) | | Work | `JobService.start` / `next_pending` / `mark`, one asset deliberately skipped | [jobs.md](jobs.md) | | Labels | `AnnotationService.add` — a box, a polygon and a whole-frame tag per asset | [annotations.md](annotations.md) | | Trunk | `DatasetService.promote` — five assets, not six | [datasets.md](datasets.md) | @@ -49,24 +50,27 @@ keys on `content_hash`, and the frames are generated deterministically from thei same pictures land in the same folds on every machine. The smoke test asserts exactly that — comparing folds by content hash, never by id. -## The one place it reaches below a service +## Every step goes through the service that owns it -Creating an `Asset` has no door yet. `SourceService` (#18) has landed, but it registers *where* -data comes from, not the assets themselves; the pipeline that hashes, deduplicates, extracts -dimensions and materializes assets into a batch is #20, with `IngestJob` (#19) around it. Until -#20 lands, `_add_assets` writes, by hand, the row that ingest will write: +That sentence used to carry an exception. Creating an `Asset` had no door, so the example wrote +the row by hand — through the same public port a service uses, commented as the one place it +reached below one, and flagged in three files as something #20 would delete. It did: ```python -hashes = [workspace.blob_store.put(BytesIO(frame_bytes(i))) for i in range(count)] -with workspace.unit_of_work() as uow: - uow.assets.add(Asset(project_id=..., content_hash=..., uri=..., width=..., height=...)) +incoming = _write_frames(dest / "incoming", FRAME_COUNT) +source = sources.register_images(project.id, incoming) +ingested = ingest.ingest(source.id, batch_name="batch-001") ``` -This is the only step in the file that does not go through the service that owns its entity, it -is commented as such where it lives, and it disappears when #20 lands. The blobs are written -before the transaction opens on purpose: `BlobStore.put` is not transactional and a rollback -cannot unwrite it — but a blob nothing points at is harmless (content-addressed, deduplicated, -and never deleted), while a row pointing at bytes that were never stored would not be. +The frames go to disk, the folder is registered as an origin, and [ingest](ingest.md) hashes +them, stores the bytes once and puts them in a draft batch — which is what a real caller does with +a real folder of photographs. `BatchService.create` disappeared from the example along with it: +the ingest is what makes the batch now. + +The frames are still generated by the example's own six-line PNG encoder rather than by Pillow, +which is a dependency these days. Their bytes are fixed, and that is the point: an asset's +identity is the SHA-256 of its content and the split keys on content hash, so the same pictures +land in the same folds on every machine. ## Why three classes for "two classes" diff --git a/docs/ingest.md b/docs/ingest.md new file mode 100644 index 00000000..5a44773b --- /dev/null +++ b/docs/ingest.md @@ -0,0 +1,123 @@ +# Ingest + +Where a registered [source](sources.md) becomes rows. `IngestService` hashes every item, stores +the bytes once, records what the decoder made of them, and puts the result in a draft +[batch](batches.md) somebody can approve. Nothing else in the kernel creates an `Asset` — +`examples/sdk_end_to_end.py` used to, and no longer does. + +```python +source = sources.register_images(project.id, Path("~/dashcam/monday").expanduser()) +result = ingest.ingest(source.id, batch_name="monday") + +result.created # assets new to this project +result.deduplicated # items whose bytes the project already held +result.failures # one line per item that could not be read at all +``` + +## One `ingest`, where registration has two methods + +`SourceService` split `register_images` from `register_video` because their arguments genuinely +differ: a clip needs a decomposition rate and gets probed, a directory needs neither. Here they do +not differ at all. The source already carries its kind, its path and its rate, so the caller +passes one id and the service branches on `SourceKind`. A second entry point would ask callers to +re-state what the source already knows. + +## Identity is content; origin is provenance + +An asset **is** its bytes: `content_hash` is the SHA-256 of the file, and the same bytes ingested +twice are one asset over one blob. `uq_asset_project_content_hash` is the index under that rule. +Per project, not global — two projects ingesting one photograph are two assets sharing one blob, +which is exactly what makes `project_id` the asset's parent. + +`source_id`, `frame_index` and `frame_timestamp` are a different kind of fact. They record where +the bytes were **first** seen, and a second sighting never rewrites them — the rule +`Source.registered_at` already follows. One image in two registered folders keeps the first +folder's path in its `uri` and the first source's id in its origin, because the alternative is an +asset whose recorded origin depends on which ingest happened to run last. + +The deliberate consequence: an asset records **one** origin, and a duplicate across two sources +loses the second. A join table is the honest upgrade if that information is ever wanted; it would +be its own migration and is not needed by anything today. + +Re-running an ingest is therefore not an error and not a no-op worth avoiding: it creates nothing, +reports every item as deduplicated, and is how a folder that grew by three files is caught up. + +## What the two paths do + +| | image directory | video | +| --- | --- | --- | +| what is read | every file at the **top level**, in filename order | one frame per extraction slot, at the rate the source records | +| decoded by | `ImageProcessor.probe` — dimensions and format from the bytes | `VideoProcessor.frames` — ffmpeg, deterministic within one build | +| `uri` | the file's absolute path | `/path/clip.mp4#frame=7` | +| frame position | none | `frame_index` and `frame_timestamp` | +| damage | one report line per file; the run carries on | the frames ffmpeg managed are kept, plus one report line | + +Subdirectories are stepped over and recorded nowhere. Recursion is not a per-run option but a +question about what *the source is* — "the same source yields the same assets" — so it belongs to +a future `register_images(..., recursive=True)` rather than here, where it would silently change +what an already-registered source means. There is 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. + +**Frames are not re-probed.** `VideoProcessor` guarantees every frame is a complete image in +`FRAME_FORMAT` at the dimensions `probe` reported, and that promise is asserted in the port's own +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. + +## 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`. +2. **No transaction.** Decode, hash and `BlobStore.put` every item. +3. Write the asset rows, reusing whatever content the project already holds. +4. Put them in the batch (through `BatchService`), then mark the job `completed`. + +Then, and only after the last block has exited, `IngestCompleted` goes on the bus — the rule every +emitter in this kernel follows. + +Step 2 is outside a transaction because 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". The blob writes are 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 the job record is for. + +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. + +## Failure splits by remedy, not by severity + +A file that is not an image, or one whose bytes will not decode, is *reported*: one +`IngestFailure` carrying the item's name, the reason, and which of the two it was. The run carries +on, because an operator with five thousand files needs the other four thousand nine hundred. The +name and the reason are kept apart so a report renders as a table rather than as a list of +sentences, and `IngestFailureKind` exists so it can be **grouped** — real data loss must not be +buried under ordinary operator noise. + +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. + +## The target batch + +With no `batch_id`, the run creates a draft named `batch_name` or, failing that, after the +source's own file or folder. With one, that batch must still be a draft — checked **before** +anything is decoded, because finding out afterwards would mean finding out after the work. + +Membership is everything the run ingested, deduplicated assets included: a duplicate is not new +data, but it is part of what the run was asked to gather. Order is ingest order, which is filename +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. diff --git a/docs/media.md b/docs/media.md index 64891d18..5e33df41 100644 --- a/docs/media.md +++ b/docs/media.md @@ -333,12 +333,12 @@ kernel — FastAPI, Typer, MCP, uvicorn — not third-party libraries, and ffmpe ## What is deliberately not here yet -- **No `Asset` field.** `ImageMetadata` and `VideoMetadata` are returned, not stored; putting - `format` and origin on the asset row belongs with the ingest pipeline. -- **No blob write.** `thumbnail()` and `frames()` hand back bytes. Storing them - content-addressed, recording a `thumbnail_hash` and writing a frame's `index`/`timestamp` onto - an asset are the ingest and thumbnail-cache tasks. - -`Source` used to be on that list and no longer is: registering a clip records its original rate -and the decomposition parameters chosen for it, built on `VideoMetadata` exactly as anticipated. -See [sources.md](sources.md). +- **No thumbnail write.** `thumbnail()` hands back bytes. Storing them content-addressed and + recording an `asset.thumbnail_hash` is the thumbnail-cache task, for the M5 gallery. + +Two things used to be on that list and no longer are. `Source` came off it with registration, +which records a clip's original rate and the decomposition parameters chosen for it, built on +`VideoMetadata` exactly as anticipated — see [sources.md](sources.md). The `Asset` fields came off +it with [ingest](ingest.md): what a probe reported is now stored as `asset.format`, and a frame's +`index`/`timestamp` land on the asset as `frame_index`/`frame_timestamp` beside the source it was +cut from. Both ports are called from exactly one place, and that is where. diff --git a/docs/persistence.md b/docs/persistence.md index 9b3f6ef7..4567a143 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -47,8 +47,16 @@ never raises `ProjectNameTaken`. But a rule with no backstop is a wish, so the store carries the constraint too: `uq_project_workspace_name` on `project (workspace_id, name COLLATE NOCASE)`, alongside -`uq_schema_project_version` and `uq_member_dataset_asset`. The invariant then survives a -service bug, a forgotten code path, and a second process. +`uq_schema_project_version`, `uq_member_dataset_asset`, `uq_release_dataset_tag`, +`uq_asset_project_content_hash` and `uq_source_project_kind_path_fps`. The invariant then survives +a service bug, a forgotten code path, and a second process. + +The last of those is the only index here whose terms are not all columns: its fourth is +`coalesce(json_extract(video, '$.extraction_fps'), 0)`. SQLite treats NULLs in a unique index as +distinct, so a nullable column would let every image directory collide with nothing at all — and +an index is not a query, so no service gains a JSON path from it. It is also the one index that +cannot use `checkfirst`, because SQLAlchemy cannot *reflect* an expression-based index; migration +008 issues `CREATE UNIQUE INDEX IF NOT EXISTS` compiled from the same `Index` object instead. The two layers reach different distances on purpose. `COLLATE NOCASE` folds ASCII only; `WorkspaceService` compares with Unicode `casefold` over an NFC-normalized, stripped string. @@ -81,6 +89,16 @@ Foreign keys are declared `ON DELETE CASCADE` — and the store issues `PRAGMA foreign_keys = ON` for every connection, because SQLite ships with foreign keys **off**. Without that pragma every constraint here would be decorative. +`CASCADE` is the rule because the child is normally *part of* the parent. `ingest_job.batch_id` +is the exception and states the other case: a run is a record of work done, not a child of the +batch it filled, so deleting the batch nulls the link rather than erasing the run. The same +argument applies to `asset.source_id` — a source is a receipt and an asset is data — except that +one is not a foreign key at all: SQLite spells a key added by `ALTER TABLE` inline on the column +while `create_all` spells one as a table constraint, the two texts differ, and `asset` is the one +table that could not be rebuilt to escape that (four cascading children, and rows that were +legitimately already there). It is documented on the column, and a future `SourceService.delete` +has to clear it by hand. + ## Migrations and `format_version` There is no alembic. A local-first, single-file, single-writer store does not need a @@ -96,8 +114,9 @@ MIGRATIONS: list[Migration] = [ Migration(version=5, name="annotation_attributes", upgrade=...), Migration(version=6, name="release_manifest_pointer", upgrade=...), Migration(version=7, name="source_provenance", upgrade=...), + Migration(version=8, name="ingest_pipeline", upgrade=...), ] -FORMAT_VERSION: int = MIGRATIONS[-1].version # 7 +FORMAT_VERSION: int = MIGRATIONS[-1].version # 8 ``` `initialize()` reads the version stamped in `_visionset_meta` and runs whatever is @@ -139,12 +158,24 @@ one, where `batch_asset.position`, which was there from migration 001, does not. And a column arriving by `ALTER` must be declared **last** on its row class, because SQLite appends it: `batch.schema_version`, `annotation_job_asset.position` and `annotation.attributes` -all sit at the end of their tables for that reason alone. `source` is exempt — migration 007 -rebuilds it from `_tables` rather than altering it, so both paths run the same `CREATE TABLE` and -the rule has nothing to bite on. Declared anywhere else, the +all sit at the end of their tables for that reason alone, and migration 008 put +`asset.format`, `asset.source_id`, `asset.frame_index` and `asset.frame_timestamp` there too. +`source` is exempt — migration 007 rebuilds it from `_tables` rather than altering it, so both +paths run the same `CREATE TABLE` and the rule has nothing to bite on — but **that exemption +expires for any column added after 007**, because a database this build wrote is already stamped +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. +**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. +Migration 008 met this twice and answered it twice: `ingest_job` was **rebuilt** so `batch_id` +could keep a real key (no children, provably empty), while `asset.source_id` was added without +one, because `asset` has four cascading children and rows that were legitimately already there. +SQLite also refuses to *drop* such a column, which is why the undo in +`_downgrade_to_version_one` rebuilds `ingest_job` rather than altering it. + **Migrations 006 and 007 drop a table**, and the bar they had to clear is worth writing down. `release` gained three `NOT NULL` columns with no honest default — an `ALTER` would have baked `manifest_hash DEFAULT ''` into every fresh database forever — and, decisively, a pre-#12 release @@ -164,6 +195,14 @@ Migration 007 carries one extra obligation that 006 did not. `ingest_job.source_ **silently, without raising**. A rebuild of a table with children has to count the children too. `release` had none, which is why the precedent alone was not enough. +Migration 008 is the worked example of the **other** answer to that question. `asset` has four +cascading children *and* rows that are perfectly legitimate — M1's example wrote assets through +the same public port a service uses — so there is nothing to refuse and nothing that could be +dropped. It alters instead, and every column it adds is nullable and honest: NULL means "this row +predates the ingest pipeline", where a `server_default` would invent a format nobody probed. What +it does refuse is data its two new unique indexes cannot accept, counted before either index is +created so an `IntegrityError` never escapes `initialize()`. + 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`, diff --git a/docs/sources.md b/docs/sources.md index 87eaa806..46c98e36 100644 --- a/docs/sources.md +++ b/docs/sources.md @@ -80,16 +80,26 @@ Two things the key deliberately leaves out: That second rule has a corollary worth knowing: re-registering an already-known clip still needs ffmpeg, because the fresh probe is what keeps the record honest. -### The gap this leaves, and when to close it - -`docs/persistence.md` says a rule with no backstop is a wish, and every other uniqueness rule in -this store has a unique index behind it. This one does not. Two concurrent registrations of one -folder can both pass the pre-check and both insert. - -That is tolerated today because no row references a source, so a duplicate is inert. It stops -being tolerable when ingest gives `asset.source_id` a target and the winner of a race starts -deciding an asset's recorded origin — **that is when this needs an index under it**. It sits -alongside the store's other known concurrency gap, the untranslated `OperationalError`. +### The gap this left, and how it was closed + +This rule shipped without a backstop, which `docs/persistence.md` calls a wish: two concurrent +registrations of one folder could both pass the pre-check and both insert. It was tolerable only +because nothing referenced a source, so a duplicate was inert — and [ingest](ingest.md) ended +that, by giving `asset.source_id` a target and letting the winner of a race decide an asset's +recorded origin. + +`uq_source_project_kind_path_fps` went in with it, over +`(project_id, kind, path, coalesce(json_extract(video, '$.extraction_fps'), 0))`. The fourth term +is an expression rather than a column, and it is `coalesce`d rather than left to be NULL, because +SQLite treats NULLs in a unique index as **distinct** — an image directory, whose `video` is NULL, +would otherwise never collide with itself, which is most of what the index is for. `0` cannot be +mistaken for a real rate: `extraction_fps` is `gt=0`. + +The two layers do what they do everywhere else in this store. The pre-check is what produces a +friendly answer; the index is the guarantee. A caller that loses the race sees a raw +`ConstraintViolated`, and the remedy is to call the same method again, which finds the winner's +row and returns it. The store's other known concurrency gap — the untranslated `OperationalError` +— is still open. ## Paths are canonicalized once @@ -116,11 +126,14 @@ successful registration as proof the file will decode is wrong. ## What is deliberately not here yet -- **No delete.** A source disappears with its project's cascade and no sooner. Nothing yet - references one, so there is no orphan to reason about; when ingest gives `asset.source_id` a - target, deletion becomes a real question with a real answer. +- **No delete.** A source disappears with its project's cascade and no sooner. The question ingest + raised does now have an answer, though: an asset **outlives** the receipt it came from. Deleting + a source must not take the asset, its annotations, its dataset membership or the releases naming + it, so a future `SourceService.delete` clears `asset.source_id` rather than cascading through it + — and it has to do that itself, because that column is deliberately not a foreign key (see + `adapters/_tables.py` for why, and what it costs). - **No event.** Registering a source announces nothing. `IngestCompleted` is the event this area - will emit, and the ingest pipeline owns it. + emits, and [the ingest pipeline](ingest.md) owns it. - **No remote kinds.** `SourceKind` has two members and grows by a deliberate kernel change with a service method behind it — see the enum's own docstring for why it is an enum where `DatasetChange.operation` is a plain `str`. diff --git a/examples/sdk_end_to_end.py b/examples/sdk_end_to_end.py index 5d31947d..d1701f59 100644 --- a/examples/sdk_end_to_end.py +++ b/examples/sdk_end_to_end.py @@ -10,16 +10,17 @@ uv run python examples/sdk_end_to_end.py [DESTINATION] The images are generated here at runtime, from the standard library alone. That -is not a convenience: VisionSet never commits fixture media, and M1 has no image -library to lean on (Pillow arrives with the media processor in M2, #16). - -**One call in here reaches below a service, on purpose.** Creating an ``Asset`` -has no door yet. #18's ``SourceService`` registers *where* data comes from, not -the assets themselves; the pipeline that hashes, deduplicates and materializes -into a batch is #20, with #19's ``IngestJob`` around it. Until #20 lands, -``_add_assets`` writes the row that ingest would write, through the same public -port a service uses. Every other step below goes through the service that -owns it, which is how the rest of the SDK is meant to be used. +is not a convenience: VisionSet never commits fixture media, and an example that +makes its own pixels needs no fixture at all. Pillow is a dependency now, and +this deliberately still does not use it — the encoder below is six lines and its +bytes are fixed, which is what keeps the split folds the same on every machine. + +**Every step now goes through the service that owns it.** That sentence used to +carry an exception: creating an ``Asset`` had no door, so this file wrote the row +by hand through the same public port a service uses. #20's ``IngestService`` +closed it. The frames are written to disk, the directory is registered as a +source, and the ingest hashes them, stores them once and puts them in a batch — +which is what a real caller does with a real folder of photographs. """ from __future__ import annotations @@ -29,7 +30,6 @@ import sys import zlib from dataclasses import dataclass -from io import BytesIO from pathlib import Path from uuid import UUID @@ -57,10 +57,12 @@ AnnotationService, BatchService, DatasetService, + IngestService, JobService, ProjectService, ReleaseService, SchemaService, + SourceService, WorkspaceService, ) @@ -120,6 +122,8 @@ class Summary: project_id: UUID dataset_id: UUID + source_id: UUID + ingest_job_id: UUID schema_version: int asset_ids: tuple[UUID, ...] skipped_asset_id: UUID @@ -177,37 +181,20 @@ def frame_bytes(index: int) -> bytes: return png(FRAME_WIDTH, FRAME_HEIGHT, (40 + index * 30, 90, 200 - index * 20)) -# --- the one write that has no service yet -------------------------------- +# --- offering the frames to the project ----------------------------------- -def _add_assets(workspace: WorkspaceService, project_id: UUID, count: int) -> list[Asset]: - """Put ``count`` generated frames into the workspace and register them. +def _write_frames(directory: Path, count: int) -> Path: + """Put ``count`` generated frames on disk, the way a camera would have. - THE ONE PLACE THIS EXAMPLE REACHES BELOW A SERVICE. M1 has no ingest, so - there is no door to an Asset yet; #20 (M2) builds the pipeline that hashes, - deduplicates, extracts dimensions and materializes assets into a batch, and - this function disappears when it lands. Until then it does by hand exactly - what ingest will do: store the bytes, then record the row that names them. - - The blobs are written *before* the transaction opens. ``BlobStore.put`` is - not transactional and a rollback cannot unwrite it — but a blob nothing - points at is harmless (content-addressed, deduplicated, and never deleted), - while a row pointing at bytes that were never stored would not be. + Named ``frame-000.png`` upward, because ingest reads a directory in filename + order and that order becomes the batch's membership order — so the labels + below can be indexed by position and still mean what they say. """ - hashes = [workspace.blob_store.put(BytesIO(frame_bytes(index))) for index in range(count)] - with workspace.unit_of_work() as uow: - return [ - uow.assets.add( - Asset( - project_id=project_id, - content_hash=content_hash, - uri=f"synthetic://road-signs/frame-{index:03d}.png", - width=FRAME_WIDTH, - height=FRAME_HEIGHT, - ) - ) - for index, content_hash in enumerate(hashes) - ] + directory.mkdir(parents=True, exist_ok=True) + for index in range(count): + (directory / f"frame-{index:03d}.png").write_bytes(frame_bytes(index)) + return directory # --- labels --------------------------------------------------------------- @@ -284,6 +271,8 @@ def main(dest: Path) -> Summary: projects = ProjectService(workspace) schemas = SchemaService(workspace) + sources = SourceService(workspace) + ingest = IngestService(workspace) batches = BatchService(workspace) jobs = JobService(workspace) annotations = AnnotationService(workspace) @@ -302,13 +291,26 @@ def main(dest: Path) -> Summary: schema = schemas.create_version(project.id, CLASSES) _say(f"schema v{schema.version}: {', '.join(c.name for c in schema.classes)}") - # (4) Six generated frames. See the docstring of _add_assets. - assets = _add_assets(workspace, project.id, FRAME_COUNT) - _say(f"{len(assets)} synthetic {FRAME_WIDTH}x{FRAME_HEIGHT} frames stored") - - # (5) A batch is the unit of annotation work. In draft it is still - # editable; membership freezes at approval. - batch = batches.create(project.id, "batch-001", [asset.id for asset in assets]) + # (4) Six generated frames on disk, and the directory holding them + # registered as where this project's data comes from. Registration does + # not walk the folder: what is in it is read at ingest, because a count + # taken now would be stale by the time anything used it. + incoming = _write_frames(dest / "incoming", FRAME_COUNT) + source = sources.register_images(project.id, incoming) + _say(f"source {source.kind.value} registered at {source.path}") + + # (5) Ingest hashes every file, stores the bytes once (content-addressed, + # so re-running this creates nothing), records what the decoder made of + # them, and puts the lot in a draft batch. Membership order is ingest + # order. A batch is the unit of annotation work; in draft it is still + # editable, and membership freezes at approval. + ingested = ingest.ingest(source.id, batch_name="batch-001") + assets = list(ingested.assets) + batch = batches.get(ingested.batch_id) + _say( + f"{ingested.created} assets ingested into batch {batch.name!r} " + f"({ingested.deduplicated} already known, {ingested.failed} unreadable)" + ) # (6) Approval pins the active schema version to the batch forever and # cuts it into jobs. The partition is exact: disjoint, and their union is @@ -376,6 +378,8 @@ def main(dest: Path) -> Summary: return Summary( project_id=project.id, dataset_id=dataset.id, + source_id=source.id, + ingest_job_id=ingested.job_id, schema_version=schema.version, asset_ids=tuple(asset.id for asset in assets), skipped_asset_id=skipped_asset_id, @@ -434,7 +438,7 @@ def _clear_previous_run(dest: Path) -> None: return if not dest.is_dir(): raise SystemExit(f"refusing to run: {dest} exists and is not a directory") - stray = {entry.name for entry in dest.iterdir()} - {"visionset.db", "blobs"} + stray = {entry.name for entry in dest.iterdir()} - {"visionset.db", "blobs", "incoming"} if stray: raise SystemExit( f"refusing to remove {dest}: it holds {', '.join(sorted(stray))}, " diff --git a/src/visionset/kernel/__init__.py b/src/visionset/kernel/__init__.py index b1362e6f..38418d76 100644 --- a/src/visionset/kernel/__init__.py +++ b/src/visionset/kernel/__init__.py @@ -24,6 +24,7 @@ EmptyRelease, EntityAlreadyExists, EntityNotFound, + IngestJobNotFound, InvalidAnnotation, InvalidAttributeValue, InvalidName, @@ -75,6 +76,7 @@ "EmptyRelease", "EntityAlreadyExists", "EntityNotFound", + "IngestJobNotFound", "InvalidAnnotation", "InvalidAttributeValue", "InvalidName", diff --git a/src/visionset/kernel/adapters/_tables.py b/src/visionset/kernel/adapters/_tables.py index 33de1420..7fce96cb 100644 --- a/src/visionset/kernel/adapters/_tables.py +++ b/src/visionset/kernel/adapters/_tables.py @@ -105,6 +105,12 @@ class SourceRow(Base): otherwise emit different DDL. Migration 7 drops and re-creates this table from this class instead, so both paths run the same ``CREATE TABLE`` and the rule has nothing to bite on. + + **That exemption expires for any column added after migration 7.** A + database this build wrote is already stamped at 7 and will never re-run it, + so a later column reaches it by ``ALTER TABLE`` after all and has to be + declared last like everywhere else. Nothing has needed one yet: migration 8 + puts a unique index over this table and adds no column to it. """ __tablename__ = "source" @@ -123,7 +129,47 @@ class SourceRow(Base): video: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) +#: One origin is one source: the backstop under ``SourceService``'s idempotency +#: rule, which shipped in #18 with nothing underneath it and acquired teeth here +#: — see that service's module docstring for why ingest is when it had to. +#: +#: The fourth term is an **expression**, not a column, and it is ``coalesce``d +#: rather than left to be NULL. SQLite treats NULLs in a unique index as +#: distinct, so an image directory — whose ``video`` is NULL — would never +#: collide with itself, which is most of what this index is for. ``0`` cannot be +#: confused with a real rate: ``VideoProvenance.extraction_fps`` is ``gt=0``. +#: +#: This is SQL reading a JSON column, which the module docstring above reserves +#: for values "nothing ever queries". An index is not a query: no service gains +#: a JSON path, ``_source_to_domain`` still rehydrates ``VideoProvenance`` whole, +#: and the doctrine's purpose — no service building SQL over JSON — is intact. +#: The alternative was a redundant ``extraction_fps`` column written by the +#: mapper and read by nobody, which would need this same ``json_extract`` to +#: backfill itself and could later stop being maintained without failing. +SOURCE_ORIGIN_UNIQUE = Index( + "uq_source_project_kind_path_fps", + SourceRow.project_id, + SourceRow.kind, + SourceRow.path, + text("coalesce(json_extract(video, '$.extraction_fps'), 0)"), + unique=True, +) + + 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. + """ + __tablename__ = "ingest_job" id: Mapped[UUID] = mapped_column(SaUuid, primary_key=True) @@ -132,6 +178,13 @@ class IngestJobRow(Base): ) state: Mapped[str] = mapped_column(String, nullable=False) 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 + #: rather than nullable for a migration's convenience. ``SET NULL`` and not + #: ``CASCADE``: deleting the batch does not un-happen the run. + batch_id: Mapped[UUID | None] = mapped_column( + SaUuid, ForeignKey("batch.id", ondelete="SET NULL"), nullable=True + ) class AssetRow(Base): @@ -146,6 +199,55 @@ class AssetRow(Base): uri: Mapped[str] = mapped_column(String, nullable=False) width: Mapped[int | None] = mapped_column(Integer, nullable=True) height: Mapped[int | None] = mapped_column(Integer, nullable=True) + # The four below arrive by ``ALTER TABLE`` in migration 8 and are therefore + # declared last, in the order that migration adds them. #21's + # ``thumbnail_hash`` goes after them, for the same reason. + # + # All four are nullable, and that is not a shortcut. ``asset`` could not be + # rebuilt the way migrations 6 and 7 rebuilt their tables: four tables carry + # ``ON DELETE CASCADE`` keys into it (``batch_asset``, ``annotation``, + # ``dataset_member``, ``annotation_job_asset``), and under + # ``PRAGMA foreign_keys = ON`` a ``DROP TABLE`` runs an implicit ``DELETE`` + # that takes all four silently. Nor could the rows simply be refused: unlike + # a pre-#12 release or a pre-#18 source, an asset written before the ingest + # pipeline is *legitimate* — M1's example wrote them through the same public + # port a service uses. NULL here means "this asset predates the pipeline", + # which is true, where any ``server_default`` would be a fiction. + #: The decoded format, never the filename's suffix. + format: Mapped[str | None] = mapped_column(String, nullable=True) + #: The source these bytes were *first* seen in. + #: + #: **Deliberately not a foreign key**, and it is the only reference in this + #: schema that is not. SQLite spells a key added by ``ALTER TABLE`` inline on + #: the column, while ``create_all`` spells one as a table constraint; the two + #: texts differ, so a fresh database and a migrated one would disagree — and + #: ``asset`` is the one table that cannot be rebuilt to escape that (four + #: cascading children, and rows that are legitimately already there). What + #: would be gained is small: there is no ``SourceService.delete``, and on the + #: one deletion path that does exist — a project's cascade — the asset and + #: the source both die by their own ``project_id`` keys. When a source + #: delete is added it must clear these itself, and say so where it is + #: written. Unindexed too: nothing lists assets by source. + source_id: Mapped[UUID | None] = mapped_column(SaUuid, nullable=True) + #: Position in the *extracted* sequence, for an asset cut out of a clip. + frame_index: Mapped[int | None] = mapped_column(Integer, nullable=True) + #: Seconds into the clip. The locator that survives a different rate. + frame_timestamp: Mapped[float | None] = mapped_column(Float, nullable=True) + + +#: The same bytes are the same asset: the backstop under the ingest pipeline's +#: deduplication, a claim ``Asset``'s docstring has made since M1 with nothing +#: enforcing it. +#: +#: Per project, not global. Two projects ingesting one photograph are two assets +#: over one blob — that is what makes ``project_id`` the parent — and a release +#: keyed on content hash still sees them as the same content. +#: +#: ``ix_asset_content_hash`` stays beside it: neither subsumes the other, and +#: removing an index is its own migration for a benefit nobody has measured. +ASSET_CONTENT_UNIQUE = Index( + "uq_asset_project_content_hash", AssetRow.project_id, AssetRow.content_hash, unique=True +) class BatchRow(Base): diff --git a/src/visionset/kernel/adapters/migrations.py b/src/visionset/kernel/adapters/migrations.py index 9d379827..d2066bf4 100644 --- a/src/visionset/kernel/adapters/migrations.py +++ b/src/visionset/kernel/adapters/migrations.py @@ -29,7 +29,9 @@ does not clear it — and note that under ``PRAGMA foreign_keys = ON``, which this store sets on every connection, ``DROP TABLE`` runs an implicit ``DELETE`` that cascades to children *silently*. A rebuild of a table with children has to count -those too. Migration 7 does; migration 6 had none to count. +those too. Migration 7 does; migration 6 had none to count. Migration 8 is the +worked example of the other answer: ``asset`` has four cascading children *and* +rows that are perfectly legitimate, so it alters and keeps them. Migrations only run forward. A workspace stamped ahead of this build is rejected (``WorkspaceFormatTooNew``) rather than silently downgraded. @@ -42,14 +44,18 @@ from typing import cast from sqlalchemy import Column, Connection, Table, inspect, text -from sqlalchemy.schema import CreateColumn +from sqlalchemy.schema import CreateColumn, CreateIndex from visionset.kernel.adapters._tables import ( + ASSET_CONTENT_UNIQUE, PROJECT_NAME_UNIQUE, + SOURCE_ORIGIN_UNIQUE, AnnotationJobAssetRow, AnnotationRow, + AssetRow, Base, BatchRow, + IngestJobRow, ReleaseRow, SourceRow, ) @@ -222,6 +228,119 @@ def _rebuild_source_with_provenance(connection: Connection) -> None: table.create(connection) +#: What a workspace written before the ingest pipeline may hold that the two new +#: unique indexes will not accept. Raw text, because these are *duplicate* +#: groups rather than rows and the mapped columns cannot express the grouping. +_DUPLICATE_COUNTS = { + "asset": ( + "SELECT count(*) FROM (SELECT project_id, content_hash FROM asset " + "GROUP BY project_id, content_hash HAVING count(*) > 1)" + ), + "source": ( + "SELECT count(*) FROM (SELECT project_id, kind, path, " + "coalesce(json_extract(video, '$.extraction_fps'), 0) AS fps FROM source " + "GROUP BY project_id, kind, path, fps HAVING count(*) > 1)" + ), +} + + +def _rebuild_ingest_job_with_its_batch_link(connection: Connection) -> None: + """Re-create ``ingest_job`` so its new ``batch_id`` can carry a real key. + + The third rebuild in this file, and the cheapest: no children to cascade to, + and nothing has ever written a row here. Migration 7 refuses any workspace + that holds one, and no build before this one had an ``IngestService``, so at + this point the table is empty on every path. "Nothing could" is a claim about + a build rather than about a file, so it is counted, and a workspace that + somehow holds one is refused rather than quietly emptied. + """ + # ``__table__`` is declared as the general ``FromClause``; for a mapped class + # it is always the ``Table``, which is what ``drop``/``create`` need. + table = cast(Table, IngestJobRow.__table__) + stored = {existing["name"] for existing in inspect(connection).get_columns(table.name)} + if "batch_id" in stored: + return + if connection.execute(text("SELECT count(*) FROM ingest_job")).scalar_one(): + raise WorkspaceCorrupt( + "this workspace holds ingest job rows written before IngestService existed. They " + "record no batch, and the assets they claim to have produced cannot be identified; " + "there is nothing to migrate them to, so run the ingests again instead." + ) + table.drop(connection) + table.create(connection) + + +def _add_ingest_origin_and_uniqueness(connection: Connection) -> None: + """Give an asset its origin, a run its batch, and both rules an index. + + **``asset`` is altered and ``ingest_job`` is rebuilt**, and the split is not + arbitrary. A key added by ``ALTER TABLE ... ADD COLUMN`` is spelled inline on + the column, while ``create_all`` spells one as a table constraint: the two + texts differ, so any column carrying a foreign key has to arrive by a + rebuild or not carry one at all. ``ingest_job.batch_id`` needs its key — + batches *are* deleted, and a run pointing at one that is gone is a lie — and + that table has no children and is provably empty here, so it is rebuilt (and + counted rather than assumed, on migrations 6 and 7's terms). + ``asset.source_id`` cannot have either: ``asset`` has four ``ON DELETE + CASCADE`` children (``batch_asset``, ``annotation``, ``dataset_member``, + ``annotation_job_asset``) and under ``PRAGMA foreign_keys = ON`` a + ``DROP TABLE`` runs an implicit ``DELETE`` that takes all four *silently*, + and unlike a pre-#12 release or a pre-#18 source its rows are perfectly + legitimate — M1's example wrote them through the same public port a service + uses. So it is altered, and the column is a plain reference; see + ``_tables.AssetRow`` for what that costs and why it is little. + + Every added column is nullable and honest: NULL means "this row predates the + ingest pipeline", which is true, where a ``server_default`` would invent a + format nobody probed. + + Idempotent the way 3 to 5 are for the columns (the inspector check inside + ``_add_column`` *is* the ``checkfirst``), the way 6 and 7 are for the rebuild + (an inspector check ahead of it), and the way 2 is for the indexes, which + share one object each with ``_tables`` rather than restating DDL. + + **Two duplicate pre-checks, because an index can fail on data already on + disk.** Both uniqueness rules existed before this migration as service-level + pre-checks with nothing underneath them — ``docs/persistence.md`` calls such + a rule a wish — so a workspace written by an earlier build may hold rows the + index refuses. It is refused with a sentence naming both counts rather than + letting an ``IntegrityError`` escape from ``initialize()``. Nothing is + dropped on that path, so it is a stop rather than a rescue: duplicate assets + cannot be merged automatically, because each may carry its own annotations. + """ + counts = { + name: connection.execute(text(query)).scalar_one() + for name, query in _DUPLICATE_COUNTS.items() + } + if any(counts.values()): + raise WorkspaceCorrupt( + f"this workspace holds {counts['asset']} group(s) of duplicate assets (one project, " + f"one content hash) and {counts['source']} group(s) of duplicate sources (one " + "project, kind, path and extraction rate). Ingest makes content the identity of an " + "asset and an origin the identity of a source, so neither set can be kept as it " + "stands; each duplicate may carry its own annotations, so merge or remove them by " + "hand first." + ) + # ``.c`` is typed as the generic column collection; each entry is a real + # ``Column``, which is what ``CreateColumn`` needs. + for column in ( + AssetRow.__table__.c.format, + AssetRow.__table__.c.source_id, + AssetRow.__table__.c.frame_index, + AssetRow.__table__.c.frame_timestamp, + ): + _add_column(connection, cast(Column[object], column)) + _rebuild_ingest_job_with_its_batch_link(connection) + ASSET_CONTENT_UNIQUE.create(connection, checkfirst=True) + # Not ``checkfirst``, unlike every other index here, and the reason is worth + # the two lines: ``checkfirst`` asks the inspector, and SQLAlchemy cannot + # reflect an **expression-based** index — it skips it with a warning, reports + # the index as absent and re-issues the ``CREATE``, which then fails on + # every fresh database. ``IF NOT EXISTS`` asks SQLite instead. The DDL is + # still compiled from the one object in ``_tables``, so nothing is restated. + connection.execute(CreateIndex(SOURCE_ORIGIN_UNIQUE, if_not_exists=True)) + + MIGRATIONS: list[Migration] = [ Migration(version=1, name="initial_schema", upgrade=_create_initial_schema), Migration( @@ -254,6 +373,11 @@ def _rebuild_source_with_provenance(connection: Connection) -> None: name="source_provenance", upgrade=_rebuild_source_with_provenance, ), + Migration( + version=8, + name="ingest_pipeline", + upgrade=_add_ingest_origin_and_uniqueness, + ), ] FORMAT_VERSION: int = MIGRATIONS[-1].version diff --git a/src/visionset/kernel/domain/__init__.py b/src/visionset/kernel/domain/__init__.py index eb5f8bba..10d52966 100644 --- a/src/visionset/kernel/domain/__init__.py +++ b/src/visionset/kernel/domain/__init__.py @@ -32,7 +32,13 @@ Geometry, PolygonGeometry, ) -from visionset.kernel.domain.ingest import IngestJob, IngestState +from visionset.kernel.domain.ingest import ( + IngestFailure, + IngestFailureKind, + IngestJob, + IngestResult, + IngestState, +) from visionset.kernel.domain.media import ( ImageFormat, ImageMetadata, @@ -130,7 +136,10 @@ "ImageFormat", "ImageMetadata", "IngestCompleted", + "IngestFailure", + "IngestFailureKind", "IngestJob", + "IngestResult", "IngestState", "LabelClass", "Manifest", diff --git a/src/visionset/kernel/domain/asset.py b/src/visionset/kernel/domain/asset.py index 47f31b77..ec2c9490 100644 --- a/src/visionset/kernel/domain/asset.py +++ b/src/visionset/kernel/domain/asset.py @@ -5,7 +5,9 @@ from typing import Literal from uuid import UUID, uuid4 -from pydantic import BaseModel, Field, field_validator +from pydantic import BaseModel, Field, field_validator, model_validator + +from visionset.kernel.domain.media import ImageFormat _SHA256_HEX = re.compile(r"^[0-9a-f]{64}$") @@ -18,6 +20,19 @@ class Asset(BaseModel): is expressed in the asset's native reference frame — pixels for images. Only the "image" modality exists today; the field is typed to extend (video, pointcloud, ...) without changing the wire format. + + **Origin is provenance, not identity.** ``source_id`` and the frame fields + record where these bytes were *first* seen, and are never rewritten when the + same content arrives again — the rule ``Source.registered_at`` already + follows. One image appearing in two registered directories is one asset with + one origin, because the alternative is an asset whose recorded origin + depends on which ingest happened to run last. + + Every origin field is optional, and each for its own reason. + ``frame_index``/``frame_timestamp`` exist only for an asset cut out of a + clip. ``format`` and ``source_id`` are absent on a row written before the + ingest pipeline existed, and no default could be honest: the store cannot + invent a format nobody probed. """ id: UUID = Field(default_factory=uuid4) @@ -27,6 +42,14 @@ class Asset(BaseModel): uri: str width: int | None = Field(default=None, ge=1) height: int | None = Field(default=None, ge=1) + #: What the bytes turned out to be, as the ``ImageProcessor`` read them. + format: ImageFormat | None = None + #: The registered origin these bytes were first seen in. + source_id: UUID | None = None + #: Position in the *extracted* sequence, for an asset cut out of a clip. + frame_index: int | None = Field(default=None, ge=0) + #: Seconds into the clip — the locator that survives a re-decomposition. + frame_timestamp: float | None = Field(default=None, ge=0) @field_validator("content_hash") @classmethod @@ -34,3 +57,19 @@ def _content_hash_is_sha256_hex(cls, value: str) -> str: if not _SHA256_HEX.fullmatch(value): raise ValueError("content_hash must be 64 lowercase hex chars (SHA-256)") return value + + @model_validator(mode="after") + def _frame_origin_is_whole(self) -> Asset: + """A frame is located by an index *and* a timestamp, inside a source. + + Half a locator is worse than none. An index with no timestamp cannot be + matched against a re-decomposition at another rate; a timestamp with no + index cannot be ordered; and either without a ``source_id`` names a + position in a clip nobody can point at. + """ + missing_index = self.frame_index is None + if missing_index != (self.frame_timestamp is None): + raise ValueError("frame_index and frame_timestamp are given together or not at all") + if not missing_index and self.source_id is None: + raise ValueError("a frame origin needs the source it was extracted from") + return self diff --git a/src/visionset/kernel/domain/ingest.py b/src/visionset/kernel/domain/ingest.py index 0a485518..1ba0c05c 100644 --- a/src/visionset/kernel/domain/ingest.py +++ b/src/visionset/kernel/domain/ingest.py @@ -1,10 +1,28 @@ # usage: from visionset.kernel.domain import IngestJob +"""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. +""" + from __future__ import annotations from enum import StrEnum from uuid import UUID, uuid4 -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field + +from visionset.kernel.domain.asset import Asset class IngestState(StrEnum): @@ -21,3 +39,82 @@ class IngestJob(BaseModel): 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 + + +class IngestFailureKind(StrEnum): + """Why one item did not become an asset, split by what to do about it. + + An enum rather than a plain ``str``, on exactly ``SourceKind``'s terms: the + set is closed, no writer outside this build produces a value, and the kernel + branches on it. What makes it worth a type at all is that a report has to be + **grouped**, not read — ``CorruptMedia``'s docstring is explicit that a + report unable to separate the two would bury real data loss under ordinary + operator noise, and a reason sentence cannot be grouped on. + """ + + #: Intact, and not something VisionSet accepts. Operator noise, usually. + UNSUPPORTED = "unsupported" + #: A format we do accept, whose bytes will not decode. Data loss. + CORRUPT = "corrupt" + + +class IngestFailure(BaseModel): + """One item an ingest run could not turn into an asset. + + ``name`` is the run's own name for the item — a path for a file on disk, + ``clip.mp4#frame=42`` for a frame — and never the exception's, which + ``MediaError`` documents as reporting rather than identity. ``reason`` never + repeats the name, which is what lets a report be a table instead of a list + of sentences. + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + name: str + kind: IngestFailureKind + reason: str + + +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. + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + job_id: UUID + project_id: UUID + source_id: UUID + batch_id: UUID + #: Everything this run put in the batch, in ingest order. + assets: tuple[Asset, ...] = () + #: The subset of ``assets`` that was new to the project. + created_asset_ids: tuple[UUID, ...] = () + failures: tuple[IngestFailure, ...] = () + + @property + def asset_ids(self) -> tuple[UUID, ...]: + """Every asset in the batch after this run, in ingest order.""" + return tuple(asset.id for asset in self.assets) + + @property + def created(self) -> int: + """How many assets did not exist in this project before the run.""" + return len(self.created_asset_ids) + + @property + def deduplicated(self) -> int: + """How many items resolved to content the project already held.""" + return len(self.assets) - len(self.created_asset_ids) + + @property + def failed(self) -> int: + """How many items could not be read at all.""" + return len(self.failures) diff --git a/src/visionset/kernel/errors.py b/src/visionset/kernel/errors.py index d170255f..67849ef5 100644 --- a/src/visionset/kernel/errors.py +++ b/src/visionset/kernel/errors.py @@ -312,6 +312,15 @@ class SourceNotFound(VisionSetError): """ +class IngestJobNotFound(VisionSetError): + """No ingest job with that id lives in this workspace. + + Deliberately **not** ``JobNotFound``, which is an *annotation* job. The two + are different entities with different lifecycles, and a single ``except`` + catching both would be catching two things because they share a word. + """ + + class SchemaVersionConflict(VisionSetError): """Two writers raced for the same next version number, and this one lost. diff --git a/src/visionset/kernel/services/__init__.py b/src/visionset/kernel/services/__init__.py index 286950a1..7eaa6c20 100644 --- a/src/visionset/kernel/services/__init__.py +++ b/src/visionset/kernel/services/__init__.py @@ -10,6 +10,7 @@ from visionset.kernel.services.annotation_service import AnnotationService from visionset.kernel.services.batch_service import BatchService from visionset.kernel.services.dataset_service import DatasetService +from visionset.kernel.services.ingest_service import IngestService from visionset.kernel.services.job_service import JobService from visionset.kernel.services.project_service import ProjectService from visionset.kernel.services.release_service import ReleaseService @@ -27,6 +28,7 @@ "AnnotationService", "BatchService", "DatasetService", + "IngestService", "JobService", "ProjectService", "ReleaseService", diff --git a/src/visionset/kernel/services/batch_service.py b/src/visionset/kernel/services/batch_service.py index 39b1e0e7..0b49f219 100644 --- a/src/visionset/kernel/services/batch_service.py +++ b/src/visionset/kernel/services/batch_service.py @@ -144,7 +144,7 @@ def add_assets(self, batch_id: UUID, asset_ids: Sequence[UUID]) -> Batch: AssetNotFound: an asset id is not in this project. """ with self._workspace.unit_of_work() as uow: - batch = self._require_draft(uow, batch_id) + batch = self.require_draft(uow, batch_id) added = _require_assets(uow, batch.project_id, asset_ids) return uow.batches.update( batch.model_copy(update={"asset_ids": _deduplicated([*batch.asset_ids, *added])}) @@ -162,7 +162,7 @@ def remove_assets(self, batch_id: UUID, asset_ids: Sequence[UUID]) -> Batch: BatchNotEditable: the batch is past ``draft``. """ with self._workspace.unit_of_work() as uow: - batch = self._require_draft(uow, batch_id) + batch = self.require_draft(uow, batch_id) dropped = set(asset_ids) return uow.batches.update( batch.model_copy( @@ -347,7 +347,18 @@ def require_batch(self, uow: UnitOfWork, batch_id: UUID) -> Batch: self._require_project(uow, batch.project_id) return batch - def _require_draft(self, uow: UnitOfWork, batch_id: UUID) -> Batch: + def require_draft(self, uow: UnitOfWork, batch_id: UUID) -> Batch: + """The batch, refusing it if its membership is already frozen. + + Public, and taking a ``uow``, for the reason :meth:`require_batch` is: + ``IngestService`` has to know a target batch will accept members + *before* it decodes five thousand files, and discovering that inside a + later ``add_assets`` would mean finding out after the work. + + Raises: + BatchNotFound: no such batch in this workspace. + BatchNotEditable: the batch is past ``draft``. + """ batch = self.require_batch(uow, batch_id) if batch.state is not BatchState.DRAFT: raise BatchNotEditable( diff --git a/src/visionset/kernel/services/ingest_service.py b/src/visionset/kernel/services/ingest_service.py new file mode 100644 index 00000000..9553bbf4 --- /dev/null +++ b/src/visionset/kernel/services/ingest_service.py @@ -0,0 +1,411 @@ +# usage: from visionset.kernel.services import IngestService +"""Ingest: the one door that turns a registered source into assets. + +#18 recorded *where* data comes from. This is what reads it — hashing every +item, storing the bytes once, writing the row that names them, and putting the +result in a draft batch somebody can approve. It is the last write in the kernel +that had no service behind it: until now the only way to make an ``Asset`` was +``examples/sdk_end_to_end.py`` reaching below a service, which this task deletes. + +**One ``ingest``, where ``SourceService`` has two ``register_*``.** That split +was made because the arguments genuinely differed — a clip needs a rate and a +probe, a directory needs neither. Here they do not differ at all: the source +already carries its kind, its path and its decomposition rate, so the branch is +on ``SourceKind`` and the caller passes one id. A second entry point would ask +callers to re-state something the source already knows. + +**Identity is content; origin is provenance.** Two registered directories +holding the same photograph produce one blob and one asset, and that asset keeps +the origin of the first sighting — the rule ``Source.registered_at`` already +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 +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. + +**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 +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. +""" + +from __future__ import annotations + +from io import BytesIO +from pathlib import Path +from uuid import UUID + +from visionset.kernel.domain import ( + Asset, + Batch, + IngestCompleted, + IngestFailure, + IngestFailureKind, + IngestJob, + IngestResult, + IngestState, + Project, + Source, + SourceKind, + normalize_name, +) +from visionset.kernel.errors import ( + CorruptMedia, + IngestJobNotFound, + MediaError, + ProjectNotFound, +) +from visionset.kernel.ports import FRAME_FORMAT, UnitOfWork +from visionset.kernel.services.batch_service import BatchService +from visionset.kernel.services.source_service import SourceService +from visionset.kernel.services.workspace_service import WorkspaceService + + +class IngestService: + """Materialize the sources of one project into assets, and into batches.""" + + def __init__(self, workspace: WorkspaceService) -> None: + self._workspace = workspace + self._sources = SourceService(workspace) + self._batches = BatchService(workspace) + + # --- reading ----------------------------------------------------------- + + def get(self, job_id: UUID) -> IngestJob: + """The ingest job with that id. + + Raises: + IngestJobNotFound: no such ingest job in this workspace. + """ + with self._workspace.unit_of_work() as uow: + return self.require_job(uow, job_id) + + # --- ingesting --------------------------------------------------------- + + def ingest( + self, + source_id: UUID, + *, + batch_id: UUID | None = None, + batch_name: str | None = None, + ) -> IngestResult: + """Read the source, store what it holds, and put it all in one batch. + + A directory source is read at its top level, in filename order; anything + below a subdirectory is not looked at. A video source is decomposed at + the rate the source itself records, and each frame becomes an asset + carrying the position it came from. + + The batch is either an existing draft named by ``batch_id`` — checked to + be editable *before* anything is decoded, because finding out afterwards + would mean finding out after the work — or one this call creates, named + ``batch_name`` or, failing that, after the source's own file or folder. + Its membership is everything this run ingested, the assets that were + already in the project included: a duplicate is not new data, but it is + part of what this run was asked to gather. + + ``UnsupportedMedia`` and ``CorruptMedia`` are collected rather than + raised — see ``IngestResult.failures``, where each keeps the item's name + and the remedy apart. + + Raises: + SourceNotFound: no such source in this workspace. + BatchNotFound: ``batch_id`` names no batch in this workspace. + BatchNotEditable: the target batch is past ``draft``. + InvalidName: ``batch_name`` is blank once stripped. + FileNotFoundError: the source's path is no longer on disk. + NotADirectoryError: a directory source's path is now a file. + WorkspaceCorrupt: a video source carries no provenance. + MediaToolUnavailable: ffmpeg is not installed on this machine. The + job records it and is marked failed before it is re-raised — one + broken machine is not five thousand broken files. + """ + with self._workspace.unit_of_work() as uow: + 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)) + + try: + candidates, failures = self._read(source) + 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.model_copy( + update={"state": IngestState.COMPLETED, "batch_id": batch.id}, + ) + ) + except Exception as exc: + 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, + project_id=source.project_id, + source_id=source.id, + asset_count=len(assets), + ) + ) + return IngestResult( + job_id=job.id, + project_id=source.project_id, + source_id=source.id, + batch_id=batch.id, + assets=tuple(assets), + created_asset_ids=tuple(created), + failures=tuple(failures), + ) + + # --- the run, phase by phase ------------------------------------------- + + def _read(self, source: Source) -> 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) + + def _read_directory(self, source: Source) -> 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 + what *the source is* — "the same source yields the same assets" — so it + belongs to a future ``register_images(..., recursive=True)`` rather than + here, where it would silently change what an already-registered source + means. Subdirectories are stepped over and recorded nowhere. + + 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. + """ + candidates: list[Asset] = [] + failures: list[IngestFailure] = [] + directory = Path(source.path) + for path in sorted(item for item in directory.iterdir() if item.is_file()): + try: + with path.open("rb") as handle: + # Probe first: a file that is going to be refused should + # never leave a blob behind. The ``seek(0)`` between is not + # decoration — ``ImageProcessor`` promises to seek to 0 and + # not to close, and ``BlobStore.put`` promises neither, so + # rewinding for it is the caller's job. + metadata = self._workspace.image_processor.probe(handle, name=str(path)) + handle.seek(0) + 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, + ) + ) + return candidates, failures + + def _read_video(self, source: Source) -> 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 + is a complete image in ``FRAME_FORMAT`` at the dimensions ``probe`` + reported, and that guarantee is asserted where it belongs, in the port's + own tests. Decoding every frame a second time to re-confirm it would + also mean putting our own encoder's output into an operator's per-file + report — a failure nobody could act on. + + Damage arrives once and terminally: ffmpeg yields the frames it managed + and *then* says the bytes ran out, so the refusal is caught around the + 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. + """ + provenance = source.require_video() + candidates: list[Asset] = [] + failures: list[IngestFailure] = [] + clip = Path(source.path) + frames = self._workspace.video_processor.frames( + clip, fps=provenance.extraction_fps, name=clip.name + ) + try: + for frame in frames: + content_hash = self._workspace.blob_store.put(BytesIO(frame.content)) + candidates.append( + Asset( + project_id=source.project_id, + content_hash=content_hash, + uri=f"{source.path}#frame={frame.index}", + width=provenance.metadata.width, + height=provenance.metadata.height, + format=FRAME_FORMAT, + source_id=source.id, + frame_index=frame.index, + frame_timestamp=frame.timestamp, + ) + ) + except MediaError as exc: + failures.append(_failure(source.path, exc)) + return candidates, failures + + def _store(self, project_id: UUID, candidates: list[Asset]) -> tuple[list[Asset], list[UUID]]: + """Write the rows, reusing whatever content the project already holds. + + The project's assets are read once, into a map keyed by content hash, + rather than queried per item — ``Repository`` has one query shape, and a + service never builds SQL. The whole-project read is affordable at this + scale and the fix when it stops being is a port method, not an import. + + The map is updated as the run proceeds, so two identical files inside one + directory become one asset rather than a pair the new unique index would + refuse at commit. + """ + assets: list[Asset] = [] + created: list[UUID] = [] + seen: set[UUID] = set() + with self._workspace.unit_of_work() as uow: + known = {asset.content_hash: asset for asset in uow.assets.list(project_id)} + for candidate in candidates: + stored = known.get(candidate.content_hash) + if stored is None: + stored = uow.assets.add(candidate) + known[stored.content_hash] = stored + created.append(stored.id) + if stored.id not in seen: + seen.add(stored.id) + assets.append(stored) + return assets, created + + def _materialize( + self, project_id: UUID, name: str, batch_id: UUID | None, assets: list[Asset] + ) -> Batch: + """Put the run's assets in their batch, through the service that owns it. + + After the rows and not before, so a run that dies during the decode + leaves no empty draft batch behind and its job's ``batch_id`` stays NULL + — which is what that column being nullable actually means. + """ + asset_ids = [asset.id for asset in assets] + if batch_id is None: + 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.""" + 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}) + ) + + # --- lookups shared by the operations above ---------------------------- + + def require_job(self, uow: UnitOfWork, job_id: UUID) -> IngestJob: + """The job, checked through its source so workspaces stay separate. + + Public, and taking a ``uow``, for the reason + ``SourceService.require_source`` is: #19 has to resolve a job inside its + own transaction before it writes progress against it, and a second + spelling of this ladder is a second place for it to be got wrong. + + Raises: + IngestJobNotFound: no such ingest job in this workspace. + """ + job = uow.ingest_jobs.get(job_id) + if job is not None: + source = uow.sources.get(job.source_id) + if source is not None: + project = uow.projects.get(source.project_id) + if project is not None and project.workspace_id == self._workspace.workspace_id: + return job + raise IngestJobNotFound( + f"no ingest job {job_id} in workspace {self._workspace.workspace.name!r}" + ) + + def _target_name( + self, uow: UnitOfWork, source: Source, batch_id: UUID | None, batch_name: str | None + ) -> str: + """The name a created batch would take, and the gate on an existing one. + + Both refusals belong here, before the decode: a blank name and a frozen + batch are things a caller can fix, and finding either out after five + thousand files have been hashed helps nobody. + """ + if batch_id is not None: + batch = self._batches.require_draft(uow, batch_id) + self._require_project(uow, batch.project_id) + return batch.name + return normalize_name(batch_name or Path(source.path).name, what="batch") + + def _require_project(self, uow: UnitOfWork, project_id: UUID) -> Project: + """The project, or refuse because this workspace does not have it.""" + project = uow.projects.get(project_id) + if project is None or project.workspace_id != self._workspace.workspace_id: + raise ProjectNotFound( + f"no project {project_id} in workspace {self._workspace.workspace.name!r}" + ) + return project + + # ``list`` shadows the builtin for every annotation after it in a class + # body, so it is declared last. See ``BatchService`` for the precedent. + + def list(self, source_id: UUID) -> list[IngestJob]: + """Every run of that source, in the order they started. + + Parented on the source rather than on the project, because that is the + one query shape ``Repository`` has and ``ingest_job.source_id`` is what + the row hangs from. A project's runs are reached through its sources. + + Raises: + SourceNotFound: no such source in this workspace. + """ + with self._workspace.unit_of_work() as uow: + source = self._sources.require_source(uow, source_id) + return uow.ingest_jobs.list(source.id) + + +def _failure(name: str, exc: MediaError) -> IngestFailure: + """One report line, with the item's name kept apart from the remedy. + + The name comes from the loop's own item and never from ``exc.name``: + ``MediaError`` documents its own as reporting rather than identity, and the + caller here already knows exactly what it was reading. + + The two branches are the whole family — ``UnsupportedMedia`` is "intact and + not for us", ``CorruptMedia`` is "for us and broken" — and a third member + would have to say which of those a report should file it under. + """ + kind = ( + IngestFailureKind.CORRUPT + if isinstance(exc, CorruptMedia) + else IngestFailureKind.UNSUPPORTED + ) + return IngestFailure(name=name, kind=kind, reason=exc.reason) diff --git a/src/visionset/kernel/services/source_service.py b/src/visionset/kernel/services/source_service.py index df95a801..1032e14e 100644 --- a/src/visionset/kernel/services/source_service.py +++ b/src/visionset/kernel/services/source_service.py @@ -24,16 +24,16 @@ place** rather than left describing a file that is gone. ``registered_at`` is never rewritten; it is the first registration. -**The idempotency has no constraint underneath it, and that is a deliberate, -named gap.** ``docs/persistence.md`` says a rule with no backstop is a wish, and -every other uniqueness rule in this store has a unique index behind it. This one -does not, because the natural key includes a float parameter and a nullable one -at that, and because two concurrent registrations of one folder are not yet able -to hurt anything: no row references a source, so a duplicate is inert. That -stops being true at #20, when ``asset.source_id`` gets a target and the winner of -a race starts deciding an asset's recorded origin. **#20 is where this needs an -index under it**, alongside the store's other known concurrency gap (#80, the -untranslated ``OperationalError``). +**That idempotency now has a constraint underneath it.** It shipped without one, +as a named gap: no row referenced a source, so a duplicate born of two concurrent +registrations was inert. Ingest ended that — ``asset.source_id`` has a target, so +the winner of such a race would decide an asset's recorded origin — and +``uq_source_project_kind_path_fps`` went in with it. The two layers do what they +do everywhere else in this store: the pre-check below is what produces a friendly +answer, and the index is the guarantee. A caller that loses the race sees a raw +``ConstraintViolated``, and the remedy is to call the same method again, which +finds the winner's row and returns it. The store's other known concurrency gap +(#80, the untranslated ``OperationalError``) is still open. Composition follows the rule in ``docs/workspaces.md``: this service takes an open ``WorkspaceService`` and nothing else, and reaches ``video_processor`` diff --git a/tests/examples/test_sdk_end_to_end.py b/tests/examples/test_sdk_end_to_end.py index 668ab9d7..3c9b2bbb 100644 --- a/tests/examples/test_sdk_end_to_end.py +++ b/tests/examples/test_sdk_end_to_end.py @@ -124,7 +124,19 @@ def test_every_service_that_should_announce_itself_did(summary: Any) -> None: assert events.count("release_published") == 2 # the publish and the reissue # One per call rather than one per box: five annotated assets, five calls. assert events.count("annotations_written") == 5 - # Declared but emitted by nobody until M2 wires ingest (#20). - assert "ingest_completed" not in events + assert events.count("ingest_completed") == 1 + assert events.index("ingest_completed") < events.index("batch_approved") assert events.index("batch_approved") < events.index("batch_completed") assert events.index("batch_completed") < events.index("release_published") + + +def test_the_example_ingests_rather_than_writing_assets_by_hand(summary: Any) -> None: + """#20 closed the one place this example reached below a service. + + The assets exist because a directory was registered and read, not because + the example wrote the rows itself — so there is a source and an ingest job + behind them, and every asset carries the origin it came from. + """ + assert summary.source_id is not None + assert summary.ingest_job_id is not None + assert len(summary.asset_ids) == 6 diff --git a/tests/kernel/test_events.py b/tests/kernel/test_events.py index 419b6e01..419585d7 100644 --- a/tests/kernel/test_events.py +++ b/tests/kernel/test_events.py @@ -478,8 +478,15 @@ def test_publishing_a_release_announces_what_it_froze(tmp_path: Path) -> None: fixture.close() -def test_nothing_in_m1_emits_ingest_completed(tmp_path: Path) -> None: - """Declared in #13, wired in M2 — and not before, by anything, quietly.""" +def test_the_annotation_cycle_announces_no_ingest(tmp_path: Path) -> None: + """#20 wired the emitter, so this no longer says "nobody". It says what is + still true and still worth a guard: nothing on the batch → job → annotation + → release path is an ingest, and none of it may quietly claim to be one. + + The positive coverage lives in ``tests/kernel/test_ingest_service.py``, + where a real run can be made without giving this file's fixture — which + builds its assets by hand, deliberately — a media dependency. + """ fixture = Fixture(tmp_path) fixture.to_release() diff --git a/tests/kernel/test_ingest_service.py b/tests/kernel/test_ingest_service.py new file mode 100644 index 00000000..2f00c1fd --- /dev/null +++ b/tests/kernel/test_ingest_service.py @@ -0,0 +1,832 @@ +"""`IngestService`: content identity, recorded origin, and the per-file report. + +Two things shape this file, both inherited from `test_source_service.py`. + +The ffmpeg requirement arrives through `write_video`, which calls `require_ffmpeg` +itself — there is deliberately no module-level skip. Most of what is asserted here +(dedup, the report, batch targeting, the not-found ladders) is about stills and has +to run on a machine with no ffmpeg at all. + +Assertions are about the contract rather than the implementation: "one blob" is +counted on disk because that is the acceptance criterion in the issue, and "the +same asset" compares ids rather than row counts. +""" + +from pathlib import Path +from uuid import UUID, uuid4 + +import pytest +from pydantic import ValidationError +from tests.fixtures.media import ( + GeneratedVideo, + write_corrupt_image, + write_corrupt_video, + write_image, + write_image_in_unsupported_format, + write_images, + write_rotated_video, + write_unsupported_file, + write_video, +) + +from visionset.kernel import ( + BatchNotEditable, + BatchNotFound, + IngestJobNotFound, + InvalidName, + MediaToolUnavailable, + SourceNotFound, +) +from visionset.kernel.domain import ( + Asset, + BatchState, + ImageFormat, + IngestCompleted, + IngestFailureKind, + IngestState, + VideoFrame, + VideoMetadata, +) +from visionset.kernel.ports import FRAME_FORMAT +from visionset.kernel.services import ( + BatchService, + IngestService, + ProjectService, + SourceService, + WorkspaceService, +) + + +class _NoFfmpeg: + """A `VideoProcessor` on a machine with no decoder installed. + + Injected through the composition point rather than monkeypatched, because + that seam is exactly what `WorkspaceService` documents it is for. + """ + + def probe(self, source: Path, *, name: str | None = None) -> VideoMetadata: + raise MediaToolUnavailable("ffmpeg is not installed; install it and try again") + + def frames( + self, source: Path, *, fps: float = 1.0, name: str | None = None + ) -> "list[VideoFrame]": + raise MediaToolUnavailable("ffmpeg is not installed; install it and try again") + + +class Fixture: + """A workspace with one project, a directory to fill, and every service.""" + + def __init__(self, tmp_path: Path, name: str = "ws") -> None: + self.tmp_path = tmp_path + self.root = tmp_path / name + self.workspace = WorkspaceService.init(self.root) + self.projects = ProjectService(self.workspace) + self.sources = SourceService(self.workspace) + self.batches = BatchService(self.workspace) + self.ingest = IngestService(self.workspace) + self.project = self.projects.create(f"{name}-project") + self.stills = tmp_path / f"{name}-stills" + self.stills.mkdir() + + def clip(self, name: str = "clip.mp4", **kwargs: object) -> GeneratedVideo: + return write_video(self.tmp_path / name, **kwargs) # type: ignore[arg-type] + + def blob_count(self) -> int: + """Blobs on disk. `put` is content-addressed, so this counts distinct bytes.""" + return len([path for path in (self.root / "blobs").rglob("*") if path.is_file()]) + + def assets(self) -> list[Asset]: + with self.workspace.unit_of_work() as uow: + return uow.assets.list(self.project.id) + + def close(self) -> None: + self.workspace.close() + + +# --- a directory of stills ------------------------------------------------ + + +def test_a_directory_of_stills_becomes_assets_in_a_draft_batch(tmp_path: Path) -> None: + 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) + + assert result.created == 3 + assert result.deduplicated == 0 + assert result.failed == 0 + batch = fixture.batches.get(result.batch_id) + assert batch.state is BatchState.DRAFT + assert batch.asset_ids == list(result.asset_ids) + fixture.close() + + +def test_membership_order_is_the_order_the_directory_was_read_in(tmp_path: Path) -> None: + """Filename order, so a re-run of an unchanged directory lines up with the first.""" + fixture = Fixture(tmp_path) + paths = write_images(fixture.stills, count=4) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id) + + assert [asset.uri for asset in result.assets] == [str(path) for path in sorted(paths)] + fixture.close() + + +def test_every_asset_records_the_dimensions_and_format_the_decoder_reported( + tmp_path: Path, +) -> None: + fixture = Fixture(tmp_path) + write_image(fixture.stills / "a.png", size=(40, 30), seed=1) + write_image(fixture.stills / "b.jpg", size=(40, 30), seed=2) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id) + + by_name = {Path(asset.uri).name: asset for asset in result.assets} + assert by_name["a.png"].format is ImageFormat.PNG + assert by_name["b.jpg"].format is ImageFormat.JPEG + assert (by_name["a.png"].width, by_name["a.png"].height) == (40, 30) + fixture.close() + + +def test_the_bytes_decide_the_format_and_not_the_name(tmp_path: Path) -> None: + """A JPEG somebody renamed is still a JPEG, and the row says so.""" + fixture = Fixture(tmp_path) + written = write_image(fixture.stills / "truth.jpg") + written.rename(fixture.stills / "liar.png") + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id) + + assert result.assets[0].format is ImageFormat.JPEG + fixture.close() + + +def test_an_exif_rotated_still_is_stored_at_its_displayed_size(tmp_path: Path) -> None: + """#16 applies orientation rather than reporting it; this is where it lands.""" + fixture = Fixture(tmp_path) + write_image(fixture.stills / "upright.jpg", size=(32, 24), orientation=6) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id) + + assert (result.assets[0].width, result.assets[0].height) == (24, 32) + fixture.close() + + +def test_a_still_records_its_source_and_no_frame_position(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=1) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + asset = fixture.ingest.ingest(source.id).assets[0] + + assert asset.source_id == source.id + assert asset.frame_index is None + assert asset.frame_timestamp is None + fixture.close() + + +def test_a_subdirectory_is_stepped_over_and_reported_nowhere(tmp_path: Path) -> None: + """Top level only — recursion is a question about what the source *is*.""" + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=2) + write_images(fixture.stills / "nested", count=3) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id) + + assert result.created == 2 + assert result.failed == 0 + fixture.close() + + +def test_an_empty_directory_produces_an_empty_draft_batch(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id) + + assert result.assets == () + assert fixture.batches.get(result.batch_id).asset_ids == [] + assert fixture.ingest.get(result.job_id).state is IngestState.COMPLETED + fixture.close() + + +# --- a clip --------------------------------------------------------------- + + +@pytest.mark.parametrize(("rate", "expected"), [(1.0, 2), (5.0, 10), (10.0, 20)]) +def test_a_clip_becomes_one_asset_per_extracted_frame( + tmp_path: Path, rate: float, expected: int +) -> None: + """The fixture is 10 fps for 2 s, so these three land exactly rather than round.""" + fixture = Fixture(tmp_path) + clip = fixture.clip() + source = fixture.sources.register_video(fixture.project.id, clip.path, extraction_fps=rate) + + result = fixture.ingest.ingest(source.id) + + assert result.created == expected + fixture.close() + + +def test_every_frame_records_where_in_the_clip_it_came_from(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + clip = fixture.clip() + source = fixture.sources.register_video(fixture.project.id, clip.path, extraction_fps=5.0) + + result = fixture.ingest.ingest(source.id) + + assert [asset.frame_index for asset in result.assets] == list(range(10)) + assert [asset.frame_timestamp for asset in result.assets] == [ + pytest.approx(index / 5.0) for index in range(10) + ] + fixture.close() + + +def test_a_frame_uri_names_the_clip_and_the_frame(tmp_path: Path) -> None: + """The spelling `MediaError`'s docstring already fixed for a decoded frame.""" + 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) + + assert [asset.uri for asset in result.assets] == [ + f"{source.path}#frame=0", + f"{source.path}#frame=1", + ] + fixture.close() + + +def test_a_frame_takes_its_size_from_the_probe_and_its_format_from_the_port( + tmp_path: Path, +) -> None: + """Frames are not re-decoded to confirm what the port already guarantees. + + `VideoProcessor` owes every frame in `FRAME_FORMAT` at the dimensions `probe` + reported, and that promise is asserted in the port's own tests. Paying a + Pillow decode per frame to re-check it would also route our own encoder's + output into an operator's per-file report. + """ + fixture = Fixture(tmp_path) + clip = fixture.clip() + source = fixture.sources.register_video(fixture.project.id, clip.path, extraction_fps=1.0) + + asset = fixture.ingest.ingest(source.id).assets[0] + + assert (asset.width, asset.height) == (clip.width, clip.height) + assert asset.format is FRAME_FORMAT + fixture.close() + + +def test_a_rotated_clip_yields_frames_at_their_displayed_size(tmp_path: Path) -> None: + """#17 applies the display matrix; a 64x48 file held upright ingests as 48x64.""" + fixture = Fixture(tmp_path) + clip = write_rotated_video(tmp_path / "upright.mp4") + source = fixture.sources.register_video(fixture.project.id, clip.path, extraction_fps=1.0) + + asset = fixture.ingest.ingest(source.id).assets[0] + + assert (asset.width, asset.height) == (clip.height, clip.width) + fixture.close() + + +def test_a_truncated_clip_keeps_what_decoded_and_reports_the_break(tmp_path: Path) -> None: + """Partial success is the contract: ffmpeg yields, then says the bytes ran out.""" + fixture = Fixture(tmp_path) + clip = write_corrupt_video(tmp_path / "broken.mp4") + source = fixture.sources.register_video(fixture.project.id, clip.path, extraction_fps=10.0) + + result = fixture.ingest.ingest(source.id) + + assert 0 < result.created < clip.frame_count + assert [failure.kind for failure in result.failures] == [IngestFailureKind.CORRUPT] + assert result.failures[0].name == source.path + assert fixture.ingest.get(result.job_id).state is IngestState.COMPLETED + fixture.close() + + +# --- identity is the content hash ----------------------------------------- + + +def test_the_same_image_in_two_sources_is_one_blob_and_one_asset(tmp_path: Path) -> None: + """The issue's first acceptance criterion, asserted on the disk and on the rows.""" + fixture = Fixture(tmp_path) + other = tmp_path / "second" + write_image(fixture.stills / "shared.png", seed=7) + write_image(other / "shared.png", seed=7) + first = fixture.sources.register_images(fixture.project.id, fixture.stills) + second = fixture.sources.register_images(fixture.project.id, other) + + original = fixture.ingest.ingest(first.id) + again = fixture.ingest.ingest(second.id) + + assert fixture.blob_count() == 1 + assert len(fixture.assets()) == 1 + assert again.created == 0 + assert again.deduplicated == 1 + assert again.asset_ids == original.asset_ids + fixture.close() + + +def test_the_first_origin_wins_when_the_same_bytes_arrive_again(tmp_path: Path) -> None: + """Origin is provenance, not identity, so a second sighting does not rewrite it.""" + fixture = Fixture(tmp_path) + other = tmp_path / "second" + write_image(fixture.stills / "shared.png", seed=7) + write_image(other / "shared.png", seed=7) + first = fixture.sources.register_images(fixture.project.id, fixture.stills) + second = fixture.sources.register_images(fixture.project.id, other) + fixture.ingest.ingest(first.id) + + asset = fixture.ingest.ingest(second.id).assets[0] + + assert asset.source_id == first.id + assert asset.uri == str(fixture.stills / "shared.png") + fixture.close() + + +def test_a_duplicate_still_joins_the_batch_of_the_run_that_found_it(tmp_path: Path) -> None: + """A duplicate is not new data, but it is part of what this run was asked to gather.""" + fixture = Fixture(tmp_path) + other = tmp_path / "second" + write_image(fixture.stills / "shared.png", seed=7) + write_image(other / "shared.png", seed=7) + first = fixture.sources.register_images(fixture.project.id, fixture.stills) + second = fixture.sources.register_images(fixture.project.id, other) + original = fixture.ingest.ingest(first.id) + + again = fixture.ingest.ingest(second.id) + + assert fixture.batches.get(again.batch_id).asset_ids == list(original.asset_ids) + fixture.close() + + +def test_two_identical_files_in_one_directory_become_one_asset(tmp_path: Path) -> None: + """Within-run dedup. Without it the pair reaches the unique index and the run dies.""" + fixture = Fixture(tmp_path) + write_image(fixture.stills / "a.png", seed=3) + write_image(fixture.stills / "b.png", seed=3) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id) + + assert result.created == 1 + assert len(result.assets) == 1 + assert fixture.blob_count() == 1 + fixture.close() + + +def test_re_ingesting_one_source_creates_nothing(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=3) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + first = fixture.ingest.ingest(source.id) + + again = fixture.ingest.ingest(source.id) + + assert again.created == 0 + assert again.deduplicated == 3 + assert again.asset_ids == first.asset_ids + assert fixture.blob_count() == 3 + fixture.close() + + +def test_two_projects_ingesting_one_image_are_two_assets_over_one_blob(tmp_path: Path) -> None: + """The unique index is per project — that is what makes `project_id` the parent.""" + fixture = Fixture(tmp_path) + write_image(fixture.stills / "shared.png", seed=7) + other = fixture.projects.create("second-project") + here = fixture.sources.register_images(fixture.project.id, fixture.stills) + there = fixture.sources.register_images(other.id, fixture.stills) + + mine = fixture.ingest.ingest(here.id) + theirs = fixture.ingest.ingest(there.id) + + assert mine.asset_ids != theirs.asset_ids + assert mine.assets[0].content_hash == theirs.assets[0].content_hash + assert fixture.blob_count() == 1 + fixture.close() + + +# --- the backstops under it ----------------------------------------------- + + +def test_the_store_refuses_a_second_asset_with_the_same_content(tmp_path: Path) -> None: + """The rule is the service's; the index is what makes it more than a wish.""" + from visionset.kernel import ConstraintViolated + + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=1) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + ingested = fixture.ingest.ingest(source.id).assets[0] + + with pytest.raises(ConstraintViolated), fixture.workspace.unit_of_work() as uow: + uow.assets.add(ingested.model_copy(update={"id": uuid4()})) + fixture.close() + + +def test_the_store_refuses_a_second_source_for_one_origin(tmp_path: Path) -> None: + """The index #18 named as owed, now that `asset.source_id` has a target.""" + from visionset.kernel import ConstraintViolated + + fixture = Fixture(tmp_path) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + with pytest.raises(ConstraintViolated), fixture.workspace.unit_of_work() as uow: + uow.sources.add(source.model_copy(update={"id": uuid4()})) + fixture.close() + + +def test_one_clip_at_two_rates_is_still_two_sources(tmp_path: Path) -> None: + """The fourth index term is doing work rather than decorating the other three.""" + fixture = Fixture(tmp_path) + clip = fixture.clip() + + slow = fixture.sources.register_video(fixture.project.id, clip.path, extraction_fps=1.0) + fast = fixture.sources.register_video(fixture.project.id, clip.path, extraction_fps=5.0) + + assert slow.id != fast.id + fixture.close() + + +# --- the target batch ----------------------------------------------------- + + +def test_a_created_batch_is_named_after_its_source_by_default(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id) + + assert fixture.batches.get(result.batch_id).name == fixture.stills.name + fixture.close() + + +def test_a_caller_may_name_the_batch_the_run_creates(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id, batch_name="monday-dashcam") + + assert fixture.batches.get(result.batch_id).name == "monday-dashcam" + fixture.close() + + +def test_an_existing_draft_batch_is_added_to_rather_than_replaced(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=2) + first = fixture.sources.register_images(fixture.project.id, fixture.stills) + original = fixture.ingest.ingest(first.id) + more = tmp_path / "more" + write_images(more, count=2, first_seed=50) + second = fixture.sources.register_images(fixture.project.id, more) + + again = fixture.ingest.ingest(second.id, batch_id=original.batch_id) + + assert again.batch_id == original.batch_id + batch = fixture.batches.get(original.batch_id) + assert batch.asset_ids == [*original.asset_ids, *again.asset_ids] + fixture.close() + + +def test_ingesting_into_a_frozen_batch_is_refused_before_anything_is_decoded( + tmp_path: Path, +) -> None: + """Fail-fast asserted rather than assumed: no blobs, no assets, no job left running.""" + fixture = Fixture(tmp_path) + 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) + more = tmp_path / "more" + write_images(more, count=2, first_seed=90) + second = fixture.sources.register_images(fixture.project.id, more) + blobs_before = fixture.blob_count() + + with pytest.raises(BatchNotEditable): + fixture.ingest.ingest(second.id, batch_id=opened.batch_id) + + assert fixture.blob_count() == blobs_before + assert len(fixture.assets()) == 2 + assert fixture.ingest.list(second.id) == [] + fixture.close() + + +def test_ingesting_into_a_batch_of_another_workspace_reads_as_missing(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + elsewhere = Fixture(tmp_path, name="other") + stranger = elsewhere.batches.create(elsewhere.project.id, "theirs") + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + with pytest.raises(BatchNotFound): + fixture.ingest.ingest(source.id, batch_id=stranger.id) + + elsewhere.close() + fixture.close() + + +def test_a_blank_batch_name_is_refused_before_anything_is_decoded(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=2) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + with pytest.raises(InvalidName): + fixture.ingest.ingest(source.id, batch_name=" ") + + assert fixture.blob_count() == 0 + fixture.close() + + +# --- the per-file report -------------------------------------------------- + + +def test_a_file_that_is_not_an_image_is_reported_and_the_run_carries_on(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=2) + notes = write_unsupported_file(fixture.stills / "notes.txt") + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id) + + assert result.created == 2 + assert [failure.name for failure in result.failures] == [str(notes)] + fixture.close() + + +def test_a_refused_file_leaves_no_blob_behind(tmp_path: Path) -> None: + """Probe before put, which is the whole reason for that ordering.""" + fixture = Fixture(tmp_path) + write_unsupported_file(fixture.stills / "notes.txt") + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + fixture.ingest.ingest(source.id) + + assert fixture.blob_count() == 0 + fixture.close() + + +def test_a_report_line_keeps_the_name_and_the_reason_apart(tmp_path: Path) -> None: + """So a report renders as a table rather than as a list of sentences.""" + fixture = Fixture(tmp_path) + notes = write_unsupported_file(fixture.stills / "notes.txt") + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + failure = fixture.ingest.ingest(source.id).failures[0] + + assert failure.name == str(notes) + assert str(notes) not in failure.reason + assert notes.name not in failure.reason + fixture.close() + + +def test_the_report_separates_data_loss_from_operator_noise(tmp_path: Path) -> None: + """Why `IngestFailureKind` exists: a reason sentence cannot be grouped on.""" + fixture = Fixture(tmp_path) + write_unsupported_file(fixture.stills / "notes.txt") + write_image_in_unsupported_format(fixture.stills / "old.bmp") + write_corrupt_image(fixture.stills / "half.png") + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id) + + by_name = {Path(failure.name).name: failure.kind for failure in result.failures} + assert by_name == { + "notes.txt": IngestFailureKind.UNSUPPORTED, + "old.bmp": IngestFailureKind.UNSUPPORTED, + "half.png": IngestFailureKind.CORRUPT, + } + fixture.close() + + +def test_per_file_failures_do_not_fail_the_job(tmp_path: Path) -> None: + """The remedy split, asserted at the level where it decides something.""" + fixture = Fixture(tmp_path) + write_unsupported_file(fixture.stills / "notes.txt") + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id) + + assert fixture.ingest.get(result.job_id).state is IngestState.COMPLETED + fixture.close() + + +# --- the job record ------------------------------------------------------- + + +def test_a_completed_run_leaves_a_job_pointing_at_its_source_and_its_batch( + tmp_path: Path, +) -> None: + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=1) + 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 + assert job.source_id == source.id + assert job.batch_id == result.batch_id + assert job.error is None + fixture.close() + + +def test_a_missing_decoder_fails_the_job_and_is_re_raised(tmp_path: Path) -> None: + """One broken machine is not five thousand broken files — hence no report line.""" + 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, + ), + ) + ) + + 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.batch_id is None + workspace.close() + + +def test_a_source_that_has_been_deleted_fails_the_job_and_is_re_raised(tmp_path: Path) -> None: + 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) + + assert fixture.ingest.list(source.id)[0].state is IngestState.FAILED + fixture.close() + + +def test_every_run_of_one_source_is_listed_in_order(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=1) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + first = fixture.ingest.ingest(source.id) + second = fixture.ingest.ingest(source.id) + + assert [job.id for job in fixture.ingest.list(source.id)] == [first.job_id, second.job_id] + fixture.close() + + +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.""" + fixture = Fixture(tmp_path) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + result = fixture.ingest.ingest(source.id) + + with fixture.workspace.unit_of_work() as uow: + assert fixture.ingest.require_job(uow, result.job_id).id == result.job_id + fixture.close() + + +# --- scope ---------------------------------------------------------------- + + +def test_ingesting_an_unknown_source_is_refused(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + + with pytest.raises(SourceNotFound): + fixture.ingest.ingest(uuid4()) + fixture.close() + + +def test_getting_an_unknown_ingest_job_is_refused(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + + with pytest.raises(IngestJobNotFound): + fixture.ingest.get(uuid4()) + fixture.close() + + +def test_an_ingest_job_in_another_workspace_reads_as_missing(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + elsewhere = Fixture(tmp_path, name="other") + source = elsewhere.sources.register_images(elsewhere.project.id, elsewhere.stills) + theirs = elsewhere.ingest.ingest(source.id) + + with pytest.raises(IngestJobNotFound): + fixture.ingest.get(theirs.job_id) + + elsewhere.close() + fixture.close() + + +# --- announcements -------------------------------------------------------- + + +def test_a_completed_ingest_announces_itself(tmp_path: Path) -> None: + """The tripwire M1 left on purpose, flipped: something emits this now.""" + fixture = Fixture(tmp_path) + seen: list[IngestCompleted] = [] + fixture.workspace.event_bus.subscribe(IngestCompleted, seen.append) + write_images(fixture.stills, count=3) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + + result = fixture.ingest.ingest(source.id) + + assert len(seen) == 1 + assert seen[0].ingest_job_id == result.job_id + assert seen[0].project_id == fixture.project.id + assert seen[0].source_id == source.id + assert seen[0].asset_count == 3 + fixture.close() + + +def test_a_run_that_failed_announces_nothing(tmp_path: Path) -> None: + fixture = Fixture(tmp_path) + seen: list[IngestCompleted] = [] + fixture.workspace.event_bus.subscribe(IngestCompleted, seen.append) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + fixture.stills.rmdir() + + with pytest.raises(FileNotFoundError): + fixture.ingest.ingest(source.id) + + assert seen == [] + fixture.close() + + +def test_the_announcement_follows_the_commit(tmp_path: Path) -> None: + """A subscriber reading the workspace back must find the work already there.""" + fixture = Fixture(tmp_path) + write_images(fixture.stills, count=2) + source = fixture.sources.register_images(fixture.project.id, fixture.stills) + found: list[int] = [] + + def count_members(event: IngestCompleted) -> None: + with fixture.workspace.unit_of_work() as uow: + found.append(len(uow.assets.list(event.project_id))) + + fixture.workspace.event_bus.subscribe(IngestCompleted, count_members) + fixture.ingest.ingest(source.id) + + assert found == [2] + fixture.close() + + +# --- the domain invariant ------------------------------------------------- + + +def _asset(**overrides: object) -> Asset: + fields: dict[str, object] = { + "project_id": uuid4(), + "content_hash": "a" * 64, + "uri": "file:///x.png", + } + fields.update(overrides) + return Asset(**fields) # type: ignore[arg-type] + + +def test_a_frame_index_without_a_timestamp_is_refused() -> None: + with pytest.raises(ValidationError, match="together or not at all"): + _asset(source_id=uuid4(), frame_index=0) + + +def test_a_frame_timestamp_without_an_index_is_refused() -> None: + with pytest.raises(ValidationError, match="together or not at all"): + _asset(source_id=uuid4(), frame_timestamp=0.0) + + +def test_a_frame_position_without_a_source_is_refused() -> None: + with pytest.raises(ValidationError, match="needs the source"): + _asset(frame_index=0, frame_timestamp=0.0) + + +def test_an_asset_may_carry_a_source_and_no_frame_position() -> None: + source_id: UUID = uuid4() + + assert _asset(source_id=source_id).source_id == source_id diff --git a/tests/kernel/test_migrations.py b/tests/kernel/test_migrations.py index cd3533b6..a4ced587 100644 --- a/tests/kernel/test_migrations.py +++ b/tests/kernel/test_migrations.py @@ -76,6 +76,30 @@ def _downgrade_to_version_one(store: SqliteMetadataStore) -> None: ) ) connection.execute(text("CREATE INDEX ix_source_project_id ON source (project_id)")) + # Migration 8's own undo. The index goes before the columns it names, + # because SQLite refuses to drop a column an index still references. + connection.execute(text("drop index if exists uq_asset_project_content_hash")) + connection.execute(text("alter table asset drop column frame_timestamp")) + connection.execute(text("alter table asset drop column frame_index")) + connection.execute(text("alter table asset drop column source_id")) + connection.execute(text("alter table asset drop column format")) + # ``ingest_job.batch_id`` sits in a foreign-key clause, and SQLite + # 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. + connection.execute(text("drop table ingest_job")) + connection.execute( + text( + "CREATE TABLE ingest_job (" + " id CHAR(32) NOT NULL," + " source_id CHAR(32) NOT NULL," + " state VARCHAR NOT NULL," + " error VARCHAR," + " PRIMARY KEY (id)," + " FOREIGN KEY(source_id) REFERENCES source (id) ON DELETE CASCADE)" + ) + ) + connection.execute(text("CREATE INDEX ix_ingest_job_source_id ON ingest_job (source_id)")) connection.execute(text("update _visionset_meta set format_version = 1")) @@ -233,6 +257,155 @@ def test_migration_seven_refuses_a_workspace_that_still_holds_pre_provenance_row store.close() +def test_migration_eight_gives_an_asset_its_origin(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("asset")} + assert {"format", "source_id", "frame_index", "frame_timestamp"} <= columns + store.close() + + +def test_migration_eight_links_an_ingest_job_to_its_batch(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_id" in columns + store.close() + + +def test_migration_eight_makes_content_unique_within_a_project(tmp_path: Path) -> None: + """Per project, not globally: two projects may hold one photograph.""" + store = SqliteMetadataStore(tmp_path / "visionset.db") + store.initialize() + index = next(sql for sql in _schema(store) if "uq_asset_project_content_hash" in sql) + assert "project_id" in index + store.close() + + +def test_migration_eight_puts_an_index_under_the_source_idempotency_rule(tmp_path: Path) -> None: + """The fourth term is the expression, and it is the half that can rot silently. + + A three-column index over ``(project_id, kind, path)`` would look right and + would refuse a clip's second extraction rate, which is a legitimate second + source; a nullable fourth column would collide with nothing at all, because + SQLite treats NULLs in a unique index as distinct. + """ + store = SqliteMetadataStore(tmp_path / "visionset.db") + store.initialize() + index = next(sql for sql in _schema(store) if "uq_source_project_kind_path_fps" in sql) + assert "json_extract" in index + assert "coalesce" in index + store.close() + + +def test_migration_eight_refuses_a_workspace_holding_duplicate_assets(tmp_path: Path) -> None: + """Refused, not merged: each duplicate may carry its own annotations.""" + store = SqliteMetadataStore(tmp_path / "visionset.db") + store.initialize() + _downgrade_to_version_one(store) + 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')") + ) + for asset_id in ("a1", "a2"): + connection.execute( + text( + "insert into asset (id, project_id, modality, content_hash, uri) " + f"values ('{asset_id}', 'p', 'image', 'deadbeef', '/x.png')" + ) + ) + + with pytest.raises(WorkspaceCorrupt, match="duplicate assets"): + store.initialize() + + # And the rows are still there — refused, not emptied. + with store.engine.connect() as connection: + assert connection.execute(text("select count(*) from asset")).scalar_one() == 2 + store.close() + + +def test_migration_eight_refuses_a_workspace_holding_duplicate_sources(tmp_path: Path) -> None: + """The rule #18 shipped as a pre-check, meeting data written while it was one.""" + 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')") + ) + # The index has to go before the rows it would refuse, which is exactly + # the situation a workspace written by the previous build is in. + connection.execute(text("drop index uq_source_project_kind_path_fps")) + for source_id in ("s1", "s2"): + connection.execute( + text( + "insert into source (id, project_id, kind, path, registered_at, " + f"capture_params) values ('{source_id}', 'p', 'image_directory', '/in', " + "'2026-07-27T00:00:00+00:00', '{}')" + ) + ) + connection.execute(text("update _visionset_meta set format_version = 7")) + + with pytest.raises(WorkspaceCorrupt, match="duplicate sources"): + store.initialize() + + with store.engine.connect() as connection: + assert connection.execute(text("select count(*) from source")).scalar_one() == 2 + store.close() + + +def test_migration_eight_refuses_a_workspace_that_still_holds_a_pre_ingest_job( + tmp_path: Path, +) -> None: + """``ingest_job`` is rebuilt, so like migrations 6 and 7 it counts first. + + Set up at generation 7 rather than 1, so that migration 8's own count is + what fires: from generation 1 migration 7 refuses the workspace before this + ever runs, because a pre-#18 source is what such a job always hangs from. + """ + 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', '{}')" + ) + ) + # Rebuilt rather than altered, for the reason migration 8 rebuilds it: + # SQLite will not drop a column that a foreign-key clause names. + connection.execute(text("drop table ingest_job")) + connection.execute( + text( + "CREATE TABLE ingest_job (" + " id CHAR(32) NOT NULL," + " source_id CHAR(32) NOT NULL," + " state VARCHAR NOT NULL," + " error VARCHAR," + " PRIMARY KEY (id)," + " FOREIGN KEY(source_id) REFERENCES source (id) ON DELETE CASCADE)" + ) + ) + connection.execute( + text("insert into ingest_job (id, source_id, state) values ('j', 's', 'pending')") + ) + connection.execute(text("update _visionset_meta set format_version = 7")) + + with pytest.raises(WorkspaceCorrupt, match="before IngestService existed"): + store.initialize() + + with store.engine.connect() as connection: + assert connection.execute(text("select count(*) from ingest_job")).scalar_one() == 1 + 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.