From c6a729a4211b35e197a3867861814c91a8d0ba84 Mon Sep 17 00:00:00 2001 From: Armando Anaya Date: Mon, 27 Jul 2026 12:10:57 -0700 Subject: [PATCH] =?UTF-8?q?feat(kernel):=20SQLite=20concurrency=20posture?= =?UTF-8?q?=20=E2=80=94=20WAL,=20busy=5Ftimeout,=20OperationalError=20?= =?UTF-8?q?=E2=86=92=20WorkspaceBusy=20(#80)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the last untranslated adapter exception before background ingest makes lock contention real. - WAL journal mode, set in `initialize()` rather than on every connection: switching to WAL writes the file header, and a connect-time pragma would leave a 4 KB mark on any stranger's file merely inspected by `format_version` — breaking the invariant that `open` creates nothing when it refuses. - `busy_timeout`, 5 s by default, on every connection. Keyword-only with a default so the class still satisfies `MetadataStoreFactory`. - One translation site, `_translated`, that every entry point routes through: `IntegrityError` → `ConstraintViolated`, `SQLITE_BUSY`/`SQLITE_LOCKED` → `WorkspaceBusy`, everything else → `WorkspaceCorrupt`. `unit_of_work` now catches `DatabaseError` rather than only `IntegrityError`, closing a hole where a non-constraint failure inside a transaction escaped raw. - Contention is told from damage by SQLite's result code, not by message text. - WAL sidecars are part of the workspace layout: `_undo_init` and both examples' cleanup now account for them. No migration: FORMAT_VERSION stays 10, VERSION stays 0.0.1.dev0, no openapi.json drift, no dependency change. --- docs/ingest.md | 3 +- docs/persistence.md | 48 +++- docs/sources.md | 8 +- docs/workspaces.md | 59 +++-- examples/ingest_end_to_end.py | 5 +- examples/sdk_end_to_end.py | 6 +- src/visionset/kernel/__init__.py | 2 + .../kernel/adapters/sqlite_metadata_store.py | 147 ++++++++++--- src/visionset/kernel/errors.py | 29 ++- .../kernel/services/annotation_service.py | 2 +- .../kernel/services/ingest_service.py | 5 +- .../kernel/services/source_service.py | 11 +- .../kernel/services/workspace_service.py | 16 ++ tests/kernel/test_concurrency.py | 205 ++++++++++++++++++ tests/kernel/test_metadata_store.py | 19 ++ 15 files changed, 492 insertions(+), 73 deletions(-) create mode 100644 tests/kernel/test_concurrency.py diff --git a/docs/ingest.md b/docs/ingest.md index 897b6cae..b6b4b391 100644 --- a/docs/ingest.md +++ b/docs/ingest.md @@ -150,7 +150,8 @@ 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 +SQLite store starts making every other writer wait out its `busy_timeout` and fail with +`WorkspaceBusy`. 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. diff --git a/docs/persistence.md b/docs/persistence.md index b09a84bb..22dc946e 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -89,6 +89,48 @@ 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. +## Connection posture + +Three settings, and they are applied in two different places for one reason: + +| Setting | Where | Why there | +| --- | --- | --- | +| `PRAGMA foreign_keys = ON` | every connection | Per connection, and SQLite forgets it on close. Reads nothing, writes nothing. | +| `PRAGMA busy_timeout = 5000` | every connection | Likewise per connection. Configurable: `SqliteMetadataStore(path, busy_timeout_ms=…)`. | +| `PRAGMA journal_mode = WAL` | `initialize()` only | **Switching to WAL writes the file header.** | + +That last row is the interesting one. WAL is recorded in the database header and persists, so +it only ever needs setting once — but turning it on *grows an empty file to a full page*, and +`WorkspaceService.open` reads `format_version` before it has decided the file is a workspace +at all. A connect-time pragma would therefore leave a 4 KB mark on any stranger's file merely +inspected, breaking the invariant that **`open` creates nothing when it refuses**. Setting it +in `initialize()` puts it exactly where the caller has already established the file is ours to +write to, and re-running it there is how a workspace written before WAL converts on its next +open. It also has to sit outside the migration transaction: SQLite refuses to change journal +mode inside one. + +The consequences for callers are in +[workspaces.md § Concurrency, plainly](workspaces.md#concurrency-plainly): readers never +block, writers are serialized and wait up to the timeout, and a wait that runs out is +`WorkspaceBusy`. + +### No SQLAlchemy exception escapes + +Every `DatabaseError` the engine raises goes through one function in the adapter, and the +order of its tests is the dispatch: + +| SQLAlchemy raises | Becomes | Because | +| --- | --- | --- | +| `IntegrityError` | `ConstraintViolated` | A constraint refused the write. Ends the transaction. | +| `OperationalError` with a `SQLITE_BUSY` / `SQLITE_LOCKED` result code | `WorkspaceBusy` | Contention. Transient — retry. | +| any other `DatabaseError`, including the rest of `OperationalError` | `WorkspaceCorrupt` | Cannot open the file, disk I/O error, disk full, not a database. Waiting will not fix it. | + +Contention is told apart from damage by SQLite's **result code** (`sqlite_errorname`), not by +the wording of the message, so a reworded SQLite release cannot silently reroute a lock into +`WorkspaceCorrupt`. `WorkspaceCorrupt` is the widest of the three deliberately: it is where +everything unusable-for-a-reason-you-cannot-wait-out lands, and splitting it further would +invent errors nobody catches. + `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 @@ -131,10 +173,10 @@ missing: | lower | the pending migrations run; the file is restamped | | higher | `WorkspaceFormatTooNew` — migrations only run forward | | not a readable database | `WorkspaceCorrupt` | +| held by another writer past the timeout | `WorkspaceBusy` | -`OperationalError` — "database is locked", "unable to open database file" — is deliberately -*not* translated. Those are environmental, not structural, and calling them corruption would -be a lie. It is a known gap: a SQLAlchemy exception can still escape on a locked file. +`initialize()` also switches the file to WAL, for the reason given under +[Connection posture](#connection-posture). **Adding a migration:** append a `Migration` with the next version and an `upgrade` taking a live `Connection`. Never edit an existing migration — a workspace already stamped at that diff --git a/docs/sources.md b/docs/sources.md index 46c98e36..c08cc679 100644 --- a/docs/sources.md +++ b/docs/sources.md @@ -27,8 +27,8 @@ anything used it. `register_video` probes the file through the workspace's `VideoProcessor` and stores the answer. The probe runs **before** the transaction opens: it is an out-of-process decoder, and holding a -write transaction open across a subprocess is how a single-writer SQLite store ends up reporting -"database is locked". +write transaction open across a subprocess is how a single-writer SQLite store ends up making +every other writer wait out its `busy_timeout` and fail with `WorkspaceBusy`. ## What a source records @@ -98,8 +98,8 @@ 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. +row and returns it. A caller that instead waits out the store's `busy_timeout` sees +`WorkspaceBusy`, and the remedy is the same. ## Paths are canonicalized once diff --git a/docs/workspaces.md b/docs/workspaces.md index f3ee620f..bad97915 100644 --- a/docs/workspaces.md +++ b/docs/workspaces.md @@ -7,13 +7,19 @@ happens in the context of exactly one, so `WorkspaceService` is both the way in ``` / visionset.db the metadata store — its presence is what makes a directory a workspace + visionset.db-wal SQLite's write-ahead log — present only while the workspace is open + visionset.db-shm its shared-memory index — likewise blobs/ FilesystemBlobStore root, sharded // ``` -Nothing else is written. WAL is deliberately **not** enabled in M1, so those two entries -are the whole format — if WAL is ever adopted, `visionset.db-wal` and `-shm` become part -of it and this page has to say so, because a user who copies only `visionset.db` under WAL -loses committed data. +Nothing else is written. The store runs in **WAL mode**, which is why the two sidecars are +part of the format: `close()` checkpoints them into `visionset.db` and removes them, so a +workspace at rest is still just the database and the blobs — but a workspace that is *open*, +or one whose process was killed, is all four entries. + +**Copying a workspace that is open loses committed data** if you take only `visionset.db`. +Everything written since the last checkpoint lives in `visionset.db-wal` until then. Close the +workspace first, or copy all three files together. Three of the five ports have no line in that layout, and that is the point: the [event bus](events.md) is in-process and the two [media processors](media.md) are decoders, so none of @@ -177,27 +183,46 @@ than the index, so nothing slips through. ## Concurrency, plainly -One SQLite file, rollback-journal mode, no cross-process lock, one engine per open -`WorkspaceService`. +One SQLite file in **WAL mode**, a **5 s `busy_timeout`** on every connection, no cross-process +lock, one engine per open `WorkspaceService`. **Guaranteed, under any number of processes:** a workspace can never contain two `project` rows whose names are equal under ASCII case folding. That holds because the guarantee is an index evaluated inside SQLite's write transaction, not the service's `SELECT`. Every write is a transaction, so no half-finished operation can be observed. -**Not guaranteed:** which error the loser of a race sees. Two processes can both pass the -pre-check; then either the second insert hits the index and raises `ConstraintViolated` — a -caller re-raises that as `ProjectNameTaken`, so user-visible behavior stays correct — or the -writes interleave and the loser eventually gets `database is locked` as an untranslated -SQLAlchemy `OperationalError`. That leak is a known M1 gap: mapping it to `WorkspaceCorrupt` -would be a lie, and inventing an error for transient failure with no caller would be -speculative. +**Readers never block, and are never blocked.** That is what WAL buys, and it is the reason +this milestone adopted it: a background ingest holding a write transaction does not stall the +request handlers reading beside it. A reader sees the last committed state, not the writer's +work in progress. + +**Writers are serialized, and wait up to `busy_timeout` for their turn.** SQLite permits one +writer at a time regardless of journal mode. A write that waits the timeout out raises +`WorkspaceBusy` — a domain error, translated in the adapter like every other (see +[persistence.md](persistence.md#connection-posture)), and transient: the remedy is to retry. +It is deliberately *not* `WorkspaceCorrupt`, which is +where the adapter files the failures retrying cannot fix. + +**Still not guaranteed:** which error the loser of a name race sees. Two processes can both +pass the pre-check; then either the second insert hits the index and raises +`ConstraintViolated` — a caller re-raises that as `ProjectNameTaken`, so user-visible +behavior stays correct — or one of them waits out the timeout and gets `WorkspaceBusy`. Both +are honest answers, and both are now domain errors. Opening the same path twice yields two independent engines with no shared cache and no -in-process lock: **VisionSet is single-writer by convention, not by enforcement.** Hardenings -in the order they would be taken: `PRAGMA busy_timeout`, `BEGIN IMMEDIATE` for write -transactions (which removes the race entirely), WAL (changes the on-disk format), an advisory -lock file. +in-process lock: **VisionSet is single-writer by convention, not by enforcement.** A long +write transaction is therefore still a thing to avoid rather than a thing the store defends +against — which is why, for example, `SourceService` probes a clip *outside* its write +transaction. `busy_timeout` shortens the window; it does not make holding a transaction +across a subprocess acceptable. + +Two hardenings remain untaken, and one of them is declined rather than merely pending. +`BEGIN IMMEDIATE` for write transactions would make every contended write wait instead of +ever failing fast — but `unit_of_work()` serves reads and writes alike, with no read-only +variant, so an immediate transaction would take the write lock for *every* read and serialize +exactly the concurrency WAL was adopted for. It stays off unless the unit of work grows a +read-only form. An advisory lock file, which would turn "single-writer by convention" into +enforcement, is still open. ## How later services are composed diff --git a/examples/ingest_end_to_end.py b/examples/ingest_end_to_end.py index d4f4a2ca..bd6b79de 100644 --- a/examples/ingest_end_to_end.py +++ b/examples/ingest_end_to_end.py @@ -364,7 +364,10 @@ 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") - ours = {"visionset.db", "blobs", "clips", "incoming"} + # The two ``-wal``/``-shm`` entries are SQLite's WAL sidecars. A clean close + # removes them, so they are only here if a previous run was killed — which + # is exactly when this function has to be able to clean up. + ours = {"visionset.db", "visionset.db-wal", "visionset.db-shm", "blobs", "clips", "incoming"} stray = {entry.name for entry in dest.iterdir()} - ours if stray: raise SystemExit( diff --git a/examples/sdk_end_to_end.py b/examples/sdk_end_to_end.py index d1701f59..025472a9 100644 --- a/examples/sdk_end_to_end.py +++ b/examples/sdk_end_to_end.py @@ -438,7 +438,11 @@ 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", "incoming"} + # The two ``-wal``/``-shm`` entries are SQLite's WAL sidecars. A clean close + # removes them, so they are only here if a previous run was killed — which + # is exactly when this function has to be able to clean up. + ours = {"visionset.db", "visionset.db-wal", "visionset.db-shm", "blobs", "incoming"} + stray = {entry.name for entry in dest.iterdir()} - ours 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 38418d76..e23e6e17 100644 --- a/src/visionset/kernel/__init__.py +++ b/src/visionset/kernel/__init__.py @@ -53,6 +53,7 @@ UnsupportedMedia, VisionSetError, WorkspaceAlreadyExists, + WorkspaceBusy, WorkspaceCorrupt, WorkspaceFormatTooNew, WorkspaceNotEmpty, @@ -105,6 +106,7 @@ "UnsupportedMedia", "VisionSetError", "WorkspaceAlreadyExists", + "WorkspaceBusy", "WorkspaceCorrupt", "WorkspaceFormatTooNew", "WorkspaceNotEmpty", diff --git a/src/visionset/kernel/adapters/sqlite_metadata_store.py b/src/visionset/kernel/adapters/sqlite_metadata_store.py index dd573469..a8cb793f 100644 --- a/src/visionset/kernel/adapters/sqlite_metadata_store.py +++ b/src/visionset/kernel/adapters/sqlite_metadata_store.py @@ -5,19 +5,21 @@ row/model translation lives in ``_mappers`` so that nothing SQLAlchemy-shaped escapes this package. -That last part includes exceptions. SQLAlchemy's ``IntegrityError`` and -``DatabaseError`` are translated here into ``ConstraintViolated`` and -``WorkspaceCorrupt``, because a service that had to catch them would need a -SQLAlchemy import to do it — which is exactly the leak this package exists to -prevent. +That last part includes exceptions. Every ``DatabaseError`` SQLAlchemy raises is +translated by :func:`_translated` — into ``ConstraintViolated``, +``WorkspaceBusy`` or ``WorkspaceCorrupt`` — because a service that had to catch +the originals would need a SQLAlchemy import to do it, which is exactly the leak +this package exists to prevent. There is one such function and every entry point +routes through it, so "no SQLAlchemy exception escapes" is a property of one +place rather than a habit spread across several. """ from __future__ import annotations -from collections.abc import Iterator +from collections.abc import Callable, Iterator from contextlib import contextmanager from pathlib import Path -from typing import Any +from typing import Any, Final from uuid import UUID from sqlalchemy import ( @@ -42,11 +44,26 @@ ConstraintViolated, EntityAlreadyExists, EntityNotFound, + VisionSetError, + WorkspaceBusy, WorkspaceCorrupt, WorkspaceFormatTooNew, ) from visionset.kernel.ports.metadata_store import UNINITIALIZED, UnitOfWork +#: How long a connection waits for a lock before giving up, in milliseconds. +#: Long enough to absorb the write bursts of a background ingest running beside +#: request handlers, short enough that a wedged writer surfaces as an error +#: rather than as a request that never returns. +DEFAULT_BUSY_TIMEOUT_MS: Final = 5_000 + +#: SQLite's own names for "someone else has it". ``sqlite_errorname`` reports +#: the *extended* result code, so a prefix test covers ``SQLITE_BUSY_SNAPSHOT`` +#: and ``SQLITE_BUSY_TIMEOUT`` without listing them. Matching the name rather +#: than the message text means a reworded SQLite release cannot silently reroute +#: contention into ``WorkspaceCorrupt``. +_BUSY_ERROR_NAMES: Final = ("SQLITE_BUSY", "SQLITE_LOCKED") + def _constraint_violated(exc: IntegrityError) -> ConstraintViolated: """Translate SQLite's constraint complaint into a domain error. @@ -59,36 +76,65 @@ def _constraint_violated(exc: IntegrityError) -> ConstraintViolated: return ConstraintViolated(str(exc.orig)) -@contextmanager -def _readable(db_path: Path) -> Iterator[None]: - """Report an unreadable database as ``WorkspaceCorrupt``, not as SQLAlchemy. +def _is_busy(exc: OperationalError) -> bool: + """Whether this is contention rather than damage.""" + name = getattr(exc.orig, "sqlite_errorname", "") + if name: + return name.startswith(_BUSY_ERROR_NAMES) + # A DBAPI that does not carry the result code — nothing does today, but the + # engine URL is not a promise — leaves only the wording to go on. + return "is locked" in str(exc.orig) + - ``OperationalError`` is re-raised untranslated on purpose: "database is - locked" and "unable to open database file" are environmental, and calling - them corruption would be a lie. Surfacing them as a domain error needs a - port vocabulary for transient failure, which nothing needs yet. +def _translated(exc: DatabaseError, db_path: Path) -> VisionSetError: + """The adapter's entire exception vocabulary, in one place. + + ``IntegrityError`` and ``OperationalError`` are both ``DatabaseError`` + subclasses, so the order of these tests *is* the dispatch. Everything that + reaches the last line is a database this build cannot work with, which is + what ``WorkspaceCorrupt`` means — including the environmental failures + (cannot open the file, disk I/O error, disk full) that waiting will not fix. """ + if isinstance(exc, IntegrityError): + return _constraint_violated(exc) + if isinstance(exc, OperationalError) and _is_busy(exc): + return WorkspaceBusy( + f"{db_path} is held by another writer and the wait ran out: {exc.orig}" + ) + return WorkspaceCorrupt(f"{db_path} is not a readable VisionSet metadata store: {exc.orig}") + + +@contextmanager +def _readable(db_path: Path) -> Iterator[None]: + """Let no SQLAlchemy exception out of a read or a schema change.""" try: yield - except OperationalError: - raise - except IntegrityError as exc: - raise _constraint_violated(exc) from exc except DatabaseError as exc: - raise WorkspaceCorrupt( - f"{db_path} is not a readable VisionSet metadata store: {exc.orig}" - ) from exc + raise _translated(exc, db_path) from exc + +def _connection_posture(busy_timeout_ms: int) -> Callable[[Any, Any], None]: + """Build the ``connect`` listener that says what a connection is. -def _enable_foreign_keys(dbapi_connection: Any, _: Any) -> None: - """SQLite ships with foreign keys OFF, per connection. + Both pragmas are per connection — SQLite forgets them on close — so they are + re-issued on every connection the engine opens, including the one that runs + migrations, since that is the same engine. Neither writes to the file, which + is what lets them run against a database this build has not yet vouched for. - Without this every ``ForeignKey`` in ``_tables`` is decorative: orphan rows - would insert happily and cascades would never fire. + ``journal_mode`` is deliberately *not* here; see :meth:`SqliteMetadataStore. + initialize`. """ - cursor = dbapi_connection.cursor() - cursor.execute("PRAGMA foreign_keys = ON") - cursor.close() + + def listener(dbapi_connection: Any, _: Any) -> None: + cursor = dbapi_connection.cursor() + # Without this every ``ForeignKey`` in ``_tables`` is decorative: SQLite + # ships with foreign keys off, so orphan rows would insert happily and + # cascades would never fire. + cursor.execute("PRAGMA foreign_keys = ON") + cursor.execute(f"PRAGMA busy_timeout = {busy_timeout_ms}") + cursor.close() + + return listener def _stored_format_version(connection: Connection) -> int | None: @@ -123,6 +169,11 @@ def _sync_children(self, entity: T) -> None: self._mapping.sync_children(self._session, entity) def _flush(self) -> None: + # Only the constraint case is caught here, so that a caller wrapping a + # single ``add()`` sees the right type at the point of the call. Anything + # broader — a lock, an unusable file — propagates to ``unit_of_work``, + # which translates it on the way out; there is nothing to gain from + # naming those twice. try: self._session.flush() except IntegrityError as exc: @@ -192,14 +243,26 @@ def __init__(self, session: Session) -> None: class SqliteMetadataStore: - def __init__(self, db_path: Path) -> None: + """One SQLite file, in WAL mode, with a bounded wait for contention. + + ``busy_timeout_ms`` is keyword-only with a default so that the class itself + still satisfies ``MetadataStoreFactory`` — a plain ``Callable[[Path], + MetadataStore]`` — and can go on being passed to ``WorkspaceService.init`` + and ``open`` as a bare class reference. A caller who wants a different wait + supplies ``partial(SqliteMetadataStore, busy_timeout_ms=...)`` as the + factory rather than growing the port's signature for a tuning knob. + """ + + def __init__(self, db_path: Path, *, busy_timeout_ms: int = DEFAULT_BUSY_TIMEOUT_MS) -> None: db_path.parent.mkdir(parents=True, exist_ok=True) self._db_path = db_path #: Built through ``URL.create`` rather than an f-string: a path containing #: ``#`` or ``?`` parses as a URL fragment or query and the engine would #: silently target a *different* file (a truncated sibling of this one). self._engine: Engine = create_engine(URL.create("sqlite", database=str(db_path))) - event.listen(self._engine, "connect", _enable_foreign_keys) + #: Registered on this engine, never on the ``Engine`` class: a global + #: listener would reach engines this store knows nothing about. + event.listen(self._engine, "connect", _connection_posture(busy_timeout_ms)) @property def engine(self) -> Engine: @@ -218,7 +281,20 @@ def initialize(self) -> None: A fresh file gets every migration and is stamped at ``FORMAT_VERSION``; an existing one gets only what it is missing. A file stamped ahead of this build raises rather than being opened on a guess. + + Journal mode is set here rather than on every connection, and the reason + is an invariant one file up: ``WorkspaceService.open`` creates nothing + when it refuses. Switching a database to WAL *writes its header* — an + empty file grows to a full page — so a connect-time pragma would leave a + 4 KB mark on any stranger's file merely inspected by ``format_version``. + By the time this method runs, the caller has established that the file is + ours to write to. WAL is recorded in the header and persists, so setting + it once is what it takes, and re-running it is how a workspace written + before WAL converts on its next open. It also has to sit outside the + migration transaction: SQLite refuses to change journal mode inside one. """ + with _readable(self._db_path), self._engine.connect() as connection: + connection.exec_driver_sql("PRAGMA journal_mode = WAL") with _readable(self._db_path), self._engine.begin() as connection: stored = _stored_format_version(connection) if stored is not None and stored > FORMAT_VERSION: @@ -240,10 +316,13 @@ def unit_of_work(self) -> Iterator[UnitOfWork]: try: with session.begin(): yield SqlUnitOfWork(session) - except IntegrityError as exc: - # A constraint can also fire at commit time, i.e. after the last - # repository call has already returned. - raise _constraint_violated(exc) from exc + except DatabaseError as exc: + # Broader than the repositories' own translation, because this is + # the last place anything can be caught: a constraint and a lock + # can both fire at commit time, i.e. after the final repository + # call has already returned cleanly. + raise _translated(exc, self._db_path) from exc def close(self) -> None: + """Dispose the engine, which checkpoints and removes the WAL sidecars.""" self._engine.dispose() diff --git a/src/visionset/kernel/errors.py b/src/visionset/kernel/errors.py index 67849ef5..297eeb15 100644 --- a/src/visionset/kernel/errors.py +++ b/src/visionset/kernel/errors.py @@ -62,12 +62,33 @@ class WorkspaceAlreadyExists(VisionSetError): class WorkspaceCorrupt(VisionSetError): - """The workspace layout is present but unusable. + """The workspace layout is present but unusable, whatever the cause. A metadata store that is not a readable database, that carries no VisionSet - schema, or that does not hold exactly one workspace row. Distinct from - ``NotAWorkspace`` (nothing there) and from ``WorkspaceFormatTooNew`` - (readable, merely newer than this build). + schema, or that does not hold exactly one workspace row — and also the + environmental failures that leave nothing to work with: a file that cannot + be opened, a disk I/O error, a full disk. Distinct from ``NotAWorkspace`` + (nothing there), from ``WorkspaceFormatTooNew`` (readable, merely newer than + this build), and from ``WorkspaceBusy`` (fine, just held by someone else). + + "Corrupt" is the widest of these on purpose: it is where the adapter files + everything it could not open, read or write for a reason the caller cannot + fix by waiting. Splitting the causes further would invent errors nobody + catches — the remedy for all of them is to look at the file and the disk. + """ + + +class WorkspaceBusy(VisionSetError): + """Another connection holds the workspace and the wait ran out. + + Transient, unlike ``WorkspaceCorrupt``: nothing is wrong with the file, + something else is writing to it. The remedy is to retry, which is the whole + reason this is its own error rather than a flavour of corruption — a surface + can answer it with a retry-after where corruption gets a hard failure. + + Raised when a write waited out the store's ``busy_timeout`` behind another + writer. Under WAL a reader never reaches here, because readers do not + contend with the writer at all. """ diff --git a/src/visionset/kernel/services/annotation_service.py b/src/visionset/kernel/services/annotation_service.py index eeabf582..16aa3675 100644 --- a/src/visionset/kernel/services/annotation_service.py +++ b/src/visionset/kernel/services/annotation_service.py @@ -268,7 +268,7 @@ def _pinned_schema(self, batch: Batch) -> AnnotationSchema: A nested read, which is why every caller does it before its first write: ``unit_of_work()`` opens a fresh session per call, and a second *writer* - on the same file is how "database is locked" happens. + on the same file is how a ``WorkspaceBusy`` happens. ``Batch.schema_version`` is ``None`` only while a batch is a draft, and the caller has already established that this one is ``in_annotation``. diff --git a/src/visionset/kernel/services/ingest_service.py b/src/visionset/kernel/services/ingest_service.py index 0108be79..5a8c9b9d 100644 --- a/src/visionset/kernel/services/ingest_service.py +++ b/src/visionset/kernel/services/ingest_service.py @@ -22,8 +22,9 @@ **The long middle of the run is in no transaction.** Decoding is a Pillow pass over thousands of files or an out-of-process ffmpeg, and holding a write -transaction open across either is how a single-writer SQLite store starts -reporting "database is locked" (#80). So the run resolves what it needs, closes +transaction open across either is how a single-writer SQLite store starts making +every other writer wait out its ``busy_timeout`` and fail with ``WorkspaceBusy``. +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 diff --git a/src/visionset/kernel/services/source_service.py b/src/visionset/kernel/services/source_service.py index 1032e14e..243ffc39 100644 --- a/src/visionset/kernel/services/source_service.py +++ b/src/visionset/kernel/services/source_service.py @@ -32,8 +32,8 @@ 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. +finds the winner's row and returns it. A caller that instead waits out the +store's ``busy_timeout`` sees ``WorkspaceBusy``, and the remedy is the same. Composition follows the rule in ``docs/workspaces.md``: this service takes an open ``WorkspaceService`` and nothing else, and reaches ``video_processor`` @@ -116,9 +116,10 @@ def register_video( The probe runs **before** the transaction opens. It is an out-of-process decoder, and holding a write transaction open across a subprocess is how - a single-writer SQLite store ends up reporting "database is locked" — - the same reason ``examples/sdk_end_to_end.py`` puts its blob writes - outside the ``unit_of_work``. + a single-writer SQLite store ends up making every other writer wait out + its ``busy_timeout`` and fail with ``WorkspaceBusy`` — the same reason + ``examples/sdk_end_to_end.py`` puts its blob writes outside the + ``unit_of_work``. The consequence is worth knowing: re-registering an already-known clip still needs ffmpeg, because the freshly probed metadata is what keeps the diff --git a/src/visionset/kernel/services/workspace_service.py b/src/visionset/kernel/services/workspace_service.py index 9e78f7b4..601a8f13 100644 --- a/src/visionset/kernel/services/workspace_service.py +++ b/src/visionset/kernel/services/workspace_service.py @@ -13,8 +13,15 @@ The layout is flat, and the database is the marker:: /visionset.db the metadata store; holds format_version + /visionset.db-wal SQLite's write-ahead log, only while open + /visionset.db-shm its shared-memory index, only while open /blobs/ the content-addressed blob store +The store runs in WAL mode, so those two sidecars exist for as long as the +workspace is open and are checkpointed away by ``close()``. They still belong to +the workspace while they are there: a copy taken mid-run that includes only +``visionset.db`` is missing whatever has not been checkpointed yet. + Three of the five ports have no line in that layout, and that is the point: the event bus is in-process and the two media processors are decoders, so none of them leaves anything behind. They are composed here anyway, because a workspace @@ -77,6 +84,13 @@ #: The metadata store. Its presence is what makes a directory a workspace. DB_FILENAME = "visionset.db" +#: SQLite's WAL sidecars, named after the database file. A clean ``close()`` +#: checkpoints and removes them, so they are only ever found beside a workspace +#: that is open right now or was killed while open — but they are part of the +#: workspace either way, which is why everything that enumerates its contents +#: has to know about them. +DB_SIDECAR_FILENAMES = (f"{DB_FILENAME}-wal", f"{DB_FILENAME}-shm") + #: Root of the content-addressed blob store, relative to the workspace directory. BLOBS_DIRNAME = "blobs" @@ -452,6 +466,8 @@ def _undo_init(root: Path, *, created_root: bool) -> None: shutil.rmtree(root, ignore_errors=True) return (root / DB_FILENAME).unlink(missing_ok=True) + for sidecar in DB_SIDECAR_FILENAMES: + (root / sidecar).unlink(missing_ok=True) shutil.rmtree(root / BLOBS_DIRNAME, ignore_errors=True) except OSError: pass diff --git a/tests/kernel/test_concurrency.py b/tests/kernel/test_concurrency.py new file mode 100644 index 00000000..fdaf001d --- /dev/null +++ b/tests/kernel/test_concurrency.py @@ -0,0 +1,205 @@ +"""The store's connection posture, and what two connections do to each other. + +The suite's only threaded tests. Everything here sequences on `threading.Event` +rather than on sleeps, and every thread is joined with a timeout and then +asserted dead — a concurrency test that hangs is a concurrency test nobody runs. + +Two `SqliteMetadataStore` instances over one file is the shape under test: that +is two engines with no shared cache and no in-process lock, which is what two +*processes* look like from SQLite's side. `WorkspaceService` composes exactly one +store per open workspace, so this is the only place that arrangement is built on +purpose. +""" + +from __future__ import annotations + +import threading +from collections.abc import Iterator +from contextlib import contextmanager +from pathlib import Path +from uuid import uuid4 + +import pytest +from sqlalchemy import text + +from visionset.kernel import ConstraintViolated, WorkspaceBusy, WorkspaceCorrupt +from visionset.kernel.adapters import SqliteMetadataStore +from visionset.kernel.adapters.sqlite_metadata_store import DEFAULT_BUSY_TIMEOUT_MS +from visionset.kernel.domain import Project, Workspace +from visionset.kernel.ports import UNINITIALIZED + +#: Every wait in this file. Long enough that a loaded CI runner does not trip it, +#: short enough that a genuine deadlock fails the suite instead of stalling it. +TIMEOUT_SECONDS = 30.0 + + +def _store(tmp_path: Path, **kwargs: int) -> SqliteMetadataStore: + store = SqliteMetadataStore(tmp_path / "visionset.db", **kwargs) + store.initialize() + return store + + +def _pragma(store: SqliteMetadataStore, name: str) -> object: + with store.engine.connect() as connection: + return connection.execute(text(f"pragma {name}")).scalar() + + +def _seed(store: SqliteMetadataStore) -> None: + with store.unit_of_work() as uow: + uow.workspaces.add(Workspace(name="w", root_dir="/tmp/w")) + + +@contextmanager +def _write_held_open(store: SqliteMetadataStore) -> Iterator[None]: + """Hold an open write transaction on another thread for the body's duration. + + Enters only once the write has actually landed — so the SQLite write lock is + genuinely held, not merely about to be — and joins the thread on the way out. + A failure inside the thread is re-raised here rather than printed and lost. + """ + writing = threading.Event() + release = threading.Event() + failure: list[BaseException] = [] + + def hold() -> None: + try: + with store.unit_of_work() as uow: + uow.workspaces.add(Workspace(name="held", root_dir="/tmp/held")) + writing.set() + release.wait(TIMEOUT_SECONDS) + except BaseException as exc: # noqa: BLE001 - re-raised on the main thread + failure.append(exc) + finally: + writing.set() + + thread = threading.Thread(target=hold, name="held-write") + thread.start() + try: + assert writing.wait(TIMEOUT_SECONDS), "the holding thread never took the write lock" + if failure: + raise failure[0] + yield + finally: + release.set() + thread.join(TIMEOUT_SECONDS) + assert not thread.is_alive(), "the holding thread never finished" + + +def test_a_fresh_store_is_in_wal_mode(tmp_path: Path) -> None: + store = _store(tmp_path) + assert _pragma(store, "journal_mode") == "wal" + store.close() + + +def test_wal_persists_in_the_file_rather_than_in_the_process(tmp_path: Path) -> None: + """Journal mode lives in the header, which is what converts an old workspace.""" + store = _store(tmp_path) + store.close() + + reopened = SqliteMetadataStore(tmp_path / "visionset.db") + assert _pragma(reopened, "journal_mode") == "wal" + reopened.close() + + +def test_reading_the_format_version_does_not_switch_a_stray_file_to_wal(tmp_path: Path) -> None: + """WAL is set by `initialize()`, because turning it on writes the header. + + `WorkspaceService.open` creates nothing when it refuses, and it asks for + `format_version` before it has decided the file is ours. A connect-time + journal-mode pragma would leave a 4 KB page on every file it merely looked + at — which is what this asserts did not happen. + """ + stray = tmp_path / "visionset.db" + stray.touch() + + store = SqliteMetadataStore(stray) + assert store.format_version == UNINITIALIZED + store.close() + + assert stray.stat().st_size == 0 + + +def test_a_clean_close_leaves_no_sidecars_behind(tmp_path: Path) -> None: + """`close()` disposes the engine, which checkpoints the WAL and removes it.""" + store = _store(tmp_path) + _seed(store) + store.close() + + assert sorted(p.name for p in tmp_path.iterdir()) == ["visionset.db"] + + +def test_every_connection_carries_the_busy_timeout(tmp_path: Path) -> None: + store = _store(tmp_path) + # Read twice, on two separate connections: the pragma is per connection, so a + # listener that fired only for the first one would show up right here. + assert _pragma(store, "busy_timeout") == DEFAULT_BUSY_TIMEOUT_MS + assert _pragma(store, "busy_timeout") == DEFAULT_BUSY_TIMEOUT_MS + store.close() + + +def test_the_busy_timeout_is_configurable(tmp_path: Path) -> None: + store = _store(tmp_path, busy_timeout_ms=250) + assert _pragma(store, "busy_timeout") == 250 + store.close() + + +def test_foreign_keys_survive_the_wal_posture(tmp_path: Path) -> None: + """The pragma migration 007 depends on, guarded against a listener rewrite.""" + store = _store(tmp_path) + assert _pragma(store, "foreign_keys") == 1 + with pytest.raises(ConstraintViolated, match="FOREIGN KEY"), store.unit_of_work() as uow: + uow.projects.add(Project(workspace_id=uuid4(), name="orphan")) + store.close() + + +def test_a_reader_proceeds_while_a_write_transaction_is_open(tmp_path: Path) -> None: + """The whole point of WAL: a reader is not blocked by the writer. + + Under the rollback journal this store used before, the read below waits out + the busy timeout and then fails. + """ + writer = _store(tmp_path) + _seed(writer) + reader = SqliteMetadataStore(tmp_path / "visionset.db", busy_timeout_ms=250) + + with _write_held_open(writer), reader.unit_of_work() as uow: + names = [row.name for row in uow.workspaces.list()] + + # The pre-write state, read without waiting: the held row is still + # uncommitted, so this is isolation working rather than staleness. + assert names == ["w"] + reader.close() + writer.close() + + +def test_a_write_that_outlives_the_busy_timeout_is_reported_busy(tmp_path: Path) -> None: + """Contention becomes `WorkspaceBusy` — never a SQLAlchemy exception.""" + holder = _store(tmp_path) + _seed(holder) + contender = SqliteMetadataStore(tmp_path / "visionset.db", busy_timeout_ms=0) + + # `pytest.raises` is listed before the unit of work so that it is still open + # when the transaction exits: with `busy_timeout_ms=0` the lock can be + # reported either at the write or at the commit, and both have to be caught. + with ( + _write_held_open(holder), + pytest.raises(WorkspaceBusy) as caught, + contender.unit_of_work() as uow, + ): + uow.workspaces.add(Workspace(name="third", root_dir="/tmp/third")) + + assert "sqlalchemy" not in type(caught.value).__module__ + assert "held by another writer" in str(caught.value) + contender.close() + holder.close() + + +def test_contention_and_damage_are_different_errors() -> None: + """`WorkspaceBusy` is not a `WorkspaceCorrupt`, in either direction. + + They arrive at the same `except` clause in the adapter and are told apart + only by the SQLite result code, so nothing but this keeps them from + collapsing into one during a later edit. + """ + assert not issubclass(WorkspaceBusy, WorkspaceCorrupt) + assert not issubclass(WorkspaceCorrupt, WorkspaceBusy) diff --git a/tests/kernel/test_metadata_store.py b/tests/kernel/test_metadata_store.py index c3443051..c4d9ad31 100644 --- a/tests/kernel/test_metadata_store.py +++ b/tests/kernel/test_metadata_store.py @@ -605,6 +605,25 @@ def test_a_file_that_is_not_a_database_is_reported_as_corrupt(tmp_path: Path) -> store.close() +def test_a_database_that_cannot_be_opened_is_corrupt_rather_than_busy(tmp_path: Path) -> None: + """The other half of the `OperationalError` split. + + SQLite answers a directory where a database was wanted with `SQLITE_CANTOPEN` + — an `OperationalError`, same class as the lock, told apart only by its + result code. It is not contention and retrying will not help, so it belongs + with the damage rather than with `WorkspaceBusy`. + """ + db_path = tmp_path / "visionset.db" + db_path.mkdir() + + store = SqliteMetadataStore(db_path) + with pytest.raises(WorkspaceCorrupt) as caught: + _ = store.format_version + assert "sqlalchemy" not in type(caught.value).__module__ + assert "unable to open database file" in str(caught.value) + store.close() + + def test_deleting_a_parent_cascades_to_its_children(tmp_path: Path) -> None: store = _store(tmp_path) with store.unit_of_work() as uow: