clearnode/eth_listener.go at main · erc7824/clearnode · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
package main

import (
	"context"
	"sync/atomic"
	"time"

	"github.com/erc7824/go-nitrolite"
	"github.com/ethereum/go-ethereum"
	"github.com/ethereum/go-ethereum/accounts/abi/bind"
	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/core/types"
	"github.com/ethereum/go-ethereum/event"
	"github.com/ipfs/go-log/v2"
)

var logger = log.Logger("base-event-listener")

const (
	maxBackOffCount = 5
)

func init() {
	log.SetAllLoggers(log.LevelDebug)
	log.SetLogLevel("base-event-listener", "debug")

	var err error
	custodyAbi, err = nitrolite.CustodyMetaData.GetAbi()
	if err != nil {
		panic(err)
	}
}

type LogHandler func(l types.Log)

// listenEvents listens for blockchain events and processes them with the provided handler
func listenEvents(
	ctx context.Context,
	client bind.ContractBackend,
	contractAddress common.Address,
	chainID uint32,
	lastBlock uint64,
	handler LogHandler,
) {
	var backOffCount atomic.Uint64
	var currentCh chan types.Log
	var eventSubscription event.Subscription

	logger.Infow("starting listening events", "chainID", chainID, "contractAddress", contractAddress.String())
	for {
		if eventSubscription == nil {
			waitForBackOffTimeout(int(backOffCount.Load()))

			currentCh = make(chan types.Log, 100)

			watchFQ := ethereum.FilterQuery{
				Addresses: []common.Address{contractAddress},
			}
			eventSub, err := client.SubscribeFilterLogs(ctx, watchFQ, currentCh)
			if err != nil {
				logger.Errorw("failed to subscribe on events", "error", err, "chainID", chainID, "contractAddress", contractAddress.String())
				backOffCount.Add(1)
				continue
			}

			eventSubscription = eventSub
			logger.Infow("watching events", "chainID", chainID, "contractAddress", contractAddress.String())
			backOffCount.Store(0)
		}

		select {
		case eventLog := <-currentCh:
			lastBlock = eventLog.BlockNumber
			logger.Debugw("received new event", "chainID", chainID, "contractAddress", contractAddress.String(), "blockNumber", lastBlock, "logIndex", eventLog.Index)
			handler(eventLog)
		case err := <-eventSubscription.Err():
			if err != nil {
				logger.Errorw("event subscription error", "error", err, "chainID", chainID, "contractAddress", contractAddress.String())
				eventSubscription.Unsubscribe()
			} else {
				logger.Debugw("subscription closed, resubscribing", "chainID", chainID, "contractAddress", contractAddress.String())
			}

			eventSubscription = nil
		}
	}
}

// waitForBackOffTimeout implements exponential backoff between retries
func waitForBackOffTimeout(backOffCount int) {
	if backOffCount > maxBackOffCount {
		logger.Fatalw("back off limit reached, exiting", "backOffCollisionCount", backOffCount)
		return
	}

	if backOffCount > 0 {
		logger.Infow("backing off before subscribing on contract events", "backOffCollisionCount", backOffCount)
		<-time.After(time.Duration(2^backOffCount-1) * time.Second)
	}
}