Stop a consumer gracefully

When you stop an application for a deploy or restart, a consumer may still be processing a message. Give it time to finish that work and record the result before closing its database connection.

Pass a cancellable context to Consume, cancel it to request shutdown, and wait for Consume to return before closing the pool. The example below makes that order visible so you can check it in a running process.

Before you begin

Use the local Go project and PostgreSQL database from the Quickstart. Its producer writes WelcomeEmailV1 messages to the signup.welcome-email stream. This guide keeps that stream and the email-sender consumer group.

1. Connect shutdown to the consumer’s lifetime

Replace cmd/consume/main.go with this complete example. The consumer handler simulates three seconds of work and prints the message id. It does not send an email.

package main

import (
  "context"
  "errors"
  "fmt"
  "os"
  "time"

  sqlstreams "github.com/allegedlyreliable/sqlstreams/client"
)

type WelcomeEmailV1 struct {
  UserId string `json:"user_id"`
}

func (WelcomeEmailV1) SchemaVersion() int { return 1 }

func main() {
  if err := run(); err != nil {
    fmt.Fprintln(os.Stderr, err)
    os.Exit(1)
  }
}

func run() error {
  sessionCtx, stop := sqlstreams.LifecycleContext(nil)
  defer stop()

  pool, err := sqlstreams.NewPostgresPool(sessionCtx, "username", "password", "localhost", "quickstart", nil)
  if err != nil {
    return err
  }
  defer func() {
    pool.Close()
    fmt.Println("database pool closed")
  }()

  client, err := sqlstreams.NewClient(sessionCtx, pool, nil)
  if err != nil {
    return err
  }

  emails := client.Stream[WelcomeEmailV1]("signup.welcome-email")
  registered, err := emails.Get(sessionCtx)
  if err != nil {
    return err
  }
  if registered == nil {
    return errors.New("stream not registered -- run the producer first")
  }

  consumer, err := emails.Consumer("email-sender").Register(sessionCtx, nil)
  if err != nil {
    return err
  }

  err = consumer.Consume(sessionCtx, receiveEmail, nil)
  fmt.Println("Consume returned")
  return err
}

func receiveEmail(ctx context.Context, email *WelcomeEmailV1) error {
  metadata, _ := sqlstreams.MetaFromContext(ctx)
  fmt.Printf("started message %d for %s\n", metadata.Id, email.UserId)

  timer := time.NewTimer(3 * time.Second)
  defer timer.Stop()
  select {
  case <-ctx.Done():
    fmt.Printf("cancelled message %d\n", metadata.Id)
    return ctx.Err()
  case <-timer.C:
    fmt.Printf("finished message %d\n", metadata.Id)
    return nil
  }
}

LifecycleContext cancels sessionCtx on SIGINT or SIGTERM. Consume stays in the foreground while shutdown finishes. Only after it returns can run exit and execute the deferred pool closure.

When applying this to your own application, keep the pool open until all consumer sessions that use it have returned. If your application already handles shutdown signals, pass its cancellable shutdown context to Consume.

2. Run the consumer and stop it

Produce four messages by running this command four times:

go run ./cmd/produce

Start the consumer:

go run ./cmd/consume

As soon as you see started message, press Ctrl-C once. The consumer handler should finish its simulated work before Consume returns and the pool closes.

This is the captured output from a run with message 1. Your timestamps and message ids will differ. The plain lines come from the example’s fmt calls, and the level=INFO lines come from LifecycleContext.

started message 1 for user-123
^Ctime=2026-09-14T20:45:20.710-04:00 level=INFO msg="graceful shutdown started (send signal again to force exit)" signal=interrupt
finished message 1
Consume returned
database pool closed
time=2026-09-14T20:45:23.728-04:00 level=INFO msg="graceful shutdown complete"

3. Use the consumer handler’s context for its work

When replacing the simulated wait with database calls or HTTP requests, pass the consumer handler’s ctx to those operations. Keep sessionCtx for the consumer session. The two contexts have different jobs.

Cancelling sessionCtx starts shutdown. A consumer handler already running gets time to finish under its own timeout. If that timeout expires, its ctx is cancelled and the consumer handler should return promptly. The timer example selects on ctx.Done() so it can return without completing the wait.

Go cannot forcibly stop application code that ignores cancellation. Check that your database and HTTP calls use the consumer handler’s context before relying on graceful shutdown in your application.

Related topics

• Consumer groups — Why processing can repeat after a consumer handler succeeds.
• Consumer leases — When unfinished work becomes available for another instance to claim.
• Consumer stopped log — Inspect the session summary and understand what its counters measure.