From 3d20e9c5123baee58091ab3d6a8f014aad7c68c7 Mon Sep 17 00:00:00 2001
From: Jay <jay@wisehodl.dev>
Date: Sun, 9 Aug 2026 18:54:35 -0400
Subject: [PATCH] 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 | 28 +++++++---------------------
 util.go    | 34 ----------------------------------
 2 files changed, 7 insertions(+), 55 deletions(-)
 delete mode 100644 util.go

diff --git a/request.go b/request.go
index bf621ff..1edcfbb 100644
--- a/request.go
+++ b/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() {
diff --git a/util.go b/util.go
deleted file mode 100644
index dc891c7..0000000
--- a/util.go
+++ /dev/null
@@ -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:]
-		}
-	}
-}