fix: update pin status from GET /pins even if failed (#2235)

Update pin status for pins regardless of status. Previously we only
updated if the state was positive.

Pickup will not retry pins that timeout, so the change here ensures we
inform users promptly where that happens.

Where the user has a pending pin, the request will now synchronously go
check cluster/pickup, update the db and return the response. Previously
the worker would respond early, and poll for a positive pin status from
the upstream for up to 30s. Just checking the upstream once per request
seems sufficient. We may want to rate limit this end point in the
future, but it's pretty low volume currently.

fixes #2234 

License: MIT

---------

Signed-off-by: Oli Evans <oli@protocol.ai>

Author: olizilla

packages/api/src/pins.js

@@ -1,6 +1,6 @@
 // @ts-nocheck
 import { JSONResponse } from './utils/json-response.js'
-import { getPins, PIN_OK_STATUS, waitAndUpdateOkPins } from './utils/pin.js'
+import { getPins, PIN_OK_STATUS, waitAndUpdateOkPins, fetchAndUpdatePins } from './utils/pin.js'
 import { PSAErrorDB, PSAErrorResourceNotFound, PSAErrorInvalidData, PSAErrorRequiredData, PinningServiceApiError } from './errors.js'
 import {
   INVALID_REQUEST_ID,
@@ -144,22 +144,11 @@ export async function pinGet (request, env, ctx) {
     throw new PSAErrorResourceNotFound()
   }
 
-  /** @type {(() => Promise<any>)[]} */
-  const tasks = []
-
   const inProgress = pinRequest.pins.filter((p) => p.status === 'PinQueued' || p.status === 'Pinning')
-  if (inProgress.length > 0) {
-    tasks.push(
-      waitAndUpdateOkPins.bind(
-        null,
-        pinRequest.contentCid,
-        env.cluster,
-        env.db)
-    )
-  }
 
-  if (ctx.waitUntil) {
-    tasks.forEach(t => ctx.waitUntil(t()))
+  if (inProgress.length > 0) {
+    await fetchAndUpdatePins(pinRequest.contentCid, env.cluster, env.db)
+    pinRequest = await env.db.getPsaPinRequest(authToken._id, request.params.requestId)
   }
 
   /** @type { PsaPinStatusResponse } */

packages/api/src/utils/pin.js

@@ -1,3 +1,4 @@
+import retry from 'p-retry'
 /**
  * @typedef {import('@nftstorage/ipfs-cluster').API.TrackerStatus} TrackerStatus
  * @typedef {'Undefined'
@@ -155,15 +156,32 @@ export async function waitOkPins (cid, cluster, waitTime = MAX_PIN_STATUS_CHECK_
  */
 export async function waitAndUpdateOkPins (cid, cluster, db, waitTime = MAX_PIN_STATUS_CHECK_TIME, checkInterval = PIN_STATUS_CHECK_INTERVAL) {
   const okPins = await waitOkPins(cid, cluster, waitTime, checkInterval)
-  const pins = okPins.map((pin) => {
-    return {
-      id: pin._id,
-      status: pin.status,
-      contentCid: cid,
-      location: pin.location
-    }
-  })
-  // @ts-ignore
+  const pins = toPinsUpsert(cid, okPins)
   await db.upsertPins(pins)
   return okPins
 }
+
+/**
+ * @param {string} contentCid
+ * @param {import('@web3-storage/db/db-client-types').PinUpsertInput[]} pins
+ * @returns {import('@web3-storage/db/db-client-types').PinsUpsertInput[]}
+ */
+export function toPinsUpsert (contentCid, pins) {
+  return pins.map(pin => ({
+    id: pin._id,
+    status: pin.status,
+    contentCid,
+    location: pin.location
+  }))
+}
+
+/**
+ * @param {string} contentCid
+ * @param {import('@nftstorage/ipfs-cluster').Cluster} cluster
+ * @param {import('@web3-storage/db').DBClient} db
+ */
+export async function fetchAndUpdatePins (contentCid, cluster, db) {
+  const statuses = await retry(() => getPins(contentCid, cluster), { retries: 3 })
+  const upserts = toPinsUpsert(contentCid, statuses)
+  return retry(() => db.upsertPins(upserts), { retries: 3 })
+}

packages/db/db-client-types.ts

@@ -95,7 +95,7 @@ export type PinItem = PinUpsertInput & {
 }
 
 export type PinsUpsertInput = {
-  id: string
+  id?: string
   status: definitions['pin']['status']
   contentCid: definitions['pin']['content_cid']
   location: Location