fix(webapp): read the alert channel and the watch target on the primary

Author: kathiekiwi

.server-changes/dashboard-agent-watch-reads.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Unsubscribing from watch notifications now takes effect immediately, so no alert arrives after you opt out. You can also set a watch on a run the moment you trigger it, instead of being told the run doesn't exist.

apps/webapp/app/services/dashboardAgentWatchChecks.server.ts

@@ -4,7 +4,7 @@
  */
 
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
-import { $replica } from "~/db.server";
+import { $replica, prisma } from "~/db.server";
 import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 import { ReportPresenter } from "~/presenters/v3/reports/ReportPresenter.server";
 import { engine } from "~/v3/runEngine.server";
@@ -52,6 +52,30 @@ export async function watchQueueExists(environmentId: string, queueName: string)
   return queue !== null;
 }
 
+/** The same run read on the primary, for a target that may have been created a moment ago. */
+export async function readWatchRunOnPrimary(
+  runFriendlyId: string,
+  environmentId: string
+): Promise<WatchRunRow | null> {
+  const run = await runStore.findRunOnPrimary(
+    { friendlyId: runFriendlyId, runtimeEnvironmentId: environmentId },
+    { select: WATCH_RUN_SELECT }
+  );
+  return run ?? null;
+}
+
+/** The same queue read on the primary. */
+export async function watchQueueExistsOnPrimary(
+  environmentId: string,
+  queueName: string
+): Promise<boolean> {
+  const queue = await prisma.taskQueue.findFirst({
+    where: { runtimeEnvironmentId: environmentId, name: queueName },
+    select: { id: true },
+  });
+  return queue !== null;
+}
+
 /** How far back the ClickHouse depth fallback looks when the live counter is down. */
 const DEPTH_FALLBACK_MINUTES = 10;
 const DEPTH_FALLBACK_BUCKET_SECONDS = 60;
@@ -290,3 +314,18 @@ export function watchCheckDeps(
     readHealth: () => readWatchHealth(environment),
   };
 }
+
+/**
+ * Creation-time deps. The target reads go to the primary, so a run or queue created moments
+ * ago is visible instead of failing as a non-existent target inside the replication window.
+ */
+export function watchCreationCheckDeps(
+  environment: AuthenticatedEnvironment,
+  now: Date = new Date()
+): WatchCheckDeps {
+  return {
+    ...watchCheckDeps(environment, now),
+    readRun: (runId) => readWatchRunOnPrimary(runId, environment.id),
+    queueExists: (queue) => watchQueueExistsOnPrimary(environment.id, queue),
+  };
+}

apps/webapp/app/services/dashboardAgentWatches.server.ts

@@ -41,7 +41,7 @@ import {
   type WatchCheckDeps,
   type WatchCheckOutcome,
 } from "~/services/dashboardAgentWatchChecks";
-import { watchCheckDeps } from "~/services/dashboardAgentWatchChecks.server";
+import { watchCreationCheckDeps } from "~/services/dashboardAgentWatchChecks.server";
 import {
   mintDashboardAgentWatchBatchToken,
   mintDashboardAgentWatchToken,
@@ -214,7 +214,8 @@ export async function createDashboardAgentWatch(params: {
 }): Promise<CreateDashboardAgentWatchResult> {
   const { environment, userId, chatId, spec } = params;
   const now = params.now ?? new Date();
-  const buildCheckDeps = params.deps?.checkDeps ?? watchCheckDeps;
+  // Creation reads the target on the primary; the polling checks stay on the replica.
+  const buildCheckDeps = params.deps?.checkDeps ?? watchCreationCheckDeps;
   const scheduleTick = params.deps?.scheduleTick ?? scheduleWatchTick;
   const isDashboardAgentConfigured = params.deps?.configured ?? isDashboardAgentConfiguredDefault;
   const checkDeps = buildCheckDeps(environment, now);

apps/webapp/app/v3/services/alerts/deliverDashboardAgentWatchAlert.server.ts

@@ -180,9 +180,10 @@ export class DeliverDashboardAgentWatchAlertService {
 /** One channel's delivery. A retry here can only re-send this channel. */
 export class DeliverDashboardAgentWatchChannelAlertService {
   async call(payload: DashboardAgentWatchChannelAlertPayload): Promise<void> {
-    // Re-read the channel rather than trusting the fan-out's snapshot: an
-    // unsubscribe between fan-out and delivery should stop the alert.
-    const channel = await $replica.projectAlertChannel.findFirst({
+    // Re-read the channel rather than trusting the fan-out's snapshot: an unsubscribe
+    // between fan-out and delivery should stop the alert. The primary, since the
+    // unsubscribe writes there and replica lag would send the mail anyway.
+    const channel = await prisma.projectAlertChannel.findFirst({
       where: {
         id: payload.channelId,
         projectId: payload.projectId,

apps/webapp/test/dashboardAgentWatchAlertFanout.test.ts

@@ -23,6 +23,8 @@ const WEBHOOK_CHANNEL = {
 
 const ctx = vi.hoisted(() => ({
   channels: [] as Array<{ id: string; type: string; properties: unknown }>,
+  /** Set to model replica lag: what the replica still sees. Null means "same as primary". */
+  replicaChannels: null as Array<{ id: string; type: string; properties: unknown }> | null,
   gateAllowed: true,
   webhookFails: false,
 }));
@@ -37,6 +39,11 @@ const safeWebhookFetch = vi.hoisted(() =>
 );
 
 vi.mock("~/db.server", () => {
+  const channelReader = (rows: () => Array<{ id: string; type: string; properties: unknown }>) => ({
+    findMany: async () => rows(),
+    findFirst: async ({ where }: { where: { id: string } }) =>
+      rows().find((channel) => channel.id === where.id) ?? null,
+  });
   const db = {
     runtimeEnvironment: {
       findFirst: async () => ({
@@ -51,11 +58,6 @@ vi.mock("~/db.server", () => {
         },
       }),
     },
-    projectAlertChannel: {
-      findMany: async () => ctx.channels,
-      findFirst: async ({ where }: { where: { id: string } }) =>
-        ctx.channels.find((channel) => channel.id === where.id) ?? null,
-    },
     organizationIntegration: {
       findFirst: async () => ({
         id: "int_1",
@@ -65,7 +67,14 @@ vi.mock("~/db.server", () => {
       }),
     },
   };
-  return { prisma: db, $replica: db, sqlDatabaseSchema: undefined };
+  return {
+    prisma: { ...db, projectAlertChannel: channelReader(() => ctx.channels) },
+    $replica: {
+      ...db,
+      projectAlertChannel: channelReader(() => ctx.replicaChannels ?? ctx.channels),
+    },
+    sqlDatabaseSchema: undefined,
+  };
 });
 
 vi.mock("~/v3/canAccessDashboardAgent.server", () => ({
@@ -111,6 +120,7 @@ const payload = {
 
 beforeEach(() => {
   ctx.channels = [EMAIL_CHANNEL, SLACK_CHANNEL, WEBHOOK_CHANNEL];
+  ctx.replicaChannels = null;
   ctx.gateAllowed = true;
   ctx.webhookFails = false;
   enqueue.mockClear();
@@ -195,6 +205,18 @@ describe("dashboard agent watch alert per-channel delivery", () => {
     expect(bodies[1].created).toBe(bodies[0].created);
   });
 
+  test("an unsubscribe the replica hasn't caught up on still stops the email", async () => {
+    ctx.channels = [];
+    ctx.replicaChannels = [EMAIL_CHANNEL];
+
+    await new DeliverDashboardAgentWatchChannelAlertService().call({
+      ...payload,
+      channelId: "chan_email",
+    });
+
+    expect(sendAlertEmail).not.toHaveBeenCalled();
+  });
+
   test("an unsubscribed channel delivers nothing", async () => {
     ctx.channels = [];
     await new DeliverDashboardAgentWatchChannelAlertService().call({

apps/webapp/test/dashboardAgentWatchCreationReads.test.ts

@@ -0,0 +1,71 @@
+import { beforeEach, describe, expect, test, vi } from "vitest";
+import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+
+// The replica lags: it has neither the just-triggered run nor the just-created queue.
+const ctx = vi.hoisted(() => ({
+  primaryReads: [] as string[],
+  replicaReads: [] as string[],
+}));
+
+vi.mock("~/db.server", () => ({
+  prisma: {
+    taskQueue: {
+      findFirst: async () => {
+        ctx.primaryReads.push("queue");
+        return { id: "queue_1" };
+      },
+    },
+  },
+  $replica: {
+    taskQueue: {
+      findFirst: async () => {
+        ctx.replicaReads.push("queue");
+        return null;
+      },
+    },
+  },
+  sqlDatabaseSchema: undefined,
+}));
+
+vi.mock("~/v3/runStore.server", () => ({
+  runStore: {
+    findRunOnPrimary: async () => {
+      ctx.primaryReads.push("run");
+      return { friendlyId: "run_1", status: "PENDING" };
+    },
+    findRun: async () => {
+      ctx.replicaReads.push("run");
+      return null;
+    },
+  },
+}));
+
+const { watchCheckDeps, watchCreationCheckDeps } =
+  await import("~/services/dashboardAgentWatchChecks.server");
+
+const environment = { id: "env_1" } as AuthenticatedEnvironment;
+
+beforeEach(() => {
+  ctx.primaryReads = [];
+  ctx.replicaReads = [];
+});
+
+describe("the watch target reads", () => {
+  test("creation reads the run and the queue on the primary", async () => {
+    const deps = watchCreationCheckDeps(environment);
+
+    expect(await deps.readRun("run_1")).not.toBeNull();
+    expect(await deps.queueExists("task/my-task")).toBe(true);
+    expect(ctx.primaryReads).toEqual(["run", "queue"]);
+    expect(ctx.replicaReads).toEqual([]);
+  });
+
+  test("polling keeps both reads on the replica", async () => {
+    const deps = watchCheckDeps(environment);
+
+    expect(await deps.readRun("run_1")).toBeNull();
+    expect(await deps.queueExists("task/my-task")).toBe(false);
+    expect(ctx.replicaReads).toEqual(["run", "queue"]);
+    expect(ctx.primaryReads).toEqual([]);
+  });
+});