fix(responsemanager): fix flaky tests

Author: hannahhoward

responsemanager/client.go

@@ -224,7 +224,12 @@ func (rm *ResponseManager) GetUpdates(p peer.ID, requestID graphsync.RequestID,
 
 // FinishTask marks a task from the task queue as done
 func (rm *ResponseManager) FinishTask(task *peertask.Task, err error) {
-	rm.send(&finishTaskRequest{task, err}, nil)
+	done := make(chan struct{}, 1)
+	rm.send(&finishTaskRequest{task, err, done}, nil)
+	select {
+	case <-rm.ctx.Done():
+	case <-done:
+	}
 }
 
 // CloseWithNetworkError closes a request due to a network error

responsemanager/messages.go

@@ -85,10 +85,15 @@ func (rur *responseUpdateRequest) handle(rm *ResponseManager) {
 type finishTaskRequest struct {
 	task *peertask.Task
 	err  error
+	done chan struct{}
 }
 
 func (ftr *finishTaskRequest) handle(rm *ResponseManager) {
 	rm.finishTask(ftr.task, ftr.err)
+	select {
+	case <-rm.ctx.Done():
+	case ftr.done <- struct{}{}:
+	}
 }
 
 type startTaskRequest struct {

responsemanager/responsemanager_test.go

@@ -291,6 +291,7 @@ func TestValidationAndExtensions(t *testing.T) {
 		// request fails with base loader reading from block store that's missing data
 		responseManager.ProcessRequests(td.ctx, td.p, td.requests)
 		td.assertCompleteRequestWith(graphsync.RequestFailedContentNotFound)
+		td.taskqueue.WaitForNoActiveTasks()
 
 		err := td.peristenceOptions.Register("chainstore", td.persistence)
 		require.NoError(t, err)
@@ -632,6 +633,8 @@ func TestValidationAndExtensions(t *testing.T) {
 				td.verifyNResponses(blockCount)
 				td.assertPausedRequest()
 
+				td.taskqueue.WaitForNoActiveTasks()
+
 				// send update
 				responseManager.ProcessRequests(td.ctx, td.p, td.updateRequests)