fix(ethmonitor): confirm cached parents and track canonical incarnations

Author: pkieltyka

ethmonitor/bootstrap.go

@@ -38,8 +38,9 @@ func (c *Chain) bootstrapBlocks(blocks Blocks) error {
 		return nil
 	}
 
-	if len(blocks) == 1 {
+	if len(blocks) == 1 && blocks[0].Event != Added {
 		c.blocks = blocks.Copy()
+		c.blocks[0].canonicalState = nil
 		return nil
 	}
 
@@ -51,7 +52,7 @@ func (c *Chain) bootstrapBlocks(blocks Blocks) error {
 
 	for _, b := range blocks {
 		if b.Event == Added {
-			err := c.push(b)
+			_, err := c.push(b)
 			if err != nil {
 				return fmt.Errorf("ethmonitor: bootstrap failed to build canonical chain: %w", err)
 			}
@@ -101,5 +102,6 @@ func (b *Block) UnmarshalJSON(data []byte) error {
 	b.Event = s.Event
 	b.Logs = s.Logs
 	b.OK = s.OK
+	b.canonicalState = nil
 	return nil
 }

ethmonitor/canonical_state_test.go

@@ -0,0 +1,179 @@
+package ethmonitor
+
+import (
+	"encoding/json"
+	"fmt"
+	"math/big"
+	"testing"
+
+	"github.com/0xsequence/ethkit/go-ethereum/common"
+	"github.com/0xsequence/ethkit/go-ethereum/core/types"
+	"github.com/stretchr/testify/require"
+)
+
+func canonicalTestBlock(num int64) *Block {
+	return &Block{
+		Block: types.NewBlockWithHeader(&types.Header{
+			Number:     big.NewInt(num),
+			BlockHash:  common.BigToHash(big.NewInt(num)),
+			ParentHash: common.BigToHash(big.NewInt(num - 1)),
+			Time:       uint64(num),
+		}),
+		Event: Added,
+		OK:    true,
+	}
+}
+
+func pushCanonicalTestBlock(t *testing.T, chain *Chain, input *Block) *Block {
+	t.Helper()
+	chain.push(input)
+	block := chain.Head()
+	require.NotNil(t, block)
+	require.Equal(t, input.Hash(), block.Hash())
+	return block
+}
+
+// The interface keeps the before-fix proof executable without the new method.
+func canonicalTestState(t *testing.T, block *Block) (uint64, bool) {
+	t.Helper()
+	state, ok := any(block).(interface{ CanonicalState() (uint64, bool) })
+	if !ok {
+		return 0, false
+	}
+	return state.CanonicalState()
+}
+
+func TestBlockCanonicalStateRetention(t *testing.T) {
+	for _, depth := range []int64{1, 3} {
+		t.Run(fmt.Sprintf("evictionDepth=%d", depth), func(t *testing.T) {
+			chain := newChain(10, false)
+			block := pushCanonicalTestBlock(t, chain, canonicalTestBlock(100))
+			copy := chain.Blocks().Copy()[0]
+			incarnation, canonical := canonicalTestState(t, block)
+			require.Positive(t, incarnation)
+			require.True(t, canonical)
+			for num := int64(101); num <= 109+depth; num++ {
+				pushCanonicalTestBlock(t, chain, canonicalTestBlock(num))
+			}
+			require.Nil(t, chain.GetBlock(block.Hash()))
+			for _, snapshot := range []*Block{block, copy} {
+				got, canonical := canonicalTestState(t, snapshot)
+				require.Equal(t, incarnation, got)
+				require.True(t, canonical, "retention eviction was mistaken for removal")
+			}
+		})
+	}
+}
+
+func TestBlockCanonicalStateReadoption(t *testing.T) {
+	chain := newChain(10, false)
+	input := canonicalTestBlock(100)
+	block := pushCanonicalTestBlock(t, chain, input)
+	shallow := *block
+	copy := chain.Blocks().Copy()[0]
+	incarnation, canonical := canonicalTestState(t, block)
+	require.Positive(t, incarnation)
+	require.True(t, canonical)
+	removed := *chain.pop()
+	removed.Event = Removed
+	for _, snapshot := range []*Block{block, &shallow, copy, &removed} {
+		got, canonical := canonicalTestState(t, snapshot)
+		require.Equal(t, incarnation, got)
+		require.False(t, canonical, "snapshot did not observe the actual removal")
+	}
+
+	// Reusing either the original input or a removed snapshot must create a fresh
+	// owned incarnation without mutating old Added/Removed event copies.
+	for _, reused := range []*Block{input, copy} {
+		fresh := pushCanonicalTestBlock(t, chain, reused)
+		got, canonical := canonicalTestState(t, fresh)
+		require.Greater(t, got, incarnation)
+		require.True(t, canonical)
+		for _, snapshot := range []*Block{block, &shallow, copy, &removed} {
+			got, canonical := canonicalTestState(t, snapshot)
+			require.Equal(t, incarnation, got)
+			require.False(t, canonical, "fresh readoption revived an old event")
+		}
+		chain.pop()
+	}
+}
+
+func TestBlockCanonicalStateBootstrap(t *testing.T) {
+	for _, count := range []int{1, 3} {
+		for _, serialized := range []bool{false, true} {
+			t.Run(fmt.Sprintf("blocks=%d/JSON=%v", count, serialized), func(t *testing.T) {
+				inputs := make(Blocks, count)
+				for i := range inputs {
+					inputs[i] = canonicalTestBlock(int64(100 + i))
+				}
+				chain := newChain(10, true)
+				if serialized {
+					data, err := json.Marshal(inputs)
+					require.NoError(t, err)
+					require.NoError(t, chain.BootstrapFromBlocksJSON(data))
+				} else {
+					require.NoError(t, chain.BootstrapFromBlocks(inputs))
+				}
+				for _, block := range chain.Blocks().Copy() {
+					incarnation, canonical := canonicalTestState(t, block)
+					require.Positive(t, incarnation)
+					require.True(t, canonical)
+					data, err := json.Marshal(block)
+					require.NoError(t, err)
+					require.NotContains(t, string(data), "incarnation")
+					require.NoError(t, json.Unmarshal(data, block))
+					incarnation, canonical = canonicalTestState(t, block)
+					require.Zero(t, incarnation, "serialized state preserved runtime ownership")
+					require.False(t, canonical)
+				}
+			})
+		}
+	}
+}
+
+func TestBlockCanonicalStateConcurrentRemoval(t *testing.T) {
+	chain := newChain(10, false)
+	input := canonicalTestBlock(100)
+	block := pushCanonicalTestBlock(t, chain, input)
+	incarnation, _ := canonicalTestState(t, block)
+	require.Positive(t, incarnation)
+	state := any(block).(interface{ CanonicalState() (uint64, bool) })
+	start, done := make(chan struct{}), make(chan struct{})
+	result := make(chan error, 1)
+	go func() {
+		close(start)
+		removed := false
+		for {
+			got, canonical := state.CanonicalState()
+			if got != incarnation || (removed && canonical) {
+				result <- fmt.Errorf("old incarnation changed or revived: id=%d canonical=%v", got, canonical)
+				return
+			}
+			removed = removed || !canonical
+			select {
+			case <-done:
+				result <- nil
+				return
+			default:
+			}
+		}
+	}()
+	<-start
+	for i := 0; i < 100; i++ {
+		chain.pop()
+		pushCanonicalTestBlock(t, chain, input)
+	}
+	close(done)
+	require.NoError(t, <-result)
+	got, canonical := state.CanonicalState()
+	require.Equal(t, incarnation, got)
+	require.False(t, canonical)
+}
+
+func TestBlockCanonicalStateUntracked(t *testing.T) {
+	for _, block := range []*Block{nil, canonicalTestBlock(100)} {
+		incarnation, canonical := canonicalTestState(t, block)
+		require.Zero(t, incarnation)
+		require.False(t, canonical)
+	}
+}

ethmonitor/chain.go

@@ -7,6 +7,7 @@ import (
 	"math/big"
 	"slices"
 	"sync"
+	"sync/atomic"
 
 	"github.com/0xsequence/ethkit/go-ethereum"
 	"github.com/0xsequence/ethkit/go-ethereum/common"
@@ -28,6 +29,7 @@ type Chain struct {
 
 	mu               sync.RWMutex
 	averageBlockTime float64 // in seconds
+	lastIncarnation  uint64
 }
 
 func newChain(retentionLimit int, bootstrapMode bool) *Chain {
@@ -59,7 +61,7 @@ func newChain(retentionLimit int, bootstrapMode bool) *Chain {
 // }
 
 // Push to the top of the stack
-func (c *Chain) push(nextBlock *Block) error {
+func (c *Chain) push(nextBlock *Block) (*Block, error) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
@@ -70,12 +72,12 @@ func (c *Chain) push(nextBlock *Block) error {
 
 		// Assert pointing at prev block
 		if nextBlock.ParentHash() != headBlock.Hash() {
-			return ErrUnexpectedParentHash
+			return nil, ErrUnexpectedParentHash
 		}
 
 		// Assert block numbers are in sequence
 		if nextBlock.NumberU64() != headBlock.NumberU64()+1 {
-			return ErrUnexpectedBlockNumber
+			return nil, ErrUnexpectedBlockNumber
 		}
 
 		// Update average block time
@@ -86,14 +88,20 @@ func (c *Chain) push(nextBlock *Block) error {
 		}
 	}
 
+	// Each adoption owns its state so reusing an input cannot revive old events.
+	c.lastIncarnation++
+	block := *nextBlock
+	block.canonicalState = &blockCanonicalState{incarnation: c.lastIncarnation}
+	block.canonicalState.canonical.Store(true)
+
 	// Add to head of stack
-	c.blocks = append(c.blocks, nextBlock)
+	c.blocks = append(c.blocks, &block)
 	if len(c.blocks) > c.retentionLimit {
 		c.blocks[0] = nil
 		c.blocks = c.blocks[1:]
 	}
 
-	return nil
+	return &block, nil
 }
 
 // Pop from the top of the stack
@@ -107,6 +115,9 @@ func (c *Chain) pop() *Block {
 
 	n := len(c.blocks) - 1
 	block := c.blocks[n]
+	if block.canonicalState != nil {
+		block.canonicalState.canonical.Store(false)
+	}
 	c.blocks[n] = nil
 	c.blocks = c.blocks[:n]
 	return block
@@ -215,6 +226,9 @@ const (
 	Removed
 )
 
+// Block contains a monitored block and its event data.
+// Construct values with keyed composite literals: private canonical state makes
+// positional literals unsupported.
 type Block struct {
 	*types.Block
 
@@ -228,6 +242,26 @@ type Block struct {
 
 	// OK flag which represents the block is ready for broadcasting
 	OK bool
+
+	canonicalState *blockCanonicalState
+}
+
+type blockCanonicalState struct {
+	incarnation uint64
+	canonical   atomic.Bool
+}
+
+// CanonicalState reports the monitor-assigned incarnation and whether it has
+// remained canonical without a known removal. Zero means the block is untracked.
+// Retention eviction preserves this state; in-memory copies share removal updates.
+// Incarnations are local to a monitor chain. Serialized blocks are untracked
+// until accepted by a monitor.
+func (b *Block) CanonicalState() (incarnation uint64, canonical bool) {
+	if b == nil || b.canonicalState == nil {
+		return 0, false
+	}
+	state := b.canonicalState
+	return state.incarnation, state.canonical.Load()
 }
 
 type Blocks []*Block
@@ -560,10 +594,11 @@ func (blocks Blocks) Copy() Blocks {
 		}
 
 		nb[i] = &Block{
-			Block: b.Block,
-			Event: b.Event,
-			Logs:  logs,
-			OK:    b.OK,
+			Block:          b.Block,
+			Event:          b.Event,
+			Logs:           logs,
+			OK:             b.OK,
+			canonicalState: b.canonicalState,
 		}
 	}
 

ethmonitor/ethmonitor.go

@@ -640,10 +640,10 @@ func (m *Monitor) monitor(ctx context.Context, wg *sync.WaitGroup) error {
 			// fetch the next block, either via the stream or via a poll
 			nextBlock, nextBlockPayload, miss, err := m.fetchNextBlock(ctx, false)
 
-			// with prefetching, a cached block may be a stale entry from an abandoned
-			// fork, written by a prefetch worker of this or a peer monitor. confirm a
-			// parent mismatch with the node before treating it as a reorg.
-			if err == nil && m.prefetch != nil && headBlock != nil && nextBlock.ParentHash() != headBlock.Hash() {
+			// A cached block may be a stale entry from an abandoned fork, written by
+			// this or a peer monitor even when local prefetching is disabled. Confirm
+			// a parent mismatch with the node before treating it as a reorg.
+			if err == nil && m.cache != nil && headBlock != nil && nextBlock.ParentHash() != headBlock.Hash() {
 				nextBlock, nextBlockPayload, miss, err = m.refetchNextBlock(ctx, nextBlock)
 			}
 
@@ -730,9 +730,11 @@ func (m *Monitor) buildCanonicalChain(ctx context.Context, nextBlock *types.Bloc
 
 	if headBlock == nil || nextBlock.ParentHash() == headBlock.Hash() {
 		// block-chaining it up
-		block := &Block{Event: Added, Block: nextBlock}
-		events = append(events, block)
-		return events, m.chain.push(block)
+		block, err := m.chain.push(&Block{Event: Added, Block: nextBlock})
+		if err != nil {
+			return events, err
+		}
+		return append(events, block), nil
 	}
 
 	// next block doest match prevHash, therefore we must pop our previous block and recursively
@@ -778,8 +780,7 @@ func (m *Monitor) buildCanonicalChain(ctx context.Context, nextBlock *types.Bloc
 		return events, err
 	}
 
-	block := &Block{Event: Added, Block: nextBlock}
-	err = m.chain.push(block)
+	block, err := m.chain.push(&Block{Event: Added, Block: nextBlock})
 	if err != nil {
 		return events, err
 	}