chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
/**
* The lazy object store: a repository's pack indexes in memory, object bytes
* fetched with ranged blob reads as a caller asks for them. Both the
* repository browser and the write guard's ancestry check read through it,
* so neither holds a bundle chain whole.
*/
import { applyDelta, inflateAt, readEntryHeader } from './objects.js';
import { BASE_EXTERNAL, BASE_NONE, decodePackIndex } from './pack-index.js';
import { concatChunks, indexPartsOf } from './record.js';
/**
* Two ranged reads close together cost more than one read spanning the gap;
* reads farther apart than this stay separate.
*/
const COALESCE_GAP_BYTES = 16 * 1024;
/**
* What both read modes offer the browsing methods. meta answers from the
* index (or the in-memory map) without fetching; get and getMany
* materialize objects; peek answers only from what is already materialized.
* @typedef {Object} ObjectStore
* @property {(sha: string) => {type: string, size: number}|null} meta
* @property {(sha: string) => import('./objects.js').GitObject|null} peek
* @property {(sha: string) => Promise<import('./objects.js').GitObject|null>} get
* @property {(shas: Iterable<string>) => Promise<void>} getMany
* @property {() => Promise<string[]>} prefetchCommits - materialize every
* commit and tag in the chain, and name them
*/
/**
* Merge sorted byte ranges, spanning gaps smaller than the coalesce
* threshold.
* @param {Array<{start: number, end: number}>} ranges
* @returns {Array<{start: number, end: number}>}
*/
function mergeRanges(ranges) {
const sorted = [...ranges].sort((a, b) => a.start - b.start);
/** @type {Array<{start: number, end: number}>} */
const merged = [];
for (const range of sorted) {
const last = merged[merged.length - 1];
if (last && range.start <= last.end + COALESCE_GAP_BYTES) {
last.end = Math.max(last.end, range.end);
} else {
merged.push({ ...range });
}
}
return merged;
}
/**
* The lazy read mode: object indexes in memory, object bytes fetched with
* ranged blob reads as screens ask for them.
* @param {import('./record.js').GitRepoRecord} repo
* @param {(cid: string) => Promise<Uint8Array>} fetchBlob
* @param {(cid: string, start: number, end: number) => Promise<Uint8Array>} fetchBlobRange
* @returns {Promise<ObjectStore>}
*/
export async function createLazyStore(repo, fetchBlob, fetchBlobRange) {
const bundles = await Promise.all(
repo.bundles.map(async (bundle) => {
const indexBlobs = indexPartsOf(bundle);
const index = decodePackIndex(
concatChunks(
await Promise.all(
indexBlobs.map((part) => fetchBlob(part.ref.$link)),
),
),
);
/** @type {number[]} */
const chunkStarts = [];
let total = 0;
for (const part of bundle.parts) {
chunkStarts.push(total);
total += part.size;
}
/** @type {Map<number, number>} */
const byOffset = new Map();
index.entries.forEach((entry, i) => {
byOffset.set(entry.offset, i);
});
return {
parts: bundle.parts,
index,
chunkStarts,
packLength: total - index.packStart,
byOffset,
};
}),
);
/** @type {Map<string, {bundle: number, entry: number}>} */
const locations = new Map();
bundles.forEach((bundle, b) => {
bundle.index.entries.forEach((entry, i) => {
if (!locations.has(entry.sha)) {
locations.set(entry.sha, { bundle: b, entry: i });
}
});
});
/** Materialized objects, by id. */
/** @type {Map<string, import('./objects.js').GitObject>} */
const objects = new Map();
/**
* A pack entry's byte extent inside its bundle. The compressed stream
* runs to the next entry, or to the pack's 20-byte trailer for the last
* one.
* @param {number} b
* @param {number} i
*/
function extent(b, i) {
const { index, packLength } = bundles[b];
const next = index.entries[i + 1]?.offset ?? packLength - 20;
return {
start: index.packStart + index.entries[i].offset,
end: index.packStart + next,
};
}
/**
* Read one byte range out of a bundle's chunked parts. A server that
* answers 200 sends the whole chunk; the slice below keeps that correct.
* @param {number} b
* @param {number} start
* @param {number} end
* @returns {Promise<Uint8Array>}
*/
async function fetchBundleRange(b, start, end) {
const { parts, chunkStarts } = bundles[b];
/** @type {Promise<Uint8Array>[]} */
const pieces = [];
for (let c = 0; c < parts.length; c++) {
const chunkStart = chunkStarts[c];
const chunkEnd = chunkStart + parts[c].size;
if (chunkEnd <= start) continue;
if (chunkStart >= end) break;
pieces.push(
fetchBlobRange(
parts[c].ref.$link,
Math.max(start, chunkStart) - chunkStart,
Math.min(end, chunkEnd) - chunkStart,
),
);
}
return concatChunks(await Promise.all(pieces));
}
/**
* The entry plus its in-bundle delta chain, deepest base last. The chain
* stops at an already-materialized base and at an external one, whose id
* only the entry bytes hold.
* @param {number} b
* @param {number} i
* @returns {number[]}
*/
function chainEntries(b, i) {
const { index, byOffset } = bundles[b];
const chain = [i];
let current = i;
for (;;) {
const base = index.entries[current].base;
if (base === BASE_NONE || base === BASE_EXTERNAL) return chain;
const baseEntry = byOffset.get(base);
if (baseEntry === undefined) {
throw new Error('pack index names a delta base it does not list');
}
if (objects.has(index.entries[baseEntry].sha)) return chain;
chain.push(baseEntry);
current = baseEntry;
}
}
/**
* Fetch and materialize the given entries of one bundle, in one coalesced
* set of ranged reads. Entries are processed in offset order, which puts
* every in-pack delta base before the entries that need it.
* @param {number} b
* @param {number[]} entryIndices
*/
async function materialize(b, entryIndices) {
const { index } = bundles[b];
const wanted = [...new Set(entryIndices)]
.filter((i) => !objects.has(index.entries[i].sha))
.sort((a, b2) => a - b2);
if (wanted.length === 0) return;
const extents = wanted.map((i) => extent(b, i));
const merged = mergeRanges(extents);
const buffers = await Promise.all(
merged.map((range) => fetchBundleRange(b, range.start, range.end)),
);
/** @param {{start: number, end: number}} range */
const bytesFor = (range) => {
for (let m = 0; m < merged.length; m++) {
if (range.start >= merged[m].start && range.end <= merged[m].end) {
return buffers[m].subarray(
range.start - merged[m].start,
range.end - merged[m].start,
);
}
}
throw new Error('entry extent missing from fetched ranges');
};
for (let w = 0; w < wanted.length; w++) {
const i = wanted[w];
const row = index.entries[i];
if (objects.has(row.sha)) continue;
const raw = bytesFor(extents[w]);
const header = readEntryHeader(raw);
const { data } = inflateAt(raw, header.dataStart);
/** @type {import('./objects.js').GitObject} */
let object;
if (header.type === 'ofs-delta' || header.type === 'ref-delta') {
const baseSha =
header.baseSha ??
bundles[b].index.entries[
/** @type {number} */ (
bundles[b].byOffset.get(
row.offset - /** @type {number} */ (header.baseRel),
)
)
].sha;
const base = objects.get(baseSha) ?? (await get(baseSha));
if (!base) {
throw new Error(`delta base ${baseSha} is not in the chain`);
}
object = { type: base.type, data: applyDelta(data, base.data) };
} else {
object = { type: header.type, data };
}
objects.set(row.sha, object);
}
}
/**
* In-flight materializations by id, so concurrent readers of one object
* share a single fetch.
* @type {Map<string, Promise<import('./objects.js').GitObject|null>>}
*/
const pending = new Map();
/** @type {ObjectStore['get']} */
function get(sha) {
const hit = objects.get(sha);
if (hit) return Promise.resolve(hit);
const location = locations.get(sha);
if (!location) return Promise.resolve(null);
let waiting = pending.get(sha);
if (!waiting) {
waiting = (async () => {
try {
await materialize(
location.bundle,
chainEntries(location.bundle, location.entry),
);
return objects.get(sha) ?? null;
} finally {
pending.delete(sha);
}
})();
pending.set(sha, waiting);
}
return waiting;
}
/** @type {ObjectStore['getMany']} */
async function getMany(shas) {
/** @type {Map<number, number[]>} */
const perBundle = new Map();
for (const sha of shas) {
if (objects.has(sha)) continue;
const location = locations.get(sha);
if (!location) continue;
const list = perBundle.get(location.bundle) ?? [];
list.push(...chainEntries(location.bundle, location.entry));
perBundle.set(location.bundle, list);
}
await Promise.all(
[...perBundle].map(([b, entries]) => materialize(b, entries)),
);
}
return {
meta(sha) {
const location = locations.get(sha);
if (!location) return null;
const row = bundles[location.bundle].index.entries[location.entry];
return { type: row.type, size: row.size };
},
peek(sha) {
return objects.get(sha) ?? null;
},
get,
getMany,
async prefetchCommits() {
/** @type {string[]} */
const shas = [];
for (const bundle of bundles) {
for (const entry of bundle.index.entries) {
if (entry.type === 'commit' || entry.type === 'tag') {
shas.push(entry.sha);
}
}
}
await getMany(shas);
return shas;
},
};
}