Currently the synchronization between block receipt and sessions is not quite air-tight, leading to some race conditions.

Background

When IPFS wants a block it

• Checks the blockstore
• If the block is not in the blockstore, asks a Bitswap Session for the block
  The Session
  • Subscribes to a "notifier" for updates about the block
  • Registers "interest" in the block with the SessionInterestManager

When Bitswap receives a block it

• Puts the block to the blockstore
• Notifies all Sessions that have subscribed to updates about the block (through the "notifier")
• Notifies all Sessions that have registered "interest" in the block

There are overlapping concerns handled in different places (SessionInterestManager vs "notifier") and non-atomic operations, and the code is complex and hard to follow.

Proposal

Create a unified Notifier interface that can synchronize operations and ensure atomicity.

Notifier

• NewSubscription(sesid uint64) Subscription
• Publish(msg ResponseMessage) error
  • Put blocks to blockstore
  • Inform listeners that are interested in blocks / HAVEs / DONT_HAVEs

Subscription

• Messages() <-chan ResponseMessage
• Subscribe(keys ...cid.Cid)
  • Listen for incoming messages with blocks / HAVEs / DONT_HAVEs for CIDs
    When messages are received, send them on the output channel
  • Check blockstore for each CID
    For each CID that is found in the blockstore
    • If there's still a listener for the CID, inform listener
    • If there's no listener (listener already received block) do nothing
• Unsubscribe(keys ...cid.Cid)
  • If there are no more sessions interested in the keys, send CANCEL to all peers
• Remove()
  • Remove all session subscriptions
  • If there are no more sessions interested in the keys, send CANCEL to all peers

ResponseMessage is a struct with public fields for blocks, HAVEs and DONT_HAVEs.

When IPFS asks a Bitswap Session for a block, the Session will call Subscribe() on the Notifier:

• When the Session gets HAVE it adds the peer to the session
• When the Session gets HAVE / DONT_HAVE it updates its want tables
• When Session gets a block it will send the block on the output channel

When a request is cancelled the Session will call Unsubscribe() for those keys.
When the Session shuts down it will call Unsubscribe() with the session id.