Retrouver une opération et reprendre après une coupure

Cette recette locale crée une écriture auprès d'un service HTTP, coupe le serveur avant réception du reçu, puis retrouve et reprend l'opération par les API publiques. Le fournisseur reconnaît la clé d'idempotence : deux tentatives produisent une seule écriture. Aucun modèle ni compte externe.

Prérequis : Touba, Node.js 24 et macOS/Linux. La panne SIGKILL concerne uniquement le serveur et le handler créés par cet exercice dans un dossier neuf.

1. Déclarer le parcours

Terminal

mkdir -p src .touba/air

src/orders.tb

module tutorial.orders
limits run
    max_run_ms as 40000
    max_tool_calls as 3
tool CreateOrder
    gives Text
    requires network
strategy Run
    gives Text
    uses CreateOrder
    do
        CreateOrder
            into result
        give result

touba.project.json

{
  "project_version":"touba-project-v1",
  "language":"javascript",
  "sdk":"@touba/sdk",
  "artifact_layout":"touba-artifacts-v1",
  "entrypoints":["strategy:tutorial.orders/Run"]
}

2. Observer l'effet avant de perdre son reçu

Ce handler utilise le protocole processus. Le premier appel attend après l'écriture externe ; le second rend son reçu. Le service local de l'étape 3 déduplique réellement les demandes par effect_id.

handler.mjs

import {writeFileSync} from 'node:fs';
let input = '';
for await (const chunk of process.stdin) input += chunk;
const request = JSON.parse(input);
const response = await fetch(process.argv[2], {
  method: 'POST', body: JSON.stringify({effect_id: request.effect.effect_id}),
});
if (!response.ok) throw new Error('Order service refused the request');
const result = await response.json();
if (request.effect.attempt_index === 1) {
  writeFileSync('interrupted-handler.pid', String(process.pid));
  Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0);
}
process.stdout.write(JSON.stringify({ok: true, value: result.value,
  effect_receipt: {effect_id: request.effect.effect_id,
    request_digest: request.effect.request_digest, outcome: 'applied',
    receipt_ref: result.receipt_ref}}));

3. Exécuter et vérifier la reprise

La recette conserve session_id et message_id avant de récupérer operation_id. Elle lit la révision et le digest par /recovery. Les comptes métier et les identités de l'application restent gérés par l'hôte.

exercise.mjs

import assert from 'node:assert/strict';
import {createServer, request as httpRequest} from 'node:http';
import {spawn} from 'node:child_process';
import {once} from 'node:events';
import {existsSync, readFileSync, writeFileSync} from 'node:fs';
import {setTimeout as delay} from 'node:timers/promises';

const orders = new Map();
let attempts = 0;
const provider = createServer(async (request, response) => {
  let body = ''; for await (const chunk of request) body += chunk;
  const {effect_id} = JSON.parse(body); attempts++;
  if (!orders.has(effect_id)) orders.set(effect_id, {
    value: 'ORDER-42', receipt_ref: 'receipt:order-42',
  });
  response.writeHead(200, {'content-type':'application/json'});
  response.end(JSON.stringify(orders.get(effect_id)));
});
await new Promise(resolve => provider.listen(0, '127.0.0.1', resolve));
writeFileSync('touba.manifest.json', JSON.stringify({manifest_version:'touba-manifest-v1', bindings:{
  'tool:tutorial.orders/CreateOrder': {
    handler:{kind:'process',command:process.execPath,args:['handler.mjs',
      `http://127.0.0.1:${provider.address().port}/orders`],cwd:'.'},
    grants:['network'], timeout_ms:10000, max_payload_bytes:65536,
    effect_contract:{kind:'external_write',retry_mode:'idempotency_key'},
  },
}}));
const children = new Set();
let handlerPid;
const token = 'local-tutorial-token';
async function start() {
  const child = spawn('touba', ['serve','--project-root',process.cwd(),'--port','0',
    '--auth-token-env','TUTORIAL_SERVER_TOKEN'], {
    env:{...process.env,TUTORIAL_SERVER_TOKEN:token}, stdio:['ignore','pipe','pipe'], detached:true,
  });
  children.add(child);
  let buffer = '', errors = '';
  child.stderr.on('data', chunk => { errors += chunk; });
  const address = await new Promise((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error(`Server timeout: ${errors}`)), 10000);
    child.once('exit', () => {clearTimeout(timer); reject(new Error(errors));});
    child.stdout.on('data', chunk => {
      buffer += chunk;
      const lines = buffer.split('\n'); buffer = lines.pop();
      for (const line of lines) {
        let event; try {event = JSON.parse(line);} catch {continue;}
        if (event.event === 'touba_project_server_started') {
          clearTimeout(timer); resolve(`http://127.0.0.1:${event.port}`);
        }
      }
    });
  });
  return {child, address};
}
async function json(address, path, body) {
  const response = await fetch(address + path, {
    method:body === undefined ? 'GET' : 'POST',
    headers:{authorization:`Bearer ${token}`,'content-type':'application/json'},
    ...(body === undefined ? {} : {body:JSON.stringify(body)}),
  });
  const value = await response.json();
  assert.ok(response.ok, JSON.stringify(value)); return value;
}
try {
  const first = await start();
  const actor = {kind:'service',id:'tutorial'};
  const session = await json(first.address, '/interactions/sessions', {
    schema_version:'touba-interaction-session-create-v1', target_kind:'strategy',
    target_ref:'strategy:tutorial.orders/Run', presentation:'workflow',
    effect_policy:'allow_declared_external_effects', requested_by:actor,
  });
  const path = `/interactions/sessions/${encodeURIComponent(session.session_id)}`;
  const message = {schema_version:'touba-interaction-message-request-v1',
    message_id:'message:create-order-42',display_text:'Create order',inputs:{},requested_by:actor};
  await new Promise((resolve, reject) => {
    const request = httpRequest(first.address + path + '/messages', {method:'POST',
      headers:{authorization:`Bearer ${token}`,'content-type':'application/json'}}, response => {
      if (response.statusCode !== 202) {response.resume(); reject(new Error('Admission failed')); return;}
      response.destroy(); resolve(); // Response body deliberately lost.
    });
    request.on('error', reject); request.end(JSON.stringify(message));
  });
  const deadline = Date.now() + 8000;
  while (!existsSync('interrupted-handler.pid') && Date.now() < deadline) await delay(20);
  assert.ok(existsSync('interrupted-handler.pid'));
  handlerPid = Number(readFileSync('interrupted-handler.pid','utf8'));
  process.kill(-first.child.pid,'SIGKILL'); await once(first.child,'exit'); children.delete(first.child);
  try {process.kill(handlerPid,'SIGKILL');} catch (error) {if(error.code !== 'ESRCH') throw error;}
  handlerPid = undefined;

  const second = await start();
  const restored = await json(second.address, path);
  const operation = restored.operations.find(item => item.message_id === message.message_id);
  assert.ok(operation); assert.equal(restored.operations.length,1);
  const duplicate = await json(second.address, path + '/messages', message);
  assert.equal(duplicate.operation_id, operation.operation_id);
  const recovery = await json(second.address, `/operations/${encodeURIComponent(operation.operation_id)}/recovery`);
  assert.equal(recovery.status,'interrupted');
  assert.equal(recovery.effects.reconciliation,'retry_safe');
  const resume = {schema_version:'touba-operation-resume-request-v1',operation_id:operation.operation_id,
    checkpoint_revision:recovery.freshness.state_revision,
    checkpoint_digest:recovery.correlations.checkpoint_digest,resume_request_id:'resume:order-42'};
  const result = await json(second.address,'/runs/sandbox/resume',resume);
  assert.equal(result.result,'ORDER-42');
  await json(second.address,'/runs/sandbox/resume',resume);
  assert.equal(attempts,2); assert.equal(orders.size,1);
  console.log('RECOVERY_WITH_ONE_BUSINESS_WRITE_OK');
} finally {
  if (handlerPid) {try {process.kill(handlerPid,'SIGKILL');} catch (error) {if(error.code !== 'ESRCH') throw error;}}
  for (const child of children) {
    if (child.exitCode === null && child.signalCode === null) {process.kill(-child.pid,'SIGTERM'); await once(child,'exit');}
  }
  await new Promise(resolve => provider.close(resolve));
}

Terminal

touba compile src/orders.tb --out .touba/air/orders.air.json
node exercise.mjs

Résultat : RECOVERY_WITH_ONE_BUSINESS_WRITE_OK. Les dossiers de l'exercice restent disponibles pour inspection. La base du service est ici en mémoire et survit au processus Touba pendant le test. En production, le service externe doit conserver durablement ses propres clés d'idempotence.

Le guide de reprise explique le budget de 40 secondes et les autres mécanismes de continuation.