Observability

Turbine reports every statement it runs through query events ($on) and every line it logs through one logger. Optional per-minute metrics ($observe) feed a local dashboard, and an OpenTelemetry recipe builds traces from the same hooks. None of it needs an extra dependency.

Query events, db.$on('query')#

Every statement emits an event after it runs, whether it succeeded or failed. Unsubscribe with $off.

db.$on('query', (event) => {
  if (event.duration > 200) {
    console.warn(`slow: ${event.model}.${event.action} took ${event.duration.toFixed(1)}ms`);
  }
});

interface QueryEvent {
  sql: string;        // the generated SQL text
  params: unknown[];  // bound parameter values (redacted by default, see below)
  duration: number;   // wall-clock ms (performance.now)
  model: string;      // table name, e.g. 'users'
  action: string;     // operation, e.g. 'findMany'
  rows: number;       // rows returned / affected
  timestamp: Date;
  error?: Error;      // present when the query failed
  strategy?: 'auto-batched';            // the 'auto' relation strategy re-planned this query
  tag?: string;                         // the enclosing db.$tag() label, see below
  batch?: 'pipeline' | 'transaction';   // ran inside pipeline() or $transaction([...])
  comment?: string;                     // the sqlComment it was sent with; sql stays commentless
  retried?: true;                       // the failed first attempt of a read Turbine re-sent
}

• A listener that throws never fails the query. With logging: true, the error is logged as query-listener-error.
• $on is a passive tap. Middleware wraps a call and can transform its result.

What emits an event#

• '$raw' is exported as RAW_QUERY_MODEL. BEGIN, COMMIT and savepoints emit no event.
• A pipeline sends its statements in one round trip, so each event's duration is an even share of the batch time.
• A read that Turbine re-sent after a lost connection emits two events: the failed attempt with retried: true, then the retry.

Attributing queries to a feature, db.$tag()#

model and action say what ran, not why. Wrap a code path in $tag and every event inside it carries the label:

const order = await db.$tag('checkout', async () => {
  const cart = await db.carts.findUnique({ where: { id: cartId } });
  await db.raw`SELECT pg_advisory_xact_lock(${cart.id})`;
  return db.orders.create({ data: { cartId: cart.id } });
});
// all three events carry tag: 'checkout'

• The scope follows async context, so awaited helpers, transactions and concurrent requests keep their own tags. An inner $tag replaces the outer one.
• $tag returns what the callback returns.
• A label is 1 to 128 characters (MAX_QUERY_TAG_LENGTH), or it throws ValidationError (E003).
• Name a code path, not a request: a label lands on every event, and with sqlComment.tag in the SQL too.

Seeing the parameter values, logQueryParams#

Params are redacted by default: each entry of event.params is '[REDACTED]'. Turn real values on with logQueryParams:

const db = turbine({
  connectionString: process.env.DATABASE_URL,
  logQueryParams: process.env.NODE_ENV !== 'production',
});
 
db.$on('query', (e) => console.log(e.sql, e.params));
// SELECT * FROM "users" WHERE "email" = $1 LIMIT 1   [ 'alice@example.com' ]

Only event.params changes. When logQueryParams is unset, errorMessages: 'verbose' also reveals params.

Log output, logger#

Every line Turbine writes goes through one logger, console by default, with each line prefixed [turbine]. Pass your own:

import type { TurbineLogCode, TurbineLogEntry, TurbineLogger } from 'turbine-orm';
 
const logger: TurbineLogger = {
  warn(message: string, entry: TurbineLogEntry) {
    if ((entry.code as TurbineLogCode) === 'unordered-page') return; // known and accepted
    appLog.warn(message, entry);
  },
  error: (message, entry) => appLog.error(message, entry),
};
 
const db = turbine({ connectionString: process.env.DATABASE_URL, logger });

The entry carries a stable code, plus table, operation or error when relevant. Match on code; the message text can change.

• debug, info, warn and error are all optional. A missing method drops that level, so logger: {} silences Turbine.
• A logger method that throws is ignored.
• console and winston take the message first and work as is. pino and bunyan take the object first, so wrap them: warn: (message, entry) => pinoLog.warn(entry, message).
• No log line carries a bound value, whatever logQueryParams says.
• A once-per-process warning is deduplicated across clients.
• logging: true adds lifecycle lines at debug level.
• introspect() and generate() take the same logger option.

Log codes#

A row that names no condition logs in every mode. Every engine, PowDB included, logs through the client's logger.

Diagnostics, diagnostics#

diagnostics controls per-query advice and a self-check of Turbine's SQL cache.

const db = turbine({
  connectionString: process.env.DATABASE_URL,
  diagnostics: 'production',
});

When unset, the mode is 'production' if NODE_ENV is 'production' or there is no process (an edge runtime), else 'development'. It is read once, at construction. Set it explicitly where NODE_ENV is unset, such as a worker, or that runtime runs in development mode.

An object overrides one part:

diagnostics: { mode: 'production', advice: true }
diagnostics: { mode: 'production', cacheCheck: { sample: 0.001 } }

• cacheCheck is 'every-hit', 'off' or { sample } with sample in (0, 1]. A mismatch is a Turbine bug: the query throws InvariantError before anything is sent.
• An invalid mode, key or sample throws ValidationError (E003) at construction.
• When cacheCheck is unset, TURBINE_DISABLE_CACHE_CHECK=1 turns the check off, and TURBINE_CACHE_CHECK_SAMPLE sets a sample rate in production mode.
• Construction warnings (a misspelled config key, prepared statements behind a pooler) fire in every mode.

SQL comments, sqlComment#

sqlComment appends sqlcommenter key/values to each statement, so pg_stat_activity, server logs and database APM tools can tell which service, model or feature sent it.

const db = turbine({
  connectionString: process.env.DATABASE_URL,
  sqlComment: {
    static: { application: 'billing' }, // the same pairs on every statement
    model: true, // the table and the method
    tag: true, // the label of the innermost db.$tag()
  },
});
 
await db.$tag('checkout', () => db.users.findUnique({ where: { id: 1 } }));

SELECT "users".* FROM "users" WHERE "id" = $1 LIMIT 1 /*action='findUnique',application='billing',model='users',tag='checkout'*/

• Values are URL-encoded, so no value can end the comment. Keys are lower-case letters, digits and underscores, starting with a letter.
• A bad static value, a value over 256 characters or an unknown option throws ValidationError (E003) at construction. A bad dynamic value is left out and logged once as sql-comment-dropped; the query still runs.
• Model-method statements, pipelines and transaction batches carry comments. Raw SQL and transaction control do not.
• Events keep sql without the comment and put it in e.comment. Cache hit rates do not change.
• A statement carrying any dynamic value is sent unnamed, so it is parsed and planned every time. static, model and tag comments keep prepared statements.
• dynamic is refused on MySQL (UnsupportedFeatureError, E017), and sqlComment on PowDB.

Metrics, db.$observe()#

$observe aggregates query events per minute and writes them to a _turbine_metrics table. Point it at a separate database:

const handle = await db.$observe({
  connectionString: process.env.TURBINE_OBSERVE_URL!, // metrics DB
  flushIntervalMs: 60_000, // default: 60s
  retentionDays: 30,       // default: 30
});
 
// on shutdown
await handle.stop(); // final flush, then closes the metrics pool

• Each flush writes count, avg, p50, p95, p99 and errors per model:action and minute. Rows merge additively, so several app instances can share a table.
• It uses its own one-connection pool and creates the table on start. Rows older than retentionDays are pruned on each flush.
• Flush errors are swallowed, and the timer does not keep the process alive.
• Calling $observe again replaces the previous one.
• A batch holds only model, action, counts and latencies: no SQL text and no parameter values. Nothing is collected until you call $observe.

CREATE TABLE IF NOT EXISTS _turbine_metrics (
  id          BIGSERIAL PRIMARY KEY,
  bucket      TIMESTAMPTZ NOT NULL,  -- minute bucket
  model       TEXT NOT NULL,
  action      TEXT NOT NULL,
  count       INTEGER NOT NULL DEFAULT 0,
  avg_ms      REAL NOT NULL DEFAULT 0,
  p50_ms      REAL NOT NULL DEFAULT 0,
  p95_ms      REAL NOT NULL DEFAULT 0,
  p99_ms      REAL NOT NULL DEFAULT 0,
  error_count INTEGER NOT NULL DEFAULT 0,
  UNIQUE(bucket, model, action)
);

Custom sinks#

Pass a sink to send the aggregates elsewhere. With a sink, connectionString is optional.

interface MetricsFlushBatch {
  rows: Array<{
    bucket: Date;   // the minute bucket
    model: string;  // table name
    action: string; // operation
    count: number;
    avg: number;
    p50: number;
    p95: number;
    p99: number;
    errors: number;
  }>;
}
 
interface ObserveSink {
  init?(): Promise<void>;                     // once at startup
  flush(batch: MetricsFlushBatch): Promise<void>; // each aggregate batch
  stop?(): Promise<void>;                     // on shutdown
}

HttpJsonSink POSTs each batch as JSON to an HTTP endpoint. A failed request drops that batch without retrying.

import { HttpJsonSink } from 'turbine-orm';
 
const handle = await db.$observe({
  sink: new HttpJsonSink({
    url: 'https://metrics.internal/ingest',
    headers: { authorization: `Bearer ${process.env.METRICS_TOKEN}` },
    // fetchFunction: customFetch, // optional override
  }),
});

Dashboard, npx turbine observe#

A local, read-only dashboard over _turbine_metrics: latency over time and a breakdown per model.action.

npx turbine observe --metrics-url postgres://...
TURBINE_OBSERVE_URL=postgres://... npx turbine observe --port 5000 --no-open

It binds 127.0.0.1:4984 and reads metrics only, never your application database. It shares Studio's security model: a non-loopback host needs --allow-remote, the printed URL holds a one-time ticket, and a foreign Host header gets 403.

OpenTelemetry#

Turbine imports nothing from @opentelemetry/*. $use opens a span per operation and $on('query') records a span per SQL statement. Install @opentelemetry/api and an SDK; the recipe needs the context manager the SDK registers.

import { context, trace, SpanKind, SpanStatusCode } from '@opentelemetry/api';
import { RAW_QUERY_MODEL } from 'turbine-orm';
import { db } from './db.js';
 
const DB_SYSTEM = 'postgresql'; // 'mysql', 'sqlite' or 'microsoft.sql_server' on the other engines
const tracer = trace.getTracer('turbine-orm');
 
// One span per Turbine operation. The statements it runs become its children.
db.$use((params, next) =>
  tracer.startActiveSpan(`${params.model}.${params.action}`, { kind: SpanKind.INTERNAL }, async (span) => {
    try {
      return await next(params);
    } catch (err) {
      span.recordException(err as Error);
      span.setStatus({ code: SpanStatusCode.ERROR });
      throw err;
    } finally {
      span.end();
    }
  }),
);
 
// One span per SQL statement, back-dated from the event's duration.
db.$on('query', (e) => {
  const end = e.timestamp.getTime();
  const table = e.model === RAW_QUERY_MODEL ? undefined : e.model;
  const span = tracer.startSpan(
    table ? `${e.action} ${table}` : e.action,
    {
      kind: SpanKind.CLIENT,
      startTime: end - e.duration,
      attributes: {
        'db.system.name': DB_SYSTEM,
        'db.operation.name': e.action,
        'db.query.text': e.sql,
        ...(table ? { 'db.collection.name': table } : {}),
        ...(e.tag ? { 'turbine.tag': e.tag } : {}),
        ...(e.batch ? { 'turbine.batch': e.batch } : {}),
        ...(e.retried ? { 'turbine.retried': true } : {}),
      },
    },
    context.active(),
  );
  if (e.error) {
    span.setAttribute('error.type', e.error.name);
    span.setStatus({ code: SpanStatusCode.ERROR, message: e.error.message });
  }
  span.end(end);
});

A request that creates an order produces this tree:

POST /orders                 your span
  orders.create              INTERNAL, from $use
    create orders            CLIENT, from $on('query')

The statement span is CLIENT and the operation span INTERNAL, so a backend counts each database call once. The operation span is wider: it also covers SQL generation and row parsing.

Why the tree comes out right#

A query event is delivered in the async context of the code that awaited the statement, so context.active() in the listener is the span active at that await. You pass nothing through Turbine.

$use wraps model methods only. These statements get no operation span and nest directly under whatever span is active where you await them:

• A lazy db.sql query runs where it is awaited, not where it was built.
• A transaction adds no span of its own. Wrap it in your own span if you want one.
• With relationLoadStrategy: 'batched' or the 'auto' fallback, every follow-up statement is a child of the one operation span. Take the table from db.query.text, since a follow-up carries its parent query's model.

What the attributes mean#

The db.* and error.type names follow the stable OpenTelemetry database semantic conventions (v1.33.0 and later, where db.system became db.system.name and db.statement became db.query.text). Pin the version your backend expects. The turbine.* attributes are Turbine's own.

• Duration covers the driver round trip and the wait for a pooled connection. Each span ends at the event's timestamp.
• Pipeline spans share the batch time evenly; see What emits an event.
• Parameter values never reach db.query.text, except with db.rawUnsafe, where you write the text. The recipe does not record e.params; the conventions treat db.query.parameter.<key> as opt-in.
• A retried read shows as two statement spans. Alert on the operation span to count failed calls, not attempts.
• recordException sends the error's message and stack to your backend. See errorMessages.

Carrying the trace into the database#

To link a statement in pg_stat_activity or the server log back to its trace, add the active span's W3C traceparent as a sqlComment:

import { trace, TraceFlags } from '@opentelemetry/api';
import type { SqlCommentOptions } from 'turbine-orm';
 
const sqlComment: SqlCommentOptions = {
  dynamic: () => {
    const ctx = trace.getActiveSpan()?.spanContext();
    // Unsampled: add nothing, so the statement keeps its prepared statement.
    if (!ctx || (ctx.traceFlags & TraceFlags.SAMPLED) === 0) return undefined;
    return { traceparent: `00-${ctx.traceId}-${ctx.spanId}-${ctx.traceFlags.toString(16).padStart(2, '0')}` };
  },
};
 
const db = turbine({ sqlComment });

SELECT "orders".* FROM "orders" WHERE "id" = $1 LIMIT 1 /*traceparent='00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01'*/

• A sampled statement is sent unnamed and planned every time, which costs tens of microseconds per statement. Returning undefined for unsampled spans keeps the rest prepared.
• db.query.text and the event's sql stay commentless.
• pg_stat_statements keeps one row per statement, since a comment is not part of its query id.
• Raw SQL carries no comment, dynamic is refused on MySQL, and sqlComment on PowDB.

See also#

• Studio: the local database UI with the same security model.
• Middleware: the active counterpart to $on.
• Typed Errors: what lands in event.error.