Merge pull request #922 from statechannels/eth-resub

Periodically resubscribe to chain event logs

.github/workflows/testground-nightly.yml

@@ -31,13 +31,13 @@ jobs:
         run:    testground plan import --from ./go-nitro-testground
         working-directory: "code"
 
-      - name: Run 3 min Test 
+      - name: Run 10 min Test 
         run: | 
          testground --endpoint=${{secrets.TG_SERVER_URL}}  run s --wait \
          -tp=isNightly=true -p=go-nitro-testground -t=virtual-payment \
          -b=docker:go -r=local:docker \
          -tp=numOfHubs=1 -tp=numOfPayers=10 -tp=numOfPayees=1  -i=12 \
-         -tp=paymentTestDuration=180  -tp=concurrentPaymentJobs=10 \
+         -tp=paymentTestDuration=600  -tp=concurrentPaymentJobs=10 \
          --tp=networkLatency=15 --tp=networkJitter=2 \
          --metadata-repo "${{github.repository}}" \
          --metadata-branch "${{github.event.pull_request.head.ref}}" \

client/engine/chainservice/eth_chainservice.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"log"
+	"time"
 
 	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
@@ -37,6 +38,12 @@ type EthChainService struct {
 	logger                   *log.Logger
 }
 
+// RESUB_INTERVAL is how often we resubscribe to log events.
+// We do this to avoid https://github.com/ethereum/go-ethereum/issues/23845
+// We use 4:30 as the default filter timeout is 5 minutes.
+// See https://github.com/ethereum/go-ethereum/blob/e14164d516600e9ac66f9060892e078f5c076229/eth/filters/filter_system.go#L43
+const RESUB_INTERVAL = 4*time.Minute + 30*time.Second
+
 // NewEthChainService constructs a chain service that submits transactions to a NitroAdjudicator
 // and listens to events from an eventSource
 func NewEthChainService(chain ethChain, na *NitroAdjudicator.NitroAdjudicator,
@@ -109,27 +116,42 @@ func (ecs *EthChainService) SendTransaction(tx protocols.ChainTransaction) error
 		return fmt.Errorf("unexpected transaction type %T", tx)
 	}
 }
-
 func (ecs *EthChainService) subcribeToEvents() error {
+
+	go ecs.listenForLogEvents()
+	return nil
+}
+
+func (ecs *EthChainService) listenForLogEvents() {
 	// Subsribe to Adjudicator events
 	query := ethereum.FilterQuery{
 		Addresses: []common.Address{ecs.naAddress},
 	}
 	logs := make(chan ethTypes.Log)
 	sub, err := ecs.chain.SubscribeFilterLogs(context.Background(), query, logs)
 	if err != nil {
-		return err
+		panic(err)
 	}
-	go ecs.listenForLogEvents(sub, logs)
-	return nil
-}
-
-func (ecs *EthChainService) listenForLogEvents(sub ethereum.Subscription, logs chan ethTypes.Log) {
 	for {
 		select {
 		case err := <-sub.Err():
-			// TODO should we try resubscribing to chain events
-			ecs.logger.Printf("event subscription error: %v", err)
+			if err != nil {
+				panic(err)
+			}
+
+			// If the error is nil then the subscription was closed and we need to re-subscribe.
+			// This is a workaround for https://github.com/ethereum/go-ethereum/issues/23845
+			var sErr error
+			sub, sErr = ecs.chain.SubscribeFilterLogs(context.Background(), query, logs)
+			if sErr != nil {
+				panic(err)
+			}
+			ecs.logger.Println("resubscribed to filtered logs")
+
+		case <-time.After(RESUB_INTERVAL):
+			// Due to https://github.com/ethereum/go-ethereum/issues/23845 we can't rely on a long running subscription.
+			// We unsub here and recreate the subscription in the next iteration of the select.
+			sub.Unsubscribe()
 		case chainEvent := <-logs:
 			switch chainEvent.Topics[0] {
 			case depositedTopic:
@@ -168,11 +190,13 @@ func (ecs *EthChainService) listenForLogEvents(sub ethereum.Subscription, logs c
 
 				event := ConcludedEvent{commonEvent: commonEvent{channelID: ce.ChannelId, BlockNum: chainEvent.BlockNumber}}
 				ecs.out <- event
+
 			default:
 				ecs.logger.Printf("Unknown chain event")
 			}
 		}
 	}
+
 }
 
 // EventFeed returns the out chan, and narrows the type so that external consumers may only receive on it.