From 1373b9e548793e28fb0f060886c4ec55c24cdd16 Mon Sep 17 00:00:00 2001 From: riotpiaole <19826264+Riotpiaole@users.noreply.github.com> Date: Sun, 21 Jun 2026 17:08:48 -0700 Subject: [PATCH] feat(reaper): add redelivery/DLQ sweep Per-queue ticker that scans vis_index for expired in-flight messages and drives each through the atomic reap.lua check-and-act, routing maxed-out messages to their queue's configured DLQ via the existing SendMessage path. Safe to run from multiple replicas against the same queue. --- internal/core/reaper/reaper.go | 116 +++++++++++++ internal/core/reaper/reaper_test.go | 254 ++++++++++++++++++++++++++++ 2 files changed, 370 insertions(+) create mode 100644 internal/core/reaper/reaper.go create mode 100644 internal/core/reaper/reaper_test.go diff --git a/internal/core/reaper/reaper.go b/internal/core/reaper/reaper.go new file mode 100644 index 0000000..3464cd8 --- /dev/null +++ b/internal/core/reaper/reaper.go @@ -0,0 +1,116 @@ +// Package reaper implements the goroutine-based redelivery/DLQ sweep from +// design.md §5: a per-queue ticker scans vis_index for expired in-flight +// messages and drives each through the atomic reap.lua check-and-act. +package reaper + +import ( + "context" + "fmt" + "time" + + goredis "github.com/redis/go-redis/v9" + + "github.com/rockliang/kafka-management-service/internal/core/queue" + kmsvcredis "github.com/rockliang/kafka-management-service/internal/redis" +) + +const ( + defaultSweepInterval = 5 * time.Second + sweepBatchSize = 100 +) + +// DLQSender is the subset of queue.SendMessageService a Reaper needs to +// route a maxed-out message to its queue's dead-letter target. Abstracted so +// tests can substitute a fake without a real Kafka broker. +type DLQSender interface { + SendMessage(ctx context.Context, in queue.SendMessageInput) (queue.SendMessageOutput, error) +} + +// Reaper sweeps one or more queues' vis_index ZSETs for expired in-flight +// messages. Safe to run concurrently from multiple replicas against the same +// queue: reap.lua's check-and-act is atomic per receipt handle, so only the +// first caller for a given expired entry gets a non-"gone" outcome. +type Reaper struct { + Redis *goredis.Client + DLQSender DLQSender + + // Interval is the sleep between sweep ticks for Run; defaults to 5s. + Interval time.Duration +} + +// Sweep runs one pass over queueName's vis_index, returning the number of +// expired entries it took action on (redelivered or DLQ-routed; entries +// another replica already won the race for don't count). +func (r *Reaper) Sweep(ctx context.Context, queueName string) (int, error) { + meta, ok, err := kmsvcredis.GetQueueMeta(ctx, r.Redis, queueName) + if err != nil { + return 0, fmt.Errorf("reaper sweep %s: %w", queueName, err) + } + if !ok { + return 0, nil + } + + handles, err := kmsvcredis.ExpiredVisIndexEntries(ctx, r.Redis, queueName, sweepBatchSize) + if err != nil { + return 0, fmt.Errorf("reaper sweep %s: %w", queueName, err) + } + + swept := 0 + for _, handle := range handles { + result, err := kmsvcredis.Reap(ctx, r.Redis, queueName, handle, meta.MaxReceiveCount) + if err != nil { + return swept, fmt.Errorf("reaper sweep %s: %w", queueName, err) + } + + switch result.Outcome { + case kmsvcredis.ReapOutcomeGone: + continue + case kmsvcredis.ReapOutcomeDLQ: + if err := r.routeToDLQ(ctx, meta, result); err != nil { + return swept, fmt.Errorf("reaper sweep %s: %w", queueName, err) + } + swept++ + case kmsvcredis.ReapOutcomeRedeliver: + swept++ + } + } + return swept, nil +} + +// routeToDLQ produces a maxed-out message onto its queue's configured +// dead-letter queue. A queue with no DLQ configured just drops the message +// here, matching reap.lua having already deleted all of its state. +func (r *Reaper) routeToDLQ(ctx context.Context, meta kmsvcredis.QueueMeta, result kmsvcredis.ReapResult) error { + if meta.DLQQueueName == "" || r.DLQSender == nil { + return nil + } + _, err := r.DLQSender.SendMessage(ctx, queue.SendMessageInput{ + QueueName: meta.DLQQueueName, + Body: result.Body, + MessageGroupID: result.GroupID, + }) + return err +} + +// Run sweeps queueName on a ticker until ctx is done. Intended to be started +// as its own goroutine per queue (design.md §5); sweep errors are logged-ish +// via the returned channel-free design (transient Redis/Kafka hiccups are +// retried on the next tick rather than stopping the loop). +func (r *Reaper) Run(ctx context.Context, queueName string, onSweepError func(error)) { + interval := r.Interval + if interval <= 0 { + interval = defaultSweepInterval + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if _, err := r.Sweep(ctx, queueName); err != nil && onSweepError != nil { + onSweepError(err) + } + } + } +} diff --git a/internal/core/reaper/reaper_test.go b/internal/core/reaper/reaper_test.go new file mode 100644 index 0000000..066ae3b --- /dev/null +++ b/internal/core/reaper/reaper_test.go @@ -0,0 +1,254 @@ +package reaper + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + goredis "github.com/redis/go-redis/v9" + "github.com/twmb/franz-go/pkg/kfake" + + "github.com/rockliang/kafka-management-service/internal/core/queue" + "github.com/rockliang/kafka-management-service/internal/kafka" + kmsvcredis "github.com/rockliang/kafka-management-service/internal/redis" +) + +// newTestKafka mirrors internal/core/queue's test setup: an in-memory, +// wire-protocol-compatible fake Kafka cluster (kfake) instead of +// testcontainers/Docker, exercising the real franz-go producer/consumer/ +// admin code paths the DLQ-routing path depends on. +func newTestKafka(t *testing.T) []string { + t.Helper() + cluster, err := kfake.NewCluster(kfake.NumBrokers(1)) + if err != nil { + t.Fatalf("starting kfake cluster: %v", err) + } + t.Cleanup(cluster.Close) + return cluster.ListenAddrs() +} + +func newTestRedisClient(t *testing.T) *goredis.Client { + t.Helper() + mr, err := miniredis.Run() + if err != nil { + t.Fatalf("starting miniredis: %v", err) + } + t.Cleanup(mr.Close) + return goredis.NewClient(&goredis.Options{Addr: mr.Addr()}) +} + +// seedQueue creates a queue's shard-0 topic, queue meta, and shard map +// directly — standing in for the queue-operator, exercised in its own test +// suite (internal/operator). +func seedQueue(t *testing.T, ctx context.Context, admin *kafka.Admin, rdb *goredis.Client, name string, maxReceiveCount int32, dlqName string) kafka.Shard { + t.Helper() + shard := kafka.Shard{ID: "0", Topic: kafka.ShardTopicName(name, false, "0"), HashRangeStart: 0, HashRangeEnd: kafka.FullHashRangeEnd, Phase: "Active"} + if err := admin.CreateTopic(ctx, shard.Topic, kafka.TopicConfig{PartitionCount: 1, ReplicationFactor: 1, RetentionSeconds: 345600, MinInsyncReplicas: 1}); err != nil { + t.Fatalf("create shard topic: %v", err) + } + if err := kmsvcredis.PutQueueMeta(ctx, rdb, name, kmsvcredis.QueueMeta{ + VisibilityTimeoutSeconds: 30, + MaxReceiveCount: maxReceiveCount, + DLQQueueName: dlqName, + PartitionsPerShard: 1, + RetentionSeconds: 345600, + }); err != nil { + t.Fatalf("put queue meta: %v", err) + } + if err := kmsvcredis.PutShardMap(ctx, rdb, name, []kafka.Shard{shard}); err != nil { + t.Fatalf("put shard map: %v", err) + } + return shard +} + +// putExpiredInFlight records an in-flight message whose visibility deadline +// is already in the past, so it's immediately eligible for the sweep — +// standing in for ReceiveMessage handing out a message and the caller never +// acking it. PutInFlight is called with a normal positive visibility timeout +// (it sets the in-flight hash's TTL to timeout+1min; a negative timeout +// would give a negative TTL, which deletes the hash immediately rather than +// leaving it in place for the reaper to find) and the vis_index score is +// then backdated directly to simulate the deadline having already elapsed. +func putExpiredInFlight(t *testing.T, ctx context.Context, rdb *goredis.Client, queueName, receiptHandle string, rec kmsvcredis.InFlightRecord) { + t.Helper() + if err := kmsvcredis.PutInFlight(ctx, rdb, queueName, receiptHandle, rec, time.Minute); err != nil { + t.Fatalf("put expired in-flight: %v", err) + } + expired := time.Now().Add(-time.Hour).UnixMilli() + if err := rdb.ZAdd(ctx, kmsvcredis.VisIndexKey(queueName), goredis.Z{Score: float64(expired), Member: receiptHandle}).Err(); err != nil { + t.Fatalf("backdate vis_index: %v", err) + } +} + +func TestSweepRoutesMaxedOutMessageToDLQ(t *testing.T) { + ctx := context.Background() + brokers := newTestKafka(t) + rdb := newTestRedisClient(t) + + admin, err := kafka.NewAdmin(brokers) + if err != nil { + t.Fatalf("new admin: %v", err) + } + t.Cleanup(admin.Close) + + shard := seedQueue(t, ctx, admin, rdb, "orders", 1, "orders-dlq") + dlqShard := seedQueue(t, ctx, admin, rdb, "orders-dlq", 3, "") + + producer, err := kafka.NewProducer(brokers) + if err != nil { + t.Fatalf("new producer: %v", err) + } + t.Cleanup(producer.Close) + router := &queue.ShardRouter{Redis: rdb} + sender := &queue.SendMessageService{Redis: rdb, Producer: producer, Router: router} + + putExpiredInFlight(t, ctx, rdb, "orders", "h1", kmsvcredis.InFlightRecord{ + ShardID: shard.ID, Topic: shard.Topic, Partition: 0, Offset: 0, + ReceiveCount: 1, Body: "maxed-out", + }) + + r := &Reaper{Redis: rdb, DLQSender: sender} + swept, err := r.Sweep(ctx, "orders") + if err != nil { + t.Fatalf("sweep: %v", err) + } + if swept != 1 { + t.Fatalf("swept = %d, want 1", swept) + } + + // The in-flight record and vis_index entry must be fully cleaned up. + if _, ok, err := kmsvcredis.GetInFlight(ctx, rdb, "orders", "h1"); err != nil || ok { + t.Fatalf("in-flight record still present after DLQ-route: ok=%v err=%v", ok, err) + } + + consumer, err := kafka.NewConsumer(brokers, "verify-dlq", dlqShard.Topic) + if err != nil { + t.Fatalf("new consumer: %v", err) + } + t.Cleanup(consumer.Close) + pollCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + var body string + for body == "" { + records, err := consumer.Poll(pollCtx) + if err != nil { + t.Fatalf("poll dlq topic: %v", err) + } + for _, rec := range records { + body = string(rec.Value) + } + if pollCtx.Err() != nil { + break + } + } + if body != "maxed-out" { + t.Fatalf("dlq message body = %q, want %q", body, "maxed-out") + } +} + +func TestSweepRedeliversWhenUnderMaxReceiveCount(t *testing.T) { + ctx := context.Background() + brokers := newTestKafka(t) + rdb := newTestRedisClient(t) + + admin, err := kafka.NewAdmin(brokers) + if err != nil { + t.Fatalf("new admin: %v", err) + } + t.Cleanup(admin.Close) + + shard := seedQueue(t, ctx, admin, rdb, "orders", 3, "") + putExpiredInFlight(t, ctx, rdb, "orders", "h1", kmsvcredis.InFlightRecord{ + ShardID: shard.ID, Topic: shard.Topic, Partition: 0, Offset: 0, + ReceiveCount: 1, Body: "retry-me", + }) + + r := &Reaper{Redis: rdb} + swept, err := r.Sweep(ctx, "orders") + if err != nil { + t.Fatalf("sweep: %v", err) + } + if swept != 1 { + t.Fatalf("swept = %d, want 1", swept) + } + + rec, ok, err := kmsvcredis.GetInFlight(ctx, rdb, "orders", "h1") + if err != nil || !ok { + t.Fatalf("in-flight record missing after redeliver: ok=%v err=%v", ok, err) + } + if rec.ReceiveCount != 2 { + t.Fatalf("receiveCount = %d, want 2", rec.ReceiveCount) + } + + handle, ok, err := kmsvcredis.PopRedeliverable(ctx, rdb, "orders") + if err != nil || !ok || handle != "h1" { + t.Fatalf("redeliverable = (%q, %v), want (%q, true)", handle, ok, "h1") + } +} + +// TestConcurrentSweepsProduceNoDuplicateDLQWrites covers acceptance criterion +// 3 of Task 7 at the reaper level (Task 4's Redis-layer concurrency test +// covers reap.lua's atomicity directly): two reaper instances racing the +// same expired entry must yield exactly one DLQ write. +func TestConcurrentSweepsProduceNoDuplicateDLQWrites(t *testing.T) { + ctx := context.Background() + brokers := newTestKafka(t) + rdb := newTestRedisClient(t) + + admin, err := kafka.NewAdmin(brokers) + if err != nil { + t.Fatalf("new admin: %v", err) + } + t.Cleanup(admin.Close) + + shard := seedQueue(t, ctx, admin, rdb, "orders", 1, "orders-dlq") + seedQueue(t, ctx, admin, rdb, "orders-dlq", 3, "") + + producer, err := kafka.NewProducer(brokers) + if err != nil { + t.Fatalf("new producer: %v", err) + } + t.Cleanup(producer.Close) + router := &queue.ShardRouter{Redis: rdb} + + var mu sync.Mutex + sendCount := 0 + countingSender := countingDLQSender{ + inner: &queue.SendMessageService{Redis: rdb, Producer: producer, Router: router}, + onSend: func() { + mu.Lock() + sendCount++ + mu.Unlock() + }, + } + + putExpiredInFlight(t, ctx, rdb, "orders", "h1", kmsvcredis.InFlightRecord{ + ShardID: shard.ID, Topic: shard.Topic, Partition: 0, Offset: 0, + ReceiveCount: 1, Body: "race-me", + }) + + r1 := &Reaper{Redis: rdb, DLQSender: countingSender} + r2 := &Reaper{Redis: rdb, DLQSender: countingSender} + + var wg sync.WaitGroup + wg.Add(2) + go func() { defer wg.Done(); _, _ = r1.Sweep(ctx, "orders") }() + go func() { defer wg.Done(); _, _ = r2.Sweep(ctx, "orders") }() + wg.Wait() + + if sendCount != 1 { + t.Fatalf("dlq send count = %d, want exactly 1", sendCount) + } +} + +type countingDLQSender struct { + inner DLQSender + onSend func() +} + +func (c countingDLQSender) SendMessage(ctx context.Context, in queue.SendMessageInput) (queue.SendMessageOutput, error) { + c.onSend() + return c.inner.SendMessage(ctx, in) +}