@flowrensic/node

Pre-release. This package is an early preview and not ready for production. Its API can change in any version.

Replay the histories your workflows already produced, against the code you are about to ship.

Flowrensic keeps what a Temporal Engine recorded. This package fetches a set to disk and reports back what your replay found. The replay itself stays yours: call Temporal's replayer directly, in your suite, with your own bundler options. Histories in a repo go stale on the next deploy and nobody can say which are still safe to delete.

Server: flowrensic/flowrensic on Docker Hub. Temporal is the supported engine.

Who the caller is

Every call to Flowrensic says who sends it, in three headers: Flowrensic-Current-Env, Flowrensic-Env-Access-Key and Flowrensic-Current-Ephemeral-Run. Each helper here takes them as the options currentEnv, envAccessKey and currentEphemeralRun, beside serverBaseUrl. The package reads no environment variable: pass the values from your own config. An env of kind ephemeral takes currentEphemeralRun, such as the CI build number, and without one the run is default.

Replay

import { fetchReplaySet, publishVerdicts, readReplaySet } from '@flowrensic/node';
import { historyFromJSON } from '@temporalio/common/lib/proto-utils.js';
import { Worker } from '@temporalio/worker';

await fetchReplaySet({
  serverBaseUrl: 'http://flowrensic:8080',
  replayHistoryWorkflow: 'OrderWorkflow',
  replayHistoryOutDir: 'replay-set',
  replayHistoryFromEnvs: ['prod', 'staging'],
  currentEnv: 'ci',
  envAccessKey: config.envAccessKey,
  currentEphemeralRun: config.buildNumber,
});

const set = await readReplaySet('replay-set');
const histories = await Promise.all(
  set.histories.map(async (entry) => ({
    workflowId: entry.run_id,
    history: historyFromJSON(JSON.parse(await readFile(join('replay-set', entry.file), 'utf8'))),
  })),
);

const verdicts = [];
for await (const result of Worker.runReplayHistories({ workflowsPath }, histories)) {
  verdicts.push({
    runId: result.workflowId,
    workflowType: set.workflow_type,
    passed: result.error === undefined,
    error: result.error?.message,
  });
}

await publishVerdicts({
  serverBaseUrl: 'http://flowrensic:8080',
  codeVersion: config.gitSha,
  verdicts,
  currentEnv: 'ci',
  envAccessKey: config.envAccessKey,
  currentEphemeralRun: config.buildNumber,
});

The set holds the histories of each env named, one env after the other, and replay-set.json lists the envs and gives each history its env. Each env named must be the current env or one its entry in FLOWRENSIC_KNOWN_ENVS may read. The current env must be of kind ephemeral: every fetch belongs to an ephemeral run, and a persistent env is refused.

Verdicts go back against the run and the version, so a break points at the change that caused it.

flowrensic replay fetch in the CLI writes the same directory and the same manifest, for a pipeline that would rather fetch in a step than in a test.

The server picks the set: recent histories, versions still running somewhere, outcomes that differ. Newest-ten on a healthy workflow type is ten completions and not one failure.

Activity telemetry

import { createAttemptSink, flowrensicInterceptor } from '@flowrensic/node';

const sink = createAttemptSink({ serverBaseUrl: 'http://flowrensic:8080', currentEnv: 'prod', envAccessKey, workerName: 'orders-worker' });

const worker = await Worker.create({
  // ...
  interceptors: { activity: [flowrensicInterceptor(sink)] },
});

workerName names the worker, the same name for every copy of it. The Workflow types page shows it beside each type the worker runs, with the task queue, and can filter by it. Left out, the worker has no name.

One event per attempt, not per activity: the gap between the two is the retry count. Each attempt carries the namespace Temporal gives the activity, and the server keeps the namespaces of each environment apart. There is no namespace setting.

Activity inbound only, and publishing failures are swallowed. Telemetry must never fail the activity it measures.

Query and signal counts

The Workflow types page counts the queries each workflow type answers. The worker needs the workflow module and the sinks:

import { createRequire } from 'node:module';
import { flowrensicQuerySinks } from '@flowrensic/node';

const queries = flowrensicQuerySinks({ serverBaseUrl: 'http://flowrensic:8080', currentEnv: 'prod', envAccessKey });

const worker = await Worker.create({
  // ...
  interceptors: {
    activity: [flowrensicInterceptor(sink)],
    workflowModules: [createRequire(import.meta.url).resolve('@flowrensic/node/workflow')],
  },
  sinks: queries.sinks,
});
// Before the process ends:
await queries.buffer.close();

The client interceptor sends the queries and signals that fail or get no answer. Put it in the plugins of the client, so it reads the namespace of the client:

const interceptor = flowrensicClientInterceptor({ serverBaseUrl: 'http://flowrensic:8080', currentEnv: 'prod', envAccessKey });
const client = new Client({ connection, namespace: 'orders', plugins: [interceptor] });
// Before the process ends:
await interceptor.close();

Peers

@temporalio/worker, @temporalio/common, @temporalio/activity, @temporalio/workflow, 1.24 or newer, for the interceptors. The replay runs on your SDK, not on one this package pins.