From b07574077142245a5e7a24a80267b826cdfedcac Mon Sep 17 00:00:00 2001 From: eleith Date: Thu, 5 Nov 2026 13:30:51 -0800 Subject: [PATCH] refactor(analytics): make collection changes restart-only --- .changeset/restart-only-collection.md | 7 ++ apps/web/src/lib/server/application.test.ts | 41 +++++++-- apps/web/src/lib/server/application.ts | 20 +++-- .../src/lib/server/visits/recorder.test.ts | 43 ++++----- apps/web/src/lib/server/visits/recorder.ts | 88 +++++++++++-------- apps/worker/e2e/startup.js | 41 +++++---- .../worker/src/services/config-reload.test.ts | 78 +++++++++------- apps/worker/src/services/config-reload.ts | 15 +++- apps/worker/src/services/retention.test.ts | 59 +------------ apps/worker/src/services/retention.ts | 12 +-- apps/worker/src/services/startup.test.ts | 3 - apps/worker/src/services/startup.ts | 4 +- .../src/workflows/retain-visits.test.ts | 17 ++-- apps/worker/src/workflows/retain-visits.ts | 1 - packages/config/src/config.schema.json | 2 +- packages/config/src/schema.ts | 2 +- packages/db/src/index.ts | 8 +- .../0004_remove_visit_collection.ts | 15 ++++ packages/db/src/migrations/index.ts | 4 +- packages/db/src/storage.test.ts | 43 +++------ packages/db/src/types.ts | 6 -- packages/db/src/visits.ts | 46 +--------- 22 files changed, 248 insertions(+), 307 deletions(-) create mode 100644 .changeset/restart-only-collection.md create mode 100644 packages/db/src/migrations/0004_remove_visit_collection.ts diff --git a/.changeset/restart-only-collection.md b/.changeset/restart-only-collection.md new file mode 100644 index 0000000..a6af12d --- /dev/null +++ b/.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. diff --git a/apps/web/src/lib/server/application.test.ts b/apps/web/src/lib/server/application.test.ts index 1beb0ac..a080514 100644 --- a/apps/web/src/lib/server/application.test.ts +++ b/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?.(); }); diff --git a/apps/web/src/lib/server/application.ts b/apps/web/src/lib/server/application.ts index 7f966b8..1290d1c 100644 --- a/apps/web/src/lib/server/application.ts +++ b/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 { return loadConfigFile(path, captured); @@ -59,14 +59,18 @@ export async function createApplication(options: ConfigOptions = {}): Promise { 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(); const poolClosed = Promise.withResolvers(); - 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); diff --git a/apps/web/src/lib/server/visits/recorder.ts b/apps/web/src/lib/server/visits/recorder.ts index a549ffe..33a092a 100644 --- a/apps/web/src/lib/server/visits/recorder.ts +++ b/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; } -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(); private pool: Piscina | undefined; private ready = false; private outstanding = 0; private stopping: Promise | 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 => { 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; } - }; + } } diff --git a/apps/worker/e2e/startup.js b/apps/worker/e2e/startup.js index 215797c..a44f7fc 100644 --- a/apps/worker/e2e/startup.js +++ b/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' ); diff --git a/apps/worker/src/services/config-reload.test.ts b/apps/worker/src/services/config-reload.test.ts index c6c6334..d95a8af 100644 --- a/apps/worker/src/services/config-reload.test.ts +++ b/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()), 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(); - 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(); + 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(); }); diff --git a/apps/worker/src/services/config-reload.ts b/apps/worker/src/services/config-reload.ts index 7d6b227..339e3de 100644 --- a/apps/worker/src/services/config-reload.ts +++ b/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 { + 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) { diff --git a/apps/worker/src/services/retention.test.ts b/apps/worker/src/services/retention.test.ts index cd077fa..4981010 100644 --- a/apps/worker/src/services/retention.test.ts +++ b/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(); - const release = Promise.withResolvers(); - 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' diff --git a/apps/worker/src/services/retention.ts b/apps/worker/src/services/retention.ts index f285aea..210e659 100644 --- a/apps/worker/src/services/retention.ts +++ b/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 { 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 { 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(); diff --git a/apps/worker/src/services/startup.test.ts b/apps/worker/src/services/startup.test.ts index 1472c29..c59b143 100644 --- a/apps/worker/src/services/startup.test.ts +++ b/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(); } diff --git a/apps/worker/src/services/startup.ts b/apps/worker/src/services/startup.ts index 005cce1..c70be28 100644 --- a/apps/worker/src/services/startup.ts +++ b/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(); diff --git a/apps/worker/src/workflows/retain-visits.test.ts b/apps/worker/src/workflows/retain-visits.test.ts index 277fc1c..4863c3c 100644 --- a/apps/worker/src/workflows/retain-visits.test.ts +++ b/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(); - const resumed = Promise.withResolvers(); - 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(); }); diff --git a/apps/worker/src/workflows/retain-visits.ts b/apps/worker/src/workflows/retain-visits.ts index 8a5b642..fb7668c 100644 --- a/apps/worker/src/workflows/retain-visits.ts +++ b/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 { return applyRetention(context); diff --git a/packages/config/src/config.schema.json b/packages/config/src/config.schema.json index e38a1de..70f4b3d 100644 --- a/packages/config/src/config.schema.json +++ b/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" }, diff --git a/packages/config/src/schema.ts b/packages/config/src/schema.ts index 73fb769..a2ecbb1 100644 --- a/packages/config/src/schema.ts +++ b/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.' } ) }, diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index 15bc07b..ac53c74 100644 --- a/packages/db/src/index.ts +++ b/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( diff --git a/packages/db/src/migrations/0004_remove_visit_collection.ts b/packages/db/src/migrations/0004_remove_visit_collection.ts new file mode 100644 index 0000000..2fe58c6 --- /dev/null +++ b/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): Promise { + await db.schema.dropTable('visit_collection').execute(); + }, + async down(db: Kysely): Promise { + 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); + } +}; diff --git a/packages/db/src/migrations/index.ts b/packages/db/src/migrations/index.ts index 22d8cf9..ef6c571 100644 --- a/packages/db/src/migrations/index.ts +++ b/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 }; diff --git a/packages/db/src/storage.test.ts b/packages/db/src/storage.test.ts index 95e6e14..481afb5 100644 --- a/packages/db/src/storage.test.ts +++ b/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 { diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index b94c5e4..bb66cf9 100644 --- a/packages/db/src/types.ts +++ b/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, 'id'>; diff --git a/packages/db/src/visits.ts b/packages/db/src/visits.ts index ecfa43d..4133c55 100644 --- a/packages/db/src/visits.ts +++ b/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, - visit: ClassifiedVisit -): Promise { - 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, - enabled: boolean, - now = Date.now() -): Promise { - const collectAfter = enabled ? sql`coalesce(collect_after, ${now})` : null; - await db - .updateTable('visit_collection') - .set({ collect_after: collectAfter }) - .where('id', '=', 1) - .execute(); +export async function insertVisit(db: Kysely, visit: ClassifiedVisit): Promise { + const row = await db.insertInto('visits').values(visit).returning('id').executeTakeFirstOrThrow(); + return row.id; } export async function pruneVisits(db: Kysely, before: number): Promise { @@ -43,20 +19,6 @@ export async function pruneVisits(db: Kysely, before: number): Promise return result.numDeletedRows === BigInt(batchSize); } -export async function purgeVisitsIfDisabled(db: Kysely): Promise { - 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): Promise { await db.deleteFrom('visits').execute(); } -- 2.54.0