feat: wire the circuit breaker into the proxy fetch path (#90)

Fixes #74

## Why
`internal/proxy/circuit.go` implemented and tested a circuit breaker, but nothing ever called it — a repeatedly-failing upstream was still hit on every request.

## Changes
- Construct a `CircuitBreaker` in `NewEngine`.
- In `Engine.Fetch`: short-circuit when the breaker is open (serve stale from the store if present, otherwise return 503), `RecordFailure` on each `UpstreamError`, and `RecordSuccess` on a successful fetch.

## Validation
- `go test ./internal/proxy/` and `make e2e` pass.

---------

Co-authored-by: BenVincent <benvin@main.unkin.net>
Reviewed-on: #90
Co-authored-by: Ben Vincent <ben@unkin.net>
Co-committed-by: Ben Vincent <ben@unkin.net>

internal/database/artifacts.go

@@ -138,16 +138,22 @@ func (db *DB) InsertAccessLogBatch(ctx context.Context, entries []AccessLogEntry
 	return err
 }
 
-func (db *DB) FindOrphanedBlobs(ctx context.Context) ([]models.Blob, error) {
+// FindOrphanedBlobs returns blobs no longer referenced by any artifact or
+// local file, restricted to those created before now()-minAge. The age cutoff
+// is a grace period that avoids a TOCTOU race with in-flight dedup uploads,
+// which insert the blob row before the referencing artifact/local_files row.
+func (db *DB) FindOrphanedBlobs(ctx context.Context, minAge time.Duration) ([]models.Blob, error) {
+	cutoff := time.Now().Add(-minAge)
 	rows, err := db.Pool.Query(ctx, `
 		SELECT b.content_hash, b.s3_key, b.size_bytes, b.content_type, b.created_at
 		FROM blobs b
-		WHERE b.content_hash NOT IN (
+		WHERE b.created_at < $1
+		  AND b.content_hash NOT IN (
 			SELECT content_hash FROM artifacts
 			UNION
 			SELECT content_hash FROM local_files
 		)
-	`)
+	`, cutoff)
 	if err != nil {
 		return nil, err
 	}

internal/gc/gc.go

@@ -9,6 +9,11 @@ import (
 	"git.unkin.net/unkin/artifactapi/internal/storage"
 )
 
+// blobGracePeriod is how old an orphaned blob must be before GC will delete
+// it. This avoids racing in-flight dedup uploads that insert the blob row
+// before the referencing artifact/local_files row exists.
+const blobGracePeriod = 1 * time.Hour
+
 type Collector struct {
 	db       *database.DB
 	store    *storage.S3
@@ -38,7 +43,7 @@ func (c *Collector) Run(ctx context.Context) {
 func (c *Collector) sweep(ctx context.Context) {
 	start := time.Now()
 
-	orphaned, err := c.db.FindOrphanedBlobs(ctx)
+	orphaned, err := c.db.FindOrphanedBlobs(ctx, blobGracePeriod)
 	if err != nil {
 		slog.Error("gc: find orphaned blobs", "error", err)
 		return

internal/proxy/engine.go

@@ -2,6 +2,8 @@ package proxy
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -252,7 +254,7 @@ func (e *Engine) headUpstream(ctx context.Context, remote models.Remote, path st
 	}
 	if resp.StatusCode == http.StatusUnauthorized {
 		resp.Body.Close()
-		token, terr := fetchBearerToken(ctx, resp.Header.Get("Www-Authenticate"), remote)
+		token, _, terr := fetchBearerToken(ctx, resp.Header.Get("Www-Authenticate"), remote)
 		if terr == nil && token != "" {
 			resp, err = doHead(http.Header{"Authorization": []string{"Bearer " + token}})
 			if err != nil {
@@ -533,6 +535,11 @@ const (
 	bearerTokenTTLMargin  = 10 * time.Second
 )
 
+func sha256Hash(data []byte) string {
+	h := sha256.Sum256(data)
+	return hex.EncodeToString(h[:])
+}
+
 // cachedBearerToken returns a bearer token for the given challenge, reusing a
 // Redis-cached token for the same remote+challenge while it is still valid.
 func (e *Engine) cachedBearerToken(ctx context.Context, wwwAuth string, remote models.Remote) (string, error) {