12.2K 146 1.0K

Đã đăng vào khoảng 11 giờ trước 30 phút đọc

#7 Bước 5 — Outbox Relay: đưa event từ PostgreSQL sang Kafka Phần 1

Tài liệu thực hành. Kết thúc bước này, dòng event bạn tạo ở bước 4 §16.3 sẽ rời khỏi bảng outbox và xuất hiện trong Kafka UI tại localhost:8080.

Kiến trúc tổng thể: ARCHITECTURE.md — bước này là hiện thực của §7 Outbox Pattern

Ghi chú về module path

Bước 2–4 viết github.com/yourname/community làm ví dụ. Dự án thật đã khởi tạo ở bước 1 với github.com/chuongtd/community, nên từ tài liệu này trở đi tôi ghi đúng path thật. Khi đọc lại code bước 2–4, thay yourname → chuongtd.

Mục tiêu

Kết thúc bước này bạn có:

eventbus/kafka — cài đặt Publisher bằng franz-go, có phân loại lỗi tạm thời / vĩnh viễn
outbox.Relay — worker nền đọc bảng theo lô, publish, đánh dấu, lùi lịch thử lại theo backoff mũ
Cột next_attempt_at và dead_at — hai thứ biến "thử lại" và "bỏ cuộc" từ ý định thành trạng thái có thể truy vấn
cmd/relay — binary thứ hai của hệ thống
Ba bài diễn tập sự cố: Kafka sập, event hỏng, event chết

Từ vựng. Đề bài dùng chữ PENDING / PROCESSED; lược đồ ở bước 3 không có cột status mà dùng dấu thời gian. Ánh xạ như sau, và tôi dùng cách nói của lược đồ trong suốt tài liệu:

Đề bài	Trong bảng `outbox`
`PENDING`	`published_at IS NULL AND dead_at IS NULL AND (next_attempt_at IS NULL OR next_attempt_at <= now())`
`PROCESSED`	`published_at IS NOT NULL`
Chờ thử lại	`next_attempt_at > now()`
Bỏ cuộc (dead letter)	`dead_at IS NOT NULL`

Dùng dấu thời gian thay cho chuỗi trạng thái vì nó trả lời được nhiều câu hỏi hơn với cùng một lượng dữ liệu: status = 'PROCESSED' chỉ nói đã xong, còn published_at nói đã xong lúc nào — và đó chính là số liệu bạn cần khi đo độ trễ hệ thống.

Mục lục

Phạm vi bước này
Nhìn từ trên xuống
Polling, CDC hay LISTEN/NOTIFY?
Migration 000005 — hai cột cho việc thử lại
Cập nhật outbox/query.sql
eventbus/kafka — Publisher
outbox/relay.go
Cấu hình: một Config, hai tiến trình
cmd/relay/main.go
Lệnh tắt
Chạy thử đầu-cuối
Ba bài diễn tập sự cố
Vận hành
Kiểm chứng bằng test
Những thứ cố ý chưa có
Checklist hoàn thành

Phụ lục A — Chạy nhiều relay mà không hỏng thứ tự
Phụ lục B — Phương án "lease" khi transaction quá dài

1. Phạm vi bước này

1.1 Cái đã có sẵn

Thành phần	Từ bước
Bảng `outbox` với `attempts`, `last_error`, `published_at`	3
`EnqueueEvent`, `MarkPublished`, `CountUnpublished` (sqlc)	3
`contracts.Envelope` + `Validate()` + `DecodePayload`	2
`eventbus.Publisher` interface, `ErrPermanent`, `IsPermanent`	2
`postgres.TxManager.Do`	2
`outbox.Writer` — ghi event trong transaction	4
Event thật (`identity.user.registered.v1`) đang nằm trong bảng	4
6 topic Kafka đã tạo, Kafka UI chạy	1

1.2 Cái sẽ viết

db/migrations/
  000005_add_outbox_retry.up.sql            ← MỚI
  000005_add_outbox_retry.down.sql          ← MỚI

internal/platform/
  eventbus/kafka/
    publisher.go                            ← MỚI
    publisher_test.go                       ← MỚI
  outbox/
    query.sql                               ← SỬA
    relay.go                                ← MỚI
    relay_test.go                           ← MỚI
  config/config.go                          ← SỬA (§8)

cmd/relay/main.go                           ← MỚI

1.3 ⚠️ Bước này có thêm migration — và vì sao

Bước 4 tôi đã chứng minh rằng không cần migration nào và không thêm. Bước này thì ngược lại: cần thật.

Bảng outbox hiện có attempts và last_error — đủ để ghi lại rằng một event đã thất bại, nhưng không đủ để làm gì đó với thông tin ấy. Cụ thể, thiếu hai câu trả lời:

Câu hỏi	Trả lời bằng gì	Có chưa
"Đã thử mấy lần rồi?"	`attempts`	✅
"Vì sao hỏng?"	`last_error`	✅
"Bao giờ thì thử lại?"	—	❌
"Khi nào thì thôi không thử nữa?"	—	❌

Không có câu trả lời thứ ba, mọi event hỏng đều bị thử lại ở đúng tần suất poll — một event lỗi với chu kỳ 1 giây là 86.400 lần gọi Kafka thất bại mỗi ngày, kèm 86.400 dòng log ERROR. Không có câu trả lời thứ tư, một event hỏng vĩnh viễn sẽ ở lại trong tập PENDING mãi mãi, khiến chỉ số cảnh báo quan trọng nhất của hệ thống (ARCHITECTURE.md §11.4: số dòng outbox chưa publish) kêu suốt và mất hết ý nghĩa.

Đây là kiểu thiếu sót chỉ lộ ra khi viết đoạn code thật sự tiêu thụ lược đồ. Bước 3 thiết kế bảng outbox cho người ghi; bước 5 là lần đầu có người đọc, và người đọc cần thêm hai cột.

2. Nhìn từ trên xuống

┌─ cmd/api (tiến trình 1) ──────────────────────────────────────────┐
│                                                                   │
│  POST /auth/register                                              │
│    └─ txm.Do:  INSERT users  +  INSERT outbox    ← MỘT transaction│
│         COMMIT                                                    │
│    └─ 201 Created                                                 │
│                                                                   │
│  API KHÔNG nói chuyện với Kafka. Kafka sập → API vẫn trả 201.     │
└───────────────────────────────────────────────────────────────────┘
                              │
                     bảng outbox (PostgreSQL)
                              │
┌─ cmd/relay (tiến trình 2) ─ vòng lặp mỗi 1 giây ──────────────────┐
│                                                                   │
│  BEGIN                                                            │
│    ①  SELECT ... WHERE published_at IS NULL                       │
│              AND dead_at IS NULL                                  │
│              AND (next_attempt_at IS NULL OR next_attempt_at<=now)│
│        ORDER BY id LIMIT 100                                      │
│        FOR UPDATE SKIP LOCKED         ← khoá giữ tới hết BEGIN..  │
│                                                                   │
│    ②  gom theo topic                                              │
│                                                                   │
│    ③  ProduceSync → Kafka            ← gọi MẠNG bên trong tx (§7.4)│
│           thành công ─┐   thất bại ─┬─ vĩnh viễn ─┐               │
│                       │             │             │               │
│    ④  MarkPublished ◄─┘   ScheduleRetry ◄─┘  MarkDead ◄┘          │
│                            (backoff mũ)      (dead_at = now)      │
│  COMMIT                                                           │
└───────────────────────────────────────────────────────────────────┘
                              │
                           Kafka
                              │
                    Kafka UI (localhost:8080)

Ba chi tiết quyết định tính đúng đắn, sẽ được giải thích riêng ở phần sau:

① và ④ phải nằm trong cùng một transaction — nếu không, FOR UPDATE là vô nghĩa (§7.3).
③ nằm giữa hai thứ đó, nghĩa là ta cố ý gọi mạng bên trong transaction — có lý do và có cái giá (§7.4).
Cửa sổ trùng lặp là khoảng từ lúc broker xác nhận (③) tới lúc COMMIT. Chết trong khoảng đó → event được gửi lại ở lượt sau. Đây là bản chất at-least-once, và là lý do bước 6 phải có idempotency guard.

3. Polling, CDC hay `LISTEN/NOTIFY`?

Đề bài gợi ý "polling hoặc CDC/Debezium, đơn giản hoá thì polling trước". Đúng — nhưng đáng biết cả ba lựa chọn để sau này đổi có cơ sở.

Cách	Cơ chế	Độ trễ	Hạ tầng thêm	Đổi được lúc nào
Polling (bước này)	Worker `SELECT` định kỳ	~½ chu kỳ poll (≈500ms)	Không	—
`LISTEN/NOTIFY`	Writer `NOTIFY` khi commit, relay đang `LISTEN` thì thức dậy ngay	~1ms	Không	Bất cứ lúc nào, cộng thêm vào polling
CDC (Debezium)	Đọc WAL của Postgres, đẩy thẳng vào Kafka	~10ms	Kafka Connect + Debezium + slot replication	Khi polling thành nút thắt thật

Vì sao polling trước, và có thể là mãi mãi. Một relay đọc lô 100 dòng, mỗi vòng tốn khoảng 5–10ms → xấp xỉ 2.000 event/giây trên một instance. Phần lớn hệ thống không bao giờ chạm tới con số đó. Polling cũng là cách duy nhất tự phục hồi: dù có bỏ lỡ tín hiệu nào, vòng lặp sau vẫn quét lại toàn bộ tập PENDING.

LISTEN/NOTIFY là nâng cấp rẻ nhất nếu 500ms độ trễ làm phiền bạn. Thêm một dòng vào outbox.Writer:

-- chạy trong CÙNG transaction với INSERT outbox
SELECT pg_notify('outbox_new', '');

rồi cho relay LISTEN outbox_new và dùng tín hiệu đó để rút ngắn thời gian ngủ. Đừng bỏ polling đi — NOTIFY chỉ được giao cho các session đang kết nối tại thời điểm commit; relay vừa khởi động lại sẽ không nhận được gì. Polling là lưới an toàn, NOTIFY là đường tăng tốc.

CDC đổi bản chất bài toán, không chỉ hiệu năng. Debezium đọc write-ahead log nên không cần SELECT, không cần FOR UPDATE, và không tạo tải đọc lên bảng. Cái giá: một replication slot (nếu consumer chết, WAL tích lại và có thể làm đầy đĩa Postgres — đây là sự cố kinh điển của Debezium), thêm Kafka Connect vào vận hành, và mất khả năng định hình message trong Go. Đổi khi có số đo cho thấy polling không đủ, đừng đổi vì nghe hay hơn.

4. Migration `000005` — hai cột cho việc thử lại

Số hiệu. Migration được đánh số theo thứ tự thời gian, không theo chủ đề. Bảng posts từng được nhắc tới với số 000005 trong ARCHITECTURE.md; vì bước này tới trước nên posts lùi xuống 000006. Tôi đã sửa lại các tài liệu cũ cho khớp. Đừng "để dành" số cho việc chưa làm — migrate chạy tuần tự, một khoảng trống là một lỗi chờ sẵn.

db/migrations/000005_add_outbox_retry.up.sql:

-- next_attempt_at: sớm nhất là lúc nào thì được thử lại.
-- NULL = chưa từng thất bại, sẵn sàng ngay. Cùng quy ước "NULL nghĩa là
-- chưa xảy ra" với published_at, nên đọc lược đồ không phải nhớ hai luật.
ALTER TABLE outbox ADD COLUMN next_attempt_at TIMESTAMPTZ;

-- dead_at: đã bỏ cuộc, cần người xem. Dòng có dead_at rời khỏi tập PENDING
-- nên không còn làm nhiễu chỉ số "hàng chờ còn bao nhiêu".
ALTER TABLE outbox ADD COLUMN dead_at TIMESTAMPTZ;

COMMENT ON COLUMN outbox.next_attempt_at IS
    'NULL = sẵn sàng gửi. Có giá trị = đang bị lùi lịch sau một lần thất bại.';
COMMENT ON COLUMN outbox.dead_at IS
    'Đã vượt số lần thử tối đa hoặc gặp lỗi vĩnh viễn. Đây LÀ dead letter queue.';

-- Index cũ chỉ loại các dòng đã gửi; giờ phải loại cả các dòng đã chết.
DROP INDEX IF EXISTS idx_outbox_unpublished;

CREATE INDEX idx_outbox_pending
    ON outbox (id)
    WHERE published_at IS NULL AND dead_at IS NULL;

-- Cho màn hình / cảnh báo "có gì chết không".
CREATE INDEX idx_outbox_dead
    ON outbox (dead_at)
    WHERE dead_at IS NOT NULL;

db/migrations/000005_add_outbox_retry.down.sql:

DROP INDEX IF EXISTS idx_outbox_dead;
DROP INDEX IF EXISTS idx_outbox_pending;

CREATE INDEX idx_outbox_unpublished
    ON outbox (id)
    WHERE published_at IS NULL;

ALTER TABLE outbox DROP COLUMN IF EXISTS dead_at;
ALTER TABLE outbox DROP COLUMN IF EXISTS next_attempt_at;

Chạy:

migrate -path db/migrations -database "postgres://app:secret@localhost:5432/community?sslmode=disable" up

Kiểm chứng:

docker compose -f deployments/docker-compose.yml exec postgres psql -U app -d community -c "\d outbox"

Phải thấy hai cột mới và ba index (outbox_pkey, idx_outbox_pending, idx_outbox_dead, outbox_event_id_key).

4.1 Vì sao `next_attempt_at` để NULL được thay vì `NOT NULL DEFAULT now()`

Hai lý do, một về ngữ nghĩa và một về vận hành.

Ngữ nghĩa: bảng này đã có hai cột theo quy ước NULL = chưa xảy ra (published_at, last_error). Cột thứ ba theo cùng quy ước thì người đọc lược đồ không phải nhớ ngoại lệ.

Vận hành: ALTER TABLE ... ADD COLUMN với cột nullable, không default là thao tác chỉ sửa metadata — tức thời, không đụng tới một dòng dữ liệu nào, không giữ khoá lâu. Trên bảng outbox của môi trường dev thì khác biệt bằng không, nhưng thói quen này là thứ phân biệt một migration chạy được lúc 3 giờ sáng với một migration khoá bảng 40 giây trên production.

4.2 ⚠️ Vì sao dead letter ở đây là một bảng, không phải một topic

Bước 1 đã tạo identity.DLQ, post.DLQ, social.DLQ. Câu hỏi tự nhiên: sao relay không đẩy event hỏng vào đó?

Vì DLQ của relay và DLQ của consumer giải quyết hai vấn đề khác nhau, và trộn chúng lại làm cái đầu tiên trở nên vô nghĩa.

	DLQ của consumer (bước 6)	"DLQ" của relay (bước này)
Event đã tới Kafka chưa?	Rồi	Chưa — đó chính là vấn đề
Nó đang nằm ở đâu	Chỉ trong Kafka	Trong PostgreSQL, an toàn, bền
Thất bại ở khâu nào	Xử lý	Giao hàng
Chuyển vào DLQ topic có ích không	Có — đưa nó ra khỏi luồng chính để partition không bị chặn	Không

Ba lý do cụ thể cho ô cuối cùng:

Nghịch lý gà–trứng. Lỗi giao hàng vĩnh viễn hay gặp nhất là MESSAGE_TOO_LARGE. Copy đúng message ấy sang topic identity.DLQ sẽ thất bại vì đúng cái lỗi đó. Cơ chế xử lý sự cố mà chính nó cũng chết vì sự cố thì không phải cơ chế.
Không thêm được độ bền nào. Event đã nằm trong PostgreSQL — có volume, có backup, có transaction. Chép nó sang Kafka để "khỏi mất" là đi lùi.
Phát lại phức tạp hơn hẳn. Từ bảng, phát lại là một câu UPDATE (§13.2). Từ một DLQ topic, bạn cần một consumer riêng, biết cách đọc, biết cách bơm ngược vào topic gốc, và tự lo chuyện đọc trùng.

Nên: cột dead_at chính là dead letter queue của relay. Ba topic .DLQ giữ nguyên, chúng dành cho consumer ở bước 6 — nơi chúng thật sự đúng việc.

5. Cập nhật `outbox/query.sql`

Đính chính bước 3. §9.3 khai báo FetchUnpublished, MarkFailed, CountUnpublished. Sau khi có hai cột mới, ba query đó không còn mô tả đúng việc chúng làm nữa: điều kiện "chưa gửi" giờ gồm ba mệnh đề chứ không phải một, và "thất bại" tách thành hai kết cục khác nhau. Dưới đây là file thay thế trọn vẹn.

internal/platform/outbox/query.sql:

-- name: EnqueueEvent :exec
-- Gọi TRONG transaction của nghiệp vụ. Đây là toàn bộ ý nghĩa của Outbox Pattern.
INSERT INTO outbox (
    event_id, event_type, topic, aggregate_id, correlation_id, payload, occurred_at
) VALUES ($1, $2, $3, $4, $5, $6, $7);

-- name: FetchPending :many
-- Lọc theo TRẠNG THÁI, không theo vị trí con trỏ — xem bước 3 §5.2.
-- SKIP LOCKED cho phép chạy nhiều relay song song (đọc kèm Phụ lục A).
-- PHẢI gọi trong transaction, nếu không FOR UPDATE không giữ được gì (§7.3).
SELECT id, event_id, event_type, topic, aggregate_id,
       correlation_id, payload, occurred_at, attempts
FROM outbox
WHERE published_at IS NULL
  AND dead_at IS NULL
  AND (next_attempt_at IS NULL OR next_attempt_at <= now())
ORDER BY id
LIMIT $1
FOR UPDATE SKIP LOCKED;

-- name: MarkPublished :exec
-- Đánh dấu cả lô bằng một lệnh thay vì N lệnh.
UPDATE outbox SET published_at = now() WHERE id = ANY($1::bigint[]);

-- name: ScheduleRetry :exec
-- Backoff mũ tính NGAY TRONG SQL, dựa trên attempts của từng dòng:
--   1s, 2s, 4s, 8s, ... trần 5 phút.
-- Nhờ vậy một lệnh xử lý được cả lô dù mỗi dòng đang ở lần thử khác nhau.
-- (Trong UPDATE, mọi biểu thức SET đọc giá trị CŨ của dòng, nên `attempts`
--  ở vế phải là số lần đã thử TRƯỚC lần này.)
UPDATE outbox
SET attempts        = attempts + 1,
    last_error      = $2,
    next_attempt_at = now() + LEAST(
        POWER(2, LEAST(attempts, 10)) * INTERVAL '1 second',
        INTERVAL '5 minutes'
    )
WHERE id = ANY($1::bigint[]);

-- name: MarkDead :exec
-- Bỏ cuộc: lỗi vĩnh viễn, hoặc đã vượt số lần thử tối đa.
UPDATE outbox
SET attempts   = attempts + 1,
    last_error = $2,
    dead_at    = now()
WHERE id = ANY($1::bigint[]);

-- name: CountPending :one
-- Chỉ số giám sát quan trọng nhất của hệ thống (ARCHITECTURE.md §11.4).
SELECT count(*) FROM outbox
WHERE published_at IS NULL AND dead_at IS NULL;

-- name: CountDead :one
SELECT count(*) FROM outbox WHERE dead_at IS NOT NULL;

-- name: OldestPendingAge :one
-- Độ trễ thật của hệ thống: event cũ nhất chưa gửi đã chờ bao lâu.
-- Trả NULL khi hàng chờ rỗng.
SELECT now() - min(occurred_at) FROM outbox
WHERE published_at IS NULL AND dead_at IS NULL;

-- name: DeletePublishedBefore :execrows
-- CHỈ xoá dòng đã gửi. Dòng dead_at phải ở lại tới khi có người xử lý.
DELETE FROM outbox WHERE published_at IS NOT NULL AND published_at < $1;

Sinh lại code:

sqlc generate

Kiểm chứng nhanh các hàm mới có mặt:

Select-String -Path internal/platform/outbox/gen/query.sql.go -Pattern "^func \(q \*Queries\)"

func (q *Queries) CountDead(ctx context.Context) (int64, error)
func (q *Queries) CountPending(ctx context.Context) (int64, error)
func (q *Queries) DeletePublishedBefore(ctx context.Context, publishedAt time.Time) (int64, error)
func (q *Queries) EnqueueEvent(ctx context.Context, arg EnqueueEventParams) error
func (q *Queries) FetchPending(ctx context.Context, limit int32) ([]FetchPendingRow, error)
func (q *Queries) MarkDead(ctx context.Context, arg MarkDeadParams) error
func (q *Queries) MarkPublished(ctx context.Context, ids []int64) error
func (q *Queries) OldestPendingAge(ctx context.Context) (interface{}, error)
func (q *Queries) ScheduleRetry(ctx context.Context, arg ScheduleRetryParams) error

OldestPendingAge trả interface{} — vì sqlc không suy được kiểu của biểu thức now() - min(...) (Postgres trả interval, không có kiểu Go tương ứng gọn gàng). Đây là hàm dành cho câu SQL gõ tay lúc điều tra, không phải cho code nghiệp vụ; nếu thấy vướng mắt thì bỏ nó khỏi query.sql và dùng thẳng psql. Tôi giữ lại để câu truy vấn quan trọng này nằm cùng chỗ với các câu còn lại.

6. `eventbus/kafka` — Publisher

New-Item -ItemType Directory -Force internal/platform/eventbus/kafka

6.1 Vấn đề chữ ký: relay cần kết quả của từng event

eventbus.Publisher từ bước 2 trả về một error cho cả lô:

Publish(ctx context.Context, topic string, events ...contracts.Envelope) error

Với relay, chữ ký đó không đủ. Xét một lô 100 event trong đó đúng một event vượt quá max.message.bytes của broker:

Nếu coi cả lô là hỏng → 99 event khoẻ mạnh bị lùi lịch cùng nhau, và sau MaxAttempts lần thì cả 100 dòng cùng bị đánh dấu chết. Một dòng xấu giết 99 dòng tốt.
Nếu coi cả lô là xong → event quá khổ biến mất không dấu vết.

Nên relay cần biết chính xác event nào trượt. Cách giải: relay tự khai báo interface nó cần, đúng mẫu đã dùng ở bước 4 §13.1. kafka.Publisher cài đặt cả hai, và cái rộng hơn định nghĩa cái hẹp hơn:

func (p *Publisher) Publish(ctx context.Context, topic string, events ...contracts.Envelope) error {
	return errors.Join(p.PublishEach(ctx, topic, events)...)
}

errors.Join với toàn nil trả về nil, nên một dòng này vừa giữ nguyên hợp đồng cũ vừa xoá bỏ trùng lặp.

6.2 `publisher.go`

internal/platform/eventbus/kafka/publisher.go:

// Package kafka cài đặt eventbus.Publisher bằng franz-go.
//
// Chỉ tiến trình relay dùng package này. Module nghiệp vụ ghi vào outbox
// và không bao giờ nhìn thấy Kafka (ARCHITECTURE.md §7).
package kafka

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/twmb/franz-go/pkg/kerr"
	"github.com/twmb/franz-go/pkg/kgo"

	"github.com/chuongtd/community/internal/contracts"
	"github.com/chuongtd/community/internal/platform/eventbus"
)

type Config struct {
	Brokers  []string
	ClientID string

	// ProduceTimeout là trần thời gian cho MỘT lô, tính cả các lần
	// franz-go tự thử lại bên trong. Relay giữ một transaction mở
	// trong suốt lời gọi này, nên con số ở đây là con số quan trọng —
	// xem relay.go §7.4.
	ProduceTimeout time.Duration
}

type Publisher struct {
	cl      *kgo.Client
	timeout time.Duration
}

func NewPublisher(cfg Config) (*Publisher, error) {
	if len(cfg.Brokers) == 0 {
		return nil, errors.New("kafka: thiếu danh sách broker")
	}
	if cfg.ProduceTimeout <= 0 {
		cfg.ProduceTimeout = 10 * time.Second
	}
	if cfg.ClientID == "" {
		cfg.ClientID = "community"
	}

	cl, err := kgo.NewClient(
		kgo.SeedBrokers(cfg.Brokers...),

		// ClientID hiện trong log broker và trong Kafka UI. Cùng lý lẽ với
		// application_name của Postgres (bước 2 §3.3): lúc có sự cố, câu hỏi
		// đầu tiên luôn là "ai đang làm việc này".
		kgo.ClientID(cfg.ClientID),

		// acks=all — chỉ tính là thành công khi mọi replica trong ISR đã ghi.
		// Đây cũng là mặc định của franz-go; viết ra để người đọc thấy đó là
		// lựa chọn có ý thức. Với acks=1, broker leader chết ngay sau khi trả
		// lời sẽ làm mất event mà relay đã đánh dấu published — mất vĩnh viễn,
		// vì dòng outbox không còn nằm trong tập PENDING nữa.
		kgo.RequiredAcks(kgo.AllISRAcks()),

		// Trần cho toàn bộ vòng đời một record, kể cả các lần thử lại nội bộ.
		// Không có nó, ProduceSync có thể treo tới khi ctx hết hạn.
		kgo.RecordDeliveryTimeout(cfg.ProduceTimeout),
		kgo.ProduceRequestTimeout(cfg.ProduceTimeout/2),
		kgo.RequestRetries(3),
	)
	if err != nil {
		return nil, fmt.Errorf("kafka: tạo client: %w", err)
	}

	// Ping để lỗi cấu hình lộ ra lúc khởi động — cùng lý lẽ với pool.Ping
	// ở bước 2 §3.1. Xem ghi chú §6.5 về việc này với một relay.
	pingCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if err := cl.Ping(pingCtx); err != nil {
		cl.Close()
		return nil, fmt.Errorf("kafka: không kết nối được tới %v: %w", cfg.Brokers, err)
	}

	return &Publisher{cl: cl, timeout: cfg.ProduceTimeout}, nil
}

func (p *Publisher) Close() { p.cl.Close() }

// Publish giữ nguyên hợp đồng eventbus.Publisher: một lỗi cho cả lô.
func (p *Publisher) Publish(ctx context.Context, topic string, events ...contracts.Envelope) error {
	return errors.Join(p.PublishEach(ctx, topic, events)...)
}

// PublishEach trả về một lỗi cho MỖI event, cùng chỉ số với events.
// errs[i] == nil nghĩa là event thứ i đã được broker xác nhận ghi.
//
// Độ dài trả về luôn bằng len(events).
func (p *Publisher) PublishEach(ctx context.Context, topic string, events []contracts.Envelope) []error {
	errs := make([]error, len(events))
	if len(events) == 0 {
		return errs
	}
	if topic == "" {
		fill(errs, eventbus.Permanent(errors.New("kafka: thiếu topic")))
		return errs
	}

	records := make([]*kgo.Record, len(events))
	for i, e := range events {
		rec, err := toRecord(topic, e)
		if err != nil {
			errs[i] = err // đã được đánh dấu Permanent bên trong toRecord
			continue
		}
		records[i] = rec
	}

	// Mặc định là "chưa có kết quả". Nếu vì lý do nào đó broker không trả
	// lời về một record, ta thà thử lại (gửi trùng) còn hơn đánh dấu đã gửi
	// một event chưa bao giờ rời khỏi máy này.
	send := make([]*kgo.Record, 0, len(records))
	index := make(map[*kgo.Record]int, len(records))
	for i, rec := range records {
		if rec == nil {
			continue
		}
		errs[i] = errors.New("kafka: không nhận được kết quả produce")
		index[rec] = i
		send = append(send, rec)
	}
	if len(send) == 0 {
		return errs
	}

	ctx, cancel := context.WithTimeout(ctx, p.timeout)
	defer cancel()

	for _, res := range p.cl.ProduceSync(ctx, send...) {
		i, ok := index[res.Record]
		if !ok {
			continue
		}
		errs[i] = classify(res.Err)
	}
	return errs
}

func fill(errs []error, err error) {
	for i := range errs {
		errs[i] = err
	}
}

6.3 `Envelope` → `kgo.Record`

// toRecord chuyển envelope thành record Kafka.
//
// Toàn bộ envelope đi vào Value; Key và Headers là bản sao có chủ đích của
// vài trường bên trong nó, phục vụ hai việc khác nhau — xem §6.4.
func toRecord(topic string, e contracts.Envelope) (*kgo.Record, error) {
	// Kiểm tra ở BIÊN GIỚI ra khỏi hệ thống. Một envelope thiếu trường lọt
	// vào Kafka sẽ làm mọi consumer thất bại, và lúc đó rất khó lần ngược
	// về người đã phát nó.
	if err := e.Validate(); err != nil {
		return nil, eventbus.Permanent(fmt.Errorf("kafka: %w", err))
	}

	body, err := json.Marshal(e)
	if err != nil {
		// Envelope đã ở trong bảng; mã hoá lại vẫn hỏng ở lần sau.
		return nil, eventbus.Permanent(
			fmt.Errorf("kafka: mã hoá envelope %s: %w", e.EventID, err))
	}

	rec := &kgo.Record{
		Topic: topic,

		// ★ Key = AggregateID. Đây là dòng biến lời hứa thứ tự của
		// ARCHITECTURE.md §5.2 thành hành vi thật: Kafka băm Key để chọn
		// partition, nên mọi event của cùng một thực thể luôn rơi vào cùng
		// một partition, và trong một partition thì thứ tự được bảo đảm.
		//
		// Bỏ dòng này (Key = nil) thì record được rải đều mọi partition,
		// và "user X đổi tên rồi bị xoá" có thể tới consumer theo thứ tự
		// ngược lại. Bug loại đó không bao giờ tái hiện được trên máy dev.
		Key:   []byte(e.AggregateID),
		Value: body,

		Headers: []kgo.RecordHeader{
			{Key: "event_id", Value: []byte(e.EventID)},
			{Key: "event_type", Value: []byte(e.EventType)},
			{Key: "correlation_id", Value: []byte(e.CorrelationID)},
		},
	}
	return rec, nil
}

6.4 Vì sao lặp lại dữ liệu trong header

Ba header trên đã có sẵn trong Value. Lặp lại là cố ý:

Người dùng	Dùng header để làm gì
Bạn, trong Kafka UI	Thấy ngay `event_type` ở cột metadata mà không phải bung JSON từng message một. Khi đang dò tìm giữa vài nghìn message, khác biệt này rất lớn
Công cụ vận hành	Lọc/định tuyến theo `event_type` mà không cần giải mã payload — nghĩa là không cần biết schema
Truy vết sự cố	`correlation_id` nối HTTP request → dòng outbox → message Kafka → log consumer thành một chuỗi grep được

Cái giá là vài chục byte mỗi message. Đáng.

Không đưa gì khác vào header. Header không được kiểm tra kiểu, không có version, và cám dỗ "để tạm một trường ở header cho nhanh" sẽ tạo ra một hợp đồng thứ hai chạy song song với contracts/v1 mà không ai ghi lại. Header là bản sao chỉ-đọc của metadata, không phải chỗ chứa dữ liệu.

6.5 Phân loại lỗi Kafka

// permanentErrors là các lỗi mà thử lại chắc chắn cho kết quả y hệt.
// Danh sách này CỐ Ý ngắn: theo bước 2 §6.1, mặc định an toàn là retry,
// nên chỉ đưa vào đây thứ đã chắc chắn.
var permanentErrors = []error{
	kerr.MessageTooLarge,     // record vượt max.message.bytes của broker
	kerr.RecordListTooLarge,  // cả lô vượt giới hạn
	kerr.InvalidRecord,       // broker từ chối định dạng record
	kerr.InvalidTopicException, // tên topic sai luật đặt tên
}

func classify(err error) error {
	if err == nil {
		return nil
	}
	for _, p := range permanentErrors {
		if errors.Is(err, p) {
			return eventbus.Permanent(err)
		}
	}
	return err
}

⚠️ Vì sao UNKNOWN_TOPIC_OR_PARTITION không nằm trong danh sách

Nó trông vĩnh viễn: topic không tồn tại, thử lại thì vẫn không tồn tại. Nhưng cách sửa nằm ngoài tiến trình này — ai đó chạy kafka-topics.sh --create, hoặc một dev.ps1 topics bị quên sau khi docker compose down -v. Ngay khi topic có mặt, event cũ chưa gửi phải tự chảy tiếp mà không cần ai can thiệp vào database.

Coi nó là tạm thời cho đúng hành vi đó, và cơ chế MaxAttempts vẫn đảm bảo nó không thử lại vô hạn: sau MaxAttempts lần, dòng chuyển sang dead_at kèm last_error nói rõ topic nào thiếu.

Đây là ứng dụng trực tiếp của nguyên tắc ở bước 2 §6.1: chọn mặc định sao cho đoán sai gây hậu quả nhẹ hơn. Đoán sai theo hướng này = thử lại thừa vài lần. Đoán sai theo hướng kia = chôn event chỉ vì ai đó quên tạo topic.

6.6 Ghi chú: producer idempotent

franz-go bật idempotent producer theo mặc định. Nó đánh số thứ tự các record gửi từ cùng một client và broker loại bỏ bản trùng — nên khi franz-go tự thử lại bên trong (RequestRetries(3)), bạn không bị ghi trùng vào Kafka.

Điều này che được các lần trùng do mạng chập chờn, nhưng không che được cửa sổ trùng ở §2 (broker xác nhận xong → relay chết → COMMIT không chạy). Lần chạy sau là một client mới, với chuỗi số mới, nên broker coi đó là record hoàn toàn khác.

Kết luận không đổi: hệ thống này là at-least-once, và consumer bắt buộc phải idempotent. Đó là bước 6.

6.7 Ghi chú: `Ping` lúc khởi động với một relay

NewPublisher gọi Ping và trả lỗi nếu broker không trả lời — nghĩa là relay không khởi động được khi Kafka đang sập. Nghe có vẻ mâu thuẫn với mục đích của outbox pattern (chịu được Kafka sập), nên cần nói rõ đây là đánh đổi có chủ đích:

Lỗi thường gặp hơn nhiều là gõ sai địa chỉ broker trong .env. Chỉ fail-fast mới bắt được nó; nếu bỏ qua, relay chạy im lặng và bạn phát hiện sau vài giờ khi nhận ra chẳng có event nào đi đâu cả.
Kafka sập trong lúc relay đang chạy vẫn được xử lý đúng — publish thất bại, event lùi lịch, không mất gì. Đó mới là kịch bản outbox pattern sinh ra để giải quyết.
Trường hợp còn lại (Kafka sập và relay khởi động lại cùng lúc) biến thành crash-loop. Với restart: unless-stopped của Docker hoặc restartPolicy của Kubernetes, relay tự lên lại ngay khi Kafka lên. Không cần code gì thêm.

7. `outbox/relay.go`

7.1 Khai báo phụ thuộc

internal/platform/outbox/relay.go:

package outbox

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"time"
	"unicode/utf8"

	"github.com/jackc/pgx/v5"

	"github.com/chuongtd/community/internal/contracts"
	"github.com/chuongtd/community/internal/platform/eventbus"
	"github.com/chuongtd/community/internal/platform/outbox/gen"
)

// Publisher là đúng phần Kafka mà relay cần — không hơn.
//
// Khai báo ở ĐÂY, phía dùng, chứ không ở phía cài đặt. Nhờ vậy toàn bộ
// test của relay chạy với một publisher giả và KHÔNG cần Kafka (§14.2).
type Publisher interface {
	PublishEach(ctx context.Context, topic string, events []contracts.Envelope) []error
}

// Transactor là phần TxManager mà relay cần.
type Transactor interface {
	Do(ctx context.Context, fn func(tx pgx.Tx) error) error
}

type RelayConfig struct {
	// BatchSize là số dòng đọc mỗi lượt. Càng lớn thì càng ít lượt đi về
	// DB, nhưng transaction cũng mở lâu hơn (§7.4).
	BatchSize int

	// PollInterval là thời gian ngủ khi hàng chờ rỗng.
	PollInterval time.Duration

	// MaxAttempts là số lần thử trước khi bỏ cuộc. 10 lần với backoff mũ
	// trần 5 phút ≈ 25 phút cố gắng — đủ để một sự cố Kafka ngắn tự khỏi.
	MaxAttempts int

	// MaxErrorLen cắt bớt last_error. Một lỗi Kafka có thể dài vài KB;
	// nhân với số dòng hỏng thì bảng phình rất nhanh (§7.6).
	MaxErrorLen int
}

func (c *RelayConfig) setDefaults() {
	if c.BatchSize <= 0 {
		c.BatchSize = 100
	}
	if c.PollInterval <= 0 {
		c.PollInterval = time.Second
	}
	if c.MaxAttempts <= 0 {
		c.MaxAttempts = 10
	}
	if c.MaxErrorLen <= 0 {
		c.MaxErrorLen = 500
	}
}

type Relay struct {
	tx  Transactor
	pub Publisher
	log *slog.Logger
	cfg RelayConfig
}

func NewRelay(tx Transactor, pub Publisher, log *slog.Logger, cfg RelayConfig) *Relay {
	cfg.setDefaults()
	return &Relay{tx: tx, pub: pub, log: log, cfg: cfg}
}

7.2 Vòng lặp

// Run chạy tới khi ctx bị huỷ. Chặn luồng gọi.
func (r *Relay) Run(ctx context.Context) error {
	r.log.Info("relay bắt đầu",
		"batch_size", r.cfg.BatchSize,
		"poll_interval", r.cfg.PollInterval,
		"max_attempts", r.cfg.MaxAttempts)

	// Timer 0 để lượt đầu chạy ngay, không phải đợi hết một chu kỳ.
	t := time.NewTimer(0)
	defer t.Stop()

	for {
		select {
		case <-ctx.Done():
			r.log.Info("relay dừng theo tín hiệu")
			return nil
		case <-t.C:
		}

		n, err := r.tick(ctx)

		switch {
		case errors.Is(err, context.Canceled):
			// Đang tắt máy giữa chừng: transaction đã rollback, các dòng
			// vẫn ở tập PENDING. Không có gì hỏng, không cần log ERROR.
			r.log.Info("relay dừng giữa lô, các event sẽ được gửi lại")
			return nil

		case err != nil:
			// KHÔNG return. Postgres nấc một nhịp không phải lý do để
			// giết cả tiến trình — lượt sau thử lại.
			r.log.Error("relay: lô thất bại", "err", err)
			t.Reset(r.cfg.PollInterval)

		case n >= r.cfg.BatchSize:
			// Lô đầy ⇒ gần như chắc chắn còn nữa. Chạy tiếp NGAY.
			t.Reset(0)

		default:
			t.Reset(r.cfg.PollInterval)
		}
	}
}

Vì sao "lô đầy thì chạy tiếp ngay" không phải chi tiết vụn vặt

Bỏ nhánh đó đi, relay xử lý tối đa BatchSize event mỗi PollInterval. Với mặc định 100 và 1 giây, trần thông lượng là 100 event/giây — bất kể Kafka và Postgres còn rảnh bao nhiêu.

Một đợt import 50.000 event sẽ mất 8 phút 20 giây, và trong suốt thời gian đó relay ngủ 99% thời gian. Có nhánh đó, cùng khối lượng ấy chảy hết trong khoảng 25 giây, giới hạn bởi tốc độ thật của Kafka.

PollInterval là thời gian ngủ khi rỗi, không phải nhịp tim của hệ thống. Nhầm hai khái niệm này là cách phổ biến nhất để tự đặt trần thông lượng cho chính mình mà không biết.

7.3 ⚠️ Vì sao `SELECT` và `UPDATE` phải cùng một transaction

func (r *Relay) tick(ctx context.Context) (int, error) {
	var fetched int

	err := r.tx.Do(ctx, func(tx pgx.Tx) error {
		q := gen.New(tx)

		rows, err := q.FetchPending(ctx, int32(r.cfg.BatchSize))
		if err != nil {
			return fmt.Errorf("outbox: đọc hàng chờ: %w", err)
		}
		fetched = len(rows)
		if fetched == 0 {
			return nil
		}
		return r.publishAndMark(ctx, q, rows)
	})

	return fetched, err
}

Bản phác ở ARCHITECTURE.md §7.4 viết r.db.Query(...) — tức là chạy ngoài transaction. Viết như vậy thì FOR UPDATE SKIP LOCKED không bảo vệ được gì, và đây là điểm tôi muốn nói rõ vì lỗi này im lặng tuyệt đối.

Khoá dòng trong PostgreSQL tồn tại cho tới hết transaction. Chạy SELECT ... FOR UPDATE ngoài transaction tường minh nghĩa là nó nằm trong một transaction ngầm chỉ dài đúng bằng câu lệnh đó: khoá được lấy rồi nhả ngay khi câu lệnh trả về.

Relay A                          Relay B
────────────────────────────────────────────────────────
SELECT ... FOR UPDATE
  → nhận id 1..100
  → khoá NHẢ ngay tại đây ❌
                                 SELECT ... FOR UPDATE
                                   → SKIP LOCKED không thấy gì bị khoá
                                   → cũng nhận id 1..100  ❌
publish 1..100                   publish 1..100
                                 → mọi event bị gửi HAI lần

Với transaction tường minh, khoá được giữ từ FetchPending cho tới COMMIT — tức là suốt cả quá trình publish và đánh dấu. Relay B gọi SELECT trong khoảng đó sẽ bỏ qua 100 dòng đang khoá và nhận dòng 101 trở đi. Đó chính là điều SKIP LOCKED được sinh ra để làm.

Quy tắc rút gọn: FOR UPDATE ngoài transaction tường minh gần như luôn là bug. Nó biên dịch được, chạy được, cho kết quả đúng trên máy dev một mình, và sai ngay khi có instance thứ hai.

7.4 ⚠️ Cái giá của việc gọi mạng trong transaction

Kết cấu trên đặt một lời gọi mạng (ProduceSync) vào giữa một transaction đang mở. Ở bước 4 §3 tôi đã cảnh báo đúng điều ngược lại: băm bcrypt phải nằm ngoài transaction. Hai chỗ khác nhau ở đâu?

	`cmd/api` băm mật khẩu	`cmd/relay` publish Kafka
Ai chờ	Người dùng thật, đang nhìn màn hình	Không ai
Pool dùng chung với	Mọi request khác	Không gì cả — relay chạy một luồng
Có cách khác không	Có — băm trước rồi mới `BEGIN`	Không — bỏ khoá ra khỏi tay là mở cửa cho gửi trùng
Hậu quả khi chậm	Cạn pool ở ~40 req/s, toàn hệ thống 500	Relay chậm lại, hàng chờ dài ra, không ai thấy

Nói ngắn: bcrypt-trong-transaction là lãng phí không cần thiết; publish-trong-transaction là cái giá của tính đúng đắn. Việc cần làm không phải né nó mà là giới hạn nó:

Trần thời gian cứng. RecordDeliveryTimeout = ProduceTimeout (mặc định 10s). Không có nó, một broker treo sẽ giữ transaction mở vô hạn.
Lô vừa phải. 100 dòng ≈ vài chục ms trên mạng LAN. Tăng lên 10.000 để "chạy nhanh hơn" chỉ làm transaction dài hơn mà không nhanh hơn bao nhiêu.
Pool riêng và nhỏ. Relay dùng MaxConns = 5 theo bảng ngân sách ở bước 2 §3.2. Nó không dùng chung pool với api.

⚠️ idle_in_transaction_session_timeout

Trong lúc ProduceSync chờ Kafka, session Postgres của relay ở trạng thái idle in transaction. PostgreSQL mặc định idle_in_transaction_session_timeout = 0 (tắt), nhưng rất nhiều dịch vụ managed đặt sẵn 60 giây — và một số công ty đặt xuống 10 giây trong postgresql.conf để chống transaction bỏ quên.

Nếu ProduceTimeout lớn hơn ngưỡng đó, Postgres sẽ giết session ngay giữa lô. Triệu chứng là lỗi FATAL: terminating connection due to idle-in-transaction timeout, xuất hiện chỉ khi Kafka chậm — nghĩa là đúng lúc bạn đang bận xử lý sự cố Kafka.

Kiểm tra một lần cho yên tâm:
SHOW idle_in_transaction_session_timeout;
Quy tắc: ProduceTimeout phải nhỏ hơn rõ rệt giá trị đó. Nếu ngưỡng là 60s thì 10s là hợp lý.

Khi cả ba giới hạn trên vẫn không đủ, Phụ lục B có phương án "lease" — rút transaction xuống còn một câu UPDATE, đổi lại cửa sổ trùng lặp dài hơn.

7.5 Publish và đánh dấu

func (r *Relay) publishAndMark(ctx context.Context, q *gen.Queries, rows []gen.FetchPendingRow) error {
	type item struct {
		row gen.FetchPendingRow
		env contracts.Envelope
	}

	var (
		// Một lô có thể chứa event của nhiều topic, nhưng PublishEach chỉ
		// nhận một topic. Gom lại để mỗi topic đúng một lượt đi về mạng.
		byTopic = make(map[string][]item)

		published []int64
		retry     = make(map[string][]int64) // lỗi → các id lùi lịch
		dead      = make(map[string][]int64) // lỗi → các id bỏ cuộc
	)

	// Gom các id theo NỘI DUNG LỖI: thường cả lô hỏng vì cùng một nguyên
	// nhân (Kafka sập), nên map này gần như luôn chỉ có một phần tử và ta
	// ghi được cả lô bằng một câu UPDATE.
	fail := func(row gen.FetchPendingRow, err error) {
		msg := truncate(err.Error(), r.cfg.MaxErrorLen)
		if eventbus.IsPermanent(err) || int(row.Attempts)+1 >= r.cfg.MaxAttempts {
			dead[msg] = append(dead[msg], row.ID)
			return
		}
		retry[msg] = append(retry[msg], row.ID)
	}

	for _, row := range rows {
		env, err := toEnvelope(row)
		if err != nil {
			// Dòng hỏng ngay trong bảng. Thử lại bao nhiêu lần cũng thế.
			fail(row, err)
			continue
		}
		byTopic[row.Topic] = append(byTopic[row.Topic], item{row: row, env: env})
	}

	for topic, items := range byTopic {
		envs := make([]contracts.Envelope, len(items))
		for i, it := range items {
			envs[i] = it.env
		}

		errs := r.pub.PublishEach(ctx, topic, envs)
		if len(errs) != len(items) {
			// Publisher vi phạm hợp đồng. Trả lỗi để CẢ transaction rollback:
			// không dòng nào bị đánh dấu, lượt sau làm lại từ đầu.
			return fmt.Errorf("outbox: publisher trả %d kết quả cho %d event (topic %s)",
				len(errs), len(items), topic)
		}

		for i, it := range items {
			if errs[i] == nil {
				published = append(published, it.row.ID)
				continue
			}
			fail(it.row, errs[i])
		}
	}

	if len(published) > 0 {
		if err := q.MarkPublished(ctx, published); err != nil {
			return fmt.Errorf("outbox: đánh dấu đã gửi: %w", err)
		}
	}

	for msg, ids := range retry {
		msg := msg // cần địa chỉ ổn định cho tham số *string
		if err := q.ScheduleRetry(ctx, gen.ScheduleRetryParams{Ids: ids, LastError: &msg}); err != nil {
			return fmt.Errorf("outbox: lùi lịch thử lại: %w", err)
		}
		r.log.Warn("outbox: lùi lịch thử lại", "count", len(ids), "err", msg)
	}

	for msg, ids := range dead {
		msg := msg
		if err := q.MarkDead(ctx, gen.MarkDeadParams{Ids: ids, LastError: &msg}); err != nil {
			return fmt.Errorf("outbox: đánh dấu chết: %w", err)
		}
		// ERROR chứ không phải WARN: dòng chết cần một con người đọc.
		r.log.Error("outbox: event bỏ cuộc, cần xử lý thủ công",
			"ids", ids, "count", len(ids), "err", msg)
	}

	if len(published) > 0 {
		r.log.Info("outbox: đã gửi", "count", len(published), "topics", len(byTopic))
	}
	return nil
}

msg := msg bên trong vòng lặp. ScheduleRetryParams.LastError là *string (cột last_error cho phép NULL), nên cần một địa chỉ. Từ Go 1.22 biến vòng lặp đã là biến mới mỗi vòng nên &msg trực tiếp cũng an toàn; tôi vẫn viết tường minh vì đoạn code này sẽ được đọc bởi người từng bị cắn bởi hành vi cũ, và một dòng thừa rẻ hơn một phút phân vân.

7.6 Hai hàm phụ

func toEnvelope(row gen.FetchPendingRow) (contracts.Envelope, error) {
	e := contracts.Envelope{
		EventID:     row.EventID.String(),
		EventType:   row.EventType,
		AggregateID: row.AggregateID,
		OccurredAt:  row.OccurredAt,
		Payload:     json.RawMessage(row.Payload),
	}
	if row.CorrelationID != nil {
		e.CorrelationID = *row.CorrelationID
	}

	if err := e.Validate(); err != nil {
		return contracts.Envelope{}, eventbus.Permanent(
			fmt.Errorf("outbox: dòng id=%d không hợp lệ: %w", row.ID, err))
	}
	return e, nil
}

// truncate cắt s xuống tối đa max BYTE, không làm hỏng ký tự UTF-8.
func truncate(s string, max int) string {
	if max <= 0 || len(s) <= max {
		return s
	}
	// Cắt thẳng theo byte có thể chẻ đôi một ký tự tiếng Việt và tạo ra
	// chuỗi UTF-8 không hợp lệ. PostgreSQL TỪ CHỐI ghi chuỗi như vậy vào
	// cột TEXT ("invalid byte sequence for encoding UTF8") — nghĩa là cả
	// transaction rollback, và relay lặp lại vĩnh viễn cùng một lỗi mà
	// thông báo lỗi thì nói về encoding chứ không nói về Kafka.
	for max > 0 && !utf8.RuneStart(s[max]) {
		max--
	}
	return s[:max] + "…"
}

Vì sao cắt last_error lại quan trọng. Một lỗi từ franz-go khi cả cụm Kafka không trả lời có thể dài vài KB (kèm danh sách broker, mã lỗi, dấu vết thử lại). Nhân với 100 dòng mỗi lô, mỗi giây, trong một sự cố kéo dài 10 phút: khoảng 1 GB ghi vào bảng outbox, cho một thông tin mà 200 ký tự đầu đã nói hết. Sự cố Kafka lúc đó kéo thêm sự cố đầy đĩa.

8. Cấu hình: một `Config`, hai tiến trình

Đính chính bước 4

Bước 4 §4 khai JWTSecret với thẻ env:"JWT_SECRET,required" và kiểm độ dài ngay trong Load(). Lúc đó hệ thống chỉ có một binary nên không có gì sai.

Bây giờ có binary thứ hai, và cmd/relay không dùng JWT một chút nào — nhưng vẫn sẽ từ chối khởi động nếu thiếu JWT_SECRET. Đó là lỗi thiết kế: nó buộc bạn phải rải một bí mật vào một tiến trình không có nhu cầu dùng nó, tức là tăng bề mặt lộ bí mật để đổi lấy không gì cả.

Bài học tổng quát: "bắt buộc" là thuộc tính của tiến trình, không phải của struct cấu hình. Thẻ ,required không diễn đạt được điều đó.

Sửa internal/platform/config/config.go:

type Config struct {
	// Mọi tiến trình đều cần.
	DatabaseURL string `env:"DATABASE_URL,required"`
	LogLevel    string `env:"LOG_LEVEL" envDefault:"info"`

	// cmd/api.
	HTTPPort  int           `env:"HTTP_PORT" envDefault:"8000"`
	JWTSecret string        `env:"JWT_SECRET"`
	JWTTTL    time.Duration `env:"JWT_TTL" envDefault:"24h"`

	// cmd/relay.
	KafkaBrokers        []string      `env:"KAFKA_BROKERS" envSeparator:","`
	KafkaProduceTimeout time.Duration `env:"KAFKA_PRODUCE_TIMEOUT" envDefault:"10s"`
	RelayBatchSize      int           `env:"RELAY_BATCH_SIZE" envDefault:"100"`
	RelayPollInterval   time.Duration `env:"RELAY_POLL_INTERVAL" envDefault:"1s"`
	RelayMaxAttempts    int           `env:"RELAY_MAX_ATTEMPTS" envDefault:"10"`
}

const MinJWTSecretLen = 32

// Load đọc biến môi trường và kiểm những gì MỌI tiến trình đều cần.
func Load() (Config, error) {
	var cfg Config
	if err := env.Parse(&cfg); err != nil {
		return Config{}, fmt.Errorf("config: %w", err)
	}
	return cfg, nil
}

// ValidateAPI kiểm những gì riêng cmd/api cần.
func (c Config) ValidateAPI() error {
	if len(c.JWTSecret) < MinJWTSecretLen {
		return fmt.Errorf(
			"config: cmd/api cần JWT_SECRET dài ít nhất %d ký tự (đang có %d)",
			MinJWTSecretLen, len(c.JWTSecret))
	}
	if c.JWTTTL <= 0 {
		return errors.New("config: JWT_TTL phải lớn hơn 0")
	}
	if c.HTTPPort <= 0 || c.HTTPPort > 65535 {
		return fmt.Errorf("config: HTTP_PORT không hợp lệ: %d", c.HTTPPort)
	}
	return nil
}

// ValidateRelay kiểm những gì riêng cmd/relay cần.
func (c Config) ValidateRelay() error {
	if len(c.KafkaBrokers) == 0 {
		return errors.New("config: cmd/relay cần KAFKA_BROKERS (ví dụ localhost:9092)")
	}
	if c.RelayBatchSize <= 0 {
		return errors.New("config: RELAY_BATCH_SIZE phải lớn hơn 0")
	}
	if c.RelayPollInterval <= 0 {
		return errors.New("config: RELAY_POLL_INTERVAL phải lớn hơn 0")
	}
	if c.RelayMaxAttempts <= 0 {
		return errors.New("config: RELAY_MAX_ATTEMPTS phải lớn hơn 0")
	}
	return nil
}

Thêm dòng cfg.ValidateAPI() vào cmd/api/main.go ngay sau config.Load():

	cfg, err := config.Load()
	if err != nil {
		return err
	}
	if err := cfg.ValidateAPI(); err != nil {   // ← THÊM
		return err
	}

Bổ sung .env và .env.example:

# ── Relay ────────────────────────────────────────────────
KAFKA_PRODUCE_TIMEOUT=10s
RELAY_BATCH_SIZE=100
RELAY_POLL_INTERVAL=1s
RELAY_MAX_ATTEMPTS=10

JWT_SECRET vẫn không có giá trị mặc định — lý lẽ ở bước 4 §4.1 không đổi. Thay đổi duy nhất là ai kiểm nó: giờ là cmd/api, tiến trình thật sự dùng tới, thay vì mọi tiến trình có liên kết gói config.

9. `cmd/relay/main.go`

New-Item -ItemType Directory -Force cmd/relay

package main

import (
	"context"
	"log/slog"
	"os"
	"os/signal"
	"syscall"

	"github.com/chuongtd/community/internal/platform/config"
	"github.com/chuongtd/community/internal/platform/eventbus/kafka"
	"github.com/chuongtd/community/internal/platform/logger"
	"github.com/chuongtd/community/internal/platform/outbox"
	"github.com/chuongtd/community/internal/platform/postgres"
)

func main() {
	if err := run(); err != nil {
		slog.Error("relay dừng vì lỗi", "err", err)
		os.Exit(1)
	}
}

func run() error {
	cfg, err := config.Load()
	if err != nil {
		return err
	}
	if err := cfg.ValidateRelay(); err != nil {
		return err
	}

	log := logger.New(cfg.LogLevel)
	slog.SetDefault(log)

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	// Pool riêng, nhỏ. Relay chạy một luồng nên chỉ dùng một kết nối tại
	// một thời điểm; phần dư là chỗ cho câu truy vấn giám sát gõ tay.
	// Ngân sách kết nối: bước 2 §3.2.
	pgCfg := postgres.DefaultConfig(cfg.DatabaseURL, "community-relay")
	pgCfg.MaxConns = 5
	pgCfg.MinConns = 1

	pool, err := postgres.Connect(ctx, pgCfg)
	if err != nil {
		return err
	}
	defer pool.Close()

	pub, err := kafka.NewPublisher(kafka.Config{
		Brokers:        cfg.KafkaBrokers,
		ClientID:       "community-relay",
		ProduceTimeout: cfg.KafkaProduceTimeout,
	})
	if err != nil {
		return err
	}
	defer pub.Close()

	relay := outbox.NewRelay(
		postgres.NewTxManager(pool),
		pub,
		log,
		outbox.RelayConfig{
			BatchSize:    cfg.RelayBatchSize,
			PollInterval: cfg.RelayPollInterval,
			MaxAttempts:  cfg.RelayMaxAttempts,
		},
	)

	// Run tự trả về nil khi ctx bị huỷ, nên không cần khối shutdown riêng
	// như cmd/api: relay không giữ kết nối đến từ bên ngoài để mà đóng.
	return relay.Run(ctx)
}

Vì sao relay là binary riêng chứ không phải một goroutine trong cmd/api. Ba lý do, theo thứ tự quan trọng:

Vòng đời khác nhau. api scale theo lượng request; relay không được scale tuỳ tiện vì nhiều instance làm hỏng thứ tự phát (Phụ lục A). Nhét chung nghĩa là mỗi lần thêm pod API là thêm một relay, không ai để ý.

Bán kính sự cố. Relay treo vì Kafka không được phép kéo theo API. Tách tiến trình là cách rẻ nhất để đảm bảo điều đó.

Triển khai độc lập. Sửa logic retry không cần khởi động lại tầng phục vụ người dùng.

10. Lệnh tắt

Thêm vào Makefile:

relay:                    ## Chạy outbox relay
	go run ./cmd/relay

outbox:                   ## Xem nhanh tình trạng hàng chờ outbox
	@$(COMPOSE) exec -T postgres psql -U app -d community -c "\
		SELECT count(*) FILTER (WHERE published_at IS NULL AND dead_at IS NULL) AS pending, \
		       count(*) FILTER (WHERE published_at IS NOT NULL)                 AS published, \
		       count(*) FILTER (WHERE dead_at IS NOT NULL)                      AS dead \
		FROM outbox;"

Và vào dev.ps1 — thêm hai nhánh vào khối switch:

    "relay"  { go run ./cmd/relay }
    "outbox" {
        Invoke-Compose exec -T postgres psql -U app -d community -c @"
SELECT count(*) FILTER (WHERE published_at IS NULL AND dead_at IS NULL) AS pending,
       count(*) FILTER (WHERE published_at IS NOT NULL)                 AS published,
       count(*) FILTER (WHERE dead_at IS NOT NULL)                      AS dead
FROM outbox;
"@
    }

và cập nhật dòng default:

    default  { Write-Host "Dung: .\dev.ps1 [up|down|stop|logs|ps|topics|smoke|api|relay|outbox|reset]" }

11. Chạy thử đầu-cuối

Đây là phần kiểm chứng của bước này: đăng ký user → thấy event trên Kafka UI.

11.1 Khởi động hạ tầng

.\dev.ps1 up
.\dev.ps1 ps

Đợi postgres và kafka đều (healthy). Rồi đảm bảo topic tồn tại (nếu bạn từng chạy down -v thì topic đã mất cùng container):

.\dev.ps1 topics

Chạy migration mới:

migrate -path db/migrations -database "postgres://app:secret@localhost:5432/community?sslmode=disable" up
sqlc generate
go build ./...

11.2 Ba terminal

Terminal 1 — API:

.\dev.ps1 api

Terminal 2 — Relay:

.\dev.ps1 relay

{"time":"...","level":"INFO","msg":"relay bắt đầu","batch_size":100,"poll_interval":"1s","max_attempts":10}

Nếu bảng đang còn event từ bước 4 chưa gửi, bạn sẽ thấy nó bay đi ngay lập tức:

{"time":"...","level":"INFO","msg":"outbox: đã gửi","count":1,"topics":1}

Terminal 3 — gọi API:

$body = @{
    username = "hoangnh"
    name     = "Nguyễn Huy Hoàng"
    email    = "hoangnh@example.com"
    password = "matkhau-rat-dai-va-an-toan"
} | ConvertTo-Json

Invoke-RestMethod -Uri http://localhost:8000/auth/register `
    -Method Post `
    -ContentType "application/json; charset=utf-8" `
    -Body ([Text.Encoding]::UTF8.GetBytes($body)) `
    -Headers @{ "X-Request-ID" = "thu-nghiem-relay-001" }

[Text.Encoding]::UTF8.GetBytes($body) — bắt buộc, không phải trang trí. PowerShell 5.1 gửi chuỗi bằng ISO-8859-1, nên Nguyễn Huy Hoàng sẽ tới server thành NguyÃªn Huy HoÃ ng. Chi tiết ở bước 4 §16.2.

11.3 Relay phản ứng

Trong vòng 1 giây, terminal 2 in:

{"time":"...","level":"INFO","msg":"outbox: đã gửi","count":1,"topics":1}

11.4 Kiểm tra bảng đã được đánh dấu

docker compose -f deployments/docker-compose.yml exec postgres psql -U app -d community -c "SELECT id, event_type, topic, aggregate_id, attempts, published_at IS NOT NULL AS sent, dead_at FROM outbox ORDER BY id DESC LIMIT 5;"

 id |          event_type           |  topic   |             aggregate_id             | attempts | sent | dead_at
----+-------------------------------+----------+--------------------------------------+----------+------+---------
  1 | identity.user.registered.v1   | identity | 3f2a...-...-...                      |        0 | t    |

attempts = 0 và sent = t: đi thẳng, không phải thử lại lần nào.

11.5 ★ Thấy event trên Kafka UI

Mở http://localhost:8080 → Topics → identity → tab Messages.

Bạn sẽ thấy một message với:

Chỗ nhìn	Giá trị mong đợi
Key	UUID của user vừa đăng ký — bằng đúng `aggregate_id` trong bảng
Headers	`event_id`, `event_type = identity.user.registered.v1`, `correlation_id = thu-nghiem-relay-001`
Value	Toàn bộ envelope dạng JSON
Partition	0, 1 hoặc 2 — Kafka chọn theo băm của Key

Value:

{
  "event_id": "9c1e...",
  "event_type": "identity.user.registered.v1",
  "aggregate_id": "3f2a...",
  "occurred_at": "2026-08-10T09:31:22.418Z",
  "correlation_id": "thu-nghiem-relay-001",
  "payload": {
    "user_id": "3f2a...",
    "username": "hoangnh",
    "name": "Nguyễn Huy Hoàng",
    "email": "hoangnh@example.com",
    "registered_at": "2026-08-10T09:31:22.418Z"
  }
}

Ba điều đáng dừng lại một chút:

correlation_id bằng đúng X-Request-ID bạn gửi. Sợi chỉ đã xuyên qua HTTP → PostgreSQL → Kafka. Ở bước 6 nó sẽ đi tiếp vào log consumer, và lúc đó grep thu-nghiem-relay-001 cho bạn toàn bộ hành trình của một request.
Không có password_hash ở đâu cả. Đây là luật từ bước 4 §10.1 đang có hiệu lực, và giờ bạn thấy vì sao nó quan trọng: message này sẽ sống trong Kafka 7 ngày và trong log của mọi consumer lâu hơn thế.
Key không rỗng. Đó là §6.3 đang làm việc. Mọi event sau này của cùng user sẽ vào cùng partition, nên "đăng ký rồi đổi tên" không bao giờ tới consumer theo thứ tự ngược.

Không muốn mở trình duyệt thì dùng console consumer:

docker compose -f deployments/docker-compose.yml exec kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic identity --from-beginning --property print.key=true --property print.headers=true

12. Ba bài diễn tập sự cố

Mục 11 chứng minh đường đi thuận lợi. Ba bài dưới đây chứng minh những thứ Outbox Pattern sinh ra để giải quyết — và chúng đáng làm bằng tay ít nhất một lần, vì cảm giác "API vẫn 201 trong khi Kafka đã chết" khó truyền đạt bằng chữ.

12.1 Kafka sập giữa chừng

Đây là bài quan trọng nhất.

# Giữ nguyên terminal 1 (api) và 2 (relay) đang chạy.
docker compose -f deployments/docker-compose.yml stop kafka

Đăng ký một user khác (đổi username và email):

$body = @{ username="user2"; name="Người Thứ Hai"; email="user2@example.com"; password="matkhau-rat-dai-va-an-toan" } | ConvertTo-Json
Invoke-RestMethod -Uri http://localhost:8000/auth/register -Method Post `
    -ContentType "application/json; charset=utf-8" -Body ([Text.Encoding]::UTF8.GetBytes($body))

Quan sát 1 — API vẫn trả 201 Created với token bình thường. Người dùng không biết gì về sự cố. Đây chính là điều đánh đổi ở ARCHITECTURE.md §7.1 mua về: tầng ghi không phụ thuộc vào Kafka.

Quan sát 2 — relay ghi WARN và lùi lịch:

{"level":"WARN","msg":"outbox: lùi lịch thử lại","count":1,"err":"unable to dial: dial tcp ..."}

Quan sát 3 — theo dõi backoff nới rộng dần:

docker compose -f deployments/docker-compose.yml exec postgres psql -U app -d community -c "SELECT id, attempts, next_attempt_at - now() AS con_bao_lau, left(last_error, 60) FROM outbox WHERE published_at IS NULL AND dead_at IS NULL;"

Chạy vài lần, cách nhau ~10 giây. attempts tăng 1, 2, 3… và con_bao_lau giãn ra 1s → 2s → 4s → 8s. Relay đang tự giảm nhịp, không đập cửa Kafka 86.400 lần một ngày.

Bật lại:

docker compose -f deployments/docker-compose.yml start kafka

Sau khi Kafka healthy và tới lượt next_attempt_at, relay tự gửi được:

{"level":"INFO","msg":"outbox: đã gửi","count":1,"topics":1}

Không mất event nào, không cần ai can thiệp. Đây là toàn bộ lý do tồn tại của bảng outbox — nếu cmd/api gọi thẳng Kafka, event này đã bốc hơi vào lúc quan sát 1.

Lưu ý về restart: unless-stopped. Nếu relay chạy trong Docker (chưa phải bây giờ), docker compose stop kafka cũng có thể làm relay khởi động lại và thất bại ở Ping — hành vi đã giải thích ở §6.7. Chạy go run từ host như hướng dẫn này thì không gặp, vì relay đã qua Ping từ trước.

12.2 Event đi vào topic không tồn tại → chết sau N lần

Đặt MaxAttempts nhỏ để không phải chờ:

$env:RELAY_MAX_ATTEMPTS = "2"
go run ./cmd/relay

Ở terminal khác, làm hỏng một dòng có chủ đích:

docker compose -f deployments/docker-compose.yml exec postgres psql -U app -d community -c "INSERT INTO outbox (event_id, event_type, topic, aggregate_id, payload) VALUES (gen_random_uuid(), 'identity.user.registered.v1', 'topic-khong-ton-tai', 'test-agg', '{\"user_id\":\"test\"}');"

Relay thử, thất bại vì UNKNOWN_TOPIC_OR_PARTITION, thử lại, rồi bỏ cuộc:

{"level":"ERROR","msg":"outbox: event bỏ cuộc, cần xử lý thủ công","ids":[3],"count":1,"err":"UNKNOWN_TOPIC_OR_PARTITION: ..."}

Kiểm tra:

.\dev.ps1 outbox

 pending | published | dead
---------+-----------+------
       0 |         2 |    1

pending = 0 mới là điểm cần thấy. Dòng hỏng đã rời khỏi hàng chờ, nên chỉ số cảnh báo quan trọng nhất trở lại 0 và tiếp tục có ý nghĩa. Nếu không có dead_at, con số đó sẽ mắc kẹt ở 1 mãi mãi và bạn sẽ học cách phớt lờ nó — đó là lúc cảnh báo chết, chứ không phải lúc nó bị tắt.

12.3 Nhiều topic trong cùng một lô

Thêm một event giả vào topic post để thấy phần gom nhóm ở §7.5 làm việc:

docker compose -f deployments/docker-compose.yml exec postgres psql -U app -d community -c "INSERT INTO outbox (event_id, event_type, topic, aggregate_id, payload) VALUES (gen_random_uuid(), 'post.post.created.v1', 'post', 'bai-viet-1', '{\"post_id\":\"bai-viet-1\"}');"

Log relay:

{"level":"INFO","msg":"outbox: đã gửi","count":1,"topics":1}

Muốn thấy topics: 2 thì chèn hai dòng vào hai topic trong cùng một câu INSERT — chúng hiện ra cùng lúc nên relay nhặt cả hai trong một lô, và gọi PublishEach hai lần, mỗi topic một lần.

Nhớ dọn các dòng thử nghiệm:

docker compose -f deployments/docker-compose.yml exec postgres psql -U app -d community -c "DELETE FROM outbox WHERE aggregate_id IN ('test-agg','bai-viet-1');"

13. Vận hành

13.1 Bốn câu SQL nên thuộc

-- 1. Hàng chờ còn bao nhiêu.
--    Chỉ số cảnh báo số 1 (ARCHITECTURE.md §11.4). Cảnh báo khi > 1000.
SELECT count(*) FROM outbox WHERE published_at IS NULL AND dead_at IS NULL;

-- 2. Độ trễ thật: event cũ nhất chưa gửi đã chờ bao lâu.
--    Hữu ích hơn câu 1, vì 5000 dòng vừa mới ghi là bình thường,
--    còn 1 dòng chờ 40 phút là relay đã chết.
SELECT now() - min(occurred_at) AS do_tre
FROM outbox WHERE published_at IS NULL AND dead_at IS NULL;

-- 3. Có gì chết không, và vì sao.
SELECT id, event_type, topic, attempts, dead_at, left(last_error, 120) AS loi
FROM outbox WHERE dead_at IS NOT NULL ORDER BY dead_at DESC LIMIT 50;

-- 4. Đang bị lùi lịch — dấu hiệu Kafka chập chờn dù chưa có gì chết.
SELECT count(*) AS dang_cho, max(attempts) AS lan_thu_nhieu_nhat
FROM outbox
WHERE published_at IS NULL AND dead_at IS NULL AND next_attempt_at > now();

Câu 2 quan trọng hơn câu 1 và hay bị bỏ qua. Ngưỡng theo số lượng báo động giả mỗi khi có đợt ghi lớn, và im lặng khi hệ thống rảnh — mà relay chết lúc hệ thống rảnh thì cũng chết y như lúc bận. Ngưỡng theo tuổi thì không có hai nhược điểm đó: chờ quá 30 giây là có vấn đề, bất kể hàng chờ dài bao nhiêu.

13.2 Phát lại event đã chết

-- Xem đã, luôn luôn xem trước.
SELECT id, event_type, topic, attempts, last_error FROM outbox WHERE dead_at IS NOT NULL;

-- Hồi sinh: xoá dấu chết, reset bộ đếm, cho vào hàng chờ ngay.
UPDATE outbox
SET dead_at         = NULL,
    attempts        = 0,
    next_attempt_at = NULL,
    last_error      = NULL
WHERE id = ANY(ARRAY[3, 7, 11]);

Relay sẽ nhặt chúng ở lượt poll kế tiếp.

⚠️ Sửa nguyên nhân trước, hồi sinh sau

Câu UPDATE trên không sửa gì cả — nó chỉ đưa event trở lại hàng chờ. Nếu topic vẫn chưa tồn tại, hay message vẫn quá khổ, event sẽ chết lại đúng như cũ, chỉ tốn thêm MaxAttempts lần thử.

Thứ tự đúng: đọc last_error → sửa nguyên nhân (tạo topic, nâng max.message.bytes, sửa dữ liệu) → xác nhận đã sửa → rồi mới hồi sinh.

13.3 Dọn dẹp

-- Giữ 7 ngày để còn điều tra khi có sự cố (ARCHITECTURE.md §7.6).
DELETE FROM outbox
WHERE published_at IS NOT NULL AND published_at < now() - INTERVAL '7 days';

Không bao giờ xoá dòng dead_at IS NOT NULL bằng job tự động. Đó là những event chưa từng rời khỏi hệ thống; xoá chúng là mất dữ liệu thật, im lặng, không hoàn tác được. Dòng chết chỉ được xoá bởi một con người đã đọc và quyết định.

Job này thuộc bước 6 (cùng với job dọn consumed_events). Bây giờ chạy tay khi cần là đủ — bảng còn vài chục dòng.

13.4 Bao nhiêu relay?

Mặc định: một. Một relay xử lý khoảng 2.000 event/giây, và quan trọng hơn: nhiều relay làm hỏng thứ tự phát theo cách rất khó phát hiện. Đọc Phụ lục A trước khi chạy instance thứ hai.

Kiểm tra đang có mấy relay nối vào DB:

SELECT application_name, count(*)
FROM pg_stat_activity
WHERE datname = 'community' AND application_name = 'community-relay'
GROUP BY 1;

cài đặt kafka

Ghi chú về module path

Mục tiêu

Mục lục

1. Phạm vi bước này

1.1 Cái đã có sẵn

1.2 Cái sẽ viết

1.3 ⚠️ Bước này có thêm migration — và vì sao

2. Nhìn từ trên xuống

3. Polling, CDC hay LISTEN/NOTIFY?

4. Migration 000005 — hai cột cho việc thử lại

4.1 Vì sao next_attempt_at để NULL được thay vì NOT NULL DEFAULT now()

4.2 ⚠️ Vì sao dead letter ở đây là một bảng, không phải một topic

5. Cập nhật outbox/query.sql

6. eventbus/kafka — Publisher

6.1 Vấn đề chữ ký: relay cần kết quả của từng event

6.2 publisher.go

6.3 Envelope → kgo.Record

6.4 Vì sao lặp lại dữ liệu trong header

6.5 Phân loại lỗi Kafka

⚠️ Vì sao UNKNOWN_TOPIC_OR_PARTITION không nằm trong danh sách

6.6 Ghi chú: producer idempotent

6.7 Ghi chú: Ping lúc khởi động với một relay

7. outbox/relay.go

7.1 Khai báo phụ thuộc

7.2 Vòng lặp

Vì sao "lô đầy thì chạy tiếp ngay" không phải chi tiết vụn vặt

7.3 ⚠️ Vì sao SELECT và UPDATE phải cùng một transaction

7.4 ⚠️ Cái giá của việc gọi mạng trong transaction

⚠️ idle_in_transaction_session_timeout

7.5 Publish và đánh dấu

7.6 Hai hàm phụ

8. Cấu hình: một Config, hai tiến trình

Đính chính bước 4

9. cmd/relay/main.go

10. Lệnh tắt

11. Chạy thử đầu-cuối

11.1 Khởi động hạ tầng

11.2 Ba terminal

11.3 Relay phản ứng

11.4 Kiểm tra bảng đã được đánh dấu

11.5 ★ Thấy event trên Kafka UI

12. Ba bài diễn tập sự cố

12.1 Kafka sập giữa chừng

12.2 Event đi vào topic không tồn tại → chết sau N lần

12.3 Nhiều topic trong cùng một lô

13. Vận hành

13.1 Bốn câu SQL nên thuộc

13.2 Phát lại event đã chết

⚠️ Sửa nguyên nhân trước, hồi sinh sau

13.3 Dọn dẹp

13.4 Bao nhiêu relay?

Mục lục

3. Polling, CDC hay `LISTEN/NOTIFY`?

4. Migration `000005` — hai cột cho việc thử lại

4.1 Vì sao `next_attempt_at` để NULL được thay vì `NOT NULL DEFAULT now()`

5. Cập nhật `outbox/query.sql`

6. `eventbus/kafka` — Publisher

6.2 `publisher.go`

6.3 `Envelope` → `kgo.Record`

⚠️ Vì sao `UNKNOWN_TOPIC_OR_PARTITION` không nằm trong danh sách

6.7 Ghi chú: `Ping` lúc khởi động với một relay

7. `outbox/relay.go`

7.3 ⚠️ Vì sao `SELECT` và `UPDATE` phải cùng một transaction

⚠️ `idle_in_transaction_session_timeout`

8. Cấu hình: một `Config`, hai tiến trình

9. `cmd/relay/main.go`