diff --git a/AGENTS.md b/AGENTS.md index 53cdfef..794990e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,8 +31,8 @@ To refresh: `curl -sL -o openapi.yaml https://docs.kalshi.com/openapi.yaml` (sam Where the spec and production disagree, production wins, and the difference gets documented in the README. Known cases: -- Error bodies. The spec's `ErrorResponse` is flat `{"code","message","details"}`. Production returns `{"error":{"code":...,"message":...}}` for most errors and `{"msg":"..."}` for parameter-binding 400s. `newAPIError` parses all three; keep it that way if the spec changes. `APIError.RawBody` always has the body. -- Hosts. Spec primary REST host is `external-api.kalshi.com`; `api.elections.kalshi.com` is listed as also supported and is the client default. AsyncAPI names `external-api-ws.kalshi.com`; the client defaults to `api.elections.kalshi.com` for WS too. Both answer. +- Error bodies. The spec's `ErrorResponse` is flat `{"code","message","details"}`. Production returns `{"error":{"code":...,"message":...}}` for most errors and `{"msg":"..."}` for parameter-binding 400s. `newAPIError` parses all three (from up to 64 KiB of the body); keep it that way if the spec changes. `APIError.RawBody` always has the first 512 bytes. +- Hosts. The client defaults to the spec's primary hosts: `external-api.kalshi.com` (REST) and `external-api-ws.kalshi.com` (WS). The shared `api.elections.kalshi.com` serves both protocols, is listed as also supported, and was the default before v0.6.2; `BaseURL` / `WSHost` select it. - `client_order_id` deduplication is documented in Kalshi's quick-start guide, not in the OpenAPI field description. A replay the server already applied returns 409. ## Conventions @@ -47,7 +47,7 @@ Where the spec and production disagree, production wins, and the difference gets - `CreateV2` / `BatchCreateV2` pick at runtime based on whether every order has a `client_order_id`. - Endpoints whose spec response is empty return `error` only. - Prefer fixed-point `_fp` and `_dollars` string fields. Legacy integer price/count fields are being removed by Kalshi; do not add new ones. -- WebSocket: command replies are matched by `id`; `get_snapshot` is the exception (answered by `orderbook_snapshot` frames keyed by `sid`). Multi-channel `Subscribe` expects one `subscribed` reply per channel. All socket writes go through `writeMu`. +- WebSocket: command replies are matched by `id`; `get_snapshot` is the exception (answered by `orderbook_snapshot` frames keyed by `sid`). Multi-channel `Subscribe` expects one `subscribed` reply per channel. Data-frame writes go through `writeSem` (a 1-slot channel so waiters can honor their context) and are bounded by `WSWriteTimeout`; control frames use gorilla's `WriteControl`, which serializes itself. - Match the surrounding code. No drive-by refactors, reformatting, or comment rewrites in files you are not otherwise changing. ## Tests are mandatory diff --git a/CHANGELOG.md b/CHANGELOG.md index 75f1316..0c6d155 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,26 @@ All notable changes to this project are documented here. The client tracks [Kals **Versioning.** The module follows semver. While it is at major version 0, a **minor** release may contain breaking changes; when it does, they are listed first under a `### Breaking` heading with migration notes, and CI refuses a release that has API-incompatible changes (per `gorelease`) without that section, or that has one on a patch bump. Patch releases never break. From v1.0.0 on, breaking changes require a major bump. +## [0.6.2] — 2026-09-12 + +Default hosts move to Kalshi's recommended `external-api` endpoints; WebSocket writes and `Close` are bounded; `DoConcurrent` no longer spawns a goroutine per index; long error bodies decode; the repository has a license. No API-incompatible changes. + +### Fixed + +- **A WebSocket write could block past the caller's deadline and hang `Close`** ([#15](https://github.com/UTXOnly/oddrip/issues/15)). `Subscribe` / `Unsubscribe` / `ListSubscriptions` / `UpdateSubscription` wrote to the socket with no deadline while holding the write lock, so on a wedged or backpressured socket one command blocked indefinitely, every later command queued behind it, context deadlines had no effect, and `Close` never reached its bounded wait. Each frame is now written with the sooner of the context deadline and `WSWriteTimeout` (default 10s), and waiting for the write slot honors the context. A write that times out fails the connection with an error wrapping the new `ErrWSWriteTimeout` (check with `errors.Is`): `Messages()` closes and `Err()` reports it, the same path as `ErrWSSlowConsumer`; the caller whose own deadline cut the write gets `context.DeadlineExceeded`. A caller that gives up waiting for the slot gets `ctx.Err()` and the connection stays healthy. `Close` now sends the close frame as a bounded control write, skips it when a command write is in flight, and returns within about `WSWriteTimeout` plus five seconds even when the peer is not reading. +- **`DoConcurrent` started one goroutine per index even with a bounded `maxInFlight`** ([#14](https://github.com/UTXOnly/oddrip/issues/14)). The limit gated the active `fn` calls behind a semaphore, but every goroutine was created up front and parked on it, so a large `n` with a small limit still cost `n` goroutine stacks. A positive `maxInFlight` now starts at most `min(n, maxInFlight)` workers that take the next index as their current call returns, and the result buffer is sized to the workers rather than `n`. Results are still index-ordered, and a cancelled context still returns the results collected so far plus `ctx.Err()`, after which workers take no further indices and never block on the result channel. `maxInFlight <= 0` is unchanged (all `n` calls at once), except that a context already cancelled on entry no longer invokes `fn` — matching what bounded mode already did. +- **`APIError` lost `Code` / `Message` / `Details` when a valid error body exceeded 512 bytes** ([#16](https://github.com/UTXOnly/oddrip/issues/16)). Only the 512-byte `RawBody` snippet was read, and the structured fields were decoded from that truncated buffer, so long validation details or gateway metadata reduced the error to `api error STATUS`. The structured fields are now decoded from up to 64 KiB of the body; `RawBody` is still the first 512 bytes, and bodies over 64 KiB are read no further and are not decoded. + +### Changed + +- **Default hosts are now `external-api.kalshi.com` (REST) and `external-api-ws.kalshi.com` (WebSocket)** ([#18](https://github.com/UTXOnly/oddrip/issues/18)), the production endpoints Kalshi recommends and the primary hosts in both specs. The shared `api.elections.kalshi.com` host that was the default through 0.6.1 remains supported for both protocols; pass `oddrip.BaseURL("https://api.elections.kalshi.com/trade-api/v2")` and `oddrip.WSHost("api.elections.kalshi.com")` to keep using it. Request signing is unaffected — the signed message is `timestamp + METHOD + path` and excludes the host. If you allowlist egress hosts, add the new ones. The WebSocket example maps `external-api.kalshi.com` in `BASE_URL` to the `-ws` host. + +### Added + +- `WSWriteTimeout(d)` option (default 10s; `<= 0` uses the default) and `ErrWSWriteTimeout` ([#15](https://github.com/UTXOnly/oddrip/issues/15)). +- `LICENSE` — MIT ([#17](https://github.com/UTXOnly/oddrip/issues/17)). The vendored `openapi.yaml` / `asyncapi.yaml` are Kalshi's published specifications and are not covered by it; the README says so. +- **Tests:** `DoConcurrent` goroutine bound at large `n`, cancellation while workers are blocked, `n == 0`; WebSocket write-slot cancellation, cancelled and live callers contending, write timeout as a terminal error, a caller deadline cutting a write, a blocked writer not holding other callers, bounded `Close` with a stuck writer and with a full socket; default REST request destination and WebSocket dial URL; the signed path is identical across the external-api, shared, and demo base URLs; long, malformed, and oversized error bodies through `newAPIError`. + ## [0.6.1] — 2026-09-12 Error bodies from production now decode; specs synced to OpenAPI 3.30.0; four missing query filters; malformed WebSocket frames fail the connection. No API-incompatible changes. diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..54951bb --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Brian Hartford + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md index 3ad758a..701191d 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ Go client for the [Kalshi Trade API](https://docs.kalshi.com/): REST plus WebSocket market data. Tracks vendored OpenAPI **3.30.0** / AsyncAPI **2.0.0**. ```bash -go get github.com/UTXOnly/oddrip/oddrip@v0.6.1 +go get github.com/UTXOnly/oddrip/oddrip@v0.6.2 ``` Go 1.24+. Import the client as `github.com/UTXOnly/oddrip/oddrip` and types as `github.com/UTXOnly/oddrip/oddrip/types`. `oddrip.Version` matches the module tag. @@ -26,7 +26,7 @@ client := oddrip.New( ) ``` -Default base URL is `https://api.elections.kalshi.com/trade-api/v2`. The spec lists `https://external-api.kalshi.com/trade-api/v2` as the primary production host and both as supported; pass `oddrip.BaseURL(...)` to switch, or for demo (`https://demo-api.kalshi.co/trade-api/v2`). `oddrip.HTTPClient(...)` swaps the transport. Auth is RSA-PSS (PKCS#8 or PKCS#1 PEM) over `timestamp + METHOD + path` (query string excluded); the same signer is used for REST and the WebSocket handshake. +Default base URL is `https://external-api.kalshi.com/trade-api/v2`, the production host Kalshi recommends. The older shared host `https://api.elections.kalshi.com/trade-api/v2` (the default before v0.6.2) is still supported; pass `oddrip.BaseURL(...)` to use it, or for demo (`https://demo-api.kalshi.co/trade-api/v2`). Switching hosts does not affect signing — the signed message covers the path only. `oddrip.HTTPClient(...)` swaps the transport. Auth is RSA-PSS (PKCS#8 or PKCS#1 PEM) over `timestamp + METHOD + path` (query string excluded); the same signer is used for REST and the WebSocket handshake. ## REST @@ -68,7 +68,7 @@ for { } ``` -Non-2xx responses are `*oddrip.APIError` with `StatusCode`, `Code`, `Message`, and `RawBody` (first 512 bytes). Kalshi's production error bodies nest `code` / `message` under `"error"` (the spec shows them flat) and parameter-binding 400s use `{"msg": ...}`; all three shapes are parsed. `RequestID` is read from a `Request-Id` header Kalshi does not currently send. An empty ticker or ID path parameter returns `oddrip.ErrEmptyPathParam` without sending — an empty order ID would otherwise hit CancelAll. +Non-2xx responses are `*oddrip.APIError` with `StatusCode`, `Code`, `Message`, and `RawBody` (first 512 bytes). `Code`, `Message`, and `Details` are decoded from up to 64 KiB of the body; a larger body is not decoded and leaves them empty. Kalshi's production error bodies nest `code` / `message` under `"error"` (the spec shows them flat) and parameter-binding 400s use `{"msg": ...}`; all three shapes are parsed. `RequestID` is read from a `Request-Id` header Kalshi does not currently send. An empty ticker or ID path parameter returns `oddrip.ErrEmptyPathParam` without sending — an empty order ID would otherwise hit CancelAll. ## Retries @@ -87,7 +87,7 @@ client := oddrip.New(oddrip.RetryConfigOption(oddrip.RetryConfig{MaxAttempts: 1} ## Concurrent requests -The client is safe for concurrent use. `DoConcurrent(ctx, n, maxInFlight, fn)` runs at most `maxInFlight` calls at once (`0` is unbounded) and returns results in index order. Cancel returns the results collected so far plus `ctx.Err()`. +The client is safe for concurrent use. `DoConcurrent(ctx, n, maxInFlight, fn)` runs at most `maxInFlight` calls at once (`0` is unbounded) and returns results in index order. A positive `maxInFlight` also caps the worker goroutines at `min(n, maxInFlight)`, so `n` can be large without creating `n` goroutines. Cancel returns the results collected so far plus `ctx.Err()`. ```go results, err := oddrip.DoConcurrent(ctx, len(tickers), 8, func(i int) (*types.GetMarketResponse, error) { @@ -141,7 +141,7 @@ if err := conn.Err(); !errors.Is(err, oddrip.ErrWSClosed) { } ``` -Commands: `Subscribe`, `Unsubscribe`, `ListSubscriptions`, `UpdateSubscription`. Channel names and `WSType*` constants are in `types`. CF Benchmarks channels take `IndexIDs` (`[]string{"all"}` for every index). Command rejections are returned as `*oddrip.WSError`. Default endpoint is `wss://api.elections.kalshi.com/trade-api/ws/v2`; the AsyncAPI names `external-api-ws.kalshi.com` as the production host — both accept connections. Point elsewhere with `WSHost` / `WSPath` / `WSScheme`. +Commands: `Subscribe`, `Unsubscribe`, `ListSubscriptions`, `UpdateSubscription`. Channel names and `WSType*` constants are in `types`. CF Benchmarks channels take `IndexIDs` (`[]string{"all"}` for every index). Command rejections are returned as `*oddrip.WSError`. Default endpoint is `wss://external-api-ws.kalshi.com/trade-api/ws/v2`, the production host Kalshi recommends; the older shared `api.elections.kalshi.com` (the default before v0.6.2) still accepts connections. Point elsewhere with `WSHost` / `WSPath` / `WSScheme` — note the recommended REST and WebSocket hosts differ (`external-api` vs `external-api-ws`), so a `BaseURL` override does not imply a `WSHost` one. - If `Messages()` falls behind, the connection fails with `ErrWSSlowConsumer` (buffer default 4096) rather than dropping deltas. Reconnect and re-snapshot any local book. - Errors scoped to a subscription arrive on `Messages()` as `Type: "error"` with a `SID`, not as a returned `*WSError`. Codes 10 (channel error) and 25 (subscription buffer overflow) are terminal for that subscription — resubscribe. Decode into `types.ErrorMsg`. @@ -150,7 +150,8 @@ Commands: `Subscribe`, `Unsubscribe`, `ListSubscriptions`, `UpdateSubscription`. - `get_snapshot` needs `SID` or a one-element `Sids`. It returns when the first `orderbook_snapshot` for that subscription arrives (`Type` is `"orderbook_snapshot"`); the frames also go to `Messages()`. - `indexlist` / `underlying_list` replies are not `Type: "ok"`. - A multi-channel `Subscribe` that fails partway returns the accepted SIDs alongside the `*WSError`. -- `Close` is idempotent. Commands are safe to call concurrently. +- Every socket write is bounded by `WSWriteTimeout` (default 10s) or the command's context deadline, whichever is sooner. A write that times out fails the connection: `Err()` wraps `ErrWSWriteTimeout` and `Messages()` closes, same as a slow consumer; the command whose own deadline cut the write returns `context.DeadlineExceeded`. A command that gives up while waiting its turn to write returns `ctx.Err()` and leaves the connection healthy. +- `Close` is idempotent and returns within about `WSWriteTimeout` plus five seconds even if the peer has stopped reading. Commands are safe to call concurrently. ## Examples @@ -165,3 +166,7 @@ CI runs `gofmt`, `go mod tidy`, `go vet`, `staticcheck`, `govulncheck`, and `go 3. Update the `@vX.Y.Z` pin in this README. Semver. While at v0, a minor release may break; those changes go first under `### Breaking`. CI runs `gorelease` against the previous tag and refuses a release that has API-incompatible changes without that heading, or that declares one on a patch bump. + +## License + +MIT — see [LICENSE](LICENSE). `openapi.yaml` and `asyncapi.yaml` are Kalshi's published API specifications ([OpenAPI](https://docs.kalshi.com/openapi.yaml), [AsyncAPI](https://docs.kalshi.com/asyncapi.yaml)), vendored unmodified as the contract this client is built against. They are Kalshi's documents and are not covered by this repository's license; Kalshi's own terms apply to them and to use of the API. diff --git a/cmd/example/README.md b/cmd/example/README.md index ddbf95b..fb86683 100644 --- a/cmd/example/README.md +++ b/cmd/example/README.md @@ -26,7 +26,7 @@ KALSHI_ACCESS_KEY=$(cat key_id) KALSHI_PRIVATE_KEY_PATH=./private_key.pem go run |----------|----------|-------------| | `KALSHI_ACCESS_KEY` | for auth | API key ID | | `KALSHI_PRIVATE_KEY_PATH` | for auth | Path to the PEM file | -| `BASE_URL` | no | Default: demo. Production: `https://api.elections.kalshi.com/trade-api/v2` | +| `BASE_URL` | no | Default: demo. Production: `https://external-api.kalshi.com/trade-api/v2` (or the shared `https://api.elections.kalshi.com/trade-api/v2`) | | `LIVE` | no | `1` places two 1¢ bids on the open 15m BTC market (`KXBTC15M`) and cancels the second. Runs against `BASE_URL` (demo unless you set production). Leaves the first order resting. Both orders use fixed `client_order_id`s (`example-resting`, `example-cancel`), so a second run may be rejected with 409. | Without auth, only public endpoints run. diff --git a/cmd/example/websocket_example/README.md b/cmd/example/websocket_example/README.md index 9f0d1d0..469d588 100644 --- a/cmd/example/websocket_example/README.md +++ b/cmd/example/websocket_example/README.md @@ -2,7 +2,7 @@ Subscribe / list / update / unsubscribe / receive. No orders. Writes `ws_calls.log` in the current directory. -Uses the same credentials as the REST example (`KALSHI_ACCESS_KEY`, `KALSHI_PRIVATE_KEY_PATH`). Default is demo (`wss://demo-api.kalshi.co/trade-api/ws/v2`). Demo often returns `websocket: bad handshake`; set `BASE_URL=https://api.elections.kalshi.com/trade-api/v2` (WS URL is derived from `BASE_URL`). +Uses the same credentials as the REST example (`KALSHI_ACCESS_KEY`, `KALSHI_PRIVATE_KEY_PATH`). Default is demo (`wss://demo-api.kalshi.co/trade-api/ws/v2`). Demo often returns `websocket: bad handshake`; set `BASE_URL=https://external-api.kalshi.com/trade-api/v2` for production. The WS URL is derived from `BASE_URL`: `external-api.kalshi.com` maps to `external-api-ws.kalshi.com`; any other host (the shared `api.elections.kalshi.com`, demo) is used as-is. Exercises ticker, orderbook_delta, trade, market_lifecycle_v2, multi-market ticker with `send_initial_snapshot`, `ListSubscriptions`, `add_markets` / `delete_markets`. Does not call `get_snapshot`, Pyth, or CF Benchmarks. diff --git a/cmd/example/websocket_example/main.go b/cmd/example/websocket_example/main.go index d25160b..e5b13a3 100644 --- a/cmd/example/websocket_example/main.go +++ b/cmd/example/websocket_example/main.go @@ -75,7 +75,13 @@ func wsURLFromBase(baseURL string) (host, path string) { if path == u.Path { path = "/trade-api/ws/v2" } - return u.Host, path + host = u.Host + // The recommended production hosts differ per protocol; the shared and + // demo hosts serve both. + if host == "external-api.kalshi.com" { + host = "external-api-ws.kalshi.com" + } + return host, path } func logSection(log *os.File, title string, body string) { diff --git a/oddrip/client.go b/oddrip/client.go index 1560598..8680423 100644 --- a/oddrip/client.go +++ b/oddrip/client.go @@ -15,7 +15,7 @@ import ( "github.com/UTXOnly/oddrip/oddrip/internal/retry" ) -const defaultBaseURL = "https://api.elections.kalshi.com/trade-api/v2" +const defaultBaseURL = "https://external-api.kalshi.com/trade-api/v2" type RetryConfig struct { MaxAttempts int diff --git a/oddrip/client_test.go b/oddrip/client_test.go index 8956dd3..188c7da 100644 --- a/oddrip/client_test.go +++ b/oddrip/client_test.go @@ -11,6 +11,7 @@ import ( "testing" "time" + "github.com/UTXOnly/oddrip/oddrip/internal/auth" "github.com/UTXOnly/oddrip/oddrip/types" ) @@ -963,3 +964,47 @@ func TestRetryPolicy_TransportError(t *testing.T) { t.Fatalf("DecreaseV2 attempts = %d, want 1 (no replay)", got) } } + +func TestDefaultBaseURL(t *testing.T) { + mt := &mockTransport{statusCode: 200, body: []byte(`{"exchange_active":true,"trading_active":true}`)} + client := New(HTTPClient(&http.Client{Transport: mt})) + if _, err := client.Exchange.GetStatus(context.Background()); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got, want := mt.req.URL.String(), "https://external-api.kalshi.com/trade-api/v2/exchange/status"; got != want { + t.Fatalf("request URL = %q, want %q", got, want) + } +} + +// The signed message is timestamp + METHOD + path, so switching hosts (the +// recommended external-api host, the shared api.elections host, demo) must +// not change what is signed. +func TestSignedPathExcludesHost(t *testing.T) { + var signed []string + signer := &auth.KalshiSigner{KeyID: "k", SignRequest: func(method, path string, _ int64) (string, error) { + signed = append(signed, method+" "+path) + return "sig", nil + }} + for _, base := range []string{"", "https://api.elections.kalshi.com/trade-api/v2", "https://demo-api.kalshi.co/trade-api/v2/"} { + mt := &mockTransport{statusCode: 200, body: []byte(`{"balance":0}`)} + opts := []Option{HTTPClient(&http.Client{Transport: mt}), Auth(signer)} + if base != "" { + opts = append(opts, BaseURL(base)) + } + client := New(opts...) + if _, err := client.Portfolio.GetBalance(context.Background(), nil); err != nil { + t.Fatalf("base %q: unexpected error: %v", base, err) + } + if got := mt.req.Header.Get("KALSHI-ACCESS-SIGNATURE"); got != "sig" { + t.Fatalf("base %q: signature header = %q", base, got) + } + } + for i, got := range signed { + if want := "GET /trade-api/v2/portfolio/balance"; got != want { + t.Fatalf("signed[%d] = %q, want %q", i, got, want) + } + } + if len(signed) != 3 { + t.Fatalf("signed %d requests, want 3", len(signed)) + } +} diff --git a/oddrip/concurrent.go b/oddrip/concurrent.go index c4cada3..b2bd6fc 100644 --- a/oddrip/concurrent.go +++ b/oddrip/concurrent.go @@ -1,6 +1,9 @@ package oddrip -import "context" +import ( + "context" + "sync/atomic" +) type ConcurrentResult[T any] struct { Value T @@ -8,32 +11,40 @@ type ConcurrentResult[T any] struct { } // DoConcurrent runs fn for i in [0, n) with at most maxInFlight calls active at -// once (maxInFlight <= 0 means unbounded). Results are index-ordered. If ctx is -// cancelled, the results collected so far are returned along with ctx.Err(). +// once (maxInFlight <= 0 means unbounded). A positive maxInFlight bounds the +// goroutines too: at most min(n, maxInFlight) workers are started, each taking +// the next index when its current call returns, so a large n does not create n +// goroutines. Results are index-ordered. If ctx is cancelled, the results +// collected so far are returned along with ctx.Err(); workers stop taking new +// indices, and a call already inside fn finishes in the background (have fn +// honor ctx to cut it short). func DoConcurrent[T any](ctx context.Context, n, maxInFlight int, fn func(i int) (T, error)) ([]ConcurrentResult[T], error) { results := make([]ConcurrentResult[T], n) type pair struct { i int r ConcurrentResult[T] } - ch := make(chan pair, n) - var sem chan struct{} - if maxInFlight > 0 { - sem = make(chan struct{}, maxInFlight) + workers := n + if maxInFlight > 0 && maxInFlight < n { + workers = maxInFlight } - for i := 0; i < n; i++ { - go func(idx int) { - if sem != nil { + ch := make(chan pair, workers) + var next atomic.Int64 + for w := 0; w < workers; w++ { + go func() { + for { + idx := next.Add(1) - 1 + if idx >= int64(n) || ctx.Err() != nil { + return + } + val, err := fn(int(idx)) select { - case sem <- struct{}{}: - defer func() { <-sem }() + case ch <- pair{int(idx), ConcurrentResult[T]{Value: val, Err: err}}: case <-ctx.Done(): return } } - val, err := fn(idx) - ch <- pair{idx, ConcurrentResult[T]{Value: val, Err: err}} - }(i) + }() } for i := 0; i < n; i++ { select { diff --git a/oddrip/concurrent_test.go b/oddrip/concurrent_test.go index b9647cf..ec238b3 100644 --- a/oddrip/concurrent_test.go +++ b/oddrip/concurrent_test.go @@ -147,3 +147,162 @@ func TestDoConcurrentCancel(t *testing.T) { time.Sleep(5 * time.Millisecond) } } + +// waitUntil polls cond until it holds or the deadline passes. +func waitUntil(t *testing.T, what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for !cond() { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %s", what) + } + time.Sleep(time.Millisecond) + } +} + +// TestDoConcurrentLargeN checks that a bounded maxInFlight bounds the worker +// goroutines as well as the active fn calls (#14): with n=10000 and a limit of +// 4, only the workers exist while fn is blocked, the peak in-flight count is +// the limit, and the dynamically claimed indices still land in order. +func TestDoConcurrentLargeN(t *testing.T) { + before := runtime.NumGoroutine() + const n, limit = 10000, 4 + var inFlight, peak atomic.Int32 + release := make(chan struct{}) + type out struct { + results []ConcurrentResult[int] + err error + } + done := make(chan out, 1) + go func() { + r, err := DoConcurrent(context.Background(), n, limit, func(i int) (int, error) { + cur := inFlight.Add(1) + defer inFlight.Add(-1) + for p := peak.Load(); cur > p && !peak.CompareAndSwap(p, cur); p = peak.Load() { + } + <-release + return i, nil + }) + done <- out{r, err} + }() + + waitUntil(t, "workers to block in fn", func() bool { return inFlight.Load() == limit }) + if now := runtime.NumGoroutine(); now-before >= 100 { + t.Fatalf("goroutines while fn blocked: before=%d now=%d; n=%d with maxInFlight=%d should not start n goroutines", before, now, n, limit) + } + close(release) + + var got out + select { + case got = <-done: + case <-time.After(5 * time.Second): + t.Fatal("DoConcurrent did not return") + } + if got.err != nil { + t.Fatal(got.err) + } + if len(got.results) != n { + t.Fatalf("len(results) = %d, want %d", len(got.results), n) + } + for i, r := range got.results { + if r.Err != nil || r.Value != i { + t.Fatalf("results[%d] = {%d %v}, want {%d nil}", i, r.Value, r.Err, i) + } + } + if got := peak.Load(); got != limit { + t.Fatalf("peak in-flight = %d, want %d", got, limit) + } + waitUntil(t, "workers to exit", func() bool { return runtime.NumGoroutine() <= before+2 }) +} + +// A limit above n starts n workers, not maxInFlight. +func TestDoConcurrentLimitAboveN(t *testing.T) { + before := runtime.NumGoroutine() + const n, limit = 2, 1000 + var inFlight atomic.Int32 + release := make(chan struct{}) + done := make(chan error, 1) + go func() { + _, err := DoConcurrent(context.Background(), n, limit, func(i int) (int, error) { + inFlight.Add(1) + <-release + return i, nil + }) + done <- err + }() + + waitUntil(t, "workers to block in fn", func() bool { return inFlight.Load() == n }) + if now := runtime.NumGoroutine(); now-before > n+1 { + t.Fatalf("goroutines while fn blocked: before=%d now=%d; want at most %d extra", before, now, n+1) + } + close(release) + if err := <-done; err != nil { + t.Fatal(err) + } +} + +// Cancelling while every worker is blocked inside fn (on something other than +// ctx) must return promptly with ctx.Err(). Once fn is released the workers must +// neither block sending a result nobody reads nor take another index. +func TestDoConcurrentCancelWhileBlocked(t *testing.T) { + before := runtime.NumGoroutine() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + const n, limit = 1000, 4 + var entered atomic.Int32 + release := make(chan struct{}) + type out struct { + results []ConcurrentResult[int] + err error + } + done := make(chan out, 1) + go func() { + r, err := DoConcurrent(ctx, n, limit, func(i int) (int, error) { + entered.Add(1) + <-release + return i + 1, nil + }) + done <- out{r, err} + }() + + waitUntil(t, "workers to block in fn", func() bool { return entered.Load() == limit }) + cancel() + + var got out + select { + case got = <-done: + case <-time.After(2 * time.Second): + t.Fatal("DoConcurrent did not return after cancel while fn was blocked") + } + if !errors.Is(got.err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled", got.err) + } + if len(got.results) != n { + t.Fatalf("len(results) = %d, want %d", len(got.results), n) + } + for i, r := range got.results { + if r.Value != 0 || r.Err != nil { + t.Fatalf("results[%d] = {%d %v}, want zero value: nothing completed before cancel", i, r.Value, r.Err) + } + } + + close(release) + waitUntil(t, "workers to exit", func() bool { return runtime.NumGoroutine() <= before+2 }) + if got := entered.Load(); got != limit { + t.Errorf("fn entered %d times, want %d: workers took new indices after cancel", got, limit) + } +} + +func TestDoConcurrentZero(t *testing.T) { + results, err := DoConcurrent(context.Background(), 0, 4, func(i int) (int, error) { + t.Errorf("fn called with i=%d for n=0", i) + return 0, nil + }) + if err != nil { + t.Fatal(err) + } + if len(results) != 0 { + t.Fatalf("len(results) = %d, want 0", len(results)) + } +} diff --git a/oddrip/errors.go b/oddrip/errors.go index 69bbb9f..4996420 100644 --- a/oddrip/errors.go +++ b/oddrip/errors.go @@ -10,15 +10,23 @@ import ( "github.com/UTXOnly/oddrip/oddrip/types" ) -const maxBodySnippet = 512 +const ( + // maxBodySnippet bounds APIError.RawBody. + maxBodySnippet = 512 + // maxErrorBody bounds how much of an error body is read to decode the + // structured fields. A body longer than this is truncated, so it does not + // parse as JSON and the fields stay empty. + maxErrorBody = 64 << 10 +) // APIError is a non-2xx response. StatusCode and RawBody (the first 512 bytes // of the body) are always set. Code, Message, and Details are filled from the // body when it is one of the shapes Kalshi emits: the spec's flat // ErrorResponse, the same object nested under "error" (what production -// returns for most errors), or {"msg": "..."} (parameter-binding 400s). -// RequestID is the Request-Id header when present; production does not -// currently send one. +// returns for most errors), or {"msg": "..."} (parameter-binding 400s). They +// are decoded from up to 64 KiB of the body; bodies over 64 KiB are not +// decoded and leave them empty. RequestID is the Request-Id header when +// present; production does not currently send one. type APIError struct { StatusCode int Code string @@ -47,8 +55,8 @@ type errorBody struct { func newAPIError(resp *http.Response) *APIError { e := &APIError{StatusCode: resp.StatusCode, RequestID: resp.Header.Get("Request-Id")} - buf, _ := io.ReadAll(io.LimitReader(resp.Body, maxBodySnippet)) - e.RawBody = string(buf) + buf, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBody)) + e.RawBody = string(buf[:min(len(buf), maxBodySnippet)]) var body errorBody if json.NewDecoder(bytes.NewReader(buf)).Decode(&body) != nil { return e diff --git a/oddrip/errors_test.go b/oddrip/errors_test.go index 342dbe8..9ff7094 100644 --- a/oddrip/errors_test.go +++ b/oddrip/errors_test.go @@ -118,3 +118,165 @@ func TestNewAPIError_RawBodyTruncatedAndRequestID(t *testing.T) { t.Errorf("RequestID = %q, want req-123", e.RequestID) } } + +// nestedErrorBody returns a production-shape error body padded to exactly n +// bytes with a long details string. +func nestedErrorBody(n int) string { + const head = `{"error":{"code":"invalid_parameters","message":"invalid parameters","details":"` + const tail = `"}}` + return head + strings.Repeat("d", n-len(head)-len(tail)) + tail +} + +func TestNewAPIError_LongBodyDecodesStructuredFields(t *testing.T) { + long := strings.Repeat("d", maxBodySnippet*2) + cases := []struct { + name string + body string + wantCode string + wantMessage string + wantDetails string + }{ + { + name: "spec flat shape", + body: `{"code":"invalid_parameters","message":"invalid parameters","details":"` + long + `"}`, + wantCode: "invalid_parameters", + wantMessage: "invalid parameters", + wantDetails: long, + }, + { + name: "production nested under error", + body: `{"error":{"code":"invalid_parameters","message":"invalid parameters","details":"` + long + `"}}`, + wantCode: "invalid_parameters", + wantMessage: "invalid parameters", + wantDetails: long, + }, + { + // Gateway metadata ahead of the error object pushes code and + // message past the RawBody snippet entirely. + name: "nested after long metadata", + body: `{"trace":"` + long + `","error":{"code":"not_found","message":"not found"}}`, + wantCode: "not_found", + wantMessage: "not found", + }, + { + name: "parameter binding msg shape", + body: `{"msg":"` + long + `"}`, + wantMessage: long, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if len(tc.body) <= maxBodySnippet { + t.Fatalf("fixture is %d bytes, must exceed %d", len(tc.body), maxBodySnippet) + } + resp := &http.Response{ + StatusCode: 400, + Header: http.Header{}, + Body: io.NopCloser(strings.NewReader(tc.body)), + } + e := newAPIError(resp) + if e.Code != tc.wantCode { + t.Errorf("Code = %q, want %q", e.Code, tc.wantCode) + } + if e.Message != tc.wantMessage { + t.Errorf("Message = %q, want %q", e.Message, tc.wantMessage) + } + if e.Details != tc.wantDetails { + t.Errorf("Details length = %d, want %d", len(e.Details), len(tc.wantDetails)) + } + if len(e.RawBody) != maxBodySnippet { + t.Errorf("RawBody length = %d, want %d", len(e.RawBody), maxBodySnippet) + } + if !strings.HasPrefix(tc.body, e.RawBody) { + t.Errorf("RawBody is not a prefix of the body") + } + }) + } +} + +func TestNewAPIError_MalformedLongBody(t *testing.T) { + full := nestedErrorBody(maxBodySnippet * 2) + cases := []struct { + name string + body string + }{ + {name: "truncated json", body: full[:len(full)-3]}, + {name: "not json", body: "" + strings.Repeat("x", maxBodySnippet*2) + ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + resp := &http.Response{ + StatusCode: 502, + Header: http.Header{}, + Body: io.NopCloser(strings.NewReader(tc.body)), + } + e := newAPIError(resp) + if e.StatusCode != 502 { + t.Errorf("StatusCode = %d, want 502", e.StatusCode) + } + if e.Code != "" || e.Message != "" || e.Details != "" || e.Service != "" { + t.Errorf("structured fields set from malformed body: %+v", e) + } + if len(e.RawBody) != maxBodySnippet { + t.Errorf("RawBody length = %d, want %d", len(e.RawBody), maxBodySnippet) + } + if !strings.HasPrefix(tc.body, e.RawBody) { + t.Errorf("RawBody is not a prefix of the body") + } + if got := e.Error(); got != "api error 502" { + t.Errorf("Error() = %q, want %q", got, "api error 502") + } + }) + } +} + +// countingReader records how many bytes newAPIError pulls from the body. +type countingReader struct { + r io.Reader + n int +} + +func (c *countingReader) Read(p []byte) (int, error) { + n, err := c.r.Read(p) + c.n += n + return n, err +} + +func TestNewAPIError_ErrorBodyReadIsBounded(t *testing.T) { + cases := []struct { + name string + size int + wantMessage string + }{ + {name: "at limit decodes", size: maxErrorBody, wantMessage: "invalid parameters"}, + {name: "one byte over is not decoded", size: maxErrorBody + 1}, + {name: "far over is not decoded", size: 4 * maxErrorBody}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + body := nestedErrorBody(tc.size) + if len(body) != tc.size { + t.Fatalf("fixture is %d bytes, want %d", len(body), tc.size) + } + cr := &countingReader{r: strings.NewReader(body)} + resp := &http.Response{ + StatusCode: 500, + Header: http.Header{}, + Body: io.NopCloser(cr), + } + e := newAPIError(resp) + if cr.n > maxErrorBody { + t.Errorf("read %d bytes from the body, want at most %d", cr.n, maxErrorBody) + } + if e.Message != tc.wantMessage { + t.Errorf("Message = %q, want %q", e.Message, tc.wantMessage) + } + if len(e.RawBody) != maxBodySnippet { + t.Errorf("RawBody length = %d, want %d", len(e.RawBody), maxBodySnippet) + } + if !strings.HasPrefix(body, e.RawBody) { + t.Errorf("RawBody is not a prefix of the body") + } + }) + } +} diff --git a/oddrip/version.go b/oddrip/version.go index 25874ea..ecca44f 100644 --- a/oddrip/version.go +++ b/oddrip/version.go @@ -1,4 +1,4 @@ package oddrip // Version is the module release. CI tags v and publishes a release when main carries a version that is not yet tagged. -const Version = "0.6.1" +const Version = "0.6.2" diff --git a/oddrip/ws.go b/oddrip/ws.go index 346725e..e3c5061 100644 --- a/oddrip/ws.go +++ b/oddrip/ws.go @@ -17,13 +17,18 @@ import ( "github.com/UTXOnly/oddrip/oddrip/types" ) -const defaultWSHost = "api.elections.kalshi.com" +const defaultWSHost = "external-api-ws.kalshi.com" const defaultWSPath = "/trade-api/ws/v2" const ( defaultWSBufferSize = 4096 defaultWSPingInterval = 30 * time.Second defaultWSReadTimeout = 90 * time.Second + defaultWSWriteTimeout = 10 * time.Second + // wsCloseWait bounds how long Close waits for the read loop after the + // socket is closed. The read loop exits as soon as the socket does, so + // this is a backstop, not an expected wait. + wsCloseWait = 5 * time.Second ) var ( @@ -33,24 +38,36 @@ var ( // ErrWSMalformedFrame is the terminal error when a text frame is not valid // JSON; Err() wraps it with the decode error. ErrWSMalformedFrame = errors.New("websocket malformed frame") + // ErrWSWriteTimeout is the terminal error when a command frame could not + // be written within WSWriteTimeout or the caller's context deadline, + // whichever came first; Err() wraps it with the socket error. A timed-out + // write leaves the socket unusable, so the connection is failed and + // Messages() closes, the same as ErrWSSlowConsumer. + ErrWSWriteTimeout = errors.New("websocket write timeout") ) type WSConn struct { - conn *websocket.Conn - auth AuthProvider - host string - path string - readTimeout time.Duration - nextID atomic.Int64 - mu sync.Mutex - closed bool - readErr error - writeMu sync.Mutex - pendMu sync.Mutex - pending map[int]chan *wsEnvelope // command replies, by command id - snapshots map[int][]chan *wsEnvelope // get_snapshot waiters, by sid - msgChan chan *types.WSMessage - readDone chan struct{} + conn *websocket.Conn + auth AuthProvider + host string + path string + readTimeout time.Duration + writeTimeout time.Duration + nextID atomic.Int64 + mu sync.Mutex + closed bool + readErr error + // writeSem serializes data-frame writes (gorilla allows one WriteMessage + // at a time). It is a 1-slot channel rather than a mutex so a waiter can + // give up when its context ends or the connection dies. Control frames + // (ping, pong, close) go through WriteControl, which gorilla serializes + // internally and which is safe alongside a data write in progress. + writeSem chan struct{} + pendMu sync.Mutex + pending map[int]chan *wsEnvelope // command replies, by command id + snapshots map[int][]chan *wsEnvelope // get_snapshot waiters, by sid + msgChan chan *types.WSMessage + readDone chan struct{} } type wsEnvelope struct { @@ -70,6 +87,7 @@ type wsOpts struct { bufferSize int pingInterval time.Duration readTimeout time.Duration + writeTimeout time.Duration } func WSScheme(scheme string) WSOption { @@ -113,10 +131,19 @@ func WSReadTimeout(d time.Duration) WSOption { } } -func (c *Client) ConnectWS(ctx context.Context, opts ...WSOption) (*WSConn, error) { - if c.auth == nil { - return nil, ErrWSAuthRequired +// WSWriteTimeout bounds every socket write. A command frame that cannot be +// written within this long (or the caller's context deadline, if sooner) +// fails the connection with an error wrapping ErrWSWriteTimeout; the close +// frame sent by Close is bounded by it too. Writes are never unbounded: +// <= 0 uses the default of 10s. +func WSWriteTimeout(d time.Duration) WSOption { + return func(o *wsOpts) { + o.writeTimeout = d } +} + +// wsConfig applies opts over the defaults and normalizes them. +func wsConfig(opts []WSOption) wsOpts { cfg := wsOpts{ scheme: "wss", host: defaultWSHost, @@ -124,6 +151,7 @@ func (c *Client) ConnectWS(ctx context.Context, opts ...WSOption) (*WSConn, erro bufferSize: defaultWSBufferSize, pingInterval: defaultWSPingInterval, readTimeout: defaultWSReadTimeout, + writeTimeout: defaultWSWriteTimeout, } for _, o := range opts { o(&cfg) @@ -134,8 +162,24 @@ func (c *Client) ConnectWS(ctx context.Context, opts ...WSOption) (*WSConn, erro if cfg.bufferSize <= 0 { cfg.bufferSize = defaultWSBufferSize } - u := url.URL{Scheme: cfg.scheme, Host: cfg.host, Path: cfg.path} - req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil) + if cfg.writeTimeout <= 0 { + cfg.writeTimeout = defaultWSWriteTimeout + } + return cfg +} + +// url is the dial destination. +func (o wsOpts) url() string { + return (&url.URL{Scheme: o.scheme, Host: o.host, Path: o.path}).String() +} + +func (c *Client) ConnectWS(ctx context.Context, opts ...WSOption) (*WSConn, error) { + if c.auth == nil { + return nil, ErrWSAuthRequired + } + cfg := wsConfig(opts) + u := cfg.url() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) if err != nil { return nil, err } @@ -145,20 +189,22 @@ func (c *Client) ConnectWS(ctx context.Context, opts ...WSOption) (*WSConn, erro dialer := websocket.Dialer{ HandshakeTimeout: 10 * time.Second, } - conn, _, err := dialer.DialContext(ctx, u.String(), req.Header) + conn, _, err := dialer.DialContext(ctx, u, req.Header) if err != nil { return nil, fmt.Errorf("ws dial: %w", err) } ws := &WSConn{ - conn: conn, - auth: c.auth, - host: cfg.host, - path: cfg.path, - readTimeout: cfg.readTimeout, - pending: make(map[int]chan *wsEnvelope), - snapshots: make(map[int][]chan *wsEnvelope), - msgChan: make(chan *types.WSMessage, cfg.bufferSize), - readDone: make(chan struct{}), + conn: conn, + auth: c.auth, + host: cfg.host, + path: cfg.path, + readTimeout: cfg.readTimeout, + writeTimeout: cfg.writeTimeout, + writeSem: make(chan struct{}, 1), + pending: make(map[int]chan *wsEnvelope), + snapshots: make(map[int][]chan *wsEnvelope), + msgChan: make(chan *types.WSMessage, cfg.bufferSize), + readDone: make(chan struct{}), } ws.nextID.Store(1) ws.resetDeadline() @@ -288,6 +334,9 @@ func (ws *WSConn) nextIDVal() int { // orderbook_snapshot frame for that sid, which is how get_snapshot is answered. // An error reply ends the wait; the replies collected before it are returned // alongside the *WSError so callers can report partial success. +// +// The write is bounded by ctx and writeTimeout (see write); nothing here waits +// past the caller's deadline. func (ws *WSConn) sendAndWait(ctx context.Context, id int, payload interface{}, expectCount int, snapshotSID int) ([]*wsEnvelope, error) { data, err := json.Marshal(payload) if err != nil { @@ -318,13 +367,7 @@ func (ws *WSConn) sendAndWait(ctx context.Context, id int, payload interface{}, ws.pendMu.Unlock() }() - ws.writeMu.Lock() - err = ws.conn.WriteMessage(websocket.TextMessage, data) - ws.writeMu.Unlock() - if err != nil { - if e := ws.Err(); e != nil { - return nil, e - } + if err := ws.write(ctx, data); err != nil { return nil, err } var out []*wsEnvelope @@ -358,6 +401,63 @@ func (ws *WSConn) sendAndWait(ctx context.Context, id int, payload interface{}, } } +// write sends one text frame. Waiting for the write slot is bounded by ctx +// (and by the connection dying); the socket write itself by the sooner of the +// ctx deadline and writeTimeout. +// +// A caller whose ctx ends while another command holds the slot gets ctx.Err() +// and the connection is untouched: its frame never started. Once a frame is +// being written, a timeout from either bound leaves gorilla's writer state +// corrupt (every later write would fail), so the connection is failed with an +// error wrapping ErrWSWriteTimeout; Messages() closes and Err() reports it. +// The caller gets ctx.Err() if its own deadline was the cause, else that +// error. Other write errors are returned as-is: the read loop surfaces the +// socket failure behind them on its own. A timeout is the case it cannot see, +// since a peer that has stopped reading may keep sending. +func (ws *WSConn) write(ctx context.Context, data []byte) error { + if err := ctx.Err(); err != nil { + return err + } + select { + case ws.writeSem <- struct{}{}: + case <-ctx.Done(): + return ctx.Err() + case <-ws.readDone: + return ws.closedErr() + } + defer func() { <-ws.writeSem }() + + deadline := time.Now().Add(ws.writeTimeout) + ctxBound := false + if d, ok := ctx.Deadline(); ok && d.Before(deadline) { + deadline, ctxBound = d, true + } + ws.conn.SetWriteDeadline(deadline) + err := ws.conn.WriteMessage(websocket.TextMessage, data) + if err == nil { + return nil + } + if e := ws.Err(); e != nil { + return e + } + var ne net.Error + if errors.As(err, &ne) && ne.Timeout() { + ws.setErr(fmt.Errorf("%w: %v", ErrWSWriteTimeout, err)) + ws.conn.Close() + if cerr := ctx.Err(); cerr != nil { + return cerr + } + if ctxBound { + // The socket deadline was the caller's. Its context timer can + // fire a moment after the socket's, so ctx.Err() may still be + // nil here; the write was cut at that deadline regardless. + return context.DeadlineExceeded + } + return ws.Err() + } + return err +} + // removeSnapshotWaiter must be called with pendMu held. func (ws *WSConn) removeSnapshotWaiter(sid int, ch chan *wsEnvelope) { waiters := ws.snapshots[sid] @@ -527,13 +627,22 @@ func (ws *WSConn) Done() <-chan struct{} { // Err is nil while the connection is healthy. After the read loop exits it is // the terminal read error, ErrWSSlowConsumer, an error wrapping -// ErrWSMalformedFrame, or ErrWSClosed after Close(). +// ErrWSMalformedFrame or ErrWSWriteTimeout, or ErrWSClosed after Close(). func (ws *WSConn) Err() error { ws.mu.Lock() defer ws.mu.Unlock() return ws.readErr } +// Close sends a close frame, closes the socket, and waits for the read loop +// to exit. It is idempotent, and pending commands return ErrWSClosed. +// +// Close is bounded even when the peer has stopped reading: it returns within +// about WSWriteTimeout plus five seconds, and in practice as soon as the +// close frame is written or skipped. The close frame is a control write with +// its own deadline, so Close never queues behind a command write; if one is +// in progress the frame is skipped (a jammed socket would not deliver it +// anyway) and the socket is closed directly, which also unblocks that write. func (ws *WSConn) Close() error { ws.mu.Lock() if ws.closed { @@ -546,21 +655,24 @@ func (ws *WSConn) Close() error { ws.readErr = ErrWSClosed } ws.mu.Unlock() - if !healthy { - ws.conn.Close() - <-ws.readDone - return nil + var err error + if healthy { + select { + case ws.writeSem <- struct{}{}: + err = ws.conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""), time.Now().Add(ws.writeTimeout)) + <-ws.writeSem + default: + // A command write is in flight and bounded by its own deadline; + // do not wait for it. + } } - ws.writeMu.Lock() - err := ws.conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "")) - ws.writeMu.Unlock() - if e := ws.conn.Close(); e != nil && err == nil { + if e := ws.conn.Close(); e != nil && err == nil && healthy { err = e } select { case <-ws.readDone: return err - case <-time.After(5 * time.Second): + case <-time.After(wsCloseWait): return fmt.Errorf("close: read loop did not exit: %w", err) } } diff --git a/oddrip/ws_test.go b/oddrip/ws_test.go index 2570994..449a5ad 100644 --- a/oddrip/ws_test.go +++ b/oddrip/ws_test.go @@ -9,6 +9,7 @@ import ( "net/http" "net/http/httptest" "net/url" + "strings" "sync" "testing" "time" @@ -89,6 +90,16 @@ func TestConnectWS_Subscribe_Integration(t *testing.T) { } } +func TestConnectWS_DefaultURL(t *testing.T) { + if got, want := wsConfig(nil).url(), "wss://external-api-ws.kalshi.com/trade-api/ws/v2"; got != want { + t.Fatalf("default dial URL = %q, want %q", got, want) + } + // The shared host that was the default through 0.6.1 is an override away. + if got, want := wsConfig([]WSOption{WSHost("api.elections.kalshi.com")}).url(), "wss://api.elections.kalshi.com/trade-api/ws/v2"; got != want { + t.Fatalf("WSHost override URL = %q, want %q", got, want) + } +} + func TestConnectWS_DialFails(t *testing.T) { client := New(Auth(&mockWSAuth{})) ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) @@ -751,6 +762,347 @@ func TestWS_MalformedFrame_FailsConnection(t *testing.T) { } } +func TestWSWriteTimeout_Option(t *testing.T) { + for _, tc := range []struct { + opt time.Duration + want time.Duration + }{ + {0, defaultWSWriteTimeout}, + {-1, defaultWSWriteTimeout}, + {3 * time.Second, 3 * time.Second}, + } { + ws := wsTestConnect(t, wsTestServer(t, wsDrain), WSWriteTimeout(tc.opt)) + if ws.writeTimeout != tc.want { + t.Errorf("WSWriteTimeout(%v): writeTimeout = %v, want %v", tc.opt, ws.writeTimeout, tc.want) + } + } + ws := wsTestConnect(t, wsTestServer(t, wsDrain)) + if ws.writeTimeout != defaultWSWriteTimeout { + t.Errorf("default writeTimeout = %v, want %v", ws.writeTimeout, defaultWSWriteTimeout) + } +} + +// A caller whose context ends while another command holds the write slot gets +// ctx.Err() without its frame ever starting, and the connection stays healthy +// for the next command. +func TestWS_Write_ContextEndsWaitingForSlot(t *testing.T) { + ws := wsTestConnect(t, wsTestServer(t, wsSubscribeEcho)) + params := types.SubscribeParams{Channels: []string{types.WSChannelTicker}} + ws.writeSem <- struct{}{} // another command is mid-write + + cancelled, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := ws.Subscribe(cancelled, params); err != context.Canceled { + t.Fatalf("Subscribe with cancelled ctx = %v, want context.Canceled", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + start := time.Now() + _, err := ws.Subscribe(ctx, params) + if err != context.DeadlineExceeded { + t.Fatalf("Subscribe while slot held = %v, want context.DeadlineExceeded", err) + } + if d := time.Since(start); d > time.Second { + t.Errorf("Subscribe returned after %v, want promptly after its 50ms deadline", d) + } + if err := ws.Err(); err != nil { + t.Fatalf("Err() = %v, want nil: giving up on the slot must not fail the connection", err) + } + ws.pendMu.Lock() + n := len(ws.pending) + ws.pendMu.Unlock() + if n != 0 { + t.Errorf("waiters left registered: %d", n) + } + + <-ws.writeSem + subs, err := ws.Subscribe(wsTestCtx(t), params) + if err != nil || len(subs) != 1 { + t.Fatalf("Subscribe after slot released: subs=%+v err=%v", subs, err) + } +} + +// Cancelled and live callers contending for the write slot: every live call +// completes and every cancelled one returns ctx.Err() without being sent. +func TestWS_Subscribe_ConcurrentWithCancelled(t *testing.T) { + ws := wsTestConnect(t, wsTestServer(t, wsSubscribeEcho)) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + cancelled, cancelNow := context.WithCancel(ctx) + cancelNow() + + const n = 20 + errs := make([]error, n) + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + c := ctx + if i%2 == 1 { + c = cancelled + } + subs, err := ws.Subscribe(c, types.SubscribeParams{Channels: []string{fmt.Sprintf("ch%d", i)}}) + if err == nil && len(subs) != 1 { + err = fmt.Errorf("got %d subscribed", len(subs)) + } + errs[i] = err + }(i) + } + wg.Wait() + for i, err := range errs { + if i%2 == 1 { + if err != context.Canceled { + t.Errorf("cancelled Subscribe %d = %v, want context.Canceled", i, err) + } + } else if err != nil { + t.Errorf("Subscribe %d: %v", i, err) + } + } + if err := ws.Err(); err != nil { + t.Errorf("Err() = %v, want nil", err) + } +} + +// A write that cannot complete within WSWriteTimeout fails the connection: +// the command returns an error wrapping ErrWSWriteTimeout, Err() reports it, +// and Messages() closes, the same path as a slow consumer. +func TestWS_WriteTimeout_FailsConnection(t *testing.T) { + ws := wsTestConnect(t, wsNoReadServer(t), WSWriteTimeout(200*time.Millisecond)) + wsJamSocket(t, ws) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + start := time.Now() + _, err := ws.Subscribe(ctx, wsJamParams()) + if !errors.Is(err, ErrWSWriteTimeout) { + t.Fatalf("Subscribe = %v, want ErrWSWriteTimeout", err) + } + if err == ErrWSWriteTimeout { + t.Fatalf("Subscribe should wrap the socket error, got bare sentinel") + } + if d := time.Since(start); d > 3*time.Second { + t.Errorf("write failed after %v with a 200ms write timeout", d) + } + select { + case <-ws.Done(): + case <-time.After(2 * time.Second): + t.Fatal("Done() did not close") + } + if e := ws.Err(); !errors.Is(e, ErrWSWriteTimeout) || e == ErrWSWriteTimeout { + t.Fatalf("Err() = %v, want wrapped ErrWSWriteTimeout", e) + } + if _, ok := <-ws.Messages(); ok { + t.Error("Messages() not closed") + } + start = time.Now() + if _, serr := ws.Subscribe(wsTestCtx(t), types.SubscribeParams{Channels: []string{types.WSChannelTicker}}); !errors.Is(serr, ErrWSWriteTimeout) { + t.Errorf("Subscribe after failure = %v, want ErrWSWriteTimeout", serr) + } + if d := time.Since(start); d > time.Second { + t.Errorf("Subscribe after failure took %v, want immediate", d) + } + if err := ws.Close(); err != nil { + t.Errorf("Close: %v", err) + } + if e := ws.Err(); !errors.Is(e, ErrWSWriteTimeout) { + t.Errorf("Err() after Close = %v, want ErrWSWriteTimeout", e) + } +} + +// The caller's context deadline bounds the write too: with the default write +// timeout, a command with a short deadline against a peer that is not reading +// returns ctx.Err() at that deadline. The frame was cut off mid-write, which +// leaves the socket unusable, so the connection is failed as well. +func TestWS_WriteTimeout_CallerDeadline(t *testing.T) { + ws := wsTestConnect(t, wsNoReadServer(t)) + wsJamSocket(t, ws) + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel() + + start := time.Now() + _, err := ws.Subscribe(ctx, wsJamParams()) + if err != context.DeadlineExceeded { + t.Fatalf("Subscribe = %v, want context.DeadlineExceeded", err) + } + if d := time.Since(start); d > 3*time.Second { + t.Errorf("Subscribe returned after %v with a 200ms deadline", d) + } + select { + case <-ws.Done(): + case <-time.After(2 * time.Second): + t.Fatal("Done() did not close") + } + if e := ws.Err(); !errors.Is(e, ErrWSWriteTimeout) { + t.Fatalf("Err() = %v, want ErrWSWriteTimeout", e) + } +} + +// While one command's write is blocked in the socket, another command with a +// shorter deadline gives up on the write slot at that deadline instead of +// queueing behind the blocked write until it times out. +func TestWS_Write_BlockedWriterDoesNotHoldOthers(t *testing.T) { + ws := wsTestConnect(t, wsNoReadServer(t), WSWriteTimeout(time.Second)) + wsJamSocket(t, ws) + + first := make(chan error, 1) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _, err := ws.Subscribe(ctx, wsJamParams()) + first <- err + }() + wsWaitSlotHeld(t, ws) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + start := time.Now() + _, err := ws.Subscribe(ctx, types.SubscribeParams{Channels: []string{types.WSChannelTicker}}) + if err != context.DeadlineExceeded { + t.Fatalf("second Subscribe = %v, want context.DeadlineExceeded", err) + } + if d := time.Since(start); d > time.Second { + t.Errorf("second Subscribe returned after %v, want promptly after its 100ms deadline", d) + } + select { + case err := <-first: + if !errors.Is(err, ErrWSWriteTimeout) { + t.Fatalf("blocked Subscribe = %v, want ErrWSWriteTimeout", err) + } + case <-time.After(5 * time.Second): + t.Fatal("blocked Subscribe did not return") + } +} + +// Close does not wait for a command write that is stuck in the socket: it +// skips the close frame, closes the socket, and that write returns. +func TestWS_Close_BoundedWithStuckWriter(t *testing.T) { + ws := wsTestConnect(t, wsNoReadServer(t), WSWriteTimeout(2*time.Second)) + wsJamSocket(t, ws) + + stuck := make(chan error, 1) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _, err := ws.Subscribe(ctx, wsJamParams()) + stuck <- err + }() + wsWaitSlotHeld(t, ws) + + start := time.Now() + if err := ws.Close(); err != nil { + t.Errorf("Close: %v", err) + } + if d := time.Since(start); d > time.Second { + t.Errorf("Close took %v with a command write stuck", d) + } + select { + case err := <-stuck: + if err != ErrWSClosed { + t.Errorf("stuck Subscribe after Close = %v, want ErrWSClosed", err) + } + case <-time.After(2 * time.Second): + t.Fatal("stuck Subscribe did not return after Close") + } + select { + case <-ws.Done(): + default: + t.Error("Done() not closed") + } + if err := ws.Err(); err != ErrWSClosed { + t.Errorf("Err() = %v, want ErrWSClosed", err) + } +} + +// With the socket already full and no command in flight, the close frame +// itself cannot be delivered; Close still returns within the write timeout. +func TestWS_Close_BoundedWhenSocketFull(t *testing.T) { + ws := wsTestConnect(t, wsNoReadServer(t), WSWriteTimeout(200*time.Millisecond)) + wsJamSocket(t, ws) + + start := time.Now() + err := ws.Close() + if d := time.Since(start); d > 2*time.Second { + t.Errorf("Close took %v with the socket full and a 200ms write timeout", d) + } + // The close frame is a few bytes; whether it squeezes into the jammed + // send buffer depends on kernel drain timing. Either it fits (nil) or + // its control write hits the 200ms deadline; anything else is a bug. + var ne net.Error + if err != nil && (!errors.As(err, &ne) || !ne.Timeout()) { + t.Errorf("Close = %v, want nil or the close frame's write timeout", err) + } + select { + case <-ws.Done(): + default: + t.Error("Done() not closed") + } + if err := ws.Err(); err != ErrWSClosed { + t.Errorf("Err() = %v, want ErrWSClosed", err) + } + if err := ws.Close(); err != nil { + t.Errorf("second Close: %v", err) + } +} + +// wsJamSocket fills the socket underneath gorilla: it writes raw bytes until +// the kernel stops accepting them, so the next frame write blocks. Loopback +// buffers vary by platform (Windows autotunes the receive window to 16MB), so +// the fill runs until a write times out rather than to a fixed size. The peer +// never reads, so it does not matter that these bytes are not frames. +func wsJamSocket(t *testing.T, ws *WSConn) { + t.Helper() + nc := ws.conn.NetConn() + buf := make([]byte, 1<<20) + giveUp := time.Now().Add(10 * time.Second) + for { + nc.SetWriteDeadline(time.Now().Add(100 * time.Millisecond)) + _, err := nc.Write(buf) + var ne net.Error + if errors.As(err, &ne) && ne.Timeout() { + break + } + if err != nil { + t.Fatalf("jam write: %v", err) + } + if time.Now().After(giveUp) { + t.Fatal("socket never filled") + } + } + nc.SetWriteDeadline(time.Time{}) +} + +// wsJamParams is a command large enough that it cannot slip into whatever +// room the kernel frees after wsJamSocket, yet cheap to marshal — the payload +// is built before the write, so its cost must not eat a caller's deadline. +func wsJamParams() types.SubscribeParams { + return types.SubscribeParams{ + Channels: []string{types.WSChannelTicker}, + MarketTickers: []string{strings.Repeat("x", 1<<20)}, + } +} + +// wsNoReadServer accepts the connection and never reads from it. +func wsNoReadServer(t *testing.T) *httptest.Server { + t.Helper() + block := make(chan struct{}) + t.Cleanup(func() { close(block) }) + return wsTestServer(t, func(*websocket.Conn) { <-block }) +} + +// wsWaitSlotHeld returns once a command holds the write slot. +func wsWaitSlotHeld(t *testing.T, ws *WSConn) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for len(ws.writeSem) == 0 { + if time.Now().After(deadline) { + t.Fatal("no command took the write slot") + } + time.Sleep(time.Millisecond) + } +} + func wsTestServer(t *testing.T, handler func(*websocket.Conn)) *httptest.Server { t.Helper() upgrader := websocket.Upgrader{}