mirror of
https://github.com/HeyPuter/puter.git
synced 2026-10-24 23:58:47 +00:00
feat(workers): internal teardown that takes the script down before the row, and destroy reports Cloudflare refusals (PUT-2219) (#4297)
release-please / release-please (push) Canceled after 0s
Backend Tests / changes (push) Canceled after 0s
Backend Tests / typecheck (push) Canceled after 0s
Backend Tests / test shard 1/4 (push) Canceled after 0s
Backend Tests / test shard 2/4 (push) Canceled after 0s
Backend Tests / test shard 3/4 (push) Canceled after 0s
Backend Tests / test shard 4/4 (push) Canceled after 0s
Backend Tests / test (main) (push) Canceled after 0s
Backend Tests / coverage (push) Canceled after 0s
Maintain Release Merge PR / update-release-pr (push) Canceled after 0s
Notify HeyPuter / notify (push) Canceled after 0s
release-please / release-please (push) Canceled after 0s
Backend Tests / changes (push) Canceled after 0s
Backend Tests / typecheck (push) Canceled after 0s
Backend Tests / test shard 1/4 (push) Canceled after 0s
Backend Tests / test shard 2/4 (push) Canceled after 0s
Backend Tests / test shard 3/4 (push) Canceled after 0s
Backend Tests / test shard 4/4 (push) Canceled after 0s
Backend Tests / test (main) (push) Canceled after 0s
Backend Tests / coverage (push) Canceled after 0s
Maintain Release Merge PR / update-release-pr (push) Canceled after 0s
Notify HeyPuter / notify (push) Canceled after 0s
This commit is contained in:
1 parent
c8424f6775
commit
aa47dcd3e5
3 files changed
+887
-41
No files matched your search
@@ -17,6 +17,7 @@
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
|
||||
/**
|
||||
* 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<typeof fetch>;
|
||||
|
||||
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 <T>(fn: () => Promise<T>) => {
|
||||
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(
|
||||
|
||||
@@ -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<string, unknown>;
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -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 = <T>(promise: Promise<T>, ms: number): Promise<T> => {
|
||||
let timer: NodeJS.Timeout | undefined;
|
||||
const deadline = new Promise<never>((_, 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<string, true>({
|
||||
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<string, unknown>): Promise<unknown> {
|
||||
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<AdminDestroyResult> {
|
||||
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<void> {
|
||||
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<string, unknown>): Promise<unknown> {
|
||||
@@ -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<Record<string, unknown>> {
|
||||
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<string, unknown>;
|
||||
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<void> {
|
||||
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,
|
||||
);
|
||||
|
||||
Reference in new issue
Block a user