Add cancellation for collection launches
This commit is contained in:
@@ -0,0 +1,283 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import { newFixture, type Fixture } from '../helpers/fixture.js';
|
||||
import { prepareNativeHost } from '../helpers/native-host.js';
|
||||
import type { CollectionMode } from '../../src/lib/deployment-plan.js';
|
||||
|
||||
vi.mock('../../src/lib/server/integrations/rcon.js', () => ({
|
||||
rconGetPlayerCount: vi.fn(async () => 0),
|
||||
rconPlutoniumVersion: vi.fn(async () => ({ revision: 5354, error: null })),
|
||||
rconStatusResult: vi.fn(async () => ({ ok: true, endpoint: { host: 'agent', port: 28960 }, rawReply: '',
|
||||
status: { players: 0, map: 'zombie_theater', gametype: 'zom', roster: [] } }))
|
||||
}));
|
||||
vi.mock('../../src/lib/server/integrations/plutonium.js', () => ({
|
||||
checkRelease: vi.fn(async () => {}), freshRelease: () => 5354, probeVersion: vi.fn(async () => 5354)
|
||||
}));
|
||||
vi.mock('../../src/lib/server/pipeline/worker.js', () => ({ kick: vi.fn(async () => {}), kickProvisioning: vi.fn(async () => {}) }));
|
||||
|
||||
let fixture: Fixture;
|
||||
let hostId: string;
|
||||
let launchNumber: number;
|
||||
|
||||
beforeEach(async () => {
|
||||
vi.resetModules(); vi.clearAllMocks();
|
||||
launchNumber = 0;
|
||||
fixture = await newFixture([]);
|
||||
vi.stubEnv('RCON_PASSWORD', 'private-rcon-password');
|
||||
vi.stubEnv('DEPLOYMENT_CREATE_CONCURRENCY', '1');
|
||||
hostId = (await prepareNativeHost()).host.id;
|
||||
const backend = await import('../../src/lib/server/backends/native.js');
|
||||
vi.spyOn(backend, 'nativeRequest').mockImplementation(async (_instance, kind) => {
|
||||
if (kind !== 'instance.inspect') throw new Error(`Unexpected operation: ${kind}`);
|
||||
return { healthy: true, state: 'running', exists: true } as never;
|
||||
});
|
||||
});
|
||||
afterEach(async () => { await fixture.cleanup(); vi.unstubAllEnvs(); });
|
||||
|
||||
async function launch(count = 2, mapSlug = 'theater', mode: CollectionMode = 'fill', savedKeys = false) {
|
||||
const { saveBlueprint } = await import('../../src/lib/server/fleet/launch-recipes.js');
|
||||
const { previewCollection, queueCollection, getCollection } = await import('../../src/lib/server/fleet/collections.js');
|
||||
const blueprint = saveBlueprint({ name: `Cancellation fixture ${++launchNumber}`, entries: [{ game: 't5', mapSlug, count }] });
|
||||
const preview = previewCollection(blueprint.id, mode);
|
||||
const keys = Object.fromEntries(preview.servers.map(server => [server.name!, `${blueprint.id}-${server.name}`]));
|
||||
const queued = queueCollection({ previewId: preview.id, ...(savedKeys ? {} : { keys }) });
|
||||
if (!queued.collectionId) throw new Error('Expected collection work');
|
||||
return getCollection(queued.collectionId)!;
|
||||
}
|
||||
|
||||
async function advance() {
|
||||
const { processNativeProvisioning } = await import('../../src/lib/server/provisioning/native.js');
|
||||
await processNativeProvisioning(() => null);
|
||||
}
|
||||
|
||||
async function operationFor(deploymentId: string) {
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
const row = db().prepare('SELECT operation_id FROM native_provisioning WHERE deployment_id=?').get(deploymentId) as { operation_id: string | null } | undefined;
|
||||
const { getHostOperation } = await import('../../src/lib/server/hosts/index.js');
|
||||
return row?.operation_id ? getHostOperation(row.operation_id)! : null;
|
||||
}
|
||||
|
||||
async function acknowledge(deploymentId: string) {
|
||||
const operation = await operationFor(deploymentId);
|
||||
if (!operation) throw new Error('Missing durable native operation');
|
||||
const { reportHostOperation } = await import('../../src/lib/server/hosts/index.js');
|
||||
reportHostOperation(hostId, operation.id, { state: 'completed', result: { state: 'running', healthy: true } });
|
||||
await advance();
|
||||
}
|
||||
|
||||
describe('collection cancellation', () => {
|
||||
it('cancels every queued member, releases saved keys, and remains idempotent after reopening the database', async () => {
|
||||
const { saveServerKey, serverKeyPool } = await import('../../src/lib/server/fleet/server-keys.js');
|
||||
for (const key of ['first', 'second', 'third']) saveServerKey('t5', key);
|
||||
const launchDetail = await launch(3, 'theater', 'fill', true);
|
||||
const { cancelCollection, getCollection, listCollections } = await import('../../src/lib/server/fleet/collections.js');
|
||||
const { db, closeDb } = await import('../../src/lib/server/db/index.js');
|
||||
expect(launchDetail.canCancel).toBe(true);
|
||||
expect(serverKeyPool('t5')).toEqual({ total: 3, available: 0 });
|
||||
expect(cancelCollection(launchDetail.collection.id)).toBe(true);
|
||||
const cancelled = getCollection(launchDetail.collection.id)!;
|
||||
expect(cancelled.canCancel).toBe(false);
|
||||
expect(cancelled.collection).toMatchObject({ cancelled_at: expect.any(String), total: 3, cancelled: 3,
|
||||
done: 0, active: 0, failed: 0, cleanupActive: 0, cleanupFailed: 0 });
|
||||
expect(cancelled.servers.every(server => server.state === 'cancelled' && server.cancellation?.state === 'done')).toBe(true);
|
||||
expect(serverKeyPool('t5')).toEqual({ total: 3, available: 3 });
|
||||
expect(db().prepare('SELECT * FROM server_key_leases').all()).toEqual([]);
|
||||
closeDb();
|
||||
expect(cancelCollection(launchDetail.collection.id)).toBe(true);
|
||||
expect(getCollection(launchDetail.collection.id)).toEqual(cancelled);
|
||||
expect(listCollections()[0]).toEqual(cancelled.collection);
|
||||
expect(db().prepare('SELECT * FROM native_creation_cancellations').all()).toHaveLength(3);
|
||||
await advance();
|
||||
expect(db().prepare("SELECT * FROM ec_host_operations WHERE kind LIKE 'instance.%'").all()).toEqual([]);
|
||||
});
|
||||
|
||||
it('rolls back all child cancellation and key releases if one cancellation fails', async () => {
|
||||
const detail = await launch(2);
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
const { cancelCollection, getCollection } = await import('../../src/lib/server/fleet/collections.js');
|
||||
db().exec(`CREATE TRIGGER reject_second_cancellation BEFORE UPDATE OF state ON server_deployments
|
||||
WHEN NEW.state='cancelled' AND OLD.id='${detail.servers[1].id}'
|
||||
BEGIN SELECT RAISE(ABORT, 'Cancellation fixture failure'); END`);
|
||||
expect(() => cancelCollection(detail.collection.id)).toThrow('Cancellation fixture failure');
|
||||
expect(getCollection(detail.collection.id)?.collection.cancelled_at).toBeNull();
|
||||
expect(getCollection(detail.collection.id)?.servers.map(server => server.state)).toEqual(['queued', 'queued']);
|
||||
expect(db().prepare('SELECT * FROM native_creation_cancellations').all()).toEqual([]);
|
||||
expect(db().prepare('SELECT * FROM server_key_leases').all()).toHaveLength(2);
|
||||
expect(db().prepare("SELECT * FROM server_deployments WHERE action='destroy'").all()).toEqual([]);
|
||||
db().exec('DROP TRIGGER reject_second_cancellation');
|
||||
expect(cancelCollection(detail.collection.id)).toBe(true);
|
||||
});
|
||||
|
||||
it('cleans up unfinished members that are waiting for health, interrupted, or failed', async () => {
|
||||
const detail = await launch(4);
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
const { cancelCollection, getCollection } = await import('../../src/lib/server/fleet/collections.js');
|
||||
const { serverKeyPool } = await import('../../src/lib/server/fleet/server-keys.js');
|
||||
['waiting', 'awaiting-health', 'interrupted', 'failed'].forEach((state, index) => {
|
||||
db().prepare('UPDATE server_deployments SET state=?,error=? WHERE id=?').run(state, 'Previous launch problem', detail.servers[index].id);
|
||||
});
|
||||
expect(getCollection(detail.collection.id)?.collection.failed).toBe(2);
|
||||
expect(cancelCollection(detail.collection.id)).toBe(true);
|
||||
const result = getCollection(detail.collection.id)!;
|
||||
expect(result.collection).toMatchObject({ total: 4, cancelled: 4, active: 0, failed: 0 });
|
||||
expect(result.servers.every(server => server.state === 'cancelled' && server.error === null && server.cancellation?.state === 'done')).toBe(true);
|
||||
expect(serverKeyPool('t5').available).toBe(4);
|
||||
});
|
||||
|
||||
it('preserves successful members, retained servers, and another launch', async () => {
|
||||
const retained = await launch(1);
|
||||
await advance(); await acknowledge(retained.servers[0].id);
|
||||
const detail = await launch(4);
|
||||
const unrelated = await launch(1, 'moon');
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
const { cancelCollection, getCollection } = await import('../../src/lib/server/fleet/collections.js');
|
||||
// Completed jobs, including successful jobs with warnings, are never teardown candidates.
|
||||
db().prepare("UPDATE server_deployments SET state='done' WHERE id=?").run(detail.servers[0].id);
|
||||
db().prepare("UPDATE server_deployments SET state='warning' WHERE id=?").run(detail.servers[1].id);
|
||||
expect(detail.retainedServers).toEqual([retained.servers[0].server]);
|
||||
expect(cancelCollection(detail.collection.id)).toBe(true);
|
||||
const result = getCollection(detail.collection.id)!;
|
||||
expect(result.collection).toMatchObject({ done: 2, cancelled: 1, failed: 0 });
|
||||
expect(result.servers.map(server => server.state)).toEqual(['done', 'warning', 'cancelled']);
|
||||
expect(result.servers.slice(0, 2).every(server => server.cancellation === null)).toBe(true);
|
||||
expect(getCollection(retained.collection.id)?.collection).toMatchObject({ cancelled_at: null, done: 1 });
|
||||
expect(db().prepare('SELECT name FROM servers WHERE name=?').get(retained.servers[0].server)).toEqual({ name: retained.servers[0].server });
|
||||
expect(getCollection(unrelated.collection.id)?.collection).toMatchObject({ cancelled_at: null, active: 1 });
|
||||
});
|
||||
|
||||
it.each(['completed', 'failed'] as const)('tracks running member cleanup through a %s agent result', async outcome => {
|
||||
const detail = await launch(2);
|
||||
await advance();
|
||||
const original = (await operationFor(detail.servers[0].id))!;
|
||||
expect(original.kind).toBe('instance.create');
|
||||
const { cancelCollection, getCollection } = await import('../../src/lib/server/fleet/collections.js');
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
const { reportHostOperation } = await import('../../src/lib/server/hosts/index.js');
|
||||
const { serverKeyPool } = await import('../../src/lib/server/fleet/server-keys.js');
|
||||
expect(cancelCollection(detail.collection.id)).toBe(true);
|
||||
const cancelled = getCollection(detail.collection.id)!;
|
||||
expect(cancelled.collection).toMatchObject({ active: 0, failed: 0, cancelled: 2, cleanupActive: 1, cleanupFailed: 0 });
|
||||
const cleanup = cancelled.servers[0].cancellation!;
|
||||
expect(cleanup).toMatchObject({ cleanupId: expect.any(String), state: 'running', error: null });
|
||||
const command = (await operationFor(cleanup.cleanupId))!;
|
||||
expect(command.kind).toBe('instance.cancel-create');
|
||||
expect(command.payload).toEqual({ createOperationId: original.id });
|
||||
expect(serverKeyPool('t5').available).toBe(1);
|
||||
expect(db().prepare('SELECT * FROM ec_host_ports').all()).toHaveLength(2);
|
||||
reportHostOperation(hostId, command.id, outcome === 'completed'
|
||||
? { state: 'completed', result: { exists: false } }
|
||||
: { state: 'failed', error: 'Agent refused cleanup before mutation' });
|
||||
await advance();
|
||||
const settled = getCollection(detail.collection.id)!;
|
||||
expect(settled.collection).toMatchObject({ active: 0, failed: 0, cancelled: 2, cleanupActive: 0, cleanupFailed: outcome === 'failed' ? 1 : 0 });
|
||||
expect(settled.servers[0].cancellation).toMatchObject({ cleanupId: cleanup.cleanupId,
|
||||
state: outcome === 'completed' ? 'done' : 'failed', error: outcome === 'completed' ? null : 'Agent refused cleanup before mutation' });
|
||||
expect(serverKeyPool('t5').available).toBe(outcome === 'completed' ? 2 : 1);
|
||||
expect(db().prepare('SELECT * FROM ec_host_ports').all()).toHaveLength(outcome === 'completed' ? 0 : 2);
|
||||
expect(cancelCollection(detail.collection.id)).toBe(true);
|
||||
expect(db().prepare('SELECT * FROM native_creation_cancellations').all()).toHaveLength(2);
|
||||
});
|
||||
|
||||
it('skips replaced instances and preserves cleanup already requested for an individual member', async () => {
|
||||
const detail = await launch(3);
|
||||
await advance();
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
const { cancelCreation } = await import('../../src/lib/server/provisioning/cancel-creation.js');
|
||||
const { cancelCollection, getCollection } = await import('../../src/lib/server/fleet/collections.js');
|
||||
db().prepare('UPDATE native_instances SET generation=generation+1 WHERE server=?').run(detail.servers[0].server);
|
||||
const independentCleanup = cancelCreation(detail.servers[1].id)!;
|
||||
expect(independentCleanup).toBeTruthy();
|
||||
expect(cancelCollection(detail.collection.id)).toBe(true);
|
||||
const result = getCollection(detail.collection.id)!;
|
||||
expect(result.servers[0]).toMatchObject({ state: 'running', cancellation: null });
|
||||
expect(result.servers[1].cancellation?.cleanupId).toBe(independentCleanup);
|
||||
expect(result.servers[2]).toMatchObject({ state: 'cancelled', cancellation: { state: 'done' } });
|
||||
expect(db().prepare('SELECT * FROM native_creation_cancellations').all()).toHaveLength(2);
|
||||
expect(db().prepare('SELECT generation,retired_at FROM native_instances WHERE server=?').get(detail.servers[0].server))
|
||||
.toEqual({ generation: 2, retired_at: null });
|
||||
});
|
||||
|
||||
it('requests cleanup for a member whose individual job was cancelled without removing its instance', async () => {
|
||||
const detail = await launch(1);
|
||||
await advance();
|
||||
const { cancelProvisioningJob } = await import('../../src/lib/server/provisioning/cancellation.js');
|
||||
const { cancelCollection, getCollection } = await import('../../src/lib/server/fleet/collections.js');
|
||||
const { serverKeyPool } = await import('../../src/lib/server/fleet/server-keys.js');
|
||||
expect(cancelProvisioningJob(detail.servers[0].id)).toBe(true);
|
||||
expect(serverKeyPool('t5').available).toBe(0);
|
||||
expect(getCollection(detail.collection.id)?.canCancel).toBe(true);
|
||||
expect(cancelCollection(detail.collection.id)).toBe(true);
|
||||
const result = getCollection(detail.collection.id)!;
|
||||
expect(result.collection).toMatchObject({ cancelled: 1, cleanupActive: 1 });
|
||||
const cleanup = result.servers[0].cancellation!;
|
||||
expect(await operationFor(cleanup.cleanupId)).toMatchObject({ kind: 'instance.cancel-create' });
|
||||
await acknowledge(cleanup.cleanupId);
|
||||
expect(serverKeyPool('t5').available).toBe(1);
|
||||
expect(getCollection(detail.collection.id)?.collection.cleanupActive).toBe(0);
|
||||
});
|
||||
|
||||
it.each(['waiting', 'shared', 'started', 'manual'] as const)('handles an exact collection with a %s removal', async removalState => {
|
||||
const retained = await launch(1);
|
||||
await advance(); await acknowledge(retained.servers[0].id);
|
||||
if (removalState === 'manual') {
|
||||
const { destroyServer } = await import('../../src/lib/server/provisioning/instances.js');
|
||||
expect(destroyServer(retained.servers[0].server).ok).toBe(true);
|
||||
}
|
||||
const detail = await launch(1, 'moon', 'exact');
|
||||
const removal = detail.servers.find(server => server.action === 'destroy')!;
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
const { cancelCollection, getCollection } = await import('../../src/lib/server/fleet/collections.js');
|
||||
if (removalState === 'shared') {
|
||||
db().prepare('INSERT INTO server_launch_collections(id,name,created_at,mode) VALUES(?,?,?,?)')
|
||||
.run('other-collection', 'Another exact launch', new Date().toISOString(), 'exact');
|
||||
db().prepare('INSERT INTO server_collection_removals(collection_id,deployment_id,position) VALUES(?,?,0)')
|
||||
.run('other-collection', removal.id);
|
||||
} else if (removalState === 'started') {
|
||||
await advance();
|
||||
expect(await operationFor(removal.id)).toMatchObject({ kind: 'instance.remove' });
|
||||
}
|
||||
expect(cancelCollection(detail.collection.id)).toBe(true);
|
||||
const result = getCollection(detail.collection.id)!;
|
||||
expect(result.servers.find(server => server.id === removal.id)?.state).toBe(removalState === 'waiting' ? 'cancelled' : removalState === 'started' ? 'running' : 'waiting');
|
||||
expect(result.servers.find(server => server.action === 'create')?.state).toBe('cancelled');
|
||||
expect(db().prepare('SELECT name FROM servers WHERE name=?').get(removal.server)).toEqual({ name: removal.server });
|
||||
if (removalState === 'shared') {
|
||||
expect(getCollection('other-collection')?.canCancel).toBe(true);
|
||||
expect(cancelCollection('other-collection')).toBe(true);
|
||||
expect(getCollection('other-collection')?.servers[0].state).toBe('cancelled');
|
||||
}
|
||||
});
|
||||
|
||||
it('returns API errors for missing or completed collections and accepts repeated cancellations', async () => {
|
||||
const completed = await launch(1);
|
||||
await advance(); await acknowledge(completed.servers[0].id);
|
||||
const { cancelCollection, getCollection } = await import('../../src/lib/server/fleet/collections.js');
|
||||
const { POST } = await import('../../src/routes/api/deployment-collections/[id]/cancel/+server.js');
|
||||
const { kickProvisioning } = await import('../../src/lib/server/pipeline/worker.js');
|
||||
expect(cancelCollection('missing')).toBe(false);
|
||||
expect(cancelCollection(completed.collection.id)).toBe(false);
|
||||
expect(getCollection(completed.collection.id)?.canCancel).toBe(false);
|
||||
expect((await POST({ params: { id: 'missing' } })).status).toBe(404);
|
||||
expect((await POST({ params: { id: completed.collection.id } })).status).toBe(409);
|
||||
expect(kickProvisioning).not.toHaveBeenCalled();
|
||||
const pending = await launch(2);
|
||||
for (let repeat = 0; repeat < 2; repeat++) {
|
||||
const response = await POST({ params: { id: pending.collection.id } });
|
||||
expect(response.status).toBe(200);
|
||||
expect(await response.json()).toEqual({ ok: true });
|
||||
}
|
||||
expect(kickProvisioning).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('migrates existing collections with no cancellation marker and preserves their jobs', async () => {
|
||||
const detail = await launch(1);
|
||||
const { db, closeDb } = await import('../../src/lib/server/db/index.js');
|
||||
db().exec('ALTER TABLE server_launch_collections DROP COLUMN cancelled_at');
|
||||
closeDb();
|
||||
const { getCollection, cancelCollection } = await import('../../src/lib/server/fleet/collections.js');
|
||||
const migrated = getCollection(detail.collection.id)!;
|
||||
expect(migrated.collection.cancelled_at).toBeNull();
|
||||
expect(migrated.canCancel).toBe(true);
|
||||
expect(migrated.servers[0]).toMatchObject({ id: detail.servers[0].id, state: 'queued', cancellation: null });
|
||||
expect(cancelCollection(detail.collection.id)).toBe(true);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user