Ingestion clock skew detection (#279)

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

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(),
         };

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,19 +157,25 @@ 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 }),
       };
 
       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

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;
+    },
+  };
+}

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;

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;

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 {

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)
    */

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', () => {

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();
+        });
+    });
 });