diff --git a/handshake.go b/handshake.go index 831646b..bd80469 100644 --- a/handshake.go +++ b/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 { diff --git a/zz_ceiling_more_test.go b/zz_ceiling_more_test.go index bdce072..5368cbf 100644 --- a/zz_ceiling_more_test.go +++ b/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 } -