Khử Trùng Lặp Và Chuyển Vùng Dự Phòng Cho Subscriber (Subscriber Failover)

Why this matters in production

Trong các hệ thống phân tán chịu lỗi cao, việc triển khai nhiều thực thể tiêu thụ dữ liệu (multiple subscriber instances) chạy song song trên các zone hoặc regions khác nhau là một thiết kế bắt buộc. Thiết kế này giúp hệ thống sẵn sàng thay thế các worker bị sập mà không làm gián đoạn luồng xử lý tin nhắn.

Tuy nhiên, khi xảy ra sự cố sập một vùng (regional failover) hoặc sập một zone:

• Các subscriber instances tại vùng bị lỗi sẽ đột ngột biến mất hoặc mất kết nối mạng với Pub/Sub server.
• Các thông điệp mà các instances lỗi này đang "thuê" (lease) để xử lý dở dang sẽ bị rơi vào trạng thái lấp lửng.
• Nếu không hiểu rõ cơ chế quản lý vòng đời tin nhắn của client library (lease management) và cách các instance còn lại cạnh tranh để nhận lại các tin nhắn này (lease competition), bạn sẽ gặp phải hiện tượng nghẽn cổ chai nghiêm trọng hoặc xử lý trùng lặp tràn lan, làm mất tác dụng của kiến trúc dự phòng.

---

Internal Model: Cơ Chế Quản Lý Lease và Tranh Chấp Khi Failover

Để hiểu cách thức chuyển đổi dự phòng diễn ra ở phía Subscriber, trước hết ta cần mổ xẻ cơ chế quản lý thời gian thuê tin nhắn (Lease Management) của Pub/Sub Client Library.

sequenceDiagram
    autonumber
    participant PubSub as Pub/Sub Server
    participant Worker1 as Subscriber Worker 1 (Sập vùng)
    participant Worker2 as Subscriber Worker 2 (Hoạt động)

    PubSub->>Worker1: Giao Message M1 (ACK Deadline ban đầu: 10s)
    Note over Worker1: Worker 1 đang xử lý M1...
    Worker1->>PubSub: Gửi modifyAckDeadline (Gia hạn thêm 10s)
    Note over Worker1: Region sập vật lý! Worker 1 chết đột ngột.
    Note over PubSub: Chờ hết 10s gia hạn cuối cùng...<br/>Không có modifyAckDeadline mới từ Worker 1.
    Note over PubSub: Giải phóng M1 về trạng thái chưa phân phối
    PubSub->>Worker2: Giao lại Message M1 (ACK Deadline: 10s)
    Worker2->>Worker2: Xử lý thành công M1
    Worker2->>PubSub: Gửi ACK M1

1. Cơ Chế Thuê Tin Nhắn Tự Động (Lease Management)

Khi bạn sử dụng các thư viện SDK chính thức của Google (như Java, Go, Python Client Libraries) để thực hiện StreamingPull:

• Khi client nhận được một nhóm tin nhắn từ Pub/Sub server, server sẽ thiết lập một khoảng thời gian chờ xác nhận ban đầu (mặc định là ackDeadline của Subscription, ví dụ 10 giây).
• Nếu việc xử lý tin nhắn trong ứng dụng kéo dài hơn 10 giây đó, client library sẽ chạy một tiến trình nền tự động gửi các yêu cầu modifyAckDeadline lên Pub/Sub server để kéo dài thời gian thuê (lease extension). Tiến trình này thường được gọi là Lease Manager.
• Nó giúp bảo vệ tin nhắn không bị phân phối lại cho các instances khác trong khi worker hiện tại vẫn đang tích cực xử lý.
• Giới hạn tối đa cho việc gia hạn tự động này (Max Extension Period) có thể cấu hình được trên client (ví dụ tối đa 1 giờ).

2. Hiện Tượng Tranh Chấp Lease (Lease Competition) Khi Có Sự Cố

Khi một thực thể tiêu thụ (ví dụ Worker 1 nằm ở us-central1) bị sập do sự cố vùng:

1. Đứt kết nối đột ngột: Giao thức kết nối dài hạn StreamingPull giữa Worker 1 và Pub/Sub bị ngắt.
2. Hết hạn thời gian thuê: Pub/Sub server sẽ dừng nhận được các gói tin gia hạn modifyAckDeadline từ Worker 1. Server sẽ chờ cho đến khi thời hạn ACK deadline cuối cùng của các tin nhắn đang được giữ bởi Worker 1 trôi qua.
3. Giải phóng và Tranh chấp: Ngay khi deadline kết thúc, Pub/Sub server sẽ giải phóng (evict) các tin nhắn này và đưa chúng trở lại hàng đợi.
4. Các subscriber instances khác đang hoạt động ở vùng lành lặn (ví dụ Worker 2 tại us-east1) đang duy trì kết nối StreamingPull mở sẵn sẽ ngay lập tức nhận được các tin nhắn này. Chúng "cạnh tranh" với nhau để giành quyền xử lý các tin nhắn mồ côi.

3. Nguy Hiểm Của Việc Cấu Hình Sai Giới Hạn Luồng (Flow Control)

Trong kịch bản failover, nếu số lượng tin nhắn bị kẹt từ vùng lỗi quá lớn:

• Nếu các worker còn lại không giới hạn số lượng tin nhắn được phép nhận đồng thời (Flow Control), chúng sẽ cố gắng kéo sạch toàn bộ lượng tin nhắn được giải phóng về bộ nhớ đệm nội bộ của mình.
• Hệ quả: Bộ nhớ RAM của worker bị tràn (Out of Memory - OOM), hoặc worker bị quá tải CPU không thể xử lý kịp, dẫn đến việc không thể gửi request gia hạn lease (modifyAckDeadline) cho chính các tin nhắn mà nó vừa kéo về.
• Các tin nhắn này lại tiếp tục hết hạn deadline trên server và lại được phân phối tiếp cho các worker khác, tạo ra một vòng lặp sập dây chuyền (cascading failure loop) trên toàn bộ cụm Subscriber.

---

Constraints, Trade-offs & Failure Modes

1. Phân Tách Mạng và Kịch Bản Não Phân Đôi (Network Partition / Split-Brain)

Một kịch bản lỗi cực kỳ khó chịu trong thực tế vận hành:

• Giả sử Worker 1 vẫn đang sống và xử lý tin nhắn M1, nhưng kết nối mạng đi ra ngoài của nó bị nghẽn khiến nó không thể gửi gói tin gia hạn modifyAckDeadline về cho Pub/Sub server.
• Pub/Sub server coi như Worker 1 đã chết, liền giải phóng M1 và gửi nó cho Worker 2.
• Lúc này, cả Worker 1 và Worker 2 đều đang xử lý song song thông điệp M1 tại cùng một thời điểm. Đây là trạng thái não phân đôi (split-brain).
• Hệ quả: Nếu ứng dụng không có cơ chế kiểm tra chéo, cả hai worker sẽ cùng ghi dữ liệu trùng lặp vào database downstream.

---

Production Patterns & Implementation Guidance

1. Thiết Kế Bộ Khử Trùng Lặp Tầng Ứng Dụng (Application-Level Deduplicator) Với Redis

Để chống lại hiện tượng trùng lặp tuyệt đối (kể cả trong các kịch bản split-brain hoặc sập vùng vượt quá cửa sổ 10 phút của Dataflow), bạn nên triển khai một bộ lọc trùng lặp ở tầng ứng dụng sử dụng một database lưu trữ phân tán có tốc độ cao như Redis với cơ chế nguyên tử (atomic operations).

package main

import (
	"context"
	"fmt"
	"time"
	"github.com/go-redis/redis/v8"
)

type Deduplicator struct {
	rdb *redis.Client
	ttl time.Duration
}

func NewDeduplicator(redisAddr string, ttl time.Duration) *Deduplicator {
	rdb := redis.NewClient(&redis.Options{
		Addr: redisAddr,
	})
	return &Deduplicator{rdb: rdb, ttl: ttl}
}

// IsDuplicate kiểm tra xem messageID đã được xử lý chưa bằng lệnh SETNX nguyên tử
func (d *Deduplicator) IsDuplicate(ctx context.Context, messageID string) (bool, error) {
	key := fmt.Sprintf("dedup:msg:%s", messageID)
	
	// SETNX (Set if Not Exists) thiết lập key nếu chưa có và trả về true.
	// Nếu key đã tồn tại, trả về false (nghĩa là message đã bị xử lý trước đó).
	inserted, err := d.rdb.SetNX(ctx, key, "processing", d.ttl).Result()
	if err != nil {
		return false, err
	}
	
	// Nếu không chèn được key mới -> Đây là tin nhắn trùng lặp
	return !inserted, nil
}

// MarkComplete cập nhật trạng thái của message là đã xử lý thành công
func (d *Deduplicator) MarkComplete(ctx context.Context, messageID string) error {
	key := fmt.Sprintf("dedup:msg:%s", messageID)
	return d.rdb.Set(ctx, key, "completed", d.ttl).Err()
}

Cách áp dụng trong luồng xử lý của Subscriber:

1. Nhận message từ Pub/Sub.
2. Gọi IsDuplicate(ctx, message.ID).
   • Nếu trả về true: Bỏ qua việc xử lý, lập tức gọi message.Ack() để xóa tin nhắn khỏi Pub/Sub.
   • Nếu trả về false: Tiến hành xử lý logic nghiệp vụ.
3. Sau khi xử lý thành công, gọi MarkComplete(ctx, message.ID) và gọi message.Ack().

2. Cấu Hình Tối Ưu Hóa Flow Control Cho Client Resilient

Khi khởi tạo Subscriber client, bạn bắt buộc phải thiết lập các tham số giới hạn luồng (Flow Control) để ngăn chặn việc sập dây chuyền khi failover:

// Cấu hình Go Pub/Sub Receive Settings cho Subscriber
sub.ReceiveSettings.MaxOutstandingMessages = 100 // Tối đa 100 tin nhắn chưa ACK trong bộ nhớ
sub.ReceiveSettings.MaxOutstandingBytes = 1e8    // Tối đa 100MB bộ đệm dữ liệu

// Thiết lập thời gian gia hạn lease tối đa
sub.ReceiveSettings.MaxExtension = 5 * time.Minute // Không gia hạn quá 5 phút

• MaxOutstandingMessages: Khống chế số lượng tin nhắn tối đa mà một thực thể subscriber nạp vào bộ nhớ đệm tại một thời điểm. Giúp tránh tràn RAM khi có bão tin nhắn dồn về sau sự cố vùng.
• MaxExtension: Giới hạn thời gian gia hạn tối đa. Nếu một tin nhắn bị kẹt quá thời gian này (ví dụ do lỗi vòng lặp vô hạn trong code xử lý), client library sẽ tự động dừng gửi lệnh modifyAckDeadline, trả tin nhắn về cho server để thực thể khác xử lý lại.

---

References

• Pub/Sub Subscriber Flow Control Settings
• Managing message leases in Google Cloud Pub/Sub
• Distributed locking patterns with Redis (Redlock)