chadtmiller.com/pds.js

Switch repository

Loading…

// Record writes that arrive at the same time, as a CI runner sends them: a
// burst of creates, then updates and deletes of the same records. Each write
// is its own commit, and the commits must form one chain.

import { PersonalDataServer } from '@pdsjs/core';
import { parseCarFile } from '@pdsjs/core/car';
import { buildMst } from '@pdsjs/core/mst';
import { cborDecode, parseCommitBlock } from '@pdsjs/core/repo';
import { beforeEach, describe, expect, test } from 'vitest';
import {
  createMemorySharedStorage,
  createMemoryStorage,
} from './helpers/memory-storage.js';

const DID = 'did:plc:concurrenttestaaaaaaaaaa';
const PASSWORD = 'test-password';
const COLLECTION = 'dev.knotless.check';

describe('concurrent record writes', () => {
  /** @type {ReturnType<typeof createMemoryStorage>} */
  let storage;
  /** @type {PersonalDataServer} */
  let pds;
  /** @type {string} */
  let token;

  beforeEach(async () => {
    storage = createMemoryStorage();
    pds = new PersonalDataServer({
      actorStorage: /** @type {any} */ (storage),
      sharedStorage: /** @type {any} */ (createMemorySharedStorage()),
      blobs: {
        async get() {
          return null;
        },
        async put() {},
        async delete() {},
      },
      jwtSecret: 'test-secret',
      hostname: 'pds.example.com',
      password: PASSWORD,
    });

    await pds.fetch(
      new Request(`https://pds.example.com/init?did=${DID}`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          did: DID,
          privateKey: Buffer.from(
            crypto.getRandomValues(new Uint8Array(32)),
          ).toString('hex'),
          handle: 'concurrent.test',
          password: PASSWORD,
        }),
      }),
    );

    const sessionRes = await pds.fetch(
      new Request(
        'https://pds.example.com/xrpc/com.atproto.server.createSession',
        {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify({ identifier: DID, password: PASSWORD }),
        },
      ),
    );
    const sessionBody = await sessionRes.json();
    expect(sessionBody.accessJwt, 'session setup failed').toBeTruthy();
    token = sessionBody.accessJwt;
  });

  /**
   * @param {string} method
   * @param {object} body
   */
  async function write(method, body) {
    const res = await pds.fetch(
      new Request(`https://pds.example.com/xrpc/com.atproto.repo.${method}`, {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          Authorization: `Bearer ${token}`,
        },
        body: JSON.stringify({ repo: DID, collection: COLLECTION, ...body }),
      }),
    );
    return { status: res.status, body: await res.json() };
  }

  /** @param {string} status */
  const check = (status) => ({ status, createdAt: '2026-10-23T23:54:00Z' });

  /**
   * Every commit event in sequence order, decoded, with its commit block, as
   * the event carries it: a later commit drops it from storage.
   */
  function commitChain() {
    const events = storage._events.filter((e) => e.commit_cid !== '');
    return events.map((e) => {
      const evt = cborDecode(e.evt);
      const bytes = /** @type {Uint8Array} */ (
        parseCarFile(evt.blocks).blocks.get(e.commit_cid)
      );
      const { data, prev } = parseCommitBlock(bytes);
      return { seq: e.seq, evt, data, prev };
    });
  }

  /**
   * Each commit names the one before it: as `prev` in the signed block, and
   * as `since` and `prevData` on its firehose event.
   */
  async function expectContinuousChain() {
    const chain = commitChain();
    const seqs = chain.map((c) => c.seq);
    expect(new Set(seqs).size).toBe(seqs.length);
    expect([...seqs].sort((a, b) => a - b)).toEqual(seqs);
    for (let i = 1; i < chain.length; i++) {
      const before = chain[i - 1];
      const after = chain[i];
      expect(after.evt.since, `commit ${i} since`).toBe(before.evt.rev);
      expect(after.evt.prevData?.toString(), `commit ${i} prevData`).toBe(
        before.data,
      );
      expect(after.prev, `commit ${i} prev`).toBe(before.evt.commit);
    }
    const head = await storage.getLatestCommit();
    expect(head?.cid).toBe(chain[chain.length - 1].evt.commit);
    expect(chain[chain.length - 1].data).toBe(
      await buildMst(await storage.listAllRecords(), async () => {}),
    );
    return chain;
  }

  test('a burst of creates, updates and deletes commits as one chain', async () => {
    const first = Array.from({ length: 8 }, (_, i) => `check${i}`);
    const created = await Promise.all(
      first.map((rkey) =>
        write('createRecord', { rkey, record: check('running') }),
      ),
    );
    for (const res of created) expect(res.status).toBe(200);

    const second = Array.from({ length: 4 }, (_, i) => `later${i}`);
    const mixed = await Promise.all([
      ...first
        .slice(0, 5)
        .map((rkey) => write('putRecord', { rkey, record: check('success') })),
      ...first.slice(5).map((rkey) => write('deleteRecord', { rkey })),
      ...second.map((rkey) =>
        write('putRecord', { rkey, record: check('running') }),
      ),
    ]);
    for (const res of mixed) expect(res.status).toBe(200);

    const chain = await expectContinuousChain();
    expect(chain.length).toBe(first.length + mixed.length);

    for (const rkey of [...first.slice(0, 5), ...second]) {
      const held = await storage.getRecord(`at://${DID}/${COLLECTION}/${rkey}`);
      expect(held, rkey).toBeTruthy();
    }
    for (const rkey of first.slice(5)) {
      expect(
        await storage.getRecord(`at://${DID}/${COLLECTION}/${rkey}`),
      ).toBeNull();
    }
    expect((await storage.listAllRecords()).length).toBe(9);
  });

  test('concurrent creates at one key commit one create and refuse the rest', async () => {
    const results = await Promise.all(
      ['a', 'b', 'c', 'd'].map((status) =>
        write('createRecord', { rkey: 'same', record: check(status) }),
      ),
    );
    expect(results.map((res) => res.status).sort()).toEqual([
      200, 400, 400, 400,
    ]);

    const chain = await expectContinuousChain();
    expect(chain.map((c) => c.evt.ops[0].action)).toEqual(['create']);
  });
});