Готовность и корректное завершение API

Livez, Readyz и точка допуска нового задания

GET /livez сообщает, способен ли процесс обслуживать запросы; GET /readyz сообщает, следует ли направлять ему новые задания. Во время начала штатного завершения процесс ещё жив, хотя новые задания принимать уже нельзя. Readiness помогает балансировщику убрать экземпляр из маршрута, но это действие может запаздывать. Клиент с сохранённым адресом также способен послать прямой POST /enqueue. Поэтому обработчик проверяет допуск сам, а не полагается только на ответ /readyz.

У допуска есть точка принятия решения: короткая критическая секция под одним sync.Mutex. Если запрос вошёл в неё до stopIntake, он уже принят и увеличил число активных обработок. Если после — получает отказ. Переключение accepting=false использует тот же mutex. Это определяет порядок даже тогда, когда goroutine исполняются одновременно; отдельный atomic.Bool.Load перед записью такого общего решения не даёт. Счётчик active полезен для наблюдения, но ожидание HTTP handlers ниже обеспечит Shutdown, а не автоматическое свойство этого счётчика.

type Server struct {
    mu        sync.Mutex
    accepting bool
    active    int
    store     JobStore
}

func (s *Server) beginJob() bool {
    s.mu.Lock()
    defer s.mu.Unlock()
    if !s.accepting { return false }
    s.active++
    return true
}

func (s *Server) endJob() {
    s.mu.Lock()
    s.active--
    s.mu.Unlock()
}

func (s *Server) stopIntake() {
    s.mu.Lock()
    s.accepting = false
    s.mu.Unlock()
}

func (s *Server) readyzHandler(w http.ResponseWriter, r *http.Request) {
    s.mu.Lock()
    ready := s.accepting
    s.mu.Unlock()
    if !ready { w.WriteHeader(http.StatusServiceUnavailable); return }
    w.WriteHeader(http.StatusOK)
}

enqueueHandler первым делом вызывает beginJob; при false отвечает 503, при true выполняет обработку и через defer s.endJob() снимает активный счётчик. Если stopIntake случится после успешного beginJob, завершение записи после переключения — не дефект: это именно ранее принятая работа, ради которой выполняется graceful drain. Новое задание после точки переключения не попадёт в SaveJob. Обе ветки нужны: внешний readiness-сигнал управляет маршрутизацией, внутренняя проверка обеспечивает правило самого API.

func TestDirectPostIsRejectedAfterStopIntake(t *testing.T) {
    store := &fakeStore{}
    srv := &Server{accepting: true, store: store}
    srv.stopIntake()

    rec := httptest.NewRecorder()
    req := httptest.NewRequest(http.MethodPost, "/enqueue",
        strings.NewReader(`{"id":"job-1"}`))
    srv.enqueueHandler(rec, req)

    if rec.Code != http.StatusServiceUnavailable {
        t.Fatalf("want 503, got %d", rec.Code)
    }
    if store.saved { t.Fatal("rejected POST reached SaveJob") }
}

Тест проверяет прямой вызов handler после stop-intake и отсутствие записи, а не работу балансировщика. Он не доказывает исход запроса, который уже вошёл до переключения. Такую последовательность нужно контролировать отдельными каналами: дождаться входа в SaveJob, переключить допуск, затем освободить сохранение. Race detector может найти небезопасные обращения к памяти, но не доказывает правильность логического порядка, если все обращения уже синхронизированы. /livez не привязан к accepting; пока его endpoint ещё достижим и процесс здоров, он может отвечать 200. После закрытия HTTP listener запрос к нему уже не обязан достигнуть handler.

Принятый запрос: допуск, подтверждённая запись, затем 202

RFC 9110 задаёт 202 Accepted как намеренно неокончательный ответ: запрос принят для обработки, но обработка ещё не завершилась. Сам статус не обещает ни устойчивой записи, ни будущего успешного выполнения. В этом Media API установлен более сильный локальный контракт: если клиент получил 202, SaveJob уже получил подтверждение от выбранного устойчивого хранилища. Если подтверждения нет, обработчик не имеет права отправить 202. Конкретная долговечность подтверждения зависит от конфигурации хранилища и его модели отказов, а не от HTTP-статуса.

У согласованной точки допуска есть практическое следствие. beginJob и stopIntake упорядочены одним mutex; запрос, допущенный раньше переключения, вправе закончить SaveJob и ответить 202 позднее. Успешный допуск не означает, что клиент обязательно увидит ответ: клиент может закрыть соединение. Это отдельная неопределённость для клиента, не причина переименовывать уже принятый запрос в «новый после stop-intake». При ошибке декодирования вернётся 400; при ошибке записи — не 202.

func (s *Server) enqueueHandler(w http.ResponseWriter, r *http.Request) {
    if !s.beginJob() {
        http.Error(w, "service is shutting down", http.StatusServiceUnavailable)
        return
    }
    defer s.endJob()

    var job JobRequest
    if err := json.NewDecoder(r.Body).Decode(&job); err != nil {
        http.Error(w, "bad request", http.StatusBadRequest)
        return
    }
    if err := s.store.SaveJob(r.Context(), job); err != nil {
        http.Error(w, "failed to persist job", http.StatusInternalServerError)
        return
    }
    w.WriteHeader(http.StatusAccepted)
}

r.Context() может быть отменён клиентом или при ином завершении запроса; сам нормальный вызов http.Server.Shutdown не отменяет контекст уже активного handler. Если хранилище соблюдает этот контекст, отмена клиента может прервать ещё не подтверждённую запись, и handler не должен утверждать 202. Если приложению нужна политика «закончить запись, несмотря на уход клиента», это отдельное решение с собственным ограниченным контекстом и явной ответственностью за продолжающуюся работу. Нельзя объяснять такой выбор вымышленным действием Shutdown.

func Test202FollowsSuccessfulSave(t *testing.T) {
    store := &fakeStore{}
    srv := &Server{accepting: true, store: store}
    req := httptest.NewRequest(http.MethodPost, "/enqueue",
        strings.NewReader(`{"id":"job-42","type":"transcode"}`))
    rec := httptest.NewRecorder()
    srv.enqueueHandler(rec, req)
    if rec.Code != http.StatusAccepted || !store.saved {
        t.Fatalf("want save then 202; code=%d saved=%t", rec.Code, store.saved)
    }

    store.failSave = true
    rec = httptest.NewRecorder()
    srv.enqueueHandler(rec, httptest.NewRequest(http.MethodPost, "/enqueue",
        strings.NewReader(`{"id":"job-43"}`)))
    if rec.Code == http.StatusAccepted { t.Fatal("202 on save failure") }
}

Тестовый fakeStore.saved подтверждает лишь порядок вызовов и ветвление при ошибке. Даже если реальный SaveJob получил ответ sql.Tx.Commit или XADD, из одного факта ответа нельзя вывести crash durability при любой конфигурации: важны настройки синхронной записи, репликации и подтверждения. Здесь предполагается, что адаптер JobStore возвращает nil только после подтверждения, достаточного для сформулированного контракта Media API. Его физическую устойчивость следует проверять отдельно на выбранном хранилище и модели сбоя. Так различаются проверяемое локальное правило «save до 202» и гарантия внешней системы.

Shutdown осушает HTTP, а не убивает активный handler

После stopIntake приложение вызывает http.Server.Shutdown с ограниченным временем ожидания. Shutdown закрывает listeners, закрывает idle connections и ждёт, пока активные HTTP connections станут idle. Это HTTP drain, а не сигнал завершить произвольный фоновый worker. Уже допущенный handler может продолжать SaveJob и возвращать 202 во время ожидания. Его r.Context() не отменяется просто потому, что начался штатный Shutdown.

Важно отделить сигнал остановки от контекста drain. Если внешний контекст уже отменён сигналом процесса, дочерний context.WithTimeout от него тоже отменится немедленно. Поэтому короткий drain-контекст создаётся отдельно; после истечения его срока Shutdown возвращает ошибку контекста. Это означает, что ожидание не завершилось в срок. Оно не означает, что Shutdown насильно прервал активные handlers или закрыл их соединения. Активная работа может ещё идти, даже когда вызов уже вернул ошибку.

func drainHTTP(srv *http.Server, app *Server) error {
    app.stopIntake()
    drainCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
    defer cancel()
    if err := srv.Shutdown(drainCtx); err != nil {
        return fmt.Errorf("HTTP drain incomplete: %w", err)
    }
    return nil
}

Вызвавший код должен решить, что делать с неполным drain: продлить ожидание в рамках бюджета остановки, зафиксировать ошибку и оставить процессу время, либо явно вызвать Close/завершить процесс. Последние действия — отдельная политика приложения; они могут оборвать соединение и не дать клиенту увидеть 202, даже если запись была подтверждена. Нельзя приписывать этот обрыв самому Shutdown. Если оркестратор завершит процесс по внешнему дедлайну, его эффект также отличается от результата метода Go.

Следующий тест не полагается на Sleep(10ms). Канал entered закрывается только после входа в SaveJob; до него stopIntake и Shutdown не запускаются. release удерживает запись, пока не поступит сигнал начала shutdown. Таким образом, проверяется именно ранее допущенный активный handler, а не случайный POST, которому успели отказать до входа. В тестовом fixture SaveJob возвращает nil после освобождения; это не проверка физической долговечности.

type blockingStore struct { entered, release chan struct{} }
func (s *blockingStore) SaveJob(ctx context.Context, job JobRequest) error {
    close(s.entered)
    <-s.release
    return nil
}

func TestAdmittedHandlerFinishesDuringShutdown(t *testing.T) {
    store := &blockingStore{make(chan struct{}), make(chan struct{})}
    app := &Server{accepting: true, store: store}
    mux := http.NewServeMux()
    mux.HandleFunc("/enqueue", app.enqueueHandler)
    server := httptest.NewServer(mux)
    releaseOnce := sync.Once{}
    release := func() { releaseOnce.Do(func() { close(store.release) }) }
    defer server.Close()
    defer release()

    result := make(chan int, 1)
    go func() {
        resp, err := http.Post(server.URL+"/enqueue", "application/json",
            strings.NewReader(`{"id":"job-99"}`))
        if err != nil { result <- 0; return }
        defer resp.Body.Close()
        result <- resp.StatusCode
    }()
    select { case <-store.entered: case <-time.After(5*time.Second): t.Fatal("handler did not enter SaveJob") }
    app.stopIntake()
    started := make(chan struct{})
    server.Config.RegisterOnShutdown(func() { close(started) })
    drained := make(chan error, 1)
    go func() {
        ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
        defer cancel()
        drained <- server.Config.Shutdown(ctx)
    }()
    select { case <-started: case <-time.After(5*time.Second): t.Fatal("shutdown did not start") }
    release()
    select {
    case got := <-result:
        if got != http.StatusAccepted { t.Fatalf("want 202, got %d", got) }
    case <-time.After(6 * time.Second):
        t.Fatal("client result did not arrive")
    }
    select {
    case err := <-drained:
        if err != nil { t.Fatalf("drain: %v", err) }
    case <-time.After(6 * time.Second):
        t.Fatal("Shutdown result did not arrive")
    }
}

Каналы задают порядок: сначала вход в SaveJob, затем запуск Shutdown, после него освобождение записи. Ограниченные ожидания entered, started, result и drained дают явную ошибку вместо бесконечного ожидания этих сигналов; прочие сбои всё ещё ограничены внешним таймаутом тестового запуска. Этот тест не моделирует истечение самого drain deadline и не обещает, что всякая запись успеет за пять секунд. Отказ нового POST после stopIntake проверяется отдельно в разделе 1. Даже после успешного HTTP drain произвольный worker может оставаться активным: для него нужен свой механизм остановки/ожидания, а не вывод из результата Shutdown.

Два таймлайна: HTTP drain и независимый worker

Полученный клиентом 202 по контракту Media API означает подтверждённую запись задания, но не его выполнение. Shutdown управляет HTTP-сервером, а не жизненным циклом отдельного worker. Для корректного вывода нужно знать, где worker запущен, продолжает ли он работу и какие обещания даёт очередь. Слова «worker продолжит» и «задача восстановится» без этих условий скрывают важные отказные ветки.

Таймлайн A — worker продолжает независимо. API допустил POST, хранилище подтвердило устойчивую запись в пределах своей настроенной модели отказа, клиент получил 202. Затем API прекратил приём новых заданий и успешно выполнил HTTP drain. Отдельный процесс worker не останавливался вместе с API: он берёт ранее записанное задание, обрабатывает и подтверждает его очереди. Успех HTTP drain здесь не доказал завершения задания; завершение произошло благодаря работающему worker и его отдельному подтверждению. Если worker медленнее, обработка может продолжиться и после остановки API.

Таймлайн B — worker прерван. После подтверждённой записи и 202 worker получил задание, но завершился до надёжного Ack. Допустим, очередь действительно сохраняет запись при сбое, использует visibility timeout или иной документированный механизм повторной доставки и позднее доступен другой worker. Тогда после истечения видимости задание может быть выдано снова. Если первый worker успел выполнить внешний эффект до падения, повтор создаёт риск дубля: обработка должна быть идемпотентной или защищённой отдельным ключом. Без устойчивой очереди, политики redelivery и доступного worker обещать дальнейшую обработку нельзя. Ни исход A, ни исход B не предопределён одним вызовом Shutdown.

Статичная схема (PNG)

// Фрагмент работы с подтверждением, при условии очереди с redelivery.
func (w *Worker) handleOne(ctx context.Context, job Job) error {
    if err := w.process(ctx, job); err != nil {
        return fmt.Errorf("process failed; job remains unacked: %w", err)
    }
    if err := w.queue.Ack(ctx, job.ID); err != nil {
        return fmt.Errorf("process completed but Ack failed: %w", err)
    }
    return nil
}

Этот фрагмент не обещает автоматического повтора: он лишь показывает точку Ack и возвращает его ошибку вызывающему коду. При ошибке process запись остаётся без Ack, но дальнейшая судьба определяется конкретной очередью. При ошибке Ack эффект process мог уже произойти, поэтому без идемпотентности повтор опасен. Нельзя игнорировать ошибку Ack или считать, что любой возвращённый error означает отсутствие предметного эффекта.

func TestEnqueueThenDequeueInSameMemoryQueue(t *testing.T) {
    q := newInMemoryQueue()
    job := Job{ID: "job-77", Type: "transcode"}
    if err := q.Enqueue(context.Background(), job); err != nil {
        t.Fatalf("enqueue: %v", err)
    }
    got, err := q.Dequeue(context.Background())
    if err != nil { t.Fatalf("dequeue: %v", err) }
    if got.ID != job.ID { t.Fatalf("want %s, got %s", job.ID, got.ID) }
}

Тест доказывает лишь, что один и тот же объект очереди в памяти возвращает ранее помещённый элемент. Он не запускает worker, не останавливает процесс, не создаёт новую очередь и не проверяет восстановление после рестарта. Для такого утверждения нужен отдельный интеграционный тест с выбранным устойчивым хранилищем и контролируемым перезапуском. Конкретная схема статуса задания или уведомления клиенту — отдельный механизм: тот же HTTP 202 не превращается в ответ о результате обработки. На этих двух таймлайнах видно точное разделение обязанностей: API отвечает за допуск и подтверждение записи, worker — за попытку обработки и Ack, очередь — за те гарантии хранения и повторной доставки, которые реально настроены.