WIP

bridgesync/bridgesync.go

@@ -22,7 +22,7 @@ var (
 	maxRetryAttemptsAfterError = 5
 )
 
-type LocalBridgeSync struct {
+type BridgeSync struct {
 	processor *processor
 	driver    *sync.EVMDriver
 }
@@ -37,7 +37,7 @@ func NewL1(
 	rd sync.ReorgDetector,
 	ethClient EthClienter,
 	initialBlock uint64,
-) (*LocalBridgeSync, error) {
+) (*BridgeSync, error) {
 	return new(
 		ctx,
 		dbPath,
@@ -62,7 +62,7 @@ func NewL2(
 	rd sync.ReorgDetector,
 	ethClient EthClienter,
 	initialBlock uint64,
-) (*LocalBridgeSync, error) {
+) (*BridgeSync, error) {
 	return new(
 		ctx,
 		dbPath,
@@ -87,7 +87,7 @@ func new(
 	ethClient EthClienter,
 	initialBlock uint64,
 	dbPrefix, reorgDetectorID string,
-) (*LocalBridgeSync, error) {
+) (*BridgeSync, error) {
 	processor, err := newProcessor(ctx, dbPath, dbPrefix)
 	if err != nil {
 		return nil, err
@@ -130,15 +130,19 @@ func new(
 	if err != nil {
 		return nil, err
 	}
-	return &LocalBridgeSync{
+	return &BridgeSync{
 		processor: processor,
 		driver:    driver,
 	}, nil
 }
 
 // Start starts the synchronization process
-func (s *LocalBridgeSync) Start(ctx context.Context) {
+func (s *BridgeSync) Start(ctx context.Context) {
 	s.driver.Sync(ctx)
 }
 
+func (s *BridgeSync) GetLastProcessedBlock(ctx context.Context) (uint64, error) {
+	return s.GetLastProcessedBlock(ctx)
+}
+
 // TODO: expose methods from the processor for consumers

l1bridge2infoindexsync/downloader.go

@@ -0,0 +1,67 @@
+package l1bridge2infoindexsync
+
+import (
+	"context"
+	"errors"
+	"math/big"
+
+	"github.com/0xPolygon/cdk/bridgesync"
+	"github.com/0xPolygon/cdk/l1infotreesync"
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+type downloader struct {
+	l1Bridge *bridgesync.BridgeSync
+	l1Info   *l1infotreesync.L1InfoTreeSync
+	l1Client ethereum.ChainReader
+}
+
+func newDownloader(
+	l1Bridge *bridgesync.BridgeSync,
+	l1Info *l1infotreesync.L1InfoTreeSync,
+	l1Client ethereum.ChainReader,
+) *downloader {
+	return &downloader{
+		l1Bridge: l1Bridge,
+		l1Info:   l1Info,
+		l1Client: l1Client,
+	}
+}
+
+func (d *downloader) getLastFinalisedL1Block(ctx context.Context) (uint64, error) {
+	b, err := d.l1Client.BlockByNumber(ctx, big.NewInt(int64(rpc.FinalizedBlockNumber)))
+	if err != nil {
+		return 0, err
+	}
+	return b.NumberU64(), nil
+}
+
+func (d *downloader) getLastProcessedBlockBridge(ctx context.Context) (uint64, error) {
+	return d.l1Bridge.GetLastProcessedBlock(ctx)
+}
+
+func (d *downloader) getLastProcessedBlockL1InfoTree(ctx context.Context) (uint64, error) {
+	return d.l1Info.GetLastProcessedBlock(ctx)
+}
+
+func (d *downloader) getLastL1InfoIndexUntilBlock(ctx context.Context, blockNum uint64) (uint32, error) {
+	info, err := d.l1Info.GetLatestInfoUntilBlock(ctx, blockNum)
+	if err != nil {
+		return 0, err
+	}
+	return info.L1InfoTreeIndex, nil
+}
+
+func (d *downloader) getMainnetExitRootAtL1InfoTreeIndex(ctx context.Context, index uint32) (common.Hash, error) {
+	leaf, err := d.l1Info.GetInfoByIndex(ctx, index)
+	if err != nil {
+		return common.Hash{}, err
+	}
+	return leaf.MainnetExitRoot, nil
+}
+
+func (d *downloader) getBridgeIndex(ctx context.Context, mainnetExitRoot common.Hash) (uint32, error) {
+	return 0, errors.New("need to implement root -> index @ bridgesync")
+}

l1bridge2infoindexsync/driver.go

@@ -0,0 +1,114 @@
+package l1bridge2infoindexsync
+
+import (
+	"context"
+
+	"github.com/0xPolygon/cdk/log"
+)
+
+type driver struct {
+	downloader *downloader
+	processor  *processor
+}
+
+func newDriver(
+	downloader *downloader,
+	processor *processor) *driver {
+	return &driver{
+		downloader: downloader,
+		processor:  processor,
+	}
+}
+
+func (d *driver) sync(ctx context.Context) {
+	lpbProcessor, err := d.processor.GetLastProcessedBlock(ctx)
+	if err != nil {
+		log.Fatal("TODO")
+	}
+	lastProcessedL1InfoIndex, err := d.processor.getLastL1InfoTreeIndexProcessed(ctx)
+	if err != nil {
+		log.Fatal("TODO")
+	}
+	for {
+		syncUntilBlock, shouldWait, err := d.getTargetSynchronizationBlock(ctx, lpbProcessor)
+		if err != nil {
+			log.Fatal("TODO")
+		}
+		if shouldWait {
+			// TODO: wait using ticker
+			continue
+		}
+		lastL1InfoTreeIndex, err := d.downloader.getLastL1InfoIndexUntilBlock(ctx, syncUntilBlock)
+		if err != nil {
+			log.Fatal("TODO")
+		}
+		relations := []bridge2L1InfoRelation{}
+		for i := lastProcessedL1InfoIndex + 1; i <= lastL1InfoTreeIndex; i++ {
+			relation, err := d.getRelation(ctx, i)
+			if err != nil {
+				log.Fatal("TODO")
+			}
+			relations = append(relations, relation)
+		}
+		if err := d.processor.addBridge2L1InfoRelations(ctx, syncUntilBlock, relations); err != nil {
+			log.Fatal("TODO")
+		}
+		lpbProcessor = syncUntilBlock
+		lastProcessedL1InfoIndex = 0 // TODO
+	}
+}
+
+func (d *driver) getTargetSynchronizationBlock(ctx context.Context, lpbProcessor uint64) (syncUntilBlock uint64, shouldWait bool, err error) {
+	lastFinalised, err := d.downloader.getLastFinalisedL1Block(ctx) // TODO: configure finality, but then we need to deal with reorgs?
+	if err != nil {
+		return
+	}
+	if lpbProcessor >= lastFinalised {
+		shouldWait = true
+		return
+	}
+	lpbInfo, err := d.downloader.getLastProcessedBlockL1InfoTree(ctx)
+	if err != nil {
+		return
+	}
+	if lpbProcessor >= lpbInfo {
+		shouldWait = true
+		return
+	}
+	lpbBridge, err := d.downloader.getLastProcessedBlockBridge(ctx)
+	if err != nil {
+		return
+	}
+	if lpbProcessor >= lpbBridge {
+		shouldWait = true
+		return
+	}
+
+	// Bridge, L1Info and L1 ahead of procesor. Pick the smallest block num as target
+	if lastFinalised <= lpbInfo {
+		syncUntilBlock = lastFinalised
+	} else {
+		syncUntilBlock = lpbInfo
+	}
+	if lpbBridge < syncUntilBlock {
+		syncUntilBlock = lpbBridge
+	}
+	return
+}
+
+func (d *driver) getRelation(ctx context.Context, l1InfoIndex uint32) (bridge2L1InfoRelation, error) {
+	mer, err := d.downloader.getMainnetExitRootAtL1InfoTreeIndex(ctx, l1InfoIndex)
+	if err != nil {
+		return bridge2L1InfoRelation{}, err
+	}
+
+	bridgeIndex, err := d.downloader.getBridgeIndex(ctx, mer)
+	if err != nil {
+		return bridge2L1InfoRelation{}, err
+	}
+
+	return bridge2L1InfoRelation{
+		bridgeIndex:     bridgeIndex,
+		l1InfoTreeIndex: l1InfoIndex,
+	}, nil
+}

l1bridge2infoindexsync/l1bridge2infoindexsync.go

@@ -0,0 +1,15 @@
+package l1bridge2infoindexsync
+
+import (
+	"context"
+	"errors"
+)
+
+type L1Bridge2InfoIndexSync struct {
+}
+
+func New() (*L1Bridge2InfoIndexSync, error) {
+	return nil, errors.New("not implemented")
+}
+
+func (s *L1Bridge2InfoIndexSync) Start(ctx context.Context) {}

l1bridge2infoindexsync/processor.go

@@ -0,0 +1,58 @@
+package l1bridge2infoindexsync
+
+import (
+	"context"
+	"errors"
+
+	"github.com/0xPolygon/cdk/common"
+	"github.com/ledgerwatch/erigon-lib/kv"
+)
+
+var (
+	lastBlokcKey = []byte("lb")
+)
+
+type processor struct {
+	db             kv.RwDB
+	lastBlockTable string
+}
+
+type bridge2L1InfoRelation struct {
+	bridgeIndex     uint32
+	l1InfoTreeIndex uint32
+}
+
+// GetLastProcessedBlock returns the last processed block oby the processor, including blocks
+// that don't have events
+func (p *processor) GetLastProcessedBlock(ctx context.Context) (uint64, error) {
+	tx, err := p.db.BeginRo(ctx)
+	if err != nil {
+		return 0, err
+	}
+	defer tx.Rollback()
+	return p.getLastProcessedBlockWithTx(tx)
+}
+
+func (p *processor) getLastProcessedBlockWithTx(tx kv.Tx) (uint64, error) {
+	if blockNumBytes, err := tx.GetOne(p.lastBlockTable, lastBlokcKey); err != nil {
+		return 0, err
+	} else if blockNumBytes == nil {
+		return 0, nil
+	} else {
+		return common.BytesToUint64(blockNumBytes), nil
+	}
+}
+
+func (p *processor) getLastL1InfoTreeIndexProcessed(ctx context.Context) (uint32, error) {
+	return 0, errors.New("not implemented")
+}
+
+func (p *processor) updateLastProcessedBlock(ctx context.Context, blockNum uint64) error {
+	return errors.New("not implemented")
+}
+
+func (p *processor) addBridge2L1InfoRelations(ctx context.Context, lastProcessedBlcok uint64, relations []bridge2L1InfoRelation) error {
+	// Note that indexes could be repeated as the L1 Info tree update can be produced by a rollup and not mainnet.
+	// Hence if the index already exist, do not update as it's better to have the lowest indes possible for the relation
+	return errors.New("not implemented")
+}