// @pdsjs/spaces/sequence - space revisions, which only the authority assigns. // // Each repo carries its own rev. The authority also puts every accepted update // on one line for the whole space: a spaceRev that strictly increases across // writers, with prevSpaceRev naming the update before it. listRepos pages by // spaceRev, and a syncer that sees a gap in prevSpaceRev catches up from there. import { createTid, tidToMs } from '@pdsjs/core/repo'; const TID_CHARS = '234567abcdefghijklmnopqrstuvwxyz'; /** How far ahead of the clock a reported repo rev may be. */ export const FUTURE_REV_LIMIT_MS = 5 * 60 * 1000; // One allocation per space at a time: two updates must not both read the same // head and claim the same predecessor. Sound for the reason writer.js gives: a // space's authority is exactly one process. /** @type {WeakMap>>} */ const queues = new WeakMap(); /** * @param {unknown} value * @returns {value is string} */ export function isTid(value) { return ( typeof value === 'string' && value.length === 13 && [...value].every((c) => TID_CHARS.includes(c)) && TID_CHARS.indexOf(value[0]) < 16 ); } /** * Whether a repo rev claims a time too far ahead of this server's clock. * @param {string} repoRev * @param {number} [now] * @returns {boolean} */ export function isFutureRev(repoRev, now = Date.now()) { const ms = tidToMs(repoRev); return ms !== null && ms > now + FUTURE_REV_LIMIT_MS; } /** * Record a writer's update and give it the next space revision. * * An update at or behind the writer's last accepted repo rev is not * resequenced: the answer is null, and the caller forwards nothing. * * @param {import('@pdsjs/core/ports').SpaceStoragePort} spaceStorage * @param {{space: string, did: string, repoRev: string, data: Uint8Array}} update * @returns {Promise<{spaceRev: string, prevSpaceRev: string|null}|null>} */ export function recordWriterUpdate( spaceStorage, { space, did, repoRev, data }, ) { let inStorage = queues.get(spaceStorage); if (!inStorage) { inStorage = new Map(); queues.set(spaceStorage, inStorage); } const run = async () => { const current = await spaceStorage.getSpaceWriter(space, did); if (current && current.repoRev >= repoRev) return null; const prevSpaceRev = await spaceStorage.getSpaceRevHead(space); const spaceRev = nextTid(prevSpaceRev); await spaceStorage.putSpaceWriter(space, did, repoRev, data, spaceRev); return { spaceRev, prevSpaceRev }; }; const prior = inStorage.get(space) ?? Promise.resolve(); const next = prior.then(run, run); inStorage.set( space, next.then( () => undefined, () => undefined, ), ); return next; } /** * A TID later than `after`. A writer recorded before space revisions existed * is sequenced at its repo rev, which another clock produced, so the clock * here alone does not guarantee that. * * @param {string|null} after * @returns {string} */ export function nextTid(after) { const tid = createTid(); if (after === null || tid > after) return tid; let micros = 0; for (const c of after.slice(0, 11)) micros = micros * 32 + TID_CHARS.indexOf(c); micros += 1; let out = ''; for (let i = 0; i < 11; i++) { out = TID_CHARS[micros % 32] + out; micros = Math.floor(micros / 32); } return `${out}22`; }