Expose fetch observation metadata (#58)

Author: andrew

README.md

@@ -351,6 +351,32 @@ io.Copy(dst, artifact.Body)
 
 The fetcher uses DNS caching (5-minute refresh), connection pooling, and a 5-minute timeout suited for large artifacts. It retries on rate limits and server errors with exponential backoff and jitter.
 
+### Observing artifact responses
+
+Use `FetchObserved` when the response metadata and downloaded content digests need to be retained:
+
+```go
+artifact, err := f.FetchObserved(ctx, url)
+if err != nil {
+    log.Fatal(err)
+}
+defer artifact.Body.Close()
+
+if _, err := io.Copy(dst, artifact.Body); err != nil {
+    log.Fatal(err)
+}
+if !artifact.Observation.Complete {
+    log.Fatal("artifact body did not reach EOF")
+}
+
+fmt.Println(artifact.Observation.RequestedURL)
+fmt.Println(artifact.Observation.FinalURL)
+fmt.Println(artifact.Observation.ByteCount)
+fmt.Println(artifact.Observation.Digests["sha256"])
+```
+
+The observation includes the time to receive the final response headers, status, declared size, media type, and an allow-list of response headers: `Accept-Ranges`, `Cache-Control`, `Content-Disposition`, `Content-Encoding`, `Content-Length`, `Content-Range`, `Digest`, `ETag`, `Expires`, and `Last-Modified`. SHA-256 and SHA-512 digests use lowercase hexadecimal encoding. Byte counts and digests remain unset until the stream reaches EOF, so a partial download cannot appear complete. Request and authentication headers are not copied into the observation.
+
 ### Per-request headers
 
 Use `FetchWithHeaders` to pass HTTP headers for a single request. This is useful when the auth token varies per request or is obtained dynamically (e.g. Docker Hub token exchange):

fetch/circuit_breaker.go

@@ -105,6 +105,37 @@ func (cbf *CircuitBreakerFetcher) FetchWithHeaders(ctx context.Context, fetchURL
 	return artifact, fetchErr
 }
 
+// FetchObserved wraps the underlying fetcher's FetchObserved with circuit breaker logic.
+func (cbf *CircuitBreakerFetcher) FetchObserved(ctx context.Context, fetchURL string) (*ObservedArtifact, error) {
+	return cbf.FetchObservedWithHeaders(ctx, fetchURL, nil)
+}
+
+// FetchObservedWithHeaders wraps the underlying fetcher's FetchObservedWithHeaders with circuit breaker logic.
+func (cbf *CircuitBreakerFetcher) FetchObservedWithHeaders(ctx context.Context, fetchURL string, headers http.Header) (*ObservedArtifact, error) {
+	registry := extractRegistry(fetchURL)
+	breaker := cbf.getBreaker(registry)
+
+	if !breaker.Ready() {
+		return nil, fmt.Errorf("circuit breaker open for registry %s: %w", registry, ErrUpstreamDown)
+	}
+
+	var artifact *ObservedArtifact
+	var fetchErr error
+	err := breaker.Call(func() error {
+		artifact, fetchErr = cbf.fetcher.FetchObservedWithHeaders(ctx, fetchURL, headers)
+		if errors.Is(fetchErr, ErrNotFound) {
+			return nil
+		}
+		return fetchErr
+	}, 0)
+
+	if err != nil {
+		return nil, err
+	}
+
+	return artifact, fetchErr
+}
+
 // Head wraps the underlying fetcher's Head with circuit breaker logic.
 func (cbf *CircuitBreakerFetcher) Head(ctx context.Context, headURL string) (size int64, contentType string, err error) {
 	registry := extractRegistry(headURL)

fetch/circuit_breaker_test.go

@@ -36,6 +36,27 @@ func TestCircuitBreakerFetch_Success(t *testing.T) {
 	}
 }
 
+func TestCircuitBreakerFetchObserved_Success(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte("test content"))
+	}))
+	defer server.Close()
+
+	cbFetcher := NewCircuitBreakerFetcher(NewFetcher())
+	artifact, err := cbFetcher.FetchObserved(context.Background(), server.URL+"/test.tar.gz")
+	if err != nil {
+		t.Fatalf("FetchObserved failed: %v", err)
+	}
+	defer func() { _ = artifact.Body.Close() }()
+
+	if _, err := io.ReadAll(artifact.Body); err != nil {
+		t.Fatalf("ReadAll failed: %v", err)
+	}
+	if !artifact.Observation.Complete {
+		t.Error("observation is incomplete after reading the response body")
+	}
+}
+
 func TestCircuitBreakerHead_Success(t *testing.T) {
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if r.Method != http.MethodHead {

fetch/fetcher.go

@@ -248,6 +248,27 @@ func (f *Fetcher) Fetch(ctx context.Context, url string) (*Artifact, error) {
 // FetchWithHeaders downloads an artifact from the given URL with additional HTTP headers.
 // The caller must close the returned Artifact.Body when done.
 func (f *Fetcher) FetchWithHeaders(ctx context.Context, url string, headers http.Header) (*Artifact, error) {
+	artifact, _, err := f.fetch(ctx, url, headers, false)
+	return artifact, err
+}
+
+// FetchObserved downloads an artifact and records metadata about the response.
+// The caller must read the body to EOF before treating the observation as complete.
+func (f *Fetcher) FetchObserved(ctx context.Context, url string) (*ObservedArtifact, error) {
+	return f.FetchObservedWithHeaders(ctx, url, nil)
+}
+
+// FetchObservedWithHeaders downloads an artifact with additional HTTP headers and
+// records metadata about the response. Request headers are not copied into the observation.
+func (f *Fetcher) FetchObservedWithHeaders(ctx context.Context, url string, headers http.Header) (*ObservedArtifact, error) {
+	artifact, observation, err := f.fetch(ctx, url, headers, true)
+	if err != nil {
+		return nil, err
+	}
+	return &ObservedArtifact{Artifact: artifact, Observation: observation}, nil
+}
+
+func (f *Fetcher) fetch(ctx context.Context, url string, headers http.Header, observe bool) (*Artifact, *FetchObservation, error) {
 	var lastErr error
 
 	for attempt := 0; attempt <= f.maxRetries; attempt++ {
@@ -259,21 +280,21 @@ func (f *Fetcher) FetchWithHeaders(ctx context.Context, url string, headers http
 
 			select {
 			case <-ctx.Done():
-				return nil, ctx.Err()
+				return nil, nil, ctx.Err()
 			case <-time.After(delay):
 			}
 		}
 
-		artifact, err := f.doFetch(ctx, url, headers)
+		artifact, observation, err := f.doFetch(ctx, url, headers, observe)
 		if err == nil {
-			return artifact, nil
+			return artifact, observation, nil
 		}
 
 		lastErr = err
 
 		// Don't retry on not found or client errors
 		if errors.Is(err, ErrNotFound) {
-			return nil, err
+			return nil, nil, err
 		}
 
 		// Retry on rate limit and server errors
@@ -282,16 +303,16 @@ func (f *Fetcher) FetchWithHeaders(ctx context.Context, url string, headers http
 		}
 
 		// Don't retry on other errors (network issues will be wrapped)
-		return nil, err
+		return nil, nil, err
 	}
 
-	return nil, lastErr
+	return nil, nil, lastErr
 }
 
-func (f *Fetcher) doFetch(ctx context.Context, url string, headers http.Header) (*Artifact, error) {
+func (f *Fetcher) doFetch(ctx context.Context, url string, headers http.Header, observe bool) (*Artifact, *FetchObservation, error) {
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
-		return nil, fmt.Errorf("creating request: %w", err)
+		return nil, nil, fmt.Errorf("creating request: %w", err)
 	}
 
 	req.Header.Set("User-Agent", f.userAgent)
@@ -311,9 +332,11 @@ func (f *Fetcher) doFetch(ctx context.Context, url string, headers http.Header)
 		}
 	}
 
+	startedAt := time.Now()
 	resp, err := f.client.Do(req)
+	responseTime := time.Since(startedAt)
 	if err != nil {
-		return nil, fmt.Errorf("fetching artifact: %w", err)
+		return nil, nil, fmt.Errorf("fetching artifact: %w", err)
 	}
 
 	switch {
@@ -325,29 +348,44 @@ func (f *Fetcher) doFetch(ctx context.Context, url string, headers http.Header)
 			}
 		}
 
-		return &Artifact{
+		artifact := &Artifact{
 			Body:        resp.Body,
 			Size:        size,
 			ContentType: resp.Header.Get("Content-Type"),
 			ETag:        resp.Header.Get("ETag"),
-		}, nil
+		}
+		if !observe {
+			return artifact, nil, nil
+		}
+
+		observation := &FetchObservation{
+			RequestedURL: url,
+			FinalURL:     resp.Request.URL.String(),
+			ResponseTime: responseTime,
+			StatusCode:   resp.StatusCode,
+			Headers:      copyObservedHeaders(resp.Header),
+			DeclaredSize: size,
+			MediaType:    resp.Header.Get("Content-Type"),
+		}
+		artifact.Body = newObservedBody(resp.Body, observation)
+		return artifact, observation, nil
 
 	case resp.StatusCode == http.StatusNotFound:
 		_ = resp.Body.Close()
-		return nil, ErrNotFound
+		return nil, nil, ErrNotFound
 
 	case resp.StatusCode == http.StatusTooManyRequests:
 		_ = resp.Body.Close()
-		return nil, ErrRateLimited
+		return nil, nil, ErrRateLimited
 
 	case resp.StatusCode >= serverErrThreshold:
 		_ = resp.Body.Close()
-		return nil, ErrUpstreamDown
+		return nil, nil, ErrUpstreamDown
 
 	default:
 		body, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrBodySize))
 		_ = resp.Body.Close()
-		return nil, fmt.Errorf("unexpected status %d: %s", resp.StatusCode, string(body))
+		return nil, nil, fmt.Errorf("unexpected status %d: %s", resp.StatusCode, string(body))
 	}
 }
 

fetch/observation.go

@@ -0,0 +1,95 @@
+package fetch
+
+import (
+	"crypto/sha256"
+	"crypto/sha512"
+	"encoding/hex"
+	"hash"
+	"io"
+	"net/http"
+	"time"
+)
+
+var observedResponseHeaders = []string{
+	"Accept-Ranges",
+	"Cache-Control",
+	"Content-Disposition",
+	"Content-Encoding",
+	"Content-Length",
+	"Content-Range",
+	"Digest",
+	"ETag",
+	"Expires",
+	"Last-Modified",
+}
+
+// FetchObservation contains metadata about a successful artifact response.
+// ByteCount and Digests are populated, and Complete is set to true, only after
+// the response body reaches EOF.
+type FetchObservation struct {
+	RequestedURL string
+	FinalURL     string
+	ResponseTime time.Duration // Time until the final response headers arrived.
+	StatusCode   int
+	Headers      http.Header // Allow-listed response headers only.
+	DeclaredSize int64
+	MediaType    string
+	ByteCount    int64
+	Digests      map[string]string
+	Complete     bool
+}
+
+// ObservedArtifact contains an artifact and its fetch observation.
+type ObservedArtifact struct {
+	*Artifact
+	Observation *FetchObservation
+}
+
+type observedBody struct {
+	body        io.ReadCloser
+	observation *FetchObservation
+	sha256      hash.Hash
+	sha512      hash.Hash
+	byteCount   int64
+}
+
+func newObservedBody(body io.ReadCloser, observation *FetchObservation) io.ReadCloser {
+	return &observedBody{
+		body:        body,
+		observation: observation,
+		sha256:      sha256.New(),
+		sha512:      sha512.New(),
+	}
+}
+
+func (b *observedBody) Read(p []byte) (int, error) {
+	n, err := b.body.Read(p)
+	if n > 0 {
+		b.byteCount += int64(n)
+		_, _ = b.sha256.Write(p[:n])
+		_, _ = b.sha512.Write(p[:n])
+	}
+	if err == io.EOF && !b.observation.Complete {
+		b.observation.ByteCount = b.byteCount
+		b.observation.Digests = map[string]string{
+			"sha256": hex.EncodeToString(b.sha256.Sum(nil)),
+			"sha512": hex.EncodeToString(b.sha512.Sum(nil)),
+		}
+		b.observation.Complete = true
+	}
+	return n, err
+}
+
+func (b *observedBody) Close() error {
+	return b.body.Close()
+}
+
+func copyObservedHeaders(headers http.Header) http.Header {
+	observed := make(http.Header)
+	for _, name := range observedResponseHeaders {
+		if values := headers.Values(name); len(values) > 0 {
+			observed[http.CanonicalHeaderKey(name)] = append([]string(nil), values...)
+		}
+	}
+	return observed
+}