L2 Data Migration Exploration

celo_migration_test_datadir/celo/chaindata/CURRENT

@@ -0,0 +1 @@
+MANIFEST-000000

celo_migration_test_datadir/celo/chaindata/LOG

@@ -0,0 +1,6 @@
+=============== Feb 21, 2024 (EST) ===============
+00:35:42.679318 log@legend F·NumFile S·FileSize N·Entry C·BadEntry B·BadBlock Ke·KeyError D·DroppedEntry L·Level Q·SeqNum T·TimeElapsed
+00:35:42.698146 db@open opening
+00:35:42.699133 version@stat F·[] S·0B[] Sc·[]
+00:35:42.702693 db@janitor F·2 G·0
+00:35:42.702728 db@open done T·4.557125ms

celo_migration_test_datadir/celo/nodes/CURRENT

@@ -0,0 +1 @@
+MANIFEST-000000

celo_migration_test_datadir/celo/nodes/LOG

@@ -0,0 +1,6 @@
+=============== Feb 21, 2024 (EST) ===============
+00:35:42.742714 log@legend F·NumFile S·FileSize N·Entry C·BadEntry B·BadBlock Ke·KeyError D·DroppedEntry L·Level Q·SeqNum T·TimeElapsed
+00:35:42.756416 db@open opening
+00:35:42.756815 version@stat F·[] S·0B[] Sc·[]
+00:35:42.763625 db@janitor F·2 G·0
+00:35:42.763648 db@open done T·7.208291ms

celo_migration_test_datadir/keystore/UTC--2024-02-21T05-35-42.765873000Z--dc1c520f9be93efadb20d7f6568238f90d41ce9f

@@ -0,0 +1 @@
+{"address":"dc1c520f9be93efadb20d7f6568238f90d41ce9f","privatekey":"77b44b86363a599a06216b7c777b51e9becf1c9737d15c5b8e2c042351919724","id":"3603a13c-8ca7-4e9e-9a79-68333b6482e3","version":3}

e2e_test/e2e_test.go

@@ -12,6 +12,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/celo-org/celo-blockchain"
 	"github.com/celo-org/celo-blockchain/common"
 	"github.com/celo-org/celo-blockchain/common/hexutil"
 	"github.com/celo-org/celo-blockchain/core/types"
@@ -40,6 +41,102 @@ func init() {
 	log.Root().SetHandler(log.DiscardHandler())
 }
 
+func TestSubscribeMergedLogs(t *testing.T) {
+	ac := test.AccountConfig(1, 2)
+	gingerbreadBlock := common.Big1
+	gc, ec, err := test.BuildConfig(ac, gingerbreadBlock)
+	require.NoError(t, err)
+	network, shutdown, err := test.NewNetwork(ac, gc, ec)
+	require.NoError(t, err)
+	defer shutdown()
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*300)
+	defer cancel()
+
+	accounts := test.Accounts(ac.DeveloperAccounts(), gc.ChainConfig())
+
+	// Send one celo from external account 0 to 1 via node 0.
+	tx, err := accounts[0].SendCeloViaGoldToken(ctx, accounts[1].Address, 1, network[0])
+	require.NoError(t, err)
+
+	ch := make(chan types.Log)
+	c, err := rpc.DialContext(ctx, network[0].WSEndpoint())
+	require.NoError(t, err)
+	c.EthSubscribe(ctx, ch, "mergedLogs", map[string]interface{}{
+		"fromBlock": hexutil.EncodeBig(big.NewInt(0)),
+		"toBlock":   hexutil.EncodeBig(big.NewInt(1000)),
+	})
+
+	for len(ch) > 0 {
+		log := <-ch
+		fmt.Println(log)
+	}
+
+	// Wait for the whole network to process the transaction.
+	err = network.AwaitTransactions(ctx, tx)
+	require.NoError(t, err)
+}
+
+func TestSubscribeLogs(t *testing.T) {
+	ac := test.AccountConfig(3, 2)
+	gingerbreadBlock := common.Big1
+	gc, ec, err := test.BuildConfig(ac, gingerbreadBlock)
+	require.NoError(t, err)
+	network, shutdown, err := test.NewNetwork(ac, gc, ec)
+	require.NoError(t, err)
+	defer shutdown()
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
+
+	accounts := test.Accounts(ac.DeveloperAccounts(), gc.ChainConfig())
+
+	// Send one celo from external account 0 to 1 via node 0.
+	tx, err := accounts[0].SendCelo(ctx, accounts[1].Address, 1, network[0])
+	require.NoError(t, err)
+
+	ch := make(chan types.Log)
+	network[0].WsClient.SubscribeFilterLogs(ctx, celo.FilterQuery{
+		FromBlock: big.NewInt(0),
+		ToBlock:   big.NewInt(1000),
+	}, ch)
+	for len(ch) > 0 {
+		<-ch
+	}
+	// Wait for the whole network to process the transaction.
+	err = network.AwaitTransactions(ctx, tx)
+	require.NoError(t, err)
+}
+
+// TODO(Alec)
+func TestMigrateDataDir(t *testing.T) {
+	ac := test.AccountConfig(1, 2)
+	gingerbreadBlock := big.NewInt(20)
+	gc, ec, err := test.BuildConfig(ac, gingerbreadBlock)
+	require.NoError(t, err)
+	// gc.Hardforks = genesis.HardforkConfig{
+	// 	ChurritoBlock:      big.NewInt(5),
+	// 	DonutBlock:         big.NewInt(10),
+	// 	EspressoBlock:      big.NewInt(15),
+	// 	GingerbreadBlock:   gingerbreadBlock,
+	// 	GingerbreadP2Block: big.NewInt(25),
+	// }
+	network, _, err := test.NewNetwork(ac, gc, ec)
+	require.NoError(t, err)
+	// defer shutdown()
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*100)
+	defer cancel()
+
+	accounts := test.Accounts(ac.DeveloperAccounts(), gc.ChainConfig())
+
+	// Send one celo from external account 0 to 1 via node 0.
+	tx, err := accounts[0].SendCelo(ctx, accounts[1].Address, 1, network[0])
+	require.NoError(t, err)
+
+	err = network.AwaitTransactions(ctx, tx)
+	require.NoError(t, err)
+	err = network.AwaitBlock(ctx, 30)
+	require.NoError(t, err)
+}
+
 // This test starts a network submits a transaction and waits for the whole
 // network to process the transaction.
 func TestSendCelo(t *testing.T) {

eth/filters/api.go

@@ -249,6 +249,76 @@ func (api *PublicFilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, er
 	return rpcSub, nil
 }
 
+// TODO(Alec)
+func (api *PublicFilterAPI) MergedLogs(ctx context.Context, crit FilterCriteria) (*rpc.Subscription, error) {
+	notifier, supported := rpc.NotifierFromContext(ctx)
+	if !supported {
+		return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
+	}
+
+	var (
+		rpcSub      = notifier.CreateSubscription()
+		matchedLogs = make(chan []*types.Log)
+	)
+
+	migrationBlock := big.NewInt(500) // last block of cel1, migrationBlock + 1 is first block of cel2
+
+	// TODO(Alec) handle cases where FromBlock and ToBlock are negative or nil
+	// TODO(Alec) there seems to be a mismatch between the interpretation of nil From blocks in the
+	// FilterQuery type definition and how they're handled by SubscribeLogs.
+	// TODO(Alec) assuming FromBlock and ToBlock are inclusive
+
+	queries := [2]ethereum.FilterQuery{
+		ethereum.FilterQuery(crit),
+	}
+	if queries[0].FromBlock.Cmp(queries[0].ToBlock) > 0 {
+		return nil, fmt.Errorf("invalid from and to block combination: from > to") // Error copied from SubscribeLogs
+	}
+	if queries[0].FromBlock.Cmp(migrationBlock) <= 0 {
+		if queries[0].ToBlock.Cmp(migrationBlock) > 0 {
+			queries = [2]ethereum.FilterQuery{
+				{
+					FromBlock: queries[0].FromBlock,
+					ToBlock:   migrationBlock,
+				},
+				{
+					FromBlock: migrationBlock.Add(migrationBlock, big.NewInt(1)),
+					ToBlock:   queries[0].ToBlock,
+				},
+			}
+		}
+	}
+
+	go func() {
+		for _, query := range queries {
+			logsSub, err := api.events.SubscribeLogs(query, matchedLogs)
+			if err != nil {
+				log.Error("Error returned from SubscribeLogs", err)
+				// TODO(Alec) what else do we need to do here?
+				return
+			}
+
+			for {
+				select {
+				case logs := <-matchedLogs:
+					for _, log := range logs {
+						notifier.Notify(rpcSub.ID, &log) // TODO(Alec) maybe you can send an error through here
+					}
+					// TODO(Alec) how can we break this loop when matchedLogs is empty to continue to next query?
+				case <-rpcSub.Err(): // client send an unsubscribe request
+					logsSub.Unsubscribe()
+					return
+				case <-notifier.Closed(): // connection dropped
+					logsSub.Unsubscribe()
+					return
+				}
+			}
+		}
+	}()
+
+	return rpcSub, nil
+}
+
 // Logs creates a subscription that fires for all new log that match the given filter criteria.
 func (api *PublicFilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subscription, error) {
 	notifier, supported := rpc.NotifierFromContext(ctx)

eth/filters/filter_system.go

@@ -184,6 +184,7 @@ func (es *EventSystem) subscribe(sub *subscription) *Subscription {
 	return &Subscription{ID: sub.id, f: sub, es: es}
 }
 
+// TODO(Alec)
 // SubscribeLogs creates a subscription that will write all logs matching the
 // given criteria to the given logs channel. Default value for the from and to
 // block is "latest". If the fromBlock > toBlock an error is returned.

rpc/handler.go

@@ -316,6 +316,7 @@ func (h *handler) handleCallMsg(ctx *callProc, msg *jsonrpcMessage) *jsonrpcMess
 
 // handleCall processes method calls.
 func (h *handler) handleCall(cp *callProc, msg *jsonrpcMessage) *jsonrpcMessage {
+	// TODO(Alec) line between two wordls
 	if msg.isSubscribe() {
 		return h.handleSubscribe(cp, msg)
 	}

rpc/service.go

@@ -59,7 +59,7 @@ type callback struct {
 }
 
 func (r *serviceRegistry) registerName(name string, rcvr interface{}) error {
-	rcvrVal := reflect.ValueOf(rcvr)
+	rcvrVal := reflect.ValueOf(rcvr) // 1
 	if name == "" {
 		return fmt.Errorf("no service name for type %s", rcvrVal.Type().String())
 	}
@@ -180,6 +180,7 @@ func (c *callback) makeArgTypes() {
 	}
 }
 
+// TODO(Alec) This is where RPC method is actually called
 // call invokes the callback.
 func (c *callback) call(ctx context.Context, method string, args []reflect.Value) (res interface{}, errRes error) {
 	// Create the argument slice.