diff --git a/pkg/appview/handlers/images.go b/pkg/appview/handlers/images.go index 43fccee..d0e505f 100644 --- a/pkg/appview/handlers/images.go +++ b/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++ } diff --git a/pkg/appview/handlers/purge_enqueue_test.go b/pkg/appview/handlers/purge_enqueue_test.go new file mode 100644 index 0000000..6030c11 --- /dev/null +++ b/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") + } +} diff --git a/pkg/appview/holdpurge/purge.go b/pkg/appview/holdpurge/purge.go index f41af6f..d1f33bd 100644 --- a/pkg/appview/holdpurge/purge.go +++ b/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 } diff --git a/pkg/appview/holdpurge/queue.go b/pkg/appview/holdpurge/queue.go new file mode 100644 index 0000000..938d522 --- /dev/null +++ b/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) } diff --git a/pkg/appview/holdpurge/queue_test.go b/pkg/appview/holdpurge/queue_test.go new file mode 100644 index 0000000..6af899c --- /dev/null +++ b/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") + } +} diff --git a/pkg/appview/server.go b/pkg/appview/server.go index 5a8d86f..18114f7 100644 --- a/pkg/appview/server.go +++ b/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. //