fix(transport): close transport under the send lock (#124)

* fix(transport): close transport under the send lock

Ktor's byte channels are single-writer: closing the socket while another
coroutine sits inside writeFully/flush mutates one kotlinx.io segment list
from two coroutines and corrupts it — Segment.compact "Check failed.", or an
NPE in the TLS writeRecord path as the cio-tls-closer coroutine flushes
close_notify through the same channel (KTOR-7729, open upstream).

Frame writes were already serialized, but every teardown path closed the
transport outside that lock, and the keepalive was a second writer racing the
close.

- Route all four teardown paths — disconnect(), abort(), handleFatalError, and
  the broker-initiated DISCONNECT — through one shutdownTransport(): take
  sendMutex, cancel and join the read loop and keepalive while holding it, then
  write any DISCONNECT and close, all in one lock acquisition.
- Bound the quiesce wait at 2s so a writer wedged on a dead peer cannot stall a
  reconnect; past that the close proceeds and the DISCONNECT is skipped.
- Guard TcpTransport.close() and WebSocketTransport.close() the same way for
  direct SPI users.
- Run teardown NonCancellable. handleFatalError cancelled the read loop before
  sending its DISCONNECT and closing, so when invoked from that loop the
  cancellation it had just requested aborted the rest of the teardown at the
  next suspension point, swallowed as best-effort, leaving the socket open.

TeardownRaceTest holds a write open via FakeTransport.sendGate and asserts each
path waits for it, and that teardown still closes when the writer never
completes.

Fixes #123

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(client): bound every wait in transport teardown

The quiesce wait was bounded, but two waits after it were not, so a dead peer
could still hold teardown open indefinitely and block a reconnect:

- The DISCONNECT write inside the lock. The same full socket buffer that blocks
  a publish blocks this write, and the close that frees the socket sat behind
  it. Abandoning the write does not reopen the race: this coroutine holds the
  lock, so the abandoned write and the close are sequential on one writer.
- The background-job joins. Cancelling a job does not interrupt a
  NonCancellable section, so a read loop already inside its own teardown runs
  that to completion first and stacks its budget onto ours. Giving up on the
  join is safe — sendMutex is held for the rest of the teardown, so a job that
  outlives the wait cannot reach the transport's writer either way.

Also reset sendsInFlight in FakeTransport.reset() so a reused instance does not
carry a stale counter into later assertions.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

Author: jamesarich

AGENTS.md

@@ -129,6 +129,23 @@ Build system: Kotlin DSL (`build.gradle.kts`) with version catalog (`gradle/libs
 
 Send PINGREQ every `keepAliveSeconds * 0.75` if no other packet was sent. If no PINGRESP within `keepAliveSeconds`, treat connection as dead and trigger reconnect (if enabled) or disconnect.
 
+### Single-writer discipline
+
+Ktor's byte channels are single-writer: two coroutines touching one channel corrupt its kotlinx.io
+segment list (`Segment.compact` "Check failed.", or an NPE in the TLS `writeRecord` path — see
+[KTOR-7729](https://youtrack.jetbrains.com/issue/KTOR-7729), open upstream). **Closing counts as
+writing** — `socket.close()` cancels that same channel and, under TLS, hands it to ktor's
+`cio-tls-closer` coroutine to flush close_notify.
+
+So `sendMutex` guards teardown as well as sends. `MqttConnection.shutdownTransport` is the only
+path that closes the transport: it takes `sendMutex`, cancels and joins the read loop and keepalive
+while holding it, then writes any DISCONNECT and closes — all under one lock acquisition, in a
+`NonCancellable` block because the loops it cancels are usually its own caller. Both transports
+guard `close()` the same way for direct SPI users. Every wait is bounded (2s) so a writer wedged on
+a dead peer cannot stall a reconnect.
+
+Anything new that writes to the transport, or closes it, has to join this discipline.
+
 ### TCP vs WebSocket framing
 
 - **TCP** (`TcpTransport.receive()`): Parse fixed header byte → decode variable-length remaining length → read exactly that many bytes. Handle partial reads correctly — this is the trickiest part.

CHANGELOG.md

@@ -7,6 +7,31 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- **Teardown no longer races an in-flight write on the transport's byte channel.** Closing the
+  socket while another coroutine sat inside `writeFully`/`flush` mutated one kotlinx.io segment
+  list from two coroutines and corrupted it — surfacing as `Segment.compact` "Check failed.", or a
+  `NullPointerException` inside ktor's TLS `writeRecord` as the `cio-tls-closer` coroutine flushed
+  close_notify through the same channel. Ktor's byte channels are single-writer and callers must
+  serialize (YouTrack [KTOR-7729](https://youtrack.jetbrains.com/issue/KTOR-7729), open upstream).
+  Frame writes were already serialized, but every teardown path — `disconnect()`, `abort()`, the
+  fatal-error handler, and a broker-initiated DISCONNECT — closed the transport outside that lock,
+  and the keepalive was a second writer racing the close. Teardown now takes the send lock first,
+  cancels and joins the read loop and keepalive while holding it (so the keepalive can only be
+  cancelled parked in `delay`, never mid-write), then writes any DISCONNECT and closes, all under
+  the one lock. `TcpTransport.close()` and `WebSocketTransport.close()` do the same for direct SPI
+  users. The wait is bounded at 2 seconds so a writer wedged on a dead peer cannot stall a
+  reconnect; past that the close proceeds regardless and the DISCONNECT is skipped.
+
+  Most visible under reconnect churn on flaky mobile networks with many concurrent publishers.
+  Fixes [#123](https://github.com/meshtastic/MQTTastic-Client-KMP/issues/123).
+
+- Fatal-error teardown could skip closing the transport. `handleFatalError` cancelled the read loop
+  before sending its DISCONNECT and closing, so when it was invoked *from* that read loop the very
+  cancellation it had just requested aborted the rest of the teardown at the next suspension point
+  and the exception was swallowed as best-effort. Teardown now runs non-cancellably.
+
 ## [0.8.0] - 2026-07-29
 
 ### Added

core/src/commonMain/kotlin/org/meshtastic/mqtt/MqttConnection.kt

@@ -22,6 +22,7 @@ import kotlinx.coroutines.Job
 import kotlinx.coroutines.NonCancellable
 import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
@@ -36,6 +37,7 @@ import kotlinx.coroutines.sync.Semaphore
 import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withContext
 import kotlinx.coroutines.withTimeout
+import kotlinx.coroutines.withTimeoutOrNull
 import kotlinx.io.bytestring.ByteString
 import org.meshtastic.mqtt.packet.Auth
 import org.meshtastic.mqtt.packet.ConnAck
@@ -357,7 +359,7 @@ internal class MqttConnection(
     /** Release the socket and reset state after a failed CONNECT/CONNACK handshake. */
     private suspend fun abandonHandshake() {
         try {
-            transport.close()
+            shutdownTransport()
         } catch (
             @Suppress("TooGenericExceptionCaught", "SwallowedException") _: Exception,
         ) {
@@ -421,14 +423,14 @@ internal class MqttConnection(
     suspend fun disconnect(reasonCode: ReasonCode = ReasonCode.SUCCESS) {
         log.debug(TAG) { "Disconnecting with reasonCode=$reasonCode" }
         failAllPendingAcks()
-        stopBackgroundJobs()
 
         try {
             if (transport.isConnected) {
                 // MQTT 3.1.1: DISCONNECT has no variable header/payload — always send empty
                 // MQTT 5.0: send with reason code and optional properties
-                sendPacket(Disconnect(reasonCode = reasonCode))
-                transport.close()
+                shutdownTransport(Disconnect(reasonCode = reasonCode))
+            } else {
+                shutdownTransport()
             }
         } finally {
             resetConnectionState()
@@ -445,12 +447,9 @@ internal class MqttConnection(
     suspend fun abort() {
         log.debug(TAG) { "Aborting connection (no DISCONNECT)" }
         failAllPendingAcks()
-        stopBackgroundJobs()
 
         try {
-            if (transport.isConnected) {
-                transport.close()
-            }
+            shutdownTransport()
         } finally {
             resetConnectionState()
         }
@@ -700,21 +699,30 @@ internal class MqttConnection(
 
     /** Send a packet over the transport, guarded by [sendMutex] for wire-level serialization. */
     private suspend fun sendPacket(packet: MqttPacket) {
-        sendMutex.withLock {
-            val bytes = packet.encode(version)
-
-            // Enforce broker's Maximum Packet Size on outbound packets (§3.2.2.3.6)
-            serverMaximumPacketSize?.let { maxSize ->
-                require(bytes.size <= maxSize) {
-                    "Outbound ${packet.packetType} (${bytes.size} bytes) exceeds server " +
-                        "Maximum Packet Size ($maxSize) (§3.2.2.3.6)"
-                }
-            }
+        sendMutex.withLock { sendPacketLocked(packet) }
+    }
 
-            log.trace(TAG) { "Sending ${packet.packetType} (${bytes.size} bytes)" }
-            transport.send(bytes)
-            lastSendMark = timeSource.markNow()
+    /**
+     * Write a packet to the transport. The caller must already hold [sendMutex].
+     *
+     * Only [sendPacket] and [shutdownTransport] call this: teardown sends its DISCONNECT
+     * inside the same lock acquisition that closes the socket, so nothing can slip onto the
+     * wire — or into the underlying byte channel — between the two.
+     */
+    private suspend fun sendPacketLocked(packet: MqttPacket) {
+        val bytes = packet.encode(version)
+
+        // Enforce broker's Maximum Packet Size on outbound packets (§3.2.2.3.6)
+        serverMaximumPacketSize?.let { maxSize ->
+            require(bytes.size <= maxSize) {
+                "Outbound ${packet.packetType} (${bytes.size} bytes) exceeds server " +
+                    "Maximum Packet Size ($maxSize) (§3.2.2.3.6)"
+            }
         }
+
+        log.trace(TAG) { "Sending ${packet.packetType} (${bytes.size} bytes)" }
+        transport.send(bytes)
+        lastSendMark = timeSource.markNow()
     }
 
     /** Register a pending ack deferred for the given packet ID, guarded by [acksMutex]. */
@@ -922,37 +930,19 @@ internal class MqttConnection(
             }
 
         log.error(TAG) { "Fatal error — tearing down connection" }
-        stopBackgroundJobs()
         try {
-            if (transport.isConnected) {
-                if (version.supportsProperties) {
-                    // §4.13: Send DISCONNECT with appropriate reason code before closing
-                    sendPacket(Disconnect(reasonCode = effectiveReasonCode))
-                }
-                // MQTT 3.1.1: no DISCONNECT with reason code — just close the transport
-            }
-        } catch (
-            @Suppress("SwallowedException") _: kotlin.coroutines.cancellation.CancellationException,
-        ) {
-            // Scope cancelled during best-effort DISCONNECT — continue cleanup
-        } catch (
-            @Suppress("TooGenericExceptionCaught", "SwallowedException") _: Exception,
-        ) {
-            // Best-effort DISCONNECT — transport may already be broken
-        }
-        try {
-            transport.close()
-        } catch (
-            @Suppress("SwallowedException") _: kotlin.coroutines.cancellation.CancellationException,
-        ) {
-            // Scope cancelled during best-effort close — continue cleanup
+            // §4.13: MQTT 5.0 announces the reason before closing; 3.1.1 has no DISCONNECT
+            // reason code, so it just closes.
+            shutdownTransport(
+                disconnect = if (version.supportsProperties) Disconnect(reasonCode = effectiveReasonCode) else null,
+            )
         } catch (
             @Suppress("TooGenericExceptionCaught", "SwallowedException") _: Exception,
         ) {
-            // Best-effort transport close
+            // Best-effort teardown — the transport may already be broken
         }
         // Use NonCancellable to ensure cleanup completes even when the calling job
-        // (read loop or keepalive) was cancelled by stopBackgroundJobs() above.
+        // (read loop or keepalive) was cancelled by shutdownTransport() above.
         withContext(NonCancellable) {
             failAllPendingAcks()
             val reason =
@@ -1071,8 +1061,7 @@ internal class MqttConnection(
                         ),
                     )
                 }
-                stopBackgroundJobs()
-                transport.close()
+                shutdownTransport()
                 failAllPendingAcks()
                 _connectionState.value =
                     ConnectionState.Disconnected(
@@ -1239,11 +1228,96 @@ internal class MqttConnection(
 
     // --- Private: Helpers ---
 
-    private fun stopBackgroundJobs() {
-        keepAliveJob?.cancel()
-        readLoopJob?.cancel()
+    /**
+     * Close the transport without ever overlapping an in-flight write.
+     *
+     * The transport's byte channel is single-writer. Ktor's is not merely undocumented on this
+     * point — closing a socket while another coroutine sits inside `writeFully`/`flush` mutates
+     * one kotlinx.io segment list from two coroutines and corrupts it, surfacing as
+     * `Segment.compact` "Check failed." or a NullPointerException deep inside the TLS record
+     * writer (YouTrack KTOR-7729, still open upstream). So teardown has to join the same
+     * single-writer discipline as [sendPacket] rather than run beside it.
+     *
+     * Order matters:
+     * 1. Take [sendMutex]. No writer is inside `transport.send` once it is held.
+     * 2. Cancel the read loop and keepalive *while holding it*, so the keepalive can only ever be
+     *    cancelled parked in `delay` — never midway through a write — and join them so a cancelled
+     *    writer cannot resume after the socket is gone.
+     * 3. Write [disconnect], if any, and close the transport, still under the lock.
+     *
+     * If a writer is wedged (a dead peer with a full socket buffer blocks `writeFully` until the
+     * TCP timeout), waiting for it forever would hang teardown and, with auto-reconnect, the
+     * client. So the wait is bounded by [WRITER_QUIESCE_TIMEOUT_MS]; past that the socket is closed
+     * regardless — which is also what unblocks the stuck writer — and the DISCONNECT is skipped,
+     * since writing it is exactly the unsafe act being avoided.
+     *
+     * Every other wait here is bounded for the same reason: the job joins by
+     * [JOB_JOIN_TIMEOUT_MS], and the DISCONNECT write itself, which the same dead peer would
+     * otherwise block indefinitely while holding teardown open.
+     *
+     * The body is [NonCancellable] because the read loop and keepalive both call this and it
+     * cancels them: on a plain context every suspension point after step 2 would abort and leak
+     * the socket.
+     */
+    private suspend fun shutdownTransport(disconnect: Disconnect? = null) {
+        // Captured before entering NonCancellable — inside it, the current Job is the
+        // withContext coroutine rather than the read-loop/keepalive job we must not join.
+        val callerJob = currentCoroutineContext()[Job]
+        withContext(NonCancellable) {
+            val quiesced =
+                withTimeoutOrNull(WRITER_QUIESCE_TIMEOUT_MS) {
+                    sendMutex.lock()
+                    true
+                } != null
+            if (!quiesced) {
+                log.warn(TAG) {
+                    "Writer still in flight after ${WRITER_QUIESCE_TIMEOUT_MS}ms — closing transport anyway"
+                }
+            }
+            try {
+                stopBackgroundJobs(callerJob)
+                if (disconnect != null && quiesced && transport.isConnected) {
+                    try {
+                        // Bounded: the same dead peer that blocks a publish blocks this write, and
+                        // the close that follows is what frees the socket. Abandoning the write
+                        // does not reopen the race — this coroutine holds the lock, so the
+                        // abandoned write and the close are sequential on one writer.
+                        withTimeoutOrNull(WRITER_QUIESCE_TIMEOUT_MS) { sendPacketLocked(disconnect) }
+                    } catch (
+                        @Suppress("TooGenericExceptionCaught", "SwallowedException") _: Exception,
+                    ) {
+                        // Best-effort DISCONNECT — the transport may already be broken
+                    }
+                }
+                transport.close()
+            } finally {
+                if (quiesced) sendMutex.unlock()
+            }
+        }
+    }
+
+    /**
+     * Cancel and join the read loop and keepalive jobs.
+     *
+     * [callerJob] is the job this teardown is running on, if any; it is cancelled but not joined,
+     * because the read loop and the keepalive both reach here through [handleFatalError] and a job
+     * cannot join itself.
+     */
+    private suspend fun stopBackgroundJobs(callerJob: Job?) {
+        val jobs = listOfNotNull(keepAliveJob, readLoopJob)
         keepAliveJob = null
         readLoopJob = null
+        jobs.forEach { it.cancel() }
+        jobs.filter { it !== callerJob }.forEach { job ->
+            // Bounded, because cancelling a job does not interrupt a NonCancellable section: a
+            // read loop already inside its own teardown runs that to completion first, and an
+            // unbounded join would stack its budget onto ours and delay a reconnect. Giving up
+            // on the join is safe — [sendMutex] is held for the rest of the teardown, so a job
+            // that outlives this wait cannot reach the transport's writer either way.
+            if (withTimeoutOrNull(JOB_JOIN_TIMEOUT_MS) { job.join() } == null) {
+                log.warn(TAG) { "Background job did not finish within ${JOB_JOIN_TIMEOUT_MS}ms — closing anyway" }
+            }
+        }
     }
 
     /** Build a CONNECT packet from [config] per §3.1. */
@@ -1425,6 +1499,23 @@ internal class MqttConnection(
         /** Timeout for waiting on acknowledgement packets (30 seconds). */
         const val ACK_TIMEOUT_MS = 30_000L
 
+        /**
+         * How long teardown waits for an in-flight write to finish before closing anyway.
+         *
+         * Long enough that a normal write completes first — that is what keeps the socket close
+         * off the byte channel's writer — and short enough that a writer wedged on a dead peer
+         * cannot stall a reconnect. See [shutdownTransport].
+         */
+        const val WRITER_QUIESCE_TIMEOUT_MS = 2_000L
+
+        /**
+         * How long teardown waits for a cancelled background job to finish.
+         *
+         * Cancellation does not interrupt a `NonCancellable` section, so a job already inside its
+         * own teardown finishes that first; without a bound its budget would stack onto ours.
+         */
+        const val JOB_JOIN_TIMEOUT_MS = 2_000L
+
         /** Keepalive factor: send PINGREQ at 75% of keep alive interval. */
         const val KEEPALIVE_FACTOR = 750 // keepAliveSeconds * 750 = milliseconds at 75%