appview: run the hold purge on a worker pool, not the request context

Deleting a tag while over quota could leave the user in the worst available
state. DeleteTagHandler deleted the tag and manifest rows first, then called
PurgeOnHold, which bounded itself at 10s against the *request* context. The
UpCloud load balancer in front of the appview cuts at its default backend
timeout at about the same moment, wins the race, hands the client a 504 and
cancels that context, killing the purge partway. The appview logged a warning
and returned 200.

So: gateway error, nothing freed, still locked out, the image gone from the UI
so the purge cannot be retried through it, blobs orphaned until the hold's GC,
and the appview considering it a success. Measured on production at
10.002367218s.

Purges now go to a fixed pool of 4 workers rooted at context.Background(), so
they survive the request ending. Following the shape of the hold's startJob
helper, minus the progress fragment, since nobody is watching a purge.

The buffer is bounded at 256 and sheds with an ERROR rather than growing: an
unbounded queue turns a slow hold into an appview memory leak. Submissions are
deduplicated on holdDID|manifestURI so a double-clicked delete does one purge
and one service-token fetch. The channel send happens under the mutex that
guards close, so a concurrent drain cannot send on a closed channel, and the
drain is wired into both exit paths before logging shuts down.

Failures are now classified and surfaced instead of swallowed: transient ones
retry three times under a 90s budget (the hold's purge is idempotent), an
unauthorized third-party hold logs at DEBUG since it is expected, and anything
else that exhausts its retries logs at ERROR naming the manifest and hold, which
is enough to re-drive by hand.

Deliberately not reordered. Purge-first-then-delete requires waiting for the
purge to know whether to delete, which puts the 10s call straight back on the
request. So the orphaned-blob window remains, materially narrower but real: a
purge that exhausts its retries still leaves blobs referenced by nothing until
the hold's GC, and there is no row left to say so. Closing that needs a durable
pending-purge record, which was judged out of proportion here.

server.go in this commit also carries one line belonging to the next one, the
token handler's display-name wiring, since the two changes landed in the same
file concurrently.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PDqoCE1j3njokkZ9b1C5n9

pkg/appview/handlers/images.go

@@ -101,7 +101,17 @@ func (h *DeleteTagHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			if err := db.DeleteManifest(h.DB, user.DID, repo, digest); err != nil {
 				slog.Warn("delete-tag: cascade DB delete failed", "did", user.DID, "digest", digest, "error", err)
 			}
-			holdpurge.PurgeOnHold(r.Context(), h.Refresher, user.DID, user.PDSEndpoint, holdDID, atproto.BuildManifestURI(user.DID, digest))
+			// Hand the purge to the background queue and return. It must
+			// not run on r.Context(): the proxy in front of the appview
+			// cuts the request at ~10s, which used to cancel the purge and
+			// hand the user a 504 for a delete that had already succeeded.
+			holdpurge.Enqueue(holdpurge.Request{
+				Refresher:   h.Refresher,
+				UserDID:     user.DID,
+				PDSEndpoint: user.PDSEndpoint,
+				HoldDID:     holdDID,
+				ManifestURI: atproto.BuildManifestURI(user.DID, digest),
+			})
 		}
 	}
 
@@ -251,10 +261,18 @@ func (h *DeleteManifestHandler) ServeHTTP(w http.ResponseWriter, r *http.Request
 	}
 
 	// Tell the hold to drop its layer/scan/image-config records for this
-	// manifest. Best-effort — failures here only mean the hold's lazy GC
-	// will clean up later, so we don't reflect the failure to the user.
-	manifestURI := atproto.BuildManifestURI(user.DID, digest)
-	holdpurge.PurgeOnHold(r.Context(), h.Refresher, user.DID, user.PDSEndpoint, holdDID, manifestURI)
+	// manifest. Queued, not inline: the call outlives the proxy's request
+	// window often enough that running it here returned a 504 to the user
+	// (and cancelled the purge) for a delete that had already succeeded.
+	// The response below therefore reports the deletion, not the reclaim —
+	// the space comes back once the queued purge lands.
+	holdpurge.Enqueue(holdpurge.Request{
+		Refresher:   h.Refresher,
+		UserDID:     user.DID,
+		PDSEndpoint: user.PDSEndpoint,
+		HoldDID:     holdDID,
+		ManifestURI: atproto.BuildManifestURI(user.DID, digest),
+	})
 
 	w.WriteHeader(http.StatusOK)
 }
@@ -357,8 +375,16 @@ func (h *DeleteUntaggedManifestsHandler) ServeHTTP(w http.ResponseWriter, r *htt
 			continue
 		}
 
-		manifestURI := atproto.BuildManifestURI(user.DID, digest)
-		holdpurge.PurgeOnHold(r.Context(), h.Refresher, user.DID, user.PDSEndpoint, holdDID, manifestURI)
+		// Queued rather than inline — a bulk delete of N manifests would
+		// otherwise serialize N hold round trips onto one request and blow
+		// through the proxy's timeout well before the loop finished.
+		holdpurge.Enqueue(holdpurge.Request{
+			Refresher:   h.Refresher,
+			UserDID:     user.DID,
+			PDSEndpoint: user.PDSEndpoint,
+			HoldDID:     holdDID,
+			ManifestURI: atproto.BuildManifestURI(user.DID, digest),
+		})
 
 		deleted++
 	}

pkg/appview/handlers/purge_enqueue_test.go

@@ -0,0 +1,51 @@
+package handlers
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// TestDeleteHandlersEnqueuePurgeRatherThanBlock guards the fix for the delete
+// 504: the hold purge is a cross-service call that regularly outlives the
+// proxy's request window, so a delete handler must hand it to the background
+// queue instead of running it inline on r.Context().
+//
+// Running it inline destroyed the DB rows first and then died when the proxy
+// cancelled the request, leaving the user with a 504, no space reclaimed, and
+// no way to retry through the UI.
+//
+// This is a source-level assertion because the inline call is only reachable
+// through a full OAuth + PDS round trip, which this package has no fixture for;
+// the runtime behaviour is covered in pkg/appview/holdpurge.
+func TestDeleteHandlersEnqueuePurgeRatherThanBlock(t *testing.T) {
+	files, err := filepath.Glob("*.go")
+	if err != nil {
+		t.Fatalf("glob: %v", err)
+	}
+
+	sawEnqueue := false
+	for _, file := range files {
+		if strings.HasSuffix(file, "_test.go") {
+			continue
+		}
+		src, err := os.ReadFile(file)
+		if err != nil {
+			t.Fatalf("read %s: %v", file, err)
+		}
+		text := string(src)
+
+		if strings.Contains(text, "holdpurge.PurgeOnHold(") {
+			t.Errorf("%s calls holdpurge.PurgeOnHold inline; request handlers must "+
+				"use holdpurge.Enqueue so the purge survives the request ending", file)
+		}
+		if strings.Contains(text, "holdpurge.Enqueue(") {
+			sawEnqueue = true
+		}
+	}
+
+	if !sawEnqueue {
+		t.Error("no handler enqueues a hold purge; the delete path lost its purge entirely")
+	}
+}

pkg/appview/holdpurge/purge.go

@@ -2,12 +2,19 @@
 // image-config) associated with a deleted manifest. It lives in its own leaf
 // package so both the web-UI handlers (pkg/appview/handlers) and the storage
 // routing layer (pkg/appview/storage) can call it without an import cycle.
+//
+// The call is a cross-service HTTP round trip that regularly takes longer than
+// the reverse proxy in front of the appview is willing to wait, so UI handlers
+// must not make it inline on the request goroutine — see queue.go and
+// Enqueue().
 package holdpurge
 
 import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
+	"fmt"
 	"io"
 	"log/slog"
 	"net/http"
@@ -18,86 +25,124 @@ import (
 	"atcr.io/pkg/auth/oauth"
 )
 
+// AttemptTimeout bounds a single purge attempt. The queue gives a job a longer
+// total budget than this so a retry still has room to run.
+const AttemptTimeout = 10 * time.Second
+
+// ErrNotAuthorized means the acting user holds no captain/crew-admin right on
+// the target hold. Expected whenever a sailor pushes to a third-party hold, so
+// it is not worth retrying and not worth an operator's attention: the hold's
+// lazy GC reclaims the blobs on its own schedule.
+var ErrNotAuthorized = errors.New("not authorized to purge on this hold")
+
+// permanentError marks a failure that will never succeed on retry (a malformed
+// request, a missing refresher). Transport failures and hold-side 5xx are left
+// unwrapped so the queue retries them.
+type permanentError struct{ err error }
+
+func (e permanentError) Error() string { return e.err.Error() }
+func (e permanentError) Unwrap() error { return e.err }
+
+func permanent(err error) error { return permanentError{err} }
+
+// IsPermanent reports whether err is known not to be worth retrying.
+func IsPermanent(err error) bool {
+	if errors.Is(err, ErrNotAuthorized) {
+		return true
+	}
+	var p permanentError
+	return errors.As(err, &p)
+}
+
 // purgeManifestRequest is the JSON body sent to io.atcr.hold.purgeManifest.
 type purgeManifestRequest struct {
 	ManifestURI string `json:"manifestUri"`
 }
 
-// PurgeOnHold tells the hold to delete the layer, scan, and image-config
-// records associated with a single manifest. This is best-effort: callers
-// should treat all errors as "log and continue" because lazy GC on the hold
-// will catch up either way (and on third-party holds the user may not even
-// have the captain/crew-admin permission needed for the call to succeed).
+// PurgeOnHold is the fire-and-forget form of Purge: it logs the outcome and
+// discards the error. Kept for callers that already run detached and have
+// nowhere to report a failure (pkg/appview/storage).
+//
+// Prefer Enqueue() from a request handler — this call blocks for up to
+// AttemptTimeout, which is long enough for a reverse proxy to give up on the
+// request first.
+func PurgeOnHold(ctx context.Context, refresher *oauth.Refresher, userDID, pdsEndpoint, holdDID, manifestURI string) {
+	err := Purge(ctx, refresher, userDID, pdsEndpoint, holdDID, manifestURI)
+	switch {
+	case err == nil:
+	case errors.Is(err, ErrNotAuthorized):
+		slog.Debug("PurgeOnHold: not authorized on hold (lazy GC will handle)",
+			"hold_did", holdDID, "manifest", manifestURI)
+	default:
+		slog.Warn("PurgeOnHold: purge failed",
+			"hold_did", holdDID, "manifest", manifestURI, "user_did", userDID, "error", err)
+	}
+}
+
+// Purge tells the hold to delete the layer, scan, and image-config records
+// associated with a single manifest, returning the outcome so the caller can
+// retry or escalate. The hold's implementation is idempotent, so a retry after
+// an ambiguous failure is safe.
 //
 // holdDID identifies which hold owns the manifest's blobs (typically the
 // `hold_endpoint` column on the manifests row, or a freshly-resolved value
 // from the manifest record). userDID + pdsEndpoint are the OAuth-acting
 // user — the service token is minted from their PDS with audience = holdDID.
-func PurgeOnHold(ctx context.Context, refresher *oauth.Refresher, userDID, pdsEndpoint, holdDID, manifestURI string) {
+//
+// ctx must not be a request context: this makes two network round trips and
+// routinely outlives the HTTP request that triggered it.
+func Purge(ctx context.Context, refresher *oauth.Refresher, userDID, pdsEndpoint, holdDID, manifestURI string) error {
 	if holdDID == "" || manifestURI == "" {
-		return
+		return nil
 	}
 	if refresher == nil {
-		slog.Debug("PurgeOnHold: OAuth refresher unavailable; skipping",
-			"hold_did", holdDID, "manifest", manifestURI)
-		return
+		// Nothing about a later attempt would supply one.
+		return permanent(errors.New("OAuth refresher unavailable"))
 	}
 
-	timeoutCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	timeoutCtx, cancel := context.WithTimeout(ctx, AttemptTimeout)
 	defer cancel()
 
 	holdURL, err := atproto.ResolveHoldURL(timeoutCtx, holdDID)
 	if err != nil {
-		slog.Warn("PurgeOnHold: failed to resolve hold URL",
-			"hold_did", holdDID, "error", err)
-		return
+		return fmt.Errorf("resolve hold URL: %w", err)
 	}
 
 	serviceToken, err := auth.GetOrFetchServiceToken(timeoutCtx, refresher, userDID, holdDID, pdsEndpoint)
 	if err != nil {
-		slog.Warn("PurgeOnHold: failed to mint service token",
-			"hold_did", holdDID, "user_did", userDID, "error", err)
-		return
+		return fmt.Errorf("mint service token: %w", err)
 	}
 
 	body, err := json.Marshal(purgeManifestRequest{ManifestURI: manifestURI})
 	if err != nil {
-		slog.Warn("PurgeOnHold: failed to marshal request",
-			"hold_did", holdDID, "error", err)
-		return
+		return permanent(fmt.Errorf("marshal request: %w", err))
 	}
 
 	req, err := http.NewRequestWithContext(timeoutCtx, http.MethodPost,
 		holdURL+atproto.HoldPurgeManifest, bytes.NewReader(body))
 	if err != nil {
-		slog.Warn("PurgeOnHold: failed to create request",
-			"hold_did", holdDID, "error", err)
-		return
+		return permanent(fmt.Errorf("create request: %w", err))
 	}
 	req.Header.Set("Authorization", "Bearer "+serviceToken)
 	req.Header.Set("Content-Type", "application/json")
 
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
-		slog.Warn("PurgeOnHold: request failed",
-			"hold_did", holdDID, "manifest", manifestURI, "error", err)
-		return
+		return fmt.Errorf("post %s: %w", atproto.HoldPurgeManifest, err)
 	}
 	defer resp.Body.Close()
 
 	if resp.StatusCode == http.StatusForbidden || resp.StatusCode == http.StatusUnauthorized {
-		// Sailor pushing to a third-party hold won't have captain/crew-admin
-		// rights; that's expected. Lazy GC on that hold will reclaim later.
-		slog.Debug("PurgeOnHold: not authorized on hold (lazy GC will handle)",
-			"hold_did", holdDID, "manifest", manifestURI, "status", resp.StatusCode)
-		return
+		return ErrNotAuthorized
 	}
 	if resp.StatusCode != http.StatusOK {
-		body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
-		slog.Warn("PurgeOnHold: hold returned non-OK status",
-			"hold_did", holdDID, "manifest", manifestURI,
-			"status", resp.StatusCode, "body", string(body))
-		return
+		msg, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
+		err := fmt.Errorf("hold returned %d: %s", resp.StatusCode, string(msg))
+		// 4xx other than the auth pair describes the request, not the moment.
+		if resp.StatusCode >= 400 && resp.StatusCode < 500 {
+			return permanent(err)
+		}
+		return err
 	}
 
 	var out struct {
@@ -107,9 +152,9 @@ func PurgeOnHold(ctx context.Context, refresher *oauth.Refresher, userDID, pdsEn
 		ImageConfigDeleted bool `json:"imageConfigDeleted"`
 	}
 	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
-		slog.Warn("PurgeOnHold: failed to parse response",
-			"hold_did", holdDID, "manifest", manifestURI, "error", err)
-		return
+		// The purge itself already happened; only our reading of the reply
+		// failed. Retrying would be harmless but pointless.
+		return permanent(fmt.Errorf("parse response: %w", err))
 	}
 
 	slog.Info("PurgeOnHold: purge succeeded",
@@ -119,4 +164,5 @@ func PurgeOnHold(ctx context.Context, refresher *oauth.Refresher, userDID, pdsEn
 		"scan_deleted", out.ScanDeleted,
 		"image_config_deleted", out.ImageConfigDeleted,
 	)
+	return nil
 }

pkg/appview/holdpurge/queue.go

@@ -0,0 +1,283 @@
+package holdpurge
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"sync"
+	"time"
+
+	"atcr.io/pkg/auth/oauth"
+)
+
+// A manifest purge is a two-hop network call (resolve the hold, mint a service
+// token, then POST) that regularly takes longer than the UpCloud load balancer
+// in front of the appview will wait. Running it inline on a delete handler
+// produced the worst possible outcome: the proxy returned 504 to the user at
+// ~10s and cancelled r.Context(), which killed the purge mid-flight, while the
+// appview logged a WARN and returned 200 to a client that had already given up.
+// The rows were gone, the space was not freed, and nothing retried.
+//
+// So the purge runs here instead, on a detached background worker, in the same
+// shape as pkg/hold/admin's startJob helper (and its peer gc.startBackground):
+// the handler hands off and returns immediately, and the work runs under its
+// own context.Background() timeout so the request ending cannot cancel it.
+// Unlike startJob there is no progress fragment to poll — nobody is watching a
+// purge — so instead of a job registry keyed by operation this keeps a fixed
+// worker pool with an in-flight dedupe set keyed by the manifest being purged.
+
+const (
+	// defaultWorkers bounds how many purges run at once. Rapid deletes queue
+	// rather than spawning a goroutine (and a service-token fetch) apiece.
+	defaultWorkers = 4
+
+	// defaultBuffer bounds how much work may be waiting. A full buffer sheds
+	// load loudly rather than growing without limit.
+	defaultBuffer = 256
+
+	// jobBudget is the total time one queued purge may take across all of its
+	// attempts, including backoff.
+	jobBudget = 90 * time.Second
+
+	// maxAttempts includes the first try. The hold's purge is idempotent, so
+	// retrying an ambiguous failure is safe.
+	maxAttempts = 3
+
+	// retryBackoff is the wait before attempt N+1.
+	retryBackoff = 3 * time.Second
+)
+
+// Request is one manifest's worth of purge work. It carries every value the
+// purge needs, because the queue deliberately does not carry the request
+// context: nothing in the purge path reads request-scoped context values (the
+// acting user and the OAuth refresher are the only request-derived inputs, and
+// both are fields here), so detaching loses nothing.
+type Request struct {
+	Refresher   *oauth.Refresher
+	UserDID     string
+	PDSEndpoint string
+	HoldDID     string
+	ManifestURI string
+}
+
+func (r Request) key() string { return r.HoldDID + "|" + r.ManifestURI }
+
+// Queue runs manifest purges on a bounded pool of background workers.
+type Queue struct {
+	jobs chan Request
+
+	// purge is the unit of work, and attempts/backoff its retry schedule.
+	// Fields rather than package constants so tests can drive the worker
+	// without a live hold and without waiting out the real backoff.
+	purge    func(context.Context, Request) error
+	attempts int
+	backoff  time.Duration
+
+	// ctx is rooted at context.Background(), never at a request. Cancelled
+	// only by Wait, to abort in-flight attempts once the drain grace period
+	// has elapsed.
+	ctx    context.Context
+	cancel context.CancelFunc
+	wg     sync.WaitGroup
+
+	mu       sync.Mutex
+	closed   bool
+	inflight map[string]struct{}
+}
+
+// NewQueue starts a queue with the given worker count and buffer depth. Both
+// fall back to the package defaults when non-positive.
+func NewQueue(workers, buffer int) *Queue {
+	return newQueue(workers, buffer, func(ctx context.Context, req Request) error {
+		return Purge(ctx, req.Refresher, req.UserDID, req.PDSEndpoint, req.HoldDID, req.ManifestURI)
+	})
+}
+
+func newQueue(workers, buffer int, purge func(context.Context, Request) error) *Queue {
+	if workers <= 0 {
+		workers = defaultWorkers
+	}
+	if buffer <= 0 {
+		buffer = defaultBuffer
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	q := &Queue{
+		jobs:     make(chan Request, buffer),
+		purge:    purge,
+		attempts: maxAttempts,
+		backoff:  retryBackoff,
+		ctx:      ctx,
+		cancel:   cancel,
+		inflight: make(map[string]struct{}, buffer),
+	}
+	q.wg.Add(workers)
+	for i := 0; i < workers; i++ {
+		go q.run()
+	}
+	return q
+}
+
+// Submit hands a purge to the background workers and returns immediately. It
+// reports whether the request was accepted.
+//
+// It is rejected, and false returned, when the manifest is already queued or
+// running (a double-clicked delete), when the queue is saturated, or after the
+// queue has been drained for shutdown. A rejection is not silent: it is logged,
+// and the hold's lazy GC still reclaims the blobs.
+func (q *Queue) Submit(req Request) bool {
+	if req.HoldDID == "" || req.ManifestURI == "" {
+		// Nothing to purge — no hold recorded for this manifest. Lazy GC only.
+		return false
+	}
+
+	q.mu.Lock()
+	if q.closed {
+		q.mu.Unlock()
+		slog.Warn("holdpurge: queue is shutting down, purge skipped",
+			"hold_did", req.HoldDID, "manifest", req.ManifestURI)
+		return false
+	}
+	key := req.key()
+	if _, dup := q.inflight[key]; dup {
+		q.mu.Unlock()
+		slog.Debug("holdpurge: purge already queued for manifest",
+			"hold_did", req.HoldDID, "manifest", req.ManifestURI)
+		return false
+	}
+	// The send happens under the same lock that guards close(), so a worker
+	// draining concurrently can never make this a send on a closed channel.
+	select {
+	case q.jobs <- req:
+		q.inflight[key] = struct{}{}
+		q.mu.Unlock()
+		return true
+	default:
+		q.mu.Unlock()
+		// Shedding is deliberate: an unbounded queue would turn a slow hold
+		// into an appview memory leak. Loud, because a saturated queue means
+		// purges are being lost.
+		slog.Error("holdpurge: purge queue full, manifest not purged (hold GC will reclaim later)",
+			"hold_did", req.HoldDID, "manifest", req.ManifestURI, "user_did", req.UserDID,
+			"queue_depth", len(q.jobs))
+		return false
+	}
+}
+
+// run is one worker. It exits when the jobs channel is closed and drained.
+func (q *Queue) run() {
+	defer q.wg.Done()
+	for req := range q.jobs {
+		q.process(req)
+		q.mu.Lock()
+		delete(q.inflight, req.key())
+		q.mu.Unlock()
+	}
+}
+
+// process runs one purge under a detached budget, retrying transient failures.
+func (q *Queue) process(req Request) {
+	ctx, cancel := context.WithTimeout(q.ctx, jobBudget)
+	defer cancel()
+
+	var err error
+	for attempt := 1; attempt <= q.attempts; attempt++ {
+		err = q.purge(ctx, req)
+		if err == nil {
+			return
+		}
+		if IsPermanent(err) {
+			break
+		}
+		if attempt == q.attempts {
+			break
+		}
+		slog.Warn("holdpurge: purge attempt failed, retrying",
+			"hold_did", req.HoldDID, "manifest", req.ManifestURI,
+			"attempt", attempt, "error", err)
+
+		timer := time.NewTimer(q.backoff)
+		select {
+		case <-timer.C:
+		case <-ctx.Done():
+			timer.Stop()
+			slog.Error("holdpurge: purge abandoned, blobs remain until hold GC",
+				"hold_did", req.HoldDID, "manifest", req.ManifestURI,
+				"user_did", req.UserDID, "error", ctx.Err())
+			return
+		}
+		timer.Stop()
+	}
+
+	if errors.Is(err, ErrNotAuthorized) {
+		// A sailor on a third-party hold has no purge right. Expected, and
+		// the hold's own GC handles it.
+		slog.Debug("holdpurge: not authorized on hold (lazy GC will handle)",
+			"hold_did", req.HoldDID, "manifest", req.ManifestURI)
+		return
+	}
+
+	// This is the state the user complained about: the manifest is deleted but
+	// the space is not back. There is no durable retry queue in the appview, so
+	// this log line is the only record — keep it at ERROR and keep the manifest
+	// URI in it, so an operator can re-drive the purge by hand (or wait for the
+	// hold's GC to reclaim the blobs).
+	slog.Error("holdpurge: purge failed after retries, blobs remain until hold GC",
+		"hold_did", req.HoldDID, "manifest", req.ManifestURI,
+		"user_did", req.UserDID, "attempts", q.attempts, "error", err)
+}
+
+// Wait stops accepting new work and blocks until the in-flight purges finish or
+// the grace period elapses, whichever comes first. Called from the appview's
+// shutdown path so a purge started by the last request before SIGTERM is not
+// silently dropped on the floor. Safe to call more than once.
+func (q *Queue) Wait(grace time.Duration) {
+	q.mu.Lock()
+	if q.closed {
+		q.mu.Unlock()
+		return
+	}
+	q.closed = true
+	pending := len(q.jobs)
+	close(q.jobs)
+	q.mu.Unlock()
+
+	if pending > 0 {
+		slog.Info("holdpurge: draining purge queue", "pending", pending, "grace", grace)
+	}
+
+	done := make(chan struct{})
+	go func() {
+		q.wg.Wait()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(grace):
+		slog.Warn("holdpurge: purge queue did not drain in time, abandoning",
+			"grace", grace)
+	}
+	// Unblocks any attempt still in flight past the grace period so its
+	// goroutine cannot outlive the process's shutdown path.
+	q.cancel()
+}
+
+var (
+	defaultQueueOnce sync.Once
+	defaultQueue     *Queue
+)
+
+// Default is the process-wide purge queue used by the UI delete handlers.
+func Default() *Queue {
+	defaultQueueOnce.Do(func() {
+		defaultQueue = NewQueue(defaultWorkers, defaultBuffer)
+	})
+	return defaultQueue
+}
+
+// Enqueue submits a purge to the default queue. This is what a delete handler
+// should call: it returns without waiting for the hold.
+func Enqueue(req Request) bool { return Default().Submit(req) }
+
+// Wait drains the default queue. Called from the appview shutdown path.
+func Wait(grace time.Duration) { Default().Wait(grace) }

pkg/appview/holdpurge/queue_test.go

@@ -0,0 +1,367 @@
+package holdpurge
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+func testRequest(manifest string) Request {
+	return Request{
+		UserDID:     "did:plc:testuser",
+		PDSEndpoint: "https://pds.example",
+		HoldDID:     "did:web:hold.example",
+		ManifestURI: manifest,
+	}
+}
+
+// captureLogs swaps the default slog handler for the duration of a test and
+// returns a function yielding everything logged so far.
+func captureLogs(t *testing.T, level slog.Level) func() string {
+	t.Helper()
+	var mu sync.Mutex
+	buf := &bytes.Buffer{}
+	prev := slog.Default()
+	slog.SetDefault(slog.New(slog.NewTextHandler(&lockedWriter{mu: &mu, buf: buf}, &slog.HandlerOptions{Level: level})))
+	t.Cleanup(func() { slog.SetDefault(prev) })
+	return func() string {
+		mu.Lock()
+		defer mu.Unlock()
+		return buf.String()
+	}
+}
+
+type lockedWriter struct {
+	mu  *sync.Mutex
+	buf *bytes.Buffer
+}
+
+func (w *lockedWriter) Write(p []byte) (int, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.buf.Write(p)
+}
+
+// TestSubmitReturnsWithoutWaitingForPurge is the shape of the fix: the delete
+// handler hands the purge off and returns. Before, it blocked on the hold for
+// up to AttemptTimeout, which is longer than the proxy in front of the appview
+// waits — so the user got a 504 for a delete that had succeeded.
+func TestSubmitReturnsWithoutWaitingForPurge(t *testing.T) {
+	release := make(chan struct{})
+	started := make(chan struct{})
+	done := make(chan struct{})
+
+	q := newQueue(1, 4, func(ctx context.Context, req Request) error {
+		close(started)
+		<-release
+		close(done)
+		return nil
+	})
+
+	start := time.Now()
+	if !q.Submit(testRequest("at://did:plc:testuser/io.atcr.manifest/abc")) {
+		t.Fatal("Submit rejected the purge")
+	}
+	elapsed := time.Since(start)
+
+	// Generous bound: the point is "does not wait for the hold", and the
+	// worker below is blocked indefinitely until we release it.
+	if elapsed > time.Second {
+		t.Fatalf("Submit blocked for %v; it must hand off and return", elapsed)
+	}
+
+	select {
+	case <-started:
+	case <-time.After(5 * time.Second):
+		t.Fatal("purge never started on a worker")
+	}
+
+	close(release)
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("purge never finished")
+	}
+	q.Wait(5 * time.Second)
+}
+
+// TestPurgeRunsAfterRequestContextCancelled is the defect itself: the proxy
+// cutting the request cancelled the context the purge was running on, so the
+// purge died mid-flight. The queued purge must run on a context rooted at
+// context.Background(), unaffected by the request ending.
+func TestPurgeRunsAfterRequestContextCancelled(t *testing.T) {
+	type observation struct {
+		errAtStart error
+		errLater   error
+	}
+	observed := make(chan observation, 1)
+
+	q := newQueue(1, 4, func(ctx context.Context, req Request) error {
+		o := observation{errAtStart: ctx.Err()}
+		// Give the cancelled request context every chance to propagate.
+		time.Sleep(50 * time.Millisecond)
+		o.errLater = ctx.Err()
+		observed <- o
+		return nil
+	})
+
+	// Stand in for the request goroutine: enqueue, then die. Submit takes no
+	// context at all, which is the structural half of the fix — there is no
+	// longer a way for a handler to hand the purge its request deadline.
+	_, cancelRequest := context.WithCancel(context.Background())
+	if !q.Submit(testRequest("at://did:plc:testuser/io.atcr.manifest/cancelled")) {
+		t.Fatal("Submit rejected the purge")
+	}
+	cancelRequest()
+
+	select {
+	case o := <-observed:
+		if o.errAtStart != nil {
+			t.Fatalf("purge ran on an already-cancelled context: %v", o.errAtStart)
+		}
+		if o.errLater != nil {
+			t.Fatalf("purge context was cancelled by the request ending: %v", o.errLater)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("purge did not run after the request context was cancelled")
+	}
+	q.Wait(5 * time.Second)
+}
+
+// TestPurgeFailureIsSurfaced: a purge that never succeeds must not disappear.
+// It retries, and the final failure is logged at ERROR with the manifest URI,
+// because the appview has no durable record of work still owed to the hold.
+func TestPurgeFailureIsSurfaced(t *testing.T) {
+	logs := captureLogs(t, slog.LevelDebug)
+
+	var attempts int
+	var mu sync.Mutex
+	finished := make(chan struct{})
+
+	q := newQueue(1, 4, func(ctx context.Context, req Request) error {
+		mu.Lock()
+		attempts++
+		n := attempts
+		mu.Unlock()
+		if n == maxAttempts {
+			defer close(finished)
+		}
+		return errors.New("hold unreachable")
+	})
+	q.backoff = time.Millisecond
+
+	if !q.Submit(testRequest("at://did:plc:testuser/io.atcr.manifest/doomed")) {
+		t.Fatal("Submit rejected the purge")
+	}
+
+	select {
+	case <-finished:
+	case <-time.After(5 * time.Second):
+		t.Fatal("purge did not exhaust its attempts")
+	}
+	q.Wait(5 * time.Second)
+
+	mu.Lock()
+	got := attempts
+	mu.Unlock()
+	if got != maxAttempts {
+		t.Errorf("attempts = %d, want %d", got, maxAttempts)
+	}
+
+	out := logs()
+	if !strings.Contains(out, "level=ERROR") {
+		t.Errorf("failed purge was not logged at ERROR:\n%s", out)
+	}
+	if !strings.Contains(out, "purge failed after retries") {
+		t.Errorf("failed purge did not name itself in the log:\n%s", out)
+	}
+	if !strings.Contains(out, "io.atcr.manifest/doomed") {
+		t.Errorf("failed purge log does not identify the manifest:\n%s", out)
+	}
+}
+
+// A permanent failure (no OAuth refresher, a malformed request, a 4xx from the
+// hold) must not burn retries, but must still be surfaced.
+func TestPermanentFailureIsNotRetried(t *testing.T) {
+	logs := captureLogs(t, slog.LevelDebug)
+
+	var mu sync.Mutex
+	var attempts int
+	q := newQueue(1, 4, func(ctx context.Context, req Request) error {
+		mu.Lock()
+		attempts++
+		mu.Unlock()
+		return permanent(errors.New("malformed"))
+	})
+	q.backoff = time.Millisecond
+
+	if !q.Submit(testRequest("at://did:plc:testuser/io.atcr.manifest/permanent")) {
+		t.Fatal("Submit rejected the purge")
+	}
+	q.Wait(5 * time.Second)
+
+	mu.Lock()
+	got := attempts
+	mu.Unlock()
+	if got != 1 {
+		t.Errorf("attempts = %d, want 1 for a permanent failure", got)
+	}
+	if !strings.Contains(logs(), "level=ERROR") {
+		t.Errorf("permanent failure was not surfaced:\n%s", logs())
+	}
+}
+
+// A sailor purging on a third-party hold has no right to; that is expected and
+// handled by the hold's own GC, so it must not page anyone.
+func TestNotAuthorizedIsNotAnError(t *testing.T) {
+	logs := captureLogs(t, slog.LevelDebug)
+
+	q := newQueue(1, 4, func(ctx context.Context, req Request) error {
+		return ErrNotAuthorized
+	})
+	q.backoff = time.Millisecond
+
+	if !q.Submit(testRequest("at://did:plc:testuser/io.atcr.manifest/thirdparty")) {
+		t.Fatal("Submit rejected the purge")
+	}
+	q.Wait(5 * time.Second)
+
+	if strings.Contains(logs(), "level=ERROR") {
+		t.Errorf("an unauthorized third-party purge should not log at ERROR:\n%s", logs())
+	}
+}
+
+// Rapid repeat deletes of the same manifest must not each spawn work.
+func TestSubmitDeduplicatesInFlightManifest(t *testing.T) {
+	release := make(chan struct{})
+	var mu sync.Mutex
+	var runs int
+
+	q := newQueue(1, 8, func(ctx context.Context, req Request) error {
+		mu.Lock()
+		runs++
+		mu.Unlock()
+		<-release
+		return nil
+	})
+
+	req := testRequest("at://did:plc:testuser/io.atcr.manifest/dupe")
+	if !q.Submit(req) {
+		t.Fatal("first Submit rejected")
+	}
+	if q.Submit(req) {
+		t.Error("second Submit for an in-flight manifest should be rejected")
+	}
+
+	close(release)
+	q.Wait(5 * time.Second)
+
+	mu.Lock()
+	defer mu.Unlock()
+	if runs != 1 {
+		t.Errorf("runs = %d, want 1", runs)
+	}
+}
+
+// A saturated queue sheds load rather than growing without bound, and says so.
+func TestSubmitShedsWhenQueueIsFull(t *testing.T) {
+	logs := captureLogs(t, slog.LevelDebug)
+
+	release := make(chan struct{})
+	q := newQueue(1, 1, func(ctx context.Context, req Request) error {
+		<-release
+		return nil
+	})
+
+	// One job occupies the worker, one fills the single buffer slot, the
+	// third has nowhere to go.
+	accepted := 0
+	for i := 0; i < 3; i++ {
+		if q.Submit(testRequest(fmt.Sprintf("at://did:plc:testuser/io.atcr.manifest/%d", i))) {
+			accepted++
+		}
+		// Let the worker pick the first job up so the buffer is the limit.
+		if i == 0 {
+			time.Sleep(50 * time.Millisecond)
+		}
+	}
+	if accepted > 2 {
+		t.Errorf("accepted %d submissions into a queue of depth 1", accepted)
+	}
+	if !strings.Contains(logs(), "purge queue full") {
+		t.Errorf("shedding was silent:\n%s", logs())
+	}
+
+	close(release)
+	q.Wait(5 * time.Second)
+}
+
+// Wait must drain the queued purges rather than let SIGTERM drop them, and it
+// must return even when a purge is wedged.
+func TestWaitDrainsAndDoesNotLeak(t *testing.T) {
+	var mu sync.Mutex
+	var completed int
+	q := newQueue(2, 8, func(ctx context.Context, req Request) error {
+		time.Sleep(20 * time.Millisecond)
+		mu.Lock()
+		completed++
+		mu.Unlock()
+		return nil
+	})
+
+	for i := 0; i < 4; i++ {
+		if !q.Submit(testRequest(fmt.Sprintf("at://did:plc:testuser/io.atcr.manifest/drain%d", i))) {
+			t.Fatalf("Submit %d rejected", i)
+		}
+	}
+
+	q.Wait(5 * time.Second)
+
+	mu.Lock()
+	got := completed
+	mu.Unlock()
+	if got != 4 {
+		t.Errorf("completed = %d, want 4 (shutdown dropped queued purges)", got)
+	}
+
+	// Post-drain submissions are refused rather than panicking on a closed
+	// channel, and a second Wait is a no-op.
+	if q.Submit(testRequest("at://did:plc:testuser/io.atcr.manifest/after")) {
+		t.Error("Submit after Wait should be rejected")
+	}
+	q.Wait(time.Second)
+}
+
+// Wait must not hang forever on a wedged purge; it gives up and cancels the
+// worker's context so the goroutine cannot outlive shutdown.
+func TestWaitGivesUpOnWedgedPurge(t *testing.T) {
+	observedCancel := make(chan struct{})
+	q := newQueue(1, 2, func(ctx context.Context, req Request) error {
+		<-ctx.Done()
+		close(observedCancel)
+		return ctx.Err()
+	})
+	q.backoff = time.Millisecond
+
+	if !q.Submit(testRequest("at://did:plc:testuser/io.atcr.manifest/wedged")) {
+		t.Fatal("Submit rejected")
+	}
+
+	start := time.Now()
+	q.Wait(100 * time.Millisecond)
+	if elapsed := time.Since(start); elapsed > 3*time.Second {
+		t.Fatalf("Wait blocked for %v past its grace period", elapsed)
+	}
+
+	select {
+	case <-observedCancel:
+	case <-time.After(5 * time.Second):
+		t.Fatal("wedged purge was never cancelled after the drain grace period")
+	}
+}

pkg/appview/server.go

@@ -26,6 +26,7 @@ import (
 	"atcr.io/pkg/appview/authgate"
 	"atcr.io/pkg/appview/db"
 	"atcr.io/pkg/appview/holdhealth"
+	"atcr.io/pkg/appview/holdpurge"
 	"atcr.io/pkg/appview/jetstream"
 	appviewlabeler "atcr.io/pkg/appview/labeler"
 	"atcr.io/pkg/appview/leases"
@@ -630,6 +631,11 @@ func NewAppViewServer(cfg *Config, branding *BrandingOverrides) (*AppViewServer,
 		// the audience names the front door actually used.
 		tokenHandler.SetServices(cfg.Auth.Services)
 
+		// cfg.Auth.Services is port-stripped so audiences match the routing
+		// host; the raw list is what a `docker login` command has to name. The
+		// display list feeds the plain-text 401 guidance only.
+		tokenHandler.SetServiceDisplayNames(cfg.Server.RegistryDomains)
+
 		tokenHandler.SetOAuthSessionValidator(s.Refresher)
 
 		// Auth-phase gate: crew reconciliation for any token request, plus
@@ -780,15 +786,21 @@ func (s *AppViewServer) ServeWithListener(listener net.Listener) error {
 		defer cancel()
 
 		if err := s.httpServer.Shutdown(shutdownCtx); err != nil && err != http.ErrServerClosed {
+			s.drainPurgeQueue()
 			logging.Shutdown()
 			return fmt.Errorf("server shutdown error: %w", err)
 		}
+
+		// After Shutdown: no request can enqueue new purges by this point, so
+		// the drain is finite.
+		s.drainPurgeQueue()
 	case err := <-serveErr:
 		s.healthWorker.Stop()
 		if s.workerCancel != nil {
 			s.workerCancel()
 		}
 		s.waitForLeasedWorkers()
+		s.drainPurgeQueue()
 		logging.Shutdown()
 		if err != nil {
 			return fmt.Errorf("server error: %w", err)
@@ -990,6 +1002,19 @@ func (s *AppViewServer) handleDIDDocument(w http.ResponseWriter, r *http.Request
 	}
 }
 
+// drainPurgeQueue waits for the backgrounded hold purges to finish before the
+// process exits.
+//
+// A purge is started by a delete handler and deliberately outlives its request
+// (see pkg/appview/holdpurge), so without this a SIGTERM arriving just after a
+// delete would drop the purge on the floor: the manifest gone, the blobs still
+// counted against the user's quota until the hold's GC catches up. Bounded, so
+// a wedged hold cannot hold the shutdown open.
+func (s *AppViewServer) drainPurgeQueue() {
+	const purgeDrainTimeout = 15 * time.Second
+	holdpurge.Wait(purgeDrainTimeout)
+}
+
 // waitForLeasedWorkers gives the leased background workers a moment to stop and
 // release their leases before the process exits.
 //