From aa47dcd3e507912fd134e7b4406cfcde3e042187 Mon Sep 17 00:00:00 2001 From: Juan Fernando Castro Date: Wed, 21 Oct 2026 21:50:54 -0400 Subject: [PATCH] feat(workers): internal teardown that takes the script down before the row, and destroy reports Cloudflare refusals (PUT-2219) (#4297) --- .../workers/WorkerDriver.cloudflare.test.ts | 576 +++++++++++++++++- .../drivers/workers/WorkerDriver.test.ts | 169 ++++- src/backend/drivers/workers/WorkerDriver.ts | 183 +++++- 3 files changed, 887 insertions(+), 41 deletions(-) diff --git a/src/backend/drivers/workers/WorkerDriver.cloudflare.test.ts b/src/backend/drivers/workers/WorkerDriver.cloudflare.test.ts index 6ddc57c80..aff7a8ebb 100644 --- a/src/backend/drivers/workers/WorkerDriver.cloudflare.test.ts +++ b/src/backend/drivers/workers/WorkerDriver.cloudflare.test.ts @@ -17,6 +17,7 @@ * along with this program. If not, see . */ + /** * WorkerDriver with a deploy backend configured. * @@ -47,7 +48,9 @@ import { PuterServer } from '../../server.js'; import type { FSEntry } from '../../stores/fs/FSEntry.js'; import { setupTestServer } from '../../testUtil.js'; import { generateDefaultFsentries } from '../../util/userProvisioning.js'; +import { resolveCallableMethods } from '../meta.js'; import { + INTERNAL_ADMIN_DESTROY, INTERNAL_ADMISSION_BYPASS, type WorkerDriver, } from './WorkerDriver.js'; @@ -60,8 +63,12 @@ let server: PuterServer; let target: WorkerDriver; let fetchSpy: MockInstance; -const edgeResponse = (body: unknown) => - ({ json: async () => body }) as unknown as Response; +const edgeResponse = (body: unknown, status = 200) => + ({ + ok: status >= 200 && status < 300, + status, + json: async () => body, + }) as unknown as Response; beforeAll(async () => { server = await setupTestServer({ @@ -159,6 +166,30 @@ const deleteCalls = () => (call) => (call[1] as RequestInit | undefined)?.method === 'DELETE', ); +const deployedToken = () => { + const [, init] = putCalls().at(-1)!; + const form = (init as RequestInit).body as FormData; + const metadata = JSON.parse(form.get('metadata') as string); + return metadata.bindings.find( + (b: { name: string }) => b.name === 'puter_auth', + ).text as string; +}; + +/** A fresh user with one deployed worker and the token it was deployed with. */ +const deployWorker = async (label: string) => { + const { user, actor } = await makeUser(); + const path = `/${user.username}/${label}.js`; + const entry = await writeSource(actor, user.id, path, 'src'); + const name = `${label}-${user.username}`; + await inCtx(actor, () => + target.create({ appId: '', workerName: name, filePath: path }), + ); + return { user, actor, entry, name, token: deployedToken() }; +}; + +const workerSession = (userId: number, workerName: string) => + server.stores.session.getWorker(userId, { appUid: null, workerName }); + // -- create ---------------------------------------------------------- describe('WorkerDriver.create with a configured deploy backend', () => { @@ -596,17 +627,220 @@ describe('WorkerDriver.destroy with a configured deploy backend', () => { target.destroy({ workerName: name }), ); - expect(result).toEqual({ success: true }); + expect(result).toEqual({ + success: true, + errors: [], + messages: [], + result: { id: name }, + }); const [url, init] = deleteCalls()[0]!; expect(url).toBe(`${SCRIPTS_BASE}/${name}/`); expect((init as RequestInit).headers).toEqual({ Authorization: `Bearer ${AUTH_KEY}`, }); + expect((init as RequestInit).signal).toBeInstanceOf(AbortSignal); + expect( + await server.stores.subdomain.getBySubdomain( + `workers.puter.${name}`, + ), + ).toBeFalsy(); + }); + + it('reports a refused edge delete as 502 and keeps the row and the worker session', async () => { + const { user, actor, name, token } = await deployWorker('refused'); + + fetchSpy.mockResolvedValueOnce( + edgeResponse( + { + success: false, + errors: [{ code: 10000, message: 'Internal error' }], + }, + 500, + ), + ); + await expect( + inCtx(actor, () => target.destroy({ workerName: name })), + ).rejects.toMatchObject({ + statusCode: 502, + legacyCode: 'worker_takedown_failed', + message: + 'Cloudflare refused to remove the script; nothing was deleted.', + fields: { status: 500 }, + }); + + expect( + await server.stores.subdomain.getBySubdomain( + `workers.puter.${name}`, + ), + ).toBeTruthy(); + expect(await workerSession(user.id, name)).toBeTruthy(); + expect( + (await server.services.auth.authenticate(token)).actor, + ).toBeTruthy(); + }); + + it('keeps the row when the edge answers 200 with success:false', async () => { + const { actor, name } = await deployWorker('softfail'); + + fetchSpy.mockResolvedValueOnce( + edgeResponse({ success: false, errors: [{ code: 10001 }] }), + ); + await expect( + inCtx(actor, () => target.destroy({ workerName: name })), + ).rejects.toMatchObject({ statusCode: 502 }); + + expect( + await server.stores.subdomain.getBySubdomain( + `workers.puter.${name}`, + ), + ).toBeTruthy(); + }); + + it('reports an edge delete that times out as 504 and keeps the row', async () => { + const { actor, name } = await deployWorker('slow'); + + fetchSpy.mockRejectedValueOnce( + new DOMException('The operation was aborted', 'TimeoutError'), + ); + await expect( + inCtx(actor, () => target.destroy({ workerName: name })), + ).rejects.toMatchObject({ + statusCode: 504, + legacyCode: 'worker_takedown_timeout', + }); + + expect( + await server.stores.subdomain.getBySubdomain( + `workers.puter.${name}`, + ), + ).toBeTruthy(); + }); + + it('reports a transport failure as 504 and keeps the row', async () => { + const { actor, name } = await deployWorker('unreachable'); + + fetchSpy.mockRejectedValueOnce(new TypeError('fetch failed')); + await expect( + inCtx(actor, () => target.destroy({ workerName: name })), + ).rejects.toMatchObject({ + statusCode: 504, + legacyCode: 'worker_takedown_timeout', + }); + + expect( + await server.stores.subdomain.getBySubdomain( + `workers.puter.${name}`, + ), + ).toBeTruthy(); + }); + + it('lets any other failure before the edge answers surface as itself', async () => { + const { actor, name } = await deployWorker('ownerr'); + + fetchSpy.mockRejectedValueOnce(new Error('config unavailable')); + const err = (await inCtx(actor, () => + target.destroy({ workerName: name }), + ).catch((e: unknown) => e)) as Error & { statusCode?: number }; + expect(err.message).toBe('config unavailable'); + expect(err.statusCode).toBeUndefined(); + + expect( + await server.stores.subdomain.getBySubdomain( + `workers.puter.${name}`, + ), + ).toBeTruthy(); + }); + + it('tears down in order: script, then sessions, then row', async () => { + const { user, actor, name } = await deployWorker('order'); + const store = server.stores.subdomain; + let seenAtRowDelete: { edgeDeletes: number; sessionGone: boolean } = { + edgeDeletes: -1, + sessionGone: false, + }; + const deleteSpy = vi + .spyOn(store, 'deleteByUuid') + .mockImplementation(async (...args) => { + seenAtRowDelete = { + edgeDeletes: deleteCalls().length, + sessionGone: !(await workerSession(user.id, name)), + }; + deleteSpy.mockRestore(); + return store.deleteByUuid(...args); + }); + + await inCtx(actor, () => target.destroy({ workerName: name })); + + expect(seenAtRowDelete).toEqual({ edgeDeletes: 1, sessionGone: true }); + }); + + it('keeps the row when the sessions cannot be dropped, and a retry finishes the job', async () => { + const { actor, name } = await deployWorker('retry'); + vi.spyOn( + server.services.auth, + 'deleteWorkerSessionsByName', + ).mockRejectedValueOnce(new Error('session store unavailable')); + + await expect( + inCtx(actor, () => target.destroy({ workerName: name })), + ).rejects.toThrow('session store unavailable'); + expect(deleteCalls()).toHaveLength(1); + expect( + await server.stores.subdomain.getBySubdomain( + `workers.puter.${name}`, + ), + ).toBeTruthy(); + + const retried = await inCtx(actor, () => + target.destroy({ workerName: name }), + ); + expect(retried).toEqual({ + success: true, + errors: [], + messages: [], + result: { id: name }, + }); + expect(deleteCalls()).toHaveLength(2); + expect( + await server.stores.subdomain.getBySubdomain( + `workers.puter.${name}`, + ), + ).toBeFalsy(); + }); + + it('treats a script the edge no longer has (code 10007) as taken down', async () => { + const { user, actor, name } = await deployWorker('gone10007'); + + fetchSpy.mockResolvedValueOnce( + edgeResponse( + { + success: false, + errors: [ + { + code: 10007, + message: 'workers.api.error.script_not_found', + }, + ], + }, + 404, + ), + ); + const result = await inCtx(actor, () => + target.destroy({ workerName: name }), + ); + + expect(result).toEqual({ + success: true, + errors: [], + messages: [], + result: { id: name }, + }); expect( await server.stores.subdomain.getBySubdomain( `workers.puter.${name}`, ), ).toBeFalsy(); + expect(await workerSession(user.id, name)).toBeNull(); }); it('lowercases the requested name before looking the worker up', async () => { @@ -630,6 +864,319 @@ describe('WorkerDriver.destroy with a configured deploy backend', () => { }); }); +// -- operator teardown ------------------------------------------------ + +describe('WorkerDriver operator teardown (INTERNAL_ADMIN_DESTROY)', () => { + const rowOf = (name: string) => + server.stores.subdomain.getBySubdomain(`workers.puter.${name}`, { + primary: true, + }); + + /** Collects every `subdomain.delete` announced while `fn` runs. */ + const announced = async (fn: () => Promise) => { + const seen: unknown[] = []; + const handler = (_key: string, data: unknown) => { + seen.push(data); + }; + server.clients.event.on('subdomain.delete', handler as never); + try { + return { result: await fn(), seen }; + } finally { + server.clients.event.off('subdomain.delete', handler as never); + } + }; + + it('takes the script down before the row, drops the session, and announces the delete once', async () => { + const { user, name, token } = await deployWorker('op'); + const row = (await rowOf(name))!; + const store = server.stores.subdomain; + let seenAtRowDelete = { edgeDeletes: -1, sessionGone: false }; + const deleteSpy = vi + .spyOn(store, 'deleteByUuid') + .mockImplementation(async (...args) => { + seenAtRowDelete = { + edgeDeletes: deleteCalls().length, + sessionGone: !(await workerSession(user.id, name)), + }; + deleteSpy.mockRestore(); + return store.deleteByUuid(...args); + }); + + const { result, seen } = await announced(() => + target[INTERNAL_ADMIN_DESTROY]({ uuid: row.uuid }), + ); + + expect(result).toEqual({ + subdomain: `workers.puter.${name}`, + userId: user.id, + workerName: name, + }); + const [url] = deleteCalls()[0]!; + expect(url).toBe(`${SCRIPTS_BASE}/${name}/`); + expect(seenAtRowDelete).toEqual({ edgeDeletes: 1, sessionGone: true }); + expect(await rowOf(name)).toBeNull(); + expect(await workerSession(user.id, name)).toBeNull(); + expect(await server.services.auth.authenticate(token)).toEqual({ + invalid: true, + }); + expect(seen).toEqual([ + { subdomain: `workers.puter.${name}`, uid: row.uuid }, + ]); + + // The driver's own listener hears this and leaves the script alone. + await new Promise((r) => setTimeout(r, 50)); + expect(deleteCalls()).toHaveLength(1); + }); + + it('reports a refused edge delete as 502 and keeps the row and session', async () => { + const { user, name, token } = await deployWorker('oprefused'); + const row = (await rowOf(name))!; + + fetchSpy.mockResolvedValueOnce( + edgeResponse( + { + success: false, + errors: [{ code: 10000, message: 'edge-only detail' }], + }, + 500, + ), + ); + const { seen } = await announced(async () => { + const err = (await target[INTERNAL_ADMIN_DESTROY]({ + uuid: row.uuid, + }).catch((e: unknown) => e)) as { + statusCode: number; + legacyCode: string; + message: string; + fields: unknown; + }; + expect(err).toMatchObject({ + statusCode: 502, + legacyCode: 'worker_takedown_failed', + message: + 'Cloudflare refused to remove the script; nothing was deleted.', + fields: { status: 500 }, + }); + // Cloudflare's own text stays out of what the caller sees. + expect(JSON.stringify([err.message, err.fields])).not.toContain( + 'edge-only detail', + ); + }); + + expect(await rowOf(name)).toMatchObject({ uuid: row.uuid }); + expect(await workerSession(user.id, name)).toBeTruthy(); + expect( + (await server.services.auth.authenticate(token)).actor, + ).toBeTruthy(); + expect(seen).toEqual([]); + }); + + it('keeps the row when the edge answers 200 with success:false', async () => { + const { name } = await deployWorker('opsoft'); + const row = (await rowOf(name))!; + + fetchSpy.mockResolvedValueOnce( + edgeResponse({ success: false, errors: [] }), + ); + await expect( + target[INTERNAL_ADMIN_DESTROY]({ uuid: row.uuid }), + ).rejects.toMatchObject({ statusCode: 502 }); + + expect(await rowOf(name)).toMatchObject({ uuid: row.uuid }); + }); + + it('treats script-not-found (code 10007) as taken down', async () => { + const { name } = await deployWorker('op10007'); + const row = (await rowOf(name))!; + + fetchSpy.mockResolvedValueOnce( + edgeResponse( + { + success: false, + errors: [ + { + code: 10007, + message: 'workers.api.error.script_not_found', + }, + ], + }, + 404, + ), + ); + await expect( + target[INTERNAL_ADMIN_DESTROY]({ uuid: row.uuid }), + ).resolves.toMatchObject({ workerName: name }); + + expect(await rowOf(name)).toBeNull(); + }); + + it('treats a bare 404 with no usable body as a refusal and keeps the row', async () => { + const { name } = await deployWorker('op404'); + const row = (await rowOf(name))!; + + fetchSpy.mockResolvedValueOnce({ + ok: false, + status: 404, + json: async () => { + throw new SyntaxError('not json'); + }, + } as unknown as Response); + await expect( + target[INTERNAL_ADMIN_DESTROY]({ uuid: row.uuid }), + ).rejects.toMatchObject({ statusCode: 502, fields: { status: 404 } }); + + expect(await rowOf(name)).toMatchObject({ uuid: row.uuid }); + }); + + it('treats a 404 that could not be routed (code 7003) as a refusal and keeps the row', async () => { + const { name } = await deployWorker('op7003'); + const row = (await rowOf(name))!; + + fetchSpy.mockResolvedValueOnce( + edgeResponse( + { + success: false, + errors: [ + { + code: 7003, + message: 'Could not route to /accounts/x/workers', + }, + ], + }, + 404, + ), + ); + await expect( + target[INTERNAL_ADMIN_DESTROY]({ uuid: row.uuid }), + ).rejects.toMatchObject({ + statusCode: 502, + legacyCode: 'worker_takedown_failed', + }); + + expect(await rowOf(name)).toMatchObject({ uuid: row.uuid }); + }); + + it('reports a timed-out edge delete as 504 with no edge text, and keeps the row', async () => { + const { name } = await deployWorker('opslow'); + const row = (await rowOf(name))!; + + fetchSpy.mockRejectedValueOnce( + new DOMException('edge-only timeout detail', 'TimeoutError'), + ); + const err = (await target[INTERNAL_ADMIN_DESTROY]({ + uuid: row.uuid, + }).catch((e: unknown) => e)) as { + message: string; + fields: unknown; + }; + expect(err).toMatchObject({ + statusCode: 504, + legacyCode: 'worker_takedown_timeout', + }); + expect(JSON.stringify([err.message, err.fields ?? null])).not.toContain( + 'edge-only', + ); + + expect(await rowOf(name)).toMatchObject({ uuid: row.uuid }); + }); + + it('keeps the row when the sessions cannot be dropped, and a retry finishes the job', async () => { + const { name } = await deployWorker('opretry'); + const row = (await rowOf(name))!; + vi.spyOn( + server.services.auth, + 'deleteWorkerSessionsByName', + ).mockRejectedValueOnce(new Error('session store unavailable')); + + const { seen } = await announced(async () => + expect( + target[INTERNAL_ADMIN_DESTROY]({ uuid: row.uuid }), + ).rejects.toThrow('session store unavailable'), + ); + expect(deleteCalls()).toHaveLength(1); + expect(await rowOf(name)).toMatchObject({ uuid: row.uuid }); + expect(seen).toEqual([]); + + await target[INTERNAL_ADMIN_DESTROY]({ uuid: row.uuid }); + expect(deleteCalls()).toHaveLength(2); + expect(await rowOf(name)).toBeNull(); + }); + + it('skips only its own announcement: a re-created worker of the same name is still taken down', async () => { + const { user, actor, name } = await deployWorker('opsame'); + const first = (await rowOf(name))!; + await target[INTERNAL_ADMIN_DESTROY]({ uuid: first.uuid }); + await new Promise((r) => setTimeout(r, 50)); + expect(deleteCalls()).toHaveLength(1); + + // Same name, new row: a delete announced for it is not the one above. + await inCtx(actor, () => + target.create({ + appId: '', + workerName: name, + filePath: `/${user.username}/opsame.js`, + }), + ); + const second = (await rowOf(name))!; + expect(second.uuid).not.toBe(first.uuid); + await server.stores.subdomain.deleteByUuid(second.uuid); + server.clients.event.emit( + 'subdomain.delete', + { subdomain: `workers.puter.${name}`, uid: second.uuid }, + {}, + ); + + await waitFor( + async () => deleteCalls().length === 2, + 'edge delete for the re-created worker', + ); + }); + + it('returns 404 for a row that is not a worker, and leaves it alone', async () => { + const { user } = await makeUser(); + const site = await server.stores.subdomain.create({ + userId: user.id, + subdomain: `site-${user.username}`, + rootDirId: null, + associatedAppId: null, + }); + + await expect( + target[INTERNAL_ADMIN_DESTROY]({ uuid: site.uuid }), + ).rejects.toMatchObject({ statusCode: 404, legacyCode: 'not_found' }); + + expect(deleteCalls()).toEqual([]); + expect( + await server.stores.subdomain.getByUuid(site.uuid, { + primary: true, + }), + ).toBeTruthy(); + }); + + it('returns 404 for an unknown or missing uuid', async () => { + await expect( + target[INTERNAL_ADMIN_DESTROY]({ uuid: uuidv4() }), + ).rejects.toMatchObject({ statusCode: 404 }); + await expect( + target[INTERNAL_ADMIN_DESTROY]({} as { uuid: string }), + ).rejects.toMatchObject({ statusCode: 404 }); + expect(deleteCalls()).toEqual([]); + }); + + it('is not on the /drivers/call surface', () => { + const callable = resolveCallableMethods(target); + expect([...callable].sort()).toEqual([ + 'create', + 'destroy', + 'getFilePaths', + 'getLoggingUrl', + ]); + expect(callable.has(String(INTERNAL_ADMIN_DESTROY.description))).toBe( + false, + ); + }); +}); + // -- worker rows deleted elsewhere ------------------------------------ describe('WorkerDriver on subdomain.delete', () => { @@ -1027,29 +1574,6 @@ describe('WorkerDriver hot reload', () => { // -- worker sessions -------------------------------------------------- describe('WorkerDriver keeps a worker`s token tied to the worker', () => { - const deployedToken = () => { - const [, init] = putCalls().at(-1)!; - const form = (init as RequestInit).body as FormData; - const metadata = JSON.parse(form.get('metadata') as string); - return metadata.bindings.find( - (b: { name: string }) => b.name === 'puter_auth', - ).text as string; - }; - - const deployWorker = async (label: string) => { - const { user, actor } = await makeUser(); - const path = `/${user.username}/${label}.js`; - const entry = await writeSource(actor, user.id, path, 'src'); - const name = `${label}-${user.username}`; - await inCtx(actor, () => - target.create({ appId: '', workerName: name, filePath: path }), - ); - return { user, actor, entry, name, token: deployedToken() }; - }; - - const workerSession = (userId: number, workerName: string) => - server.stores.session.getWorker(userId, { appUid: null, workerName }); - it('binds the deployed token to the worker row, and a redeploy keeps it', async () => { const { user, actor, name, token } = await deployWorker('bind'); const row = await server.stores.subdomain.getBySubdomain( diff --git a/src/backend/drivers/workers/WorkerDriver.test.ts b/src/backend/drivers/workers/WorkerDriver.test.ts index 722ed4484..cabbd7599 100644 --- a/src/backend/drivers/workers/WorkerDriver.test.ts +++ b/src/backend/drivers/workers/WorkerDriver.test.ts @@ -1,4 +1,4 @@ -// This test checks everything in WorkerDriver up to the cloudflare level. +// This test checks everything in WorkerDriver up to the cloudflare level. // Full deployments are not tested as this would require Cloudflare Workerd, // however the interactions with the Puter API are import { @@ -13,10 +13,18 @@ import { import { Actor, makeActor as resolveActor } from '../../core/actor.js'; import { runWithContext } from '../../core/context.js'; import { PuterServer } from '../../server.js'; -import { setupTestServer } from '../../testUtil.js'; +import { + setupPuterTestEnv, + setupTestServer, + type PuterTestEnv, +} from '../../testUtil.js'; import { decodeCursor } from '../../util/pagination.js'; import { generateDefaultFsentries } from '../../util/userProvisioning.js'; -import { INTERNAL_ADMISSION_BYPASS } from './WorkerDriver.js'; +import { resolveCallableMethods } from '../meta.js'; +import { + INTERNAL_ADMIN_DESTROY, + INTERNAL_ADMISSION_BYPASS, +} from './WorkerDriver.js'; import type { WorkerDriver } from './WorkerDriver.js'; describe('WorkerDriver', () => { @@ -252,7 +260,10 @@ describe('WorkerDriver', () => { it('lets an app bind a worker to an app it created', async () => { const { generated, builderActor } = await seedApps(); await expect( - inCtx(() => deploy(generated.uid, 'wk-sandboxed'), builderActor), + inCtx( + () => deploy(generated.uid, 'wk-sandboxed'), + builderActor, + ), ).rejects.toMatchObject({ statusCode: 503 }); }); @@ -266,14 +277,20 @@ describe('WorkerDriver', () => { it('rejects an app binding a worker to an app it did not create', async () => { const { unrelated, builderActor } = await seedApps(); await expect( - inCtx(() => deploy(unrelated.uid, 'wk-unrelated'), builderActor), + inCtx( + () => deploy(unrelated.uid, 'wk-unrelated'), + builderActor, + ), ).rejects.toMatchObject({ statusCode: 403 }); }); it('rejects an app binding a worker to an unknown app', async () => { const { builderActor } = await seedApps(); await expect( - inCtx(() => deploy('app-does-not-exist', 'wk-ghost'), builderActor), + inCtx( + () => deploy('app-does-not-exist', 'wk-ghost'), + builderActor, + ), ).rejects.toMatchObject({ statusCode: 403 }); }); @@ -906,7 +923,9 @@ describe('WorkerDriver', () => { try { for (const body of forgeries) { const forged = JSON.parse(body) as Record; - expect(Object.getOwnPropertySymbols(forged)).toHaveLength(0); + expect(Object.getOwnPropertySymbols(forged)).toHaveLength( + 0, + ); await expect( runWithContext({ actor: unverified() }, () => strictDriver.create({ @@ -927,3 +946,139 @@ describe('WorkerDriver', () => { }); }); }); + +// Boots local workerd mode, which sticks for the rest of this file: runs last. +describe('WorkerDriver operator teardown on local workerd', () => { + let env: PuterTestEnv; + let driver: WorkerDriver; + + beforeAll(async () => { + env = await setupPuterTestEnv({ + workers: { localServer: true }, + } as never); + driver = env.server.drivers.workers as unknown as WorkerDriver; + }, 60_000); + + afterAll(async () => { + await env?.shutdown(); + }); + + const seedWorker = async (label: string) => { + const owner = await env.server.stores.user.create({ + username: `wklocal-${label}`, + uuid: `wklocal-uuid-${label}`, + password: null, + email: null, + }); + const row = await env.server.stores.subdomain.create({ + userId: owner.id, + subdomain: `workers.puter.${label}`, + rootDirId: null, + associatedAppId: null, + }); + return { owner, row }; + }; + + /** A worker deployed for real through the driver, so Miniflare holds it. */ + const deployLocal = async (label: string) => { + const owner = await env.server.stores.user.getByUsername( + env.users.user.username, + ); + const ownerActor = resolveActor({ + user: { + uuid: owner!.uuid, + id: owner!.id, + username: owner!.username, + email: `${owner!.username}@test.com`, + email_confirmed: true, + }, + }); + const path = `/${owner!.username}/${label}.js`; + const source = `router.get('/', () => new Response('${label}'));`; + await runWithContext({ actor: ownerActor }, () => + env.server.services.fs.write(owner!.id, { + fileMetadata: { + path, + size: Buffer.byteLength(source), + contentType: 'application/javascript', + overwrite: true, + }, + fileContent: Buffer.from(source), + }), + ); + const deployed = (await runWithContext({ actor: ownerActor }, () => + driver.create({ appId: '', workerName: label, filePath: path }), + )) as { success?: boolean }; + expect(deployed.success).toBe(true); + const row = await env.server.stores.subdomain.getBySubdomain( + `workers.puter.${label}`, + { primary: true }, + ); + return { owner: owner!, row: row! }; + }; + + it('takes the local script down, drops the row, and reports the owner', async () => { + const { owner, row } = await deployLocal('torn'); + const local = env.server.services.localworkerservice; + const deleteLocal = vi.spyOn(local, 'cfDeleteLocal'); + const served = await local.cfCallLocal( + 'torn', + new Request('http://worker.localhost/'), + ); + expect(await served.text()).toBe('torn'); + + const result = await driver[INTERNAL_ADMIN_DESTROY]({ + uuid: row.uuid, + }); + + expect(deleteLocal).toHaveBeenCalledWith('torn'); + expect(result).toEqual({ + subdomain: 'workers.puter.torn', + userId: owner.id, + workerName: 'torn', + }); + expect( + await env.server.stores.subdomain.getByUuid(row.uuid, { + primary: true, + }), + ).toBeNull(); + // The Miniflare instance is disposed with the row: nothing answers now. + const after = await local.cfCallLocal( + 'torn', + new Request('http://worker.localhost/'), + ); + expect(after.status).toBe(404); + }, 60_000); + + it('is absent from the callable surface and unreachable over /drivers/call', async () => { + const { row } = await seedWorker('kept'); + expect(resolveCallableMethods(driver).has('destroy')).toBe(true); + + for (const method of [ + String(INTERNAL_ADMIN_DESTROY.description), + 'INTERNAL_ADMIN_DESTROY', + `Symbol(${INTERNAL_ADMIN_DESTROY.description})`, + ]) { + expect(resolveCallableMethods(driver).has(method)).toBe(false); + const res = await fetch(new URL('/drivers/call', env.apiOrigin), { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + Authorization: `Bearer ${env.users.user.token}`, + Origin: env.apiOrigin, + }, + body: JSON.stringify({ + interface: 'workers', + method, + args: { uuid: row.uuid }, + }), + }); + expect(res.status, method).toBe(404); + } + expect( + await env.server.stores.subdomain.getByUuid(row.uuid, { + primary: true, + }), + ).toBeTruthy(); + }); +}); diff --git a/src/backend/drivers/workers/WorkerDriver.ts b/src/backend/drivers/workers/WorkerDriver.ts index a0339b743..decc0307d 100644 --- a/src/backend/drivers/workers/WorkerDriver.ts +++ b/src/backend/drivers/workers/WorkerDriver.ts @@ -40,6 +40,7 @@ import { } from '../../services/metering/consts.js'; import type { DriverConcurrentConfig, DriverRateLimitConfig } from '../meta.js'; import { PuterDriver } from '../types.js'; +import { BoundedTtlMap } from '../../util/boundedTtlMap.js'; import { awaitLater } from '../../util/concurrency.js'; import { isUniqueViolation } from '../../util/dbError.js'; import { loadFileInput } from '../util/fileInput.js'; @@ -92,6 +93,58 @@ export const INTERNAL_ADMISSION_BYPASS = Symbol( */ export const INTERNAL_DEPLOY_TARGET = Symbol('workers.internalDeployTarget'); +/** Operator teardown key; a symbol keeps it off the `/drivers/call` surface. */ +export const INTERNAL_ADMIN_DESTROY = Symbol('workers.internalAdminDestroy'); + +/** What the operator teardown reports back, for the caller's audit trail. */ +export interface AdminDestroyResult { + subdomain: string; + userId: number; + workerName: string; +} + +/** The edge refused a script delete: neither success nor "no such script". */ +export class WorkerTakedownRefusedError extends Error { + constructor( + readonly status: number, + readonly detail: string | undefined, + ) { + super( + `Cloudflare refused the script delete (status ${status})${ + detail ? `: ${detail}` : '' + }`, + ); + this.name = 'WorkerTakedownRefusedError'; + } +} + +const CF_DELETE_TIMEOUT_MS = 15_000; +// Cloudflare Workers API code 10007: Worker or workers.dev subdomain not found. +const CF_SCRIPT_NOT_FOUND_CODE = 10007; + +// What `fetch` rejects with when the edge never answered: timeout, transport. +const isEdgeTransportError = (err: unknown): boolean => + err instanceof TypeError || + (err instanceof Error && + (err.name === 'TimeoutError' || err.name === 'AbortError')); + +const withDeadline = (promise: Promise, ms: number): Promise => { + let timer: NodeJS.Timeout | undefined; + const deadline = new Promise((_, reject) => { + timer = setTimeout( + () => + reject( + new DOMException( + 'The operation was aborted due to timeout', + 'TimeoutError', + ), + ), + ms, + ); + }); + return Promise.race([promise, deadline]).finally(() => clearTimeout(timer)); +}; + export interface InternalDeployTarget { /** Script name to deploy under, in place of `workerName`. */ scriptName: string; @@ -289,6 +342,11 @@ export class WorkerDriver extends PuterDriver { #cfBaseUrl = ''; #hotReloadSubscribed = false; + // Row uids this process tore down itself; its own announcement skips them. + #tornDown = new BoundedTtlMap({ + maxEntries: 1000, + ttlMs: 30_000, + }); static currentPreambleVersion(): string | null { return preambleVersion; @@ -568,6 +626,7 @@ export class WorkerDriver extends PuterDriver { return cfResult; } + /** Tear down the caller's worker; a refused edge delete is a 502. */ async destroy(args: Record): Promise { const actor = requireContextUserActor(); assertActorEmailVerified(actor, this.config); @@ -595,12 +654,60 @@ export class WorkerDriver extends PuterDriver { this.#requireCfConfig(); - const cfResult = await this.#cfDelete(workerName); - await this.stores.subdomain.deleteByUuid(row.uuid, { - userId: actor.user.id, - }); + await this.#teardown(row, { userId: actor.user.id }); + // Stable regardless of the edge body: puter.js needs `result` truthy. + return { + success: true, + errors: [], + messages: [], + result: { id: workerName }, + }; + } + + /** Operator teardown by row uuid: no actor, no ownership check. */ + async [INTERNAL_ADMIN_DESTROY](input: { + uuid: string; + }): Promise { + const uuid = String(input?.uuid ?? ''); + const row = uuid + ? await this.stores.subdomain.getByUuid(uuid, { primary: true }) + : null; + const subdomain = String(row?.subdomain ?? ''); + if (!row || !subdomain.startsWith(WORKER_SUBDOMAIN_PREFIX)) + throw new HttpError(404, 'Worker not found', { + legacyCode: 'not_found', + }); + + this.#requireCfConfig(); + + await this.#teardown(row); + return { + subdomain, + userId: Number(row.user_id), + workerName: subdomain.slice(WORKER_SUBDOMAIN_PREFIX.length), + }; + } + + /** Script, sessions, then row: a failed step leaves the row for a retry. */ + async #teardown( + row: SubdomainRow, + { userId }: { userId?: number } = {}, + ): Promise { + const subdomain = String(row.subdomain); + const workerName = subdomain.slice(WORKER_SUBDOMAIN_PREFIX.length); + await this.#takeDownScript(workerName); await this.services.auth.deleteWorkerSessionsByName(workerName); - return cfResult; + await this.stores.subdomain.deleteByUuid(row.uuid, { userId }); + this.#tornDown.set(row.uuid, true); + try { + this.clients.event.emit( + 'subdomain.delete', + { subdomain, uid: row.uuid }, + {}, + ); + } catch { + // Non-critical. + } } async getFilePaths(args: Record): Promise { @@ -858,16 +965,70 @@ export class WorkerDriver extends PuterDriver { return { success: false, errors, url: null }; } + /** Edge delete: success or code 10007 resolves, anything else refuses. */ async #cfDelete(workerName: string): Promise> { if (USE_LOCAL_WORKERD) { - return this.services.localworkerservice.cfDeleteLocal(workerName); + return withDeadline( + this.services.localworkerservice.cfDeleteLocal(workerName), + CF_DELETE_TIMEOUT_MS, + ); } const cfg = this.#workerConfig(); const res = await fetch(`${this.#cfBaseUrl}/scripts/${workerName}/`, { method: 'DELETE', headers: { Authorization: `Bearer ${cfg.XAUTHKEY}` }, + signal: AbortSignal.timeout(CF_DELETE_TIMEOUT_MS), }); - return (await res.json()) as Record; + let json: { + success?: unknown; + errors?: Array<{ code?: unknown; message?: unknown }>; + } = {}; + try { + json = (await res.json()) ?? {}; + } catch { + // No usable body; the status decides. + } + if (res.ok && json.success !== false) return json; + const errors = Array.isArray(json.errors) ? json.errors : []; + if (errors.some((e) => Number(e?.code) === CF_SCRIPT_NOT_FOUND_CODE)) + return { success: true, errors: [], result: null }; + const first = errors[0]?.message; + throw new WorkerTakedownRefusedError( + res.status, + typeof first === 'string' ? first : undefined, + ); + } + + /** Refusal is 502, timeout or transport 504; edge text is only logged. */ + async #takeDownScript(workerName: string): Promise { + try { + await this.#cfDelete(workerName); + } catch (err) { + if (err instanceof WorkerTakedownRefusedError) { + console.warn( + `[workers] edge refused delete of ${workerName}: ${err.message}`, + ); + throw new HttpError( + 502, + 'Cloudflare refused to remove the script; nothing was deleted.', + { + legacyCode: 'worker_takedown_failed', + cause: err, + fields: { status: err.status }, + }, + ); + } + if (!isEdgeTransportError(err)) throw err; + console.warn( + `[workers] edge delete of ${workerName} did not complete`, + err, + ); + throw new HttpError( + 504, + 'The script delete did not complete; the worker is kept and a retry finishes it.', + { legacyCode: 'worker_takedown_timeout', cause: err }, + ); + } } // -- Helpers ------------------------------------------------------ @@ -1093,9 +1254,15 @@ export class WorkerDriver extends PuterDriver { // owner's account) still has a deployed script to take down. this.clients.event.on( 'subdomain.delete', - (_key: string, data: { subdomain?: unknown }) => { + (_key: string, data: { subdomain?: unknown; uid?: unknown }) => { const subdomain = String(data?.subdomain ?? ''); if (!subdomain.startsWith(WORKER_SUBDOMAIN_PREFIX)) return; + // Its own announcement: that row is already fully torn down. + const uid = String(data?.uid ?? ''); + if (uid && this.#tornDown.has(uid)) { + this.#tornDown.delete(uid); + return; + } const workerName = subdomain.slice( WORKER_SUBDOMAIN_PREFIX.length, );