feat: graphsync provider

Author: tchardin

package.json

@@ -14,6 +14,7 @@
     "@types/bl": "^5.0.2",
     "@types/mime": "^2.0.3",
     "@types/uuid": "^8.3.4",
+    "@types/varint": "^6.0.0",
     "aegir": "^36.2.3",
     "blockstore-core": "^1.0.5",
     "ipfs-unixfs-importer": "^9.0.6",
@@ -39,6 +40,7 @@
     "peer-id": "^0.16.0",
     "uint8arrays": "^3.0.0",
     "uuid": "^8.3.2",
+    "varint": "^6.0.0",
     "varint-decoder": "^1.0.0"
   }
 }

src/graphsync.ts

@@ -17,6 +17,7 @@ import {
   PROTOCOL,
   GraphSyncBlock,
   decodeBlock,
+  GraphSyncRequest,
   GraphSyncResponse,
   decodeMessage,
   newRequest,
@@ -28,6 +29,7 @@ import type {
   NodeReifier,
 } from "./traversal";
 import {Node, walkBlocks, parseContext, unixfsReifier} from "./traversal";
+import {responseBuilder} from "./response-builder";
 
 export class GraphSync {
   started = false;
@@ -83,6 +85,10 @@ export class GraphSync {
       req.incomingResponseHook(resp);
     }
   }
+  async _handleRequest(peer: PeerId, req: GraphSyncRequest) {
+    const {stream} = await this.network.dialProtocol(peer, PROTOCOL);
+    await pipe(responseBuilder(req, this.blocks), lp.encode(), stream);
+  }
   async _handler(props: HandlerProps) {
     const source = props.stream.source as AsyncIterable<BufferList>;
     for await (const chunk of lp.decode()(source)) {
@@ -96,6 +102,11 @@ export class GraphSync {
       if (msg.rsp) {
         msg.rsp.forEach((resp) => this._handleResponse(resp));
       }
+      if (msg.req) {
+        msg.req.forEach((req) =>
+          this._handleRequest(props.connection.remotePeer, req)
+        );
+      }
     }
   }
 }
@@ -127,14 +138,21 @@ export class Request extends EventEmitter {
     this.loader = new AsyncLoader(blocks, this.incomingBlockHook.bind(this));
   }
 
-  open(peer: PeerId, extensions?: {[key: string]: any}) {
-    this.loader.setWaitNotify(async () => {
-      this.loader.notifyWaiting = false;
-      const {stream} = await this.dialer.dialProtocol(peer, PROTOCOL);
-      await pipe(
-        [newRequest(this.id, this.root, this.selector, extensions)],
-        stream
-      );
+  open(peer: PeerId, extensions?: {[key: string]: any}): Promise<void> {
+    return new Promise((resolve, reject) => {
+      this.loader.setWaitNotify(async () => {
+        this.loader.notifyWaiting = false;
+        try {
+          const {stream} = await this.dialer.dialProtocol(peer, PROTOCOL);
+          pipe(
+            [newRequest(this.id, this.root, this.selector, extensions)],
+            stream
+          );
+          resolve();
+        } catch (e) {
+          reject(e);
+        }
+      });
     });
   }
 

src/messages.ts

@@ -4,8 +4,7 @@ import {Buffer} from "buffer";
 import {parse as uuidParse} from "uuid";
 import * as dagCBOR from "@ipld/dag-cbor";
 import type BufferList from "bl/BufferList";
-// @ts-ignore (no types)
-import vd from "varint-decoder";
+import * as varint from "varint";
 import lp from "it-length-prefixed";
 import {SelectorNode, decoderFor} from "./traversal";
 
@@ -72,7 +71,7 @@ export const statuses = {
   [ResponseStatusCode.RequestCancelled]: "RequestCancelled",
 };
 
-enum GraphSyncLinkAction {
+export enum GraphSyncLinkAction {
   Present = "p",
   DuplicateNotSent = "d",
   Missing = "m",
@@ -86,7 +85,7 @@ type GraphSyncPriority = number;
 
 type GraphSyncMetadatum = [CID, GraphSyncLinkAction];
 
-type GraphSyncMetadata = GraphSyncMetadatum[];
+export type GraphSyncMetadata = GraphSyncMetadatum[];
 
 enum GraphSyncRequestType {
   New = "n",
@@ -114,13 +113,13 @@ export type GraphSyncResponse = {
   ext?: GraphSyncExtensions;
 };
 
-type GraphSyncMessage = {
+export type GraphSyncMessage = {
   req?: GraphSyncRequest[];
   rsp?: GraphSyncResponse[];
   blk?: GraphSyncBlock[];
 };
 
-type GraphSyncMessageRoot = {
+export type GraphSyncMessageRoot = {
   gs2: GraphSyncMessage;
 };
 
@@ -130,20 +129,21 @@ export function newRequest(
   sel: SelectorNode,
   ext?: GraphSyncExtensions
 ): BufferList {
+  const req: GraphSyncRequest = {
+    id: uuidParse(id) as Uint8Array,
+    type: GraphSyncRequestType.New,
+    pri: 0,
+    root,
+    sel,
+  };
+  if (ext) {
+    req.ext = ext;
+  }
   return lp.encode.single(
     new Buffer(
-      dagCBOR.encode({
+      dagCBOR.encode<GraphSyncMessageRoot>({
         gs2: {
-          requests: [
-            {
-              id: uuidParse(id),
-              type: GraphSyncRequestType.New,
-              pri: 0,
-              root,
-              sel,
-              ext,
-            },
-          ],
+          req: [req],
         },
       })
     )
@@ -158,10 +158,12 @@ export async function decodeBlock(
   block: GraphSyncBlock,
   hashers: {[key: number]: hasher.MultihashHasher<any>}
 ): Promise<Block<any>> {
-  const values = vd(block[0]);
-  const cidVersion = values[0];
-  const multicodec = values[1];
-  const multihash = values[2];
+  let offset = 0;
+  const cidVersion = varint.decode(block[0], offset) as any;
+  offset += varint.decode.bytes;
+  const multicodec = varint.decode(block[0], offset);
+  offset += varint.decode.bytes;
+  const multihash = varint.decode(block[0], offset);
   const hasher = hashers[multihash];
   if (!hasher) {
     throw new Error("Unsuported hasher");

src/resolver.ts

@@ -6,6 +6,7 @@ import {
   allSelector,
   Node,
   Kind,
+  LinkLoader,
   parseContext,
   walkBlocks,
   SelectorNode,
@@ -65,50 +66,16 @@ export function getPeerID(addr: Multiaddr): PeerId {
   return PeerId.createFromB58String(parts[idx]);
 }
 
+// Iterate an IPLD traversal and resolve UnixFS blocks
 export async function* resolve(
-  path: string,
-  provider: Multiaddr,
-  exchange: GraphSync
-): AsyncIterable<any> {
-  const {root, selector: sel} = unixfsPathSelector(path);
-  const pid = getPeerID(provider);
-  exchange.network.peerStore.addressBook.add(pid, [provider]);
-  const request = exchange.request(root, sel);
-  const id = Date.now();
-  const voucher = {
-    ID: id,
-    PayloadCID: root,
-    Params: {
-      Selector: sel,
-      PieceCID: null,
-      PricePerByte: new Uint8Array(),
-      PaymentInterval: 1000,
-      PaymentIntervalIncrease: 0,
-      UnsealPrice: new Uint8Array(),
-    },
-  };
-  request.open(pid, {
-    [EXTENSION]: {
-      IsRq: true,
-      Request: {
-        BCid: root,
-        Type: 0,
-        Pull: true,
-        Paus: false,
-        Part: false,
-        Stor: sel,
-        Vouch: voucher,
-        VTyp: "RetrievalDealProposal/1",
-        XferID: id,
-        RestartChannel: ["", "", 0],
-      },
-      Response: null,
-    },
-  });
+  root: CID,
+  selector: SelectorNode,
+  loader: LinkLoader
+): AsyncIterable<Uint8Array> {
   for await (const blk of walkBlocks(
     new Node(root),
-    parseContext().parseSelector(sel),
-    request
+    parseContext().parseSelector(selector),
+    loader
   )) {
     // if not cbor or dagpb just return the bytes
     switch (blk.cid.code) {
@@ -133,8 +100,6 @@ export async function* resolve(
       // ignore
     }
   }
-  // tell the loader we're done receiving blocks for this traversal
-  request.close();
 }
 
 type FetchInit = {
@@ -145,7 +110,44 @@ type FetchInit = {
 
 export async function fetch(url: string, init: FetchInit): Promise<Response> {
   const {headers, exchange, provider} = init;
-  const content = resolve(url, provider, exchange);
+
+  const {root, selector: sel} = unixfsPathSelector(url);
+  const pid = getPeerID(provider);
+  exchange.network.peerStore.addressBook.add(pid, [provider]);
+  const request = exchange.request(root, sel);
+  const id = Date.now();
+  const voucher = {
+    ID: id,
+    PayloadCID: root,
+    Params: {
+      Selector: sel,
+      PieceCID: null,
+      PricePerByte: new Uint8Array(),
+      PaymentInterval: 1000,
+      PaymentIntervalIncrease: 0,
+      UnsealPrice: new Uint8Array(),
+    },
+  };
+  request.open(pid, {
+    [EXTENSION]: {
+      IsRq: true,
+      Request: {
+        BCid: root,
+        Type: 0,
+        Pull: true,
+        Paus: false,
+        Part: false,
+        Stor: sel,
+        Vouch: voucher,
+        VTyp: "RetrievalDealProposal/1",
+        XferID: id,
+        RestartChannel: ["", "", 0],
+      },
+      Response: null,
+    },
+  });
+
+  const content = resolve(root, sel, request);
   const iterator = content[Symbol.asyncIterator]();
 
   try {
@@ -171,6 +173,7 @@ export async function fetch(url: string, init: FetchInit): Promise<Response> {
         console.log(e);
         writer.abort((e as Error).message);
       }
+      request.close();
     }
     write();
     return new Response(readable, {