@@ -71,45 +71,54 @@ func (rh *ReorgHandler) Start() {
7171 log .Debug ().Msgf ("Reorg handler running" )
7272 go func () {
7373 for range ticker .C {
74- lookbackFrom := new (big.Int ).Add (rh .lastCheckedBlock , big .NewInt (int64 (rh .blocksPerScan )))
75- blockHeaders , err := rh .storage .MainStorage .LookbackBlockHeaders (rh .rpc .GetChainID (), rh .blocksPerScan , lookbackFrom )
74+ // need to include lastCheckedBlock to check if next block's parent matches
75+ lookbackFrom := new (big.Int ).Add (rh .lastCheckedBlock , big .NewInt (int64 (rh .blocksPerScan - 1 )))
76+ mostRecentBlockChecked , err := rh .RunFromBlock (lookbackFrom )
7677 if err != nil {
77- log .Error ().Err (err ).Msg ("Error getting recent block headers " )
78+ log .Error ().Err (err ).Msg ("Error during reorg handling " )
7879 continue
7980 }
80- if len (blockHeaders ) == 0 {
81- log .Warn ().Msg ("No block headers found" )
81+ if mostRecentBlockChecked == nil {
8282 continue
8383 }
84- mostRecentBlockHeader := blockHeaders [0 ]
85- reorgEndIndex := findReorgEndIndex (blockHeaders )
86- if reorgEndIndex == - 1 {
87- rh .lastCheckedBlock = mostRecentBlockHeader .Number
88- rh .storage .OrchestratorStorage .SetLastReorgCheckedBlockNumber (rh .rpc .GetChainID (), mostRecentBlockHeader .Number )
89- metrics .ReorgHandlerLastCheckedBlock .Set (float64 (mostRecentBlockHeader .Number .Int64 ()))
90- continue
91- }
92- metrics .ReorgCounter .Inc ()
93- forkPoint , err := rh .findForkPoint (blockHeaders [reorgEndIndex :])
94- if err != nil {
95- log .Error ().Err (err ).Msg ("Error while finding fork point" )
96- continue
97- }
98- err = rh .handleReorg (forkPoint , lookbackFrom )
99- if err != nil {
100- log .Error ().Err (err ).Msg ("Error while handling reorg" )
101- continue
102- }
103- rh .lastCheckedBlock = mostRecentBlockHeader .Number
104- rh .storage .OrchestratorStorage .SetLastReorgCheckedBlockNumber (rh .rpc .GetChainID (), mostRecentBlockHeader .Number )
105- metrics .ReorgHandlerLastCheckedBlock .Set (float64 (mostRecentBlockHeader .Number .Int64 ()))
84+
85+ rh .lastCheckedBlock = mostRecentBlockChecked
86+ rh .storage .OrchestratorStorage .SetLastReorgCheckedBlockNumber (rh .rpc .GetChainID (), mostRecentBlockChecked )
87+ metrics .ReorgHandlerLastCheckedBlock .Set (float64 (mostRecentBlockChecked .Int64 ()))
10688 }
10789 }()
10890
10991 // Keep the program running (otherwise it will exit)
11092 select {}
11193}
11294
95+ func (rh * ReorgHandler ) RunFromBlock (lookbackFrom * big.Int ) (lastCheckedBlock * big.Int , err error ) {
96+ blockHeaders , err := rh .storage .MainStorage .LookbackBlockHeaders (rh .rpc .GetChainID (), rh .blocksPerScan , lookbackFrom )
97+ if err != nil {
98+ return nil , fmt .Errorf ("error getting recent block headers: %w" , err )
99+ }
100+ if len (blockHeaders ) == 0 {
101+ log .Warn ().Msg ("No block headers found during reorg handling" )
102+ return nil , nil
103+ }
104+ mostRecentBlockHeader := blockHeaders [0 ]
105+ reorgEndIndex := findReorgEndIndex (blockHeaders )
106+ if reorgEndIndex == - 1 {
107+ return mostRecentBlockHeader .Number , nil
108+ }
109+ reorgEndBlock := blockHeaders [reorgEndIndex ].Number
110+ metrics .ReorgCounter .Inc ()
111+ forkPoint , err := rh .findFirstForkedBlockNumber (blockHeaders [reorgEndIndex :])
112+ if err != nil {
113+ return nil , fmt .Errorf ("error while finding fork point: %w" , err )
114+ }
115+ err = rh .handleReorg (forkPoint , reorgEndBlock )
116+ if err != nil {
117+ return nil , fmt .Errorf ("error while handling reorg: %w" , err )
118+ }
119+ return mostRecentBlockHeader .Number , nil
120+ }
121+
113122func findReorgEndIndex (reversedBlockHeaders []common.BlockHeader ) (index int ) {
114123 for i := 0 ; i < len (reversedBlockHeaders )- 1 ; i ++ {
115124 currentBlock := reversedBlockHeaders [i ]
@@ -129,20 +138,21 @@ func findReorgEndIndex(reversedBlockHeaders []common.BlockHeader) (index int) {
129138 return - 1
130139}
131140
132- func (rh * ReorgHandler ) findForkPoint (reversedBlockHeaders []common.BlockHeader ) (forkPoint * big.Int , err error ) {
141+ func (rh * ReorgHandler ) findFirstForkedBlockNumber (reversedBlockHeaders []common.BlockHeader ) (forkPoint * big.Int , err error ) {
133142 newBlocksByNumber , err := rh .getNewBlocksByNumber (reversedBlockHeaders )
134143 if err != nil {
135144 return nil , err
136145 }
137146
138- for i := 0 ; i < len (reversedBlockHeaders )- 1 ; i ++ {
147+ // skip first because that is the reorg end block
148+ for i := 1 ; i < len (reversedBlockHeaders ); i ++ {
139149 blockHeader := reversedBlockHeaders [i ]
140150 block , ok := (* newBlocksByNumber )[blockHeader .Number .String ()]
141151 if ! ok {
142152 return nil , fmt .Errorf ("block not found: %s" , blockHeader .Number .String ())
143153 }
144- if block .Hash == blockHeader .Hash {
145- previousBlock := reversedBlockHeaders [i + 1 ]
154+ if blockHeader . ParentHash == block .ParentHash && blockHeader . Hash == block .Hash {
155+ previousBlock := reversedBlockHeaders [i - 1 ]
146156 return previousBlock .Number , nil
147157 }
148158 }
@@ -151,7 +161,7 @@ func (rh *ReorgHandler) findForkPoint(reversedBlockHeaders []common.BlockHeader)
151161 if err != nil {
152162 return nil , fmt .Errorf ("error getting next headers batch: %w" , err )
153163 }
154- return rh .findForkPoint (nextHeadersBatch )
164+ return rh .findFirstForkedBlockNumber (nextHeadersBatch )
155165}
156166
157167func (rh * ReorgHandler ) getNewBlocksByNumber (reversedBlockHeaders []common.BlockHeader ) (* map [string ]common.Block , error ) {
@@ -171,13 +181,15 @@ func (rh *ReorgHandler) getNewBlocksByNumber(reversedBlockHeaders []common.Block
171181}
172182
173183func (rh * ReorgHandler ) handleReorg (reorgStart * big.Int , reorgEnd * big.Int ) error {
184+ log .Debug ().Msgf ("Handling reorg from block %s to %s" , reorgStart .String (), reorgEnd .String ())
174185 blockRange := make ([]* big.Int , 0 , new (big.Int ).Sub (reorgEnd , reorgStart ).Int64 ())
175186 for i := new (big.Int ).Set (reorgStart ); i .Cmp (reorgEnd ) <= 0 ; i .Add (i , big .NewInt (1 )) {
176187 blockRange = append (blockRange , new (big.Int ).Set (i ))
177188 }
178189
179190 results := rh .worker .Run (blockRange )
180191 data := make ([]common.BlockData , 0 , len (results ))
192+ blocksToDelete := make ([]* big.Int , 0 , len (results ))
181193 for _ , result := range results {
182194 if result .Error != nil {
183195 return fmt .Errorf ("cannot fix reorg: failed block %s: %w" , result .BlockNumber .String (), result .Error )
@@ -188,10 +200,11 @@ func (rh *ReorgHandler) handleReorg(reorgStart *big.Int, reorgEnd *big.Int) erro
188200 Transactions : result .Data .Transactions ,
189201 Traces : result .Data .Traces ,
190202 })
203+ blocksToDelete = append (blocksToDelete , result .BlockNumber )
191204 }
192205 // TODO make delete and insert atomic
193- if err := rh .storage .MainStorage .DeleteBlockData (rh .rpc .GetChainID (), blockRange ); err != nil {
194- return fmt .Errorf ("error deleting data for blocks %v: %w" , blockRange , err )
206+ if err := rh .storage .MainStorage .DeleteBlockData (rh .rpc .GetChainID (), blocksToDelete ); err != nil {
207+ return fmt .Errorf ("error deleting data for blocks %v: %w" , blocksToDelete , err )
195208 }
196209 if err := rh .storage .MainStorage .InsertBlockData (& data ); err != nil {
197210 return fmt .Errorf ("error saving data to main storage: %w" , err )
0 commit comments