From 81552a076abe3e05d24fa69b6bd83d768f0e8644 Mon Sep 17 00:00:00 2001 From: Polliog Date: Fri, 17 Jul 2026 13:41:48 +0200 Subject: [PATCH 01/17] add ingestion clock skew tracker --- packages/backend/src/config/index.ts | 8 ++ .../backend/src/modules/ingestion/skew.ts | 66 +++++++++++++++++ .../backend/src/modules/metering/types.ts | 4 +- .../src/tests/modules/ingestion/skew.test.ts | 73 +++++++++++++++++++ 4 files changed, 150 insertions(+), 1 deletion(-) create mode 100644 packages/backend/src/modules/ingestion/skew.ts create mode 100644 packages/backend/src/tests/modules/ingestion/skew.test.ts diff --git a/packages/backend/src/config/index.ts b/packages/backend/src/config/index.ts index 0b135b4b..c05506a1 100644 --- a/packages/backend/src/config/index.ts +++ b/packages/backend/src/config/index.ts @@ -66,6 +66,14 @@ const configSchema = z.object({ METERING_FLUSH_INTERVAL_MS: z.string().default('5000').transform(Number), METERING_FLUSH_MAX_BUFFER: z.string().default('500').transform(Number), + // Ingestion clock skew detection (#279). A client-supplied `time` far from the + // server clock is stored as sent but can never be seen by a threshold alert + // rule, whose largest window is 24h. Detected and counted, never rejected: + // backfilling historical logs is legitimate. Set either to 0 to disable that + // direction. + INGESTION_SKEW_PAST_MS: z.string().default('86400000').transform(Number), // 24h: the max alert time_window + INGESTION_SKEW_FUTURE_MS: z.string().default('300000').transform(Number), // 5m: room for NTP jitter + // Capability usage-quota evaluator (#214). Periodic job that flags over-quota orgs. QUOTA_EVALUATOR_ENABLED: z.string().default('true').transform((val) => val === 'true'), QUOTA_EVALUATOR_INTERVAL_MS: z.string().default('60000').transform(Number), diff --git a/packages/backend/src/modules/ingestion/skew.ts b/packages/backend/src/modules/ingestion/skew.ts new file mode 100644 index 00000000..baf7457a --- /dev/null +++ b/packages/backend/src/modules/ingestion/skew.ts @@ -0,0 +1,66 @@ +/** + * Ingestion clock skew detection (#279). + * + * A log whose client-supplied `time` sits far from the server clock is stored + * and is visible in the UI, but threshold alert rules count logs in + * [now - time_window, now] and so can never see it. The largest configurable + * window is 24h (alerts/routes.ts), which is where the past default comes from: + * past that point, no threshold rule in the product can match the log. + * + * Observation only. Skewed records are written unchanged. + */ +import { config } from '../../config/index.js'; + +export interface SkewSummary { + /** Number of skewed records in the batch. */ + count: number; + /** Worst past delta seen, in ms. 0 when nothing was skewed into the past. */ + maxPastMs: number; + /** Worst future delta seen, in ms. 0 when nothing was skewed into the future. */ + maxFutureMs: number; +} + +export interface SkewTracker { + observe(time: Date | undefined): void; + summary(): SkewSummary | null; +} + +/** + * Single-pass tracker. `now` is read once per batch by the caller so that a + * large batch is measured against one instant, not a drifting one. + */ +export function createSkewTracker(now: number): SkewTracker { + const pastMs = config.INGESTION_SKEW_PAST_MS; + const futureMs = config.INGESTION_SKEW_FUTURE_MS; + + let count = 0; + let maxPastMs = 0; + let maxFutureMs = 0; + + return { + observe(time: Date | undefined): void { + if (!time) return; + + // NaN for an Invalid Date. Both comparisons below are false for NaN, so a + // malformed timestamp is never miscounted as skew: that is a separate + // problem and counting it here would make this signal dishonest. + const delta = now - time.getTime(); + + if (pastMs > 0 && delta > pastMs) { + count++; + if (delta > maxPastMs) maxPastMs = delta; + return; + } + + if (futureMs > 0 && delta < -futureMs) { + count++; + const ahead = -delta; + if (ahead > maxFutureMs) maxFutureMs = ahead; + } + }, + + summary(): SkewSummary | null { + return count > 0 ? { count, maxPastMs, maxFutureMs } : null; + }, + }; +} diff --git a/packages/backend/src/modules/metering/types.ts b/packages/backend/src/modules/metering/types.ts index 34680053..946559c4 100644 --- a/packages/backend/src/modules/metering/types.ts +++ b/packages/backend/src/modules/metering/types.ts @@ -14,7 +14,9 @@ export type MeteringEventType = | 'ingestion.pii_rejected' | 'ingestion.detection_enqueue_failed' | 'ingestion.exception_enqueue_failed' - | 'ingestion.identifier_failed'; + | 'ingestion.identifier_failed' + // Clock skew at ingestion (#279): not billed, surfaced in admin stats and per project. + | 'ingestion.timestamp_skew'; export interface MeteringEvent { type: MeteringEventType; diff --git a/packages/backend/src/tests/modules/ingestion/skew.test.ts b/packages/backend/src/tests/modules/ingestion/skew.test.ts new file mode 100644 index 00000000..34d20fd4 --- /dev/null +++ b/packages/backend/src/tests/modules/ingestion/skew.test.ts @@ -0,0 +1,73 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; + +// The tracker reads thresholds from config at construction, so config is mocked +// per test rather than importing the real .env.test values. +const mockConfig = { INGESTION_SKEW_PAST_MS: 86400000, INGESTION_SKEW_FUTURE_MS: 300000 }; +vi.mock('../../../config/index.js', () => ({ config: mockConfig })); + +const { createSkewTracker } = await import('../../../modules/ingestion/skew.js'); + +const NOW = new Date('2026-07-17T12:00:00.000Z').getTime(); +const ago = (ms: number) => new Date(NOW - ms); + +describe('createSkewTracker', () => { + beforeEach(() => { + mockConfig.INGESTION_SKEW_PAST_MS = 86400000; + mockConfig.INGESTION_SKEW_FUTURE_MS = 300000; + }); + + it('returns null when nothing is skewed', () => { + const t = createSkewTracker(NOW); + t.observe(ago(0)); + t.observe(ago(60_000)); + t.observe(ago(86_399_000)); + expect(t.summary()).toBeNull(); + }); + + it('counts a log older than the past threshold and records the worst delta', () => { + const t = createSkewTracker(NOW); + t.observe(ago(97_200_000)); // 27h in the past, the #279 case + t.observe(ago(90_000_000)); // 25h, skewed but less extreme + expect(t.summary()).toEqual({ count: 2, maxPastMs: 97_200_000, maxFutureMs: 0 }); + }); + + it('counts a log ahead of the future threshold and records the worst delta', () => { + const t = createSkewTracker(NOW); + t.observe(ago(-600_000)); // 10m in the future + expect(t.summary()).toEqual({ count: 1, maxPastMs: 0, maxFutureMs: 600_000 }); + }); + + it('tracks both directions in one batch', () => { + const t = createSkewTracker(NOW); + t.observe(ago(97_200_000)); + t.observe(ago(-600_000)); + t.observe(ago(1000)); + expect(t.summary()).toEqual({ count: 2, maxPastMs: 97_200_000, maxFutureMs: 600_000 }); + }); + + it('does not count an Invalid Date as skew', () => { + const t = createSkewTracker(NOW); + t.observe(new Date('garbage')); + expect(t.summary()).toBeNull(); + }); + + it('does not count a missing time', () => { + const t = createSkewTracker(NOW); + t.observe(undefined); + expect(t.summary()).toBeNull(); + }); + + it('disables past detection when the threshold is 0', () => { + mockConfig.INGESTION_SKEW_PAST_MS = 0; + const t = createSkewTracker(NOW); + t.observe(ago(97_200_000)); + expect(t.summary()).toBeNull(); + }); + + it('disables future detection when the threshold is 0', () => { + mockConfig.INGESTION_SKEW_FUTURE_MS = 0; + const t = createSkewTracker(NOW); + t.observe(ago(-600_000)); + expect(t.summary()).toBeNull(); + }); +}); From e270b2a5505a14b2c7d78a9f2e49d451014c4042 Mon Sep 17 00:00:00 2001 From: Polliog Date: Fri, 17 Jul 2026 13:43:16 +0200 Subject: [PATCH 02/17] detect clock skew on log ingestion --- .../backend/src/modules/ingestion/service.ts | 25 +++++++- .../ingestion/ingestion-service.test.ts | 64 +++++++++++++++++++ 2 files changed, 87 insertions(+), 2 deletions(-) diff --git a/packages/backend/src/modules/ingestion/service.ts b/packages/backend/src/modules/ingestion/service.ts index a3f814e7..b0dc425d 100644 --- a/packages/backend/src/modules/ingestion/service.ts +++ b/packages/backend/src/modules/ingestion/service.ts @@ -17,6 +17,7 @@ import { hooks, HookExecutionError } from '../../hooks/index.js'; import type { BeforeIngestContext } from '../../hooks/index.js'; import { hub } from '@logtide/core'; import { isInternalLoggingEnabled } from '../../utils/internal-logger.js'; +import { createSkewTracker } from './skew.js'; export interface IngestRejection { /** Index of the record in the submitted batch. */ @@ -156,10 +157,13 @@ export class IngestionService { // Convert logs to reservoir LogRecord format // Note: reservoir handles null byte sanitization internally + // Clock skew (#279): observed inside the existing map so a large batch costs + // one extra subtraction per record and one Date.now() per batch. + const skewTracker = createSkewTracker(Date.now()); let records: BeforeIngestContext['records'] = logs.map((log) => { // Extract hostname if not already set in metadata const hostname = log.metadata?.hostname || extractHostname(log); - + const metadata = { ...log.metadata, ...(hostname && { hostname }), @@ -167,8 +171,11 @@ export class IngestionService { const hasMetadata = Object.keys(metadata).length > 0; + const time = typeof log.time === 'string' ? new Date(log.time) : log.time; + skewTracker.observe(time); + return { - time: typeof log.time === 'string' ? new Date(log.time) : log.time, + time, projectId, service: sanitizeForPostgres(log.service), level: log.level as ReservoirLogLevel, @@ -180,6 +187,20 @@ export class IngestionService { }; }); + // Clock skew signal (#279). Fire-and-forget, never rejects: the records above + // are written unchanged. Guarded on organizationId like the other ingestion + // health counters, since anonymous ingestion has no org to attribute to. + const skew = skewTracker.summary(); + if (skew && organizationId) { + metering.record({ + type: 'ingestion.timestamp_skew', + quantity: skew.count, + organizationId, + projectId, + metadata: { maxPastMs: skew.maxPastMs, maxFutureMs: skew.maxFutureMs }, + }); + } + // Lifecycle hook (#216): last interception point before the reservoir // write. Hooks may reject (throw) or mutate/replace `records`. // hasHandlers guard keeps the OSS no-hooks path at zero overhead diff --git a/packages/backend/src/tests/modules/ingestion/ingestion-service.test.ts b/packages/backend/src/tests/modules/ingestion/ingestion-service.test.ts index 771d3495..f3b3db1b 100644 --- a/packages/backend/src/tests/modules/ingestion/ingestion-service.test.ts +++ b/packages/backend/src/tests/modules/ingestion/ingestion-service.test.ts @@ -473,4 +473,68 @@ describe('IngestionService', () => { expect(result.rejected).toEqual([]); }); }); + + describe('clock skew detection (#279)', () => { + it('records a skew counter and still stores the log', async () => { + const recordSpy = vi.spyOn(metering, 'record'); + + const skewed = new Date(Date.now() - 27 * 60 * 60 * 1000).toISOString(); + const result = await ingestionService.ingestLogs( + [{ time: skewed, service: 'ubuntu', level: 'critical', message: 'skewed log' }], + projectId, + ); + + // The invariant that matters: skew never rejects. + expect(result.received).toBe(1); + expect(result.rejected).toEqual([]); + + const skewCall = recordSpy.mock.calls.find( + ([e]) => e.type === 'ingestion.timestamp_skew', + ); + expect(skewCall).toBeDefined(); + expect(skewCall![0].quantity).toBe(1); + expect(skewCall![0].projectId).toBe(projectId); + expect(skewCall![0].metadata).toMatchObject({ maxFutureMs: 0 }); + expect((skewCall![0].metadata as { maxPastMs: number }).maxPastMs).toBeGreaterThan( + 24 * 60 * 60 * 1000, + ); + + recordSpy.mockRestore(); + }); + + it('records no skew counter for a fresh log', async () => { + const recordSpy = vi.spyOn(metering, 'record'); + + await ingestionService.ingestLogs( + [{ time: new Date().toISOString(), service: 'ubuntu', level: 'info', message: 'fresh' }], + projectId, + ); + + expect( + recordSpy.mock.calls.find(([e]) => e.type === 'ingestion.timestamp_skew'), + ).toBeUndefined(); + + recordSpy.mockRestore(); + }); + + it('counts only the skewed records in a mixed batch', async () => { + const recordSpy = vi.spyOn(metering, 'record'); + + await ingestionService.ingestLogs( + [ + { time: new Date(Date.now() - 27 * 60 * 60 * 1000).toISOString(), service: 'a', level: 'info', message: 'old' }, + { time: new Date().toISOString(), service: 'a', level: 'info', message: 'fresh' }, + { time: new Date(Date.now() - 30 * 60 * 60 * 1000).toISOString(), service: 'a', level: 'info', message: 'older' }, + ], + projectId, + ); + + const skewCall = recordSpy.mock.calls.find( + ([e]) => e.type === 'ingestion.timestamp_skew', + ); + expect(skewCall![0].quantity).toBe(2); + + recordSpy.mockRestore(); + }); + }); }); From 73b4e80d3309550a5759f1ea7901be53c1b3befe Mon Sep 17 00:00:00 2001 From: Polliog Date: Fri, 17 Jul 2026 13:49:32 +0200 Subject: [PATCH 03/17] surface skew counter in admin health --- packages/backend/src/modules/admin/service.ts | 2 ++ .../src/tests/modules/admin/routes.test.ts | 22 +++++++++++++++++++ packages/frontend/src/lib/api/admin.ts | 1 + 3 files changed, 25 insertions(+) diff --git a/packages/backend/src/modules/admin/service.ts b/packages/backend/src/modules/admin/service.ts index d1a77899..b2ac283b 100644 --- a/packages/backend/src/modules/admin/service.ts +++ b/packages/backend/src/modules/admin/service.ts @@ -1234,6 +1234,7 @@ export class AdminService { 'ingestion.detection_enqueue_failed', 'ingestion.exception_enqueue_failed', 'ingestion.identifier_failed', + 'ingestion.timestamp_skew', ]) .groupBy('type') .execute(); @@ -1249,6 +1250,7 @@ export class AdminService { detectionEnqueueFailed: byType['ingestion.detection_enqueue_failed'] ?? 0, exceptionEnqueueFailed: byType['ingestion.exception_enqueue_failed'] ?? 0, identifierFailed: byType['ingestion.identifier_failed'] ?? 0, + timestampSkew: byType['ingestion.timestamp_skew'] ?? 0, }, enrichment: enrichmentService.getStatus(), }; diff --git a/packages/backend/src/tests/modules/admin/routes.test.ts b/packages/backend/src/tests/modules/admin/routes.test.ts index 27fd29d2..7ac86949 100644 --- a/packages/backend/src/tests/modules/admin/routes.test.ts +++ b/packages/backend/src/tests/modules/admin/routes.test.ts @@ -844,6 +844,28 @@ describe('Admin Routes', () => { expect(body.counters24h.identifierFailed).toBe(0); expect(body.enrichment).toBeDefined(); }); + + it('reports the timestamp skew counter', async () => { + await db + .insertInto('metering_events') + .values({ + organization_id: testOrg.id, + project_id: testProject.id, + type: 'ingestion.timestamp_skew', + quantity: 7, + metadata: { maxPastMs: 97200000, maxFutureMs: 0 }, + }) + .execute(); + + const response = await app.inject({ + method: 'GET', + url: '/api/v1/admin/stats/ingestion-health', + headers: { authorization: `Bearer ${adminToken}` }, + }); + + expect(response.statusCode).toBe(200); + expect(response.json().counters24h.timestampSkew).toBe(7); + }); }); describe('PATCH /api/v1/admin/users/:id/role', () => { diff --git a/packages/frontend/src/lib/api/admin.ts b/packages/frontend/src/lib/api/admin.ts index 75880776..fadc5a2e 100644 --- a/packages/frontend/src/lib/api/admin.ts +++ b/packages/frontend/src/lib/api/admin.ts @@ -390,6 +390,7 @@ export interface IngestionHealthStats { detectionEnqueueFailed: number; exceptionEnqueueFailed: number; identifierFailed: number; + timestampSkew: number; }; enrichment: { ipReputation: EnrichmentSourceStatus & { totalIps: number }; From 3d0013634f8be337dd365702a6c7fa69747956cc Mon Sep 17 00:00:00 2001 From: Polliog Date: Fri, 17 Jul 2026 13:51:15 +0200 Subject: [PATCH 04/17] add timestamp skew counter to ingestion health card --- packages/frontend/src/routes/dashboard/admin/+page.svelte | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/frontend/src/routes/dashboard/admin/+page.svelte b/packages/frontend/src/routes/dashboard/admin/+page.svelte index 9f574a51..8d1865fc 100644 --- a/packages/frontend/src/routes/dashboard/admin/+page.svelte +++ b/packages/frontend/src/routes/dashboard/admin/+page.svelte @@ -492,6 +492,7 @@
Detection enqueue failures: 0 ? 'font-semibold text-destructive' : ''}>{c.detectionEnqueueFailed}
Exception enqueue failures: 0 ? 'font-semibold text-destructive' : ''}>{c.exceptionEnqueueFailed}
Identifier failures: {c.identifierFailed}
+
Timestamp skew: {c.timestampSkew}
GeoIP: {ingestionHealth.enrichment.geoIp.ready ? 'ready' : 'unavailable'} IP reputation: {ingestionHealth.enrichment.ipReputation.ready ? 'ready' : 'unavailable'} From 0e58a3669d032140efc8a0c1f4dbe389cf4ecd00 Mon Sep 17 00:00:00 2001 From: Polliog Date: Fri, 17 Jul 2026 13:55:29 +0200 Subject: [PATCH 05/17] add per-project ingestion health endpoint --- .../backend/src/modules/projects/routes.ts | 22 +++++ .../backend/src/modules/projects/service.ts | 53 +++++++++++ .../src/tests/modules/projects/routes.test.ts | 92 +++++++++++++++++++ 3 files changed, 167 insertions(+) diff --git a/packages/backend/src/modules/projects/routes.ts b/packages/backend/src/modules/projects/routes.ts index a51672ce..de235571 100644 --- a/packages/backend/src/modules/projects/routes.ts +++ b/packages/backend/src/modules/projects/routes.ts @@ -158,6 +158,28 @@ export async function projectsRoutes(fastify: FastifyInstance) { } }); + // Get project ingestion health (#279: clock skew warning surface) + fastify.get('/:id/ingestion-health', async (request: any, reply) => { + try { + const { id } = projectIdSchema.parse(request.params); + + // Access control + org scoping: getProjectById joins organization_members + // on the caller, so a project outside the caller's orgs reads as 404. + const project = await projectsService.getProjectById(id, request.user.id); + if (!project) { + return reply.status(404).send({ error: 'Project not found' }); + } + + const health = await projectsService.getIngestionHealth(project.organizationId, id); + return reply.send(health); + } catch (error) { + if (error instanceof z.ZodError) { + return reply.status(400).send({ error: 'Invalid project ID' }); + } + throw error; + } + }); + // Restore a soft-deleted project fastify.post('/:id/restore', async (request: any, reply) => { try { diff --git a/packages/backend/src/modules/projects/service.ts b/packages/backend/src/modules/projects/service.ts index d625f001..e652df5f 100644 --- a/packages/backend/src/modules/projects/service.ts +++ b/packages/backend/src/modules/projects/service.ts @@ -38,6 +38,16 @@ export interface UpdateProjectInput { statusPagePassword?: string; } +export interface ProjectSkewHealth { + /** Total skewed records seen in the last 24h. */ + count24h: number; + /** Worst past delta across the window, in ms. */ + maxPastMs: number; + /** Worst future delta across the window, in ms. */ + maxFutureMs: number; + lastSeenAt: string; +} + // All columns returned on every Project fetch const PROJECT_COLUMNS = [ 'id', @@ -232,6 +242,49 @@ export class ProjectsService { return mapProject(project); } + /** + * Per-project ingestion health (#279). Reads the clock skew counter written by + * ingestLogs. Scoped by (organization_id, project_id, time) so it rides + * idx_metering_org_project_time. Volume is one row per skewed batch, so the + * aggregation is done here rather than in SQL. + */ + async getIngestionHealth( + organizationId: string, + projectId: string, + ): Promise<{ skew: ProjectSkewHealth | null }> { + const since = new Date(Date.now() - 24 * 60 * 60 * 1000); + + const rows = await db + .selectFrom('metering_events') + .select(['quantity', 'metadata', 'time']) + .where('organization_id', '=', organizationId) + .where('project_id', '=', projectId) + .where('type', '=', 'ingestion.timestamp_skew') + .where('time', '>=', since) + .execute(); + + if (rows.length === 0) { + return { skew: null }; + } + + let count24h = 0; + let maxPastMs = 0; + let maxFutureMs = 0; + let lastSeenAt = rows[0].time; + + for (const row of rows) { + count24h += Number(row.quantity); + const meta = (row.metadata ?? {}) as { maxPastMs?: number; maxFutureMs?: number }; + if ((meta.maxPastMs ?? 0) > maxPastMs) maxPastMs = meta.maxPastMs ?? 0; + if ((meta.maxFutureMs ?? 0) > maxFutureMs) maxFutureMs = meta.maxFutureMs ?? 0; + if (row.time > lastSeenAt) lastSeenAt = row.time; + } + + return { + skew: { count24h, maxPastMs, maxFutureMs, lastSeenAt: lastSeenAt.toISOString() }, + }; + } + /** * Update a project (only active projects can be updated) */ diff --git a/packages/backend/src/tests/modules/projects/routes.test.ts b/packages/backend/src/tests/modules/projects/routes.test.ts index 66261651..f608addb 100644 --- a/packages/backend/src/tests/modules/projects/routes.test.ts +++ b/packages/backend/src/tests/modules/projects/routes.test.ts @@ -469,4 +469,96 @@ describe('Projects Routes', () => { expect(response.statusCode).toBe(401); }); }); + + describe('GET /:id/ingestion-health', () => { + let otherOrgProjectId: string; + + beforeEach(async () => { + // Second organization the test user is not a member of. + const otherOrg = await createTestOrganization(); + const otherProject = await createTestProject({ organizationId: otherOrg.id }); + otherOrgProjectId = otherProject.id; + }); + + it('returns null skew when nothing was recorded', async () => { + const response = await app.inject({ + method: 'GET', + url: `/api/v1/projects/${testProject.id}/ingestion-health`, + headers: { Authorization: `Bearer ${authToken}` }, + }); + + expect(response.statusCode).toBe(200); + expect(response.json()).toEqual({ skew: null }); + }); + + it('aggregates skew events from the last 24h', async () => { + await db + .insertInto('metering_events') + .values([ + { + organization_id: testOrganization.id, + project_id: testProject.id, + type: 'ingestion.timestamp_skew', + quantity: 4, + metadata: { maxPastMs: 97200000, maxFutureMs: 0 }, + time: new Date(Date.now() - 60 * 60 * 1000), + }, + { + organization_id: testOrganization.id, + project_id: testProject.id, + type: 'ingestion.timestamp_skew', + quantity: 6, + metadata: { maxPastMs: 90000000, maxFutureMs: 600000 }, + time: new Date(Date.now() - 30 * 60 * 1000), + }, + ]) + .execute(); + + const response = await app.inject({ + method: 'GET', + url: `/api/v1/projects/${testProject.id}/ingestion-health`, + headers: { Authorization: `Bearer ${authToken}` }, + }); + + expect(response.statusCode).toBe(200); + const { skew } = response.json(); + expect(skew.count24h).toBe(10); + expect(skew.maxPastMs).toBe(97200000); // worst across events, not last + expect(skew.maxFutureMs).toBe(600000); + expect(new Date(skew.lastSeenAt).getTime()).toBeGreaterThan(Date.now() - 31 * 60 * 1000); + }); + + it('ignores skew events older than 24h', async () => { + await db + .insertInto('metering_events') + .values({ + organization_id: testOrganization.id, + project_id: testProject.id, + type: 'ingestion.timestamp_skew', + quantity: 99, + metadata: { maxPastMs: 97200000, maxFutureMs: 0 }, + time: new Date(Date.now() - 25 * 60 * 60 * 1000), + }) + .execute(); + + const response = await app.inject({ + method: 'GET', + url: `/api/v1/projects/${testProject.id}/ingestion-health`, + headers: { Authorization: `Bearer ${authToken}` }, + }); + + expect(response.json()).toEqual({ skew: null }); + }); + + it('does not leak skew from another organization', async () => { + // otherOrgProjectId belongs to an org the test user is not a member of. + const response = await app.inject({ + method: 'GET', + url: `/api/v1/projects/${otherOrgProjectId}/ingestion-health`, + headers: { Authorization: `Bearer ${authToken}` }, + }); + + expect(response.statusCode).toBe(404); + }); + }); }); From 658b8322d2e05a5de386971642f8d4493c3c339f Mon Sep 17 00:00:00 2001 From: Polliog Date: Fri, 17 Jul 2026 14:01:45 +0200 Subject: [PATCH 06/17] warn on project overview when logs are skewed --- packages/frontend/src/lib/api/projects.ts | 17 +++++++ .../projects/IngestionSkewBanner.svelte | 48 +++++++++++++++++++ .../projects/IngestionSkewBanner.test.ts | 39 +++++++++++++++ .../projects/[id]/overview/+page.svelte | 11 ++++- 4 files changed, 114 insertions(+), 1 deletion(-) create mode 100644 packages/frontend/src/lib/components/projects/IngestionSkewBanner.svelte create mode 100644 packages/frontend/src/lib/components/projects/IngestionSkewBanner.test.ts diff --git a/packages/frontend/src/lib/api/projects.ts b/packages/frontend/src/lib/api/projects.ts index 5758b21d..b1dbe97d 100644 --- a/packages/frontend/src/lib/api/projects.ts +++ b/packages/frontend/src/lib/api/projects.ts @@ -16,6 +16,13 @@ export interface UpdateProjectInput { statusPagePassword?: string; } +export interface ProjectSkewHealth { + count24h: number; + maxPastMs: number; + maxFutureMs: number; + lastSeenAt: string; +} + export class ProjectsAPI { constructor(private getToken: () => string | null) {} @@ -81,6 +88,16 @@ export class ProjectsAPI { return response.json(); } + async getIngestionHealth(id: string): Promise<{ skew: ProjectSkewHealth | null }> { + const response = await this.request(`/projects/${id}/ingestion-health`); + + if (!response.ok) { + throw new Error('Failed to fetch project ingestion health'); + } + + return response.json(); + } + async getProject(id: string): Promise<{ project: Project }> { const response = await this.request(`/projects/${id}`); diff --git a/packages/frontend/src/lib/components/projects/IngestionSkewBanner.svelte b/packages/frontend/src/lib/components/projects/IngestionSkewBanner.svelte new file mode 100644 index 00000000..a9ce04ab --- /dev/null +++ b/packages/frontend/src/lib/components/projects/IngestionSkewBanner.svelte @@ -0,0 +1,48 @@ + + +{#if show && skew} + + + Logs are arriving with an out-of-range timestamp + +

+ In the last 24 hours, {skew.count24h.toLocaleString('en-US')} + {skew.count24h === 1 ? 'log' : 'logs'} arrived with a timestamp up to {direction}. + They are stored and searchable, but threshold alert rules only count logs inside their + time window, so these logs cannot trigger an alert. +

+

+ This usually means the shipper is sending a time field that does not match + the current instant. Omitting the field entirely makes LogTide use the server ingestion + time instead. +

+ + +{/if} diff --git a/packages/frontend/src/lib/components/projects/IngestionSkewBanner.test.ts b/packages/frontend/src/lib/components/projects/IngestionSkewBanner.test.ts new file mode 100644 index 00000000..6d14a051 --- /dev/null +++ b/packages/frontend/src/lib/components/projects/IngestionSkewBanner.test.ts @@ -0,0 +1,39 @@ +import { describe, it, expect } from 'vitest'; +import { render, screen } from '@testing-library/svelte'; +import IngestionSkewBanner from './IngestionSkewBanner.svelte'; + +describe('IngestionSkewBanner', () => { + it('renders nothing when there is no skew', () => { + const { container } = render(IngestionSkewBanner, { props: { skew: null } }); + expect(container.querySelector('[role="alert"]')).toBeNull(); + }); + + it('renders nothing when the count is zero', () => { + const { container } = render(IngestionSkewBanner, { + props: { skew: { count24h: 0, maxPastMs: 0, maxFutureMs: 0, lastSeenAt: '2026-07-17T09:00:00.000Z' } }, + }); + expect(container.querySelector('[role="alert"]')).toBeNull(); + }); + + it('reports past skew with a localized count and an hour figure', () => { + render(IngestionSkewBanner, { + props: { + skew: { count24h: 1234, maxPastMs: 97200000, maxFutureMs: 0, lastSeenAt: '2026-07-17T09:00:00.000Z' }, + }, + }); + + expect(screen.getByRole('alert')).toBeInTheDocument(); + expect(screen.getByText(/1,234/)).toBeInTheDocument(); + expect(screen.getByText(/27 hours in the past/)).toBeInTheDocument(); + }); + + it('reports future skew', () => { + render(IngestionSkewBanner, { + props: { + skew: { count24h: 5, maxPastMs: 0, maxFutureMs: 3600000, lastSeenAt: '2026-07-17T09:00:00.000Z' }, + }, + }); + + expect(screen.getByText(/1 hour ahead of the server clock/)).toBeInTheDocument(); + }); +}); diff --git a/packages/frontend/src/routes/dashboard/projects/[id]/overview/+page.svelte b/packages/frontend/src/routes/dashboard/projects/[id]/overview/+page.svelte index 9777dd28..9322be19 100644 --- a/packages/frontend/src/routes/dashboard/projects/[id]/overview/+page.svelte +++ b/packages/frontend/src/routes/dashboard/projects/[id]/overview/+page.svelte @@ -12,6 +12,9 @@ import TopServicesWidget from '$lib/components/dashboard/TopServicesWidget.svelte'; import RecentErrorsWidget from '$lib/components/dashboard/RecentErrorsWidget.svelte'; import Spinner from '$lib/components/Spinner.svelte'; + import IngestionSkewBanner from '$lib/components/projects/IngestionSkewBanner.svelte'; + import { projectsAPI } from '$lib/api/projects'; + import type { ProjectSkewHealth } from '$lib/api/projects'; import Activity from '@lucide/svelte/icons/activity'; import AlertTriangle from '@lucide/svelte/icons/alert-triangle'; @@ -24,6 +27,7 @@ let activity = $state(null); let topServices = $state([]); let recentErrors = $state([]); + let skew = $state(null); let loading = $state(true); let error = $state(''); let lastLoadedKey = $state(null); @@ -45,17 +49,19 @@ try { const orgId = $currentOrganization.id; - const [statsData, activityData, servicesData, errorsData] = await Promise.all([ + const [statsData, activityData, servicesData, errorsData, healthData] = await Promise.all([ dashboardAPI.getStats(orgId, projectId), dashboardAPI.getActivityOverview(orgId, projectId), dashboardAPI.getTopServices(orgId, projectId), dashboardAPI.getRecentErrors(orgId, projectId), + projectsAPI.getIngestionHealth(projectId).catch(() => ({ skew: null })), ]); stats = statsData; activity = activityData; topServices = servicesData; recentErrors = errorsData; + skew = healthData.skew; lastLoadedKey = `${orgId}-${projectId}`; } catch (e) { @@ -76,6 +82,7 @@ activity = null; topServices = []; recentErrors = []; + skew = null; lastLoadedKey = null; return; } @@ -185,6 +192,8 @@

{:else if stats} + +
Date: Fri, 17 Jul 2026 14:06:51 +0200 Subject: [PATCH 07/17] hoist skew banner outside empty-state branch on project overview --- .../routes/dashboard/projects/[id]/overview/+page.svelte | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/packages/frontend/src/routes/dashboard/projects/[id]/overview/+page.svelte b/packages/frontend/src/routes/dashboard/projects/[id]/overview/+page.svelte index 9322be19..212426ba 100644 --- a/packages/frontend/src/routes/dashboard/projects/[id]/overview/+page.svelte +++ b/packages/frontend/src/routes/dashboard/projects/[id]/overview/+page.svelte @@ -169,6 +169,10 @@
+ {#if !loading && !error} + + {/if} + {#if loading}
@@ -192,8 +196,6 @@

{:else if stats} - -
Date: Fri, 17 Jul 2026 14:10:07 +0200 Subject: [PATCH 08/17] changelog: ingestion clock skew detection --- .env.example | 7 +++++++ CHANGELOG.md | 1 + 2 files changed, 8 insertions(+) diff --git a/.env.example b/.env.example index 6e463390..f92464a9 100644 --- a/.env.example +++ b/.env.example @@ -35,6 +35,13 @@ SMTP_FROM=noreply@logtide.local RATE_LIMIT_MAX=1000 RATE_LIMIT_WINDOW=60000 +# Ingestion +# Ingestion clock skew detection (#279). Logs whose `time` is further than this +# from the server clock are counted and surfaced, never rejected. Set to 0 to +# disable a direction. +INGESTION_SKEW_PAST_MS=86400000 +INGESTION_SKEW_FUTURE_MS=300000 + # Environment NODE_ENV=development diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a19436f..3b9a5983 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- **Ingestion clock skew detection** (#279). Logs whose client-supplied `time` is more than 24 hours in the past or more than 5 minutes ahead of the server clock are counted and surfaced, on the project overview page for tenants and in admin ingestion health for operators. Such logs are stored and searchable but can never be counted by a threshold alert rule, whose largest window is 24 hours, which previously made a misconfigured shipper look like an alerting bug. The logs are never rejected: backfilling historical data stays valid. Thresholds are configurable via `INGESTION_SKEW_PAST_MS` and `INGESTION_SKEW_FUTURE_MS`; set either to 0 to disable that direction. - **Scheduled email digest reports, completed and enabled** (#154): the digest feature whose foundation landed in #209 (and shipped disabled since 1.0.0-beta) is now finished and active. The report grew from log volume only to five sections, each with a quiet empty state: log volume with period-over-period trend, top 5 services by error+critical count with delta vs the previous period (engine-agnostic via `reservoir.topValues`, so TimescaleDB/ClickHouse/MongoDB all work), new error groups first seen in the period (`error_groups.first_seen`), a security summary (windowed detection totals and top triggered Sigma rules read from raw `detection_events` to avoid continuous-aggregate staleness, plus a point-in-time open/investigating incident count), and monitor uptime (aggregated from `monitor_uptime_daily`, section omitted for orgs without monitors). Emails are now real HTML built on the shared `email-templates.ts` component library (inline CSS, escaped user-controlled strings, en-US number formatting) with the full report duplicated in the plaintext fallback, and the base template footer learned an `unsubscribeUrl` variant so digest recipients (who may have no account) get a one-click unsubscribe instead of the channel-settings link. Scheduling was redesigned: instead of one cron per organization registered at worker boot (which could never pick up config changes at runtime, because graphile-worker cron items are fixed once the runner starts), a single static hourly `digest-dispatch` cron sweeps enabled `digest_configs`, matches `delivery_hour`/`delivery_day_of_week` against the current UTC hour, and enqueues `digest-generation` jobs with hour-keyed dedup keys and a `last_sent_at` double-fire guard (migration 053), so config CRUD is live on both queue backends with no restart. New management surface: session-auth CRUD under `/api/v1/digests` (config upsert/delete, recipient add/remove/resubscribe with token rotation on resubscribe, membership for reads and owner/admin for writes, audit-logged as `digest.*`, recipients capped by a new `digests.max_recipients` capability with the canonical 4-case tests), a public `POST /api/v1/digests/unsubscribe` where the 32-byte random token is the credential (idempotent, returns a masked email), an "Email Digests" settings page (schedule card with UTC delivery time and recipient management) and a public `/unsubscribe` page consumed by the email footer link. The worker-side disable comment from the beta is gone: the dispatch cron registers at boot ## [1.1.0] - 2026-07-08 From 7f05b7479740b0ae6ce19269e1cadfb1c554ec51 Mon Sep 17 00:00:00 2001 From: Polliog Date: Fri, 17 Jul 2026 14:34:34 +0200 Subject: [PATCH 09/17] note metering_enabled requirement for skew counters --- .env.example | 3 ++- packages/backend/src/config/index.ts | 4 +++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/.env.example b/.env.example index f92464a9..f545e470 100644 --- a/.env.example +++ b/.env.example @@ -38,7 +38,8 @@ RATE_LIMIT_WINDOW=60000 # Ingestion # Ingestion clock skew detection (#279). Logs whose `time` is further than this # from the server clock are counted and surfaced, never rejected. Set to 0 to -# disable a direction. +# disable a direction. Requires METERING_ENABLED=true (defaults to true): with +# metering off, the counters are never written and stay empty on both surfaces. INGESTION_SKEW_PAST_MS=86400000 INGESTION_SKEW_FUTURE_MS=300000 diff --git a/packages/backend/src/config/index.ts b/packages/backend/src/config/index.ts index c05506a1..d67c441c 100644 --- a/packages/backend/src/config/index.ts +++ b/packages/backend/src/config/index.ts @@ -70,7 +70,9 @@ const configSchema = z.object({ // server clock is stored as sent but can never be seen by a threshold alert // rule, whose largest window is 24h. Detected and counted, never rejected: // backfilling historical logs is legitimate. Set either to 0 to disable that - // direction. + // direction. Counters are written via the metering recorder, so they require + // METERING_ENABLED=true; with metering off, detection still runs but nothing + // is ever recorded and both read surfaces stay empty. INGESTION_SKEW_PAST_MS: z.string().default('86400000').transform(Number), // 24h: the max alert time_window INGESTION_SKEW_FUTURE_MS: z.string().default('300000').transform(Number), // 5m: room for NTP jitter From eb679e6d556282cb504b7b5845d57c3be3dc2119 Mon Sep 17 00:00:00 2001 From: Polliog Date: Fri, 17 Jul 2026 14:34:34 +0200 Subject: [PATCH 10/17] fix stale index name in ingestion health comment --- packages/backend/src/modules/projects/service.ts | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/packages/backend/src/modules/projects/service.ts b/packages/backend/src/modules/projects/service.ts index e652df5f..5c135c1e 100644 --- a/packages/backend/src/modules/projects/service.ts +++ b/packages/backend/src/modules/projects/service.ts @@ -244,9 +244,10 @@ export class ProjectsService { /** * Per-project ingestion health (#279). Reads the clock skew counter written by - * ingestLogs. Scoped by (organization_id, project_id, time) so it rides - * idx_metering_org_project_time. Volume is one row per skewed batch, so the - * aggregation is done here rather than in SQL. + * ingestLogs. Filters on (organization_id, type, time) plus project_id, so the + * planner favors idx_metering_org_type_time: type is highly selective while + * (org, project, 24h) alone matches every metering row for the project. Volume + * is one row per skewed batch, so the aggregation is done here rather than in SQL. */ async getIngestionHealth( organizationId: string, From 5962194d97f14df2a46adc9bf3aa282a8e522cbc Mon Sep 17 00:00:00 2001 From: Polliog Date: Fri, 17 Jul 2026 14:34:37 +0200 Subject: [PATCH 11/17] add skew writer/reader contract and config default tests --- .../backend/src/tests/config/config.test.ts | 14 +++++++ .../src/tests/modules/ingestion/skew.test.ts | 2 +- .../src/tests/modules/projects/routes.test.ts | 37 +++++++++++++++++++ 3 files changed, 52 insertions(+), 1 deletion(-) diff --git a/packages/backend/src/tests/config/config.test.ts b/packages/backend/src/tests/config/config.test.ts index 0857e3b3..9f536ed1 100644 --- a/packages/backend/src/tests/config/config.test.ts +++ b/packages/backend/src/tests/config/config.test.ts @@ -64,6 +64,20 @@ describe('Config Validation', () => { } }); + it('should default ingestion clock skew thresholds to 24h past / 5m future', () => { + const result = configSchema.safeParse({ + DATABASE_URL: 'postgresql://user:pass@localhost:5432/db', + API_KEY_SECRET: 'a'.repeat(32), + }); + expect(result.success).toBe(true); + if (result.success) { + // 24h: the largest alert time_window, the whole reason a log past it + // can never be evaluated by a threshold rule. + expect(result.data.INGESTION_SKEW_PAST_MS).toBe(86400000); + expect(result.data.INGESTION_SKEW_FUTURE_MS).toBe(300000); + } + }); + it('should accept custom PORT and HOST', () => { const result = configSchema.safeParse({ DATABASE_URL: 'postgresql://user:pass@localhost:5432/db', diff --git a/packages/backend/src/tests/modules/ingestion/skew.test.ts b/packages/backend/src/tests/modules/ingestion/skew.test.ts index 34d20fd4..6afaa019 100644 --- a/packages/backend/src/tests/modules/ingestion/skew.test.ts +++ b/packages/backend/src/tests/modules/ingestion/skew.test.ts @@ -1,4 +1,4 @@ -import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import { describe, it, expect, vi, beforeEach } from 'vitest'; // The tracker reads thresholds from config at construction, so config is mocked // per test rather than importing the real .env.test values. diff --git a/packages/backend/src/tests/modules/projects/routes.test.ts b/packages/backend/src/tests/modules/projects/routes.test.ts index f608addb..5a634399 100644 --- a/packages/backend/src/tests/modules/projects/routes.test.ts +++ b/packages/backend/src/tests/modules/projects/routes.test.ts @@ -2,6 +2,8 @@ import { describe, it, expect, beforeEach, afterAll, beforeAll } from 'vitest'; import Fastify, { FastifyInstance } from 'fastify'; import { db } from '../../../database/index.js'; import { projectsRoutes } from '../../../modules/projects/routes.js'; +import { ingestionService } from '../../../modules/ingestion/service.js'; +import { meteringRecorder } from '../../../modules/metering/index.js'; import { createTestContext, createTestUser, createTestProject, createTestOrganization } from '../../helpers/factories.js'; import crypto from 'crypto'; @@ -560,5 +562,40 @@ describe('Projects Routes', () => { expect(response.statusCode).toBe(404); }); + + it('reads the exact keys a real ingest writes (writer/reader contract)', async () => { + // #279 repro: a log ~27h in the past, past the 24h skew threshold. + const skewedTime = new Date(Date.now() - 27 * 60 * 60 * 1000).toISOString(); + const ingestResult = await ingestionService.ingestLogs( + [{ time: skewedTime, service: 'ubuntu', level: 'critical', message: 'skewed log' }], + testProject.id, + ); + expect(ingestResult.received).toBe(1); + expect(ingestResult.rejected).toEqual([]); + + // Force the buffered metering event through to the DB rather than + // waiting for the recorder's interval flush. + await meteringRecorder.flush(); + + const response = await app.inject({ + method: 'GET', + url: `/api/v1/projects/${testProject.id}/ingestion-health`, + headers: { Authorization: `Bearer ${authToken}` }, + }); + + expect(response.statusCode).toBe(200); + const { skew } = response.json(); + expect(skew.count24h).toBe(1); + expect(skew.maxPastMs).toBeGreaterThan(24 * 60 * 60 * 1000); + + // Required invariant: a skewed log is stored, never rejected. + const storedLog = await db + .selectFrom('logs') + .selectAll() + .where('project_id', '=', testProject.id) + .where('message', '=', 'skewed log') + .executeTakeFirst(); + expect(storedLog).toBeDefined(); + }); }); }); From 14bc3b3316743cb06811bf594ea6359f640235b8 Mon Sep 17 00:00:00 2001 From: Polliog Date: Fri, 17 Jul 2026 15:01:25 +0200 Subject: [PATCH 12/17] hardcode ingestion clock skew thresholds, drop env knobs --- .env.example | 8 ----- packages/backend/src/config/index.ts | 10 ------- .../backend/src/modules/ingestion/skew.ts | 23 ++++++++++----- .../backend/src/tests/config/config.test.ts | 14 --------- .../src/tests/modules/ingestion/skew.test.ts | 29 ++----------------- 5 files changed, 17 insertions(+), 67 deletions(-) diff --git a/.env.example b/.env.example index f545e470..6e463390 100644 --- a/.env.example +++ b/.env.example @@ -35,14 +35,6 @@ SMTP_FROM=noreply@logtide.local RATE_LIMIT_MAX=1000 RATE_LIMIT_WINDOW=60000 -# Ingestion -# Ingestion clock skew detection (#279). Logs whose `time` is further than this -# from the server clock are counted and surfaced, never rejected. Set to 0 to -# disable a direction. Requires METERING_ENABLED=true (defaults to true): with -# metering off, the counters are never written and stay empty on both surfaces. -INGESTION_SKEW_PAST_MS=86400000 -INGESTION_SKEW_FUTURE_MS=300000 - # Environment NODE_ENV=development diff --git a/packages/backend/src/config/index.ts b/packages/backend/src/config/index.ts index d67c441c..0b135b4b 100644 --- a/packages/backend/src/config/index.ts +++ b/packages/backend/src/config/index.ts @@ -66,16 +66,6 @@ const configSchema = z.object({ METERING_FLUSH_INTERVAL_MS: z.string().default('5000').transform(Number), METERING_FLUSH_MAX_BUFFER: z.string().default('500').transform(Number), - // Ingestion clock skew detection (#279). A client-supplied `time` far from the - // server clock is stored as sent but can never be seen by a threshold alert - // rule, whose largest window is 24h. Detected and counted, never rejected: - // backfilling historical logs is legitimate. Set either to 0 to disable that - // direction. Counters are written via the metering recorder, so they require - // METERING_ENABLED=true; with metering off, detection still runs but nothing - // is ever recorded and both read surfaces stay empty. - INGESTION_SKEW_PAST_MS: z.string().default('86400000').transform(Number), // 24h: the max alert time_window - INGESTION_SKEW_FUTURE_MS: z.string().default('300000').transform(Number), // 5m: room for NTP jitter - // Capability usage-quota evaluator (#214). Periodic job that flags over-quota orgs. QUOTA_EVALUATOR_ENABLED: z.string().default('true').transform((val) => val === 'true'), QUOTA_EVALUATOR_INTERVAL_MS: z.string().default('60000').transform(Number), diff --git a/packages/backend/src/modules/ingestion/skew.ts b/packages/backend/src/modules/ingestion/skew.ts index baf7457a..5a7922b1 100644 --- a/packages/backend/src/modules/ingestion/skew.ts +++ b/packages/backend/src/modules/ingestion/skew.ts @@ -4,12 +4,22 @@ * A log whose client-supplied `time` sits far from the server clock is stored * and is visible in the UI, but threshold alert rules count logs in * [now - time_window, now] and so can never see it. The largest configurable - * window is 24h (alerts/routes.ts), which is where the past default comes from: - * past that point, no threshold rule in the product can match the log. + * window is 24h (alerts/routes.ts), which is where the past threshold comes + * from: past that point, no threshold rule in the product can match the log. * * Observation only. Skewed records are written unchanged. + * + * Always on: these thresholds are load-bearing product invariants, not + * tunable knobs, so they are hardcoded rather than read from config. There is + * no "disable" state. */ -import { config } from '../../config/index.js'; + +/** 24h: the largest alert `time_window` (alerts/routes.ts). Past this point, + * no threshold rule in the product can ever count the log. Derived, not guessed. */ +const PAST_THRESHOLD_MS = 86400000; + +/** 5m: room for NTP jitter. */ +const FUTURE_THRESHOLD_MS = 300000; export interface SkewSummary { /** Number of skewed records in the batch. */ @@ -30,9 +40,6 @@ export interface SkewTracker { * large batch is measured against one instant, not a drifting one. */ export function createSkewTracker(now: number): SkewTracker { - const pastMs = config.INGESTION_SKEW_PAST_MS; - const futureMs = config.INGESTION_SKEW_FUTURE_MS; - let count = 0; let maxPastMs = 0; let maxFutureMs = 0; @@ -46,13 +53,13 @@ export function createSkewTracker(now: number): SkewTracker { // problem and counting it here would make this signal dishonest. const delta = now - time.getTime(); - if (pastMs > 0 && delta > pastMs) { + if (delta > PAST_THRESHOLD_MS) { count++; if (delta > maxPastMs) maxPastMs = delta; return; } - if (futureMs > 0 && delta < -futureMs) { + if (delta < -FUTURE_THRESHOLD_MS) { count++; const ahead = -delta; if (ahead > maxFutureMs) maxFutureMs = ahead; diff --git a/packages/backend/src/tests/config/config.test.ts b/packages/backend/src/tests/config/config.test.ts index 9f536ed1..0857e3b3 100644 --- a/packages/backend/src/tests/config/config.test.ts +++ b/packages/backend/src/tests/config/config.test.ts @@ -64,20 +64,6 @@ describe('Config Validation', () => { } }); - it('should default ingestion clock skew thresholds to 24h past / 5m future', () => { - const result = configSchema.safeParse({ - DATABASE_URL: 'postgresql://user:pass@localhost:5432/db', - API_KEY_SECRET: 'a'.repeat(32), - }); - expect(result.success).toBe(true); - if (result.success) { - // 24h: the largest alert time_window, the whole reason a log past it - // can never be evaluated by a threshold rule. - expect(result.data.INGESTION_SKEW_PAST_MS).toBe(86400000); - expect(result.data.INGESTION_SKEW_FUTURE_MS).toBe(300000); - } - }); - it('should accept custom PORT and HOST', () => { const result = configSchema.safeParse({ DATABASE_URL: 'postgresql://user:pass@localhost:5432/db', diff --git a/packages/backend/src/tests/modules/ingestion/skew.test.ts b/packages/backend/src/tests/modules/ingestion/skew.test.ts index 6afaa019..0b3e31e9 100644 --- a/packages/backend/src/tests/modules/ingestion/skew.test.ts +++ b/packages/backend/src/tests/modules/ingestion/skew.test.ts @@ -1,21 +1,10 @@ -import { describe, it, expect, vi, beforeEach } from 'vitest'; - -// The tracker reads thresholds from config at construction, so config is mocked -// per test rather than importing the real .env.test values. -const mockConfig = { INGESTION_SKEW_PAST_MS: 86400000, INGESTION_SKEW_FUTURE_MS: 300000 }; -vi.mock('../../../config/index.js', () => ({ config: mockConfig })); - -const { createSkewTracker } = await import('../../../modules/ingestion/skew.js'); +import { describe, it, expect } from 'vitest'; +import { createSkewTracker } from '../../../modules/ingestion/skew.js'; const NOW = new Date('2026-07-17T12:00:00.000Z').getTime(); const ago = (ms: number) => new Date(NOW - ms); describe('createSkewTracker', () => { - beforeEach(() => { - mockConfig.INGESTION_SKEW_PAST_MS = 86400000; - mockConfig.INGESTION_SKEW_FUTURE_MS = 300000; - }); - it('returns null when nothing is skewed', () => { const t = createSkewTracker(NOW); t.observe(ago(0)); @@ -56,18 +45,4 @@ describe('createSkewTracker', () => { t.observe(undefined); expect(t.summary()).toBeNull(); }); - - it('disables past detection when the threshold is 0', () => { - mockConfig.INGESTION_SKEW_PAST_MS = 0; - const t = createSkewTracker(NOW); - t.observe(ago(97_200_000)); - expect(t.summary()).toBeNull(); - }); - - it('disables future detection when the threshold is 0', () => { - mockConfig.INGESTION_SKEW_FUTURE_MS = 0; - const t = createSkewTracker(NOW); - t.observe(ago(-600_000)); - expect(t.summary()).toBeNull(); - }); }); From a877ef2a9fa14a81f91e9b217d802f920b6cf04b Mon Sep 17 00:00:00 2001 From: Polliog Date: Fri, 17 Jul 2026 15:03:49 +0200 Subject: [PATCH 13/17] exempt ingestion health counters from metering toggle --- .../backend/src/modules/metering/recorder.ts | 10 +++++- .../tests/modules/metering/recorder.test.ts | 33 ++++++++++++++++++- 2 files changed, 41 insertions(+), 2 deletions(-) diff --git a/packages/backend/src/modules/metering/recorder.ts b/packages/backend/src/modules/metering/recorder.ts index 76b8feec..84485b04 100644 --- a/packages/backend/src/modules/metering/recorder.ts +++ b/packages/backend/src/modules/metering/recorder.ts @@ -29,7 +29,15 @@ export class MeteringRecorder { } record(event: MeteringEvent): void { - if (!config.METERING_ENABLED) return; + // METERING_ENABLED gates usage/resource metering (billing-shaped). The + // `ingestion.*` types are operational health counters, not usage: they are + // already excluded from tenant usage breakdowns by the `NOT LIKE + // 'ingestion.%'` filter in breakdown.ts. A billing toggle must not be able + // to silently blind operators to ingestion health (including the + // pii_rejected safety counter), so those types bypass this gate. The + // hardCap drop-guard below still applies: it is a memory safety valve, + // not a feature toggle. + if (!config.METERING_ENABLED && !event.type.startsWith('ingestion.')) return; if (this.buffer.length >= this.hardCap) { this.dropped++; return; diff --git a/packages/backend/src/tests/modules/metering/recorder.test.ts b/packages/backend/src/tests/modules/metering/recorder.test.ts index e1055632..8b08388a 100644 --- a/packages/backend/src/tests/modules/metering/recorder.test.ts +++ b/packages/backend/src/tests/modules/metering/recorder.test.ts @@ -1,5 +1,6 @@ -import { describe, it, expect, beforeEach } from 'vitest'; +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; import { db } from '../../../database/index.js'; +import { config } from '../../../config/index.js'; import { MeteringRecorder } from '../../../modules/metering/recorder.js'; import { createTestContext } from '../../helpers/factories.js'; @@ -62,4 +63,34 @@ describe('MeteringRecorder', () => { expect(rows).toHaveLength(1); expect(Number(rows[0].quantity)).toBe(99); }); + + describe('METERING_ENABLED gate', () => { + const original = config.METERING_ENABLED; + + afterEach(() => { + config.METERING_ENABLED = original; + }); + + it('still buffers an ingestion.* health counter when metering is disabled', () => { + config.METERING_ENABLED = false; + const rec = new MeteringRecorder({ maxBuffer: 1000 }); + rec.record({ type: 'ingestion.timestamp_skew', quantity: 1, organizationId: orgId, projectId }); + expect(rec.bufferSize).toBe(1); + }); + + it('does not buffer a usage event when metering is disabled', () => { + config.METERING_ENABLED = false; + const rec = new MeteringRecorder({ maxBuffer: 1000 }); + rec.record({ type: 'logs.ingested.events', quantity: 1, organizationId: orgId, projectId }); + expect(rec.bufferSize).toBe(0); + }); + + it('buffers both ingestion.* and usage events when metering is enabled', () => { + config.METERING_ENABLED = true; + const rec = new MeteringRecorder({ maxBuffer: 1000 }); + rec.record({ type: 'ingestion.timestamp_skew', quantity: 1, organizationId: orgId, projectId }); + rec.record({ type: 'logs.ingested.events', quantity: 1, organizationId: orgId, projectId }); + expect(rec.bufferSize).toBe(2); + }); + }); }); From 4d0c095312c593e31e7e2c458a3b3746fa6a5256 Mon Sep 17 00:00:00 2001 From: Polliog Date: Fri, 17 Jul 2026 15:07:42 +0200 Subject: [PATCH 14/17] push skew aggregation into sql, fix index comment aggregate maxPastMs/maxFutureMs/count/lastSeenAt in one query instead of a JS loop over every row. cast metadata jsonb fields to float8 before MAX to avoid a lexicographic max bug, with a regression test for it. --- .../backend/src/modules/projects/service.ts | 67 +++++++++++-------- .../src/tests/modules/projects/routes.test.ts | 37 ++++++++++ 2 files changed, 77 insertions(+), 27 deletions(-) diff --git a/packages/backend/src/modules/projects/service.ts b/packages/backend/src/modules/projects/service.ts index 5c135c1e..d1a4a4c4 100644 --- a/packages/backend/src/modules/projects/service.ts +++ b/packages/backend/src/modules/projects/service.ts @@ -1,3 +1,4 @@ +import { sql } from 'kysely'; import { db } from '../../database/connection.js'; import type { Project, StatusPageVisibility } from '@logtide/shared'; import bcrypt from 'bcrypt'; @@ -244,10 +245,12 @@ export class ProjectsService { /** * Per-project ingestion health (#279). Reads the clock skew counter written by - * ingestLogs. Filters on (organization_id, type, time) plus project_id, so the - * planner favors idx_metering_org_type_time: type is highly selective while - * (org, project, 24h) alone matches every metering row for the project. Volume - * is one row per skewed batch, so the aggregation is done here rather than in SQL. + * ingestLogs, aggregated in SQL so the query returns exactly one row rather than + * pulling every skew row into JS. Filters on (organization_id, type, time) plus + * project_id, so the planner favors idx_metering_org_type_time: type is highly + * selective while (org, project, 24h) alone matches every metering row for the + * project. metadata->>'x' yields text, so each field is cast to float8 before + * MAX to avoid a lexicographic (string) max. */ async getIngestionHealth( organizationId: string, @@ -255,34 +258,44 @@ export class ProjectsService { ): Promise<{ skew: ProjectSkewHealth | null }> { const since = new Date(Date.now() - 24 * 60 * 60 * 1000); - const rows = await db - .selectFrom('metering_events') - .select(['quantity', 'metadata', 'time']) - .where('organization_id', '=', organizationId) - .where('project_id', '=', projectId) - .where('type', '=', 'ingestion.timestamp_skew') - .where('time', '>=', since) - .execute(); - - if (rows.length === 0) { + const query = sql<{ + count24h: number | string | null; + max_past_ms: number | string | null; + max_future_ms: number | string | null; + last_seen_at: Date | string | null; + }>` + SELECT + COALESCE(SUM(quantity), 0)::float8 AS count24h, + COALESCE(MAX((metadata->>'maxPastMs')::float8), 0) AS max_past_ms, + COALESCE(MAX((metadata->>'maxFutureMs')::float8), 0) AS max_future_ms, + MAX(time) AS last_seen_at + FROM metering_events + WHERE organization_id = ${organizationId} + AND project_id = ${projectId} + AND type = 'ingestion.timestamp_skew' + AND time >= ${since} + `; + const result = await query.execute(db); + const row = result.rows[0]; + + // An aggregate over zero matching rows still returns one row, but every + // aggregate column is NULL except the COALESCE'd ones; last_seen_at (MAX(time)) + // has no COALESCE, so it is the reliable "no rows" signal. + if (!row || row.last_seen_at === null) { return { skew: null }; } - let count24h = 0; - let maxPastMs = 0; - let maxFutureMs = 0; - let lastSeenAt = rows[0].time; - - for (const row of rows) { - count24h += Number(row.quantity); - const meta = (row.metadata ?? {}) as { maxPastMs?: number; maxFutureMs?: number }; - if ((meta.maxPastMs ?? 0) > maxPastMs) maxPastMs = meta.maxPastMs ?? 0; - if ((meta.maxFutureMs ?? 0) > maxFutureMs) maxFutureMs = meta.maxFutureMs ?? 0; - if (row.time > lastSeenAt) lastSeenAt = row.time; - } + const toNumber = (value: number | string | null): number => + typeof value === 'number' ? value : parseFloat(value ?? '0'); + const lastSeenAt = row.last_seen_at instanceof Date ? row.last_seen_at : new Date(row.last_seen_at); return { - skew: { count24h, maxPastMs, maxFutureMs, lastSeenAt: lastSeenAt.toISOString() }, + skew: { + count24h: toNumber(row.count24h), + maxPastMs: toNumber(row.max_past_ms), + maxFutureMs: toNumber(row.max_future_ms), + lastSeenAt: lastSeenAt.toISOString(), + }, }; } diff --git a/packages/backend/src/tests/modules/projects/routes.test.ts b/packages/backend/src/tests/modules/projects/routes.test.ts index 5a634399..d4100d62 100644 --- a/packages/backend/src/tests/modules/projects/routes.test.ts +++ b/packages/backend/src/tests/modules/projects/routes.test.ts @@ -597,5 +597,42 @@ describe('Projects Routes', () => { .executeTakeFirst(); expect(storedLog).toBeDefined(); }); + + it('takes the numeric max of maxPastMs, not the lexicographic (text) max', async () => { + // "9500000" > "10000000" as text (compares '9' vs '1' first), but + // 10000000 is the larger number. A MAX() over uncasted metadata->>'x' + // text would wrongly report 9500000 here. + await db + .insertInto('metering_events') + .values([ + { + organization_id: testOrganization.id, + project_id: testProject.id, + type: 'ingestion.timestamp_skew', + quantity: 1, + metadata: { maxPastMs: 9500000, maxFutureMs: 0 }, + time: new Date(Date.now() - 60 * 60 * 1000), + }, + { + organization_id: testOrganization.id, + project_id: testProject.id, + type: 'ingestion.timestamp_skew', + quantity: 1, + metadata: { maxPastMs: 10000000, maxFutureMs: 0 }, + time: new Date(Date.now() - 30 * 60 * 1000), + }, + ]) + .execute(); + + const response = await app.inject({ + method: 'GET', + url: `/api/v1/projects/${testProject.id}/ingestion-health`, + headers: { Authorization: `Bearer ${authToken}` }, + }); + + expect(response.statusCode).toBe(200); + const { skew } = response.json(); + expect(skew.maxPastMs).toBe(10000000); + }); }); }); From 082e9efcd23f1954f9e94caeb9c69daab7508884 Mon Sep 17 00:00:00 2001 From: Polliog Date: Fri, 17 Jul 2026 15:10:34 +0200 Subject: [PATCH 15/17] render lastSeenAt as relative time in skew banner --- .../projects/IngestionSkewBanner.svelte | 21 +++++++++ .../projects/IngestionSkewBanner.test.ts | 44 ++++++++++++++++++- 2 files changed, 64 insertions(+), 1 deletion(-) diff --git a/packages/frontend/src/lib/components/projects/IngestionSkewBanner.svelte b/packages/frontend/src/lib/components/projects/IngestionSkewBanner.svelte index a9ce04ab..eee20146 100644 --- a/packages/frontend/src/lib/components/projects/IngestionSkewBanner.svelte +++ b/packages/frontend/src/lib/components/projects/IngestionSkewBanner.svelte @@ -25,6 +25,24 @@ ? `${humanize(skew.maxPastMs)} in the past` : `${humanize(skew.maxFutureMs)} ahead of the server clock`, ); + + // How long ago the most recent skewed log arrived. Relative phrasing avoids a + // locale-dependent absolute timestamp and tells the user whether this is still + // happening or they already fixed it and are looking at the tail of the window. + function timeAgo(iso: string): string { + const diffMs = Date.now() - new Date(iso).getTime(); + const minutes = Math.floor(diffMs / 60000); + if (minutes < 1) { + return 'just now'; + } + if (minutes < 60) { + return `${minutes.toLocaleString('en-US')} ${minutes === 1 ? 'minute' : 'minutes'} ago`; + } + const hours = Math.floor(diffMs / 3600000); + return `${hours.toLocaleString('en-US')} ${hours === 1 ? 'hour' : 'hours'} ago`; + } + + const lastSeen = $derived(!skew ? '' : timeAgo(skew.lastSeenAt)); {#if show && skew} @@ -43,6 +61,9 @@ the current instant. Omitting the field entirely makes LogTide use the server ingestion time instead.

+

+ Most recent at {lastSeen}. +

{/if} diff --git a/packages/frontend/src/lib/components/projects/IngestionSkewBanner.test.ts b/packages/frontend/src/lib/components/projects/IngestionSkewBanner.test.ts index 6d14a051..bfa43e14 100644 --- a/packages/frontend/src/lib/components/projects/IngestionSkewBanner.test.ts +++ b/packages/frontend/src/lib/components/projects/IngestionSkewBanner.test.ts @@ -1,8 +1,17 @@ -import { describe, it, expect } from 'vitest'; +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; import { render, screen } from '@testing-library/svelte'; import IngestionSkewBanner from './IngestionSkewBanner.svelte'; describe('IngestionSkewBanner', () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-07-17T09:00:00.000Z')); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + it('renders nothing when there is no skew', () => { const { container } = render(IngestionSkewBanner, { props: { skew: null } }); expect(container.querySelector('[role="alert"]')).toBeNull(); @@ -36,4 +45,37 @@ describe('IngestionSkewBanner', () => { expect(screen.getByText(/1 hour ahead of the server clock/)).toBeInTheDocument(); }); + + it('renders the most recent skew time as a relative phrase (plural minutes)', () => { + render(IngestionSkewBanner, { + props: { + // 5 minutes before the fixed system time. + skew: { count24h: 5, maxPastMs: 3600000, maxFutureMs: 0, lastSeenAt: '2026-07-17T08:55:00.000Z' }, + }, + }); + + expect(screen.getByText(/Most recent at 5 minutes ago\./)).toBeInTheDocument(); + }); + + it('renders the most recent skew time with singular minute', () => { + render(IngestionSkewBanner, { + props: { + // 1 minute before the fixed system time. + skew: { count24h: 5, maxPastMs: 3600000, maxFutureMs: 0, lastSeenAt: '2026-07-17T08:59:00.000Z' }, + }, + }); + + expect(screen.getByText(/Most recent at 1 minute ago\./)).toBeInTheDocument(); + }); + + it('renders the most recent skew time in hours', () => { + render(IngestionSkewBanner, { + props: { + // 2 hours before the fixed system time. + skew: { count24h: 5, maxPastMs: 3600000, maxFutureMs: 0, lastSeenAt: '2026-07-17T07:00:00.000Z' }, + }, + }); + + expect(screen.getByText(/Most recent at 2 hours ago\./)).toBeInTheDocument(); + }); }); From 956a52882f45688778e17e957ad4202e5aa05f1c Mon Sep 17 00:00:00 2001 From: Polliog Date: Fri, 17 Jul 2026 15:11:31 +0200 Subject: [PATCH 16/17] pin lastSeenAt assertion to newest fixture row --- .../src/tests/modules/projects/routes.test.ts | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/packages/backend/src/tests/modules/projects/routes.test.ts b/packages/backend/src/tests/modules/projects/routes.test.ts index d4100d62..1a61370f 100644 --- a/packages/backend/src/tests/modules/projects/routes.test.ts +++ b/packages/backend/src/tests/modules/projects/routes.test.ts @@ -494,6 +494,9 @@ describe('Projects Routes', () => { }); it('aggregates skew events from the last 24h', async () => { + const olderEventTime = new Date(Date.now() - 60 * 60 * 1000); + const newestEventTime = new Date(Date.now() - 30 * 60 * 1000); + await db .insertInto('metering_events') .values([ @@ -503,7 +506,7 @@ describe('Projects Routes', () => { type: 'ingestion.timestamp_skew', quantity: 4, metadata: { maxPastMs: 97200000, maxFutureMs: 0 }, - time: new Date(Date.now() - 60 * 60 * 1000), + time: olderEventTime, }, { organization_id: testOrganization.id, @@ -511,7 +514,7 @@ describe('Projects Routes', () => { type: 'ingestion.timestamp_skew', quantity: 6, metadata: { maxPastMs: 90000000, maxFutureMs: 600000 }, - time: new Date(Date.now() - 30 * 60 * 1000), + time: newestEventTime, }, ]) .execute(); @@ -527,7 +530,11 @@ describe('Projects Routes', () => { expect(skew.count24h).toBe(10); expect(skew.maxPastMs).toBe(97200000); // worst across events, not last expect(skew.maxFutureMs).toBe(600000); - expect(new Date(skew.lastSeenAt).getTime()).toBeGreaterThan(Date.now() - 31 * 60 * 1000); + // Pins lastSeenAt to the newest fixture row's time (not "now"), with a + // small tolerance for DB round-tripping. + expect(Math.abs(new Date(skew.lastSeenAt).getTime() - newestEventTime.getTime())).toBeLessThan( + 5000, + ); }); it('ignores skew events older than 24h', async () => { From 84b4c999d9036e27b181d54d9cc7e704abeb9d2a Mon Sep 17 00:00:00 2001 From: Polliog Date: Fri, 17 Jul 2026 15:13:12 +0200 Subject: [PATCH 17/17] changelog: skew always on, metering gate fix --- CHANGELOG.md | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3b9a5983..0df15280 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,8 +8,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed +- **Ingestion health counters no longer depend on the metering toggle** (#279). `METERING_ENABLED=false` silently suppressed every `ingestion.*` counter, including `ingestion.pii_rejected`, the safety counter for records dropped because PII masking failed. That toggle governs usage and resource metering, while the `ingestion.*` counters are operational health signals already excluded from tenant usage breakdowns by a prefix filter, so they now bypass it. The buffer hard cap still applies to them: that is a memory valve, not a feature switch. + ### Added -- **Ingestion clock skew detection** (#279). Logs whose client-supplied `time` is more than 24 hours in the past or more than 5 minutes ahead of the server clock are counted and surfaced, on the project overview page for tenants and in admin ingestion health for operators. Such logs are stored and searchable but can never be counted by a threshold alert rule, whose largest window is 24 hours, which previously made a misconfigured shipper look like an alerting bug. The logs are never rejected: backfilling historical data stays valid. Thresholds are configurable via `INGESTION_SKEW_PAST_MS` and `INGESTION_SKEW_FUTURE_MS`; set either to 0 to disable that direction. +- **Ingestion clock skew detection** (#279). Logs whose client-supplied `time` is more than 24 hours in the past or more than 5 minutes ahead of the server clock are counted and surfaced, on the project overview page for tenants and in admin ingestion health for operators. Such logs are stored and searchable but can never be counted by a threshold alert rule, whose largest window is 24 hours, which previously made a misconfigured shipper look like an alerting bug: ingestion worked, rules were enabled, channels tested fine, and no alert ever fired. The 24 hour threshold is derived rather than chosen, being the maximum configurable alert `time_window`. The logs are never rejected, so backfilling historical data stays valid, and detection is always on with no configuration to get wrong. Detection runs inside the existing record-building pass in `ingestLogs`, the single choke point for every log path (batch, Fluent Bit, OTLP, receivers), at a cost of one subtraction per record. - **Scheduled email digest reports, completed and enabled** (#154): the digest feature whose foundation landed in #209 (and shipped disabled since 1.0.0-beta) is now finished and active. The report grew from log volume only to five sections, each with a quiet empty state: log volume with period-over-period trend, top 5 services by error+critical count with delta vs the previous period (engine-agnostic via `reservoir.topValues`, so TimescaleDB/ClickHouse/MongoDB all work), new error groups first seen in the period (`error_groups.first_seen`), a security summary (windowed detection totals and top triggered Sigma rules read from raw `detection_events` to avoid continuous-aggregate staleness, plus a point-in-time open/investigating incident count), and monitor uptime (aggregated from `monitor_uptime_daily`, section omitted for orgs without monitors). Emails are now real HTML built on the shared `email-templates.ts` component library (inline CSS, escaped user-controlled strings, en-US number formatting) with the full report duplicated in the plaintext fallback, and the base template footer learned an `unsubscribeUrl` variant so digest recipients (who may have no account) get a one-click unsubscribe instead of the channel-settings link. Scheduling was redesigned: instead of one cron per organization registered at worker boot (which could never pick up config changes at runtime, because graphile-worker cron items are fixed once the runner starts), a single static hourly `digest-dispatch` cron sweeps enabled `digest_configs`, matches `delivery_hour`/`delivery_day_of_week` against the current UTC hour, and enqueues `digest-generation` jobs with hour-keyed dedup keys and a `last_sent_at` double-fire guard (migration 053), so config CRUD is live on both queue backends with no restart. New management surface: session-auth CRUD under `/api/v1/digests` (config upsert/delete, recipient add/remove/resubscribe with token rotation on resubscribe, membership for reads and owner/admin for writes, audit-logged as `digest.*`, recipients capped by a new `digests.max_recipients` capability with the canonical 4-case tests), a public `POST /api/v1/digests/unsubscribe` where the 32-byte random token is the credential (idempotent, returns a masked email), an "Email Digests" settings page (schedule card with UTC delivery time and recipient management) and a public `/unsubscribe` page consumed by the email footer link. The worker-side disable comment from the beta is gone: the dispatch cron registers at boot ## [1.1.0] - 2026-07-08