fix memory leaks and add godot restart between entities

- use streaming uploads for S3 instead of loading files into memory
- add HTTP/HTTPS agents with connection pooling limits
- consume response bodies on error to free connections
- clean up file streams in finally blocks
- restart godot after each entity to prevent memory accumulation
- add better error logging for failed asset processing
- handle entities without GLTF as success with 0 assets

Author: kuruk-mm

.env.default

@@ -19,4 +19,9 @@ FETCH_MAX_RETRIES=3
 FETCH_INITIAL_DELAY_MS=1000
 FETCH_MAX_DELAY_MS=30000
 FETCH_TIMEOUT_MS=60000
-FETCH_BACKOFF_MULTIPLIER=2
+FETCH_BACKOFF_MULTIPLIER=2
+
+# Asset server configuration
+ASSET_SERVER_URL=http://localhost:8080
+ASSET_SERVER_TIMEOUT_MS=600000
+ASSET_SERVER_CONCURRENT_BUNDLES=4

package-lock.json

@@ -26,6 +26,7 @@
     "@aws-sdk/client-s3": "^3.726.1",
     "@aws-sdk/client-sns": "^3.699.0",
     "@aws-sdk/client-sqs": "^3.699.0",
+    "@aws-sdk/lib-storage": "^3.726.1",
     "@dcl/schemas": "^7.4.1",
     "@well-known-components/env-config-provider": "^1.1.1",
     "@well-known-components/http-server": "^2.0.0",

package.json

@@ -1,5 +1,9 @@
 import { IFetchComponent } from '@well-known-components/http-server'
 import { ILoggerComponent } from '@well-known-components/interfaces'
+import { exec } from 'child_process'
+import { promisify } from 'util'
+
+const execAsync = promisify(exec)
 
 export type AssetType = 'scene' | 'wearable' | 'emote' | 'texture'
 
@@ -9,6 +13,7 @@ export interface IAssetServerComponent {
   processAssets(params: ProcessAssetsParams): Promise<ProcessAssetsResponse>
   getBatchStatus(batchId: string): Promise<BatchStatus>
   waitForCompletion(batchId: string, timeoutMs?: number): Promise<BatchStatus>
+  restartGodot(): Promise<boolean>
 }
 
 export type ProcessSceneParams = {
@@ -85,6 +90,8 @@ export function createAssetServerComponent(
   async function isReady(): Promise<boolean> {
     try {
       const response = await fetch.fetch(`${baseUrl}/health`)
+      // Consume response body to free the connection
+      await response.text().catch(() => {})
       return response.ok
     } catch {
       return false
@@ -173,17 +180,65 @@ export function createAssetServerComponent(
         return status
       }
 
-      await new Promise((resolve) => setTimeout(resolve, pollIntervalMs))
+      // Use a cancellable sleep pattern
+      await new Promise<void>((resolve) => {
+        const timeoutId = setTimeout(() => resolve(), pollIntervalMs)
+        // Ensure timeout is unreferenced so it doesn't keep the process alive
+        if (timeoutId.unref) {
+          timeoutId.unref()
+        }
+      })
     }
 
     throw new Error(`Timeout waiting for batch ${batchId} to complete after ${timeoutMs}ms`)
   }
 
+  async function restartGodot(): Promise<boolean> {
+    const port = process.env.ASSET_SERVER_PORT || '8080'
+    logger.info('Restarting Godot asset-server...')
+
+    try {
+      // Kill existing Godot process - use simpler pattern
+      try {
+        await execAsync('pkill -9 -f "decentraland.godot.client" || true')
+        logger.info('Sent kill signal to Godot process')
+      } catch {
+        // Ignore errors - process might not exist
+      }
+
+      // Wait for process to fully terminate
+      await new Promise((resolve) => setTimeout(resolve, 2000))
+
+      // Start new Godot process in background
+      const godotCmd = `/app/decentraland.godot.client.x86_64 --headless --asset-server --asset-server-port ${port}`
+      const child = exec(godotCmd)
+
+      // Detach from child process - we don't need to track it
+      child.unref()
+
+      // Wait for server to be ready (up to 60 seconds)
+      for (let i = 0; i < 60; i++) {
+        await new Promise((resolve) => setTimeout(resolve, 1000))
+        if (await isReady()) {
+          logger.info('Godot asset-server is ready after restart')
+          return true
+        }
+      }
+
+      logger.error('Godot asset-server failed to become ready after restart')
+      return false
+    } catch (error) {
+      logger.error('Failed to restart Godot', { error: error instanceof Error ? error.message : String(error) })
+      return false
+    }
+  }
+
   return {
     isReady,
     processScene,
     processAssets,
     getBatchStatus,
-    waitForCompletion
+    waitForCompletion,
+    restartGodot
   }
 }

src/adapters/asset-server.ts

@@ -1,6 +1,23 @@
 import { IFetchComponent } from '@well-known-components/http-server'
 import { IConfigComponent, ILoggerComponent } from '@well-known-components/interfaces'
 import * as nodeFetch from 'node-fetch'
+import http from 'http'
+import https from 'https'
+
+// Create HTTP agents with limited connection pooling to prevent memory leaks
+const httpAgent = new http.Agent({
+  keepAlive: true,
+  maxSockets: 10,
+  maxFreeSockets: 5,
+  timeout: 60000
+})
+
+const httpsAgent = new https.Agent({
+  keepAlive: true,
+  maxSockets: 10,
+  maxFreeSockets: 5,
+  timeout: 60000
+})
 
 // Error codes that indicate transient network failures
 const RETRYABLE_ERROR_CODES = new Set([
@@ -84,17 +101,28 @@ export async function createFetchComponent(deps?: {
         const timeoutId = setTimeout(() => controller.abort(), fetchConfig.timeoutMs)
 
         try {
-          // Merge abort signal with any existing signal
+          // Select appropriate agent based on protocol
+          const isHttps = urlString.startsWith('https')
+          const agent = isHttps ? httpsAgent : httpAgent
+
+          // Merge abort signal and agent with any existing options
           const mergedInit: nodeFetch.RequestInit = {
             ...init,
-            signal: controller.signal as nodeFetch.RequestInit['signal']
+            signal: controller.signal as nodeFetch.RequestInit['signal'],
+            agent
           }
 
           const response = await nodeFetch.default(url, mergedInit)
           clearTimeout(timeoutId)
 
           // Check for retryable HTTP status
           if (isRetryableStatus(response.status) && attempt < fetchConfig.maxRetries) {
+            // Consume and discard the response body to free up the connection
+            try {
+              await response.text()
+            } catch {
+              // Ignore errors when consuming body
+            }
             const delay = calculateDelay(attempt, fetchConfig)
             logger?.warn(
               `Retryable HTTP status ${response.status} for ${urlString}, attempt ${attempt + 1}/${fetchConfig.maxRetries + 1}, retrying in ${delay}ms`

src/adapters/fetch.ts

@@ -65,24 +65,28 @@ export function createMonitoringReporter(
       return
     }
 
+    let controller: AbortController | null = new AbortController()
+    const timeoutId = setTimeout(() => controller?.abort(), 5000)
+
     try {
       const url = `${monitoringUrl}${endpoint}`
-      const controller = new AbortController()
-      const timeoutId = setTimeout(() => controller.abort(), 5000)
 
-      await fetch.fetch(url, {
+      const response = await fetch.fetch(url, {
         method: 'POST',
         headers: { 'Content-Type': 'application/json' },
         body: JSON.stringify({ ...data, secret: monitoringSecret }),
         signal: controller.signal
       })
-
-      clearTimeout(timeoutId)
+      // Consume response body to free the connection
+      await response.text().catch(() => {})
     } catch (error) {
       // Silently ignore - monitoring should never block pipeline
       logger.debug('Monitoring report failed (non-blocking)', {
         error: error instanceof Error ? error.message : 'Unknown error'
       })
+    } finally {
+      clearTimeout(timeoutId)
+      controller = null
     }
   }
 

src/adapters/monitoring-reporter.ts

@@ -1,5 +1,7 @@
 import { IBaseComponent } from '@well-known-components/interfaces'
-import { S3Client, PutObjectCommand } from '@aws-sdk/client-s3'
+import { S3Client } from '@aws-sdk/client-s3'
+import { Upload } from '@aws-sdk/lib-storage'
+import { createReadStream } from 'fs'
 import { mkdir, readFile, writeFile } from 'fs/promises'
 import path from 'path'
 import { AppComponents } from '../types'
@@ -28,20 +30,32 @@ export async function createS3StorageComponent(
   return {
     storeFile: async function (key: string, filePath: string) {
       const keyWithPrefix = `${formattedPrefix}${key}`
+      let fileStream: ReturnType<typeof createReadStream> | null = null
 
       try {
-        const fileContent = await readFile(filePath)
-        const command = new PutObjectCommand({
-          Bucket: bucketName,
-          Key: keyWithPrefix,
-          Body: fileContent
+        // Use streaming upload to avoid loading entire file into memory
+        fileStream = createReadStream(filePath)
+
+        const upload = new Upload({
+          client: s3Client,
+          params: {
+            Bucket: bucketName,
+            Key: keyWithPrefix,
+            Body: fileStream
+          }
         })
 
-        await s3Client.send(command)
+        await upload.done()
         logger.info(`Stored file ${keyWithPrefix} in S3`)
       } catch (error) {
         logger.error(`Error storing file ${keyWithPrefix} in S3`)
         logger.error(error as any)
+      } finally {
+        // Ensure stream is closed
+        if (fileStream) {
+          fileStream.destroy()
+          fileStream = null
+        }
       }
     },
 
@@ -52,20 +66,23 @@ export async function createS3StorageComponent(
         let success = false
 
         while (attempt < 3 && !success) {
+          let fileStream: ReturnType<typeof createReadStream> | null = null
           try {
-            // Read the file content
-            const fileContent = await readFile(filePath)
-
-            // Create and send the S3 command
-            const command = new PutObjectCommand({
-              Bucket: bucketName,
-              Key: keyWithPrefix,
-              Body: fileContent
+            // Use streaming upload to avoid loading entire file into memory
+            fileStream = createReadStream(filePath)
+
+            const upload = new Upload({
+              client: s3Client,
+              params: {
+                Bucket: bucketName,
+                Key: keyWithPrefix,
+                Body: fileStream
+              }
             })
 
-            await s3Client.send(command)
+            await upload.done()
 
-            // If the command succeeds, log the success and exit the retry loop
+            // If the upload succeeds, log the success and exit the retry loop
             logger.info(`Successfully stored file ${keyWithPrefix} in S3`)
             success = true
           } catch (error) {
@@ -77,6 +94,12 @@ export async function createS3StorageComponent(
               logger.error(`Failed to store file ${keyWithPrefix} after 3 attempts`)
               logger.error(error as any)
             }
+          } finally {
+            // Ensure stream is closed
+            if (fileStream) {
+              fileStream.destroy()
+              fileStream = null
+            }
           }
         }
 

src/adapters/storage.ts

@@ -16,6 +16,7 @@ import { AwsCredentialIdentity } from '@smithy/types'
 import { createMonitoringReporter } from './adapters/monitoring-reporter'
 import { getProcessMethod } from './service'
 import { createAssetServerComponent } from './adapters/asset-server'
+import { rm } from 'fs/promises'
 
 // Helper function to convert URN with token ID to pointer without token ID
 function urnToPointer(urn: string): string {
@@ -63,6 +64,8 @@ async function handleProfile(
 
     const profileResponse = await fetch.fetch(profileUrl)
     if (!profileResponse.ok) {
+      // Consume body to free the connection
+      await profileResponse.text().catch(() => {})
       throw new Error(`Failed to fetch profile: ${profileResponse.statusText}`)
     }
 
@@ -111,6 +114,8 @@ async function handleProfile(
     })
 
     if (!entitiesResponse.ok) {
+      // Consume body to free the connection
+      await entitiesResponse.text().catch(() => {})
       throw new Error(`Failed to fetch entities: ${entitiesResponse.statusText}`)
     }
 
@@ -201,6 +206,8 @@ async function handleProfile(
           if (result.status === 'completed' && result.zip_path) {
             const s3Key = `${asset.gltfHash}-mobile.zip`
             await storage.storeFile(s3Key, result.zip_path)
+            // Clean up temp file from asset-server
+            await rm(result.zip_path, { force: true }).catch(() => {})
             logger.info(`Completed: ${asset.pointer} ${variant} -> ${s3Key}`)
             return { success: true, hash: asset.gltfHash }
           } else {
@@ -250,6 +257,8 @@ async function handleEntityId(
       })
 
       if (!response.ok) {
+        // Consume body to free the connection
+        await response.text().catch(() => {})
         throw new Error(`Failed to fetch entity ID: ${response.statusText}`)
       }
 
@@ -282,6 +291,8 @@ async function handleEntityId(
       })
 
       if (!response.ok) {
+        // Consume body to free the connection
+        await response.text().catch(() => {})
         throw new Error(`Failed to fetch world data: ${response.statusText}`)
       }