Модуль 4. Outbox — goroutine + pgx FOR UPDATE SKIP LOCKED

Цель модуля: Гарантировать, что после git push вебхук дойдёт до внешней CI/CD-системы при любых сбоях сети. Применить Transactional Outbox (arch m3) к Git-специфичному событию RepositoryPushed на чистом Go: goroutine-воркер + pgx FOR UPDATE SKIP LOCKED + HMAC-подпись (arch m7) + retry/DLQ, без Symfony Messenger и без Redis Streams.

Перед началом: Arch m2 (Идемпотентность), m3 (Outbox), m7 (HMAC) · Git-hosting модули 1–3 Инструменты: Go 1.22+, pgx v5, net/http, hmac/sha256, Toxiproxy, k6 Трудоёмкость: ~5 ч (теория 1.5 ч · практика 3 ч · ADR 0.5 ч)

🔥 Инцидент: push прошёл, вебхук потерян

Пользователь делает git push в репозиторий. CI/CD ждёт вебхук RepositoryPushed, чтобы запустить сборку. Push принимается, packfile записан, refs обновлены — но вебхук не пришёл. Внешняя CI-система была недоступна ровно 2 секунды, Go-обработчик получил net/http: timeout awaiting response headers и просто залогировал ошибку — а откатывать уже принятый push бессмысленно.

В коде привычный Dual-Write:

go
// ❌ Антипаттерн: вебхук шлётся после коммита БД
func (h *PushHandler) Handle(w http.ResponseWriter, r *http.Request) {
    tx, _ := h.db.Begin(r.Context())
    h.repo.UpdateRefs(tx, push)        // (1) COMMIT refs в БД
    tx.Commit(r.Context())

    // (2) Вебхук — ПОСЛЕ коммита, если упадёт → потерян
    resp, err := h.http.Post(push.WebhookURL, "application/json", payload)
    if err != nil {
        log.Printf("webhook failed: %v", err) // всё: push остался, вебхук ушёл в void
    }
}

CI/CD упал после коммита — вебхук потерян, push остался. Сборка не запустилась, команда часами ждёт зелёного пайплайна, который не существует.

Знакомая история из arch m3: две независимые операции, одна прошла — вторая нет. Решение то же — Transactional Outbox, но на Go без брокера вообще: goroutine + pgx + net/http. На Symfony это Symfony Messenger + Redis Streams (или Kafka) — отдельная инфраструктура. На Go воркер — это 30 строк кода в internal/outbox/worker.go.

📖 Теория: Outbox на чистом Go, от Dual-Write к идемпотентной доставке

Глоссарий модуля:

Transactional Outbox — событие сохраняется в той же транзакции БД, что и бизнес-данные; отдельный воркер гарантирует доставку.
RepositoryPushed — доменное событие push: aggregate_id = repo_id, payload = {ref, before, after, commits[]}.
HMAC-подпись вебхука — HMAC-SHA256(secret, payload) в заголовке X-Signature, как в arch m7; получатель проверяет подлинность.
Идемпотентность вебхука — X-Webhook-Id (уникальный ключ на стороне получателя) защищает от дублей при at-least-once доставке (arch m2).
FOR UPDATE SKIP LOCKED — Postgres-расширение: блокирует строку от конкурентных воркеров, а занятые пропускает — основа параллельного polling-worker'а.
DLQ (Dead-Letter Queue) — очередь для событий, не доставленных после всех попыток; разбираются вручную.

4.1. Transactional Outbox для push (связь с arch m3)

Push-хендлер выполняет две записи в одной транзакции pgx:

go
// internal/repository/git_storage.go
func (s *GitStorage) AcceptPush(ctx context.Context, push *Push) error {
    tx, err := s.db.Begin(ctx)
    if err != nil {
        return err
    }
    defer tx.Rollback(ctx) // safe: no-op если Commit

    // (1) метаданные репозитория: updated_at, last_commit_sha
    if _, err := tx.Exec(ctx,
        `UPDATE repositories SET updated_at = NOW(), last_commit_sha = $1 WHERE id = $2`,
        push.After, push.RepoID,
    ); err != nil {
        return err
    }

    // (2) событие Outbox — в ТОЙ ЖЕ транзакции
    payload, _ := json.Marshal(map[string]any{
        "ref":     push.Ref,
        "before":  push.Before,
        "after":   push.After,
        "commits": push.Commits,
    })
    if _, err := tx.Exec(ctx,
        `INSERT INTO outbox_events (aggregate_id, event_type, payload, webhook_url, status)
         VALUES ($1, 'RepositoryPushed', $2, $3, 'pending')`,
        push.RepoID, payload, push.WebhookURL,
    ); err != nil {
        return err
    }

    return tx.Commit(ctx) // ОДИН COMMIT = атомарно обе строки
}

Публикацией вебхука занимается отдельный воркер — Dual-Write устранён, refs и событие коммитятся атомарно.

В Symfony это был $em->wrapInTransaction(function () use ...) { persist($repo); persist(OutboxEvent); }). В Go мы пишем tx.Begin / tx.Exec / tx.Commit руками — короче, но без абстракций ORM. Цена — больше SQL-кода, плюс — полная прозрачность.

4.2. Воркер доставки: goroutine + `FOR UPDATE SKIP LOCKED`

Воркер — это обычная goroutine с бесконечным циклом, запускаемая в main.go через go outboxWorker.Run(ctx). Каждую секунду: поллит таблицу, захватывает строки, шлёт вебхук, обновляет статус.

go
// internal/outbox/worker.go
type Worker struct {
    db     *pgxpool.Pool
    secret []byte
    client *http.Client
}

func (w *Worker) Run(ctx context.Context) {
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            return // graceful shutdown
        case <-ticker.C:
            if err := w.poll(ctx); err != nil {
                log.Printf("outbox poll error: %v", err)
            }
        }
    }
}

func (w *Worker) poll(ctx context.Context) error {
    tx, err := w.db.Begin(ctx)
    if err != nil {
        return err
    }
    defer tx.Rollback(ctx)

    // FOR UPDATE SKIP LOCKED — конкурентные воркеры пропускают занятые строки
    rows, err := tx.Query(ctx,
        `SELECT id, payload, webhook_url, attempts
         FROM outbox_events
         WHERE status = 'pending' AND attempts < 5
         ORDER BY created_at
         LIMIT 10
         FOR UPDATE SKIP LOCKED`)
    if err != nil {
        return err
    }
    defer rows.Close()

    for rows.Next() {
        var ev OutboxEvent
        if err := rows.Scan(&ev.ID, &ev.Payload, &ev.WebhookURL, &ev.Attempts); err != nil {
            return err
        }
        // ВАЖНО: публикация ВНУТРИ транзакции или после — см. 4.3
        if err := w.publish(ctx, &ev); err != nil {
            tx.Exec(ctx, `UPDATE outbox_events SET attempts = attempts + 1, last_error = $2 WHERE id = $1`,
                ev.ID, err.Error())
            continue
        }
        tx.Exec(ctx, `UPDATE outbox_events SET status = 'published', published_at = NOW() WHERE id = $1`, ev.ID)
    }
    return tx.Commit(ctx)
}

LIMIT 10 + FOR UPDATE SKIP LOCKED — пачка по 10 событий за одну транзакцию. Если запустить 3 воркера (go w.Run(ctx) × 3), каждый возьмёт свой набор — SKIP LOCKED обеспечивает это автоматически без согласованного блокирования.

4.3. Retry + DLQ

При ошибке сети: attempts = attempts + 1, last_error = err.Error(). При attempts >= 5 → статус failed, событие в DLQ (отдельная таблица или Kafka-топик), алерт в PagerDuty. Cron (или ещё одна goroutine) удаляет published старше 30 дней по индексу created_at.

go
if ev.Attempts >= 5 {
    tx.Exec(ctx, `UPDATE outbox_events SET status = 'failed', last_error = $2 WHERE id = $1`,
        ev.ID, "max attempts reached — DLQ")
    // + shove в DLQ: kafka.Produce("dlq-events", ev) или INSERT INTO outbox_dlq
}

На Symfony это делал Symfony Messenger из коробки (retry policy, failure transport). На Go — 10 строк if-логики. Проще в коде, но responsibility за DLQ полностью в ваших руках: алерт, повтор (replay) DLQ — всё это пишем сами.

4.4. HMAC-подпись вебхука (связь с arch m7)

Каждая доставка несёт заголовок X-Signature: HMAC-SHA256(secret, payload). Получатель проверяет подлинность, как в arch m7.

go
// internal/outbox/publisher.go
func (w *Worker) publish(ctx context.Context, ev *OutboxEvent) error {
    mac := hmac.New(sha256.New, w.secret)
    mac.Write(ev.Payload)
    signature := hex.EncodeToString(mac.Sum(nil))

    req, _ := http.NewRequestWithContext(ctx, "POST", ev.WebhookURL, bytes.NewReader(ev.Payload))
    req.Header.Set("Content-Type", "application/json")
    req.Header.Set("X-Signature", signature)
    req.Header.Set("X-Webhook-Id", ev.ID)  // идемпотентность (4.5)

    resp, err := w.client.Do(req)
    if err != nil {
        return err // network error — ретрай!
    }
    defer resp.Body.Close()

    if resp.StatusCode >= 200 && resp.StatusCode < 300 {
        return nil // success → mark published
    }
    return fmt.Errorf("webhook returned %d", resp.StatusCode)
}

На Symfony это был hash_hmac('sha256', $payload, $secret). На Go — hmac.New(sha256.New, secret) с Write + Sum. Больше строк, но без зависимостей — стандартная библиотека.

4.5. Идемпотентность вебхука (связь с arch m2)

Каждая доставка несёт X-Webhook-Id = outbox.id. Получатель хранит обработанные ID в UNIQUE-колонке: повторная доставка того же ID → 409/no-op. Это ровно паттерн Idempotency-Key из arch m2, перенесённый на HTTP-сторону получателя.

Упал воркер между шагом publish и UPDATE status='published' — строка останется pending, другой воркер возьмёт её и отправит повторно. Это at-least-once: получатель обязан быть идемпотентным через X-Webhook-Id.

Если CI-приёмник не идемпотентен (запускает сборку по каждому приходу) — ретраи будут давать дубли сборок. Это ответственность не отправителя, а приёмника — но архитектурно вы должны это требовать в контракте.

4.6. Воркер — это goroutine, не cron job

На Symfony мы сравнивали Messenger (long-running command) с cron. На Go вопрос острее: можно time.Sleep(1 * time.Second) в goroutine (воркер, latency ~1s), можно cron.New() раз в минуту (latency до 60s), можно LISTEN/NOTIFY (минимальная latency, но держит коннект и нет retain). Наш выбор — goroutine + ticker 1s: long-lived процесс внутри Go-бинарника, живёт рядом с http.ListenAndServe в main.go, выключается через ctx.Done() (graceful shutdown). В Symfony long-running command надо было запускать через systemd/supervisor и следить за restart — в Go это бесплатно.

4.7. pgx v5 vs database/sql

В курсе мы выбираем pgx v5: нативный FOR UPDATE SKIP LOCKED, pgxpool.Pool с настраиваемым connection pool, context.Context во всех методах (cancel'ится при grace shutdown), батчинг (COPY), prepared statements (важно в Модуле 7). database/sql + lib/pq — интерфейс-стандарт, но медленнее и без pgx-специфики.

4.8. Компромиссы (Trade-offs)

Подход	Плюс	Минус	Когда выбирать
Outbox + goroutine polling (наш)	без брокера, транзакции PG, at-least-once	таблица растёт → TTL; polling latency ~1s	всегда для webhook delivery
Прямой HTTP после коммита	простота	Dual-Write: вебхук теряется при сбое CI	антипаттерн (никогда)
LISTEN/NOTIFY вместо polling	realtime delivery, без polling overhead	держит коннект, нет retain при рестарте	small-scale, тестируемо

Вывод: Polling Outbox с goroutine + pgx — вариант по умолчанию для Go-воркера. Не нужен Messenger, не нужен Redis Streams — чистый Go + pgx + net/http + hmac. Это и есть упрощение Symfony-курса: одно приложение (бинарник) вместо app + Messenger worker + Redis/Kafka.

🛠 Практические задания

Сценарий: После git push внешний CI ждёт вебхук. Цель: 0 потерянных вебхуков при 50% недоступности CI-эндпоинта.

План:

Таблица outbox_events (arch m3 схема) + индексы (status, created_at), TTL 30 дней.
GitStorage.AcceptPush() внутри одной транзакции pgx сохраняет refs метаданные + INSERT outbox_events.
outbox.Worker: goroutine с ticker.NewTicker(1s) + FOR UPDATE SKIP LOCKED + HMAC-подпись + POST + mark published.
Retry (increment attempts) + DLQ при attempts >= 5 + алерт.
Идемпотентность: X-Webhook-Id + stub-приёмник с UNIQUE-колонкой.
Тест Toxiproxy: CI-эндпоинт недоступен 50% времени → 0 потерянных, outbox_pending_count → 0.
ADR-GG4 «Outbox на чистом Go: нет Messenger, нет Redis Streams».

Задача 1: `GitStorage.AcceptPush` в одной транзакции pgx

Push-хендлер сохраняет метаданные репозитория и RepositoryPushed одним tx.Commit(). aggregate_id = repo_id — будущий partition key.

✅ Готово, когда:

UPDATE repositories + INSERT outbox_events в одной транзакции pgx
payload содержит {ref, before, after, commits[]}
aggregate_id = repo_id помечен как partition key (в схеме БД)

Задача 2: outbox.Worker (goroutine + FOR UPDATE SKIP LOCKED + HMAC)

Запустить в main.go:

go
outboxWorker := outbox.NewWorker(db, secret)
go outboxWorker.Run(ctx) // запускаем goroutine до http.ListenAndServe

Воркер в одной транзакции: захват строки → POST с X-Signature: HMAC-SHA256 → UPDATE status='published'.

✅ Готово, когда:

Захват через SELECT ... FOR UPDATE SKIP LOCKED
X-Signature = HMAC-SHA256 от payload (arch m7)
Успех → status='published', published_at=NOW(); ошибка → attempts+1, last_error
attempts >= 5 → status='failed' + DLQ + алерт
Graceful shutdown: ctx.Done() останавливает воркер, tx.Rollback сбрасывает незавершённую транзакцию

Задача 3: Идемпотентный приёмник

Stub CI-эндпоинт хранит webhook_id в UNIQUE-колонке. Повтор → 409, no-op.

✅ Готово, когда:

Приёмник проверяет X-Webhook-Id по UNIQUE-индексу
Дубль не вызывает повторного запуска сборки (arch m2)

Задача 4: Тест Toxiproxy + k6

Toxiproxy отбрасывает 50% запросов к CI. k6 шлёт 100 push-ов. SLO: потерянных вебхуков = 0, outbox_pending_count → 0 за 5 минут, приёмник корректно гасит дубли через X-Webhook-Id.

✅ Готово, когда:

Toxiproxy: 50% timeout между воркером и CI-эндпоинтом
k6: 100 push-ов параллельно
Потерянных событий = 0; дубли гасятся приёмником

🧩 Проверь себя

1. Зачем заголовок X-Signature с HMAC-SHA256 на Go?

2. Что делает goroutine воркера при graceful shutdown?

3. Какое ключевое отличие Outbox на Go от Symfony-Messenger?

4. Почему отправка вебхука после коммита метаданных push — это Dual-Write?

5. Воркер Outbox на Go захватывает строки через…

🔗 Мостик к Модулю 5 (Cache-Aside)

После push мы надёжно рассылаем вебхуки через goroutine-воркер. Но каждый просмотр файлов в UI бьёт по диску: git ls-tree делает random-read, и 1000 одновременных просмотров популярного репозитория упираются в IOPS, p99 = 4 секунды. На Symfony это решалось кэшем Redis (Cache-Aside + SWR). В Модуле 5 мы закэшируем деревья файлов в Redis через go-redis + точно такую же Cache-Aside логику, и свяжем инвалидацию с тем же событием RepositoryPushed из Outbox — но с одной особенностью: на Go и кэш-клиент, и инвалидация живут в одном бинарнике с API, без отдельного consumer process.

← Модуль 3: Bulkhead — chan struct{} вместо пулов FPM Модуль 5: Cache-Aside — go-redis для деревьев файлов →

Модуль 4. Outbox — goroutine + pgx FOR UPDATE SKIP LOCKED

🔥 Инцидент: push прошёл, вебхук потерян

📖 Теория: Outbox на чистом Go, от Dual-Write к идемпотентной доставке

4.1. Transactional Outbox для push (связь с arch m3)

4.2. Воркер доставки: goroutine + FOR UPDATE SKIP LOCKED

4.3. Retry + DLQ

4.4. HMAC-подпись вебхука (связь с arch m7)

4.5. Идемпотентность вебхука (связь с arch m2)

4.6. Воркер — это goroutine, не cron job

4.7. pgx v5 vs database/sql

4.8. Компромиссы (Trade-offs)

🛠 Практические задания

Задача 1: GitStorage.AcceptPush в одной транзакции pgx

Задача 2: outbox.Worker (goroutine + FOR UPDATE SKIP LOCKED + HMAC)

Задача 3: Идемпотентный приёмник

Задача 4: Тест Toxiproxy + k6

🧩 Проверь себя

🔗 Мостик к Модулю 5 (Cache-Aside)

4.2. Воркер доставки: goroutine + `FOR UPDATE SKIP LOCKED`

Задача 1: `GitStorage.AcceptPush` в одной транзакции pgx