-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add DLQ for reprocessing of failed blocks
- Loading branch information
1 parent
67dbd31
commit 49c8b35
Showing
13 changed files
with
357 additions
and
11 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -22,4 +22,7 @@ go.work | |
go.work.sum | ||
|
||
# Binary | ||
indexer | ||
indexer | ||
bin | ||
|
||
.idea |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,79 @@ | ||
package ingester | ||
|
||
import ( | ||
"fmt" | ||
"slices" | ||
"sync" | ||
|
||
"github.com/duneanalytics/blockchain-ingester/models" | ||
pq "github.com/emirpasic/gods/queues/priorityqueue" | ||
"github.com/emirpasic/gods/utils" | ||
) | ||
|
||
type DLQ struct { | ||
priorityQueue pq.Queue // structure not thread safe | ||
mutex sync.Mutex | ||
} | ||
|
||
// DLQBlock This is generic so that metadata about retries can be maintained in an envelope during processing for when | ||
// a block needs to make its way back onto the DLQ later | ||
type DLQBlock[T any] struct { | ||
Value T | ||
Retries int | ||
// TODO last error? | ||
// TODO last error time? | ||
// TODO next retry time? | ||
|
||
} | ||
|
||
func (b *DLQBlock[T]) String() string { | ||
return fmt.Sprintf("%+v", *b) | ||
} | ||
|
||
func MapDLQBlock[T, U any](b DLQBlock[T], mapper func(T) U) DLQBlock[U] { | ||
return DLQBlock[U]{ | ||
Value: mapper(b.Value), | ||
Retries: b.Retries, | ||
} | ||
} | ||
|
||
func NewDLQ() *DLQ { | ||
return &DLQ{priorityQueue: *pq.NewWith(byRetries)} | ||
} | ||
|
||
// Comparator function (sort by retry count in ascending order) | ||
func byRetries(a, b interface{}) int { | ||
return utils.IntComparator(a.(DLQBlock[int64]).Retries, b.(DLQBlock[int64]).Retries) | ||
} | ||
|
||
func (dlq *DLQ) AddBlockGaps(gaps []models.BlockGap) { | ||
// queue these in reverse so that recent blocks are retried first | ||
slices.SortFunc(gaps, func(a, b models.BlockGap) int { | ||
return -utils.IntComparator(a.FirstMissing, b.FirstMissing) | ||
}) | ||
|
||
dlq.mutex.Lock() | ||
defer dlq.mutex.Unlock() | ||
|
||
for _, gap := range gaps { | ||
for i := gap.FirstMissing; i <= gap.LastMissing; i++ { | ||
dlq.priorityQueue.Enqueue(DLQBlock[int64]{Value: i, Retries: 0}) | ||
} | ||
} | ||
} | ||
|
||
func (dlq *DLQ) AddBlock(blockNumber int64, retries int) { | ||
dlq.mutex.Lock() | ||
defer dlq.mutex.Unlock() | ||
|
||
dlq.priorityQueue.Enqueue(DLQBlock[int64]{Value: blockNumber, Retries: retries}) | ||
} | ||
|
||
// TODO: rate limiting of some sort | ||
func (dlq *DLQ) GetNextBlock() (value DLQBlock[int64], ok bool) { | ||
dlq.mutex.Lock() | ||
defer dlq.mutex.Unlock() | ||
|
||
dequeue, ok := dlq.priorityQueue.Dequeue() | ||
return dequeue.(DLQBlock[int64]), ok | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.