@@ -209,7 +209,7 @@ func TestRoundTrip(t *testing.T) {
209209 for opens < 2 || completes < 2 || len (sentIncrements ) < 21 || len (receivedIncrements ) < 21 {
210210 select {
211211 case <- ctx .Done ():
212- t .Fatal ("Did not complete succcessful data transfer" )
212+ t .Fatal ("Did not complete successful data transfer" )
213213 case <- finished :
214214 completes ++
215215 case <- opened :
@@ -343,7 +343,7 @@ func TestMultipleRoundTripMultipleStores(t *testing.T) {
343343 for opens < 2 * data .requestCount || completes < 2 * data .requestCount {
344344 select {
345345 case <- ctx .Done ():
346- t .Fatal ("Did not complete succcessful data transfer" )
346+ t .Fatal ("Did not complete successful data transfer" )
347347 case <- finished :
348348 completes ++
349349 case <- opened :
@@ -481,7 +481,7 @@ func TestManyReceiversAtOnce(t *testing.T) {
481481 for opens < 2 * data .receiverCount || completes < 2 * data .receiverCount {
482482 select {
483483 case <- ctx .Done ():
484- t .Fatal ("Did not complete succcessful data transfer" )
484+ t .Fatal ("Did not complete successful data transfer" )
485485 case <- finished :
486486 completes ++
487487 case <- opened :
@@ -497,6 +497,101 @@ func TestManyReceiversAtOnce(t *testing.T) {
497497 }
498498}
499499
500+ // TestPushRequestAutoRestart tests that if the connection for a push request
501+ // goes down, it will automatically restart (given the right config options)
502+ func TestPushRequestAutoRestart (t * testing.T ) {
503+ ctx := context .Background ()
504+ ctx , cancel := context .WithTimeout (ctx , 10 * time .Second )
505+ defer cancel ()
506+
507+ gsData := testutil .NewGraphsyncTestingData (ctx , t , nil , nil )
508+ netRetry := network .RetryParameters (time .Second , time .Second , 5 , 1 )
509+ gsData .DtNet1 = network .NewFromLibp2pHost (gsData .Host1 , netRetry )
510+ host1 := gsData .Host1 // initiator, data sender
511+ host2 := gsData .Host2 // data recipient
512+
513+ tp1 := gsData .SetupGSTransportHost1 ()
514+ tp2 := gsData .SetupGSTransportHost2 ()
515+
516+ restartConf := PushChannelRestartConfig (100 * time .Millisecond , 1 , 10 , 200 * time .Millisecond )
517+ dt1 , err := NewDataTransfer (gsData .DtDs1 , gsData .TempDir1 , gsData .DtNet1 , tp1 , gsData .StoredCounter1 , restartConf )
518+ require .NoError (t , err )
519+ testutil .StartAndWaitForReady (ctx , t , dt1 )
520+ dt2 , err := NewDataTransfer (gsData .DtDs2 , gsData .TempDir2 , gsData .DtNet2 , tp2 , gsData .StoredCounter2 )
521+ require .NoError (t , err )
522+ testutil .StartAndWaitForReady (ctx , t , dt2 )
523+
524+ received := make (chan struct {})
525+ finished := make (chan struct {}, 2 )
526+ var subscriber datatransfer.Subscriber = func (event datatransfer.Event , channelState datatransfer.ChannelState ) {
527+ //t.Logf("%s: %s\n", datatransfer.Events[event.Code], datatransfer.Statuses[channelState.Status()])
528+
529+ if event .Code == datatransfer .DataReceived {
530+ received <- struct {}{}
531+ }
532+
533+ if channelState .Status () == datatransfer .Completed {
534+ finished <- struct {}{}
535+ }
536+ }
537+ dt1 .SubscribeToEvents (subscriber )
538+ dt2 .SubscribeToEvents (subscriber )
539+ voucher := testutil.FakeDTType {Data : "applesauce" }
540+ sv := testutil .NewStubbedValidator ()
541+
542+ sourceDagService := gsData .DagService1
543+ destDagService := gsData .DagService2
544+
545+ root , origBytes := testutil .LoadUnixFSFile (ctx , t , sourceDagService , loremFile )
546+ rootCid := root .(cidlink.Link ).Cid
547+
548+ require .NoError (t , dt1 .RegisterVoucherType (& testutil.FakeDTType {}, sv ))
549+ require .NoError (t , dt2 .RegisterVoucherType (& testutil.FakeDTType {}, sv ))
550+ chid , err := dt1 .OpenPushDataChannel (ctx , host2 .ID (), & voucher , rootCid , gsData .AllSelector )
551+ require .NoError (t , err )
552+
553+ // Wait for a block to be received
554+ <- received
555+
556+ // Break connection
557+ t .Logf ("Breaking connection to peer" )
558+ require .NoError (t , gsData .Mn .UnlinkPeers (host1 .ID (), host2 .ID ()))
559+ require .NoError (t , gsData .Mn .DisconnectPeers (host1 .ID (), host2 .ID ()))
560+
561+ t .Logf ("Sleep for a second" )
562+ time .Sleep (1 * time .Second )
563+
564+ // Restore connection
565+ t .Logf ("Restore connection" )
566+ require .NoError (t , gsData .Mn .LinkAll ())
567+ time .Sleep (200 * time .Millisecond )
568+ conn , err := gsData .Mn .ConnectPeers (host1 .ID (), host2 .ID ())
569+ require .NoError (t , err )
570+ require .NotNil (t , conn )
571+
572+ t .Logf ("Waiting for auto-restart on push channel %s" , chid )
573+
574+ (func () {
575+ finishedCount := 0
576+ for {
577+ select {
578+ case <- ctx .Done ():
579+ t .Fatal ("Did not complete successful data transfer" )
580+ return
581+ case <- received :
582+ case <- finished :
583+ finishedCount ++
584+ if finishedCount == 2 {
585+ return
586+ }
587+ }
588+ }
589+ })()
590+
591+ // Verify that the file was transferred to the destination node
592+ testutil .VerifyHasFile (ctx , t , destDagService , root , origBytes )
593+ }
594+
500595func TestRoundTripCancelledRequest (t * testing.T ) {
501596 ctx := context .Background ()
502597 testCases := map [string ]struct {
@@ -751,7 +846,7 @@ func TestSimulatedRetrievalFlow(t *testing.T) {
751846 for providerFinished != nil || clientFinished != nil {
752847 select {
753848 case <- ctx .Done ():
754- t .Fatal ("Did not complete succcessful data transfer" )
849+ t .Fatal ("Did not complete successful data transfer" )
755850 case <- providerFinished :
756851 providerFinished = nil
757852 case <- clientFinished :
@@ -868,7 +963,7 @@ func TestPauseAndResume(t *testing.T) {
868963 pauseInitiators < 1 || pauseResponders < 1 || resumeInitiators < 1 || resumeResponders < 1 {
869964 select {
870965 case <- ctx .Done ():
871- t .Fatal ("Did not complete succcessful data transfer" )
966+ t .Fatal ("Did not complete successful data transfer" )
872967 case <- finished :
873968 completes ++
874969 case <- opened :
@@ -968,7 +1063,7 @@ func TestUnrecognizedVoucherRoundTrip(t *testing.T) {
9681063 for opens < 1 || finishes < 1 {
9691064 select {
9701065 case <- ctx .Done ():
971- t .Fatal ("Did not complete succcessful data transfer" )
1066+ t .Fatal ("Did not complete successful data transfer" )
9721067 case <- finished :
9731068 finishes ++
9741069 case <- opened :
0 commit comments