persistent-storage file download

.github/workflows/ci.yml

@@ -154,7 +154,6 @@ jobs:
           bash -x start_ocean.sh 2>&1 > start-node.log &
         env:
           NODE_VERSION: main
-
       - name: Install deps & build
         run: npm ci && npm run build:metadata
 

src/@types/Provider.ts

@@ -141,6 +141,7 @@ export const PROTOCOL_COMMANDS = {
   PERSISTENT_STORAGE_UPLOAD_FILE: 'persistentStorageUploadFile',
   PERSISTENT_STORAGE_GET_FILE_OBJECT: 'persistentStorageGetFileObject',
   PERSISTENT_STORAGE_DELETE_FILE: 'persistentStorageDeleteFile',
+  PERSISTENT_STORAGE_DOWNLOAD_FILE: 'persistentStorageDownloadFile',
   SERVICE_GET_TEMPLATES: 'serviceGetTemplates',
   SERVICE_START: 'serviceStart',
   SERVICE_STOP: 'serviceStop',

src/services/providers/BaseProvider.ts

@@ -1123,6 +1123,35 @@ export class BaseProvider {
     )
   }
 
+  /**
+   * Downloads a file stored in a persistent-storage bucket, dispatching to the HTTP
+   * or P2P transport based on `nodeUri`.
+   * @param {OceanNode} nodeUri The provider target (HTTP URL, or peerId / multiaddr for P2P).
+   * @param {SignerOrAuthTokenOrSignature} signerOrAuthToken Signer, JWT auth token, or precomputed signature used to authenticate the request.
+   * @param {string} bucketId The bucket holding the file.
+   * @param {string} fileName The name of the file to download.
+   * @param {number} [offset=0] Byte offset to resume the download from. Must be a non-negative safe integer. Over HTTP this becomes a `Range` request (server must answer 206 starting at `offset`); over P2P it is sent in the request payload.
+   * @param {AbortSignal} [signal] Abort signal that cancels the download.
+   * @return {Promise<ComputeResultStream>} An async-iterable stream of the file body starting at `offset`.
+   */
+  public async downloadPersistentStorageFile(
+    nodeUri: OceanNode,
+    signerOrAuthToken: SignerOrAuthTokenOrSignature,
+    bucketId: string,
+    fileName: string,
+    offset: number = 0,
+    signal?: AbortSignal
+  ): Promise<ComputeResultStream> {
+    return this.getImpl(nodeUri).downloadPersistentStorageFile(
+      nodeUri,
+      signerOrAuthToken,
+      bucketId,
+      fileName,
+      offset,
+      signal
+    )
+  }
+
   // ── Service on Demand ────────────────────────────────────────────────
 
   public async getServiceTemplates(

src/services/providers/HttpProvider.ts

@@ -1592,6 +1592,72 @@ export class HttpProvider {
     return response.json()
   }
 
+  /**
+   * Downloads a file stored in a persistent-storage bucket over HTTP.
+   * @param {string} nodeUri The provider URI.
+   * @param {SignerOrAuthTokenOrSignature} signerOrAuthToken Signer, JWT auth token, or precomputed signature used to authenticate the request.
+   * @param {string} bucketId The bucket holding the file.
+   * @param {string} fileName The name of the file to download.
+   * @param {number} [offset=0] Byte offset to resume the download from. When greater than 0 a `Range: bytes=<offset>-` header is sent and the server must answer with `206 Partial Content` starting at that offset. Must be a non-negative safe integer.
+   * @param {AbortSignal} [signal] Abort signal that cancels the in-flight request.
+   * @return {Promise<ComputeResultStream>} An async-iterable stream of the file body starting at `offset`.
+   */
+  public async downloadPersistentStorageFile(
+    nodeUri: string,
+    signerOrAuthToken: SignerOrAuthTokenOrSignature,
+    bucketId: string,
+    fileName: string,
+    offset: number = 0,
+    signal?: AbortSignal
+  ): Promise<ComputeResultStream> {
+    if (!Number.isSafeInteger(offset) || offset < 0) {
+      throw new Error(`Invalid offset: ${offset}. Must be a non-negative safe integer.`)
+    }
+    const routeBase =
+      this.baseUrl(nodeUri) +
+      `/api/services/persistentStorage/buckets/${encodeURIComponent(
+        bucketId
+      )}/files/${encodeURIComponent(fileName)}`
+    const authPayload = await this.getSignedCommandParams(
+      nodeUri,
+      signerOrAuthToken,
+      PROTOCOL_COMMANDS.PERSISTENT_STORAGE_DOWNLOAD_FILE,
+      signal
+    )
+    const query = this.buildQuery({ ...authPayload })
+    const headers: Record<string, string> = {}
+    if (typeof signerOrAuthToken === 'string') {
+      headers.Authorization = signerOrAuthToken
+    }
+    if (offset > 0) headers.Range = `bytes=${offset}-`
+    const response = await fetch(`${routeBase}?${query.toString()}`, {
+      method: 'GET',
+      headers,
+      signal
+    })
+    if (!response.ok) throw new Error(await response.text())
+    if (offset > 0) {
+      // A server that ignores the Range header answers 200 with the full body;
+      // consuming it as if it started at `offset` would corrupt a resumed download,
+      // so require a 206 whose Content-Range begins exactly at the requested offset.
+      if (response.status !== 206) {
+        throw new Error(
+          `Persistent storage range request was not honored: expected 206 Partial Content, got ${response.status}`
+        )
+      }
+      const contentRange = response.headers.get('content-range')
+      const match = contentRange?.match(/bytes\s+(\d+)-/i)
+      if (!match || Number(match[1]) !== offset) {
+        throw new Error(
+          `Persistent storage range request returned an unexpected Content-Range: ${
+            contentRange ?? 'none'
+          } (requested offset ${offset})`
+        )
+      }
+    }
+    return responseBodyToAsyncIterable(response.body)
+  }
+
   public async uploadPersistentStorageFile(
     nodeUri: string,
     signerOrAuthToken: SignerOrAuthTokenOrSignature,

src/services/providers/P2pProvider.ts

@@ -3567,6 +3567,119 @@ export class P2pProvider {
     )
   }
 
+  /**
+   * Downloads a file stored in a persistent-storage bucket over libp2p.
+   *
+   * The transfer mirrors `getComputeResult`: `dialAndStream` takes a concurrency
+   * slot up front, the first frame is a status JSON (an `httpStatus >= 400` throws
+   * and releases the slot immediately), and the returned generator streams the file
+   * body, applying flow control so a slow consumer does not overrun the buffer. If
+   * the caller stops reading early, or `signal` aborts, or the idle timeout fires,
+   * the stream is reset so the peer stops sending, and the concurrency slot is
+   * released from the generator's `finally`.
+   * @param {OceanNode} nodeUri The provider node (peerId / multiaddr).
+   * @param {SignerOrAuthTokenOrSignature} signerOrAuthToken Signer, JWT auth token, or precomputed signature used to authenticate the request.
+   * @param {string} bucketId The bucket holding the file.
+   * @param {string} fileName The name of the file to download.
+   * @param {number} [offset=0] Byte offset to resume the download from, sent to the node in the request payload. Must be a non-negative safe integer.
+   * @param {AbortSignal} [signal] Abort signal that cancels the download mid-flight and tears down the stream.
+   * @return {Promise<ComputeResultStream>} An async-iterable stream of the file body starting at `offset`.
+   */
+  public async downloadPersistentStorageFile(
+    nodeUri: OceanNode,
+    signerOrAuthToken: SignerOrAuthTokenOrSignature,
+    bucketId: string,
+    fileName: string,
+    offset: number = 0,
+    signal?: AbortSignal
+  ): Promise<ComputeResultStream> {
+    if (!Number.isSafeInteger(offset) || offset < 0) {
+      throw new Error(`Invalid offset: ${offset}. Must be a non-negative safe integer.`)
+    }
+    const { consumerAddress, nonce, signature } = await this.getSignedCommandParams(
+      nodeUri,
+      signerOrAuthToken,
+      PROTOCOL_COMMANDS.PERSISTENT_STORAGE_DOWNLOAD_FILE,
+      signal
+    )
+    const payload: Record<string, any> = {
+      command: PROTOCOL_COMMANDS.PERSISTENT_STORAGE_DOWNLOAD_FILE,
+      bucketId,
+      fileName,
+      offset,
+      consumerAddress
+    }
+
+    if (typeof signerOrAuthToken === 'string') {
+      payload.authorization = signerOrAuthToken
+    } else {
+      payload.nonce = nonce
+      payload.signature = signature
+    }
+
+    // A stored file is a bulk transfer like a compute result, so this mirrors
+    // `getComputeResult`: `dialAndStream` takes a concurrency slot and the generator
+    // below releases it from its `finally` rather than this method releasing on return.
+    const { firstBytes, frames, stream, release } = await this.dialAndStream(
+      nodeUri,
+      payload,
+      signal
+    )
+
+    let status: Record<string, any>
+    try {
+      // First frame is always a status JSON
+      status = JSON.parse(new TextDecoder().decode(firstBytes))
+      if (typeof status?.httpStatus === 'number' && status.httpStatus >= 400) {
+        throw new Error(
+          status.error ?? `P2P persistent storage download error: ${status.httpStatus}`
+        )
+      }
+    } catch (e) {
+      // Nothing is going to consume the generator, so hand the slot back here.
+      release()
+      throw e
+    }
+
+    const idleTimeout = this.streamIdleTimeoutMs()
+    return (async function* () {
+      let completed = false
+      try {
+        while (true) {
+          // Flow control — see `LP_RESUME_BELOW_BYTES`. A file download is exactly the
+          // transfer where a consumer writing to disk falls behind the sender, and an unread
+          // backlog past `maxBufferSize` is silently dropped by `byteStream`, desynchronising
+          // the frame parser and handing out corrupt, out-of-sequence frames.
+          if (frames.pendingBytes <= LP_RESUME_BELOW_BYTES) {
+            resumeReads(stream)
+          }
+          // Every frame is read under the caller's signal as well as the idle timeout,
+          // so a download in progress can be cancelled mid-flight.
+          const chunk = await readFrame(frames, signal, idleTimeout)
+          pauseReads(stream)
+          yield chunk
+        }
+      } catch (e) {
+        // Truncation and a clean end throw the same error type; only the clean end
+        // may finish the stream, or the consumer writes a short file.
+        if (!frames.isCleanEnd(e)) throw e
+        completed = true
+      } finally {
+        // Never leave the read side paused once nobody is reading any more.
+        resumeReads(stream)
+        // Cancelled, broken, or left early by the consumer: reset the stream so the peer
+        // stops producing a file body nobody will read.
+        if (!completed) {
+          abortResponseStream(
+            stream,
+            new Error('P2P persistent storage download is no longer being read')
+          )
+        }
+        release()
+      }
+    })()
+  }
+
   public async uploadPersistentStorageFile(
     nodeUri: OceanNode,
     signerOrAuthToken: SignerOrAuthTokenOrSignature,

test/integration/Provider.test.ts

@@ -126,6 +126,7 @@ describe('Provider persistent storage tests', function () {
   let nodeUri: string
   let bucketId: string
   const fileName = `oceanjs-persistent-storage-${Date.now()}.txt`
+  const fileContent = `persistent-storage-content-${Date.now()}`
 
   before(async () => {
     ownerSigner = (await provider.getSigner(0)) as Signer
@@ -158,9 +159,7 @@ describe('Provider persistent storage tests', function () {
       bucketId,
       fileName,
       (async function* () {
-        yield isP2pUri(nodeUri)
-          ? new TextEncoder().encode(`persistent-storage-${Date.now()}`)
-          : `persistent-storage-${Date.now()}`
+        yield isP2pUri(nodeUri) ? new TextEncoder().encode(fileContent) : fileContent
       })()
     )
 
@@ -188,6 +187,21 @@ describe('Provider persistent storage tests', function () {
     assert(fileObject?.fileName === fileName, 'File object has wrong file name')
   })
 
+  it('downloads the uploaded file bytes', async () => {
+    const stream = await ProviderInstance.downloadPersistentStorageFile(
+      nodeUri,
+      ownerSigner,
+      bucketId,
+      fileName
+    )
+    const chunks: Uint8Array[] = []
+    for await (const chunk of stream) {
+      chunks.push(chunk)
+    }
+    const downloaded = new TextDecoder().decode(Buffer.concat(chunks))
+    assert(downloaded === fileContent, 'Downloaded bytes do not match uploaded content')
+  })
+
   it('creates a bucket with a label and renames it', async () => {
     const created = await ProviderInstance.createPersistentStorageBucket(
       nodeUri,