feat: add block session support to @helia/interface

[achingbrain]

There are no implementations yet but the usage pattern will be something like:

// unixfs cat command
export async function * cat (cid: CID, blockstore: Blocks, options: Partial<CatOptions> = {}): AsyncIterable<Uint8Array> {
  // create a session for the CID if support is available
  const blocks = await (blockstore.createSession != null ? blockstore.createSession(cid, options) : blockstore)
  const opts: CatOptions = mergeOptions(defaultOptions, options)

  // resolve and export using the session, if created, otherwise fall back to regular blockstore access
  const resolved = await resolve(cid, opts.path, blocks, opts)
  const result = await exporter(resolved.cid, blocks, opts)

  if (result.type !== 'file' && result.type !== 'raw') {
    throw new NotAFileError()
  }

  if (result.content == null) {
    throw new NoContentError()
  }

  yield * result.content(opts)
}

Alternatively the user can control session creation:

import { unixfs } from '@helia/unixfs'
import { createHelia } from 'helia' // or http
import { CID } from 'multiformats/cid'

const node = await createHelia()
const rootCid = CID.parse('Qmfoo')
const sessionBlockstore = await node.blockstore.createSession(rootCid, {
  signal: AbortSignal.timeout(5000)
})

// all operations will use the same session
const fs = unixfs({ blockstore: sessionBlockstore })

for await (const entry of fs.ls(rootCid) {
  if (entry.type !== 'file') {
    continue
  }

  for await (const buf of fs.cat(entry.cid)) {
    // ...
  }
}

Removes the BlockAnnouncer/BlockRetriever single-method interface BlockBroker split because we would have to add another BlockSessionFactory interface for this which starts getting unwieldy. Instead just have all the methods be optional and filter the brokers before use.

[SgtPooki]

+1 to combining blockBroker method types.

Could you explain how a session would become aware of providers obtained from a delegated-routing call?

[SgtPooki]

To be clear, the signal passed here is for the entire retrival session?

E.g. Your example would abort if the entire session lasted longer than 5000ms, correct?

  const sessionBlockstore = await node.blockstore.createSession(rootCid, {
  signal: AbortSignal.timeout(5000)
})

[SgtPooki]

Also, do we want to return Blockstore explitly and not Blocks?

[achingbrain]

To be clear, the signal passed here is for the entire retrival session?

No, the signal is just for the session creation, e.g. when we do a findProvs and send WANTHAVE to create the initial set of providers.

Also, do we want to return Blockstore explitly and not Blocks?

It returns Blockstore intentionally so that you can't create a session from a session.

[SgtPooki]

Not really related to this section of code but: how do these session blockstores get providers from delegated-routing to the sessions?

[achingbrain]

Similar to #398 (comment) - you accept an instance of the Helia Routing as a component and query it:

E.g.:

import { createHeliaHttp from } '@helia/http'
import { createDelegatedRoutingV1HttpApiClient } from '@helia/delegated-routing-v1-http-api-client'
import { Routing } from '@helia/interface'

interface MyBlockBrokerComponents {
  routing: Routing
}

interface MyBlockBrokerInit {
  // broker-specific config
}

class MyBlockBroker {
  private routing: Routing

  constructor (components: MyBlockBrokerComponents, init: MyBlockBrokerInit) {
    this.routing = components.routing
  }
  
  createSession (root: CID): Promise<BlockBroker> {
    for await (const prov of this.routing.findProviders(root)) {
      if (prov.protocols.includes('my-block-broker-protocol')) {
        // do block broker stuff
      }
    }

   // ...more code here
  }
}

function createMyBlockBroker(init: MyBlockBrokerInit) {
  return (components) => new MyBlockBroker(components, init)
}

const node = createHeliaHttp({
  blockBrokers: [
    createMyBlockBroker({
      // ...init stuff here
    })
  ]
  routers: [
    createDelegatedRoutingV1HttpApiClient('https://example.com')
  ]
})

// this will call the broker's createSession method
const sessionBlockstore = await node.blockstore.createSesssion(cid)

[aschmahmann]

Out of curiosity what's the reason for putting the session information in the blockstore rather than passing it along with say the signals?

My concern here (which might not be valid) is that it feels like we did something similar in the Go code and needing users to remember to create new block fetching stacks for every session ended up leading to the sessions not being plumbed through the stack well and therefore not having one "code session" per "logical session".

We're currently trying to alter how we do this to allow it to happen via signal (or "context" in Go). See ipfs/boxo#567 and ipfs/kubo#7198.

Every language/environment is different so maybe these concerns would be less likely to surface here.

[achingbrain]

My feeling is that if people are using the API incorrectly it's probably because the correct usage isn't obvious enough, either through basic things like argument names and positions or documentation or both.

We already have people not passing AbortSignals into long-lived operations so I think a session identifier there would also get missed.

The primary interaction point for people with DAGs of various shapes are the @helia/* modules. Since these are (mostly?) stateless they are essentially disposable so I think pushing the session/no session decision to the user (e.g. do I use helia.blockstore or helia.blockstore.createSession(rootCid)) and having them interact with separate @helia/unixfs instances for each session will create a smaller amount of cognitive load.

[achingbrain]

Could you explain how a session would become aware of providers obtained from a delegated-routing call?

HTTP Delegated Routers are configured as a router, and when you use the routing methods it'll query them.

E.g.:

import { createHeliaHttp from } '@helia/http'
import { createDelegatedRoutingV1HttpApiClient } from '@helia/delegated-routing-v1-http-api-client'

const node = createHeliaHttp({
  routers: [
    createDelegatedRoutingV1HttpApiClient('https://example.com')
  ]
})

// this will call the http router internally:
for await (const prov of node.routing.findProviders(cid)) {
  if (prov.protocols.includes('transport-ipfs-gateway-http')) {
    // do http things
  }

  if (prov.protocols.includes('transport-bitswap')) {
    // do bitswap things
  }

  // etc
}

[SgtPooki]

approved, but left some comments for improvement

[SgtPooki]

do we want allSettled instead of all? do we want a put to fail if a single announcement fails?

[achingbrain]

Perhaps allSettled. Maybe we'd want to throw if all blocker provers failed to announce?

[SgtPooki]

@achingbrain anything blocking the merging of this? were you able to test this successfully in helia-http-gateway or should I take a look at that?