feat: gateway: auto-cleanup of installed filters when ws connection ends

gateway/handler.go

@@ -61,6 +61,7 @@ func Handler(
 	m := mux.NewRouter()
 
 	opts = append(opts, jsonrpc.WithReverseClient[lapi.EthSubscriberMethods]("Filecoin"), jsonrpc.WithServerErrors(lapi.RPCErrors))
+
 	serveRpc := func(path string, hnd interface{}) {
 		rpcServer := jsonrpc.NewServer(opts...)
 		rpcServer.Register("Filecoin", hnd)
@@ -106,7 +107,11 @@ type statefulCallHandler struct {
 }
 
 func (h statefulCallHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	r = r.WithContext(context.WithValue(r.Context(), statefulCallTrackerKey, newStatefulCallTracker()))
+	tracker := newStatefulCallTracker()
+	defer func() {
+		go tracker.cleanup()
+	}()
+	r = r.WithContext(context.WithValue(r.Context(), statefulCallTrackerKey, tracker))
 	h.next.ServeHTTP(w, r)
 }
 

gateway/node.go

@@ -7,6 +7,7 @@ import (
 
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
+	logger "github.com/ipfs/go-log/v2"
 	"go.opencensus.io/stats"
 	"golang.org/x/time/rate"
 
@@ -31,6 +32,8 @@ import (
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
 )
 
+var log = logger.Logger("gateway")
+
 const (
 	DefaultLookbackCap            = time.Hour * 24
 	DefaultStateWaitLookbackLimit = abi.ChainEpoch(20)

gateway/proxy_eth.go

@@ -6,7 +6,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"sync"
-	"time"
 
 	"github.com/ipfs/go-cid"
 	"golang.org/x/xerrors"
@@ -446,11 +445,8 @@ func (gw *Node) EthGetFilterChanges(ctx context.Context, id ethtypes.EthFilterID
 	if err != nil {
 		return nil, xerrors.Errorf("EthGetFilterChanges not supported: %w", err)
 	}
-	ft.lk.Lock()
-	_, ok := ft.userFilters[id]
-	ft.lk.Unlock()
 
-	if !ok {
+	if !ft.hasFilter(id) {
 		return nil, filter.ErrFilterNotFound
 	}
 
@@ -466,11 +462,8 @@ func (gw *Node) EthGetFilterLogs(ctx context.Context, id ethtypes.EthFilterID) (
 	if err != nil {
 		return nil, xerrors.Errorf("EthGetFilterLogs not supported: %w", err)
 	}
-	ft.lk.Lock()
-	_, ok := ft.userFilters[id]
-	ft.lk.Unlock()
 
-	if !ok {
+	if !ft.hasFilter(id) {
 		return nil, nil
 	}
 
@@ -482,7 +475,7 @@ func (gw *Node) EthNewFilter(ctx context.Context, filter *ethtypes.EthFilterSpec
 		return ethtypes.EthFilterID{}, err
 	}
 
-	return addUserFilterLimited(ctx, "EthNewFilter", func() (ethtypes.EthFilterID, error) {
+	return gw.addUserFilterLimited(ctx, "EthNewFilter", func() (ethtypes.EthFilterID, error) {
 		return gw.target.EthNewFilter(ctx, filter)
 	})
 }
@@ -492,7 +485,7 @@ func (gw *Node) EthNewBlockFilter(ctx context.Context) (ethtypes.EthFilterID, er
 		return ethtypes.EthFilterID{}, err
 	}
 
-	return addUserFilterLimited(ctx, "EthNewBlockFilter", func() (ethtypes.EthFilterID, error) {
+	return gw.addUserFilterLimited(ctx, "EthNewBlockFilter", func() (ethtypes.EthFilterID, error) {
 		return gw.target.EthNewBlockFilter(ctx)
 	})
 }
@@ -502,7 +495,7 @@ func (gw *Node) EthNewPendingTransactionFilter(ctx context.Context) (ethtypes.Et
 		return ethtypes.EthFilterID{}, err
 	}
 
-	return addUserFilterLimited(ctx, "EthNewPendingTransactionFilter", func() (ethtypes.EthFilterID, error) {
+	return gw.addUserFilterLimited(ctx, "EthNewPendingTransactionFilter", func() (ethtypes.EthFilterID, error) {
 		return gw.target.EthNewPendingTransactionFilter(ctx)
 	})
 }
@@ -517,6 +510,7 @@ func (gw *Node) EthUninstallFilter(ctx context.Context, id ethtypes.EthFilterID)
 	if err != nil {
 		return false, xerrors.Errorf("EthUninstallFilter not supported: %w", err)
 	}
+
 	ft.lk.Lock()
 	defer ft.lk.Unlock()
 
@@ -526,6 +520,7 @@ func (gw *Node) EthUninstallFilter(ctx context.Context, id ethtypes.EthFilterID)
 
 	ok, err := gw.target.EthUninstallFilter(ctx, id)
 	if err != nil {
+		// don't delete the filter, it's "stuck" so should still count towards the limit
 		return false, err
 	}
 
@@ -545,12 +540,12 @@ func (gw *Node) EthSubscribe(ctx context.Context, p jsonrpc.RawParams) (ethtypes
 	}
 
 	if gw.subHnd == nil {
-		return ethtypes.EthSubscriptionID{}, xerrors.New("subscription support not enabled")
+		return ethtypes.EthSubscriptionID{}, xerrors.New("EthSubscribe not supported: subscription support not enabled")
 	}
 
 	ethCb, ok := jsonrpc.ExtractReverseClient[api.EthSubscriberMethods](ctx)
 	if !ok {
-		return ethtypes.EthSubscriptionID{}, xerrors.Errorf("connection doesn't support callbacks")
+		return ethtypes.EthSubscriptionID{}, xerrors.Errorf("EthSubscribe not supported: connection doesn't support callbacks")
 	}
 
 	ft, err := getStatefulTracker(ctx)
@@ -581,7 +576,11 @@ func (gw *Node) EthSubscribe(ctx context.Context, p jsonrpc.RawParams) (ethtypes
 		return ethtypes.EthSubscriptionID{}, err
 	}
 
-	ft.userSubscriptions[sub] = time.Now()
+	ft.userSubscriptions[sub] = func() {
+		if _, err := gw.target.EthUnsubscribe(ctx, sub); err != nil {
+			log.Warnf("error unsubscribing after connection end: %v", err)
+		}
+	}
 
 	return sub, err
 }
@@ -605,6 +604,7 @@ func (gw *Node) EthUnsubscribe(ctx context.Context, id ethtypes.EthSubscriptionI
 
 	ok, err := gw.target.EthUnsubscribe(ctx, id)
 	if err != nil {
+		// don't delete the subscription, it's "stuck" so should still count towards the limit
 		return false, err
 	}
 
@@ -679,31 +679,40 @@ func (gw *Node) EthTraceFilter(ctx context.Context, filter ethtypes.EthTraceFilt
 
 var EthMaxFiltersPerConn = 16 // todo make this configurable
 
-func addUserFilterLimited(ctx context.Context, callName string, cb func() (ethtypes.EthFilterID, error)) (ethtypes.EthFilterID, error) {
+func (gw *Node) addUserFilterLimited(
+	ctx context.Context,
+	callName string,
+	install func() (ethtypes.EthFilterID, error),
+) (ethtypes.EthFilterID, error) {
 	ft, err := getStatefulTracker(ctx)
 	if err != nil {
 		return ethtypes.EthFilterID{}, xerrors.Errorf("%s not supported: %w", callName, err)
 	}
+
 	ft.lk.Lock()
 	defer ft.lk.Unlock()
 
 	if len(ft.userFilters) >= EthMaxFiltersPerConn {
 		return ethtypes.EthFilterID{}, xerrors.New("too many filters")
 	}
 
-	id, err := cb()
+	id, err := install()
 	if err != nil {
 		return id, err
 	}
 
-	ft.userFilters[id] = time.Now()
+	ft.userFilters[id] = func() {
+		if _, err := gw.target.EthUninstallFilter(ctx, id); err != nil {
+			log.Warnf("error uninstalling filter after connection end: %v", err)
+		}
+	}
 
 	return id, nil
 }
 
 func getStatefulTracker(ctx context.Context) (*statefulCallTracker, error) {
 	if jsonrpc.GetConnectionType(ctx) != jsonrpc.ConnectionTypeWS {
-		return nil, xerrors.New("stateful tracking is only available for websockets connections")
+		return nil, xerrors.New("stateful methods are only available on websockets connections")
 	}
 
 	if ct, ok := ctx.Value(statefulCallTrackerKey).(*statefulCallTracker); !ok {
@@ -713,17 +722,39 @@ func getStatefulTracker(ctx context.Context) (*statefulCallTracker, error) {
 	}
 }
 
+type cleanup func()
+
 type statefulCallTracker struct {
 	lk sync.Mutex
 
-	userFilters       map[ethtypes.EthFilterID]time.Time
-	userSubscriptions map[ethtypes.EthSubscriptionID]time.Time
+	userFilters       map[ethtypes.EthFilterID]cleanup
+	userSubscriptions map[ethtypes.EthSubscriptionID]cleanup
+}
+
+func (ft *statefulCallTracker) cleanup() {
+	ft.lk.Lock()
+	defer ft.lk.Unlock()
+
+	for _, cleanup := range ft.userFilters {
+		cleanup()
+	}
+	for _, cleanup := range ft.userSubscriptions {
+		cleanup()
+	}
+}
+
+func (ft *statefulCallTracker) hasFilter(id ethtypes.EthFilterID) bool {
+	ft.lk.Lock()
+	defer ft.lk.Unlock()
+
+	_, ok := ft.userFilters[id]
+	return ok
 }
 
 // called per request (ws connection)
 func newStatefulCallTracker() *statefulCallTracker {
 	return &statefulCallTracker{
-		userFilters:       make(map[ethtypes.EthFilterID]time.Time),
-		userSubscriptions: make(map[ethtypes.EthSubscriptionID]time.Time),
+		userFilters:       make(map[ethtypes.EthFilterID]cleanup),
+		userSubscriptions: make(map[ethtypes.EthSubscriptionID]cleanup),
 	}
 }