Feat/port fixes from v13.3

messagequeue/messagequeue.go

@@ -248,6 +248,11 @@ func (mq *MessageQueue) sendMessage() {
 		log.Infof("cant open message sender to peer %s: %s", mq.p, err)
 		// TODO: cant connect, what now?
 		mq.publishError(metadata, fmt.Errorf("cant open message sender to peer %s: %w", mq.p, err))
+		select {
+		case <-mq.done:
+		default:
+			mq.Shutdown()
+		}
 		return
 	}
 

requestmanager/client.go

@@ -191,6 +191,8 @@ func (rm *RequestManager) NewRequest(ctx context.Context,
 	select {
 	case <-rm.ctx.Done():
 		return rm.emptyResponse()
+	case <-ctx.Done():
+		return rm.emptyResponse()
 	case receivedInProgressRequest = <-inProgressRequestChan:
 	}
 
@@ -284,6 +286,8 @@ func (rm *RequestManager) CancelRequest(ctx context.Context, requestID graphsync
 	select {
 	case <-rm.ctx.Done():
 		return errors.New("context cancelled")
+	case <-ctx.Done():
+		return ctx.Err()
 	case err := <-terminated:
 		return err
 	}
@@ -306,6 +310,8 @@ func (rm *RequestManager) UnpauseRequest(ctx context.Context, requestID graphsyn
 	select {
 	case <-rm.ctx.Done():
 		return errors.New("context cancelled")
+	case <-ctx.Done():
+		return ctx.Err()
 	case err := <-response:
 		return err
 	}
@@ -318,6 +324,8 @@ func (rm *RequestManager) PauseRequest(ctx context.Context, requestID graphsync.
 	select {
 	case <-rm.ctx.Done():
 		return errors.New("context cancelled")
+	case <-ctx.Done():
+		return ctx.Err()
 	case err := <-response:
 		return err
 	}
@@ -330,6 +338,8 @@ func (rm *RequestManager) UpdateRequest(ctx context.Context, requestID graphsync
 	select {
 	case <-rm.ctx.Done():
 		return errors.New("context cancelled")
+	case <-ctx.Done():
+		return ctx.Err()
 	case err := <-response:
 		return err
 	}

requestmanager/requestmanager_test.go

@@ -181,6 +181,21 @@ func TestCancelRequestImperativeNoMoreBlocks(t *testing.T) {
 	require.True(t, ok)
 }
 
+func TestCommandsWithCancelledContext(t *testing.T) {
+	ctx := context.Background()
+	managerCtx, managerCancel := context.WithCancel(ctx)
+	defer managerCancel()
+	td := newTestData(managerCtx, t)
+	peers := testutil.GeneratePeers(1)
+
+	cancelledCtx, cancel := context.WithCancel(ctx)
+	cancel()
+
+	returnedResponseChan, returnedErrorChan := td.requestManager.NewRequest(cancelledCtx, peers[0], td.blockChain.TipLink, td.blockChain.Selector())
+	testutil.VerifyEmptyErrors(managerCtx, t, returnedErrorChan)
+	testutil.VerifyEmptyResponse(managerCtx, t, returnedResponseChan)
+}
+
 func TestCancelManagerExitsGracefully(t *testing.T) {
 	ctx := context.Background()
 	managerCtx, managerCancel := context.WithCancel(ctx)

responsemanager/client.go

@@ -166,6 +166,8 @@ func (rm *ResponseManager) UnpauseResponse(ctx context.Context, requestID graphs
 	select {
 	case <-rm.ctx.Done():
 		return errors.New("context cancelled")
+	case <-ctx.Done():
+		return ctx.Err()
 	case err := <-response:
 		return err
 	}
@@ -178,6 +180,8 @@ func (rm *ResponseManager) PauseResponse(ctx context.Context, requestID graphsyn
 	select {
 	case <-rm.ctx.Done():
 		return errors.New("context cancelled")
+	case <-ctx.Done():
+		return ctx.Err()
 	case err := <-response:
 		return err
 	}
@@ -190,6 +194,8 @@ func (rm *ResponseManager) CancelResponse(ctx context.Context, requestID graphsy
 	select {
 	case <-rm.ctx.Done():
 		return errors.New("context cancelled")
+	case <-ctx.Done():
+		return ctx.Err()
 	case err := <-response:
 		return err
 	}
@@ -202,6 +208,8 @@ func (rm *ResponseManager) UpdateResponse(ctx context.Context, requestID graphsy
 	select {
 	case <-rm.ctx.Done():
 		return errors.New("context cancelled")
+	case <-ctx.Done():
+		return ctx.Err()
 	case err := <-response:
 		return err
 	}