diff --git a/api/v1/keepers/accrue.ts b/api/v1/keepers/accrue.ts index 47a12f3f..e0b8f4f1 100644 --- a/api/v1/keepers/accrue.ts +++ b/api/v1/keepers/accrue.ts @@ -24,7 +24,20 @@ export default async function handler(req: VercelRequest, res: VercelResponse) { // *all* traffic, not just correctly-authenticated traffic, if it only ran // after a successful auth check, unauthenticated/wrong-token spam would // be entirely unbounded, since a 401 would return before this ever runs. - if (!(await checkRateLimit(req, res))) return; + // Wrapped, unlike a plain `await`: checkRateLimit talks to Upstash, and an + // outage there would otherwise escape as a bare unhandled 500 with no + // [accrual-keeper] log line, before the run (and its own store-outage + // signalling) ever starts. Fails closed on purpose even though the run + // itself is more tolerant: this is the abuse backstop on an endpoint that + // signs real transactions, and the next scheduled tick retries anyway. + try { + if (!(await checkRateLimit(req, res))) return; + } catch (err) { + console.error("[accrual-keeper] rate limit check failed:", err); + return res + .status(503) + .json({ error: "Rate limiter unavailable; refusing to run" }); + } if (!isCronSecretConfigured()) { return res.status(503).json({ error: "CRON_SECRET is not configured" }); diff --git a/api/v1/keepers/rebalance.ts b/api/v1/keepers/rebalance.ts index 0d842f56..71147c7f 100644 --- a/api/v1/keepers/rebalance.ts +++ b/api/v1/keepers/rebalance.ts @@ -26,7 +26,20 @@ export default async function handler(req: VercelRequest, res: VercelResponse) { // if it only ran after a successful auth check, unauthenticated/wrong-token // spam would be entirely unbounded against an endpoint holding full vault // admin authority. - if (!(await checkRateLimit(req, res))) return; + // Wrapped, unlike a plain `await`: checkRateLimit talks to Upstash, and an + // outage there would otherwise escape as a bare unhandled 500 with no + // [migration-keeper] log line, before the run (and its own store-outage + // signalling) ever starts. Fails closed on purpose even though the run + // itself is more tolerant: this is the abuse backstop on an endpoint that + // signs real transactions, and the next scheduled tick retries anyway. + try { + if (!(await checkRateLimit(req, res))) return; + } catch (err) { + console.error("[migration-keeper] rate limit check failed:", err); + return res + .status(503) + .json({ error: "Rate limiter unavailable; refusing to run" }); + } if (!isCronSecretConfigured()) { return res.status(503).json({ error: "CRON_SECRET is not configured" }); diff --git a/apps/docs/operations/accrual-keeper.md b/apps/docs/operations/accrual-keeper.md index ba1636f3..6cd8def1 100644 --- a/apps/docs/operations/accrual-keeper.md +++ b/apps/docs/operations/accrual-keeper.md @@ -90,11 +90,51 @@ 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:::`) 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. + +Two things differ, both following from what a duplicate `accrue()` actually +costs: it re-syncs a cached value from the adapter's live position and +produces the same result however many times it lands, so at most one extra +Soroban fee, never incorrect accounting. + +- **Fallback.** Where the migration keeper refuses to run on a deployment + without a shared store, this keeper falls back to a per-invocation one and + keeps running. On a deployed environment that fallback is logged as a + warning, not an info line: it reinstates the duplicate-submission gap, and + that should be visible to whoever watches these logs. +- **Store outages.** Where the migration keeper skips a vault it cannot + verify, this keeper proceeds and warns. Halting all accrual for the length + of a KV outage would leave every vault's TVL/APY stale in order to avoid a + duplicate that costs a fee. A record that is readable and says "still in + flight" is still respected; only an unreadable one is proceeded past. + +`deps.submitAccrual` is handed the lease's hooks; an injected submitter that +ignores them loses cross-invocation dedup, and the run warns at startup when +one is in use. + +## 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. diff --git a/apps/docs/operations/environment-variables.md b/apps/docs/operations/environment-variables.md index f1df8711..a8ef1471 100644 --- a/apps/docs/operations/environment-variables.md +++ b/apps/docs/operations/environment-variables.md @@ -8,22 +8,25 @@ ## API: serverless (`api/v1/`) and Fastify (`apps/api-local`) -| Variable | Required | Default | Description | -| ---------------------------------------- | ------------ | ---------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `STELLAR_NETWORK` | No | `"testnet"` | Selects the network the API talks to. Any value other than `"mainnet"` resolves to testnet. Controls which `CONTRACT_ADDRESSES`/`STELLAR_NETWORKS` entry (`packages/shared/src/constants.ts`) is used for every contract call the API makes. | -| `DEFINDEX_VAULT_ID` | No | `""` | Overrides the DeFindex vault contract address at runtime. When empty, the address from `CONTRACT_ADDRESSES.testnet.defindex.vault` in `packages/shared/src/constants.ts` is used. Blend and vault contract addresses are always sourced from constants. | -| `PORT` | No | `3001` | Fastify server port (local dev only). | -| `ALLOWED_ORIGIN` | No | `"https://usemeridian.vercel.app"` | CORS allowed origin for the Fastify server. Set to your frontend domain in production if running Fastify as a standalone server. | -| `REDIS_URL` | No | `""` | Redis URL for `@fastify/rate-limit` in `apps/api-local` (ioredis). Unset: in-memory store (single process); production: set for distributed rate limits. | -| `CRON_SECRET` | Yes | `""` | Bearer token required by scheduled keeper endpoints in production and preview deployments. Only true local dev (no `VERCEL_ENV` set) is permissive without it. | -| `MERIDIAN_KEEPER_SECRET_KEY` | Yes (keeper) | `""` | Stellar secret seed for the funded account that submits Blend `accrue()` transactions. Store in a secrets manager or deployment environment variables; never commit it. | -| `MERIDIAN_KEEPER_MAX_ATTEMPTS` | No | `3` | Maximum attempts per submission. Shared by both the accrue keeper and the migration keeper (`rebalance.ts`), not accrue-specific despite the name; sizing it affects both. | -| `MERIDIAN_KEEPER_RETRY_BASE_DELAY_MS` | No | `1000` | Initial exponential-backoff delay for transient keeper failures. Shared by both the accrue keeper and the migration keeper. | -| `MERIDIAN_KEEPER_RPC_TIMEOUT_MS` | No | `10000` | Timeout for keeper RPC calls, in milliseconds. Shared by both the accrue keeper and the migration keeper. Fully governs submission calls; discovery reads are additionally capped at a hardcoded 10s ceiling shared with the rest of `stellar-sdk-helpers`, so values above `10000` only extend the submission side. | -| `MERIDIAN_MIGRATION_KEEPER_SECRET_KEY` | Yes (keeper) | `""` | Stellar secret seed for the migration keeper. Must be the vault's actual admin address; `migrate_adapter` is admin-gated, unlike the permissionless `accrue()`, so this key carries full vault admin authority. Deliberately separate from `MERIDIAN_KEEPER_SECRET_KEY`. See `apps/docs/operations/migration-keeper.md`. | -| `MERIDIAN_MIGRATION_MAX_SLIPPAGE_BPS` | No | `100` | `max_slippage_bps` passed to every `migrate_adapter` call. The config loader rejects `10000` (unlimited slippage). | -| `MERIDIAN_MIGRATION_MIN_IMPROVEMENT_BPS` | No | `50` | Minimum rate improvement, in basis points, a candidate protocol must clear before the keeper migrates to it. | -| `MERIDIAN_ADAPTER__ID` | No | `""` | Candidate adapter contract address the migration keeper may migrate the vault to, one var per protocol (e.g. `MERIDIAN_ADAPTER_BLEND_ID`, `MERIDIAN_ADAPTER_DEFINDEX_ID`). Not a fixed list: any `` is picked up automatically, adding a new protocol needs no code change. Unset excludes that protocol from consideration, no fallback default. | +| Variable | Required | Default | Description | +| ---------------------------------------- | ------------ | ---------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `STELLAR_NETWORK` | No | `"testnet"` | Selects the network the API talks to. Any value other than `"mainnet"` resolves to testnet. Controls which `CONTRACT_ADDRESSES`/`STELLAR_NETWORKS` entry (`packages/shared/src/constants.ts`) is used for every contract call the API makes. | +| `DEFINDEX_VAULT_ID` | No | `""` | Overrides the DeFindex vault contract address at runtime. When empty, the address from `CONTRACT_ADDRESSES.testnet.defindex.vault` in `packages/shared/src/constants.ts` is used. Blend and vault contract addresses are always sourced from constants. | +| `PORT` | No | `3001` | Fastify server port (local dev only). | +| `ALLOWED_ORIGIN` | No | `"https://usemeridian.vercel.app"` | CORS allowed origin for the Fastify server. Set to your frontend domain in production if running Fastify as a standalone server. | +| `REDIS_URL` | No | `""` | Redis URL for `@fastify/rate-limit` in `apps/api-local` (ioredis). Unset: in-memory store (single process); production: set for distributed rate limits. | +| `UPSTASH_REDIS_REST_URL` | Yes (prod) | `""` | Upstash Redis REST endpoint. Backs distributed rate limiting (`api/_lib/middleware.ts`) and the keepers' cross-invocation submission records. The API refuses to start without it when `VERCEL_ENV=production`; the migration keeper refuses on _any_ deployment, preview included, since preview also signs real transactions. | +| `UPSTASH_REDIS_REST_TOKEN` | Yes (prod) | `""` | Auth token for `UPSTASH_REDIS_REST_URL`. Same requirement and same consumers. | +| `CRON_SECRET` | Yes | `""` | Bearer token required by scheduled keeper endpoints in production and preview deployments. Only true local dev (no `VERCEL_ENV` set) is permissive without it. | +| `MERIDIAN_KEEPER_SECRET_KEY` | Yes (keeper) | `""` | Stellar secret seed for the funded account that submits Blend `accrue()` transactions. Store in a secrets manager or deployment environment variables; never commit it. | +| `MERIDIAN_KEEPER_MAX_ATTEMPTS` | No | `3` | Maximum attempts per submission. Shared by both the accrue keeper and the migration keeper (`rebalance.ts`), not accrue-specific despite the name; sizing it affects both. | +| `MERIDIAN_KEEPER_RETRY_BASE_DELAY_MS` | No | `1000` | Initial exponential-backoff delay for transient keeper failures. Shared by both the accrue keeper and the migration keeper. | +| `MERIDIAN_KEEPER_RPC_TIMEOUT_MS` | No | `10000` | Timeout for keeper RPC calls, in milliseconds. Shared by both the accrue keeper and the migration keeper. Fully governs submission calls; discovery reads are additionally capped at a hardcoded 10s ceiling shared with the rest of `stellar-sdk-helpers`, so values above `10000` only extend the submission side. | +| `MERIDIAN_KEEPER_SUBMISSION_TTL_MS` | No | `360000` | How long a recorded, still-unconfirmed keeper submission keeps blocking a new one for the same target, in milliseconds. Defaults to the 300s transaction validity window plus 60s of clock-skew margin; past it the transaction can never land, so the record is cleared and a retry is allowed. Rejected below `300000`, since a shorter window would expire the record while its transaction can still land. Shared by both keepers. See `apps/docs/operations/migration-keeper.md`. | +| `MERIDIAN_MIGRATION_KEEPER_SECRET_KEY` | Yes (keeper) | `""` | Stellar secret seed for the migration keeper. Must be the vault's actual admin address; `migrate_adapter` is admin-gated, unlike the permissionless `accrue()`, so this key carries full vault admin authority. Deliberately separate from `MERIDIAN_KEEPER_SECRET_KEY`. See `apps/docs/operations/migration-keeper.md`. | +| `MERIDIAN_MIGRATION_MAX_SLIPPAGE_BPS` | No | `100` | `max_slippage_bps` passed to every `migrate_adapter` call. The config loader rejects `10000` (unlimited slippage). | +| `MERIDIAN_MIGRATION_MIN_IMPROVEMENT_BPS` | No | `50` | Minimum rate improvement, in basis points, a candidate protocol must clear before the keeper migrates to it. | +| `MERIDIAN_ADAPTER__ID` | No | `""` | Candidate adapter contract address the migration keeper may migrate the vault to, one var per protocol (e.g. `MERIDIAN_ADAPTER_BLEND_ID`, `MERIDIAN_ADAPTER_DEFINDEX_ID`). Not a fixed list: any `` is picked up automatically, adding a new protocol needs no code change. Unset excludes that protocol from consideration, no fallback default. | ## Deploy scripts (`scripts/`) diff --git a/apps/docs/operations/migration-keeper.md b/apps/docs/operations/migration-keeper.md index cb31a06f..52514ec7 100644 --- a/apps/docs/operations/migration-keeper.md +++ b/apps/docs/operations/migration-keeper.md @@ -180,32 +180,125 @@ 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 lease + +Held in Upstash Redis (`packages/stellar-sdk-helpers/src/keeper-state.ts`), +one record per vault, keyed `meridian:keeper:migration::`. +It is taken in two steps: + +1. **Claim** (`SET NX`) **before the transaction is built.** A plain "is + there a record?" read would not be a claim: two genuinely concurrent + invocations, a scheduled run overlapping a manual `workflow_dispatch`, + could both read nothing and both broadcast. The claim carries no hash + yet, and expires after `DEFAULT_CLAIM_TTL_MS` (60s), which only has to + cover build + simulate + sign. +2. **Record the hash as soon as the transaction is signed**, before + `sendTransaction` is called, with a compare-and-set against the claim. + The hash comes from the signed transaction itself, so a transaction that + reaches the mempool and then times out, or comes back `TRY_AGAIN_LATER`, + is always covered. Recording only after a successful send would miss + exactly the cases that produce duplicates. + +The cost of step 2 is deliberate: a crash between signing and broadcasting +leaves a record for a transaction that never went out. That is bounded, not +a lockup, the record ages out at the transaction's own validity window, +which is exactly when it becomes provably unable to land, so the worst case +is a delayed retry rather than a duplicate migration. + +Every write after the claim is conditional on the exact record this run put +there. Without that, a slow run could clear a record a newer run had already +replaced, handing a third run a clean slate to rebroadcast into. A run that +loses its lease (claim expired, another run took the key) stops touching it +and logs that it did. + +At the start of every run, an existing record is **resolved against the +network**, never trusted on its own word: + +| State of the record | Meaning | Action | +| ----------------------------------------------- | ------------------------------------ | ---------------------------- | +| claim only, inside the claim window | another run is mid-build | **skip this vault this run** | +| claim only, past the claim window | that run died before signing | clear, evaluate again | +| hash, lookup `SUCCESS` | the migration landed | clear, evaluate again | +| hash, lookup `FAILED` | it failed on-chain | clear, retry allowed | +| hash, not found, older than the validity window | provably dead, it can never land now | clear, retry allowed | +| hash, 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. `MERIDIAN_KEEPER_SUBMISSION_TTL_MS` defaults to +`360000` (the 300s transaction validity window plus 60s of clock-skew +margin) and is **rejected below 300000**: a shorter TTL would clear the +record while its transaction could still land, turning the expiry rule into +a duplicate generator. Records also carry a Redis-side expiry, so a run that +dies before it can clear one cannot leave it behind indefinitely. + +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. Every store and status call is time-bounded, so a +black-holed connection can't hang the run past its `maxDuration` budget. + +Because a per-process fallback cannot dedup across invocations at all, the +migration keeper **refuses to run on any deployment** (production _and_ +preview) without `UPSTASH_REDIS_REST_URL`/`UPSTASH_REDIS_REST_TOKEN`. +Preview is included deliberately: preview deployments sign real transactions +off a real key, and `api/_lib/middleware.ts` only fails closed on +production. Local dev falls back to a per-invocation store and says so. + +`deps.submitMigration` is handed the lease's hooks. An injected submitter +that forwards them to `submitKeeperOperation` keeps full dedup; one that +ignores them keeps only the claim and the on-chain re-check, and the run +warns at startup that it is in that state. + +### 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 lease: it covers the case where a migration +already landed and the record has since been cleared or aged out. 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. + +The two keepers also differ on what an unreadable store means, on purpose. +This keeper stops; the accrue keeper proceeds and warns. Stopping accrual +for the length of a KV outage would leave every vault's TVL/APY stale to +avoid a duplicate that costs one Soroban fee, which is the wrong trade in +that direction and the right one here. diff --git a/packages/stellar-sdk-helpers/src/accrual-keeper.test.ts b/packages/stellar-sdk-helpers/src/accrual-keeper.test.ts index d4d1c686..7a44c067 100644 --- a/packages/stellar-sdk-helpers/src/accrual-keeper.test.ts +++ b/packages/stellar-sdk-helpers/src/accrual-keeper.test.ts @@ -99,6 +99,12 @@ import { type KeeperLogger, } from "./accrual-keeper"; import type { KnownPoolMeta } from "./known-pools"; +import { + createInMemoryKeeperStateStore, + submissionStateKey, + type KeeperStateStore, + type SubmissionRecord, +} from "./keeper-state"; const NETWORK = { network: "testnet" as const, @@ -112,6 +118,7 @@ const CONFIG: BlendAccrualKeeperConfig = { maxAttempts: 3, baseDelayMs: 1, rpcTimeoutMs: 100, + submissionTtlMs: 360_000, }; const VAULT: KnownPoolMeta = { @@ -144,6 +151,18 @@ const DEFINDEX_ADAPTER: DiscoveredAdapter = { protocol: "defindex", }; +// Counts only the warnings a test cares about: injecting `submitAccrual` +// also warns once that dedup depends on the injected submitter forwarding +// the hooks, which is unrelated to whatever a given test is asserting. +function warningsMatching(log: KeeperLogger, fragment: string) { + return (log.warn as ReturnType).mock.calls.filter(([message]) => + String(message).includes(fragment) + ); +} + +// Hash of the signed transaction, known before submission. +const SIGNED_HASH = "deadbeef"; + function logger(): KeeperLogger { return { info: vi.fn(), @@ -180,9 +199,19 @@ beforeEach(() => { (sim: { kind?: string }) => sim.kind === "success" ); stellarMocks.assembleTransaction.mockImplementation((tx: unknown) => ({ - build: () => ({ tx, sign: stellarMocks.signPrepared }), + build: () => ({ + tx, + sign: stellarMocks.signPrepared, + // The keeper records the signed transaction's own hash before it is + // ever sent, so the built transaction has to expose one. + hash: () => Buffer.from(SIGNED_HASH, "hex"), + }), })); 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 }); }); @@ -640,7 +669,16 @@ describe("runBlendAccrualKeeper", () => { }); expect(submitAccrual).toHaveBeenCalledOnce(); - expect(submitAccrual).toHaveBeenCalledWith(BLEND_ADAPTER, 1); + // The third argument is this run's submission-lease hooks: an injected + // submitter is expected to forward them to keep cross-invocation dedup. + expect(submitAccrual).toHaveBeenCalledWith( + BLEND_ADAPTER, + 1, + expect.objectContaining({ + onSigned: expect.any(Function), + onResolved: expect.any(Function), + }) + ); expect(result.successes).toEqual([ { vaultId: "meridian-usdc", @@ -732,7 +770,9 @@ describe("runBlendAccrualKeeper", () => { }); expect(submitAccrual).toHaveBeenCalledTimes(2); - expect(log.warn).toHaveBeenCalledOnce(); + expect(warningsMatching(log, "transient failure; retrying")).toHaveLength( + 1 + ); expect(result.failures).toEqual([]); expect(result.successes[0]).toMatchObject({ hash: "HASH2", attempts: 2 }); }); @@ -755,9 +795,13 @@ describe("runBlendAccrualKeeper", () => { submitAccrual, }); - expect(submitAccrual).toHaveBeenNthCalledWith(1, BLEND_ADAPTER, 1); - expect(submitAccrual).toHaveBeenNthCalledWith(2, BLEND_ADAPTER, 2); - expect(submitAccrual).toHaveBeenNthCalledWith(3, BLEND_ADAPTER, 3); + const hooks = expect.objectContaining({ + onSigned: expect.any(Function), + onResolved: expect.any(Function), + }); + expect(submitAccrual).toHaveBeenNthCalledWith(1, BLEND_ADAPTER, 1, hooks); + expect(submitAccrual).toHaveBeenNthCalledWith(2, BLEND_ADAPTER, 2, hooks); + expect(submitAccrual).toHaveBeenNthCalledWith(3, BLEND_ADAPTER, 3, hooks); expect(sleep).toHaveBeenNthCalledWith(1, 1); expect(sleep).toHaveBeenNthCalledWith(2, 2); expect(result.successes[0]).toMatchObject({ @@ -1279,13 +1323,15 @@ describe("runBlendAccrualKeeper", () => { ]); }); - it("retries try-again-later responses from the default transaction path", async () => { + it("rechecks the signed transaction after try-again-later instead of rebroadcasting", async () => { + // TRY_AGAIN_LATER is not "nothing happened": the node may already be + // processing this transaction. Building a second one would give it a + // different hash, and both could land. const sleep = vi.fn(); const server = makeServer({ sendTransaction: vi .fn() - .mockResolvedValueOnce({ status: "TRY_AGAIN_LATER" }) - .mockResolvedValueOnce({ hash: "RETRY_HASH", status: "PENDING" }), + .mockResolvedValueOnce({ status: "TRY_AGAIN_LATER" }), }); stellarMocks.getRpcServer.mockReturnValue(server); stellarMocks.waitForTransaction.mockResolvedValue({ ledger: 654 }); @@ -1299,17 +1345,47 @@ describe("runBlendAccrualKeeper", () => { sleep, }); - expect(server.sendTransaction).toHaveBeenCalledTimes(2); + expect(server.sendTransaction).toHaveBeenCalledOnce(); expect(sleep).toHaveBeenCalledWith(1); + expect(stellarMocks.waitForTransaction).toHaveBeenCalledWith( + server, + SIGNED_HASH, + { timeoutMs: 20000 } + ); expect(result.successes).toMatchObject([ { - hash: "RETRY_HASH", + hash: SIGNED_HASH, ledger: 654, attempts: 2, }, ]); }); + it("tracks the signed hash when the send call itself times out", async () => { + // The send may have reached the network before the client gave up. + const server = makeServer({ + sendTransaction: vi + .fn() + .mockRejectedValueOnce(new Error("Soroban RPC timed out after 100ms")), + }); + stellarMocks.getRpcServer.mockReturnValue(server); + stellarMocks.waitForTransaction.mockResolvedValue({ ledger: 655 }); + + const result = await runBlendAccrualKeeper(CONFIG, { + logger: logger(), + sleep: vi.fn(), + discoverAdapters: async () => ({ + adapters: [BLEND_ADAPTER], + failures: [], + }), + }); + + expect(server.sendTransaction).toHaveBeenCalledOnce(); + expect(result.successes).toMatchObject([ + { hash: SIGNED_HASH, attempts: 2 }, + ]); + }); + it("skips an adapter instead of starting it once the run deadline has passed", async () => { const submitAccrual = vi.fn(); @@ -1367,3 +1443,188 @@ describe("runBlendAccrualKeeper", () => { ]); }); }); + +describe("runBlendAccrualKeeper cross-invocation dedup", () => { + const KEY = submissionStateKey( + "accrual", + "testnet", + BLEND_ADAPTER.vaultId, + BLEND_ADAPTER.adapterId + ); + + async function store(seed?: SubmissionRecord) { + const inner = createInMemoryKeeperStateStore(); + if (seed) await inner.claim(KEY, seed, 600_000); + return inner; + } + + function run(stateStore: KeeperStateStore, log = logger()) { + return runBlendAccrualKeeper(CONFIG, { + logger: log, + sleep: vi.fn(), + stateStore, + discoverAdapters: async () => ({ + adapters: [BLEND_ADAPTER], + failures: [], + }), + }); + } + + it("skips an adapter whose prior accrue() is still unconfirmed instead of sending a second one", async () => { + // The record is the only thing that survives a killed invocation; + // without it the next cron tick would 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 = await store({ + hash: "INFLIGHT_HASH", + updatedAtMs: Date.now() - 1_000, + }); + + const result = await run(stateStore); + + expect(server.sendTransaction).not.toHaveBeenCalled(); + expect(result.successes).toEqual([]); + expect(result.failures).toEqual([]); + expect(result.skipped).toMatchObject([ + { + adapterId: "CADAPTERBLEND", + reason: expect.stringContaining("still unconfirmed"), + }, + ]); + // Left in place: it is still genuinely in flight. + expect(await stateStore.get(KEY)).not.toBeNull(); + }); + + it("skips an adapter another run has claimed but not yet signed for", async () => { + // Two concurrent invocations (a scheduled run overlapping a manual + // workflow_dispatch) would otherwise both read "no record" and both + // broadcast; the claim is what makes the check exclusive. + const server = makeServer(); + stellarMocks.getRpcServer.mockReturnValue(server); + const stateStore = await store({ hash: null, updatedAtMs: Date.now() }); + + const result = await run(stateStore); + + expect(server.sendTransaction).not.toHaveBeenCalled(); + expect(result.skipped).toMatchObject([ + { reason: expect.stringContaining("already preparing") }, + ]); + }); + + 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 = await store({ + hash: "LANDED_HASH", + updatedAtMs: Date.now() - 1_000, + }); + + const result = await run(stateStore); + + 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(await stateStore.get(KEY)).toBeNull(); + }); + + it("ages out a record whose transaction can no longer land, rather than blocking forever", async () => { + const server = makeServer({ + getTransaction: vi.fn(async () => ({ status: "NOT_FOUND" })), + }); + stellarMocks.getRpcServer.mockReturnValue(server); + const stateStore = await store({ + hash: "DEAD_HASH", + updatedAtMs: Date.now() - CONFIG.submissionTtlMs - 1_000, + }); + + const result = await run(stateStore); + + expect(result.successes).toMatchObject([{ hash: "HASH" }]); + expect(server.sendTransaction).toHaveBeenCalledOnce(); + }); + + it("records the signed hash before the transaction is sent, not after", async () => { + // Recording after the send returns would miss the case that matters: + // a send that times out may already have reached the mempool. + let recordedAtSend: string | null | undefined; + const stateStore = await store(); + const server = makeServer({ + getTransaction: vi.fn(async () => ({ status: "NOT_FOUND" })), + sendTransaction: vi.fn(async () => { + recordedAtSend = (await stateStore.get(KEY))?.record.hash; + return { hash: "FRESH_HASH", status: "PENDING" }; + }), + }); + stellarMocks.getRpcServer.mockReturnValue(server); + + await run(stateStore); + + expect(recordedAtSend).toBe(SIGNED_HASH); + expect(await stateStore.get(KEY)).toBeNull(); + }); + + it("proceeds when the store cannot be read, rather than halting all accrual", async () => { + // Deliberately the opposite of the migration keeper: a duplicate + // accrue() re-syncs a cached value and costs one fee, while halting + // leaves every vault's TVL/APY stale for the length of the outage. + const inner = await store(); + const stateStore: KeeperStateStore = { + ...inner, + get: async () => { + throw new Error("KV unavailable"); + }, + }; + const server = makeServer(); + stellarMocks.getRpcServer.mockReturnValue(server); + const log = logger(); + + const result = await run(stateStore, log); + + expect(server.sendTransaction).toHaveBeenCalledOnce(); + expect(result.successes).toHaveLength(1); + expect( + warningsMatching(log, "proceeding without cross-invocation dedup") + ).toHaveLength(1); + }); + + it("releases the claim when nothing was ever signed", async () => { + // A stale adapter aborts before the transaction is built; leaving the + // claim behind would lock the adapter out for the claim's whole window + // over a transaction that never existed. + stellarMocks.simulateView.mockResolvedValue("CADAPTERDEFINDEX_NEW"); + const stateStore = await store(); + const server = makeServer(); + stellarMocks.getRpcServer.mockReturnValue(server); + + await run(stateStore); + + expect(await stateStore.get(KEY)).toBeNull(); + }); + + 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 run(await store()); + + 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"), + }, + ]); + }); +}); diff --git a/packages/stellar-sdk-helpers/src/accrual-keeper.ts b/packages/stellar-sdk-helpers/src/accrual-keeper.ts index 23151475..c4238cef 100644 --- a/packages/stellar-sdk-helpers/src/accrual-keeper.ts +++ b/packages/stellar-sdk-helpers/src/accrual-keeper.ts @@ -5,6 +5,7 @@ import { simulateView } from "./tx"; import type { StellarNetwork } from "./types"; import { consoleLogger, + errorMessage, parsePositiveInt, redactedErrorMessage, retryOutcome, @@ -14,12 +15,23 @@ import { type KeeperLogger, } from "./keeper-retry"; import { + assertAdapterUnchanged, expectString, + isStaleAdapterError, isTransientKeeperError, submitKeeperOperation, SubmissionInFlightError, type KeeperRpcServer, + type KeeperSubmissionHooks, } from "./keeper-tx"; +import { + loadKeeperStateStore, + parseSubmissionTtlMs, + resolvePriorSubmission, + submissionStateKey, + SubmissionLease, + type KeeperStateStore, +} from "./keeper-state"; export type { KeeperFailure, KeeperLogger } from "./keeper-retry"; @@ -60,6 +72,7 @@ export interface BlendAccrualKeeperConfig { maxAttempts: number; baseDelayMs: number; rpcTimeoutMs: number; + submissionTtlMs: number; } export interface DiscoveredAdapter { @@ -118,10 +131,18 @@ export interface BlendAccrualKeeperDeps { adapters: DiscoveredAdapter[]; failures: KeeperFailure[]; }>; + // `hooks` carries this run's submission lease: an override that forwards + // it to submitKeeperOperation keeps cross-invocation dedup; one that + // ignores it falls back to the claim held for the duration of the run, and + // is warned about at run start. submitAccrual?: ( adapter: DiscoveredAdapter, - attempt: number + attempt: number, + hooks: KeeperSubmissionHooks ) => Promise>; + // Cross-invocation submission tracking (#515). Defaults to whatever the + // environment provides (Upstash Redis when configured); injected in tests. + stateStore?: KeeperStateStore; logger?: KeeperLogger; sleep?: (ms: number) => Promise; deadlineAt?: number; @@ -154,6 +175,7 @@ export function loadBlendAccrualKeeperConfig( DEFAULT_RPC_TIMEOUT_MS, "MERIDIAN_KEEPER_RPC_TIMEOUT_MS" ), + submissionTtlMs: parseSubmissionTtlMs(env), }; } @@ -262,12 +284,32 @@ export async function discoverLiveAdapters( return { adapters, failures }; } -function submitAccrualTransaction( +async function submitAccrualTransaction( adapter: DiscoveredAdapter, config: BlendAccrualKeeperConfig, server: KeeperRpcServer, - priorHash?: string + priorHash?: string, + hooks?: KeeperSubmissionHooks ): Promise> { + // The accrue and migration keepers act on the same vault's adapter with no + // coordination between them: 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 accrue() the detached + // one, a silently ineffective call (a detached adapter is still a valid + // contract, so nothing errors) whose yield never reaches the vault. + // Re-reading the vault's live adapter here is the same guard the migration + // keeper already runs before building its own transaction. Skipped when + // rechecking an already-sent transaction (priorHash), which must keep + // tracking that hash rather than re-deciding whether to send it. + if (!priorHash) { + await assertAdapterUnchanged( + server, + adapter.vaultContractId, + config.network.passphrase, + adapter.adapterId + ); + } + return submitKeeperOperation( adapter.adapterId, "accrue", @@ -279,7 +321,8 @@ function submitAccrualTransaction( confirmationTimeoutMs: CONFIRMATION_TIMEOUT_MS, }, server, - priorHash + priorHash, + hooks ); } @@ -292,6 +335,22 @@ export async function runBlendAccrualKeeper( const startedAt = new Date().toISOString(); const deadlineAt = deps.deadlineAt ?? Date.now() + FUNCTION_BUDGET_MS; const server = getRpcServer(config.network.rpcUrl, config.rpcTimeoutMs); + // Same mechanism as the migration keeper's, deliberately: a duplicate + // accrue() only costs a wasted fee, but having both keepers behave + // identically is what makes the execution model reasonable to audit. The + // one difference is the fallback, see loadKeeperStateStore's requireShared. + const stateStore = + deps.stateStore ?? + loadKeeperStateStore(process.env, { + keeper: "accrual", + requireShared: false, + logger, + }); + if (deps.submitAccrual) { + logger.warn( + "[accrual-keeper] submitAccrual is overridden; cross-invocation dedup depends on the injected submitter forwarding the provided hooks" + ); + } const discovery = deps.discoverAdapters ? await deps.discoverAdapters() : await discoverLiveAdapters({ @@ -350,22 +409,97 @@ export async function runBlendAccrualKeeper( }); continue; } - // Scoped to this run only: an unconfirmed hash from a prior invocation - // (e.g. the previous cron tick) is not recoverable here, so a run that - // exhausts its retries mid-confirmation can send a fresh accrue() next - // time. Accepted: accrue() only refreshes a cached value from live - // on-chain state, so a duplicate costs a wasted fee, not bad accounting. + // Resolved against the network, never trusted from the record alone: a + // hash that landed (or failed, or aged past the transaction's validity + // window) clears and lets this run proceed; only a genuinely still-in- + // flight one blocks. See keeper-state.ts. + const stateKey = submissionStateKey( + "accrual", + config.network.network, + adapter.vaultId, + adapter.adapterId + ); + const priorContext = { + vaultId: adapter.vaultId, + adapterId: adapter.adapterId, + }; + const prior = await resolvePriorSubmission({ + store: stateStore, + key: stateKey, + server, + ttlMs: config.submissionTtlMs, + rpcTimeoutMs: config.rpcTimeoutMs, + logger, + context: priorContext, + }); + if (prior.state === "in-flight" || prior.state === "claimed") { + skipped.push({ + vaultId: adapter.vaultId, + vaultContractId: adapter.vaultContractId, + adapterId: adapter.adapterId, + protocol: adapter.protocol, + reason: + prior.state === "in-flight" + ? "a prior accrue() submission is still unconfirmed; skipped to avoid a duplicate" + : "another run is already preparing an accrue() for this adapter; skipped to avoid a duplicate", + }); + logger.warn("[accrual-keeper] skipping adapter; prior submission", { + vaultId: adapter.vaultId, + adapterId: adapter.adapterId, + state: prior.state, + }); + continue; + } + if (prior.state === "unknown") { + // Deliberately fail *open* here, unlike the migration keeper: halting + // all accrual for the length of a store outage would leave every + // vault's TVL/APY stale, which is worse than the duplicate it avoids. + // A duplicate accrue() re-syncs a cached value from live state and + // costs one Soroban fee. The migration keeper's duplicate costs + // slippage twice, which is why only it stops. + logger.warn( + "[accrual-keeper] proceeding without cross-invocation dedup; prior submission state unknown", + { ...priorContext, reason: prior.reason } + ); + } + + // Taken before anything is built, so two concurrent invocations can't + // both read "no record" and both broadcast. + const acquired = await SubmissionLease.acquire({ + store: stateStore, + key: stateKey, + submissionTtlMs: config.submissionTtlMs, + logger, + context: priorContext, + }); + if ("error" in acquired) { + skipped.push({ + vaultId: adapter.vaultId, + vaultContractId: adapter.vaultContractId, + adapterId: adapter.adapterId, + protocol: adapter.protocol, + reason: `could not take the submission lease (${acquired.error}); skipped this run`, + }); + continue; + } + const lease = acquired.lease; + const submissionHooks = lease.hooks; + + // In-run tracking (priorHash) still exists alongside the lease: it's + // what keeps a retry inside this same run rechecking one hash instead of + // re-reading the store on every attempt. let priorHash: string | undefined; try { const result = await withKeeperRetry( (attempt) => deps.submitAccrual - ? deps.submitAccrual(adapter, attempt) + ? deps.submitAccrual(adapter, attempt, submissionHooks) : submitAccrualTransaction( adapter, config, server, - priorHash + priorHash, + submissionHooks ).catch((err: unknown) => { if (err instanceof SubmissionInFlightError) { priorHash = err.sentHash; @@ -402,6 +536,30 @@ export async function runBlendAccrualKeeper( attempts: result.attempts, }); } catch (err) { + if (isStaleAdapterError(err)) { + // The migration keeper moved this vault to a different adapter while + // this run was working. Accruing the detached one would succeed and + // do nothing useful; the new adapter gets picked up by the next run's + // discovery. A benign race, so a skip rather than a failure that + // would page someone. + skipped.push({ + vaultId: adapter.vaultId, + vaultContractId: adapter.vaultContractId, + adapterId: adapter.adapterId, + protocol: adapter.protocol, + reason: + "vault's adapter changed since discovery; skipped to avoid accruing a detached adapter", + }); + logger.info( + "[accrual-keeper] accrue skipped; adapter changed since discovery", + { + vaultId: adapter.vaultId, + adapterId: adapter.adapterId, + detail: errorMessage(err), + } + ); + continue; + } const { attempts, transient } = retryOutcome(err, isTransientKeeperError); const failure: KeeperFailure = { vaultId: adapter.vaultId, @@ -415,6 +573,11 @@ export async function runBlendAccrualKeeper( }; failures.push(failure); logger.error("[accrual-keeper] accrue failed", { ...failure }); + } finally { + // A claim that never became a signed transaction (a stale adapter, a + // simulation error) must not keep the next run out for the claim's + // full window. + await lease.releaseIfUnsent(); } } diff --git a/packages/stellar-sdk-helpers/src/index.ts b/packages/stellar-sdk-helpers/src/index.ts index 1841e127..8f378208 100644 --- a/packages/stellar-sdk-helpers/src/index.ts +++ b/packages/stellar-sdk-helpers/src/index.ts @@ -5,6 +5,7 @@ export * from "./defilamma"; export * from "./defindex"; export * from "./horizon"; export * from "./keeper-retry"; +export * from "./keeper-state"; export * from "./keeper-tx"; export * from "./known-pools"; export * from "./migration-keeper"; diff --git a/packages/stellar-sdk-helpers/src/keeper-state.test.ts b/packages/stellar-sdk-helpers/src/keeper-state.test.ts new file mode 100644 index 00000000..fcdabdbd --- /dev/null +++ b/packages/stellar-sdk-helpers/src/keeper-state.test.ts @@ -0,0 +1,719 @@ +import { describe, expect, it, vi } from "vitest"; + +import { + DEFAULT_CLAIM_TTL_MS, + DEFAULT_SUBMISSION_TTL_MS, + createInMemoryKeeperStateStore, + createUpstashKeeperStateStore, + loadKeeperStateStore, + parseSubmissionTtlMs, + resolvePriorSubmission, + serializeRecord, + submissionStateKey, + SubmissionLease, + type KeeperStateStore, + type SubmissionRecord, +} from "./keeper-state"; +import { TX_VALIDITY_WINDOW_MS } from "./keeper-tx"; +import type { KeeperLogger } from "./keeper-retry"; + +function logger(): KeeperLogger { + return { info: vi.fn(), warn: vi.fn(), error: vi.fn() }; +} + +const KEY = "meridian:keeper:migration:testnet:meridian-usdc"; + +async function seeded(record?: SubmissionRecord) { + const store = createInMemoryKeeperStateStore(); + if (record) await store.claim(KEY, record, 600_000); + return store; +} + +function lookup(response: unknown) { + return { getTransaction: vi.fn(async () => response as never) }; +} + +describe("submissionStateKey", () => { + it("namespaces by keeper and network so records can never be read across either", () => { + // A testnet run blocking a mainnet one, or the accrue keeper reading the + // migration keeper's record, would both be silent and confusing. + expect(submissionStateKey("migration", "testnet", "meridian-usdc")).toBe( + KEY + ); + expect( + submissionStateKey("accrual", "mainnet", "meridian-usdc", "CADAPTER") + ).toBe("meridian:keeper:accrual:mainnet:meridian-usdc:CADAPTER"); + }); +}); + +describe("parseSubmissionTtlMs", () => { + it("defaults to the transaction validity window plus clock-skew margin", () => { + expect(parseSubmissionTtlMs({})).toBe(DEFAULT_SUBMISSION_TTL_MS); + expect(DEFAULT_SUBMISSION_TTL_MS).toBeGreaterThan(TX_VALIDITY_WINDOW_MS); + }); + + it("reads an operator override at or above the validity window", () => { + expect( + parseSubmissionTtlMs({ MERIDIAN_KEEPER_SUBMISSION_TTL_MS: "600000" }) + ).toBe(600_000); + }); + + it("rejects a TTL shorter than the transaction's own validity window", () => { + // Anything shorter turns "aged out, so provably dead" into a duplicate + // generator: the record clears while the transaction can still land. + expect(() => + parseSubmissionTtlMs({ MERIDIAN_KEEPER_SUBMISSION_TTL_MS: "90000" }) + ).toThrow(/must be at least 300000/); + }); + + it("rejects a non-positive override", () => { + expect(() => + parseSubmissionTtlMs({ MERIDIAN_KEEPER_SUBMISSION_TTL_MS: "0" }) + ).toThrow(/must be a positive integer/); + }); +}); + +describe("resolvePriorSubmission", () => { + it("reports none when nothing was recorded", async () => { + const result = await resolvePriorSubmission({ + store: await seeded(), + key: KEY, + server: lookup({ status: "NOT_FOUND" }), + ttlMs: DEFAULT_SUBMISSION_TTL_MS, + logger: logger(), + }); + + expect(result).toEqual({ state: "none" }); + }); + + it("clears the record when the recorded transaction confirmed successfully", async () => { + const store = await seeded({ hash: "HASH", updatedAtMs: Date.now() }); + + const result = await resolvePriorSubmission({ + store, + key: KEY, + server: lookup({ status: "SUCCESS", ledger: 42 }), + ttlMs: DEFAULT_SUBMISSION_TTL_MS, + logger: logger(), + }); + + expect(result).toEqual({ state: "landed", hash: "HASH", ledger: 42 }); + expect(await store.get(KEY)).toBeNull(); + }); + + it("clears the record and allows an immediate retry when the transaction failed on-chain", async () => { + const store = await seeded({ hash: "HASH", updatedAtMs: Date.now() }); + + const result = await resolvePriorSubmission({ + store, + key: KEY, + server: lookup({ status: "FAILED" }), + ttlMs: DEFAULT_SUBMISSION_TTL_MS, + logger: logger(), + }); + + expect(result).toEqual({ state: "failed", hash: "HASH" }); + expect(await store.get(KEY)).toBeNull(); + }); + + it("keeps blocking while an unfound transaction is still inside its validity window", async () => { + const now = 1_000_000; + const store = await seeded({ hash: "HASH", updatedAtMs: now - 5_000 }); + + const result = await resolvePriorSubmission({ + store, + key: KEY, + server: lookup({ status: "NOT_FOUND" }), + ttlMs: DEFAULT_SUBMISSION_TTL_MS, + logger: logger(), + now, + }); + + expect(result).toEqual({ state: "in-flight", hash: "HASH", ageMs: 5_000 }); + expect(await store.get(KEY)).not.toBeNull(); + }); + + it("ages out an unfound transaction that can no longer land, so nothing waits on a human", async () => { + const now = 1_000_000; + const store = await seeded({ + hash: "HASH", + updatedAtMs: now - DEFAULT_SUBMISSION_TTL_MS - 1, + }); + + const result = await resolvePriorSubmission({ + store, + key: KEY, + server: lookup({ status: "NOT_FOUND" }), + ttlMs: DEFAULT_SUBMISSION_TTL_MS, + logger: logger(), + now, + }); + + expect(result).toEqual({ state: "expired", hash: "HASH" }); + expect(await store.get(KEY)).toBeNull(); + }); + + it("reports a hash-less claim as claimed, without touching the network", async () => { + const server = lookup({ status: "NOT_FOUND" }); + const now = 1_000_000; + + const result = await resolvePriorSubmission({ + store: await seeded({ hash: null, updatedAtMs: now - 1_000 }), + key: KEY, + server, + ttlMs: DEFAULT_SUBMISSION_TTL_MS, + logger: logger(), + now, + }); + + expect(result).toEqual({ state: "claimed", ageMs: 1_000 }); + // Nothing was signed, so there is no hash to ask the network about. + expect(server.getTransaction).not.toHaveBeenCalled(); + }); + + it("ages a stale claim out on the short claim window, not the submission window", async () => { + // A run that died mid-build never signed anything, so there is no + // transaction that could still land; blocking for the full submission + // TTL would be five minutes of nothing. + const now = 1_000_000; + const store = await seeded({ + hash: null, + updatedAtMs: now - DEFAULT_CLAIM_TTL_MS - 1, + }); + + const result = await resolvePriorSubmission({ + store, + key: KEY, + server: lookup({ status: "NOT_FOUND" }), + ttlMs: DEFAULT_SUBMISSION_TTL_MS, + logger: logger(), + now, + }); + + expect(result).toEqual({ state: "expired", hash: null }); + expect(await store.get(KEY)).toBeNull(); + }); + + it("treats an unreadable store as unknown, never as 'nothing was submitted'", async () => { + const log = logger(); + const store: KeeperStateStore = { + ...(await seeded()), + get: async () => { + throw new Error("KV unavailable"); + }, + }; + + const result = await resolvePriorSubmission({ + store, + key: KEY, + server: lookup({ status: "NOT_FOUND" }), + ttlMs: DEFAULT_SUBMISSION_TTL_MS, + logger: log, + }); + + expect(result).toMatchObject({ state: "unknown" }); + expect(log.warn).toHaveBeenCalledWith( + "[keeper-state] could not read prior submission record", + expect.objectContaining({ error: "KV unavailable" }) + ); + }); + + it("treats a failed status lookup as unknown rather than assuming the transaction is dead", async () => { + const store = await seeded({ hash: "HASH", updatedAtMs: Date.now() }); + + const result = await resolvePriorSubmission({ + store, + key: KEY, + server: { + getTransaction: vi.fn(async () => { + throw new Error("rpc unavailable"); + }), + }, + ttlMs: DEFAULT_SUBMISSION_TTL_MS, + logger: logger(), + }); + + expect(result).toMatchObject({ state: "unknown" }); + // Still recorded: the run couldn't prove anything either way. + expect(await store.get(KEY)).not.toBeNull(); + }); + + it("bounds the status lookup instead of hanging the run on a black-holed connection", async () => { + const result = await resolvePriorSubmission({ + store: await seeded({ hash: "HASH", updatedAtMs: Date.now() }), + key: KEY, + server: { getTransaction: () => new Promise(() => undefined) }, + ttlMs: DEFAULT_SUBMISSION_TTL_MS, + rpcTimeoutMs: 5, + logger: logger(), + }); + + expect(result).toMatchObject({ state: "unknown" }); + }); + + it("blocks on an unrecognised status instead of treating it as resolved", async () => { + const result = await resolvePriorSubmission({ + store: await seeded({ hash: "HASH", updatedAtMs: Date.now() }), + key: KEY, + server: lookup({ status: "PENDING_SOMETHING_NEW" }), + ttlMs: DEFAULT_SUBMISSION_TTL_MS, + logger: logger(), + }); + + expect(result).toMatchObject({ state: "in-flight" }); + }); + + it("leaves a record another run has replaced alone instead of clearing it", async () => { + // The race this prevents: run A resolves an old hash as failed and + // clears the key, run B has since written a new hash there, A's clear + // wipes it, and run C sees a clean slate and rebroadcasts. + const store = await seeded({ hash: "OLD_HASH", updatedAtMs: Date.now() }); + const log = logger(); + const server = { + getTransaction: vi.fn(async () => { + // B writes a newer record while A's lookup is in flight. + const current = await store.get(KEY); + await store.replace( + KEY, + { hash: "NEW_HASH", updatedAtMs: Date.now() }, + 600_000, + current!.revision + ); + return { status: "FAILED" } as never; + }), + }; + + const result = await resolvePriorSubmission({ + store, + key: KEY, + server, + ttlMs: DEFAULT_SUBMISSION_TTL_MS, + logger: log, + }); + + expect(result).toMatchObject({ state: "failed", hash: "OLD_HASH" }); + expect((await store.get(KEY))?.record.hash).toBe("NEW_HASH"); + expect(log.info).toHaveBeenCalledWith( + "[keeper-state] record changed before it could be cleared", + expect.any(Object) + ); + }); +}); + +describe("SubmissionLease", () => { + async function acquire(store: KeeperStateStore, log = logger()) { + return SubmissionLease.acquire({ + store, + key: KEY, + submissionTtlMs: DEFAULT_SUBMISSION_TTL_MS, + logger: log, + }); + } + + it("takes the target exclusively, so a concurrent run cannot also claim it", async () => { + const store = await seeded(); + + const first = await acquire(store); + const second = await acquire(store); + + expect("lease" in first).toBe(true); + expect(second).toMatchObject({ + error: expect.stringContaining("already holds"), + }); + }); + + it("refuses the lease when the store is unreachable", async () => { + const store: KeeperStateStore = { + ...(await seeded()), + claim: async () => { + throw new Error("KV unavailable"); + }, + }; + + expect(await acquire(store)).toMatchObject({ + error: "submission state store unavailable", + }); + }); + + it("upgrades the claim to the signed hash and clears it once resolved", async () => { + const store = await seeded(); + const acquired = await acquire(store); + if (!("lease" in acquired)) throw new Error("expected a lease"); + + await acquired.lease.hooks.onSigned?.("SIGNED"); + expect((await store.get(KEY))?.record.hash).toBe("SIGNED"); + + await acquired.lease.hooks.onResolved?.("SIGNED"); + expect(await store.get(KEY)).toBeNull(); + }); + + it("releases a claim that never became a signed transaction", async () => { + const store = await seeded(); + const acquired = await acquire(store); + if (!("lease" in acquired)) throw new Error("expected a lease"); + + await acquired.lease.releaseIfUnsent(); + + expect(await store.get(KEY)).toBeNull(); + }); + + it("keeps a signed transaction's record when the run ends, rather than releasing it", async () => { + const store = await seeded(); + const acquired = await acquire(store); + if (!("lease" in acquired)) throw new Error("expected a lease"); + + await acquired.lease.hooks.onSigned?.("SIGNED"); + await acquired.lease.releaseIfUnsent(); + + expect((await store.get(KEY))?.record.hash).toBe("SIGNED"); + }); + + it("stops touching the key after losing the lease to another run", async () => { + // The claim expired and another run took the key: this run must not + // overwrite or clear what now belongs to someone else. + const store = await seeded(); + const log = logger(); + const acquired = await acquire(store, log); + if (!("lease" in acquired)) throw new Error("expected a lease"); + + const stolen: SubmissionRecord = { hash: "OTHER", updatedAtMs: Date.now() }; + const current = await store.get(KEY); + await store.replace(KEY, stolen, 600_000, current!.revision); + + await acquired.lease.hooks.onSigned?.("SIGNED"); + expect((await store.get(KEY))?.record.hash).toBe("OTHER"); + expect(log.warn).toHaveBeenCalledWith( + "[keeper-state] lost the submission lease", + expect.any(Object) + ); + + await acquired.lease.hooks.onResolved?.("SIGNED"); + expect((await store.get(KEY))?.record.hash).toBe("OTHER"); + }); + + it("never throws when the store write fails, since the transaction is already signed", async () => { + // Throwing here would surface as a submission error, and the retry loop + // answers those by broadcasting a second transaction. + const log = logger(); + const inner = await seeded(); + const store: KeeperStateStore = { + ...inner, + replace: async () => { + throw new Error("KV write failed"); + }, + deleteIf: async () => { + throw new Error("KV delete failed"); + }, + }; + const acquired = await acquire(store, log); + if (!("lease" in acquired)) throw new Error("expected a lease"); + + await expect( + acquired.lease.hooks.onSigned?.("SIGNED") + ).resolves.toBeUndefined(); + await expect( + acquired.lease.hooks.onResolved?.("SIGNED") + ).resolves.toBeUndefined(); + expect(log.warn).toHaveBeenCalledTimes(2); + }); + + it("retries recording the signed hash instead of giving up on the first transient failure", async () => { + // A swallowed one-shot failure here would leave the claim (short TTL, no + // hash) as the only record of an actually in-flight transaction, aging + // out well before the transaction itself could no longer land. + const log = logger(); + const inner = await seeded(); + let attempts = 0; + const store: KeeperStateStore = { + ...inner, + replace: async (...args) => { + attempts += 1; + if (attempts < 3) throw new Error("KV write failed"); + return inner.replace(...args); + }, + }; + const acquired = await acquire(store, log); + if (!("lease" in acquired)) throw new Error("expected a lease"); + + await acquired.lease.hooks.onSigned?.("SIGNED"); + + expect(attempts).toBe(3); + expect((await store.get(KEY))?.record.hash).toBe("SIGNED"); + expect(log.warn).not.toHaveBeenCalled(); + }); +}); + +describe("createInMemoryKeeperStateStore", () => { + it("expires a record once its TTL has passed", async () => { + vi.useFakeTimers(); + try { + const store = createInMemoryKeeperStateStore(); + await store.claim(KEY, { hash: "HASH", updatedAtMs: Date.now() }, 1_000); + expect(await store.get(KEY)).not.toBeNull(); + vi.advanceTimersByTime(1_001); + expect(await store.get(KEY)).toBeNull(); + } finally { + vi.useRealTimers(); + } + }); + + it("rejects a conditional write whose expected revision no longer matches", async () => { + const store = createInMemoryKeeperStateStore(); + const claimed = await store.claim( + KEY, + { hash: null, updatedAtMs: 1 }, + 1_000 + ); + + await store.replace( + KEY, + { hash: "A", updatedAtMs: 2 }, + 1_000, + claimed!.revision + ); + + expect( + await store.replace( + KEY, + { hash: "B", updatedAtMs: 3 }, + 1_000, + claimed!.revision + ) + ).toBeNull(); + expect(await store.deleteIf(KEY, claimed!.revision)).toBe(false); + expect((await store.get(KEY))?.record.hash).toBe("A"); + }); + + it("ignores a stored value that isn't a usable record", async () => { + const store = createInMemoryKeeperStateStore(); + await store.claim(KEY, { hash: 7 as never, updatedAtMs: 1 }, 1_000); + expect(await store.get(KEY)).toBeNull(); + }); +}); + +describe("createUpstashKeeperStateStore", () => { + function fetchMock(response: unknown, ok = true, status = 200) { + return vi.fn(async () => ({ + ok, + status, + json: async () => response, + })) as unknown as typeof fetch; + } + + const RECORD: SubmissionRecord = { hash: "HASH", updatedAtMs: 5 }; + + it("reads a record back through the REST API", async () => { + const fetchImpl = fetchMock({ result: serializeRecord(RECORD) }); + const store = createUpstashKeeperStateStore({ + url: "https://redis.example/", + token: "tok", + fetchImpl, + }); + + expect((await store.get(KEY))?.record).toEqual(RECORD); + expect(fetchImpl).toHaveBeenCalledWith( + // Trailing slash trimmed, so the command never posts to a double slash. + "https://redis.example", + expect.objectContaining({ + method: "POST", + body: JSON.stringify(["GET", KEY]), + headers: expect.objectContaining({ Authorization: "Bearer tok" }), + }) + ); + }); + + it("claims with SET NX and a millisecond expiry, and reports a lost race", async () => { + const taken = createUpstashKeeperStateStore({ + url: "https://redis.example", + token: "tok", + // Upstash returns null for a SET NX that didn't apply. + fetchImpl: fetchMock({ result: null }), + }); + expect(await taken.claim(KEY, RECORD, 1_500)).toBeNull(); + + const fetchImpl = fetchMock({ result: "OK" }); + const free = createUpstashKeeperStateStore({ + url: "https://redis.example", + token: "tok", + fetchImpl, + }); + expect(await free.claim(KEY, RECORD, 1_500)).toMatchObject({ + record: RECORD, + }); + expect(fetchImpl).toHaveBeenCalledWith( + "https://redis.example", + expect.objectContaining({ + body: JSON.stringify([ + "SET", + KEY, + serializeRecord(RECORD), + "NX", + "PX", + 1500, + ]), + }) + ); + }); + + it("makes replace and delete conditional on the stored value, not just the key", async () => { + // A plain SET/DEL would let a slow run clobber a newer run's record. + const fetchImpl = fetchMock({ result: 1 }); + const store = createUpstashKeeperStateStore({ + url: "https://redis.example", + token: "tok", + fetchImpl, + }); + + await store.replace(KEY, RECORD, 1_500, "OLD"); + await store.deleteIf(KEY, "OLD"); + + const bodies = ( + fetchImpl as unknown as ReturnType + ).mock.calls.map(([, init]) => String((init as RequestInit).body)); + expect(bodies[0]).toContain('"EVAL"'); + expect(bodies[0]).toContain("OLD"); + expect(bodies[1]).toContain('"EVAL"'); + expect(bodies[1]).toContain("OLD"); + }); + + it("reports a conditional write that lost the race", async () => { + const store = createUpstashKeeperStateStore({ + url: "https://redis.example", + token: "tok", + fetchImpl: fetchMock({ result: 0 }), + }); + + expect(await store.replace(KEY, RECORD, 1_500, "OLD")).toBeNull(); + expect(await store.deleteIf(KEY, "OLD")).toBe(false); + }); + + it("treats an unparseable or malformed stored value as no record", async () => { + for (const result of [ + "not json", + JSON.stringify({ hash: 7, updatedAtMs: 1 }), + JSON.stringify({ hash: "H" }), + null, + ]) { + const store = createUpstashKeeperStateStore({ + url: "https://redis.example", + token: "tok", + fetchImpl: fetchMock({ result }), + }); + expect(await store.get(KEY)).toBeNull(); + } + }); + + it("reports an HTTP failure by status alone, never echoing the credential", async () => { + const store = createUpstashKeeperStateStore({ + url: "https://redis.example", + token: "super-secret-token", + fetchImpl: fetchMock({}, false, 503), + }); + + await expect(store.get(KEY)).rejects.toThrow( + "Upstash Redis request failed with HTTP 503" + ); + await expect(store.get(KEY)).rejects.not.toThrow(/super-secret-token/); + }); + + it("surfaces a Redis-level error response", async () => { + const store = createUpstashKeeperStateStore({ + url: "https://redis.example", + token: "tok", + fetchImpl: fetchMock({ error: "WRONGTYPE" }), + }); + + await expect(store.get(KEY)).rejects.toThrow( + "Upstash Redis error: WRONGTYPE" + ); + }); + + it("bounds a hung request instead of stalling the run past its budget", async () => { + // The worst moment for an unbounded KV call is right after a + // transaction was broadcast. + const store = createUpstashKeeperStateStore({ + url: "https://redis.example", + token: "tok", + fetchImpl: (() => + new Promise(() => undefined)) as unknown as typeof fetch, + timeoutMs: 5, + }); + + await expect(store.get(KEY)).rejects.toThrow(/timed out/); + }); +}); + +describe("loadKeeperStateStore", () => { + it("uses Upstash when the same credentials the rate limiter uses are present", async () => { + const fetchImpl = vi.fn(async () => ({ + ok: true, + status: 200, + json: async () => ({ result: null }), + })) as unknown as typeof fetch; + + const store = loadKeeperStateStore( + { + UPSTASH_REDIS_REST_URL: "https://redis.example", + UPSTASH_REDIS_REST_TOKEN: "tok", + }, + { keeper: "migration", requireShared: true, logger: logger(), fetchImpl } + ); + await store.get(KEY); + + expect(fetchImpl).toHaveBeenCalledOnce(); + }); + + it("refuses to run the migration keeper on any deployment without a shared store", () => { + // Preview counts: preview deployments sign real transactions off a real + // key, and middleware.ts only fails closed on production, so this is + // the guard that actually covers preview. + for (const env of ["production", "preview"]) { + expect(() => + loadKeeperStateStore( + { VERCEL_ENV: env }, + { keeper: "migration", requireShared: true, logger: logger() } + ) + ).toThrow( + /UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN are required/ + ); + } + }); + + it("warns, not just informs, when a deployed accrue keeper falls back", () => { + // Falling back reinstates the duplicate-submission gap; that belongs in + // logs someone can alert on, not buried at info level. + const log = logger(); + loadKeeperStateStore( + { VERCEL_ENV: "production" }, + { keeper: "accrual", requireShared: false, logger: log } + ); + + expect(log.warn).toHaveBeenCalledWith( + expect.stringContaining("cross-invocation dedup is inactive"), + { store: "in-memory", env: "production" } + ); + }); + + it("stays quiet at info level in local dev, where the fallback is expected", () => { + const log = logger(); + loadKeeperStateStore( + {}, + { keeper: "migration", requireShared: true, logger: log } + ); + + expect(log.warn).not.toHaveBeenCalled(); + expect(log.info).toHaveBeenCalledWith( + expect.stringContaining("cross-invocation dedup is inactive"), + { store: "in-memory" } + ); + }); + + it("ignores blank credentials rather than building a store that cannot work", () => { + const log = logger(); + loadKeeperStateStore( + { UPSTASH_REDIS_REST_URL: " ", UPSTASH_REDIS_REST_TOKEN: "tok" }, + { keeper: "accrual", requireShared: false, logger: log } + ); + expect(log.info).toHaveBeenCalled(); + }); +}); diff --git a/packages/stellar-sdk-helpers/src/keeper-state.ts b/packages/stellar-sdk-helpers/src/keeper-state.ts new file mode 100644 index 00000000..7701b3f0 --- /dev/null +++ b/packages/stellar-sdk-helpers/src/keeper-state.ts @@ -0,0 +1,658 @@ +// Cross-invocation submission tracking for scheduled keepers (#515). +// +// keeper-tx.ts's `priorHash` only lives inside a single invocation: if the +// process is killed (or a run exhausts its retries) while a transaction is +// sent but unconfirmed, the next cron tick has no memory of it. For +// `accrue()` that costs a wasted fee; for `migrate_adapter` it costs real +// slippage twice, since each call is its own slippage-bounded transaction. +// +// The unit of state is a lease on one keeper target, taken in two steps: +// +// 1. `SET NX` a hash-less claim *before* the transaction is built. This is +// what makes "no record" mean "nobody is working on this target": a +// plain read beforehand would let two genuinely concurrent invocations +// (a scheduled run overlapping a manual `workflow_dispatch`) both see +// nothing and both broadcast. Claims are short-lived, so a crash during +// build costs a bounded delay, not a stuck target. +// 2. Compare-and-set the real transaction hash into that claim as soon as +// the transaction is *signed*, before `sendTransaction` is called. The +// hash is derivable from the signed transaction, so a transaction that +// reaches the mempool and then times out, or comes back +// TRY_AGAIN_LATER, is always covered by a record. The cost is that a +// crash between signing and broadcasting leaves a record for a +// transaction that never went out; that record ages out at the +// transaction's own validity window, which is exactly when it becomes +// provably unable to land. +// +// Every write after the claim is conditional on the exact record this run +// put there (`revision`), so a slow run can never clobber or clear a newer +// run's record and hand a third run a clean slate to rebroadcast into. +// +// A record is never trusted on its word: every run resolves it by looking +// the hash up on-network, so "still unconfirmed" is an observed answer, not +// an assumption. See apps/docs/operations/migration-keeper.md for the state +// machine. + +import { withRaceTimeout, withRetry } from "@meridian/shared"; +import { + errorMessage, + parsePositiveInt, + type KeeperLogger, +} from "./keeper-retry"; +import { TX_VALIDITY_WINDOW_MS, type KeeperSubmissionHooks } from "./keeper-tx"; + +// A submitted transaction can never land more than TX_VALIDITY_WINDOW_MS +// after it was built (keeper-tx.ts builds with that as its `setTimeout`). +// The extra 60s is margin for clock skew between this process and the +// network, and for the gap between building and broadcasting. +export const DEFAULT_SUBMISSION_TTL_MS = TX_VALIDITY_WINDOW_MS + 60_000; + +// How long an unupgraded claim (no hash yet) blocks the target. Only has to +// cover build + simulate + sign, which is bounded by the keepers' own +// function budget; deliberately far shorter than the submission TTL, since +// a claim that never became a transaction is blocking for nothing. +export const DEFAULT_CLAIM_TTL_MS = 60_000; + +// Every store round trip is bounded: an unbounded KV call can stall a run +// past the platform's maxDuration, and the worst moment for that is right +// after a transaction was broadcast. +export const DEFAULT_STORE_TIMEOUT_MS = 5_000; + +// Recording a signed transaction's hash (SubmissionLease.write) is retried +// on a transient store failure rather than swallowed on the first error: a +// swallowed failure here leaves the claim (short TTL, no hash) as the only +// record of a transaction that is actually in flight, aging out on the +// claim window instead of the transaction's real validity window and +// letting a concurrent run rebroadcast it. +const WRITE_RETRY_ATTEMPTS = 3; +const WRITE_RETRY_BASE_DELAY_MS = 200; + +export interface SubmissionRecord { + // null while a run has claimed the target but has not signed a + // transaction for it yet. + hash: string | null; + updatedAtMs: number; +} + +// The record plus the exact serialized bytes it was read/written as, which +// every conditional write compares against. Without it, "delete this key" +// is unconditional and races: run A resolving an old hash could clear the +// record run B had already replaced it with, leaving run C free to +// rebroadcast. +export interface StoredRecord { + record: SubmissionRecord; + revision: string; +} + +export interface KeeperStateStore { + get(key: string): Promise; + /** SET NX. Returns the stored claim on success, null if someone else holds the key. */ + claim( + key: string, + record: SubmissionRecord, + ttlMs: number + ): Promise; + /** Compare-and-set against `expectedRevision`. Null when the stored value moved on. */ + replace( + key: string, + record: SubmissionRecord, + ttlMs: number, + expectedRevision: string + ): Promise; + /** Compare-and-delete against `expectedRevision`. False when the stored value moved on. */ + deleteIf(key: string, expectedRevision: string): Promise; +} + +// Structural rather than `Pick` so this module +// never depends on the SDK's enum objects, which the keeper tests mock away. +export interface KeeperTxLookup { + getTransaction( + hash: string + ): Promise<{ status?: string; ledger?: number } | null | undefined>; +} + +export type PriorSubmission = + | { state: "none" } + | { state: "landed"; hash: string; ledger?: number } + | { state: "failed"; hash: string } + | { state: "expired"; hash: string | null } + | { state: "in-flight"; hash: string; ageMs: number } + // Another run has claimed this target but has not signed anything yet. + | { state: "claimed"; ageMs: number } + // The store or the RPC lookup itself failed, so whether a prior + // submission is still in flight is unknown. Deliberately distinct from + // "none": treating an unreadable store as "nothing was submitted" would + // turn a KV outage into exactly the duplicate submission this module + // exists to prevent. What a keeper does with it depends on what its own + // duplicate costs, see `blockOnUnknown` in each keeper. + | { state: "unknown"; reason: string }; + +export function serializeRecord(record: SubmissionRecord): string { + return JSON.stringify({ hash: record.hash, updatedAtMs: record.updatedAtMs }); +} + +export function parseSubmissionTtlMs( + env: Record +): number { + const ttlMs = parsePositiveInt( + env.MERIDIAN_KEEPER_SUBMISSION_TTL_MS, + DEFAULT_SUBMISSION_TTL_MS, + "MERIDIAN_KEEPER_SUBMISSION_TTL_MS" + ); + // A TTL shorter than the transaction's own validity window turns the + // "aged out, so provably dead" expiry into a duplicate generator: the + // record would be cleared while the original transaction can still land. + if (ttlMs < TX_VALIDITY_WINDOW_MS) { + throw new Error( + `MERIDIAN_KEEPER_SUBMISSION_TTL_MS must be at least ${TX_VALIDITY_WINDOW_MS} (a submitted transaction stays valid that long, so a shorter record would expire while it can still land)` + ); + } + return ttlMs; +} + +/** + * Key for one keeper target's submission lease. Namespaced by keeper and + * network so the accrue and migration keepers can never read each other's + * records, and so a testnet run can never block a mainnet one. + */ +export function submissionStateKey( + keeper: "accrual" | "migration", + network: string, + ...target: string[] +): string { + return ["meridian", "keeper", keeper, network, ...target].join(":"); +} + +/** + * Resolves whatever the store holds for `key` against the network, clearing + * the record whenever the underlying transaction's fate becomes known. + * + * Never throws: a keeper's dedup check failing must not take the run down + * with it, so a store or lookup failure surfaces as `unknown` for the caller + * to decide about. + */ +export async function resolvePriorSubmission(options: { + store: KeeperStateStore; + key: string; + server: KeeperTxLookup; + ttlMs: number; + logger: KeeperLogger; + claimTtlMs?: number; + rpcTimeoutMs?: number; + context?: Record; + now?: number; +}): Promise { + const { store, key, server, ttlMs, logger } = options; + const claimTtlMs = options.claimTtlMs ?? DEFAULT_CLAIM_TTL_MS; + const rpcTimeoutMs = options.rpcTimeoutMs ?? DEFAULT_STORE_TIMEOUT_MS; + const context = options.context ?? {}; + const now = options.now ?? Date.now(); + + let stored: StoredRecord | null; + try { + stored = await store.get(key); + } catch (err) { + logger.warn("[keeper-state] could not read prior submission record", { + ...context, + error: errorMessage(err), + }); + return { state: "unknown", reason: "submission state store unavailable" }; + } + if (!stored) return { state: "none" }; + + const { record, revision } = stored; + const ageMs = now - record.updatedAtMs; + + // A claim with no hash: another run is mid-build, or died mid-build. It + // ages out on the much shorter claim window, since nothing was ever + // signed and there is no transaction that could still land. + if (record.hash === null) { + if (ageMs > claimTtlMs) { + await clearRecord(store, key, revision, logger, context); + return { state: "expired", hash: null }; + } + return { state: "claimed", ageMs }; + } + + let lookup: { status?: string; ledger?: number } | null | undefined; + try { + lookup = await withRaceTimeout( + () => server.getTransaction(record.hash as string), + rpcTimeoutMs, + "Soroban RPC" + ); + } catch (err) { + logger.warn("[keeper-state] could not look up prior submission", { + ...context, + hash: record.hash, + error: errorMessage(err), + }); + return { + state: "unknown", + reason: "prior submission status could not be checked", + }; + } + + const status = lookup?.status; + if (status === "SUCCESS") { + await clearRecord(store, key, revision, logger, context); + return { + state: "landed", + hash: record.hash, + ...(lookup?.ledger !== undefined && { ledger: lookup.ledger }), + }; + } + if (status === "FAILED") { + await clearRecord(store, key, revision, logger, context); + return { state: "failed", hash: record.hash }; + } + + // NOT_FOUND (or any status this client doesn't recognise): the network has + // no opinion yet. Age it out against the transaction's own validity window + // rather than waiting on a human, so a record can never block forever. + if (ageMs > ttlMs) { + await clearRecord(store, key, revision, logger, context); + return { state: "expired", hash: record.hash }; + } + return { state: "in-flight", hash: record.hash, ageMs }; +} + +/** Conditional clear. Never throws; the store's own TTL is the backstop. */ +async function clearRecord( + store: KeeperStateStore, + key: string, + revision: string, + logger: KeeperLogger, + context: Record +): Promise { + try { + const cleared = await store.deleteIf(key, revision); + if (!cleared) { + // Benign and worth seeing: another run replaced the record between + // this run reading it and clearing it. Leaving it alone is the whole + // point of the conditional delete. + logger.info("[keeper-state] record changed before it could be cleared", { + ...context, + }); + } + return cleared; + } catch (err) { + logger.warn("[keeper-state] could not clear submission record", { + ...context, + error: errorMessage(err), + }); + return false; + } +} + +/** + * One run's exclusive hold on one keeper target, from before the + * transaction is built until its fate is known. + * + * Every method is failure-tolerant by design: once a transaction is signed, + * a store error must never surface as a submission error, because the retry + * loop answers those by broadcasting a second transaction, exactly the + * duplicate this exists to prevent. + */ +export class SubmissionLease { + private held: StoredRecord | null; + + private constructor( + private readonly store: KeeperStateStore, + private readonly key: string, + private readonly submissionTtlMs: number, + private readonly logger: KeeperLogger, + private readonly context: Record, + claim: StoredRecord + ) { + this.held = claim; + } + + /** + * Takes the lease with SET NX. Returns null when another run already holds + * the target (the caller skips it) or when the store is unreachable and + * `blockOnUnknown` says a duplicate is not worth risking. + */ + static async acquire(options: { + store: KeeperStateStore; + key: string; + submissionTtlMs: number; + logger: KeeperLogger; + claimTtlMs?: number; + context?: Record; + now?: number; + }): Promise<{ lease: SubmissionLease } | { error: string }> { + const context = options.context ?? {}; + const now = options.now ?? Date.now(); + let claim: StoredRecord | null; + try { + claim = await options.store.claim( + options.key, + { hash: null, updatedAtMs: now }, + options.claimTtlMs ?? DEFAULT_CLAIM_TTL_MS + ); + } catch (err) { + options.logger.warn("[keeper-state] could not claim the target", { + ...context, + error: errorMessage(err), + }); + return { error: "submission state store unavailable" }; + } + if (!claim) { + return { error: "another run already holds this target" }; + } + return { + lease: new SubmissionLease( + options.store, + options.key, + options.submissionTtlMs, + options.logger, + context, + claim + ), + }; + } + + /** Whether this run still holds a claim that never became a transaction. */ + get unsent(): boolean { + return this.held !== null && this.held.record.hash === null; + } + + /** + * Records the signed transaction's hash, conditional on this run still + * holding what it wrote last. Called before `sendTransaction`, so a + * transaction that reaches the mempool is covered even if the send call + * times out or is deferred. + */ + private async write(hash: string, now = Date.now()): Promise { + if (!this.held) return; + const revision = this.held.revision; + try { + const next = await withRetry( + () => + this.store.replace( + this.key, + { hash, updatedAtMs: now }, + this.submissionTtlMs, + revision + ), + WRITE_RETRY_ATTEMPTS, + WRITE_RETRY_BASE_DELAY_MS + ); + if (!next) { + // Lost the lease (claim expired and someone else took it). Stop + // touching the key: it now belongs to another run. + this.logger.warn("[keeper-state] lost the submission lease", { + ...this.context, + hash, + }); + this.held = null; + return; + } + this.held = next; + } catch (err) { + // Every retry failed: the claim (short TTL, no hash) is still what's + // stored, so this signed, in-flight transaction is only covered until + // the much shorter claim window ages out, not its real validity + // window. Logged loudly since a concurrent run that outlives the + // claim TTL will read "expired" and rebroadcast. + this.logger.warn( + "[keeper-state] could not record submission after retries; the claim covering it will age out early", + { + ...this.context, + hash, + error: errorMessage(err), + } + ); + } + } + + /** Clears this run's record once the transaction's fate is known. */ + private async clear(): Promise { + if (!this.held) return; + const revision = this.held.revision; + this.held = null; + await clearRecord( + this.store, + this.key, + revision, + this.logger, + this.context + ); + } + + /** + * Releases a claim that never became a signed transaction (a simulation + * error, a deadline, a rejected build), so the next run isn't blocked for + * the claim's full window over a target nothing was ever sent for. + */ + async releaseIfUnsent(): Promise { + if (this.unsent) await this.clear(); + } + + /** Hooks to hand to submitKeeperOperation (see keeper-tx.ts). */ + get hooks(): KeeperSubmissionHooks { + return { + onSigned: (hash) => this.write(hash), + onResolved: () => this.clear(), + }; + } +} + +/** + * Per-process store. Useful in tests and local dev, but note what it is not: + * keeper invocations are separate serverless executions, so nothing written + * here survives to the next run. It keeps the code path identical without + * pretending to provide cross-invocation dedup; only a shared store does. + */ +export function createInMemoryKeeperStateStore(): KeeperStateStore { + const entries = new Map(); + + function read(key: string): string | null { + const entry = entries.get(key); + if (!entry) return null; + if (Date.now() > entry.expiresAt) { + entries.delete(key); + return null; + } + return entry.value; + } + + return { + async get(key) { + const value = read(key); + return value === null ? null : hydrate(value); + }, + async claim(key, record, ttlMs) { + if (read(key) !== null) return null; + const value = serializeRecord(record); + entries.set(key, { value, expiresAt: Date.now() + ttlMs }); + return { record, revision: value }; + }, + async replace(key, record, ttlMs, expectedRevision) { + if (read(key) !== expectedRevision) return null; + const value = serializeRecord(record); + entries.set(key, { value, expiresAt: Date.now() + ttlMs }); + return { record, revision: value }; + }, + async deleteIf(key, expectedRevision) { + if (read(key) !== expectedRevision) return false; + entries.delete(key); + return true; + }, + }; +} + +function hydrate(value: string): StoredRecord | null { + let parsed: unknown; + try { + parsed = JSON.parse(value); + } catch { + return null; + } + if (!parsed || typeof parsed !== "object") return null; + const { hash, updatedAtMs } = parsed as Partial; + if (hash !== null && typeof hash !== "string") return null; + if (hash === "") return null; + if (typeof updatedAtMs !== "number" || !Number.isFinite(updatedAtMs)) { + return null; + } + return { record: { hash, updatedAtMs }, revision: value }; +} + +// Compare-and-set / compare-and-delete need to be atomic against the stored +// value, which the REST API can only express through a script. +const CAS_SET = + "if redis.call('GET', KEYS[1]) == ARGV[1] then redis.call('SET', KEYS[1], ARGV[2], 'PX', ARGV[3]) return 1 else return 0 end"; +const CAS_DEL = + "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) else return 0 end"; + +/** + * Upstash Redis store, over the REST API the rest of this repo already + * points at for rate limiting (`api/_lib/middleware.ts`), reusing the same + * `UPSTASH_REDIS_REST_URL`/`UPSTASH_REDIS_REST_TOKEN` pair. + * + * Spoken over plain `fetch` rather than `@upstash/redis` on purpose: this + * package is the shared Stellar helper library, imported by the web build as + * well as the API, and a handful of Redis commands don't justify pulling a + * client dependency into it. + * + * Every record is written with a Redis-side expiry as well, 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. + */ +export function createUpstashKeeperStateStore(options: { + url: string; + token: string; + fetchImpl?: typeof fetch; + timeoutMs?: number; +}): KeeperStateStore { + const url = options.url.replace(/\/+$/, ""); + const fetchImpl = options.fetchImpl ?? fetch; + const timeoutMs = options.timeoutMs ?? DEFAULT_STORE_TIMEOUT_MS; + + async function command(args: (string | number)[]): Promise { + const response = await withRaceTimeout( + () => + fetchImpl(url, { + method: "POST", + headers: { + Authorization: `Bearer ${options.token}`, + "Content-Type": "application/json", + }, + body: JSON.stringify(args), + // Belt and braces with the race below: this also frees the socket + // rather than leaving a hung request running past the run. + signal: AbortSignal.timeout(timeoutMs), + }), + timeoutMs, + "Upstash Redis" + ); + if (!response.ok) { + // Deliberately status-only: the response body can echo the command, + // and the URL/token never appear in the message at all. + throw new Error( + `Upstash Redis request failed with HTTP ${response.status}` + ); + } + const body = (await response.json()) as { + result?: unknown; + error?: string; + }; + if (body.error) throw new Error(`Upstash Redis error: ${body.error}`); + return body.result ?? null; + } + + return { + async get(key) { + const value = await command(["GET", key]); + return typeof value === "string" ? hydrate(value) : null; + }, + async claim(key, record, ttlMs) { + const value = serializeRecord(record); + // PX, not EX: these windows are derived from millisecond transaction + // bounds, and rounding up to whole seconds would keep a dead record + // blocking longer than the transaction it tracks could live. + const result = await command([ + "SET", + key, + value, + "NX", + "PX", + expiry(ttlMs), + ]); + return result === null ? null : { record, revision: value }; + }, + async replace(key, record, ttlMs, expectedRevision) { + const value = serializeRecord(record); + const result = await command([ + "EVAL", + CAS_SET, + 1, + key, + expectedRevision, + value, + String(expiry(ttlMs)), + ]); + return Number(result) === 1 ? { record, revision: value } : null; + }, + async deleteIf(key, expectedRevision) { + const result = await command(["EVAL", CAS_DEL, 1, key, expectedRevision]); + return Number(result) === 1; + }, + }; +} + +function expiry(ttlMs: number): number { + return Math.max(1, Math.ceil(ttlMs)); +} + +/** + * Picks the submission state store from the environment. + * + * `requireShared` is the migration keeper: a duplicate `migrate_adapter` + * costs real slippage twice, so on any deployed environment it refuses to + * run without a shared store rather than silently degrading to a + * per-process one that cannot dedup across invocations at all. Preview + * counts as deployed: preview deployments sign real transactions off a real + * key (see `api/_lib/middleware.ts`), so "not production" is not the same as + * "not real". + */ +export function loadKeeperStateStore( + env: Record, + options: { + keeper: "accrual" | "migration"; + requireShared: boolean; + logger: KeeperLogger; + fetchImpl?: typeof fetch; + timeoutMs?: number; + } +): KeeperStateStore { + const url = env.UPSTASH_REDIS_REST_URL?.trim(); + const token = env.UPSTASH_REDIS_REST_TOKEN?.trim(); + if (url && token) { + return createUpstashKeeperStateStore({ + url, + token, + ...(options.fetchImpl && { fetchImpl: options.fetchImpl }), + ...(options.timeoutMs !== undefined && { timeoutMs: options.timeoutMs }), + }); + } + const deployed = Boolean(env.VERCEL_ENV); + if (options.requireShared && deployed) { + throw new Error( + `Refusing to run the migration keeper on a ${env.VERCEL_ENV} deployment: UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN are required (the in-memory fallback is per-invocation and cannot prevent a duplicate migrate_adapter)` + ); + } + // Warn, not info, on anything deployed: falling back reinstates the exact + // duplicate-submission gap this module exists to close, and that should be + // visible in logs rather than buried at info level. Local dev, where the + // fallback is the expected state, stays quiet. + const message = `[${options.keeper}-keeper] no shared submission state store configured; cross-invocation dedup is inactive for this run`; + if (deployed) { + options.logger.warn(message, { store: "in-memory", env: env.VERCEL_ENV }); + } else { + options.logger.info(message, { store: "in-memory" }); + } + return createInMemoryKeeperStateStore(); +} diff --git a/packages/stellar-sdk-helpers/src/keeper-tx.ts b/packages/stellar-sdk-helpers/src/keeper-tx.ts index f2508272..e4a06b1a 100644 --- a/packages/stellar-sdk-helpers/src/keeper-tx.ts +++ b/packages/stellar-sdk-helpers/src/keeper-tx.ts @@ -12,10 +12,22 @@ import { } from "@stellar/stellar-sdk"; import { withRaceTimeout } from "@meridian/shared"; import { BASE_FEE } from "./internal"; -import { describeSendError, simErrorMessage, waitForTransaction } from "./tx"; +import { + describeSendError, + simErrorMessage, + simulateView, + waitForTransaction, +} from "./tx"; import type { StellarNetwork } from "./types"; import { errorMessage } from "./keeper-retry"; +// Transactions are built with this validity window (`setTimeout` below), so +// it is also the point past which a submitted transaction can never land. +// Exported because the submission-record TTL in keeper-state.ts is derived +// from it: a record that expired sooner would clear while its transaction +// could still be landing. +export const TX_VALIDITY_WINDOW_MS = 300_000; + // A real rpc.Server satisfies this directly (no cast needed); a narrower // Pick instead of the hand-written interface this used to be means the // signatures can never silently drift from the real SDK's. @@ -113,6 +125,96 @@ export function expectString( return value; } +// Thrown by the stale-adapter guard below: the caller's view of which +// adapter a vault is using is out of date, something else already changed +// it. Shared by both keepers: the migration keeper uses it to avoid +// migrating off an adapter the vault no longer has, and the accrual keeper +// to avoid accruing on an adapter the vault has already migrated away from +// (a silently ineffective call, since a detached adapter is still a valid +// contract). Both treat it as a benign, expected race, a skip rather than a +// failure. +// +// Detected downstream by message text, not `instanceof`: withKeeperRetry +// wraps whatever it catches in a KeeperRetryError (keeper-retry.ts), which +// preserves the message but not the original error's type. Same approach +// isDefinitiveOnChainFailure already uses for the equivalent problem. +export const STALE_ADAPTER_MESSAGE = "Vault's adapter changed since discovery"; + +export class StaleAdapterError extends Error { + constructor(expected: string, actual: string) { + super( + `${STALE_ADAPTER_MESSAGE} (expected ${expected}, now ${actual}); skipping to avoid a stale call` + ); + this.name = "StaleAdapterError"; + } +} + +export function isStaleAdapterError(err: unknown): boolean { + return errorMessage(err).includes(STALE_ADAPTER_MESSAGE); +} + +/** + * Re-reads the vault's live `get_adapter()` and throws StaleAdapterError if + * it no longer matches what this run discovered. A cheap, best-effort guard + * that narrows, but cannot close, the window between deciding to act on an + * adapter and the transaction landing: it catches the common case of "a + * prior run already changed this vault's adapter" for one simulate call. + */ +export async function assertAdapterUnchanged( + server: KeeperRpcServer, + vaultContractId: string, + networkPassphrase: string, + expectedAdapterId: string +): Promise { + const liveAdapterId = expectString( + await simulateView( + server as never, + vaultContractId, + networkPassphrase, + "get_adapter" + ), + "get_adapter", + vaultContractId + ); + if (liveAdapterId !== expectedAdapterId) { + throw new StaleAdapterError(expectedAdapterId, liveAdapterId); + } +} + +// Lifecycle hooks around the two moments that matter for cross-invocation +// dedup: +// +// `onSigned` fires as soon as a transaction is signed and its hash is known, +// *before* `sendTransaction`. Recording after the send returns would miss +// the cases that matter most: a send that times out, or comes back +// TRY_AGAIN_LATER, may already have put the transaction in the mempool with +// no record of it anywhere. +// +// `onResolved` fires once that hash's fate is known: confirmed, definitively +// failed on-chain, or rejected outright at submission. +// +// Both are invoked defensively: a throwing hook must never surface as a +// submission error, since the retry loop would answer that by broadcasting a +// second transaction, exactly the duplicate the hooks exist to prevent. +// Implementations are expected to log their own failures (see +// keeper-state.ts). +export interface KeeperSubmissionHooks { + onSigned?: (hash: string) => Promise; + onResolved?: (hash: string) => Promise; +} + +async function runHook( + hook: ((hash: string) => Promise) | undefined, + hash: string +): Promise { + if (!hook) return; + try { + await hook(hash); + } catch { + // Deliberately swallowed, see KeeperSubmissionHooks above. + } +} + export interface KeeperTxConfig { network: StellarNetwork; secretKey: string; @@ -127,23 +229,27 @@ export interface KeeperTxConfig { // transaction on this call, only re-throws to keep tracking the *same* hash, // until the prior transaction's fate is actually known (confirmed success or // confirmed on-chain failure). Callers must persist `priorHash` across their -// own retry attempts (see accrual-keeper.ts and migration-keeper.ts). +// own retry attempts (see accrual-keeper.ts and migration-keeper.ts), and +// persist it across invocations through `hooks` (see keeper-state.ts). export async function submitKeeperOperation( contractId: string, method: string, args: xdr.ScVal[], config: KeeperTxConfig, server: KeeperRpcServer, - priorHash?: string + priorHash?: string, + hooks?: KeeperSubmissionHooks ): Promise<{ hash: string; ledger: number }> { if (priorHash) { try { const confirmed = await waitForTransaction(server, priorHash, { timeoutMs: config.confirmationTimeoutMs, }); + await runHook(hooks?.onResolved, priorHash); return { hash: priorHash, ledger: confirmed.ledger }; } catch (err) { if (isDefinitiveOnChainFailure(err)) { + await runHook(hooks?.onResolved, priorHash); throw new SubmissionFailedError(err); } throw new SubmissionInFlightError(priorHash, err); @@ -162,7 +268,7 @@ export async function submitKeeperOperation( networkPassphrase: config.network.passphrase, }) .addOperation(contract.call(method, ...args)) - .setTimeout(300) + .setTimeout(TX_VALIDITY_WINDOW_MS / 1000) .build(); const sim = await withRaceTimeout( @@ -180,27 +286,54 @@ export async function submitKeeperOperation( const prepared = rpc.assembleTransaction(tx, sim).build(); prepared.sign(keypair); - const sent = await withRaceTimeout( - () => server.sendTransaction(prepared), - config.rpcTimeoutMs, - "Soroban RPC" - ); + // Known before the network is touched at all: from here on, every path + // out of this function has a hash to track, so no failure mode can leave a + // transaction in the mempool with nothing recorded against it. + const signedHash = prepared.hash().toString("hex"); + await runHook(hooks?.onSigned, signedHash); + + let sent: Awaited>; + try { + sent = await withRaceTimeout( + () => server.sendTransaction(prepared), + config.rpcTimeoutMs, + "Soroban RPC" + ); + } catch (err) { + // The send may well have reached the network before this timed out. + // Never rebuild after this point: a fresh transaction would have a + // different hash and could land alongside this one. Tracking the same + // hash is what the retry path is for. + throw new SubmissionInFlightError(signedHash, err); + } if (sent.status === "ERROR") { + // Rejected outright: this transaction is not in flight and never will + // be, so release the record rather than blocking the target until it + // ages out. + await runHook(hooks?.onResolved, signedHash); throw new Error( `Transaction rejected at submission: ${describeSendError(sent)}` ); } if (sent.status === "TRY_AGAIN_LATER") { - throw new Error("Transaction could not be submitted yet (try again later)"); + // Explicitly *not* a clean "nothing happened": the node may already be + // processing this transaction. Same treatment as a timeout, recheck the + // hash instead of building a second transaction. + throw new SubmissionInFlightError( + signedHash, + new Error("Transaction could not be submitted yet (try again later)") + ); } try { const confirmed = await waitForTransaction(server, sent.hash, { timeoutMs: config.confirmationTimeoutMs, }); + await runHook(hooks?.onResolved, sent.hash); return { hash: sent.hash, ledger: confirmed.ledger }; } catch (err) { if (isDefinitiveOnChainFailure(err)) { + await runHook(hooks?.onResolved, sent.hash); throw new SubmissionFailedError(err); } throw new SubmissionInFlightError(sent.hash, err); diff --git a/packages/stellar-sdk-helpers/src/migration-keeper.test.ts b/packages/stellar-sdk-helpers/src/migration-keeper.test.ts index 592e174d..67ecd6d0 100644 --- a/packages/stellar-sdk-helpers/src/migration-keeper.test.ts +++ b/packages/stellar-sdk-helpers/src/migration-keeper.test.ts @@ -105,6 +105,12 @@ import { type MigrationKeeperConfig, } from "./migration-keeper"; import type { KeeperLogger } from "./keeper-retry"; +import { + createInMemoryKeeperStateStore, + submissionStateKey, + type KeeperStateStore, + type SubmissionRecord, +} from "./keeper-state"; import type { KnownPoolMeta } from "./known-pools"; const NETWORK = { @@ -121,6 +127,7 @@ const CONFIG: MigrationKeeperConfig = { rpcTimeoutMs: 100, minImprovementBps: 50, maxSlippageBps: 100, + submissionTtlMs: 360_000, candidateAdapters: { defindex: "CDEFINDEXADAPTER" }, }; @@ -140,6 +147,9 @@ const DISCOVERED_VAULT: DiscoveredVault = { currentPoolId: "CBLENDPOOL", }; +// Hash of the signed transaction, known before submission. +const SIGNED_HASH = "deadbeef"; + function logger(): KeeperLogger { return { info: vi.fn(), warn: vi.fn(), error: vi.fn() }; } @@ -165,7 +175,12 @@ beforeEach(() => { stellarMocks.isSimulationError.mockReturnValue(false); stellarMocks.isSimulationSuccess.mockReturnValue(true); stellarMocks.assembleTransaction.mockReturnValue({ - build: () => ({ sign: stellarMocks.signPrepared }), + build: () => ({ + sign: stellarMocks.signPrepared, + // The keeper records the signed transaction's own hash before it is + // ever sent, so the built transaction has to expose one. + hash: () => Buffer.from(SIGNED_HASH, "hex"), + }), }); }); @@ -452,10 +467,16 @@ describe("runMigrationKeeper", () => { submitMigration, }); + // The fourth argument is this run's submission-lease hooks: an injected + // submitter is expected to forward them to keep cross-invocation dedup. expect(submitMigration).toHaveBeenCalledWith( DISCOVERED_VAULT, "CDEFINDEXADAPTER", - 1 + 1, + expect.objectContaining({ + onSigned: expect.any(Function), + onResolved: expect.any(Function), + }) ); expect(result.migrations).toEqual([ { @@ -1174,3 +1195,189 @@ describe("runMigrationKeeper", () => { ]); }); }); + +describe("runMigrationKeeper cross-invocation dedup", () => { + const KEY = submissionStateKey("migration", "testnet", "meridian-usdc"); + + async function store(seed?: SubmissionRecord) { + const inner = createInMemoryKeeperStateStore(); + if (seed) await inner.claim(KEY, seed, 600_000); + return inner; + } + + const rateSource = () => + vi.fn(async ({ protocol }: { protocol: string }) => + protocol === "blend" ? 500 : 700 + ); + + function run(stateStore: KeeperStateStore, rates = rateSource()) { + return runMigrationKeeper(CONFIG, { + logger: logger(), + sleep: vi.fn(), + stateStore, + discoverVaults: async () => ({ + vaults: [DISCOVERED_VAULT], + failures: [], + }), + rateSource: rates, + resolveCandidatePool: async () => "CDEFINDEXPOOL", + }); + } + + it("does not send a second migrate_adapter while a prior one is still unconfirmed", async () => { + // The whole point of #515: unlike accrue(), a duplicate here costs real + // slippage a second time, not a flat fee. + const server = makeServer({ + getTransaction: vi.fn(async () => ({ status: "NOT_FOUND" })), + }); + stellarMocks.getRpcServer.mockReturnValue(server); + stellarMocks.simulateView.mockResolvedValue( + DISCOVERED_VAULT.currentAdapterId + ); + const rates = rateSource(); + + const result = await run( + await store({ hash: "INFLIGHT_HASH", updatedAtMs: Date.now() - 1_000 }), + rates + ); + + expect(server.sendTransaction).not.toHaveBeenCalled(); + // Blocked before evaluation, so the rate lookups (and the deadline + // budget they spend) are never paid for a vault that cannot migrate. + expect(rates).not.toHaveBeenCalled(); + expect(result.migrations).toEqual([]); + expect(result.failures).toEqual([]); + expect(result.skipped).toMatchObject([ + { + vaultId: "meridian-usdc", + reason: expect.stringContaining("still unconfirmed"), + }, + ]); + }); + + it("skips a vault another run has claimed but not yet signed for", async () => { + // A plain "no record" read is not a claim: two concurrent invocations + // could otherwise both pass the check and both broadcast. + const server = makeServer(); + stellarMocks.getRpcServer.mockReturnValue(server); + stellarMocks.simulateView.mockResolvedValue( + DISCOVERED_VAULT.currentAdapterId + ); + + const result = await run( + await store({ hash: null, updatedAtMs: Date.now() }) + ); + + expect(server.sendTransaction).not.toHaveBeenCalled(); + expect(result.skipped).toMatchObject([ + { reason: expect.stringContaining("already preparing") }, + ]); + }); + + it("resolves a prior submission that landed and evaluates again", async () => { + const server = makeServer({ + getTransaction: vi.fn(async () => ({ status: "SUCCESS", ledger: 5 })), + sendTransaction: vi.fn(async () => ({ + hash: "SUBMITTED_HASH", + status: "PENDING", + })), + }); + stellarMocks.getRpcServer.mockReturnValue(server); + stellarMocks.waitForTransaction.mockResolvedValue({ ledger: 321 }); + stellarMocks.simulateView.mockResolvedValue( + DISCOVERED_VAULT.currentAdapterId + ); + const stateStore = await store({ + hash: "LANDED_HASH", + updatedAtMs: Date.now() - 1_000, + }); + + const result = await run(stateStore); + + expect(result.migrations).toMatchObject([{ hash: "SUBMITTED_HASH" }]); + expect(await stateStore.get(KEY)).toBeNull(); + }); + + it("clears a record whose transaction is past its validity window instead of blocking on it", async () => { + const server = makeServer({ + getTransaction: vi.fn(async () => ({ status: "NOT_FOUND" })), + sendTransaction: vi.fn(async () => ({ + hash: "SUBMITTED_HASH", + status: "PENDING", + })), + }); + stellarMocks.getRpcServer.mockReturnValue(server); + stellarMocks.waitForTransaction.mockResolvedValue({ ledger: 321 }); + stellarMocks.simulateView.mockResolvedValue( + DISCOVERED_VAULT.currentAdapterId + ); + + const result = await run( + await store({ + hash: "DEAD_HASH", + updatedAtMs: Date.now() - CONFIG.submissionTtlMs - 1, + }) + ); + + expect(result.migrations).toMatchObject([{ hash: "SUBMITTED_HASH" }]); + }); + + it("records the signed hash before the transaction is sent, so a killed run still blocks the next one", async () => { + let recordedAtSend: string | null | undefined; + const stateStore = await store(); + const server = makeServer({ + getTransaction: vi.fn(async () => ({ status: "NOT_FOUND" })), + sendTransaction: vi.fn(async () => { + recordedAtSend = (await stateStore.get(KEY))?.record.hash; + return { hash: "SUBMITTED_HASH", status: "PENDING" }; + }), + }); + stellarMocks.getRpcServer.mockReturnValue(server); + stellarMocks.simulateView.mockResolvedValue( + DISCOVERED_VAULT.currentAdapterId + ); + stellarMocks.waitForTransaction.mockResolvedValue({ ledger: 321 }); + + await run(stateStore); + + expect(recordedAtSend).toBe(SIGNED_HASH); + expect(await stateStore.get(KEY)).toBeNull(); + }); + + it("skips the vault when the prior submission's status cannot be checked", async () => { + // A store or RPC outage must not be read as "nothing was submitted": + // that is precisely the assumption that produces a double migration. + const server = makeServer({ + getTransaction: vi.fn(async () => { + throw new Error("rpc unavailable"); + }), + }); + stellarMocks.getRpcServer.mockReturnValue(server); + + const result = await run( + await store({ hash: "UNKNOWN_HASH", updatedAtMs: Date.now() }) + ); + + expect(server.sendTransaction).not.toHaveBeenCalled(); + expect(result.failures).toEqual([]); + expect(result.skipped).toMatchObject([ + { reason: expect.stringContaining("could not be verified") }, + ]); + }); + + it("releases the claim when the migration is abandoned before anything is signed", async () => { + // The stale-adapter guard aborts before the transaction is built. + // Leaving the claim behind would lock the vault out of migrating for + // the claim's whole window over a transaction that never existed. + stellarMocks.simulateView.mockResolvedValue("CSOMEOTHERADAPTER"); + const stateStore = await store(); + stellarMocks.getRpcServer.mockReturnValue(makeServer()); + + const result = await run(stateStore); + + expect(result.skipped).toMatchObject([ + { reason: expect.stringContaining("adapter changed since discovery") }, + ]); + expect(await stateStore.get(KEY)).toBeNull(); + }); +}); diff --git a/packages/stellar-sdk-helpers/src/migration-keeper.ts b/packages/stellar-sdk-helpers/src/migration-keeper.ts index 886cd6c2..61c51a98 100644 --- a/packages/stellar-sdk-helpers/src/migration-keeper.ts +++ b/packages/stellar-sdk-helpers/src/migration-keeper.ts @@ -34,12 +34,23 @@ import { type RetryConfig, } from "./keeper-retry"; import { + assertAdapterUnchanged, expectString, + isStaleAdapterError, isTransientKeeperError, submitKeeperOperation, SubmissionInFlightError, type KeeperRpcServer, + type KeeperSubmissionHooks, } from "./keeper-tx"; +import { + loadKeeperStateStore, + parseSubmissionTtlMs, + resolvePriorSubmission, + submissionStateKey, + SubmissionLease, + type KeeperStateStore, +} from "./keeper-state"; const DEFAULT_MAX_ATTEMPTS = 3; const DEFAULT_BASE_DELAY_MS = 1_000; @@ -115,6 +126,7 @@ export interface MigrationKeeperConfig { rpcTimeoutMs: number; minImprovementBps: number; maxSlippageBps: number; + submissionTtlMs: number; candidateAdapters: Record; } @@ -174,10 +186,15 @@ export interface MigrationKeeperDeps { }>; rateSource?: RateSourceFn; resolveCandidatePool?: (adapterId: string) => Promise; + // `hooks` carries this run's submission lease: an override that forwards + // it to submitKeeperOperation keeps cross-invocation dedup; one that + // ignores it falls back to the claim held for the duration of the run and + // the on-chain adapter re-check, and is warned about at run start. submitMigration?: ( vault: DiscoveredVault, toAdapterId: string, - attempt: number + attempt: number, + hooks: KeeperSubmissionHooks ) => Promise< Omit< MigrationSuccess, @@ -190,6 +207,9 @@ export interface MigrationKeeperDeps { | "improvementBps" > >; + // Cross-invocation submission tracking (#515). Defaults to whatever the + // environment provides (Upstash Redis when configured); injected in tests. + stateStore?: KeeperStateStore; logger?: KeeperLogger; sleep?: (ms: number) => Promise; deadlineAt?: number; @@ -290,6 +310,7 @@ export function loadMigrationKeeperConfig( "MERIDIAN_MIGRATION_MIN_IMPROVEMENT_BPS" ), maxSlippageBps, + submissionTtlMs: parseSubmissionTtlMs(env), candidateAdapters: parseCandidateAdapters(env), }; } @@ -650,33 +671,6 @@ async function findBestCandidate( }; } -// Thrown by the stale-adapter guard below: this run's discovery data is out -// of date, something else already changed the vault's adapter. This is the -// benign, expected outcome the guard exists to catch (the cross-invocation -// race documented in migration-keeper.md, tracked in #515), not a genuine -// operational problem, so it's reported as a skip, not a KeeperFailure. -// -// Detected downstream by message text, not `instanceof`: withKeeperRetry -// wraps whatever it catches in a KeeperRetryError (keeper-retry.ts), which -// preserves the message but not the original error's type, so by the time -// this reaches runMigrationKeeper's catch block, `err` is a -// KeeperRetryError, never a StaleAdapterError. Same approach -// isDefinitiveOnChainFailure already uses for the equivalent problem. -const STALE_ADAPTER_MESSAGE = "Vault's adapter changed since discovery"; - -class StaleAdapterError extends Error { - constructor(expected: string, actual: string) { - super( - `${STALE_ADAPTER_MESSAGE} (expected ${expected}, now ${actual}); skipping to avoid a stale migration` - ); - this.name = "StaleAdapterError"; - } -} - -function isStaleAdapterError(err: unknown): boolean { - return errorMessage(err).includes(STALE_ADAPTER_MESSAGE); -} - async function submitMigrationTransaction( vaultContractId: string, expectedCurrentAdapterId: string, @@ -684,30 +678,23 @@ async function submitMigrationTransaction( maxSlippageBps: number, config: MigrationKeeperConfig, server: KeeperRpcServer, - priorHash?: string + priorHash?: string, + hooks?: KeeperSubmissionHooks ): Promise<{ hash: string; ledger: number }> { // Only checked before building a brand-new transaction, never when // rechecking an already-sent one (priorHash set): a cheap, best-effort - // guard against the cross-invocation race documented in - // migration-keeper.md, this run's discovery data could be stale if - // another invocation already migrated this vault since. Narrows, doesn't - // close, the window: a genuine TOCTOU gap remains between this check and - // the transaction actually landing, real cross-invocation dedup is - // tracked separately in #515. + // guard against this run's discovery data being stale because another + // invocation already migrated this vault since. It also covers the one + // window the submission record can't (broadcast succeeded, then the + // process died before the record was written), so the two guards are + // complementary rather than redundant, see migration-keeper.md. if (!priorHash) { - const liveAdapterId = expectString( - await simulateView( - server as never, - vaultContractId, - config.network.passphrase, - "get_adapter" - ), - "get_adapter", - vaultContractId + await assertAdapterUnchanged( + server, + vaultContractId, + config.network.passphrase, + expectedCurrentAdapterId ); - if (liveAdapterId !== expectedCurrentAdapterId) { - throw new StaleAdapterError(expectedCurrentAdapterId, liveAdapterId); - } } return submitKeeperOperation( @@ -724,7 +711,8 @@ async function submitMigrationTransaction( confirmationTimeoutMs: CONFIRMATION_TIMEOUT_MS, }, server, - priorHash + priorHash, + hooks ); } @@ -737,6 +725,22 @@ export async function runMigrationKeeper( const startedAt = new Date().toISOString(); const deadlineAt = deps.deadlineAt ?? Date.now() + FUNCTION_BUDGET_MS; const server = getRpcServer(config.network.rpcUrl, config.rpcTimeoutMs); + // Shared, not per-run, state: this is the only thing that survives a + // killed invocation, so it's what stops the next cron tick from sending a + // second migrate_adapter while the first is still landing (#515). + const stateStore = + deps.stateStore ?? + loadKeeperStateStore(process.env, { + keeper: "migration", + requireShared: true, + logger, + }); + if (deps.submitMigration) { + logger.warn( + "[migration-keeper] submitMigration is overridden; cross-invocation dedup depends on the injected submitter forwarding the provided hooks", + { vaultScope: "all" } + ); + } const rateSource = deps.rateSource ?? defaultRateSource; const resolveCandidatePool = deps.resolveCandidatePool ?? @@ -792,6 +796,62 @@ export async function runMigrationKeeper( continue; } + // Checked before evaluation, not just before submission: a vault whose + // prior migration is still in flight isn't going to be migrated this + // run either way, so there's no reason to spend the rate lookups (and + // the deadline budget they consume) reaching that conclusion. + const stateKey = submissionStateKey( + "migration", + config.network.network, + vault.vaultId + ); + const priorContext = { vaultId: vault.vaultId, keeper: "migration-keeper" }; + const prior = await resolvePriorSubmission({ + store: stateStore, + key: stateKey, + server, + ttlMs: config.submissionTtlMs, + rpcTimeoutMs: config.rpcTimeoutMs, + logger, + context: priorContext, + }); + // Every blocking state is fatal for this vault this run: unlike the + // accrue keeper, this one has no cheap-duplicate escape hatch, so an + // unverifiable store is a reason to stop, not to guess. + if ( + prior.state === "in-flight" || + prior.state === "claimed" || + prior.state === "unknown" + ) { + const reason = + prior.state === "in-flight" + ? "a prior migrate_adapter submission is still unconfirmed; skipped to avoid a duplicate migration" + : prior.state === "claimed" + ? "another run is already preparing a migration for this vault; skipped to avoid a duplicate migration" + : `prior submission state could not be verified (${prior.reason}); skipped rather than risk a duplicate migration`; + skipped.push({ vaultId: vault.vaultId, reason }); + logger.warn("[migration-keeper] migration skipped; prior submission", { + vaultId: vault.vaultId, + state: prior.state, + ...(prior.state === "in-flight" && { + hash: prior.hash, + ageMs: prior.ageMs, + }), + }); + continue; + } + if (prior.state !== "none") { + // landed / failed / expired: the record is already cleared, this run + // is free to evaluate again. Logged because "the previous run's + // transaction turned out to have landed after all" is exactly the + // sequence that's impossible to reconstruct afterwards otherwise. + logger.info("[migration-keeper] prior submission resolved", { + vaultId: vault.vaultId, + state: prior.state, + hash: prior.hash, + }); + } + let evaluation: { best: BestCandidate | null; skipReason?: string }; try { evaluation = await findBestCandidate( @@ -858,12 +918,37 @@ export async function runMigrationKeeper( }); continue; } + // Taken before anything is built: a plain "no record" read is not a + // claim on the vault, so two concurrent invocations could otherwise both + // pass the check above and both broadcast. + const acquired = await SubmissionLease.acquire({ + store: stateStore, + key: stateKey, + submissionTtlMs: config.submissionTtlMs, + logger, + context: priorContext, + }); + if ("error" in acquired) { + skipped.push({ + vaultId: vault.vaultId, + reason: `could not take the submission lease (${acquired.error}); skipped rather than risk a duplicate migration`, + }); + continue; + } + const lease = acquired.lease; + const submissionHooks = lease.hooks; + let priorHash: string | undefined; try { const result = await withKeeperRetry( (attempt) => deps.submitMigration - ? deps.submitMigration(vault, best.adapterId, attempt) + ? deps.submitMigration( + vault, + best.adapterId, + attempt, + submissionHooks + ) : submitMigrationTransaction( vault.vaultContractId, vault.currentAdapterId, @@ -871,7 +956,8 @@ export async function runMigrationKeeper( config.maxSlippageBps, config, server, - priorHash + priorHash, + submissionHooks ).catch((err: unknown) => { if (err instanceof SubmissionInFlightError) { priorHash = err.sentHash; @@ -951,6 +1037,11 @@ export async function runMigrationKeeper( }; failures.push(failure); logger.error("[migration-keeper] migrate_adapter failed", { ...failure }); + } finally { + // A claim that never became a signed transaction (stale adapter, a + // simulation error, an exhausted deadline) must not keep the next run + // out for the claim's full window. + await lease.releaseIfUnsent(); } }