When a Deploy Drops the Webhook: Reconciling Gemini Long-Running Operations with a Belt-and-Suspenders Design

// operation-ledger.ts
export type OpStatus = "pending" | "succeeded" | "failed";
 
export interface OpRecord {
  opName: string;        // The operation resource name returned by Gemini
  jobKind: string;       // A usage identifier such as "batch-review-classify"
  status: OpStatus;
  submittedAt: number;   // epoch ms
  settledAt?: number;    // When the terminal state was confirmed
  settledBy?: "webhook" | "reconcile";
  attempts: number;      // How many times reconciliation looked it up
  sideEffectDone: boolean; // Whether the result write-back completed
}
 
const TTL_SECONDS = 60 * 60 * 24 * 14; // Keep 14 days for auditing
 
export async function putOp(kv: KVNamespace, rec: OpRecord): Promise<void> {
  await kv.put(`op:${rec.opName}`, JSON.stringify(rec), {
    expirationTtl: TTL_SECONDS,
    metadata: { status: rec.status, submittedAt: rec.submittedAt },
  });
}
 
export async function getOp(kv: KVNamespace, opName: string): Promise<OpRecord | null> {
  const raw = await kv.get(`op:${opName}`);
  return raw ? (JSON.parse(raw) as OpRecord) : null;
}
 
export async function registerSubmission(
  kv: KVNamespace,
  opName: string,
  jobKind: string,
): Promise<void> {
  await putOp(kv, {
    opName,
    jobKind,
    status: "pending",
    submittedAt: Date.now(),
    attempts: 0,
    sideEffectDone: false,
  });
}

// webhook-handler.ts
import { getOp, putOp } from "./operation-ledger";
import { verifySignature } from "./verify-signature";
import { runSideEffect } from "./side-effect";
 
export async function handleWebhook(req: Request, env: Env): Promise<Response> {
  const raw = await req.text();
  if (!(await verifySignature(raw, req.headers, env.WEBHOOK_SECRET))) {
    return new Response("invalid signature", { status: 401 });
  }
 
  const event = JSON.parse(raw) as { opName: string; state: string };
  // Only handle terminal states. Intermediate events can be ignored.
  const terminal =
    event.state === "SUCCEEDED" ? "succeeded" :
    event.state === "FAILED" ? "failed" : null;
  if (!terminal) return new Response("ignored (non-terminal)", { status: 200 });
 
  await settle(env.OPS_KV, event.opName, terminal, "webhook");
  // Return 200 only after the side effect is confirmed.
  // An early ACK stops the retries and drops the event.
  return new Response("ok", { status: 200 });
}
 
// Terminal confirmation + side effect. Called by both webhook and reconcile.
export async function settle(
  kv: KVNamespace,
  opName: string,
  status: "succeeded" | "failed",
  by: "webhook" | "reconcile",
): Promise<void> {
  const rec = await getOp(kv, opName);
  if (!rec) return;                 // Not in the ledger = not our concern
  if (rec.status !== "pending") return; // Already closed = ignore idempotently
 
  if (status === "succeeded" && !rec.sideEffectDone) {
    await runSideEffect(opName);    // Write the result back exactly once
  }
  rec.status = status;
  rec.settledAt = Date.now();
  rec.settledBy = by;
  rec.sideEffectDone = status === "succeeded";
  await putOp(kv, rec);
}

// reconcile.ts
import { getOp, putOp } from "./operation-ledger";
import { settle } from "./webhook-handler";
import { fetchOperationState } from "./gemini-ops";
 
const GRACE_MS = 90_000;       // Wait 90s after submit for the webhook
const STUCK_MS = 6 * 3600_000; // 6h pending = treat as an anomaly, alert
 
export async function reconcile(env: Env): Promise<void> {
  const now = Date.now();
  let cursor: string | undefined;
 
  do {
    const page = await env.OPS_KV.list({ prefix: "op:", cursor, limit: 1000 });
    cursor = page.list_complete ? undefined : page.cursor;
 
    for (const key of page.keys) {
      const meta = key.metadata as { status?: string; submittedAt?: number } | undefined;
      // Sift coarsely on metadata. Never load the body for non-pending entries.
      if (meta?.status !== "pending") continue;
      if (meta.submittedAt && now - meta.submittedAt < GRACE_MS) continue;
 
      const rec = await getOp(env.OPS_KV, key.name.slice(3));
      if (!rec || rec.status !== "pending") continue;
 
      const state = await fetchOperationState(rec.opName, env); // operations.get
      if (state === "SUCCEEDED" || state === "FAILED") {
        await settle(env.OPS_KV, rec.opName, state === "SUCCEEDED" ? "succeeded" : "failed", "reconcile");
        continue;
      }
 
      // Still running. Advance the attempt count and alert if it ran too long.
      rec.attempts += 1;
      await putOp(env.OPS_KV, rec);
      if (now - rec.submittedAt > STUCK_MS) {
        await alertStuck(rec); // To Slack, etc. Never auto-close it.
      }
    }
  } while (cursor);
}

// A second scan added inside the reconcile loop
if (rec.status === "succeeded" && !rec.sideEffectDone) {
  await runSideEffect(rec.opName);
  rec.sideEffectDone = true;
  await putOp(env.OPS_KV, rec);
}