.changeset/restart-only-collection.md

@@ -0,0 +1,7 @@
+---
+'@goto/web': minor
+'@goto/worker': minor
+'@goto/cli': minor
+---
+
+Make analytics collection on/off startup-only and remove the collection-control table and insertion fence. Stop and drain both services before changing between zero and positive retention, then restart; changes between positive retention periods remain live. Keep whole-event admission drops with distinct bounded warnings for queue capacity and payload size.

apps/web/src/lib/server/application.test.ts

@@ -25,7 +25,6 @@ let reload: ReloadOptions;
 const stop = vi.fn();
 const stopRecorder = vi.fn();
 const recordVisit = vi.fn();
-const setEnabled = vi.fn();
 
 function prepared(
 	destination = 'https://example.org/first',
@@ -46,7 +45,7 @@ beforeEach(() => {
 	vi.clearAllMocks();
 	vi.spyOn(console, 'warn').mockImplementation(() => {});
 	vi.mocked(loadConfigFile).mockResolvedValue(prepared());
-	vi.mocked(createVisitRecorder).mockReturnValue({ recordVisit, setEnabled, stop: stopRecorder });
+	vi.mocked(createVisitRecorder).mockReturnValue({ recordVisit, stop: stopRecorder });
 	vi.mocked(startConfigReloading).mockImplementation(async (options) => {
 		reload = options as ReloadOptions;
 		return stop;
@@ -69,7 +68,7 @@ test('uses a fixed path/environment and stops monitoring with the application',
 	expect(application.currentSnapshot().configuration.auth).not.toHaveProperty('secret');
 	await application.close?.();
 	expect(stop).toHaveBeenCalledOnce();
-	expect(createVisitRecorder).toHaveBeenCalledExactlyOnceWith('/deployment/data/goto.sqlite', true);
+	expect(createVisitRecorder).toHaveBeenCalledExactlyOnceWith('/deployment/data/goto.sqlite');
 	expect(application.recordVisit).toBe(recordVisit);
 	expect(stopRecorder).toHaveBeenCalledOnce();
 });
@@ -99,7 +98,7 @@ test('publishes valid changes, ignores equivalent data, and retains last good da
 test.each([
 	{ site: { url: 'https://different-admin.example.test' } },
 	{ auth: { credentials: { username: 'changed', password: 'private-changed-password' } } },
-	{ database: { path: './data/different.sqlite' }, analytics: { retention_months: 0 } }
+	{ database: { path: './data/different.sqlite' } }
 ])('rejects the whole candidate when startup settings change: %j', async (changes) => {
 	const application = await createApplication(options);
 	const initial = application.currentSnapshot();
@@ -107,19 +106,43 @@ test.each([
 	expect(reload.onChange(prepared('https://example.org/blocked', changes), prepared())).toBe(false);
 
 	expect(application.currentSnapshot()).toBe(initial);
-	expect(setEnabled).not.toHaveBeenCalled();
 	expect(console.warn).toHaveBeenCalledExactlyOnceWith(
 		'Configuration change requires restart; keeping current configuration'
 	);
 	await application.close?.();
 });
 
-test('disables live collection without replacing the recorder', async () => {
+test('rejects live collection disable and the rest of its candidate until coordinated restart', async () => {
 	const application = await createApplication(options);
-	const next = prepared('https://example.org/first', { analytics: { retention_months: 0 } });
+	const initial = application.currentSnapshot();
+	const next = prepared('https://example.org/blocked', { analytics: { retention_months: 0 } });
+	expect(reload.onChange(next, prepared())).toBe(false);
+	expect(application.currentSnapshot()).toBe(initial);
+	expect(createVisitRecorder).toHaveBeenCalledOnce();
+	expect(console.warn).toHaveBeenCalledExactlyOnceWith(
+		'Analytics collection change requires coordinated restart; keeping current configuration'
+	);
+	await application.close?.();
+});
+
+test('zero startup creates no recorder and rejects live enabling', async () => {
+	const initial = prepared('https://example.org/first', { analytics: { retention_months: 0 } });
+	vi.mocked(loadConfigFile).mockResolvedValueOnce(initial);
+	const application = await createApplication(options);
+	const snapshot = application.currentSnapshot();
+	expect(createVisitRecorder).not.toHaveBeenCalled();
+	expect(application.recordVisit).toBeUndefined();
+	expect(reload.onChange(prepared('https://example.org/blocked'), initial)).toBe(false);
+	expect(application.currentSnapshot()).toBe(snapshot);
+	await application.close?.();
+	expect(stopRecorder).not.toHaveBeenCalled();
+});
+
+test('changes positive retention live without replacing the recorder', async () => {
+	const application = await createApplication(options);
+	const next = prepared('https://example.org/second', { analytics: { retention_months: 9 } });
 	expect(reload.onChange(next, prepared())).toBe(true);
-	expect(setEnabled).toHaveBeenCalledExactlyOnceWith(false);
-	expect(application.currentSnapshot().configuration.analytics.retention_months).toBe(0);
+	expect(application.currentSnapshot().configuration.analytics.retention_months).toBe(9);
 	expect(createVisitRecorder).toHaveBeenCalledOnce();
 	await application.close?.();
 });

apps/web/src/lib/server/application.ts

@@ -31,7 +31,8 @@ function startupSettings({ configuration }: PreparedConfiguration) {
 	return {
 		siteUrl: configuration.site.url,
 		auth: configuration.auth,
-		databasePath: configuration.database.path
+		databasePath: configuration.database.path,
+		collectionEnabled: configuration.analytics.retention_months !== 0
 	};
 }
 
@@ -47,8 +48,7 @@ export async function createApplication(options: ConfigOptions = {}): Promise<Ap
 	let snapshot = prepareSnapshot(initial);
 	const authentication = createAuthentication(snapshot.configuration);
 	const databasePath = resolve(captured.cwd, startup.databasePath);
-	const collectionEnabled = snapshot.configuration.analytics.retention_months !== 0;
-	const recorder = createVisitRecorder(databasePath, collectionEnabled);
+	const recorder = startup.collectionEnabled ? createVisitRecorder(databasePath) : undefined;
 
 	function loadConfiguration(path: string): Promise<PreparedConfiguration> {
 		return loadConfigFile(path, captured);
@@ -59,14 +59,18 @@ export async function createApplication(options: ConfigOptions = {}): Promise<Ap
 		previous: PreparedConfiguration
 	): boolean {
 		const nextSettings = startupSettings(next);
+		if (nextSettings.collectionEnabled !== startup.collectionEnabled) {
+			reportReloadIssue(
+				'Analytics collection change requires coordinated restart; keeping current configuration'
+			);
+			return false;
+		}
 		if (!isDeepStrictEqual(nextSettings, startup)) {
 			reportReloadIssue('Configuration change requires restart; keeping current configuration');
 			return false;
 		}
 		if (!isDeepStrictEqual(next, previous)) {
 			const candidate = prepareSnapshot(next);
-			const collectionEnabled = candidate.configuration.analytics.retention_months !== 0;
-			recorder.setEnabled(collectionEnabled);
 			snapshot = candidate;
 		}
 		return true;
@@ -87,7 +91,7 @@ export async function createApplication(options: ConfigOptions = {}): Promise<Ap
 	try {
 		stopReloading = await startConfigReloading(reloadOptions);
 	} catch (error) {
-		await recorder.stop();
+		await recorder?.stop();
 		throw error;
 	}
 	const application: Application = {
@@ -95,14 +99,14 @@ export async function createApplication(options: ConfigOptions = {}): Promise<Ap
 			return snapshot;
 		},
 		authentication,
-		recordVisit: recorder.recordVisit,
 		async close() {
 			try {
 				await stopReloading();
 			} finally {
-				await recorder.stop();
+				await recorder?.stop();
 			}
 		}
 	};
+	if (recorder) application.recordVisit = recorder.recordVisit;
 	return application;
 }

apps/web/src/lib/server/visits/recorder.test.ts

@@ -44,59 +44,46 @@ afterEach(() => {
 	vi.restoreAllMocks();
 });
 
-test('bounds unsettled visits and releases capacity after a failed insert', async () => {
-	const recorder = createVisitRecorder('/data/goto.sqlite', true);
+test('bounds unsettled visits and reports queue and insert failures separately', async () => {
+	const recorder = createVisitRecorder('/data/goto.sqlite');
 	recorder.recordVisit(visit);
 	expect(writes).toHaveLength(0);
 	initialized.resolve();
 	await initialized.promise;
 	for (let index = 0; index < 1000; index++) recorder.recordVisit(visit);
 	recorder.recordVisit(visit);
+	recorder.recordVisit(visit);
 	expect(writes).toHaveLength(1000);
 	writes[0]!.reject(new Error('private database detail'));
 	await Promise.resolve();
 	recorder.recordVisit(visit);
 	expect(writes).toHaveLength(1001);
-	expect(console.warn).toHaveBeenCalledExactlyOnceWith(
-		'Visit tracking unavailable; some visits may not be recorded'
-	);
+	expect(vi.mocked(console.warn).mock.calls).toEqual([
+		['Visit recording queue full (1000 outstanding); dropping analytics visit'],
+		['Visit tracking unavailable; some visits may not be recorded']
+	]);
 	await recorder.stop();
 });
 
-test('drops oversized events without changing the payload', async () => {
-	const recorder = createVisitRecorder('/data/goto.sqlite', true);
+test('drops whole oversized events and warns once without logging their payload', async () => {
+	const recorder = createVisitRecorder('/data/goto.sqlite');
 	initialized.resolve();
 	await initialized.promise;
 	const oversized = { ...visit, user_agent: 'x'.repeat(32 * 1024) };
 	recorder.recordVisit(oversized);
+	recorder.recordVisit(oversized);
 	expect(writes).toHaveLength(0);
 	expect(oversized.user_agent).toHaveLength(32 * 1024);
-	await recorder.stop();
-});
-
-test('disabled collection starts no worker and can later enable and pause admission', async () => {
-	const recorder = createVisitRecorder('/data/goto.sqlite', false);
-	recorder.recordVisit(visit);
-	expect(Piscina).not.toHaveBeenCalled();
-	recorder.setEnabled(true);
-	initialized.resolve();
-	await initialized.promise;
-	recorder.recordVisit(visit);
-	expect(writes).toHaveLength(1);
-	recorder.setEnabled(false);
-	recorder.recordVisit(visit);
-	expect(writes).toHaveLength(1);
-	recorder.setEnabled(true);
-	recorder.recordVisit(visit);
-	expect(writes).toHaveLength(2);
-	expect(Piscina).toHaveBeenCalledOnce();
+	expect(console.warn).toHaveBeenCalledExactlyOnceWith(
+		'Visit payload exceeds 32 KiB; dropping analytics visit'
+	);
 	await recorder.stop();
 });
 
 test('stops admission and waits for database and pool closure', async () => {
 	const closed = Promise.withResolvers<void>();
 	const poolClosed = Promise.withResolvers<void>();
-	const recorder = createVisitRecorder('/data/goto.sqlite', true);
+	const recorder = createVisitRecorder('/data/goto.sqlite');
 	initialized.resolve();
 	await initialized.promise;
 	pool.run.mockImplementation(() => closed.promise);
@@ -126,7 +113,7 @@ test('stops admission and waits for database and pool closure', async () => {
 });
 
 test('initialization failures disable capture without propagating private errors', async () => {
-	const recorder = createVisitRecorder('/data/goto.sqlite', true);
+	const recorder = createVisitRecorder('/data/goto.sqlite');
 	initialized.reject(new Error('private database path'));
 	await Promise.resolve();
 	recorder.recordVisit(visit);

apps/web/src/lib/server/visits/recorder.ts

@@ -3,63 +3,72 @@ import type { NewVisit } from '@goto/db';
 import { closeVisitPool, createVisitPool, MAX_OUTSTANDING } from './pool.js';
 
 const MAX_VISIT_BYTES = 32 * 1024;
+const WARNINGS = {
+	queue: 'Visit recording queue full (1000 outstanding); dropping analytics visit',
+	size: 'Visit payload exceeds 32 KiB; dropping analytics visit',
+	unavailable: 'Visit tracking unavailable; some visits may not be recorded'
+};
+
+type RecordingWarning = keyof typeof WARNINGS;
 
 export interface VisitRecorder {
 	recordVisit(visit: NewVisit): void;
-	setEnabled(enabled: boolean): void;
 	stop(): Promise<void>;
 }
 
-export function createVisitRecorder(databasePath: string, enabled: boolean): VisitRecorder {
-	return new Recorder(databasePath, enabled);
+export function createVisitRecorder(databasePath: string): VisitRecorder {
+	return new Recorder(databasePath);
 }
 
-function visitBytes(visit: NewVisit): number {
-	return [visit.domain, visit.path, visit.ip, visit.user_agent, visit.accept_language].reduce(
-		(bytes, value) => bytes + Buffer.byteLength(value ?? ''),
-		8
-	);
+function exceedsVisitSizeLimit(visit: NewVisit): boolean {
+	const textFields = [visit.domain, visit.path, visit.ip, visit.user_agent, visit.accept_language];
+	let bytes = 8;
+	for (const value of textFields) {
+		bytes += Buffer.byteLength(value ?? '');
+		if (bytes > MAX_VISIT_BYTES) return true;
+	}
+	return false;
 }
 
 class Recorder implements VisitRecorder {
-	private readonly databasePath: string;
-	private enabled: boolean;
-	private reported = false;
+	private readonly reported = new Set<RecordingWarning>();
 	private pool: Piscina | undefined;
 	private ready = false;
 	private outstanding = 0;
 	private stopping: Promise<void> | undefined;
 
-	constructor(databasePath: string, enabled: boolean) {
-		this.databasePath = databasePath;
-		this.enabled = enabled;
-		if (enabled) this.startPool();
+	constructor(databasePath: string) {
+		this.startPool(databasePath);
 	}
 
 	readonly recordVisit = (visit: NewVisit): void => {
 		const pool = this.pool;
-		if (!this.enabled || !this.ready || !pool || this.stopping) return;
-		const bytes = visitBytes(visit);
-		if (this.outstanding >= MAX_OUTSTANDING || bytes > MAX_VISIT_BYTES) {
-			this.report();
-			return;
-		}
+		if (!this.ready || !pool || this.stopping) return;
+		if (this.dropVisitIfOverLimit(visit)) return;
 		this.submitVisit(pool, visit);
 	};
 
-	readonly setEnabled = (enabled: boolean): void => {
-		if (this.stopping) return;
-		this.enabled = enabled;
-		if (enabled) this.startPool();
-	};
-
 	readonly stop = (): Promise<void> => {
 		if (!this.stopping) {
-			this.stopping = this.pool ? closeVisitPool(this.pool).catch(this.report) : Promise.resolve();
+			this.stopping = this.pool
+				? closeVisitPool(this.pool).catch(this.reportUnavailable)
+				: Promise.resolve();
 		}
 		return this.stopping;
 	};
 
+	private dropVisitIfOverLimit(visit: NewVisit): boolean {
+		if (this.outstanding >= MAX_OUTSTANDING) {
+			this.report('queue');
+			return true;
+		}
+		if (exceedsVisitSizeLimit(visit)) {
+			this.report('size');
+			return true;
+		}
+		return false;
+	}
+
 	private submitVisit(pool: Piscina, visit: NewVisit): void {
 		this.outstanding += 1;
 		try {
@@ -70,10 +79,9 @@ class Recorder implements VisitRecorder {
 		}
 	}
 
-	private startPool(): void {
-		if (this.pool || this.stopping) return;
+	private startPool(databasePath: string): void {
 		try {
-			const pool = createVisitPool(this.databasePath);
+			const pool = createVisitPool(databasePath);
 			this.pool = pool;
 			pool.on('error', this.onPoolFailure);
 			const initialized = pool.run(null, { name: 'ready' });
@@ -89,7 +97,7 @@ class Recorder implements VisitRecorder {
 
 	private readonly onRecordFailure = (): void => {
 		this.onRecorded();
-		this.report();
+		this.reportUnavailable();
 	};
 
 	private readonly onReady = (): void => {
@@ -97,17 +105,21 @@ class Recorder implements VisitRecorder {
 	};
 
 	private readonly onPoolFailure = (): void => {
-		this.report();
+		this.reportUnavailable();
 		void this.stop();
 	};
 
-	private readonly report = (): void => {
-		if (this.reported) return;
-		this.reported = true;
+	private readonly reportUnavailable = (): void => {
+		this.report('unavailable');
+	};
+
+	private report(reason: RecordingWarning): void {
+		if (this.reported.has(reason)) return;
+		this.reported.add(reason);
 		try {
-			console.warn('Visit tracking unavailable; some visits may not be recorded');
+			console.warn(WARNINGS[reason]);
 		} catch {
 			return;
 		}
-	};
+	}
 }

apps/worker/e2e/startup.js

@@ -25,21 +25,20 @@ async function seedHistory(directory) {
 	const db = await openDb(join(directory, 'data/goto.sqlite'));
 	try {
 		await runMigrations(db);
+		const middle = new Date();
+		middle.setUTCMonth(middle.getUTCMonth() - 6);
 		await insertVisit(db, { ...visit, visited_at: Date.UTC(2020, 0, 1) });
+		await insertVisit(db, { ...visit, visited_at: middle.getTime() });
 		await insertVisit(db, visit);
 	} finally {
 		await db.destroy();
 	}
 }
 
-function visitState(directory) {
+function visitCount(directory) {
 	const db = new DatabaseSync(join(directory, 'data/goto.sqlite'), { readOnly: true });
 	try {
-		return {
-			count: db.prepare('SELECT count(*) AS count FROM visits').get().count,
-			collectAfter: db.prepare('SELECT collect_after FROM visit_collection WHERE id = 1').get()
-				.collect_after
-		};
+		return db.prepare('SELECT count(*) AS count FROM visits').get().count;
 	} finally {
 		db.close();
 	}
@@ -70,7 +69,7 @@ async function checkStartup(mode, port = 0) {
 			JSON.stringify({
 				site: { url: 'http://admin.example.test' },
 				auth: { credentials: { username: 'smoke', password: 'private-worker-smoke-password' } },
-				analytics: { retention_months: mode === 'retention-change' ? 3 : 0 },
+				analytics: { retention_months: mode === 'retention-change' ? 9 : 0 },
 				domains: {}
 			})
 		);
@@ -141,9 +140,8 @@ async function checkStartup(mode, port = 0) {
 			});
 			assert.equal(response.status, 200);
 			assert.deepEqual(await response.json(), { status: 'ok' });
-			const visits = visitState(directory);
-			assert.equal(visits.count, mode === 'retention-change' ? 1 : 0);
-			if (mode !== 'retention-change') assert.equal(visits.collectAfter, null);
+			const visits = visitCount(directory);
+			assert.equal(visits, mode === 'retention-change' ? 2 : 0);
 			const database = new DatabaseSync(join(directory, 'data/worker.sqlite'), {
 				readOnly: true
 			});
@@ -167,13 +165,24 @@ async function checkStartup(mode, port = 0) {
 				);
 				const path = join(directory, 'config/goto.yaml');
 				const configuration = JSON.parse(await readFile(path, 'utf8'));
+				configuration.analytics.retention_months = 3;
+				await saveConfiguration(path, configuration);
+				await waitUntil(() => visitCount(directory) === 1, stopped.signal);
 				configuration.analytics.retention_months = 0;
 				await saveConfiguration(path, configuration);
-				await waitUntil(() => visitState(directory).count === 0, stopped.signal);
-				assert.equal(visitState(directory).collectAfter, null);
-				configuration.analytics.retention_months = 9;
-				await saveConfiguration(path, configuration);
-				await waitUntil(() => visitState(directory).collectAfter !== null, stopped.signal);
+				await waitUntil(
+					() => output.includes('Analytics collection change requires coordinated restart'),
+					stopped.signal
+				);
+				assert.equal(visitCount(directory), 1, 'Live collection disable must not purge history');
+				const stillRunning = await fetch(`http://127.0.0.1:${ready.port}/healthz`, {
+					signal: AbortSignal.timeout(2000)
+				});
+				assert.equal(
+					stillRunning.status,
+					200,
+					'Collection change must not auto-restart one service'
+				);
 				child.kill('SIGTERM');
 			} else {
 				child.kill('SIGTERM');
@@ -215,5 +224,5 @@ try {
 	await new Promise((resolve) => occupied.close(resolve));
 }
 console.log(
-	'Compiled/source startup, config restart, idle zero policy, live retention, health, partial-startup cleanup, and shutdown passed'
+	'Compiled/source startup, config restart, idle zero policy, live positive retention, restart-only collection, health, partial-startup cleanup, and shutdown passed'
 );

apps/worker/src/services/config-reload.test.ts

@@ -4,10 +4,11 @@ import { afterEach, beforeEach, expect, test, vi } from 'vitest';
 import { applyConfig, watchConfiguration } from './config-reload.js';
 import type { WorkerContext } from './context.js';
 import type { ShutdownResources } from './lifecycle.js';
-import { configureRetention, requestRetention } from './retention.js';
+import { logger } from './logger.js';
+import { requestRetention } from './retention.js';
 
 vi.mock('./logger.js', () => ({ logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() } }));
-vi.mock('./retention.js', () => ({ configureRetention: vi.fn(), requestRetention: vi.fn() }));
+vi.mock('./retention.js', () => ({ requestRetention: vi.fn() }));
 vi.mock('@goto/config', async (importOriginal) => ({
 	...(await importOriginal<typeof import('@goto/config')>()),
 	startConfigReloading: vi.fn()
@@ -34,41 +35,62 @@ function context(): WorkerContext {
 	return { config: configuration().configuration, db: {} as WorkerContext['db'] };
 }
 
-beforeEach(() => {
-	vi.resetAllMocks();
-});
+beforeEach(() => vi.resetAllMocks());
 
 afterEach(() => {
 	vi.restoreAllMocks();
 	vi.useRealTimers();
 });
 
-test('requests restart for either database path and rejects the whole candidate', async () => {
+test('requests restart for either database path when collection remains enabled', async () => {
 	const initial = configuration();
 	const restart = vi.fn();
 	const ctx = context();
 	const accepted = ctx.config;
 	expect(await applyConfig(configuration(), initial, restart, ctx, engine)).toBe(true);
 	ctx.config = accepted;
-	expect(await applyConfig(configuration('./other.sqlite', 0), initial, restart, ctx, engine)).toBe(
+	expect(await applyConfig(configuration('./other.sqlite'), initial, restart, ctx, engine)).toBe(
 		false
 	);
 	expect(
-		await applyConfig(configuration(undefined, 0, './other.sqlite'), initial, restart, ctx, engine)
+		await applyConfig(configuration(undefined, 3, './other.sqlite'), initial, restart, ctx, engine)
 	).toBe(false);
 	expect(restart).toHaveBeenCalledTimes(2);
 	expect(ctx.config).toBe(accepted);
-	expect(configureRetention).not.toHaveBeenCalled();
+	expect(requestRetention).not.toHaveBeenCalled();
 });
 
-test('waits for zero-policy purge and reconciles a later re-enable', async () => {
+test.each([
+	[3, 0],
+	[0, 9]
+])(
+	'collection %s to %s rejects the whole candidate without auto-restart',
+	async (before, after) => {
+		const initial = configuration(undefined, before);
+		const next = configuration('./changed-worker.sqlite', after);
+		const ctx = context();
+		ctx.config = initial.configuration;
+		const restart = vi.fn();
+		expect(await applyConfig(next, initial, restart, ctx, engine)).toBe(false);
+		expect(ctx.config).toBe(initial.configuration);
+		expect(restart).not.toHaveBeenCalled();
+		expect(requestRetention).not.toHaveBeenCalled();
+		expect(logger.warn).toHaveBeenCalledExactlyOnceWith(
+			'Analytics collection change requires coordinated restart; keeping current configuration'
+		);
+	}
+);
+
+test('waits for a positive policy change to be pruned and accepts a later positive change', async () => {
+	const initial = configuration(undefined, 9);
 	const ctx = context();
-	const purged = Promise.withResolvers<void>();
-	vi.mocked(requestRetention).mockReturnValueOnce(purged.promise);
-	const disabled = configuration(undefined, 0);
-	const applying = applyConfig(disabled, configuration(), vi.fn(), ctx, engine);
+	ctx.config = initial.configuration;
+	const pruned = Promise.withResolvers<void>();
+	vi.mocked(requestRetention).mockReturnValueOnce(pruned.promise);
+	const next = configuration();
+	const applying = applyConfig(next, initial, vi.fn(), ctx, engine);
 	await Promise.resolve();
-	expect(ctx.config.analytics.retention_months).toBe(0);
+	expect(ctx.config.analytics.retention_months).toBe(3);
 	expect(requestRetention).toHaveBeenCalledExactlyOnceWith(ctx, engine);
 	let accepted = false;
 	void applying.then(() => {
@@ -76,32 +98,22 @@ test('waits for zero-policy purge and reconciles a later re-enable', async () =>
 	});
 	await Promise.resolve();
 	expect(accepted).toBe(false);
-	purged.resolve();
+	pruned.resolve();
 	expect(await applying).toBe(true);
-	expect(await applyConfig(configuration(undefined, 9), disabled, vi.fn(), ctx, engine)).toBe(true);
-	expect(ctx.config.analytics.retention_months).toBe(9);
+	expect(await applyConfig(configuration(undefined, 12), next, vi.fn(), ctx, engine)).toBe(true);
+	expect(ctx.config.analytics.retention_months).toBe(12);
 	expect(requestRetention).toHaveBeenCalledTimes(2);
 });
 
-test('reconciles positive policy changes immediately', async () => {
+test('a failed positive policy reconciliation requests shutdown', async () => {
+	const initial = configuration(undefined, 9);
 	const ctx = context();
-	await applyConfig(configuration(), configuration(undefined, 9), vi.fn(), ctx, engine);
-	expect(requestRetention).toHaveBeenCalledOnce();
-	await applyConfig(configuration(undefined, 12), configuration(), vi.fn(), ctx, engine);
-	expect(configureRetention).toHaveBeenCalledTimes(2);
-	expect(requestRetention).toHaveBeenCalledTimes(2);
-});
-
-test('a failed policy reconciliation keeps zero effective and requests shutdown', async () => {
-	const ctx = context();
-	const initial = configuration();
+	ctx.config = initial.configuration;
 	const failure = new Error('Private storage detail');
 	vi.mocked(requestRetention).mockRejectedValueOnce(failure);
 	const stop = vi.fn();
-	await expect(applyConfig(configuration(undefined, 0), initial, stop, ctx, engine)).rejects.toBe(
-		failure
-	);
-	expect(ctx.config.analytics.retention_months).toBe(0);
+	await expect(applyConfig(configuration(), initial, stop, ctx, engine)).rejects.toBe(failure);
+	expect(ctx.config.analytics.retention_months).toBe(3);
 	expect(stop).toHaveBeenCalledOnce();
 });
 

apps/worker/src/services/config-reload.ts

@@ -3,7 +3,7 @@ import { logger } from './logger.js';
 import type { ShutdownResources } from './lifecycle.js';
 import type { OpenWorkflow } from 'openworkflow';
 import type { WorkerContext } from './context.js';
-import { configureRetention, requestRetention } from './retention.js';
+import { requestRetention } from './retention.js';
 
 interface ConfigurationSource {
 	path: string;
@@ -18,6 +18,16 @@ export async function applyConfig(
 	context: WorkerContext,
 	engine: OpenWorkflow
 ): Promise<boolean> {
+	const months = next.configuration.analytics.retention_months;
+	const previousMonths = previous.configuration.analytics.retention_months;
+	const collectionEnabled = months !== 0;
+	const previousCollectionEnabled = previousMonths !== 0;
+	if (collectionEnabled !== previousCollectionEnabled) {
+		logger.warn(
+			'Analytics collection change requires coordinated restart; keeping current configuration'
+		);
+		return false;
+	}
 	const nextDatabase = next.configuration.database;
 	const previousDatabase = previous.configuration.database;
 	if (
@@ -30,11 +40,8 @@ export async function applyConfig(
 	}
 	const currentMonths = context.config.analytics.retention_months;
 	context.config = next.configuration;
-	const months = next.configuration.analytics.retention_months;
-	const previousMonths = previous.configuration.analytics.retention_months;
 	try {
 		if (months !== previousMonths || months !== currentMonths) {
-			await configureRetention(context, engine);
 			await requestRetention(context, engine);
 		}
 	} catch (error) {

apps/worker/src/services/retention.test.ts

@@ -1,10 +1,10 @@
 import { validateConfig } from '@goto/config';
-import { openDb, runMigrations, setVisitCollection } from '@goto/db';
+import { openDb, runMigrations } from '@goto/db';
 import type { OpenWorkflow } from 'openworkflow';
 import { expect, test, vi } from 'vitest';
 import {
 	applyRetention,
-	configureRetention,
+	configureRetentionSchedule,
 	requestRetention,
 	retentionCutoff
 } from './retention.js';
@@ -58,7 +58,7 @@ test('zero pauses the native schedule and queues one purge only when data exists
 	const sdk = engine as unknown as OpenWorkflow;
 	try {
 		ctx.config.analytics.retention_months = 0;
-		await configureRetention(ctx, sdk);
+		await configureRetentionSchedule(ctx, sdk);
 		expect(engine.scheduleWorkflow).toHaveBeenCalledExactlyOnceWith(
 			retainVisits,
 			retentionSchedule.input,
@@ -88,9 +88,6 @@ test('zero pauses the native schedule and queues one purge only when data exists
 		expect(engine.runWorkflow).toHaveBeenCalledExactlyOnceWith(retainVisits, {});
 		await applyRetention(ctx);
 		expect(await ctx.db.selectFrom('visits').selectAll().execute()).toEqual([]);
-		expect(
-			await ctx.db.selectFrom('visit_collection').select('collect_after').executeTakeFirstOrThrow()
-		).toEqual({ collect_after: null });
 	} finally {
 		await ctx.db.destroy();
 	}
@@ -109,7 +106,7 @@ test('positive retention resumes the schedule and prunes every expired batch usi
 				{ ...visit, visited_at: cutoff }
 			])
 			.execute();
-		await configureRetention(ctx, engine as unknown as OpenWorkflow);
+		await configureRetentionSchedule(ctx, engine as unknown as OpenWorkflow);
 		expect(engine.schedules.resume).toHaveBeenCalledExactlyOnceWith(retentionSchedule.id);
 		expect(engine.schedules.pause).not.toHaveBeenCalled();
 		await applyRetention(ctx, now);
@@ -121,54 +118,6 @@ test('positive retention resumes the schedule and prunes every expired batch usi
 	}
 });
 
-test('finishes an interrupted zero-policy purge before reopening collection', async () => {
-	const ctx = await context();
-	try {
-		const now = Date.now();
-		await setVisitCollection(ctx.db, false);
-		await ctx.db
-			.insertInto('visits')
-			.values({ ...visit, visited_at: now })
-			.execute();
-		ctx.config.analytics.retention_months = 9;
-		await applyRetention(ctx, now);
-		expect(await ctx.db.selectFrom('visits').selectAll().execute()).toEqual([]);
-		const state = await ctx.db
-			.selectFrom('visit_collection')
-			.select('collect_after')
-			.executeTakeFirstOrThrow();
-		expect(state.collect_after).toBeTypeOf('number');
-	} finally {
-		await ctx.db.destroy();
-	}
-});
-
-test('a positive job waiting for its connection cannot reopen collection after zero', async () => {
-	const ctx = await context();
-	const acquired = Promise.withResolvers<void>();
-	const release = Promise.withResolvers<void>();
-	try {
-		await setVisitCollection(ctx.db, false);
-		const holding = ctx.db.transaction().execute(async () => {
-			acquired.resolve();
-			await release.promise;
-		});
-		await Promise.race([acquired.promise, holding]);
-		const applying = applyRetention(ctx);
-		ctx.config.analytics.retention_months = 0;
-		release.resolve();
-		await Promise.all([holding, applying]);
-		const state = await ctx.db
-			.selectFrom('visit_collection')
-			.select('collect_after')
-			.executeTakeFirstOrThrow();
-		expect(state.collect_after).toBeNull();
-	} finally {
-		release.resolve();
-		await ctx.db.destroy();
-	}
-});
-
 test('handles leap-day boundaries without overflowing into March', () => {
 	expect(new Date(retentionCutoff(3, Date.parse('2024-05-31T01:00:00Z'))).toISOString()).toBe(
 		'2024-02-29T01:00:00.000Z'

apps/worker/src/services/retention.ts

@@ -1,18 +1,17 @@
 import { setImmediate } from 'node:timers/promises';
 import type { GotoConfiguration } from '@goto/config';
-import { pruneVisits, purgeVisits, purgeVisitsIfDisabled, setVisitCollection } from '@goto/db';
+import { pruneVisits, purgeVisits } from '@goto/db';
 import type { OpenWorkflow } from 'openworkflow';
 import { retainVisits, retentionSchedule } from '../workflows/specs.js';
 import type { WorkerContext } from './context.js';
 
 export type RetentionMonths = GotoConfiguration['analytics']['retention_months'];
 
-export async function configureRetention(
+export async function configureRetentionSchedule(
 	context: WorkerContext,
 	engine: OpenWorkflow
 ): Promise<void> {
 	const enabled = context.config.analytics.retention_months !== 0;
-	if (!enabled) await setVisitCollection(context.db, false);
 	await engine.scheduleWorkflow(retainVisits, retentionSchedule.input, retentionSchedule);
 	if (enabled) await engine.schedules.resume(retentionSchedule.id);
 	else await engine.schedules.pause(retentionSchedule.id);
@@ -33,16 +32,9 @@ export async function requestRetention(
 export async function applyRetention(context: WorkerContext, now = Date.now()): Promise<void> {
 	const months = context.config.analytics.retention_months;
 	if (months === 0) {
-		await setVisitCollection(context.db, false);
 		await purgeVisits(context.db);
 		return;
 	}
-	await context.db.transaction().execute(async (transaction) => {
-		await purgeVisitsIfDisabled(transaction);
-		if (context.config.analytics.retention_months !== 0) {
-			await setVisitCollection(transaction, true);
-		}
-	});
 	const before = retentionCutoff(months, now);
 	while (await pruneVisits(context.db, before)) {
 		await setImmediate();

apps/worker/src/services/startup.test.ts

@@ -53,9 +53,6 @@ test('starts an idle zero-policy service, migrates visit storage, and exposes he
 		const db = await openDb(join(directory, 'data/visits.sqlite'), { readOnly: true });
 		try {
 			expect(await db.selectFrom('visits').selectAll().execute()).toEqual([]);
-			expect(
-				await db.selectFrom('visit_collection').select('collect_after').executeTakeFirstOrThrow()
-			).toEqual({ collect_after: null });
 		} finally {
 			await db.destroy();
 		}

apps/worker/src/services/startup.ts

@@ -11,7 +11,7 @@ import { createShutdown, type ShutdownResources } from './lifecycle.js';
 import { watchConfiguration } from './config-reload.js';
 import { logger } from './logger.js';
 import type { WorkerContext } from './context.js';
-import { configureRetention, requestRetention } from './retention.js';
+import { configureRetentionSchedule, requestRetention } from './retention.js';
 import { registerRetentionWorkflow } from '../workflows/retain-visits.js';
 
 export interface WorkerService {
@@ -80,7 +80,7 @@ async function startWorkflowWorker(
 	const engineLogger = createEngineLogger(logger);
 	const engine = new OpenWorkflow({ backend, logger: engineLogger });
 	registerRetentionWorkflow(engine, context);
-	await configureRetention(context, engine);
+	await configureRetentionSchedule(context, engine);
 	const workflowWorker = engine.newWorker({ concurrency: 1 });
 	resources.workflowWorker = workflowWorker;
 	await workflowWorker.start();

apps/worker/src/workflows/retain-visits.test.ts

@@ -55,19 +55,12 @@ test('scheduled cleanup uses durable jitter; startup and policy runs do not', as
 	expect(applyRetention).toHaveBeenCalledExactlyOnceWith(ctx);
 });
 
-test('a scheduled run does no cleanup if collection is disabled during its delay', async () => {
+test('a prior scheduled run does no cleanup when the service starts with collection disabled', async () => {
 	const ctx = context();
-	const { step, sleep } = steps();
-	const sleeping = Promise.withResolvers<void>();
-	const resumed = Promise.withResolvers<void>();
-	sleep.mockImplementationOnce(async () => {
-		sleeping.resolve();
-		await resumed.promise;
-	});
-	const running = runRetentionWorkflow(ctx, { scheduled: true }, step);
-	await sleeping.promise;
 	ctx.config.analytics.retention_months = 0;
-	resumed.resolve();
-	await running;
+	const { step, names, sleep } = steps();
+	await runRetentionWorkflow(ctx, { scheduled: true }, step);
+	expect(names).toEqual([]);
+	expect(sleep).not.toHaveBeenCalled();
 	expect(applyRetention).not.toHaveBeenCalled();
 });

apps/worker/src/workflows/retain-visits.ts

@@ -30,7 +30,6 @@ export async function runRetentionWorkflow(
 		const seconds = await step.run({ name: 'choose-jitter' }, chooseJitter);
 		const duration: `${number}s` = `${seconds}s`;
 		await step.sleep('retention-jitter', duration);
-		if (collectionDisabled(context)) return 'completed';
 	}
 	function applyPolicy(): Promise<void> {
 		return applyRetention(context);

packages/config/src/config.schema.json

@@ -290,7 +290,7 @@
 			"properties": {
 				"retention_months": {
 					"default": 3,
-					"description": "Calendar months of analytics history. Zero disables collection and purges retained visits.",
+					"description": "Calendar months of analytics history. Zero disables collection and purges retained visits at startup. Stop both services before changing between zero and positive retention, then restart. Positive retention changes reload live.",
 					"anyOf": [
 						{ "const": 0, "type": "number" },
 						{ "const": 3, "type": "number" },

packages/config/src/schema.ts

@@ -191,7 +191,7 @@ const AnalyticsSchema = Type.Object(
 			{
 				default: 3,
 				description:
-					'Calendar months of analytics history. Zero disables collection and purges retained visits.'
+					'Calendar months of analytics history. Zero disables collection and purges retained visits at startup. Stop both services before changing between zero and positive retention, then restart. Positive retention changes reload live.'
 			}
 		)
 	},

packages/db/src/index.ts

@@ -6,13 +6,7 @@ import { NodeSqliteDialect } from './node-sqlite-dialect.js';
 import type { Database } from './types.js';
 
 export { runMigrations } from './migrate.js';
-export {
-	insertVisit,
-	pruneVisits,
-	purgeVisits,
-	purgeVisitsIfDisabled,
-	setVisitCollection
-} from './visits.js';
+export { insertVisit, pruneVisits, purgeVisits } from './visits.js';
 export type { ClassifiedVisit, Database, NewVisit, Visit } from './types.js';
 
 export async function openDb(

packages/db/src/migrations/0004_remove_visit_collection.ts

@@ -0,0 +1,15 @@
+import { sql, type Kysely, type Migration } from 'kysely';
+
+export const removeVisitCollection: Migration = {
+	async up(db: Kysely<unknown>): Promise<void> {
+		await db.schema.dropTable('visit_collection').execute();
+	},
+	async down(db: Kysely<unknown>): Promise<void> {
+		await db.schema
+			.createTable('visit_collection')
+			.addColumn('id', 'integer', (column) => column.primaryKey())
+			.addColumn('collect_after', 'integer')
+			.execute();
+		await sql`INSERT INTO visit_collection (id, collect_after) VALUES (1, 0)`.execute(db);
+	}
+};

packages/db/src/migrations/index.ts

@@ -1,9 +1,11 @@
 import { initial } from './0001_visits.js';
 import { addVisitClassification } from './0002_visit_classification.js';
 import { visitRetention } from './0003_visit_retention.js';
+import { removeVisitCollection } from './0004_remove_visit_collection.js';
 
 export const migrations = {
 	'0001_visits': initial,
 	'0002_visit_classification': addVisitClassification,
-	'0003_visit_retention': visitRetention
+	'0003_visit_retention': visitRetention,
+	'0004_remove_visit_collection': removeVisitCollection
 };

packages/db/src/storage.test.ts

@@ -11,7 +11,6 @@ import {
 	pruneVisits,
 	purgeVisits,
 	runMigrations,
-	setVisitCollection,
 	type ClassifiedVisit
 } from './index.js';
 import { migrations } from './migrations/index.js';
@@ -52,8 +51,12 @@ test('opens a real WAL database and applies versioned migrations once', async ()
 		expect(await runMigrations(db)).toEqual([
 			'0001_visits',
 			'0002_visit_classification',
-			'0003_visit_retention'
+			'0003_visit_retention',
+			'0004_remove_visit_collection'
 		]);
+		const tables = await db.introspection.getTables();
+		const names = tables.map((table) => table.name);
+		expect(names).not.toContain('visit_collection');
 		expect(await runMigrations(db)).toEqual([]);
 	} finally {
 		await db.destroy();
@@ -63,7 +66,7 @@ test('opens a real WAL database and applies versioned migrations once', async ()
 test('persists classified visits for the same logical path on different domains', async () => {
 	const path = join(directory, 'visits.sqlite');
 	const writer = await openDb(path);
-	const ids: (number | undefined)[] = [];
+	const ids: number[] = [];
 	try {
 		await runMigrations(writer);
 		ids.push(await insertVisit(writer, visit));
@@ -119,7 +122,11 @@ test('adds classification columns without changing existing visit data', async (
 			VALUES (${visit.domain}, ${visit.path}, ${visit.visited_at}, ${visit.ip}, ${visit.user_agent}, ${visit.accept_language})
 		`.execute(db);
 
-		expect(await runMigrations(db)).toEqual(['0002_visit_classification', '0003_visit_retention']);
+		expect(await runMigrations(db)).toEqual([
+			'0002_visit_classification',
+			'0003_visit_retention',
+			'0004_remove_visit_collection'
+		]);
 		expect(await db.selectFrom('visits').selectAll().executeTakeFirstOrThrow()).toEqual({
 			...visit,
 			id: 1,
@@ -154,7 +161,8 @@ test('a failed migration rolls back schema and bookkeeping so startup can retry'
 		expect(await runMigrations(db)).toEqual([
 			'0001_visits',
 			'0002_visit_classification',
-			'0003_visit_retention'
+			'0003_visit_retention',
+			'0004_remove_visit_collection'
 		]);
 		expect(await runMigrations(db)).toEqual([]);
 	} finally {
@@ -189,31 +197,6 @@ test('prunes expired visits in bounded passes and zero-retention purge removes t
 	}
 });
 
-test('a collection fence rejects queued writes after purge and after re-enabling', async () => {
-	const path = join(directory, 'collection.sqlite');
-	const writer = await openDb(path);
-	await runMigrations(writer);
-	const maintenance = await openDb(path);
-	try {
-		await insertVisit(writer, visit);
-		await setVisitCollection(maintenance, false);
-		await purgeVisits(maintenance);
-		expect(await insertVisit(writer, visit)).toBeUndefined();
-		const resumedAt = visit.visited_at + 10;
-		await setVisitCollection(maintenance, true, resumedAt);
-		expect(await insertVisit(writer, visit)).toBeUndefined();
-		expect(await insertVisit(writer, { ...visit, visited_at: resumedAt })).toBeUndefined();
-		await setVisitCollection(maintenance, true, resumedAt + 100);
-		const resumed = { ...visit, visited_at: resumedAt + 1 };
-		expect(await insertVisit(writer, resumed)).toBeTypeOf('number');
-		const rows = await writer.selectFrom('visits').select('visited_at').execute();
-		expect(rows).toEqual([{ visited_at: resumed.visited_at }]);
-	} finally {
-		await maintenance.destroy();
-		await writer.destroy();
-	}
-});
-
 test('destroy closes an opened database without an intervening query', async () => {
 	const close = vi.spyOn(DatabaseSync.prototype, 'close');
 	try {

packages/db/src/types.ts

@@ -12,14 +12,8 @@ export interface VisitTable {
 	os: string;
 }
 
-export interface VisitCollectionTable {
-	id: number;
-	collect_after: number | null;
-}
-
 export interface Database {
 	visits: VisitTable;
-	visit_collection: VisitCollectionTable;
 }
 
 export type ClassifiedVisit = Omit<Insertable<VisitTable>, 'id'>;

packages/db/src/visits.ts

@@ -1,33 +1,9 @@
-import { sql, type Kysely } from 'kysely';
+import type { Kysely } from 'kysely';
 import type { ClassifiedVisit, Database } from './types.js';
 
-export async function insertVisit(
-	db: Kysely<Database>,
-	visit: ClassifiedVisit
-): Promise<number | undefined> {
-	const insert = sql<{ id: number }>`
-		INSERT INTO visits (domain, path, visited_at, ip, user_agent, accept_language, browser, os)
-		SELECT ${visit.domain}, ${visit.path}, ${visit.visited_at}, ${visit.ip},
-			${visit.user_agent}, ${visit.accept_language}, ${visit.browser}, ${visit.os}
-		FROM visit_collection
-		WHERE id = 1 AND collect_after < ${visit.visited_at}
-		RETURNING id
-	`;
-	const result = await insert.execute(db);
-	return result.rows[0]?.id;
-}
-
-export async function setVisitCollection(
-	db: Kysely<Database>,
-	enabled: boolean,
-	now = Date.now()
-): Promise<void> {
-	const collectAfter = enabled ? sql<number>`coalesce(collect_after, ${now})` : null;
-	await db
-		.updateTable('visit_collection')
-		.set({ collect_after: collectAfter })
-		.where('id', '=', 1)
-		.execute();
+export async function insertVisit(db: Kysely<Database>, visit: ClassifiedVisit): Promise<number> {
+	const row = await db.insertInto('visits').values(visit).returning('id').executeTakeFirstOrThrow();
+	return row.id;
 }
 
 export async function pruneVisits(db: Kysely<Database>, before: number): Promise<boolean> {
@@ -43,20 +19,6 @@ export async function pruneVisits(db: Kysely<Database>, before: number): Promise
 	return result.numDeletedRows === BigInt(batchSize);
 }
 
-export async function purgeVisitsIfDisabled(db: Kysely<Database>): Promise<void> {
-	await db
-		.deleteFrom('visits')
-		.where((expression) => {
-			const disabled = expression
-				.selectFrom('visit_collection')
-				.select('id')
-				.where('id', '=', 1)
-				.where('collect_after', 'is', null);
-			return expression.exists(disabled);
-		})
-		.execute();
-}
-
 export async function purgeVisits(db: Kysely<Database>): Promise<void> {
 	await db.deleteFrom('visits').execute();
 }