feat: support want-have & want-block in engine

Author: dirkmc

src/decision-engine/index.js

@@ -28,8 +28,8 @@ class Ledger {
     this.accounting.bytesRecv += n
   }
 
-  wants (cid, priority) {
-    this.wantlist.add(cid, priority)
+  wants (cid, priority, wantType) {
+    this.wantlist.add(cid, priority, wantType)
   }
 
   cancelWant (cid) {

src/decision-engine/ledger.js

@@ -66,11 +66,12 @@ class Bitswap {
 
   // handle messages received through the network
   async _receiveMessage (peerId, incoming) {
+    // TODO: Need to add block to blockstore _before_ calling engine.messageReceived
     try {
       await this.engine.messageReceived(peerId, incoming)
     } catch (err) {
-      // Only logging the issue to process as much as possible
-      // of the message. Currently `messageReceived` does not
+      // Log instead of throwing an error so as to process as much as
+      // possible of the message. Currently `messageReceived` does not
       // throw any errors, but this could change in the future.
       this._log('failed to receive message', incoming)
     }
@@ -279,22 +280,28 @@ class Bitswap {
   async putMany (blocks) { // eslint-disable-line require-await
     const self = this
 
-    return this.blockstore.putMany(async function * () {
+    // Add any new blocks to the blockstore
+    const newBlocks = []
+    await this.blockstore.putMany(async function * () {
       for await (const block of blocks) {
         if (await self.blockstore.has(block.cid)) {
           continue
         }
 
         yield block
-
-        self.notifications.hasBlock(block)
-        self.engine.receivedBlocks([block.cid])
-        // Note: Don't wait for provide to finish before returning
-        self.network.provide(block.cid).catch((err) => {
-          self._log.error('Failed to provide: %s', err.message)
-        })
+        newBlocks.push(block)
       }
     }())
+
+    // Notify listeners that we have received the new blocks
+    for (const block of newBlocks) {
+      self.notifications.hasBlock(block)
+      self.engine.receivedBlocks([block])
+      // Note: Don't wait for provide to finish before returning
+      self.network.provide(block.cid).catch((err) => {
+        self._log.error('Failed to provide: %s', err.message)
+      })
+    }
   }
 
   /**

src/index.js

@@ -3,9 +3,10 @@
 const WantlistEntry = require('../wantlist').Entry
 
 module.exports = class BitswapMessageEntry {
-  constructor (cid, priority, cancel) {
-    this.entry = new WantlistEntry(cid, priority)
+  constructor (cid, priority, wantType, cancel, sendDontHave) {
+    this.entry = new WantlistEntry(cid, priority, wantType)
     this.cancel = Boolean(cancel)
+    this.sendDontHave = Boolean(sendDontHave)
   }
 
   get cid () {
@@ -24,13 +25,23 @@ module.exports = class BitswapMessageEntry {
     this.entry.priority = val
   }
 
+  get wantType () {
+    return this.entry.wantType
+  }
+
+  set wantType (val) {
+    this.entry.wantType = val
+  }
+
   get [Symbol.toStringTag] () {
     const cidStr = this.cid.toString('base58btc')
     return `BitswapMessageEntry ${cidStr} <cancel: ${this.cancel}, priority: ${this.priority}>`
   }
 
   equals (other) {
     return (this.cancel === other.cancel) &&
+           (this.sendDontHave === other.sendDontHave) &&
+           (this.wantType === other.wantType) &&
            this.entry.equals(other.entry)
   }
 }

src/types/message/entry.js

@@ -15,24 +15,43 @@ class BitswapMessage {
     this.full = full
     this.wantlist = new Map()
     this.blocks = new Map()
+    this.blockPresences = new Map()
     this.pendingBytes = 0
   }
 
   get empty () {
     return this.blocks.size === 0 &&
-           this.wantlist.size === 0
+           this.wantlist.size === 0 &&
+           this.blockPresences.size === 0
   }
 
-  addEntry (cid, priority, cancel) {
+  addEntry (cid, priority, wantType, cancel, sendDontHave) {
+    if (wantType == null) {
+      wantType = BitswapMessage.WantType.Block
+    }
     const cidStr = cid.toString('base58btc')
 
     const entry = this.wantlist.get(cidStr)
 
     if (entry) {
-      entry.priority = priority
-      entry.cancel = Boolean(cancel)
+      // Only change priority if want is of the same type
+      if (entry.wantType === wantType) {
+        entry.priority = priority
+      }
+      // Only change from "dont cancel" to "do cancel"
+      if (cancel) {
+        entry.cancel = Boolean(cancel)
+      }
+      // Only change from "dont send" to "do send" DONT_HAVE
+      if (sendDontHave) {
+        entry.sendDontHave = Boolean(sendDontHave)
+      }
+      // want-block overrides existing want-have
+      if (wantType === BitswapMessage.WantType.Block && entry.wantType === BitswapMessage.WantType.Have) {
+        entry.wantType = wantType
+      }
     } else {
-      this.wantlist.set(cidStr, new Entry(cid, priority, cancel))
+      this.wantlist.set(cidStr, new Entry(cid, priority, wantType, cancel, sendDontHave))
     }
   }
 
@@ -41,10 +60,24 @@ class BitswapMessage {
     this.blocks.set(cidStr, block)
   }
 
+  addHave (cid) {
+    const cidStr = cid.toString('base58btc')
+    if (!this.blockPresences.has(cidStr)) {
+      this.blockPresences.set(cidStr, BitswapMessage.BlockPresenceType.Have)
+    }
+  }
+
+  addDontHave (cid) {
+    const cidStr = cid.toString('base58btc')
+    if (!this.blockPresences.has(cidStr)) {
+      this.blockPresences.set(cidStr, BitswapMessage.BlockPresenceType.DontHave)
+    }
+  }
+
   cancel (cid) {
     const cidStr = cid.toString('base58btc')
     this.wantlist.delete(cidStr)
-    this.addEntry(cid, 0, true)
+    this.addEntry(cid, 0, BitswapMessage.WantType.Block, true, false)
   }
 
   setPendingBytes (size) {
@@ -106,6 +139,13 @@ class BitswapMessage {
       })
     })
 
+    for (const [cidStr, bpType] of this.blockPresences) {
+      msg.blockPresences.push({
+        cid: new CID(cidStr).buffer,
+        type: bpType
+      })
+    }
+
     if (this.pendingBytes > 0) {
       msg.pendingBytes = this.pendingBytes
     }
@@ -115,8 +155,10 @@ class BitswapMessage {
 
   equals (other) {
     if (this.full !== other.full ||
+        this.pendingBytes !== other.pendingBytes ||
         !isMapEqual(this.wantlist, other.wantlist) ||
-        !isMapEqual(this.blocks, other.blocks)
+        !isMapEqual(this.blocks, other.blocks) ||
+        !isMapEqual(this.blockPresences, other.blockPresences)
     ) {
       return false
     }
@@ -141,7 +183,7 @@ BitswapMessage.deserialize = async (raw) => {
     decoded.wantlist.entries.forEach((entry) => {
       // note: entry.block is the CID here
       const cid = new CID(entry.block)
-      msg.addEntry(cid, entry.priority, entry.cancel)
+      msg.addEntry(cid, entry.priority, entry.wantType, entry.cancel, entry.sendDontHave)
     })
   }
 
@@ -178,5 +220,18 @@ BitswapMessage.deserialize = async (raw) => {
   return msg
 }
 
+BitswapMessage.blockPresenceSize = (cid) => {
+  // It doesn't really matter if this is not exactly right
+  return cid.buffer.length + 1
+}
+
 BitswapMessage.Entry = Entry
+BitswapMessage.WantType = {
+  Block: pbm.Message.Wantlist.WantType.Block,
+  Have: pbm.Message.Wantlist.WantType.Have
+}
+BitswapMessage.BlockPresenceType = {
+  Have: pbm.Message.BlockPresenceType.Have,
+  DontHave: pbm.Message.BlockPresenceType.DontHave
+}
 module.exports = BitswapMessage

src/types/message/index.js

@@ -1,12 +1,13 @@
 'use strict'
 
 class WantListEntry {
-  constructor (cid, priority) {
+  constructor (cid, priority, wantType) {
     // Keep track of how many requests we have for this key
     this._refCounter = 1
 
     this.cid = cid
     this.priority = priority || 1
+    this.wantType = wantType
   }
 
   inc () {
@@ -30,7 +31,8 @@ class WantListEntry {
   equals (other) {
     return (this._refCounter === other._refCounter) &&
       this.cid.equals(other.cid) &&
-      this.priority === other.priority
+      this.priority === other.priority &&
+      this.wantType === other.wantType
   }
 }
 

src/types/wantlist/entry.js

@@ -13,15 +13,23 @@ class Wantlist {
     return this.set.size
   }
 
-  add (cid, priority) {
+  add (cid, priority, wantType) {
+    // Have to import here to avoid circular reference
+    const Message = require('../message')
+
     const cidStr = cid.toString('base58btc')
     const entry = this.set.get(cidStr)
 
     if (entry) {
       entry.inc()
       entry.priority = priority
+
+      // We can only overwrite want-have with want-block
+      if (entry.wantType === Message.WantType.Have && wantType === Message.WantType.Block) {
+        entry.wantType = wantType
+      }
     } else {
-      this.set.set(cidStr, new Entry(cid, priority))
+      this.set.set(cidStr, new Entry(cid, priority, wantType))
       if (this._stats) {
         this._stats.push(null, 'wantListSize', 1)
       }

src/types/wantlist/index.js

@@ -20,7 +20,7 @@ module.exports = class WantManager {
 
   _addEntries (cids, cancel, force) {
     const entries = cids.map((cid, i) => {
-      return new Message.Entry(cid, CONSTANTS.kMaxPriority - i, cancel)
+      return new Message.Entry(cid, CONSTANTS.kMaxPriority - i, Message.WantType.Block, cancel)
     })
 
     entries.forEach((e) => {

src/want-manager/index.js

@@ -79,8 +79,8 @@ describe('bitswap with mocks', function () {
 
       const msg = new Message(false)
 
-      msg.addEntry(b1.cid, 1, false)
-      msg.addEntry(b2.cid, 1, false)
+      msg.addEntry(b1.cid, 1)
+      msg.addEntry(b2.cid, 1)
 
       await bs._receiveMessage(other, msg)