fix(iot): пустое тело сообщения → DLQ (вместо молчаливого подтверждения)
Убран ранний дроп пустого тела; теперь оно идёт через processTelemetry → errInvalidEnvelope → DLQ (или дроп после maxReceiveCount, если DLQ недоступна).
This commit is contained in:
@@ -100,12 +100,14 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *
|
|||||||
}
|
}
|
||||||
var pending []pendingAck
|
var pending []pendingAck
|
||||||
for _, msg := range resp.Messages {
|
for _, msg := range resp.Messages {
|
||||||
if msg.Body == nil || *msg.Body == "" {
|
// nil/пустое тело не подтверждаем молча: прогоняем через processTelemetry,
|
||||||
log.Warn("consumer: empty message body", "message_id", aws.ToString(msg.MessageId))
|
// который вернёт errInvalidEnvelope → DLQ (ревью 2026-10-10: пустое тело
|
||||||
pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)})
|
// подтверждалось без DLQ).
|
||||||
continue
|
body := ""
|
||||||
|
if msg.Body != nil {
|
||||||
|
body = *msg.Body
|
||||||
}
|
}
|
||||||
if err := processTelemetry(ctx, *msg.Body, store, log); err != nil {
|
if err := processTelemetry(ctx, body, store, log); err != nil {
|
||||||
// Битый envelope → DLQ (а не бесконечный ретрай и не потеря):
|
// Битый envelope → DLQ (а не бесконечный ретрай и не потеря):
|
||||||
// кладём тело в DLQ, затем подтверждаем основное сообщение.
|
// кладём тело в DLQ, затем подтверждаем основное сообщение.
|
||||||
if errors.Is(err, errInvalidEnvelope) {
|
if errors.Is(err, errInvalidEnvelope) {
|
||||||
@@ -115,7 +117,7 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *
|
|||||||
// зациклиться навсегда (находка #2): после maxReceiveCount
|
// зациклиться навсегда (находка #2): после maxReceiveCount
|
||||||
// дропаем битое с громким логом.
|
// дропаем битое с громким логом.
|
||||||
if dlqURL != "" {
|
if dlqURL != "" {
|
||||||
if derr := sendToDLQ(ctx, sqsClient, dlqURL, *msg.Body, log); derr == nil {
|
if derr := sendToDLQ(ctx, sqsClient, dlqURL, body, log); derr == nil {
|
||||||
pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)})
|
pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)})
|
||||||
continue
|
continue
|
||||||
} else {
|
} else {
|
||||||
|
|||||||
Reference in New Issue
Block a user