Metrics

Custom integrations can implement HookMetricEmit directly or use river.HookMetricEmitFunc. Use a type switch with a default case so that the integration continues to work when a future River version adds another metric type. The current payload types are described in the next section.

metricChan := make(chan rivertype.Metric, 100)

metricPlugin := river.HookMetricEmitFunc(func(
    _ context.Context,
    params *rivertype.HookMetricEmitParams,
) {
    select {
    case metricChan <- params.Metric:
    default:
        // Drop the metric rather than block River's job-fetch loop.
    }
})

go func() {
    for metric := range metricChan {
        switch metric := metric.(type) {
        case *rivertype.JobGetAvailableDurationMetric:
            recordFetchDuration(metric.Queue, metric.Duration)

        case *rivertype.JobGetAvailableCountMetric:
            recordFetchCount(metric.Queue, metric.Count)

        default:
            // Ignore metrics added by future River versions.
        }
    }
}()

riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    Plugins: []rivertype.Plugin{metricPlugin},
    // ...
})

Metric hooks run synchronously in hot paths, including while River is fetching jobs. They must not perform network access, persistence, or expensive processing inline. Use a metrics library that records asynchronously, or hand metrics to a bounded channel as shown above. If the channel is full, dropping a metric is preferable to slowing down the job-fetch loop.

Time worker execution with a plugin, or subscribe to completed and failed jobs. Ruby does not currently emit the internal fetch metrics exposed by Go, Rust, and TypeScript:

class TimingPlugin
  def work(job, operation)
    started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
    operation.call
  ensure
    Metrics.observe(job.kind, Process.clock_gettime(Process::CLOCK_MONOTONIC) - started)
  end
end

config = River::Config.new(plugins: [TimingPlugin.new])

Metrics represents your application’s metrics library; use bounded, nonblocking recording.

Implement Hook::metric_emit to forward metric values to your monitoring system. Match the variants you need and ignore the rest:

struct MetricsHook;

impl riverqueue::Hook for MetricsHook {
    async fn metric_emit(&self, metric: riverqueue::Metric) -> Result<(), riverqueue::BoxError> {
        match metric {
            riverqueue::Metric::JobGetAvailableDuration(duration) => record_fetch_duration(duration),
            riverqueue::Metric::JobGetAvailableCount(count) => record_fetch_count(count),
            _ => {},
        }

        Ok(())
    }
}

let client = riverqueue::Client::builder(pool).hook(MetricsHook).build()?;

The recording functions are application-defined. Keep callbacks cheap and avoid blocking I/O in the fetch path.

Use the onMetric hook to forward measurements to your monitoring system. This example records fetch duration and the number of jobs fetched:

const client = new Client(driver, {
  hooks: {
    onMetric: (metric) => {
      if (metric.name === "job_get_available_duration") {
        recordFetchDuration(metric.queue, metric.duration.total("milliseconds"));
      } else if (metric.name === "job_get_available_count") {
        recordFetchCount(metric.queue, metric.count);
      }
    },
  },
});

The same values are published through Node’s riverqueue:metric diagnostics channel. The recording functions are application-defined.

The example above handles the two metric payloads River currently emits after a successful job-fetch operation:

• JobGetAvailableDurationMetric: How long it took to lock a batch of available jobs. Its fields contain duration and queue name. The timing of this metric growing out of hand is an early sign of database degradation (e.g. from a long-running query) that may lead to queue degradation.
• JobGetAvailableCountMetric: How many jobs were locked in the batch. Its fields contain the count and queue name.

Each payload implements rivertype.Metric and has a corresponding name, MetricNameJobGetAvailableDuration or MetricNameJobGetAvailableCount. More metric types may be added in future River versions.

Use plugins for attempt durations and subscriptions for job outcomes. Ruby has no dedicated internal metric hook yet.

Metric::JobGetAvailableDuration measures successful fetch time; Metric::JobGetAvailableCount reports the number claimed. These variants carry a duration or count, without a queue label. Keep a wildcard arm for future metric variants.

job_get_available_duration and job_get_available_count include the queue. job_completion_requeued and job_completion_dropped report failures to persist completions. See the metric reference.

For a ready-made integration that handles these payloads, install otelriver:

go get -u github.com/riverqueue/rivercontrib/otelriver

Add it to Config.Plugins:

import (
    "github.com/riverqueue/river"
    "github.com/riverqueue/river/riverdriver/riverpgxv5"
    "github.com/riverqueue/river/rivertype"
    "github.com/riverqueue/rivercontrib/otelriver"
)

riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    Plugins: []rivertype.Plugin{
        otelriver.NewMiddleware(nil),
    },
    // ...
})
if err != nil {
    panic(err)
}

Despite its constructor’s name, otelriver.NewMiddleware implements both River middleware and HookMetricEmit. Registering it under Config.Plugins activates both interfaces, enabling its existing insertion and work instrumentation along with the internal River metrics. An existing configuration under Config.Middleware should be moved to Config.Plugins to receive the internal metrics.

otelriver exports the job-fetch payloads as:

• river.job_get_available_duration: A histogram of successful job-fetch durations, tagged by queue.
• river.job_get_available_count: A histogram of the number of jobs locked by successful fetches, tagged by queue.

See OpenTelemetry for provider configuration and the other traces and metrics available through otelriver.

otelriver is a Go package. In Ruby, instrument a work plugin with your application’s OpenTelemetry library and propagate trace context through job metadata. There is no bundled Ruby equivalent yet.

otelriver is a Go package. In Rust, use River’s hooks and middleware to connect to your application’s tracing or OpenTelemetry stack. There is no bundled Rust equivalent yet.

otelriver is a Go package. In TypeScript, use work middleware to create spans and onMetric or Node diagnostics channels to record metrics. River core has no OpenTelemetry dependency.