-
Notifications
You must be signed in to change notification settings - Fork 1.7k
/
head_tracker.go
361 lines (325 loc) · 11.3 KB
/
head_tracker.go
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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
package headtracker
import (
"context"
"errors"
"fmt"
"sync"
"time"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
"github.com/smartcontractkit/chainlink-common/pkg/logger"
"github.com/smartcontractkit/chainlink-common/pkg/services"
"github.com/smartcontractkit/chainlink-common/pkg/utils/mailbox"
htrktypes "github.com/smartcontractkit/chainlink/v2/common/headtracker/types"
"github.com/smartcontractkit/chainlink/v2/common/types"
)
var (
promCurrentHead = promauto.NewGaugeVec(prometheus.GaugeOpts{
Name: "head_tracker_current_head",
Help: "The highest seen head number",
}, []string{"evmChainID"})
promOldHead = promauto.NewCounterVec(prometheus.CounterOpts{
Name: "head_tracker_very_old_head",
Help: "Counter is incremented every time we get a head that is much lower than the highest seen head ('much lower' is defined as a block that is EVM.FinalityDepth or greater below the highest seen head)",
}, []string{"evmChainID"})
)
// HeadsBufferSize - The buffer is used when heads sampling is disabled, to ensure the callback is run for every head
const HeadsBufferSize = 10
type HeadTracker[
HTH htrktypes.Head[BLOCK_HASH, ID],
S types.Subscription,
ID types.ID,
BLOCK_HASH types.Hashable,
] struct {
services.StateMachine
log logger.SugaredLogger
headBroadcaster types.HeadBroadcaster[HTH, BLOCK_HASH]
headSaver types.HeadSaver[HTH, BLOCK_HASH]
mailMon *mailbox.Monitor
client htrktypes.Client[HTH, S, ID, BLOCK_HASH]
chainID ID
config htrktypes.Config
htConfig htrktypes.HeadTrackerConfig
backfillMB *mailbox.Mailbox[HTH]
broadcastMB *mailbox.Mailbox[HTH]
headListener types.HeadListener[HTH, BLOCK_HASH]
chStop services.StopChan
wgDone sync.WaitGroup
getNilHead func() HTH
}
// NewHeadTracker instantiates a new HeadTracker using HeadSaver to persist new block numbers.
func NewHeadTracker[
HTH htrktypes.Head[BLOCK_HASH, ID],
S types.Subscription,
ID types.ID,
BLOCK_HASH types.Hashable,
](
lggr logger.Logger,
client htrktypes.Client[HTH, S, ID, BLOCK_HASH],
config htrktypes.Config,
htConfig htrktypes.HeadTrackerConfig,
headBroadcaster types.HeadBroadcaster[HTH, BLOCK_HASH],
headSaver types.HeadSaver[HTH, BLOCK_HASH],
mailMon *mailbox.Monitor,
getNilHead func() HTH,
) types.HeadTracker[HTH, BLOCK_HASH] {
chStop := make(chan struct{})
lggr = logger.Named(lggr, "HeadTracker")
return &HeadTracker[HTH, S, ID, BLOCK_HASH]{
headBroadcaster: headBroadcaster,
client: client,
chainID: client.ConfiguredChainID(),
config: config,
htConfig: htConfig,
log: logger.Sugared(lggr),
backfillMB: mailbox.NewSingle[HTH](),
broadcastMB: mailbox.New[HTH](HeadsBufferSize),
chStop: chStop,
headListener: NewHeadListener[HTH, S, ID, BLOCK_HASH](lggr, client, config, chStop),
headSaver: headSaver,
mailMon: mailMon,
getNilHead: getNilHead,
}
}
// Start starts HeadTracker service.
func (ht *HeadTracker[HTH, S, ID, BLOCK_HASH]) Start(ctx context.Context) error {
return ht.StartOnce("HeadTracker", func() error {
ht.log.Debugw("Starting HeadTracker", "chainID", ht.chainID)
latestChain, err := ht.headSaver.Load(ctx)
if err != nil {
return err
}
if latestChain.IsValid() {
ht.log.Debugw(
fmt.Sprintf("HeadTracker: Tracking logs from last block %v with hash %s", latestChain.BlockNumber(), latestChain.BlockHash()),
"blockNumber", latestChain.BlockNumber(),
"blockHash", latestChain.BlockHash(),
)
}
// NOTE: Always try to start the head tracker off with whatever the
// latest head is, without waiting for the subscription to send us one.
//
// In some cases the subscription will send us the most recent head
// anyway when we connect (but we should not rely on this because it is
// not specced). If it happens this is fine, and the head will be
// ignored as a duplicate.
initialHead, err := ht.getInitialHead(ctx)
if err != nil {
if errors.Is(err, ctx.Err()) {
return nil
}
ht.log.Errorw("Error getting initial head", "err", err)
} else if initialHead.IsValid() {
if err := ht.handleNewHead(ctx, initialHead); err != nil {
return fmt.Errorf("error handling initial head: %w", err)
}
} else {
ht.log.Debug("Got nil initial head")
}
ht.wgDone.Add(3)
go ht.headListener.ListenForNewHeads(ht.handleNewHead, ht.wgDone.Done)
go ht.backfillLoop()
go ht.broadcastLoop()
ht.mailMon.Monitor(ht.broadcastMB, "HeadTracker", "Broadcast", ht.chainID.String())
return nil
})
}
// Close stops HeadTracker service.
func (ht *HeadTracker[HTH, S, ID, BLOCK_HASH]) Close() error {
return ht.StopOnce("HeadTracker", func() error {
close(ht.chStop)
ht.wgDone.Wait()
return ht.broadcastMB.Close()
})
}
func (ht *HeadTracker[HTH, S, ID, BLOCK_HASH]) Name() string {
return ht.log.Name()
}
func (ht *HeadTracker[HTH, S, ID, BLOCK_HASH]) HealthReport() map[string]error {
report := map[string]error{ht.Name(): ht.Healthy()}
services.CopyHealth(report, ht.headListener.HealthReport())
return report
}
func (ht *HeadTracker[HTH, S, ID, BLOCK_HASH]) Backfill(ctx context.Context, headWithChain HTH, depth uint) (err error) {
if uint(headWithChain.ChainLength()) >= depth {
return nil
}
baseHeight := headWithChain.BlockNumber() - int64(depth-1)
if baseHeight < 0 {
baseHeight = 0
}
return ht.backfill(ctx, headWithChain.EarliestHeadInChain(), baseHeight)
}
func (ht *HeadTracker[HTH, S, ID, BLOCK_HASH]) LatestChain() HTH {
return ht.headSaver.LatestChain()
}
func (ht *HeadTracker[HTH, S, ID, BLOCK_HASH]) getInitialHead(ctx context.Context) (HTH, error) {
head, err := ht.client.HeadByNumber(ctx, nil)
if err != nil {
return ht.getNilHead(), fmt.Errorf("failed to fetch initial head: %w", err)
}
loggerFields := []interface{}{"head", head}
if head.IsValid() {
loggerFields = append(loggerFields, "blockNumber", head.BlockNumber(), "blockHash", head.BlockHash())
}
ht.log.Debugw("Got initial head", loggerFields...)
return head, nil
}
func (ht *HeadTracker[HTH, S, ID, BLOCK_HASH]) handleNewHead(ctx context.Context, head HTH) error {
prevHead := ht.headSaver.LatestChain()
ht.log.Debugw(fmt.Sprintf("Received new head %v", head.BlockNumber()),
"blockHash", head.BlockHash(),
"parentHeadHash", head.GetParentHash(),
"blockTs", head.GetTimestamp(),
"blockTsUnix", head.GetTimestamp().Unix(),
"blockDifficulty", head.BlockDifficulty(),
)
err := ht.headSaver.Save(ctx, head)
if ctx.Err() != nil {
return nil
} else if err != nil {
return fmt.Errorf("failed to save head: %#v: %w", head, err)
}
if !prevHead.IsValid() || head.BlockNumber() > prevHead.BlockNumber() {
promCurrentHead.WithLabelValues(ht.chainID.String()).Set(float64(head.BlockNumber()))
headWithChain := ht.headSaver.Chain(head.BlockHash())
if !headWithChain.IsValid() {
return fmt.Errorf("HeadTracker#handleNewHighestHead headWithChain was unexpectedly nil")
}
ht.backfillMB.Deliver(headWithChain)
ht.broadcastMB.Deliver(headWithChain)
} else if head.BlockNumber() == prevHead.BlockNumber() {
if head.BlockHash() != prevHead.BlockHash() {
ht.log.Debugw("Got duplicate head", "blockNum", head.BlockNumber(), "head", head.BlockHash(), "prevHead", prevHead.BlockHash())
} else {
ht.log.Debugw("Head already in the database", "head", head.BlockHash())
}
} else {
ht.log.Debugw("Got out of order head", "blockNum", head.BlockNumber(), "head", head.BlockHash(), "prevHead", prevHead.BlockNumber())
prevUnFinalizedHead := prevHead.BlockNumber() - int64(ht.config.FinalityDepth())
if head.BlockNumber() < prevUnFinalizedHead {
promOldHead.WithLabelValues(ht.chainID.String()).Inc()
ht.log.Criticalf("Got very old block with number %d (highest seen was %d). This is a problem and either means a very deep re-org occurred, one of the RPC nodes has gotten far out of sync, or the chain went backwards in block numbers. This node may not function correctly without manual intervention.", head.BlockNumber(), prevHead.BlockNumber())
ht.SvcErrBuffer.Append(errors.New("got very old block"))
}
}
return nil
}
func (ht *HeadTracker[HTH, S, ID, BLOCK_HASH]) broadcastLoop() {
defer ht.wgDone.Done()
samplingInterval := ht.htConfig.SamplingInterval()
if samplingInterval > 0 {
ht.log.Debugf("Head sampling is enabled - sampling interval is set to: %v", samplingInterval)
debounceHead := time.NewTicker(samplingInterval)
defer debounceHead.Stop()
for {
select {
case <-ht.chStop:
return
case <-debounceHead.C:
item := ht.broadcastMB.RetrieveLatestAndClear()
if !item.IsValid() {
continue
}
ht.headBroadcaster.BroadcastNewLongestChain(item)
}
}
} else {
ht.log.Info("Head sampling is disabled - callback will be called on every head")
for {
select {
case <-ht.chStop:
return
case <-ht.broadcastMB.Notify():
for {
item, exists := ht.broadcastMB.Retrieve()
if !exists {
break
}
ht.headBroadcaster.BroadcastNewLongestChain(item)
}
}
}
}
}
func (ht *HeadTracker[HTH, S, ID, BLOCK_HASH]) backfillLoop() {
defer ht.wgDone.Done()
ctx, cancel := ht.chStop.NewCtx()
defer cancel()
for {
select {
case <-ht.chStop:
return
case <-ht.backfillMB.Notify():
for {
head, exists := ht.backfillMB.Retrieve()
if !exists {
break
}
{
err := ht.Backfill(ctx, head, uint(ht.config.FinalityDepth()))
if err != nil {
ht.log.Warnw("Unexpected error while backfilling heads", "err", err)
} else if ctx.Err() != nil {
break
}
}
}
}
}
}
// backfill fetches all missing heads up until the base height
func (ht *HeadTracker[HTH, S, ID, BLOCK_HASH]) backfill(ctx context.Context, head types.Head[BLOCK_HASH], baseHeight int64) (err error) {
headBlockNumber := head.BlockNumber()
if headBlockNumber <= baseHeight {
return nil
}
mark := time.Now()
fetched := 0
l := ht.log.With("blockNumber", headBlockNumber,
"n", headBlockNumber-baseHeight,
"fromBlockHeight", baseHeight,
"toBlockHeight", headBlockNumber-1)
l.Debug("Starting backfill")
defer func() {
if ctx.Err() != nil {
l.Warnw("Backfill context error", "err", ctx.Err())
return
}
l.Debugw("Finished backfill",
"fetched", fetched,
"time", time.Since(mark),
"err", err)
}()
for i := head.BlockNumber() - 1; i >= baseHeight; i-- {
// NOTE: Sequential requests here mean it's a potential performance bottleneck, be aware!
existingHead := ht.headSaver.Chain(head.GetParentHash())
if existingHead.IsValid() {
head = existingHead
continue
}
head, err = ht.fetchAndSaveHead(ctx, i, head.GetParentHash())
fetched++
if ctx.Err() != nil {
ht.log.Debugw("context canceled, aborting backfill", "err", err, "ctx.Err", ctx.Err())
break
} else if err != nil {
return fmt.Errorf("fetchAndSaveHead failed: %w", err)
}
}
return
}
func (ht *HeadTracker[HTH, S, ID, BLOCK_HASH]) fetchAndSaveHead(ctx context.Context, n int64, hash BLOCK_HASH) (HTH, error) {
ht.log.Debugw("Fetching head", "blockHeight", n, "blockHash", hash)
head, err := ht.client.HeadByHash(ctx, hash)
if err != nil {
return ht.getNilHead(), err
} else if !head.IsValid() {
return ht.getNilHead(), errors.New("got nil head")
}
err = ht.headSaver.Save(ctx, head)
if err != nil {
return ht.getNilHead(), err
}
return head, nil
}