festipod/backend/services/keyed-serializer.test.js

const { test, expect, describe } = require('bun:test');
const { createKeyedSerializer, SerializedWorkTimeout, DEFAULT_TIMEOUT_MS } = require('./keyed-serializer');

const tick = ms => new Promise(resolve => setTimeout(resolve, ms));

describe('createKeyedSerializer — one work at a time per key', () => {
  test('works for one key never overlap and run in submission order, even when the first is slower', async () => {
    const serialize = createKeyedSerializer();
    const trace = [];
    const work = (name, ms) => async () => {
      trace.push(`start ${name}`);
      await tick(ms);
      trace.push(`end ${name}`);
      return name;
    };
    const results = await Promise.all([serialize('E', work('first', 30)), serialize('E', work('second', 1))]);
    expect(results).toEqual(['first', 'second']);
    expect(trace).toEqual(['start first', 'end first', 'start second', 'end second']);
  });

  test('different keys run concurrently', async () => {
    const serialize = createKeyedSerializer();
    const trace = [];
    const work = (name, ms) => async () => {
      trace.push(`start ${name}`);
      await tick(ms);
      trace.push(`end ${name}`);
    };
    await Promise.all([serialize('E1', work('a', 20)), serialize('E2', work('b', 1))]);
    expect(trace.slice(0, 2).sort()).toEqual(['start a', 'start b']);
    expect(trace.indexOf('end b')).toBeLessThan(trace.indexOf('end a'));
  });

  test('a failing work rejects its own caller but does not block the next one for its key', async () => {
    const serialize = createKeyedSerializer();
    const first = serialize('E', async () => {
      throw new Error('pod down');
    });
    const second = serialize('E', async () => 'ran');
    await expect(first).rejects.toThrow('pod down');
    await expect(second).resolves.toBe('ran');
  });

  test('an idle key holds no memory', async () => {
    const serialize = createKeyedSerializer();
    await Promise.all([serialize('E1', async () => {}), serialize('E2', async () => {})]);
    await tick(0);
    expect(serialize.pending()).toBe(0);
  });

  test('THE RACE IT CLOSES: the decision is taken on a read made when acting, so the newest state wins', async () => {
    // Two notifications about one event: it was public, then made private. Each handler reads the
    // event then acts. Unserialized, the first handler's read (public) can precede the second's while
    // its action lands last: grant-after-revoke. Serialized with the read INSIDE, the last action
    // follows a read of the newest state.
    const scenario = async runFor => {
      let podState = 'public';
      const acl = [];
      const handler = actDelay => async () => {
        const seen = podState; // the read, taken when acting
        await tick(actDelay); // the pod's latency before the action lands
        acl.push(seen === 'public' ? 'grant' : 'revoke');
      };
      const first = runFor(handler(30)); // slow: its action lands late
      await tick(5); // the first handler has read "public" and is acting
      podState = 'private'; // the owner switches the event; its notification arrives
      const second = runFor(handler(1));
      await Promise.all([first, second]);
      return acl[acl.length - 1];
    };
    // Unserialized (the baseline): the stale grant lands last.
    expect(await scenario(work => work())).toBe('grant');
    // Serialized per object: the second handler reads after the first has acted.
    const serialize = createKeyedSerializer();
    expect(await scenario(work => serialize('E', work))).toBe('revoke');
  });
});

describe('createKeyedSerializer — bounded: a stalled work never holds its key forever', () => {
  test('a work that never settles is abandoned after the timeout: its caller is told, the next work runs', async () => {
    const serialize = createKeyedSerializer({ timeoutMs: 50 });
    const stalled = serialize('E', () => new Promise(() => {})); // a pod request that never answers
    let ran = false;
    const next = serialize('E', async () => {
      ran = true;
    });
    await expect(stalled).rejects.toBeInstanceOf(SerializedWorkTimeout);
    await next;
    expect(ran).toBe(true);
    await tick(0);
    expect(serialize.pending()).toBe(0);
  });

  test('the abandoned work sees its signal aborted, so it can give up before writing', async () => {
    const serialize = createKeyedSerializer({ timeoutMs: 20 });
    let wrote = false;
    const slow = serialize('E', async signal => {
      await tick(60); // the read that took too long
      if (signal.aborted) return;
      wrote = true;
    });
    await expect(slow).rejects.toBeInstanceOf(SerializedWorkTimeout);
    await tick(80);
    expect(wrote).toBe(false);
  });

  test('a work within the limit is untouched, and the default limit is 30 s', async () => {
    const serialize = createKeyedSerializer({ timeoutMs: 100 });
    await expect(serialize('E', async signal => (signal.aborted ? 'aborted' : 'done'))).resolves.toBe('done');
    expect(DEFAULT_TIMEOUT_MS).toBe(30000);
  });
});

describe('createKeyedSerializer — a write that lands after its timeout is re-decided', () => {
  test('onLateSettle runs once the abandoned work settles, and its re-decision runs after the newer work', async () => {
    const serialize = createKeyedSerializer({ timeoutMs: 20 });
    const order = [];
    const redecided = new Promise(resolve => {
      const slow = serialize(
        'E',
        async () => {
          await tick(60); // a write sent before the timeout, landing after it
          order.push('late write landed');
          return { wroteLate: true };
        },
        { onLateSettle: value => value.wroteLate && serialize('E', async () => resolve(order.push('re-decision'))) }
      );
      slow.catch(() => order.push('timed out'));
      serialize('E', async () => order.push('newer work'));
    });
    await redecided;
    expect(order).toEqual(['timed out', 'newer work', 'late write landed', 're-decision']);
  });

  test('onLateSettle is not called for a work that finished in time', async () => {
    const serialize = createKeyedSerializer({ timeoutMs: 100 });
    let called = false;
    await serialize('E', async () => {}, { onLateSettle: () => (called = true) });
    await tick(5);
    expect(called).toBe(false);
  });
});

describe('createKeyedSerializer — onLateSettle receives what the late work resolved', () => {
  test('the value of a late work is handed over (undefined when it rejected), so the caller can tell a late write', async () => {
    const serialize = createKeyedSerializer({ timeoutMs: 20 });
    const seen = [];
    const done = new Promise(resolve => {
      serialize('A', async () => (await tick(50), { wroteLate: false }), { onLateSettle: v => seen.push(['A', v]) }).catch(() => {});
      serialize('B', async () => { await tick(50); throw new Error('pod down'); }, { onLateSettle: v => { seen.push(['B', v]); resolve(); } }).catch(() => {});
    });
    await done;
    await tick(10);
    expect(seen).toEqual([['A', { wroteLate: false }], ['B', undefined]]);
  });
});