fix: verify workflow status after rejected guard updates

Author: richiemcilroy

scripts/loops/watchdog.test.ts

@@ -28,6 +28,8 @@ function fakeProvider(status = "Draft") {
 	const attemptedWrites: string[] = [];
 	let brokenWorkflow: string | undefined;
 	let startedWorkflow: string | undefined;
+	let startedOnWrite = false;
+	let startError: Error | undefined;
 	const api: Pick<LoopsApi, "request"> = {
 		async request<T>(path: string, method = "GET", body?: unknown): Promise<T> {
 			if (path === "api-key") return { teamName: "Cap Software, Inc." } as T;
@@ -40,6 +42,10 @@ function fakeProvider(status = "Draft") {
 				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:
@@ -61,7 +67,10 @@ function fakeProvider(status = "Draft") {
 			}
 			return {
 				name: target.journey.name,
-				status,
+				status:
+					target.workflowId === startedWorkflow && startedOnWrite
+						? "Sending"
+						: status,
 				mailingListId: target.mailingListId,
 				rootNodeId: "trigger",
 				nodes: { trigger: { nextNodeIds: [target.guardId] } },
@@ -73,8 +82,9 @@ function fakeProvider(status = "Draft") {
 		guards,
 		writes,
 		attemptedWrites,
-		startDuringWrite: (id: string) => {
+		startDuringWrite: (id: string, error?: Error) => {
 			startedWorkflow = id;
+			startError = error;
 		},
 		fail: (id: string) => {
 			brokenWorkflow = id;
@@ -102,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, {

scripts/loops/watchdog.ts

@@ -8,7 +8,7 @@ import {
 import { journeys } from "../../emails/flows";
 import registry from "../../emails/resources.json";
 import type { Journey } from "../../emails/types";
-import { LoopsApi, LoopsApiError } from "./api";
+import { LoopsApi } from "./api";
 
 export const healthUrl = "https://cap.so/api/cron/sync-loops/health";
 
@@ -109,6 +109,8 @@ 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<{
@@ -119,6 +121,7 @@ export async function enforceDeliverySafety(
 				nodes: Record<string, { nextNodeIds: string[] }>;
 			}>(path);
 			assert.equal(workflow.name, target.journey.name);
+			sending = workflow.status === "Sending";
 			assert(
 				["Draft", "Sending", "Paused", "PausedAndQueueing"].includes(
 					workflow.status,
@@ -171,14 +174,15 @@ export async function enforceDeliverySafety(
 				guard.appliesDownstream
 			)
 				nextFilter = undefined;
-			if (nextFilter && workflow.status === "Sending") {
+			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 },
@@ -188,15 +192,17 @@ export async function enforceDeliverySafety(
 				assert.equal(confirmed.appliesDownstream, true);
 			}
 			results.push({ workflow: target.journey.key, action });
-		} catch (error) {
-			const sending =
-				error instanceof LoopsApiError &&
-				error.status === 400 &&
-				typeof error.details === "object" &&
-				error.details !== null &&
-				"message" in error.details &&
-				error.details.message ===
-					"This operation is not allowed while the workflow is sending.";
+		} catch {
+			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",