chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
A commit deletes the blocks it leaves unreachable, and getRepo streams the table
.changeset/collect-blocks.mdnew+20−0
| @@ -0,0 +1,20 @@ | ||
--- | ||
'@pdsjs/core': patch | ||
'@pdsjs/cloudflare': patch | ||
'@pdsjs/storage-sqlite': patch | ||
--- | ||
| ||
A commit deletes the blocks it leaves unreachable: the tree nodes it replaced, | ||
the records it replaced or deleted that no other record shares, and the | ||
previous commit, as the reference PDS does. Until now every block was kept, so | ||
a deleted record stayed in storage, readable by its CID to anything with the | ||
account's storage, extensions included. | ||
| ||
`com.atproto.sync.getRepo` then reads the block table a page at a time and | ||
streams it, rather than walking the tree block by block and building the whole | ||
CAR first. The first `getRepo` after upgrading sweeps the blocks stored before, | ||
once, holding writes while it does. A rebuilt tree or an imported repo is swept | ||
the same way. | ||
| ||
`com.atproto.sync.getRecord` sends the path from the root to the record, not | ||
every node of the tree. | ||
packages/cloudflare/src/index.js+76−2
| @@ -135,6 +135,23 @@ function safeOne(cursor) { | ||
return rows.length > 0 ? rows[0] : null; | ||
} | ||
| ||
/** Durable Object SQL binds at most 100 parameters to one statement. */ | ||
const MAX_BOUND = 100; | ||
| ||
/** | ||
* @template T | ||
* @param {T[]} items | ||
* @returns {T[][]} | ||
*/ | ||
function chunks(items) { | ||
/** @type {T[][]} */ | ||
const out = []; | ||
for (let i = 0; i < items.length; i += MAX_BOUND) { | ||
out.push(items.slice(i, i + MAX_BOUND)); | ||
} | ||
return out; | ||
} | ||
| ||
/** | ||
* @typedef {Object} BlobLike | ||
* @property {ArrayBuffer} buffer | ||
| @@ -158,11 +175,12 @@ function normalizeBlob(value) { | ||
} | ||
| ||
/** | ||
* Create storage adapter for Cloudflare D1 (Durable Object SQLite) | ||
* Create storage adapter for Cloudflare D1 (Durable Object SQLite). Exported | ||
* for the tests, which run it over better-sqlite3. | ||
* @param {D1SqlInterface} sql - D1 SQL interface from state.storage.sql | ||
* @returns {ActorStoragePort} | ||
*/ | ||
function createActorStorage(sql) { | ||
export function createActorStorage(sql) { | ||
// Initialize schema | ||
sql.exec(` | ||
CREATE TABLE IF NOT EXISTS metadata ( | ||
| @@ -183,6 +201,7 @@ function createActorStorage(sql) { | ||
value BLOB | ||
); | ||
CREATE INDEX IF NOT EXISTS idx_records_collection ON records(collection, rkey); | ||
CREATE INDEX IF NOT EXISTS idx_records_cid ON records(cid); | ||
| ||
CREATE TABLE IF NOT EXISTS commits ( | ||
seq INTEGER PRIMARY KEY, | ||
| @@ -306,6 +325,61 @@ function createActorStorage(sql) { | ||
); | ||
}, | ||
| ||
async deleteBlocks(cids) { | ||
for (const part of chunks(cids)) { | ||
const marks = part.map(() => '?').join(','); | ||
sql.exec(`DELETE FROM blocks WHERE cid IN (${marks})`, ...part); | ||
} | ||
}, | ||
| ||
async listBlocks(cursor, limit) { | ||
const rows = ( | ||
cursor | ||
? sql.exec( | ||
'SELECT cid, data FROM blocks WHERE cid > ? ORDER BY cid LIMIT ?', | ||
cursor, | ||
limit, | ||
) | ||
: sql.exec('SELECT cid, data FROM blocks ORDER BY cid LIMIT ?', limit) | ||
).toArray(); | ||
const blocks = rows.map((r) => ({ | ||
cid: String(r.cid), | ||
data: /** @type {Uint8Array} */ (normalizeBlob(r.data)), | ||
})); | ||
const last = blocks.length === limit ? blocks[limit - 1].cid : null; | ||
return { blocks, cursor: last }; | ||
}, | ||
| ||
async recordCidsAmong(cids) { | ||
/** @type {string[]} */ | ||
const found = []; | ||
for (const part of chunks(cids)) { | ||
const marks = part.map(() => '?').join(','); | ||
const rows = sql | ||
.exec( | ||
`SELECT DISTINCT cid FROM records WHERE cid IN (${marks})`, | ||
...part, | ||
) | ||
.toArray(); | ||
found.push(...rows.map((r) => String(r.cid))); | ||
} | ||
return found; | ||
}, | ||
| ||
async getBlocksCollected() { | ||
const row = safeOne( | ||
sql.exec("SELECT value FROM metadata WHERE key = 'blocksCollected'"), | ||
); | ||
return String(row?.value) === '1'; | ||
}, | ||
| ||
async setBlocksCollected(collected) { | ||
sql.exec( | ||
"INSERT OR REPLACE INTO metadata (key, value) VALUES ('blocksCollected', ?)", | ||
collected ? '1' : '0', | ||
); | ||
}, | ||
| ||
async getLatestCommit() { | ||
const row = /** @type {CommitRow|null} */ ( | ||
safeOne( | ||
packages/cloudflare/test/blocks.test.jsnew+9−0
| @@ -0,0 +1,9 @@ | ||
import { runBlockStorageConformance } from '../../../test/helpers/block-storage-conformance.js'; | ||
import { createD1Shim } from '../../../test/helpers/d1-shim.js'; | ||
import { createActorStorage } from '../src/index.js'; | ||
| ||
runBlockStorageConformance('cloudflare (D1 shim)', async () => { | ||
const Database = (await import('better-sqlite3')).default; | ||
const { sql } = createD1Shim(new Database(':memory:')); | ||
return /** @type {any} */ (createActorStorage(/** @type {any} */ (sql))); | ||
}); | ||
packages/core/src/handlers/account-backup.js+19−2
| @@ -17,6 +17,9 @@ import { | ||
} from '../backup.js'; | ||
import { base64UrlDecode } from '../crypto.js'; | ||
| ||
/** Heads a snapshot tries before giving up, should commits keep landing. */ | ||
const SNAPSHOT_TRIES = 3; | ||
| ||
/** | ||
* Count the bytes of a stream as they pass, and give the reader a place to | ||
* checkpoint between them. One blob is more than one request to the target, | ||
| @@ -156,7 +159,7 @@ export function createBackupHandlers(ctx) { | ||
try { | ||
const did = await getDid(); | ||
if (!did) throw new Error('No account on this server yet'); | ||
const commit = await actorStorage.getLatestCommit(); | ||
let commit = await actorStorage.getLatestCommit(); | ||
if (!commit) throw new Error('No repository to back up'); | ||
| ||
const state = await readBackupState(); | ||
| @@ -229,7 +232,21 @@ export function createBackupHandlers(ctx) { | ||
}; | ||
| ||
try { | ||
const car = await buildFullRepoCar(commit.cid); | ||
/** @type {Uint8Array|null} */ | ||
let car = null; | ||
for (let tries = 1; !car; tries++) { | ||
try { | ||
car = await buildFullRepoCar(commit.cid); | ||
} catch (err) { | ||
// A commit landing during the walk drops nodes of the tree it has | ||
// not reached yet: the snapshot is of the new head instead. | ||
const head = await actorStorage.getLatestCommit(); | ||
if (tries === SNAPSHOT_TRIES || !head || head.cid === commit.cid) { | ||
throw err; | ||
} | ||
commit = head; | ||
} | ||
} | ||
run.carBytes = car.byteLength; | ||
await checkpoint(true); | ||
| ||
packages/core/src/handlers/xrpc-sync.js+135−12
| @@ -6,17 +6,26 @@ | ||
// getBlob and listBlobs are in the same namespace, but they read blob storage | ||
// and not the repo. They are with the other blob endpoints. | ||
| ||
import { collectMstBlocks } from '../mst.js'; | ||
import { buildCarFile, cborDecode, cidToString } from '../repo.js'; | ||
import { mstGet } from '../mst.js'; | ||
import { | ||
buildCarFile, | ||
carBlock, | ||
carHeader, | ||
cborDecode, | ||
cidToString, | ||
} from '../repo.js'; | ||
| ||
/** | ||
* @typedef {Object} SyncContext | ||
* @property {import('../ports.js').ActorStoragePort} actorStorage | ||
* @property {() => Promise<string|null>} getDid | ||
* @property {() => Promise<boolean>} isDeactivated | ||
* @property {<T>(fn: () => Promise<T>) => Promise<T>} [withRepoLock] - Runs | ||
* `fn` after the repo's writes queued before it; the sweep before getRepo | ||
* holds it. Absent, `fn` runs at once. | ||
*/ | ||
| ||
/** Blocks buildFullRepoCar reads between two waits on a timer. */ | ||
/** Blocks read between two waits on a timer, or in one page of the table. */ | ||
const BLOCKS_PER_WAIT = 500; | ||
| ||
/** @returns {Response} */ | ||
| @@ -110,6 +119,14 @@ export function createSyncHandlers(ctx) { | ||
return repoNotFound(); | ||
} | ||
| ||
if (await blocksCollected()) { | ||
// Swept, the head may have moved on while this waited for the lock. | ||
const head = (await actorStorage.getLatestCommit()) ?? commit; | ||
return new Response(streamRepoCar(head.cid), { | ||
headers: { 'Content-Type': 'application/vnd.ipld.car' }, | ||
}); | ||
} | ||
| ||
let car; | ||
try { | ||
car = await buildFullRepoCar(commit.cid); | ||
| @@ -131,7 +148,17 @@ export function createSyncHandlers(ctx) { | ||
| ||
/** | ||
* The whole repo as one CAR: the commit, every MST node, and every record. | ||
* Serves com.atproto.sync.getRepo and is the repo half of a backup snapshot. | ||
* Serves com.atproto.sync.getRepo from a store that keeps every block it | ||
* was given, and is the repo half of a backup snapshot. | ||
* @param {string} commitCid | ||
* @returns {Promise<Uint8Array>} | ||
*/ | ||
async function buildFullRepoCar(commitCid) { | ||
return buildCarFile(commitCid, await repoBlocks(commitCid)); | ||
} | ||
| ||
/** | ||
* Every block the commit reaches, by walking its tree. | ||
* | ||
* Throws if any of those blocks is absent. A caller must not answer with a | ||
* part of a repo, because a relay accepts it as the whole one. | ||
| @@ -140,9 +167,9 @@ export function createSyncHandlers(ctx) { | ||
* Object delivers no other request until its code waits on something other | ||
* than a promise already settled, and a whole repo is thousands of blocks. | ||
* @param {string} commitCid | ||
* @returns {Promise<Uint8Array>} | ||
* @returns {Promise<Array<{cid: string, data: Uint8Array}>>} | ||
*/ | ||
async function buildFullRepoCar(commitCid) { | ||
async function repoBlocks(commitCid) { | ||
/** @type {Array<{cid: string, data: Uint8Array}>} */ | ||
const blocks = []; | ||
const taken = new Set(); | ||
| @@ -196,7 +223,93 @@ export function createSyncHandlers(ctx) { | ||
if (commit?.data instanceof Uint8Array) { | ||
await collectNode(cidToString(commit.data)); | ||
} | ||
return buildCarFile(commitCid, blocks); | ||
return blocks; | ||
} | ||
| ||
/** | ||
* Whether the block table holds only what the current commit reaches, so | ||
* that it can be read as the repo; sweeping it first, once, if not. | ||
* | ||
* Each commit drops what it leaves unreachable, but blocks stored before | ||
* commits did, or left by a rebuilt tree or an imported CAR, stay until | ||
* this sweeps them. It holds the repo lock, so no commit stores a block it | ||
* has not counted, and waits on a timer between pages, as the walk does. | ||
* @returns {Promise<boolean>} | ||
*/ | ||
async function blocksCollected() { | ||
const store = actorStorage; | ||
if ( | ||
!store.listBlocks || | ||
!store.deleteBlocks || | ||
!store.getBlocksCollected || | ||
!store.setBlocksCollected | ||
) { | ||
return false; | ||
} | ||
if (await store.getBlocksCollected()) return true; | ||
const listBlocks = store.listBlocks.bind(store); | ||
const deleteBlocks = store.deleteBlocks.bind(store); | ||
const setBlocksCollected = store.setBlocksCollected.bind(store); | ||
const sweep = async () => { | ||
// Another request may have swept while this one waited for the lock. | ||
if (await store.getBlocksCollected?.()) return true; | ||
const commit = await actorStorage.getLatestCommit(); | ||
if (!commit) return false; | ||
const reached = new Set( | ||
(await repoBlocks(commit.cid)).map((block) => block.cid), | ||
); | ||
for (const record of await actorStorage.listAllRecords()) { | ||
reached.add(record.cid); | ||
} | ||
/** @type {string|null} */ | ||
let cursor = null; | ||
do { | ||
const page = await listBlocks(cursor, BLOCKS_PER_WAIT); | ||
const unreached = page.blocks | ||
.map((block) => block.cid) | ||
.filter((cid) => !reached.has(cid)); | ||
await deleteBlocks(unreached); | ||
cursor = page.cursor; | ||
await new Promise((resolve) => setTimeout(resolve, 0)); | ||
} while (cursor); | ||
await setBlocksCollected(true); | ||
return true; | ||
}; | ||
return ctx.withRepoLock ? ctx.withRepoLock(sweep) : sweep(); | ||
} | ||
| ||
/** | ||
* The repo as a CAR read from the block table a page at a time, as the | ||
* reference PDS reads its own, and sent as it is read: the next page is | ||
* read only once the last has gone out, so other requests run between. | ||
* | ||
* A commit that lands meanwhile drops blocks the stream may not have sent | ||
* yet. The reference's stream does the same; a consumer finds the CAR does | ||
* not hold its root's tree and asks again. | ||
* @param {string} commitCid | ||
* @returns {ReadableStream<Uint8Array>} | ||
*/ | ||
function streamRepoCar(commitCid) { | ||
const listBlocks = | ||
/** @type {NonNullable<typeof actorStorage.listBlocks>} */ ( | ||
actorStorage.listBlocks | ||
).bind(actorStorage); | ||
/** @type {string|null} */ | ||
let cursor = null; | ||
let started = false; | ||
return new ReadableStream({ | ||
async pull(controller) { | ||
if (!started) { | ||
started = true; | ||
controller.enqueue(carHeader(commitCid)); | ||
return; | ||
} | ||
const page = await listBlocks(cursor, BLOCKS_PER_WAIT); | ||
for (const block of page.blocks) controller.enqueue(carBlock(block)); | ||
cursor = page.cursor; | ||
if (!cursor) controller.close(); | ||
}, | ||
}); | ||
} | ||
| ||
/** | ||
| @@ -261,12 +374,22 @@ export function createSyncHandlers(ctx) { | ||
const decoded = cborDecode(commitData); | ||
if (decoded.data) { | ||
const mstRootCid = cidToString(decoded.data); | ||
// Include all MST blocks (proof path) | ||
const mstBlocks = await collectMstBlocks(mstRootCid, async (cid) => { | ||
return await actorStorage.getBlock(cid); | ||
// The proof is the path from the root to the key: the nodes a | ||
// lookup reads. | ||
/** @type {string[]} */ | ||
const path = []; | ||
/** @param {string} cid */ | ||
const getBlock = (cid) => { | ||
path.push(cid); | ||
return actorStorage.getBlock(cid); | ||
}; | ||
// A lookup only reads: it never calls putBlock. | ||
const reader = /** @type {import('../mst.js').MstStore} */ ({ | ||
getBlock, | ||
}); | ||
for (const block of mstBlocks) { | ||
await addBlock(block.cid); | ||
await mstGet(mstRootCid, `${collection}/${rkey}`, reader); | ||
for (const cid of path) { | ||
await addBlock(cid); | ||
} | ||
} | ||
} | ||
packages/core/src/loader.js+3−0
| @@ -70,6 +70,9 @@ export async function loadRepositoryFromCar(carBytes, actorStorage) { | ||
for (const [cid, data] of blocks) { | ||
await actorStorage.putBlock(cid, data); | ||
} | ||
// A CAR may carry blocks its commit does not reach: the next getRepo | ||
// sweeps them before reading the table as the repo. | ||
await actorStorage.setBlocksCollected?.(false); | ||
| ||
// Set metadata | ||
await actorStorage.setDid(did); | ||
packages/core/src/mst.js+32−0
| @@ -240,6 +240,38 @@ export async function collectMstBlocks(rootCid, getBlock) { | ||
return blocks; | ||
} | ||
| ||
/** | ||
* Those of `read`, nodes of the old tree a change read, that the tree at | ||
* `root` no longer reaches. | ||
* | ||
* A changed tree is the nodes the change made and subtrees of the old one. | ||
* An old node still in it hangs off a made node, or off an old node the | ||
* change read on its way down, since a change reads every node on the path | ||
* to what it changes. So a walk from `root` through only those two sets | ||
* reaches every read node still in the tree. | ||
* @param {string|null} root | ||
* @param {Map<string, Uint8Array>} made - Nodes the change made, by CID | ||
* @param {Map<string, Uint8Array>} read | ||
* @returns {string[]} | ||
*/ | ||
export function unreachedNodes(root, made, read) { | ||
const reached = new Set(); | ||
const next = root ? [root] : []; | ||
while (next.length) { | ||
const cid = /** @type {string} */ (next.pop()); | ||
if (reached.has(cid)) continue; | ||
reached.add(cid); | ||
const data = made.get(cid) ?? read.get(cid); | ||
if (!data) continue; | ||
const node = cborDecode(data); | ||
if (node.l) next.push(cidToString(node.l)); | ||
for (const entry of node.e || []) { | ||
if (entry.t) next.push(cidToString(entry.t)); | ||
} | ||
} | ||
return [...read.keys()].filter((cid) => !reached.has(cid)); | ||
} | ||
| ||
/** | ||
* Get MST proof blocks for a specific record | ||
* Returns all blocks needed to verify the record exists in the MST | ||
packages/core/src/pds.js+21−0
| @@ -44,6 +44,7 @@ import { | ||
mstGet, | ||
mstInvert, | ||
mstUpdate, | ||
unreachedNodes, | ||
} from './mst.js'; | ||
import { mintDpopNonce, parseDpopProof } from './oauth.js'; | ||
import { createPasskeyCeremony } from './passkey.js'; | ||
| @@ -392,6 +393,7 @@ export class PersonalDataServer { | ||
actorStorage: this.actorStorage, | ||
getDid: () => this.getDid(), | ||
isDeactivated: () => this._migration.isDeactivated(), | ||
withRepoLock: (fn) => this.withRepoLock(fn), | ||
}); | ||
| ||
this._blob = createBlobHandlers({ | ||
| @@ -1451,6 +1453,9 @@ export class PersonalDataServer { | ||
return { rev: prev.rev, cid: prev.cid, changed: false }; | ||
} | ||
| ||
// The old tree stays behind, unread: the next getRepo sweeps it. | ||
await this.actorStorage.setBlocksCollected?.(false); | ||
| ||
// Every node of the rebuilt tree is kept. A commit stores only the nodes | ||
// it created because the rest are already known good; here the old tree | ||
// is what is in doubt, so nothing is pruned against it. | ||
| @@ -1579,6 +1584,7 @@ export class PersonalDataServer { | ||
) { | ||
// Previous commit block is unavailable: fall back to a full rebuild | ||
// rather than silently starting a fresh tree | ||
await this.actorStorage.setBlocksCollected?.(false); | ||
mstRoot = await buildMst( | ||
await this.actorStorage.listAllRecords(), | ||
mstStore.putBlock, | ||
| @@ -1737,6 +1743,21 @@ export class PersonalDataServer { | ||
await this.actorStorage.putCommit(seq, commitCid, rev, prev?.cid || null); | ||
await this.actorStorage.putEvent(seq, did, commitCid, eventEncoded); | ||
| ||
// What this commit left unreachable goes, as the reference PDS drops a | ||
// commit's removed CIDs: the old tree's nodes it replaced, the records it | ||
// replaced or deleted that no other record shares, and the previous | ||
// commit. Kept, a deleted record stays readable by its CID, and the | ||
// block table cannot be read as the repo (see getRepo). | ||
if (this.actorStorage.deleteBlocks && this.actorStorage.recordCidsAmong) { | ||
const replaced = [...new Set(opPrev.values())]; | ||
const shared = new Set(await this.actorStorage.recordCidsAmong(replaced)); | ||
await this.actorStorage.deleteBlocks([ | ||
...unreachedNodes(mstRoot, pending, readNodes), | ||
...replaced.filter((cid) => !shared.has(cid)), | ||
...(prev ? [prev.cid] : []), | ||
]); | ||
} | ||
| ||
// Broadcast to connected WebSocket subscribers | ||
if (this.webSocket?.broadcast) { | ||
const frame = this._firehose.formatFirehoseEvent({ | ||
packages/core/src/ports.js+13−0
| @@ -271,6 +271,19 @@ | ||
* @property {(uri: string) => Promise<void>} deleteRecord | ||
* @property {(cid: string) => Promise<Uint8Array|null>} getBlock | ||
* @property {(cid: string, data: Uint8Array) => Promise<void>} putBlock | ||
* @property {(cids: string[]) => Promise<void>} [deleteBlocks] | ||
* Optional, with the three below: a store that has them drops the blocks | ||
* each commit leaves unreachable, and getRepo reads its block table rather | ||
* than walking the tree. Absent, every block is kept. | ||
* @property {(cursor: string|null, limit: number) => Promise<{blocks: Array<{cid: string, data: Uint8Array}>, cursor: string|null}>} [listBlocks] | ||
* Every stored block, in CID order, `limit` at a time after `cursor`. | ||
* @property {(cids: string[]) => Promise<string[]>} [recordCidsAmong] | ||
* Those of `cids` some record still has as its CID. | ||
* @property {() => Promise<boolean>} [getBlocksCollected] | ||
* Whether the block table holds only what the current commit reaches: | ||
* false until the first sweep, and again after the tree is rebuilt or a | ||
* repo imported. | ||
* @property {(collected: boolean) => Promise<void>} [setBlocksCollected] | ||
* @property {() => Promise<{seq: number, cid: string, rev: string}|null>} getLatestCommit | ||
* @property {(seq: number, cid: string, rev: string, prev: string|null) => Promise<void>} putCommit | ||
* @property {(cursor: number|null, limit: number) => Promise<{events: EventResult[], cursor: number|null}>} getEvents | ||
packages/core/src/repo.js+33−20
| @@ -522,38 +522,51 @@ export async function getKeyDepth(key) { | ||
* @returns {Uint8Array} CAR file bytes | ||
*/ | ||
export function buildCarFile(rootCid, blocks) { | ||
const parts = []; | ||
return concatBytes([carHeader(rootCid), ...blocks.map(carBlock)]); | ||
} | ||
| ||
// Header: { version: 1, roots: [...] }. Most callers have a single root; | ||
// permissioned repos (proposal 0016) declare two — a signed commit and an | ||
// index — so an array is accepted as well. | ||
/** | ||
* A CAR file's header, which a stream of {@link carBlock}s follows. | ||
* | ||
* Header: { version: 1, roots: [...] }. Most callers have a single root; | ||
* permissioned repos (proposal 0016) declare two — a signed commit and an | ||
* index — so an array is accepted as well. | ||
* @param {string|string[]} rootCid | ||
* @returns {Uint8Array} | ||
*/ | ||
export function carHeader(rootCid) { | ||
const rootCids = Array.isArray(rootCid) ? rootCid : [rootCid]; | ||
const header = cborEncodeDagCbor({ | ||
version: 1, | ||
roots: rootCids.map((cid) => new CID(cidToBytes(cid))), | ||
}); | ||
parts.push(varint(header.length)); | ||
parts.push(header); | ||
| ||
// Blocks: varint(len) + cid + data | ||
for (const block of blocks) { | ||
const cidBytes = cidToBytes(block.cid); | ||
const blockLen = cidBytes.length + block.data.length; | ||
parts.push(varint(blockLen)); | ||
parts.push(cidBytes); | ||
parts.push(block.data); | ||
} | ||
return concatBytes([varint(header.length), header]); | ||
} | ||
| ||
/** | ||
* One block of a CAR file: varint(len) + cid + data. | ||
* @param {{cid: string, data: Uint8Array}} block | ||
* @returns {Uint8Array} | ||
*/ | ||
export function carBlock(block) { | ||
const cidBytes = cidToBytes(block.cid); | ||
const blockLen = cidBytes.length + block.data.length; | ||
return concatBytes([varint(blockLen), cidBytes, block.data]); | ||
} | ||
| ||
// Concatenate all parts | ||
/** | ||
* @param {Uint8Array[]} parts | ||
* @returns {Uint8Array} | ||
*/ | ||
function concatBytes(parts) { | ||
const totalLen = parts.reduce((sum, p) => sum + p.length, 0); | ||
const car = new Uint8Array(totalLen); | ||
const out = new Uint8Array(totalLen); | ||
let offset = 0; | ||
for (const part of parts) { | ||
car.set(part, offset); | ||
out.set(part, offset); | ||
offset += part.length; | ||
} | ||
| ||
return car; | ||
return out; | ||
} | ||
| ||
/** | ||