diff --git a/TODO.md b/TODO.md index c70f11b..404abd1 100644 --- a/TODO.md +++ b/TODO.md @@ -29,6 +29,13 @@ P2: security: referer blacklist # Completed Steps +- 2026-10-04 upstream host semaphores and variant `.meta` files no longer + outlive their use (closes #87): the fetcher counts the fetches holding or + waiting for a slot of each upstream host's semaphore and removes the host's + semaphore once none is left, so fetches from many hosts no longer leave one + semaphore each until restart; `VariantStorage.Delete` removes the variant's + `.meta` file along with it, a missing `.meta` file not being an error, and + `DeleteWithMeta`, which eviction called for that, is gone. - 2026-10-04 `README.md` matches the code (closes #74): "Storage" names the cache directories pixa uses (`cache/sources`, `cache/metadata`, `cache/variants`) and how files are named in each, and the comments in diff --git a/internal/httpfetcher/fetch_internal_test.go b/internal/httpfetcher/fetch_internal_test.go index c0d6e9d..83c78c0 100644 --- a/internal/httpfetcher/fetch_internal_test.go +++ b/internal/httpfetcher/fetch_internal_test.go @@ -191,7 +191,23 @@ func fetchBody(t *testing.T, f *HTTPFetcher, path string) string { // semLen reports how many per-host semaphore slots are currently held. func semLen(f *HTTPFetcher, host string) int { - return len(f.getHostSemaphore(host)) + f.hostSemMu.Lock() + defer f.hostSemMu.Unlock() + + sem, ok := f.hostSems[host] + if !ok { + return 0 + } + + return len(sem.slots) +} + +// hostSemCount reports how many hosts have a semaphore in hostSems. +func hostSemCount(f *HTTPFetcher) int { + f.hostSemMu.Lock() + defer f.hostSemMu.Unlock() + + return len(f.hostSems) } func TestFetchRedirectToPrivateIPBlocked(t *testing.T) { diff --git a/internal/httpfetcher/httpfetcher.go b/internal/httpfetcher/httpfetcher.go index d65f2da..3cfee1c 100644 --- a/internal/httpfetcher/httpfetcher.go +++ b/internal/httpfetcher/httpfetcher.go @@ -160,10 +160,13 @@ func DefaultConfig() *Config { // HTTPFetcher implements Fetcher with SSRF protection and connection limits // per host and for all hosts together. type HTTPFetcher struct { - client *http.Client - config *Config - hostSems map[string]chan struct{} // per-host semaphores - hostSemMu sync.Mutex // protects hostSems map + client *http.Client + config *Config + // hostSems holds the semaphore of each host with a fetch holding or + // waiting for one of its slots; the entry is removed when the host's + // last such fetch gives its slot back or stops waiting. + hostSems map[string]*hostSemaphore + hostSemMu sync.Mutex // protects hostSems and each entry's count // allHostsSemaphore has one slot per connection allowed to all hosts // together (config.MaxConnections). allHostsSemaphore chan struct{} @@ -171,6 +174,14 @@ type HTTPFetcher struct { connectionWaitTimeout time.Duration } +// hostSemaphore is one host's connection slots +// (config.MaxConnectionsPerHost) and the number of fetches holding or +// waiting for one of them. +type hostSemaphore struct { + slots chan struct{} + count int +} + // New creates a new HTTPFetcher with SSRF protection. func New(config *Config) *HTTPFetcher { if config == nil { @@ -211,7 +222,7 @@ func New(config *Config) *HTTPFetcher { return &HTTPFetcher{ client: client, config: config, - hostSems: make(map[string]chan struct{}), + hostSems: make(map[string]*hostSemaphore), allHostsSemaphore: make(chan struct{}, config.MaxConnections), connectionWaitTimeout: ConnectionWaitTimeout, } @@ -307,6 +318,8 @@ func (f *HTTPFetcher) acquireConnection( select { case hostSem <- struct{}{}: case <-ctx.Done(): + f.putHostSemaphore(host) + return nil, ctx.Err() } @@ -314,32 +327,55 @@ func (f *HTTPFetcher) acquireConnection( case f.allHostsSemaphore <- struct{}{}: case <-time.After(f.connectionWaitTimeout): <-hostSem + f.putHostSemaphore(host) return nil, ErrTooManyConnections case <-ctx.Done(): <-hostSem + f.putHostSemaphore(host) return nil, ctx.Err() } return func() { <-hostSem + f.putHostSemaphore(host) <-f.allHostsSemaphore }, nil } -// getHostSemaphore returns the semaphore for a host, creating it if necessary. +// getHostSemaphore returns the semaphore for a host, creating it if +// necessary, and counts the caller among the fetches using it. The caller +// calls putHostSemaphore once it holds no slot and waits for none. func (f *HTTPFetcher) getHostSemaphore(host string) chan struct{} { f.hostSemMu.Lock() defer f.hostSemMu.Unlock() sem, ok := f.hostSems[host] if !ok { - sem = make(chan struct{}, f.config.MaxConnectionsPerHost) + sem = &hostSemaphore{ + slots: make(chan struct{}, f.config.MaxConnectionsPerHost), + } f.hostSems[host] = sem } - return sem + sem.count++ + + return sem.slots +} + +// putHostSemaphore stops counting the caller among the fetches using the +// host's semaphore, and removes the semaphore when no fetch uses it. +func (f *HTTPFetcher) putHostSemaphore(host string) { + f.hostSemMu.Lock() + defer f.hostSemMu.Unlock() + + sem := f.hostSems[host] + + sem.count-- + if sem.count == 0 { + delete(f.hostSems, host) + } } // buildResult validates the upstream response and assembles a FetchResult diff --git a/internal/httpfetcher/max_connections_internal_test.go b/internal/httpfetcher/max_connections_internal_test.go index d23816c..a776460 100644 --- a/internal/httpfetcher/max_connections_internal_test.go +++ b/internal/httpfetcher/max_connections_internal_test.go @@ -5,6 +5,7 @@ import ( "errors" "net" "strconv" + "sync" "testing" "time" ) @@ -119,6 +120,98 @@ func TestFetchFreesHostSlotWhenContextEndsWaitingForConnection(t *testing.T) { } } +// TestFetchRemovesIdleHostSemaphores checks that a host's semaphore is +// removed once no fetch holds or waits for one of its slots: after 100 +// concurrent fetches from 50 hosts have all finished, no semaphore is left. +func TestFetchRemovesIdleHostSemaphores(t *testing.T) { + t.Parallel() + + srv := startUpstream(t) + f, _ := newServerFetcher(t, srv, nil) + ctx := testContext(t) + + var wg sync.WaitGroup + + for i := range 100 { + wg.Go(func() { + res, err := f.Fetch(ctx, imageURLOnPort(1+i%50)) + if err != nil { + t.Errorf("Fetch() error = %v", err) + + return + } + + _ = res.Content.Close() + }) + } + + wg.Wait() + + if n := hostSemCount(f); n != 0 { + t.Errorf("%d host semaphores left after every fetch finished, want 0", n) + } +} + +// TestFetchRemovesHostSemaphoreWhenNoConnection checks that a fetch that +// ends without a connection leaves no semaphore behind: when it is refused +// after waiting for a connection shared by all hosts, when its context ends +// while it waits for its host's slot, and when its context ends while it +// waits for a connection shared by all hosts, long before the 10 second +// wait timeout. +func TestFetchRemovesHostSemaphoreWhenNoConnection(t *testing.T) { + t.Parallel() + + srv := startUpstream(t) + + cfg := DefaultConfig() + cfg.MaxConnections = 1 + cfg.MaxConnectionsPerHost = 1 + + f, _ := newServerFetcher(t, srv, cfg) + f.connectionWaitTimeout = 100 * time.Millisecond + + open, err := f.Fetch(testContext(t), imageURLOnPort(81)) + if err != nil { + t.Fatalf("first Fetch() error = %v", err) + } + + _, err = f.Fetch(testContext(t), imageURLOnPort(82)) + if !errors.Is(err, ErrTooManyConnections) { + t.Fatalf("Fetch() from another host: error = %v, "+ + "want ErrTooManyConnections", err) + } + + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) + defer cancel() + + _, err = f.Fetch(ctx, imageURLOnPort(81)) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Fetch() from the busy host: error = %v, "+ + "want context.DeadlineExceeded", err) + } + + // Back to the 10 second wait, so the next fetch's context ends first. + f.connectionWaitTimeout = ConnectionWaitTimeout + + ctx, cancel = context.WithTimeout(t.Context(), 100*time.Millisecond) + defer cancel() + + _, err = f.Fetch(ctx, imageURLOnPort(83)) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Fetch() from a host with nothing open: error = %v, "+ + "want context.DeadlineExceeded", err) + } + + err = open.Content.Close() + if err != nil { + t.Fatalf("close first body: %v", err) + } + + if n := hostSemCount(f); n != 0 { + t.Errorf("%d host semaphores left after every fetch finished, want 0", n) + } +} + // TestFetchReleasesConnectionOnError checks that a fetch that fails after // taking its connection gives it back: with MaxConnections at 1, the slot // must be free after the failure and the next fetch must succeed. diff --git a/internal/imgcache/eviction.go b/internal/imgcache/eviction.go index 5d185ea..9f90b8f 100644 --- a/internal/imgcache/eviction.go +++ b/internal/imgcache/eviction.go @@ -283,7 +283,7 @@ func (c *Cache) evictVariant(ctx context.Context, cacheKey VariantKey) error { c.metaCache.Remove(cacheKey) - err = c.variants.DeleteWithMeta(cacheKey) + err = c.variants.Delete(cacheKey) if err != nil { return err } diff --git a/internal/imgcache/storage.go b/internal/imgcache/storage.go index 7e1cb38..395178d 100644 --- a/internal/imgcache/storage.go +++ b/internal/imgcache/storage.go @@ -564,7 +564,8 @@ func (s *VariantStorage) Exists(key VariantKey) bool { return err == nil } -// Delete removes content at the given key. +// Delete removes the content at the given key together with its .meta +// sidecar file. A missing file is not an error. func (s *VariantStorage) Delete(key VariantKey) error { path := s.keyToPath(key) @@ -573,18 +574,7 @@ func (s *VariantStorage) Delete(key VariantKey) error { return fmt.Errorf("failed to delete content: %w", err) } - return nil -} - -// DeleteWithMeta removes the content at the given key together with -// its .meta sidecar file. A missing file is not an error. -func (s *VariantStorage) DeleteWithMeta(key VariantKey) error { - err := s.Delete(key) - if err != nil { - return err - } - - metaPath := s.keyToPath(key) + ".meta" + metaPath := path + ".meta" err = os.Remove(metaPath) if err != nil && !os.IsNotExist(err) { diff --git a/internal/imgcache/storage_internal_test.go b/internal/imgcache/storage_internal_test.go index 0465bb1..c57e1ee 100644 --- a/internal/imgcache/storage_internal_test.go +++ b/internal/imgcache/storage_internal_test.go @@ -438,3 +438,73 @@ func TestVariantStorage_StoreLogsFailedMetaWrite(t *testing.T) { t.Errorf("log missing %s; got %q", want, logBuf.String()) } } + +// storeTestVariant stores one variant, with its .meta file, in a new +// VariantStorage and returns the storage and the variant's key. +func storeTestVariant(t *testing.T) (*VariantStorage, VariantKey) { + t.Helper() + + storage, err := NewVariantStorage(t.TempDir(), slog.New(slog.DiscardHandler)) + if err != nil { + t.Fatalf("NewVariantStorage() error = %v", err) + } + + key := CacheKey(&ImageRequest{SourceHost: testHostCDN, SourcePath: testPathCat}) + + _, err = storage.Store(key, bytes.NewReader([]byte("variant data")), "image/webp") + if err != nil { + t.Fatalf("Store() error = %v", err) + } + + return storage, key +} + +// TestVariantStorage_DeleteRemovesMeta verifies that Delete removes the +// variant's .meta file along with the variant file. +func TestVariantStorage_DeleteRemovesMeta(t *testing.T) { + t.Parallel() + + storage, key := storeTestVariant(t) + metaPath := storage.keyToPath(key) + ".meta" + + _, err := os.Stat(metaPath) + if err != nil { + t.Fatalf("Store() wrote no .meta file: %v", err) + } + + err = storage.Delete(key) + if err != nil { + t.Fatalf("Delete() error = %v", err) + } + + if storage.Exists(key) { + t.Error("Exists() = true after delete, want false") + } + + _, err = os.Stat(metaPath) + if !os.IsNotExist(err) { + t.Errorf(".meta file left after Delete() (stat err=%v)", err) + } +} + +// TestVariantStorage_DeleteWithoutMeta verifies that Delete succeeds for a +// variant whose .meta file is missing. +func TestVariantStorage_DeleteWithoutMeta(t *testing.T) { + t.Parallel() + + storage, key := storeTestVariant(t) + + err := os.Remove(storage.keyToPath(key) + ".meta") + if err != nil { + t.Fatalf("removing .meta file: %v", err) + } + + err = storage.Delete(key) + if err != nil { + t.Fatalf("Delete() error = %v, want nil", err) + } + + if storage.Exists(key) { + t.Error("Exists() = true after delete, want false") + } +}