DurableStore
KIT · interface · 类型合同
App implements an exclusive lock across every writer and validates/serializes stored state.
导入
ts
import type { DurableStore } from '@jangts/agent-core/kits/durable';js
/** @typedef {import('@jangts/agent-core/kits/durable').DurableStore} DurableStore */其他公开入口：@jangts/agent-core/types、@jangts/agent-core/structs、@jangts/agent-core/kits、@jangts/agent-core/kits/durable。
定义
ts
interface DurableStore<State>泛型参数
| 类型参数 | 约束 | 默认类型 |
|---|---|---|
State | — | — |
字段结构
| 字段 | 类型 | 必需 | 说明 |
|---|---|---|---|
exclusive | <T>(id: string, operation: () => Promise<T>) => Promise<T> | 必需 | Runs operation under the exclusive lock for id; other writers must wait or fail. |
load | (id: string) => Promise<DurableCheckpoint<State> | undefined> | 必需 | Returns the stored checkpoint, or undefined when the task has never been saved. |
save | (id: string, checkpoint: DurableCheckpoint<State>) => Promise<void> | 必需 | Persists a checkpoint; must be durable before the promise resolves. |
成员目录
| 成员 | 用途 |
|---|---|
exclusive | Runs operation under the exclusive lock for id; other writers must wait or fail. |
load | Returns the stored checkpoint, or undefined when the task has never been saved. |
save | Persists a checkpoint; must be durable before the promise resolves. |
成员详情
exclusive
Runs operation under the exclusive lock for id; other writers must wait or fail.
ts
DurableStore.exclusive<T>(id: string, operation: () => Promise<T>): Promise<T>Runs operation under the exclusive lock for id; other writers must wait or fail.
| 类型参数 | 约束 | 默认类型 |
|---|---|---|
T | — | — |
| 参数 | 类型 | 传递 | 说明 |
|---|---|---|---|
id | string | 必需 | — |
operation | () => Promise<T> | 必需 | — |
返回： Promise<T>。
等待后的值： Awaited<T>。
load
Returns the stored checkpoint, or undefined when the task has never been saved.
ts
DurableStore.load(id: string): Promise<DurableCheckpoint<State> | undefined>Returns the stored checkpoint, or undefined when the task has never been saved.
| 参数 | 类型 | 传递 | 说明 |
|---|---|---|---|
id | string | 必需 | — |
返回： Promise<DurableCheckpoint<State> \| undefined>。
等待后的值： DurableCheckpoint<State> \| undefined。
返回值的可选结构： 选择其中一种结构，各分支的字段要求分别列出。
分支 1
undefined
分支 2
| 字段 | 类型 | 必需 | 说明 |
|---|---|---|---|
version | 2 | 必需 | Checkpoint format version; other versions are rejected on load. |
definitionId | string | 必需 | Definition that produced the checkpoint; a mismatch invalidates it. |
revision | string | 必需 | Definition revision that produced the checkpoint; a mismatch invalidates it. |
nextStepIndex | number | 必需 | Zero-based index of the next step to run. |
attemptCount | number | 必需 | Attempts already spent; one is spent per step entry. |
maxAttempts | number | 必需 | Total attempts allowed for the whole task. |
deadlineAtMs | number | 必需 | Absolute deadline in epoch milliseconds; expiry cancels the task. |
status | "running" | "cancelled" | "completed" | "ready" | 必需 | 见类型定义及调用场景。 |
state | State | 必需 | App-owned task state; only isState decides whether it is still valid. |
modelBudget | { maxCalls: number; reservedCalls: number; maxCallsPerAttempt: number; } | undefined | 可省略 | Durable, conservative reservations. An interrupted attempt never refunds its allowance. |
save
Persists a checkpoint; must be durable before the promise resolves.
ts
DurableStore.save(id: string, checkpoint: DurableCheckpoint<State>): Promise<void>Persists a checkpoint; must be durable before the promise resolves.
| 参数 | 类型 | 传递 | 说明 |
|---|---|---|---|
id | string | 必需 | — |
checkpoint | DurableCheckpoint<State> | 必需 | — |
checkpoint 的字段
| 字段 | 类型 | 必需 | 说明 |
|---|---|---|---|
version | 2 | 必需 | Checkpoint format version; other versions are rejected on load. |
definitionId | string | 必需 | Definition that produced the checkpoint; a mismatch invalidates it. |
revision | string | 必需 | Definition revision that produced the checkpoint; a mismatch invalidates it. |
nextStepIndex | number | 必需 | Zero-based index of the next step to run. |
attemptCount | number | 必需 | Attempts already spent; one is spent per step entry. |
maxAttempts | number | 必需 | Total attempts allowed for the whole task. |
deadlineAtMs | number | 必需 | Absolute deadline in epoch milliseconds; expiry cancels the task. |
status | "running" | "cancelled" | "completed" | "ready" | 必需 | 见类型定义及调用场景。 |
state | State | 必需 | App-owned task state; only isState decides whether it is still valid. |
modelBudget | { maxCalls: number; reservedCalls: number; maxCallsPerAttempt: number; } | undefined | 可省略 | Durable, conservative reservations. An interrupted attempt never refunds its allowance. |
返回： Promise<void>。
等待后的值： void。
使用示例
- Durable：显式检查点与受控恢复：App 提供实际检查点存储和排他锁，任务沿同一定义与累计额度显式运行或恢复。
先用隔离内存检查点理解一次执行
入门 · Node.js
这是最小存储合同的隔离演示；真正跨进程/跨请求恢复必须换成具备排他与持久保证的 App store。
ts
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);js
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);同一 run 增加中断后的真实对账策略
配置 · Node.js
示例不擅自把“无法确认是否提交”解释成失败或重新执行；生产对账需查询真实业务系统。
ts
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);js
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);