fix(messagev2): adjust schema per feedback

Author: rvagg

message/ipldbind/message.go

@@ -45,19 +45,19 @@ func (gse GraphSyncExtensions) ToExtensionsList() []graphsync.ExtensionData {
 type GraphSyncRequest struct {
 	Id          []byte
 	RequestType graphsync.RequestType
-	Priority    graphsync.Priority
+	Priority    *graphsync.Priority
 	Root        *cid.Cid
 	Selector    *datamodel.Node
-	Extensions  GraphSyncExtensions
+	Extensions  *GraphSyncExtensions
 }
 
 // GraphSyncResponse is an struct to capture data on a response sent back
 // in a GraphSyncMessage.
 type GraphSyncResponse struct {
 	Id         []byte
 	Status     graphsync.ResponseStatusCode
-	Metadata   []message.GraphSyncLinkMetadatum
-	Extensions GraphSyncExtensions
+	Metadata   *[]message.GraphSyncLinkMetadatum
+	Extensions *GraphSyncExtensions
 }
 
 // GraphSyncBlock is a container for representing extension data for bindnode,

message/ipldbind/schema.ipldsch

@@ -1,7 +1,15 @@
-type GraphSyncExtensions {String:nullable Any}
+################################################################################
+###                  GraphSync messaging protocol version 2                  ###
+################################################################################
+
+# UUID bytes
 type GraphSyncRequestID bytes
+
 type GraphSyncPriority int
 
+# Extensions as a name:data map where the data is any arbitrary, valid Node
+type GraphSyncExtensions { String : nullable Any }
+
 type GraphSyncLinkAction enum {
    # Present means the linked block was present on this machine, and is included
    # in this message
@@ -18,6 +26,9 @@ type GraphSyncLinkAction enum {
    # TODO: | DuplicateDAGSkipped ("s")
 } representation string
 
+# Metadata for each "link" in the DAG being communicated, each block gets one of
+# these and missing blocks also get one, with an explanation as per
+# GraphSyncLinkAction
 type GraphSyncMetadatum struct {
   link         Link
   action       GraphSyncLinkAction
@@ -57,38 +68,47 @@ type GraphSyncRequestType enum {
    | Cancel  ("c")
    # Update means the extensions contain an update about this request
    | Update  ("u")
-   # Restart means restart this request from the begging, respecting the any DoNotSendCids/DoNotSendBlocks contained
-   # in the extensions -- essentially a cancel followed by a new
+   # Restart means restart this request from the begging, respecting the any
+   # DoNotSendCids/DoNotSendBlocks contained in the extensions--essentially a
+   # cancel followed by a new
    # TODO: | Restart ("r")
 } representation string
 
 type GraphSyncRequest struct {
-  id                GraphSyncRequestID   # unique id set on the requester side
-  requestType       GraphSyncRequestType # the request type
-  priority          GraphSyncPriority    # the priority (normalized). default to 1
-  root     nullable Link                 # a CID for the root node in the query
-  selector nullable Any                  # see https://github.com/ipld/specs/blob/master/selectors/selectors.md
-  extensions        GraphSyncExtensions  # side channel information
-} representation tuple
+  id                  GraphSyncRequestID   (rename "id")   # unique id set on the requester side
+  requestType         GraphSyncRequestType (rename "type") # the request type
+  priority   optional GraphSyncPriority    (rename "pri")  # the priority (normalized). default to 1
+  root       optional Link                 (rename "root") # a CID for the root node in the query
+  selector   optional Any                  (rename "sel")  # see https://github.com/ipld/specs/blob/master/selectors/selectors.md
+  extensions optional GraphSyncExtensions  (rename "ext")  # side channel information
+} representation map
 
 type GraphSyncResponse struct {
-  id          GraphSyncRequestID          # the request id we are responding to
-  status      GraphSyncResponseStatusCode # a status code.
-  metadata    GraphSyncMetadata           # metadata about response
-  extensions  GraphSyncExtensions         # side channel information
-} representation tuple
+  id          GraphSyncRequestID           (rename "reqid") # the request id we are responding to
+  status      GraphSyncResponseStatusCode  (rename "stat")  # a status code.
+  metadata    optional GraphSyncMetadata   (rename "meta")  # metadata about response
+  extensions  optional GraphSyncExtensions (rename "ext")   # side channel information
+} representation map
 
+# Block data and CID prefix that can be used to reconstruct the entire CID from
+# the hash of the bytes
 type GraphSyncBlock struct {
   prefix  Bytes # CID prefix (cid version, multicodec and multihash prefix (type + length)
   data    Bytes
 } representation tuple
 
+# We expect each message to contain at least one of the fields, typically either
+# just requests, or responses and possibly blocks with it
 type GraphSyncMessage struct {
   requests  optional [GraphSyncRequest]  (rename "req")
   responses optional [GraphSyncResponse] (rename "rsp")
   blocks    optional [GraphSyncBlock]    (rename "blk")
 } representation map
 
+# Parent keyed union to hold the message, the root of the structure that can be
+# used to version the messaging format outside of the protocol and makes the
+# data itself more self-descriptive (i.e. `{"gs2":...` will appear at the front
+# of every msg)
 type GraphSyncMessageRoot union {
   | GraphSyncMessage "gs2"
 } representation keyed

message/v2/message.go

@@ -70,13 +70,15 @@ func (mh *MessageHandler) toIPLD(gsm message.GraphSyncMessage) (*ipldbind.GraphS
 			if root == cid.Undef {
 				rootPtr = nil
 			}
+			priority := request.Priority()
+			ext := ipldbind.NewGraphSyncExtensions(request)
 			ibmRequests = append(ibmRequests, ipldbind.GraphSyncRequest{
 				Id:          request.ID().Bytes(),
 				Root:        rootPtr,
 				Selector:    selPtr,
-				Priority:    request.Priority(),
+				Priority:    &priority,
 				RequestType: request.Type(),
-				Extensions:  ipldbind.NewGraphSyncExtensions(request),
+				Extensions:  &ext,
 			})
 		}
 		ibm.Requests = &ibmRequests
@@ -90,11 +92,13 @@ func (mh *MessageHandler) toIPLD(gsm message.GraphSyncMessage) (*ipldbind.GraphS
 			if !ok {
 				return nil, fmt.Errorf("unexpected metadata type")
 			}
+			md := glsm.RawMetadata()
+			ext := ipldbind.NewGraphSyncExtensions(response)
 			ibmResponses = append(ibmResponses, ipldbind.GraphSyncResponse{
 				Id:         response.RequestID().Bytes(),
 				Status:     response.Status(),
-				Metadata:   glsm.RawMetadata(),
-				Extensions: ipldbind.NewGraphSyncExtensions(response),
+				Metadata:   &md,
+				Extensions: &ext,
 			})
 		}
 		ibm.Responses = &ibmResponses
@@ -161,8 +165,13 @@ func (mh *MessageHandler) fromIPLD(ibm *ipldbind.GraphSyncMessageRoot) (message.
 				continue
 			}
 
+			var ext []graphsync.ExtensionData
+			if req.Extensions != nil {
+				ext = req.Extensions.ToExtensionsList()
+			}
+
 			if req.RequestType == graphsync.RequestTypeUpdate {
-				requests[id] = message.NewUpdateRequest(id, req.Extensions.ToExtensionsList()...)
+				requests[id] = message.NewUpdateRequest(id, ext...)
 				continue
 			}
 
@@ -176,7 +185,12 @@ func (mh *MessageHandler) fromIPLD(ibm *ipldbind.GraphSyncMessageRoot) (message.
 				selector = *req.Selector
 			}
 
-			requests[id] = message.NewRequest(id, root, selector, graphsync.Priority(req.Priority), req.Extensions.ToExtensionsList()...)
+			var priority graphsync.Priority
+			if req.Priority != nil {
+				priority = graphsync.Priority(*req.Priority)
+			}
+
+			requests[id] = message.NewRequest(id, root, selector, priority, ext...)
 		}
 	}
 
@@ -188,10 +202,18 @@ func (mh *MessageHandler) fromIPLD(ibm *ipldbind.GraphSyncMessageRoot) (message.
 			if err != nil {
 				return message.GraphSyncMessage{}, err
 			}
-			responses[id] = message.NewResponse(id,
-				graphsync.ResponseStatusCode(res.Status),
-				res.Metadata,
-				res.Extensions.ToExtensionsList()...)
+
+			var md []message.GraphSyncLinkMetadatum
+			if res.Metadata != nil {
+				md = *res.Metadata
+			}
+
+			var ext []graphsync.ExtensionData
+			if res.Extensions != nil {
+				ext = res.Extensions.ToExtensionsList()
+			}
+
+			responses[id] = message.NewResponse(id, graphsync.ResponseStatusCode(res.Status), md, ext...)
 		}
 	}
 

message/v2/message_test.go

@@ -54,7 +54,7 @@ func TestAppendingRequests(t *testing.T) {
 	require.NoError(t, err)
 
 	gsrIpld := (*gsmIpld.Gs2.Requests)[0]
-	require.Equal(t, priority, gsrIpld.Priority)
+	require.Equal(t, priority, *gsrIpld.Priority)
 	require.Equal(t, request.Type(), graphsync.RequestTypeNew)
 	require.Equal(t, root, *gsrIpld.Root)
 	require.Equal(t, selector, *gsrIpld.Selector)

responsemanager/responsemanager_test.go

@@ -136,6 +136,11 @@ func TestCancellationQueryInProgress(t *testing.T) {
 
 	td.assertRequestCleared()
 
+	td.cancel()
+	select {
+	case <-td.ctx.Done():
+	}
+
 	tracing := td.collectTracing(t)
 	traceStrings := tracing.TracesToStrings()
 	require.Contains(t, traceStrings, "processRequests(0)")

responsemanager/server.go

@@ -52,8 +52,8 @@ func (rm *ResponseManager) terminateRequest(key responseKey) {
 	}
 	rm.connManager.Unprotect(key.p, key.requestID.Tag())
 	delete(rm.inProgressResponses, key)
-	ipr.cancelFn()
 	ipr.span.End()
+	ipr.cancelFn()
 }
 
 func (rm *ResponseManager) processUpdate(ctx context.Context, key responseKey, update gsmsg.GraphSyncRequest) {