diff --git a/crates/inferproxy/src/lib.rs b/crates/inferproxy/src/lib.rs index 7309520..b987212 100644 --- a/crates/inferproxy/src/lib.rs +++ b/crates/inferproxy/src/lib.rs @@ -108,28 +108,34 @@ pub fn serve(listener: UnixListener, upstream: String, limits: Limits) -> io::Re } fn handle(client: UnixStream, upstream: String, open: Arc) { - let _guard = OpenGuard::new(open); let server = match TcpStream::connect(&upstream) { Ok(s) => s, Err(_) => return, }; - let _ = forward(&client, &server); - drop(_guard); + let c2s = { + let _guard = OpenGuard::new(open); + match forward(&client, &server) { + Ok(c2s) => c2s, + Err(_) => return, + } + }; + let _ = client.shutdown(Shutdown::Both); + let _ = server.shutdown(Shutdown::Both); + let _ = c2s.join(); } -fn forward(client: &UnixStream, server: &TcpStream) -> io::Result<()> { +fn forward(client: &UnixStream, server: &TcpStream) -> io::Result> { let mut c2s_in = client.try_clone()?; let mut s2c_out = client.try_clone()?; let mut s2c_in = server.try_clone()?; let mut c2s_out = server.try_clone()?; - thread::scope(|s| { - s.spawn(move || { - let _ = copy(&mut c2s_in, &mut c2s_out); - let _ = c2s_out.shutdown(Shutdown::Write); - }); - s.spawn(move || { - let _ = copy(&mut s2c_in, &mut s2c_out); - }); + let c2s = thread::spawn(move || { + let _ = copy(&mut c2s_in, &mut c2s_out); + let _ = c2s_out.shutdown(Shutdown::Write); }); - Ok(()) + let _ = thread::spawn(move || { + let _ = copy(&mut s2c_in, &mut s2c_out); + }) + .join(); + Ok(c2s) } diff --git a/crates/inferproxy/tests/forward.rs b/crates/inferproxy/tests/forward.rs index ef2249e..9e2cbab 100644 --- a/crates/inferproxy/tests/forward.rs +++ b/crates/inferproxy/tests/forward.rs @@ -154,3 +154,44 @@ fn an_unreachable_upstream_closes_the_client() { let path = start("127.0.0.1:1".to_string(), Limits::default()); assert!(was_refused(&path)); } + +/// An upstream that answers at once and closes, whether or not the client has finished sending. +/// This is what `llama-server` does, and what a server that dies mid-answer looks like. +fn answer_and_close_upstream() -> String { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + thread::spawn(move || { + for stream in listener.incoming() { + let mut stream = stream.unwrap(); + let mut first = [0u8; 1]; + let _ = stream.read(&mut first); + stream.write_all(b"answer").unwrap(); + // dropping `stream` closes it + } + }); + addr +} + +#[test] +fn the_upstreams_close_reaches_a_client_that_is_still_sending() { + // The HTTP client in loopd never half-closes: it keeps its sending side open until it drops + // the connection. When the server closes, the proxy must close towards the client at once, + // or the client only learns of a dead server from its own timeout. + let path = start(answer_and_close_upstream(), Limits::default()); + let mut s = UnixStream::connect(&path).unwrap(); + s.set_read_timeout(Some(Duration::from_secs(2))).unwrap(); + s.write_all(b"request").unwrap(); + let started = std::time::Instant::now(); + let mut got = Vec::new(); + let result = s.read_to_end(&mut got); + assert!( + result.is_ok(), + "the read must end with EOF, not a timeout: {result:?}" + ); + assert_eq!(got, b"answer"); + assert!( + started.elapsed() < Duration::from_millis(1000), + "EOF took {:?}", + started.elapsed() + ); +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 82780f1..7c0b722 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -27,7 +27,8 @@ reviewer adds findings under "Reviews" once per milestone. | M2a/09-llama-chat | 2026-09-18 | done | 1 | pass | none | Wrote crates/loopd/src/llama/chat.rs (`chat`, with the head wait in a separate `wait_for_head`) and registered `pub mod chat;`. `chat` builds the body (build error -> Protocol), opens and POSTs, then `wait_for_head` loops `read_head` at `poll_ms`: a Timeout is classified Idle/Busy/Unavailable by `received_any` then a `slots()` poll, emits `Waiting { slot_busy }` on every poll, keeps a per-state `since` that resets on state change, and returns `WaitTimeout`/`LoadTimeout`/`Stalled` at the right limits; 200 streams via `Events`+`Assembler` mapping `Timeout->Stalled`, `Truncated->StreamClosedEarly`, else `Protocol`, then `finish(false)`; non-200 returns `Http { status, error_text }`. All 13 chat tests pass five runs in a row. Table-to-test map: state (table 1) Busy -> a_busy_slot_is_waited_out / a_slot_that_stays_busy_is_a_wait_timeout, Idle-not-processing -> only_the_requests_own_slot_counts_as_busy, Unavailable -> an_unavailable_server_is_a_load_timeout, Idle-received_any -> a_slow_but_steady_stream_is_not_a_stall (turn1 head is 738 B, pieces are 3140/8=392 B, so the head-wait times out with a partial head); thresholds (table 2) -> a_slot_that_stays_busy_is_a_wait_timeout / an_unavailable_server_is_a_load_timeout / only_the_requests_own_slot_counts_as_busy; stream (table 3) Data/Done/None -> the recorded completion and trickle tests, Timeout -> silence_in_mid_stream_is_a_stall, Truncated -> a_stream_that_is_cut_is_closed_early_wherever_the_cut_falls, other -> garbage_in_the_stream_is_a_protocol_error; clock-restart -> the_wait_clocks_restart_when_the_state_changes. One path fix: info and request live under crate::llama, not crate::, so the imports use crate::llama::{info::..., request::...}. | Ornith-1.5-35B-A3B | | M2a/10-llama-cap | 2026-09-18 | done | 1 | pass | none | Added `Client::end_reasoning` to info.rs: POSTs `{"id","action":"reasoning_end","model"}` to `/v1/chat/completions/control` via `call`, reads `success` as a bool from the server's JSON (ignoring `message`), non-200 stays an Err through `call`, a missing/non-bool `success` is Protocol. Threaded the cap into `chat` step 4: after passing a chunk's events on, when `assembler.in_reasoning()`, a local `cap_at: Option` holds where the cap fired (None while it has not fired); on `tokens >= thinking_cap` it calls `end_reasoning(assembler.id())` once, remembers `tokens` and emits `ThinkingCapped` on `Ok(true)`, returns `ThinkingOverrun` on `Ok(false)`/`Err`, and after firing returns `ThinkingOverrun` once `tokens >= at + thinking_overrun`; `finish(cap_at.is_some())`. The `the_allowance_is_exact` test passes with `>=` in both rows (63 is not `20+44`, and is `>= 20+43`). One guard: a reasoning chunk with no id at cap time is Protocol rather than a panic. 6 cap tests + 13 chat tests pass; `make gate` prints `gate: ok`. | Ornith-1.5-35B-A3B | | M2a/12-selftest | 2026-09-18 | done | 1 | pass | Ornith-1.5-35B-A3B | -| M2a/13-verify-device | 2026-09-18 | done | 1 | pass | none | Wrote no library code. Put the two given files in place (`crates/loopd/tests/device.rs`, and a `Makefile` whose only difference from the old one is the new `verify-device` target), confirmed `make gate` prints `gate: ok` with `device` at `0 passed; 0 failed; 4 ignored`, and `curl http://straylight:11434/health` returned `{"status":"ok"}`. `make verify-device` ran the four `#[ignore]` checks one at a time against the real server on slot 0: self-test, capped thinking block, a request surviving its own proxy being killed and restarted, and a second turn reusing the cache of a first turn that contained thinking. All 4 passed in ~21s (well under two minutes). | ? | Wrote crates/loopd/src/selftest.rs (SelfTestError with Display/std::error::Error/From, and `run` doing the three checks in order) and registered `pub mod selftest;` in lib.rs; rewrote main.rs into `loopd selftest --config `. Check 1 calls props() and compares chat-template sha256, then n_ctx, then total_slots, a sha256 error being Hash so a wrong server is sent no prompt; check 2 posts one read_file tool with chat_with_retry and requires finish_reason ToolCalls, a first call named read_file whose arguments parse as JSON carrying a string path, wrapping an InferError as Infer; check 3 runs turn 1 then an extension of it and maps a cache Loss to CacheMiss. The copied test's cache_outcome/CacheOutcome live at crate::llama::info, so the import follows that. main.rs parses args as &[&str] via a two-step String->&str collect; unknown/missing args are exit 2 and a config load failure is `loopd: ` exit 1. Real server via inferproxy against straylight: minimal.toml gave three step lines and `selftest: ok` exit 0; setting slots=3 gave `selftest: FAILED: slot count: expected 3, got 2` exit 1. | Ornith-1.5-35B-A3B | +| M2a/13-verify-device | 2026-09-18 | done | 1 | pass | none | +| M2a/14-inferproxy-close | 2026-09-18 | done | 2 | fail | none | Made the proxy close towards the client as soon as the server-to-client copy ends, for any reason. `forward` now joins only the s2c thread and returns the c2s JoinHandle, so it returns when the server stops sending instead of waiting for the client to stop sending too; `handle` drops the OpenGuard inside a block scope, then shuts down the client (Both) and server (Both) so the client's read returns EOF at once and the c2s thread ends, then joins c2s. This is rule 4 of task 02 (drop the open-place before closing the client). The half-close when the client stops sending first is unchanged. The copied `forward.rs` is byte-identical to the plan. 7 passed ten runs in a row; `make gate` prints `gate: ok`. | ? | Wrote no library code. Put the two given files in place (`crates/loopd/tests/device.rs`, and a `Makefile` whose only difference from the old one is the new `verify-device` target), confirmed `make gate` prints `gate: ok` with `device` at `0 passed; 0 failed; 4 ignored`, and `curl http://straylight:11434/health` returned `{"status":"ok"}`. `make verify-device` ran the four `#[ignore]` checks one at a time against the real server on slot 0: self-test, capped thinking block, a request surviving its own proxy being killed and restarted, and a second turn reusing the cache of a first turn that contained thinking. All 4 passed in ~21s (well under two minutes). | ? | Wrote crates/loopd/src/selftest.rs (SelfTestError with Display/std::error::Error/From, and `run` doing the three checks in order) and registered `pub mod selftest;` in lib.rs; rewrote main.rs into `loopd selftest --config `. Check 1 calls props() and compares chat-template sha256, then n_ctx, then total_slots, a sha256 error being Hash so a wrong server is sent no prompt; check 2 posts one read_file tool with chat_with_retry and requires finish_reason ToolCalls, a first call named read_file whose arguments parse as JSON carrying a string path, wrapping an InferError as Infer; check 3 runs turn 1 then an extension of it and maps a cache Loss to CacheMiss. The copied test's cache_outcome/CacheOutcome live at crate::llama::info, so the import follows that. main.rs parses args as &[&str] via a two-step String->&str collect; unknown/missing args are exit 2 and a config load failure is `loopd: ` exit 1. Real server via inferproxy against straylight: minimal.toml gave three step lines and `selftest: ok` exit 0; setting slots=3 gave `selftest: FAILED: slot count: expected 3, got 2` exit 1. | Ornith-1.5-35B-A3B | ## Reviews