chadtmiller.com/pds.js

Switch repository

Loading…

// @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<object, Map<string, Promise<unknown>>>} */
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`;
}