300 lines
20 KiB
TypeScript
300 lines
20 KiB
TypeScript
import { randomUUID } from 'node:crypto';
|
|
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 { RebalanceMove } from '../../src/lib/rebalancing.js';
|
|
|
|
vi.mock('../../src/lib/server/pipeline/worker.js', () => ({ kick: vi.fn(async () => {}), kickProvisioning: vi.fn(async () => {}) }));
|
|
vi.mock('../../src/lib/server/integrations/plutonium.js', () => ({ checkRelease: vi.fn(async () => {}), freshRelease: () => 5354, probeVersion: vi.fn(async () => 5354) }));
|
|
vi.mock('../../src/lib/server/integrations/rcon.js', () => ({
|
|
rconStatusResult: vi.fn(async () => ({ ok: true, endpoint: { host: 'agent', port: 28960 },
|
|
status: { players: 0, map: 'zombie_theater', gametype: 'zom', roster: [] } })),
|
|
rconGetPlayerCount: vi.fn(async () => 0), rconPlutoniumVersion: vi.fn(async () => ({ revision: 5354, error: null }))
|
|
}));
|
|
|
|
let fixture: Fixture, sourceId: string, targetId: string, libraryId: string;
|
|
const names = Array.from({ length: 4 }, (_, index) => `t5_zombie_theater_${index + 1}`);
|
|
const image = 'example.test/plutainer:rebalance';
|
|
const gib = 1024 ** 3;
|
|
const clockStep = () => vi.setSystemTime(Date.now() + 1000);
|
|
const settings = () => ({ hostIds: [sourceId, targetId], game: 'all' as const, maxMoves: 5 });
|
|
|
|
async function heartbeat(hostId: string, cpuUsagePercent: number) {
|
|
const { db } = await import('../../src/lib/server/db/index.js');
|
|
const { heartbeat } = await import('../../src/lib/server/hosts/index.js');
|
|
const instances = db().prepare('SELECT id,generation FROM native_instances WHERE host_id=? AND retired_at IS NULL')
|
|
.all(hostId) as { id: string; generation: number }[];
|
|
heartbeat(hostId, { protocolVersion: 1, cpuCount: 16, cpuUsagePercent, memoryTotal: 32 * gib,
|
|
memoryAvailable: 32 * gib - (2 + instances.length * 2) * gib, dockerRunning: true,
|
|
directoriesWritable: true, storageAvailable: 100 * gib, runtimeImages: [image],
|
|
libraries: [{ id: libraryId, state: 'ready' }], instanceResources: instances.map(instance => ({
|
|
instanceId: instance.id, generation: instance.generation, uptimeSeconds: 3600, cpuCores: 3, memoryBytes: 2 * gib })) });
|
|
}
|
|
|
|
beforeEach(async () => {
|
|
vi.resetModules(); vi.clearAllMocks();
|
|
vi.useFakeTimers({ toFake: ['Date'] }); vi.setSystemTime(new Date('2026-10-08T16:00:00.000Z'));
|
|
fixture = await newFixture([]); vi.stubEnv('PLUTAINER_IMAGE', image); vi.stubEnv('RCON_PASSWORD', 'preserved-password');
|
|
const { db } = await import('../../src/lib/server/db/index.js');
|
|
const at = new Date().toISOString(), launched = new Date(Date.now() - 600_000).toISOString();
|
|
for (const [index, name] of names.entries()) db().prepare(`INSERT INTO servers(name,game,map,game_mode,map_slug,
|
|
config_file,game_id,image,game_port,container_state,container_checked_at,last_players,last_map,last_seen_at,last_checked_at,last_roster)
|
|
VALUES(?,'t5','zombie_theater','zombie','theater','dedicated_sp.cfg','t5sp',?,?,'running',?,0,'zombie_theater',?,?,'[]')`)
|
|
.run(name, image, 28960 + index, at, at, at);
|
|
const prepared = await prepareNativeHost(names); sourceId = prepared.host.id; libraryId = prepared.versions.get('t5')!;
|
|
const hosts = await import('../../src/lib/server/hosts/index.js');
|
|
targetId = hosts.createHost({ name: 'Empty destination', portMin: 29000, portMax: 29020 }).id;
|
|
hosts.enrollHost(hosts.createEnrollment(targetId).token, { protocolVersion: 1, cpuCount: 16, cpuUsagePercent: 6.25,
|
|
memoryTotal: 32 * gib, memoryAvailable: 30 * gib, dockerRunning: true, directoriesWritable: true,
|
|
storageAvailable: 100 * gib, runtimeImages: [image], libraries: [{ id: libraryId, state: 'ready' }] });
|
|
const libraries = await import('../../src/lib/server/game-files/index.js');
|
|
libraries.updateHostLibraryCaches(targetId, [{ libraryVersionId: libraryId, status: 'ready',
|
|
manifestSha256: libraries.getLibraryVersion(libraryId)!.manifestSha256 }]);
|
|
const { nativeInstance } = await import('../../src/lib/server/backends/native.js');
|
|
const { beginDeployment } = await import('../../src/lib/server/provisioning/history.js');
|
|
const { captureRecipe, recipeStorage } = await import('../../src/lib/server/fleet/launch-recipes.js');
|
|
const { reserveServerKey } = await import('../../src/lib/server/fleet/server-keys.js');
|
|
const { recordScript } = await import('../../src/lib/server/fleet/inventory.js');
|
|
for (const name of names) {
|
|
const instance = nativeInstance(name)!;
|
|
reserveServerKey(name, 't5', `saved-key-${name}`);
|
|
for (const protocol of ['udp', 'tcp'] as const) hosts.reserveHostPort(sourceId, instance.id, { port: instance.port, bindAddress: '0.0.0.0', protocol });
|
|
const id = beginDeployment(name, 'create', 'fixture');
|
|
const operation = hosts.queueHostOperation(sourceId, 'instance.create', { name, rconPassword: 'preserved-password' },
|
|
{ instanceId: instance.id, generation: 1, idempotencyKey: randomUUID() });
|
|
hosts.reportHostOperation(sourceId, operation.id, { state: 'completed', result: { state: 'running', healthy: true } });
|
|
db().prepare("UPDATE server_deployments SET state='done',submitted_at=?,finished_at=? WHERE id=?").run(launched, launched, id);
|
|
db().prepare('UPDATE native_provisioning SET operation_id=?,verified_at=? WHERE deployment_id=?').run(operation.id, launched, id);
|
|
const recipe = captureRecipe('t5', 'theater', 'preserved-password');
|
|
const stored = recipeStorage({ ...recipe, image, native: { hostId: sourceId, libraryId, image } });
|
|
db().prepare('INSERT INTO server_launch_recipes(deployment_id,recipe,recipe_summary) VALUES(?,?,?)').run(id, stored.recipe, stored.recipeSummary);
|
|
for (const file of recipe.files) recordScript(name, file.path, file.versionId, file.sha256, file.size, 'native-create');
|
|
}
|
|
await heartbeat(sourceId, 75); await heartbeat(targetId, 6.25);
|
|
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 host request ${kind}`);
|
|
return { state: 'running', healthy: true, exists: true } as never;
|
|
});
|
|
});
|
|
afterEach(async () => { vi.useRealTimers(); vi.restoreAllMocks(); await fixture.cleanup(); vi.unstubAllEnvs(); });
|
|
|
|
async function start() {
|
|
const service = await import('../../src/lib/server/rebalancing/index.js');
|
|
const preview = service.previewRebalance(settings());
|
|
expect(preview.moves.length, JSON.stringify(preview.exclusions)).toBeGreaterThanOrEqual(2);
|
|
return service.startRebalance(preview.id);
|
|
}
|
|
async function currentMove(runId: string): Promise<RebalanceMove> {
|
|
const { getRebalanceRun } = await import('../../src/lib/server/rebalancing/index.js');
|
|
return getRebalanceRun(runId)!.moves.find(move => move.state === 'moving')!;
|
|
}
|
|
async function provision() {
|
|
const { processProvisioning } = await import('../../src/lib/server/provisioning/worker.js');
|
|
await processProvisioning();
|
|
}
|
|
async function finishMove(move: RebalanceMove) {
|
|
const { hostMigration } = await import('../../src/lib/server/provisioning/host-drain.js');
|
|
const hosts = await import('../../src/lib/server/hosts/index.js');
|
|
const { db } = await import('../../src/lib/server/db/index.js');
|
|
const removal = hostMigration(move.operationId!)!.remove_operation_id;
|
|
expect(removal).toBeTruthy();
|
|
clockStep(); hosts.reportHostOperation(sourceId, removal!, { state: 'completed' });
|
|
await provision(); await provision();
|
|
const created = db().prepare('SELECT operation_id FROM native_provisioning WHERE deployment_id=?').get(move.operationId) as { operation_id: string | null };
|
|
expect(created.operation_id).toBeTruthy();
|
|
clockStep(); hosts.reportHostOperation(targetId, created.operation_id!, { state: 'completed', result: { state: 'running', healthy: true } });
|
|
clockStep(); await provision();
|
|
expect(db().prepare('SELECT state FROM server_deployments WHERE id=?').get(move.operationId)).toEqual({ state: 'done' });
|
|
}
|
|
|
|
describe('manual rebalance runs', () => {
|
|
it('previews without reserving capacity and rejects stale plans', async () => {
|
|
const { db } = await import('../../src/lib/server/db/index.js');
|
|
const { previewRebalance, startRebalance } = await import('../../src/lib/server/rebalancing/index.js');
|
|
const ports = db().prepare('SELECT * FROM ec_host_ports').all();
|
|
const commands = db().prepare('SELECT id FROM ec_host_operations').all();
|
|
const preview = previewRebalance(settings());
|
|
expect(preview.moves.length).toBeGreaterThanOrEqual(2);
|
|
expect(db().prepare('SELECT * FROM ec_host_ports').all()).toEqual(ports);
|
|
expect(db().prepare('SELECT id FROM ec_host_operations').all()).toEqual(commands);
|
|
expect(db().prepare('SELECT * FROM rebalancing_runs').all()).toEqual([]);
|
|
db().prepare("UPDATE servers SET desired_state='stopped' WHERE name=?").run(preview.moves[0].server);
|
|
expect(() => startRebalance(preview.id)).toThrow('fleet changed');
|
|
});
|
|
|
|
it('starts idempotently, rejects concurrent runs, and queues exactly one move', async () => {
|
|
const service = await import('../../src/lib/server/rebalancing/index.js');
|
|
const preview = service.previewRebalance(settings());
|
|
const competing = service.previewRebalance(settings());
|
|
const run = service.startRebalance(preview.id);
|
|
expect(service.startRebalance(preview.id).id).toBe(run.id);
|
|
expect(() => service.startRebalance(competing.id)).toThrow('current rebalance');
|
|
expect(await service.runRebalancing()).toBe(true);
|
|
expect(await service.runRebalancing()).toBe(false);
|
|
const moves = service.getRebalanceRun(run.id)!.moves;
|
|
expect(moves.filter(move => move.operationId)).toHaveLength(1);
|
|
expect(moves.filter(move => move.state === 'planned').length).toBeGreaterThan(0);
|
|
});
|
|
|
|
it('rejects expired previews but returns an already-started run on a late retry', async () => {
|
|
const service = await import('../../src/lib/server/rebalancing/index.js');
|
|
const first = service.previewRebalance(settings());
|
|
const expired = service.previewRebalance(settings());
|
|
const run = service.startRebalance(first.id);
|
|
vi.setSystemTime(Date.now() + 6 * 60_000);
|
|
expect(service.startRebalance(first.id).id).toBe(run.id);
|
|
expect(() => service.startRebalance(expired.id)).toThrow('Preview expired');
|
|
});
|
|
|
|
it('stops before removal without changing placement or releasing the key, and allows a later host drain', async () => {
|
|
const run = await start();
|
|
const service = await import('../../src/lib/server/rebalancing/index.js');
|
|
await service.runRebalancing();
|
|
const move = await currentMove(run.id);
|
|
const stopped = service.stopRebalance(run.id);
|
|
expect(stopped.state).toBe('stopped');
|
|
expect(stopped.moves.every(move => move.state === 'skipped')).toBe(true);
|
|
await provision();
|
|
const { nativeInstance } = await import('../../src/lib/server/backends/native.js');
|
|
const { getServerKey } = await import('../../src/lib/server/fleet/server-keys.js');
|
|
expect(nativeInstance(move.server)?.host_id).toBe(sourceId);
|
|
expect(getServerKey(move.server)).toBe(`saved-key-${move.server}`);
|
|
const { db } = await import('../../src/lib/server/db/index.js');
|
|
expect(db().prepare("SELECT id FROM ec_host_operations WHERE kind='instance.remove'").all()).toEqual([]);
|
|
const { updateHost } = await import('../../src/lib/server/hosts/index.js');
|
|
updateHost(sourceId, { mode: 'draining' });
|
|
const { reconcileHostDrains } = await import('../../src/lib/server/provisioning/host-drain.js');
|
|
reconcileHostDrains();
|
|
expect(db().prepare("SELECT id FROM server_deployments WHERE server=? AND triggered_by='host-drain'").get(move.server)).toBeTruthy();
|
|
});
|
|
|
|
it('finishes a committed move after Stop while withdrawing the remaining moves', async () => {
|
|
const run = await start();
|
|
const service = await import('../../src/lib/server/rebalancing/index.js');
|
|
await service.runRebalancing(); await provision();
|
|
const move = await currentMove(run.id);
|
|
expect(service.stopRebalance(run.id).state).toBe('stopping');
|
|
const { cancelProvisioningJob } = await import('../../src/lib/server/provisioning/cancellation.js');
|
|
expect(cancelProvisioningJob(move.operationId!)).toBe(false);
|
|
await finishMove(move); await service.runRebalancing();
|
|
const finished = service.getRebalanceRun(run.id)!;
|
|
expect(finished.state).toBe('stopped');
|
|
expect(finished.moves.filter(move => move.state === 'done')).toHaveLength(1);
|
|
expect(finished.moves.filter(move => move.state === 'skipped')).toHaveLength(run.moves.length - 1);
|
|
const { nativeInstance } = await import('../../src/lib/server/backends/native.js');
|
|
const { getHost } = await import('../../src/lib/server/hosts/index.js');
|
|
expect(nativeInstance(move.server)).toMatchObject({ id: move.instanceId, generation: 2, host_id: targetId });
|
|
expect(getHost(sourceId)?.mode).toBe('active');
|
|
});
|
|
|
|
it('treats cancellation from Provisioning as stopping the rest of the run', async () => {
|
|
const run = await start();
|
|
const service = await import('../../src/lib/server/rebalancing/index.js');
|
|
await service.runRebalancing();
|
|
const move = await currentMove(run.id);
|
|
const { cancelProvisioningJob } = await import('../../src/lib/server/provisioning/cancellation.js');
|
|
expect(cancelProvisioningJob(move.operationId!)).toBe(true);
|
|
expect(service.getRebalanceRun(run.id)?.state).toBe('stopping');
|
|
await service.runRebalancing();
|
|
expect(service.getRebalanceRun(run.id)?.state).toBe('stopped');
|
|
expect(service.getRebalanceRun(run.id)?.moves.every(move => move.state === 'skipped')).toBe(true);
|
|
});
|
|
|
|
it('waits for both hosts to report fresh measurements before admitting the next move', async () => {
|
|
const run = await start();
|
|
const service = await import('../../src/lib/server/rebalancing/index.js');
|
|
await service.runRebalancing(); await provision();
|
|
await finishMove(await currentMove(run.id));
|
|
expect(await service.runRebalancing()).toBe(false);
|
|
expect(service.getRebalanceRun(run.id)?.message).toContain('fresh source and destination');
|
|
expect(service.getRebalanceRun(run.id)?.moves.filter(move => move.operationId)).toHaveLength(1);
|
|
clockStep(); await heartbeat(sourceId, 56.25);
|
|
expect(await service.runRebalancing()).toBe(false);
|
|
await heartbeat(targetId, 25);
|
|
expect(await service.runRebalancing()).toBe(true);
|
|
expect(service.getRebalanceRun(run.id)?.moves.filter(move => move.operationId)).toHaveLength(2);
|
|
});
|
|
|
|
it('skips an uncommitted move that becomes occupied and continues other approved work', async () => {
|
|
const run = await start();
|
|
const service = await import('../../src/lib/server/rebalancing/index.js');
|
|
await service.runRebalancing();
|
|
const move = await currentMove(run.id);
|
|
const { db } = await import('../../src/lib/server/db/index.js');
|
|
db().prepare('UPDATE servers SET last_players=2 WHERE name=?').run(move.server);
|
|
await service.runRebalancing();
|
|
const fresh = service.getRebalanceRun(run.id)!;
|
|
expect(fresh.state).toBe('running');
|
|
expect(fresh.moves.find(item => item.id === move.id)?.state).toBe('skipped');
|
|
expect(fresh.moves.filter(item => item.state === 'moving')).toHaveLength(1);
|
|
});
|
|
|
|
it('halts remaining moves after a provisioning failure', async () => {
|
|
const run = await start();
|
|
const service = await import('../../src/lib/server/rebalancing/index.js');
|
|
await service.runRebalancing();
|
|
const move = await currentMove(run.id);
|
|
const { db } = await import('../../src/lib/server/db/index.js');
|
|
db().prepare("UPDATE server_deployments SET state='failed',error='Recovery needed' WHERE id=?").run(move.operationId);
|
|
expect(await service.runRebalancing()).toBe(false);
|
|
const failed = service.getRebalanceRun(run.id)!;
|
|
expect(failed.state).toBe('failed');
|
|
expect(failed.moves.find(item => item.id === move.id)).toMatchObject({ state: 'failed', message: 'Recovery needed' });
|
|
expect(failed.moves.filter(item => item.state === 'skipped')).toHaveLength(run.moves.length - 1);
|
|
});
|
|
|
|
it('keeps the source running until a global startup slot is available', async () => {
|
|
const run = await start();
|
|
const service = await import('../../src/lib/server/rebalancing/index.js');
|
|
await service.runRebalancing();
|
|
const move = await currentMove(run.id);
|
|
const { db } = await import('../../src/lib/server/db/index.js');
|
|
const { beginDeployment } = await import('../../src/lib/server/provisioning/history.js');
|
|
const occupied = beginDeployment('another-server', 'create', 'fixture');
|
|
await provision();
|
|
const { hostMigration } = await import('../../src/lib/server/provisioning/host-drain.js');
|
|
expect(hostMigration(move.operationId!)?.remove_operation_id).toBeNull();
|
|
db().prepare("UPDATE server_deployments SET state='done' WHERE id=?").run(occupied);
|
|
await provision();
|
|
expect(hostMigration(move.operationId!)?.remove_operation_id).toEqual(expect.any(String));
|
|
});
|
|
|
|
it('retains its global startup reservation between source removal and destination creation', async () => {
|
|
const run = await start();
|
|
const service = await import('../../src/lib/server/rebalancing/index.js');
|
|
await service.runRebalancing(); await provision();
|
|
const move = await currentMove(run.id);
|
|
const { hostMigration } = await import('../../src/lib/server/provisioning/migrations.js');
|
|
const { reportHostOperation } = await import('../../src/lib/server/hosts/index.js');
|
|
reportHostOperation(sourceId, hostMigration(move.operationId!)!.remove_operation_id!, { state: 'completed' });
|
|
await provision();
|
|
const { db } = await import('../../src/lib/server/db/index.js');
|
|
expect(db().prepare('SELECT state FROM server_deployments WHERE id=?').get(move.operationId)).toEqual({ state: 'waiting' });
|
|
const { queueNativeCreation } = await import('../../src/lib/server/provisioning/native.js');
|
|
const extra = queueNativeCreation({ game: 't5', gameMode: 'zombie', mapSlug: 'theater',
|
|
hostId: targetId, libraryId, image, plutoKey: 'another-key' });
|
|
expect(extra.ok).toBe(true);
|
|
if (!extra.ok) throw new Error(extra.message);
|
|
await provision();
|
|
expect(db().prepare('SELECT operation_id FROM native_provisioning WHERE deployment_id=?').get(extra.deploymentId)).toBeUndefined();
|
|
expect(db().prepare('SELECT operation_id FROM native_provisioning WHERE deployment_id=?').get(move.operationId))
|
|
.toEqual({ operation_id: expect.any(String) });
|
|
});
|
|
|
|
it.each([{ state: 'running', healthy: false }, { state: 'stopped', healthy: false }])('holds removal when fresh inspection returns $state with healthy=$healthy', async health => {
|
|
const run = await start();
|
|
const service = await import('../../src/lib/server/rebalancing/index.js');
|
|
await service.runRebalancing();
|
|
const move = await currentMove(run.id);
|
|
const { nativeRequest } = await import('../../src/lib/server/backends/native.js');
|
|
vi.mocked(nativeRequest).mockResolvedValueOnce(health);
|
|
await provision();
|
|
const { hostMigration } = await import('../../src/lib/server/provisioning/host-drain.js');
|
|
expect(hostMigration(move.operationId!)?.remove_operation_id).toBeNull();
|
|
const { db } = await import('../../src/lib/server/db/index.js');
|
|
expect(db().prepare('SELECT stage FROM server_deployments WHERE id=?').get(move.operationId))
|
|
.toEqual({ stage: expect.stringContaining('no longer healthy') });
|
|
});
|
|
});
|