feat(keepers): dedup keeper submissions across invocations

apps/docs/operations/accrual-keeper.md

@@ -90,11 +90,40 @@ HTTP 500 so the scheduled run is observable instead of silently passing.
 
 If a submitted `accrue()` transaction is still unconfirmed when a retry
 attempt times out, the keeper re-checks that same transaction hash instead of
-sending a new one, within a single run. This tracking does not persist across
-separate keeper invocations: if a run exhausts its retries while a submission
-is still unconfirmed, the next scheduled run has no memory of it and may send
-a fresh `accrue()` transaction for the same adapter. This is an accepted,
-bounded gap rather than a fund-safety issue: `accrue()` only refreshes a
-cached value from the adapter's live position and produces the same result no
-matter how many times it lands, so a duplicate costs at most one extra
-Soroban fee, not incorrect accounting.
+sending a new one, within a single run.
+
+That tracking also persists **across** invocations (#515). The submitted
+hash is recorded in the shared store (Upstash Redis, keyed
+`meridian:keeper:accrual:<network>:<vaultId>:<adapterId>`) as soon as the
+transaction is broadcast, and every run resolves an existing record against
+the network before submitting anything: landed, failed, or aged out past the
+transaction's validity window clears it, and only a genuinely still-in-flight
+one skips the adapter for that run. The mechanism, its state machine, and
+`MERIDIAN_KEEPER_SUBMISSION_TTL_MS` are documented in full in
+[Migration Keeper](./migration-keeper.md#cross-invocation-duplicate-protection);
+this keeper uses exactly the same code path, deliberately, so both keepers'
+execution model is the same thing to reason about.
+
+The one difference is the fallback. Where the migration keeper refuses to run
+in production without a shared store, this keeper falls back to a
+per-invocation in-memory one (logging that dedup is inactive) and keeps
+running: a duplicate `accrue()` only refreshes a cached value from the
+adapter's live position and produces the same result no matter how many times
+it lands, so it costs at most one extra Soroban fee, not incorrect
+accounting. The migration keeper's duplicate costs real slippage twice, which
+is why only it fails closed.
+
+## Racing The Migration Keeper
+
+Both keepers act on the same vault's adapter independently. This keeper can
+read `get_adapter()` at discovery, have the migration keeper switch the vault
+to a different adapter before this submission lands, and then call `accrue()`
+on the now-detached adapter, which succeeds and does nothing useful (a
+detached adapter is still a valid contract, so nothing errors) while the
+yield it would have accrued never reaches the vault.
+
+Before building a new `accrue()` transaction, the keeper therefore re-reads
+the vault's live `get_adapter()` and skips the adapter if the vault has
+already moved on. The next run's discovery picks up the new adapter. The skip
+is reported in `skipped[]`, not `failures[]`: it is a benign race, and the
+new adapter is accrued on the following tick.

apps/docs/operations/migration-keeper.md

@@ -180,32 +180,91 @@ failure (e.g. slippage exceeded) is reported immediately without retrying,
 and the run stops starting new work once it's within `vercel.json`'s
 `maxDuration` budget rather than risk being killed mid-retry.
 
-The in-flight-transaction tracking (`priorHash`) only covers a single
-invocation, exactly like the accrue keeper's own version of this gap (see
-`apps/docs/operations/accrual-keeper.md`). If the process is killed (or a
-run exhausts its retries) while a `migrate_adapter` transaction is sent but
-still unconfirmed, the next scheduled run has no memory of it: discovery
-reads whatever adapter is live on-chain at that point and evaluates fresh,
-so it will not deliberately resend the exact same migration, but if the
-prior transaction is still landing when the next run fires, a second,
-independent `migrate_adapter` call can still go out before the first
-confirms. Unlike `accrue()`, this isn't free: each call is its own
-slippage-bounded transaction, so a genuine double-migration costs real
-slippage twice. This is an accepted, bounded gap
-covered by the same cross-invocation persistence work needed for the accrue
-keeper, not something this keeper solves on its own (tracked in #515, which
-also needs to account for the accrue keeper racing against this one: both
-act on the same vault's adapter independently, with no coordination between
-them, see #515 for the full scope once `migrate_adapter` is actually live
-on the vault).
-
-Before building a brand-new transaction (not when rechecking an
-already-sent one), the keeper re-reads the vault's live `get_adapter()` and
-compares it against what discovery saw for this run. A mismatch means
-something else already changed the vault's adapter since this run started,
-and the migration is skipped rather than submitted against stale
-assumptions. This narrows the cross-invocation race window; it does not
-close it, a mismatch can still occur between this check and the
-transaction actually landing on-chain (an unavoidable TOCTOU gap without a
-contract-level compare-and-swap), but it catches the common case of "a
-prior run's migration already landed" for free.
+## Cross-Invocation Duplicate Protection
+
+`priorHash` (in `keeper-tx.ts`) only tracks an unconfirmed transaction
+_within_ one invocation. That alone is not enough here: if the process is
+killed, or a run exhausts its retries while a `migrate_adapter` transaction
+is sent but unconfirmed, the next scheduled run would have no memory of it
+and could send a second, independent migration while the first is still
+landing. Unlike `accrue()`, that isn't free, each call is its own
+slippage-bounded transaction, so a double-migration costs real slippage
+twice.
+
+Two guards close that, and they cover different failure windows:
+
+**1. A shared submission record** (`packages/stellar-sdk-helpers/src/keeper-state.ts`).
+One record per vault, in Upstash Redis, keyed
+`meridian:keeper:migration:<network>:<vaultId>`, holding just the submitted
+transaction hash and the time it was broadcast.
+
+The record is written **only after** `sendTransaction` returns a hash, never
+before. There is deliberately no "about to send" state, so a crash between
+deciding to migrate and actually broadcasting leaves nothing behind that
+could block the next run.
+
+At the start of every run, an existing record is **resolved against the
+network**, never trusted on its own word:
+
+| Lookup of the recorded hash                           | Meaning                              | Action                           |
+| ----------------------------------------------------- | ------------------------------------ | -------------------------------- |
+| `SUCCESS`                                             | the migration landed                 | clear the record, evaluate again |
+| `FAILED`                                              | it failed on-chain                   | clear the record, retry allowed  |
+| not found, older than the transaction validity window | provably dead, it can never land now | clear the record, retry allowed  |
+| not found, still inside that window                   | genuinely still in flight            | **skip this vault this run**     |
+| the store or the lookup itself errored                | unknown                              | **skip this vault this run**     |
+
+So a record can never block a vault indefinitely: it either resolves to a
+real outcome or ages out. The window comes from the transaction's own time
+bounds, `submitKeeperOperation` builds with `.setTimeout(300)`, so
+`MERIDIAN_KEEPER_SUBMISSION_TTL_MS` defaults to `360000` (300s plus 60s of
+clock-skew margin). Every record is also written with a Redis-side expiry of
+the same length, so even a run that dies before it can clear a record cannot
+leave one behind past the point where its transaction could still land.
+
+An unreadable store is treated as _unknown_, not as "nothing was submitted":
+reading a KV outage as "safe to migrate" would produce exactly the duplicate
+this exists to prevent. Migrations pause (visibly, in `skipped[]`) until the
+store is reachable again.
+
+Because a per-process fallback cannot dedup across invocations at all, the
+migration keeper **refuses to run in production** without
+`UPSTASH_REDIS_REST_URL`/`UPSTASH_REDIS_REST_TOKEN`, the same pair
+`api/_lib/middleware.ts` already requires there for distributed rate
+limiting. Outside production it falls back to a per-invocation in-memory
+store and logs that dedup is inactive for the run.
+
+**2. The on-chain adapter re-check.** Before building a brand-new transaction
+(not when rechecking an already-sent one), the keeper re-reads the vault's
+live `get_adapter()` and compares it against what discovery saw for this run.
+A mismatch means something else already changed the vault's adapter, and the
+migration is skipped rather than submitted against stale assumptions.
+
+This is not redundant with the record: it covers the one window the record
+cannot, where the broadcast succeeded but the process died before the record
+was written. In that case the next run has no record, but it does see the
+vault already sitting on the new adapter, and skips. Conversely, the record
+covers what the re-check cannot, an unconfirmed transaction that has not yet
+changed the adapter. A TOCTOU gap still remains between the re-check and the
+transaction landing (unavoidable without a contract-level compare-and-swap),
+which is why both guards exist rather than either alone.
+
+Skips from either guard land in `skipped[]`, not `failures[]`: both are
+benign, expected races, and a keeper that returned HTTP 500 every time one
+fired would page someone for correct behavior.
+
+## Coordination With The Accrue Keeper
+
+The two keepers act on the same vault's adapter independently. The accrue
+keeper can read `get_adapter()` at discovery, have this keeper switch the
+vault to a different adapter before its submission lands, and then call
+`accrue()` on the now-detached adapter, a silently ineffective call (a
+detached adapter is still a valid contract, so nothing errors) whose yield
+never reaches the vault.
+
+The accrue keeper therefore runs the same live-`get_adapter()` re-check
+before building its own transaction, and skips when the vault has moved on
+(see `apps/docs/operations/accrual-keeper.md`). No lock or shared ordering
+between the two keepers is introduced: each independently refuses to act on
+an adapter the vault no longer uses, which is enough to make the race benign
+without coupling their schedules.

packages/stellar-sdk-helpers/src/accrual-keeper.test.ts

@@ -99,6 +99,7 @@ import {
   type KeeperLogger,
 } from "./accrual-keeper";
 import type { KnownPoolMeta } from "./known-pools";
+import { submissionStateKey, type SubmissionRecord } from "./keeper-state";
 
 const NETWORK = {
   network: "testnet" as const,
@@ -112,6 +113,7 @@ const CONFIG: BlendAccrualKeeperConfig = {
   maxAttempts: 3,
   baseDelayMs: 1,
   rpcTimeoutMs: 100,
+  submissionTtlMs: 360_000,
 };
 
 const VAULT: KnownPoolMeta = {
@@ -183,6 +185,10 @@ beforeEach(() => {
     build: () => ({ tx, sign: stellarMocks.signPrepared }),
   }));
   stellarMocks.simulateView.mockReset();
+  // The pre-submit "the vault still uses this adapter" guard reads
+  // get_adapter() fresh on the default submission path; keep it matching
+  // BLEND_ADAPTER unless a test is specifically exercising a mismatch.
+  stellarMocks.simulateView.mockResolvedValue(BLEND_ADAPTER.adapterId);
   stellarMocks.waitForTransaction.mockResolvedValue({ ledger: 999 });
 });
 
@@ -1367,3 +1373,202 @@ describe("runBlendAccrualKeeper", () => {
     ]);
   });
 });
+
+describe("runBlendAccrualKeeper cross-invocation dedup", () => {
+  function store(initial?: Record<string, SubmissionRecord>) {
+    const records = new Map<string, SubmissionRecord>(
+      Object.entries(initial ?? {})
+    );
+    return {
+      records,
+      get: vi.fn(async (key: string) => records.get(key) ?? null),
+      set: vi.fn(async (key: string, record: SubmissionRecord) => {
+        records.set(key, record);
+      }),
+      delete: vi.fn(async (key: string) => {
+        records.delete(key);
+      }),
+    };
+  }
+
+  const KEY = submissionStateKey(
+    "accrual",
+    "testnet",
+    BLEND_ADAPTER.vaultId,
+    BLEND_ADAPTER.adapterId
+  );
+
+  it("skips an adapter whose prior accrue() is still unconfirmed instead of sending a second one", async () => {
+    // The gap this closes: the record is the only thing that survives a
+    // killed invocation, so without it the next cron tick would happily
+    // broadcast a duplicate while the first transaction is still landing.
+    const server = makeServer({
+      getTransaction: vi.fn(async () => ({ status: "NOT_FOUND" })),
+    });
+    stellarMocks.getRpcServer.mockReturnValue(server);
+    const stateStore = store({
+      [KEY]: { hash: "INFLIGHT_HASH", submittedAtMs: Date.now() - 1_000 },
+    });
+
+    const result = await runBlendAccrualKeeper(CONFIG, {
+      logger: logger(),
+      sleep: vi.fn(),
+      stateStore,
+      discoverAdapters: async () => ({
+        adapters: [BLEND_ADAPTER],
+        failures: [],
+      }),
+    });
+
+    expect(server.sendTransaction).not.toHaveBeenCalled();
+    expect(result.successes).toEqual([]);
+    expect(result.failures).toEqual([]);
+    expect(result.skipped).toMatchObject([
+      {
+        vaultId: "meridian-usdc",
+        adapterId: "CADAPTERBLEND",
+        reason: expect.stringContaining("still unconfirmed"),
+      },
+    ]);
+    // The record is left in place: it's still genuinely in flight.
+    expect(stateStore.records.get(KEY)).toBeDefined();
+  });
+
+  it("clears a prior submission that actually landed and submits again", async () => {
+    const server = makeServer({
+      getTransaction: vi.fn(async () => ({ status: "SUCCESS", ledger: 12 })),
+    });
+    stellarMocks.getRpcServer.mockReturnValue(server);
+    const stateStore = store({
+      [KEY]: { hash: "LANDED_HASH", submittedAtMs: Date.now() - 1_000 },
+    });
+
+    const result = await runBlendAccrualKeeper(CONFIG, {
+      logger: logger(),
+      sleep: vi.fn(),
+      stateStore,
+      discoverAdapters: async () => ({
+        adapters: [BLEND_ADAPTER],
+        failures: [],
+      }),
+    });
+
+    expect(result.successes).toMatchObject([{ hash: "HASH" }]);
+    expect(server.sendTransaction).toHaveBeenCalledOnce();
+    // Cleared once resolved, and again once this run's own submission
+    // confirmed, so nothing is left to block the next tick.
+    expect(stateStore.records.get(KEY)).toBeUndefined();
+  });
+
+  it("ages out a record whose transaction can no longer land, rather than blocking forever", async () => {
+    // NOT_FOUND past the transaction's own validity window means it is
+    // provably dead; without this the record would block every subsequent
+    // run until a human intervened.
+    const server = makeServer({
+      getTransaction: vi.fn(async () => ({ status: "NOT_FOUND" })),
+    });
+    stellarMocks.getRpcServer.mockReturnValue(server);
+    const stateStore = store({
+      [KEY]: {
+        hash: "DEAD_HASH",
+        submittedAtMs: Date.now() - CONFIG.submissionTtlMs - 1_000,
+      },
+    });
+
+    const result = await runBlendAccrualKeeper(CONFIG, {
+      logger: logger(),
+      sleep: vi.fn(),
+      stateStore,
+      discoverAdapters: async () => ({
+        adapters: [BLEND_ADAPTER],
+        failures: [],
+      }),
+    });
+
+    expect(result.successes).toMatchObject([{ hash: "HASH" }]);
+    expect(server.sendTransaction).toHaveBeenCalledOnce();
+  });
+
+  it("records the broadcast hash before waiting for confirmation, not after", async () => {
+    // The wait is exactly what times out, so a record written after it
+    // would be missing in the case it exists for.
+    let recordedWhilePending: SubmissionRecord | undefined;
+    const server = makeServer({
+      getTransaction: vi.fn(async () => ({ status: "NOT_FOUND" })),
+      sendTransaction: vi.fn(async () => ({
+        hash: "FRESH_HASH",
+        status: "PENDING",
+      })),
+    });
+    stellarMocks.getRpcServer.mockReturnValue(server);
+    const stateStore = store();
+    stellarMocks.waitForTransaction.mockImplementation(async () => {
+      recordedWhilePending = stateStore.records.get(KEY);
+      return { ledger: 7 };
+    });
+
+    await runBlendAccrualKeeper(CONFIG, {
+      logger: logger(),
+      sleep: vi.fn(),
+      stateStore,
+      discoverAdapters: async () => ({
+        adapters: [BLEND_ADAPTER],
+        failures: [],
+      }),
+    });
+
+    expect(recordedWhilePending).toMatchObject({ hash: "FRESH_HASH" });
+    expect(stateStore.records.get(KEY)).toBeUndefined();
+  });
+
+  it("skips rather than guesses when the submission state store cannot be read", async () => {
+    const stateStore = store();
+    stateStore.get.mockRejectedValue(new Error("KV unavailable"));
+    const server = makeServer();
+    stellarMocks.getRpcServer.mockReturnValue(server);
+
+    const result = await runBlendAccrualKeeper(CONFIG, {
+      logger: logger(),
+      sleep: vi.fn(),
+      stateStore,
+      discoverAdapters: async () => ({
+        adapters: [BLEND_ADAPTER],
+        failures: [],
+      }),
+    });
+
+    expect(server.sendTransaction).not.toHaveBeenCalled();
+    expect(result.skipped).toMatchObject([
+      { reason: expect.stringContaining("could not be verified") },
+    ]);
+  });
+
+  it("skips accruing an adapter the vault has already migrated away from", async () => {
+    // The accrue/migrate race folded into #515: accrue() on a detached
+    // adapter succeeds and does nothing, so it must not be reported as a
+    // success (nor as a failure, it is a benign race).
+    stellarMocks.simulateView.mockResolvedValue("CADAPTERDEFINDEX_NEW");
+    const server = makeServer();
+    stellarMocks.getRpcServer.mockReturnValue(server);
+
+    const result = await runBlendAccrualKeeper(CONFIG, {
+      logger: logger(),
+      sleep: vi.fn(),
+      stateStore: store(),
+      discoverAdapters: async () => ({
+        adapters: [BLEND_ADAPTER],
+        failures: [],
+      }),
+    });
+
+    expect(server.sendTransaction).not.toHaveBeenCalled();
+    expect(result.successes).toEqual([]);
+    expect(result.failures).toEqual([]);
+    expect(result.skipped).toMatchObject([
+      {
+        adapterId: "CADAPTERBLEND",
+        reason: expect.stringContaining("adapter changed since discovery"),
+      },
+    ]);
+  });
+});