security: fix send-on-closed-channel panic; remove unbounded buffer.

Before, sends into the event buffer were not performed under mutex lock, meaning that the channel could be closed while a send is in progress, causing a panic. A relay flooding events while a Close was in progress could trigger a panic.
Also, a slow consumer in addition to a flooding relay would cause the buffered pipe to grow unboundedly, which could cause memory to run out.
Both of these were resolved by removing the buffered pipe altogether, sending directly into the events channel, and sending under mutex.
The tradeoff we accept is that to secure the code against remote vulnerabilities and ensure no messages are dropped, the consumer must drain the subscription channels to prevent backpressure and deadlocks.

request.go

@@ -144,7 +144,6 @@ type request struct {
 	active         bool
 	firstEventSeen bool
 
-	buffer chan ReqEvent
 	events chan ReqEvent
 	closed chan ReqClosed
 
@@ -214,19 +213,12 @@ func (m *RequestManager) newStream(
 		}
 		id = fmt.Sprintf("%s:%s", label, generateID())
 	}
-	buffer := make(chan ReqEvent, 64)
+	events := make(chan ReqEvent, 64)
 	closed := make(chan ReqClosed, 1)
 
-	events := make(chan ReqEvent)
-	go func() {
-		bufferedPipe(buffer, events)
-		close(events)
-	}()
-
 	req := &request{
 		id:      id,
 		filters: filters,
-		buffer:  buffer,
 		events:  events,
 		closed:  closed,
 		isQuery: isQuery,
@@ -310,7 +302,7 @@ func (m *RequestManager) Cancel(id string) error {
 	}
 
 	req.deregisterOnce.Do(func() {
-		close(req.buffer)
+		close(req.events)
 		close(req.closed)
 	})
 	delete(m.reqs, id)
@@ -330,7 +322,7 @@ func (m *RequestManager) Close() {
 			go m.envoy.Send(envelope.EncloseClose(id))
 		}
 		req.deregisterOnce.Do(func() {
-			close(req.buffer)
+			close(req.events)
 			close(req.closed)
 		})
 		delete(m.reqs, id)
@@ -363,7 +355,7 @@ func (m *RequestManager) activate(req *request) {
 func (m *RequestManager) deregister(req *request) {
 	req.active = false
 	req.deregisterOnce.Do(func() {
-		close(req.buffer)
+		close(req.events)
 		close(req.closed)
 	})
 	delete(m.reqs, req.id)
@@ -424,17 +416,15 @@ func (m *RequestManager) dispatchInbox(msg InboxMessage) {
 	}
 }
 
-// routeEvent, routeEOSE, and routeClosed use blocking sends into req.buffer,
-// which reads eagerly into a slice buffer and cannot block the router.
 func (m *RequestManager) routeEvent(msg InboxMessage) {
 	subID, event, err := envelope.FindSubscriptionEvent(msg.Data)
 	if err != nil {
 		return
 	}
 	m.mu.Lock()
+	defer m.mu.Unlock()
 	req, ok := m.reqs[subID]
 	if !ok {
-		m.mu.Unlock()
 		return
 	}
 	if !req.firstEventSeen {
@@ -444,18 +434,14 @@ func (m *RequestManager) routeEvent(msg InboxMessage) {
 		go m.observer.Record(m.envoy.URL(),
 			FirstEventReceived{ID: reqSubID, ReceivedAt: receivedAt})
 	}
-	m.mu.Unlock()
 
-	req.buffer <- ReqEvent{
+	req.events <- ReqEvent{
 		Peer:       msg.URL,
 		ReceivedAt: msg.ReceivedAt,
 		Data:       event,
 	}
 }
 
-// routeEvent, routeEOSE, and routeClosed are always called sequentially from
-// the same routeInbox goroutine via dispatchInbox. This makes it safe for
-// routeEOSE to close req.buffer: no concurrent routeEvent send can race it.
 func (m *RequestManager) routeEOSE(msg InboxMessage) {
 	subID, err := envelope.FindEOSE(msg.Data)
 	if err != nil {
@@ -480,7 +466,7 @@ func (m *RequestManager) routeEOSE(msg InboxMessage) {
 		// manually cleanup query state
 		// specifically, do not close req.closed or events can be missed
 		req.active = false
-		close(req.buffer)
+		close(req.events)
 		delete(m.reqs, req.id)
 		if m.envoy.IsConnected() {
 			go func() {

util.go

@@ -1,34 +0,0 @@
-package prism
-
-func bufferedPipe[T any](input <-chan T, output chan<- T) {
-	var buffer []T
-
-	for {
-		var outOrNil chan<- T
-		var next T
-
-		// toggle send channel
-		if len(buffer) > 0 {
-			outOrNil = output
-			next = buffer[0]
-		} else if input == nil {
-			// input closed
-			return
-		}
-
-		select {
-		case item, ok := <-input:
-			if !ok {
-				// input is closed, set input nil
-				input = nil
-				continue
-			}
-			buffer = append(buffer, item)
-		case outOrNil <- next:
-			// discard element, set to zero to free memory
-			var zero T
-			buffer[0] = zero
-			buffer = buffer[1:]
-		}
-	}
-}