Repository: eclipse-paho/paho.golang

Issue:
DefaultPinger doesn't detect connection loss when consistently sending packets
**Describe the bug**
When a TCP connection is lost (for instance, when switching between wifi networks) the `DefaultPinger` doesn't detect the issue or re-establish the connection.

This happens because `DefaultPinger` calculates the `pingDue` time based on `lastPacketSent`. When packets are consistently being sent (but no replies are ever received) the `DefaultPinger` keeps delaying it's ping message indefinitely.

**To reproduce**
Establish an MQTT connection, send messages on a more frequent cadence than `keepAlive`, and then disrupt the TCP connection.

**Debug output**
```
time                           lastPingSent                   lastPacketSent                 lastPingResponse               pingDue
2025-02-19T07:26:14.930876194Z 0001-01-01T00:00:00Z.          0001-01-01T00:00:00Z           0001-01-01T00:00:00Z           0001-01-01T00:00:30Z
2025-02-19T07:26:44.932315842Z 2025-02-19T07:26:14.932301934Z 2025-02-19T07:26:39.972580833Z 2025-02-19T07:26:14.980298158Z 2025-02-19T07:27:09.972580833Z
2025-02-19T07:27:09.973104232Z 2025-02-19T07:26:14.932301934Z 2025-02-19T07:27:04.974105149Z 2025-02-19T07:26:14.980298158Z 2025-02-19T07:27:34.974105149Z
2025-02-19T07:27:34.974607106Z 2025-02-19T07:26:14.932301934Z 2025-02-19T07:27:29.976906357Z 2025-02-19T07:26:14.980298158Z 2025-02-19T07:27:59.976906357Z
2025-02-19T07:27:59.977479867Z 2025-02-19T07:26:14.932301934Z 2025-02-19T07:27:59.977238334Z 2025-02-19T07:26:14.980298158Z 2025-02-19T07:28:29.977238334Z
2025-02-19T07:28:29.978458208Z 2025-02-19T07:26:14.932301934Z 2025-02-19T07:28:29.976478509Z 2025-02-19T07:26:14.980298158Z 2025-02-19T07:28:59.976478509Z
2025-02-19T07:28:59.977752918Z 2025-02-19T07:26:14.932301934Z 2025-02-19T07:28:59.97761318Z  2025-02-19T07:26:14.980298158Z 2025-02-19T07:29:29.97761318Z
2025-02-19T07:29:29.978714351Z 2025-02-19T07:26:14.932301934Z 2025-02-19T07:29:25.262394398Z 2025-02-19T07:26:14.980298158Z 2025-02-19T07:29:55.262394398Z
2025-02-19T07:29:55.26762407Z  2025-02-19T07:26:14.932301934Z 2025-02-19T07:29:45.35189184Z  2025-02-19T07:26:14.980298158Z 2025-02-19T07:30:15.35189184Z
2025-02-19T07:30:15.361808698Z 2025-02-19T07:26:14.932301934Z 2025-02-19T07:30:05.603714165Z 2025-02-19T07:26:14.980298158Z 2025-02-19T07:30:35.603714165Z
2025-02-19T07:30:35.604367247Z 2025-02-19T07:26:14.932301934Z 2025-02-19T07:30:25.865425995Z 2025-02-19T07:26:14.980298158Z 2025-02-19T07:30:55.865425995Z
2025-02-19T07:30:55.874694767Z 2025-02-19T07:26:14.932301934Z 2025-02-19T07:30:45.94657684Z  2025-02-19T07:26:14.980298158Z 2025-02-19T07:31:15.94657684Z
2025-02-19T07:31:15.955379912Z 2025-02-19T07:26:14.932301934Z 2025-02-19T07:31:06.031695079Z 2025-02-19T07:26:14.980298158Z 2025-02-19T07:31:36.031695079Z
```

**Expected behaviour**
`DefaultPinger` should detect the lost network connection and re-establish the MQTT connection.

**Software used:**
github.com/eclipse/paho.golang v0.21.0

**Additional context**



Hints from issue comments:
@MattBrittan: Interesting. In this situation I'd expect the loss of connection to be picked when an attempt is made to write to (or, generally, read from) the connection. I think there might be a small bug here;

At QOS 0 [we](https://github.com/eclipse-paho/paho.golang/blob/a13d64c6e20199977a3b70fa728a1b3e0af5e908/paho/client.go#L867C15-L867C19):

```
c.debug.Println("sending QoS0 message")
		if _, err := pb.WriteTo(c.config.Conn); err != nil {
			go c.error(err)
			return nil, err
		}
```

but at [QOS 1/2](https://github.com/eclipse-paho/paho.golang/blob/a13d64c6e20199977a3b70fa728a1b3e0af5e908/paho/client.go#L892) we don't call `go c.error(err)` (and I think we probably should). Can you please confirm what QOS you are using when publishing?

Generally I'd also expect the connection loss to be picked in [`incomming`](https://github.com/eclipse-paho/paho.golang/blob/a13d64c6e20199977a3b70fa728a1b3e0af5e908/paho/client.go#L493).

The ping process exists to detect half-open connections; sending a packet should be enough to detect this (should get a TCP timeout) so I don't believe that an additional ping is needed in this case. I [do test this](https://github.com/eclipse-paho/paho.golang/blob/master/autopaho/examples/docker/readme.md#simulating-network-connection-loss) and the connection loss is picked up in that situation, so what is happening in your case must vary. 

I'd appreciate it if you could share more information about how you are using the library (i.e. are you using `autopaho`?), debug logs would also be useful.@AndrewBase: I'm using QOS1 and autopaho. I'll try to get you code and logs tomorrow.

From what I understand the TCP timeout can take up to 15 minutes to detect a dead connection. gRPC fixed a similar issue by setting TCP_USER_TIMEOUT on the socket on Linux:
https://github.com/grpc/grpc-go/pull/2307
https://www.evanjones.ca/tcp-connection-timeouts.html
https://blog.cloudflare.com/when-tcp-sockets-refuse-to-die/#busy-estab-socket-is-not-forever@MattBrittan: > I'll try to get you code and logs tomorrow.

Don't worry too much about this (as you are using `autopaho` rather than something custom). 

If you have time, try chucking a `go c.error(err)` [here](https://github.com/eclipse-paho/paho.golang/blob/a13d64c6e20199977a3b70fa728a1b3e0af5e908/paho/client.go#L892). I'm not sure if `pb.WriteTo` will fail or not (I guess this is where having logs might be nice, I develop on a windows box so would need to spin up a VM to try and test this). It would be interesting to know if the connection is reestablished after a longer wait?@AndrewBase: Partially redacted log messages:
[logs.txt](https://github.com/user-attachments/files/18893218/logs.txt)

Code for how we create the paho client:
```
	tlsConfig := &tls.Config{
		Certificates: tlsCertificate,
		RootCAs:      certpool,
	}

	// Router can be used to route incoming messages (from subscription topics) to channels.
	router := paho.NewStandardRouter()
	router.DefaultHandler(func(p *paho.Publish) { fmt.Printf("defaulthandler received message with topic: %s\n", p.Topic) })

	brokerUrl, err := url.Parse(broker)
	if err != nil {
		return nil, fmt.Errorf("error parsing broker URL: %w", err)
	}
	cliCfg := autopaho.ClientConfig{
		ServerUrls:        []*url.URL{brokerUrl},
		TlsCfg:            tlsConfig,
		KeepAlive:         20,                     // Keepalive message recommended in paho.golang documentation to be sent every 20 seconds.
		ConnectRetryDelay: 500 * time.Millisecond, // Set to 1/2 second. NW-1808, was set to the default of 10 seconds but seemed to cause issues with publishing telemetry data.
		OnConnectionUp: func(cm *autopaho.ConnectionManager, connAck *paho.Connack) {
			subscriptions := make([]paho.SubscribeOptions, 0)
			for topic, QOS := range topicsQOSMap {
				subscription := paho.SubscribeOptions{
					Topic: topic,
					QoS:   QOS,
				}
				subscriptions = append(subscriptions, subscription)
			}
			if len(subscriptions) == 0 {
				return
			}
			if _, err := cm.Subscribe(context.Background(), &paho.Subscribe{Subscriptions: subscriptions}); err != nil {
				slog.Error(fmt.Sprintf("error subscribing to topics %v: %v", topicsQOSMap, err))
			}
		},
		ClientConfig: paho.ClientConfig{
			ClientID:      clientID,
			OnClientError: func(err error) { slog.Error(fmt.Sprintf("client error: %s\n", err)) },
			OnServerDisconnect: func(d *paho.Disconnect) {
				slog.Warn(fmt.Sprintf("server requested disconnect; reason code: %s\n", string(d.ReasonCode)))
			},
			OnPublishReceived: []func(paho.PublishReceived) (bool, error){
				func(pr paho.PublishReceived) (bool, error) {
					router.Route(pr.Packet.Packet())
					return true, nil // We assume that the router handles all messages.
				},
			},
		},
	}
	cliCfg.Debug = newMqttLogger(*slog.Default(), slog.LevelDebug)
	cliCfg.PahoDebug = newMqttLogger(*slog.Default(), slog.LevelDebug)

	// Connect to the server - this will return immediately after initiating the connection process
	client, err := autopaho.NewConnection(ctx, cliCfg)
	if err != nil {
		return nil, fmt.Errorf("error creating MQTT connection: %w", err)
	}
```@MattBrittan: Interesting, it's not what I suspected (there is no indication in the log that the loss of connection has been detected). This seems to be an issue with the TCP stack (would generally expect the loss of connection to be noted).

The [MQTT Spec](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Keep_Alive_1) only requires the client to send a PING when the gap between outgoing packets exceeds the Keep Aliveperiod. This is what this library currently does.

I've taken a look at other libraries. The [Go paho.mqtt.golang v3.1.1 client](https://github.com/eclipse-paho/paho.mqtt.golang/blob/514b7fa0af8f25cf2b354961ef3ec550a65016b8/ping.go#L54C10-L54C23) sends a ping based on both sent and received packets, as do the [python](https://github.com/eclipse-paho/paho.mqtt.python/blob/d45de3737879cfe7a6acc361631fa5cb1ef584bb/src/paho/mqtt/client.py#L3271) and [c](https://github.com/eclipse-paho/paho.mqtt.c/blob/2150ba29d9df24ad1733c460eb099f292af84ee5/src/MQTTProtocolClient.c#L774) clients. As such, I think it would be reasonable for this library to take the same approach to catch this kind of issue.

Happy to receive a PR for this, otherwise I'll try to have a look within a couple of weeks.@MattBrittan: Please test with the changes in @master and let me know if this resolves your issue.@MattBrittan: I believe that #292 resolved this but there has been no further updates on the issue. As such I'm going to close this, please feel free to reopen if the issue is not resolved.

Solution patch:
diff --git a/paho/client.go b/paho/client.go
index a772da8..cb03847 100644
--- a/paho/client.go
+++ b/paho/client.go
@@ -494,6 +494,7 @@ func (c *Client) incoming(ctx context.Context) {
 				go c.error(err)
 				return
 			}
+			c.config.PingHandler.PacketReceived()
 			switch recv.Type {
 			case packets.CONNACK:
 				c.debug.Println("received CONNACK (unexpected)")
diff --git a/paho/pinger.go b/paho/pinger.go
index 0eb28cc..67c1e84 100644
--- a/paho/pinger.go
+++ b/paho/pinger.go
@@ -36,6 +36,9 @@ type Pinger interface {
 	// PacketSent is called when a packet is sent to the server.
 	PacketSent()
 
+	// PacketReceived is called when a packet (possibly a PINGRESP) is received from the server
+	PacketReceived()
+
 	// PingResp is called when a PINGRESP is received from the server.
 	PingResp()
 
@@ -46,8 +49,9 @@ type Pinger interface {
 
 // DefaultPinger is the default implementation of Pinger.
 type DefaultPinger struct {
-	lastPacketSent   time.Time
-	lastPingResponse time.Time
+	lastPacketSent     time.Time
+	lastPacketReceived time.Time
+	lastPingResponse   time.Time
 
 	debug log.Logger
 
@@ -99,7 +103,16 @@ func (p *DefaultPinger) Run(ctx context.Context, conn net.Conn, keepAlive uint16
 		case t := <-timer.C:
 			p.mu.Lock()
 			lastPingResponse := p.lastPingResponse
-			pingDue := p.lastPacketSent.Add(interval)
+			// The MQTT Spec only requires that a ping be sent if no control packets have been SENT within the keepalive
+			// period (MQTT-3.1.2-20). Only sending PING in that one case can cause issues if the only activity is
+			// outgoing messages, a half-open connection should result in a TCP timeout but this can take a long time
+			//(issue #288). To address this we PING if we have not both sent, and received, packets within keepAlive.
+			var pingDue time.Time
+			if p.lastPacketSent.Before(p.lastPacketReceived) {
+				pingDue = p.lastPacketSent.Add(interval)
+			} else {
+				pingDue = p.lastPacketReceived.Add(interval)
+			}
 			p.mu.Unlock()
 
 			if !lastPingSent.IsZero() && lastPingSent.After(lastPingResponse) {
@@ -136,6 +149,12 @@ func (p *DefaultPinger) PacketSent() {
 	p.lastPacketSent = time.Now()
 }
 
+func (p *DefaultPinger) PacketReceived() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.lastPacketReceived = time.Now()
+}
+
 func (p *DefaultPinger) PingResp() {
 	p.mu.Lock()
 	defer p.mu.Unlock()



Test patch:
diff --git a/autopaho/auto_test.go b/autopaho/auto_test.go
index f52faf2..6b5e71d 100644
--- a/autopaho/auto_test.go
+++ b/autopaho/auto_test.go
@@ -38,10 +38,12 @@ import (
 
 const shortDelay = 500 * time.Millisecond // Used when something should happen pretty quickly (increase when debugging)
 const longerDelay = time.Second           // Longer delay than above (for things like test wide context timeout)
+const longestDelay = 5 * longerDelay      // Used when queuing a lot of messages (extra time needed, especially when race detector active)
 
 // When debugging uncomment the below (to prevent tests terminating too quickly)
 // const shortDelay = time.Hour
 // const longerDelay = time.Hour
+// const longestDelay = time.Hour
 
 const dummyURL = "tcp://127.0.0.1:1883"
 
diff --git a/autopaho/queue_test.go b/autopaho/queue_test.go
index a4ea8c1..53ec9f1 100644
--- a/autopaho/queue_test.go
+++ b/autopaho/queue_test.go
@@ -141,7 +141,7 @@ func TestQueuedMessages(t *testing.T) {
 		},
 	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), longerDelay)
+	ctx, cancel := context.WithTimeout(context.Background(), longestDelay)
 	defer cancel()
 	cm, err := NewConnection(ctx, config)
 	if err != nil {
@@ -204,8 +204,8 @@ func TestQueuedMessages(t *testing.T) {
 	// Wait for all messages to be received
 	select {
 	case <-got200Messages:
-	case <-time.After(5 * longerDelay):
-		t.Fatal("timeout awaiting messages")
+	case <-time.After(longestDelay):
+		t.Fatalf("%s: timeout awaiting messages (got %d) ", time.Now(), len(receivedPublish)) // this is a data race but delay should prevent issues
 	}
 
 	// Disconnect
diff --git a/internal/basictestserver/server.go b/internal/basictestserver/server.go
index 81e7dc0..507ca6e 100644
--- a/internal/basictestserver/server.go
+++ b/internal/basictestserver/server.go
@@ -44,19 +44,24 @@ type TestServer struct {
 }
 
 func New(logger Logger) *TestServer {
+	// net.Pipe does not implement net.buffersWriter so packets may be mixed up if not wrapped
+	conn, clientConn := net.Pipe()
 	t := &TestServer{
-		stop:      make(chan struct{}),
-		done:      make(chan struct{}),
-		responses: make(map[byte]packets.Packet),
-		logger:    logger,
+		conn:       packets.NewThreadSafeConn(conn),
+		clientConn: packets.NewThreadSafeConn(clientConn),
+		stop:       make(chan struct{}),
+		done:       make(chan struct{}),
+		responses:  make(map[byte]packets.Packet),
+		logger:     logger,
 	}
-	t.conn, t.clientConn = net.Pipe()
+
 	if logger == nil {
 		t.logger = log.Default()
 	}
 	return t
 }
 
+// ClientConn returns a connection to be used when creating the Paho Client
 func (t *TestServer) ClientConn() net.Conn {
 	return t.clientConn
 }
@@ -65,6 +70,7 @@ func (t *TestServer) SetResponse(pt byte, p packets.Packet) {
 	t.responses[pt] = p
 }
 
+// SendPacket is used to send a packet to the client
 func (t *TestServer) SendPacket(p packets.Packet) error {
 	_, err := p.WriteTo(t.conn)
 
diff --git a/paho/default_pinger_test.go b/paho/default_pinger_test.go
index b6c0e8d..ca1d4f8 100644
--- a/paho/default_pinger_test.go
+++ b/paho/default_pinger_test.go
@@ -97,7 +97,7 @@ func TestDefaultPingerSuccess(t *testing.T) {
 	}
 }
 
-func TestDefaultPingerPacketSent(t *testing.T) {
+func TestDefaultPingerPacketSentReceived(t *testing.T) {
 	fakeClientConn, fakeServerConn := net.Pipe()
 
 	pinger := NewDefaultPinger()
@@ -110,7 +110,7 @@ func TestDefaultPingerPacketSent(t *testing.T) {
 		pingResult <- pinger.Run(ctx, fakeClientConn, 3)
 	}()
 
-	// keep calling PacketSent() in a goroutine to check that the Pinger avoids sending PINGREQs when not needed
+	// keep calling PacketSent() (and PacketReceived) in a goroutine to check that the Pinger avoids sending PINGREQs when not needed
 	stop := make(chan struct{})
 	go func() {
 		for {
@@ -119,8 +119,9 @@ func TestDefaultPingerPacketSent(t *testing.T) {
 				return
 			default:
 			}
-			// keep calling PacketSent()
+			// keep calling PacketSent() and PacketReceived
 			pinger.PacketSent()
+			pinger.PacketReceived()
 		}
 	}()
 	defer close(stop)
@@ -138,7 +139,7 @@ func TestDefaultPingerPacketSent(t *testing.T) {
 			if recv.Type == packets.PINGREQ {
 				count++
 				pinger.PingResp()
-				if count > 1 { // we allow the count to be 1 because the first PINGREQ is sent immediately
+				if count == 2 { // we allow the count to be 1 because the first PINGREQ is sent immediately
 					close(tooManyPingreqs)
 				}
 			}
@@ -156,6 +157,125 @@ func TestDefaultPingerPacketSent(t *testing.T) {
 	}
 }
 
+// TestDefaultPingerPacketSentOnly - If packets are being sent, but nothing received, then a pingreq should be sent
+func TestDefaultPingerPacketSentOnly(t *testing.T) {
+	fakeClientConn, fakeServerConn := net.Pipe()
+
+	pinger := NewDefaultPinger()
+	pinger.SetDebug(paholog.NewTestLogger(t, "DefaultPinger:"))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	pingResult := make(chan error, 1)
+	go func() {
+		pingResult <- pinger.Run(ctx, fakeClientConn, 3)
+	}()
+
+	// keep calling PacketSent() (and PacketReceived) in a goroutine to check that the Pinger avoids sending PINGREQs when not needed
+	stop := make(chan struct{})
+	go func() {
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+			// keep calling PacketSent()
+			pinger.PacketSent()
+		}
+	}()
+	defer close(stop)
+
+	// keep reading from fakeServerConn and call PingResp() when a PINGREQ is received
+	// if more than one PINGREQ is received, the test will fail
+	count := 0
+	twoPingreqs := make(chan struct{})
+	go func() {
+		for {
+			recv, err := packets.ReadPacket(fakeServerConn)
+			if err != nil {
+				return
+			}
+			if recv.Type == packets.PINGREQ {
+				count++
+				pinger.PingResp()
+				if count == 2 {
+					close(twoPingreqs)
+				}
+			}
+		}
+	}()
+	defer fakeServerConn.Close()
+
+	select {
+	case <-twoPingreqs:
+		// pass
+	case err := <-pingResult:
+		t.Errorf("expected DefaultPinger to not return error, got %v", err)
+	case <-time.After(10 * time.Second):
+		t.Error("expected DefaultPinger to send PINGREQs when no packets received")
+	}
+}
+
+// TestDefaultPingerPacketReceiveOnly - If packets are being received, but nothing sent, then a pingreq should be sent
+func TestDefaultPingerPacketReceiveOnly(t *testing.T) {
+	fakeClientConn, fakeServerConn := net.Pipe()
+
+	pinger := NewDefaultPinger()
+	pinger.SetDebug(paholog.NewTestLogger(t, "DefaultPinger:"))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	pingResult := make(chan error, 1)
+	go func() {
+		pingResult <- pinger.Run(ctx, fakeClientConn, 3)
+	}()
+
+	// keep calling PacketSent() (and PacketReceived) in a goroutine to check that the Pinger avoids sending PINGREQs when not needed
+	stop := make(chan struct{})
+	go func() {
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+			pinger.PacketReceived()
+		}
+	}()
+	defer close(stop)
+
+	// keep reading from fakeServerConn and call PingResp() when a PINGREQ is received
+	// if more than one PINGREQ is received, the test will fail
+	count := 0
+	twoPingreqs := make(chan struct{})
+	go func() {
+		for {
+			recv, err := packets.ReadPacket(fakeServerConn)
+			if err != nil {
+				return
+			}
+			if recv.Type == packets.PINGREQ {
+				count++
+				pinger.PingResp()
+				if count == 2 {
+					close(twoPingreqs)
+				}
+			}
+		}
+	}()
+	defer fakeServerConn.Close()
+
+	select {
+	case <-twoPingreqs:
+		// pass
+	case err := <-pingResult:
+		t.Errorf("expected DefaultPinger to not return error, got %v", err)
+	case <-time.After(10 * time.Second):
+		t.Error("expected DefaultPinger to send PINGREQs when no packets sent")
+	}
+}
+
 func TestDefaultPingerStartStop(t *testing.T) {
 	fakeServerConn, fakeClientConn := net.Pipe()