Durable

KIT · constant · 运行时值

导入

TypeScript

import { Durable } from '@jangts/agent-core/kits/durable';

JavaScript

import { Durable } from '@jangts/agent-core/kits/durable';

其他公开入口：@jangts/agent-core/types、@jangts/agent-core/kits、@jangts/agent-core/kits/durable。

定义

成员目录

成员	用途
defineDurableTask	完整输入输出见下方重载。

成员详情

defineDurableTask

Durable.defineDurableTask<State, App>(definition: DurableTaskDefinition<State, App>, options?: DurableTaskOptions<State>): DurableTask<State, App>

Compile a durable task definition into a resumable runner.

Each run call locks the task id in the store, loads the checkpoint and resumes at nextStepIndex. A checkpoint from another definition id or revision, or one that fails the predicates below, is rejected instead of being repaired, because silently accepting it would replay App side effects under a different contract.

类型参数	约束	默认类型
State	—	—
App	—	—

参数	类型	传递	说明
definition	DurableTaskDefinition<State, App>	必需	—
options	DurableTaskOptions<State>	可省略	—

definition 的字段

字段	类型	必需	说明
id	string	必需	只读。见类型定义及调用场景。
revision	string	必需	只读。Definition revision; stored in the checkpoint so resumed work matches its code.
steps	readonly DurableStep<State, App>[]	必需	只读。Steps in execution order; the checkpoint stores an index into this list.
isState	(state: unknown) => state is State	必需	Validates a loaded checkpoint state before any step runs.
authorize	((checkpoint: Readonly<DurableCheckpoint<State>>, context: OperationContext<App>) => import("../common.js").MaybePromise<void>) | undefined	可省略	Recheck current identity, source versions and grants on every explicit resume.

options 的字段

字段	类型	必需	说明
now	(() => number) | undefined	可省略	只读。Clock override for deadlines and timestamps; defaults to Date.now.
store	DurableStore<State> | undefined	可省略	只读。Default store; a call may pass its own store instead.

返回： DurableTask<State, App>。

返回值结构

字段	类型	必需	说明
run	(request: { readonly taskId: string; readonly store?: DurableStore<State> | undefined; }, context: OperationContext<App>) => Promise<State>	必需	Runs or resumes the task.

使用示例

• Durable：显式检查点与受控恢复：App 提供实际检查点存储和排他锁，任务沿同一定义与累计额度显式运行或恢复。

先用隔离内存检查点理解一次执行

入门 · Node.js

这是最小存储合同的隔离演示；真正跨进程/跨请求恢复必须换成具备排他与持久保证的 App store。

TypeScript

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);

JavaScript

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

示例不擅自把“无法确认是否提交”解释成失败或重新执行；生产对账需查询真实业务系统。

TypeScript

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);

JavaScript

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);

相关策略源码

• 状态、累计预算、期限和对账顺序
• 当前状态与重放策略枚举

相关定义

• defineDurableTask
• DurableTask
• DurableTaskDefinition
• DurableTaskOptions