diff --git a/apps/alerting/alchemy.run.ts b/apps/alerting/alchemy.run.ts index 06a0cc97..9c05df2f 100644 --- a/apps/alerting/alchemy.run.ts +++ b/apps/alerting/alchemy.run.ts @@ -64,9 +64,16 @@ export const createAlertingWorker = ({ stage, mapleDb }: CreateAlertingWorkerOpt // Ref stages attach MAPLE_DB via worker.bind below. ...(mapleDb ? { MAPLE_DB: mapleDb } : {}), AI_TRIAGE_WORKFLOW: aiTriageWorkflow, - EMAIL: Cloudflare.Email.SendEmail("email", { - allowedSenderAddresses: ["notifications@noreply.maple.dev"], - }), + // Production only: preview/stg workers run the same email crons against + // their own DB branches, so a binding here means every live stage sends + // its own copy of onboarding/digest/alert emails to real users. + ...(stage.kind === "prd" + ? { + EMAIL: Cloudflare.Email.SendEmail("email", { + allowedSenderAddresses: ["notifications@noreply.maple.dev"], + }), + } + : {}), TINYBIRD_HOST: requireEnv("TINYBIRD_HOST"), TINYBIRD_TOKEN: Redacted.make(requireEnv("TINYBIRD_TOKEN")), // Alert-rule evaluation runs Tinybird-scoped raw SQL through diff --git a/apps/api/alchemy.run.ts b/apps/api/alchemy.run.ts index 70a7bc13..a4507c8d 100644 --- a/apps/api/alchemy.run.ts +++ b/apps/api/alchemy.run.ts @@ -143,9 +143,16 @@ export const createMapleApi = ({ stage, domains }: CreateMapleApiOptions) => simple: { limit: 600, period: 60 }, }), API_V2_RATE_LIMIT_PARTITION: formatMapleStage(stage), - EMAIL: Cloudflare.Email.SendEmail("email", { - allowedSenderAddresses: ["notifications@noreply.maple.dev"], - }), + // Production only: preview/stg workers run the same email crons against + // their own DB branches, so a binding here means every live stage sends + // its own copy of onboarding/digest/alert emails to real users. + ...(stage.kind === "prd" + ? { + EMAIL: Cloudflare.Email.SendEmail("email", { + allowedSenderAddresses: ["notifications@noreply.maple.dev"], + }), + } + : {}), TINYBIRD_HOST: requireEnv("TINYBIRD_HOST"), TINYBIRD_TOKEN: Redacted.make(requireEnv("TINYBIRD_TOKEN")), ...optionalSecret("TINYBIRD_SIGNING_KEY"), diff --git a/apps/api/src/lib/EmailService.ts b/apps/api/src/lib/EmailService.ts index 8890f4eb..bb964c89 100644 --- a/apps/api/src/lib/EmailService.ts +++ b/apps/api/src/lib/EmailService.ts @@ -47,7 +47,14 @@ export class EmailService extends Context.Service).EMAIL - const isConfigured = binding !== undefined + // Real sends are production-only: preview/stg 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 (wrangler dev, manual deploys). + const emailAllowed = + env.MAPLE_ENVIRONMENT === "production" || env.MAPLE_EMAIL_ALLOW_NONPROD === "true" + const isConfigured = binding !== undefined && emailAllowed const send = Effect.fn("EmailService.send")(function* ( to: string, @@ -67,6 +74,14 @@ export class EmailService extends Context.Service binding.send({ diff --git a/apps/api/src/lib/Env.ts b/apps/api/src/lib/Env.ts index 9a06eff8..26052e0b 100644 --- a/apps/api/src/lib/Env.ts +++ b/apps/api/src/lib/Env.ts @@ -26,6 +26,10 @@ export interface EnvShape { readonly MAPLE_INGEST_KEY_LOOKUP_HMAC_KEY: Redacted.Redacted readonly MAPLE_INGEST_PUBLIC_URL: string readonly MAPLE_APP_BASE_URL: string + /** Deployment environment (`production`, `staging`, `pr-`, `development`) — set by alchemy from the stage. */ + readonly MAPLE_ENVIRONMENT: string + /** Escape hatch: allow real email sends outside production (e.g. a dedicated stg test run). */ + readonly MAPLE_EMAIL_ALLOW_NONPROD: string readonly CLERK_SECRET_KEY: Option.Option> readonly CLERK_PUBLISHABLE_KEY: Option.Option readonly CLERK_JWT_KEY: Option.Option> @@ -100,6 +104,8 @@ const envConfig = Config.all({ MAPLE_INGEST_KEY_LOOKUP_HMAC_KEY: Config.redacted("MAPLE_INGEST_KEY_LOOKUP_HMAC_KEY"), MAPLE_INGEST_PUBLIC_URL: stringWithDefault("MAPLE_INGEST_PUBLIC_URL", "http://127.0.0.1:3474"), MAPLE_APP_BASE_URL: stringWithDefault("MAPLE_APP_BASE_URL", "http://127.0.0.1:3471"), + MAPLE_ENVIRONMENT: stringWithDefault("MAPLE_ENVIRONMENT", "development"), + MAPLE_EMAIL_ALLOW_NONPROD: stringWithDefault("MAPLE_EMAIL_ALLOW_NONPROD", "false"), CLERK_SECRET_KEY: optionalRedacted("CLERK_SECRET_KEY"), CLERK_PUBLISHABLE_KEY: optionalString("CLERK_PUBLISHABLE_KEY"), CLERK_JWT_KEY: optionalRedacted("CLERK_JWT_KEY"), diff --git a/apps/api/src/services/AlertDeliveryDispatch.test.ts b/apps/api/src/services/AlertDeliveryDispatch.test.ts index 01ce1b23..e74de140 100644 --- a/apps/api/src/services/AlertDeliveryDispatch.test.ts +++ b/apps/api/src/services/AlertDeliveryDispatch.test.ts @@ -260,4 +260,54 @@ describe("dispatchDelivery", () => { assert.include(error.message, "EMAIL binding is missing") }), ) + + it.effect("email: succeeds with annotation when only some members fail", () => + Effect.gen(function* () { + const sent: string[] = [] + const deps: DispatchDeps = { + sendEmail: (to) => + to === "oncall@acme.test" + ? Effect.fail( + new AlertDeliveryError({ + message: "mailbox unavailable", + destinationType: "email", + }), + ) + : Effect.sync(() => { + sent.push(to) + }), + } + + const result = yield* dispatchDelivery(emailContext, "{}", failingFetch, 5_000, LINK, CHAT, deps) + + assert.deepStrictEqual(sent, ["ops@acme.test"]) + assert.include(result.providerMessage, "Emailed 1 of 2 members") + assert.include(result.providerMessage, "oncall@acme.test") + assert.include(result.providerMessage, "mailbox unavailable") + }), + ) + + it.effect("email: fails with the first member error when every member fails", () => + Effect.gen(function* () { + const deps: DispatchDeps = { + sendEmail: () => + Effect.fail( + new AlertDeliveryError({ + message: "Cloudflare Email send timed out after 15s", + destinationType: "email", + }), + ), + } + + const error = yield* Effect.flip( + dispatchDelivery(emailContext, "{}", failingFetch, 5_000, LINK, CHAT, deps), + ) + + assert.instanceOf(error, AlertDeliveryError) + assert.include(error.message, "failed for all 2 members") + // The verbatim member error must survive aggregation so retryability + // classification (timeout detection) keeps working upstream. + assert.include(error.message, "timed out") + }), + ) }) diff --git a/apps/api/src/services/AlertDeliveryDispatch.ts b/apps/api/src/services/AlertDeliveryDispatch.ts index cc802e81..25d82127 100644 --- a/apps/api/src/services/AlertDeliveryDispatch.ts +++ b/apps/api/src/services/AlertDeliveryDispatch.ts @@ -881,12 +881,49 @@ export const dispatchDelivery = ( email: (config) => Effect.gen(function* () { const { subject, html } = yield* buildAlertEmailContent(context, linkUrl, chatUrl) - yield* Effect.forEach( - config.members, - (member) => deps.sendEmail(member.email, subject, html), - { discard: true }, + const outcomes = yield* Effect.forEach(config.members, (member) => + deps.sendEmail(member.email, subject, html).pipe( + Effect.match({ + onSuccess: () => ({ member, error: null as string | null }), + onFailure: (error) => ({ member, error: error.message }), + }), + ), ) + const failures = outcomes.filter((o) => o.error != null) const count = config.members.length + + if (failures.length === count && count > 0) { + // Nobody received the email — safe to fail retryable; the retry + // re-sends to members who all got nothing. Keep the first failure + // message verbatim so timeout classification survives aggregation. + return yield* Effect.fail( + makeDeliveryError( + `Email delivery failed for all ${count} member${count === 1 ? "" : "s"}: ${failures[0]!.error}`, + "email", + ), + ) + } + + if (failures.length > 0) { + // Partial success is terminal: there is no per-member attempt + // state, so retrying the event would re-email the members who + // already received it. Report success and surface the failures. + yield* Effect.logWarning("Alert email delivered to a subset of members").pipe( + Effect.annotateLogs({ + failedCount: failures.length, + memberCount: count, + firstError: failures[0]!.error, + }), + ) + return { + providerMessage: `Emailed ${count - failures.length} of ${count} members; failed: ${failures + .map((f) => `${f.member.email} (${f.error})`) + .join(", ")}`, + providerReference: null, + responseCode: null, + } as DispatchResult + } + return { providerMessage: `Emailed ${count} member${count === 1 ? "" : "s"}`, providerReference: null, diff --git a/apps/api/src/services/AlertsService.test.ts b/apps/api/src/services/AlertsService.test.ts index a2b215bb..fb6f8449 100644 --- a/apps/api/src/services/AlertsService.test.ts +++ b/apps/api/src/services/AlertsService.test.ts @@ -706,6 +706,161 @@ describe("AlertsService", () => { }).pipe(Effect.provide(makeLayer(testDb, makeWarehouseStub(state), { fetch: okFetch }))) }) + it.effect("suppresses trigger and resolve notifications while an incident flaps", () => { + const testDb = createTestDb(trackedDbs) + const breachedRows = [ + { + count: 200, + avgDuration: 40, + p50Duration: 20, + p95Duration: 120, + p99Duration: 240, + errorRate: 10, + satisfiedCount: 180, + toleratingCount: 10, + apdexScore: 0.925, + }, + ] as ReadonlyArray> + const healthyRows = [ + { + count: 200, + avgDuration: 20, + p50Duration: 10, + p95Duration: 80, + p99Duration: 160, + errorRate: 0.5, + satisfiedCount: 195, + toleratingCount: 3, + apdexScore: 0.9825, + }, + ] as ReadonlyArray> + const state = { tracesAggregateRows: breachedRows } + + return Effect.gen(function* () { + yield* TestClock.setTime(DEFAULT_CLOCK_EPOCH_MS) + const alerts = yield* AlertsService + const orgId = asOrgId("org_flap") + const userId = asUserId("user_flap") + const destination = yield* createWebhookDestination(alerts, orgId, userId) + yield* createErrorRateRule(alerts, orgId, userId, destination.id) + + const tick = Effect.gen(function* () { + yield* alerts.runSchedulerTick() + yield* TestClock.adjust(Duration.minutes(1)) + }) + + // Flap 1: open (trigger delivered) then resolve (resolve delivered). + yield* tick + yield* tick + state.tracesAggregateRows = healthyRows + yield* tick + yield* tick + + // Flap 2 within the renotify interval: incident opens again, but both + // its trigger and its resolve notifications are suppressed. + state.tracesAggregateRows = breachedRows + yield* tick + yield* tick + state.tracesAggregateRows = healthyRows + yield* tick + yield* tick + + const incidents = yield* alerts.listIncidents(orgId) + const events = yield* alerts.listDeliveryEvents(orgId) + + assert.lengthOf(incidents.incidents, 2) + assert.isTrue(incidents.incidents.every((incident) => incident.status === "resolved")) + // Only the first flap emailed: one trigger + one resolve, nothing for flap 2. + assert.deepStrictEqual( + events.events.map((event: { eventType: string }) => event.eventType).sort(), + ["resolve", "trigger"], + ) + }).pipe(Effect.provide(makeLayer(testDb, makeWarehouseStub(state), { fetch: okFetch }))) + }) + + it.effect("queues at most one renotify per interval while deliveries keep failing", () => { + const testDb = createTestDb(trackedDbs) + const state = { + tracesAggregateRows: [ + { + count: 200, + avgDuration: 40, + p50Duration: 20, + p95Duration: 120, + p99Duration: 240, + errorRate: 10, + satisfiedCount: 180, + toleratingCount: 10, + apdexScore: 0.925, + }, + ] as ReadonlyArray>, + } + const failingFetch: typeof fetch = (async () => + new Response("boom", { status: 500 })) as unknown as typeof fetch + + return Effect.gen(function* () { + yield* TestClock.setTime(DEFAULT_CLOCK_EPOCH_MS) + const alerts = yield* AlertsService + const orgId = asOrgId("org_renotify_gate") + const userId = asUserId("user_renotify_gate") + const destination = yield* createWebhookDestination(alerts, orgId, userId) + yield* alerts.createRule( + orgId, + userId, + adminRoles, + new AlertRuleUpsertRequest({ + name: "Checkout error rate", + severity: "critical", + enabled: true, + serviceNames: ["checkout"], + signalType: "error_rate", + comparator: "gt", + threshold: 5, + windowMinutes: 5, + minimumSampleCount: 10, + consecutiveBreachesRequired: 2, + consecutiveHealthyRequired: 2, + renotifyIntervalMinutes: 5, + destinationIds: [destination.id], + }), + ) + + // Trigger opens on the second tick; every delivery attempt fails with a + // retryable 500 for the whole test, so lastNotifiedAt can only advance + // via the queue-time gate. + for (let minute = 0; minute < 9; minute += 1) { + yield* alerts.runSchedulerTick() + yield* TestClock.adjust(Duration.minutes(1)) + } + + const events = yield* alerts.listDeliveryEvents(orgId) + const renotifyFirstAttempts = events.events.filter( + (event: { eventType: string; attemptNumber: number }) => + event.eventType === "renotify" && event.attemptNumber === 1, + ) + // Trigger at t=1min → renotify due at t=6min. Ticks at 7/8 min must NOT + // re-queue a fresh renotify chain even though nothing was delivered. + assert.lengthOf(renotifyFirstAttempts, 1) + + const incidents = yield* alerts.listIncidents(orgId) + assert.lengthOf(incidents.incidents, 1) + assert.isNotNull(incidents.incidents[0]?.lastNotifiedAt) + + // One interval after the queue-time advance, a second renotify chain starts. + for (let minute = 0; minute < 3; minute += 1) { + yield* alerts.runSchedulerTick() + yield* TestClock.adjust(Duration.minutes(1)) + } + const eventsAfter = yield* alerts.listDeliveryEvents(orgId) + const renotifyChains = new Set( + eventsAfter.events + .filter((event: { eventType: string }) => event.eventType === "renotify") + .map((event: { deliveryKey: string }) => event.deliveryKey.split(":").at(-1)), + ) + assert.strictEqual(renotifyChains.size, 2) + }).pipe(Effect.provide(makeLayer(testDb, makeWarehouseStub(state), { fetch: failingFetch }))) + }) + it.effect("skips unchanged alert_rule_states writes and refreshes on the 5-minute heartbeat", () => { const testDb = createTestDb(trackedDbs) // Healthy from the start: errorRate 1 stays below the threshold of 5. diff --git a/apps/api/src/services/AlertsService.ts b/apps/api/src/services/AlertsService.ts index 503982f6..13ec465c 100644 --- a/apps/api/src/services/AlertsService.ts +++ b/apps/api/src/services/AlertsService.ts @@ -85,7 +85,7 @@ import { type AlertRuleRow, alertRuleStates, } from "@maple/db" -import { and, asc, desc, eq, inArray, isNotNull, isNull, lt, lte, ne, or, sql } from "drizzle-orm" +import { and, asc, desc, eq, gte, inArray, isNotNull, isNull, lt, lte, ne, or, sql } from "drizzle-orm" import { Array as Arr, Cause, @@ -3787,6 +3787,36 @@ export class AlertsService extends Context.Service= normalized.consecutiveBreachesRequired ) { + // Flap suppression: a metric oscillating around the threshold opens + // a fresh incident per flap, which would email an identical trigger + // notification every few minutes. If the previous incident for this + // (rule, group) was notified within the renotify interval, open the + // incident but skip the trigger notification and carry the prior + // lastNotifiedAt forward — the renotify gate then enforces one + // email per interval while the flapping persists. + const priorNotified = + (yield* dbExecute((db) => + db + .select({ lastNotifiedAt: alertIncidents.lastNotifiedAt }) + .from(alertIncidents) + .where( + and( + eq(alertIncidents.orgId, row.orgId), + eq(alertIncidents.ruleId, row.id), + eq(alertIncidents.groupKey, groupKey), + eq(alertIncidents.status, "resolved"), + isNotNull(alertIncidents.lastNotifiedAt), + gte( + alertIncidents.lastNotifiedAt, + new Date(timestamp - normalized.renotifyIntervalMinutes * 60_000), + ), + ), + ) + .orderBy(desc(alertIncidents.lastNotifiedAt)) + .limit(1), + ))[0] ?? null + const flapSuppressedAt = priorNotified?.lastNotifiedAt ?? null + const incidentId = decodeAlertIncidentIdSync(makeUuid()) const incident: AlertIncidentRow = { id: incidentId, @@ -3809,21 +3839,35 @@ export class AlertsService extends Context.Service db.insert(alertIncidents).values(incident)) - yield* queueIncidentNotifications( - row.orgId, - normalized, - incident, - evaluation, - "trigger", - timestamp, - ) + if (flapSuppressedAt != null) { + yield* Effect.logInfo("Skipping trigger notification for flapping incident").pipe( + Effect.annotateLogs({ + ruleId: row.id, + incidentId, + groupKey, + priorNotifiedAt: flapSuppressedAt.toISOString(), + }), + ) + } else { + yield* queueIncidentNotifications( + row.orgId, + normalized, + incident, + evaluation, + "trigger", + timestamp, + ) + } return { transition: "opened" as const, incidentId, @@ -3843,6 +3887,18 @@ export class AlertsService extends Context.Service db .update(alertIncidents) @@ -3852,14 +3908,12 @@ export class AlertsService extends Context.Service void, edgeBackend?: ReturnType, fingerprintRows?: () => ReadonlyArray>, + dispatcher?: (typeof NotificationDispatcher)["Service"], ) => { const testDb = createTestDb(createdDbs) const envLive = Env.layer.pipe(Layer.provide(testConfig())) const databaseLive = testDb.layer - const dispatcherStub = Layer.succeed(NotificationDispatcher, { - dispatch: () => Effect.succeed({ delivered: 0, failed: 0 }), - }) + const dispatcherStub = Layer.succeed( + NotificationDispatcher, + dispatcher ?? { + dispatch: () => Effect.succeed({ delivered: 0, failed: 0 }), + }, + ) return ErrorsService.layer.pipe( Layer.provide( Layer.mergeAll( @@ -938,6 +943,91 @@ describe("ErrorsService.runTick", () => { }).pipe(Effect.provide(makeErrorsLayer(() => rows))) }) + it.effect("overlapping ticks dispatch each incident notification exactly once", () => { + let rows: ReadonlyArray> = [scanRow()] + const dispatched: string[] = [] + const countingDispatcher: (typeof NotificationDispatcher)["Service"] = { + dispatch: (_orgId, _destinationIds, context) => + Effect.sync(() => { + dispatched.push(context.deliveryKey) + return { delivered: 1, failed: 0 } + }), + } + return Effect.gen(function* () { + const errors = yield* ErrorsService + const database = yield* Database + yield* TestClock.setTime(TICK_MS) + yield* seedIssue(asIssueId(randomUUID())) + // Enabled policy with a destination so incident open/resolve actually + // dispatches (the dispatcher itself is stubbed — no destination row needed). + yield* database.execute((db) => + db.insert(errorNotificationPolicies).values({ + orgId: ORG, + enabled: true, + destinationIdsJson: ["7d31c9e1-0000-4000-8000-000000000001"], + notifyOnResolve: true, + updatedAt: new Date(TICK_MS), + updatedBy: "test", + }), + ) + + yield* errors.runTick() + assert.lengthOf( + dispatched.filter((key) => key.endsWith(":open")), + 1, + ) + + // Stale window elapsed, no fresh errors: two OVERLAPPING ticks race the + // open→resolved flip. The CAS lets exactly one dispatch the resolve. + rows = [] + yield* TestClock.setTime(TICK_MS + 31 * 60_000) + const resolveResults = yield* Effect.all([errors.runTick(), errors.runTick()], { + concurrency: 2, + }) + assert.strictEqual( + resolveResults.reduce((s, r) => s + r.incidentsResolved, 0), + 1, + ) + assert.lengthOf( + dispatched.filter((key) => key.endsWith(":resolve")), + 1, + ) + + // Errors return: two OVERLAPPING ticks race the reopen. The state-row + // CAS lets exactly one insert the incident and dispatch its open. + const reopenMs = TICK_MS + 32 * 60_000 + rows = [ + scanRow({ + firstSeen: toWarehouseDateTime(reopenMs - 60_000), + lastSeen: toWarehouseDateTime(reopenMs - 1_000), + }), + ] + yield* TestClock.setTime(reopenMs) + const reopenResults = yield* Effect.all([errors.runTick(), errors.runTick()], { + concurrency: 2, + }) + assert.strictEqual( + reopenResults.reduce((s, r) => s + r.incidentsOpened, 0), + 1, + ) + assert.lengthOf( + dispatched.filter((key) => key.endsWith(":open")), + 2, + ) + + const issue = (yield* loadIssuesByFingerprint(SCAN_FINGERPRINT))[0]! + const incidents = yield* loadIncidentsForIssue(issue.id) + assert.lengthOf(incidents, 2) + const states = yield* database.execute((db) => + db.select().from(errorIssueStates).where(eq(errorIssueStates.issueId, issue.id)), + ) + assert.strictEqual( + states[0]?.openIncidentId, + incidents.find((incident) => incident.status === "open")?.id, + ) + }).pipe(Effect.provide(makeErrorsLayer(() => rows, undefined, undefined, undefined, countingDispatcher))) + }) + it.effect( "a re-fired fingerprint on a done issue reopens it to triage with a regression incident", () => { diff --git a/apps/api/src/services/ErrorsService.ts b/apps/api/src/services/ErrorsService.ts index e0f244c8..0b87c9e1 100644 --- a/apps/api/src/services/ErrorsService.ts +++ b/apps/api/src/services/ErrorsService.ts @@ -2482,23 +2482,14 @@ const make: Effect.Effect< ? "regression" : "first_seen" const incidentId = newErrorIncidentId() - yield* dbExecute((db) => - db.insert(errorIncidents).values({ - id: incidentId, - orgId, - issueId, - status: "open", - reason, - firstTriggeredAt: new Date(firstSeenMs), - lastTriggeredAt: new Date(lastSeenMs), - resolvedAt: null, - occurrenceCount: row.count, - createdAt: new Date(windowEndMs), - updatedAt: new Date(windowEndMs), - }), - ) - yield* dbExecute((db) => + // CAS the open slot BEFORE creating the incident or dispatching: + // overlapping ticks can both read openIncidentId == null, and the + // notify path below dispatches immediately (no outbox), so only the + // upsert winner may proceed. setWhere keeps the conflict-update a + // no-op when another tick already claimed the slot; RETURNING then + // yields zero rows for the loser. + const claimed = yield* dbExecute((db) => db .insert(errorIssueStates) .values({ @@ -2517,7 +2508,37 @@ const make: Effect.Effect< openIncidentId: incidentId, updatedAt: new Date(windowEndMs), }, - }), + setWhere: isNull(errorIssueStates.openIncidentId), + }) + .returning({ openIncidentId: errorIssueStates.openIncidentId }), + ) + const wonOpenSlot = claimed[0]?.openIncidentId === incidentId + + if (!wonOpenSlot) { + // Lost the race: a concurrent tick opened the incident for this + // same scan window and already dispatched its notification. Do + // not bump occurrence counts here — the winner counted this + // window's occurrences on insert. + yield* Effect.logInfo("Skipping duplicate error incident open (lost CAS)").pipe( + Effect.annotateLogs({ orgId, issueId }), + ) + return { touched: 1, opened: 0 } + } + + yield* dbExecute((db) => + db.insert(errorIncidents).values({ + id: incidentId, + orgId, + issueId, + status: "open", + reason, + firstTriggeredAt: new Date(firstSeenMs), + lastTriggeredAt: new Date(lastSeenMs), + resolvedAt: null, + occurrenceCount: row.count, + createdAt: new Date(windowEndMs), + updatedAt: new Date(windowEndMs), + }), ) yield* notifyIncidentOpened(orgId, policy, { @@ -2599,9 +2620,12 @@ const make: Effect.Effect< ), ), ) - yield* Effect.forEach(staleIncidents, (incident) => + const resolveOutcomes = yield* Effect.forEach(staleIncidents, (incident) => Effect.gen(function* () { - yield* dbExecute((db) => + // CAS the status flip: overlapping ticks both list the incident as + // stale, and the resolve notification dispatches immediately — only + // the tick that wins the open→resolved transition may notify. + const flipped = yield* dbExecute((db) => db .update(errorIncidents) .set({ @@ -2609,8 +2633,12 @@ const make: Effect.Effect< resolvedAt: new Date(windowEndMs), updatedAt: new Date(windowEndMs), }) - .where(eq(errorIncidents.id, incident.id)), + .where(and(eq(errorIncidents.id, incident.id), eq(errorIncidents.status, "open"))) + .returning({ id: errorIncidents.id }), ) + if (flipped.length === 0) { + return { resolved: 0 } + } yield* dbExecute((db) => db .update(errorIssueStates) @@ -2645,9 +2673,10 @@ const make: Effect.Effect< }) } } + return { resolved: 1 } }), ) - const incidentsResolved = staleIncidents.length + const incidentsResolved = resolveOutcomes.reduce((s, r) => s + r.resolved, 0) let issuesArchived = 0 let issuesDeleted = 0 @@ -2758,6 +2787,10 @@ const make: Effect.Effect< } }) + // Overlapping ticks are tolerated (there is no per-org claim lock): scan + // bookkeeping may repeat under overlap, but incident open/resolve + // transitions — and the notifications they dispatch — are CAS-guarded in + // processOrg, so users never receive duplicate incident emails. const runTick: ErrorsServiceShape["runTick"] = Effect.fn("ErrorsService.runTick")(function* () { const endMs = yield* Clock.currentTimeMillis const startMs = endMs - TICK_WINDOW_MS