From 2dcb9bd0fc56329d2b3996ce7895edc092a69cd4 Mon Sep 17 00:00:00 2001 From: coderconnoisseur Date: Fri, 16 Oct 2026 00:14:08 +0530 Subject: [PATCH 1/9] pkg: epoch-stamped rewind handling and handler panic recovery --- pkg/kafkaconsumer/runner.go | 42 ++++++++++---- pkg/kafkaconsumer/runner_test.go | 38 +++++++++++++ pkg/kafkaconsumer/tracker.go | 41 ++++++++++--- pkg/kafkaconsumer/tracker_test.go | 95 +++++++++++++++++++++++++++---- 4 files changed, 187 insertions(+), 29 deletions(-) diff --git a/pkg/kafkaconsumer/runner.go b/pkg/kafkaconsumer/runner.go index 0f92b47..924d99c 100644 --- a/pkg/kafkaconsumer/runner.go +++ b/pkg/kafkaconsumer/runner.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "log" + "runtime/debug" "strconv" "sync" "time" @@ -56,6 +57,12 @@ type Config struct { OnDLQ func(msg kafka.Message, err error) } +// job is a fetched message plus the tracker epoch it was tracked in. +type job struct { + msg kafka.Message + epoch uint64 +} + // runner holds the state shared by the fetch loop and the workers of one Run. type runner struct { r Reader @@ -97,14 +104,14 @@ func Run(ctx context.Context, r Reader, cfg Config, h Handler) error { stopFetchFn: stopFetch, } - work := make(chan kafka.Message) // unbuffered: backpressure on fetching + work := make(chan job) // unbuffered: backpressure on fetching var wg sync.WaitGroup for range cfg.Workers { wg.Add(1) go func() { defer wg.Done() - for msg := range work { - rn.process(msg) + for j := range work { + rn.process(j) } }() } @@ -163,7 +170,7 @@ func (c *Config) applyDefaults() { } } -func (rn *runner) fetchLoop(ctx context.Context, work chan<- kafka.Message) { +func (rn *runner) fetchLoop(ctx context.Context, work chan<- job) { for { msg, err := rn.r.FetchMessage(ctx) if err != nil { @@ -172,9 +179,9 @@ func (rn *runner) fetchLoop(ctx context.Context, work chan<- kafka.Message) { } return } - rn.tracker.track(msg) + epoch := rn.tracker.track(msg) select { - case work <- msg: + case work <- job{msg, epoch}: case <-ctx.Done(): // Tracked but never handed off: it stays pending, which blocks // commits past it, so it is redelivered on restart. @@ -204,7 +211,8 @@ func (rn *runner) err() error { // process drives one message to a terminal state and commits what that // unblocks. If the message cannot reach a terminal state (drain timeout, or // the DLQ write failed) it is left uncommitted. -func (rn *runner) process(msg kafka.Message) { +func (rn *runner) process(j job) { + msg := j.msg err := rn.handleWithRetry(msg) if err != nil { if rn.hctx.Err() != nil { @@ -214,12 +222,12 @@ func (rn *runner) process(msg kafka.Message) { return } } - rn.complete(msg) + rn.complete(msg, j.epoch) } func (rn *runner) handleWithRetry(msg kafka.Message) error { for attempt := 1; ; attempt++ { - err := rn.h(rn.hctx, msg) + err := rn.handle(msg) if err == nil || IsPermanent(err) || attempt > rn.cfg.MaxRetries || rn.hctx.Err() != nil { return err } @@ -232,6 +240,18 @@ func (rn *runner) handleWithRetry(msg kafka.Message) error { } } +// handle runs one handler attempt. A panic becomes a permanent error, so the +// message is dead-lettered instead of crashing the process. +func (rn *runner) handle(msg kafka.Message) (err error) { + defer func() { + if r := recover(); r != nil { + log.Printf("kafkaconsumer: handler panic on %s/%d@%d: %v\n%s", msg.Topic, msg.Partition, msg.Offset, r, debug.Stack()) + err = Permanent(fmt.Errorf("handler panic: %v", r)) + } + }() + return rn.h(rn.hctx, msg) +} + // deadLetter routes a failed message to the DLQ (or drops it when no DLQ is // configured). It reports whether the message reached a terminal state. func (rn *runner) deadLetter(msg kafka.Message, cause error) bool { @@ -297,13 +317,13 @@ func (rn *runner) sleep(attempt int) bool { // complete marks msg terminal and commits the highest offset that became // committable. Holding commitMu across done+commit keeps commits serialized // and per-partition monotonic: a later done() can only pop higher offsets. -func (rn *runner) complete(msg kafka.Message) { +func (rn *runner) complete(msg kafka.Message, epoch uint64) { rn.commitMu.Lock() defer rn.commitMu.Unlock() if rn.stopped { return } - toCommit, ok := rn.tracker.done(msg) + toCommit, ok := rn.tracker.done(msg, epoch) if !ok { return } diff --git a/pkg/kafkaconsumer/runner_test.go b/pkg/kafkaconsumer/runner_test.go index 914dcdb..16a75aa 100644 --- a/pkg/kafkaconsumer/runner_test.go +++ b/pkg/kafkaconsumer/runner_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "strconv" + "strings" "sync" "sync/atomic" "testing" @@ -319,6 +320,43 @@ func TestRunPermanentErrorSkipsRetries(t *testing.T) { } } +func TestRunHandlerPanicGoesToDLQAndRunContinues(t *testing.T) { + r := newFakeReader(seq("t", 3)...) + dlq := &fakeWriter{} + var panics atomic.Int32 + cancel, errc := startRun(r, Config{Workers: 1, MaxRetries: 3, RetryBackoff: time.Millisecond, DLQ: dlq, DLQTopic: "t.dlq"}, + func(_ context.Context, m kafka.Message) error { + if m.Offset == 0 { + panics.Add(1) + panic("nil map write") + } + return nil + }) + + r.waitCommitted(t, 2) + cancel() + if err := waitRun(t, errc); err != nil { + t.Fatalf("Run returned %v, want nil", err) + } + if got := panics.Load(); got != 1 { + t.Fatalf("panicking handler called %d times, want 1", got) + } + w := dlq.written() + if len(w) != 1 || string(w[0].Key) != "k0" { + t.Fatalf("DLQ = %d messages, want offset 0's message", len(w)) + } + var xerr string + for _, h := range w[0].Headers { + if h.Key == "x-error" { + xerr = string(h.Value) + } + } + if !strings.Contains(xerr, "handler panic") { + t.Fatalf("x-error = %q, want it to contain %q", xerr, "handler panic") + } + assertMonotonic(t, r.committed()) +} + func TestRunNilDLQDropsAndCommits(t *testing.T) { r := newFakeReader(seq("t", 1)...) var dlqHook atomic.Int32 diff --git a/pkg/kafkaconsumer/tracker.go b/pkg/kafkaconsumer/tracker.go index 4bfcd6d..d5f1fe4 100644 --- a/pkg/kafkaconsumer/tracker.go +++ b/pkg/kafkaconsumer/tracker.go @@ -1,6 +1,9 @@ package kafkaconsumer import ( + "cmp" + "log" + "slices" "sync" "github.com/segmentio/kafka-go" @@ -16,6 +19,10 @@ type partitionKey struct { type partitionState struct { pending []kafka.Message done map[int64]struct{} + // epoch advances on every rewind; completions carrying an older epoch + // belong to copies that were superseded by a re-fetch. + epoch uint64 + last int64 // highest offset tracked in the current epoch, -1 if none } // offsetTracker decides which offset is safe to commit: a message may only @@ -30,28 +37,48 @@ func newOffsetTracker() *offsetTracker { return &offsetTracker{parts: make(map[partitionKey]*partitionState)} } -// track records a fetched message. Messages of a partition must be tracked -// in fetch (increasing offset) order. -func (t *offsetTracker) track(msg kafka.Message) { +// track records a fetched message and returns the epoch to pass to done. +// An offset at or below the last tracked one means the reader rewound (e.g. +// after a consumer-group rebalance it re-fetches from the committed offset): +// the partition starts a new epoch and its old pending state is discarded. +func (t *offsetTracker) track(msg kafka.Message) uint64 { t.mu.Lock() defer t.mu.Unlock() key := partitionKey{msg.Topic, msg.Partition} ps, ok := t.parts[key] if !ok { - ps = &partitionState{done: make(map[int64]struct{})} + ps = &partitionState{done: make(map[int64]struct{}), last: -1} t.parts[key] = ps } + if msg.Offset <= ps.last { + log.Printf("kafkaconsumer: %s/%d rewound from %d to %d; ignoring in-flight completions of the old fetch", + msg.Topic, msg.Partition, ps.last, msg.Offset) + ps.epoch++ + clear(ps.pending) // release payloads for GC + ps.pending = ps.pending[:0] + clear(ps.done) + } ps.pending = append(ps.pending, msg) + ps.last = msg.Offset + return ps.epoch } // done marks msg finished and pops the contiguous finished prefix of its // partition. It returns the last popped message — the one to commit — and -// true, or false when msg's predecessors are still in flight. -func (t *offsetTracker) done(msg kafka.Message) (kafka.Message, bool) { +// true, or false when msg's predecessors are still in flight. A completion +// from an older epoch, or for an offset not pending, changes nothing: after +// a rewind the re-fetched copy is processed again and commits instead. +func (t *offsetTracker) done(msg kafka.Message, epoch uint64) (kafka.Message, bool) { t.mu.Lock() defer t.mu.Unlock() ps, ok := t.parts[partitionKey{msg.Topic, msg.Partition}] - if !ok { + if !ok || epoch != ps.epoch { + return kafka.Message{}, false + } + // pending is in increasing offset order within an epoch. + if _, found := slices.BinarySearchFunc(ps.pending, msg.Offset, func(m kafka.Message, off int64) int { + return cmp.Compare(m.Offset, off) + }); !found { return kafka.Message{}, false } ps.done[msg.Offset] = struct{}{} diff --git a/pkg/kafkaconsumer/tracker_test.go b/pkg/kafkaconsumer/tracker_test.go index df67049..b7aa133 100644 --- a/pkg/kafkaconsumer/tracker_test.go +++ b/pkg/kafkaconsumer/tracker_test.go @@ -13,11 +13,12 @@ func msgAt(topic string, partition int, offset int64) kafka.Message { func TestTrackerInOrderCompletionCommitsEach(t *testing.T) { tr := newOffsetTracker() + var e uint64 for i := int64(0); i < 3; i++ { - tr.track(msgAt("t", 0, i)) + e = tr.track(msgAt("t", 0, i)) } for i := int64(0); i < 3; i++ { - got, ok := tr.done(msgAt("t", 0, i)) + got, ok := tr.done(msgAt("t", 0, i), e) if !ok || got.Offset != i { t.Fatalf("done(%d) = (%d, %v), want (%d, true)", i, got.Offset, ok, i) } @@ -27,12 +28,12 @@ func TestTrackerInOrderCompletionCommitsEach(t *testing.T) { func TestTrackerOutOfOrderCompletionWaitsForPredecessor(t *testing.T) { tr := newOffsetTracker() tr.track(msgAt("t", 0, 1)) - tr.track(msgAt("t", 0, 2)) + e := tr.track(msgAt("t", 0, 2)) - if got, ok := tr.done(msgAt("t", 0, 2)); ok { + if got, ok := tr.done(msgAt("t", 0, 2), e); ok { t.Fatalf("done(2) before 1 returned commit for offset %d, want nothing", got.Offset) } - got, ok := tr.done(msgAt("t", 0, 1)) + got, ok := tr.done(msgAt("t", 0, 1), e) if !ok || got.Offset != 2 { t.Fatalf("done(1) = (%d, %v), want (2, true)", got.Offset, ok) } @@ -41,14 +42,14 @@ func TestTrackerOutOfOrderCompletionWaitsForPredecessor(t *testing.T) { func TestTrackerPartitionsAreIndependent(t *testing.T) { tr := newOffsetTracker() tr.track(msgAt("t", 0, 10)) // stays pending - tr.track(msgAt("t", 1, 20)) - tr.track(msgAt("u", 0, 30)) + e1 := tr.track(msgAt("t", 1, 20)) + eu := tr.track(msgAt("u", 0, 30)) - got, ok := tr.done(msgAt("t", 1, 20)) + got, ok := tr.done(msgAt("t", 1, 20), e1) if !ok || got.Partition != 1 || got.Offset != 20 { t.Fatalf("partition 1 blocked by partition 0: got (%d/%d, %v)", got.Partition, got.Offset, ok) } - got, ok = tr.done(msgAt("u", 0, 30)) + got, ok = tr.done(msgAt("u", 0, 30), eu) if !ok || got.Topic != "u" || got.Offset != 30 { t.Fatalf("topic u blocked by topic t: got (%s/%d, %v)", got.Topic, got.Offset, ok) } @@ -57,8 +58,9 @@ func TestTrackerPartitionsAreIndependent(t *testing.T) { func TestTrackerConcurrentUse(t *testing.T) { tr := newOffsetTracker() const n = 200 + var e uint64 for i := int64(0); i < n; i++ { - tr.track(msgAt("t", 0, i)) + e = tr.track(msgAt("t", 0, i)) } var ( @@ -70,7 +72,7 @@ func TestTrackerConcurrentUse(t *testing.T) { wg.Add(1) go func(off int64) { defer wg.Done() - if m, ok := tr.done(msgAt("t", 0, off)); ok { + if m, ok := tr.done(msgAt("t", 0, off), e); ok { mu.Lock() if m.Offset > highest { highest = m.Offset @@ -84,3 +86,74 @@ func TestTrackerConcurrentUse(t *testing.T) { t.Fatalf("highest committable offset = %d, want %d", highest, n-1) } } + +// partState returns the tracker's internal state for partition 0 of topic t. +func partState(t *testing.T, tr *offsetTracker) *partitionState { + t.Helper() + ps, ok := tr.parts[partitionKey{"t", 0}] + if !ok { + t.Fatal("partition t/0 not tracked") + } + return ps +} + +func TestTrackerRewindIgnoresStaleCompletions(t *testing.T) { + tr := newOffsetTracker() + var e0 uint64 + for i := int64(0); i < 3; i++ { + e0 = tr.track(msgAt("t", 0, i)) + } + // Generation change: the broker re-delivers from committed offset 1. + e1 := tr.track(msgAt("t", 0, 1)) + if e1 == e0 { + t.Fatalf("rewind kept epoch %d, want a new one", e0) + } + + for _, off := range []int64{1, 2} { + if got, ok := tr.done(msgAt("t", 0, off), e0); ok { + t.Fatalf("stale done(%d, e0) committed offset %d, want nothing", off, got.Offset) + } + if ps := partState(t, tr); len(ps.done) != 0 { + t.Fatalf("stale done(%d, e0) left done entries %v", off, ps.done) + } + } + + got, ok := tr.done(msgAt("t", 0, 1), e1) + if !ok || got.Offset != 1 { + t.Fatalf("done(1', e1) = (%d, %v), want (1, true)", got.Offset, ok) + } + if e := tr.track(msgAt("t", 0, 2)); e != e1 { + t.Fatalf("track(2') epoch = %d, want %d (no rewind)", e, e1) + } + got, ok = tr.done(msgAt("t", 0, 2), e1) + if !ok || got.Offset != 2 { + t.Fatalf("done(2', e1) = (%d, %v), want (2, true)", got.Offset, ok) + } + + ps := partState(t, tr) + if len(ps.done) != 0 || len(ps.pending) != 0 { + t.Fatalf("leftover state: pending %v, done %v", offsets(ps.pending), ps.done) + } +} + +func TestTrackerDoneUnknownOffsetIsNoOp(t *testing.T) { + tr := newOffsetTracker() + tr.track(msgAt("t", 0, 5)) + e := tr.track(msgAt("t", 0, 7)) + + for _, off := range []int64{3, 6, 9} { + if got, ok := tr.done(msgAt("t", 0, off), e); ok { + t.Fatalf("done(%d) of untracked offset committed %d, want nothing", off, got.Offset) + } + } + ps := partState(t, tr) + if len(ps.done) != 0 { + t.Fatalf("untracked done created entries %v", ps.done) + } + if got := offsets(ps.pending); len(got) != 2 || got[0] != 5 || got[1] != 7 { + t.Fatalf("pending = %v, want [5 7]", got) + } + if got, ok := tr.done(msgAt("u", 0, 0), 0); ok { + t.Fatalf("done on unknown partition committed %d, want nothing", got.Offset) + } +} From 0c4fe3f945acadd03ac7095939e88cef9b7dbc52 Mon Sep 17 00:00:00 2001 From: coderconnoisseur Date: Fri, 16 Oct 2026 00:19:55 +0530 Subject: [PATCH 2/9] services: env-var configuration; matcher pool x25 and 500ms deadline defaults --- services/api-gateway/cmd/api-gateway/main.go | 52 +++++++------ services/api-gateway/internal/app/app.go | 11 ++- services/api-gateway/internal/app/config.go | 46 +++++++++++ .../api-gateway/internal/app/config_test.go | 36 +++++++++ .../cmd/assignment-service/main.go | 8 +- .../assignment-service/internal/app/app.go | 10 +-- .../assignment-service/internal/app/config.go | 46 +++++++++++ .../internal/app/config_test.go | 36 +++++++++ .../cmd/matching-service/main.go | 8 +- services/matching-service/internal/app/app.go | 32 ++++---- .../matching-service/internal/app/config.go | 77 +++++++++++++++++++ .../internal/app/config_test.go | 57 ++++++++++++++ .../internal/kafka/consumer.go | 12 +-- .../cmd/pricing-service/main.go | 8 +- services/pricing-service/internal/app/app.go | 11 ++- .../pricing-service/internal/app/config.go | 46 +++++++++++ .../internal/app/config_test.go | 36 +++++++++ 17 files changed, 467 insertions(+), 65 deletions(-) create mode 100644 services/api-gateway/internal/app/config.go create mode 100644 services/api-gateway/internal/app/config_test.go create mode 100644 services/assignment-service/internal/app/config.go create mode 100644 services/assignment-service/internal/app/config_test.go create mode 100644 services/matching-service/internal/app/config.go create mode 100644 services/matching-service/internal/app/config_test.go create mode 100644 services/pricing-service/internal/app/config.go create mode 100644 services/pricing-service/internal/app/config_test.go diff --git a/services/api-gateway/cmd/api-gateway/main.go b/services/api-gateway/cmd/api-gateway/main.go index 56ca3b4..02e548b 100644 --- a/services/api-gateway/cmd/api-gateway/main.go +++ b/services/api-gateway/cmd/api-gateway/main.go @@ -1,40 +1,44 @@ package main import ( - // "fmt" "context" + "log" + "net/http" "os" "os/signal" - "ridepulse/services/api-gateway/internal/app" "time" - // "github.com/gin-gonic/gin" - "net/http" - // "golang.org/x/tools/go/analysis/passes/defers" + "ridepulse/services/api-gateway/internal/app" ) -func main(){ - r:=app.NewApp() - - serv:=&http.Server{ - Addr:":8080", - Handler:r, - + +func main() { + cfg, err := app.LoadConfig(os.Getenv) + if err != nil { + log.Fatalf("api-gateway: invalid configuration: %v", err) } - go func(){//start server in a goroutine - if err:=serv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Printf("api-gateway config: http=%s brokers=%v", cfg.HTTPAddr, cfg.KafkaBrokers) + + r, closePublisher := app.NewApp(cfg) + + serv := &http.Server{ + Addr: cfg.HTTPAddr, + Handler: r, + } + go func() { // start server in a goroutine + if err := serv.ListenAndServe(); err != nil && err != http.ErrServerClosed { panic(err) } }() - //i dont want server to shut down immediately - //so i am adding a channel to listen for interrupt signal - //it will complete graceful shutdown - - quit:=make(chan os.Signal,1) - signal.Notify(quit,os.Interrupt) - <-quit // signal arrived here + // Block until an interrupt arrives, then shut down gracefully. + quit := make(chan os.Signal, 1) + signal.Notify(quit, os.Interrupt) + <-quit - ctx,cancel:= context.WithTimeout(context.Background(),5*time.Second)//graceful shutdown + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() serv.Shutdown(ctx) - -} \ No newline at end of file + // In-flight handlers have drained; flush and close the Kafka writer. + if err := closePublisher(); err != nil { + log.Printf("api-gateway publisher close: %v", err) + } +} diff --git a/services/api-gateway/internal/app/app.go b/services/api-gateway/internal/app/app.go index 3418a95..5541124 100644 --- a/services/api-gateway/internal/app/app.go +++ b/services/api-gateway/internal/app/app.go @@ -5,7 +5,10 @@ import ( "github.com/gin-gonic/gin" ) -func NewApp() *gin.Engine{ - publisher:=NewKafkaPublisher([]string {"localhost:9092"}) - return server.NewRouter(publisher) -} \ No newline at end of file + +// NewApp wires the router and returns it with a close func that releases the +// Kafka publisher; call it after the HTTP server has shut down. +func NewApp(cfg Config) (*gin.Engine, func() error) { + publisher := NewKafkaPublisher(cfg.KafkaBrokers) + return server.NewRouter(publisher), publisher.Writer.Close +} diff --git a/services/api-gateway/internal/app/config.go b/services/api-gateway/internal/app/config.go new file mode 100644 index 0000000..7610a69 --- /dev/null +++ b/services/api-gateway/internal/app/config.go @@ -0,0 +1,46 @@ +package app + +import ( + "fmt" + "strings" +) + +// Config holds the gateway's environment-derived settings. +type Config struct { + HTTPAddr string + KafkaBrokers []string +} + +// LoadConfig reads HTTP_ADDR and KAFKA_BROKERS via getenv. +func LoadConfig(getenv func(string) string) (Config, error) { + brokers, err := parseBrokers(envOr(getenv, "KAFKA_BROKERS", "localhost:9092")) + if err != nil { + return Config{}, err + } + return Config{ + HTTPAddr: envOr(getenv, "HTTP_ADDR", ":8080"), + KafkaBrokers: brokers, + }, nil +} + +func envOr(getenv func(string) string, key, def string) string { + if v := strings.TrimSpace(getenv(key)); v != "" { + return v + } + return def +} + +// parseBrokers splits a comma-separated broker list, trimming spaces and +// rejecting empty entries. +func parseBrokers(raw string) ([]string, error) { + parts := strings.Split(raw, ",") + brokers := make([]string, 0, len(parts)) + for _, p := range parts { + p = strings.TrimSpace(p) + if p == "" { + return nil, fmt.Errorf("KAFKA_BROKERS %q contains an empty entry", raw) + } + brokers = append(brokers, p) + } + return brokers, nil +} diff --git a/services/api-gateway/internal/app/config_test.go b/services/api-gateway/internal/app/config_test.go new file mode 100644 index 0000000..e119334 --- /dev/null +++ b/services/api-gateway/internal/app/config_test.go @@ -0,0 +1,36 @@ +package app + +import ( + "reflect" + "testing" +) + +func env(m map[string]string) func(string) string { + return func(k string) string { return m[k] } +} + +func TestLoadConfig(t *testing.T) { + tests := []struct { + name string + env map[string]string + want Config + wantErr bool + }{ + {"defaults", nil, Config{HTTPAddr: ":8080", KafkaBrokers: []string{"localhost:9092"}}, false}, + {"overrides", map[string]string{"HTTP_ADDR": ":9000", "KAFKA_BROKERS": "a:1, b:2"}, + Config{HTTPAddr: ":9000", KafkaBrokers: []string{"a:1", "b:2"}}, false}, + {"empty broker entry", map[string]string{"KAFKA_BROKERS": "a:1,,b:2"}, Config{}, true}, + {"trailing comma", map[string]string{"KAFKA_BROKERS": "a:1,"}, Config{}, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := LoadConfig(env(tt.env)) + if (err != nil) != tt.wantErr { + t.Fatalf("err = %v, wantErr %v", err, tt.wantErr) + } + if !tt.wantErr && !reflect.DeepEqual(got, tt.want) { + t.Fatalf("got %+v, want %+v", got, tt.want) + } + }) + } +} diff --git a/services/assignment-service/cmd/assignment-service/main.go b/services/assignment-service/cmd/assignment-service/main.go index 00c0f81..e2c3870 100644 --- a/services/assignment-service/cmd/assignment-service/main.go +++ b/services/assignment-service/cmd/assignment-service/main.go @@ -14,7 +14,13 @@ func main() { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() - if err := app.Run(ctx); err != nil { + cfg, err := app.LoadConfig(os.Getenv) + if err != nil { + log.Fatalf("assignment-service: invalid configuration: %v", err) + } + log.Printf("assignment-service config: http=%s brokers=%v", cfg.HTTPAddr, cfg.KafkaBrokers) + + if err := app.Run(ctx, cfg); err != nil { stop() log.Fatal(err) } diff --git a/services/assignment-service/internal/app/app.go b/services/assignment-service/internal/app/app.go index ff3cbdf..33015f0 100644 --- a/services/assignment-service/internal/app/app.go +++ b/services/assignment-service/internal/app/app.go @@ -15,26 +15,24 @@ import ( ) const ( - kafkaBroker = "localhost:9092" - httpAddr = ":2114" shutdownTimeout = 5 * time.Second ) // Run serves the HTTP API and consumes RideMatched events until ctx is // cancelled, then shuts everything down. -func Run(ctx context.Context) error { +func Run(ctx context.Context, cfg Config) error { metrics.Register() mem := store.NewMemory() - server := httpapi.NewServer(httpAddr, mem) + server := httpapi.NewServer(cfg.HTTPAddr, mem) go func() { - log.Printf("assignment-service listening on %s", httpAddr) + log.Printf("assignment-service listening on %s", cfg.HTTPAddr) if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { log.Printf("assignment-service http server: %v", err) } }() - consumer := kafka.NewKafkaConsumer([]string{kafkaBroker}) + consumer := kafka.NewKafkaConsumer(cfg.KafkaBrokers) runErr := consumer.Run(ctx, func(_ context.Context, event domain.RideMatchedEvent) error { var latencyMs int64 if event.RequestedAt > 0 { diff --git a/services/assignment-service/internal/app/config.go b/services/assignment-service/internal/app/config.go new file mode 100644 index 0000000..a258c4b --- /dev/null +++ b/services/assignment-service/internal/app/config.go @@ -0,0 +1,46 @@ +package app + +import ( + "fmt" + "strings" +) + +// Config holds assignment-service's environment-derived settings. +type Config struct { + HTTPAddr string + KafkaBrokers []string +} + +// LoadConfig reads HTTP_ADDR and KAFKA_BROKERS via getenv. +func LoadConfig(getenv func(string) string) (Config, error) { + brokers, err := parseBrokers(envOr(getenv, "KAFKA_BROKERS", "localhost:9092")) + if err != nil { + return Config{}, err + } + return Config{ + HTTPAddr: envOr(getenv, "HTTP_ADDR", ":2114"), + KafkaBrokers: brokers, + }, nil +} + +func envOr(getenv func(string) string, key, def string) string { + if v := strings.TrimSpace(getenv(key)); v != "" { + return v + } + return def +} + +// parseBrokers splits a comma-separated broker list, trimming spaces and +// rejecting empty entries. +func parseBrokers(raw string) ([]string, error) { + parts := strings.Split(raw, ",") + brokers := make([]string, 0, len(parts)) + for _, p := range parts { + p = strings.TrimSpace(p) + if p == "" { + return nil, fmt.Errorf("KAFKA_BROKERS %q contains an empty entry", raw) + } + brokers = append(brokers, p) + } + return brokers, nil +} diff --git a/services/assignment-service/internal/app/config_test.go b/services/assignment-service/internal/app/config_test.go new file mode 100644 index 0000000..b07876d --- /dev/null +++ b/services/assignment-service/internal/app/config_test.go @@ -0,0 +1,36 @@ +package app + +import ( + "reflect" + "testing" +) + +func env(m map[string]string) func(string) string { + return func(k string) string { return m[k] } +} + +func TestLoadConfig(t *testing.T) { + tests := []struct { + name string + env map[string]string + want Config + wantErr bool + }{ + {"defaults", nil, Config{HTTPAddr: ":2114", KafkaBrokers: []string{"localhost:9092"}}, false}, + {"overrides", map[string]string{"HTTP_ADDR": ":9000", "KAFKA_BROKERS": "a:1, b:2"}, + Config{HTTPAddr: ":9000", KafkaBrokers: []string{"a:1", "b:2"}}, false}, + {"empty broker entry", map[string]string{"KAFKA_BROKERS": "a:1,,b:2"}, Config{}, true}, + {"trailing comma", map[string]string{"KAFKA_BROKERS": "a:1,"}, Config{}, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := LoadConfig(env(tt.env)) + if (err != nil) != tt.wantErr { + t.Fatalf("err = %v, wantErr %v", err, tt.wantErr) + } + if !tt.wantErr && !reflect.DeepEqual(got, tt.want) { + t.Fatalf("got %+v, want %+v", got, tt.want) + } + }) + } +} diff --git a/services/matching-service/cmd/matching-service/main.go b/services/matching-service/cmd/matching-service/main.go index f75c180..71479a3 100644 --- a/services/matching-service/cmd/matching-service/main.go +++ b/services/matching-service/cmd/matching-service/main.go @@ -14,7 +14,13 @@ func main() { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() - if err := app.Run(ctx); err != nil { + cfg, err := app.LoadConfig(os.Getenv) + if err != nil { + log.Fatalf("matching-service: invalid configuration: %v", err) + } + log.Printf("matching-service config: metrics=%s brokers=%v redis=%s workers=%d deadline=%s", cfg.MetricsAddr, cfg.KafkaBrokers, cfg.RedisAddr, cfg.MatchWorkers, cfg.MatchDeadline) + + if err := app.Run(ctx, cfg); err != nil { stop() log.Fatal(err) } diff --git a/services/matching-service/internal/app/app.go b/services/matching-service/internal/app/app.go index 988367f..9e13461 100644 --- a/services/matching-service/internal/app/app.go +++ b/services/matching-service/internal/app/app.go @@ -18,11 +18,12 @@ import ( // Deadlines for the per-ride processing pipeline. // -// matchDeadline is the total wall-clock budget the matcher has for finding a -// driver. It must be strictly greater than matching.BatchDeadline so that at -// least one batch attempt fits inside it. In practice we size it as -// roughly (matching.BatchDeadline * desiredBatchAttempts) so contention on a -// hotspot can traverse multiple batches before giving up. +// defaultMatchDeadline is the default total wall-clock budget the matcher has +// for finding a driver (Config.MatchDeadline, MATCH_DEADLINE). It must be +// strictly greater than matching.BatchDeadline so that at least one batch +// attempt fits inside it. A successful match's p95 was ~80 ms, so 500 ms +// still fits two full batches and halves the time a worker is held on a +// failure compared with the previous 800 ms. // // publishDeadline is a SEPARATE budget for emitting ride.matched to Kafka. // It was previously sharing the match deadline via a single 300 ms context, @@ -30,16 +31,15 @@ import ( // enough for a Kafka WriteMessages round-trip with RequireOne, and the // "match succeeded but publish timed out" failure mode was silent. const ( - matchDeadline = 800 * time.Millisecond - publishDeadline = 300 * time.Millisecond + defaultMatchDeadline = 500 * time.Millisecond + publishDeadline = 300 * time.Millisecond - metricsAddr = ":2112" shutdownTimeout = 5 * time.Second ) // Run serves metrics and consumes RidePriced events until ctx is cancelled, // then shuts everything down. -func Run(ctx context.Context) error { +func Run(ctx context.Context, cfg Config) error { metrics.Register() // Dedicated mux so the scrape endpoint can't collide with anything @@ -47,27 +47,27 @@ func Run(ctx context.Context) error { mux := http.NewServeMux() mux.Handle("/metrics", promhttp.Handler()) server := &http.Server{ - Addr: metricsAddr, + Addr: cfg.MetricsAddr, Handler: mux, ReadTimeout: 5 * time.Second, WriteTimeout: 5 * time.Second, } go func() { - log.Printf("Prometheus metrics running on %s", metricsAddr) + log.Printf("Prometheus metrics running on %s", cfg.MetricsAddr) if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { log.Printf("matching-service metrics server: %v", err) } }() // --- Dependencies --- - driverRepo := redis.NewDriverRepository("localhost:6379") + driverRepo := redis.NewDriverRepository(cfg.RedisAddr) matcher := matching.New(driverRepo) - consumer := kafka.NewKafkaConsumer([]string{"localhost:9092"}) - publisher := kafka.NewKafkaPublisher([]string{"localhost:9092"}) + consumer := kafka.NewKafkaConsumer(cfg.KafkaBrokers, cfg.MatchWorkers) + publisher := kafka.NewKafkaPublisher(cfg.KafkaBrokers) runErr := consumer.Run(ctx, func(ctx context.Context, event domain.RidePricedEvent) error { - matchCtx, matchCancel := context.WithTimeout(ctx, matchDeadline) + matchCtx, matchCancel := context.WithTimeout(ctx, cfg.MatchDeadline) defer matchCancel() match, err := matcher.Match(matchCtx, event) @@ -85,7 +85,7 @@ func Run(ctx context.Context) error { } // Publish gets its OWN budget derived from the handler ctx, not from - // matchCtx. A match that used most of matchDeadline must not starve + // matchCtx. A match that used most of the match deadline must not starve // the Kafka write of its round-trip budget. A publish failure is // returned so the message is retried and then dead-lettered. publishCtx, publishCancel := context.WithTimeout(ctx, publishDeadline) diff --git a/services/matching-service/internal/app/config.go b/services/matching-service/internal/app/config.go new file mode 100644 index 0000000..08d4556 --- /dev/null +++ b/services/matching-service/internal/app/config.go @@ -0,0 +1,77 @@ +package app + +import ( + "fmt" + "runtime" + "strconv" + "strings" + "time" + + "ridepulse/services/matching-service/internal/matching" +) + +// Config holds matching-service's environment-derived settings. +type Config struct { + MetricsAddr string + KafkaBrokers []string + RedisAddr string + MatchWorkers int + MatchDeadline time.Duration +} + +// LoadConfig reads the service settings via getenv and validates them. +func LoadConfig(getenv func(string) string) (Config, error) { + brokers, err := parseBrokers(envOr(getenv, "KAFKA_BROKERS", "localhost:9092")) + if err != nil { + return Config{}, err + } + + workers := runtime.NumCPU() * 25 + if raw := strings.TrimSpace(getenv("MATCH_WORKERS")); raw != "" { + workers, err = strconv.Atoi(raw) + if err != nil || workers <= 0 { + return Config{}, fmt.Errorf("MATCH_WORKERS %q must be a positive integer", raw) + } + } + + deadline := defaultMatchDeadline + if raw := strings.TrimSpace(getenv("MATCH_DEADLINE")); raw != "" { + deadline, err = time.ParseDuration(raw) + if err != nil { + return Config{}, fmt.Errorf("MATCH_DEADLINE %q is not a valid duration: %w", raw, err) + } + if deadline <= matching.BatchDeadline { + return Config{}, fmt.Errorf("MATCH_DEADLINE %q must be greater than the batch deadline %s", raw, matching.BatchDeadline) + } + } + + return Config{ + MetricsAddr: envOr(getenv, "METRICS_ADDR", ":2112"), + KafkaBrokers: brokers, + RedisAddr: envOr(getenv, "REDIS_ADDR", "localhost:6379"), + MatchWorkers: workers, + MatchDeadline: deadline, + }, nil +} + +func envOr(getenv func(string) string, key, def string) string { + if v := strings.TrimSpace(getenv(key)); v != "" { + return v + } + return def +} + +// parseBrokers splits a comma-separated broker list, trimming spaces and +// rejecting empty entries. +func parseBrokers(raw string) ([]string, error) { + parts := strings.Split(raw, ",") + brokers := make([]string, 0, len(parts)) + for _, p := range parts { + p = strings.TrimSpace(p) + if p == "" { + return nil, fmt.Errorf("KAFKA_BROKERS %q contains an empty entry", raw) + } + brokers = append(brokers, p) + } + return brokers, nil +} diff --git a/services/matching-service/internal/app/config_test.go b/services/matching-service/internal/app/config_test.go new file mode 100644 index 0000000..d90bdfd --- /dev/null +++ b/services/matching-service/internal/app/config_test.go @@ -0,0 +1,57 @@ +package app + +import ( + "reflect" + "runtime" + "testing" + "time" +) + +func env(m map[string]string) func(string) string { + return func(k string) string { return m[k] } +} + +func TestLoadConfig(t *testing.T) { + defaults := Config{ + MetricsAddr: ":2112", + KafkaBrokers: []string{"localhost:9092"}, + RedisAddr: "localhost:6379", + MatchWorkers: runtime.NumCPU() * 25, + MatchDeadline: 500 * time.Millisecond, + } + tests := []struct { + name string + env map[string]string + want Config + wantErr bool + }{ + {"defaults", nil, defaults, false}, + {"overrides", map[string]string{ + "METRICS_ADDR": ":9", "KAFKA_BROKERS": "a:1, b:2", "REDIS_ADDR": "r:6", + "MATCH_WORKERS": "7", "MATCH_DEADLINE": "1s", + }, Config{ + MetricsAddr: ":9", KafkaBrokers: []string{"a:1", "b:2"}, RedisAddr: "r:6", + MatchWorkers: 7, MatchDeadline: time.Second, + }, false}, + {"empty broker entry", map[string]string{"KAFKA_BROKERS": "a:1,,b:2"}, Config{}, true}, + {"workers zero", map[string]string{"MATCH_WORKERS": "0"}, Config{}, true}, + {"workers negative", map[string]string{"MATCH_WORKERS": "-3"}, Config{}, true}, + {"workers not a number", map[string]string{"MATCH_WORKERS": "many"}, Config{}, true}, + {"deadline not a duration", map[string]string{"MATCH_DEADLINE": "500"}, Config{}, true}, + {"deadline equals batch deadline", map[string]string{"MATCH_DEADLINE": "200ms"}, Config{}, true}, + {"deadline below batch deadline", map[string]string{"MATCH_DEADLINE": "100ms"}, Config{}, true}, + {"deadline just above batch deadline", map[string]string{"MATCH_DEADLINE": "201ms"}, + func() Config { c := defaults; c.MatchDeadline = 201 * time.Millisecond; return c }(), false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := LoadConfig(env(tt.env)) + if (err != nil) != tt.wantErr { + t.Fatalf("err = %v, wantErr %v", err, tt.wantErr) + } + if !tt.wantErr && !reflect.DeepEqual(got, tt.want) { + t.Fatalf("got %+v, want %+v", got, tt.want) + } + }) + } +} diff --git a/services/matching-service/internal/kafka/consumer.go b/services/matching-service/internal/kafka/consumer.go index ace084e..f0632c4 100644 --- a/services/matching-service/internal/kafka/consumer.go +++ b/services/matching-service/internal/kafka/consumer.go @@ -5,7 +5,6 @@ import ( "encoding/json" "fmt" "log" - "runtime" "time" "github.com/segmentio/kafka-go" @@ -18,11 +17,12 @@ import ( const dlqTopic = "ride.priced.dlq" type KafkaConsumer struct { - reader *kafka.Reader - dlq *kafka.Writer + reader *kafka.Reader + dlq *kafka.Writer + workers int } -func NewKafkaConsumer(brokers []string) *KafkaConsumer { +func NewKafkaConsumer(brokers []string, workers int) *KafkaConsumer { r := kafka.NewReader(kafka.ReaderConfig{ Brokers: brokers, GroupID: "matching-service", @@ -46,7 +46,7 @@ func NewKafkaConsumer(brokers []string) *KafkaConsumer { // Default batch timeout (1s) would stall every synchronous DLQ write. BatchTimeout: 10 * time.Millisecond, } - return &KafkaConsumer{reader: r, dlq: dlq} + return &KafkaConsumer{reader: r, dlq: dlq, workers: workers} } // Run consumes RidePriced events until ctx is cancelled, calling handler for @@ -58,7 +58,7 @@ func (c *KafkaConsumer) Run( handler func(ctx context.Context, event domain.RidePricedEvent) error, ) error { cfg := kafkaconsumer.Config{ - Workers: runtime.NumCPU() * 5, + Workers: c.workers, DLQ: c.dlq, DLQTopic: dlqTopic, OnRetry: func(kafka.Message, int, error) { metrics.Retries.Inc() }, diff --git a/services/pricing-service/cmd/pricing-service/main.go b/services/pricing-service/cmd/pricing-service/main.go index c6852e6..c1c7b77 100644 --- a/services/pricing-service/cmd/pricing-service/main.go +++ b/services/pricing-service/cmd/pricing-service/main.go @@ -15,7 +15,13 @@ func main() { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() - if err := app.Run(ctx); err != nil { + cfg, err := app.LoadConfig(os.Getenv) + if err != nil { + log.Fatalf("pricing-service: invalid configuration: %v", err) + } + log.Printf("pricing-service config: metrics=%s brokers=%v", cfg.MetricsAddr, cfg.KafkaBrokers) + + if err := app.Run(ctx, cfg); err != nil { stop() log.Fatal(err) } diff --git a/services/pricing-service/internal/app/app.go b/services/pricing-service/internal/app/app.go index 35c0d39..49727bd 100644 --- a/services/pricing-service/internal/app/app.go +++ b/services/pricing-service/internal/app/app.go @@ -17,14 +17,13 @@ import ( ) const ( - metricsAddr = ":2115" publishTimeout = 300 * time.Millisecond shutdownTimeout = 5 * time.Second ) // Run serves metrics and consumes RideRequested events until ctx is // cancelled, then shuts everything down. -func Run(ctx context.Context) error { +func Run(ctx context.Context, cfg Config) error { metrics.Register() // Prometheus scrape endpoint on its own port. Uses a dedicated ServeMux so @@ -32,24 +31,24 @@ func Run(ctx context.Context) error { mux := http.NewServeMux() mux.Handle("/metrics", promhttp.Handler()) server := &http.Server{ - Addr: metricsAddr, + Addr: cfg.MetricsAddr, Handler: mux, ReadTimeout: 5 * time.Second, WriteTimeout: 5 * time.Second, } go func() { - log.Printf("pricing-service metrics listening on %s", metricsAddr) + log.Printf("pricing-service metrics listening on %s", cfg.MetricsAddr) if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { log.Printf("pricing-service metrics server: %v", err) } }() consumer := kafka.NewKafkaConsumer( - []string{"localhost:9092"}, + cfg.KafkaBrokers, "pricing-service", "ride.requested", ) - publisher := kafka.NewKafkaPublisher([]string{"localhost:9092"}) + publisher := kafka.NewKafkaPublisher(cfg.KafkaBrokers) runErr := consumer.Run(ctx, func(ctx context.Context, event domain.RideRequestedEvent) error { // Counts handler attempts (retries included); pricing_retries_total diff --git a/services/pricing-service/internal/app/config.go b/services/pricing-service/internal/app/config.go new file mode 100644 index 0000000..248209a --- /dev/null +++ b/services/pricing-service/internal/app/config.go @@ -0,0 +1,46 @@ +package app + +import ( + "fmt" + "strings" +) + +// Config holds pricing-service's environment-derived settings. +type Config struct { + MetricsAddr string + KafkaBrokers []string +} + +// LoadConfig reads METRICS_ADDR and KAFKA_BROKERS via getenv. +func LoadConfig(getenv func(string) string) (Config, error) { + brokers, err := parseBrokers(envOr(getenv, "KAFKA_BROKERS", "localhost:9092")) + if err != nil { + return Config{}, err + } + return Config{ + MetricsAddr: envOr(getenv, "METRICS_ADDR", ":2115"), + KafkaBrokers: brokers, + }, nil +} + +func envOr(getenv func(string) string, key, def string) string { + if v := strings.TrimSpace(getenv(key)); v != "" { + return v + } + return def +} + +// parseBrokers splits a comma-separated broker list, trimming spaces and +// rejecting empty entries. +func parseBrokers(raw string) ([]string, error) { + parts := strings.Split(raw, ",") + brokers := make([]string, 0, len(parts)) + for _, p := range parts { + p = strings.TrimSpace(p) + if p == "" { + return nil, fmt.Errorf("KAFKA_BROKERS %q contains an empty entry", raw) + } + brokers = append(brokers, p) + } + return brokers, nil +} diff --git a/services/pricing-service/internal/app/config_test.go b/services/pricing-service/internal/app/config_test.go new file mode 100644 index 0000000..7ea522c --- /dev/null +++ b/services/pricing-service/internal/app/config_test.go @@ -0,0 +1,36 @@ +package app + +import ( + "reflect" + "testing" +) + +func env(m map[string]string) func(string) string { + return func(k string) string { return m[k] } +} + +func TestLoadConfig(t *testing.T) { + tests := []struct { + name string + env map[string]string + want Config + wantErr bool + }{ + {"defaults", nil, Config{MetricsAddr: ":2115", KafkaBrokers: []string{"localhost:9092"}}, false}, + {"overrides", map[string]string{"METRICS_ADDR": ":9000", "KAFKA_BROKERS": "a:1, b:2"}, + Config{MetricsAddr: ":9000", KafkaBrokers: []string{"a:1", "b:2"}}, false}, + {"empty broker entry", map[string]string{"KAFKA_BROKERS": "a:1,,b:2"}, Config{}, true}, + {"trailing comma", map[string]string{"KAFKA_BROKERS": "a:1,"}, Config{}, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := LoadConfig(env(tt.env)) + if (err != nil) != tt.wantErr { + t.Fatalf("err = %v, wantErr %v", err, tt.wantErr) + } + if !tt.wantErr && !reflect.DeepEqual(got, tt.want) { + t.Fatalf("got %+v, want %+v", got, tt.want) + } + }) + } +} From 0044ca8f904fbbd134ea2f060ab6a063745b3ef0 Mon Sep 17 00:00:00 2001 From: coderconnoisseur Date: Fri, 16 Oct 2026 00:23:07 +0530 Subject: [PATCH 3/9] matching: publish ride.unmatched on business match failure --- services/matching-service/internal/app/app.go | 92 ++++++++++----- .../matching-service/internal/app/app_test.go | 109 ++++++++++++++++++ .../matching-service/internal/domain/ride.go | 11 +- .../internal/kafka/publisher.go | 31 ++++- .../internal/metrics/metrics.go | 9 ++ 5 files changed, 222 insertions(+), 30 deletions(-) diff --git a/services/matching-service/internal/app/app.go b/services/matching-service/internal/app/app.go index 9e13461..dbcb217 100644 --- a/services/matching-service/internal/app/app.go +++ b/services/matching-service/internal/app/app.go @@ -66,32 +66,7 @@ func Run(ctx context.Context, cfg Config) error { consumer := kafka.NewKafkaConsumer(cfg.KafkaBrokers, cfg.MatchWorkers) publisher := kafka.NewKafkaPublisher(cfg.KafkaBrokers) - runErr := consumer.Run(ctx, func(ctx context.Context, event domain.RidePricedEvent) error { - matchCtx, matchCancel := context.WithTimeout(ctx, cfg.MatchDeadline) - defer matchCancel() - - match, err := matcher.Match(matchCtx, event) - if err != nil { - if isBusinessFailure(err, matchCtx.Err()) { - // No driver / match deadline is a business outcome, not a - // retryable failure: log it and commit the offset. - log.Printf("matching failed for ride %s: %v", event.RideID, err) - return nil - } - // Redis/infra error: return it so the runner retries (and then - // dead-letters). Retries are safe: Match replays an existing - // assignment via ride:driver instead of matching again. - return err - } - - // Publish gets its OWN budget derived from the handler ctx, not from - // matchCtx. A match that used most of the match deadline must not starve - // the Kafka write of its round-trip budget. A publish failure is - // returned so the message is retried and then dead-lettered. - publishCtx, publishCancel := context.WithTimeout(ctx, publishDeadline) - defer publishCancel() - return publisher.PublishRideMatched(publishCtx, match) - }) + runErr := consumer.Run(ctx, newHandler(matcher, publisher, cfg.MatchDeadline)) shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownTimeout) defer cancel() @@ -120,3 +95,68 @@ func isBusinessFailure(err, matchCtxErr error) bool { return errors.Is(err, matching.ErrNoDriversAvailable) || (errors.Is(err, context.DeadlineExceeded) && matchCtxErr != nil) } + +// matcherAPI and eventPublisher are the handler's collaborators, narrowed so +// the handler can be tested with fakes. +type matcherAPI interface { + Match(ctx context.Context, event domain.RidePricedEvent) (domain.RideMatchedEvent, error) +} + +type eventPublisher interface { + PublishRideMatched(ctx context.Context, event domain.RideMatchedEvent) error + PublishRideUnmatched(ctx context.Context, event domain.RideUnmatchedEvent) error +} + +// newHandler builds the per-ride handler: match, then publish ride.matched, +// or on a business failure publish ride.unmatched so the rider is told. +func newHandler(matcher matcherAPI, publisher eventPublisher, matchDeadline time.Duration) func(context.Context, domain.RidePricedEvent) error { + return func(ctx context.Context, event domain.RidePricedEvent) error { + matchCtx, matchCancel := context.WithTimeout(ctx, matchDeadline) + defer matchCancel() + + match, err := matcher.Match(matchCtx, event) + if err != nil { + if isBusinessFailure(err, matchCtx.Err()) { + // No driver / match deadline is a business outcome, not a + // retryable failure: log it, tell the rider, commit the offset. + log.Printf("matching failed for ride %s: %v", event.RideID, err) + reason := unmatchedReason(err) + publishCtx, publishCancel := context.WithTimeout(ctx, publishDeadline) + defer publishCancel() + // A publish failure is returned so the runner retries it. + if perr := publisher.PublishRideUnmatched(publishCtx, domain.RideUnmatchedEvent{ + RideID: event.RideID, + Reason: reason, + RequestedAt: event.RequestedAt, + FailedAt: time.Now().UnixNano(), + }); perr != nil { + return perr + } + metrics.MatchUnmatchedPublished.WithLabelValues(reason).Inc() + return nil + } + // Redis/infra error: return it so the runner retries (and then + // dead-letters). Retries are safe: Match replays an existing + // assignment via ride:driver instead of matching again. + return err + } + + // Publish gets its OWN budget derived from the handler ctx, not from + // matchCtx. A match that used most of the match deadline must not starve + // the Kafka write of its round-trip budget. A publish failure is + // returned so the message is retried and then dead-lettered. + publishCtx, publishCancel := context.WithTimeout(ctx, publishDeadline) + defer publishCancel() + return publisher.PublishRideMatched(publishCtx, match) + } +} + +// unmatchedReason maps a business-failure error to the ride.unmatched reason. +// Mirrors isBusinessFailure: anything that is not ErrNoDriversAvailable is the +// outer match deadline. +func unmatchedReason(err error) string { + if errors.Is(err, matching.ErrNoDriversAvailable) { + return "no_drivers" + } + return "match_deadline" +} diff --git a/services/matching-service/internal/app/app_test.go b/services/matching-service/internal/app/app_test.go index 7ed9dee..103619c 100644 --- a/services/matching-service/internal/app/app_test.go +++ b/services/matching-service/internal/app/app_test.go @@ -5,7 +5,9 @@ import ( "errors" "fmt" "testing" + "time" + "ridepulse/services/matching-service/internal/domain" "ridepulse/services/matching-service/internal/matching" ) @@ -29,3 +31,110 @@ func TestIsBusinessFailure(t *testing.T) { } } } + +func TestUnmatchedReason(t *testing.T) { + cases := []struct { + name string + err error + want string + }{ + {"no drivers", matching.ErrNoDriversAvailable, "no_drivers"}, + {"wrapped no drivers", fmt.Errorf("match: %w", matching.ErrNoDriversAvailable), "no_drivers"}, + {"deadline", context.DeadlineExceeded, "match_deadline"}, + } + for _, tc := range cases { + if got := unmatchedReason(tc.err); got != tc.want { + t.Errorf("%s: unmatchedReason(%v) = %q, want %q", tc.name, tc.err, got, tc.want) + } + } +} + +type fakeMatcher struct { + match domain.RideMatchedEvent + err error + waitForDone bool // block until the match ctx expires, like a real slow match +} + +func (f fakeMatcher) Match(ctx context.Context, _ domain.RidePricedEvent) (domain.RideMatchedEvent, error) { + if f.waitForDone { + <-ctx.Done() + } + return f.match, f.err +} + +type fakePublisher struct { + matched []domain.RideMatchedEvent + unmatched []domain.RideUnmatchedEvent + unmatchedErr error +} + +func (f *fakePublisher) PublishRideMatched(_ context.Context, e domain.RideMatchedEvent) error { + f.matched = append(f.matched, e) + return nil +} + +func (f *fakePublisher) PublishRideUnmatched(_ context.Context, e domain.RideUnmatchedEvent) error { + if f.unmatchedErr != nil { + return f.unmatchedErr + } + f.unmatched = append(f.unmatched, e) + return nil +} + +func TestHandlerPublishesUnmatchedOnBusinessFailure(t *testing.T) { + pub := &fakePublisher{} + h := newHandler(fakeMatcher{err: matching.ErrNoDriversAvailable}, pub, time.Second) + before := time.Now().UnixNano() + if err := h(context.Background(), domain.RidePricedEvent{RideID: "r1", RequestedAt: 42}); err != nil { + t.Fatalf("handler error: %v", err) + } + if len(pub.matched) != 0 || len(pub.unmatched) != 1 { + t.Fatalf("matched=%d unmatched=%d, want 0/1", len(pub.matched), len(pub.unmatched)) + } + got := pub.unmatched[0] + if got.RideID != "r1" || got.Reason != "no_drivers" || got.RequestedAt != 42 || got.FailedAt < before { + t.Errorf("unexpected event: %+v", got) + } +} + +func TestHandlerUnmatchedOnMatchDeadline(t *testing.T) { + pub := &fakePublisher{} + h := newHandler(fakeMatcher{err: context.DeadlineExceeded, waitForDone: true}, pub, time.Millisecond) + if err := h(context.Background(), domain.RidePricedEvent{RideID: "r2"}); err != nil { + t.Fatalf("handler error: %v", err) + } + if len(pub.unmatched) != 1 || pub.unmatched[0].Reason != "match_deadline" { + t.Fatalf("unmatched = %+v, want one match_deadline", pub.unmatched) + } +} + +func TestHandlerReturnsUnmatchedPublishError(t *testing.T) { + boom := errors.New("kafka down") + pub := &fakePublisher{unmatchedErr: boom} + h := newHandler(fakeMatcher{err: matching.ErrNoDriversAvailable}, pub, time.Second) + if err := h(context.Background(), domain.RidePricedEvent{RideID: "r3"}); !errors.Is(err, boom) { + t.Fatalf("err = %v, want publish error", err) + } +} + +func TestHandlerInfraErrorPublishesNothing(t *testing.T) { + pub := &fakePublisher{} + h := newHandler(fakeMatcher{err: errors.New("redis down")}, pub, time.Second) + if err := h(context.Background(), domain.RidePricedEvent{RideID: "r4"}); err == nil { + t.Fatal("want error") + } + if len(pub.unmatched) != 0 || len(pub.matched) != 0 { + t.Errorf("published on infra error: %+v %+v", pub.matched, pub.unmatched) + } +} + +func TestHandlerPublishesMatchedOnSuccess(t *testing.T) { + pub := &fakePublisher{} + h := newHandler(fakeMatcher{match: domain.RideMatchedEvent{RideID: "r5", DriverID: "d"}}, pub, time.Second) + if err := h(context.Background(), domain.RidePricedEvent{RideID: "r5"}); err != nil { + t.Fatal(err) + } + if len(pub.matched) != 1 || len(pub.unmatched) != 0 { + t.Errorf("matched=%d unmatched=%d", len(pub.matched), len(pub.unmatched)) + } +} diff --git a/services/matching-service/internal/domain/ride.go b/services/matching-service/internal/domain/ride.go index 61cca9e..78f9cea 100644 --- a/services/matching-service/internal/domain/ride.go +++ b/services/matching-service/internal/domain/ride.go @@ -20,4 +20,13 @@ type RideMatchedEvent struct{ DriverID string MatchedAT int64 RequestedAt int64 // carried through from RideRequestedEvent -} \ No newline at end of file +} + +// RideUnmatchedEvent is published when matching ends in a business failure +// (no driver found, or the match deadline expired). +type RideUnmatchedEvent struct{ + RideID string + Reason string // "no_drivers" or "match_deadline" + RequestedAt int64 // UnixNano, carried through from RidePricedEvent + FailedAt int64 // UnixNano +} diff --git a/services/matching-service/internal/kafka/publisher.go b/services/matching-service/internal/kafka/publisher.go index ae459d1..8fecf03 100644 --- a/services/matching-service/internal/kafka/publisher.go +++ b/services/matching-service/internal/kafka/publisher.go @@ -2,11 +2,13 @@ package kafka import( "context" "encoding/json" + "errors" "github.com/segmentio/kafka-go" "ridepulse/services/matching-service/internal/domain" ) type KafkaPublisher struct{ writer *kafka.Writer; + unmatchedWriter *kafka.Writer } func NewKafkaPublisher (brokers[] string)*KafkaPublisher{ writer:=&kafka.Writer{ @@ -16,7 +18,15 @@ func NewKafkaPublisher (brokers[] string)*KafkaPublisher{ RequiredAcks: kafka.RequireOne, BatchTimeout: 10_000_000,//10ms } - return &KafkaPublisher{writer} + unmatched:=&kafka.Writer{ + Addr: kafka.TCP(brokers...), + Topic:"ride.unmatched", + Balancer: &kafka.Hash{}, + RequiredAcks: kafka.RequireOne, + BatchTimeout: 10_000_000,//10ms + AllowAutoTopicCreation: true,// new topic + } + return &KafkaPublisher{writer, unmatched} } func (p *KafkaPublisher) PublishRideMatched( @@ -34,7 +44,22 @@ func (p *KafkaPublisher) PublishRideMatched( return p.writer.WriteMessages(ctx, msg) } -// Close flushes and closes the underlying writer. +// PublishRideUnmatched writes a ride.unmatched event keyed by ride ID. +func (p *KafkaPublisher) PublishRideUnmatched( + ctx context.Context, + event domain.RideUnmatchedEvent, +)error{ + val,err:=json.Marshal(event) + if err!=nil{ + return err + } + return p.unmatchedWriter.WriteMessages(ctx, kafka.Message{ + Key: []byte(event.RideID), + Value: val, + }) +} + +// Close flushes and closes both writers. func (p *KafkaPublisher) Close() error { - return p.writer.Close() + return errors.Join(p.writer.Close(), p.unmatchedWriter.Close()) } diff --git a/services/matching-service/internal/metrics/metrics.go b/services/matching-service/internal/metrics/metrics.go index eea9f7e..8ed4430 100644 --- a/services/matching-service/internal/metrics/metrics.go +++ b/services/matching-service/internal/metrics/metrics.go @@ -84,6 +84,14 @@ var ( }, ) + MatchUnmatchedPublished = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "match_unmatched_published_total", + Help: "ride.unmatched events published, by reason.", + }, + []string{"reason"}, + ) + MatchLatency = prometheus.NewHistogram( prometheus.HistogramOpts{ Name: "matching_latency_seconds", @@ -131,6 +139,7 @@ func Register() { MatchSuccess, MatchFailures, MatchIdempotentReplays, + MatchUnmatchedPublished, MatchLatency, NearbyDrivers, Goroutines, From 8c32a16b459d0eb185b8c62434091db8af984fac Mon Sep 17 00:00:00 2001 From: coderconnoisseur Date: Fri, 16 Oct 2026 00:26:59 +0530 Subject: [PATCH 4/9] assignment: dedupe ride.matched, consume ride.unmatched, ride status API --- .../assignment-service/internal/app/app.go | 91 +++++++++++++++---- .../internal/domain/ride.go | 25 +++++ .../internal/httpapi/server.go | 49 +++++++++- .../internal/httpapi/server_test.go | 69 ++++++++++++++ .../internal/kafka/consumer.go | 72 ++++++++++++--- .../internal/metrics/metrics.go | 29 ++++++ .../internal/store/memory.go | 74 ++++++++++++--- .../internal/store/memory_test.go | 78 ++++++++++++++++ 8 files changed, 439 insertions(+), 48 deletions(-) create mode 100644 services/assignment-service/internal/httpapi/server_test.go create mode 100644 services/assignment-service/internal/store/memory_test.go diff --git a/services/assignment-service/internal/app/app.go b/services/assignment-service/internal/app/app.go index 33015f0..08f97f1 100644 --- a/services/assignment-service/internal/app/app.go +++ b/services/assignment-service/internal/app/app.go @@ -5,6 +5,7 @@ import ( "errors" "log" "net/http" + "sync" "time" "ridepulse/services/assignment-service/internal/domain" @@ -18,8 +19,8 @@ const ( shutdownTimeout = 5 * time.Second ) -// Run serves the HTTP API and consumes RideMatched events until ctx is -// cancelled, then shuts everything down. +// Run serves the HTTP API and consumes RideMatched and RideUnmatched events +// until ctx is cancelled, then shuts everything down. func Run(ctx context.Context, cfg Config) error { metrics.Register() mem := store.NewMemory() @@ -32,33 +33,87 @@ func Run(ctx context.Context, cfg Config) error { } }() - consumer := kafka.NewKafkaConsumer(cfg.KafkaBrokers) - runErr := consumer.Run(ctx, func(_ context.Context, event domain.RideMatchedEvent) error { + matched := kafka.NewKafkaConsumer(cfg.KafkaBrokers) + unmatched := kafka.NewUnmatchedConsumer(cfg.KafkaBrokers) + + // Both consumers share a derived ctx: if one fails fatally the other is + // cancelled, and the first error is returned once both have stopped. + runCtx, cancelRun := context.WithCancel(ctx) + defer cancelRun() + var ( + wg sync.WaitGroup + errOnce sync.Once + firstErr error + ) + runner := func(run func(context.Context) error) { + defer wg.Done() + if err := run(runCtx); err != nil { + errOnce.Do(func() { firstErr = err }) + cancelRun() + } + } + wg.Add(2) + go runner(func(c context.Context) error { return matched.Run(c, handleMatched(mem)) }) + go runner(func(c context.Context) error { return unmatched.Run(c, handleUnmatched(mem)) }) + wg.Wait() + runErr := firstErr + + shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownTimeout) + defer cancel() + if err := server.Shutdown(shutdownCtx); err != nil { + log.Printf("assignment-service http shutdown: %v", err) + } + if err := matched.Close(); err != nil { + log.Printf("assignment-service matched consumer close: %v", err) + } + if err := unmatched.Close(); err != nil { + log.Printf("assignment-service unmatched consumer close: %v", err) + } + log.Printf("assignment-service stopped") + return runErr +} + +// handleMatched records a matched ride. Metrics are observed only for the +// first delivery; redeliveries count as duplicates. +func handleMatched(mem *store.Memory) func(context.Context, domain.RideMatchedEvent) error { + return func(_ context.Context, event domain.RideMatchedEvent) error { var latencyMs int64 if event.RequestedAt > 0 { latencyMs = (time.Now().UnixNano() - event.RequestedAt) / int64(time.Millisecond) - metrics.EndToEndLatency.Observe(float64(latencyMs) / 1000.0) - } else { - metrics.MissingRequestedAt.Inc() } - mem.Put(domain.Assignment{ + isNew := mem.PutMatched(domain.Assignment{ RideID: event.RideID, DriverID: event.DriverID, MatchedAT: event.MatchedAT, LatencyMs: latencyMs, }) + if !isNew { + metrics.AssignmentDuplicates.Inc() + return nil + } + if event.RequestedAt > 0 { + metrics.EndToEndLatency.Observe(float64(latencyMs) / 1000.0) + } else { + metrics.MissingRequestedAt.Inc() + } metrics.AssignmentsTotal.Inc() return nil - }) - - shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownTimeout) - defer cancel() - if err := server.Shutdown(shutdownCtx); err != nil { - log.Printf("assignment-service http shutdown: %v", err) } - if err := consumer.Close(); err != nil { - log.Printf("assignment-service consumer close: %v", err) +} + +// handleUnmatched records a failed ride, ignoring duplicates and rides that +// already have an outcome. +func handleUnmatched(mem *store.Memory) func(context.Context, domain.RideUnmatchedEvent) error { + return func(_ context.Context, event domain.RideUnmatchedEvent) error { + if !mem.PutUnmatched(event.RideID, event.Reason, event.FailedAt) { + metrics.UnmatchedDuplicates.Inc() + return nil + } + metrics.UnmatchedTotal.WithLabelValues(event.Reason).Inc() + if event.RequestedAt > 0 { + elapsed := time.Now().UnixNano() - event.RequestedAt + metrics.UnmatchedEndToEndLatency.Observe(float64(elapsed) / float64(time.Second)) + } + return nil } - log.Printf("assignment-service stopped") - return runErr } diff --git a/services/assignment-service/internal/domain/ride.go b/services/assignment-service/internal/domain/ride.go index 19cf482..6ea5504 100644 --- a/services/assignment-service/internal/domain/ride.go +++ b/services/assignment-service/internal/domain/ride.go @@ -14,6 +14,15 @@ type RideMatchedEvent struct { RequestedAt int64 } +// RideUnmatchedEvent is published by matching-service on ride.unmatched when +// a ride could not be matched. FailedAt is a UnixNano timestamp. +type RideUnmatchedEvent struct { + RideID string + Reason string + RequestedAt int64 + FailedAt int64 +} + // Assignment is what we store for each successfully matched ride. type Assignment struct { RideID string `json:"ride_id"` @@ -21,3 +30,19 @@ type Assignment struct { MatchedAT int64 `json:"matched_at"` LatencyMs int64 `json:"latency_ms"` } + +// Ride outcome statuses reported by the store and the HTTP API. +const ( + StatusMatched = "matched" + StatusUnmatched = "unmatched" +) + +// Outcome is the terminal result recorded for a ride: the assignment when +// matched, or the reason and failure time when unmatched. +type Outcome struct { + RideID string + Status string + Assignment Assignment + Reason string + FailedAt int64 +} diff --git a/services/assignment-service/internal/httpapi/server.go b/services/assignment-service/internal/httpapi/server.go index 7a41c41..05a224c 100644 --- a/services/assignment-service/internal/httpapi/server.go +++ b/services/assignment-service/internal/httpapi/server.go @@ -9,6 +9,7 @@ import ( "github.com/prometheus/client_golang/prometheus/promhttp" + "ridepulse/services/assignment-service/internal/domain" "ridepulse/services/assignment-service/internal/store" ) @@ -17,11 +18,13 @@ import ( // // GET /metrics — Prometheus scrape endpoint // GET /assignments/ — look up a single assignment (JSON) -// GET /count — total assignments stored (plaintext) +// GET /rides/ — matched/unmatched outcome of a ride (JSON) +// GET /count — total matched assignments stored (plaintext) func NewServer(addr string, mem *store.Memory) *http.Server { mux := http.NewServeMux() mux.Handle("/metrics", promhttp.Handler()) mux.HandleFunc("/assignments/", queryHandler(mem)) + mux.HandleFunc("/rides/", rideHandler(mem)) mux.HandleFunc("/count", countHandler(mem)) return &http.Server{ @@ -49,6 +52,50 @@ func queryHandler(mem *store.Memory) http.HandlerFunc { } } +type matchedBody struct { + RideID string `json:"ride_id"` + Status string `json:"status"` + DriverID string `json:"driver_id"` + MatchedAT int64 `json:"matched_at"` + LatencyMs int64 `json:"latency_ms"` +} + +type unmatchedBody struct { + RideID string `json:"ride_id"` + Status string `json:"status"` + Reason string `json:"reason"` + FailedAt int64 `json:"failed_at"` +} + +func rideHandler(mem *store.Memory) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + rideID := strings.TrimPrefix(r.URL.Path, "/rides/") + if rideID == "" { + http.Error(w, "missing ride id", http.StatusBadRequest) + return + } + o, ok := mem.Status(rideID) + if !ok { + http.Error(w, "not found", http.StatusNotFound) + return + } + var body any + if o.Status == domain.StatusMatched { + body = matchedBody{ + RideID: o.RideID, + Status: o.Status, + DriverID: o.Assignment.DriverID, + MatchedAT: o.Assignment.MatchedAT, + LatencyMs: o.Assignment.LatencyMs, + } + } else { + body = unmatchedBody{RideID: o.RideID, Status: o.Status, Reason: o.Reason, FailedAt: o.FailedAt} + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(body) + } +} + func countHandler(mem *store.Memory) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { fmt.Fprintf(w, "%d\n", mem.Count()) diff --git a/services/assignment-service/internal/httpapi/server_test.go b/services/assignment-service/internal/httpapi/server_test.go new file mode 100644 index 0000000..649b540 --- /dev/null +++ b/services/assignment-service/internal/httpapi/server_test.go @@ -0,0 +1,69 @@ +package httpapi + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "reflect" + "testing" + + "ridepulse/services/assignment-service/internal/domain" + "ridepulse/services/assignment-service/internal/store" +) + +func get(mem *store.Memory, path string) *httptest.ResponseRecorder { + rec := httptest.NewRecorder() + NewServer("", mem).Handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, path, nil)) + return rec +} + +func TestRideStatusShapes(t *testing.T) { + mem := store.NewMemory() + mem.PutMatched(domain.Assignment{RideID: "m1", DriverID: "d1", MatchedAT: 100, LatencyMs: 42}) + mem.PutUnmatched("u1", "no_drivers", 200) + + cases := []struct { + path string + want map[string]any + }{ + {"/rides/m1", map[string]any{ + "ride_id": "m1", "status": "matched", "driver_id": "d1", "matched_at": 100.0, "latency_ms": 42.0, + }}, + {"/rides/u1", map[string]any{ + "ride_id": "u1", "status": "unmatched", "reason": "no_drivers", "failed_at": 200.0, + }}, + } + for _, tc := range cases { + rec := get(mem, tc.path) + if rec.Code != http.StatusOK { + t.Fatalf("%s code = %d", tc.path, rec.Code) + } + var got map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(got, tc.want) { + t.Fatalf("%s body = %v, want %v", tc.path, got, tc.want) + } + } + + if rec := get(mem, "/rides/nope"); rec.Code != http.StatusNotFound { + t.Fatalf("unknown code = %d, want 404", rec.Code) + } +} + +func TestAssignmentsAndCountUnchanged(t *testing.T) { + mem := store.NewMemory() + mem.PutMatched(domain.Assignment{RideID: "m1", DriverID: "d1"}) + mem.PutUnmatched("u1", "no_drivers", 200) + + if rec := get(mem, "/assignments/m1"); rec.Code != http.StatusOK { + t.Fatalf("assignments matched code = %d", rec.Code) + } + if rec := get(mem, "/assignments/u1"); rec.Code != http.StatusNotFound { + t.Fatalf("assignments unmatched code = %d, want 404", rec.Code) + } + if rec := get(mem, "/count"); rec.Body.String() != "1\n" { + t.Fatalf("count = %q, want 1", rec.Body.String()) + } +} diff --git a/services/assignment-service/internal/kafka/consumer.go b/services/assignment-service/internal/kafka/consumer.go index 40cadc7..5c53b0a 100644 --- a/services/assignment-service/internal/kafka/consumer.go +++ b/services/assignment-service/internal/kafka/consumer.go @@ -14,18 +14,28 @@ import ( "ridepulse/services/assignment-service/internal/metrics" ) -const dlqTopic = "ride.matched.dlq" +const ( + groupID = "assignment-service" -type KafkaConsumer struct { - reader *kafka.Reader - dlq *kafka.Writer + matchedTopic = "ride.matched" + matchedDLQTopic = "ride.matched.dlq" + unmatchedTopic = "ride.unmatched" + unmatchedDLQTopic = "ride.unmatched.dlq" +) + +// consumer is the topic-agnostic reader + DLQ writer pair shared by the +// typed consumers below. Both topics use the same consumer group. +type consumer struct { + reader *kafka.Reader + dlq *kafka.Writer + dlqTopic string } -func NewKafkaConsumer(brokers []string) *KafkaConsumer { +func newConsumer(brokers []string, topic, dlqTopic string) *consumer { r := kafka.NewReader(kafka.ReaderConfig{ Brokers: brokers, - GroupID: "assignment-service", - Topic: "ride.matched", + GroupID: groupID, + Topic: topic, // Commits are issued by the runner after processing; the interval // makes them asynchronous batches instead of one round trip each. CommitInterval: time.Second, @@ -45,37 +55,69 @@ func NewKafkaConsumer(brokers []string) *KafkaConsumer { // Default batch timeout (1s) would stall every synchronous DLQ write. BatchTimeout: 10 * time.Millisecond, } - return &KafkaConsumer{reader: r, dlq: dlq} + return &consumer{reader: r, dlq: dlq, dlqTopic: dlqTopic} } -// Run consumes RideMatched events until ctx is cancelled, calling handler for +// run consumes events of type T until ctx is cancelled, calling handler for // each. A decode failure is dead-lettered without retries; a handler error is // retried and then dead-lettered. Offsets are committed only once a message // has been handled or dead-lettered. -func (c *KafkaConsumer) Run( +func run[T any]( ctx context.Context, - handler func(ctx context.Context, event domain.RideMatchedEvent) error, + c *consumer, + what string, + handler func(ctx context.Context, event T) error, ) error { cfg := kafkaconsumer.Config{ Workers: 4, DLQ: c.dlq, - DLQTopic: dlqTopic, + DLQTopic: c.dlqTopic, OnDLQ: func(kafka.Message, error) { metrics.DLQMessages.Inc() }, } return kafkaconsumer.Run(ctx, c.reader, cfg, func(ctx context.Context, msg kafka.Message) error { - var event domain.RideMatchedEvent + var event T if err := json.Unmarshal(msg.Value, &event); err != nil { - return kafkaconsumer.Permanent(fmt.Errorf("decode RideMatched: %w", err)) + return kafkaconsumer.Permanent(fmt.Errorf("decode %s: %w", what, err)) } return handler(ctx, event) }) } // Close closes the reader and the DLQ writer. -func (c *KafkaConsumer) Close() error { +func (c *consumer) Close() error { rerr := c.reader.Close() if err := c.dlq.Close(); err != nil { return err } return rerr } + +// KafkaConsumer consumes ride.matched. +type KafkaConsumer struct{ *consumer } + +func NewKafkaConsumer(brokers []string) *KafkaConsumer { + return &KafkaConsumer{newConsumer(brokers, matchedTopic, matchedDLQTopic)} +} + +// Run consumes RideMatched events until ctx is cancelled. +func (c *KafkaConsumer) Run( + ctx context.Context, + handler func(ctx context.Context, event domain.RideMatchedEvent) error, +) error { + return run(ctx, c.consumer, "RideMatched", handler) +} + +// UnmatchedConsumer consumes ride.unmatched. +type UnmatchedConsumer struct{ *consumer } + +func NewUnmatchedConsumer(brokers []string) *UnmatchedConsumer { + return &UnmatchedConsumer{newConsumer(brokers, unmatchedTopic, unmatchedDLQTopic)} +} + +// Run consumes RideUnmatched events until ctx is cancelled. +func (c *UnmatchedConsumer) Run( + ctx context.Context, + handler func(ctx context.Context, event domain.RideUnmatchedEvent) error, +) error { + return run(ctx, c.consumer, "RideUnmatched", handler) +} diff --git a/services/assignment-service/internal/metrics/metrics.go b/services/assignment-service/internal/metrics/metrics.go index 57bf00a..55d48fa 100644 --- a/services/assignment-service/internal/metrics/metrics.go +++ b/services/assignment-service/internal/metrics/metrics.go @@ -25,6 +25,31 @@ var ( Help: "Matched events with no RequestedAt timestamp (indicates upstream schema drift).", }) + // AssignmentDuplicates counts redelivered ride.matched events for rides + // already recorded; they are not counted in AssignmentsTotal. + AssignmentDuplicates = prometheus.NewCounter(prometheus.CounterOpts{ + Name: "assignment_duplicates_total", + Help: "Duplicate ride.matched events ignored (at-least-once redelivery).", + }) + + UnmatchedTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "unmatched_total", + Help: "Rides that failed matching (ride.unmatched consumptions), by reason.", + }, []string{"reason"}) + + UnmatchedDuplicates = prometheus.NewCounter(prometheus.CounterOpts{ + Name: "unmatched_duplicates_total", + Help: "ride.unmatched events ignored because the ride already had an outcome.", + }) + + // UnmatchedEndToEndLatency: time from api-gateway POST /rides receipt to + // the matching failure being consumed. Same buckets as EndToEndLatency. + UnmatchedEndToEndLatency = prometheus.NewHistogram(prometheus.HistogramOpts{ + Name: "unmatched_end_to_end_latency_seconds", + Help: "Time from RideRequested emission to RideUnmatched consumption.", + Buckets: []float64{0.05, 0.1, 0.2, 0.3, 0.5, 0.8, 1.2, 2.0, 3.0, 5.0, 10.0}, + }) + DLQMessages = prometheus.NewCounter(prometheus.CounterOpts{ Name: "assignment_dlq_messages_total", Help: "Messages routed to the dead-letter topic.", @@ -36,6 +61,10 @@ func Register() { AssignmentsTotal, EndToEndLatency, MissingRequestedAt, + AssignmentDuplicates, + UnmatchedTotal, + UnmatchedDuplicates, + UnmatchedEndToEndLatency, DLQMessages, ) } diff --git a/services/assignment-service/internal/store/memory.go b/services/assignment-service/internal/store/memory.go index c6dcd4a..85c01cb 100644 --- a/services/assignment-service/internal/store/memory.go +++ b/services/assignment-service/internal/store/memory.go @@ -6,37 +6,83 @@ import ( "ridepulse/services/assignment-service/internal/domain" ) -// Memory is a simple thread-safe map keyed by ride ID. +// Memory is a simple thread-safe store of ride outcomes keyed by ride ID. // -// This is deliberately in-memory: assignments live only for the run of a -// single benchmark and are read back via the /assignments/ -// endpoint for verification. Persistence is not the point of this -// service — closing the pipeline so RPS benchmarks can measure end-to-end -// success is the point. +// This is deliberately in-memory: outcomes live only for the run of a +// single benchmark and are read back via the HTTP API for verification. +// Persistence is not the point of this service; closing the pipeline so +// RPS benchmarks can measure end-to-end success is the point. +// +// A ride is either matched or unmatched. Matched wins: it overwrites an +// earlier unmatched outcome, and a later unmatched one is ignored. type Memory struct { - mu sync.RWMutex - byRideID map[string]domain.Assignment + mu sync.RWMutex + matched map[string]domain.Assignment + unmatched map[string]domain.Outcome } func NewMemory() *Memory { - return &Memory{byRideID: make(map[string]domain.Assignment)} + return &Memory{ + matched: make(map[string]domain.Assignment), + unmatched: make(map[string]domain.Outcome), + } +} + +// PutMatched records a's ride as matched and reports whether it was new. +// A duplicate for an already-matched ride changes nothing and returns false. +func (m *Memory) PutMatched(a domain.Assignment) (isNew bool) { + m.mu.Lock() + defer m.mu.Unlock() + if _, ok := m.matched[a.RideID]; ok { + return false + } + m.matched[a.RideID] = a + delete(m.unmatched, a.RideID) + return true } -func (m *Memory) Put(a domain.Assignment) { +// PutUnmatched records rideID as unmatched and reports whether it was new. +// It is ignored if the ride is already matched or already unmatched. +func (m *Memory) PutUnmatched(rideID, reason string, failedAt int64) (isNew bool) { m.mu.Lock() - m.byRideID[a.RideID] = a - m.mu.Unlock() + defer m.mu.Unlock() + if _, ok := m.matched[rideID]; ok { + return false + } + if _, ok := m.unmatched[rideID]; ok { + return false + } + m.unmatched[rideID] = domain.Outcome{ + RideID: rideID, + Status: domain.StatusUnmatched, + Reason: reason, + FailedAt: failedAt, + } + return true } +// Get returns the assignment only when the ride is matched. func (m *Memory) Get(rideID string) (domain.Assignment, bool) { m.mu.RLock() - a, ok := m.byRideID[rideID] + a, ok := m.matched[rideID] m.mu.RUnlock() return a, ok } +// Status returns the full outcome for a ride, matched or unmatched. +func (m *Memory) Status(rideID string) (domain.Outcome, bool) { + m.mu.RLock() + defer m.mu.RUnlock() + if a, ok := m.matched[rideID]; ok { + return domain.Outcome{RideID: rideID, Status: domain.StatusMatched, Assignment: a}, true + } + o, ok := m.unmatched[rideID] + return o, ok +} + +// Count returns the number of matched rides (the benchmark reads this). func (m *Memory) Count() int { m.mu.RLock() defer m.mu.RUnlock() - return len(m.byRideID) + return len(m.matched) } diff --git a/services/assignment-service/internal/store/memory_test.go b/services/assignment-service/internal/store/memory_test.go new file mode 100644 index 0000000..ecdeb12 --- /dev/null +++ b/services/assignment-service/internal/store/memory_test.go @@ -0,0 +1,78 @@ +package store + +import ( + "testing" + + "ridepulse/services/assignment-service/internal/domain" +) + +func TestPutMatchedDedupes(t *testing.T) { + m := NewMemory() + first := domain.Assignment{RideID: "r1", DriverID: "d1", MatchedAT: 1, LatencyMs: 10} + if !m.PutMatched(first) { + t.Fatal("first PutMatched should be new") + } + if m.PutMatched(domain.Assignment{RideID: "r1", DriverID: "d2"}) { + t.Fatal("duplicate PutMatched should not be new") + } + got, ok := m.Get("r1") + if !ok || got != first { + t.Fatalf("Get = %+v, %v; want original %+v", got, ok, first) + } + if m.Count() != 1 { + t.Fatalf("Count = %d, want 1", m.Count()) + } +} + +func TestMatchedOverridesUnmatched(t *testing.T) { + m := NewMemory() + if !m.PutUnmatched("r1", "no_drivers", 5) { + t.Fatal("PutUnmatched should be new") + } + if !m.PutMatched(domain.Assignment{RideID: "r1", DriverID: "d1"}) { + t.Fatal("PutMatched after unmatched should be new") + } + st, ok := m.Status("r1") + if !ok || st.Status != domain.StatusMatched || st.Assignment.DriverID != "d1" { + t.Fatalf("Status = %+v, %v; want matched by d1", st, ok) + } + if m.Count() != 1 { + t.Fatalf("Count = %d, want 1", m.Count()) + } +} + +func TestUnmatchedIgnoredAfterMatchedOrUnmatched(t *testing.T) { + m := NewMemory() + m.PutMatched(domain.Assignment{RideID: "r1", DriverID: "d1"}) + if m.PutUnmatched("r1", "no_drivers", 5) { + t.Fatal("PutUnmatched after matched should be ignored") + } + if st, _ := m.Status("r1"); st.Status != domain.StatusMatched { + t.Fatalf("status = %q, want matched", st.Status) + } + + if !m.PutUnmatched("r2", "no_drivers", 7) { + t.Fatal("first PutUnmatched should be new") + } + if m.PutUnmatched("r2", "match_deadline", 9) { + t.Fatal("duplicate PutUnmatched should be ignored") + } + st, _ := m.Status("r2") + if st.Reason != "no_drivers" || st.FailedAt != 7 { + t.Fatalf("Status = %+v, want original reason and failedAt", st) + } +} + +func TestCountAndGetMatchedOnly(t *testing.T) { + m := NewMemory() + m.PutUnmatched("r1", "no_drivers", 5) + if m.Count() != 0 { + t.Fatalf("Count = %d, want 0 (unmatched not counted)", m.Count()) + } + if _, ok := m.Get("r1"); ok { + t.Fatal("Get should not return an unmatched ride") + } + if _, ok := m.Status("missing"); ok { + t.Fatal("Status should report unknown ride as missing") + } +} From 0878b3fc8ef33df1864d461f9d86512178f38a1d Mon Sep 17 00:00:00 2001 From: coderconnoisseur Date: Fri, 16 Oct 2026 00:29:48 +0530 Subject: [PATCH 5/9] rider-sim: batched ticker to exceed timer-resolution RPS cap --- .../cmd/rider-simulator/main.go | 6 +- .../internal/simulator/simulator.go | 77 +++++++++++++++---- .../internal/simulator/simulator_test.go | 51 ++++++++++++ 3 files changed, 117 insertions(+), 17 deletions(-) create mode 100644 services/rider-simulator/internal/simulator/simulator_test.go diff --git a/services/rider-simulator/cmd/rider-simulator/main.go b/services/rider-simulator/cmd/rider-simulator/main.go index fafa89b..ae52db5 100644 --- a/services/rider-simulator/cmd/rider-simulator/main.go +++ b/services/rider-simulator/cmd/rider-simulator/main.go @@ -4,7 +4,9 @@ import ( "context" "log" "os" + "os/signal" "strconv" + "syscall" "ridepulse/services/rider-simulator/internal/simulator" ) @@ -14,7 +16,8 @@ import ( // Falls back to the previous hardcoded defaults so a plain // `go run cmd/rider-simulator/main.go` still works. func main() { - ctx := context.Background() + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() url := envDefault("RIDER_URL", "http://localhost:8080/rides") rps := envInt("RIDER_RPS", 500) concurrency := envInt("RIDER_CONCURRENCY", 200) @@ -22,7 +25,6 @@ func main() { s := simulator.New(url) log.Printf("Starting rider simulator: url=%s rps=%d concurrency=%d", url, rps, concurrency) s.Run(ctx, rps, concurrency) - select {} } func envDefault(key, def string) string { diff --git a/services/rider-simulator/internal/simulator/simulator.go b/services/rider-simulator/internal/simulator/simulator.go index d458480..51e4239 100644 --- a/services/rider-simulator/internal/simulator/simulator.go +++ b/services/rider-simulator/internal/simulator/simulator.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "log" + "math" "math/rand/v2" "net/http" "sync/atomic" @@ -55,8 +56,9 @@ type RideRequest struct { } var requests int64 -var errors int64 +var errCount int64 var totalLatency int64 +var dropped int64 func (s *Simulator) sendRide(ctx context.Context) { req := RideRequest{} @@ -65,8 +67,8 @@ func (s *Simulator) sendRide(ctx context.Context) { req.Drop.Lat = boxMinLat + rand.Float64()*boxSpan req.Drop.Lng = boxMinLng + rand.Float64()*boxSpan body, _ := json.Marshal(req) - start:= time.Now() - resp,err:= s.client.Post( + start := time.Now() + resp, err := s.client.Post( s.url, "application/json", bytes.NewBuffer(body), @@ -75,39 +77,84 @@ func (s *Simulator) sendRide(ctx context.Context) { atomic.AddInt64(&requests, 1) atomic.AddInt64(&totalLatency, latency.Microseconds()) if err != nil || (resp != nil && resp.StatusCode >= 400) { - atomic.AddInt64(&errors, 1) + atomic.AddInt64(&errCount, 1) } if resp != nil { resp.Body.Close() } } + +// tickInterval is well above the ~15 ms Windows timer resolution, so the +// ticker fires reliably; each tick sends a batch sized by the Pacer. +const tickInterval = 20 * time.Millisecond + +// Pacer converts elapsed wall time into a whole number of sends at a target +// rate, carrying the fractional remainder to the next tick so the long-run +// rate is exact and a late tick catches up. +type Pacer struct { + rps float64 + acc float64 +} + +func NewPacer(rps int) *Pacer { + return &Pacer{rps: float64(rps)} +} + +// Next returns how many requests are due after elapsed time has passed. +func (p *Pacer) Next(elapsed time.Duration) int { + p.acc += p.rps * elapsed.Seconds() + n := int(math.Floor(p.acc)) + p.acc -= float64(n) + return n +} + +// Run sends rps requests per second until ctx is cancelled. If maxConcurrency +// requests are already in flight, the send is skipped and counted as dropped +// rather than blocking the ticker. func (s *Simulator) Run(ctx context.Context, rps int, maxConcurrency int) { - interval := time.Second / time.Duration(rps) // semaphore to avoid goroutine explosion sem := make(chan struct{}, maxConcurrency) // metrics go func() { ticker := time.NewTicker(1 * time.Second) - for range ticker.C { + defer ticker.Stop() + for { + select { + case <-ticker.C: + case <-ctx.Done(): + return + } r := atomic.SwapInt64(&requests, 0) - e := atomic.SwapInt64(&errors, 0) + e := atomic.SwapInt64(&errCount, 0) l := atomic.SwapInt64(&totalLatency, 0) + d := atomic.SwapInt64(&dropped, 0) var avg int64 if r > 0 { avg = l / r } - log.Printf("RPS: %d | Errors: %d | Avg Latency(us): %d", r, e, avg) + log.Printf("RPS: %d | Errors: %d | Dropped: %d | Avg Latency(us): %d", r, e, d, avg) } }() - ticker := time.NewTicker(interval) + pacer := NewPacer(rps) + ticker := time.NewTicker(tickInterval) + defer ticker.Stop() + last := time.Now() for { select { - case <-ticker.C: - sem <- struct{}{} // acquire slot - go func() { - defer func() { <-sem }() // release - s.sendRide(ctx) - }() + case now := <-ticker.C: + n := pacer.Next(now.Sub(last)) + last = now + for i := 0; i < n; i++ { + select { + case sem <- struct{}{}: // acquire slot + go func() { + defer func() { <-sem }() // release + s.sendRide(ctx) + }() + default: + atomic.AddInt64(&dropped, 1) + } + } case <-ctx.Done(): return } diff --git a/services/rider-simulator/internal/simulator/simulator_test.go b/services/rider-simulator/internal/simulator/simulator_test.go new file mode 100644 index 0000000..c8ad2d3 --- /dev/null +++ b/services/rider-simulator/internal/simulator/simulator_test.go @@ -0,0 +1,51 @@ +package simulator + +import ( + "testing" + "time" +) + +func TestPacerYieldsExactRPS(t *testing.T) { + // Tick gaps sum to exactly 1 s; the late patterns include one very late tick. + steady := func() []time.Duration { + d := make([]time.Duration, 50) + for i := range d { + d[i] = 20 * time.Millisecond + } + return d + }() + late := []time.Duration{ + 20 * time.Millisecond, 35 * time.Millisecond, 5 * time.Millisecond, + 140 * time.Millisecond, 20 * time.Millisecond, 250 * time.Millisecond, + 30 * time.Millisecond, 500 * time.Millisecond, + } + for name, gaps := range map[string][]time.Duration{"steady": steady, "late": late} { + var total time.Duration + for _, g := range gaps { + total += g + } + if total != time.Second { + t.Fatalf("%s: gaps sum to %v, want 1s", name, total) + } + for _, rps := range []int{50, 1000, 3000} { + p := NewPacer(rps) + sent := 0 + for _, g := range gaps { + sent += p.Next(g) + } + if sent < rps-1 || sent > rps+1 { + t.Errorf("%s rps=%d: sent %d, want %d±1", name, rps, sent, rps) + } + } + } +} + +func TestPacerCarriesFraction(t *testing.T) { + p := NewPacer(50) // 1 request per 20 ms, but 10 ms ticks yield 0.5 each + if n := p.Next(10 * time.Millisecond); n != 0 { + t.Fatalf("first tick = %d, want 0", n) + } + if n := p.Next(10 * time.Millisecond); n != 1 { + t.Fatalf("second tick = %d, want 1", n) + } +} From 6e6369fcf4c0cc25d129dcab698685bbe8fafbe9 Mon Sep 17 00:00:00 2001 From: coderconnoisseur Date: Fri, 16 Oct 2026 00:33:40 +0530 Subject: [PATCH 6/9] docs: unmatched rides, env config, rewind-safe commits --- README.md | 61 +++++++++++---- docs/at-least-once-consumers.md | 133 +++++++++++++++++++++++--------- docs/driver-state-model.md | 56 ++++++++++++-- docs/rps-benchmarking-setup.md | 108 ++++++++++++++++++++++---- 4 files changed, 283 insertions(+), 75 deletions(-) diff --git a/README.md b/README.md index d35d6c0..5ddb177 100644 --- a/README.md +++ b/README.md @@ -39,8 +39,8 @@ RidePulse follows an event-driven microservices design where each component has | **Rider Simulator** | Generates ride requests at configurable RPS against the HTTP gateway | | **API Gateway** | Accepts `POST /rides`, publishes `ride.requested` to Kafka | | **Pricing Service** | Consumes `ride.requested`, computes fare, publishes `ride.priced`; failures retried, then dead-lettered to `ride.requested.dlq` | -| **Matching Service** | Consumes `ride.priced`, runs geospatial search + atomic driver locking, assigns the driver (idempotent per ride), publishes `ride.matched`; infrastructure failures retried, then dead-lettered to `ride.priced.dlq` | -| **Assignment Service** | Consumes `ride.matched`, records the assignment, exposes end-to-end latency metrics; dead-letters to `ride.matched.dlq` | +| **Matching Service** | Consumes `ride.priced`, runs geospatial search + atomic driver locking, assigns the driver (idempotent per ride), publishes `ride.matched`; business failures (`no_drivers`, `match_deadline`) publish `ride.unmatched`; infrastructure failures retried, then dead-lettered to `ride.priced.dlq` | +| **Assignment Service** | Consumes `ride.matched` and `ride.unmatched`, records each ride's outcome once (duplicates ignored), exposes end-to-end latency metrics and `GET /rides/`; dead-letters to `ride.matched.dlq` / `ride.unmatched.dlq` | | **Driver Simulator** | Continuously updates live driver locations in Redis GEO; keeps assigned drivers out of the GEO set and releases them after `DRIVER_TRIP_DURATION` | | **Kafka** | Decouples every stage of the pipeline | | **Redis** | GEO indexing, distributed driver locking (Lua safe-unlock) and driver assignment state | @@ -71,14 +71,15 @@ RiderSimulator ──HTTP POST /rides──▶ API Gateway │ (batch losers explicitly released) ├── atomic assign: driver:assigned, ride:driver, │ remove from GEO (replays return same driver) - ▼ - Kafka: ride.matched + ├─ matched ──▶ Kafka: ride.matched + └─ no driver / deadline ──▶ Kafka: ride.unmatched │ ▼ Assignment Service │ ▼ - records assignment; end_to_end_latency + records the ride's outcome once; end_to_end_latency + (GET /rides/ returns matched or unmatched) ``` --- @@ -109,13 +110,13 @@ Full write-up: [shuffle-nearest-first-fix.md](docs/shuffle-nearest-first-fix.md) ### 4. Progressive Parallel Locking with hierarchical deadlines -Each match has its own 800 ms budget; each batch of 5 parallel `SETNX` attempts has a 200 ms budget nested inside it; Kafka publish gets its own independent 300 ms budget so a slow match can't starve the publish RTT. +Each match has its own budget (`MATCH_DEADLINE`, default 500 ms; the write-up below was done at 800 ms); each batch of 5 parallel `SETNX` attempts has a 200 ms budget nested inside it; Kafka publish gets its own independent 300 ms budget so a slow match can't starve the publish RTT. Full write-up: [deadline-fix.md](docs/deadline-fix.md). ### 5. At-least-once consumers with ordered commits and DLQs -Pricing, matching and assignment share one runner (`pkg/kafkaconsumer`). An offset is committed only after its message, and every earlier message of the partition, was handled or dead-lettered, so a crash redelivers instead of losing work. Failed handlers are retried with backoff, then written to `.dlq` with `x-error` / `x-original-*` headers; shutdown drains in-flight messages for up to 10 s. +Pricing, matching and assignment share one runner (`pkg/kafkaconsumer`). An offset is committed only after its message, and every earlier message of the partition, was handled or dead-lettered, so a crash redelivers instead of losing work. Failed handlers are retried with backoff, then written to `.dlq` with `x-error` / `x-original-*` headers; a handler panic is dead-lettered without retries instead of crashing the process; a consumer-group rewind starts a new tracker epoch so stale completions cannot corrupt commits; shutdown drains in-flight messages for up to 10 s. Full write-up: [at-least-once-consumers.md](docs/at-least-once-consumers.md). @@ -136,9 +137,10 @@ Every stage exposes Prometheus metrics on its own port. `RequestedAt` is stamped Key exports: - `gateway_request_duration_seconds`, `gateway_requests_total`, `gateway_publish_errors_total` - `pricing_processing_duration_seconds`, `pricing_events_published_total` -- `matching_latency_seconds`, `matching_success_total`, `matching_failures_total` +- `matching_latency_seconds`, `matching_success_total`, `matching_failures_total`, `match_unmatched_published_total{reason}` - `driver_lock_conflict_total`, `driver_lock_released_phantom_total` -- `end_to_end_latency_seconds`, `assignments_total` +- `end_to_end_latency_seconds`, `assignments_total` (each ride counted once), `assignment_duplicates_total` +- `unmatched_total{reason}`, `unmatched_duplicates_total`, `unmatched_end_to_end_latency_seconds` - `pricing_retries_total`, `matching_retries_total`, `pricing_dlq_messages_total`, `matching_dlq_messages_total`, `assignment_dlq_messages_total`, `match_idempotent_replays_total` --- @@ -162,7 +164,7 @@ Headline from that run (Windows dev box, single-partition Kafka, single Redis, ~ | Peak "healthy" throughput (>95 % success) | ~50 RPS | | Successful `matching_latency_seconds` p95 | stayed 33–84 ms across every step | -The primary bottleneck is matcher pool starvation from slow-fail matches: under contention, ~30 % of matches consume workers for the full 800 ms `matchDeadline`, blowing past `NumCPU × 5` worker capacity. Fixing this (fail-fast on batch 1 saturation, multi-partition Kafka topics, larger pool) is the next set of unlocks. +The primary bottleneck in that run was matcher pool starvation from slow-fail matches: under contention, ~30 % of matches consumed workers for the full 800 ms match deadline, blowing past the `NumCPU × 5` worker capacity of the time. The matcher defaults have since changed (`MATCH_WORKERS` = `NumCPU × 25`, `MATCH_DEADLINE` = 500 ms) but the new values have **not been measured**; the ladder must be re-run. Fail-fast on batch 1 saturation and multi-partition Kafka topics are still open. See [rps-benchmarking-setup.md](docs/rps-benchmarking-setup.md) for the runbook, ladder shape, PromQL queries, and interpretation notes. @@ -170,11 +172,12 @@ See [rps-benchmarking-setup.md](docs/rps-benchmarking-setup.md) for the runbook, ## Known limitations and next steps -- **Matcher pool starvation.** The pool is `runtime.NumCPU() * 5` workers and each match may hold one for up to the 800 ms `matchDeadline`. Under contention, slow-fail matches occupy workers until the deadline and the queue backs up. Next: fail fast when the first batch is fully contended instead of waiting out the deadline, then re-measure pool size. -- **Single-partition Kafka topics.** One partition per topic caps consumer parallelism per service. Next: partition by ride ID and run multiple consumers per stage. -- **Hardcoded endpoints.** Services connect to `localhost:9092` and `localhost:6379` directly; only the rider simulator and driver simulator (`REDIS_ADDR`) read their config from environment variables. Next: env-based config so services can run inside Compose. -- **In-memory assignment store.** The assignment service keeps results in a map for benchmark verification; nothing is persisted. -- **Test coverage is unit-level.** Unit tests cover the tier-preserving shuffle, the at-least-once consumer runner (ordered commits, retries, DLQ, drain) and the Redis Lua scripts (locking, assignment, release) against an in-process Redis (miniredis). Nothing yet exercises real Kafka/Redis end to end. Next: integration tests against containerized Kafka and Redis. +- **Matcher pool starvation: new defaults, still unmeasured.** The pool now defaults to `runtime.NumCPU() * 25` workers (`MATCH_WORKERS`) and the match deadline to 500 ms (`MATCH_DEADLINE`), down from `NumCPU * 5` and 800 ms. Slow-fail matches still hold a worker until the deadline, and no ladder has been run with these values, so they are a starting point, not a result. Next: re-run the ladder, and fail fast when the first batch is fully contended instead of waiting out the deadline. +- **Single-partition Kafka topics.** One partition per topic caps consumer parallelism per service. Next: create topics with more partitions, partition by ride ID and run multiple consumers per stage. Until then the rewind handling below is not exercised. +- **Rewind detection is partial.** The offset tracker detects a rewind only when a re-fetched offset is at or below the last offset it tracked. A partition re-assigned starting *above* that offset is not detected, so the committed offset can move backwards and messages are redelivered. That stays within at-least-once, and handlers absorb duplicates. +- **Two readers, one consumer group.** assignment-service reads `ride.matched` and `ride.unmatched` with two readers in the same group (`assignment-service`). Partitions are assigned per topic correctly, but a join or leave by either reader triggers a rebalance of both. +- **In-memory assignment store.** The assignment service keeps ride outcomes (matched and unmatched) in maps for benchmark verification; nothing is persisted, so after a restart a redelivered event is counted as new. +- **Test coverage is unit-level.** Unit tests cover the tier-preserving shuffle, the at-least-once consumer runner (ordered commits, retries, DLQ, drain, rewind epochs, panic recovery) and the Redis Lua scripts (locking, assignment, release) against an in-process Redis (miniredis). Nothing yet exercises real Kafka/Redis end to end: the unmatched path, the epoch-stamped commits and the new matcher defaults have not been run against a live cluster. Next: integration tests against containerized Kafka and Redis. --- @@ -243,6 +246,24 @@ Prometheus scrape targets: Confirm all four are `up` at http://localhost:9090/targets. +**Configuration.** Every service reads its endpoints from environment variables; unset or blank values fall back to the defaults below, so the commands above work unchanged on one host. `KAFKA_BROKERS` is a comma-separated list. An invalid value makes the service exit at startup. + +| Variable | Default | Used by | Meaning | +|---|---|---|---| +| `KAFKA_BROKERS` | `localhost:9092` | api-gateway, pricing, matching, assignment | Comma-separated broker list. | +| `REDIS_ADDR` | `localhost:6379` | matching, driver-simulator | Redis address. | +| `HTTP_ADDR` | `:8080` (api-gateway), `:2114` (assignment) | api-gateway, assignment | HTTP listen address (also serves `/metrics`). | +| `METRICS_ADDR` | `:2115` (pricing), `:2112` (matching) | pricing, matching | Metrics listen address. | +| `MATCH_WORKERS` | `runtime.NumCPU() * 25` | matching | Matcher worker-pool size; positive integer. | +| `MATCH_DEADLINE` | `500ms` | matching | Total per-match budget (Go duration); must exceed the 200 ms batch deadline. | +| `DRIVER_COUNT` | `1000` | driver-simulator | Number of simulated drivers. | +| `DRIVER_TRIP_DURATION` | `5s` | driver-simulator | How long an assigned driver stays busy (Go duration). | +| `RIDER_URL` | `http://localhost:8080/rides` | rider-simulator | Gateway endpoint to POST rides to. | +| `RIDER_RPS` | `500` | rider-simulator | Target requests per second. | +| `RIDER_CONCURRENCY` | `200` | rider-simulator | Max in-flight requests; sends beyond it are counted as dropped. | + +The `MATCH_WORKERS` and `MATCH_DEADLINE` defaults have not been benchmarked; treat them as untested until the ladder is re-run. + ### 3. Send load **Automated ladder (recommended):** @@ -277,15 +298,21 @@ histogram_quantile(0.95, sum by (le) (rate(end_to_end_latency_seconds_bucket[1m] # Where matches are failing (contention vs deadline) rate(driver_lock_conflict_total[1m]) rate(driver_lock_released_phantom_total[1m]) + +# Rides that found no driver, by reason (no_drivers | match_deadline) +sum by (reason) (rate(unmatched_total[1m])) ``` Verify pipeline closure by querying assignment-service directly: ```bash -curl http://localhost:2114/count # total assignments observed -curl http://localhost:2114/assignments/ # look up a specific ride +curl http://localhost:2114/count # total matched rides observed +curl http://localhost:2114/assignments/ # look up a matched ride (404 if unmatched) +curl http://localhost:2114/rides/ # outcome of any ride: matched or unmatched ``` +`GET /rides/` returns `{"ride_id","status":"matched","driver_id","matched_at","latency_ms"}` or `{"ride_id","status":"unmatched","reason","failed_at"}` (`reason` is `no_drivers` or `match_deadline`), and 404 for a ride with no outcome yet. + ### 5. Run the checks CI runs Each service is its own Go module (tied together by `services/go.work` for local development). CI runs `go vet`, `go build`, and `go test` in every module; the unit tests need no Kafka or Redis: diff --git a/docs/at-least-once-consumers.md b/docs/at-least-once-consumers.md index 19aa98f..72a32f4 100644 --- a/docs/at-least-once-consumers.md +++ b/docs/at-least-once-consumers.md @@ -3,7 +3,8 @@ > **Status:** local-only note. Not intended for the public repo. > Documents the offset-commit and failure-handling bug shared by the three > Kafka consumers, the shared runner that replaced it -> (`pkg/kafkaconsumer`), and what is still not covered. +> (`pkg/kafkaconsumer`), the epoch mechanism that handles consumer-group +> rewinds, handler panic recovery, and what is still not covered. --- @@ -64,7 +65,7 @@ fetched before N on the same partition has also reached a terminal state. A slow message at the head holds the watermark back; later messages that already finished stay in the `done` set and are committed in one go as soon as the head completes. Commits are serialized under a mutex, so the -committed offset of a partition is monotonic. +committed offset of a partition is monotonic within an epoch (see 3f). A message reaches a terminal state in exactly one of three ways: @@ -96,6 +97,7 @@ The DLQ topic is the original topic plus `.dlq`: | `ride.requested` | `ride.requested.dlq` | pricing-service | | `ride.priced` | `ride.priced.dlq` | matching-service | | `ride.matched` | `ride.matched.dlq` | assignment-service | +| `ride.unmatched` | `ride.unmatched.dlq` | assignment-service | The DLQ record keeps the original key and value and any original headers, and appends four more: @@ -142,7 +144,50 @@ the DLQ writer and the publisher (matching also closes Redis). Fatal errors (a fetch error not caused by shutdown, a synchronous commit error, a DLQ write failure) take the same drain path and are returned to `main`. -### 3e. Reader settings +### 3e. Handler panics + +Each handler attempt runs under a `recover`. A panic is logged with its +stack, wrapped as `Permanent(handler panic: ...)` and treated like any other +permanent failure: no retries, straight to the DLQ, offset committed once +the DLQ write succeeds. A handler that panics deterministically therefore +costs one DLQ record per message instead of a crash loop. The panic text ends +up in the record's `x-error` header, and the service's `*_dlq_messages_total` +counter increments (the `OnRetry` hook does not fire, since there is no +retry). Only panics on the handler goroutine are caught; a panic in a +goroutine the handler itself starts still crashes the process. + +### 3f. Rewinds and epochs + +A consumer-group rebalance makes kafka-go re-fetch from the last committed +offset, which can be lower than offsets the tracker already holds. Left +alone, the old pending entries and the new copies would collide and the +watermark would stall. + +`partitionState` therefore carries an `epoch` and `last`, the highest offset +tracked in the current epoch. `track` compares each fetched offset with +`last`: + +- offset greater than `last`: ordinary forward fetch, same epoch; +- offset at or below `last`: the partition rewound. The tracker logs it, + increments the epoch, discards the old `pending` and `done` state and starts + tracking from the re-fetched message. + +`track` returns the epoch, the runner carries it with the message to its +worker, and `done(msg, epoch)` ignores a completion whose epoch is not the +current one (or whose offset is no longer pending). Copies that were in +flight when the rewind happened still run to completion, but cannot move the +watermark or trigger a commit; the re-fetched copy is processed again and +commits normally. The cost is that such messages are handled twice, which the +handlers already tolerate (section 5). + +What this does **not** detect: a rewind is only noticed when the re-fetched +offset is `<= last`. If a partition is re-assigned to this consumer starting +*above* the last tracked offset, nothing looks wrong to the tracker, the old +pending entries stay, and the committed offset can end up behind where it +was. Messages between the two are redelivered. That is still at-least-once, +but it is redelivery the tracker did not recognize and could not suppress. + +### 3g. Reader settings All three readers use `CommitInterval: 1s`. `CommitMessages` therefore queues the offset and a background loop flushes it once a second, instead @@ -164,15 +209,28 @@ why the handlers have to tolerate duplicates (section 5). | Service | Workers | Decode failure | Handler failure | |---|---|---|---| | pricing | `NumCPU` | Permanent → `ride.requested.dlq` | Publish error retried, then DLQ | -| matching | `NumCPU × 5` | Permanent → `ride.priced.dlq` | See below | -| assignment | 4 | Permanent → `ride.matched.dlq` | Cannot fail after decode | +| matching | `MATCH_WORKERS` (default `NumCPU × 25`) | Permanent → `ride.priced.dlq` | See below | +| assignment (`ride.matched`) | 4 | Permanent → `ride.matched.dlq` | Cannot fail after decode | +| assignment (`ride.unmatched`) | 4 | Permanent → `ride.unmatched.dlq` | Cannot fail after decode | + +All handler panics, in any service, take the permanent path (3e). + +assignment-service runs two runners, one per topic, with two readers in the +same consumer group (`assignment-service`). Each topic's partitions are +assigned correctly, but a reader joining or leaving triggers a rebalance +that affects both. If one runner fails fatally the other is cancelled and +`Run` returns the first error. **Matching** splits `Match` errors in `app.go`: - `ErrNoDriversAvailable`, and `context.DeadlineExceeded` when it is the - outer match deadline (`matchCtx.Err() != nil`), are business failures. The handler logs, returns `nil`, and the offset is committed. - The ride is currently dropped: nothing tells the rider or downstream - systems. Emitting a `ride.unmatched` event is roadmap item 1c. + outer match deadline (`matchCtx.Err() != nil`), are business failures. + The handler logs and publishes a `ride.unmatched` event (reason + `no_drivers` or `match_deadline`, 300 ms publish budget), then returns + `nil` and the offset is committed. If that publish fails the error is + returned, so the runner retries and eventually dead-letters it like any + other infrastructure error. See + [driver-state-model.md](driver-state-model.md#4b-the-unmatched-path). - Any other `Match` error (Redis failures, and any other `DeadlineExceeded` such as the detached assign timeout) is returned, so the runner retries it and eventually dead-letters it. Retrying is safe because @@ -186,30 +244,29 @@ why the handlers have to tolerate duplicates (section 5). ## 5. Trade-offs and known behaviour - **Duplicates are possible.** A crash after the handler succeeded but - before the 1 s commit flush redelivers the message. Pricing is a pure - function plus a publish, so a duplicate `ride.priced` is harmless to - pricing itself but flows downstream. Matching deduplicates by ride - through `ride:driver:`. **Assignment-service has no dedupe**: a - duplicate `ride.matched` overwrites the in-memory record for the ride - (it is keyed by ride ID) but increments `assignments_total` and the - latency histogram again. Full dedupe there is roadmap item 1e. + before the 1 s commit flush redelivers the message, as does a rewind + (3f). Pricing is a pure function plus a publish, so a duplicate + `ride.priced` is harmless to pricing itself but flows downstream. Matching + deduplicates by ride through `ride:driver:`. Assignment-service + deduplicates in its store: the first `ride.matched` for a ride is recorded + and counted in `assignments_total` and `end_to_end_latency_seconds`; later + ones only increment `assignment_duplicates_total`. A ride has one outcome: + matched wins over unmatched (a `ride.matched` replaces an earlier + unmatched record, a later `ride.unmatched` is ignored and counted in + `unmatched_duplicates_total`). The store is in memory, so after an + assignment-service restart a redelivered event is new again. - **One slow message delays commits, not processing.** Other workers keep going; the watermark simply does not advance past the stuck message, so a crash replays everything after it. Retries are bounded (4 attempts, ≤ 1 s backoff), so a stuck head is bounded too. -- **Handler panics are not recovered.** A panic in a handler crashes the - process, and the message is redelivered on restart. If it panics - deterministically, that is a crash loop rather than a DLQ entry. -- **Partition rebalances are not handled by the tracker.** It assumes one - assigned partition set for the life of the process. If a rebalance moved a - partition away and back, the broker may rewind to a lower offset than the - tracker has seen. After a consumer-group generation change kafka-go rewinds - to the last committed offset, and the tracker does not handle re-fetched - offsets: the commit watermark can stall (nothing is lost, because the - messages are redelivered on restart) and pending entries grow until the - process restarts. The topics are single-partition and each service runs as - a single instance today, so this cannot happen yet. An epoch-stamped tracker - reset is planned before the Phase 2 multi-partition work. +- **Rewind detection is partial.** Epochs (3f) handle a rewind to an offset + `<= last`. A partition re-assigned starting above the last tracked offset + is not detected, and the committed offset can move backwards, causing + redelivery. At-least-once still holds. The topics are single-partition and + each service runs as a single instance today, so neither case has been + seen outside unit tests. +- **Panic recovery covers the handler goroutine only** (3e). A panic in a + goroutine the handler spawns is not caught. - **Drain timeout abandons work.** After `DrainTimeout` the runner returns without waiting for stuck handlers; they are cancelled and their messages are redelivered. @@ -223,6 +280,8 @@ why the handlers have to tolerate duplicates (section 5). | `pricing_retries_total`, `matching_retries_total` | Handler attempts that failed and were retried. | | `pricing_dlq_messages_total`, `matching_dlq_messages_total`, `assignment_dlq_messages_total` | Messages written to the service's DLQ topic. Should stay at zero in a healthy run. | | `match_idempotent_replays_total` | Matching saw a ride that already had a driver (retry or redelivery). Non-zero is evidence of redelivery being absorbed. | +| `match_unmatched_published_total{reason}` | `ride.unmatched` events matching published, by `no_drivers` / `match_deadline`. | +| `assignment_duplicates_total`, `unmatched_duplicates_total` | Redelivered `ride.matched` / `ride.unmatched` events assignment-service recognized and ignored. | | `pricing_events_consumed_total` | Handler **attempts**, retries included. Compare with `pricing_events_published_total`. | ```promql @@ -242,14 +301,14 @@ clean shutdown and restart. ## 7. What remains -- **1c — `ride.unmatched`.** Business failures in matching are committed and - forgotten. -- **1e — dedupe in assignment-service.** Duplicates currently double-count - metrics. -- **Rebalance-safe tracker.** Needed before running more than one instance - or more than one partition. -- **Handler panic recovery** into the retry/DLQ path. -- **Phase 2 throughput work** (multi-partition topics, matcher pool - starvation from slow-fail matches). +- **Full rewind detection.** Re-assignment above the last tracked offset is + not recognized (3f). +- **Phase 2 throughput work.** Multi-partition topics, and the batch-1 + fail-fast for slow-fail matches in matching. The `MATCH_WORKERS` / + `MATCH_DEADLINE` defaults (`NumCPU × 25`, 500 ms) are unmeasured. +- **Persistent assignment store**, so dedupe survives a restart. +- **No end-to-end run yet.** The epoch handling, panic path, `ride.unmatched` + flow and assignment dedupe are covered by hermetic unit tests only; none of + it has been run against real Kafka and Redis. - **Benchmark ladder must be re-run.** `tools/benchmark/benchmark-results.csv` predates these changes; see the note in the README. diff --git a/docs/driver-state-model.md b/docs/driver-state-model.md index 7b1a517..5b193fb 100644 --- a/docs/driver-state-model.md +++ b/docs/driver-state-model.md @@ -112,7 +112,7 @@ match again. Consumer lag is nowhere near that today. `AssignDriver` runs on a context detached from the match deadline (`context.WithoutCancel`, then a 200 ms timeout). Without that, a match -that spent most of its 800 ms budget could have the assign cut off and the +that spent most of its match budget (`MATCH_DEADLINE`, default 500 ms) could have the assign cut off and the caller would believe the ride failed while Redis might have assigned it. Be honest about what the 200 ms bounds: the go-redis client is built with @@ -133,9 +133,48 @@ error classification is what makes that safe: only the outer match deadline failure and return the error. matching-service's handler returns it, so the runner retries; the retry replays via the record if the assign had in fact landed. -- `ErrNoDriversAvailable` / `context.DeadlineExceeded` → business failure - (offset committed, ride currently dropped; `ride.unmatched` is roadmap - item 1c). +- `ErrNoDriversAvailable` / `context.DeadlineExceeded` (outer match + deadline) → business failure: matching publishes `ride.unmatched` and + commits the offset (section 4b). + +### 4b. The unmatched path + +When matching ends without a driver, the handler publishes a +`ride.unmatched` event instead of dropping the ride: + +| Field | Meaning | +|---|---| +| `RideID` | The ride (also the Kafka key). | +| `Reason` | `no_drivers` (`ErrNoDriversAvailable`) or `match_deadline` (the outer match deadline expired). | +| `RequestedAt` | UnixNano stamp from the gateway, carried through for latency. | +| `FailedAt` | UnixNano time the failure was decided. | + +Publishing has its own 300 ms budget, separate from the match deadline. If +the publish fails the handler returns the error, the runner retries it and +eventually dead-letters the `ride.priced` message, as for any other +infrastructure failure; the offset is committed only after the event is out. +`match_unmatched_published_total{reason}` counts successful publishes. The +writer is created with `AllowAutoTopicCreation`, so the first unmatched ride +creates the topic; assignment-service dead-letters undecodable events to +`ride.unmatched.dlq`. + +An unmatched ride writes no Redis state: no `ride:driver` record, no assigned +key, nothing locked (phantom locks are released as usual). That has two +consequences: + +- A redelivered `ride.priced` for an unmatched ride runs `Match` again, and + may now succeed if drivers have been released in the meantime. The ride + can then produce `ride.matched` after `ride.unmatched`. +- assignment-service therefore treats matched as final: `ride.matched` + replaces an earlier unmatched record, a later `ride.unmatched` for a + matched ride is ignored (`unmatched_duplicates_total`), and so is a second + `ride.unmatched` for the same ride. `GET /rides/` on + assignment-service reports whichever outcome is current. + +assignment-service counts unmatched rides in `unmatched_total{reason}` and +`unmatched_end_to_end_latency_seconds`, separately from `assignments_total` +and `end_to_end_latency_seconds`, which now count each matched ride once +(redeliveries go to `assignment_duplicates_total`). --- @@ -191,12 +230,13 @@ should be about `DRIVER_COUNT` (locked drivers are still in the set). ## 7. What remains -- **1c — `ride.unmatched`** for rides that find no driver. -- **1e — dedupe in assignment-service**, so a replayed `ride.matched` does - not double-count. +- **Persistent outcome store.** assignment-service dedupes in memory, so a + restart forgets which rides it has seen. - **Phase 2 throughput work.** With drivers now busy for the trip duration, the saturation point of the ladder moves; the supply model is also a new tuning dimension (`DRIVER_COUNT` against `DRIVER_TRIP_DURATION`). - **Hard Redis time bounds** (`ContextTimeoutEnabled`), see section 4. - **Benchmark ladder must be re-run** before any number is compared with the - committed CSV. + committed CSV. The matcher defaults (`MATCH_WORKERS` = `NumCPU × 25`, + `MATCH_DEADLINE` = 500 ms) are unmeasured, and the unmatched path has not + been exercised against real Kafka and Redis. diff --git a/docs/rps-benchmarking-setup.md b/docs/rps-benchmarking-setup.md index 2f5d7d3..e783140 100644 --- a/docs/rps-benchmarking-setup.md +++ b/docs/rps-benchmarking-setup.md @@ -45,6 +45,16 @@ correctness during a benchmark: - `GET /assignments/` — returns the stored assignment (JSON) or 404. - `GET /count` — plaintext count of assignments observed so far. +- `GET /rides/` — outcome of a ride, matched or unmatched (added + with the unmatched path, below), or 404. + +> **Update.** assignment-service now also consumes `ride.unmatched` (second +> reader, same consumer group), records each ride's outcome once and ignores +> duplicates. `assignments_total` and `end_to_end_latency_seconds` count each +> matched ride once; redeliveries go to `assignment_duplicates_total`. +> Unmatched rides are counted in `unmatched_total{reason}` and +> `unmatched_end_to_end_latency_seconds`. See +> [driver-state-model.md](driver-state-model.md#4b-the-unmatched-path). Layout: @@ -271,6 +281,47 @@ RIDER_RPS=300 RIDER_CONCURRENCY=120 go run cmd/rider-simulator/main.go `RIDER_URL` (defaults `http://localhost:8080/rides`) is the third env-var handle. +The rider simulator logs once a second: + +``` +RPS: | Errors: | Dropped: | Avg Latency(us): +``` + +`Dropped` counts sends skipped because `RIDER_CONCURRENCY` requests were +already in flight. The simulator never blocks its ticker on a slow gateway, +so a non-zero `Dropped` means the offered load was below the requested +`RIDER_RPS` and the step is measuring the concurrency cap rather than the +pipeline. Raise `RIDER_CONCURRENCY` (or investigate gateway latency) until it +stays at zero. `actual_rps_in` in the CSV is measured at the gateway and +already reflects any shortfall. + +#### Environment variables (all services) + +Every service now reads its endpoints from the environment; unset or blank +values use the defaults. Invalid values make the service exit at startup. + +| Variable | Default | Used by | Meaning | +|---|---|---|---| +| `KAFKA_BROKERS` | `localhost:9092` | api-gateway, pricing, matching, assignment | Comma-separated broker list. | +| `REDIS_ADDR` | `localhost:6379` | matching, driver-simulator | Redis address. | +| `HTTP_ADDR` | `:8080` (api-gateway), `:2114` (assignment) | api-gateway, assignment | HTTP listen address (also serves `/metrics`). | +| `METRICS_ADDR` | `:2115` (pricing), `:2112` (matching) | pricing, matching | Metrics listen address. | +| `MATCH_WORKERS` | `runtime.NumCPU() * 25` | matching | Matcher worker-pool size; positive integer. | +| `MATCH_DEADLINE` | `500ms` | matching | Total per-match budget (Go duration); must exceed the 200 ms batch deadline. | +| `RIDER_URL` | `http://localhost:8080/rides` | rider-simulator | Gateway endpoint. | +| `RIDER_RPS` | `500` | rider-simulator | Target requests per second. | +| `RIDER_CONCURRENCY` | `200` | rider-simulator | Max in-flight requests. | +| `DRIVER_COUNT` | `1000` | driver-simulator | Number of simulated drivers. | +| `DRIVER_TRIP_DURATION` | `5s` | driver-simulator | How long an assigned driver stays busy. | + +**The ladder must be re-run.** Besides the driver-supply change below, the +matcher defaults moved from `NumCPU * 5` workers and an 800 ms deadline to +`NumCPU * 25` and 500 ms, and the rider simulator now paces with a batched +ticker. None of these values has been measured: they are a starting point. +Record `MATCH_WORKERS`, `MATCH_DEADLINE`, `DRIVER_COUNT` and +`DRIVER_TRIP_DURATION` with every result set, and vary one at a time. The +committed CSV predates all of it. + #### Driver supply and DLQ topics (added with the correctness work) Drivers are no longer instantly reusable: a matched driver is removed from @@ -289,7 +340,8 @@ record them with the results. `tools/benchmark/benchmark-results.csv` predates this change and is not comparable; re-run the ladder. Each consumer dead-letters failures to `.dlq` (`ride.requested.dlq`, -`ride.priced.dlq`, `ride.matched.dlq`), auto-created on first write. A +`ride.priced.dlq`, `ride.matched.dlq`, `ride.unmatched.dlq`), auto-created on +first write; a handler panic is dead-lettered too. A healthy run leaves them empty; check `pricing_dlq_messages_total`, `matching_dlq_messages_total` and `assignment_dlq_messages_total` after each step. See [at-least-once-consumers.md](at-least-once-consumers.md) and @@ -303,6 +355,9 @@ curl -s http://localhost:2114/count # Fetch a specific ride's assignment curl -s http://localhost:2114/assignments/ | jq + +# Outcome of any ride: status "matched" or "unmatched" (with reason) +curl -s http://localhost:2114/rides/ | jq ``` If `/count` is much lower than the number of `POST /rides` you sent, @@ -360,9 +415,25 @@ rate(driver_lock_released_phantom_total[1m]) - Rising `driver_lock_released_phantom_total` → phantom-lock guard is actively doing work. Good sign, expected under contention. - Rising `matching_failures_total` with high `driver_lock_conflict_total` - → the outer match deadline (800 ms) isn't enough to traverse all four - batches for these rides. Consider larger radius or fewer batches - before failing. + → the outer match deadline (`MATCH_DEADLINE`, default 500 ms) isn't + enough to traverse all the batches for these rides. Consider larger radius + or fewer batches before failing. + +Failures are now visible downstream as well: matching publishes +`ride.unmatched` and assignment-service counts it. + +``` +sum by (reason) (rate(match_unmatched_published_total[1m])) +sum by (reason) (rate(unmatched_total[1m])) +histogram_quantile(0.95, sum by (le) (rate(unmatched_end_to_end_latency_seconds_bucket[1m]))) +``` + +`reason="no_drivers"` is supply exhaustion (check `DRIVER_COUNT` against +`DRIVER_TRIP_DURATION`); `reason="match_deadline"` is the match budget running +out. The two counters should track each other; a gap means unmatched events +are in flight or in `ride.unmatched.dlq`. Note that `success_rate_pct` in the +CSV is `assignments_rps / actual_rps_in`, so unmatched rides count as +failures there. ### 4e. Retries, dead letters and replays @@ -373,11 +444,15 @@ increase(pricing_dlq_messages_total[5m]) increase(matching_dlq_messages_total[5m]) increase(assignment_dlq_messages_total[5m]) rate(match_idempotent_replays_total[1m]) +rate(assignment_duplicates_total[1m]) +rate(unmatched_duplicates_total[1m]) ``` Any DLQ increase invalidates the step's success rate; inspect the `x-error` header on the DLQ record. `match_idempotent_replays_total` above zero means a retry or redelivery was absorbed without assigning a second driver. +`assignment_duplicates_total` and `unmatched_duplicates_total` above zero mean +assignment-service absorbed a redelivery without double-counting. ### 4f. Publish-error attribution @@ -421,10 +496,11 @@ You will very likely hit one of these first: at 1000 drivers × 500 ms tick. Under matcher load, GEORADIUS queries compete with those writes on a single Redis connection. Fix: pipeline the GEOADDs and/or use a Redis pool. -3. **Windows ticker granularity in rider-simulator** — `time.Second / - 1000 = 1 ms` ticker won't fire that fast on Windows (~15 ms - resolution). At requested-RPS > ~600 you'll see actual RPS plateau - even though the code says otherwise. +3. **Rider-simulator concurrency cap** — the simulator ticks every 20 ms + (above Windows' ~15 ms timer resolution) and sends a batch sized to the + target rate, so the old plateau from a 1 ms ticker is gone. If + `RIDER_CONCURRENCY` is too low for the gateway's latency, sends are + skipped instead; watch the `Dropped` counter in its log. Whichever of these shows up first is your real first bottleneck. Everything downstream that "seemed" slow was masked by it. @@ -485,14 +561,20 @@ of scope for the "can we measure at all?" pass: explicitly create topics with `partitions=8` (or so) in a bootstrap step in `docker-compose.yml`, and run multiple matching-service / pricing-service instances. -- **Env-var config**. Most hosts are still hardcoded (`localhost:9092`, - `localhost:6379`, etc.; driver-simulator now reads `REDIS_ADDR`). Fine for a single-host benchmark; blocks - running any service on a different box. +- ~~Env-var config~~ is done: every service reads `KAFKA_BROKERS`, + `REDIS_ADDR`, `HTTP_ADDR` / `METRICS_ADDR` and, for matching, + `MATCH_WORKERS` / `MATCH_DEADLINE` (table in section 3c). +- **Fail-fast on batch 1.** A fully contended first batch still waits out the + match deadline and holds a worker. +- **Re-run the ladder** with the new matcher defaults before trusting any + number. - ~~Graceful shutdown~~, ~~silent-drop consumer retries~~ and ~~match-failed vs publish-failed~~ are addressed by the shared consumer runner; see [at-least-once-consumers.md](at-least-once-consumers.md). - Still open: `ride.unmatched` for rides with no driver, dedupe in - assignment-service, and rebalance-safe offset tracking. + `ride.unmatched` for rides with no driver, dedupe in assignment-service + and epoch-stamped offset tracking have since been added. Still open: + detection of a re-assignment above the last tracked offset, and a + persistent assignment store. - **Grafana dashboards**. Prometheus's own UI is fine for reading metrics, but a permanent dashboard would make comparing benchmark runs faster. From 8cb12493970a4dd0376a000cd105d5d7ae7d1042 Mon Sep 17 00:00:00 2001 From: coderconnoisseur Date: Fri, 16 Oct 2026 00:35:07 +0530 Subject: [PATCH 7/9] docs: rider-simulator env fallback is not fatal --- README.md | 2 +- docs/rps-benchmarking-setup.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 5ddb177..9138985 100644 --- a/README.md +++ b/README.md @@ -246,7 +246,7 @@ Prometheus scrape targets: Confirm all four are `up` at http://localhost:9090/targets. -**Configuration.** Every service reads its endpoints from environment variables; unset or blank values fall back to the defaults below, so the commands above work unchanged on one host. `KAFKA_BROKERS` is a comma-separated list. An invalid value makes the service exit at startup. +**Configuration.** Every service reads its endpoints from environment variables; unset or blank values fall back to the defaults below, so the commands above work unchanged on one host. `KAFKA_BROKERS` is a comma-separated list. Invalid values make the services exit at startup, except rider-simulator, which falls back to the default for a non-numeric or non-positive `RIDER_RPS` / `RIDER_CONCURRENCY`; double-check these when running the ladder. | Variable | Default | Used by | Meaning | |---|---|---|---| diff --git a/docs/rps-benchmarking-setup.md b/docs/rps-benchmarking-setup.md index e783140..a6aad7e 100644 --- a/docs/rps-benchmarking-setup.md +++ b/docs/rps-benchmarking-setup.md @@ -298,7 +298,7 @@ already reflects any shortfall. #### Environment variables (all services) Every service now reads its endpoints from the environment; unset or blank -values use the defaults. Invalid values make the service exit at startup. +values use the defaults. Invalid values make the services exit at startup, except rider-simulator, which falls back to the default for a non-numeric or non-positive `RIDER_RPS` / `RIDER_CONCURRENCY`; double-check these when running the ladder. | Variable | Default | Used by | Meaning | |---|---|---|---| From dba4be393e86745e452e24f69032657426c86062 Mon Sep 17 00:00:00 2001 From: coderconnoisseur Date: Fri, 16 Oct 2026 00:44:57 +0530 Subject: [PATCH 8/9] assignment: separate consumer group for ride.unmatched; test metric gating --- README.md | 2 +- docs/at-least-once-consumers.md | 19 ++++-- docs/rps-benchmarking-setup.md | 3 +- services/assignment-service/go.mod | 1 + .../internal/app/app_test.go | 65 +++++++++++++++++++ .../internal/kafka/consumer.go | 15 +++-- 6 files changed, 91 insertions(+), 14 deletions(-) create mode 100644 services/assignment-service/internal/app/app_test.go diff --git a/README.md b/README.md index 9138985..8c96624 100644 --- a/README.md +++ b/README.md @@ -175,7 +175,7 @@ See [rps-benchmarking-setup.md](docs/rps-benchmarking-setup.md) for the runbook, - **Matcher pool starvation: new defaults, still unmeasured.** The pool now defaults to `runtime.NumCPU() * 25` workers (`MATCH_WORKERS`) and the match deadline to 500 ms (`MATCH_DEADLINE`), down from `NumCPU * 5` and 800 ms. Slow-fail matches still hold a worker until the deadline, and no ladder has been run with these values, so they are a starting point, not a result. Next: re-run the ladder, and fail fast when the first batch is fully contended instead of waiting out the deadline. - **Single-partition Kafka topics.** One partition per topic caps consumer parallelism per service. Next: create topics with more partitions, partition by ride ID and run multiple consumers per stage. Until then the rewind handling below is not exercised. - **Rewind detection is partial.** The offset tracker detects a rewind only when a re-fetched offset is at or below the last offset it tracked. A partition re-assigned starting *above* that offset is not detected, so the committed offset can move backwards and messages are redelivered. That stays within at-least-once, and handlers absorb duplicates. -- **Two readers, one consumer group.** assignment-service reads `ride.matched` and `ride.unmatched` with two readers in the same group (`assignment-service`). Partitions are assigned per topic correctly, but a join or leave by either reader triggers a rebalance of both. +- **Two readers, two consumer groups.** assignment-service reads `ride.matched` in group `assignment-service` and `ride.unmatched` in group `assignment-service-unmatched`. A shared group would make the leader's metadata request for both topics fail as a whole while `ride.unmatched` does not exist yet (it is created lazily on the first business failure), stalling the success path too; with separate groups a missing `ride.unmatched` topic only delays the unmatched reader, and rebalances are independent. - **In-memory assignment store.** The assignment service keeps ride outcomes (matched and unmatched) in maps for benchmark verification; nothing is persisted, so after a restart a redelivered event is counted as new. - **Test coverage is unit-level.** Unit tests cover the tier-preserving shuffle, the at-least-once consumer runner (ordered commits, retries, DLQ, drain, rewind epochs, panic recovery) and the Redis Lua scripts (locking, assignment, release) against an in-process Redis (miniredis). Nothing yet exercises real Kafka/Redis end to end: the unmatched path, the epoch-stamped commits and the new matcher defaults have not been run against a live cluster. Next: integration tests against containerized Kafka and Redis. diff --git a/docs/at-least-once-consumers.md b/docs/at-least-once-consumers.md index 72a32f4..dc28a3a 100644 --- a/docs/at-least-once-consumers.md +++ b/docs/at-least-once-consumers.md @@ -215,11 +215,16 @@ why the handlers have to tolerate duplicates (section 5). All handler panics, in any service, take the permanent path (3e). -assignment-service runs two runners, one per topic, with two readers in the -same consumer group (`assignment-service`). Each topic's partitions are -assigned correctly, but a reader joining or leaving triggers a rebalance -that affects both. If one runner fails fatally the other is cancelled and -`Run` returns the first error. +assignment-service runs two runners, one per topic, each with its own +consumer group: `assignment-service` for `ride.matched` and +`assignment-service-unmatched` for `ride.unmatched`. With a shared group the +leader's metadata request for both topics fails as a whole while +`ride.unmatched` does not exist yet (it is created lazily on the first +business failure), which would leave the `ride.matched` reader without +partitions too. With two groups a not-yet-existing `ride.unmatched` topic +only delays the unmatched reader, and rebalances are independent. If one +runner fails fatally the other is cancelled and `Run` returns the first +error. **Matching** splits `Match` errors in `app.go`: @@ -293,8 +298,8 @@ rate(match_idempotent_replays_total[1m]) To inspect dead letters, consume the `*.dlq` topic and read the `x-error` and `x-original-*` headers. To check what is committed, describe the -consumer groups `pricing-service`, `matching-service` and -`assignment-service` with the Kafka CLI: lag should return to zero after a +consumer groups `pricing-service`, `matching-service`, +`assignment-service` and `assignment-service-unmatched` with the Kafka CLI: lag should return to zero after a clean shutdown and restart. --- diff --git a/docs/rps-benchmarking-setup.md b/docs/rps-benchmarking-setup.md index a6aad7e..8cf308d 100644 --- a/docs/rps-benchmarking-setup.md +++ b/docs/rps-benchmarking-setup.md @@ -49,7 +49,8 @@ correctness during a benchmark: with the unmatched path, below), or 404. > **Update.** assignment-service now also consumes `ride.unmatched` (second -> reader, same consumer group), records each ride's outcome once and ignores +> reader, own consumer group `assignment-service-unmatched`, so a not-yet-created +> topic only delays that reader), records each ride's outcome once and ignores > duplicates. `assignments_total` and `end_to_end_latency_seconds` count each > matched ride once; redeliveries go to `assignment_duplicates_total`. > Unmatched rides are counted in `unmatched_total{reason}` and diff --git a/services/assignment-service/go.mod b/services/assignment-service/go.mod index fc7ef4c..30b9f3a 100644 --- a/services/assignment-service/go.mod +++ b/services/assignment-service/go.mod @@ -12,6 +12,7 @@ require ( github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/klauspost/compress v1.18.0 // indirect github.com/kr/text v0.2.0 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/pierrec/lz4/v4 v4.1.15 // indirect github.com/prometheus/client_model v0.6.2 // indirect diff --git a/services/assignment-service/internal/app/app_test.go b/services/assignment-service/internal/app/app_test.go new file mode 100644 index 0000000..9c1c9b9 --- /dev/null +++ b/services/assignment-service/internal/app/app_test.go @@ -0,0 +1,65 @@ +package app + +import ( + "context" + "testing" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + + "ridepulse/services/assignment-service/internal/domain" + "ridepulse/services/assignment-service/internal/metrics" + "ridepulse/services/assignment-service/internal/store" +) + +// counterDelta runs fn and returns how far c moved. Reading before and after +// keeps the test independent of other tests using the package-level metrics. +func counterDelta(c prometheus.Collector, fn func()) float64 { + before := testutil.ToFloat64(c) + fn() + return testutil.ToFloat64(c) - before +} + +func TestHandleMatchedCountsEachRideOnce(t *testing.T) { + h := handleMatched(store.NewMemory()) + event := domain.RideMatchedEvent{RideID: "ride-metric-m", DriverID: "d1", RequestedAt: 1} + + var dups float64 + total := counterDelta(metrics.AssignmentsTotal, func() { + dups = counterDelta(metrics.AssignmentDuplicates, func() { + for i := 0; i < 2; i++ { + if err := h(context.Background(), event); err != nil { + t.Fatalf("handleMatched: %v", err) + } + } + }) + }) + if total != 1 { + t.Errorf("assignments_total moved by %v, want 1", total) + } + if dups != 1 { + t.Errorf("assignment_duplicates_total moved by %v, want 1", dups) + } +} + +func TestHandleUnmatchedCountsEachRideOnce(t *testing.T) { + h := handleUnmatched(store.NewMemory()) + event := domain.RideUnmatchedEvent{RideID: "ride-metric-u", Reason: "no_drivers", RequestedAt: 1, FailedAt: 2} + + var dups float64 + total := counterDelta(metrics.UnmatchedTotal.WithLabelValues("no_drivers"), func() { + dups = counterDelta(metrics.UnmatchedDuplicates, func() { + for i := 0; i < 2; i++ { + if err := h(context.Background(), event); err != nil { + t.Fatalf("handleUnmatched: %v", err) + } + } + }) + }) + if total != 1 { + t.Errorf("unmatched_total moved by %v, want 1", total) + } + if dups != 1 { + t.Errorf("unmatched_duplicates_total moved by %v, want 1", dups) + } +} diff --git a/services/assignment-service/internal/kafka/consumer.go b/services/assignment-service/internal/kafka/consumer.go index 5c53b0a..e7b023b 100644 --- a/services/assignment-service/internal/kafka/consumer.go +++ b/services/assignment-service/internal/kafka/consumer.go @@ -15,7 +15,12 @@ import ( ) const ( - groupID = "assignment-service" + // Each topic gets its own consumer group. With one shared group, the + // group leader's metadata request for both topics fails as a whole while + // ride.unmatched does not exist yet (it is created lazily on the first + // business failure), leaving both readers without partitions. + matchedGroupID = "assignment-service" + unmatchedGroupID = "assignment-service-unmatched" matchedTopic = "ride.matched" matchedDLQTopic = "ride.matched.dlq" @@ -24,14 +29,14 @@ const ( ) // consumer is the topic-agnostic reader + DLQ writer pair shared by the -// typed consumers below. Both topics use the same consumer group. +// typed consumers below. Each topic has its own consumer group. type consumer struct { reader *kafka.Reader dlq *kafka.Writer dlqTopic string } -func newConsumer(brokers []string, topic, dlqTopic string) *consumer { +func newConsumer(brokers []string, groupID, topic, dlqTopic string) *consumer { r := kafka.NewReader(kafka.ReaderConfig{ Brokers: brokers, GroupID: groupID, @@ -96,7 +101,7 @@ func (c *consumer) Close() error { type KafkaConsumer struct{ *consumer } func NewKafkaConsumer(brokers []string) *KafkaConsumer { - return &KafkaConsumer{newConsumer(brokers, matchedTopic, matchedDLQTopic)} + return &KafkaConsumer{newConsumer(brokers, matchedGroupID, matchedTopic, matchedDLQTopic)} } // Run consumes RideMatched events until ctx is cancelled. @@ -111,7 +116,7 @@ func (c *KafkaConsumer) Run( type UnmatchedConsumer struct{ *consumer } func NewUnmatchedConsumer(brokers []string) *UnmatchedConsumer { - return &UnmatchedConsumer{newConsumer(brokers, unmatchedTopic, unmatchedDLQTopic)} + return &UnmatchedConsumer{newConsumer(brokers, unmatchedGroupID, unmatchedTopic, unmatchedDLQTopic)} } // Run consumes RideUnmatched events until ctx is cancelled. From 351dda647a99e6e12d5ecc0d941da0c2c397bdce Mon Sep 17 00:00:00 2001 From: coderconnoisseur Date: Fri, 16 Oct 2026 00:46:26 +0530 Subject: [PATCH 9/9] matching: size redis pool to workers, retry assign errors past deadline --- README.md | 1 + docs/at-least-once-consumers.md | 8 +++++--- docs/driver-state-model.md | 3 ++- docs/rps-benchmarking-setup.md | 1 + .../cmd/matching-service/main.go | 2 +- services/matching-service/internal/app/app.go | 7 ++++++- .../matching-service/internal/app/app_test.go | 1 + .../matching-service/internal/app/config.go | 11 +++++++++++ .../internal/app/config_test.go | 12 +++++++++++- .../internal/matching/matching.go | 17 +++++++++++++---- .../internal/redis/driver_repository.go | 6 ++++-- .../internal/redis/driver_repository_test.go | 2 +- 12 files changed, 57 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index 8c96624..19bcdd3 100644 --- a/README.md +++ b/README.md @@ -255,6 +255,7 @@ Confirm all four are `up` at http://localhost:9090/targets. | `HTTP_ADDR` | `:8080` (api-gateway), `:2114` (assignment) | api-gateway, assignment | HTTP listen address (also serves `/metrics`). | | `METRICS_ADDR` | `:2115` (pricing), `:2112` (matching) | pricing, matching | Metrics listen address. | | `MATCH_WORKERS` | `runtime.NumCPU() * 25` | matching | Matcher worker-pool size; positive integer. | +| `REDIS_POOL_SIZE` | `MATCH_WORKERS * 5` | matching | Redis connection pool size; positive integer. Each worker runs up to 5 parallel `SETNX` attempts, while the go-redis default is only 10 x GOMAXPROCS. | | `MATCH_DEADLINE` | `500ms` | matching | Total per-match budget (Go duration); must exceed the 200 ms batch deadline. | | `DRIVER_COUNT` | `1000` | driver-simulator | Number of simulated drivers. | | `DRIVER_TRIP_DURATION` | `5s` | driver-simulator | How long an assigned driver stays busy (Go duration). | diff --git a/docs/at-least-once-consumers.md b/docs/at-least-once-consumers.md index dc28a3a..7e04a3e 100644 --- a/docs/at-least-once-consumers.md +++ b/docs/at-least-once-consumers.md @@ -236,9 +236,11 @@ error. returned, so the runner retries and eventually dead-letters it like any other infrastructure error. See [driver-state-model.md](driver-state-model.md#4b-the-unmatched-path). -- Any other `Match` error (Redis failures, and any other - `DeadlineExceeded` such as the detached assign timeout) is returned, so the runner - retries it and eventually dead-letters it. Retrying is safe because +- Any other `Match` error is returned. That covers + Redis failures, any `DeadlineExceeded` that is not the outer match deadline, + and every `AssignDriver` failure (wrapped in `matching.ErrAssign`, such as + the detached assign timeout), which is retried even if the outer deadline + has expired by then. The runner retries and eventually dead-letters these. Retrying is safe because `Match` replays an existing assignment instead of matching again (see [driver-state-model.md](driver-state-model.md)). - Publishing `ride.matched` has its own 300 ms budget; a publish failure is diff --git a/docs/driver-state-model.md b/docs/driver-state-model.md index 5b193fb..dadd7a1 100644 --- a/docs/driver-state-model.md +++ b/docs/driver-state-model.md @@ -124,7 +124,8 @@ Redis can hold a match handler for up to that longer socket timeout. If that matters, enable `ContextTimeoutEnabled` on the client. The handler's error classification is what makes that safe: only the outer match deadline (`matchCtx.Err() != nil`) is treated as a business failure; any other -`DeadlineExceeded` is retried, and the retry replays through `ride:driver`. +`DeadlineExceeded`, and any assign error (`matching.ErrAssign`, even if the +outer deadline has expired), is retried, and the retry replays through `ride:driver`. ### Error handling around it diff --git a/docs/rps-benchmarking-setup.md b/docs/rps-benchmarking-setup.md index 8cf308d..3d8a4e0 100644 --- a/docs/rps-benchmarking-setup.md +++ b/docs/rps-benchmarking-setup.md @@ -308,6 +308,7 @@ values use the defaults. Invalid values make the services exit at startup, excep | `HTTP_ADDR` | `:8080` (api-gateway), `:2114` (assignment) | api-gateway, assignment | HTTP listen address (also serves `/metrics`). | | `METRICS_ADDR` | `:2115` (pricing), `:2112` (matching) | pricing, matching | Metrics listen address. | | `MATCH_WORKERS` | `runtime.NumCPU() * 25` | matching | Matcher worker-pool size; positive integer. | +| `REDIS_POOL_SIZE` | `MATCH_WORKERS * 5` | matching | Redis connection pool size; positive integer. Each worker runs up to 5 parallel `SETNX` attempts, while the go-redis default is only 10 x GOMAXPROCS. | | `MATCH_DEADLINE` | `500ms` | matching | Total per-match budget (Go duration); must exceed the 200 ms batch deadline. | | `RIDER_URL` | `http://localhost:8080/rides` | rider-simulator | Gateway endpoint. | | `RIDER_RPS` | `500` | rider-simulator | Target requests per second. | diff --git a/services/matching-service/cmd/matching-service/main.go b/services/matching-service/cmd/matching-service/main.go index 71479a3..ae7130b 100644 --- a/services/matching-service/cmd/matching-service/main.go +++ b/services/matching-service/cmd/matching-service/main.go @@ -18,7 +18,7 @@ func main() { if err != nil { log.Fatalf("matching-service: invalid configuration: %v", err) } - log.Printf("matching-service config: metrics=%s brokers=%v redis=%s workers=%d deadline=%s", cfg.MetricsAddr, cfg.KafkaBrokers, cfg.RedisAddr, cfg.MatchWorkers, cfg.MatchDeadline) + log.Printf("matching-service config: metrics=%s brokers=%v redis=%s workers=%d redis_pool=%d deadline=%s", cfg.MetricsAddr, cfg.KafkaBrokers, cfg.RedisAddr, cfg.MatchWorkers, cfg.RedisPoolSize, cfg.MatchDeadline) if err := app.Run(ctx, cfg); err != nil { stop() diff --git a/services/matching-service/internal/app/app.go b/services/matching-service/internal/app/app.go index dbcb217..7d4058e 100644 --- a/services/matching-service/internal/app/app.go +++ b/services/matching-service/internal/app/app.go @@ -60,7 +60,7 @@ func Run(ctx context.Context, cfg Config) error { }() // --- Dependencies --- - driverRepo := redis.NewDriverRepository(cfg.RedisAddr) + driverRepo := redis.NewDriverRepository(cfg.RedisAddr, cfg.RedisPoolSize) matcher := matching.New(driverRepo) consumer := kafka.NewKafkaConsumer(cfg.KafkaBrokers, cfg.MatchWorkers) @@ -91,7 +91,12 @@ func Run(ctx context.Context, cfg Config) error { // DeadlineExceeded counts only when the outer match deadline expired // (matchCtxErr != nil); any other deadline, such as the detached assign // timeout or a pool wait, is retried, which is safe via ride:driver replay. +// An AssignDriver failure (matching.ErrAssign) is always retried, even if the +// outer deadline has since expired. func isBusinessFailure(err, matchCtxErr error) bool { + if errors.Is(err, matching.ErrAssign) { + return false + } return errors.Is(err, matching.ErrNoDriversAvailable) || (errors.Is(err, context.DeadlineExceeded) && matchCtxErr != nil) } diff --git a/services/matching-service/internal/app/app_test.go b/services/matching-service/internal/app/app_test.go index 103619c..bf2d16f 100644 --- a/services/matching-service/internal/app/app_test.go +++ b/services/matching-service/internal/app/app_test.go @@ -24,6 +24,7 @@ func TestIsBusinessFailure(t *testing.T) { {"inner deadline, match ctx alive", context.DeadlineExceeded, nil, false}, {"redis error", errors.New("dial tcp: connection refused"), nil, false}, {"canceled", context.Canceled, nil, false}, + {"assign timeout after outer deadline", fmt.Errorf("%w: %w", matching.ErrAssign, context.DeadlineExceeded), context.DeadlineExceeded, false}, } for _, tc := range cases { if got := isBusinessFailure(tc.err, tc.ctx); got != tc.want { diff --git a/services/matching-service/internal/app/config.go b/services/matching-service/internal/app/config.go index 08d4556..a95f370 100644 --- a/services/matching-service/internal/app/config.go +++ b/services/matching-service/internal/app/config.go @@ -16,6 +16,7 @@ type Config struct { KafkaBrokers []string RedisAddr string MatchWorkers int + RedisPoolSize int MatchDeadline time.Duration } @@ -34,6 +35,15 @@ func LoadConfig(getenv func(string) string) (Config, error) { } } + // Each worker runs up to BatchSize parallel SETNX attempts. + poolSize := workers * matching.BatchSize + if raw := strings.TrimSpace(getenv("REDIS_POOL_SIZE")); raw != "" { + poolSize, err = strconv.Atoi(raw) + if err != nil || poolSize <= 0 { + return Config{}, fmt.Errorf("REDIS_POOL_SIZE %q must be a positive integer", raw) + } + } + deadline := defaultMatchDeadline if raw := strings.TrimSpace(getenv("MATCH_DEADLINE")); raw != "" { deadline, err = time.ParseDuration(raw) @@ -50,6 +60,7 @@ func LoadConfig(getenv func(string) string) (Config, error) { KafkaBrokers: brokers, RedisAddr: envOr(getenv, "REDIS_ADDR", "localhost:6379"), MatchWorkers: workers, + RedisPoolSize: poolSize, MatchDeadline: deadline, }, nil } diff --git a/services/matching-service/internal/app/config_test.go b/services/matching-service/internal/app/config_test.go index d90bdfd..a701d07 100644 --- a/services/matching-service/internal/app/config_test.go +++ b/services/matching-service/internal/app/config_test.go @@ -5,6 +5,8 @@ import ( "runtime" "testing" "time" + + "ridepulse/services/matching-service/internal/matching" ) func env(m map[string]string) func(string) string { @@ -17,6 +19,7 @@ func TestLoadConfig(t *testing.T) { KafkaBrokers: []string{"localhost:9092"}, RedisAddr: "localhost:6379", MatchWorkers: runtime.NumCPU() * 25, + RedisPoolSize: runtime.NumCPU() * 25 * matching.BatchSize, MatchDeadline: 500 * time.Millisecond, } tests := []struct { @@ -31,8 +34,15 @@ func TestLoadConfig(t *testing.T) { "MATCH_WORKERS": "7", "MATCH_DEADLINE": "1s", }, Config{ MetricsAddr: ":9", KafkaBrokers: []string{"a:1", "b:2"}, RedisAddr: "r:6", - MatchWorkers: 7, MatchDeadline: time.Second, + MatchWorkers: 7, RedisPoolSize: 7 * matching.BatchSize, MatchDeadline: time.Second, }, false}, + {"pool size override", map[string]string{"REDIS_POOL_SIZE": "42"}, + func() Config { c := defaults; c.RedisPoolSize = 42; return c }(), false}, + {"pool size follows workers", map[string]string{"MATCH_WORKERS": "3"}, + func() Config { c := defaults; c.MatchWorkers = 3; c.RedisPoolSize = 3 * matching.BatchSize; return c }(), false}, + {"pool size zero", map[string]string{"REDIS_POOL_SIZE": "0"}, Config{}, true}, + {"pool size negative", map[string]string{"REDIS_POOL_SIZE": "-1"}, Config{}, true}, + {"pool size not a number", map[string]string{"REDIS_POOL_SIZE": "big"}, Config{}, true}, {"empty broker entry", map[string]string{"KAFKA_BROKERS": "a:1,,b:2"}, Config{}, true}, {"workers zero", map[string]string{"MATCH_WORKERS": "0"}, Config{}, true}, {"workers negative", map[string]string{"MATCH_WORKERS": "-3"}, Config{}, true}, diff --git a/services/matching-service/internal/matching/matching.go b/services/matching-service/internal/matching/matching.go index 4e9c3a3..e545987 100644 --- a/services/matching-service/internal/matching/matching.go +++ b/services/matching-service/internal/matching/matching.go @@ -3,6 +3,7 @@ package matching import ( "context" "errors" + "fmt" "math/rand" "sync" "time" @@ -21,8 +22,11 @@ import ( // outer deadline). const BatchDeadline = 200 * time.Millisecond +// BatchSize is the number of drivers locked in parallel per batch. It is +// exported so the Redis pool can be sized to workers * BatchSize. +const BatchSize = 5 + const ( - batchSize = 5 batchDeadline = BatchDeadline releaseTimeout = 200 * time.Millisecond assignTimeout = 200 * time.Millisecond @@ -39,6 +43,11 @@ const ( // is a business outcome, not an infrastructure failure, so it is not retried. var ErrNoDriversAvailable = errors.New("no drivers available") +// ErrAssign wraps any AssignDriver failure other than ports.ErrLockLost. It is +// an infrastructure error: callers retry it even if the match deadline has +// expired (a retry replays via ride:driver). +var ErrAssign = errors.New("assign driver") + type Service struct { repo ports.DriverRepository } @@ -127,7 +136,7 @@ func (s *Service) Match(ctx context.Context, event domain.RidePricedEvent) (doma // per tier instead of being pinned to a fixed 5. shuffleWithinTiers(drivers, driverTierSize) - for i := 0; i < len(drivers); i += batchSize { + for i := 0; i < len(drivers); i += BatchSize { select { case <-ctx.Done(): @@ -136,7 +145,7 @@ func (s *Service) Match(ctx context.Context, event domain.RidePricedEvent) (doma default: } - end := i + batchSize + end := i + BatchSize if end > len(drivers) { end = len(drivers) } @@ -159,7 +168,7 @@ func (s *Service) Match(ctx context.Context, event domain.RidePricedEvent) (doma if err != nil { s.releaseLock(winnerID, event.RideID) metrics.MatchFailures.Inc() - return domain.RideMatchedEvent{}, err + return domain.RideMatchedEvent{}, fmt.Errorf("%w: %w", ErrAssign, err) } metrics.MatchSuccess.Inc() return matchedEvent(event, assignedID), nil diff --git a/services/matching-service/internal/redis/driver_repository.go b/services/matching-service/internal/redis/driver_repository.go index a4254a9..3e9ed3f 100644 --- a/services/matching-service/internal/redis/driver_repository.go +++ b/services/matching-service/internal/redis/driver_repository.go @@ -15,9 +15,11 @@ type DriverRepository struct { client *redis.Client } -func NewDriverRepository(addr string) *DriverRepository { +// NewDriverRepository connects to addr with a pool of poolSize connections. +func NewDriverRepository(addr string, poolSize int) *DriverRepository { rdb := redis.NewClient(&redis.Options{ - Addr: addr, + Addr: addr, + PoolSize: poolSize, }) return &DriverRepository{client: rdb} } diff --git a/services/matching-service/internal/redis/driver_repository_test.go b/services/matching-service/internal/redis/driver_repository_test.go index 6d4b800..2fa193c 100644 --- a/services/matching-service/internal/redis/driver_repository_test.go +++ b/services/matching-service/internal/redis/driver_repository_test.go @@ -15,7 +15,7 @@ import ( func newTestRepo(t *testing.T) (*DriverRepository, *miniredis.Miniredis) { t.Helper() mr := miniredis.RunT(t) - repo := NewDriverRepository(mr.Addr()) + repo := NewDriverRepository(mr.Addr(), 10) t.Cleanup(func() { _ = repo.Close() }) return repo, mr }