at-container-registry/pkg/appview/holdpurge/queue.go

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) }