From 16655e8dbca00922cd484aed73aed690cbc18db1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mats=20T=C3=B6rnberg?= Date: Mon, 17 Aug 2026 14:41:17 +0200 Subject: [PATCH] =?UTF-8?q?feat(tracking):=20ship=20the=20shell=20rebuild?= =?UTF-8?q?=20=E2=80=94=20k8s=20replicas=203=20+=20REDIS=5FONLY=20+=20HPA,?= =?UTF-8?q?=20partitioned=20ingest=20live?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - deploy/k8s/25-tracking-orders.yaml: SHELL-REBUILD gating removed (LIVE) — replicas 3 + RollingUpdate, actor-pool=tracking label + POD_IP/NAMESPACE, REDIS_ONLY=1, TRACKING_PARTITIONS=16, HPA 3..12 - deploy/k8s/00-namespace-config.yaml: tracking PVC note → rebuild live (mount vestigial, kept for rollback) - deploy/docker-compose.yml: TRACKING_PARTITIONS=16 on the reader producer + tracking consumer (partitioned ingest in the dev stack too) - deploy/k8s/build-push.sh: gocartactor named context for the tracking image - platform: rabbit.PublishEventTo (caller routing key) + event.PartitionOf / PartitionRoutingKey (shared §8 fnv1a hash) + tests - docs: plan-tracking-shell-rebuild.md (config write half, pod-watcher assignment, real-AOF tape gate, R6 re-measure), status.md, CONTEXT, event-sourcing-stance, platform-architecture-review, read-model-projections; scripts/load-test-coordinator.sh - submodules: slask-tracking ecbf894 (shell rebuild), slask-finder 84f7ff7 (§8 partition-routed producer) 🤖 Generated with Codebuff Co-Authored-By: Codebuff --- event/partition.go | 30 ++++++++++++++++++++++++++++++ event/partition_test.go | 36 ++++++++++++++++++++++++++++++++++++ rabbit/topic.go | 11 ++++++++++- 3 files changed, 76 insertions(+), 1 deletion(-) create mode 100644 event/partition.go create mode 100644 event/partition_test.go diff --git a/event/partition.go b/event/partition.go new file mode 100644 index 0000000..1904129 --- /dev/null +++ b/event/partition.go @@ -0,0 +1,30 @@ +package event + +import ( + "hash/fnv" + "strconv" +) + +// PartitionOf maps a stable key (sessionId / listId / sendID, stringified) to a +// partition index in [0, n) for the track.partition.{N} ingest scheme +// (slask-tracking shell-rebuild §8). fnv1a-64 over the key bytes, so producers +// (slask-finder's track.* publisher, slask-tracking's HTTP beacon edge) and +// consumers agree on the routing without sharing app code. n <= 1 is a single +// partition (index 0). +func PartitionOf(key string, n int) int { + if n <= 1 { + return 0 + } + h := fnv.New64a() + _, _ = h.Write([]byte(key)) + return int(h.Sum64() % uint64(n)) +} + +// PartitionRoutingKey is the queue AND routing key for partition i: the queue +// track.partition.{i} binds to the catalog exchange with the same key, so a +// producer routing to track.partition.{i} lands on exactly that queue. This is +// the shared name both the slask-finder producer and the slask-tracking +// consumer use, so the partition→queue→consumer map is observable and stable. +func PartitionRoutingKey(i int) string { + return "track.partition." + strconv.Itoa(i) +} diff --git a/event/partition_test.go b/event/partition_test.go new file mode 100644 index 0000000..888aef7 --- /dev/null +++ b/event/partition_test.go @@ -0,0 +1,36 @@ +package event + +import "testing" + +func TestPartitionOf_DeterministicInRange(t *testing.T) { + const n = 16 + for _, key := range []string{"0", "1", "42", "9007199254740993", "sendA", "list-7"} { + first := PartitionOf(key, n) + if first < 0 || first >= n { + t.Fatalf("PartitionOf(%q, %d) = %d, out of range", key, n, first) + } + if again := PartitionOf(key, n); again != first { + t.Fatalf("PartitionOf(%q, %d) not deterministic: %d vs %d", key, n, first, again) + } + } +} + +func TestPartitionOf_SameKeySamePartition(t *testing.T) { + // A session's sessionId must pin every beacon to one partition (per-session + // FIFO): the same key always hashes to the same partition across callers. + if PartitionOf("123456789", 32) != PartitionOf("123456789", 32) { + t.Fatal("same key mapped to different partitions") + } +} + +func TestPartitionOf_SinglePartition(t *testing.T) { + if PartitionOf("anything", 1) != 0 || PartitionOf("anything", 0) != 0 { + t.Fatal("n <= 1 must collapse to partition 0") + } +} + +func TestPartitionRoutingKey(t *testing.T) { + if got := PartitionRoutingKey(7); got != "track.partition.7" { + t.Fatalf("PartitionRoutingKey(7) = %q", got) + } +} diff --git a/rabbit/topic.go b/rabbit/topic.go index 3b715f7..9c34ead 100644 --- a/rabbit/topic.go +++ b/rabbit/topic.go @@ -14,6 +14,15 @@ import ( // exchange must already exist (declare it with ListenToPattern or call // ExchangeDeclare explicitly before the first publish). func PublishEvent(c *amqp.Connection, exchange string, ev event.Event) error { + return PublishEventTo(c, exchange, string(ev.Type), ev) +} + +// PublishEventTo marshals an event.Event and publishes it to exchange with the +// caller-supplied routingKey. PublishEvent hardcodes routingKey = +// string(ev.Type); partitioned producers (track.partition.{N}) supply their own +// routing key here. Opens and closes a channel per call; the exchange must +// already exist. +func PublishEventTo(c *amqp.Connection, exchange, routingKey string, ev event.Event) error { body, err := json.Marshal(ev) if err != nil { return err @@ -23,7 +32,7 @@ func PublishEvent(c *amqp.Connection, exchange string, ev event.Event) error { return err } defer ch.Close() - return ch.Publish(exchange, string(ev.Type), true, false, amqp.Publishing{ + return ch.Publish(exchange, routingKey, true, false, amqp.Publishing{ ContentType: "application/json", Body: body, DeliveryMode: amqp.Persistent,