Process the MQTT ingest on a bounded worker pool
paho delivers on one goroutine, so bills were committed strictly one after another. Each is a full Postgres transaction — advisory lock, dedup, stock row locks, availability check, four inserts, commit — which is 10-30ms, so the ceiling was roughly 30-100 bills a second and a shop's backlog draining after an outage took minutes to land. A fixed pool behind a bounded queue, rather than a goroutine per message. Unbounded concurrency would open a transaction per message and exhaust the connection pool under a storm, stalling every one of them at once — a slow minute turned into a dead one. When the queue fills, submit blocks: paho stops acknowledging, the broker's in-flight window fills, it stops sending, and the backpressure reaches the till, which holds its bills and retries. Slow, but nothing is dropped. Heartbeats get their own pool. Sharing one would let a backlog of bills delay presence, so every till would appear to go dark at exactly the moment the system was busiest — the worst time to be blind to which counters are alive. Payloads are copied on the way in. paho reuses its buffer once a handler returns and the work now happens after that, so a queued bill would otherwise be read as whatever message arrived next — silently, and as valid JSON often enough to commit the wrong sale. One bug found by its own test: submit-after-stop selected between a done-channel and the job channel, and once both were ready Go picks at random. Picking the send panics on a closed channel. It would have shown up in production as an occasional crash during shutdown and nowhere else. Now guarded by an RWMutex held across the send, so the queue cannot be closed under one in progress. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Suriya
@@ -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
|
||||
|
||||
174
messaging/posworkers.go
Normal file
174
messaging/posworkers.go
Normal file
@@ -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() {}
|
||||
247
messaging/posworkers_test.go
Normal file
247
messaging/posworkers_test.go
Normal file
@@ -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()
|
||||
}
|
||||
Reference in New Issue
Block a user