Snapsync "static sync" with p2p

**This is a squash batch of 23 commits:**

- Ashraf Basic snap algo .. Did something + #1925
- p2p client and looking at the syncService
- finishing p2p handler - question
- protobuf changes
- snap server implements handler methods directly
- fix snap server tests after refactor
- refactor syncer to loop over iterator, add client message handling
- p2p complete - no data exchange
- Managed to sync ~200MB data
- Fix: Trie iteration
- test describing how storage of big contracts are responded
- Storage ranges - fix unexpected contract change
- Tweak message size params, trim logs
- limit memory usage & verification
- verification bug fix
- Add sync mode p2p CLI flag (#2186)
- Return struct instead of interface
- Change flag usage string
- Remove SnapServer interface
- Revert makefile
- Minor fixes
- Update docs
- Less logs, verification fixed
- code improvements

Makefile

@@ -133,7 +133,9 @@ node1: juno-cached
 	--p2p-peers=/ip4/127.0.0.1/tcp/7777/p2p/12D3KooWLdURCjbp1D7hkXWk6ZVfcMDPtsNnPHuxoTcWXFtvrxGG \
 	--p2p-addr=/ip4/0.0.0.0/tcp/7778 \
 	--p2p-private-key="8aeffc26c3c371565dbe634c5248ae26f4fa5c33bc8f7328ac95e73fb94eaf263550f02449521f7cf64af17d248c5f170be46c06986a29803124c0819cb8fac3" \
-	--metrics-port=9091
+	--metrics-port=9091 \
+	--pprof \
+	--pprof-port=9096 \
 
 #	--p2p-peers=/ip4/127.0.0.1/tcp/7778/p2p/12D3KooWDQVMmK6cQrfFcWUoFF8Ch5vYegfwiP5Do2SFC2NAXeBk \
 

adapters/p2p2core/felt.go

@@ -31,7 +31,6 @@ func adapt(v interface{ GetElements() []byte }) *felt.Felt {
 	if v == nil || reflect.ValueOf(v).IsNil() {
 		return nil
 	}
-
 	return new(felt.Felt).SetBytes(v.GetElements())
 }
 

blockchain/blockchain.go

@@ -76,18 +76,29 @@ type Blockchain struct {
 	network  *utils.Network
 	database db.DB
 
+	snapshots []*snapshotRecord
+
 	listener EventListener
 
 	cachedPending atomic.Pointer[Pending]
 }
 
 func New(database db.DB, network *utils.Network) *Blockchain {
 	RegisterCoreTypesToEncoder()
-	return &Blockchain{
+	bc := &Blockchain{
 		database: database,
 		network:  network,
 		listener: &SelectiveListener{},
 	}
+
+	// TODO: Used only for testing though...
+	// TODO: the following is only used for snap sync, uncomment when we need it again
+	// err := bc.seedSnapshot()
+	// if err != nil {
+	// 	fmt.Printf("Error seeding snapshot %s", err)
+	// }
+
+	return bc
 }
 
 func (b *Blockchain) WithListener(listener EventListener) *Blockchain {
@@ -348,7 +359,7 @@ func (b *Blockchain) SetL1Head(update *core.L1Head) error {
 func (b *Blockchain) Store(block *core.Block, blockCommitments *core.BlockCommitments,
 	stateUpdate *core.StateUpdate, newClasses map[felt.Felt]core.Class,
 ) error {
-	return b.database.Update(func(txn db.Transaction) error {
+	err := b.database.Update(func(txn db.Transaction) error {
 		if err := verifyBlock(txn, block); err != nil {
 			return err
 		}
@@ -388,6 +399,50 @@ func (b *Blockchain) Store(block *core.Block, blockCommitments *core.BlockCommit
 		heightBin := core.MarshalBlockNumber(block.Number)
 		return txn.Set(db.ChainHeight.Key(), heightBin)
 	})
+	if err != nil {
+		return err
+	}
+
+	// TODO: the following is only used for snap sync, uncomment when we need it again
+	// err = b.seedSnapshot()
+	// if err != nil {
+	// 	return err
+	// }
+
+	return nil
+}
+
+func (b *Blockchain) StoreRaw(blockNumber uint64, stateDiff *core.StateDiff) error {
+	return b.database.Update(func(txn db.Transaction) error {
+		return core.NewState(txn).UpdateNoVerify(blockNumber, stateDiff, make(map[felt.Felt]core.Class))
+	})
+}
+
+func (b *Blockchain) PutClasses(blockNumber uint64, classHashes map[felt.Felt]*felt.Felt, newClasses map[felt.Felt]core.Class) error {
+	return b.database.Update(func(txn db.Transaction) error {
+		v1ClassHashes := map[felt.Felt]*felt.Felt{}
+		for ch, class := range newClasses {
+			if class.Version() == 1 {
+				v1ClassHashes[ch] = classHashes[ch]
+			}
+		}
+
+		return core.NewState(txn).UpdateNoVerify(blockNumber, &core.StateDiff{
+			DeclaredV1Classes: v1ClassHashes,
+		}, newClasses)
+	})
+}
+
+func (b *Blockchain) PutContracts(address, nonces, classHash []*felt.Felt) error {
+	return b.database.Update(func(txn db.Transaction) error {
+		return core.NewState(txn).UpdateContractNoLog(address, nonces, classHash)
+	})
+}
+
+func (b *Blockchain) PutStorage(storage map[felt.Felt]map[felt.Felt]*felt.Felt) error {
+	return b.database.Update(func(txn db.Transaction) error {
+		return core.NewState(txn).UpdateContractStorages(storage)
+	})
 }
 
 // VerifyBlock assumes the block has already been sanity-checked.
@@ -805,6 +860,16 @@ func (b *Blockchain) HeadState() (core.StateReader, StateCloser, error) {
 	return core.NewState(txn), txn.Discard, nil
 }
 
+func (b *Blockchain) HeadStateFreakingState() (*core.State, StateCloser, error) {
+	b.listener.OnRead("HeadState")
+	txn, err := b.database.NewTransaction(false)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return core.NewState(txn), txn.Discard, nil
+}
+
 // StateAtBlockNumber returns a StateReader that provides a stable view to the state at the given block number
 func (b *Blockchain) StateAtBlockNumber(blockNumber uint64) (core.StateReader, StateCloser, error) {
 	b.listener.OnRead("StateAtBlockNumber")

blockchain/blockchain_test.go

@@ -28,6 +28,7 @@ func TestNew(t *testing.T) {
 	gw := adaptfeeder.New(client)
 	t.Run("empty blockchain's head is nil", func(t *testing.T) {
 		chain := blockchain.New(pebble.NewMemTest(t), &utils.Mainnet)
+		defer chain.Close()
 		assert.Equal(t, &utils.Mainnet, chain.Network())
 		b, err := chain.Head()
 		assert.Nil(t, b)
@@ -43,8 +44,10 @@ func TestNew(t *testing.T) {
 		testDB := pebble.NewMemTest(t)
 		chain := blockchain.New(testDB, &utils.Mainnet)
 		assert.NoError(t, chain.Store(block0, &emptyCommitments, stateUpdate0, nil))
+		chain.Close()
 
 		chain = blockchain.New(testDB, &utils.Mainnet)
+		defer chain.Close()
 		b, err := chain.Head()
 		require.NoError(t, err)
 		assert.Equal(t, block0, b)
@@ -56,6 +59,7 @@ func TestHeight(t *testing.T) {
 	gw := adaptfeeder.New(client)
 	t.Run("return nil if blockchain is empty", func(t *testing.T) {
 		chain := blockchain.New(pebble.NewMemTest(t), &utils.Sepolia)
+		defer chain.Close()
 		_, err := chain.Height()
 		assert.Error(t, err)
 	})
@@ -69,16 +73,19 @@ func TestHeight(t *testing.T) {
 		testDB := pebble.NewMemTest(t)
 		chain := blockchain.New(testDB, &utils.Mainnet)
 		assert.NoError(t, chain.Store(block0, &emptyCommitments, stateUpdate0, nil))
+		chain.Close()
 
 		chain = blockchain.New(testDB, &utils.Mainnet)
 		height, err := chain.Height()
 		require.NoError(t, err)
 		assert.Equal(t, block0.Number, height)
+		chain.Close()
 	})
 }
 
 func TestBlockByNumberAndHash(t *testing.T) {
 	chain := blockchain.New(pebble.NewMemTest(t), &utils.Sepolia)
+	defer chain.Close()
 	t.Run("same block is returned for both GetBlockByNumber and GetBlockByHash", func(t *testing.T) {
 		client := feeder.NewTestClient(t, &utils.Mainnet)
 		gw := adaptfeeder.New(client)
@@ -115,6 +122,7 @@ func TestVerifyBlock(t *testing.T) {
 	require.NoError(t, err)
 
 	chain := blockchain.New(pebble.NewMemTest(t), &utils.Mainnet)
+	defer chain.Close()
 
 	t.Run("error if chain is empty and incoming block number is not 0", func(t *testing.T) {
 		block := &core.Block{Header: &core.Header{Number: 10}}
@@ -177,6 +185,7 @@ func TestSanityCheckNewHeight(t *testing.T) {
 	require.NoError(t, err)
 
 	chain := blockchain.New(pebble.NewMemTest(t), &utils.Mainnet)
+	defer chain.Close()
 
 	client := feeder.NewTestClient(t, &utils.Mainnet)
 
@@ -222,6 +231,7 @@ func TestStore(t *testing.T) {
 
 	t.Run("add block to empty blockchain", func(t *testing.T) {
 		chain := blockchain.New(pebble.NewMemTest(t), &utils.Mainnet)
+		defer chain.Close()
 		require.NoError(t, chain.Store(block0, &emptyCommitments, stateUpdate0, nil))
 
 		headBlock, err := chain.Head()
@@ -249,6 +259,7 @@ func TestStore(t *testing.T) {
 		require.NoError(t, err)
 
 		chain := blockchain.New(pebble.NewMemTest(t), &utils.Mainnet)
+		defer chain.Close()
 		require.NoError(t, chain.Store(block0, &emptyCommitments, stateUpdate0, nil))
 		require.NoError(t, chain.Store(block1, &emptyCommitments, stateUpdate1, nil))
 
@@ -290,6 +301,7 @@ func TestStoreL1HandlerTxnHash(t *testing.T) {
 
 func TestBlockCommitments(t *testing.T) {
 	chain := blockchain.New(pebble.NewMemTest(t), &utils.Mainnet)
+	defer chain.Close()
 	client := feeder.NewTestClient(t, &utils.Mainnet)
 	gw := adaptfeeder.New(client)
 
@@ -315,6 +327,7 @@ func TestBlockCommitments(t *testing.T) {
 
 func TestTransactionAndReceipt(t *testing.T) {
 	chain := blockchain.New(pebble.NewMemTest(t), &utils.Mainnet)
+	defer chain.Close()
 
 	client := feeder.NewTestClient(t, &utils.Mainnet)
 	gw := adaptfeeder.New(client)
@@ -403,6 +416,7 @@ func TestTransactionAndReceipt(t *testing.T) {
 func TestState(t *testing.T) {
 	testDB := pebble.NewMemTest(t)
 	chain := blockchain.New(testDB, &utils.Mainnet)
+	defer chain.Close()
 
 	client := feeder.NewTestClient(t, &utils.Mainnet)
 	gw := adaptfeeder.New(client)
@@ -466,6 +480,7 @@ func TestState(t *testing.T) {
 func TestEvents(t *testing.T) {
 	testDB := pebble.NewMemTest(t)
 	chain := blockchain.New(testDB, &utils.Goerli2)
+	defer chain.Close()
 
 	client := feeder.NewTestClient(t, &utils.Goerli2)
 	gw := adaptfeeder.New(client)
@@ -585,6 +600,7 @@ func TestEvents(t *testing.T) {
 func TestRevert(t *testing.T) {
 	testdb := pebble.NewMemTest(t)
 	chain := blockchain.New(testdb, &utils.Mainnet)
+	defer chain.Close()
 
 	client := feeder.NewTestClient(t, &utils.Mainnet)
 	gw := adaptfeeder.New(client)
@@ -673,13 +689,15 @@ func TestL1Update(t *testing.T) {
 			got, err := chain.L1Head()
 			require.NoError(t, err)
 			assert.Equal(t, head, got)
+			chain.Close()
 		})
 	}
 }
 
 func TestPending(t *testing.T) {
 	testDB := pebble.NewMemTest(t)
 	chain := blockchain.New(testDB, &utils.Mainnet)
+	defer chain.Close()
 	client := feeder.NewTestClient(t, &utils.Mainnet)
 	gw := adaptfeeder.New(client)
 
@@ -821,6 +839,7 @@ func TestPending(t *testing.T) {
 func TestStorePendingIncludesNumber(t *testing.T) {
 	network := utils.Mainnet
 	chain := blockchain.New(pebble.NewMemTest(t), &network)
+	defer chain.Close()
 
 	// Store pending genesis.
 	require.NoError(t, chain.StorePending(&blockchain.Pending{