encode: flush headers immediately for server-sent events responses (#7905)

* encode: flush headers immediately for server-sent events responses

The encode middleware withholds the response header until the first body
write so it can sniff content-type and apply the minimum_length threshold.
For a text/event-stream response the upstream typically writes headers and
flushes to establish the event stream before any event body is available,
so the client never received the headers and the stream stalled; the same
buffering also delayed individual events.

When WriteHeader sees a text/event-stream content type, initialize encoding
and write the header through immediately. Forcing the header out also marks
the response as started, so subsequent event writes bypass the minimum_length
buffering and stream to the client as they arrive.

Fixes #6293

* encode: add WriteHeader benchmark covering SSE fast path

* encode: replace mime.ParseMediaType with bound-checked SSE check

WriteHeader runs an SSE Content-Type check on every call once headers
haven't been written yet. mime.ParseMediaType parses the full media
type, including parameters, even when nothing matches, which shows up
on the hot header-write path.

Replace it with a bound-checked manual prefix/boundary check (isSSE),
skipping parameter parsing for the common non-SSE case.

* encode: reject content types with junk after text/event-stream

isSSE accepted any suffix after a space, so a value like
"text/event-stream nonsense" was treated as an SSE response. After the
media type, skip optional whitespace and require either the end of the
value or a parameter separator. The check remains allocation-free, so
the hot-path motivation for the manual matcher is preserved.

---------

Co-authored-by: SillyZir <269283839+SillyZir@users.noreply.github.com>
Co-authored-by: Kévin Dunglas <kevin@les-tilleuls.coop>

modules/caddyhttp/encode/encode.go

@@ -34,6 +34,8 @@ import (
 	"github.com/caddyserver/caddy/v2/modules/caddyhttp"
 )
 
+const sseMediaType = "text/event-stream"
+
 func init() {
 	caddy.RegisterModule(Encode{})
 }
@@ -262,12 +264,14 @@ func (rw *responseWriter) WriteHeader(status int) {
 		rw.disabled = true // partial representations must not be dynamically re-encoded
 	}
 
+	h := rw.Header()
+
 	// See #5849 and RFC 9110 section 15.4.5 (https://www.rfc-editor.org/rfc/rfc9110.html#section-15.4.5) - 304
 	// Not Modified must have certain headers set as if it was a 200 response, and according to the issue
 	// we would miss the Vary header in this case when compression was also enabled; note that we set this
 	// header in the responseWriter.init() method but that is only called if we are writing a response body
-	if status == http.StatusNotModified && !hasVaryValue(rw.Header(), "Accept-Encoding") {
-		rw.Header().Add("Vary", "Accept-Encoding")
+	if status == http.StatusNotModified && !hasVaryValue(h, "Accept-Encoding") {
+		h.Add("Vary", "Accept-Encoding")
 	}
 
 	// write status immediately if status is 2xx and the request is CONNECT
@@ -283,6 +287,29 @@ func (rw *responseWriter) WriteHeader(status int) {
 	if 100 <= status && status <= 199 {
 		rw.ResponseWriter.WriteHeader(status)
 	}
+
+	// write header immediately for server-sent events responses, since the
+	// body may not be written for a while and the client needs the headers
+	// to establish the event stream; see #6293
+	if !rw.wroteHeader && (status < 100 || status > 199) && isSSE(h.Get("Content-Type")) {
+		rw.init()
+		rw.ResponseWriter.WriteHeader(status)
+		rw.wroteHeader = true
+	}
+}
+
+func isSSE(contentType string) bool {
+	if len(contentType) < len(sseMediaType) || !strings.EqualFold(contentType[:len(sseMediaType)], sseMediaType) {
+		return false
+	}
+
+	// After the media type, allow only optional whitespace followed by the
+	// end of the value or a parameter separator, so a longer type such as
+	// "text/event-streamfoo" or garbage like "text/event-stream nonsense"
+	// does not match. TrimLeft on the tail does not allocate.
+	rest := strings.TrimLeft(contentType[len(sseMediaType):], " \t")
+
+	return rest == "" || rest[0] == ';'
 }
 
 // Match determines, if encoding should be done based on the ResponseMatcher.

modules/caddyhttp/encode/encode_test.go

@@ -20,6 +20,92 @@ func BenchmarkOpenResponseWriter(b *testing.B) {
 	}
 }
 
+// discardResponseWriter is a minimal http.ResponseWriter used to isolate
+// WriteHeader's own cost from a real transport.
+type discardResponseWriter struct {
+	header http.Header
+}
+
+func (w *discardResponseWriter) Header() http.Header         { return w.header }
+func (w *discardResponseWriter) Write(p []byte) (int, error) { return len(p), nil }
+func (w *discardResponseWriter) WriteHeader(int)             {}
+
+// BenchmarkResponseWriterWriteHeader covers the branches inside WriteHeader:
+// the plain/common case, the SSE Content-Type check (both when it doesn't
+// match and when it does and rw.init() runs), CONNECT 2xx, informational
+// (1xx), and 304 Not Modified (Vary bookkeeping).
+func BenchmarkResponseWriterWriteHeader(b *testing.B) {
+	benchCases := []struct {
+		name        string
+		encoding    string
+		isConnect   bool
+		status      int
+		contentType string
+	}{
+		{name: "plain", encoding: "test", status: http.StatusOK},
+		{name: "html", encoding: "test", status: http.StatusOK, contentType: "text/html; charset=utf-8"},
+		{name: "event-stream", encoding: "gzip", status: http.StatusOK, contentType: "text/event-stream"},
+		{name: "connect", encoding: "test", isConnect: true, status: http.StatusOK},
+		{name: "informational", encoding: "test", status: http.StatusEarlyHints},
+		{name: "not-modified", encoding: "test", status: http.StatusNotModified},
+	}
+
+	for _, bc := range benchCases {
+		b.Run(bc.name, func(b *testing.B) {
+			enc := new(Encode)
+			if bc.name == "event-stream" {
+				enc.writerPools = map[string]*sync.Pool{
+					"gzip": {New: func() any { return mockEncoder{} }},
+				}
+				ctx, cancel := caddy.NewContext(caddy.Context{Context: context.Background()})
+				defer cancel()
+				if err := enc.Provision(ctx); err != nil {
+					b.Fatalf("Provision() error = %v", err)
+				}
+			}
+
+			w := &discardResponseWriter{header: make(http.Header)}
+			rw := enc.openResponseWriter(bc.encoding, w, bc.isConnect)
+
+			for b.Loop() {
+				for k := range rw.Header() {
+					delete(rw.Header(), k)
+				}
+				if bc.contentType != "" {
+					rw.Header().Set("Content-Type", bc.contentType)
+				}
+				rw.wroteHeader = false
+				rw.statusCode = 0
+				rw.disabled = false
+				rw.w = nil
+				rw.WriteHeader(bc.status)
+			}
+		})
+	}
+}
+
+func TestIsSSE(t *testing.T) {
+	for _, tc := range []struct {
+		contentType string
+		want        bool
+	}{
+		{"", false},
+		{"text/plain", false},
+		{"text/event-stream", true},
+		{"Text/Event-Stream", true},
+		{"text/event-stream; charset=utf-8", true},
+		{"text/event-stream ; charset=utf-8", true},
+		{"text/event-stream  ", true},
+		{"text/event-streamfoo", false},
+		{"text/event-stream nonsense", false},
+		{"text/event-stream x; charset=utf-8", false},
+	} {
+		if got := isSSE(tc.contentType); got != tc.want {
+			t.Errorf("isSSE(%q) = %v, want %v", tc.contentType, got, tc.want)
+		}
+	}
+}
+
 func TestPreferOrder(t *testing.T) {
 	testCases := []struct {
 		name     string

modules/caddyhttp/encode/sse_test.go

@@ -0,0 +1,134 @@
+// Copyright 2015 Matthew Holt and The Caddy Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package encode_test
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/caddyserver/caddy/v2"
+	"github.com/caddyserver/caddy/v2/caddyconfig"
+	"github.com/caddyserver/caddy/v2/modules/caddyhttp"
+	"github.com/caddyserver/caddy/v2/modules/caddyhttp/encode"
+	caddygzip "github.com/caddyserver/caddy/v2/modules/caddyhttp/encode/gzip"
+)
+
+// recordingWriter records the moment WriteHeader reaches the underlying
+// writer, so a test can distinguish "headers flushed to the client" from
+// "headers still buffered inside the encoder".
+type recordingWriter struct {
+	http.ResponseWriter
+	wroteHeader bool
+	status      int
+}
+
+func (rw *recordingWriter) WriteHeader(status int) {
+	if !rw.wroteHeader {
+		rw.wroteHeader = true
+		rw.status = status
+	}
+	rw.ResponseWriter.WriteHeader(status)
+}
+
+func (rw *recordingWriter) Flush() {
+	if f, ok := rw.ResponseWriter.(http.Flusher); ok {
+		f.Flush()
+	}
+}
+
+func newSSEEncodeHandler(t *testing.T) *encode.Encode {
+	t.Helper()
+	enc := &encode.Encode{
+		EncodingsRaw: caddy.ModuleMap{
+			"gzip": caddyconfig.JSON(caddygzip.Gzip{}, nil),
+		},
+		Prefer: []string{"gzip"},
+		// A large minimum_length means a normal small response would be
+		// buffered (its header withheld) until enough bytes arrive; the SSE
+		// path must bypass this so the handshake reaches the client.
+		MinLength: 4096,
+	}
+	ctx, cancel := caddy.NewContext(caddy.Context{Context: t.Context()})
+	t.Cleanup(cancel)
+	if err := enc.Provision(ctx); err != nil {
+		t.Fatalf("Provision() error = %v", err)
+	}
+	if err := enc.Validate(); err != nil {
+		t.Fatalf("Validate() error = %v", err)
+	}
+	return enc
+}
+
+// An SSE upstream typically writes headers and flushes to establish the
+// event stream before any event body is available. The encode middleware
+// must let those headers reach the client immediately rather than holding
+// them for minimum_length content sniffing. See #6293.
+func TestSSEHeadersFlushedBeforeBody(t *testing.T) {
+	enc := newSSEEncodeHandler(t)
+	r := httptest.NewRequest(http.MethodGet, "/", nil)
+	r.Header.Set("Accept-Encoding", "gzip")
+	baseRec := httptest.NewRecorder()
+	rec := &recordingWriter{ResponseWriter: baseRec}
+
+	next := caddyhttp.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) error {
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.WriteHeader(http.StatusOK)
+		// Flush the way a real SSE handler does — through the response
+		// controller (encode's writer implements FlushError, not Flush).
+		if err := http.NewResponseController(w).Flush(); err != nil {
+			t.Errorf("flush failed: %v", err)
+		}
+		// Before any event body is written, the client must already have the
+		// headers AND the flush must have reached the underlying writer.
+		if !rec.wroteHeader {
+			t.Error("SSE response headers were not written to the client before the body")
+		}
+		if rec.status != http.StatusOK {
+			t.Errorf("underlying status = %d, want 200", rec.status)
+		}
+		if !baseRec.Flushed {
+			t.Error("underlying ResponseRecorder was not flushed for the SSE handshake")
+		}
+		return nil
+	})
+
+	if err := enc.ServeHTTP(rec, r, next); err != nil {
+		t.Fatalf("ServeHTTP() error = %v", err)
+	}
+}
+
+// A normal (non-SSE) small response is still allowed to buffer its header
+// for content sniffing — the SSE change must not force every response to
+// flush its header early. This guards the scope of the fix.
+func TestNonSSESmallResponseStillBuffersHeader(t *testing.T) {
+	enc := newSSEEncodeHandler(t)
+	r := httptest.NewRequest(http.MethodGet, "/", nil)
+	r.Header.Set("Accept-Encoding", "gzip")
+	rec := &recordingWriter{ResponseWriter: httptest.NewRecorder()}
+
+	next := caddyhttp.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) error {
+		w.Header().Set("Content-Type", "text/plain")
+		w.WriteHeader(http.StatusOK)
+		if rec.wroteHeader {
+			t.Error("non-SSE response flushed its header early; SSE bypass leaked to normal responses")
+		}
+		return nil
+	})
+
+	if err := enc.ServeHTTP(rec, r, next); err != nil {
+		t.Fatalf("ServeHTTP() error = %v", err)
+	}
+}