| 1 | package extension |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "testing" |
| 7 | "time" |
| 8 | ) |
| 9 | |
| 10 | // TestHandlerPoolSaturated verifies the bounded inbound concurrency: with all |
| 11 | // 32 handler slots occupied, the next request is answered -32099 (server |
| 12 | // busy) while the parked requests still complete afterwards. |
| 13 | func TestHandlerPoolSaturated(t *testing.T) { |
| 14 | release := make(chan struct{}) |
| 15 | interceptors := map[string]InterceptorFunc{ |
| 16 | "tool.before": func(ctx context.Context, _ string, _ json.RawMessage) (*InterceptResult, error) { |
| 17 | select { |
| 18 | case <-release: |
| 19 | case <-ctx.Done(): |
| 20 | } |
| 21 | return Continue(), nil |
| 22 | }, |
| 23 | } |
| 24 | host, _ := startFakeHost(t, basicHandler(), Options{Interceptors: interceptors}) |
| 25 | host.handshake(t) |
| 26 | |
| 27 | channels := make([]chan hostResponse, 0, maxConcurrentHandlers+1) |
| 28 | for i := 0; i < maxConcurrentHandlers+1; i++ { |
| 29 | _, ch := host.startRequest(MethodExtensionIntercept, InterceptParams{ |
| 30 | Event: EventToolBefore, Seq: uint64(i + 1), Payload: json.RawMessage(`{}`), |
| 31 | }) |
| 32 | channels = append(channels, ch) |
| 33 | } |
| 34 | |
| 35 | // Exactly one request — the one finding no handler slot — is rejected |
| 36 | // immediately; the rest stay parked on release. |
| 37 | busyIdx := -1 |
| 38 | deadline := time.Now().Add(5 * time.Second) |
| 39 | for busyIdx < 0 && time.Now().Before(deadline) { |
| 40 | for i, ch := range channels { |
| 41 | select { |
| 42 | case resp := <-ch: |
| 43 | if resp.Err == nil || resp.Err.Code != CodeServerBusy { |
| 44 | t.Fatalf("request %d: unexpected early response %+v", i, resp) |
| 45 | } |
| 46 | busyIdx = i |
| 47 | default: |
| 48 | } |
| 49 | } |
| 50 | if busyIdx < 0 { |
| 51 | time.Sleep(5 * time.Millisecond) |
| 52 | } |
| 53 | } |
| 54 | if busyIdx < 0 { |
| 55 | t.Fatal("no request was answered server-busy") |
| 56 | } |
| 57 | for i, ch := range channels { |
| 58 | if i == busyIdx { |
| 59 | continue |
| 60 | } |
| 61 | select { |
| 62 | case resp := <-ch: |
| 63 | t.Fatalf("request %d: expected parked, got %+v", i, resp) |
| 64 | default: |
| 65 | } |
| 66 | } |
| 67 | |
| 68 | close(release) |
| 69 | for i, ch := range channels { |
| 70 | if i == busyIdx { |
| 71 | continue |
| 72 | } |
| 73 | select { |
| 74 | case resp := <-ch: |
| 75 | if resp.Err != nil { |
| 76 | t.Fatalf("parked request %d errored: %+v", i, resp.Err) |
| 77 | } |
| 78 | case <-time.After(5 * time.Second): |
| 79 | t.Fatalf("parked request %d did not complete after release", i) |
| 80 | } |
| 81 | } |
| 82 | } |
| 83 |