drop consumers if their out buffer fills up (#417)

right now if these buffers fill up we just start dropping events, and
then when the buffers drain we just resume sending events. this seems
quite wrong.

events/events.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"time"
 
 	comatproto "github.com/bluesky-social/indigo/api/atproto"
 	label "github.com/bluesky-social/indigo/api/label"
@@ -73,11 +74,20 @@ func (em *EventManager) broadcastEvent(evt *XRPCStreamEvent) {
 			select {
 			case s.outgoing <- evt:
 			case <-s.done:
+			default:
+				log.Warnw("dropping slow consumer due to event overflow", "bufferSize", len(s.outgoing), "ident", s.ident)
 				go func(torem *Subscriber) {
-					em.rmSubscriber(torem)
+					select {
+					case torem.outgoing <- &XRPCStreamEvent{
+						Error: &ErrorFrame{
+							Error: "ConsumerTooSlow",
+						},
+					}:
+					case <-time.After(time.Second * 5):
+						log.Warnw("failed to send error frame to backed up consumer", "ident", torem.ident)
+					}
+					torem.cleanup()
 				}(s)
-			default:
-				log.Warnf("event overflow (%d)", len(s.outgoing))
 			}
 			s.broadcastCounter.Inc()
 		}
@@ -100,6 +110,8 @@ type Subscriber struct {
 
 	done chan struct{}
 
+	cleanup func()
+
 	ident            string
 	enqueuedCounter  prometheus.Counter
 	broadcastCounter prometheus.Counter
@@ -164,15 +176,15 @@ func (em *EventManager) Subscribe(ctx context.Context, ident string, filter func
 		broadcastCounter: eventsBroadcast.WithLabelValues(ident),
 	}
 
-	cleanup := func() {
+	sub.cleanup = sync.OnceFunc(func() {
 		close(done)
 		em.rmSubscriber(sub)
 		close(sub.outgoing)
-	}
+	})
 
 	if since == nil {
 		em.addSubscriber(sub)
-		return sub.outgoing, cleanup, nil
+		return sub.outgoing, sub.cleanup, nil
 	}
 
 	out := make(chan *XRPCStreamEvent, em.bufferSize)
@@ -243,7 +255,7 @@ func (em *EventManager) Subscribe(ctx context.Context, ident string, filter func
 		}
 	}()
 
-	return out, cleanup, nil
+	return out, sub.cleanup, nil
 }
 
 func sequenceForEvent(evt *XRPCStreamEvent) int64 {