fix: make the asset-bundle cache safe across processes

The content-key lock was per-process: a second sdk-commands start on the
same project — another terminal, a stale process, CI running preview
twice — still converted concurrently and interleaved bundle files into
one cache directory, which readCache would then serve as a mix.

Conversions now stage into a private directory and publish with a single
rename. A reader sees the entry whole or not at all, and two previews
racing the same content produce identical bytes, so the loser discards
its own rather than merging into a directory it does not own.

An advisory lock keeps the second preview off work already running.
Every failure mode in it — stale lock, timeout, unwritable directory —
degrades to converting twice, which the atomic publish already makes
safe, so correctness never rests on the lock and it can stay
approximate. A killed conversion's staging is swept on the next start,
age-gated so a live conversion's directory is never taken out from
under it.

Each mechanism is mutation-checked: publishing without staging fails the
two publish tests, never inspecting a held lock's age fails the wait
test, and sweeping without the age gate fails the sweep test.

Author: eordano

packages/@dcl/sdk-commands/src/commands/start/asset-bundles.ts

@@ -30,6 +30,12 @@ export type AssetBundles = {
   invalidate(): void
 }
 
+/** Past this, a lock's owner is presumed dead and its staging is litter. */
+const STALE_LOCK_MS = 10 * 60_000
+/** Past this, converting twice beats waiting any longer on another preview. */
+const WAIT_FOR_LOCK_MS = 2 * 60_000
+const POLL_LOCK_MS = 250
+
 /** "windows" | "mac" | "linux" — the platforms abgen's export lane accepts. */
 export function hostPlatform(): string {
   return process.platform === 'win32' ? 'windows' : process.platform === 'darwin' ? 'mac' : 'linux'
@@ -63,6 +69,8 @@ export async function setupAssetBundles(
   const converted = new Map<string, ConvertedScene>()
   const platform = hostPlatform()
 
+  await sweepStaging(components, cacheRoot)
+
   // One conversion per distinct content, and only the newest publishes.
   //
   // Two separate faults lived here. A second invalidate() while a conversion
@@ -197,12 +205,98 @@ async function convertScene(
   input: { files: SceneFile[]; contentKey: string }
 ): Promise<ConvertedScene | undefined> {
   const { files, contentKey } = input
-  const cached = await readCache(components, { ...job, contentKey })
+  const full = { ...job, contentKey }
+  const cached = await readCache(components, full)
   if (cached) {
     components.logger.log(`asset-bundles: scene already converted (${cached.bundles.size} bundles, cached)`)
     return cached
   }
 
+  const lock = await acquireLock(components, full)
+  try {
+    // The lock may have been held by another preview that has since finished,
+    // so the miss above can be out of date by the time we get here.
+    const justFinished = await readCache(components, full)
+    if (justFinished) {
+      components.logger.log(
+        `asset-bundles: scene converted by another preview (${justFinished.bundles.size} bundles, cached)`
+      )
+      return justFinished
+    }
+    return await runConversion(components, abgen, full, files)
+  } finally {
+    await lock.release()
+  }
+}
+
+/**
+ * Keeps a second preview of the same project off work already in progress.
+ *
+ * Advisory only, and deliberately so: every failure mode here — a stale lock, a
+ * timeout, an unwritable directory — degrades to converting twice, which
+ * writeCache's atomic publish already makes safe. Correctness never rests on
+ * the lock, so it can be as approximate as it likes.
+ */
+async function acquireLock(
+  components: Pick<CliComponents, 'fs' | 'logger'>,
+  job: { cacheRoot: string; platform: string; contentKey: string }
+): Promise<{ release(): Promise<void> }> {
+  const lockPath = `${cacheDir(job)}.lock`
+  const noop = { release: async () => undefined }
+  const drop = {
+    release: async () => {
+      try {
+        await components.fs.unlink(lockPath)
+      } catch {
+        // Someone judged it stale and took it. Theirs to remove now.
+      }
+    }
+  }
+
+  const deadline = Date.now() + WAIT_FOR_LOCK_MS
+  for (;;) {
+    try {
+      await components.fs.mkdir(job.cacheRoot, { recursive: true })
+      await components.fs.writeFile(lockPath, `${process.pid}\n`, { flag: 'wx' })
+      return drop
+    } catch (error: any) {
+      if (error?.code !== 'EEXIST') return noop
+    }
+
+    const age = await lockAge(components, lockPath)
+    if (age === undefined) continue // released while we looked; try to take it
+    if (age > STALE_LOCK_MS) {
+      // Its owner died mid-conversion. Nothing here is transactional enough to
+      // be worth a handshake — the cost of being wrong is a second conversion.
+      try {
+        await components.fs.unlink(lockPath)
+      } catch {
+        // Lost the race to another waiter; it holds the lock now.
+      }
+      continue
+    }
+    if (Date.now() > deadline) {
+      components.logger.warn('asset-bundles: another preview is still converting this scene; converting anyway')
+      return noop
+    }
+    await new Promise((resolve) => setTimeout(resolve, POLL_LOCK_MS))
+  }
+}
+
+async function lockAge(components: Pick<CliComponents, 'fs'>, lockPath: string): Promise<number | undefined> {
+  try {
+    return Date.now() - (await components.fs.stat(lockPath)).mtimeMs
+  } catch {
+    return undefined
+  }
+}
+
+async function runConversion(
+  components: Pick<CliComponents, 'fs' | 'logger'>,
+  abgen: AbgenModule,
+  job: { projectRoot: string; entityId: string; platform: string; cacheRoot: string; contentKey: string },
+  files: SceneFile[]
+): Promise<ConvertedScene | undefined> {
   const started = Date.now()
   printProgressInfo(components.logger, 'asset-bundles: converting the scene (cached after the first run)...')
 
@@ -233,7 +327,7 @@ async function convertScene(
     manifest,
     bundles: new Map(result.bundles.map((b) => [b.name, b.data]))
   }
-  await writeCache(components, { ...job, contentKey }, scene)
+  await writeCache(components, job, scene)
   components.logger.log(`asset-bundles: scene converted (${scene.bundles.size} bundles, ${elapsed}s)`)
   return scene
 }
@@ -291,18 +385,59 @@ async function readCache(
   }
 }
 
+/**
+ * Publishes the conversion as one indivisible step.
+ *
+ * Written into a private staging directory and moved into place with a single
+ * rename, which is atomic on every filesystem this runs on. A reader therefore
+ * sees the entry complete or not at all, and two previews converting the same
+ * content race harmlessly: the loser's bytes are identical to the winner's, so
+ * it discards them rather than merging into a directory someone else owns.
+ */
 async function writeCache(
   components: Pick<CliComponents, 'fs'>,
   job: { cacheRoot: string; entityId: string; platform: string; contentKey: string },
   scene: ConvertedScene
 ): Promise<void> {
   const dir = cacheDir(job)
-  await components.fs.mkdir(path.join(dir, 'bundles'), { recursive: true })
+  const staging = `${dir}.tmp-${crypto.randomBytes(6).toString('hex')}`
+  await components.fs.mkdir(path.join(staging, 'bundles'), { recursive: true })
   for (const [name, data] of scene.bundles) {
-    await components.fs.writeFile(path.join(dir, 'bundles', name), data)
+    await components.fs.writeFile(path.join(staging, 'bundles', name), data)
+  }
+  await components.fs.writeFile(path.join(staging, 'manifest.json'), scene.manifest)
+
+  try {
+    await components.fs.rename(staging, dir)
+  } catch {
+    await components.fs.rm(staging, { recursive: true, force: true })
+  }
+}
+
+/**
+ * Clears staging directories a killed conversion left behind.
+ *
+ * Age-gated because a young one is not litter — it belongs to a preview that is
+ * converting right now, and removing it would delete that conversion's output
+ * from under it.
+ */
+async function sweepStaging(components: Pick<CliComponents, 'fs'>, cacheRoot: string): Promise<void> {
+  let entries: string[]
+  try {
+    entries = await components.fs.readdir(cacheRoot)
+  } catch {
+    return // no cache yet, nothing to sweep
+  }
+  for (const entry of entries) {
+    if (!/\.tmp-[0-9a-f]+$/.test(entry)) continue
+    const full = path.join(cacheRoot, entry)
+    try {
+      if (Date.now() - (await components.fs.stat(full)).mtimeMs < STALE_LOCK_MS) continue
+      await components.fs.rm(full, { recursive: true, force: true })
+    } catch {
+      // Being removed by whoever owns it, or not ours to remove.
+    }
   }
-  // Written last: its presence is what marks the cache complete.
-  await components.fs.writeFile(path.join(dir, 'manifest.json'), scene.manifest)
 }
 
 /** The slice of @dcl/abgen-node this command uses. */

test/sdk-commands/commands/start/asset-bundles.spec.ts

@@ -20,8 +20,12 @@ function makeComponents({ cached = false }: { cached?: boolean } = {}) {
     fileExists: jest.fn(async () => cached),
     readFile: jest.fn(async (p: string) => (p.endsWith('manifest.json') ? '{"exitCode":0}' : Buffer.from('bytes'))),
     readdir: jest.fn(async () => ['a_mac']),
-    mkdir: jest.fn(async () => undefined),
-    writeFile: jest.fn(async () => undefined)
+    mkdir: jest.fn(async (_p: string, _opts?: any) => undefined),
+    writeFile: jest.fn(async (_p: string, _data?: any, _opts?: any) => undefined),
+    rename: jest.fn(async (_from: string, _to: string) => undefined),
+    rm: jest.fn(async (_p: string, _opts?: any) => undefined),
+    unlink: jest.fn(async (_p: string) => undefined),
+    stat: jest.fn(async (_p: string) => ({ mtimeMs: Date.now() }))
   }
   const config = { getString: jest.fn(async () => undefined), requireString: jest.fn(), getNumber: jest.fn() }
   const fetch = { fetch: jest.fn() }
@@ -239,3 +243,98 @@ describe('start/asset-bundles reconversion', () => {
     expect(assetBundles!.get(hostPlatform())?.bundles.has('old_mac')).toBeFalsy()
   })
 })
+
+describe('start/asset-bundles cache publishing', () => {
+  const files = require('../../../../packages/@dcl/sdk-commands/src/logic/project-files')
+  afterEach(() => {
+    jest.clearAllMocks()
+    files.getPublishableFiles.mockResolvedValue(['scene.json', 'models/a.glb'])
+  })
+
+  const ok = {
+    code: 0,
+    bundles: [{ name: 'a_mac', data: Buffer.from('B') }],
+    events: [],
+    errors: [],
+    manifest: '{"exitCode":0}'
+  }
+
+  it('stages the conversion and publishes it with a single rename', async () => {
+    const { components, fs } = makeComponents()
+    convert.mockResolvedValue(ok)
+
+    const assetBundles = await setupAssetBundles(components, PROJECT)
+    await assetBundles!.ready
+
+    // Nothing may be written straight into the directory a reader looks at.
+    const written = fs.writeFile.mock.calls.map((c: any) => c[0])
+    const bundles = written.filter((p: string) => p.includes('bundles'))
+    expect(bundles.length).toBeGreaterThan(0)
+    for (const p of written) {
+      if (p.endsWith('.lock')) continue
+      expect(p).toContain('.tmp-')
+    }
+
+    const [from, to] = fs.rename.mock.calls[0]
+    expect(from).toContain('.tmp-')
+    expect(to).not.toContain('.tmp-')
+    expect(to.endsWith(`_${hostPlatform()}`)).toBe(true)
+  })
+
+  it('discards its own output when another preview published first', async () => {
+    const { components, fs } = makeComponents()
+    convert.mockResolvedValue(ok)
+    fs.rename.mockRejectedValue(Object.assign(new Error('not empty'), { code: 'ENOTEMPTY' }))
+
+    const assetBundles = await setupAssetBundles(components, PROJECT)
+
+    // The scene still resolves: the winner's bytes are ours byte for byte.
+    await expect(assetBundles!.ready).resolves.toBeDefined()
+    expect(fs.rm).toHaveBeenCalledWith(expect.stringContaining('.tmp-'), { recursive: true, force: true })
+  })
+
+  it('takes over a lock whose owner died', async () => {
+    const { components, fs } = makeComponents()
+    convert.mockResolvedValue(ok)
+    fs.writeFile.mockRejectedValueOnce(Object.assign(new Error('exists'), { code: 'EEXIST' }))
+    fs.stat.mockResolvedValueOnce({ mtimeMs: Date.now() - 60 * 60_000 })
+
+    const assetBundles = await setupAssetBundles(components, PROJECT)
+    await assetBundles!.ready
+
+    expect(fs.unlink).toHaveBeenCalledWith(expect.stringContaining('.lock'))
+    expect(convert).toHaveBeenCalledTimes(1)
+  })
+
+  it('uses what a live lock holder produced rather than converting again', async () => {
+    const { components, fs } = makeComponents()
+    convert.mockResolvedValue(ok)
+    fs.writeFile.mockRejectedValueOnce(Object.assign(new Error('exists'), { code: 'EEXIST' }))
+    // Absent for the pre-lock miss, present once the holder has published.
+    fs.fileExists.mockResolvedValueOnce(false).mockResolvedValue(true)
+
+    const assetBundles = await setupAssetBundles(components, PROJECT)
+
+    await expect(assetBundles!.ready).resolves.toBeDefined()
+    expect(convert).not.toHaveBeenCalled()
+    // Held and fresh, so it was waited on. Without this the test passes even
+    // if the lock is ignored outright, since the re-read alone finds the entry.
+    expect(fs.stat).toHaveBeenCalledWith(expect.stringContaining('.lock'))
+  })
+
+  it('sweeps abandoned staging directories but leaves live ones alone', async () => {
+    const { components, fs } = makeComponents()
+    convert.mockResolvedValue(ok)
+    fs.readdir.mockResolvedValue(['abc_mac', 'abc_mac.tmp-0011aabb', 'abc_mac.tmp-ffee2233'])
+    fs.stat.mockImplementation(async (p: string) => ({
+      mtimeMs: p.endsWith('0011aabb') ? Date.now() - 60 * 60_000 : Date.now()
+    }))
+
+    await setupAssetBundles(components, PROJECT)
+
+    const swept = fs.rm.mock.calls.map((c: any) => c[0])
+    expect(swept).toContain(`${PROJECT}/.dcl-optimized-assets/abc_mac.tmp-0011aabb`)
+    expect(swept).not.toContain(`${PROJECT}/.dcl-optimized-assets/abc_mac.tmp-ffee2233`)
+    expect(swept).not.toContain(`${PROJECT}/.dcl-optimized-assets/abc_mac`)
+  })
+})