fix: clear subscriberConn in runMux defer to prevent stale reference and msgType refactor

Author: nugaon

pkg/pubsub/mode.go

@@ -375,8 +375,14 @@ func (m *GSOCEphemeralMode) CreateSubscriberConn(stream p2p.Stream, overlay swar
 
 // runMux reads broker messages from the shared p2p stream and broadcasts each to all
 // registered WS sessions. It exits when the stream closes or returns an error.
+// On exit it immediately clears m.subscriberConn so new Connect calls open a fresh stream.
 func (m *GSOCEphemeralMode) runMux(stream p2p.Stream) {
-	defer m.subscriberConn.closeAll()
+	defer func() {
+		m.subscriberConn.closeAll()
+		m.mu.Lock()
+		m.subscriberConn = nil
+		m.mu.Unlock()
+	}()
 	for {
 		msg, err := m.ReadBrokerMessage(stream)
 		if err != nil {
@@ -397,14 +403,13 @@ func (m *GSOCEphemeralMode) GetSubscriberConn() *SubscriberConn {
 	return m.subscriberConn
 }
 
-// RemoveSubscriberConn decrements the ref count for the connection.
-// When the last WS session exits, it closes the stream, stopping the mux goroutine.
+// RemoveSubscriberConn decrements the ref count for conn.
+// When the last WS session exits it closes the stream, stopping the mux goroutine.
+// If the mux already died and cleared m.subscriberConn, refs are still tracked on conn
+// so the stream is closed exactly once when refs reach zero.
 func (m *GSOCEphemeralMode) RemoveSubscriberConn(conn *SubscriberConn) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
-	if m.subscriberConn != conn {
-		return
-	}
 	conn.refs--
 	if conn.refs <= 0 {
 		m.subscriberConn = nil

pkg/pubsub/pubsub.go

@@ -34,7 +34,8 @@ const (
 	ModeGSOCEphemeral ModeID = 1
 
 	// Service-level broker message types.
-	MsgTypePing byte = 0x03
+	// 0x01 is reserved for ping across all modes; mode-specific types start at 0x02.
+	MsgTypePing byte = 0x01
 
 	// streamPingInterval is how often the broker sends a keepalive ping to each subscriber.
 	streamPingInterval = 30 * time.Second