From 92d794415a1056314c806fd1762237c7fec24da9 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Tue, 4 Nov 2025 22:48:07 -0600 Subject: [PATCH] don't use in-memory for holddid caching, just reference from db --- CLAUDE.md | 16 +- pkg/appview/db/queries.go | 29 ++++ pkg/appview/storage/context.go | 3 +- pkg/appview/storage/context_test.go | 5 + pkg/appview/storage/hold_cache.go | 98 ------------ pkg/appview/storage/hold_cache_test.go | 150 ------------------ pkg/appview/storage/routing_repository.go | 40 +++-- .../storage/routing_repository_test.go | 87 +++++++--- 8 files changed, 124 insertions(+), 304 deletions(-) delete mode 100644 pkg/appview/storage/hold_cache.go delete mode 100644 pkg/appview/storage/hold_cache_test.go diff --git a/CLAUDE.md b/CLAUDE.md index f8a4d2b..f09859d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -349,12 +349,13 @@ Later (subsequent docker push): - Implements `distribution.Repository` interface - Uses RegistryContext to pass DID, PDS endpoint, hold DID, OAuth refresher, etc. -**hold_cache.go**: In-memory hold DID cache -- Caches `(DID, repository) → holdDid` for pull operations -- TTL: 10 minutes (covers typical pull operations) -- Cleanup: Background goroutine runs every 5 minutes -- **NOTE:** Simple in-memory cache for MVP. For production: use Redis or similar -- Prevents expensive PDS manifest lookups on every blob request during pull +**Database-based hold DID lookups**: +- Queries SQLite `manifests` table for hold DID (indexed, fast) +- No in-memory caching needed - database IS the cache +- Persistent across restarts, multi-instance safe +- Pull operations use hold DID from latest manifest (historical reference) +- Push operations use fresh discovery from profile/default +- Function: `db.GetLatestHoldDIDForRepo(did, repository)` in `pkg/appview/db/queries.go` **proxy_blob_store.go**: External storage proxy (routes to hold via XRPC) - Resolves hold DID → HTTP URL for XRPC requests (did:web resolution) @@ -604,7 +605,8 @@ See `.env.hold.example` for all available options. Key environment variables: **General:** - Middleware is in `pkg/appview/middleware/` (auth.go, registry.go) -- Storage routing is in `pkg/appview/storage/` (routing_repository.go, proxy_blob_store.go, hold_cache.go) +- Storage routing is in `pkg/appview/storage/` (routing_repository.go, proxy_blob_store.go) +- Hold DID lookups use database queries (no in-memory caching) - Storage drivers imported as `_ "github.com/distribution/distribution/v3/registry/storage/driver/s3-aws"` - Hold service reuses distribution's driver factory for multi-backend support diff --git a/pkg/appview/db/queries.go b/pkg/appview/db/queries.go index 3bededc..236b8c7 100644 --- a/pkg/appview/db/queries.go +++ b/pkg/appview/db/queries.go @@ -724,6 +724,30 @@ func GetNewestManifestForRepo(db *sql.DB, did, repository string) (*Manifest, er return &m, nil } +// GetLatestHoldDIDForRepo returns the hold DID from the most recent manifest for a repository +// Returns empty string if no manifests exist (e.g., first push) +// This is used instead of the in-memory cache to determine which hold to use for blob operations +func GetLatestHoldDIDForRepo(db *sql.DB, did, repository string) (string, error) { + var holdDID string + err := db.QueryRow(` + SELECT hold_endpoint + FROM manifests + WHERE did = ? AND repository = ? + ORDER BY created_at DESC + LIMIT 1 + `, did, repository).Scan(&holdDID) + + if err == sql.ErrNoRows { + // No manifests yet - return empty string (first push case) + return "", nil + } + if err != nil { + return "", err + } + + return holdDID, nil +} + // GetRepositoriesForDID returns all unique repository names for a DID // Used by backfill to reconcile annotations for all repositories func GetRepositoriesForDID(db *sql.DB, did string) ([]string, error) { @@ -1576,6 +1600,11 @@ func (m *MetricsDB) IncrementPushCount(did, repository string) error { return IncrementPushCount(m.db, did, repository) } +// GetLatestHoldDIDForRepo returns the hold DID from the most recent manifest for a repository +func (m *MetricsDB) GetLatestHoldDIDForRepo(did, repository string) (string, error) { + return GetLatestHoldDIDForRepo(m.db, did, repository) +} + // GetFeaturedRepositories fetches top repositories sorted by stars and pulls func GetFeaturedRepositories(db *sql.DB, limit int, currentUserDID string) ([]FeaturedRepository, error) { query := ` diff --git a/pkg/appview/storage/context.go b/pkg/appview/storage/context.go index 14fc138..4028db5 100644 --- a/pkg/appview/storage/context.go +++ b/pkg/appview/storage/context.go @@ -8,10 +8,11 @@ import ( "atcr.io/pkg/auth/oauth" ) -// DatabaseMetrics interface for tracking pull/push counts +// DatabaseMetrics interface for tracking pull/push counts and querying hold DIDs type DatabaseMetrics interface { IncrementPullCount(did, repository string) error IncrementPushCount(did, repository string) error + GetLatestHoldDIDForRepo(did, repository string) (string, error) } // ReadmeCache interface for README content caching diff --git a/pkg/appview/storage/context_test.go b/pkg/appview/storage/context_test.go index dce13c8..0ca4164 100644 --- a/pkg/appview/storage/context_test.go +++ b/pkg/appview/storage/context_test.go @@ -29,6 +29,11 @@ func (m *mockDatabaseMetrics) IncrementPushCount(did, repository string) error { return nil } +func (m *mockDatabaseMetrics) GetLatestHoldDIDForRepo(did, repository string) (string, error) { + // Return empty string for mock - tests can override if needed + return "", nil +} + func (m *mockDatabaseMetrics) getPullCount() int { m.mu.Lock() defer m.mu.Unlock() diff --git a/pkg/appview/storage/hold_cache.go b/pkg/appview/storage/hold_cache.go deleted file mode 100644 index 60a6100..0000000 --- a/pkg/appview/storage/hold_cache.go +++ /dev/null @@ -1,98 +0,0 @@ -package storage - -import ( - "sync" - "time" -) - -// HoldCache caches hold DIDs for (DID, repository) pairs -// This avoids expensive ATProto lookups on every blob request during pulls -// -// NOTE: This is a simple in-memory cache for MVP. For production deployments: -// - Use Redis or similar for distributed caching -// - Consider implementing cache size limits -// - Monitor memory usage under high load -type HoldCache struct { - mu sync.RWMutex - cache map[string]*holdCacheEntry -} - -type holdCacheEntry struct { - holdDID string - expiresAt time.Time -} - -var globalHoldCache = &HoldCache{ - cache: make(map[string]*holdCacheEntry), -} - -func init() { - // Start background cleanup goroutine - go func() { - ticker := time.NewTicker(5 * time.Minute) - defer ticker.Stop() - for range ticker.C { - globalHoldCache.Cleanup() - } - }() -} - -// GetGlobalHoldCache returns the global hold cache instance -func GetGlobalHoldCache() *HoldCache { - return globalHoldCache -} - -// Set stores a hold DID for a (DID, repository) pair with a TTL -func (c *HoldCache) Set(did, repository, holdDID string, ttl time.Duration) { - c.mu.Lock() - defer c.mu.Unlock() - - key := did + ":" + repository - c.cache[key] = &holdCacheEntry{ - holdDID: holdDID, - expiresAt: time.Now().Add(ttl), - } -} - -// Get retrieves a hold DID for a (DID, repository) pair -// Returns empty string and false if not found or expired -func (c *HoldCache) Get(did, repository string) (string, bool) { - c.mu.RLock() - defer c.mu.RUnlock() - - key := did + ":" + repository - entry, ok := c.cache[key] - if !ok { - return "", false - } - - // Check if expired - if time.Now().After(entry.expiresAt) { - // Don't delete here (would need write lock), let cleanup handle it - return "", false - } - - return entry.holdDID, true -} - -// Cleanup removes expired entries (called automatically every 5 minutes) -func (c *HoldCache) Cleanup() { - c.mu.Lock() - defer c.mu.Unlock() - - now := time.Now() - removed := 0 - for key, entry := range c.cache { - if now.After(entry.expiresAt) { - delete(c.cache, key) - removed++ - } - } - - // Log cleanup stats for monitoring - if removed > 0 || len(c.cache) > 100 { - // Log if we removed entries OR if cache is growing large - // This helps identify if cache size is becoming a concern - println("Hold cache cleanup: removed", removed, "entries, remaining", len(c.cache)) - } -} diff --git a/pkg/appview/storage/hold_cache_test.go b/pkg/appview/storage/hold_cache_test.go deleted file mode 100644 index 94e7e00..0000000 --- a/pkg/appview/storage/hold_cache_test.go +++ /dev/null @@ -1,150 +0,0 @@ -package storage - -import ( - "testing" - "time" -) - -func TestHoldCache_SetAndGet(t *testing.T) { - cache := &HoldCache{ - cache: make(map[string]*holdCacheEntry), - } - - did := "did:plc:test123" - repo := "myapp" - holdDID := "did:web:hold01.atcr.io" - ttl := 10 * time.Minute - - // Set a value - cache.Set(did, repo, holdDID, ttl) - - // Get the value - should succeed - gotHoldDID, ok := cache.Get(did, repo) - if !ok { - t.Fatal("Expected Get to return true, got false") - } - if gotHoldDID != holdDID { - t.Errorf("Expected hold DID %q, got %q", holdDID, gotHoldDID) - } -} - -func TestHoldCache_GetNonExistent(t *testing.T) { - cache := &HoldCache{ - cache: make(map[string]*holdCacheEntry), - } - - // Get non-existent value - _, ok := cache.Get("did:plc:nonexistent", "repo") - if ok { - t.Error("Expected Get to return false for non-existent key") - } -} - -func TestHoldCache_ExpiredEntry(t *testing.T) { - cache := &HoldCache{ - cache: make(map[string]*holdCacheEntry), - } - - did := "did:plc:test123" - repo := "myapp" - holdDID := "did:web:hold01.atcr.io" - - // Set with very short TTL - cache.Set(did, repo, holdDID, 10*time.Millisecond) - - // Wait for expiration - time.Sleep(20 * time.Millisecond) - - // Get should return false - _, ok := cache.Get(did, repo) - if ok { - t.Error("Expected Get to return false for expired entry") - } -} - -func TestHoldCache_Cleanup(t *testing.T) { - cache := &HoldCache{ - cache: make(map[string]*holdCacheEntry), - } - - // Add multiple entries with different TTLs - cache.Set("did:plc:1", "repo1", "hold1", 10*time.Millisecond) - cache.Set("did:plc:2", "repo2", "hold2", 1*time.Hour) - cache.Set("did:plc:3", "repo3", "hold3", 10*time.Millisecond) - - // Wait for some to expire - time.Sleep(20 * time.Millisecond) - - // Run cleanup - cache.Cleanup() - - // Verify expired entries are removed - if _, ok := cache.Get("did:plc:1", "repo1"); ok { - t.Error("Expected expired entry 1 to be removed") - } - if _, ok := cache.Get("did:plc:3", "repo3"); ok { - t.Error("Expected expired entry 3 to be removed") - } - - // Verify non-expired entry remains - if _, ok := cache.Get("did:plc:2", "repo2"); !ok { - t.Error("Expected non-expired entry to remain") - } -} - -func TestHoldCache_ConcurrentAccess(t *testing.T) { - cache := &HoldCache{ - cache: make(map[string]*holdCacheEntry), - } - - done := make(chan bool) - - // Concurrent writes - for i := 0; i < 10; i++ { - go func(id int) { - did := "did:plc:concurrent" - repo := "repo" + string(rune(id)) - holdDID := "hold" + string(rune(id)) - cache.Set(did, repo, holdDID, 1*time.Minute) - done <- true - }(i) - } - - // Concurrent reads - for i := 0; i < 10; i++ { - go func(id int) { - repo := "repo" + string(rune(id)) - cache.Get("did:plc:concurrent", repo) - done <- true - }(i) - } - - // Wait for all goroutines - for i := 0; i < 20; i++ { - <-done - } -} - -func TestHoldCache_KeyFormat(t *testing.T) { - cache := &HoldCache{ - cache: make(map[string]*holdCacheEntry), - } - - did := "did:plc:test" - repo := "myrepo" - holdDID := "did:web:hold" - - cache.Set(did, repo, holdDID, 1*time.Minute) - - // Verify the key is stored correctly (did:repo) - expectedKey := did + ":" + repo - if _, exists := cache.cache[expectedKey]; !exists { - t.Errorf("Expected key %q to exist in cache", expectedKey) - } -} - -// TODO: Add more comprehensive tests: -// - Test GetGlobalHoldCache() -// - Test cache size monitoring -// - Benchmark cache performance under load -// - Test cleanup goroutine timing diff --git a/pkg/appview/storage/routing_repository.go b/pkg/appview/storage/routing_repository.go index ad6a4ff..c19471e 100644 --- a/pkg/appview/storage/routing_repository.go +++ b/pkg/appview/storage/routing_repository.go @@ -1,6 +1,6 @@ // Package storage implements the storage routing layer for AppView. // It routes manifests to ATProto PDS (as io.atcr.manifest records) and -// blobs to hold services via XRPC, with hold DID caching for efficient pulls. +// blobs to hold services via XRPC, with database-based hold DID lookups. // All storage operations are proxied - AppView stores nothing locally. package storage @@ -8,7 +8,6 @@ import ( "context" "log/slog" "sync" - "time" "github.com/distribution/distribution/v3" ) @@ -50,17 +49,6 @@ func (r *RoutingRepository) Manifests(ctx context.Context, options ...distributi manifestStore := r.manifestStore r.mu.Unlock() - // After any manifest operation, cache the hold DID for blob fetches - // We use a goroutine to avoid blocking, and check after a short delay to allow the operation to complete - go func() { - time.Sleep(100 * time.Millisecond) // Brief delay to let manifest fetch complete - if holdDID := manifestStore.GetLastFetchedHoldDID(); holdDID != "" { - // Cache for 10 minutes - should cover typical pull operations - GetGlobalHoldCache().Set(r.Ctx.DID, r.Ctx.Repository, holdDID, 10*time.Minute) - slog.Debug("Cached hold DID", "component", "storage/routing", "did", r.Ctx.DID, "repo", r.Ctx.Repository, "hold", holdDID) - } - }() - return manifestStore, nil } @@ -76,17 +64,23 @@ func (r *RoutingRepository) Blobs(ctx context.Context) distribution.BlobStore { return blobStore } - // For pull operations, check if we have a cached hold DID from a recent manifest fetch + // For pull operations, check database for hold DID from the most recent manifest // This ensures blobs are fetched from the hold recorded in the manifest, not re-discovered holdDID := r.Ctx.HoldDID // Default to discovery-based DID + holdSource := "discovery" - if cachedHoldDID, ok := GetGlobalHoldCache().Get(r.Ctx.DID, r.Ctx.Repository); ok { - // Use cached hold DID from manifest - holdDID = cachedHoldDID - slog.Debug("Using cached hold from manifest", "component", "storage/blobs", "did", r.Ctx.DID, "repo", r.Ctx.Repository, "hold", cachedHoldDID) - } else { - // No cached hold, use discovery-based DID (for push or first pull) - slog.Debug("Using discovery-based hold", "component", "storage/blobs", "did", r.Ctx.DID, "repo", r.Ctx.Repository, "hold", holdDID) + if r.Ctx.Database != nil { + // Query database for the latest manifest's hold DID + if dbHoldDID, err := r.Ctx.Database.GetLatestHoldDIDForRepo(r.Ctx.DID, r.Ctx.Repository); err == nil && dbHoldDID != "" { + // Use hold DID from database (pull case - use historical reference) + holdDID = dbHoldDID + holdSource = "database" + slog.Debug("Using hold from database manifest", "component", "storage/blobs", "did", r.Ctx.DID, "repo", r.Ctx.Repository, "hold", dbHoldDID) + } else if err != nil { + // Log error but don't fail - fall back to discovery-based DID + slog.Warn("Failed to query database for hold DID", "component", "storage/blobs", "error", err) + } + // If dbHoldDID is empty (no manifests yet), fall through to use discovery-based DID } if holdDID == "" { @@ -94,7 +88,9 @@ func (r *RoutingRepository) Blobs(ctx context.Context) distribution.BlobStore { panic("hold DID not set in RegistryContext - ensure default_hold_did is configured in middleware") } - // Update context with the correct hold DID (may be cached or discovered) + slog.Debug("Using hold DID for blobs", "component", "storage/blobs", "did", r.Ctx.DID, "repo", r.Ctx.Repository, "hold", holdDID, "source", holdSource) + + // Update context with the correct hold DID (may be from database or discovered) r.Ctx.HoldDID = holdDID // Create and cache proxy blob store diff --git a/pkg/appview/storage/routing_repository_test.go b/pkg/appview/storage/routing_repository_test.go index 5f1b4f3..3685806 100644 --- a/pkg/appview/storage/routing_repository_test.go +++ b/pkg/appview/storage/routing_repository_test.go @@ -4,7 +4,6 @@ import ( "context" "sync" "testing" - "time" "github.com/distribution/distribution/v3" "github.com/stretchr/testify/assert" @@ -13,6 +12,27 @@ import ( "atcr.io/pkg/atproto" ) +// mockDatabase is a simple mock for testing +type mockDatabase struct { + holdDID string + err error +} + +func (m *mockDatabase) IncrementPullCount(did, repository string) error { + return nil +} + +func (m *mockDatabase) IncrementPushCount(did, repository string) error { + return nil +} + +func (m *mockDatabase) GetLatestHoldDIDForRepo(did, repository string) (string, error) { + if m.err != nil { + return "", m.err + } + return m.holdDID, nil +} + func TestNewRoutingRepository(t *testing.T) { ctx := &RegistryContext{ DID: "did:plc:test123", @@ -89,38 +109,36 @@ func TestRoutingRepository_ManifestStoreCaching(t *testing.T) { assert.NotNil(t, repo.manifestStore) } -// TestRoutingRepository_Blobs_WithCache tests blob store with cached hold DID -func TestRoutingRepository_Blobs_WithCache(t *testing.T) { - // Pre-populate the hold cache - cache := GetGlobalHoldCache() - cachedHoldDID := "did:web:cached.hold.io" - cache.Set("did:plc:test123", "myapp", cachedHoldDID, 10*time.Minute) +// TestRoutingRepository_Blobs_WithDatabase tests blob store with database hold DID +func TestRoutingRepository_Blobs_WithDatabase(t *testing.T) { + dbHoldDID := "did:web:database.hold.io" ctx := &RegistryContext{ DID: "did:plc:test123", Repository: "myapp", HoldDID: "did:web:default.hold.io", // Discovery-based hold (should be overridden) ATProtoClient: atproto.NewClient("https://pds.example.com", "did:plc:test123", ""), + Database: &mockDatabase{holdDID: dbHoldDID}, } repo := NewRoutingRepository(nil, ctx) blobStore := repo.Blobs(context.Background()) assert.NotNil(t, blobStore) - // Verify the hold DID was updated to use the cached value - assert.Equal(t, cachedHoldDID, repo.Ctx.HoldDID, "should use cached hold DID") + // Verify the hold DID was updated to use the database value + assert.Equal(t, dbHoldDID, repo.Ctx.HoldDID, "should use database hold DID") } -// TestRoutingRepository_Blobs_WithoutCache tests blob store with discovery-based hold -func TestRoutingRepository_Blobs_WithoutCache(t *testing.T) { +// TestRoutingRepository_Blobs_WithoutDatabase tests blob store with discovery-based hold +func TestRoutingRepository_Blobs_WithoutDatabase(t *testing.T) { discoveryHoldDID := "did:web:discovery.hold.io" - // Use a different DID/repo to avoid cache contamination from other tests ctx := &RegistryContext{ DID: "did:plc:nocache456", Repository: "uncached-app", HoldDID: discoveryHoldDID, ATProtoClient: atproto.NewClient("https://pds.example.com", "did:plc:nocache456", ""), + Database: nil, // No database } repo := NewRoutingRepository(nil, ctx) @@ -131,6 +149,26 @@ func TestRoutingRepository_Blobs_WithoutCache(t *testing.T) { assert.Equal(t, discoveryHoldDID, repo.Ctx.HoldDID, "should use discovery-based hold DID") } +// TestRoutingRepository_Blobs_DatabaseEmptyFallback tests fallback when database returns empty hold DID +func TestRoutingRepository_Blobs_DatabaseEmptyFallback(t *testing.T) { + discoveryHoldDID := "did:web:discovery.hold.io" + + ctx := &RegistryContext{ + DID: "did:plc:test123", + Repository: "newapp", + HoldDID: discoveryHoldDID, + ATProtoClient: atproto.NewClient("https://pds.example.com", "did:plc:test123", ""), + Database: &mockDatabase{holdDID: ""}, // Empty string (no manifests yet) + } + + repo := NewRoutingRepository(nil, ctx) + blobStore := repo.Blobs(context.Background()) + + assert.NotNil(t, blobStore) + // Verify the hold DID falls back to discovery-based + assert.Equal(t, discoveryHoldDID, repo.Ctx.HoldDID, "should fall back to discovery-based hold DID when database returns empty") +} + // TestRoutingRepository_BlobStoreCaching tests that blob store is cached func TestRoutingRepository_BlobStoreCaching(t *testing.T) { ctx := &RegistryContext{ @@ -254,26 +292,23 @@ func TestRoutingRepository_ConcurrentAccess(t *testing.T) { assert.NotNil(t, cachedBlobStore) } -// TestRoutingRepository_HoldCachePopulation tests that hold DID cache is populated after manifest fetch -// Note: This test verifies the goroutine behavior with a delay -func TestRoutingRepository_HoldCachePopulation(t *testing.T) { +// TestRoutingRepository_Blobs_Priority tests that database hold DID takes priority over discovery +func TestRoutingRepository_Blobs_Priority(t *testing.T) { + dbHoldDID := "did:web:database.hold.io" + discoveryHoldDID := "did:web:discovery.hold.io" + ctx := &RegistryContext{ DID: "did:plc:test123", Repository: "myapp", - HoldDID: "did:web:hold01.atcr.io", + HoldDID: discoveryHoldDID, // Discovery-based hold ATProtoClient: atproto.NewClient("https://pds.example.com", "did:plc:test123", ""), + Database: &mockDatabase{holdDID: dbHoldDID}, // Database has a different hold DID } repo := NewRoutingRepository(nil, ctx) + blobStore := repo.Blobs(context.Background()) - // Create manifest store (which triggers the cache population goroutine) - _, err := repo.Manifests(context.Background()) - require.NoError(t, err) - - // Wait for goroutine to complete (it has a 100ms sleep) - time.Sleep(200 * time.Millisecond) - - // Note: We can't easily verify the cache was populated without a real manifest fetch - // The actual caching happens in GetLastFetchedHoldDID() which requires manifest operations - // This test primarily verifies the Manifests() call doesn't panic with the goroutine + assert.NotNil(t, blobStore) + // Database hold DID should take priority over discovery + assert.Equal(t, dbHoldDID, repo.Ctx.HoldDID, "database hold DID should take priority over discovery") }