mirror of
https://tangled.org/evan.jarrett.net/at-container-registry
synced 2026-09-20 01:04:15 +00:00
clean up GC implementation
This commit is contained in:
Evan Jarrett
@@ -54,7 +54,7 @@ registration:
|
||||
# Create a wildcard crew record allowing any authenticated user to join.
|
||||
allow_all_crew: false
|
||||
# URL to fetch avatar image from during bootstrap.
|
||||
profile_avatar_url: https://imgs.blue/evan.jarrett.net/1TpTOdtS60GdJWBYEqtK22y688jajbQ9a5kbYRFtwuqrkBAE
|
||||
profile_avatar_url: https://atcr.io/web-app-manifest-192x192.png
|
||||
# Post to Bluesky when users push images. Synced to captain record on startup.
|
||||
enable_bluesky_posts: false
|
||||
# Deployment region, auto-detected from cloud metadata or S3 config.
|
||||
@@ -75,6 +75,10 @@ database:
|
||||
admin:
|
||||
# Enable the web-based admin panel for crew and storage management.
|
||||
enabled: true
|
||||
# Garbage collection settings.
|
||||
gc:
|
||||
# Enable nightly garbage collection of orphaned blobs and records.
|
||||
enabled: false
|
||||
# Storage quota tiers. Empty disables quota enforcement.
|
||||
quota:
|
||||
# Quota tiers keyed by rank name. Each tier has a human-readable quota limit.
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
# Config Blob Storage Decision
|
||||
|
||||
## Background
|
||||
|
||||
OCI image manifests reference two types of blobs:
|
||||
|
||||
1. **Layers** — filesystem diffs (tar+gzip), typically large, content-addressed and shared across users
|
||||
2. **Config blob** — small JSON (~2-15KB) containing image metadata: architecture, OS, environment variables, entrypoint, Dockerfile build history, and labels
|
||||
|
||||
In ATCR, manifests are stored in the user's PDS while all blobs (layers and config) are stored in S3 via the hold service. The hold tracks layers with `io.atcr.hold.layer` records but has no equivalent tracking for config blobs.
|
||||
|
||||
## Considered: Storing Config Blobs in PDS
|
||||
|
||||
Config blobs are unique per image build — unlike layers which are deduplicated across users, a config blob contains the specific Dockerfile history, env vars, and labels for that build. This makes them conceptually "user data" that could belong in the user's PDS alongside the manifest.
|
||||
|
||||
The proposal was to add a `ConfigBlob` field to `ManifestRecord`, uploading the config blob to PDS during push (the data is already fetched from S3 for label extraction). The config would remain in S3 as well since the distribution library puts it there during the blob push phase.
|
||||
|
||||
Potential benefits:
|
||||
- Manifests become more self-contained in PDS
|
||||
- Config metadata (entrypoint, env, history) available without S3 access (e.g., for web UI)
|
||||
- Aligns with the principle that user-specific data belongs in the user's PDS
|
||||
|
||||
## Decision: Keep Config Blobs in S3 Only
|
||||
|
||||
Config blobs can contain sensitive data:
|
||||
|
||||
- **Environment variables** — `ENV DATABASE_URL=...`, `ENV API_KEY=...` set in Dockerfiles
|
||||
- **Build history** — `history[].created_by` reveals exact Dockerfile commands, internal registry URLs, build arguments
|
||||
- **Labels** — may contain internal metadata not intended for public consumption
|
||||
|
||||
ATProto has no private data. The current storage split creates a useful privacy boundary:
|
||||
|
||||
| Storage | Visibility | Contains |
|
||||
|---------|-----------|----------|
|
||||
| PDS | Public (anyone) | Manifest structure, tags, repo names, annotations |
|
||||
| Hold/S3 | Auth-gated | Layers + config — actual image content |
|
||||
|
||||
This boundary enables **semi-private repos**: the public PDS metadata tells you what images exist (names, tags, sizes), but you cannot reconstruct or run the image without hold access. Storing config in PDS would break this — build secrets and Dockerfile history would be publicly readable even when the hold restricts blob access.
|
||||
|
||||
We considered making PDS storage optional (only for fully public holds or allow-all-crew holds), but an optional field that can't be relied upon adds complexity without clear benefit — the config must live in S3 regardless for the pull path.
|
||||
|
||||
## Current Status
|
||||
|
||||
Config blobs remain in S3 behind hold authorization. GC handles config digests to prevent orphaned deletion (config digests are included in the referenced set alongside layer digests).
|
||||
|
||||
## Revisit If
|
||||
|
||||
- ATProto adds private data support
|
||||
- A concrete use case emerges that requires PDS-native config access
|
||||
File diff suppressed because one or more lines are too long
@@ -19,7 +19,10 @@
|
||||
<symbol id="database" viewBox="0 0 24 24"><ellipse cx="12" cy="5" rx="9" ry="3"/><path d="M3 5V19A9 3 0 0 0 21 19V5"/><path d="M3 12A9 3 0 0 0 21 12"/></symbol>
|
||||
<symbol id="download" viewBox="0 0 24 24"><path d="M12 15V3"/><path d="M21 15v4a2 2 0 0 1-2 2H5a2 2 0 0 1-2-2v-4"/><path d="m7 10 5 5 5-5"/></symbol>
|
||||
<symbol id="eye" viewBox="0 0 24 24"><path d="M2.062 12.348a1 1 0 0 1 0-.696 10.75 10.75 0 0 1 19.876 0 1 1 0 0 1 0 .696 10.75 10.75 0 0 1-19.876 0"/><circle cx="12" cy="12" r="3"/></symbol>
|
||||
<symbol id="file-plus" viewBox="0 0 24 24"><path d="M6 22a2 2 0 0 1-2-2V4a2 2 0 0 1 2-2h8a2.4 2.4 0 0 1 1.704.706l3.588 3.588A2.4 2.4 0 0 1 20 8v12a2 2 0 0 1-2 2z"/><path d="M14 2v5a1 1 0 0 0 1 1h5"/><path d="M9 15h6"/><path d="M12 18v-6"/></symbol>
|
||||
<symbol id="file-x" viewBox="0 0 24 24"><path d="M6 22a2 2 0 0 1-2-2V4a2 2 0 0 1 2-2h8a2.4 2.4 0 0 1 1.704.706l3.588 3.588A2.4 2.4 0 0 1 20 8v12a2 2 0 0 1-2 2z"/><path d="M14 2v5a1 1 0 0 0 1 1h5"/><path d="m14.5 12.5-5 5"/><path d="m9.5 12.5 5 5"/></symbol>
|
||||
<symbol id="fingerprint" viewBox="0 0 24 24"><path d="M12 10a2 2 0 0 0-2 2c0 1.02-.1 2.51-.26 4"/><path d="M14 13.12c0 2.38 0 6.38-1 8.88"/><path d="M17.29 21.02c.12-.6.43-2.3.5-3.02"/><path d="M2 12a10 10 0 0 1 18-6"/><path d="M2 16h.01"/><path d="M21.8 16c.2-2 .131-5.354 0-6"/><path d="M5 19.5C5.5 18 6 15 6 12a6 6 0 0 1 .34-2"/><path d="M8.65 22c.21-.66.45-1.32.57-2"/><path d="M9 6.8a6 6 0 0 1 9 5.2v2"/></symbol>
|
||||
<symbol id="git-merge" viewBox="0 0 24 24"><circle cx="18" cy="18" r="3"/><circle cx="6" cy="6" r="3"/><path d="M6 21V9a9 9 0 0 0 9 9"/></symbol>
|
||||
<symbol id="github" viewBox="0 0 24 24"><path d="M15 22v-4a4.8 4.8 0 0 0-1-3.5c3 0 6-2 6-5.5.08-1.25-.27-2.48-1-3.5.28-1.15.28-2.35 0-3.5 0 0-1 0-3 1.5-2.64-.5-5.36-.5-8 0C6 2 5 2 5 2c-.3 1.15-.3 2.35 0 3.5A5.403 5.403 0 0 0 4 9c0 3.5 3 5.5 6 5.5-.39.49-.68 1.05-.85 1.65-.17.6-.22 1.23-.15 1.85v4"/><path d="M9 18c-4.51 2-5-2-7-2"/></symbol>
|
||||
<symbol id="hard-drive" viewBox="0 0 24 24"><line x1="22" x2="2" y1="12" y2="12"/><path d="M5.45 5.11 2 12v6a2 2 0 0 0 2 2h16a2 2 0 0 0 2-2v-6l-3.45-6.89A2 2 0 0 0 16.76 4H7.24a2 2 0 0 0-1.79 1.11z"/><line x1="6" x2="6.01" y1="16" y2="16"/><line x1="10" x2="10.01" y1="16" y2="16"/></symbol>
|
||||
<symbol id="info" viewBox="0 0 24 24"><circle cx="12" cy="12" r="10"/><path d="M12 16v-4"/><path d="M12 8h.01"/></symbol>
|
||||
@@ -46,5 +49,6 @@
|
||||
<symbol id="user" viewBox="0 0 24 24"><path d="M19 21v-2a4 4 0 0 0-4-4H9a4 4 0 0 0-4 4v2"/><circle cx="12" cy="7" r="4"/></symbol>
|
||||
<symbol id="user-plus" viewBox="0 0 24 24"><path d="M16 21v-2a4 4 0 0 0-4-4H6a4 4 0 0 0-4 4v2"/><circle cx="9" cy="7" r="4"/><line x1="19" x2="19" y1="8" y2="14"/><line x1="22" x2="16" y1="11" y2="11"/></symbol>
|
||||
<symbol id="x-circle" viewBox="0 0 24 24"><circle cx="12" cy="12" r="10"/><path d="m15 9-6 6"/><path d="m9 9 6 6"/></symbol>
|
||||
<symbol id="zap" viewBox="0 0 24 24"><path d="M4 14a1 1 0 0 1-.78-1.63l9.9-10.2a.5.5 0 0 1 .86.46l-1.92 6.02A1 1 0 0 0 13 10h7a1 1 0 0 1 .78 1.63l-9.9 10.2a.5.5 0 0 1-.86-.46l1.92-6.02A1 1 0 0 0 11 14z"/></symbol>
|
||||
<symbol id="helm" viewBox="0 0 24 24"><path d="M12.337 0c-.475 0-.861 1.016-.861 2.269 0 .527.069 1.011.183 1.396a8.514 8.514 0 0 0-3.961 1.22 5.229 5.229 0 0 0-.595-1.093c-.606-.866-1.34-1.436-1.79-1.43a.381.381 0 0 0-.217.066c-.39.273-.123 1.326.596 2.353.267.381.559.705.84.948a8.683 8.683 0 0 0-1.528 1.716h1.734a7.179 7.179 0 0 1 5.381-2.421 7.18 7.18 0 0 1 5.382 2.42h1.733a8.687 8.687 0 0 0-1.32-1.53c.35-.249.735-.643 1.078-1.133.719-1.027.986-2.08.596-2.353a.382.382 0 0 0-.217-.065c-.45-.007-1.184.563-1.79 1.43a4.897 4.897 0 0 0-.676 1.325 8.52 8.52 0 0 0-3.899-1.42c.12-.39.193-.887.193-1.429 0-1.253-.386-2.269-.862-2.269zM1.624 9.443v5.162h1.358v-1.968h1.64v1.968h1.357V9.443H4.62v1.838H2.98V9.443zm5.912 0v5.162h3.21v-1.108H8.893v-.95h1.64v-1.142h-1.64v-.84h1.853V9.443zm4.698 0v5.162h3.218v-1.362h-1.86v-3.8zm4.706 0v5.162h1.364v-2.643l1.357 1.225 1.35-1.232v2.65h1.365V9.443h-.614l-2.1 1.914-2.109-1.914zm-11.82 7.28a8.688 8.688 0 0 0 1.412 1.548 5.206 5.206 0 0 0-.841.948c-.719 1.027-.985 2.08-.596 2.353.39.273 1.289-.338 2.007-1.364a5.23 5.23 0 0 0 .595-1.092 8.514 8.514 0 0 0 3.961 1.219 5.01 5.01 0 0 0-.183 1.396c0 1.253.386 2.269.861 2.269.476 0 .862-1.016.862-2.269 0-.542-.072-1.04-.193-1.43a8.52 8.52 0 0 0 3.9-1.42c.121.4.352.865.675 1.327.719 1.026 1.617 1.637 2.007 1.364.39-.273.123-1.326-.596-2.353-.343-.49-.727-.885-1.077-1.135a8.69 8.69 0 0 0 1.202-1.36h-1.771a7.174 7.174 0 0 1-5.227 2.252 7.174 7.174 0 0 1-5.226-2.252z" fill="currentColor" stroke="none"/></symbol>
|
||||
</svg>
|
||||
|
Before Width: | Height: | Size: 11 KiB After Width: | Height: | Size: 12 KiB |
+21
-1
@@ -23,6 +23,7 @@ import (
|
||||
"time"
|
||||
|
||||
"atcr.io/pkg/atproto"
|
||||
"atcr.io/pkg/hold/gc"
|
||||
"atcr.io/pkg/hold/pds"
|
||||
"atcr.io/pkg/hold/quota"
|
||||
|
||||
@@ -63,6 +64,7 @@ type AdminSession struct {
|
||||
type AdminUI struct {
|
||||
pds *pds.HoldPDS
|
||||
quotaMgr *quota.Manager
|
||||
gc *gc.GarbageCollector
|
||||
clientApp *indigooauth.ClientApp
|
||||
templates map[string]*template.Template
|
||||
config AdminConfig
|
||||
@@ -76,7 +78,7 @@ type AdminUI struct {
|
||||
type adminContextKey struct{}
|
||||
|
||||
// NewAdminUI creates a new admin UI instance
|
||||
func NewAdminUI(ctx context.Context, holdPDS *pds.HoldPDS, quotaMgr *quota.Manager, cfg AdminConfig) (*AdminUI, error) {
|
||||
func NewAdminUI(ctx context.Context, holdPDS *pds.HoldPDS, quotaMgr *quota.Manager, garbageCollector *gc.GarbageCollector, cfg AdminConfig) (*AdminUI, error) {
|
||||
if !cfg.Enabled {
|
||||
return nil, nil
|
||||
}
|
||||
@@ -139,6 +141,7 @@ func NewAdminUI(ctx context.Context, holdPDS *pds.HoldPDS, quotaMgr *quota.Manag
|
||||
ui := &AdminUI{
|
||||
pds: holdPDS,
|
||||
quotaMgr: quotaMgr,
|
||||
gc: garbageCollector,
|
||||
clientApp: clientApp,
|
||||
templates: templates,
|
||||
config: cfg,
|
||||
@@ -226,6 +229,15 @@ func parseTemplates() (map[string]*template.Template, error) {
|
||||
return s[:n] + "..."
|
||||
},
|
||||
"formatBytes": formatHumanBytes,
|
||||
"formatDuration": func(d time.Duration) string {
|
||||
if d < time.Second {
|
||||
return fmt.Sprintf("%dms", d.Milliseconds())
|
||||
}
|
||||
if d < time.Minute {
|
||||
return fmt.Sprintf("%.1fs", d.Seconds())
|
||||
}
|
||||
return d.Round(time.Second).String()
|
||||
},
|
||||
"formatTime": func(t time.Time) string {
|
||||
return t.Format("2006-01-02 15:04")
|
||||
},
|
||||
@@ -364,6 +376,7 @@ func (ui *AdminUI) RegisterRoutes(r chi.Router) {
|
||||
r.Get("/admin/api/tab/crew", ui.handleCrewTab)
|
||||
r.Get("/admin/api/tab/settings", ui.handleSettingsTab)
|
||||
r.Get("/admin/api/tab/relays", ui.handleRelaysTab)
|
||||
r.Get("/admin/api/tab/storage", ui.handleGCTab)
|
||||
|
||||
// Backward-compat redirects for old bookmarks
|
||||
r.Get("/admin/crew", func(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -375,6 +388,9 @@ func (ui *AdminUI) RegisterRoutes(r chi.Router) {
|
||||
r.Get("/admin/relays", func(w http.ResponseWriter, r *http.Request) {
|
||||
http.Redirect(w, r, "/admin#relays", http.StatusFound)
|
||||
})
|
||||
r.Get("/admin/storage", func(w http.ResponseWriter, r *http.Request) {
|
||||
http.Redirect(w, r, "/admin#storage", http.StatusFound)
|
||||
})
|
||||
|
||||
// Crew sub-pages (full page, unchanged)
|
||||
r.Get("/admin/crew/add", ui.handleCrewAddForm)
|
||||
@@ -395,6 +411,10 @@ func (ui *AdminUI) RegisterRoutes(r chi.Router) {
|
||||
r.Post("/admin/relays/crawl", ui.handleRelayCrawl)
|
||||
r.Post("/admin/relays/crawl-all", ui.handleRelayCrawlAll)
|
||||
|
||||
// GC POSTs
|
||||
r.Post("/admin/api/gc/preview", ui.handleGCPreview)
|
||||
r.Post("/admin/api/gc/run", ui.handleGCRun)
|
||||
|
||||
// API endpoints (for HTMX)
|
||||
r.Get("/admin/api/stats", ui.handleStatsAPI)
|
||||
r.Get("/admin/api/top-users", ui.handleTopUsersAPI)
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
package admin
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"atcr.io/pkg/hold/gc"
|
||||
)
|
||||
|
||||
// gcTabData is the data passed to the storage/GC tab partial
|
||||
type gcTabData struct {
|
||||
Enabled bool
|
||||
Running bool
|
||||
|
||||
HasPreview bool
|
||||
LastPreviewAge string
|
||||
|
||||
HasResult bool
|
||||
LastResultAge string
|
||||
}
|
||||
|
||||
// handleGCTab returns the storage/GC tab content (HTMX partial)
|
||||
func (ui *AdminUI) handleGCTab(w http.ResponseWriter, r *http.Request) {
|
||||
if ui.gc == nil {
|
||||
ui.renderTemplate(w, "partials/tab_storage.html", gcTabData{})
|
||||
return
|
||||
}
|
||||
|
||||
cfg := ui.gc.GetConfig()
|
||||
_, previewAt := ui.gc.LastPreview()
|
||||
_, resultAt := ui.gc.LastResult()
|
||||
|
||||
data := gcTabData{
|
||||
Enabled: cfg.Enabled,
|
||||
Running: ui.gc.IsRunning(),
|
||||
HasPreview: !previewAt.IsZero(),
|
||||
LastPreviewAge: timeAgo(previewAt),
|
||||
HasResult: !resultAt.IsZero(),
|
||||
LastResultAge: timeAgo(resultAt),
|
||||
}
|
||||
|
||||
ui.renderTemplate(w, "partials/tab_storage.html", data)
|
||||
}
|
||||
|
||||
// handleGCPreview runs a GC preview (analysis only, no mutations)
|
||||
func (ui *AdminUI) handleGCPreview(w http.ResponseWriter, r *http.Request) {
|
||||
if ui.gc == nil {
|
||||
ui.renderTemplate(w, "partials/gc_error.html", struct{ Error string }{"GC not available"})
|
||||
return
|
||||
}
|
||||
|
||||
preview, err := ui.gc.Preview(r.Context())
|
||||
if err != nil {
|
||||
slog.Error("GC preview failed", "error", err)
|
||||
ui.renderTemplate(w, "partials/gc_error.html", struct{ Error string }{err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
session := getSessionFromContext(r.Context())
|
||||
slog.Info("GC preview completed via admin panel",
|
||||
"orphanedRecords", len(preview.OrphanedRecords),
|
||||
"orphanedBlobs", len(preview.OrphanedBlobs),
|
||||
"missingRecords", len(preview.MissingRecords),
|
||||
"duration", preview.Duration,
|
||||
"by", session.DID)
|
||||
|
||||
ui.renderTemplate(w, "partials/gc_preview.html", struct {
|
||||
Preview *gc.GCPreview
|
||||
}{Preview: preview})
|
||||
}
|
||||
|
||||
// handleGCRun executes an actual GC run
|
||||
func (ui *AdminUI) handleGCRun(w http.ResponseWriter, r *http.Request) {
|
||||
if ui.gc == nil {
|
||||
ui.renderTemplate(w, "partials/gc_error.html", struct{ Error string }{"GC not available"})
|
||||
return
|
||||
}
|
||||
|
||||
result, err := ui.gc.Run(r.Context())
|
||||
if err != nil {
|
||||
slog.Error("GC run failed", "error", err)
|
||||
ui.renderTemplate(w, "partials/gc_error.html", struct{ Error string }{err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
session := getSessionFromContext(r.Context())
|
||||
slog.Info("GC run completed via admin panel",
|
||||
"blobsDeleted", result.BlobsDeleted,
|
||||
"bytesReclaimed", result.BytesReclaimed,
|
||||
"recordsDeleted", result.RecordsDeleted,
|
||||
"recordsReconciled", result.RecordsReconciled,
|
||||
"duration", result.Duration,
|
||||
"by", session.DID)
|
||||
|
||||
ui.renderTemplate(w, "partials/gc_result.html", struct {
|
||||
Result *gc.GCResult
|
||||
}{Result: result})
|
||||
}
|
||||
|
||||
// timeAgo returns a human-readable relative time string
|
||||
func timeAgo(t time.Time) string {
|
||||
if t.IsZero() {
|
||||
return ""
|
||||
}
|
||||
d := time.Since(t)
|
||||
switch {
|
||||
case d < time.Minute:
|
||||
return "just now"
|
||||
case d < time.Hour:
|
||||
mins := int(d.Minutes())
|
||||
if mins == 1 {
|
||||
return "1 minute ago"
|
||||
}
|
||||
return fmt.Sprintf("%d minutes ago", mins)
|
||||
case d < 24*time.Hour:
|
||||
hours := int(d.Hours())
|
||||
if hours == 1 {
|
||||
return "1 hour ago"
|
||||
}
|
||||
return fmt.Sprintf("%d hours ago", hours)
|
||||
default:
|
||||
days := int(d.Hours() / 24)
|
||||
if days == 1 {
|
||||
return "1 day ago"
|
||||
}
|
||||
return fmt.Sprintf("%d days ago", days)
|
||||
}
|
||||
}
|
||||
File diff suppressed because one or more lines are too long
@@ -19,7 +19,10 @@
|
||||
<symbol id="database" viewBox="0 0 24 24"><ellipse cx="12" cy="5" rx="9" ry="3"/><path d="M3 5V19A9 3 0 0 0 21 19V5"/><path d="M3 12A9 3 0 0 0 21 12"/></symbol>
|
||||
<symbol id="download" viewBox="0 0 24 24"><path d="M12 15V3"/><path d="M21 15v4a2 2 0 0 1-2 2H5a2 2 0 0 1-2-2v-4"/><path d="m7 10 5 5 5-5"/></symbol>
|
||||
<symbol id="eye" viewBox="0 0 24 24"><path d="M2.062 12.348a1 1 0 0 1 0-.696 10.75 10.75 0 0 1 19.876 0 1 1 0 0 1 0 .696 10.75 10.75 0 0 1-19.876 0"/><circle cx="12" cy="12" r="3"/></symbol>
|
||||
<symbol id="file-plus" viewBox="0 0 24 24"><path d="M6 22a2 2 0 0 1-2-2V4a2 2 0 0 1 2-2h8a2.4 2.4 0 0 1 1.704.706l3.588 3.588A2.4 2.4 0 0 1 20 8v12a2 2 0 0 1-2 2z"/><path d="M14 2v5a1 1 0 0 0 1 1h5"/><path d="M9 15h6"/><path d="M12 18v-6"/></symbol>
|
||||
<symbol id="file-x" viewBox="0 0 24 24"><path d="M6 22a2 2 0 0 1-2-2V4a2 2 0 0 1 2-2h8a2.4 2.4 0 0 1 1.704.706l3.588 3.588A2.4 2.4 0 0 1 20 8v12a2 2 0 0 1-2 2z"/><path d="M14 2v5a1 1 0 0 0 1 1h5"/><path d="m14.5 12.5-5 5"/><path d="m9.5 12.5 5 5"/></symbol>
|
||||
<symbol id="fingerprint" viewBox="0 0 24 24"><path d="M12 10a2 2 0 0 0-2 2c0 1.02-.1 2.51-.26 4"/><path d="M14 13.12c0 2.38 0 6.38-1 8.88"/><path d="M17.29 21.02c.12-.6.43-2.3.5-3.02"/><path d="M2 12a10 10 0 0 1 18-6"/><path d="M2 16h.01"/><path d="M21.8 16c.2-2 .131-5.354 0-6"/><path d="M5 19.5C5.5 18 6 15 6 12a6 6 0 0 1 .34-2"/><path d="M8.65 22c.21-.66.45-1.32.57-2"/><path d="M9 6.8a6 6 0 0 1 9 5.2v2"/></symbol>
|
||||
<symbol id="git-merge" viewBox="0 0 24 24"><circle cx="18" cy="18" r="3"/><circle cx="6" cy="6" r="3"/><path d="M6 21V9a9 9 0 0 0 9 9"/></symbol>
|
||||
<symbol id="github" viewBox="0 0 24 24"><path d="M15 22v-4a4.8 4.8 0 0 0-1-3.5c3 0 6-2 6-5.5.08-1.25-.27-2.48-1-3.5.28-1.15.28-2.35 0-3.5 0 0-1 0-3 1.5-2.64-.5-5.36-.5-8 0C6 2 5 2 5 2c-.3 1.15-.3 2.35 0 3.5A5.403 5.403 0 0 0 4 9c0 3.5 3 5.5 6 5.5-.39.49-.68 1.05-.85 1.65-.17.6-.22 1.23-.15 1.85v4"/><path d="M9 18c-4.51 2-5-2-7-2"/></symbol>
|
||||
<symbol id="hard-drive" viewBox="0 0 24 24"><line x1="22" x2="2" y1="12" y2="12"/><path d="M5.45 5.11 2 12v6a2 2 0 0 0 2 2h16a2 2 0 0 0 2-2v-6l-3.45-6.89A2 2 0 0 0 16.76 4H7.24a2 2 0 0 0-1.79 1.11z"/><line x1="6" x2="6.01" y1="16" y2="16"/><line x1="10" x2="10.01" y1="16" y2="16"/></symbol>
|
||||
<symbol id="info" viewBox="0 0 24 24"><circle cx="12" cy="12" r="10"/><path d="M12 16v-4"/><path d="M12 8h.01"/></symbol>
|
||||
@@ -46,5 +49,6 @@
|
||||
<symbol id="user" viewBox="0 0 24 24"><path d="M19 21v-2a4 4 0 0 0-4-4H9a4 4 0 0 0-4 4v2"/><circle cx="12" cy="7" r="4"/></symbol>
|
||||
<symbol id="user-plus" viewBox="0 0 24 24"><path d="M16 21v-2a4 4 0 0 0-4-4H6a4 4 0 0 0-4 4v2"/><circle cx="9" cy="7" r="4"/><line x1="19" x2="19" y1="8" y2="14"/><line x1="22" x2="16" y1="11" y2="11"/></symbol>
|
||||
<symbol id="x-circle" viewBox="0 0 24 24"><circle cx="12" cy="12" r="10"/><path d="m15 9-6 6"/><path d="m9 9 6 6"/></symbol>
|
||||
<symbol id="zap" viewBox="0 0 24 24"><path d="M4 14a1 1 0 0 1-.78-1.63l9.9-10.2a.5.5 0 0 1 .86.46l-1.92 6.02A1 1 0 0 0 13 10h7a1 1 0 0 1 .78 1.63l-9.9 10.2a.5.5 0 0 1-.86-.46l1.92-6.02A1 1 0 0 0 11 14z"/></symbol>
|
||||
<symbol id="helm" viewBox="0 0 24 24"><path d="M12.337 0c-.475 0-.861 1.016-.861 2.269 0 .527.069 1.011.183 1.396a8.514 8.514 0 0 0-3.961 1.22 5.229 5.229 0 0 0-.595-1.093c-.606-.866-1.34-1.436-1.79-1.43a.381.381 0 0 0-.217.066c-.39.273-.123 1.326.596 2.353.267.381.559.705.84.948a8.683 8.683 0 0 0-1.528 1.716h1.734a7.179 7.179 0 0 1 5.381-2.421 7.18 7.18 0 0 1 5.382 2.42h1.733a8.687 8.687 0 0 0-1.32-1.53c.35-.249.735-.643 1.078-1.133.719-1.027.986-2.08.596-2.353a.382.382 0 0 0-.217-.065c-.45-.007-1.184.563-1.79 1.43a4.897 4.897 0 0 0-.676 1.325 8.52 8.52 0 0 0-3.899-1.42c.12-.39.193-.887.193-1.429 0-1.253-.386-2.269-.862-2.269zM1.624 9.443v5.162h1.358v-1.968h1.64v1.968h1.357V9.443H4.62v1.838H2.98V9.443zm5.912 0v5.162h3.21v-1.108H8.893v-.95h1.64v-1.142h-1.64v-.84h1.853V9.443zm4.698 0v5.162h3.218v-1.362h-1.86v-3.8zm4.706 0v5.162h1.364v-2.643l1.357 1.225 1.35-1.232v2.65h1.365V9.443h-.614l-2.1 1.914-2.109-1.914zm-11.82 7.28a8.688 8.688 0 0 0 1.412 1.548 5.206 5.206 0 0 0-.841.948c-.719 1.027-.985 2.08-.596 2.353.39.273 1.289-.338 2.007-1.364a5.23 5.23 0 0 0 .595-1.092 8.514 8.514 0 0 0 3.961 1.219 5.01 5.01 0 0 0-.183 1.396c0 1.253.386 2.269.861 2.269.476 0 .862-1.016.862-2.269 0-.542-.072-1.04-.193-1.43a8.52 8.52 0 0 0 3.9-1.42c.121.4.352.865.675 1.327.719 1.026 1.617 1.637 2.007 1.364.39-.273.123-1.326-.596-2.353-.343-.49-.727-.885-1.077-1.135a8.69 8.69 0 0 0 1.202-1.36h-1.771a7.174 7.174 0 0 1-5.227 2.252 7.174 7.174 0 0 1-5.226-2.252z" fill="currentColor" stroke="none"/></symbol>
|
||||
</svg>
|
||||
|
Before Width: | Height: | Size: 11 KiB After Width: | Height: | Size: 12 KiB |
@@ -13,6 +13,9 @@
|
||||
<a href="/admin#relays" class="btn btn-sm btn-ghost admin-tab-mobile" data-tab="relays">
|
||||
{{ icon "radio-tower" "size-4" }} Relays
|
||||
</a>
|
||||
<a href="/admin#storage" class="btn btn-sm btn-ghost admin-tab-mobile" data-tab="storage">
|
||||
{{ icon "hard-drive" "size-4" }} Storage
|
||||
</a>
|
||||
</div>
|
||||
{{end}}
|
||||
|
||||
@@ -24,6 +27,7 @@
|
||||
<li data-tab="crew"><a href="/admin#crew">{{ icon "anchor" "size-5" }} Crew</a></li>
|
||||
<li data-tab="settings"><a href="/admin#settings">{{ icon "settings" "size-5" }} Settings</a></li>
|
||||
<li data-tab="relays"><a href="/admin#relays">{{ icon "radio-tower" "size-5" }} Relays</a></li>
|
||||
<li data-tab="storage"><a href="/admin#storage">{{ icon "hard-drive" "size-5" }} Storage</a></li>
|
||||
</ul>
|
||||
</aside>
|
||||
{{end}}
|
||||
|
||||
@@ -30,9 +30,16 @@
|
||||
hx-swap="innerHTML">
|
||||
</div>
|
||||
|
||||
<!-- Storage / GC (loads on first activation) -->
|
||||
<div id="tab-storage" class="admin-panel hidden"
|
||||
hx-get="/admin/api/tab/storage"
|
||||
hx-trigger="tab:storage from:body once"
|
||||
hx-swap="innerHTML">
|
||||
</div>
|
||||
|
||||
<script>
|
||||
(function() {
|
||||
var validTabs = ['dashboard', 'crew', 'settings', 'relays'];
|
||||
var validTabs = ['dashboard', 'crew', 'settings', 'relays', 'storage'];
|
||||
|
||||
function switchAdminTab(tabId) {
|
||||
if (validTabs.indexOf(tabId) === -1) tabId = 'dashboard';
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
{{define "partials/gc_error.html"}}
|
||||
<div class="alert alert-error">
|
||||
{{ icon "alert-triangle" "size-5" }}
|
||||
<span>{{.Error}}</span>
|
||||
</div>
|
||||
{{end}}
|
||||
@@ -0,0 +1,161 @@
|
||||
{{define "partials/gc_preview.html"}}
|
||||
<div class="space-y-6">
|
||||
<!-- Summary cards -->
|
||||
<div class="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-4 gap-4">
|
||||
<div class="stats shadow bg-base-100">
|
||||
<div class="stat">
|
||||
<div class="stat-title">Orphaned Records</div>
|
||||
<div class="stat-value {{if .Preview.OrphanedRecords}}text-error{{end}}">{{len .Preview.OrphanedRecords}}</div>
|
||||
<div class="stat-desc">Layer records with no manifest</div>
|
||||
</div>
|
||||
</div>
|
||||
<div class="stats shadow bg-base-100">
|
||||
<div class="stat">
|
||||
<div class="stat-title">Orphaned Blobs</div>
|
||||
<div class="stat-value {{if .Preview.OrphanedBlobs}}text-error{{end}}">{{len .Preview.OrphanedBlobs}}</div>
|
||||
<div class="stat-desc">S3 blobs with no layer record</div>
|
||||
</div>
|
||||
</div>
|
||||
<div class="stats shadow bg-base-100">
|
||||
<div class="stat">
|
||||
<div class="stat-title">Missing Records</div>
|
||||
<div class="stat-value {{if .Preview.MissingRecords}}text-warning{{end}}">{{len .Preview.MissingRecords}}</div>
|
||||
<div class="stat-desc">Would be reconciled</div>
|
||||
</div>
|
||||
</div>
|
||||
<div class="stats shadow bg-base-100">
|
||||
<div class="stat">
|
||||
<div class="stat-title">Referenced Blobs</div>
|
||||
<div class="stat-value text-success">{{.Preview.ReferencedBlobs}}</div>
|
||||
<div class="stat-desc">{{formatDuration .Preview.Duration}}</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- Meta info -->
|
||||
<p class="text-sm text-base-content/50">
|
||||
Checked {{.Preview.UsersChecked}} users, {{.Preview.ManifestsChecked}} manifests, {{.Preview.TotalLayerRecords}} layer records, {{.Preview.TotalBlobs}} S3 blobs.
|
||||
</p>
|
||||
|
||||
<!-- Orphaned Records table -->
|
||||
{{if .Preview.OrphanedRecords}}
|
||||
<div class="collapse collapse-arrow bg-base-100 shadow-sm">
|
||||
<input type="checkbox" />
|
||||
<div class="collapse-title font-medium">
|
||||
{{ icon "file-x" "size-4 inline" }} Orphaned Layer Records ({{len .Preview.OrphanedRecords}})
|
||||
</div>
|
||||
<div class="collapse-content">
|
||||
<div class="overflow-x-auto">
|
||||
<table class="table table-sm">
|
||||
<thead>
|
||||
<tr>
|
||||
<th>RKey</th>
|
||||
<th>Digest</th>
|
||||
<th>Manifest</th>
|
||||
<th>Size</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{{range .Preview.OrphanedRecords}}
|
||||
<tr>
|
||||
<td><code class="text-xs font-mono">{{.Rkey}}</code></td>
|
||||
<td><code class="text-xs font-mono">{{truncate .Digest 24}}</code></td>
|
||||
<td><code class="text-xs font-mono break-all">{{truncate .ManifestURI 50}}</code></td>
|
||||
<td class="whitespace-nowrap">{{formatBytes .Size}}</td>
|
||||
</tr>
|
||||
{{end}}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
{{end}}
|
||||
|
||||
<!-- Orphaned Blobs table -->
|
||||
{{if .Preview.OrphanedBlobs}}
|
||||
<div class="collapse collapse-arrow bg-base-100 shadow-sm">
|
||||
<input type="checkbox" />
|
||||
<div class="collapse-title font-medium">
|
||||
{{ icon "trash-2" "size-4 inline" }} Orphaned Blobs ({{len .Preview.OrphanedBlobs}})
|
||||
</div>
|
||||
<div class="collapse-content">
|
||||
<div class="overflow-x-auto">
|
||||
<table class="table table-sm">
|
||||
<thead>
|
||||
<tr>
|
||||
<th>Digest</th>
|
||||
<th>Size</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{{range .Preview.OrphanedBlobs}}
|
||||
<tr>
|
||||
<td><code class="text-xs font-mono">{{truncate .Digest 30}}</code></td>
|
||||
<td class="whitespace-nowrap">{{formatBytes .Size}}</td>
|
||||
</tr>
|
||||
{{end}}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
{{end}}
|
||||
|
||||
<!-- Missing Records table (reconcile mode only) -->
|
||||
{{if .Preview.MissingRecords}}
|
||||
<div class="collapse collapse-arrow bg-base-100 shadow-sm">
|
||||
<input type="checkbox" />
|
||||
<div class="collapse-title font-medium">
|
||||
{{ icon "file-plus" "size-4 inline" }} Missing Layer Records ({{len .Preview.MissingRecords}})
|
||||
</div>
|
||||
<div class="collapse-content">
|
||||
<div class="overflow-x-auto">
|
||||
<table class="table table-sm">
|
||||
<thead>
|
||||
<tr>
|
||||
<th>Digest</th>
|
||||
<th>Manifest</th>
|
||||
<th>User</th>
|
||||
<th>Size</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{{range .Preview.MissingRecords}}
|
||||
<tr>
|
||||
<td><code class="text-xs font-mono">{{truncate .Digest 24}}</code></td>
|
||||
<td><code class="text-xs font-mono break-all">{{truncate .ManifestURI 50}}</code></td>
|
||||
<td><code class="text-xs font-mono">{{truncate .UserDID 24}}</code></td>
|
||||
<td class="whitespace-nowrap">{{formatBytes .Size}}</td>
|
||||
</tr>
|
||||
{{end}}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
{{end}}
|
||||
|
||||
<!-- Run GC button (only if there are actions to take) -->
|
||||
{{if or .Preview.OrphanedRecords .Preview.OrphanedBlobs .Preview.MissingRecords}}
|
||||
<div class="flex items-center gap-3 mt-6">
|
||||
<button class="btn btn-error gap-2"
|
||||
hx-post="/admin/api/gc/run"
|
||||
hx-target="#gc-results"
|
||||
hx-swap="innerHTML"
|
||||
hx-confirm="Are you sure you want to run garbage collection?"
|
||||
hx-indicator="#gc-loading">
|
||||
{{ icon "zap" "size-4" }}
|
||||
Run GC
|
||||
</button>
|
||||
</div>
|
||||
{{end}}
|
||||
|
||||
<!-- Nothing to clean -->
|
||||
{{if and (not .Preview.OrphanedRecords) (not .Preview.OrphanedBlobs) (not .Preview.MissingRecords)}}
|
||||
<div class="alert alert-success">
|
||||
{{ icon "check-circle" "size-5" }}
|
||||
<span>Storage is clean. No orphaned data found.</span>
|
||||
</div>
|
||||
{{end}}
|
||||
</div>
|
||||
{{end}}
|
||||
@@ -0,0 +1,39 @@
|
||||
{{define "partials/gc_result.html"}}
|
||||
<div class="space-y-6">
|
||||
<div class="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-3 gap-4">
|
||||
<div class="stats shadow bg-base-100">
|
||||
<div class="stat">
|
||||
<div class="stat-title">Records Deleted</div>
|
||||
<div class="stat-value">{{.Result.RecordsDeleted}}</div>
|
||||
<div class="stat-desc">of {{.Result.OrphanedRecords}} orphaned</div>
|
||||
</div>
|
||||
</div>
|
||||
<div class="stats shadow bg-base-100">
|
||||
<div class="stat">
|
||||
<div class="stat-title">Blobs Deleted</div>
|
||||
<div class="stat-value">{{.Result.BlobsDeleted}}</div>
|
||||
<div class="stat-desc">of {{.Result.OrphanedBlobs}} orphaned</div>
|
||||
</div>
|
||||
</div>
|
||||
<div class="stats shadow bg-base-100">
|
||||
<div class="stat">
|
||||
<div class="stat-title">Space Reclaimed</div>
|
||||
<div class="stat-value text-lg">{{formatBytes .Result.BytesReclaimed}}</div>
|
||||
<div class="stat-desc">in {{formatDuration .Result.Duration}}</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{{if gt .Result.RecordsReconciled 0}}
|
||||
<div class="alert alert-info">
|
||||
{{ icon "git-merge" "size-5" }}
|
||||
<span>Reconciled {{.Result.RecordsReconciled}} missing layer records.</span>
|
||||
</div>
|
||||
{{end}}
|
||||
|
||||
<p class="text-sm text-base-content/50">
|
||||
Checked {{.Result.UsersChecked}} users, {{.Result.ManifestsChecked}} manifests.
|
||||
{{.Result.ReferencedBlobs}} blobs are referenced and retained.
|
||||
</p>
|
||||
</div>
|
||||
{{end}}
|
||||
@@ -0,0 +1,60 @@
|
||||
{{define "partials/tab_storage.html"}}
|
||||
<div class="flex justify-between items-center min-h-12 mb-6">
|
||||
<h1 class="text-2xl font-bold">Storage</h1>
|
||||
</div>
|
||||
|
||||
<!-- GC Configuration -->
|
||||
<div class="card bg-base-100 shadow-sm mb-6">
|
||||
<div class="card-body">
|
||||
<h2 class="card-title text-lg">Garbage Collection</h2>
|
||||
<div class="flex flex-wrap gap-2 mt-2">
|
||||
{{if .Enabled}}
|
||||
<span class="badge badge-success gap-1">{{ icon "check-circle" "size-3" }} Enabled</span>
|
||||
{{else}}
|
||||
<span class="badge badge-error gap-1">{{ icon "x-circle" "size-3" }} Disabled</span>
|
||||
{{end}}
|
||||
|
||||
{{if .Running}}
|
||||
<span class="badge badge-warning gap-1">
|
||||
<span class="loading loading-spinner loading-xs"></span>
|
||||
Running
|
||||
</span>
|
||||
{{end}}
|
||||
</div>
|
||||
|
||||
{{if .HasPreview}}
|
||||
<p class="text-sm text-base-content/50 mt-2">Last scan: {{.LastPreviewAge}}</p>
|
||||
{{end}}
|
||||
{{if .HasResult}}
|
||||
<p class="text-sm text-base-content/50">Last run: {{.LastResultAge}}</p>
|
||||
{{end}}
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- Scan button -->
|
||||
<div class="flex gap-3 mb-6">
|
||||
<button class="btn btn-primary gap-2"
|
||||
hx-post="/admin/api/gc/preview"
|
||||
hx-target="#gc-results"
|
||||
hx-swap="innerHTML"
|
||||
hx-indicator="#gc-loading"
|
||||
{{if .Running}}disabled{{end}}>
|
||||
{{ icon "search" "size-4" }}
|
||||
Scan for Orphans
|
||||
</button>
|
||||
</div>
|
||||
|
||||
<div id="gc-loading" class="htmx-indicator mb-4">
|
||||
<div class="flex items-center gap-3 p-4 bg-base-100 rounded-lg shadow-sm">
|
||||
<span class="loading loading-spinner loading-md text-primary"></span>
|
||||
<div>
|
||||
<p class="font-medium">Scanning storage...</p>
|
||||
<p class="text-sm text-base-content/50">This may take a few minutes for large holds.</p>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div id="gc-results">
|
||||
<!-- Preview or run results will be swapped in here -->
|
||||
</div>
|
||||
{{end}}
|
||||
@@ -16,6 +16,7 @@ import (
|
||||
"github.com/spf13/viper"
|
||||
|
||||
"atcr.io/pkg/config"
|
||||
"atcr.io/pkg/hold/gc"
|
||||
"atcr.io/pkg/hold/quota"
|
||||
)
|
||||
|
||||
@@ -29,6 +30,7 @@ type Config struct {
|
||||
Registration RegistrationConfig `yaml:"registration" comment:"Auto-registration and bootstrap settings."`
|
||||
Database DatabaseConfig `yaml:"database" comment:"Embedded PDS database settings."`
|
||||
Admin AdminConfig `yaml:"admin" comment:"Admin panel settings."`
|
||||
GC gc.Config `yaml:"gc" comment:"Garbage collection settings."`
|
||||
Quota quota.Config `yaml:"quota" comment:"Storage quota tiers. Empty disables quota enforcement."`
|
||||
Scanner ScannerConfig `yaml:"scanner" comment:"Vulnerability scanner settings. Empty disables scanning."`
|
||||
configPath string `yaml:"-"` // internal: path to YAML file for subsystem config loading
|
||||
@@ -183,6 +185,8 @@ func setHoldDefaults(v *viper.Viper) {
|
||||
v.SetDefault("storage.bucket", "")
|
||||
v.SetDefault("storage.endpoint", "")
|
||||
|
||||
// GC defaults
|
||||
v.SetDefault("gc.enabled", false)
|
||||
// Scanner defaults
|
||||
v.SetDefault("scanner.secret", "")
|
||||
|
||||
@@ -237,6 +241,9 @@ func LoadConfig(yamlPath string) (*Config, error) {
|
||||
_ = v.BindEnv("storage.bucket", "S3_BUCKET")
|
||||
_ = v.BindEnv("storage.endpoint", "S3_ENDPOINT")
|
||||
|
||||
// Bind legacy GC env vars (backward compat)
|
||||
_ = v.BindEnv("gc.enabled", "GC_ENABLED")
|
||||
|
||||
// Unmarshal into config struct
|
||||
cfg := &Config{}
|
||||
if err := v.Unmarshal(cfg, config.UnmarshalOption()); err != nil {
|
||||
|
||||
+4
-19
@@ -3,10 +3,7 @@
|
||||
// layer records in the hold's embedded PDS.
|
||||
package gc
|
||||
|
||||
import (
|
||||
"os"
|
||||
"time"
|
||||
)
|
||||
import "time"
|
||||
|
||||
// Hardcoded defaults - keep configuration simple
|
||||
const (
|
||||
@@ -18,20 +15,8 @@ const (
|
||||
gcGracePeriod = 7 * 24 * time.Hour
|
||||
)
|
||||
|
||||
// Config holds GC configuration, loaded from environment variables
|
||||
// Config holds GC configuration
|
||||
type Config struct {
|
||||
// Enabled controls whether GC is active (GC_ENABLED, default: true)
|
||||
Enabled bool
|
||||
|
||||
// DryRun logs what would be deleted without actually deleting (GC_DRY_RUN, default: true)
|
||||
// Remove after initial validation
|
||||
DryRun bool
|
||||
}
|
||||
|
||||
// LoadConfigFromEnv loads GC configuration from environment variables
|
||||
func LoadConfigFromEnv() Config {
|
||||
return Config{
|
||||
Enabled: os.Getenv("GC_ENABLED") != "false", // Default true
|
||||
DryRun: os.Getenv("GC_DRY_RUN") != "false", // Default true
|
||||
}
|
||||
// Enabled controls whether the nightly GC background process runs.
|
||||
Enabled bool `yaml:"enabled" comment:"Enable nightly garbage collection of orphaned blobs and records."`
|
||||
}
|
||||
|
||||
+549
-150
@@ -5,7 +5,6 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"regexp"
|
||||
@@ -19,6 +18,49 @@ import (
|
||||
storagedriver "github.com/distribution/distribution/v3/registry/storage/driver"
|
||||
)
|
||||
|
||||
// maxPreviewItems caps per-category detail slices to prevent memory/HTML bloat
|
||||
const maxPreviewItems = 10000
|
||||
|
||||
// OrphanedRecordDetail holds info about a single orphaned layer record
|
||||
type OrphanedRecordDetail struct {
|
||||
Rkey string `json:"rkey"`
|
||||
Digest string `json:"digest"`
|
||||
ManifestURI string `json:"manifestUri"`
|
||||
UserDID string `json:"userDid"`
|
||||
MediaType string `json:"mediaType"`
|
||||
Size int64 `json:"size"`
|
||||
}
|
||||
|
||||
// OrphanedBlobDetail holds info about a single orphaned blob in S3
|
||||
type OrphanedBlobDetail struct {
|
||||
Digest string `json:"digest"`
|
||||
Size int64 `json:"size"`
|
||||
}
|
||||
|
||||
// MissingRecordDetail holds info about a layer record that should exist but doesn't
|
||||
type MissingRecordDetail struct {
|
||||
Digest string `json:"digest"`
|
||||
ManifestURI string `json:"manifestUri"`
|
||||
UserDID string `json:"userDid"`
|
||||
MediaType string `json:"mediaType"`
|
||||
Size int64 `json:"size"`
|
||||
}
|
||||
|
||||
// GCPreview contains detailed analysis results (no mutations performed)
|
||||
type GCPreview struct {
|
||||
OrphanedRecords []OrphanedRecordDetail `json:"orphanedRecords"`
|
||||
OrphanedBlobs []OrphanedBlobDetail `json:"orphanedBlobs"`
|
||||
MissingRecords []MissingRecordDetail `json:"missingRecords"`
|
||||
|
||||
TotalLayerRecords int `json:"totalLayerRecords"`
|
||||
TotalBlobs int `json:"totalBlobs"`
|
||||
ReferencedBlobs int `json:"referencedBlobs"`
|
||||
UsersChecked int `json:"usersChecked"`
|
||||
ManifestsChecked int `json:"manifestsChecked"`
|
||||
Reconcile bool `json:"reconcile"`
|
||||
Duration time.Duration `json:"duration"`
|
||||
}
|
||||
|
||||
// GarbageCollector handles cleanup of orphaned blobs from storage
|
||||
type GarbageCollector struct {
|
||||
pds *pds.HoldPDS
|
||||
@@ -30,17 +72,48 @@ type GarbageCollector struct {
|
||||
stopCh chan struct{}
|
||||
// wg tracks the background goroutine
|
||||
wg sync.WaitGroup
|
||||
|
||||
// mu protects running state and last results
|
||||
mu sync.Mutex
|
||||
running bool
|
||||
|
||||
// Last results (for admin panel display)
|
||||
lastPreview *GCPreview
|
||||
lastPreviewAt time.Time
|
||||
lastResult *GCResult
|
||||
lastResultAt time.Time
|
||||
}
|
||||
|
||||
// GCResult contains statistics from a GC run
|
||||
type GCResult struct {
|
||||
BlobsDeleted int64 `json:"blobs_deleted"`
|
||||
BytesReclaimed int64 `json:"bytes_reclaimed"`
|
||||
RecordsDeleted int64 `json:"records_deleted"`
|
||||
OrphanedRecords int64 `json:"orphaned_records"`
|
||||
OrphanedBlobs int64 `json:"orphaned_blobs"`
|
||||
ReferencedBlobs int64 `json:"referenced_blobs"`
|
||||
Duration time.Duration `json:"duration"`
|
||||
BlobsDeleted int64 `json:"blobs_deleted"`
|
||||
BytesReclaimed int64 `json:"bytes_reclaimed"`
|
||||
RecordsDeleted int64 `json:"records_deleted"`
|
||||
OrphanedRecords int64 `json:"orphaned_records"`
|
||||
OrphanedBlobs int64 `json:"orphaned_blobs"`
|
||||
ReferencedBlobs int64 `json:"referenced_blobs"`
|
||||
RecordsReconciled int64 `json:"records_reconciled"`
|
||||
ManifestsChecked int64 `json:"manifests_checked"`
|
||||
UsersChecked int64 `json:"users_checked"`
|
||||
Duration time.Duration `json:"duration"`
|
||||
}
|
||||
|
||||
// manifestInfo holds a parsed manifest fetched from a user's PDS
|
||||
type manifestInfo struct {
|
||||
URI string // AT-URI of the manifest
|
||||
UserDID string // DID of the user who owns it
|
||||
Record *atproto.ManifestRecord // Parsed manifest data
|
||||
}
|
||||
|
||||
// analysisResult holds intermediate data from record analysis, shared between Run and Preview
|
||||
type analysisResult struct {
|
||||
referenced map[string]bool
|
||||
orphanedRkeys []string // rkeys for deletion in Run
|
||||
orphanedDetails []OrphanedRecordDetail // details for display in Preview
|
||||
missingDetails []MissingRecordDetail // details for creation in Run / display in Preview
|
||||
usersChecked int64
|
||||
manifestsChecked int64
|
||||
totalRecords int
|
||||
}
|
||||
|
||||
// NewGarbageCollector creates a new GC instance
|
||||
@@ -54,24 +127,31 @@ func NewGarbageCollector(holdPDS *pds.HoldPDS, driver storagedriver.StorageDrive
|
||||
}
|
||||
}
|
||||
|
||||
// Start begins the GC background process
|
||||
// It runs GC immediately on startup, then periodically according to gcInterval
|
||||
// tryStart attempts to mark GC as running. Returns false if already running.
|
||||
func (gc *GarbageCollector) tryStart() bool {
|
||||
gc.mu.Lock()
|
||||
defer gc.mu.Unlock()
|
||||
if gc.running {
|
||||
return false
|
||||
}
|
||||
gc.running = true
|
||||
return true
|
||||
}
|
||||
|
||||
// finish marks GC as no longer running
|
||||
func (gc *GarbageCollector) finish() {
|
||||
gc.mu.Lock()
|
||||
gc.running = false
|
||||
gc.mu.Unlock()
|
||||
}
|
||||
|
||||
// Start begins the GC background process with a nightly ticker
|
||||
func (gc *GarbageCollector) Start(ctx context.Context) {
|
||||
if !gc.cfg.Enabled {
|
||||
gc.logger.Info("GC disabled")
|
||||
return
|
||||
}
|
||||
|
||||
// Run on startup
|
||||
gc.logger.Info("Running GC on startup", "dryRun", gc.cfg.DryRun)
|
||||
result, err := gc.Run(ctx)
|
||||
if err != nil {
|
||||
gc.logger.Error("Startup GC failed", "error", err)
|
||||
} else {
|
||||
gc.logResult(result)
|
||||
}
|
||||
|
||||
// Start background ticker for nightly runs
|
||||
gc.wg.Add(1)
|
||||
go func() {
|
||||
defer gc.wg.Done()
|
||||
@@ -88,10 +168,10 @@ func (gc *GarbageCollector) Start(ctx context.Context) {
|
||||
gc.logger.Info("GC context cancelled")
|
||||
return
|
||||
case <-ticker.C:
|
||||
gc.logger.Info("Running nightly GC", "dryRun", gc.cfg.DryRun)
|
||||
gc.logger.Info("Running nightly GC")
|
||||
result, err := gc.Run(ctx)
|
||||
if err != nil {
|
||||
gc.logger.Error("Nightly GC failed", "error", err)
|
||||
gc.logger.Warn("Nightly GC skipped or failed", "error", err)
|
||||
} else {
|
||||
gc.logResult(result)
|
||||
}
|
||||
@@ -110,87 +190,226 @@ func (gc *GarbageCollector) Stop() {
|
||||
|
||||
// Run executes a single GC cycle
|
||||
func (gc *GarbageCollector) Run(ctx context.Context) (*GCResult, error) {
|
||||
if !gc.tryStart() {
|
||||
return nil, fmt.Errorf("GC operation already in progress")
|
||||
}
|
||||
defer gc.finish()
|
||||
|
||||
start := time.Now()
|
||||
result := &GCResult{}
|
||||
|
||||
gc.logger.Info("Starting GC run", "dryRun", gc.cfg.DryRun)
|
||||
gc.logger.Info("Starting GC run")
|
||||
|
||||
// Phase 1: Build referenced set from layer records
|
||||
referenced, orphanedRecords, err := gc.buildReferencedSet(ctx, result)
|
||||
// Phase 1: Analyze records (build referenced set, find orphans, identify missing)
|
||||
analysis, err := gc.analyzeRecords(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("phase 1 (build referenced set) failed: %w", err)
|
||||
return nil, fmt.Errorf("phase 1 (analyze records) failed: %w", err)
|
||||
}
|
||||
|
||||
result.OrphanedRecords = int64(len(analysis.orphanedRkeys))
|
||||
result.UsersChecked = analysis.usersChecked
|
||||
result.ManifestsChecked = analysis.manifestsChecked
|
||||
|
||||
gc.logger.Info("Phase 1 complete",
|
||||
"referenced", len(referenced),
|
||||
"orphanedRecords", len(orphanedRecords))
|
||||
"referenced", len(analysis.referenced),
|
||||
"orphanedRecords", len(analysis.orphanedRkeys),
|
||||
"missingRecords", len(analysis.missingDetails))
|
||||
|
||||
// Reconcile: create missing layer records
|
||||
if len(analysis.missingDetails) > 0 {
|
||||
gc.reconcileMissingRecords(ctx, analysis.missingDetails, result)
|
||||
}
|
||||
|
||||
// Phase 2: Delete orphaned layer records
|
||||
if err := gc.deleteOrphanedRecords(ctx, orphanedRecords, result); err != nil {
|
||||
if err := gc.deleteOrphanedRecords(ctx, analysis.orphanedRkeys, result); err != nil {
|
||||
gc.logger.Error("Phase 2 (delete orphaned records) failed", "error", err)
|
||||
// Continue to phase 3 - we can still clean up blobs
|
||||
}
|
||||
|
||||
// Phase 3: Walk storage and delete unreferenced blobs
|
||||
if err := gc.deleteOrphanedBlobs(ctx, referenced, result); err != nil {
|
||||
if err := gc.deleteOrphanedBlobs(ctx, analysis.referenced, result); err != nil {
|
||||
return nil, fmt.Errorf("phase 3 (delete orphaned blobs) failed: %w", err)
|
||||
}
|
||||
|
||||
result.Duration = time.Since(start)
|
||||
result.ReferencedBlobs = int64(len(referenced))
|
||||
result.ReferencedBlobs = int64(len(analysis.referenced))
|
||||
|
||||
// Store last result for admin panel
|
||||
gc.mu.Lock()
|
||||
gc.lastResult = result
|
||||
gc.lastResultAt = time.Now()
|
||||
gc.mu.Unlock()
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// buildReferencedSet iterates layer records and builds a set of referenced digests
|
||||
// Returns: referenced digest set, list of orphaned record rkeys, error
|
||||
func (gc *GarbageCollector) buildReferencedSet(ctx context.Context, result *GCResult) (map[string]bool, []string, error) {
|
||||
referenced := make(map[string]bool)
|
||||
var orphanedRecords []string
|
||||
// Preview runs read-only analysis and returns detailed results for the admin panel.
|
||||
// It never creates, deletes, or modifies any data.
|
||||
func (gc *GarbageCollector) Preview(ctx context.Context) (*GCPreview, error) {
|
||||
if !gc.tryStart() {
|
||||
return nil, fmt.Errorf("GC operation already in progress")
|
||||
}
|
||||
defer gc.finish()
|
||||
|
||||
start := time.Now()
|
||||
|
||||
gc.logger.Info("Starting GC preview")
|
||||
|
||||
// Phase 1: Analyze records
|
||||
analysis, err := gc.analyzeRecords(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("analyze records: %w", err)
|
||||
}
|
||||
|
||||
// Phase 2: Walk S3 to find orphaned blobs (read-only)
|
||||
orphanedBlobs, totalBlobs, err := gc.scanOrphanedBlobDetails(ctx, analysis.referenced)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("scan orphaned blobs: %w", err)
|
||||
}
|
||||
|
||||
preview := &GCPreview{
|
||||
OrphanedRecords: analysis.orphanedDetails,
|
||||
OrphanedBlobs: orphanedBlobs,
|
||||
MissingRecords: analysis.missingDetails,
|
||||
TotalLayerRecords: analysis.totalRecords,
|
||||
TotalBlobs: totalBlobs,
|
||||
ReferencedBlobs: len(analysis.referenced),
|
||||
UsersChecked: int(analysis.usersChecked),
|
||||
ManifestsChecked: int(analysis.manifestsChecked),
|
||||
Reconcile: true,
|
||||
Duration: time.Since(start),
|
||||
}
|
||||
|
||||
gc.logger.Info("GC preview complete",
|
||||
"orphanedRecords", len(preview.OrphanedRecords),
|
||||
"orphanedBlobs", len(preview.OrphanedBlobs),
|
||||
"missingRecords", len(preview.MissingRecords),
|
||||
"referencedBlobs", preview.ReferencedBlobs,
|
||||
"duration", preview.Duration)
|
||||
|
||||
// Store for admin panel
|
||||
gc.mu.Lock()
|
||||
gc.lastPreview = preview
|
||||
gc.lastPreviewAt = time.Now()
|
||||
gc.mu.Unlock()
|
||||
|
||||
return preview, nil
|
||||
}
|
||||
|
||||
// analyzeRecords performs Phase 1 analysis: builds referenced set, finds orphaned records,
|
||||
// and identifies missing layer records. Pure analysis — no mutations.
|
||||
// Discovers users, fetches manifests, scans records, identifies missing records.
|
||||
func (gc *GarbageCollector) analyzeRecords(ctx context.Context) (*analysisResult, error) {
|
||||
result := &analysisResult{
|
||||
referenced: make(map[string]bool),
|
||||
}
|
||||
|
||||
recordsIndex := gc.pds.RecordsIndex()
|
||||
if recordsIndex == nil {
|
||||
return nil, nil, fmt.Errorf("records index not available")
|
||||
return nil, fmt.Errorf("records index not available")
|
||||
}
|
||||
|
||||
// Step 1: Discover all user DIDs
|
||||
userDIDs, err := gc.discoverUserDIDs(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("discover user DIDs: %w", err)
|
||||
}
|
||||
gc.logger.Info("Discovered user DIDs", "count", len(userDIDs))
|
||||
|
||||
// Step 2: Fetch manifests from each user's PDS
|
||||
knownManifests := make(map[string]*manifestInfo)
|
||||
fetchedUsers := make(map[string]bool)
|
||||
|
||||
for _, did := range userDIDs {
|
||||
result.usersChecked++
|
||||
|
||||
manifests, err := gc.fetchUserManifests(ctx, did)
|
||||
if err != nil {
|
||||
gc.logger.Warn("Failed to fetch manifests for user, treating their records as referenced",
|
||||
"did", did, "error", err)
|
||||
continue
|
||||
}
|
||||
|
||||
fetchedUsers[did] = true
|
||||
for _, m := range manifests {
|
||||
result.manifestsChecked++
|
||||
knownManifests[m.URI] = m
|
||||
|
||||
// Add all layer digests to referenced set
|
||||
for _, layer := range m.Record.Layers {
|
||||
result.referenced[layer.Digest] = true
|
||||
}
|
||||
// Add config digest to referenced set (config blobs are in S3 but
|
||||
// don't get layer records — just protect them from deletion)
|
||||
if m.Record.Config != nil && m.Record.Config.Digest != "" {
|
||||
result.referenced[m.Record.Config.Digest] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
gc.logger.Info("Fetched manifests from user PDS instances",
|
||||
"knownManifests", len(knownManifests),
|
||||
"fetchedUsers", len(fetchedUsers))
|
||||
|
||||
// Step 3: Scan existing layer records to find orphans and build coveredPairs
|
||||
coveredPairs := make(map[string]bool)
|
||||
|
||||
cursor := ""
|
||||
batchSize := 1000
|
||||
totalRecords := 0
|
||||
|
||||
for {
|
||||
records, nextCursor, err := recordsIndex.ListRecords(atproto.LayerCollection, batchSize, cursor, true)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to list layer records: %w", err)
|
||||
return nil, fmt.Errorf("failed to list layer records: %w", err)
|
||||
}
|
||||
|
||||
for _, rec := range records {
|
||||
totalRecords++
|
||||
result.totalRecords++
|
||||
|
||||
// Decode the layer record
|
||||
layer, err := gc.decodeLayerRecord(ctx, rec)
|
||||
if err != nil {
|
||||
gc.logger.Warn("Failed to decode layer record", "rkey", rec.Rkey, "error", err)
|
||||
continue
|
||||
}
|
||||
|
||||
// Grace period: skip records from last 7 days
|
||||
// Track this (manifest, digest) pair as covered
|
||||
pairKey := layer.Manifest + "|" + layer.Digest
|
||||
coveredPairs[pairKey] = true
|
||||
|
||||
// Grace period: skip recent records
|
||||
recordTime := tidToTime(rec.Rkey)
|
||||
if time.Since(recordTime) < gcGracePeriod {
|
||||
// Recent record - assume referenced, skip checking
|
||||
referenced[layer.Digest] = true
|
||||
result.referenced[layer.Digest] = true
|
||||
continue
|
||||
}
|
||||
|
||||
// Cross-check: does the manifest still exist?
|
||||
if gc.manifestExists(ctx, layer.Manifest) {
|
||||
referenced[layer.Digest] = true
|
||||
// Check if this layer's manifest is known
|
||||
if _, known := knownManifests[layer.Manifest]; known {
|
||||
result.referenced[layer.Digest] = true
|
||||
} else {
|
||||
result.OrphanedRecords++
|
||||
orphanedRecords = append(orphanedRecords, rec.Rkey)
|
||||
gc.logger.Debug("Found orphaned layer record",
|
||||
"rkey", rec.Rkey,
|
||||
"digest", layer.Digest,
|
||||
"manifest", layer.Manifest)
|
||||
// Manifest not in our fetched set — check if the user's PDS was unreachable
|
||||
parts := parseATURI(layer.Manifest)
|
||||
if parts != nil && !fetchedUsers[parts.DID] {
|
||||
// User's PDS was unreachable — safe default: assume referenced
|
||||
result.referenced[layer.Digest] = true
|
||||
} else {
|
||||
// User's PDS was reachable but manifest not found — orphaned
|
||||
result.orphanedRkeys = append(result.orphanedRkeys, rec.Rkey)
|
||||
if len(result.orphanedDetails) < maxPreviewItems {
|
||||
result.orphanedDetails = append(result.orphanedDetails, OrphanedRecordDetail{
|
||||
Rkey: rec.Rkey,
|
||||
Digest: layer.Digest,
|
||||
ManifestURI: layer.Manifest,
|
||||
UserDID: layer.UserDID,
|
||||
MediaType: layer.MediaType,
|
||||
Size: layer.Size,
|
||||
})
|
||||
}
|
||||
gc.logger.Debug("Found orphaned layer record",
|
||||
"rkey", rec.Rkey,
|
||||
"digest", layer.Digest,
|
||||
"manifest", layer.Manifest)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -199,35 +418,257 @@ func (gc *GarbageCollector) buildReferencedSet(ctx context.Context, result *GCRe
|
||||
}
|
||||
cursor = nextCursor
|
||||
|
||||
// Progress logging
|
||||
if totalRecords%10000 == 0 {
|
||||
gc.logger.Info("Phase 1 progress", "processed", totalRecords)
|
||||
if result.totalRecords%10000 == 0 {
|
||||
gc.logger.Info("Phase 1 progress", "processed", result.totalRecords)
|
||||
}
|
||||
}
|
||||
|
||||
gc.logger.Info("Scanned layer records", "total", totalRecords)
|
||||
return referenced, orphanedRecords, nil
|
||||
gc.logger.Info("Scanned layer records", "total", result.totalRecords, "coveredPairs", len(coveredPairs))
|
||||
|
||||
// Step 4: Identify missing layer records (uncovered manifest+layer pairs)
|
||||
for _, m := range knownManifests {
|
||||
for _, layer := range m.Record.Layers {
|
||||
pairKey := m.URI + "|" + layer.Digest
|
||||
if coveredPairs[pairKey] {
|
||||
continue
|
||||
}
|
||||
if len(result.missingDetails) < maxPreviewItems {
|
||||
result.missingDetails = append(result.missingDetails, MissingRecordDetail{
|
||||
Digest: layer.Digest,
|
||||
ManifestURI: m.URI,
|
||||
UserDID: m.UserDID,
|
||||
MediaType: layer.MediaType,
|
||||
Size: layer.Size,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// scanOrphanedBlobDetails walks S3 and returns details of unreferenced blobs.
|
||||
// Read-only — no deletions. Returns orphaned blob details and total blob count.
|
||||
func (gc *GarbageCollector) scanOrphanedBlobDetails(ctx context.Context, referenced map[string]bool) ([]OrphanedBlobDetail, int, error) {
|
||||
var orphaned []OrphanedBlobDetail
|
||||
totalBlobs := 0
|
||||
blobsPath := "/docker/registry/v2/blobs"
|
||||
|
||||
err := gc.driver.Walk(ctx, blobsPath, func(fi storagedriver.FileInfo) error {
|
||||
if fi.IsDir() {
|
||||
return nil
|
||||
}
|
||||
if !strings.HasSuffix(fi.Path(), "/data") {
|
||||
return nil
|
||||
}
|
||||
|
||||
digest := extractDigestFromPath(fi.Path())
|
||||
if digest == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
totalBlobs++
|
||||
|
||||
if !referenced[digest] {
|
||||
if len(orphaned) < maxPreviewItems {
|
||||
orphaned = append(orphaned, OrphanedBlobDetail{
|
||||
Digest: digest,
|
||||
Size: fi.Size(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
return nil, 0, fmt.Errorf("walk storage failed: %w", err)
|
||||
}
|
||||
|
||||
return orphaned, totalBlobs, nil
|
||||
}
|
||||
|
||||
// reconcileMissingRecords creates layer records for manifest+layer pairs that are missing.
|
||||
func (gc *GarbageCollector) reconcileMissingRecords(ctx context.Context, missing []MissingRecordDetail, result *GCResult) {
|
||||
for _, m := range missing {
|
||||
record := atproto.NewLayerRecord(
|
||||
m.Digest,
|
||||
m.Size,
|
||||
m.MediaType,
|
||||
m.UserDID,
|
||||
m.ManifestURI,
|
||||
)
|
||||
if _, _, err := gc.pds.CreateLayerRecord(ctx, record); err != nil {
|
||||
gc.logger.Error("Failed to create reconciled layer record",
|
||||
"digest", m.Digest,
|
||||
"manifest", m.ManifestURI,
|
||||
"error", err)
|
||||
continue
|
||||
}
|
||||
result.RecordsReconciled++
|
||||
gc.logger.Info("Created missing layer record",
|
||||
"digest", m.Digest,
|
||||
"manifest", m.ManifestURI,
|
||||
"user", m.UserDID)
|
||||
}
|
||||
|
||||
if result.RecordsReconciled > 0 {
|
||||
gc.logger.Info("Reconciliation complete", "recordsCreated", result.RecordsReconciled)
|
||||
}
|
||||
}
|
||||
|
||||
// discoverUserDIDs returns all DIDs that may have manifests referencing this hold.
|
||||
// Union of: captain owner + crew members + distinct DIDs from layer records.
|
||||
func (gc *GarbageCollector) discoverUserDIDs(ctx context.Context) ([]string, error) {
|
||||
seen := make(map[string]bool)
|
||||
|
||||
// Captain owner DID
|
||||
_, captain, err := gc.pds.GetCaptainRecord(ctx)
|
||||
if err != nil {
|
||||
gc.logger.Warn("Failed to get captain record", "error", err)
|
||||
} else if captain.Owner != "" {
|
||||
seen[captain.Owner] = true
|
||||
}
|
||||
|
||||
// Crew member DIDs
|
||||
crew, err := gc.pds.ListCrewMembers(ctx)
|
||||
if err != nil {
|
||||
gc.logger.Warn("Failed to list crew members", "error", err)
|
||||
} else {
|
||||
for _, m := range crew {
|
||||
if m.Record.Member != "" {
|
||||
seen[m.Record.Member] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Distinct DIDs from existing layer records (catches non-crew pushers on public holds)
|
||||
recordsIndex := gc.pds.RecordsIndex()
|
||||
if recordsIndex != nil {
|
||||
dids, err := recordsIndex.DistinctDIDs(atproto.LayerCollection)
|
||||
if err != nil {
|
||||
gc.logger.Warn("Failed to get distinct DIDs from records", "error", err)
|
||||
} else {
|
||||
for _, did := range dids {
|
||||
seen[did] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
dids := make([]string, 0, len(seen))
|
||||
for did := range seen {
|
||||
dids = append(dids, did)
|
||||
}
|
||||
return dids, nil
|
||||
}
|
||||
|
||||
// fetchUserManifests fetches all io.atcr.manifest records for a user from their PDS,
|
||||
// filtered to manifests that reference this hold.
|
||||
func (gc *GarbageCollector) fetchUserManifests(ctx context.Context, userDID string) ([]*manifestInfo, error) {
|
||||
pdsEndpoint, err := atproto.ResolveDIDToPDS(ctx, userDID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("resolve PDS for %s: %w", userDID, err)
|
||||
}
|
||||
return gc.fetchUserManifestsFromEndpoint(ctx, userDID, pdsEndpoint, gc.pds.DID())
|
||||
}
|
||||
|
||||
// fetchUserManifestsFromEndpoint fetches manifests from a specific PDS endpoint.
|
||||
// Separated from fetchUserManifests for testability (avoids DID resolution).
|
||||
func (gc *GarbageCollector) fetchUserManifestsFromEndpoint(ctx context.Context, userDID, pdsEndpoint, holdDID string) ([]*manifestInfo, error) {
|
||||
var manifests []*manifestInfo
|
||||
cursor := ""
|
||||
client := &http.Client{Timeout: 30 * time.Second}
|
||||
|
||||
for {
|
||||
reqURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.listRecords?repo=%s&collection=%s&limit=100",
|
||||
pdsEndpoint, userDID, atproto.ManifestCollection)
|
||||
if cursor != "" {
|
||||
reqURL += "&cursor=" + cursor
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, "GET", reqURL, nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create request: %w", err)
|
||||
}
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("http request: %w", err)
|
||||
}
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
resp.Body.Close()
|
||||
return nil, fmt.Errorf("listRecords returned status %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
var listResult struct {
|
||||
Records []struct {
|
||||
URI string `json:"uri"`
|
||||
CID string `json:"cid"`
|
||||
Value json.RawMessage `json:"value"`
|
||||
} `json:"records"`
|
||||
Cursor string `json:"cursor,omitempty"`
|
||||
}
|
||||
|
||||
if err := json.NewDecoder(resp.Body).Decode(&listResult); err != nil {
|
||||
resp.Body.Close()
|
||||
return nil, fmt.Errorf("decode response: %w", err)
|
||||
}
|
||||
resp.Body.Close()
|
||||
|
||||
for _, rec := range listResult.Records {
|
||||
var manifest atproto.ManifestRecord
|
||||
if err := json.Unmarshal(rec.Value, &manifest); err != nil {
|
||||
gc.logger.Warn("Failed to parse manifest record",
|
||||
"uri", rec.URI, "error", err)
|
||||
continue
|
||||
}
|
||||
|
||||
if gc.manifestBelongsToHold(&manifest, holdDID) {
|
||||
manifests = append(manifests, &manifestInfo{
|
||||
URI: rec.URI,
|
||||
UserDID: userDID,
|
||||
Record: &manifest,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
if listResult.Cursor == "" {
|
||||
break
|
||||
}
|
||||
cursor = listResult.Cursor
|
||||
}
|
||||
|
||||
return manifests, nil
|
||||
}
|
||||
|
||||
// manifestBelongsToHold checks if a manifest references this hold via HoldDID or legacy HoldEndpoint.
|
||||
func (gc *GarbageCollector) manifestBelongsToHold(manifest *atproto.ManifestRecord, holdDID string) bool {
|
||||
if manifest.HoldDID == holdDID {
|
||||
return true
|
||||
}
|
||||
// Legacy: check holdEndpoint converted to DID
|
||||
if manifest.HoldEndpoint != "" {
|
||||
resolved := atproto.ResolveHoldDIDFromURL(manifest.HoldEndpoint)
|
||||
return resolved == holdDID
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// deleteOrphanedRecords removes layer records whose manifests no longer exist
|
||||
func (gc *GarbageCollector) deleteOrphanedRecords(ctx context.Context, orphanedRkeys []string, result *GCResult) error {
|
||||
for _, rkey := range orphanedRkeys {
|
||||
if gc.cfg.DryRun {
|
||||
gc.logger.Info("DRY-RUN: Would delete layer record", "rkey", rkey)
|
||||
} else {
|
||||
if err := gc.pds.DeleteLayerRecord(ctx, rkey); err != nil {
|
||||
gc.logger.Error("Failed to delete layer record", "rkey", rkey, "error", err)
|
||||
continue
|
||||
}
|
||||
result.RecordsDeleted++
|
||||
gc.logger.Debug("Deleted orphaned layer record", "rkey", rkey)
|
||||
if err := gc.pds.DeleteLayerRecord(ctx, rkey); err != nil {
|
||||
gc.logger.Error("Failed to delete layer record", "rkey", rkey, "error", err)
|
||||
continue
|
||||
}
|
||||
result.RecordsDeleted++
|
||||
gc.logger.Debug("Deleted orphaned layer record", "rkey", rkey)
|
||||
}
|
||||
|
||||
gc.logger.Info("Phase 2 complete",
|
||||
"orphaned", len(orphanedRkeys),
|
||||
"deleted", result.RecordsDeleted,
|
||||
"dryRun", gc.cfg.DryRun)
|
||||
"deleted", result.RecordsDeleted)
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -259,21 +700,15 @@ func (gc *GarbageCollector) deleteOrphanedBlobs(ctx context.Context, referenced
|
||||
|
||||
result.OrphanedBlobs++
|
||||
|
||||
if gc.cfg.DryRun {
|
||||
gc.logger.Info("DRY-RUN: Would delete blob",
|
||||
"digest", digest,
|
||||
"size", fi.Size())
|
||||
} else {
|
||||
if err := gc.driver.Delete(ctx, fi.Path()); err != nil {
|
||||
gc.logger.Error("Failed to delete blob", "path", fi.Path(), "error", err)
|
||||
return nil // Continue with other blobs
|
||||
}
|
||||
result.BlobsDeleted++
|
||||
result.BytesReclaimed += fi.Size()
|
||||
gc.logger.Debug("Deleted orphaned blob",
|
||||
"digest", digest,
|
||||
"size", fi.Size())
|
||||
if err := gc.driver.Delete(ctx, fi.Path()); err != nil {
|
||||
gc.logger.Error("Failed to delete blob", "path", fi.Path(), "error", err)
|
||||
return nil // Continue with other blobs
|
||||
}
|
||||
result.BlobsDeleted++
|
||||
result.BytesReclaimed += fi.Size()
|
||||
gc.logger.Debug("Deleted orphaned blob",
|
||||
"digest", digest,
|
||||
"size", fi.Size())
|
||||
|
||||
return nil
|
||||
})
|
||||
@@ -285,8 +720,7 @@ func (gc *GarbageCollector) deleteOrphanedBlobs(ctx context.Context, referenced
|
||||
gc.logger.Info("Phase 3 complete",
|
||||
"orphanedBlobs", result.OrphanedBlobs,
|
||||
"deleted", result.BlobsDeleted,
|
||||
"reclaimed", result.BytesReclaimed,
|
||||
"dryRun", gc.cfg.DryRun)
|
||||
"reclaimed", result.BytesReclaimed)
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -309,28 +743,6 @@ func (gc *GarbageCollector) decodeLayerRecord(ctx context.Context, rec pds.Recor
|
||||
return &layer, nil
|
||||
}
|
||||
|
||||
// manifestExists checks if a manifest still exists at the given AT-URI
|
||||
func (gc *GarbageCollector) manifestExists(ctx context.Context, manifestURI string) bool {
|
||||
// Parse AT-URI: at://did:plc:xxx/io.atcr.manifest/abc123
|
||||
parts := parseATURI(manifestURI)
|
||||
if parts == nil {
|
||||
gc.logger.Debug("Could not parse manifest URI", "uri", manifestURI)
|
||||
return false // Can't parse, assume orphaned
|
||||
}
|
||||
|
||||
// Check if the manifest record still exists via XRPC
|
||||
exists, err := gc.checkManifestViaXRPC(ctx, parts.DID, parts.Collection, parts.Rkey)
|
||||
if err != nil {
|
||||
// Network error - assume manifest exists (safe default)
|
||||
gc.logger.Warn("Failed to check manifest existence, assuming exists",
|
||||
"uri", manifestURI,
|
||||
"error", err)
|
||||
return true
|
||||
}
|
||||
|
||||
return exists
|
||||
}
|
||||
|
||||
// atURIParts contains parsed components of an AT-URI
|
||||
type atURIParts struct {
|
||||
DID string
|
||||
@@ -361,47 +773,6 @@ func parseATURI(uri string) *atURIParts {
|
||||
}
|
||||
}
|
||||
|
||||
// checkManifestViaXRPC checks if a manifest record exists by querying the user's PDS
|
||||
func (gc *GarbageCollector) checkManifestViaXRPC(ctx context.Context, did, collection, rkey string) (bool, error) {
|
||||
// Resolve DID to PDS endpoint
|
||||
pdsEndpoint, err := atproto.ResolveDIDToPDS(ctx, did)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("resolve PDS: %w", err)
|
||||
}
|
||||
|
||||
// Build XRPC URL
|
||||
url := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",
|
||||
pdsEndpoint, did, collection, rkey)
|
||||
|
||||
// Make request with timeout
|
||||
client := &http.Client{Timeout: 10 * time.Second}
|
||||
req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("create request: %w", err)
|
||||
}
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("http request: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
// Consume body to allow connection reuse
|
||||
_, _ = io.Copy(io.Discard, resp.Body)
|
||||
|
||||
switch resp.StatusCode {
|
||||
case http.StatusOK:
|
||||
return true, nil
|
||||
case http.StatusNotFound, http.StatusBadRequest:
|
||||
// Record doesn't exist
|
||||
return false, nil
|
||||
default:
|
||||
// Read error body for debugging
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
return false, fmt.Errorf("unexpected status %d: %s", resp.StatusCode, string(body))
|
||||
}
|
||||
}
|
||||
|
||||
// tidToTime extracts the timestamp from a TID (Timestamp ID)
|
||||
// TIDs are 13-character base32 encoded timestamps with counter
|
||||
func tidToTime(tid string) time.Time {
|
||||
@@ -435,12 +806,40 @@ func (gc *GarbageCollector) logResult(result *GCResult) {
|
||||
"referencedBlobs", result.ReferencedBlobs,
|
||||
"orphanedRecords", result.OrphanedRecords,
|
||||
"recordsDeleted", result.RecordsDeleted,
|
||||
"recordsReconciled", result.RecordsReconciled,
|
||||
"manifestsChecked", result.ManifestsChecked,
|
||||
"usersChecked", result.UsersChecked,
|
||||
"orphanedBlobs", result.OrphanedBlobs,
|
||||
"blobsDeleted", result.BlobsDeleted,
|
||||
"bytesReclaimed", result.BytesReclaimed,
|
||||
"dryRun", gc.cfg.DryRun)
|
||||
"bytesReclaimed", result.BytesReclaimed)
|
||||
|
||||
// Also log as JSON for easier parsing
|
||||
resultJSON, _ := json.Marshal(result)
|
||||
gc.logger.Debug("GC result JSON", "result", string(resultJSON))
|
||||
}
|
||||
|
||||
// LastPreview returns the most recent preview result and when it was generated
|
||||
func (gc *GarbageCollector) LastPreview() (*GCPreview, time.Time) {
|
||||
gc.mu.Lock()
|
||||
defer gc.mu.Unlock()
|
||||
return gc.lastPreview, gc.lastPreviewAt
|
||||
}
|
||||
|
||||
// LastResult returns the most recent GC run result and when it was generated
|
||||
func (gc *GarbageCollector) LastResult() (*GCResult, time.Time) {
|
||||
gc.mu.Lock()
|
||||
defer gc.mu.Unlock()
|
||||
return gc.lastResult, gc.lastResultAt
|
||||
}
|
||||
|
||||
// IsRunning returns whether a GC operation is currently in progress
|
||||
func (gc *GarbageCollector) IsRunning() bool {
|
||||
gc.mu.Lock()
|
||||
defer gc.mu.Unlock()
|
||||
return gc.running
|
||||
}
|
||||
|
||||
// GetConfig returns the current GC configuration
|
||||
func (gc *GarbageCollector) GetConfig() Config {
|
||||
return gc.cfg
|
||||
}
|
||||
|
||||
+269
-42
@@ -1,8 +1,15 @@
|
||||
package gc
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"atcr.io/pkg/atproto"
|
||||
)
|
||||
|
||||
func TestExtractDigestFromPath(t *testing.T) {
|
||||
@@ -176,53 +183,273 @@ func TestTidToTime(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadConfigFromEnv(t *testing.T) {
|
||||
// Test default values
|
||||
t.Run("default values", func(t *testing.T) {
|
||||
// Clear any existing env vars
|
||||
t.Setenv("GC_ENABLED", "")
|
||||
t.Setenv("GC_DRY_RUN", "")
|
||||
|
||||
cfg := LoadConfigFromEnv()
|
||||
|
||||
// Default: enabled
|
||||
if !cfg.Enabled {
|
||||
t.Error("expected Enabled to be true by default")
|
||||
}
|
||||
|
||||
// Default: dry run enabled
|
||||
if !cfg.DryRun {
|
||||
t.Error("expected DryRun to be true by default")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("disabled via env", func(t *testing.T) {
|
||||
t.Setenv("GC_ENABLED", "false")
|
||||
t.Setenv("GC_DRY_RUN", "false")
|
||||
|
||||
cfg := LoadConfigFromEnv()
|
||||
|
||||
func TestConfig(t *testing.T) {
|
||||
t.Run("zero value is disabled", func(t *testing.T) {
|
||||
cfg := Config{}
|
||||
if cfg.Enabled {
|
||||
t.Error("expected Enabled to be false when GC_ENABLED=false")
|
||||
}
|
||||
|
||||
if cfg.DryRun {
|
||||
t.Error("expected DryRun to be false when GC_DRY_RUN=false")
|
||||
t.Error("expected zero-value Enabled to be false")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("enabled via env", func(t *testing.T) {
|
||||
t.Setenv("GC_ENABLED", "true")
|
||||
t.Setenv("GC_DRY_RUN", "true")
|
||||
|
||||
cfg := LoadConfigFromEnv()
|
||||
|
||||
t.Run("explicit enabled", func(t *testing.T) {
|
||||
cfg := Config{Enabled: true}
|
||||
if !cfg.Enabled {
|
||||
t.Error("expected Enabled to be true when GC_ENABLED=true")
|
||||
}
|
||||
|
||||
if !cfg.DryRun {
|
||||
t.Error("expected DryRun to be true when GC_DRY_RUN=true")
|
||||
t.Error("expected Enabled to be true")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestManifestBelongsToHold(t *testing.T) {
|
||||
gc := &GarbageCollector{}
|
||||
holdDID := "did:web:hold01.atcr.io"
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
manifest *atproto.ManifestRecord
|
||||
want bool
|
||||
}{
|
||||
{
|
||||
name: "matching HoldDID",
|
||||
manifest: &atproto.ManifestRecord{
|
||||
HoldDID: "did:web:hold01.atcr.io",
|
||||
},
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "non-matching HoldDID",
|
||||
manifest: &atproto.ManifestRecord{
|
||||
HoldDID: "did:web:other-hold.atcr.io",
|
||||
},
|
||||
want: false,
|
||||
},
|
||||
{
|
||||
name: "legacy HoldEndpoint matching",
|
||||
manifest: &atproto.ManifestRecord{
|
||||
HoldEndpoint: "https://hold01.atcr.io",
|
||||
},
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "legacy HoldEndpoint non-matching",
|
||||
manifest: &atproto.ManifestRecord{
|
||||
HoldEndpoint: "https://other-hold.atcr.io",
|
||||
},
|
||||
want: false,
|
||||
},
|
||||
{
|
||||
name: "HoldDID takes precedence over endpoint",
|
||||
manifest: &atproto.ManifestRecord{
|
||||
HoldDID: "did:web:hold01.atcr.io",
|
||||
HoldEndpoint: "https://other-hold.atcr.io",
|
||||
},
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "empty manifest",
|
||||
manifest: &atproto.ManifestRecord{},
|
||||
want: false,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got := gc.manifestBelongsToHold(tt.manifest, holdDID)
|
||||
if got != tt.want {
|
||||
t.Errorf("manifestBelongsToHold() = %v, want %v", got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestFetchUserManifests(t *testing.T) {
|
||||
holdDID := "did:web:hold.example.com"
|
||||
|
||||
// Create a mock PDS server
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
collection := r.URL.Query().Get("collection")
|
||||
if collection != atproto.ManifestCollection {
|
||||
t.Errorf("unexpected collection: %s", collection)
|
||||
}
|
||||
|
||||
// Return manifests: one for this hold, one for another hold, one multi-arch
|
||||
response := map[string]any{
|
||||
"records": []map[string]any{
|
||||
{
|
||||
"uri": "at://did:plc:user1/io.atcr.manifest/abc123",
|
||||
"cid": "bafyrei1",
|
||||
"value": map[string]any{
|
||||
"$type": "io.atcr.manifest",
|
||||
"repository": "myapp",
|
||||
"digest": "sha256:abc123",
|
||||
"holdDid": holdDID,
|
||||
"mediaType": "application/vnd.oci.image.manifest.v1+json",
|
||||
"layers": []map[string]any{
|
||||
{"digest": "sha256:layer1", "size": 1000, "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip"},
|
||||
{"digest": "sha256:layer2", "size": 2000, "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip"},
|
||||
},
|
||||
"config": map[string]any{
|
||||
"digest": "sha256:config1",
|
||||
"size": 500,
|
||||
"mediaType": "application/vnd.oci.image.config.v1+json",
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
"uri": "at://did:plc:user1/io.atcr.manifest/def456",
|
||||
"cid": "bafyrei2",
|
||||
"value": map[string]any{
|
||||
"$type": "io.atcr.manifest",
|
||||
"repository": "otherapp",
|
||||
"digest": "sha256:def456",
|
||||
"holdDid": "did:web:other-hold.example.com",
|
||||
"mediaType": "application/vnd.oci.image.manifest.v1+json",
|
||||
"layers": []map[string]any{
|
||||
{"digest": "sha256:layer3", "size": 3000, "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip"},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
"uri": "at://did:plc:user1/io.atcr.manifest/ghi789",
|
||||
"cid": "bafyrei3",
|
||||
"value": map[string]any{
|
||||
"$type": "io.atcr.manifest",
|
||||
"repository": "multiarch",
|
||||
"digest": "sha256:ghi789",
|
||||
"holdDid": holdDID,
|
||||
"mediaType": "application/vnd.oci.image.index.v1+json",
|
||||
"manifests": []map[string]any{
|
||||
{"digest": "sha256:abc123", "size": 500, "mediaType": "application/vnd.oci.image.manifest.v1+json"},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(response)
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
gc := &GarbageCollector{
|
||||
pds: nil, // Not used directly in fetchUserManifests when we bypass DID resolution
|
||||
logger: newTestLogger(),
|
||||
}
|
||||
|
||||
// Call fetchUserManifestsFromEndpoint (bypasses DID resolution)
|
||||
manifests, err := gc.fetchUserManifestsFromEndpoint(
|
||||
t.Context(), "did:plc:user1", server.URL, holdDID)
|
||||
if err != nil {
|
||||
t.Fatalf("fetchUserManifestsFromEndpoint() error = %v", err)
|
||||
}
|
||||
|
||||
// Should only include manifests for this hold (the regular one and the multi-arch)
|
||||
if len(manifests) != 2 {
|
||||
t.Fatalf("expected 2 manifests for this hold, got %d", len(manifests))
|
||||
}
|
||||
|
||||
// First should be the regular manifest with layers
|
||||
m1 := manifests[0]
|
||||
if m1.URI != "at://did:plc:user1/io.atcr.manifest/abc123" {
|
||||
t.Errorf("unexpected URI: %s", m1.URI)
|
||||
}
|
||||
if m1.UserDID != "did:plc:user1" {
|
||||
t.Errorf("unexpected UserDID: %s", m1.UserDID)
|
||||
}
|
||||
if len(m1.Record.Layers) != 2 {
|
||||
t.Errorf("expected 2 layers, got %d", len(m1.Record.Layers))
|
||||
}
|
||||
if m1.Record.Config == nil || m1.Record.Config.Digest != "sha256:config1" {
|
||||
t.Error("expected config digest sha256:config1")
|
||||
}
|
||||
|
||||
// Second should be the multi-arch manifest (no layers, has manifests)
|
||||
m2 := manifests[1]
|
||||
if len(m2.Record.Layers) != 0 {
|
||||
t.Errorf("multi-arch manifest should have no layers, got %d", len(m2.Record.Layers))
|
||||
}
|
||||
if len(m2.Record.Manifests) != 1 {
|
||||
t.Errorf("expected 1 manifest reference, got %d", len(m2.Record.Manifests))
|
||||
}
|
||||
}
|
||||
|
||||
func TestFetchUserManifests_Pagination(t *testing.T) {
|
||||
holdDID := "did:web:hold.example.com"
|
||||
callCount := 0
|
||||
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
callCount++
|
||||
cursor := r.URL.Query().Get("cursor")
|
||||
|
||||
var response map[string]any
|
||||
if cursor == "" {
|
||||
// First page
|
||||
response = map[string]any{
|
||||
"records": []map[string]any{
|
||||
{
|
||||
"uri": "at://did:plc:user1/io.atcr.manifest/page1",
|
||||
"cid": "bafyrei1",
|
||||
"value": map[string]any{
|
||||
"$type": "io.atcr.manifest",
|
||||
"holdDid": holdDID,
|
||||
"layers": []map[string]any{{"digest": "sha256:l1", "size": 100}},
|
||||
},
|
||||
},
|
||||
},
|
||||
"cursor": "page2cursor",
|
||||
}
|
||||
} else {
|
||||
// Second page (no more cursor)
|
||||
response = map[string]any{
|
||||
"records": []map[string]any{
|
||||
{
|
||||
"uri": "at://did:plc:user1/io.atcr.manifest/page2",
|
||||
"cid": "bafyrei2",
|
||||
"value": map[string]any{
|
||||
"$type": "io.atcr.manifest",
|
||||
"holdDid": holdDID,
|
||||
"layers": []map[string]any{{"digest": "sha256:l2", "size": 200}},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(response)
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
gc := &GarbageCollector{logger: newTestLogger()}
|
||||
|
||||
manifests, err := gc.fetchUserManifestsFromEndpoint(
|
||||
t.Context(), "did:plc:user1", server.URL, holdDID)
|
||||
if err != nil {
|
||||
t.Fatalf("error = %v", err)
|
||||
}
|
||||
|
||||
if callCount != 2 {
|
||||
t.Errorf("expected 2 HTTP calls (pagination), got %d", callCount)
|
||||
}
|
||||
if len(manifests) != 2 {
|
||||
t.Fatalf("expected 2 manifests, got %d", len(manifests))
|
||||
}
|
||||
}
|
||||
|
||||
func TestFetchUserManifests_PDSError(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
fmt.Fprint(w, "internal error")
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
gc := &GarbageCollector{logger: newTestLogger()}
|
||||
|
||||
_, err := gc.fetchUserManifestsFromEndpoint(
|
||||
t.Context(), "did:plc:user1", server.URL, "did:web:hold.example.com")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for 500 response")
|
||||
}
|
||||
}
|
||||
|
||||
func newTestLogger() *slog.Logger {
|
||||
return slog.Default().With("component", "gc-test")
|
||||
}
|
||||
|
||||
@@ -454,6 +454,28 @@ func (ri *RecordsIndex) QuotasByDID(collection string) (map[string]QuotaResult,
|
||||
return result, rows.Err()
|
||||
}
|
||||
|
||||
// DistinctDIDs returns all unique DIDs that have records in a collection.
|
||||
// Used by GC to discover all users who have pushed to this hold.
|
||||
func (ri *RecordsIndex) DistinctDIDs(collection string) ([]string, error) {
|
||||
rows, err := ri.db.Query(`
|
||||
SELECT DISTINCT did FROM records WHERE collection = ? AND did IS NOT NULL
|
||||
`, collection)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to query distinct DIDs: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var dids []string
|
||||
for rows.Next() {
|
||||
var did string
|
||||
if err := rows.Scan(&did); err != nil {
|
||||
return nil, fmt.Errorf("failed to scan DID: %w", err)
|
||||
}
|
||||
dids = append(dids, did)
|
||||
}
|
||||
return dids, rows.Err()
|
||||
}
|
||||
|
||||
// QuotaResult holds aggregated quota data from SQL queries
|
||||
type QuotaResult struct {
|
||||
UniqueBlobs int
|
||||
|
||||
@@ -740,3 +740,62 @@ func TestRecordsIndex_QuotaForDID_IgnoresNullDigest(t *testing.T) {
|
||||
t.Errorf("Expected total size 100, got %d", totalSize)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRecordsIndex_DistinctDIDs tests retrieving unique DIDs from a collection
|
||||
func TestRecordsIndex_DistinctDIDs(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
ri, err := NewRecordsIndex(filepath.Join(tmpDir, "records.db"))
|
||||
if err != nil {
|
||||
t.Fatalf("NewRecordsIndex() error = %v", err)
|
||||
}
|
||||
defer ri.Close()
|
||||
|
||||
// Add layer records for multiple users (some with duplicate DIDs)
|
||||
ri.IndexRecord("io.atcr.hold.layer", "r1", "cid1", "did:plc:alice", "sha256:aaa", 100)
|
||||
ri.IndexRecord("io.atcr.hold.layer", "r2", "cid2", "did:plc:alice", "sha256:bbb", 200)
|
||||
ri.IndexRecord("io.atcr.hold.layer", "r3", "cid3", "did:plc:bob", "sha256:ccc", 500)
|
||||
// Non-layer record should be excluded
|
||||
ri.IndexRecord("io.atcr.hold.crew", "r4", "cid4", "did:plc:charlie", "", 0)
|
||||
|
||||
dids, err := ri.DistinctDIDs("io.atcr.hold.layer")
|
||||
if err != nil {
|
||||
t.Fatalf("DistinctDIDs() error = %v", err)
|
||||
}
|
||||
|
||||
if len(dids) != 2 {
|
||||
t.Fatalf("Expected 2 distinct DIDs, got %d: %v", len(dids), dids)
|
||||
}
|
||||
|
||||
didSet := make(map[string]bool)
|
||||
for _, d := range dids {
|
||||
didSet[d] = true
|
||||
}
|
||||
|
||||
if !didSet["did:plc:alice"] {
|
||||
t.Error("Expected did:plc:alice in results")
|
||||
}
|
||||
if !didSet["did:plc:bob"] {
|
||||
t.Error("Expected did:plc:bob in results")
|
||||
}
|
||||
if didSet["did:plc:charlie"] {
|
||||
t.Error("did:plc:charlie should not be in layer DIDs")
|
||||
}
|
||||
}
|
||||
|
||||
// TestRecordsIndex_DistinctDIDs_Empty tests DistinctDIDs with no records
|
||||
func TestRecordsIndex_DistinctDIDs_Empty(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
ri, err := NewRecordsIndex(filepath.Join(tmpDir, "records.db"))
|
||||
if err != nil {
|
||||
t.Fatalf("NewRecordsIndex() error = %v", err)
|
||||
}
|
||||
defer ri.Close()
|
||||
|
||||
dids, err := ri.DistinctDIDs("io.atcr.hold.layer")
|
||||
if err != nil {
|
||||
t.Fatalf("DistinctDIDs() error = %v", err)
|
||||
}
|
||||
if len(dids) != 0 {
|
||||
t.Errorf("Expected empty DIDs slice, got %d entries", len(dids))
|
||||
}
|
||||
}
|
||||
|
||||
+3
-5
@@ -188,11 +188,9 @@ func NewHoldServer(cfg *Config) (*HoldServer, error) {
|
||||
}
|
||||
|
||||
// Initialize garbage collector
|
||||
gcConfig := gc.LoadConfigFromEnv()
|
||||
s.garbageCollector = gc.NewGarbageCollector(s.PDS, driver, gcConfig)
|
||||
s.garbageCollector = gc.NewGarbageCollector(s.PDS, driver, cfg.GC)
|
||||
slog.Info("Garbage collector initialized",
|
||||
"enabled", gcConfig.Enabled,
|
||||
"dryRun", gcConfig.DryRun)
|
||||
"enabled", cfg.GC.Enabled)
|
||||
}
|
||||
|
||||
// Setup HTTP routes with chi router
|
||||
@@ -230,7 +228,7 @@ func NewHoldServer(cfg *Config) (*HoldServer, error) {
|
||||
ConfigPath: cfg.ConfigPath(),
|
||||
}
|
||||
|
||||
s.adminUI, err = admin.NewAdminUI(context.Background(), s.PDS, s.QuotaManager, adminCfg)
|
||||
s.adminUI, err = admin.NewAdminUI(context.Background(), s.PDS, s.QuotaManager, s.garbageCollector, adminCfg)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to initialize admin panel: %w", err)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user