fix: no limit on response duration, only stall guards

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

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 {

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
+}

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)
+	}
+}

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() }

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)
+	}
+}

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)
+	}
+}

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
 }
 

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() }