diff --git a/packages/chord/test/next/examples/60-session-sync/main.ts b/packages/chord/test/next/examples/60-session-sync/main.ts new file mode 100644 index 000000000..04c922d8f --- /dev/null +++ b/packages/chord/test/next/examples/60-session-sync/main.ts @@ -0,0 +1,180 @@ +// Session sync as decided (decisions S3, S5, S7, W9, "The outbox is optional, internal and versioned", "p3 meeting, +// 2026-11-02"), played with Chord's real pieces: a laptop store with an outbox, `store.outbox.read`/`trim`, +// `store.export` with `encodeJsonl`, `importStore`, and batch apply, `applyBatch` (internal: exported from +// `chord/next` marked internal, imported here from `src/next/store/apply-batch.ts`; it takes the `ChordStore` behind +// `openStore`'s `Store`). Radius and the laptop's sync client are stand-ins written for this example (`radius.ts`, +// `sync.ts`), not Chord APIs: the sync layer is not built yet (slices.md, "The sync layer"). +// Steps: 1. a session on the laptop; 2. the handshake and a first upload; 3. a second sync moves only the tail; +// 4. a bootstrap from an archive for a session whose outbox was turned on late; 5. "continue on Radius" and the epoch +// fence (the split), then the same without the fence (silent divergence); 6. back to the laptop with a pull. +// Run from packages/chord: +// node --conditions=source --experimental-strip-types test/next/examples/60-session-sync/main.ts +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { Readable } from "node:stream"; +import { BACKGROUND_CONTEXT as context } from "../../../../src/context/index.ts"; +import { decodeJsonl, importStore, openStore, type Store, sqliteBackend } from "../../../../src/next/store/index.ts"; +import { openNodeSqliteDatabase } from "../../../../src/sqlite/node.ts"; +import { RadiusStandIn } from "./radius.ts"; +import { describe, withSession } from "./session.ts"; +import { archiveOf, type LocalSession, pull, push, type Synced, sync } from "./sync.ts"; + +const radius = new RadiusStandIn(); +const directory = await mkdtemp(join(tmpdir(), "chord-sync-")); + +/** One sync's result as a line. */ +function shown(result: Synced): string { + switch (result.kind) { + case "uploaded": + return `uploaded ${result.from}..${result.to}, Radius at ${result.lastSeq}, trimmed through it`; + case "nothing": + return `nothing to send, Radius at ${result.lastSeq}`; + case "bootstrapped": + return `archive uploaded (${result.bytes} bytes), Radius imported it at ${result.lastSeq}`; + case "behind": + return `Radius is ahead at ${result.lastSeq}`; + case "split": + return `refused for an old epoch: this copy is now ${result.session.id}`; + case "refused": + return `refused: ${result.message}`; + } +} + +/** A new in-memory store (outbox on) imported from archive bytes. */ +async function copyOf(bytes: Uint8Array): Promise { + const backend = sqliteBackend(await openNodeSqliteDatabase(":memory:")); + return importStore(backend, decodeJsonl(Readable.from([bytes])), context, { outbox: true }); +} + +/** Radius's view of a session as a line. */ +function atRadius(id: string): string { + const known = radius.handshake()[id]!; + return `Radius ${id}: lastSeq ${known.lastSeq}, log ${radius.logRange(id)}, epoch ${known.epoch} (${known.driver})`; +} + +try { + // 1. The laptop: SQLite in memory with the outbox on from its first commit; one session, a few commits. + const store = await openStore(sqliteBackend(await openNodeSqliteDatabase(":memory:")), context, { outbox: true }); + const s1: LocalSession = { id: "s1", store, epoch: 1, acked: 0 }; + await withSession(store, async (session) => { + await session.rename("Fix the parser", context); + await session.say({ role: "user", text: "the parser drops comments" }, context); + await session.say({ role: "assistant", text: "found it in lexer.ts" }, context); + }); + console.log("1. laptop s1: lastSeq", store.lastSeq, "outbox floor", store.outbox!.floor); + + // 2. The handshake: Radius has no s1 (lastSeq 0), so the whole outbox goes, 1..lastSeq; then trimmed. + console.log("2. sync:", shown(await sync(s1, radius))); + console.log(" ", atRadius("s1"), "- laptop outbox floor", store.outbox!.floor); + + // 3. Two more commits; the next sync sends only them. + await withSession(store, async (session) => { + await session.say({ role: "user", text: "and strings too" }, context); + await session.say({ role: "assistant", text: "fixed both" }, context); + }); + console.log("3. laptop s1: lastSeq", store.lastSeq, "- sync:", shown(await sync(s1, radius))); + console.log(" ", atRadius("s1")); + console.log(" laptop:", await describe(store)); + console.log(" Radius:", await describe(radius.replica("s1"))); + + // 4. Session s2 ran without an outbox (a file here, since it is reopened), then sync was turned on: the outbox starts + // after its lastSeq, so Radius, which has nothing, needs an archive first; batches go on from there. + const path = join(directory, "s2.sqlite"); + const before = await openStore(sqliteBackend(await openNodeSqliteDatabase(path)), context); + await withSession(before, async (session) => { + await session.rename("Release notes", context); + await session.say({ role: "user", text: "draft the notes for 1.4" }, context); + }); + await before.close(context); + const store2 = await openStore(sqliteBackend(await openNodeSqliteDatabase(path)), context, { outbox: true }); + const s2: LocalSession = { id: "s2", store: store2, epoch: 1, acked: 0 }; + console.log( + "4. laptop s2: lastSeq", + store2.lastSeq, + "outbox floor", + store2.outbox!.floor, + "- sync:", + shown(await sync(s2, radius)), + ); + await withSession(store2, async (session) => { + await session.say({ role: "assistant", text: "here is a draft" }, context); + }); + console.log(" laptop s2: lastSeq", store2.lastSeq, "- sync:", shown(await sync(s2, radius))); + console.log(" ", atRadius("s2")); + console.log(" Radius:", await describe(radius.replica("s2"))); + await store2.close(context); + + // 5. "Continue on Radius": the laptop uploads its tail (nothing left here), Radius bumps the epoch and drives, + // writing a commit on its replica. The laptop keeps its copy as a replica and writes nothing. + const handoff = await archiveOf(store); + console.log( + "5. laptop s1 sync:", + shown(await sync(s1, radius)), + "- Radius takes over, epoch", + radius.takeOver("s1"), + ); + await radius.drive("s1", (replica) => + withSession(replica, async (session) => { + await session.say({ role: "assistant", text: "(on Radius) added a test" }, context); + }), + ); + console.log(" ", atRadius("s1")); + // A laptop process that missed the handoff (here a copy of the laptop's store at it, so the laptop itself can take + // the session back in step 6) writes at epoch 1 and uploads the new commit after its last acknowledgement. + const stale: LocalSession = { id: "s1", store: await copyOf(handoff), epoch: 1, acked: 6 }; + await withSession(stale.store, async (session) => { + await session.say({ role: "assistant", text: "(stale laptop) added a different test" }, context); + }); + const fenced = await push(stale, radius); + console.log(" stale laptop: lastSeq", stale.store.lastSeq, "- push:", shown(fenced)); + // The split (S7): no rollback; the copy is a new session with its origin, synced on its own from an archive. + if (fenced.kind === "split") { + const split = fenced.session; + console.log( + " split", + split.id, + "origin", + JSON.stringify(split.origin), + "- sync:", + shown(await sync(split, radius)), + ); + console.log(" ", atRadius(split.id)); + } + await stale.store.close(context); + + // 5b. The same without the epoch fence: the stale upload of seq 7 does not follow Radius's 7 and is refused, so far + // so good; but the next handshake finds both at 7, sends nothing and trims the laptop's commit 7 as acknowledged. + radius.checkEpochs = false; + const unfenced: LocalSession = { id: "s1", store: await copyOf(handoff), epoch: 1, acked: 6 }; + await withSession(unfenced.store, async (session) => { + await session.say({ role: "assistant", text: "(stale laptop) added a different test" }, context); + }); + console.log("5b. no epoch check - stale push:", shown(await push(unfenced, radius))); + console.log(" stale sync:", shown(await sync(unfenced, radius)), "- outbox floor", unfenced.store.outbox!.floor); + console.log(" stale: ", await describe(unfenced.store)); + console.log(" Radius:", await describe(radius.replica("s1"))); + await unfenced.store.close(context); + radius.checkEpochs = true; + + // 6. Back to the laptop: it takes the epoch (Radius stops driving), pulls Radius's commits after its lastSeq + // (batch apply on the laptop's store; no host open on it), and drives again. + s1.epoch = radius.handBack("s1", "laptop"); + const pulled = await pull(s1, radius); + console.log(`6. laptop takes epoch ${s1.epoch}, pulls ${pulled.from}..${pulled.to}: lastSeq ${store.lastSeq}`); + const [laptopView, radiusView] = [await describe(store), await describe(radius.replica("s1"))]; + console.log(" laptop:", laptopView); + console.log(" Radius:", radiusView); + console.log(" same values:", laptopView === radiusView); + await withSession(store, async (session) => { + await session.rename("Fix the parser (done)", context); + }); + console.log(" laptop writes, sync:", shown(await sync(s1, radius))); + console.log(" ", atRadius("s1")); + console.log(" same values:", (await describe(store)) === (await describe(radius.replica("s1")))); + + await store.close(context); +} finally { + await radius.close(); + await rm(directory, { recursive: true, force: true }); +} diff --git a/packages/chord/test/next/examples/60-session-sync/radius.ts b/packages/chord/test/next/examples/60-session-sync/radius.ts new file mode 100644 index 000000000..177ca5cf9 --- /dev/null +++ b/packages/chord/test/next/examples/60-session-sync/radius.ts @@ -0,0 +1,159 @@ +// STAND-IN for Radius, not a Chord API: what example 60 needs of the server side of session sync, in memory. Per +// session id it keeps a replica store (SQLite in memory, as a Durable Object's SQLite would be), the commit log (what +// R2 would hold; the replica's outbox is moved into it and trimmed after every change), the driver and its epoch. +// Batches and archives cross a "wire" as JSON and JSONL bytes, so the replica checks them as a trust boundary. +// `checkEpochs = false` turns the epoch fence off, to show what the gap analysis warned about. +import { Readable } from "node:stream"; +import { BACKGROUND_CONTEXT as context } from "../../../../src/context/index.ts"; +import { ChordError } from "../../../../src/next/data/index.ts"; +// Internal: the sync layer's batch apply (exported from `chord/next` marked internal) takes the `ChordStore` that +// `openStore` returns behind the app-facing `Store` type. +import { applyBatch } from "../../../../src/next/store/apply-batch.ts"; +import { + decodeJsonl, + importStore, + type OutboxCommit, + openStore, + type Store, + sqliteBackend, +} from "../../../../src/next/store/index.ts"; +import type { ChordStore } from "../../../../src/next/store/store.ts"; +import { openNodeSqliteDatabase } from "../../../../src/sqlite/node.ts"; + +/** What the handshake reports per session: Radius's applied head, the epoch and who drives. */ +export type Known = { readonly lastSeq: number; readonly epoch: number; readonly driver: string }; + +/** A batch as it travels: the commits `from..to` and the uploader's epoch (S5: every batch carries `from` and `to`). */ +export type Batch = { + readonly from: number; + readonly to: number; + readonly epoch: number; + readonly commits: readonly OutboxCommit[]; +}; + +/** An upload's answer: applied (the new head is the acknowledgement) or refused, for an old epoch or by the replica. */ +export type Uploaded = + | { readonly kind: "applied"; readonly lastSeq: number; readonly epoch: number } + | { readonly kind: "refused"; readonly reason: "epoch"; readonly epoch: number } + | { readonly kind: "refused"; readonly reason: "batch"; readonly message: string }; + +type Entry = { + readonly replica: Store; + /** The commit log (R2's part): every commit after `logFloor` (the seq it was bootstrapped at). */ + readonly log: OutboxCommit[]; + readonly logFloor: number; + epoch: number; + driver: string; +}; + +/** The stand-in: sessions by id, for one owner. */ +export class RadiusStandIn { + /** The epoch fence (S7); off only to show the divergence without it. */ + checkEpochs = true; + readonly #sessions = new Map(); + + /** The handshake: every session Radius has, with its `lastSeq`, epoch and driver. */ + handshake(): Record { + const known: Record = {}; + for (const [id, { replica, epoch, driver }] of this.#sessions) + known[id] = { lastSeq: replica.lastSeq, epoch, driver }; + return known; + } + + /** Applies an uploaded batch (JSON) to the session's replica; a first batch from seq 1 makes the session. */ + async upload(id: string, driver: string, wire: string): Promise { + const batch = JSON.parse(wire) as Batch; + let entry = this.#sessions.get(id); + if (entry === undefined && batch.from === 1) entry = this.#create(id, driver, await openReplica()); + if (entry === undefined) return { kind: "refused", reason: "batch", message: "unknown session: bootstrap first" }; + if (this.checkEpochs && batch.epoch !== entry.epoch) + return { kind: "refused", reason: "epoch", epoch: entry.epoch }; + try { + await applyBatch(entry.replica as ChordStore, batch.commits, context); + } catch (error) { + if (!(error instanceof ChordError)) throw error; + return { kind: "refused", reason: "batch", message: error.message }; + } + await this.#moveToLog(entry); + return { kind: "applied", lastSeq: entry.replica.lastSeq, epoch: entry.epoch }; + } + + /** Makes the session from an archive (JSONL bytes): its replica imported, the log starting at its `lastSeq`. */ + async bootstrap(id: string, driver: string, bytes: Uint8Array): Promise { + const backend = sqliteBackend(await openNodeSqliteDatabase(":memory:")); + const replica = await importStore(backend, decodeJsonl(Readable.from([bytes])), context, { outbox: true }); + const entry = this.#create(id, driver, replica); + return { lastSeq: replica.lastSeq, epoch: entry.epoch, driver }; + } + + /** "Continue on Radius": a new epoch, Radius drives; an upload at the old epoch is refused from now on. */ + takeOver(id: string): number { + const entry = this.#entry(id); + entry.epoch += 1; + entry.driver = "radius"; + return entry.epoch; + } + + /** Radius as the driver: `work` writes on the replica (a host on it, closed again), its commits go to the log. */ + async drive(id: string, work: (replica: Store) => Promise): Promise { + const entry = this.#entry(id); + if (entry.driver !== "radius") throw new Error(`radius: ${id} is driven by ${entry.driver}`); + await work(entry.replica); + await this.#moveToLog(entry); + } + + /** "Back to the laptop": a new epoch for `driver`; Radius stops writing. */ + handBack(id: string, driver: string): number { + const entry = this.#entry(id); + entry.epoch += 1; + entry.driver = driver; + return entry.epoch; + } + + /** The log's commits after `afterSeq` (JSON), for a pull. */ + pull(id: string, afterSeq: number): string { + const entry = this.#entry(id); + if (afterSeq < entry.logFloor) throw new Error(`radius: ${id}'s log starts after ${entry.logFloor}`); + return JSON.stringify(entry.log.filter((commit) => commit.seq > afterSeq)); + } + + /** The session's replica, to read it in the example. */ + replica(id: string): Store { + return this.#entry(id).replica; + } + + /** The seqs the log holds, as `from..to`. */ + logRange(id: string): string { + const { log, logFloor } = this.#entry(id); + return log.length === 0 ? `empty after ${logFloor}` : `${log[0]!.seq}..${log.at(-1)!.seq}`; + } + + /** Closes every replica. */ + async close(): Promise { + for (const { replica } of this.#sessions.values()) await replica.close(context); + } + + #create(id: string, driver: string, replica: Store): Entry { + const entry: Entry = { replica, log: [], logFloor: replica.lastSeq, epoch: 1, driver }; + this.#sessions.set(id, entry); + return entry; + } + + #entry(id: string): Entry { + const entry = this.#sessions.get(id); + if (entry === undefined) throw new Error(`radius: no session ${id}`); + return entry; + } + + /** The replica's new commits into the log (the DO writing its commit log to R2), then trimmed from its outbox. */ + async #moveToLog(entry: Entry): Promise { + const outbox = entry.replica.outbox!; + entry.log.push(...(await outbox.read(outbox.floor, 10_000, context))); + await outbox.trim(entry.replica.lastSeq, context); + } +} + +/** A replica as Radius makes it for a session uploaded from seq 1: empty, its outbox on. */ +async function openReplica(): Promise { + return openStore(sqliteBackend(await openNodeSqliteDatabase(":memory:")), context, { outbox: true }); +} diff --git a/packages/chord/test/next/examples/60-session-sync/session.ts b/packages/chord/test/next/examples/60-session-sync/session.ts new file mode 100644 index 000000000..b1d4c9e30 --- /dev/null +++ b/packages/chord/test/next/examples/60-session-sync/session.ts @@ -0,0 +1,83 @@ +// The session example 60 syncs: one object per host with a persisted title document and a `messages` table, and a +// helper running a host on a store for the length of some work (one open host per store; a closed host holds no +// claims, so a batch may apply to its store afterwards). +import { BACKGROUND_CONTEXT as context } from "../../../../src/context/index.ts"; +import { arr, int, local, method, obj, str } from "../../../../src/next/contracts/index.ts"; +import { defineState } from "../../../../src/next/documents/index.ts"; +import { createHost, defineFacet, type Reference } from "../../../../src/next/host/index.ts"; +import { defineTable, query, type Store } from "../../../../src/next/store/index.ts"; + +const Messages = defineTable("messages", { + version: 1, + columns: { n: int(), role: str(), text: str() }, + key: ["n"], +}); + +const Title = defineState("title", { + schema: obj({ text: str() }), + version: 1, + initial: () => ({ text: "untitled" }), + persist: true, + checkpointWhen: () => false, +}); + +/** The session's contract: its title, a message (one commit each), a rename (one commit) and the transcript. */ +export const Session = local("example.session", { + version: 1, + params: {}, + members: { + title: obj({ text: str() }), + say: method(obj({ role: str(), text: str() }), int()), + rename: method(str()), + transcript: method(obj({}), arr(str())), + }, +}); + +const facet = defineFacet((host) => { + host.serve(Session, (_, self) => { + const messages = self.table(Messages); + const title = self.state(Title); + const all = query(Messages).orderBy((m) => m.n, "asc"); + return { + title, + say: async ({ role, text }, ctx) => { + const n = (await messages.count(all, ctx)) + 1; + await self.tx(async (tx) => tx.put(messages, { n, role, text }), ctx); + return n; + }, + rename: async (text, ctx) => { + await title.change((draft) => { + draft.text = text; + }, ctx); + return null; + }, + transcript: async (_, ctx) => { + const lines: string[] = []; + for await (const m of messages.rows(all, ctx)) lines.push(`${m.n} ${m.role}: ${m.text}`); + return lines; + }, + }; + }); +}); + +/** Runs a host serving the session on `store` until `work` is done, then closes the host (the store stays open). */ +export async function withSession( + store: Store, + work: (session: Reference) => Promise, +): Promise { + const host = createHost({ store, onError: (error) => console.error("background error:", error) }); + try { + await host.load([{ name: "@example/session", facet }], context); + return await work(await host.get(Session, {}, context)); + } finally { + await host.close(context); + } +} + +/** The session as one line: its title and its messages. */ +export function describe(store: Store): Promise { + return withSession(store, async (session) => { + const lines = await session.transcript({}, context); + return `"${session.title.value.text}" [${lines.join(" | ")}]`; + }); +} diff --git a/packages/chord/test/next/examples/60-session-sync/sync.ts b/packages/chord/test/next/examples/60-session-sync/sync.ts new file mode 100644 index 000000000..e88c7e0b9 --- /dev/null +++ b/packages/chord/test/next/examples/60-session-sync/sync.ts @@ -0,0 +1,101 @@ +// STAND-IN for the laptop's sync layer, not a Chord API: the decided handshake (ask Radius for `lastSeq`, send the +// outbox from `lastSeq + 1`, trim through the answer), the bootstrap from an archive, the eager upload of new commits +// after the last acknowledgement (S5), the split on an old epoch (S7) and the pull of Radius's commits. Built on +// Chord's real pieces: `store.outbox.read`/`trim`, `store.export`, `encodeJsonl`, and `applyBatch` (internal). +import { BACKGROUND_CONTEXT as context } from "../../../../src/context/index.ts"; +import { applyBatch } from "../../../../src/next/store/apply-batch.ts"; +import { encodeJsonl, type OutboxCommit, type Store } from "../../../../src/next/store/index.ts"; +import type { ChordStore } from "../../../../src/next/store/store.ts"; +import type { Batch, RadiusStandIn } from "./radius.ts"; + +/** A session on the laptop: its id, store, the epoch it drives in (a new session: 1) and the `lastSeq` Radius last acknowledged. */ +export type LocalSession = { + readonly id: string; + readonly store: Store; + epoch: number; + acked: number; + /** A split-off copy's origin: the session it was and the last seq they share (S7). */ + readonly origin?: { readonly id: string; readonly seq: number }; +}; + +/** What a sync or a push did. */ +export type Synced = + | { readonly kind: "uploaded"; readonly from: number; readonly to: number; readonly lastSeq: number } + | { readonly kind: "nothing"; readonly lastSeq: number } + | { readonly kind: "bootstrapped"; readonly lastSeq: number; readonly bytes: number } + | { readonly kind: "behind"; readonly lastSeq: number } + | { readonly kind: "split"; readonly session: LocalSession } + | { readonly kind: "refused"; readonly message: string }; + +const DRIVER = "laptop"; + +/** The handshake, then the outbox from Radius's `lastSeq + 1`; an archive when Radius has less than the outbox. */ +export async function sync(session: LocalSession, radius: RadiusStandIn): Promise { + const { store } = session; + const outbox = store.outbox!; + const known = radius.handshake()[session.id]; + if (known === undefined && outbox.floor > 0) return bootstrap(session, radius); + if (known !== undefined && known.lastSeq < outbox.floor) return bootstrap(session, radius); + // Radius ahead: a pull-back (S3, same epoch) or a split (S7); step 6 pulls explicitly. + if (known !== undefined && known.lastSeq > store.lastSeq) return { kind: "behind", lastSeq: known.lastSeq }; + session.acked = known?.lastSeq ?? 0; + await outbox.trim(session.acked, context); + return push(session, radius); +} + +/** Uploads the outbox after the last acknowledgement, without asking first (an eager or resubmitted upload, S5). */ +export async function push(session: LocalSession, radius: RadiusStandIn): Promise { + const { store } = session; + const commits = await store.outbox!.read(session.acked, 10_000, context); + if (commits.length === 0) return { kind: "nothing", lastSeq: session.acked }; + const batch: Batch = { from: commits[0]!.seq, to: commits.at(-1)!.seq, epoch: session.epoch, commits }; + const answer = await radius.upload(session.id, DRIVER, JSON.stringify(batch)); + switch (answer.kind) { + case "applied": + session.acked = answer.lastSeq; + session.epoch = answer.epoch; + await store.outbox!.trim(answer.lastSeq, context); + return { kind: "uploaded", from: batch.from, to: batch.to, lastSeq: answer.lastSeq }; + case "refused": + if (answer.reason === "batch") return { kind: "refused", message: answer.message }; + // S7: no rollback; this copy goes on as a new session that shares the old one's commits through `acked`. + return { + kind: "split", + session: { + id: `${session.id}~split@${session.acked}`, + store, + epoch: 1, + acked: 0, + origin: { id: session.id, seq: session.acked }, + }, + }; + } +} + +/** Uploads the store's archive; Radius imports it and batches go on from its `lastSeq`. */ +async function bootstrap(session: LocalSession, radius: RadiusStandIn): Promise { + const bytes = await archiveOf(session.store); + const known = await radius.bootstrap(session.id, DRIVER, bytes); + session.acked = known.lastSeq; + session.epoch = known.epoch; + await session.store.outbox!.trim(known.lastSeq, context); + return { kind: "bootstrapped", lastSeq: known.lastSeq, bytes: bytes.length }; +} + +/** Applies Radius's commits after the local `lastSeq` (no host may be open on the store: its claims refuse a batch). */ +export async function pull(session: LocalSession, radius: RadiusStandIn): Promise<{ from: number; to: number }> { + const commits = JSON.parse(radius.pull(session.id, session.store.lastSeq)) as OutboxCommit[]; + const from = session.store.lastSeq + 1; + await applyBatch(session.store as ChordStore, commits, context); + // Batch apply writes the commits to this store's outbox too; Radius has them, so they are acknowledged. + session.acked = session.store.lastSeq; + await session.store.outbox!.trim(session.acked, context); + return { from, to: session.store.lastSeq }; +} + +/** The store's archive as JSONL bytes (the commit queue held while it is read). */ +export async function archiveOf(store: Store): Promise { + const parts: Uint8Array[] = []; + for await (const part of encodeJsonl(store.export(context))) parts.push(part); + return Buffer.concat(parts); +}