mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
feat(chord): stream sync bodies and bound their memory
Batch and snapshot bodies are byte streams end to end, so no side holds a whole batch, snapshot or store: uploads end a batch at BATCH_BYTES, pulls apply in chunks of APPLY_BYTES, snapshots stream from export() into importStore(). SyncFileSystem gains openBinaryReader and appendFile so FileSystemSyncRemote streams bodies to and from disk. test/bench/sync-memory.ts measures the heap in workerd Durable Objects under a 128 MB cap: every sync step peaks under 40 MB, up to 256 MB of history.
This commit is contained in:
Cristina Poncela Cubeiro
parent
1a93bad3ed
commit
2a70f6e820
12 changed files with 1168 additions and 149 deletions
|
|
@ -1,53 +1,139 @@
|
|||
// Bodies on the remote, encoded and decoded as streams, so neither side holds a whole batch or snapshot in memory:
|
||||
// gzip-compressed JSON lines, one outbox commit (a batch) or one archive part (a snapshot) per line. gzip, because
|
||||
// `CompressionStream` offers it in Node, Workers, and browsers alike, and a browser viewer reads batches too.
|
||||
|
||||
import type { ArchivePart } from "../store/archive.ts";
|
||||
import { decodeJsonl, encodeJsonl } from "../store/jsonl.ts";
|
||||
import type { OutboxCommit } from "../store/outbox.ts";
|
||||
import { decodeLogLines, encodeLogLine, LogFormatError } from "./lines.ts";
|
||||
import { decodeLogLine, encodeLogLine, LogFormatError } from "./lines.ts";
|
||||
|
||||
/** A body on the remote: its bytes in chunks, read once. */
|
||||
export type SyncBody = AsyncIterable<Uint8Array>;
|
||||
|
||||
/** The bytes a chunk of encoded lines collects before it is compressed. */
|
||||
const CHUNK = 64 * 1024;
|
||||
|
||||
/** A batch body: `commits` as gzip-compressed JSON lines, read from `commits` only as the body is read. */
|
||||
export function encodeBatch(commits: Iterable<OutboxCommit> | AsyncIterable<OutboxCommit>): SyncBody {
|
||||
return encodeLines(
|
||||
(async function* () {
|
||||
for await (const commit of commits) yield encodeLogLine(commit);
|
||||
})(),
|
||||
);
|
||||
}
|
||||
|
||||
/** The commits of a batch body, in order, as they arrive. */
|
||||
export async function* decodeBatch(body: SyncBody): AsyncGenerator<OutboxCommit> {
|
||||
for await (const { commit } of decodeBatchLines(body)) yield commit;
|
||||
}
|
||||
|
||||
/** The commits of a batch body with the size of each line, for callers that bound what they hold. */
|
||||
export async function* decodeBatchLines(
|
||||
body: SyncBody,
|
||||
): AsyncGenerator<{ readonly commit: OutboxCommit; readonly bytes: number }> {
|
||||
let previous = 0;
|
||||
for await (const { text, number } of lines(gunzip(body))) {
|
||||
const commit = decodeLogLine(text, number);
|
||||
if (commit.seq <= previous) throw new LogFormatError(`Log line ${number} does not increase the sequence`);
|
||||
previous = commit.seq;
|
||||
yield { commit, bytes: text.length + 1 };
|
||||
}
|
||||
}
|
||||
|
||||
/** A snapshot body: a store archive (`store.export()`), one part per JSON line, gzip-compressed. */
|
||||
export function encodeSnapshot(parts: AsyncIterable<ArchivePart>): SyncBody {
|
||||
return gzip(encodeJsonl(parts));
|
||||
}
|
||||
|
||||
/** The parts of a snapshot body as they arrive, unchecked: `importStore()` checks them (a trust boundary). */
|
||||
export function decodeSnapshot(body: SyncBody): AsyncIterable<unknown> {
|
||||
return decodeJsonl(gunzip(body));
|
||||
}
|
||||
|
||||
/** Text lines, already newline-terminated, gzip-compressed in chunks of whole lines. */
|
||||
function encodeLines(lines: AsyncIterable<string>): SyncBody {
|
||||
return gzip(
|
||||
(async function* () {
|
||||
const encoder = new TextEncoder();
|
||||
let pending = "";
|
||||
for await (const line of lines) {
|
||||
pending += line;
|
||||
if (pending.length >= CHUNK) {
|
||||
yield encoder.encode(pending);
|
||||
pending = "";
|
||||
}
|
||||
}
|
||||
if (pending.length > 0) yield encoder.encode(pending);
|
||||
})(),
|
||||
);
|
||||
}
|
||||
|
||||
/** The lines of decompressed `chunks` without their newlines; a body that ends inside a line is an error. */
|
||||
async function* lines(chunks: AsyncIterable<Uint8Array>): AsyncGenerator<{ readonly text: string; number: number }> {
|
||||
const decoder = new TextDecoder("utf-8", { fatal: true });
|
||||
let rest = "";
|
||||
let number = 0;
|
||||
const decode = (chunk?: Uint8Array): string => {
|
||||
try {
|
||||
return chunk === undefined ? decoder.decode() : decoder.decode(chunk, { stream: true });
|
||||
} catch (error) {
|
||||
throw new LogFormatError(`Log line ${number + 1} is not UTF-8`, { cause: error });
|
||||
}
|
||||
};
|
||||
for await (const chunk of chunks) {
|
||||
rest += decode(chunk);
|
||||
let start = 0;
|
||||
for (let end = rest.indexOf("\n"); end >= 0; end = rest.indexOf("\n", start)) {
|
||||
yield { text: rest.slice(start, end), number: ++number };
|
||||
start = end + 1;
|
||||
}
|
||||
rest = rest.slice(start);
|
||||
}
|
||||
rest += decode();
|
||||
if (rest.length > 0) throw new LogFormatError(`The body ends inside log line ${number + 1}`);
|
||||
}
|
||||
|
||||
function gzip(chunks: AsyncIterable<Uint8Array>): SyncBody {
|
||||
return pipe(chunks, new CompressionStream("gzip"));
|
||||
}
|
||||
|
||||
function gunzip(chunks: SyncBody): AsyncIterable<Uint8Array> {
|
||||
return pipe(chunks, new DecompressionStream("gzip"));
|
||||
}
|
||||
|
||||
/**
|
||||
* A batch as uploaded: outbox commits after one sequence as gzip-compressed JSON lines. gzip, because
|
||||
* `CompressionStream` offers it in Node, Workers, and browsers alike, and a browser viewer reads batches too.
|
||||
* `chunks` through `transform`, as they come: the source is written while the output is read, so each holds no more
|
||||
* than the stream's buffers. Stopping early, or a failure on either side, ends the source too (its `return()`, which
|
||||
* for an export gives the store's commit queue back).
|
||||
*/
|
||||
export async function encodeBatch(commits: readonly OutboxCommit[]): Promise<Uint8Array> {
|
||||
return gzip(new TextEncoder().encode(commits.map(encodeLogLine).join("")));
|
||||
}
|
||||
|
||||
export async function decodeBatch(bytes: Uint8Array): Promise<OutboxCommit[]> {
|
||||
const { commits, torn } = decodeLogLines(await gunzipText(bytes));
|
||||
if (torn > 0) throw new LogFormatError("Batch ends inside a log line");
|
||||
return commits;
|
||||
}
|
||||
|
||||
/** A snapshot as uploaded: a store archive (`store.export()`), one part per JSON line, gzip-compressed. */
|
||||
export async function encodeSnapshot(parts: AsyncIterable<ArchivePart>): Promise<Uint8Array> {
|
||||
let text = "";
|
||||
for await (const part of parts) text += `${JSON.stringify(part)}\n`;
|
||||
return gzip(new TextEncoder().encode(text));
|
||||
}
|
||||
|
||||
/** The parts of a snapshot, unchecked: `importStore()` checks them (a trust boundary). */
|
||||
export async function decodeSnapshot(bytes: Uint8Array): Promise<unknown[]> {
|
||||
const text = await gunzipText(bytes);
|
||||
if (text.length > 0 && !text.endsWith("\n")) throw new LogFormatError("Snapshot ends inside a line");
|
||||
return text
|
||||
.split("\n")
|
||||
.filter((line) => line.length > 0)
|
||||
.map((line, index) => {
|
||||
try {
|
||||
return JSON.parse(line) as unknown;
|
||||
} catch (error) {
|
||||
throw new LogFormatError(`Snapshot line ${index + 1} is not JSON`, { cause: error });
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function gzip(bytes: Uint8Array): Promise<Uint8Array> {
|
||||
return transform(bytes, new CompressionStream("gzip"));
|
||||
}
|
||||
|
||||
async function gunzipText(bytes: Uint8Array): Promise<string> {
|
||||
return new TextDecoder("utf-8", { fatal: true }).decode(await transform(bytes, new DecompressionStream("gzip")));
|
||||
}
|
||||
|
||||
async function transform(bytes: Uint8Array, stream: CompressionStream | DecompressionStream): Promise<Uint8Array> {
|
||||
const output = new Blob([bytes as Uint8Array<ArrayBuffer>]).stream().pipeThrough(stream);
|
||||
return new Uint8Array(await new Response(output).arrayBuffer());
|
||||
async function* pipe(
|
||||
chunks: AsyncIterable<Uint8Array>,
|
||||
transform: CompressionStream | DecompressionStream,
|
||||
): AsyncGenerator<Uint8Array> {
|
||||
const writer = transform.writable.getWriter();
|
||||
const reader = transform.readable.getReader();
|
||||
const writing = (async () => {
|
||||
try {
|
||||
for await (const chunk of chunks) await writer.write(chunk as Uint8Array<ArrayBuffer>);
|
||||
await writer.close();
|
||||
} catch (error) {
|
||||
await writer.abort(error).catch(() => {});
|
||||
throw error;
|
||||
}
|
||||
})();
|
||||
// Its failure reaches the reader too (the abort errors the readable side); this only keeps it from going unhandled.
|
||||
writing.catch(() => {});
|
||||
let done = false;
|
||||
try {
|
||||
while (true) {
|
||||
const result = await reader.read();
|
||||
if (result.done) break;
|
||||
yield result.value;
|
||||
}
|
||||
done = true;
|
||||
await writing;
|
||||
} finally {
|
||||
if (!done) await reader.cancel().catch(() => {});
|
||||
reader.releaseLock();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,7 +1,10 @@
|
|||
// Session sync over a Chord store's outbox: upload the commits after the last sequence a copy shares with a remote
|
||||
// into a slot the remote creates once, pull a copy forward with `applyBatch`, restore from an archive, fold history
|
||||
// into snapshots, and decide when to upload. The remote and the host's sync state are interfaces (`remote.ts`); a
|
||||
// directory-backed pair for Node is in `./node`.
|
||||
// into snapshots, and decide when to upload. The remote and the host's sync state are interfaces (`remote.ts`).
|
||||
//
|
||||
// Bounded memory: bodies are streams (`batch.ts`). An upload reads the outbox a page at a time and stops a batch at
|
||||
// `BATCH_BYTES`; a pull applies a batch in chunks of at most `APPLY_BYTES`; a snapshot is exported and imported as it
|
||||
// streams. Nothing holds a whole batch, snapshot or store, so a Durable Object's isolate (128 MB) can run it.
|
||||
|
||||
import type { Context } from "../context/index.ts";
|
||||
import { applyBatch } from "../store/apply-batch.ts";
|
||||
|
|
@ -9,11 +12,12 @@ import type { Backend } from "../store/backend.ts";
|
|||
import { importStore } from "../store/import.ts";
|
||||
import type { OutboxCommit } from "../store/outbox.ts";
|
||||
import { type ChordStore, openStore, type Store } from "../store/store.ts";
|
||||
import { decodeBatch, decodeSnapshot, encodeBatch, encodeSnapshot } from "./batch.ts";
|
||||
import { decodeBatchLines, decodeSnapshot, encodeBatch, encodeSnapshot, type SyncBody } from "./batch.ts";
|
||||
import {
|
||||
type BatchInfo,
|
||||
HistoryUnavailableError,
|
||||
LOG_START,
|
||||
type PutBatchResult,
|
||||
remoteHead,
|
||||
remoteHistory,
|
||||
type Seq,
|
||||
|
|
@ -21,8 +25,17 @@ import {
|
|||
type SessionSyncStore,
|
||||
} from "./remote.ts";
|
||||
|
||||
/** Commits read from the outbox per uploaded batch. */
|
||||
const BATCH_COMMITS = 1000;
|
||||
/** Commits read from the outbox per page at most. */
|
||||
const OUTBOX_PAGE = 100;
|
||||
/** About how much one outbox page holds: the count of the next page follows the size of the commits read. */
|
||||
const PAGE_BYTES = 1024 * 1024;
|
||||
/**
|
||||
* Where an upload ends a batch: once its lines (before compression) reach this size. One commit larger than this is
|
||||
* a batch of its own.
|
||||
*/
|
||||
export const BATCH_BYTES = 8 * 1024 * 1024;
|
||||
/** How much of a batch a pull holds before it applies it: commits are applied in chunks of about this size. */
|
||||
export const APPLY_BYTES = 4 * 1024 * 1024;
|
||||
|
||||
/**
|
||||
* What a host keeps for each copy of a session it syncs: the last sequence the copy shares with the remote. Commits
|
||||
|
|
@ -123,22 +136,30 @@ export async function uploadSession(options: {
|
|||
const head = storeHead(source);
|
||||
if (head < synced) return { type: "behind", head, syncedHead: synced };
|
||||
let to: Seq;
|
||||
let body: Uint8Array;
|
||||
let body: SyncBody;
|
||||
let archive = false;
|
||||
let close: (() => Promise<void>) | undefined;
|
||||
if (synced === LOG_START && outbox.floor > 0) {
|
||||
const snapshot = await exportSnapshot(source, context);
|
||||
to = snapshot.to;
|
||||
body = snapshot.body;
|
||||
close = snapshot.close;
|
||||
archive = true;
|
||||
} else {
|
||||
const commits = await readOutbox(source, synced, BATCH_COMMITS, context);
|
||||
if (commits.length === 0)
|
||||
const range = await outboxRange(source, synced, context);
|
||||
if (range === undefined)
|
||||
return uploaded === undefined ? { type: "idle" } : { type: "uploaded", head: uploaded };
|
||||
to = commits.at(-1)!.seq;
|
||||
body = await encodeBatch(commits);
|
||||
to = range.to;
|
||||
body = encodeBatch(outboxCommits(source, synced, to, context));
|
||||
}
|
||||
const after = synced;
|
||||
const result = await store.putBatch(sessionId, after, to, device, body, context, archive ? { archive } : {});
|
||||
let result: PutBatchResult;
|
||||
try {
|
||||
result = await store.putBatch(sessionId, after, to, device, body, context, archive ? { archive } : {});
|
||||
} finally {
|
||||
// A remote that answered without reading the archive leaves the export holding the commit queue.
|
||||
await close?.();
|
||||
}
|
||||
// A history that does not end at `after` is another one too: the remote never held this copy's past `after`.
|
||||
if (result.type === "not_head") return { type: "diverged", at: after };
|
||||
if (result.type === "exists" && result.device !== device) return { type: "diverged", at: after };
|
||||
|
|
@ -150,23 +171,123 @@ export async function uploadSession(options: {
|
|||
}
|
||||
}
|
||||
|
||||
/** The store's state at its head as a snapshot body. `export()` holds the commit queue, so the head cannot move. */
|
||||
async function exportSnapshot(
|
||||
store: Store,
|
||||
context: Context,
|
||||
): Promise<{ readonly to: Seq; readonly body: Uint8Array }> {
|
||||
/**
|
||||
* Where the next batch after `after` ends: the outbox read a page at a time, holding one page, until the lines reach
|
||||
* `BATCH_BYTES` (one commit at least). Undefined when the outbox has nothing after `after`.
|
||||
*/
|
||||
async function outboxRange(store: Store, after: Seq, context: Context): Promise<{ readonly to: Seq } | undefined> {
|
||||
let to: Seq | undefined;
|
||||
let bytes = 0;
|
||||
const pages = new OutboxPages(store, after, context);
|
||||
for (let page = await pages.next(); page.length > 0; page = await pages.next()) {
|
||||
for (const commit of page) {
|
||||
if (to !== undefined && bytes + encodedSize(commit) > BATCH_BYTES) return { to };
|
||||
bytes += encodedSize(commit);
|
||||
to = commit.seq;
|
||||
}
|
||||
if (bytes >= BATCH_BYTES) break;
|
||||
}
|
||||
return to === undefined ? undefined : { to };
|
||||
}
|
||||
|
||||
/** The outbox commits after `after` through `to`, a page at a time, as the body is read. */
|
||||
async function* outboxCommits(store: Store, after: Seq, to: Seq, context: Context): AsyncGenerator<OutboxCommit> {
|
||||
const pages = new OutboxPages(store, after, context);
|
||||
let at = after;
|
||||
while (at < to) {
|
||||
const page = await pages.next();
|
||||
if (page.length === 0) throw new OutboxUnavailableError(`The outbox ends at ${at}, before ${to}`);
|
||||
for (const commit of page) {
|
||||
if (commit.seq > to) return;
|
||||
yield commit;
|
||||
at = commit.seq;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Pages of the outbox after a sequence, each about `PAGE_BYTES`: the outbox reads by count, so the next page's count
|
||||
* follows the commits read so far (`OUTBOX_PAGE` to start, one commit at least). A run of large commits reads one at a
|
||||
* time instead of `OUTBOX_PAGE` of them at once.
|
||||
*/
|
||||
class OutboxPages {
|
||||
readonly #store: Store;
|
||||
readonly #context: Context;
|
||||
#after: Seq;
|
||||
#limit = OUTBOX_PAGE;
|
||||
#done = false;
|
||||
|
||||
constructor(store: Store, after: Seq, context: Context) {
|
||||
this.#store = store;
|
||||
this.#after = after;
|
||||
this.#context = context;
|
||||
}
|
||||
|
||||
async next(): Promise<OutboxCommit[]> {
|
||||
if (this.#done) return [];
|
||||
const page = await readOutbox(this.#store, this.#after, this.#limit, this.#context);
|
||||
if (page.length < this.#limit) this.#done = true;
|
||||
if (page.length === 0) return page;
|
||||
this.#after = page.at(-1)!.seq;
|
||||
const average = page.reduce((size, commit) => size + encodedSize(commit), 0) / page.length;
|
||||
this.#limit = Math.max(1, Math.min(OUTBOX_PAGE, Math.floor(PAGE_BYTES / average)));
|
||||
return page;
|
||||
}
|
||||
}
|
||||
|
||||
/** A commit's line size before compression, as `BATCH_BYTES` counts it. */
|
||||
function encodedSize(commit: OutboxCommit): number {
|
||||
return JSON.stringify(commit.writes).length + 64;
|
||||
}
|
||||
|
||||
/** A snapshot of a store being exported: where it ends, its body, and `close()` for when the body is not read to the end. */
|
||||
type ExportedSnapshot = { readonly to: Seq; readonly body: SyncBody; close(): Promise<void> };
|
||||
|
||||
/**
|
||||
* The store's state as a streamed snapshot body. The export holds the store's commit queue from its first part until
|
||||
* the body is read to the end or `close()` is called, so `to` (its manifest's `lastSeq`) is exactly what it holds.
|
||||
*/
|
||||
async function exportSnapshot(store: Store, context: Context): Promise<ExportedSnapshot> {
|
||||
const parts = store.export(context);
|
||||
const body = await encodeSnapshot(
|
||||
(async function* () {
|
||||
for await (const part of parts) {
|
||||
if (part.kind === "manifest") to = part.lastSeq;
|
||||
yield part;
|
||||
}
|
||||
})(),
|
||||
);
|
||||
if (to === undefined) throw new Error("A store export has no manifest");
|
||||
return { to, body };
|
||||
const first = await parts.next();
|
||||
if (first.done || first.value.kind !== "manifest") {
|
||||
await parts.return?.(undefined);
|
||||
throw new Error("A store export does not start with its manifest");
|
||||
}
|
||||
const manifest = first.value;
|
||||
return {
|
||||
to: manifest.lastSeq,
|
||||
body: encodeSnapshot(
|
||||
(async function* () {
|
||||
yield manifest;
|
||||
yield* parts;
|
||||
})(),
|
||||
),
|
||||
close: async () => {
|
||||
await parts.return?.(undefined);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply a batch body to `store` as it streams, in chunks of about `APPLY_BYTES`, skipping the commits the store has
|
||||
* (a pull resuming inside a batch). Each chunk is one `applyBatch()` transaction; a pull stopped between chunks
|
||||
* resumes after the last applied one.
|
||||
*/
|
||||
async function applyBody(store: ChordStore, body: SyncBody, context: Context): Promise<void> {
|
||||
let chunk: OutboxCommit[] = [];
|
||||
let bytes = 0;
|
||||
for await (const line of decodeBatchLines(body)) {
|
||||
if (line.commit.seq <= store.lastSeq) continue;
|
||||
chunk.push(line.commit);
|
||||
bytes += line.bytes;
|
||||
if (bytes >= APPLY_BYTES) {
|
||||
await applyBatch(store, chunk, context);
|
||||
chunk = [];
|
||||
bytes = 0;
|
||||
}
|
||||
}
|
||||
if (chunk.length > 0) await applyBatch(store, chunk, context);
|
||||
}
|
||||
|
||||
/** This copy has commits it has not uploaded: upload them first, which may find that the copy diverged. */
|
||||
|
|
@ -208,7 +329,7 @@ export async function pullSession(options: {
|
|||
if (at() > synced) throw new UnsyncedCommitsError(sessionId, at(), synced);
|
||||
const history = remoteHistory(listing, at());
|
||||
try {
|
||||
const bodies: { readonly to: Seq; readonly archive: boolean; read(): Promise<Uint8Array | undefined> }[] = [];
|
||||
const bodies: { readonly to: Seq; readonly archive: boolean; read(): Promise<SyncBody | undefined> }[] = [];
|
||||
if (history.snapshot !== undefined) {
|
||||
const snapshot = history.snapshot;
|
||||
bodies.push({ to: snapshot, archive: true, read: () => remote.getSnapshot(sessionId, snapshot, context) });
|
||||
|
|
@ -228,15 +349,13 @@ export async function pullSession(options: {
|
|||
if (body.archive) {
|
||||
// An archive restores a whole store: only into an empty backend.
|
||||
if (target !== undefined || local.kind !== "empty") throw new HistoryUnavailableError(at());
|
||||
target = await importStore(local.backend, await decodeSnapshot(bytes), context, { outbox: true });
|
||||
target = await importStore(local.backend, decodeSnapshot(bytes), context, { outbox: true });
|
||||
} else {
|
||||
if (target === undefined) {
|
||||
if (local.kind !== "empty") throw new Error("unreachable: a store copy always has a target");
|
||||
target = await openStore(local.backend, context, { outbox: true });
|
||||
}
|
||||
const from = target.lastSeq;
|
||||
const commits = (await decodeBatch(bytes)).filter((commit) => commit.seq > from);
|
||||
if (commits.length > 0) await applyBatch(target as ChordStore, commits, context);
|
||||
await applyBody(target as ChordStore, bytes, context);
|
||||
}
|
||||
await state.set(sessionId, { syncedHead: at() }, context);
|
||||
}
|
||||
|
|
@ -259,8 +378,8 @@ export async function pullSession(options: {
|
|||
}
|
||||
|
||||
/**
|
||||
* Fold a session's history into one snapshot: restore it into `scratch` (an empty backend, for example an in-memory
|
||||
* SQLite database), upload its `export()` as a snapshot, drop the commits of the batches it covers, keeping their
|
||||
* Fold a session's history into one snapshot: restore it into `scratch` (an empty backend; a file or a Durable Object's
|
||||
* SQLite storage, not `:memory:`, so the restored store is not held in memory), upload its `export()` as a snapshot, drop the commits of the batches it covers, keeping their
|
||||
* slots taken, and delete older snapshots. The remote runs this, not clients.
|
||||
*/
|
||||
export async function foldSession(options: {
|
||||
|
|
@ -282,7 +401,11 @@ export async function foldSession(options: {
|
|||
});
|
||||
try {
|
||||
const snapshot = await exportSnapshot(restored.store, context);
|
||||
await store.putSnapshot(sessionId, snapshot.to, snapshot.body, context);
|
||||
try {
|
||||
await store.putSnapshot(sessionId, snapshot.to, snapshot.body, context);
|
||||
} finally {
|
||||
await snapshot.close();
|
||||
}
|
||||
await store.foldBatches(
|
||||
sessionId,
|
||||
history.batches.filter((batch: BatchInfo) => batch.to <= snapshot.to).map((batch) => batch.after),
|
||||
|
|
|
|||
|
|
@ -10,12 +10,27 @@ export type FileResult<T> = { readonly ok: true; readonly value: T } | { readonl
|
|||
/** A directory entry. */
|
||||
export type SyncFileInfo = { readonly name: string; readonly kind: "file" | "directory" | "symlink" };
|
||||
|
||||
/** A file opened for reads at offsets: what streaming a body back needs. */
|
||||
export interface SyncFileReader {
|
||||
/** Up to `length` bytes at `offset`; fewer only at end of file. */
|
||||
read(offset: number, length: number, context: Context): Promise<FileResult<Uint8Array>>;
|
||||
close(context: Context): Promise<void>;
|
||||
}
|
||||
|
||||
/** File operations for the directory-backed sync pieces. A missing file or directory fails with `code: "not_found"`. */
|
||||
export interface SyncFileSystem {
|
||||
joinPath(parts: string[], context: Context): Promise<FileResult<string>>;
|
||||
readTextFile(path: string, context: Context): Promise<FileResult<string>>;
|
||||
readBinaryFile(path: string, context: Context): Promise<FileResult<Uint8Array>>;
|
||||
/** Open a file for bounded reads, so a body streams back without being read whole. */
|
||||
openBinaryReader(
|
||||
path: string,
|
||||
options: { noFollow?: boolean } | undefined,
|
||||
context: Context,
|
||||
): Promise<FileResult<SyncFileReader>>;
|
||||
writeFile(path: string, content: string | Uint8Array, context: Context): Promise<FileResult<void>>;
|
||||
/** Append to a file, creating it: a body is written chunk by chunk as it streams. */
|
||||
appendFile(path: string, content: string | Uint8Array, context: Context): Promise<FileResult<void>>;
|
||||
renameFile(sourcePath: string, destinationPath: string, context: Context): Promise<FileResult<void>>;
|
||||
listDir(path: string, context: Context): Promise<FileResult<SyncFileInfo[]>>;
|
||||
createDir(path: string, options: { recursive?: boolean } | undefined, context: Context): Promise<FileResult<void>>;
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import type { Context } from "../context/index.ts";
|
||||
import type { SyncBody } from "./batch.ts";
|
||||
import { isNotFound, type SyncFileSystem, unwrap } from "./files.ts";
|
||||
import {
|
||||
type BatchInfo,
|
||||
|
|
@ -12,6 +13,13 @@ import {
|
|||
} from "./remote.ts";
|
||||
|
||||
const pad = (value: number): string => String(value).padStart(16, "0");
|
||||
/** How much of a body is read from its file at a time. */
|
||||
const READ_CHUNK = 256 * 1024;
|
||||
|
||||
/** One buffer as a body. */
|
||||
async function* once(bytes: Uint8Array): SyncBody {
|
||||
yield bytes;
|
||||
}
|
||||
const SLOT_NAME = /^(\d{16})\.json$/;
|
||||
const SNAPSHOT_NAME = /^(\d{16})\.batch$/;
|
||||
const SESSION_ID = /^[A-Za-z0-9_-][A-Za-z0-9._-]*$/;
|
||||
|
|
@ -49,7 +57,7 @@ export class FileSystemSyncRemote implements SessionSyncMaintenance {
|
|||
after: Seq,
|
||||
to: Seq,
|
||||
device: string,
|
||||
body: Uint8Array,
|
||||
body: SyncBody,
|
||||
context: Context,
|
||||
options: { readonly archive?: boolean } = {},
|
||||
): Promise<PutBatchResult> {
|
||||
|
|
@ -64,23 +72,23 @@ export class FileSystemSyncRemote implements SessionSyncMaintenance {
|
|||
sessionId,
|
||||
["slots"],
|
||||
`${pad(after)}.json`,
|
||||
new TextEncoder().encode(JSON.stringify(slot)),
|
||||
once(new TextEncoder().encode(JSON.stringify(slot))),
|
||||
context,
|
||||
);
|
||||
return { type: "created" };
|
||||
}
|
||||
|
||||
async getBatch(sessionId: string, after: Seq, context: Context): Promise<Uint8Array | undefined> {
|
||||
async getBatch(sessionId: string, after: Seq, context: Context): Promise<SyncBody | undefined> {
|
||||
const slot = await this.#slot(sessionId, after, context);
|
||||
if (slot === undefined || slot.folded) return undefined;
|
||||
return this.#read(context, sessionId, "batches", `${pad(after)}.batch`);
|
||||
return this.#stream(context, sessionId, "batches", `${pad(after)}.batch`);
|
||||
}
|
||||
|
||||
getSnapshot(sessionId: string, to: Seq, context: Context): Promise<Uint8Array | undefined> {
|
||||
return this.#read(context, sessionId, "snapshots", `${pad(to)}.batch`);
|
||||
getSnapshot(sessionId: string, to: Seq, context: Context): Promise<SyncBody | undefined> {
|
||||
return this.#stream(context, sessionId, "snapshots", `${pad(to)}.batch`);
|
||||
}
|
||||
|
||||
async putSnapshot(sessionId: string, to: Seq, body: Uint8Array, context: Context): Promise<void> {
|
||||
async putSnapshot(sessionId: string, to: Seq, body: SyncBody, context: Context): Promise<void> {
|
||||
await this.#write(sessionId, ["snapshots"], `${pad(to)}.batch`, body, context);
|
||||
}
|
||||
|
||||
|
|
@ -93,7 +101,7 @@ export class FileSystemSyncRemote implements SessionSyncMaintenance {
|
|||
sessionId,
|
||||
["slots"],
|
||||
`${pad(after)}.json`,
|
||||
new TextEncoder().encode(JSON.stringify(folded)),
|
||||
once(new TextEncoder().encode(JSON.stringify(folded))),
|
||||
context,
|
||||
);
|
||||
const body = await this.#path(context, sessionId, "batches", `${pad(after)}.batch`);
|
||||
|
|
@ -145,7 +153,7 @@ export class FileSystemSyncRemote implements SessionSyncMaintenance {
|
|||
}
|
||||
|
||||
async putMeta(sessionId: string, meta: JsonObject, context: Context): Promise<void> {
|
||||
await this.#write(sessionId, [], "meta.json", new TextEncoder().encode(JSON.stringify(meta)), context);
|
||||
await this.#write(sessionId, [], "meta.json", once(new TextEncoder().encode(JSON.stringify(meta))), context);
|
||||
}
|
||||
|
||||
async getMeta(sessionId: string, context: Context): Promise<JsonObject | undefined> {
|
||||
|
|
@ -170,12 +178,38 @@ export class FileSystemSyncRemote implements SessionSyncMaintenance {
|
|||
return unwrap(result);
|
||||
}
|
||||
|
||||
/** Write through a temporary file and a rename, so a reader never sees half a file. */
|
||||
/** A body streamed back from its file, a chunk at a time; undefined when the file is missing. */
|
||||
async #stream(context: Context, sessionId: string, ...parts: string[]): Promise<SyncBody | undefined> {
|
||||
const opened = await this.#fs.openBinaryReader(
|
||||
await this.#path(context, sessionId, ...parts),
|
||||
undefined,
|
||||
context,
|
||||
);
|
||||
if (isNotFound(opened)) return undefined;
|
||||
const reader = unwrap(opened);
|
||||
return (async function* () {
|
||||
try {
|
||||
for (let offset = 0; ; ) {
|
||||
const chunk = unwrap(await reader.read(offset, READ_CHUNK, context));
|
||||
if (chunk.length === 0) return;
|
||||
offset += chunk.length;
|
||||
yield chunk;
|
||||
}
|
||||
} finally {
|
||||
await reader.close(context);
|
||||
}
|
||||
})();
|
||||
}
|
||||
|
||||
/**
|
||||
* Write a body through a temporary file and a rename, so a reader never sees half a file; the body is appended chunk
|
||||
* by chunk as it streams.
|
||||
*/
|
||||
async #write(
|
||||
sessionId: string,
|
||||
directory: readonly string[],
|
||||
name: string,
|
||||
body: Uint8Array,
|
||||
body: SyncBody,
|
||||
context: Context,
|
||||
): Promise<void> {
|
||||
unwrap(
|
||||
|
|
@ -183,7 +217,13 @@ export class FileSystemSyncRemote implements SessionSyncMaintenance {
|
|||
);
|
||||
const path = await this.#path(context, sessionId, ...directory, name);
|
||||
const temporary = await this.#path(context, sessionId, ...directory, `${name}.tmp`);
|
||||
unwrap(await this.#fs.writeFile(temporary, body, context));
|
||||
unwrap(await this.#fs.writeFile(temporary, new Uint8Array(), context));
|
||||
try {
|
||||
for await (const chunk of body) unwrap(await this.#fs.appendFile(temporary, chunk, context));
|
||||
} catch (error) {
|
||||
await this.#fs.remove(temporary, { force: true }, context);
|
||||
throw error;
|
||||
}
|
||||
unwrap(await this.#fs.renameFile(temporary, path, context));
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,8 +1,17 @@
|
|||
// Session sync over a store's outbox (`@earendil-works/chord/sync`, experimental). Portable: the remote and the host's
|
||||
// sync state are interfaces, and the directory-backed ones take their file operations (`SyncFileSystem`).
|
||||
|
||||
export { decodeBatch, decodeSnapshot, encodeBatch, encodeSnapshot } from "./batch.ts";
|
||||
export {
|
||||
decodeBatch,
|
||||
decodeBatchLines,
|
||||
decodeSnapshot,
|
||||
encodeBatch,
|
||||
encodeSnapshot,
|
||||
type SyncBody,
|
||||
} from "./batch.ts";
|
||||
export {
|
||||
APPLY_BYTES,
|
||||
BATCH_BYTES,
|
||||
FlushScheduler,
|
||||
type FlushSchedulerOptions,
|
||||
foldSession,
|
||||
|
|
@ -17,9 +26,9 @@ export {
|
|||
type UploadResult,
|
||||
uploadSession,
|
||||
} from "./client.ts";
|
||||
export type { FileResult, SyncFileInfo, SyncFileSystem } from "./files.ts";
|
||||
export type { FileResult, SyncFileInfo, SyncFileReader, SyncFileSystem } from "./files.ts";
|
||||
export { checkSessionId, FileSystemSyncRemote } from "./filesystem.ts";
|
||||
export { decodeLogLines, encodeLogLine, LOG_VERSION, LogFormatError } from "./lines.ts";
|
||||
export { decodeLogLine, encodeLogLine, LOG_VERSION, LogFormatError } from "./lines.ts";
|
||||
export {
|
||||
type BatchInfo,
|
||||
HistoryUnavailableError,
|
||||
|
|
|
|||
|
|
@ -27,29 +27,11 @@ export function encodeLogLine(commit: OutboxCommit): string {
|
|||
}
|
||||
|
||||
/**
|
||||
* Decode JSON lines. A last line without its newline is a torn append and is returned as `torn` instead of a commit;
|
||||
* anything else that does not decode is an error. A line of an unknown version is an error too: an older pi must not
|
||||
* apply writes it does not understand. The writes themselves are checked by `applyBatch()`, a trust boundary.
|
||||
* One line (without its newline) as an outbox commit; `number` names the line in errors. A line of an unknown version
|
||||
* is an error: an older pi must not apply writes it does not understand. The writes themselves are checked by
|
||||
* `applyBatch()`, a trust boundary.
|
||||
*/
|
||||
export function decodeLogLines(text: string): { readonly commits: OutboxCommit[]; readonly torn: number } {
|
||||
const commits: OutboxCommit[] = [];
|
||||
let start = 0;
|
||||
let previous = 0;
|
||||
while (start < text.length) {
|
||||
const end = text.indexOf("\n", start);
|
||||
if (end === -1) return { commits, torn: text.length - start };
|
||||
const commit = decodeLine(text.slice(start, end), commits.length + 1);
|
||||
if (commit.seq <= previous) {
|
||||
throw new LogFormatError(`Log line ${commits.length + 1} does not increase the sequence`);
|
||||
}
|
||||
previous = commit.seq;
|
||||
commits.push(commit);
|
||||
start = end + 1;
|
||||
}
|
||||
return { commits, torn: 0 };
|
||||
}
|
||||
|
||||
function decodeLine(text: string, number: number): OutboxCommit {
|
||||
export function decodeLogLine(text: string, number: number): OutboxCommit {
|
||||
let value: unknown;
|
||||
try {
|
||||
value = JSON.parse(text);
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import type { Context } from "../context/index.ts";
|
||||
import type { SyncBody } from "./batch.ts";
|
||||
|
||||
/** A store's commit sequence (`store.lastSeq`, an outbox commit's `seq`); 0 is before every commit. */
|
||||
export type Seq = number;
|
||||
|
|
@ -62,19 +63,22 @@ export type SessionChanges = {
|
|||
|
||||
/** The client's view of a remote store. Clients never delete or fold. */
|
||||
export interface SessionSyncStore {
|
||||
/** Create the slot after `after` unless it exists or the history does not end at `after`. */
|
||||
/**
|
||||
* Create the slot after `after` unless it exists or the history does not end at `after`. `body` is a stream read once;
|
||||
* a remote that answers without reading it (`exists`, `not_head`) leaves it unread.
|
||||
*/
|
||||
putBatch(
|
||||
sessionId: string,
|
||||
after: Seq,
|
||||
to: Seq,
|
||||
device: string,
|
||||
body: Uint8Array,
|
||||
body: SyncBody,
|
||||
context: Context,
|
||||
options?: { readonly archive?: boolean },
|
||||
): Promise<PutBatchResult>;
|
||||
/** A batch's body; `undefined` when its slot is folded or absent. */
|
||||
getBatch(sessionId: string, after: Seq, context: Context): Promise<Uint8Array | undefined>;
|
||||
getSnapshot(sessionId: string, to: Seq, context: Context): Promise<Uint8Array | undefined>;
|
||||
getBatch(sessionId: string, after: Seq, context: Context): Promise<SyncBody | undefined>;
|
||||
getSnapshot(sessionId: string, to: Seq, context: Context): Promise<SyncBody | undefined>;
|
||||
list(sessionId: string, context: Context): Promise<RemoteListing>;
|
||||
/**
|
||||
* The sessions changed after `cursor` (all of them without one). A remote may also list sessions that did not change,
|
||||
|
|
@ -91,7 +95,7 @@ export interface SessionSyncStore {
|
|||
|
||||
/** The remote's own side: folding history into snapshots, which clients never do. */
|
||||
export interface SessionSyncMaintenance extends SessionSyncStore {
|
||||
putSnapshot(sessionId: string, to: Seq, body: Uint8Array, context: Context): Promise<void>;
|
||||
putSnapshot(sessionId: string, to: Seq, body: SyncBody, context: Context): Promise<void>;
|
||||
/** Drop the bodies of these slots, keeping the slots taken so nobody can create them again. */
|
||||
foldBatches(sessionId: string, afters: readonly Seq[], context: Context): Promise<void>;
|
||||
deleteSnapshots(sessionId: string, snapshots: readonly Seq[], context: Context): Promise<void>;
|
||||
|
|
|
|||
169
packages/chord/test/bench/sync-memory.ts
Normal file
169
packages/chord/test/bench/sync-memory.ts
Normal file
|
|
@ -0,0 +1,169 @@
|
|||
// Memory benchmark of the sync layer inside Durable Objects (RFC 0074's budget: 96 to 128 MB per isolate): workerd
|
||||
// through Miniflare, the objects of `sync-worker.ts` on their own SQLite storage, and the isolate's V8 heap sampled
|
||||
// through the inspector while each step runs. Steps per session size: fill the origin's store (commits with its
|
||||
// outbox), upload (outbox to remote, bodies streamed into the remote's SQLite), pull (restore a replica through the
|
||||
// origin's remote over RPC), fold (restore into scratch, export a snapshot back). Each step starts after a full GC; its
|
||||
// baseline is the used heap then, and its peak the highest `usedSize` sampled until it answers.
|
||||
//
|
||||
// workerd runs with `--expose-gc` and `--max-old-space-size=128` (Cloudflare's per-isolate limit), so a step that
|
||||
// needed more than 128 MB would fail, not just report a big number.
|
||||
//
|
||||
// Run by hand from packages/chord (not in CI; about three minutes):
|
||||
// node --conditions=source --experimental-strip-types test/bench/sync-memory.ts
|
||||
|
||||
import { build } from "esbuild";
|
||||
import { convertV4MiniflareOptions, Miniflare } from "miniflare";
|
||||
import WebSocket from "ws";
|
||||
|
||||
/** Session histories: total MB of row text, and the size of each commit's row. */
|
||||
const RUNS = [
|
||||
{ sizeMb: 8, rowBytes: 64 * 1024 },
|
||||
{ sizeMb: 32, rowBytes: 64 * 1024 },
|
||||
{ sizeMb: 128, rowBytes: 64 * 1024 },
|
||||
{ sizeMb: 256, rowBytes: 64 * 1024 },
|
||||
// Few large commits: each one close to a whole batch, the case that holds the most at once.
|
||||
{ sizeMb: 64, rowBytes: 4 * 1024 * 1024 },
|
||||
];
|
||||
const INSPECTOR_PORT = 9341;
|
||||
const SAMPLE_MS = 10;
|
||||
const MB = 1024 * 1024;
|
||||
|
||||
process.env.MINIFLARE_WORKERD_V8_FLAGS = "--expose-gc --max-old-space-size=128";
|
||||
|
||||
const bundled = await build({
|
||||
entryPoints: [new URL("./sync-worker.ts", import.meta.url).pathname],
|
||||
bundle: true,
|
||||
format: "esm",
|
||||
platform: "neutral",
|
||||
write: false,
|
||||
external: ["cloudflare:*"],
|
||||
conditions: ["source", "workerd", "worker"],
|
||||
mainFields: ["module", "main"],
|
||||
target: "es2023",
|
||||
});
|
||||
|
||||
const mf = new Miniflare(
|
||||
convertV4MiniflareOptions({
|
||||
inspectorPort: INSPECTOR_PORT,
|
||||
workers: [
|
||||
{
|
||||
name: "sync-bench",
|
||||
compatibilityDate: "2026-10-01",
|
||||
compatibilityFlags: ["nodejs_compat"],
|
||||
modules: [{ type: "ESModule", path: "worker.mjs", contents: bundled.outputFiles[0]!.text }],
|
||||
durableObjects: {
|
||||
ORIGIN: { className: "Origin", useSQLite: true },
|
||||
REPLICA: { className: "Replica", useSQLite: true },
|
||||
FOLDER: { className: "Folder", useSQLite: true },
|
||||
},
|
||||
},
|
||||
],
|
||||
}),
|
||||
);
|
||||
|
||||
try {
|
||||
await mf.ready;
|
||||
await mf.dispatchFetch("http://bench/unknown");
|
||||
const [target] = (await (await fetch(`http://127.0.0.1:${INSPECTOR_PORT}/json/list`)).json()) as {
|
||||
webSocketDebuggerUrl: string;
|
||||
}[];
|
||||
const inspector = await connect(target!.webSocketDebuggerUrl);
|
||||
|
||||
const rows: string[] = [];
|
||||
for (const { sizeMb, rowBytes } of RUNS) {
|
||||
const run = `run${sizeMb}x${rowBytes}`;
|
||||
const count = (sizeMb * MB) / rowBytes;
|
||||
const label = `${sizeMb} MB in ${mb(rowBytes)} MB rows`;
|
||||
for (const step of ["fill", "upload", "pull", "fold"] as const) {
|
||||
const query = step === "fill" ? `&count=${count}&size=${rowBytes}` : "";
|
||||
const measured = await measure(inspector, () => mf.dispatchFetch(`http://bench/${step}?run=${run}${query}`));
|
||||
const body = (await measured.response.json()) as { ms: number; result: unknown };
|
||||
if (!measured.response.ok) throw new Error(`${step} ${label} failed: ${JSON.stringify(body)}`);
|
||||
rows.push(
|
||||
`| ${label} | ${step} | ${mb(measured.baseline)} | ${mb(measured.peak)} | ${mb(measured.peak - measured.baseline)} | ${body.ms} | ${summary(step, body.result)} |`,
|
||||
);
|
||||
console.log(rows.at(-1));
|
||||
}
|
||||
}
|
||||
inspector.close();
|
||||
console.log(
|
||||
[
|
||||
"",
|
||||
"| history | step | heap before (MB) | peak heap (MB) | peak above before (MB) | ms | result |",
|
||||
"|---|---|---|---|---|---|---|",
|
||||
...rows,
|
||||
].join("\n"),
|
||||
);
|
||||
} finally {
|
||||
await mf.dispose();
|
||||
}
|
||||
|
||||
/** What a step's answer must offer (Miniflare's own `Response` type, not the DOM one). */
|
||||
type StepResponse = { readonly ok: boolean; json(): Promise<unknown>; text(): Promise<string> };
|
||||
|
||||
type Inspector = { call(method: string): Promise<{ result?: { usedSize?: number } }>; close(): void };
|
||||
|
||||
async function connect(url: string): Promise<Inspector> {
|
||||
const ws = new WebSocket(url);
|
||||
await new Promise((resolve, reject) => {
|
||||
ws.once("open", resolve);
|
||||
ws.once("error", reject);
|
||||
});
|
||||
let id = 0;
|
||||
const pending = new Map<number, (value: never) => void>();
|
||||
ws.on("message", (raw) => {
|
||||
const message = JSON.parse(String(raw)) as { id?: number };
|
||||
if (message.id !== undefined) pending.get(message.id)?.(message as never);
|
||||
pending.delete(message.id ?? -1);
|
||||
});
|
||||
const inspector: Inspector = {
|
||||
call: (method) =>
|
||||
new Promise((resolve) => {
|
||||
const n = ++id;
|
||||
pending.set(n, resolve as (value: never) => void);
|
||||
ws.send(JSON.stringify({ id: n, method }));
|
||||
}),
|
||||
close: () => ws.close(),
|
||||
};
|
||||
await inspector.call("Runtime.enable");
|
||||
return inspector;
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs `step` from a settled heap: a full GC in the isolate first, its used heap then as the baseline, then samples
|
||||
* every `SAMPLE_MS` until the step answers; the peak is the highest sample.
|
||||
*/
|
||||
async function measure(
|
||||
inspector: Inspector,
|
||||
step: () => Promise<StepResponse>,
|
||||
): Promise<{ response: StepResponse; baseline: number; peak: number }> {
|
||||
const used = async () => (await inspector.call("Runtime.getHeapUsage")).result?.usedSize ?? 0;
|
||||
await (await mf.dispatchFetch("http://bench/gc")).text();
|
||||
const baseline = await used();
|
||||
let peak = baseline;
|
||||
let running = true;
|
||||
const sampling = (async () => {
|
||||
while (running) {
|
||||
peak = Math.max(peak, await used());
|
||||
await new Promise((resolve) => setTimeout(resolve, SAMPLE_MS));
|
||||
}
|
||||
})();
|
||||
const response = await step();
|
||||
running = false;
|
||||
await sampling;
|
||||
peak = Math.max(peak, await used());
|
||||
return { response, baseline, peak };
|
||||
}
|
||||
|
||||
function mb(bytes: number): string {
|
||||
return (bytes / MB).toFixed(1);
|
||||
}
|
||||
|
||||
function summary(step: string, result: unknown): string {
|
||||
const value = result as Record<string, unknown>;
|
||||
if (step === "upload") {
|
||||
const stored = value.stored as number;
|
||||
return `${value.slots} slots, ${mb(stored)} MB stored`;
|
||||
}
|
||||
return JSON.stringify(result);
|
||||
}
|
||||
449
packages/chord/test/bench/sync-worker.ts
Normal file
449
packages/chord/test/bench/sync-worker.ts
Normal file
|
|
@ -0,0 +1,449 @@
|
|||
// The Durable Objects of the sync memory benchmark (`sync-memory.ts`), one Chord store each on the object's own SQLite
|
||||
// storage (`chord/sqlite/cloudflare`), as a deployment has them:
|
||||
// - `Origin`: the copy that commits and uploads, and the remote, its slots and bodies in its SQLite (`SqlRemote`);
|
||||
// - `Replica`: a copy restored by a pull, through the origin's remote over RPC;
|
||||
// - `Folder`: folds the origin's history into a snapshot, its scratch store on its own SQLite.
|
||||
// Bodies cross objects as byte streams and are stored in parts, so what an object holds is what the sync layer holds.
|
||||
// The benchmark drives the objects over `fetch` and reads the isolate's heap through the inspector.
|
||||
|
||||
import { DurableObject } from "cloudflare:workers";
|
||||
import { BACKGROUND_CONTEXT as context } from "../../src/context/index.ts";
|
||||
import { int, str } from "../../src/contracts/index.ts";
|
||||
import { DurableObjectSqliteDatabase } from "../../src/sqlite/cloudflare.ts";
|
||||
import { sqliteBackend } from "../../src/store/backend.ts";
|
||||
import type { StagedWrite } from "../../src/store/step.ts";
|
||||
import { type ChordStore, openStore } from "../../src/store/store.ts";
|
||||
import { defineTable } from "../../src/store/table.ts";
|
||||
import { TableChanges } from "../../src/store/table-changes.ts";
|
||||
import { TableHandle } from "../../src/store/table-handle.ts";
|
||||
import { prepareTables } from "../../src/store/table-migration.ts";
|
||||
import {
|
||||
type BatchInfo,
|
||||
foldSession,
|
||||
type JsonObject,
|
||||
type PutBatchResult,
|
||||
pullSession,
|
||||
type RemoteListing,
|
||||
remoteHead,
|
||||
type Seq,
|
||||
type SessionChanges,
|
||||
type SessionSyncMaintenance,
|
||||
type SyncBody,
|
||||
type SyncState,
|
||||
type SyncStateStore,
|
||||
uploadSession,
|
||||
} from "../../src/sync/index.ts";
|
||||
|
||||
// The parts of the Workers runtime this bench uses, typed here: full Workers types would clash with the repository's
|
||||
// Node types, and `DurableObject` (packages/codemode/src/cloudflare-modules.d.ts) leaves `ctx` untyped.
|
||||
type SqlValue = string | number | null | ArrayBuffer | Uint8Array;
|
||||
type SqlCursor<T> = { toArray(): T[]; one(): T };
|
||||
type SqlStorage = { exec<T = Record<string, SqlValue>>(query: string, ...bindings: SqlValue[]): SqlCursor<T> };
|
||||
type ObjectStorage = {
|
||||
readonly sql: SqlStorage;
|
||||
transaction<T>(closure: () => Promise<T>): Promise<T>;
|
||||
transactionSync<T>(closure: () => T): T;
|
||||
};
|
||||
type ObjectState = { readonly storage: ObjectStorage };
|
||||
type ObjectId = unknown;
|
||||
type Namespace<T> = { idFromName(name: string): ObjectId; get(id: ObjectId): T };
|
||||
|
||||
/** A Durable Object of this bench: the runtime's base class with its `ctx` typed. */
|
||||
const BenchObject = DurableObject as unknown as abstract new (
|
||||
ctx: ObjectState,
|
||||
env: Env,
|
||||
) => { readonly ctx: ObjectState; readonly env: Env };
|
||||
|
||||
type Env = {
|
||||
readonly ORIGIN: Namespace<Origin>;
|
||||
readonly REPLICA: Namespace<Replica>;
|
||||
readonly FOLDER: Namespace<Folder>;
|
||||
};
|
||||
|
||||
const Rows = defineTable("rows", { version: 1, columns: { id: str(), n: int(), text: str() }, key: ["id"] });
|
||||
const TABLE = "bench.sync:rows";
|
||||
const OPEN = { admit: () => true, adopt: () => {} };
|
||||
const owner = { label: "bench", scope: {}, assertLive: () => {}, fail: () => {} };
|
||||
/** Where a stored body is split into rows: below the object's 2 MB value limit. */
|
||||
const PART = 512 * 1024;
|
||||
|
||||
/** The isolate's own GC (the benchmark starts workerd with `--expose-gc`), so a step starts from a settled heap. */
|
||||
const collect = (): void => (globalThis as { gc?: () => void }).gc?.();
|
||||
|
||||
/** A body as a byte stream, for RPC. */
|
||||
function toStream(body: SyncBody): ReadableStream<Uint8Array> {
|
||||
const iterator = body[Symbol.asyncIterator]();
|
||||
return new ReadableStream<Uint8Array>({
|
||||
async pull(controller) {
|
||||
const next = await iterator.next();
|
||||
if (next.done) controller.close();
|
||||
else controller.enqueue(next.value);
|
||||
},
|
||||
async cancel() {
|
||||
await iterator.return?.();
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/** A byte stream from RPC as a body. */
|
||||
async function* fromStream(stream: ReadableStream<Uint8Array>): SyncBody {
|
||||
const reader = stream.getReader();
|
||||
try {
|
||||
while (true) {
|
||||
const next = await reader.read();
|
||||
if (next.done) return;
|
||||
yield next.value;
|
||||
}
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
}
|
||||
}
|
||||
|
||||
type SlotRow = { after: number; upto: number; device: string; archive: number; folded: number };
|
||||
|
||||
/**
|
||||
* A remote in an object's SQLite storage, in tables of its own beside the store's: slots, and bodies written and read
|
||||
* back a part at a time as they stream.
|
||||
*/
|
||||
class SqlRemote implements SessionSyncMaintenance {
|
||||
readonly #sql: SqlStorage;
|
||||
|
||||
constructor(sql: SqlStorage) {
|
||||
this.#sql = sql;
|
||||
sql.exec(`CREATE TABLE IF NOT EXISTS r_slots (after INTEGER PRIMARY KEY, upto INTEGER, device TEXT, archive INTEGER, folded INTEGER);
|
||||
CREATE TABLE IF NOT EXISTS r_parts (kind TEXT, at INTEGER, part INTEGER, data BLOB, PRIMARY KEY (kind, at, part));
|
||||
CREATE TABLE IF NOT EXISTS r_snapshots (upto INTEGER PRIMARY KEY);`);
|
||||
}
|
||||
|
||||
async putBatch(
|
||||
_session: string,
|
||||
after: Seq,
|
||||
to: Seq,
|
||||
device: string,
|
||||
body: SyncBody,
|
||||
_context: unknown,
|
||||
options: { readonly archive?: boolean } = {},
|
||||
): Promise<PutBatchResult> {
|
||||
const existing = this.#sql
|
||||
.exec<{ upto: number; device: string }>("SELECT upto, device FROM r_slots WHERE after = ?", after)
|
||||
.toArray()[0];
|
||||
if (existing) return { type: "exists", to: existing.upto, device: existing.device };
|
||||
const head = remoteHead(await this.list());
|
||||
if (after !== head) return { type: "not_head", head };
|
||||
await this.#write("batch", after, body);
|
||||
this.#sql.exec("INSERT INTO r_slots VALUES (?, ?, ?, ?, 0)", after, to, device, options.archive ? 1 : 0);
|
||||
return { type: "created" };
|
||||
}
|
||||
|
||||
async getBatch(_session: string, after: Seq): Promise<SyncBody | undefined> {
|
||||
const slot = this.#sql.exec<{ folded: number }>("SELECT folded FROM r_slots WHERE after = ?", after).toArray()[0];
|
||||
return slot && !slot.folded ? this.#read("batch", after) : undefined;
|
||||
}
|
||||
|
||||
async getSnapshot(_session: string, to: Seq): Promise<SyncBody | undefined> {
|
||||
return this.#sql.exec("SELECT 1 FROM r_snapshots WHERE upto = ?", to).toArray().length > 0
|
||||
? this.#read("snapshot", to)
|
||||
: undefined;
|
||||
}
|
||||
|
||||
async putSnapshot(_session: string, to: Seq, body: SyncBody): Promise<void> {
|
||||
await this.#write("snapshot", to, body);
|
||||
this.#sql.exec("INSERT OR REPLACE INTO r_snapshots VALUES (?)", to);
|
||||
}
|
||||
|
||||
async foldBatches(_session: string, afters: readonly Seq[]): Promise<void> {
|
||||
for (const after of afters) {
|
||||
this.#sql.exec("UPDATE r_slots SET folded = 1 WHERE after = ?", after);
|
||||
this.#sql.exec("DELETE FROM r_parts WHERE kind = 'batch' AND at = ?", after);
|
||||
}
|
||||
}
|
||||
|
||||
async deleteSnapshots(_session: string, snapshots: readonly Seq[]): Promise<void> {
|
||||
for (const to of snapshots) {
|
||||
this.#sql.exec("DELETE FROM r_snapshots WHERE upto = ?", to);
|
||||
this.#sql.exec("DELETE FROM r_parts WHERE kind = 'snapshot' AND at = ?", to);
|
||||
}
|
||||
}
|
||||
|
||||
async list(): Promise<RemoteListing> {
|
||||
const batches = this.#sql
|
||||
.exec<SlotRow>("SELECT after, upto, device, archive, folded FROM r_slots ORDER BY after")
|
||||
.toArray()
|
||||
.map(
|
||||
(slot): BatchInfo => ({
|
||||
after: slot.after,
|
||||
to: slot.upto,
|
||||
device: slot.device,
|
||||
...(slot.archive ? { archive: true as const } : {}),
|
||||
...(slot.folded ? { folded: true as const } : {}),
|
||||
}),
|
||||
);
|
||||
const snapshots = this.#sql
|
||||
.exec<{ upto: number }>("SELECT upto FROM r_snapshots")
|
||||
.toArray()
|
||||
.map((row) => row.upto);
|
||||
return { batches, snapshots };
|
||||
}
|
||||
|
||||
async sessions(): Promise<SessionChanges> {
|
||||
return { sessions: [], cursor: "", more: false };
|
||||
}
|
||||
async putMeta(): Promise<void> {}
|
||||
async getMeta(): Promise<JsonObject | undefined> {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/** Bytes the remote stores, all bodies together. */
|
||||
stored(): number {
|
||||
return this.#sql.exec<{ n: number }>("SELECT coalesce(sum(length(data)), 0) AS n FROM r_parts").one().n;
|
||||
}
|
||||
|
||||
/** The body's chunks gathered into parts of `PART` bytes, one row each, as it streams. */
|
||||
async #write(kind: string, at: number, body: SyncBody): Promise<void> {
|
||||
this.#sql.exec("DELETE FROM r_parts WHERE kind = ? AND at = ?", kind, at);
|
||||
let part = 0;
|
||||
let pending: Uint8Array[] = [];
|
||||
let size = 0;
|
||||
const flush = () => {
|
||||
const data = new Uint8Array(size);
|
||||
let offset = 0;
|
||||
for (const chunk of pending) {
|
||||
data.set(chunk, offset);
|
||||
offset += chunk.length;
|
||||
}
|
||||
this.#sql.exec("INSERT INTO r_parts VALUES (?, ?, ?, ?)", kind, at, part++, data);
|
||||
pending = [];
|
||||
size = 0;
|
||||
};
|
||||
for await (const chunk of body) {
|
||||
pending.push(chunk);
|
||||
size += chunk.length;
|
||||
if (size >= PART) flush();
|
||||
}
|
||||
if (size > 0) flush();
|
||||
}
|
||||
|
||||
/** The body's parts, read back one row at a time. */
|
||||
#read(kind: string, at: number): SyncBody {
|
||||
const sql = this.#sql;
|
||||
return (async function* () {
|
||||
for (let part = 0; ; part++) {
|
||||
const row = sql
|
||||
.exec<{ data: ArrayBuffer }>(
|
||||
"SELECT data FROM r_parts WHERE kind = ? AND at = ? AND part = ?",
|
||||
kind,
|
||||
at,
|
||||
part,
|
||||
)
|
||||
.toArray()[0];
|
||||
if (!row) return;
|
||||
yield new Uint8Array(row.data);
|
||||
}
|
||||
})();
|
||||
}
|
||||
}
|
||||
|
||||
/** The origin's remote seen from another object: its calls over RPC, bodies as byte streams. */
|
||||
class RpcRemote implements SessionSyncMaintenance {
|
||||
readonly #origin: Origin;
|
||||
|
||||
constructor(origin: Origin) {
|
||||
this.#origin = origin;
|
||||
}
|
||||
|
||||
async putBatch(): Promise<PutBatchResult> {
|
||||
throw new Error("The benchmark uploads only at the origin");
|
||||
}
|
||||
async getBatch(_session: string, after: Seq): Promise<SyncBody | undefined> {
|
||||
const stream = await this.#origin.getBatch(after);
|
||||
return stream === null ? undefined : fromStream(stream);
|
||||
}
|
||||
async getSnapshot(_session: string, to: Seq): Promise<SyncBody | undefined> {
|
||||
const stream = await this.#origin.getSnapshot(to);
|
||||
return stream === null ? undefined : fromStream(stream);
|
||||
}
|
||||
async putSnapshot(_session: string, to: Seq, body: SyncBody): Promise<void> {
|
||||
await this.#origin.putSnapshot(to, toStream(body));
|
||||
}
|
||||
async foldBatches(_session: string, afters: readonly Seq[]): Promise<void> {
|
||||
await this.#origin.foldBatches([...afters]);
|
||||
}
|
||||
async deleteSnapshots(_session: string, snapshots: readonly Seq[]): Promise<void> {
|
||||
await this.#origin.deleteSnapshots([...snapshots]);
|
||||
}
|
||||
async list(): Promise<RemoteListing> {
|
||||
return (await this.#origin.list()) as RemoteListing;
|
||||
}
|
||||
async sessions(): Promise<SessionChanges> {
|
||||
return { sessions: [], cursor: "", more: false };
|
||||
}
|
||||
async putMeta(): Promise<void> {}
|
||||
async getMeta(): Promise<JsonObject | undefined> {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
class MemoryState implements SyncStateStore {
|
||||
#state: SyncState | undefined;
|
||||
async get(): Promise<SyncState | undefined> {
|
||||
return this.#state;
|
||||
}
|
||||
async set(_session: string, state: SyncState): Promise<void> {
|
||||
this.#state = state;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* This object's SQLite storage as a store backend. Durable Object SQLite refuses `PRAGMA encoding` and
|
||||
* `pragma_encoding` (`SQLITE_AUTH`), which `openStore` reads for its UTF-8 check; its databases are always UTF-8, so
|
||||
* the benchmark answers that one query itself. (A store bug, to fix in `openStore` or the Cloudflare adapter.)
|
||||
*/
|
||||
function backendOf(storage: ObjectStorage) {
|
||||
const rewrite = (query: string) =>
|
||||
query
|
||||
.replace(/^PRAGMA encoding$/, "SELECT 'UTF-8' AS encoding")
|
||||
.replace(/pragma_encoding e\b/, "(SELECT 'UTF-8' AS encoding) e");
|
||||
const sql = storage.sql;
|
||||
const wrapped = {
|
||||
sql: { exec: (query: string, ...bindings: SqlValue[]) => sql.exec(rewrite(query), ...bindings) },
|
||||
transaction: <T>(closure: () => Promise<T>) => storage.transaction(closure),
|
||||
transactionSync: <T>(closure: () => T) => storage.transactionSync(closure),
|
||||
};
|
||||
return sqliteBackend(new DurableObjectSqliteDatabase(wrapped as never));
|
||||
}
|
||||
|
||||
/** Deterministic text that compresses about as transcript text does (about 2:1), not to nothing. */
|
||||
function text(size: number, seed: number): string {
|
||||
let state = seed * 2654435761 + 1;
|
||||
const words = ["the", "agent", "reads", "a", "file", "and", "writes", "tests", "for", "it", "then", "runs"];
|
||||
let out = "";
|
||||
while (out.length < size) {
|
||||
state ^= state << 13;
|
||||
state ^= state >>> 17;
|
||||
state ^= state << 5;
|
||||
out += `${words[(state >>> 0) % words.length]} ${(state >>> 8) & 0xffff} `;
|
||||
}
|
||||
return out.slice(0, size);
|
||||
}
|
||||
|
||||
export class Origin extends BenchObject {
|
||||
#remote = new SqlRemote(this.ctx.storage.sql);
|
||||
|
||||
/** `count` commits of one row of `size` characters each. */
|
||||
async fill(count: number, size: number): Promise<{ lastSeq: number }> {
|
||||
collect();
|
||||
const store = await openStore(backendOf(this.ctx.storage), context, { outbox: true });
|
||||
try {
|
||||
const chord = store as ChordStore;
|
||||
await prepareTables([{ definition: Rows, table: TABLE }], chord, OPEN, "bench.sync", context);
|
||||
const handle = new TableHandle(Rows, TABLE, chord, owner);
|
||||
for (let index = 0; index < count; index++) {
|
||||
const staged: StagedWrite[] = [];
|
||||
new TableChanges(handle, staged).put({ id: `r${index}`, n: index, text: text(size, index) });
|
||||
await chord.commit(staged, OPEN, context);
|
||||
}
|
||||
return { lastSeq: store.lastSeq };
|
||||
} finally {
|
||||
await store.close(context);
|
||||
}
|
||||
}
|
||||
|
||||
async upload(): Promise<unknown> {
|
||||
collect();
|
||||
const store = await openStore(backendOf(this.ctx.storage), context, { outbox: true });
|
||||
try {
|
||||
const result = await uploadSession({
|
||||
sessionId: "s",
|
||||
device: "origin",
|
||||
source: store,
|
||||
store: this.#remote,
|
||||
state: new MemoryState(),
|
||||
context,
|
||||
});
|
||||
return { result, slots: (await this.#remote.list()).batches.length, stored: this.#remote.stored() };
|
||||
} finally {
|
||||
await store.close(context);
|
||||
}
|
||||
}
|
||||
|
||||
async getBatch(after: number): Promise<ReadableStream<Uint8Array> | null> {
|
||||
const body = await this.#remote.getBatch("s", after);
|
||||
return body === undefined ? null : toStream(body);
|
||||
}
|
||||
async getSnapshot(to: number): Promise<ReadableStream<Uint8Array> | null> {
|
||||
const body = await this.#remote.getSnapshot("s", to);
|
||||
return body === undefined ? null : toStream(body);
|
||||
}
|
||||
async putSnapshot(to: number, stream: ReadableStream<Uint8Array>): Promise<void> {
|
||||
await this.#remote.putSnapshot("s", to, fromStream(stream));
|
||||
}
|
||||
async foldBatches(afters: number[]): Promise<void> {
|
||||
await this.#remote.foldBatches("s", afters);
|
||||
}
|
||||
async deleteSnapshots(snapshots: number[]): Promise<void> {
|
||||
await this.#remote.deleteSnapshots("s", snapshots);
|
||||
}
|
||||
async list(): Promise<RemoteListing> {
|
||||
return this.#remote.list();
|
||||
}
|
||||
}
|
||||
|
||||
export class Replica extends BenchObject {
|
||||
/** Restores the session from the origin's remote into this object's empty storage. */
|
||||
async pull(origin: string): Promise<unknown> {
|
||||
collect();
|
||||
const remote = new RpcRemote(this.env.ORIGIN.get(this.env.ORIGIN.idFromName(origin)));
|
||||
const pulled = await pullSession({
|
||||
sessionId: "s",
|
||||
store: remote,
|
||||
local: { kind: "empty", backend: backendOf(this.ctx.storage) },
|
||||
state: new MemoryState(),
|
||||
context,
|
||||
});
|
||||
await pulled.store.close(context);
|
||||
return { head: pulled.head };
|
||||
}
|
||||
}
|
||||
|
||||
export class Folder extends BenchObject {
|
||||
/** Folds the origin's history into a snapshot, restoring it into this object's empty storage as scratch. */
|
||||
async fold(origin: string): Promise<unknown> {
|
||||
collect();
|
||||
const remote = new RpcRemote(this.env.ORIGIN.get(this.env.ORIGIN.idFromName(origin)));
|
||||
return {
|
||||
folded: await foldSession({ sessionId: "s", store: remote, scratch: backendOf(this.ctx.storage), context }),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
export default {
|
||||
async fetch(request: Request, env: Env): Promise<Response> {
|
||||
const url = new URL(request.url);
|
||||
const run = url.searchParams.get("run") ?? "bench";
|
||||
const count = Number(url.searchParams.get("count") ?? 0);
|
||||
const size = Number(url.searchParams.get("size") ?? 0);
|
||||
const started = Date.now();
|
||||
let result: unknown;
|
||||
switch (url.pathname) {
|
||||
case "/gc":
|
||||
// Every object of the benchmark shares this isolate: one full GC here settles the heap for the next step.
|
||||
collect();
|
||||
return new Response("ok");
|
||||
case "/fill":
|
||||
result = await env.ORIGIN.get(env.ORIGIN.idFromName(run)).fill(count, size);
|
||||
break;
|
||||
case "/upload":
|
||||
result = await env.ORIGIN.get(env.ORIGIN.idFromName(run)).upload();
|
||||
break;
|
||||
case "/pull":
|
||||
result = await env.REPLICA.get(env.REPLICA.idFromName(run)).pull(run);
|
||||
break;
|
||||
case "/fold":
|
||||
result = await env.FOLDER.get(env.FOLDER.idFromName(run)).fold(run);
|
||||
break;
|
||||
default:
|
||||
return new Response("unknown step", { status: 404 });
|
||||
}
|
||||
return Response.json({ ms: Date.now() - started, result });
|
||||
},
|
||||
};
|
||||
|
|
@ -1,16 +1,20 @@
|
|||
// The sync wire format: outbox commits as versioned JSON lines, and snapshots as store archives.
|
||||
// The sync wire format, as streams: outbox commits as versioned, gzip-compressed JSON lines, snapshots as store
|
||||
// archives, and decoding that holds one line at a time.
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { BACKGROUND_CONTEXT as context } from "../../src/context/index.ts";
|
||||
import { createDocument } from "../../src/store/engine.ts";
|
||||
import type { ChordStore } from "../../src/store/store.ts";
|
||||
import {
|
||||
decodeLogLines,
|
||||
decodeBatch,
|
||||
decodeLogLine,
|
||||
decodeSnapshot,
|
||||
encodeBatch,
|
||||
encodeLogLine,
|
||||
encodeSnapshot,
|
||||
LOG_START,
|
||||
LogFormatError,
|
||||
readOutbox,
|
||||
type SyncBody,
|
||||
} from "../../src/sync/index.ts";
|
||||
import { memoryStore } from "../support.ts";
|
||||
|
||||
|
|
@ -36,44 +40,107 @@ async function seeded(): Promise<ChordStore> {
|
|||
return store;
|
||||
}
|
||||
|
||||
describe("log lines", () => {
|
||||
it("round-trips outbox commits", async () => {
|
||||
async function collect<T>(items: AsyncIterable<T>): Promise<T[]> {
|
||||
const out: T[] = [];
|
||||
for await (const item of items) out.push(item);
|
||||
return out;
|
||||
}
|
||||
|
||||
/** A body's bytes, joined. */
|
||||
async function bytesOf(body: SyncBody): Promise<Uint8Array> {
|
||||
const chunks = await collect(body);
|
||||
const out = new Uint8Array(chunks.reduce((size, chunk) => size + chunk.length, 0));
|
||||
let offset = 0;
|
||||
for (const chunk of chunks) {
|
||||
out.set(chunk, offset);
|
||||
offset += chunk.length;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Plain bytes gzip-compressed, as a body: for bodies the encoders would never write. */
|
||||
function gzipped(text: string): SyncBody {
|
||||
return (async function* () {
|
||||
const stream = new Blob([text]).stream().pipeThrough(new CompressionStream("gzip"));
|
||||
yield new Uint8Array(await new Response(stream).arrayBuffer());
|
||||
})();
|
||||
}
|
||||
|
||||
/** `bytes` as a body of chunks of `size`: decoding must not depend on where chunks split. */
|
||||
async function* chunked(bytes: Uint8Array, size: number): SyncBody {
|
||||
for (let offset = 0; offset < bytes.length; offset += size) yield bytes.subarray(offset, offset + size);
|
||||
}
|
||||
|
||||
describe("batch bodies", () => {
|
||||
it("round-trip outbox commits, whatever the chunks", async () => {
|
||||
const store = await seeded();
|
||||
try {
|
||||
const commits = await readOutbox(store, LOG_START, 100, context);
|
||||
expect(commits).toHaveLength(2);
|
||||
expect(decodeLogLines(commits.map(encodeLogLine).join(""))).toEqual({ commits, torn: 0 });
|
||||
const bytes = await bytesOf(encodeBatch(commits));
|
||||
expect(bytes[0]).toBe(0x1f);
|
||||
for (const size of [1, 7, bytes.length]) {
|
||||
expect(await collect(decodeBatch(chunked(bytes, size)))).toEqual(commits);
|
||||
}
|
||||
} finally {
|
||||
await store.close(context);
|
||||
}
|
||||
});
|
||||
|
||||
it("reports a torn last line instead of failing", () => {
|
||||
it("refuse a body that ends inside a line", async () => {
|
||||
const line = encodeLogLine({ seq: 1, nextId: 2, writes: [] });
|
||||
expect(decodeLogLines(line + line.slice(0, 5))).toEqual({
|
||||
commits: [{ seq: 1, nextId: 2, writes: [] }],
|
||||
torn: 5,
|
||||
});
|
||||
await expect(collect(decodeBatch(gzipped(line + line.slice(0, 5))))).rejects.toThrow("ends inside log line 2");
|
||||
});
|
||||
|
||||
it("refuses unknown versions, a missing next id, and sequences that do not increase", () => {
|
||||
expect(() => decodeLogLines('{"v":1,"seq":1,"nextId":2,"writes":[]}\n')).toThrow(LogFormatError);
|
||||
expect(() => decodeLogLines('{"v":2,"seq":1,"writes":[]}\n')).toThrow(LogFormatError);
|
||||
it("refuse sequences that do not increase", async () => {
|
||||
const line = encodeLogLine({ seq: 3, nextId: 4, writes: [] });
|
||||
expect(() => decodeLogLines(line + line)).toThrow("does not increase");
|
||||
await expect(collect(decodeBatch(gzipped(line + line)))).rejects.toThrow("does not increase");
|
||||
});
|
||||
|
||||
it("decode a commit only when it is read", async () => {
|
||||
const lines = [1, 2, 3].map((seq) => encodeLogLine({ seq, nextId: seq + 1, writes: [] })).join("");
|
||||
const commits = decodeBatch(gzipped(`${lines}not json\n`));
|
||||
const iterator = commits[Symbol.asyncIterator]();
|
||||
expect((await iterator.next()).value).toMatchObject({ seq: 1 });
|
||||
await iterator.return?.(undefined);
|
||||
});
|
||||
});
|
||||
|
||||
describe("snapshots", () => {
|
||||
describe("log lines", () => {
|
||||
it("refuse unknown versions and a missing next id", () => {
|
||||
expect(() => decodeLogLine('{"v":1,"seq":1,"nextId":2,"writes":[]}', 1)).toThrow(LogFormatError);
|
||||
expect(() => decodeLogLine('{"v":2,"seq":1,"writes":[]}', 1)).toThrow(LogFormatError);
|
||||
expect(decodeLogLine('{"v":2,"seq":1,"nextId":2,"writes":[]}', 1)).toEqual({ seq: 1, nextId: 2, writes: [] });
|
||||
});
|
||||
});
|
||||
|
||||
describe("snapshot bodies", () => {
|
||||
it("round-trip a store archive", async () => {
|
||||
const store = await seeded();
|
||||
try {
|
||||
const parts: unknown[] = [];
|
||||
for await (const part of store.export(context)) parts.push(part);
|
||||
expect(await decodeSnapshot(await encodeSnapshot(store.export(context)))).toEqual(parts);
|
||||
const parts = await collect(store.export(context));
|
||||
expect(await collect(decodeSnapshot(encodeSnapshot(store.export(context))))).toEqual(parts);
|
||||
expect(parts[0]).toMatchObject({ kind: "manifest", lastSeq: store.lastSeq });
|
||||
} finally {
|
||||
await store.close(context);
|
||||
}
|
||||
});
|
||||
|
||||
it("give the store's commit queue back when the body is not read to the end", async () => {
|
||||
const store = await seeded();
|
||||
try {
|
||||
const body = encodeSnapshot(store.export(context))[Symbol.asyncIterator]();
|
||||
await body.next();
|
||||
await body.return?.();
|
||||
const base = { version: 1, fingerprint: "f1", value: { key: "c" } };
|
||||
const committed = store.commit(
|
||||
[createDocument({ namespace: "chord", owner: "app.sync", kind: "note", key: "c" }, base, "{}")],
|
||||
OPEN,
|
||||
context,
|
||||
);
|
||||
await expect(committed).resolves.toBeGreaterThan(0);
|
||||
} finally {
|
||||
await store.close(context);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1,9 +1,9 @@
|
|||
// `SyncFileSystem` over Node's file system, for the sync tests (a host passes its own environment, such as Pi Durable's
|
||||
// `NodeExecutionEnv`).
|
||||
|
||||
import { mkdir, readdir, readFile, rename, rm, writeFile } from "node:fs/promises";
|
||||
import { appendFile, mkdir, open, readdir, readFile, rename, rm, writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import type { FileResult, SyncFileInfo, SyncFileSystem } from "../../src/sync/index.ts";
|
||||
import type { FileResult, SyncFileInfo, SyncFileReader, SyncFileSystem } from "../../src/sync/index.ts";
|
||||
|
||||
/** An error with a `code`: `not_found` for a missing file or directory, as `SyncFileSystem` reports it. */
|
||||
class FileError extends Error {
|
||||
|
|
@ -27,7 +27,23 @@ export const nodeFiles: SyncFileSystem = {
|
|||
joinPath: async (parts) => ({ ok: true, value: join(...parts) }),
|
||||
readTextFile: (path) => attempt(() => readFile(path, "utf8")),
|
||||
readBinaryFile: (path) => attempt(async () => new Uint8Array(await readFile(path))),
|
||||
openBinaryReader: (path) =>
|
||||
attempt(async (): Promise<SyncFileReader> => {
|
||||
const file = await open(path, "r");
|
||||
return {
|
||||
read: (offset, length) =>
|
||||
attempt(async () => {
|
||||
const buffer = new Uint8Array(length);
|
||||
const { bytesRead } = await file.read(buffer, 0, length, offset);
|
||||
return buffer.subarray(0, bytesRead);
|
||||
}),
|
||||
close: async () => {
|
||||
await file.close();
|
||||
},
|
||||
};
|
||||
}),
|
||||
writeFile: (path, content) => attempt(() => writeFile(path, content)),
|
||||
appendFile: (path, content) => attempt(() => appendFile(path, content)),
|
||||
renameFile: (source, destination) => attempt(() => rename(source, destination)),
|
||||
listDir: (path) =>
|
||||
attempt(async () =>
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ import { tmpdir } from "node:os";
|
|||
import { join } from "node:path";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { BACKGROUND_CONTEXT, type Context } from "../../src/context/index.ts";
|
||||
import { int, str } from "../../src/contracts/index.ts";
|
||||
import { int, opt, str } from "../../src/contracts/index.ts";
|
||||
import { openNodeSqliteDatabase } from "../../src/sqlite/node.ts";
|
||||
import { type Backend, sqliteBackend } from "../../src/store/backend.ts";
|
||||
import type { StagedWrite } from "../../src/store/step.ts";
|
||||
|
|
@ -67,12 +67,16 @@ async function sessionStore(path = ":memory:"): Promise<Store> {
|
|||
}
|
||||
|
||||
/** The table every copy writes: a counter per id. */
|
||||
const Counts = defineTable("counts", { version: 1, columns: { id: str(), n: int() }, key: ["id"] });
|
||||
const Counts = defineTable("counts", {
|
||||
version: 1,
|
||||
columns: { id: str(), n: int(), pad: opt(str()) },
|
||||
key: ["id"],
|
||||
});
|
||||
const TABLE = "app.sync:counts";
|
||||
const OPEN = { admit: () => true, adopt: () => {} };
|
||||
|
||||
/** One batch of the test's history: rows to put. */
|
||||
type Rows = readonly { readonly id: string; readonly n: number }[];
|
||||
type Rows = readonly { readonly id: string; readonly n: number; readonly pad?: string }[];
|
||||
|
||||
/** Commits each batch of rows to `store`, preparing the table first (its creation is a commit of its own). */
|
||||
async function commitAll(store: Store, batches: readonly Rows[]): Promise<void> {
|
||||
|
|
@ -227,9 +231,9 @@ describe("batches", () => {
|
|||
await commitAll(store, seedBatches());
|
||||
const commits = await readOutbox(store, LOG_START, 1000, context);
|
||||
expect(commits.length).toBeGreaterThan(0);
|
||||
const bytes = await encodeBatch(commits);
|
||||
expect(bytes[0]).toBe(0x1f);
|
||||
expect(await decodeBatch(bytes)).toEqual(commits);
|
||||
const decoded = [];
|
||||
for await (const commit of decodeBatch(encodeBatch(commits))) decoded.push(commit);
|
||||
expect(decoded).toEqual(commits);
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -582,6 +586,61 @@ describe("syncing a session between devices", () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe("bounded memory", () => {
|
||||
/** Rows of `size` characters each, one per commit: about `count * size` bytes of history. */
|
||||
function bigBatches(count: number, size: number): Rows[] {
|
||||
return Array.from({ length: count }, (_, index) => [{ id: `big${index}`, n: index, pad: "x".repeat(size) }]);
|
||||
}
|
||||
|
||||
it("ends a batch at BATCH_BYTES, so a large history uploads in several slots", async () => {
|
||||
const store = await remote();
|
||||
const laptop = device("laptop", store);
|
||||
// 24 commits of ~1 MB: about three batches of BATCH_BYTES (8 MB).
|
||||
await laptop.commit(bigBatches(24, 1024 * 1024));
|
||||
expect(await laptop.upload()).toEqual({ type: "uploaded", head: (await laptop.store()).lastSeq });
|
||||
const slots = (await store.list(SESSION, context)).batches;
|
||||
expect(slots.length).toBeGreaterThanOrEqual(3);
|
||||
for (const [index, slot] of slots.entries()) if (index > 0) expect(slot.after).toBe(slots[index - 1]!.to);
|
||||
const replica = device("replica", store);
|
||||
await replica.pull();
|
||||
expect(await reads(await replica.store())).toEqual(await reads(await laptop.store()));
|
||||
});
|
||||
|
||||
it("applies a large batch in chunks of APPLY_BYTES, and a pull stopped between them resumes", async () => {
|
||||
const store = await remote();
|
||||
const laptop = device("laptop", store);
|
||||
// 6 commits of ~1 MB in one batch: about two apply chunks of APPLY_BYTES (4 MB).
|
||||
await laptop.commit(bigBatches(6, 1024 * 1024));
|
||||
await laptop.upload();
|
||||
expect((await store.list(SESSION, context)).batches).toHaveLength(1);
|
||||
const viewer = device("viewer", store);
|
||||
const local = tracked(await sessionStore());
|
||||
const chord = local as ChordStore;
|
||||
let applied = 0;
|
||||
const apply = chord.replay.bind(chord);
|
||||
chord.replay = ((...args: Parameters<ChordStore["replay"]>) => {
|
||||
if (++applied === 2) return Promise.reject(new Error("power cut"));
|
||||
return apply(...args);
|
||||
}) as ChordStore["replay"];
|
||||
const pull = () =>
|
||||
pullSession({
|
||||
sessionId: SESSION,
|
||||
store,
|
||||
local: { kind: "store", store: local },
|
||||
state: viewer.state,
|
||||
context,
|
||||
});
|
||||
await expect(pull()).rejects.toThrow("power cut");
|
||||
const midway = local.lastSeq;
|
||||
expect(midway).toBeGreaterThan(0);
|
||||
expect(midway).toBeLessThan((await laptop.store()).lastSeq);
|
||||
await pull();
|
||||
// The first chunk was applied, the second refused, then the resumed pull applied the rest in more chunks.
|
||||
expect(applied).toBeGreaterThanOrEqual(3);
|
||||
expect(await reads(local)).toEqual(await reads(await laptop.store()));
|
||||
});
|
||||
});
|
||||
|
||||
describe("host pieces", () => {
|
||||
it("lists sessions and keeps the last metadata written", async () => {
|
||||
const store = await remote();
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue