feat(linktracker): create link tracker

Create a link tracker that helps dedup block sends and also track if blocks are missing from
requests

responsemanager/linktracker/linktracker.go

+package linktracker
+
+import (
+	gsmsg "github.com/ipfs/go-graphsync/message"
+	"github.com/ipld/go-ipld-prime"
+)
+
+// LinkTracker records links being traversed to determine useful information
+// in crafting responses for a peer. Specifically, if any in progress request
+// has already sent a block for a given link, don't send it again.
+// Second, keep track of whether links are missing blocks so you can determine
+// at the end if a complete response has been transmitted.
+type LinkTracker struct {
+	isMissingBlocks                   map[gsmsg.GraphSyncRequestID]struct{}
+	linksWithBlocksTraversedByRequest map[gsmsg.GraphSyncRequestID][]ipld.Link
+	traversalsWithBlocksInProgress    map[ipld.Link]int
+}
+
+// New makes a new link tracker
+func New() *LinkTracker {
+	return &LinkTracker{
+		isMissingBlocks:                   make(map[gsmsg.GraphSyncRequestID]struct{}),
+		linksWithBlocksTraversedByRequest: make(map[gsmsg.GraphSyncRequestID][]ipld.Link),
+		traversalsWithBlocksInProgress:    make(map[ipld.Link]int),
+	}
+}
+
+// ShouldSendBlockFor says whether we should send a block for a given link, based
+// on whether we have traversed it already in one of the in progress requests and
+// sent a block already.
+func (lt *LinkTracker) ShouldSendBlockFor(link ipld.Link) bool {
+	return lt.traversalsWithBlocksInProgress[link] == 0
+}
+
+// RecordLinkTraversal records that we traversed a link during a request, and
+// whether we had the block when we did it.
+func (lt *LinkTracker) RecordLinkTraversal(requestID gsmsg.GraphSyncRequestID, link ipld.Link, hasBlock bool) {
+	if hasBlock {
+		lt.linksWithBlocksTraversedByRequest[requestID] = append(lt.linksWithBlocksTraversedByRequest[requestID], link)
+		lt.traversalsWithBlocksInProgress[link]++
+	} else {
+		lt.isMissingBlocks[requestID] = struct{}{}
+	}
+}
+
+// FinishRequest records that we have completed the given request, and returns
+// true if all links traversed had blocks present.
+func (lt *LinkTracker) FinishRequest(requestID gsmsg.GraphSyncRequestID) (hasAllBlocks bool) {
+	_, ok := lt.isMissingBlocks[requestID]
+	hasAllBlocks = !ok
+	links, ok := lt.linksWithBlocksTraversedByRequest[requestID]
+	if !ok {
+		return
+	}
+	for _, link := range links {
+		lt.traversalsWithBlocksInProgress[link]--
+		if lt.traversalsWithBlocksInProgress[link] <= 0 {
+			delete(lt.traversalsWithBlocksInProgress, link)
+		}
+	}
+	delete(lt.linksWithBlocksTraversedByRequest, requestID)
+
+	return
+}

responsemanager/linktracker/linktracker_test.go

+package linktracker
+
+import (
+	"math/rand"
+	"testing"
+
+	gsmsg "github.com/ipfs/go-graphsync/message"
+	"github.com/ipfs/go-graphsync/testbridge"
+)
+
+func TestShouldSendBlocks(t *testing.T) {
+	linkTracker := New()
+	link1 := testbridge.NewMockLink()
+	link2 := testbridge.NewMockLink()
+	if !linkTracker.ShouldSendBlockFor(link1) || !linkTracker.ShouldSendBlockFor(link2) {
+		t.Fatal("Links not traversed should send blocks")
+	}
+	requestID1 := gsmsg.GraphSyncRequestID(rand.Int31())
+	requestID2 := gsmsg.GraphSyncRequestID(rand.Int31())
+
+	linkTracker.RecordLinkTraversal(requestID1, link1, true)
+	linkTracker.RecordLinkTraversal(requestID1, link2, true)
+	linkTracker.RecordLinkTraversal(requestID2, link1, true)
+
+	if linkTracker.ShouldSendBlockFor(link1) || linkTracker.ShouldSendBlockFor(link2) {
+		t.Fatal("Links already traversed with blocks should not send blocks again")
+	}
+
+	linkTracker.FinishRequest(requestID1)
+	if linkTracker.ShouldSendBlockFor(link1) || !linkTracker.ShouldSendBlockFor(link2) {
+		t.Fatal("Finishing request should resend blocks only if there are no in progress requests for that block remain")
+	}
+}
+
+func TestHasAllBlocks(t *testing.T) {
+	linkTracker := New()
+	link1 := testbridge.NewMockLink()
+	link2 := testbridge.NewMockLink()
+	requestID1 := gsmsg.GraphSyncRequestID(rand.Int31())
+	requestID2 := gsmsg.GraphSyncRequestID(rand.Int31())
+
+	linkTracker.RecordLinkTraversal(requestID1, link1, true)
+	linkTracker.RecordLinkTraversal(requestID1, link2, false)
+	linkTracker.RecordLinkTraversal(requestID2, link1, true)
+
+	hasAllBlocksRequest1 := linkTracker.FinishRequest(requestID1)
+	hasAllBlocksRequest2 := linkTracker.FinishRequest(requestID2)
+	if hasAllBlocksRequest1 || !hasAllBlocksRequest2 {
+		t.Fatal("A request has all blocks if and only if all link traversals occurred with blocks present")
+	}
+}
+
+func TestBlockBecomesAvailable(t *testing.T) {
+	linkTracker := New()
+	link1 := testbridge.NewMockLink()
+	if !linkTracker.ShouldSendBlockFor(link1) {
+		t.Fatal("Links not traversed should send blocks")
+	}
+	requestID1 := gsmsg.GraphSyncRequestID(rand.Int31())
+	requestID2 := gsmsg.GraphSyncRequestID(rand.Int31())
+
+	linkTracker.RecordLinkTraversal(requestID1, link1, false)
+	linkTracker.RecordLinkTraversal(requestID2, link1, false)
+
+	if !linkTracker.ShouldSendBlockFor(link1) {
+		t.Fatal("Links traversed without blocks should still send them if they become availabe")
+	}
+
+	linkTracker.RecordLinkTraversal(requestID1, link1, true)
+
+	if linkTracker.ShouldSendBlockFor(link1) {
+		t.Fatal("Links traversed with blocks should no longer send")
+	}
+
+	hasAllBlocks := linkTracker.FinishRequest(requestID1)
+	if hasAllBlocks {
+		t.Fatal("Even if block becomes available, traversal may be incomplete, request still should not be considered to have all blocks")
+	}
+
+	if !linkTracker.ShouldSendBlockFor(link1) {
+		t.Fatal("Block traversals should resend for requests that never traversed while block was present")
+	}
+}