feat!: return allocated bytes in store/add receipt (storacha#1213)

This PR adds a new field to the `store/add` success result: `allocated:
number` - the total bytes allocated in the space to accommodate the
stored item, it **may be zero if the item is _already_ stored in _this_
space**.

This allows us to accurately report and bill for items stored in a
space. Currently, `store/add` when the item is already stored in the
space will cause a new diff to be created, effecitvely counting the same
shard multiple times. This could happen because of accidentally
uploading the same item, or because of a retry of a big item where some
shards were successful.

The following additional changes enable this functionality:

* `StoreTable` and `UploadTable` methods now have return types that are
`Result<O, X>` - this allows us to specify and communicate 2 errors
`RecordNotFound` and `RecordKeyConflict` (explained in the following
bullets).
* In the context of these 2 tables the semantics of `insert` have
changed to be more in line with postgres/other DBs - "insert" means add
to the DB or fail (with `RecordKeyConflict`) if an item with the same
key already exists.
* We can satisfy this constraint in dynamodb with [condition
expressions](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Expressions.ConditionExpressions.html#Expressions.ConditionExpressions.PreventingOverwrites)
or
[`ReturnValues`](https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_PutItem.html#DDB-PutItem-request-ReturnValues).
* In the `UploadTable` I've renamed `insert` to `upsert` since the
behaviour in use here is more akin to "update or insert" - aka
["upsert"](https://en.wiktionary.org/wiki/upsert).
* `remove` and `get` now fail with `RecordNotFound` if the item to
delete is not available.
* Again in dynamodb we can satisfy this constraint with condition
expressions or `ReturnValues`

---------

Co-authored-by: Vasco Santos <santos.vasco10@gmail.com>

packages/capabilities/src/types.ts

@@ -438,18 +438,34 @@ export type StoreRemove = InferInvokedCapability<typeof StoreCaps.remove>
 export type StoreList = InferInvokedCapability<typeof StoreCaps.list>
 
 export type StoreAddSuccess = StoreAddSuccessDone | StoreAddSuccessUpload
-export interface StoreAddSuccessDone {
-  status: 'done'
+
+export type StoreAddSuccessStatusUpload = 'upload'
+export type StoreAddSuccessStatusDone = 'done'
+
+export interface StoreAddSuccessResult {
+  /**
+   * Status of the item to store. A "done" status indicates that it is not
+   * necessary to upload the item. An "upload" status indicates that the item
+   * should be uploaded to the provided URL.
+   */
+  status: StoreAddSuccessStatusUpload | StoreAddSuccessStatusDone
+  /**
+   * Total bytes allocated in the space to accommodate this stored item.
+   * May be zero if the item is _already_ stored in _this_ space.
+   */
+  allocated: number
+  /** DID of the space this item will be stored in. */
   with: DID
+  /** CID of the item. */
   link: UnknownLink
-  url?: undefined
-  headers?: undefined
 }
 
-export interface StoreAddSuccessUpload {
-  status: 'upload'
-  with: DID
-  link: UnknownLink
+export interface StoreAddSuccessDone extends StoreAddSuccessResult {
+  status: StoreAddSuccessStatusDone
+}
+
+export interface StoreAddSuccessUpload extends StoreAddSuccessResult {
+  status: StoreAddSuccessStatusUpload
   url: ToString<URL>
   headers: Record<string, string>
 }
@@ -497,14 +513,7 @@ export type UploadAddSuccess = Omit<UploadListItem, 'insertedAt' | 'updatedAt'>
 
 export type UploadGetSuccess = UploadListItem
 
-export type UploadRemoveSuccess = UploadDidRemove | UploadDidNotRemove
-
-export interface UploadDidRemove extends UploadAddSuccess {}
-
-export interface UploadDidNotRemove {
-  root?: undefined
-  shards?: undefined
-}
+export type UploadRemoveSuccess = UploadAddSuccess
 
 export interface UploadListSuccess extends ListResponse<UploadListItem> {}
 

packages/eslint-config-w3up/index.js

@@ -10,7 +10,7 @@ module.exports = {
     EXPERIMENTAL_useProjectService: true,
   },
   rules: {
-    "@typescript-eslint/no-floating-promises": "error",
+    '@typescript-eslint/no-floating-promises': 'error',
     '@typescript-eslint/no-unused-vars': 'off',
     '@typescript-eslint/ban-ts-comment': 'off',
 

packages/filecoin-api/test/events/aggregator.js

@@ -738,7 +738,7 @@ export const test = {
           minPieceInsertedAt: new Date().toISOString(),
         }
         const putAggregateRes = await context.aggregateStore.put(
-          aggregateRecord,
+          aggregateRecord
         )
         assert.ok(putAggregateRes.ok)
 

packages/upload-api/src/admin/store/inspect.js

@@ -24,9 +24,7 @@ const inspect = async ({ capability }, context) => {
     return { error: new UnknownProvider(capability.with) }
   }
 
-  return {
-    ok: await context.storeTable.inspect(capability.nb.link),
-  }
+  return await context.storeTable.inspect(capability.nb.link)
 }
 
 class UnknownProvider extends Provider.Failure {

packages/upload-api/src/admin/upload/inspect.js

@@ -24,9 +24,7 @@ const inspect = async ({ capability }, context) => {
     return { error: new UnknownProvider(capability.with) }
   }
 
-  return {
-    ok: await context.uploadTable.inspect(capability.nb.root),
-  }
+  return await context.uploadTable.inspect(capability.nb.root)
 }
 
 class UnknownProvider extends Provider.Failure {

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

@@ -24,7 +24,7 @@ export function storeAddProvider(context) {
       Server.DID.parse(capability.with).did()
     )
     const issuer = invocation.issuer.did()
-    const [allocated, carIsLinkedToAccount, carExists] = await Promise.all([
+    const [allocated, carExists] = await Promise.all([
       allocate(
         {
           capability: {
@@ -33,7 +33,6 @@ export function storeAddProvider(context) {
         },
         context
       ),
-      storeTable.exists(space, link),
       carStoreBucket.has(link),
     ])
 
@@ -42,21 +41,30 @@ export function storeAddProvider(context) {
       return allocated
     }
 
-    if (!carIsLinkedToAccount) {
-      await storeTable.insert({
-        space,
-        link,
-        size,
-        origin,
-        issuer,
-        invocation: invocation.cid,
-      })
+    let allocatedSize = size
+    const res = await storeTable.insert({
+      space,
+      link,
+      size,
+      origin,
+      issuer,
+      invocation: invocation.cid,
+    })
+    if (res.error) {
+      // if the insert failed with conflict then this item has already been
+      // added to the space and there is no allocation change.
+      if (res.error.name === 'RecordKeyConflict') {
+        allocatedSize = 0
+      } else {
+        return res
+      }
     }
 
     if (carExists) {
       return {
         ok: {
           status: 'done',
+          allocated: allocatedSize,
           with: space,
           link,
         },
@@ -67,6 +75,7 @@ export function storeAddProvider(context) {
     return {
       ok: {
         status: 'upload',
+        allocated: allocatedSize,
         with: space,
         link,
         url: url.toString(),

packages/upload-api/src/store/get.js

@@ -1,6 +1,7 @@
 import * as Server from '@ucanto/server'
 import * as Store from '@web3-storage/capabilities/store'
 import * as API from '../types.js'
+import { StoreItemNotFound } from './lib.js'
 
 /**
  * @param {API.StoreServiceContext} context
@@ -14,16 +15,9 @@ export function storeGetProvider(context) {
     }
     const space = Server.DID.parse(capability.with).did()
     const res = await context.storeTable.get(space, link)
-    if (!res) {
-      return {
-        error: {
-          name: 'StoreItemNotFound',
-          message: 'Store item not found',
-        },
-      }
-    }
-    return {
-      ok: res,
+    if (res.error && res.error.name === 'RecordNotFound') {
+      return Server.error(new StoreItemNotFound(space, link))
     }
+    return res
   })
 }

packages/upload-api/src/store/lib.js

@@ -0,0 +1,29 @@
+import { Failure } from '@ucanto/server'
+
+export class StoreItemNotFound extends Failure {
+  /**
+   * @param {import('@ucanto/interface').DID} space
+   * @param {import('@ucanto/interface').UnknownLink} link
+   */
+  constructor(space, link) {
+    super()
+    this.space = space
+    this.link = link
+  }
+
+  get name() {
+    return 'StoreItemNotFound'
+  }
+
+  describe() {
+    return `${this.link} not found in ${this.space}`
+  }
+
+  toJSON() {
+    return {
+      ...super.toJSON(),
+      space: this.space,
+      link: { '/': this.link.toString() },
+    }
+  }
+}

packages/upload-api/src/store/list.js

@@ -10,13 +10,6 @@ export function storeListProvider(context) {
   return Server.provide(Store.list, async ({ capability }) => {
     const { cursor, size, pre } = capability.nb
     const space = Server.DID.parse(capability.with).did()
-
-    return {
-      ok: await context.storeTable.list(space, {
-        size,
-        cursor,
-        pre,
-      }),
-    }
+    return await context.storeTable.list(space, { size, cursor, pre })
   })
 }

packages/upload-api/src/store/remove.js

@@ -1,6 +1,7 @@
 import * as Server from '@ucanto/server'
 import * as Store from '@web3-storage/capabilities/store'
 import * as API from '../types.js'
+import { StoreItemNotFound } from './lib.js'
 
 /**
  * @param {API.StoreServiceContext} context
@@ -11,41 +12,11 @@ export function storeRemoveProvider(context) {
     const { link } = capability.nb
     const space = Server.DID.parse(capability.with).did()
 
-    const item = await context.storeTable.get(space, link)
-    if (!item) {
+    const res = await context.storeTable.remove(space, link)
+    if (res.error && res.error.name === 'RecordNotFound') {
       return Server.error(new StoreItemNotFound(space, link))
     }
 
-    await context.storeTable.remove(space, link)
-
-    return Server.ok({ size: item.size })
+    return res
   })
 }
-
-class StoreItemNotFound extends Server.Failure {
-  /**
-   * @param {import('@ucanto/interface').DID} space
-   * @param {import('@ucanto/interface').UnknownLink} link
-   */
-  constructor(space, link) {
-    super()
-    this.space = space
-    this.link = link
-  }
-
-  get name() {
-    return 'StoreItemNotFound'
-  }
-
-  describe() {
-    return `${this.link} not found in ${this.space}`
-  }
-
-  toJSON() {
-    return {
-      ...super.toJSON(),
-      space: this.space,
-      link: { '/': this.link.toString() },
-    }
-  }
-}

packages/upload-api/src/types.ts

@@ -422,28 +422,64 @@ export interface DudewhereBucket {
   put: (dataCid: string, carCid: string) => Promise<void>
 }
 
+/**
+ * Indicates the requested record was not present in the table.
+ */
+export interface RecordNotFound extends Failure {
+  name: 'RecordNotFound'
+}
+
+/**
+ * Indicates the inserted record key conflicts with an existing key of a record
+ * that already exists in the table.
+ */
+export interface RecordKeyConflict extends Failure {
+  name: 'RecordKeyConflict'
+}
+
 export interface StoreTable {
-  inspect: (link: UnknownLink) => Promise<StoreInspectSuccess>
-  exists: (space: DID, link: UnknownLink) => Promise<boolean>
-  get: (space: DID, link: UnknownLink) => Promise<StoreGetSuccess | undefined>
-  insert: (item: StoreAddInput) => Promise<StoreAddOutput>
-  remove: (space: DID, link: UnknownLink) => Promise<void>
+  inspect: (link: UnknownLink) => Promise<Result<StoreInspectSuccess, Failure>>
+  exists: (space: DID, link: UnknownLink) => Promise<Result<boolean, Failure>>
+  get: (
+    space: DID,
+    link: UnknownLink
+  ) => Promise<Result<StoreGetSuccess, RecordNotFound>>
+  /** Inserts an item in the table if it does not already exist. */
+  insert: (
+    item: StoreAddInput
+  ) => Promise<Result<StoreAddOutput, RecordKeyConflict>>
+  /** Removes an item from the table but fails if the item does not exist. */
+  remove: (
+    space: DID,
+    link: UnknownLink
+  ) => Promise<Result<StoreRemoveSuccess, RecordNotFound>>
   list: (
     space: DID,
     options?: ListOptions
-  ) => Promise<ListResponse<StoreListItem>>
+  ) => Promise<Result<ListResponse<StoreListItem>, Failure>>
 }
 
 export interface UploadTable {
-  inspect: (link: UnknownLink) => Promise<UploadInspectSuccess>
-  exists: (space: DID, root: UnknownLink) => Promise<boolean>
-  get: (space: DID, link: UnknownLink) => Promise<UploadGetSuccess | undefined>
-  insert: (item: UploadAddInput) => Promise<UploadAddSuccess>
-  remove: (space: DID, root: UnknownLink) => Promise<UploadRemoveSuccess | null>
+  inspect: (link: UnknownLink) => Promise<Result<UploadInspectSuccess, Failure>>
+  exists: (space: DID, root: UnknownLink) => Promise<Result<boolean, Failure>>
+  get: (
+    space: DID,
+    link: UnknownLink
+  ) => Promise<Result<UploadGetSuccess, RecordNotFound>>
+  /**
+   * Inserts an item in the table if it does not already exist or updates an
+   * existing item if it does exist.
+   */
+  upsert: (item: UploadAddInput) => Promise<Result<UploadAddSuccess, Failure>>
+  /** Removes an item from the table but fails if the item does not exist. */
+  remove: (
+    space: DID,
+    root: UnknownLink
+  ) => Promise<Result<UploadRemoveSuccess, RecordNotFound>>
   list: (
     space: DID,
     options?: ListOptions
-  ) => Promise<ListResponse<UploadListItem>>
+  ) => Promise<Result<ListResponse<UploadListItem>, Failure>>
 }
 
 export type SpaceInfoSuccess = {

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

@@ -30,7 +30,7 @@ export function uploadAddProvider(context) {
 
     const [res] = await Promise.all([
       // Store in Database
-      uploadTable.insert({
+      uploadTable.upsert({
         space,
         root,
         shards,
@@ -40,7 +40,7 @@ export function uploadAddProvider(context) {
       writeDataCidToCarCidsMapping(dudewhereBucket, root, shards),
     ])
 
-    return { ok: res }
+    return res
   })
 }