[Feature] UUIDs, protocol versioning, v2 protocol w/ dag-cbor messaging

benchmarks/testnet/virtual.go

@@ -16,6 +16,7 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	gsmsg "github.com/ipfs/go-graphsync/message"
+	gsmsgv1 "github.com/ipfs/go-graphsync/message/v1"
 	gsnet "github.com/ipfs/go-graphsync/network"
 )
 
@@ -137,7 +138,7 @@ func (n *network) SendMessage(
 			rateLimiters[to] = rateLimiter
 		}
 
-		pbMsg, err := mes.ToProto()
+		pbMsg, err := gsmsgv1.NewMessageHandler().ToProto(peer.ID("foo"), mes)
 		if err != nil {
 			return err
 		}

cidset/cidset.go

@@ -4,32 +4,30 @@ import (
 	"errors"
 
 	"github.com/ipfs/go-cid"
+	"github.com/ipld/go-ipld-prime/datamodel"
 	"github.com/ipld/go-ipld-prime/fluent"
 	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
 	"github.com/ipld/go-ipld-prime/node/basicnode"
-
-	"github.com/ipfs/go-graphsync/ipldutil"
 )
 
 // EncodeCidSet encodes a cid set into bytes for the do-no-send-cids extension
-func EncodeCidSet(cids *cid.Set) ([]byte, error) {
+func EncodeCidSet(cids *cid.Set) datamodel.Node {
 	list := fluent.MustBuildList(basicnode.Prototype.List, int64(cids.Len()), func(la fluent.ListAssembler) {
 		_ = cids.ForEach(func(c cid.Cid) error {
 			la.AssembleValue().AssignLink(cidlink.Link{Cid: c})
 			return nil
 		})
 	})
-	return ipldutil.EncodeNode(list)
+	return list
 }
 
 // DecodeCidSet decode a cid set from data for the do-no-send-cids extension
-func DecodeCidSet(data []byte) (*cid.Set, error) {
-	list, err := ipldutil.DecodeNode(data)
-	if err != nil {
-		return nil, err
+func DecodeCidSet(data datamodel.Node) (*cid.Set, error) {
+	if data.Kind() != datamodel.Kind_List {
+		return nil, errors.New("did not receive a list of CIDs")
 	}
 	set := cid.NewSet()
-	iter := list.ListIterator()
+	iter := data.ListIterator()
 	for !iter.Done() {
 		_, next, err := iter.Next()
 		if err != nil {

cidset/cidset_test.go

@@ -15,8 +15,7 @@ func TestDecodeEncodeCidSet(t *testing.T) {
 	for _, c := range cids {
 		set.Add(c)
 	}
-	encoded, err := EncodeCidSet(set)
-	require.NoError(t, err, "encode errored")
+	encoded := EncodeCidSet(set)
 	decodedCidSet, err := DecodeCidSet(encoded)
 	require.NoError(t, err, "decode errored")
 	require.Equal(t, decodedCidSet.Len(), set.Len())

dedupkey/dedupkey.go

@@ -1,27 +1,21 @@
 package dedupkey
 
 import (
+	"github.com/ipld/go-ipld-prime/datamodel"
 	"github.com/ipld/go-ipld-prime/node/basicnode"
-
-	"github.com/ipfs/go-graphsync/ipldutil"
 )
 
 // EncodeDedupKey returns encoded cbor data for string key
-func EncodeDedupKey(key string) ([]byte, error) {
+func EncodeDedupKey(key string) (datamodel.Node, error) {
 	nb := basicnode.Prototype.String.NewBuilder()
 	err := nb.AssignString(key)
 	if err != nil {
 		return nil, err
 	}
-	nd := nb.Build()
-	return ipldutil.EncodeNode(nd)
+	return nb.Build(), nil
 }
 
 // DecodeDedupKey returns a string key decoded from cbor data
-func DecodeDedupKey(data []byte) (string, error) {
-	nd, err := ipldutil.DecodeNode(data)
-	if err != nil {
-		return "", err
-	}
-	return nd.AsString()
+func DecodeDedupKey(data datamodel.Node) (string, error) {
+	return data.AsString()
 }

docs/processes.puml

@@ -12,16 +12,17 @@ partition "Top Level Interface" {
 if (operation type) then (outgoing request or incoming response)
 partition "Graphsync Requestor Implementation" {
 :RequestManager;
-if (operation type) then (incoming response)
-partition "Verifying Queries" {
+partition "Executing Requests" {
+:TaskQueue;
 fork
-:ipld.Traverse;
+:Executor;
 fork again
-:ipld.Traverse;
+:Executor;
 fork again
-:ipld.Traverse;
+:Executor;
 end fork
 }
+if (operation type) then (verified responses)
 partition "Collecting Responses" {
 fork
 :Response Collector;
@@ -33,21 +34,21 @@ end fork
 }
 :Responses returned to client;
 stop
-else (outgoing request)
+else (request messages)
 :Send Request To Network;
 endif
 }
 else (incoming request)
 partition "Graphsync Responder Implementation" {
 :ResponseManager;
 partition "Performing Queries" {
-:PeerTaskQueue;
+:TaskQueue;
 fork
-:ipld.Traverse;
+:QueryExecutor;
 fork again
-:ipld.Traverse;
+:QueryExecutor;
 fork again
-:ipld.Traverse;
+:QueryExecutor;
 end fork
 }
 }

docs/request-execution.puml

@@ -0,0 +1,102 @@
+@startuml Request Execution
+participant "GraphSync\nTop Level\nInterface" as TLI
+participant RequestManager
+participant TaskQueue
+participant RequestExecutor as RE
+participant ReconciledLoader
+participant TraversalRecord
+participant Verifier
+participant LocalStorage
+participant Traverser
+participant Network
+
+== Initialization ==
+
+TLI -> RequestManager ** : Setup
+TLI -> RE ** : Setup
+TLI -> TaskQueue ** : Setup
+
+== Executing A Request ==
+
+par
+note over TLI : Request Initiation
+TLI -> RequestManager : New Request
+RequestManager -> RequestManager : Create Request Context
+RequestManager -> TaskQueue : Push Request
+else
+note over RE: Request Execution
+TaskQueue -> RE : Next Request\nTo Process
+RE -> RequestManager : Initiate request execution
+RequestManager -> Traverser ** : Create to manage selector traversal
+RequestManager -> ReconciledLoader ** : create to manage 
+RequestManager -> RE : Traverser + ReconciledLoader
+note over RE: Local loading phase
+loop until traversal complete, request context cancelled, or missing block locally
+Traverser -> RE : Request to load blocks\nto perform traversal
+RE -> ReconciledLoader : Load next block
+ReconciledLoader -> LocalStorage : Load Block
+LocalStorage --> ReconciledLoader : Block or missing
+ReconciledLoader -> TraversalRecord : Record link traversal
+TraversalRecord --> ReconciledLoader
+ReconciledLoader --> RE : Block or missing
+opt block is present
+RE --> Traverser : Next block to load
+end
+end
+RE -> Network : Send Graphsync Request
+RE -> ReconciledLoader : remote online
+ReconciledLoader -> Verifier ** : Create new from traversal record
+ReconciledLoader -> RE
+note over RE: Remote loading phase
+loop until traversal complete, request context cancelled, or missing block locally
+Traverser -> RE : Request to load blocks\nto perform traversal
+RE -> ReconciledLoader : Load next block
+alt on missing path for remote
+ReconciledLoader -> LocalStorage : Load Block
+LocalStorage --> ReconciledLoader : Block or missing
+else
+loop until block loaded, missing, or error
+opt new remote responses
+  alt verification not done
+    ReconciledLoader -> Verifier : verify next response
+    alt success
+    Verifier --> ReconciledLoader : verified
+    ReconciledLoader -> ReconciledLoader : wait for more responses
+    else failure
+    Verifier --> ReconciledLoader : error
+    end
+  else verification done
+    alt next response matches current block load
+
+      alt next response contains a block
+        ReconciledLoader -> LocalStorage : store remote block
+        LocalStorage --> ReconciledLoader
+        ReconciledLoader -> ReconciledLoader : block laoded from remote
+      else next response does not contain block
+        opt next response is missing
+          ReconciledLoader -> ReconciledLoader : record missing path
+        end
+        ReconciledLoader -> LocalStorage : load block
+        LocalStorage --> ReconciledLoader : block or missing
+      end
+    else next response doesn not match
+      ReconciledLoader -> ReconciledLoader : error
+    end
+  end
+end
+opt remote goes offline
+ReconciledLoader -> LocalStorage : load block
+LocalStorage --> ReconciledLoader : block or missing
+end
+end
+ReconciledLoader -> TraversalRecord : Record link traversal
+TraversalRecord --> ReconciledLoader
+ReconciledLoader --> RE : Block, missing or error
+RE -> Traverser : Next block to load
+end
+end
+else
+Network -> RequestManager : New Responses
+RequestManager -> ReconciledLoader : Ingest Responses
+end
+@enduml