wamp,transport: add Peer.Done() for cooperative shutdown; idempotent Close

Adds a Done() <-chan struct{} method to the wamp.Peer interface.
Done is closed when Close is called, before the channel returned
by Send is closed. Senders that may race with Close can now select
on both Send and Done to detect peer closure cooperatively:

	select {
	case peer.Send() <- msg:
	case <-peer.Done():
	    return // peer is closing; abandon
	}

The previous contract was: bare `peer.Send() <- msg` racing with
Close panics with "send on closed channel". That hazard is now
documented in the Peer interface doc and an opt-in safe pattern
is provided. (Migrating call sites to the cooperative pattern is
a follow-up; the realm-shutdown path in router/realm.go already
serialises Close with sends via its actor goroutine, so the
panic vector there is structural rather than user-facing.)

Also makes Close idempotent on all three peer implementations
(localPeer, rawSocketPeer, websocketPeer) via sync.Once. Calling
Close more than once is now a safe no-op rather than a "close of
closed channel" panic. The previous implementations either had no
guard at all (localPeer, rawSocketPeer) or a racy
select-with-default early-out (websocketPeer).

Tests:

  - transport: TestLocalPeerDoneClosesOnClose,
    TestLocalPeerDoneIndependentPerSide,
    TestLocalPeerCloseIdempotent — pin the Done() and idempotency
    contract on localPeer.
  - transport: existing TestCloseWebsocketPeer continues to verify
    websocketPeer Close idempotency (now genuinely safe via
    sync.Once instead of the previous racy select).

Test scaffolding (testPeer in router/broker_test.go and
wamp/peer_test.go) updated to satisfy the new interface method.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Marko Petzold

router/broker_test.go

@@ -11,17 +11,20 @@ import (
 )
 
 type testPeer struct {
-	in chan wamp.Message
+	in   chan wamp.Message
+	done chan struct{}
 }
 
 func newTestPeer() *testPeer {
 	return &testPeer{
-		in: make(chan wamp.Message, 1),
+		in:   make(chan wamp.Message, 1),
+		done: make(chan struct{}),
 	}
 }
 
 func (p *testPeer) Recv() <-chan wamp.Message { return p.in }
 func (p *testPeer) Send() chan<- wamp.Message { return p.in }
+func (p *testPeer) Done() <-chan struct{}     { return p.done }
 func (p *testPeer) Close()                    {}
 
 func (p *testPeer) IsLocal() bool { return true }

transport/localpeer.go

@@ -1,6 +1,8 @@
 package transport
 
 import (
+	"sync"
+
 	"github.com/gammazero/nexus/v3/wamp"
 )
 
@@ -32,9 +34,9 @@ func LinkedPeersQSize(queueSize int) (wamp.Peer, wamp.Peer) {
 	cToR := make(chan wamp.Message)
 
 	// router reads from and writes to client
-	r := &localPeer{rd: cToR, wr: rToC}
+	r := &localPeer{rd: cToR, wr: rToC, done: make(chan struct{})}
 	// client reads from and writes to router
-	c := &localPeer{rd: rToC, wr: cToR}
+	c := &localPeer{rd: rToC, wr: cToR, done: make(chan struct{})}
 
 	return c, r
 }
@@ -43,6 +45,9 @@ func LinkedPeersQSize(queueSize int) (wamp.Peer, wamp.Peer) {
 type localPeer struct {
 	rd <-chan wamp.Message
 	wr chan<- wamp.Message
+
+	done      chan struct{}
+	closeOnce sync.Once
 }
 
 // IsLocal returns true is the wamp.Peer is a localPeer.
@@ -54,6 +59,19 @@ func (p *localPeer) Recv() <-chan wamp.Message { return p.rd }
 // Send returns the peer's outbound message channel.
 func (p *localPeer) Send() chan<- wamp.Message { return p.wr }
 
-// Close closes the outgoing channel, waking any readers waiting on data from
-// this peer.
-func (p *localPeer) Close() { close(p.wr) }
+// Done returns a channel that is closed when the peer is closing. It
+// is closed before the Send channel, so a goroutine selecting on both
+// Send and Done is guaranteed to wake via Done before the runtime
+// observes the Send channel as closed.
+func (p *localPeer) Done() <-chan struct{} { return p.done }
+
+// Close closes the outgoing channel, waking any readers waiting on
+// data from this peer. Idempotent and safe to call concurrently — the
+// first invocation closes Done first, then the outbound channel; later
+// invocations are no-ops.
+func (p *localPeer) Close() {
+	p.closeOnce.Do(func() {
+		close(p.done)
+		close(p.wr)
+	})
+}

transport/localpeer_test.go

@@ -88,6 +88,70 @@ func TestBlockOnBlockedRouter(t *testing.T) {
 	<-done
 }
 
+// TestLocalPeerDoneClosesOnClose verifies the Peer.Done() contract:
+// Close fires the Done channel before unwinding the rest of the
+// shutdown sequence, so consumers selecting on Done observe peer
+// closure cooperatively.
+func TestLocalPeerDoneClosesOnClose(t *testing.T) {
+	c, _ := transport.LinkedPeers()
+
+	// Done starts open.
+	select {
+	case <-c.Done():
+		require.FailNow(t, "Done must not be closed before Close")
+	default:
+	}
+
+	c.Close()
+
+	// Done is closed after Close.
+	select {
+	case <-c.Done():
+	case <-time.After(time.Second):
+		require.FailNow(t, "Done must be closed after Close")
+	}
+}
+
+// TestLocalPeerDoneIndependentPerSide verifies the two peers in a
+// LinkedPeers pair have independent Done signals — closing one does
+// not signal the other.
+func TestLocalPeerDoneIndependentPerSide(t *testing.T) {
+	c, r := transport.LinkedPeers()
+	c.Close()
+
+	select {
+	case <-c.Done():
+	default:
+		require.FailNow(t, "client peer Done must be closed after client Close")
+	}
+
+	select {
+	case <-r.Done():
+		require.FailNow(t, "router peer Done must not be closed by client Close")
+	default:
+	}
+
+	r.Close()
+	select {
+	case <-r.Done():
+	default:
+		require.FailNow(t, "router peer Done must be closed after router Close")
+	}
+}
+
+// TestLocalPeerCloseIdempotent verifies that calling Close more than
+// once is a safe no-op rather than a "close of closed channel" panic.
+// The realm shutdown ordering relies on this: a session whose handler
+// closed its peer naturally may also be in the snapshot of
+// pending-to-close peers that the realm walks at the end of close().
+func TestLocalPeerCloseIdempotent(t *testing.T) {
+	c, r := transport.LinkedPeers()
+	c.Close()
+	require.NotPanics(t, func() { c.Close() }, "second Close on client peer must not panic")
+	require.NotPanics(t, func() { r.Close() }, "Close on router peer must not panic")
+	require.NotPanics(t, func() { r.Close() }, "second Close on router peer must not panic")
+}
+
 func BenchmarkClientToRouter(b *testing.B) {
 	c, r := transport.LinkedPeers()
 

transport/rawsocketpeer.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"net"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/gammazero/nexus/v3/stdlog"
@@ -35,6 +36,8 @@ type rawSocketPeer struct {
 
 	writerDone chan struct{}
 
+	closeOnce sync.Once
+
 	log stdlog.StdLog
 }
 
@@ -174,27 +177,39 @@ func (rs *rawSocketPeer) Recv() <-chan wamp.Message { return rs.rd }
 
 func (rs *rawSocketPeer) Send() chan<- wamp.Message { return rs.wr }
 
+// Done returns a channel that is closed when the peer is closing. It
+// is closed before the Send channel, so a goroutine selecting on both
+// Send and Done is guaranteed to wake via Done before the runtime
+// observes the Send channel as closed.
+func (rs *rawSocketPeer) Done() <-chan struct{} { return rs.closed }
+
 func (rs *rawSocketPeer) IsLocal() bool { return false }
 
-// Close closes the rawsocket peer. This closes the local send channel, and
-// sends a close control message to the socket to tell the other side to close.
+// Close closes the rawsocket peer. Idempotent and safe to call
+// concurrently — the first invocation closes Done first, then unwinds
+// sendHandler/recvHandler/socket; later invocations are no-ops.
 //
 // *** Do not call Send after calling Close. ***
 func (rs *rawSocketPeer) Close() {
-	// Tell sendHandler to exit, and discard any queued messages. Do not close
-	// wr channel in case there are incoming messages during close.
-	rs.cancelSender()
-	<-rs.writerDone
-	close(rs.wr)
-	for range rs.wr {
-	}
-
-	// Tell recvHandler to close.
-	close(rs.closed)
+	rs.closeOnce.Do(func() {
+		// Close the Done channel first so any Send goroutine that
+		// selected on it can wake and abandon before the wr channel
+		// is closed.
+		close(rs.closed)
+
+		// Tell sendHandler to exit, and discard any queued messages.
+		// Do not close wr channel in case there are incoming messages
+		// during close.
+		rs.cancelSender()
+		<-rs.writerDone
+		close(rs.wr)
+		for range rs.wr {
+		}
 
-	// Ignore errors since socket may have been closed by other side first in
-	// response to a goodbye message.
-	_ = rs.conn.Close()
+		// Ignore errors since socket may have been closed by other
+		// side first in response to a goodbye message.
+		_ = rs.conn.Close()
+	})
 }
 
 // sendHandler pulls messages from the write channel, and pushes them to the

transport/websocketpeer.go

@@ -7,6 +7,7 @@ import (
 	"net"
 	"net/http"
 	"net/url"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -92,6 +93,8 @@ type websocketPeer struct {
 	recvDone   chan struct{}
 	writerDone chan struct{}
 
+	closeOnce sync.Once
+
 	log stdlog.StdLog
 }
 
@@ -228,40 +231,45 @@ func (w *websocketPeer) Send() chan<- wamp.Message { return w.wr }
 
 func (w *websocketPeer) IsLocal() bool { return false }
 
-// Close closes the websocket peer. This closes the local send channel, and
-// sends a close control message to the websocket to tell the other side to
-// close.
+// Close closes the websocket peer. Idempotent and safe to call
+// concurrently — the first invocation closes Done first, then unwinds
+// sendHandler/recvHandler/socket; later invocations are no-ops.
 //
 // *** Do not call Send after calling Close. ***
 func (w *websocketPeer) Close() {
-	select {
-	case <-w.closed:
-		return
-	default:
-	}
-
-	// Tell sendHandler to exit and discard any queued messages. Do not close
-	// wr channel in case there are incoming messages during close.
-	w.cancelSender()
-	<-w.writerDone
-	close(w.wr)
-	for range w.wr {
-	}
-
-	closeMsg := websocket.FormatCloseMessage(websocket.CloseNormalClosure, "goodbye")
+	w.closeOnce.Do(func() {
+		// Close the Done channel first so any Send goroutine that
+		// selected on it can wake and abandon before the wr channel
+		// is closed.
+		close(w.closed)
+
+		// Tell sendHandler to exit and discard any queued messages.
+		// Do not close wr channel in case there are incoming messages
+		// during close.
+		w.cancelSender()
+		<-w.writerDone
+		close(w.wr)
+		for range w.wr {
+		}
 
-	// Tell recvHandler to close.
-	close(w.closed)
+		closeMsg := websocket.FormatCloseMessage(websocket.CloseNormalClosure, "goodbye")
 
-	// Ignore errors since websocket may have been closed by other side first
-	// in response to a goodbye message.
-	_ = w.conn.WriteControl(websocket.CloseMessage, closeMsg, time.Now().Add(ctrlTimeout))
-	_ = w.conn.Close()
+		// Ignore errors since websocket may have been closed by other
+		// side first in response to a goodbye message.
+		_ = w.conn.WriteControl(websocket.CloseMessage, closeMsg, time.Now().Add(ctrlTimeout))
+		_ = w.conn.Close()
 
-	// Wait for the recvHandler goroutine to exit.
-	<-w.recvDone
+		// Wait for the recvHandler goroutine to exit.
+		<-w.recvDone
+	})
 }
 
+// Done returns a channel that is closed when the peer is closing. It
+// is closed before the Send channel, so a goroutine selecting on both
+// Send and Done is guaranteed to wake via Done before the runtime
+// observes the Send channel as closed.
+func (w *websocketPeer) Done() <-chan struct{} { return w.closed }
+
 // sendHandler pulls messages from the write channel, and pushes them to the
 // websocket.
 func (w *websocketPeer) sendHandler() {

wamp/peer.go

@@ -6,18 +6,65 @@ import (
 )
 
 // Peer is the interface implemented by endpoints communicating via WAMP.
+//
+// # Lifecycle and concurrent shutdown
+//
+// Close is idempotent and safe to call concurrently with itself. It is
+// NOT fully safe to call concurrently with Send for callers that use
+// the bare-channel form `peer.Send() <- msg`: the underlying outbound
+// channel must be closed at some point so the partner peer's Recv
+// loop can terminate, and a parked send racing that close panics
+// regardless of any synchronisation the implementation does internally
+// — that is a Go runtime constraint, not an implementation detail.
+//
+// Done() exists to make the cooperative pattern available:
+//
+//	select {
+//	case peer.Send() <- msg:
+//	case <-peer.Done():
+//	    return // peer is closing; abandon
+//	}
+//
+// Implementations close Done before closing the Send channel, so a
+// goroutine entering this select after Close has begun observes Done
+// ready and exits cleanly without ever attempting the send. A goroutine
+// already parked in this select when Close starts wakes via Done in
+// the typical case — but if the partner peer's Recv buffer happens
+// to have room at the same instant, the runtime may still pick the
+// Send case; callers that need ironclad safety should arrange for
+// their Send goroutines to exit before calling Close (e.g. via a
+// goroutine lifecycle wait group). This is the same pattern the
+// router's realm-shutdown ordering already follows internally.
+//
+// Bare `peer.Send() <- msg` (no select) racing with Close panics, by
+// design — fixing it would require either an extra forwarder
+// goroutine per peer (with its own goroutine-leak hazard) or an API
+// change to a method-form Send that takes a mutex. Both have higher
+// architectural cost than the documented hazard. Callers that may
+// race with Close should always use the cooperative pattern above.
 type Peer interface {
-	// Closes the peer connection and the channel returned from Recv().
+	// Close closes the peer connection and the channel returned from
+	// Recv(). Close is idempotent; calling it more than once is a
+	// no-op. Done is closed before the Send channel is closed,
+	// see the cooperative shutdown pattern in the package doc.
 	Close()
 
 	// IsLocal returns true if the session is local.
 	IsLocal() bool
 
-	// Recv returns a channel of messages from the peer.
+	// Recv returns a channel of messages from the peer. The channel
+	// is closed when the remote side closes its connection, or when
+	// Close is called on this peer.
 	Recv() <-chan Message
 
-	// Send returns the peer's outgoing message channel.
+	// Send returns the peer's outgoing message channel. See the
+	// package doc for the cooperative-shutdown pattern that makes
+	// Send safe to use concurrently with Close.
 	Send() chan<- Message
+
+	// Done returns a channel that is closed when Close is called.
+	// Senders use it to detect peer closure cooperatively.
+	Done() <-chan struct{}
 }
 
 // RecvTimeout receives a message from a peer within the specified time.