From 9932998e5e38cf9f70779f11cccd3972d4487479 Mon Sep 17 00:00:00 2001 From: ashhart <71890694+ashhart@users.noreply.github.com> Date: Wed, 4 Nov 2026 23:25:43 +0000 Subject: [PATCH] lanes: share HTTP wait codes with load diagnostics (#2135) Publish fixed wait and refusal reasons in the load ledger, JSON snapshot, metrics and DeepSeek admission logs. Reuse engine_busy, engine_unavailable and memory_pressure from the HTTP errors; closed and stale submissions retain their original errors and share the unavailable counter. Additional codes describe prefix, state-read, cache-budget and backend conditions. Keep HTTP responses and scheduling decisions unchanged. Move the existing submission body to its own module to keep the host within the source limit. Observe only priced request ledgers through their price callback; a page-backed pool has no per-request price. Check shared HTTP and diagnostic names, repeated memory refusals, queue reason counts, preserved closed/stale errors and page-backed observation. With a load producer attached, each host unlock scans queued requests and recomputes wait reasons under the host/load locks before deduplication, including admission callbacks where needed; elapsed overhead is unmeasured. --- zig/build/server.zig | 3 + zig/src/core/engine_api.zig | 1 + zig/src/core/lane_halt_test.zig | 2 +- zig/src/core/lane_host.zig | 37 ++--- zig/src/core/lane_host_end.zig | 3 +- zig/src/core/lane_host_fetch.zig | 2 +- zig/src/core/lane_host_load.zig | 21 ++- zig/src/core/lane_host_load_test.zig | 149 +++++++++++++++++++ zig/src/core/lane_host_submit.zig | 36 +++++ zig/src/core/lane_host_tests.zig | 1 + zig/src/core/lane_job.zig | 1 + zig/src/core/lane_review_test.zig | 4 +- zig/src/core/lane_rounds.zig | 14 +- zig/src/core/lane_rounds_test.zig | 6 +- zig/src/core/load_ledger.zig | 7 + zig/src/core/wait_reason.zig | 128 ++++++++++++++++ zig/src/families/deepseek_v41/sched_host.zig | 23 ++- zig/src/server/load_metrics.zig | 18 +++ zig/src/server/load_metrics_test.zig | 28 +++- zig/src/server/wait_reason.zig | 7 +- zig/src/server/wait_reason_test.zig | 10 ++ 21 files changed, 453 insertions(+), 48 deletions(-) create mode 100644 zig/src/core/lane_host_submit.zig create mode 100644 zig/src/core/wait_reason.zig diff --git a/zig/build/server.zig b/zig/build/server.zig index 5ad8f3144f..5a0cedb588 100644 --- a/zig/build/server.zig +++ b/zig/build/server.zig @@ -79,6 +79,9 @@ pub fn toolHostTests(b: *std.Build, parser: *std.Build.Module, server: *std.Buil const profile_line = b.addRunArtifact(b.addTest(.{ .root_module = server, .filters = &.{"profile line:"} })); profile_line.has_side_effects = true; b.step("test-profile-line-host", "Profile start-up line scoping against the tree's real backend names").dependOn(&profile_line.step); + const metrics = b.addRunArtifact(b.addTest(.{ .root_module = server, .filters = &.{ "scrape", "load object", "wait reasons:" } })); + metrics.has_side_effects = true; + b.step("test-load-metrics-host", "Load ledger scrape and /v1/load carry bounded wait and refusal reasons").dependOn(&metrics.step); const admitted = b.addRunArtifact(b.addTest(.{ .root_module = server, .filters = &.{"admission"} })); admitted.has_side_effects = true; b.step("test-admission-host", "Opt-in in-flight caps and per-key rates: limits, 429s over sockets, metrics").dependOn(&admitted.step); diff --git a/zig/src/core/engine_api.zig b/zig/src/core/engine_api.zig index 97901faa80..0ed26da477 100644 --- a/zig/src/core/engine_api.zig +++ b/zig/src/core/engine_api.zig @@ -16,6 +16,7 @@ pub const EmbeddingBatch = lanes.embedding.Batch; pub const embedding_vector = @import("embedding_vector.zig"); pub const load_ledger = @import("load_ledger.zig"); pub const kv_ledger = @import("kv_ledger.zig"); +pub const wait_reason = @import("wait_reason.zig"); /// One image of a request (``Request.images``): the tower's input and the placeholder rows its output replaces. pub const Image = lanes.stream.Image; diff --git a/zig/src/core/lane_halt_test.zig b/zig/src/core/lane_halt_test.zig index e83462fca7..3a81c003eb 100644 --- a/zig/src/core/lane_halt_test.zig +++ b/zig/src/core/lane_halt_test.zig @@ -46,7 +46,7 @@ fn haltCase(release_at_stop: ?*bool) !void { fn admit(ctx: *anyopaque, ask: api.Rounds.Ask, _: *[]const u8) api.Rounds.Verdict { const p = self(ctx); for (p.seqs.items) |x| if (!x.decoding) return .wait; - p.seqs.append(gpa, .{ .key = ask.key, .n = ask.prompt.len }) catch return .refuse; + p.seqs.append(gpa, .{ .key = ask.key, .n = ask.prompt.len }) catch return .{ .refuse = .backend }; return .admit; } fn admitted(_: *anyopaque) anyerror!void {} diff --git a/zig/src/core/lane_host.zig b/zig/src/core/lane_host.zig index 7d8b32840f..8975ad3b07 100644 --- a/zig/src/core/lane_host.zig +++ b/zig/src/core/lane_host.zig @@ -30,6 +30,7 @@ const rounds = @import("lane_rounds.zig"); const scoring = @import("lane_host_score.zig"); const lane_stream = @import("lane_stream.zig"); const lane_halt = @import("lane_halt.zig"); +const reasons = @import("wait_reason.zig"); pub const LaneHost = struct { gpa: Allocator, @@ -91,6 +92,7 @@ pub const LaneHost = struct { load: load.State = .{}, // this host's part of the server's load ledger (lane_host_load.zig) capacity: ?load.Source = null, // how the engine holds KV, set before start; null: no capacity group kv: ?@import("kv_ledger.zig").Ledger = null, // Info.ledger's budget or the paged pool's (lane_host_ledger.zig) + refusals: reasons.Refusals = .{}, // refused requests by fixed reason, under mutex pub const ThreadInit = struct { ctx: *anyopaque, run: *const fn (ctx: *anyopaque) anyerror!void }; @@ -191,26 +193,7 @@ pub const LaneHost = struct { } fn submitFn(ctx: *anyopaque, id: Id, request: *const Request, sink: Sink) api.SubmitError!void { - const h = self(ctx); - if (@import("lane_prompt_only.zig").refuses(h, request)) return error.InvalidRequest; - const job = h.gpa.create(Job) catch return error.Busy; - job.* = .{ .host = h, .id = id, .request = request, .sink = sink, .submitted = std.Io.Clock.awake.now(h.io).toNanoseconds() }; - h.mutex.lockUncancelable(h.io); - defer h.unlock(); - const unavailable = h.closing or (if (h.health) |source| !source.read(source.ctx) else false); - if (unavailable or h.stale or pressure.refuses(h)) { - h.gpa.destroy(job); - return if (unavailable) error.Closed else if (h.stale) error.Stale else error.MemoryPressure; - } - // foreground before background, each in arrival order (the Python job queue's priority) - var at = h.queued.items.len; - while (!request.background and at > 0 and h.queued.items[at - 1].request.background) at -= 1; - h.queued.insert(h.gpa, at, job) catch { - h.gpa.destroy(job); - return error.Busy; - }; - h.load.started += 1; - h.wake.signal(h.io); + return @import("lane_host_submit.zig").submit(self(ctx), id, request, sink); } fn cancelFn(ctx: *anyopaque, id: Id) void { @@ -345,6 +328,7 @@ pub const LaneHost = struct { return .none; }; if (ledger.never(h, h.queued.items[at])) { // past the whole budget alone: no release will ever make it fit + h.refusals.note(.prompt_too_large); const job = h.queued.orderedRemove(at); h.unlock(); h.finish(job, .failed, ledger.too_big); @@ -359,6 +343,7 @@ pub const LaneHost = struct { job.seq = h.admissions; h.admissions += 1; h.admitted.append(h.gpa, job) catch { + h.refusals.note(.engine_busy); h.unlock(); h.finish(job, .failed, "out of memory"); return .failed; @@ -407,7 +392,10 @@ pub const LaneHost = struct { return switch (e) { error.Cancelled => h.cancel(job), error.LaneMemory => ends.requeue(h, job), - else => ends.dropFailed(h, job, e), + else => blk: { + h.refused(.backend_prompt); + break :blk ends.dropFailed(h, job, e); + }, }; } @@ -492,6 +480,13 @@ pub const LaneHost = struct { return x.text(x.ctx, e) orelse @errorName(e); } + /// One refused request, by its fixed reason; the caller does not hold the lock. + pub fn refused(h: *LaneHost, why: reasons.Refusal) void { + h.lock(); + defer h.unlock(); + h.refusals.note(why); + } + pub fn drop(h: *LaneHost, job: *Job, message: []const u8) bool { h.remove(job); if (job.started and !job.stream.finished) h.core.discard(&job.stream); diff --git a/zig/src/core/lane_host_end.zig b/zig/src/core/lane_host_end.zig index 7b03038f85..c68e11e03f 100644 --- a/zig/src/core/lane_host_end.zig +++ b/zig/src/core/lane_host_end.zig @@ -65,6 +65,7 @@ pub fn deliverPending(h: *LaneHost) void { /// A round the backend failed: the failure hook hears the error and every admitted stream gets its name. pub fn failRound(h: *LaneHost, err: anyerror) void { failed(h, err); + h.refused(.backend_round); failAll(h, @errorName(err)); } @@ -123,7 +124,7 @@ pub fn requeue(h: *LaneHost, job: *Job) bool { if (waits) { h.load.requeued += 1; h.memory_lanes = others; // under the lock: a publish reads it - } + } else h.refusals.note(.lane_memory_alone); h.unlock(); if (!waits) { h.finish(job, .failed, if (others > 0) "out of memory" else alone_words); diff --git a/zig/src/core/lane_host_fetch.zig b/zig/src/core/lane_host_fetch.zig index a1f7007bfd..414ea23fa8 100644 --- a/zig/src/core/lane_host_fetch.zig +++ b/zig/src/core/lane_host_fetch.zig @@ -43,7 +43,7 @@ pub fn queue(h: *LaneHost) void { } /// Whether queued `job` waits for its read: still going and inside the bound (the lock held). -pub fn held(h: *LaneHost, job: *const Job) bool { +pub fn held(h: *const LaneHost, job: *const Job) bool { const r = (h.cache orelse return false).disk orelse return false; const f = r.fetch orelse return false; if (job.fetch.writing) |name| { diff --git a/zig/src/core/lane_host_load.zig b/zig/src/core/lane_host_load.zig index 4eae5b4ac3..3f9deaa180 100644 --- a/zig/src/core/lane_host_load.zig +++ b/zig/src/core/lane_host_load.zig @@ -195,7 +195,11 @@ pub fn yielded(h: *LaneHost) void { /// The `requests` group from the host's lists; the caller holds the host mutex. fn requests(h: *const LaneHost) ledger.Requests { var waiting_tokens: u64 = 0; - for (h.queued.items) |job| waiting_tokens += job.request.prompt.len; + var waits: @import("wait_reason.zig").Waits = .{}; + for (h.queued.items) |job| { + waiting_tokens += job.request.prompt.len; + waits.note(waitReason(h, job)); + } const left = filling(h); return .{ .running = @intCast(h.admitted.items.len), @@ -206,9 +210,24 @@ fn requests(h: *const LaneHost) ledger.Requests { .started = h.load.started, .requeued = h.load.requeued, .yields = h.load.yields, + .waiting_reasons = waits.counts, + .refusals = h.refusals, }; } +/// Why a queued request is not running: one fixed code, in the order admission decides (lock held). +fn waitReason(h: *const LaneHost, job: *const Job) @import("wait_reason.zig").Wait { + if (h.stale) return .engine_unavailable; + if (job.prefix.leader != null) return .prefix_cut; + if (@import("lane_host_fetch.zig").held(h, job)) return .state_read; + if (@import("lane_host_pressure.zig").holds(h)) return .memory_pressure; + const lanes_now = @min(h.info_.lanes, h.memory_lanes orelse h.info_.lanes); + if (h.admitted.items.len >= lanes_now) return .engine_busy; + if (@import("lane_host_ledger.zig").priced(h)) return if (@import("lane_host_ledger.zig").fits(h, job)) .engine_busy else .ledger_bytes; + if (h.info_.admit) |a| return if (a.fits(a.ctx, @intCast(job.request.prompt.len), job.request.max_tokens)) .engine_busy else .pool_room; + return .engine_busy; +} + /// Prompt tokens left in the foreground passes filling now: 0 for none, null when the backend reports no progress. fn filling(h: *const LaneHost) ?u64 { var left: u64 = 0; diff --git a/zig/src/core/lane_host_load_test.zig b/zig/src/core/lane_host_load_test.zig index 82ad661a90..3bbdfaa58f 100644 --- a/zig/src/core/lane_host_load_test.zig +++ b/zig/src/core/lane_host_load_test.zig @@ -9,16 +9,35 @@ const Id = api.Id; const Event = api.Event; const Request = api.Request; const Engine = api.Engine; +const reasons = api.wait_reason; + +/// A pool where prompts of 64 tokens or more fit only when nothing else runs. +const Pool = struct { + fn fits(_: *anyopaque, prompt: u32, _: u32) bool { + return prompt < 64; + } +}; + +fn cost(_: *anyopaque, prompt: u32, max_tokens: u32) u64 { + return @as(u64, prompt + max_tokens) * 4; +} /// How a request finished, as the host reports it. const Box = struct { mutex: std.Io.Mutex = .init, done: ?Reason = null, + tokens: usize = 0, fn event(ctx: *anyopaque, _: Id, e: *const Event) void { const b: *@This() = @ptrCast(@alignCast(ctx)); b.mutex.lockUncancelable(std.testing.io); defer b.mutex.unlock(std.testing.io); if (e.* == .finished) b.done = e.finished.reason; + if (e.* == .tokens) b.tokens += e.tokens.len; + } + fn started(b: *@This()) bool { + b.mutex.lockUncancelable(std.testing.io); + defer b.mutex.unlock(std.testing.io); + return b.tokens > 0; } fn wait(b: *@This()) Reason { while (true) { @@ -410,3 +429,133 @@ test "a request holding a lane lowers the free lanes" { try std.testing.expectEqual(@as(?u32, 1), watch.free); try std.testing.expectEqual(@as(u32, 2), ledger.read().capacity.?.lanes_free); } + +test "waits by reason: a pool-room head and a lane-full queue each carry one fixed code" { + var rig: Rig = undefined; + try rig.init(.{ .lanes = 3, .admit = .{ .ctx = @ptrCast(&rig), .fits = Pool.fits } }); + defer rig.deinit(); + var ledger: Ledger = .init(std.testing.io); + ledger.attach(rig.host.engine()); + const short = [_]u32{ 3, 1, 4 }; + const long: [64]u32 = @splat(7); + var a: Box = .{}; + var c: Box = .{}; + var d: Box = .{}; + var e: Box = .{}; + var f: Box = .{}; + const rs: Request = .{ .prompt = &short, .max_tokens = 512 }; + const rl: Request = .{ .prompt = &long, .max_tokens = 4 }; + try rig.host.engine().submit(1, &rs, a.sink()); + try rig.host.engine().submit(2, &rl, c.sink()); + try rig.host.engine().submit(3, &rs, d.sink()); + try rig.host.start(); + // one short prompt runs, a second goes beside it, and the long head waits with a lane still free + while (!d.started()) std.Io.sleep(std.testing.io, .fromMilliseconds(1), .awake) catch {}; + var q = ledger.read().requests.?; + try std.testing.expectEqual(@as(u32, 1), q.waiting); + try std.testing.expectEqual(@as(u32, 1), q.waiting_reasons[@backingInt(reasons.Wait.pool_room)]); + try std.testing.expectEqual(@as(u32, 0), q.waiting_reasons[@backingInt(reasons.Wait.engine_busy)]); + // two more short prompts: one takes the last lane, the rest wait for a lane, the long head included + try rig.host.engine().submit(4, &rs, e.sink()); + try rig.host.engine().submit(5, &rs, f.sink()); + for (0..2000) |_| { + var status: api.Status = .{}; + rig.host.engine().status(&status, &.{}); + if (status.running == 3) break; + std.Io.sleep(std.testing.io, .fromMilliseconds(1), .awake) catch {}; + } + q = ledger.read().requests.?; + try std.testing.expectEqual(@as(u32, 2), q.waiting); + try std.testing.expectEqual(@as(u32, 0), q.waiting_reasons[@backingInt(reasons.Wait.pool_room)]); + try std.testing.expectEqual(q.waiting, q.waiting_reasons[@backingInt(reasons.Wait.engine_busy)]); + try std.testing.expectEqual(q.waiting, (reasons.Waits{ .counts = q.waiting_reasons }).total()); + _ = a.wait(); + _ = c.wait(); + _ = d.wait(); + _ = e.wait(); + _ = f.wait(); + ledger.detach(false); +} + +test "repeated prompt failures release every lane and count one fixed refusal each" { + var rig: Rig = undefined; + try rig.init(.{ .lanes = 2, .ledger = .{ .ctx = @ptrCast(&rig), .total = 1 << 20, .cost = cost } }); + defer rig.deinit(); + rig.target.memory_lanes = 0; // every prompt pass finds no lane memory, even alone + var ledger: Ledger = .init(std.testing.io); + ledger.attach(rig.host.engine()); + const prompt = [_]u32{ 2, 7, 1, 8 }; + var boxes: [3]Box = .{ .{}, .{}, .{} }; + for (&boxes, 0..) |*b, i| { + const request: Request = .{ .prompt = &prompt, .max_tokens = 8 }; + try rig.host.engine().submit(@intCast(i + 1), &request, b.sink()); + } + try rig.host.start(); + for (&boxes) |*b| try std.testing.expectEqual(Reason.failed, b.wait()); + rig.host.lock(); // the last finish's publish has run by the time its event lands + rig.host.unlock(); + const q = ledger.read().requests.?; + try std.testing.expectEqual(@as(u32, 0), q.waiting); + try std.testing.expectEqual(@as(u32, 0), q.running); + try std.testing.expectEqual(@as(u64, 3), q.refusals.count(.lane_memory_alone)); + try std.testing.expect(rig.host.kv.?.idle()); // every promised lease came back + // memory returns: one request runs to its end and leaves the lane free + rig.target.memory_lanes = null; + var ok: Box = .{}; + const request: Request = .{ .prompt = &prompt, .max_tokens = 8 }; + try rig.host.engine().submit(9, &request, ok.sink()); + _ = ok.wait(); + rig.host.lock(); + rig.host.unlock(); + var status: api.Status = .{}; + rig.host.engine().status(&status, &.{}); + try std.testing.expectEqual(@as(u32, 0), status.running); + try std.testing.expectEqual(@as(u32, 0), status.waiting); + try std.testing.expect(rig.host.kv.?.idle()); + ledger.detach(false); +} + +test "page-backed waits: publishing a queued request never asks for an absent per-request price" { + var rig: Rig = undefined; + try rig.init(.{ .lanes = 1 }); + defer rig.deinit(); + const kv = @import("kv_ledger.zig"); + const Pages = struct { + fn read(_: *const anyopaque) kv.Pages { + return .{ .total = 4, .available = 4, .reclaimable = 0, .unit = 64 }; + } + fn detach(_: *anyopaque) void {} + }; + rig.host.kv = .ofPages(.{ .ctx = &rig, .read = Pages.read, .detach = Pages.detach }); + var ledger: Ledger = .init(std.testing.io); + ledger.attach(rig.host.engine()); + const prompt = [_]u32{ 2, 7, 1 }; + const request: Request = .{ .prompt = &prompt, .max_tokens = 4 }; + var box: Box = .{}; + try rig.host.engine().submit(1, &request, box.sink()); + try rig.host.start(); + _ = box.wait(); + ledger.detach(false); +} + +test "capacity refusal codes: closed and stale keep their errors and share engine_unavailable" { + var rig: Rig = undefined; + try rig.init(.{ .lanes = 1 }); + defer rig.deinit(); + var ledger: Ledger = .init(std.testing.io); + ledger.attach(rig.host.engine()); + const prompt = [_]u32{ 2, 7, 1 }; + const request: Request = .{ .prompt = &prompt, .max_tokens = 4 }; + var box: Box = .{}; + rig.host.stale = true; + try std.testing.expectError(error.Stale, rig.host.engine().submit(1, &request, box.sink())); + rig.host.stale = false; + rig.host.closing = true; + try std.testing.expectError(error.Closed, rig.host.engine().submit(2, &request, box.sink())); + const q = ledger.read().requests.?; + try std.testing.expectEqual(@as(u64, 2), q.refusals.count(.engine_unavailable)); + try std.testing.expectEqual(@as(u64, 0), q.refusals.count(.engine_busy)); + try std.testing.expectEqual(@as(u32, 0), q.waiting); + rig.host.closing = false; + ledger.detach(false); +} diff --git a/zig/src/core/lane_host_submit.zig b/zig/src/core/lane_host_submit.zig new file mode 100644 index 0000000000..d602151e2a --- /dev/null +++ b/zig/src/core/lane_host_submit.zig @@ -0,0 +1,36 @@ +//! Request submission to the lane host: the same queue order and refusal conditions, with bounded reason counts. +const std = @import("std"); +const api = @import("engine_api.zig"); +const LaneHost = @import("lane_host.zig").LaneHost; +const Job = LaneHost.Job; +const pressure = @import("lane_host_pressure.zig"); + +pub fn submit(h: *LaneHost, id: api.Id, request: *const api.Request, sink: api.Sink) api.SubmitError!void { + if (@import("lane_prompt_only.zig").refuses(h, request)) { + h.refused(.invalid_request); + return error.InvalidRequest; + } + const job = h.gpa.create(Job) catch { + h.refused(.engine_busy); + return error.Busy; + }; + job.* = .{ .host = h, .id = id, .request = request, .sink = sink, .submitted = std.Io.Clock.awake.now(h.io).toNanoseconds() }; + h.mutex.lockUncancelable(h.io); + defer h.unlock(); + const unavailable = h.closing or (if (h.health) |source| !source.read(source.ctx) else false); + if (unavailable or h.stale or pressure.refuses(h)) { + h.refusals.note(if (unavailable or h.stale) .engine_unavailable else .memory_pressure); + h.gpa.destroy(job); + return if (unavailable) error.Closed else if (h.stale) error.Stale else error.MemoryPressure; + } + // foreground before background, each in arrival order (the Python job queue's priority) + var at = h.queued.items.len; + while (!request.background and at > 0 and h.queued.items[at - 1].request.background) at -= 1; + h.queued.insert(h.gpa, at, job) catch { + h.refusals.note(.engine_busy); + h.gpa.destroy(job); + return error.Busy; + }; + h.load.started += 1; + h.wake.signal(h.io); +} diff --git a/zig/src/core/lane_host_tests.zig b/zig/src/core/lane_host_tests.zig index ab38b0e3d8..7511124049 100644 --- a/zig/src/core/lane_host_tests.zig +++ b/zig/src/core/lane_host_tests.zig @@ -10,6 +10,7 @@ test { _ = @import("lane_init_test.zig"); _ = @import("lane_failure_review_test.zig"); _ = @import("lane_host_first_test.zig"); + _ = @import("wait_reason.zig"); } test { diff --git a/zig/src/core/lane_job.zig b/zig/src/core/lane_job.zig index dba112e4d4..f51612df9b 100644 --- a/zig/src/core/lane_job.zig +++ b/zig/src/core/lane_job.zig @@ -68,6 +68,7 @@ pub const Job = struct { const store = job.host.cache orelse return; if (!job.started) return store.release(e); job.stream.reuse.saved = null; // the backend restores before its first chunk; a later keep may free it + if (job.stream.reuse_failed) job.host.refused(.restore_refused); const ok = !job.stream.reuse_failed and job.stream.cached == e.at; if (job.stream.reuse_failed or ok) store.resumedWithReason(e, job.cacheKey(), if (ok) null else job.stream.reuse_error orelse error.RestoreRejected) else store.release(e); } diff --git a/zig/src/core/lane_review_test.zig b/zig/src/core/lane_review_test.zig index a660b00eb3..165b337174 100644 --- a/zig/src/core/lane_review_test.zig +++ b/zig/src/core/lane_review_test.zig @@ -34,7 +34,7 @@ test "more than sixteen planned pieces retain their job until its final" { } fn beginAdmit(_: *anyopaque) void {} fn admit(ctx: *anyopaque, ask: api.Rounds.Ask, _: *[]const u8) api.Rounds.Verdict { - self(ctx).seqs.append(gpa, .{ .key = ask.key, .n = ask.prompt.len }) catch return .refuse; + self(ctx).seqs.append(gpa, .{ .key = ask.key, .n = ask.prompt.len }) catch return .{ .refuse = .backend }; return .admit; } fn admitted(_: *anyopaque) anyerror!void {} @@ -166,7 +166,7 @@ test "failed repeated pieces drop their job once and skip its remaining batches" } fn beginAdmit(_: *anyopaque) void {} fn admit(ctx: *anyopaque, ask: api.Rounds.Ask, _: *[]const u8) api.Rounds.Verdict { - self(ctx).seqs.append(gpa, .{ .key = ask.key, .n = ask.prompt.len }) catch return .refuse; + self(ctx).seqs.append(gpa, .{ .key = ask.key, .n = ask.prompt.len }) catch return .{ .refuse = .backend }; return .admit; } fn admitted(_: *anyopaque) anyerror!void {} diff --git a/zig/src/core/lane_rounds.zig b/zig/src/core/lane_rounds.zig index cf9d613008..ae63854939 100644 --- a/zig/src/core/lane_rounds.zig +++ b/zig/src/core/lane_rounds.zig @@ -24,7 +24,8 @@ pub const Rounds = struct { } } else struct {}; - pub const Verdict = enum { admit, wait, skip, refuse }; + pub const Refusal = enum { too_large, memory, backend }; + pub const Verdict = union(enum) { admit, wait, skip, refuse: Refusal }; pub const Piece = struct { key: usize, start: u64, end: u64, save: bool, run: bool = true }; pub const Ask = struct { key: usize, prompt: []const u32, max_tokens: u32, submitted_s: f64 }; @@ -85,7 +86,12 @@ fn round(h: *LaneHost, rd: Rounds) bool { }, .wait => break, .skip => i += 1, - .refuse => { + .refuse => |r| { + h.refusals.note(switch (r) { + .too_large => .prompt_too_large, + .memory => .lane_memory_alone, + .backend => .backend_prompt, + }); if (refused.append(h.gpa, .{ .job = job, .why = why })) |_| {} else |_| break; _ = h.queued.orderedRemove(i); }, @@ -184,6 +190,7 @@ fn round(h: *LaneHost, rd: Rounds) bool { /// A job whose piece failed: dropped and taken out of this round's finals (`drop` frees the job). fn dropPiece(h: *LaneHost, job: *Job, message: []const u8) void { + h.refused(.backend_prompt); const key = @intFromPtr(job); for (h.pieces.items) |*piece| if (piece.key == key) { piece.run = false; @@ -217,6 +224,7 @@ fn runFinals(h: *LaneHost) void { } h.core.prepareFinals(ss[0..n]) catch |e| { ends.failed(h, e); + h.refused(.backend_prompt); ends.failAll(h, h.words(e)); return; }; @@ -246,6 +254,7 @@ fn runFinals(h: *LaneHost) void { if (joined > 0) std.log.info("join drafts: {d} of {d} finished prompts drafted in one pass", .{ joined, n }); for (jobs[0..n], errs[0..n]) |job, x| { if (x) |e| { + if (e != error.Cancelled) h.refused(.backend_prompt); _ = if (e == error.Cancelled) h.cancel(job) else ends.dropFailed(h, job, e); continue; } @@ -258,6 +267,7 @@ fn runFinals(h: *LaneHost) void { const job: *Job = @ptrFromInt(key); if (job.stream.finished) continue; h.core.finishStream(&job.stream) catch |e| { + if (e != error.Cancelled) h.refused(.backend_prompt); _ = if (e == error.Cancelled) h.cancel(job) else ends.dropFailed(h, job, e); continue; }; diff --git a/zig/src/core/lane_rounds_test.zig b/zig/src/core/lane_rounds_test.zig index 4a59e38040..5711bc4ff7 100644 --- a/zig/src/core/lane_rounds_test.zig +++ b/zig/src/core/lane_rounds_test.zig @@ -39,7 +39,7 @@ test "a round planner: prompts in pieces between the other streams' decode round p.waits += 1; return .wait; }; - p.seqs.append(gpa, .{ .key = ask.key, .n = ask.prompt.len }) catch return .refuse; + p.seqs.append(gpa, .{ .key = ask.key, .n = ask.prompt.len }) catch return .{ .refuse = .backend }; return .admit; } fn admitted(_: *anyopaque) anyerror!void {} @@ -162,7 +162,7 @@ test "a failed piece run fails its requests alone: their finals of the same roun } fn beginAdmit(_: *anyopaque) void {} fn admit(ctx: *anyopaque, ask: api.Rounds.Ask, _: *[]const u8) api.Rounds.Verdict { - self(ctx).seqs.append(gpa, .{ .key = ask.key, .n = ask.prompt.len }) catch return .refuse; + self(ctx).seqs.append(gpa, .{ .key = ask.key, .n = ask.prompt.len }) catch return .{ .refuse = .backend }; return .admit; } fn admitted(_: *anyopaque) anyerror!void {} @@ -317,7 +317,7 @@ test "a failed piece fails its request alone: its final of the same round does n } fn beginAdmit(_: *anyopaque) void {} fn admit(ctx: *anyopaque, ask: api.Rounds.Ask, _: *[]const u8) api.Rounds.Verdict { - self(ctx).seqs.append(gpa, .{ .key = ask.key, .n = ask.prompt.len }) catch return .refuse; + self(ctx).seqs.append(gpa, .{ .key = ask.key, .n = ask.prompt.len }) catch return .{ .refuse = .backend }; return .admit; } fn admitted(_: *anyopaque) anyerror!void {} diff --git a/zig/src/core/load_ledger.zig b/zig/src/core/load_ledger.zig index e855f0a7a0..faabbf20d3 100644 --- a/zig/src/core/load_ledger.zig +++ b/zig/src/core/load_ledger.zig @@ -1,6 +1,7 @@ //! The server's load ledger: scheduler load a host publishes as it happens, read whole by /metrics and /v1/load. const std = @import("std"); const api = @import("engine_api.zig"); +const reasons = @import("wait_reason.zig"); /// The model routes `http_requests_active` counts apart; a fixed set, so the label never grows. pub const RouteGroup = enum { chat_completions, completions, responses, messages, embeddings, speech, other }; @@ -20,6 +21,10 @@ pub const Requests = struct { started: u64 = 0, requeued: u64 = 0, yields: u64 = 0, + /// The queued requests by fixed wait reason; their sum is `waiting`. + waiting_reasons: [reasons.wait_count]u32 = @splat(0), + /// Refused requests by fixed reason; counters, folded into `base` at detach. + refusals: reasons.Refusals = .{}, }; /// The `work` group: prompt and output tokens as they are computed, and the engine time of their prompt calls. @@ -181,6 +186,7 @@ fn fold(base: *Groups, cur: Groups) void { b.started += r.started; b.requeued += r.requeued; b.yields += r.yields; + b.refusals = reasons.Refusals.add(b.refusals, r.refusals); } if (cur.work) |w| base.work = addWork(w, base.work.?); } @@ -191,6 +197,7 @@ fn addRequests(r: Requests, base: Requests) Requests { out.started += base.started; out.requeued += base.requeued; out.yields += base.yields; + out.refusals = reasons.Refusals.add(out.refusals, base.refusals); return out; } diff --git a/zig/src/core/wait_reason.zig b/zig/src/core/wait_reason.zig new file mode 100644 index 0000000000..1048bb8810 --- /dev/null +++ b/zig/src/core/wait_reason.zig @@ -0,0 +1,128 @@ +//! One fixed code for every way a request waits or is refused: bounded values, no request data (issue #2135). +const std = @import("std"); + +/// Why a queued request is not running. Every lane host wait path maps to exactly one. +pub const Wait = enum(u8) { + engine_busy, // every lane holds an admitted request + prefix_cut, // another prompt is filling a cut this request would resume + state_read, // its disk state is still being read ahead + memory_pressure, // admission holds under the memory governor + engine_unavailable, // a Living Weights put-back has not reached the device + ledger_bytes, // the byte ledger promises no room yet + pool_room, // the family's pool has no room for its caches + kv_pool, // DeepSeek-V4.1's round planner waits for KV pool pages + memory_floor, // DeepSeek-V4.1's round planner waits for the memory floor + prefill_concurrent, // DeepSeek-V4.1's round planner waits behind a long prompt prefilling +}; + +/// Why a request was refused. Every lane host refusal path maps to exactly one. +pub const Refusal = enum(u8) { + engine_unavailable, // the host is closing, unhealthy or waiting for a device put-back + engine_busy, // no job or queue slot could be allocated + invalid_request, // the family's request checks refused it before admission + memory_pressure, // critical memory pressure (a 503) + prompt_too_large, // the prompt is past the whole ledger alone + lane_memory_alone, // no lane memory for its caches with nothing else running + backend_prompt, // its prompt pass failed + backend_round, // a shared round failed + restore_refused, // its kept state could not be restored: it prefills from 0 +}; + +pub const wait_count = @typeInfo(Wait).@"enum".field_names.len; +pub const refusal_count = @typeInfo(Refusal).@"enum".field_names.len; + +/// The code a wait reports: its tag name, fixed at compile time. +pub fn waitCode(w: Wait) []const u8 { + return @tagName(w); +} + +/// The code a refusal reports: its tag name, fixed at compile time. +pub fn refusalCode(r: Refusal) []const u8 { + return switch (r) { + .engine_busy => waitCode(.engine_busy), + .engine_unavailable => waitCode(.engine_unavailable), + .memory_pressure => waitCode(.memory_pressure), + else => @tagName(r), + }; +} + +/// Queued requests by wait reason; the sum is the host's `waiting` count. +pub const Waits = struct { + counts: [wait_count]u32 = @splat(0), + + pub fn note(w: *Waits, r: Wait) void { + w.counts[@backingInt(r)] +|= 1; + } + + pub fn count(w: Waits, r: Wait) u32 { + return w.counts[@backingInt(r)]; + } + + pub fn total(w: Waits) u32 { + var n: u32 = 0; + for (w.counts) |c| n +|= c; + return n; + } +}; + +/// Refused requests by refusal reason; a counter, so it survives a host detach. +pub const Refusals = struct { + counts: [refusal_count]u64 = @splat(0), + + pub fn note(r: *Refusals, why: Refusal) void { + r.counts[@backingInt(why)] +|= 1; + } + + pub fn count(r: Refusals, why: Refusal) u64 { + return r.counts[@backingInt(why)]; + } + + pub fn add(a: Refusals, b: Refusals) Refusals { + var out = a; + for (b.counts, 0..) |c, i| out.counts[i] +|= c; + return out; + } +}; + +test "every wait and refusal code is a fixed tag name within the label bound" { + var waits: [wait_count][]const u8 = undefined; + var refusals: [refusal_count][]const u8 = undefined; + for (std.enums.values(Wait), 0..) |w, i| waits[i] = waitCode(w); + for (std.enums.values(Refusal), 0..) |r, i| refusals[i] = refusalCode(r); + for ([_][]const []const u8{ &waits, &refusals }) |codes| for (codes, 0..) |code, i| { + try std.testing.expect(code.len > 0 and code.len <= 32); + for (code) |c| try std.testing.expect((c >= 'a' and c <= 'z') or c == '_'); + for (codes[0..i]) |prior| try std.testing.expect(!std.mem.eql(u8, prior, code)); + }; +} + +test "wait counts add by reason and keep their total" { + var w: Waits = .{}; + w.note(.engine_busy); + w.note(.engine_busy); + w.note(.prefix_cut); + try std.testing.expectEqual(@as(u32, 2), w.count(.engine_busy)); + try std.testing.expectEqual(@as(u32, 1), w.count(.prefix_cut)); + try std.testing.expectEqual(@as(u32, 0), w.count(.kv_pool)); + try std.testing.expectEqual(@as(u32, 3), w.total()); +} + +test "refusal counts add across a host detach" { + var a: Refusals = .{}; + a.note(.backend_prompt); + var b: Refusals = .{}; + b.note(.backend_prompt); + b.note(.memory_pressure); + const sum = Refusals.add(a, b); + try std.testing.expectEqual(@as(u64, 2), sum.count(.backend_prompt)); + try std.testing.expectEqual(@as(u64, 1), sum.count(.memory_pressure)); +} + +test "capacity codes share the same names for queued and refused requests" { + try std.testing.expectEqualStrings("engine_busy", waitCode(.engine_busy)); + try std.testing.expectEqualStrings("engine_unavailable", waitCode(.engine_unavailable)); + try std.testing.expectEqualStrings("memory_pressure", waitCode(.memory_pressure)); + try std.testing.expectEqualStrings(waitCode(.engine_busy), refusalCode(.engine_busy)); + try std.testing.expectEqualStrings(waitCode(.engine_unavailable), refusalCode(.engine_unavailable)); + try std.testing.expectEqualStrings(waitCode(.memory_pressure), refusalCode(.memory_pressure)); +} diff --git a/zig/src/families/deepseek_v41/sched_host.zig b/zig/src/families/deepseek_v41/sched_host.zig index 7d6bcc7d4d..c166817335 100644 --- a/zig/src/families/deepseek_v41/sched_host.zig +++ b/zig/src/families/deepseek_v41/sched_host.zig @@ -89,7 +89,7 @@ pub const SchedHost = struct { const before = h.p.counts; const v = h.p.admit(&h.round, .{ .key = ask.key, .n = ask.prompt.len, .max_new = ask.max_tokens, .submitted = ask.submitted_s }, store, h.now(), h.memory()) catch |e| { words.* = @errorName(e); - return .refuse; + return .{ .refuse = .backend }; }; h.logWait(ask, v, before); switch (v) { @@ -97,7 +97,7 @@ pub const SchedHost = struct { h.admits.put(h.gpa, ask.key, a) catch { h.p.ended(ask.key); words.* = "out of memory"; - return .refuse; + return .{ .refuse = .backend }; }; return .admit; }, @@ -108,7 +108,10 @@ pub const SchedHost = struct { .too_large => "the request needs more KV pool pages than the pool has", .memory => "the prompt's prefill cannot fit in this server's memory now (TF_DSV41_FLOOR_HARD_GIB)", }; - return .refuse; + return .{ .refuse = switch (r) { + .too_large => .too_large, + .memory => .memory, + } }; }, } } @@ -123,13 +126,19 @@ pub const SchedHost = struct { const waited = (h.waits.waiting(h.gpa, ask.key, t) catch return) orelse return; if (c.pool_waits > before.pool_waits) { const free = if (h.gt.sessions) |ss| ss.store.pool.available() else 0; - log.info("admission: request {x} ({d} prompt tokens) has waited {d:.0} s for the KV pool: {d} pages wanted, {d} free, running requests hold the rest", .{ ask.key, ask.prompt.len, waited, h.want, free }); + log.info("admission: request {x} ({d} prompt tokens) has waited {d:.0} s reason={s}: {d} pages wanted, {d} free, running requests hold the rest", .{ ask.key, ask.prompt.len, waited, api.wait_reason.waitCode(.kv_pool), h.want, free }); } else if (c.mem_waits > before.mem_waits) { - log.info("admission: request {x} ({d} prompt tokens) has waited {d:.0} s for the memory floor (TF_DSV41_FLOOR_HARD_GIB)", .{ ask.key, ask.prompt.len, waited }); - } else log.info("admission: request {x} ({d} prompt tokens) has waited {d:.0} s behind a long prompt prefilling (TF_DSV41_PREFILL_CONCURRENT)", .{ ask.key, ask.prompt.len, waited }); + log.info("admission: request {x} ({d} prompt tokens) has waited {d:.0} s reason={s} (TF_DSV41_FLOOR_HARD_GIB)", .{ ask.key, ask.prompt.len, waited, api.wait_reason.waitCode(.memory_floor) }); + } else log.info("admission: request {x} ({d} prompt tokens) has waited {d:.0} s reason={s} (TF_DSV41_PREFILL_CONCURRENT)", .{ ask.key, ask.prompt.len, waited, api.wait_reason.waitCode(.prefill_concurrent) }); }, .admit => if (h.waits.done(ask.key, t)) |waited| log.info("admission: request {x} admitted after {d:.1} s", .{ ask.key, waited }), - .refuse => _ = h.waits.done(ask.key, t), + .refuse => |r| { + _ = h.waits.done(ask.key, t); + log.info("admission: request {x} ({d} prompt tokens) refused reason={s}", .{ ask.key, ask.prompt.len, api.wait_reason.refusalCode(switch (r) { + .too_large => .prompt_too_large, + .memory => .lane_memory_alone, + }) }); + }, } } diff --git a/zig/src/server/load_metrics.zig b/zig/src/server/load_metrics.zig index 6f3d6a3e84..646db42484 100644 --- a/zig/src/server/load_metrics.zig +++ b/zig/src/server/load_metrics.zig @@ -29,6 +29,14 @@ pub fn render(w: *std.Io.Writer, s: ledger.Snapshot, id: Identity) !void { try w.print("{s}http_requests_active{{route=\"{s}\"}} {d}\n", .{ prefix, @tagName(g), s.active(g) }); if (s.requests) |r| { try gauge(w, "waiting_prompt_tokens", "gauge", "Prompt tokens of the queued requests, before any kept prefix restores part of them.", r.waiting_prompt_tokens); + try family(w, "requests_waiting_by_reason", "gauge", "Queued requests by fixed wait reason; the sum is requests_waiting."); + for (std.enums.values(api.wait_reason.Wait)) |why| { + try w.print("{s}requests_waiting_by_reason{{reason=\"{s}\"}} {d}\n", .{ prefix, api.wait_reason.waitCode(why), r.waiting_reasons[@backingInt(why)] }); + } + try family(w, "request_refusals_total", "counter", "Refused requests by fixed reason."); + for (std.enums.values(api.wait_reason.Refusal)) |why| { + try w.print("{s}request_refusals_total{{reason=\"{s}\"}} {d}\n", .{ prefix, api.wait_reason.refusalCode(why), r.refusals.count(why) }); + } if (r.filling_left) |left| try gauge(w, "filling_prompt_tokens_left", "gauge", "Prompt tokens still to prefill in the foreground passes filling now.", left); try gauge(w, "requests_started_total", "counter", "Requests the engine accepted into its queue.", r.started); try gauge(w, "requests_requeued_total", "counter", "Admitted requests that lost their lane for memory before a first token and queued again.", r.requeued); @@ -77,6 +85,16 @@ pub fn object(a: Allocator, s: ledger.Snapshot, now_s: f64, model: []const u8) A try o.put(a, "requests_requeued", try json.intValue(a, r.requeued)); try o.put(a, "background_yields", try json.intValue(a, r.yields)); try o.put(a, "waiting_prompt_tokens", try json.intValue(a, r.waiting_prompt_tokens)); + const waits = try json.newObject(a); + for (std.enums.values(api.wait_reason.Wait)) |why| { + try waits.put(a, api.wait_reason.waitCode(why), try json.intValue(a, r.waiting_reasons[@backingInt(why)])); + } + try o.put(a, "waiting_by_reason", .{ .object = waits }); + const refused = try json.newObject(a); + for (std.enums.values(api.wait_reason.Refusal)) |why| { + try refused.put(a, api.wait_reason.refusalCode(why), try json.intValue(a, r.refusals.count(why))); + } + try o.put(a, "refusals", .{ .object = refused }); if (r.filling_left) |left| try o.put(a, "filling_prompt_tokens_left", try json.intValue(a, left)); try o.put(a, "fill_unknown", .{ .bool = r.fill_unknown }); } diff --git a/zig/src/server/load_metrics_test.zig b/zig/src/server/load_metrics_test.zig index aac7832e8e..0ab167239a 100644 --- a/zig/src/server/load_metrics_test.zig +++ b/zig/src/server/load_metrics_test.zig @@ -57,7 +57,13 @@ test "a live scrape shows each group once, beside the request counts metrics.zig defer out.deinit(); var s = snapshot(false); s.http[@backingInt(ledger.RouteGroup.responses)] = 2; - s.requests = .{ .running = 3, .waiting = 2, .waiting_prompt_tokens = 9000, .filling_left = 512, .started = 40, .requeued = 1, .yields = 2 }; + var waits: api.wait_reason.Waits = .{}; + waits.note(.engine_busy); + waits.note(.pool_room); + var refused: api.wait_reason.Refusals = .{}; + refused.note(.backend_prompt); + refused.note(.backend_prompt); + s.requests = .{ .running = 3, .waiting = 2, .waiting_prompt_tokens = 9000, .filling_left = 512, .started = 40, .requeued = 1, .yields = 2, .waiting_reasons = waits.counts, .refusals = refused }; s.work = .{ .prefill_compute = 1000, .prefill_cache = 400, .decode = 300, .prefill_busy_ns = 1_500_000_000 }; s.capacity = .{ .mode = .paged, .lanes_free = 1 }; s.pool = .{ .capacity = 4096, .available = 1024, .reclaimable = 512, .promised = 256 }; @@ -70,6 +76,11 @@ test "a live scrape shows each group once, beside the request counts metrics.zig "tensorfold:engine_parked 0\n", "tensorfold:http_requests_active{route=\"responses\"} 2\n", "tensorfold:waiting_prompt_tokens 9000\n", + "tensorfold:requests_waiting_by_reason{reason=\"engine_busy\"} 1\n", + "tensorfold:requests_waiting_by_reason{reason=\"pool_room\"} 1\n", + "tensorfold:requests_waiting_by_reason{reason=\"kv_pool\"} 0\n", + "tensorfold:request_refusals_total{reason=\"backend_prompt\"} 2\n", + "tensorfold:request_refusals_total{reason=\"restore_refused\"} 0\n", "tensorfold:filling_prompt_tokens_left 512\n", "tensorfold:requests_started_total 40\n", "tensorfold:scheduler_tokens_total{mode=\"prefill_compute\"} 1000\n", @@ -120,16 +131,21 @@ test "the /v1/load object carries every group a host fills, and only server valu const a = arena.allocator(); var s = snapshot(false); s.http[@backingInt(ledger.RouteGroup.chat_completions)] = 3; - s.requests = .{ .running = 3, .waiting = 2, .waiting_prompt_tokens = 9000, .filling_left = 512, .started = 40, .requeued = 1, .yields = 2 }; + var waits: api.wait_reason.Waits = .{}; + waits.note(.engine_busy); + var refused: api.wait_reason.Refusals = .{}; + refused.note(.backend_prompt); + s.requests = .{ .running = 3, .waiting = 2, .waiting_prompt_tokens = 9000, .filling_left = 512, .started = 40, .requeued = 1, .yields = 2, .waiting_reasons = waits.counts, .refusals = refused }; s.work = .{ .prefill_compute = 1000, .prefill_cache = 400, .decode = 300, .prefill_busy_ns = 1_500_000 }; s.capacity = .{ .mode = .paged, .lanes_free = 1 }; s.pool = .{ .capacity = 4096, .available = 1024, .reclaimable = 512, .promised = 256, .run = false }; const live = try loadBody(a, s); for ([_][]const u8{ - "\"instance\":\"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\"", "\"parked\":false", "\"chat_completions\":3", - "\"running\":3", "\"waiting_prompt_tokens\":9000", "\"filling_prompt_tokens_left\":512", - "\"mode\":\"paged\"", "\"available_tokens\":1024", "\"run_available\":false", - "\"prefill_compute\":1000", "\"busy_us\":{\"prefill\":1500}", + "\"instance\":\"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\"", "\"parked\":false", "\"chat_completions\":3", + "\"running\":3", "\"waiting_prompt_tokens\":9000", "\"filling_prompt_tokens_left\":512", + "\"mode\":\"paged\"", "\"available_tokens\":1024", "\"run_available\":false", + "\"prefill_compute\":1000", "\"busy_us\":{\"prefill\":1500}", "\"waiting_by_reason\":{\"engine_busy\":1,\"prefix_cut\":0", + "\"refusals\":{\"engine_unavailable\":0,\"engine_busy\":0,\"invalid_request\":0,\"memory_pressure\":0,\"prompt_too_large\":0,\"lane_memory_alone\":0,\"backend_prompt\":1", }) |key| { if (!has(live, key)) std.debug.print("missing {s} in {s}\n", .{ key, live }); try std.testing.expect(has(live, key)); diff --git a/zig/src/server/wait_reason.zig b/zig/src/server/wait_reason.zig index 3ec4ad09b4..08ab5a64ec 100644 --- a/zig/src/server/wait_reason.zig +++ b/zig/src/server/wait_reason.zig @@ -1,5 +1,6 @@ //! One reason for every wait: the fixed code a client keys on, and the words it reads. const std = @import("std"); +const reasons = @import("engine_api").wait_reason; /// A retryable wait (HTTP 503 with Retry-After): the code clients key on and the message they read. pub const Wait = struct { code: []const u8, message: []const u8 }; @@ -15,9 +16,9 @@ pub const reload_failed: Wait = .{ .code = "engine_reload_failed", .message = "t pub const unloading_problem = "the engine is still unloading; send the request again"; /// A retryable capacity refusal: the engine had no room for this request and it can be sent again. -pub const engine_busy = "engine_busy"; -pub const engine_unavailable = "engine_unavailable"; -pub const memory_pressure = "memory_pressure"; +pub const engine_busy = reasons.waitCode(.engine_busy); +pub const engine_unavailable = reasons.waitCode(.engine_unavailable); +pub const memory_pressure = reasons.waitCode(.memory_pressure); /// Every server wait and capacity code, in one bounded set. pub const all = [_]Wait{ restarting, loading, reload_failed }; diff --git a/zig/src/server/wait_reason_test.zig b/zig/src/server/wait_reason_test.zig index a4e5030045..885d55ab35 100644 --- a/zig/src/server/wait_reason_test.zig +++ b/zig/src/server/wait_reason_test.zig @@ -210,3 +210,13 @@ test "wait reasons HTTP: a failed reload answers 503 engine_reload_failed and le try t.expectEqual(round + 1, broken.attempts); } } + +test "wait reasons: HTTP capacity codes equal load and scrape codes" { + const reasons = api.wait_reason; + try t.expectEqualStrings(wait.engine_busy, reasons.waitCode(.engine_busy)); + try t.expectEqualStrings(wait.engine_busy, reasons.refusalCode(.engine_busy)); + try t.expectEqualStrings(wait.engine_unavailable, reasons.waitCode(.engine_unavailable)); + try t.expectEqualStrings(wait.engine_unavailable, reasons.refusalCode(.engine_unavailable)); + try t.expectEqualStrings(wait.memory_pressure, reasons.waitCode(.memory_pressure)); + try t.expectEqualStrings(wait.memory_pressure, reasons.refusalCode(.memory_pressure)); +}