tests

storagemarket/network/libp2p_impl.go

@@ -65,8 +65,8 @@ func (impl *libp2pStorageMarketNetwork) NewDealStatusStream(ctx context.Context,
 
 func (impl *libp2pStorageMarketNetwork) openStream(ctx context.Context, id peer.ID, protocol protocol.ID) (network.Stream, error) {
 	b := &backoff.Backoff{
-		Min:    50 * time.Millisecond,
-		Max:    6 * time.Second,
+		Min:    100 * time.Millisecond,
+		Max:    10 * time.Second,
 		Factor: 5,
 		Jitter: true,
 	}

storagemarket/network/libp2p_impl_test.go

@@ -43,6 +43,39 @@ func (tr *testReceiver) HandleDealStatusStream(s network.DealStatusStream) {
 	}
 }
 
+func TestOpenStreamWithRetries(t *testing.T) {
+	ctx := context.Background()
+	td := shared_testutil.NewLibp2pTestData(ctx, t)
+
+	fromNetwork := network.NewFromLibp2pHost(td.Host1)
+	toNetwork := network.NewFromLibp2pHost(td.Host2)
+	toHost := td.Host2.ID()
+
+	// host1 gets no-op receiver
+	tr := &testReceiver{t: t}
+	require.NoError(t, fromNetwork.SetDelegate(tr))
+
+	// host2 gets a receiver that will start after some time -> so we can verify exponential backoff kicks in
+	require.NoError(t, td.Host2.Close())
+	achan := make(chan network.AskRequest)
+	tr2 := &testReceiver{t: t, askStreamHandler: func(s network.StorageAskStream) {
+		readq, err := s.ReadAskRequest()
+		require.NoError(t, err)
+		achan <- readq
+	}}
+	go func() {
+		select {
+		case _ = <-time.After(3 * time.Second):
+			toNetwork.SetDelegate(tr2)
+		case <-ctx.Done():
+			return
+		}
+	}()
+
+	// setup query stream host1 --> host 2
+	assertAskRequestReceived(ctx, t, fromNetwork, toHost, achan)
+}
+
 func TestAskStreamSendReceiveAskRequest(t *testing.T) {
 	ctx := context.Background()
 	td := shared_testutil.NewLibp2pTestData(ctx, t)