fix(uploadstore): preserve default transport while bounding R2 headers

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- Fixed R2 header stalls while preserving configured HTTP transport wrappers. Thanks @SebTardif.
 - Fixed member-directory pagination loops across chat, embeds, and workspace settings, and restored loading newer embedded channel messages. Thanks @SebTardif.
 - Fixed profile saves for provisioned workspaces retaining their reserved slug. Thanks @sercada.
 - Fixed embedded channel formatting, GIF, profile, and image controls by sharing working composer behavior across views.

apps/api/internal/uploadstore/r2.go

@@ -84,13 +84,7 @@ func NewR2(cfg R2Config) (*R2, error) {
 }
 
 func defaultR2HTTPClient() *http.Client {
-	transport, ok := http.DefaultTransport.(*http.Transport)
-	if !ok {
-		return &http.Client{}
-	}
-	cloned := transport.Clone()
-	cloned.ResponseHeaderTimeout = defaultR2ResponseHeaderTimeout
-	return &http.Client{Transport: cloned}
+	return &http.Client{Transport: r2Transport{}}
 }
 
 func (s *R2) Save(ctx context.Context, body io.Reader, options SaveOptions) (SavedObject, error) {

apps/api/internal/uploadstore/r2_test.go

@@ -124,13 +124,6 @@ func TestR2ConfigValidation(t *testing.T) {
 	if store.httpClient == nil || store.httpClient.Timeout != 0 {
 		t.Fatalf("expected streaming-safe client timeout, got %#v", store.httpClient)
 	}
-	transport, ok := store.httpClient.Transport.(*http.Transport)
-	if !ok {
-		t.Fatalf("expected default transport, got %#v", store.httpClient.Transport)
-	}
-	if transport.ResponseHeaderTimeout != defaultR2ResponseHeaderTimeout {
-		t.Fatalf("expected response header timeout %s, got %s", defaultR2ResponseHeaderTimeout, transport.ResponseHeaderTimeout)
-	}
 	customClient := &http.Client{}
 	store, err = NewR2(R2Config{AccountID: "account", AccessKeyID: "access", SecretAccessKey: "secret", Bucket: "bucket", HTTPClient: customClient})
 	if err != nil {

apps/api/internal/uploadstore/r2_transport.go

@@ -0,0 +1,116 @@
+package uploadstore
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptrace"
+	"sync"
+	"time"
+)
+
+// r2Transport preserves the configured default transport and its trace hooks.
+// After-write timing requires a transport that emits Go's HTTP lifecycle hooks.
+type r2Transport struct{}
+
+func (r2Transport) RoundTrip(req *http.Request) (*http.Response, error) {
+	ctx, cancel := context.WithCancelCause(req.Context())
+	lifetime := &r2ResponseLifetime{cancel: cancel}
+	ctx = httptrace.WithClientTrace(ctx, &httptrace.ClientTrace{
+		GetConn:      func(string) { lifetime.retry() },
+		WroteRequest: lifetime.wroteRequest,
+	})
+	resp, err := http.DefaultTransport.RoundTrip(req.WithContext(ctx))
+	lifetime.headersComplete()
+	if err != nil || resp == nil || resp.Body == nil {
+		// Let http.Client retain its validation and empty-body normalization.
+		lifetime.close()
+		return resp, err
+	}
+	resp.Body = &r2ResponseBody{body: resp.Body, lifetime: lifetime}
+	return resp, nil
+}
+
+type r2ResponsePhase uint8
+
+const (
+	r2Headers r2ResponsePhase = iota
+	r2Body
+	r2Closed
+)
+
+type r2ResponseLifetime struct {
+	mu         sync.Mutex
+	phase      r2ResponsePhase
+	generation uint64
+	timer      *time.Timer
+	cancel     context.CancelCauseFunc
+}
+
+func (l *r2ResponseLifetime) stopTimerLocked() {
+	l.generation++
+	if l.timer != nil {
+		l.timer.Stop()
+		l.timer = nil
+	}
+}
+
+func (l *r2ResponseLifetime) wroteRequest(info httptrace.WroteRequestInfo) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if l.phase != r2Headers || info.Err != nil {
+		return
+	}
+	l.stopTimerLocked()
+	generation := l.generation
+	l.timer = time.AfterFunc(defaultR2ResponseHeaderTimeout, func() {
+		l.mu.Lock()
+		defer l.mu.Unlock()
+		// Stop does not join an already-started callback. Check and cancel
+		// under the same lock so it cannot cancel a later attempt or phase.
+		if l.phase == r2Headers && l.generation == generation {
+			l.cancel(context.DeadlineExceeded)
+		}
+	})
+}
+
+func (l *r2ResponseLifetime) retry() {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if l.phase == r2Headers {
+		l.stopTimerLocked()
+	}
+}
+
+func (l *r2ResponseLifetime) headersComplete() {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	l.phase = r2Body
+	l.stopTimerLocked()
+}
+
+func (l *r2ResponseLifetime) close() {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	l.phase = r2Closed
+	l.stopTimerLocked()
+	l.cancel(nil)
+}
+
+type r2ResponseBody struct {
+	body     io.ReadCloser
+	lifetime *r2ResponseLifetime
+}
+
+func (b *r2ResponseBody) Read(p []byte) (int, error) {
+	n, err := b.body.Read(p)
+	if err != nil {
+		b.lifetime.close()
+	}
+	return n, err
+}
+
+func (b *r2ResponseBody) Close() error {
+	b.lifetime.close()
+	return b.body.Close()
+}