Subscriptions

Especially in mature systems, it’s useful to receive detailed information on exactly what’s happening inside a job queue to enable uses like emitting custom telemetry like logging and metrics. River clients support this through Client.Subscribe, which returns a channel emitting events about jobs moving through the client’s workers.

Along with a reference to worked jobs, events contain rich statistics like how long the job took to run, and how long it had to wait in the queue. See Event and JobStatistics.

func (c *Client[TTx]) Subscribe(kinds ...EventKind) (<-chan *Event, func()) {
    ...
}

// Event wraps an event that occurred within a River client, like a job being
// completed.
type Event struct {
    // Kind is the kind of event. Receivers should read this field and respond
    // accordingly. Subscriptions will only receive event kinds that they
    // requested when creating a subscription with Subscribe.
    Kind EventKind

    // Job contains job-related information.
    Job *rivertype.JobRow

    // JobStats are statistics about the run of a job.
    JobStats *JobStatistics
}

// JobStatistics contains information about a single execution of a job.
type JobStatistics struct {
    CompleteDuration  time.Duration // Time it took to set the job completed, discarded, or errored.
    QueueWaitDuration time.Duration // Time the job spent waiting in available state before starting execution.
    RunDuration       time.Duration // Time job spent running (measured around job worker.)
}

Subscribe takes variadic args containing the kinds of events a subscriber would like to receive:

subscribeChan, subscribeCancel :=
    riverClient.Subscribe(river.EventKindJobCompleted)
defer subscribeCancel()

See the Subscription example for complete code.

Subscribe returns a subscription channel and a cancel function, the latter making it possible to close the subscription. Some notable properties:

• Subscription channels are buffered channels of size 100 to prevent slow consumers from stalling clients as they distribute events. Receivers doing heavy lifting as they receive events (e.g. network sends) should have their own buffering system so they don’t fall behind the subscription and accidentally drop events.

• Subscription channels are closed when clients stop. Channels receive *Event, so subscribers can detect this condition by checking for a nil event received.

• Cancel functions should generally be invoked using a defer, but it’s not strictly necessary to do so unless subscribers want to close their subscription prematurely. Subscriptions are terminated automatically on shutdown.

• Clients support an arbitrary number of subscriptions, so it’s okay to Subscribe multiple times for different uses.

• Events are only distributed for jobs worked by the specific client that was subscribed to. When running multiple clients across multiple nodes, it’s necessary to subscribe to all of them to receive events for all jobs in the entirety of the cluster.

Create a subscription for completed jobs and iterate over its events. Close the subscription in ensure when the listener exits:

subscription = client.subscribe(:job_completed, buffer_size: 1_000)

begin
  subscription.each { |event| puts event.job.id }
ensure
  subscription.close
end

Subscriptions are bounded. New events are dropped when the buffer is full, so consume them promptly. Closing a subscription wakes readers and lets them drain buffered events.

Receive events asynchronously from the subscription, handling lag notifications and stopping when it closes:

use riverqueue::{EventKind, EventRecvError};

let mut events = client.subscribe(&[EventKind::JobCompleted])?;

loop {
    match events.recv().await {
        Ok(event) => {
            if let Some(event) = event.as_job() {
                println!("Completed job {}", event.job.id);
            }
        }
        Err(EventRecvError::Lagged(count)) => eprintln!("Missed {count} events"),
        Err(EventRecvError::Closed) => break,
        Err(error) => return Err(error.into()),
    }
}

Events are local to the client and emitted after persistence. Dropping the receiver unsubscribes.

Iterate over the subscription with for await to receive completion events. The using declaration closes it when the scope exits:

using events = client.subscribe({ kinds: ["job_completed"], capacity: 1_000 });

for await (const event of events) {
  if (event.kind === "subscription_lag") {
    console.warn(`Dropped ${event.dropped} events`);
  } else {
    console.log("Completed job", event.job.id);
  }
}

A full buffer drops the oldest events and reports subscription_lag. using, close(), or exiting the loop closes the subscription.

Subscribe takes variadic args so that subscribers can receive multiple kinds of events:

subscribeChan, subscribeCancel := riverClient.Subscribe(
    river.EventKindJobCompleted,
    river.EventKindJobFailed,
)
defer subscribeCancel()

Pass multiple event kinds to subscribe to receive them through the same subscription:

subscription = client.subscribe(:job_completed, :job_failed, buffer_size: 1_000)

Include each event kind in the slice passed to subscribe:

let events = client.subscribe(&[
    riverqueue::EventKind::JobCompleted,
    riverqueue::EventKind::JobFailed,
])?;

List all the event kinds you want in the subscription’s kinds array:

using events = client.subscribe({ kinds: ["job_completed", "job_failed"] });

See EventKind for Go’s event constants.

Use symbols such as :job_completed, :job_failed, :job_cancelled, and :job_snoozed.

See EventKind for Rust’s event variants.

Use strings such as "job_completed", "job_failed", "job_cancelled", and "job_snoozed". The event’s kind narrows its TypeScript type.