From 1e3386fac80f0464378147d075b639cf3b34c30f Mon Sep 17 00:00:00 2001 From: Suriya Date: Mon, 3 Aug 2026 19:44:47 +0530 Subject: [PATCH] Elect a single MQTT consumer among replicas MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deployed as a StatefulSet with three replicas, and MQTT has no queue groups — every subscriber receives every message. All three pods would commit the same bill and publish three acks. Nothing double-counts, because the ingest deduplicates on the till's UUID and holds an advisory lock, but it is three times the database work and three times the traffic for one sale. Ordinal 0 consumes; the others stay idle. A StatefulSet already guarantees stable unique ordinals, so this is a deterministic election with no lock, no lease and no new dependency. If that pod dies the set recreates it and tills hold their bills meanwhile, which is what they are built to do. POS_MQTT_CONSUMER=always/never overrides it for deployments that are not a StatefulSet. Anything without an ordinal name — a Deployment pod, a bare container, local development — consumes, because a lone instance that silently refused to would be a far more confusing failure than one that did. The client id now defaults to the pod name rather than a constant. Two connections sharing an id evict each other in a reconnect loop that looks exactly like a flapping network, and takes a while to recognise as anything else. Co-Authored-By: Claude Opus 5 --- messaging/posmqtt.go | 75 +++++++++++++++++++++++++++++++-------- messaging/posmqtt_test.go | 41 +++++++++++++++++++++ 2 files changed, 102 insertions(+), 14 deletions(-) diff --git a/messaging/posmqtt.go b/messaging/posmqtt.go index 9957eb5..02002cf 100644 --- a/messaging/posmqtt.go +++ b/messaging/posmqtt.go @@ -16,22 +16,17 @@ import ( mqtt "github.com/eclipse/paho.mqtt.golang" ) -// Plain-MQTT ingest for the Nearle POS terminals. +// MQTT ingest for the Nearle POS terminals. // -// The sibling of posconsumer.go, and which one you want depends entirely on -// what is listening on the other end: +// The broker is Eclipse Mosquitto, shared with the rider fleet. An audit of the +// estate found no reachable NATS and no MQTT gateway on the NATS boxes that do +// exist, so a NATS consumer that briefly lived here was deleted rather than +// left to rot — a client for a protocol nothing speaks is worse than none. // -// - **This file** talks MQTT to a broker like Mosquitto or EMQX — the kind -// already running at the rider app's `66.116.225.226:1883`. -// - **posconsumer.go** talks the NATS protocol to a NATS server, which -// exposes MQTT through a gateway but speaks NATS itself on 4222. +// Enabled with MQTT_URL. Unset, the terminals reach the same service over HTTP +// instead, and this file does nothing. // -// They are not interchangeable: a NATS client cannot connect to Mosquitto, and -// an MQTT client cannot use NATS' native subjects. Both call the same -// PosService, so whichever is running, a bill lands identically. -// -// Enabled with MQTT_URL. Both may run at once, which is what a migration -// between brokers looks like. +// Only one replica consumes: see posConsumerElected. const ( // Namespaced under `nearle/` alongside the rider app's @@ -60,13 +55,34 @@ func StartPosMqttConsumer(svc services.PosService) (*PosMqttConsumer, error) { return nil, nil } + // Only one replica consumes. + // + // MQTT has no queue groups — every subscriber receives every message, so + // three replicas would each commit the same bill and publish three acks. + // The ingest is idempotent, so nothing double-counts, but it is three times + // the database work and three times the traffic for one sale. + // + // A StatefulSet gives pods stable ordinal names, so ordinal 0 is a + // deterministic election with no coordination and no extra dependency. If + // that pod dies the set recreates it; tills hold their bills and re-send in + // the meantime, which is exactly what they are built to do. + if !posConsumerElected() { + log.Printf("pos: replica %q is not the elected consumer, MQTT ingest idle here", + os.Getenv("HOSTNAME")) + return nil, nil + } + c := &PosMqttConsumer{svc: svc} opts := mqtt.NewClientOptions(). AddBroker(url). // Stable, so the broker resumes this session and redelivers anything // in flight rather than treating every restart as a new subscriber. - SetClientID(getEnvDefault("MQTT_CLIENT_ID", "nearle-pos-ingest")). + // Defaults to the pod name so replicas can never collide: a second + // connection with the same client id evicts the first, and the two then + // fight in a reconnect loop that looks like a flapping network. + SetClientID(getEnvDefault("MQTT_CLIENT_ID", + getEnvDefault("HOSTNAME", "nearle-pos-ingest"))). SetCleanSession(false). SetAutoReconnect(true). SetMaxReconnectInterval(30 * time.Second). @@ -261,6 +277,37 @@ func (c *PosMqttConsumer) Close() { c.client.Disconnect(uint(quiesce)) } +// posConsumerElected decides whether this replica runs the MQTT ingest. +// +// Rules, in order: +// +// - POS_MQTT_CONSUMER=always or =never settles it outright, for deployments +// that are not a StatefulSet or that want the consumer somewhere specific. +// - A StatefulSet pod name ending in `-0` is elected. Ordinals are stable and +// unique, so this needs no lock, no lease and no coordination. +// - Anything else — a bare container, a Deployment, local development — +// is elected, because a single instance that refused to consume would be a +// far more confusing failure than one that did. +func posConsumerElected() bool { + switch strings.ToLower(strings.TrimSpace(os.Getenv("POS_MQTT_CONSUMER"))) { + case "always", "true", "yes": + return true + case "never", "false", "no": + return false + } + + host := strings.TrimSpace(os.Getenv("HOSTNAME")) + if i := strings.LastIndex(host, "-"); i >= 0 { + if ordinal := host[i+1:]; ordinal != "" && strings.Trim(ordinal, "0123456789") == "" { + // A StatefulSet ordinal. Only the first replica consumes. + return ordinal == "0" + } + } + + // Not an ordinal-named pod, so there is nothing to elect against. + return true +} + func getEnvDefault(key, fallback string) string { if v := strings.TrimSpace(os.Getenv(key)); v != "" { return v diff --git a/messaging/posmqtt_test.go b/messaging/posmqtt_test.go index f3a40ea..1aaaaa2 100644 --- a/messaging/posmqtt_test.go +++ b/messaging/posmqtt_test.go @@ -380,3 +380,44 @@ func TestTopicIdentityRejectsShortTopics(t *testing.T) { t.Errorf("topicIdentity = %q/%q, want 12/T4A9", store, terminal) } } + +// MQTT has no queue groups: every subscriber gets every message. Three replicas +// all consuming would commit the same bill three times and publish three acks — +// harmless, because the ingest is idempotent, but three times the work. +func TestOnlyTheFirstStatefulSetReplicaConsumes(t *testing.T) { + cases := []struct { + name string + hostname string + override string + want bool + }{ + {"statefulset ordinal 0", "fiesta-0", "", true}, + {"statefulset ordinal 1", "fiesta-1", "", false}, + {"statefulset ordinal 2", "fiesta-2", "", false}, + {"double-digit ordinal", "fiesta-10", "", false}, + + // A Deployment pod has a random suffix, not an ordinal. Refusing to + // consume there would be a far more confusing failure than consuming. + {"deployment pod", "fiesta-7d4f9c8b6d-x2k9p", "", true}, + {"bare container", "a1b2c3d4e5f6", "", true}, + {"no hostname", "", "", true}, + + // The override settles it outright either way. + {"forced on", "fiesta-2", "always", true}, + {"forced off", "fiesta-0", "never", false}, + {"forced on via true", "fiesta-5", "true", true}, + {"forced off via false", "fiesta-0", "false", false}, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + t.Setenv("HOSTNAME", c.hostname) + t.Setenv("POS_MQTT_CONSUMER", c.override) + + if got := posConsumerElected(); got != c.want { + t.Errorf("posConsumerElected() = %v, want %v (hostname %q, override %q)", + got, c.want, c.hostname, c.override) + } + }) + } +}