From c0156e1c3dfa0b660c351e25415a3dc1c17fdb8f Mon Sep 17 00:00:00 2001 From: Travis Rich Date: Tue, 20 Oct 2026 22:36:31 -0400 Subject: [PATCH] Layout F: convert a collection from today's layout A layout.convert job per collection walks its versions in order, from a cursor (version, record tree, key) within a budget, and writes the records each version's new leaves hold into the collection's F packs and index, as stored. Nodes don't change, so version hashes don't. A record is written once however many versions hold it, and a job run again writes nothing twice (convertTree writes only what KeyedLayout.missing finds absent). Versions committed while it runs are converted in turn; the merges of its index run as it goes (layout.maintain with layout 'f'). Once the head is converted it switches the collection to F in one statement, only if the head is still the last version converted and no push is committing. A commit reads the layout when it starts and keeps it (a parallel commit in its plan, which its units and assembly open), and publishes only if the collection is still in it (publishVersion's recordLayout); otherwise it fails with 503, push again. Commits wait 90 s after the switch, as isolates note a collection's layout for a minute. POST /api/admin/convert {collection} starts it; GET ?collection= its state. --- packages/protocol/src/index.ts | 7 + packages/protocol/src/repo/convert.ts | 126 +++++++++ packages/protocol/src/repo/layouts/keyed.ts | 11 + packages/server/src/api/admin.ts | 24 ++ packages/server/src/handlers.ts | 15 +- packages/server/src/jobs.ts | 2 + packages/server/src/layouts/convert.ts | 206 +++++++++++++++ packages/server/src/layouts/errors.ts | 20 ++ packages/server/src/ports.ts | 7 + packages/server/src/push/delta.ts | 6 + packages/server/src/push/outcome.ts | 4 + packages/server/src/push/parallel.ts | 24 +- packages/server/src/stores.ts | 55 ++-- packages/server/src/versions/commit.ts | 22 +- packages/server/src/versions/fork.ts | 1 + packages/server/src/versions/publish.ts | 22 +- packages/server/test/commit.test.ts | 1 + packages/server/test/convert.test.ts | 279 ++++++++++++++++++++ 18 files changed, 797 insertions(+), 35 deletions(-) create mode 100644 packages/protocol/src/repo/convert.ts create mode 100644 packages/server/src/layouts/convert.ts create mode 100644 packages/server/src/layouts/errors.ts create mode 100644 packages/server/test/convert.test.ts diff --git a/packages/protocol/src/index.ts b/packages/protocol/src/index.ts index 6e622013..ac6ed28e 100644 --- a/packages/protocol/src/index.ts +++ b/packages/protocol/src/index.ts @@ -32,6 +32,13 @@ export { type TypedLeaf, } from './repo/layouts/index.js' export { LiveHashes, markRange, recordRoots, type TypedRoot } from './repo/layouts/reclaim.js' +export { + type ConvertBudget, + type ConvertProgress, + type ConvertTree, + convertTree, + versionTrees, +} from './repo/convert.js' export { fsck, type FsckCursor, diff --git a/packages/protocol/src/repo/convert.ts b/packages/protocol/src/repo/convert.ts new file mode 100644 index 00000000..0a458ed3 --- /dev/null +++ b/packages/protocol/src/repo/convert.ts @@ -0,0 +1,126 @@ +import { type RecordEntry, recordTree } from '../tree/node.js' +import { newNodes } from '../tree/read.js' +/** + * Converting a collection from today's layout to layout F (planning: layout-f-pr.md, + * "PR 2"). A version's record trees are walked against the version before's; each + * leaf it adds is read from its body (today's layout) and the records F's index + * doesn't hold yet are written to the set's packs, as stored (an out-of-line + * record stays a pointer). Nodes don't change, so neither do roots or version + * hashes. A record is written once however many versions hold it, and a walk that + * runs again (a retried job) writes nothing twice. + */ +import { compareUtf8 } from '../utf8.js' +import { KeyedLayout, type RecordSet } from './layouts/index.js' +import type { Repo } from './repo.js' +import { RepoSource } from './repo.js' + +/** One record tree of a version: a set's tree of a type, and the version before's. */ +export interface ConvertTree { + set: RecordSet + type: string + base: string | null + root: string | null +} + +export interface ConvertBudget { + /** Leaves to read at most before stopping (one at least). */ + leaves: number + /** Stop past this time (ms since the epoch). */ + until: number +} + +export interface ConvertProgress { + /** The last key of the last leaf done when the budget ran out; null when the tree is done. */ + after: string | null + leaves: number + records: number + /** Records the index already held (an earlier version's, or a job run again). */ + skipped: number +} + +/** + * Write the records of `tree`'s new leaves (from `from`, today's layout) into + * `to`'s set (layout F), from `after` on, within `budget`. + */ +export async function convertTree( + from: Repo, + to: Repo, + tree: ConvertTree, + after: string | null, + budget: ConvertBudget, +): Promise { + const store = to.records?.[tree.set] + if (!(store instanceof KeyedLayout)) throw new Error('convertTree: the target is not on layout F') + if (from.records) throw new Error("convertTree: the source is not on today's layout") + const out: ConvertProgress = { after: null, leaves: 0, records: 0, skipped: 0 } + if (tree.root === null) return out + const source = new RepoSource(recordTree, from) + const writer = store.writer() + try { + for await (const n of newNodes(source, tree.base, tree.root, { after })) { + if (n.level !== 0) continue + const node = await from.decoded(recordTree, n.hash) + if (node.kind !== 'leaf') continue + const entries = node.entries as RecordEntry[] + const missing = new Set(await store.missing(tree.type, entries)) + out.skipped += entries.length - missing.size + if (missing.size) { + // The stored body, streamed (an out-of-line record's pointer stays a pointer). + let i = 0 + for await (const line of from.bodyLineStream({ hash: n.hash, entries }, undefined, { + pointers: true, + })) { + if (missing.has(i)) { + writer.add(tree.type, entries[i]!, line) + out.records++ + } + if (++i % 1024 === 0) await writer.drain?.() + } + if (i !== entries.length) + throw new Error( + `convertTree: leaf ${n.hash} has ${i} lines for ${entries.length} entries`, + ) + await writer.drain?.() + } + out.leaves++ + if (out.leaves >= budget.leaves || Date.now() >= budget.until) { + out.after = n.lastKey + break + } + } + } finally { + await writer.finish() + } + // A leaf past the cursor's last key can't come before it. + if (out.after !== null && after !== null && compareUtf8(out.after, after) <= 0) + throw new Error('convertTree: the walk went backwards') + return out +} + +/** + * A version's record trees in a fixed order (public then private, types by slug), + * each with the same set's tree of the type in the version before (`prev`, null + * for the first): what `convertTree` walks. + */ +export async function versionTrees( + repo: Repo, + version: string, + prev: string | null, +): Promise { + const setsOf = async (hash: string | null) => { + if (hash === null) + return { public: {}, private: {} } as Record> + const root = await repo.root(hash) + const priv = root.private ? await repo.privateSet(root.private) : null + const roots = (types: Record | undefined) => + Object.fromEntries(Object.entries(types ?? {}).map(([slug, t]) => [slug, t.root])) + return { public: roots(root.public.types), private: roots(priv?.types) } + } + const now = await setsOf(version) + const before = await setsOf(prev) + const out: ConvertTree[] = [] + for (const set of ['public', 'private'] as const) + for (const type of Object.keys(now[set]).sort(compareUtf8)) + out.push({ set, type, base: before[set][type] ?? null, root: now[set][type] ?? null }) + return out +} diff --git a/packages/protocol/src/repo/layouts/keyed.ts b/packages/protocol/src/repo/layouts/keyed.ts index 85455784..e3e010b1 100644 --- a/packages/protocol/src/repo/layouts/keyed.ts +++ b/packages/protocol/src/repo/layouts/keyed.ts @@ -238,6 +238,17 @@ export class KeyedLayout implements RecordStore { return keys.map((k) => found.get(k)) } + /** + * Which of `entries` (of type `type`) the index doesn't hold, by index into + * `entries`, from the newest manifest: a conversion writes only these, so a job + * that runs again writes nothing twice. + */ + async missing(type: string, entries: readonly RecordEntry[]): Promise { + if (!entries.length) return [] + const found = await this.#find(await this.manifests.latest(), type, entries) + return found.flatMap((x, i) => (x === undefined ? [i] : [])) + } + async leafLines(leaf: TypedLeaf): Promise { if (!leaf.entries.length) return [] try { diff --git a/packages/server/src/api/admin.ts b/packages/server/src/api/admin.ts index aeae199e..77ab4b70 100644 --- a/packages/server/src/api/admin.ts +++ b/packages/server/src/api/admin.ts @@ -17,6 +17,8 @@ * POST /api/admin/fsck stewards: {collection, fileBytes?} checks its repository * GET /api/admin/fsck?collection= stewards: the last check's report * POST /api/admin/usage/rebuild stewards: {day} recomputes them from the usage log + * POST /api/admin/convert stewards: {collection} converts it to layout F + * GET /api/admin/convert?collection= stewards: the conversion's state * * A steward is a KF Auth user whose role is 'admin', read fresh on each check. */ @@ -27,6 +29,7 @@ import type { AppEnv } from '../app.js' import { usageFor } from '../billing/usage.js' import { chunks } from '../db/chunks.js' import * as schema from '../db/schema.js' +import { conversionState, startConversion } from '../layouts/convert.js' import { forgetDenylist } from '../lib/limits.js' import { jsonError } from './access.js' @@ -252,6 +255,27 @@ export function adminRoutes() { return c.json({ ok: true, queued: true }, 202) }) + app.post('/api/admin/convert', async (c) => { + const denied = await stewardOnly(c) + if (denied) return denied + const b = (await c.req.json().catch(() => null)) as { collection?: unknown } | null + const col = await collectionBySlugs(c, b?.collection) + if (!col) return jsonError(c, 404, 'Collection not found (send {"collection": "owner/slug"})') + if (col.recordLayout !== 'today') + return jsonError(c, 409, `Collection is already on layout ${col.recordLayout}`) + return c.json(await startConversion(c.var.ports, col.id), 202) + }) + + app.get('/api/admin/convert', async (c) => { + const denied = await stewardOnly(c) + if (denied) return denied + const col = await collectionBySlugs(c, c.req.query('collection')) + if (!col) return jsonError(c, 404, 'Collection not found (?collection=owner/slug)') + const state = await conversionState(c.var.ports, col.id) + if (!state) return jsonError(c, 404, 'Not converted') + return c.json({ ...state, recordLayout: col.recordLayout }) + }) + app.get('/api/admin/fsck', async (c) => { const denied = await stewardOnly(c) if (denied) return denied diff --git a/packages/server/src/handlers.ts b/packages/server/src/handlers.ts index 1c20c0b6..a55f8a07 100644 --- a/packages/server/src/handlers.ts +++ b/packages/server/src/handlers.ts @@ -30,6 +30,7 @@ import './files/files.js' import { reconcileDue } from './billing/reconcile.js' import { cleanupTick } from './cleanup/runs.js' import './cleanup/layout-reclaim.js' +import './layouts/convert.js' import * as schema from './db/schema.js' import './push/compact.js' import './push/parallel.js' @@ -66,9 +67,17 @@ registerJob('version.published', async (job, ports) => { } }) -/** A job's record set, and the collection's layout F store for it (null: not on F). */ +/** + * A job's record set, and the collection's layout F store for it (null: not on F). + * `layout: 'f'` opens it on F whatever the collection's layout: a conversion's + * merges run while the collection is still on today's (layouts/convert.ts). + */ async function setLayout(ports: Ports, job: JobMessage): Promise { - const repo = await ports.stores.forCollection(String(job.collectionId)) + const id = String(job.collectionId) + const repo = + job.layout === 'f' + ? await ports.stores.forCollectionAs(id, 'f') + : await ports.stores.forCollection(id) const records = repo.records?.[job.set === 'private' ? 'private' : 'public'] return records instanceof KeyedLayout ? records : null } @@ -86,6 +95,7 @@ registerJob('layout.maintain', async (job, ports) => { set: String(job.set ?? 'public'), plan: plan.id, range: i, + ...(job.layout === 'f' ? { layout: 'f' } : {}), })), ), }) @@ -96,6 +106,7 @@ registerJob('layout.maintain', async (job, ports) => { type: 'layout.maintain', collectionId: String(job.collectionId), set: String(job.set ?? 'public'), + ...(job.layout === 'f' ? { layout: 'f' } : {}), }) }) diff --git a/packages/server/src/jobs.ts b/packages/server/src/jobs.ts index 53947425..8ea61d2e 100644 --- a/packages/server/src/jobs.ts +++ b/packages/server/src/jobs.ts @@ -55,6 +55,8 @@ export const BULK_JOBS: ReadonlySet = new Set([ // Its reclaim, a key range a job (cleanup/layout-reclaim.ts), and merges too large for one job. 'layout.reclaim', 'layout.merge', + // Converting a collection from today's layout to F (layouts/convert.ts). + 'layout.convert', ]) export const isBulk = (type: string) => BULK_JOBS.has(type) diff --git a/packages/server/src/layouts/convert.ts b/packages/server/src/layouts/convert.ts new file mode 100644 index 00000000..f1f133b8 --- /dev/null +++ b/packages/server/src/layouts/convert.ts @@ -0,0 +1,206 @@ +/** + * Converting a collection from today's record layout to layout F (planning: + * layout-f-pr.md, "PR 2"), as bounded jobs: + * + * layout.convert from a cursor (a version, one of its record trees, a key in it), + * write the records each version's new leaves hold into the + * collection's F packs and index, until a budget runs out; queue + * the next. Once the head is converted and no push is committing, + * switch the collection to F. + * + * Nodes don't change, so neither do roots or version hashes: only where records live. + * Until the switch the collection reads and commits in today's layout as before, and + * versions committed meanwhile are converted in turn. A record is written once however + * many versions hold it, and a job that runs again writes nothing twice + * (`convertTree`). + * + * The switch is one statement: the layout moves to `f` only if the head is the last + * version converted and no push of the collection is committing. A commit reads the + * layout when it starts and keeps it (a parallel one in its plan), and publishes only + * if the collection is still in it (`publishVersion`), so no version is published in + * the layout it isn't in. Isolates note a collection's layout for a minute + * (`stores.ts`), so for a while after the switch commits wait (LayoutSwitchingError): + * no version only F can read appears while an isolate may still read in today's. + * + * Old leaf bodies stay where they are: nothing reads them once the collection is on + * F, and the storage cleanup's mark still covers them (a follow-up). Mirrors of the + * collection get F's files from the next version they copy. + */ +import { convertTree, type LayoutName, versionTrees } from '@underlay/protocol' +import { and, asc, eq, gt, sql } from 'drizzle-orm' + +import * as schema from '../db/schema.js' +import { registerJob } from '../jobs.js' +import type { Ports } from '../ports.js' +import { LayoutSwitchingError } from './errors.js' + +export const convertConfig = { + /** Leaves one job reads at most. */ + leavesPerJob: 2000, + /** How long one job converts before it queues the rest. */ + budgetMs: 4 * 60_000, + /** Commits wait this long after the switch: stores note a layout for a minute. */ + switchPauseMs: 90_000, + /** Seconds before a job that couldn't switch (a push committing) tries again. */ + retrySeconds: 30, +} + +export interface ConvertState { + collectionId: string + status: 'converting' | 'switched' + startedAt: string + /** The job step this state is at: a job of any other step is a duplicate. */ + step: number + /** The last version fully converted (0: none), and its id and hash. */ + seq: number + versionId: string | null + versionHash: string | null + /** In the next version: the record tree being converted (versionTrees' order), and the key it's done through. */ + tree: number + after: string | null + stats: { versions: number; leaves: number; records: number; skipped: number; jobs: number } + switchedAt?: string +} + +const stateKey = (collectionId: string) => `convert/${collectionId}.json` + +export async function conversionState( + ports: Ports, + collectionId: string, +): Promise { + const o = await ports.stores.internal.get(stateKey(collectionId)) + return o ? (JSON.parse(await o.text()) as ConvertState) : null +} + +const save = (ports: Ports, s: ConvertState) => + ports.stores.internal.put(stateKey(s.collectionId), JSON.stringify(s), { + contentType: 'application/json', + }) + +/** Start converting a collection on today's layout (a no-op while one is under way). */ +export async function startConversion(ports: Ports, collectionId: string): Promise { + const [c] = await ports.db + .select({ layout: schema.collections.recordLayout }) + .from(schema.collections) + .where(eq(schema.collections.id, collectionId)) + if (!c) throw new Error(`Collection ${collectionId} not found`) + const known = await conversionState(ports, collectionId) + if (known?.status === 'converting') return known + if (c.layout !== 'today') + throw new Error(`Collection ${collectionId} is already on layout ${c.layout}`) + const state: ConvertState = { + collectionId, + status: 'converting', + startedAt: new Date().toISOString(), + step: 0, + seq: 0, + versionId: null, + versionHash: null, + tree: 0, + after: null, + stats: { versions: 0, leaves: 0, records: 0, skipped: 0, jobs: 0 }, + } + await save(ports, state) + await ports.jobs.enqueue({ type: 'layout.convert', collectionId, step: 0 }) + return state +} + +/** + * Throws LayoutSwitchingError when a commit to a collection on `layout` must wait: + * it was switched to F under switchPauseMs ago. + */ +export async function checkSwitching( + ports: Ports, + collectionId: string, + layout: LayoutName, +): Promise { + if (layout !== 'f') return + const s = await conversionState(ports, collectionId) + if (s?.switchedAt && Date.now() - Date.parse(s.switchedAt) < convertConfig.switchPauseMs) + throw new LayoutSwitchingError() +} + +/** Queue the merges of the collection's F index (it isn't on F yet: `layout: 'f'`). */ +const queueMaintain = (ports: Ports, collectionId: string) => + ports.jobs.enqueueBatch( + (['public', 'private'] as const).map((set) => ({ + type: 'layout.maintain', + collectionId, + set, + layout: 'f', + })), + ) + +registerJob('layout.convert', async (job, ports) => { + const collectionId = String(job.collectionId) + const state = await conversionState(ports, collectionId) + if (state?.status !== 'converting' || state.step !== Number(job.step)) return + const from = await ports.stores.forCollectionAs(collectionId, 'today') + const to = await ports.stores.forCollectionAs(collectionId, 'f') + const until = Date.now() + convertConfig.budgetMs + let leaves = 0 + const next = async (delaySeconds?: number) => { + state.step++ + state.stats.jobs++ + await save(ports, state) + await queueMaintain(ports, collectionId) + await ports.jobs.enqueue( + { type: 'layout.convert', collectionId, step: state.step }, + delaySeconds ? { delaySeconds } : undefined, + ) + } + + const versions = await ports.db + .select({ id: schema.versions.id, seq: schema.versions.seq, hash: schema.versions.hash }) + .from(schema.versions) + .where(and(eq(schema.versions.collectionId, collectionId), gt(schema.versions.seq, state.seq))) + .orderBy(asc(schema.versions.seq)) + .limit(100) + for (const v of versions) { + const trees = await versionTrees(from, v.hash, state.versionHash) + for (; state.tree < trees.length; state.tree++) { + const p = await convertTree(from, to, trees[state.tree]!, state.after, { + leaves: Math.max(1, convertConfig.leavesPerJob - leaves), + until, + }) + leaves += p.leaves + state.stats.leaves += p.leaves + state.stats.records += p.records + state.stats.skipped += p.skipped + state.after = p.after + // Stopped inside the tree: the next job carries on from its key. + if (p.after !== null) return next() + } + state.seq = v.seq + state.versionId = v.id + state.versionHash = v.hash + state.tree = 0 + state.after = null + state.stats.versions++ + if (leaves >= convertConfig.leavesPerJob || Date.now() >= until) return next() + } + if (versions.length === 100) return next() + + // Every version is converted: switch, if the head is still the last of them and no + // push is committing (one statement, so neither can change in between). + const switched = await ports.db + .update(schema.collections) + .set({ recordLayout: 'f' }) + .where( + and( + eq(schema.collections.id, collectionId), + eq(schema.collections.recordLayout, 'today'), + sql`${schema.collections.headVersionId} IS ${state.versionId}`, + sql`NOT EXISTS (SELECT 1 FROM ${schema.pushSessions} WHERE ${schema.pushSessions.collectionId} = ${collectionId} AND ${schema.pushSessions.status} = 'committing')`, + ), + ) + .returning({ id: schema.collections.id }) + if (!switched.length) return next(convertConfig.retrySeconds) + ports.stores.forget(collectionId) + state.status = 'switched' + state.switchedAt = new Date().toISOString() + state.step++ + await save(ports, state) + await queueMaintain(ports, collectionId) + console.log(`[convert] ${collectionId} switched to layout F: ${JSON.stringify(state.stats)}`) +}) diff --git a/packages/server/src/layouts/errors.ts b/packages/server/src/layouts/errors.ts new file mode 100644 index 00000000..7a872931 --- /dev/null +++ b/packages/server/src/layouts/errors.ts @@ -0,0 +1,20 @@ +/** Errors of a collection's record layout changing under a commit (layouts/convert.ts). */ + +/** The collection's layout changed while this commit wrote in the old one: push again. */ +export class LayoutChangedError extends Error { + constructor() { + super("The collection's storage layout changed while this push was committing") + this.name = 'LayoutChangedError' + } +} + +/** + * The collection was just switched to layout F: commits wait a minute, until no + * isolate still reads it in the old layout (stores note a layout for a minute). + */ +export class LayoutSwitchingError extends Error { + constructor() { + super("The collection's storage layout is switching; push again in a minute") + this.name = 'LayoutSwitchingError' + } +} diff --git a/packages/server/src/ports.ts b/packages/server/src/ports.ts index d2165eb3..9d4e8980 100644 --- a/packages/server/src/ports.ts +++ b/packages/server/src/ports.ts @@ -41,6 +41,13 @@ export interface Jobs { export interface Stores { /** The repository of a collection's primary placement, in the collection's record layout. */ forCollection(collectionId: string): Promise + /** + * The same placement in a given layout, whatever the collection's: a conversion + * reads today's and writes F's, and a commit keeps the layout it planned in. + */ + forCollectionAs(collectionId: string, layout: LayoutName): Promise + /** Forget this isolate's note of a collection's placement and layout (it changed). */ + forget(collectionId: string): void /** The record layout a new collection gets (a fork takes its parent's). */ newCollections: LayoutName /** The repository at a storage location. */ diff --git a/packages/server/src/push/delta.ts b/packages/server/src/push/delta.ts index ae7107be..b220127c 100644 --- a/packages/server/src/push/delta.ts +++ b/packages/server/src/push/delta.ts @@ -26,6 +26,7 @@ import { import { eq } from 'drizzle-orm' import * as schema from '../db/schema.js' +import { checkSwitching } from '../layouts/convert.js' import type { Ports } from '../ports.js' import { type BaseVersion, commitVersion, type TypeInput } from '../versions/commit.js' import { deltaChanges, isPrivateSchema, stageOutOfLine } from './changes.js' @@ -342,7 +343,11 @@ export async function commitDeltaSession( const schemaErr = schemaSetError(inputs.schemas) if (schemaErr) return { status: 'schema_refused', error: schemaErr } const runs = await sessionRuns(ports, session.id) + // The layout as the database has it now, not as this isolate last noted it (a + // conversion may have just switched it): the commit keeps it to the end. + ports.stores.forget(session.collectionId) const repo = await ports.stores.forCollection(session.collectionId) + await checkSwitching(ports, session.collectionId, repo.layout) const root = base ? await repo.root(base.hash) : null const priv = root?.private ? await repo.privateSet(root.private) : null const sets = { pub: root?.public ?? emptySet(), priv: priv ?? emptySet() } @@ -356,6 +361,7 @@ export async function commitDeltaSession( } return commitVersion(ports, { collectionId: session.collectionId, + recordLayout: repo.layout, fence, base, types: await typeInputsForDelta(ports, session, inputs, runs, { diff --git a/packages/server/src/push/outcome.ts b/packages/server/src/push/outcome.ts index 6fc72aaf..ae333baf 100644 --- a/packages/server/src/push/outcome.ts +++ b/packages/server/src/push/outcome.ts @@ -9,6 +9,7 @@ import { eq } from 'drizzle-orm' import { FenceError, StorageBusyError } from '../cleanup/fence.js' import * as schema from '../db/schema.js' +import { LayoutChangedError, LayoutSwitchingError } from '../layouts/errors.js' import type { Ports } from '../ports.js' import type { CommitResult } from '../versions/commit.js' import { SESSION_TTL_MS, transition } from './session.js' @@ -53,6 +54,9 @@ export async function commitOutcome( }, } } + if (err instanceof LayoutChangedError || err instanceof LayoutSwitchingError) { + return { status: 503, body: { error: `${err.message}. Push again.`, statusCode: 503 } } + } if (!(err instanceof MissingFilesError)) throw err return { status: 422, diff --git a/packages/server/src/push/parallel.ts b/packages/server/src/push/parallel.ts index 71317af3..73f2b896 100644 --- a/packages/server/src/push/parallel.ts +++ b/packages/server/src/push/parallel.ts @@ -38,6 +38,7 @@ import { bodyOfRecord, type Change, type DecodedNode, + type LayoutName, compileSchema, boundaryBytes, compareUtf8, @@ -153,6 +154,12 @@ interface UnitSpec { interface CommitPlan { id: string + /** + * The record layout the commit writes in, read when it was planned: its units and + * assembly write in it whatever their isolates last noted, and it publishes only + * if the collection is still in it. Absent in plans made before conversions. + */ + layout?: LayoutName /** * The storage fence epoch read before the units wrote anything. Assembly * publishes under it, so a deletion window during the commit fails the push @@ -250,6 +257,12 @@ export function baseLevelFor(count: number): number { return level } +/** The collection's repository in the layout its commit was planned in. */ +const planRepo = (ports: Ports, collectionId: string, plan: CommitPlan) => + plan.layout + ? ports.stores.forCollectionAs(collectionId, plan.layout) + : ports.stores.forCollection(collectionId) + /** The base tree's nodes at `level` (or its root, when that's lower), in order. */ async function baseNodes(repo: Repo, root: string | null, level: number): Promise { if (root === null) return [] @@ -557,6 +570,7 @@ export async function planParallel( const dir = planDir(session.id, planId) const plan: CommitPlan = { id: planId, + layout: ctx.repo.layout, base: ctx.base, fence: ctx.fence, types: {}, @@ -903,7 +917,7 @@ async function runPendingUnit(ports: Ports, unit: UnitRow, session: SessionRow): const runs = await unitRuns(ports, unit.slicesKey!) const t = plan.types[unit.type]! const set = unit.set as SetName - const repo = await ports.stores.forCollection(session.collectionId) + const repo = await planRepo(ports, session.collectionId, plan) // Writes in flight hold compressed leaves; halved from the default so two units fit an isolate. const sink = new RepoSink(repo, { bodyOf: bodyOfRecord, @@ -1137,7 +1151,7 @@ export async function assembleParallel(ports: Ports, sessionId: string): Promise } const plan = await getJson(ports, `${planDir(sessionId, planId)}/plan.json`) - const repo = await ports.stores.forCollection(session.collectionId) + const repo = await planRepo(ports, session.collectionId, plan) // Base leaves keep large records as pointers, as in units (runPendingUnit). A source // a tree and a sink a set (told each tree's type): layout F reads and writes each // set's records apart, and finds them by type. @@ -1294,6 +1308,7 @@ async function publish(ports: Ports, session: SessionRow, plan: CommitPlan, a: A const outcome = await commitOutcome(ports, session.id, () => commitVersion(ports, { collectionId: session.collectionId, + ...(plan.layout ? { recordLayout: plan.layout } : {}), // Plans written before the fence existed fail if any window has opened since. fence: plan.fence ?? 0, base: plan.base, @@ -1354,7 +1369,7 @@ async function startValidation( plan: CommitPlan, built: Assembled['built'], ) { - const repo = await ports.stores.forCollection(session.collectionId) + const repo = await planRepo(ports, session.collectionId, plan) const tasks: ValidationTask[] = [] for (const [type, t] of Object.entries(plan.types)) { if (!t.validate) continue @@ -1457,7 +1472,8 @@ export async function runValidation( const t = tasks[task]! const inputs = await loadInputs(ports, sessionId) const validate = compileSchema(inputs.schemas[t.type]!) - const repo = await ports.stores.forCollection(session.collectionId) + const plan = await getJson(ports, `${planDir(sessionId, planId)}/plan.json`) + const repo = await planRepo(ports, session.collectionId, plan) const source = new RepoSource(recordTree, repo, { set: t.set, type: t.type }) const result: ValidationResult = { total: 0, errors: [] } for await (const e of entriesInRange(source, t.root, t.after, t.through)) { diff --git a/packages/server/src/stores.ts b/packages/server/src/stores.ts index daf13401..dc3b7f8e 100644 --- a/packages/server/src/stores.ts +++ b/packages/server/src/stores.ts @@ -114,34 +114,41 @@ export function createStores(db: Db, cache: Cache, platform: PlatformStorage): S const located = (locationId: string, collectionId: string, layout: LayoutName) => layout === 'f' ? forCollectionRuns(locationId, collectionId) : forLocation(locationId) + const primary = async (collectionId: string) => { + const hit = primaryCache.get(collectionId) + if (hit && Date.now() - hit.at < PRIMARY_TTL_MS) return hit + const [row] = await db + .select({ + locationId: schema.placements.locationId, + layout: schema.collections.recordLayout, + }) + .from(schema.placements) + .innerJoin(schema.collections, eq(schema.collections.id, schema.placements.collectionId)) + .where( + and( + eq(schema.placements.collectionId, collectionId), + eq(schema.placements.role, 'primary'), + ), + ) + .limit(1) + if (!row) throw new Error(`Collection ${collectionId} has no primary placement`) + const found = { locationId: row.locationId, layout: row.layout, at: Date.now() } + primaryCache.set(collectionId, found) + return found + } + return { forLocation, newCollections: platform.newCollections ?? 'today', async forCollection(collectionId: string): Promise { - const hit = primaryCache.get(collectionId) - if (hit && Date.now() - hit.at < PRIMARY_TTL_MS) - return located(hit.locationId, collectionId, hit.layout) - const [row] = await db - .select({ - locationId: schema.placements.locationId, - layout: schema.collections.recordLayout, - }) - .from(schema.placements) - .innerJoin(schema.collections, eq(schema.collections.id, schema.placements.collectionId)) - .where( - and( - eq(schema.placements.collectionId, collectionId), - eq(schema.placements.role, 'primary'), - ), - ) - .limit(1) - if (!row) throw new Error(`Collection ${collectionId} has no primary placement`) - primaryCache.set(collectionId, { - locationId: row.locationId, - layout: row.layout, - at: Date.now(), - }) - return located(row.locationId, collectionId, row.layout) + const p = await primary(collectionId) + return located(p.locationId, collectionId, p.layout) + }, + async forCollectionAs(collectionId: string, layout: LayoutName): Promise { + return located((await primary(collectionId)).locationId, collectionId, layout) + }, + forget(collectionId: string): void { + primaryCache.delete(collectionId) }, internal: new PrefixedStore(platform.bucket, platform.internalPrefix), fileBytes: files, diff --git a/packages/server/src/versions/commit.ts b/packages/server/src/versions/commit.ts index fa2851ec..79e15274 100644 --- a/packages/server/src/versions/commit.ts +++ b/packages/server/src/versions/commit.ts @@ -38,11 +38,13 @@ import { setRecordTotals, signEntry, type TreeSummary, + type LayoutName, } from '@underlay/protocol' import { eq } from 'drizzle-orm' import { FenceError, writeFence } from '../cleanup/fence.js' import * as schema from '../db/schema.js' +import { LayoutChangedError } from '../layouts/errors.js' import type { Ports } from '../ports.js' import { publishCollectionInfo } from './collection-info.js' import { collectionFileSizes, fileSizes } from './file-refs.js' @@ -62,6 +64,12 @@ export interface BaseVersion { export interface CommitInput { collectionId: string + /** + * The record layout the commit writes in: a parallel commit's plan's, which its + * units wrote in. Else the collection's, read now. It publishes only if the + * collection is still in it. + */ + recordLayout?: LayoutName /** * The storage fence epoch read when this commit's write phase began (before a * parallel commit's units, or around a retry loop). Read here when absent. @@ -115,13 +123,20 @@ export type CommitResult = export async function commitVersion(ports: Ports, input: CommitInput): Promise { const { db } = ports - const repo = await ports.stores.forCollection(input.collectionId) const [collection] = await db .select() .from(schema.collections) .where(eq(schema.collections.id, input.collectionId)) .limit(1) if (!collection) throw new Error(`Collection ${input.collectionId} not found`) + // The layout from the row just read, not an isolate's note of it (a conversion + // may have switched it in the last minute). + const recordLayout = input.recordLayout ?? collection.recordLayout + if (recordLayout !== collection.recordLayout) { + ports.stores.forget(input.collectionId) + throw new LayoutChangedError() + } + const repo = await ports.stores.forCollectionAs(input.collectionId, recordLayout) const fence = input.fence ?? (await writeFence(db)) const built = await buildVersion(repo, { @@ -221,6 +236,7 @@ export async function commitVersion(ports: Ports, input: CommitInput): Promise { +): Promise<{ + ok: boolean + headVersionId: string | null + fenced: boolean + /** The collection's layout changed since the version was written: redo it. */ + layoutMoved: boolean +}> { const v = p.version const now = Date.now() const createdAt = v.createdAt?.getTime() ?? now @@ -145,7 +156,9 @@ export async function publishVersion( pushSessionId: lit(v.pushSessionId ?? null).as('push_session_id'), }) .from(schema.collections) - .where(and(headIsBase, fenceHolds(p.fence))) as never, + .where( + and(headIsBase, eq(schema.collections.recordLayout, p.recordLayout), fenceHolds(p.fence)), + ) as never, ), // 2. Move the head, only if it is still the base and the row from 1 exists. db @@ -224,7 +237,7 @@ export async function publishVersion( await db.batch(statements as unknown as Parameters[0]) const [row] = await db - .select({ head: schema.collections.headVersionId }) + .select({ head: schema.collections.headVersionId, layout: schema.collections.recordLayout }) .from(schema.collections) .where(eq(schema.collections.id, v.collectionId)) .limit(1) @@ -233,5 +246,6 @@ export async function publishVersion( ok, headVersionId: row?.head ?? null, fenced: !ok && (await fenceMoved(db, p.fence)), + layoutMoved: !ok && row !== undefined && row.layout !== p.recordLayout, } } diff --git a/packages/server/test/commit.test.ts b/packages/server/test/commit.test.ts index 5f35390c..5086ee63 100644 --- a/packages/server/test/commit.test.ts +++ b/packages/server/test/commit.test.ts @@ -598,6 +598,7 @@ describe('publishVersion', () => { const fork = { parentCollectionId: parent.id, parentSeq: 1, sets: 'public' as const } const input = (id: string, baseVersionId: string | null) => ({ fence: 0, + recordLayout: child.recordLayout, version: row(child.id, id), baseVersionId, collectionUpdate: { publicFilesRoot: null, summary: null }, diff --git a/packages/server/test/convert.test.ts b/packages/server/test/convert.test.ts new file mode 100644 index 00000000..7e25a361 --- /dev/null +++ b/packages/server/test/convert.test.ts @@ -0,0 +1,279 @@ +/** + * Converting a collection from today's layout to layout F (layouts/convert.ts): + * every version keeps its hash and reads the same, a push made while converting is + * converted too, the switch waits for a committing push, commits wait out the + * switch, and a version written in one layout never publishes in the other. + */ +import { + fsck, + iterate, + type LayoutName, + type RecordSet, + RepoSource, + recordTree, +} from '@underlay/protocol' +import { eq } from 'drizzle-orm' +import { afterAll, afterEach, describe, expect, it } from 'vitest' + +import * as schema from '../src/db/schema.js' +import { conversionState, convertConfig, startConversion } from '../src/layouts/convert.js' +import { parallelConfig } from '../src/push/parallel.js' +import { publishVersion } from '../src/versions/publish.js' +import { cleanup, type Harness, harness } from './harness.js' + +afterAll(cleanup) + +const defaults = { parallel: { ...parallelConfig }, convert: { ...convertConfig } } +afterEach(() => { + Object.assign(parallelConfig, defaults.parallel) + Object.assign(convertConfig, defaults.convert) +}) + +const Author = { type: 'object', properties: { name: { type: 'string' } } } +const rec = (id: string, name: string, priv = false) => ({ + id, + type: 'Author', + data: { name }, + ...(priv ? { private: true } : {}), +}) +const id = (i: number) => `a${String(i).padStart(5, '0')}` +const PATH = '/api/collections/org/conv' + +async function push( + h: Harness, + user: string, + open: Record, + upserts: unknown[], + deletes: string[] = [], + o: { parallel?: boolean; drain?: boolean } = {}, +) { + const res = await h.request(`${PATH}/push`, { method: 'POST', user, json: open }) + expect(res.status).toBe(200) + const sid = ((await res.json()) as { session_id: string }).session_id + if (upserts.length) + expect( + (await h.request(`${PATH}/push/${sid}/records`, { method: 'POST', user, ndjson: upserts })) + .status, + ).toBe(200) + if (deletes.length) + expect( + ( + await h.request(`${PATH}/push/${sid}/deletes`, { + method: 'POST', + user, + ndjson: deletes.map((d) => ({ type: 'Author', id: d })), + }) + ).status, + ).toBe(200) + Object.assign( + parallelConfig, + o.parallel ? { aboveBytes: 0, unitBytes: 32_000, baseLevel: 0 } : { aboveBytes: Infinity }, + ) + const commit = await h.request(`${PATH}/push/${sid}/commit`, { method: 'POST', user }) + if (o.drain !== false) await h.drain() + return commit +} + +async function versions(h: Harness, collectionId: string) { + const rows = await h.ports.db + .select({ seq: schema.versions.seq, hash: schema.versions.hash }) + .from(schema.versions) + .where(eq(schema.versions.collectionId, collectionId)) + return rows.sort((a, b) => a.seq - b.seq).map((r) => r.hash) +} + +/** Every record of version `hash`, with its body, read in `layout` from its set alone. */ +async function records(h: Harness, collectionId: string, hash: string, layout: LayoutName) { + const repo = await h.ports.stores.forCollectionAs(collectionId, layout) + const root = await repo.root(hash) + const priv = root.private ? await repo.privateSet(root.private) : null + const out: string[] = [] + const sets: [RecordSet, typeof root.public | null][] = [ + ['public', root.public], + ['private', priv], + ] + for (const [set, s] of sets) + for (const [type, t] of Object.entries(s?.types ?? {})) + for await (const e of iterate(new RepoSource(recordTree, repo, { set, type }), t.root, { + payloads: true, + })) + out.push(`${set} ${type} ${e.key} ${e.body}`) + return out +} + +async function layoutOf(h: Harness, collectionId: string) { + const [c] = await h.ports.db + .select({ layout: schema.collections.recordLayout }) + .from(schema.collections) + .where(eq(schema.collections.id, collectionId)) + return c!.layout +} + +describe("converting a collection from today's layout to F", () => { + it('keeps every version and record, and converts a push made meanwhile', async () => { + const h = await harness() + const user = await h.member() + const c = await h.collection('conv', { recordLayout: 'today' }) + const big = rec(id(9999), 'x'.repeat(70_000)) + const first = [ + ...Array.from({ length: 3000 }, (_, i) => rec(id(i), `n${i}`)), + ...Array.from({ length: 20 }, (_, i) => rec(id(5000 + i), `p${i}`, true)), + big, + ] + await push(h, user, { schemas: { Author } }, first) + await push( + h, + user, + {}, + Array.from({ length: 200 }, (_, i) => rec(id(i * 7), `again${i}`)), + Array.from({ length: 50 }, (_, i) => id(i * 11 + 1)), + ) + await push( + h, + user, + {}, + Array.from({ length: 400 }, (_, i) => rec(id(i * 5 + 2), `third${i}`)), + [], + { parallel: true }, + ) + // A few leaves a job: the conversion stops and carries on inside trees. + Object.assign(convertConfig, { leavesPerJob: 2, switchPauseMs: 0 }) + await startConversion(h.ports, c.id) + // Committed before the conversion's first job runs: on today's layout, converted too. + const fourth = await push(h, user, {}, [rec(id(4), 'fourth')], [], { drain: false }) + expect(fourth.status).toBe(201) + const hashes = await versions(h, c.id) + expect(hashes).toHaveLength(4) + const before = await Promise.all(hashes.map((v) => records(h, c.id, v, 'today'))) + await h.drain() + + const state = await conversionState(h.ports, c.id) + expect(state).toMatchObject({ status: 'switched', seq: 4 }) + expect(state!.stats.jobs).toBeGreaterThan(3) + expect(state!.stats.records).toBe(new Set(before.flat()).size) + expect(await layoutOf(h, c.id)).toBe('f') + expect(await versions(h, c.id)).toEqual(hashes) + // Every version reads the same from F's packs as from its leaf bodies. + for (const [i, v] of hashes.entries()) expect(await records(h, c.id, v, 'f')).toEqual(before[i]) + const keys = [...h.bucket.objects.keys()] + expect(keys.some((k) => k.startsWith(`repo/runs/${c.id}/public/packs/`))).toBe(true) + expect(keys.some((k) => k.startsWith(`repo/runs/${c.id}/private/packs/`))).toBe(true) + const repo = await h.ports.stores.forCollection(c.id) + expect(repo.layout).toBe('f') + const report = await fsck(repo, { collectionId: c.id, trustedKeys: [h.signer.publicKey] }) + expect(report.errors).toEqual([]) + + // Then it commits on F, serially and in parallel. + await push(h, user, {}, [rec(id(10), 'fifth')]) + await push( + h, + user, + {}, + Array.from({ length: 300 }, (_, i) => rec(id(i * 3), `sixth${i}`)), + [], + { parallel: true }, + ) + const after = await versions(h, c.id) + expect(after).toHaveLength(6) + const head = await records(h, c.id, after.at(-1)!, 'f') + expect(head).toContain( + `public Author ${id(10)} ${JSON.stringify({ id: id(10), type: 'Author', data: { name: 'fifth' } })}`, + ) + expect(head.filter((l) => l.includes('"sixth'))).toHaveLength(300) + // A job that runs again after the switch writes nothing. + expect( + (await fsck(repo, { collectionId: c.id, trustedKeys: [h.signer.publicKey] })).errors, + ).toEqual([]) + }, 60_000) + + it('switches only once no push is committing', async () => { + const h = await harness() + const user = await h.member() + const c = await h.collection('conv', { recordLayout: 'today' }) + await push(h, user, { schemas: { Author } }, [rec(id(1), 'one')]) + // A session mid-commit (as a parallel commit's units would leave it). + const sid = crypto.randomUUID() + await h.ports.db.insert(schema.pushSessions).values({ + id: sid, + collectionId: c.id, + userId: user, + status: 'committing', + expiresAt: new Date(Date.now() + 3_600_000), + } as typeof schema.pushSessions.$inferInsert) + Object.assign(convertConfig, { retrySeconds: 3600 }) + await startConversion(h.ports, c.id) + await h.drain() + expect(await layoutOf(h, c.id)).toBe('today') + expect((await conversionState(h.ports, c.id))?.status).toBe('converting') + // The push finishes; the waiting job switches. + await h.ports.db + .update(schema.pushSessions) + .set({ status: 'failed' }) + .where(eq(schema.pushSessions.id, sid)) + await h.ports.db.update(schema.jobs).set({ runAt: new Date(0) }) + await h.drain() + expect(await layoutOf(h, c.id)).toBe('f') + }) + + it('makes commits wait out the switch', async () => { + const h = await harness() + const user = await h.member() + const c = await h.collection('conv', { recordLayout: 'today' }) + await push(h, user, { schemas: { Author } }, [rec(id(1), 'one')]) + await startConversion(h.ports, c.id) + await h.drain() + expect(await layoutOf(h, c.id)).toBe('f') + const res = await push(h, user, {}, [rec(id(2), 'two')]) + expect(res.status).toBe(503) + expect(await versions(h, c.id)).toHaveLength(1) + convertConfig.switchPauseMs = 0 + expect((await push(h, user, {}, [rec(id(2), 'two')])).status).toBe(201) + }) + + it('never publishes a version in a layout the collection has left', async () => { + const h = await harness() + const c = await h.collection('conv', { recordLayout: 'today' }) + await h.ports.db + .update(schema.collections) + .set({ recordLayout: 'f' }) + .where(eq(schema.collections.id, c.id)) + const versionId = crypto.randomUUID() + const published = await publishVersion(h.ports.db, { + fence: 0, + recordLayout: 'today', + version: { + id: versionId, + collectionId: c.id, + seq: 1, + semver: 'v1.0.0', + major: 1, + minor: 0, + patch: 0, + hash: 'ulv2:' + '0'.repeat(64), + baseSemver: null, + message: null, + pushedBy: null, + appId: null, + actorId: null, + recordCount: 0, + publicRecordCount: 0, + fileCount: 0, + totalBytes: 0, + publicFileCount: 0, + publicTotalBytes: 0, + typeCounts: {}, + publicTypeCounts: {}, + hasPrivate: false, + publicRefsRoot: null, + privateRefsRoot: null, + changes: { added: 0, removed: 0, updated: 0 }, + }, + baseVersionId: null, + collectionUpdate: { publicFilesRoot: null, summary: null }, + schemaHashes: [], + usage: [], + }) + expect(published).toMatchObject({ ok: false, layoutMoved: true }) + expect(await versions(h, c.id)).toEqual([]) + }) +})