From 9e600649a6ca8bf6cd15ff0cd8d29c3b601d5635 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Thu, 8 Jan 2026 23:31:30 -0600 Subject: [PATCH] begin s3 garbage collection implementation, more envvar cleanup --- .env.appview.example | 11 - .env.example | 13 - .env.hold.example | 12 + CLAUDE.md | 3 +- cmd/appview/serve.go | 4 +- cmd/hold/main.go | 20 ++ deploy/.env.prod.template | 16 -- deploy/docker-compose.prod.yml | 3 - docs/appview.md | 15 -- pkg/appview/config.go | 26 +- pkg/appview/config_test.go | 27 +- pkg/hold/gc/config.go | 37 +++ pkg/hold/gc/gc.go | 446 +++++++++++++++++++++++++++++++++ pkg/hold/gc/gc_test.go | 228 +++++++++++++++++ pkg/hold/pds/layer.go | 19 ++ pkg/hold/pds/server.go | 30 +++ 16 files changed, 804 insertions(+), 106 deletions(-) create mode 100644 pkg/hold/gc/config.go create mode 100644 pkg/hold/gc/gc.go create mode 100644 pkg/hold/gc/gc_test.go diff --git a/.env.appview.example b/.env.appview.example index 4445db0..fd0659c 100644 --- a/.env.appview.example +++ b/.env.appview.example @@ -21,10 +21,6 @@ ATCR_HTTP_ADDR=:5000 # Production: Set to your public URL (e.g., https://atcr.io) # ATCR_BASE_URL=http://127.0.0.1:5000 -# Service name (used for JWT service/issuer fields) -# Default: Derived from base URL hostname, or "atcr.io" -# ATCR_SERVICE_NAME=atcr.io - # ============================================================================== # Storage Configuration # ============================================================================== @@ -49,9 +45,6 @@ ATCR_DEFAULT_HOLD_DID=did:web:127.0.0.1:8080 # Default: /var/lib/atcr/auth/private-key.crt # ATCR_AUTH_CERT_PATH=/var/lib/atcr/auth/private-key.crt -# JWT token expiration in seconds (default: 300 = 5 minutes) -# ATCR_TOKEN_EXPIRATION=300 - # Path to OAuth client P-256 signing key (auto-generated on first run) # Used for confidential OAuth client authentication (production only) # Localhost deployments always use public OAuth clients (no key needed) @@ -130,7 +123,3 @@ ATCR_LOG_LEVEL=debug # ATProto relay endpoint for backfill sync API # Default: https://relay1.us-east.bsky.network # ATCR_RELAY_ENDPOINT=https://relay1.us-east.bsky.network - -# Backfill interval (default: 1h) -# Examples: 30m, 1h, 2h, 24h -# ATCR_BACKFILL_INTERVAL=1h diff --git a/.env.example b/.env.example index fc653eb..ae81f93 100644 --- a/.env.example +++ b/.env.example @@ -45,10 +45,6 @@ # Production: Set to your public URL (e.g., https://atcr.io) # ATCR_BASE_URL=https://atcr.io -# Service name for JWT issuer/service fields -# Default: Derived from ATCR_BASE_URL hostname, or "atcr.io" -# ATCR_SERVICE_NAME=atcr.io - # ============================================================================== # APPVIEW - STORAGE CONFIGURATION (REQUIRED) # ============================================================================== @@ -72,10 +68,6 @@ ATCR_DEFAULT_HOLD_DID=did:web:127.0.0.1:8080 # Default: /var/lib/atcr/auth/private-key.crt # ATCR_AUTH_CERT_PATH=/var/lib/atcr/auth/private-key.crt -# JWT token expiration in seconds -# Default: 300 (5 minutes) -# ATCR_TOKEN_EXPIRATION=300 - # Path to OAuth client P-256 signing key (auto-generated for production) # Used for confidential OAuth client authentication # Localhost deployments always use public OAuth clients (no key needed) @@ -110,11 +102,6 @@ ATCR_DEFAULT_HOLD_DID=did:web:127.0.0.1:8080 # Default: https://relay1.us-east.bsky.network # ATCR_RELAY_ENDPOINT=https://relay1.us-east.bsky.network -# Backfill sync interval -# Default: 1h -# Examples: 30m, 1h, 2h, 24h -# ATCR_BACKFILL_INTERVAL=1h - # ============================================================================== # APPVIEW - HEALTH CHECKS # ============================================================================== diff --git a/.env.hold.example b/.env.hold.example index 8f7fbcb..e8ef19a 100644 --- a/.env.hold.example +++ b/.env.hold.example @@ -151,3 +151,15 @@ ATCR_LOG_LEVEL=debug # Basic auth credentials (optional) # ATCR_LOG_SHIPPER_USERNAME= # ATCR_LOG_SHIPPER_PASSWORD= + +# ============================================================================== +# Garbage Collection +# ============================================================================== + +# Enable garbage collection for orphaned blobs (default: true) +# GC runs on startup and then nightly (every 24 hours) +GC_ENABLED=true + +# Dry-run mode: log what would be deleted without actually deleting (default: true) +# Set to false after validating the GC logs show correct behavior +GC_DRY_RUN=true diff --git a/CLAUDE.md b/CLAUDE.md index 54719bc..5e9647a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -230,7 +230,7 @@ ATCR uses three distinct token types in its authentication flow: - **Issued by:** AppView after OAuth login - **Stored in:** Docker credential helper (`~/.atcr/credential-helper-token.json`) - **Used for:** Docker client → AppView authentication -- **Lifetime:** 15 minutes (configurable via `ATCR_TOKEN_EXPIRATION`) +- **Lifetime:** 5 minutes - **Format:** JWT with DID claim **3. Service Tokens** @@ -666,7 +666,6 @@ See `.env.appview.example` for all available options. Key environment variables: **Authentication:** - `ATCR_AUTH_KEY_PATH` - JWT signing key path (default: `/var/lib/atcr/auth/private-key.pem`) -- `ATCR_TOKEN_EXPIRATION` - JWT expiration in seconds (default: 300) **UI:** - `ATCR_UI_DATABASE_PATH` - SQLite database path (default: `/var/lib/atcr/ui.db`) diff --git a/cmd/appview/serve.go b/cmd/appview/serve.go index a06e762..36c8b89 100644 --- a/cmd/appview/serve.go +++ b/cmd/appview/serve.go @@ -565,8 +565,8 @@ func initializeJetstream(database *sql.DB, jetstreamCfg *appview.JetstreamConfig } }() - // Start periodic backfill scheduler - interval := jetstreamCfg.BackfillInterval + // Start periodic backfill scheduler (hardcoded 1h interval) + interval := 1 * time.Hour go func() { ticker := time.NewTicker(interval) diff --git a/cmd/hold/main.go b/cmd/hold/main.go index 2ccf14e..fd9b753 100644 --- a/cmd/hold/main.go +++ b/cmd/hold/main.go @@ -12,6 +12,7 @@ import ( "atcr.io/pkg/hold" "atcr.io/pkg/hold/admin" + "atcr.io/pkg/hold/gc" "atcr.io/pkg/hold/oci" "atcr.io/pkg/hold/pds" "atcr.io/pkg/hold/quota" @@ -122,6 +123,7 @@ func main() { // Create blob store adapter and XRPC handlers var ociHandler *oci.XRPCHandler + var garbageCollector *gc.GarbageCollector if holdPDS != nil { // Create storage driver from config ctx := context.Background() @@ -142,6 +144,13 @@ func main() { // Create OCI XRPC handler (multipart upload endpoints) ociHandler = oci.NewXRPCHandler(holdPDS, *s3Service, driver, cfg.Server.DisablePresignedURLs, cfg.Registration.EnableBlueskyPosts, nil, quotaMgr) + + // Initialize garbage collector + gcConfig := gc.LoadConfigFromEnv() + garbageCollector = gc.NewGarbageCollector(holdPDS, driver, gcConfig) + slog.Info("Garbage collector initialized", + "enabled", gcConfig.Enabled, + "dryRun", gcConfig.DryRun) } // Setup HTTP routes with chi router @@ -238,6 +247,11 @@ func main() { } } + // Start garbage collector (runs on startup + nightly) + if garbageCollector != nil { + garbageCollector.Start(context.Background()) + } + // Wait for signal or server error select { case err := <-serverErr: @@ -257,6 +271,12 @@ func main() { } } + // Stop garbage collector + if garbageCollector != nil { + garbageCollector.Stop() + slog.Info("Garbage collector stopped") + } + // Close broadcaster database connection if broadcaster != nil { if err := broadcaster.Close(); err != nil { diff --git a/deploy/.env.prod.template b/deploy/.env.prod.template index af8d1de..3d943cb 100644 --- a/deploy/.env.prod.template +++ b/deploy/.env.prod.template @@ -142,10 +142,6 @@ S3_ENDPOINT=https://6vmss.upcloudobjects.com # Uncomment to override if you want to use a different hold service as the default # ATCR_DEFAULT_HOLD_DID=did:web:some-other-hold.example.com -# JWT token expiration in seconds -# Default: 300 (5 minutes) -ATCR_TOKEN_EXPIRATION=300 - # OAuth client display name (shown in authorization screens) # Default: AT Container Registry # ATCR_CLIENT_NAME=AT Container Registry @@ -178,11 +174,6 @@ ATCR_BACKFILL_ENABLED=true # Default: https://relay1.us-east.bsky.network ATCR_RELAY_ENDPOINT=https://relay1.us-east.bsky.network -# Backfill interval -# Examples: 30m, 1h, 2h, 24h -# Default: 1h -ATCR_BACKFILL_INTERVAL=1h - # ============================================================================== # Optional: Filesystem Storage (alternative to S3) # ============================================================================== @@ -195,13 +186,6 @@ ATCR_BACKFILL_INTERVAL=1h # STORAGE_DRIVER=filesystem # STORAGE_ROOT_DIR=/var/lib/atcr/hold -# ============================================================================== -# Advanced Configuration -# ============================================================================== - -# Override service name (defaults to APPVIEW_DOMAIN) -# ATCR_SERVICE_NAME=atcr.io - # ============================================================================== # CHECKLIST # ============================================================================== diff --git a/deploy/docker-compose.prod.yml b/deploy/docker-compose.prod.yml index 7fc6da6..570dd95 100644 --- a/deploy/docker-compose.prod.yml +++ b/deploy/docker-compose.prod.yml @@ -48,7 +48,6 @@ services: # Server configuration ATCR_HTTP_ADDR: :5000 ATCR_BASE_URL: https://${APPVIEW_DOMAIN:-atcr.io} - ATCR_SERVICE_NAME: ${APPVIEW_DOMAIN:-atcr.io} # Storage configuration (derived from HOLD_DOMAIN) ATCR_DEFAULT_HOLD_DID: ${ATCR_DEFAULT_HOLD_DID:-did:web:${HOLD_DOMAIN:-hold01.atcr.io}} @@ -56,7 +55,6 @@ services: # Authentication ATCR_AUTH_KEY_PATH: /var/lib/atcr/auth/private-key.pem ATCR_AUTH_CERT_PATH: /var/lib/atcr/auth/private-key.crt - ATCR_TOKEN_EXPIRATION: ${ATCR_TOKEN_EXPIRATION:-300} # UI configuration ATCR_UI_DATABASE_PATH: /var/lib/atcr/ui.db @@ -69,7 +67,6 @@ services: JETSTREAM_URL: ${JETSTREAM_URL:-wss://jetstream2.us-west.bsky.network/subscribe} ATCR_BACKFILL_ENABLED: ${ATCR_BACKFILL_ENABLED:-true} ATCR_RELAY_ENDPOINT: ${ATCR_RELAY_ENDPOINT:-https://relay1.us-east.bsky.network} - ATCR_BACKFILL_INTERVAL: ${ATCR_BACKFILL_INTERVAL:-1h} volumes: # Persistent data: auth keys, UI database, OAuth tokens, Jetstream cache - atcr-appview-data:/var/lib/atcr diff --git a/docs/appview.md b/docs/appview.md index d2ea1de..6a77b65 100644 --- a/docs/appview.md +++ b/docs/appview.md @@ -110,11 +110,6 @@ Or via Docker Compose (recommended). - **Production:** Set to your public URL (e.g., `https://atcr.example.com`) - **Example:** `https://atcr.io`, `http://127.0.0.1:5000` -#### `ATCR_SERVICE_NAME` -- **Default:** Derived from `ATCR_BASE_URL` hostname, or `atcr.io` -- **Description:** Service name used for JWT `service` and `issuer` fields. Controls token scope. -- **Example:** `atcr.io`, `registry.example.com` - ### Storage Configuration #### `ATCR_DEFAULT_HOLD_DID` ⚠️ REQUIRED @@ -138,11 +133,6 @@ Or via Docker Compose (recommended). - **Description:** Path to JWT signing certificate. Auto-generated if missing. - **Note:** Paired with `ATCR_AUTH_KEY_PATH` -#### `ATCR_TOKEN_EXPIRATION` -- **Default:** `300` (5 minutes) -- **Description:** JWT token expiration in seconds. Registry JWTs are short-lived for security. -- **Recommendation:** Keep between 300-900 seconds (5-15 minutes) - ### Web UI Configuration #### `ATCR_UI_DATABASE_PATH` @@ -200,11 +190,6 @@ Jetstream provides real-time indexing of ATProto records (manifests, tags) into - **Description:** ATProto relay endpoint for backfill sync API - **Note:** Used when `ATCR_BACKFILL_ENABLED=true` -#### `ATCR_BACKFILL_INTERVAL` -- **Default:** `1h` -- **Description:** How often to run backfill sync -- **Format:** Duration string (e.g., `30m`, `1h`, `2h`, `24h`) - ### Legacy Configuration #### `TEST_MODE` diff --git a/pkg/appview/config.go b/pkg/appview/config.go index 3f8bb85..413b8fb 100644 --- a/pkg/appview/config.go +++ b/pkg/appview/config.go @@ -102,9 +102,6 @@ type JetstreamConfig struct { // BackfillEnabled controls whether backfill is enabled (from env: ATCR_BACKFILL_ENABLED, default: true) BackfillEnabled bool `yaml:"backfill_enabled"` - // BackfillInterval is the backfill interval (from env: ATCR_BACKFILL_INTERVAL, default: 1h) - BackfillInterval time.Duration `yaml:"backfill_interval"` - // RelayEndpoint is the relay endpoint for sync API (from env: ATCR_RELAY_ENDPOINT, default: https://relay1.us-east.bsky.network) RelayEndpoint string `yaml:"relay_endpoint"` } @@ -117,11 +114,11 @@ type AuthConfig struct { // CertPath is the JWT certificate path (from env: ATCR_AUTH_CERT_PATH, default: "/var/lib/atcr/auth/private-key.crt") CertPath string `yaml:"cert_path"` - // TokenExpiration is the JWT expiration duration (from env: ATCR_TOKEN_EXPIRATION, default: 300s) + // TokenExpiration is the JWT expiration duration (5 minutes) TokenExpiration time.Duration `yaml:"token_expiration"` // ServiceName is the service name used for JWT issuer and service fields - // Derived from ATCR_SERVICE_NAME env var or extracted from base URL (e.g., "atcr.io") + // Derived from base URL hostname (e.g., "atcr.io") ServiceName string `yaml:"service_name"` } @@ -176,20 +173,14 @@ func LoadConfigFromEnv() (*Config, error) { // Jetstream configuration cfg.Jetstream.URL = getEnvOrDefault("JETSTREAM_URL", "wss://jetstream2.us-west.bsky.network/subscribe") cfg.Jetstream.BackfillEnabled = os.Getenv("ATCR_BACKFILL_ENABLED") != "false" - cfg.Jetstream.BackfillInterval = getDurationOrDefault("ATCR_BACKFILL_INTERVAL", 1*time.Hour) cfg.Jetstream.RelayEndpoint = getEnvOrDefault("ATCR_RELAY_ENDPOINT", "https://relay1.us-east.bsky.network") // Auth configuration cfg.Auth.KeyPath = getEnvOrDefault("ATCR_AUTH_KEY_PATH", "/var/lib/atcr/auth/private-key.pem") cfg.Auth.CertPath = getEnvOrDefault("ATCR_AUTH_CERT_PATH", "/var/lib/atcr/auth/private-key.crt") - // Parse token expiration (default: 300 seconds = 5 minutes) - expirationStr := getEnvOrDefault("ATCR_TOKEN_EXPIRATION", "300") - expirationSecs, err := strconv.Atoi(expirationStr) - if err != nil { - return nil, fmt.Errorf("invalid ATCR_TOKEN_EXPIRATION: %w", err) - } - cfg.Auth.TokenExpiration = time.Duration(expirationSecs) * time.Second + // Token expiration: 5 minutes (not configurable) + cfg.Auth.TokenExpiration = 5 * time.Minute // Derive service name from base URL or env var (used for JWT issuer and service) cfg.Auth.ServiceName = getServiceName(cfg.Server.BaseURL) @@ -336,14 +327,9 @@ func buildHealthConfig() configuration.Health { } } -// getServiceName extracts service name from base URL or uses env var +// getServiceName extracts service name from base URL hostname func getServiceName(baseURL string) string { - // Check env var first - if serviceName := os.Getenv("ATCR_SERVICE_NAME"); serviceName != "" { - return serviceName - } - - // Try to extract from base URL + // Extract from base URL parsed, err := url.Parse(baseURL) if err == nil && parsed.Hostname() != "" { hostname := parsed.Hostname() diff --git a/pkg/appview/config_test.go b/pkg/appview/config_test.go index 4a34339..eb7d01e 100644 --- a/pkg/appview/config_test.go +++ b/pkg/appview/config_test.go @@ -8,59 +8,39 @@ import ( func Test_getServiceName(t *testing.T) { tests := []struct { - name string - baseURL string - envService string - setEnv bool - want string + name string + baseURL string + want string }{ - { - name: "env var set", - baseURL: "http://127.0.0.1:5000", - envService: "custom.registry.io", - setEnv: true, - want: "custom.registry.io", - }, { name: "localhost - use default", baseURL: "http://localhost:5000", - setEnv: false, want: "atcr.io", }, { name: "127.0.0.1 - use default", baseURL: "http://127.0.0.1:5000", - setEnv: false, want: "atcr.io", }, { name: "custom domain", baseURL: "https://registry.example.com", - setEnv: false, want: "registry.example.com", }, { name: "domain with port", baseURL: "https://registry.example.com:443", - setEnv: false, want: "registry.example.com", }, { name: "invalid URL - use default", baseURL: "://invalid", - setEnv: false, want: "atcr.io", }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if tt.setEnv { - t.Setenv("ATCR_SERVICE_NAME", tt.envService) - } else { - os.Unsetenv("ATCR_SERVICE_NAME") - } - got := getServiceName(tt.baseURL) if got != tt.want { t.Errorf("getServiceName() = %v, want %v", got, tt.want) @@ -214,7 +194,6 @@ func TestLoadConfigFromEnv(t *testing.T) { // Clear other env vars to use defaults os.Unsetenv("ATCR_BASE_URL") - os.Unsetenv("ATCR_SERVICE_NAME") got, err := LoadConfigFromEnv() if (err != nil) != tt.wantError { diff --git a/pkg/hold/gc/config.go b/pkg/hold/gc/config.go new file mode 100644 index 0000000..38f2114 --- /dev/null +++ b/pkg/hold/gc/config.go @@ -0,0 +1,37 @@ +// Package gc implements garbage collection for the hold service. +// It periodically cleans up orphaned blobs from S3 storage based on +// layer records in the hold's embedded PDS. +package gc + +import ( + "os" + "time" +) + +// Hardcoded defaults - keep configuration simple +const ( + // gcInterval is how often GC runs (nightly) + gcInterval = 24 * time.Hour + + // gcGracePeriod is how old a layer record must be before it's considered for GC. + // Records created in the last 7 days are skipped (GDPR/CCPA compliant). + gcGracePeriod = 7 * 24 * time.Hour +) + +// Config holds GC configuration, loaded from environment variables +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 + } +} diff --git a/pkg/hold/gc/gc.go b/pkg/hold/gc/gc.go new file mode 100644 index 0000000..7f4ffaf --- /dev/null +++ b/pkg/hold/gc/gc.go @@ -0,0 +1,446 @@ +package gc + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "regexp" + "strings" + "sync" + "time" + + "atcr.io/pkg/atproto" + "atcr.io/pkg/hold/pds" + "github.com/bluesky-social/indigo/atproto/syntax" + storagedriver "github.com/distribution/distribution/v3/registry/storage/driver" +) + +// GarbageCollector handles cleanup of orphaned blobs from storage +type GarbageCollector struct { + pds *pds.HoldPDS + driver storagedriver.StorageDriver + cfg Config + logger *slog.Logger + + // stopCh signals the background goroutine to stop + stopCh chan struct{} + // wg tracks the background goroutine + wg sync.WaitGroup +} + +// 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"` +} + +// NewGarbageCollector creates a new GC instance +func NewGarbageCollector(holdPDS *pds.HoldPDS, driver storagedriver.StorageDriver, cfg Config) *GarbageCollector { + return &GarbageCollector{ + pds: holdPDS, + driver: driver, + cfg: cfg, + logger: slog.Default().With("component", "gc"), + stopCh: make(chan struct{}), + } +} + +// Start begins the GC background process +// It runs GC immediately on startup, then periodically according to gcInterval +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() + + ticker := time.NewTicker(gcInterval) + defer ticker.Stop() + + for { + select { + case <-gc.stopCh: + gc.logger.Info("GC background process stopped") + return + case <-ctx.Done(): + gc.logger.Info("GC context cancelled") + return + case <-ticker.C: + gc.logger.Info("Running nightly GC", "dryRun", gc.cfg.DryRun) + result, err := gc.Run(ctx) + if err != nil { + gc.logger.Error("Nightly GC failed", "error", err) + } else { + gc.logResult(result) + } + } + } + }() + + gc.logger.Info("GC background process started", "interval", gcInterval) +} + +// Stop gracefully stops the GC background process +func (gc *GarbageCollector) Stop() { + close(gc.stopCh) + gc.wg.Wait() +} + +// Run executes a single GC cycle +func (gc *GarbageCollector) Run(ctx context.Context) (*GCResult, error) { + start := time.Now() + result := &GCResult{} + + gc.logger.Info("Starting GC run", "dryRun", gc.cfg.DryRun) + + // Phase 1: Build referenced set from layer records + referenced, orphanedRecords, err := gc.buildReferencedSet(ctx, result) + if err != nil { + return nil, fmt.Errorf("phase 1 (build referenced set) failed: %w", err) + } + + gc.logger.Info("Phase 1 complete", + "referenced", len(referenced), + "orphanedRecords", len(orphanedRecords)) + + // Phase 2: Delete orphaned layer records + if err := gc.deleteOrphanedRecords(ctx, orphanedRecords, 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 { + return nil, fmt.Errorf("phase 3 (delete orphaned blobs) failed: %w", err) + } + + result.Duration = time.Since(start) + result.ReferencedBlobs = int64(len(referenced)) + + 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 + + recordsIndex := gc.pds.RecordsIndex() + if recordsIndex == nil { + return nil, nil, fmt.Errorf("records index not available") + } + + 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) + } + + for _, rec := range records { + 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 + recordTime := tidToTime(rec.Rkey) + if time.Since(recordTime) < gcGracePeriod { + // Recent record - assume referenced, skip checking + referenced[layer.Digest] = true + continue + } + + // Cross-check: does the manifest still exist? + if gc.manifestExists(ctx, layer.Manifest) { + 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) + } + } + + if nextCursor == "" { + break + } + cursor = nextCursor + + // Progress logging + if totalRecords%10000 == 0 { + gc.logger.Info("Phase 1 progress", "processed", totalRecords) + } + } + + gc.logger.Info("Scanned layer records", "total", totalRecords) + return referenced, orphanedRecords, nil +} + +// 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) + } + } + + gc.logger.Info("Phase 2 complete", + "orphaned", len(orphanedRkeys), + "deleted", result.RecordsDeleted, + "dryRun", gc.cfg.DryRun) + + return nil +} + +// deleteOrphanedBlobs walks storage and deletes blobs not in the referenced set +func (gc *GarbageCollector) deleteOrphanedBlobs(ctx context.Context, referenced map[string]bool, result *GCResult) error { + blobsPath := "/docker/registry/v2/blobs" + + err := gc.driver.Walk(ctx, blobsPath, func(fi storagedriver.FileInfo) error { + if fi.IsDir() { + return nil + } + + // Only process data files + if !strings.HasSuffix(fi.Path(), "/data") { + return nil + } + + // Extract digest from path + digest := extractDigestFromPath(fi.Path()) + if digest == "" { + return nil + } + + // Check if referenced by any layer record + if referenced[digest] { + return nil + } + + 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()) + } + + return nil + }) + + if err != nil { + return fmt.Errorf("walk storage failed: %w", err) + } + + gc.logger.Info("Phase 3 complete", + "orphanedBlobs", result.OrphanedBlobs, + "deleted", result.BlobsDeleted, + "reclaimed", result.BytesReclaimed, + "dryRun", gc.cfg.DryRun) + + return nil +} + +// decodeLayerRecord reads and decodes a layer record from the PDS +func (gc *GarbageCollector) decodeLayerRecord(ctx context.Context, rec pds.Record) (*atproto.LayerRecord, error) { + // Get the record from the repo + recordPath := rec.Collection + "/" + rec.Rkey + _, recBytes, err := gc.pds.GetRecordBytes(ctx, recordPath) + if err != nil { + return nil, fmt.Errorf("get record bytes: %w", err) + } + + // Decode the layer record + var layer atproto.LayerRecord + if err := layer.UnmarshalCBOR(bytes.NewReader(*recBytes)); err != nil { + return nil, fmt.Errorf("unmarshal CBOR: %w", err) + } + + 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 + Collection string + Rkey string +} + +// parseATURI parses an AT-URI into its components +// Format: at://did:plc:xxx/collection/rkey +func parseATURI(uri string) *atURIParts { + if !strings.HasPrefix(uri, "at://") { + return nil + } + + // Remove at:// prefix + path := strings.TrimPrefix(uri, "at://") + + // Split by / + parts := strings.SplitN(path, "/", 3) + if len(parts) != 3 { + return nil + } + + return &atURIParts{ + DID: parts[0], + Collection: parts[1], + Rkey: parts[2], + } +} + +// 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 { + // TIDs are base32-sortable timestamps + // Use indigo's syntax package for proper parsing + t, err := syntax.ParseTID(tid) + if err != nil { + // Return zero time - will be older than grace period + return time.Time{} + } + return t.Time() +} + +// extractDigestFromPath extracts a digest from a storage path +// Path format: /docker/registry/v2/blobs/{algorithm}/{xx}/{hash}/data +// Returns: {algorithm}:{hash} +func extractDigestFromPath(path string) string { + // Match pattern: /blobs/{alg}/{xx}/{hash}/data + re := regexp.MustCompile(`/blobs/([^/]+)/[^/]+/([^/]+)/data$`) + matches := re.FindStringSubmatch(path) + if len(matches) != 3 { + return "" + } + return matches[1] + ":" + matches[2] +} + +// logResult logs the GC result in a structured format +func (gc *GarbageCollector) logResult(result *GCResult) { + gc.logger.Info("GC run complete", + "duration", result.Duration, + "referencedBlobs", result.ReferencedBlobs, + "orphanedRecords", result.OrphanedRecords, + "recordsDeleted", result.RecordsDeleted, + "orphanedBlobs", result.OrphanedBlobs, + "blobsDeleted", result.BlobsDeleted, + "bytesReclaimed", result.BytesReclaimed, + "dryRun", gc.cfg.DryRun) + + // Also log as JSON for easier parsing + resultJSON, _ := json.Marshal(result) + gc.logger.Debug("GC result JSON", "result", string(resultJSON)) +} diff --git a/pkg/hold/gc/gc_test.go b/pkg/hold/gc/gc_test.go new file mode 100644 index 0000000..082c7eb --- /dev/null +++ b/pkg/hold/gc/gc_test.go @@ -0,0 +1,228 @@ +package gc + +import ( + "testing" + "time" +) + +func TestExtractDigestFromPath(t *testing.T) { + tests := []struct { + name string + path string + expected string + }{ + { + name: "valid sha256 path", + path: "/docker/registry/v2/blobs/sha256/ab/abc123def456/data", + expected: "sha256:abc123def456", + }, + { + name: "valid sha256 path with full hash", + path: "/docker/registry/v2/blobs/sha256/e3/e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855/data", + expected: "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", + }, + { + name: "invalid path - no data suffix", + path: "/docker/registry/v2/blobs/sha256/ab/abc123def456", + expected: "", + }, + { + name: "invalid path - wrong structure", + path: "/some/other/path/data", + expected: "", + }, + { + name: "empty path", + path: "", + expected: "", + }, + { + name: "uploads temp path (should not match)", + path: "/docker/registry/v2/uploads/temp-uuid/data", + expected: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := extractDigestFromPath(tt.path) + if result != tt.expected { + t.Errorf("extractDigestFromPath(%q) = %q, want %q", tt.path, result, tt.expected) + } + }) + } +} + +func TestParseATURI(t *testing.T) { + tests := []struct { + name string + uri string + expectNil bool + did string + collection string + rkey string + }{ + { + name: "valid AT-URI", + uri: "at://did:plc:abc123/io.atcr.manifest/xyz789", + expectNil: false, + did: "did:plc:abc123", + collection: "io.atcr.manifest", + rkey: "xyz789", + }, + { + name: "valid AT-URI with did:web", + uri: "at://did:web:example.com/io.atcr.manifest/manifest123", + expectNil: false, + did: "did:web:example.com", + collection: "io.atcr.manifest", + rkey: "manifest123", + }, + { + name: "invalid - no at:// prefix", + uri: "did:plc:abc123/io.atcr.manifest/xyz789", + expectNil: true, + }, + { + name: "invalid - missing rkey", + uri: "at://did:plc:abc123/io.atcr.manifest", + expectNil: true, + }, + { + name: "invalid - empty string", + uri: "", + expectNil: true, + }, + { + name: "invalid - http URL", + uri: "https://example.com/xrpc/com.atproto.repo.getRecord", + expectNil: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := parseATURI(tt.uri) + if tt.expectNil { + if result != nil { + t.Errorf("parseATURI(%q) = %+v, want nil", tt.uri, result) + } + return + } + + if result == nil { + t.Errorf("parseATURI(%q) = nil, want non-nil", tt.uri) + return + } + + if result.DID != tt.did { + t.Errorf("parseATURI(%q).DID = %q, want %q", tt.uri, result.DID, tt.did) + } + if result.Collection != tt.collection { + t.Errorf("parseATURI(%q).Collection = %q, want %q", tt.uri, result.Collection, tt.collection) + } + if result.Rkey != tt.rkey { + t.Errorf("parseATURI(%q).Rkey = %q, want %q", tt.uri, result.Rkey, tt.rkey) + } + }) + } +} + +func TestTidToTime(t *testing.T) { + // Test with known TID format + // TIDs are base32-encoded timestamps with counter + tests := []struct { + name string + tid string + expectZero bool + minAge time.Duration // Minimum expected age (roughly) + }{ + { + name: "valid TID from 2024", + tid: "3l7nqy25tks2c", // A real TID from around 2024 + expectZero: false, + }, + { + name: "invalid TID - too short", + tid: "abc", + expectZero: true, + }, + { + name: "invalid TID - empty", + tid: "", + expectZero: true, + }, + { + name: "invalid TID - not base32", + tid: "!!!!!!!!!!!!!!", + expectZero: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := tidToTime(tt.tid) + if tt.expectZero { + if !result.IsZero() { + t.Errorf("tidToTime(%q) = %v, want zero time", tt.tid, result) + } + return + } + + if result.IsZero() { + t.Errorf("tidToTime(%q) = zero time, want non-zero", tt.tid) + } + }) + } +} + +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() + + 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.Run("enabled via env", func(t *testing.T) { + t.Setenv("GC_ENABLED", "true") + t.Setenv("GC_DRY_RUN", "true") + + cfg := LoadConfigFromEnv() + + 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") + } + }) +} diff --git a/pkg/hold/pds/layer.go b/pkg/hold/pds/layer.go index 9edd5f2..60e6819 100644 --- a/pkg/hold/pds/layer.go +++ b/pkg/hold/pds/layer.go @@ -49,6 +49,25 @@ func (p *HoldPDS) GetLayerRecord(ctx context.Context, rkey string) (*atproto.Lay return nil, fmt.Errorf("GetLayerRecord not yet implemented - use via XRPC listRecords instead") } +// DeleteLayerRecord deletes a layer record by rkey +// This deletes from both the repo (MST) and the records index +func (p *HoldPDS) DeleteLayerRecord(ctx context.Context, rkey string) error { + // Delete from repo (MST) + if err := p.repomgr.DeleteRecord(ctx, p.uid, atproto.LayerCollection, rkey); err != nil { + return fmt.Errorf("failed to delete from repo: %w", err) + } + + // Delete from index + if p.recordsIndex != nil { + if err := p.recordsIndex.DeleteRecord(atproto.LayerCollection, rkey); err != nil { + // Log but don't fail - index will resync on backfill + fmt.Printf("Warning: failed to delete from records index: %v\n", err) + } + } + + return nil +} + // ListLayerRecords lists layer records with pagination // Returns records, next cursor (empty if no more), and error // Note: This is a simplified implementation. For production, consider adding filters diff --git a/pkg/hold/pds/server.go b/pkg/hold/pds/server.go index 5cc6fd4..31414d7 100644 --- a/pkg/hold/pds/server.go +++ b/pkg/hold/pds/server.go @@ -152,6 +152,36 @@ func (p *HoldPDS) UID() models.Uid { return p.uid } +// GetRecordBytes retrieves raw CBOR bytes for a record +// recordPath format: "collection/rkey" +func (p *HoldPDS) GetRecordBytes(ctx context.Context, recordPath string) (cid.Cid, *[]byte, error) { + session, err := p.carstore.ReadOnlySession(p.uid) + if err != nil { + return cid.Undef, nil, fmt.Errorf("failed to create session: %w", err) + } + + head, err := p.carstore.GetUserRepoHead(ctx, p.uid) + if err != nil { + return cid.Undef, nil, fmt.Errorf("failed to get repo head: %w", err) + } + + if !head.Defined() { + return cid.Undef, nil, fmt.Errorf("repo is empty") + } + + repoHandle, err := repo.OpenRepo(ctx, session, head) + if err != nil { + return cid.Undef, nil, fmt.Errorf("failed to open repo: %w", err) + } + + recordCID, recBytes, err := repoHandle.GetRecordBytes(ctx, recordPath) + if err != nil { + return cid.Undef, nil, fmt.Errorf("failed to get record: %w", err) + } + + return recordCID, recBytes, nil +} + // Bootstrap initializes the hold with the captain record, owner as first crew member, and profile func (p *HoldPDS) Bootstrap(ctx context.Context, storageDriver driver.StorageDriver, ownerDID string, public bool, allowAllCrew bool, avatarURL, region string) error { if ownerDID == "" {