Fixup: Fix synchronization issues in the test

Maybe I will replace this with e2e alltogether.
To be consistent I should probably extend other
p2p clients. On the other hand I dislike methods
for testing only.

Author: martintomazic

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

@@ -33,6 +33,9 @@ type Client interface {
 		request *GetCheckpointChunkRequest,
 		cp *Checkpoint,
 	) (*GetCheckpointChunkResponse, rpc.PeerFeedback, error)
+
+	// IsReady is true when protocol client is aware of at least one remote peer.
+	IsReady() bool
 }
 
 // Checkpoint contains checkpoint metadata together with peer information.
@@ -114,6 +117,10 @@ func (c *client) getBestPeers(opts ...rpc.BestPeersOption) []core.PeerID {
 	return append(c.mgr.GetBestPeers(opts...), c.fallbackMgr.GetBestPeers(opts...)...)
 }
 
+func (c *client) IsReady() bool {
+	return len(c.getBestPeers()) > 0
+}
+
 // NewClient creates a new checkpoint sync protocol client.
 //
 // Previously, it was part of the storage sync protocol. To enable seamless rolling

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

@@ -3,6 +3,7 @@ package diffsync
 import (
 	"context"
 
+	"github.com/libp2p/go-libp2p/core"
 	"github.com/oasisprotocol/oasis-core/go/common"
 	"github.com/oasisprotocol/oasis-core/go/p2p/protocol"
 	"github.com/oasisprotocol/oasis-core/go/p2p/rpc"
@@ -23,6 +24,9 @@ type Client interface {
 	// GetDiff requests a write log of entries that must be applied to get from the first given root
 	// to the second one.
 	GetDiff(ctx context.Context, request *GetDiffRequest) (*GetDiffResponse, rpc.PeerFeedback, error)
+
+	// IsReady is true when protocol client is aware of at least one remote peer.
+	IsReady() bool
 }
 
 type client struct {
@@ -33,8 +37,7 @@ type client struct {
 
 func (c *client) GetDiff(ctx context.Context, request *GetDiffRequest) (*GetDiffResponse, rpc.PeerFeedback, error) {
 	var rsp GetDiffResponse
-	peers := append(c.mgr.GetBestPeers(), c.fallbackMgr.GetBestPeers()...)
-	pf, err := c.rc.CallOne(ctx, peers, MethodGetDiff, request, &rsp,
+	pf, err := c.rc.CallOne(ctx, c.getBestPeers(), MethodGetDiff, request, &rsp,
 		rpc.WithMaxPeerResponseTime(MaxGetDiffResponseTime),
 	)
 	if err != nil {
@@ -43,6 +46,14 @@ func (c *client) GetDiff(ctx context.Context, request *GetDiffRequest) (*GetDiff
 	return &rsp, pf, nil
 }
 
+func (c *client) getBestPeers(opts ...rpc.BestPeersOption) []core.PeerID {
+	return append(c.mgr.GetBestPeers(opts...), c.fallbackMgr.GetBestPeers(opts...)...)
+}
+
+func (c *client) IsReady() bool {
+	return len(c.getBestPeers()) > 0
+}
+
 // NewClient creates a new diff sync protocol client.
 //
 // Previously, it was part of the storage sync protocol. To enable seamless rolling

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

@@ -117,8 +117,8 @@ func testLegacyHostClient(ctx context.Context, t *testing.T, host p2pApi.Service
 	require := require.New(t)
 
 	client := synclegacy.NewClient(host, chainContext, runtimeID)
-	time.Sleep(2 * time.Second)
-
+	err := waitReady(ctx, client.IsReady)
+	require.NoError(err, "Wait for p2p client readiness")
 	// Test diff part of the storagesync protocol.
 	rsp, _, err := client.GetDiff(ctx, &synclegacy.GetDiffRequest{})
 	require.NoError(err, "Fetch storage diff from p2p")
@@ -146,15 +146,19 @@ func testNewClients(ctx context.Context, t *testing.T, host p2pApi.Service, back
 
 	// Test diff sync protocol.
 	diffClient := diffsync.NewClient(host, chainContext, runtimeID)
-	time.Sleep(2 * time.Second)
+	err := waitReady(ctx, diffClient.IsReady)
+	require.NoError(err, "Wait for p2p client readiness")
+
 	rsp2, _, err := diffClient.GetDiff(ctx, &diffsync.GetDiffRequest{})
 	require.NoError(err, "Fetch storage diff from p2p")
 	err = assertEqualGetDiffResponse(ctx, backend, rsp2.WriteLog)
 	require.NoError(err, "Assert expected storage diff response")
 
 	// Test checkpoint sync protocol.
 	cpsClient := checkpointsync.NewClient(host, chainContext, runtimeID)
-	time.Sleep(2 * time.Second)
+	err = waitReady(ctx, cpsClient.IsReady)
+	require.NoError(err, "Wait for p2p client readiness")
+
 	cps, err := cpsClient.GetCheckpoints(ctx, &checkpointsync.GetCheckpointsRequest{
 		Version: 1,
 	})
@@ -307,6 +311,27 @@ func getAvailablePort() (int, error) {
 	return addr.Port, nil
 }
 
+func waitReady(ctx context.Context, isReady func() bool) error {
+	ctx, cancel := context.WithTimeout(ctx, 1*time.Minute)
+	defer cancel()
+
+	backoff := 20 * time.Millisecond
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+
+		if isReady() {
+			return nil
+		}
+
+		time.Sleep(backoff)
+		backoff *= 2
+	}
+}
+
 type backendMock struct{}
 
 func (bm *backendMock) SyncGet(context.Context, *syncer.GetRequest) (*syncer.ProofResponse, error) {

go/worker/storage/p2p/synclegacy/client.go

@@ -36,6 +36,9 @@ type Client interface {
 		request *GetCheckpointChunkRequest,
 		cp *Checkpoint,
 	) (*GetCheckpointChunkResponse, rpc.PeerFeedback, error)
+
+	// IsReady is true when protocol client is aware of at least one remote peer.
+	IsReady() bool
 }
 
 // Checkpoint contains checkpoint metadata together with peer information.
@@ -129,6 +132,10 @@ func GetStorageSyncProtocolID(chainContext string, runtimeID common.Namespace) c
 	return protocol.NewRuntimeProtocolID(chainContext, runtimeID, StorageSyncProtocolID, StorageSyncProtocolVersion)
 }
 
+func (c *client) IsReady() bool {
+	return len(c.mgrC.GetBestPeers()) > 0
+}
+
 // NewClient creates a new storage sync protocol client.
 func NewClient(p2p rpc.P2P, chainContext string, runtimeID common.Namespace) Client {
 	// Use two separate clients and managers for the same protocol. This is to make sure that peers