chore: handle graphs with data on internal and leaf nodes

Author: achingbrain

src/exporter/file.js

@@ -60,7 +60,6 @@ function streamBytes (dag, node, fileSize, offset, length) {
   }
 
   const end = offset + length
-  let streamPosition = 0
 
   function getData ({ node, start }) {
     try {
@@ -70,18 +69,24 @@ function streamBytes (dag, node, fileSize, offset, length) {
         return Buffer.alloc(0)
       }
 
-      const block = extractDataFromBlock(file.data, start, offset, end)
-
-      streamPosition += block.length
-
-      return block
+      return extractDataFromBlock(file.data, start, offset, end)
     } catch (error) {
       throw new Error(`Failed to unmarshal node - ${error.message}`)
     }
   }
 
+  // as we step through the children, keep track of where we are in the stream
+  // so we can filter out nodes we're not interested in
+  let streamPosition = 0
+
   function visitor ({ node }) {
     const file = UnixFS.unmarshal(node.data)
+    const nodeHasData = Boolean(file.data && file.data.length)
+
+    // handle case where data is present on leaf nodes and internal nodes
+    if (nodeHasData && node.links.length) {
+      streamPosition += file.data.length
+    }
 
     // work out which child nodes contain the requested data
     const filteredLinks = node.links
@@ -96,7 +101,7 @@ function streamBytes (dag, node, fileSize, offset, length) {
 
         return child
       })
-      .filter((child, index) => {
+      .filter((child) => {
         return (offset >= child.start && offset < child.end) || // child has offset byte
           (end > child.start && end <= child.end) || // child has end byte
           (offset < child.start && end > child.end) // child is between offset and end bytes

test/exporter.js

@@ -14,12 +14,17 @@ const CID = require('cids')
 const loadFixture = require('aegir/fixtures')
 const doUntil = require('async/doUntil')
 const waterfall = require('async/waterfall')
+const parallel = require('async/parallel')
 const series = require('async/series')
 const fs = require('fs')
 const path = require('path')
 const push = require('pull-pushable')
 const toPull = require('stream-to-pull-stream')
 const toStream = require('pull-stream-to-stream')
+const {
+  DAGNode,
+  DAGLink
+} = require('ipld-dag-pb')
 
 const unixFSEngine = require('./../src')
 const exporter = unixFSEngine.exporter
@@ -635,6 +640,79 @@ module.exports = (repo) => {
         })
       )
     })
+
+    it('exports file with data on internal and leaf nodes', function (done) {
+      waterfall([
+        (cb) => createAndPersistNode(ipld, 'raw', [0x04, 0x05, 0x06, 0x07], [], cb),
+        (leaf, cb) => createAndPersistNode(ipld, 'file', [0x00, 0x01, 0x02, 0x03], [
+          leaf
+        ], cb),
+        (file, cb) => {
+          pull(
+            exporter(file.multihash, ipld),
+            pull.asyncMap((file, cb) => readFile(file, cb)),
+            pull.through(buffer => {
+              expect(buffer).to.deep.equal(Buffer.from([0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07]))
+            }),
+            pull.collect(cb)
+          )
+        }
+      ], done)
+    })
+
+    it('exports file with data on some internal and leaf nodes', function (done) {
+      // create a file node with three children:
+      // where:
+      //   i = internal node without data
+      //   d = internal node with data
+      //   l = leaf node with data
+      //             i
+      //          /  |  \
+      //         l   d   i
+      //             |     \
+      //             l      l
+      waterfall([
+        (cb) => {
+          // create leaves
+          parallel([
+            (next) => createAndPersistNode(ipld, 'raw', [0x00, 0x01, 0x02, 0x03], [], next),
+            (next) => createAndPersistNode(ipld, 'raw', [0x08, 0x09, 0x10, 0x11], [], next),
+            (next) => createAndPersistNode(ipld, 'raw', [0x12, 0x13, 0x14, 0x15], [], next)
+          ], cb)
+        },
+        (leaves, cb) => {
+          parallel([
+            (next) => createAndPersistNode(ipld, 'raw', [0x04, 0x05, 0x06, 0x07], [leaves[1]], next),
+            (next) => createAndPersistNode(ipld, 'raw', null, [leaves[2]], next)
+          ], (error, internalNodes) => {
+            if (error) {
+              return cb(error)
+            }
+
+            createAndPersistNode(ipld, 'file', null, [
+              leaves[0],
+              internalNodes[0],
+              internalNodes[1]
+            ], cb)
+          })
+        },
+        (file, cb) => {
+          pull(
+            exporter(file.multihash, ipld),
+            pull.asyncMap((file, cb) => readFile(file, cb)),
+            pull.through(buffer => {
+              expect(buffer).to.deep.equal(
+                Buffer.from([
+                  0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07,
+                  0x08, 0x09, 0x10, 0x11, 0x12, 0x13, 0x14, 0x15
+                ])
+              )
+            }),
+            pull.collect(cb)
+          )
+        }
+      ], done)
+    })
   })
 }
 
@@ -670,3 +748,26 @@ function readFile (file, done) {
     })
   )
 }
+
+function createAndPersistNode (ipld, type, data, children, callback) {
+  const file = new UnixFS(type, data ? Buffer.from(data) : undefined)
+  const links = []
+
+  children.forEach(child => {
+    const leaf = UnixFS.unmarshal(child.data)
+
+    file.addBlockSize(leaf.fileSize())
+
+    links.push(new DAGLink('', child.size, child.multihash))
+  })
+
+  DAGNode.create(file.marshal(), links, (error, node) => {
+    if (error) {
+      return callback(error)
+    }
+
+    ipld.put(node, {
+      cid: new CID(node.multihash)
+    }, (error) => callback(error, node))
+  })
+}