fix: quarantine broken-warehouse orgs, skip revoked Cloudflare integrations, gate non-prod alerting crons

apps/alerting/alchemy.run.ts

@@ -94,6 +94,9 @@ export const createAlertingWorker = ({ stage, mapleDb }: CreateAlertingWorkerOpt
 				EMAIL_FROM: process.env.EMAIL_FROM?.trim() || "Maple <notifications@noreply.maple.dev>",
 				...optionalPlain("MAPLE_ENDPOINT"),
 				...optionalPlain("MAPLE_ENVIRONMENT", resolveDeploymentEnvironment(stage)),
+			// Non-prod stages skip all crons (they share live org data via the prod
+			// DB); set to "1" on a stage to deliberately exercise crons there.
+			...optionalPlain("MAPLE_ALERTING_ALLOW_NONPROD"),
 				...optionalPlain("COMMIT_SHA"),
 				MAPLE_INGEST_KEY: Redacted.make(requireEnv("MAPLE_OTEL_INGEST_KEY")),
 				...optionalSecret("MAPLE_ROOT_PASSWORD"),

apps/alerting/src/worker.ts

@@ -130,7 +130,9 @@ const buildLayer = (_env: Record<string, unknown>) => {
 	)
 
 	const DigestServiceLive = DigestService.layer.pipe(
-		Layer.provide(Layer.mergeAll(BaseLive, WarehouseQueryServiceLive, EmailServiceLive)),
+		Layer.provide(
+			Layer.mergeAll(BaseLive, WarehouseQueryServiceLive, EdgeCacheServiceLive, EmailServiceLive),
+		),
 	)
 
 	const OnboardingServiceLive = OnboardingService.layer.pipe(Layer.provide(BaseLive))
@@ -345,6 +347,21 @@ export default {
 		env: Record<string, unknown>,
 		ctx: ExecutionContextLike,
 	): Promise<void> {
+		// Non-prod stages (stg, PR previews) share live org data — stg's Hyperdrive
+		// points at the prod database — so their crons would iterate real orgs with
+		// stage-local Tinybird/Clerk credentials: every tick fails per-org and floods
+		// the error dashboards (and historically sent duplicate emails, see #237).
+		// Same gating philosophy as the prd-only EMAIL binding, with an explicit
+		// override for deliberately exercising crons on a non-prod stage.
+		const environment = typeof env.MAPLE_ENVIRONMENT === "string" ? env.MAPLE_ENVIRONMENT : ""
+		const allowNonProd =
+			env.MAPLE_ALERTING_ALLOW_NONPROD === "1" || env.MAPLE_ALERTING_ALLOW_NONPROD === "true"
+		if (environment !== "production" && !allowNonProd) {
+			console.log(
+				`Skipping alerting cron on non-production stage (MAPLE_ENVIRONMENT=${environment || "unset"}, cron=${event.cron}); set MAPLE_ALERTING_ALLOW_NONPROD=1 to run crons here`,
+			)
+			return
+		}
 		const program = Match.value(event.cron).pipe(
 			Match.when("*/5 * * * *", () =>
 				Effect.all([anomalyTick, cloudflareAnalyticsTick, planetScaleTick], {

apps/api/src/app.ts

@@ -228,7 +228,9 @@ const AiTriageServiceLive = AiTriageService.layer.pipe(Layer.provideMerge(CoreSe
 const InvestigationServiceLive = InvestigationService.layer.pipe(Layer.provideMerge(CoreServicesLive))
 
 const DigestServiceLive = DigestService.layer.pipe(
-	Layer.provideMerge(Layer.mergeAll(InfraLive, WarehouseQueryServiceLive, EmailServiceLive)),
+	Layer.provideMerge(
+		Layer.mergeAll(InfraLive, WarehouseQueryServiceLive, EdgeCacheServiceLive, EmailServiceLive),
+	),
 )
 
 // VCS service wiring for the fetch-path worker. VcsSyncService (the sync

apps/api/src/lib/warehouse-org-quarantine.test.ts

@@ -0,0 +1,77 @@
+import { assert, describe, it } from "@effect/vitest"
+import { WarehouseAuthError, WarehouseConfigError, WarehouseUpstreamError } from "@maple/domain/http"
+import { makeEdgeCacheService, makeMemoryBackend } from "@maple/query-engine/caching"
+import { Cause, Effect } from "effect"
+import {
+	causeHasWarehouseConfigClassError,
+	isOrgWarehouseQuarantined,
+	quarantineOnConfigClassCause,
+	quarantineOrgWarehouse,
+} from "./warehouse-org-quarantine"
+
+const authError = new WarehouseAuthError({ message: "invalid authentication token", pipeName: "p" })
+
+describe("causeHasWarehouseConfigClassError", () => {
+	it("matches a direct auth failure", () => {
+		assert.isTrue(causeHasWarehouseConfigClassError(Cause.fail(authError)))
+	})
+
+	it("matches a config failure", () => {
+		assert.isTrue(
+			causeHasWarehouseConfigClassError(
+				Cause.fail(new WarehouseConfigError({ message: "unknown database", pipeName: "p" })),
+			),
+		)
+	})
+
+	it("matches an auth error wrapped as the cause of a service error", () => {
+		const wrapped = { _tag: "@maple/http/errors/AnomalyPersistenceError", cause: authError }
+		assert.isTrue(causeHasWarehouseConfigClassError(Cause.fail(wrapped)))
+	})
+
+	it("matches an auth error raised as a defect", () => {
+		assert.isTrue(causeHasWarehouseConfigClassError(Cause.die(authError)))
+	})
+
+	it("does not match transient upstream failures", () => {
+		assert.isFalse(
+			causeHasWarehouseConfigClassError(
+				Cause.fail(new WarehouseUpstreamError({ message: "503", pipeName: "p" })),
+			),
+		)
+	})
+
+	it("does not match untagged errors", () => {
+		assert.isFalse(causeHasWarehouseConfigClassError(Cause.fail(new Error("boom"))))
+	})
+})
+
+describe("org quarantine", () => {
+	it.effect("round-trips through the edge cache", () =>
+		Effect.gen(function* () {
+			const edgeCache = makeEdgeCacheService(makeMemoryBackend())
+			assert.isFalse(yield* isOrgWarehouseQuarantined(edgeCache, "org_a"))
+			yield* quarantineOrgWarehouse(edgeCache, "org_a", 1_000)
+			assert.isTrue(yield* isOrgWarehouseQuarantined(edgeCache, "org_a"))
+			assert.isFalse(yield* isOrgWarehouseQuarantined(edgeCache, "org_b"))
+		}),
+	)
+
+	it.effect("quarantines only on config-class causes", () =>
+		Effect.gen(function* () {
+			const edgeCache = makeEdgeCacheService(makeMemoryBackend())
+			const transient = yield* quarantineOnConfigClassCause(
+				edgeCache,
+				"org_a",
+				Cause.fail(new WarehouseUpstreamError({ message: "503", pipeName: "p" })),
+				1_000,
+			)
+			assert.isFalse(transient)
+			assert.isFalse(yield* isOrgWarehouseQuarantined(edgeCache, "org_a"))
+
+			const config = yield* quarantineOnConfigClassCause(edgeCache, "org_a", Cause.fail(authError), 1_000)
+			assert.isTrue(config)
+			assert.isTrue(yield* isOrgWarehouseQuarantined(edgeCache, "org_a"))
+		}),
+	)
+})

apps/api/src/lib/warehouse-org-quarantine.ts

@@ -0,0 +1,74 @@
+import type { EdgeCacheServiceShape } from "@maple/query-engine/caching"
+import { Cause, Effect, Option } from "effect"
+
+/**
+ * Per-org quarantine for the cross-org alerting ticks (errors / anomalies /
+ * digests). An org whose warehouse rejects queries with an auth/config-class
+ * error (e.g. a Tinybird token whose workspace no longer exists) fails
+ * identically on every tick — retrying each 1–5 min only burns quota and floods
+ * the error dashboards. When a tick's per-org failure carries such an error,
+ * the org is parked in the edge cache and skipped until the TTL expires, which
+ * doubles as the automatic retry once the org's config is repaired.
+ *
+ * Scoped to warehouse-connectivity errors only: transient upstream failures
+ * (`WarehouseUpstreamError`) and genuine query bugs keep retrying every tick.
+ */
+
+const QUARANTINE_BUCKET = "warehouse-org-quarantine"
+const QUARANTINE_TTL_S = 6 * 60 * 60
+
+type EdgeCache = EdgeCacheServiceShape
+
+/**
+ * Errors that indicate the org's warehouse is unusable until an operator fixes
+ * its configuration — not until the next retry.
+ */
+const QUARANTINE_ERROR_TAGS: ReadonlySet<string> = new Set([
+	"@maple/http/errors/WarehouseAuthError",
+	"@maple/http/errors/WarehouseConfigError",
+])
+
+const hasQuarantineTag = (value: unknown, depth = 0): boolean => {
+	if (depth > 8 || typeof value !== "object" || value === null) return false
+	const candidate = value as { readonly _tag?: unknown; readonly cause?: unknown }
+	if (typeof candidate._tag === "string" && QUARANTINE_ERROR_TAGS.has(candidate._tag)) return true
+	// Services wrap warehouse failures in their own tagged errors (e.g.
+	// AnomalyPersistenceError) with the original error as `cause` — walk it.
+	return hasQuarantineTag(candidate.cause, depth + 1)
+}
+
+/** Does this cause (or anything wrapped inside it) carry an auth/config-class warehouse error? */
+export const causeHasWarehouseConfigClassError = <E>(cause: Cause.Cause<E>): boolean =>
+	cause.reasons.some((reason) =>
+		Cause.isFailReason(reason)
+			? hasQuarantineTag(reason.error)
+			: Cause.isDieReason(reason) && hasQuarantineTag(reason.defect),
+	)
+
+/** Best-effort check — a cache failure never blocks the tick, it just disables the skip. */
+export const isOrgWarehouseQuarantined = (edgeCache: EdgeCache, orgId: string) =>
+	edgeCache
+		.rawGet<number>(QUARANTINE_BUCKET, orgId)
+		.pipe(
+			Effect.map(Option.isSome),
+			Effect.orElseSucceed(() => false),
+		)
+
+/** Park the org for `QUARANTINE_TTL_S`. Best-effort; failures are ignored. */
+export const quarantineOrgWarehouse = (edgeCache: EdgeCache, orgId: string, nowMs: number) =>
+	edgeCache.rawPut(QUARANTINE_BUCKET, orgId, nowMs, QUARANTINE_TTL_S).pipe(Effect.ignore)
+
+/**
+ * Shared per-org failure seam for the alerting ticks: when the failure is
+ * config-class, quarantine the org and log at Info (expected, parked); anything
+ * else is logged at Error as before. Returns whether the org was quarantined.
+ */
+export const quarantineOnConfigClassCause = <E>(
+	edgeCache: EdgeCache,
+	orgId: string,
+	cause: Cause.Cause<E>,
+	nowMs: number,
+) =>
+	causeHasWarehouseConfigClassError(cause)
+		? quarantineOrgWarehouse(edgeCache, orgId, nowMs).pipe(Effect.as(true))
+		: Effect.succeed(false)

apps/api/src/services/AnomalyDetectionService.ts

@@ -36,6 +36,10 @@ import {
 import { and, desc, eq, gte, inArray, isNull, lt, lte, ne, or, sql } from "drizzle-orm"
 import { CH, parseWarehouseDateTime } from "@maple/query-engine"
 import { EdgeCacheService } from "@maple/query-engine/caching"
+import {
+	isOrgWarehouseQuarantined,
+	quarantineOnConfigClassCause,
+} from "../lib/warehouse-org-quarantine"
 import { Array as Arr, Cause, Clock, Context, Effect, Layer, Option, Ref, Schedule, Schema } from "effect"
 import type { TenantContext } from "./AuthService"
 import { AI_TRIAGE_WORKFLOW_BINDING, maybeEnqueueTriage } from "../lib/ai-triage-enqueue"
@@ -1844,7 +1848,17 @@ const make = Effect.gen(function* () {
 			const results = yield* Effect.forEach(
 				orgsToProcess,
 				(org) =>
-					processOrg(org, nowMs, runRetention).pipe(
+					Effect.gen(function* () {
+						// Orgs whose warehouse rejected queries with an auth/config-class
+						// error are parked (see warehouse-org-quarantine.ts).
+						if (yield* isOrgWarehouseQuarantined(edgeCache, org)) {
+							yield* Effect.logInfo("Skipping org with quarantined warehouse").pipe(
+								Effect.annotateLogs({ orgId: org }),
+							)
+							return emptyStats
+						}
+						return yield* processOrg(org, nowMs, runRetention)
+					}).pipe(
 						// Isolate genuine per-org failures/defects so one bad org can't fail
 						// the whole tick. Interrupts (isolate teardown) are NOT per-org
 						// failures — re-raise them so the tick cancels promptly instead of
@@ -1853,9 +1867,21 @@ const make = Effect.gen(function* () {
 							Cause.hasInterruptsOnly(cause)
 								? Effect.interrupt
 								: Effect.gen(function* () {
-										yield* Effect.logError("Anomaly tick failed for org").pipe(
-											Effect.annotateLogs({ orgId: org, error: Cause.pretty(cause) }),
+										const quarantined = yield* quarantineOnConfigClassCause(
+											edgeCache,
+											org,
+											cause,
+											nowMs,
 										)
+										if (quarantined) {
+											yield* Effect.logInfo(
+												"Org warehouse rejected queries with a config-class error; quarantined",
+											).pipe(Effect.annotateLogs({ orgId: org, error: Cause.pretty(cause) }))
+										} else {
+											yield* Effect.logError("Anomaly tick failed for org").pipe(
+												Effect.annotateLogs({ orgId: org, error: Cause.pretty(cause) }),
+											)
+										}
 										yield* Ref.update(orgFailures, (n) => n + 1)
 										return emptyStats
 									}),

apps/api/src/services/CloudflareAnalyticsService.test.ts

@@ -12,6 +12,7 @@ import {
 import { ConfigProvider, Effect, Layer, Schema } from "effect"
 import { TestClock } from "effect/testing"
 import { FetchHttpClient } from "effect/unstable/http"
+import { eq } from "drizzle-orm"
 import { encryptAes256Gcm, parseBase64Aes256GcmKey } from "../lib/Crypto"
 import { Database } from "../lib/DatabaseLive"
 import { Env } from "../lib/Env"
@@ -845,6 +846,32 @@ describe("CloudflareAnalyticsService", () => {
 		}).pipe(Effect.provide(makeLayer(testDb, captured)))
 	})
 
+	it.effect("pollAllOrgs skips connections stamped revoked; resetOrgState clears the stamp", () => {
+		const testDb = createTestDb(trackedDbs)
+		const captured: CapturedIngest[] = []
+		return Effect.gen(function* () {
+			yield* TestClock.setTime(T0)
+			yield* seedConnection()
+			const database = yield* Database
+			yield* database.execute((db) =>
+				db
+					.update(oauthConnections)
+					.set({ revokedAt: new Date(T0 - MIN) })
+					.where(eq(oauthConnections.orgId, ORG)),
+			)
+			const service = yield* CloudflareAnalyticsService
+			const result = yield* service.pollAllOrgs()
+			assert.strictEqual(result.perOrg.length, 0)
+			assert.strictEqual(result.rowsIngested, 0)
+
+			yield* service.resetOrgState(ORG)
+			const rows = yield* database.execute((db) =>
+				db.select().from(oauthConnections).where(eq(oauthConnections.orgId, ORG)),
+			)
+			assert.isNull(rows[0]!.revokedAt)
+		}).pipe(Effect.provide(makeLayer(testDb, captured)))
+	})
+
 	it.effect("resetOrgState re-enables disabled rows and clears error/discovery state", () => {
 		const testDb = createTestDb(trackedDbs)
 		const captured: CapturedIngest[] = []

apps/api/src/services/CloudflareAnalyticsService.ts

@@ -1695,6 +1695,9 @@ export class CloudflareAnalyticsService extends Context.Service<
 				// failures record and retry next tick.
 				if (Result.isFailure(tokenResult)) {
 					const error = tokenResult.failure
+					if (error instanceof IntegrationsRevokedError) {
+						yield* oauth.markConnectionRevoked(orgId)
+					}
 					yield* recordOrgError(orgId, error.message, now, {
 						disable: error instanceof IntegrationsRevokedError,
 					})
@@ -1715,6 +1718,9 @@ export class CloudflareAnalyticsService extends Context.Service<
 					const zonesResult = yield* Effect.result(listZones(accessToken, accountId, apiBaseUrl))
 					if (Result.isFailure(zonesResult)) {
 						const error = zonesResult.failure
+						if (error instanceof IntegrationsRevokedError) {
+							yield* oauth.markConnectionRevoked(orgId)
+						}
 						yield* recordOrgError(orgId, error.message, now, {
 							disable: error instanceof IntegrationsRevokedError,
 						})
@@ -1803,7 +1809,10 @@ export class CloudflareAnalyticsService extends Context.Service<
 						).pipe(
 							Effect.catchTag("@maple/http/errors/IntegrationsRevokedError", (error) =>
 								// Stop this org's loop; the seam disables + records health org-wide.
-								Ref.set(revokedRef, true).pipe(
+								// Also stamp the connection so pollAllOrgs skips it until reconnect
+								// (a mid-poll 401 never goes through the refresh path that stamps).
+								oauth.markConnectionRevoked(orgId).pipe(
+									Effect.andThen(Ref.set(revokedRef, true)),
 									Effect.as(
 										allPartsFailed(item, "revoked", error.message, {
 											orgWide: true,
@@ -1919,7 +1928,9 @@ export class CloudflareAnalyticsService extends Context.Service<
 				db
 					.selectDistinct({ orgId: oauthConnections.orgId, scope: oauthConnections.scope })
 					.from(oauthConnections)
-					.where(eq(oauthConnections.provider, "cloudflare")),
+					// Revoked grants fail identically every tick until the org reconnects
+					// (which clears revokedAt) — skip them instead of re-erroring forever.
+					.where(and(eq(oauthConnections.provider, "cloudflare"), isNull(oauthConnections.revokedAt))),
 			)
 			// Orgs whose grant lacks the analytics scopes can't be polled — the status endpoint
 			// already surfaces that (analyticsCapable: false), so skipping them here avoids
@@ -2239,6 +2250,15 @@ export class CloudflareAnalyticsService extends Context.Service<
 					})
 					.where(eq(cloudflareAnalyticsState.orgId, orgId)),
 			)
+			// Manual recovery path: also clear the revocation stamp so pollAllOrgs
+			// picks the org up again (reconnect clears it via upsertConnection, but
+			// resetOrgState may be invoked without a fresh OAuth round-trip).
+			yield* dbExecute((db) =>
+				db
+					.update(oauthConnections)
+					.set({ revokedAt: null })
+					.where(and(eq(oauthConnections.orgId, orgId), eq(oauthConnections.provider, "cloudflare"))),
+			)
 		})
 
 		return {

apps/api/src/services/CloudflareOAuthService.ts

@@ -118,6 +118,11 @@ export interface CloudflareOAuthServiceShape {
 	readonly disconnect: (
 		orgId: OrgId,
 	) => Effect.Effect<{ readonly disconnected: boolean }, IntegrationsPersistenceError>
+	/**
+	 * Stamp the org's connection as revoked so pollers stop retrying it every
+	 * tick; cleared automatically when the org reconnects. Best-effort.
+	 */
+	readonly markConnectionRevoked: (orgId: OrgId) => Effect.Effect<void>
 }
 
 export class CloudflareOAuthService extends Context.Service<
@@ -339,6 +344,7 @@ export class CloudflareOAuthService extends Context.Service<
 			getStatus,
 			getValidAccessToken,
 			disconnect,
+			markConnectionRevoked: oauth.markConnectionRevoked,
 		} satisfies CloudflareOAuthServiceShape
 	}),
 }) {