Manifest block size (#53)

* add `blockSize` and `size` to manifest

* fix tests

* fix decoding/encoding

* re-export coders

* style

dagger/blocktype.nim

@@ -16,6 +16,9 @@ import pkg/questionable/results
 
 import ./errors
 
+const
+  BlockSize* = 4096 # file chunk read size
+
 type
   Block* = object of RootObj
     cid*: Cid
@@ -36,15 +39,17 @@ func init*(
     hash = ? MultiHash.digest($mcodec, data).mapFailure
     cid = ? Cid.init(version, codec, hash).mapFailure
 
-  success Block(
+  # TODO: If the hash is `>=` to the data,
+  # use the Cid as a container!
+  Block(
     cid: cid,
-    data: @data)
+    data: @data).success
 
 func init*(
   T: type Block,
   cid: Cid,
   data: openArray[byte],
-  verify: bool = false): ?!T =
+  verify: bool = true): ?!T =
 
   let
     mhash = ? cid.mhash.mapFailure

dagger/manifest.nim

@@ -1,3 +1,4 @@
+import ./manifest/coders
 import ./manifest/manifest
 
-export manifest
+export manifest, coders

dagger/manifest/coders.nim

@@ -17,7 +17,7 @@ import pkg/questionable/results
 import pkg/chronicles
 import pkg/chronos
 
-import ./types
+import ./manifest
 import ../errors
 
 const
@@ -27,21 +27,44 @@ type
   ManifestCoderType*[codec: static MultiCodec] = object
   DagPBCoder* = ManifestCoderType[multiCodec("dag-pb")]
 
-func encode*(_: DagPBCoder, b: Manifest): ?!seq[byte] =
+const
+  # TODO: move somewhere better?
+  ManifestContainers* = {
+    $DagPBCodec: DagPBCoder()
+  }.toTable
+
+func encode*(_: DagPBCoder, manifest: Manifest): ?!seq[byte] =
   ## Encode the manifest into a ``ManifestCodec``
   ## multicodec container (Dag-pb) for now
   ##
 
   var pbNode = initProtoBuffer()
 
-  for c in b.blocks:
+  for c in manifest.blocks:
     var pbLink = initProtoBuffer()
     pbLink.write(1, c.data.buffer) # write Cid links
     pbLink.finish()
     pbNode.write(2, pbLink)
 
-  let cid = !b.rootHash
+  # NOTE: The `Data` field in the the `dag-pb`
-  pbNode.write(1, cid.data.buffer) # set the rootHash Cid as the data field
+  # contains the following protobuf `Message`
+  #
+  # ```protobuf
+  #   Message Header {
+  #     optional bytes rootHash = 1;    # the root (tree) hash
+  #     optional uint32 blockSize = 2;  # size of a single block
+  #     optional uint32 blocksLen = 3;  # total amount of blocks
+  #   }
+  # ```
+  #
+
+  let cid = !manifest.rootHash
+  var header = initProtoBuffer()
+  header.write(1, cid.data.buffer)
+  header.write(2, manifest.blockSize.uint32)
+  header.write(3, manifest.blocks.len.uint32)
+
+  pbNode.write(1, header.buffer) # set the rootHash Cid as the data field
   pbNode.finish()
 
   return pbNode.buffer.success
@@ -52,13 +75,27 @@ func decode*(_: DagPBCoder, data: openArray[byte]): ?!Manifest =
 
   var
     pbNode = initProtoBuffer(data)
-    cidBuf: seq[byte]
+    pbHeader: ProtoBuffer
+    rootHash: seq[byte]
+    blockSize: uint32
+    blocksLen: uint32
     blocks: seq[Cid]
 
-  if pbNode.getField(1, cidBuf).isErr:
+  # Decode `Header` message
-    return failure("Unable to decode Cid from manifest!")
+  if pbNode.getField(1, pbHeader).isErr:
+    return failure("Unable to decode `Header` from dag-pb manifest!")
 
-  let cid = ? Cid.init(cidBuf).mapFailure
+  # Decode `Header` contents
+  if pbHeader.getField(1, rootHash).isErr:
+    return failure("Unable to decode `rootHash` from manifest!")
+
+  if pbHeader.getField(2, blockSize).isErr:
+    return failure("Unable to decode `blockSize` from manifest!")
+
+  if pbHeader.getField(3, blocksLen).isErr:
+    return failure("Unable to decode `blocksLen` from manifest!")
+
+  let rootHashCid = ? Cid.init(rootHash).mapFailure
   var linksBuf: seq[seq[byte]]
   if pbNode.getRepeatedField(2, linksBuf).isOk:
     for pbLinkBuf in linksBuf:
@@ -70,4 +107,31 @@ func decode*(_: DagPBCoder, data: openArray[byte]): ?!Manifest =
       if pbLink.getField(1, blockBuf).isOk:
         blocks.add(? Cid.init(blockBuf).mapFailure)
 
-  Manifest(rootHash: cid.some, blocks: blocks).success
+  if blocksLen.int != blocks.len:
+    return failure("Total blocks and length of blocks in header don't match!")
+
+  Manifest(
+    rootHash: rootHashCid.some,
+    blockSize: blockSize.int,
+    blocks: blocks,
+    hcodec: (? rootHashCid.mhash.mapFailure).mcodec,
+    codec: rootHashCid.mcodec,
+    version: rootHashCid.cidver).success
+
+proc encode*(self: var Manifest, encoder = ManifestContainers[$DagPBCodec]): ?!seq[byte] =
+  ## Encode a manifest using `encoder`
+  ##
+
+  if self.rootHash.isNone:
+    ? self.makeRoot()
+
+  encoder.encode(self)
+
+func decode*(
+  _: type Manifest,
+  data: openArray[byte],
+  decoder = ManifestContainers[$DagPBCodec]): ?!Manifest =
+  ## Decode a manifest using `decoder`
+  ##
+
+  decoder.decode(data)

dagger/manifest/manifest.nim

@@ -16,30 +16,61 @@ import pkg/questionable
 import pkg/questionable/results
 import pkg/chronicles
 
-import ./types
-import ./coders
 import ../errors
+import ../blocktype
 
-export coders, Manifest
+template EmptyDigests: untyped =
+  var
+    emptyDigests {.global, threadvar.}:
+      array[CIDv0..CIDv1, Table[MultiCodec, MultiHash]]
 
-const
+  once:
-  ManifestContainers* = {
+    emptyDigests = [
-    $DagPBCodec: DagPBCoder()
+      CIDv0: {
-  }.toTable
+        multiCodec("sha2-256"): Cid
+        .init("bafybeihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku")
+        .get()
+        .mhash
+        .get()
+      }.toTable,
+      CIDv1: {
+        multiCodec("sha2-256"): Cid
+        .init("QmdfTbBqBPQ7VNxZEYEj14VmRuZBkqFbiwReogJgS1zR1n")
+        .get()
+        .mhash
+        .get()
+      }.toTable,
+    ]
+
+  emptyDigests
+
+type
+  Manifest* = object of RootObj
+    rootHash*: ?Cid       # root (tree) hash of the contained data set
+    blockSize*: int       # size of each contained block (might not be needed if blocks are len-prefixed)
+    blocks*: seq[Cid]     # block Cid
+    version*: CidVersion  # Cid version
+    hcodec*: MultiCodec   # Multihash codec
+    codec*: MultiCodec    # Data set codec
 
 func len*(self: Manifest): int =
   self.blocks.len
 
+func size*(self: Manifest): int =
+  self.blocks.len * self.blockSize
+
 func `[]`*(self: Manifest, i: Natural): Cid =
   self.blocks[i]
 
 func `[]=`*(self: var Manifest, i: Natural, item: Cid) =
+  self.rootHash = Cid.none
   self.blocks[i] = item
 
 func `[]`*(self: Manifest, i: BackwardsIndex): Cid =
   self.blocks[self.len - i.int]
 
 func `[]=`*(self: var Manifest, i: BackwardsIndex, item: Cid) =
+  self.rootHash = Cid.none
   self.blocks[self.len - i.int] = item
 
 proc add*(self: var Manifest, cid: Cid) =
@@ -60,7 +91,11 @@ template hashBytes(mh: MultiHash): seq[byte] =
 
   mh.data.buffer[mh.dpos..(mh.dpos + mh.size - 1)]
 
-proc makeRoot(self: var Manifest): ?!void =
+proc makeRoot*(self: var Manifest): ?!void =
+  ## Create a tree hash root of the contained
+  ## block hashes
+  ##
+
   var
     stack: seq[MultiHash]
 
@@ -96,30 +131,13 @@ proc cid*(self: var Manifest): ?!Cid =
 
   (!self.rootHash).success
 
-proc encode*(self: var Manifest, encoder = ManifestContainers[$DagPBCodec]): ?!seq[byte] =
-  ## Encode a manifest using `encoder`
-  ##
-
-  if self.rootHash.isNone:
-    ? self.makeRoot()
-
-  encoder.encode(self)
-
-func decode*(
-  _: type Manifest,
-  data: openArray[byte],
-  decoder = ManifestContainers[$DagPBCodec]): ?!Manifest =
-  ## Decode a manifest using `decoder`
-  ##
-
-  decoder.decode(data)
-
 proc init*(
   T: type Manifest,
   blocks: openArray[Cid] = [],
   version = CIDv1,
   hcodec = multiCodec("sha2-256"),
-  codec = multiCodec("raw")): ?!T =
+  codec = multiCodec("raw"),
+  blockSize = BlockSize): ?!T =
   ## Create a manifest using array of `Cid`s
   ##
 
@@ -131,4 +149,10 @@ proc init*(
     version: version,
     codec: codec,
     hcodec: hcodec,
+    blockSize: blockSize
     ).success
+
+proc init*(
+  T: type Manifest,
+  data: openArray[byte]): ?!T =
+  Manifest.decode(data)

dagger/manifest/types.nim

@@ -1,46 +0,0 @@
-## Nim-Dagger
-## Copyright (c) 2022 Status Research & Development GmbH
-## Licensed under either of
-##  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
-##  * MIT license ([LICENSE-MIT](LICENSE-MIT))
-## at your option.
-## This file may not be copied, modified, or distributed except according to
-## those terms.
-
-{.push raises: [Defect].}
-
-import pkg/libp2p
-import pkg/questionable
-
-template EmptyDigests*: untyped =
-  var
-    emptyDigests {.global, threadvar.}:
-      array[CIDv0..CIDv1, Table[MultiCodec, MultiHash]]
-
-  once:
-    emptyDigests = [
-      CIDv0: {
-        multiCodec("sha2-256"): Cid
-        .init("bafybeihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku")
-        .get()
-        .mhash
-        .get()
-      }.toTable,
-      CIDv1: {
-        multiCodec("sha2-256"): Cid
-        .init("QmdfTbBqBPQ7VNxZEYEj14VmRuZBkqFbiwReogJgS1zR1n")
-        .get()
-        .mhash
-        .get()
-      }.toTable,
-    ]
-
-  emptyDigests
-
-type
-  Manifest* = object of RootObj
-    rootHash*: ?Cid
-    blocks*: seq[Cid]
-    version*: CidVersion
-    hcodec*: MultiCodec
-    codec*: MultiCodec

dagger/node.nim

@@ -29,9 +29,6 @@ import ./blockexchange
 logScope:
   topics = "dagger node"
 
-const
-  FileChunkSize* = 4096 # file chunk read size
-
 type
   DaggerError = object of CatchableError
 

tests/dagger/testnode.nim

@@ -112,7 +112,7 @@ suite "Test Node":
     var data: seq[byte]
     while true:
       var
-        buf = newSeq[byte](FileChunkSize)
+        buf = newSeq[byte](BlockSize)
         res = await stream.readOnce(addr buf[0], buf.len)
 
       if res <= 0: