handshake: close accepted streams — port 444 leaked a goroutine per connection

handshake.go

@@ -528,6 +528,38 @@ func (hm *Manager) handleConnection(stream coreapi.Stream) {
 			slog.Error("handshake: recovered from panic in connection handler", "panic", r)
 		}
 	}()
+
+	// Close the stream on every exit path. Nothing closed it before — Close
+	// appeared nowhere in this file — so each accepted connection leaked:
+	//
+	//   * on the timeout branch below, the reader goroutine stays parked in
+	//     stream.Read forever, retaining its 64 KiB buf, the stream, and the
+	//     underlying daemon Connection (RecvBuf/SendBuf channels). Only a
+	//     Close can unblock it, and there wasn't one.
+	//   * on the success path the daemon-side Connection was never released
+	//     either.
+	//
+	// Port 444 accepts from any peer, so a peer that connects and sends
+	// nothing leaked ~64 KiB plus a goroutine stack and a full Connection
+	// every handshakeRecvTimeout, uncapped. That is the same shape as the
+	// dead idle-timeout that OOM-killed the service fleet.
+	//
+	// Closing here is what frees the parked reader: Close ends at
+	// CloseConnection, which closes RecvBuf, so the blocked receive returns
+	// !ok and the goroutine exits with io.EOF.
+	//
+	// The handshakeCloseDelay pause is why that constant exists — it was
+	// declared for exactly this ("delay before closing after send to let
+	// data flush") and had no remaining reference, stranded when the close
+	// path was removed. processMessage writes its reply synchronously, so
+	// without the pause we would close from under an in-flight response.
+	// Each handleConnection already runs in its own goroutine (see the
+	// accept loop), so this delays nothing but its own teardown.
+	defer func() {
+		time.Sleep(handshakeCloseDelay)
+		_ = stream.Close()
+	}()
+
 	const maxMsgSize = 64 * 1024
 
 	type readResult struct {

zz_ceiling_more_test.go

@@ -4,6 +4,7 @@ package handshake
 
 import (
 	"io"
+	"sync"
 	"testing"
 	"time"
 
@@ -194,7 +195,8 @@ func TestTrustedPeers_ReturnsAllEntries(t *testing.T) {
 // -----------------------------------------------------------------------
 
 type slowStream struct {
-	done chan struct{}
+	done      chan struct{}
+	closeOnce sync.Once
 }
 
 func newSlowStream() *slowStream { return &slowStream{done: make(chan struct{})} }
@@ -203,11 +205,24 @@ func (s *slowStream) Read(p []byte) (int, error) {
 	<-s.done
 	return 0, io.EOF
 }
-func (s *slowStream) Write(p []byte) (int, error)      { return len(p), nil }
-func (s *slowStream) Close() error                     { close(s.done); return nil }
-func (s *slowStream) LocalAddr() coreapi.Addr           { return coreapi.Addr{} }
+func (s *slowStream) Write(p []byte) (int, error) { return len(p), nil }
+
+// Close is idempotent, matching the real transport. The production
+// *Connection guards teardown with a sync.Once (CloseRecvBuf →
+// c.CloseOnce.Do), so calling Close twice is safe there. This double
+// stood in for it with a bare close(s.done), which panics on the second
+// call — so once handleConnection started closing the stream itself (as
+// it must, or every accepted connection leaks a goroutine), the test's
+// own `defer stream.Close()` became a second close and paniced.
+//
+// The bug was in the double's fidelity, not in closing twice.
+func (s *slowStream) Close() error {
+	s.closeOnce.Do(func() { close(s.done) })
+	return nil
+}
+func (s *slowStream) LocalAddr() coreapi.Addr          { return coreapi.Addr{} }
 func (s *slowStream) LocalPort() uint16                { return 0 }
-func (s *slowStream) RemoteAddr() coreapi.Addr          { return coreapi.Addr{} }
+func (s *slowStream) RemoteAddr() coreapi.Addr         { return coreapi.Addr{} }
 func (s *slowStream) RemotePort() uint16               { return 0 }
 func (s *slowStream) SetDeadline(time.Time) error      { return nil }
 func (s *slowStream) SetReadDeadline(time.Time) error  { return nil }
@@ -317,4 +332,3 @@ func TestSaveTrust_MkdirAllErrorIsNonFatal(t *testing.T) {
 	hm.mu.Unlock()
 	hm.saveTrust() // logs error, must not panic
 }
-