fix: check for blob/accept receipts before blob/add is concluded

packages/upload-api/src/blob/add.js

@@ -5,7 +5,6 @@ import * as Blob from '@web3-storage/capabilities/blob'
 import * as W3sBlob from '@web3-storage/capabilities/web3.storage/blob'
 import * as HTTP from '@web3-storage/capabilities/http'
 import * as API from '../types.js'
-
 import { createConcludeInvocation } from '../ucan/conclude.js'
 import { AwaitError } from './lib.js'
 

packages/upload-client/package.json

@@ -94,6 +94,7 @@
     "@types/varint": "^6.0.1",
     "@ucanto/principal": "^9.0.1",
     "@ucanto/server": "^10.0.0",
+    "@web3-storage/content-claims": "^5.0.0",
     "@web3-storage/eslint-config-w3up": "workspace:^",
     "assert": "^2.0.0",
     "blockstore-core": "^3.0.0",

packages/upload-client/src/blob.js

@@ -1,14 +1,15 @@
+import { CAR } from '@ucanto/transport'
 import { sha256 } from 'multiformats/hashes/sha2'
 import { ed25519 } from '@ucanto/principal'
 import { conclude } from '@web3-storage/capabilities/ucan'
 import * as UCAN from '@web3-storage/capabilities/ucan'
-import { Receipt } from '@ucanto/core'
+import { Delegation, Receipt } from '@ucanto/core'
 import * as W3sBlobCapabilities from '@web3-storage/capabilities/web3.storage/blob'
 import * as BlobCapabilities from '@web3-storage/capabilities/blob'
 import * as HTTPCapabilities from '@web3-storage/capabilities/http'
 import { SpaceDID } from '@web3-storage/capabilities/utils'
 import retry, { AbortError } from 'p-retry'
-import { servicePrincipal, connection } from './service.js'
+import { receiptsEndpoint, servicePrincipal, connection } from './service.js'
 import { REQUEST_RETRIES } from './constants.js'
 
 /**
@@ -26,6 +27,40 @@ function createUploadProgressHandler(url, handler) {
   return onUploadProgress
 }
 
+// FIXME this code was copied over from w3up-client and modified to parameterise receiptsEndpoint
+/**
+ * Get a receipt for an executed task by its CID.
+ *
+ * @param {import('multiformats').UnknownLink} taskCid
+ * @param {import('./types.js').RequestOptions} [options]
+ */
+async function getReceipt(taskCid, options = {}) {
+  // Fetch receipt from endpoint
+  const url = new URL(
+    taskCid.toString(),
+    options.receiptsEndpoint ?? receiptsEndpoint
+  )
+  /* c8 ignore next */
+  const fetchReceipt = options.fetch ?? globalThis.fetch.bind(globalThis)
+  const workflowResponse = await fetchReceipt(url)
+  /* c8 ignore start */
+  if (!workflowResponse.ok) {
+    throw new Error(
+      `no receipt available for requested task ${taskCid.toString()}`
+    )
+  }
+  /* c8 ignore stop */
+  // Get receipt from Message Archive
+  const agentMessageBytes = new Uint8Array(await workflowResponse.arrayBuffer())
+  // Decode message
+  const agentMessage = await CAR.request.decode({
+    body: agentMessageBytes,
+    headers: {},
+  })
+  // Get receipt from the potential multiple receipts in the message
+  return agentMessage.receipts.get(taskCid.toString())
+}
+
 // FIXME this code has been copied over from upload-api
 /**
  * @param {import('@ucanto/interface').Invocation} concludeFx
@@ -166,7 +201,7 @@ export function createConcludeInvocation(id, serviceDid, receipt) {
  * The issuer needs the `blob/add` delegated capability.
  * @param {Blob|Uint8Array} data Blob data.
  * @param {import('./types.js').RequestOptions} [options]
- * @returns {Promise<import('multiformats').MultihashDigest>}
+ * @returns {Promise<import('./types.js').BlobAddOk>}
  */
 export async function add(
   { issuer, with: resource, proofs, audience },
@@ -303,7 +338,57 @@ export async function add(
     })
   }
 
-  return multihash
+  // Ensure the blob has been accepted
+  const acceptReceipt = await retry(
+    async () => {
+      try {
+        return await getReceipt(nextTasks.accept.task.link(), options)
+      } catch (err) {
+        throw new Error(`failed ${BlobCapabilities.add.can} invocation`, {
+          cause: err,
+        })
+      }
+    },
+    {
+      onFailedAttempt: console.warn,
+      retries: options.retries ?? REQUEST_RETRIES,
+    }
+  )
+
+  // @ts-ignore Property
+  if (!acceptReceipt?.out.ok?.site) {
+    throw new Error(`failed ${BlobCapabilities.add.can} invocation`, {
+      cause: 'failed to get blob/accept receipt',
+    })
+  }
+
+  const blocks = new Map(
+    [...acceptReceipt.iterateIPLDBlocks()].map((block) => [
+      `${block.cid}`,
+      block,
+    ])
+  )
+  const site = Delegation.view(
+    {
+      root: /** @type {import('@ucanto/interface').UCANLink} */ (
+        // @ts-ignore Property
+        acceptReceipt.out.ok.site
+      ),
+      blocks,
+    },
+    null
+  )
+  /* c8 ignore next 5 */
+  if (!site) {
+    throw new Error(`failed ${BlobCapabilities.add.can} invocation`, {
+      cause: 'failed to get blob/accept receipt',
+    })
+  }
+
+  return {
+    multihash,
+    site: site.link(),
+  }
 }
 
 /**

packages/upload-client/src/index.js

@@ -10,10 +10,10 @@ import * as Upload from './upload.js'
 import * as UnixFS from './unixfs.js'
 import * as CAR from './car.js'
 import { ShardingStream, defaultFileComparator } from './sharding.js'
-import { codec as carCodec } from '@ucanto/transport/car'
 
 export { Blob, Index, Store, Upload, UnixFS, CAR }
 export * from './sharding.js'
+export { receiptsEndpoint } from './service.js'
 
 /**
  * Uploads a file to the service and returns the root data CID for the
@@ -144,9 +144,9 @@ async function uploadBlockStream(
           async transform(car, controller) {
             const bytes = new Uint8Array(await car.arrayBuffer())
             // Invoke blob/add and write bytes to write target
-            const multihash = await Blob.add(conf, bytes, options)
+            const { multihash } = await Blob.add(conf, bytes, options)
             // Should this be raw instead?
-            const cid = Link.create(carCodec.code, multihash)
+            const cid = Link.create(CAR.code, multihash)
             let piece
             if (pieceHasher) {
               const multihashDigest = await pieceHasher.digest(bytes)
@@ -199,20 +199,15 @@ async function uploadBlockStream(
   /* c8 ignore next */
   if (!root) throw new Error('missing root CID')
 
-  const index = ShardedDAGIndex.create(root)
-  for (const [i, shard] of shards.entries()) {
-    const slices = shardIndexes[i]
-    index.shards.set(shard.multihash, slices)
-  }
-  const indexBytes = await index.archive()
+  const indexBytes = await indexShardedDAG(root, shards, shardIndexes)
   /* c8 ignore next 3 */
   if (!indexBytes.ok) {
     throw new Error('failed to archive DAG index', { cause: indexBytes.error })
   }
 
   // Store the index in the space
-  const indexDigest = await Blob.add(conf, indexBytes.ok, options)
-  const indexLink = Link.create(carCodec.code, indexDigest)
+  const { multihash } = await Blob.add(conf, indexBytes.ok, options)
+  const indexLink = Link.create(CAR.code, multihash)
 
   // Register the index with the service
   await Index.add(conf, indexLink, options)
@@ -221,3 +216,19 @@ async function uploadBlockStream(
 
   return root
 }
+
+/**
+ * Indexes a sharded DAG
+ *
+ * @param {import('multiformats').Link} root
+ * @param {import('./types.js').CARLink[]} shards
+ * @param {Array<Map<import('./types.js').SliceDigest, import('./types.js').Position>>} shardIndexes
+ */
+export async function indexShardedDAG(root, shards, shardIndexes) {
+  const index = ShardedDAGIndex.create(root)
+  for (const [i, shard] of shards.entries()) {
+    const slices = shardIndexes[i]
+    index.shards.set(shard.multihash, slices)
+  }
+  return await index.archive()
+}

packages/upload-client/src/service.js

@@ -4,6 +4,7 @@ import * as DID from '@ipld/dag-ucan/did'
 
 export const serviceURL = new URL('https://up.web3.storage')
 export const servicePrincipal = DID.parse('did:web:web3.storage')
+export const receiptsEndpoint = 'https://up.web3.storage/receipt/'
 
 /** @type {import('@ucanto/interface').ConnectionView<import('./types.js').Service>} */
 export const connection = connect({

packages/upload-client/src/types.ts

@@ -2,7 +2,13 @@ import type {
   FetchOptions as IpfsUtilsFetchOptions,
   ProgressStatus as XHRProgressStatus,
 } from 'ipfs-utils/src/types.js'
-import { Link, UnknownLink, Version, MultihashHasher } from 'multiformats'
+import {
+  MultihashDigest,
+  Link,
+  UnknownLink,
+  Version,
+  MultihashHasher,
+} from 'multiformats'
 import { Block } from '@ipld/unixfs'
 import {
   ServiceMethod,
@@ -12,6 +18,7 @@ import {
   DID,
   Principal,
   Failure,
+  UCANLink,
 } from '@ucanto/interface'
 import {
   UCANConclude,
@@ -305,6 +312,7 @@ export interface RequestOptions
     UploadProgressTrackable {
   fetch?: typeof globalThis.fetch
   nonce?: string
+  receiptsEndpoint?: URL
 }
 
 export interface ListRequestOptions extends RequestOptions, Pageable {}
@@ -372,3 +380,8 @@ export interface FileLike extends BlobLike {
    */
   name: string
 }
+
+export interface BlobAddOk {
+  multihash: MultihashDigest
+  site: UCANLink
+}