fix: race condition when requesting the same block twice

[achingbrain]

When we call blockstore.putMany, some implementations will batch up all the puts and write them at once. This means that blockstore.has might not return true for a little while - if another request for a given block comes in before blockstore.has returns true it'll get added to the want list. If the block then finishes it's batch and finally a remote peer supplies the wanted block, the notifications that complete the second block request will never get sent and the process will hang indefinitely.

The change made here is to separate the sending of notifications out from putting things into the blockstore. If the blockstore has a block, but the block is still in the wantlist, send notifications that we now have the block.

[achingbrain]

Fixes ipfs/js-ipfs#2814

[dirkmc]

It's my understanding that the async function asyncFn passed to blockstore.putMany(asyncFn) supplies the blocks to be put - but doesn't make guarantee about when those blocks will be commited to the blockstore.

So I think it may make more sense to send notifications after blockstore.putMany() has completed.

[achingbrain]

I thought similar, but you'd have to iterate over the list of blocks twice and they may not be available to the source any more (e.g. it may not be an array).

const iter = function * () {
  yield 'cid'
}

function main (source) {
  for (const cid of source) {
    console.info(cid)
  }

  for (const cid of source) {
    console.info(cid)
  }
}

main(iter())
// prints 'cid' once, not twice

Either that or you store the blocks in a list for processing after putMany, which may exhaust available memory, or you store just the CIDs and retrieve the blocks after the putMany, though they may not be available and the list may be unreasonably long, etc..

[dirkmc]

The solution I am using in the new message types PR is to store them in an array:
https://github.com/ipfs/js-ipfs-bitswap/pull/211/files/f2b6be5ffbae25bdd911496a356b02b4490714ce#diff-1fdf421c05c1140f6d71444ea2b27638R285

I don't think it will use much more memory, because it's just a list of pointers, right?

[achingbrain]

It's going to be a factor of the size of the added blocks - if you add a 10GB .iso file to IPFS, sticking them all in an array before sending notifications will cause the process to consume 10GB of memory.

If instead each block is processed one at a time (or in small batches) from reading the file to writing it out to disk it'll consume a lot less.

This isn't the only problem site, obviously - if our final blockstore batch size is unbounded it'll crash there too.

[achingbrain]

Also, I’m not sure bitswap is the right place to solve this problem. The blockstore has created it by making the behaviour of putMany non-obvious (to me at least), it should make the caller’s life a bit easier by limiting batch sizes, caching things that have been batched up, etc, though this is way out of scope for this PR.

[dirkmc]

IIUC the blocks are a parameter to bitswap.putMany(blocks), so they're not going to get cleaned up until the function returns. If we create an array of pointers to the blocks, that will use a negligible amount of memory compared to the blocks themselves.

I agree that the blockstore is probably a better place to tackle memory issues.

Just to clarify, when we call yield on line 292 I don't believe that means that the block has been committed to the blockstore. That's why I'm suggesting we only send out notifications after blockstore.putMany() completes.

[achingbrain]

bitswap.putMany(blocks) takes an iterable as an argument, which might be an array in which case yes, they are all in memory at once so storing references to them in an array does not hurt, but it also might be a readable stream or a generator in which case they are pulled one block at a time. It's for this case that we do not want to stick them all in an array.

[dirkmc]

Ahh I see you're right blocks is an async iterable.

For correctness I believe we need to send notifications after the block has been committed, ie at the end of blockstore.putMany().

In order to be able to send notifications we could check the type of blocks and then

• if it's an array: store a list of refs to each block that was put
• if it's an async iterable: store a list of CIDs then call datastore.getBlocks() for notifications. Probably most datastore implementations will have some caching that will make this perform reasonably well

[achingbrain]

Looking back on this blockstore.putMany() conflates batching with streaming which I think is a mistake.

The intention (as I see it) of putMany is to allow streaming blocks into the blockstore, but it can actually end up batching those blocks, worst case storing them until the stream ends - and in our case forcing the caller to store something to do with the whole stream (e.g. the list of CIDs to notify of)

I think we should be explicit in our interfaces and remove the batching semantic from blockstore.putMany() and instead expose a blockstore.putBatch() or similar method for when that behaviour is desirable.

The batching is done at the repo level, I'm more than happy to pull that out.

[achingbrain]

It's been pulled out. [email protected] .putMany() now streams to the datastore instead of using batches.

[dirkmc]

@achingbrain what do you think about this approach? #216

I still need to fix the test but want to get your thoughts first

[achingbrain]

Further improvements in this branch:

• Upgrade to use the streaming API from interface-datastore
  • Does not assume that only arrays of CIDs are being passed any more, only uses the AsyncIterable interface contract to access data
• Actually dial remote nodes with bitswap 1.2.0
• Stop using store.has as it'll be removed in the next interface-datastore release

[hugomrdias]

LGTM but i would keep .has and would try to find another way to surface the issues that come with relying in the .has output without a transaction

[hugomrdias]

LGTM great work!!

[achingbrain]

@dirkmc can this be merged & released?

[vasco-santos]

This gets fixed in #217

The peer address just needs to be added to the AddressBook before the operation

[achingbrain]

Could you create an issue to track this please?

[vasco-santos]

I already updated #217 to add the peer multiaddrs to the AddressBook before the sendMessage and removed the dial operation from the test.

We are removing the PeerInfo from [email protected]. multiaddrs for the peer should be in the AddressBook before any attempt to dial using a PeerId. Considering this case, the nodes needed to "discover" themselves, in order to send a message. The "discover" here is adding the multiaddrs to the AddressBook and then use the sendMessage with the PeerId.

So, the goal here is to use the new libp2p and not to accept PeerInfo in the sendMessage. I can create an issue to track this, but I am considering this as part of updating libp2p in bitswap.

it('dials to peer using Bitswap 1.2.0', async () => {
    networkA = new Network(p2pA, bitswapMockA)

    // only supports 1.2.0
    networkB = new Network(p2pB, bitswapMockB)
    networkB.protocols = ['/ipfs/bitswap/1.2.0']

    networkA.start()
    networkB.start()

    const deferred = pDefer()

    bitswapMockB._receiveMessage = () => {
      deferred.resolve()
    }

    p2pA.peerStore.addressBook.set(p2pB.peerId, p2pB.multiaddrs)
    await networkA.sendMessage(p2pB.peerId, new Message(true))

    return deferred
  })

[dirkmc]

Some good cleanup in here as well, thanks! 👍