Accept missing blocks without failing data transfer (#291)

* feat(channels): add missing cids field

* feat(graphsync): handle missings cids via events, integration test

* style(lint): mod tidy

* feat(channels): put retrievals with missing CIDs in "partially complete"

* style(graphsync): add explanatory comment

* feat(deps): update to v0.12.0 graphsync

* ci(circle): update to go 1.16

* build(go.mod): update to go 1.16

Author: hannahhoward

.circleci/config.yml

@@ -1,12 +1,12 @@
 version: 2.1
 orbs:
-  go: gotest/[email protected].9
+  go: gotest/[email protected].14
   codecov: codecov/[email protected]
 
 executors:
   golang:
     docker:
-      - image: circleci/golang:1.15
+      - image: circleci/golang:1.16
     resource_class: small
 
 commands:

channelmonitor/channelmonitor_test.go

@@ -621,3 +621,7 @@ func (m *mockChannelState) ReceivedCidsLen() int {
 func (m *mockChannelState) ReceivedCidsTotal() int64 {
 	panic("implement me")
 }
+
+func (m *mockChannelState) MissingCids() []cid.Cid {
+	panic("implement me")
+}

channels/channel_state.go

@@ -56,6 +56,9 @@ type channelState struct {
 	// stages tracks the timeline of events related to a data transfer, for
 	// traceability purposes.
 	stages *datatransfer.ChannelStages
+
+	// missingCids are the set of CIDS that were missing and skipped over in the data transfer
+	missingCids []cid.Cid
 }
 
 // EmptyChannelState is the zero value for channel state, meaning not present
@@ -193,6 +196,10 @@ func (c channelState) OtherPeer() peer.ID {
 	return c.sender
 }
 
+func (c channelState) MissingCids() []cid.Cid {
+	return c.missingCids
+}
+
 // Stages returns the current ChannelStages object, or an empty object.
 // It is unsafe for the caller to modify the return value, and changes may not
 // be persisted. It should be treated as immutable.
@@ -230,6 +237,7 @@ func fromInternalChannelState(c internal.ChannelState, voucherDecoder DecoderByT
 		voucherDecoder:       voucherDecoder,
 		receivedCids:         receivedCidsReader,
 		stages:               c.Stages,
+		missingCids:          c.MissingCids,
 	}
 }
 

channels/channels.go

@@ -373,6 +373,11 @@ func (c *Channels) ReceiveDataError(chid datatransfer.ChannelID, err error) erro
 	return c.send(chid, datatransfer.ReceiveDataError, err)
 }
 
+// CIDMissing indicates the sender is missing a section of the graph in the response
+func (c *Channels) CIDMissing(chid datatransfer.ChannelID, cid cid.Cid) error {
+	return c.send(chid, datatransfer.CIDMissing, cid)
+}
+
 // HasChannel returns true if the given channel id is being tracked
 func (c *Channels) HasChannel(chid datatransfer.ChannelID) (bool, error) {
 	return c.stateMachines.Has(chid)

channels/channels_fsm.go

@@ -1,6 +1,7 @@
 package channels
 
 import (
+	"github.com/ipfs/go-cid"
 	logging "github.com/ipfs/go-log/v2"
 	cbg "github.com/whyrusleeping/cbor-gen"
 
@@ -103,6 +104,24 @@ var ChannelEvents = fsm.Events{
 			chst.AddLog("")
 			return nil
 		}),
+
+	fsm.Event(datatransfer.CIDMissing).FromMany(transferringStates...).ToJustRecord().
+		Action(func(chst *internal.ChannelState, missing cid.Cid) error {
+			// TODO: find a more efficient way to do this
+			var found bool
+			for _, c := range chst.MissingCids {
+				if c.Equals(missing) {
+					found = true
+					break
+				}
+			}
+			if !found {
+				chst.MissingCids = append(chst.MissingCids, missing)
+			}
+			chst.AddLog("")
+			return nil
+		}),
+
 	fsm.Event(datatransfer.Disconnected).FromAny().ToNoChange().Action(func(chst *internal.ChannelState, err error) error {
 		chst.Message = err.Error()
 		chst.AddLog("data transfer disconnected: %s", chst.Message)
@@ -225,6 +244,12 @@ var ChannelEvents = fsm.Events{
 		return nil
 	}),
 
+	fsm.Event(datatransfer.CleanupCompletePartial).
+		From(datatransfer.Completing).To(datatransfer.PartiallyCompleted).Action(func(chst *internal.ChannelState) error {
+		chst.AddLog("")
+		return nil
+	}),
+
 	// will kickoff state handlers for channels that were cleaning up
 	fsm.Event(datatransfer.CompleteCleanupOnRestart).FromAny().ToNoChange().Action(func(chst *internal.ChannelState) error {
 		chst.AddLog("")
@@ -247,6 +272,9 @@ func cleanupConnection(ctx fsm.Context, env ChannelEnvironment, channel internal
 	}
 	env.CleanupChannel(datatransfer.ChannelID{ID: channel.TransferID, Initiator: channel.Initiator, Responder: channel.Responder})
 	env.Unprotect(otherParty, datatransfer.ChannelID{ID: channel.TransferID, Initiator: channel.Initiator, Responder: channel.Responder}.String())
+	if channel.Status == datatransfer.Completing && len(channel.MissingCids) > 0 {
+		return ctx.Trigger(datatransfer.CleanupCompletePartial)
+	}
 	return ctx.Trigger(datatransfer.CleanupComplete)
 }
 
@@ -262,6 +290,7 @@ var ChannelFinalityStates = []fsm.StateKey{
 	datatransfer.Cancelled,
 	datatransfer.Completed,
 	datatransfer.Failed,
+	datatransfer.PartiallyCompleted,
 }
 
 // IsChannelTerminated returns true if the channel is in a finality state

channels/channels_test.go

@@ -241,6 +241,37 @@ func TestChannels(t *testing.T) {
 		require.ElementsMatch(t, []cid.Cid{cids[0], cids[1]}, state.ReceivedCids())
 	})
 
+	t.Run("missing cids", func(t *testing.T) {
+		ds := dss.MutexWrap(datastore.NewMapDatastore())
+		dir := os.TempDir()
+		cidLists, err := cidlists.NewCIDLists(dir)
+		require.NoError(t, err)
+		channelList, err := channels.New(ds, cidLists, notifier, decoderByType, decoderByType, &fakeEnv{}, peers[0])
+		require.NoError(t, err)
+		err = channelList.Start(ctx)
+		require.NoError(t, err)
+
+		_, err = channelList.CreateNew(peers[0], tid1, cids[0], selector, fv1, peers[0], peers[0], peers[1])
+		require.NoError(t, err)
+		state := checkEvent(ctx, t, received, datatransfer.Open)
+		require.Equal(t, datatransfer.Requested, state.Status())
+
+		err = channelList.CIDMissing(datatransfer.ChannelID{Initiator: peers[0], Responder: peers[1], ID: tid1}, cids[0])
+		require.NoError(t, err)
+		state = checkEvent(ctx, t, received, datatransfer.CIDMissing)
+		require.Equal(t, []cid.Cid{cids[0]}, state.MissingCids())
+
+		err = channelList.CIDMissing(datatransfer.ChannelID{Initiator: peers[0], Responder: peers[1], ID: tid1}, cids[1])
+		require.NoError(t, err)
+		state = checkEvent(ctx, t, received, datatransfer.CIDMissing)
+		require.Equal(t, []cid.Cid{cids[0], cids[1]}, state.MissingCids())
+
+		err = channelList.CIDMissing(datatransfer.ChannelID{Initiator: peers[0], Responder: peers[1], ID: tid1}, cids[0])
+		require.NoError(t, err)
+		state = checkEvent(ctx, t, received, datatransfer.CIDMissing)
+		require.Equal(t, []cid.Cid{cids[0], cids[1]}, state.MissingCids())
+	})
+
 	t.Run("pause/resume", func(t *testing.T) {
 		state, err := channelList.GetByID(ctx, datatransfer.ChannelID{Initiator: peers[0], Responder: peers[1], ID: tid1})
 		require.NoError(t, err)

channels/internal/internalchannel.go

@@ -68,6 +68,9 @@ type ChannelState struct {
 	//
 	// EXPERIMENTAL; subject to change.
 	Stages *datatransfer.ChannelStages
+
+	// MissingCids are the set of CIDS that were missing and skipped over in the data transfer
+	MissingCids []cid.Cid
 }
 
 // AddLog takes an fmt string with arguments, and adds the formatted string to

channels/internal/internalchannel_cbor_gen.go

@@ -110,6 +110,13 @@ const (
 
 	// Opened is fired when a request for data is sent from this node to a peer
 	Opened
+
+	// CIDMissing is fired when the sender is missing a section of the graph in the response
+	CIDMissing
+
+	// CleanupCompletePartial causes a completing request to transition to a PartiallyCompleted state
+	// rather than a full Completed state
+	CleanupCompletePartial
 )
 
 // Events are human readable names for data transfer events
@@ -144,6 +151,8 @@ var Events = map[EventCode]string{
 	ReceiveDataError:            "ReceiveDataError",
 	TransferRequestQueued:       "TransferRequestQueued",
 	RequestCancelled:            "RequestCancelled",
+	CIDMissing:                  "CIDMissing",
+	CleanupCompletePartial:      "CleanupCompletePartial",
 }
 
 // Event is a struct containing information about a data transfer event

events.go

@@ -1,6 +1,6 @@
 module github.com/filecoin-project/go-data-transfer
 
-go 1.15
+go 1.16
 
 require (
 	github.com/bep/debounce v1.2.0
@@ -10,10 +10,10 @@ require (
 	github.com/hannahhoward/go-pubsub v0.0.0-20200423002714-8d62886cc36e
 	github.com/ipfs/go-block-format v0.0.3
 	github.com/ipfs/go-blockservice v0.2.1
-	github.com/ipfs/go-cid v0.0.7
+	github.com/ipfs/go-cid v0.1.0
 	github.com/ipfs/go-datastore v0.5.1
 	github.com/ipfs/go-ds-badger v0.3.0
-	github.com/ipfs/go-graphsync v0.11.5
+	github.com/ipfs/go-graphsync v0.12.0
 	github.com/ipfs/go-ipfs-blockstore v1.1.2
 	github.com/ipfs/go-ipfs-blocksutil v0.0.1
 	github.com/ipfs/go-ipfs-chunker v0.0.5
@@ -24,8 +24,8 @@ require (
 	github.com/ipfs/go-ipld-format v0.2.0
 	github.com/ipfs/go-log/v2 v2.3.0
 	github.com/ipfs/go-merkledag v0.5.1
-	github.com/ipfs/go-unixfs v0.2.4
-	github.com/ipld/go-ipld-prime v0.14.3
+	github.com/ipfs/go-unixfs v0.3.1
+	github.com/ipld/go-ipld-prime v0.14.4
 	github.com/jbenet/go-random v0.0.0-20190219211222-123a90aedc0c
 	github.com/jpillora/backoff v1.0.0
 	github.com/libp2p/go-libp2p v0.16.0
@@ -37,7 +37,7 @@ require (
 	go.opentelemetry.io/otel/sdk v1.2.0
 	go.opentelemetry.io/otel/trace v1.3.0
 	go.uber.org/atomic v1.9.0
-	golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6
+	golang.org/x/exp v0.0.0-20210615023648-acb5c1269671
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1
 )