en
new
0
Avatar
Thư Viện Kiến Thức @hhoang
12.3K 152 1.0K
Published Thursday, 4:47 p.m. 20 min read
23

#9 Bước 6 — Consumer, Idempotency và DLQ Phần 1

Bước 6 — Consumer, Idempotency và DLQ

Tài liệu thực hành. Kết thúc bước này, vòng tròn event-driven đóng lại: đăng ký user ở cmd/api → event qua cmd/relay lên Kafka → cmd/worker nghe được và tự tạo dòng user_stats → GET /users/{id}/stats trả về số liệu.

Kiến trúc tổng thể: ARCHITECTURE.md — bước này là hiện thực của §8 Idempotency và §11 Xử lý lỗi, retry và DLQ

Vì sao bước này là bước 6

Bạn nói chưa biết hướng tiếp theo. Thực ra năm bài trước đã chỉ vào đúng một chỗ — có ba món nợ đang treo, và cả ba cùng được trả bằng một việc:

Món nợ Ghi ở
"Hệ thống là at-least-once, consumer bắt buộc phải idempotent" — nhắc 4 lần, chưa có dòng code nào Bước 5 §6.6, ARCHITECTURE §7.5
"stats tự tạo dòng của mình khi nghe identity.user.registered.v1" — đã quyết, chưa làm Bước 4 §18.2
"Bảng consumed_events và query EnsureUserStats chỉ có nghĩa khi có consumer thật" Bước 3 §13

Ngoài ra còn một lý do mạnh hơn: hiện tại hệ thống chưa chứng minh được gì cả. Bước 5 đưa event lên Kafka, nhưng không ai đọc. Một hệ event-driven mà không có consumer thì chỉ là một hệ ghi log sang chỗ khác. Bước này là bước đầu tiên event gây ra hệ quả.

Sau bước 6, xương sống kiến trúc hoàn chỉnh và mọi việc còn lại (module post, social, feed, search) chỉ là lặp lại đúng mẫu này — thêm module, thêm handler, không sửa hạ tầng. Đó là lúc bạn có thể nói kiến trúc đã xong.

Mục tiêu

  • platform/idempotency/guard.go — chống xử lý trùng bằng bảng consumed_events
  • eventbus/kafka/consumer.go — consumer group, commit thủ công, retry, DLQ
  • modules/stats hoàn chỉnh: repository, subscriber, endpoint đọc
  • cmd/worker — binary thứ ba
  • platform/cleanup — job dọn outbox và consumed_events
  • Ba bài diễn tập: giao trùng, message độc, worker chết giữa lô

Không thêm migration nào. Bảng consumed_events (000004) và user_stats (000002) đã dựng ở bước 3, và bước này không cần thêm cột nào — khác với bước 5, nơi tôi phải thêm hai cột vì lược đồ thiếu thật. Kiểm tra một lần cho chắc:

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

Mục lục

  1. Ba câu hỏi phải trả lời trước khi viết consumer
  2. Nhìn từ trên xuống
  3. platform/idempotency
  4. eventbus/kafka/consumer.go
  5. modules/stats
  6. cmd/worker/main.go
  7. platform/cleanup — trả nợ hai job dọn dẹp
  8. Chạy thử đầu-cuối
  9. Ba bài diễn tập sự cố
  10. Kiểm chứng bằng test
  11. Vận hành
  12. Những thứ cố ý chưa có
  13. Checklist hoàn thành

1. Ba câu hỏi phải trả lời trước khi viết consumer

Phần này không có code. Nó là ba khái niệm mà nếu hiểu sai, consumer của bạn sẽ mất dữ liệu trong im lặng — và đây là loại sai lầm mà test trên máy dev không bao giờ phát hiện.

1.1 Offset là mốc nước, không phải phiếu xác nhận từng message

Trực giác tự nhiên: "xử lý xong message nào thì báo cho Kafka biết message đó xong." Kafka không hoạt động như vậy.

Offset commit là một con số cho mỗi partition, mang nghĩa: "tôi đã xong mọi thứ trước vị trí này." Không có cách nào nói "xong 5 và 7, chưa xong 6".

partition 0:  [4] [5] [6] [7] [8] ...
                       ↑
              commit offset = 6
              ⇒ "4 và 5 xong rồi, đọc lại từ 6"

Hệ quả trực tiếp và rất dễ mắc:

// ❌ SAI — xử lý xong cái nào commit cái đó, bỏ qua cái lỗi
for _, rec := range records {
    if err := handle(rec); err != nil {
        log.Error("bỏ qua", "err", err)   // ← "chỉ một message thôi mà"
        continue
    }
    commit(rec)
}

Giả sử message 5 lỗi, 6 và 7 thành công. Bạn commit 7 (thực chất là 8). Message 5 biến mất vĩnh viễn — nó nằm dưới mốc nước, Kafka sẽ không bao giờ giao lại. Dòng log.Error là dấu vết duy nhất còn lại, và nó nằm lẫn trong vài triệu dòng log khác.

Quy tắc: chỉ commit offset khi mọi message trước nó đã được xử lý hoặc đã được đưa vào DLQ. "Đã đưa vào DLQ" cũng là một kết cục hợp lệ — nó có nghĩa là message đã rời khỏi luồng chính một cách có ghi nhận, không phải bị bỏ rơi.

1.2 ⚠️ Auto-commit là at-most-once đội lốt

Đây là bug mất dữ liệu phổ biến nhất của hệ Kafka, và nó là mặc định trong hầu hết client.

Auto-commit commit offset theo đồng hồ (mặc định 5 giây), không quan tâm handler của bạn đã làm gì:

t=0.0s   poll → nhận message 100..200
t=0.1s   handler bắt đầu xử lý 100
t=5.0s   ⏰ auto-commit → ghi offset 200   ← đã commit CẢ 100..200
t=5.1s   handler xử lý tới 150 thì process bị kill
         ⇒ khởi động lại: đọc từ 200
         ⇒ message 150..200 KHÔNG BAO GIỜ được xử lý

Không có lỗi, không có log, không có cảnh báo. Bạn phát hiện khi có người thắc mắc vì sao số liệu của họ thiếu.

Nên trong consumer.go có dòng này, và nó không phải tuỳ chọn:

kgo.DisableAutoCommit(),

Đổi lại, ta tự commit sau khi xử lý xong — nghĩa là at-least-once: chết trước lúc commit thì message được giao lại. Đó là lý do mục 3 tồn tại.

Ba lựa chọn, không có lựa chọn thứ tư:

Cơ chế Hậu quả
At-most-once Commit trước khi xử lý Chết giữa chừng → mất message
At-least-once Commit sau khi xử lý Chết giữa chừng → trùng message
Exactly-once Transaction Kafka + Kafka Chỉ đúng khi cả nguồn và đích đều là Kafka. Đích của ta là PostgreSQL, nên không dùng được

Ta chọn at-least-once và khử trùng ở phía đích. Đó là kiến trúc duy nhất đúng khi đích là database — và cũng là lý do bảng consumed_events được thiết kế từ bước 3.

1.3 ⚠️ "Thử lại một message" nghĩa là chặn cả partition

Trong hàng đợi truyền thống (RabbitMQ, SQS), message là độc lập: một message lỗi được đẩy lại vào hàng, các message khác chảy tiếp.

Kafka không phải hàng đợi. Partition là một sổ ghi có thứ tự, đọc bằng một con trỏ. Không có "đẩy lại vào hàng". Muốn thử lại message 5 thì con trỏ phải đứng ở 5 — nghĩa là 6, 7, 8 đứng chờ theo.

Ba lựa chọn, và mỗi cái đánh đổi một thứ khác nhau:

Cách Giữ thứ tự Giữ thông lượng Dùng khi
Thử lại tại chỗ, ngân sách nhỏ (bước này) ✅ ✅ nếu lỗi hiếm Lỗi tạm thời ngắn: DB nấc, deadlock
Chuyển sang retry topic ❌ ✅ Lỗi cần chờ lâu: dịch vụ ngoài sập (Phụ lục B)
Chặn tới khi thành công ✅ ❌ tắc hoàn toàn Không bao giờ

Cách thứ ba là poison message kinh điển: một message hỏng chặn vĩnh viễn mọi message phía sau trong partition đó. Consumer vẫn "chạy", CPU vẫn quay, log vẫn có dòng — nhưng lag tăng đều và không gì được xử lý. Đây là sự cố production hay gặp nhất của hệ Kafka.

Bước này chọn cách thứ nhất với ngân sách retry rất chặt (~700ms), và bất kỳ thứ gì vượt ngân sách đó đi thẳng vào DLQ. Lý do "rất chặt" nằm ở §4.5 — nó không phải chuyện thẩm mỹ mà là ràng buộc cứng của consumer group.

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

┌─ cmd/api ─────────────────────────────────────────────────────────┐
│  POST /auth/register → INSERT users + INSERT outbox  (1 tx)       │
│  GET  /users/{id}/stats → SELECT user_stats     ← MỚI ở bước này  │
└───────────────────────────────────────────────────────────────────┘
                     │ bảng outbox
┌─ cmd/relay ───────────────────────────────────────────────────────┐
│  poll outbox → ProduceSync → Kafka topic "identity"               │
└───────────────────────────────────────────────────────────────────┘
                     │ Kafka
┌─ cmd/worker (MỚI) ────────────────────────────────────────────────┐
│  PollRecords(topic "identity", group "stats-service")             │
│    │                                                              │
│    ├─ ① giải mã Value → contracts.Envelope                        │
│    │      hỏng → DLQ ngay (retry vô ích)                          │
│    │                                                              │
│    ├─ ② Route theo EventType                                      │
│    │      không khớp → BỎ QUA, coi như xong (bước 2 §6.3)          │
│    │                                                              │
│    ├─ ③ stats.OnUserRegistered                                    │
│    │      └─ txm.Do:  guard.Mark(event_id, "stats-service")       │
│    │                  ├─ trùng → thoát êm, không làm gì           │
│    │                  └─ mới  → EnsureUserStats(user_id)          │
│    │                 COMMIT          ← ★ MỘT transaction          │
│    │                                                              │
│    ├─ ④ lỗi tạm thời → thử lại 100ms, 200ms, 400ms                │
│    │      vẫn lỗi / lỗi vĩnh viễn → identity.DLQ                  │
│    │                                                              │
│    └─ ⑤ CommitRecords  ← chỉ sau khi ③ hoặc ④ đã xong             │
│                                                                   │
│  + ticker mỗi 1h: dọn outbox > 7 ngày, consumed_events > 30 ngày  │
└───────────────────────────────────────────────────────────────────┘

Ba chi tiết quyết định tính đúng đắn:

  1. guard.Mark và việc thật phải cùng một transaction (§3). Tách ra là tự tạo lại đúng cái bug mà guard sinh ra để ngăn.
  2. Commit offset là bước cuối cùng (§1.1). Commit sớm một nhịp = mất dữ liệu.
  3. DLQ phải publish xong trước khi commit (§4.6). Ngược lại = message rơi vào khoảng trống giữa hai hệ thống.

3. platform/idempotency

3.1 guard.go

Query MarkConsumed đã sinh từ bước 3 §9.2. Việc còn lại là bọc nó lại cho chỗ gọi đọc được.

internal/platform/idempotency/guard.go:

// Package idempotency chống việc xử lý cùng một event nhiều lần.
//
// Kafka giao at-least-once và relay có retry, nên event SẼ bị giao lại
// vào một ngày nào đó (ARCHITECTURE.md §8).
package idempotency

import (
	"context"
	"fmt"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"

	"github.com/chuongtd/community/internal/platform/idempotency/gen"
)

// Guard chưa có trường nào. Giữ là struct để sau này thêm metric hay
// logger mà không phải sửa chữ ký ở mọi chỗ gọi.
type Guard struct{}

func NewGuard() *Guard { return &Guard{} }

// Mark ghi nhận rằng consumer đã xử lý event này.
//
// Trả về true  → đây là lần đầu, HÃY làm việc thật.
// Trả về false → đã xử lý rồi, hãy thoát êm.
//
// Tham số tx là BẮT BUỘC, và đó là chủ đích: việc thật phải nằm trong
// cùng transaction với lời gọi này (xem §3.2). Không có transaction thì
// không gọi được hàm này — cùng cách ép Luật 3 mà outbox.Writer dùng.
func (g *Guard) Mark(ctx context.Context, tx pgx.Tx, eventID, consumer string) (bool, error) {
	if consumer == "" {
		return false, fmt.Errorf("idempotency: thiếu tên consumer")
	}

	// Cột event_id kiểu UUID; Envelope giữ nó dạng chuỗi.
	id, err := uuid.Parse(eventID)
	if err != nil {
		return false, fmt.Errorf("idempotency: event_id không phải UUID (%q): %w", eventID, err)
	}

	rows, err := gen.New(tx).MarkConsumed(ctx, gen.MarkConsumedParams{
		EventID:  id,
		Consumer: consumer,
	})
	if err != nil {
		return false, fmt.Errorf("idempotency: ghi consumed_events: %w", err)
	}

	// MarkConsumed là INSERT ... ON CONFLICT DO NOTHING (:execrows).
	// 1 dòng bị ảnh hưởng = chèn mới = lần đầu. 0 dòng = đã có sẵn.
	return rows == 1, nil
}

3.2 Vì sao tx là tham số, không phải pool

Đây là toàn bộ giá trị của guard, và nó rất dễ làm sai theo cách trông không sai.

// ❌ SAI — hai transaction riêng
ok, _ := guard.Mark(ctx, pool, eventID, consumer)   // COMMIT ở đây
if !ok { return nil }
repo.EnsureUserStats(ctx, pool, userID)             // ← process chết TRƯỚC dòng này

Nếu process chết ở đúng khoảng giữa: event đã được đánh dấu "đã xử lý", nhưng công việc chưa làm. Kafka giao lại, guard trả false, handler bỏ qua. Dòng user_stats không bao giờ được tạo.

Bạn vừa dùng một cơ chế chống-trùng để tạo ra một vụ mất dữ liệu. Và nó không để lại dấu vết nào: log sẽ ghi "bỏ qua event trùng", đúng như thiết kế.

Cùng một transaction thì hai khả năng duy nhất là cả hai xảy ra hoặc không cái nào xảy ra. Trường hợp thứ hai dẫn tới giao lại, và giao lại là chuyện hệ thống này đã tính sẵn.

3.3 ⚠️ Guard bảo vệ ghi database, không bảo vệ tác dụng phụ

Giới hạn này cần nói rõ vì nó sẽ cắn bạn ở handler thứ ba hoặc thứ tư.

return s.tx.Do(ctx, func(tx pgx.Tx) error {
    ok, err := s.guard.Mark(ctx, tx, e.EventID, ConsumerName)
    if err != nil || !ok {
        return err
    }
    if err := s.repo.EnsureUserStats(ctx, tx, userID); err != nil {
        return err
    }
    return s.mailer.SendWelcome(ctx, p.Email)   // ⚠️ KHÔNG rollback được
})

Email đã gửi thì transaction rollback không gọi nó về. Nếu COMMIT thất bại sau đó, event được giao lại → guard cho qua (vì lần trước rollback) → người dùng nhận email thứ hai.

Ba cách xử lý, theo thứ tự nên ưu tiên:

Cách Làm thế nào
Đẩy tác dụng phụ ra event riêng Handler chỉ ghi DB + ghi outbox notification.welcome.requested.v1. Một consumer khác gửi mail, và nó tự lo idempotency của mình. Đây là cách đúng nhất và cũng chính là Outbox Pattern áp dụng lần thứ hai
Gọi API có khoá idempotency Phần lớn dịch vụ mail/payment nhận Idempotency-Key; truyền event_id vào đó
Chấp nhận Với việc vô hại (làm nóng cache) thì gửi trùng không sao — nhưng phải là quyết định có ghi lại, không phải chuyện bỏ sót

Quy tắc: trong tx.Do chỉ được làm những việc mà ROLLBACK hoàn tác được.

3.4 ⚠️ Đổi tên consumer group = xử lý lại toàn bộ

Cột consumer trong consumed_events là tên consumer group. Đổi tên nó có hai hậu quả xảy ra đồng thời, và chúng cộng dồn:

  1. Kafka coi đây là group mới → không có offset đã lưu → đọc lại từ đầu topic (theo ConsumeResetOffset)
  2. Guard cũng coi đây là consumer mới → mọi event trong 7 ngày retention được xử lý lại như thể lần đầu

Với EnsureUserStats thì vô hại (ON CONFLICT DO NOTHING). Với posts_count = posts_count + 1 thì mọi counter bị cộng thêm một lần nữa.

Nên tên group được khai là hằng số ở một chỗ duy nhất, kèm cảnh báo:

// internal/modules/stats/subscriber/subscriber.go

// ConsumerName là tên consumer group của module stats.
//
// ⚠️ KHÔNG ĐỔI giá trị này. Nó vừa là group id của Kafka, vừa là khoá
// trong bảng consumed_events. Đổi nó = đọc lại toàn bộ topic VÀ mất
// toàn bộ lịch sử chống trùng cùng lúc (xem bước 6 §3.4).
const ConsumerName = "stats-service"

Nếu thật sự cần đổi (ví dụ tách module), quy trình an toàn là sao chép lịch sử sang tên mới trước khi chạy:

INSERT INTO consumed_events (event_id, consumer, processed_at)
SELECT event_id, 'stats-service-v2', processed_at
FROM consumed_events WHERE consumer = 'stats-service'
ON CONFLICT DO NOTHING;

4. eventbus/kafka/consumer.go

4.1 Một client cho mỗi consumer group

Interface eventbus.Subscriber từ bước 2 §6 cho phép đăng ký nhiều (topic, group):

Subscribe(topic, group string, h Handler)

Nhưng một kgo.Client chỉ thuộc về một consumer group — group id là tham số lúc tạo client, không đổi được. Đây không phải hạn chế của franz-go mà là bản chất giao thức: heartbeat, rebalance, offset đều gắn với một group.

Nên Consumer giữ một client cho mỗi group, và Run chạy một goroutine cho mỗi client:

Consumer
├── group "stats-service"  → kgo.Client #1 → topics [identity]      → goroutine 1
└── group "search-indexer" → kgo.Client #2 → topics [identity,post] → goroutine 2

Hai group nhận bản sao độc lập của mọi message, có offset riêng, rebalance riêng. Đó chính là điều ARCHITECTURE.md §6.3 hứa: thêm consumer mới không ảnh hưởng consumer cũ.

4.2 Cấu hình và khởi tạo

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

package kafka

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

	"github.com/twmb/franz-go/pkg/kgo"
	"golang.org/x/sync/errgroup"

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

type ConsumerConfig struct {
	Brokers  []string
	ClientID string

	// MaxAttempts là số lần gọi handler cho MỘT message trước khi bỏ vào DLQ.
	MaxAttempts int

	// RetryBase là khoảng chờ của lần thử lại đầu tiên; các lần sau nhân đôi.
	// Tổng thời gian chờ phải nhỏ hơn RebalanceTimeout rất nhiều — xem §4.5.
	RetryBase time.Duration

	// PollMax là số record tối đa mỗi lượt poll.
	PollMax int

	// RebalanceTimeout là thời gian tối đa một thành viên được phép "im lặng"
	// trước khi group đá nó ra.
	RebalanceTimeout time.Duration
}

func (c *ConsumerConfig) setDefaults() {
	if c.MaxAttempts <= 0 {
		c.MaxAttempts = 4
	}
	if c.RetryBase <= 0 {
		c.RetryBase = 100 * time.Millisecond
	}
	if c.PollMax <= 0 {
		c.PollMax = 500
	}
	if c.RebalanceTimeout <= 0 {
		c.RebalanceTimeout = 60 * time.Second
	}
	if c.ClientID == "" {
		c.ClientID = "community-worker"
	}
}

type subscription struct {
	topic   string
	handler eventbus.Handler
}

// Consumer cài eventbus.Subscriber bằng franz-go.
type Consumer struct {
	cfg  ConsumerConfig
	dlq  *Publisher // dùng lại Publisher của bước 5 để đẩy vào DLQ
	log  *slog.Logger
	subs map[string][]subscription // group → các đăng ký
}

var _ eventbus.Subscriber = (*Consumer)(nil)

func NewConsumer(cfg ConsumerConfig, dlq *Publisher, log *slog.Logger) *Consumer {
	cfg.setDefaults()
	return &Consumer{
		cfg:  cfg,
		dlq:  dlq,
		log:  log,
		subs: make(map[string][]subscription),
	}
}

// Subscribe đăng ký handler. Phải gọi TRƯỚC Run.
func (c *Consumer) Subscribe(topic, group string, h eventbus.Handler) {
	c.subs[group] = append(c.subs[group], subscription{topic: topic, handler: h})
}

Vì sao dùng lại *Publisher của bước 5 cho DLQ thay vì tạo client producer riêng: nó đã có acks=all, đã có timeout, đã Ping lúc khởi động. DLQ là nơi ít được phép mất dữ liệu nhất trong toàn hệ thống — message vào đó là message đang có vấn đề, mất nó là mất luôn manh mối. Không có lý do gì để nó chạy với cấu hình yếu hơn đường chính.

4.3 Vòng lặp

// Run chạy tới khi ctx bị huỷ. Chặn luồng gọi.
func (c *Consumer) Run(ctx context.Context) error {
	if len(c.subs) == 0 {
		return errors.New("kafka: chưa có đăng ký nào, gọi Subscribe trước Run")
	}

	// Sắp xếp để thứ tự khởi động tất định — chỉ để log dễ đọc.
	groups := make([]string, 0, len(c.subs))
	for g := range c.subs {
		groups = append(groups, g)
	}
	sort.Strings(groups)

	g, ctx := errgroup.WithContext(ctx)
	for _, group := range groups {
		group := group
		g.Go(func() error { return c.runGroup(ctx, group) })
	}
	return g.Wait()
}

func (c *Consumer) runGroup(ctx context.Context, group string) error {
	subs := c.subs[group]

	// Gom handler theo topic. Nhiều handler cùng topic thì gọi lần lượt.
	routes := make(map[string][]eventbus.Handler)
	topics := make([]string, 0, len(subs))
	for _, s := range subs {
		if _, seen := routes[s.topic]; !seen {
			topics = append(topics, s.topic)
		}
		routes[s.topic] = append(routes[s.topic], s.handler)
	}

	cl, err := kgo.NewClient(
		kgo.SeedBrokers(c.cfg.Brokers...),
		kgo.ClientID(c.cfg.ClientID+"/"+group),
		kgo.ConsumerGroup(group),
		kgo.ConsumeTopics(topics...),

		// Group MỚI thì đọc từ đầu topic. Xem ghi chú §4.4.
		kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()),

		// ★ Không thể thương lượng. Xem §1.2.
		kgo.DisableAutoCommit(),

		// ★ Không thể thương lượng. Xem §4.5.
		kgo.BlockRebalanceOnPoll(),

		kgo.RebalanceTimeout(c.cfg.RebalanceTimeout),
		kgo.FetchMaxWait(time.Second),

		// Trước khi nhả partition, commit những gì đã xử lý xong. Không có
		// bước này, mọi message đã xử lý từ lần commit cuối sẽ được giao lại
		// cho thành viên mới — đúng, nhưng lãng phí.
		kgo.OnPartitionsRevoked(func(ctx context.Context, cl *kgo.Client, parts map[string][]int32) {
			c.log.Info("kafka: nhả partition", "group", group, "parts", parts)
			if err := cl.CommitUncommittedOffsets(ctx); err != nil {
				c.log.Error("kafka: commit trước khi nhả thất bại", "group", group, "err", err)
			}
		}),
		kgo.OnPartitionsAssigned(func(_ context.Context, _ *kgo.Client, parts map[string][]int32) {
			c.log.Info("kafka: nhận partition", "group", group, "parts", parts)
		}),
	)
	if err != nil {
		return fmt.Errorf("kafka: tạo consumer %s: %w", group, err)
	}
	// Bản Close này nhả partition tử tế thay vì để group phải chờ hết
	// SessionTimeout mới phát hiện ra ta đã đi. Bắt buộc dùng khi đã bật
	// BlockRebalanceOnPoll.
	defer cl.CloseAllowingRebalance()

	c.log.Info("kafka: consumer bắt đầu", "group", group, "topics", topics)

	for {
		fetches := cl.PollRecords(ctx, c.cfg.PollMax)

		if fetches.IsClientClosed() {
			return nil
		}
		if err := ctx.Err(); err != nil {
			return nil // đang tắt máy, không phải lỗi
		}

		// Lỗi tầng fetch (mất kết nối, không có leader): franz-go tự thử lại
		// ở lượt poll sau. Chỉ log, đừng thoát.
		fetches.EachError(func(topic string, part int32, err error) {
			if errors.Is(err, context.Canceled) {
				return
			}
			c.log.Warn("kafka: lỗi fetch", "group", group, "topic", topic, "partition", part, "err", err)
		})

		var done []*kgo.Record

		// Xử lý TUẦN TỰ trong từng partition — đó là điều kiện để giữ
		// thứ tự mà aggregate_id đã mua được (Phụ lục A).
		fetches.EachPartition(func(p kgo.FetchTopicPartition) {
			for _, rec := range p.Records {
				if ctx.Err() != nil {
					return
				}
				if err := c.handleRecord(ctx, group, routes[rec.Topic], rec); err != nil {
					// Tới đây nghĩa là KHÔNG xử lý được và cũng KHÔNG đưa
					// vào DLQ được. Dừng partition này tại đúng chỗ: những
					// record trước đã vào `done`, record này thì không.
					c.log.Error("kafka: dừng partition tại record lỗi",
						"group", group, "topic", rec.Topic,
						"partition", rec.Partition, "offset", rec.Offset, "err", err)
					return
				}
				done = append(done, rec)
			}
		})

		if len(done) > 0 {
			// ★ Commit là bước CUỐI CÙNG. CommitRecords ghi offset+1 của
			// record cao nhất trong mỗi partition.
			if err := cl.CommitRecords(ctx, done...); err != nil {
				// Không commit được → không mất gì, chỉ là lượt sau xử lý
				// lại. Guard sẽ khử trùng.
				c.log.Error("kafka: commit thất bại, sẽ xử lý lại", "group", group, "err", err)
			}
		}

		// Mở cửa cho rebalance đã bị chặn trong lúc xử lý. Xem §4.5.
		cl.AllowRebalance()
	}
}

4.4 ConsumeResetOffset — chọn có ý thức

Tuỳ chọn này chỉ có tác dụng khi group chưa có offset đã lưu (lần đầu chạy, hoặc offset đã hết hạn).

Lựa chọn Nghĩa Phù hợp khi
AtStart() (chọn) Đọc lại toàn bộ topic Consumer xây trạng thái từ event. Nó phải xử lý được cả lịch sử, nếu không số liệu sẽ thiếu đúng phần trước khi nó ra đời
AtEnd() Chỉ event từ giờ trở đi Consumer làm việc nhất thời: gửi thông báo, đẩy websocket. Gửi lại thông báo của tuần trước là spam

stats thuộc loại thứ nhất, nên AtStart(). Và điều đó chỉ an toàn nhờ guard: đọc lại 7 ngày event mà không cộng trùng counter nào.

Hai lựa chọn này là hai loại consumer khác nhau về bản chất, không phải hai giá trị cấu hình. Chọn sai theo hướng AtEnd() cho consumer xây trạng thái là cách tạo ra số liệu thiếu vĩnh viễn mà không ai biết vì sao.

4.5 ⚠️ Ngân sách retry phải nhỏ hơn RebalanceTimeout

Đây là cạm bẫy khó nhất của mục này, và nó tạo ra một vòng lặp tự nuôi.

BlockRebalanceOnPoll khiến rebalance bị hoãn cho tới khi bạn gọi AllowRebalance(). Cần thế vì nếu không:

poll → nhận partition 0, record 100..200
    │  ... đang xử lý ...
    │  ⟳ REBALANCE: partition 0 bị giao cho thành viên khác
    ↓
CommitRecords(100..200)   ← commit offset cho partition KHÔNG CÒN thuộc về mình

Lời commit đó bị từ chối hoặc ghi từ một generation đã cũ. Trong lúc đó thành viên mới đã đọc lại từ offset cũ. Kết quả là hai consumer cùng xử lý một khoảng record, và bạn không có cách nào biết.

Nhưng cái giá của việc chặn rebalance là: thời gian xử lý bị tính vào ngân sách im lặng của thành viên. Chặn lâu hơn RebalanceTimeout thì group kết luận bạn đã chết:

Ngân sách retry:  4 lần × backoff tới 5 phút = 10 phút
RebalanceTimeout: 60 giây
                          ↓
① 60s: group đá worker ra, giao partition cho thành viên khác
② worker xử lý xong, CommitRecords → THẤT BẠI (không còn là thành viên)
③ worker rejoin, có thể nhận lại đúng partition đó
④ đọc lại từ offset cũ → gặp lại đúng message hỏng
⑤ về ①

Vòng này quay mãi. Triệu chứng: log đầy commit thất bại và nhận partition, consumer lag không giảm, CPU thấp. Rất khó chẩn đoán vì không thành phần nào báo lỗi rõ ràng — mỗi thành phần đều đang làm đúng việc của nó.

Quy tắc: MaxAttempts × tổng backoff × PollMax phải nhỏ hơn nhiều RebalanceTimeout.

Mặc định của tài liệu này:

Tham số Giá trị Ghi chú
MaxAttempts 4 tức 3 lần chờ
Backoff 100ms, 200ms, 400ms tổng 700ms cho một message
RebalanceTimeout 60s dư ~85 lần

Ngân sách 700ms nghe ít, và đúng là ít — nó chỉ đủ cho lỗi tạm thời thật sự ngắn: một deadlock Postgres, một nhịp mất kết nối. Mọi thứ cần chờ lâu hơn là việc của Phụ lục B, nơi thời gian chờ nằm ngoài vòng lặp poll.

Trường hợp xấu nhất phải tính theo cả lô, không phải một message: 500 record cùng hỏng × 700ms ≈ 350 giây > 60 giây. Nếu bạn lo kịch bản này, hạ PollMax xuống 100 (≈70s, vẫn sát) hoặc chuyển hẳn sang retry topic. Với lỗi hiếm và rải rác thì mặc định trên an toàn.

4.6 DLQ — khi nào và làm sao

// handleRecord xử lý một record tới kết cục cuối cùng.
//
// Trả nil  → record đã xử lý xong HOẶC đã vào DLQ ⇒ được phép commit.
// Trả error → chưa xử lý được và cũng chưa vào DLQ ⇒ KHÔNG được commit.
func (c *Consumer) handleRecord(ctx context.Context, group string, handlers []eventbus.Handler, rec *kgo.Record) error {
	var env contracts.Envelope
	if err := json.Unmarshal(rec.Value, &env); err != nil {
		// Không giải mã được thì không có event_id, không route được,
		// và lần sau vẫn thế. Vào DLQ ngay.
		return c.deadLetter(ctx, group, rec, 0, fmt.Errorf("giải mã envelope: %w", err))
	}
	if err := env.Validate(); err != nil {
		return c.deadLetter(ctx, group, rec, 0, err)
	}

	log := c.log.With(
		"group", group,
		"event_id", env.EventID,
		"event_type", env.EventType,
		// Sợi chỉ từ bước 4: grep một correlation_id là thấy toàn bộ
		// hành trình HTTP → outbox → Kafka → tới đây.
		"correlation_id", env.CorrelationID,
	)

	var lastErr error
	for attempt := 1; attempt <= c.cfg.MaxAttempts; attempt++ {
		lastErr = nil
		for _, h := range handlers {
			if err := h(ctx, env); err != nil {
				lastErr = err
				break
			}
		}
		if lastErr == nil {
			return nil
		}

		// Đang tắt máy: KHÔNG tính là thất bại của message. Trả lỗi để
		// không commit — lượt chạy sau xử lý lại.
		if ctx.Err() != nil {
			return lastErr
		}
		if eventbus.IsPermanent(lastErr) {
			log.Warn("lỗi vĩnh viễn, vào DLQ ngay", "err", lastErr)
			return c.deadLetter(ctx, group, rec, attempt, lastErr)
		}
		if attempt == c.cfg.MaxAttempts {
			break
		}

		wait := c.cfg.RetryBase << (attempt - 1) // 100ms, 200ms, 400ms
		log.Warn("lỗi tạm thời, sẽ thử lại", "attempt", attempt, "wait", wait, "err", lastErr)

		select {
		case <-ctx.Done():
			return lastErr
		case <-time.After(wait):
		}
	}

	log.Error("hết số lần thử, vào DLQ", "attempts", c.cfg.MaxAttempts, "err", lastErr)
	return c.deadLetter(ctx, group, rec, c.cfg.MaxAttempts, lastErr)
}

// deadLetter chuyển record sang topic DLQ tương ứng.
//
// Value giữ NGUYÊN từng byte; mọi thông tin chẩn đoán đi vào header.
// Nhờ vậy công cụ phát lại chỉ cần copy Value về topic gốc, không phải
// bóc lớp bọc nào — và message không phình lên quá giới hạn broker.
func (c *Consumer) deadLetter(ctx context.Context, group string, rec *kgo.Record, attempts int, cause error) error {
	dlqTopic := contracts.DLQ(rec.Topic)

	headers := make([]kgo.RecordHeader, 0, len(rec.Headers)+6)
	headers = append(headers, rec.Headers...)
	headers = append(headers,
		kgo.RecordHeader{Key: "dlq_original_topic", Value: []byte(rec.Topic)},
		kgo.RecordHeader{Key: "dlq_original_partition", Value: []byte(fmt.Sprint(rec.Partition))},
		kgo.RecordHeader{Key: "dlq_original_offset", Value: []byte(fmt.Sprint(rec.Offset))},
		kgo.RecordHeader{Key: "dlq_consumer_group", Value: []byte(group)},
		kgo.RecordHeader{Key: "dlq_attempts", Value: []byte(fmt.Sprint(attempts))},
		kgo.RecordHeader{Key: "dlq_error", Value: []byte(truncate(cause.Error(), 900))},
	)

	out := &kgo.Record{
		Topic:   dlqTopic,
		Key:     rec.Key, // giữ Key để message của cùng aggregate vẫn cạnh nhau
		Value:   rec.Value,
		Headers: headers,
	}

	if err := c.dlq.cl.ProduceSync(ctx, out).FirstErr(); err != nil {
		// ★ KHÔNG nuốt lỗi này. Trả về để KHÔNG commit offset: message
		// vẫn nằm trong topic gốc và sẽ được thử lại. Nuốt nó ở đây là
		// đánh mất message ở khoảng trống giữa hai topic.
		return fmt.Errorf("kafka: đẩy vào %s thất bại: %w", dlqTopic, err)
	}

	c.log.Error("kafka: đã chuyển vào DLQ",
		"dlq_topic", dlqTopic, "original_topic", rec.Topic,
		"partition", rec.Partition, "offset", rec.Offset,
		"group", group, "attempts", attempts, "err", cause)
	return nil
}

truncate là đúng hàm đã viết ở bước 5 §7.6 — cắt theo ranh giới rune UTF-8. Chuyển nó vào internal/platform/stringsx để cả outbox và kafka dùng chung, hoặc chép lại một bản; với một hàm mười dòng thì cả hai đều được, miễn là có cắt. Header Kafka không phải cột TEXT nên chuỗi hỏng không làm INSERT thất bại, nhưng một lỗi vài KB nhân với mọi message hỏng vẫn là chỗ tốn dung lượng vô ích.

⚠️ Nếu chính lời publish vào DLQ cũng thất bại vĩnh viễn?

Ví dụ: message sát trần max.message.bytes, thêm ~300 byte header là vượt. Lúc đó handleRecord luôn trả lỗi, không bao giờ commit, và partition tắc vĩnh viễn — đúng cái poison message ở §1.3.

Đây là một ngã ba thật, không có lựa chọn nào hoàn hảo. Tài liệu này chọn tắc, và đó là lựa chọn có ý thức:

  • Tắc thì ồn ào: lag tăng, cảnh báo kêu, có người tới xem. Sự cố được phát hiện.
  • Bỏ qua để chảy tiếp thì im lặng: message biến mất, mọi chỉ số bình thường.

Và điều làm lựa chọn này chấp nhận được là event vẫn còn ở bảng outbox trong 7 ngày, kèm event_id để tra. Đó là chỗ mà thiết kế của bước 5 trả cổ tức: outbox không chỉ là hàng chờ gửi, nó là sổ ghi gốc của mọi event từng phát ra. Với dữ liệu quan trọng, "tắc và ầm ĩ" luôn tốt hơn "chảy và mất".

Nếu hệ thống của bạn ưu tiên thông lượng hơn tính toàn vẹn, hãy đảo lựa chọn — nhưng viết lý do vào code, đừng để nó là hệ quả tình cờ của một dòng continue.

5. modules/stats

Module này là loại đặc biệt đã được ARCHITECTURE.md §4.4 mô tả: gần như chỉ có consumer. Nó không có endpoint ghi nào; mọi thay đổi đến từ event.

5.1 repository

internal/modules/stats/repository/postgres.go:

package repository

import (
	"context"
	"errors"
	"fmt"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/chuongtd/community/internal/modules/stats/domain"
	"github.com/chuongtd/community/internal/modules/stats/repository/gen"
)

type Postgres struct {
	pool *pgxpool.Pool
}

func NewPostgres(pool *pgxpool.Pool) *Postgres { return &Postgres{pool: pool} }

// EnsureUserStats tạo dòng thống kê rỗng. Nhận tx vì nó luôn được gọi
// cùng transaction với guard.Mark (§3.2).
func (r *Postgres) EnsureUserStats(ctx context.Context, tx pgx.Tx, userID uuid.UUID) error {
	if err := gen.New(tx).EnsureUserStats(ctx, userID); err != nil {
		return fmt.Errorf("stats: tạo dòng user_stats: %w", err)
	}
	return nil
}

// GetUserStats đọc thống kê. Dùng pool vì đây là đường ĐỌC, không cần
// transaction — và không được ghé vào transaction của ai khác.
func (r *Postgres) GetUserStats(ctx context.Context, userID uuid.UUID) (domain.UserStats, error) {
	row, err := gen.New(r.pool).GetUserStats(ctx, userID)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return domain.UserStats{}, domain.ErrStatsNotFound
		}
		return domain.UserStats{}, fmt.Errorf("stats: đọc user_stats: %w", err)
	}
	return domain.UserStats{
		UserID:              row.UserID,
		PostsCount:          row.PostsCount,
		TotalPostViews:      row.TotalPostViews,
		FollowersCount:      row.FollowersCount,
		FollowingUsersCount: row.FollowingUsersCount,
		BookmarksCount:      row.BookmarksCount,
		UpdatedAt:           row.UpdatedAt,
	}, nil
}

internal/modules/stats/domain/stats.go:

package domain

import (
	"errors"
	"time"

	"github.com/google/uuid"
)

var ErrStatsNotFound = errors.New("không tìm thấy thống kê")

// UserStats là entity thuần Go: không tag json, không tag db.
// Cùng lý lẽ với domain.User ở bước 4 §11.
type UserStats struct {
	UserID              uuid.UUID
	PostsCount          int32
	TotalPostViews      int32
	FollowersCount      int32
	FollowingUsersCount int32
	BookmarksCount      int32
	UpdatedAt           time.Time
}

Số cột thật tuỳ theo migration 000002 của bạn — bước 3 §4 có 8 counter. Đối chiếu gen/models.go rồi thêm cho khớp; sqlc đã sinh sẵn tên trường đúng nên chỉ là việc chép.

5.2 subscriber

internal/modules/stats/subscriber/subscriber.go:

// Package subscriber là trái tim của module stats: mọi thay đổi số liệu
// đều đến từ event, không từ lời gọi hàm của module khác (Luật 1).
package subscriber

import (
	"context"
	"fmt"
	"log/slog"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"

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

// ConsumerName là tên consumer group của module stats.
//
// ⚠️ KHÔNG ĐỔI giá trị này. Nó vừa là group id của Kafka, vừa là khoá
// trong bảng consumed_events. Đổi nó = đọc lại toàn bộ topic VÀ mất
// toàn bộ lịch sử chống trùng cùng lúc (bước 6 §3.4).
const ConsumerName = "stats-service"

// Ba interface dưới đây khai báo ĐÚNG những gì subscriber cần — khai ở
// phía dùng, theo mẫu bước 4 §13.1. Nhờ vậy test dùng được bản giả.
type Transactor interface {
	Do(ctx context.Context, fn func(tx pgx.Tx) error) error
}

type Guard interface {
	Mark(ctx context.Context, tx pgx.Tx, eventID, consumer string) (bool, error)
}

type Repository interface {
	EnsureUserStats(ctx context.Context, tx pgx.Tx, userID uuid.UUID) error
}

type Subscriber struct {
	tx    Transactor
	guard Guard
	repo  Repository
	log   *slog.Logger
}

func New(tx Transactor, guard Guard, repo Repository, log *slog.Logger) *Subscriber {
	return &Subscriber{tx: tx, guard: guard, repo: repo, log: log}
}

internal/modules/stats/subscriber/on_user_registered.go:

package subscriber

// OnUserRegistered tạo dòng user_stats cho người dùng mới.
//
// Đây là lời hứa của bước 4 §18.2: module stats tự tạo dòng của mình,
// identity không bao giờ ghi vào bảng này.
func (s *Subscriber) OnUserRegistered(ctx context.Context, e contracts.Envelope) error {
	p, err := contracts.DecodePayload[v1.UserRegistered](e)
	if err != nil {
		// Payload hỏng: thử lại bao nhiêu lần cũng ra kết quả này.
		return eventbus.Permanent(err)
	}

	userID, err := uuid.Parse(p.UserID)
	if err != nil {
		return eventbus.Permanent(
			fmt.Errorf("stats: user_id không phải UUID (%q): %w", p.UserID, err))
	}

	return s.tx.Do(ctx, func(tx pgx.Tx) error {
		ok, err := s.guard.Mark(ctx, tx, e.EventID, ConsumerName)
		if err != nil {
			return err
		}
		if !ok {
			s.log.Debug("stats: bỏ qua event trùng", "event_id", e.EventID)
			return nil
		}

		// ★ CÙNG transaction với Mark. Xem §3.2.
		return s.repo.EnsureUserStats(ctx, tx, userID)
	})
}

5.3 ⚠️ Handler này không cần guard — vậy sao vẫn dùng?

Câu hỏi này phải được đặt ra, vì phép thử ở ARCHITECTURE.md §8.4 là "chạy hai lần liên tiếp có ra kết quả khác không?", và EnsureUserStats là INSERT ... ON CONFLICT DO NOTHING — chạy 100 lần vẫn ra một dòng. Theo phép thử, guard ở đây là thừa.

Tôi vẫn dùng, và cần nói thẳng lý do nào không đủ thuyết phục trước:

  • ❌ "Cho nhất quán" — nhất quán không phải lý do để chạy thêm một INSERT.
  • ❌ "Sau này cần" — chính lý lẽ tôi đã bác bỏ bốn lần ở các bước trước khi hoãn guard.go tới đúng bước này.

Lý do thật, và nó khác hai cái trên:

Guard bảo vệ thân hàm tương lai, không phải dòng code hôm nay. OnUserRegistered là chỗ tự nhiên nhất để ai đó thêm việc — một counter đăng ký toàn hệ thống, một dòng audit, một bản ghi welcome. Người thêm việc đó đang nghĩ về nghiệp vụ mới, không nghĩ về ngữ nghĩa giao hàng của Kafka. Nếu guard đã ở đó, việc mới tự động được bảo vệ. Nếu chưa, nó phải tự nhớ ra một chuyện thuộc tầng hạ tầng — và đó là loại chuyện người ta không nhớ.

Cái giá: một INSERT vào index đã mở sẵn, trong transaction đã mở sẵn — cỡ vài chục micro-giây, cộng một dòng ~40 byte giữ 30 ngày.

Còn phép thử §8.4 vẫn đúng và vẫn phải dùng — nó chỉ trả lời một câu hỏi hẹp hơn bạn tưởng: "thao tác này có tự an toàn khi giao trùng không?" Câu trả lời "có" nghĩa là bạn được phép bỏ guard, không phải nên bỏ. Bỏ là hợp lý khi handler chỉ có đúng một UPDATE ... SET x = $2 và sẽ không bao giờ có gì thêm.

Handler Tự idempotent? Cần guard?
EnsureUserStats (ON CONFLICT DO NOTHING) Có Không bắt buộc — dùng để bảo vệ việc thêm sau
posts_count = posts_count + 1 Không Bắt buộc
SET avatar_url = $2 Có Không
Gửi email chào mừng Không, và guard cũng không cứu được (§3.3) Cần cơ chế khác

5.4 module.go — một điểm lắp ráp, hai binary

internal/modules/stats/module.go:

package stats

import (
	"log/slog"

	"github.com/go-chi/chi/v5"
	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/chuongtd/community/internal/contracts"
	v1 "github.com/chuongtd/community/internal/contracts/v1"
	statshttp "github.com/chuongtd/community/internal/modules/stats/transport/http"
	"github.com/chuongtd/community/internal/modules/stats/repository"
	"github.com/chuongtd/community/internal/modules/stats/subscriber"
	"github.com/chuongtd/community/internal/platform/eventbus"
)

type Module struct {
	sub *subscriber.Subscriber
	api *statshttp.Handler
}

func New(pool *pgxpool.Pool, tx subscriber.Transactor, guard subscriber.Guard, log *slog.Logger) *Module {
	repo := repository.NewPostgres(pool)
	return &Module{
		sub: subscriber.New(tx, guard, repo, log),
		api: statshttp.NewHandler(repo, log),
	}
}

// RegisterHTTP — chỉ endpoint ĐỌC. Module này không có đường ghi qua HTTP.
func (m *Module) RegisterHTTP(r chi.Router) {
	r.Get("/users/{id}/stats", m.api.GetUserStats)
}

// RegisterEvents khai báo module này nghe gì.
//
// Đọc đúng hàm này là biết module tham gia vào hệ thống ra sao — không
// phải đi tìm khắp package.
func (m *Module) RegisterEvents(bus eventbus.Subscriber) {
	bus.Subscribe(contracts.TopicIdentity, subscriber.ConsumerName,
		eventbus.Route(map[string]eventbus.Handler{
			v1.TypeUserRegistered: m.sub.OnUserRegistered,
		}))
}

Một Module, hai binary, hai mặt khác nhau

cmd/api gọi RegisterHTTP và không bao giờ gọi RegisterEvents. cmd/worker làm ngược lại.

Đây không phải chi tiết vụn: nó có nghĩa là tầng phục vụ người dùng không giữ kết nối Kafka nào. api chỉ đọc user_stats từ Postgres. Kafka sập, worker chết, số liệu ngừng cập nhật — nhưng GET /users/{id}/stats vẫn trả 200 với dữ liệu cuối cùng đã biết. Bán kính sự cố bị giới hạn bởi chính cách lắp ráp.

Và eventbus.Route ở đây là hàm viết từ bước 2 §6.2 — nó bỏ qua trong im lặng mọi EventType không khớp. Ngày module post thêm post.post.pinned.v1, stats không hề gãy. Đó là lời hứa của bước 2 lần đầu được kiểm chứng bằng code chạy thật.

Backend

All Rights Reserved

Table of contents


Viblo
Let's register a Viblo Account to get more interesting posts.
Register