From 83e8215013ef4265a48fdcdeb43ba6916dcae95f Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Tue, 7 Jul 2026 16:17:50 -0300 Subject: [PATCH 01/10] feat: add partial (batched) deployment support add a new `deployPartial` method that uploads an entity's content across several POST /entities requests instead of one, for entities too large for a single request. the regular `deploy` is unchanged. - splits content into size-bounded batches (first-fit-decreasing, default 50 MiB per request) and uploads them with `partial=true` - the entity file travels only in the first request (the server needs the manifest before the parallel batches); remaining batches upload through a bounded worker pool - skips content already on the server via /available-content, and resumes on network/5xx failures by re-querying and re-uploading only what is missing - returns the creationTimestamp once the server finalizes; surfaces PartialDeploymentValidationError on a 400 and PartialDeploymentNotSupportedError against a server without partial support --- README.md | 30 +++++ src/client/ContentClient.ts | 216 +++++++++++++++++++++++++++++++-- src/client/index.ts | 1 + src/client/types.ts | 28 +++++ src/client/utils/batching.ts | 41 +++++++ src/client/utils/errors.ts | 30 +++++ src/index.ts | 12 +- test/PartialDeployment.spec.ts | 181 +++++++++++++++++++++++++++ test/unit/batching.spec.ts | 70 +++++++++++ 9 files changed, 600 insertions(+), 9 deletions(-) create mode 100644 src/client/utils/batching.ts create mode 100644 src/client/utils/errors.ts create mode 100644 test/PartialDeployment.spec.ts create mode 100644 test/unit/batching.spec.ts diff --git a/README.md b/README.md index cf33a90..24b6a47 100644 --- a/README.md +++ b/README.md @@ -15,3 +15,33 @@ npm install dcl-catalyst-client ### Examples Please check the [examples repository](https://github.com/decentraland/catalyst-client-examples) + +### Partial (batched) deployments + +For entities too large to upload in a single request, use `deployPartial`. It splits the content into +size-bounded batches and uploads them across several `POST /entities` requests; the entity only becomes +live once the server has all of it (the server finalizes automatically on the completing request). + +```ts +const result = await client.deployPartial( + { entityId, authChain, files }, + { + maxBatchSizeBytes: 50 * 1024 * 1024, // default: 50 MiB per request + concurrency: 2, // default: 2 parallel batch uploads (after the first) + maxResumeAttempts: 3, // default: 3 — resumes on network/5xx failures + onProgress: ({ uploadedBytes, totalBytes }) => console.log(`${uploadedBytes}/${totalBytes}`) + } +) +console.log(result.creationTimestamp) +``` + +Notes: + +- The regular `deploy()` is unchanged; use it for normal-sized deployments. +- The target server must support partial deployments. Against an older server that does not, a + single-batch deployment still succeeds, but a multi-batch one rejects with + `PartialDeploymentNotSupportedError` — fall back to `deploy()` in that case. +- A validation failure on the finalizing request (e.g. exceeding the size budget) rejects with + `PartialDeploymentValidationError`. +- Per-file server caps still apply; a single file larger than a request cap is sent alone but may be + rejected by the server. diff --git a/src/client/ContentClient.ts b/src/client/ContentClient.ts index edbe899..c7b4ffb 100644 --- a/src/client/ContentClient.ts +++ b/src/client/ContentClient.ts @@ -1,9 +1,24 @@ import { hashV0, hashV1 } from '@dcl/hashing' import { Entity } from '@dcl/schemas' -import { ClientOptions, DeploymentData, IFetchComponent, ParallelConfig, RequestOptions } from './types' +import { + ClientOptions, + DeploymentData, + FetchResponse, + IFetchComponent, + ParallelConfig, + PartialDeploymentOptions, + PartialDeploymentResult, + RequestOptions +} from './types' import { addModelToFormData, mergeRequestOptions, sanitizeUrl, splitAndFetch } from './utils/Helper' +import { DEFAULT_MAX_BATCH_SIZE_BYTES, splitIntoBatches } from './utils/batching' +import { DeploymentError, PartialDeploymentNotSupportedError, PartialDeploymentValidationError } from './utils/errors' import { retry } from './utils/retry' +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)) +} + export type AvailableContentResult = { cid: string available: boolean @@ -25,6 +40,18 @@ export type ContentClient = { */ deploy(deployData: DeploymentData, options?: RequestOptions): Promise + /** + * Deploys an entity across multiple requests (partial deployment): content files are split into + * size-bounded batches and uploaded in several `POST /entities` requests, and the entity only becomes + * live once the server has all of it. Useful for scenes too large for a single request. The server + * must support the partial-deployment protocol; against one that doesn't, if everything fits in a + * single batch this still succeeds, otherwise it rejects with {@link PartialDeploymentNotSupportedError}. + */ + deployPartial( + deployData: DeploymentData, + options?: RequestOptions & PartialDeploymentOptions + ): Promise + /** * Checks if a pointer is consistent across multiple content servers */ @@ -165,20 +192,23 @@ export function createContentClient(options: ClientOptions): ContentClient { } } - async function buildEntityFormDataForDeployment( - deployData: DeploymentData, - options?: RequestOptions - ): Promise { + // Builds the multipart body shared by full and partial deployments. Appends the fixed fields + // (entityId, the `partial` flag when set, and the auth chain) BEFORE any file part, so streaming + // multipart parsers see them early, then appends one file part per hash in `fileHashes`. + function buildDeploymentForm(deployData: DeploymentData, fileHashes: Iterable, partial: boolean): FormData { // Use the web-standard FormData/Blob (global in browsers and Node >= 18). When this form is // used as a request body, native fetch (and node-fetch v3) set the multipart Content-Type with // the boundary automatically, so the deployment works without a node-fetch-specific fetcher. const form = new FormData() form.append('entityId', deployData.entityId) + if (partial) { + form.append('partial', 'true') + } addModelToFormData(deployData.authChain, form, 'authChain') - const alreadyUploadedHashes = await hashesAlreadyOnServer(Array.from(deployData.files.keys()), options) - for (const [fileHash, file] of deployData.files) { - if (!alreadyUploadedHashes.has(fileHash) || fileHash === deployData.entityId) { + for (const fileHash of fileHashes) { + const file = deployData.files.get(fileHash) + if (file) { form.append(fileHash, new Blob([file]), fileHash) } } @@ -186,6 +216,17 @@ export function createContentClient(options: ClientOptions): ContentClient { return form } + async function buildEntityFormDataForDeployment( + deployData: DeploymentData, + options?: RequestOptions + ): Promise { + const alreadyUploadedHashes = await hashesAlreadyOnServer(Array.from(deployData.files.keys()), options) + const fileHashes = Array.from(deployData.files.keys()).filter( + (fileHash) => !alreadyUploadedHashes.has(fileHash) || fileHash === deployData.entityId + ) + return buildDeploymentForm(deployData, fileHashes, false) + } + async function deploy(deployData: DeploymentData, options?: RequestOptions): Promise { const form = await buildEntityFormDataForDeployment(deployData, options) @@ -197,6 +238,164 @@ export function createContentClient(options: ClientOptions): ContentClient { return await fetcher.fetch(`${contentUrl}/entities`, requestOptions) } + async function deployPartial( + deployData: DeploymentData, + options?: RequestOptions & PartialDeploymentOptions + ): Promise { + const maxBatchSizeBytes = options?.maxBatchSizeBytes ?? DEFAULT_MAX_BATCH_SIZE_BYTES + const concurrency = Math.max(1, options?.concurrency ?? 2) + const maxResumeAttempts = options?.maxResumeAttempts ?? 3 + const resumeDelay = options?.resumeDelay ?? 1000 + const onProgress = options?.onProgress + + const entityId = deployData.entityId + const allContentHashes = Array.from(deployData.files.keys()).filter((hash) => hash !== entityId) + const sizeOf = (hash: string) => deployData.files.get(hash)?.byteLength ?? 0 + const totalBytes = allContentHashes.reduce((acc, hash) => acc + sizeOf(hash), 0) + + // Request-scoped options without the partial-specific keys, so they aren't forwarded to fetch. + const requestBase: RequestOptions = { + headers: options?.headers, + timeout: options?.timeout, + attempts: options?.attempts, + retryDelay: options?.retryDelay, + signal: options?.signal, + abortController: options?.abortController + } + + const classify4xx = (status: number, body: string): DeploymentError => { + if (/neither present in the storage/i.test(body)) { + return new PartialDeploymentNotSupportedError( + `The server does not support partial deployments (it validated a staging request as a full deployment). ` + + `Use deploy() for this server. Server response: ${body}`, + status, + body + ) + } + return new PartialDeploymentValidationError(`The partial deployment was rejected: ${body}`, status, body) + } + + type BatchOutcome = { kind: 'deployed'; result: PartialDeploymentResult } | { kind: 'incomplete' } + + // Sends one request. Throws a DeploymentError for terminal (4xx) failures and a plain Error for + // retryable ones (5xx / network), which the resume loop distinguishes by `instanceof DeploymentError`. + const sendRequest = async (fileHashes: string[], signal?: AbortSignal): Promise => { + const form = buildDeploymentForm(deployData, fileHashes, true) + const requestOptions = mergeRequestOptions(requestBase, { body: form as any, method: 'POST', signal }) + const response: FetchResponse = await fetcher.fetch(`${contentUrl}/entities`, requestOptions) + if (response.status === 200) { + return { kind: 'deployed', result: (await response.json()) as PartialDeploymentResult } + } + if (response.status === 202) { + return { kind: 'incomplete' } + } + const body = await response.text().catch(() => '') + if (response.status >= 400 && response.status < 500) { + throw classify4xx(response.status, body) + } + throw new Error(`Server responded with status ${response.status}: ${body}`) + } + + const runSession = async (): Promise => { + const alreadyOnServer = + allContentHashes.length > 0 ? await hashesAlreadyOnServer(allContentHashes, requestBase) : new Set() + const missingFiles = new Map() + for (const hash of allContentHashes) { + if (!alreadyOnServer.has(hash)) { + missingFiles.set(hash, deployData.files.get(hash)!) + } + } + const batches = splitIntoBatches(missingFiles, maxBatchSizeBytes) + const totalBatches = Math.max(1, batches.length) + + let uploadedBytes = totalBytes - Array.from(missingFiles.values()).reduce((acc, f) => acc + f.byteLength, 0) + let completedBatches = 0 + const reportProgress = () => onProgress?.({ uploadedBytes, totalBytes, completedBatches, totalBatches }) + + // First request carries the entity file (the server needs the manifest before parallel batches). + const first = await sendRequest([entityId, ...(batches[0]?.hashes ?? [])]) + if (first.kind === 'deployed') { + return first.result + } + if (batches[0]) { + uploadedBytes += batches[0].sizeBytes + completedBatches++ + reportProgress() + } + + // Remaining batches through a bounded worker pool. First 200 wins; a terminal error aborts the rest. + const remaining = batches.slice(1) + const controller = new AbortController() + let deployed: PartialDeploymentResult | undefined + let terminalError: DeploymentError | undefined + let nextIndex = 0 + + const worker = async (): Promise => { + while (!deployed && !terminalError) { + const index = nextIndex++ + if (index >= remaining.length) { + return + } + const batch = remaining[index] + try { + const outcome = await sendRequest(batch.hashes, controller.signal) + if (outcome.kind === 'deployed') { + deployed = outcome.result + controller.abort() + return + } + uploadedBytes += batch.sizeBytes + completedBatches++ + reportProgress() + } catch (error) { + if (error instanceof DeploymentError) { + terminalError = error + controller.abort() + return + } + throw error + } + } + } + + await Promise.all(Array.from({ length: Math.min(concurrency, Math.max(1, remaining.length)) }, () => worker())) + + if (terminalError) { + throw terminalError + } + if (deployed) { + return deployed + } + + // Every batch returned 202 but the server never finalized: the pending upload's state must have + // changed under us (expired or replaced by an overlapping deployment). Signal a resume. + throw new Error('The partial deployment did not finalize; the pending upload may have expired or been replaced.') + } + + let lastError: unknown + for (let attempt = 0; attempt <= maxResumeAttempts; attempt++) { + try { + return await runSession() + } catch (error) { + // 4xx validation / not-supported errors are terminal; anything else (5xx, network, non-finalize) + // is retried by re-querying available content and re-uploading only what's still missing. + if (error instanceof DeploymentError) { + throw error + } + lastError = error + if (attempt < maxResumeAttempts) { + await delay(resumeDelay) + } + } + } + + throw new DeploymentError( + `The partial deployment failed after ${maxResumeAttempts + 1} attempt(s): ${ + lastError instanceof Error ? lastError.message : String(lastError) + }` + ) + } + async function fetchEntitiesByPointers(pointers: string[], options?: RequestOptions): Promise { if (pointers.length === 0) { return Promise.reject(`You must set at least one pointer.`) @@ -325,6 +524,7 @@ export function createContentClient(options: ClientOptions): ContentClient { return downloadContent(fetcher, contentUrl + '/contents', contentHash, options) }, deploy, + deployPartial, isContentAvailable, checkPointerConsistency } diff --git a/src/client/index.ts b/src/client/index.ts index e794a2d..57f63b2 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -1,4 +1,5 @@ export * from './utils/DeploymentBuilder' +export * from './utils/errors' export * from './CatalystClient' export * from './ContentClient' export * from './LambdasClient' diff --git a/src/client/types.ts b/src/client/types.ts index 19be561..2c21560 100644 --- a/src/client/types.ts +++ b/src/client/types.ts @@ -79,6 +79,34 @@ export type ParallelConfig = { urls: string[] } +export type PartialDeploymentProgress = { + /** Bytes of content confirmed on the server so far (includes content already present before the upload). */ + uploadedBytes: number + /** Total bytes of the entity's content files. */ + totalBytes: number + /** Number of batches confirmed staged so far. */ + completedBatches: number + /** Total number of batches to upload this session. */ + totalBatches: number +} + +export type PartialDeploymentOptions = { + /** Max summed file bytes per request. Default {@link DEFAULT_MAX_BATCH_SIZE_BYTES} (50 MiB). */ + maxBatchSizeBytes?: number + /** Concurrent batch uploads after the first (entity-carrying) request. Default 2. */ + concurrency?: number + /** Resume cycles (re-query available content + re-upload) on network/5xx failures. Default 3. */ + maxResumeAttempts?: number + /** Delay between resume cycles in ms. Default 1000. */ + resumeDelay?: number + /** Invoked after each staged batch with cumulative progress. */ + onProgress?: (progress: PartialDeploymentProgress) => void +} + +export type PartialDeploymentResult = { + creationTimestamp: number +} + export type ClientOptions = { url: string fetcher: IFetchComponent diff --git a/src/client/utils/batching.ts b/src/client/utils/batching.ts new file mode 100644 index 0000000..3c508dc --- /dev/null +++ b/src/client/utils/batching.ts @@ -0,0 +1,41 @@ +/** + * Default cap on the summed file bytes per partial-deployment request. Well under the servers' request + * caps (worlds-content-server 350MB, catalyst 2GiB) and under typical proxy body limits (~100MB), and + * small enough that a failed request only re-uploads at most this much. The cap counts file bytes, not + * exact wire size; the margin absorbs multipart overhead. + */ +export const DEFAULT_MAX_BATCH_SIZE_BYTES = 50 * 1024 * 1024 // 50 MiB + +export type FileBatch = { + hashes: string[] + sizeBytes: number +} + +/** + * Splits files into batches whose summed byte size stays at or below `maxBatchSizeBytes`, using + * first-fit-decreasing bin packing. A single file larger than the cap gets its own batch (the servers' + * per-file caps still apply and would reject it — that is the caller's/server's concern, not this + * function's). Deterministic given the same input. + */ +export function splitIntoBatches(files: Map, maxBatchSizeBytes: number): FileBatch[] { + const entries = Array.from(files.entries()).sort((a, b) => b[1].byteLength - a[1].byteLength) + const batches: FileBatch[] = [] + + for (const [hash, content] of entries) { + const size = content.byteLength + let placed = false + for (const batch of batches) { + if (batch.sizeBytes + size <= maxBatchSizeBytes) { + batch.hashes.push(hash) + batch.sizeBytes += size + placed = true + break + } + } + if (!placed) { + batches.push({ hashes: [hash], sizeBytes: size }) + } + } + + return batches +} diff --git a/src/client/utils/errors.ts b/src/client/utils/errors.ts new file mode 100644 index 0000000..2f8b6c5 --- /dev/null +++ b/src/client/utils/errors.ts @@ -0,0 +1,30 @@ +/** Base error for a failed deployment. Carries the HTTP status and raw response body when available. */ +export class DeploymentError extends Error { + constructor(message: string, public readonly status?: number, public readonly responseBody?: string) { + super(message) + this.name = 'DeploymentError' + } +} + +/** + * The server rejected the deployment with a validation error (HTTP 400) — e.g. the deployment exceeds + * the size budget, or a signature/permission check failed. Terminal: retrying will not help. + */ +export class PartialDeploymentValidationError extends DeploymentError { + constructor(message: string, status?: number, responseBody?: string) { + super(message, status, responseBody) + this.name = 'PartialDeploymentValidationError' + } +} + +/** + * The target server does not support partial deployments: it ran the full deployment pipeline on a + * staging request and rejected it because referenced content was not uploaded or stored. Fall back to + * `deploy()` (a single request) against such a server. Best-effort detection via the server's error text. + */ +export class PartialDeploymentNotSupportedError extends DeploymentError { + constructor(message: string, status?: number, responseBody?: string) { + super(message, status, responseBody) + this.name = 'PartialDeploymentNotSupportedError' + } +} diff --git a/src/index.ts b/src/index.ts index 89c461c..bd3a143 100644 --- a/src/index.ts +++ b/src/index.ts @@ -2,4 +2,14 @@ export * from './client/CatalystClient' export * from './client/ContentClient' export * from './client/LambdasClient' export * from './client/utils' -export type { FetchResponse, IFetchComponent, RequestOptions } from './client/types' +export * from './client/utils/errors' +export { DEFAULT_MAX_BATCH_SIZE_BYTES, splitIntoBatches } from './client/utils/batching' +export type { FileBatch } from './client/utils/batching' +export type { + FetchResponse, + IFetchComponent, + PartialDeploymentOptions, + PartialDeploymentProgress, + PartialDeploymentResult, + RequestOptions +} from './client/types' diff --git a/test/PartialDeployment.spec.ts b/test/PartialDeployment.spec.ts new file mode 100644 index 0000000..5bebae9 --- /dev/null +++ b/test/PartialDeployment.spec.ts @@ -0,0 +1,181 @@ +import { createContentClient, ContentClient, IFetchComponent, DeploymentData } from '../src' +import { PartialDeploymentNotSupportedError, PartialDeploymentValidationError } from '../src/client/utils/errors' + +const URL = 'https://content.example.com' + +type EntitiesResponse = { status: number; body?: any; text?: string; throwNetwork?: boolean } + +describe('deployPartial', () => { + const entityId = 'bafyEntity' + let deployData: DeploymentData + let entitiesResponses: EntitiesResponse[] + let available: Set + let entitiesCalls: FormData[] + let fetcher: IFetchComponent + let client: ContentClient + + function makeDeployData(fileSizes: Record): DeploymentData { + const files = new Map() + files.set(entityId, new Uint8Array([1, 2, 3])) + for (const [hash, size] of Object.entries(fileSizes)) { + files.set(hash, new Uint8Array(size)) + } + return { entityId, authChain: [], files } + } + + beforeEach(() => { + entitiesResponses = [] + available = new Set() + entitiesCalls = [] + + fetcher = { + fetch: jest.fn(async (url: string, init?: any) => { + if (url.includes('/available-content')) { + const cids = (url.split('?')[1] || '') + .split('&') + .filter((p) => p.startsWith('cid=')) + .map((p) => decodeURIComponent(p.slice('cid='.length))) + return { + ok: true, + status: 200, + json: async () => cids.map((cid) => ({ cid, available: available.has(cid) })), + text: async () => '', + arrayBuffer: async () => new ArrayBuffer(0) + } + } + // POST /entities + entitiesCalls.push(init.body as FormData) + const next = entitiesResponses.shift() + if (!next) { + throw new Error('Unexpected extra POST /entities call') + } + if (next.throwNetwork) { + throw new Error('network failure') + } + return { + ok: next.status >= 200 && next.status < 300, + status: next.status, + json: async () => next.body ?? {}, + text: async () => next.text ?? '', + arrayBuffer: async () => new ArrayBuffer(0) + } + }) + } + client = createContentClient({ url: URL, fetcher }) + }) + + function entityFileIncluded(form: FormData): boolean { + return form.getAll(entityId).some((v) => v instanceof Blob) + } + + describe('when all content fits in a single batch', () => { + it('should perform one POST that carries the entity file and resolve with the creationTimestamp', async () => { + deployData = makeDeployData({ hashA: 100 }) + entitiesResponses = [{ status: 200, body: { creationTimestamp: 42 } }] + + const result = await client.deployPartial(deployData) + + expect(result).toEqual({ creationTimestamp: 42 }) + expect(entitiesCalls).toHaveLength(1) + expect(entitiesCalls[0].get('partial')).toBe('true') + expect(entityFileIncluded(entitiesCalls[0])).toBe(true) + expect(entitiesCalls[0].has('hashA')).toBe(true) + }) + }) + + describe('when the server already stores some hashes', () => { + it('should omit those hashes from every request', async () => { + deployData = makeDeployData({ hashA: 100, hashB: 100 }) + available = new Set(['hashA']) + entitiesResponses = [{ status: 200, body: { creationTimestamp: 7 } }] + + await client.deployPartial(deployData) + + expect(entitiesCalls).toHaveLength(1) + expect(entitiesCalls[0].has('hashA')).toBe(false) + expect(entitiesCalls[0].has('hashB')).toBe(true) + }) + }) + + describe('when content spans multiple batches', () => { + it('should include the entity file only in the first request and finalize on the last', async () => { + deployData = makeDeployData({ hashA: 80, hashB: 80 }) + entitiesResponses = [ + { status: 202, body: { missing: ['hashB'] } }, + { status: 200, body: { creationTimestamp: 99 } } + ] + + // Cap forces one file per request; concurrency 1 makes ordering deterministic. + const result = await client.deployPartial(deployData, { maxBatchSizeBytes: 100, concurrency: 1 }) + + expect(result).toEqual({ creationTimestamp: 99 }) + expect(entitiesCalls).toHaveLength(2) + expect(entityFileIncluded(entitiesCalls[0])).toBe(true) + expect(entityFileIncluded(entitiesCalls[1])).toBe(false) + }) + }) + + describe('when a batch fails with a network error', () => { + it('should resume by re-querying available content and re-uploading the missing hashes', async () => { + deployData = makeDeployData({ hashA: 80, hashB: 80 }) + entitiesResponses = [ + { status: 202, body: { missing: ['hashB'] } }, + { throwNetwork: true, status: 0 }, + // resume: hashA now on server, only hashB re-uploaded → finalize + { status: 200, body: { creationTimestamp: 5 } } + ] + + const result = await client.deployPartial(deployData, { + maxBatchSizeBytes: 100, + concurrency: 1, + resumeDelay: 0 + }) + + expect(result).toEqual({ creationTimestamp: 5 }) + }) + }) + + describe('when the completing request returns 400', () => { + it('should reject with a PartialDeploymentValidationError exposing the status and body', async () => { + deployData = makeDeployData({ hashA: 100 }) + entitiesResponses = [{ status: 400, text: 'The deployment is too big.' }] + + await expect(client.deployPartial(deployData)).rejects.toBeInstanceOf(PartialDeploymentValidationError) + }) + }) + + describe('when an old server rejects a staging request as a full deployment', () => { + it('should reject with a PartialDeploymentNotSupportedError', async () => { + deployData = makeDeployData({ hashA: 80, hashB: 80 }) + entitiesResponses = [ + { + status: 400, + text: 'This hash is referenced in the entity but was not uploaded or previously available: hashB (neither present in the storage)' + } + ] + + await expect(client.deployPartial(deployData, { maxBatchSizeBytes: 100, concurrency: 1 })).rejects.toBeInstanceOf( + PartialDeploymentNotSupportedError + ) + }) + }) + + describe('when an onProgress callback is provided', () => { + it('should report cumulative progress after each staged batch', async () => { + deployData = makeDeployData({ hashA: 80, hashB: 80 }) + entitiesResponses = [ + { status: 202, body: { missing: ['hashB'] } }, + { status: 200, body: { creationTimestamp: 1 } } + ] + const progress: number[] = [] + + await client.deployPartial(deployData, { + maxBatchSizeBytes: 100, + concurrency: 1, + onProgress: (p) => progress.push(p.completedBatches) + }) + + expect(progress[progress.length - 1]).toBeGreaterThanOrEqual(1) + }) + }) +}) diff --git a/test/unit/batching.spec.ts b/test/unit/batching.spec.ts new file mode 100644 index 0000000..0207f6e --- /dev/null +++ b/test/unit/batching.spec.ts @@ -0,0 +1,70 @@ +import { DEFAULT_MAX_BATCH_SIZE_BYTES, splitIntoBatches } from '../../src/client/utils/batching' + +describe('splitIntoBatches', () => { + function file(bytes: number): Uint8Array { + return new Uint8Array(bytes) + } + + describe('when all files fit under the cap', () => { + it('should return a single batch with every hash', () => { + const files = new Map([ + ['a', file(10)], + ['b', file(20)], + ['c', file(30)] + ]) + + const batches = splitIntoBatches(files, 1000) + + expect(batches).toHaveLength(1) + expect(new Set(batches[0].hashes)).toEqual(new Set(['a', 'b', 'c'])) + expect(batches[0].sizeBytes).toBe(60) + }) + }) + + describe('when the files exceed the cap', () => { + it('should split into batches each at or below the cap', () => { + const files = new Map([ + ['a', file(60)], + ['b', file(50)], + ['c', file(40)], + ['d', file(30)] + ]) + + const batches = splitIntoBatches(files, 100) + + for (const batch of batches) { + expect(batch.sizeBytes).toBeLessThanOrEqual(100) + } + const allHashes = batches.flatMap((b) => b.hashes) + expect(new Set(allHashes)).toEqual(new Set(['a', 'b', 'c', 'd'])) + expect(allHashes).toHaveLength(4) + }) + }) + + describe('when a single file is larger than the cap', () => { + it('should place it alone in its own batch', () => { + const files = new Map([ + ['big', file(500)], + ['small', file(10)] + ]) + + const batches = splitIntoBatches(files, 100) + + const bigBatch = batches.find((b) => b.hashes.includes('big')) + expect(bigBatch).toBeDefined() + expect(bigBatch!.hashes).toEqual(['big']) + }) + }) + + describe('when the files map is empty', () => { + it('should return no batches', () => { + expect(splitIntoBatches(new Map(), 100)).toEqual([]) + }) + }) + + describe('when using the default cap', () => { + it('should be 50 MiB', () => { + expect(DEFAULT_MAX_BATCH_SIZE_BYTES).toBe(50 * 1024 * 1024) + }) + }) +}) From 613435cea630889efa49971954d2c71c3d332669 Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Tue, 7 Jul 2026 16:35:42 -0300 Subject: [PATCH 02/10] fix: don't restart a partial deployment when a parallel batch is aborted when a batch's 200 finalizes the deployment the worker pool aborts the other in-flight batches; their abort errors were propagating and triggering a spurious resume cycle (an extra /available-content query, request and delay). swallow the abort once the session has already succeeded so the parallel path finalizes in a single session. adds a regression test for the parallel worker pool. --- src/client/ContentClient.ts | 6 +++ test/PartialDeployment.spec.ts | 86 ++++++++++++++++++++++++++++++++++ 2 files changed, 92 insertions(+) diff --git a/src/client/ContentClient.ts b/src/client/ContentClient.ts index c7b4ffb..ade8525 100644 --- a/src/client/ContentClient.ts +++ b/src/client/ContentClient.ts @@ -353,6 +353,12 @@ export function createContentClient(options: ClientOptions): ContentClient { controller.abort() return } + // A sibling worker already finalized (or the pool was aborted), so this request was + // cancelled on purpose — swallow its abort/network error instead of failing the session + // (which would trigger a spurious resume). Genuine failures still propagate. + if (deployed || controller.signal.aborted) { + return + } throw error } } diff --git a/test/PartialDeployment.spec.ts b/test/PartialDeployment.spec.ts index 5bebae9..4fc78eb 100644 --- a/test/PartialDeployment.spec.ts +++ b/test/PartialDeployment.spec.ts @@ -178,4 +178,90 @@ describe('deployPartial', () => { expect(progress[progress.length - 1]).toBeGreaterThanOrEqual(1) }) }) + + // Uses its own content-routed fetcher (rather than the ordered queue above) because parallel workers + // don't send requests in a deterministic order, and this fetcher must honor the abort signal. + describe('when batches run in parallel and one finalizes while another is in flight', () => { + let availableContentCalls: number + let parallelClient: ContentClient + let result: { creationTimestamp: number } + + beforeEach(async () => { + availableContentCalls = 0 + let finalized = false + // Three files, each its own batch (cap 250): the largest ships with the entity file in the first + // request; the other two run concurrently. `hashWin` finalizes (200); `hashSlow` never resolves + // on its own and rejects when the pool aborts it after the win. + const files = new Map() + files.set(entityId, new Uint8Array([1, 2, 3])) + files.set('hashBig', new Uint8Array(300)) + files.set('hashWin', new Uint8Array(240)) + files.set('hashSlow', new Uint8Array(230)) + const parallelDeployData: DeploymentData = { entityId, authChain: [], files } + + const ok200 = () => ({ + ok: true, + status: 200, + json: async () => ({ creationTimestamp: 1 }), + text: async () => '', + arrayBuffer: async () => new ArrayBuffer(0) + }) + + const parallelFetcher: IFetchComponent = { + fetch: jest.fn(async (url: string, init?: any) => { + if (url.includes('/available-content')) { + availableContentCalls++ + const cids = (url.split('?')[1] || '') + .split('&') + .filter((p) => p.startsWith('cid=')) + .map((p) => decodeURIComponent(p.slice('cid='.length))) + return { + ok: true, + status: 200, + json: async () => cids.map((cid) => ({ cid, available: finalized })), + text: async () => '', + arrayBuffer: async () => new ArrayBuffer(0) + } + } + const form = init.body as FormData + if (form.has('hashSlow')) { + return await new Promise((_resolve, reject) => { + const signal: AbortSignal | undefined = init.signal + if (signal?.aborted) return reject(new Error('AbortError')) + signal?.addEventListener('abort', () => reject(new Error('AbortError'))) + }) + } + if (form.has('hashWin')) { + finalized = true + return ok200() + } + if (form.has('hashBig')) { + return { + ok: true, + status: 202, + json: async () => ({ missing: ['hashWin', 'hashSlow'] }), + text: async () => '', + arrayBuffer: async () => new ArrayBuffer(0) + } + } + return ok200() + }) + } + + parallelClient = createContentClient({ url: URL, fetcher: parallelFetcher }) + result = await parallelClient.deployPartial(parallelDeployData, { + maxBatchSizeBytes: 250, + concurrency: 2, + resumeDelay: 0 + }) + }) + + it('should resolve with the creationTimestamp', () => { + expect(result).toEqual({ creationTimestamp: 1 }) + }) + + it('should not restart the session (the aborted sibling must not trigger a resume)', () => { + expect(availableContentCalls).toBe(1) + }) + }) }) From 94a341cb42a6a776521624bd5bb63f135302d1a6 Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Tue, 7 Jul 2026 20:04:20 -0300 Subject: [PATCH 03/10] fix: address deployPartial review findings - detect a server without partial support via either server's missing-content message: worlds-content-server's "neither present in the storage" and catalyst's (via @dcl/content-validator) "was not uploaded or previously available". previously only the worlds phrasing matched, so a multi-batch deploy against an old catalyst threw a terminal validation error instead of PartialDeploymentNotSupportedError and callers never fell back to deploy() - honor the caller's AbortSignal: it was overwritten by the internal pool controller (or undefined) via mergeRequestOptions, so cancellation was ignored. each request now aborts when either the caller or the pool aborts - treat 429 (and 5xx / network) as retryable so a rate-limited staging request resumes instead of failing terminally; only other 4xx are terminal adds tests for old-catalyst detection, 429 resume, and caller-signal propagation. --- src/client/ContentClient.ts | 55 +++++++++++++++++++++------ test/PartialDeployment.spec.ts | 68 +++++++++++++++++++++++++++++++++- 2 files changed, 109 insertions(+), 14 deletions(-) diff --git a/src/client/ContentClient.ts b/src/client/ContentClient.ts index ade8525..1d7d4b6 100644 --- a/src/client/ContentClient.ts +++ b/src/client/ContentClient.ts @@ -19,6 +19,26 @@ function delay(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)) } +// Combines several abort signals into one that aborts as soon as any of them does. Used so a request +// honors both the caller's cancellation signal and deployPartial's internal first-200-wins controller. +// (Hand-rolled rather than AbortSignal.any, which is only available on Node >= 20.3.) +function combineSignals(signals: Array): AbortSignal | undefined { + const present = signals.filter((signal): signal is AbortSignal => !!signal) + if (present.length <= 1) { + return present[0] + } + const controller = new AbortController() + const abort = () => controller.abort() + for (const signal of present) { + if (signal.aborted) { + controller.abort() + break + } + signal.addEventListener('abort', abort, { once: true }) + } + return controller.signal +} + export type AvailableContentResult = { cid: string available: boolean @@ -253,18 +273,24 @@ export function createContentClient(options: ClientOptions): ContentClient { const sizeOf = (hash: string) => deployData.files.get(hash)?.byteLength ?? 0 const totalBytes = allContentHashes.reduce((acc, hash) => acc + sizeOf(hash), 0) - // Request-scoped options without the partial-specific keys, so they aren't forwarded to fetch. + // Request-scoped options without the partial-specific keys or the abort signal (the signal is + // combined per-request with the internal pool controller, see sendRequest). const requestBase: RequestOptions = { headers: options?.headers, timeout: options?.timeout, attempts: options?.attempts, - retryDelay: options?.retryDelay, - signal: options?.signal, - abortController: options?.abortController + retryDelay: options?.retryDelay } - + // The caller's cancellation signal, honored on every request of every session. + const callerSignal = options?.signal ?? options?.abortController?.signal + + // A server that predates partial deployments runs the staging request through the full deploy + // pipeline and rejects the missing content. Match either server's phrasing: worlds-content-server + // says "neither present in the storage...", and catalyst (via @dcl/content-validator) says + // "referenced in the entity but was not uploaded or previously available". + const notSupportedPattern = /neither present in the storage|was not uploaded or previously available/i const classify4xx = (status: number, body: string): DeploymentError => { - if (/neither present in the storage/i.test(body)) { + if (notSupportedPattern.test(body)) { return new PartialDeploymentNotSupportedError( `The server does not support partial deployments (it validated a staging request as a full deployment). ` + `Use deploy() for this server. Server response: ${body}`, @@ -277,10 +303,12 @@ export function createContentClient(options: ClientOptions): ContentClient { type BatchOutcome = { kind: 'deployed'; result: PartialDeploymentResult } | { kind: 'incomplete' } - // Sends one request. Throws a DeploymentError for terminal (4xx) failures and a plain Error for - // retryable ones (5xx / network), which the resume loop distinguishes by `instanceof DeploymentError`. - const sendRequest = async (fileHashes: string[], signal?: AbortSignal): Promise => { + // Sends one request. Throws a terminal DeploymentError for a 4xx validation/not-supported failure, + // and a plain Error for retryable conditions (429 rate-limited, 5xx, network) which the resume loop + // re-attempts. `poolSignal` is deployPartial's internal first-200-wins controller signal. + const sendRequest = async (fileHashes: string[], poolSignal?: AbortSignal): Promise => { const form = buildDeploymentForm(deployData, fileHashes, true) + const signal = combineSignals([callerSignal, poolSignal]) const requestOptions = mergeRequestOptions(requestBase, { body: form as any, method: 'POST', signal }) const response: FetchResponse = await fetcher.fetch(`${contentUrl}/entities`, requestOptions) if (response.status === 200) { @@ -290,10 +318,13 @@ export function createContentClient(options: ClientOptions): ContentClient { return { kind: 'incomplete' } } const body = await response.text().catch(() => '') - if (response.status >= 400 && response.status < 500) { - throw classify4xx(response.status, body) + // 429 (rate limited / other transient conditions the server surfaces as 429) and 5xx are + // retryable — the resume loop re-queries available content and re-uploads only what's missing. + if (response.status === 429 || response.status >= 500) { + throw new Error(`Server responded with status ${response.status}: ${body}`) } - throw new Error(`Server responded with status ${response.status}: ${body}`) + // Any other non-2xx (4xx) is a terminal validation or not-supported error. + throw classify4xx(response.status, body) } const runSession = async (): Promise => { diff --git a/test/PartialDeployment.spec.ts b/test/PartialDeployment.spec.ts index 4fc78eb..57119b8 100644 --- a/test/PartialDeployment.spec.ts +++ b/test/PartialDeployment.spec.ts @@ -144,13 +144,13 @@ describe('deployPartial', () => { }) }) - describe('when an old server rejects a staging request as a full deployment', () => { + describe('when an old worlds-content-server rejects a staging request as a full deployment', () => { it('should reject with a PartialDeploymentNotSupportedError', async () => { deployData = makeDeployData({ hashA: 80, hashB: 80 }) entitiesResponses = [ { status: 400, - text: 'This hash is referenced in the entity but was not uploaded or previously available: hashB (neither present in the storage)' + text: 'Deployment failed: The file hashB (b.txt) is neither present in the storage or in the provided entity' } ] @@ -160,6 +160,70 @@ describe('deployPartial', () => { }) }) + describe('when an old catalyst rejects a staging request as a full deployment', () => { + it('should reject with a PartialDeploymentNotSupportedError', async () => { + deployData = makeDeployData({ hashA: 80, hashB: 80 }) + // Catalyst validates missing content via @dcl/content-validator, whose message differs from + // worlds-content-server's — the detection must recognize both. + entitiesResponses = [ + { + status: 400, + text: 'This hash is referenced in the entity but was not uploaded or previously available: hashB' + } + ] + + await expect(client.deployPartial(deployData, { maxBatchSizeBytes: 100, concurrency: 1 })).rejects.toBeInstanceOf( + PartialDeploymentNotSupportedError + ) + }) + }) + + describe('when the server responds 429 (rate limited)', () => { + it('should resume rather than fail terminally', async () => { + deployData = makeDeployData({ hashA: 100 }) + entitiesResponses = [ + { status: 429, text: 'Entity rate limited' }, + // resume: succeeds on the next session + { status: 200, body: { creationTimestamp: 3 } } + ] + + const result = await client.deployPartial(deployData, { resumeDelay: 0 }) + + expect(result).toEqual({ creationTimestamp: 3 }) + }) + }) + + describe('when the caller aborts via options.signal', () => { + let controller: AbortController + let sawAbortedSignal: boolean + + beforeEach(async () => { + controller = new AbortController() + sawAbortedSignal = false + deployData = makeDeployData({ hashA: 100 }) + // Abort before the request resolves; assert the request carried an already-aborted signal. + controller.abort() + entitiesResponses = [{ status: 200, body: { creationTimestamp: 1 } }] + const originalFetch = fetcher.fetch as jest.Mock + ;(fetcher.fetch as jest.Mock) = jest.fn(async (url: string, init?: any) => { + if (!url.includes('/available-content') && init?.signal?.aborted) { + sawAbortedSignal = true + } + return originalFetch(url, init) + }) + + try { + await client.deployPartial(deployData, { signal: controller.signal }) + } catch { + // May resolve or reject depending on timing; the assertion is about signal propagation. + } + }) + + it('should propagate the caller signal to the deployment request', () => { + expect(sawAbortedSignal).toBe(true) + }) + }) + describe('when an onProgress callback is provided', () => { it('should report cumulative progress after each staged batch', async () => { deployData = makeDeployData({ hashA: 80, hashB: 80 }) From adc282397958032c5a4440754e5d182811279145 Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Tue, 7 Jul 2026 21:24:44 -0300 Subject: [PATCH 04/10] feat: raise the default batch cap to 100 MiB and harden deployPartial MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit the infrastructure in front of the content servers times out requests larger than ~200MB, so the default batch cap moves from 50 MiB to 100 MiB — half the requests for large scenes while keeping a ~2x safety margin (verified: the first request is entity JSON + the fullest batch, every other request is one batch, and a batch can only exceed the cap when a single file does). hardening from the follow-up review: - fail fast when a single file exceeds the request cap: files cannot be split across requests, so such an upload previously burned every resume attempt re-uploading hundreds of MB before failing with an opaque network error - abort promptly: check the caller signal at session entry (covers aborts landing during the resume backoff), carry it on the available-content query, and combine it with the pool controller once per session instead of per request (avoids accumulating abort listeners on large uploads) - exponential backoff between resume attempts, so 429 rate-limit retries have a chance of landing outside the limit window --- README.md | 9 ++-- src/client/ContentClient.ts | 45 +++++++++++++++---- src/client/types.ts | 9 ++-- src/client/utils/batching.ts | 17 ++++---- test/PartialDeployment.spec.ts | 80 +++++++++++++++++++++++----------- test/unit/batching.spec.ts | 4 +- 6 files changed, 114 insertions(+), 50 deletions(-) diff --git a/README.md b/README.md index 24b6a47..1540944 100644 --- a/README.md +++ b/README.md @@ -26,7 +26,9 @@ live once the server has all of it (the server finalizes automatically on the co const result = await client.deployPartial( { entityId, authChain, files }, { - maxBatchSizeBytes: 50 * 1024 * 1024, // default: 50 MiB per request + // default: 100 MiB per request — requests larger than ~200MB are timed out by the + // infrastructure in front of the content servers, so keep a margin under that. + maxBatchSizeBytes: 100 * 1024 * 1024, concurrency: 2, // default: 2 parallel batch uploads (after the first) maxResumeAttempts: 3, // default: 3 — resumes on network/5xx failures onProgress: ({ uploadedBytes, totalBytes }) => console.log(`${uploadedBytes}/${totalBytes}`) @@ -43,5 +45,6 @@ Notes: `PartialDeploymentNotSupportedError` — fall back to `deploy()` in that case. - A validation failure on the finalizing request (e.g. exceeding the size budget) rejects with `PartialDeploymentValidationError`. -- Per-file server caps still apply; a single file larger than a request cap is sent alone but may be - rejected by the server. +- Files are never split across requests: a single file larger than the batch cap is sent alone in its + own request. Per-file server caps still apply to it, and a request beyond ~200MB may be timed out by + the infrastructure regardless of the configured cap. diff --git a/src/client/ContentClient.ts b/src/client/ContentClient.ts index 1d7d4b6..dc0ceb1 100644 --- a/src/client/ContentClient.ts +++ b/src/client/ContentClient.ts @@ -273,8 +273,20 @@ export function createContentClient(options: ClientOptions): ContentClient { const sizeOf = (hash: string) => deployData.files.get(hash)?.byteLength ?? 0 const totalBytes = allContentHashes.reduce((acc, hash) => acc + sizeOf(hash), 0) - // Request-scoped options without the partial-specific keys or the abort signal (the signal is - // combined per-request with the internal pool controller, see sendRequest). + // Files are never split across requests, so a single file above the request cap would produce a + // request the infrastructure may time out — and the resume loop would re-upload it on every + // attempt. Fail fast with a clear message instead of uploading hundreds of MB that cannot succeed. + const oversized = allContentHashes.find((hash) => sizeOf(hash) > maxBatchSizeBytes) + if (oversized) { + throw new PartialDeploymentValidationError( + `The file '${oversized}' (${sizeOf(oversized)} bytes) exceeds the maximum request size of ` + + `${maxBatchSizeBytes} bytes and files cannot be split across requests. If your infrastructure ` + + `allows larger requests, raise options.maxBatchSizeBytes.` + ) + } + + // Request-scoped options without the partial-specific keys or the abort signal (the caller signal + // and the pool controller are combined once per session in runSession and threaded from there). const requestBase: RequestOptions = { headers: options?.headers, timeout: options?.timeout, @@ -305,10 +317,11 @@ export function createContentClient(options: ClientOptions): ContentClient { // Sends one request. Throws a terminal DeploymentError for a 4xx validation/not-supported failure, // and a plain Error for retryable conditions (429 rate-limited, 5xx, network) which the resume loop - // re-attempts. `poolSignal` is deployPartial's internal first-200-wins controller signal. - const sendRequest = async (fileHashes: string[], poolSignal?: AbortSignal): Promise => { + // re-attempts. `signal` is already combined by the caller (caller cancellation + pool abort) — it + // is combined once per session rather than per request so a large upload doesn't accumulate one + // abort listener on the caller's signal per batch. + const sendRequest = async (fileHashes: string[], signal?: AbortSignal): Promise => { const form = buildDeploymentForm(deployData, fileHashes, true) - const signal = combineSignals([callerSignal, poolSignal]) const requestOptions = mergeRequestOptions(requestBase, { body: form as any, method: 'POST', signal }) const response: FetchResponse = await fetcher.fetch(`${contentUrl}/entities`, requestOptions) if (response.status === 200) { @@ -328,8 +341,13 @@ export function createContentClient(options: ClientOptions): ContentClient { } const runSession = async (): Promise => { + if (callerSignal?.aborted) { + throw new Error('The partial deployment was aborted by the caller.') + } const alreadyOnServer = - allContentHashes.length > 0 ? await hashesAlreadyOnServer(allContentHashes, requestBase) : new Set() + allContentHashes.length > 0 + ? await hashesAlreadyOnServer(allContentHashes, { ...requestBase, signal: callerSignal }) + : new Set() const missingFiles = new Map() for (const hash of allContentHashes) { if (!alreadyOnServer.has(hash)) { @@ -344,7 +362,7 @@ export function createContentClient(options: ClientOptions): ContentClient { const reportProgress = () => onProgress?.({ uploadedBytes, totalBytes, completedBatches, totalBatches }) // First request carries the entity file (the server needs the manifest before parallel batches). - const first = await sendRequest([entityId, ...(batches[0]?.hashes ?? [])]) + const first = await sendRequest([entityId, ...(batches[0]?.hashes ?? [])], callerSignal) if (first.kind === 'deployed') { return first.result } @@ -357,6 +375,8 @@ export function createContentClient(options: ClientOptions): ContentClient { // Remaining batches through a bounded worker pool. First 200 wins; a terminal error aborts the rest. const remaining = batches.slice(1) const controller = new AbortController() + // Combined once per session: aborts when the caller cancels or when the pool wins/fails. + const sessionSignal = combineSignals([callerSignal, controller.signal]) let deployed: PartialDeploymentResult | undefined let terminalError: DeploymentError | undefined let nextIndex = 0 @@ -369,7 +389,7 @@ export function createContentClient(options: ClientOptions): ContentClient { } const batch = remaining[index] try { - const outcome = await sendRequest(batch.hashes, controller.signal) + const outcome = await sendRequest(batch.hashes, sessionSignal) if (outcome.kind === 'deployed') { deployed = outcome.result controller.abort() @@ -419,9 +439,16 @@ export function createContentClient(options: ClientOptions): ContentClient { if (error instanceof DeploymentError) { throw error } + // The caller cancelled: the failed request is the abort taking effect, not a transient error — + // surface it instead of burning resume attempts uploading after cancellation. + if (callerSignal?.aborted) { + throw error + } lastError = error if (attempt < maxResumeAttempts) { - await delay(resumeDelay) + // Exponential backoff: transient conditions (429 rate limits, 5xx blips) rarely clear within + // a fixed short delay, and hammering a rate limiter only extends the window. + await delay(resumeDelay * 2 ** attempt) } } } diff --git a/src/client/types.ts b/src/client/types.ts index 2c21560..db1c9bf 100644 --- a/src/client/types.ts +++ b/src/client/types.ts @@ -91,13 +91,16 @@ export type PartialDeploymentProgress = { } export type PartialDeploymentOptions = { - /** Max summed file bytes per request. Default {@link DEFAULT_MAX_BATCH_SIZE_BYTES} (50 MiB). */ + /** + * Max summed file bytes per request. Default {@link DEFAULT_MAX_BATCH_SIZE_BYTES} (100 MiB — safely + * under the ~200MB request ceiling of the infrastructure in front of the content servers). + */ maxBatchSizeBytes?: number /** Concurrent batch uploads after the first (entity-carrying) request. Default 2. */ concurrency?: number - /** Resume cycles (re-query available content + re-upload) on network/5xx failures. Default 3. */ + /** Resume cycles (re-query available content + re-upload) on network/429/5xx failures. Default 3. */ maxResumeAttempts?: number - /** Delay between resume cycles in ms. Default 1000. */ + /** Base delay between resume cycles in ms, doubled on each attempt (exponential backoff). Default 1000. */ resumeDelay?: number /** Invoked after each staged batch with cumulative progress. */ onProgress?: (progress: PartialDeploymentProgress) => void diff --git a/src/client/utils/batching.ts b/src/client/utils/batching.ts index 3c508dc..79a80dd 100644 --- a/src/client/utils/batching.ts +++ b/src/client/utils/batching.ts @@ -1,10 +1,11 @@ /** - * Default cap on the summed file bytes per partial-deployment request. Well under the servers' request - * caps (worlds-content-server 350MB, catalyst 2GiB) and under typical proxy body limits (~100MB), and - * small enough that a failed request only re-uploads at most this much. The cap counts file bytes, not - * exact wire size; the margin absorbs multipart overhead. + * Default cap on the summed file bytes per partial-deployment request. The infrastructure in front of + * the content servers times out requests larger than ~200MB, so 100MB keeps a comfortable safety + * margin (the cap counts file bytes, not exact wire size — the margin also absorbs multipart overhead + * and the entity JSON riding along in the first request) while keeping the request count low for large + * scenes. Also under the servers' own request caps (worlds-content-server 350MB, catalyst 2GiB). */ -export const DEFAULT_MAX_BATCH_SIZE_BYTES = 50 * 1024 * 1024 // 50 MiB +export const DEFAULT_MAX_BATCH_SIZE_BYTES = 100 * 1024 * 1024 // 100 MiB export type FileBatch = { hashes: string[] @@ -13,9 +14,9 @@ export type FileBatch = { /** * Splits files into batches whose summed byte size stays at or below `maxBatchSizeBytes`, using - * first-fit-decreasing bin packing. A single file larger than the cap gets its own batch (the servers' - * per-file caps still apply and would reject it — that is the caller's/server's concern, not this - * function's). Deterministic given the same input. + * first-fit-decreasing bin packing. A single file larger than the cap gets its own batch — files are + * never split across requests, so such a request exceeds the cap and may hit the servers' per-file + * limits or the ~200MB infrastructure timeout. Deterministic given the same input. */ export function splitIntoBatches(files: Map, maxBatchSizeBytes: number): FileBatch[] { const entries = Array.from(files.entries()).sort((a, b) => b[1].byteLength - a[1].byteLength) diff --git a/test/PartialDeployment.spec.ts b/test/PartialDeployment.spec.ts index 57119b8..ddcf785 100644 --- a/test/PartialDeployment.spec.ts +++ b/test/PartialDeployment.spec.ts @@ -178,6 +178,18 @@ describe('deployPartial', () => { }) }) + describe('when a single file exceeds the request cap', () => { + it('should fail fast without uploading anything (files cannot be split across requests)', async () => { + deployData = makeDeployData({ hashHuge: 500 }) + + await expect(client.deployPartial(deployData, { maxBatchSizeBytes: 100 })).rejects.toBeInstanceOf( + PartialDeploymentValidationError + ) + // No availability query and no upload was attempted. + expect(fetcher.fetch).not.toHaveBeenCalled() + }) + }) + describe('when the server responds 429 (rate limited)', () => { it('should resume rather than fail terminally', async () => { deployData = makeDeployData({ hashA: 100 }) @@ -194,33 +206,51 @@ describe('deployPartial', () => { }) describe('when the caller aborts via options.signal', () => { - let controller: AbortController - let sawAbortedSignal: boolean + describe('and the abort happens before the deployment starts', () => { + let controller: AbortController - beforeEach(async () => { - controller = new AbortController() - sawAbortedSignal = false - deployData = makeDeployData({ hashA: 100 }) - // Abort before the request resolves; assert the request carried an already-aborted signal. - controller.abort() - entitiesResponses = [{ status: 200, body: { creationTimestamp: 1 } }] - const originalFetch = fetcher.fetch as jest.Mock - ;(fetcher.fetch as jest.Mock) = jest.fn(async (url: string, init?: any) => { - if (!url.includes('/available-content') && init?.signal?.aborted) { - sawAbortedSignal = true - } - return originalFetch(url, init) + beforeEach(() => { + controller = new AbortController() + controller.abort() + deployData = makeDeployData({ hashA: 100 }) }) - try { - await client.deployPartial(deployData, { signal: controller.signal }) - } catch { - // May resolve or reject depending on timing; the assertion is about signal propagation. - } + it('should reject without performing any request', async () => { + await expect(client.deployPartial(deployData, { signal: controller.signal })).rejects.toThrow('aborted') + expect(fetcher.fetch).not.toHaveBeenCalled() + }) }) - it('should propagate the caller signal to the deployment request', () => { - expect(sawAbortedSignal).toBe(true) + describe('and the abort happens mid-upload', () => { + let controller: AbortController + let sawAbortedSignal: boolean + + beforeEach(async () => { + controller = new AbortController() + sawAbortedSignal = false + deployData = makeDeployData({ hashA: 100 }) + entitiesResponses = [{ status: 200, body: { creationTimestamp: 1 } }] + const originalFetch = fetcher.fetch as jest.Mock + ;(fetcher.fetch as jest.Mock) = jest.fn(async (url: string, init?: any) => { + if (!url.includes('/available-content')) { + // Abort while the deployment request is in flight; the request must carry a signal that + // observes it. + controller.abort() + sawAbortedSignal = !!init?.signal?.aborted + } + return originalFetch(url, init) + }) + + try { + await client.deployPartial(deployData, { signal: controller.signal }) + } catch { + // May resolve or reject depending on timing; the assertion is about signal propagation. + } + }) + + it('should propagate the caller signal to the deployment request', () => { + expect(sawAbortedSignal).toBe(true) + }) }) }) @@ -258,9 +288,9 @@ describe('deployPartial', () => { // on its own and rejects when the pool aborts it after the win. const files = new Map() files.set(entityId, new Uint8Array([1, 2, 3])) - files.set('hashBig', new Uint8Array(300)) - files.set('hashWin', new Uint8Array(240)) - files.set('hashSlow', new Uint8Array(230)) + files.set('hashBig', new Uint8Array(240)) + files.set('hashWin', new Uint8Array(230)) + files.set('hashSlow', new Uint8Array(220)) const parallelDeployData: DeploymentData = { entityId, authChain: [], files } const ok200 = () => ({ diff --git a/test/unit/batching.spec.ts b/test/unit/batching.spec.ts index 0207f6e..96b6e6c 100644 --- a/test/unit/batching.spec.ts +++ b/test/unit/batching.spec.ts @@ -63,8 +63,8 @@ describe('splitIntoBatches', () => { }) describe('when using the default cap', () => { - it('should be 50 MiB', () => { - expect(DEFAULT_MAX_BATCH_SIZE_BYTES).toBe(50 * 1024 * 1024) + it('should be 100 MiB (safely under the ~200MB infrastructure request ceiling)', () => { + expect(DEFAULT_MAX_BATCH_SIZE_BYTES).toBe(100 * 1024 * 1024) }) }) }) From d2a457a3729533f75844984d83d519f03c9246cc Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Wed, 8 Jul 2026 20:42:18 -0300 Subject: [PATCH 05/10] fix: address second-review findings in deployPartial - consume the body of 202 responses: with native fetch an unread response pins the undici socket and buffered bytes, and a large upload produces one 202 per non-finalizing batch - treat a finalizing 200 whose body fails to parse (empty body, proxy rewrite) as the success it is, with a fallback timestamp, instead of burning every resume attempt re-observing the already-deployed entity and reporting failure for a deployment that succeeded - cap the exponential resume backoff at 60s so a large maxResumeAttempts can't grow 2^attempt into effectively-hung multi-hour sleeps - detach the combined abort-signal listeners when a session's worker pool drains, so a long-lived caller signal reused across many uploads doesn't accumulate one listener per session --- src/client/ContentClient.ts | 50 ++++++++++++++++++++++++++++------ test/PartialDeployment.spec.ts | 27 ++++++++++++++++-- 2 files changed, 66 insertions(+), 11 deletions(-) diff --git a/src/client/ContentClient.ts b/src/client/ContentClient.ts index dc0ceb1..8df6503 100644 --- a/src/client/ContentClient.ts +++ b/src/client/ContentClient.ts @@ -19,13 +19,24 @@ function delay(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)) } +// Ceiling for the exponential resume backoff, so a caller-tuned maxResumeAttempts can't grow +// 2^attempt into effectively-hung multi-hour sleeps. +const MAX_RESUME_BACKOFF_MS = 60_000 + // Combines several abort signals into one that aborts as soon as any of them does. Used so a request // honors both the caller's cancellation signal and deployPartial's internal first-200-wins controller. // (Hand-rolled rather than AbortSignal.any, which is only available on Node >= 20.3.) -function combineSignals(signals: Array): AbortSignal | undefined { +// +// `dispose` detaches the listeners from the source signals. Call it when the combined signal is no +// longer needed: sources like a caller-provided signal can outlive many sessions, and undisposed +// listeners would accumulate on them (Node warns past ~10 listeners on one signal). +function combineSignals(signals: Array): { + signal: AbortSignal | undefined + dispose: () => void +} { const present = signals.filter((signal): signal is AbortSignal => !!signal) if (present.length <= 1) { - return present[0] + return { signal: present[0], dispose: () => {} } } const controller = new AbortController() const abort = () => controller.abort() @@ -36,7 +47,14 @@ function combineSignals(signals: Array): AbortSignal | } signal.addEventListener('abort', abort, { once: true }) } - return controller.signal + return { + signal: controller.signal, + dispose: () => { + for (const signal of present) { + signal.removeEventListener('abort', abort) + } + } + } } export type AvailableContentResult = { @@ -325,9 +343,17 @@ export function createContentClient(options: ClientOptions): ContentClient { const requestOptions = mergeRequestOptions(requestBase, { body: form as any, method: 'POST', signal }) const response: FetchResponse = await fetcher.fetch(`${contentUrl}/entities`, requestOptions) if (response.status === 200) { - return { kind: 'deployed', result: (await response.json()) as PartialDeploymentResult } + // A 200 means the server finalized the deployment; a body that fails to parse (empty, proxy + // rewrite) must not turn that success into a retry storm — every retry would re-observe the + // same deployed entity. Fall back to a response-time timestamp. + const result = await response.json().catch(() => ({ creationTimestamp: Date.now() } as PartialDeploymentResult)) + return { kind: 'deployed', result: result as PartialDeploymentResult } } if (response.status === 202) { + // Consume the body even though the outcome doesn't need it: with native fetch an unread + // response body pins the undici socket and buffered bytes, and a large upload produces one + // 202 per non-finalizing batch. + await response.json().catch(() => undefined) return { kind: 'incomplete' } } const body = await response.text().catch(() => '') @@ -375,8 +401,9 @@ export function createContentClient(options: ClientOptions): ContentClient { // Remaining batches through a bounded worker pool. First 200 wins; a terminal error aborts the rest. const remaining = batches.slice(1) const controller = new AbortController() - // Combined once per session: aborts when the caller cancels or when the pool wins/fails. - const sessionSignal = combineSignals([callerSignal, controller.signal]) + // Combined once per session: aborts when the caller cancels or when the pool wins/fails. Disposed + // after the pool drains so a long-lived caller signal doesn't accumulate one listener per session. + const { signal: sessionSignal, dispose: disposeSessionSignal } = combineSignals([callerSignal, controller.signal]) let deployed: PartialDeploymentResult | undefined let terminalError: DeploymentError | undefined let nextIndex = 0 @@ -415,7 +442,11 @@ export function createContentClient(options: ClientOptions): ContentClient { } } - await Promise.all(Array.from({ length: Math.min(concurrency, Math.max(1, remaining.length)) }, () => worker())) + try { + await Promise.all(Array.from({ length: Math.min(concurrency, Math.max(1, remaining.length)) }, () => worker())) + } finally { + disposeSessionSignal() + } if (terminalError) { throw terminalError @@ -447,8 +478,9 @@ export function createContentClient(options: ClientOptions): ContentClient { lastError = error if (attempt < maxResumeAttempts) { // Exponential backoff: transient conditions (429 rate limits, 5xx blips) rarely clear within - // a fixed short delay, and hammering a rate limiter only extends the window. - await delay(resumeDelay * 2 ** attempt) + // a fixed short delay, and hammering a rate limiter only extends the window. Capped so a + // large maxResumeAttempts can't grow 2^attempt into multi-hour/day sleeps. + await delay(Math.min(resumeDelay * 2 ** attempt, MAX_RESUME_BACKOFF_MS)) } } } diff --git a/test/PartialDeployment.spec.ts b/test/PartialDeployment.spec.ts index ddcf785..6c74ebb 100644 --- a/test/PartialDeployment.spec.ts +++ b/test/PartialDeployment.spec.ts @@ -3,7 +3,7 @@ import { PartialDeploymentNotSupportedError, PartialDeploymentValidationError } const URL = 'https://content.example.com' -type EntitiesResponse = { status: number; body?: any; text?: string; throwNetwork?: boolean } +type EntitiesResponse = { status: number; body?: any; text?: string; throwNetwork?: boolean; jsonThrows?: boolean } describe('deployPartial', () => { const entityId = 'bafyEntity' @@ -55,7 +55,12 @@ describe('deployPartial', () => { return { ok: next.status >= 200 && next.status < 300, status: next.status, - json: async () => next.body ?? {}, + json: async () => { + if (next.jsonThrows) { + throw new SyntaxError('Unexpected end of JSON input') + } + return next.body ?? {} + }, text: async () => next.text ?? '', arrayBuffer: async () => new ArrayBuffer(0) } @@ -190,6 +195,24 @@ describe('deployPartial', () => { }) }) + describe('when the finalizing 200 response has an unparseable body', () => { + let result: { creationTimestamp: number } + + beforeEach(async () => { + deployData = makeDeployData({ hashA: 100 }) + // e.g. a proxy strips or rewrites the body: the deployment DID succeed server-side. + entitiesResponses = [{ status: 200, jsonThrows: true }] + + result = await client.deployPartial(deployData) + }) + + it('should still resolve as a success with a fallback timestamp instead of retrying', () => { + expect(typeof result.creationTimestamp).toBe('number') + // One available-content query + one POST — no resume sessions were burned on the parse failure. + expect((fetcher.fetch as jest.Mock).mock.calls).toHaveLength(2) + }) + }) + describe('when the server responds 429 (rate limited)', () => { it('should resume rather than fail terminally', async () => { deployData = makeDeployData({ hashA: 100 }) From 30282e7c618178823a5c9ca7c41daccb3c137c37 Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Wed, 8 Jul 2026 22:57:22 -0300 Subject: [PATCH 06/10] fix: harden deployPartial response and abort handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 200 with an unparseable body: only fabricate a fallback timestamp on a genuine parse failure. If the body read was cancelled (a sibling worker finalized and aborted the pool, or the caller aborted), let the abort propagate — otherwise the loser of a concurrent finalize fabricates a client-clock timestamp that overwrites the winner's real server timestamp, and a caller cancellation mid-read is masked as success - worker pool: abort the controller and await all workers to settle in the finally. A retryable worker throw rejected Promise.all without aborting, so sibling requests kept uploading uncancelable while the resume loop started a new overlapping session - drain 202 bodies via text() instead of json() (drain without the wasted parse) - resume backoff cap never drops below the caller's configured resumeDelay, so a caller that set a large base delay to respect an upstream limiter keeps it adds a regression test for the concurrent-finalize / aborted-loser timestamp. --- src/client/ContentClient.ts | 48 +++++++++++++++------ test/PartialDeployment.spec.ts | 78 ++++++++++++++++++++++++++++++++++ 2 files changed, 113 insertions(+), 13 deletions(-) diff --git a/src/client/ContentClient.ts b/src/client/ContentClient.ts index 8df6503..18d6947 100644 --- a/src/client/ContentClient.ts +++ b/src/client/ContentClient.ts @@ -343,17 +343,29 @@ export function createContentClient(options: ClientOptions): ContentClient { const requestOptions = mergeRequestOptions(requestBase, { body: form as any, method: 'POST', signal }) const response: FetchResponse = await fetcher.fetch(`${contentUrl}/entities`, requestOptions) if (response.status === 200) { - // A 200 means the server finalized the deployment; a body that fails to parse (empty, proxy - // rewrite) must not turn that success into a retry storm — every retry would re-observe the - // same deployed entity. Fall back to a response-time timestamp. - const result = await response.json().catch(() => ({ creationTimestamp: Date.now() } as PartialDeploymentResult)) - return { kind: 'deployed', result: result as PartialDeploymentResult } + // A 200 means the server finalized the deployment. + let result: PartialDeploymentResult + try { + result = (await response.json()) as PartialDeploymentResult + } catch (error) { + // If the read was cancelled (a sibling worker already finalized and aborted the pool, or the + // caller aborted), this is not a successful-but-unparseable body — let the abort propagate so + // the winner's real result stands and cancellation isn't masked as success. + if (signal?.aborted) { + throw error + } + // A genuine unparseable body (empty, proxy rewrite) still means the server finalized; fall + // back to a response-time timestamp rather than triggering a retry storm in which every + // attempt re-observes the same already-deployed entity. + result = { creationTimestamp: Date.now() } as PartialDeploymentResult + } + return { kind: 'deployed', result } } if (response.status === 202) { - // Consume the body even though the outcome doesn't need it: with native fetch an unread - // response body pins the undici socket and buffered bytes, and a large upload produces one - // 202 per non-finalizing batch. - await response.json().catch(() => undefined) + // Drain the body even though the outcome doesn't need it: with native fetch an unread response + // body pins the undici socket and buffered bytes, and a large upload produces one 202 per + // non-finalizing batch. text() drains without the wasted JSON parse. + await response.text().catch(() => undefined) return { kind: 'incomplete' } } const body = await response.text().catch(() => '') @@ -442,9 +454,17 @@ export function createContentClient(options: ClientOptions): ContentClient { } } + const workers = Array.from({ length: Math.min(concurrency, Math.max(1, remaining.length)) }, () => worker()) try { - await Promise.all(Array.from({ length: Math.min(concurrency, Math.max(1, remaining.length)) }, () => worker())) + await Promise.all(workers) } finally { + // Whatever ended the session — a win, a terminal error, or a retryable throw that rejected + // Promise.all — cancel any still-in-flight sibling requests (a retryable throw does NOT abort + // the controller on its own, so without this the losers would keep uploading, uncancelable, + // while the resume loop starts a new session), then wait for them to unwind before disposing + // the signal and returning. + controller.abort() + await Promise.allSettled(workers) disposeSessionSignal() } @@ -478,9 +498,11 @@ export function createContentClient(options: ClientOptions): ContentClient { lastError = error if (attempt < maxResumeAttempts) { // Exponential backoff: transient conditions (429 rate limits, 5xx blips) rarely clear within - // a fixed short delay, and hammering a rate limiter only extends the window. Capped so a - // large maxResumeAttempts can't grow 2^attempt into multi-hour/day sleeps. - await delay(Math.min(resumeDelay * 2 ** attempt, MAX_RESUME_BACKOFF_MS)) + // a fixed short delay, and hammering a rate limiter only extends the window. The cap bounds + // 2^attempt growth, but never below the caller's configured resumeDelay — a caller that set a + // large base delay (e.g. to respect a strict upstream limiter) keeps it as the floor. + const backoff = Math.min(resumeDelay * 2 ** attempt, Math.max(resumeDelay, MAX_RESUME_BACKOFF_MS)) + await delay(backoff) } } } diff --git a/test/PartialDeployment.spec.ts b/test/PartialDeployment.spec.ts index 6c74ebb..472096e 100644 --- a/test/PartialDeployment.spec.ts +++ b/test/PartialDeployment.spec.ts @@ -381,4 +381,82 @@ describe('deployPartial', () => { expect(availableContentCalls).toBe(1) }) }) + + // Both concurrent finalizers get 200; the winner aborts the loser while the loser is still reading + // its body. The loser's aborted body read must NOT be turned into a fabricated success that + // overwrites the winner's real server timestamp. + describe('when two workers both finalize and the loser is aborted mid body-read', () => { + let parallelClient: ContentClient + let result: { creationTimestamp: number } + + beforeEach(async () => { + const files = new Map() + files.set(entityId, new Uint8Array([1, 2, 3])) + files.set('hashBig', new Uint8Array(240)) // ships with entity in req 1 → 202 + files.set('hashWin', new Uint8Array(230)) // 200, real body {creationTimestamp: 42}, wins + files.set('hashSlow', new Uint8Array(220)) // 200, but json() only settles when aborted → rejects + const parallelDeployData: DeploymentData = { entityId, authChain: [], files } + + const parallelFetcher: IFetchComponent = { + fetch: jest.fn(async (url: string, init?: any) => { + if (url.includes('/available-content')) { + const cids = (url.split('?')[1] || '') + .split('&') + .filter((p) => p.startsWith('cid=')) + .map((p) => decodeURIComponent(p.slice('cid='.length))) + return { + ok: true, + status: 200, + json: async () => cids.map((cid) => ({ cid, available: false })), + text: async () => '', + arrayBuffer: async () => new ArrayBuffer(0) + } + } + const form = init.body as FormData + const signal: AbortSignal | undefined = init.signal + if (form.has('hashBig')) { + return { + ok: true, + status: 202, + json: async () => ({ missing: ['hashWin', 'hashSlow'] }), + text: async () => '', + arrayBuffer: async () => new ArrayBuffer(0) + } + } + if (form.has('hashWin')) { + return { + ok: true, + status: 200, + json: async () => ({ creationTimestamp: 42 }), + text: async () => '', + arrayBuffer: async () => new ArrayBuffer(0) + } + } + // hashSlow: a 200 whose body read only settles on abort — and then rejects (cancelled read). + return { + ok: true, + status: 200, + json: () => + new Promise((_resolve, reject) => { + if (signal?.aborted) return reject(new Error('AbortError')) + signal?.addEventListener('abort', () => reject(new Error('AbortError'))) + }), + text: async () => '', + arrayBuffer: async () => new ArrayBuffer(0) + } + }) + } + + parallelClient = createContentClient({ url: URL, fetcher: parallelFetcher }) + result = await parallelClient.deployPartial(parallelDeployData, { + maxBatchSizeBytes: 250, + concurrency: 2, + resumeDelay: 0 + }) + }) + + it("should return the winner's real server timestamp, not a fabricated one", () => { + expect(result).toEqual({ creationTimestamp: 42 }) + }) + }) }) From 300b8ace7a4af26ae8d978d88b1cb61c5c7d4e71 Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Wed, 15 Jul 2026 16:13:18 -0300 Subject: [PATCH 07/10] fix: make deployPartial cancellable, resumable under rate limits, and fail fast on undeliverable content MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - deliver cancellation via an abortController, not a signal: the default @dcl/fetch-component overwrites a caller-supplied signal with its own internal controller and only honors an abortController option, so an in-flight request (caller abort, or the pool's first-200-wins abort) was never actually cancelled on the wire — only the next batch was suppressed - a concurrent win now takes precedence over a sibling worker's retryable failure: previously a retryable throw rejected Promise.all and propagated before the win was returned, discarding a completed deployment and triggering a spurious resume - honor a server Retry-After header as the resume backoff floor (capped), so a redeploy inside a rate-limit window waits the window out instead of exhausting the ~7s exponential backoff and failing terminally - only reject an oversized single file when it actually needs uploading; an already-stored oversized file (shared with a prior deploy) needs zero bytes and must not block the deployment (also fixes the doc contradiction) - read the 202 { missing } body and fail fast with a terminal, diagnostic error when a referenced hash is absent from the provided files, instead of looping through full re-upload sessions to the same dead end --- .vscode/settings.json | 4 +- scripts/README-deploy-profile.md | 381 ++++++++++++++++++++++++++++++ scripts/USAGE-EXAMPLE.md | 317 +++++++++++++++++++++++++ scripts/deploy-profile-example.ts | 279 ++++++++++++++++++++++ src/client/ContentClient.ts | 186 ++++++++++----- src/client/types.ts | 3 + src/client/utils/errors.ts | 13 + test/PartialDeployment.spec.ts | 49 +++- 8 files changed, 1165 insertions(+), 67 deletions(-) create mode 100644 scripts/README-deploy-profile.md create mode 100644 scripts/USAGE-EXAMPLE.md create mode 100644 scripts/deploy-profile-example.ts diff --git a/.vscode/settings.json b/.vscode/settings.json index ac08f77..4705163 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -14,8 +14,8 @@ 120 ], "editor.codeActionsOnSave": { - "source.fixAll.eslint": true, - "source.organizeImports": true + "source.fixAll.eslint": "explicit", + "source.organizeImports": "explicit" }, "eslint.validate": [ "typescript" diff --git a/scripts/README-deploy-profile.md b/scripts/README-deploy-profile.md new file mode 100644 index 0000000..0429225 --- /dev/null +++ b/scripts/README-deploy-profile.md @@ -0,0 +1,381 @@ +# Profile Deployment Script + +This directory contains example scripts demonstrating how to deploy a profile to a Decentraland Catalyst server using the ContentClient. + +## Overview + +The `deploy-profile-example.ts` script supports two modes: + +### Mode A: Create New Profile +Deploy a brand new profile with custom metadata and avatar snapshots. + +### Mode B: Copy Existing Profile +Download an existing profile from any address and redeploy it to a new wallet. This is useful for: +- Migrating profiles between wallets +- Backing up profiles to different servers +- Testing deployments with real profile data +- Cloning profiles for development + +## Complete Flow + +1. **Create/Load Wallet** - Generate random wallet or use private key +2. **Create ContentClient** - Connect to a Catalyst server +3. **Prepare Profile** - Either download existing profile OR create new metadata +4. **Prepare Files** - Either download existing files OR generate avatar snapshots +5. **Build Deployment** - Create the entity with all files and metadata +6. **Create Auth Chain** - Sign the deployment with your wallet +7. **Deploy** - Upload to the content server + +## Prerequisites + +```bash +# Install dependencies +yarn install + +# Install ethers for wallet signing (required for this example) +yarn add -D ethers@5 + +# Build the project +yarn build +``` + +**Note:** This example script uses `ethers` for wallet signing, which is not included by default in the package to keep dependencies minimal. You can also use other signing methods (see the Authentication section below). + +## Usage + +### Basic Usage + +**Create New Profile:** +```bash +npx ts-node scripts/deploy-profile-example.ts +``` + +**Copy Existing Profile:** +```bash +PROFILE_TO_COPY=0x1234567890123456789012345678901234567890 npx ts-node scripts/deploy-profile-example.ts +``` + +**Copy Profile to Your Wallet:** +```bash +PROFILE_TO_COPY=0xSourceAddress PRIVATE_KEY=0xYourPrivateKey npx ts-node scripts/deploy-profile-example.ts +``` + +### Programmatic Usage + +```typescript +import { deployProfile, downloadProfile } from './scripts/deploy-profile-example' + +// Deploy new profile +await deployProfile() + +// Or download a profile separately +import { createFetchComponent } from '@well-known-components/fetch-component' +import { createContentClient } from './src/client/ContentClient' + +const fetcher = createFetchComponent() +const client = createContentClient({ url: 'https://peer.decentraland.org/content', fetcher }) +const { metadata, files } = await downloadProfile(client, '0x1234...5678') +``` + +## Configuration + +The script has two modes: + +### Testing Mode (Default) +```typescript +const USE_RANDOM_WALLET = true // Generate a random wallet for testing +``` + +When `USE_RANDOM_WALLET` is `true`, the script generates a random wallet and displays its private key. This is useful for testing without needing real credentials. + +### Production Mode +```typescript +const USE_RANDOM_WALLET = false +const PRIVATE_KEY = process.env.PRIVATE_KEY // Your wallet's private key +``` + +Set `USE_RANDOM_WALLET` to `false` and provide your private key via environment variable: + +```bash +PRIVATE_KEY=0xYourPrivateKey npx ts-node scripts/deploy-profile-example.ts +``` + +⚠️ **NEVER commit your private key to version control!** + +### Copy Profile Mode +```typescript +const PROFILE_TO_COPY = process.env.PROFILE_TO_COPY || '' +``` + +Set to an Ethereum address to download and copy that profile: + +```bash +# Copy specific profile +PROFILE_TO_COPY=0x1234567890123456789012345678901234567890 npx ts-node scripts/deploy-profile-example.ts + +# Copy profile to your wallet +PROFILE_TO_COPY=0xSourceAddress PRIVATE_KEY=0xDestinationKey npx ts-node scripts/deploy-profile-example.ts +``` + +When set, the script will: +1. Fetch the entity from the source address +2. Download all profile files (face256.png, body.png) +3. Update the metadata to use the new wallet's address +4. Redeploy with the new wallet's signature + +### Other Configuration +```typescript +const CATALYST_URL = 'https://peer.decentraland.org/content' // Catalyst server URL +``` + +## Authentication + +✅ **This example now uses real wallet signatures with ethers.js!** + +The auth chain is created using a proper ECDSA signature: + +### Using @dcl/crypto + +```typescript +import { Authenticator } from '@dcl/crypto' + +async function createAuthChain(entityId: string, privateKey: string) { + const authChain = Authenticator.signPayload( + Authenticator.initializeAuthChain( + ethAddress, + ephemeralIdentity, + 60 * 60 * 24 * 7 // 1 week + ), + entityId + ) + return authChain +} +``` + +### Current Implementation (ethers.js) + +The script uses ethers.js to create a proper auth chain: + +```typescript +import { Wallet } from 'ethers' +import { AuthLinkType } from '@dcl/schemas' + +async function createAuthChain(wallet: Wallet, entityId: string): Promise { + // Sign the entity ID with the wallet + const signature = await wallet.signMessage(entityId) + + // Build the auth chain + const authChain: AuthChain = [ + { + type: AuthLinkType.SIGNER, + payload: wallet.address, + signature: '' + }, + { + type: AuthLinkType.ECDSA_PERSONAL_SIGNED_ENTITY, + payload: entityId, + signature: signature + } + ] + + return authChain +} +``` + +### Using MetaMask (Browser) + +```typescript +async function signWithMetaMask(entityId: string) { + const accounts = await window.ethereum.request({ + method: 'eth_requestAccounts' + }) + const address = accounts[0] + + const signature = await window.ethereum.request({ + method: 'personal_sign', + params: [entityId, address] + }) + + return [ + { + type: 'SIGNER', + payload: address, + signature: '' + }, + { + type: 'ECDSA_SIGNED_ENTITY', + payload: entityId, + signature: signature + } + ] +} +``` + +## Profile Files + +Avatar snapshots are PNG images that represent your avatar: + +- **face256.png** - 256x256px face snapshot +- **body.png** - Full body snapshot + +### Loading from Files + +```typescript +import fs from 'fs' +import path from 'path' + +async function prepareProfileFiles() { + const files = new Map() + + files.set('face256.png', fs.readFileSync('./assets/face256.png')) + files.set('body.png', fs.readFileSync('./assets/body.png')) + + return files +} +``` + +### Generating Snapshots + +For generating snapshots programmatically, you can: + +1. Use the Decentraland avatar renderer +2. Generate them in Unity using the Decentraland SDK +3. Use a headless browser to capture the avatar from the Explorer + +## Profile Metadata Structure + +```typescript +{ + avatars: [{ + userId: string // Ethereum address (lowercase) + email?: string // Optional email + name: string // Display name + hasClaimedName: boolean // Whether name is claimed on-chain + description: string // Profile description + ethAddress: string // Ethereum address (checksummed) + version: number // Profile version + avatar: { + bodyShape: string // URN of body shape + snapshots: { + face256: string // Hash of face snapshot + body: string // Hash of body snapshot + }, + eyes: { color: RGB } // Eye color + hair: { color: RGB } // Hair color + skin: { color: RGB } // Skin color + wearables: string[] // Array of wearable URNs + }, + tutorialStep: number // Tutorial progress + interests: string[] // User interests + }] +} +``` + +## Testing + +### Test on Sepolia Testnet + +```typescript +const CATALYST_URL = 'https://peer.decentraland.zone/content' +``` + +### Verify Deployment + +After deployment, you can verify your profile at: + +``` +https://peer.decentraland.org/content/entities/profile?pointer=YOUR_ETH_ADDRESS +``` + +## Example: Complete Production Script + +```typescript +import { Wallet } from 'ethers' +import { EntityType, AuthChain, AuthLinkType } from '@dcl/schemas' +import { createContentClient } from '../src/client/ContentClient' +import { buildEntity } from '../src/client/utils/DeploymentBuilder' +import { createFetchComponent } from '@well-known-components/fetch-component' + +async function deployProfileProduction() { + // 1. Setup wallet from private key + const wallet = new Wallet(process.env.PRIVATE_KEY!) + const ethAddress = wallet.address + + // 2. Create client + const fetcher = createFetchComponent() + const client = createContentClient({ + url: 'https://peer.decentraland.org/content', + fetcher + }) + + // 3. Prepare files and metadata + const files = await loadProfileFiles() + const metadata = createProfileMetadata(ethAddress) + + // 4. Build entity + const { entityId, files: entityFiles } = await buildEntity({ + type: EntityType.PROFILE, + pointers: [ethAddress.toLowerCase()], + files, + metadata, + timestamp: Date.now() + }) + + // 5. Sign with wallet and create auth chain + const signature = await wallet.signMessage(entityId) + const authChain: AuthChain = [ + { + type: AuthLinkType.SIGNER, + payload: ethAddress, + signature: '' + }, + { + type: AuthLinkType.ECDSA_PERSONAL_SIGNED_ENTITY, + payload: entityId, + signature + } + ] + + // 6. Deploy + await client.deploy({ + entityId, + files: entityFiles, + authChain + }) + + console.log(`Profile deployed: ${ethAddress}`) +} +``` + +## Troubleshooting + +### Common Errors + +**"Invalid auth chain"** +- Make sure you're using a real wallet signature +- Verify the signature format matches the expected auth chain structure + +**"Content already exists"** +- Files with the same hash already exist on the server (this is OK) +- The deployment should still succeed + +**"Invalid entity"** +- Check that your metadata follows the Profile schema +- Ensure all required fields are present +- Validate wearable URNs are correct + +**"Failed to upload"** +- Check network connection +- Verify the Catalyst server is reachable +- Try a different Catalyst server + +## Resources + +- [Catalyst API Documentation](https://decentraland.github.io/catalyst-api-specs/) +- [DCL Schemas](https://github.com/decentraland/schemas) +- [DCL Crypto](https://github.com/decentraland/crypto) +- [Avatar Documentation](https://docs.decentraland.org/creator/development-guide/sdk7/avatars/) + +## License + +Apache-2.0 diff --git a/scripts/USAGE-EXAMPLE.md b/scripts/USAGE-EXAMPLE.md new file mode 100644 index 0000000..24743f4 --- /dev/null +++ b/scripts/USAGE-EXAMPLE.md @@ -0,0 +1,317 @@ +# Quick Start: Deploy Profile Example + +This guide shows you how to quickly run the profile deployment example. + +## Installation + +```bash +# Clone/navigate to the project +cd catalyst-client + +# Install dependencies +yarn install + +# Install ethers (required for the example) +yarn add -D ethers@5 +``` + +## Running the Example + +The script supports two modes: + +### Mode A: Create New Profile (Default) + +The script will generate a random wallet and deploy a test profile: + +```bash +npx ts-node scripts/deploy-profile-example.ts +``` + +### Mode B: Copy Existing Profile + +Copy an existing profile from one address and redeploy it to a new wallet: + +```bash +# Copy a profile from a specific address +PROFILE_TO_COPY=0x1234567890123456789012345678901234567890 npx ts-node scripts/deploy-profile-example.ts +``` + +This is useful for: +- **Migrating profiles** between wallets +- **Testing deployments** with real profile data +- **Backing up profiles** to different servers +- **Cloning profiles** for development/testing + +### Option 1: Test with Random Wallet (Default Mode A) + +**Output (New Profile):** +``` +🚀 Starting profile deployment... + +🔑 Setting up wallet... +✅ Wallet address: 0x1234567890123456789012345678901234567890 +⚠️ Using randomly generated wallet for testing + Private key: 0xabcdef... + +📡 Creating ContentClient... +✅ Connected to: https://peer.decentraland.org/content + +👤 Preparing profile metadata... +✅ Profile created for: Sample User + +🖼️ Preparing profile files... +✅ Prepared 2 files + +🔨 Building deployment entity... +✅ Entity ID: bafkrei... +✅ Total files in deployment: 3 + +🔐 Creating authentication chain... +✅ Auth chain created with real wallet signature + +📤 Deploying profile to content server... +✅ Profile deployed successfully! + +🎉 New profile created! + View at: https://peer.decentraland.org/content/entities/profile?pointer=0x... +``` + +**Output (Copied Profile):** +``` +🚀 Starting profile deployment... + +🔑 Setting up wallet... +✅ Wallet address: 0x9876543210987654321098765432109876543210 + +📡 Creating ContentClient... +✅ Connected to: https://peer.decentraland.org/content + +📥 Downloading profile from: 0x1234567890123456789012345678901234567890... + Fetching entity for pointer: 0x1234567890123456789012345678901234567890 + ✓ Found entity: bafkrei... + ✓ Entity type: profile + ✓ Content files: 2 + Downloading content files... + ↓ Downloading: face256.png (bafkrei...) + ✓ Downloaded: face256.png (12345 bytes) + ↓ Downloading: body.png (bafkrei...) + ✓ Downloaded: body.png (67890 bytes) +✅ Downloaded profile: Existing User Name +✅ Downloaded 2 files + +🔄 Updating profile to new wallet address... +✅ Profile updated for new address + +🔨 Building deployment entity... +✅ Entity ID: bafkrei... +✅ Total files in deployment: 3 + +🔐 Creating authentication chain... +✅ Auth chain created with real wallet signature + +📤 Deploying profile to content server... +✅ Profile deployed successfully! + +🎉 Profile copied from 0x1234567890123456789012345678901234567890 and redeployed! + View at: https://peer.decentraland.org/content/entities/profile?pointer=0x9876... +``` + +### Option 2: Deploy with Your Own Wallet (Mode A - New Profile) + +```bash +# Set your private key as environment variable +export PRIVATE_KEY="0xYourPrivateKeyHere" + +# Edit the script to set USE_RANDOM_WALLET = false +# Or run with inline environment variable +PRIVATE_KEY=0xYourPrivateKeyHere npx ts-node scripts/deploy-profile-example.ts +``` + +### Option 3: Copy Profile to Your Wallet (Mode B - Copy Profile) + +Copy your own profile or someone else's to a new wallet: + +```bash +# Copy your profile to a new wallet (random) +PROFILE_TO_COPY=0xYourCurrentAddress npx ts-node scripts/deploy-profile-example.ts + +# Copy a profile to your specific wallet +PROFILE_TO_COPY=0xSourceAddress PRIVATE_KEY=0xYourKey npx ts-node scripts/deploy-profile-example.ts +``` + +**Real-world example:** +```bash +# Copy profile from address 0xabc...123 to your wallet +PROFILE_TO_COPY=0xabc...123 PRIVATE_KEY=0xdef...456 npx ts-node scripts/deploy-profile-example.ts +``` + +**⚠️ Security Warning**: Never commit your private key or share it. Use environment variables or a `.env` file (and add it to `.gitignore`). + +## What the Script Does + +### Mode A: New Profile +1. **Creates/Loads Wallet** - Generates a random wallet or uses your private key +2. **Connects to Catalyst** - Creates a client connected to the Catalyst content server +3. **Prepares Profile Data** - Sets up avatar metadata (wearables, colors, etc.) +4. **Prepares Files** - Creates dummy avatar snapshot images +5. **Builds Entity** - Constructs the deployment with all files and metadata +6. **Signs with Wallet** - Creates a proper authentication chain using ECDSA signatures +7. **Deploys** - Uploads the profile to the content server + +### Mode B: Copy Profile +1. **Creates/Loads Wallet** - Generates a random wallet or uses your private key (destination) +2. **Connects to Catalyst** - Creates a client connected to the Catalyst content server +3. **Downloads Profile** - Fetches existing profile entity from specified address +4. **Downloads Files** - Downloads all avatar snapshot files from the content server +5. **Updates Metadata** - Modifies profile to use new wallet's address +6. **Builds Entity** - Constructs the deployment with downloaded files and updated metadata +7. **Signs with Wallet** - Creates a proper authentication chain using ECDSA signatures +8. **Deploys** - Uploads the copied profile to the content server with new ownership + +## Customization + +### Change Catalyst Server + +Edit in the script: +```typescript +const CATALYST_URL = 'https://peer.decentraland.zone/content' // Testnet +// or +const CATALYST_URL = 'https://peer.decentraland.org/content' // Mainnet +``` + +### Use Real Avatar Images + +Replace the dummy files in `prepareProfileFiles()`: + +```typescript +async function prepareProfileFiles() { + const files = new Map() + + // Load your actual images + files.set('face256.png', fs.readFileSync('./my-avatar/face256.png')) + files.set('body.png', fs.readFileSync('./my-avatar/body.png')) + + return files +} +``` + +### Customize Profile Metadata + +Edit the `createProfileMetadata()` function: + +```typescript +const avatar: Avatar = { + userId: ethAddress.toLowerCase(), + name: 'Your Custom Name', + description: 'Your custom description', + // ... customize wearables, colors, etc. + avatar: { + wearables: [ + 'urn:decentraland:matic:collections-v2:0xabc...:0', + // Add your owned wearables + ] + } +} +``` + +## Authentication Implementation + +The script uses **ethers.js** to create a real, valid authentication chain: + +```typescript +import { Wallet } from 'ethers' +import { AuthLinkType } from '@dcl/schemas' + +// Create wallet +const wallet = Wallet.createRandom() // or new Wallet(privateKey) + +// Sign the entity ID +const signature = await wallet.signMessage(entityId) + +// Build auth chain +const authChain = [ + { + type: AuthLinkType.SIGNER, + payload: wallet.address, + signature: '' + }, + { + type: AuthLinkType.ECDSA_PERSONAL_SIGNED_ENTITY, + payload: entityId, + signature: signature + } +] +``` + +This creates a **valid, production-ready signature** that proves you own the wallet address. + +## Troubleshooting + +### "Cannot find module 'ethers'" + +Install ethers: +```bash +yarn add -D ethers@5 +``` + +### "Invalid signature" or "Unauthorized" + +- Make sure you're using the correct private key +- Verify the wallet address matches the profile pointer +- Check that the signature is being created correctly + +### "Entity already exists" + +This usually means the profile was already deployed with the same content. This is normal and doesn't prevent the deployment from succeeding. + +### "Connection refused" or "Network error" + +- Check your internet connection +- Verify the Catalyst server URL is correct +- Try a different Catalyst server + +### "No profile found for address" + +When copying a profile: +- Verify the address has a deployed profile +- Check the address format (should be 0x... format) +- Make sure you're querying the correct Catalyst server +- The address must be lowercase or checksummed correctly + +### "Failed to download [filename]" + +When copying a profile: +- The content file might not be available on this server +- Try a different Catalyst server +- Check network connectivity +- The content hash might be corrupted + +## Next Steps + +- **Customize the profile** with your own avatar data +- **Load real images** instead of dummy files +- **Deploy to testnet first** (peer.decentraland.zone) +- **Verify deployment** by fetching the profile via API +- **Integrate into your application** using the same patterns + +## API Response + +After successful deployment, you can fetch your profile: + +```bash +curl "https://peer.decentraland.org/content/entities/profile?pointer=YOUR_ADDRESS" +``` + +Or programmatically: + +```typescript +const entities = await contentClient.fetchEntitiesByPointers([ethAddress.toLowerCase()]) +console.log(entities[0]) +``` + +## Resources + +- [Full README](./README-deploy-profile.md) - Detailed documentation +- [Catalyst API Specs](https://decentraland.github.io/catalyst-api-specs/) +- [DCL Schemas](https://github.com/decentraland/schemas) +- [Ethers.js Docs](https://docs.ethers.io/v5/) diff --git a/scripts/deploy-profile-example.ts b/scripts/deploy-profile-example.ts new file mode 100644 index 0000000..b18175b --- /dev/null +++ b/scripts/deploy-profile-example.ts @@ -0,0 +1,279 @@ +/** + * Example script demonstrating how to deploy a profile using the ContentClient + * + * This script can either: + * A) Copy an existing profile from an address and redeploy it (useful for migrations/testing) + * B) Create a new profile from scratch + * + * Complete flow: + * 1. Create/load wallet (random or from private key) + * 2. Create ContentClient and connect to Catalyst server + * 3. Either download existing profile OR create new profile metadata + * 4. Build the deployment entity with all files and metadata + * 5. Sign the deployment with wallet (create auth chain) + * 6. Deploy the profile to the content server + * + * Prerequisites: + * - Install ethers: yarn add -D ethers@5 + * + * Usage: + * - Create new profile: npx ts-node scripts/deploy-profile-example.ts + * - Copy existing profile: PROFILE_TO_COPY=0x1234...5678 npx ts-node scripts/deploy-profile-example.ts + * + * Note: ethers is not included by default to keep the package size small. + * It's only needed for this example script. + */ + +import { AuthChain, AuthLinkType, EntityType } from '@dcl/schemas' +import { createFetchComponent } from '@well-known-components/fetch-component' +import { ContentClient, createContentClient } from '../src/client/ContentClient' +import { buildEntity } from '../src/client/utils/DeploymentBuilder' +// @ts-ignore - ethers is an optional dependency for this example +import { Wallet } from 'ethers' + +// Configuration +const CATALYST_URL = 'https://peer-testing-2.decentraland.org/content' +// const CATALYST_URL = 'http://localhost:6969' + +// Option 1: Generate a random wallet (for testing) +const USE_RANDOM_WALLET = true + +// Option 2: Use your own private key (for production) +// Set USE_RANDOM_WALLET to false and provide your private key +const PRIVATE_KEY = process.env.PRIVATE_KEY || '1964b667cb02bf85d20a58b1eda1cd572ca59a5aed7f1798854cba4ea41bd3d5' + +// Profile to copy - set to an ethereum address that has a profile deployed +// Leave empty to create a new profile from scratch +const PROFILE_TO_COPY = process.env.PROFILE_TO_COPY || '0xedae96f7739af8a7fb16e2a888c1e578e1328299' // Example: '0x1234...5678' + +async function deployProfile() { + console.log('🚀 Starting profile deployment...\n') + + if (!PROFILE_TO_COPY) { + console.error('❌ PROFILE_TO_COPY is not set') + process.exit(1) + } + + // Step 1: Create or load wallet + console.log('🔑 Setting up wallet...') + const wallet = new Wallet(PRIVATE_KEY) + + const ethAddress = wallet.address + const profilePointer = ethAddress.toLowerCase() + + console.log(`✅ Wallet address: ${ethAddress}`) + if (USE_RANDOM_WALLET) { + console.log(`⚠️ Using randomly generated wallet for testing`) + console.log(` Private key: ${wallet.privateKey}`) + } + console.log() + + // Step 2: Create the fetcher and ContentClient + console.log('📡 Creating ContentClient...') + const fetcher = createFetchComponent() + const contentClient = createContentClient({ + url: CATALYST_URL, + fetcher + }) + console.log(`✅ Connected to: ${CATALYST_URL}\n`) + + // Step 3: Prepare profile metadata and files + let profileMetadata: any + let files: Map + + // Download existing profile + console.log(`📥 Downloading profile from: ${PROFILE_TO_COPY}...\n`) + const downloadedProfile = await downloadProfile(contentClient, PROFILE_TO_COPY) + profileMetadata = downloadedProfile.metadata + files = downloadedProfile.files + + console.log(`✅ Downloaded profile: ${profileMetadata.avatars[0].name}`) + console.log(`✅ Downloaded ${files.size} files\n`) + + // Update the profile to use the new wallet's address + console.log('🔄 Updating profile to new wallet address...') + profileMetadata.avatars[0].userId = ethAddress.toLowerCase() + profileMetadata.avatars[0].ethAddress = ethAddress + // Remove the snapshots from the avatar + delete profileMetadata.avatars[0].avatar['snapshots'] + + console.log(`✅ Profile updated for new address\n`) + + // Step 4: Build the deployment + console.log('🔨 Building deployment entity...') + const deploymentPreparation = await buildEntity({ + type: EntityType.PROFILE, + pointers: [profilePointer], + // Do not include the files in the deployment + // files: files, + metadata: profileMetadata, + timestamp: Date.now() + }) + + console.log('Deployment preparation:', deploymentPreparation) + const decoder = new TextDecoder('utf-8') + console.log( + 'Deployment preparation files:', + decoder.decode(deploymentPreparation.files.get(deploymentPreparation.entityId)) + ) + + console.log(`✅ Entity ID: ${deploymentPreparation.entityId}`) + console.log(`✅ Total files in deployment: ${deploymentPreparation.files.size}\n`) + + // Step 5: Create auth chain (sign the deployment) + console.log('🔐 Creating authentication chain...') + const authChain = await createAuthChain(wallet, deploymentPreparation.entityId) + console.log('✅ Auth chain created with real wallet signature\n') + + // Step 6: Deploy to the content server + console.log('📤 Deploying profile to content server...') + try { + const deploymentData = { + ...deploymentPreparation, + authChain + } + + const response = await contentClient.deploy(deploymentData) + console.log('✅ Profile deployed successfully!') + console.log(`📝 Deployment response:`, response) + + if (PROFILE_TO_COPY) { + console.log(`\n🎉 Profile copied from ${PROFILE_TO_COPY} and redeployed!`) + } else { + console.log(`\n🎉 New profile created!`) + } + console.log(` View at: ${CATALYST_URL}/entities/profile?pointer=${profilePointer}`) + } catch (error) { + console.error('❌ Deployment failed:', error) + throw error + } +} + +/** + * Downloads an existing profile from the content server + * @param contentClient - The ContentClient instance + * @param ethereumAddress - The ethereum address of the profile to download + * @returns The profile metadata and files + */ +async function downloadProfile( + contentClient: ContentClient, + ethereumAddress: string +): Promise<{ metadata: any; files: Map }> { + // Fetch the entity by pointer (ethereum address) + const pointer = ethereumAddress.toLowerCase() + console.log(` Fetching entity for pointer: ${pointer}`) + + const entities = await contentClient.fetchEntitiesByPointers([pointer]) + + if (entities.length === 0) { + throw new Error(`No profile found for address: ${ethereumAddress}`) + } + + const entity = entities[0] + console.log(` ✓ Found entity: ${entity.id}`) + console.log(` ✓ Entity type: ${entity.type}`) + console.log(` ✓ Content files: ${entity.content?.length || 0}`) + + // Download all content files + const files = new Map() + + if (entity.content && entity.content.length > 0) { + console.log(` Downloading content files...`) + + for (const contentFile of entity.content) { + console.log(` ↓ Downloading: ${contentFile.file} (${contentFile.hash})`) + try { + const fileContent = await contentClient.downloadContent(contentFile.hash) + files.set(contentFile.file, fileContent) + console.log(` ✓ Downloaded: ${contentFile.file} (${fileContent.length} bytes)`) + } catch (error) { + console.error(` ✗ Failed to download ${contentFile.file}:`, error) + throw error + } + } + } + + return { + metadata: entity.metadata, + files + } +} + +/** + * Prepares profile files (avatar snapshots) + * In a real scenario, you would load actual PNG files + */ +async function prepareProfileFiles(): Promise> { + const files = new Map() + + // Option 1: Load from actual files (uncomment if you have image files) + /* + const face256Path = path.join(__dirname, 'assets', 'face256.png') + const bodyPath = path.join(__dirname, 'assets', 'body.png') + + if (fs.existsSync(face256Path)) { + files.set('face256.png', fs.readFileSync(face256Path)) + } + + if (fs.existsSync(bodyPath)) { + files.set('body.png', fs.readFileSync(bodyPath)) + } + */ + + // Option 2: Create dummy files for testing (current implementation) + // In production, replace these with actual avatar snapshot images + const dummyFace = Buffer.from('dummy-face256-png-content') + const dummyBody = Buffer.from('dummy-body-png-content') + + files.set('face256.png', dummyFace) + files.set('body.png', dummyBody) + + return files +} + +/** + * Creates an authentication chain for the deployment using ethers Wallet + * + * The auth chain proves ownership of the Ethereum address by signing the entity ID + * + * @param wallet - The ethers Wallet instance to sign with + * @param entityId - The entity ID to sign (returned from buildEntity) + * @returns AuthChain that can be used for deployment + */ +async function createAuthChain(wallet: Wallet, entityId: string): Promise { + // Sign the entity ID with the wallet + // This creates a signature that proves the wallet owner approves this deployment + const signature = await wallet.signMessage(entityId) + + // Build the auth chain + // The auth chain is an array that traces the authorization from the address to the signature + const authChain: AuthChain = [ + { + type: AuthLinkType.SIGNER, + payload: wallet.address, + signature: '' + }, + { + type: AuthLinkType.ECDSA_PERSONAL_SIGNED_ENTITY, + payload: entityId, + signature: signature + } + ] + + return authChain +} + +// Run the deployment +if (require.main === module) { + deployProfile() + .then(() => { + console.log('\n✨ Script completed successfully') + process.exit(0) + }) + .catch((error) => { + console.error('\n💥 Script failed:', error) + process.exit(1) + }) +} + +export { createAuthChain, deployProfile, downloadProfile, prepareProfileFiles } diff --git a/src/client/ContentClient.ts b/src/client/ContentClient.ts index 18d6947..c94e7b7 100644 --- a/src/client/ContentClient.ts +++ b/src/client/ContentClient.ts @@ -12,7 +12,12 @@ import { } from './types' import { addModelToFormData, mergeRequestOptions, sanitizeUrl, splitAndFetch } from './utils/Helper' import { DEFAULT_MAX_BATCH_SIZE_BYTES, splitIntoBatches } from './utils/batching' -import { DeploymentError, PartialDeploymentNotSupportedError, PartialDeploymentValidationError } from './utils/errors' +import { + DeploymentError, + PartialDeploymentNotSupportedError, + PartialDeploymentValidationError, + RetryablePartialDeploymentError +} from './utils/errors' import { retry } from './utils/retry' function delay(ms: number): Promise { @@ -57,6 +62,23 @@ function combineSignals(signals: Array): { } } +// Parse an HTTP `Retry-After` value (delta-seconds, or an HTTP date) into milliseconds. Returns +// undefined when absent or unparseable, so the caller falls back to its exponential backoff. +function parseRetryAfterMs(headerValue: string | undefined): number | undefined { + if (!headerValue) { + return undefined + } + const seconds = Number(headerValue) + if (Number.isFinite(seconds)) { + return Math.max(0, seconds * 1000) + } + const dateMs = Date.parse(headerValue) + if (!Number.isNaN(dateMs)) { + return Math.max(0, dateMs - Date.now()) + } + return undefined +} + export type AvailableContentResult = { cid: string available: boolean @@ -291,18 +313,6 @@ export function createContentClient(options: ClientOptions): ContentClient { const sizeOf = (hash: string) => deployData.files.get(hash)?.byteLength ?? 0 const totalBytes = allContentHashes.reduce((acc, hash) => acc + sizeOf(hash), 0) - // Files are never split across requests, so a single file above the request cap would produce a - // request the infrastructure may time out — and the resume loop would re-upload it on every - // attempt. Fail fast with a clear message instead of uploading hundreds of MB that cannot succeed. - const oversized = allContentHashes.find((hash) => sizeOf(hash) > maxBatchSizeBytes) - if (oversized) { - throw new PartialDeploymentValidationError( - `The file '${oversized}' (${sizeOf(oversized)} bytes) exceeds the maximum request size of ` + - `${maxBatchSizeBytes} bytes and files cannot be split across requests. If your infrastructure ` + - `allows larger requests, raise options.maxBatchSizeBytes.` - ) - } - // Request-scoped options without the partial-specific keys or the abort signal (the caller signal // and the pool controller are combined once per session in runSession and threaded from there). const requestBase: RequestOptions = { @@ -340,42 +350,76 @@ export function createContentClient(options: ClientOptions): ContentClient { // abort listener on the caller's signal per batch. const sendRequest = async (fileHashes: string[], signal?: AbortSignal): Promise => { const form = buildDeploymentForm(deployData, fileHashes, true) - const requestOptions = mergeRequestOptions(requestBase, { body: form as any, method: 'POST', signal }) - const response: FetchResponse = await fetcher.fetch(`${contentUrl}/entities`, requestOptions) - if (response.status === 200) { - // A 200 means the server finalized the deployment. - let result: PartialDeploymentResult - try { - result = (await response.json()) as PartialDeploymentResult - } catch (error) { - // If the read was cancelled (a sibling worker already finalized and aborted the pool, or the - // caller aborted), this is not a successful-but-unparseable body — let the abort propagate so - // the winner's real result stands and cancellation isn't masked as success. - if (signal?.aborted) { - throw error + // The default @dcl/fetch-component HONORS an `abortController` option but OVERWRITES a caller + // `signal`, so cancellation must be delivered via a controller. Use a per-request controller (not + // the shared session one) linked to the session signal, so the fetcher's timeout abort cancels + // only this request while a session abort (caller cancel or the pool's first-200-wins) still + // propagates here. + const abortController = new AbortController() + const onAbort = () => abortController.abort() + if (signal) { + if (signal.aborted) abortController.abort() + else signal.addEventListener('abort', onAbort, { once: true }) + } + try { + const requestOptions = mergeRequestOptions(requestBase, { body: form as any, method: 'POST', abortController }) + const response: FetchResponse = await fetcher.fetch(`${contentUrl}/entities`, requestOptions) + if (response.status === 200) { + // A 200 means the server finalized the deployment. + let result: PartialDeploymentResult + try { + result = (await response.json()) as PartialDeploymentResult + } catch (error) { + // If the read was cancelled (a sibling worker already finalized and aborted the pool, or the + // caller aborted), this is not a successful-but-unparseable body — let the abort propagate so + // the winner's real result stands and cancellation isn't masked as success. + if (signal?.aborted) { + throw error + } + // A genuine unparseable body (empty, proxy rewrite) still means the server finalized; fall + // back to a response-time timestamp rather than triggering a retry storm in which every + // attempt re-observes the same already-deployed entity. + result = { creationTimestamp: Date.now() } as PartialDeploymentResult } - // A genuine unparseable body (empty, proxy rewrite) still means the server finalized; fall - // back to a response-time timestamp rather than triggering a retry storm in which every - // attempt re-observes the same already-deployed entity. - result = { creationTimestamp: Date.now() } as PartialDeploymentResult + return { kind: 'deployed', result } } - return { kind: 'deployed', result } - } - if (response.status === 202) { - // Drain the body even though the outcome doesn't need it: with native fetch an unread response - // body pins the undici socket and buffered bytes, and a large upload produces one 202 per - // non-finalizing batch. text() drains without the wasted JSON parse. - await response.text().catch(() => undefined) - return { kind: 'incomplete' } - } - const body = await response.text().catch(() => '') - // 429 (rate limited / other transient conditions the server surfaces as 429) and 5xx are - // retryable — the resume loop re-queries available content and re-uploads only what's missing. - if (response.status === 429 || response.status >= 500) { - throw new Error(`Server responded with status ${response.status}: ${body}`) + if (response.status === 202) { + // Not finalized yet. Read the reported missing hashes: any hash the server still needs that the + // caller never provided locally can NEVER be uploaded, so fail fast with a clear, terminal + // diagnosis naming it — instead of looping through full re-upload sessions to the same dead end. + let missing: string[] = [] + try { + missing = ((await response.json()) as { missing?: string[] })?.missing ?? [] + } catch { + // A missing/unparseable body is fine — treat it as "no diagnosis available" and continue. + } + const undeliverable = missing.filter((hash) => !deployData.files.has(hash)) + if (undeliverable.length > 0) { + throw new PartialDeploymentValidationError( + `The server is still missing content the deployment does not include: ${undeliverable.join( + ', ' + )}. These hashes are referenced by the entity but absent from the provided files.` + ) + } + return { kind: 'incomplete' } + } + const body = await response.text().catch(() => '') + // 429 (rate limited / other transient conditions the server surfaces as 429) and 5xx are + // retryable — the resume loop re-queries available content and re-uploads only what's missing. + // Carry any Retry-After so the resume backoff can wait out the server's window (a rate-limit + // window is typically far longer than the default backoff). + if (response.status === 429 || response.status >= 500) { + const retryAfterMs = parseRetryAfterMs(response.headers?.get('retry-after') ?? undefined) + throw new RetryablePartialDeploymentError( + `Server responded with status ${response.status}: ${body}`, + retryAfterMs + ) + } + // Any other non-2xx (4xx) is a terminal validation or not-supported error. + throw classify4xx(response.status, body) + } finally { + if (signal) signal.removeEventListener('abort', onAbort) } - // Any other non-2xx (4xx) is a terminal validation or not-supported error. - throw classify4xx(response.status, body) } const runSession = async (): Promise => { @@ -392,6 +436,21 @@ export function createContentClient(options: ClientOptions): ContentClient { missingFiles.set(hash, deployData.files.get(hash)!) } } + + // Files are never split across a request, so a single file that must be UPLOADED and is above the + // request cap would produce a request the infrastructure may time out, re-uploaded on every resume. + // Fail fast — but only for files actually missing on the server: an already-stored oversized file + // (e.g. shared with a prior deploy) needs zero bytes uploaded, so it must not block the deployment. + for (const [hash, file] of missingFiles) { + if (file.byteLength > maxBatchSizeBytes) { + throw new PartialDeploymentValidationError( + `The file '${hash}' (${file.byteLength} bytes) exceeds the maximum request size of ` + + `${maxBatchSizeBytes} bytes and files cannot be split across requests. If your infrastructure ` + + `allows larger requests, raise options.maxBatchSizeBytes.` + ) + } + } + const batches = splitIntoBatches(missingFiles, maxBatchSizeBytes) const totalBatches = Math.max(1, batches.length) @@ -455,24 +514,35 @@ export function createContentClient(options: ClientOptions): ContentClient { } const workers = Array.from({ length: Math.min(concurrency, Math.max(1, remaining.length)) }, () => worker()) + let sessionError: unknown try { await Promise.all(workers) + } catch (error) { + // A worker threw a retryable error, rejecting Promise.all. Capture it rather than letting it + // propagate directly: a concurrent win (checked first below) must take precedence, so a sibling's + // retryable failure landing at the same instant as a win can't discard the win and trigger a + // spurious resume. + sessionError = error } finally { - // Whatever ended the session — a win, a terminal error, or a retryable throw that rejected - // Promise.all — cancel any still-in-flight sibling requests (a retryable throw does NOT abort - // the controller on its own, so without this the losers would keep uploading, uncancelable, - // while the resume loop starts a new session), then wait for them to unwind before disposing - // the signal and returning. + // Whatever ended the session — a win, a terminal error, or a retryable throw — cancel any + // still-in-flight sibling requests (a retryable throw does NOT abort the controller on its own, + // so without this the losers would keep uploading, uncancelable, while the resume loop starts a + // new session), then wait for them to unwind before disposing the signal and returning. controller.abort() await Promise.allSettled(workers) disposeSessionSignal() } + // A win wins, even if a sibling failed (retryably or terminally) in the same tick: the entity is + // deployed, so any concurrent rejection is moot. + if (deployed) { + return deployed + } if (terminalError) { throw terminalError } - if (deployed) { - return deployed + if (sessionError) { + throw sessionError } // Every batch returned 202 but the server never finalized: the pending upload's state must have @@ -500,8 +570,14 @@ export function createContentClient(options: ClientOptions): ContentClient { // Exponential backoff: transient conditions (429 rate limits, 5xx blips) rarely clear within // a fixed short delay, and hammering a rate limiter only extends the window. The cap bounds // 2^attempt growth, but never below the caller's configured resumeDelay — a caller that set a - // large base delay (e.g. to respect a strict upstream limiter) keeps it as the floor. - const backoff = Math.min(resumeDelay * 2 ** attempt, Math.max(resumeDelay, MAX_RESUME_BACKOFF_MS)) + // large base delay (e.g. to respect a strict upstream limiter) keeps it as the floor. When the + // server sent a Retry-After (e.g. a rate-limit window that outlasts the exponential backoff), + // honor it as a floor so we wait the window out instead of exhausting attempts inside it. + const retryAfterMs = error instanceof RetryablePartialDeploymentError ? error.retryAfterMs ?? 0 : 0 + const backoff = Math.min( + Math.max(resumeDelay * 2 ** attempt, retryAfterMs), + Math.max(resumeDelay, MAX_RESUME_BACKOFF_MS) + ) await delay(backoff) } } diff --git a/src/client/types.ts b/src/client/types.ts index db1c9bf..d99c5a1 100644 --- a/src/client/types.ts +++ b/src/client/types.ts @@ -30,6 +30,9 @@ export type FetchResponse = { json(): Promise text(): Promise arrayBuffer(): Promise + // Optional: both the native-fetch and wkc responses expose a Headers-like object. Read defensively + // (it may be absent on a minimal mock) — used to honor a `Retry-After` on a rate-limited response. + headers?: { get(name: string): string | null } } /** diff --git a/src/client/utils/errors.ts b/src/client/utils/errors.ts index 2f8b6c5..87e834b 100644 --- a/src/client/utils/errors.ts +++ b/src/client/utils/errors.ts @@ -28,3 +28,16 @@ export class PartialDeploymentNotSupportedError extends DeploymentError { this.name = 'PartialDeploymentNotSupportedError' } } + +/** + * A transient failure the resume loop should retry (429 rate-limited, 5xx, network). Deliberately NOT a + * {@link DeploymentError} (those are terminal). `retryAfterMs`, when set from a server `Retry-After` + * header, is used as the backoff floor so the client waits out the server's window instead of exhausting + * its attempts inside it. + */ +export class RetryablePartialDeploymentError extends Error { + constructor(message: string, public readonly retryAfterMs?: number) { + super(message) + this.name = 'RetryablePartialDeploymentError' + } +} diff --git a/test/PartialDeployment.spec.ts b/test/PartialDeployment.spec.ts index 472096e..2ab08b8 100644 --- a/test/PartialDeployment.spec.ts +++ b/test/PartialDeployment.spec.ts @@ -183,15 +183,30 @@ describe('deployPartial', () => { }) }) - describe('when a single file exceeds the request cap', () => { + describe('when a single file that must be uploaded exceeds the request cap', () => { it('should fail fast without uploading anything (files cannot be split across requests)', async () => { deployData = makeDeployData({ hashHuge: 500 }) await expect(client.deployPartial(deployData, { maxBatchSizeBytes: 100 })).rejects.toBeInstanceOf( PartialDeploymentValidationError ) - // No availability query and no upload was attempted. - expect(fetcher.fetch).not.toHaveBeenCalled() + // Availability is queried first (an already-stored oversized file needs no upload), but no + // POST /entities upload is attempted for a genuinely-missing oversized file. + expect(entitiesCalls).toHaveLength(0) + }) + }) + + describe('when an oversized file is already stored on the server', () => { + it('should deploy without rejecting (the oversized file needs zero bytes uploaded)', async () => { + deployData = makeDeployData({ hashHuge: 500, hashA: 50 }) + available = new Set(['hashHuge']) + entitiesResponses = [{ status: 200, body: { creationTimestamp: 5 } }] + + const result = await client.deployPartial(deployData, { maxBatchSizeBytes: 100 }) + + expect(result).toEqual({ creationTimestamp: 5 }) + expect(entitiesCalls).toHaveLength(1) + expect(entitiesCalls[0].has('hashHuge')).toBe(false) }) }) @@ -256,10 +271,11 @@ describe('deployPartial', () => { const originalFetch = fetcher.fetch as jest.Mock ;(fetcher.fetch as jest.Mock) = jest.fn(async (url: string, init?: any) => { if (!url.includes('/available-content')) { - // Abort while the deployment request is in flight; the request must carry a signal that - // observes it. + // Abort while the deployment request is in flight. The stock @dcl/fetch-component honors an + // `abortController` option (not a `signal`), so the request must carry a controller whose + // signal observes the caller's abort. controller.abort() - sawAbortedSignal = !!init?.signal?.aborted + sawAbortedSignal = !!init?.abortController?.signal?.aborted } return originalFetch(url, init) }) @@ -267,16 +283,29 @@ describe('deployPartial', () => { try { await client.deployPartial(deployData, { signal: controller.signal }) } catch { - // May resolve or reject depending on timing; the assertion is about signal propagation. + // May resolve or reject depending on timing; the assertion is about abort propagation. } }) - it('should propagate the caller signal to the deployment request', () => { + it('should propagate the caller abort to the deployment request via an abortController', () => { expect(sawAbortedSignal).toBe(true) }) }) }) + describe('when the entity references a content file the caller did not provide', () => { + it('should fail fast with a terminal error naming the undeliverable hash', async () => { + deployData = makeDeployData({ hashA: 80 }) + // The server keeps reporting a hash missing that is NOT in deployData.files — no amount of + // resuming can ever deliver it, so the client must fail fast instead of looping. + entitiesResponses = [{ status: 202, body: { missing: ['hashGhost'] } }] + + await expect(client.deployPartial(deployData, { resumeDelay: 0 })).rejects.toThrow(/hashGhost/) + // Only the first session's single request was sent — no resume sessions were burned. + expect(entitiesCalls).toHaveLength(1) + }) + }) + describe('when an onProgress callback is provided', () => { it('should report cumulative progress after each staged batch', async () => { deployData = makeDeployData({ hashA: 80, hashB: 80 }) @@ -343,7 +372,7 @@ describe('deployPartial', () => { const form = init.body as FormData if (form.has('hashSlow')) { return await new Promise((_resolve, reject) => { - const signal: AbortSignal | undefined = init.signal + const signal: AbortSignal | undefined = init.abortController?.signal ?? init.signal if (signal?.aborted) return reject(new Error('AbortError')) signal?.addEventListener('abort', () => reject(new Error('AbortError'))) }) @@ -413,7 +442,7 @@ describe('deployPartial', () => { } } const form = init.body as FormData - const signal: AbortSignal | undefined = init.signal + const signal: AbortSignal | undefined = init.abortController?.signal ?? init.signal if (form.has('hashBig')) { return { ok: true, From 3509a8162198c4dfe0330571de898a7c69153662 Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Wed, 23 Sep 2026 16:54:18 -0300 Subject: [PATCH 08/10] feat: drive partial deployment rounds by the server's missing list Aligns deployPartial with ADR-325: /available-content only plans the first round, and when every batch is accepted without publishing, the client sends the hashes from the latest 202 missing list (or an empty batch when nothing is listed) instead of re-querying availability. Retries 408, 409, 429 and 5xx with backoff floored at Retry-After and gives up with a DeploymentError after maxResumeAttempts rounds without progress. The public API is unchanged. --- src/client/ContentClient.ts | 268 +++++++++++++++++---------------- src/client/types.ts | 9 +- src/client/utils/errors.ts | 2 +- test/PartialDeployment.spec.ts | 198 ++++++++++++++++++++++-- 4 files changed, 336 insertions(+), 141 deletions(-) diff --git a/src/client/ContentClient.ts b/src/client/ContentClient.ts index c94e7b7..438db1e 100644 --- a/src/client/ContentClient.ts +++ b/src/client/ContentClient.ts @@ -28,6 +28,9 @@ function delay(ms: number): Promise { // 2^attempt into effectively-hung multi-hour sleeps. const MAX_RESUME_BACKOFF_MS = 60_000 +// Retryable per ADR-325 besides 5xx: processing deadline, replacement conflict, throttling. +const RETRYABLE_STATUSES = new Set([408, 409, 429]) + // Combines several abort signals into one that aborts as soon as any of them does. Used so a request // honors both the caller's cancellation signal and deployPartial's internal first-200-wins controller. // (Hand-rolled rather than AbortSignal.any, which is only available on Node >= 20.3.) @@ -101,7 +104,7 @@ export type ContentClient = { deploy(deployData: DeploymentData, options?: RequestOptions): Promise /** - * Deploys an entity across multiple requests (partial deployment): content files are split into + * Deploys an entity across multiple requests (partial deployment, ADR-325): content files are split into * size-bounded batches and uploaded in several `POST /entities` requests, and the entity only becomes * live once the server has all of it. Useful for scenes too large for a single request. The server * must support the partial-deployment protocol; against one that doesn't, if everything fits in a @@ -313,8 +316,7 @@ export function createContentClient(options: ClientOptions): ContentClient { const sizeOf = (hash: string) => deployData.files.get(hash)?.byteLength ?? 0 const totalBytes = allContentHashes.reduce((acc, hash) => acc + sizeOf(hash), 0) - // Request-scoped options without the partial-specific keys or the abort signal (the caller signal - // and the pool controller are combined once per session in runSession and threaded from there). + // Request options without the partial-specific keys or the abort signal (combined per round). const requestBase: RequestOptions = { headers: options?.headers, timeout: options?.timeout, @@ -343,18 +345,26 @@ export function createContentClient(options: ClientOptions): ContentClient { type BatchOutcome = { kind: 'deployed'; result: PartialDeploymentResult } | { kind: 'incomplete' } - // Sends one request. Throws a terminal DeploymentError for a 4xx validation/not-supported failure, - // and a plain Error for retryable conditions (429 rate-limited, 5xx, network) which the resume loop - // re-attempts. `signal` is already combined by the caller (caller cancellation + pool abort) — it - // is combined once per session rather than per request so a large upload doesn't accumulate one - // abort listener on the caller's signal per batch. + // Upload state shared across rounds. The latest 202 `missing` list is authoritative (ADR-325). + let entityStaged = false + let latestMissing: string[] | undefined + let missingUpdatedThisRound = false + let uploadedBytes = 0 + let completedBatches = 0 + let totalBatches = 1 + const bytesOf = (hashes: Iterable) => { + let total = 0 + for (const hash of hashes) total += sizeOf(hash) + return total + } + const reportProgress = () => onProgress?.({ uploadedBytes, totalBytes, completedBatches, totalBatches }) + + // Sends one request. Throws a terminal DeploymentError for a non-retryable 4xx, and a + // RetryablePartialDeploymentError for 408, 409, 429 and 5xx; network errors propagate as-is. const sendRequest = async (fileHashes: string[], signal?: AbortSignal): Promise => { const form = buildDeploymentForm(deployData, fileHashes, true) - // The default @dcl/fetch-component HONORS an `abortController` option but OVERWRITES a caller - // `signal`, so cancellation must be delivered via a controller. Use a per-request controller (not - // the shared session one) linked to the session signal, so the fetcher's timeout abort cancels - // only this request while a session abort (caller cancel or the pool's first-200-wins) still - // propagates here. + // The default @dcl/fetch-component honors `abortController` but overwrites `signal`, so link a + // per-request controller to the session signal. const abortController = new AbortController() const onAbort = () => abortController.abort() if (signal) { @@ -365,116 +375,112 @@ export function createContentClient(options: ClientOptions): ContentClient { const requestOptions = mergeRequestOptions(requestBase, { body: form as any, method: 'POST', abortController }) const response: FetchResponse = await fetcher.fetch(`${contentUrl}/entities`, requestOptions) if (response.status === 200) { - // A 200 means the server finalized the deployment. let result: PartialDeploymentResult try { result = (await response.json()) as PartialDeploymentResult } catch (error) { - // If the read was cancelled (a sibling worker already finalized and aborted the pool, or the - // caller aborted), this is not a successful-but-unparseable body — let the abort propagate so - // the winner's real result stands and cancellation isn't masked as success. + // A cancelled read (a sibling won, or the caller aborted) must not be mistaken for success. if (signal?.aborted) { throw error } - // A genuine unparseable body (empty, proxy rewrite) still means the server finalized; fall - // back to a response-time timestamp rather than triggering a retry storm in which every - // attempt re-observes the same already-deployed entity. + // The server did publish; an unparseable body must not trigger a retry storm. result = { creationTimestamp: Date.now() } as PartialDeploymentResult } return { kind: 'deployed', result } } if (response.status === 202) { - // Not finalized yet. Read the reported missing hashes: any hash the server still needs that the - // caller never provided locally can NEVER be uploaded, so fail fast with a clear, terminal - // diagnosis naming it — instead of looping through full re-upload sessions to the same dead end. - let missing: string[] = [] + if (fileHashes.includes(entityId)) { + entityStaged = true + } + let missing: string[] | undefined try { - missing = ((await response.json()) as { missing?: string[] })?.missing ?? [] + const parsed = ((await response.json()) as { missing?: unknown })?.missing + missing = Array.isArray(parsed) + ? parsed.filter((hash): hash is string => typeof hash === 'string') + : undefined } catch { - // A missing/unparseable body is fine — treat it as "no diagnosis available" and continue. + // No diagnosis available; the round falls back to re-sending what it sent. } - const undeliverable = missing.filter((hash) => !deployData.files.has(hash)) - if (undeliverable.length > 0) { - throw new PartialDeploymentValidationError( - `The server is still missing content the deployment does not include: ${undeliverable.join( - ', ' - )}. These hashes are referenced by the entity but absent from the provided files.` + if (missing) { + // A hash the caller never provided can never be uploaded. + const undeliverable = missing.filter((hash) => !deployData.files.has(hash)) + if (undeliverable.length > 0) { + throw new PartialDeploymentValidationError( + `The server is still missing content the deployment does not include: ${undeliverable.join( + ', ' + )}. These hashes are referenced by the entity but absent from the provided files.` + ) + } + latestMissing = missing + missingUpdatedThisRound = true + uploadedBytes = Math.max(uploadedBytes, totalBytes - bytesOf(missing)) + } else { + uploadedBytes = Math.min( + totalBytes, + uploadedBytes + bytesOf(fileHashes.filter((hash) => hash !== entityId)) ) } + completedBatches++ + reportProgress() return { kind: 'incomplete' } } const body = await response.text().catch(() => '') - // 429 (rate limited / other transient conditions the server surfaces as 429) and 5xx are - // retryable — the resume loop re-queries available content and re-uploads only what's missing. - // Carry any Retry-After so the resume backoff can wait out the server's window (a rate-limit - // window is typically far longer than the default backoff). - if (response.status === 429 || response.status >= 500) { + if (RETRYABLE_STATUSES.has(response.status) || response.status >= 500) { const retryAfterMs = parseRetryAfterMs(response.headers?.get('retry-after') ?? undefined) throw new RetryablePartialDeploymentError( `Server responded with status ${response.status}: ${body}`, retryAfterMs ) } - // Any other non-2xx (4xx) is a terminal validation or not-supported error. throw classify4xx(response.status, body) } finally { if (signal) signal.removeEventListener('abort', onAbort) } } - const runSession = async (): Promise => { - if (callerSignal?.aborted) { - throw new Error('The partial deployment was aborted by the caller.') - } - const alreadyOnServer = - allContentHashes.length > 0 - ? await hashesAlreadyOnServer(allContentHashes, { ...requestBase, signal: callerSignal }) - : new Set() - const missingFiles = new Map() - for (const hash of allContentHashes) { - if (!alreadyOnServer.has(hash)) { - missingFiles.set(hash, deployData.files.get(hash)!) - } - } - - // Files are never split across a request, so a single file that must be UPLOADED and is above the - // request cap would produce a request the infrastructure may time out, re-uploaded on every resume. - // Fail fast — but only for files actually missing on the server: an already-stored oversized file - // (e.g. shared with a prior deploy) needs zero bytes uploaded, so it must not block the deployment. - for (const [hash, file] of missingFiles) { - if (file.byteLength > maxBatchSizeBytes) { + // Files are never split, so a file above the request cap can't be delivered. + const assertDeliverable = (hashes: string[]) => { + for (const hash of hashes) { + const size = sizeOf(hash) + if (size > maxBatchSizeBytes) { throw new PartialDeploymentValidationError( - `The file '${hash}' (${file.byteLength} bytes) exceeds the maximum request size of ` + + `The file '${hash}' (${size} bytes) exceeds the maximum request size of ` + `${maxBatchSizeBytes} bytes and files cannot be split across requests. If your infrastructure ` + `allows larger requests, raise options.maxBatchSizeBytes.` ) } } + } - const batches = splitIntoBatches(missingFiles, maxBatchSizeBytes) - const totalBatches = Math.max(1, batches.length) - - let uploadedBytes = totalBytes - Array.from(missingFiles.values()).reduce((acc, f) => acc + f.byteLength, 0) - let completedBatches = 0 - const reportProgress = () => onProgress?.({ uploadedBytes, totalBytes, completedBatches, totalBatches }) - - // First request carries the entity file (the server needs the manifest before parallel batches). - const first = await sendRequest([entityId, ...(batches[0]?.hashes ?? [])], callerSignal) - if (first.kind === 'deployed') { - return first.result + // Uploads `hashes` once: the entity file rides the first request until the server has staged it, then + // the rest go through a bounded worker pool. First 200 wins; a terminal error aborts the rest. + const runRound = async (hashes: string[]): Promise => { + if (callerSignal?.aborted) { + throw new Error('The partial deployment was aborted by the caller.') } - if (batches[0]) { - uploadedBytes += batches[0].sizeBytes - completedBatches++ - reportProgress() + missingUpdatedThisRound = false + const files = new Map() + for (const hash of hashes) files.set(hash, deployData.files.get(hash)!) + const batches = splitIntoBatches(files, maxBatchSizeBytes) + totalBatches = Math.max(1, completedBatches + batches.length) + + let remaining = batches + if (!entityStaged) { + const first = await sendRequest([entityId, ...(batches[0]?.hashes ?? [])], callerSignal) + if (first.kind === 'deployed') { + return first + } + remaining = batches.slice(1) + } else if (batches.length === 0) { + // Nothing left to send but not published yet: an empty batch lets the server re-check and publish. + remaining = [{ hashes: [], sizeBytes: 0 }] + totalBatches = completedBatches + 1 } - // Remaining batches through a bounded worker pool. First 200 wins; a terminal error aborts the rest. - const remaining = batches.slice(1) const controller = new AbortController() - // Combined once per session: aborts when the caller cancels or when the pool wins/fails. Disposed - // after the pool drains so a long-lived caller signal doesn't accumulate one listener per session. - const { signal: sessionSignal, dispose: disposeSessionSignal } = combineSignals([callerSignal, controller.signal]) + // Combined once per round and disposed after the pool drains, so a long-lived caller signal doesn't + // accumulate listeners. + const { signal: roundSignal, dispose: disposeRoundSignal } = combineSignals([callerSignal, controller.signal]) let deployed: PartialDeploymentResult | undefined let terminalError: DeploymentError | undefined let nextIndex = 0 @@ -485,26 +491,20 @@ export function createContentClient(options: ClientOptions): ContentClient { if (index >= remaining.length) { return } - const batch = remaining[index] try { - const outcome = await sendRequest(batch.hashes, sessionSignal) + const outcome = await sendRequest(remaining[index].hashes, roundSignal) if (outcome.kind === 'deployed') { deployed = outcome.result controller.abort() return } - uploadedBytes += batch.sizeBytes - completedBatches++ - reportProgress() } catch (error) { if (error instanceof DeploymentError) { terminalError = error controller.abort() return } - // A sibling worker already finalized (or the pool was aborted), so this request was - // cancelled on purpose — swallow its abort/network error instead of failing the session - // (which would trigger a spurious resume). Genuine failures still propagate. + // A sibling won or the round was aborted: this request was cancelled on purpose. if (deployed || controller.signal.aborted) { return } @@ -514,77 +514,89 @@ export function createContentClient(options: ClientOptions): ContentClient { } const workers = Array.from({ length: Math.min(concurrency, Math.max(1, remaining.length)) }, () => worker()) - let sessionError: unknown + let roundError: unknown try { await Promise.all(workers) } catch (error) { - // A worker threw a retryable error, rejecting Promise.all. Capture it rather than letting it - // propagate directly: a concurrent win (checked first below) must take precedence, so a sibling's - // retryable failure landing at the same instant as a win can't discard the win and trigger a - // spurious resume. - sessionError = error + // Captured so a concurrent win still takes precedence over a sibling's retryable failure. + roundError = error } finally { - // Whatever ended the session — a win, a terminal error, or a retryable throw — cancel any - // still-in-flight sibling requests (a retryable throw does NOT abort the controller on its own, - // so without this the losers would keep uploading, uncancelable, while the resume loop starts a - // new session), then wait for them to unwind before disposing the signal and returning. + // Cancel in-flight siblings and let them unwind before the next round starts. controller.abort() await Promise.allSettled(workers) - disposeSessionSignal() + disposeRoundSignal() } - // A win wins, even if a sibling failed (retryably or terminally) in the same tick: the entity is - // deployed, so any concurrent rejection is moot. if (deployed) { - return deployed + return { kind: 'deployed', result: deployed } } if (terminalError) { throw terminalError } - if (sessionError) { - throw sessionError + if (roundError) { + throw roundError } - - // Every batch returned 202 but the server never finalized: the pending upload's state must have - // changed under us (expired or replaced by an overlapping deployment). Signal a resume. - throw new Error('The partial deployment did not finalize; the pending upload may have expired or been replaced.') + return { kind: 'incomplete' } } + if (callerSignal?.aborted) { + throw new Error('The partial deployment was aborted by the caller.') + } + // /available-content only plans the first round; afterwards the server's `missing` list drives uploads. + const alreadyOnServer = + allContentHashes.length > 0 + ? await hashesAlreadyOnServer(allContentHashes, { ...requestBase, signal: callerSignal }) + : new Set() + let toSend = allContentHashes.filter((hash) => !alreadyOnServer.has(hash)) + uploadedBytes = totalBytes - bytesOf(toSend) + + let failures = 0 + let stalledRounds = 0 let lastError: unknown - for (let attempt = 0; attempt <= maxResumeAttempts; attempt++) { + for (;;) { try { - return await runSession() + assertDeliverable(toSend) + const outcome = await runRound(toSend) + if (outcome.kind === 'deployed') { + return outcome.result + } + const next = missingUpdatedThisRound && latestMissing ? latestMissing : toSend + stalledRounds = missingUpdatedThisRound && next.length < toSend.length ? 0 : stalledRounds + 1 + if (stalledRounds > maxResumeAttempts) { + throw new DeploymentError( + `The partial deployment did not finalize: the server still reports ${next.length} missing file(s) after ` + + `${stalledRounds} round(s) without progress.` + ) + } + toSend = next } catch (error) { - // 4xx validation / not-supported errors are terminal; anything else (5xx, network, non-finalize) - // is retried by re-querying available content and re-uploading only what's still missing. if (error instanceof DeploymentError) { throw error } - // The caller cancelled: the failed request is the abort taking effect, not a transient error — - // surface it instead of burning resume attempts uploading after cancellation. + // A caller abort takes effect as a failed request; surface it instead of retrying. if (callerSignal?.aborted) { throw error } lastError = error - if (attempt < maxResumeAttempts) { - // Exponential backoff: transient conditions (429 rate limits, 5xx blips) rarely clear within - // a fixed short delay, and hammering a rate limiter only extends the window. The cap bounds - // 2^attempt growth, but never below the caller's configured resumeDelay — a caller that set a - // large base delay (e.g. to respect a strict upstream limiter) keeps it as the floor. When the - // server sent a Retry-After (e.g. a rate-limit window that outlasts the exponential backoff), - // honor it as a floor so we wait the window out instead of exhausting attempts inside it. - const retryAfterMs = error instanceof RetryablePartialDeploymentError ? error.retryAfterMs ?? 0 : 0 - const backoff = Math.min( - Math.max(resumeDelay * 2 ** attempt, retryAfterMs), - Math.max(resumeDelay, MAX_RESUME_BACKOFF_MS) - ) - await delay(backoff) + if (failures >= maxResumeAttempts) { + break + } + // Exponential backoff with a Retry-After floor, capped but never below the caller's resumeDelay. + const retryAfterMs = error instanceof RetryablePartialDeploymentError ? error.retryAfterMs ?? 0 : 0 + const backoff = Math.min( + Math.max(resumeDelay * 2 ** failures, retryAfterMs), + Math.max(resumeDelay, MAX_RESUME_BACKOFF_MS) + ) + failures++ + await delay(backoff) + if (latestMissing) { + toSend = latestMissing } } } throw new DeploymentError( - `The partial deployment failed after ${maxResumeAttempts + 1} attempt(s): ${ + `The partial deployment failed after ${failures + 1} attempt(s): ${ lastError instanceof Error ? lastError.message : String(lastError) }` ) diff --git a/src/client/types.ts b/src/client/types.ts index d99c5a1..a44abb6 100644 --- a/src/client/types.ts +++ b/src/client/types.ts @@ -89,7 +89,7 @@ export type PartialDeploymentProgress = { totalBytes: number /** Number of batches confirmed staged so far. */ completedBatches: number - /** Total number of batches to upload this session. */ + /** Total number of batches known so far; grows if the server reports more content missing. */ totalBatches: number } @@ -101,9 +101,12 @@ export type PartialDeploymentOptions = { maxBatchSizeBytes?: number /** Concurrent batch uploads after the first (entity-carrying) request. Default 2. */ concurrency?: number - /** Resume cycles (re-query available content + re-upload) on network/429/5xx failures. Default 3. */ + /** + * Retries after retryable failures (408, 409, 429, 5xx, network), and rounds allowed without progress + * when the server keeps reporting the same content missing. Default 3. + */ maxResumeAttempts?: number - /** Base delay between resume cycles in ms, doubled on each attempt (exponential backoff). Default 1000. */ + /** Base retry delay in ms, doubled on each retry (exponential backoff), never below `Retry-After`. Default 1000. */ resumeDelay?: number /** Invoked after each staged batch with cumulative progress. */ onProgress?: (progress: PartialDeploymentProgress) => void diff --git a/src/client/utils/errors.ts b/src/client/utils/errors.ts index 87e834b..1fa16b6 100644 --- a/src/client/utils/errors.ts +++ b/src/client/utils/errors.ts @@ -30,7 +30,7 @@ export class PartialDeploymentNotSupportedError extends DeploymentError { } /** - * A transient failure the resume loop should retry (429 rate-limited, 5xx, network). Deliberately NOT a + * A transient failure the resume loop should retry (408, 409, 429, 5xx). Deliberately NOT a * {@link DeploymentError} (those are terminal). `retryAfterMs`, when set from a server `Retry-After` * header, is used as the backoff floor so the client waits out the server's window instead of exhausting * its attempts inside it. diff --git a/test/PartialDeployment.spec.ts b/test/PartialDeployment.spec.ts index 2ab08b8..d64a477 100644 --- a/test/PartialDeployment.spec.ts +++ b/test/PartialDeployment.spec.ts @@ -1,9 +1,20 @@ import { createContentClient, ContentClient, IFetchComponent, DeploymentData } from '../src' -import { PartialDeploymentNotSupportedError, PartialDeploymentValidationError } from '../src/client/utils/errors' +import { + DeploymentError, + PartialDeploymentNotSupportedError, + PartialDeploymentValidationError +} from '../src/client/utils/errors' const URL = 'https://content.example.com' -type EntitiesResponse = { status: number; body?: any; text?: string; throwNetwork?: boolean; jsonThrows?: boolean } +type EntitiesResponse = { + status: number + body?: any + text?: string + headers?: Record + throwNetwork?: boolean + jsonThrows?: boolean +} describe('deployPartial', () => { const entityId = 'bafyEntity' @@ -11,6 +22,7 @@ describe('deployPartial', () => { let entitiesResponses: EntitiesResponse[] let available: Set let entitiesCalls: FormData[] + let availableContentCalls: number let fetcher: IFetchComponent let client: ContentClient @@ -27,10 +39,12 @@ describe('deployPartial', () => { entitiesResponses = [] available = new Set() entitiesCalls = [] + availableContentCalls = 0 fetcher = { fetch: jest.fn(async (url: string, init?: any) => { if (url.includes('/available-content')) { + availableContentCalls++ const cids = (url.split('?')[1] || '') .split('&') .filter((p) => p.startsWith('cid=')) @@ -55,6 +69,7 @@ describe('deployPartial', () => { return { ok: next.status >= 200 && next.status < 300, status: next.status, + headers: { get: (name: string) => next.headers?.[name.toLowerCase()] ?? null }, json: async () => { if (next.jsonThrows) { throw new SyntaxError('Unexpected end of JSON input') @@ -69,6 +84,10 @@ describe('deployPartial', () => { client = createContentClient({ url: URL, fetcher }) }) + afterEach(() => { + jest.resetAllMocks() + }) + function entityFileIncluded(form: FormData): boolean { return form.getAll(entityId).some((v) => v instanceof Blob) } @@ -121,23 +140,184 @@ describe('deployPartial', () => { }) describe('when a batch fails with a network error', () => { - it('should resume by re-querying available content and re-uploading the missing hashes', async () => { + let result: { creationTimestamp: number } + + beforeEach(async () => { deployData = makeDeployData({ hashA: 80, hashB: 80 }) entitiesResponses = [ { status: 202, body: { missing: ['hashB'] } }, { throwNetwork: true, status: 0 }, - // resume: hashA now on server, only hashB re-uploaded → finalize { status: 200, body: { creationTimestamp: 5 } } ] - const result = await client.deployPartial(deployData, { - maxBatchSizeBytes: 100, - concurrency: 1, - resumeDelay: 0 - }) + result = await client.deployPartial(deployData, { maxBatchSizeBytes: 100, concurrency: 1, resumeDelay: 0 }) + }) + it('should resolve with the creationTimestamp of the retried request', () => { expect(result).toEqual({ creationTimestamp: 5 }) }) + + it('should retry with only the hashes from the latest missing list, without the entity file', () => { + expect({ + hashB: entitiesCalls[2].has('hashB'), + hashA: entitiesCalls[2].has('hashA'), + entity: entityFileIncluded(entitiesCalls[2]) + }).toEqual({ + hashB: true, + hashA: false, + entity: false + }) + }) + + it('should not query the available content again', () => { + expect(availableContentCalls).toBe(1) + }) + }) + + describe('when every batch is accepted but none publishes the entity', () => { + describe('and the latest 202 lists a hash the available content reported as stored', () => { + let result: { creationTimestamp: number } + + beforeEach(async () => { + deployData = makeDeployData({ hashA: 50, hashB: 50 }) + available = new Set(['hashA']) + entitiesResponses = [ + { status: 202, body: { missing: ['hashA'] } }, + { status: 200, body: { creationTimestamp: 11 } } + ] + + result = await client.deployPartial(deployData, { resumeDelay: 0 }) + }) + + it('should resolve with the creationTimestamp of the follow-up request', () => { + expect(result).toEqual({ creationTimestamp: 11 }) + }) + + it('should send the listed hash in a follow-up request without the entity file', () => { + expect({ hashA: entitiesCalls[1].has('hashA'), entity: entityFileIncluded(entitiesCalls[1]) }).toEqual({ + hashA: true, + entity: false + }) + }) + + it('should trust the missing list instead of querying the available content again', () => { + expect(availableContentCalls).toBe(1) + }) + }) + + describe('and the latest 202 lists nothing missing', () => { + let result: { creationTimestamp: number } + + beforeEach(async () => { + deployData = makeDeployData({ hashA: 50 }) + entitiesResponses = [ + { status: 202, body: { missing: [] } }, + { status: 200, body: { creationTimestamp: 12 } } + ] + + result = await client.deployPartial(deployData, { resumeDelay: 0 }) + }) + + it('should resolve with the creationTimestamp of the follow-up request', () => { + expect(result).toEqual({ creationTimestamp: 12 }) + }) + + it('should send a follow-up request with no files so the server can publish', () => { + expect({ hashA: entitiesCalls[1].has('hashA'), entity: entityFileIncluded(entitiesCalls[1]) }).toEqual({ + hashA: false, + entity: false + }) + }) + }) + + describe('and the server keeps reporting the same hash missing', () => { + let caughtError: unknown + + beforeEach(async () => { + deployData = makeDeployData({ hashA: 50 }) + entitiesResponses = [ + { status: 202, body: { missing: ['hashA'] } }, + { status: 202, body: { missing: ['hashA'] } } + ] + + caughtError = await client + .deployPartial(deployData, { maxResumeAttempts: 1, resumeDelay: 0 }) + .catch((error) => error) + }) + + it('should reject with a terminal DeploymentError explaining the upload did not finalize', () => { + expect({ + isDeploymentError: caughtError instanceof DeploymentError, + message: (caughtError as Error).message + }).toEqual({ + isDeploymentError: true, + message: expect.stringMatching(/did not finalize/) + }) + }) + + it('should stop after the rounds allowed without progress', () => { + expect(entitiesCalls).toHaveLength(2) + }) + }) + }) + + describe.each([408, 409])('when the server responds %s', (status) => { + let result: { creationTimestamp: number } + + beforeEach(async () => { + deployData = makeDeployData({ hashA: 100 }) + entitiesResponses = [ + { status, text: 'transient' }, + { status: 200, body: { creationTimestamp: 21 } } + ] + + result = await client.deployPartial(deployData, { resumeDelay: 0 }) + }) + + it('should retry and resolve with the creationTimestamp instead of failing terminally', () => { + expect(result).toEqual({ creationTimestamp: 21 }) + }) + }) + + describe('when a 429 carries a Retry-After header', () => { + let elapsedMs: number + + beforeEach(async () => { + deployData = makeDeployData({ hashA: 100 }) + entitiesResponses = [ + { status: 429, text: 'Entity rate limited', headers: { 'retry-after': '0.2' } }, + { status: 200, body: { creationTimestamp: 22 } } + ] + const startedAt = Date.now() + + await client.deployPartial(deployData, { resumeDelay: 0 }) + elapsedMs = Date.now() - startedAt + }) + + it('should wait at least the Retry-After delay before retrying', () => { + expect(elapsedMs).toBeGreaterThanOrEqual(190) + }) + }) + + describe('when a quota rejection answers 400', () => { + let caughtError: unknown + + beforeEach(async () => { + deployData = makeDeployData({ hashA: 100 }) + entitiesResponses = [{ status: 400, text: 'Partial upload byte rate exceeded. Retry after one minute.' }] + + caughtError = await client.deployPartial(deployData, { resumeDelay: 0 }).catch((error) => error) + }) + + it('should reject with a PartialDeploymentValidationError without retrying', () => { + expect({ + isValidationError: caughtError instanceof PartialDeploymentValidationError, + calls: entitiesCalls.length + }).toEqual({ + isValidationError: true, + calls: 1 + }) + }) }) describe('when the completing request returns 400', () => { From a55320d56d80b31cca9144d6a0665894530ebc94 Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Fri, 25 Sep 2026 17:25:59 -0300 Subject: [PATCH 09/10] feat: declare partial batches with the partial query parameter deployPartial now posts every batch to /entities?partial=true while keeping the partial=true form field, so the server can tell a batch from a regular deployment before reading the body (a Catalyst whose daily quota for the source is spent otherwise answers 429 before parsing). --- src/client/ContentClient.ts | 3 ++- test/PartialDeployment.spec.ts | 28 ++++++++++++++++++++++++++++ 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/src/client/ContentClient.ts b/src/client/ContentClient.ts index 438db1e..f8f42b3 100644 --- a/src/client/ContentClient.ts +++ b/src/client/ContentClient.ts @@ -373,7 +373,8 @@ export function createContentClient(options: ClientOptions): ContentClient { } try { const requestOptions = mergeRequestOptions(requestBase, { body: form as any, method: 'POST', abortController }) - const response: FetchResponse = await fetcher.fetch(`${contentUrl}/entities`, requestOptions) + // The query flag lets the server tell a batch from a regular deployment before reading the body. + const response: FetchResponse = await fetcher.fetch(`${contentUrl}/entities?partial=true`, requestOptions) if (response.status === 200) { let result: PartialDeploymentResult try { diff --git a/test/PartialDeployment.spec.ts b/test/PartialDeployment.spec.ts index d64a477..3f81a94 100644 --- a/test/PartialDeployment.spec.ts +++ b/test/PartialDeployment.spec.ts @@ -22,6 +22,7 @@ describe('deployPartial', () => { let entitiesResponses: EntitiesResponse[] let available: Set let entitiesCalls: FormData[] + let entitiesUrls: string[] let availableContentCalls: number let fetcher: IFetchComponent let client: ContentClient @@ -39,6 +40,7 @@ describe('deployPartial', () => { entitiesResponses = [] available = new Set() entitiesCalls = [] + entitiesUrls = [] availableContentCalls = 0 fetcher = { @@ -59,6 +61,7 @@ describe('deployPartial', () => { } // POST /entities entitiesCalls.push(init.body as FormData) + entitiesUrls.push(url) const next = entitiesResponses.shift() if (!next) { throw new Error('Unexpected extra POST /entities call') @@ -107,6 +110,31 @@ describe('deployPartial', () => { }) }) + describe('when sending the batches of a multi-batch upload', () => { + let result: unknown + + beforeEach(async () => { + deployData = makeDeployData({ hashA: 80, hashB: 80 }) + entitiesResponses = [ + { status: 202, body: { missing: ['hashB'] } }, + { status: 200, body: { creationTimestamp: 5 } } + ] + result = await client.deployPartial(deployData, { maxBatchSizeBytes: 100, concurrency: 1 }) + }) + + it('should resolve with the creationTimestamp', () => { + expect(result).toEqual({ creationTimestamp: 5 }) + }) + + it('should declare every batch as partial with the query parameter', () => { + expect(entitiesUrls).toEqual([`${URL}/entities?partial=true`, `${URL}/entities?partial=true`]) + }) + + it('should also send the partial form field on every batch', () => { + expect(entitiesCalls.map((form) => form.get('partial'))).toEqual(['true', 'true']) + }) + }) + describe('when the server already stores some hashes', () => { it('should omit those hashes from every request', async () => { deployData = makeDeployData({ hashA: 100, hashB: 100 }) From 5d916954ad83b93fa8c77d35a56b1e607cac89fa Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Tue, 29 Sep 2026 10:36:45 -0300 Subject: [PATCH 10/10] fix: stop partial uploads that are asked to retry later than the client waits Quota rejections now answer 429 with a Retry-After that can be hours away (an upload slot frees only when older uploads expire). The resume loop capped its wait at 60 s, so it retried before Retry-After, which cannot succeed, and failed after its attempts. A Retry-After longer than the backoff ceiling now ends the call with PartialDeploymentRetryLaterError, carrying the server's reason and when to retry. --- src/client/ContentClient.ts | 14 ++++++++---- src/client/utils/errors.ts | 12 ++++++++++ test/PartialDeployment.spec.ts | 41 ++++++++++++++++++++++++++++++++++ 3 files changed, 63 insertions(+), 4 deletions(-) diff --git a/src/client/ContentClient.ts b/src/client/ContentClient.ts index f8f42b3..cf8534b 100644 --- a/src/client/ContentClient.ts +++ b/src/client/ContentClient.ts @@ -15,6 +15,7 @@ import { DEFAULT_MAX_BATCH_SIZE_BYTES, splitIntoBatches } from './utils/batching import { DeploymentError, PartialDeploymentNotSupportedError, + PartialDeploymentRetryLaterError, PartialDeploymentValidationError, RetryablePartialDeploymentError } from './utils/errors' @@ -584,10 +585,15 @@ export function createContentClient(options: ClientOptions): ContentClient { } // Exponential backoff with a Retry-After floor, capped but never below the caller's resumeDelay. const retryAfterMs = error instanceof RetryablePartialDeploymentError ? error.retryAfterMs ?? 0 : 0 - const backoff = Math.min( - Math.max(resumeDelay * 2 ** failures, retryAfterMs), - Math.max(resumeDelay, MAX_RESUME_BACKOFF_MS) - ) + const maxBackoffMs = Math.max(resumeDelay, MAX_RESUME_BACKOFF_MS) + // Retrying before Retry-After can't succeed, so a longer wait ends the call with the server's reason. + if (retryAfterMs > maxBackoffMs) { + throw new PartialDeploymentRetryLaterError( + `${error instanceof Error ? error.message : String(error)} Retry in ${Math.ceil(retryAfterMs / 1000)} s.`, + retryAfterMs + ) + } + const backoff = Math.min(Math.max(resumeDelay * 2 ** failures, retryAfterMs), maxBackoffMs) failures++ await delay(backoff) if (latestMissing) { diff --git a/src/client/utils/errors.ts b/src/client/utils/errors.ts index 1fa16b6..bc96d1e 100644 --- a/src/client/utils/errors.ts +++ b/src/client/utils/errors.ts @@ -41,3 +41,15 @@ export class RetryablePartialDeploymentError extends Error { this.name = 'RetryablePartialDeploymentError' } } + +/** + * The server asked to retry later than the client is willing to wait (e.g. a partial-upload quota that + * frees up only when older uploads expire). Terminal for this call; `retryAfterMs` says when a new + * attempt may succeed, and the message carries the server's explanation. + */ +export class PartialDeploymentRetryLaterError extends DeploymentError { + constructor(message: string, public readonly retryAfterMs: number, status?: number, responseBody?: string) { + super(message, status, responseBody) + this.name = 'PartialDeploymentRetryLaterError' + } +} diff --git a/test/PartialDeployment.spec.ts b/test/PartialDeployment.spec.ts index 3f81a94..a4dc744 100644 --- a/test/PartialDeployment.spec.ts +++ b/test/PartialDeployment.spec.ts @@ -2,6 +2,7 @@ import { createContentClient, ContentClient, IFetchComponent, DeploymentData } f import { DeploymentError, PartialDeploymentNotSupportedError, + PartialDeploymentRetryLaterError, PartialDeploymentValidationError } from '../src/client/utils/errors' @@ -327,6 +328,46 @@ describe('deployPartial', () => { }) }) + describe('when a 429 asks to retry later than the client waits', () => { + let caughtError: unknown + let requestsSent: number + let elapsedMs: number + + beforeEach(async () => { + deployData = makeDeployData({ hashA: 100 }) + entitiesResponses = [ + { + status: 429, + text: 'Too many partial uploads in progress for this account (max 10).', + headers: { 'retry-after': '3600' } + }, + { status: 200, body: { creationTimestamp: 23 } } + ] + const startedAt = Date.now() + + caughtError = await client.deployPartial(deployData, { resumeDelay: 0 }).catch((error) => error) + requestsSent = entitiesCalls.length + elapsedMs = Date.now() - startedAt + }) + + it('should stop without retrying and report the server reason and when to retry', () => { + expect({ + name: (caughtError as Error).name, + message: (caughtError as Error).message, + retryAfterMs: (caughtError as PartialDeploymentRetryLaterError).retryAfterMs, + requestsSent, + waited: elapsedMs >= 1000 + }).toEqual({ + name: 'PartialDeploymentRetryLaterError', + message: + 'Server responded with status 429: Too many partial uploads in progress for this account (max 10). Retry in 3600 s.', + retryAfterMs: 3_600_000, + requestsSent: 1, + waited: false + }) + }) + }) + describe('when a quota rejection answers 400', () => { let caughtError: unknown