Repository navigation
fix(plugin): refresh the served catalog on the configured TTL instead of publishing once at install #18235
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
fix(plugin): refresh the served catalog on the configured TTL instead of publishing once at install #18235
Changes from all commits
334fb40
d4e7c50
b19cbd5
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -17,12 +17,14 @@ import { buildProviderPayload, collectCatalog } from "./catalog.js"; | |
| import { usesModelsComboSource } from "./combos-source.js"; | ||
| import { | ||
| UNREACHABLE_COOLDOWN_MS, | ||
| MIN_REFRESH_INTERVAL_MS, | ||
| memoryCacheKey, | ||
| readDiskSnapshot, | ||
| snapshotIdentityFingerprint, | ||
| writeDiskSnapshot, | ||
| type CatalogSnapshot, | ||
| } from "./cache.js"; | ||
| import { startRefreshLoop } from "./refresh-loop.js"; | ||
| import { assertContext } from "./compat.js"; | ||
| import { type ApiKeyOrigin, resolveApiKey, warnIfMissing } from "./credentials.js"; | ||
| import { createSourceErrorReporter } from "./enrichment-report.js"; | ||
|
|
@@ -136,6 +138,11 @@ export default Plugin.define({ | |
| unreachableUntil: 0, | ||
| }; | ||
|
|
||
| // Per-installation stopped flag for the background refresh loop. Declared | ||
| // up front so the closures below can reference it without a temporal | ||
| // dead zone, and so a tick that lands after teardown publishes nothing. | ||
| let stopped = false; | ||
|
|
||
| // The credential the host holds wins over one written in config, so a | ||
| // user who connected the integration from the UI never has to paste a | ||
| // key into `opencode.json`. Reading it is async and the transforms must | ||
|
|
@@ -164,9 +171,16 @@ export default Plugin.define({ | |
| if (credentialChecked && apiKeyOrigin !== "missing") return; | ||
| const next = await resolveApiKey(ctx, X, parsed.apiKey, log); | ||
| const moved = next.key !== resolved.apiKey; | ||
| const previousCacheKey = cacheKey; | ||
| resolved.apiKey = next.key; | ||
| apiKeyOrigin = next.origin; | ||
| if (moved) ({ cacheKey, identityFingerprint } = credentialsOf()); | ||
| if (moved) { | ||
| ({ cacheKey, identityFingerprint } = credentialsOf()); | ||
| // Drop the stale entry so the memory cache does not grow across a | ||
| // credential change. | ||
| state.entries.delete(previousCacheKey); | ||
| state.inFlight.delete(previousCacheKey); | ||
| } | ||
| if (!credentialChecked) warnIfMissing(next, X, log); | ||
| else if (moved) log.info(`[omniroute-v2] API key picked up from the ${next.origin} source`); | ||
| credentialChecked = true; | ||
|
|
@@ -304,7 +318,7 @@ export default Plugin.define({ | |
| enrichment: previous?.enrichment ?? new Map(), | ||
| fetchedAt: Date.now(), | ||
| }; | ||
| if (models.length > 0) { | ||
| if (!stopped && models.length > 0) { | ||
| state.entries.set(cacheKey, snapshot); | ||
| await writeDiskSnapshot(X, snapshot, identityFingerprint, log); | ||
| } | ||
|
|
@@ -334,6 +348,7 @@ export default Plugin.define({ | |
| SourceResult<OmniRouteEnrichmentMap>, | ||
| ] | ||
| ): Promise<void> { | ||
| if (stopped) return; | ||
| if (state.entries.get(cacheKey) !== base) return; | ||
| // Per source: a success replaces (even with an empty answer — that is | ||
| // the gateway's answer), a failure keeps what we had. | ||
|
|
@@ -362,7 +377,7 @@ export default Plugin.define({ | |
| ); | ||
| const optionalChanged = state.optionalFingerprint !== optionalFingerprint; | ||
| state.optionalFingerprint = optionalFingerprint; | ||
| if (optionalChanged) { | ||
| if (optionalChanged && !stopped) { | ||
| try { | ||
| await ctx.provider.reload(); | ||
| } catch (err) { | ||
|
|
@@ -373,10 +388,11 @@ export default Plugin.define({ | |
| } | ||
| } | ||
|
|
||
| function loadSnapshot(): Promise<CatalogSnapshot> { | ||
| function loadSnapshot(force = false): Promise<CatalogSnapshot> { | ||
| const now = Date.now(); | ||
| const hit = state.entries.get(cacheKey); | ||
| if (hit && hit.fetchedAt + resolved.modelCacheTtlMs > now) return Promise.resolve(hit); | ||
| if (!force && hit && hit.fetchedAt + resolved.modelCacheTtlMs > now) | ||
| return Promise.resolve(hit); | ||
| // Cooldown after a total models failure: skip the network until it | ||
| // lapses. Serves last-known-good when one exists; otherwise the refresh | ||
| // below still runs (nothing to serve, no point pretending). | ||
|
|
@@ -427,10 +443,13 @@ export default Plugin.define({ | |
| // setup publishes first, then the host replays the callback (during | ||
| // registration and on every reload) and reads the published snapshot. | ||
| let latest: { info: unknown; models: unknown[] } | undefined; | ||
| const refreshAndPublish = async (): Promise<void> => { | ||
| const refreshAndPublish = async (options?: { force?: boolean }): Promise<void> => { | ||
| if (stopped) return; | ||
| await ensureCredential(); | ||
| if (stopped) return; | ||
| await ensureWarmSnapshot(); | ||
| const snapshot = await loadSnapshot(); | ||
| const snapshot = await loadSnapshot(options?.force === true); | ||
| if (stopped) return; | ||
| let effective = snapshot; | ||
| if (snapshot.models.length === 0) { | ||
| const stale = state.entries.get(cacheKey); | ||
|
|
@@ -468,7 +487,7 @@ export default Plugin.define({ | |
| const fingerprint = catalogContentFingerprint(effective.models, effective.combos); | ||
| const changed = state.fingerprint !== undefined && state.fingerprint !== fingerprint; | ||
| state.fingerprint = fingerprint; | ||
| if (changed) { | ||
| if (changed && !stopped) { | ||
| await Promise.resolve(); | ||
| try { | ||
| await ctx.provider.reload(); | ||
|
|
@@ -621,5 +640,53 @@ export default Plugin.define({ | |
| ); | ||
| } | ||
| } | ||
|
|
||
| // Background refresh loop: keep the served catalog aligned with the | ||
| // gateway. The first publish above already populated `latest`; the loop | ||
| // only re-runs refreshAndPublish on the configured TTL. | ||
| // Registry of live loops, keyed by providerId, scoped to globalThis so it | ||
| // survives module re-evaluation (the host loads the plugin up to ~20 | ||
| // times per process). Defense in depth behind the Cleanup returned below. | ||
| const registryKey = Symbol.for("omniroute.opencode-plugin-v2.refresh-loops"); | ||
| const g = globalThis as unknown as Record< | ||
| symbol, | ||
| Map<string, { stop: () => void; state: RefreshState }> | ||
| >; | ||
| if (!g[registryKey]) g[registryKey] = new Map(); | ||
| const registry = g[registryKey]; | ||
| // Stop any loop already registered for this providerId. | ||
| const existing = registry.get(X); | ||
| if (existing) { | ||
| try { | ||
| existing.stop(); | ||
| } catch { | ||
| // ignore | ||
| } | ||
| registry.delete(X); | ||
| } | ||
|
|
||
| const stopLoop = startRefreshLoop({ | ||
| intervalMs: Math.max(MIN_REFRESH_INTERVAL_MS, resolved.modelCacheTtlMs), | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Bound the interval before passing it to The options schema accepts a TTL above
|
||
| refresh: () => refreshAndPublish({ force: true }), | ||
| logger: log, | ||
| }); | ||
|
|
||
| // Cleanup returned to the host: flag the installation as stopped (a tick | ||
| // already in flight then publishes nothing), clear the timer and drop the | ||
| // registry entry. The entry carries `state` only so tests can inspect the | ||
| // memory cache. | ||
| const stop = () => { | ||
| stopped = true; | ||
| try { | ||
| stopLoop(); | ||
| } catch { | ||
| // ignore | ||
| } | ||
| registry.delete(X); | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Delete only this installation’s registry entry. If the host calls an older setup’s cleanup after a replacement is registered,
|
||
| }; | ||
|
|
||
| registry.set(X, { stop, state }); | ||
|
|
||
| return stop; | ||
| }, | ||
| }); | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| export interface RefreshLoopOptions { | ||
| intervalMs: number; | ||
| refresh: () => Promise<void>; | ||
| logger: { warn: (message: string) => void; debug?: (message: string) => void }; | ||
| } | ||
|
|
||
| /** | ||
| * Run `refresh` on a fixed interval without overlapping invocations. | ||
| * | ||
| * Returns a `stop` function that sets the stopped flag and clears the | ||
| * interval. `unref()` is guarded so the timer never keeps the process | ||
| * alive and works on both Node 22 and Node 24. | ||
| */ | ||
| export function startRefreshLoop(options: RefreshLoopOptions): () => void { | ||
| const { intervalMs, refresh, logger } = options; | ||
| let stopped = false; | ||
| let inFlight = false; | ||
|
|
||
| const tick = async (): Promise<void> => { | ||
| if (stopped) return; | ||
| if (inFlight) return; | ||
| inFlight = true; | ||
| try { | ||
| await refresh(); | ||
| } catch (err) { | ||
| logger.warn( | ||
| `[omniroute-v2] background refresh failed: ${err instanceof Error ? err.message : String(err)}` | ||
| ); | ||
| } finally { | ||
| inFlight = false; | ||
| } | ||
| }; | ||
|
|
||
| const timer = setInterval(() => { | ||
| void tick(); | ||
| }, intervalMs); | ||
| if (typeof timer.unref === "function") timer.unref(); | ||
|
|
||
| return () => { | ||
| stopped = true; | ||
| clearInterval(timer); | ||
| }; | ||
| } |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Script executed:
sed -n '130,190p;440,500p;630,700p' @omniroute/opencode-plugin-v2/src/index.tsRepository: cheaper-inference/OmniRoute
Length of output: 7892
Script executed:
Repository: cheaper-inference/OmniRoute
Length of output: 7864
Script executed:
Repository: cheaper-inference/OmniRoute
Length of output: 5990
Stop the previous installation before the replacement publishes.
existing.stop()correctly sets the previous setup'sstoppedflag. However, the replacement callsawait refreshAndPublish()before it invokesexisting.stop(). An old refresh that has already passed its stop guard can therefore reachctx.provider.reload()while the replacement is publishing. Move the existing-loop replacement, or otherwise invoke the old cleanup, before the replacement's initial refresh.