fixup! feat(graphsync): unify req & resp Pause, Unpause & Cancel by RequestID

Author: rvagg

requestmanager/server.go

@@ -233,7 +233,7 @@ func (rm *RequestManager) cancelRequest(requestID graphsync.RequestID, onTermina
 	if !ok {
 		if onTerminated != nil {
 			select {
-			case onTerminated <- graphsync.RequestNotFoundErr{}:
+			case onTerminated <- &graphsync.RequestNotFoundErr{}:
 			case <-rm.ctx.Done():
 			}
 		}
@@ -380,7 +380,7 @@ func (rm *RequestManager) validateRequest(requestID graphsync.RequestID, p peer.
 func (rm *RequestManager) unpause(id graphsync.RequestID, extensions []graphsync.ExtensionData) error {
 	inProgressRequestStatus, ok := rm.inProgressRequestStatuses[id]
 	if !ok {
-		return graphsync.RequestNotFoundErr{}
+		return &graphsync.RequestNotFoundErr{}
 	}
 	if inProgressRequestStatus.state != graphsync.Paused {
 		return errors.New("request is not paused")
@@ -394,7 +394,7 @@ func (rm *RequestManager) unpause(id graphsync.RequestID, extensions []graphsync
 func (rm *RequestManager) pause(id graphsync.RequestID) error {
 	inProgressRequestStatus, ok := rm.inProgressRequestStatuses[id]
 	if !ok {
-		return graphsync.RequestNotFoundErr{}
+		return &graphsync.RequestNotFoundErr{}
 	}
 	if inProgressRequestStatus.state == graphsync.Paused {
 		return errors.New("request is already paused")

responsemanager/responsemanager_test.go

@@ -218,22 +218,33 @@ func TestStats(t *testing.T) {
 	responseManager := td.nullTaskQueueResponseManager()
 	td.requestHooks.Register(selectorvalidator.SelectorValidator(100))
 	responseManager.Startup()
-	responseManager.ProcessRequests(td.ctx, td.p, td.requests)
+
+	p1 := td.p
+	reqid1 := td.requestID
+	req1 := td.requests
+
 	p2 := testutil.GeneratePeers(1)[0]
-	responseManager.ProcessRequests(td.ctx, p2, td.requests)
-	peerState := responseManager.PeerState(td.p)
+	reqid2 := graphsync.NewRequestID()
+	req2 := []gsmsg.GraphSyncRequest{
+		gsmsg.NewRequest(reqid2, td.blockChain.TipLink.(cidlink.Link).Cid, td.blockChain.Selector(), graphsync.Priority(0), td.extension),
+	}
+
+	responseManager.ProcessRequests(td.ctx, p1, req1)
+	responseManager.ProcessRequests(td.ctx, p2, req2)
+
+	peerState := responseManager.PeerState(p1)
 	require.Len(t, peerState.RequestStates, 1)
-	require.Equal(t, peerState.RequestStates[td.requestID], graphsync.Queued)
+	require.Equal(t, peerState.RequestStates[reqid1], graphsync.Queued)
 	require.Len(t, peerState.Pending, 1)
-	require.Equal(t, peerState.Pending[0], td.requestID)
+	require.Equal(t, peerState.Pending[0], reqid1)
 	require.Len(t, peerState.Active, 0)
 	// no inconsistencies
 	require.Len(t, peerState.Diagnostics(), 0)
 	peerState = responseManager.PeerState(p2)
 	require.Len(t, peerState.RequestStates, 1)
-	require.Equal(t, peerState.RequestStates[td.requestID], graphsync.Queued)
+	require.Equal(t, peerState.RequestStates[reqid2], graphsync.Queued)
 	require.Len(t, peerState.Pending, 1)
-	require.Equal(t, peerState.Pending[0], td.requestID)
+	require.Equal(t, peerState.Pending[0], reqid2)
 	require.Len(t, peerState.Active, 0)
 	// no inconsistencies
 	require.Len(t, peerState.Diagnostics(), 0)

responsemanager/server.go

@@ -3,7 +3,6 @@ package responsemanager
 import (
 	"context"
 	"errors"
-	"fmt"
 	"math"
 	"time"
 
@@ -124,7 +123,7 @@ func (rm *ResponseManager) processUpdate(ctx context.Context, requestID graphsyn
 func (rm *ResponseManager) unpauseRequest(requestID graphsync.RequestID, extensions ...graphsync.ExtensionData) error {
 	inProgressResponse, ok := rm.inProgressResponses[requestID]
 	if !ok {
-		return graphsync.RequestNotFoundErr{}
+		return &graphsync.RequestNotFoundErr{}
 	}
 	if inProgressResponse.state != graphsync.Paused {
 		return errors.New("request is not paused")
@@ -148,7 +147,7 @@ func (rm *ResponseManager) abortRequest(ctx context.Context, requestID graphsync
 		rm.responseQueue.Remove(queueTopic{response.peer, requestID}, response.peer)
 	}
 	if !ok || response.state == graphsync.CompletingSend {
-		return graphsync.RequestNotFoundErr{}
+		return &graphsync.RequestNotFoundErr{}
 	}
 
 	_, span := otel.Tracer("graphsync").Start(trace.ContextWithSpan(ctx, response.span),
@@ -354,7 +353,7 @@ func (rm *ResponseManager) getUpdates(requestID graphsync.RequestID) []gsmsg.Gra
 func (rm *ResponseManager) pauseRequest(requestID graphsync.RequestID) error {
 	inProgressResponse, ok := rm.inProgressResponses[requestID]
 	if !ok || inProgressResponse.state == graphsync.CompletingSend {
-		return graphsync.RequestNotFoundErr{}
+		return &graphsync.RequestNotFoundErr{}
 	}
 	if inProgressResponse.state == graphsync.Paused {
 		return errors.New("request is already paused")
@@ -371,7 +370,6 @@ func (rm *ResponseManager) peerState(p peer.ID) peerstate.PeerState {
 	rm.responseQueue.WithPeerTopics(p, func(peerTopics *peertracker.PeerTrackerTopics) {
 		requestStates := make(graphsync.RequestStates)
 		for key, ipr := range rm.inProgressResponses {
-			fmt.Printf("%v <> %v == %v\n", ipr.peer.Pretty(), p.Pretty(), ipr.peer == p)
 			if ipr.peer == p {
 				requestStates[key] = ipr.state
 			}