diff --git a/packages/agent/test/swm-publish-profile-mutex.test.ts b/packages/agent/test/swm-publish-profile-mutex.test.ts new file mode 100644 index 0000000000..3d602f86c8 --- /dev/null +++ b/packages/agent/test/swm-publish-profile-mutex.test.ts @@ -0,0 +1,201 @@ +/** + * `publishProfileTail` serialization tests — PR #700 round-2 mutex. + * + * The mutex lives at `packages/agent/src/dkg-agent.ts:4085-4101`. + * Every `publishProfile()` caller chains onto the prior tail's + * promise via `.catch(swallow).then(() => publishProfileImpl())`, + * so the four production callers (startup, heartbeat, key + * rotation, key revocation) can never race on + * `ProfileManager.currentKcId` or the agent registry triples. + * + * #716 review-consolidation audit flagged the mutex itself as + * "Critical — never proven to serialize concurrent calls". The + * helper code is tiny but the correctness gate it provides is the + * widest in PR #700's concurrency-critical fan-out (rotate / revoke + * mid-heartbeat would otherwise rewrite the same registry triples + * concurrently). This file pins: + * + * 1. **Serialization** — N concurrent `publishProfile()` calls + * invoke `publishProfileImpl` exactly once each and never + * overlap. + * + * 2. **Error isolation** — a failing `publishProfileImpl` does + * not poison the tail; the next caller still gets to run. + * + * 3. **Return propagation** — each caller's awaited promise + * resolves to its OWN `publishProfileImpl` return value, not + * a sibling's. + */ +import { afterEach, describe, expect, it } from 'vitest'; +import { MockChainAdapter } from '@origintrail-official/dkg-chain'; +import { DKGAgent } from '../src/index.js'; + +/** + * Reach into the agent to replace the private `publishProfileImpl` + * with a controllable stub. The stub records: + * - the order of calls + * - the maximum number of concurrent invocations observed + * + * If the mutex is honoured, `maxConcurrency` stays at 1 across N + * overlapping `publishProfile()` calls. If a future refactor drops + * the tail-chain or breaks the `.then(...)` ordering, concurrency + * spikes to N and the assertion below fires loud. + */ +interface PublishProfileInternals { + publishProfileImpl(): Promise; + publishProfileTail: Promise; +} + +async function bootAgent(): Promise<{ agent: DKGAgent; internals: PublishProfileInternals }> { + const agent = await DKGAgent.create({ + name: 'PublishProfileMutexTest', + chainAdapter: new MockChainAdapter(), + }); + const internals = agent as unknown as PublishProfileInternals; + return { agent, internals }; +} + +describe('DKGAgent.publishProfile — tail-chain mutex serialization (PR #700 round 2)', () => { + let agent: DKGAgent | null = null; + afterEach(async () => { + if (agent) { + await agent.stop().catch(() => undefined); + agent = null; + } + }); + + it('serializes N concurrent publishProfile() calls — max-in-flight stays at 1', async () => { + const boot = await bootAgent(); + agent = boot.agent; + const internals = boot.internals; + + // Per-call stub that sleeps a controllable amount. We pick a + // short delay (15 ms) — enough that all calls overlap in the + // event loop if they were ever allowed to run concurrently, but + // short enough that the test stays well under any reasonable + // CI timeout even with `maxWorkers: 1` agent test config. + const SLEEP_MS = 15; + const N = 5; + let inFlight = 0; + let maxConcurrency = 0; + const callOrder: number[] = []; + let nextCallId = 0; + + internals.publishProfileImpl = async function stubImpl(): Promise { + const myId = nextCallId++; + callOrder.push(myId); + inFlight++; + try { + if (inFlight > maxConcurrency) maxConcurrency = inFlight; + await new Promise((resolve) => setTimeout(resolve, SLEEP_MS)); + return { ok: true, callId: myId }; + } finally { + inFlight--; + } + } as PublishProfileInternals['publishProfileImpl']; + + // Fire N concurrent publishProfile() calls. Each must wait for + // the prior tail to settle. + const promises = Array.from({ length: N }, () => agent!.publishProfile()); + const results = await Promise.all(promises); + + // The whole point: never more than 1 publishProfileImpl in flight. + expect(maxConcurrency).toBe(1); + + // And: every call ran exactly once, in submission order + // (the chain is FIFO — each new caller appends to the tail). + expect(callOrder).toEqual([0, 1, 2, 3, 4]); + + // Each caller awaits ITS OWN run — not a sibling's return value + // — even though they're chained. + expect(results.map((r) => (r as { callId: number }).callId)).toEqual([0, 1, 2, 3, 4]); + }); + + it('error isolation: a failing publishProfileImpl does not poison the tail for subsequent callers', async () => { + // Pinned by the explicit `.catch(swallow)` in the mutex body. + // Without that, a single rejected publish would wedge every + // future publishProfile() in `await publishProfileTail` because + // the tail promise would stay rejected forever (and `.then()` on + // a rejected promise without a `.catch` propagates). + const boot = await bootAgent(); + agent = boot.agent; + const internals = boot.internals; + + let invocation = 0; + internals.publishProfileImpl = async function stubImpl(): Promise { + const myInvocation = invocation++; + if (myInvocation === 1) { + throw new Error('synthetic failure to test error isolation'); + } + return { ok: true, invocation: myInvocation }; + } as PublishProfileInternals['publishProfileImpl']; + + // Codex review feedback: collecting the three promises with + // `Promise.allSettled` attaches a handler to each one BEFORE + // awaiting. Without it, the middle promise can reject between + // the call site at line `agent.publishProfile()` and the + // subsequent `await expect(...).rejects.toThrow(...)` call, + // surfacing as an unhandled rejection under Vitest/Node and + // making the test flaky on busy CI runners. + const settled = await Promise.allSettled([ + agent.publishProfile(), + agent.publishProfile(), + agent.publishProfile(), + ]); + + expect(settled[0].status).toBe('fulfilled'); + expect((settled[0] as PromiseFulfilledResult).value).toEqual({ + ok: true, + invocation: 0, + }); + + expect(settled[1].status).toBe('rejected'); + expect((settled[1] as PromiseRejectedResult).reason).toBeInstanceOf(Error); + expect(((settled[1] as PromiseRejectedResult).reason as Error).message).toMatch( + /synthetic failure to test error isolation/, + ); + + // The crucial assertion: the third call must still run even + // though the second tail rejected. The `.catch()` in + // `publishProfile()` swallows the prior error before chaining. + expect(settled[2].status).toBe('fulfilled'); + expect((settled[2] as PromiseFulfilledResult).value).toEqual({ + ok: true, + invocation: 2, + }); + + // And: the next *fresh* call (after the bad ones settled) also + // succeeds — proves the tail is healthy long-term. + await expect(agent.publishProfile()).resolves.toEqual({ ok: true, invocation: 3 }); + }); + + it('returns each caller their OWN publishProfileImpl result (return-value isolation across chained tails)', async () => { + // Subtle correctness boundary: the chained-promise shape could + // accidentally collapse N callers to receiving the same return + // value if the mutex was implemented as a single shared promise + // (e.g. `return this.publishProfileTail`). The actual + // implementation captures `run` per-caller and returns it, so + // each awaited promise resolves to its OWN implementation call. + const boot = await bootAgent(); + agent = boot.agent; + const internals = boot.internals; + + let counter = 0; + internals.publishProfileImpl = async function stubImpl(): Promise { + const myId = counter++; + // Stagger so the chained-promise shape can't accidentally + // resolve all callers to the last result via shared state. + await new Promise((resolve) => setTimeout(resolve, 5)); + return { uniqueResult: `result-${myId}` }; + } as PublishProfileInternals['publishProfileImpl']; + + const results = await Promise.all([ + agent.publishProfile(), + agent.publishProfile(), + agent.publishProfile(), + ]); + + expect(results.map((r) => (r as { uniqueResult: string }).uniqueResult)) + .toEqual(['result-0', 'result-1', 'result-2']); + }); +}); diff --git a/packages/agent/test/swm-sender-key-parallel-fanout.test.ts b/packages/agent/test/swm-sender-key-parallel-fanout.test.ts index 0e1b6839a0..e32d6772bb 100644 --- a/packages/agent/test/swm-sender-key-parallel-fanout.test.ts +++ b/packages/agent/test/swm-sender-key-parallel-fanout.test.ts @@ -29,6 +29,7 @@ import { SWM_SENDER_KEY_PACKAGE_VERSION, SWM_SENDER_KEY_PACKAGE_ACK_TYPE, encodeSwmSenderKeyPackageAck, + decodeSwmSenderKeyPackage, generateWorkspaceRecipientEncryptionKey, type OperationContext, } from '@origintrail-official/dkg-core'; @@ -245,4 +246,238 @@ describe('createAndDistributeSwmSenderKeyEpoch: parallel fanout latency', () => }); expect(state).toBeDefined(); }); + + it('1-of-N partial fail: throw cites only the agent whose keys all failed; non-failed peers do not appear in the error', async () => { + // The aggregation logic at `dkg-agent.ts:5998-6042` separates per- + // agent outcomes: a fatal agent is one where EVERY key failed. The + // throw must: + // - include exactly the fatal agent(s) — not the successful ones + // - count them correctly ("N agent(s)" in the message) + // - leave the other recipients' deliveries observable as + // successes (e.g. their epoch state) + // + // This pins the "M of N agents fatal" branch the existing all-fail + // and all-soft tests don't reach. + const boot = await bootAgent(); + agent = boot.agent; + const internals = boot.internals; + + const recipientA = makeFakeRecipient(); + const recipientB = makeFakeRecipient(); // <-- this one's keys will fail + const recipientC = makeFakeRecipient(); + + // Messenger returns ACCEPTED for A and C, REJECTED for B. We + // discriminate on the recipient peerId since each fake recipient + // has a deterministic peerId derived from its agentAddress. + installStubMessenger(internals, async (peerId): Promise => { + const acceptedEnvelope = encodeSwmSenderKeyPackageAck({ + version: SWM_SENDER_KEY_PACKAGE_VERSION, + type: SWM_SENDER_KEY_PACKAGE_ACK_TYPE, + accepted: true, + }); + const rejectedEnvelope = encodeSwmSenderKeyPackageAck({ + version: SWM_SENDER_KEY_PACKAGE_VERSION, + type: SWM_SENDER_KEY_PACKAGE_ACK_TYPE, + accepted: false, + reason: 'simulated per-recipient fatal', + }); + const isBfailure = peerId === recipientB.peerId; + return { + delivered: true, + response: isBfailure ? rejectedEnvelope : acceptedEnvelope, + attempts: 1, + messageId: `m-test-${peerId.slice(-6)}`, + }; + }); + + const sender = agentFromPrivateKey( + ethers.Wallet.createRandom().privateKey, + 'sender', + ) as AgentKeyRecord & { privateKey: string }; + + let thrown: Error | null = null; + try { + await internals.createAndDistributeSwmSenderKeyEpoch({ + contextGraphId: 'test-cg/fanout-1ofN', + sender, + recipients: [recipientA, recipientB, recipientC], + membershipHash: 'sha256:fanout-1ofN', + ctx: { operationId: 'test-op', operationName: 'share' }, + }); + } catch (err) { + thrown = err as Error; + } + + // Must throw — recipient B is fatal even though A and C succeeded. + expect(thrown).not.toBeNull(); + // Aggregation count must be EXACTLY 1 — not 3 (every agent), not + // 0 (none). + expect(thrown!.message).toMatch(/rejected by 1 agent\(s\)/); + // Identity of the fatal agent must be present in the throw. + expect(thrown!.message.toLowerCase()).toContain(recipientB.agentAddress.toLowerCase()); + // Identities of the successful agents MUST NOT be present (would + // leak diagnostic noise and mislead operators). + expect(thrown!.message.toLowerCase()).not.toContain(recipientA.agentAddress.toLowerCase()); + expect(thrown!.message.toLowerCase()).not.toContain(recipientC.agentAddress.toLowerCase()); + // The simulated per-recipient reason should bubble up via the + // failure list (proves the per-key reasons are forwarded). + expect(thrown!.message).toContain('simulated per-recipient fatal'); + }); + + it('per-AGENT (not per-key) aggregation: an agent with 2 keys where 1 accepts and 1 rejects is NOT fatal', async () => { + // Codex review feedback on #740: the 1-of-N test above uses one + // key per agent, so it does not actually exercise the + // "fatal only when EVERY key for an agent fails" aggregation + // rule documented at `dkg-agent.ts:5998-6042`. A regression + // that started aggregating by key (instead of by agent) would + // pass that test silently — any one key rejection would still + // throw, even if other keys for the SAME agent succeeded. + // + // To pin the per-agent semantics, build recipientB with TWO + // keys (same `agentAddress` + `peerId`, distinct + // `recipientKeyId` + `publicKeyBytes`) and have the messenger + // accept one and reject the other. The expected production + // behavior: B is logged as a partial-delivery warning but NOT + // added to `fatalAgents`, so the fanout call resolves + // successfully overall. recipientA with a single all-fail key + // is the actual fatal — the only one cited in the throw. + const boot = await bootAgent(); + agent = boot.agent; + const internals = boot.internals; + + // Build recipientB with two keys for the SAME agent. + const wallet = ethers.Wallet.createRandom(); + const agentAddress = wallet.address; + const recipientId = `did:dkg:agent:${agentAddress.toLowerCase()}`; + const peerId = `12D3KooWFakeTestPeer${ethers.id(agentAddress).slice(2, 18)}`; + const keyAId = `${recipientId}#x25519-keyA-${ethers.id(`${agentAddress}|A`).slice(2, 10)}`; + const keyBId = `${recipientId}#x25519-keyB-${ethers.id(`${agentAddress}|B`).slice(2, 10)}`; + const keyA = generateWorkspaceRecipientEncryptionKey(recipientId, keyAId); + const keyB = generateWorkspaceRecipientEncryptionKey(recipientId, keyBId); + const recipientB_keyA: FakeRecipient = { + agentAddress, + peerId, + recipientKeyId: keyAId, + recipientId, + purpose: WORKSPACE_RECIPIENT_ENCRYPTION_KEY_PURPOSE, + encryptionKeyAlgorithm: WORKSPACE_AGENT_ENCRYPTION_KEY_ALGORITHM_X25519, + publicKeyBytes: keyA.publicKeyBytes!, + }; + const recipientB_keyB: FakeRecipient = { + agentAddress, + peerId, + recipientKeyId: keyBId, + recipientId, + purpose: WORKSPACE_RECIPIENT_ENCRYPTION_KEY_PURPOSE, + encryptionKeyAlgorithm: WORKSPACE_AGENT_ENCRYPTION_KEY_ALGORITHM_X25519, + publicKeyBytes: keyB.publicKeyBytes!, + }; + // A separate agent whose only key always fails — the genuine + // fatal, used as the control to keep the throw observable. + const recipientFatal = makeFakeRecipient(); + // And one fully-successful agent to keep the all-accept path + // active in this scenario. + const recipientHappy = makeFakeRecipient(); + + // Messenger discrimination: + // - recipientFatal: always reject (genuine fatal) + // - recipientB peerId: reject the call carrying keyAId, accept + // the call carrying keyBId. Both calls go to the SAME peerId + // so we need a STABLE per-key discriminator — Codex review + // feedback on the prior revision: keying off the per-peer + // call ordinal was order-dependent because the fanout's + // `createSignedSwmSenderKeyPackage` runs per-recipient in + // parallel and the two sends can race to the messenger in + // either order. The `SwmSenderKeyPackage` proto encodes + // `recipientKeyId` as a top-level plaintext field (the + // recipient needs it to pick the right decryption key), so + // we decode the payload and key the decision off that. + // - everyone else: accept. + const callsByPeer = new Map(); + const seenKeyIds: string[] = []; + installStubMessenger(internals, async (sendPeerId, _protocolId, payload): Promise => { + const acceptedEnvelope = encodeSwmSenderKeyPackageAck({ + version: SWM_SENDER_KEY_PACKAGE_VERSION, + type: SWM_SENDER_KEY_PACKAGE_ACK_TYPE, + accepted: true, + }); + const rejectedEnvelope = encodeSwmSenderKeyPackageAck({ + version: SWM_SENDER_KEY_PACKAGE_VERSION, + type: SWM_SENDER_KEY_PACKAGE_ACK_TYPE, + accepted: false, + reason: 'simulated key-level rejection', + }); + callsByPeer.set(sendPeerId, (callsByPeer.get(sendPeerId) ?? 0) + 1); + + if (sendPeerId === recipientFatal.peerId) { + return { + delivered: true, + response: rejectedEnvelope, + attempts: 1, + messageId: `m-fatal-${sendPeerId.slice(-6)}`, + }; + } + if (sendPeerId === peerId) { + // Decode the package to read `recipientKeyId` directly — + // robust against the per-recipient send race. + const pkg = decodeSwmSenderKeyPackage(payload); + seenKeyIds.push(pkg.recipientKeyId); + return { + delivered: true, + response: pkg.recipientKeyId === keyAId ? rejectedEnvelope : acceptedEnvelope, + attempts: 1, + messageId: `m-mixed-${pkg.recipientKeyId.slice(-8)}`, + }; + } + return { + delivered: true, + response: acceptedEnvelope, + attempts: 1, + messageId: `m-happy-${sendPeerId.slice(-6)}`, + }; + }); + + const sender = agentFromPrivateKey( + ethers.Wallet.createRandom().privateKey, + 'sender', + ) as AgentKeyRecord & { privateKey: string }; + + let thrown: Error | null = null; + try { + await internals.createAndDistributeSwmSenderKeyEpoch({ + contextGraphId: 'test-cg/per-agent-mixed', + sender, + // Order matters for the per-peer ordinal discrimination: + // B_keyA is sent BEFORE B_keyB, so the messenger's "first + // call to recipientB's peerId" reliably maps to keyA. + recipients: [recipientHappy, recipientB_keyA, recipientB_keyB, recipientFatal], + membershipHash: 'sha256:per-agent-mixed', + ctx: { operationId: 'test-op', operationName: 'share' }, + }); + } catch (err) { + thrown = err as Error; + } + + // We expect a throw — recipientFatal is the only ALL-fail agent. + expect(thrown).not.toBeNull(); + + // Pin per-AGENT semantics: exactly 1 fatal agent (not 2). A + // regression that counted per-key would surface "2 agent(s)" + // because recipientB had a key-level rejection too. + expect(thrown!.message).toMatch(/rejected by 1 agent\(s\)/); + + // The throw must cite recipientFatal but NOT recipientB — + // recipientB had partial success and is intentionally not + // listed as fatal under the per-agent rule. + expect(thrown!.message.toLowerCase()).toContain(recipientFatal.agentAddress.toLowerCase()); + expect(thrown!.message.toLowerCase()).not.toContain(agentAddress.toLowerCase()); + expect(thrown!.message.toLowerCase()).not.toContain(recipientHappy.agentAddress.toLowerCase()); + + // Sanity: recipientB's peerId was called exactly twice and the + // per-key discrimination correctly saw BOTH keyAId and keyBId + // (order doesn't matter — that's the whole point of decoding + // the payload instead of using a per-peer call ordinal). + expect(callsByPeer.get(peerId)).toBe(2); + expect([...seenKeyIds].sort()).toEqual([keyAId, keyBId].sort()); + }); });