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

14. Kiểm chứng bằng test

14.1 truncate — ranh giới UTF-8

internal/platform/outbox/relay_test.go:

package outbox

import (
	"strings"
	"testing"
	"unicode/utf8"

	"github.com/stretchr/testify/require"
)

func TestTruncateGiuUTF8HopLe(t *testing.T) {
	// "ễ" chiếm 3 byte. Cắt ở byte 1 hoặc 2 sẽ tạo ra chuỗi hỏng.
	s := strings.Repeat("Nguyễn", 100)

	for max := 1; max <= 60; max++ {
		got := truncate(s, max)
		require.True(t, utf8.ValidString(got),
			"cắt ở %d byte tạo ra chuỗi UTF-8 hỏng: %q", max, got)
	}
}

func TestTruncateKhongDungDenChuoiNgan(t *testing.T) {
	require.Equal(t, "ngắn", truncate("ngắn", 500))
}

Test này bảo vệ một lỗi rất cụ thể: chuỗi UTF-8 hỏng khiến INSERT thất bại, transaction rollback, và relay lặp lại vô hạn với một thông báo lỗi nói về encoding — trong khi vấn đề thật là Kafka. Xem §7.6.

14.2 Relay — Postgres thật, Kafka giả

Vì Publisher là interface do chính relay.go khai báo, ba test dưới đây chạy không cần Kafka.

// fakePublisher trả kết quả dựng sẵn theo chỉ số event.
type fakePublisher struct {
	calls  []string // các topic đã được gọi, theo thứ tự
	result func(topic string, i int) error
}

func (f *fakePublisher) PublishEach(_ context.Context, topic string, events []contracts.Envelope) []error {
	f.calls = append(f.calls, topic)
	errs := make([]error, len(events))
	for i := range events {
		if f.result != nil {
			errs[i] = f.result(topic, i)
		}
	}
	return errs
}

func TestGuiThanhCongThiDanhDauPublished(t *testing.T) {
	pool := testPool(t)          // helper từ bước 2 §8.3
	seedOutbox(t, pool, "identity", 3)

	pub := &fakePublisher{}
	r := NewRelay(postgres.NewTxManager(pool), pub, testLogger(), RelayConfig{BatchSize: 10})

	n, err := r.tick(context.Background())
	require.NoError(t, err)
	require.Equal(t, 3, n)

	pending, published, dead := counts(t, pool)
	require.Equal(t, 0, pending)
	require.Equal(t, 3, published)
	require.Equal(t, 0, dead)
}

func TestLoiTamThoiThiLuiLichChuKhongChet(t *testing.T) {
	pool := testPool(t)
	seedOutbox(t, pool, "identity", 1)

	pub := &fakePublisher{result: func(string, int) error {
		return errors.New("broker không trả lời")
	}}
	r := NewRelay(postgres.NewTxManager(pool), pub, testLogger(), RelayConfig{
		BatchSize: 10, MaxAttempts: 5,
	})

	_, err := r.tick(context.Background())
	require.NoError(t, err) // lỗi publish KHÔNG làm hỏng cả lô

	var attempts int
	var nextAt *time.Time
	require.NoError(t, pool.QueryRow(context.Background(),
		`SELECT attempts, next_attempt_at FROM outbox LIMIT 1`).Scan(&attempts, &nextAt))

	require.Equal(t, 1, attempts)
	require.NotNil(t, nextAt, "phải được lùi lịch")
	require.True(t, nextAt.After(time.Now()), "next_attempt_at phải ở tương lai")

	// Và lượt ngay sau đó KHÔNG nhặt lại nó nữa.
	n, err := r.tick(context.Background())
	require.NoError(t, err)
	require.Zero(t, n, "dòng đang bị lùi lịch không được nhặt lại ngay")
}

func TestLoiVinhVienThiChetNgayLanDau(t *testing.T) {
	pool := testPool(t)
	seedOutbox(t, pool, "identity", 1)

	pub := &fakePublisher{result: func(string, int) error {
		return eventbus.Permanent(errors.New("message quá lớn"))
	}}
	r := NewRelay(postgres.NewTxManager(pool), pub, testLogger(), RelayConfig{
		BatchSize: 10, MaxAttempts: 100, // dù cho phép 100 lần
	})

	_, err := r.tick(context.Background())
	require.NoError(t, err)

	_, _, dead := counts(t, pool)
	require.Equal(t, 1, dead, "lỗi vĩnh viễn không được thử lại lần nào")
}

func TestMotLoNhieuTopicThiGoiRiengTungTopic(t *testing.T) {
	pool := testPool(t)
	seedOutbox(t, pool, "identity", 2)
	seedOutbox(t, pool, "post", 3)

	pub := &fakePublisher{}
	r := NewRelay(postgres.NewTxManager(pool), pub, testLogger(), RelayConfig{BatchSize: 10})

	n, err := r.tick(context.Background())
	require.NoError(t, err)
	require.Equal(t, 5, n)

	require.ElementsMatch(t, []string{"identity", "post"}, pub.calls)

	_, published, _ := counts(t, pool)
	require.Equal(t, 5, published)
}

Test thứ hai là test có giá trị nhất trong nhóm. Nó khẳng định hai điều mà mắt thường không thấy khi đọc code: publish thất bại không làm cả lô rollback, và dòng vừa thất bại không bị nhặt lại ngay ở lượt kế tiếp. Thiếu tính chất thứ hai, relay sẽ quay vòng ở tốc độ tối đa quanh một event hỏng — CPU 100%, log ngập, và không ai hiểu vì sao.

14.3 Publisher — kiểm phần không cần broker

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

func TestKeyLaAggregateID(t *testing.T) {
	e, err := contracts.NewEnvelope("identity.user.registered.v1", "user-42", map[string]string{"a": "b"})
	require.NoError(t, err)

	rec, err := toRecord("identity", e)
	require.NoError(t, err)

	require.Equal(t, []byte("user-42"), rec.Key,
		"Key PHẢI là aggregate_id, nếu không thứ tự trong partition sẽ hỏng")
}

func TestEnvelopeHongThiLoiVinhVien(t *testing.T) {
	_, err := toRecord("identity", contracts.Envelope{}) // thiếu mọi trường
	require.Error(t, err)
	require.True(t, eventbus.IsPermanent(err),
		"envelope hỏng thử lại bao nhiêu lần cũng thế")
}

func TestPhanLoaiLoi(t *testing.T) {
	require.True(t, eventbus.IsPermanent(classify(kerr.MessageTooLarge)))
	require.False(t, eventbus.IsPermanent(classify(kerr.UnknownTopicOrPartition)),
		"topic có thể được tạo sau — xem §6.5")
	require.Nil(t, classify(nil))
}

14.4 Chạy

go test ./internal/... -race -count=1

-race đáng giá ở đây: relay là code chạy nền có timer, và nếu sau này bạn thêm goroutine để publish song song, race detector sẽ là thứ đầu tiên phát hiện ra sai lầm.

15. Những thứ cố ý chưa có

16. Checklist hoàn thành

• [ ] migrate ... up chạy sạch; \d outbox có next_attempt_at và dead_at
• [ ] \d outbox có idx_outbox_pending và idx_outbox_dead; không còn idx_outbox_unpublished
• [ ] sqlc generate sinh ra FetchPending, ScheduleRetry, MarkDead, CountPending
• [ ] go build ./... sạch, go vet ./... sạch
• [ ] go test ./internal/... -race -count=1 xanh
• [ ] .\dev.ps1 relay khởi động và in relay bắt đầu
• [ ] Đăng ký một user → relay in outbox: đã gửi trong vòng 1 giây
• [ ] SELECT published_at FROM outbox — không còn NULL
• [ ] Kafka UI (localhost:8080) → topic identity → tab Messages: thấy event
• [ ] Message có Key = user id và header event_type
• [ ] correlation_id trong message bằng X-Request-ID đã gửi
• [ ] Payload không chứa password_hash
• [ ] Diễn tập §12.1: dừng Kafka → API vẫn 201 → bật lại → event tự đi
• [ ] Diễn tập §12.2: event hỏng chuyển sang dead, pending về 0

Cấu trúc thư mục lúc này:

project/
├── cmd/
│   ├── api/main.go
│   ├── relay/main.go               ← MỚI
│   └── smoketest/main.go
├── db/migrations/
│   ├── 000001_create_users.*.sql
│   ├── 000002_create_user_stats.*.sql
│   ├── 000003_create_outbox.*.sql
│   ├── 000004_create_consumed_events.*.sql
│   └── 000005_add_outbox_retry.*.sql   ← MỚI
├── internal/
│   ├── contracts/
│   ├── modules/identity/
│   └── platform/
│       ├── config/                 ← sửa: ValidateAPI / ValidateRelay
│       ├── eventbus/
│       │   ├── bus.go
│       │   ├── inmem/
│       │   └── kafka/publisher.go  ← MỚI
│       └── outbox/
│           ├── query.sql           ← sửa
│           ├── writer.go
│           ├── relay.go            ← MỚI
│           └── gen/
└── ...

Commit:

git add .
git commit -m "feat(outbox): relay đẩy event sang Kafka, có backoff và dead letter"

Phụ lục A — Chạy nhiều relay mà không hỏng thứ tự

FOR UPDATE SKIP LOCKED khiến nhiều relay chia nhau công việc mà không gửi trùng. Nhưng nó không giữ được thứ tự phát, và đây là điều gần như không tài liệu nào nói rõ.

Relay A: SELECT → nhận id 1..100   (trong đó id=7   là user X)
Relay B: SELECT → nhận id 101..200 (trong đó id=150 là user X)

B chạy nhanh hơn, COMMIT trước.
→ Kafka nhận id=150 TRƯỚC id=7.
→ Cùng Key (user X) nên cùng partition.
→ Consumer thấy "user X bị xoá" trước "user X đổi tên".

Với một relay, ORDER BY id bảo đảm thứ tự toàn cục và tình huống trên không xảy ra được.

Nếu thật sự cần nhiều instance, chia theo aggregate thay vì để chúng bốc ngẫu nhiên. Mỗi relay chỉ lấy phần của mình:

-- Relay thứ i trong N instance. Truyền i, N vào từ biến môi trường.
SELECT ...
FROM outbox
WHERE published_at IS NULL
  AND dead_at IS NULL
  AND (next_attempt_at IS NULL OR next_attempt_at <= now())
  AND abs(hashtext(aggregate_id)) % @shard_count = @shard_index
ORDER BY id
LIMIT $1
FOR UPDATE SKIP LOCKED;

Vì mọi event của một aggregate_id luôn thuộc về đúng một shard, thứ tự trong phạm vi một thực thể được giữ nguyên — và đó là thứ tự duy nhất ARCHITECTURE.md §5.2 hứa hẹn.

Cái giá: shard_count trở thành hằng số vận hành. Đổi từ 2 lên 3 phải dừng hết relay rồi bật lại đồng loạt; nếu hai instance chạy với shard_count khác nhau, có aggregate không ai nhận và có aggregate hai người cùng nhận.

Khuyến nghị: chạy một relay cho tới khi có số đo chứng minh nó là nút thắt. Một relay là thiết kế đơn giản hơn và bảo đảm mạnh hơn.

Phụ lục B — Phương án "lease" khi transaction quá dài

§7.4 giữ transaction mở suốt lời gọi Kafka. Nếu Postgres của bạn có idle_in_transaction_session_timeout thấp và không đổi được, hoặc DBA phản đối transaction dài, đây là phương án thay thế.

Ý tưởng: thay khoá bằng "thuê" — đẩy next_attempt_at về tương lai để giành quyền xử lý, rồi commit ngay.

-- name: ClaimPending :many
WITH claimed AS (
    SELECT id 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
)
UPDATE outbox o
-- Giữ chỗ 30 giây: relay khác sẽ không thấy các dòng này.
SET next_attempt_at = now() + INTERVAL '30 seconds'
FROM claimed c
WHERE o.id = c.id
RETURNING o.id, o.event_id, o.event_type, o.topic, o.aggregate_id,
          o.correlation_id, o.payload, o.occurred_at, o.attempts;

Vòng đời đổi thành: ClaimPending (transaction ngắn, chỉ một câu lệnh) → COMMIT → publish ngoài transaction → MarkPublished (transaction ngắn thứ hai).

Cả hai đều đúng dưới giả định at-least-once; bản lease chỉ đơn giản là trùng nhiều hơn. Đó là lý do bản chính được chọn làm mặc định — và cũng là lý do idempotency ở bước 6 không phải tuỳ chọn.

Bước tiếp theo

Bước 6 — Consumer và Idempotency: cài eventbus/kafka/consumer.go (consumer group, commit offset, retry, DLQ topic), viết platform/idempotency/guard.go dựa trên bảng consumed_events từ bước 3 §6, và cho module stats nghe identity.user.registered.v1 để tự tạo dòng user_stats của mình — lời hứa còn nợ từ bước 4 §18.

Đó cũng là bước lời cảnh báo lặp đi lặp lại trong tài liệu này — "hệ thống là at-least-once, consumer bắt buộc phải idempotent" — chuyển từ một câu chữ thành một ràng buộc khoá chính trong database.