mirror of
https://github.com/vercel-labs/fx.git
synced 2026-10-28 10:15:17 +08:00
Wait briefly for a failed model catalog's retry
When the model catalog failed to load, the retry a submitted prompt starts ran behind the failed state, so the prompt's first request went out without the model's limits. With no output limit, some providers stop a reply at 4,096 tokens. A request now waits up to two seconds for that retry, as it waits for a first load. A retry that lands in time gives the request the catalog's limits. A slower one leaves the request as it was before, so a retry that hangs never holds a prompt for longer than that.
This commit is contained in:
2 files changed
+85
-6
No files matched your search
@@ -42,6 +42,10 @@ const picker_revalidate_floor_ms: i64 = 5_000;
|
||||
const failure_backoff_base_ms: i64 = 5_000;
|
||||
const failure_backoff_cap_ms: i64 = 5 * std.time.ms_per_min;
|
||||
|
||||
/// How long a request waits for a failed catalog's retry before it goes out
|
||||
/// without the catalog's limits.
|
||||
const failed_catalog_retry_wait_ms: i64 = 2_000;
|
||||
|
||||
fn failureBackoffMs(failure_streak: u32) i64 {
|
||||
var delay = failure_backoff_base_ms;
|
||||
var step: u32 = 1;
|
||||
@@ -689,6 +693,7 @@ pub const Runtime = struct {
|
||||
model: []const u8,
|
||||
cancel_flag: *std.atomic.Value(bool),
|
||||
) model_capabilities.ResolveError!model_capabilities.Capabilities {
|
||||
var retry_wait_until_ms: ?i64 = null;
|
||||
while (true) {
|
||||
self.finishThreadIfDone();
|
||||
self.mutex.lockUncancelable(io_mod.getIo());
|
||||
@@ -709,9 +714,19 @@ pub const Runtime = struct {
|
||||
}
|
||||
return model_capabilities.resolveCapabilities(model, metadata);
|
||||
}
|
||||
// A request waits a moment for a failed catalog's retry, such as
|
||||
// the one a submitted prompt starts, so a retry that lands quickly
|
||||
// gives this request its limits. A slow retry never holds it.
|
||||
const retry_in_flight = state == .failed and self.fetch_in_flight;
|
||||
self.mutex.unlock(io_mod.getIo());
|
||||
const wait_for_retry = retry_in_flight and wait: {
|
||||
const now = model_catalog.clockMs();
|
||||
const until = retry_wait_until_ms orelse now + failed_catalog_retry_wait_ms;
|
||||
retry_wait_until_ms = until;
|
||||
break :wait now < until;
|
||||
};
|
||||
|
||||
switch (state) {
|
||||
switch (if (wait_for_retry) .loading else state) {
|
||||
.loading => {
|
||||
if (cancel_flag.load(.seq_cst)) {
|
||||
debug_trace.logf(
|
||||
@@ -1347,7 +1362,7 @@ test "model picker revalidates past the reopen floor while activity waits for th
|
||||
}
|
||||
}
|
||||
|
||||
test "model cache retries a failed catalog behind the failed state so requests never wait on it" {
|
||||
test "model cache retries a failed catalog behind the failed state so requests wait on it only briefly" {
|
||||
const public: credentials.CatalogAccess = .{ .public_only = .no_credential };
|
||||
var runtime = Runtime.init(std.testing.allocator, "/v1/models");
|
||||
defer runtime.deinit();
|
||||
@@ -1365,10 +1380,17 @@ test "model cache retries a failed catalog behind the failed state so requests n
|
||||
try std.testing.expect(runtime.isFailed());
|
||||
try std.testing.expect(!runtime.isLoading());
|
||||
|
||||
// A set cancel flag fails any wait at once, so a returned value proves
|
||||
// the request resolved without waiting on the retry.
|
||||
// A set cancel flag fails any wait at once, so the error proves the
|
||||
// request waits on the retry.
|
||||
var cancelled = std.atomic.Value(bool).init(true);
|
||||
_ = try runtime.resolveForRequest("first", &cancelled);
|
||||
try std.testing.expectError(error.Cancelled, runtime.resolveForRequest("first", &cancelled));
|
||||
// A retry that has not landed holds a request only for the bound.
|
||||
var live = std.atomic.Value(bool).init(false);
|
||||
const started_ms = model_catalog.clockMs();
|
||||
const unresolved = try runtime.resolveForRequest("first", &live);
|
||||
try std.testing.expect(model_catalog.clockMs() - started_ms >= failed_catalog_retry_wait_ms);
|
||||
try std.testing.expectEqual(@as(?u32, null), unresolved.max_output_tokens);
|
||||
try std.testing.expect(fetchInFlight(&runtime));
|
||||
try runtime.openMenu();
|
||||
try std.testing.expectEqual(ModelMenuLoadState.loading, runtime.menu.load_state);
|
||||
|
||||
|
||||
@@ -43,6 +43,7 @@ import {
|
||||
startFakeGateway,
|
||||
TmuxSession,
|
||||
type FakeGatewayModel,
|
||||
type FakeGatewayOptions,
|
||||
type FakeGatewayResponse,
|
||||
tmuxAvailable,
|
||||
} from "./tmux-helpers";
|
||||
@@ -1504,7 +1505,7 @@ async function launchRouteRecoveryTui(
|
||||
responses: FakeGatewayResponse[],
|
||||
options: {
|
||||
model?: string;
|
||||
models?: FakeGatewayModel[];
|
||||
models?: FakeGatewayOptions["models"];
|
||||
settings?: Record<string, unknown>;
|
||||
} = {},
|
||||
) {
|
||||
@@ -1941,6 +1942,62 @@ describe.skipIf(!tmuxAvailable())("TUI gateway stream lifecycle", () => {
|
||||
TIMEOUT * 2,
|
||||
);
|
||||
|
||||
test(
|
||||
"a prompt's request waits for the retry of a model catalog that failed to load",
|
||||
async () => {
|
||||
const model = "zai/glm-5.3";
|
||||
const finalText = "Catalog retry reached the first request.";
|
||||
let catalogUp = false;
|
||||
const { queuedGateway, stderrPath } = await launchRouteRecoveryTui(
|
||||
"fx-tui-catalog-retry-",
|
||||
[fakeGatewayFinalText(finalText)],
|
||||
{
|
||||
model,
|
||||
models: async () => {
|
||||
if (!catalogUp) return new Response("catalog unavailable", { status: 503 });
|
||||
// Slower than building the request, which goes out first unless it
|
||||
// waits, and well inside the two seconds a request waits.
|
||||
await Bun.sleep(750);
|
||||
return [{
|
||||
id: model,
|
||||
type: "language",
|
||||
tags: ["reasoning", "tool-use"],
|
||||
context_window: 1_000_000,
|
||||
max_tokens: 1_000_000,
|
||||
}];
|
||||
},
|
||||
settings: { model },
|
||||
},
|
||||
);
|
||||
await waitForCondition(
|
||||
() => queuedGateway.modelRequests.length >= 1,
|
||||
"the failed startup catalog load",
|
||||
);
|
||||
// The first retry after a failure waits five seconds; this only paces
|
||||
// the prompt past it, and the request count below proves the retry ran.
|
||||
await Bun.sleep(5_500);
|
||||
const failedLoads = queuedGateway.modelRequests.length;
|
||||
catalogUp = true;
|
||||
|
||||
await session!.sendText("hi");
|
||||
await session!.waitForText(finalText, TIMEOUT);
|
||||
await session!.waitForComposer(TIMEOUT);
|
||||
|
||||
expect(queuedGateway.modelRequests.length).toBe(failedLoads + 1);
|
||||
expect(queuedGateway.requests).toHaveLength(1);
|
||||
// The full-window catalog limit is bounded, as the retried catalog says.
|
||||
expect(JSON.parse(queuedGateway.requests[0]!.body).maxOutputTokens).toBe(32_768);
|
||||
expect(session!.isAlive()).toBe(true);
|
||||
expect(readFileSync(stderrPath, "utf8")).toBe("");
|
||||
|
||||
await session!.sendText("/quit");
|
||||
expect(await session!.waitForSessionEnd(TIMEOUT)).toBe(true);
|
||||
await session!.kill();
|
||||
session = null;
|
||||
},
|
||||
TIMEOUT * 2,
|
||||
);
|
||||
|
||||
test(
|
||||
"live token counter includes submitted input, reasoning, and streamed text",
|
||||
async () => {
|
||||
|
||||
Reference in new issue
Block a user