Durable: explicit checkpoints and controlled recovery
The App provides real checkpoint storage and an exclusive lock; a task runs or resumes explicitly along the same definition and accumulated budget.
defineDurableTask({id,revision,isState,steps}, options?) returns run({taskId,store},context). store must provide exclusive, load, and save that are valid across all writers. A checkpoint has version=2 and contains definitionId/revision, nextStepIndex, attemptCount/maxAttempts, deadlineAtMs, status, and state; it must be created by the App, and the task never seeds or looks up business data automatically.
ready starts the next step, running means that step was interrupted, completed returns the result after validation, and cancelled is rejected. safe means the step may be redone; reconcile means recovery must go through reconcile, and it is rejected when missing. The attempt count is incremented and saved before the handler, and the new state and next step are saved on success; resuming does not reset the accumulated count or the deadline. Every save waits inside the lock, but cancellation cannot undo an external commit that has already happened.
The App starts or resumes tasks through explicit calls and provides durable storage, idempotency keys, side-effect reconciliation, re-authorization, and storage transactions.
Progressive examples for the same business case
First understand one execution with an isolated in-memory checkpoint
Beginner · Node.js
This is an isolated demonstration of the minimal storage contract; real cross-process or cross-request recovery must use an App store with exclusivity and durability guarantees.
import { Durable } from '@jangts/agent-core/kits/durable';
import type { DurableCheckpoint, DurableStore } from '@jangts/agent-core/kits/durable';
type State = { count: number };
const records = new Map<string, DurableCheckpoint<State>>([
['demo-task', { version: 2, definitionId: 'count', revision: '1', nextStepIndex: 0, attemptCount: 0, maxAttempts: 3, deadlineAtMs: Date.now() + 60_000, status: 'ready', state: { count: 0 } }],
]);
let tail = Promise.resolve();
// 隔离示例：进程内串行锁和内存记录，不是生产持久化实现。
const store: DurableStore<State> = {
exclusive(_id, operation) { const result = tail.then(operation); tail = result.then(() => {}, () => {}); return result; },
async load(id) { const record = records.get(id); return record && structuredClone(record); },
async save(id, record) { records.set(id, structuredClone(record)); },
};
const isState = (value: unknown): value is State => !!value && typeof value === 'object' && 'count' in value && typeof value.count === 'number';
const task = Durable.defineDurableTask<State, undefined>({
id: 'count', revision: '1', isState,
steps: [{ id: 'increment', replayPolicy: 'safe', async run(state) { return { count: state.count + 1 }; } }],
});
const context = { app: undefined, signal: new AbortController().signal };
console.log(await task.run({ taskId: 'demo-task', store }, context));
console.log(records.get('demo-task')?.attemptCount);import { Durable } from '@jangts/agent-core/kits/durable';
const records = new Map([
['demo-task', { version: 2, definitionId: 'count', revision: '1', nextStepIndex: 0, attemptCount: 0, maxAttempts: 3, deadlineAtMs: Date.now() + 60_000, status: 'ready', state: { count: 0 } }],
]);
let tail = Promise.resolve();
// 隔离示例：进程内串行锁和内存记录，不是生产持久化实现。
const store = {
exclusive(_id, operation) { const result = tail.then(operation); tail = result.then(() => { }, () => { }); return result; },
async load(id) { const record = records.get(id); return record && structuredClone(record); },
async save(id, record) { records.set(id, structuredClone(record)); },
};
const isState = (value) => !!value && typeof value === 'object' && 'count' in value && typeof value.count === 'number';
const task = Durable.defineDurableTask({
id: 'count', revision: '1', isState,
steps: [{ id: 'increment', replayPolicy: 'safe', async run(state) { return { count: state.count + 1 }; } }],
});
const context = { app: undefined, signal: new AbortController().signal };
console.log(await task.run({ taskId: 'demo-task', store }, context));
console.log(records.get('demo-task')?.attemptCount);Add a real reconciliation policy for interruptions to the same run
Configuration · Node.js
The example does not presume to interpret "cannot confirm whether it committed" as failure or re-execution; production reconciliation must query the real business system.
import { Durable } from '@jangts/agent-core/kits/durable';
import type { DurableCheckpoint, DurableStore } from '@jangts/agent-core/kits/durable';
type State = { count: number };
const records = new Map<string, DurableCheckpoint<State>>([
['demo-task', { version: 2, definitionId: 'count', revision: '1', nextStepIndex: 0, attemptCount: 0, maxAttempts: 3, deadlineAtMs: Date.now() + 60_000, status: 'ready', state: { count: 0 } }],
]);
let tail = Promise.resolve();
// 隔离示例：进程内串行锁和内存记录，不是生产持久化实现。
const store: DurableStore<State> = {
exclusive(_id, operation) { const result = tail.then(operation); tail = result.then(() => {}, () => {}); return result; },
async load(id) { const record = records.get(id); return record && structuredClone(record); },
async save(id, record) { records.set(id, structuredClone(record)); },
};
const isState = (value: unknown): value is State => !!value && typeof value === 'object' && 'count' in value && typeof value.count === 'number';
// 假设业务服务以 taskId 为幂等键保存真实提交结果；这里用内存 Map 演示。
const committed = new Map<string, number>();
const task = Durable.defineDurableTask<State, { key: string }>({
id: 'count', revision: '1', isState,
steps: [{ id: 'commit', replayPolicy: 'reconcile',
async run(state, { app }) {
const value = committed.get(app.key) ?? state.count + 1;
committed.set(app.key, value);
return { count: value };
},
async reconcile(state, { app }) {
const value = committed.get(app.key);
if (value === undefined) throw new Error('没有可核实的提交结果，需要 App 决定后续动作');
return { count: value };
},
}],
});
const context = { app: { key: 'demo-task' }, signal: new AbortController().signal };
console.log(await task.run({ taskId: 'demo-task', store }, context));
console.log(await task.run({ taskId: 'demo-task', store }, context)); // 已完成记录不会再次执行步骤
console.log(committed.size);import { Durable } from '@jangts/agent-core/kits/durable';
const records = new Map([
['demo-task', { version: 2, definitionId: 'count', revision: '1', nextStepIndex: 0, attemptCount: 0, maxAttempts: 3, deadlineAtMs: Date.now() + 60_000, status: 'ready', state: { count: 0 } }],
]);
let tail = Promise.resolve();
// 隔离示例：进程内串行锁和内存记录，不是生产持久化实现。
const store = {
exclusive(_id, operation) { const result = tail.then(operation); tail = result.then(() => { }, () => { }); return result; },
async load(id) { const record = records.get(id); return record && structuredClone(record); },
async save(id, record) { records.set(id, structuredClone(record)); },
};
const isState = (value) => !!value && typeof value === 'object' && 'count' in value && typeof value.count === 'number';
// 假设业务服务以 taskId 为幂等键保存真实提交结果；这里用内存 Map 演示。
const committed = new Map();
const task = Durable.defineDurableTask({
id: 'count', revision: '1', isState,
steps: [{ id: 'commit', replayPolicy: 'reconcile',
async run(state, { app }) {
const value = committed.get(app.key) ?? state.count + 1;
committed.set(app.key, value);
return { count: value };
},
async reconcile(state, { app }) {
const value = committed.get(app.key);
if (value === undefined)
throw new Error('没有可核实的提交结果，需要 App 决定后续动作');
return { count: value };
},
}],
});
const context = { app: { key: 'demo-task' }, signal: new AbortController().signal };
console.log(await task.run({ taskId: 'demo-task', store }, context));
console.log(await task.run({ taskId: 'demo-task', store }, context)); // 已完成记录不会再次执行步骤
console.log(committed.size);