Đếm Click An Toàn: Idempotency, DLQ và Partial Batch Failure
Consumer ở bài 09 mới chỉ ghi log, và nó ngây thơ về hai chuyện của xử lý bất đồng bộ thực tế. Thứ nhất, hệ thống nhắn tin phân tán thường giao ít nhất một lần, nên cùng một sự kiện có thể tới hai lần; nếu cứ cộng bộ đếm mỗi lần nhận thì đếm sai. Thứ hai, khi xử lý lỗi thì sự kiện đi đâu, thử lại bao nhiêu lần, và nếu vẫn hỏng thì có biến mất không? Bài này biến consumer thành một aggregator thật giải cả hai.
Mục tiêu
Chèn một hàng đợi SQS giữa EventBridge và Lambda để có batch, retry và dead-letter queue. Đếm click vào DynamoDB sao cho nhận trùng sự kiện không đếm sai (idempotent), gắn DLQ để sự kiện lỗi không mất, và báo lỗi theo từng message để chỉ thử lại cái hỏng. Ta test thật cả ba: đếm đúng, gửi trùng đếm một lần, và sự kiện lỗi rơi vào DLQ.
Vì sao chèn SQS vào giữa
Bài 09 nối EventBridge thẳng vào Lambda. Thêm một hàng đợi SQS vào giữa cho ta ba thứ mà đường thẳng không có sẵn. SQS gom message thành batch để Lambda xử lý nhiều cái một lần, rẻ hơn. Nó giữ message và thử lại theo lịch nếu xử lý lỗi, thay vì mất. Và nó có dead-letter queue: sau một số lần thử thất bại, message được chuyển sang một hàng đợi riêng để điều tra sau, không kẹt mãi ở hàng chính.
resolve ──PutEvents──▶ EventBridge bus
│ rule khop link.clicked
▼
ClickQueue (SQS) ──▶ ClickAggregator (Lambda, batch)
│ │ ghi DynamoDB (transaction)
│ qua 3 lan loi │
▼ ▼
ClickDLQ (SQS) META.clicks += 1
(dieu tra sau) STAT#<ngay>.count += 1
CLICK#<eventId> (marker idempotency)
Trong SAM, EventBridge rule trỏ target sang SQS, và Lambda nhận từ SQS:
ClickQueue:
Type: AWS::SQS::Queue
Properties:
VisibilityTimeout: 30
RedrivePolicy:
deadLetterTargetArn: !GetAtt ClickDLQ.Arn
maxReceiveCount: 3
maxReceiveCount: 3 nghĩa là một message được thử tối đa ba lần; lần thứ tư mà vẫn chưa xử lý xong thì SQS chuyển nó sang ClickDLQ.
Đếm idempotent bằng transaction
Phần đếm idempotent là chỗ cần cẩn thận nhất. Mỗi sự kiện EventBridge mang một id duy nhất. Ý tưởng: ghi một marker CLICK#<id> đánh dấu "đã xử lý sự kiện này", và chỉ cho ghi marker nếu nó chưa tồn tại. Nếu ghép việc ghi marker với việc cộng bộ đếm vào cùng một transaction, thì cộng bộ đếm và đánh dấu đã xử lý xảy ra cùng lúc hoặc không cái nào xảy ra:
await ddb.send(
new TransactWriteCommand({
TransactItems: [
{
Update: {
TableName: TABLE,
Key: { PK: `LINK#${code}`, SK: "META" },
UpdateExpression: "ADD clicks :one",
ExpressionAttributeValues: { ":one": 1 },
},
},
{
Update: {
TableName: TABLE,
Key: { PK: `LINK#${code}`, SK: `STAT#${date}` },
UpdateExpression: "ADD #c :one",
ExpressionAttributeNames: { "#c": "count" },
ExpressionAttributeValues: { ":one": 1 },
},
},
{
Put: {
TableName: TABLE,
Item: { PK: `CLICK#${id}`, SK: "CLICK", ttl },
ConditionExpression: "attribute_not_exists(PK)",
},
},
],
})
);
Transaction này làm ba việc: cộng bộ đếm tổng trên item META, cộng bộ đếm ngày trên item STAT#<ngày>, và ghi marker với điều kiện attribute_not_exists(PK). Nếu sự kiện tới lần hai, marker đã có, điều kiện fail, và DynamoDB hủy toàn bộ transaction. Không có bộ đếm nào bị cộng thêm. Khi đó lệnh ném TransactionCanceledException, và handler hiểu đó là sự kiện trùng nên bỏ qua an toàn:
if ((e as { name?: string }).name === "TransactionCanceledException") {
console.log("duplicate, skip", { id });
continue;
}
Marker mang một thuộc tính ttl đặt 24 giờ sau, và bảng bật Time To Live trên thuộc tính đó, nên DynamoDB tự xóa marker cũ, bảng không phình mãi. Đây chính là mẫu mà thư viện AWS Lambda Powertools đóng gói trong tiện ích Idempotency của nó; ở đây ta làm thẳng bằng transaction để thấy rõ cơ chế, còn Powertools cho cách viết ngắn hơn nếu không muốn tự quản lý marker.
Báo lỗi theo từng message
Khi Lambda nhận một batch nhiều message từ SQS, nếu một message hỏng mà báo cả batch là lỗi thì những message tốt cũng bị thử lại, sinh ra xử lý trùng. Cách đúng là báo từng message hỏng. Tài liệu mô tả cơ chế này: hàm "catches failures for individual messages and returns their identifiers in a batchItemFailures response, signaling Lambda to retry only the failed messages rather than the entire batch."
const batchItemFailures: SQSBatchItemFailure[] = [];
for (const record of event.Records) {
try {
// ...xu ly record...
} catch (err) {
batchItemFailures.push({ itemIdentifier: record.messageId });
}
}
return { batchItemFailures };
Để bật cơ chế này, event source khai FunctionResponseTypes: [ReportBatchItemFailures]. Message được báo lỗi sẽ quay lại hàng đợi và thử lại; những message khác trong batch coi như xong.
Thử thật: đếm và phân tích theo ngày
Tạo một link và mở ba lần qua API. Sự kiện đi EventBridge tới SQS tới aggregator, và bộ đếm lên:
$ for i in 1 2 3; do curl -s -o /dev/null "$API/$CODE"; done
$ aws dynamodb get-item --table-name url-shortener \
--key '{"PK":{"S":"LINK#7knfT08"},"SK":{"S":"META"}}' --query 'Item.clicks.N'
"3"
Vì aggregator cộng cả STAT#<ngày>, ta có sẵn số liệu theo ngày. Bảng dưới là ảnh chụp item collection của link sau khi đã chạy nốt phần idempotency ở mục kế (mục đó gửi thêm một sự kiện gắn ngày 2026-05-26), nên bộ đếm tổng đã là 4 và có thêm dòng STAT#2026-05-26:
$ aws dynamodb query --table-name url-shortener \
--key-condition-expression "PK = :pk" \
--expression-attribute-values '{":pk":{"S":"LINK#7knfT08"}}' \
--query 'Items[].{SK:SK.S,clicks:clicks.N,count:count.N}' --output table
+------------------+----------+--------+
| SK | clicks | count |
+------------------+----------+--------+
| META | 4 | None |
| STAT#2026-05-25 | None | 3 |
| STAT#2026-05-26 | None | 1 |
+------------------+----------+--------+
Bộ đếm ngày dùng phần ngày (UTC) của thời điểm click làm sort key, nên dashboard sau này vẽ biểu đồ theo thời gian chỉ là query item collection rồi lọc các dòng STAT#.
Thử thật: idempotency
Để buộc tình huống "cùng một sự kiện tới hai lần", gửi thẳng vào hàng đợi một message có id cố định, hai lần:
BODY='{"id":"fixed-test-id-xyz","detail":{"code":"7knfT08","at":"2026-05-26T10:00:00Z"}}'
aws sqs send-message --queue-url "$QURL" --message-body "$BODY"
aws sqs send-message --queue-url "$QURL" --message-body "$BODY"
Bộ đếm tổng chỉ tăng đúng một, từ 3 lên 4, dù gửi hai lần. Và trong bảng trên, STAT#2026-05-26 bằng 1 chứ không phải 2. Lần thứ hai bị marker CLICK#fixed-test-id-xyz chặn ngay trong transaction, nên không cộng. Đếm trùng bị loại tại gốc.
Thử thật: dead-letter queue
Gửi một sự kiện hỏng, thiếu code, để handler ném lỗi:
aws sqs send-message --queue-url "$QURL" \
--message-body '{"id":"bad-event-1","detail":{"at":"2026-05-25T10:00:00Z"}}'
Theo dõi DLQ, message lỗi xuất hiện sau khoảng 90 giây, đúng bằng ba lần thử cách nhau một chu kỳ visibility 30 giây:
t=80s: DLQ có 0 message
t=90s: DLQ có 1 message
$ aws sqs receive-message --queue-url "$DLQURL" --query 'Messages[0].Body'
{"id":"bad-event-1","detail":{"at":"2026-05-25T10:00:00Z"}}
Sự kiện hỏng không kẹt mãi ở hàng chính, cũng không biến mất. Sau ba lần thử thất bại, SQS chuyển nó nguyên vẹn sang DLQ, nơi ta có thể đọc lại để tìm nguyên nhân, sửa, rồi đẩy lại nếu cần. Còn những sự kiện tốt vẫn được xử lý bình thường suốt thời gian đó vì lỗi được báo theo từng message.
🧹 Dọn dẹp
# xoa link demo + cac STAT cua no
for sk in META STAT#2026-05-25 STAT#2026-05-26; do
aws dynamodb delete-item --table-name url-shortener \
--key "{\"PK\":{\"S\":\"LINK#$CODE\"},\"SK\":{\"S\":\"$sk\"}}"
done
aws cognito-idp admin-delete-user --user-pool-id "$POOL" --username agg@example.com
aws sqs purge-queue --queue-url "$DLQURL"
Marker CLICK#<id> còn lại tự hết hạn qua TTL. Giữ stack cho bài sau.
Tổng kết
Consumer giờ là một aggregator thật. Chèn SQS vào giữa cho ta batch, retry và DLQ. Đếm bằng transaction ghép việc cộng bộ đếm với một marker idempotency, nên sự kiện giao trùng không đếm sai. Báo lỗi theo từng message để chỉ thử lại cái hỏng, và sau số lần định trước thì sự kiện lỗi rơi vào DLQ thay vì biến mất. Bộ đếm tổng và bộ đếm theo ngày cùng nằm trong item collection của link, sẵn cho dashboard.
Dữ liệu click giờ chính xác và có theo ngày, nhưng dashboard vẫn phải hỏi lại server mới biết số mới. Bài sau làm phần realtime: dùng WebSocket API của API Gateway để khi một lượt click được đếm, server đẩy số mới xuống trình duyệt đang mở dashboard, không cần tải lại trang.