diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a19436f..0df15280 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +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: 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 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/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/modules/ingestion/skew.ts b/packages/backend/src/modules/ingestion/skew.ts new file mode 100644 index 00000000..5a7922b1 --- /dev/null +++ b/packages/backend/src/modules/ingestion/skew.ts @@ -0,0 +1,73 @@ +/** + * 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 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. + */ + +/** 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. */ + 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 { + 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 (delta > PAST_THRESHOLD_MS) { + count++; + if (delta > maxPastMs) maxPastMs = delta; + return; + } + + if (delta < -FUTURE_THRESHOLD_MS) { + 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/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/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/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..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'; @@ -38,6 +39,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 +243,62 @@ export class ProjectsService { return mapProject(project); } + /** + * Per-project ingestion health (#279). Reads the clock skew counter written by + * 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, + projectId: string, + ): Promise<{ skew: ProjectSkewHealth | null }> { + const since = new Date(Date.now() - 24 * 60 * 60 * 1000); + + 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 }; + } + + 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: toNumber(row.count24h), + maxPastMs: toNumber(row.max_past_ms), + maxFutureMs: toNumber(row.max_future_ms), + lastSeenAt: lastSeenAt.toISOString(), + }, + }; + } + /** * Update a project (only active projects can be updated) */ 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/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(); + }); + }); }); 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..0b3e31e9 --- /dev/null +++ b/packages/backend/src/tests/modules/ingestion/skew.test.ts @@ -0,0 +1,48 @@ +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', () => { + 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(); + }); +}); 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); + }); + }); }); diff --git a/packages/backend/src/tests/modules/projects/routes.test.ts b/packages/backend/src/tests/modules/projects/routes.test.ts index 66261651..1a61370f 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'; @@ -469,4 +471,175 @@ 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 () => { + const olderEventTime = new Date(Date.now() - 60 * 60 * 1000); + const newestEventTime = new Date(Date.now() - 30 * 60 * 1000); + + 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: olderEventTime, + }, + { + organization_id: testOrganization.id, + project_id: testProject.id, + type: 'ingestion.timestamp_skew', + quantity: 6, + metadata: { maxPastMs: 90000000, maxFutureMs: 600000 }, + time: newestEventTime, + }, + ]) + .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); + // 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 () => { + 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); + }); + + 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(); + }); + + 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); + }); + }); }); 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 }; 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..eee20146 --- /dev/null +++ b/packages/frontend/src/lib/components/projects/IngestionSkewBanner.svelte @@ -0,0 +1,69 @@ + + +{#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. +

+

+ 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 new file mode 100644 index 00000000..bfa43e14 --- /dev/null +++ b/packages/frontend/src/lib/components/projects/IngestionSkewBanner.test.ts @@ -0,0 +1,81 @@ +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(); + }); + + 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(); + }); + + 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(); + }); +}); 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'} 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..212426ba 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; } @@ -162,6 +169,10 @@
+ {#if !loading && !error} + + {/if} + {#if loading}