From 0a079652a372ddd8d0a8eb0921bf24451be81ce0 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Tue, 10 Nov 2026 02:18:00 +0100 Subject: [PATCH 1/4] fix(infra): one warehouse env for every querying Worker, and small Worker fixes Worker env - The managed ClickHouse read endpoint (`CLICKHOUSE_URL`, provider, user, database, password) was bound on api only, so alerting ticks and agent tools always read the default upstream even when the dashboard read another. It is now one shared `clickhouseEnv` in `@maple/infra/env`, merged into api, alerting and ai. api's resulting env is unchanged. - `bun dev` hands the scraper `MAPLE_INGEST_URL` from the ingest route. Its default is a fixed port that ingest's dev script does not listen on. - Remove `MAPLE_EMAIL_ALLOW_NONPROD`. No Worker bound it, and the send binding only exists on prd, so the "not configured" branch always answered first and the override message could not be reached. `bindEmailSender` is the one gate. api queues - The audit and cancellation retry limits were each written twice under "must match" comments. Both now come from `queue-retry-limits.ts`, a dependency-free module the consumer settings and the batch handlers share. - Drop the three `renamedFrom` shims on the queue consumers: every long-lived stage has deployed past them many times since 2026-09-07. status - A miss on the host lookup is cached for the same window as a hit. The shared host has no page, so every `/s/` request paid one uncached KV read before the slug lookup. Resolving the slug first was not used: it would let a host with its own page serve another page by slug. A miss never replaces a cached page id. electric-sync - Remove the v1 `alert_destinations` shape. The web app has requested `alert_destinations_v2` since 2026-10-13 and nothing else asks for v1. - Point the publication comments at the timestamped migration directories. Comments - scraper: say that the in-request retry is narrower than the batch reporter's on purpose (delivery is at-least-once). No behavior change. - chat-bot: the webhook route comment said nothing delivers by webhook and the Worker has no hostname. Both stopped being true. --- alchemy.run.ts | 2 ++ apps/ai/src/worker.ts | 3 ++ apps/alerting/src/worker.ts | 6 ++-- apps/api/src/audit-events-runtime.ts | 13 +++----- apps/api/src/cancellation-review-runtime.ts | 7 +---- apps/api/src/queue-retry-limits.ts | 8 +++++ apps/api/src/resources/env.ts | 7 ++--- apps/api/src/worker/consumers.ts | 23 ++++++-------- apps/chat-bot/src/routes/webhook.ts | 6 ++-- .../src/electric/ElectricClient.test.ts | 29 ++++++++---------- .../electric-sync/src/shapes/registry.test.ts | 5 ++-- apps/electric-sync/src/shapes/registry.ts | 24 ++++----------- apps/scraper/src/ApiClient.ts | 5 +++- apps/scraper/src/Env.ts | 10 +++---- apps/status/src/handler.test.ts | 24 ++++++++++++++- apps/status/src/snapshot.ts | 30 +++++++++++++++---- apps/status/src/test-support.ts | 10 +++++-- docs/electric-sync.md | 3 +- packages/backend/src/platform/EmailService.ts | 16 ++-------- packages/backend/src/platform/Env.ts | 5 ---- packages/infra/src/env.ts | 12 ++++++++ 21 files changed, 137 insertions(+), 111 deletions(-) create mode 100644 apps/api/src/queue-retry-limits.ts diff --git a/alchemy.run.ts b/alchemy.run.ts index 4415dd10c3..ea2c9e57c4 100644 --- a/alchemy.run.ts +++ b/alchemy.run.ts @@ -271,6 +271,8 @@ const createDevProcess = (app: DevApp, route: Portless.Route) => PORT: Output.map(Output.asOutput(route.port), String), PORTLESS_URL: Portless.routeUrl(app), MAPLE_API_URL: Portless.routeUrl("api"), + // The scraper's default is a fixed port; ingest's dev script listens on its route's port. + MAPLE_INGEST_URL: Portless.routeUrl("ingest"), // The alchemy CLI sets NODE_ENV=production and children inherit it; vite and // astro read it for `import.meta.env.DEV`/`PROD`, so dev servers must override it. NODE_ENV: "development", diff --git a/apps/ai/src/worker.ts b/apps/ai/src/worker.ts index 7d34085f76..e2fa0a4736 100644 --- a/apps/ai/src/worker.ts +++ b/apps/ai/src/worker.ts @@ -15,6 +15,7 @@ import { import { appUrlsEnv, authEnv, + clickhouseEnv, githubAppSourceEnv, ingestKeyCryptoEnv, merge, @@ -43,6 +44,8 @@ import { AiObservabilityLive } from "./worker/observability" const configuredEnv = ({ stage, region, domains, profile }: MapleStackContext) => merge( tinybirdEnv, + // Same upstream as api, so a tool answers from the warehouse the dashboard reads. + clickhouseEnv, authEnv, appUrlsEnv(domains), selfObservabilityEnv(stage, region), diff --git a/apps/alerting/src/worker.ts b/apps/alerting/src/worker.ts index 67e44a2a14..b973d7ec27 100644 --- a/apps/alerting/src/worker.ts +++ b/apps/alerting/src/worker.ts @@ -20,6 +20,7 @@ import { apnsEnv, appUrlsEnv, authEnv, + clickhouseEnv, cloudflareOAuthEnv, githubAppSourceEnv, ingestKeyCryptoEnv, @@ -56,11 +57,12 @@ export type AlertingWorkerEnv = Record const configuredEnv = (stage: MapleStage, region: MapleRegion, domains: MapleDomains) => merge( tinybirdEnv, + // Same upstream as api, so a tick evaluates the warehouse the dashboard reads. + clickhouseEnv, authEnv, ingestKeyCryptoEnv, appUrlsEnv(domains), - // MAPLE_ENVIRONMENT must stay stage-derived, not overridable: it gates both the - // non-prod cron skip and `EmailService.emailAllowed`. + // MAPLE_ENVIRONMENT must stay stage-derived, not overridable: it gates the non-prod cron skip. selfObservabilityEnv(stage, region), // "1" runs crons on a non-prod stage (which otherwise skips them). optionalPlain("MAPLE_ALERTING_ALLOW_NONPROD"), diff --git a/apps/api/src/audit-events-runtime.ts b/apps/api/src/audit-events-runtime.ts index 3c3418061e..b1f9aa4165 100644 --- a/apps/api/src/audit-events-runtime.ts +++ b/apps/api/src/audit-events-runtime.ts @@ -13,6 +13,7 @@ import { auditEventToRow, decodeAuditLogEvent, } from "@maple/backend/services/audit/audit-event" +import { AUDIT_EVENTS_MAX_RETRIES } from "./queue-retry-limits" /** * The consumer writes through `WarehouseQueryService.ingest`, which pins every @@ -25,14 +26,6 @@ export const AuditEventsLive = WarehouseQueryService.layer.pipe( Layer.provide(Layer.mergeAll(EventBaseLive, EdgeCacheServiceLive)), ) -/** - * Must match `maxRetries` on the audit-events consumer in `worker.ts`. - * Cloudflare routes the message to the DLQ after this many retries without - * telling us; the check below is what makes the hand-off visible in logs at - * the moment it happens. - */ -const AUDIT_EVENTS_MAX_RETRIES = 5 - /** * Best-effort identity for the exhaustion log. The body reached us as queue * JSON and may be anything at all, so these read defensively rather than @@ -54,7 +47,9 @@ interface DecodedMessage { /** * Retrying past the limit is what hands the message to the DLQ; acking there - * would silently discard it instead. + * would silently discard it instead. Cloudflare makes that hand-off without + * telling us, so the final-attempt check is what puts it in the logs. The + * limit is the consumer's own `maxRetries` (`worker/consumers.ts`). */ const retryOrExhaust = (message: Message, cause: unknown) => { const isFinalAttempt = message.attempts > AUDIT_EVENTS_MAX_RETRIES diff --git a/apps/api/src/cancellation-review-runtime.ts b/apps/api/src/cancellation-review-runtime.ts index 164e85f336..f0459eab2d 100644 --- a/apps/api/src/cancellation-review-runtime.ts +++ b/apps/api/src/cancellation-review-runtime.ts @@ -6,6 +6,7 @@ import { EventBaseLive } from "@maple/backend/platform/DatabasePgLive" import type { QueueBatch } from "@maple/backend/platform/queue-batch" import { CancellationReviewJob } from "@maple/backend/services/cancellation-review/CancellationReviewQueue" import { CancellationReviewService } from "@maple/backend/services/cancellation-review/CancellationReviewService" +import { CANCELLATION_REVIEW_MAX_RETRIES } from "./queue-retry-limits" /** Not `maple-api`: a review's minute of upstream reads would skew the request service's percentiles. */ export const cancellationReviewTelemetry = eventTelemetry({ serviceName: "maple-cancellation-reviews" }) @@ -18,12 +19,6 @@ export const CancellationReviewLive = CancellationReviewService.layer.pipe( Layer.provide(Layer.mergeAll(EventBaseLive, EdgeCacheServiceLive)), ) -/** - * Must match `maxRetries` on the cancellation-reviews consumer - * (`worker/consumers.ts`). The attempt past it logs the drop and acks. - */ -const CANCELLATION_REVIEW_MAX_RETRIES = 5 - /** * 1, 2, 4, 8, 16 minutes: about half an hour for Clerk, Autumn, Postgres or * Slack to come back. A review waiting out another delivery's claim (a diff --git a/apps/api/src/queue-retry-limits.ts b/apps/api/src/queue-retry-limits.ts new file mode 100644 index 0000000000..e1178a062a --- /dev/null +++ b/apps/api/src/queue-retry-limits.ts @@ -0,0 +1,8 @@ +/** + * Retry limits a queue consumer's settings and its batch handler must agree on: Cloudflare stops + * redelivering after `maxRetries` without saying so, and the handler logs the drop on that attempt. + * Dependency-free, so `worker/consumers.ts` reads them without loading a runtime module. + */ +export const AUDIT_EVENTS_MAX_RETRIES = 5 + +export const CANCELLATION_REVIEW_MAX_RETRIES = 5 diff --git a/apps/api/src/resources/env.ts b/apps/api/src/resources/env.ts index fe5110bc1a..64650ad2ca 100644 --- a/apps/api/src/resources/env.ts +++ b/apps/api/src/resources/env.ts @@ -9,6 +9,7 @@ import { apnsEnv, appUrlsEnv, authEnv, + clickhouseEnv, cloudflareOAuthEnv, githubAppSourceEnv, ingestKeyCryptoEnv, @@ -28,11 +29,7 @@ export const apiConfiguredEnv = (stage: MapleStage, region: MapleRegion, domains merge( tinybirdEnv, // ClickHouse (BYO warehouse); `tinybird` unless an org config overrides it. - optionalPlain("CLICKHOUSE_URL"), - plainWithDefault("CLICKHOUSE_PROVIDER", "tinybird"), - optionalPlain("CLICKHOUSE_USER"), - optionalPlain("CLICKHOUSE_DATABASE"), - optionalSecret("CLICKHOUSE_PASSWORD"), + clickhouseEnv, // Dev-only; the runtime ignores it outside MAPLE_ENVIRONMENT=development. optionalPlain("MAPLE_IGNORE_ORG_CLICKHOUSE"), // Dev only: on a deploy a pin would point the whole API at one tenant. The local stack's diff --git a/apps/api/src/worker/consumers.ts b/apps/api/src/worker/consumers.ts index e47cf5a780..7846c68d51 100644 --- a/apps/api/src/worker/consumers.ts +++ b/apps/api/src/worker/consumers.ts @@ -5,7 +5,6 @@ * layer graph and one Postgres socket for the batch. */ import * as Cloudflare from "@alchemy.run/cloudflare" -import { renamedFrom } from "alchemy/Rename" import { Effect, Layer, Stream } from "effect" import { DatabasePgLive } from "@maple/backend/platform/DatabasePgLive" import { AuditEventsDlq, AuditEventsQueue } from "@maple/infra/cloudflare" @@ -17,13 +16,13 @@ import { } from "../resources/queues" import type { ApiPortsLayer } from "./bindings" import type { VcsSyncIsolateCaches } from "@maple/backend/services/integrations/vcs/vcs-sync-isolate-caches" +import { AUDIT_EVENTS_MAX_RETRIES, CANCELLATION_REVIEW_MAX_RETRIES } from "../queue-retry-limits" import { runEvent } from "./events" import { auditEventsModule, cancellationReviewModule, planetScaleWebhookModule } from "./modules" import { consumeVcsSyncBatch } from "./vcs-sync" import { VCS_SYNC_BATCH_SIZE, VCS_SYNC_MAX_RETRIES } from "./vcs-sync-dispatch" -// Consumer settings. The audit consumer's `maxRetries` must stay in sync with -// AUDIT_EVENTS_MAX_RETRIES in audit-events-runtime.ts, which logs the drop on +// Consumer settings. Each `maxRetries` is the constant its batch handler reads to log the drop on // the final attempt. The VCS one reads its constants from the batch route, which caps a request at them. const VCS_SYNC_CONSUMER = { batchSize: VCS_SYNC_BATCH_SIZE, @@ -33,12 +32,11 @@ const VCS_SYNC_CONSUMER = { } satisfies Cloudflare.Queues.MessagesProps const PLANETSCALE_WEBHOOKS_CONSUMER = VCS_SYNC_CONSUMER // A review is up to a minute of upstream reads and one model call: one per -// invocation, few at once. The consumer sets each retry's delay; `maxRetries` -// must match CANCELLATION_REVIEW_MAX_RETRIES in cancellation-review-runtime.ts. +// invocation, few at once. The consumer sets each retry's delay. const CANCELLATION_REVIEWS_CONSUMER = { batchSize: 1, maxConcurrency: 2, - maxRetries: 5, + maxRetries: CANCELLATION_REVIEW_MAX_RETRIES, maxWaitTime: "1 second", } satisfies Cloudflare.Queues.MessagesProps // Audit entries tolerate a few seconds of delivery latency; batch wider and @@ -46,17 +44,14 @@ const CANCELLATION_REVIEWS_CONSUMER = { const auditEventsConsumer = (deadLetterQueue: string | undefined): Cloudflare.Queues.MessagesProps => ({ batchSize: 25, maxConcurrency: 2, - maxRetries: 5, + maxRetries: AUDIT_EVENTS_MAX_RETRIES, maxWaitTime: "5 seconds", deadLetterQueue, }) /** * Attaches the five consumers to the host at plan time and their listeners - * at runtime. Needs `Queues.EventSourceLive`. `renamedFrom` carries the - * consumer resources over from the ids the api factory declared them under, - * so the deploy migrates their state rows instead of re-creating the - * consumers; drop it once every stage has deployed past it. + * at runtime. Needs `Queues.EventSourceLive`. */ export const registerQueueConsumers = ( ports: ApiPortsLayer, @@ -70,7 +65,7 @@ export const registerQueueConsumers = ( Effect.flatMap(Stream.runCollect(stream), (messages) => consumeVcsSyncBatch(ports, env, vcsSyncCaches, messages), ), - ).pipe(renamedFrom({ fqn: "vcs-sync-consumer" })) + ) // Same handler, own consumer: a scheduled sync's backlog drains here without holding up webhooks. const vcsBackfillQueue = yield* VcsBackfillQueue yield* Cloudflare.Queues.consumeQueueMessages(vcsBackfillQueue, VCS_SYNC_CONSUMER, (stream) => @@ -96,7 +91,7 @@ export const registerQueueConsumers = ( ), ), ), - ).pipe(renamedFrom({ fqn: "planetscale-webhooks-consumer" })) + ) const cancellationReviewsQueue = yield* CancellationReviewsQueue yield* Cloudflare.Queues.consumeQueueMessages( cancellationReviewsQueue, @@ -136,5 +131,5 @@ export const registerQueueConsumers = ( ), ), ), - ).pipe(renamedFrom({ fqn: "audit-events-consumer" })) + ) }) diff --git a/apps/chat-bot/src/routes/webhook.ts b/apps/chat-bot/src/routes/webhook.ts index 2b6b076508..35492d4036 100644 --- a/apps/chat-bot/src/routes/webhook.ts +++ b/apps/chat-bot/src/routes/webhook.ts @@ -6,10 +6,8 @@ * verify a signature, parse a body or know a payload: a connector that cannot * authenticate its own caller is a connector that has not been written yet. * - * Nothing registered today delivers this way — the first connector is a socket - * one — so this exists as the contract's other half, tested against a fake - * connector. It is also why the Worker takes no public hostname yet; see - * `../worker.ts`. + * Slack delivers this way (its connector is `kind: "webhook"`), which is why the + * Worker takes a public hostname where the stage has one; see `../worker.ts`. */ import { connectors as registeredConnectors } from "@maple/chat-platform/connectors" import { ConfigProvider, Effect, Option, Schema, Array as Arr } from "effect" diff --git a/apps/electric-sync/src/electric/ElectricClient.test.ts b/apps/electric-sync/src/electric/ElectricClient.test.ts index 41a99eca75..bf5fa96405 100644 --- a/apps/electric-sync/src/electric/ElectricClient.test.ts +++ b/apps/electric-sync/src/electric/ElectricClient.test.ts @@ -199,21 +199,18 @@ describe("buildUpstreamShapeUrl", () => { assert.isNull(params.get("replica")) }) - it.each(["alert_destinations", "alert_destinations_v2"] as const)( - "pins %s's column projection and drops the secret columns", - (shape) => { - const columns = columnsOf(buildUrl(shape).params) - // PK + org scope must be present for Electric. - assert.include(columns, "id") - assert.include(columns, "org_id") - // The public config the browser renders is allowed… - assert.include(columns, "config_json") - // …but the encrypted secret columns must never be projected. - assert.notInclude(columns, "secret_ciphertext") - assert.notInclude(columns, "secret_iv") - assert.notInclude(columns, "secret_tag") - }, - ) + it("pins alert_destinations_v2's column projection and drops the secret columns", () => { + const columns = columnsOf(buildUrl("alert_destinations_v2").params) + // PK + org scope must be present for Electric. + assert.include(columns, "id") + assert.include(columns, "org_id") + // The public config the browser renders is allowed… + assert.include(columns, "config_json") + // …but the encrypted secret columns must never be projected. + assert.notInclude(columns, "secret_ciphertext") + assert.notInclude(columns, "secret_iv") + assert.notInclude(columns, "secret_tag") + }) it("projects only safe API-key display fields", () => { const { params } = buildUrl("api_keys") @@ -230,7 +227,7 @@ describe("buildUpstreamShapeUrl", () => { it("never lets the client widen a column-restricted shape back to secrets", () => { const columns = columnsOf( - buildUrl("alert_destinations", { + buildUrl("alert_destinations_v2", { query: "columns=id%2Corg_id%2Csecret_ciphertext%2Csecret_iv%2Csecret_tag", }).params, ) diff --git a/apps/electric-sync/src/shapes/registry.test.ts b/apps/electric-sync/src/shapes/registry.test.ts index caad0635f6..42c7eaf5ec 100644 --- a/apps/electric-sync/src/shapes/registry.test.ts +++ b/apps/electric-sync/src/shapes/registry.test.ts @@ -12,10 +12,11 @@ describe("isShapeName", () => { it("accepts whitelisted shapes and rejects everything else", () => { assert.isTrue(isSubscriptionName("dashboards")) assert.isTrue(isSubscriptionName("alert_rules")) - assert.isTrue(isSubscriptionName("alert_destinations")) assert.isTrue(isSubscriptionName("alert_destinations_v2")) + // Retired once every client moved to `_v2`. + assert.isFalse(isSubscriptionName("alert_destinations")) assert.isTrue(isSubscriptionName("api_keys")) - // Pruned from both the whitelist and the publication (0022) once their + // Pruned from both the whitelist and the publication (electric_publication_prune) once their // client collections were removed — they must not resolve as shapes. assert.isFalse(isSubscriptionName("error_issues")) assert.isFalse(isSubscriptionName("actors")) diff --git a/apps/electric-sync/src/shapes/registry.ts b/apps/electric-sync/src/shapes/registry.ts index 0771b497c4..2974eddb83 100644 --- a/apps/electric-sync/src/shapes/registry.ts +++ b/apps/electric-sync/src/shapes/registry.ts @@ -19,13 +19,14 @@ import { Schema, Record as Rec } from "effect" // requires the PK in the projection). // // Every table listed here MUST also be a member of `electric_publication_default` -// (packages/db/drizzle/0009 + later publication migrations) — Electric runs with +// (packages/db/drizzle/20260704212005_electric_publication + later publication +// migrations): Electric runs with // ELECTRIC_MANUAL_TABLE_PUBLISHING=true and will not publish a table itself, so a // shape over an unpublished table never receives changes. The reverse also holds: // a published table with no shape here (and no client collection) is pure // replication cost, so drop it from the publication instead of leaving it. That // is why the errors vertical (error_issues / actors / open_error_incidents) and -// scrape_target_checks are gone — see 0022_electric_publication_prune. +// scrape_target_checks are gone (20260727105910_electric_publication_prune). const SUBSCRIPTIONS = { dashboards: { table: "dashboards" }, alert_rules: { table: "alert_rules" }, @@ -58,23 +59,8 @@ const SUBSCRIPTIONS = { // workspace); the encrypted webhook secrets live in separate `secret_*` columns // that MUST NOT reach the browser, so the projection drops them (and the // unused `created_by`/`updated_by`). The PK `id` is required in the projection. - alert_destinations: { - table: "alert_destinations", - columns: [ - "id", - "org_id", - "name", - "type", - "enabled", - "config_json", - "last_tested_at", - "last_test_error", - "created_at", - "updated_at", - ], - }, - // `_v2` adds the auto-disable state, so the browser can say why Maple stopped - // delivering. The v1 shape stays for clients still on the previous build. + // `_v2` because `columns` gained the auto-disable state, so the browser can say + // why Maple stopped delivering. alert_destinations_v2: { table: "alert_destinations", columns: [ diff --git a/apps/scraper/src/ApiClient.ts b/apps/scraper/src/ApiClient.ts index 10b9c6fb99..63eb9bd521 100644 --- a/apps/scraper/src/ApiClient.ts +++ b/apps/scraper/src/ApiClient.ts @@ -41,7 +41,10 @@ const encodeUptimeReports = Schema.encodeSync(UptimeCheckReportList) const decodeJson = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Unknown)) const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown)) -/** Not idempotent: a timeout may follow a batch that landed, so only an explicit refusal retries. */ +/** + * The in-request retry covers an explicit refusal only: a timeout may follow a batch that landed. + * The batch reporter redelivers every transient failure until it lands, so delivery is at-least-once. + */ const isRefusal = (status: number | null): boolean => status === 429 || status === 503 export class ApiClient extends Context.Service()("@maple/scraper/ApiClient", { diff --git a/apps/scraper/src/Env.ts b/apps/scraper/src/Env.ts index 3b2d1ec926..18b2200964 100644 --- a/apps/scraper/src/Env.ts +++ b/apps/scraper/src/Env.ts @@ -34,11 +34,11 @@ export interface ScraperEnvConfig { readonly PORT: number } -// Defaults target the local dev stack (`bun dev`: api on 3472, ingest on -// 3474, the docker-compose dev token) so the scraper boots without extra -// configuration instead of crashing the turbo dev TUI. Production overrides -// all three (apps/scraper/alchemy.run.ts); a missing override -// degrades to visible per-reconcile warnings, never a crash loop. +// Defaults target a standalone run (api on 3472, ingest on 3474, the +// docker-compose dev token) so the scraper boots without extra configuration. +// `bun dev` passes both URLs from its routes (root alchemy.run.ts) and +// production overrides all three (apps/scraper/alchemy.run.ts); a missing +// override degrades to visible per-reconcile warnings, never a crash loop. // // Numeric settings ARE validated at layer build, unlike the URLs: a missing // URL fails loudly every reconcile, but SCRAPER_CONCURRENCY=0 builds a diff --git a/apps/status/src/handler.test.ts b/apps/status/src/handler.test.ts index 5ede6e4d14..2147d0f9b4 100644 --- a/apps/status/src/handler.test.ts +++ b/apps/status/src/handler.test.ts @@ -7,7 +7,7 @@ import { makeSnapshot, } from "@maple/status-page/fixtures" import { beforeEach, describe, expect, it } from "vitest" -import { cacheKey } from "./snapshot" +import { CACHE_FRESH_MS, cacheKey } from "./snapshot" import { type FakeStorage, makeFakeStorage, @@ -111,6 +111,28 @@ describe("serving a page", () => { expect((await send(fake, "https://status.maple.workers.dev/s/unknown/")).status).toBe(404) }) + it("asks KV for the shared host once per cache window, not on every slug request", async () => { + const url = "https://status.maple.workers.dev/s/acme/" + const hostKey = StatusPageKeys.host("status.maple.workers.dev") + const hostReads = () => Arr.filter(fake.kvReads, (key) => key === hostKey).length + expect((await send(fake, url)).status).toBe(200) + expect((await send(fake, url)).status).toBe(200) + expect(hostReads()).toBe(1) + expect((await send(fake, url, { nowMs: FIXTURE_NOW + CACHE_FRESH_MS })).status).toBe(200) + expect(hostReads()).toBe(2) + }) + + it("serves a host that gains a page once its remembered miss has aged out", async () => { + const page = "https://status.late.example/" + expect((await send(fake, page)).status).toBe(404) + fake.kv.set(StatusPageKeys.host("status.late.example"), FIXTURE_PAGE_ID) + expect((await send(fake, page, { nowMs: FIXTURE_NOW + CACHE_FRESH_MS })).status).toBe(200) + }) + + it("keeps a host's own page ahead of the slug path, so it never serves another page", async () => { + expect((await send(fake, "https://status.acme.example/s/acme/")).status).toBe(404) + }) + it("stops serving a renamed page under its old slug", async () => { fake.kv.set(StatusPageKeys.slug("old-acme"), FIXTURE_PAGE_ID) expect((await send(fake, "https://status.maple.workers.dev/s/old-acme/")).status).toBe(404) diff --git a/apps/status/src/snapshot.ts b/apps/status/src/snapshot.ts index f97a86691c..2652141bb9 100644 --- a/apps/status/src/snapshot.ts +++ b/apps/status/src/snapshot.ts @@ -48,9 +48,13 @@ const decodeFound = ( ): Option.Option> => raw === null ? Option.none() : Option.map(decode(raw), (value) => ({ raw, value })) +/** Cached in place of a value when KV answered that the key does not exist. */ +const KV_MISS = "missing" + /** * Cache (fresh) -> KV -> R2 -> cache (any age). An undecodable value is a miss, and a value read - * from KV or R2 refreshes the cache. With `kvMissIsFinal`, a KV `null` ends the search. + * from KV or R2 refreshes the cache. With `kvMissIsFinal`, a KV `null` ends the search, and with + * `rememberMiss` that answer is cached for as long as a hit is. */ const readThrough = ( env: StatusEnv, @@ -59,6 +63,7 @@ const readThrough = ( readonly kvKey: string readonly r2Key: string | null readonly kvMissIsFinal: boolean + readonly rememberMiss: boolean readonly decode: (raw: string) => Option.Option }, ): Effect.Effect> => @@ -66,8 +71,9 @@ const readThrough = ( const { storage } = env const cached = yield* storage.cacheGet(input.cacheKey).pipe(Effect.orElseSucceed(() => null)) const fromCache = decodeFound(cached?.value ?? null, input.decode) - if (cached !== null && Option.isSome(fromCache) && env.nowMs - cached.storedAt < CACHE_FRESH_MS) { - return found(fromCache.value.value) + if (cached !== null && env.nowMs - cached.storedAt < CACHE_FRESH_MS) { + if (Option.isSome(fromCache)) return found(fromCache.value.value) + if (input.rememberMiss && cached.value === KV_MISS) return MISSING } const remember = (hit: Found) => storage @@ -77,7 +83,15 @@ const readThrough = ( const kv = yield* Effect.result(storage.kvGet(input.kvKey)) const fromKv = Result.isSuccess(kv) ? decodeFound(kv.success, input.decode) : Option.none() if (Option.isSome(fromKv)) return yield* remember(fromKv.value) - if (input.kvMissIsFinal && Result.isSuccess(kv) && kv.success === null) return MISSING + if (input.kvMissIsFinal && Result.isSuccess(kv) && kv.success === null) { + // Never over a cached value: that copy is the fallback for when KV and R2 both fail. + if (input.rememberMiss && Option.isNone(fromCache)) { + yield* storage + .cachePut(input.cacheKey, { value: KV_MISS, storedAt: env.nowMs }) + .pipe(Effect.ignore) + } + return MISSING + } const r2 = input.r2Key === null ? null : yield* Effect.result(storage.r2Get(input.r2Key)) const fromR2 = @@ -93,13 +107,17 @@ const decodePageId = (raw: string): Option.Option => { return /^[0-9a-f-]{36}$/u.test(id) ? Option.some(id) : Option.none() } -/** The page a hostname serves: KV `host:`, R2 `hosts/` when KV fails, then any cached answer. */ +/** + * The page a hostname serves: KV `host:`, R2 `hosts/` when KV fails, then any cached answer. + * A miss is remembered: the shared host has no page, and every `/s/` request asks it first. + */ export const pageIdForHost = (env: StatusEnv, hostname: string): Effect.Effect> => readThrough(env, { cacheKey: cacheKey("host", Str.toLowerCase(hostname)), kvKey: StatusPageKeys.host(hostname), r2Key: StatusPageKeys.hostObject(hostname), kvMissIsFinal: true, + rememberMiss: true, decode: decodePageId, }) @@ -110,6 +128,7 @@ export const pageIdForSlug = (env: StatusEnv, slug: string): Effect.Effect (lookup._tag === "found" ? Option.some(lookup.value) : Option.none()))) diff --git a/apps/status/src/test-support.ts b/apps/status/src/test-support.ts index 14c32ebff8..2e562c45df 100644 --- a/apps/status/src/test-support.ts +++ b/apps/status/src/test-support.ts @@ -39,8 +39,14 @@ export const makeFakeStorage = () => { const unlisted = new Set() // Runs once inside the next `list`, to land a concurrent write while it is in flight. const duringList: ListHook = { run: null } + // Every key a KV `get` asked for, in order. + const kvReads = new Array() const storage: StatusStorage = { - kvGet: (key) => guard("kv", key, () => kv.get(key) ?? null), + kvGet: (key) => + guard("kv", key, () => { + kvReads.push(key) + return kv.get(key) ?? null + }), kvPut: (key, value) => guard("kv-put", key, () => void kv.set(key, value)), kvList: (prefix, limit) => guard("kv", prefix, () => { @@ -66,7 +72,7 @@ export const makeFakeStorage = () => { } // Shared across requests, so two posts get distinct entry ids. const ids = { next: 0 } - return { storage, kv, r2, cache, failing, limits, unlisted, duringList, ids } + return { storage, kv, kvReads, r2, cache, failing, limits, unlisted, duringList, ids } } export type FakeStorage = ReturnType diff --git a/docs/electric-sync.md b/docs/electric-sync.md index a693e6b1a3..760196f66b 100644 --- a/docs/electric-sync.md +++ b/docs/electric-sync.md @@ -61,8 +61,7 @@ writes: endpoint captures the Postgres txid on the mutating statement | `alert_rules` | alert_rules | none | | `alert_rule_states` | alert_rule_states | none | | `alert_incidents` | alert_incidents | none | -| `alert_destinations` | alert_destinations | columns: drops the encrypted `secret_*` (previous builds only) | -| `alert_destinations_v2` | alert_destinations | columns: as v1, plus `disabled_at` / `disabled_reason` | +| `alert_destinations_v2` | alert_destinations | columns: drops the encrypted `secret_*` | | `api_keys` | api_keys | columns: drops `key_hash` / `metadata_json` | | `investigation_v2` | investigations | scoped to one `id` (client supplies the value as `$2`); columns | diff --git a/packages/backend/src/platform/EmailService.ts b/packages/backend/src/platform/EmailService.ts index f261841620..b4353e0363 100644 --- a/packages/backend/src/platform/EmailService.ts +++ b/packages/backend/src/platform/EmailService.ts @@ -36,13 +36,9 @@ export class EmailService extends Context.Service const senderOption = yield* Effect.serviceOption(EmailSender) const sender = Option.flatten(senderOption) - // Real sends are production-only: non-prod stages share real user data - // (branched DBs, Clerk members), so a live binding there would deliver - // duplicate copies of every cron-driven email. The alchemy configs no - // longer attach EMAIL outside prd; this guard covers any binding that - // still reaches a non-prod worker (`alchemy dev`, manual deploys). - const emailAllowed = env.MAPLE_ENVIRONMENT === "production" || env.MAPLE_EMAIL_ALLOW_NONPROD - const isConfigured = Option.isSome(sender) && emailAllowed + // Real sends are production-only: non-prod stages share real user data, so a + // binding there would send duplicates. `bindEmailSender` is the one gate. + const isConfigured = Option.isSome(sender) const send = Effect.fn("EmailService.send")(function* ( to: string, @@ -60,12 +56,6 @@ export class EmailService extends Context.Service }) } - if (!emailAllowed) { - return yield* new EmailDeliveryError({ - message: `Email suppressed: sends are disabled in ${env.MAPLE_ENVIRONMENT} (set MAPLE_EMAIL_ALLOW_NONPROD=true to override)`, - }) - } - const result = yield* sender.value .send({ from: fromEmail, to, subject, html, ...options }) .pipe( diff --git a/packages/backend/src/platform/Env.ts b/packages/backend/src/platform/Env.ts index 1571dcb1bc..fb4c3ba431 100644 --- a/packages/backend/src/platform/Env.ts +++ b/packages/backend/src/platform/Env.ts @@ -54,8 +54,6 @@ export interface EnvConfig { readonly MAPLE_ENVIRONMENT: string /** The regional instance this worker belongs to, derived by the stack from the stage (`prd-eu` → `eu`). */ readonly MAPLE_REGION: MapleRegion - /** Escape hatch: allow real email sends outside production (e.g. a deliberate test run on a dev stage). */ - readonly MAPLE_EMAIL_ALLOW_NONPROD: boolean /** Route every org to the managed warehouse; honoured only in development. */ readonly MAPLE_IGNORE_ORG_CLICKHOUSE: boolean readonly CLERK_SECRET_KEY: Option.Option @@ -247,9 +245,6 @@ const envConfig = Config.all({ MAPLE_API_BASE_URL: stringWithDefault("MAPLE_API_BASE_URL", "http://127.0.0.1:3472"), MAPLE_ENVIRONMENT: stringWithDefault("MAPLE_ENVIRONMENT", "development"), MAPLE_REGION: stringWithDefault("MAPLE_REGION", "us").pipe(Config.map(parseMapleRegion)), - MAPLE_EMAIL_ALLOW_NONPROD: stringWithDefault("MAPLE_EMAIL_ALLOW_NONPROD", "false").pipe( - Config.map((value) => value === "true"), - ), MAPLE_IGNORE_ORG_CLICKHOUSE: stringWithDefault("MAPLE_IGNORE_ORG_CLICKHOUSE", "false").pipe( Config.map((value) => value === "1" || value === "true"), ), diff --git a/packages/infra/src/env.ts b/packages/infra/src/env.ts index ecd9442fe6..d3ceb04829 100644 --- a/packages/infra/src/env.ts +++ b/packages/infra/src/env.ts @@ -107,6 +107,18 @@ export const tinybirdEnv: Config.Config = merge( optionalPlain("TINYBIRD_RAW_SQL_JWT_RPS_LIMIT"), ) +/** + * The managed ClickHouse read endpoint; `WarehouseQueryService` reads through it when + * `CLICKHOUSE_URL` is set. Shared by every Worker that queries, so they all read one warehouse. + */ +export const clickhouseEnv: Config.Config = merge( + optionalPlain("CLICKHOUSE_URL"), + plainWithDefault("CLICKHOUSE_PROVIDER", "tinybird"), + optionalPlain("CLICKHOUSE_USER"), + optionalPlain("CLICKHOUSE_DATABASE"), + optionalSecret("CLICKHOUSE_PASSWORD"), +) + /** Ingest-key envelope encryption + lookup HMAC. Required wherever ingest keys are read. */ export const ingestKeyCryptoEnv: Config.Config = merge( requireSecretEntry("MAPLE_INGEST_KEY_ENCRYPTION_KEY"), From 4b71a82b8c7f1c2bb5976250d8b3ff95843bbc85 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Tue, 10 Nov 2026 02:22:22 +0100 Subject: [PATCH 2/4] fix(email): keep the production-only send guard, drop only its unused override The override flag was bound nowhere, but the environment check behind it is the second gate for a binding that reaches a non-prod Worker. --- apps/alerting/src/worker.ts | 3 ++- packages/backend/src/platform/EmailService.ts | 16 +++++++++++++--- 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/apps/alerting/src/worker.ts b/apps/alerting/src/worker.ts index b973d7ec27..893fa3c296 100644 --- a/apps/alerting/src/worker.ts +++ b/apps/alerting/src/worker.ts @@ -62,7 +62,8 @@ const configuredEnv = (stage: MapleStage, region: MapleRegion, domains: MapleDom authEnv, ingestKeyCryptoEnv, appUrlsEnv(domains), - // MAPLE_ENVIRONMENT must stay stage-derived, not overridable: it gates the non-prod cron skip. + // MAPLE_ENVIRONMENT must stay stage-derived, not overridable: it gates both the + // non-prod cron skip and `EmailService.emailAllowed`. selfObservabilityEnv(stage, region), // "1" runs crons on a non-prod stage (which otherwise skips them). optionalPlain("MAPLE_ALERTING_ALLOW_NONPROD"), diff --git a/packages/backend/src/platform/EmailService.ts b/packages/backend/src/platform/EmailService.ts index b4353e0363..325b0fcbf1 100644 --- a/packages/backend/src/platform/EmailService.ts +++ b/packages/backend/src/platform/EmailService.ts @@ -36,9 +36,13 @@ export class EmailService extends Context.Service const senderOption = yield* Effect.serviceOption(EmailSender) const sender = Option.flatten(senderOption) - // Real sends are production-only: non-prod stages share real user data, so a - // binding there would send duplicates. `bindEmailSender` is the one gate. - const isConfigured = Option.isSome(sender) + // Real sends are production-only: non-prod stages share real user data + // (branched DBs, Clerk members), so a live binding there would deliver + // duplicate copies of every cron-driven email. The alchemy configs no + // longer attach EMAIL outside prd; this guard covers any binding that + // still reaches a non-prod worker (`alchemy dev`, manual deploys). + const emailAllowed = env.MAPLE_ENVIRONMENT === "production" + const isConfigured = Option.isSome(sender) && emailAllowed const send = Effect.fn("EmailService.send")(function* ( to: string, @@ -56,6 +60,12 @@ export class EmailService extends Context.Service }) } + if (!emailAllowed) { + return yield* new EmailDeliveryError({ + message: `Email suppressed: sends are disabled in ${env.MAPLE_ENVIRONMENT}`, + }) + } + const result = yield* sender.value .send({ from: fromEmail, to, subject, html, ...options }) .pipe( From 15a826e2273c5f55ee790f1ae86425541c8725c6 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Tue, 10 Nov 2026 04:18:24 +0100 Subject: [PATCH 3/4] fix(chat-bot): keep the webhook route's comment free of vendor names The vendor isolation test guards apps/chat-bot/src, comments included. --- apps/chat-bot/src/routes/webhook.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/chat-bot/src/routes/webhook.ts b/apps/chat-bot/src/routes/webhook.ts index 35492d4036..b69998d623 100644 --- a/apps/chat-bot/src/routes/webhook.ts +++ b/apps/chat-bot/src/routes/webhook.ts @@ -6,8 +6,8 @@ * verify a signature, parse a body or know a payload: a connector that cannot * authenticate its own caller is a connector that has not been written yet. * - * Slack delivers this way (its connector is `kind: "webhook"`), which is why the - * Worker takes a public hostname where the stage has one; see `../worker.ts`. + * A connector of `kind: "webhook"` is delivered this way, which is why the Worker + * takes a public hostname where the stage has one; see `../worker.ts`. */ import { connectors as registeredConnectors } from "@maple/chat-platform/connectors" import { ConfigProvider, Effect, Option, Schema, Array as Arr } from "effect" From 05e9853efb430809f981a12c5673424471b46354 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Tue, 10 Nov 2026 10:43:45 +0100 Subject: [PATCH 4/4] docs(infra): the consumer rename note and the scraper's delivery comment say what holds The queue consumers no longer carry a rename mapping, so the infra doc describes when one is needed instead of saying they have it. The scraper comment promised at-least-once delivery, which a bounded buffer and a timed shutdown flush do not give. --- apps/scraper/src/ApiClient.ts | 3 ++- docs/infra.md | 9 ++++----- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/apps/scraper/src/ApiClient.ts b/apps/scraper/src/ApiClient.ts index 63eb9bd521..26805c13c8 100644 --- a/apps/scraper/src/ApiClient.ts +++ b/apps/scraper/src/ApiClient.ts @@ -43,7 +43,8 @@ const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown)) /** * The in-request retry covers an explicit refusal only: a timeout may follow a batch that landed. - * The batch reporter redelivers every transient failure until it lands, so delivery is at-least-once. + * The batch reporter retries a transiently failed batch, so a batch can land twice. It can also + * be dropped: the reporter's buffer is bounded and its shutdown flush is timed. */ const isRefusal = (status: number | null): boolean => status === 429 || status === 503 diff --git a/docs/infra.md b/docs/infra.md index 8cde419eb7..e33a7d52d1 100644 --- a/docs/infra.md +++ b/docs/infra.md @@ -237,11 +237,10 @@ What each kind of Worker keeps beside the module: - **Queues** (`api`): `Cloudflare.Queues.consumeQueueMessages(queue, settings, handler)` in `impl`, under `Queues.EventSourceLive`, yields the `Consumer` resource at plan time and the listener at runtime (`src/worker/consumers.ts`, over the declarations in - `src/resources/queues.ts`). The consumers carry `renamedFrom({ fqn })` with the ids the - retired api factory declared them under (`vcs-sync-consumer`, …). Alchemy migrates the - state rows, so the deploy plans a noop instead of re-creating each consumer. Without it, - the delete of the old row would have removed the physical consumer the new row had - adopted. Drop the decoration once every stage has deployed past it. + `src/resources/queues.ts`). Renaming a consumer's id needs `renamedFrom({ fqn })` from + `alchemy/Rename` until every stage has deployed past it: alchemy then migrates the state + row and plans a noop. Without it, the delete of the old row removes the physical consumer + the new row had adopted. - **Background telemetry** (`api`): queue batches and cron ticks run under their own SDK instance (`eventTelemetry` in `@maple/infra/worker-telemetry`, provided around the event), so `maple-vcs-sync` and `maple-planetscale-webhooks` keep their own service names.