diff --git a/.github/workflows/loops-safety.yml b/.github/workflows/loops-safety.yml index 006420afe3..27f24d5c3f 100644 --- a/.github/workflows/loops-safety.yml +++ b/.github/workflows/loops-safety.yml @@ -24,7 +24,7 @@ jobs: - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2.2.0 with: bun-version: "1.4.0" - - name: Hold lifecycle delivery if synchronization is unhealthy + - name: Check lifecycle delivery and report required manual pauses env: LOOPS_API_KEY: ${{ secrets.LOOPS_API_KEY }} LOOPS_HEALTH_SECRET: ${{ secrets.LOOPS_HEALTH_SECRET }} diff --git a/packages/database/loops/lifecycle.ts b/packages/database/loops/lifecycle.ts index 5eb5d0fe16..5e08a99975 100644 --- a/packages/database/loops/lifecycle.ts +++ b/packages/database/loops/lifecycle.ts @@ -22,6 +22,19 @@ export type LoopsRuntimeConfig = { teammateJoinedAt?: string | null; }; +export function enrollmentWindow(signupAt: string, config: LoopsRuntimeConfig) { + return { + recentSignup: + config.enrollmentEnabled && + Date.parse(signupAt) >= config.enrollmentAfter.getTime(), + recentJoin: + config.enrollmentEnabled && + Boolean(config.teammateJoinedAt) && + Date.parse(isoDate(config.teammateJoinedAt ?? "")) >= + config.enrollmentAfter.getTime(), + }; +} + export function lifecycleUpdate( source: LoopsProfileSource, remote: LifecycleContact, @@ -42,20 +55,15 @@ export function lifecycleUpdate( !["unsubscribed", "suppressed"].includes(remote.capConsent ?? ""); const subscribed = globallySubscribed && remote.mailingLists[config.listId] === true; - const recentSignup = - Date.parse(profile.capSignupAt) >= config.enrollmentAfter.getTime(); + const enrollment = enrollmentWindow(profile.capSignupAt, config); const imported = Boolean(remote.capImportedAt); - const recentJoin = - teammate && - Boolean(config.teammateJoinedAt) && - Date.parse(isoDate(config.teammateJoinedAt ?? "")) >= - config.enrollmentAfter.getTime(); const eligible = config.enrollmentEnabled && source.signedUp && subscribed && !source.pendingInvite && - ((!imported && recentSignup) || recentJoin) && + ((!imported && enrollment.recentSignup) || + (teammate && enrollment.recentJoin)) && audience !== "unknown"; const { subscribed: _subscribed, diff --git a/packages/database/loops/worker.ts b/packages/database/loops/worker.ts index 689bc62165..0f8864ef90 100644 --- a/packages/database/loops/worker.ts +++ b/packages/database/loops/worker.ts @@ -12,6 +12,7 @@ import { LoopsRequestError, } from "./client"; import { + enrollmentWindow, type LifecycleContact, type LoopsRuntimeConfig, lifecycleUpdate, @@ -263,8 +264,10 @@ export async function runLoopsSync(customerCopy: CustomerCopy) { profile: profileFingerprint(localProfile), signedUp: source.signedUp, pendingInvite: source.pendingInvite, - enrollmentEnabled: config.enrollmentEnabled, - enrollmentAfter: config.enrollmentAfter.toISOString(), + enrollment: enrollmentWindow( + localProfile.capSignupAt, + jobConfig, + ), listId: config.listId, teammateJoinedAt: job.teammateJoinedAt, }), diff --git a/scripts/loops/README.md b/scripts/loops/README.md index 9613cb1302..f2345e3c3d 100644 --- a/scripts/loops/README.md +++ b/scripts/loops/README.md @@ -80,13 +80,17 @@ Deploy the generated schema before code that selects the new columns. The unship With the schema available and test configuration set, `bun scripts/loops/seed.ts` reports how many completed signups would be queued. `--apply` queues them without sending anything. `bun scripts/loops/sync.ts --apply` processes a bounded batch using the same worker as the cron route. Test mode fails closed without an allowlist. Set `LOOPS_SYNC_MODE=production` only as part of the reviewed cutover. Seed old completed accounts once after the final suppression reconciliation; subsequent signups enter the queue through the application. +Sync fingerprints include the contact's effective signup and teammate-join eligibility, rather than the raw enrollment switch or cutoff date. Enabling enrollment or moving a future cutoff does not invalidate every historical profile. A signup or join crossing the cutoff, or disabling an eligible contact's enrollment, still changes its fingerprint and requires synchronization. + ## Independent delivery check `.github/workflows/loops-safety.yml` runs every five minutes, independently of the Cap cron worker. It is disabled until the repository variable `LOOPS_WATCHDOG_ENABLED` is explicitly set to `true`. Configure GitHub secrets `LOOPS_API_KEY` and `LOOPS_HEALTH_SECRET`, and configure the same health secret in Cap. Scheduled Actions run from the default branch, so this protection is not deployed while the PR remains unmerged. The authenticated, read-only `/api/cron/sync-loops/health` endpoint reports queue counts without contact data. It reports unhealthy when production sync or enrollment is disabled, the queue is empty, a due job is over five minutes late, or a job has failed three times. Expected holds for incomplete signup and removed/changed identities do not count as delivery failures. An unchanged successful refresh clears old failures; a newly queued change does not inherit an older attempt's retry delay. -The checker rejects unavailable, malformed or stale health responses. On failure or audience drift it adds mutually exclusive subscription conditions to each journey's first audience guard, with scope set to all following nodes, then reads the guards back. One failed update does not stop attempts on the other journeys. An update failure or existing hold fails the Action so it remains visible. Assign an owner to GitHub Actions failure notifications and verify that notification before activation. +The checker rejects unavailable, malformed or stale health responses. For draft or paused journeys, it can add mutually exclusive subscription conditions to the first audience guard, with scope set to all following nodes, then read the guards back. One failed update does not stop attempts on the other journeys. An update failure or existing hold fails the Action so it remains visible. + +For a running journey, Loops rejects audience edits with HTTP 400: "This operation is not allowed while the workflow is sending." The checker reports `manual-pause-required`, fails the Action, and explicitly warns that delivery has not stopped. Pause the affected workflows in the Loops UI before applying a hold or changing their guards. This monitor does not automatically stop active delivery. Assign an owner to GitHub Actions failure notifications and verify that notification before activation; do not launch on an assumption of automatic pausing. The hold persists after recovery. It does not change workflow status, contact subscriptions or campaign schedules. The current four production drafts have this hold applied and independently verified. To inspect them without writes: @@ -94,7 +98,7 @@ The hold persists after recovery. It does not change workflow status, contact su bun run emails:check-loops --structure-only --require-held ``` -For an emergency hold, with the Loops API key in the environment: +For an emergency, first pause active workflows in the Loops UI. Then, with the Loops API key in the environment, apply persistent audience holds to the stopped workflows: ```sh bun scripts/loops/watchdog.ts --hold --apply @@ -108,7 +112,7 @@ bun scripts/loops/watchdog.ts --resume --apply Resume requires a fresh authenticated health response and refuses changed audience rules. It does not start a draft or unpause a workflow. Contacts that have already exited because of a hold are not automatically replayed; assess recovery separately without bulk re-enrolling history. -GitHub scheduling can be delayed, and a Loops API outage can prevent guard updates. A message already being sent may still arrive. Test a running owned-account journey across an outage in the deployed environment before relying on this protection. Manually pause Loops if the check cannot apply a hold. Campaigns require a fresh health check before scheduling and manual cancellation or pause during an outage; the checker only guards the four registered journeys. +GitHub scheduling can be delayed, and a Loops API outage can prevent even stopped-workflow guard updates. Running workflows keep sending until they are manually paused; a message already being sent may still arrive. Verify the alert and manual-response procedure with an owned-account journey before relying on it. Campaigns require a fresh health check before scheduling and manual cancellation or pause during an outage; the checker only inspects the four registered journeys. ## Before any activation @@ -125,7 +129,7 @@ Activation is deliberately outside this migration's approved scope. These gates 1. Review the final custom drafts and audiences, and configure rotated credentials in the intended environment. September 11 controlled deliveries passed SPF, DKIM and DMARC with inherited `p=quarantine`; the earlier DMARC warning is no longer an observed blocker. Recheck sending-domain status at cutover. 2. The native Stripe connection passed owned live-customer creation/update tests, including name sync, list assignment, global unsubscribe preservation and a recipient Preference Center mailing-list opt-out surviving a later native update. Import/reconcile the remaining Bento negative records with workflows off. Then deploy the reviewed schema/code in test mode and prove the actual signup, purchase and invite routes with owned accounts. 3. Set a future `LOOPS_ENROLLMENT_AFTER` timestamp, verify the cron schedule and permissions in the deployed environment, and seed the completed-account sync jobs. Keep enrollment disabled while inspecting the resulting contacts. Preserve existing opt-outs and suppressions; no separate consent-capture step is needed. -4. Configure and enable the independent delivery check above, verify its deployed health endpoint and failure notifications, and prove purchase/invite/opt-out transitions remove contacts before later promotional steps. Exercise its hold and explicit recovery with an owned-account journey. Downstream filters use the last synced fields; `capVerifiedAt` is not a native expiry guarantee. Pause workflows manually if the checker cannot reach Loops, and handle scheduled campaigns separately. +4. Configure and enable the independent delivery check above, verify its deployed health endpoint and failure notifications, and prove purchase/invite/opt-out transitions remove contacts before later promotional steps. Exercise failure detection, manual pause, hold and explicit recovery with an owned-account journey. The September 11 live test confirmed that the API cannot change guards while Sending; automatic stopping remains unresolved. Approve an operational plan with manual pausing, or keep delivery off until a supported automatic control is available. Downstream filters use the last synced fields; `capVerifiedAt` is not a native expiry guarantee. Handle scheduled campaigns separately. 5. Reconcile a fresh Bento delta at cutover, including all opt-outs and changed entitlements. Confirm no campaign/flow is queued to send twice, check overlap with Resend recording emails, then disable old Bento marketing automations only as part of the approved cutover. Preserve source history and suppression evidence. 6. Test the deployed Cap signup/purchase/invite/opt-out path through Loops to an owned inbox, including a sync outage and retries. The completed synthetic profile-to-Loops tests do not replace this production integration check. 7. After explicit activation approval, choose a future enrollment boundary with enough setup time. While that boundary is still in the future, enable production sync/enrollment and the independent delivery check, confirm healthy queue results, explicitly remove the delivery holds, and start the reviewed Loops workflows. Complete those steps before the boundary so the first eligible signup is processed by an already-running workflow; if setup overruns, move the boundary forward before allowing enrollment. This release admits new signups after the boundary, not a numerically limited cohort. Monitor deliveries, complaints, opt-outs and duplicate suppression. Do not bulk enroll imported history or replay stage changes emitted while workflows were drafts or held. Rollback stops new enrollment and pauses Loops before considering re-enabling Bento; never run both senders for the same journey. Retire Bento and rotate remaining credentials after reconciliation. diff --git a/scripts/loops/lifecycle.test.ts b/scripts/loops/lifecycle.test.ts index d1363a805f..371dd80521 100644 --- a/scripts/loops/lifecycle.test.ts +++ b/scripts/loops/lifecycle.test.ts @@ -2,6 +2,7 @@ import { describe, expect, test } from "bun:test"; import { customerCopy } from "../../emails/customer-copy"; import type { LoopsClient } from "../../packages/database/loops/client"; import { + enrollmentWindow, type LifecycleContact, type LoopsRuntimeConfig, lifecycleUpdate, @@ -21,6 +22,53 @@ const config: LoopsRuntimeConfig = { allowedEmails: new Set([email]), }; +describe("enrollment fingerprint", () => { + test("historical accounts stay unchanged when enrollment is enabled or the future cutoff moves", () => { + const signup = "2026-09-10T10:00:00.000Z"; + const held = enrollmentWindow(signup, { + ...config, + enrollmentEnabled: false, + }); + expect(held).toEqual({ recentSignup: false, recentJoin: false }); + expect(enrollmentWindow(signup, config)).toEqual(held); + expect( + enrollmentWindow(signup, { + ...config, + enrollmentAfter: new Date("2026-09-12T00:00:00Z"), + }), + ).toEqual(held); + }); + + test("eligible signups change when enrollment is disabled or the cutoff crosses their signup", () => { + const signup = "2026-09-11T10:00:00.000Z"; + expect(enrollmentWindow(signup, config)).toEqual({ + recentSignup: true, + recentJoin: false, + }); + expect( + enrollmentWindow(signup, { ...config, enrollmentEnabled: false }), + ).toEqual({ recentSignup: false, recentJoin: false }); + expect( + enrollmentWindow(signup, { + ...config, + enrollmentAfter: new Date("2026-09-11T10:00:00.001Z"), + }), + ).toEqual({ recentSignup: false, recentJoin: false }); + }); + + test("new teammate joins change historical accounts at the exact cutoff", () => { + const signup = "2026-09-10T10:00:00.000Z"; + const joined = { ...config, teammateJoinedAt: "2026-09-11 00:00:00" }; + expect(enrollmentWindow(signup, joined)).toEqual({ + recentSignup: false, + recentJoin: true, + }); + expect( + enrollmentWindow(signup, { ...joined, enrollmentEnabled: false }), + ).toEqual({ recentSignup: false, recentJoin: false }); + }); +}); + function fixture(): LoopsProfileSource { return { input: { diff --git a/scripts/loops/watchdog.test.ts b/scripts/loops/watchdog.test.ts index 9a088f6f68..4350caf7c3 100644 --- a/scripts/loops/watchdog.test.ts +++ b/scripts/loops/watchdog.test.ts @@ -3,7 +3,7 @@ import { heldWorkflowAudience, workflowAudience, } from "../../emails/delivery-safety"; -import type { LoopsApi } from "./api"; +import { type LoopsApi, LoopsApiError } from "./api"; import { checkSyncHealth, enforceDeliverySafety, @@ -12,7 +12,7 @@ import { safetyTargets, } from "./watchdog"; -function fakeProvider() { +function fakeProvider(status = "Draft") { const guards = new Map( safetyTargets.map((target) => [ target.workflowId, @@ -25,7 +25,11 @@ function fakeProvider() { ]), ); const writes: string[] = []; + const attemptedWrites: string[] = []; let brokenWorkflow: string | undefined; + let startedWorkflow: string | undefined; + let startedOnWrite = false; + let startError: Error | undefined; const api: Pick = { async request(path: string, method = "GET", body?: unknown): Promise { if (path === "api-key") return { teamName: "Cap Software, Inc." } as T; @@ -37,6 +41,16 @@ function fakeProvider() { const guard = guards.get(target.workflowId); if (!guard) throw new Error("Unknown guard"); if (method === "POST") { + attemptedWrites.push(target.workflowId); + if (target.workflowId === startedWorkflow) { + startedOnWrite = true; + if (startError) throw startError; + } + if (status === "Sending" || target.workflowId === startedWorkflow) + throw new LoopsApiError(400, path, { + message: + "This operation is not allowed while the workflow is sending.", + }); if (target.workflowId === brokenWorkflow) throw new Error("Unavailable"); const update = body as { @@ -53,6 +67,10 @@ function fakeProvider() { } return { name: target.journey.name, + status: + target.workflowId === startedWorkflow && startedOnWrite + ? "Sending" + : status, mailingListId: target.mailingListId, rootNodeId: "trigger", nodes: { trigger: { nextNodeIds: [target.guardId] } }, @@ -63,6 +81,11 @@ function fakeProvider() { api, guards, writes, + attemptedWrites, + startDuringWrite: (id: string, error?: Error) => { + startedWorkflow = id; + startError = error; + }, fail: (id: string) => { brokenWorkflow = id; }, @@ -89,6 +112,22 @@ describe("independent delivery safety", () => { expect(results.every((result) => result.action === "held")).toBe(true); }); + test("concurrent activation is detected even when the provider error format changes", async () => { + for (const error of [ + new LoopsApiError(400, "nodes/guard", { error: "Workflow is active" }), + new SyntaxError("Unexpected response encoding"), + ]) { + const provider = fakeProvider(); + provider.startDuringWrite(safetyTargets[0].workflowId, error); + const result = await enforceDeliverySafety(provider.api, { + healthy: false, + apply: true, + }); + expect(result[0].action).toBe("manual-pause-required"); + expect(provider.attemptedWrites).toHaveLength(4); + } + }); + test("healthy operation never rewrites audiences", async () => { const provider = fakeProvider(); const result = await enforceDeliverySafety(provider.api, { @@ -99,7 +138,7 @@ describe("independent delivery safety", () => { expect(provider.writes).toHaveLength(0); }); - test("an outage blocks every downstream journey and recovery never resumes it automatically", async () => { + test("an outage holds stopped journeys and recovery never resumes them automatically", async () => { const provider = fakeProvider(); const result = await enforceDeliverySafety(provider.api, { healthy: false, @@ -123,6 +162,85 @@ describe("independent delivery safety", () => { expect(provider.writes).toHaveLength(4); }); + test("an unhealthy running journey requires manual pause without attempting forbidden edits", async () => { + const provider = fakeProvider("Sending"); + const result = await enforceDeliverySafety(provider.api, { + healthy: false, + apply: true, + }); + expect( + result.every((item) => item.action === "manual-pause-required"), + ).toBe(true); + expect(provider.attemptedWrites).toHaveLength(0); + for (const guard of provider.guards.values()) + expect(filterIsHeld(guard.audienceFilter)).toBe(false); + }); + + test("explicit holds on running journeys require manual pause even during a dry run", async () => { + for (const apply of [true, false]) { + const provider = fakeProvider("Sending"); + const result = await enforceDeliverySafety(provider.api, { + healthy: true, + hold: true, + apply, + }); + expect( + result.every((item) => item.action === "manual-pause-required"), + ).toBe(true); + expect(provider.attemptedWrites).toHaveLength(0); + } + }); + + test("a journey starting between read and write still reports the manual pause requirement", async () => { + const provider = fakeProvider(); + provider.startDuringWrite(safetyTargets[0].workflowId); + const result = await enforceDeliverySafety(provider.api, { + healthy: false, + apply: true, + }); + expect(result[0].action).toBe("manual-pause-required"); + expect(result.slice(1).every((item) => item.action === "held")).toBe(true); + }); + + test("unknown workflow states fail without edits", async () => { + const provider = fakeProvider("Unknown"); + const result = await enforceDeliverySafety(provider.api, { + healthy: false, + apply: true, + }); + expect(result.every((item) => item.action === "error")).toBe(true); + expect(provider.attemptedWrites).toHaveLength(0); + }); + + test("healthy running journeys are not rewritten when already resumed", async () => { + const provider = fakeProvider("Sending"); + const result = await enforceDeliverySafety(provider.api, { + healthy: true, + apply: true, + resume: true, + }); + expect(result.every((item) => item.action === "resumed")).toBe(true); + expect(provider.attemptedWrites).toHaveLength(0); + }); + + test("removing a hold requires pausing a running workflow first", async () => { + const provider = fakeProvider("Sending"); + for (const target of safetyTargets) { + const guard = provider.guards.get(target.workflowId); + if (!guard) throw new Error("Missing fixture"); + guard.audienceFilter = heldWorkflowAudience(target.journey); + } + const result = await enforceDeliverySafety(provider.api, { + healthy: true, + apply: true, + resume: true, + }); + expect( + result.every((item) => item.action === "manual-pause-required"), + ).toBe(true); + expect(provider.attemptedWrites).toHaveLength(0); + }); + test("resuming requires explicit apply and a fresh healthy result", async () => { const provider = fakeProvider(); await expect( @@ -219,7 +337,7 @@ describe("health response boundaries", () => { expect(healthyReport(invalid, now)).toBe(false); }); - test("HTTP errors, malformed JSON and network failures hold delivery", async () => { + test("HTTP errors, malformed JSON and network failures fail the health check", async () => { expect( await checkSyncHealth( "secret", diff --git a/scripts/loops/watchdog.ts b/scripts/loops/watchdog.ts index 6739626735..acb85933ae 100644 --- a/scripts/loops/watchdog.ts +++ b/scripts/loops/watchdog.ts @@ -109,15 +109,25 @@ export async function enforceDeliverySafety( assert.equal(identity.teamName, registry.teamName); const results: { workflow: string; action: string }[] = []; for (const target of options.targets ?? safetyTargets) { + let sending = false; + let attemptedUpdate = false; try { const path = `workflows/${target.workflowId}`; const workflow = await api.request<{ name: string; + status: string; mailingListId: string; rootNodeId: string; nodes: Record; }>(path); assert.equal(workflow.name, target.journey.name); + sending = workflow.status === "Sending"; + assert( + ["Draft", "Sending", "Paused", "PausedAndQueueing"].includes( + workflow.status, + ), + "Unknown workflow status", + ); assert.equal(workflow.mailingListId, target.mailingListId); assert.deepEqual(workflow.nodes[workflow.rootNodeId]?.nextNodeIds, [ target.guardId, @@ -158,7 +168,21 @@ export async function enforceDeliverySafety( }; action = options.apply ? "held" : "would-hold"; } + if ( + nextFilter && + isDeepStrictEqual(nextFilter, guard.audienceFilter) && + guard.appliesDownstream + ) + nextFilter = undefined; + if (nextFilter && sending) { + results.push({ + workflow: target.journey.key, + action: "manual-pause-required", + }); + continue; + } if (nextFilter && options.apply) { + attemptedUpdate = true; await api.request(nodePath, "POST", { expectedRevisionId: guard.workflowRevisionId, payload: { audienceFilter: nextFilter, appliesDownstream: true }, @@ -169,7 +193,20 @@ export async function enforceDeliverySafety( } results.push({ workflow: target.journey.key, action }); } catch { - results.push({ workflow: target.journey.key, action: "error" }); + if (attemptedUpdate && !sending) { + try { + const latest = await api.request<{ status: string }>( + `workflows/${target.workflowId}`, + ); + sending = latest.status === "Sending"; + } catch { + sending = false; + } + } + results.push({ + workflow: target.journey.key, + action: sending ? "manual-pause-required" : "error", + }); } } return results; @@ -191,6 +228,13 @@ if (import.meta.main) { { healthy, apply: values.apply, resume: values.resume, hold: values.hold }, ); console.log(JSON.stringify({ healthy, results })); + const manualPause = results.filter( + (result) => result.action === "manual-pause-required", + ); + if (manualPause.length) + console.error( + `Pause these workflows in Loops now: ${manualPause.map((result) => result.workflow).join(", ")}. The Loops API cannot edit audience filters while a workflow is sending. Delivery has not been stopped.`, + ); const successfulActions = values.hold ? ["held", "already-held"] : ["healthy", "resumed"];