Non-blocking subscriber teardown

Author: jazware

pkg/server/server.go

@@ -296,7 +296,7 @@ func (s *Server) Emit(ctx context.Context, e *models.Event, asJSON, compBytes []
 			defer sub.lk.Unlock()
 
 			// Don't emit events to subscribers that are replaying and are too far behind
-			if sub.cursor != nil && sub.seq < e.TimeUS-cutoverThresholdUS {
+			if sub.cursor != nil && sub.seq < e.TimeUS-cutoverThresholdUS || sub.tearingDown {
 				return
 			}
 

pkg/server/subscriber.go

@@ -20,14 +20,15 @@ type WantedCollections struct {
 }
 
 type Subscriber struct {
-	ws     *websocket.Conn
-	conLk  sync.Mutex
-	realIP string
-	lk     sync.Mutex
-	seq    int64
-	outbox chan *[]byte
-	hello  chan struct{}
-	id     int64
+	ws          *websocket.Conn
+	conLk       sync.Mutex
+	realIP      string
+	lk          sync.Mutex
+	seq         int64
+	outbox      chan *[]byte
+	hello       chan struct{}
+	id          int64
+	tearingDown bool
 
 	// Subscriber options
 
@@ -99,11 +100,16 @@ func emitToSubscriber(ctx context.Context, log *slog.Logger, sub *Subscriber, ti
 		default:
 			// Drop slow subscribers if they're live tailing and fall too far behind
 			log.Error("failed to send event to subscriber, dropping", "error", "buffer full", "subscriber", sub.id)
-			sub.Terminate("consuming too slowly")
-			err := sub.ws.Close()
-			if err != nil {
-				log.Error("failed to close subscriber connection", "error", err)
-			}
+
+			// Tearing down a subscriber can block, so do it in a goroutine
+			go func() {
+				sub.tearingDown = true
+				sub.Terminate("consuming too slowly")
+				err := sub.ws.Close()
+				if err != nil {
+					log.Error("failed to close subscriber connection", "error", err)
+				}
+			}()
 		}
 	}