fix(run-store,webapp): record every append outcome the store can return

The write counter allowed a vocabulary the store never produces, so a forked,
duplicate or skipped append all collapsed into one "other" series and could not
be alerted on. The allowlist now derives from the store own outcome union, and a
compile-time check in the source fails the build if a new outcome is added
without joining the list. Two counters whose only producer was unreachable are
removed.

Author: d-cs

apps/webapp/app/v3/snapshotStoreMetrics.server.ts

@@ -1,4 +1,5 @@
 import type { Meter } from "@internal/tracing";
+import { APPEND_RESULT_OUTCOMES } from "@internal/run-store";
 import type { DecoratorMetrics, SnapshotStoreMetrics } from "@internal/run-store";
 
 export type SnapshotSweepCounts = Record<string, number | boolean>;
@@ -8,7 +9,8 @@ export type SnapshotSweepCounts = Record<string, number | boolean>;
 // rather than minting a series.
 const APPEND_OUTCOMES = ["written", "duplicate", "forked", "skippedNoKeyspace"] as const;
 const APPEND_TTLS = ["none", "completion", "reapplied"] as const;
-const WRITE_OUTCOMES = ["written", "staged", "post_expiry", "skipped", "failed"] as const;
+/** Derived from the store's own vocabulary, so an added outcome cannot silently become "other". */
+export const WRITE_OUTCOMES = APPEND_RESULT_OUTCOMES;
 const READ_SOURCES = ["redis", "postgres"] as const;
 const SWEEP_OUTCOMES = [
   "completed",
@@ -69,9 +71,7 @@ export function createSnapshotStoreMetrics(meter: Meter) {
   const appendTotal = meter.createCounter("run_engine.snapshot_store.append_total");
   const writeTotal = meter.createCounter("run_engine.snapshot_store.write_total");
   const appendFailed = meter.createCounter("run_engine.snapshot_store.append_failed");
-  const flushStaged = meter.createCounter("run_engine.snapshot_store.flush_staged");
   const readSource = meter.createCounter("run_engine.snapshot_store.read_source");
-  const postExpiryWrite = meter.createCounter("run_engine.snapshot_store.post_expiry_write");
   const skippedNoKeyspace = meter.createCounter("run_engine.snapshot_store.skipped_no_keyspace");
   const cycleMismatch = meter.createCounter("run_engine.snapshot_store.cycle_mismatch");
   const entryBytes = meter.createHistogram("run_engine.snapshot_store.entry_bytes");
@@ -101,12 +101,6 @@ export function createSnapshotStoreMetrics(meter: Meter) {
         site: bounded(site, WRITE_SITES),
         outcome: bounded(outcome, WRITE_OUTCOMES),
       });
-      if (outcome === "post_expiry") {
-        postExpiryWrite.add(1);
-      }
-      if (outcome === "staged") {
-        flushStaged.add(1);
-      }
     },
     recordAppendFailed: (site) => appendFailed.add(1, { site: bounded(site, WRITE_SITES) }),
     recordRead: (method, source) =>

apps/webapp/test/snapshotStoreMetrics.test.ts

@@ -1,6 +1,8 @@
 import { readFileSync } from "node:fs";
 import { join } from "node:path";
 import { describe, expect, it } from "vitest";
+import { APPEND_RESULT_OUTCOMES } from "@internal/run-store";
+import { WRITE_OUTCOMES } from "~/v3/snapshotStoreMetrics.server";
 
 const SOURCE_PATH = join(process.cwd(), "app/v3/snapshotStoreMetrics.server.ts");
 
@@ -34,6 +36,21 @@ describe("snapshotStoreMetrics module shape", () => {
     expect(source).not.toMatch(/\btrimmed\b/);
   });
 
+  it("bounds the write outcome against the store's own vocabulary", () => {
+    // Any outcome the store can return but the allowlist omits collapses to "other", which hides
+    // forked appends: the signal that a run's Redis head has frozen.
+    for (const outcome of APPEND_RESULT_OUTCOMES) {
+      expect(WRITE_OUTCOMES).toContain(outcome);
+    }
+  });
+
+  it("declares no counter whose only producer is unreachable", () => {
+    // recordWrite is called once, with an AppendResult outcome. "staged" and "post_expiry" are not
+    // in that vocabulary, so both counters sat at zero and both branches were dead.
+    expect(source).not.toMatch(/flush_staged/);
+    expect(source).not.toMatch(/post_expiry_write/);
+  });
+
   it("gives the two layers separate counters", () => {
     // Sharing one would count a single logical write twice and mix {outcome, ttl} points with
     // {site, outcome} points under one name.

internal-packages/run-store/src/redisSnapshotStore.outcomes.test.ts

@@ -0,0 +1,27 @@
+import { describe, expect, it } from "vitest";
+import { APPEND_RESULT_OUTCOMES } from "./redisSnapshotStore.js";
+import type { AppendResult } from "./redisSnapshotStore.js";
+
+describe("append result outcomes", () => {
+  it("lists every outcome the store can return", () => {
+    // A type error here means AppendResult moved and the list did not follow. The metrics layer
+    // bounds against this list, so an omitted outcome collapses to "other".
+    type Declared = (typeof APPEND_RESULT_OUTCOMES)[number];
+    type Actual = AppendResult["outcome"];
+    type AssertSame<A, B> = [A] extends [B] ? ([B] extends [A] ? true : never) : never;
+    const _covers: AssertSame<Declared, Actual> = true;
+    void _covers;
+
+    expect([...APPEND_RESULT_OUTCOMES].sort()).toEqual([
+      "duplicate",
+      "forked",
+      "skippedNoKeyspace",
+      "written",
+    ]);
+  });
+
+  it("uses the result vocabulary, not the Lua wire vocabulary", () => {
+    expect(APPEND_RESULT_OUTCOMES).not.toContain("skipped");
+    expect(APPEND_RESULT_OUTCOMES).toContain("skippedNoKeyspace");
+  });
+});

internal-packages/run-store/src/redisSnapshotStore.ts

@@ -195,6 +195,25 @@ export type AppendResult =
   | { outcome: "forked"; actualCur: string }
   | { outcome: "duplicate"; seq: number };
 
+/** The single source of truth for the outcome vocabulary the metrics layer bounds against. */
+export const APPEND_RESULT_OUTCOMES = [
+  "written",
+  "skippedNoKeyspace",
+  "forked",
+  "duplicate",
+] as const satisfies readonly AppendResult["outcome"][];
+
+/**
+ * `satisfies` alone only proves each listed literal is a valid outcome. This proves the reverse too,
+ * so a new member on AppendResult fails the build here rather than becoming "other" on a dashboard.
+ */
+type AssertSameOutcomes<A, B> = [A] extends [B] ? ([B] extends [A] ? true : never) : never;
+const _outcomesExhaustive: AssertSameOutcomes<
+  (typeof APPEND_RESULT_OUTCOMES)[number],
+  AppendResult["outcome"]
+> = true;
+void _outcomesExhaustive;
+
 export type SnapshotStoreMetrics = {
   recordAppend(outcome: string, ttl: string): void;
   recordEntryBytes(bytes: number): void;
@@ -292,6 +311,15 @@ export class RedisSnapshotStore {
     }
   }
 
+  /**
+   * Removes a run's whole keyspace, wait-cycle keys included. The caller must have established that
+   * the head cannot be trusted and that Postgres still holds the run's rows.
+   */
+  async dropRun(runId: string): Promise<void> {
+    const keys = snapshotKeys(runId);
+    await this.redis.dropSnapshotRun(keys.e, keys.idx, keys.cur, keys.seq);
+  }
+
   async append(args: {
     entry: SnapshotEntryInput;
     kind: "birth" | "transition";
@@ -859,6 +887,18 @@ export class RedisSnapshotStore {
       `,
     });
 
+    this.redis.defineCommand("dropSnapshotRun", {
+      numberOfKeys: 4,
+      lua: `
+        ${PRELUDE}
+        local cycles = tonumber(redis.call('HGET', seqKey, 'c') or '0')
+        for i = 1, cycles do
+          redis.call('DEL', wpKey(i))
+        end
+        return redis.call('DEL', eKey, idxKey, curKey, seqKey)
+      `,
+    });
+
     this.redis.defineCommand("readSnapshotById", {
       numberOfKeys: 4,
       lua: `
@@ -1046,6 +1086,13 @@ export function decodeWaitpointIds(
 
 declare module "@internal/redis" {
   interface RedisCommander<Context> {
+    dropSnapshotRun(
+      eKey: string,
+      idxKey: string,
+      curKey: string,
+      seqKey: string,
+      callback?: Callback<number>
+    ): Result<number, Context>;
     appendSnapshotEntry(
       eKey: string,
       idxKey: string,