app.storage

Content-addressed blob store — pluggable backends.

BlobStore is a Protocol; get_blob_store() returns the concrete implementation chosen by settings.blob_backend:

• fs (default): bytes live at <blob_root>/<sha[:2]>/<sha>. Writes are atomic via os.replace from a sibling temp file. Reads stream the file directly. This is the v0 backend and the test default.

• s3: bytes live at s3://<bucket>/<prefix><sha[:2]>/<sha>. Reads download lazily to the local S3 cache and return paths into that cache, so callers that need a filesystem path (pycolmap, Pillow) work transparently. Configured via settings.blob_s3_*.

Refcounting is the caller’s responsibility (typically a service) inside a transaction so blob lifecycle stays consistent with referencing rows — that contract is identical across backends.

class app.storage.blobs.BlobStore(*args, **kwargs)

Bases: Protocol

Sha256-keyed binary store. All methods are sha-addressed; the backend chooses where bytes physically live.

exists(sha)

Parameters:

sha (str)

Return type:

bool

put_stream(reader, *, chunk_size=Ellipsis)

Parameters:

• reader (BinaryIO)

• chunk_size (int)

Return type:

tuple[str, int]

put_bytes(data)

Parameters:

data (bytes)

Return type:

tuple[str, int]

open(sha)

Parameters:

sha (str)

Return type:

BinaryIO

aiter_chunks(sha, *, chunk_size=Ellipsis)

Parameters:

• sha (str)

• chunk_size (int)

Return type:

AsyncIterator[bytes]

delete(sha)

Parameters:

sha (str)

Return type:

None

local_path(sha)

Return a local filesystem path for the blob’s bytes.

For filesystem backends this is the canonical storage path. For remote backends (S3) the bytes are downloaded into the local cache on first access; subsequent calls return the cached path. Callers that need to hand a real path to a native library (pycolmap, Pillow, OpenCV) should use this.

Parameters:

sha (str)

Return type:

Path

class app.storage.blobs.FSBlobStore(settings=None)

Bases: object

Default backend — bytes on the local filesystem under <blob_root>/<sha[:2]>/<sha>.

Parameters:

settings (Settings | None)

backend: str = 'fs'

is_singleton: bool = False

path_for(sha)

Parameters:

sha (str)

Return type:

Path

local_path(sha)

Parameters:

sha (str)

Return type:

Path

exists(sha)

Parameters:

sha (str)

Return type:

bool

put_stream(reader, *, chunk_size=1048576)

Parameters:

• reader (BinaryIO)

• chunk_size (int)

Return type:

tuple[str, int]

put_bytes(data)

Parameters:

data (bytes)

Return type:

tuple[str, int]

open(sha)

Parameters:

sha (str)

Return type:

BinaryIO

async aiter_chunks(sha, *, chunk_size=1048576)

Parameters:

• sha (str)

• chunk_size (int)

Return type:

AsyncIterator[bytes]

delete(sha)

Parameters:

sha (str)

Return type:

None

class app.storage.blobs.S3BlobStore(settings=None)

Bases: object

Bytes live at s3://<bucket>/<prefix><sha[:2]>/<sha>.

Reads download lazily into <workspace>/_blob_cache/<sha[:2]>/<sha> so local_path() is a true filesystem path. The cache is content- addressed by sha so it never goes stale; deletes on the bucket are mirrored locally on next delete().

Parameters:

settings (Settings | None)

backend: str = 's3'

is_singleton: bool = False

bucket: str

prefix: str

cache_root: Path

exists(sha)

Parameters:

sha (str)

Return type:

bool

put_stream(reader, *, chunk_size=1048576)

Parameters:

• reader (BinaryIO)

• chunk_size (int)

Return type:

tuple[str, int]

put_bytes(data)

Parameters:

data (bytes)

Return type:

tuple[str, int]

local_path(sha)

Parameters:

sha (str)

Return type:

Path

open(sha)

Parameters:

sha (str)

Return type:

BinaryIO

async aiter_chunks(sha, *, chunk_size=1048576)

Parameters:

• sha (str)

• chunk_size (int)

Return type:

AsyncIterator[bytes]

delete(sha)

Parameters:

sha (str)

Return type:

None

class app.storage.blobs.InMemoryBlobStore(settings=None)

Bases: object

All bytes live in a process-local dict.

Used by ephemeral=true mode: zero disk persistence, ideal for tests, demos, and embedded use. Callers that need a real filesystem path (pycolmap, Pillow) trigger lazy materialization into a per-store temp dir via local_path(); that materialization survives until shutdown() is called.

Marked is_singleton = True so the factory caches a single instance per process — bytes live in self._bytes and would be unreachable to a writer-via-call-1 / reader-via-call-2 pattern if the factory built a fresh instance each time. The L15 regression-guard in the contract suite pins this invariant.

Parameters:

settings (Settings | None)

backend: str = 'memory'

is_singleton: bool = True

exists(sha)

Parameters:

sha (str)

Return type:

bool

put_stream(reader, *, chunk_size=1048576)

Parameters:

• reader (BinaryIO)

• chunk_size (int)

Return type:

tuple[str, int]

put_bytes(data)

Parameters:

data (bytes)

Return type:

tuple[str, int]

open(sha)

Parameters:

sha (str)

Return type:

BinaryIO

async aiter_chunks(sha, *, chunk_size=1048576)

Parameters:

• sha (str)

• chunk_size (int)

Return type:

AsyncIterator[bytes]

local_path(sha)

Parameters:

sha (str)

Return type:

Path

delete(sha)

Parameters:

sha (str)

Return type:

None

shutdown()

Drop all bytes and remove the scratch dir. Idempotent.

Return type:

None

app.storage.blobs.get_blob_store(settings=None)

Build a blob store per settings.blob_backend.

Backends declare their lifetime via is_singleton: False (FS / S3) constructs a fresh stateless wrapper per call; True (memory) caches one process-local instance because the bytes live in instance state — building a fresh store on every call would drop writes silently.

Callers that need to inject a custom backend in tests should pass a constructed instance through dependency injection rather than registering globally; reset_memory_blob_store_for_tests() drops cached singletons between test cases.

Parameters:

settings (Settings | None)

Return type:

BlobStore

app.storage.blobs.reset_memory_blob_store_for_tests()

Drop cached singleton instances. Call between test cases that mutate settings.blob_backend so each gets a fresh store.

Return type:

None

class app.storage.blobs.TempUploadStore(settings=None)

Bases: object

Working area for in-flight chunked uploads, separate from finalized blob storage. Each upload gets its own file under workspaces/_uploads/{upload_id}.

Parameters:

settings (Settings | None)

path_for(upload_id)

Parameters:

upload_id (str)

Return type:

Path

append(upload_id, offset, data)

Parameters:

• upload_id (str)

• offset (int)

• data (bytes)

Return type:

int

size(upload_id)

Parameters:

upload_id (str)

Return type:

int

hash_and_size(upload_id)

Parameters:

upload_id (str)

Return type:

tuple[str, int]

finalize_into(upload_id, blob_store)

Parameters:

• upload_id (str)

• blob_store (BlobStore)

Return type:

tuple[str, int]

discard(upload_id)

Parameters:

upload_id (str)

Return type:

None

app.storage.blobs.hash_iter(chunks)

Parameters:

chunks (Iterable[bytes])

Return type:

tuple[str, int]

app.storage.blobs.safe_rmtree(p)

Parameters:

p (Path)

Return type:

None

Sealed-snapshot writer.

Workers write live progress to sparse/; periodically they “seal” the current state into snapshots/{seq}/ via atomic dir rename so the API can read without racing the writer.

Sealing protocol:

1. tmp = snapshots/.tmp_{seq}

2. copy live sparse/ (and any sidecar JSONs) into tmp

3. write tmp/.complete last

4. os.replace(tmp, target) (atomic on the same FS)

5. update latest pointer (text file holding seq number)

API readers:

• list seq dirs (filter to those with .complete marker)

• read inside any sealed dir; data is immutable

class app.storage.snapshots.SnapshotStore(root)

Bases: object

Parameters:

root (Path)

seal(*, seq, source_dir, summary=None)

Parameters:

• seq (int)

• source_dir (Path)

• summary (dict | None)

Return type:

Path

list_sealed()

Return type:

list[int]

latest()

Return type:

int | None

path_for(seq)

Parameters:

seq (int)

Return type:

Path

gc(*, keep_last=3)

Parameters:

keep_last (int)

Return type:

list[int]

MappingInput checkpoint storage helpers.

pycolmap.MappingInput.save/load writes a binary container with magic PCMAPINx00 (v1) holding the pre-mapping state needed to resume an incremental run. We persist these per-job under jobs/{job_id}/checkpoints/{seq}.pcmapin so the worker can reload from the latest checkpoint on resume without re-running upstream stages.

The pycolmap binding is required to read/write — for tests we expose a trivial adapter that round-trips a payload through a regular file so the storage code path can be exercised independently.

class app.storage.mapping_input.CheckpointRef(seq: 'int', path: 'Path', summary: 'dict')

Bases: object

Parameters:

• seq (int)

• path (Path)

• summary (dict)

seq: int

path: Path

summary: dict

app.storage.mapping_input.checkpoint_root(job_dir)

Parameters:

job_dir (Path)

Return type:

Path

app.storage.mapping_input.write_checkpoint(job_dir, *, seq, payload, summary=None)

Parameters:

• job_dir (Path)

• seq (int)

• payload (bytes)

• summary (dict | None)

Return type:

CheckpointRef

app.storage.mapping_input.list_checkpoints(job_dir)

Parameters:

job_dir (Path)

Return type:

list[CheckpointRef]

app.storage.mapping_input.latest_checkpoint(job_dir)

Parameters:

job_dir (Path)

Return type:

CheckpointRef | None

app.storage.mapping_input.gc_checkpoints(job_dir, *, keep_last=5)

Parameters:

• job_dir (Path)

• keep_last (int)

Return type:

list[int]

Global S3 LRU cache.

Cache layout: <s3_cache_root>/<bucket>/<sha-of-key>__<etag> so each distinct (bucket, key, etag) tuple lives at a unique path. Stale ETags are evicted; LRU eviction by total bytes when over budget.

Cache is shared across projects/tenants — content addressed by ETag.

class app.storage.s3_cache.CachedObject(bucket: 'str', key: 'str', etag: 'str', path: 'Path', size: 'int')

Bases: object

Parameters:

• bucket (str)

• key (str)

• etag (str)

• path (Path)

• size (int)

bucket: str

key: str

etag: str

path: Path

size: int

class app.storage.s3_cache.S3Cache(settings=None)

Bases: object

Filesystem-backed LRU.

Atime-driven LRU on the entry’s manifest (.meta.json) so we don’t depend on filesystem atime, which is often disabled.

Parameters:

settings (Settings | None)

path_for(bucket, key, etag)

Parameters:

• bucket (str)

• key (str)

• etag (str)

Return type:

Path

lookup(bucket, key, etag)

Parameters:

• bucket (str)

• key (str)

• etag (str)

Return type:

CachedObject | None

insert(*, bucket, key, etag, src_bytes)

Parameters:

• bucket (str)

• key (str)

• etag (str)

• src_bytes (bytes)

Return type:

CachedObject

total_bytes()

Return type:

int

evict_to(*, max_bytes)

Evict least-recently-used entries until under budget. Returns bytes freed.

Parameters:

max_bytes (int)

Return type:

int

clear()

Return type:

None

Workspace GC + storage accounting.

GC policy (Phase 5). Old completed jobs (status in succeeded, failed, cancelled, cancelled_dirty) past ttl_days are swept in this order:

1. Drop the job’s dense/ directory first.

2. Drop the job’s snapshots/ directory next (job-scoped output).

3. Drop the job’s sparse/ directory last.

4. Drop the job’s log.jsonl / events.jsonl only if drop_db_rows is set.

Reconstruction-level artifacts are NOT touched by job GC because reconstructions can outlive the job that produced them and may be referenced by other (still active) jobs. A separate gc_orphan_* job sweeps reconstructions whose dataset_snapshot_hash no longer matches any current dataset.

Pinned jobs (Job.pinned=True) are skipped at every step.

Caller passes a now for testability; otherwise datetime.now(UTC).

app.storage.workspace.workspace_total_bytes(root)

Parameters:

root (Path)

Return type:

int

async app.storage.workspace.gc_completed_jobs(session, *, ttl_days, now=None, drop_db_rows=False)

Sweep completed-and-old jobs. Returns a summary dict for tests.

Parameters:

• session (AsyncSession)

• ttl_days (int)

• now (datetime | None)

• drop_db_rows (bool)

Return type:

dict

Model-artifact filesystem layout + lazy download/verify.

app.storage.models.models_root(settings=None)

Parameters:

settings (Settings | None)

Return type:

Path

app.storage.models.artifact_path(family, name, version, filename='weights.pth')

Parameters:

• family (str)

• name (str)

• version (str)

• filename (str)

Return type:

Path

app.storage.models.verify_sha(path, expected_sha)

Parameters:

• path (Path)

• expected_sha (str)

Return type:

bool

app.storage.models.install_artifact(*, family, name, version, src_bytes, expected_sha)

Parameters:

• family (str)

• name (str)

• version (str)

• src_bytes (bytes)

• expected_sha (str)

Return type:

Path