From 53e196a261063381e72b693363dc37ac7a01bbe8 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Sun, 4 Jan 2026 15:53:44 -0600 Subject: [PATCH] start researching quotas based on layer size per DID --- docs/QUOTAS.md | 1477 ++++------------- lexicons/io/atcr/hold/layer.json | 13 +- pkg/appview/handlers/storage.go | 129 ++ pkg/appview/routes/routes.go | 5 + pkg/appview/storage/manifest_store.go | 9 +- pkg/appview/templates/pages/settings.html | 59 + .../templates/partials/storage_stats.html | 12 + pkg/atproto/cbor_gen.go | 106 +- pkg/atproto/endpoints.go | 6 + pkg/atproto/lexicon.go | 33 +- pkg/atproto/lexicon_test.go | 85 +- pkg/hold/oci/xrpc.go | 27 +- pkg/hold/pds/layer.go | 103 ++ pkg/hold/pds/layer_test.go | 135 +- pkg/hold/pds/xrpc.go | 31 + 15 files changed, 886 insertions(+), 1344 deletions(-) create mode 100644 pkg/appview/handlers/storage.go create mode 100644 pkg/appview/templates/partials/storage_stats.html diff --git a/docs/QUOTAS.md b/docs/QUOTAS.md index d30effc..0e1bdc1 100644 --- a/docs/QUOTAS.md +++ b/docs/QUOTAS.md @@ -1,994 +1,470 @@ # ATCR Quota System -This document describes ATCR's storage quota implementation, inspired by Harbor's proven approach to per-project blob tracking with deduplication. +This document describes ATCR's storage quota implementation using ATProto records for per-user layer tracking. ## Table of Contents - [Overview](#overview) -- [Harbor's Approach (Reference Implementation)](#harbors-approach-reference-implementation) -- [Storage Options](#storage-options) -- [Quota Data Model](#quota-data-model) -- [Push Flow (Detailed)](#push-flow-detailed) +- [Quota Model](#quota-model) +- [Layer Record Schema](#layer-record-schema) +- [Quota Calculation](#quota-calculation) +- [Push Flow](#push-flow) - [Delete Flow](#delete-flow) - [Garbage Collection](#garbage-collection) -- [Quota Reconciliation](#quota-reconciliation) - [Configuration](#configuration) -- [Trade-offs & Design Decisions](#trade-offs--design-decisions) - [Future Enhancements](#future-enhancements) ## Overview ATCR implements per-user storage quotas to: 1. **Limit storage consumption** on shared hold services -2. **Track actual S3 costs** (what new data was added) -3. **Benefit from deduplication** (users only pay once per layer) -4. **Provide transparency** (show users their storage usage) +2. **Provide transparency** (show users their storage usage) +3. **Enable fair billing** (users pay for what they use) -**Key principle:** Users pay for layers they've uploaded, but only ONCE per layer regardless of how many images reference it. +**Key principle:** Users pay for layers they reference, deduplicated per-user. If you push the same layer in multiple images, you only pay once. ### Example Scenario ``` Alice pushes myapp:v1 (layers A, B, C - each 100MB) -→ Alice's quota: +300MB (all new layers) +→ Creates 3 layer records in hold's PDS +→ Alice's quota: 300MB (3 unique layers) Alice pushes myapp:v2 (layers A, B, D) -→ Layers A, B already claimed by Alice -→ Layer D is new (100MB) -→ Alice's quota: +100MB (only D is new) -→ Total: 400MB +→ Creates 3 more layer records (A, B again, plus D) +→ Alice's quota: 400MB (4 unique layers: A, B, C, D) +→ Layers A, B appear twice in records but deduplicated in quota calc Bob pushes his-app:latest (layers A, E) -→ Layer A already exists in S3 (uploaded by Alice) -→ Bob claims it for first time → +100MB to Bob's quota -→ Layer E is new → +100MB to Bob's quota -→ Bob's quota: 200MB +→ Creates 2 layer records for Bob +→ Bob's quota: 200MB (2 unique layers: A, E) +→ Layer A shared with Alice in S3, but Bob pays for his own usage -Physical S3 storage: 500MB (A, B, C, D, E) -Claimed storage: 600MB (Alice: 400MB, Bob: 200MB) -Deduplication savings: 100MB (layer A shared) +Physical S3 storage: 500MB (A, B, C, D, E - deduplicated globally) +Alice's quota: 400MB +Bob's quota: 200MB ``` -## Harbor's Approach (Reference Implementation) +## Quota Model -Harbor is built on distribution/distribution (same as ATCR) and implements quotas as middleware. Their approach: +### Everyone Pays for What They Upload -### Key Insights from Harbor +Each user is charged for all unique layers they reference, regardless of whether those layers exist in S3 from other users' uploads. -1. **"Shared blobs are only computed once per project"** - - Each project tracks which blobs it has uploaded - - Same blob used in multiple images counts only once per project - - Different projects claiming the same blob each pay for it +**Why this model?** +- **Simple mental model**: "I pushed 500MB of layers, I use 500MB of quota" +- **Predictable**: Your quota doesn't change based on others' actions +- **Clean deletion**: Delete manifest → layer records removed → quota freed +- **No cross-user dependencies**: Users are isolated -2. **Quota checked when manifest is pushed** - - Blobs upload first (presigned URLs, can't intercept) - - Manifest pushed last → quota check happens here - - Can reject manifest if quota exceeded (orphaned blobs cleaned by GC) +**Trade-off:** +- Total claimed storage can exceed physical S3 storage +- This is acceptable - deduplication is an operational benefit for ATCR, not a billing feature -3. **Middleware-based implementation** - - distribution/distribution has NO built-in quota support - - Harbor added it as request preprocessing middleware - - Uses database (PostgreSQL) or Redis for quota storage +### ATProto-Native Storage -4. **Per-project ownership model** - - Blobs are physically deduplicated globally - - Quota accounting is logical (per-project claims) - - Total claimed storage can exceed physical storage +Layer tracking uses ATProto records stored in the hold's embedded PDS: +- **Collection**: `io.atcr.hold.layer` +- **Repository**: Hold's DID (e.g., `did:web:hold01.atcr.io`) +- **Records**: One per manifest-layer relationship (TID-based keys) -### References +This approach: +- Keeps quota data in ATProto (no separate database) +- Enables standard ATProto sync/query mechanisms +- Provides full audit trail of layer usage -- Harbor Quota Documentation: https://goharbor.io/docs/1.10/administration/configure-project-quotas/ -- Harbor Source: https://github.com/goharbor/harbor (see `src/controller/quota`) +## Layer Record Schema -## Storage Options +### LayerRecord -The hold service needs to store quota data somewhere. Two options: +```go +// pkg/atproto/lexicon.go -### Option 1: S3-Based Storage (Recommended for BYOS) - -Store quota metadata alongside blobs in the same S3 bucket: - -``` -Bucket structure: -/docker/registry/v2/blobs/sha256/ab/abc123.../data ← actual blobs -/atcr/quota/did:plc:alice.json ← quota tracking -/atcr/quota/did:plc:bob.json -``` - -**Pros:** -- ✅ No separate database needed -- ✅ Single S3 bucket (better UX - no second bucket to configure) -- ✅ Quota data lives with the blobs -- ✅ Hold service stays relatively stateless -- ✅ Works with any S3-compatible service (Storj, Minio, Upcloud, Fly.io) - -**Cons:** -- ❌ Slower than local database (network round-trip) -- ❌ Eventual consistency issues -- ❌ Race conditions on concurrent updates -- ❌ Extra S3 API costs (GET/PUT per upload) - -**Performance:** -- Each blob upload: 1 HEAD (blob exists?) + 1 GET (quota) + 1 PUT (update quota) -- Typical latency: 100-200ms total overhead -- For high-throughput registries, consider SQLite - -### Option 2: SQLite Database (Recommended for Shared Holds) - -Local database in hold service: - -```bash -/var/lib/atcr/hold-quota.db -``` - -**Pros:** -- ✅ Fast local queries (no network latency) -- ✅ ACID transactions (no race conditions) -- ✅ Efficient for high-throughput registries -- ✅ Can use foreign keys and joins - -**Cons:** -- ❌ Makes hold service stateful (persistent volume needed) -- ❌ Not ideal for ephemeral BYOS deployments -- ❌ Backup/restore complexity -- ❌ Multi-instance scaling requires shared database - -**Schema:** -```sql -CREATE TABLE user_quotas ( - did TEXT PRIMARY KEY, - quota_limit INTEGER NOT NULL DEFAULT 10737418240, -- 10GB - quota_used INTEGER NOT NULL DEFAULT 0, - updated_at TIMESTAMP -); - -CREATE TABLE claimed_layers ( - did TEXT NOT NULL, - digest TEXT NOT NULL, - size INTEGER NOT NULL, - claimed_at TIMESTAMP, - PRIMARY KEY(did, digest) -); -``` - -### Recommendation - -- **BYOS (user-owned holds):** S3-based (keeps hold service ephemeral) -- **Shared holds (multi-user):** SQLite (better performance and consistency) -- **High-traffic production:** SQLite or PostgreSQL (Harbor uses this) - -## Quota Data Model - -### Quota File Format (S3-based) - -```json -{ - "did": "did:plc:alice123", - "limit": 10737418240, - "used": 5368709120, - "claimed_layers": { - "sha256:abc123...": 104857600, - "sha256:def456...": 52428800, - "sha256:789ghi...": 209715200 - }, - "last_updated": "2025-10-09T12:34:56Z", - "version": 1 +type LayerRecord struct { + Type string `json:"$type"` // "io.atcr.hold.layer" + Digest string `json:"digest"` // Layer digest (sha256:abc123...) + Size int64 `json:"size"` // Size in bytes + MediaType string `json:"mediaType"` // e.g., "application/vnd.oci.image.layer.v1.tar+gzip" + Manifest string `json:"manifest"` // at://did:plc:alice/io.atcr.manifest/abc123 + UserDID string `json:"userDid"` // User's DID for quota grouping + CreatedAt string `json:"createdAt"` // ISO 8601 timestamp } ``` -**Fields:** -- `did`: User's ATProto DID -- `limit`: Maximum storage in bytes (default: 10GB) -- `used`: Current storage usage in bytes (sum of claimed_layers) -- `claimed_layers`: Map of digest → size for all layers user has uploaded -- `last_updated`: Timestamp of last quota update -- `version`: Schema version for future migrations +### Record Key -### Why Track Individual Layers? +Records use TID (timestamp-based ID) as the rkey. This means: +- Multiple records can exist for the same layer (from different manifests) +- Deduplication happens at query time, not storage time +- Simple append-only writes on manifest push -**Q: Can't we just track a counter?** +### Example Records -**A: We need layer tracking for:** +``` +Manifest A (layers X, Y, Z) → creates 3 records +Manifest B (layers X, W) → creates 2 records -1. **Deduplication detection** - - Check if user already claimed a layer → free upload - - Example: Updating an image reuses most layers +io.atcr.hold.layer collection: +┌──────────────┬────────┬──────┬───────────────────────────────────┬─────────────────┐ +│ rkey (TID) │ digest │ size │ manifest │ userDid │ +├──────────────┼────────┼──────┼───────────────────────────────────┼─────────────────┤ +│ 3jui7...001 │ X │ 100 │ at://did:plc:alice/.../manifestA │ did:plc:alice │ +│ 3jui7...002 │ Y │ 200 │ at://did:plc:alice/.../manifestA │ did:plc:alice │ +│ 3jui7...003 │ Z │ 150 │ at://did:plc:alice/.../manifestA │ did:plc:alice │ +│ 3jui7...004 │ X │ 100 │ at://did:plc:alice/.../manifestB │ did:plc:alice │ ← duplicate digest +│ 3jui7...005 │ W │ 300 │ at://did:plc:alice/.../manifestB │ did:plc:alice │ +└──────────────┴────────┴──────┴───────────────────────────────────┴─────────────────┘ +``` -2. **Accurate deletes** - - When manifest deleted, only decrement unclaimed layers - - User may have 5 images sharing layer A - deleting 1 image doesn't free layer A +## Quota Calculation -3. **Quota reconciliation** - - Verify quota matches reality by listing user's manifests - - Recalculate from layers in manifests vs claimed_layers map +### Query: User's Unique Storage -4. **Auditing** - - "Show me what I'm storing" - - Users can see which layers consume their quota +```sql +-- Calculate quota by deduplicating layers +SELECT SUM(size) FROM ( + SELECT DISTINCT digest, size + FROM io.atcr.hold.layer + WHERE userDid = ? +) +``` -## Push Flow (Detailed) +Using the example above: +- Layer X appears twice but counted once: 100 +- Layers Y, Z, W counted once each: 200 + 150 + 300 +- **Total: 750 bytes** + +### Implementation + +```go +// pkg/hold/quota/quota.go + +type QuotaManager struct { + pds *pds.Server // Hold's embedded PDS +} + +// GetUsage calculates a user's current quota usage +func (q *QuotaManager) GetUsage(ctx context.Context, userDID string) (int64, error) { + // List all layer records for this user + records, err := q.pds.ListRecords(ctx, LayerCollection, userDID) + if err != nil { + return 0, err + } + + // Deduplicate by digest + uniqueLayers := make(map[string]int64) // digest -> size + for _, record := range records { + var layer LayerRecord + if err := json.Unmarshal(record.Value, &layer); err != nil { + continue + } + if layer.UserDID == userDID { + uniqueLayers[layer.Digest] = layer.Size + } + } + + // Sum unique layer sizes + var total int64 + for _, size := range uniqueLayers { + total += size + } + + return total, nil +} + +// CheckQuota returns true if user has space for additional bytes +func (q *QuotaManager) CheckQuota(ctx context.Context, userDID string, additional int64, limit int64) (bool, int64, error) { + current, err := q.GetUsage(ctx, userDID) + if err != nil { + return false, 0, err + } + + return current+additional <= limit, current, nil +} +``` + +### Quota Response + +```go +type QuotaInfo struct { + Used int64 `json:"used"` // Current usage (deduplicated) + Limit int64 `json:"limit"` // User's quota limit + Available int64 `json:"available"` // Remaining space +} +``` + +## Push Flow ### Step-by-Step: User Pushes Image ``` -┌──────────┐ ┌──────────┐ ┌──────────┐ -│ Client │ │ Hold │ │ S3 │ -│ (Docker) │ │ Service │ │ Bucket │ -└──────────┘ └──────────┘ └──────────┘ - │ │ │ - │ 1. PUT /v2/.../blobs/ │ │ - │ upload?digest=sha256:abc│ │ - ├───────────────────────────>│ │ - │ │ │ - │ │ 2. Check if blob exists │ - │ │ (Stat/HEAD request) │ - │ ├───────────────────────────>│ - │ │<───────────────────────────┤ - │ │ 200 OK (exists) or │ - │ │ 404 Not Found │ - │ │ │ - │ │ 3. Read user quota │ - │ │ GET /atcr/quota/{did} │ - │ ├───────────────────────────>│ - │ │<───────────────────────────┤ - │ │ quota.json │ - │ │ │ - │ │ 4. Calculate quota impact │ - │ │ - If digest in │ - │ │ claimed_layers: 0 │ - │ │ - Else: size │ - │ │ │ - │ │ 5. Check quota limit │ - │ │ used + impact <= limit? │ - │ │ │ - │ │ 6. Update quota │ - │ │ PUT /atcr/quota/{did} │ - │ ├───────────────────────────>│ - │ │<───────────────────────────┤ - │ │ 200 OK │ - │ │ │ - │ 7. Presigned URL │ │ - │<───────────────────────────┤ │ - │ {url: "https://s3..."} │ │ - │ │ │ - │ 8. Upload blob to S3 │ │ - ├────────────────────────────┼───────────────────────────>│ - │ │ │ - │ 9. 200 OK │ │ - │<───────────────────────────┼────────────────────────────┤ - │ │ │ +┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ +│ Client │ │ AppView │ │ Hold │ │ User PDS │ +│ (Docker) │ │ │ │ Service │ │ │ +└──────────┘ └──────────┘ └──────────┘ └──────────┘ + │ │ │ │ + │ 1. Upload blobs │ │ │ + ├─────────────────────>│ │ │ + │ │ 2. Route to hold │ │ + │ ├─────────────────────>│ │ + │ │ │ 3. Store in S3 │ + │ │ │ │ + │ 4. PUT manifest │ │ │ + ├─────────────────────>│ │ │ + │ │ │ │ + │ │ 5. Calculate quota │ │ + │ │ impact for new │ │ + │ │ layers │ │ + │ │ │ │ + │ │ 6. Check quota limit │ │ + │ ├─────────────────────>│ │ + │ │<─────────────────────┤ │ + │ │ │ │ + │ │ 7. Store manifest │ │ + │ ├──────────────────────┼─────────────────────>│ + │ │ │ │ + │ │ 8. Create layer │ │ + │ │ records │ │ + │ ├─────────────────────>│ │ + │ │ │ 9. Write to │ + │ │ │ hold's PDS │ + │ │ │ │ + │ 10. 201 Created │ │ │ + │<─────────────────────┤ │ │ ``` -### Implementation (Pseudocode) +### Implementation ```go -// cmd/hold/main.go - HandlePutPresignedURL +// pkg/appview/storage/routing_repository.go -func (s *HoldService) HandlePutPresignedURL(w http.ResponseWriter, r *http.Request) { - var req PutPresignedURLRequest - json.NewDecoder(r.Body).Decode(&req) +func (r *RoutingRepository) PutManifest(ctx context.Context, manifest distribution.Manifest) error { + // Parse manifest to get layers + layers := extractLayers(manifest) - // Step 1: Check if blob already exists in S3 - blobPath := fmt.Sprintf("/docker/registry/v2/blobs/%s/%s/%s/data", - algorithm, digest[:2], digest) - - _, err := s.driver.Stat(ctx, blobPath) - blobExists := (err == nil) - - // Step 2: Read quota from S3 (or SQLite) - quota, err := s.quotaManager.GetQuota(req.DID) + // Get user's current unique layers from hold + existingLayers, err := r.holdClient.GetUserLayers(ctx, r.userDID) if err != nil { - // First upload - create quota with defaults - quota = &Quota{ - DID: req.DID, - Limit: s.config.QuotaDefaultLimit, - Used: 0, - ClaimedLayers: make(map[string]int64), + return err + } + existingSet := makeDigestSet(existingLayers) + + // Calculate quota impact (only new unique layers) + var quotaImpact int64 + for _, layer := range layers { + if !existingSet[layer.Digest] { + quotaImpact += layer.Size } } - // Step 3: Calculate quota impact - quotaImpact := req.Size // Default: assume new layer - - if _, alreadyClaimed := quota.ClaimedLayers[req.Digest]; alreadyClaimed { - // User already uploaded this layer before - quotaImpact = 0 - log.Printf("Layer %s already claimed by %s, no quota impact", - req.Digest, req.DID) - } else if blobExists { - // Blob exists in S3 (uploaded by another user) - // But this user is claiming it for first time - // Still counts against their quota - log.Printf("Layer %s exists globally but new to %s, quota impact: %d", - req.Digest, req.DID, quotaImpact) - } else { - // Brand new blob - will be uploaded to S3 - log.Printf("New layer %s for %s, quota impact: %d", - req.Digest, req.DID, quotaImpact) - } - - // Step 4: Check quota limit - if quota.Used + quotaImpact > quota.Limit { - http.Error(w, fmt.Sprintf( - "quota exceeded: used=%d, impact=%d, limit=%d", - quota.Used, quotaImpact, quota.Limit, - ), http.StatusPaymentRequired) // 402 - return - } - - // Step 5: Update quota (optimistic - before upload completes) - quota.Used += quotaImpact - if quotaImpact > 0 { - quota.ClaimedLayers[req.Digest] = req.Size - } - quota.LastUpdated = time.Now() - - if err := s.quotaManager.SaveQuota(quota); err != nil { - http.Error(w, "failed to update quota", http.StatusInternalServerError) - return - } - - // Step 6: Generate presigned URL - presignedURL, err := s.getUploadURL(ctx, req.Digest, req.Size, req.DID) + // Check quota + ok, current, err := r.quotaManager.CheckQuota(ctx, r.userDID, quotaImpact, r.quotaLimit) if err != nil { - // Rollback quota update on error - quota.Used -= quotaImpact - delete(quota.ClaimedLayers, req.Digest) - s.quotaManager.SaveQuota(quota) - - http.Error(w, "failed to generate presigned URL", http.StatusInternalServerError) - return + return err + } + if !ok { + return fmt.Errorf("quota exceeded: used=%d, impact=%d, limit=%d", + current, quotaImpact, r.quotaLimit) } - // Step 7: Return presigned URL + quota info - resp := PutPresignedURLResponse{ - URL: presignedURL, - ExpiresAt: time.Now().Add(15 * time.Minute), - QuotaInfo: QuotaInfo{ - Used: quota.Used, - Limit: quota.Limit, - Available: quota.Limit - quota.Used, - Impact: quotaImpact, - AlreadyClaimed: quotaImpact == 0, - }, + // Store manifest in user's PDS + manifestURI, err := r.atprotoClient.PutManifest(ctx, manifest) + if err != nil { + return err } - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(resp) + // Create layer records in hold's PDS + for _, layer := range layers { + record := LayerRecord{ + Type: "io.atcr.hold.layer", + Digest: layer.Digest, + Size: layer.Size, + MediaType: layer.MediaType, + Manifest: manifestURI, + UserDID: r.userDID, + CreatedAt: time.Now().Format(time.RFC3339), + } + if err := r.holdClient.CreateLayerRecord(ctx, record); err != nil { + log.Printf("Warning: failed to create layer record: %v", err) + // Continue - reconciliation will fix + } + } + + return nil } ``` -### Race Condition Handling +### Quota Check Timing -**Problem:** Two concurrent uploads of the same blob +Quota is checked when the **manifest is pushed** (after blobs are uploaded): +- Blobs upload first via presigned URLs +- Manifest pushed last triggers quota check +- If quota exceeded, manifest is rejected (orphaned blobs cleaned by GC) -``` -Time User A User B -0ms Upload layer X (100MB) -10ms Upload layer X (100MB) -20ms Check exists: NO Check exists: NO -30ms Quota impact: 100MB Quota impact: 100MB -40ms Update quota A: +100MB Update quota B: +100MB -50ms Generate presigned URL Generate presigned URL -100ms Upload to S3 completes Upload to S3 (overwrites A's) -``` - -**Result:** Both users charged 100MB, but only 100MB stored in S3. - -**Mitigation strategies:** - -1. **Accept eventual consistency** (recommended for S3-based) - - Run periodic reconciliation to fix discrepancies - - Small inconsistency window (minutes) is acceptable - - Reconciliation uses PDS as source of truth - -2. **Optimistic locking** (S3 ETags) - ```go - // Use S3 ETags for conditional writes - oldETag := getQuotaFileETag(did) - err := putQuotaFileWithCondition(quota, oldETag) - if err == PreconditionFailed { - // Retry with fresh read - } - ``` - -3. **Database transactions** (SQLite-based) - ```sql - BEGIN TRANSACTION; - SELECT * FROM user_quotas WHERE did = ? FOR UPDATE; - UPDATE user_quotas SET used = used + ? WHERE did = ?; - COMMIT; - ``` +This matches Harbor's approach and is the industry standard. ## Delete Flow -### Manifest Deletion via AppView UI +### Manifest Deletion -When a user deletes a manifest through the AppView web interface: +When a user deletes a manifest: ``` ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ -│ User │ │ AppView │ │ Hold │ │ PDS │ -│ UI │ │ Database │ │ Service │ │ │ +│ User │ │ AppView │ │ Hold │ │ User PDS │ +│ UI │ │ │ │ Service │ │ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │ │ │ │ DELETE manifest │ │ │ ├─────────────────────>│ │ │ │ │ │ │ - │ │ 1. Get manifest │ │ - │ │ and layers │ │ - │ │ │ │ - │ │ 2. Check which │ │ - │ │ layers still │ │ - │ │ referenced by │ │ - │ │ user's other │ │ - │ │ manifests │ │ - │ │ │ │ - │ │ 3. DELETE manifest │ │ - │ │ from PDS │ │ + │ │ 1. Delete manifest │ │ + │ │ from user's PDS │ │ │ ├──────────────────────┼─────────────────────>│ │ │ │ │ - │ │ 4. POST /quota/decrement │ + │ │ 2. Delete layer │ │ + │ │ records for this │ │ + │ │ manifest │ │ │ ├─────────────────────>│ │ - │ │ {layers: [...]} │ │ + │ │ │ 3. Remove records │ + │ │ │ where manifest │ + │ │ │ == deleted URI │ │ │ │ │ - │ │ │ 5. Update quota │ - │ │ │ Remove unclaimed │ - │ │ │ layers │ - │ │ │ │ - │ │ 6. 200 OK │ │ - │ │<─────────────────────┤ │ - │ │ │ │ - │ │ 7. Delete from DB │ │ - │ │ │ │ - │ 8. Success │ │ │ + │ 4. 204 No Content │ │ │ │<─────────────────────┤ │ │ - │ │ │ │ ``` -### AppView Implementation +### Implementation ```go // pkg/appview/handlers/manifest.go func (h *ManifestHandler) DeleteManifest(w http.ResponseWriter, r *http.Request) { - did := r.Context().Value("auth.did").(string) + userDID := auth.GetDID(r.Context()) repository := chi.URLParam(r, "repository") digest := chi.URLParam(r, "digest") - // Step 1: Get manifest and its layers from database - manifest, err := db.GetManifest(h.db, digest) - if err != nil { - http.Error(w, "manifest not found", 404) + // Get manifest URI before deletion + manifestURI := fmt.Sprintf("at://%s/%s/%s", userDID, ManifestCollection, digest) + + // Delete manifest from user's PDS + if err := h.atprotoClient.DeleteRecord(ctx, ManifestCollection, digest); err != nil { + http.Error(w, "failed to delete manifest", 500) return } - layers, err := db.GetLayersForManifest(h.db, manifest.ID) - if err != nil { - http.Error(w, "failed to get layers", 500) - return - } - - // Step 2: For each layer, check if user still references it - // in other manifests - layersToDecrement := []LayerInfo{} - - for _, layer := range layers { - // Query: does this user have other manifests using this layer? - stillReferenced, err := db.CheckLayerReferencedByUser( - h.db, did, repository, layer.Digest, manifest.ID, - ) - - if err != nil { - http.Error(w, "failed to check layer references", 500) - return - } - - if !stillReferenced { - // This layer is no longer used by user - layersToDecrement = append(layersToDecrement, LayerInfo{ - Digest: layer.Digest, - Size: layer.Size, - }) - } - } - - // Step 3: Delete manifest from user's PDS - atprotoClient := atproto.NewClient(manifest.PDSEndpoint, did, accessToken) - err = atprotoClient.DeleteRecord(ctx, atproto.ManifestCollection, manifestRKey) - if err != nil { - http.Error(w, "failed to delete from PDS", 500) - return - } - - // Step 4: Notify hold service to decrement quota - if len(layersToDecrement) > 0 { - holdClient := &http.Client{} - - decrementReq := QuotaDecrementRequest{ - DID: did, - Layers: layersToDecrement, - } - - body, _ := json.Marshal(decrementReq) - resp, err := holdClient.Post( - manifest.HoldEndpoint + "/quota/decrement", - "application/json", - bytes.NewReader(body), - ) - - if err != nil || resp.StatusCode != 200 { - log.Printf("Warning: failed to update quota on hold service: %v", err) - // Continue anyway - GC reconciliation will fix it - } - } - - // Step 5: Delete from AppView database - err = db.DeleteManifest(h.db, did, repository, digest) - if err != nil { - http.Error(w, "failed to delete from database", 500) - return + // Delete associated layer records from hold's PDS + if err := h.holdClient.DeleteLayerRecords(ctx, manifestURI); err != nil { + log.Printf("Warning: failed to delete layer records: %v", err) + // Continue - reconciliation will clean up } w.WriteHeader(http.StatusNoContent) } ``` -### Hold Service Decrement Endpoint +### Hold Service: Delete Layer Records ```go -// cmd/hold/main.go +// pkg/hold/pds/xrpc.go -type QuotaDecrementRequest struct { - DID string `json:"did"` - Layers []LayerInfo `json:"layers"` -} - -type LayerInfo struct { - Digest string `json:"digest"` - Size int64 `json:"size"` -} - -func (s *HoldService) HandleQuotaDecrement(w http.ResponseWriter, r *http.Request) { - var req QuotaDecrementRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - http.Error(w, "invalid request", 400) - return - } - - // Read current quota - quota, err := s.quotaManager.GetQuota(req.DID) - if err != nil { - http.Error(w, "quota not found", 404) - return - } - - // Decrement quota for each layer - for _, layer := range req.Layers { - if size, claimed := quota.ClaimedLayers[layer.Digest]; claimed { - // Remove from claimed layers - delete(quota.ClaimedLayers, layer.Digest) - quota.Used -= size - - log.Printf("Decremented quota for %s: layer %s (%d bytes)", - req.DID, layer.Digest, size) - } else { - log.Printf("Warning: layer %s not in claimed_layers for %s", - layer.Digest, req.DID) - } - } - - // Ensure quota.Used doesn't go negative (defensive) - if quota.Used < 0 { - log.Printf("Warning: quota.Used went negative for %s, resetting to 0", req.DID) - quota.Used = 0 - } - - // Save updated quota - quota.LastUpdated = time.Now() - if err := s.quotaManager.SaveQuota(quota); err != nil { - http.Error(w, "failed to save quota", 500) - return - } - - // Return updated quota info - json.NewEncoder(w).Encode(map[string]any{ - "used": quota.Used, - "limit": quota.Limit, - }) -} -``` - -### SQL Query: Check Layer References - -```sql --- pkg/appview/db/queries.go - --- Check if user still references this layer in other manifests -SELECT COUNT(*) -FROM layers l -JOIN manifests m ON l.manifest_id = m.id -WHERE m.did = ? -- User's DID - AND l.digest = ? -- Layer digest - AND m.id != ? -- Exclude the manifest being deleted -``` - -## Garbage Collection - -### Background: Orphaned Blobs - -Orphaned blobs accumulate when: -1. Manifest push fails after blobs uploaded (presigned URLs bypass hold) -2. Quota exceeded - manifest rejected, blobs already in S3 -3. User deletes manifest - blobs no longer referenced - -**GC periodically cleans these up.** - -### GC Cron Implementation - -Similar to AppView's backfill worker, the hold service can run periodic GC: - -```go -// cmd/hold/gc/gc.go - -type GarbageCollector struct { - driver storagedriver.StorageDriver - appviewURL string - holdURL string - quotaManager *quota.Manager -} - -// Run garbage collection -func (gc *GarbageCollector) Run(ctx context.Context) error { - log.Println("Starting garbage collection...") - - // Step 1: Get list of referenced blobs from AppView - referenced, err := gc.getReferencedBlobs() - if err != nil { - return fmt.Errorf("failed to get referenced blobs: %w", err) - } - - referencedSet := make(map[string]bool) - for _, digest := range referenced { - referencedSet[digest] = true - } - - log.Printf("AppView reports %d referenced blobs", len(referenced)) - - // Step 2: Walk S3 blobs - deletedCount := 0 - reclaimedBytes := int64(0) - - err = gc.driver.Walk(ctx, "/docker/registry/v2/blobs", func(fileInfo storagedriver.FileInfo) error { - if fileInfo.IsDir() { - return nil // Skip directories - } - - // Extract digest from path - // Path: /docker/registry/v2/blobs/sha256/ab/abc123.../data - digest := extractDigestFromPath(fileInfo.Path()) - - if !referencedSet[digest] { - // Unreferenced blob - delete it - size := fileInfo.Size() - - if err := gc.driver.Delete(ctx, fileInfo.Path()); err != nil { - log.Printf("Failed to delete blob %s: %v", digest, err) - return nil // Continue anyway - } - - deletedCount++ - reclaimedBytes += size - - log.Printf("GC: Deleted unreferenced blob %s (%d bytes)", digest, size) - } - - return nil - }) - - if err != nil { - return fmt.Errorf("failed to walk blobs: %w", err) - } - - log.Printf("GC complete: deleted %d blobs, reclaimed %d bytes", - deletedCount, reclaimedBytes) - - return nil -} - -// Get referenced blobs from AppView -func (gc *GarbageCollector) getReferencedBlobs() ([]string, error) { - // Query AppView for all blobs referenced by manifests - // stored in THIS hold service - url := fmt.Sprintf("%s/internal/blobs/referenced?hold=%s", - gc.appviewURL, url.QueryEscape(gc.holdURL)) - - resp, err := http.Get(url) - if err != nil { - return nil, err - } - defer resp.Body.Close() - - var result struct { - Blobs []string `json:"blobs"` - } - - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { - return nil, err - } - - return result.Blobs, nil -} -``` - -### AppView Internal API - -```go -// pkg/appview/handlers/internal.go - -// Get all referenced blobs for a specific hold -func (h *InternalHandler) GetReferencedBlobs(w http.ResponseWriter, r *http.Request) { - holdEndpoint := r.URL.Query().Get("hold") - if holdEndpoint == "" { - http.Error(w, "missing hold parameter", 400) - return - } - - // Query database for all layers in manifests stored in this hold - query := ` - SELECT DISTINCT l.digest - FROM layers l - JOIN manifests m ON l.manifest_id = m.id - WHERE m.hold_endpoint = ? - ` - - rows, err := h.db.Query(query, holdEndpoint) - if err != nil { - http.Error(w, "database error", 500) - return - } - defer rows.Close() - - blobs := []string{} - for rows.Next() { - var digest string - if err := rows.Scan(&digest); err != nil { - continue - } - blobs = append(blobs, digest) - } - - json.NewEncoder(w).Encode(map[string]any{ - "blobs": blobs, - "count": len(blobs), - "hold": holdEndpoint, - }) -} -``` - -### GC Cron Schedule - -```go -// cmd/hold/main.go - -func main() { - // ... service setup ... - - // Start GC cron if enabled - if os.Getenv("GC_ENABLED") == "true" { - gcInterval := 24 * time.Hour // Daily by default - - go func() { - ticker := time.NewTicker(gcInterval) - defer ticker.Stop() - - for range ticker.C { - if err := garbageCollector.Run(context.Background()); err != nil { - log.Printf("GC error: %v", err) - } - } - }() - - log.Printf("GC cron started: runs every %v", gcInterval) - } - - // Start server... -} -``` - -## Quota Reconciliation - -### PDS as Source of Truth - -**Key insight:** Manifest records in PDS are publicly readable (no OAuth needed for reads). - -Each manifest contains: -- Repository name -- Digest -- Layers array with digest + size -- Hold endpoint - -The hold service can query the PDS to calculate the user's true quota: - -``` -1. List all io.atcr.manifest records for user -2. Filter manifests where holdEndpoint == this hold service -3. Extract unique layers (deduplicate by digest) -4. Sum layer sizes = true quota usage -5. Compare to quota file -6. Fix discrepancies -``` - -### Implementation - -```go -// cmd/hold/quota/reconcile.go - -type Reconciler struct { - quotaManager *Manager - atprotoResolver *atproto.Resolver - holdURL string -} - -// ReconcileUser recalculates quota from PDS manifests -func (r *Reconciler) ReconcileUser(ctx context.Context, did string) error { - log.Printf("Reconciling quota for %s", did) - - // Step 1: Resolve user's PDS endpoint - identity, err := r.atprotoResolver.ResolveIdentity(ctx, did) - if err != nil { - return fmt.Errorf("failed to resolve DID: %w", err) - } - - // Step 2: Create unauthenticated ATProto client - // (manifest records are public - no OAuth needed) - client := atproto.NewClient(identity.PDSEndpoint, did, "") - - // Step 3: List all manifest records for this user - manifests, err := client.ListRecords(ctx, atproto.ManifestCollection, 1000) - if err != nil { - return fmt.Errorf("failed to list manifests: %w", err) - } - - // Step 4: Filter manifests stored in THIS hold service - // and extract unique layers - uniqueLayers := make(map[string]int64) // digest -> size - - for _, record := range manifests { - var manifest atproto.ManifestRecord - if err := json.Unmarshal(record.Value, &manifest); err != nil { - log.Printf("Warning: failed to parse manifest: %v", err) - continue - } - - // Only count manifests stored in this hold - if manifest.HoldEndpoint != r.holdURL { - continue - } - - // Add config blob - if manifest.Config.Digest != "" { - uniqueLayers[manifest.Config.Digest] = manifest.Config.Size - } - - // Add layer blobs - for _, layer := range manifest.Layers { - uniqueLayers[layer.Digest] = layer.Size - } - } - - // Step 5: Calculate true quota usage - trueUsage := int64(0) - for _, size := range uniqueLayers { - trueUsage += size - } - - log.Printf("User %s true usage from PDS: %d bytes (%d unique layers)", - did, trueUsage, len(uniqueLayers)) - - // Step 6: Compare with current quota file - quota, err := r.quotaManager.GetQuota(did) - if err != nil { - log.Printf("No existing quota for %s, creating new", did) - quota = &Quota{ - DID: did, - Limit: r.quotaManager.DefaultLimit, - ClaimedLayers: make(map[string]int64), - } - } - - // Step 7: Fix discrepancies - if quota.Used != trueUsage || len(quota.ClaimedLayers) != len(uniqueLayers) { - log.Printf("Quota mismatch for %s: recorded=%d, actual=%d (diff=%d)", - did, quota.Used, trueUsage, trueUsage - quota.Used) - - // Update quota to match PDS truth - quota.Used = trueUsage - quota.ClaimedLayers = uniqueLayers - quota.LastUpdated = time.Now() - - if err := r.quotaManager.SaveQuota(quota); err != nil { - return fmt.Errorf("failed to save reconciled quota: %w", err) - } - - log.Printf("Reconciled quota for %s: %d bytes", did, trueUsage) - } else { - log.Printf("Quota for %s is accurate", did) - } - - return nil -} - -// ReconcileAll reconciles all users (run periodically) -func (r *Reconciler) ReconcileAll(ctx context.Context) error { - // Get list of all users with quota files - users, err := r.quotaManager.ListUsers() +func (s *Server) DeleteLayerRecords(ctx context.Context, manifestURI string) error { + // List all layer records + records, err := s.ListRecords(ctx, LayerCollection, "") if err != nil { return err } - log.Printf("Starting reconciliation for %d users", len(users)) - - for _, did := range users { - if err := r.ReconcileUser(ctx, did); err != nil { - log.Printf("Failed to reconcile %s: %v", did, err) - // Continue with other users + // Delete records matching this manifest + for _, record := range records { + var layer LayerRecord + if err := json.Unmarshal(record.Value, &layer); err != nil { + continue + } + if layer.Manifest == manifestURI { + if err := s.DeleteRecord(ctx, LayerCollection, record.RKey); err != nil { + log.Printf("Failed to delete layer record %s: %v", record.RKey, err) + } } } - log.Println("Reconciliation complete") return nil } ``` -### Reconciliation Cron +### Quota After Deletion + +After deleting a manifest: +- Layer records for that manifest are removed +- Quota recalculated with `SELECT DISTINCT` query +- If layer was only in deleted manifest → quota decreases +- If layer exists in other manifests → quota unchanged (still deduplicated) + +## Garbage Collection + +### Orphaned Blobs + +Orphaned blobs accumulate when: +1. Manifest push fails after blobs uploaded +2. Quota exceeded - manifest rejected +3. User deletes manifest - blobs may no longer be referenced + +### GC Process ```go -// cmd/hold/main.go +// pkg/hold/gc/gc.go -func main() { - // ... setup ... - - // Start reconciliation cron - if os.Getenv("QUOTA_RECONCILE_ENABLED") == "true" { - reconcileInterval := 24 * time.Hour // Daily - - go func() { - ticker := time.NewTicker(reconcileInterval) - defer ticker.Stop() - - for range ticker.C { - if err := reconciler.ReconcileAll(context.Background()); err != nil { - log.Printf("Reconciliation error: %v", err) - } - } - }() - - log.Printf("Quota reconciliation cron started: runs every %v", reconcileInterval) +func (gc *GarbageCollector) Run(ctx context.Context) error { + // Step 1: Get all referenced digests from layer records + records, err := gc.pds.ListRecords(ctx, LayerCollection, "") + if err != nil { + return err } - // ... start server ... + referenced := make(map[string]bool) + for _, record := range records { + var layer LayerRecord + if err := json.Unmarshal(record.Value, &layer); err != nil { + continue + } + referenced[layer.Digest] = true + } + + log.Printf("Found %d referenced blobs", len(referenced)) + + // Step 2: Walk S3 blobs and delete unreferenced + var deleted, reclaimed int64 + err = gc.driver.Walk(ctx, "/docker/registry/v2/blobs", func(fi storagedriver.FileInfo) error { + if fi.IsDir() { + return nil + } + + digest := extractDigestFromPath(fi.Path()) + if !referenced[digest] { + size := fi.Size() + if err := gc.driver.Delete(ctx, fi.Path()); err != nil { + log.Printf("Failed to delete %s: %v", digest, err) + return nil + } + deleted++ + reclaimed += size + log.Printf("GC: deleted %s (%d bytes)", digest, size) + } + return nil + }) + + log.Printf("GC complete: deleted %d blobs, reclaimed %d bytes", deleted, reclaimed) + return err } ``` -### Why PDS as Source of Truth Works +### GC Schedule -1. **Manifests are canonical** - If manifest exists in PDS, user owns those layers -2. **Public reads** - No OAuth needed, just resolve DID → PDS endpoint -3. **ATProto durability** - PDS is user's authoritative data store -4. **AppView is cache** - AppView database might lag or have inconsistencies -5. **Reconciliation fixes drift** - Periodic sync from PDS ensures accuracy - -**Example reconciliation scenarios:** - -- **Orphaned quota entries:** User deleted manifest from PDS, but hold quota still has it - → Reconciliation removes from claimed_layers - -- **Missing quota entries:** User pushed manifest, but quota update failed - → Reconciliation adds to claimed_layers - -- **Race condition duplicates:** Two concurrent pushes double-counted a layer - → Reconciliation fixes to actual usage +```bash +# Environment variable +GC_ENABLED=true +GC_INTERVAL=24h # Daily by default +``` ## Configuration @@ -997,293 +473,68 @@ func main() { ```bash # .env.hold -# ============================================================================ # Quota Configuration -# ============================================================================ - -# Enable quota enforcement QUOTA_ENABLED=true +QUOTA_DEFAULT_LIMIT=10737418240 # 10GB in bytes -# Default quota limit per user (bytes) -# 10GB = 10737418240 -# 50GB = 53687091200 -# 100GB = 107374182400 -QUOTA_DEFAULT_LIMIT=10737418240 - -# Storage backend for quota data -# Options: s3, sqlite -QUOTA_STORAGE_BACKEND=s3 - -# For S3-based storage: -# Quota files stored in same bucket as blobs -QUOTA_STORAGE_PREFIX=/atcr/quota/ - -# For SQLite-based storage: -QUOTA_DB_PATH=/var/lib/atcr/hold-quota.db - -# ============================================================================ # Garbage Collection -# ============================================================================ - -# Enable periodic garbage collection GC_ENABLED=true - -# GC interval (default: 24h) GC_INTERVAL=24h - -# AppView URL for GC reference checking -APPVIEW_URL=https://atcr.io - -# ============================================================================ -# Quota Reconciliation -# ============================================================================ - -# Enable quota reconciliation from PDS -QUOTA_RECONCILE_ENABLED=true - -# Reconciliation interval (default: 24h) -QUOTA_RECONCILE_INTERVAL=24h - -# ============================================================================ -# Hold Service Identity (Required) -# ============================================================================ - -# Public URL of this hold service -HOLD_PUBLIC_URL=https://hold1.example.com - -# Owner DID (for auto-registration) -HOLD_OWNER=did:plc:xyz123 ``` -### AppView Configuration +### Quota Limits by Bytes -```bash -# .env.appview - -# Internal API endpoint for hold services -# Used for GC reference checking -ATCR_INTERNAL_API_ENABLED=true - -# Optional: authentication token for internal APIs -ATCR_INTERNAL_API_TOKEN=secret123 -``` - -## Trade-offs & Design Decisions - -### 1. Claimed Storage vs Physical Storage - -**Decision:** Track claimed storage (logical accounting) - -**Why:** -- Predictable for users: "you pay for what you upload" -- No complex cross-user dependencies -- Delete always gives you quota back -- Matches Harbor's proven model - -**Trade-off:** -- Total claimed can exceed physical storage -- Users might complain "I uploaded 10GB but S3 only has 6GB" - -**Mitigation:** -- Show deduplication savings metric -- Educate users: "You claimed 10GB, but deduplication saved 4GB" - -### 2. S3 vs SQLite for Quota Storage - -**Decision:** Support both, recommend based on use case - -**S3 Pros:** -- No database to manage -- Quota data lives with blobs -- Better for ephemeral BYOS - -**SQLite Pros:** -- Faster (no network) -- ACID transactions (no race conditions) -- Better for high-traffic shared holds - -**Trade-off:** -- S3: eventual consistency, race conditions -- SQLite: stateful service, scaling challenges - -**Mitigation:** -- Reconciliation fixes S3 inconsistencies -- SQLite can use shared DB for multi-instance - -### 3. Optimistic Quota Update - -**Decision:** Update quota BEFORE upload completes - -**Why:** -- Prevent race conditions (two users uploading simultaneously) -- Can reject before presigned URL generated -- Simpler flow - -**Trade-off:** -- If upload fails, quota already incremented (user "paid" for nothing) - -**Mitigation:** -- Reconciliation from PDS fixes orphaned quota entries -- Acceptable for MVP (upload failures are rare) - -### 4. AppView as Intermediary - -**Decision:** AppView notifies hold service on deletes - -**Why:** -- AppView already has manifest/layer database -- Can efficiently check if layer still referenced -- Hold service doesn't need to query PDS on every delete - -**Trade-off:** -- AppView → Hold dependency -- Network hop on delete - -**Mitigation:** -- If notification fails, reconciliation fixes quota -- Eventually consistent is acceptable - -### 5. PDS as Source of Truth - -**Decision:** Use PDS manifests for reconciliation - -**Why:** -- Manifests in PDS are canonical user data -- Public reads (no OAuth for reconciliation) -- AppView database might lag or be inconsistent - -**Trade-off:** -- Reconciliation requires PDS queries (slower) -- Limited to 1000 manifests per query - -**Mitigation:** -- Run reconciliation daily (not real-time) -- Paginate if user has >1000 manifests +| Size | Bytes | +|------|-------| +| 1 GB | 1073741824 | +| 5 GB | 5368709120 | +| 10 GB | 10737418240 | +| 50 GB | 53687091200 | +| 100 GB | 107374182400 | ## Future Enhancements ### 1. Quota API Endpoints ``` -GET /quota/usage - Get current user's quota -GET /quota/breakdown - Get storage by repository -POST /quota/limit - Update user's quota limit (admin) -GET /quota/stats - Get hold-wide statistics +GET /xrpc/io.atcr.hold.getQuota?did={userDID} - Get user's quota usage +GET /xrpc/io.atcr.hold.getQuotaBreakdown - Storage by repository ``` ### 2. Quota Alerts -Notify users when approaching limit: -- Email/webhook at 80%, 90%, 95% -- Reject uploads at 100% (currently implemented) -- Grace period: allow 105% temporarily +- Warning thresholds at 80%, 90%, 95% +- Email/webhook notifications +- Grace period before hard enforcement ### 3. Tiered Quotas -Different limits based on user tier: -- Free: 10GB -- Pro: 100GB -- Enterprise: unlimited +| Tier | Limit | +|------|-------| +| Free | 10 GB | +| Pro | 100 GB | +| Enterprise | Unlimited | -### 4. Quota Purchasing +### 4. Rate Limiting -Allow users to buy additional storage: -- Stripe integration -- $0.10/GB/month pricing -- Dynamic limit updates +Pull rate limits (Docker Hub style): +- Anonymous: 100 pulls per 6 hours per IP +- Authenticated: 200 pulls per 6 hours +- Paid: Unlimited -### 5. Cross-Hold Deduplication +### 5. Quota Purchasing -If multiple holds share same S3 bucket: -- Track blob ownership globally -- Split costs proportionally -- More complex, but maximizes deduplication - -### 6. Manifest-Based Quota (Alternative Model) - -Instead of tracking layers, track manifests: -- Simpler: just count manifest sizes -- No deduplication benefits for users -- Might be acceptable for some use cases - -### 7. Redis-Based Quota (High Performance) - -For high-traffic registries: -- Use Redis instead of S3/SQLite -- Sub-millisecond quota checks -- Harbor-proven approach - -### 8. Quota Visualizations - -Web UI showing: -- Storage usage over time -- Top consumers by repository -- Deduplication savings graph -- Layer size distribution - -## Appendix: SQL Queries - -### Check if User Still References Layer - -```sql --- After deleting manifest, check if user has other manifests using this layer -SELECT COUNT(*) -FROM layers l -JOIN manifests m ON l.manifest_id = m.id -WHERE m.did = ? -- User's DID - AND l.digest = ? -- Layer digest to check - AND m.id != ? -- Exclude the manifest being deleted -``` - -### Get All Unique Layers for User - -```sql --- Calculate true quota usage for a user -SELECT DISTINCT l.digest, l.size -FROM layers l -JOIN manifests m ON l.manifest_id = m.id -WHERE m.did = ? - AND m.hold_endpoint = ? -``` - -### Get Referenced Blobs for Hold - -```sql --- For GC: get all blobs still referenced by any user of this hold -SELECT DISTINCT l.digest -FROM layers l -JOIN manifests m ON l.manifest_id = m.id -WHERE m.hold_endpoint = ? -``` - -### Get Storage Stats by Repository - -```sql --- User's storage broken down by repository -SELECT - m.repository, - COUNT(DISTINCT m.id) as manifest_count, - COUNT(DISTINCT l.digest) as unique_layers, - SUM(l.size) as total_size -FROM manifests m -JOIN layers l ON l.manifest_id = m.id -WHERE m.did = ? - AND m.hold_endpoint = ? -GROUP BY m.repository -ORDER BY total_size DESC -``` +- Stripe integration for additional storage +- $0.10/GB/month pricing (industry standard) ## References - **Harbor Quotas:** https://goharbor.io/docs/1.10/administration/configure-project-quotas/ -- **Harbor Source:** https://github.com/goharbor/harbor - **ATProto Spec:** https://atproto.com/specs/record - **OCI Distribution Spec:** https://github.com/opencontainers/distribution-spec -- **S3 API Reference:** https://docs.aws.amazon.com/AmazonS3/latest/API/ -- **Distribution GC:** https://github.com/distribution/distribution/blob/main/registry/storage/garbagecollect.go --- -**Document Version:** 1.0 -**Last Updated:** 2025-10-09 -**Author:** Generated from implementation research and Harbor analysis +**Document Version:** 2.0 +**Last Updated:** 2026-01-04 +**Model:** Per-user layer tracking with ATProto records diff --git a/lexicons/io/atcr/hold/layer.json b/lexicons/io/atcr/hold/layer.json index d94e6e1..b56f391 100644 --- a/lexicons/io/atcr/hold/layer.json +++ b/lexicons/io/atcr/hold/layer.json @@ -8,7 +8,7 @@ "description": "Represents metadata about a container layer stored in the hold. Stored in the hold's embedded PDS for tracking and analytics.", "record": { "type": "object", - "required": ["digest", "size", "mediaType", "repository", "userDid", "userHandle", "createdAt"], + "required": ["digest", "size", "mediaType", "manifest", "userDid", "createdAt"], "properties": { "digest": { "type": "string", @@ -24,21 +24,16 @@ "description": "Media type (e.g., application/vnd.oci.image.layer.v1.tar+gzip)", "maxLength": 128 }, - "repository": { + "manifest": { "type": "string", - "description": "Repository this layer belongs to", - "maxLength": 255 + "format": "at-uri", + "description": "AT-URI of the manifest that included this layer (e.g., at://did:plc:xyz/io.atcr.manifest/abc123)" }, "userDid": { "type": "string", "format": "did", "description": "DID of user who uploaded this layer" }, - "userHandle": { - "type": "string", - "format": "handle", - "description": "Handle of user (for display purposes)" - }, "createdAt": { "type": "string", "format": "datetime", diff --git a/pkg/appview/handlers/storage.go b/pkg/appview/handlers/storage.go new file mode 100644 index 0000000..62490b7 --- /dev/null +++ b/pkg/appview/handlers/storage.go @@ -0,0 +1,129 @@ +package handlers + +import ( + "encoding/json" + "fmt" + "html/template" + "log/slog" + "net/http" + + "atcr.io/pkg/appview/middleware" + "atcr.io/pkg/appview/storage" + "atcr.io/pkg/atproto" + "atcr.io/pkg/auth/oauth" +) + +// StorageHandler handles the storage quota API endpoint +// Returns an HTML partial for HTMX to swap into the settings page +type StorageHandler struct { + Templates *template.Template + Refresher *oauth.Refresher +} + +// QuotaStats mirrors the hold service response +type QuotaStats struct { + UserDID string `json:"userDid"` + UniqueBlobs int `json:"uniqueBlobs"` + TotalSize int64 `json:"totalSize"` +} + +func (h *StorageHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + user := middleware.GetUser(r) + if user == nil { + http.Error(w, "Unauthorized", http.StatusUnauthorized) + return + } + + // Create ATProto client with session provider + client := atproto.NewClientWithSessionProvider(user.PDSEndpoint, user.DID, h.Refresher) + + // Get user's sailor profile to find their default hold + profile, err := storage.GetProfile(r.Context(), client) + if err != nil { + slog.Warn("Failed to get profile for storage quota", "did", user.DID, "error", err) + h.renderError(w, "Failed to load profile") + return + } + + if profile == nil || profile.DefaultHold == "" { + // No default hold configured - can't check quota + h.renderNoHold(w) + return + } + + // Resolve hold URL from DID + holdURL := atproto.ResolveHoldURL(profile.DefaultHold) + if holdURL == "" { + slog.Warn("Failed to resolve hold URL", "did", user.DID, "holdDid", profile.DefaultHold) + h.renderError(w, "Failed to resolve hold service") + return + } + + // Call the hold's quota endpoint + quotaURL := fmt.Sprintf("%s%s?userDid=%s", holdURL, atproto.HoldGetQuota, user.DID) + resp, err := http.Get(quotaURL) + if err != nil { + slog.Warn("Failed to fetch quota from hold", "did", user.DID, "holdURL", holdURL, "error", err) + h.renderError(w, "Failed to connect to hold service") + return + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + slog.Warn("Hold returned error for quota", "did", user.DID, "status", resp.StatusCode) + h.renderError(w, "Hold service returned an error") + return + } + + var stats QuotaStats + if err := json.NewDecoder(resp.Body).Decode(&stats); err != nil { + slog.Warn("Failed to decode quota response", "did", user.DID, "error", err) + h.renderError(w, "Failed to parse quota data") + return + } + + // Render the stats partial + h.renderStats(w, stats) +} + +func (h *StorageHandler) renderStats(w http.ResponseWriter, stats QuotaStats) { + data := struct { + UniqueBlobs int + TotalSize int64 + HumanSize string + }{ + UniqueBlobs: stats.UniqueBlobs, + TotalSize: stats.TotalSize, + HumanSize: humanizeBytes(stats.TotalSize), + } + + w.Header().Set("Content-Type", "text/html") + if err := h.Templates.ExecuteTemplate(w, "storage_stats", data); err != nil { + slog.Error("Failed to render storage stats template", "error", err) + http.Error(w, "Failed to render template", http.StatusInternalServerError) + } +} + +func (h *StorageHandler) renderError(w http.ResponseWriter, message string) { + w.Header().Set("Content-Type", "text/html") + fmt.Fprintf(w, `
%s
`, message) +} + +func (h *StorageHandler) renderNoHold(w http.ResponseWriter) { + w.Header().Set("Content-Type", "text/html") + fmt.Fprint(w, `
No hold configured. Set a default hold above to see storage usage.
`) +} + +// humanizeBytes converts bytes to human-readable format +func humanizeBytes(bytes int64) string { + const unit = 1024 + if bytes < unit { + return fmt.Sprintf("%d B", bytes) + } + div, exp := int64(unit), 0 + for n := bytes / unit; n >= unit; n /= unit { + div *= unit + exp++ + } + return fmt.Sprintf("%.1f %cB", float64(bytes)/float64(div), "KMGTPE"[exp]) +} diff --git a/pkg/appview/routes/routes.go b/pkg/appview/routes/routes.go index b7ac28a..89a59cf 100644 --- a/pkg/appview/routes/routes.go +++ b/pkg/appview/routes/routes.go @@ -174,6 +174,11 @@ func RegisterUIRoutes(router chi.Router, deps UIDependencies) { RegistryURL: registryURL, }).ServeHTTP) + r.Get("/api/storage", (&uihandlers.StorageHandler{ + Templates: deps.Templates, + Refresher: deps.Refresher, + }).ServeHTTP) + r.Post("/api/profile/default-hold", (&uihandlers.UpdateDefaultHoldHandler{ Refresher: deps.Refresher, }).ServeHTTP) diff --git a/pkg/appview/storage/manifest_store.go b/pkg/appview/storage/manifest_store.go index 7b4bc33..4df2e80 100644 --- a/pkg/appview/storage/manifest_store.go +++ b/pkg/appview/storage/manifest_store.go @@ -325,11 +325,12 @@ func (s *ManifestStore) notifyHoldAboutManifest(ctx context.Context, manifestRec serviceToken := s.ctx.ServiceToken // Build notification request + // Note: userHandle is resolved from userDid on the hold side (cached, 24-hour TTL) notifyReq := map[string]any{ - "repository": s.ctx.Repository, - "userDid": s.ctx.DID, - "userHandle": s.ctx.Handle, - "operation": operation, + "repository": s.ctx.Repository, + "userDid": s.ctx.DID, + "manifestDigest": manifestDigest, + "operation": operation, } // For push operations, include full manifest data diff --git a/pkg/appview/templates/pages/settings.html b/pkg/appview/templates/pages/settings.html index dab2b46..059b66e 100644 --- a/pkg/appview/templates/pages/settings.html +++ b/pkg/appview/templates/pages/settings.html @@ -29,6 +29,15 @@ + +
+

Storage Usage

+

Estimated storage usage on your default hold.

+
+

Loading...

+
+
+

Default Hold

@@ -200,6 +209,56 @@