1/**
2 * The concrete object store: DO SQLite rows + R2 overflow (DESIGN.md §3.3).
3 *
4 * Implements the git layer's `ObjectSource` contract over the `objects`
5 * table, dispatching byte reads on the `location` column:
6 *
7 * - `'row'` — the `zdata` BLOB (zlib(content), no loose header),
8 * - `'r2'` — a streamed R2 `get` of `r2_key` (which may point into a fork
9 * parent's prefix — immutable, shared),
10 * - `'pack'` — v1.1 compaction; reads die with {@link NotImplementedError}
11 * until the compaction alarm lands (schema-ready, DESIGN.md §10).
12 *
13 * Plus the push-side surface the protocol layer needs: staged inserts with
14 * the row/R2 placement policy, batched metadata reads, and the
15 * staged-∪-live membership oracle behind the receive-pack connectivity
16 * check (DESIGN.md §3.6 step 5).
17 *
18 * All reads on the `ObjectSource` surface see **live objects only**
19 * (`staged_push IS NULL`); staged rows become visible to fetches only after
20 * the final `transactionSync` flips them live.
21 */
22import type { BlobBody, BlobStoreError, BlobStoreShape } from "../BlobStore.ts";
23import { RuntimeContext } from "../../RuntimeContext.ts";
24import * as Effect from "effect/Effect";
25import * as Schema from "effect/Schema";
26import * as Stream from "effect/Stream";
27import type { ObjectType, Oid } from "../Protocol/ObjectCodec.ts";
28import {
29 concatBytes,
30 encodeTypeSize,
31 type PackEntryType,
32} from "../Protocol/ObjectCodec.ts";
33import type { ManifestEntry } from "../Protocol/Store.ts";
34import {
35 StoreError,
36 type ObjectLocation,
37 type ObjectMeta,
38 type ObjectSource,
39} from "../Protocol/Store.ts";
40import * as Zlib from "../Protocol/Zlib.ts";
41import { objectKey, packKeyOf } from "./Keys.ts";
42import type { ObjectMetaRow, SqlClient } from "./Sql.ts";
43
44// ─────────────────────────────────────────────────────────────────────────────
45// Constants & errors
46// ─────────────────────────────────────────────────────────────────────────────
47
48/**
49 * Compressed-size threshold above which an object's bytes are offloaded to
50 * R2 (`location = 'r2'`) instead of a row BLOB — 1 MiB, a safety margin
51 * under the 2 MB SQLite row cap (DESIGN.md §3.3).
52 */
53export const R2_OFFLOAD_THRESHOLD = 1_048_576;
54
55/**
56 * Per-object uncompressed size cap for v1 — 64 MiB. Delta application needs
57 * base + result in memory; this keeps worst-case ingest memory bounded.
58 * Enforced by `PackParser` (rejected with `unpack object too large`);
59 * exported here as the storage layer's authoritative constant.
60 */
61export const MAX_OBJECT_SIZE = 67_108_864;
62
63/**
64 * Window size for coalesced reads out of compacted packs (DESIGN §15
65 * bottleneck 2). Touching one object in a window fetches the whole window,
66 * so a fetch that walks thousands of objects costs a handful of R2 GETs
67 * instead of thousands.
68 */
69export const WINDOW_BYTES = 4 * 1024 * 1024;
70
71/**
72 * Objects per batch on the emission/closure paths — one SQL statement and
73 * one emitted chunk per batch instead of per object (DESIGN §22).
74 */
75export const EMIT_BATCH = 256;
76
77/**
78 * Rows per multi-row `INSERT` while staging a push: 6 bindings per row, and
79 * Durable Object SQLite allows at most 100 bound parameters per statement
80 * (`MAX_IN_PARAMS`) — 16 rows is 96. Still a 16x cut in statement
81 * executions over one row per statement (see `insertStagedBatch`).
82 */
83export const STAGE_INSERT_ROWS = 16;
84
85const chunkArray = <T>(
86 items: ReadonlyArray<T>,
87 size: number,
88): Array<Array<T>> => {
89 const out: Array<Array<T>> = [];
90 for (let i = 0; i < items.length; i += size) {
91 out.push(items.slice(i, i + size) as Array<T>);
92 }
93 return out;
94};
95
96/**
97 * Isolate-wide budget for retained pack windows — 32 MiB (8 windows).
98 *
99 * The cache is a module-level LRU shared by every repo instance in the
100 * isolate, not a per-instance map: Durable Object instances of one class
101 * share an isolate and its 128 MiB, so per-instance caches multiply with
102 * the number of active repos. Measured (DESIGN §22.4): a 36 MiB push into a
103 * second repo was rejected with "isolate exceeded its memory limit" while
104 * the first repo's caches were resident. Keys carry the repo id
105 * (`packKey`), so sharing is safe.
106 */
107export const WINDOW_CACHE_BYTES = 32 * 1024 * 1024;
108
109/**
110 * Windows fetched ahead of the one being emitted (DESIGN §22.4). Blob
111 * storage GETs cost ~150–250 ms each regardless of size, and a 40 MiB
112 * pack is ten of them; fetched one at a time they were half of the
113 * dynamic transfer time. 3 ahead + 1 current stays within the cache.
114 */
115export const PACK_LOOKAHEAD = 3;
116
117const windows = new Map<string, { start: number; bytes: Uint8Array }>();
118let windowBytes = 0;
119const retainWindow = (
120 cacheKey: string,
121 slab: { start: number; bytes: Uint8Array },
122) => {
123 windows.set(cacheKey, slab);
124 windowBytes += slab.bytes.byteLength;
125 while (windowBytes > WINDOW_CACHE_BYTES && windows.size > 1) {
126 const oldest = windows.keys().next().value;
127 if (oldest === undefined) break;
128 windowBytes -= windows.get(oldest)!.bytes.byteLength;
129 windows.delete(oldest);
130 }
131};
132
133/**
134 * Raised (as a defect, via `Effect.die`) when a v1 code path reaches a
135 * feature the design explicitly defers — currently only reads of
136 * `location = 'pack'` objects, which exist starting with v1.1 compaction.
137 */
138export class NotImplementedError extends Schema.TaggedError<NotImplementedError>()(
139 "NotImplementedError",
140 { feature: Schema.String },
141) {}
142
143// ─────────────────────────────────────────────────────────────────────────────
144// Contract
145// ─────────────────────────────────────────────────────────────────────────────
146
147/**
148 * A pack entry resolved by `PackParser`, ready to stage: the object id, its
149 * numeric type, uncompressed size, and the zlib-compressed bytes to store
150 * (the pack's compressed span verbatim for non-delta entries; a fresh
151 * `deflate(content)` for delta-resolved ones).
152 */
153/**
154 * The live-object predicate (DESIGN §22.14). A push's rows are live the
155 * moment its `pushes` row is committed; the per-row `staged_push` column
156 * is nulled afterwards, off the response path (`flipPush`), and by GC as a
157 * safety net. Reads never wait for that rewrite.
158 */
159export const LIVE_OBJECTS =
160 "(staged_push IS NULL OR staged_push IN (SELECT push_id FROM pushes WHERE state = 'committed'))";
161
162export interface StagedObject {
163 readonly oid: Oid;
164 /**
165 * Set when the bytes already live in a pack in blob storage (a promoted
166 * wire pack, DESIGN §22.5): the row is staged as `location='pack'` with
167 * these coordinates and `zdata` is NOT stored.
168 */
169 readonly pack?:
170 | { readonly packId: string; readonly offset: number }
171 | undefined;
172 /**
173 * Compressed span when `zdata` is not carried (a promoted row whose
174 * bytes live in the pack): the insert must record the real size.
175 */
176 readonly zsize?: number | undefined;
177 readonly type: ObjectType;
178 /** Uncompressed content size in bytes. */
179 readonly size: number;
180 /** `zlib(content)` without the loose header. */
181 readonly zdata: Uint8Array;
182}
183
184/** Outcome of {@link ObjectStore.insertStaged}. */
185export type InsertOutcome = "inserted" | "exists";
186
187/**
188 * The full storage surface of a repo's objects: the read-only
189 * `ObjectSource` the protocol layer consumes, plus batched metadata reads,
190 * staged inserts, and the connectivity-check membership oracle.
191 */
192export interface ObjectStore extends ObjectSource {
193 /**
194 * Inflated content for many objects, one statement per ~256 oids — the
195 * closure walk reads whole tree levels this way instead of one SQL
196 * round trip per tree (DESIGN §22). Missing oids are simply absent.
197 */
198 readonly readContentBatch: (
199 oids: ReadonlyArray<Oid>,
200 ) => Effect.Effect<ReadonlyMap<Oid, Uint8Array>, StoreError>;
201 readonly packEntries: (
202 entries: ReadonlyArray<ManifestEntry>,
203 ) => Stream.Stream<Uint8Array, StoreError>;
204 /**
205 * Batched metadata for many oids (live objects only). Chunked `IN`
206 * queries, ≤ 100 params each. Missing oids are simply absent from the map.
207 */
208 readonly getMetaBatch: (
209 oids: ReadonlyArray<Oid>,
210 ) => Effect.Effect<ReadonlyMap<Oid, ObjectMeta>, StoreError>;
211 /**
212 * Stages one resolved object for `pushId` (DESIGN.md §3.6 step 4).
213 *
214 * Placement policy: compressed size > 1 MiB ⇒ the bytes are written to R2
215 * **first** (`{repoId}/objects/{oid}`, content-addressed so re-runs are
216 * idempotent), then the metadata row; otherwise the row carries the BLOB.
217 *
218 * Existing oids are skipped (idempotent re-push). A row left staged by a
219 * **different** (crashed) push is adopted — re-stamped with this `pushId`
220 * so the connectivity check and the final live-flip see it.
221 */
222 readonly insertStaged: (
223 pushId: string,
224 object: StagedObject,
225 ) => Effect.Effect<InsertOutcome, StoreError>;
226 /**
227 * Batched sibling of {@link ObjectStore.insertStaged} — the ingest hot
228 * path (DESIGN.md §16.6).
229 *
230 * Staging objects one at a time costs two statements each (an existence
231 * probe and an insert), so a 13.7k-object push ran ~27k statements and
232 * ~13.7k transactions inside a single-threaded Durable Object. This
233 * inserts a whole batch in **one** `transactionSync` with
234 * `INSERT OR IGNORE` (no probe), then adopts any rows left staged by a
235 * crashed push with one chunked `UPDATE`.
236 *
237 * Oversize objects (> 1 MiB compressed) still go through the per-object
238 * path, since each needs its own R2 write first.
239 */
240 readonly insertStagedBatch: (
241 pushId: string,
242 objects: ReadonlyArray<StagedObject>,
243 ) => Effect.Effect<void, StoreError>;
244 /** A live object's type and content for thin-delta resolution, or `undefined`. */
245 /** Read only live objects or objects staged by this push, with a caller-selected size cap. */
246 readonly readPrepared: (
247 pushId: string,
248 oid: Oid,
249 maxBytes: number,
250 ) => Effect.Effect<
251 { readonly type: ObjectType; readonly content: Uint8Array } | undefined,
252 StoreError
253 >;
254 readonly readBase: (
255 oid: Oid,
256 ) => Effect.Effect<
257 { readonly type: ObjectType; readonly content: Uint8Array } | undefined,
258 StoreError
259 >;
260 /**
261 * The connectivity-check membership oracle: returns the subset of `oids`
262 * that exist in **neither** the live store **nor** this push's staged
263 * rows. An empty result means every referenced object is present and the
264 * push is safe to commit.
265 */
266 readonly missingObjects: (
267 oids: ReadonlyArray<Oid>,
268 pushId: string,
269 ) => Effect.Effect<ReadonlyArray<Oid>, StoreError>;
270}
271
272/**
273 * Dependencies of {@link makeObjectStore}. Plain values (not Layers) so
274 * `RepoObject.ts` wires the store inside its inner per-instance effect.
275 */
276export interface ObjectStoreOptions {
277 /** The repo DO's SQL client (see `makeSqlClient`). */
278 readonly sql: SqlClient;
279 /** The swappable bulk-byte store (R2 by default — see `BlobStore.ts`). */
280 readonly blobs: BlobStoreShape;
281 /** The repo's ULID — the R2 key prefix (DESIGN.md §3.2). */
282 readonly repoId: string;
283}
284
285// ─────────────────────────────────────────────────────────────────────────────
286// Implementation
287// ─────────────────────────────────────────────────────────────────────────────
288
289const VALID_LOCATIONS: ReadonlySet<string> = new Set(["row", "r2", "pack"]);
290
291const isObjectType = (type: number): type is ObjectType =>
292 type === 1 || type === 2 || type === 3 || type === 4;
293
294const metaFromRow = (
295 row: ObjectMetaRow,
296): Effect.Effect<ObjectMeta, StoreError> => {
297 if (!isObjectType(row.type) || !VALID_LOCATIONS.has(row.location)) {
298 return Effect.fail(
299 new StoreError({
300 reason: `corrupt objects row for ${row.oid}: type=${row.type} location=${row.location}`,
301 }),
302 );
303 }
304 return Effect.succeed({
305 oid: row.oid,
306 type: row.type,
307 size: row.size,
308 zsize: row.zsize,
309 location: row.location as ObjectLocation,
310 });
311};
312
313/**
314 * Returns a plain `ArrayBuffer` view of the bytes for SQLite BLOB binding
315 * (workerd binds `ArrayBuffer`, not typed arrays). Zero-copy when the view
316 * spans its whole buffer.
317 */
318const toArrayBuffer = (bytes: Uint8Array): ArrayBuffer =>
319 bytes.byteOffset === 0 &&
320 bytes.byteLength === bytes.buffer.byteLength &&
321 bytes.buffer instanceof ArrayBuffer
322 ? bytes.buffer
323 : (bytes.buffer.slice(
324 bytes.byteOffset,
325 bytes.byteOffset + bytes.byteLength,
326 ) as ArrayBuffer);
327
328/** Row projection used by the byte-read dispatch. */
329interface ZDataRow extends Record<
330 string,
331 string | number | ArrayBuffer | null
332> {
333 readonly oid: string;
334 readonly location: string;
335 readonly zdata: ArrayBuffer | null;
336 readonly r2_key: string | null;
337 /** sha1 of the containing pack when `location='pack'` (DESIGN §12.1). */
338 readonly pack_id: string | null;
339 /** Byte offset of this object's zdata span inside that pack. */
340 readonly pack_offset: number | null;
341 /** Length of the zdata span (mirrors `zsize`). */
342 readonly zsize: number;
343}
344
345/**
346 * Creates the {@link ObjectStore} for one repo. Plain factory — no Layers —
347 * so `RepoObject.ts` can wire it in its inner effect with the ambient
348 * `state`/bucket handles.
349 */
350export const makeObjectStore = (options: ObjectStoreOptions): ObjectStore => {
351 const { sql, blobs, repoId } = options;
352
353 /** Discharges the blob store's `RuntimeContext` coloring and closes the
354 * error union into `StoreError` (we run inside the Repo DO). */
355 const runBlob = <A>(
356 effect: Effect.Effect<A, BlobStoreError, RuntimeContext>,
357 what: string,
358 ): Effect.Effect<A, StoreError> =>
359 effect.pipe(
360 Effect.mapError(
361 (error) => new StoreError({ reason: `${what}: ${error.reason}` }),
362 ),
363 Effect.provide(RuntimeContext.phantom),
364 );
365
366 const zlibToStore = (error: Zlib.ZlibError): StoreError =>
367 new StoreError({ reason: error.reason });
368
369 const readRow = (
370 oid: Oid,
371 pushId: string | null = null,
372 ): Effect.Effect<ZDataRow | undefined, StoreError> =>
373 sql.first<ZDataRow>(
374 `SELECT oid, location, zdata, r2_key, pack_id, pack_offset, zsize
375 FROM objects WHERE oid = ? AND (${LIVE_OBJECTS} OR staged_push = ?)`,
376 oid,
377 pushId,
378 );
379
380 const requireRow = Effect.fn(function* (
381 oid: Oid,
382 pushId: string | null = null,
383 ) {
384 const row = yield* readRow(oid, pushId);
385 if (row === undefined) {
386 return yield* Effect.fail(
387 new StoreError({ reason: `object not found: ${oid}` }),
388 );
389 }
390 return row;
391 });
392
393 /** Fetches the R2 body for an `'r2'`-located row. */
394 const r2Body = Effect.fn(function* (oid: Oid, r2Key: string | null) {
395 if (r2Key === null) {
396 return yield* Effect.fail(
397 new StoreError({
398 reason: `object ${oid} has location='r2' but no r2_key`,
399 }),
400 );
401 }
402 const body = yield* runBlob(blobs.get(r2Key), `blob get ${r2Key}`);
403 if (body === null) {
404 return yield* Effect.fail(
405 new StoreError({ reason: `R2 object missing: ${r2Key}` }),
406 );
407 }
408 return body;
409 });
410
411 /**
412 * Coalesced window cache over compacted packs (DESIGN §15 bottleneck 2).
413 *
414 * A pack read is one ranged R2 GET per object if done naively — and a
415 * fetch that walks a few thousand objects then costs a few thousand
416 * serialized round trips. Measured: **68 s to serve a 53 KiB fetch** over
417 * a 1202-object pack. Instead, reads are served out of window-aligned
418 * slabs: touching any object in a window fetches the whole window once,
419 * and every other object inside it is then free. A small pack collapses
420 * to a single GET.
421 *
422 * Bounded by construction: at most {@link MAX_CACHED_WINDOWS} windows of
423 * {@link WINDOW_BYTES} are retained (insertion-ordered LRU), so the cache
424 * cannot outgrow the isolate's memory budget.
425 */
426
427 const readWindow = Effect.fn(function* (key: string, windowIndex: number) {
428 const cacheKey = `${key}:${windowIndex}`;
429 const hit = windows.get(cacheKey);
430 if (hit !== undefined) {
431 // Refresh recency (Map iterates in insertion order).
432 windows.delete(cacheKey);
433 windows.set(cacheKey, hit);
434 return hit;
435 }
436 const start = windowIndex * WINDOW_BYTES;
437 const body = yield* runBlob(
438 blobs.get(key, { offset: start, length: WINDOW_BYTES }),
439 `blob window get ${key}`,
440 );
441 if (body === null) {
442 return yield* Effect.fail(
443 new StoreError({ reason: `pack missing: ${key}` }),
444 );
445 }
446 const bytes = yield* body.bytes.pipe(
447 Effect.mapError(
448 (error) =>
449 new StoreError({ reason: `R2 read ${key}: ${error.message}` }),
450 ),
451 );
452 const slab = { start, bytes };
453 retainWindow(cacheKey, slab);
454 return slab;
455 });
456
457 /**
458 * Reads an object's zdata span out of a compacted R2 pack (DESIGN §12.1),
459 * via the window cache above.
460 *
461 * `pack_offset` addresses the object's **zdata** directly (not the pack
462 * entry header), so the slice is exactly the stored compressed bytes —
463 * the same bytes a `location='row'` object holds in its BLOB. Entry
464 * headers are regenerated from `(type, size)` at emission time.
465 */
466 const packBytes = Effect.fn(function* (oid: Oid, row: ZDataRow) {
467 if (row.pack_id === null || row.pack_offset === null) {
468 return yield* Effect.fail(
469 new StoreError({
470 reason: `object ${oid} has location='pack' but no pack coordinates`,
471 }),
472 );
473 }
474 const key = packKeyOf(repoId, row.pack_id);
475 const offset = row.pack_offset;
476 const windowIndex = Math.floor(offset / WINDOW_BYTES);
477 const slab = yield* readWindow(key, windowIndex);
478 const relative = offset - slab.start;
479 if (relative >= 0 && relative + row.zsize <= slab.bytes.length) {
480 // Copy out: the slab is retained by the cache and must not be aliased.
481 return slab.bytes.slice(relative, relative + row.zsize);
482 }
483 // The object straddles the window boundary (or the window was short) —
484 // fall back to an exact ranged read for this one object.
485 const body = yield* runBlob(
486 blobs.get(key, { offset, length: row.zsize }),
487 `blob ranged get ${key}`,
488 );
489 if (body === null) {
490 return yield* Effect.fail(
491 new StoreError({ reason: `pack missing: ${key}` }),
492 );
493 }
494 return yield* body.bytes.pipe(
495 Effect.mapError(
496 (error) =>
497 new StoreError({ reason: `R2 read ${key}: ${error.message}` }),
498 ),
499 );
500 });
501
502 /** Reads the stored compressed bytes fully into memory. */
503 const readZBytes = Effect.fn(function* (
504 oid: Oid,
505 pushId: string | null = null,
506 ) {
507 const row = yield* requireRow(oid, pushId);
508 switch (row.location) {
509 case "row": {
510 if (row.zdata === null) {
511 return yield* Effect.fail(
512 new StoreError({
513 reason: `object ${oid} has location='row' but NULL zdata`,
514 }),
515 );
516 }
517 return new Uint8Array(row.zdata);
518 }
519 case "r2": {
520 const body = yield* r2Body(oid, row.r2_key);
521 return yield* body.bytes.pipe(
522 Effect.mapError(
523 (error) =>
524 new StoreError({
525 reason: `R2 read ${row.r2_key}: ${error.message}`,
526 }),
527 ),
528 );
529 }
530 case "pack":
531 return yield* packBytes(oid, row);
532 default:
533 return yield* Effect.fail(
534 new StoreError({
535 reason: `object ${oid} has unknown location '${row.location}'`,
536 }),
537 );
538 }
539 });
540
541 const getMetaBatch = (
542 oids: ReadonlyArray<Oid>,
543 ): Effect.Effect<ReadonlyMap<Oid, ObjectMeta>, StoreError> =>
544 Effect.gen(function* () {
545 const unique = Array.from(new Set(oids));
546 const map = new Map<Oid, ObjectMeta>();
547 if (unique.length === 0) {
548 return map;
549 }
550 const rows = yield* sql.inChunks<ObjectMetaRow>(
551 (ph) =>
552 `SELECT oid, type, size, zsize, location FROM objects WHERE oid IN (${ph}) AND ${LIVE_OBJECTS}`,
553 unique,
554 );
555 for (const row of rows) {
556 map.set(row.oid, yield* metaFromRow(row));
557 }
558 return map;
559 });
560
561 const insertStagedOne = Effect.fn(function* (
562 pushId: string,
563 object: StagedObject,
564 ) {
565 const existing = yield* sql.first<{ staged_push: string | null }>(
566 `SELECT staged_push FROM objects WHERE oid = ?`,
567 object.oid,
568 );
569 if (existing !== undefined) {
570 if (existing.staged_push !== null && existing.staged_push !== pushId) {
571 // Adopt a row left staged by a crashed push so this push's
572 // connectivity check and live-flip see it.
573 yield* sql.run(
574 `UPDATE objects SET staged_push = ? WHERE oid = ? AND staged_push IS NOT NULL`,
575 pushId,
576 object.oid,
577 );
578 }
579 return "exists" as const;
580 }
581 if (object.zdata.byteLength > R2_OFFLOAD_THRESHOLD) {
582 // R2 write FIRST, then the metadata row — a crash between the two
583 // leaves only a harmless content-addressed R2 orphan.
584 const key = objectKey(repoId, object.oid);
585 yield* runBlob(blobs.put(key, object.zdata), `blob put ${key}`);
586 yield* sql.run(
587 `INSERT OR IGNORE INTO objects (oid, type, size, zsize, location, zdata, r2_key, staged_push)
588 VALUES (?, ?, ?, ?, 'r2', NULL, ?, ?)`,
589 object.oid,
590 object.type,
591 object.size,
592 object.zdata.byteLength,
593 key,
594 pushId,
595 );
596 } else {
597 yield* sql.run(
598 `INSERT OR IGNORE INTO objects (oid, type, size, zsize, location, zdata, staged_push)
599 VALUES (?, ?, ?, ?, 'row', ?, ?)`,
600 object.oid,
601 object.type,
602 object.size,
603 object.zdata.byteLength,
604 toArrayBuffer(object.zdata),
605 pushId,
606 );
607 }
608 return "inserted" as const;
609 });
610
611 const readZDataOf = (oid: Oid): Stream.Stream<Uint8Array, StoreError> =>
612 Stream.unwrap(
613 Effect.gen(function* () {
614 const row = yield* requireRow(oid);
615 switch (row.location) {
616 case "row": {
617 if (row.zdata === null) {
618 return yield* Effect.fail(
619 new StoreError({
620 reason: `object ${oid} has location='row' but NULL zdata`,
621 }),
622 );
623 }
624 return Stream.succeed(new Uint8Array(row.zdata));
625 }
626 case "r2": {
627 const body = yield* r2Body(oid, row.r2_key);
628 return body.stream.pipe(
629 Stream.mapError(
630 (error) =>
631 new StoreError({
632 reason: `R2 stream ${row.r2_key}: ${error.message}`,
633 }),
634 ),
635 );
636 }
637 case "pack":
638 return Stream.succeed(yield* packBytes(oid, row));
639 default:
640 return yield* Effect.fail(
641 new StoreError({
642 reason: `object ${oid} has unknown location '${row.location}'`,
643 }),
644 );
645 }
646 }),
647 );
648
649 const self: ObjectStore = {
650 // ── ObjectSource (live objects only) ────────────────────────────────────
651 has: (oid) =>
652 sql
653 .first<{ one: number }>(
654 `SELECT 1 AS one FROM objects WHERE oid = ? AND ${LIVE_OBJECTS}`,
655 oid,
656 )
657 .pipe(Effect.map((row) => row !== undefined)),
658
659 filterExisting: (oids) =>
660 Effect.gen(function* () {
661 const unique = Array.from(new Set(oids));
662 if (unique.length === 0) {
663 return [];
664 }
665 const rows = yield* sql.inChunks<{ oid: string }>(
666 (ph) =>
667 `SELECT oid FROM objects WHERE oid IN (${ph}) AND ${LIVE_OBJECTS}`,
668 unique,
669 );
670 return rows.map((row) => row.oid);
671 }),
672
673 getMeta: (oid) =>
674 Effect.gen(function* () {
675 const row = yield* sql.first<ObjectMetaRow>(
676 `SELECT oid, type, size, zsize, location FROM objects WHERE oid = ? AND ${LIVE_OBJECTS}`,
677 oid,
678 );
679 if (row === undefined) {
680 return undefined;
681 }
682 return yield* metaFromRow(row);
683 }),
684
685 readZData: readZDataOf,
686
687 readContent: (oid) =>
688 readZBytes(oid).pipe(
689 Effect.flatMap((zdata) =>
690 Zlib.inflate(zdata).pipe(Effect.mapError(zlibToStore)),
691 ),
692 ),
693
694 readContentBatch: (oids) =>
695 Effect.gen(function* () {
696 const out = new Map<Oid, Uint8Array>();
697 const unique = Array.from(new Set(oids));
698 if (unique.length === 0) return out;
699 const rows = yield* sql.inChunks<ZDataRow>(
700 (ph) =>
701 `SELECT oid, location, zdata, r2_key, pack_id, pack_offset, zsize
702 FROM objects WHERE oid IN (${ph}) AND ${LIVE_OBJECTS}`,
703 unique,
704 );
705 const inflate = (z: Uint8Array) =>
706 Zlib.inflate(z).pipe(Effect.mapError(zlibToStore));
707 // Pack-resident rows in pack order so consecutive window reads hit.
708 const packRows = rows
709 .filter((row) => row.location === "pack")
710 .sort(
711 (a, b) =>
712 (a.pack_id ?? "").localeCompare(b.pack_id ?? "") ||
713 (a.pack_offset ?? 0) - (b.pack_offset ?? 0),
714 );
715 for (const row of rows) {
716 if (row.location === "row" && row.zdata !== null) {
717 out.set(row.oid, yield* inflate(new Uint8Array(row.zdata)));
718 } else if (row.location === "r2") {
719 out.set(
720 row.oid,
721 yield* readZBytes(row.oid).pipe(Effect.flatMap(inflate)),
722 );
723 }
724 }
725 for (const row of packRows) {
726 out.set(row.oid, yield* inflate(yield* packBytes(row.oid, row)));
727 }
728 return out;
729 }),
730
731 packEntries: (entries) =>
732 Stream.suspend(() => {
733 // Locality order (DESIGN §22): a non-delta pack is order-free, so
734 // emit SQLite rows in manifest order, then every pack-resident
735 // object sorted by (pack, offset) so the window cache streams each
736 // 4 MiB window exactly once, then the rare R2-offloaded objects.
737 const byOid = new Map(entries.map((entry) => [entry.oid, entry]));
738 const rowEntries: Array<ManifestEntry> = [];
739 const packList: Array<ManifestEntry> = [];
740 const r2Entries: Array<ManifestEntry> = [];
741 for (const entry of entries) {
742 (entry.location === "row"
743 ? rowEntries
744 : entry.location === "pack"
745 ? packList
746 : r2Entries
747 ).push(entry);
748 }
749 const header = (entry: ManifestEntry) =>
750 encodeTypeSize(entry.type as PackEntryType, entry.size);
751
752 const rowStream = Stream.fromIterable(
753 chunkArray(rowEntries, EMIT_BATCH),
754 ).pipe(
755 Stream.mapEffect((run) =>
756 Effect.gen(function* () {
757 const rows = yield* sql.inChunks<{
758 oid: string;
759 zdata: ArrayBuffer | null;
760 }>(
761 (ph) =>
762 `SELECT oid, zdata FROM objects WHERE oid IN (${ph}) AND ${LIVE_OBJECTS}`,
763 run.map((entry) => entry.oid),
764 );
765 const zdataOf = new Map(rows.map((row) => [row.oid, row.zdata]));
766 const pieces: Array<Uint8Array> = [];
767 for (const entry of run) {
768 const zdata = zdataOf.get(entry.oid);
769 if (zdata === undefined || zdata === null) {
770 // The manifest said 'row', but compaction may have moved
771 // the bytes to a pack between closure and emission (the
772 // alarm runs concurrently with fetches). Re-dispatch on
773 // the CURRENT location; only a truly absent object fails.
774 pieces.push(header(entry), yield* readZBytes(entry.oid));
775 continue;
776 }
777 pieces.push(header(entry), new Uint8Array(zdata));
778 }
779 return concatBytes(pieces);
780 }),
781 ),
782 );
783
784 // Pack runs (DESIGN §22.4): coords sorted by (pack, offset) and
785 // grouped by 4 MiB window. Each group is ONE window fetch (with the
786 // next window fetched concurrently — a cache hit on the following
787 // step) and one synchronous slice loop: no per-object Effect call,
788 // no per-object cache lookup. Objects that straddle a window edge
789 // (rare) fall back to the ranged read in `packBytes`; pack order
790 // is free, so they simply follow their window's run.
791 const packStream = Stream.unwrap(
792 Effect.gen(function* () {
793 if (packList.length === 0) return Stream.empty;
794 const coords = yield* sql.inChunks<{
795 oid: string;
796 pack_id: string | null;
797 pack_offset: number | null;
798 zsize: number;
799 }>(
800 (ph) =>
801 `SELECT oid, pack_id, pack_offset, zsize FROM objects WHERE oid IN (${ph}) AND ${LIVE_OBJECTS}`,
802 packList.map((entry) => entry.oid),
803 );
804 coords.sort(
805 (a, b) =>
806 (a.pack_id ?? "").localeCompare(b.pack_id ?? "") ||
807 (a.pack_offset ?? 0) - (b.pack_offset ?? 0),
808 );
809 type Coord = (typeof coords)[number];
810 type Run = {
811 readonly key: string;
812 readonly window: number;
813 readonly inside: Array<Coord>;
814 readonly straddling: Array<Coord>;
815 };
816 const runs: Array<Run> = [];
817 for (const coord of coords) {
818 if (coord.pack_id === null || coord.pack_offset === null) {
819 return yield* Effect.fail(
820 new StoreError({
821 reason: `object ${coord.oid} is location='pack' without coordinates`,
822 }),
823 );
824 }
825 const key = packKeyOf(repoId, coord.pack_id);
826 const window = Math.floor(coord.pack_offset / WINDOW_BYTES);
827 let run = runs[runs.length - 1];
828 if (
829 run === undefined ||
830 run.key !== key ||
831 run.window !== window
832 ) {
833 run = { key, window, inside: [], straddling: [] };
834 runs.push(run);
835 }
836 (coord.pack_offset + coord.zsize <= (window + 1) * WINDOW_BYTES
837 ? run.inside
838 : run.straddling
839 ).push(coord);
840 }
841 const rowOf = (coord: Coord): ZDataRow => ({
842 oid: coord.oid,
843 location: "pack",
844 zdata: null,
845 r2_key: null,
846 pack_id: coord.pack_id,
847 pack_offset: coord.pack_offset,
848 zsize: coord.zsize,
849 });
850 // Lookahead: the next PACK_LOOKAHEAD windows are fetched
851 // concurrently with this one; each is a cache hit when its
852 // turn comes. Bounded by WINDOW_CACHE_BYTES (8 windows).
853 return Stream.fromIterable(
854 runs.map(
855 (run, i) =>
856 [run, runs.slice(i + 1, i + 1 + PACK_LOOKAHEAD)] as const,
857 ),
858 ).pipe(
859 Stream.mapEffect(([run, ahead]) =>
860 Effect.gen(function* () {
861 const [slab] = yield* Effect.all(
862 [
863 readWindow(run.key, run.window),
864 ...ahead.map((next) =>
865 readWindow(next.key, next.window).pipe(Effect.ignore),
866 ),
867 ],
868 { concurrency: 1 + PACK_LOOKAHEAD },
869 );
870 const pieces = yield* Effect.sync(() => {
871 const out: Array<Uint8Array> = [];
872 for (const coord of run.inside) {
873 const entry = byOid.get(coord.oid);
874 if (entry === undefined) continue;
875 const at = coord.pack_offset! - slab.start;
876 out.push(
877 header(entry),
878 slab.bytes.subarray(at, at + coord.zsize),
879 );
880 }
881 return out;
882 });
883 for (const coord of run.straddling) {
884 const entry = byOid.get(coord.oid);
885 if (entry === undefined) continue;
886 pieces.push(
887 header(entry),
888 yield* packBytes(coord.oid, rowOf(coord)),
889 );
890 }
891 return concatBytes(pieces);
892 }),
893 ),
894 );
895 }),
896 );
897
898 const r2Stream = Stream.fromIterable(r2Entries).pipe(
899 Stream.flatMap((entry) =>
900 Stream.succeed(header(entry)).pipe(
901 Stream.concat(readZDataOf(entry.oid)),
902 ),
903 ),
904 );
905
906 return rowStream.pipe(
907 Stream.concat(packStream),
908 Stream.concat(r2Stream),
909 );
910 }),
911
912 // ── Store-side surface ──────────────────────────────────────────────────
913 getMetaBatch,
914
915 insertStagedBatch: Effect.fn(function* (
916 pushId: string,
917 objects: ReadonlyArray<StagedObject>,
918 ) {
919 if (objects.length === 0) return;
920 const inline: Array<StagedObject> = [];
921 const promoted: Array<StagedObject> = [];
922 for (const object of objects) {
923 if (object.pack !== undefined) {
924 promoted.push(object);
925 } else if (object.zdata.byteLength > R2_OFFLOAD_THRESHOLD) {
926 yield* insertStagedOne(pushId, object);
927 } else {
928 inline.push(object);
929 }
930 }
931 if (promoted.length > 0) {
932 // Rows that point into a pack already in blob storage: no BLOB
933 // written to SQLite at all. 7 bindings per row → 14 rows per
934 // statement under the 100-parameter cap.
935 const sorted = [...promoted].sort((a, b) =>
936 a.oid < b.oid ? -1 : a.oid > b.oid ? 1 : 0,
937 );
938 const written = yield* sql.transactionSync((raw) => {
939 let changed = 0;
940 for (let at = 0; at < sorted.length; at += 14) {
941 const part = sorted.slice(at, at + 14);
942 const values = part
943 .map(() => "(?, ?, ?, ?, 'pack', NULL, ?, ?, ?)")
944 .join(", ");
945 const bindings: Array<string | number> = [];
946 for (const object of part) {
947 bindings.push(
948 object.oid,
949 object.type,
950 object.size,
951 object.zsize ?? object.zdata.byteLength,
952 object.pack!.packId,
953 object.pack!.offset,
954 pushId,
955 );
956 }
957 changed += raw.exec(
958 `INSERT OR IGNORE INTO objects (oid, type, size, zsize, location, zdata, pack_id, pack_offset, staged_push) VALUES ${values}`,
959 ...bindings,
960 ).rowsWritten;
961 }
962 return changed;
963 });
964 if (written < sorted.length) {
965 yield* sql.inChunks(
966 (ph) =>
967 `UPDATE objects SET staged_push = ? WHERE oid IN (${ph})
968 AND staged_push IS NOT NULL AND staged_push != ?
969 AND staged_push NOT IN (SELECT push_id FROM pushes WHERE state = 'committed')`,
970 sorted.map((object) => object.oid),
971 { prefix: [pushId], suffix: [pushId] },
972 );
973 }
974 }
975 if (inline.length === 0) return;
976
977 // One transaction for the whole batch: `INSERT OR IGNORE` needs no
978 // existence probe, and a single commit replaces one per object.
979 // Sorted by oid: the table is WITHOUT ROWID keyed by oid, so inserting
980 // in key order turns a batch into sequential B-tree appends instead of
981 // random page splits.
982 const rows = inline
983 .map((object) => ({
984 oid: object.oid,
985 type: object.type,
986 size: object.size,
987 zsize: object.zdata.byteLength,
988 zdata: toArrayBuffer(object.zdata),
989 }))
990 .sort((a, b) => (a.oid < b.oid ? -1 : a.oid > b.oid ? 1 : 0));
991 // Multi-row INSERTs (DESIGN §22.4): one statement per STAGE_INSERT_ROWS
992 // rows instead of one per row. On production CPU each statement
993 // execution costs far more than binding six more parameters, and
994 // staging was the largest slice of a push's ingest (7 s of 30 s).
995 const written = yield* sql.transactionSync((raw) => {
996 let changed = 0;
997 for (let at = 0; at < rows.length; at += STAGE_INSERT_ROWS) {
998 const part = rows.slice(at, at + STAGE_INSERT_ROWS);
999 const values = part.map(() => "(?, ?, ?, ?, 'row', ?, ?)").join(", ");
1000 const bindings: Array<string | number | ArrayBuffer> = [];
1001 for (const row of part) {
1002 bindings.push(
1003 row.oid,
1004 row.type,
1005 row.size,
1006 row.zsize,
1007 row.zdata,
1008 pushId,
1009 );
1010 }
1011 const cursor = raw.exec(
1012 `INSERT OR IGNORE INTO objects (oid, type, size, zsize, location, zdata, staged_push) VALUES ${values}`,
1013 ...bindings,
1014 );
1015 changed += cursor.rowsWritten;
1016 }
1017 return changed;
1018 });
1019 // Every row was new: nothing to adopt. Only when INSERT OR IGNORE
1020 // skipped rows can some belong to a crashed push and need re-stamping.
1021 if (written >= rows.length) return;
1022 yield* sql.inChunks(
1023 (ph) =>
1024 `UPDATE objects SET staged_push = ? WHERE oid IN (${ph})
1025 AND staged_push IS NOT NULL AND staged_push != ?
1026 AND staged_push NOT IN (SELECT push_id FROM pushes WHERE state = 'committed')`,
1027 inline.map((object) => object.oid),
1028 { prefix: [pushId], suffix: [pushId] },
1029 );
1030 }),
1031 readPrepared: (pushId, oid, maxBytes) =>
1032 Effect.gen(function* () {
1033 if (!Number.isSafeInteger(maxBytes) || maxBytes < 0)
1034 return yield* new StoreError({ reason: "invalid object size limit" });
1035 const row = yield* sql.first<ObjectMetaRow>(
1036 `SELECT oid, type, size, zsize, location FROM objects WHERE oid = ? AND (${LIVE_OBJECTS} OR staged_push = ?)`,
1037 oid,
1038 pushId,
1039 );
1040 if (row === undefined) return undefined;
1041 const meta = yield* metaFromRow(row);
1042 if (meta.size > maxBytes)
1043 return yield* new StoreError({
1044 reason: `object exceeds ${maxBytes} byte read limit`,
1045 });
1046 const zdata = yield* readZBytes(oid, pushId);
1047 const content = yield* Zlib.inflate(zdata).pipe(
1048 Effect.mapError(zlibToStore),
1049 );
1050 return { type: meta.type, content };
1051 }),
1052
1053 readBase: (oid) =>
1054 self
1055 .getMeta(oid)
1056 .pipe(
1057 Effect.flatMap((meta) =>
1058 meta === undefined
1059 ? Effect.succeed(undefined)
1060 : self
1061 .readContent(oid)
1062 .pipe(
1063 Effect.map((content) => ({ type: meta.type, content })),
1064 ),
1065 ),
1066 ),
1067 insertStaged: insertStagedOne,
1068
1069 missingObjects: (oids, pushId) =>
1070 Effect.gen(function* () {
1071 const unique = Array.from(new Set(oids));
1072 if (unique.length === 0) {
1073 return [];
1074 }
1075 const rows = yield* sql.inChunks<{ oid: string }>(
1076 (ph) =>
1077 `SELECT oid FROM objects WHERE oid IN (${ph}) AND (${LIVE_OBJECTS} OR staged_push = ?)`,
1078 unique,
1079 { suffix: [pushId] },
1080 );
1081 const present = new Set(rows.map((row) => row.oid));
1082 return unique.filter((oid) => !present.has(oid));
1083 }),
1084 };
1085 return self;
1086};