feat: expose executor job observability

services/orchestrator/executor-job-store.test.mjs

@@ -19,6 +19,18 @@ function record(overrides = {}) {
   };
 }
 
+function transactionalPool(query) {
+  return {
+    query,
+    async connect() {
+      return {
+        query,
+        release() {},
+      };
+    },
+  };
+}
+
 test('PostgreSQL Executor Job Store creates an isolated schema and job table', async () => {
   const queries = [];
   const store = createPostgresExecutorJobStore({
@@ -37,53 +49,85 @@ test('PostgreSQL Executor Job Store creates an isolated schema and job table', a
   assert.match(queries[1].sql, /"memind_orchestrator_test"\."executor_jobs"/);
   assert.match(queries[1].sql, /idempotency_key VARCHAR\(200\) NOT NULL UNIQUE/);
   assert.match(queries[2].sql, /executor_jobs_updated_at_idx/);
+  assert.match(queries[3].sql, /"memind_orchestrator_test"\."executor_job_events"/);
+  assert.match(queries[3].sql, /UNIQUE \(job_id, sequence\)/);
+  assert.match(queries[5].sql, /executor_job_snapshot_imported/);
 });
 
 test('PostgreSQL Executor Job Store atomically creates and reloads idempotent state', async () => {
   const stored = record();
+  const blockedEvent = {
+    version: 'executor-job-event-v1',
+    eventId: 'event-1',
+    jobId: 'job-1',
+    sequence: 1,
+    type: 'executor_job_blocked',
+    timestamp: 1000,
+    data: { status: 'blocked' },
+  };
   const calls = [];
   const store = createPostgresExecutorJobStore({
-    pool: {
-      async query(sql, params = []) {
-        calls.push({ sql, params });
-        if (sql.includes('INSERT INTO')) {
-          return { rows: [{ state_json: stored }] };
-        }
-        if (sql.includes('UPDATE')) {
-          return { rows: [{ state_json: { ...stored, status: 'cancelled' } }] };
-        }
-        if (sql.includes('WHERE job_id =')) {
-          return { rows: [{ state_json: JSON.stringify(stored) }] };
-        }
-        if (sql.includes('WHERE idempotency_key =')) {
-          return { rows: [{ state_json: stored }] };
-        }
+    pool: transactionalPool(async (sql, params = []) => {
+      calls.push({ sql, params });
+      if (sql.includes('INSERT INTO') && sql.includes('executor_jobs')) {
+        return { rows: [{ state_json: stored }] };
+      }
+      if (sql.includes('INSERT INTO') && sql.includes('executor_job_events')) {
         return { rows: [] };
-      },
-    },
+      }
+      if (sql.includes('SELECT job_id') && sql.includes('FOR UPDATE')) {
+        return { rows: [{ job_id: 'job-1' }] };
+      }
+      if (sql.includes('UPDATE')) {
+        return { rows: [{ state_json: { ...stored, status: 'cancelled' } }] };
+      }
+      if (sql.includes('MAX(sequence)')) {
+        return { rows: [{ next_sequence: 2 }] };
+      }
+      if (sql.includes('SELECT event_json')) {
+        return { rows: [{ event_json: blockedEvent }] };
+      }
+      if (sql.includes('WHERE job_id =')) {
+        return { rows: [{ state_json: JSON.stringify(stored) }] };
+      }
+      if (sql.includes('WHERE idempotency_key =')) {
+        return { rows: [{ state_json: stored }] };
+      }
+      return { rows: [] };
+    }),
   });
 
-  const created = await store.createIfAbsent(stored);
+  const created = await store.createIfAbsent(stored, { initialEvent: blockedEvent });
   assert.equal(created.created, true);
   assert.deepEqual(created.record, stored);
   assert.deepEqual(await store.getById('job-1'), stored);
   assert.deepEqual(await store.getByIdempotencyKey('idem-1'), stored);
-  assert.equal((await store.update({ ...stored, status: 'cancelled' })).status, 'cancelled');
-  assert.equal(JSON.parse(calls[0].params[3]).jobId, 'job-1');
+  assert.equal(
+    (await store.update(
+      { ...stored, status: 'cancelled' },
+      { event: { ...blockedEvent, eventId: 'event-2', type: 'executor_job_cancelled' } },
+    )).status,
+    'cancelled',
+  );
+  assert.deepEqual(await store.listEvents('job-1'), [blockedEvent]);
+  const jobInsert = calls.find((call) => (
+    call.sql.includes('INSERT INTO') && call.sql.includes('executor_jobs')
+  ));
+  assert.equal(JSON.parse(jobInsert.params[3]).jobId, 'job-1');
+  assert.equal(calls.filter((call) => call.sql.trim() === 'BEGIN').length, 2);
+  assert.equal(calls.filter((call) => call.sql.trim() === 'COMMIT').length, 2);
 });
 
 test('PostgreSQL Executor Job Store returns the existing idempotent record after conflict', async () => {
   const stored = record();
   const store = createPostgresExecutorJobStore({
-    pool: {
-      async query(sql) {
-        if (sql.includes('INSERT INTO')) return { rows: [] };
-        if (sql.includes('WHERE idempotency_key =')) {
-          return { rows: [{ state_json: stored }] };
-        }
-        return { rows: [] };
-      },
-    },
+    pool: transactionalPool(async (sql) => {
+      if (sql.includes('INSERT INTO')) return { rows: [] };
+      if (sql.includes('WHERE idempotency_key =')) {
+        return { rows: [{ state_json: stored }] };
+      }
+      return { rows: [] };
+    }),
   });
   assert.deepEqual(
     await store.createIfAbsent(stored),
@@ -91,6 +135,37 @@ test('PostgreSQL Executor Job Store returns the existing idempotent record after
   );
 });
 
+test('PostgreSQL Executor Job Store rolls back job creation when its initial event fails', async () => {
+  const calls = [];
+  const eventError = new Error('event insert unavailable');
+  const store = createPostgresExecutorJobStore({
+    pool: transactionalPool(async (sql) => {
+      calls.push(sql.trim());
+      if (sql.includes('INSERT INTO') && sql.includes('executor_jobs')) {
+        return { rows: [{ state_json: record() }] };
+      }
+      if (sql.includes('INSERT INTO') && sql.includes('executor_job_events')) {
+        throw eventError;
+      }
+      return { rows: [] };
+    }),
+  });
+
+  await assert.rejects(
+    store.createIfAbsent(record(), {
+      initialEvent: {
+        eventId: 'event-failing',
+        jobId: 'job-1',
+        timestamp: 1000,
+      },
+    }),
+    eventError,
+  );
+  assert.ok(calls.includes('BEGIN'));
+  assert.ok(calls.includes('ROLLBACK'));
+  assert.ok(!calls.includes('COMMIT'));
+});
+
 test('Executor Job persistence supports explicit non-durable memory mode', async () => {
   const persistence = await createExecutorJobPersistence({ mode: 'memory' });
   assert.equal(persistence.kind, 'memory');