Refactor async loading for simplicity and correctness

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

docs/responder-sequence.puml

@@ -1,24 +1,20 @@
 @startuml Responding To A Request
 participant "GraphSync\nTop Level\nInterface" as TLI
 participant ResponseManager
-participant "Query Executor" as QW
-participant PeerTaskQueue
+participant "QueryExecutor" as QW
+participant TaskQueue
 participant PeerTracker
 participant Traverser
 participant ResponseAssembler
 participant LinkTracker
-participant ResponseBuilder
-participant "Intercepted Loader" as ILoader
 participant Loader
 participant "Message Sending\nLayer" as Message
 
 == Initialization ==
 
 TLI -> ResponseManager ** : Setup
-ResponseManager -> QW ** : Create
-activate QW
-TLI -> PeerTaskQueue ** : Setup
-TLI -> PeerResponseManager ** : Setup
+TLI -> QW ** : Setup
+TLI -> TaskQueue ** : Setup
 
 == Responding To Request ==
 
@@ -27,38 +23,32 @@ loop until shutdown
 note over TLI : Request Queueing Loop
 TLI -> ResponseManager : Process requests
 alt new request
-ResponseManager -> PeerTaskQueue : Push Request
-PeerTaskQueue -> PeerTracker ** : Create for peer\n as neccesary
-PeerTaskQueue -> PeerTracker : Push Request
 ResponseManager -> ResponseManager : Create Request Context
+ResponseManager -> TaskQueue : Push Request
 else cancel request
 ResponseManager -> ResponseManager : Cancel Request Context
 end
 end
 else
 loop until shutdown
 note over QW: Request Processing Loop
-QW -> PeerTaskQueue : Pop Request
-PeerTaskQueue -> PeerTracker : Pop Request
-PeerTracker -> PeerTaskQueue : Next Request\nTo Process
-PeerTaskQueue -> QW : Next Request\nTo Process
+TaskQueue -> QW : Next Request\nTo Process
+activate QW
 QW -> QW : Process incoming request hooks
-QW -> ILoader ** : Create w/ Request, Peer, and Loader
 QW -> Traverser ** : Create to manage selector traversal
 loop until traversal complete or request context cancelled
 note over Traverser: Selector Traversal Loop
-Traverser -> ILoader : Request to load blocks\nto perform traversal
-ILoader -> Loader : Load blocks\nfrom local storage
-Loader -> ILoader : Blocks From\nlocal storage or error
-ILoader -> Traverser : Blocks to continue\n traversal or error
-ILoader -> QW : Block or error to Send Back
+Traverser -> QW : Request to load blocks\nto perform traversal
+QW -> Loader : Load blocks\nfrom local storage
+Loader -> QW : Blocks From\nlocal storage or error
+QW -> Traverser : Blocks to continue\n traversal or error
 QW -> QW: Processing outgoing block hooks
 QW -> ResponseAssembler: Add outgoing responses
 activate ResponseAssembler
 ResponseAssembler -> LinkTracker ** : Create for peer if not already present
 ResponseAssembler -> LinkTracker : Notify block or\n error, ask whether\n block is duplicate
 LinkTracker -> ResponseAssembler : Whether to\n send block
-ResponseAssembler -> ResponseBuilder : Aggregate Response Metadata & Block
+ResponseAssembler -> ResponseAssembler : Aggregate Response Metadata & Blocks
 ResponseAssembler -> Message : Send aggregate response
 deactivate ResponseAssembler
 end
@@ -67,7 +57,7 @@ QW -> ResponseAssembler : Request Finished
 activate ResponseAssembler
 ResponseAssembler -> LinkTracker : Query If Errors\n Were Present
 LinkTracker -> ResponseAssembler : True/False\n if errors present
-ResponseAssembler -> ResponseBuilder : Aggregate request finishing
+ResponseAssembler -> ResponseAssembler : Aggregate request finishing
 ResponseAssembler -> Message : Send aggregate response
 deactivate ResponseAssembler
 end

graphsync.go

@@ -127,14 +127,29 @@ func (e RequestNotFoundErr) Error() string {
 }
 
 // RemoteMissingBlockErr indicates that the remote peer was missing a block
-// in the selector requested. It is a non-terminal error in the error stream
+// in the selector requested, and we also don't have it locally.
+// It is a  -terminal error in the error stream
 // for a request and does NOT cause a request to fail completely
 type RemoteMissingBlockErr struct {
 	Link ipld.Link
+	Path ipld.Path
 }
 
 func (e RemoteMissingBlockErr) Error() string {
-	return fmt.Sprintf("remote peer is missing block: %s", e.Link.String())
+	return fmt.Sprintf("remote peer is missing block (%s) at path %s", e.Link.String(), e.Path)
+}
+
+// RemoteIncorrectResponseError indicates that the remote peer sent a response
+// to a traversal that did not correspond with the expected next link
+// in the selector traversal based on verification of data up to this point
+type RemoteIncorrectResponseError struct {
+	LocalLink  ipld.Link
+	RemoteLink ipld.Link
+	Path       ipld.Path
+}
+
+func (e RemoteIncorrectResponseError) Error() string {
+	return fmt.Sprintf("expected link (%s) at path %s does not match link sent by remote (%s), possible malicious responder", e.LocalLink, e.Path, e.RemoteLink)
 }
 
 var (
@@ -223,6 +238,8 @@ type LinkMetadataIterator func(cid.Cid, LinkAction)
 
 // LinkMetadata is used to access link metadata through an Iterator
 type LinkMetadata interface {
+	// Length returns the number of metadata entries
+	Length() int64
 	// Iterate steps over individual metadata one by one using the provided
 	// callback
 	Iterate(LinkMetadataIterator)