feat: support partial (multi-request) scene uploads

.env.default

@@ -81,3 +81,7 @@ SHARED_SECRET_MAX_ATTEMPTS_PER_MINUTE=3
 ######################################
 # How long to keep UNDEPLOYED scenes before permanent deletion (ms). Default: 7 days
 SCENE_EVICTION_TTL_MS=604800000
+# How long a partial (multi-request) deployment may stay pending before it is reclaimed (ms). Default 24h.
+PENDING_DEPLOYMENT_TTL=86400000
+# Max concurrent non-expired pending (partial) uploads a single deployer may have in flight. Default 10.
+MAX_PENDING_DEPLOYMENTS_PER_DEPLOYER=10

docs/database-schema.md

@@ -11,7 +11,20 @@ The database contains the following main tables:
 3. **`world_permissions`** - Stores deployment and streaming permission grants
 4. **`world_permission_parcels`** - Stores parcel-level permission restrictions (normalized)
 5. **`blocked`** - Stores blocked wallet addresses
-6. **`migrations`** - Tracks executed database migrations (internal, managed automatically)
+6. **`pending_scenes`** - Stores partial (multi-request) deployments still being uploaded (not yet live)
+7. **`migrations`** - Tracks executed database migrations (internal, managed automatically)
+
+### Table: `pending_scenes`
+
+Staging area for partial deployments: a scene's content can be uploaded across several `POST /entities`
+requests (with `partial=true`), and the entity only becomes a live `world_scenes` row once every
+referenced file is present. Intentionally has **no** foreign key to `worlds` — a half-uploaded world must
+not create a `worlds` row (which would leak into listings and world-validity checks) before it goes live.
+The authoritative entity bytes live in content storage under the entity id; the `entity` JSONB here is a
+copy used by garbage collection. Columns: `entity_id` (PK), `world_name`, `parcels` (TEXT[]), `entity`
+(JSONB), `deployer`, `created_at`/`updated_at` (TIMESTAMPTZ). At most one non-expired pending scene may
+exist per world + overlapping parcels; rows expire after `PENDING_DEPLOYMENT_TTL` (default 24h) and are
+removed by the daily eviction job. `created_at` anchors the deployment-TTL check for staged uploads.
 
 ---
 

src/adapters/eviction-job.ts

@@ -5,9 +5,9 @@ const DEFAULT_TTL_MS = 7 * 24 * 60 * 60 * 1000 // 7 days
 const ONE_DAY_MS = 24 * 60 * 60 * 1000
 
 export async function createEvictionJob(
-  components: Pick<AppComponents, 'config' | 'logs' | 'worlds'>
+  components: Pick<AppComponents, 'config' | 'logs' | 'worlds' | 'pendingScenesManager'>
 ): Promise<IJobComponent> {
-  const { config, logs, worlds } = components
+  const { config, logs, worlds, pendingScenesManager } = components
   const logger = logs.getLogger('eviction-job')
   const evictionTtlMs = (await config.getNumber('SCENE_EVICTION_TTL_MS')) ?? DEFAULT_TTL_MS
 
@@ -16,7 +16,9 @@ export async function createEvictionJob(
     async () => {
       logger.info('Running eviction job...')
       const evicted = await worlds.evictUndeployedWorlds(evictionTtlMs)
-      logger.info(`Eviction completed. Deleted ${evicted} scene(s).`)
+      // The pending-scenes manager owns the PENDING_DEPLOYMENT_TTL, so expiry uses its configured value.
+      const expiredPending = await pendingScenesManager.deleteExpired()
+      logger.info(`Eviction completed. Deleted ${evicted} scene(s) and ${expiredPending} expired pending upload(s).`)
     },
     ONE_DAY_MS,
     { repeat: true, onError: (err) => logger.error(`Eviction job failed: ${err}`) }

src/adapters/pending-scenes-manager.ts

@@ -0,0 +1,148 @@
+import SQL from 'sql-template-strings'
+import { Entity } from '@dcl/schemas'
+import { InvalidRequestError } from '@dcl/http-commons'
+import { AppComponents, IPendingScenesManager, PendingScene, UpsertPendingScene } from '../types'
+
+const DEFAULT_PENDING_DEPLOYMENT_TTL_MS = 24 * 60 * 60 * 1000 // 24 hours
+
+type PendingSceneRow = {
+  entity_id: string
+  world_name: string
+  parcels: string[]
+  entity: Entity
+  deployer: string
+  created_at: Date
+  updated_at: Date
+}
+
+function toPendingScene(row: PendingSceneRow): PendingScene {
+  return {
+    entityId: row.entity_id,
+    worldName: row.world_name,
+    parcels: row.parcels,
+    entity: row.entity,
+    deployer: row.deployer,
+    createdAt: row.created_at,
+    updatedAt: row.updated_at
+  }
+}
+
+export async function createPendingScenesManager(
+  components: Pick<AppComponents, 'config' | 'database' | 'logs'>
+): Promise<IPendingScenesManager> {
+  const { config, database, logs } = components
+  const logger = logs.getLogger('pending-scenes-manager')
+  const ttlMs = (await config.getNumber('PENDING_DEPLOYMENT_TTL')) ?? DEFAULT_PENDING_DEPLOYMENT_TTL_MS
+
+  async function getByEntityId(entityId: string): Promise<PendingScene | undefined> {
+    const cutoff = new Date(Date.now() - ttlMs)
+    const result = await database.query<PendingSceneRow>(
+      SQL`SELECT entity_id, world_name, parcels, entity, deployer, created_at, updated_at
+          FROM pending_scenes
+          WHERE entity_id = ${entityId} AND created_at >= ${cutoff}
+          LIMIT 1`
+    )
+    return result.rows.length > 0 ? toPendingScene(result.rows[0]) : undefined
+  }
+
+  async function upsert(input: UpsertPendingScene): Promise<PendingScene> {
+    const worldName = input.worldName.toLowerCase()
+    const expiryCutoff = new Date(Date.now() - ttlMs)
+
+    return await database.withAsyncContextTransaction(async () => {
+      // Serialize the "replace overlapping + insert" critical section per world (also across
+      // processes) so two concurrent uploads for the same world+parcels can't both insert.
+      await database.query(SQL`SELECT pg_advisory_xact_lock(hashtextextended(${'pending_scenes:' + worldName}, 0))`)
+
+      // The single pending slot per parcel set goes to the NEWEST scene (Decentraland deployment
+      // ordering: greater entity.timestamp, tie broken by greater entity id). Reject rather than
+      // replace when a strictly-newer overlapping upload is already in flight, so a stale/older upload
+      // can't evict a newer competitor's staged content (and two clients can't ping-pong evicting each
+      // other). A resume (same entity id) is excluded and never conflicts with itself.
+      const newer = await database.query(SQL`
+        SELECT 1 FROM pending_scenes
+        WHERE world_name = ${worldName}
+          AND parcels && ${input.parcels}::text[]
+          AND entity_id != ${input.entityId}
+          AND created_at >= ${expiryCutoff}
+          AND ( (entity->>'timestamp')::bigint > ${input.entity.timestamp}
+                OR ((entity->>'timestamp')::bigint = ${input.entity.timestamp} AND entity_id > ${input.entityId}) )
+        LIMIT 1
+      `)
+      if (newer.rowCount > 0) {
+        throw new InvalidRequestError('A newer partial upload is already in progress for one or more of these parcels.')
+      }
+
+      // Purge expired rows and replace any non-expired pending scene of this world whose parcels
+      // overlap the new one (a different entity id). Having rejected the newer-conflict above, every
+      // remaining overlapping row is strictly older, so replacing it is the intended "newest wins".
+      await database.query(SQL`
+        DELETE FROM pending_scenes
+        WHERE created_at < ${expiryCutoff}
+           OR (world_name = ${worldName} AND parcels && ${input.parcels}::text[] AND entity_id != ${input.entityId})
+      `)
+
+      const result = await database.query<PendingSceneRow>(SQL`
+        INSERT INTO pending_scenes (entity_id, world_name, parcels, entity, deployer, created_at, updated_at)
+        VALUES (${input.entityId}, ${worldName}, ${input.parcels}::text[], ${input.entity}::jsonb, ${input.deployer.toLowerCase()}, now(), now())
+        ON CONFLICT (entity_id) DO UPDATE SET updated_at = now()
+        RETURNING entity_id, world_name, parcels, entity, deployer, created_at, updated_at
+      `)
+      return toPendingScene(result.rows[0])
+    })
+  }
+
+  async function deleteByEntityId(entityId: string): Promise<void> {
+    await database.query(SQL`DELETE FROM pending_scenes WHERE entity_id = ${entityId}`)
+  }
+
+  async function countActiveByDeployer(deployer: string): Promise<number> {
+    const cutoff = new Date(Date.now() - ttlMs)
+    const result = await database.query<{ count: string }>(
+      SQL`SELECT COUNT(*) as count FROM pending_scenes
+          WHERE deployer = ${deployer.toLowerCase()} AND created_at >= ${cutoff}`
+    )
+    return parseInt(result.rows[0].count, 10)
+  }
+
+  async function deleteExpired(): Promise<number> {
+    const cutoff = new Date(Date.now() - ttlMs)
+    const result = await database.query(SQL`DELETE FROM pending_scenes WHERE created_at < ${cutoff}`)
+    const removed = result.rowCount ?? 0
+    if (removed > 0) {
+      logger.info(`Removed ${removed} expired pending scene(s)`)
+    }
+    return removed
+  }
+
+  async function getActivePendingKeys(): Promise<Set<string>> {
+    const cutoff = new Date(Date.now() - ttlMs)
+    // Project only the content hashes out of the entity JSONB instead of shipping every pending
+    // scene's full manifest: GC calls this once per delete batch, so on a large sweep the payload
+    // size matters more than the (tiny) row count. The jsonb_typeof guard keeps a row whose `content`
+    // is absent, null, or a non-array from erroring `jsonb_array_elements` ('cannot extract elements
+    // from a scalar') — one such row would otherwise fail the whole query and wedge GC server-wide.
+    const result = await database.query<{ entity_id: string; hashes: string[] | null }>(
+      SQL`SELECT entity_id,
+                 ARRAY(
+                   SELECT jsonb_array_elements(entity->'content')->>'hash'
+                   WHERE jsonb_typeof(entity->'content') = 'array'
+                 ) AS hashes
+          FROM pending_scenes
+          WHERE created_at >= ${cutoff}`
+    )
+    const keys = new Set<string>()
+    for (const row of result.rows) {
+      // The staged entity JSON, its auth-chain blob, and every content file it references are all
+      // referenced by the in-flight upload even though no world_scenes row exists yet.
+      keys.add(row.entity_id)
+      keys.add(`${row.entity_id}.auth`)
+      for (const hash of row.hashes ?? []) {
+        keys.add(hash)
+      }
+    }
+    return keys
+  }
+
+  return { getByEntityId, upsert, deleteByEntityId, deleteExpired, getActivePendingKeys, countActiveByDeployer }
+}

src/components.ts

@@ -23,6 +23,8 @@ import { createWorldsIndexerComponent } from './adapters/worlds-indexer'
 
 import { createValidator } from './logic/validations'
 import { createEntityDeployer } from './adapters/entity-deployer'
+import { createPendingScenesManager } from './adapters/pending-scenes-manager'
+import { createPartialDeploymentsComponent } from './logic/partial-deployments'
 import { createMigrationExecutor } from './adapters/migration-executor'
 import { createNameDenyListChecker } from './adapters/name-deny-list-checker'
 import { createDatabaseComponent } from './adapters/database-component'
@@ -217,6 +219,20 @@ export async function initComponents(): Promise<AppComponents> {
     worldsManager
   })
 
+  const pendingScenesManager = await createPendingScenesManager({ config, database, logs })
+
+  const partialDeployments = await createPartialDeploymentsComponent({
+    config,
+    coordinates,
+    entityDeployer,
+    limitsManager,
+    logs,
+    pendingScenesManager,
+    storage,
+    validator,
+    worldsManager
+  })
+
   const migrationExecutor = createMigrationExecutor({ logs, database: database, nameOwnership, storage, worldsManager })
 
   const notificationService = await createNotificationsClientComponent({ config, fetch, logs })
@@ -244,7 +260,7 @@ export async function initComponents(): Promise<AppComponents> {
 
   const worlds = createWorldsComponent({ worldsManager, snsClient })
 
-  const evictionJob = await createEvictionJob({ config, logs, worlds })
+  const evictionJob = await createEvictionJob({ config, logs, worlds, pendingScenesManager })
 
   const denyList = await createDenyListComponent({ config, fetch, logs })
   const bans = await createBansComponent({ config, fetch, logs })
@@ -302,6 +318,8 @@ export async function initComponents(): Promise<AppComponents> {
     nats,
     notificationService,
     participantKicker,
+    partialDeployments,
+    pendingScenesManager,
     peersRegistry,
     permissions,
     permissionsManager,

src/controllers/handlers/deploy-entity-handler.ts

@@ -1,6 +1,7 @@
 import { Entity } from '@dcl/schemas'
 import { IHttpServerComponent } from '@dcl/core-commons'
-import { FormDataContext, readUploadedFile, toDeploymentFile } from '../../logic/multipart'
+import { bufferToStream, streamToBuffer } from '@dcl/catalyst-storage'
+import { FormDataContext, toDeploymentFile } from '../../logic/multipart'
 import { DeploymentFile, HandlerContextWithPath } from '../../types'
 import { extractAuthChain } from '../../logic/extract-auth-chain'
 import { InvalidRequestError } from '@dcl/http-commons'
@@ -19,18 +20,41 @@ function parseEntityJson(raw: string) {
 }
 
 export async function deployEntity(
-  ctx: FormDataContext & HandlerContextWithPath<'config' | 'entityDeployer' | 'storage' | 'validator', '/entities'>
+  ctx: FormDataContext &
+    HandlerContextWithPath<
+      'config' | 'entityDeployer' | 'logs' | 'partialDeployments' | 'pendingScenesManager' | 'storage' | 'validator',
+      '/entities'
+    >
 ): Promise<IHttpServerComponent.IResponse> {
   const entityId = requireString(ctx.formData.fields.entityId?.value[0])
   const authChain = extractAuthChain(ctx)
 
-  const entityFile = ctx.formData.files[entityId]
+  // A `partial=true` field marks a staging request of a multi-request (partial) deployment: content may
+  // be uploaded across several requests and the world only becomes live once all of it is present.
+  const isPartial = ctx.formData.fields.partial?.value[0] === 'true'
+
+  // Resolve the entity file. It must be uploaded on the first request; a later partial (resume) request
+  // may omit it, in which case it is read back from storage where the first request stored it.
+  let entityFile: DeploymentFile | undefined = ctx.formData.files[entityId]
+    ? toDeploymentFile(ctx.formData.files[entityId])
+    : undefined
+  if (!entityFile && isPartial) {
+    const stored = await ctx.components.storage.retrieve(entityId)
+    if (stored) {
+      const buf = await streamToBuffer(await stored.asStream())
+      entityFile = { size: buf.length, getStream: () => bufferToStream(buf), asBuffer: async () => buf }
+    }
+  }
   if (!entityFile) {
-    throw new InvalidRequestError(`Entity file "${entityId}" is missing from the request.`)
+    throw new InvalidRequestError(
+      isPartial
+        ? `The first partial request for an entity must include the entity file "${entityId}".`
+        : `Entity file "${entityId}" is missing from the request.`
+    )
   }
 
   // The entity JSON is small, so it is safe to read fully into memory.
-  const entityRaw = (await readUploadedFile(entityFile)).toString()
+  const entityRaw = (await entityFile.asBuffer()).toString()
   const entityMetadataJson = parseEntityJson(entityRaw)
 
   const entity: Entity = {
@@ -42,6 +66,36 @@ export async function deployEntity(
   for (const filesKey in ctx.formData.files) {
     uploadedFiles.set(filesKey, toDeploymentFile(ctx.formData.files[filesKey]))
   }
+  // Ensure the entity file is in the map even when it came from storage (partial resume request).
+  if (!uploadedFiles.has(entityId)) {
+    uploadedFiles.set(entityId, entityFile)
+  }
+
+  const baseUrl = (await ctx.components.config.getString('HTTP_BASE_URL')) || `https://${ctx.url.host}`
+
+  if (isPartial) {
+    const result = await ctx.components.partialDeployments.stage({
+      baseUrl,
+      entity,
+      entityRaw,
+      authChain,
+      files: uploadedFiles
+    })
+    if (result.complete) {
+      return {
+        status: 200,
+        body: {
+          creationTimestamp: Date.now(),
+          ...result.result
+        }
+      }
+    }
+    return { status: 202, body: { missing: result.missing ?? [] } }
+  }
+
+  // Vanilla (single-request) deployment — behaves exactly as before, plus TTL anchoring on and cleanup
+  // of any pending upload that happens to exist for this entity.
+  const pending = await ctx.components.pendingScenesManager.getByEntityId(entityId)
 
   const contentHashesInStorage = await ctx.components.storage.existMultiple(
     Array.from(new Set((entity.content || []).map(($) => $.hash)))
@@ -52,15 +106,15 @@ export async function deployEntity(
     entity,
     files: uploadedFiles,
     authChain,
-    contentHashesInStorage
+    contentHashesInStorage,
+    pendingCreatedAt: pending?.createdAt
   })
 
   if (!validationResult.ok()) {
     throw new InvalidRequestError(`Deployment failed: ${validationResult.errors.join(', ')}`)
   }
 
   // Store the entity
-  const baseUrl = (await ctx.components.config.getString('HTTP_BASE_URL')) || `https://${ctx.url.host}`
   const message = await ctx.components.entityDeployer.deployEntity(
     baseUrl,
     entity,
@@ -70,6 +124,20 @@ export async function deployEntity(
     authChain
   )
 
+  // If this entity had a pending (partial) upload, it is now fully deployed via the vanilla path, so
+  // drop its staging row. Only when one exists (the common vanilla deploy has none), and best-effort:
+  // the deployment already committed, so a cleanup failure must not turn a successful deploy into a
+  // 5xx — the stale row would otherwise expire on its own via PENDING_DEPLOYMENT_TTL.
+  if (pending) {
+    try {
+      await ctx.components.pendingScenesManager.deleteByEntityId(entityId)
+    } catch (error) {
+      ctx.components.logs
+        .getLogger('deploy-entity')
+        .warn(`Failed to delete pending scene after a successful deploy: ${error}`, { entityId })
+    }
+  }
+
   return {
     status: 200,
     body: {