From 5feac7b72f2d259fdf8e062b724562cc6549980c Mon Sep 17 00:00:00 2001 From: alexcos20 Date: Wed, 2 Sep 2026 08:33:40 +0300 Subject: [PATCH 1/4] add ps download file --- .github/workflows/ci.yml | 4 ++ src/@types/Provider.ts | 1 + src/services/providers/BaseProvider.ts | 18 +++++ src/services/providers/HttpProvider.ts | 34 ++++++++++ src/services/providers/P2pProvider.ts | 92 ++++++++++++++++++++++++++ test/integration/Provider.test.ts | 20 +++++- 6 files changed, 166 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b76cc071c..958693c57 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -61,6 +61,8 @@ jobs: working-directory: ${{ github.workspace }}/barge run: | bash -x start_ocean.sh 2>&1 > start_ocean.log & + env: + NODE_VERSION: pr-1466 - name: Install deps & build run: npm ci && npm run build:metadata - name: Wait for node to be ready @@ -150,6 +152,8 @@ jobs: working-directory: ${{ github.workspace }}/barge run: | bash -x start_ocean.sh 2>&1 > start-node.log & + env: + NODE_VERSION: pr-1466 - name: Install deps & build run: npm ci && npm run build:metadata diff --git a/src/@types/Provider.ts b/src/@types/Provider.ts index 52d4dee57..f96a3d80b 100644 --- a/src/@types/Provider.ts +++ b/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', diff --git a/src/services/providers/BaseProvider.ts b/src/services/providers/BaseProvider.ts index e700c9253..5382a16a4 100644 --- a/src/services/providers/BaseProvider.ts +++ b/src/services/providers/BaseProvider.ts @@ -1123,6 +1123,24 @@ export class BaseProvider { ) } + public async downloadPersistentStorageFile( + nodeUri: OceanNode, + signerOrAuthToken: SignerOrAuthTokenOrSignature, + bucketId: string, + fileName: string, + offset: number = 0, + signal?: AbortSignal + ): Promise { + return this.getImpl(nodeUri).downloadPersistentStorageFile( + nodeUri, + signerOrAuthToken, + bucketId, + fileName, + offset, + signal + ) + } + // ── Service on Demand ──────────────────────────────────────────────── public async getServiceTemplates( diff --git a/src/services/providers/HttpProvider.ts b/src/services/providers/HttpProvider.ts index e9ecec3e6..130d05dba 100644 --- a/src/services/providers/HttpProvider.ts +++ b/src/services/providers/HttpProvider.ts @@ -1592,6 +1592,40 @@ export class HttpProvider { return response.json() } + public async downloadPersistentStorageFile( + nodeUri: string, + signerOrAuthToken: SignerOrAuthTokenOrSignature, + bucketId: string, + fileName: string, + offset: number = 0, + signal?: AbortSignal + ): Promise { + 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 = {} + 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()) + return responseBodyToAsyncIterable(response.body) + } + public async uploadPersistentStorageFile( nodeUri: string, signerOrAuthToken: SignerOrAuthTokenOrSignature, diff --git a/src/services/providers/P2pProvider.ts b/src/services/providers/P2pProvider.ts index 829859515..ea559e28c 100644 --- a/src/services/providers/P2pProvider.ts +++ b/src/services/providers/P2pProvider.ts @@ -3567,6 +3567,98 @@ export class P2pProvider { ) } + public async downloadPersistentStorageFile( + nodeUri: OceanNode, + signerOrAuthToken: SignerOrAuthTokenOrSignature, + bucketId: string, + fileName: string, + offset: number = 0, + signal?: AbortSignal + ): Promise { + const { consumerAddress, nonce, signature } = await this.getSignedCommandParams( + nodeUri, + signerOrAuthToken, + PROTOCOL_COMMANDS.PERSISTENT_STORAGE_DOWNLOAD_FILE, + signal + ) + const payload: Record = { + 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 + 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, diff --git a/test/integration/Provider.test.ts b/test/integration/Provider.test.ts index 511a40fef..4b9648eb6 100644 --- a/test/integration/Provider.test.ts +++ b/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, From 36150c37b5f910a7e2e1764cdd78a37708b35c4b Mon Sep 17 00:00:00 2001 From: alexcos20 Date: Wed, 2 Sep 2026 08:51:41 +0300 Subject: [PATCH 2/4] fixes --- src/services/providers/BaseProvider.ts | 11 +++++++++ src/services/providers/HttpProvider.ts | 34 ++++++++++++++++++++++++++ src/services/providers/P2pProvider.ts | 23 +++++++++++++++++ 3 files changed, 68 insertions(+) diff --git a/src/services/providers/BaseProvider.ts b/src/services/providers/BaseProvider.ts index 5382a16a4..41934dfd6 100644 --- a/src/services/providers/BaseProvider.ts +++ b/src/services/providers/BaseProvider.ts @@ -1123,6 +1123,17 @@ 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} An async-iterable stream of the file body starting at `offset`. + */ public async downloadPersistentStorageFile( nodeUri: OceanNode, signerOrAuthToken: SignerOrAuthTokenOrSignature, diff --git a/src/services/providers/HttpProvider.ts b/src/services/providers/HttpProvider.ts index 130d05dba..61aadd669 100644 --- a/src/services/providers/HttpProvider.ts +++ b/src/services/providers/HttpProvider.ts @@ -1592,6 +1592,16 @@ 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=-` 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} An async-iterable stream of the file body starting at `offset`. + */ public async downloadPersistentStorageFile( nodeUri: string, signerOrAuthToken: SignerOrAuthTokenOrSignature, @@ -1600,6 +1610,11 @@ export class HttpProvider { offset: number = 0, signal?: AbortSignal ): Promise { + 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( @@ -1623,6 +1638,25 @@ export class HttpProvider { 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) } diff --git a/src/services/providers/P2pProvider.ts b/src/services/providers/P2pProvider.ts index ea559e28c..b0eb5b50e 100644 --- a/src/services/providers/P2pProvider.ts +++ b/src/services/providers/P2pProvider.ts @@ -3567,6 +3567,24 @@ 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} An async-iterable stream of the file body starting at `offset`. + */ public async downloadPersistentStorageFile( nodeUri: OceanNode, signerOrAuthToken: SignerOrAuthTokenOrSignature, @@ -3575,6 +3593,11 @@ export class P2pProvider { offset: number = 0, signal?: AbortSignal ): Promise { + 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, From ea9ed2341fcf0cb442793e75e0943050a2a1240c Mon Sep 17 00:00:00 2001 From: alexcos20 Date: Wed, 2 Sep 2026 08:53:35 +0300 Subject: [PATCH 3/4] fix lint --- src/services/providers/HttpProvider.ts | 4 +--- src/services/providers/P2pProvider.ts | 4 +--- 2 files changed, 2 insertions(+), 6 deletions(-) diff --git a/src/services/providers/HttpProvider.ts b/src/services/providers/HttpProvider.ts index 61aadd669..b0821a888 100644 --- a/src/services/providers/HttpProvider.ts +++ b/src/services/providers/HttpProvider.ts @@ -1611,9 +1611,7 @@ export class HttpProvider { signal?: AbortSignal ): Promise { if (!Number.isSafeInteger(offset) || offset < 0) { - throw new Error( - `Invalid offset: ${offset}. Must be a non-negative safe integer.` - ) + throw new Error(`Invalid offset: ${offset}. Must be a non-negative safe integer.`) } const routeBase = this.baseUrl(nodeUri) + diff --git a/src/services/providers/P2pProvider.ts b/src/services/providers/P2pProvider.ts index b0eb5b50e..ad7b37594 100644 --- a/src/services/providers/P2pProvider.ts +++ b/src/services/providers/P2pProvider.ts @@ -3594,9 +3594,7 @@ export class P2pProvider { signal?: AbortSignal ): Promise { if (!Number.isSafeInteger(offset) || offset < 0) { - throw new Error( - `Invalid offset: ${offset}. Must be a non-negative safe integer.` - ) + throw new Error(`Invalid offset: ${offset}. Must be a non-negative safe integer.`) } const { consumerAddress, nonce, signature } = await this.getSignedCommandParams( nodeUri, From ac0c11d282cc519f06170aad7f5de427ae6d9f93 Mon Sep 17 00:00:00 2001 From: Alex Coseru Date: Wed, 2 Sep 2026 12:04:52 +0300 Subject: [PATCH 4/4] Change NODE_VERSION from pr-1466 to main --- .github/workflows/ci.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 958693c57..4d388f4a1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -62,7 +62,7 @@ jobs: run: | bash -x start_ocean.sh 2>&1 > start_ocean.log & env: - NODE_VERSION: pr-1466 + NODE_VERSION: main - name: Install deps & build run: npm ci && npm run build:metadata - name: Wait for node to be ready @@ -153,7 +153,7 @@ jobs: run: | bash -x start_ocean.sh 2>&1 > start-node.log & env: - NODE_VERSION: pr-1466 + NODE_VERSION: main - name: Install deps & build run: npm ci && npm run build:metadata