OpenTelemetry

@naikidev/commiq-otel instruments commiq stores with OpenTelemetry tracing. Each command becomes a span, and events emitted during that command are recorded as span events, giving end-to-end visibility in any OTel-compatible backend.

Installation

pnpm add @naikidev/commiq-otel @opentelemetry/api

npm install @naikidev/commiq-otel @opentelemetry/api

yarn add @naikidev/commiq-otel @opentelemetry/api

bun add @naikidev/commiq-otel @opentelemetry/api

@opentelemetry/api and @naikidev/commiq are both peer dependencies. Core is a peer rather than a direct dependency because a nested second copy of core meant otel instrumented a different store instance than the one the app used, and spans silently stopped appearing.

Basic usage

import { createCommandDef, createStore, sealStore } from "@naikidev/commiq";
import { instrumentStore } from "@naikidev/commiq-otel";

const increment = createCommandDef("increment");
const store = createStore({ count: 0 });
store.addCommandHandler(increment, (ctx) => {
  ctx.setState({ count: ctx.state.count + 1 });
});

const instrumentation = instrumentStore(store, { storeName: "counter" });

const sealed = sealStore(store);
sealed.queue(increment);

instrumentStore(store, options)

function instrumentStore(
  store: Streamable,
  options: InstrumentOptions,
): StoreInstrumentation;

store needs only openStream/closeStream, so a StoreImpl or a SealedStore both work.

StoreInstrumentation is both callable and a Disposable:

instrumentation();          // uninstrument
instrumentation.destroy();  // identical

Both are idempotent and interchangeable. The callable form keeps v1 call sites working; destroy() lets otel join the same generic teardown loop as every other commiq plugin:

import type { Disposable } from "@naikidev/commiq";

const plugins: Disposable[] = [instrumentation, devtools, effects];
for (const plugin of plugins) plugin.destroy();

Disposal removes the stream listener and ends any in-flight spans as abandoned.

InstrumentOptions

Cross-store propagation

Each instrumentStore call keeps its own state. Two independent stores never share span parents by default, so interleaved work — SSR, multi-tenant Node, concurrent requests — cannot fabricate causal edges.

To trace a causal chain that genuinely crosses stores, pass one shared registry to every participant:

import { createTraceRegistry, instrumentStore } from "@naikidev/commiq-otel";

const registry = createTraceRegistry({ maxEntries: 512 });

instrumentStore(orders, { storeName: "orders", registry });
instrumentStore(payments, { storeName: "payments", registry });

A command queued from an event handler in another store is then parented to the span of the command that emitted the event — even if that command has already finished.

Scope the registry to the trust boundary. One registry per request or per tenant on a server; a module-level registry is fine in a browser, where there is only one user.

TraceRegistry

The registry stores immutable SpanContext values — never live Span objects. In v1 every non-command event registered an alias holding a strongly-referenced ended Span, and nothing removed it: roughly 72,000 retained entries an hour at 20 events a second, and far worse now that stateChanged fires per setState.

Tracing model

Each command creates a span living from commandStarted to commandHandled, commandHandlingError or commandInterrupted:

commiq.command:increment
  ├─ span event: commiq.correlation
  ├─ span event: stateChanged
  └─ span event: itemAdded

Events emitted during a command are recorded as span events on that command's span. Events with no resolvable live parent create short standalone commiq.event:<name> spans.

What gets recorded

eventHandlingError and unhandledError come from core's error channel and are new in v2. A throwing event handler no longer suppresses commandHandled, so the command span still closes OK while the failure gets its own error span — the command genuinely succeeded; a downstream reaction did not.

Span attributes

Command spans (commiq.command:<name>):

Standalone event spans (commiq.event:<name>):

Error spans: commiq.event_handler:<name> carries commiq.store and commiq.event.name; commiq.error:<source> carries commiq.store and commiq.error.source.

Correlation ids are not in this table by default — see Cardinality and PII.

Abandoned commands

A command that never settles — a hung await, navigation mid-flight — would leave an unended span, and unended spans are never exported. The whole trace silently vanished in exactly the case you most need it.

Such spans are now ended with SpanStatusCode.ERROR plus:

• commiq.command.abandoned — true
• commiq.command.abandoned_reason — "timeout", "overflow" or "disposed"

The sweep timer only exists while commands are in flight and is cleared on disposal. Set maxCommandDurationMs: 0 to disable it — accepting that a hung command then loses its trace.

An abandoned span is a real signal in your backend: an ERROR span with commiq.command.abandoned=true means a command in your app never finished. Alert on it.

Cardinality and PII

Span names are low-cardinality: commiq.command:<name>, commiq.event:<name>.

Correlation ids

Correlation ids are high-cardinality by nature — one distinct value per span — which drives cost on backends that index attributes. By default they are recorded on a commiq.correlation span event instead:

caused_by is omitted when the cause is null.

Set recordCorrelationIds: true to record them as span attributes instead, if your backend does not index or derive metrics from attributes. That was v1's behaviour.

Error text

import { defaultSanitizeError, instrumentStore } from "@naikidev/commiq-otel";

instrumentStore(store, {
  storeName: "users",
  sanitizeError: (error) => (error instanceof Error ? error.message : defaultSanitizeError(error)),
});

A sanitizeError that throws, or returns a non-string, falls back to the error type — it cannot break instrumentation.

No command payload and no state is ever attached to spans, in either version.

Cleanup

const instrumentation = instrumentStore(store, { storeName: "counter" });

instrumentation.destroy();

Disposal removes the stream listener and ends every in-flight span with abandoned_reason: "disposed", so partial traces still reach the backend.

Full example

import { NodeSDK } from "@opentelemetry/sdk-node";
import { ConsoleSpanExporter } from "@opentelemetry/sdk-trace-node";
import { createCommandDef, createStore, sealStore } from "@naikidev/commiq";
import { createTraceRegistry, instrumentStore } from "@naikidev/commiq-otel";

const sdk = new NodeSDK({ traceExporter: new ConsoleSpanExporter() });
sdk.start();

const increment = createCommandDef("increment");
const store = createStore({ count: 0 });
store.addCommandHandler(increment, (ctx) => {
  ctx.setState({ count: ctx.state.count + 1 });
});

const registry = createTraceRegistry();
const instrumentation = instrumentStore(store, {
  storeName: "counter",
  tracerName: "my-app",
  registry,
});

const sealed = sealStore(store);
await sealed.queue(increment);

instrumentation.destroy();
await sdk.shutdown();

queue() returns a CommandHandle that never rejects, so awaiting it is safe without a try/catch — it resolves with a status of "handled", "failed", "interrupted", "invalid" or "discarded".

Exported types

InstrumentOptions, StoreInstrumentation, TraceRegistry, TraceRegistryOptions, TraceLink, ErrorSanitizer.

import type { ErrorSanitizer, InstrumentOptions } from "@naikidev/commiq-otel";

const sanitize: ErrorSanitizer = (error) => (error instanceof RangeError ? "RangeError" : "Error");
const otelOptions: InstrumentOptions = { storeName: "cart", sanitizeError: sanitize };

Testing locally

ConsoleSpanExporter is enough for quick debugging without any backend.