DurableStore

KIT · interface · Type contract

App implements an exclusive lock across every writer and validates/serializes stored state.

Import

TypeScript

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

JavaScript

/** @typedef {import('@jangts/agent-core/kits/durable').DurableStore} DurableStore */

Other public entry points:@jangts/agent-core/types, @jangts/agent-core/structs, @jangts/agent-core/kits, @jangts/agent-core/kits/durable.

Definition

interface DurableStore<State>

Type parameters

Type parameter	Constraint	Default type
State	—	—

Field structure

Field	Type	Required	Description
exclusive	<T>(id: string, operation: () => Promise<T>) => Promise<T>	Required	Runs operation under the exclusive lock for id; other writers must wait or fail.
load	(id: string) => Promise<DurableCheckpoint<State> | undefined>	Required	Returns the stored checkpoint, or undefined when the task has never been saved.
save	(id: string, checkpoint: DurableCheckpoint<State>) => Promise<void>	Required	Persists a checkpoint; must be durable before the promise resolves.

Member index

Member	Purpose
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.

Member details

exclusive

Runs operation under the exclusive lock for id; other writers must wait or fail.

DurableStore.exclusive<T>(id: string, operation: () => Promise<T>): Promise<T>

Runs operation under the exclusive lock for id; other writers must wait or fail.

Type parameter	Constraint	Default type
T	—	—

Parameter	Type	Passing	Description
id	string	Required	—
operation	() => Promise<T>	Required	—

Returns: Promise<T>.

Awaited value: Awaited<T>.

load

Returns the stored checkpoint, or undefined when the task has never been saved.

DurableStore.load(id: string): Promise<DurableCheckpoint<State> | undefined>

Returns the stored checkpoint, or undefined when the task has never been saved.

Parameter	Type	Passing	Description
id	string	Required	—

Returns: Promise<DurableCheckpoint<State> \| undefined>.

Awaited value: DurableCheckpoint<State> \| undefined.

Optional shapes of Return value: choose one shape; required fields for each branch are listed separately.

Branch 1

undefined

Branch 2

Field	Type	Required	Description
version	2	Required	Checkpoint format version; other versions are rejected on load.
definitionId	string	Required	Definition that produced the checkpoint; a mismatch invalidates it.
revision	string	Required	Definition revision that produced the checkpoint; a mismatch invalidates it.
nextStepIndex	number	Required	Zero-based index of the next step to run.
attemptCount	number	Required	Attempts already spent; one is spent per step entry.
maxAttempts	number	Required	Total attempts allowed for the whole task.
deadlineAtMs	number	Required	Absolute deadline in epoch milliseconds; expiry cancels the task.
status	"running" | "cancelled" | "completed" | "ready"	Required	See the type definition and usage context.
state	State	Required	App-owned task state; only isState decides whether it is still valid.
modelBudget	{ maxCalls: number; reservedCalls: number; maxCallsPerAttempt: number; } | undefined	Optional	Durable, conservative reservations. An interrupted attempt never refunds its allowance.

save

Persists a checkpoint; must be durable before the promise resolves.

DurableStore.save(id: string, checkpoint: DurableCheckpoint<State>): Promise<void>

Persists a checkpoint; must be durable before the promise resolves.

Parameter	Type	Passing	Description
id	string	Required	—
checkpoint	DurableCheckpoint<State>	Required	—

Fields of checkpoint

Field	Type	Required	Description
version	2	Required	Checkpoint format version; other versions are rejected on load.
definitionId	string	Required	Definition that produced the checkpoint; a mismatch invalidates it.
revision	string	Required	Definition revision that produced the checkpoint; a mismatch invalidates it.
nextStepIndex	number	Required	Zero-based index of the next step to run.
attemptCount	number	Required	Attempts already spent; one is spent per step entry.
maxAttempts	number	Required	Total attempts allowed for the whole task.
deadlineAtMs	number	Required	Absolute deadline in epoch milliseconds; expiry cancels the task.
status	"running" | "cancelled" | "completed" | "ready"	Required	See the type definition and usage context.
state	State	Required	App-owned task state; only isState decides whether it is still valid.
modelBudget	{ maxCalls: number; reservedCalls: number; maxCallsPerAttempt: number; } | undefined	Optional	Durable, conservative reservations. An interrupted attempt never refunds its allowance.

Returns: Promise<void>.

Awaited value: void.

Examples

• 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.

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.

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

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.

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

Related policy source

• State, accumulated budget, deadline, and reconciliation order
• Current status and replay policy enumerations

Related definitions

• DurableCheckpoint