mirror of
https://github.com/git-pkgs/proxy.git
synced 2026-11-09 15:49:16 -05:00
84111c387a
Instances sharing a Postgres database each fetched a missed artifact from upstream; only callers within one instance shared a fetch. The instance that fetches now holds a lease on the miss in a new fetch_leases table and renews it while it downloads. Another instance missing the same key polls the cache and serves the stored copy, or takes the lease over when the holder releases it without storing anything or stops renewing it. Metadata misses use the same lease when metadata caching is on with a TTL above zero; without a stored copy there is nothing to wait for. proxy_fetch_lease_waits_total counts the misses that waited. Reclaim deletes leases left by stopped instances. Refs #543
94 lines
3.1 KiB
Go
94 lines
3.1 KiB
Go
package server
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"github.com/git-pkgs/proxy/internal/database"
|
|
"github.com/git-pkgs/proxy/internal/storage"
|
|
)
|
|
|
|
// An object no record points at any more waits in the pending_deletes queue
|
|
// for a grace period before it is deleted, so a request that read the record
|
|
// before it changed can still open the object, and a signed URL to it stays
|
|
// valid. Reclaim runs whether or not a cache size limit is set.
|
|
const (
|
|
reclaimInterval = 1 * time.Minute
|
|
reclaimBatch = 100
|
|
reclaimMinGrace = 1 * time.Hour
|
|
// reclaimTickBudget bounds how long one tick keeps deleting batches, so
|
|
// a long queue, such as one left by a retention sweep, drains faster than
|
|
// one batch per interval without one tick running on indefinitely.
|
|
reclaimTickBudget = 30 * time.Second
|
|
)
|
|
|
|
func (s *Server) startReclaimLoop(ctx context.Context) {
|
|
grace := max(reclaimMinGrace, s.cfg.ParseDirectServeTTL())
|
|
|
|
ticker := time.NewTicker(reclaimInterval)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
s.runExclusive(ctx, "reclaim", func() {
|
|
reclaimDue(ctx, s.db, s.storage, s.logger, time.Now().Add(-grace), reclaimTickBudget)
|
|
// Leases left by instances that stopped while fetching.
|
|
if _, err := s.db.DeleteExpiredFetchLeases(reclaimMinGrace); err != nil {
|
|
s.logger.Warn("reclaim: failed to delete expired fetch leases", "error", err)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
}
|
|
|
|
// reclaimDue deletes batches of objects queued before cutoff until a batch
|
|
// comes back short, a delete fails or budget has passed. Stopping on a
|
|
// failed delete keeps a storage outage to one batch of attempts per tick.
|
|
func reclaimDue(ctx context.Context, db *database.DB, store storage.Storage, logger *slog.Logger,
|
|
cutoff time.Time, budget time.Duration) {
|
|
start := time.Now()
|
|
for ctx.Err() == nil && time.Since(start) < budget {
|
|
listed, failed := reclaimStorage(ctx, db, store, logger, cutoff)
|
|
if listed < reclaimBatch || failed > 0 {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// reclaimStorage deletes up to one batch of objects queued before cutoff and
|
|
// reports how many it listed and how many deletes failed. A delete that
|
|
// fails is queued again, behind the rest, so objects the backend keeps
|
|
// refusing cannot fill every batch.
|
|
func reclaimStorage(ctx context.Context, db *database.DB, store storage.Storage, logger *slog.Logger,
|
|
cutoff time.Time) (listed, failed int) {
|
|
paths, err := db.GetDuePendingDeletes(cutoff, reclaimBatch)
|
|
if err != nil {
|
|
logger.Warn("reclaim: failed to list pending deletes", "error", err)
|
|
return 0, 0
|
|
}
|
|
|
|
for _, path := range paths {
|
|
if ctx.Err() != nil {
|
|
return len(paths), failed
|
|
}
|
|
if err := store.Delete(ctx, path); err != nil {
|
|
if ctx.Err() != nil {
|
|
return len(paths), failed
|
|
}
|
|
failed++
|
|
logger.Warn("reclaim: failed to delete object, will retry", "path", path, "error", err)
|
|
if err := db.QueuePendingDelete(path); err != nil {
|
|
logger.Warn("reclaim: failed to requeue object", "path", path, "error", err)
|
|
}
|
|
continue
|
|
}
|
|
if err := db.RemovePendingDelete(path); err != nil {
|
|
logger.Warn("reclaim: failed to dequeue deleted object", "path", path, "error", err)
|
|
}
|
|
}
|
|
return len(paths), failed
|
|
}
|