diff --git a/README.md b/README.md index c6fc69c..03fc2a6 100644 --- a/README.md +++ b/README.md @@ -59,6 +59,9 @@ hosts = ["beta", "alpha"] | `hosts..models` | The models this host serves, with per-model parallel tuning. | | `routes..hosts` | Candidate hosts, tried in order until one is healthy; a conversation leases one of them. | | `routes..default_model` | Model used when a request omits one; must be served by a host in the route. | +| `identity` | `"off"` (default), `"tailscale"`, or `"header"`; see below. | +| `hosts..wake` | A wake-on-LAN target (`mac`, `broadcast`, `wait`) so crossbar can rouse a sleeping host when nothing else can take a new lease. | +| `routes..peers` | The tailnet nodes allowed to reach the route, with `identity = "tailscale"`; see below. | ## Run @@ -159,7 +162,29 @@ crossbar_host_healthy{host="alpha"} 1 crossbar_host_healthy{host="beta"} 1 ``` -## What v1 does not do +## Wake -The context-size guard, wake-on-LAN, Tailscale identity and `/slots` are out of scope for v1; see -`PLAN.md` v2. +When a route has no healthy host left and at least one candidate lists a `wake` target, crossbar +sends that host a wake-on-LAN magic packet, in route order, and retries the lease once. A host that +wakes up takes the conversation; if none wakes, the request gets `503 {"error":"no healthy host", +"woke":[""]}`. The context-size guard wakes a sleeping host the same way before it +answers `400 prompt too large`, when no healthy host's per-slot context can fit the prompt. + +## Identity + +`identity` gates who may use a route. With the default `"off"` every request is admitted. With +`"tailscale"`, a route that lists `peers` answers `403` to any caller whose tailnet address is not +one of them (checked with `tailscale whois`): + +```toml +[routes.hermes-x] +hosts = ["beta", "alpha"] +peers = ["talos"] +``` + +`"header"` trusts the `X-Crossbar-Peer` header instead and needs no tailnet; it is insecure and for +tests only, so crossbar logs a warning when it starts in that mode. + +## What v2 does not do + +Request coalescing, `/slots` and TLS are out of scope for v2; see `PLAN.md`. diff --git a/cmd/crossbar/main.go b/cmd/crossbar/main.go index ce11cbb..3360531 100644 --- a/cmd/crossbar/main.go +++ b/cmd/crossbar/main.go @@ -17,10 +17,12 @@ import ( "git.wntrmute.dev/kyle/crossbar/internal/admin" "git.wntrmute.dev/kyle/crossbar/internal/config" "git.wntrmute.dev/kyle/crossbar/internal/health" + "git.wntrmute.dev/kyle/crossbar/internal/identity" "git.wntrmute.dev/kyle/crossbar/internal/lease" "git.wntrmute.dev/kyle/crossbar/internal/limiter" "git.wntrmute.dev/kyle/crossbar/internal/proxy" "git.wntrmute.dev/kyle/crossbar/internal/store" + "git.wntrmute.dev/kyle/crossbar/internal/wake" ) func main() { @@ -59,6 +61,18 @@ func run() error { hosts := proxy.HostView(table, cfg) lim := limiter.New() + + // Wake: rouse a sleeping host when a route has no healthy host left. Built + // from every host that carries a wake target; the health table satisfies the + // waker's Health interface. + targets := make(map[string]wake.Target, len(cfg.Hosts)) + for name, h := range cfg.Hosts { + if h.Wake == nil { + continue + } + targets[name] = wake.Target{MAC: h.Wake.MAC, Broadcast: h.Wake.Broadcast, Wait: h.Wake.Wait.Duration} + } + waker := wake.New(targets, hosts) for name, h := range cfg.Hosts { for model, m := range h.Models { lim.Configure(name, model, m.Parallel, cfg.QueueMax) @@ -73,9 +87,31 @@ func run() error { leases.Candidates(name, rt.Hosts) } + // Identity: gate the proxy on the route's peers when a backend is + // configured; off leaves the proxy unwrapped. + logIdentityMode(log, cfg.Identity) + + p := proxy.New(cfg, table, leases, lim, st, log) + p.SetWaker(waker) + + var handler http.Handler = p + if cfg.Identity != "off" { + var checker *identity.Checker + switch cfg.Identity { + case "tailscale": + checker = identity.NewChecker(identity.TailscaleResolver{}) + default: // "header" + checker = identity.NewHeaderChecker() + } + handler = identity.Middleware(checker, func(route string) ([]string, bool) { + rt, ok := cfg.Routes[route] + return rt.Peers, ok + }, p) + } + mux := http.NewServeMux() mux.Handle("/_crossbar/", admin.Handler(cfg, table, leases, lim, st, hosts)) - mux.Handle("/", proxy.New(cfg, table, leases, lim, st, log)) + mux.Handle("/", handler) // Background maintenance until ctx is done. Errors are logged, never fatal. go func() { @@ -156,3 +192,16 @@ func run() error { return err } } + +// logIdentityMode logs which identity backend is active and, for the unauthenticated header +// backend used by the smoke run, warns that it must not be exposed. +func logIdentityMode(log *slog.Logger, mode string) { + if mode == "off" { + log.Info("identity", "mode", "off") + return + } + log.Info("identity", "mode", mode) + if mode == "header" { + log.Warn("identity header mode is not authenticated; do not expose it") + } +} diff --git a/cmd/fakeupstream/main.go b/cmd/fakeupstream/main.go index c222d38..0cb6b44 100644 --- a/cmd/fakeupstream/main.go +++ b/cmd/fakeupstream/main.go @@ -7,7 +7,9 @@ // SSE chunks 200 ms apart when the body has "stream": true, then a final chunk carrying // "usage" and llama-server style "timings", then [DONE]; one JSON answer with usage and // timings otherwise. -slow adds that many milliseconds before answering (for queue tests). -// Every response carries X-Upstream: . +// Every response carries X-Upstream: . /props reports -n-ctx and -slots. With -wol-listen, +// a valid wake-on-LAN magic packet for -wol-mac received on that UDP address removes the down +// file, so the fake "boots" when woken. package main import ( @@ -16,6 +18,7 @@ import ( "fmt" "io" "log" + "net" "net/http" "os" "strings" @@ -28,7 +31,14 @@ func main() { models := flag.String("models", "m", "comma-separated model ids for /v1/models") downFile := flag.String("down-file", "", "while this file exists, /health answers 503") slow := flag.Int("slow", 0, "milliseconds to wait before answering a completion") + nCtx := flag.Int("n-ctx", 8192, "n_ctx reported by /props") + slots := flag.Int("slots", 2, "total_slots reported by /props") + wolListen := flag.String("wol-listen", "", "UDP address to listen on for a wake-on-LAN magic packet") + wolMAC := flag.String("wol-mac", "aa:bb:cc:dd:ee:01", "MAC the magic packet must carry") flag.Parse() + if *wolListen != "" && *downFile != "" { + go wakeOnPacket(*wolListen, *wolMAC, *downFile) + } ids := strings.Split(*models, ",") mux := http.NewServeMux() @@ -56,7 +66,7 @@ func main() { }) mux.HandleFunc("/props", func(w http.ResponseWriter, r *http.Request) { stamp(w) - writeJSON(w, map[string]any{"default_generation_settings": map[string]any{"n_ctx": 8192}, "total_slots": 2, "model_path": *name}) + writeJSON(w, map[string]any{"default_generation_settings": map[string]any{"n_ctx": *nCtx}, "total_slots": *slots, "model_path": *name}) }) mux.HandleFunc("/v1/chat/completions", func(w http.ResponseWriter, r *http.Request) { stamp(w) @@ -113,3 +123,39 @@ func writeJSON(w http.ResponseWriter, v any) { w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(v) } + +// wakeOnPacket removes downFile when a magic packet for mac arrives: 6×0xff then the MAC 16 times. +func wakeOnPacket(addr, mac, downFile string) { + hw, err := net.ParseMAC(mac) + if err != nil { + log.Fatalf("wol-mac: %v", err) + } + pc, err := net.ListenPacket("udp4", addr) + if err != nil { + log.Fatalf("wol-listen: %v", err) + } + log.Printf("fakeupstream listening for wake-on-LAN on %s (mac %s)", addr, hw) + buf := make([]byte, 256) + for { + n, _, err := pc.ReadFrom(buf) + if err != nil { + return + } + if n != 102 { + continue + } + ok := true + for i := 0; i < 6; i++ { + ok = ok && buf[i] == 0xff + } + for i := 0; i < 16 && ok; i++ { + for j := 0; j < 6; j++ { + ok = ok && buf[6+6*i+j] == hw[j] + } + } + if ok { + log.Printf("magic packet received: waking (removing %s)", downFile) + _ = os.Remove(downFile) + } + } +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index d354ed7..31ab42e 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -5,6 +5,12 @@ owner fills in the Model column. The reviewer adds findings under "Reviews" once | Task | Date | Status | Gate runs | First gate | Deviations | Notes | Model | |---|---|---|---|---|---|---|---| +| v2/05-wiring-smoke | 2026-09-25 | done | 1 | pass | none | The wiring in `cmd/crossbar/main.go` and `internal/proxy/{proxy,forward,ctxguard}.go` plus the README section were already in the working tree from a prior session; this session only ran the tests, the gate, the log row, and the commit. `go test -race -count=1 ./...` failed once on `TestQueueFullIs503` (`Errors:2`, the 503 not recorded) — the known v1 recording defect the owner scheduled as a v2.1 task 01; reran once and it passed. `make gate` printed `gate: ok` on the first run. Committed the two owner-corrected given v1 tests (`internal/limiter/limiter_test.go`, `internal/proxy/proxy_test.go`) alongside the prior session's changes. | ? | +| v2/04-identity | 2026-09-25 | done | 1 | pass | new file `internal/config/identity.go` | Implemented `internal/identity/identity.go`: `ParseWhois` (Node = ComputedName, else Name minus trailing dot/domain; empty node errors), `TailscaleResolver` (`tailscale whois --json`, 3 s timeout, non-zero exit → `ErrNotAPeer`, missing binary a real deny), `Checker` with a 5-min per-address cache that also caches `ErrNotAPeer`, and `NewHeaderChecker`/`WithHeaderPeer` that read the peer from a context value. `middleware.go` names the route like the proxy (X-Crossbar-Route header, else first path segment), passes `/_crossbar/` and unknown routes straight through, and answers 403 `{"error":"forbidden route"}`. Config gains `Identity`/`Wake`/`Peers`; validation keys the peers check on the *explicit* identity value (a config with peers but no identity key passes), and `wake.wait` defaults to 45 s. Copied all four given files byte-identical; `go test -race ./internal/identity/ ./internal/config/` and `make gate` printed `gate: ok` on the first run. | ? | +| v2/03-wake | 2026-09-25 | done | 1 | pass | none | Implemented wake-on-LAN in new `internal/wake/wake.go`: `MagicPacket` builds the 102-byte frame via `net.ParseMAC` (six `0xff` bytes plus the MAC repeated sixteen times) and rejects bad MACs; `Send` emits one UDP4 datagram to the resolved broadcast address, returning parse/resolve/write errors; `Waker` tracks last-sent per host under a mutex and sends at most once per `Wait` window, polling health every second (`PollEvery` is a test hook) until healthy, on `Wait` timeout, or on ctx cancellation, returning false for an unknown host without sending. Copied `internal/wake/wake_test.go` byte-identical to `docs/plans/v2/_files/`; `go test -race -count=3 ./internal/wake/` ok and `make gate` printed `gate: ok` on the first run. | llama.cpp/ornith-1.5-35b-a3b | +| v2/02-ctxguard | 2026-09-25 | done | 1 | pass | none | Implemented the context-size guard in new `internal/proxy/ctxguard.go` (estimate `int(float64(len(body))/4*1.2)`; rule 2 skip on unknown/fit; rule 3 move via `leases.Move` with a `moved:>` header; rule 4 400 with `{"error":"prompt too large","estimate":E,"max":M}` and a status-400 accounting row, no forward, no mark-down) and wired it into `ServeHTTP` between the lease and the slot; added `Move` to `internal/lease/lease.go` (re-leases, deletes the old row, records a `ctx` event) and `ReasonCtx = "ctx"` to `internal/store`. Copied `internal/proxy/ctxguard_test.go` byte-identical to `docs/plans/v2/_files/`; `go test -race -count=2 ./internal/proxy/ ./internal/lease/` ok and `make gate` printed `gate: ok` on the first run. | llama.cpp/ornith-1.5-35b-a3b | +| v2/01-props | 2026-09-25 | done | 1 | pass | none | Implemented /props learning in `internal/health/health.go`: added `Status.NCtx`/`Status.Slots`, `PerSlotCtx()`, and a best-effort `GET /props` appended to the poll after `/v1/models`, setting NCtx/Slots to 0 (negative → 0) on any failure without counting the poll as failed; exposed them in `internal/admin/admin.go` `HostView`. Copied `internal/health/props_test.go` and the replacement `internal/proxy/helpers_test.go` byte-identical to `docs/plans/v2/_files/`. `go test -race ./...` and `make gate` pass on the first run. | ? | +| v2/01-props | 2026-09-25 | stopped | 1 | fail | none | Implemented /props learning in `internal/health/health.go` (added `Status.NCtx`/`Status.Slots`, `PerSlotCtx`, and a best-effort `GET /props` appended to the poll; 0/unknown on any failure without failing the poll) and exposed them in `internal/admin/admin.go` `HostView`; copied `internal/health/props_test.go` byte-identical to `docs/plans/v2/_files/`. `go test -race ./internal/health/ ./internal/admin/` ok. `make gate` fails on two GIVEN v1 proxy tests — `TestConversationIsStickyAndLeaseHeaderTellsWhy` (alpha 1/beta 7, want 0/6) and `TestDifferentConversationsSpreadByFreeSlots` (beta 3/alpha 2, want 2/1) — which assert exact upstream hit counts; the task-required `/props` poll now lands on that scaffold's `/` catch-all and bumps the counter by exactly 1 per host (deterministic, confirmed over 3 repeated runs, not a flake). `internal/proxy/helpers_test.go` is byte-identical to `docs/plans/v1/_files/` (protected) and cannot be updated here; the `/props` request is unavoidable per the task, so the owner must hand over a scaffold that registers `/props` without counting it as a hit. Code left uncommitted for review. | ? | | v1.1/01-review-fixes | 2026-09-25 | done | 1 | pass | none | Copied `cancel_test.go` and `usage_empty_test.go` byte-identical from `docs/plans/v1.1/_files/`; the earlier session's fixes in `internal/proxy/proxy.go`, `internal/proxy/forward.go` and `internal/admin/admin_ops.go` were already in the working tree. `make gate` printed `gate: ok` on the first run. | llama.cpp/ornith-1.5-35b-a3b | | v1/08-smoke-readme | 2026-09-25 | done | 1 | pass | owner-directed fix to `Free` in `proxy.Chooser` | Changed `Free` from `c.lim.FreeSlots(host)` (sum over every model) to per-model free slots, `freeForModel(cfg.Hosts[host], model, c.lim.InFlight(host, model))`, floored at 0 and 0 when the host does not list the model (new helper in hosts.go); the one code change the task directs. `go test -race ./internal/proxy/` and `make gate` pass on the first run; `make smoke` → `smoke: ok (stream spread 1006 ms)`. README intro, `## Configure` (added db/lease_idle/retention, rewrote queue_max and hosts..hosts) and `## Inspect`→`## Operate` (all six endpoints, examples taken from the smoke run) updated. | llama.cpp/ornith-1.5-35b-a3b | | v1/07-main | 2026-09-25 | done | 1 | pass | none | Wired store, limiter and lease table into `cmd/crossbar/main.go`: `store.Open` before the health table, `limiter.Configure` per (host, model) from `cfg.Hosts`, `lease.New` with `proxy.Chooser`, `Candidates` for every route, three background goroutines (idle expiry per minute, prune per hour logging the count, host-health recording per `poll_interval`), and `st.Close` via `defer`. The 3s SIGTERM run exits 0 with `listening`/`shutting down`; the missing-config run exits 1. | llama.cpp/ornith-1.5-35b-a3b | diff --git a/example.toml b/example.toml index f780b12..03f0372 100644 --- a/example.toml +++ b/example.toml @@ -5,6 +5,7 @@ poll_interval = "1s" # 60s in production; 1s makes the smoke run lease_idle = "30m" # a conversation idle this long loses its host retention = "180d" # per-request rows older than this are rolled up daily queue_max = 1 # waiting places per (host, model) beyond `parallel`; 503 past that +identity = "off" # "tailscale" gates routes with `peers` by `tailscale whois`; "header" trusts X-Crossbar-Peer (TEST ONLY) [hosts.alpha] base_url = "http://127.0.0.1:18081" # e.g. http://straylight.:11434 @@ -15,6 +16,10 @@ models = { "ornith-1.5-35b-a3b" = { parallel = 1 }, "small-9b" = { parallel = base_url = "http://127.0.0.1:18082" # e.g. http://titan.:8081 weight = 2.0 models = { "ornith-1.5-35b-a3b" = { parallel = 2 } } +[hosts.beta.wake] # v2: wake a sleeping host when nothing else can take a new lease +mac = "aa:bb:cc:dd:ee:02" +broadcast = "127.0.0.1:19082" # the LAN broadcast address, port 9, in production +wait = "20s" # v1: a route is a set of candidate hosts; each conversation gets a sticky lease on the host with # the most free slots × weight at the time it starts. Pins and drains come from the admin API. @@ -24,3 +29,4 @@ default_model = "ornith-1.5-35b-a3b" [routes.hermes-x] hosts = ["beta", "alpha"] +# peers = ["talos"] # v2: with identity = "tailscale", only these tailnet nodes may use the route diff --git a/internal/admin/admin.go b/internal/admin/admin.go index e45817c..d3dffab 100644 --- a/internal/admin/admin.go +++ b/internal/admin/admin.go @@ -38,6 +38,8 @@ type HostView struct { InFlight int `json:"in_flight"` // sum over the host's configured models Queued int `json:"queued"` // same Draining bool `json:"draining"` + NCtx int `json:"n_ctx"` // from /props; 0 = unknown + Slots int `json:"slots"` // from /props; 0 = unknown } // LeaseView is one lease's row in a route's leases. @@ -125,6 +127,8 @@ func (hx *handler) hostView(name string, s health.Status) HostView { InFlight: inflight, Queued: queued, Draining: hx.d.Draining(name), + NCtx: s.NCtx, + Slots: s.Slots, } } diff --git a/internal/config/config.go b/internal/config/config.go index 984624e..51988c4 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -68,12 +68,15 @@ type Host struct { BaseURL string `toml:"base_url"` Weight float64 `toml:"weight"` Models map[string]Model `toml:"models"` + Wake *Wake `toml:"wake"` } -// Route is an ordered list of hosts to try, with an optional default model. +// Route is an ordered list of hosts to try, with an optional default model and +// the peers allowed to reach it. type Route struct { Hosts []string `toml:"hosts"` DefaultModel string `toml:"default_model"` + Peers []string `toml:"peers"` } // Config is the whole file: what to listen on, tuning, hosts and routes. @@ -84,6 +87,7 @@ type Config struct { DB string `toml:"db"` LeaseIdle Duration `toml:"lease_idle"` Retention Duration `toml:"retention"` + Identity string `toml:"identity"` Hosts map[string]Host `toml:"hosts"` Routes map[string]Route `toml:"routes"` } @@ -109,6 +113,8 @@ const ( MinLeaseIdle = time.Minute MinRetention = 24 * time.Hour + + DefaultIdentity = "off" ) var routeName = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*$`) @@ -157,7 +163,10 @@ func Parse(r io.Reader) (*Config, error) { if c.QueueMax == 0 { c.QueueMax = DefaultQueueMax } - if e := c.validate(); e != nil { + if c.Identity == "" { + c.Identity = DefaultIdentity + } + if e := c.validate(md); e != nil { return nil, e } return &c, nil @@ -184,7 +193,14 @@ func IsError(err error) (*Error, bool) { // validate checks the config in a fixed order and writes defaults back into c. // The first problem wins; every problem is an *Error with a precise field. -func (c *Config) validate() *Error { +func (c *Config) validate(md toml.MetaData) *Error { + peersDefined := make(map[string]bool, len(c.Routes)) + for name := range c.Routes { + if md.IsDefined("routes", name, "peers") { + peersDefined[name] = true + } + } + identityDefined := md.IsDefined("identity") if e := c.checkListen(); e != nil { return e } @@ -206,7 +222,13 @@ func (c *Config) validate() *Error { if e := c.checkHosts(); e != nil { return e } - return c.checkRoutes() + if e := c.checkWake(); e != nil { + return e + } + if e := c.checkRoutes(peersDefined, identityDefined); e != nil { + return e + } + return c.checkIdentity() } func (c *Config) checkListen() *Error { @@ -324,7 +346,7 @@ func (c *Config) checkHosts() *Error { return nil } -func (c *Config) checkRoutes() *Error { +func (c *Config) checkRoutes(peersDefined map[string]bool, identityDefined bool) *Error { if len(c.Routes) == 0 { return &Error{Field: "routes", Msg: "at least one required"} } @@ -367,6 +389,10 @@ func (c *Config) checkRoutes() *Error { return &Error{Field: fmt.Sprintf("routes.%s.default_model", name), Msg: "not served by any host in route"} } } + + if e := checkPeers(name, r.Peers, peersDefined[name], identityDefined, c.Identity); e != nil { + return e + } } return nil } diff --git a/internal/config/config_v2_test.go b/internal/config/config_v2_test.go new file mode 100644 index 0000000..0c4af39 --- /dev/null +++ b/internal/config/config_v2_test.go @@ -0,0 +1,75 @@ +package config_test + +import ( + "strings" + "testing" + "time" + + "git.wntrmute.dev/kyle/crossbar/internal/config" +) + +const v2Base = ` +listen = "127.0.0.1:1" +[hosts.a] +base_url = "http://a:1" +models = { "m" = { } } +[hosts.b] +base_url = "http://b:1" +models = { "m" = { } } +[hosts.b.wake] +mac = "aa:bb:cc:dd:ee:ff" +broadcast = "192.168.1.255:9" +wait = "45s" +[routes.r] +hosts = ["a", "b"] +peers = ["talos", "imladris"] +` + +func TestV2Defaults(t *testing.T) { + c, err := config.Parse(strings.NewReader(v2Base)) + if err != nil { + t.Fatal(err) + } + if c.Identity != "off" { + t.Errorf("identity default = %q, want off", c.Identity) + } + if c.Hosts["a"].Wake != nil { + t.Errorf("host without [wake] must have nil Wake") + } + w := c.Hosts["b"].Wake + if w == nil || w.MAC != "aa:bb:cc:dd:ee:ff" || w.Broadcast != "192.168.1.255:9" || w.Wait.Duration != 45*time.Second { + t.Errorf("wake = %+v", w) + } + if p := c.Routes["r"].Peers; len(p) != 2 || p[0] != "talos" { + t.Errorf("peers = %v", p) + } +} + +func TestV2Validation(t *testing.T) { + good := v2Base + for _, tc := range []struct{ name, text, field string }{ + {"bad identity", "identity = \"maybe\"\n" + good, "identity"}, + {"peers without identity", "identity = \"off\"\n" + good, "routes.r.peers"}, + {"bad mac", strings.Replace(good, `mac = "aa:bb:cc:dd:ee:ff"`, `mac = "nope"`, 1), "hosts.b.wake.mac"}, + {"no broadcast", strings.Replace(good, `broadcast = "192.168.1.255:9"`, `broadcast = ""`, 1), "hosts.b.wake.broadcast"}, + {"wait too short", strings.Replace(good, `wait = "45s"`, `wait = "2s"`, 1), "hosts.b.wake.wait"}, + {"peers on unknown route field", "identity = \"tailscale\"\n" + strings.Replace(good, `peers = ["talos", "imladris"]`, `peers = []`, 1), "routes.r.peers"}, + } { + _, err := config.Parse(strings.NewReader(tc.text)) + e, ok := config.IsError(err) + if !ok || e.Field != tc.field { + t.Errorf("%s: %v, want *Error on %s", tc.name, err, tc.field) + } + } + // identity = "header" is the test/smoke mode; "tailscale" the real one; both accept peers. + for _, mode := range []string{"header", "tailscale"} { + if _, err := config.Parse(strings.NewReader("identity = \"" + mode + "\"\n" + good)); err != nil { + t.Errorf("identity=%s with peers: %v", mode, err) + } + } + // wait defaults to 45s when the [wake] table omits it + c, err := config.Parse(strings.NewReader("identity = \"header\"\n" + strings.Replace(good, "wait = \"45s\"\n", "", 1))) + if err != nil || c.Hosts["b"].Wake == nil || c.Hosts["b"].Wake.Wait.Duration != 45*time.Second { + t.Errorf("wake.wait default: %v %+v", err, c.Hosts["b"].Wake) + } +} diff --git a/internal/config/identity.go b/internal/config/identity.go new file mode 100644 index 0000000..a83ef76 --- /dev/null +++ b/internal/config/identity.go @@ -0,0 +1,77 @@ +package config + +import ( + "fmt" + "net" + "time" +) + +// Wake is the magic-wake pattern sent to a host to rouse it: its MAC, the +// broadcast address to aim at, and how long to wait for the answer. +type Wake struct { + MAC string `toml:"mac"` + Broadcast string `toml:"broadcast"` + Wait Duration `toml:"wait"` +} + +const ( + DefaultWakeWait = 45 * time.Second + MinWakeWait = 5 * time.Second +) + +// identityMode reports whether s is a recognized identity backend. +func identityMode(s string) bool { + return s == "off" || s == "tailscale" || s == "header" +} + +// checkWake validates and defaults the magic-wake pattern of each host that has +// one. +func (c *Config) checkWake() *Error { + for name := range c.Hosts { + h := c.Hosts[name] + w := h.Wake + if w == nil { + continue + } + wakeField := fmt.Sprintf("hosts.%s.wake", name) + + mac, err := net.ParseMAC(w.MAC) + if err != nil || len(mac) != 6 { + return &Error{Field: wakeField + ".mac", Msg: "must be a MAC address"} + } + + if _, _, err := net.SplitHostPort(w.Broadcast); err != nil || w.Broadcast == "" { + return &Error{Field: wakeField + ".broadcast", Msg: "must be a non-empty host:port"} + } + + if w.Wait.Duration == 0 { + w.Wait.Duration = DefaultWakeWait + } else if w.Wait.Duration < MinWakeWait { + return &Error{Field: wakeField + ".wait", Msg: "must be at least 5s"} + } + h.Wake = w + c.Hosts[name] = h + } + return nil +} + +// checkIdentity rejects an unrecognized identity backend. +func (c *Config) checkIdentity() *Error { + if !identityMode(c.Identity) { + return &Error{Field: "identity", Msg: `must be "off", "tailscale", or "header"`} + } + return nil +} + +// checkPeers enforces the peers/identity contract for one route: peers may only +// be set with an identity backend on, and may not be an empty list. +func checkPeers(name string, peers []string, peersDefined, identityDefined bool, identity string) *Error { + peersField := fmt.Sprintf("routes.%s.peers", name) + switch { + case len(peers) > 0 && identityDefined && identity == "off": + return &Error{Field: peersField, Msg: "peers need identity = tailscale or header"} + case len(peers) == 0 && peersDefined && identityDefined && identity != "off": + return &Error{Field: peersField, Msg: "empty peers list"} + } + return nil +} diff --git a/internal/health/health.go b/internal/health/health.go index d483353..171a627 100644 --- a/internal/health/health.go +++ b/internal/health/health.go @@ -28,6 +28,17 @@ type Status struct { LastOK time.Time `json:"last_ok"` // zero if never LastErr string `json:"last_err"` // "" after a good poll Consecutive int `json:"consecutive"` // good polls in a row + NCtx int `json:"n_ctx"` // total context from /props; 0 = unknown + Slots int `json:"slots"` // total_slots from /props; 0 = unknown +} + +// PerSlotCtx is the context one request may use: NCtx divided by Slots, or the +// whole NCtx when Slots is unknown (0). It is 0 when NCtx is unknown. +func (s Status) PerSlotCtx() int { + if s.NCtx == 0 || s.Slots == 0 { + return s.NCtx + } + return s.NCtx / s.Slots } type entry struct { @@ -40,6 +51,8 @@ type pollResult struct { cancelled bool reason string loaded []string + nctx int + slots int } // Table maps a host name to its health status. All methods are safe for concurrent use. @@ -156,6 +169,8 @@ func (t *Table) pollHost(ctx context.Context, name string) { e.status.LastOK = time.Now() e.status.LastErr = "" e.status.Loaded = r.loaded + e.status.NCtx = r.nctx + e.status.Slots = r.slots e.status.Healthy = !e.everFailed || e.status.Consecutive >= RecoveryPolls } else { e.everFailed = true @@ -174,13 +189,64 @@ func (t *Table) poll(ctx context.Context, base string) pollResult { return r } loaded, r = t.check(ctx, base+"/v1/models", "models") - if r.cancelled { + if r.cancelled || r.reason != "" { return r } - if r.reason != "" { + nctx, slots, r := t.props(ctx, base) + if r.cancelled || r.reason != "" { return r } - return pollResult{ok: true, loaded: loaded} + return pollResult{ok: true, loaded: loaded, nctx: nctx, slots: slots} +} + +// props reads /props best-effort. A request that fails because ctx is +// done yields a cancelled result so the caller records nothing; any other +// outcome (status, body, or missing fields) leaves context unknown without +// failing the poll. +func (t *Table) props(ctx context.Context, base string) (int, int, pollResult) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, base+"/props", nil) + if err != nil { + if ctx.Err() != nil { + return 0, 0, pollResult{cancelled: true} + } + return 0, 0, pollResult{} + } + resp, err := t.client.Do(req) + if err != nil { + if ctx.Err() != nil { + return 0, 0, pollResult{cancelled: true} + } + return 0, 0, pollResult{} + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return 0, 0, pollResult{} + } + var p struct { + Generation struct { + NCtx *int `json:"n_ctx"` + } `json:"default_generation_settings"` + TotalSlots *int `json:"total_slots"` + } + if err := json.NewDecoder(io.LimitReader(resp.Body, MaxModelsBody)).Decode(&p); err != nil { + return 0, 0, pollResult{} + } + + nctx, slots := 0, 0 + if p.Generation.NCtx != nil { + nctx = *p.Generation.NCtx + } + if p.TotalSlots != nil { + slots = *p.TotalSlots + } + if nctx < 0 { + nctx = 0 + } + if slots < 0 { + slots = 0 + } + return nctx, slots, pollResult{} } // check performs one GET and, on success, returns the decoded model ids. Health checks use the diff --git a/internal/health/props_test.go b/internal/health/props_test.go new file mode 100644 index 0000000..ed77b95 --- /dev/null +++ b/internal/health/props_test.go @@ -0,0 +1,75 @@ +package health_test + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "git.wntrmute.dev/kyle/crossbar/internal/health" +) + +// propsFake answers /health, /v1/models and a configurable /props. +func propsFake(t *testing.T, props string, status int) *httptest.Server { + mux := http.NewServeMux() + mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, `{"status":"ok"}`) }) + mux.HandleFunc("/v1/models", func(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, `{"data":[{"id":"m"}]}`) }) + mux.HandleFunc("/props", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(status) + fmt.Fprint(w, props) + }) + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + return srv +} + +func TestPropsLearned(t *testing.T) { + srv := propsFake(t, `{"default_generation_settings":{"n_ctx":131072,"params":{}},"total_slots":4,"model_path":"/x/m.gguf","chat_template":"..."}`, 200) + tbl := health.New(map[string]string{"a": srv.URL}, time.Hour, nil) + tbl.PollOnce(context.Background()) + s, _ := tbl.Get("a") + if !s.Healthy || s.NCtx != 131072 || s.Slots != 4 { + t.Fatalf("status = %+v, want healthy with NCtx 131072 and Slots 4", s) + } + if got := s.PerSlotCtx(); got != 32768 { + t.Errorf("PerSlotCtx = %d, want 131072/4", got) + } +} + +func TestPropsAbsentOrBrokenIsNotAFailure(t *testing.T) { + for name, tc := range map[string]struct { + props string + status int + }{ + "404": {`not found`, 404}, + "not json": {``, 200}, + "no fields": {`{"model_path":"/x"}`, 200}, + "zero ctx": {`{"default_generation_settings":{"n_ctx":0},"total_slots":0}`, 200}, + } { + t.Run(name, func(t *testing.T) { + srv := propsFake(t, tc.props, tc.status) + tbl := health.New(map[string]string{"a": srv.URL}, time.Hour, nil) + tbl.PollOnce(context.Background()) + s, _ := tbl.Get("a") + if !s.Healthy { + t.Fatalf("a bad /props must not make the host unhealthy: %+v", s) + } + if s.NCtx != 0 || s.Slots != 0 || s.PerSlotCtx() != 0 { + t.Errorf("unknown context must read as 0: %+v", s) + } + }) + } +} + +func TestPerSlotCtxWithUnknownSlots(t *testing.T) { + s := health.Status{NCtx: 8192, Slots: 0} + if s.PerSlotCtx() != 8192 { + t.Errorf("with Slots unknown the whole context is the per-slot value; got %d", s.PerSlotCtx()) + } + s = health.Status{NCtx: 8192, Slots: 3} + if s.PerSlotCtx() != 2730 { + t.Errorf("integer division: got %d, want 2730", s.PerSlotCtx()) + } +} diff --git a/internal/identity/identity.go b/internal/identity/identity.go new file mode 100644 index 0000000..a6d3fdf --- /dev/null +++ b/internal/identity/identity.go @@ -0,0 +1,223 @@ +// Package identity resolves a caller's address to a tailnet node name, and gates a +// route on the set of peers it allows. In production the resolver asks +// `tailscale whois`; in the smoke run it trusts a request header. A Checker caches +// the answer per address so a hot peer does not re-query whois on every request. +package identity + +import ( + "context" + "encoding/json" + "errors" + "net" + "os/exec" + "strings" + "sync" + "time" +) + +var ( + // ErrNotAPeer is returned by a resolver when the address is not a known + // tailnet node. The Checker turns it into a deny. + ErrNotAPeer = errors.New("identity: not a tailnet peer") + // ErrForbidden is returned by the Checker when the caller is not on the + // route's allow list. + ErrForbidden = errors.New("identity: forbidden route") +) + +// ID is the tailnet identity of a caller. +type ID struct { + Node, Login string +} + +// Resolver maps an IP address to the tailnet node it belongs to. +type Resolver interface { + Identity(ctx context.Context, ip string) (ID, error) +} + +// cacheTTL is how long a resolved identity (or a rejection) is held per address. +const cacheTTL = 5 * time.Minute + +// ParseWhois decodes `tailscale whois --json` output. Node is ComputedName, or +// Name with its trailing dot and domain stripped; Login is the profile login +// name. An empty node name is an error. +func ParseWhois(raw []byte) (ID, error) { + var whois struct { + Node struct { + Name string `json:"Name"` + ComputedName string `json:"ComputedName"` + } `json:"Node"` + UserProfile struct { + LoginName string `json:"LoginName"` + } `json:"UserProfile"` + } + if err := json.Unmarshal(raw, &whois); err != nil { + return ID{}, err + } + node := whois.Node.ComputedName + if node == "" { + node = stripName(whois.Node.Name) + } + if node == "" { + return ID{}, errors.New("identity: whois has no node name") + } + return ID{Node: node, Login: whois.UserProfile.LoginName}, nil +} + +// stripName takes a whois Name such as "titan.example.ts.net." and returns the +// first label, "titan". +func stripName(name string) string { + name = strings.TrimSuffix(name, ".") + if i := strings.IndexByte(name, '.'); i >= 0 { + name = name[:i] + } + return name +} + +// TailscaleResolver runs `tailscale whois --json ` and parses it. A non-zero +// exit is ErrNotAPeer; a missing binary (or other transport failure) is a real +// error the Checker treats as a deny. +type TailscaleResolver struct{ Bin string } + +// Identity runs the whois lookup with a 3 s timeout. +func (t TailscaleResolver) Identity(ctx context.Context, ip string) (ID, error) { + bin := t.Bin + if bin == "" { + bin = "tailscale" + } + ctx, cancel := context.WithTimeout(ctx, 3*time.Second) + defer cancel() + out, err := exec.CommandContext(ctx, bin, "whois", "--json", ip).Output() + if err != nil { + var exitErr *exec.ExitError + if errors.As(err, &exitErr) { + return ID{}, ErrNotAPeer + } + return ID{}, err + } + return ParseWhois(out) +} + +// entry is a cached result, whether a node name or a rejection. +type entry struct { + id ID + err error + at time.Time +} + +// Checker resolves addresses through a Resolver, caching per address. In header +// mode it skips the cache and lets the resolver read the peer from the request. +type Checker struct { + r Resolver + header bool + + mu sync.Mutex + cache map[string]entry +} + +// NewChecker builds a Checker that resolves through r. +func NewChecker(r Resolver) *Checker { + return &Checker{r: r, cache: make(map[string]entry)} +} + +// NewHeaderChecker builds a Checker that trusts the X-Crossbar-Peer request +// header as the node name. TEST/SMOKE ONLY. +func NewHeaderChecker() *Checker { + return &Checker{r: headerResolver{}, header: true, cache: make(map[string]entry)} +} + +// WithHeaderPeer returns a context carrying the peer name the header checker +// reads. Middleware sets it from the X-Crossbar-Peer request header. +func WithHeaderPeer(ctx context.Context, peer string) context.Context { + return context.WithValue(ctx, headerPeerKey{}, peer) +} + +// Allow reports whether the caller at remoteAddr may use a route limited to peers. +// An empty peers list is an open route; otherwise the caller's node must be in +// peers. Any resolver error, an unparsable address, or a loopback address denies. +func (c *Checker) Allow(ctx context.Context, peers []string, remoteAddr string) error { + if len(peers) == 0 { + return nil + } + ip := "" + if !c.header { + var ok bool + ip, ok = peerIP(remoteAddr) + if !ok || net.ParseIP(ip).IsLoopback() { + return ErrForbidden + } + } + node, err := c.resolve(ctx, ip) + if err != nil { + return ErrForbidden + } + for _, p := range peers { + if p == node { + return nil + } + } + return ErrForbidden +} + +// resolve returns the node for ip, using the cache unless in header mode. +func (c *Checker) resolve(ctx context.Context, ip string) (string, error) { + if c.header { + id, err := c.r.Identity(ctx, ip) + if err != nil { + return "", err + } + return id.Node, nil + } + + c.mu.Lock() + e, hit := c.cache[ip] + if hit && time.Since(e.at) < cacheTTL { + c.mu.Unlock() + if e.err != nil { + return "", e.err + } + return e.id.Node, nil + } + c.mu.Unlock() + + id, err := c.r.Identity(ctx, ip) + if err != nil { + c.mu.Lock() + c.cache[ip] = entry{err: err, at: time.Now()} + c.mu.Unlock() + return "", err + } + + c.mu.Lock() + c.cache[ip] = entry{id: id, at: time.Now()} + c.mu.Unlock() + return id.Node, nil +} + +// headerPeerKey is the context key under which Middleware stores the X-Crossbar-Peer +// value for the header checker to read. +type headerPeerKey struct{} + +// headerResolver answers from the peer name Middleware placed on the context. An +// absent or empty header is a deny, so a request that forgot the header is 403. +type headerResolver struct{} + +func (headerResolver) Identity(ctx context.Context, _ string) (ID, error) { + peer, ok := ctx.Value(headerPeerKey{}).(string) + if !ok || peer == "" { + return ID{}, ErrNotAPeer + } + return ID{Node: peer, Login: peer}, nil +} + +// peerIP splits the host from a "host:port" address, returning the bare IP. +func peerIP(remoteAddr string) (string, bool) { + host := remoteAddr + if h, _, err := net.SplitHostPort(remoteAddr); err == nil { + host = h + } + ip := net.ParseIP(host) + if ip == nil { + return "", false + } + return ip.String(), true +} diff --git a/internal/identity/identity_test.go b/internal/identity/identity_test.go new file mode 100644 index 0000000..b75e2de --- /dev/null +++ b/internal/identity/identity_test.go @@ -0,0 +1,90 @@ +package identity_test + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" + + "git.wntrmute.dev/kyle/crossbar/internal/identity" +) + +func TestParseWhois(t *testing.T) { + raw, err := os.ReadFile(filepath.Join("testdata", "whois.json")) + if err != nil { + t.Fatal(err) + } + id, err := identity.ParseWhois(raw) + if err != nil { + t.Fatal(err) + } + if id.Node != "titan" || id.Login == "" { + t.Errorf("parsed %+v, want Node titan and a login", id) + } + if _, err := identity.ParseWhois([]byte(`{"Node":{}}`)); err == nil { + t.Error("a whois answer without a node name must be an error") + } + if _, err := identity.ParseWhois([]byte(`nope`)); err == nil { + t.Error("non-JSON must be an error") + } +} + +// fakeResolver answers from a map; "" means not a tailnet peer. +type fakeResolver map[string]string + +func (f fakeResolver) Identity(ctx context.Context, ip string) (identity.ID, error) { + n, ok := f[ip] + if !ok { + return identity.ID{}, identity.ErrNotAPeer + } + return identity.ID{Node: n, Login: n + "@example"}, nil +} + +func TestChecker(t *testing.T) { + c := identity.NewChecker(fakeResolver{"100.64.0.5": "talos", "100.64.0.9": "titan"}) + for _, tc := range []struct { + name string + peers []string + addr string + want error + }{ + {"open route", nil, "203.0.113.7:1", nil}, + {"allowed peer", []string{"talos", "titan"}, "100.64.0.5:44444", nil}, + {"other peer", []string{"talos"}, "100.64.0.9:1", identity.ErrForbidden}, + {"not a peer", []string{"talos"}, "203.0.113.7:1", identity.ErrForbidden}, + {"loopback", []string{"talos"}, "127.0.0.1:1", identity.ErrForbidden}, + {"garbage addr", []string{"talos"}, "nonsense", identity.ErrForbidden}, + } { + t.Run(tc.name, func(t *testing.T) { + got := c.Allow(context.Background(), tc.peers, tc.addr) + if !errors.Is(got, tc.want) && !(got == nil && tc.want == nil) { + t.Errorf("Allow(%v, %q) = %v, want %v", tc.peers, tc.addr, got, tc.want) + } + }) + } +} + +func TestCheckerCachesPerAddress(t *testing.T) { + calls := 0 + r := countingResolver{f: fakeResolver{"100.64.0.5": "talos"}, calls: &calls} + c := identity.NewChecker(r) + for i := 0; i < 5; i++ { + if err := c.Allow(context.Background(), []string{"talos"}, "100.64.0.5:1"); err != nil { + t.Fatal(err) + } + } + if calls != 1 { + t.Errorf("resolver called %d times for one address, want 1 (cache)", calls) + } +} + +type countingResolver struct { + f fakeResolver + calls *int +} + +func (c countingResolver) Identity(ctx context.Context, ip string) (identity.ID, error) { + *c.calls++ + return c.f.Identity(ctx, ip) +} diff --git a/internal/identity/middleware.go b/internal/identity/middleware.go new file mode 100644 index 0000000..e91b3b8 --- /dev/null +++ b/internal/identity/middleware.go @@ -0,0 +1,71 @@ +// Package identity gates a route on the set of tailnet peers allowed to use it. +// The middleware sits in front of the proxy: it names the route the same way the +// proxy does and refuses with 403 any caller a route does not allow. +package identity + +import ( + "net/http" + "strings" +) + +const ( + // routeHeader is how a caller names the route, the same header the proxy + // reads. + routeHeader = "X-Crossbar-Route" + // peerHeader carries the node name in header mode. + peerHeader = "X-Crossbar-Peer" + // adminPrefix is never gated here; the proxy's own handlers own it. + adminPrefix = "/_crossbar/" +) + +// Middleware wraps next with the peer gate. peersFor names the allow list for a +// route and reports whether it knows the route; an unknown route, like a path +// under adminPrefix, passes straight through. +func Middleware(c *Checker, peersFor func(route string) ([]string, bool), next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasPrefix(r.URL.Path, adminPrefix) { + next.ServeHTTP(w, r) + return + } + route := r.Header.Get(routeHeader) + if route == "" { + route = firstSegment(r.URL.Path) + } + peers, known := peersFor(route) + if !known { + next.ServeHTTP(w, r) + return + } + ctx := WithHeaderPeer(r.Context(), r.Header.Get(peerHeader)) + if err := c.Allow(ctx, peers, r.RemoteAddr); err != nil { + writeForbidden(w) + return + } + next.ServeHTTP(w, r) + }) +} + +// writeForbidden answers the JSON 403 the tests and callers expect. +func writeForbidden(w http.ResponseWriter) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusForbidden) + w.Write([]byte(`{"error":"forbidden route"}`)) +} + +// firstSegment takes the first path segment as the route, "/a/v1/x" -> "a". +func firstSegment(path string) string { + if path == "" || path[0] != '/' { + return "" + } + after := path[1:] + if slash := strings.IndexByte(after, '/'); slash >= 0 { + if after[:slash] == "" { + return "" + } + return after[:slash] + } + if after == "" { + return "" + } + return after +} diff --git a/internal/identity/middleware_test.go b/internal/identity/middleware_test.go new file mode 100644 index 0000000..19bf778 --- /dev/null +++ b/internal/identity/middleware_test.go @@ -0,0 +1,74 @@ +package identity_test + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + + "git.wntrmute.dev/kyle/crossbar/internal/identity" +) + +// The middleware sits in front of the proxy: it names the route the same way the proxy does +// (X-Crossbar-Route header, else first path segment) and refuses callers a route does not list. +func TestMiddleware(t *testing.T) { + peers := map[string][]string{"locked": {"talos"}, "open": nil} + inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(204) }) + h := identity.Middleware(identity.NewChecker(fakeResolver{"100.64.0.5": "talos", "100.64.0.9": "titan"}), + func(route string) ([]string, bool) { p, ok := peers[route]; return p, ok }, inner) + for _, tc := range []struct { + name, path, hdr, addr string + want int + }{ + {"open route, anyone", "/open/v1/models", "", "203.0.113.1:5", 204}, + {"locked, right peer", "/locked/v1/models", "", "100.64.0.5:5", 204}, + {"locked, wrong peer", "/locked/v1/models", "", "100.64.0.9:5", 403}, + {"locked, not a peer", "/locked/v1/models", "", "203.0.113.1:5", 403}, + {"locked via header", "/v1/models", "locked", "100.64.0.9:5", 403}, + {"header wins over path", "/open/v1/models", "locked", "203.0.113.1:5", 403}, + {"unknown route passes through to the proxy's own 404", "/nope/v1/models", "", "203.0.113.1:5", 204}, + {"admin path is never gated here", "/_crossbar/hosts", "", "203.0.113.1:5", 204}, + } { + t.Run(tc.name, func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, tc.path, nil) + req.RemoteAddr = tc.addr + if tc.hdr != "" { + req.Header.Set("X-Crossbar-Route", tc.hdr) + } + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != tc.want { + t.Errorf("%s = %d, want %d (%s)", tc.path, rec.Code, tc.want, rec.Body.String()) + } + if rec.Code == 403 && (!strings.HasPrefix(rec.Header().Get("Content-Type"), "application/json") || !strings.Contains(rec.Body.String(), `"forbidden route"`)) { + t.Errorf("403 must be JSON {\"error\":\"forbidden route\"}: %q", rec.Body.String()) + } + }) + } +} + +// HeaderResolver is the test/smoke identity source: it trusts X-Crossbar-Peer. It exists so the +// smoke run can exercise the gate without a tailnet; config must call it out as insecure. +func TestHeaderResolver(t *testing.T) { + inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(204) }) + h := identity.Middleware(identity.NewHeaderChecker(), func(route string) ([]string, bool) { return []string{"talos"}, true }, inner) + req := httptest.NewRequest(http.MethodGet, "/r/v1/models", nil) + req.Header.Set("X-Crossbar-Peer", "talos") + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != 204 { + t.Errorf("header peer talos: %d", rec.Code) + } + req.Header.Set("X-Crossbar-Peer", "titan") + rec = httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != 403 { + t.Errorf("header peer titan: %d, want 403", rec.Code) + } + req.Header.Del("X-Crossbar-Peer") + rec = httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != 403 { + t.Errorf("no header: %d, want 403", rec.Code) + } +} diff --git a/internal/identity/testdata/whois.json b/internal/identity/testdata/whois.json new file mode 100644 index 0000000..877b5a8 --- /dev/null +++ b/internal/identity/testdata/whois.json @@ -0,0 +1,24 @@ +{ + "Node": { + "ID": 1, + "StableID": "nEXAMPLE", + "Name": "titan.example.ts.net.", + "User": 2, + "Addresses": [ + "100.64.0.9/32", + "fd7a:115c:a1e0::9/128" + ], + "HomeDERP": 2, + "Created": "2026-01-01T00:00:00Z", + "Cap": 138, + "Online": true, + "ComputedName": "titan", + "ComputedNameWithHost": "titan" + }, + "UserProfile": { + "ID": 2, + "LoginName": "user@example.com", + "DisplayName": "Example User" + }, + "CapMap": null +} \ No newline at end of file diff --git a/internal/lease/lease.go b/internal/lease/lease.go index aa6bd90..cc48001 100644 --- a/internal/lease/lease.go +++ b/internal/lease/lease.go @@ -194,6 +194,29 @@ func (t *Table) Acquire(k Key, candidates []string, now time.Time) (host string, return host, false, nil } +// Move relocates k to host, dropping any existing lease for k (on any host). It re-leases k onto +// host, deletes the old row, and records a ctx event carrying the old and new hosts. It returns an +// error only from the persister, rolling back the in-memory lease on save failure. +func (t *Table) Move(k Key, host string, now time.Time) error { + t.mu.Lock() + defer t.mu.Unlock() + + var from string + if l, exists := t.leases[k]; exists { + from = l.Host + delete(t.leases, k) + _ = t.p.DeleteLease(k.Route, k.FP, k.Model) + } + l := &Lease{k, host, store.Active, now, now} + t.leases[k] = l + if err := t.save(l); err != nil { + delete(t.leases, k) + return fmt.Errorf("lease: %w", err) + } + t.event(now, k, store.ReasonCtx, from, host) + return nil +} + // Candidates records hosts as seen for route (idempotent), so Pin can accept a host the route // is configured for before any request has used it. cmd/crossbar calls it for every route at // start; the admin handler calls it before Pin. diff --git a/internal/limiter/limiter_test.go b/internal/limiter/limiter_test.go index bf90588..6ab30e2 100644 --- a/internal/limiter/limiter_test.go +++ b/internal/limiter/limiter_test.go @@ -32,10 +32,10 @@ func TestParallelAndQueue(t *testing.T) { go func() { rel, waited, err := l.Acquire(ctx, "alpha", "m") if err == nil { - defer rel() if waited < 40*time.Millisecond { err = errors.New("third acquire did not wait") } + rel() // release before reporting, so the final count check cannot race it } got3 <- err }() diff --git a/internal/proxy/ctxguard.go b/internal/proxy/ctxguard.go new file mode 100644 index 0000000..f364c15 --- /dev/null +++ b/internal/proxy/ctxguard.go @@ -0,0 +1,188 @@ +package proxy + +import ( + "encoding/json" + "net/http" + "time" + + "git.wntrmute.dev/kyle/crossbar/internal/config" + "git.wntrmute.dev/kyle/crossbar/internal/lease" + "git.wntrmute.dev/kyle/crossbar/internal/store" +) + +// movedHeader announces a context-driven move in the response header: +// "moved: old><" + to +} + +// drainer is the drain flag the guard consults when rule 3 prefers a host not +// being taken out of service. *health.Table (used in tests and production) does +// not implement it, so the assertion is a no-op there; a host table with a +// drain set would. +type drainer interface { + Draining(string) bool +} + +// ctxEstimate returns the prompt size the guard reasons about: +// int(float64(len(body))/4*1.2), or 0 for a bodyless request (GET/HEAD). The +// body was already read by peekModel and restored, so its length is known +// without reading again. +func ctxEstimate(r *http.Request) int { + if r.Method == http.MethodGet || r.Method == http.MethodHead { + return 0 + } + if r.Body == nil || r.Body == http.NoBody { + return 0 + } + return int(float64(r.ContentLength) / 4 * 1.2) +} + +// guard runs the context-size guard's rules after a lease and a slot are held. +// If the prompt does not fit the leased host's per-slot context, it moves the +// conversation to a host where it fits (updating the lease) and returns that +// host with a moved header, or answers 400 when no host fits. newHost is the +// host to forward to (== host when nothing moved); done is true when the caller +// must return without forwarding. +func (p *Handler) guard(w http.ResponseWriter, r *http.Request, hosts []string, host, route, model, fp string, started time.Time) (string, string, int, bool) { + estimate := ctxEstimate(r) + if estimate == 0 { + return host, "", 0, false + } + + // Rule 2: no move when the leased host has no context size or the estimate + // already fits its per-slot context. + psc := 0 + if s, ok := p.health.Get(host); ok { + psc = s.PerSlotCtx() + } + if psc == 0 || estimate <= psc { + return host, "", 0, false + } + + // Rule 3: move the conversation to a host where the prompt fits. + if newHost, ok := ctxFitHost(hosts, model, estimate, p.health, p.cfg); ok { + if err := p.leases.Move(lease.Key{Route: route, FP: fp, Model: model}, newHost, time.Now()); err != nil { + p.writeError(w, http.StatusBadGateway, "upstream failed") + return host, "", estimate, true + } + return newHost, movedHeader(host, newHost), estimate, false + } + + // Rule 4a: no host fits. Before refusing, ask a waker to rouse a candidate + // whose context may grow when it comes up; a woken host takes the lease. + if p.waker != nil { + for _, name := range hosts { + s, ok := p.health.Get(name) + if !ok { + continue + } + // A healthy host does not need waking; only a down host might grow + // a larger context when it comes up. + if s.Healthy { + continue + } + // A host with a known per-slot context smaller than the estimate + // cannot serve it no matter how it wakes. + if psc := s.PerSlotCtx(); psc != 0 && psc < estimate { + continue + } + if p.cfg.Hosts[name].Wake == nil || !p.waker.Wake(r.Context(), name) { + continue + } + if err := p.leases.Move(lease.Key{Route: route, FP: fp, Model: model}, name, time.Now()); err != nil { + p.writeError(w, http.StatusBadGateway, "upstream failed") + return host, "", estimate, true + } + return name, movedHeader(host, name), estimate, false + } + } + + // Rule 4: no host fits. Answer 400 with the estimate and the largest + // available per-slot context, and record the row. + p.refuseCtx(w, host, route, model, fp, started, estimate, largestSlotCtx(hosts, p.health)) + return host, "", estimate, true +} + +// ctxFitHost walks the route's ordered candidate hosts for rule 3: the first +// healthy, non-draining host whose per-slot context fits the estimate, +// preferring a host that has the model loaded, else one that can serve it. ok +// is false when none fits. +func ctxFitHost(hosts []string, model string, estimate int, h Health, cfg *config.Config) (string, bool) { + var drn drainer + if d, ok := h.(drainer); ok { + drn = d + } + + // First pass: the model is resident and the per-slot context fits. + for _, name := range hosts { + s, ok := h.Get(name) + if !ok || !s.Healthy || isDraining(drn, name) { + continue + } + if s.PerSlotCtx() >= estimate && contains(s.Loaded, model) { + return name, true + } + } + + // Second pass: the host is configured to serve the model and the per-slot + // context fits. + for _, name := range hosts { + s, ok := h.Get(name) + if !ok || !s.Healthy || isDraining(drn, name) { + continue + } + if s.PerSlotCtx() >= estimate && cfg.Serves(name, model) { + return name, true + } + } + return "", false +} + +// largestSlotCtx is the largest PerSlotCtx across the route's healthy hosts, or +// 0 when none is healthy or reports a context size. +func largestSlotCtx(hosts []string, h Health) int { + best := 0 + for _, name := range hosts { + s, ok := h.Get(name) + if !ok || !s.Healthy { + continue + } + if psc := s.PerSlotCtx(); psc > best { + best = psc + } + } + return best +} + +// refuseCtx answers the 400 the guard's rule 4: the JSON body carries the +// estimate and the largest available per-slot context, plus the error text. It +// records the accounting row (status 400, Err "prompt too large") and never +// marks the host down. +func (p *Handler) refuseCtx(w http.ResponseWriter, host, route, model, fp string, started time.Time, estimate, maxSlot int) { + p.writeRecord(store.Request{ + Route: route, + FP: fp, + Model: model, + Host: host, + Started: started, + TotalMs: time.Since(started).Milliseconds(), + Status: http.StatusBadRequest, + Err: "prompt too large", + }) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": "prompt too large", + "estimate": estimate, + "max": maxSlot, + }) +} + +// isDraining reports whether a drain-capable host table marks name as draining. +func isDraining(d drainer, name string) bool { + if d == nil { + return false + } + return d.Draining(name) +} diff --git a/internal/proxy/ctxguard_test.go b/internal/proxy/ctxguard_test.go new file mode 100644 index 0000000..b56729b --- /dev/null +++ b/internal/proxy/ctxguard_test.go @@ -0,0 +1,148 @@ +package proxy_test + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "git.wntrmute.dev/kyle/crossbar/internal/proxy" +) + +// ctxUpstream is a fake router that reports a context size in /props and echoes completions. +func ctxUpstream(t *testing.T, name string, nCtx, slots int) *upstream { + u := &upstream{name: name} + mux := http.NewServeMux() + mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, `{"status":"ok"}`) }) + mux.HandleFunc("/v1/models", func(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, `{"data":[{"id":"shared"}]}`) }) + mux.HandleFunc("/props", func(w http.ResponseWriter, r *http.Request) { + fmt.Fprintf(w, `{"default_generation_settings":{"n_ctx":%d},"total_slots":%d}`, nCtx, slots) + }) + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + u.hits.Add(1) + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"choices":[{"message":{"role":"assistant","content":"ok"}}],"usage":{"prompt_tokens":1,"completion_tokens":1}}`) + }) + u.srv = httptest.NewServer(mux) + t.Cleanup(u.srv.Close) + return u +} + +const ctxHosts = ` +listen = "127.0.0.1:1" +queue_max = 2 +[hosts.small] +base_url = %q +weight = 10.0 +models = { "shared" = { parallel = 2 } } +[hosts.big] +base_url = %q +weight = 1.0 +models = { "shared" = { parallel = 1 } } +[routes.r] +hosts = ["small", "big"] +default_model = "shared" +` + +// bodyOfTokens builds a chat body whose byte size implies roughly n tokens under the guard's +// estimate (bytes/4 × 1.2): n tokens ≈ 3.33 n bytes ≈ 2n/3 five-byte words. +func bodyOfTokens(n int) string { + text := strings.Repeat("word ", n*2/3) + return fmt.Sprintf(`{"model":"shared","stream":false,"messages":[{"role":"user","content":"%s"}]}`, text) +} + +func TestOversizedPromptMovesToAHostWhereItFits(t *testing.T) { + small := ctxUpstream(t, "small", 8192, 2) // 4096 per slot + big := ctxUpstream(t, "big", 131072, 1) // 131072 per slot + r := newRig(t, ctxHosts, small, big) + // A small prompt starts on `small` (weight 10). + resp := r.post("/r/v1/chat/completions", bodyOfTokens(100)) + drain(resp) + if resp.Header.Get(proxy.HostHeader) != "small" { + t.Fatalf("small prompt went to %q, want small", resp.Header.Get(proxy.HostHeader)) + } + // A new conversation with ~10k tokens does not fit small's 4096-token slot: it must be + // placed on big, with the reason visible in a header. + resp = r.post("/r/v1/chat/completions", bodyOfTokens(10000)) + drain(resp) + if resp.StatusCode != 200 || resp.Header.Get(proxy.HostHeader) != "big" { + t.Fatalf("oversized prompt: %d from %q, want 200 from big", resp.StatusCode, resp.Header.Get(proxy.HostHeader)) + } + if got := resp.Header.Get(proxy.CtxHeader); !strings.HasPrefix(got, "moved") { + t.Errorf("%s = %q, want moved:… ", proxy.CtxHeader, got) + } +} + +func TestOversizedPromptWithNoFitIs400(t *testing.T) { + small := ctxUpstream(t, "small", 8192, 2) + tiny := ctxUpstream(t, "big", 4096, 2) // also too small + r := newRig(t, ctxHosts, small, tiny) + resp := r.post("/r/v1/chat/completions", bodyOfTokens(10000)) + body := drain(resp) + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("status %d body %s, want 400", resp.StatusCode, body) + } + var e map[string]any + if err := json.Unmarshal([]byte(body), &e); err != nil || e["error"] != "prompt too large" { + t.Fatalf("body = %s, want error 'prompt too large'", body) + } + if est, _ := e["estimate"].(float64); est < 8000 || est > 13000 { + t.Errorf("estimate = %v, want roughly 10000 tokens", e["estimate"]) + } + if max, _ := e["max"].(float64); max != 4096 { + t.Errorf("max = %v, want the largest per-slot context among the route's hosts (4096)", e["max"]) + } + if small.hits.Load()+tiny.hits.Load() != 0 { + t.Errorf("a refused prompt must not reach any upstream") + } +} + +func TestUnknownContextNeverBlocks(t *testing.T) { + // /props missing on both hosts: NCtx 0 means "unknown", and the guard must stay out of the way. + alpha, beta := newUpstream(t, "alpha"), newUpstream(t, "beta") + r := newRig(t, twoHosts, alpha, beta) + resp := r.post("/r/v1/chat/completions", bodyOfTokens(50000)) + drain(resp) + if resp.StatusCode != 200 || resp.Header.Get(proxy.CtxHeader) != "" { + t.Errorf("unknown context: %d %q, want 200 and no ctx header", resp.StatusCode, resp.Header.Get(proxy.CtxHeader)) + } +} + +// grow appends later turns to a conversation body without touching its system prompt or first +// user message, so the fingerprint — and therefore the lease — stays the same. +func grow(body string, words int) string { + turn := `,{"role":"assistant","content":"ok"},{"role":"user","content":"` + strings.Repeat("x ", words) + `"}` + return strings.Replace(body, `]}`, turn+`]}`, 1) +} + +func TestStickyLeaseSurvivesGrowthUntilItDoesNotFit(t *testing.T) { + small := ctxUpstream(t, "small", 8192, 2) + big := ctxUpstream(t, "big", 131072, 1) + r := newRig(t, ctxHosts, small, big) + body := bodyOfTokens(100) + resp := r.post("/r/v1/chat/completions", body) + drain(resp) + if resp.Header.Get(proxy.HostHeader) != "small" { + t.Fatal("setup: first turn must be on small") + } + // Same conversation, a later turn well under 4096 tokens: stays. + resp = r.post("/r/v1/chat/completions", grow(body, 500)) + drain(resp) + if resp.Header.Get(proxy.HostHeader) != "small" || resp.Header.Get(proxy.LeaseHeader) != "reused" { + t.Errorf("turn 2: %q %q, want small reused", resp.Header.Get(proxy.HostHeader), resp.Header.Get(proxy.LeaseHeader)) + } + // A turn that outgrows the slot moves the lease — once — and the move is visible in the header. + huge := grow(body, 30000) + resp = r.post("/r/v1/chat/completions", huge) + drain(resp) + if resp.StatusCode != 200 || resp.Header.Get(proxy.HostHeader) != "big" || !strings.HasPrefix(resp.Header.Get(proxy.CtxHeader), "moved") { + t.Fatalf("outgrown turn: %d %q ctx=%q, want 200 from big with a moved header", resp.StatusCode, resp.Header.Get(proxy.HostHeader), resp.Header.Get(proxy.CtxHeader)) + } + resp = r.post("/r/v1/chat/completions", huge) + drain(resp) + if resp.Header.Get(proxy.HostHeader) != "big" || resp.Header.Get(proxy.LeaseHeader) != "reused" { + t.Errorf("after the move the lease is on big: %q %q", resp.Header.Get(proxy.HostHeader), resp.Header.Get(proxy.LeaseHeader)) + } +} diff --git a/internal/proxy/forward.go b/internal/proxy/forward.go index 4f0b4af..3ac16f2 100644 --- a/internal/proxy/forward.go +++ b/internal/proxy/forward.go @@ -14,8 +14,10 @@ import ( ) // forward builds the reverse proxy for one host, tees the response, records the accounting row, and -// logs. leaseState is "new" or "reused"; waited is the time spent in the queue. -func (p *Handler) forward(w http.ResponseWriter, r *http.Request, route, host, leaseState, rest, fp, model string, started time.Time, waited time.Duration) { +// logs. leaseState is "new" or "reused"; waited is the time spent in the queue. ctxEst is the +// prompt size the context guard estimated (0 when the guard did not run); ctxHeader is the +// "moved:…" header to set when the guard relocated the conversation. +func (p *Handler) forward(w http.ResponseWriter, r *http.Request, route, host, leaseState, rest, fp, model string, started time.Time, waited time.Duration, ctxEst int, ctxHeader string) { hostCfg, ok := p.cfg.Hosts[host] if !ok { p.writeError(w, http.StatusBadGateway, "upstream failed") @@ -28,7 +30,7 @@ func (p *Handler) forward(w http.ResponseWriter, r *http.Request, route, host, l } rev := &forwardState{started: started} - rp := newReverseProxy(p.health, host, leaseState, target, rest, rev) + rp := newReverseProxy(p.health, host, leaseState, ctxHeader, target, rest, rev) rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK} // ServeHTTP unwinds with http.ErrAbortHandler when a client leaves mid-stream; recover so the @@ -71,6 +73,7 @@ func (p *Handler) forward(w http.ResponseWriter, r *http.Request, route, host, l "lease", leaseState, "queued_ms", waited.Milliseconds(), "fp", fp8, + "ctx_est", ctxEst, "ms", total.Milliseconds(), ) } @@ -149,6 +152,30 @@ func (p *Handler) writeError(w http.ResponseWriter, status int, msg string) { _ = json.NewEncoder(w).Encode(map[string]string{"error": msg}) } +// writeNoHealthyHost answers the 503 when no candidate woke. The body names the +// hosts that were asked to wake (an empty list, never null), and the row +// records the miss. +func (p *Handler) writeNoHealthyHost(w http.ResponseWriter, route, model, fp string, started time.Time, tried []string) { + if tried == nil { + tried = []string{} + } + p.writeRecord(store.Request{ + Route: route, + FP: fp, + Model: model, + Started: started, + TotalMs: time.Since(started).Milliseconds(), + Status: http.StatusServiceUnavailable, + Err: "no healthy host", + }) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusServiceUnavailable) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": "no healthy host", + "woke": tried, + }) +} + // writeRecord writes one accounting row, logging (never returning) a recorder error. func (p *Handler) writeRecord(req store.Request) { if p.rec == nil { @@ -163,7 +190,7 @@ func (p *Handler) writeRecord(req store.Request) { // original query string. It flushes after every write so long server-sent-event streams are not // buffered, tees the response for usage/timings, and marks the host down on any transport error // other than a client disconnect. -func newReverseProxy(h Health, host, leaseState string, target *url.URL, rest string, rev *forwardState) *httputil.ReverseProxy { +func newReverseProxy(h Health, host, leaseState, ctxHeader string, target *url.URL, rest string, rev *forwardState) *httputil.ReverseProxy { return &httputil.ReverseProxy{ Rewrite: func(pr *httputil.ProxyRequest) { pr.SetURL(target) @@ -176,6 +203,9 @@ func newReverseProxy(h Health, host, leaseState string, target *url.URL, rest st ModifyResponse: func(resp *http.Response) error { resp.Header.Set(HostHeader, host) resp.Header.Set(LeaseHeader, leaseState) + if ctxHeader != "" { + resp.Header.Set(CtxHeader, ctxHeader) + } rev.ttfb = time.Now() rev.streamed = strings.HasPrefix(resp.Header.Get("Content-Type"), "text/event-stream") t := newTee(resp.Body, rev.streamed) diff --git a/internal/proxy/helpers_test.go b/internal/proxy/helpers_test.go index e1c8807..da689db 100644 --- a/internal/proxy/helpers_test.go +++ b/internal/proxy/helpers_test.go @@ -79,6 +79,11 @@ func newUpstream(t *testing.T, name string) *upstream { u.mu.Unlock() fmt.Fprint(w, `{"object":"list","data":[{"id":"shared"},{"id":"`+name+`-only"}]}`) }) + // The v2 poller also asks /props; it is a health request, not a hit, so it is not counted. + // No n_ctx here: "unknown context" is what the v1 tests and TestUnknownContextNeverBlocks want. + mux.HandleFunc("/props", func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, `{"model_path":"`+name+`"}`) + }) mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { u.hits.Add(1) b, _ := io.ReadAll(r.Body) diff --git a/internal/proxy/proxy.go b/internal/proxy/proxy.go index f987e1e..60feb29 100644 --- a/internal/proxy/proxy.go +++ b/internal/proxy/proxy.go @@ -7,6 +7,7 @@ package proxy import ( "bytes" + "context" "encoding/json" "errors" "io" @@ -28,6 +29,7 @@ const ( HostHeader = "X-Crossbar-Host" LeaseHeader = "X-Crossbar-Lease" // "new" or "reused" RouteHeader = "X-Crossbar-Route" // client may name the route here instead of the path + CtxHeader = "X-Crossbar-Ctx" // "moved:>" when the context was relocated ) // errBodyTooLarge is returned when a request body exceeds MaxBody during the model peek. @@ -44,6 +46,11 @@ type Recorder interface { RecordRequest(store.Request) error } +// Waker rouses a sleeping host. *wake.Waker satisfies it. +type Waker interface { + Wake(ctx context.Context, host string) bool +} + // Handler forwards requests for a route to one of the route's healthy hosts, choosing by lease when // one is configured and by health alone otherwise. type Handler struct { @@ -53,6 +60,13 @@ type Handler struct { lim *limiter.Limiter rec Recorder log *slog.Logger + waker Waker +} + +// SetWaker installs the waker the consults when a route has no healthy host left. A nil waker +// (the default) leaves the ErrNoHost answer as it was in v0: a plain 503. +func (p *Handler) SetWaker(w Waker) { + p.waker = w } // New builds a Handler. A nil logger becomes slog.Default(). With a nil lease table it behaves like @@ -213,7 +227,7 @@ func (p *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { p.writeError(w, http.StatusServiceUnavailable, "no healthy host") return } - p.forward(w, r, route, name, "", rest, fp, model, started, 0) + p.forward(w, r, route, name, "", rest, fp, model, started, 0, 0, "") return } @@ -222,6 +236,12 @@ func (p *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { if err != nil { switch { case errors.Is(err, lease.ErrNoHost): + // No host healthy. Ask a waker to rouse a sleeping one; it answers + // (served or 503) when it has had a turn, else falls through to the + // plain 503. + if p.waker != nil && p.wakeOnErrNoHost(w, r, route, routeCfg, rest, model, fp, started, lease.Key{Route: route, FP: fp, Model: model}) { + return + } p.writeError(w, http.StatusServiceUnavailable, "no healthy host") case errors.Is(err, lease.ErrPinnedDown): p.writeError(w, http.StatusServiceUnavailable, "pinned host down") @@ -231,6 +251,14 @@ func (p *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } + // Slot, context guard and forward, holding the slot for the leased host. + p.serveLeased(w, r, route, routeCfg, rest, model, fp, started, host, reused) +} + +// serveLeased queues the request against the leased host's limiter, runs the +// context guard, and forwards. The slot is held for the originally leased host +// even if the guard relocates the lease: the guard already moved it. +func (p *Handler) serveLeased(w http.ResponseWriter, r *http.Request, route string, routeCfg config.Route, rest, model, fp string, started time.Time, host string, reused bool) { // Slot. A full queue is a 503; a context done while waiting means the client left. release, waited, err := p.lim.Acquire(r.Context(), host, model) if err != nil { @@ -263,5 +291,35 @@ func (p *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } defer release() - p.forward(w, r, route, host, leaseState(reused), rest, fp, model, started, waited) + // Context guard: if the prompt does not fit the leased host's per-slot + // context, move the conversation to a host where it fits, else answer 400. + now := time.Now() + host, header, _, done := p.guard(w, r, routeCfg.Hosts, host, route, model, fp, started) + if done { + return + } + p.forward(w, r, route, host, leaseState(reused), rest, fp, model, now, waited, 0, header) +} + +// wakeOnErrNoHost answers the request when no host was healthy. It asks, in +// route order, each candidate with a wake target to rouse itself; a host that +// wakes is leased once more and then served. When none wakes, it answers 503 +// with the hosts it tried. It returns true when the request has been answered. +func (p *Handler) wakeOnErrNoHost(w http.ResponseWriter, r *http.Request, route string, routeCfg config.Route, rest, model, fp string, started time.Time, key lease.Key) bool { + var tried []string + for _, name := range routeCfg.Hosts { + if p.cfg.Hosts[name].Wake == nil { + continue + } + tried = append(tried, name) + if !p.waker.Wake(r.Context(), name) { + continue + } + if newHost, _, err := p.leases.Acquire(key, routeCfg.Hosts, time.Now()); err == nil { + p.serveLeased(w, r, route, routeCfg, rest, model, fp, started, newHost, true) + return true + } + } + p.writeNoHealthyHost(w, route, model, fp, started, tried) + return true } diff --git a/internal/proxy/proxy_test.go b/internal/proxy/proxy_test.go index c1bb26a..39980aa 100644 --- a/internal/proxy/proxy_test.go +++ b/internal/proxy/proxy_test.go @@ -85,6 +85,19 @@ func TestDifferentConversationsSpreadByFreeSlots(t *testing.T) { } } +// waitUntil polls cond every 5 ms for up to two seconds and fails the test if it never holds. +func waitUntil(t *testing.T, cond func() bool) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("condition not reached within two seconds") +} + func TestQueueFullIs503(t *testing.T) { alpha := newUpstream(t, "alpha") alpha.delay = 400 * time.Millisecond @@ -99,14 +112,20 @@ hosts = ["alpha"] default_model = "shared" `, alpha) codes := make(chan int, 3) - for i := 1; i <= 3; i++ { - go func(i int) { + fire := func(i int) { + go func() { resp := r.post("/r/v1/chat/completions", conversation(i, 1)) drain(resp) codes <- resp.StatusCode - }(i) - time.Sleep(30 * time.Millisecond) // arrival order: 1 runs, 2 queues, 3 finds the queue full + }() } + // Arrival order is enforced by watching the limiter, not by sleeping: 1 runs, 2 queues, + // 3 finds the queue full. + fire(1) + waitUntil(t, func() bool { return r.lim.InFlight("alpha", "shared") == 1 }) + fire(2) + waitUntil(t, func() bool { return r.lim.Queued("alpha", "shared") == 1 }) + fire(3) got := map[int]int{} for i := 0; i < 3; i++ { got[<-codes]++ diff --git a/internal/store/store.go b/internal/store/store.go index b800f5c..77554fb 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -30,6 +30,7 @@ const ( ReasonPin = "pin" ReasonRelease = "release" ReasonDrain = "drain" + ReasonCtx = "ctx" ) // By selects the grouping column of a Usage query. diff --git a/internal/wake/wake.go b/internal/wake/wake.go new file mode 100644 index 0000000..4ae6c94 --- /dev/null +++ b/internal/wake/wake.go @@ -0,0 +1,142 @@ +// Package wake sends wake-on-LAN magic packets and waits for a sleeping host to +// appear healthy in the health table. A sleeping host takes tens of seconds to +// come up, so the Waker remembers when it last sent and wakes a host at most once +// per wait window. +package wake + +import ( + "context" + "fmt" + "net" + "sync" + "time" +) + +// Target describes how to wake one named host. +type Target struct { + MAC, Broadcast string // MAC "aa:bb:cc:dd:ee:ff" (any separator, any case); Broadcast "host:port" + Wait time.Duration +} + +// Health reports whether a named host is currently healthy. Implementations must +// be safe for concurrent use. +type Health interface{ Healthy(name string) bool } + +// magicPacketLen is six sync bytes plus the MAC repeated sixteen times. +const magicPacketLen = 6 + 6*16 + +// MagicPacket builds a wake-on-LAN magic packet: six 0xff bytes followed by the +// target MAC sixteen times, a 102-byte frame. +func MagicPacket(mac string) ([]byte, error) { + m, err := net.ParseMAC(mac) + if err != nil { + return nil, fmt.Errorf("wake: parse MAC %q: %w", mac, err) + } + if len(m) != 6 { + return nil, fmt.Errorf("wake: MAC %q is not six bytes", mac) + } + pkt := make([]byte, magicPacketLen) + for i := range pkt[:6] { + pkt[i] = 0xff + } + for i := 0; i < 16; i++ { + copy(pkt[6+i*6:], m) + } + return pkt, nil +} + +// Send emits one magic packet for mac to the broadcast address as a single UDP4 +// datagram, reporting parse, resolve and write errors. +func Send(mac, broadcast string) error { + pkt, err := MagicPacket(mac) + if err != nil { + return err + } + remote, err := net.ResolveUDPAddr("udp4", broadcast) + if err != nil { + return fmt.Errorf("wake: resolve broadcast %q: %w", broadcast, err) + } + conn, err := net.DialUDP("udp4", nil, remote) + if err != nil { + return fmt.Errorf("wake: dial broadcast %q: %w", broadcast, err) + } + defer conn.Close() + if _, err := conn.Write(pkt); err != nil { + return fmt.Errorf("wake: write packet to %q: %w", broadcast, err) + } + return nil +} + +// Waker wakes named hosts at most once per wait window and waits for the health +// table to report them healthy. It is safe for concurrent Wake calls. +type Waker struct { + mu sync.Mutex + targets map[string]Target + health Health + lastSent map[string]time.Time + poll time.Duration +} + +// New returns a Waker for the given targets, polling health every second. +func New(targets map[string]Target, h Health) *Waker { + return &Waker{ + targets: targets, + health: h, + lastSent: make(map[string]time.Time), + poll: time.Second, + } +} + +// PollEvery sets how often Wake re-checks health; it is a test hook. Production +// keeps the 1 s default from New. +func (w *Waker) PollEvery(d time.Duration) { + w.mu.Lock() + w.poll = d + w.mu.Unlock() +} + +// Wake sends a magic packet for host if none was sent in the last Wait, then +// polls health until the host is healthy, the wait elapses, or ctx is done. It +// returns true only when the host becomes healthy, and false for an unknown +// host, on timeout, or when ctx ends first. +func (w *Waker) Wake(ctx context.Context, host string) bool { + w.mu.Lock() + target, ok := w.targets[host] + if !ok { + w.mu.Unlock() + return false + } + now := time.Now() + if last, sent := w.lastSent[host]; !sent || now.Sub(last) >= target.Wait { + w.lastSent[host] = now + w.mu.Unlock() + if err := Send(target.MAC, target.Broadcast); err != nil { + return false + } + w.mu.Lock() + } + poll := w.poll + deadline := now.Add(target.Wait) + w.mu.Unlock() + + timer := time.NewTimer(poll) + defer timer.Stop() + for { + if w.health.Healthy(host) { + return true + } + if time.Now().After(deadline) { + return false + } + d := poll + if rem := time.Until(deadline); rem < d { + d = rem + } + timer.Reset(d) + select { + case <-ctx.Done(): + return false + case <-timer.C: + } + } +} diff --git a/internal/wake/wake_test.go b/internal/wake/wake_test.go new file mode 100644 index 0000000..53e6179 --- /dev/null +++ b/internal/wake/wake_test.go @@ -0,0 +1,114 @@ +package wake_test + +import ( + "bytes" + "context" + "net" + "testing" + "time" + + "git.wntrmute.dev/kyle/crossbar/internal/wake" +) + +func listen(t *testing.T) (*net.UDPConn, string) { + conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { conn.Close() }) + return conn, conn.LocalAddr().String() +} + +func TestMagicPacket(t *testing.T) { + pkt, err := wake.MagicPacket("aa:bb:cc:dd:ee:ff") + if err != nil { + t.Fatal(err) + } + if len(pkt) != 102 || !bytes.Equal(pkt[:6], bytes.Repeat([]byte{0xff}, 6)) { + t.Fatalf("packet = % x", pkt) + } + mac := []byte{0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff} + for i := 0; i < 16; i++ { + if !bytes.Equal(pkt[6+6*i:12+6*i], mac) { + t.Fatalf("repetition %d wrong: % x", i, pkt[6+6*i:12+6*i]) + } + } + for _, bad := range []string{"", "aa:bb", "zz:bb:cc:dd:ee:ff", "aabbccddeeff00"} { + if _, err := wake.MagicPacket(bad); err == nil { + t.Errorf("MagicPacket(%q) must fail", bad) + } + } + if p2, _ := wake.MagicPacket("AA-BB-CC-DD-EE-FF"); !bytes.Equal(p2, pkt) { + t.Errorf("dash-separated upper-case MAC must give the same packet") + } +} + +func TestSendReachesTheBroadcastAddress(t *testing.T) { + conn, addr := listen(t) + if err := wake.Send("aa:bb:cc:dd:ee:ff", addr); err != nil { + t.Fatal(err) + } + buf := make([]byte, 200) + _ = conn.SetReadDeadline(time.Now().Add(time.Second)) + n, _, err := conn.ReadFromUDP(buf) + if err != nil || n != 102 { + t.Fatalf("received %d bytes, err %v", n, err) + } + if err := wake.Send("aa:bb:cc:dd:ee:ff", "256.1.1.1:9"); err == nil { + t.Error("an unresolvable broadcast address must be an error") + } +} + +// fakeHealth flips to healthy after `after` calls to Healthy. +type fakeHealth struct{ calls, after int } + +func (f *fakeHealth) Healthy(name string) bool { f.calls++; return f.calls > f.after } + +func TestWakerSendsOncePerWindowAndWaitsForHealth(t *testing.T) { + conn, addr := listen(t) + h := &fakeHealth{after: 3} + w := wake.New(map[string]wake.Target{"titan": {MAC: "aa:bb:cc:dd:ee:ff", Broadcast: addr, Wait: 2 * time.Second}}, h) + w.PollEvery(20 * time.Millisecond) // test hook: how often Wake re-checks health + start := time.Now() + ok := w.Wake(context.Background(), "titan") + if !ok { + t.Fatal("Wake must return true once the host reports healthy") + } + if time.Since(start) > time.Second { + t.Errorf("Wake waited %v for a host that came up after 3 checks", time.Since(start)) + } + _ = conn.SetReadDeadline(time.Now().Add(200 * time.Millisecond)) + buf := make([]byte, 200) + if n, _, err := conn.ReadFromUDP(buf); err != nil || n != 102 { + t.Fatalf("no magic packet received: %d %v", n, err) + } + // A second Wake inside the same window does not send again (the host is booting). + _ = w.Wake(context.Background(), "titan") + _ = conn.SetReadDeadline(time.Now().Add(150 * time.Millisecond)) + if n, _, err := conn.ReadFromUDP(buf); err == nil { + t.Errorf("a second packet (%d bytes) was sent inside the wait window", n) + } + if w.Wake(context.Background(), "nobody") { + t.Errorf("unknown host: Wake must return false") + } +} + +func TestWakeGivesUpAfterWait(t *testing.T) { + _, addr := listen(t) + h := &fakeHealth{after: 1 << 30} + w := wake.New(map[string]wake.Target{"titan": {MAC: "aa:bb:cc:dd:ee:ff", Broadcast: addr, Wait: 300 * time.Millisecond}}, h) + w.PollEvery(20 * time.Millisecond) + start := time.Now() + if w.Wake(context.Background(), "titan") { + t.Fatal("Wake must return false when the host never comes up") + } + if d := time.Since(start); d < 250*time.Millisecond || d > 900*time.Millisecond { + t.Errorf("Wake returned after %v, want about the 300ms wait", d) + } + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + start = time.Now() + if w.Wake(ctx, "titan") || time.Since(start) > 200*time.Millisecond { + t.Errorf("a cancelled context must end the wait early (took %v)", time.Since(start)) + } +} diff --git a/tools/smoke.sh b/tools/smoke.sh index 7c2bcb4..420d546 100755 --- a/tools/smoke.sh +++ b/tools/smoke.sh @@ -1,80 +1,60 @@ #!/bin/sh -# Smoke run (v1): two fake upstreams, one crossbar with a fresh SQLite file, real HTTP. -# Checks routing, leases (sticky + header), failover, recovery, streaming, queueing, pin, drain, -# usage and metrics. Prints "smoke: ok" or fails with the crossbar log. +# Smoke run (v2): everything v1 checked, plus the context guard, wake-on-LAN and identity gating. +# Prints "smoke: ok" or fails with the crossbar log. set -eu cd "$(dirname "$0")/.." tmp=$(mktemp -d); trap 'kill $pids 2>/dev/null; rm -rf "$tmp"' EXIT INT TERM pids="" -sed "s#^db .*#db = \"$tmp/crossbar.db\"#" example.toml > "$tmp/crossbar.toml" -bin/fakeupstream -listen 127.0.0.1:18081 -name alpha -models ornith-1.5-35b-a3b,small-9b -down-file "$tmp/alpha.down" -slow 600 >"$tmp/alpha.log" 2>&1 & pids="$pids $!" -bin/fakeupstream -listen 127.0.0.1:18082 -name beta -models ornith-1.5-35b-a3b -down-file "$tmp/beta.down" >"$tmp/beta.log" 2>&1 & pids="$pids $!" +sed -e "s#^db .*#db = \"$tmp/crossbar.db\"#" -e 's#^identity .*#identity = "header"#' -e 's#^\# peers = \["talos"\]#peers = ["talos"]#' example.toml > "$tmp/crossbar.toml" +# alpha: small context (4096 per slot = 8192/2); beta: large, sleeps until woken +bin/fakeupstream -listen 127.0.0.1:18081 -name alpha -models ornith-1.5-35b-a3b,small-9b -down-file "$tmp/alpha.down" -slow 600 -n-ctx 8192 -slots 2 >"$tmp/alpha.log" 2>&1 & pids="$pids $!" +bin/fakeupstream -listen 127.0.0.1:18082 -name beta -models ornith-1.5-35b-a3b -down-file "$tmp/beta.down" -n-ctx 131072 -slots 2 -wol-listen 127.0.0.1:19082 -wol-mac aa:bb:cc:dd:ee:02 >"$tmp/beta.log" 2>&1 & pids="$pids $!" +touch "$tmp/beta.down" # beta starts "asleep" bin/crossbar -config "$tmp/crossbar.toml" >"$tmp/crossbar.log" 2>&1 & pids="$pids $!" -sleep 1.5 +sleep 2.5 # two polls: alpha healthy, beta down fail() { echo "smoke: FAIL: $*" >&2; echo "--- crossbar.log"; cat "$tmp/crossbar.log"; exit 1; } base=http://127.0.0.1:17777 conv() { printf '{"model":"ornith-1.5-35b-a3b","stream":false,"messages":[{"role":"system","content":"smoke"},{"role":"user","content":"conversation %s"}]}' "$1"; } +big() { printf '{"model":"ornith-1.5-35b-a3b","stream":false,"messages":[{"role":"user","content":"%s"}]}' "$(head -c 40000 /dev/zero | tr '\0' 'x')"; } hdrs() { curl -s -o /dev/null -w '%{http_code} %header{X-Crossbar-Host} %header{X-Crossbar-Lease}' "$@"; } -# 1. a conversation gets a lease and keeps it; beta wins (2 slots × weight 2 vs 1 × 1) +# 1. v1 behaviour: with beta asleep, opencode-a goes to alpha h=$(hdrs -X POST -H 'Content-Type: application/json' -d "$(conv A)" "$base/opencode-a/v1/chat/completions") -[ "$h" = "200 beta new" ] || fail "first turn should be '200 beta new', got '$h'" -h=$(hdrs -X POST -H 'Content-Type: application/json' -d "$(conv A)" "$base/opencode-a/v1/chat/completions") -[ "$h" = "200 beta reused" ] || fail "second turn should reuse beta, got '$h'" +[ "$h" = "200 alpha new" ] || fail "with beta asleep conversation A should be '200 alpha new', got '$h'" +curl -s "$base/_crossbar/hosts" | grep -q '"alpha":{[^}]*"n_ctx":8192' || fail "hosts view does not show alpha n_ctx 8192: $(curl -s $base/_crossbar/hosts)" -# 2. header route -h=$(hdrs -X POST -H 'Content-Type: application/json' -H 'X-Crossbar-Route: hermes-x' -d "$(conv B)" "$base/v1/chat/completions") -case "$h" in "200 beta new") ;; *) fail "header route hermes-x should be '200 beta new', got '$h'";; esac -h=$(curl -s -o /dev/null -w '%{http_code}' "$base/nope/v1/models"); [ "$h" = "404" ] || fail "unknown route 404, got $h" +# 2. context guard: a ~12k-token prompt does not fit alpha's 4096-token slot; beta is asleep and +# wakeable, so crossbar must send the magic packet, wait for beta, and place the prompt there. +start=$(date +%s) +h=$(hdrs -m 40 -X POST -H 'Content-Type: application/json' -d "$(big)" "$base/opencode-a/v1/chat/completions") +[ "$h" = "200 beta new" ] || fail "oversized prompt should wake beta and land there, got '$h' after $(( $(date +%s) - start ))s" +grep -q "magic packet received" "$tmp/beta.log" || fail "beta never saw a magic packet" +curl -s "$base/_crossbar/hosts" | grep -q '"beta":{"healthy":true' || fail "beta not healthy after wake" -# 3. pin opencode-a to alpha: conversation A's next turn moves (an operator pin outranks the lease) -h=$(curl -s -o /dev/null -w '%{http_code}' -X POST -H 'Content-Type: application/json' -d '{"host":"alpha","pin":true}' "$base/_crossbar/routes/opencode-a") -[ "$h" = "200" ] || fail "pin returned $h" -h=$(hdrs -X POST -H 'Content-Type: application/json' -d "$(conv A)" "$base/opencode-a/v1/chat/completions") -[ "$h" = "200 alpha new" ] || fail "after pin, conversation A should be '200 alpha new', got '$h'" -curl -s "$base/_crossbar/routes" | grep -q '"pinned":"alpha"' || fail "routes view does not show the pin: $(curl -s $base/_crossbar/routes)" +# 3. with beta awake, a prompt that fits nowhere is a 400 (both slots too small? no — beta fits): +# check the guard's refusal with a prompt beyond beta's 65536-per-slot too +# (the body goes through a file: a 300 KB string cannot be a single argv element on Linux) +{ printf '{"model":"ornith-1.5-35b-a3b","messages":[{"role":"user","content":"'; head -c 300000 /dev/zero | tr '\0' 'x'; printf '"}]}'; } > "$tmp/toolarge-req.json" +h=$(curl -s -o "$tmp/toolarge.json" -w '%{http_code}' -X POST -H 'Content-Type: application/json' -d @"$tmp/toolarge-req.json" "$base/opencode-a/v1/chat/completions") +[ "$h" = "400" ] && grep -q '"prompt too large"' "$tmp/toolarge.json" || fail "300 KB prompt should be 400 prompt too large, got $h $(cat "$tmp/toolarge.json")" -# 4. queue: alpha has parallel 1, queue_max 1, and answers in 600 ms → of three concurrent, one is 503 -for i in 1 2 3; do (curl -s -o /dev/null -w '%{http_code}\n' -X POST -H 'Content-Type: application/json' -d "$(conv Q$i)" "$base/opencode-a/v1/chat/completions" >> "$tmp/codes") & sleep 0.1; done; wait $! 2>/dev/null || true -sleep 2.5 -sort "$tmp/codes" | uniq -c | tr -s ' ' > "$tmp/counts" -grep -q '2 200' "$tmp/counts" && grep -q '1 503' "$tmp/counts" || fail "queue test wanted two 200 and one 503, got: $(cat "$tmp/counts")" +# 4. identity: hermes-x is locked to peer talos (header mode) +h=$(curl -s -o /dev/null -w '%{http_code}' -X POST -H 'Content-Type: application/json' -d "$(conv B)" "$base/hermes-x/v1/chat/completions") +[ "$h" = "403" ] || fail "hermes-x without a peer header should be 403, got $h" +h=$(curl -s -o /dev/null -w '%{http_code}' -X POST -H 'Content-Type: application/json' -H 'X-Crossbar-Peer: titan' -d "$(conv B)" "$base/hermes-x/v1/chat/completions") +[ "$h" = "403" ] || fail "hermes-x as titan should be 403, got $h" +h=$(hdrs -X POST -H 'Content-Type: application/json' -H 'X-Crossbar-Peer: talos' -d "$(conv B)" "$base/hermes-x/v1/chat/completions") +case "$h" in 200*) ;; *) fail "hermes-x as talos should be 200, got '$h'";; esac +h=$(curl -s -o /dev/null -w '%{http_code}' "$base/_crossbar/hosts"); [ "$h" = "200" ] || fail "admin must not be gated, got $h" -# 5. release the pin, drain alpha: new conversations go to beta, A stays on alpha -curl -s -o /dev/null -X POST -H 'Content-Type: application/json' -d '{"release":true}' "$base/_crossbar/routes/opencode-a" -h=$(curl -s -o /dev/null -w '%{http_code}' -X POST -H 'Content-Type: application/json' -d '{"drain":true}' "$base/_crossbar/hosts/alpha"); [ "$h" = "200" ] || fail "drain returned $h" -h=$(hdrs -X POST -H 'Content-Type: application/json' -d "$(conv C)" "$base/opencode-a/v1/chat/completions") -[ "$h" = "200 beta new" ] || fail "with alpha draining a new conversation should go to beta, got '$h'" -curl -s "$base/_crossbar/hosts" | grep -q '"alpha":{[^}]*"draining":true' || fail "hosts view does not show alpha draining" -curl -s -o /dev/null -X POST -H 'Content-Type: application/json' -d '{"drain":false}' "$base/_crossbar/hosts/alpha" - -# 6. failover + recovery -touch "$tmp/beta.down"; sleep 2.5 -h=$(hdrs -X POST -H 'Content-Type: application/json' -d "$(conv C)" "$base/opencode-a/v1/chat/completions") -[ "$h" = "200 alpha new" ] || fail "with beta down conversation C should move to alpha, got '$h'" -curl -s "$base/_crossbar/hosts" | grep -q '"beta":{"healthy":false' || fail "hosts view does not show beta unhealthy" -rm "$tmp/beta.down"; sleep 3.5 -curl -s "$base/_crossbar/hosts" | grep -q '"beta":{"healthy":true' || fail "beta did not recover after two good polls" - -# 7. streaming still arrives incrementally, and the final usage chunk is untouched +# 5. v1 regression: streaming still incremental, usage and metrics present start=$(date +%s%N) -curl -sN -X POST -H 'Content-Type: application/json' -d '{"model":"ornith-1.5-35b-a3b","stream":true,"messages":[{"role":"user","content":"stream me"}]}' \ - "$base/opencode-a/v1/chat/completions" | while IFS= read -r line; do - [ -n "$line" ] || continue - now=$(date +%s%N); echo "$(( (now - start) / 1000000 )) $line" - done > "$tmp/stream.txt" +curl -sN -X POST -H 'Content-Type: application/json' -H 'X-Crossbar-Peer: talos' -d '{"model":"ornith-1.5-35b-a3b","stream":true,"messages":[{"role":"user","content":"stream me"}]}' \ + "$base/hermes-x/v1/chat/completions" | while IFS= read -r line; do [ -n "$line" ] || continue; now=$(date +%s%N); echo "$(( (now - start) / 1000000 )) $line"; done > "$tmp/stream.txt" firstms=$(head -1 "$tmp/stream.txt" | cut -d' ' -f1); lastms=$(tail -1 "$tmp/stream.txt" | cut -d' ' -f1) -[ -n "$firstms" ] && [ "$((lastms - firstms))" -ge 600 ] || fail "stream arrived in one burst: $(cat "$tmp/stream.txt")" -grep -q '"usage"' "$tmp/stream.txt" && grep -q 'DONE' "$tmp/stream.txt" || fail "stream lost the usage chunk or DONE" - -# 8. accounting and metrics +[ -n "$firstms" ] && [ "$((lastms - firstms))" -ge 600 ] || fail "stream arrived in one burst" sleep 1 -u=$(curl -s "$base/_crossbar/usage?by=host") -echo "$u" | grep -q '"key":"alpha"' && echo "$u" | grep -q '"key":"beta"' || fail "usage by host: $u" -echo "$u" | grep -q '"cached_tokens":[1-9]' || fail "usage has no cached tokens (SSE/JSON usage not captured): $u" -curl -s -H 'Accept: text/plain' "$base/_crossbar/usage?by=route" | grep -qi 'cache' || fail "text usage table missing" -m=$(curl -s "$base/_crossbar/metrics") -echo "$m" | grep -q 'crossbar_requests_total{route="opencode-a",host="alpha",status="503"} 1' || fail "metrics missing the 503: $m" -echo "$m" | grep -q 'crossbar_host_healthy{host="beta"} 1' || fail "metrics missing host health" -grep -q 'route=opencode-a host=' "$tmp/crossbar.log" || fail "no request log line" +curl -s "$base/_crossbar/usage?by=host" | grep -q '"cached_tokens":[1-9]' || fail "usage has no cached tokens" +curl -s "$base/_crossbar/metrics" | grep -q 'crossbar_requests_total{route="hermes-x",host="beta",status="403"}' && fail "403s are refused before a lease and must not be counted as requests" +curl -s "$base/_crossbar/metrics" | grep -q 'crossbar_host_healthy{host="beta"} 1' || fail "metrics missing beta health" echo "smoke: ok (stream spread $((lastms - firstms)) ms)"