From 5236a741405f4014042bf1c466da57802419c34b Mon Sep 17 00:00:00 2001 From: Zac Lou <97340247+ZacLou@users.noreply.github.com> Date: Thu, 3 Sep 2026 15:08:22 +0800 Subject: [PATCH 1/5] fix(migration-keeper): keep active snapshot under alternating rates The migration keeper re-derives the best-rate candidate from scratch on every scheduled run with no memory of an already-begun migration. When the top candidate flips between two adapters across consecutive runs, each run overwrites the prior begin_migration snapshot and resets the ledger-gap cooldown, so a migration can in principle never reach migrate_adapter despite a real, sustained improvement opportunity. Fix: before falling through to begin_migration for the freshly-derived best candidate, check whether an active migration snapshot already exists for a different adapter. If it does and that adapter still clears minImprovementBps against the current rate, override best to the snapshotted adapter so the existing migrate_adapter path completes it instead of resetting the cooldown. Only let the snapshot lapse when its adapter's rate genuinely stops clearing the threshold. Two new tests cover both halves: preservation when the snapshot still qualifies, and replacement when it has genuinely decayed. --- .../src/migration-keeper.test.ts | 126 +++++++++++++++ .../src/migration-keeper.ts | 145 +++++++++++++++++- 2 files changed, 270 insertions(+), 1 deletion(-) diff --git a/packages/stellar-sdk-helpers/src/migration-keeper.test.ts b/packages/stellar-sdk-helpers/src/migration-keeper.test.ts index 5b71b674..748c114f 100644 --- a/packages/stellar-sdk-helpers/src/migration-keeper.test.ts +++ b/packages/stellar-sdk-helpers/src/migration-keeper.test.ts @@ -1405,6 +1405,132 @@ describe("runMigrationKeeper", () => { ]); }); + // #699: When an active migration snapshot exists for a configured + // candidate adapter that differs from the freshly-derived "best" + // candidate, but the snapshotted adapter still clears + // minImprovementBps, the keeper should prefer completing the existing + // migration instead of overwriting the snapshot with a new + // begin_migration for a different adapter. Without this, alternating + // rates between two adapters across scheduled runs would reset the + // ledger-gap cooldown on every run, and a real migration opportunity + // could never reach migrate_adapter. + it("preserves an active migration snapshot whose adapter still clears the improvement threshold (#699)", async () => { + const server = makeServer(); + stellarMocks.getRpcServer.mockReturnValue(server); + stellarMocks.waitForTransaction.mockResolvedValue({ ledger: 321 }); + + // Two candidate adapters: "defindex" (the existing CONFIG candidate) + // and "blendv2" (a second one that will hold the active snapshot). + const configWithTwoCandidates: MigrationKeeperConfig = { + ...CONFIG, + candidateAdapters: { + defindex: "CDEFINDEXADAPTER", + blendv2: "CBLENDV2ADAPTER", + }, + }; + + // The snapshot on-chain is for blendv2 (CBLENDV2ADAPTER), not the + // best candidate defindex (CDEFINDEXADAPTER) that findBestCandidate + // will derive this run. + mockLiveAdapterAndActiveSnapshot( + DISCOVERED_VAULT.currentAdapterId, + "CBLENDV2ADAPTER" + ); + + // Rates: current (blend) = 500, defindex = 700 (best, improvement 200), + // blendv2 = 650 (still clears minImprovementBps=50, improvement 150). + // findBestCandidate picks defindex (700), but the #699 check should + // override to blendv2 (650) because that's what the snapshot holds. + const rateSource = vi.fn(async ({ adapterId }: { adapterId: string }) => { + if (adapterId === "CBLENDADAPTER") return 500; + if (adapterId === "CDEFINDEXADAPTER") return 700; + if (adapterId === "CBLENDV2ADAPTER") return 650; + return null; + }); + + const result = await runMigrationKeeper(configWithTwoCandidates, { + logger: logger(), + discoverVaults: async () => ({ + vaults: [DISCOVERED_VAULT], + failures: [], + }), + rateSource, + resolveCandidatePool: async () => "CSOMEPOOL", + sleep: vi.fn(), + }); + + // migrate_adapter should have been submitted (not begin_migration), + // and the migration target should be the snapshotted adapter + // (CBLENDV2ADAPTER), not the "best" one (CDEFINDEXADAPTER). + expect(server.sendTransaction).toHaveBeenCalledOnce(); + expect(result.migrations).toHaveLength(1); + expect(result.migrations[0]).toMatchObject({ + vaultId: "meridian-usdc", + fromAdapterId: "CBLENDADAPTER", + toAdapterId: "CBLENDV2ADAPTER", + toProtocol: "blendv2", + improvementBps: 150, + }); + expect(result.skipped).toEqual([]); + }); + + // #699: When the snapshotted adapter's rate has genuinely decayed below + // the threshold, the keeper should let the snapshot lapse and pick the + // fresh best candidate instead — this is the "only let the snapshot + // lapse once the snapshotted adapter's rate genuinely stops clearing + // the threshold" half of the fix. + it("replaces a stale migration snapshot whose adapter no longer clears the threshold (#699)", async () => { + const server = makeServer(); + stellarMocks.getRpcServer.mockReturnValue(server); + stellarMocks.waitForTransaction.mockResolvedValue({ ledger: 321 }); + + const configWithTwoCandidates: MigrationKeeperConfig = { + ...CONFIG, + candidateAdapters: { + defindex: "CDEFINDEXADAPTER", + blendv2: "CBLENDV2ADAPTER", + }, + }; + + // Snapshot is for blendv2, but blendv2's rate has dropped to 510 + // (only 10 bps over current=500, below minImprovementBps=50). + // defindex is still 700 (200 bps over current), so findBestCandidate + // picks defindex, and the #699 check should NOT override it. + mockLiveAdapterAndActiveSnapshot( + DISCOVERED_VAULT.currentAdapterId, + "CBLENDV2ADAPTER" + ); + + const rateSource = vi.fn(async ({ adapterId }: { adapterId: string }) => { + if (adapterId === "CBLENDADAPTER") return 500; + if (adapterId === "CDEFINDEXADAPTER") return 700; + if (adapterId === "CBLENDV2ADAPTER") return 510; + return null; + }); + + const result = await runMigrationKeeper(configWithTwoCandidates, { + logger: logger(), + discoverVaults: async () => ({ + vaults: [DISCOVERED_VAULT], + failures: [], + }), + rateSource, + resolveCandidatePool: async () => "CSOMEPOOL", + sleep: vi.fn(), + }); + + // begin_migration should have been submitted for the fresh best + // (defindex), not migrate_adapter for the stale snapshot (blendv2). + expect(server.sendTransaction).toHaveBeenCalledOnce(); + expect(result.migrations).toEqual([]); + expect(result.skipped).toMatchObject([ + { + vaultId: "meridian-usdc", + reason: expect.stringContaining("begin_migration submitted"), + }, + ]); + }); + it("releases the submission lease after begin_migration so a later run isn't blocked", async () => { const server = makeServer(); stellarMocks.getRpcServer.mockReturnValue(server); diff --git a/packages/stellar-sdk-helpers/src/migration-keeper.ts b/packages/stellar-sdk-helpers/src/migration-keeper.ts index 7cefdf82..6e9d437f 100644 --- a/packages/stellar-sdk-helpers/src/migration-keeper.ts +++ b/packages/stellar-sdk-helpers/src/migration-keeper.ts @@ -976,6 +976,10 @@ export async function runMigrationKeeper( // on the real path. let hasMatchingSnapshot = deps.submitMigration != null; let snapshotReadFailed = false; + // #699: Capture the snapshot's adapter so the preservation check below + // can decide whether to keep it even when findBestCandidate returned a + // different "best" candidate this run. + let existingSnapshotAdapter: string | null = null; if (!hasMatchingSnapshot) { try { const snapshot = (await simulateView( @@ -984,7 +988,8 @@ export async function runMigrationKeeper( config.network.passphrase, "get_migration_snapshot" )) as { adapter: string } | null; - hasMatchingSnapshot = snapshot?.adapter === best.adapterId; + existingSnapshotAdapter = snapshot?.adapter ?? null; + hasMatchingSnapshot = existingSnapshotAdapter === best.adapterId; } catch (err) { // A genuine "no snapshot" (or one for a different adapter) traps // with MigrationNotInitialized, which is not a transient failure by @@ -1001,6 +1006,144 @@ export async function runMigrationKeeper( } } + // #699: If an active snapshot exists for a different adapter than the + // freshly-derived "best", check whether that snapshotted adapter still + // clears minImprovementBps against the current rate. If it does, + // prefer completing the existing migration over switching to a newer + // candidate: without this, alternating rates between two adapters + // across scheduled runs would overwrite the already-begun snapshot on + // each run and reset the ledger-gap cooldown, so a real, sustained + // migration opportunity could in principle never reach + // migrate_adapter. Only let the snapshot lapse (and pick a new + // candidate) once the snapshotted adapter's rate genuinely stops + // clearing the threshold. + if ( + !hasMatchingSnapshot && + existingSnapshotAdapter && + existingSnapshotAdapter !== vault.currentAdapterId + ) { + // Reverse-lookup the snapshot adapter's protocol from the configured + // candidate adapters. If it's no longer configured, we can't query + // its rate, so fall through to begin_migration for the fresh best. + const snapshotProtocol = Object.entries(config.candidateAdapters).find( + ([, id]) => id === existingSnapshotAdapter + )?.[0]; + + if (snapshotProtocol) { + try { + // The snapshotted adapter's rate and the vault's current rate + // are queried concurrently: they're independent reads, so + // running them in sequence would needlessly double the latency + // of this check. Each is retried individually via + // withKeeperRetry, same as findBestCandidate's own candidate + // evaluation. + const [snapshotRateResult, currentRateResult] = await Promise.all([ + withKeeperRetry( + async () => + rateSource({ + protocol: snapshotProtocol, + adapterId: existingSnapshotAdapter, + poolId: await resolveCandidatePool(existingSnapshotAdapter), + ...(vault.assetId !== undefined && { + assetId: vault.assetId, + }), + }), + { + maxAttempts: config.maxAttempts, + baseDelayMs: config.baseDelayMs, + deadlineAt, + }, + logger, + { + vaultId: vault.vaultId, + adapterId: existingSnapshotAdapter, + protocol: snapshotProtocol, + stage: "evaluate", + }, + sleepFn, + isTransientKeeperError, + "migration-keeper" + ), + withKeeperRetry( + () => + rateSource({ + protocol: vault.currentProtocol, + adapterId: vault.currentAdapterId, + poolId: vault.currentPoolId, + ...(vault.assetId !== undefined && { + assetId: vault.assetId, + }), + }), + { + maxAttempts: config.maxAttempts, + baseDelayMs: config.baseDelayMs, + deadlineAt, + }, + logger, + { + vaultId: vault.vaultId, + adapterId: vault.currentAdapterId, + protocol: vault.currentProtocol, + stage: "evaluate", + }, + sleepFn, + isTransientKeeperError, + "migration-keeper" + ), + ]); + + const snapshotRate = snapshotRateResult.value; + const currentRate = currentRateResult.value; + + if ( + isUsableRate(snapshotRate) && + isUsableRate(currentRate) && + snapshotRate - currentRate >= config.minImprovementBps + ) { + // The snapshotted adapter still clears the threshold: override + // best so the existing begin_migration / migrate_adapter flow + // below completes it instead of overwriting the snapshot with + // a begin_migration for the new best candidate. + logger.info( + "[migration-keeper] preserving active migration snapshot (#699)", + { + vaultId: vault.vaultId, + snapshotAdapterId: existingSnapshotAdapter, + snapshotProtocol, + overriddenBestAdapterId: best.adapterId, + overriddenBestProtocol: best.protocol, + snapshotImprovementBps: snapshotRate - currentRate, + overriddenBestImprovementBps: best.improvementBps, + } + ); + best.adapterId = existingSnapshotAdapter; + best.protocol = snapshotProtocol; + best.improvementBps = snapshotRate - currentRate; + hasMatchingSnapshot = true; + } + // If the snapshot's adapter no longer clears the threshold, + // hasMatchingSnapshot stays false and begin_migration for the + // fresh best runs below — exactly the desired behaviour when the + // snapshotted adapter's rate has genuinely decayed. + } catch (err) { + // A transient failure during the snapshot's rate lookup must not + // block the migration entirely: the fresh best from + // findBestCandidate is still valid, so fall through to + // begin_migration for it. The snapshot will be overwritten, + // which is acceptable — the alternative (skipping the migration) + // is strictly worse for liveness. + logger.warn( + "[migration-keeper] snapshot rate lookup failed; falling through to fresh candidate (#699)", + { + vaultId: vault.vaultId, + snapshotAdapterId: existingSnapshotAdapter, + error: errorMessage(err), + } + ); + } + } + } + if (snapshotReadFailed) { failures.push({ vaultId: vault.vaultId, From 661ac4890e89f16b960a56044b5b15c9a07c4c73 Mon Sep 17 00:00:00 2001 From: Zac Lou <97340247+ZacLou@users.noreply.github.com> Date: Sat, 5 Sep 2026 01:20:50 +0800 Subject: [PATCH 2/5] refactor(#699): hoist snapshot read, remove duplicate post-hoc block --- .../src/migration-keeper.ts | 378 ++++++++---------- 1 file changed, 170 insertions(+), 208 deletions(-) diff --git a/packages/stellar-sdk-helpers/src/migration-keeper.ts b/packages/stellar-sdk-helpers/src/migration-keeper.ts index 6e9d437f..1a577d3d 100644 --- a/packages/stellar-sdk-helpers/src/migration-keeper.ts +++ b/packages/stellar-sdk-helpers/src/migration-keeper.ts @@ -125,6 +125,29 @@ function isUsableRate(rate: number | null): rate is number { return rate !== null && Number.isFinite(rate); } +/** Returns true when `rate - currentRate >= minImprovementBps`. Extracted + * so the fresh-candidate path and the snapshot-preservation path share one + * comparison — a future change to threshold semantics (e.g. rounding-aware + * or percentage-based) only changes the comparison in one place. */ +function clearsImprovementThreshold( + rate: number, + currentRate: number, + minImprovementBps: number +): boolean { + return rate - currentRate >= minImprovementBps; +} + +/** A candidate that may already have an on-chain migration snapshot. When + * `findBestCandidate` receives one, it evaluates the pinned candidate in + * the same concurrent batch as every other candidate. If the pinned + * candidate clears the improvement threshold, it is preferred over any + * other candidate regardless of improvement magnitude — completing the + * existing migration avoids resetting the ledger-gap cooldown (see #699). */ +interface PinnedCandidate { + adapterId: string; + protocol: string; +} + export interface MigrationKeeperConfig { network: StellarNetwork; secretKey: string; @@ -520,7 +543,13 @@ async function findBestCandidate( resolveCandidatePool: (adapterId: string) => Promise, logger: KeeperLogger, sleepFn: (ms: number) => Promise, - deadlineAt: number + deadlineAt: number, + /** An existing on-chain migration snapshot to evaluate alongside fresh + * candidates in the same concurrent batch (#699). When the pinned + * candidate clears the improvement threshold, it is preferred over + * every other candidate — completing the existing migration avoids + * resetting the ledger-gap cooldown. */ + pinned?: PinnedCandidate ): Promise<{ best: BestCandidate | null; skipReason?: string }> { // Nothing to compare against: don't pay for a retried rate lookup (up to // maxAttempts, with backoff) just to discover there was never a candidate @@ -540,10 +569,16 @@ async function findBestCandidate( // to compare regardless of what the current rate turns out to be, so // don't pay for that retried lookup only to discover there was never // anything to evaluate it against. + // + // When a pinned candidate is present it is evaluated in the same + // concurrent batch rather than as a separate post-hoc check, so that + // future paths needing to reason about "the current best vs. an existing + // on-chain commitment" reuse one generalized candidate-evaluation pass + // without duplicating rate fetches or threshold comparisons (#699). const candidates = Object.entries(config.candidateAdapters).filter( ([, adapterId]) => adapterId !== vault.currentAdapterId ); - if (candidates.length === 0) { + if (candidates.length === 0 && !pinned) { return { best: null, skipReason: "every configured candidate is the vault's current adapter", @@ -588,34 +623,56 @@ async function findBestCandidate( return { best: null, skipReason: "current rate unavailable" }; } - // Evaluated concurrently, like discovery above: each candidate's pool - // resolution and rate lookup is independent of every other candidate, so - // running them one at a time would let the deadline budget get eaten by - // earlier candidates before later ones are even attempted. + // Evaluated concurrently: regular candidates plus the pinned (snapshot) + // candidate when one exists. Each entry's pool resolution and rate + // lookup is independent of every other entry, so running them one at a + // time would let the deadline budget get eaten by earlier entries before + // later ones are even attempted. Including the pinned candidate here + // rather than in a separate post-hoc block means its rate is fetched + // once, not twice, and the improvement threshold is compared in one + // logical place. + const evaluate = (protocol: string, adapterId: string) => + withKeeperRetry( + async () => { + const poolId = await resolveCandidatePool(adapterId); + return rateSource({ + protocol, + adapterId, + poolId, + ...(vault.assetId !== undefined && { assetId: vault.assetId }), + }); + }, + { + maxAttempts: config.maxAttempts, + baseDelayMs: config.baseDelayMs, + deadlineAt, + }, + logger, + { vaultId: vault.vaultId, adapterId, protocol, stage: "evaluate" }, + sleepFn, + isTransientKeeperError, + "migration-keeper" + ); + + // Tag each entry with its kind and the mark that drives the preference + // rule below (pinned entries win when eligible, other entries compete + // by improvement magnitude). + type CandidateEntry = { + protocol: string; + adapterId: string; + pinned: boolean; + }; + const entries: CandidateEntry[] = candidates.map( + ([protocol, adapterId]) => ({ protocol, adapterId, pinned: false }) + ); + if (pinned) { + entries.push({ protocol: pinned.protocol, adapterId: pinned.adapterId, pinned: true }); + } + const settled = await Promise.allSettled( - candidates.map(async ([protocol, adapterId]) => { - const result = await withKeeperRetry( - async () => { - const poolId = await resolveCandidatePool(adapterId); - return rateSource({ - protocol, - adapterId, - poolId, - ...(vault.assetId !== undefined && { assetId: vault.assetId }), - }); - }, - { - maxAttempts: config.maxAttempts, - baseDelayMs: config.baseDelayMs, - deadlineAt, - }, - logger, - { vaultId: vault.vaultId, adapterId, protocol, stage: "evaluate" }, - sleepFn, - isTransientKeeperError, - "migration-keeper" - ); - return { protocol, adapterId, rate: result.value }; + entries.map(async (entry) => { + const result = await evaluate(entry.protocol, entry.adapterId); + return { protocol: entry.protocol, adapterId: entry.adapterId, rate: result.value, pinned: entry.pinned }; }) ); @@ -625,14 +682,20 @@ async function findBestCandidate( // another. Only surface the failure if nothing usable came out of any // candidate at all. let best: BestCandidate | null = null; + // A pinned candidate that clears the threshold wins, full stop — the + // policy is "complete the existing migration rather than reset the + // cooldown", and the improvement delta between it and a different "best" + // is irrelevant because switching would restart the timer. This flag + // prevents a higher-improvement non-pinned candidate from overriding + // a pinned candidate that also clears the threshold. + let pinnedClears = false; let firstFailure: { protocol: string; adapterId: string; reason: unknown } | undefined; let anyRateKnown = false; for (let i = 0; i < settled.length; i++) { const outcome = settled[i]; - const target = candidates[i]; - if (!outcome || !target) continue; - const [protocol, adapterId] = target; + const entry = entries[i]; + if (!outcome || !entry) continue; if (outcome.status === "rejected") { // Only the first rejection becomes the vault's reported // CandidateEvaluationError (KeeperFailure is per-vault, not @@ -642,11 +705,11 @@ async function findBestCandidate( // first. logger.warn("[migration-keeper] candidate evaluation failed", { vaultId: vault.vaultId, - adapterId, - protocol, + adapterId: entry.adapterId, + protocol: entry.protocol, error: errorMessage(outcome.reason), }); - firstFailure ??= { protocol, adapterId, reason: outcome.reason }; + firstFailure ??= { protocol: entry.protocol, adapterId: entry.adapterId, reason: outcome.reason }; continue; } const { rate } = outcome.value; @@ -654,14 +717,28 @@ async function findBestCandidate( anyRateKnown = true; const improvementBps = rate - currentRate; - if (improvementBps < config.minImprovementBps) continue; + if (!clearsImprovementThreshold(rate, currentRate, config.minImprovementBps)) continue; + + // Pinned candidate clearing the threshold wins unconditionally: + // completing the existing migration is the only way to reach + // migrate_adapter without resetting the cooldown, and the alternative + // of picking a higher-improvement candidate would overwrite the + // snapshot and restart the timer (see #699). The pinned candidate must + // not be discarded in favour of a non-pinned one even when the latter + // has a marginally higher improvement — "the existing migration + // survives" is the entire point of this mechanism. + if (entry.pinned) { + pinnedClears = true; + best = { protocol: entry.protocol, adapterId: entry.adapterId, improvementBps }; + break; + } if (!best || improvementBps > best.improvementBps) { - best = { protocol, adapterId, improvementBps }; + best = { protocol: entry.protocol, adapterId: entry.adapterId, improvementBps }; } } if (best) { - return { best }; + return { best, ...(pinnedClears && { skipReason: undefined }) }; } // A failed candidate must never block a valid decision reached from a // different candidate, this applies just as much when that decision is @@ -871,6 +948,44 @@ export async function runMigrationKeeper( }); } + // Read the on-chain migration snapshot before findBestCandidate so the + // snapshotted adapter can be evaluated alongside fresh candidates in + // the same concurrent batch rather than in a separate post-hoc check + // (#699). The snapshotted adapter is passed as a "pinned" candidate: + // findBestCandidate evaluates it in the same Promise.allSettled, + // reusing the rate fetch and threshold comparison, and when the pinned + // candidate clears the threshold it wins unconditionally — completing + // the existing migration instead of resetting the ledger-gap cooldown. + let hasMatchingSnapshot = deps.submitMigration != null; + let snapshotReadFailed = false; + let existingSnapshotAdapter: string | null = null; + let pinned: PinnedCandidate | undefined; + + if (!hasMatchingSnapshot) { + try { + const snapshot = (await simulateView( + server as never, + vault.vaultContractId, + config.network.passphrase, + "get_migration_snapshot" + )) as { adapter: string } | null; + existingSnapshotAdapter = snapshot?.adapter ?? null; + if ( + existingSnapshotAdapter && + existingSnapshotAdapter !== vault.currentAdapterId + ) { + const snapshotProtocol = Object.entries( + config.candidateAdapters + ).find(([, id]) => id === existingSnapshotAdapter)?.[0]; + if (snapshotProtocol) { + pinned = { adapterId: existingSnapshotAdapter, protocol: snapshotProtocol }; + } + } + } catch (err) { + snapshotReadFailed = isTransientKeeperError(err); + } + } + let evaluation: { best: BestCandidate | null; skipReason?: string }; try { evaluation = await findBestCandidate( @@ -880,7 +995,8 @@ export async function runMigrationKeeper( resolveCandidatePool, logger, sleepFn, - deadlineAt + deadlineAt, + pinned ); } catch (err) { const { adapterId, protocol, attempts, transient } = @@ -974,175 +1090,21 @@ export async function runMigrationKeeper( // network/mocks regardless of the injected override, same as // assertAdapterUnchanged inside submitMigrationTransaction only runs // on the real path. - let hasMatchingSnapshot = deps.submitMigration != null; - let snapshotReadFailed = false; - // #699: Capture the snapshot's adapter so the preservation check below - // can decide whether to keep it even when findBestCandidate returned a - // different "best" candidate this run. - let existingSnapshotAdapter: string | null = null; - if (!hasMatchingSnapshot) { - try { - const snapshot = (await simulateView( - server as never, - vault.vaultContractId, - config.network.passphrase, - "get_migration_snapshot" - )) as { adapter: string } | null; - existingSnapshotAdapter = snapshot?.adapter ?? null; - hasMatchingSnapshot = existingSnapshotAdapter === best.adapterId; - } catch (err) { - // A genuine "no snapshot" (or one for a different adapter) traps - // with MigrationNotInitialized, which is not a transient failure by - // isTransientKeeperError's classification — that's the case this - // falls through to begin_migration for. A real RPC/network failure - // reading the snapshot must NOT be treated the same way: assuming - // "no snapshot" and firing begin_migration would reset a possibly - // already cooldown-elapsed snapshot's ledger_seq back to now, - // pushing a ready-to-migrate vault's migration back a full - // MIN_LEDGER_GAP for no reason. Report it as a retryable failure - // instead, same as this file's other on-chain read checks. - hasMatchingSnapshot = false; - snapshotReadFailed = isTransientKeeperError(err); - } - } - - // #699: If an active snapshot exists for a different adapter than the - // freshly-derived "best", check whether that snapshotted adapter still - // clears minImprovementBps against the current rate. If it does, - // prefer completing the existing migration over switching to a newer - // candidate: without this, alternating rates between two adapters - // across scheduled runs would overwrite the already-begun snapshot on - // each run and reset the ledger-gap cooldown, so a real, sustained - // migration opportunity could in principle never reach - // migrate_adapter. Only let the snapshot lapse (and pick a new - // candidate) once the snapshotted adapter's rate genuinely stops - // clearing the threshold. - if ( - !hasMatchingSnapshot && - existingSnapshotAdapter && - existingSnapshotAdapter !== vault.currentAdapterId - ) { - // Reverse-lookup the snapshot adapter's protocol from the configured - // candidate adapters. If it's no longer configured, we can't query - // its rate, so fall through to begin_migration for the fresh best. - const snapshotProtocol = Object.entries(config.candidateAdapters).find( - ([, id]) => id === existingSnapshotAdapter - )?.[0]; - - if (snapshotProtocol) { - try { - // The snapshotted adapter's rate and the vault's current rate - // are queried concurrently: they're independent reads, so - // running them in sequence would needlessly double the latency - // of this check. Each is retried individually via - // withKeeperRetry, same as findBestCandidate's own candidate - // evaluation. - const [snapshotRateResult, currentRateResult] = await Promise.all([ - withKeeperRetry( - async () => - rateSource({ - protocol: snapshotProtocol, - adapterId: existingSnapshotAdapter, - poolId: await resolveCandidatePool(existingSnapshotAdapter), - ...(vault.assetId !== undefined && { - assetId: vault.assetId, - }), - }), - { - maxAttempts: config.maxAttempts, - baseDelayMs: config.baseDelayMs, - deadlineAt, - }, - logger, - { - vaultId: vault.vaultId, - adapterId: existingSnapshotAdapter, - protocol: snapshotProtocol, - stage: "evaluate", - }, - sleepFn, - isTransientKeeperError, - "migration-keeper" - ), - withKeeperRetry( - () => - rateSource({ - protocol: vault.currentProtocol, - adapterId: vault.currentAdapterId, - poolId: vault.currentPoolId, - ...(vault.assetId !== undefined && { - assetId: vault.assetId, - }), - }), - { - maxAttempts: config.maxAttempts, - baseDelayMs: config.baseDelayMs, - deadlineAt, - }, - logger, - { - vaultId: vault.vaultId, - adapterId: vault.currentAdapterId, - protocol: vault.currentProtocol, - stage: "evaluate", - }, - sleepFn, - isTransientKeeperError, - "migration-keeper" - ), - ]); - - const snapshotRate = snapshotRateResult.value; - const currentRate = currentRateResult.value; - - if ( - isUsableRate(snapshotRate) && - isUsableRate(currentRate) && - snapshotRate - currentRate >= config.minImprovementBps - ) { - // The snapshotted adapter still clears the threshold: override - // best so the existing begin_migration / migrate_adapter flow - // below completes it instead of overwriting the snapshot with - // a begin_migration for the new best candidate. - logger.info( - "[migration-keeper] preserving active migration snapshot (#699)", - { - vaultId: vault.vaultId, - snapshotAdapterId: existingSnapshotAdapter, - snapshotProtocol, - overriddenBestAdapterId: best.adapterId, - overriddenBestProtocol: best.protocol, - snapshotImprovementBps: snapshotRate - currentRate, - overriddenBestImprovementBps: best.improvementBps, - } - ); - best.adapterId = existingSnapshotAdapter; - best.protocol = snapshotProtocol; - best.improvementBps = snapshotRate - currentRate; - hasMatchingSnapshot = true; - } - // If the snapshot's adapter no longer clears the threshold, - // hasMatchingSnapshot stays false and begin_migration for the - // fresh best runs below — exactly the desired behaviour when the - // snapshotted adapter's rate has genuinely decayed. - } catch (err) { - // A transient failure during the snapshot's rate lookup must not - // block the migration entirely: the fresh best from - // findBestCandidate is still valid, so fall through to - // begin_migration for it. The snapshot will be overwritten, - // which is acceptable — the alternative (skipping the migration) - // is strictly worse for liveness. - logger.warn( - "[migration-keeper] snapshot rate lookup failed; falling through to fresh candidate (#699)", - { - vaultId: vault.vaultId, - snapshotAdapterId: existingSnapshotAdapter, - error: errorMessage(err), - } - ); - } - } - } + // + // Re-evaluated from best after findBestCandidate, which already + // evaluated the snapshotted adapter as a pinned candidate in the + // same concurrent batch (#699). When the pinned candidate won, + // best.adapterId already holds the snapshotted adapter's ID; when a + // non-pinned candidate won or no snapshot exists, the snapshot + // doesn't match and falls through to begin_migration below. The + // on-chain get_migration_snapshot read was moved to BEFORE + // findBestCandidate (see the pinned construction above) so the + // snapshotted adapter's rate is fetched once, not twice, and the + // improvement-threshold comparison is in one logical place. + hasMatchingSnapshot = + hasMatchingSnapshot || + (existingSnapshotAdapter != null && + existingSnapshotAdapter === best.adapterId); if (snapshotReadFailed) { failures.push({ From 286dfe9e407677096a8a34b69238de436d73bd30 Mon Sep 17 00:00:00 2001 From: Zac Lou <97340247+ZacLou@users.noreply.github.com> Date: Sat, 5 Sep 2026 11:09:02 +0800 Subject: [PATCH 3/5] fix(#699): deduplicate pinned candidate from regular candidates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove the double-evaluation bug where the pinned snapshot adapter was evaluated twice — once as a regular candidate (never filtered out from the candidates list) and once as a pinned entry, doubling resolveCandidatePool/rateSource RPC calls. Fix: exclude the pinned adapter from the regular candidates filter. Also remove the dead pinnedClears flag whose sole consumer was a conditional spread producing a skipReason: undefined that tripped CI's exactOptionalPropertyTypes: true. The early break in the loop already guarantees pinned-candidate priority. --- .../src/migration-keeper.ts | 68 ++++++++++++++----- 1 file changed, 51 insertions(+), 17 deletions(-) diff --git a/packages/stellar-sdk-helpers/src/migration-keeper.ts b/packages/stellar-sdk-helpers/src/migration-keeper.ts index 1a577d3d..9fbf2189 100644 --- a/packages/stellar-sdk-helpers/src/migration-keeper.ts +++ b/packages/stellar-sdk-helpers/src/migration-keeper.ts @@ -575,8 +575,15 @@ async function findBestCandidate( // future paths needing to reason about "the current best vs. an existing // on-chain commitment" reuse one generalized candidate-evaluation pass // without duplicating rate fetches or threshold comparisons (#699). + // + // The pinned adapter is excluded from the regular `candidates` list so + // it is never evaluated twice: it appears once as a pinned entry (added + // below), not once as a regular candidate and once as pinned. When the + // snapshot adapter is not in the configured candidates at all, + // candidates is unchanged (no entry to de-duplicate). const candidates = Object.entries(config.candidateAdapters).filter( - ([, adapterId]) => adapterId !== vault.currentAdapterId + ([, adapterId]) => + adapterId !== vault.currentAdapterId && adapterId !== pinned?.adapterId ); if (candidates.length === 0 && !pinned) { return { @@ -662,17 +669,28 @@ async function findBestCandidate( adapterId: string; pinned: boolean; }; - const entries: CandidateEntry[] = candidates.map( - ([protocol, adapterId]) => ({ protocol, adapterId, pinned: false }) - ); + const entries: CandidateEntry[] = candidates.map(([protocol, adapterId]) => ({ + protocol, + adapterId, + pinned: false, + })); if (pinned) { - entries.push({ protocol: pinned.protocol, adapterId: pinned.adapterId, pinned: true }); + entries.push({ + protocol: pinned.protocol, + adapterId: pinned.adapterId, + pinned: true, + }); } const settled = await Promise.allSettled( entries.map(async (entry) => { const result = await evaluate(entry.protocol, entry.adapterId); - return { protocol: entry.protocol, adapterId: entry.adapterId, rate: result.value, pinned: entry.pinned }; + return { + protocol: entry.protocol, + adapterId: entry.adapterId, + rate: result.value, + pinned: entry.pinned, + }; }) ); @@ -685,10 +703,9 @@ async function findBestCandidate( // A pinned candidate that clears the threshold wins, full stop — the // policy is "complete the existing migration rather than reset the // cooldown", and the improvement delta between it and a different "best" - // is irrelevant because switching would restart the timer. This flag - // prevents a higher-improvement non-pinned candidate from overriding - // a pinned candidate that also clears the threshold. - let pinnedClears = false; + // is irrelevant because switching would restart the timer. The early + // `break` below prevents a higher-improvement non-pinned candidate from + // overriding a pinned candidate that also clears the threshold. let firstFailure: { protocol: string; adapterId: string; reason: unknown } | undefined; let anyRateKnown = false; @@ -709,7 +726,11 @@ async function findBestCandidate( protocol: entry.protocol, error: errorMessage(outcome.reason), }); - firstFailure ??= { protocol: entry.protocol, adapterId: entry.adapterId, reason: outcome.reason }; + firstFailure ??= { + protocol: entry.protocol, + adapterId: entry.adapterId, + reason: outcome.reason, + }; continue; } const { rate } = outcome.value; @@ -717,7 +738,10 @@ async function findBestCandidate( anyRateKnown = true; const improvementBps = rate - currentRate; - if (!clearsImprovementThreshold(rate, currentRate, config.minImprovementBps)) continue; + if ( + !clearsImprovementThreshold(rate, currentRate, config.minImprovementBps) + ) + continue; // Pinned candidate clearing the threshold wins unconditionally: // completing the existing migration is the only way to reach @@ -728,17 +752,24 @@ async function findBestCandidate( // has a marginally higher improvement — "the existing migration // survives" is the entire point of this mechanism. if (entry.pinned) { - pinnedClears = true; - best = { protocol: entry.protocol, adapterId: entry.adapterId, improvementBps }; + best = { + protocol: entry.protocol, + adapterId: entry.adapterId, + improvementBps, + }; break; } if (!best || improvementBps > best.improvementBps) { - best = { protocol: entry.protocol, adapterId: entry.adapterId, improvementBps }; + best = { + protocol: entry.protocol, + adapterId: entry.adapterId, + improvementBps, + }; } } if (best) { - return { best, ...(pinnedClears && { skipReason: undefined }) }; + return { best }; } // A failed candidate must never block a valid decision reached from a // different candidate, this applies just as much when that decision is @@ -978,7 +1009,10 @@ export async function runMigrationKeeper( config.candidateAdapters ).find(([, id]) => id === existingSnapshotAdapter)?.[0]; if (snapshotProtocol) { - pinned = { adapterId: existingSnapshotAdapter, protocol: snapshotProtocol }; + pinned = { + adapterId: existingSnapshotAdapter, + protocol: snapshotProtocol, + }; } } } catch (err) { From 656636eaff497a33ccd870741ed182e1f3356a1d Mon Sep 17 00:00:00 2001 From: Zac Lou <97340247+ZacLou@users.noreply.github.com> Date: Sun, 6 Sep 2026 00:53:01 +0800 Subject: [PATCH 4/5] test(migration-keeper): pinned-candidate edge cases (#705) --- .../src/migration-keeper.test.ts | 96 +++++++++++++++++++ 1 file changed, 96 insertions(+) diff --git a/packages/stellar-sdk-helpers/src/migration-keeper.test.ts b/packages/stellar-sdk-helpers/src/migration-keeper.test.ts index 748c114f..0fb85751 100644 --- a/packages/stellar-sdk-helpers/src/migration-keeper.test.ts +++ b/packages/stellar-sdk-helpers/src/migration-keeper.test.ts @@ -1531,6 +1531,102 @@ describe("runMigrationKeeper", () => { ]); }); + // Coverage for the branch where candidates is empty but a pinned snapshot + // exists (the pinned adapter is the only non-current candidate). + it("migrates via pinned candidate when it is the only non-current candidate (#699)", async () => { + const server = makeServer(); + stellarMocks.getRpcServer.mockReturnValue(server); + stellarMocks.waitForTransaction.mockResolvedValue({ ledger: 321 }); + + // Only one candidate adapter, which is also the pinned snapshot adapter. + const configWithOneCandidate: MigrationKeeperConfig = { + ...CONFIG, + candidateAdapters: { + blendv2: "CBLENDV2ADAPTER", + }, + }; + + mockLiveAdapterAndActiveSnapshot( + DISCOVERED_VAULT.currentAdapterId, + "CBLENDV2ADAPTER" + ); + + const rateSource = vi.fn(async ({ adapterId }: { adapterId: string }) => { + if (adapterId === "CBLENDADAPTER") return 500; + if (adapterId === "CBLENDV2ADAPTER") return 600; // 100 bps improvement, clears 50 + return null; + }); + + const result = await runMigrationKeeper(configWithOneCandidate, { + logger: logger(), + discoverVaults: async () => ({ + vaults: [DISCOVERED_VAULT], + failures: [], + }), + rateSource, + resolveCandidatePool: async () => "CSOMEPOOL", + sleep: vi.fn(), + }); + + // migrate_adapter should have been submitted for the pinned adapter. + expect(server.sendTransaction).toHaveBeenCalledOnce(); + expect(result.migrations).toHaveLength(1); + expect(result.migrations[0]).toMatchObject({ + vaultId: "meridian-usdc", + fromAdapterId: "CBLENDADAPTER", + toAdapterId: "CBLENDV2ADAPTER", + toProtocol: "blendv2", + improvementBps: 100, + }); + }); + + // Coverage for clearsImprovementThreshold at the exact boundary. + it("preserves a snapshot whose improvement is exactly at the threshold boundary (#699)", async () => { + const server = makeServer(); + stellarMocks.getRpcServer.mockReturnValue(server); + stellarMocks.waitForTransaction.mockResolvedValue({ ledger: 321 }); + + const configWithTwoCandidates: MigrationKeeperConfig = { + ...CONFIG, + candidateAdapters: { + defindex: "CDEFINDEXADAPTER", + blendv2: "CBLENDV2ADAPTER", + }, + }; + + mockLiveAdapterAndActiveSnapshot( + DISCOVERED_VAULT.currentAdapterId, + "CBLENDV2ADAPTER" + ); + + // blendv2 rate = 550, current = 500, improvement = 50 (exactly at threshold). + const rateSource = vi.fn(async ({ adapterId }: { adapterId: string }) => { + if (adapterId === "CBLENDADAPTER") return 500; + if (adapterId === "CDEFINDEXADAPTER") return 700; + if (adapterId === "CBLENDV2ADAPTER") return 550; + return null; + }); + + const result = await runMigrationKeeper(configWithTwoCandidates, { + logger: logger(), + discoverVaults: async () => ({ + vaults: [DISCOVERED_VAULT], + failures: [], + }), + rateSource, + resolveCandidatePool: async () => "CSOMEPOOL", + sleep: vi.fn(), + }); + + // The pinned candidate clears the threshold (>=), so it should be preferred. + expect(server.sendTransaction).toHaveBeenCalledOnce(); + expect(result.migrations).toHaveLength(1); + expect(result.migrations[0]).toMatchObject({ + toAdapterId: "CBLENDV2ADAPTER", + improvementBps: 50, + }); + }); + it("releases the submission lease after begin_migration so a later run isn't blocked", async () => { const server = makeServer(); stellarMocks.getRpcServer.mockReturnValue(server); From 79122d2fa1fc500e4ec20e85735aacf4961a1c1a Mon Sep 17 00:00:00 2001 From: ZacLou Date: Sun, 6 Sep 2026 08:21:28 +0800 Subject: [PATCH 5/5] fix(#699): address review feedback on pinned-candidate handling - Fail fast when a pinned snapshot candidate's rate fetch is rejected, preventing a non-pinned candidate from winning due to an RPC blip. - Skip findBestCandidate entirely when the on-chain snapshot read fails, avoiding wasted deadline/RPC budget. - Remove redundant existingSnapshotAdapter variable; use pinned?.adapterId. - Remove now-dead snapshotReadFailed post-evaluation block. Closes review feedback on PR #705. --- .../src/migration-keeper.ts | 66 +++++++++++-------- 1 file changed, 39 insertions(+), 27 deletions(-) diff --git a/packages/stellar-sdk-helpers/src/migration-keeper.ts b/packages/stellar-sdk-helpers/src/migration-keeper.ts index 9fbf2189..1607173e 100644 --- a/packages/stellar-sdk-helpers/src/migration-keeper.ts +++ b/packages/stellar-sdk-helpers/src/migration-keeper.ts @@ -694,6 +694,23 @@ async function findBestCandidate( }) ); + // When a pinned candidate exists but its rate fetch failed, we cannot + // safely fall back to a non-pinned winner: the pinned entry represents + // an in-progress migration that must complete, and letting a transient + // RPC blip on the pinned adapter allow another candidate to win would + // reset the cooldown. Surface the failure so it is retried. + if (pinned) { + const pinnedIndex = entries.findIndex((e) => e.pinned); + const pinnedOutcome = pinnedIndex >= 0 ? settled[pinnedIndex] : undefined; + if (pinnedOutcome?.status === "rejected") { + throw new CandidateEvaluationError( + pinned.protocol, + pinned.adapterId, + pinnedOutcome.reason + ); + } + } + // A candidate that failed to evaluate must never discard a different // candidate that succeeded: an unrelated RPC blip on one protocol // shouldn't block a genuine, already-computed migration opportunity on @@ -989,7 +1006,6 @@ export async function runMigrationKeeper( // the existing migration instead of resetting the ledger-gap cooldown. let hasMatchingSnapshot = deps.submitMigration != null; let snapshotReadFailed = false; - let existingSnapshotAdapter: string | null = null; let pinned: PinnedCandidate | undefined; if (!hasMatchingSnapshot) { @@ -1000,17 +1016,14 @@ export async function runMigrationKeeper( config.network.passphrase, "get_migration_snapshot" )) as { adapter: string } | null; - existingSnapshotAdapter = snapshot?.adapter ?? null; - if ( - existingSnapshotAdapter && - existingSnapshotAdapter !== vault.currentAdapterId - ) { + const snapshotAdapter = snapshot?.adapter ?? null; + if (snapshotAdapter && snapshotAdapter !== vault.currentAdapterId) { const snapshotProtocol = Object.entries( config.candidateAdapters - ).find(([, id]) => id === existingSnapshotAdapter)?.[0]; + ).find(([, id]) => id === snapshotAdapter)?.[0]; if (snapshotProtocol) { pinned = { - adapterId: existingSnapshotAdapter, + adapterId: snapshotAdapter, protocol: snapshotProtocol, }; } @@ -1020,6 +1033,23 @@ export async function runMigrationKeeper( } } + // If reading the on-chain snapshot failed, don't waste deadline/RPC + // budget evaluating candidates only to discard the result below. + if (snapshotReadFailed) { + failures.push({ + vaultId: vault.vaultId, + vaultContractId: vault.vaultContractId, + adapterId: vault.currentAdapterId, + protocol: vault.currentProtocol, + stage: "evaluate", + attempts: 1, + transient: true, + error: + "could not read the migration snapshot; skipped rather than risk resetting an existing cooldown", + }); + continue; + } + let evaluation: { best: BestCandidate | null; skipReason?: string }; try { evaluation = await findBestCandidate( @@ -1136,25 +1166,7 @@ export async function runMigrationKeeper( // snapshotted adapter's rate is fetched once, not twice, and the // improvement-threshold comparison is in one logical place. hasMatchingSnapshot = - hasMatchingSnapshot || - (existingSnapshotAdapter != null && - existingSnapshotAdapter === best.adapterId); - - if (snapshotReadFailed) { - failures.push({ - vaultId: vault.vaultId, - vaultContractId: vault.vaultContractId, - adapterId: best.adapterId, - protocol: best.protocol, - stage: "submit", - attempts: 1, - transient: true, - error: - "could not read the migration snapshot; skipped rather than risk resetting an existing cooldown", - }); - await lease.releaseIfUnsent(); - continue; - } + hasMatchingSnapshot || pinned?.adapterId === best.adapterId; if (!hasMatchingSnapshot) { try {