diff --git a/README.md b/README.md index c6593a2..78da010 100644 --- a/README.md +++ b/README.md @@ -207,8 +207,11 @@ These terms are the names the code uses: - the upstream answered with an error status - the body's first bytes were not those of the format its `Content-Type` names + - a check of a cached source failed in any way (see Cache TTL) - **Image cache**: the part of pixa that keeps the sources, the variants and the negative cache (see Storage). +- **Cache TTL**: how long pixa serves a cached source and its variants before it + checks the source with upstream again: 24 hours (see Cache TTL). - **Disk cache limit** (`cache_max_bytes`): the size in bytes that the cached sources and variants may reach together. - **Image proxy**: the part of pixa that serves a variant from the image cache @@ -226,10 +229,14 @@ name. The image cache keeps these tables in the database: -- `source_content`: a row for each source, by source key, with its size and when - it was last used +- `source_content`: a row for each source, by source key, with: + - its size and when it was last used + - when it was fetched, and when upstream last sent it or confirmed it + unchanged + - the `ETag` and `Last-Modified` headers upstream sent with it - `variant_content`: a row for each variant, by variant key (column - `cache_key`), with its size and when it was last used + `cache_key`), with its size, when it was last used, and when its source was + last checked with upstream - `cache_usage`: the total size of the cached sources and variants, which eviction reads - `negative_cache`: a row for each source URL in the negative cache, with when @@ -275,6 +282,36 @@ written. A time handed over beyond that is dropped. Its image then only looks less recently used to eviction. `script/bench` measures how many hits pixa serves per second (see Load Test). +### Cache TTL + +pixa serves a cached source and its variants for 24 hours after upstream last +sent the source or confirmed it unchanged. After that, the next request for an +image of that source checks the source with upstream first: + +- pixa asks upstream to send the source only if it changed. It sends the + source's `ETag` as `If-None-Match` and its `Last-Modified` as + `If-Modified-Since`, when upstream gave them. With neither, upstream sends the + whole image again. +- When upstream answers 304 Not Modified, pixa serves the cached source and its + variants for another 24 hours. Nothing is downloaded or converted again. +- When upstream sends the image, it replaces the cached source. Each variant + made from the old image is converted again from the new one when it is next + requested. +- When the check fails, pixa serves the cached image. A failure can be an error + status, a timeout or no connection. The failure puts the source URL in the + negative cache, so pixa does not ask upstream again for 5 minutes, however + often it serves the image. + +pixa evicts a source and its variants 30 days after upstream last sent the +source or confirmed it unchanged. So a source whose checks keep failing is +served for at most 30 days. Eviction to stay under the disk cache limit can +remove images sooner. + +A variant file's modification time is when its source was last checked with +upstream. So a cache hit still reads no database row to know whether the variant +is older than 24 hours. A request that checks its source with upstream counts as +a miss. + ### Routes pixa answers these routes; any other path answers 404. A path in this list asked diff --git a/TODO.md b/TODO.md index 8416fe5..6f270b3 100644 --- a/TODO.md +++ b/TODO.md @@ -30,6 +30,20 @@ P2: security: per-origin rate limiting # Completed Steps +- 2026-11-11 a cached source is checked with upstream once it is 24 hours old + (closes #69, #83). Before, it and its variants were served until eviction, so + a changed image at the same URL was never picked up. + - `source_content` keeps the source's `fetched_at`, `checked_at`, and the + `ETag` and `Last-Modified` upstream sent. + - The check sends them as `If-None-Match` and `If-Modified-Since`. A 304 + renews the source; new bytes replace it. + - A variant file's modification time is its source's `checked_at`, so a hit + still makes no database call. A variant made before the source's + `fetched_at` is converted again. + - A failed check serves the cached image and puts the source URL in the + negative cache, so upstream is asked at most once per 5 minutes. + - The eviction goroutine evicts sources and variants whose `checked_at` is + 30 days old (`CacheMaxAge`). - 2026-10-27 the reconciliation pass skips a directory it cannot list and a file whose info it cannot read, and goes on (closes #343). It stopped before its later checks at a directory it could not list, or at a source or variant file diff --git a/internal/db/migrations/001_schema.sql b/internal/db/migrations/001_schema.sql index 8aa1696..d7fe2aa 100644 --- a/internal/db/migrations/001_schema.sql +++ b/internal/db/migrations/001_schema.sql @@ -3,14 +3,24 @@ -- Source images, one row per source URL -- Files stored at: cache/sources///, where source_key --- is the SHA-256 of the source's host, path and query +-- is the SHA-256 of the source's host, path and query. +-- fetched_at is when the cached bytes were fetched. checked_at is when +-- upstream last sent them or answered that they had not changed. +-- upstream_etag and upstream_last_modified are the ETag and Last-Modified +-- headers upstream sent with the bytes, or '' when it sent none. CREATE TABLE IF NOT EXISTS source_content ( source_key TEXT PRIMARY KEY, size_bytes INTEGER NOT NULL, - last_accessed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP + last_accessed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + fetched_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + checked_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + upstream_etag TEXT NOT NULL DEFAULT '', + upstream_last_modified TEXT NOT NULL DEFAULT '' ); CREATE INDEX IF NOT EXISTS idx_source_content_last_accessed ON source_content(last_accessed_at); +CREATE INDEX IF NOT EXISTS idx_source_content_checked + ON source_content(checked_at); -- Processed variant blobs -- Files stored at: cache/variants///. Tracked here @@ -18,14 +28,19 @@ CREATE INDEX IF NOT EXISTS idx_source_content_last_accessed -- without a directory scan, and so LRU eviction has a timestamp to order -- on. A cache hit does not read this table: it takes the content type -- from the file's first bytes, and its last_accessed_at is written --- later, in a batch. +-- later, in a batch. checked_at is the checked_at of the source the +-- variant was made from, when it was made or last renewed; the variant +-- file's modification time holds the same time for cache hits. CREATE TABLE IF NOT EXISTS variant_content ( cache_key TEXT PRIMARY KEY, size_bytes INTEGER NOT NULL, - last_accessed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP + last_accessed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + checked_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_variant_content_last_accessed ON variant_content(last_accessed_at); +CREATE INDEX IF NOT EXISTS idx_variant_content_checked + ON variant_content(checked_at); -- Total cache usage: the sum of size_bytes over source_content and -- variant_content, kept by the triggers below in the same statement diff --git a/internal/handlers/client_gone_internal_test.go b/internal/handlers/client_gone_internal_test.go index 108a5a9..df91306 100644 --- a/internal/handlers/client_gone_internal_test.go +++ b/internal/handlers/client_gone_internal_test.go @@ -26,7 +26,7 @@ type heldFetcher struct { // Fetch sends on f.started, waits for f.release to be closed, and returns // httpfetcher.ErrUpstreamError. func (f heldFetcher) Fetch( - context.Context, imagerequest.Source, + context.Context, imagerequest.Source, string, string, ) (*httpfetcher.FetchResult, error) { f.started <- struct{}{} diff --git a/internal/handlers/handlers_internal_test.go b/internal/handlers/handlers_internal_test.go index de83bdf..ad0adb0 100644 --- a/internal/handlers/handlers_internal_test.go +++ b/internal/handlers/handlers_internal_test.go @@ -131,7 +131,7 @@ func newMockFetcher(fs fs.FS) *mockFetcher { } func (f *mockFetcher) Fetch( - _ context.Context, source imagerequest.Source, + _ context.Context, source imagerequest.Source, _, _ string, ) (*httpfetcher.FetchResult, error) { path := source.Host + source.Path diff --git a/internal/handlers/image_errors_internal_test.go b/internal/handlers/image_errors_internal_test.go index f4fee3c..c8c2a33 100644 --- a/internal/handlers/image_errors_internal_test.go +++ b/internal/handlers/image_errors_internal_test.go @@ -428,7 +428,7 @@ type failingFetcher struct { // Fetch returns f's error. func (f failingFetcher) Fetch( - context.Context, imagerequest.Source, + context.Context, imagerequest.Source, string, string, ) (*httpfetcher.FetchResult, error) { return nil, f.err } diff --git a/internal/handlers/referer_blocklist_internal_test.go b/internal/handlers/referer_blocklist_internal_test.go index 98691c4..88a8c28 100644 --- a/internal/handlers/referer_blocklist_internal_test.go +++ b/internal/handlers/referer_blocklist_internal_test.go @@ -36,10 +36,11 @@ type countingFetcher struct { // Fetch counts the fetch and passes it on. func (f *countingFetcher) Fetch( ctx context.Context, source imagerequest.Source, + ifNoneMatch, ifModifiedSince string, ) (*httpfetcher.FetchResult, error) { f.fetches.Add(1) - return f.Fetcher.Fetch(ctx, source) + return f.Fetcher.Fetch(ctx, source, ifNoneMatch, ifModifiedSince) } // newRefererRoutes returns both image routes of a Handlers whose diff --git a/internal/httpfetcher/conditional_internal_test.go b/internal/httpfetcher/conditional_internal_test.go new file mode 100644 index 0000000..7200204 --- /dev/null +++ b/internal/httpfetcher/conditional_internal_test.go @@ -0,0 +1,76 @@ +package httpfetcher + +import ( + "errors" + "io" + "net/http" + "net/http/httptest" + "sync" + "testing" +) + +// TestFetch_SendsConditionalHeadersAndHandlesNotModified checks that Fetch +// sends a non-empty ifNoneMatch and ifModifiedSince as request headers, that +// upstream's 304 to such a request is a result with status 304 and no +// content, and that a 304 to a request with neither is ErrUpstreamError. +func TestFetch_SendsConditionalHeadersAndHandlesNotModified(t *testing.T) { + t.Parallel() + + const ( + etag = `"abc"` + lastModified = "Mon, 02 Nov 2026 10:00:00 GMT" + ) + + var ( + mu sync.Mutex + received http.Header + ) + + srv := httptest.NewServer(http.HandlerFunc( + func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + received = r.Header.Clone() + mu.Unlock() + + w.WriteHeader(http.StatusNotModified) + })) + t.Cleanup(srv.Close) + + f, _ := newServerFetcher(t, srv, nil) + + res, err := f.Fetch(testContext(t), upstreamSource("/image"), etag, lastModified) + if err != nil { + t.Fatalf("conditional Fetch() error = %v", err) + } + + body, err := io.ReadAll(res.Content) + _ = res.Content.Close() + + mu.Lock() + t.Logf("status = %d, body = %q, request headers = %v", res.StatusCode, body, received) + gotIfNoneMatch := received.Get("If-None-Match") + gotIfModifiedSince := received.Get("If-Modified-Since") + mu.Unlock() + + if err != nil || len(body) != 0 { + t.Errorf("304 content = %q, error = %v; want empty", body, err) + } + + if res.StatusCode != http.StatusNotModified { + t.Errorf("StatusCode = %d, want 304", res.StatusCode) + } + + if gotIfNoneMatch != etag || gotIfModifiedSince != lastModified { + t.Errorf("If-None-Match = %q, If-Modified-Since = %q; want %q and %q", + gotIfNoneMatch, gotIfModifiedSince, etag, lastModified) + } + + if held := len(f.allHostsSemaphore); held != 0 { + t.Errorf("connection still held after the 304: %d held", held) + } + + _, err = f.Fetch(testContext(t), upstreamSource("/image"), "", "") + if !errors.Is(err, ErrUpstreamError) { + t.Errorf("unconditional Fetch() answered 304: error = %v, want ErrUpstreamError", err) + } +} diff --git a/internal/httpfetcher/dial_context_internal_test.go b/internal/httpfetcher/dial_context_internal_test.go index b632a1f..e40139a 100644 --- a/internal/httpfetcher/dial_context_internal_test.go +++ b/internal/httpfetcher/dial_context_internal_test.go @@ -51,12 +51,12 @@ func TestDialContextReplacesOnlyTheDialer(t *testing.T) { t.Errorf("body = %q, want %q", body, imagePayload) } - _, err := f.Fetch(testContext(t), imageSourceOn("127.0.0.1")) + _, err := f.Fetch(testContext(t), imageSourceOn("127.0.0.1"), "", "") if !errors.Is(err, ErrSSRFBlocked) { t.Errorf("Fetch(loopback URL) error = %v, want ErrSSRFBlocked", err) } - _, err = f.Fetch(testContext(t), upstreamSource("/redirect/private")) + _, err = f.Fetch(testContext(t), upstreamSource("/redirect/private"), "", "") if !errors.Is(err, ErrSSRFBlocked) { t.Errorf("Fetch(/redirect/private) error = %v, want ErrSSRFBlocked", err) } diff --git a/internal/httpfetcher/fetch_internal_test.go b/internal/httpfetcher/fetch_internal_test.go index 7c5199f..fee497b 100644 --- a/internal/httpfetcher/fetch_internal_test.go +++ b/internal/httpfetcher/fetch_internal_test.go @@ -160,7 +160,7 @@ func testContext(t *testing.T) context.Context { func fetchImage(t *testing.T, f *HTTPFetcher, path string) *FetchResult { t.Helper() - res, err := f.Fetch(testContext(t), upstreamSource(path)) + res, err := f.Fetch(testContext(t), upstreamSource(path), "", "") if err != nil { t.Fatalf("Fetch(%s) error = %v", path, err) } @@ -172,7 +172,7 @@ func fetchImage(t *testing.T, f *HTTPFetcher, path string) *FetchResult { func fetchExpectError(t *testing.T, f *HTTPFetcher, path string) error { t.Helper() - res, err := f.Fetch(testContext(t), upstreamSource(path)) + res, err := f.Fetch(testContext(t), upstreamSource(path), "", "") if err == nil { _ = res.Content.Close() @@ -224,7 +224,7 @@ func TestFetchRedirectToPrivateIPBlocked(t *testing.T) { srv := startUpstream(t) f, dialer := newServerFetcher(t, srv, nil) - _, err := f.Fetch(testContext(t), upstreamSource("/redirect/private")) + _, err := f.Fetch(testContext(t), upstreamSource("/redirect/private"), "", "") if !errors.Is(err, ErrSSRFBlocked) { t.Fatalf("Fetch() error = %v, want ErrSSRFBlocked", err) } @@ -624,7 +624,7 @@ func TestFetchSchemeFollowsAllowHTTP(t *testing.T) { DialContext: dialer.dialContext, }) - res, err := f.Fetch(testContext(t), upstreamSource("/image")) + res, err := f.Fetch(testContext(t), upstreamSource("/image"), "", "") if err == nil { _ = res.Content.Close() } diff --git a/internal/httpfetcher/httpfetcher.go b/internal/httpfetcher/httpfetcher.go index d532288..7f8a8f7 100644 --- a/internal/httpfetcher/httpfetcher.go +++ b/internal/httpfetcher/httpfetcher.go @@ -80,8 +80,13 @@ var ( // Fetcher retrieves content from upstream origins. type Fetcher interface { - // Fetch retrieves the content of the source image. - Fetch(ctx context.Context, source imagerequest.Source) (*FetchResult, error) + // Fetch retrieves the content of the source image. A non-empty + // ifNoneMatch or ifModifiedSince is sent as the request header of that + // name, and upstream may then answer 304 Not Modified with no content. + Fetch( + ctx context.Context, source imagerequest.Source, + ifNoneMatch, ifModifiedSince string, + ) (*FetchResult, error) } // FetchResult contains the result of fetching from upstream. @@ -223,9 +228,13 @@ func New(config *Config) *HTTPFetcher { // Fetch retrieves the source image with SSRF protection, over https, or over // http when AllowHTTP is set. When MaxConnections fetches are already in // progress, it waits up to ConnectionWaitTimeout for one to finish, then -// fails with ErrTooManyConnections. +// fails with ErrTooManyConnections. A non-empty ifNoneMatch or +// ifModifiedSince is sent as the If-None-Match or If-Modified-Since header. +// Upstream's 304 Not Modified to such a request is a result with StatusCode +// 304 and empty Content; to any other request it is ErrUpstreamError. func (f *HTTPFetcher) Fetch( ctx context.Context, source imagerequest.Source, + ifNoneMatch, ifModifiedSince string, ) (*FetchResult, error) { sourceURL := source.URL(f.config.AllowHTTP) @@ -252,17 +261,7 @@ func (f *HTTPFetcher) Fetch( req := &http.Request{ Method: http.MethodGet, URL: sourceURL, - Header: make(http.Header), - } - - req.Header.Set("User-Agent", userAgent) - req.Header.Set("Accept", strings.Join(allowedContentTypes(), ", ")) - - // The ID of the request this fetch serves, so the fetch can be found in - // the upstream host's logs - requestID := middleware.GetReqID(ctx) - if requestID != "" { - req.Header.Set(middleware.RequestIDHeader, requestID) + Header: requestHeaders(ctx, ifNoneMatch, ifModifiedSince), } // Use httptrace to capture connection details @@ -291,6 +290,20 @@ func (f *HTTPFetcher) Fetch( return nil, fmt.Errorf("upstream request failed: %w", err) } + // The deferred release gives the connection back, as no body follows + conditional := ifNoneMatch != "" || ifModifiedSince != "" + if conditional && resp.StatusCode == http.StatusNotModified { + _ = resp.Body.Close() + + return &FetchResult{ + Content: http.NoBody, + Headers: resp.Header, + StatusCode: resp.StatusCode, + FetchDurationMs: fetchDuration.Milliseconds(), + RemoteAddr: remoteAddr, + }, nil + } + result, err := f.buildResult(resp, remoteAddr, fetchDuration, release) if err != nil { return nil, err @@ -303,6 +316,33 @@ func (f *HTTPFetcher) Fetch( return result, nil } +// requestHeaders returns the headers of a fetch: User-Agent, Accept, the +// ID of the request the fetch serves, so the fetch can be found in the +// upstream host's logs, and a non-empty ifNoneMatch or ifModifiedSince. +func requestHeaders( + ctx context.Context, ifNoneMatch, ifModifiedSince string, +) http.Header { + header := make(http.Header) + + header.Set("User-Agent", userAgent) + header.Set("Accept", strings.Join(allowedContentTypes(), ", ")) + + requestID := middleware.GetReqID(ctx) + if requestID != "" { + header.Set(middleware.RequestIDHeader, requestID) + } + + if ifNoneMatch != "" { + header.Set("If-None-Match", ifNoneMatch) + } + + if ifModifiedSince != "" { + header.Set("If-Modified-Since", ifModifiedSince) + } + + return header +} + // acquireConnection takes a slot for host, then one of the slots shared by // all hosts, and returns the func that gives both back. The host's slot // comes first, so fetches queued for one busy host hold no shared slot. diff --git a/internal/httpfetcher/httpfetcher_internal_test.go b/internal/httpfetcher/httpfetcher_internal_test.go index e1400a2..20d8968 100644 --- a/internal/httpfetcher/httpfetcher_internal_test.go +++ b/internal/httpfetcher/httpfetcher_internal_test.go @@ -210,7 +210,7 @@ func TestMockFetcher_FetchesFile(t *testing.T) { m := NewMock(mockFS) result, err := m.Fetch(context.Background(), - imagerequest.Source{Host: testHost, Path: "/images/photo.jpg"}) + imagerequest.Source{Host: testHost, Path: "/images/photo.jpg"}, "", "") if err != nil { t.Fatalf("Fetch() error = %v", err) } @@ -242,7 +242,7 @@ func TestMockFetcher_MissingFileReturnsUpstreamError(t *testing.T) { m := NewMock(mockFS) _, err := m.Fetch(context.Background(), - imagerequest.Source{Host: testHost, Path: "/missing.jpg"}) + imagerequest.Source{Host: testHost, Path: "/missing.jpg"}, "", "") if !errors.Is(err, ErrUpstreamError) { t.Errorf("Fetch() error = %v, want ErrUpstreamError", err) } @@ -259,7 +259,7 @@ func TestMockFetcher_RespectsContextCancellation(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() - _, err := m.Fetch(ctx, imagerequest.Source{Host: testHost, Path: "/photo.jpg"}) + _, err := m.Fetch(ctx, imagerequest.Source{Host: testHost, Path: "/photo.jpg"}, "", "") if !errors.Is(err, context.Canceled) { t.Errorf("Fetch() error = %v, want context.Canceled", err) } diff --git a/internal/httpfetcher/max_connections_internal_test.go b/internal/httpfetcher/max_connections_internal_test.go index bb80228..0bc855f 100644 --- a/internal/httpfetcher/max_connections_internal_test.go +++ b/internal/httpfetcher/max_connections_internal_test.go @@ -45,12 +45,12 @@ func TestFetchLimitsConnectionsToAllHostsTogether(t *testing.T) { f, _ := newServerFetcher(t, srv, cfg) f.connectionWaitTimeout = 100 * time.Millisecond - first, err := f.Fetch(testContext(t), imageSourceOnPort(81)) + first, err := f.Fetch(testContext(t), imageSourceOnPort(81), "", "") if err != nil { t.Fatalf("first Fetch() error = %v", err) } - second, err := f.Fetch(testContext(t), imageSourceOnPort(82)) + second, err := f.Fetch(testContext(t), imageSourceOnPort(82), "", "") if err != nil { t.Fatalf("second Fetch() error = %v", err) } @@ -59,7 +59,7 @@ func TestFetchLimitsConnectionsToAllHostsTogether(t *testing.T) { start := time.Now() - _, err = f.Fetch(testContext(t), imageSourceOnPort(83)) + _, err = f.Fetch(testContext(t), imageSourceOnPort(83), "", "") if !errors.Is(err, ErrTooManyConnections) { t.Fatalf("third Fetch() error = %v, want ErrTooManyConnections", err) } @@ -78,7 +78,7 @@ func TestFetchLimitsConnectionsToAllHostsTogether(t *testing.T) { t.Fatalf("close first body: %v", err) } - third, err := f.Fetch(testContext(t), imageSourceOnPort(83)) + third, err := f.Fetch(testContext(t), imageSourceOnPort(83), "", "") if err != nil { t.Fatalf("Fetch() after a response was closed: error = %v", err) } @@ -105,7 +105,7 @@ func TestFetchFreesHostSlotWhenContextEndsWaitingForConnection(t *testing.T) { f, _ := newServerFetcher(t, srv, cfg) - first, err := f.Fetch(testContext(t), imageSourceOnPort(81)) + first, err := f.Fetch(testContext(t), imageSourceOnPort(81), "", "") if err != nil { t.Fatalf("first Fetch() error = %v", err) } @@ -115,7 +115,7 @@ func TestFetchFreesHostSlotWhenContextEndsWaitingForConnection(t *testing.T) { ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) defer cancel() - _, err = f.Fetch(ctx, imageSourceOnPort(82)) + _, err = f.Fetch(ctx, imageSourceOnPort(82), "", "") if !errors.Is(err, context.DeadlineExceeded) { t.Fatalf("second Fetch() error = %v, want context.DeadlineExceeded", err) } @@ -140,7 +140,7 @@ func TestFetchRemovesIdleHostSemaphores(t *testing.T) { for i := range 100 { wg.Go(func() { - res, err := f.Fetch(ctx, imageSourceOnPort(1+i%50)) + res, err := f.Fetch(ctx, imageSourceOnPort(1+i%50), "", "") if err != nil { t.Errorf("Fetch() error = %v", err) @@ -179,12 +179,12 @@ func TestFetchRemovesHostSemaphoreWhenNoConnection(t *testing.T) { f, _ := newServerFetcher(t, srv, cfg) f.connectionWaitTimeout = 100 * time.Millisecond - open, err := f.Fetch(testContext(t), imageSourceOnPort(81)) + open, err := f.Fetch(testContext(t), imageSourceOnPort(81), "", "") if err != nil { t.Fatalf("first Fetch() error = %v", err) } - _, err = f.Fetch(testContext(t), imageSourceOnPort(82)) + _, err = f.Fetch(testContext(t), imageSourceOnPort(82), "", "") if !errors.Is(err, ErrTooManyConnections) { t.Fatalf("Fetch() from another host: error = %v, "+ "want ErrTooManyConnections", err) @@ -193,7 +193,7 @@ func TestFetchRemovesHostSemaphoreWhenNoConnection(t *testing.T) { ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) defer cancel() - _, err = f.Fetch(ctx, imageSourceOnPort(81)) + _, err = f.Fetch(ctx, imageSourceOnPort(81), "", "") if !errors.Is(err, context.DeadlineExceeded) { t.Fatalf("Fetch() from the busy host: error = %v, "+ "want context.DeadlineExceeded", err) @@ -205,7 +205,7 @@ func TestFetchRemovesHostSemaphoreWhenNoConnection(t *testing.T) { ctx, cancel = context.WithTimeout(t.Context(), 100*time.Millisecond) defer cancel() - _, err = f.Fetch(ctx, imageSourceOnPort(83)) + _, err = f.Fetch(ctx, imageSourceOnPort(83), "", "") if !errors.Is(err, context.DeadlineExceeded) { t.Fatalf("Fetch() from a host with nothing open: error = %v, "+ "want context.DeadlineExceeded", err) @@ -255,7 +255,7 @@ func TestFetchReleasesConnectionOnError(t *testing.T) { f, _ := newServerFetcher(t, srv, cfg) f.connectionWaitTimeout = 100 * time.Millisecond - _, err := f.Fetch(testContext(t), tc.source) + _, err := f.Fetch(testContext(t), tc.source, "", "") if !errors.Is(err, tc.want) { t.Fatalf("Fetch() error = %v, want %v", err, tc.want) } diff --git a/internal/httpfetcher/mock.go b/internal/httpfetcher/mock.go index 1e3268b..5101cfc 100644 --- a/internal/httpfetcher/mock.go +++ b/internal/httpfetcher/mock.go @@ -24,9 +24,10 @@ func NewMock(fsys fs.FS) *MockFetcher { return &MockFetcher{fs: fsys} } -// Fetch retrieves the source's content from the mock filesystem. +// Fetch retrieves the source's content from the mock filesystem. It ignores +// ifNoneMatch and ifModifiedSince, and always sends the content. func (m *MockFetcher) Fetch( - ctx context.Context, source imagerequest.Source, + ctx context.Context, source imagerequest.Source, _, _ string, ) (*FetchResult, error) { // Check context cancellation select { diff --git a/internal/httpfetcher/request_id_internal_test.go b/internal/httpfetcher/request_id_internal_test.go index b6e87b8..36d4f09 100644 --- a/internal/httpfetcher/request_id_internal_test.go +++ b/internal/httpfetcher/request_id_internal_test.go @@ -35,7 +35,7 @@ func TestFetchSendsRequestID(t *testing.T) { ctx := context.WithValue(testContext(t), middleware.RequestIDKey, requestID) - res, err := f.Fetch(ctx, upstreamSource("/image")) + res, err := f.Fetch(ctx, upstreamSource("/image"), "", "") if err != nil { t.Fatalf("Fetch() error = %v", err) } diff --git a/internal/imageproxy/cache_ttl_internal_test.go b/internal/imageproxy/cache_ttl_internal_test.go new file mode 100644 index 0000000..44bb7b7 --- /dev/null +++ b/internal/imageproxy/cache_ttl_internal_test.go @@ -0,0 +1,394 @@ +package imageproxy + +import ( + "bytes" + "context" + "errors" + "fmt" + "image/color" + "io" + "net/http" + "os" + "sync" + "testing" + "time" + + "sneak.berlin/go/pixa/internal/httpfetcher" + "sneak.berlin/go/pixa/internal/imagerequest" +) + +// Validators the test upstream sends with its image. +const ( + testETag = `"v1"` + testLastModified = "Mon, 02 Nov 2026 10:00:00 GMT" +) + +// errConnectionRefused stands for an upstream that cannot be reached. +var errConnectionRefused = errors.New("connection refused") + +// conditionalUpstream serves one JPEG image with an ETag or a Last-Modified +// header, or both. It answers 304 Not Modified to a fetch whose If-None-Match +// is its ETag, or, when it has no ETag, whose If-Modified-Since is its +// Last-Modified. While err is set, every fetch fails with it. It records the +// conditional headers of every fetch. +type conditionalUpstream struct { + mu sync.Mutex + image []byte + etag string + lastModified string + err error + + ifNoneMatch []string + ifModifiedSince []string +} + +func (u *conditionalUpstream) Fetch( + _ context.Context, _ imagerequest.Source, + ifNoneMatch, ifModifiedSince string, +) (*httpfetcher.FetchResult, error) { + u.mu.Lock() + defer u.mu.Unlock() + + u.ifNoneMatch = append(u.ifNoneMatch, ifNoneMatch) + u.ifModifiedSince = append(u.ifModifiedSince, ifModifiedSince) + + if u.err != nil { + return nil, u.err + } + + unchanged := ifNoneMatch == u.etag + if u.etag == "" { + unchanged = ifModifiedSince == u.lastModified + } + + if unchanged { + return &httpfetcher.FetchResult{ + Content: http.NoBody, + StatusCode: http.StatusNotModified, + }, nil + } + + headers := http.Header{} + if u.etag != "" { + headers.Set("ETag", u.etag) + } + + if u.lastModified != "" { + headers.Set("Last-Modified", u.lastModified) + } + + return &httpfetcher.FetchResult{ + Content: io.NopCloser(bytes.NewReader(u.image)), + ContentLength: int64(len(u.image)), + ContentType: testContentTypeJPEG, + Headers: headers, + StatusCode: http.StatusOK, + }, nil +} + +// change replaces the upstream image and its ETag. +func (u *conditionalUpstream) change(image []byte, etag string) { + u.mu.Lock() + defer u.mu.Unlock() + + u.image = image + u.etag = etag +} + +// fail makes every later fetch fail with err. +func (u *conditionalUpstream) fail(err error) { + u.mu.Lock() + defer u.mu.Unlock() + + u.err = err +} + +// fetches returns how many fetches upstream was asked for. +func (u *conditionalUpstream) fetches() int { + u.mu.Lock() + defer u.mu.Unlock() + + return len(u.ifNoneMatch) +} + +// lastConditionalHeaders returns the If-None-Match and If-Modified-Since of +// the last fetch. +func (u *conditionalUpstream) lastConditionalHeaders() (string, string) { + u.mu.Lock() + defer u.mu.Unlock() + + last := len(u.ifNoneMatch) - 1 + + return u.ifNoneMatch[last], u.ifModifiedSince[last] +} + +// setupConditionalUpstreamProxy returns a proxy whose upstream is a +// conditionalUpstream serving a red JPEG with an ETag. +func setupConditionalUpstreamProxy( + t *testing.T, +) (*testProxy, *TestFixtures, *conditionalUpstream) { + t.Helper() + + svc, fixtures := setupTestProxy(t) + upstream := &conditionalUpstream{ + image: generateTestJPEG(t, 100, 100, color.RGBA{255, 0, 0, 255}), + etag: testETag, + } + svc.fetcher = upstream + + return svc, fixtures, upstream +} + +// TestProxy_Get_FreshVariantAsksUpstreamNothing checks that a variant whose +// source was checked with upstream less than the cache TTL ago is served from +// the cache without asking upstream. +func TestProxy_Get_FreshVariantAsksUpstreamNothing(t *testing.T) { + t.Parallel() + + svc, fixtures, upstream := setupConditionalUpstreamProxy(t) + variant := photoVariant(fixtures, 85, imagerequest.FitCover) + + first, _ := getImage(t, svc, variant) + second, status := getImage(t, svc, variant) + + t.Logf("cache status = %s, upstream fetches = %d", status, upstream.fetches()) + + if status != CacheHit { + t.Errorf("CacheStatus = %q, want %q", status, CacheHit) + } + + if !bytes.Equal(first, second) { + t.Error("the cached variant differs from the one first served") + } + + if n := upstream.fetches(); n != 1 { + t.Errorf("upstream fetches = %d, want 1", n) + } +} + +// TestProxy_Get_UnchangedSourceAfterTTLIsNotDownloadedAgain caches a variant, +// sets its source's check time and the variant's file time two days back, +// and requests it again. Upstream must be asked with the validator it sent, +// answer 304, and send no bytes; the variant must be served again without +// being made again, and then be served from the cache until the TTL passes. +func TestProxy_Get_UnchangedSourceAfterTTLIsNotDownloadedAgain(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + etag string + lastModified string + wantIfNoneMatch string + wantIfModifiedSince string + }{ + {"ETag", testETag, "", testETag, ""}, + {"Last-Modified only", "", testLastModified, "", testLastModified}, + {"both", testETag, testLastModified, testETag, testLastModified}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + svc, fixtures, upstream := setupConditionalUpstreamProxy(t) + upstream.etag = tt.etag + upstream.lastModified = tt.lastModified + variant := photoVariant(fixtures, 85, imagerequest.FitCover) + + first, _ := getImage(t, svc, variant) + setBackSourceAndVariant(t, svc, variant, time.Now().AddDate(0, 0, -2)) + + second, status := getImage(t, svc, variant) + + ifNoneMatch, ifModifiedSince := upstream.lastConditionalHeaders() + counters := readCounters(t, svc) + t.Logf("cache status = %s, If-None-Match = %q, If-Modified-Since = %q, "+ + "counters = %+v", status, ifNoneMatch, ifModifiedSince, counters) + + if ifNoneMatch != tt.wantIfNoneMatch { + t.Errorf("If-None-Match = %q, want %q", ifNoneMatch, tt.wantIfNoneMatch) + } + + if ifModifiedSince != tt.wantIfModifiedSince { + t.Errorf("If-Modified-Since = %q, want %q", + ifModifiedSince, tt.wantIfModifiedSince) + } + + if !bytes.Equal(first, second) { + t.Error("the variant served after the check differs from the cached one") + } + + if counters.upstreamFetchCount != 1 || counters.transformCount != 1 { + t.Errorf("upstream downloads = %d, transforms = %d; want 1 and 1", + counters.upstreamFetchCount, counters.transformCount) + } + + _, status = getImage(t, svc, variant) + if status != CacheHit { + t.Errorf("CacheStatus after the check = %q, want %q", status, CacheHit) + } + + if n := upstream.fetches(); n != 2 { + t.Errorf("upstream fetches = %d, want 2", n) + } + }) + } +} + +// TestProxy_Get_ChangedSourceAfterTTLReplacesSourceAndVariants caches two +// variants of one source, changes the image upstream, and sets the source's +// check time and both variants' file times two days back. Each variant must +// then be made again from the new image, which is downloaded once. +func TestProxy_Get_ChangedSourceAfterTTLReplacesSourceAndVariants(t *testing.T) { + t.Parallel() + + svc, fixtures, upstream := setupConditionalUpstreamProxy(t) + variants := []*imagerequest.Request{ + photoVariant(fixtures, 85, imagerequest.FitCover), + photoVariant(fixtures, 40, imagerequest.FitCover), + } + + oldImages := make([][]byte, len(variants)) + for i, variant := range variants { + oldImages[i], _ = getImage(t, svc, variant) + } + + newSource := generateTestJPEG(t, 120, 90, color.RGBA{0, 0, 255, 255}) + upstream.change(newSource, `"v2"`) + + twoDaysAgo := time.Now().AddDate(0, 0, -2) + for _, variant := range variants { + setBackSourceAndVariant(t, svc, variant, twoDaysAgo) + } + + for i, variant := range variants { + image, _ := getImage(t, svc, variant) + if bytes.Equal(image, oldImages[i]) { + t.Errorf("variant %d is still made from the old image", i) + } + } + + counters := readCounters(t, svc) + t.Logf("counters = %+v, upstream fetches = %d", counters, upstream.fetches()) + + if counters.upstreamFetchCount != 2 { + t.Errorf("upstream downloads = %d, want 2", counters.upstreamFetchCount) + } + + cached, err := os.ReadFile(sourceFilePath(svc, variants[0].Source.Key())) + if err != nil { + t.Fatalf("reading the cached source: %v", err) + } + + if !bytes.Equal(cached, newSource) { + t.Error("the cached source is not the new image") + } +} + +// TestProxy_Get_UpstreamFailureAfterTTLServesCachedImage caches a variant, +// sets its source's check time and the variant's file time two days back, and +// makes upstream fail. The cached variant must still be served, and a variant +// not yet cached must be made from the cached source. +func TestProxy_Get_UpstreamFailureAfterTTLServesCachedImage(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + }{ + {"error status", fmt.Errorf("%w: status 503", httpfetcher.ErrUpstreamError)}, + {"timeout", httpfetcher.ErrUpstreamTimeout}, + {"no connection", errConnectionRefused}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + svc, fixtures, upstream := setupConditionalUpstreamProxy(t) + variant := photoVariant(fixtures, 85, imagerequest.FitCover) + + first, _ := getImage(t, svc, variant) + setBackSourceAndVariant(t, svc, variant, time.Now().AddDate(0, 0, -2)) + upstream.fail(tt.err) + + second, _ := getImage(t, svc, variant) + t.Logf("upstream fetches = %d", upstream.fetches()) + + if upstream.fetches() != 2 { + t.Errorf("upstream fetches = %d, want 2", upstream.fetches()) + } + + if !bytes.Equal(first, second) { + t.Error("the variant served while upstream fails differs from the cached one") + } + + getImage(t, svc, photoVariant(fixtures, 40, imagerequest.FitCover)) + }) + } +} + +// TestProxy_Get_FailingUpstreamIsAskedOnceWhileImageIsServedOften caches a +// variant, sets its source's check time and the variant's file time two days +// back, and makes every fetch time out. Serving the variant ten times, and a +// variant not yet cached, must ask upstream once: the failed check puts the +// source URL in the negative cache. +func TestProxy_Get_FailingUpstreamIsAskedOnceWhileImageIsServedOften(t *testing.T) { + t.Parallel() + + svc, fixtures, upstream := setupConditionalUpstreamProxy(t) + variant := photoVariant(fixtures, 85, imagerequest.FitCover) + + getImage(t, svc, variant) + setBackSourceAndVariant(t, svc, variant, time.Now().AddDate(0, 0, -2)) + upstream.fail(httpfetcher.ErrUpstreamTimeout) + + for range 10 { + getImage(t, svc, variant) + } + + getImage(t, svc, photoVariant(fixtures, 40, imagerequest.FitCover)) + + t.Logf("upstream fetches = %d", upstream.fetches()) + + if n := upstream.fetches(); n != 2 { + t.Errorf("upstream fetches = %d, want 2: the first fetch and one check", n) + } +} + +// setBackSourceAndVariant sets back to when the times the cache keeps for +// req's source and for the variant req asks for: the source's fetch and check +// times, both last access times, the variant's check time, and the variant +// file's modification time. +func setBackSourceAndVariant( + t *testing.T, svc *testProxy, req *imagerequest.Request, when time.Time, +) { + t.Helper() + + key := req.Source.Key() + setSourceLastAccessed(t, svc.db, key, when) + + _, err := svc.db.ExecContext(t.Context(), + `UPDATE source_content SET fetched_at = ?, checked_at = ? + WHERE source_key = ?`, when.UTC(), when.UTC(), string(key)) + if err != nil { + t.Fatalf("setting back the source's fetch and check times: %v", err) + } + + variantKey := imagerequest.CacheKey(req) + setVariantLastAccessed(t, svc.db, variantKey, when) + + _, err = svc.db.ExecContext(t.Context(), + `UPDATE variant_content SET checked_at = ? WHERE cache_key = ?`, + when.UTC(), string(variantKey)) + if err != nil { + t.Fatalf("setting back the variant's check time: %v", err) + } + + path := variantFilePath(svc, variantKey) + + err = os.Chtimes(path, when, when) + if err != nil { + t.Fatalf("setting back the times of %s: %v", path, err) + } +} diff --git a/internal/imageproxy/concurrent_misses_internal_test.go b/internal/imageproxy/concurrent_misses_internal_test.go index 4c7409f..a5c5e03 100644 --- a/internal/imageproxy/concurrent_misses_internal_test.go +++ b/internal/imageproxy/concurrent_misses_internal_test.go @@ -37,6 +37,7 @@ type heldFetcher struct { func (f *heldFetcher) Fetch( ctx context.Context, source imagerequest.Source, + ifNoneMatch, ifModifiedSince string, ) (*httpfetcher.FetchResult, error) { f.fetches.Add(1) @@ -48,7 +49,7 @@ func (f *heldFetcher) Fetch( return nil, ctx.Err() } - return f.upstream.Fetch(ctx, source) + return f.upstream.Fetch(ctx, source, ifNoneMatch, ifModifiedSince) } // releaseFetches lets every held fetch, and every later one, go on. @@ -489,7 +490,7 @@ func TestProxy_Get_WaiterGetsProcessingTimedOut(t *testing.T) { type panickingFetcher struct{} func (panickingFetcher) Fetch( - context.Context, imagerequest.Source, + context.Context, imagerequest.Source, string, string, ) (*httpfetcher.FetchResult, error) { panic("upstream fetcher panicked") } diff --git a/internal/imageproxy/imageproxy.go b/internal/imageproxy/imageproxy.go index ae17947..fe62e49 100644 --- a/internal/imageproxy/imageproxy.go +++ b/internal/imageproxy/imageproxy.go @@ -10,6 +10,7 @@ import ( "fmt" "io" "log/slog" + "net/http" "runtime/debug" "time" @@ -157,7 +158,9 @@ const ( // // A cache hit opens the variant file and makes no database call: the cache // hands its access time to the goroutine that writes access times, without -// waiting for it, and its hit is counted in memory. A variant file that +// waiting for it, and its hit is counted in memory. A variant whose source +// was last checked with upstream CacheTTL or more ago is a miss, which checks +// the source first (see processFromSourceOrFetch). A variant file that // cannot be served, such as one whose first bytes are not those of an output // format, is a miss, and the variant is made again. func (p *Proxy) Get( @@ -178,7 +181,7 @@ func (p *Proxy) Get( }, nil } - if !errors.Is(err, imgcache.ErrNotFound) { + if !errors.Is(err, imgcache.ErrNotFound) && !errors.Is(err, imgcache.ErrStale) { p.log.Warn("cached variant not served, making it again", "cache_key", cacheKey, "error", err) } @@ -337,13 +340,56 @@ func (p *Proxy) loadCachedSource( return reader, size, format } -// processFromSourceOrFetch processes an image, using cached source content -// if available. +// processFromSourceOrFetch makes the variant req asks for, from the cached +// source when there is one and from a fetched source otherwise. +// +// A cached source last checked with upstream CacheTTL or more ago is checked +// first: upstream is asked to send it only if it changed since. When it +// changed, the new bytes replace the cached source and the variant is made +// from them. When it did not change, or the check fails, the cached source +// is used: the variant's cached file is served again if it was made from the +// cached source's bytes, and otherwise the variant is made from them. func (p *Proxy) processFromSourceOrFetch( ctx context.Context, req *imagerequest.Request, cacheKey imagerequest.VariantKey, ) (*processedVariant, error) { + cached, err := p.cache.LookupSource(ctx, req.Source) + if err != nil { + if !errors.Is(err, imgcache.ErrNotFound) { + p.log.Warn("failed to look up cached source, fetching", "error", err) + } + + return p.fetchAndProcess(ctx, req, cacheKey) + } + + if p.cache.Stale(cached.CheckedAt) { + fetched, err := p.fetchSource(ctx, req, cached.ETag, cached.LastModified) + + switch { + case errors.Is(err, errSourceNotModified): + cached.CheckedAt = time.Now().UTC() + p.renewSource(ctx, req.Source, cached.CheckedAt) + case err != nil: + // Any failure, a timeout included, puts the source URL in the + // negative cache, so upstream is asked at most once per + // NegativeTTL while it fails, however often the image is served + if !errors.Is(err, ErrNegativeCached) { + p.log.Warn("checking source with upstream failed, using cached source", + "request_id", middleware.GetReqID(ctx), + "host", req.Source.Host, "path", req.Source.Path, "error", err) + p.storeNegative(ctx, req) + } + default: + return p.processFetched(ctx, req, cacheKey, fetched) + } + } + + variant := p.renewVariant(ctx, cacheKey, cached) + if variant != nil { + return variant, nil + } + source, sourceSize, sourceFormat := p.loadCachedSource(ctx, req.Source) // Fetch from upstream if we don't have source data or it's empty @@ -356,18 +402,113 @@ func (p *Proxy) processFromSourceOrFetch( // Process using cached source; nothing was fetched from upstream. The // image processor reads the source only once it has a processing slot, // so a request waiting for one holds none of it in memory. - return p.processAndStore(ctx, req, cacheKey, source, sourceFormat, sourceSize) + return p.processAndStore(ctx, req, cacheKey, cached.CheckedAt, + source, sourceFormat, sourceSize) } -// fetchAndProcess fetches from upstream, processes, and caches the result. -// It counts the fetch with the bytes read from upstream, including when -// reading the response or a later step fails. It fetches nothing while the -// source URL is in the negative cache. +// renewSource records that upstream answered at checkedAt that src's cached +// source had not changed. A failure to record it is only logged: the source +// is then checked again on its next use. +func (p *Proxy) renewSource( + ctx context.Context, src imagerequest.Source, checkedAt time.Time, +) { + err := p.cache.RenewSource(ctx, src, checkedAt) + if err != nil { + p.log.Warn("failed to renew cached source", "source_key", src.Key(), + "error", err) + } +} + +// renewVariant returns the cached variant of cacheKey when it was made from +// the bytes of the cached source that source describes, and renews it to +// source.CheckedAt (see imgcache.Cache.RenewVariant). It returns nil when +// there is no such variant or it cannot be read. +func (p *Proxy) renewVariant( + ctx context.Context, cacheKey imagerequest.VariantKey, + source imgcache.SourceInfo, +) *processedVariant { + reader, _, contentType, err := p.cache.RenewVariant(ctx, cacheKey, source) + if err != nil { + if !errors.Is(err, imgcache.ErrNotFound) { + p.log.Warn("cached variant not renewed, making it again", + "cache_key", cacheKey, "error", err) + } + + return nil + } + + defer func() { _ = reader.Close() }() + + data, err := io.ReadAll(reader) + if err != nil { + p.log.Warn("failed to read cached variant, making it again", + "cache_key", cacheKey, "error", err) + + return nil + } + + return &processedVariant{data: data, contentType: contentType} +} + +// fetchAndProcess fetches the source from upstream, stores it, and makes +// and stores the variant from it. func (p *Proxy) fetchAndProcess( ctx context.Context, req *imagerequest.Request, cacheKey imagerequest.VariantKey, ) (*processedVariant, error) { + fetched, err := p.fetchSource(ctx, req, "", "") + if err != nil { + return nil, err + } + + return p.processFetched(ctx, req, cacheKey, fetched) +} + +// processFetched makes and stores the variant from the source fetchSource +// fetched, then closes the fetch's response. +func (p *Proxy) processFetched( + ctx context.Context, + req *imagerequest.Request, + cacheKey imagerequest.VariantKey, + fetched *fetchedSource, +) (*processedVariant, error) { + // Closing the response frees the upstream connection. It is closed only + // after processing, so the fetcher's connection limit also bounds the + // fetched sources held in memory while their requests wait for a + // processing slot. + defer func() { _ = fetched.response.Close() }() + + return p.processAndStore(ctx, req, cacheKey, fetched.info.CheckedAt, + bytes.NewReader(fetched.data), fetched.format, int64(len(fetched.data))) +} + +// errSourceNotModified is returned by fetchSource when upstream answers 304 +// Not Modified: the cached source has not changed. +var errSourceNotModified = errors.New("upstream source not modified") + +// fetchedSource is a source fetchSource fetched and stored in the cache. +type fetchedSource struct { + data []byte + format imagerequest.Format + // info is what the cache keeps about the source + info imgcache.SourceInfo + // response is the fetch's response, still open; closing it frees the + // upstream connection + response io.Closer +} + +// fetchSource fetches req's source from upstream and stores it in the cache. +// A non-empty ifNoneMatch or ifModifiedSince asks upstream to send the source +// only if it changed since; errSourceNotModified says it did not. It counts +// the fetch with the bytes read from upstream, including when reading the +// response or a later step fails. It fetches nothing while the source URL is +// in the negative cache. The caller closes the returned response. +func (p *Proxy) fetchSource( + ctx context.Context, + req *imagerequest.Request, + ifNoneMatch, ifModifiedSince string, +) (*fetchedSource, error) { // Check negative cache - skip fetching for recently-failed URLs negHit, err := p.cache.Failure(ctx, req.Source) if err != nil { @@ -387,7 +528,8 @@ func (p *Proxy) fetchAndProcess( p.log.Debug("fetching from upstream", "host", req.Source.Host, "path", req.Source.Path, "query", req.Source.Query) - fetchResult, err := p.fetcher.Fetch(ctx, req.Source) + fetchResult, err := p.fetcher.Fetch(ctx, req.Source, + ifNoneMatch, ifModifiedSince) if err != nil { // Store negative cache for certain errors if isNegativeCacheable(err) { @@ -397,11 +539,15 @@ func (p *Proxy) fetchAndProcess( return nil, fmt.Errorf("upstream fetch failed: %w", err) } - // Closing the body frees the upstream connection. It is closed only - // after processing, so the fetcher's connection limit also bounds the - // fetched sources held in memory while their requests wait for a - // processing slot. - defer func() { _ = fetchResult.Content.Close() }() + if fetchResult.StatusCode == http.StatusNotModified { + _ = fetchResult.Content.Close() + + p.log.Info("upstream source not modified", + "request_id", middleware.GetReqID(ctx), "host", req.Source.Host, + "path", req.Source.Path, "fetch_ms", fetchResult.FetchDurationMs) + + return nil, errSourceNotModified + } // Read and validate the source content sourceData, err := io.ReadAll(fetchResult.Content) @@ -411,31 +557,12 @@ func (p *Proxy) fetchAndProcess( p.incrementUpstreamFetch(fetchBytes) if err != nil { + _ = fetchResult.Content.Close() + return nil, fmt.Errorf("failed to read upstream response: %w", err) } - // Calculate download bitrate - var downloadRate string - - if fetchResult.FetchDurationMs > 0 { - seconds := float64(fetchResult.FetchDurationMs) / 1000.0 //nolint:mnd // ms to seconds - bitsPerSecond := float64(fetchBytes*8) / seconds //nolint:mnd // bytes to bits - downloadRate = humanize.SI(bitsPerSecond, "bps") - } - - // Log upstream fetch details - p.log.Info("upstream fetched", - "request_id", middleware.GetReqID(ctx), - "host", req.Source.Host, - "path", req.Source.Path, - "bytes", fetchBytes, - "fetch_ms", fetchResult.FetchDurationMs, - "rate", downloadRate, - "remote_addr", fetchResult.RemoteAddr, - "http", fetchResult.HTTPVersion, - "tls", fetchResult.TLSVersion, - "cipher", fetchResult.TLSCipherSuite, - ) + p.logUpstreamFetch(ctx, req, fetchResult, fetchBytes) // The source's format, detected from its first bytes, which must match // its content type. A body that does not is the upstream's error, so it @@ -448,18 +575,70 @@ func (p *Proxy) fetchAndProcess( "content_type", fetchResult.ContentType, "error", err) p.storeNegative(ctx, req) + _ = fetchResult.Content.Close() + return nil, fmt.Errorf("content validation failed: %w", err) } + info := fetchedSourceInfo(fetchResult) + // Store source content - _, err = p.cache.PutSource(ctx, req.Source, bytes.NewReader(sourceData)) + _, err = p.cache.PutSource(ctx, req.Source, bytes.NewReader(sourceData), info) if err != nil { p.log.Warn("failed to store source content", "error", err) // Continue even if caching fails } - return p.processAndStore( - ctx, req, cacheKey, bytes.NewReader(sourceData), sourceFormat, fetchBytes, + return &fetchedSource{ + data: sourceData, + format: sourceFormat, + info: info, + response: fetchResult.Content, + }, nil +} + +// fetchedSourceInfo returns what the cache keeps about a source upstream +// just sent: fetched and checked now, with the ETag and Last-Modified +// headers of fetchResult. +func fetchedSourceInfo(fetchResult *httpfetcher.FetchResult) imgcache.SourceInfo { + now := time.Now().UTC() + upstreamHeaders := http.Header(fetchResult.Headers) + + return imgcache.SourceInfo{ + FetchedAt: now, + CheckedAt: now, + ETag: upstreamHeaders.Get("ETag"), + LastModified: upstreamHeaders.Get("Last-Modified"), + } +} + +// logUpstreamFetch logs a fetch of req's source that read fetchBytes bytes, +// with its download rate and connection details. +func (p *Proxy) logUpstreamFetch( + ctx context.Context, + req *imagerequest.Request, + fetchResult *httpfetcher.FetchResult, + fetchBytes int64, +) { + var downloadRate string + + if fetchResult.FetchDurationMs > 0 { + seconds := float64(fetchResult.FetchDurationMs) / 1000.0 //nolint:mnd // ms to seconds + bitsPerSecond := float64(fetchBytes*8) / seconds //nolint:mnd // bytes to bits + downloadRate = humanize.SI(bitsPerSecond, "bps") + } + + p.log.Info("upstream fetched", + "request_id", middleware.GetReqID(ctx), + "host", req.Source.Host, + "path", req.Source.Path, + "bytes", fetchBytes, + "fetch_ms", fetchResult.FetchDurationMs, + "rate", downloadRate, + "remote_addr", fetchResult.RemoteAddr, + "http", fetchResult.HTTPVersion, + "tls", fetchResult.TLSVersion, + "cipher", fetchResult.TLSCipherSuite, ) } @@ -494,11 +673,13 @@ type processedVariant struct { } // processAndStore processes the image read from source, which is in -// sourceFormat, and stores the result. +// sourceFormat, and stores the result. sourceCheckedAt is when the source +// was last checked with upstream (see imgcache.Cache.PutVariant). func (p *Proxy) processAndStore( ctx context.Context, req *imagerequest.Request, cacheKey imagerequest.VariantKey, + sourceCheckedAt time.Time, source io.Reader, sourceFormat imagerequest.Format, fetchBytes int64, @@ -559,7 +740,8 @@ func (p *Proxy) processAndStore( ) // Store variant to cache - err = p.cache.PutVariant(ctx, cacheKey, bytes.NewReader(processedData)) + err = p.cache.PutVariant(ctx, cacheKey, sourceCheckedAt, + bytes.NewReader(processedData)) if err != nil { p.log.Warn("failed to store variant", "error", err) // Continue even if caching fails diff --git a/internal/imageproxy/imageproxy_internal_test.go b/internal/imageproxy/imageproxy_internal_test.go index be2e46a..78a03eb 100644 --- a/internal/imageproxy/imageproxy_internal_test.go +++ b/internal/imageproxy/imageproxy_internal_test.go @@ -3,9 +3,7 @@ package imageproxy import ( "context" "io" - "os" "testing" - "time" "sneak.berlin/go/pixa/internal/imagerequest" "sneak.berlin/go/pixa/internal/magic" @@ -263,97 +261,6 @@ func TestProxy_Get_Caching(t *testing.T) { } } -// TestProxy_Get_ServesSourceAndVariantCachedTwoDaysAgo caches a source and -// a variant of it and sets back two days every time pixa stored for them. -// For the source: last_accessed_at in source_content, and the access and -// modification times of its file. For the variant: last_accessed_at in -// variant_content, and the access and modification times of its file. It -// then checks that the variant is still served from the cache and that a new -// variant is made from the cached source without fetching it again. -// -// Left at the present: the files' status change and birth times, which a -// program cannot set, and the times of the directories holding the files, -// which other entries share. -func TestProxy_Get_ServesSourceAndVariantCachedTwoDaysAgo(t *testing.T) { - t.Parallel() - - svc, fixtures := setupTestProxy(t) - variant := photoVariant(fixtures, 85, imagerequest.FitCover) - - resp, err := svc.Get(t.Context(), variant) - if err != nil { - t.Fatalf("Get() first request error = %v", err) - } - - _ = resp.Content.Close() - - twoDaysAgo := time.Now().AddDate(0, 0, -2) - backdateSource(t, svc, variant, twoDaysAgo) - backdateVariant(t, svc, variant, twoDaysAgo) - - resp, err = svc.Get(t.Context(), variant) - if err != nil { - t.Fatalf("Get() of the cached variant error = %v", err) - } - - _ = resp.Content.Close() - - if resp.CacheStatus != CacheHit { - t.Errorf("cached variant CacheStatus = %q, want %q", - resp.CacheStatus, CacheHit) - } - - // Another quality is a new variant, made from the cached source - resp, err = svc.Get(t.Context(), photoVariant(fixtures, 40, imagerequest.FitCover)) - if err != nil { - t.Fatalf("Get() of a new variant error = %v", err) - } - - _ = resp.Content.Close() - - fetches := readCounters(t, svc).upstreamFetchCount - if fetches != 1 { - t.Errorf("upstream fetches = %d, want 1", fetches) - } -} - -// backdateSource sets every time pixa stored for req's cached source to when, -// as listed on TestProxy_Get_ServesSourceAndVariantCachedTwoDaysAgo. -func backdateSource( - t *testing.T, svc *testProxy, req *imagerequest.Request, when time.Time, -) { - t.Helper() - - key := req.Source.Key() - setSourceLastAccessed(t, svc.db, key, when) - - path := sourceFilePath(svc, key) - - err := os.Chtimes(path, when, when) - if err != nil { - t.Fatalf("setting back the times of %s: %v", path, err) - } -} - -// backdateVariant sets every time pixa stored for the cached variant req asks -// for to when, as listed on -// TestProxy_Get_ServesSourceAndVariantCachedTwoDaysAgo. -func backdateVariant( - t *testing.T, svc *testProxy, req *imagerequest.Request, when time.Time, -) { - t.Helper() - - key := imagerequest.CacheKey(req) - path := variantFilePath(svc, key) - - setVariantLastAccessed(t, svc.db, key, when) - - err := os.Chtimes(path, when, when) - if err != nil { - t.Fatalf("setting back the times of %s: %v", path, err) - } -} - func TestProxy_Get_DifferentSizes(t *testing.T) { t.Parallel() diff --git a/internal/imageproxy/stats_internal_test.go b/internal/imageproxy/stats_internal_test.go index 14fa7cf..c9cda93 100644 --- a/internal/imageproxy/stats_internal_test.go +++ b/internal/imageproxy/stats_internal_test.go @@ -94,7 +94,7 @@ type fakeUpstream struct { } func (u *fakeUpstream) Fetch( - context.Context, imagerequest.Source, + context.Context, imagerequest.Source, string, string, ) (*httpfetcher.FetchResult, error) { return &httpfetcher.FetchResult{ Content: io.NopCloser(u), diff --git a/internal/imgcache/cache.go b/internal/imgcache/cache.go index ab8a5d2..bf2dc9f 100644 --- a/internal/imgcache/cache.go +++ b/internal/imgcache/cache.go @@ -13,7 +13,9 @@ import ( "errors" "fmt" "io" + "io/fs" "log/slog" + "os" "path/filepath" "time" @@ -38,11 +40,30 @@ const cleanExpiredBatchSize = 1000 // cache. const DefaultNegativeTTL = 5 * time.Minute +// DefaultCacheTTL is how long after upstream last sent a cached source, or +// answered that it had not changed, pixad serves the source and its +// variants without asking upstream again. +const DefaultCacheTTL = 24 * time.Hour + +// CacheMaxAge is how long after upstream last sent a cached source, or +// answered that it had not changed, the source and its variants are evicted +// (see EvictPastMaxAge). Until then a source whose checks fail is served. +const CacheMaxAge = 30 * 24 * time.Hour + +// ErrStale is returned by Variant for a variant whose source was last +// checked with upstream CacheTTL or more ago. +var ErrStale = errors.New("cached variant is older than the cache TTL") + // Config holds cache configuration. type Config struct { StateDir string NegativeTTL time.Duration + // CacheTTL is how long a cached source and its variants are served + // before upstream is asked again whether the source changed. Zero means + // DefaultCacheTTL. + CacheTTL time.Duration + // MaxBytes is the disk cache size limit in bytes that eviction // enforces. Zero turns the disk cache off entirely: no cache // directories are created, lookups always miss, stores are no-ops, @@ -113,6 +134,10 @@ func NewCache( log = slog.Default() } + if config.CacheTTL == 0 { + config.CacheTTL = DefaultCacheTTL + } + c := &Cache{ db: db, config: config, @@ -158,11 +183,13 @@ func NewCache( // Variant opens a cached variant and returns a reader, its size and its // content type, read from the variant file's first bytes. It opens that one -// file and makes no database call. A variant it opens has its access time -// handed to the background writer without waiting (see touchVariant). A -// variant not in the cache, or any variant of a disabled cache, is -// ErrNotFound; a variant file whose first bytes are not those of an output -// format is another error. +// file and makes no database call: the file's modification time is when its +// source was last checked with upstream (see PutVariant), and a variant +// whose source was checked CacheTTL or more ago is ErrStale. A variant it +// opens has its access time handed to the background writer without waiting +// (see touchVariant). A variant not in the cache, or any variant of a +// disabled cache, is ErrNotFound; a variant file whose first bytes are not +// those of an output format is another error. func (c *Cache) Variant( cacheKey imagerequest.VariantKey, ) (io.ReadCloser, int64, string, error) { @@ -170,11 +197,17 @@ func (c *Cache) Variant( return nil, 0, "", ErrNotFound } - file, size, err := c.variants.Open(string(cacheKey)) + file, info, err := c.openVariant(cacheKey) if err != nil { return nil, 0, "", err } + if c.Stale(info.ModTime()) { + _ = file.Close() + + return nil, 0, "", ErrStale + } + format, err := readFormat(file) if err != nil { _ = file.Close() @@ -184,15 +217,81 @@ func (c *Cache) Variant( c.touchVariant(cacheKey) - return file, size, format.MIMEType(), nil + return file, info.Size(), format.MIMEType(), nil } -// PutVariant stores a processed variant by its cache key and records -// it in the size accounting. On a disabled cache it is a no-op. The -// accounting insert is best-effort (the startup reconciliation pass -// adopts any variant file that misses its accounting row). +// RenewVariant opens a cached variant made from the source bytes that source +// describes, whatever its age, and returns it as Variant does. It moves the +// variant file's modification time, and the checked_at of its row, forward +// to source.CheckedAt, so Variant serves the file until CacheTTL after that +// check and eviction keeps it until CacheMaxAge after it. A variant made +// before source.FetchedAt was made from older bytes, and is ErrNotFound, as +// is a variant not in the cache. +func (c *Cache) RenewVariant( + ctx context.Context, cacheKey imagerequest.VariantKey, source SourceInfo, +) (io.ReadCloser, int64, string, error) { + if c.config.MaxBytes == 0 { + return nil, 0, "", ErrNotFound + } + + file, info, err := c.openVariant(cacheKey) + if err != nil { + return nil, 0, "", err + } + + if info.ModTime().Before(source.FetchedAt) { + _ = file.Close() + + return nil, 0, "", ErrNotFound + } + + format, err := readFormat(file) + if err != nil { + _ = file.Close() + + return nil, 0, "", err + } + + if info.ModTime().Before(source.CheckedAt) { + err = c.variants.SetModTime(string(cacheKey), source.CheckedAt) + if err != nil { + _ = file.Close() + + return nil, 0, "", err + } + + _, err = c.db.ExecContext(ctx, + `UPDATE variant_content SET checked_at = ? WHERE cache_key = ?`, + source.CheckedAt.UTC(), string(cacheKey)) + if err != nil { + c.log.Warn("failed to renew variant's checked_at", + "cache_key", cacheKey, "error", err) + } + } + + c.touchVariant(cacheKey) + + return file, info.Size(), format.MIMEType(), nil +} + +// Stale reports whether a source last checked with upstream at checkedAt, +// and its variants, are due to be checked again. +func (c *Cache) Stale(checkedAt time.Time) bool { + return time.Since(checkedAt) >= c.config.CacheTTL +} + +// PutVariant stores a processed variant by its cache key and records it in +// the size accounting. sourceCheckedAt is when the source the variant was +// made from was last checked with upstream; it becomes the modification time +// of the variant's file, which Variant and RenewVariant read. On a disabled +// cache it is a no-op. The accounting insert is best-effort (the startup +// reconciliation pass adopts any variant file that misses its accounting +// row). func (c *Cache) PutVariant( - ctx context.Context, cacheKey imagerequest.VariantKey, content io.Reader, + ctx context.Context, + cacheKey imagerequest.VariantKey, + sourceCheckedAt time.Time, + content io.Reader, ) error { if c.config.MaxBytes == 0 { return nil @@ -208,15 +307,29 @@ func (c *Cache) PutVariant( return err } + // A file left with the time of its writing could be served as made from + // bytes fetched after the ones it was made from + err = c.variants.SetModTime(string(cacheKey), sourceCheckedAt) + if err != nil { + deleteErr := c.variants.Delete(string(cacheKey)) + if deleteErr != nil { + c.log.Warn("failed to remove variant file whose time was not set", + "cache_key", cacheKey, "error", deleteErr) + } + + return err + } + size := int64(len(data)) _, err = c.db.ExecContext(ctx, ` - INSERT INTO variant_content (cache_key, size_bytes) - VALUES (?, ?) + INSERT INTO variant_content (cache_key, size_bytes, checked_at) + VALUES (?, ?, ?) ON CONFLICT(cache_key) DO UPDATE SET size_bytes = excluded.size_bytes, - last_accessed_at = CURRENT_TIMESTAMP - `, string(cacheKey), size) + last_accessed_at = CURRENT_TIMESTAMP, + checked_at = excluded.checked_at + `, string(cacheKey), size, sourceCheckedAt.UTC()) if err != nil { c.log.Warn("failed to record variant in size accounting", "cache_key", cacheKey, "error", err) @@ -272,16 +385,74 @@ func (c *Cache) Source( return file, size, format, nil } -// PutSource stores fetched source content: it writes the file named by -// the source's key, then the source's row. When the row cannot be written -// the file is removed and the store fails, as a file with no row is never -// served or evicted; if that removal fails too, the next reconciliation -// pass removes the file. On a disabled cache it is a no-op returning an -// empty key. +// SourceInfo is what the cache keeps in a cached source's row besides its +// size and access time. +type SourceInfo struct { + // FetchedAt is when the cached bytes were fetched. + FetchedAt time.Time + // CheckedAt is when upstream last sent the bytes or answered that they + // had not changed. + CheckedAt time.Time + // ETag is the ETag header upstream sent with the bytes, or "". + ETag string + // LastModified is the Last-Modified header upstream sent with the + // bytes, or "". + LastModified string +} + +// LookupSource returns what the cache keeps about the cached source of src. +// A source with no row is ErrNotFound, as is any source of a disabled cache. +func (c *Cache) LookupSource( + ctx context.Context, src imagerequest.Source, +) (SourceInfo, error) { + if c.config.MaxBytes == 0 { + return SourceInfo{}, ErrNotFound + } + + var info SourceInfo + + err := c.db.QueryRowContext(ctx, ` + SELECT fetched_at, checked_at, upstream_etag, upstream_last_modified + FROM source_content WHERE source_key = ? + `, string(src.Key())).Scan( + &info.FetchedAt, &info.CheckedAt, &info.ETag, &info.LastModified) + if errors.Is(err, sql.ErrNoRows) { + return SourceInfo{}, ErrNotFound + } + + if err != nil { + return SourceInfo{}, fmt.Errorf("failed to look up source: %w", err) + } + + return info, nil +} + +// RenewSource records that upstream answered at checkedAt that the cached +// source of src had not changed. +func (c *Cache) RenewSource( + ctx context.Context, src imagerequest.Source, checkedAt time.Time, +) error { + _, err := c.db.ExecContext(ctx, + `UPDATE source_content SET checked_at = ? WHERE source_key = ?`, + checkedAt.UTC(), string(src.Key())) + if err != nil { + return fmt.Errorf("failed to renew source: %w", err) + } + + return nil +} + +// PutSource stores fetched source content with what info says about it: it +// writes the file named by the source's key, then the source's row. When the +// row cannot be written the file is removed and the store fails, as a file +// with no row is never served or evicted; if that removal fails too, the +// next reconciliation pass removes the file. On a disabled cache it is a +// no-op returning an empty key. func (c *Cache) PutSource( ctx context.Context, src imagerequest.Source, content io.Reader, + info SourceInfo, ) (imagerequest.SourceKey, error) { if c.config.MaxBytes == 0 { return "", nil @@ -302,12 +473,18 @@ func (c *Cache) PutSource( size := int64(len(data)) _, err = c.db.ExecContext(ctx, ` - INSERT INTO source_content (source_key, size_bytes) - VALUES (?, ?) + INSERT INTO source_content (source_key, size_bytes, fetched_at, + checked_at, upstream_etag, upstream_last_modified) + VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(source_key) DO UPDATE SET size_bytes = excluded.size_bytes, - last_accessed_at = CURRENT_TIMESTAMP - `, string(key), size) + last_accessed_at = CURRENT_TIMESTAMP, + fetched_at = excluded.fetched_at, + checked_at = excluded.checked_at, + upstream_etag = excluded.upstream_etag, + upstream_last_modified = excluded.upstream_last_modified + `, string(key), size, info.FetchedAt.UTC(), info.CheckedAt.UTC(), info.ETag, + info.LastModified) if err != nil { deleteErr := c.srcContent.Delete(string(key)) if deleteErr != nil { @@ -449,6 +626,26 @@ func (c *Cache) cleanExpiredBatch(ctx context.Context) (int64, error) { return deleted, nil } +// openVariant opens the file of a cached variant and returns it with its +// information, which gives its size and its modification time. +func (c *Cache) openVariant( + cacheKey imagerequest.VariantKey, +) (*os.File, fs.FileInfo, error) { + file, _, err := c.variants.Open(string(cacheKey)) + if err != nil { + return nil, nil, err + } + + info, err := file.Stat() + if err != nil { + _ = file.Close() + + return nil, nil, fmt.Errorf("failed to stat variant: %w", err) + } + + return file, info, nil +} + // readFormat reads the first bytes of a source or variant file, without // moving its offset, and returns the format they show. Sources and variants // are in the same six formats. diff --git a/internal/imgcache/cache_hit_internal_test.go b/internal/imgcache/cache_hit_internal_test.go index 828ce50..0dc7dcb 100644 --- a/internal/imgcache/cache_hit_internal_test.go +++ b/internal/imgcache/cache_hit_internal_test.go @@ -49,7 +49,7 @@ func TestVariant_ContentTypeFromFirstBytes(t *testing.T) { key := imagerequest.VariantKey("aabbccdd10" + string(format)) - err = cache.PutVariant(t.Context(), key, bytes.NewReader(image)) + err = cache.PutVariant(t.Context(), key, time.Now(), bytes.NewReader(image)) if err != nil { t.Fatalf("PutVariant(%s) error = %v", format, err) } diff --git a/internal/imgcache/eviction.go b/internal/imgcache/eviction.go index 6746411..4081bec 100644 --- a/internal/imgcache/eviction.go +++ b/internal/imgcache/eviction.go @@ -131,6 +131,56 @@ func (c *Cache) EvictToLimit(ctx context.Context) error { } } +// EvictPastMaxAge evicts each source and variant whose checked_at is +// CacheMaxAge or more ago, evictionBatchSize of each at a time, until none +// is left. It is a no-op when the cache is disabled. +func (c *Cache) EvictPastMaxAge(ctx context.Context) error { + if c.config.MaxBytes == 0 { + return nil + } + + cutoff := time.Now().UTC().Add(-CacheMaxAge) + + for { + sources, err := queryStringColumn[imagerequest.SourceKey](ctx, c.db, ` + SELECT source_key FROM source_content WHERE checked_at < ? LIMIT ? + `, "sources past the maximum age", "source key", cutoff, evictionBatchSize) + if err != nil { + return err + } + + for _, key := range sources { + err = c.evictSourceBlob(ctx, key) + if err != nil { + return err + } + } + + variants, err := queryStringColumn[imagerequest.VariantKey](ctx, c.db, ` + SELECT cache_key FROM variant_content WHERE checked_at < ? LIMIT ? + `, "variants past the maximum age", "variant key", cutoff, evictionBatchSize) + if err != nil { + return err + } + + for _, key := range variants { + err = c.evictVariant(ctx, key) + if err != nil { + return err + } + } + + if len(sources) > 0 || len(variants) > 0 { + c.log.Info("evicted cache content past the maximum age", + "sources", len(sources), "variants", len(variants)) + } + + if len(sources) < evictionBatchSize && len(variants) < evictionBatchSize { + return nil + } + } +} + // evictBatch fetches one batch of LRU candidates across variants and // source blobs and evicts them oldest-first until excessBytes are // freed or the batch is exhausted. It returns the bytes freed. A @@ -344,8 +394,9 @@ func (c *Cache) notifyWritePressure() { } // StartEviction launches the background eviction goroutine, which -// removes the expired negative cache rows at startup and again on every -// periodic tick thereafter, reconciles the database accounting with the +// removes the expired negative cache rows, and the sources and variants +// past CacheMaxAge, at startup and again on every periodic tick +// thereafter, reconciles the database accounting with the // cache directories at startup and again on the first periodic tick // that comes at least ten times as long after a reconciliation pass // ended as that pass took, and evicts to the configured limit on the given @@ -392,6 +443,7 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) { nextReconciliation := c.runReconciliationPass(ctx) c.runCleanExpired(ctx) + c.runEvictPastMaxAge(ctx) c.runEvictionPass(ctx) ticker := time.NewTicker(interval) @@ -417,6 +469,7 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) { } c.runCleanExpired(ctx) + c.runEvictPastMaxAge(ctx) case <-c.evictionPressure: } @@ -474,6 +527,20 @@ func (c *Cache) runCleanExpired(ctx context.Context) { } } +// runEvictPastMaxAge runs EvictPastMaxAge, logging a failure instead of +// propagating it (the loop must keep running). It does nothing once ctx is +// cancelled. +func (c *Cache) runEvictPastMaxAge(ctx context.Context) { + if ctx.Err() != nil { + return + } + + err := c.EvictPastMaxAge(ctx) + if err != nil { + c.log.Warn("eviction past the maximum age failed", "error", err) + } +} + // reconcileAccounting synchronizes the database size accounting with // the actual contents of the cache directories. It runs at startup and // then on the first periodic eviction tick that comes at least ten times @@ -609,16 +676,18 @@ func (c *Cache) reconcileVariantFile( } // adoptVariantFile inserts an accounting row for a variant file that -// has none, using the file's size and modification time. +// has none, using the file's size and modification time, which is also +// the variant's checked_at. func (c *Cache) adoptVariantFile( ctx context.Context, info fs.FileInfo, cacheKey imagerequest.VariantKey, ) error { modTime := info.ModTime().UTC().Format(sqliteTimestampLayout) _, err := c.db.ExecContext(ctx, ` - INSERT INTO variant_content (cache_key, size_bytes, last_accessed_at) - VALUES (?, ?, ?) - `, string(cacheKey), info.Size(), modTime) + INSERT INTO variant_content (cache_key, size_bytes, last_accessed_at, + checked_at) + VALUES (?, ?, ?, ?) + `, string(cacheKey), info.Size(), modTime, info.ModTime().UTC()) if err != nil { return fmt.Errorf("failed to adopt variant file into accounting: %w", err) } diff --git a/internal/imgcache/eviction_internal_test.go b/internal/imgcache/eviction_internal_test.go index fe64af1..cdab8e2 100644 --- a/internal/imgcache/eviction_internal_test.go +++ b/internal/imgcache/eviction_internal_test.go @@ -101,7 +101,7 @@ func storeEvictionTestSource( } key, err := cache.PutSource( - context.Background(), req.Source, bytes.NewReader(content), + context.Background(), req.Source, bytes.NewReader(content), SourceInfo{}, ) if err != nil { t.Fatalf("PutSource(%s%s) failed: %v", host, path, err) @@ -117,7 +117,7 @@ func storeEvictionTestVariant( ) { t.Helper() - err := cache.PutVariant(t.Context(), key, bytes.NewReader(content)) + err := cache.PutVariant(t.Context(), key, time.Now(), bytes.NewReader(content)) if err != nil { t.Fatalf("PutVariant(%s) failed: %v", key, err) } @@ -432,13 +432,14 @@ func assertDisabledCacheWritesAreNoOps( ctx := t.Context() err := cache.PutVariant( - ctx, imagerequest.CacheKey(req), bytes.NewReader([]byte("data")), + ctx, imagerequest.CacheKey(req), time.Now(), bytes.NewReader([]byte("data")), ) if err != nil { t.Fatalf("PutVariant on disabled cache must be a no-op, got error: %v", err) } - key, err := cache.PutSource(ctx, req.Source, bytes.NewReader([]byte("data"))) + key, err := cache.PutSource(ctx, req.Source, bytes.NewReader([]byte("data")), + SourceInfo{}) if err != nil { t.Fatalf("PutSource on disabled cache must be a no-op, got error: %v", err) } diff --git a/internal/imgcache/filestore.go b/internal/imgcache/filestore.go index 779ff65..9800ab6 100644 --- a/internal/imgcache/filestore.go +++ b/internal/imgcache/filestore.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "path/filepath" + "time" ) // Storage constants. @@ -91,6 +92,17 @@ func (s *FileStore) Size(key string) (int64, error) { return stat.Size(), nil } +// SetModTime sets the modification time of the file for key to modTime, +// leaving its access time as it is. +func (s *FileStore) SetModTime(key string, modTime time.Time) error { + err := os.Chtimes(s.keyToPath(key), time.Time{}, modTime) + if err != nil { + return fmt.Errorf("failed to set modification time: %w", err) + } + + return nil +} + // Delete removes the file for key. A missing file is not an error. func (s *FileStore) Delete(key string) error { err := os.Remove(s.keyToPath(key)) diff --git a/internal/imgcache/max_age_internal_test.go b/internal/imgcache/max_age_internal_test.go new file mode 100644 index 0000000..70314e2 --- /dev/null +++ b/internal/imgcache/max_age_internal_test.go @@ -0,0 +1,88 @@ +package imgcache + +import ( + "bytes" + "os" + "slices" + "testing" + "time" + + "sneak.berlin/go/pixa/internal/imagerequest" +) + +// TestEvictPastMaxAge stores a source and a variant last checked with +// upstream 31 days ago and a source and a variant checked 29 days ago. +// EvictPastMaxAge must remove the rows and files of the first two and keep +// the other two. +func TestEvictPastMaxAge(t *testing.T) { + t.Parallel() + + cache, _ := newEvictionTestCache(t, 1<<30) + old := time.Now().Add(-31 * 24 * time.Hour) + recent := time.Now().Add(-29 * 24 * time.Hour) + + oldSource := imagerequest.Source{Host: testHostCDN, Path: "/old.jpg"} + recentSource := imagerequest.Source{Host: testHostCDN, Path: "/recent.jpg"} + + for src, checkedAt := range map[imagerequest.Source]time.Time{ + oldSource: old, recentSource: recent, + } { + _, err := cache.PutSource(t.Context(), src, bytes.NewReader([]byte("source")), + SourceInfo{FetchedAt: checkedAt, CheckedAt: checkedAt}) + if err != nil { + t.Fatalf("PutSource(%s) error = %v", src.Path, err) + } + } + + oldVariant := imagerequest.VariantKey("aabbccdd0001") + recentVariant := imagerequest.VariantKey("aabbccdd0002") + + for key, checkedAt := range map[imagerequest.VariantKey]time.Time{ + oldVariant: old, recentVariant: recent, + } { + err := cache.PutVariant(t.Context(), key, checkedAt, + bytes.NewReader([]byte("variant"))) + if err != nil { + t.Fatalf("PutVariant(%s) error = %v", key, err) + } + } + + err := cache.EvictPastMaxAge(t.Context()) + if err != nil { + t.Fatalf("EvictPastMaxAge() error = %v", err) + } + + sourceRows, err := queryStringColumn[imagerequest.SourceKey](t.Context(), + cache.db, `SELECT source_key FROM source_content`, "sources", "source") + if err != nil { + t.Fatal(err) + } + + variantRows, err := queryStringColumn[imagerequest.VariantKey](t.Context(), + cache.db, `SELECT cache_key FROM variant_content`, "variants", "variant") + if err != nil { + t.Fatal(err) + } + + t.Logf("source rows left = %v, variant rows left = %v", sourceRows, variantRows) + + if !slices.Equal(sourceRows, []imagerequest.SourceKey{recentSource.Key()}) { + t.Errorf("source rows left = %v, want only %s", sourceRows, recentSource.Key()) + } + + if !slices.Equal(variantRows, []imagerequest.VariantKey{recentVariant}) { + t.Errorf("variant rows left = %v, want only %s", variantRows, recentVariant) + } + + for path, wantKept := range map[string]bool{ + cache.srcContent.keyToPath(string(oldSource.Key())): false, + cache.srcContent.keyToPath(string(recentSource.Key())): true, + cache.variants.keyToPath(string(oldVariant)): false, + cache.variants.keyToPath(string(recentVariant)): true, + } { + _, statErr := os.Stat(path) + if kept := statErr == nil; kept != wantKept { + t.Errorf("file %s kept = %t, want %t (%v)", path, kept, wantKept, statErr) + } + } +} diff --git a/internal/imgcache/source_content_internal_test.go b/internal/imgcache/source_content_internal_test.go index 60a9be4..d3b3d70 100644 --- a/internal/imgcache/source_content_internal_test.go +++ b/internal/imgcache/source_content_internal_test.go @@ -128,7 +128,7 @@ func TestPutSource_FailedRowWriteLeavesNoFile(t *testing.T) { Source: imagerequest.Source{Host: testHostCDN, Path: testPathCat}, } key, err := cache.PutSource(t.Context(), req.Source, - bytes.NewReader([]byte("source bytes"))) + bytes.NewReader([]byte("source bytes")), SourceInfo{}) t.Logf("PutSource() = %q, error = %v", key, err) if err == nil { @@ -170,7 +170,7 @@ func TestSource_FormatFromFirstBytes(t *testing.T) { t.Fatalf("reading the %s image: %v", format, err) } - _, err = cache.PutSource(t.Context(), src, bytes.NewReader(image)) + _, err = cache.PutSource(t.Context(), src, bytes.NewReader(image), SourceInfo{}) if err != nil { t.Fatalf("PutSource(%s) error = %v", format, err) } diff --git a/internal/providers/providers.go b/internal/providers/providers.go index dda53ca..03824ce 100644 --- a/internal/providers/providers.go +++ b/internal/providers/providers.go @@ -83,6 +83,7 @@ func newCache(lc fx.Lifecycle, params cacheParams) (*imgcache.Cache, error) { cache, err := imgcache.NewCache(db, imgcache.Config{ StateDir: cfg.StateDir, NegativeTTL: imgcache.DefaultNegativeTTL, + CacheTTL: imgcache.DefaultCacheTTL, MaxBytes: maxBytes, Logger: params.Logger.Get(), }, params.MetricsRegistry)