Scrub response errors

benchmarks/testnet/network_test.go

@@ -33,7 +33,7 @@ func TestSendMessageAsyncButWaitForResponse(t *testing.T) {
 		fromWaiter peer.ID,
 		msgFromWaiter gsmsg.GraphSyncMessage) {
 
-		builder := gsmsg.NewBuilder(gsmsg.Topic(0))
+		builder := gsmsg.NewBuilder()
 		builder.AddBlock(blocks.NewBlock([]byte(expectedStr)))
 		msgToWaiter, err := builder.Build()
 		require.NoError(t, err)
@@ -62,7 +62,7 @@ func TestSendMessageAsyncButWaitForResponse(t *testing.T) {
 		}
 	}))
 
-	builder := gsmsg.NewBuilder(gsmsg.Topic(0))
+	builder := gsmsg.NewBuilder()
 	builder.AddBlock(blocks.NewBlock([]byte("data")))
 	messageSentAsync, err := builder.Build()
 	require.NoError(t, err)

impl/graphsync.go

@@ -259,7 +259,6 @@ func New(parent context.Context, network gsnet.GraphSyncNetwork,
 		responseManager,
 		outgoingBlockHooks,
 		requestUpdatedHooks,
-		responseAssembler,
 	)
 	graphSync := &GraphSync{
 		network:                     network,

impl/graphsync_test.go

@@ -135,7 +135,7 @@ func TestSendResponseToIncomingRequest(t *testing.T) {
 
 	requestID := graphsync.RequestID(rand.Int31())
 
-	builder := gsmsg.NewBuilder(gsmsg.Topic(0))
+	builder := gsmsg.NewBuilder()
 	builder.AddRequest(gsmsg.NewRequest(requestID, blockChain.TipLink.(cidlink.Link).Cid, blockChain.Selector(), graphsync.Priority(math.MaxInt32), td.extension))
 	message, err := builder.Build()
 	require.NoError(t, err)
@@ -978,13 +978,14 @@ func TestNetworkDisconnect(t *testing.T) {
 	drain(responder)
 
 	tracing := collectTracing(t)
-	traceStrings := tracing.TracesToStrings()
-	require.Contains(t, traceStrings, "response(0)->executeTask(0)")
-	// may contain multiple abortRequest traces as the network error can bubble up >1 times
-	// but these will only record if the request is still executing
-	require.Contains(t, traceStrings, "request(0)->newRequest(0)")
-	require.Contains(t, traceStrings, "request(0)->executeTask(0)")
-	require.Contains(t, traceStrings, "request(0)->terminateRequest(0)")
+	require.ElementsMatch(t, []string{
+		"response(0)->executeTask(0)",
+		"response(0)->abortRequest(0)",
+		"response(0)->executeTask(1)",
+		"request(0)->newRequest(0)",
+		"request(0)->executeTask(0)",
+		"request(0)->terminateRequest(0)",
+	}, tracing.TracesToStrings())
 	// has ContextCancelError exception recorded in the right place
 	tracing.SingleExceptionEvent(t, "request(0)->executeTask(0)", "ContextCancelError", ipldutil.ContextCancelError{}.Error(), false)
 }

message/builder.go

@@ -14,7 +14,6 @@ import (
 // requests for a given peer and then generates the corresponding
 // GraphSync message when ready to send
 type Builder struct {
-	topic              Topic
 	outgoingBlocks     map[cid.Cid]blocks.Block
 	blkSize            uint64
 	completedResponses map[graphsync.RequestID]graphsync.ResponseStatusCode
@@ -23,13 +22,9 @@ type Builder struct {
 	requests           map[graphsync.RequestID]GraphSyncRequest
 }
 
-// Topic is an identifier for notifications about this response builder
-type Topic uint64
-
 // NewBuilder generates a new Builder.
-func NewBuilder(topic Topic) *Builder {
+func NewBuilder() *Builder {
 	return &Builder{
-		topic:              topic,
 		requests:           make(map[graphsync.RequestID]GraphSyncRequest),
 		outgoingBlocks:     make(map[cid.Cid]blocks.Block),
 		completedResponses: make(map[graphsync.RequestID]graphsync.ResponseStatusCode),
@@ -87,6 +82,31 @@ func (b *Builder) Empty() bool {
 	return len(b.requests) == 0 && len(b.outgoingBlocks) == 0 && len(b.outgoingResponses) == 0
 }
 
+// ScrubResponse removes a response from a message and any blocks only referenced by that response
+func (b *Builder) ScrubResponses(requestIDs []graphsync.RequestID) uint64 {
+	for _, requestID := range requestIDs {
+		delete(b.completedResponses, requestID)
+		delete(b.extensions, requestID)
+		delete(b.outgoingResponses, requestID)
+	}
+	oldSize := b.blkSize
+	newBlkSize := uint64(0)
+	savedBlocks := make(map[cid.Cid]blocks.Block, len(b.outgoingBlocks))
+	for _, metadata := range b.outgoingResponses {
+		for _, item := range metadata {
+			block, willSendBlock := b.outgoingBlocks[item.Link]
+			_, alreadySavedBlock := savedBlocks[item.Link]
+			if item.BlockPresent && willSendBlock && !alreadySavedBlock {
+				savedBlocks[item.Link] = block
+				newBlkSize += uint64(len(block.RawData()))
+			}
+		}
+	}
+	b.blkSize = newBlkSize
+	b.outgoingBlocks = savedBlocks
+	return oldSize - newBlkSize
+}
+
 // Build assembles and encodes message data from the added requests, links, and blocks.
 func (b *Builder) Build() (GraphSyncMessage, error) {
 	responses := make(map[graphsync.RequestID]GraphSyncResponse, len(b.outgoingResponses))
@@ -107,11 +127,6 @@ func (b *Builder) Build() (GraphSyncMessage, error) {
 	}, nil
 }
 
-// Topic returns the identifier for notifications sent about this builder
-func (b *Builder) Topic() Topic {
-	return b.topic
-}
-
 func responseCode(status graphsync.ResponseStatusCode, isComplete bool) graphsync.ResponseStatusCode {
 	if !isComplete {
 		return graphsync.PartialResponse