diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5268ed622..4d388f4a1 100644 --- a/.github/workflows/ci.yml +++ b/.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 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..41934dfd6 100644 --- a/src/services/providers/BaseProvider.ts +++ b/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} 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 { + 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 b8481bca6..ad3fe844d 100644 --- a/src/services/providers/HttpProvider.ts +++ b/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=-` 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, + bucketId: string, + fileName: string, + 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( + 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()) + 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, diff --git a/src/services/providers/P2pProvider.ts b/src/services/providers/P2pProvider.ts index 7628136c4..b2a250a22 100644 --- a/src/services/providers/P2pProvider.ts +++ b/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} 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 { + 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 = { + 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,