go/worker/storage: Remove legacy storagy sync p2p protocol

Author: martintomazic

.changelog/6261.internal.md

@@ -0,0 +1,5 @@
+go/worker/storage: Remove legacy storage sync p2p protocol
+
+The node will now only server new checkpoint and diff sync
+p2p protocols. This is backward compatible since legacy
+clients still serve both legacy and new protocols.

go/oasis-test-runner/scenario/e2e/runtime/checkpoint_sync.go

@@ -342,7 +342,6 @@ func RegisterScenarios() error {
 		StorageSyncFromRegistered,
 		StorageSyncInconsistent,
 		StorageEarlyStateSync,
-		CheckpointSync,
 		// Sentry test.
 		Sentry,
 		// Keymanager tests.

go/oasis-test-runner/scenario/e2e/runtime/scenario.go

@@ -11,11 +11,9 @@ import (
 	"sync"
 	"time"
 
-	"github.com/oasisprotocol/oasis-core/go/p2p/rpc"
 	storageApi "github.com/oasisprotocol/oasis-core/go/storage/api"
 	"github.com/oasisprotocol/oasis-core/go/storage/mkvs/checkpoint"
 	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/checkpointsync"
-	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/synclegacy"
 )
 
 const (
@@ -103,7 +101,19 @@ func (n *Node) checkpointChunkFetcher(
 		defer cancel()
 
 		// Fetch chunk from peers.
-		rsp, pf, err := n.fetchChunk(chunkCtx, chunk)
+		rsp, pf, err := n.checkpointSync.GetCheckpointChunk(
+			chunkCtx,
+			&checkpointsync.GetCheckpointChunkRequest{
+				Version: chunk.Version,
+				Root:    chunk.Root,
+				Index:   chunk.Index,
+				Digest:  chunk.Digest,
+			},
+			&checkpointsync.Checkpoint{
+				Metadata: chunk.checkpoint.Metadata,
+				Peers:    chunk.checkpoint.Peers,
+			},
+		)
 		if err != nil {
 			n.logger.Error("failed to fetch chunk from peers",
 				"err", err,
@@ -114,7 +124,7 @@ func (n *Node) checkpointChunkFetcher(
 		}
 
 		// Restore fetched chunk.
-		done, err := n.localStorage.Checkpointer().RestoreChunk(chunkCtx, chunk.Index, bytes.NewBuffer(rsp))
+		done, err := n.localStorage.Checkpointer().RestoreChunk(chunkCtx, chunk.Index, bytes.NewBuffer(rsp.Chunk))
 		cancel()
 
 		switch {
@@ -154,46 +164,6 @@ func (n *Node) checkpointChunkFetcher(
 	}
 }
 
-// fetchChunk fetches chunk using checkpoint sync p2p protocol client.
-//
-// In case of no peers or error, it fallbacks to the legacy storage sync protocol.
-func (n *Node) fetchChunk(ctx context.Context, chunk *chunk) ([]byte, rpc.PeerFeedback, error) {
-	rsp1, pf, err := n.checkpointSync.GetCheckpointChunk(
-		ctx,
-		&checkpointsync.GetCheckpointChunkRequest{
-			Version: chunk.Version,
-			Root:    chunk.Root,
-			Index:   chunk.Index,
-			Digest:  chunk.Digest,
-		},
-		&checkpointsync.Checkpoint{
-			Metadata: chunk.checkpoint.Metadata,
-			Peers:    chunk.checkpoint.Peers,
-		},
-	)
-	if err == nil { // if NO error
-		return rsp1.Chunk, pf, nil
-	}
-
-	rsp2, pf, err := n.legacyStorageSync.GetCheckpointChunk(
-		ctx,
-		&synclegacy.GetCheckpointChunkRequest{
-			Version: chunk.Version,
-			Root:    chunk.Root,
-			Index:   chunk.Index,
-			Digest:  chunk.Digest,
-		},
-		&synclegacy.Checkpoint{
-			Metadata: chunk.checkpoint.Metadata,
-			Peers:    chunk.checkpoint.Peers,
-		},
-	)
-	if err != nil {
-		return nil, nil, err
-	}
-	return rsp2.Chunk, pf, nil
-}
-
 func (n *Node) handleCheckpoint(check *checkpointsync.Checkpoint, maxParallelRequests uint) (cpStatus int, rerr error) {
 	if err := n.localStorage.Checkpointer().StartRestore(n.ctx, check.Metadata); err != nil {
 		// Any previous restores were already aborted by the driver up the call stack, so
@@ -317,7 +287,9 @@ func (n *Node) getCheckpointList() ([]*checkpointsync.Checkpoint, error) {
 	ctx, cancel := context.WithTimeout(n.ctx, cpListsTimeout)
 	defer cancel()
 
-	list, err := n.fetchCheckpoints(ctx)
+	list, err := n.checkpointSync.GetCheckpoints(ctx, &checkpointsync.GetCheckpointsRequest{
+		Version: 1,
+	})
 	if err != nil {
 		n.logger.Error("failed to retrieve any checkpoints",
 			"err", err,
@@ -331,33 +303,6 @@ func (n *Node) getCheckpointList() ([]*checkpointsync.Checkpoint, error) {
 	return list, nil
 }
 
-// fetchCheckpoints fetches checkpoints using checkpoint sync p2p protocol client.
-//
-// In case of no peers or error, it fallbacks to the legacy storage sync protocol.
-func (n *Node) fetchCheckpoints(ctx context.Context) ([]*checkpointsync.Checkpoint, error) {
-	list1, err := n.checkpointSync.GetCheckpoints(ctx, &checkpointsync.GetCheckpointsRequest{
-		Version: 1,
-	})
-	if err == nil && len(list1) > 0 { // if NO error and at least one checkpoint
-		return list1, nil
-	}
-
-	list2, err := n.legacyStorageSync.GetCheckpoints(ctx, &synclegacy.GetCheckpointsRequest{
-		Version: 1,
-	})
-	if err != nil {
-		return nil, err
-	}
-	var cps []*checkpointsync.Checkpoint
-	for _, cp := range list2 {
-		cps = append(cps, &checkpointsync.Checkpoint{
-			Metadata: cp.Metadata,
-			Peers:    cp.Peers,
-		})
-	}
-	return cps, nil
-}
-
 // sortCheckpoints sorts the slice in-place (descending by version, peers, hash).
 func sortCheckpoints(s []*checkpointsync.Checkpoint) {
 	slices.SortFunc(s, func(a, b *checkpointsync.Checkpoint) int {

go/worker/storage/committee/checkpoint_sync.go

@@ -35,7 +35,6 @@ import (
 	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/checkpointsync"
 	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/diffsync"
 	storagePub "github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/pub"
-	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/synclegacy"
 )
 
 var (
@@ -130,9 +129,8 @@ type Node struct { // nolint: maligned
 
 	localStorage storageApi.LocalBackend
 
-	diffSync          diffsync.Client
-	checkpointSync    checkpointsync.Client
-	legacyStorageSync synclegacy.Client
+	diffSync       diffsync.Client
+	checkpointSync checkpointsync.Client
 
 	undefinedRound uint64
 
@@ -226,7 +224,6 @@ func NewNode(
 	})
 
 	// Advertise and serve p2p protocols.
-	commonNode.P2P.RegisterProtocolServer(synclegacy.NewServer(commonNode.ChainContext, commonNode.Runtime.ID(), localStorage))
 	commonNode.P2P.RegisterProtocolServer(diffsync.NewServer(commonNode.ChainContext, commonNode.Runtime.ID(), localStorage))
 	if config.GlobalConfig.Storage.Checkpointer.Enabled {
 		commonNode.P2P.RegisterProtocolServer(checkpointsync.NewServer(commonNode.ChainContext, commonNode.Runtime.ID(), localStorage))
@@ -236,7 +233,6 @@ func NewNode(
 	}
 
 	// Create p2p protocol clients.
-	n.legacyStorageSync = synclegacy.NewClient(commonNode.P2P, commonNode.ChainContext, commonNode.Runtime.ID())
 	n.diffSync = diffsync.NewClient(commonNode.P2P, commonNode.ChainContext, commonNode.Runtime.ID())
 	n.checkpointSync = checkpointsync.NewClient(commonNode.P2P, commonNode.ChainContext, commonNode.Runtime.ID())
 
@@ -444,29 +440,13 @@ func (n *Node) fetchDiff(round uint64, prevRoot, thisRoot storageApi.Root) {
 	ctx, cancel := context.WithCancel(n.ctx)
 	defer cancel()
 
-	wl, pf, err := n.getDiff(ctx, prevRoot, thisRoot)
+	rsp, pf, err := n.diffSync.GetDiff(ctx, &diffsync.GetDiffRequest{StartRoot: prevRoot, EndRoot: thisRoot})
 	if err != nil {
 		result.err = err
 		return
 	}
 	result.pf = pf
-	result.writeLog = wl
-}
-
-// getDiff fetches writelog using diff sync p2p protocol client.
-//
-// In case of no peers or error, it fallbacks to the legacy storage sync protocol.
-func (n *Node) getDiff(ctx context.Context, prevRoot, thisRoot storageApi.Root) (storageApi.WriteLog, rpc.PeerFeedback, error) {
-	rsp1, pf, err := n.diffSync.GetDiff(ctx, &diffsync.GetDiffRequest{StartRoot: prevRoot, EndRoot: thisRoot})
-	if err == nil { // if NO error
-		return rsp1.WriteLog, pf, nil
-	}
-
-	rsp2, pf, err := n.legacyStorageSync.GetDiff(ctx, &synclegacy.GetDiffRequest{StartRoot: prevRoot, EndRoot: thisRoot})
-	if err != nil {
-		return nil, nil, err
-	}
-	return rsp2.WriteLog, pf, nil
+	result.writeLog = rsp.WriteLog
 }
 
 func (n *Node) finalize(summary *blockSummary) {