Run a durable flow
Use a flow when the steps, values and replay history matter. Start with a small sequential graph, not an arbitrary execution engine.
On this page
Source-backed MarkdownWire the service into your app
Use the independent app/tools and test SQL credentials from Installation. Flows is an explicit app extension: the base App does not automatically provide a flows getter. The example subclass registers it through the normal service() cache.
Every producer and worker needs this boot wiring, definition store and block registry. Initializing Flows registers FlowStepJob; that queued job resolves app().flows when restored. Production apps migrate QueuedJob, FailedJob and FLOW_MODELS before running workers.
import { FileFlowDefinitionStore, Flows } from '@db3.ai/app/flows';
import { App, type AppOptions } from '@db3.ai/app/server';
import { normalizeTextBlock } from './normalizeTextBlock';
import { uppercaseTextBlock } from './uppercaseTextBlock';
/** Application-owned boot wiring shared by request producers and queue workers. */
export class FlowExampleApp extends App {
readonly #definitionRoot: string;
/** Creates the app; the caller owns its definition directory and database. */
constructor(options: AppOptions, definitionRoot: string) { super(options); this.#definitionRoot = definitionRoot; }
/** Lazily registers the durable flow-step handler and the app's block types. */
get flows(): Flows {
return this.service('flows', () => new Flows({ queue: this.queue, queueName: 'flows', maxTries: 1, definitions: new FileFlowDefinitionStore({ root: this.#definitionRoot }), blocks: [normalizeTextBlock, uppercaseTextBlock] }));
}
}
Copy the flow
The five files separate app wiring, graph data, two executable blocks and the disposable runner. Definitions are stored in temporary files by this lab; keep application definitions in a deliberate versioned location.
mkdir -p examples
cp -R node_modules/@db3.ai/app/src/flows/examples/. examples/Run and inspect it
Expect completed, output CLIENT NOTE, two steps and a recorded progress log. Blank input fails with Note text cannot be blank.. Replaying repaired input with the original definition gives FIXED NOTE; replaying the latest definition gives Updated: CLIENT NOTE. All temporary tables and files are removed.
npx tsx examples/runTextFlow.tsGive each block a small contract
defineBlock() describes named input/output ports, optional configuration and one executable function. The graph validates basic value types. Your block still validates business meaning, such as a string that contains more than whitespace.
Keep database work in app-owned blocks with scoped models. There is no generic arbitrary-SQL block. The context gives run/step identities and durable log(); avoid putting secrets or sensitive source values in it.
import { defineBlock } from '@db3.ai/app/flows';
/** Validates application meaning after the graph's basic string contract. */
export const normalizeTextBlock = defineBlock<{ text: string }, { text: string }>({
type: 'notes.normalize', name: 'Normalize note',
inputs: { text: { type: 'string', required: true } },
outputs: { text: { type: 'string', required: true } },
/** Rejects blank text and records a safe, value-free progress message. */
async run(input, context) {
const text = input.text.trim();
if (!text) throw new Error('Note text cannot be blank.');
await context.log('info', 'Note normalized');
return { text };
},
});
Separate configuration from input
The formatting block has a prefix configuration default. Configuration is resolved from the block occurrence, while text arrives through a named connection. Editor hints can help a host render a form; they do not change runtime validation.
import { defineBlock } from '@db3.ai/app/flows';
/** Formats validated text without an external provider or irreversible side effect. */
export const uppercaseTextBlock = defineBlock<{ text: string }, { text: string }, { prefix: string }>({
type: 'notes.uppercase', name: 'Uppercase note',
inputs: { text: { type: 'string', required: true } },
outputs: { text: { type: 'string', required: true } },
config: { prefix: { type: 'string', default: '' } },
/** Applies the snapshotted configuration to one deterministic output. */
run(input, context) { return { text: `${context.config.prefix}${input.text.toUpperCase()}` }; },
});
Keep the graph as source data
The definition contains stable ULIDs, ports, connections and positions. Preserve IDs when editing an existing graph. saveDefinition() validates before writing; pass expectedRevision to detect a stale file-store edit.
The current compiler accepts one connected sequential path. Branches, merges and cycles are rejected. A visual position does not define execution order. The file store writes deterministic JSON and uses content revisions; it is not a multi-writer database transaction or untrusted filesystem sandbox.
import type { FlowDefinition } from '@db3.ai/app/flows';
import { ulid } from '@db3.ai/pure/ulid';
/** Creates a two-block sequential definition with stable IDs retained on edits. */
export function createTextFlow(): FlowDefinition {
const normalize = ulid();
const uppercase = ulid();
return {
schemaVersion: 1, id: ulid(), name: 'Normalize note text',
inputs: { text: { type: 'string', required: true } }, outputs: { text: { type: 'string', required: true } },
blocks: [{ id: normalize, type: 'notes.normalize', position: { x: 0, y: 0 } }, { id: uppercase, type: 'notes.uppercase', config: { prefix: '' }, position: { x: 250, y: 0 } }],
connections: [{ id: ulid(), sourceBlockId: normalize, sourcePort: 'text', targetBlockId: uppercase, targetPort: 'text' }],
};
}
Dispatch and process the steps
run() creates durable run/step state and dispatches a flow-step job to flows. The lab drains ready jobs with a finite safety limit. Production needs a supervised queue worker consuming that same queue.
runDetails() returns the run, ordered steps and lifecycle/log events. Captured input/output has a default one-megabyte boundary. There is no automatic secret redaction or tenant authorization; scope any history/API access in your app.
import { mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { pathToFileURL } from 'node:url';
import { createGeneratedTestDatabase } from '@db3.ai/app/db/test/db';
import { FLOW_MODELS } from '@db3.ai/app/flows';
import { FailedJob, QueuedJob } from '@db3.ai/app/queue';
import { FlowExampleApp } from './FlowExampleApp';
import { createTextFlow } from './createTextFlow';
/** Drains only ready jobs, with an explicit finite guard for this small lab. */
async function drain(application: FlowExampleApp): Promise<void> {
for (let jobs = 0; jobs < 20; jobs++) if (!await application.queue.processNextJob('flows')) return;
throw new Error('Flow lab exceeded its 20-job safety limit.');
}
/** Runs a real durable graph, rejects invalid input and replays original/latest definitions. */
export async function runTextFlow() {
const root = await mkdtemp(join(tmpdir(), 'db3-flow-guide-'));
try {
const database = await createGeneratedTestDatabase('flow_guide');
const application = new FlowExampleApp({ db: database.db, queue: { queue: 'flows', retryDelaySeconds: 0, queueMonitor: false } }, root);
try {
await application.db.install(QueuedJob, FailedJob, ...FLOW_MODELS);
const flow = createTextFlow();
const stored = await application.flows.saveDefinition(flow);
const original = await application.flows.run(flow.id, { text: ' Client note ' });
await drain(application);
const originalDetails = await application.flows.runDetails(original.id!);
const failed = await application.flows.run(flow.id, { text: ' ' });
await drain(application);
const failure = await application.flows.runDetails(failed.id!);
const changed = structuredClone(flow);
changed.blocks[1]!.config = { prefix: 'Updated: ' };
await application.flows.saveDefinition(changed, { expectedRevision: stored.revision });
const replay = await application.flows.replay(failed.id!, { definition: 'original', input: { text: 'Fixed note' } });
const latest = await application.flows.replay(original.id!, { definition: 'latest' });
await drain(application);
const replayed = await application.flows.runDetails(replay.id!);
const latestDetails = await application.flows.runDetails(latest.id!);
return { status: originalDetails.run.status, output: originalDetails.run.output, steps: originalDetails.steps.length, logged: originalDetails.events.some(event => event.message === 'Note normalized'), failure: failure.run.status, failureMessage: failure.run.error?.message, repaired: replayed.run.output, latest: latestDetails.run.output, replayLinked: replayed.run.replayOf?.id === failed.id };
} finally { try { await application.close(); } finally { await database.destroy(); } }
} finally { await rm(root, { recursive: true, force: true }); }
}
if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) console.log(JSON.stringify(await runTextFlow(), null, 2));
Repair, then choose what to replay
Replay creates a new linked run. definition: "original" uses the stored graph snapshot; "latest" resolves current source. You can supply replacement input explicitly. The lab changes the second block configuration and proves the two modes differ.
The snapshot contains definitions, not a copy of executable JavaScript. Replaying an old graph still uses registered runtime code. Version block types when code changes would break historic behavior. Side-effecting blocks need idempotency: replay can repeat work.
Compose a child flow
Structural flow.input, flow.output and flow.subflow blocks are registered by the service. A subflow occurrence supplies a child flowId; its public ports come from the child definition.
The parent waits while a linked child run executes. Nested definitions are captured for original replay; latest replay resolves current source. The service integration tests cover nested waiting/failure/replay, but the first lab stays a two-block path. A nested-flow cookbook and designer integration remain follow-ups.
Testing
Create a tests/flows directory and save the test below as runTextFlow.test.ts. It imports the example you copied into examples/. Keep the same folder layout so the relative import resolves.
Run from the application root with the development dependencies from Installation. These are consumer tests, not commands that assume a framework checkout. This database lab needs the same test-only SQL credentials when run through Vitest.
import { expect, it } from 'vitest';
import { runTextFlow } from '../../examples/runTextFlow';
it('runs a persisted graph, captures safe progress and repairs failed work through explicit replay', async () => {
expect(await runTextFlow()).toEqual({ status: 'completed', output: { text: 'CLIENT NOTE' }, steps: 2, logged: true, failure: 'failed', failureMessage: 'Note text cannot be blank.', repaired: { text: 'FIXED NOTE' }, latest: { text: 'Updated: CLIENT NOTE' }, replayLinked: true });
});
Run and extend the test
Change the prefix or add another pure block. Keep the failure, repair, replay-link and original/latest assertions. This lab uses one attempt to make terminal failure immediate; configure retry policy for the actual operation rather than copying that as a production default.
npx vitest run tests/flows/runTextFlow.test.ts
npx tsc --noEmit --target ES2022 --module ESNext --moduleResolution Bundler --types node --skipLibCheck examples/*.tsCoverage and API
Tested: real file definitions and SQL queue, App boot, two blocks, input/business validation, progress, terminal failure and replay modes. Explained/reference: nested graphs, editor contracts, payload bounds, revisions and production safety. Branching, generic SQL execution, credentials in captured values and designer UI are not promised.
Persists and runs a two-block graph, inspects steps/logs, fails blank input and replays original/latest definitions.
The guide test passes against the real framework components.packages/app/src/flows/tests/examples/runTextFlow.test.tsThis test command requires the framework repository. Use the walkthrough commands in an installed application.
Environment: Disposable SQL queue, temporary file definition store, no provider or graph UI.