Skip to content

Commit b699b3a

Browse files
authored
Merge pull request #45 from abgnydn/critique/t2-server-loader
test: pin degraded-concurrency policy, station surface, model-mismatch 400
2 parents e05c3ed + ed4f44b commit b699b3a

5 files changed

Lines changed: 345 additions & 30 deletions

File tree

scripts/agent-server-test.mjs

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,25 @@ try {
6363
check('/v1/models', models.object === 'list' && models.data?.length > 0,
6464
models.data?.[0]?.id ?? '(none)')
6565

66+
// 1b. Model mismatch is a hard 400 naming both sides (the LM Studio trap:
67+
// never serve the resident model under a foreign name). Needs no tokens —
68+
// the tab only has to be connected, which check 1 already proved.
69+
const mmRes = await fetch(`${BASE}/v1/chat/completions`, {
70+
method: 'POST', headers: { 'content-type': 'application/json' },
71+
body: JSON.stringify({
72+
model: 'definitely-not-a-model',
73+
messages: [{ role: 'user', content: 'hi' }],
74+
}),
75+
})
76+
const mmBody = await mmRes.json().catch(() => ({}))
77+
check('model mismatch is 400', mmRes.status === 400,
78+
`HTTP ${mmRes.status} · ${(mmBody.error?.message ?? '').slice(0, 80)}`)
79+
check('mismatch names both models',
80+
typeof mmBody.error?.message === 'string'
81+
&& mmBody.error.message.includes('definitely-not-a-model')
82+
&& mmBody.error.message.includes(health.hosting),
83+
(mmBody.error?.message ?? '').slice(0, 80))
84+
6685
// 2. Non-streaming completion.
6786
const t0 = Date.now()
6887
const r1 = await (await fetch(`${BASE}/v1/chat/completions`, {

src/zero-tvm/map-limited.ts

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,54 @@ export function backoffMs(attempt: number, base = 500): number {
2222
return base * Math.pow(2, attempt) * (0.5 + Math.random())
2323
}
2424

25+
/**
26+
* DEGRADED CONCURRENCY TRACKER — sticky session policy for flaky networks.
27+
*
28+
* Extracted from weight-loader.ts (which touches GPUBufferUsage at module
29+
* scope and can't be imported outside a browser) so the streak policy is
30+
* unit-testable: degrade on the first transient failure, recover after N
31+
* consecutive clean fetches, and never flip-flop on a stray success.
32+
*/
33+
export interface DegradedTracker {
34+
readonly degraded: boolean
35+
markTransient(onNotice?: (msg: string) => void): void
36+
markSuccess(onNotice?: (msg: string) => void): void
37+
limitOf(): number
38+
}
39+
40+
export function createDegradedTracker(opts: {
41+
full: number
42+
degraded: number
43+
recoverAfter?: number
44+
}): DegradedTracker {
45+
const recoverAfter = opts.recoverAfter ?? 10
46+
let degraded = false
47+
let streak = 0
48+
return {
49+
get degraded() { return degraded },
50+
markTransient(onNotice) {
51+
streak = 0 // the recovery streak is consecutive, not cumulative
52+
if (!degraded) {
53+
degraded = true
54+
onNotice?.(`network unstable — dropping shard concurrency ${opts.full}${opts.degraded}`)
55+
}
56+
},
57+
markSuccess(onNotice) {
58+
if (!degraded) return
59+
streak++
60+
if (streak >= recoverAfter) {
61+
degraded = false
62+
streak = 0 // hygiene: every episode starts with markTransient (which
63+
// zeroes), so this changes no observable path — it keeps the
64+
// invariant ("streak counts this episode's clean fetches")
65+
// structural instead of order-dependent.
66+
onNotice?.(`network stable — restoring shard concurrency ${opts.degraded}${opts.full}`)
67+
}
68+
},
69+
limitOf() { return degraded ? opts.degraded : opts.full },
70+
}
71+
}
72+
2573
export interface MapLimitedOptions {
2674
/** Extra worker runs per item after its first failure (default 0). */
2775
retries?: number

src/zero-tvm/weight-loader.ts

Lines changed: 11 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ import { openMlxCheckpoint, assembleMlx, planModel, planKey, progressEvery, type
1616
import {
1717
fetchSlabSource, opfsSlabSource, type SlabDirectory, type SlabSource,
1818
} from './slab-source.js'
19-
import { mapLimited, backoffMs } from './map-limited.js'
19+
import { mapLimited, backoffMs, createDegradedTracker } from './map-limited.js'
2020

2121
// ============================================================
2222
// Model URL + cache dir
@@ -58,25 +58,11 @@ const DEGRADED_FETCH_CONCURRENCY = 3
5858
* Exponential backoff with jitter: ~0.5s, 1s, 2s, 4s (see backoffMs). */
5959
const FETCH_RETRIES = 4
6060

61-
/** Set on the first mid-stream network failure; sticky for the page session
62-
* so a retried loadWeights also runs at the reduced concurrency. */
63-
let networkDegraded = false
64-
65-
/** Count of consecutive shard fetches that completed without hitting a
66-
* transient failure while in degraded mode. Ten successes reset concurrency
67-
* back to full — a single stray reset does not flip-flop the session. */
68-
let consecutiveSuccesses = 0
69-
70-
/** Called when a shard fetch completes without transients. Resets the session
71-
* back to full concurrency after 10 consecutive clean fetches. */
72-
function markFetchSuccess(onRetry?: (msg: string) => void): void {
73-
if (!networkDegraded) return
74-
consecutiveSuccesses++
75-
if (consecutiveSuccesses >= 10) {
76-
networkDegraded = false
77-
onRetry?.(`network stable — restoring shard concurrency ${DEGRADED_FETCH_CONCURRENCY}${FETCH_CONCURRENCY}`)
78-
}
79-
}
61+
/** Session network health: sticky degrade on the first mid-stream failure so
62+
* a retried loadWeights runs at the reduced concurrency, recovering after
63+
* ten consecutive clean fetches. Policy lives in map-limited.ts (testable);
64+
* this is the one session instance. */
65+
const netHealth = createDegradedTracker({ full: FETCH_CONCURRENCY, degraded: DEGRADED_FETCH_CONCURRENCY })
8066

8167
// ============================================================
8268
// ndarray-cache.json types
@@ -245,8 +231,7 @@ async function opfsWrite(dir: OPFSDir, dataPath: string, data: ArrayBuffer): Pro
245231
* Range header to continue from the bytes already received when the server
246232
* honors it (206). A 200 answer to a Range request restarts cleanly.
247233
*
248-
* The first mid-stream failure flips the session-wide `networkDegraded` flag
249-
* (concurrency drops from FETCH_CONCURRENCY to DEGRADED_FETCH_CONCURRENCY).
234+
* The first mid-stream failure degrades session concurrency (see netHealth).
250235
*/
251236
async function fetchBufWithRetry(
252237
url: string,
@@ -261,11 +246,7 @@ async function fetchBufWithRetry(
261246
// A mid-stream / connection failure — degrade session concurrency once.
262247
const transient = (e: unknown) => {
263248
lastErr = e
264-
consecutiveSuccesses = 0 // the streak above is consecutive, not cumulative
265-
if (!networkDegraded) {
266-
networkDegraded = true
267-
onRetry?.(`network unstable — dropping shard concurrency ${FETCH_CONCURRENCY}${DEGRADED_FETCH_CONCURRENCY}`)
268-
}
249+
netHealth.markTransient(onRetry)
269250
}
270251
for (let attempt = 0; attempt <= retries; attempt++) {
271252
if (attempt > 0) {
@@ -330,7 +311,7 @@ async function fetchBufWithRetry(
330311
transient(new Error(`truncated body: got ${received} of ${expectTotal} bytes`))
331312
continue
332313
}
333-
markFetchSuccess(onRetry)
314+
netHealth.markSuccess(onRetry)
334315
const only = chunks.length === 1 ? chunks[0] : null
335316
if (only && only.byteOffset === 0 && only.byteLength === only.buffer.byteLength) {
336317
return only.buffer
@@ -776,14 +757,14 @@ export async function loadWeights(
776757

777758
try {
778759
// Concurrency is re-read between shards so the first mid-stream network
779-
// failure (networkDegraded) drops parallelism for the rest of the session.
760+
// failure drops parallelism (netHealth) for the rest of the session.
780761
// Per-shard failures are non-fatal until that shard exhausts its own
781762
// retries (FETCH_RETRIES+1 network attempts inside fetchBufWithRetry, ×2
782763
// full tier walks via mapLimited's retries) — only then does the shared
783764
// AbortSignal cancel the sibling fetches still in flight.
784765
await mapLimited(
785766
shardEntries,
786-
() => (networkDegraded ? DEGRADED_FETCH_CONCURRENCY : FETCH_CONCURRENCY),
767+
() => netHealth.limitOf(),
787768
async ([dataPath, records], _idx, signal) => {
788769
let shard: ArrayBuffer
789770
try {
Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
/**
2+
* UNIT TESTS — degraded-concurrency tracker + backoff (src/zero-tvm/map-limited.ts).
3+
*
4+
* The weight loader drops shard concurrency 8 → 3 on the first mid-stream
5+
* failure and recovers after ten consecutive clean fetches. That policy used
6+
* to live in weight-loader.ts module state — next to GPUBufferUsage reads
7+
* that make the module unimportable in Node — so no test could touch it.
8+
* createDegradedTracker carries the identical policy in the import-safe
9+
* module; these tests pin it, including the streak-reset edge the old code
10+
* got wrong twice (first as a never-resetting flag, then as a streak the
11+
* failure path never zeroed).
12+
*/
13+
14+
import { describe, test, expect } from 'vitest'
15+
import { backoffMs, createDegradedTracker } from '../../src/zero-tvm/map-limited.js'
16+
17+
describe('backoffMs', () => {
18+
test('stays inside [0.5, 1.5) × base·2^attempt', () => {
19+
for (let attempt = 0; attempt < 6; attempt++) {
20+
for (let i = 0; i < 200; i++) {
21+
const ms = backoffMs(attempt)
22+
expect(ms).toBeGreaterThanOrEqual(0.5 * 500 * 2 ** attempt)
23+
expect(ms).toBeLessThan(1.5 * 500 * 2 ** attempt)
24+
}
25+
}
26+
})
27+
28+
test('honors a custom base', () => {
29+
for (let i = 0; i < 100; i++) {
30+
const ms = backoffMs(2, 100)
31+
expect(ms).toBeGreaterThanOrEqual(200)
32+
expect(ms).toBeLessThan(600)
33+
}
34+
})
35+
36+
test('means double per attempt: E[jitter] is 1.0', () => {
37+
// Bands overlap ([0.5,1.5) spans 3x against 2x growth), so per-sample
38+
// ordering is NOT guaranteed — but the mean of attempt a+1 is 2x the
39+
// mean of attempt a. 2000 samples make ±10% watertight (SE ≈ 0.6%).
40+
for (let attempt = 0; attempt < 4; attempt++) {
41+
let sum = 0
42+
const N = 2000
43+
for (let i = 0; i < N; i++) sum += backoffMs(attempt)
44+
const mean = sum / N
45+
const expected = 500 * 2 ** attempt
46+
expect(mean).toBeGreaterThan(expected * 0.9)
47+
expect(mean).toBeLessThan(expected * 1.1)
48+
}
49+
})
50+
})
51+
52+
describe('createDegradedTracker', () => {
53+
test('starts healthy at full concurrency, successes are no-ops', () => {
54+
const t = createDegradedTracker({ full: 8, degraded: 3 })
55+
expect(t.degraded).toBe(false)
56+
expect(t.limitOf()).toBe(8)
57+
const notes: string[] = []
58+
t.markSuccess((m) => notes.push(m))
59+
expect(t.degraded).toBe(false)
60+
expect(notes).toEqual([])
61+
})
62+
63+
test('first transient degrades once and announces', () => {
64+
const t = createDegradedTracker({ full: 8, degraded: 3 })
65+
const notes: string[] = []
66+
t.markTransient((m) => notes.push(m))
67+
expect(t.degraded).toBe(true)
68+
expect(t.limitOf()).toBe(3)
69+
expect(notes).toHaveLength(1)
70+
expect(notes[0]).toContain('8 → 3')
71+
// A second failure while degraded announces nothing new.
72+
t.markTransient((m) => notes.push(m))
73+
expect(notes).toHaveLength(1)
74+
})
75+
76+
test('nine clean fetches do not recover; the tenth does', () => {
77+
const t = createDegradedTracker({ full: 8, degraded: 3 })
78+
const notes: string[] = []
79+
t.markTransient()
80+
for (let i = 0; i < 9; i++) t.markSuccess((m) => notes.push(m))
81+
expect(t.degraded).toBe(true)
82+
expect(t.limitOf()).toBe(3)
83+
expect(notes).toEqual([])
84+
t.markSuccess((m) => notes.push(m))
85+
expect(t.degraded).toBe(false)
86+
expect(t.limitOf()).toBe(8)
87+
expect(notes).toHaveLength(1)
88+
expect(notes[0]).toContain('3 → 8')
89+
})
90+
91+
test('a failure zeroes the streak: nine + failure + nine stays degraded', () => {
92+
const t = createDegradedTracker({ full: 8, degraded: 3 })
93+
t.markTransient()
94+
for (let i = 0; i < 9; i++) t.markSuccess()
95+
t.markTransient()
96+
for (let i = 0; i < 9; i++) t.markSuccess()
97+
expect(t.degraded).toBe(true)
98+
t.markSuccess()
99+
expect(t.degraded).toBe(false)
100+
})
101+
102+
test('recovery resets the streak: the next degradation needs ten again', () => {
103+
// Strictly, the reset is hygiene, not a behavior change: every degraded
104+
// episode begins with markTransient (the only path that sets degraded),
105+
// which zeroes the streak first — so the counter is 0 at each episode
106+
// start with or without the reset. It is kept because the invariant
107+
// ("streak counts clean fetches in the CURRENT episode") should hold
108+
// structurally, not incidentally via call order. This test pins the
109+
// ten-again behavior against a future refactor that breaks that order.
110+
const t = createDegradedTracker({ full: 8, degraded: 3 })
111+
t.markTransient()
112+
for (let i = 0; i < 10; i++) t.markSuccess()
113+
expect(t.degraded).toBe(false)
114+
t.markTransient()
115+
t.markSuccess()
116+
expect(t.degraded).toBe(true)
117+
for (let i = 0; i < 9; i++) t.markSuccess()
118+
expect(t.degraded).toBe(false)
119+
})
120+
121+
test('recoverAfter is configurable', () => {
122+
const t = createDegradedTracker({ full: 4, degraded: 1, recoverAfter: 2 })
123+
t.markTransient()
124+
t.markSuccess()
125+
expect(t.degraded).toBe(true)
126+
t.markSuccess()
127+
expect(t.degraded).toBe(false)
128+
expect(t.limitOf()).toBe(4)
129+
})
130+
})

0 commit comments

Comments
 (0)