Keepalive ping write can block indefinitely behind stuck publishes, defeating keepalive and delaying half-open connection detection

[dhruvjain99]

Background:
The MQTT keepalive goroutine sends a PINGREQ when no activity has been seen for the keepalive interval or no inbound packets have arrived. The code currently does this with a plain ping.Write(conn) call. If the TCP send buffer is backed up because an earlier PUBLISH is unacknowledged (half-open / stuck), the ping.Write can block indefinitely. That prevents the keepalive from progressing, delays detection of a dead connection, and defers recovery until kernel-level retransmit limits (tcp_retries2) trigger or other higher-level timeouts fire.

Problem:

• ping.Write(conn) can hang if the underlying TCP socket is under backpressure (e.g., due to unacknowledged in-flight PUBLISH).
• Keepalive logic relies on being able to send a PINGREQ and then observe a PINGRESP; if the write blocks forever, neither happens, so the connection liveness check is effectively stalled.
• This results in late disconnection detection in half-open scenarios, increasing the time to failover/reconnect.

Reproduction steps (high level):

1. Establish a connection with a PUBLISH in flight that does not get acknowledged (simulate a silent peer or dropped ACK).
2. Let the PUBLISH saturate the send buffer / create backpressure.
3. Wait for keepalive to trigger; observe that ping.Write blocks because the socket is clogged.
4. No PINGRESP is received; connection isn’t torn down until much later (e.g., due to tcp_retries2 expiry), even though the peer is effectively dead.

Proposal:

• Before calling ping.Write(conn), set a bounded write deadline so the write cannot block indefinitely.
• Example change:

// writer serializes deadline+write+clear so concurrent goroutines don't race.
type deadlineWriter struct {
    conn net.Conn
    mu   sync.Mutex
}

func newDeadlineWriter(conn net.Conn) *deadlineWriter {
    return &deadlineWriter{conn: conn}
}

// WriteWithTimeout sets a per-write deadline, does the write, then clears it.
func (w *deadlineWriter) WriteWithTimeout(packetsWriter func(io.Writer) error, timeout time.Duration) error {
    w.mu.Lock()
    defer w.mu.Unlock()

    if timeout > 0 {
        _ = w.conn.SetWriteDeadline(time.Now().Add(timeout))
    }
    // perform the actual write via callback so callers can pass arbitrary packet logic
    err := packetsWriter(w.conn)
    // clear the deadline so it does not linger
    _ = w.conn.SetWriteDeadline(time.Time{})

    return err
}

// assume you wrap the raw conn once and reuse:
safeWriter := newDeadlineWriter(conn)

// when sending ping:
if atomic.LoadInt32(&c.pingOutstanding) == 0 {
    DEBUG.Println(PNG, "keepalive sending ping")
    ping := packets.NewControlPacket(packets.Pingreq).(*packets.PingreqPacket)
    atomic.StoreInt32(&c.pingOutstanding, 1)

    // bounded write to avoid blocking forever behind stuck publishes
    err := safeWriter.WriteWithTimeout(func(w io.Writer) error {
        return ping.Write(w)
    }, 2*time.Second) // make 2s configurable if desired

    if err != nil {
        if ne, ok := err.(net.Error); ok && ne.Timeout() {
            CRITICAL.Println(PNG, "keepalive ping write timed out, disconnecting:", err)
            c.internalConnLost(fmt.Errorf("keepalive ping write timeout: %w", err))
            return
        }
        ERROR.Println(PNG, "keepalive ping write error:", err)
        // optionally treat other errors as fatal too
    } else {
        c.lastSent.Store(time.Now())
        pingSent = time.Now()
    }
}

[dhruvjain99]

Another Proposal

Add PUBACK watchdog to detect stalled QoS1/2 publishes and recover from half-open connections faster

Problems it can address

1. No application-level timeout for missing PUBACK/PUBCOMP.
2. A stalled publish can block new messages and keepalive pings (if the socket is saturated) for many minutes.
3. The connection may appear “alive” to keepalive pings even though specific publishes are never acknowledged.

Implementation Details

1. When sending a QoS1/2 publish, start a timer for a configurable PublishAckTimeout (e.g., 5–10 seconds).
2. If the expected acknowledgment is not received within that window:
3. Log the failure with the packet ID.
4. Treat the connection as unhealthy and call internalConnLost (or equivalent reconnect logic).
5. Optionally retry the publish after reconnect, depending on client QoS semantics.

[dhruvjain99]

Problem:
ping.Write(conn) can hang if the underlying TCP socket is under backpressure (e.g., due to unacknowledged in-flight PUBLISH).
Keepalive logic relies on being able to send a PINGREQ and then observe a PINGRESP; if the write blocks forever, neither happens, so the connection liveness check is effectively stalled.
This results in late disconnection detection in half-open scenarios, increasing the time to failover/reconnect.

The above problem can be reproduced using the following client code when run on an ubuntu machine or container and incoming TCP traffic from port 1883 can be blocked using iptables. The linux/kernel level configs will remain as is(default).

root@colima:/go/src/sim# cat main.go
package main

import (
	"fmt"
	"log"
	"os"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"
)

func main() {
	// Set MQTT logging levels
	mqtt.ERROR = log.New(os.Stderr, "[ERROR] ", log.LstdFlags)
	mqtt.CRITICAL = log.New(os.Stderr, "[CRIT] ", log.LstdFlags)
	mqtt.WARN = log.New(os.Stderr, "[WARN] ", log.LstdFlags)
	mqtt.DEBUG = log.New(os.Stderr, "[DEBUG] ", log.LstdFlags)

	// Define broker address and client ID
	broker := "tcp://host.docker.internal:1883" // replace with your broker
	clientID := "test-client-keepalive-10000"

	opts := mqtt.NewClientOptions().
		AddBroker(broker).
		SetClientID(clientID).
		SetKeepAlive(30 * time.Second).
		SetAutoReconnect(true).
		SetConnectRetry(true).
		SetConnectRetryInterval(10 * time.Second).
		SetCleanSession(true)

	// Optional: Message handler
	opts.OnConnect = func(c mqtt.Client) {
		fmt.Println("Connected to broker")
	}

	opts.OnConnectionLost = func(c mqtt.Client, err error) {
		fmt.Printf("Connection lost: %v\n", err)
	}

	client := mqtt.NewClient(opts)

	// Connect to the broker
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		panic(token.Error())
	}

	for {
		time.Sleep(2 * time.Second)
		// Large publish to saturate send buffer
		payload := make([]byte, 5<<20) // 5MB payload
		for i := range payload {
			payload[i] = 'A'
		}

		fmt.Println("[client] starting large publish to clog send buffer")
		go func() {
			token := client.Publish("test/topic", 1, false, payload)
			// This will block internally when send buffer is full (due to broker not reading)
			token.Wait()
			if err := token.Error(); err != nil {
				fmt.Println("[client] publish error:", err)
			} else {
				fmt.Println("[client] publish completed")
			}
		}()
	}

}

This will stall the publish indefinitely because the socket will be full which will also block PINGREQ packet thus resulting in a late half-open connection detection until the kernel level tcp_retries2 aborts the connection.

[dhruvjain99]

The above problem can be addressed by setting a reasonable writeTimeout and keepalive while initialising the client. Upon writeTimeout, the connection is reset by calling internalConnLost and i/o timeout error reason is logged.

And if the regular publishes does not saturate the socket when the payload is small, then a reasonable keepalive can help detect the late disconnection.

So until writeTimeout and keepalive is correctly configured, we won't need the initially proposed solution of added writeTimeout separately for pings.

The connection may appear “alive” to keepalive pings even though specific publishes are never acknowledged.

Do we have any mechanism in place to address this problem? If not, how about implementing above proposed puback watchdog which resets the connection upon no ack within a reasonable time - ackTimeout.

[MattBrittan]

Thanks for your work on this, I wonder if the simplest solution might be to use the approach the V5 client uses and send the keepalive in a GoRoutine (it's not something that happens all that frequently so this should not have much of an impact on performance).

If not, how about implementing above proposed puback watchdog which resets the connection upon no ack within a reasonable time - ackTimeout.

This is certainly an option, but some care would be needed. Some random thoughts below:

• Starting off a monitoring routine for each message might be a bit expensive, an alternative approach might be to periodically chack MIDS (either just check the number of ID's in use, or store a timestamp for the last packet sent re that message).
• Care is needed when connection comes up, there may be a lot of queued messages and these can take time to process (see MaxResumePubInFlight for limit on this.
• writeTimeout should address the majority of issues but can be difficult to set (if bursts of messages are sent it could be expected that it will take a while to get all acknowledgements).
• The MQTT spec dictates that the broker MUST respond to these packets (e.g. "MUST send a PUBREL packet when it receives a PUBREC packet from the receiver. ") so a missing acknowledgement would be a breach of protocol.

[dhruvjain99]

Thanks for your work on this, I wonder if the simplest solution might be to use the approach the V5 client uses and send the keepalive in a GoRoutine (it's not something that happens all that frequently so this should not have much of an impact on performance).

This should be done to handle the blocking ping case when writeTimeout is indefinite.

For the ackWatchdog, it will be started upon a successful MQTT connection and will be a single goroutine instead of one for each outgoing control packet(expecting a response). If the response timeout then, it will disconnect. The idea is to run a watchdog goroutine similar to keepalive goroutine. The watchdog will compare if the fastReconnectCheckStartTime is greater than lastReceived. If true, then if currentTime - fastReconnectCheckStartTime >= ackTimeout it will trigger disconnection otherwise skip and wait for the next iteration.

fastReconnectCheckStartTime can be set by any outgoing control packet with an expected response and if lastReceived is greater than fastReconnectCheckStartTime.

https://github.com/gojek/paho.mqtt.golang/pull/3/files

Here is the example implementation ^^

The MQTT spec dictates that the broker MUST respond to these packets (e.g. "MUST send a PUBREL packet when it receives a PUBREC packet from the receiver. ") so a missing acknowledgement would be a breach of protocol.

Agree, but this ackWatchdog option serves the purpose of initiating a fast reconnection incase there is any anomaly observed in the connection/data exchange.

This is just an extra protection / care taken to prevent any half open and reset quickly. I know keepalive and writeTimeout if used correctly also does the same.

[dhruvjain99]

We would love to contribute the above changes on the eclipse paho upstream as well.