diff --git a/src/session.ts b/src/session.ts index e2e85bd..87075d9 100644 --- a/src/session.ts +++ b/src/session.ts @@ -34,6 +34,7 @@ export class StanddownSession { readonly #maxAuditEntries: number; readonly #selfExemptionScope: 'policy' | 'session'; readonly #readOnlyAuditLog: AuditEntry[] = []; + #stateLock: Promise = Promise.resolve(); constructor( store: StateStore, @@ -211,11 +212,33 @@ export class StanddownSession { } async exportAuditLog(): Promise { - const state = await this.#loadState(); - return trimAuditLog( - [...state.auditLog, ...this.#readOnlyAuditLog], - this.#maxAuditEntries, - ).map(cloneAuditEntry); + return this.#serialize(async () => { + const state = await this.#loadState(); + return trimAuditLog( + [...state.auditLog, ...this.#readOnlyAuditLog], + this.#maxAuditEntries, + ).map(cloneAuditEntry); + }); + } + + /** + * Serializes every state read-modify-write onto a single FIFO chain. + * #withState and #failClosedWithAudit each do an async load -> mutate -> save; + * without this, two overlapping evaluations both load the pre-write snapshot + * and the second save clobbers the first, so a just-recorded stand-down can be + * lost. The adapter's own hooks coalesce via a `pending` flag, but external + * callers driving evaluate()/ingest() concurrently are not otherwise + * serialized, so this lock is what makes them safe. It also gives + * read-after-write consistency: a shouldStandDown queued after an ingest reads + * that ingest's committed state. + */ + #serialize(task: () => Promise): Promise { + const run = this.#stateLock.then(task, task); + this.#stateLock = run.then( + () => undefined, + () => undefined, + ); + return run; } async #withState( @@ -228,44 +251,46 @@ export class StanddownSession { advertiserHost?: string, opts: { persist?: boolean } = {}, ): Promise { - let state: StanddownState; + return this.#serialize(async () => { + let state: StanddownState; - try { - state = await this.#loadState(); - } catch { - return failClosedDecision('store-error'); - } + try { + state = await this.#loadState(); + } catch { + return failClosedDecision('store-error'); + } - const result = fn(state); + const result = fn(state); - if (this.#auditLog) { - const entry = auditEntry({ - time: now, - action, - advertiserHost: - advertiserHost ?? result.detection?.strongest?.advertiserHost, - detection: result.detection, - decision: result.decision, - }); + if (this.#auditLog) { + const entry = auditEntry({ + time: now, + action, + advertiserHost: + advertiserHost ?? result.detection?.strongest?.advertiserHost, + detection: result.detection, + decision: result.decision, + }); + + if (opts.persist === false) { + appendAuditEntry(this.#readOnlyAuditLog, entry, this.#maxAuditEntries); + } else { + appendAuditEntry(state.auditLog, entry, this.#maxAuditEntries); + } + } if (opts.persist === false) { - appendAuditEntry(this.#readOnlyAuditLog, entry, this.#maxAuditEntries); - } else { - appendAuditEntry(state.auditLog, entry, this.#maxAuditEntries); + return result.decision; } - } - - if (opts.persist === false) { - return result.decision; - } - try { - await this.#store.save(state); - } catch { - return failClosedDecision('store-error'); - } + try { + await this.#store.save(state); + } catch { + return failClosedDecision('store-error'); + } - return result.decision; + return result.decision; + }); } async #failClosedWithAudit( @@ -281,25 +306,27 @@ export class StanddownSession { return decision; } - try { - const state = await this.#loadState(); - appendAuditEntry( - state.auditLog, - auditEntry({ - time: now, - action, - advertiserHost, - detection, - decision, - }), - this.#maxAuditEntries, - ); - await this.#store.save(state); - } catch { - return failClosedDecision('store-error'); - } + return this.#serialize(async () => { + try { + const state = await this.#loadState(); + appendAuditEntry( + state.auditLog, + auditEntry({ + time: now, + action, + advertiserHost, + detection, + decision, + }), + this.#maxAuditEntries, + ); + await this.#store.save(state); + } catch { + return failClosedDecision('store-error'); + } - return decision; + return decision; + }); } async #loadState(): Promise { diff --git a/tests/session.test.ts b/tests/session.test.ts index 717647b..d6c7141 100644 --- a/tests/session.test.ts +++ b/tests/session.test.ts @@ -415,6 +415,33 @@ describe('StanddownSession', () => { expect(decision.standDown).toBe(true); expect(decision.degraded).toBeUndefined(); }); + + it('serializes concurrent ingests so neither loses the other write', async () => { + // A store that yields on every load/save, forcing two overlapping + // evaluations to interleave. Without the session's serialization both would + // load the same pre-write snapshot and the second save would clobber the + // first, dropping one host's stand-down record. + const store = new DelayingStore(); + const session = new StanddownSession(store); + + const [a, b] = await Promise.all([ + session.ingest( + { url: 'https://merchant-a.example/?cjevent=aaa', now: 0 }, + [cjPolicy], + ), + session.ingest( + { url: 'https://merchant-b.example/?cjevent=bbb', now: 0 }, + [cjPolicy], + ), + ]); + + expect(a).toMatchObject({ standDown: true, policyId: 'cj' }); + expect(b).toMatchObject({ standDown: true, policyId: 'cj' }); + + const state = await store.load(); + expect(state?.sessions['merchant-a.example']).toBeDefined(); + expect(state?.sessions['merchant-b.example']).toBeDefined(); + }); }); class FailingLoadStore implements StateStore { @@ -437,6 +464,24 @@ class FailingSaveStore implements StateStore { } } +class DelayingStore implements StateStore { + readonly #inner: MemoryStateStore; + + constructor(initialState?: StanddownState) { + this.#inner = new MemoryStateStore(initialState); + } + + async load(): Promise { + await Promise.resolve(); + return this.#inner.load(); + } + + async save(state: StanddownState): Promise { + await Promise.resolve(); + return this.#inner.save(state); + } +} + class CountingStore implements StateStore { readonly #inner: MemoryStateStore; saveCount = 0;