diff --git a/messaging/posmqtt.go b/messaging/posmqtt.go index 02002cf..31c2e46 100644 --- a/messaging/posmqtt.go +++ b/messaging/posmqtt.go @@ -42,6 +42,12 @@ const ( type PosMqttConsumer struct { client mqtt.Client svc services.PosService + + // Bills and registrations share a pool; heartbeats get their own, so a + // backlog of sales cannot make every till look dark at the moment the + // system is busiest. + ingest *posPool + health *posPool } // StartPosMqttConsumer connects and subscribes. @@ -72,7 +78,15 @@ func StartPosMqttConsumer(svc services.PosService) (*PosMqttConsumer, error) { return nil, nil } - c := &PosMqttConsumer{svc: svc} + // Each ingest worker holds a database transaction while it runs, so the + // real ceiling is the Postgres connection pool rather than the CPU. The + // queue is deep enough to absorb a burst and shallow enough that a genuine + // overload is felt as backpressure rather than hidden as latency. + c := &PosMqttConsumer{ + svc: svc, + ingest: newPosPool("ingest", posPoolSize("POS_INGEST_WORKERS", 8), 256), + health: newPosPool("health", posPoolSize("POS_HEALTH_WORKERS", 2), 512), + } opts := mqtt.NewClientOptions(). AddBroker(url). @@ -84,6 +98,10 @@ func StartPosMqttConsumer(svc services.PosService) (*PosMqttConsumer, error) { SetClientID(getEnvDefault("MQTT_CLIENT_ID", getEnvDefault("HOSTNAME", "nearle-pos-ingest"))). SetCleanSession(false). + // Ordered delivery keeps paho on one goroutine, which is what lets a + // full queue push back on the broker. With concurrent delivery paho + // would keep reading no matter how far behind the workers were. + SetOrderMatters(posOrderedDelivery). SetAutoReconnect(true). SetMaxReconnectInterval(30 * time.Second). SetKeepAlive(30 * time.Second). @@ -101,9 +119,9 @@ func StartPosMqttConsumer(svc services.PosService) (*PosMqttConsumer, error) { opts.SetOnConnectHandler(func(client mqtt.Client) { log.Printf("pos: connected to MQTT broker %s", url) for topic, handler := range map[string]mqtt.MessageHandler{ - topicOrders: c.handleOrders, - topicCustomers: c.handleCustomers, - topicHealth: c.handleHealth, + topicOrders: wrapHandler(c.ingest, c.handleOrders), + topicCustomers: wrapHandler(c.ingest, c.handleCustomers), + topicHealth: wrapHandler(c.health, c.handleHealth), } { if token := client.Subscribe(topic, 1, handler); token.Wait() && token.Error() != nil { log.Printf("pos: could not subscribe to %s: %v", topic, token.Error()) @@ -270,6 +288,12 @@ func (c *PosMqttConsumer) Close() { if c == nil || c.client == nil { return } + // Workers drain before the connection closes, so a bill mid-commit still + // gets its ack out. Disconnecting first would strand it: committed here, + // unacknowledged there, and sent again on the till's next attempt. + c.ingest.stop() + c.health.stop() + quiesce, err := strconv.Atoi(getEnvDefault("MQTT_QUIESCE_MS", "2000")) if err != nil || quiesce < 0 { quiesce = 2000 diff --git a/messaging/posworkers.go b/messaging/posworkers.go new file mode 100644 index 0000000..ce75d50 --- /dev/null +++ b/messaging/posworkers.go @@ -0,0 +1,174 @@ +package messaging + +import ( + "log" + "os" + "strconv" + "sync" + + mqtt "github.com/eclipse/paho.mqtt.golang" +) + +// Concurrency for the MQTT ingest. +// +// paho delivers messages on a single goroutine, so without this every bill is +// committed one after another. A bill is a full Postgres transaction — advisory +// lock, dedup check, stock row locks, availability check, four inserts, commit — +// which realistically costs 10–30ms. Serially that is 30–100 bills a second, +// and a shop-wide backlog draining after an outage would take minutes to land. +// +// ### Why a bounded pool rather than a goroutine per message +// +// paho can be told to call handlers concurrently, but it spawns without limit. +// A storm would then open a database transaction per message, exhaust the +// connection pool, and stall every one of them at once — turning a slow minute +// into a dead one. +// +// A fixed pool behind a bounded queue does the opposite. When the queue fills, +// submitting **blocks**, which is the point: paho stops acknowledging, the +// broker's in-flight window fills, and it stops sending. Backpressure travels +// all the way back to the till, which holds its bills and retries. Slow, but +// nothing is dropped and nothing is lost. +// +// ### Why bills and heartbeats have separate pools +// +// A heartbeat is one Redis write and a bill is a transaction. Sharing a queue +// would let a backlog of bills delay presence, and every till would appear to +// go dark at exactly the moment the system was busiest — the worst possible +// time to be blind to which counters are alive. + +// posPool is a fixed set of workers reading a bounded queue. +type posPool struct { + name string + jobs chan func() + wg sync.WaitGroup + once sync.Once + + // Guards the transition to closed. A plain `select` over a done-channel and + // the job channel is not enough: once both are ready Go picks between them + // at random, and picking the send panics on a closed channel. Held for + // reading across the whole of submit, so stop cannot close the queue out + // from under a send already in progress. + mu sync.RWMutex + closed bool +} + +func newPosPool(name string, workers, queue int) *posPool { + p := &posPool{ + name: name, + jobs: make(chan func(), queue), + } + + p.wg.Add(workers) + for i := 0; i < workers; i++ { + go func() { + defer p.wg.Done() + for job := range p.jobs { + job() + } + }() + } + + log.Printf("pos: %s pool started with %d workers, queue %d", name, workers, queue) + return p +} + +// submit queues work, blocking when the queue is full. +// +// Blocking is deliberate. Dropping would lose a bill outright; the terminal +// would eventually re-send it, but only after its ack timeout, and meanwhile we +// would have thrown away work we had already accepted. Blocking instead pushes +// back through paho to the broker to the till, which is exactly where the +// decision to slow down belongs. +func (p *posPool) submit(job func()) { + p.mu.RLock() + + if p.closed { + p.mu.RUnlock() + // Shutting down. Running it inline still gets the work done and its ack + // published, rather than discarding a bill that already reached us. + job() + return + } + + // The read lock is held across the send. Blocking here while the queue is + // full cannot deadlock against stop: the workers only exit once the channel + // is closed, and that happens under the write lock this send is holding + // off — so they stay alive and keep draining until this send completes. + p.jobs <- job + p.mu.RUnlock() +} + +// stop drains the queue and waits for in-flight work. +// +// Every job already accepted runs to completion, so a bill mid-commit still +// gets its ack. Without one the terminal would hold it and send it again on +// restart — harmless, but avoidable. +func (p *posPool) stop() { + p.once.Do(func() { + // The write lock waits for every submit already in progress, so the + // channel is never closed while something is mid-send. + p.mu.Lock() + p.closed = true + close(p.jobs) + p.mu.Unlock() + + p.wg.Wait() + log.Printf("pos: %s pool drained", p.name) + }) +} + +// posPoolSize reads a worker count from the environment. +// +// The default is deliberately modest. Each worker holds a database transaction +// while it runs, so the useful ceiling is the Postgres connection pool, not the +// CPU — set this above what the database can serve and the workers simply queue +// inside the driver instead, where there is no backpressure to feel. +func posPoolSize(key string, fallback int) int { + v, err := strconv.Atoi(os.Getenv(key)) + if err != nil || v <= 0 { + return fallback + } + if v > 128 { + return 128 + } + return v +} + +// posOrderedDelivery reports whether paho should preserve message order. +// +// Left on: paho then delivers on one goroutine, which hands work to the pool +// and blocks when it is full. That single delivery goroutine is what makes +// backpressure reach the broker at all — with concurrent delivery paho would +// keep reading regardless of how far behind the workers were. +const posOrderedDelivery = true + +// wrapHandler puts a paho message handler behind a pool. +// +// The payload is copied because paho reuses its buffer once the handler +// returns, and the work now happens after that. +func wrapHandler(pool *posPool, h mqtt.MessageHandler) mqtt.MessageHandler { + return func(client mqtt.Client, msg mqtt.Message) { + topic := msg.Topic() + payload := make([]byte, len(msg.Payload())) + copy(payload, msg.Payload()) + + pool.submit(func() { + h(client, copiedMessage{topic: topic, payload: payload}) + }) + } +} + +// copiedMessage carries a payload that outlives paho's buffer. +type copiedMessage struct { + topic string + payload []byte +} + +func (m copiedMessage) Duplicate() bool { return false } +func (m copiedMessage) Qos() byte { return 1 } +func (m copiedMessage) Retained() bool { return false } +func (m copiedMessage) Topic() string { return m.topic } +func (m copiedMessage) MessageID() uint16 { return 0 } +func (m copiedMessage) Payload() []byte { return m.payload } +func (m copiedMessage) Ack() {} diff --git a/messaging/posworkers_test.go b/messaging/posworkers_test.go new file mode 100644 index 0000000..c0d0a2e --- /dev/null +++ b/messaging/posworkers_test.go @@ -0,0 +1,247 @@ +package messaging + +import ( + "sync" + "sync/atomic" + "testing" + "time" + + mqtt "github.com/eclipse/paho.mqtt.golang" +) + +func TestEveryJobRuns(t *testing.T) { + pool := newPosPool("test", 4, 16) + + var done int64 + var wg sync.WaitGroup + for i := 0; i < 100; i++ { + wg.Add(1) + pool.submit(func() { + defer wg.Done() + atomic.AddInt64(&done, 1) + }) + } + + wg.Wait() + pool.stop() + + if got := atomic.LoadInt64(&done); got != 100 { + t.Errorf("ran %d jobs, want 100", got) + } +} + +func TestConcurrencyIsBounded(t *testing.T) { + // The reason the pool exists. Unbounded concurrency would open a database + // transaction per message and exhaust the connection pool under a storm, + // stalling every one of them at once. + const workers = 4 + pool := newPosPool("test", workers, 64) + + var inFlight, peak int64 + var wg sync.WaitGroup + + for i := 0; i < 200; i++ { + wg.Add(1) + pool.submit(func() { + defer wg.Done() + + now := atomic.AddInt64(&inFlight, 1) + for { + was := atomic.LoadInt64(&peak) + if now <= was || atomic.CompareAndSwapInt64(&peak, was, now) { + break + } + } + + time.Sleep(time.Millisecond) + atomic.AddInt64(&inFlight, -1) + }) + } + + wg.Wait() + pool.stop() + + if got := atomic.LoadInt64(&peak); got > workers { + t.Errorf("peak concurrency %d exceeded the %d workers", got, workers) + } +} + +func TestSubmitBlocksRatherThanDroppingWork(t *testing.T) { + // A full queue must slow the caller down, not discard a bill. Dropping + // would throw away work already accepted from the broker, and the terminal + // would only find out at its ack timeout. + pool := newPosPool("test", 1, 1) + + release := make(chan struct{}) + var ran int64 + + // Occupy the single worker. + pool.submit(func() { + <-release + atomic.AddInt64(&ran, 1) + }) + + // Fill the queue, then a third submit must block until the worker frees up. + pool.submit(func() { atomic.AddInt64(&ran, 1) }) + + blocked := make(chan struct{}) + go func() { + pool.submit(func() { atomic.AddInt64(&ran, 1) }) + close(blocked) + }() + + select { + case <-blocked: + t.Fatal("submit returned while the queue was full; work would be dropped under load") + case <-time.After(100 * time.Millisecond): + // Correctly blocked. + } + + close(release) + + select { + case <-blocked: + case <-time.After(3 * time.Second): + t.Fatal("submit never unblocked after the worker freed up") + } + + pool.stop() + + if got := atomic.LoadInt64(&ran); got != 3 { + t.Errorf("ran %d jobs, want 3 — none may be lost", got) + } +} + +func TestStopDrainsAcceptedWork(t *testing.T) { + // A bill mid-commit must still get its ack. Without one the terminal holds + // it and sends it again on restart — harmless, but avoidable. + pool := newPosPool("test", 2, 64) + + var done int64 + for i := 0; i < 50; i++ { + pool.submit(func() { + time.Sleep(time.Millisecond) + atomic.AddInt64(&done, 1) + }) + } + + pool.stop() + + if got := atomic.LoadInt64(&done); got != 50 { + t.Errorf("only %d of 50 jobs completed before shutdown finished", got) + } +} + +func TestSubmitAfterStopStillRunsTheWork(t *testing.T) { + // A message that arrived during shutdown has already been taken from the + // broker. Discarding it would lose a bill we accepted responsibility for. + pool := newPosPool("test", 2, 8) + pool.stop() + + var ran int64 + pool.submit(func() { atomic.AddInt64(&ran, 1) }) + + if got := atomic.LoadInt64(&ran); got != 1 { + t.Error("work submitted during shutdown was discarded") + } +} + +func TestStopIsIdempotent(t *testing.T) { + // Close() may be reached twice on a shutdown path; a second close of the + // jobs channel would panic and take the process down mid-drain. + pool := newPosPool("test", 2, 8) + pool.stop() + pool.stop() + pool.stop() +} + +func TestPoolSizeFallsBackAndClamps(t *testing.T) { + t.Setenv("POS_TEST_WORKERS", "") + if got := posPoolSize("POS_TEST_WORKERS", 8); got != 8 { + t.Errorf("unset = %d, want the fallback 8", got) + } + + t.Setenv("POS_TEST_WORKERS", "not a number") + if got := posPoolSize("POS_TEST_WORKERS", 8); got != 8 { + t.Errorf("garbage = %d, want the fallback 8", got) + } + + t.Setenv("POS_TEST_WORKERS", "0") + if got := posPoolSize("POS_TEST_WORKERS", 8); got != 8 { + t.Errorf("zero = %d, want the fallback 8", got) + } + + t.Setenv("POS_TEST_WORKERS", "-4") + if got := posPoolSize("POS_TEST_WORKERS", 8); got != 8 { + t.Errorf("negative = %d, want the fallback 8", got) + } + + t.Setenv("POS_TEST_WORKERS", "24") + if got := posPoolSize("POS_TEST_WORKERS", 8); got != 24 { + t.Errorf("explicit = %d, want 24", got) + } + + // Clamped: more workers than the database can serve just moves the queue + // inside the driver, where there is no backpressure to feel. + t.Setenv("POS_TEST_WORKERS", "100000") + if got := posPoolSize("POS_TEST_WORKERS", 8); got != 128 { + t.Errorf("absurd = %d, want the 128 clamp", got) + } +} + +func TestAWrappedHandlerCopiesThePayload(t *testing.T) { + // paho reuses its buffer once a handler returns, and with a pool the work + // now happens *after* that. Without a copy a queued bill would be read as + // whatever message happened to arrive next — silently, and as valid JSON + // often enough to commit the wrong sale. + pool := newPosPool("test", 1, 4) + + seen := make(chan string, 1) + wrapped := wrapHandler(pool, func(_ mqtt.Client, msg mqtt.Message) { + seen <- string(msg.Payload()) + }) + + // A buffer paho would reuse. + buffer := []byte(`{"batch_id":"original"}`) + wrapped(nil, fakeMessage{topic: "nearle/pos/12/T4A9/order", payload: buffer}) + + // Overwrite it the instant the handler returns, exactly as paho would. + for i := range buffer { + buffer[i] = 'X' + } + + select { + case got := <-seen: + if got != `{"batch_id":"original"}` { + t.Errorf("handler saw %q — the payload was not copied before queueing", got) + } + case <-time.After(3 * time.Second): + t.Fatal("the wrapped handler never ran") + } + + pool.stop() +} + +func TestAWrappedHandlerKeepsTheTopic(t *testing.T) { + // Store and terminal are read from the topic, never the body. Losing it in + // the hand-off would leave the ack with nowhere to go. + pool := newPosPool("test", 1, 4) + + seen := make(chan string, 1) + wrapped := wrapHandler(pool, func(_ mqtt.Client, msg mqtt.Message) { + seen <- msg.Topic() + }) + + wrapped(nil, fakeMessage{topic: "nearle/pos/1135/T4A9/order", payload: []byte("{}")}) + + select { + case got := <-seen: + if got != "nearle/pos/1135/T4A9/order" { + t.Errorf("topic = %q, want nearle/pos/1135/T4A9/order", got) + } + case <-time.After(3 * time.Second): + t.Fatal("the wrapped handler never ran") + } + + pool.stop() +}