From 1bb9d4d49d42a0b9dc3d8c3bd2ace5849b1f50e8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Wed, 2 Sep 2026 08:56:58 -0400 Subject: [PATCH] server: commit operations before enqueueing and answer seq_no races with PREV_HASH_MISMATCH The operation and receipt now commit before the queue send. A Postgres 23505 on operations_agent_id_seq_no_key is answered with 400 PREV_HASH_MISMATCH carrying the stored chain hash so hook retries can resync. A failed enqueue after the commit is logged as operation.enqueue_failed and the receipt is still returned. Every producer now names its message: MessageQueue.send takes the durable message id, and the BullMQ adapter passes it as the job id, so the receipt's queue_message_id is the real queue identifier. Claude-Session: https://claude.ai/code/session_01QoUjdc7HGKTsgj5GhG8HvF --- AGENTS.md | 1 + packages/server/src/adapters/bullmq.ts | 7 +- packages/server/src/adapters/interfaces.ts | 4 +- packages/server/src/scheduler.ts | 3 +- .../server/src/services/export-service.ts | 2 +- .../server/src/services/operation-service.ts | 50 ++++++-- .../server/test/operation-conflict.test.mjs | 113 ++++++++++++++++++ 7 files changed, 159 insertions(+), 21 deletions(-) create mode 100644 packages/server/test/operation-conflict.test.mjs diff --git a/AGENTS.md b/AGENTS.md index 645a713..5bd4d42 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -56,6 +56,7 @@ This repository publishes the open-source server, console, integration catalog, - Keep production source files at or below 500 lines. - Keep functions focused on one ownership boundary. - Propagate unexpected errors to the CLI or request boundary. +- Commit the operation and receipt before enqueueing. Log a failed enqueue as `operation.enqueue_failed` and still return the receipt; answer a `UNIQUE(agent_id, seq_no)` conflict with `PREV_HASH_MISMATCH` carrying the stored chain hash. - Keep technical claims executable through tests or generated validation. - Keep the mirrored Python package, distribution, and CLI version in `sdks/python/elydora/_version.py`; Setuptools metadata must read that literal dynamically. - Ship `sdks/python/elydora/py.typed` in every mirrored wheel and verify public annotations from an installed-wheel consumer. diff --git a/packages/server/src/adapters/bullmq.ts b/packages/server/src/adapters/bullmq.ts index d38e0d4..ea5233a 100644 --- a/packages/server/src/adapters/bullmq.ts +++ b/packages/server/src/adapters/bullmq.ts @@ -9,7 +9,6 @@ import { Queue as BullQueue } from 'bullmq'; import type { ConnectionOptions } from 'bullmq'; import type { MessageQueue } from './interfaces.js'; -import { generateUUIDv7 } from '../utils/uuid.js'; export const QUEUE_NAME = 'elydora-queue'; @@ -20,14 +19,14 @@ export class BullMQAdapter implements MessageQueue { this.queue = new BullQueue(QUEUE_NAME, { connection }); } - async send(body: unknown): Promise<{ messageId: string }> { - const job = await this.queue.add('message', body as Record, { + async send(messageId: string, body: unknown): Promise { + await this.queue.add('message', body as Record, { + jobId: messageId, attempts: 3, backoff: { type: 'exponential', delay: 1000 }, removeOnComplete: { count: 100 }, removeOnFail: { count: 500 }, }); - return { messageId: job.id ?? generateUUIDv7() }; } async close(): Promise { diff --git a/packages/server/src/adapters/interfaces.ts b/packages/server/src/adapters/interfaces.ts index 0221045..44432f4 100644 --- a/packages/server/src/adapters/interfaces.ts +++ b/packages/server/src/adapters/interfaces.ts @@ -117,6 +117,6 @@ export interface Cache { * Mirrors the subset of Cloudflare Queue methods used by Elydora services. */ export interface MessageQueue { - /** Enqueue a message. Returns the assigned message ID. */ - send(body: unknown): Promise<{ messageId: string }>; + /** Enqueue a message under the caller's durable message ID. */ + send(messageId: string, body: unknown): Promise; } diff --git a/packages/server/src/scheduler.ts b/packages/server/src/scheduler.ts index 01cbb7d..9a2035a 100644 --- a/packages/server/src/scheduler.ts +++ b/packages/server/src/scheduler.ts @@ -12,6 +12,7 @@ import { RedisCacheAdapter } from './adapters/redis-cache.js'; import { BullMQAdapter } from './adapters/bullmq.js'; import { createEpoch } from './services/epoch-service.js'; import { DEFAULT_EPOCH_INTERVAL_MS } from './shared/index.js'; +import { generateUUIDv7 } from './utils/uuid.js'; // --------------------------------------------------------------------------- // Initialize adapters @@ -65,7 +66,7 @@ async function runEpochCreation(): Promise { if (epoch) { try { - await queue.send({ + await queue.send(generateUUIDv7(), { type: 'tsa', epoch_id: epoch.epoch_id, org_id: epoch.org_id, diff --git a/packages/server/src/services/export-service.ts b/packages/server/src/services/export-service.ts index 7396bc2..9e7343f 100644 --- a/packages/server/src/services/export-service.ts +++ b/packages/server/src/services/export-service.ts @@ -99,7 +99,7 @@ export async function createExport( .run(); // Enqueue the export job - await queue.send({ + await queue.send(generateUUIDv7(), { type: 'export', export_id: exportId, org_id: orgId, diff --git a/packages/server/src/services/operation-service.ts b/packages/server/src/services/operation-service.ts index a87af2f..9bbaa62 100644 --- a/packages/server/src/services/operation-service.ts +++ b/packages/server/src/services/operation-service.ts @@ -201,20 +201,10 @@ export async function submitOperation( }); // ------------------------------------------------------------------ - // Step 10: Enqueue for async processing + // Step 10: Reserve the queue message ID; the send follows the commit // ------------------------------------------------------------------ - const queueMessageResult = await queue.send({ - type: 'operation', - operation_id: eor.operation_id, - org_id: eor.org_id, - agent_id: eor.agent_id, - }); - - // D1 batch insert: operation + receipt + const queueMessageId = generateUUIDv7(); const receiptId = generateUUIDv7(); - const queueMessageId = typeof queueMessageResult === 'object' && queueMessageResult !== null - ? String((queueMessageResult as Record).messageId ?? receiptId) - : receiptId; // ------------------------------------------------------------------ // Step 11: Generate EAR receipt @@ -291,11 +281,45 @@ export async function submitOperation( .bind(receiptId, eor.operation_id, r2ReceiptKey, receivedAt), ); - await db.batch(statements); + try { + await db.batch(statements); + } catch (error) { + if (!isSeqNoConflict(error)) throw error; + const winner = await db + .prepare('SELECT chain_hash FROM operations WHERE agent_id = ? ORDER BY seq_no DESC LIMIT 1') + .bind(eor.agent_id) + .first<{ chain_hash: string }>(); + if (!winner) throw new Error('seq_no conflict without a stored operation.'); + throw new AppError(400, 'PREV_HASH_MISMATCH', { + key: 'operation.prevHashMismatch', + params: { expected: winner.chain_hash, actual: eor.prev_chain_hash }, + }); + } + + try { + await queue.send(queueMessageId, { + type: 'operation', + operation_id: eor.operation_id, + org_id: eor.org_id, + agent_id: eor.agent_id, + }); + } catch (error) { + console.error('operation.enqueue_failed', { + operation_id: eor.operation_id, + queue_message_id: queueMessageId, + error: error instanceof Error ? error.message : String(error), + }); + } return { receipt: ear }; } +function isSeqNoConflict(error: unknown): boolean { + if (typeof error !== 'object' || error === null) return false; + const { code, constraint } = error as { code?: unknown; constraint?: unknown }; + return code === '23505' && constraint === 'operations_agent_id_seq_no_key'; +} + // --------------------------------------------------------------------------- // Get operation // --------------------------------------------------------------------------- diff --git a/packages/server/test/operation-conflict.test.mjs b/packages/server/test/operation-conflict.test.mjs new file mode 100644 index 0000000..4a4ca4c --- /dev/null +++ b/packages/server/test/operation-conflict.test.mjs @@ -0,0 +1,113 @@ +import assert from 'node:assert/strict'; +import { generateKeyPairSync, randomUUID } from 'node:crypto'; +import test from 'node:test'; + +import { AppError } from '../dist/middleware/error-handler.js'; +import { submitOperation } from '../dist/services/operation-service.js'; +import { jcsCanonicalise, sha256Base64url, signEd25519 } from '../dist/utils/crypto.js'; + +const GENESIS = 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; +const WINNER = 'Rxlf4j36C3KvIQ3hWuOkX698BR5iDypUFuB70JjEuvM'; + +function keyPair() { + const pair = generateKeyPairSync('ed25519'); + return { + seed: pair.privateKey.export({ format: 'jwk' }).d, + publicKey: pair.publicKey.export({ format: 'jwk' }).x, + }; +} + +async function signedEor(agentId, orgId, kid, seed) { + const payload = { tool: 'Bash' }; + const unsigned = { + op_version: '1.0', + operation_id: randomUUID(), + org_id: orgId, + agent_id: agentId, + issued_at: Date.now(), + ttl_ms: 30_000, + nonce: randomUUID(), + operation_type: 'ai.tool_use', + subject: { session_id: 'unit' }, + action: { tool: 'Bash' }, + payload, + payload_hash: await sha256Base64url(jcsCanonicalise(payload)), + prev_chain_hash: GENESIS, + agent_pubkey_kid: kid, + }; + const signature = await signEd25519(seed, new TextEncoder().encode(jcsCanonicalise(unsigned))); + return { ...unsigned, signature }; +} + +function pgConflict() { + return Object.assign(new Error('duplicate key value violates unique constraint "operations_agent_id_seq_no_key"'), { + code: '23505', + constraint: 'operations_agent_id_seq_no_key', + }); +} + +function fakes(batchError, agentId, orgId, kid, publicKey) { + const sent = []; + const db = { + prepare: (sql) => ({ + bind: () => ({ + first: async () => { + if (sql.includes('FROM agents')) return { agent_id: agentId, org_id: orgId, status: 'active' }; + if (sql.includes('FROM agent_keys')) return { kid, agent_id: agentId, public_key: publicKey, status: 'active' }; + if (sql.includes('SELECT chain_hash, seq_no')) return null; + if (sql.includes('SELECT chain_hash FROM operations')) return { chain_hash: WINNER }; + throw new Error(`unexpected query: ${sql}`); + }, + }), + }), + batch: async () => { + if (batchError) throw batchError; + return []; + }, + }; + const r2 = { put: async () => undefined }; + const kv = { get: async () => null, put: async () => undefined }; + const queue = { send: async (messageId, body) => { sent.push({ messageId, body }); } }; + return { db, r2, kv, queue, sent }; +} + +test('a seq_no conflict on insert reports the winning chain hash as PREV_HASH_MISMATCH', async () => { + const agent = keyPair(); + const server = keyPair(); + const eor = await signedEor('agent-conflict', 'org-conflict', 'agent-conflict-key-1', agent.seed); + const { db, r2, kv, queue, sent } = fakes(pgConflict(), 'agent-conflict', 'org-conflict', 'agent-conflict-key-1', agent.publicKey); + + await assert.rejects( + submitOperation(db, r2, kv, queue, eor, server.seed), + (error) => error instanceof AppError + && error.statusCode === 400 + && error.errorCode === 'PREV_HASH_MISMATCH' + && error.message.includes(`Expected prev_chain_hash "${WINNER}"`), + ); + assert.equal(sent.length, 0); +}); + +test('other insert failures propagate unchanged and nothing is enqueued', async () => { + const agent = keyPair(); + const server = keyPair(); + const eor = await signedEor('agent-conflict', 'org-conflict', 'agent-conflict-key-1', agent.seed); + const { db, r2, kv, queue, sent } = fakes(new Error('connection reset'), 'agent-conflict', 'org-conflict', 'agent-conflict-key-1', agent.publicKey); + + await assert.rejects( + submitOperation(db, r2, kv, queue, eor, server.seed), + (error) => error instanceof Error && !(error instanceof AppError) && error.message === 'connection reset', + ); + assert.equal(sent.length, 0); +}); + +test('a committed operation is enqueued with its reserved queue message id', async () => { + const agent = keyPair(); + const server = keyPair(); + const eor = await signedEor('agent-conflict', 'org-conflict', 'agent-conflict-key-1', agent.seed); + const { db, r2, kv, queue, sent } = fakes(null, 'agent-conflict', 'org-conflict', 'agent-conflict-key-1', agent.publicKey); + + const { receipt } = await submitOperation(db, r2, kv, queue, eor, server.seed); + assert.equal(sent.length, 1); + assert.equal(sent[0].messageId, receipt.queue_message_id); + assert.equal(sent[0].body.operation_id, eor.operation_id); +});