Bound a blocking send, which MSG_DONTWAIT does not (#291)

GraphcodeKit/Sources/IPC/OutboundChannel.swift

@@ -74,6 +74,7 @@ final class OutboundChannel: @unchecked Sendable {
     isSocket =
       getsockopt(fileDescriptor, SOL_SOCKET, SO_TYPE, &socketType, &typeSize) == 0
     Self.armAgainstSIGPIPE(fileDescriptor)
+    Self.boundBlockingSends(on: fileDescriptor)
     // Captured strongly on purpose: the channel must outlive the registry's reference to
     // it, because the descriptor is closed by `pump` on its way out. A weak capture would
     // let a channel released without `close()` take its thread — and its unclosed
@@ -104,6 +105,27 @@ final class OutboundChannel: @unchecked Sendable {
     #endif
   }
 
+  /// Bounds how long one `send(2)` may sit inside the kernel before it comes back with
+  /// whatever it managed to write.
+  ///
+  /// `MSG_DONTWAIT` is not enough on its own, which is the correction this exists to
+  /// make. On macOS that flag does **nothing** for `send` on a blocking AF_UNIX stream
+  /// socket: measured here, 60 KB into a peer with a 4 KB buffer was still inside `send`
+  /// after 120 seconds. So the poll loop below never ran, and the writer parked in the
+  /// syscall exactly as it had before — the actor stayed safe, because that is the
+  /// thread's doing, but `closeAndWait` was not bounded the way its comment promised.
+  ///
+  /// `SO_SNDTIMEO` is the half that works. It applies only to sending, so the reader —
+  /// which shares this open file description — is untouched, unlike `O_NONBLOCK`. With
+  /// it a full peer returns a short count after one timeout rather than never, and the
+  /// loop below does what it always claimed to.
+  private static func boundBlockingSends(on fileDescriptor: Int32) {
+    var timeout = timeval(
+      tv_sec: 0, tv_usec: suseconds_t(Int(writabilityPollMilliseconds) * 1000))
+    setsockopt(
+      fileDescriptor, SOL_SOCKET, SO_SNDTIMEO, &timeout, socklen_t(MemoryLayout<timeval>.size))
+  }
+
   /// Queues a frame and returns immediately.
   ///
   /// `supersedingKey` replaces any still-undelivered frame carrying the same key rather
@@ -254,18 +276,20 @@ final class OutboundChannel: @unchecked Sendable {
   /// Writes one length-prefixed frame, returning false if the peer failed or the channel
   /// was closed part-way through.
   ///
-  /// Every write is non-blocking *per call* (`MSG_DONTWAIT`), waiting for writability
-  /// with a bounded `poll` rather than parking inside `write(2)` until the peer drains.
-  /// That is not a refinement, it is what lets `closeAndWait` promise to return: a thread
-  /// already blocked inside a write on a unix socket is **not** reliably woken by another
-  /// thread's `shutdown`, so a wedged peer could hang the disconnecting caller for ever —
-  /// the original bug moved one layer down, where it showed up as the full test suite
-  /// hanging on a channel whose client never read. Polling in slices lets the writer
-  /// notice `isClosing` by itself instead of depending on being interrupted.
+  /// No single `send` may park here indefinitely: `SO_SNDTIMEO` (set in `init`) bounds it,
+  /// and a short count is followed by a bounded `poll` for writability. That is what lets
+  /// `closeAndWait` promise to return — a thread already blocked inside a send on a unix
+  /// socket is **not** reliably woken by another thread's `shutdown`, so a wedged peer
+  /// could otherwise hang the disconnecting caller for ever, which is how this showed up
+  /// the first time: the full test suite hanging on a channel whose client never read.
+  /// Coming back every slice lets the writer notice `isClosing` by itself instead of
+  /// depending on being interrupted.
   ///
-  /// `MSG_DONTWAIT` per call rather than `O_NONBLOCK` on the descriptor: that flag lives
-  /// on the open file description, which the daemon's *reader* shares, and a reader that
-  /// started returning `EAGAIN` would tear down every connection.
+  /// `MSG_DONTWAIT` is passed too, but do not rely on it: on macOS it does nothing for
+  /// `send` on a blocking unix socket (see `boundBlockingSends`). It is kept because it
+  /// is honoured on Linux and costs nothing where it is not. Neither is `O_NONBLOCK` on
+  /// the descriptor, which would change the *reader* that shares this open file
+  /// description — a reader returning `EAGAIN` would tear down every connection.
   private func writeFrame(_ data: Data) -> Bool {
     let length = UInt32(data.count)
     var buffer = Data(capacity: 4 + data.count)

graphcode/Tests/OutboundChannelTests.swift

@@ -130,12 +130,11 @@ struct OutboundChannelTests {
     #expect(received.contains(reply))
   }
 
-  /// Closing must not wait on a peer that has stopped reading. This is why the writer
-  /// never parks inside a blocking `write(2)`: a thread already blocked there is not
-  /// reliably woken by another thread's `shutdown`, so a wedged peer could hang the
-  /// disconnecting caller for ever — the original bug moved one layer down, from the
-  /// actor to the connection loop. Writing in non-blocking slices lets the writer notice
-  /// the close by itself.
+  /// Closing must not wait on a peer that has stopped reading. This is why no single send
+  /// may park indefinitely: a thread already blocked inside one is not reliably woken by
+  /// another thread's `shutdown`, so a wedged peer could hang the disconnecting caller for
+  /// ever — the original bug moved one layer down, from the actor to the connection loop.
+  /// Coming back every `SO_SNDTIMEO` slice lets the writer notice the close by itself.
   @Test
   func closingReleasesAWriterParkedOnAWedgedClient() {
     let (daemon, client) = makeSocketPair()
@@ -153,6 +152,31 @@ struct OutboundChannelTests {
     #expect(elapsed < 2.0, "closing a wedged channel took \(elapsed)s")
   }
 
+  /// The correction to #291. `MSG_DONTWAIT` does nothing for `send` on a blocking
+  /// AF_UNIX stream socket on macOS — measured at 60 KB into a 4 KB peer, still inside
+  /// the syscall after two minutes — so the writer parked there and the poll loop that
+  /// `closeAndWait` depends on never ran. `SO_SNDTIMEO` is what actually bounds it, and
+  /// it has to be on the descriptor before anything is written.
+  @Test
+  func theChannelBoundsHowLongOneSendMayPark() {
+    let (daemon, client) = makeSocketPair()
+    OutboundChannels.open(daemon)
+    defer {
+      OutboundChannels.close(daemon)
+      close(client)
+    }
+
+    var timeout = timeval(tv_sec: 0, tv_usec: 0)
+    var size = socklen_t(MemoryLayout<timeval>.size)
+    #expect(getsockopt(daemon, SOL_SOCKET, SO_SNDTIMEO, &timeout, &size) == 0)
+
+    let microseconds = Int(timeout.tv_sec) * 1_000_000 + Int(timeout.tv_usec)
+    #expect(microseconds > 0, "a send on this channel is unbounded")
+    // Bounded, and short enough that a close is noticed promptly rather than a slice
+    // later — anything approaching a second would put the teardown back where it was.
+    #expect(microseconds <= 200_000)
+  }
+
   /// The safety valve. With superseding in play a backlog is normally one snapshot, so
   /// reaching the budget means a peer stopped reading and stayed stopped across many
   /// distinct frames. Dropping it is the honest outcome, and `send` says so by returning