From c952afdd29eaced96da9e621ba6fca8d997ed205 Mon Sep 17 00:00:00 2001
From: MrAlders0n <MrAlders0n@smal.ca>
Date: Wed, 30 Sep 2026 10:28:03 -0400
Subject: [PATCH] fix(ingest): let paho auto-ack MQTT deliveries

Sessions are clean and client IDs random, so deferring acks never bought redelivery; it only capped delivery at the broker's inflight window and pushed backlog into the broker queue, where drops are silent.
---
 internal/ingest/ingest.go     |  8 ++------
 internal/ingest/queue_test.go | 15 ++-------------
 2 files changed, 4 insertions(+), 19 deletions(-)

diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go
index 289a0de..dcbe2bb 100644
--- a/internal/ingest/ingest.go
+++ b/internal/ingest/ingest.go
@@ -249,7 +249,6 @@ func (w *Worker) Start(ctx context.Context) {
 	}
 	queue := newMessageQueue(8, 2048, 32<<20, func(ctx context.Context, msg mqtt.Message) {
 		w.handleMessageContext(ctx, msg)
-		msg.Ack()
 	})
 	reportDrops := func() {
 		if n := queue.takeDropped(); n > 0 {
@@ -268,13 +267,13 @@ func (w *Worker) Start(ctx context.Context) {
 
 	// Isolate workers across deployments; Paho reuses this ID on reconnect.
 	// Keep it alphanumeric and within MQTT 3.1's 23-character client ID limit.
+	// Auto-ack: sessions are clean, so deferring acks only throttled the broker's inflight window.
 	opts := mqtt.NewClientOptions().
 		AddBroker(w.cfg.URL).
 		SetClientID(rand.Text()[:23]).
 		SetUsername(w.cfg.Username).
 		SetPassword(w.cfg.Password).
 		SetAutoReconnect(true).
-		SetAutoAckDisabled(true).
 		SetMaxReconnectInterval(30 * time.Second).
 		SetKeepAlive(30 * time.Second).
 		SetPingTimeout(10 * time.Second).
@@ -335,10 +334,7 @@ func (w *Worker) subscribe(client mqtt.Client, queue *messageQueue) {
 	// meshcore/{IATA}/{pubkey}/status
 	// We do NOT subscribe to /internal (Role 2 access).
 	tok := client.Subscribe("meshcore/#", 1, func(_ mqtt.Client, msg mqtt.Message) {
-		if !queue.enqueue(msg) {
-			// Counted drops must release the broker's inflight slot.
-			msg.Ack()
-		}
+		queue.enqueue(msg)
 	})
 	if tok.Wait() && tok.Error() != nil {
 		w.log.Error("subscribe error", "error", tok.Error())
diff --git a/internal/ingest/queue_test.go b/internal/ingest/queue_test.go
index 130a17a..6b43fe4 100644
--- a/internal/ingest/queue_test.go
+++ b/internal/ingest/queue_test.go
@@ -141,24 +141,13 @@ type completedSubscribe struct{ mqtt.Token }
 func (completedSubscribe) Wait() bool   { return true }
 func (completedSubscribe) Error() error { return nil }
 
-type ackMessage struct {
-	mqtt.Message
-	acked bool
-}
-
-func (m *ackMessage) Ack() { m.acked = true }
-
-func TestSubscribeReleasesRejectedMessage(t *testing.T) {
+func TestSubscribeCountsRejectedMessage(t *testing.T) {
 	w, _ := newTestWorker()
 	q := newMessageQueue(1, 1, 1, func(context.Context, mqtt.Message) { t.Error("rejected message processed") })
 	defer q.close(context.Background())
 	client := &subscribeCapture{}
 	w.subscribe(client, q)
-	m := &ackMessage{Message: queueMessage("aa", "packets", 0)}
-	client.callback(client, m)
-	if !m.acked {
-		t.Fatal("rejected message still occupies the broker inflight window")
-	}
+	client.callback(client, queueMessage("aa", "packets", 0))
 	if q.takeDropped() != 1 {
 		t.Fatal("rejected message was not counted")
 	}