Name every Cloud diagnostics 429 and keep failures and drop reports past the budget

web/app/api/observability/cloud/route.ts

@@ -16,7 +16,8 @@ export const POST = makeCloudTelemetryHandler({
       throw new Error("diagnostics_rate_limit_unconfigured");
     }
     const result = await checkRateLimit(rule, { request });
-    if (result.rateLimited || result.error === "blocked") return false;
+    // Names the firewall rule that refused, so a shared or misconfigured rule is visible in Axiom.
+    if (result.rateLimited || result.error === "blocked") return { rateLimitRule: rule };
     if (result.error) throw new Error("diagnostics_rate_limit_unavailable");
     return true;
   },

web/services/observability/cloudTelemetryExport.ts

@@ -2,6 +2,11 @@ import { createHmac } from "node:crypto";
 import type { CloudTelemetryAuth, CloudTelemetrySpan } from "./cloudTelemetryContract";
 import type { StoredCloudDiagnostic } from "./cloudTelemetryRepository";
 
+/** The keyed account hash shared by exported diagnostics and the ingest route span. */
+export function cloudTelemetryAccountKey(identityKey: string, userId: string): string {
+  return createHmac("sha256", identityKey).update(userId).digest("hex");
+}
+
 export type CloudAxiomConfiguration = {
   readonly origin: string;
   readonly token: string;
@@ -64,7 +69,7 @@ export async function exportCloudDiagnostics(
     backend_tag: row.payload.backend?.tag,
     backend_source_sha256: row.payload.backend?.sourceSha256,
     backend_revision: row.payload.backend?.revision ?? "unknown",
-    account_key: createHmac("sha256", configuration.identityKey).update(row.userId).digest("hex"),
+    account_key: cloudTelemetryAccountKey(configuration.identityKey, row.userId),
     source: row.payload.source ?? "client",
     event_id: row.eventId, operation_id: row.payload.span.operationId,
     trace_id: row.payload.span.traceId, span_id: row.payload.span.spanId,
@@ -117,7 +122,7 @@ function resourceAttributes(row: StoredCloudDiagnostic, configuration: CloudAxio
     "service.instance.id": row.payload.source === "server" ? undefined : client.instanceId,
     "os.version": row.payload.source === "server" ? undefined : client.osVersion,
     "host.arch": row.payload.source === "server" ? undefined : client.architecture,
-    "cmux.account_key": createHmac("sha256", configuration.identityKey).update(row.userId).digest("hex"),
+    "cmux.account_key": cloudTelemetryAccountKey(configuration.identityKey, row.userId),
     "cmux.observation.source": row.payload.source ?? "client",
   };
 }

web/services/observability/cloudTelemetryIngest.ts

@@ -1,19 +1,29 @@
-import { trace } from "@opentelemetry/api";
+import { SpanKind, trace, type Attributes } from "@opentelemetry/api";
 import { Effect } from "effect";
 import { readBoundedJsonObject } from "../apns/routePolicy";
 import { CLOUD_TELEMETRY_MAX_BYTES, parseCloudTelemetryUpload, type CloudTelemetryBatch, type CloudTelemetryUpload } from "./cloudTelemetryContract";
+import { cloudTelemetryAccountKey } from "./cloudTelemetryExport";
 import { sampleCloudTelemetryBatch } from "./cloudTelemetrySampling";
 
 export type CloudTelemetryIngestDependencies = {
   readonly authenticate: (request: Request) => Promise<{ readonly id: string } | null>;
-  readonly checkIngress: (request: Request) => Promise<boolean>;
+  /**
+   * The pre-authentication gate. `true` admits the request; `false` or a refusal
+   * naming the firewall rule answers 429 with refusal `ingress_rate_limited`.
+   */
+  readonly checkIngress: (request: Request) => Promise<boolean | CloudTelemetryIngressRefusal>;
   /** Stores the sampled spans. `sampleWeights` lists spans that stand for more than one submitted span. */
   readonly accept: (userId: string, batch: CloudTelemetryBatch, sampleWeights: ReadonlyMap<string, number>) => Promise<number>;
   readonly scheduleDrain: () => void;
   readonly now: () => number;
 };
 
-export class CloudTelemetryLimitError extends Error {}
+export type CloudTelemetryIngressRefusal = { readonly rateLimitRule: string };
+
+/** The account's byte budget refused the batch. `storedSpans` were kept anyway, from the reserve or earlier receipts. */
+export class CloudTelemetryLimitError extends Error {
+  constructor(readonly storedSpans = 0) { super("cloud_diagnostics_byte_budget"); }
+}
 export class CloudTelemetryConflictError extends Error {}
 
 /** The HTTP adapter runs one Effect program; delivery is owned by the durable outbox. */
@@ -25,26 +35,43 @@ export function makeCloudTelemetryHandler(dependencies: CloudTelemetryIngestDepe
 
 function ingest(request: Request, dependencies: CloudTelemetryIngestDependencies) {
   return Effect.gen(function* () {
-    const allowed = yield* Effect.tryPromise(() => dependencies.checkIngress(request));
-    if (!allowed) return response(429, "rate_limited");
+    // Identity labels gathered as the request is understood; refusals carry what is known so far.
+    const labels: Record<string, string> = {};
+    const ingress = yield* Effect.tryPromise(() => dependencies.checkIngress(request));
+    if (ingress !== true) {
+      if (ingress) labels["cmux.telemetry.rate_limit_rule"] = ingress.rateLimitRule;
+      // Before authentication and parsing, so only the cause and the rule are known.
+      return refuse(429, "rate_limited", { submitted: 0, accepted: 0, rejected: 0, coerced: 0, refusal: "ingress_rate_limited" }, labels, dependencies.now());
+    }
     const encoding = request.headers.get("content-encoding");
     if (encoding && encoding !== "identity") return response(415, "unsupported_encoding");
     if (request.headers.get("content-type")?.split(";")[0]?.trim() !== "application/json") return response(415, "unsupported_content_type");
     const user = yield* Effect.tryPromise(() => dependencies.authenticate(request));
     if (!user) return response(401, "unauthorized");
+    const accountKey = cloudTelemetryAccountKeyFromEnvironment(user.id);
+    if (accountKey) labels["cmux.account_key"] = accountKey;
     const body = yield* Effect.tryPromise(() => readBoundedJsonObject(request, CLOUD_TELEMETRY_MAX_BYTES));
     if (!body.ok) {
-      recordUploadOutcome({ submitted: 0, accepted: 0, rejected: 0, coerced: 0, refusal: body.error });
-      return response(body.error === "request_too_large" ? 413 : 400, body.error);
+      return refuse(body.error === "request_too_large" ? 413 : 400, body.error, { submitted: 0, accepted: 0, rejected: 0, coerced: 0, refusal: body.error }, labels, dependencies.now());
     }
     const parsed = parseCloudTelemetryUpload(body.value, dependencies.now());
     if (!parsed.ok) {
-      recordUploadOutcome({ submitted: parsed.submitted, accepted: 0, rejected: parsed.submitted, coerced: 0, refusal: parsed.reason });
-      return response(400, "invalid_diagnostics");
+      return refuse(400, "invalid_diagnostics", { submitted: parsed.submitted, accepted: 0, rejected: parsed.submitted, coerced: 0, refusal: parsed.reason }, labels, dependencies.now());
     }
     const { upload } = parsed;
-    yield* store(user.id, upload.batch, dependencies);
-    recordUploadOutcome(uploadCounts(upload));
+    if (upload.batch.client.instanceId) labels["cmux.client.instance_id"] = upload.batch.client.instanceId;
+    yield* store(user.id, upload.batch, dependencies).pipe(
+      Effect.tapError((error) => Effect.sync(() => {
+        const stored = error instanceof CloudTelemetryLimitError ? error.storedSpans : 0;
+        // Evidence kept from a refused batch is durable now; deliver it without waiting for the cron.
+        if (stored > 0) scheduleDrainSafely(dependencies);
+        // Budget, conflict and storage refusals follow authentication and client backoff, so each is recorded.
+        recordRefusal(storeRefusalStatus(error), {
+          ...uploadCounts(upload), accepted: 0, refusal: storeRefusal(error), ...(stored > 0 ? { stored } : {}),
+        }, labels);
+      })),
+    );
+    recordUploadOutcome(uploadCounts(upload), labels);
     // Sampled-out and rejected spans are acknowledged too: the client must not retry or re-queue them.
     return new Response(JSON.stringify({
       accepted: upload.batch.spans.length,
@@ -65,14 +92,20 @@ function store(userId: string, batch: CloudTelemetryBatch, dependencies: CloudTe
     yield* Effect.tryPromise({
       try: () => dependencies.accept(userId, { ...batch, spans: sampled.spans }, sampled.sampleWeights), catch: (error) => error,
     });
-    // A failed scheduling call must not revoke a durable receipt. The cron drain remains responsible.
-    yield* Effect.sync(() => dependencies.scheduleDrain()).pipe(Effect.catchAllDefect(() => Effect.void));
+    yield* Effect.sync(() => scheduleDrainSafely(dependencies));
   });
 }
 
+/** A failed scheduling call must not revoke a durable receipt. The cron drain remains responsible. */
+function scheduleDrainSafely(dependencies: CloudTelemetryIngestDependencies): void {
+  try { dependencies.scheduleDrain(); } catch { /* the cron drain delivers it */ }
+}
+
 type UploadOutcome = {
   readonly submitted: number; readonly accepted: number; readonly rejected: number; readonly coerced: number;
   readonly refusal?: string; readonly rejectedReasons?: string;
+  /** Spans of a refused batch that are stored anyway (reserve evidence or earlier receipts of a retry). */
+  readonly stored?: number;
 };
 
 function uploadCounts(upload: CloudTelemetryUpload): UploadOutcome {
@@ -83,16 +116,91 @@ function uploadCounts(upload: CloudTelemetryUpload): UploadOutcome {
   };
 }
 
-/** Counts on the route span, so Axiom shows partial acceptance and refusal causes. Codes only, never submitted text. */
-function recordUploadOutcome(outcome: UploadOutcome): void {
-  trace.getActiveSpan()?.setAttributes({
+function outcomeAttributes(outcome: UploadOutcome, labels: Readonly<Record<string, string>>): Attributes {
+  return {
+    ...labels,
     "cmux.telemetry.submitted_spans": outcome.submitted,
     "cmux.telemetry.accepted_spans": outcome.accepted,
     "cmux.telemetry.rejected_spans": outcome.rejected,
     "cmux.telemetry.coerced_spans": outcome.coerced,
     ...(outcome.refusal ? { "cmux.telemetry.refusal": outcome.refusal } : {}),
     ...(outcome.rejectedReasons ? { "cmux.telemetry.rejected_reasons": outcome.rejectedReasons } : {}),
-  });
+    ...(outcome.stored ? { "cmux.telemetry.stored_spans": outcome.stored } : {}),
+  };
+}
+
+/** Counts on the route span, so Axiom shows partial acceptance and refusal causes. Codes only, never submitted text. */
+function recordUploadOutcome(outcome: UploadOutcome, labels: Readonly<Record<string, string>>): void {
+  trace.getActiveSpan()?.setAttributes(outcomeAttributes(outcome, labels));
+}
+
+/**
+ * The route span is head-sampled with the app-wide 2% ratio, so a refusal also
+ * leaves its own always-kept span: refused uploads are countable in Axiom by
+ * cause, firewall rule, account and install. `cmux.telemetry.refusal_count` is
+ * the number of refusals the span stands for.
+ */
+function recordRefusal(status: number, outcome: UploadOutcome, labels: Readonly<Record<string, string>>, refusalCount = 1): void {
+  recordUploadOutcome(outcome, labels);
+  trace.getTracer("cmux.cloud-diagnostics").startSpan(CLOUD_TELEMETRY_REFUSAL_SPAN, {
+    root: true, kind: SpanKind.INTERNAL,
+    attributes: {
+      ...outcomeAttributes(outcome, labels), "cmux.priority": true, "http.response.status_code": status,
+      "cmux.telemetry.refusal_count": refusalCount,
+    },
+  }).end();
+}
+
+/**
+ * Refusals a caller can repeat without backing off (a firewall refusal before
+ * authentication, a malformed body) are coalesced per cause, rule and account:
+ * at most one kept span per key per window on each server instance, carrying the
+ * count. A flood the firewall stops therefore cannot become a flood of kept spans.
+ * Refusals still pending when an instance stops are not reported.
+ */
+const REFUSAL_COALESCE_WINDOW_MS = 10_000;
+const REFUSAL_COALESCE_MAX_KEYS = 1_000;
+const pendingRefusals = new Map<string, { readonly emittedAtMs: number; readonly suppressed: number }>();
+
+function refuse(status: number, error: string, outcome: UploadOutcome, labels: Readonly<Record<string, string>>, nowMs: number): Response {
+  const key = [outcome.refusal, labels["cmux.telemetry.rate_limit_rule"], labels["cmux.account_key"]].join("\u0000");
+  const pending = pendingRefusals.get(key);
+  if (pending && nowMs - pending.emittedAtMs < REFUSAL_COALESCE_WINDOW_MS) {
+    pendingRefusals.set(key, { ...pending, suppressed: pending.suppressed + 1 });
+    recordUploadOutcome(outcome, labels);
+  } else {
+    if (!pending && pendingRefusals.size >= REFUSAL_COALESCE_MAX_KEYS) pendingRefusals.clear();
+    pendingRefusals.set(key, { emittedAtMs: nowMs, suppressed: 0 });
+    recordRefusal(status, outcome, labels, 1 + (pending?.suppressed ?? 0));
+  }
+  return response(status, error);
+}
+
+/** The always-sampled span name for a refused upload. */
+export const CLOUD_TELEMETRY_REFUSAL_SPAN = "cmux.cloud_diagnostics.refusal";
+
+
+function storeRefusalStatus(error: unknown): number {
+  if (error instanceof CloudTelemetryLimitError) return 429;
+  if (error instanceof CloudTelemetryConflictError) return 409;
+  return 503;
+}
+
+/** The refusal code of a storage error, matching the status `errorResponse` answers. */
+function storeRefusal(error: unknown): string {
+  if (error instanceof CloudTelemetryLimitError) return "account_byte_budget";
+  if (error instanceof CloudTelemetryConflictError) return "event_conflict";
+  return "storage_unavailable";
+}
+
+/**
+ * The same keyed account hash as exported client diagnostics, so Axiom can join
+ * upload outcomes to the accounts that sent them. Without the key the account
+ * is left out, never sent raw.
+ */
+function cloudTelemetryAccountKeyFromEnvironment(userId: string): string | undefined {
+  const key = process.env.CMUX_CLOUD_TELEMETRY_ID_KEY?.trim();
+  return key && key.length >= 32 ? cloudTelemetryAccountKey(key, userId) : undefined;
 }
 
 function errorResponse(error: unknown): Response {

web/services/observability/cloudTelemetryRepository.ts

@@ -16,6 +16,21 @@ export type StoredCloudDiagnostic = {
   readonly attempts: number;
 };
 
+/** An account's routine diagnostics budget per minute: storage bytes of newly accepted spans. */
+export const CLOUD_DIAGNOSTICS_BYTES_PER_MINUTE = 262_144;
+/**
+ * Failure, timeout, cancellation and sign-in records may use this reserve above
+ * the routine budget. When routine success detail has filled the minute, a batch
+ * still stores its evidence before it answers 429, so the evidence is durable on
+ * the first attempt and the client's retry deduplicates it. The ceiling is
+ * budget plus reserve.
+ */
+export const CLOUD_DIAGNOSTICS_PRIORITY_RESERVE_BYTES = 131_072;
+
+function isPriorityCloudDiagnostic(span: CloudTelemetrySpan): boolean {
+  return span.outcome !== "success" || span.auth !== undefined;
+}
+
 /** Account quota and deduplication are transactional across all server instances. */
 export async function acceptCloudTelemetry(
   userId: string, batch: CloudTelemetryBatch,
@@ -36,9 +51,13 @@ export async function acceptCloudTelemetry(
     // The sample weight is ingest policy, not submitted evidence, so it stays outside the hash.
     const sampleWeight = sampleWeights?.get(span.eventId);
     const stored = { ...payload, backend, ...(sampleWeight && sampleWeight !== 1 ? { sampleWeight } : {}) };
-    return { id: span.eventId, payload: canonicalJSON(stored), hash: createHash("sha256").update(encoded).digest("hex") };
+    return {
+      id: span.eventId, payload: canonicalJSON(stored), hash: createHash("sha256").update(encoded).digest("hex"),
+      priority: isPriorityCloudDiagnostic(span),
+    };
   });
-  return cloudDb().transaction(async (tx) => {
+  /** Pending spans left unstored by the budget; zero accepts the whole batch. */
+  const refused = await cloudDb().transaction(async (tx) => {
     await tx.execute(sql`set local statement_timeout = '3000ms'`);
     // One account lock also makes duplicate-content checks race-free.
     await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`cloud-diagnostics:${userId}`}, 0))`);
@@ -52,21 +71,33 @@ export async function acceptCloudTelemetry(
       if (previous && previous !== row.hash) throw new CloudTelemetryConflictError();
     }
     const pending = rows.filter((row) => !existing.has(row.id));
-    if (pending.length === 0) return rows.length;
-    const bytes = pending.reduce((sum, row) => sum + Buffer.byteLength(row.payload), 0);
+    if (pending.length === 0) return 0;
     const minute = Math.floor(Date.now() / 60_000);
-    const quota = await tx.execute(sql`
-      insert into cloud_diagnostic_budgets (user_id, minute, bytes) values (${userId}, ${minute}, ${bytes})
-      on conflict (user_id, minute) do update set bytes = cloud_diagnostic_budgets.bytes + excluded.bytes
-      where cloud_diagnostic_budgets.bytes + excluded.bytes <= 262144 returning bytes
-    `);
-    if (!quota[0]) throw new CloudTelemetryLimitError();
-    await tx.execute(sql`
-      insert into cloud_diagnostic_events (user_id, event_id, payload, payload_hash)
-      values ${sql.join(pending.map((row) => sql`(${userId}, ${row.id}::uuid, ${row.payload}::jsonb, ${row.hash})`), sql`, `)}
-    `);
-    return rows.length;
+    const charge = async (charged: typeof pending, limit: number) => {
+      const bytes = charged.reduce((sum, row) => sum + Buffer.byteLength(row.payload), 0);
+      const quota = await tx.execute(sql`
+        insert into cloud_diagnostic_budgets (user_id, minute, bytes) values (${userId}, ${minute}, ${bytes})
+        on conflict (user_id, minute) do update set bytes = cloud_diagnostic_budgets.bytes + excluded.bytes
+        where cloud_diagnostic_budgets.bytes + excluded.bytes <= ${limit} returning bytes
+      `);
+      if (!quota[0]) return false;
+      await tx.execute(sql`
+        insert into cloud_diagnostic_events (user_id, event_id, payload, payload_hash)
+        values ${sql.join(charged.map((row) => sql`(${userId}, ${row.id}::uuid, ${row.payload}::jsonb, ${row.hash})`), sql`, `)}
+      `);
+      return true;
+    };
+    if (await charge(pending, CLOUD_DIAGNOSTICS_BYTES_PER_MINUTE)) return 0;
+    // Over the routine budget: keep the batch's evidence within the reserve and refuse the rest.
+    // A retry deduplicates the stored evidence, so only its routine spans are charged again.
+    const priority = pending.filter((row) => row.priority);
+    if (priority.length === 0) return pending.length;
+    const kept = await charge(priority, CLOUD_DIAGNOSTICS_BYTES_PER_MINUTE + CLOUD_DIAGNOSTICS_PRIORITY_RESERVE_BYTES);
+    return kept ? pending.length - priority.length : pending.length;
   });
+  // Thrown after the commit, so evidence kept from a refused batch stays stored.
+  if (refused > 0) throw new CloudTelemetryLimitError(rows.length - refused);
+  return rows.length;
 }
 
 export async function claimCloudDiagnostics(limit = 100, onlyOwner?: string): Promise<{ leaseId: string; rows: StoredCloudDiagnostic[] }> {