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]]); }); });