Compare commits
9
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
91f4490d34 | ||
|
|
5455f27d5b | ||
|
|
c8c53e87d6 | ||
|
|
847399a4ff | ||
|
|
a86499a0f9 | ||
|
|
ad0a0e1033 | ||
|
|
d24c52b5cc | ||
|
|
030794635e | ||
|
|
2af768011a |
+31
-5
@@ -32,7 +32,7 @@ import urllib.parse
|
||||
import urllib.request
|
||||
import uuid
|
||||
|
||||
VERSION = "1.4.10"
|
||||
VERSION = "1.4.11"
|
||||
LABEL = "io.electric-cherry"
|
||||
MAX_FILE = 32 * 1024 * 1024
|
||||
MANIFEST_MAX_ENTRIES = 10000
|
||||
@@ -918,6 +918,8 @@ class Agent:
|
||||
self.observation_keep_days = 0 # Older controllers do not authorize cleanup.
|
||||
self.library_progress = {}
|
||||
self.library_transfers = {}
|
||||
self.inventory_manifest_cache = {}
|
||||
self.inventory_manifest_lock = threading.Lock()
|
||||
self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
|
||||
self.executing_id = None
|
||||
|
||||
@@ -1191,6 +1193,25 @@ class Agent:
|
||||
continue
|
||||
return samples
|
||||
|
||||
def inventory_manifest(self, marker):
|
||||
# Inventory only: launch/download verification must still read disk.
|
||||
# Include file identity so atomic replacements invalidate even when
|
||||
# their size and modification time match the previous manifest.
|
||||
with self.inventory_manifest_lock:
|
||||
try:
|
||||
metadata = marker.stat()
|
||||
signature = (metadata.st_mtime_ns, metadata.st_size, metadata.st_dev, metadata.st_ino)
|
||||
cached = self.inventory_manifest_cache.get(marker)
|
||||
if cached is not None and cached[0] == signature:
|
||||
return cached[1]
|
||||
self.inventory_manifest_cache.pop(marker, None)
|
||||
value = json.loads(marker.read_text())
|
||||
self.inventory_manifest_cache[marker] = (signature, value)
|
||||
return value
|
||||
except (OSError, ValueError):
|
||||
self.inventory_manifest_cache.pop(marker, None)
|
||||
raise
|
||||
|
||||
def inventory(self):
|
||||
cpu_usage = self.cpu_usage_percent()
|
||||
usage = shutil.disk_usage(self.root)
|
||||
@@ -1210,13 +1231,18 @@ class Agent:
|
||||
libraries, caches = [], []
|
||||
progress = {key: value.copy() for key, value in list(self.library_progress.items())}
|
||||
progress.update({key: transfer.snapshot() for key, transfer in list(self.library_transfers.items())})
|
||||
for marker in (self.root / "libraries").glob("*/*/manifest.json"):
|
||||
manifests = list((self.root / "libraries").glob("*/*/manifest.json"))
|
||||
partial_manifests = list((self.root / "libraries").glob("*/.*.partial/source.json"))
|
||||
with self.inventory_manifest_lock:
|
||||
for marker in self.inventory_manifest_cache.keys() - set(manifests + partial_manifests):
|
||||
del self.inventory_manifest_cache[marker]
|
||||
for marker in manifests:
|
||||
# A crash between writing a manifest and its atomic promotion must
|
||||
# never expose the staging directory as a ready library.
|
||||
if marker.parent.name.startswith("."):
|
||||
continue
|
||||
try:
|
||||
value = json.loads(marker.read_text())
|
||||
value = self.inventory_manifest(marker)
|
||||
# Rechecking an existing cache must revoke readiness until its
|
||||
# entire contents have passed verification again.
|
||||
if value["libraryId"] not in progress and (marker.parent / "transfer.json").exists():
|
||||
@@ -1233,9 +1259,9 @@ class Agent:
|
||||
except (OSError, ValueError, KeyError):
|
||||
continue
|
||||
ready = {item["id"] for item in libraries}
|
||||
for marker in (self.root / "libraries").glob("*/.*.partial/source.json"):
|
||||
for marker in partial_manifests:
|
||||
try:
|
||||
value = json.loads(marker.read_text())
|
||||
value = self.inventory_manifest(marker)
|
||||
library_id = identifier(value["libraryId"])
|
||||
if library_id in ready or library_id in progress:
|
||||
continue
|
||||
|
||||
@@ -386,6 +386,68 @@ class AgentTests(unittest.TestCase):
|
||||
self.agent.journal.close()
|
||||
self.temp.cleanup()
|
||||
|
||||
def test_inventory_reuses_manifests_until_mtime_changes(self):
|
||||
marker = self.library / "manifest.json"
|
||||
saved = json.loads(marker.read_text())
|
||||
original_read = Path.read_text
|
||||
reads = []
|
||||
|
||||
def read(path, *args, **kwargs):
|
||||
if path == marker:
|
||||
reads.append(path)
|
||||
return original_read(path, *args, **kwargs)
|
||||
|
||||
with patch.object(Path, "read_text", read):
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 9)
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 9)
|
||||
self.assertEqual(len(reads), 1)
|
||||
metadata = marker.stat()
|
||||
marker.write_text(json.dumps({**saved, "size": 8}))
|
||||
module.os.utime(marker, ns=(metadata.st_atime_ns, metadata.st_mtime_ns + 1_000_000_000))
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8)
|
||||
self.assertEqual(len(reads), 2)
|
||||
|
||||
def test_inventory_reloads_atomic_replacement_with_same_size_and_mtime(self):
|
||||
marker = self.library / "manifest.json"
|
||||
saved = json.loads(marker.read_text())
|
||||
self.agent.inventory()
|
||||
metadata = marker.stat()
|
||||
module.atomic_json(marker, {**saved, "size": 8})
|
||||
module.os.utime(marker, ns=(metadata.st_atime_ns, metadata.st_mtime_ns))
|
||||
self.assertEqual(marker.stat().st_size, metadata.st_size)
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8)
|
||||
|
||||
def test_inventory_discards_invalid_and_removed_manifests(self):
|
||||
marker = self.library / "manifest.json"
|
||||
saved = json.loads(marker.read_text())
|
||||
self.agent.inventory()
|
||||
marker.write_text("invalid JSON")
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"], [])
|
||||
self.assertNotIn(marker, self.agent.inventory_manifest_cache)
|
||||
module.atomic_json(marker, saved)
|
||||
self.assertEqual(len(self.agent.inventory()["libraryCaches"]), 1)
|
||||
marker.unlink()
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"], [])
|
||||
self.assertNotIn(marker, self.agent.inventory_manifest_cache)
|
||||
module.atomic_json(marker, {**saved, "size": 8})
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8)
|
||||
|
||||
def test_cached_inventory_preserves_transfer_progress_and_integrity_checks(self):
|
||||
self.agent.inventory()
|
||||
progress = {"libraryVersionId": self.library_id, "status": "verifying", "bytesVerified": 3}
|
||||
module.atomic_json(self.library / "transfer.json", progress)
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["status"], "missing")
|
||||
module.atomic_json(self.library / "transfer.json", {**progress, "status": "failed", "bytesVerified": 5})
|
||||
cache = self.agent.inventory()["libraryCaches"][0]
|
||||
self.assertEqual(cache["status"], "failed")
|
||||
self.assertEqual(cache["bytesVerified"], 5)
|
||||
(self.library / "content/game.ff").write_bytes(b"bad-data!")
|
||||
with self.assertRaises(AgentError):
|
||||
self.agent.verify_library(self.library)
|
||||
(self.library / "manifest.json").write_text("invalid JSON")
|
||||
with self.assertRaises(ValueError):
|
||||
self.agent.verify_library(self.library)
|
||||
|
||||
def test_library_remove_preserves_other_versions_and_is_idempotent(self):
|
||||
other = self.agent.library_dir("t5", "other-library")
|
||||
other.mkdir()
|
||||
@@ -1179,6 +1241,17 @@ class AgentTests(unittest.TestCase):
|
||||
self.assertEqual(caches[0]["status"], "missing")
|
||||
self.assertEqual(caches[0]["bytesDownloaded"], len(content))
|
||||
self.assertNotIn({"id": "downloaded", "state": "ready"}, inventory["libraries"])
|
||||
(staging / "content/zone/test.ff").write_bytes(content[:5])
|
||||
original_read = Path.read_text
|
||||
|
||||
def read(path, *args, **kwargs):
|
||||
if path == staging / "source.json":
|
||||
self.fail("Unchanged partial manifest should be cached")
|
||||
return original_read(path, *args, **kwargs)
|
||||
|
||||
with patch.object(Path, "read_text", read):
|
||||
partial = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded")
|
||||
self.assertEqual(partial["bytesDownloaded"], 5)
|
||||
self.agent.library_progress["downloaded"] = {**caches[0], "status": "downloading", "bytesDownloaded": 3}
|
||||
caches = [item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded"]
|
||||
self.assertEqual(len(caches), 1)
|
||||
|
||||
@@ -595,6 +595,10 @@ missing, stale, or predates lifecycle changes. RCON status queries share one
|
||||
operation per host, with two concurrent local probes and a 20-second batch
|
||||
deadline. Individual failures remain visible per server. Both additions are
|
||||
capability-gated; older agents retain individual inspection and RCON operations.
|
||||
Agent 1.4.11 caches parsed library manifests during inventory collection until
|
||||
their modification time, size, or file identity changes. Deleted manifests leave
|
||||
the cache, while transfer progress and partial-file sizes are sampled each time.
|
||||
Launch and download integrity verification continue to read from disk.
|
||||
Agent 1.4.5 reserves two observation workers for instance health and read-only RCON,
|
||||
and one for files, logs, and host inventory. Background reads arrive in batches of
|
||||
up to four. File receipt uploads release the instance lock before contacting the
|
||||
|
||||
+5
-4
@@ -3,9 +3,10 @@ import { migrate } from '$lib/server/db/migrate.js';
|
||||
import { startWorker } from '$lib/server/pipeline/worker.js';
|
||||
import { interruptDeployments } from '$lib/server/provisioning/history.js';
|
||||
import { authenticationHandle } from '$lib/server/auth/guard.js';
|
||||
import { emit } from '$lib/server/pipeline/bus.js';
|
||||
import { invalidateFleet } from '$lib/server/pipeline/bus.js';
|
||||
import { startPerformanceMonitor } from '$lib/server/performance.js';
|
||||
import { withOperationSource } from '$lib/server/hosts/operation-context.js';
|
||||
import { withHostRequestContext } from '$lib/server/hosts/request-context.js';
|
||||
|
||||
let booted = false;
|
||||
function boot(): void {
|
||||
@@ -23,13 +24,13 @@ function boot(): void {
|
||||
export const handle: Handle = async (input) => {
|
||||
boot();
|
||||
const started = performance.now();
|
||||
const response = await withOperationSource(`${input.event.request.method} ${input.event.route.id ?? 'Web request'}`,
|
||||
() => authenticationHandle(input));
|
||||
const response = await withHostRequestContext(() => withOperationSource(`${input.event.request.method} ${input.event.route.id ?? 'Web request'}`,
|
||||
() => authenticationHandle(input)));
|
||||
const elapsed = performance.now() - started;
|
||||
response.headers.append('Server-Timing', `app;dur=${elapsed.toFixed(1)}`);
|
||||
if (process.env.EC_PERF_LOG === 'true' && elapsed > 250 && !input.event.route.id?.startsWith('/api/agent/')) {
|
||||
console.info('[request]', { route: input.event.route.id, method: input.event.request.method, status: response.status, elapsedMs: Math.round(elapsed) });
|
||||
}
|
||||
if (input.event.locals.user && !['GET','HEAD','OPTIONS'].includes(input.event.request.method) && response.status < 400) emit({ type: 'fleet-invalidated' });
|
||||
if (input.event.locals.user && !['GET','HEAD','OPTIONS'].includes(input.event.request.method) && response.status < 400) invalidateFleet();
|
||||
return response;
|
||||
};
|
||||
|
||||
@@ -391,6 +391,21 @@ export function db(): Database.Database {
|
||||
throw err;
|
||||
}
|
||||
|
||||
// Cache only runtime queries, after migrations have finished. Statements
|
||||
// belong to this connection, so reopening the DB starts with an empty cache.
|
||||
// Callers share statement options (e.g. raw()); use consistent options for
|
||||
// identical SQL and pass values to run/get/all rather than binding permanently.
|
||||
const statements = new Map<string, Database.Statement>();
|
||||
const prepare = handle.prepare.bind(handle);
|
||||
handle.prepare = ((sql: string) => {
|
||||
let statement = statements.get(sql);
|
||||
if (!statement) {
|
||||
statement = prepare(sql);
|
||||
statements.set(sql, statement);
|
||||
}
|
||||
return statement;
|
||||
}) as typeof handle.prepare;
|
||||
|
||||
_db = handle;
|
||||
return _db;
|
||||
}
|
||||
|
||||
@@ -2,19 +2,31 @@ import { actionLabels, type ProvisioningAction } from '../../provisioning.js';
|
||||
import { db } from '../db/index.js';
|
||||
import type { FleetOperation } from '../../fleet-overview.js';
|
||||
|
||||
/** Read persisted work, including recent completions for one-shot lobby watches. */
|
||||
/** Fleet clients need only active work and unresolved failures. */
|
||||
export function fleetOperations(now = Date.now()): FleetOperation[] {
|
||||
return readOperations(now);
|
||||
}
|
||||
|
||||
/** Recent completions are loaded only by the server's own page. */
|
||||
export function serverOperations(server: string, now = Date.now()): FleetOperation[] {
|
||||
return readOperations(now, server);
|
||||
}
|
||||
|
||||
function readOperations(now: number, server?: string): FleetOperation[] {
|
||||
const since = new Date(now - 86_400_000).toISOString();
|
||||
const bindings = { server: server ?? null, since };
|
||||
const scope = (column: string) => server === undefined ? '' : `AND ${column} = @server`;
|
||||
const handle = db();
|
||||
const operations: FleetOperation[] = [];
|
||||
const slots = handle.prepare('SELECT server, requested_at, error FROM restart_rollout WHERE healthy_at IS NULL')
|
||||
.all() as { server: string; requested_at: string; error: string | null }[];
|
||||
const slots = handle.prepare(`SELECT server, requested_at, error FROM restart_rollout WHERE healthy_at IS NULL ${scope('server')}`)
|
||||
.all(bindings) as { server: string; requested_at: string; error: string | null }[];
|
||||
const recovering = new Map(slots.map(slot => [slot.server, slot]));
|
||||
const representedRecovery = new Set<string>();
|
||||
const campaigns = handle.prepare(`SELECT d.id, d.action, d.started_at, ds.server,
|
||||
ds.restart_state, ds.error, ds.restarted_at, ds.players
|
||||
FROM deployments d JOIN deployment_servers ds ON ds.deployment_id = d.id
|
||||
WHERE ds.server_deleted_at IS NULL AND (ds.restart_state != 'done' OR ds.checked_at >= ? OR d.finished_at >= ? OR d.started_at >= ?)`).all(since, since, since) as {
|
||||
WHERE ds.server_deleted_at IS NULL ${scope('ds.server')} AND (ds.restart_state != 'done'
|
||||
${server === undefined ? '' : 'OR ds.checked_at >= @since OR d.finished_at >= @since OR d.started_at >= @since'})`).all(bindings) as {
|
||||
id: string; action: string; started_at: string; server: string; restart_state: string;
|
||||
error: string | null; restarted_at: string | null; players: number | null;
|
||||
}[];
|
||||
@@ -34,19 +46,19 @@ export function fleetOperations(now = Date.now()): FleetOperation[] {
|
||||
href: `/campaigns/${encodeURIComponent(row.id)}` });
|
||||
}
|
||||
// Only the newest lifecycle operation can still require action on a server.
|
||||
// Separate branches let SQLite use state and finish indexes; UNION deduplicates
|
||||
// recent warnings that are also the newest operation. Unary + keeps terminal
|
||||
// branches on finished_at and rowid instead of searching historical states.
|
||||
// Separate branches use state/rowid indexes for fleet reads and the server
|
||||
// index for detail history. Completed work is never loaded for the full fleet.
|
||||
const lifecycle = handle.prepare(`SELECT id, server, action, state, stage, error, started_at FROM server_deployments
|
||||
WHERE server_deleted_at IS NULL AND state IN ('queued', 'running', 'waiting', 'awaiting-health')
|
||||
WHERE server_deleted_at IS NULL ${scope('server')} AND state IN ('queued', 'running', 'waiting', 'awaiting-health')
|
||||
${server === undefined ? '' : `UNION
|
||||
SELECT id, server, action, state, stage, error, started_at FROM server_deployments
|
||||
WHERE server_deleted_at IS NULL AND server = @server
|
||||
AND +state IN ('done', 'warning', 'cancelled', 'failed', 'interrupted') AND finished_at >= @since`}
|
||||
UNION
|
||||
SELECT id, server, action, state, stage, error, started_at FROM server_deployments
|
||||
WHERE server_deleted_at IS NULL AND +state IN ('done', 'warning', 'cancelled') AND finished_at >= ?
|
||||
UNION
|
||||
SELECT id, server, action, state, stage, error, started_at FROM server_deployments
|
||||
WHERE server_deleted_at IS NULL AND +state IN ('failed', 'interrupted', 'warning')
|
||||
AND rowid IN (SELECT MAX(rowid) FROM server_deployments GROUP BY server)`)
|
||||
.all(since) as { id: string; server: string; action: string; state: string; stage: string;
|
||||
WHERE server_deleted_at IS NULL ${scope('server')} AND +state IN ('failed', 'interrupted')
|
||||
AND rowid IN (SELECT MAX(rowid) FROM server_deployments WHERE 1=1 ${scope('server')} GROUP BY server)`)
|
||||
.all(bindings) as { id: string; server: string; action: string; state: string; stage: string;
|
||||
error: string | null; started_at: string }[];
|
||||
for (const row of lifecycle) {
|
||||
if (row.state === 'awaiting-health') representedRecovery.add(row.server);
|
||||
@@ -59,8 +71,8 @@ export function fleetOperations(now = Date.now()): FleetOperation[] {
|
||||
detail: row.error ?? row.stage, href: `/provisioning/${encodeURIComponent(row.id)}` });
|
||||
}
|
||||
const betas = handle.prepare(`SELECT b.id, b.opened_at, bs.server, bs.stage, bs.error
|
||||
FROM betas b JOIN beta_servers bs ON bs.beta_id = b.id WHERE bs.server_deleted_at IS NULL AND b.state IN ('running', 'ready-promote')`)
|
||||
.all() as { id: number; opened_at: string; server: string; stage: string; error: string | null }[];
|
||||
FROM betas b JOIN beta_servers bs ON bs.beta_id = b.id WHERE bs.server_deleted_at IS NULL ${scope('bs.server')} AND b.state IN ('running', 'ready-promote')`)
|
||||
.all(bindings) as { id: number; opened_at: string; server: string; stage: string; error: string | null }[];
|
||||
const stages: Record<string, string> = { queued: 'Waiting for an earlier test wave', deploying: 'Deploying beta',
|
||||
'awaiting-empty': 'Waiting for an empty lobby', restarting: 'Restarting', 'awaiting-health': 'Recovering',
|
||||
'awaiting-match': 'Observing a match', ready: 'Ready for review', failed: 'Beta needs attention' };
|
||||
@@ -71,9 +83,9 @@ export function fleetOperations(now = Date.now()): FleetOperation[] {
|
||||
state: row.stage === 'failed' ? 'attention' : 'active', label: stages[row.stage] ?? row.stage,
|
||||
detail: row.error ?? recovering.get(row.server)?.error ?? `Beta ${row.id}`, href: `/betas/${row.id}` });
|
||||
}
|
||||
const maintenance = handle.prepare(`SELECT m.* FROM maintenance_jobs m WHERE m.server_deleted_at IS NULL AND m.state = 'failed'
|
||||
const maintenance = handle.prepare(`SELECT m.* FROM maintenance_jobs m WHERE m.server_deleted_at IS NULL ${scope('m.server')} AND m.state = 'failed'
|
||||
AND NOT EXISTS (SELECT 1 FROM maintenance_jobs n WHERE n.server = m.server AND n.kind = m.kind AND n.id > m.id)`)
|
||||
.all() as { id: number; server: string; kind: string; detail: string | null; requested_at: string }[];
|
||||
.all(bindings) as { id: number; server: string; kind: string; detail: string | null; requested_at: string }[];
|
||||
for (const job of maintenance) {
|
||||
operations.push({ key: `maintenance:${job.id}`, group: `maintenance:${job.id}`, kind: 'maintenance',
|
||||
server: job.server, state: 'attention', startedAt: job.requested_at,
|
||||
|
||||
@@ -1,26 +1,32 @@
|
||||
import { db } from '../db/index.js';
|
||||
import { listServers } from './servers.js';
|
||||
import { observationReader } from './status.js';
|
||||
import { getServer, listServers } from './servers.js';
|
||||
import { observationReader, serverObservation } from './status.js';
|
||||
import { config, komodoConfigured, rconConfigured } from '../config.js';
|
||||
import { runningVersions } from '../integrations/plutonium.js';
|
||||
import { listHosts } from '../hosts/index.js';
|
||||
import type { NativeInstance } from '../backends/native.js';
|
||||
import { serverOperations } from './overview.js';
|
||||
|
||||
export function fleetSnapshot() {
|
||||
export function fleetSnapshot(serverName?: string) {
|
||||
const bindings = serverName === undefined ? [] : [serverName];
|
||||
const rconHost = config.rcon.host;
|
||||
const hosts = listHosts();
|
||||
const hostsById = new Map(hosts.map(({ inventory: _inventory, ...host }) => [host.id, host]));
|
||||
const versions = new Map(runningVersions().map(version => [version.server, version]));
|
||||
const instances = new Map((db().prepare('SELECT * FROM native_instances WHERE retired_at IS NULL').all() as NativeInstance[]).map(instance => [instance.server, instance]));
|
||||
const versions = new Map(runningVersions(serverName).map(version => [version.server, version]));
|
||||
const instances = new Map((db().prepare(`SELECT * FROM native_instances WHERE retired_at IS NULL ${serverName === undefined ? '' : 'AND server = ?'}`)
|
||||
.all(...bindings) as NativeInstance[]).map(instance => [instance.server, instance]));
|
||||
const holds = new Map<string, { target: string; betaId: number }[]>();
|
||||
for (const row of db().prepare(`SELECT bsv.server, s.target_rel_path AS target, b.id AS betaId
|
||||
FROM beta_servers bsv JOIN betas b ON b.id=bsv.beta_id JOIN beta_scripts bs ON bs.beta_id=b.id
|
||||
JOIN scripts s ON s.id=bs.script_id WHERE bsv.server_deleted_at IS NULL AND b.state IN ('running','ready-promote')`).all() as
|
||||
JOIN scripts s ON s.id=bs.script_id WHERE bsv.server_deleted_at IS NULL AND b.state IN ('running','ready-promote')
|
||||
${serverName === undefined ? '' : 'AND bsv.server = ?'}`).all(...bindings) as
|
||||
{ server: string; target: string; betaId: number }[]) {
|
||||
const entries = holds.get(row.server) ?? []; entries.push({ target: row.target, betaId: row.betaId }); holds.set(row.server, entries);
|
||||
}
|
||||
const observe = observationReader();
|
||||
const servers = listServers().map((s) => {
|
||||
const observe = serverName === undefined ? observationReader() : serverObservation;
|
||||
const selected = serverName === undefined ? undefined : getServer(serverName);
|
||||
const rows = serverName === undefined ? listServers() : selected && !selected.destroyed_at ? [selected] : [];
|
||||
const servers = rows.map((s) => {
|
||||
const instance = s.native_instance_id ? instances.get(s.name) : null;
|
||||
return {
|
||||
...observe(s),
|
||||
@@ -36,7 +42,7 @@ export function fleetSnapshot() {
|
||||
return {
|
||||
servers,
|
||||
serverNow: Date.now(),
|
||||
reservedNames: (db().prepare("SELECT j.server FROM server_creation_jobs j JOIN server_deployments d ON d.id = j.deployment_id WHERE d.state IN ('queued', 'running', 'awaiting-health')").all() as { server: string }[]).map(row => row.server),
|
||||
reservedNames: serverName === undefined ? (db().prepare("SELECT j.server FROM server_creation_jobs j JOIN server_deployments d ON d.id = j.deployment_id WHERE d.state IN ('queued', 'running', 'awaiting-health')").all() as { server: string }[]).map(row => row.server) : [],
|
||||
integrations: {
|
||||
native: hosts.length > 0 || !komodoConfigured(),
|
||||
hostsReady: hosts.filter(host => host.readiness === 'ready').length,
|
||||
@@ -49,3 +55,6 @@ export function fleetSnapshot() {
|
||||
};
|
||||
}
|
||||
|
||||
export function serverSnapshot(name: string) {
|
||||
return { ...fleetSnapshot(name), operations: serverOperations(name) };
|
||||
}
|
||||
|
||||
@@ -332,9 +332,17 @@ function transferProgress(value: unknown, status: LibraryCacheState): LibraryTra
|
||||
bytesPerSecond: progress.bytesPerSecond as number | null ?? null, etaSeconds: progress.etaSeconds as number | null ?? null };
|
||||
}
|
||||
|
||||
type LibraryCacheMetadata = Pick<GameLibraryVersion, 'state' | 'sizeBytes' | 'manifestSha256'>;
|
||||
const publishedCacheMetadata = new WeakMap<ReturnType<typeof database>, Map<string, LibraryCacheMetadata>>();
|
||||
|
||||
export function updateHostLibraryCaches(hostId: string, input: unknown, options: { snapshot?: boolean } = {}): void {
|
||||
if (!Array.isArray(input) || input.length > 2000) throw new Error('Invalid host library report.');
|
||||
const handle = database();
|
||||
// Published metadata is immutable. Keep caches local to the connection and
|
||||
// avoid retaining a publication from an outer transaction that may roll back.
|
||||
const versions = handle.inTransaction ? new Map<string, LibraryCacheMetadata>()
|
||||
: publishedCacheMetadata.get(handle) ?? new Map<string, LibraryCacheMetadata>();
|
||||
if (!handle.inTransaction) publishedCacheMetadata.set(handle, versions);
|
||||
const now = new Date().toISOString();
|
||||
handle.transaction(() => {
|
||||
const reported = new Set<string>();
|
||||
@@ -344,7 +352,10 @@ export function updateHostLibraryCaches(hostId: string, input: unknown, options:
|
||||
if (typeof id !== 'string') throw new Error('Invalid library version ID.');
|
||||
if (reported.has(id)) throw new Error('Duplicate version in host cache report.');
|
||||
reported.add(id);
|
||||
const version = getLibraryVersion(id); if (!version || version.state !== 'ready') continue;
|
||||
const version = versions.get(id) ?? handle.prepare(`SELECT state, size_bytes AS sizeBytes, manifest_sha256 AS manifestSha256
|
||||
FROM game_file_versions WHERE id=?`).get(id) as LibraryCacheMetadata | undefined;
|
||||
if (!version || version.state !== 'ready') continue;
|
||||
versions.set(id, version);
|
||||
let status = raw.status ?? raw.state;
|
||||
if (!['missing', 'downloading', 'verifying', 'ready', 'failed', 'outdated'].includes(status)) throw new Error('Invalid library cache state.');
|
||||
const checksum = typeof raw.manifestSha256 === 'string' ? raw.manifestSha256 : null;
|
||||
|
||||
@@ -6,6 +6,7 @@ import { db } from '../db/index.js';
|
||||
import { ensureHostSchema } from './schema.js';
|
||||
import { successfulObservation, routineObservationSql, importantOperationSql, operationTargetSql, safeOperationTarget } from './observations.js';
|
||||
import { currentOperationSource } from './operation-context.js';
|
||||
import { parseHostInventory } from './request-context.js';
|
||||
import { fleetSettings } from '../settings.js';
|
||||
import { recordInstanceResources, validateInstanceResources } from './calibration.js';
|
||||
import { refreshHostAvailability, scheduleAvailabilityRefresh } from './availability.js';
|
||||
@@ -38,7 +39,7 @@ function presentDecommission(row: DecommissionRow | undefined): HostDecommission
|
||||
}
|
||||
function hostFromRow(row: HostRow): Host {
|
||||
const enabling = database().prepare('SELECT requested_at FROM ec_host_enable_requests WHERE host_id=?').get(row.id) as { requested_at: string } | undefined;
|
||||
const inventory: HostInventory = JSON.parse(row.inventory);
|
||||
const inventory = parseHostInventory(row.id, row.inventory);
|
||||
const connected = !!row.credential_active && !!row.last_heartbeat && Date.now() - Date.parse(row.last_heartbeat) <= HEARTBEAT_MAX_AGE_MS;
|
||||
const reasons: string[] = [];
|
||||
if (!row.enrolled_at) reasons.push('Agent has not enrolled.');
|
||||
@@ -304,9 +305,15 @@ export function enrollHost(token: unknown, rawInventory: unknown): { hostId: str
|
||||
}
|
||||
export function authenticateCredential(token: string): Host | null {
|
||||
if (!token || token.length > 200) return null;
|
||||
const row = database().prepare('SELECT id,host_id FROM ec_host_credentials WHERE token_hash=? AND revoked_at IS NULL').get(digest(token)) as { id: string; host_id: string } | undefined;
|
||||
const row = database().prepare('SELECT id,host_id,last_used_at FROM ec_host_credentials WHERE token_hash=? AND revoked_at IS NULL')
|
||||
.get(digest(token)) as { id: string; host_id: string; last_used_at: string | null } | undefined;
|
||||
if (!row) return null;
|
||||
database().prepare('UPDATE ec_host_credentials SET last_used_at=? WHERE id=?').run(now(), row.id);
|
||||
const timestamp = Date.now();
|
||||
if (row.last_used_at === null || timestamp - Date.parse(row.last_used_at) >= 60_000) {
|
||||
database().prepare(`UPDATE ec_host_credentials SET last_used_at=? WHERE id=? AND revoked_at IS NULL
|
||||
AND (last_used_at IS NULL OR last_used_at<=?)`)
|
||||
.run(new Date(timestamp).toISOString(), row.id, new Date(timestamp - 60_000).toISOString());
|
||||
}
|
||||
return getHost(row.host_id);
|
||||
}
|
||||
export function revokeCredentials(hostId: string): void {
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
import { AsyncLocalStorage } from 'node:async_hooks';
|
||||
import type { HostInventory } from '../../hosts.js';
|
||||
|
||||
type InventoryCache = Map<string, { json: string; inventory: HostInventory }>;
|
||||
const requests = new AsyncLocalStorage<{ inventories: InventoryCache | null }>();
|
||||
|
||||
export function withHostRequestContext<T>(work: () => T | Promise<T>): Promise<T> {
|
||||
const context = { inventories: new Map() as InventoryCache | null };
|
||||
return requests.run(context, async () => {
|
||||
try { return await work(); }
|
||||
finally {
|
||||
// Detached work can inherit this context. Release request data and
|
||||
// stop caching when the response is ready, even if that work continues.
|
||||
context.inventories = null;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
export function parseHostInventory(hostId: string, json: string): HostInventory {
|
||||
const cache = requests.getStore()?.inventories;
|
||||
const saved = cache?.get(hostId);
|
||||
if (saved?.json === json) return saved.inventory;
|
||||
const inventory: HostInventory = JSON.parse(json);
|
||||
// Compare the stored JSON so heartbeats and transaction rollbacks within
|
||||
// a request remain visible. Other host state is always read afresh.
|
||||
cache?.set(hostId, { json, inventory });
|
||||
return inventory;
|
||||
}
|
||||
@@ -20,13 +20,18 @@ export function freshRelease(): number | null {
|
||||
const age = latest.checkedAt ? Date.now() - Date.parse(latest.checkedAt) : NaN;
|
||||
return !latest.error && age >= 0 && age < HOUR ? latest.revision : null;
|
||||
}
|
||||
export function runningVersions(): PlutoniumVersion[] {
|
||||
const versions = db().prepare('SELECT * FROM plutonium_versions').all() as PlutoniumVersion[];
|
||||
export function runningVersions(serverName?: string): PlutoniumVersion[] {
|
||||
const bindings = serverName === undefined ? [] : [serverName];
|
||||
const versions = db().prepare(`SELECT * FROM plutonium_versions ${serverName === undefined ? '' : 'WHERE server = ?'}`)
|
||||
.all(...bindings) as PlutoniumVersion[];
|
||||
const target = freshRelease();
|
||||
const byServer = new Map(versions.map(version => [version.server, version]));
|
||||
const epochs = latestGameSubmissions();
|
||||
const recovering = new Set((db().prepare('SELECT server FROM restart_rollout WHERE healthy_at IS NULL').all() as { server: string }[]).map(row => row.server));
|
||||
return listServers().map(server => {
|
||||
const epochs = serverName === undefined ? latestGameSubmissions() : new Map([[serverName, latestGameSubmission(serverName)]]);
|
||||
const recovering = new Set((db().prepare(`SELECT server FROM restart_rollout WHERE healthy_at IS NULL ${serverName === undefined ? '' : 'AND server = ?'}`)
|
||||
.all(...bindings) as { server: string }[]).map(row => row.server));
|
||||
const selected = serverName === undefined ? undefined : getServer(serverName);
|
||||
const servers = serverName === undefined ? listServers() : selected && !selected.destroyed_at ? [selected] : [];
|
||||
return servers.map(server => {
|
||||
const value = byServer.get(server.name)
|
||||
?? { server: server.name, revision: null, checked_at: null, error: null };
|
||||
const age = value.checked_at ? Date.now() - Date.parse(value.checked_at) : NaN;
|
||||
|
||||
@@ -7,6 +7,7 @@ import { pruneHostObservations } from '../hosts/index.js';
|
||||
interface PendingRow {
|
||||
deployment_id: string; server: string; restart_state: string;
|
||||
restarted_at: string | null; files_changed: number;
|
||||
players: number | null; error: string | null;
|
||||
}
|
||||
type SweepResult = { checked: number; restarted: number; healthy: number; held: number; failed: number };
|
||||
let sweepInFlight: Promise<SweepResult> | null = null;
|
||||
@@ -39,8 +40,10 @@ async function runSweep(): Promise<SweepResult> {
|
||||
attempts = attempts + CASE WHEN ? IS NOT NULL AND checked_at IS NOT ? THEN 1 ELSE 0 END,
|
||||
restarted_at = COALESCE(restarted_at, ?) WHERE deployment_id = ? AND server = ?`)
|
||||
.run(state, operation.checked_at, operation.players, error, operation.checked_at, operation.checked_at, restartedAt, row.deployment_id, row.server);
|
||||
if (state !== row.restart_state || operation.players !== row.players || error !== row.error) {
|
||||
emit({ type: 'server-state', deploymentId: row.deployment_id, server: row.server, state, players: operation.players, error });
|
||||
}
|
||||
}
|
||||
db().prepare(`UPDATE deployments SET finished_at = ? WHERE finished_at IS NULL AND NOT EXISTS (
|
||||
SELECT 1 FROM deployment_servers WHERE deployment_id = deployments.id AND server_deleted_at IS NULL AND restart_state IN ('pending','failed','held'))`).run(nowIso());
|
||||
db().prepare(`DELETE FROM deployments WHERE started_at < ?
|
||||
|
||||
+57
-24
@@ -6,15 +6,17 @@ import { browser } from '$app/environment';
|
||||
import { statusByServer, statusReceivedAt, syncServerTime, serverTimeNow, matchEventsByServer, telemetryByServer, uiNow } from './servers.js';
|
||||
import { type FleetServer, type FleetOperation } from '$lib/fleet-overview.js';
|
||||
|
||||
export const fleet = writable<{ servers: FleetServer[]; operations: FleetOperation[]; receivedAt: number }>({
|
||||
// Each server detail view owns a scoped snapshot; it must not replace the shared fleet.
|
||||
export function createFleetStore(serverName?: string) {
|
||||
const fleet = writable<{ servers: FleetServer[]; operations: FleetOperation[]; receivedAt: number }>({
|
||||
servers: [], operations: [], receivedAt: 0
|
||||
});
|
||||
export const fleetError = writable('');
|
||||
export const fleetRefreshing = writable(false);
|
||||
export const fleetIntegrations = writable<{ rcon: boolean; komodo: boolean; native?: boolean; hostsReady?: number; hostsTotal?: number; rconTimeoutMs: number; defaultRconHost: string } | null>(null);
|
||||
});
|
||||
const fleetError = writable('');
|
||||
const fleetRefreshing = writable(false);
|
||||
const fleetIntegrations = writable<{ rcon: boolean; komodo: boolean; native?: boolean; hostsReady?: number; hostsTotal?: number; rconTimeoutMs: number; defaultRconHost: string } | null>(null);
|
||||
|
||||
const observed = new Map<string, { server: FleetServer; status: unknown; match: unknown; report: unknown; phase: string; value: FleetServer }>();
|
||||
const freshnessPhase = derived([fleet, statusByServer, telemetryByServer, uiNow], ([snapshot, live, reports, now]) => {
|
||||
const observed = new Map<string, { server: FleetServer; status: unknown; match: unknown; report: unknown; phase: string; value: FleetServer }>();
|
||||
const freshnessPhase = derived([fleet, statusByServer, telemetryByServer, uiNow], ([snapshot, live, reports, now]) => {
|
||||
const times = snapshot.servers.flatMap(server => {
|
||||
const status = live.get(server.name);
|
||||
return [status?.lastSeenAt, status?.containerCheckedAt, status?.rconStatus?.last_seen_at, status?.telemetry?.receivedAt,
|
||||
@@ -22,8 +24,8 @@ const freshnessPhase = derived([fleet, statusByServer, telemetryByServer, uiNow]
|
||||
server.telemetry?.receivedAt, reports.get(server.name)?.receivedAt];
|
||||
});
|
||||
return times.map(time => `${freshStatusTimestamp(time ?? null, now)}:${now - Date.parse(time ?? '') >= 0 && now - Date.parse(time ?? '') <= 120_000}`).join(',');
|
||||
});
|
||||
export const fleetServers = derived([fleet, statusByServer, statusReceivedAt, matchEventsByServer, telemetryByServer, freshnessPhase], ([snapshot, live, received, matches, reports]) => {
|
||||
});
|
||||
const fleetServers = derived([fleet, statusByServer, statusReceivedAt, matchEventsByServer, telemetryByServer, freshnessPhase], ([snapshot, live, received, matches, reports]) => {
|
||||
const now = get(uiNow);
|
||||
const next = snapshot.servers.map(server => {
|
||||
const candidate = live.get(server.name);
|
||||
@@ -59,24 +61,24 @@ export const fleetServers = derived([fleet, statusByServer, statusReceivedAt, ma
|
||||
const names = new Set(snapshot.servers.map(server => server.name));
|
||||
for (const name of observed.keys()) if (!names.has(name)) observed.delete(name);
|
||||
return next;
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
let snapshotServerNow = 0;
|
||||
export function hydrateFleet(snapshot: { servers: FleetServer[]; operations: FleetOperation[]; integrations: NonNullable<ReturnType<typeof getIntegrations>>; serverNow: number }) {
|
||||
let snapshotServerNow = 0;
|
||||
function hydrateFleet(snapshot: { servers: FleetServer[]; operations: FleetOperation[]; integrations: NonNullable<ReturnType<typeof getIntegrations>>; serverNow: number }) {
|
||||
if (!browser || snapshot.serverNow < snapshotServerNow) return;
|
||||
snapshotServerNow = snapshot.serverNow;
|
||||
syncServerTime(snapshot.serverNow);
|
||||
fleetIntegrations.set(snapshot.integrations);
|
||||
fleet.set({ servers: snapshot.servers, operations: snapshot.operations, receivedAt: Date.now() - Math.max(0, serverTimeNow() - snapshot.serverNow) });
|
||||
}
|
||||
function getIntegrations() { return get(fleetIntegrations); }
|
||||
}
|
||||
function getIntegrations() { return get(fleetIntegrations); }
|
||||
|
||||
|
||||
|
||||
let inFlight: Promise<void> | null = null;
|
||||
let refreshRequested = false;
|
||||
export function refreshFleet(afterCurrent = false): Promise<void> {
|
||||
let inFlight: Promise<void> | null = null;
|
||||
let refreshRequested = false;
|
||||
function refreshFleet(afterCurrent = false): Promise<void> {
|
||||
if (!browser) return Promise.resolve();
|
||||
if (inFlight) { refreshRequested ||= afterCurrent; return inFlight; }
|
||||
inFlight = (async () => {
|
||||
@@ -87,15 +89,17 @@ export function refreshFleet(afterCurrent = false): Promise<void> {
|
||||
try {
|
||||
const results = await Promise.allSettled([
|
||||
(async () => {
|
||||
const response = await fetch('/api/servers', { signal: controller.signal });
|
||||
const response = await fetch(serverName === undefined ? '/api/servers' : `/api/servers/${encodeURIComponent(serverName)}`, { signal: controller.signal });
|
||||
if (!response.ok) throw new Error('Could not refresh servers. Retrying automatically.');
|
||||
const servers = await response.json();
|
||||
if (servers.serverNow < snapshotServerNow) return;
|
||||
snapshotServerNow = servers.serverNow;
|
||||
syncServerTime(servers.serverNow); fleetIntegrations.set(servers.integrations);
|
||||
fleet.update(value => ({ ...value, servers: servers.servers, receivedAt: requestedAt }));
|
||||
fleet.update(value => ({ ...value, servers: servers.servers, receivedAt: requestedAt,
|
||||
operations: serverName === undefined ? value.operations : servers.operations }));
|
||||
})(),
|
||||
(async () => {
|
||||
if (serverName !== undefined) return;
|
||||
const response = await fetch('/api/fleet', { signal: controller.signal });
|
||||
if (!response.ok) throw new Error('Could not refresh operations. Retrying automatically.');
|
||||
const operations = await response.json();
|
||||
@@ -117,20 +121,49 @@ export function refreshFleet(afterCurrent = false): Promise<void> {
|
||||
if (refreshRequested) { refreshRequested = false; void refreshFleet(); }
|
||||
});
|
||||
return inFlight;
|
||||
}
|
||||
}
|
||||
|
||||
let stop: (() => void) | null = null;
|
||||
export function startFleetMonitor(): () => void {
|
||||
let stop: (() => void) | null = null;
|
||||
function startFleetMonitor(): () => void {
|
||||
if (!browser || stop) return () => {};
|
||||
const stopPolling = pollVisible(() => refreshFleet(), 60_000, !get(fleet).receivedAt || Date.now() - get(fleet).receivedAt > 30_000);
|
||||
let refreshTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
let stopped = false;
|
||||
let pending = false;
|
||||
let refreshing = false;
|
||||
let lastEventRefresh = -Infinity;
|
||||
const schedule = () => {
|
||||
if (stopped || refreshing || refreshTimer !== undefined || !pending || document.visibilityState === 'hidden') return;
|
||||
refreshTimer = setTimeout(async () => {
|
||||
refreshTimer = undefined;
|
||||
refreshing = true;
|
||||
try {
|
||||
// Wait for any manual/poll refresh, then fetch evidence newer than
|
||||
// the event. Do not use refreshFleet(true)'s immediate follow-up.
|
||||
if (inFlight) await inFlight;
|
||||
if (stopped || document.visibilityState === 'hidden') return;
|
||||
pending = false;
|
||||
lastEventRefresh = Date.now();
|
||||
await refreshFleet();
|
||||
} finally {
|
||||
refreshing = false;
|
||||
schedule();
|
||||
}
|
||||
}, Math.max(250, lastEventRefresh + 2000 - Date.now()));
|
||||
};
|
||||
const changed = () => {
|
||||
if (refreshTimer || document.visibilityState === 'hidden') return;
|
||||
refreshTimer = setTimeout(() => { refreshTimer = undefined; void refreshFleet(true); }, 250);
|
||||
if (document.visibilityState === 'hidden') return;
|
||||
pending = true;
|
||||
schedule();
|
||||
};
|
||||
window.addEventListener('ec:fleet-changed', changed);
|
||||
stop = () => {
|
||||
stopped = true;
|
||||
stopPolling(); clearTimeout(refreshTimer); window.removeEventListener('ec:fleet-changed', changed); stop = null;
|
||||
};
|
||||
return stop;
|
||||
}
|
||||
return { fleet, fleetServers, fleetError, fleetRefreshing, fleetIntegrations, hydrateFleet, refreshFleet, startFleetMonitor };
|
||||
}
|
||||
|
||||
export const { fleet, fleetServers, fleetError, fleetRefreshing, fleetIntegrations, hydrateFleet, refreshFleet, startFleetMonitor } = createFleetStore();
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { writable, readonly } from 'svelte/store';
|
||||
import { writable, readonly, get } from 'svelte/store';
|
||||
import { browser } from '$app/environment';
|
||||
import type { GameTelemetry } from '$lib/telemetry.js';
|
||||
|
||||
@@ -56,13 +56,17 @@ export function serverTimeNow(): number {
|
||||
return serverClock ? serverClock.time + performance.now() - serverClock.receivedAt : Date.now();
|
||||
}
|
||||
|
||||
export function syncServerTime(serverNow?: number): void {
|
||||
function updateServerClock(serverNow?: number): void {
|
||||
if (typeof serverNow === 'number' && Number.isFinite(serverNow)) {
|
||||
// A delayed HTTP snapshot must not rewind time behind a newer stream
|
||||
// update and make that update look like it came from the future.
|
||||
const time = serverClock ? Math.max(serverNow, serverTimeNow()) : serverNow;
|
||||
serverClock = { time, receivedAt: performance.now() };
|
||||
}
|
||||
}
|
||||
|
||||
export function syncServerTime(serverNow?: number): void {
|
||||
updateServerClock(serverNow);
|
||||
// Refresh immediately: a suspended tab's interval may be well behind.
|
||||
_uiNow.set(serverTimeNow());
|
||||
}
|
||||
@@ -74,35 +78,67 @@ let source: EventSource | null = null;
|
||||
export function startServerStream(): (() => void) | undefined {
|
||||
if (!browser || source) return;
|
||||
syncServerTime();
|
||||
source = new EventSource('/api/events/servers');
|
||||
source.addEventListener('server-telemetry', (evt) => {
|
||||
const stream = source = new EventSource('/api/events/servers');
|
||||
let stopped = false;
|
||||
let frame: number | null = null;
|
||||
let pendingStatus: Map<string, ServerStatus> | null = null;
|
||||
let pendingTelemetry: Map<string, GameTelemetry> | null = null;
|
||||
let pendingMatches: Map<string, GameTelemetry> | null = null;
|
||||
let receivedAt = 0;
|
||||
const flush = () => {
|
||||
frame = null;
|
||||
if (stopped) return;
|
||||
const statuses = pendingStatus, telemetry = pendingTelemetry, matches = pendingMatches;
|
||||
const statusAt = receivedAt;
|
||||
pendingStatus = pendingTelemetry = pendingMatches = null;
|
||||
// Publish the latest clock before exposing reports, including after a
|
||||
// suspended frame. Every changed store is published at most once per frame.
|
||||
syncServerTime();
|
||||
if (statuses) { statusByServer.set(statuses); statusReceivedAt.set(statusAt); }
|
||||
if (matches) matchEventsByServer.set(matches);
|
||||
if (telemetry) telemetryByServer.set(telemetry);
|
||||
};
|
||||
const schedule = () => { if (frame === null) frame = requestAnimationFrame(flush); };
|
||||
stream.addEventListener('server-telemetry', (evt) => {
|
||||
if (stopped) return;
|
||||
const data = JSON.parse((evt as MessageEvent).data) as { entries: GameTelemetry[]; snapshot?: boolean; serverNow?: number };
|
||||
syncServerTime(data.serverNow);
|
||||
matchEventsByServer.update((previous) => {
|
||||
const next = data.snapshot ? new Map<string, GameTelemetry>() : new Map(previous);
|
||||
for (const entry of data.entries) {
|
||||
if (entry.event === 'game_start' || entry.event === 'game_end') next.set(entry.server, entry);
|
||||
updateServerClock(data.serverNow);
|
||||
// Merge into one private map per frame. Snapshots replace all earlier
|
||||
// evidence; later deltas still apply in arrival order.
|
||||
if (data.snapshot) {
|
||||
pendingTelemetry = new Map();
|
||||
pendingMatches = new Map();
|
||||
}
|
||||
return next;
|
||||
pendingTelemetry ??= new Map(get(telemetryByServer));
|
||||
for (const entry of data.entries) {
|
||||
pendingTelemetry.set(entry.server, entry);
|
||||
if (entry.event === 'game_start' || entry.event === 'game_end') {
|
||||
pendingMatches ??= new Map(get(matchEventsByServer));
|
||||
pendingMatches.set(entry.server, entry);
|
||||
}
|
||||
}
|
||||
schedule();
|
||||
});
|
||||
telemetryByServer.update((previous) => {
|
||||
const next = data.snapshot ? new Map<string, GameTelemetry>() : new Map(previous);
|
||||
for (const entry of data.entries) next.set(entry.server, entry);
|
||||
return next;
|
||||
});
|
||||
});
|
||||
source.addEventListener('server-status', (evt) => {
|
||||
stream.addEventListener('server-status', (evt) => {
|
||||
if (stopped) return;
|
||||
const data = JSON.parse((evt as MessageEvent).data) as { entries: ServerStatus[]; serverNow?: number; delta?: boolean };
|
||||
syncServerTime(data.serverNow);
|
||||
statusByServer.update(previous => {
|
||||
const m = data.delta ? new Map(previous) : new Map<string, ServerStatus>();
|
||||
for (const e of data.entries) m.set(e.server, { ...e, receivedAt: Date.now() });
|
||||
return m;
|
||||
});
|
||||
statusReceivedAt.set(Date.now());
|
||||
updateServerClock(data.serverNow);
|
||||
receivedAt = Date.now();
|
||||
if (!data.delta) pendingStatus = new Map();
|
||||
pendingStatus ??= new Map(get(statusByServer));
|
||||
for (const entry of data.entries) pendingStatus.set(entry.server, { ...entry, receivedAt });
|
||||
schedule();
|
||||
});
|
||||
if (tickInterval === null) {
|
||||
tickInterval = setInterval(() => { if (typeof document === 'undefined' || document.visibilityState !== 'hidden') _uiNow.set(serverTimeNow()); }, 1000);
|
||||
}
|
||||
return () => { source?.close(); source = null; if (tickInterval) clearInterval(tickInterval); tickInterval = null; };
|
||||
return () => {
|
||||
if (stopped) return;
|
||||
stopped = true;
|
||||
if (frame !== null) cancelAnimationFrame(frame);
|
||||
pendingStatus = pendingTelemetry = pendingMatches = null;
|
||||
stream.close(); source = null;
|
||||
if (tickInterval) clearInterval(tickInterval);
|
||||
tickInterval = null;
|
||||
};
|
||||
}
|
||||
|
||||
@@ -75,7 +75,7 @@
|
||||
$effect(() => {
|
||||
if (!sessionReady) return;
|
||||
const path = $page.url.pathname;
|
||||
if (!(path === '/' || /^\/(servers|scripts|configs|plugins)\//.test(path) || path === '/betas/new')) return;
|
||||
if (!(path === '/' || /^\/(scripts|configs|plugins)\//.test(path) || path === '/betas/new')) return;
|
||||
if ('fleetSnapshot' in $page.data && $page.data.fleetSnapshot) hydrateFleet($page.data.fleetSnapshot);
|
||||
return startFleetMonitor();
|
||||
});
|
||||
|
||||
+22
-12
@@ -1,3 +1,7 @@
|
||||
<script module lang="ts">
|
||||
const serverTitleCollator = new Intl.Collator(undefined, { numeric: true });
|
||||
</script>
|
||||
|
||||
<script lang="ts">
|
||||
let { data }: { data: import('./$types').PageData } = $props();
|
||||
const snapshot = $derived($fleet.receivedAt ? $fleet : data.fleetSnapshot);
|
||||
@@ -92,15 +96,21 @@
|
||||
let problems = $derived(servers.filter(server => server.health === 'offline' || server.health === 'stale'));
|
||||
let starting = $derived(servers.filter(server => server.health === 'starting').length);
|
||||
let provisioning = $derived(servers.filter(server => server.health === 'provisioning').length);
|
||||
let filtered = $derived(servers.filter(server => matchesFleet(server, query, filter, game, pinned, activeOperations, observationNow()))
|
||||
.sort((a, b) => {
|
||||
const pinOrder = Number(pinned.includes(b.name)) - Number(pinned.includes(a.name));
|
||||
if (pinOrder) return pinOrder;
|
||||
const rank = (server: FleetServer) => sort === 'players' ? (freshServer(server, observationNow()) ? server.last_players ?? -1 : -1)
|
||||
: sort === 'round' ? currentRound(server, $telemetryByServer.get(server.name), observationNow()) ?? -1
|
||||
: sort === 'duration' ? matchDuration(server) : 0;
|
||||
return rank(b) - rank(a) || serverTitle(a.name).localeCompare(serverTitle(b.name), undefined, { numeric: true });
|
||||
}));
|
||||
let filtered = $derived.by(() => {
|
||||
const now = observationNow();
|
||||
const pinnedNames = new Set(pinned);
|
||||
return servers.filter(server => matchesFleet(server, query, filter, game, pinned, activeOperations, now))
|
||||
.map(server => ({
|
||||
server,
|
||||
title: serverTitle(server.name),
|
||||
pinned: Number(pinnedNames.has(server.name)),
|
||||
rank: sort === 'players' ? (freshServer(server, now) ? server.last_players ?? -1 : -1)
|
||||
: sort === 'round' ? currentRound(server, $telemetryByServer.get(server.name), now) ?? -1
|
||||
: sort === 'duration' ? matchDuration(server, now) : 0
|
||||
}))
|
||||
.sort((a, b) => b.pinned - a.pinned || b.rank - a.rank || serverTitleCollator.compare(a.title, b.title))
|
||||
.map(({ server }) => server);
|
||||
});
|
||||
let groups = $derived.by(() => {
|
||||
const result = new Map<string, { title: string; game: string; servers: FleetServer[]; launch: ReturnType<typeof fleetIdentity>['launch'] }>();
|
||||
for (const server of filtered) {
|
||||
@@ -116,9 +126,9 @@
|
||||
let spotlight = $derived(servers.filter(server => currentRound(server, $telemetryByServer.get(server.name), observationNow()) !== null)
|
||||
.sort((a, b) => matchDuration(b) - matchDuration(a))[0]);
|
||||
|
||||
function matchDuration(server: FleetServer): number {
|
||||
return freshServer(server, observationNow()) && server.active_since && server.game_started_at && !server.game_ended_at
|
||||
&& (server.last_players ?? 0) > 0 ? Math.max(0, observationNow() - Date.parse(server.game_started_at)) : -1;
|
||||
function matchDuration(server: FleetServer, now = observationNow()): number {
|
||||
return freshServer(server, now) && server.active_since && server.game_started_at && !server.game_ended_at
|
||||
&& (server.last_players ?? 0) > 0 ? Math.max(0, now - Date.parse(server.game_started_at)) : -1;
|
||||
}
|
||||
function togglePin(name: string) { pinned = pinned.includes(name) ? pinned.filter(item => item !== name) : [...pinned, name]; }
|
||||
function launch(selection: { game: ZombieGame; slug: string } | null = null) {
|
||||
|
||||
@@ -2,7 +2,7 @@ import { json } from '@sveltejs/kit';
|
||||
import { heartbeat, pendingCreationCancellations } from '$lib/server/hosts/index.js';
|
||||
import { authenticateAgent, hostApiError, objectBody } from '$lib/server/hosts/http.js';
|
||||
import { reconcileHostPreparation } from '$lib/server/hosts/preparation.js';
|
||||
import { reconcileHostLibraryCleanup, cleanupControllerLibraries } from '$lib/server/game-files/index.js';
|
||||
import { reconcileHostLibraryCleanup } from '$lib/server/game-files/index.js';
|
||||
import { fleetSettings } from '$lib/server/settings.js';
|
||||
export async function POST({ request }: { request: Request }) {
|
||||
const authenticated = authenticateAgent(request);
|
||||
@@ -16,7 +16,6 @@ export async function POST({ request }: { request: Request }) {
|
||||
}
|
||||
reconcileHostPreparation(host.id);
|
||||
reconcileHostLibraryCleanup(host.id);
|
||||
await cleanupControllerLibraries();
|
||||
return json({ host, cancelledCreations: pendingCreationCancellations(host.id), heartbeatSeconds: 15,
|
||||
observationKeepDays: fleetSettings().observationKeepDays, serverTime: new Date().toISOString() });
|
||||
} catch (cause) { return hostApiError(cause); }
|
||||
|
||||
@@ -1,11 +1,14 @@
|
||||
import { observationReader } from '$lib/server/fleet/status.js';
|
||||
import { listServers } from '$lib/server/fleet/servers.js';
|
||||
import { statusEntry } from '$lib/server/pipeline/worker.js';
|
||||
import { subscribe } from '$lib/server/pipeline/bus.js';
|
||||
import { subscribe, type PipelineEvent } from '$lib/server/pipeline/bus.js';
|
||||
import { listTelemetry } from '$lib/server/fleet/telemetry.js';
|
||||
|
||||
const encoder = new TextEncoder();
|
||||
const HEARTBEAT_MS = 20_000;
|
||||
// Bus subscribers receive the same event object. Reuse its complete frame
|
||||
// across connections without retaining events after delivery.
|
||||
const frames = new WeakMap<PipelineEvent, Uint8Array>();
|
||||
|
||||
export async function GET() {
|
||||
let heartbeat: NodeJS.Timeout | null = null;
|
||||
@@ -15,7 +18,12 @@ export async function GET() {
|
||||
start(controller) {
|
||||
unsubscribe = subscribe((event) => {
|
||||
if (event.type === 'server-status' || event.type === 'server-telemetry') {
|
||||
controller.enqueue(encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify({ ...event, serverNow: Date.now() })}\n\n`));
|
||||
let frame = frames.get(event);
|
||||
if (!frame) {
|
||||
frame = encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify({ ...event, serverNow: Date.now() })}\n\n`);
|
||||
frames.set(event, frame);
|
||||
}
|
||||
controller.enqueue(frame);
|
||||
}
|
||||
});
|
||||
const observation = observationReader();
|
||||
|
||||
@@ -1,6 +1,11 @@
|
||||
import { kick } from '$lib/server/pipeline/worker.js';
|
||||
import { json } from '@sveltejs/kit';
|
||||
import { destroyServer } from '$lib/server/provisioning/index.js';
|
||||
import { serverSnapshot } from '$lib/server/fleet/snapshot.js';
|
||||
|
||||
export function GET({ params }: { params: { name: string } }) {
|
||||
return json(serverSnapshot(params.name));
|
||||
}
|
||||
|
||||
export async function DELETE({ params }: { params: { name: string } }) {
|
||||
const name = params.name;
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { fleetSnapshot } from '$lib/server/fleet/snapshot.js';
|
||||
import { fleetOperations } from '$lib/server/fleet/overview.js';
|
||||
import { serverSnapshot } from '$lib/server/fleet/snapshot.js';
|
||||
|
||||
export function load() { return { fleetSnapshot: { ...fleetSnapshot(), operations: fleetOperations() } }; }
|
||||
export function load({ params }: { params: { name: string } }) {
|
||||
return { serverSnapshot: serverSnapshot(params.name) };
|
||||
}
|
||||
|
||||
@@ -1,14 +1,12 @@
|
||||
<script lang="ts">
|
||||
let { data }: { data: import('./$types').PageData } = $props();
|
||||
const snapshot = $derived($fleet.receivedAt ? $fleet : data.fleetSnapshot);
|
||||
const visibleFleet = $derived($fleet.receivedAt ? $fleetServers : data.fleetSnapshot.servers);
|
||||
import RuntimeVersion from '$ui/components/RuntimeVersion.svelte';
|
||||
import MatchHistory from '$ui/components/MatchHistory.svelte';
|
||||
import RoundTelemetry from '$ui/components/RoundTelemetry.svelte';
|
||||
import TelemetryDiagnostics from '$ui/components/TelemetryDiagnostics.svelte';
|
||||
import { pollVisible } from '$ui/poll.js';
|
||||
import { fetchCampaigns } from '$ui/campaigns.js';
|
||||
import { fleet, fleetServers, fleetError, fleetIntegrations, refreshFleet } from '$ui/stores/fleet.js';
|
||||
import { createFleetStore } from '$ui/stores/fleet.js';
|
||||
import { freshServer, fleetIdentity } from '$lib/fleet-overview.js';
|
||||
import { serverTitle } from '$ui/server-labels.js';
|
||||
import ServerIdentity from '$ui/components/ServerIdentity.svelte';
|
||||
@@ -109,9 +107,17 @@
|
||||
}
|
||||
|
||||
let name = $derived($page.params.name ?? '');
|
||||
const detail = $derived(createFleetStore(name));
|
||||
const { fleet, fleetServers, fleetError, fleetIntegrations, refreshFleet } = $derived(detail);
|
||||
$effect(() => {
|
||||
detail.hydrateFleet(data.serverSnapshot);
|
||||
return detail.startFleetMonitor();
|
||||
});
|
||||
const snapshot = $derived($fleet.receivedAt ? $fleet : data.serverSnapshot);
|
||||
const visibleFleet = $derived($fleet.receivedAt ? $fleetServers : data.serverSnapshot.servers);
|
||||
let server = $derived((visibleFleet.find(s => s.name === name) as Server | undefined) ?? null);
|
||||
const native = $derived(!!server?.native_instance_id);
|
||||
let integrations = $derived($fleetIntegrations ?? data.fleetSnapshot.integrations);
|
||||
let integrations = $derived($fleetIntegrations ?? data.serverSnapshot.integrations);
|
||||
let history = $state<DeploymentRow[]>([]);
|
||||
let inventory = $state<InventoryEntry[]>([]);
|
||||
let komodoProbe = $state<KomodoProbeResult | null>(null);
|
||||
@@ -610,7 +616,7 @@
|
||||
</div>
|
||||
{:else if $fleetError}
|
||||
<div class="ui-error" role="alert">{$fleetError}<Button onclick={() => refreshFleet(true)}>Retry</Button></div>
|
||||
{:else if $fleet.receivedAt || data.fleetSnapshot}
|
||||
{:else if $fleet.receivedAt || data.serverSnapshot}
|
||||
<EmptyState title="Server not found" description="This server may have been removed from the fleet." />
|
||||
{:else}
|
||||
<div class="loading">Loading…</div>
|
||||
|
||||
@@ -32,7 +32,7 @@ import urllib.parse
|
||||
import urllib.request
|
||||
import uuid
|
||||
|
||||
VERSION = "1.4.10"
|
||||
VERSION = "1.4.11"
|
||||
LABEL = "io.electric-cherry"
|
||||
MAX_FILE = 32 * 1024 * 1024
|
||||
MANIFEST_MAX_ENTRIES = 10000
|
||||
@@ -918,6 +918,8 @@ class Agent:
|
||||
self.observation_keep_days = 0 # Older controllers do not authorize cleanup.
|
||||
self.library_progress = {}
|
||||
self.library_transfers = {}
|
||||
self.inventory_manifest_cache = {}
|
||||
self.inventory_manifest_lock = threading.Lock()
|
||||
self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
|
||||
self.executing_id = None
|
||||
|
||||
@@ -1191,6 +1193,25 @@ class Agent:
|
||||
continue
|
||||
return samples
|
||||
|
||||
def inventory_manifest(self, marker):
|
||||
# Inventory only: launch/download verification must still read disk.
|
||||
# Include file identity so atomic replacements invalidate even when
|
||||
# their size and modification time match the previous manifest.
|
||||
with self.inventory_manifest_lock:
|
||||
try:
|
||||
metadata = marker.stat()
|
||||
signature = (metadata.st_mtime_ns, metadata.st_size, metadata.st_dev, metadata.st_ino)
|
||||
cached = self.inventory_manifest_cache.get(marker)
|
||||
if cached is not None and cached[0] == signature:
|
||||
return cached[1]
|
||||
self.inventory_manifest_cache.pop(marker, None)
|
||||
value = json.loads(marker.read_text())
|
||||
self.inventory_manifest_cache[marker] = (signature, value)
|
||||
return value
|
||||
except (OSError, ValueError):
|
||||
self.inventory_manifest_cache.pop(marker, None)
|
||||
raise
|
||||
|
||||
def inventory(self):
|
||||
cpu_usage = self.cpu_usage_percent()
|
||||
usage = shutil.disk_usage(self.root)
|
||||
@@ -1210,13 +1231,18 @@ class Agent:
|
||||
libraries, caches = [], []
|
||||
progress = {key: value.copy() for key, value in list(self.library_progress.items())}
|
||||
progress.update({key: transfer.snapshot() for key, transfer in list(self.library_transfers.items())})
|
||||
for marker in (self.root / "libraries").glob("*/*/manifest.json"):
|
||||
manifests = list((self.root / "libraries").glob("*/*/manifest.json"))
|
||||
partial_manifests = list((self.root / "libraries").glob("*/.*.partial/source.json"))
|
||||
with self.inventory_manifest_lock:
|
||||
for marker in self.inventory_manifest_cache.keys() - set(manifests + partial_manifests):
|
||||
del self.inventory_manifest_cache[marker]
|
||||
for marker in manifests:
|
||||
# A crash between writing a manifest and its atomic promotion must
|
||||
# never expose the staging directory as a ready library.
|
||||
if marker.parent.name.startswith("."):
|
||||
continue
|
||||
try:
|
||||
value = json.loads(marker.read_text())
|
||||
value = self.inventory_manifest(marker)
|
||||
# Rechecking an existing cache must revoke readiness until its
|
||||
# entire contents have passed verification again.
|
||||
if value["libraryId"] not in progress and (marker.parent / "transfer.json").exists():
|
||||
@@ -1233,9 +1259,9 @@ class Agent:
|
||||
except (OSError, ValueError, KeyError):
|
||||
continue
|
||||
ready = {item["id"] for item in libraries}
|
||||
for marker in (self.root / "libraries").glob("*/.*.partial/source.json"):
|
||||
for marker in partial_manifests:
|
||||
try:
|
||||
value = json.loads(marker.read_text())
|
||||
value = self.inventory_manifest(marker)
|
||||
library_id = identifier(value["libraryId"])
|
||||
if library_id in ready or library_id in progress:
|
||||
continue
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
{
|
||||
"version": "1.4.10",
|
||||
"version": "1.4.11",
|
||||
"protocolVersion": 1,
|
||||
"platform": "debian-amd64",
|
||||
"sha256": "c518a216ad4e9dd86ac82743b27bae54cdf1c51ef7ae3bf8d252dcc9863be1fb",
|
||||
"sha256": "4c6cc180f3c45a436878a0278d62b6840526150e039fdf51bafadef63759b131",
|
||||
"updaterProtocolVersion": 1,
|
||||
"updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8"
|
||||
}
|
||||
|
||||
@@ -48,7 +48,11 @@ beforeEach(async () => {
|
||||
vi.mocked(komodo.komodoUpdateTagsOnResource).mockResolvedValue({ ok: true });
|
||||
const rcon = await import('../../src/lib/server/integrations/rcon.js');
|
||||
vi.mocked(rcon.rconGetPlayerCount).mockResolvedValue(0);
|
||||
vi.mocked(rcon.rconPlutoniumVersion).mockResolvedValue({ revision: 5354, error: null });
|
||||
vi.mocked(rcon.rconPlutoniumVersion).mockImplementation(async () => {
|
||||
// Version evidence must be observed strictly after the launch timestamp.
|
||||
await new Promise(resolve => setTimeout(resolve, 2));
|
||||
return { revision: 5354, error: null };
|
||||
});
|
||||
vi.mocked(rcon.rconStatusResult).mockImplementation(async server => status(server));
|
||||
const { kick } = await import('../../src/lib/server/pipeline/worker.js');
|
||||
vi.mocked(kick).mockResolvedValue(undefined);
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import { newFixture, type Fixture } from '../helpers/fixture.js';
|
||||
|
||||
let fixture: Fixture;
|
||||
beforeEach(async () => {
|
||||
vi.resetModules();
|
||||
fixture = await newFixture([]);
|
||||
});
|
||||
afterEach(async () => { await fixture.cleanup(); });
|
||||
|
||||
describe('prepared statement caching', () => {
|
||||
it('reuses statements with fresh parameters and remains usable after rollback', async () => {
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
const insertSql = 'INSERT INTO settings (key, value) VALUES (?, ?)';
|
||||
const selectSql = 'SELECT value FROM settings WHERE key = ?';
|
||||
const insert = db().prepare(insertSql);
|
||||
const select = db().prepare(selectSql);
|
||||
insert.run('cache-first', 'one');
|
||||
expect(() => db().transaction(() => {
|
||||
db().prepare(insertSql).run('cache-rollback', 'discarded');
|
||||
throw new Error('rollback');
|
||||
})()).toThrow('rollback');
|
||||
db().prepare(insertSql).run('cache-second', 'two');
|
||||
|
||||
expect(db().prepare(insertSql)).toBe(insert);
|
||||
expect(db().prepare(selectSql)).toBe(select);
|
||||
expect(select.get('cache-first')).toEqual({ value: 'one' });
|
||||
expect(select.get('cache-second')).toEqual({ value: 'two' });
|
||||
expect(select.get('cache-rollback')).toBeUndefined();
|
||||
});
|
||||
|
||||
it('prepares fresh statements when the connection is reopened', async () => {
|
||||
const { db, closeDb } = await import('../../src/lib/server/db/index.js');
|
||||
const sql = 'SELECT value FROM settings WHERE key = ?';
|
||||
const oldStatement = db().prepare(sql);
|
||||
db().prepare('INSERT INTO settings (key, value) VALUES (?, ?)').run('cache-persisted', 'saved');
|
||||
closeDb();
|
||||
|
||||
const statement = db().prepare(sql);
|
||||
expect(statement).not.toBe(oldStatement);
|
||||
expect(statement.database).not.toBe(oldStatement.database);
|
||||
expect(statement.get('cache-persisted')).toEqual({ value: 'saved' });
|
||||
expect(db().prepare(sql)).toBe(statement);
|
||||
});
|
||||
});
|
||||
@@ -15,7 +15,11 @@ vi.mock('../../src/lib/server/integrations/komodo.js', () => ({
|
||||
}));
|
||||
vi.mock('../../src/lib/server/integrations/rcon.js', () => ({
|
||||
rconGetPlayerCount: vi.fn(async () => 0),
|
||||
rconPlutoniumVersion: vi.fn(async () => ({ revision: 5354, error: null })),
|
||||
rconPlutoniumVersion: vi.fn(async () => {
|
||||
// Version evidence must be observed strictly after the launch timestamp.
|
||||
await new Promise(resolve => setTimeout(resolve, 2));
|
||||
return { revision: 5354, error: null };
|
||||
}),
|
||||
rconStatusResult: vi.fn(async (server: string) => ({ ok: true, endpoint: { host: 'localhost', port: 28960 }, rawReply: '',
|
||||
status: { players: 0, map: server.startsWith('t6_') ? 'zm_transit' : server.includes('_moon_') ? 'zombie_moon' : 'zombie_theater', gametype: null, roster: [] } }))
|
||||
}));
|
||||
|
||||
@@ -0,0 +1,158 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import { get } from 'svelte/store';
|
||||
import type { FleetOperation } from '../../src/lib/fleet-overview.js';
|
||||
|
||||
vi.mock('$app/environment', () => ({ browser: true }));
|
||||
|
||||
let documentStub: EventTarget & { visibilityState: string };
|
||||
let stop: (() => void) | undefined;
|
||||
const changed = () => window.dispatchEvent(new Event('ec:fleet-changed'));
|
||||
const visibility = (state: string) => {
|
||||
documentStub.visibilityState = state;
|
||||
documentStub.dispatchEvent(new Event('visibilitychange'));
|
||||
};
|
||||
const response = (url: string) => Response.json(url === '/api/servers'
|
||||
? { serverNow: Date.now(), servers: [], integrations: {} } : { operations: [] });
|
||||
|
||||
beforeEach(() => {
|
||||
vi.resetModules();
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(new Date('2026-10-05T12:00:00Z'));
|
||||
documentStub = Object.assign(new EventTarget(), { visibilityState: 'visible' });
|
||||
vi.stubGlobal('document', documentStub);
|
||||
vi.stubGlobal('window', new EventTarget());
|
||||
vi.stubGlobal('fetch', vi.fn(async (url: string) => response(url)));
|
||||
});
|
||||
afterEach(() => {
|
||||
stop?.(); stop = undefined;
|
||||
vi.useRealTimers(); vi.unstubAllGlobals(); vi.restoreAllMocks();
|
||||
});
|
||||
async function setup() {
|
||||
const store = await import('../../src/lib/ui/stores/fleet.js');
|
||||
store.fleet.set({ servers: [], operations: [], receivedAt: Date.now() });
|
||||
stop = store.startFleetMonitor();
|
||||
return store;
|
||||
}
|
||||
|
||||
describe('fleet event refreshes', () => {
|
||||
it('refreshes server history through one scoped request without replacing the fleet snapshot', async () => {
|
||||
const store = await import('../../src/lib/ui/stores/fleet.js');
|
||||
const integrations = { rcon: true, komodo: true, rconTimeoutMs: 2000, defaultRconHost: '' };
|
||||
const snapshot = { servers: [], operations: [], integrations, serverNow: Date.now() };
|
||||
store.hydrateFleet(snapshot);
|
||||
const originalFleet = get(store.fleet);
|
||||
const detail = store.createFleetStore('one / two');
|
||||
detail.hydrateFleet(snapshot);
|
||||
stop = detail.startFleetMonitor();
|
||||
const history: FleetOperation[] = [{ key: 'deployment:done', group: 'deployment:done', kind: 'deployment',
|
||||
server: 'one / two', state: 'done', label: 'Complete', detail: 'Complete', href: '/provisioning/done', startedAt: new Date().toISOString() }];
|
||||
vi.mocked(fetch).mockResolvedValue(Response.json({ ...snapshot, operations: history }));
|
||||
changed();
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
expect(fetch).toHaveBeenCalledTimes(1);
|
||||
expect(fetch).toHaveBeenCalledWith('/api/servers/one%20%2F%20two', expect.anything());
|
||||
expect(get(detail.fleet).operations).toEqual(history);
|
||||
expect(get(store.fleet)).toBe(originalFleet);
|
||||
expect(get(store.fleet).operations).toEqual([]);
|
||||
});
|
||||
|
||||
it('isolates late responses when navigating between server detail stores', async () => {
|
||||
const { createFleetStore, fleet } = await import('../../src/lib/ui/stores/fleet.js');
|
||||
const first = createFleetStore('first');
|
||||
const second = createFleetStore('second');
|
||||
let finish!: (response: Response) => void;
|
||||
vi.mocked(fetch).mockImplementationOnce(() => new Promise(resolve => { finish = resolve; }));
|
||||
const oldRequest = first.refreshFleet();
|
||||
const newSnapshot = { serverNow: Date.now(), servers: [], integrations: {}, operations: [] };
|
||||
vi.mocked(fetch).mockResolvedValue(Response.json(newSnapshot));
|
||||
await second.refreshFleet();
|
||||
const current = get(second.fleet);
|
||||
finish(Response.json(newSnapshot)); await oldRequest;
|
||||
expect(get(second.fleet)).toBe(current);
|
||||
expect(get(fleet).receivedAt).toBe(0);
|
||||
expect(fetch).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('coalesces bursts and refreshes at most once every two seconds with a trailing refresh', async () => {
|
||||
await setup();
|
||||
changed(); changed(); changed();
|
||||
await vi.advanceTimersByTimeAsync(249);
|
||||
expect(fetch).not.toHaveBeenCalled();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(fetch).toHaveBeenCalledTimes(2);
|
||||
for (let i = 0; i < 7; i++) {
|
||||
changed();
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
expect(fetch).toHaveBeenCalledTimes(2);
|
||||
}
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
expect(fetch).toHaveBeenCalledTimes(4);
|
||||
changed();
|
||||
await vi.advanceTimersByTimeAsync(2000);
|
||||
expect(fetch).toHaveBeenCalledTimes(6);
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
expect(fetch).toHaveBeenCalledTimes(6);
|
||||
});
|
||||
|
||||
it('keeps events during a request without an immediate follow-up when it finishes', async () => {
|
||||
await setup();
|
||||
let finish!: () => void;
|
||||
const gate = new Promise<void>(resolve => { finish = resolve; });
|
||||
vi.mocked(fetch).mockImplementation(async url => { await gate; return response(String(url)); });
|
||||
changed();
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
expect(fetch).toHaveBeenCalledTimes(2);
|
||||
changed(); changed();
|
||||
await vi.advanceTimersByTimeAsync(500);
|
||||
finish();
|
||||
await vi.advanceTimersByTimeAsync(1499);
|
||||
expect(fetch).toHaveBeenCalledTimes(2);
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(fetch).toHaveBeenCalledTimes(4);
|
||||
});
|
||||
|
||||
it('waits for a slow manual refresh and coalesces events arriving during it', async () => {
|
||||
const { refreshFleet } = await setup();
|
||||
let finish!: () => void;
|
||||
const gate = new Promise<void>(resolve => { finish = resolve; });
|
||||
vi.mocked(fetch).mockImplementation(async url => { await gate; return response(String(url)); });
|
||||
const manual = refreshFleet();
|
||||
changed();
|
||||
await vi.advanceTimersByTimeAsync(3000);
|
||||
changed(); changed();
|
||||
expect(fetch).toHaveBeenCalledTimes(2);
|
||||
finish(); await manual;
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(fetch).toHaveBeenCalledTimes(4);
|
||||
await vi.advanceTimersByTimeAsync(3000);
|
||||
expect(fetch).toHaveBeenCalledTimes(4);
|
||||
});
|
||||
|
||||
it('skips queued refreshes while hidden, refreshes on return, and cancels on stop', async () => {
|
||||
await setup();
|
||||
changed(); visibility('hidden');
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
changed();
|
||||
await vi.advanceTimersByTimeAsync(3000);
|
||||
expect(fetch).not.toHaveBeenCalled();
|
||||
visibility('visible');
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(fetch).toHaveBeenCalledTimes(2);
|
||||
changed(); stop!();
|
||||
await vi.advanceTimersByTimeAsync(3000);
|
||||
expect(fetch).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('cancels a follow-up waiting for an active request when stopped', async () => {
|
||||
const { refreshFleet } = await setup();
|
||||
let finish!: () => void;
|
||||
const gate = new Promise<void>(resolve => { finish = resolve; });
|
||||
vi.mocked(fetch).mockImplementation(async url => { await gate; return response(String(url)); });
|
||||
const manual = refreshFleet();
|
||||
changed();
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
stop!(); finish(); await manual;
|
||||
await vi.advanceTimersByTimeAsync(3000);
|
||||
expect(fetch).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
});
|
||||
@@ -64,8 +64,10 @@ describe('persisted Fleet operations', () => {
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
const handle = db();
|
||||
handle.prepare("INSERT INTO maintenance_jobs(server,kind,state,target_revision,requested_at,detail) VALUES ('server','update','failed',5354,?,'Revision not verified')").run(iso());
|
||||
const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js');
|
||||
const { fleetOperations, serverOperations } = await import('../../src/lib/server/fleet/overview.js');
|
||||
expect(fleetOperations(now)).toEqual([expect.objectContaining({ kind: 'maintenance', state: 'attention', detail: 'Revision not verified' })]);
|
||||
expect(serverOperations('server', now)).toEqual(fleetOperations(now));
|
||||
expect(serverOperations('unrelated', now)).toEqual([]);
|
||||
handle.prepare("INSERT INTO maintenance_jobs(server,kind,state,target_revision,requested_at) VALUES ('server','update','queued',5354,?)").run(iso(1000));
|
||||
expect(fleetOperations(now)).toEqual([]);
|
||||
});
|
||||
@@ -76,7 +78,7 @@ describe('persisted Fleet operations', () => {
|
||||
handle.prepare("INSERT INTO servers (name, komodo_name) VALUES ('server', 'server')").run();
|
||||
handle.prepare("INSERT INTO deployments (id, action, started_at) VALUES ('one', 'promote', ?)").run(iso(-2 * 86_400_000));
|
||||
handle.prepare("INSERT INTO deployment_servers (deployment_id, server, files_changed, restart_state, players) VALUES ('one', 'server', 1, 'pending', 4)").run();
|
||||
const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js');
|
||||
const { fleetOperations, serverOperations } = await import('../../src/lib/server/fleet/overview.js');
|
||||
expect(fleetOperations(now)).toEqual([expect.objectContaining({ label: 'Waiting for an empty lobby', state: 'active', href: '/campaigns/one' })]);
|
||||
handle.prepare("UPDATE deployment_servers SET restarted_at = ?, players = NULL").run(iso());
|
||||
handle.prepare("INSERT INTO restart_rollout (server, requested_at) VALUES ('server', ?)").run(iso());
|
||||
@@ -86,18 +88,20 @@ describe('persisted Fleet operations', () => {
|
||||
expect(fleetOperations(now)[0].label).toBe('Recovering');
|
||||
db().prepare("UPDATE deployment_servers SET restart_state = 'done', checked_at = ?").run(iso());
|
||||
db().prepare('UPDATE restart_rollout SET healthy_at = ?').run(iso());
|
||||
expect(fleetOperations(now)[0].state).toBe('done');
|
||||
expect(fleetOperations(now)).toEqual([]);
|
||||
expect(serverOperations('server', now)[0].state).toBe('done');
|
||||
});
|
||||
|
||||
it('keeps lifecycle failures actionable until superseded and retains successful completions', async () => {
|
||||
it('keeps lifecycle failures actionable until superseded and retains completions only for that server', async () => {
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
const handle = db();
|
||||
const insert = handle.prepare("INSERT INTO server_deployments (id, server, action, state, stage, started_at, finished_at, triggered_by) VALUES (?, ?, ?, ?, ?, ?, ?, 'test')");
|
||||
insert.run('failed', 'server', 'create', 'failed', 'Creating container', iso(-86_400_001), iso(-86_400_001));
|
||||
const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js');
|
||||
const { fleetOperations, serverOperations } = await import('../../src/lib/server/fleet/overview.js');
|
||||
expect(fleetOperations(now)[0].state).toBe('attention');
|
||||
insert.run('retry', 'server', 'create', 'done', 'Ready', iso(), iso());
|
||||
expect(fleetOperations(now)).toEqual([expect.objectContaining({ key: 'deployment:retry', state: 'done' })]);
|
||||
expect(fleetOperations(now)).toEqual([]);
|
||||
expect(serverOperations('server', now)).toEqual([expect.objectContaining({ key: 'deployment:retry', state: 'done' })]);
|
||||
insert.run('queued', 'new-server', 'create', 'queued', 'Waiting to launch', iso(), null);
|
||||
expect(fleetOperations(now)).toContainEqual(expect.objectContaining({ server: 'new-server', label: 'Waiting to launch', state: 'active' }));
|
||||
});
|
||||
@@ -107,7 +111,7 @@ describe('persisted Fleet operations', () => {
|
||||
const handle = db();
|
||||
handle.prepare("INSERT INTO servers (name, komodo_name) VALUES ('server', 'server')").run();
|
||||
handle.prepare("INSERT INTO restart_rollout (server, requested_at, error) VALUES ('server', ?, 'RCON not answering')").run(iso());
|
||||
const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js');
|
||||
const { fleetOperations, serverOperations } = await import('../../src/lib/server/fleet/overview.js');
|
||||
expect(fleetOperations(now)[0]).toMatchObject({ kind: 'recovery', state: 'attention', detail: 'RCON not answering' });
|
||||
handle.prepare("INSERT INTO scripts (id, name, target_rel_path) VALUES (1, 'test', 'test.gsc')").run();
|
||||
handle.prepare("INSERT INTO script_versions (id, script_id, version, sha256, size, content) VALUES (1, 1, 1, 'hash', 0, X'')").run();
|
||||
@@ -115,9 +119,11 @@ describe('persisted Fleet operations', () => {
|
||||
handle.prepare("INSERT INTO beta_servers (beta_id, server, had_original, stage) VALUES (1, 'server', 0, 'awaiting-match')").run();
|
||||
expect(fleetOperations(now)).toContainEqual(expect.objectContaining({ kind: 'beta', label: 'Observing a match', href: '/betas/1' }));
|
||||
expect(fleetOperations(now)).toContainEqual(expect.objectContaining({ kind: 'recovery', state: 'attention' }));
|
||||
expect(serverOperations('server', now)).toEqual(fleetOperations(now));
|
||||
expect(serverOperations('unrelated', now)).toEqual([]);
|
||||
});
|
||||
|
||||
it('preserves lifecycle selection across cutoff, deletion, supersession, and overlapping warning branches', async () => {
|
||||
it('keeps only active work and unresolved failures across deletion and supersession', async () => {
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
const handle = db();
|
||||
const since = iso(-86_400_000);
|
||||
@@ -138,13 +144,12 @@ describe('persisted Fleet operations', () => {
|
||||
}
|
||||
}
|
||||
})();
|
||||
// The former OR query is the compatibility reference for all three branches.
|
||||
// Completed work, including warnings and cancellations, belongs to server history.
|
||||
const expected = handle.prepare(`SELECT d.id FROM server_deployments d
|
||||
WHERE d.server_deleted_at IS NULL AND (d.state IN ('queued', 'running', 'waiting', 'awaiting-health')
|
||||
OR (d.state IN ('done', 'warning', 'cancelled') AND d.finished_at >= ?)
|
||||
OR (d.state IN ('failed', 'interrupted', 'warning') AND NOT EXISTS
|
||||
OR (d.state IN ('failed', 'interrupted') AND NOT EXISTS
|
||||
(SELECT 1 FROM server_deployments n WHERE n.server = d.server AND n.rowid > d.rowid)))`)
|
||||
.all(since) as { id: string }[];
|
||||
.all() as { id: string }[];
|
||||
const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js');
|
||||
expect(fleetOperations(now).map(row => row.key).sort()).toEqual(expected.map(row => `deployment:${row.id}`).sort());
|
||||
});
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import { newFixture, type Fixture } from '../helpers/fixture.js';
|
||||
|
||||
let fixture: Fixture;
|
||||
const now = Date.parse('2026-10-05T12:00:00Z');
|
||||
const iso = (offset = 0) => new Date(now + offset).toISOString();
|
||||
beforeEach(async () => {
|
||||
vi.resetModules();
|
||||
vi.useFakeTimers({ toFake: ['Date'] }); vi.setSystemTime(now);
|
||||
fixture = await newFixture([]);
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
db().transaction(() => {
|
||||
const server = db().prepare('INSERT INTO servers(name, komodo_name) VALUES (?, ?)');
|
||||
const operation = db().prepare(`INSERT INTO server_deployments(id,server,action,state,stage,started_at,finished_at,triggered_by)
|
||||
VALUES(?,?,'restart',?,'Complete',?,?,'test')`);
|
||||
for (const name of ['one', 'two']) {
|
||||
server.run(name, name);
|
||||
for (let i = 0; i < 100; i++) operation.run(`${name}-${i}`, name, 'done', iso(-i), iso(-i));
|
||||
operation.run(`${name}-old`, name, 'done', iso(-86_400_001), iso(-86_400_001));
|
||||
operation.run(`${name}-failed`, name, 'failed', iso(), iso());
|
||||
operation.run(`${name}-active`, name, 'waiting', iso(), null);
|
||||
}
|
||||
db().prepare("INSERT INTO deployments(id,action,started_at,finished_at) VALUES('campaign','push',?,?)").run(iso(), iso());
|
||||
for (const name of ['one', 'two']) {
|
||||
db().prepare(`INSERT INTO deployment_servers(deployment_id,server,files_changed,restart_state,checked_at)
|
||||
VALUES('campaign',?,1,'done',?)`).run(name, iso());
|
||||
}
|
||||
})();
|
||||
});
|
||||
afterEach(async () => { await fixture.cleanup(); vi.useRealTimers(); vi.restoreAllMocks(); });
|
||||
|
||||
describe('fleet and server response scope', () => {
|
||||
it('omits completed history from fleet REST and SSR responses', async () => {
|
||||
const { GET } = await import('../../src/routes/api/fleet/+server.js');
|
||||
const { load } = await import('../../src/routes/+page.server.js');
|
||||
const body = await GET().json();
|
||||
expect(body.operations.map((operation: { key: string }) => operation.key).sort())
|
||||
.toEqual(['deployment:one-active', 'deployment:two-active']);
|
||||
expect(load().fleetSnapshot.operations).toEqual(body.operations);
|
||||
});
|
||||
|
||||
it('loads only the selected server and its recent history in both REST and SSR', async () => {
|
||||
const { GET } = await import('../../src/routes/api/servers/[name]/+server.js');
|
||||
const { load } = await import('../../src/routes/servers/[name]/+page.server.js');
|
||||
const registry = await import('../../src/lib/server/fleet/servers.js');
|
||||
const { fleetSnapshot } = await import('../../src/lib/server/fleet/snapshot.js');
|
||||
const expected = fleetSnapshot().servers.find(server => server.name === 'one');
|
||||
const list = vi.spyOn(registry, 'listServers');
|
||||
const snapshot = load({ params: { name: 'one' } }).serverSnapshot;
|
||||
expect(list).not.toHaveBeenCalled();
|
||||
expect(snapshot.servers).toEqual([expected]);
|
||||
expect(snapshot.operations).toHaveLength(103);
|
||||
expect(snapshot.operations.every(operation => operation.server === 'one')).toBe(true);
|
||||
expect(snapshot.operations.map(operation => operation.key)).not.toContain('deployment:one-old');
|
||||
expect(snapshot.operations).toContainEqual(expect.objectContaining({ key: 'campaign:campaign:one', state: 'done' }));
|
||||
expect(snapshot.operations).toContainEqual(expect.objectContaining({ key: 'deployment:one-failed', state: 'attention' }));
|
||||
const body = await GET({ params: { name: 'one' } }).json();
|
||||
expect(body).toEqual(JSON.parse(JSON.stringify(snapshot)));
|
||||
expect(list).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('returns no unrelated servers or history for missing and removed servers', async () => {
|
||||
const { serverSnapshot } = await import('../../src/lib/server/fleet/snapshot.js');
|
||||
expect(serverSnapshot('missing')).toMatchObject({ servers: [], operations: [] });
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
db().prepare("UPDATE servers SET destroyed_at = ? WHERE name = 'one'").run(iso());
|
||||
expect(serverSnapshot('one').servers).toEqual([]);
|
||||
db().prepare("UPDATE server_deployments SET server_deleted_at = ? WHERE server = 'one'").run(iso());
|
||||
db().prepare("UPDATE deployment_servers SET server_deleted_at = ? WHERE server = 'one'").run(iso());
|
||||
expect(serverSnapshot('one').operations).toEqual([]);
|
||||
});
|
||||
|
||||
it('uses a server index for recent lifecycle history', async () => {
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
const { serverOperations } = await import('../../src/lib/server/fleet/overview.js');
|
||||
const prepare = vi.spyOn(db(), 'prepare');
|
||||
serverOperations('one', now);
|
||||
const sql = prepare.mock.calls.map(([sql]) => sql).find(sql => sql.includes('FROM server_deployments'))!;
|
||||
prepare.mockRestore();
|
||||
const plan = db().prepare(`EXPLAIN QUERY PLAN ${sql}`).all({ server: 'one', since: iso(-86_400_000) }) as { detail: string }[];
|
||||
const details = plan.map(row => row.detail).join('\n');
|
||||
expect(details).toMatch(/SEARCH server_deployments USING (?:COVERING )?INDEX idx_server_deployments_server/);
|
||||
expect(details).not.toMatch(/SCAN server_deployments\b/);
|
||||
});
|
||||
});
|
||||
@@ -145,6 +145,79 @@ describe('game-file import lifecycle', () => {
|
||||
});
|
||||
});
|
||||
describe('verified host distribution', () => {
|
||||
it('reuses published metadata across host reports without reading or parsing a large manifest', () => {
|
||||
const version = begin();
|
||||
const manifest = JSON.stringify(Array.from({ length: 20_000 }, (_, i) => ({ ...file, path: `zone/file-${i}.ff` })));
|
||||
const size = file.size * 20_000;
|
||||
state.handle!.prepare("UPDATE game_file_versions SET state='ready',size_bytes=?,manifest=?,manifest_sha256='large-checksum' WHERE id=?")
|
||||
.run(size, manifest, version.id);
|
||||
const otherHost = createHost({ name: 'Other reporting host' });
|
||||
const parse = JSON.parse;
|
||||
const parsing = vi.spyOn(JSON, 'parse').mockImplementation((value, reviver) => {
|
||||
if (value === manifest) throw new Error('Cache reports must not deserialize manifests.');
|
||||
return parse(value, reviver);
|
||||
});
|
||||
const preparing = vi.spyOn(state.handle!, 'prepare');
|
||||
try {
|
||||
for (const id of [hostId, otherHost.id, hostId]) {
|
||||
updateHostLibraryCaches(id, [{ libraryVersionId: version.id, status: 'ready', manifestSha256: 'large-checksum' }]);
|
||||
expect(getHostLibraryCache(id, version.id)).toMatchObject({ status: 'ready', bytesDownloaded: size, bytesTotal: size });
|
||||
}
|
||||
const reads = preparing.mock.calls.map(([sql]) => sql).filter(sql => sql.includes('FROM game_file_versions'));
|
||||
expect(reads).toHaveLength(1);
|
||||
expect(state.handle!.prepare(reads[0]).columns().map(column => column.name)).toEqual(['state', 'sizeBytes', 'manifestSha256']);
|
||||
updateHostLibraryCaches(hostId, [{ libraryVersionId: version.id, status: 'ready', manifestSha256: 'wrong' }]);
|
||||
expect(getHostLibraryCache(hostId, version.id)?.status).toBe('failed');
|
||||
expect(() => updateHostLibraryCaches(hostId, [{ libraryVersionId: version.id, status: 'downloading', bytesDownloaded: size + 1 }]))
|
||||
.toThrow('Invalid library download progress');
|
||||
} finally { parsing.mockRestore(); preparing.mockRestore(); }
|
||||
});
|
||||
|
||||
it('does not cache missing or unpublished versions', () => {
|
||||
updateHostLibraryCaches(hostId, [{ libraryVersionId: 'future', status: 'ready', manifestSha256: 'checksum' }]);
|
||||
const version = begin();
|
||||
state.handle!.prepare("UPDATE game_file_versions SET id='future' WHERE id=?").run(version.id);
|
||||
updateHostLibraryCaches(hostId, [{ libraryVersionId: 'future', status: 'ready', manifestSha256: 'checksum' }]);
|
||||
expect(getHostLibraryCache(hostId, 'future')).toBeNull();
|
||||
state.handle!.prepare("UPDATE game_file_versions SET state='ready',size_bytes=123,manifest_sha256='checksum' WHERE id='future'").run();
|
||||
updateHostLibraryCaches(hostId, [{ libraryVersionId: 'future', status: 'ready', manifestSha256: 'checksum' }]);
|
||||
expect(getHostLibraryCache(hostId, 'future')).toMatchObject({ status: 'ready', bytesTotal: 123 });
|
||||
});
|
||||
|
||||
it('does not retain metadata from a rolled-back publication', () => {
|
||||
const version = begin();
|
||||
expect(() => state.handle!.transaction(() => {
|
||||
state.handle!.prepare("UPDATE game_file_versions SET state='ready',size_bytes=123,manifest_sha256='temporary' WHERE id=?").run(version.id);
|
||||
updateHostLibraryCaches(hostId, [{ libraryVersionId: version.id, status: 'ready', manifestSha256: 'temporary' }]);
|
||||
throw new Error('rollback');
|
||||
})()).toThrow('rollback');
|
||||
updateHostLibraryCaches(hostId, [{ libraryVersionId: version.id, status: 'ready', manifestSha256: 'temporary' }]);
|
||||
expect(getHostLibraryCache(hostId, version.id)).toBeNull();
|
||||
state.handle!.prepare("UPDATE game_file_versions SET state='ready',size_bytes=456,manifest_sha256='published' WHERE id=?").run(version.id);
|
||||
updateHostLibraryCaches(hostId, [{ libraryVersionId: version.id, status: 'ready', manifestSha256: 'published' }]);
|
||||
expect(getHostLibraryCache(hostId, version.id)).toMatchObject({ status: 'ready', bytesTotal: 456 });
|
||||
});
|
||||
|
||||
it('keeps metadata caches separate for each database connection', () => {
|
||||
const seed = () => {
|
||||
listLibraries();
|
||||
legacyLibrary('cache-library', 't6');
|
||||
legacyVersion('cache-version', 'cache-library', '2025-01-01', '2025-01-01');
|
||||
};
|
||||
const report = () => updateHostLibraryCaches(hostId, [{ libraryVersionId: 'cache-version', status: 'ready', manifestSha256: 'checksum' }]);
|
||||
seed(); report();
|
||||
const original = state.handle!;
|
||||
state.handle = new Database(':memory:');
|
||||
try {
|
||||
seed();
|
||||
const preparing = vi.spyOn(state.handle, 'prepare');
|
||||
try {
|
||||
report();
|
||||
expect(preparing.mock.calls.filter(([sql]) => sql.includes('FROM game_file_versions'))).toHaveLength(1);
|
||||
} finally { preparing.mockRestore(); }
|
||||
} finally { state.handle.close(); state.handle = original; }
|
||||
});
|
||||
|
||||
it('deduplicates active downloads and blocks readiness without exact manifest verification', async () => {
|
||||
const version = await published();
|
||||
const operation = queueLibraryDownload(hostId, version.id);
|
||||
|
||||
@@ -68,6 +68,23 @@ function verified(id: string, version: GameLibraryVersion) {
|
||||
}
|
||||
|
||||
describe('automatic host preparation', () => {
|
||||
it('accepts heartbeats without running controller library cleanup', async () => {
|
||||
const host = hosts.createHost({ name: 'Heartbeat host' });
|
||||
const { credential } = hosts.enrollHost(hosts.createEnrollment(host.id).token, inventory);
|
||||
const { POST } = await import('../../src/routes/api/agent/heartbeat/+server.js');
|
||||
const cleanup = vi.spyOn(libraries, 'cleanupControllerLibraries').mockRejectedValue(new Error('Cleanup is independent of heartbeats'));
|
||||
try {
|
||||
for (let i = 0; i < 3; i++) {
|
||||
const response = await POST({ request: new Request('https://controller.test/api/agent/heartbeat', {
|
||||
method: 'POST', headers: { Authorization: `Bearer ${credential}`, 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ inventory: { ...inventory, libraryCaches: [] } })
|
||||
}) });
|
||||
expect(response.status).toBe(200);
|
||||
}
|
||||
expect(cleanup).not.toHaveBeenCalled();
|
||||
} finally { cleanup.mockRestore(); }
|
||||
});
|
||||
|
||||
it('keeps re-enabled hosts disabled until a fresh runtime pull and both game libraries are verified', async () => {
|
||||
const t5 = await publish('t5');
|
||||
const t6 = await publish('t6');
|
||||
|
||||
@@ -0,0 +1,146 @@
|
||||
import { setImmediate } from 'node:timers/promises';
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import type { HostInventory } from '../../src/lib/hosts.js';
|
||||
import { newFixture, type Fixture } from '../helpers/fixture.js';
|
||||
|
||||
let fixture: Fixture;
|
||||
const epoch = Date.parse('2026-10-05T12:00:00Z');
|
||||
const inventory = { hostname: 'request-cache-host', protocolVersion: 1, dockerRunning: true, directoriesWritable: true,
|
||||
storageAvailable: 100 * 1024 ** 3, runtimeImages: ['plutainer:test'], libraries: [{ id: 'test-library', state: 'ready' }] };
|
||||
beforeEach(async () => {
|
||||
vi.resetModules(); vi.useFakeTimers({ toFake: ['Date'] }); vi.setSystemTime(epoch);
|
||||
fixture = await newFixture([]);
|
||||
});
|
||||
afterEach(async () => { vi.restoreAllMocks(); vi.useRealTimers(); await fixture.cleanup(); });
|
||||
|
||||
async function enrolled() {
|
||||
const hosts = await import('../../src/lib/server/hosts/index.js');
|
||||
const context = await import('../../src/lib/server/hosts/request-context.js');
|
||||
const host = hosts.createHost({ name: 'Request cache host' });
|
||||
const { credential } = hosts.enrollHost(hosts.createEnrollment(host.id).token, inventory);
|
||||
// Finish enrollment's independent availability refresh before counting parses.
|
||||
await setImmediate();
|
||||
return { ...hosts, ...context, hostId: host.id, credential };
|
||||
}
|
||||
|
||||
describe('agent credential usage writes', () => {
|
||||
it('writes on first use and at most once per minute, including after reopening the database', async () => {
|
||||
const api = await enrolled();
|
||||
const { db, closeDb } = await import('../../src/lib/server/db/index.js');
|
||||
db().exec(`CREATE TABLE credential_usage_writes (credential_id TEXT, at TEXT);
|
||||
CREATE TRIGGER count_credential_usage AFTER UPDATE OF last_used_at ON ec_host_credentials
|
||||
BEGIN INSERT INTO credential_usage_writes VALUES (NEW.id, NEW.last_used_at); END;`);
|
||||
const writes = () => db().prepare('SELECT at FROM credential_usage_writes').all();
|
||||
const authenticate = () => expect(api.authenticateCredential(api.credential)?.id).toBe(api.hostId);
|
||||
authenticate(); authenticate(); authenticate();
|
||||
expect(writes()).toEqual([{ at: new Date(epoch).toISOString() }]);
|
||||
vi.setSystemTime(epoch + 59_999);
|
||||
authenticate();
|
||||
closeDb(); authenticate();
|
||||
expect(writes()).toHaveLength(1);
|
||||
vi.setSystemTime(epoch + 60_000);
|
||||
authenticate(); authenticate();
|
||||
expect(writes()).toEqual([{ at: new Date(epoch).toISOString() }, { at: new Date(epoch + 60_000).toISOString() }]);
|
||||
expect(api.authenticateCredential('invalid')).toBeNull();
|
||||
expect(writes()).toHaveLength(2);
|
||||
});
|
||||
|
||||
it('checks revocation immediately and records first use of a replacement credential', async () => {
|
||||
const api = await enrolled();
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
await api.withHostRequestContext(() => {
|
||||
expect(api.authenticateCredential(api.credential)?.id).toBe(api.hostId);
|
||||
vi.setSystemTime(epoch + 1000);
|
||||
const replacement = api.rotateCredential(api.hostId);
|
||||
expect(api.authenticateCredential(api.credential)).toBeNull();
|
||||
expect(api.authenticateCredential(replacement)?.id).toBe(api.hostId);
|
||||
expect(db().prepare('SELECT last_used_at FROM ec_host_credentials WHERE revoked_at IS NULL').get())
|
||||
.toEqual({ last_used_at: new Date(epoch + 1000).toISOString() });
|
||||
api.revokeCredentials(api.hostId);
|
||||
expect(api.authenticateCredential(replacement)).toBeNull();
|
||||
expect(api.getHost(api.hostId)?.credentialActive).toBe(false);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('request-scoped host inventory parsing', () => {
|
||||
it('parses once across authentication, host reads, and operation handling within a request', async () => {
|
||||
const api = await enrolled();
|
||||
const serialized = JSON.stringify(inventory);
|
||||
const parse = vi.spyOn(JSON, 'parse');
|
||||
await api.withHostRequestContext(async () => {
|
||||
const authenticated = api.authenticateCredential(api.credential)!;
|
||||
const operation = api.queueHostOperation(api.hostId, 'instance.inspect', { name: 'server' });
|
||||
api.pendingHostOperationRows(api.hostId, 'read', false);
|
||||
api.reportHostOperationStatus(api.hostId, operation.id, { state: 'accepted' });
|
||||
await Promise.resolve();
|
||||
for (let i = 0; i < 10; i++) expect(api.getHost(api.hostId)?.inventory).toBe(authenticated.inventory);
|
||||
expect(api.requireReadyHost(api.hostId).inventory).toBe(authenticated.inventory);
|
||||
expect(api.listHosts()[0].inventory).toBe(authenticated.inventory);
|
||||
});
|
||||
expect(parse.mock.calls.filter(([value]) => value === serialized)).toHaveLength(1);
|
||||
await api.withHostRequestContext(() => api.getHost(api.hostId));
|
||||
expect(parse.mock.calls.filter(([value]) => value === serialized)).toHaveLength(2);
|
||||
});
|
||||
|
||||
it('keeps host metadata fresh and reparses changed inventory, including after rollback', async () => {
|
||||
const api = await enrolled();
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
await api.withHostRequestContext(() => {
|
||||
const original = api.getHost(api.hostId)!;
|
||||
const updated = api.updateHost(api.hostId, { name: 'Renamed host', mode: 'draining' });
|
||||
expect(updated).toMatchObject({ name: 'Renamed host', mode: 'draining', readiness: 'draining' });
|
||||
expect(updated.inventory).toBe(original.inventory);
|
||||
const heartbeat = api.heartbeat(api.hostId, { ...inventory, storageAvailable: 123 });
|
||||
expect(heartbeat.inventory).not.toBe(original.inventory);
|
||||
expect(api.getHost(api.hostId)?.inventory).toBe(heartbeat.inventory);
|
||||
expect(api.getHost(api.hostId)?.inventory.storageAvailable).toBe(123);
|
||||
expect(() => db().transaction(() => {
|
||||
db().prepare('UPDATE ec_hosts SET inventory=? WHERE id=?').run(JSON.stringify({ ...inventory, storageAvailable: 456 }), api.hostId);
|
||||
expect(api.getHost(api.hostId)?.inventory.storageAvailable).toBe(456);
|
||||
throw new Error('rollback');
|
||||
})()).toThrow('rollback');
|
||||
expect(api.getHost(api.hostId)?.inventory.storageAvailable).toBe(123);
|
||||
});
|
||||
});
|
||||
|
||||
it('isolates overlapping requests and does not cache reads outside a request', async () => {
|
||||
const api = await enrolled();
|
||||
let release!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
let first!: HostInventory;
|
||||
const request = api.withHostRequestContext(async () => {
|
||||
first = api.getHost(api.hostId)!.inventory;
|
||||
await gate;
|
||||
expect(api.getHost(api.hostId)!.inventory).toBe(first);
|
||||
});
|
||||
try {
|
||||
await api.withHostRequestContext(async () => {
|
||||
const second = api.getHost(api.hostId)!.inventory;
|
||||
expect(second).not.toBe(first);
|
||||
await Promise.resolve();
|
||||
expect(api.getHost(api.hostId)!.inventory).toBe(second);
|
||||
});
|
||||
expect(api.getHost(api.hostId)!.inventory).not.toBe(api.getHost(api.hostId)!.inventory);
|
||||
} finally { release(); await request; }
|
||||
});
|
||||
|
||||
it.each([false, true])('releases cached inventory from inherited work after request completion (failed: %s)', async failed => {
|
||||
const api = await enrolled();
|
||||
let release!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
let detached!: Promise<void>;
|
||||
let original!: HostInventory;
|
||||
const request = api.withHostRequestContext(() => {
|
||||
original = api.getHost(api.hostId)!.inventory;
|
||||
detached = gate.then(() => {
|
||||
const later = api.getHost(api.hostId)!.inventory;
|
||||
expect(later).not.toBe(original);
|
||||
expect(api.getHost(api.hostId)!.inventory).not.toBe(later);
|
||||
});
|
||||
if (failed) throw new Error('request failed');
|
||||
});
|
||||
if (failed) await expect(request).rejects.toThrow('request failed'); else await request;
|
||||
release(); await detached;
|
||||
});
|
||||
});
|
||||
@@ -70,10 +70,10 @@ describe('provisioning reads with retained lifecycle history', () => {
|
||||
expect(queries).toHaveLength(1);
|
||||
const statement = handle.prepare(queries[0]);
|
||||
expect(statement.columns().map(column => column.name)).toEqual(['id', 'server', 'action', 'state', 'stage', 'error', 'started_at']);
|
||||
const plan = handle.prepare(`EXPLAIN QUERY PLAN ${queries[0]}`).all(iso(-86_400_000)) as { detail: string }[];
|
||||
const plan = handle.prepare(`EXPLAIN QUERY PLAN ${queries[0]}`).all() as { detail: string }[];
|
||||
const details = plan.map(row => row.detail).join('\n');
|
||||
expect(details).toContain('SEARCH server_deployments USING INDEX idx_server_deployments_state');
|
||||
expect(details).toContain('SEARCH server_deployments USING INDEX idx_server_deployments_finished_at');
|
||||
expect(queries[0]).not.toContain('finished_at >=');
|
||||
expect(details).toContain('SEARCH server_deployments USING INTEGER PRIMARY KEY (rowid=?)');
|
||||
expect(details).toContain('SCAN server_deployments USING COVERING INDEX idx_server_deployments_server');
|
||||
expect(details).not.toMatch(/SCAN server_deployments(?:\n|$)/);
|
||||
|
||||
@@ -16,6 +16,53 @@ afterEach(async () => {
|
||||
});
|
||||
|
||||
describe('sweep', () => {
|
||||
it('emits server state only when state, players, or error changes', async () => {
|
||||
const { sweep } = await import('../../src/lib/server/pipeline/sweep.js');
|
||||
const { seedServerRegistry } = await import('../../src/lib/server/fleet/servers.js');
|
||||
const { db } = await import('../../src/lib/server/db/index.js');
|
||||
const { subscribe } = await import('../../src/lib/server/pipeline/bus.js');
|
||||
await seedServerRegistry();
|
||||
db().prepare("INSERT INTO deployments (id, action, started_at, triggered_by) VALUES ('changes', 'test', ?, 'ui')")
|
||||
.run(new Date().toISOString());
|
||||
db().prepare(`INSERT INTO deployment_servers (deployment_id, server, files_changed, restart_state)
|
||||
VALUES ('changes', 't5_zombie_theater_1', 1, 'pending')`).run();
|
||||
const events = vi.fn();
|
||||
const unsubscribe = subscribe(event => { if (event.type === 'server-state') events(event); });
|
||||
try {
|
||||
await sweep();
|
||||
expect(events).toHaveBeenCalledTimes(1);
|
||||
await sweep();
|
||||
expect(events).toHaveBeenCalledTimes(1);
|
||||
|
||||
// Fresh check timestamps still update the campaign, without an SSE event.
|
||||
const checkedAt = new Date().toISOString();
|
||||
db().prepare('UPDATE server_deployments SET checked_at = ?').run(checkedAt);
|
||||
await sweep();
|
||||
expect(events).toHaveBeenCalledTimes(1);
|
||||
expect(db().prepare("SELECT checked_at, attempts FROM deployment_servers WHERE deployment_id = 'changes'").get())
|
||||
.toEqual({ checked_at: checkedAt, attempts: 1 });
|
||||
|
||||
db().prepare('UPDATE server_deployments SET players = 2').run();
|
||||
await sweep();
|
||||
expect(events).toHaveBeenCalledTimes(2);
|
||||
expect(events).toHaveBeenLastCalledWith(expect.objectContaining({ state: 'pending', players: 2 }));
|
||||
db().prepare("UPDATE server_deployments SET error = 'Waiting for players'").run();
|
||||
await sweep();
|
||||
expect(events).toHaveBeenCalledTimes(3);
|
||||
expect(events).toHaveBeenLastCalledWith(expect.objectContaining({ error: 'Waiting for players' }));
|
||||
db().prepare("UPDATE server_deployments SET state = 'failed'").run();
|
||||
await sweep();
|
||||
expect(events).toHaveBeenCalledTimes(4);
|
||||
expect(events).toHaveBeenLastCalledWith(expect.objectContaining({ state: 'failed' }));
|
||||
await sweep();
|
||||
expect(events).toHaveBeenCalledTimes(4);
|
||||
db().prepare("UPDATE server_deployments SET state = 'done'").run();
|
||||
await sweep();
|
||||
expect(events).toHaveBeenCalledTimes(5);
|
||||
expect(events).toHaveBeenLastCalledWith(expect.objectContaining({ state: 'done', error: null }));
|
||||
} finally { unsubscribe(); }
|
||||
});
|
||||
|
||||
it('queues pending rows in Provisioning even when integrations are unset', async () => {
|
||||
const { sweep } = await import('../../src/lib/server/pipeline/sweep.js');
|
||||
const { seedServerRegistry } = await import('../../src/lib/server/fleet/servers.js');
|
||||
|
||||
@@ -90,6 +90,45 @@ it('timestamps initial telemetry and subsequent status and telemetry events with
|
||||
}
|
||||
});
|
||||
|
||||
it('serializes each live server event once across connections and releases cancelled subscribers', async () => {
|
||||
const { GET } = await import('../../src/routes/api/events/servers/+server.js');
|
||||
const { emit } = await import('../../src/lib/server/pipeline/bus.js');
|
||||
const readers = await Promise.all([GET(), GET()].map(async response => (await response).body!.getReader()));
|
||||
const stringify = vi.spyOn(JSON, 'stringify');
|
||||
try {
|
||||
for (const reader of readers) {
|
||||
await reader.read(); // Initial comment and per-connection snapshots.
|
||||
await reader.read();
|
||||
await reader.read();
|
||||
}
|
||||
for (const event of [
|
||||
{ type: 'server-status' as const, delta: true, entries: [] },
|
||||
{ type: 'server-telemetry' as const, entries: [{ ...payload, receivedAt: new Date().toISOString() }] }
|
||||
]) {
|
||||
stringify.mockClear();
|
||||
emit(event);
|
||||
expect(stringify).toHaveBeenCalledTimes(1);
|
||||
const [first, second] = await Promise.all(readers.map(reader => reader.read()));
|
||||
expect(first.value).toBe(second.value);
|
||||
const frame = new TextDecoder().decode(first.value);
|
||||
expect(frame.startsWith(`event: ${event.type}\ndata: `)).toBe(true);
|
||||
expect(JSON.parse(frame.split('\n')[1].slice(6))).toEqual({ ...event, serverNow: expect.any(Number) });
|
||||
}
|
||||
await readers[0].cancel();
|
||||
stringify.mockClear();
|
||||
emit({ type: 'server-status', entries: [] });
|
||||
expect(stringify).toHaveBeenCalledTimes(1);
|
||||
expect((await readers[1].read()).done).toBe(false);
|
||||
await readers[1].cancel();
|
||||
stringify.mockClear();
|
||||
emit({ type: 'server-telemetry', entries: [] });
|
||||
expect(stringify).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
stringify.mockRestore();
|
||||
await Promise.all(readers.map(reader => reader.cancel()));
|
||||
}
|
||||
});
|
||||
|
||||
it.each(['round_change', 'heartbeat'] as const)('persists %s stats, broadcasts updates, and leaves RCON safety data untouched', async (event) => {
|
||||
const payload = gameEvent(event, 17);
|
||||
const { subscribe } = await import('../../src/lib/server/pipeline/bus.js');
|
||||
|
||||
@@ -8,9 +8,16 @@ import { gameEvent } from '../helpers/telemetry.js';
|
||||
vi.mock('$app/environment', () => ({ browser: true }));
|
||||
|
||||
const epoch = Date.parse('2026-09-19T12:00:00Z');
|
||||
let source: EventTarget;
|
||||
let source: EventTarget & { close: () => void };
|
||||
let stop: (() => void) | undefined;
|
||||
let snapshot: { serverNow: number; servers: FleetServer[]; integrations: object };
|
||||
let frames: Map<number, FrameRequestCallback>;
|
||||
|
||||
function flushFrame() {
|
||||
const callbacks = [...frames.values()];
|
||||
frames.clear();
|
||||
for (const callback of callbacks) callback(performance.now());
|
||||
}
|
||||
|
||||
function server(timestamp = epoch, changes: Partial<FleetServer> = {}): FleetServer {
|
||||
return {
|
||||
@@ -35,12 +42,20 @@ function status(timestamp: number, changes: Partial<ServerStatus> = {}): ServerS
|
||||
|
||||
function emit(serverNow: number, entries: ServerStatus[]) {
|
||||
source.dispatchEvent(new MessageEvent('server-status', { data: JSON.stringify({ serverNow, entries }) }));
|
||||
flushFrame();
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.resetModules();
|
||||
vi.useFakeTimers({ toFake: ['Date', 'performance', 'setTimeout', 'clearTimeout', 'setInterval', 'clearInterval'] });
|
||||
vi.setSystemTime(epoch);
|
||||
frames = new Map();
|
||||
let frameId = 0;
|
||||
vi.stubGlobal('requestAnimationFrame', vi.fn((callback: FrameRequestCallback) => {
|
||||
frames.set(++frameId, callback);
|
||||
return frameId;
|
||||
}));
|
||||
vi.stubGlobal('cancelAnimationFrame', vi.fn((id: number) => { frames.delete(id); }));
|
||||
snapshot = { serverNow: epoch, servers: [server()], integrations: {} };
|
||||
vi.stubGlobal('fetch', vi.fn(async (url: string) => new Response(JSON.stringify(
|
||||
url === '/api/servers' ? snapshot : { operations: [] }
|
||||
@@ -66,6 +81,130 @@ async function setup() {
|
||||
return { ...clock, ...fleet };
|
||||
}
|
||||
|
||||
function queueEvent(type: string, data: object) {
|
||||
source.dispatchEvent(new MessageEvent(type, { data: JSON.stringify(data) }));
|
||||
}
|
||||
|
||||
describe('animation-frame stream batching', () => {
|
||||
it('publishes each changed store once for a burst and preserves lifecycle evidence through heartbeats', async () => {
|
||||
const stores = await setup();
|
||||
emit(epoch, [status(epoch), status(epoch, { server: 'two' })]);
|
||||
const before = get(stores.statusByServer);
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
const watched = [stores.statusByServer, stores.statusReceivedAt, stores.telemetryByServer, stores.matchEventsByServer, stores.uiNow];
|
||||
const changes = watched.map(() => vi.fn());
|
||||
const unsubscribe = watched.map((store, index) => store.subscribe(changes[index]));
|
||||
changes.forEach(change => change.mockClear());
|
||||
vi.mocked(requestAnimationFrame).mockClear();
|
||||
try {
|
||||
for (let i = 0; i < 100; i++) {
|
||||
queueEvent('server-status', { delta: true, serverNow: epoch + 1000 + i,
|
||||
entries: [status(epoch + 1000, { players: i })] });
|
||||
queueEvent('server-telemetry', { serverNow: epoch + 1000 + i,
|
||||
entries: [{ ...gameEvent(i === 0 ? 'game_start' : 'heartbeat', i, 1), receivedAt: new Date(epoch + 1000 + i).toISOString() }] });
|
||||
}
|
||||
expect(requestAnimationFrame).toHaveBeenCalledTimes(1);
|
||||
changes.forEach(change => expect(change).not.toHaveBeenCalled());
|
||||
expect(get(stores.statusByServer)).toBe(before);
|
||||
expect(before.get('t5_zombie_theater_1')?.players).toBe(0);
|
||||
flushFrame();
|
||||
changes.forEach(change => expect(change).toHaveBeenCalledTimes(1));
|
||||
expect(get(stores.statusByServer).size).toBe(2);
|
||||
expect(get(stores.statusByServer).get('t5_zombie_theater_1')?.players).toBe(99);
|
||||
expect(get(stores.telemetryByServer).get('t5_zombie_theater_1')).toMatchObject({ event: 'heartbeat', round: 99 });
|
||||
expect(get(stores.matchEventsByServer).get('t5_zombie_theater_1')?.event).toBe('game_start');
|
||||
|
||||
queueEvent('server-telemetry', { entries: [{ ...gameEvent('heartbeat', 100, 1), receivedAt: new Date(epoch + 1100).toISOString() }] });
|
||||
flushFrame();
|
||||
expect(requestAnimationFrame).toHaveBeenCalledTimes(2);
|
||||
expect(changes[2]).toHaveBeenCalledTimes(2);
|
||||
expect(changes[3]).toHaveBeenCalledTimes(1);
|
||||
} finally { unsubscribe.forEach(unsubscribe => unsubscribe()); }
|
||||
});
|
||||
|
||||
it('applies snapshots and subsequent deltas in order, including empty snapshots', async () => {
|
||||
const { statusByServer, telemetryByServer, matchEventsByServer } = await setup();
|
||||
const telemetry = (event: GameEvent['event'], name: string) => ({ ...gameEvent(event, 1, 1, name), receivedAt: new Date(epoch).toISOString() });
|
||||
emit(epoch, [status(epoch, { server: 'old' })]);
|
||||
queueEvent('server-telemetry', { entries: [telemetry('game_start', 'old')] });
|
||||
flushFrame();
|
||||
|
||||
queueEvent('server-status', { delta: true, entries: [status(epoch, { server: 'discarded' })] });
|
||||
queueEvent('server-status', { entries: [status(epoch, { server: 'two' })] });
|
||||
queueEvent('server-status', { delta: true, entries: [status(epoch, { server: 'three' })] });
|
||||
queueEvent('server-telemetry', { entries: [telemetry('game_end', 'discarded')] });
|
||||
queueEvent('server-telemetry', { snapshot: true, entries: [telemetry('heartbeat', 'two')] });
|
||||
queueEvent('server-telemetry', { entries: [telemetry('game_end', 'three')] });
|
||||
queueEvent('server-telemetry', { entries: [telemetry('heartbeat', 'three')] });
|
||||
flushFrame();
|
||||
expect([...get(statusByServer).keys()]).toEqual(['two', 'three']);
|
||||
expect([...get(telemetryByServer).keys()]).toEqual(['two', 'three']);
|
||||
expect([...get(matchEventsByServer).keys()]).toEqual(['three']);
|
||||
expect(get(matchEventsByServer).get('three')?.event).toBe('game_end');
|
||||
expect(get(telemetryByServer).get('three')?.event).toBe('heartbeat');
|
||||
|
||||
queueEvent('server-status', { entries: [] });
|
||||
queueEvent('server-telemetry', { snapshot: true, entries: [] });
|
||||
flushFrame();
|
||||
expect(get(statusByServer).size).toBe(0);
|
||||
expect(get(telemetryByServer).size).toBe(0);
|
||||
expect(get(matchEventsByServer).size).toBe(0);
|
||||
});
|
||||
|
||||
it('retains event arrival times so buffered status cannot overwrite a newer HTTP snapshot', async () => {
|
||||
const { fleetServers, refreshFleet, statusByServer, statusReceivedAt } = await setup();
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
queueEvent('server-status', { delta: true, serverNow: epoch + 1000, entries: [status(epoch + 1000, { players: 1 })] });
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
snapshot = { ...snapshot, serverNow: epoch + 2000, servers: [server(epoch + 2000, { last_players: 4 })] };
|
||||
await refreshFleet();
|
||||
queueEvent('server-status', { delta: true, serverNow: epoch + 2000, entries: [status(epoch + 2000, { server: 'two' })] });
|
||||
flushFrame();
|
||||
expect(get(statusByServer).get('t5_zombie_theater_1')?.receivedAt).toBe(epoch + 1000);
|
||||
expect(get(statusReceivedAt)).toBe(epoch + 2000);
|
||||
expect(get(fleetServers)[0].last_players).toBe(4);
|
||||
});
|
||||
|
||||
it('accounts for frame delay without rewinding the server clock', async () => {
|
||||
const { uiNow, statusByServer } = await setup();
|
||||
vi.stubGlobal('document', { visibilityState: 'hidden' });
|
||||
queueEvent('server-status', { serverNow: epoch + 5000, entries: [status(epoch + 5000)] });
|
||||
await vi.advanceTimersByTimeAsync(2000);
|
||||
queueEvent('server-telemetry', { serverNow: epoch + 4000, entries: [] });
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(get(uiNow)).toBe(epoch);
|
||||
expect(get(statusByServer).size).toBe(0);
|
||||
expect(frames.size).toBe(1);
|
||||
flushFrame();
|
||||
expect(get(uiNow)).toBe(epoch + 8000);
|
||||
expect(get(statusByServer).size).toBe(1);
|
||||
});
|
||||
|
||||
it('cancels queued frames and discards buffered events when stopping and restarting', async () => {
|
||||
const { statusByServer, telemetryByServer, startServerStream } = await setup();
|
||||
const oldSource = source;
|
||||
const close = vi.spyOn(oldSource, 'close');
|
||||
queueEvent('server-status', { entries: [status(epoch)] });
|
||||
queueEvent('server-telemetry', { entries: [{ ...gameEvent('game_start'), receivedAt: new Date(epoch).toISOString() }] });
|
||||
const callback = [...frames.values()][0];
|
||||
const oldStop = stop!;
|
||||
oldStop();
|
||||
expect(close).toHaveBeenCalledTimes(1);
|
||||
expect(cancelAnimationFrame).toHaveBeenCalledTimes(1);
|
||||
expect(frames.size).toBe(0);
|
||||
callback(0);
|
||||
expect(get(statusByServer).size).toBe(0);
|
||||
expect(get(telemetryByServer).size).toBe(0);
|
||||
stop = startServerStream();
|
||||
oldStop();
|
||||
oldSource.dispatchEvent(new MessageEvent('server-status', { data: JSON.stringify({ entries: [status(epoch)] }) }));
|
||||
queueEvent('server-status', { delta: true, entries: [status(epoch, { server: 'new' })] });
|
||||
flushFrame();
|
||||
expect([...get(statusByServer).keys()]).toEqual(['new']);
|
||||
expect(get(telemetryByServer).size).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe('Fleet report clock', () => {
|
||||
it('accepts newer navigation snapshots while keeping stream evidence newer than the snapshot', async () => {
|
||||
const { fleetServers, hydrateFleet } = await setup();
|
||||
@@ -86,6 +225,7 @@ describe('Fleet report clock', () => {
|
||||
source.dispatchEvent(new MessageEvent('server-status', { data: JSON.stringify({
|
||||
delta: true, serverNow: epoch + 1000, entries: [status(epoch + 1000, { players: 2 })]
|
||||
}) }));
|
||||
flushFrame();
|
||||
expect(get(statusByServer).size).toBe(2);
|
||||
expect(get(fleetServers).find(server => server.name === 't5_zombie_theater_1')?.last_players).toBe(2);
|
||||
const changes = vi.fn();
|
||||
@@ -103,6 +243,7 @@ describe('Fleet report clock', () => {
|
||||
source.dispatchEvent(new MessageEvent('server-telemetry', { data: JSON.stringify({ serverNow: epoch + seconds * 1000,
|
||||
entries: [{ ...gameEvent(event, 2, players), receivedAt: at(seconds),
|
||||
timing: { matchId: 1, players: {}, server: timing } }] }) }));
|
||||
flushFrame();
|
||||
return get(fleetServers)[0];
|
||||
};
|
||||
const active = { activeSince: at(1), inactiveSince: null, gameStartedAt: at(1), gameEndedAt: null };
|
||||
@@ -132,15 +273,17 @@ describe('Fleet report clock', () => {
|
||||
entries: [{ ...gameEvent('heartbeat', 2, 1), receivedAt: new Date(epoch).toISOString(),
|
||||
timing: { matchId: 1, players: {}, server: { activeSince: startedAt, inactiveSince: null,
|
||||
gameStartedAt: startedAt, gameEndedAt: null } } }] }) }));
|
||||
flushFrame();
|
||||
expect(get(fleetServers)[0]).toMatchObject({ active_since: startedAt, game_started_at: startedAt });
|
||||
});
|
||||
|
||||
it('updates count and roster immediately from telemetry and expires them together', async () => {
|
||||
it('updates count and roster on the next frame and expires them together', async () => {
|
||||
snapshot.servers = [server(epoch, { container_state: 'running', container_checked_at: new Date(epoch).toISOString(), last_players: 4 })];
|
||||
const { fleetServers, uiNow } = await setup();
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
source.dispatchEvent(new MessageEvent('server-telemetry', { data: JSON.stringify({ serverNow: epoch + 1000,
|
||||
entries: [{ ...gameEvent('heartbeat', 15, 3), receivedAt: new Date(epoch + 1000).toISOString() }] }) }));
|
||||
flushFrame();
|
||||
expect(get(fleetServers)[0]).toMatchObject({ health: 'online', last_players: 3, status_source: 'telemetry' });
|
||||
expect(get(fleetServers)[0].roster).toHaveLength(3);
|
||||
expect(matchesFleet(get(fleetServers)[0], 'Player 3', 'occupied', '', [], [], get(uiNow))).toBe(true);
|
||||
@@ -149,12 +292,15 @@ describe('Fleet report clock', () => {
|
||||
expect(matchesFleet(get(fleetServers)[0], '', 'occupied', '', [], [], get(uiNow))).toBe(false);
|
||||
});
|
||||
|
||||
it('applies game lifecycle events immediately and clears them after an empty report', async () => {
|
||||
it('applies game lifecycle events on the next frame and clears them after an empty report', async () => {
|
||||
const { fleetServers } = await setup();
|
||||
const report = (event: string, time: number) => source.dispatchEvent(new MessageEvent('server-telemetry', {
|
||||
const report = (event: string, time: number) => {
|
||||
source.dispatchEvent(new MessageEvent('server-telemetry', {
|
||||
data: JSON.stringify({ serverNow: time, entries: [{ server: 't5_zombie_theater_1', event,
|
||||
receivedAt: new Date(time).toISOString(), players: [] }] })
|
||||
}));
|
||||
flushFrame();
|
||||
};
|
||||
emit(epoch, [status(epoch, { players: 1 })]);
|
||||
report('game_end', epoch + 1000);
|
||||
expect(get(fleetServers)[0].game_ended_at).toBe(new Date(epoch + 1000).toISOString());
|
||||
@@ -249,6 +395,7 @@ describe('Fleet report clock', () => {
|
||||
source.dispatchEvent(new MessageEvent('server-telemetry', {
|
||||
data: JSON.stringify({ serverNow: epoch, entries: [], snapshot: true })
|
||||
}));
|
||||
flushFrame();
|
||||
expect(get(uiNow)).toBe(epoch);
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(get(uiNow)).toBe(epoch + 1000);
|
||||
|
||||
@@ -31,6 +31,30 @@ async function setup() {
|
||||
}
|
||||
|
||||
describe('independent provisioning scheduling', () => {
|
||||
it('runs controller library cleanup on worker ticks without overlapping a slow cleanup', async () => {
|
||||
const s = await setup();
|
||||
const { cleanupControllerLibraries } = await import('../../src/lib/server/game-files/index.js');
|
||||
const cleanup = deferred();
|
||||
vi.mocked(cleanupControllerLibraries).mockImplementationOnce(() => cleanup.promise);
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
s.startWorker();
|
||||
expect(cleanupControllerLibraries).toHaveBeenCalledTimes(1);
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(cleanupControllerLibraries).toHaveBeenCalledTimes(1);
|
||||
expect(s.process.mock.calls.length).toBeGreaterThan(1);
|
||||
cleanup.resolve();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
await vi.advanceTimersByTimeAsync(15_000);
|
||||
expect(cleanupControllerLibraries).toHaveBeenCalledTimes(2);
|
||||
} finally {
|
||||
s.stopWorker(); cleanup.resolve(); s.background.resolve();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
vi.useRealTimers();
|
||||
await vi.waitFor(() => expect(s.workerActivity()).toEqual([]));
|
||||
}
|
||||
});
|
||||
|
||||
it('admits and checks new work while campaign, beta and maintenance observers are still waiting', async () => {
|
||||
const s = await setup();
|
||||
try {
|
||||
|
||||
Reference in New Issue
Block a user