rest: flush-chain integrity — composition test, convention guards, error-path fakes (#174)

rest/cachecontrol_internal_test.go

@@ -131,7 +131,7 @@ func TestCacheControlWriter_CommitDecision(t *testing.T) {
 	// happened, and both must KEEP. Broadening the discriminator to any
 	// non-nil error would delete from the live map a stamp the wire already
 	// carries and reopen with committed=false a decision the wire already
-	// took. The delegate (flushErrorWriter, sentry_test.go) fails the flush
+	// took. The delegate (flushErrorWriter, fakewriters_test.go) fails the flush
 	// with the injected error, never the refusal sentinel; the trailing 404
 	// is the handler-reacts-to-the-failed-flush move from the rollback pin
 	// above — here it must NOT reopen the commit (on a real server that 404
@@ -170,17 +170,6 @@ func TestCacheControlWriter_CommitDecision(t *testing.T) {
 	}
 }
 
-// noFlushWriter hides the recorder's Flusher — the shape gzipResponseWriter
-// had before #167 (no FlushError, no Flusher, no Unwrap), where a delegated
-// flush always fails with http.ErrNotSupported.
-type noFlushWriter struct {
-	rr *httptest.ResponseRecorder
-}
-
-func (w *noFlushWriter) Header() http.Header         { return w.rr.Header() }
-func (w *noFlushWriter) Write(b []byte) (int, error) { return w.rr.Write(b) }
-func (w *noFlushWriter) WriteHeader(code int)        { w.rr.WriteHeader(code) }
-
 // cacheControl values are registration-time constants; a negative max-age is
 // always a programming error, so it fails at registration, not on the wire.
 func TestCacheControl_NegativeMaxAgePanics(t *testing.T) {

rest/fakewriters_test.go

@@ -0,0 +1,108 @@
+package rest
+
+// The delegate fake family (#174): every error-path pin on the chain's
+// writer wrappers injects its downstream failure through one of these fakes,
+// each modelling ONE base-writer shape the real wire can take. They live
+// together so the next pin (e.g. #178's rollback witnesses) reuses a shape
+// instead of minting a near-duplicate:
+//
+//   - noFlushWriter: the controller's REFUSAL shape — no FlushError, no
+//     Flusher, no Unwrap, so a delegated flush returns http.ErrNotSupported
+//     and nothing reaches the wire (the rollback discriminator's "nothing
+//     sent" world, #164/#163 R1).
+//   - flushErrorWriter: the GENUINE-failure shape — FlushError returns the
+//     injected error (a conn write error's analogue), so the commit really
+//     happened before the error surfaced (the "keep the latch/stamp" world,
+//     #164/#172).
+//   - failingWriter: the broken-pipe shape — every body Write fails with the
+//     injected error, the failure that makes gzip's OWN writer (gz.Flush,
+//     gz.Close) fail; its FlushError succeeds and counts, so a pin can
+//     observe whether a wrapper's flush path stopped before delegating.
+//   - flushSnapshotWriter: the wire-faithful observer — records what the
+//     wrapper pushed down and snapshots the buffer the moment Flush arrives
+//     (the bytes that would be on the wire after the flush).
+//   - noFlushUnderlying: noFlushWriter with its own buffer instead of a
+//     recorder, for pins that must inspect the bytes downstream of a refusal.
+//
+// All package-internal: the wrappers under test are unexported.
+
+import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+)
+
+// noFlushWriter hides the recorder's Flusher — the shape gzipResponseWriter
+// had before #167 (no FlushError, no Flusher, no Unwrap), where a delegated
+// flush always fails with http.ErrNotSupported.
+type noFlushWriter struct {
+	rr *httptest.ResponseRecorder
+}
+
+func (w *noFlushWriter) Header() http.Header         { return w.rr.Header() }
+func (w *noFlushWriter) Write(b []byte) (int, error) { return w.rr.Write(b) }
+func (w *noFlushWriter) WriteHeader(code int)        { w.rr.WriteHeader(code) }
+
+// flushErrorWriter is a base writer whose flush genuinely FAILS rather than
+// being refused: FlushError returns the injected error, never
+// http.ErrNotSupported. The real-server analogue is a conn write error — the
+// implied 200 commits to the wire BEFORE the error returns to the handler.
+type flushErrorWriter struct {
+	rr  *httptest.ResponseRecorder
+	err error
+}
+
+func (w *flushErrorWriter) Header() http.Header         { return w.rr.Header() }
+func (w *flushErrorWriter) Write(b []byte) (int, error) { return w.rr.Write(b) }
+func (w *flushErrorWriter) WriteHeader(code int)        { w.rr.WriteHeader(code) }
+func (w *flushErrorWriter) FlushError() error           { return w.err }
+
+// failingWriter rejects every body write with a fixed genuine error — the
+// downstream-failure shape (connection torn down, write timeout) that makes
+// the gzip layer's own writes fail: gz.Flush and gz.Close surface it, and it
+// sticks in the gzip.Writer. Its FlushError SUCCEEDS and counts calls, so a
+// pin can prove a wrapper's flush path returned early — when the failure
+// came from the wrapper's own downstream write, a delegated flush must never
+// run (flushes stays 0).
+type failingWriter struct {
+	header  http.Header
+	err     error
+	flushes int // delegated flushes that reached this base
+}
+
+func (w *failingWriter) Header() http.Header       { return w.header }
+func (w *failingWriter) Write([]byte) (int, error) { return 0, w.err }
+func (w *failingWriter) WriteHeader(int)           {}
+func (w *failingWriter) FlushError() error         { w.flushes++; return nil }
+
+// flushSnapshotWriter records what the wrapper under test pushed down to it,
+// and snapshots that buffer the moment Flush is called — the bytes that would
+// be on the wire after the flush.
+type flushSnapshotWriter struct {
+	header   http.Header
+	buf      bytes.Buffer
+	flushed  int
+	snapshot []byte // buf contents at the first Flush
+}
+
+func (w *flushSnapshotWriter) Header() http.Header         { return w.header }
+func (w *flushSnapshotWriter) Write(b []byte) (int, error) { return w.buf.Write(b) }
+func (w *flushSnapshotWriter) WriteHeader(int)             {}
+func (w *flushSnapshotWriter) Flush() {
+	if w.flushed == 0 {
+		w.snapshot = append([]byte(nil), w.buf.Bytes()...)
+	}
+	w.flushed++
+}
+
+// noFlushUnderlying is a writer the controller cannot flush — no FlushError,
+// no Flusher, no Unwrap — with its own buffer, for pins that must inspect
+// what landed downstream of the refusal.
+type noFlushUnderlying struct {
+	header http.Header
+	buf    bytes.Buffer
+}
+
+func (w *noFlushUnderlying) Header() http.Header         { return w.header }
+func (w *noFlushUnderlying) Write(b []byte) (int, error) { return w.buf.Write(b) }
+func (w *noFlushUnderlying) WriteHeader(int)             {}

rest/flushchain_internal_test.go

@@ -0,0 +1,141 @@
+package rest
+
+// Composed flush-chain integrity (#174): #167's "flush succeeds from a
+// handler" held only by composition of per-layer pins (sentry, metrics,
+// cachecontrol, gzip each pinned individually) — nothing pinned the COMPOSED
+// property, so a wrapper inserted into New tomorrow without
+// FlushError/Flusher/Unwrap would break handler flushes in production with
+// every existing test green. This test assembles the stack through chain()
+// (rest.go) — the SAME function New composes its middleware with, so the
+// assembly under test is production's by construction, not a hand-kept
+// mirror that could drift (#174 review) — with routes() swapped for a probe
+// mux whose route registers through handle() exactly like a production route
+// (routeLabel + requireScope + cacheControl). Over a real server it pins the
+// composed behavior: the mid-body flush succeeds, and the flushed prefix is
+// decodable on the wire while the handler still holds the tail. Internal
+// (package rest) because chain and the wrappers it mounts are unexported.
+
+import (
+	"compress/gzip"
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestFlushChain_NewShapedStackStreamsFlushedPrefix(t *testing.T) {
+	// Sentry ENABLED: without a client sentryMiddleware never mounts
+	// commitWriter, and the composed property would silently exclude the
+	// outermost wrapper. The premise that an enabled client mounts the
+	// wrapper UNCONDITIONALLY is itself pinned —
+	// TestSentryMiddleware_MountsCommitWriterWhenEnabled (sentry_test.go).
+	enableSentry(t)
+
+	const part1 = "first chunk, flushed mid-stream"
+	const part2 = "; tail held by the handler until the prefix decoded"
+
+	flushErr := make(chan error, 1) // handler runs on the server goroutine
+	release := make(chan struct{})  // client → handler: prefix decoded, finish
+	releaseOnce := sync.OnceFunc(func() { close(release) })
+	writeErr := make(chan error, 2)
+	probe := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, err := io.WriteString(w, part1)
+		writeErr <- err
+		flushErr <- http.NewResponseController(w).Flush()
+		<-release
+		_, err = io.WriteString(w, part2)
+		writeErr <- err
+	})
+
+	// The production assembly itself: chain() is what New composes the
+	// middleware with (rest.go), here with routes() swapped for the probe
+	// mux. handle() gives the probe the full per-route wrap set (routeLabel —
+	// the metrics sweep's never-routed contract holds for this traffic too —
+	// plus requireScope and cacheControl), so the flush traverses every
+	// writer wrapper a production route's would: commitWriter → statusWriter
+	// → gzipResponseWriter → cacheControlWriter.
+	mux := http.NewServeMux()
+	handle(mux, "GET /", "read", maxAgeRosterFamily, probe)
+	h := chain(&sentryFakeDatastore{}, mux)
+
+	srv := httptest.NewServer(h)
+	defer srv.Close()
+	// LIFO: unblock the handler before srv.Close waits on it, so a failure
+	// before the deliberate release fails the test instead of hanging it.
+	defer releaseOnce()
+
+	// Watchdog: a chain that buffers the flush instead of streaming it would
+	// leave the client blocked forever; the deadline turns that into a crisp
+	// failure (same pattern as the per-layer streaming pin in gzip_test.go).
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+"/", nil)
+	require.NoError(t, err)
+	req.Header.Set("Authorization", "Bearer cav7_sentry_read")
+	req.Header.Set("Accept-Encoding", "gzip")
+	res, err := srv.Client().Do(req)
+	require.NoError(t, err)
+	defer res.Body.Close()
+
+	require.Equal(t, http.StatusOK, res.StatusCode)
+	require.Equal(t, "gzip", res.Header.Get("Content-Encoding"),
+		"the request negotiated gzip — the full wrapper chain must be in play")
+	assert.Equal(t, "max-age=600", res.Header.Get("Cache-Control"),
+		"the first write committed the implied 200 — cacheControlWriter stamped it, proof the innermost wrapper was in the chain")
+
+	require.NoError(t, <-writeErr, "pre-flush write must succeed")
+	require.NoError(t, <-flushErr,
+		"ResponseController.Flush must succeed through the composed New-shaped chain")
+
+	// Decode the flushed prefix while the handler still blocks on release —
+	// possible only if every layer pushed its part of the flush to the wire.
+	zr, prefix := decodeStackPrefixWithin(t, res.Body, len(part1), 5*time.Second)
+	assert.Equal(t, part1, prefix,
+		"flushed prefix must decompress to exactly the pre-flush writes")
+
+	releaseOnce()
+	tail, err := io.ReadAll(zr) // EOF verifies the gzip CRC/size trailer
+	require.NoError(t, err, "tail must decode through an intact trailer")
+	require.NoError(t, <-writeErr, "post-flush write must succeed")
+	assert.Equal(t, part1+part2, prefix+string(tail),
+		"full body must decompress byte-identically to the handler output")
+}
+
+// decodeStackPrefixWithin opens a gzip reader over body and decodes exactly n
+// bytes, failing the test if that takes longer than timeout — the deadlock
+// shape a buffered (non-streamed) flush produces. Returns the open reader so
+// the caller can decode the rest of the stream. (The package-external twin
+// lives in gzip_test.go; this internal test cannot reach it.)
+func decodeStackPrefixWithin(t *testing.T, body io.Reader, n int, timeout time.Duration) (*gzip.Reader, string) {
+	t.Helper()
+	type result struct {
+		zr  *gzip.Reader
+		buf []byte
+		err error
+	}
+	done := make(chan result, 1)
+	go func() {
+		zr, err := gzip.NewReader(body)
+		if err != nil {
+			done <- result{err: err}
+			return
+		}
+		buf := make([]byte, n)
+		_, err = io.ReadFull(zr, buf)
+		done <- result{zr: zr, buf: buf, err: err}
+	}()
+	select {
+	case res := <-done:
+		require.NoError(t, res.err, "flushed prefix must be decodable as a gzip stream")
+		return res.zr, string(res.buf)
+	case <-time.After(timeout):
+		t.Fatal("flushed prefix never became decodable — some layer buffered the flush instead of streaming it to the wire")
+		return nil, ""
+	}
+}

rest/gzip.go

@@ -77,7 +77,10 @@ func GzipMiddleware(next http.Handler) http.Handler {
 					// stream). The precise per-shape faithfulness — commit-then-
 					// hijack smuggling, truncated-after-output, accurate
 					// logging — is TRACKED IN #181 and is latent until a
-					// hijacking handler exists. No handler hijacks today.
+					// hijacking handler exists. No handler hijacks today —
+					// self-announcing, not remembered: the hijack tripwire
+					// (wrapperconventions_test.go, #174) fails the suite the
+					// moment production code in this package hijacks.
 					if errors.Is(err, http.ErrHijacked) {
 						return
 					}