From 8d08a97b9168ed6f33b428c2f359ff569ee3217c Mon Sep 17 00:00:00 2001 From: Marketen Date: Thu, 15 Oct 2026 10:10:37 +0200 Subject: [PATCH] fix: no limit on response duration, only stall guards The Gateway cut every response at five minutes: an absolute server write deadline, plus total-duration deadlines on provider requests. Healthy long answers and streams ended mid-response. - The server has no absolute write deadline. Each client read and write is bounded on its own (CLIENT_IO_TIMEOUT_SECONDS, default 60), in chunks of at most 32 KiB with the deadline refreshed on progress, so a long transfer that keeps moving is never cut. Request headers keep a 30 s limit. - Provider requests have no total deadline. They end only when the provider sends nothing, neither headers nor bytes, for PROVIDER_SILENCE_TIMEOUT_SECONDS (default 900). Keepalive bytes count as activity. Tinfoil's verified transport keeps its own client; only the wait for headers is bounded around it. - Both settings must be positive whole seconds. Co-Authored-By: Claude Opus 5.5 --- README.md | 7 ++ ...fidential_chat_completions_handler_test.go | 44 ++++++++++++ .../adapters/http/middleware/read_timeout.go | 44 ++++++++++++ .../http/middleware/read_timeout_test.go | 47 ++++++++++++ .../adapters/http/middleware/write_timeout.go | 64 +++++++++++++++++ .../http/middleware/write_timeout_test.go | 37 ++++++++++ .../providers/openai/activity_timeout_test.go | 27 +++++++ .../adapters/providers/openai/client.go | 6 +- .../providers/stream/header_timeout.go | 55 ++++++++++++++ .../providers/stream/header_timeout_test.go | 55 ++++++++++++++ .../adapters/providers/stream/idle_timeout.go | 71 +++++++++++++++++++ .../providers/stream/idle_timeout_test.go | 40 +++++++++++ .../adapters/providers/tinfoil/adapter.go | 13 ++-- apps/gateway/main.go | 19 +++-- apps/gateway/timeouts.go | 34 +++++++++ apps/gateway/timeouts_test.go | 19 +++++ 16 files changed, 565 insertions(+), 17 deletions(-) create mode 100644 apps/gateway/internal/adapters/http/middleware/read_timeout.go create mode 100644 apps/gateway/internal/adapters/http/middleware/read_timeout_test.go create mode 100644 apps/gateway/internal/adapters/http/middleware/write_timeout.go create mode 100644 apps/gateway/internal/adapters/http/middleware/write_timeout_test.go create mode 100644 apps/gateway/internal/adapters/providers/openai/activity_timeout_test.go create mode 100644 apps/gateway/internal/adapters/providers/stream/header_timeout.go create mode 100644 apps/gateway/internal/adapters/providers/stream/header_timeout_test.go create mode 100644 apps/gateway/internal/adapters/providers/stream/idle_timeout.go create mode 100644 apps/gateway/internal/adapters/providers/stream/idle_timeout_test.go create mode 100644 apps/gateway/timeouts.go create mode 100644 apps/gateway/timeouts_test.go diff --git a/README.md b/README.md index ed78b47..0f89ba7 100644 --- a/README.md +++ b/README.md @@ -88,6 +88,13 @@ keep a valid `Retry-After` header. Errors include `origin` (`gateway` or details are not exposed. A configured fallback provider is tried only after a rate limit or a server-side failure, never after a rejected request. +Responses have no limit on total duration. Two stall guards apply instead: +`PROVIDER_SILENCE_TIMEOUT_SECONDS` (default 900) ends a request when the +provider sends nothing, neither headers nor bytes, for that long, and +`CLIENT_IO_TIMEOUT_SECONDS` (default 60) ends one when a single read from or +write to the client is blocked for that long. Both must be positive whole +seconds. + ## Privacy and usage controls Nexus provides per-key PII masking settings and access to privacy-focused diff --git a/apps/gateway/internal/adapters/http/handlers/confidential_chat_completions_handler_test.go b/apps/gateway/internal/adapters/http/handlers/confidential_chat_completions_handler_test.go index 15d8e32..71ce782 100644 --- a/apps/gateway/internal/adapters/http/handlers/confidential_chat_completions_handler_test.go +++ b/apps/gateway/internal/adapters/http/handlers/confidential_chat_completions_handler_test.go @@ -11,6 +11,7 @@ import ( "testing" "time" + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/http/middleware" "github.com/google/uuid" "github.com/tinfoilsh/encrypted-http-body-protocol/identity" "github.com/tinfoilsh/encrypted-http-body-protocol/protocol" @@ -108,6 +109,49 @@ func TestConfidentialChatCompletionsHandler_StreamsEncryptedResponse(t *testing. } } +func TestConfidentialStreamRefreshesNetworkWriteDeadline(t *testing.T) { + next := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + for i := 0; i < 6; i++ { + _, _ = io.WriteString(w, "data: token\n\n") + w.(http.Flusher).Flush() + time.Sleep(30 * time.Millisecond) + } + _, _ = io.WriteString(w, "data: [DONE]\n\n") + w.(http.Flusher).Flush() + }) + handler, clientIdentity := newConfidentialTestHandler(t, next) + server := httptest.NewUnstartedServer(middleware.WriteTimeout(time.Second)(http.HandlerFunc(handler.Handle))) + server.Config.WriteTimeout = 50 * time.Millisecond + server.Start() + defer server.Close() + request, requestContext, _ := newEncryptedConfidentialRequest(t, clientIdentity, envelopeAt( + time.Now(), uuid.NewString(), json.RawMessage(`{"model":"test","stream":true,"messages":[]}`), + )) + request.URL.Scheme, request.URL.Host = "http", server.Listener.Addr().String() + request.RequestURI = "" + response, err := server.Client().Do(request) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + ciphertext, err := io.ReadAll(response.Body) + if err != nil { + t.Fatalf("encrypted stream ended early: %v", err) + } + if bytes.Contains(ciphertext, []byte("data: token")) { + t.Fatal("encrypted stream exposed plaintext") + } + recorder := httptest.NewRecorder() + recorder.HeaderMap = response.Header.Clone() + recorder.Code = response.StatusCode + _, _ = recorder.Body.Write(ciphertext) + plaintext := decryptConfidentialResponse(t, recorder, requestContext) + if got, want := string(plaintext), strings.Repeat("data: token\n\n", 6)+"data: [DONE]\n\n"; got != want { + t.Fatalf("stream = %q, want %q", got, want) + } +} + func TestConfidentialChatCompletionsHandler_EncryptsAuthenticationErrors(t *testing.T) { next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if _, err := ExtractBearerToken(r); err != nil { diff --git a/apps/gateway/internal/adapters/http/middleware/read_timeout.go b/apps/gateway/internal/adapters/http/middleware/read_timeout.go new file mode 100644 index 0000000..d57660f --- /dev/null +++ b/apps/gateway/internal/adapters/http/middleware/read_timeout.go @@ -0,0 +1,44 @@ +package middleware + +import ( + "io" + "net/http" + "time" +) + +// ReadTimeout bounds each upload read, not the total upload or inference time. +// Header reading remains bounded separately by http.Server.ReadHeaderTimeout. +func ReadTimeout(timeout time.Duration) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + controller := http.NewResponseController(w) + _ = controller.SetReadDeadline(time.Time{}) + if r.Body != nil { + r.Body = &readTimeoutBody{ReadCloser: r.Body, controller: controller, timeout: timeout} + } + next.ServeHTTP(w, r) + }) + } +} + +type readTimeoutBody struct { + io.ReadCloser + controller *http.ResponseController + timeout time.Duration +} + +func (b *readTimeoutBody) Read(buffer []byte) (int, error) { + _ = b.controller.SetReadDeadline(time.Now().Add(b.timeout)) + n, err := b.ReadCloser.Read(buffer[:min(len(buffer), maxTransferChunk)]) + if err != nil { + _ = b.controller.SetReadDeadline(time.Time{}) + } + return n, err +} + +func (b *readTimeoutBody) Close() error { + _ = b.controller.SetReadDeadline(time.Now().Add(b.timeout)) + err := b.ReadCloser.Close() + _ = b.controller.SetReadDeadline(time.Time{}) + return err +} diff --git a/apps/gateway/internal/adapters/http/middleware/read_timeout_test.go b/apps/gateway/internal/adapters/http/middleware/read_timeout_test.go new file mode 100644 index 0000000..a835f5f --- /dev/null +++ b/apps/gateway/internal/adapters/http/middleware/read_timeout_test.go @@ -0,0 +1,47 @@ +package middleware + +import ( + "fmt" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +func TestActiveUploadAndModelWaitOutliveAbsoluteDeadlines(t *testing.T) { + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + if err != nil { + w.WriteHeader(400) + return + } + // Waiting for model work must not inherit the final upload deadline. + time.Sleep(200 * time.Millisecond) + fmt.Fprint(w, string(body)) + }) + server := httptest.NewUnstartedServer(ReadTimeout(150 * time.Millisecond)(WriteTimeout(150 * time.Millisecond)(handler))) + server.Config.ReadTimeout = 50 * time.Millisecond + server.Config.WriteTimeout = 50 * time.Millisecond + server.Start() + defer server.Close() + reader, writer := io.Pipe() + go func() { + defer writer.Close() + for i := 0; i < 8; i++ { + if _, err := writer.Write([]byte("x")); err != nil { + return + } + time.Sleep(25 * time.Millisecond) + } + }() + response, err := server.Client().Post(server.URL, "text/plain", reader) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + if err != nil || response.StatusCode != 200 || string(body) != "xxxxxxxx" { + t.Fatalf("active upload/model wait interrupted: %d %q %v", response.StatusCode, body, err) + } +} diff --git a/apps/gateway/internal/adapters/http/middleware/write_timeout.go b/apps/gateway/internal/adapters/http/middleware/write_timeout.go new file mode 100644 index 0000000..e4235ae --- /dev/null +++ b/apps/gateway/internal/adapters/http/middleware/write_timeout.go @@ -0,0 +1,64 @@ +package middleware + +import ( + "io" + "net/http" + "time" +) + +const maxTransferChunk = 32 << 10 + +// WriteTimeout bounds each network write, including encrypted response frames. +// It replaces an absolute server WriteTimeout, which would cut off a healthy +// long-running stream even while tokens and keepalives continue arriving. +func WriteTimeout(timeout time.Duration) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + controller := http.NewResponseController(w) + _ = controller.SetWriteDeadline(time.Time{}) + next.ServeHTTP(&writeTimeoutWriter{ResponseWriter: w, controller: controller, timeout: timeout}, r) + }) + } +} + +type writeTimeoutWriter struct { + http.ResponseWriter + controller *http.ResponseController + timeout time.Duration +} + +func (w *writeTimeoutWriter) Unwrap() http.ResponseWriter { return w.ResponseWriter } + +func (w *writeTimeoutWriter) Write(body []byte) (int, error) { + if len(body) == 0 { + _ = w.controller.SetWriteDeadline(time.Now().Add(w.timeout)) + return w.ResponseWriter.Write(body) + } + var written int + for len(body) > 0 { + chunk := body[:min(len(body), maxTransferChunk)] + _ = w.controller.SetWriteDeadline(time.Now().Add(w.timeout)) + n, err := w.ResponseWriter.Write(chunk) + written += n + if err != nil { + return written, err + } + if n != len(chunk) { + return written, io.ErrShortWrite + } + body = body[n:] + } + return written, nil +} + +func (w *writeTimeoutWriter) WriteHeader(status int) { + _ = w.controller.SetWriteDeadline(time.Now().Add(w.timeout)) + w.ResponseWriter.WriteHeader(status) +} + +func (w *writeTimeoutWriter) FlushError() error { + _ = w.controller.SetWriteDeadline(time.Now().Add(w.timeout)) + return w.controller.Flush() +} + +func (w *writeTimeoutWriter) Flush() { _ = w.FlushError() } diff --git a/apps/gateway/internal/adapters/http/middleware/write_timeout_test.go b/apps/gateway/internal/adapters/http/middleware/write_timeout_test.go new file mode 100644 index 0000000..846a4c1 --- /dev/null +++ b/apps/gateway/internal/adapters/http/middleware/write_timeout_test.go @@ -0,0 +1,37 @@ +package middleware + +import ( + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestActiveStreamOutlivesAbsoluteServerDeadline(t *testing.T) { + handler := WriteTimeout(100 * time.Millisecond)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + for i := 0; i < 6; i++ { + if _, err := fmt.Fprint(w, "data: token\n\n"); err != nil { + return + } + w.(http.Flusher).Flush() + time.Sleep(30 * time.Millisecond) + } + fmt.Fprint(w, "data: [DONE]\n\n") + })) + srv := httptest.NewUnstartedServer(handler) + srv.Config.WriteTimeout = 50 * time.Millisecond + srv.Start() + defer srv.Close() + resp, err := srv.Client().Get(srv.URL) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil || !strings.Contains(string(body), "[DONE]") { + t.Fatalf("active stream interrupted: %v %q", err, body) + } +} diff --git a/apps/gateway/internal/adapters/providers/openai/activity_timeout_test.go b/apps/gateway/internal/adapters/providers/openai/activity_timeout_test.go new file mode 100644 index 0000000..220b903 --- /dev/null +++ b/apps/gateway/internal/adapters/providers/openai/activity_timeout_test.go @@ -0,0 +1,27 @@ +package openai + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +func TestNonStreamingKeepalivesOutliveSilenceTimeout(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + for i := 0; i < 8; i++ { + fmt.Fprint(w, "\n") + w.(http.Flusher).Flush() + time.Sleep(25 * time.Millisecond) + } + fmt.Fprint(w, `{"choices":[{"message":{"content":"complete"}}]}`) + })) + defer server.Close() + body, err := NewClient(150*time.Millisecond).Do(context.Background(), server.URL, "fixture", map[string]any{"model": "test"}) + if err != nil || len(body) == 0 { + t.Fatalf("active JSON response ended early: %v", err) + } +} diff --git a/apps/gateway/internal/adapters/providers/openai/client.go b/apps/gateway/internal/adapters/providers/openai/client.go index cf29144..0d5761a 100644 --- a/apps/gateway/internal/adapters/providers/openai/client.go +++ b/apps/gateway/internal/adapters/providers/openai/client.go @@ -14,6 +14,7 @@ import ( "strings" "time" + providerstream "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/providers/stream" enclavenetwork "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/enclave/network" ) @@ -51,9 +52,6 @@ func (c *Client) Do(ctx context.Context, baseURL, apiKey string, body map[string // DoEndpoint uses the same bounded, enclave-aware transport for JSON APIs. func (c *Client) DoEndpoint(ctx context.Context, baseURL, endpoint, apiKey string, body map[string]any) ([]byte, error) { - ctx, cancel := context.WithTimeout(ctx, c.responseTimeout) - defer cancel() - jsonBody, err := json.Marshal(body) if err != nil { return nil, fmt.Errorf("failed to marshal request: %w", err) @@ -70,6 +68,7 @@ func (c *Client) DoEndpoint(ctx context.Context, baseURL, endpoint, apiKey strin if err != nil { return nil, fmt.Errorf("provider request failed: %w", err) } + resp.Body = providerstream.WithIdleTimeout(resp.Body, c.responseTimeout) defer resp.Body.Close() respBody, err := io.ReadAll(io.LimitReader(resp.Body, 10*1024*1024)) @@ -108,6 +107,7 @@ func (c *Client) DoStream(ctx context.Context, baseURL, apiKey string, body map[ return nil, ParseProviderResponseError(resp, body) } + resp.Body = providerstream.WithIdleTimeout(resp.Body, c.responseTimeout) return resp, nil } diff --git a/apps/gateway/internal/adapters/providers/stream/header_timeout.go b/apps/gateway/internal/adapters/providers/stream/header_timeout.go new file mode 100644 index 0000000..44bc39b --- /dev/null +++ b/apps/gateway/internal/adapters/providers/stream/header_timeout.go @@ -0,0 +1,55 @@ +package stream + +import ( + "context" + "io" + "net/http" + "sync" + "time" +) + +// DoWithHeaderTimeout bounds the wait for headers without imposing a deadline +// on the response body or changing the caller's verified HTTP transport. +func DoWithHeaderTimeout(client *http.Client, request *http.Request, timeout time.Duration) (*http.Response, error) { + if timeout <= 0 { + return client.Do(request) + } + parent := request.Context() + ctx, cancel := context.WithCancel(parent) + var mu sync.Mutex + waiting, expired := true, false + timer := time.AfterFunc(timeout, func() { + mu.Lock() + defer mu.Unlock() + if waiting { + expired = true + cancel() + } + }) + response, err := client.Do(request.WithContext(ctx)) + mu.Lock() + waiting = false + timer.Stop() + timedOut := expired + mu.Unlock() + if timedOut && parent.Err() == nil { + cancel() + if response != nil { + _ = response.Body.Close() + } + return nil, context.DeadlineExceeded + } + if err != nil { + cancel() + return response, err + } + response.Body = &cancelBody{ReadCloser: response.Body, cancel: cancel} + return response, nil +} + +type cancelBody struct { + io.ReadCloser + cancel context.CancelFunc +} + +func (b *cancelBody) Close() error { b.cancel(); return b.ReadCloser.Close() } diff --git a/apps/gateway/internal/adapters/providers/stream/header_timeout_test.go b/apps/gateway/internal/adapters/providers/stream/header_timeout_test.go new file mode 100644 index 0000000..4990674 --- /dev/null +++ b/apps/gateway/internal/adapters/providers/stream/header_timeout_test.go @@ -0,0 +1,55 @@ +package stream + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +func TestHeaderWaitDoesNotSetBodyLifetime(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(200) + w.(http.Flusher).Flush() + for i := 0; i < 8; i++ { + fmt.Fprint(w, "x") + w.(http.Flusher).Flush() + time.Sleep(25 * time.Millisecond) + } + })) + defer server.Close() + request, _ := http.NewRequest("GET", server.URL, nil) + response, err := DoWithHeaderTimeout(server.Client(), request, 150*time.Millisecond) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + if err != nil || string(body) != "xxxxxxxx" { + t.Fatalf("header timer canceled active body: %v %q", err, body) + } +} + +func TestSilentHeaderWaitReportsTimeout(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { <-r.Context().Done() })) + defer server.Close() + request, _ := http.NewRequest("GET", server.URL, nil) + _, err := DoWithHeaderTimeout(server.Client(), request, 40*time.Millisecond) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("silent header wait = %v", err) + } +} + +func TestHeaderWaitPreservesCallerCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + request, _ := http.NewRequestWithContext(ctx, "GET", "http://example.invalid", nil) + _, err := DoWithHeaderTimeout(http.DefaultClient, request, time.Second) + if !errors.Is(err, context.Canceled) { + t.Fatalf("caller cancellation = %v", err) + } +} diff --git a/apps/gateway/internal/adapters/providers/stream/idle_timeout.go b/apps/gateway/internal/adapters/providers/stream/idle_timeout.go new file mode 100644 index 0000000..b4f7fb6 --- /dev/null +++ b/apps/gateway/internal/adapters/providers/stream/idle_timeout.go @@ -0,0 +1,71 @@ +// Package stream bounds provider silence while allowing active long streams. +package stream + +import ( + "io" + "sync" + "time" +) + +type idleTimeoutError struct{} + +func (idleTimeoutError) Error() string { return "provider stream idle timeout" } +func (idleTimeoutError) Timeout() bool { return true } +func (idleTimeoutError) Temporary() bool { return true } + +type idleBody struct { + body io.ReadCloser + timeout time.Duration + timer *time.Timer + mu sync.Mutex + closed bool + expired bool + deadline time.Time +} + +func WithIdleTimeout(body io.ReadCloser, timeout time.Duration) io.ReadCloser { + if timeout <= 0 { + return body + } + b := &idleBody{body: body, timeout: timeout, deadline: time.Now().Add(timeout)} + b.timer = time.AfterFunc(timeout, func() { + b.mu.Lock() + if b.closed { + b.mu.Unlock() + return + } + // A reset can race a callback already waiting for the lock. A stale + // callback must not expire a stream that just delivered more bytes. + if remaining := time.Until(b.deadline); remaining > 0 { + b.timer.Reset(remaining) + b.mu.Unlock() + return + } + b.expired, b.closed = true, true + b.mu.Unlock() + _ = body.Close() + }) + return b +} + +func (b *idleBody) Read(p []byte) (int, error) { + n, err := b.body.Read(p) + b.mu.Lock() + defer b.mu.Unlock() + if b.expired { + return n, idleTimeoutError{} + } + if n > 0 && !b.closed { + b.deadline = time.Now().Add(b.timeout) + b.timer.Reset(b.timeout) + } + return n, err +} + +func (b *idleBody) Close() error { + b.mu.Lock() + b.closed = true + b.timer.Stop() + b.mu.Unlock() + return b.body.Close() +} diff --git a/apps/gateway/internal/adapters/providers/stream/idle_timeout_test.go b/apps/gateway/internal/adapters/providers/stream/idle_timeout_test.go new file mode 100644 index 0000000..a5aa8a6 --- /dev/null +++ b/apps/gateway/internal/adapters/providers/stream/idle_timeout_test.go @@ -0,0 +1,40 @@ +package stream + +import ( + "errors" + "io" + "net" + "testing" + "time" +) + +func TestActivityRefreshesIdleTimeout(t *testing.T) { + r, w := io.Pipe() + body := WithIdleTimeout(r, 150*time.Millisecond) + defer body.Close() + go func() { + defer w.Close() + for i := 0; i < 7; i++ { + if _, err := w.Write([]byte("x")); err != nil { + return + } + time.Sleep(30 * time.Millisecond) + } + }() + got, err := io.ReadAll(body) + if err != nil || string(got) != "xxxxxxx" { + t.Fatalf("active stream timed out: %v %q", err, got) + } +} + +func TestProviderSilenceClosesBlockedRead(t *testing.T) { + r, w := io.Pipe() + defer w.Close() + body := WithIdleTimeout(r, 20*time.Millisecond) + defer body.Close() + _, err := body.Read(make([]byte, 1)) + var timeout net.Error + if !errors.As(err, &timeout) || !timeout.Timeout() { + t.Fatalf("idle provider did not time out: %v", err) + } +} diff --git a/apps/gateway/internal/adapters/providers/tinfoil/adapter.go b/apps/gateway/internal/adapters/providers/tinfoil/adapter.go index 3f06017..13f7d22 100644 --- a/apps/gateway/internal/adapters/providers/tinfoil/adapter.go +++ b/apps/gateway/internal/adapters/providers/tinfoil/adapter.go @@ -13,6 +13,7 @@ import ( "time" "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/providers/openai" + providerstream "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/providers/stream" "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/application/ports" "github.com/dappnode/dappnode-nexus-gateway/pkg/domain" tinfoilsdk "github.com/tinfoilsh/tinfoil-go" @@ -80,20 +81,15 @@ func (a *Adapter) newVerifiedClient(ctx context.Context, model domain.PublicMode } func (a *Adapter) do(ctx context.Context, verified VerifiedClient, apiKey string, body map[string]any) ([]byte, error) { - if a.timeout > 0 { - var cancel context.CancelFunc - ctx, cancel = context.WithTimeout(ctx, a.timeout) - defer cancel() - } - req, err := newProviderRequest(ctx, verified, apiKey, body) if err != nil { return nil, err } - resp, err := verified.HTTPClient().Do(req) + resp, err := providerstream.DoWithHeaderTimeout(verified.HTTPClient(), req, a.timeout) if err != nil { return nil, fmt.Errorf("tinfoil request failed: %w", err) } + resp.Body = providerstream.WithIdleTimeout(resp.Body, a.timeout) defer resp.Body.Close() respBody, err := io.ReadAll(io.LimitReader(resp.Body, 10*1024*1024)) @@ -111,7 +107,7 @@ func (a *Adapter) doStream(ctx context.Context, verified VerifiedClient, apiKey if err != nil { return nil, err } - resp, err := verified.HTTPClient().Do(req) + resp, err := providerstream.DoWithHeaderTimeout(verified.HTTPClient(), req, a.timeout) if err != nil { return nil, fmt.Errorf("tinfoil stream request failed: %w", err) } @@ -120,6 +116,7 @@ func (a *Adapter) doStream(ctx context.Context, verified VerifiedClient, apiKey resp.Body.Close() return nil, openai.ParseProviderResponseError(resp, respBody) } + resp.Body = providerstream.WithIdleTimeout(resp.Body, a.timeout) return resp, nil } diff --git a/apps/gateway/main.go b/apps/gateway/main.go index 1e1bda4..9f4c1ff 100644 --- a/apps/gateway/main.go +++ b/apps/gateway/main.go @@ -15,6 +15,7 @@ import ( nsmattestation "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/attestation/nsm" gwhttp "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/http" "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/http/handlers" + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/http/middleware" meteringadapter "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/metering" "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/observability/metrics" "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/pii/presidio" @@ -56,7 +57,11 @@ func main() { routerURL := envOr("ROUTER_URL", "http://localhost:8083") meteringURL := envOr("METERING_URL", "http://localhost:8084") meteringToken := envOr("METERING_TOKEN", "dev-metering-token") - providerTimeout := 300 * time.Second + transportTimeouts, err := loadTransportTimeouts() + if err != nil { + log.Fatal(err) + } + providerTimeout := transportTimeouts.ProviderSilence zapLogger, err := logger.NewZapLogger(logLevel) if err != nil { @@ -130,16 +135,18 @@ func main() { } router := gwhttp.NewRouter(healthHandler, modelsHandler, chatHandler, handlers.NewSystemOneHandler(generateSvc, zapLogger), confidentialChatHandler, tinfoilHandler, attestationHandler, zapLogger) + router = middleware.ReadTimeout(transportTimeouts.ClientIO)(middleware.WriteTimeout(transportTimeouts.ClientIO)(router)) metricsMux := http.NewServeMux() metricsMux.Handle("/metrics", metrics.Handler()) srv := &http.Server{ - Addr: ":" + port, - Handler: router, - ReadTimeout: 30 * time.Second, - WriteTimeout: 300 * time.Second, - IdleTimeout: 120 * time.Second, + Addr: ":" + port, + Handler: router, + ReadHeaderTimeout: 30 * time.Second, + // Reads and writes are bounded one at a time by middleware. An absolute + // deadline would cut off healthy streams and long answers. + IdleTimeout: 120 * time.Second, } go func() { diff --git a/apps/gateway/timeouts.go b/apps/gateway/timeouts.go new file mode 100644 index 0000000..910aaec --- /dev/null +++ b/apps/gateway/timeouts.go @@ -0,0 +1,34 @@ +package main + +import ( + "fmt" + "os" + "strconv" + "time" +) + +type transportTimeouts struct { + ProviderSilence time.Duration + ClientIO time.Duration +} + +func loadTransportTimeouts() (transportTimeouts, error) { + provider, err := timeoutSeconds("PROVIDER_SILENCE_TIMEOUT_SECONDS", 15*time.Minute) + if err != nil { + return transportTimeouts{}, err + } + clientIO, err := timeoutSeconds("CLIENT_IO_TIMEOUT_SECONDS", time.Minute) + return transportTimeouts{ProviderSilence: provider, ClientIO: clientIO}, err +} + +func timeoutSeconds(name string, fallback time.Duration) (time.Duration, error) { + value := os.Getenv(name) + if value == "" { + return fallback, nil + } + seconds, err := strconv.ParseInt(value, 10, 64) + if err != nil || seconds <= 0 || seconds > int64((1<<63-1)/time.Second) { + return 0, fmt.Errorf("%s must be a positive number of seconds", name) + } + return time.Duration(seconds) * time.Second, nil +} diff --git a/apps/gateway/timeouts_test.go b/apps/gateway/timeouts_test.go new file mode 100644 index 0000000..beed609 --- /dev/null +++ b/apps/gateway/timeouts_test.go @@ -0,0 +1,19 @@ +package main + +import ( + "testing" + "time" +) + +func TestTimeoutConfigurationRejectsInvalidOrOverflowingValues(t *testing.T) { + for _, value := range []string{"0", "-1", "not-a-duration", "9223372036854775807"} { + t.Setenv("TEST_STALL_TIMEOUT", value) + if _, err := timeoutSeconds("TEST_STALL_TIMEOUT", time.Minute); err == nil { + t.Fatalf("accepted invalid duration %q", value) + } + } + t.Setenv("TEST_STALL_TIMEOUT", "900") + if got, err := timeoutSeconds("TEST_STALL_TIMEOUT", time.Minute); err != nil || got != 15*time.Minute { + t.Fatalf("configured timeout = %v, %v", got, err) + } +}