ethrpc: Buffer sub messages if subscriber is slow

node/impl/full/eth.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/zyedidia/generic/queue"
 	"sort"
 	"strconv"
 	"sync"
@@ -1352,7 +1353,7 @@ func (e *EthEvent) EthSubscribe(ctx context.Context, p jsonrpc.RawParams) (ethty
 		return ethtypes.EthSubscriptionID{}, xerrors.Errorf("connection doesn't support callbacks")
 	}
 
-	sub, err := e.SubManager.StartSubscription(e.SubscribtionCtx, ethCb.EthSubscription)
+	sub, err := e.SubManager.StartSubscription(e.SubscribtionCtx, ethCb.EthSubscription, e.uninstallFilter)
 	if err != nil {
 		return ethtypes.EthSubscriptionID{}, err
 	}
@@ -1418,18 +1419,11 @@ func (e *EthEvent) EthUnsubscribe(ctx context.Context, id ethtypes.EthSubscripti
 		return false, api.ErrNotSupported
 	}
 
-	filters, err := e.SubManager.StopSubscription(ctx, id)
+	err := e.SubManager.StopSubscription(ctx, id)
 	if err != nil {
 		return false, nil
 	}
 
-	for _, f := range filters {
-		if err := e.uninstallFilter(ctx, f); err != nil {
-			// this will leave the filter a zombie, collecting events up to the maximum allowed
-			log.Warnf("failed to remove filter when unsubscribing: %v", err)
-		}
-	}
-
 	return true, nil
 }
 
@@ -1615,7 +1609,7 @@ type EthSubscriptionManager struct {
 	subs     map[ethtypes.EthSubscriptionID]*ethSubscription
 }
 
-func (e *EthSubscriptionManager) StartSubscription(ctx context.Context, out ethSubscriptionCallback) (*ethSubscription, error) { // nolint
+func (e *EthSubscriptionManager) StartSubscription(ctx context.Context, out ethSubscriptionCallback, dropFilter func(context.Context, filter.Filter) error) (*ethSubscription, error) { // nolint
 	rawid, err := uuid.NewRandom()
 	if err != nil {
 		return nil, xerrors.Errorf("new uuid: %w", err)
@@ -1626,13 +1620,17 @@ func (e *EthSubscriptionManager) StartSubscription(ctx context.Context, out ethS
 	ctx, quit := context.WithCancel(ctx)
 
 	sub := &ethSubscription{
-		Chain:    e.Chain,
-		StateAPI: e.StateAPI,
-		ChainAPI: e.ChainAPI,
-		id:       id,
-		in:       make(chan interface{}, 200),
-		out:      out,
-		quit:     quit,
+		Chain:           e.Chain,
+		StateAPI:        e.StateAPI,
+		ChainAPI:        e.ChainAPI,
+		uninstallFilter: dropFilter,
+		id:              id,
+		in:              make(chan interface{}, 200),
+		out:             out,
+		quit:            quit,
+
+		toSend:   queue.New[[]byte](),
+		sendCond: make(chan struct{}, 1),
 	}
 
 	e.mu.Lock()
@@ -1643,37 +1641,46 @@ func (e *EthSubscriptionManager) StartSubscription(ctx context.Context, out ethS
 	e.mu.Unlock()
 
 	go sub.start(ctx)
+	go sub.startOut(ctx)
 
 	return sub, nil
 }
 
-func (e *EthSubscriptionManager) StopSubscription(ctx context.Context, id ethtypes.EthSubscriptionID) ([]filter.Filter, error) {
+func (e *EthSubscriptionManager) StopSubscription(ctx context.Context, id ethtypes.EthSubscriptionID) error {
 	e.mu.Lock()
 	defer e.mu.Unlock()
 
 	sub, ok := e.subs[id]
 	if !ok {
-		return nil, xerrors.Errorf("subscription not found")
+		return xerrors.Errorf("subscription not found")
 	}
 	sub.stop()
 	delete(e.subs, id)
 
-	return sub.filters, nil
+	return nil
 }
 
 type ethSubscriptionCallback func(context.Context, jsonrpc.RawParams) error
 
+const maxSendQueue = 20000
+
 type ethSubscription struct {
-	Chain    *store.ChainStore
-	StateAPI StateAPI
-	ChainAPI ChainAPI
-	id       ethtypes.EthSubscriptionID
-	in       chan interface{}
-	out      ethSubscriptionCallback
+	Chain           *store.ChainStore
+	StateAPI        StateAPI
+	ChainAPI        ChainAPI
+	uninstallFilter func(context.Context, filter.Filter) error
+	id              ethtypes.EthSubscriptionID
+	in              chan interface{}
+	out             ethSubscriptionCallback
 
 	mu      sync.Mutex
 	filters []filter.Filter
 	quit    func()
+
+	sendLk       sync.Mutex
+	sendQueueLen int
+	toSend       *queue.Queue[[]byte]
+	sendCond     chan struct{}
 }
 
 func (e *ethSubscription) addFilter(ctx context.Context, f filter.Filter) {
@@ -1684,6 +1691,36 @@ func (e *ethSubscription) addFilter(ctx context.Context, f filter.Filter) {
 	e.filters = append(e.filters, f)
 }
 
+// sendOut processes the final subscription queue. It's here in case the subscriber
+// is slow, and we need to buffer the messages.
+func (e *ethSubscription) startOut(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-e.sendCond:
+			e.sendLk.Lock()
+
+			for !e.toSend.Empty() {
+				front := e.toSend.Dequeue()
+				e.sendQueueLen--
+
+				e.sendLk.Unlock()
+
+				if err := e.out(ctx, front); err != nil {
+					log.Warnw("error sending subscription response, killing subscription", "sub", e.id, "error", err)
+					e.stop()
+					return
+				}
+
+				e.sendLk.Lock()
+			}
+
+			e.sendLk.Unlock()
+		}
+	}
+}
+
 func (e *ethSubscription) send(ctx context.Context, v interface{}) {
 	resp := ethtypes.EthSubscriptionResponse{
 		SubscriptionID: e.id,
@@ -1696,10 +1733,22 @@ func (e *ethSubscription) send(ctx context.Context, v interface{}) {
 		return
 	}
 
-	if err := e.out(ctx, outParam); err != nil {
-		log.Warnw("sending subscription response", "sub", e.id, "error", err)
+	e.sendLk.Lock()
+	defer e.sendLk.Unlock()
+
+	e.toSend.Enqueue(outParam)
+
+	e.sendQueueLen++
+	if e.sendQueueLen > maxSendQueue {
+		log.Warnw("subscription send queue full, killing subscription", "sub", e.id)
+		e.stop()
 		return
 	}
+
+	select {
+	case e.sendCond <- struct{}{}:
+	default: // already signalled, and we're holding the lock so we know that the event will be processed
+	}
 }
 
 func (e *ethSubscription) start(ctx context.Context) {
@@ -1743,11 +1792,23 @@ func (e *ethSubscription) start(ctx context.Context) {
 
 func (e *ethSubscription) stop() {
 	e.mu.Lock()
-	defer e.mu.Unlock()
+	if e.quit == nil {
+		e.mu.Unlock()
+		return
+	}
 
 	if e.quit != nil {
 		e.quit()
 		e.quit = nil
+		e.mu.Unlock()
+
+		for _, f := range e.filters {
+			// note: the context in actually unused in uninstallFilter
+			if err := e.uninstallFilter(context.TODO(), f); err != nil {
+				// this will leave the filter a zombie, collecting events up to the maximum allowed
+				log.Warnf("failed to remove filter when unsubscribing: %v", err)
+			}
+		}
 	}
 }