go/worker/storage: Integrate new p2p protocols

Legacy storage sync protocol is still advertised and served
to enable seamless rolling upgrades of the network.

Author: martintomazic

.changelog/6262.feature.md

@@ -0,0 +1,12 @@
+go: Split storage sync p2p protocol
+
+Storage sync protocol was split into two independent protocols (checkpoint
+and diff sync).
+
+This change was made since there may be fewer nodes that expose checkpoints
+than storage diff. Previously, this could lead to issues with state sync
+when a node was connected with peers that supported storage sync protocol
+but had no checkpoints available.
+
+This was done in backwards compatible manner, so that both protocols are still
+advertised and used. Eventually, we plan to remove legacy protocol.

go/oasis-node/cmd/debug/byzantine/node.go

@@ -22,7 +22,7 @@ import (
 	scheduler "github.com/oasisprotocol/oasis-core/go/scheduler/api"
 	storage "github.com/oasisprotocol/oasis-core/go/storage/api"
 	"github.com/oasisprotocol/oasis-core/go/worker/client"
-	storageP2P "github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/sync"
+	storageP2P "github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/synclegacy"
 )
 
 type byzantine struct {

go/worker/storage/committee/checkpoint_sync.go

@@ -13,7 +13,7 @@ import (
 
 	storageApi "github.com/oasisprotocol/oasis-core/go/storage/api"
 	"github.com/oasisprotocol/oasis-core/go/storage/mkvs/checkpoint"
-	storageSync "github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/sync"
+	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/checkpointsync"
 )
 
 const (
@@ -55,7 +55,7 @@ type chunk struct {
 	*checkpoint.ChunkMetadata
 
 	// checkpoint points to the checkpoint this chunk originated from.
-	checkpoint *storageSync.Checkpoint
+	checkpoint *checkpointsync.Checkpoint
 }
 
 type chunkHeap struct {
@@ -101,7 +101,7 @@ func (n *Node) checkpointChunkFetcher(
 		defer cancel()
 
 		// Fetch chunk from peers.
-		rsp, pf, err := n.storageSync.GetCheckpointChunk(chunkCtx, &storageSync.GetCheckpointChunkRequest{
+		rsp, pf, err := n.checkpointSync.GetCheckpointChunk(chunkCtx, &checkpointsync.GetCheckpointChunkRequest{
 			Version: chunk.Version,
 			Root:    chunk.Root,
 			Index:   chunk.Index,
@@ -157,7 +157,7 @@ func (n *Node) checkpointChunkFetcher(
 	}
 }
 
-func (n *Node) handleCheckpoint(check *storageSync.Checkpoint, maxParallelRequests uint) (cpStatus int, rerr error) {
+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
 		// things should have been going smoothly here; bail.
@@ -276,11 +276,11 @@ func (n *Node) handleCheckpoint(check *storageSync.Checkpoint, maxParallelReques
 	}
 }
 
-func (n *Node) getCheckpointList() ([]*storageSync.Checkpoint, error) {
+func (n *Node) getCheckpointList() ([]*checkpointsync.Checkpoint, error) {
 	ctx, cancel := context.WithTimeout(n.ctx, cpListsTimeout)
 	defer cancel()
 
-	list, err := n.storageSync.GetCheckpoints(ctx, &storageSync.GetCheckpointsRequest{
+	list, err := n.checkpointSync.GetCheckpoints(ctx, &checkpointsync.GetCheckpointsRequest{
 		Version: 1,
 	})
 	if err != nil {
@@ -297,8 +297,8 @@ func (n *Node) getCheckpointList() ([]*storageSync.Checkpoint, error) {
 }
 
 // sortCheckpoints sorts the slice in-place (descending by version, peers, hash).
-func sortCheckpoints(s []*storageSync.Checkpoint) {
-	slices.SortFunc(s, func(a, b *storageSync.Checkpoint) int {
+func sortCheckpoints(s []*checkpointsync.Checkpoint) {
+	slices.SortFunc(s, func(a, b *checkpointsync.Checkpoint) int {
 		return cmp.Or(
 			cmp.Compare(b.Root.Version, a.Root.Version),
 			cmp.Compare(len(b.Peers), len(a.Peers)),
@@ -307,7 +307,7 @@ func sortCheckpoints(s []*storageSync.Checkpoint) {
 	})
 }
 
-func (n *Node) checkCheckpointUsable(cp *storageSync.Checkpoint, remainingMask outstandingMask, genesisRound uint64) bool {
+func (n *Node) checkCheckpointUsable(cp *checkpointsync.Checkpoint, remainingMask outstandingMask, genesisRound uint64) bool {
 	namespace := n.commonNode.Runtime.ID()
 	if !namespace.Equal(&cp.Root.Namespace) {
 		// Not for the right runtime.
@@ -357,7 +357,7 @@ func (n *Node) syncCheckpoints(genesisRound uint64, wantOnlyGenesis bool) (*bloc
 
 	// If we only want the genesis checkpoint, filter it out.
 	if wantOnlyGenesis && len(cps) > 0 {
-		var filteredCps []*storageSync.Checkpoint
+		var filteredCps []*checkpointsync.Checkpoint
 		for _, cp := range cps {
 			if cp.Root.Version == genesisRound {
 				filteredCps = append(filteredCps, cp)

go/worker/storage/committee/checkpoint_sync_test.go

@@ -8,35 +8,35 @@ import (
 	"github.com/oasisprotocol/oasis-core/go/p2p/rpc"
 	"github.com/oasisprotocol/oasis-core/go/storage/mkvs/checkpoint"
 	"github.com/oasisprotocol/oasis-core/go/storage/mkvs/node"
-	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/sync"
+	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/checkpointsync"
 )
 
 func TestSortCheckpoints(t *testing.T) {
-	cp1 := &sync.Checkpoint{
+	cp1 := &checkpointsync.Checkpoint{
 		Metadata: &checkpoint.Metadata{
 			Root: node.Root{
 				Version: 2,
 			},
 		},
 		Peers: []rpc.PeerFeedback{rpc.NewNopPeerFeedback(), rpc.NewNopPeerFeedback()},
 	}
-	cp2 := &sync.Checkpoint{
+	cp2 := &checkpointsync.Checkpoint{
 		Metadata: &checkpoint.Metadata{
 			Root: node.Root{
 				Version: 2,
 			},
 		},
 		Peers: []rpc.PeerFeedback{rpc.NewNopPeerFeedback()},
 	}
-	cp3 := &sync.Checkpoint{
+	cp3 := &checkpointsync.Checkpoint{
 		Metadata: &checkpoint.Metadata{
 			Root: node.Root{
 				Version: 1,
 			},
 		},
 		Peers: []rpc.PeerFeedback{rpc.NewNopPeerFeedback(), rpc.NewNopPeerFeedback()},
 	}
-	cp4 := &sync.Checkpoint{
+	cp4 := &checkpointsync.Checkpoint{
 		Metadata: &checkpoint.Metadata{
 			Root: node.Root{
 				Version: 1,
@@ -45,9 +45,9 @@ func TestSortCheckpoints(t *testing.T) {
 		Peers: []rpc.PeerFeedback{rpc.NewNopPeerFeedback()},
 	}
 
-	s := []*sync.Checkpoint{cp2, cp3, cp4, cp1}
+	s := []*checkpointsync.Checkpoint{cp2, cp3, cp4, cp1}
 
 	sortCheckpoints(s)
 
-	assert.Equal(t, s, []*sync.Checkpoint{cp1, cp2, cp3, cp4})
+	assert.Equal(t, s, []*checkpointsync.Checkpoint{cp1, cp2, cp3, cp4})
 }

go/worker/storage/committee/node.go

@@ -32,8 +32,10 @@ import (
 	"github.com/oasisprotocol/oasis-core/go/worker/common/committee"
 	"github.com/oasisprotocol/oasis-core/go/worker/registration"
 	"github.com/oasisprotocol/oasis-core/go/worker/storage/api"
+	"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"
-	storageSync "github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/sync"
+	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/synclegacy"
 )
 
 var (
@@ -128,7 +130,8 @@ type Node struct { // nolint: maligned
 
 	localStorage storageApi.LocalBackend
 
-	storageSync storageSync.Client
+	diffSync       diffsync.Client
+	checkpointSync checkpointsync.Client
 
 	undefinedRound uint64
 
@@ -272,9 +275,23 @@ func NewNode(
 		node:   n,
 	})
 
-	// Register storage sync service.
-	commonNode.P2P.RegisterProtocolServer(storageSync.NewServer(commonNode.ChainContext, commonNode.Runtime.ID(), localStorage))
-	n.storageSync = storageSync.NewClient(commonNode.P2P, commonNode.ChainContext, commonNode.Runtime.ID())
+	// Advertise and serve legacy storage sync protocol.
+	// TODO #6270 will move advertisement part out of the protocol registration.
+	commonNode.P2P.RegisterProtocolServer(synclegacy.NewServer(commonNode.ChainContext, commonNode.Runtime.ID(), localStorage))
+	commonNode.P2P.RegisterProtocol(synclegacy.GetStorageSyncProtocolID(commonNode.ChainContext, commonNode.Runtime.ID()), 5, 10)
+
+	// Register diff sync protocol server.
+	commonNode.P2P.RegisterProtocolServer(diffsync.NewServer(commonNode.ChainContext, commonNode.Runtime.ID(), localStorage))
+	// Register checkpoint sync protocol server if checkpoints are enabled.
+	if checkInterval < checkpoint.CheckIntervalDisabled {
+		commonNode.P2P.RegisterProtocolServer(checkpointsync.NewServer(commonNode.ChainContext, commonNode.Runtime.ID(), localStorage))
+	}
+
+	// Create diff and checkpoint sync p2p protocol clients that have a fallback to the old legacy storage sync protocol.
+	// TODO: This automatically starts advertising the given protocol even if the server is not registered.
+	//       #6270 wil fix that by moving advertisement out of the client creation.
+	n.diffSync = diffsync.NewClient(commonNode.P2P, commonNode.ChainContext, commonNode.Runtime.ID())
+	n.checkpointSync = checkpointsync.NewClient(commonNode.P2P, commonNode.ChainContext, commonNode.Runtime.ID())
 
 	// Register storage pub service if configured.
 	if rpcRoleProvider != nil {
@@ -430,7 +447,7 @@ func (n *Node) fetchDiff(round uint64, prevRoot, thisRoot storageApi.Root) {
 	ctx, cancel := context.WithCancel(n.ctx)
 	defer cancel()
 
-	rsp, pf, err := n.storageSync.GetDiff(ctx, &storageSync.GetDiffRequest{StartRoot: prevRoot, EndRoot: thisRoot})
+	rsp, pf, err := n.diffSync.GetDiff(ctx, &diffsync.GetDiffRequest{StartRoot: prevRoot, EndRoot: thisRoot})
 	if err != nil {
 		result.err = err
 		return

go/worker/storage/p2p/checkpointsync/client.go

@@ -10,7 +10,7 @@ import (
 	"github.com/oasisprotocol/oasis-core/go/p2p/protocol"
 	"github.com/oasisprotocol/oasis-core/go/p2p/rpc"
 	"github.com/oasisprotocol/oasis-core/go/storage/mkvs/checkpoint"
-	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/sync"
+	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/synclegacy"
 )
 
 const (
@@ -125,7 +125,7 @@ func (c *client) getBestPeers(opts ...rpc.BestPeersOption) []core.PeerID {
 // separately.
 func NewClient(p2p rpc.P2P, chainContext string, runtimeID common.Namespace) Client {
 	pid := protocol.NewRuntimeProtocolID(chainContext, runtimeID, CheckpointSyncProtocolID, CheckpointSyncProtocolVersion)
-	fallbackPid := sync.GetStorageSyncProtocolID(chainContext, runtimeID)
+	fallbackPid := synclegacy.GetStorageSyncProtocolID(chainContext, runtimeID)
 	rc := rpc.NewClient(p2p.Host(), pid, fallbackPid)
 	mgr := rpc.NewPeerManager(p2p, pid)
 	rc.RegisterListener(mgr)

go/worker/storage/p2p/checkpointsync/protocol.go

@@ -1,5 +1,5 @@
 // Package checkpointsync defines wire protocol together with client/server
-// implementations for the checkpoints sync protocol, used for runtime state sync.
+// implementations for the checkpoint sync protocol, used for runtime state sync.
 package checkpointsync
 
 import (

go/worker/storage/p2p/diffsync/client.go

@@ -6,7 +6,7 @@ import (
 	"github.com/oasisprotocol/oasis-core/go/common"
 	"github.com/oasisprotocol/oasis-core/go/p2p/protocol"
 	"github.com/oasisprotocol/oasis-core/go/p2p/rpc"
-	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/sync"
+	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/synclegacy"
 )
 
 const (
@@ -54,7 +54,7 @@ func (c *client) GetDiff(ctx context.Context, request *GetDiffRequest) (*GetDiff
 // separately.
 func NewClient(p2p rpc.P2P, chainContext string, runtimeID common.Namespace) Client {
 	pid := protocol.NewRuntimeProtocolID(chainContext, runtimeID, DiffSyncProtocolID, DiffSyncProtocolVersion)
-	fallbackPid := sync.GetStorageSyncProtocolID(chainContext, runtimeID)
+	fallbackPid := synclegacy.GetStorageSyncProtocolID(chainContext, runtimeID)
 	rc := rpc.NewClient(p2p.Host(), pid, fallbackPid)
 	mgr := rpc.NewPeerManager(p2p, pid)
 	rc.RegisterListener(mgr)

go/worker/storage/p2p/sync_test/interoperable_test.go

@@ -30,7 +30,7 @@ import (
 	"github.com/oasisprotocol/oasis-core/go/storage/mkvs/writelog"
 	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/checkpointsync"
 	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/diffsync"
-	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/sync"
+	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/synclegacy"
 )
 
 var (
@@ -116,18 +116,18 @@ func test(t *testing.T, dataDir string, legacyHost bool, peerKind peerKind) {
 func testLegacyHostClient(ctx context.Context, t *testing.T, host p2pApi.Service, backend storageApi.Backend) {
 	require := require.New(t)
 
-	client := sync.NewClient(host, chainContext, runtimeID)
+	client := synclegacy.NewClient(host, chainContext, runtimeID)
 	time.Sleep(2 * time.Second)
 
 	// Test diff part of the storagesync protocol.
-	rsp, _, err := client.GetDiff(ctx, &sync.GetDiffRequest{})
+	rsp, _, err := client.GetDiff(ctx, &synclegacy.GetDiffRequest{})
 	require.NoError(err, "Fetch storage diff from p2p")
 
 	err = assertEqualGetDiffResponse(ctx, backend, rsp.WriteLog)
 	require.NoError(err, "Assert expected storage diff response")
 
 	// Test checkpoints part of the storagesync protocol.
-	cps, err := client.GetCheckpoints(ctx, &sync.GetCheckpointsRequest{
+	cps, err := client.GetCheckpoints(ctx, &synclegacy.GetCheckpointsRequest{
 		Version: 1,
 	})
 	require.NoError(err, "Fetch checkpoints from p2p")
@@ -136,7 +136,7 @@ func testLegacyHostClient(ctx context.Context, t *testing.T, host p2pApi.Service
 		Namespace: runtimeID,
 	})
 	require.NoError(err, "Fetch expected storage diff from backend")
-	getMeta := func(cp *sync.Checkpoint) *checkpoint.Metadata { return cp.Metadata }
+	getMeta := func(cp *synclegacy.Checkpoint) *checkpoint.Metadata { return cp.Metadata }
 	err = assertEqualCheckpoints(cps, want, getMeta)
 	require.NoError(err, "Assert expected checkpoints response")
 }
@@ -278,10 +278,10 @@ func mustStartNewPeer(t *testing.T, dataDir string, id int, backend storageApi.B
 
 	switch kind {
 	case legacy:
-		serverLegacy := sync.NewServer(chainContext, runtimeID, backend)
+		serverLegacy := synclegacy.NewServer(chainContext, runtimeID, backend)
 		p2p.RegisterProtocolServer(serverLegacy)
 	case all:
-		serverLegacy := sync.NewServer(chainContext, runtimeID, backend)
+		serverLegacy := synclegacy.NewServer(chainContext, runtimeID, backend)
 		p2p.RegisterProtocolServer(serverLegacy)
 		diff := diffsync.NewServer(chainContext, runtimeID, backend)
 		p2p.RegisterProtocolServer(diff)

go/worker/storage/p2p/sync/client.go …/worker/storage/p2p/synclegacy/client.gogo/worker/storage/p2p/sync/client.go renamed to go/worker/storage/p2p/synclegacy/client.go go/worker/storage/p2p/sync/client.go renamed to go/worker/storage/p2p/synclegacy/client.go

@@ -1,4 +1,4 @@
-package sync
+package synclegacy
 
 import (
 	"context"