added the ability to ignore the cursor on the block poller

streamingfast · Nov 30, 2023 · 7aa2064 · 7aa2064
1 parent f5e9f30
commit 7aa2064
Show file tree

Hide file tree

Showing 4 changed files with 27 additions and 6 deletions.
diff --git a/blockpoller/options.go b/blockpoller/options.go
@@ -16,6 +16,14 @@ func WithStoringState(stateStorePath string) Option {
 	}
 }
 
+// WithCursorPurged ensures the poller will ignore the cursor and start from the startBlockNum
+// the cursor will still be saved as the poller progresses
+func WithCursorPurged() Option {
+	return func(p *BlockPoller) {
+		p.ignoreCursor = true
+	}
+}
+
 func WithLogger(logger *zap.Logger) Option {
 	return func(p *BlockPoller) {
 		p.logger = logger

diff --git a/blockpoller/poller.go b/blockpoller/poller.go
@@ -26,6 +26,7 @@ type BlockPoller struct {
 	startBlockNumGate    uint64
 	fetchBlockRetryCount uint64
 	stateStorePath       string
+	ignoreCursor         bool
 
 	blockFetcher BlockFetcher
 	blockHandler BlockHandler
@@ -77,7 +78,7 @@ func (p *BlockPoller) Run(ctx context.Context, startBlockNum uint64, chainLatest
 
 func (p *BlockPoller) run(resolvedStartBlock bstream.BlockRef) (err error) {
 
-	p.forkDB, resolvedStartBlock, err = initState(resolvedStartBlock, p.stateStorePath, p.logger)
+	p.forkDB, resolvedStartBlock, err = initState(resolvedStartBlock, p.stateStorePath, p.ignoreCursor, p.logger)
 	if err != nil {
 		return fmt.Errorf("unable to initialize cursor: %w", err)
 	}

diff --git a/blockpoller/state_file.go b/blockpoller/state_file.go
@@ -92,18 +92,30 @@ func (p *BlockPoller) saveState(blocks []*forkable.Block) error {
 	return nil
 }
 
-func initState(resolvedStartBlock bstream.BlockRef, stateStorePath string, logger *zap.Logger) (*forkable.ForkDB, bstream.BlockRef, error) {
+func initState(resolvedStartBlock bstream.BlockRef, stateStorePath string, ignoreCursor bool, logger *zap.Logger) (*forkable.ForkDB, bstream.BlockRef, error) {
 	forkDB := forkable.NewForkDB(forkable.ForkDBWithLogger(logger))
 
+	useStartBlockFunc := func() (*forkable.ForkDB, bstream.BlockRef, error) {
+		forkDB.InitLIB(resolvedStartBlock)
+		return forkDB, resolvedStartBlock, nil
+	}
+
+	if ignoreCursor {
+		logger.Info("ignorign cursor",
+			zap.Stringer("start_block", resolvedStartBlock),
+			zap.Stringer("lib", resolvedStartBlock),
+		)
+		return useStartBlockFunc()
+	}
+
 	sf, err := getState(stateStorePath)
 	if err != nil {
 		logger.Warn("unable to load cursor file, initializing a new forkdb",
 			zap.Stringer("start_block", resolvedStartBlock),
 			zap.Stringer("lib", resolvedStartBlock),
 			zap.Error(err),
 		)
-		forkDB.InitLIB(resolvedStartBlock)
-		return forkDB, resolvedStartBlock, nil
+		return useStartBlockFunc()
 	}
 
 	forkDB.InitLIB(bstream.NewBlockRef(sf.Lib.Id, sf.Lib.Num))

diff --git a/blockpoller/state_file_test.go b/blockpoller/state_file_test.go
@@ -54,7 +54,7 @@ func TestFireBlockFinalizer_state(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, expectedStateFileCnt, string(cnt))
 
-	forkDB, startBlock, err := initState(bstream.NewBlockRef("60a", 60), dirName, zap.NewNop())
+	forkDB, startBlock, err := initState(bstream.NewBlockRef("60a", 60), dirName, false, zap.NewNop())
 	require.NoError(t, err)
 
 	blocks, reachedLib := forkDB.CompleteSegment(bstream.NewBlockRef("105a", 105))
@@ -70,7 +70,7 @@ func TestFireBlockFinalizer_noSstate(t *testing.T) {
 	require.NoError(t, err)
 	defer os.Remove(dirName)
 
-	forkDB, startBlock, err := initState(bstream.NewBlockRef("60a", 60), dirName, logger)
+	forkDB, startBlock, err := initState(bstream.NewBlockRef("60a", 60), dirName, false, logger)
 	require.NoError(t, err)
 
 	blocks, reachedLib := forkDB.CompleteSegment(bstream.NewBlockRef("60a", 60))