Implement draft idea how reorg should be restructured

clients/feeder/feeder.go

@@ -21,7 +21,10 @@
 	"github.com/stretchr/testify/require"
 )
 
-var ErrDeprecatedCompiledClass = errors.New("deprecated compiled class")
+var (
+	ErrDeprecatedCompiledClass = errors.New("deprecated compiled class")
+	ErrBlockNotFound           = errors.New("block not found")
+)
 
 type Backoff func(wait time.Duration) time.Duration
 
@@ -254,7 +257,12 @@
 				if res.StatusCode == http.StatusOK {
 					return res.Body, nil
 				} else {
-					err = errors.New(res.Status)
+					if c.isBlockNotFoundError(res) {
+						err = ErrBlockNotFound
+						break
+					} else {
+						err = errors.New(res.Status)
+					}
 				}
 
 				res.Body.Close()
@@ -270,9 +278,25 @@
 			c.log.Debugw("Failed query to feeder, retrying...", "req", req.URL.String(), "retryAfter", wait.String(), "err", err)
 		}
 	}
+
 	return nil, err
 }
 
+func (c *Client) isBlockNotFoundError(res *http.Response) bool {
+	var starknetError struct {
+		Code    string `json:"code"`
+		Message string `json:"message"` // not used for now
+	}
+
+	err := json.NewDecoder(res.Body).Decode(&starknetError)
+	if err != nil {
+		c.log.Errorw("Failed to read response body", "err", err)
+		return false
+	}
+
+	return starknetError.Code == "StarknetErrorCode.BLOCK_NOT_FOUND"
+}
+
 func (c *Client) StateUpdate(ctx context.Context, blockID string) (*starknet.StateUpdate, error) {
 	queryURL := c.buildQueryString("get_state_update", map[string]string{
 		"blockNumber": blockID,

sync/sync.go

@@ -7,7 +7,9 @@
 	"sync/atomic"
 	"time"
 
+	"github.com/NethermindEth/juno/clients/feeder"
+
 	"github.com/NethermindEth/juno/blockchain"
 	"github.com/NethermindEth/juno/core"
 	"github.com/NethermindEth/juno/core/felt"
 	"github.com/NethermindEth/juno/db"
@@ -101,16 +103,23 @@
 	return nil
 }
 
-func (s *Synchronizer) fetcherTask(ctx context.Context, height uint64, verifiers *stream.Stream,
-	resetStreams context.CancelFunc,
-) stream.Callback {
+func (s *Synchronizer) fetcherTask(ctx context.Context, height uint64, verifiers *stream.Stream, resetStreams context.CancelFunc, deepReorg chan struct{}) stream.Callback {
+	emptyCallback := func() {}
+
 	for {
 		select {
 		case <-ctx.Done():
-			return func() {}
+			return emptyCallback
 		default:
 			stateUpdate, block, err := s.starknetData.StateUpdateWithBlock(ctx, height)
 			if err != nil {
+				// in case block not found we should initiate deep reorg
+				if errors.Is(err, feeder.ErrBlockNotFound) {
+					resetStreams()
+					deepReorg <- struct{}{}
+					return emptyCallback
+				}
+
 				continue
 			}
 
@@ -266,6 +275,8 @@
 	pendingSem := make(chan struct{}, 1)
 	go s.pollPending(syncCtx, pendingSem)
 
+	// rough marker that deep reorg in progress, todo replace with smth. else
+	deepReorg := make(chan struct{}, 1)
 	for {
 		select {
 		case <-streamCtx.Done():
@@ -278,6 +289,13 @@
 				pendingSem <- struct{}{}
 				latestSem <- struct{}{}
 				return
+			case <-deepReorg:
+				// 1. fetch current height from feeder by using block=latest
+				// 2. revert all blocks till the block where it diverged
+				// 3. mark deep reorg as done
+				// 4. restart syncing process
+				// to discuss: do we need to remove "small" reorg in verifyier logic ?
+				// potential speedup: set catchUpMode before restarting sync
 			default:
 				streamCtx, streamCancel = context.WithCancel(syncCtx)
 				nextHeight = s.nextHeight()
@@ -288,7 +306,7 @@
 			curHeight, curStreamCtx, curCancel := nextHeight, streamCtx, streamCancel
 			fetchers.Go(func() stream.Callback {
 				fetchTimer := time.Now()
-				cb := s.fetcherTask(curStreamCtx, curHeight, verifiers, curCancel)
+				cb := s.fetcherTask(curStreamCtx, curHeight, verifiers, curCancel, deepReorg)
 				s.listener.OnSyncStepDone(OpFetch, curHeight, time.Since(fetchTimer))
 				return cb
 			})