fix(spool): treat client hangup after full body as complete

internal/spool/group.go

@@ -6,6 +6,7 @@ import (
 	"maps"
 	"net/http"
 	"os"
+	"strconv"
 	"sync"
 	"time"
 
@@ -306,6 +307,11 @@ func (e *entry) serve(w http.ResponseWriter, req *http.Request, filled bool) (Re
 		}
 		select {
 		case <-req.Context().Done():
+			// A client that read the whole declared body may hang up before the
+			// fill commits; that is a complete response, not a failure.
+			if headerSent && header.Get("Content-Length") == strconv.FormatInt(offset, 10) {
+				return result, nil
+			}
 			return result, errors.WithStack(req.Context().Err())
 		case <-changed:
 		}

internal/spool/group_test.go

@@ -125,6 +125,50 @@ func TestFillOutlivesItsRequests(t *testing.T) {
 	assert.Equal(t, "first second", string(body))
 }
 
+func TestClientLeavingAfterFullBodyIsComplete(t *testing.T) {
+	ctx, registry, _ := newTestRegistry(t, spool.Config{})
+	group := registry.NewGroup(spool.ReleaseCompleted)
+	release := make(chan struct{})
+	defer close(release)
+	fill := func(_ context.Context, w *spool.Writer) error {
+		w.Header().Set("Content-Length", "4")
+		w.WriteHeader(http.StatusOK)
+		_, err := io.WriteString(w, "body")
+		<-release
+		return err
+	}
+
+	reqCtx, cancel := context.WithCancel(ctx)
+	w := &signalingRecorder{ResponseRecorder: httptest.NewRecorder(), wrote: make(chan struct{}, 1)}
+	served := make(chan error, 1)
+	var result spool.Result
+	go func() {
+		req := httptest.NewRequestWithContext(reqCtx, http.MethodGet, "/object", nil)
+		var err error
+		result, err = group.Serve(w, req, key, fill)
+		served <- err
+	}()
+	<-w.wrote
+	cancel()
+	assert.NoError(t, <-served)
+	assert.Equal(t, int64(4), result.Bytes)
+	assert.Equal(t, "body", w.Body.String())
+}
+
+type signalingRecorder struct {
+	*httptest.ResponseRecorder
+	wrote chan struct{}
+}
+
+func (r *signalingRecorder) Write(p []byte) (int, error) {
+	n, err := r.ResponseRecorder.Write(p)
+	select {
+	case r.wrote <- struct{}{}:
+	default:
+	}
+	return n, err
+}
+
 func TestRequestsFollowPendingCacheWrite(t *testing.T) {
 	ctx, registry, dir := newTestRegistry(t, spool.Config{})
 	group := registry.NewGroup(spool.ReleaseCompleted)