Log MQTT packets dropped while a broker is down

CHANGELOG.md

@@ -43,6 +43,13 @@ built on meshcore-go v1.7.0.
   rather than queuing behind relayed traffic, and it ignores packets firmware discards, such as
   encrypted payloads that are not whole AES blocks.
 
+### Fixed
+
+- **Packets dropped while a broker is down are now logged.** They are still not buffered, but a
+  warning says when a broker starts dropping them, and the count is logged when it reconnects.
+  Before, they went missing without a word, including any heard in the moment before the first
+  connection.
+
 ### Upgrading
 
 - **Docker: switch the image to `ghcr.io/owlshack/owlbot`.** `ghcr.io/meshcore-go/meshcore-bot`

mqtt.go

@@ -37,6 +37,9 @@ type brokerClient struct {
 	stop       chan struct{} // closed by Stop to halt the worker; publishCh is never closed, so an in-flight send can't panic
 	workerDone chan struct{}
 	dropped    atomic.Uint64
+	// downDropped counts packets dropped since the broker went unreachable.
+	// Only publishWorker touches it.
+	downDropped uint64
 	// retrying is true while a connectAndRetry loop owns this broker. Needed
 	// now that BOTH Start and a failed token refresh can spawn one: without it
 	// two loops race on the same client pointer and both swapClient.
@@ -430,10 +433,16 @@ func (o *MqttObserver) publishWorker(bc *brokerClient) {
 
 func (o *MqttObserver) doPublish(bc *brokerClient, job publishJob) {
 	client := bc.currentClient()
-	if client == nil || !client.IsConnected() {
+	// IsConnected stays true while paho auto-reconnects, and paho then
+	// completes a QoS 0 publish without sending or storing it. Drop those here
+	// too, so the outage is counted and logged rather than silent.
+	if client == nil || !client.IsConnected() || (job.qos == 0 && !client.IsConnectionOpen()) {
 		// Publishing to a disconnected client blocks for the full
-		// publishWaitTimeout per job and stalls the worker.
-		bc.dropped.Add(1)
+		// publishWaitTimeout per job and stalls the worker. No buffer: a
+		// packet heard while the broker is down is not sent later.
+		if bc.downDropped++; bc.downDropped == 1 {
+			o.log.Warn("broker not connected, dropping packets until it is", "broker", bc.cfg.Name)
+		}
 		return
 	}
 	token := client.Publish(job.topic, job.qos, job.retain, job.payload)
@@ -443,6 +452,11 @@ func (o *MqttObserver) doPublish(bc *brokerClient, job publishJob) {
 	}
 	if err := token.Error(); err != nil {
 		o.log.Error("publish error", "broker", bc.cfg.Name, "error", err)
+		return
+	}
+	if bc.downDropped > 0 {
+		o.log.Info("broker publishing again", "broker", bc.cfg.Name, "dropped", bc.downDropped)
+		bc.downDropped = 0
 	}
 }
 

stats_test.go

@@ -810,3 +810,52 @@ func TestRefreshFailureDropsStaleClientSoRetryDials(t *testing.T) {
 	close(bc.stop)
 	waitLoopsDone(t, bc)
 }
+
+// fakeLinkClient is connected, reconnecting or down on demand.
+type fakeLinkClient struct {
+	mqtt.Client
+	up           bool
+	reconnecting bool // paho with AutoReconnect: IsConnected true, connection not open
+}
+
+func (f *fakeLinkClient) IsConnected() bool      { return f.up || f.reconnecting }
+func (f *fakeLinkClient) IsConnectionOpen() bool { return f.up }
+func (f *fakeLinkClient) Publish(string, byte, bool, any) mqtt.Token {
+	return &mqtt.DummyToken{}
+}
+
+// Packets heard while a broker is down are dropped, not buffered. That must be
+// visible: one warning when the dropping starts, and the count once publishing
+// resumes, not a line per packet.
+func TestDoPublishLogsDropsWhileBrokerDown(t *testing.T) {
+	var logs strings.Builder
+	o := &MqttObserver{log: slog.New(slog.NewTextHandler(&logs, nil))}
+	fc := &fakeLinkClient{}
+	bc := &brokerClient{cfg: BrokerConfig{Name: "b"}}
+	bc.swapClient(fc)
+
+	for range 3 {
+		o.doPublish(bc, publishJob{topic: "t"})
+	}
+	if n := strings.Count(logs.String(), "dropping packets"); n != 1 {
+		t.Fatalf("%d drop warnings for one outage, want 1:\n%s", n, logs.String())
+	}
+
+	fc.up = true
+	o.doPublish(bc, publishJob{topic: "t"})
+	o.doPublish(bc, publishJob{topic: "t"})
+	if !strings.Contains(logs.String(), "broker publishing again") || !strings.Contains(logs.String(), "dropped=3") {
+		t.Fatalf("recovery should report the 3 dropped packets once:\n%s", logs.String())
+	}
+	if n := strings.Count(logs.String(), "publishing again"); n != 1 {
+		t.Errorf("%d recovery lines, want 1", n)
+	}
+
+	// An outage mid-run looks like this to paho: still "connected" while it
+	// reconnects, and a QoS 0 publish then vanishes. It must warn again.
+	fc.up, fc.reconnecting = false, true
+	o.doPublish(bc, publishJob{topic: "t"})
+	if n := strings.Count(logs.String(), "dropping packets"); n != 2 {
+		t.Errorf("%d drop warnings after a mid-run outage, want 2", n)
+	}
+}