From 1a1688bbb3105814fac6410960c4413d1d27fa4e Mon Sep 17 00:00:00 2001 From: Mark Paine Date: Tue, 9 Dec 2025 12:36:21 +1300 Subject: [PATCH 1/3] Add MaxRollbackDepth to prevent deep reorg processing Add optional MaxRollbackDepth parameter to WalkerOptions to stop the walker service when a reorg exceeds the configured depth limit. - Added MaxRollbackDepth int64 field to WalkerOptions (0 = unlimited) - Updated undoBlocks() to check rollback depth during backtracking - Walker stops (returns running=false) if depth exceeds limit This prevents processing dangerous deep reorgs that could corrupt downstream state by stopping before emitting an undo command. --- walker/walker.go | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/walker/walker.go b/walker/walker.go index 27beb26..b3a44ad 100644 --- a/walker/walker.go +++ b/walker/walker.go @@ -48,6 +48,7 @@ type WalkerOptions struct { TipChanged chan string // from TipChaser() FullUndoBlocks bool // fully decode blocks in UndoForkBlocks (or just hash and height) BufferBlocks int // number of blocks to decode ahead of the consumer (channel size, default 10) + MaxRollbackDepth int64 // maximum rollback depth before stopping (0 = unlimited) } /* @@ -100,6 +101,7 @@ func WalkTheDoge(opts WalkerOptions) (service governor.Service, blocks chan Bloc fullUndoBlocks: opts.FullUndoBlocks, lastProcessed: opts.LastProcessedBlock, blockInterval: POLL_INTERVAL, + maxRollback: opts.MaxRollbackDepth, } if opts.TipChanged != nil { // We will receive tipChanged notifications: use a longer polling timer @@ -121,6 +123,7 @@ type dogeWalker struct { lastProcessed string // last processed block hash to begin walking from (hex) blockInterval time.Duration // interval for polling blocks (longer if tipChanged is set) isIdle bool // true if the last message we sent was 'idle' + maxRollback int64 // maximum rollback depth } func (c *dogeWalker) Run() { @@ -285,6 +288,10 @@ func (c *dogeWalker) undoBlocks(head spec.BlockHeader) (undo *UndoForkBlocks, ne } // Accumulate undo info. undo.UndoBlocks = append(undo.UndoBlocks, head.Hash) + if c.maxRollback > 0 && int64(len(undo.UndoBlocks)) > c.maxRollback { + log.Printf("DogeWalker: MaxRollbackDepth exceeded (%d > %d). Stopping service.", len(undo.UndoBlocks), c.maxRollback) + return undo, "", false // stopping + } if c.fullUndoBlocks { block, cont := c.fetchBlockData(head.Hash) if !cont { From 0b6c6a93f507cedbbb0df159de610b5964b7dc8f Mon Sep 17 00:00:00 2001 From: Mark Paine Date: Tue, 9 Dec 2025 13:25:17 +1300 Subject: [PATCH 2/3] Fix: Return nil undo when MaxRollbackDepth exceeded to prevent sending incomplete undo When MaxRollbackDepth is exceeded, undoBlocks() was returning a partially populated undo object (with UndoBlocks but without LastValidHeight/LastValidHash). This could allow an invalid undo command to be sent to consumers. Fix: - Return nil instead of incomplete undo when limit exceeded - Add nil checks in callers before sending undo commands This ensures deep reorgs exceeding the limit are properly stopped without sending invalid undo commands, while normal reorgs continue to work correctly. --- walker/walker.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/walker/walker.go b/walker/walker.go index b3a44ad..71c331b 100644 --- a/walker/walker.go +++ b/walker/walker.go @@ -171,7 +171,7 @@ func (c *dogeWalker) Run() { if head.Confirmations == -1 { // Last-processed block is longer on-chain, start with a rollback. undo, nextBlock, cont := c.undoBlocks(head) - if !cont { + if !cont || undo == nil { return // stopping } select { @@ -251,7 +251,7 @@ func (c *dogeWalker) followTheChain(height int64, nextUnprocessed string) (lastP // This block is no longer on-chain. // Roll back until we find a block that is on-chain. undo, nextBlock, cont := c.undoBlocks(head) - if !cont { + if !cont || undo == nil { return lastProcessed, false } select { @@ -290,7 +290,7 @@ func (c *dogeWalker) undoBlocks(head spec.BlockHeader) (undo *UndoForkBlocks, ne undo.UndoBlocks = append(undo.UndoBlocks, head.Hash) if c.maxRollback > 0 && int64(len(undo.UndoBlocks)) > c.maxRollback { log.Printf("DogeWalker: MaxRollbackDepth exceeded (%d > %d). Stopping service.", len(undo.UndoBlocks), c.maxRollback) - return undo, "", false // stopping + return nil, "", false // stopping - return nil to prevent sending incomplete undo } if c.fullUndoBlocks { block, cont := c.fetchBlockData(head.Hash) From 591ca409c8d3497bc0b836890867468718dc0445 Mon Sep 17 00:00:00 2001 From: Mark Paine Date: Tue, 9 Dec 2025 14:44:06 +1300 Subject: [PATCH 3/3] Add MaxRollbackWaitDuration for graceful deep reorg handling - Add MaxRollbackWaitDuration option to WalkerOptions - Implement wait and retry mechanism when MaxRollbackDepth is exceeded - When deep reorg detected, wait configured duration before retrying - Prevents restart loop by giving RPC node time to re-index - Zero value (0) means return immediately (backward compatible) - Default wait duration: 5 seconds (configured in indexer) --- walker/walker.go | 81 +++++++++++++++++++++++++++++++++--------------- 1 file changed, 56 insertions(+), 25 deletions(-) diff --git a/walker/walker.go b/walker/walker.go index 71c331b..4639d2b 100644 --- a/walker/walker.go +++ b/walker/walker.go @@ -42,13 +42,14 @@ type UndoForkBlocks struct { // Configuraton for WalkTheDoge. type WalkerOptions struct { - Chain *doge.ChainParams // chain parameters, e.g. doge.DogeMainNetChain - LastProcessedBlock string // last processed block hash to begin walking from (hex) - Client spec.Blockchain // from NewCoreRPCClient() - TipChanged chan string // from TipChaser() - FullUndoBlocks bool // fully decode blocks in UndoForkBlocks (or just hash and height) - BufferBlocks int // number of blocks to decode ahead of the consumer (channel size, default 10) - MaxRollbackDepth int64 // maximum rollback depth before stopping (0 = unlimited) + Chain *doge.ChainParams // chain parameters, e.g. doge.DogeMainNetChain + LastProcessedBlock string // last processed block hash to begin walking from (hex) + Client spec.Blockchain // from NewCoreRPCClient() + TipChanged chan string // from TipChaser() + FullUndoBlocks bool // fully decode blocks in UndoForkBlocks (or just hash and height) + BufferBlocks int // number of blocks to decode ahead of the consumer (channel size, default 10) + MaxRollbackDepth int64 // maximum rollback depth before waiting for re-index (0 = unlimited) + MaxRollbackWaitDuration time.Duration // wait duration before retrying after deep reorg (0 = return immediately) } /* @@ -94,14 +95,15 @@ func WalkTheDoge(opts WalkerOptions) (service governor.Service, blocks chan Bloc } c := dogeWalker{ // The larger this channel is, the more blocks we can decode-ahead. - output: make(chan BlockOrUndo, chanSize), - client: opts.Client, - chain: opts.Chain, - tipChanged: opts.TipChanged, - fullUndoBlocks: opts.FullUndoBlocks, - lastProcessed: opts.LastProcessedBlock, - blockInterval: POLL_INTERVAL, - maxRollback: opts.MaxRollbackDepth, + output: make(chan BlockOrUndo, chanSize), + client: opts.Client, + chain: opts.Chain, + tipChanged: opts.TipChanged, + fullUndoBlocks: opts.FullUndoBlocks, + lastProcessed: opts.LastProcessedBlock, + blockInterval: POLL_INTERVAL, + maxRollback: opts.MaxRollbackDepth, + maxRollbackWait: opts.MaxRollbackWaitDuration, } if opts.TipChanged != nil { // We will receive tipChanged notifications: use a longer polling timer @@ -114,16 +116,17 @@ func WalkTheDoge(opts WalkerOptions) (service governor.Service, blocks chan Bloc // dogeWalker is the internal state. type dogeWalker struct { governor.ServiceCtx - output chan BlockOrUndo - client spec.Blockchain - chain *doge.ChainParams - tipChanged chan string // receive from TipChaser. - stop <-chan struct{} // ctx.Done() channel. - fullUndoBlocks bool // fully decode blocks in UndoForkBlocks - lastProcessed string // last processed block hash to begin walking from (hex) - blockInterval time.Duration // interval for polling blocks (longer if tipChanged is set) - isIdle bool // true if the last message we sent was 'idle' - maxRollback int64 // maximum rollback depth + output chan BlockOrUndo + client spec.Blockchain + chain *doge.ChainParams + tipChanged chan string // receive from TipChaser. + stop <-chan struct{} // ctx.Done() channel. + fullUndoBlocks bool // fully decode blocks in UndoForkBlocks + lastProcessed string // last processed block hash to begin walking from (hex) + blockInterval time.Duration // interval for polling blocks (longer if tipChanged is set) + isIdle bool // true if the last message we sent was 'idle' + maxRollback int64 // maximum rollback depth + maxRollbackWait time.Duration // wait duration before retrying after deep reorg } func (c *dogeWalker) Run() { @@ -172,6 +175,15 @@ func (c *dogeWalker) Run() { // Last-processed block is longer on-chain, start with a rollback. undo, nextBlock, cont := c.undoBlocks(head) if !cont || undo == nil { + // Deep reorg detected - wait and retry if configured + if c.maxRollbackWait > 0 { + log.Printf("DogeWalker: Deep reorg detected, waiting %v for RPC node to re-index...", c.maxRollbackWait) + if c.Sleep(c.maxRollbackWait) { + return // stopping + } + // Retry checking the block - it might be back on-chain now + continue // retry the loop + } return // stopping } select { @@ -252,6 +264,25 @@ func (c *dogeWalker) followTheChain(height int64, nextUnprocessed string) (lastP // Roll back until we find a block that is on-chain. undo, nextBlock, cont := c.undoBlocks(head) if !cont || undo == nil { + // Deep reorg detected - wait and retry if configured + if c.maxRollbackWait > 0 { + log.Printf("DogeWalker: Deep reorg detected in followTheChain, waiting %v for RPC node to re-index...", c.maxRollbackWait) + if c.Sleep(c.maxRollbackWait) { + return lastProcessed, false // stopping + } + // Retry checking the block - it might be back on-chain now + // Re-fetch the block header to check if it's back on-chain + head, cont := c.fetchBlockHeader(nextUnprocessed) + if !cont { + return lastProcessed, false // stopping + } + if head.Confirmations != -1 { + // Block is back on-chain, continue processing + continue // retry processing this block + } + // Still off-chain, return to let Run() handle retry + return lastProcessed, false + } return lastProcessed, false } select {