← назад к разделу

Фоновые горутины — планировщики, воркеры, outbox-relay — это место, где graceful shutdown чаще всего ломается молча. Горутина продолжает работу после получения сигнала остановки, пул базы данных закрывается раньше, чем транзакция завершилась. В результате при деплое часть операций уходит без commit — состояние расходится.

Разберём, как правильно организовать остановку фоновых горутин: что проверять, в каком порядке закрывать ресурсы и почему outbox-relay требует особого внимания.

Обязательно

Почему горутины не останавливаются сами

Горутина в Go — это независимый поток выполнения. Когда приложение получает SIGTERM, горутины не узнают об этом автоматически. Без явной передачи сигнала остановки горутина продолжает цикл бесконечно.

Стандартный инструмент — context.Context. Когда главный код вызывает cancel(), все горутины, которые смотрят на ctx.Done(), получают сигнал: «пора заканчивать».

Второй инструмент — sync.WaitGroup. Он позволяет главному коду подождать, пока все горутины завершат текущую итерацию, прежде чем закрывать соединения с базой данных.

Структура фоновой горутины

Любая фоновая горутина получает два аргумента: ctx context.Context и wg *sync.WaitGroup. Вызывающая сторона регистрирует горутину в WaitGroup до запуска.

type OutboxRelay struct {
    pool     *pgxpool.Pool
    queries  *db.Queries
    producer *kafka.Writer
    interval time.Duration
}

func (r *OutboxRelay) Run(ctx context.Context, wg *sync.WaitGroup) {
    defer wg.Done()
    ticker := time.NewTicker(r.interval)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            if err := r.processOneBatch(ctx); err != nil {
                slog.WarnContext(ctx, "outbox relay batch failed", "error", err)
            }
        }
    }
}

Несколько важных деталей:

  • defer wg.Done() — первым, чтобы счётчик освобождался при любом выходе из функции.
  • select проверяет ctx.Done() перед каждым тиком, а не внутри processOneBatch. Это значит «не начинать новую итерацию», а не «прервать текущую».
  • Если SIGTERM пришёл в момент выполнения processOneBatch, горутина дожидается его завершения и только на следующем витке видит <-ctx.Done() и выходит.

Частая ошибка — написать for { processOneBatch(ctx) } без select и ctx.Done(). Такая горутина игнорирует сигнал остановки, и WaitGroup никогда не освободится.

Порядок shutdown в main

Порядок закрытия ресурсов принципиален. Если закрыть пул базы данных до того, как горутины завершили работу, pool.Close() сначала повиснет на соединении, которое горутина ещё держит в транзакции, а следующий Begin вернёт ошибку closed pool — пачка уйдёт в лог ошибкой и не отправится.

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    pool, err := pgxpool.New(ctx, cfg.DatabaseURL)
    if err != nil {
        slog.Error("pgxpool connect", "error", err)
        os.Exit(1)
    }

    appState := health.NewState()

    producer := &kafka.Writer{
        Addr:         kafka.TCP(cfg.KafkaBroker),
        Topic:        "order-events",
        Balancer:     &kafka.Hash{},
        RequiredAcks: kafka.RequireAll,
        BatchTimeout: 10 * time.Millisecond,
    }

    relay := &OutboxRelay{
        pool:     pool,
        queries:  db.New(pool),
        producer: producer,
        interval: 500 * time.Millisecond,
    }

    var wg sync.WaitGroup
    wg.Add(1)
    go relay.Run(ctx, &wg)

    srv := &http.Server{Addr: cfg.Addr, Handler: buildRouter(appState)}

    sigC := make(chan os.Signal, 1)
    signal.Notify(sigC, syscall.SIGTERM, syscall.SIGINT)
    defer signal.Stop(sigC)

    errC := make(chan error, 1)
    go func() { errC <- srv.ListenAndServe() }()

    select {
    case sig := <-sigC:
        slog.Info("получили SIGTERM, начинаем graceful shutdown", "signal", sig.String())
    case err := <-errC:
        slog.Error("server error", "error", err) // сервер не поднялся: гасим остальное тем же путём
    }

    appState.SetNotReady()    // readiness → 503 первым

    cancel()                  // сигнал горутинам: новых итераций не начинать

    shutCtx, shutCancel := context.WithTimeout(context.Background(), 25*time.Second)
    defer shutCancel()
    if err := srv.Shutdown(shutCtx); err != nil {
        slog.Error("http shutdown", "error", err)
    }

    wg.Wait()                 // дождаться завершения всех горутин

    if err := producer.Close(); err != nil {
        slog.Error("kafka writer close", "error", err)
    }
    pool.Close()              // пул — последним
    slog.Info("graceful shutdown завершён")
}

Правило простое: pool.Close() всегда строго после wg.Wait().

Две настройки writer здесь не случайны. Balancer: &kafka.Hash{} — без него ключ сообщения на выбор партиции не влияет: балансировщик по умолчанию раскладывает сообщения по кругу, события одного заказа разъезжаются по партициям, и порядок теряется (а если партиция должна совпадать с той, что выбрал бы Java-продюсер, берут kafka.Murmur2Balancer). BatchTimeout: 10ms — о нём ниже, в разборе пачки.

Outbox-relay: почему транзакцию нельзя прерывать

Outbox-relay — горутина, которая забирает события из таблицы outbox_event и публикует их в Kafka. Её особенность: незавершённая транзакция означает потенциальный дубль при публикации.

Поэтому внутри processOneBatch транзакция открывается на context.Background(), а не на родительском ctx. Даже если ctx уже отменён (пришёл SIGTERM), транзакция доводится до commit или rollback.

func (r *OutboxRelay) processOneBatch(ctx context.Context) error {
    tx, err := r.pool.Begin(context.Background()) // отдельный ctx для транзакции
    if err != nil {
        return fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback(context.Background())

    qtx := r.queries.WithTx(tx)

    events, err := qtx.LockOutboxBatch(context.Background(), 50)
    if err != nil {
        return fmt.Errorf("lock outbox batch: %w", err)
    }
    if len(events) == 0 {
        return nil
    }

    msgs := make([]kafka.Message, 0, len(events))
    for _, e := range events {
        msgs = append(msgs, kafka.Message{Key: []byte(e.AggregateID), Value: e.Payload})
    }
    if err := r.producer.WriteMessages(context.Background(), msgs...); err != nil {
        return fmt.Errorf("publish outbox batch: %w", err)
    }
    for _, e := range events {
        if err := qtx.MarkDispatched(context.Background(), e.ID); err != nil {
            return fmt.Errorf("mark dispatched %s: %w", e.ID, err)
        }
    }

    if err := tx.Commit(context.Background()); err != nil {
        return fmt.Errorf("commit outbox batch: %w", err)
    }
    return nil
}

Пачка уходит в Kafka одним вызовом WriteMessages, и это не украшение. Синхронный kafka.Writer держит вызов, пока не соберёт полную пачку (BatchSize, по умолчанию 100) или не истечёт BatchTimeout — по умолчанию секунда. Отправка по одному сообщению в цикле — это секунда на событие и почти минута на пачку из пятидесяти; одним вызовом вся пачка ждёт один таймаут, а с BatchTimeout: 10ms из main укладывается в десятки миллисекунд. У запроса LockOutboxBatch один параметр, поэтому sqlc генерирует обычный аргумент, а не структуру Params.

SQL-запрос для блокировки строк (sqlc):

-- name: LockOutboxBatch :many
SELECT id, aggregate_id, topic, payload
FROM outbox_event
WHERE dispatched_at IS NULL
ORDER BY created_at
LIMIT @limit
FOR UPDATE SKIP LOCKED;

FOR UPDATE SKIP LOCKED решает проблему параллельного запуска при rolling update. Если два экземпляра приложения запущены одновременно, они не берут одни и те же строки: каждый берёт свои и не ждёт других. Когда старый экземпляр завершает пачку и commit, строки помечены dispatched_at IS NOT NULL — новый их пропустит. Если старый упал без commit — строки разблокированы, новый подхватит.

Для обычных горутин без критичных транзакций (например, удаление устаревших записей) можно передавать ctx напрямую: прерывание допустимо, следующий запуск доделает.

Несколько горутин — один WaitGroup

Если сервис запускает несколько фоновых задач, все они регистрируются в одном WaitGroup:

var wg sync.WaitGroup

wg.Add(1)
go outboxRelay.Run(ctx, &wg)

wg.Add(1)
go productConsumer.Run(ctx, &wg)

wg.Add(1)
go expiredOrderCleaner.Run(ctx, &wg)

// при shutdown — один общий вызов
cancel()
wg.Wait()
pool.Close()

Все горутины завершаются параллельно, shutdown ждёт медленнейшей. Если OutboxRelay доводит пачку за 3 секунды, а ProductConsumer — за 8 секунд, суммарное ожидание 8 секунд, а не 11.

Задача по расписанию в нескольких подах

Пересчёт остатков в три часа ночи должен выполняться один раз, а подов три, и у каждого свой тикер. Координацию отдают базе: задача берёт advisory lock на время работы, и пока он занят, остальные поды пропускают запуск. Остановка пода освобождает блокировку вместе с соединением, поэтому «залипшего» замка не бывает:

func (j *RecalculateStock) RunOnce(ctx context.Context) error {
    conn, err := j.pool.Acquire(ctx)
    if err != nil {
        return err
    }
    defer conn.Release()

    var locked bool
    if err := conn.QueryRow(ctx, "SELECT pg_try_advisory_lock($1)", recalcStockLockID).Scan(&locked); err != nil {
        return err
    }
    if !locked {
        return nil // другой под уже считает
    }
    defer conn.Exec(context.WithoutCancel(ctx), "SELECT pg_advisory_unlock($1)", recalcStockLockID)

    return j.recalculate(ctx)
}

Блокировка висит на соединении, поэтому его держат явно через Acquire до конца работы, а не берут из пула на каждый запрос. Если задача длиннее бюджета остановки, её делают возобновляемой: пачками с фиксацией прогресса, чтобы следующий под продолжил с места обрыва.

Бюджет времени

У graceful shutdown есть общий лимит — обычно 60 секунд, из которых Kubernetes оставляет около 10 секунд на preStop. Фоновые горутины должны укладываться в ~20 секунд.

Если пачка из 50 событий занимает слишком долго — уменьшите размер пачки до 10–15 событий. Увеличивать бюджет не нужно: пачка должна быть маленькой по дизайну.

Примерная раскладка:

Шаг остановкиСколько ждём
preStop sleep10 с
http.Server.Shutdown≤ 25 с (параллельно с отменой горутин)
ожидание горутин по WaitGroup≤ 20 с
pool.Close~1 с
итого≤ 56 с — влезает в 60 с terminationGracePeriodSeconds

На практике HTTP-drain заканчивается раньше, чем outbox завершает пачку — они частично перекрываются.

Ожидание горутин ограничивают явно, иначе одна зависшая пачка съест весь бюджет пода:

done := make(chan struct{})
go func() {
    wg.Wait()
    close(done)
}()
select {
case <-done:
    slog.Info("фоновые горутины завершились")
case <-time.After(20 * time.Second):
    slog.Warn("фоновые горутины не уложились в бюджет, закрываем пул") // незавершённая пачка вернётся по аренде
}
pool.Close()
Дополнительно: при первом чтении можно пропустить

Глубже: лишняя итерация после отменырасширенное

Если к моменту select готовы оба канала — тик уже случился и ctx уже отменён, — Go выберет случайную ветку, и горутина может начать ещё одну пачку после сигнала остановки. Обычно это не страшно: пачка короткая, а транзакция всё равно на своём контексте. Если лишняя итерация недопустима, добавьте проверку ctx.Err() != nil первой строкой ветки тикера. Второй тонкий момент — time.NewTicker не ждёт окончания итерации: пачка, которая длилась дольше интервала, увидит следующий тик сразу же (лишние тики тикер отбрасывает, но один держит наготове), поэтому долгие пачки удобнее запускать через time.NewTimer с перезаводом после каждой.

Коротко

  • Горутины не останавливаются сами при SIGTERM — им нужен context.Context с cancel().
  • ctx.Done() проверяется в select перед каждым тиком: «не начинать новую итерацию», не прерывать текущую.
  • defer wg.Done() — первой строкой в горутине, wg.Add(1) — до go в вызывающем коде.
  • pool.Close() строго после wg.Wait() — иначе Close повиснет на занятом соединении, а следующая итерация получит closed pool.
  • Внутри критичной секции outbox-relay транзакция открывается на context.Background(), не на отменённом ctx.
  • FOR UPDATE SKIP LOCKED защищает от двойной обработки одних строк при rolling update.
  • Несколько горутин — один общий WaitGroup, завершаются параллельно.
  • Размер пачки держите маленьким: фоновые горутины должны укладываться в ~20 секунд.
  • Пачку в Kafka отправляйте одним WriteMessages и с коротким BatchTimeout: синхронный writer ждёт таймаут на каждый вызов, а ключ влияет на партицию только с Balancer: &kafka.Hash{}.

Что пощупать

Relay из этой статьи в практикуме remodov/marketplace-system-go работает в сервисе заказов: OutboxRelay.Run крутит тикер, каждая пачка идёт в своей транзакции под контекстом с таймаутом, который не отменяется вместе с родительским (context.WithoutCancel), а main ждёт горутину через WaitGroup после остановки HTTP-сервера. Так остановка сервиса не рвёт транзакцию посреди пачки: либо строки помечены и отправлены, либо откатились целиком.

Код: relay_outbox.go, main.go.

Что почитать дальше