Check cached sources with upstream after the cache TTL (closes #69) #350
No files matched your search
@@ -6,8 +6,9 @@ results.
|
||||
|
||||
A web page asks pixa for an image at a size and in a format. pixa fetches the
|
||||
original image, makes the requested image from it, and serves that. It keeps
|
||||
both on disk. Later requests for the same image are served from disk, without a
|
||||
fetch or a conversion.
|
||||
both on disk. Later requests for the same image are served from disk. After 24
|
||||
hours pixa first asks the server that holds the original whether it changed (see
|
||||
Cache TTL).
|
||||
|
||||
In this document, the upstream host is the server that holds an original image.
|
||||
|
||||
@@ -232,8 +233,8 @@ These terms are the names the code uses:
|
||||
- **Variant**: the image pixa serves for an image request, made from its source
|
||||
with its transform.
|
||||
- **Variant key**: the SHA-256 hash of a variant's source, size, format, quality
|
||||
and fit. It names the variant's file and its row in the database, and gives
|
||||
its `ETag`.
|
||||
and fit. It names the variant's file and its row in the database. The
|
||||
variant's `ETag` is made from it and the file's modification time.
|
||||
- **Signed URL**: a `/v1/image/` URL whose `sig` parameter signs its image
|
||||
request and its expiry `exp` with `signing_key`.
|
||||
- **Encrypted URL**: a `/v1/e/` URL whose token holds its image request and any
|
||||
@@ -244,9 +245,12 @@ These terms are the names the code uses:
|
||||
after a fetch failed in one of these ways:
|
||||
- the upstream answered with an error status;
|
||||
- the body's first bytes were not those of the format its `Content-Type`
|
||||
names.
|
||||
names;
|
||||
- a check of a cached image 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 uses a cached source and serves 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
|
||||
@@ -517,8 +521,10 @@ seconds left until then, at most one year. So no browser or proxy cache keeps
|
||||
the image after pixa would refuse the URL. A URL with no expiry gets one year.
|
||||
`immutable` only stops a client revalidating while its copy is fresh. The
|
||||
response also carries `X-Pixa-Cache`: `HIT` for an image served from the cache,
|
||||
and `MISS` for one made for this request. The `ETag` is the first 16 characters
|
||||
of the variant key, in quotes.
|
||||
and `MISS` for one made for this request or older than 24 hours (see Cache TTL).
|
||||
The `ETag` is made from the first 16 characters of the variant key and the
|
||||
variant file's modification time (see Cache TTL). So a variant made again from a
|
||||
changed source gets a new `ETag`.
|
||||
|
||||
### Signed URLs
|
||||
|
||||
@@ -824,10 +830,17 @@ and `state.sqlite3-shm` beside the file while pixa runs.
|
||||
|
||||
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 and when it was last used
|
||||
- its source's check time when the variant was made or last renewed
|
||||
- the `ETag` and `Last-Modified` headers upstream sent with the source it
|
||||
was made from
|
||||
- `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
|
||||
@@ -881,6 +894,48 @@ within `cache_max_bytes`. It runs after each store, and every five minutes. It
|
||||
also removes the expired rows of the negative cache every five minutes, at most
|
||||
1000 per statement, so a request waits behind 1000 deletions at most.
|
||||
|
||||
### Cache TTL
|
||||
|
||||
pixa uses a cached source for 24 hours after upstream last sent it or confirmed
|
||||
it unchanged. During that time it serves the source's variants from the cache.
|
||||
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 uses the cached source and serves
|
||||
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 pixa uses a source whose checks keep
|
||||
failing for at most 30 days. Eviction to stay under the disk cache limit can
|
||||
remove images sooner.
|
||||
|
||||
That eviction can remove a source and keep its variants. Each variant's row
|
||||
keeps the `ETag` and `Last-Modified` of the source it was made from. So pixa
|
||||
checks such a variant with upstream in the same way after 24 hours:
|
||||
|
||||
- When upstream answers 304 Not Modified, pixa serves the cached variant for
|
||||
another 24 hours. Nothing is downloaded or converted again.
|
||||
- When upstream sends the image, pixa caches it as the source and converts the
|
||||
variant again.
|
||||
- When the check fails, pixa serves the cached variant and puts the source URL
|
||||
in the negative cache.
|
||||
|
||||
A variant file's modification time is its source's check time when the variant
|
||||
was made or last renewed. So a cache hit 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.
|
||||
|
||||
### Configuration
|
||||
|
||||
Every setting can be given as an environment variable, in a config file
|
||||
|
||||
@@ -3,14 +3,24 @@
|
||||
|
||||
-- Source images, one row per source URL
|
||||
-- Files stored at: cache/sources/<ab>/<cd>/<source_key>, 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/<ab>/<cd>/<cache_key>. Tracked here
|
||||
@@ -18,14 +28,24 @@ 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.
|
||||
-- upstream_etag and upstream_last_modified are those of the source bytes
|
||||
-- the variant was made from, so the variant can still be checked with
|
||||
-- upstream after its source is evicted.
|
||||
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,
|
||||
upstream_etag TEXT NOT NULL DEFAULT '',
|
||||
upstream_last_modified TEXT NOT NULL DEFAULT ''
|
||||
);
|
||||
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
|
||||
|
||||
@@ -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{}{}
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -257,6 +257,90 @@ func TestHandleImage_ConditionalRequest_IfNoneMatch_DifferentETag(t *testing.T)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHandleImage_ConditionalRequest_EarlierETagGetsChangedImage serves an
|
||||
// image, changes it upstream, and requests it again with the ETag first
|
||||
// served, once the cache TTL has passed. The changed image must be served,
|
||||
// not 304 Not Modified.
|
||||
func TestHandleImage_ConditionalRequest_EarlierETagGetsChangedImage(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const host = "goodhost.example.com"
|
||||
|
||||
upstream := fstest.MapFS{
|
||||
host + photoPath: &fstest.MapFile{
|
||||
Data: generateTestJPEG(t, 100, 100, color.RGBA{255, 0, 0, 255}),
|
||||
},
|
||||
}
|
||||
|
||||
// A cache TTL of one nanosecond has passed by the second request
|
||||
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.Config{
|
||||
StateDir: t.TempDir(),
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
CacheTTL: time.Nanosecond,
|
||||
MaxBytes: 1 << 30,
|
||||
}, prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
t.Fatalf("imgcache.NewCache() error = %v", err)
|
||||
}
|
||||
|
||||
svc, err := imageproxy.New(&imageproxy.Config{
|
||||
Cache: cache,
|
||||
Fetcher: newMockFetcher(upstream),
|
||||
Processor: imageprocessor.New(imageprocessor.Params{
|
||||
MaxInputBytes: 50 << 20,
|
||||
MaxConcurrentProcessing: 2,
|
||||
}),
|
||||
MaxResponseSize: 50 << 20,
|
||||
Logger: slog.New(slog.DiscardHandler),
|
||||
}, prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
t.Fatalf("imageproxy.New() error = %v", err)
|
||||
}
|
||||
|
||||
h := &Handlers{
|
||||
imgSvc: svc,
|
||||
log: slog.New(slog.DiscardHandler),
|
||||
signer: signature.New(testSigningKey),
|
||||
allowlist: allowlist.New([]string{host}),
|
||||
}
|
||||
|
||||
r := chi.NewRouter()
|
||||
r.Get("/v1/image/*", h.HandleImage())
|
||||
|
||||
target := "/v1/image/" + host + photoPath + "/50x50.jpeg"
|
||||
|
||||
first := httptest.NewRecorder()
|
||||
r.ServeHTTP(first,
|
||||
httptest.NewRequestWithContext(t.Context(), http.MethodGet, target, nil))
|
||||
|
||||
etag := first.Header().Get("ETag")
|
||||
if first.Code != http.StatusOK || etag == "" {
|
||||
t.Fatalf("first request: status = %d, ETag = %q, want %d and an ETag",
|
||||
first.Code, etag, http.StatusOK)
|
||||
}
|
||||
|
||||
upstream[host+photoPath] = &fstest.MapFile{
|
||||
Data: generateTestJPEG(t, 100, 100, color.RGBA{0, 0, 255, 255}),
|
||||
}
|
||||
|
||||
req := httptest.NewRequestWithContext(t.Context(), http.MethodGet, target, nil)
|
||||
req.Header.Set("If-None-Match", etag)
|
||||
|
||||
second := httptest.NewRecorder()
|
||||
r.ServeHTTP(second, req)
|
||||
t.Logf("first ETag %s; with it after the change: status %d, ETag %s, %d bytes",
|
||||
etag, second.Code, second.Header().Get("ETag"), second.Body.Len())
|
||||
|
||||
if second.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want %d with the changed image",
|
||||
second.Code, http.StatusOK)
|
||||
}
|
||||
|
||||
if bytes.Equal(second.Body.Bytes(), first.Body.Bytes()) {
|
||||
t.Error("the image served is the one made before the change")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleImage_ETagHeader(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,493 @@
|
||||
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")
|
||||
|
||||
// upstreamFailure is one way a fetch from upstream fails.
|
||||
type upstreamFailure struct {
|
||||
name string
|
||||
err error
|
||||
}
|
||||
|
||||
// upstreamFailures returns the ways a check of a cached image with upstream
|
||||
// can fail: an error status, a timeout and no connection.
|
||||
func upstreamFailures() []upstreamFailure {
|
||||
return []upstreamFailure{
|
||||
{"error status", fmt.Errorf("%w: status 503", httpfetcher.ErrUpstreamError)},
|
||||
{"timeout", httpfetcher.ErrUpstreamTimeout},
|
||||
{"no connection", errConnectionRefused},
|
||||
}
|
||||
}
|
||||
|
||||
// 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, with the source still cached and with the source's
|
||||
// row and file removed, as eviction does. 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
|
||||
sourceEvicted bool
|
||||
etag string
|
||||
lastModified string
|
||||
wantIfNoneMatch string
|
||||
wantIfModifiedSince string
|
||||
}{
|
||||
{"ETag", false, testETag, "", testETag, ""},
|
||||
{"Last-Modified only", false, "", testLastModified, "", testLastModified},
|
||||
{"both", false, testETag, testLastModified, testETag, testLastModified},
|
||||
{"ETag, source evicted", true, testETag, "", testETag, ""},
|
||||
{"Last-Modified only, source evicted", true, "", testLastModified, "",
|
||||
testLastModified},
|
||||
{"both, source evicted", true, 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))
|
||||
|
||||
if tt.sourceEvicted {
|
||||
evictSource(t, svc, variant.Source)
|
||||
}
|
||||
|
||||
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()
|
||||
|
||||
for _, failure := range upstreamFailures() {
|
||||
t.Run(failure.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(failure.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)
|
||||
}
|
||||
}
|
||||
|
||||
// TestProxy_Get_UpstreamFailureAfterTTLServesVariantOfEvictedSource caches a
|
||||
// variant, sets its file time two days back, removes its source's row and
|
||||
// file, as eviction does, and makes upstream fail. The cached variant must
|
||||
// still be served.
|
||||
func TestProxy_Get_UpstreamFailureAfterTTLServesVariantOfEvictedSource(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
for _, failure := range upstreamFailures() {
|
||||
t.Run(failure.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))
|
||||
evictSource(t, svc, variant.Source)
|
||||
upstream.fail(failure.err)
|
||||
|
||||
second, _ := getImage(t, svc, variant)
|
||||
t.Logf("upstream fetches = %d", upstream.fetches())
|
||||
|
||||
if !bytes.Equal(first, second) {
|
||||
t.Error("the variant served while upstream fails differs from the cached one")
|
||||
}
|
||||
|
||||
if n := upstream.fetches(); n != 2 {
|
||||
t.Errorf("upstream fetches = %d, want 2", n)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestProxy_Get_FailingUpstreamIsAskedOnceForVariantOfEvictedSource caches a
|
||||
// variant, sets its file time two days back, removes its source's row and
|
||||
// file, as eviction does, and makes every fetch time out. Serving the variant
|
||||
// ten times must ask upstream once: the failed check puts the source URL in
|
||||
// the negative cache.
|
||||
func TestProxy_Get_FailingUpstreamIsAskedOnceForVariantOfEvictedSource(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))
|
||||
evictSource(t, svc, variant.Source)
|
||||
upstream.fail(httpfetcher.ErrUpstreamTimeout)
|
||||
|
||||
for range 10 {
|
||||
image, _ := getImage(t, svc, variant)
|
||||
if !bytes.Equal(image, first) {
|
||||
t.Fatal("the variant served while upstream fails differs from the cached one")
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
// evictSource removes the row and then the file of src's cached source, as
|
||||
// the cache's eviction does, and leaves its variants cached.
|
||||
func evictSource(t *testing.T, svc *testProxy, src imagerequest.Source) {
|
||||
t.Helper()
|
||||
|
||||
_, err := svc.db.ExecContext(t.Context(),
|
||||
`DELETE FROM source_content WHERE source_key = ?`, string(src.Key()))
|
||||
if err != nil {
|
||||
t.Fatalf("deleting the source's row: %v", err)
|
||||
}
|
||||
|
||||
err = os.Remove(sourceFilePath(svc, src.Key()))
|
||||
if err != nil {
|
||||
t.Fatalf("removing the source's file: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
@@ -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(
|
||||
@@ -165,20 +168,20 @@ func (p *Proxy) Get(
|
||||
) (*ImageResponse, error) {
|
||||
cacheKey := imagerequest.CacheKey(req)
|
||||
|
||||
reader, size, contentType, err := p.cache.Variant(cacheKey)
|
||||
cached, err := p.cache.Variant(cacheKey)
|
||||
if err == nil {
|
||||
p.cache.IncrementStats(true)
|
||||
|
||||
return &ImageResponse{
|
||||
Content: reader,
|
||||
ContentLength: size,
|
||||
ContentType: contentType,
|
||||
Content: cached.Content,
|
||||
ContentLength: cached.Size,
|
||||
ContentType: cached.ContentType,
|
||||
CacheStatus: CacheHit,
|
||||
ETag: formatETag(cacheKey),
|
||||
ETag: formatETag(cacheKey, cached.CheckedAt),
|
||||
}, 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)
|
||||
}
|
||||
@@ -285,7 +288,7 @@ func (p *Proxy) processOrWait(
|
||||
ContentLength: int64(len(variant.data)),
|
||||
ContentType: variant.contentType,
|
||||
FetchedBytes: variant.fetchedBytes,
|
||||
ETag: formatETag(cacheKey),
|
||||
ETag: formatETag(cacheKey, variant.checkedAt),
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -337,13 +340,50 @@ 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. When
|
||||
// the source is not cached but the variant is, the variant is checked with
|
||||
// upstream the same way (see fetchAndProcess).
|
||||
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:
|
||||
p.recordFailedCheck(ctx, req, err)
|
||||
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 +396,199 @@ 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, 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.
|
||||
// recordFailedCheck logs a failed check of req's source with upstream, after
|
||||
// which the image is still served from the cache. It puts the source URL in
|
||||
// the negative cache, whatever the failure, a timeout included, so upstream
|
||||
// is asked at most once per NegativeTTL while it fails, however often the
|
||||
// image is served. A failure that was a negative cache hit records nothing.
|
||||
func (p *Proxy) recordFailedCheck(
|
||||
ctx context.Context, req *imagerequest.Request, err error,
|
||||
) {
|
||||
if errors.Is(err, ErrNegativeCached) {
|
||||
return
|
||||
}
|
||||
|
||||
p.log.Warn("checking source with upstream failed, serving from the cache",
|
||||
"request_id", middleware.GetReqID(ctx),
|
||||
"host", req.Source.Host, "path", req.Source.Path, "error", err)
|
||||
p.storeNegative(ctx, req)
|
||||
}
|
||||
|
||||
// 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 {
|
||||
cached, 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
|
||||
}
|
||||
|
||||
return p.readCachedVariant(cacheKey, cached)
|
||||
}
|
||||
|
||||
// cachedVariantIgnoringTTL returns the cached variant of cacheKey whatever
|
||||
// its age, with what its row keeps about the source bytes it was made from
|
||||
// (see imgcache.Cache.LookupVariant). It returns nil when the variant is not
|
||||
// cached or cannot be read. The SourceInfo is empty when the variant is not
|
||||
// returned, or its row is missing or cannot be read.
|
||||
func (p *Proxy) cachedVariantIgnoringTTL(
|
||||
ctx context.Context, cacheKey imagerequest.VariantKey,
|
||||
) (*processedVariant, imgcache.SourceInfo) {
|
||||
cached, err := p.cache.VariantIgnoringTTL(cacheKey)
|
||||
if err != nil {
|
||||
if !errors.Is(err, imgcache.ErrNotFound) {
|
||||
p.log.Warn("cached variant not served", "cache_key", cacheKey,
|
||||
"error", err)
|
||||
}
|
||||
|
||||
return nil, imgcache.SourceInfo{}
|
||||
}
|
||||
|
||||
variant := p.readCachedVariant(cacheKey, cached)
|
||||
if variant == nil {
|
||||
return nil, imgcache.SourceInfo{}
|
||||
}
|
||||
|
||||
madeFrom, err := p.cache.LookupVariant(ctx, cacheKey)
|
||||
if err != nil && !errors.Is(err, imgcache.ErrNotFound) {
|
||||
p.log.Warn("failed to look up cached variant", "cache_key", cacheKey,
|
||||
"error", err)
|
||||
}
|
||||
|
||||
return variant, madeFrom
|
||||
}
|
||||
|
||||
// readCachedVariant reads the cached variant of cacheKey from its opened file,
|
||||
// and closes the file. It returns nil when the variant cannot be read.
|
||||
func (p *Proxy) readCachedVariant(
|
||||
cacheKey imagerequest.VariantKey, file imgcache.VariantFile,
|
||||
) *processedVariant {
|
||||
defer func() { _ = file.Content.Close() }()
|
||||
|
||||
data, err := io.ReadAll(file.Content)
|
||||
if err != nil {
|
||||
p.log.Warn("failed to read cached variant", "cache_key", cacheKey,
|
||||
"error", err)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
return &processedVariant{
|
||||
data: data,
|
||||
contentType: file.ContentType,
|
||||
checkedAt: file.CheckedAt,
|
||||
}
|
||||
}
|
||||
|
||||
// fetchAndProcess fetches the source from upstream, stores it, and makes
|
||||
// and stores the variant from it.
|
||||
//
|
||||
// The variant can still be cached after eviction removed its source. The
|
||||
// fetch then sends the ETag and Last-Modified of the source bytes the variant
|
||||
// was made from, so upstream sends the source only if it changed since. When
|
||||
// it did not change, the cached variant is renewed and served, with no
|
||||
// download and no conversion. When the fetch fails, the cached variant is
|
||||
// served whatever its age, and the failure is recorded as a failed check (see
|
||||
// recordFailedCheck).
|
||||
func (p *Proxy) fetchAndProcess(
|
||||
ctx context.Context,
|
||||
req *imagerequest.Request,
|
||||
cacheKey imagerequest.VariantKey,
|
||||
) (*processedVariant, error) {
|
||||
variant, madeFrom := p.cachedVariantIgnoringTTL(ctx, cacheKey)
|
||||
|
||||
fetched, err := p.fetchSource(ctx, req, madeFrom.ETag, madeFrom.LastModified)
|
||||
|
||||
switch {
|
||||
case err == nil:
|
||||
return p.processFetched(ctx, req, cacheKey, fetched)
|
||||
case variant == nil:
|
||||
return nil, err
|
||||
case errors.Is(err, errSourceNotModified):
|
||||
variant.checkedAt = time.Now().UTC()
|
||||
|
||||
err = p.cache.SetVariantCheckedAt(ctx, cacheKey, variant.checkedAt)
|
||||
if err != nil {
|
||||
p.log.Warn("failed to renew cached variant", "cache_key", cacheKey,
|
||||
"error", err)
|
||||
}
|
||||
default:
|
||||
p.recordFailedCheck(ctx, req, err)
|
||||
}
|
||||
|
||||
return variant, nil
|
||||
}
|
||||
|
||||
// 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,
|
||||
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 +608,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 +619,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 +637,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 +655,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,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -491,14 +750,19 @@ type processedVariant struct {
|
||||
data []byte
|
||||
contentType string
|
||||
fetchedBytes int64
|
||||
// checkedAt is its source's check time when the variant was made or last
|
||||
// renewed, which its cached file has as its modification time
|
||||
checkedAt time.Time
|
||||
}
|
||||
|
||||
// processAndStore processes the image read from source, which is in
|
||||
// sourceFormat, and stores the result.
|
||||
// sourceFormat, and stores the result. sourceInfo describes the source's
|
||||
// bytes, and is stored with the variant (see imgcache.Cache.PutVariant).
|
||||
func (p *Proxy) processAndStore(
|
||||
ctx context.Context,
|
||||
req *imagerequest.Request,
|
||||
cacheKey imagerequest.VariantKey,
|
||||
sourceInfo imgcache.SourceInfo,
|
||||
source io.Reader,
|
||||
sourceFormat imagerequest.Format,
|
||||
fetchBytes int64,
|
||||
@@ -559,7 +823,8 @@ func (p *Proxy) processAndStore(
|
||||
)
|
||||
|
||||
// Store variant to cache
|
||||
err = p.cache.PutVariant(ctx, cacheKey, bytes.NewReader(processedData))
|
||||
err = p.cache.PutVariant(ctx, cacheKey, sourceInfo,
|
||||
bytes.NewReader(processedData))
|
||||
if err != nil {
|
||||
p.log.Warn("failed to store variant", "error", err)
|
||||
// Continue even if caching fails
|
||||
@@ -569,6 +834,7 @@ func (p *Proxy) processAndStore(
|
||||
data: processedData,
|
||||
contentType: processResult.ContentType,
|
||||
fetchedBytes: fetchBytes,
|
||||
checkedAt: sourceInfo.CheckedAt,
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -580,13 +846,16 @@ func isNegativeCacheable(err error) bool {
|
||||
// etagHashLength is the number of hash characters to use for ETags.
|
||||
const etagHashLength = 16
|
||||
|
||||
// formatETag formats a VariantKey as a quoted ETag value.
|
||||
func formatETag(key imagerequest.VariantKey) string {
|
||||
// formatETag returns the quoted ETag of the variant of key: the first
|
||||
// etagHashLength characters of key, then checkedAt in Unix nanoseconds.
|
||||
// checkedAt is its source's check time when the variant was made or last
|
||||
// renewed, so a variant made again from new source bytes gets another ETag.
|
||||
func formatETag(key imagerequest.VariantKey, checkedAt time.Time) string {
|
||||
hash := string(key)
|
||||
// Use first 16 characters of hash for a shorter but still unique ETag
|
||||
if len(hash) > etagHashLength {
|
||||
hash = hash[:etagHashLength]
|
||||
}
|
||||
|
||||
return `"` + hash + `"`
|
||||
return fmt.Sprintf(`"%s-%d"`, hash, checkedAt.UnixNano())
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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),
|
||||
|
||||
+329
-38
@@ -3,8 +3,12 @@
|
||||
// - the variants made from them
|
||||
// - the negative cache: the source URLs whose fetch failed recently
|
||||
//
|
||||
// Each image is a file with a row in the database. Eviction removes the
|
||||
// least recently used images to keep the cache under its size limit.
|
||||
// Each image is a file with a row in the database. A source is used, and its
|
||||
// variants are served, until the cache TTL has passed since upstream last sent
|
||||
// the source or confirmed it unchanged. Then the source is due to be checked
|
||||
// with upstream again (see Stale). Eviction removes:
|
||||
// - the least recently used images, to keep the cache under its size limit
|
||||
// - the images not checked with upstream for EvictionAfterLastCheck
|
||||
package imgcache
|
||||
|
||||
import (
|
||||
@@ -13,7 +17,9 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/fs"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
@@ -38,11 +44,31 @@ 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 uses the source and serves its
|
||||
// variants without asking upstream again.
|
||||
const DefaultCacheTTL = 24 * time.Hour
|
||||
|
||||
// EvictionAfterLastCheck 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 EvictNotCheckedRecently). Until then a source whose checks
|
||||
// fail is used.
|
||||
const EvictionAfterLastCheck = 30 * 24 * time.Hour
|
||||
|
||||
// ErrStale is returned by Variant for a variant whose file's modification
|
||||
// time is CacheTTL or more ago (see Variant).
|
||||
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 is used, and its variants 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 +139,10 @@ func NewCache(
|
||||
log = slog.Default()
|
||||
}
|
||||
|
||||
if config.CacheTTL == 0 {
|
||||
config.CacheTTL = DefaultCacheTTL
|
||||
}
|
||||
|
||||
c := &Cache{
|
||||
db: db,
|
||||
config: config,
|
||||
@@ -156,43 +186,167 @@ func NewCache(
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// 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.
|
||||
func (c *Cache) Variant(
|
||||
cacheKey imagerequest.VariantKey,
|
||||
) (io.ReadCloser, int64, string, error) {
|
||||
// VariantFile is the open file of a cached variant, as Variant,
|
||||
// VariantIgnoringTTL and RenewVariant return it.
|
||||
type VariantFile struct {
|
||||
// Content reads the file. The caller closes it.
|
||||
Content io.ReadCloser
|
||||
// Size is the file's size in bytes.
|
||||
Size int64
|
||||
// ContentType is the MIME type of the variant's format, read from the
|
||||
// file's first bytes.
|
||||
ContentType string
|
||||
// CheckedAt is the file's modification time: its source's check time
|
||||
// when the variant was made or last renewed.
|
||||
CheckedAt time.Time
|
||||
}
|
||||
|
||||
// Variant opens a cached variant and returns its file (see VariantFile). It
|
||||
// opens that one file and makes no database call: the file's modification
|
||||
// time is its source's check time when the variant was made or last renewed
|
||||
// (see PutVariant), and a variant whose time is 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) (VariantFile, error) {
|
||||
if c.config.MaxBytes == 0 {
|
||||
return nil, 0, "", ErrNotFound
|
||||
return VariantFile{}, ErrNotFound
|
||||
}
|
||||
|
||||
file, size, err := c.variants.Open(string(cacheKey))
|
||||
file, info, err := c.variants.Open(string(cacheKey))
|
||||
if err != nil {
|
||||
return nil, 0, "", err
|
||||
return VariantFile{}, err
|
||||
}
|
||||
|
||||
if c.Stale(info.ModTime()) {
|
||||
_ = file.Close()
|
||||
|
||||
return VariantFile{}, ErrStale
|
||||
}
|
||||
|
||||
return c.finishOpeningVariant(cacheKey, file, info)
|
||||
}
|
||||
|
||||
// VariantIgnoringTTL opens a cached variant whatever its age, and returns it
|
||||
// as Variant does. The image proxy reads it for a variant whose source is not
|
||||
// cached, before it checks the source with upstream (see LookupVariant).
|
||||
func (c *Cache) VariantIgnoringTTL(
|
||||
cacheKey imagerequest.VariantKey,
|
||||
) (VariantFile, error) {
|
||||
if c.config.MaxBytes == 0 {
|
||||
return VariantFile{}, ErrNotFound
|
||||
}
|
||||
|
||||
file, info, err := c.variants.Open(string(cacheKey))
|
||||
if err != nil {
|
||||
return VariantFile{}, err
|
||||
}
|
||||
|
||||
return c.finishOpeningVariant(cacheKey, file, info)
|
||||
}
|
||||
|
||||
// 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's check time forward to source.CheckedAt (see SetVariantCheckedAt).
|
||||
// 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,
|
||||
) (VariantFile, error) {
|
||||
if c.config.MaxBytes == 0 {
|
||||
return VariantFile{}, ErrNotFound
|
||||
}
|
||||
|
||||
file, info, err := c.variants.Open(string(cacheKey))
|
||||
if err != nil {
|
||||
return VariantFile{}, err
|
||||
}
|
||||
|
||||
if info.ModTime().Before(source.FetchedAt) {
|
||||
_ = file.Close()
|
||||
|
||||
return VariantFile{}, ErrNotFound
|
||||
}
|
||||
|
||||
format, err := readFormat(file)
|
||||
if err != nil {
|
||||
_ = file.Close()
|
||||
|
||||
return nil, 0, "", err
|
||||
return VariantFile{}, err
|
||||
}
|
||||
|
||||
checkedAt := info.ModTime()
|
||||
|
||||
if checkedAt.Before(source.CheckedAt) {
|
||||
err = c.SetVariantCheckedAt(ctx, cacheKey, source.CheckedAt)
|
||||
if err != nil {
|
||||
_ = file.Close()
|
||||
|
||||
return VariantFile{}, err
|
||||
}
|
||||
|
||||
checkedAt = source.CheckedAt
|
||||
}
|
||||
|
||||
c.touchVariant(cacheKey)
|
||||
|
||||
return file, size, format.MIMEType(), nil
|
||||
return VariantFile{
|
||||
Content: file,
|
||||
Size: info.Size(),
|
||||
ContentType: format.MIMEType(),
|
||||
CheckedAt: checkedAt,
|
||||
}, 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).
|
||||
// SetVariantCheckedAt records checkedAt as when upstream last sent the source
|
||||
// bytes the cached variant of cacheKey was made from, or answered that they
|
||||
// had not changed. It sets the variant file's modification time to checkedAt,
|
||||
// so Variant serves the file until CacheTTL after it. It sets the checked_at
|
||||
// of the variant's row too, so eviction keeps the variant until
|
||||
// EvictionAfterLastCheck after it. A failure to write the row is only logged.
|
||||
// On a disabled cache it is a no-op.
|
||||
func (c *Cache) SetVariantCheckedAt(
|
||||
ctx context.Context, cacheKey imagerequest.VariantKey, checkedAt time.Time,
|
||||
) error {
|
||||
if c.config.MaxBytes == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
err := c.variants.SetModTime(string(cacheKey), checkedAt)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
_, err = c.db.ExecContext(ctx,
|
||||
`UPDATE variant_content SET checked_at = ? WHERE cache_key = ?`,
|
||||
checkedAt.UTC(), string(cacheKey))
|
||||
if err != nil {
|
||||
c.log.Warn("failed to renew variant's checked_at",
|
||||
"cache_key", cacheKey, "error", err)
|
||||
}
|
||||
|
||||
return 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. source describes the source bytes the variant was made
|
||||
// from. Its CheckedAt becomes the modification time of the variant's file,
|
||||
// which Variant and RenewVariant read. The variant's row keeps its CheckedAt,
|
||||
// ETag and LastModified (see LookupVariant). 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,
|
||||
source SourceInfo,
|
||||
content io.Reader,
|
||||
) error {
|
||||
if c.config.MaxBytes == 0 {
|
||||
return nil
|
||||
@@ -208,15 +362,33 @@ 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), source.CheckedAt)
|
||||
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,
|
||||
upstream_etag, upstream_last_modified)
|
||||
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,
|
||||
upstream_etag = excluded.upstream_etag,
|
||||
upstream_last_modified = excluded.upstream_last_modified
|
||||
`, string(cacheKey), size, source.CheckedAt.UTC(), source.ETag,
|
||||
source.LastModified)
|
||||
if err != nil {
|
||||
c.log.Warn("failed to record variant in size accounting",
|
||||
"cache_key", cacheKey, "error", err)
|
||||
@@ -255,7 +427,7 @@ func (c *Cache) Source(
|
||||
return nil, 0, "", ErrNotFound
|
||||
}
|
||||
|
||||
file, size, err := c.srcContent.Open(string(key))
|
||||
file, info, err := c.srcContent.Open(string(key))
|
||||
if err != nil {
|
||||
return nil, 0, "", err
|
||||
}
|
||||
@@ -269,19 +441,107 @@ func (c *Cache) Source(
|
||||
|
||||
c.touchSourceContent(key)
|
||||
|
||||
return file, size, format, nil
|
||||
return file, info.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
|
||||
}
|
||||
|
||||
// LookupVariant returns what the row of the cached variant of cacheKey keeps
|
||||
// about the source bytes the variant was made from: the CheckedAt, ETag and
|
||||
// LastModified they had when the variant was made or last renewed. The row
|
||||
// keeps no FetchedAt, so it is zero. With the ETag and LastModified the image
|
||||
// proxy checks the variant with upstream after its source is evicted. A
|
||||
// variant with no row is ErrNotFound, as is any variant of a disabled cache.
|
||||
func (c *Cache) LookupVariant(
|
||||
ctx context.Context, cacheKey imagerequest.VariantKey,
|
||||
) (SourceInfo, error) {
|
||||
if c.config.MaxBytes == 0 {
|
||||
return SourceInfo{}, ErrNotFound
|
||||
}
|
||||
|
||||
var info SourceInfo
|
||||
|
||||
err := c.db.QueryRowContext(ctx, `
|
||||
SELECT checked_at, upstream_etag, upstream_last_modified
|
||||
FROM variant_content WHERE cache_key = ?
|
||||
`, string(cacheKey)).Scan(&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 variant: %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 +562,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 +715,31 @@ func (c *Cache) cleanExpiredBatch(ctx context.Context) (int64, error) {
|
||||
return deleted, nil
|
||||
}
|
||||
|
||||
// finishOpeningVariant returns a variant file that FileStore.Open opened, with
|
||||
// its information info, as Variant does. It reads the content type from the
|
||||
// file's first bytes and hands the variant's access time to the background
|
||||
// writer (see touchVariant). It closes the file when its first bytes are not
|
||||
// those of an output format.
|
||||
func (c *Cache) finishOpeningVariant(
|
||||
cacheKey imagerequest.VariantKey, file *os.File, info fs.FileInfo,
|
||||
) (VariantFile, error) {
|
||||
format, err := readFormat(file)
|
||||
if err != nil {
|
||||
_ = file.Close()
|
||||
|
||||
return VariantFile{}, err
|
||||
}
|
||||
|
||||
c.touchVariant(cacheKey)
|
||||
|
||||
return VariantFile{
|
||||
Content: file,
|
||||
Size: info.Size(),
|
||||
ContentType: format.MIMEType(),
|
||||
CheckedAt: info.ModTime(),
|
||||
}, 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.
|
||||
|
||||
@@ -49,16 +49,19 @@ 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, SourceInfo{CheckedAt: time.Now()},
|
||||
bytes.NewReader(image))
|
||||
if err != nil {
|
||||
t.Fatalf("PutVariant(%s) error = %v", format, err)
|
||||
}
|
||||
|
||||
reader, size, contentType, err := cache.Variant(key)
|
||||
variant, err := cache.Variant(key)
|
||||
if err != nil {
|
||||
t.Fatalf("Variant(%s) error = %v", format, err)
|
||||
}
|
||||
|
||||
reader, size, contentType := variant.Content, variant.Size, variant.ContentType
|
||||
|
||||
served, err := io.ReadAll(reader)
|
||||
_ = reader.Close()
|
||||
|
||||
@@ -109,16 +112,19 @@ func TestVariant_RefusesFileOfNoOutputFormat(t *testing.T) {
|
||||
t.Fatalf("%s: storing the variant file: %v", tt.name, err)
|
||||
}
|
||||
|
||||
reader, _, contentType, err := cache.Variant(key)
|
||||
variant, err := cache.Variant(key)
|
||||
t.Logf("%s: Variant() error = %v", tt.name, err)
|
||||
|
||||
reader, contentType := variant.Content, variant.ContentType
|
||||
if reader != nil || contentType != "" || !errors.Is(err, tt.wantErr) {
|
||||
t.Errorf("%s: Variant() = reader %v, content type %q, error %v; "+
|
||||
"want no reader and %v", tt.name, reader, contentType, err, tt.wantErr)
|
||||
}
|
||||
}
|
||||
|
||||
reader, _, _, err := cache.Variant(testVariantKeyFour)
|
||||
variant, err := cache.Variant(testVariantKeyFour)
|
||||
|
||||
reader := variant.Content
|
||||
if reader != nil || !errors.Is(err, ErrNotFound) {
|
||||
t.Errorf("Variant() of a variant with no file = reader %v, error %v; "+
|
||||
"want no reader and %v", reader, err, ErrNotFound)
|
||||
@@ -126,7 +132,9 @@ func TestVariant_RefusesFileOfNoOutputFormat(t *testing.T) {
|
||||
|
||||
disabled, _ := newEvictionTestCache(t, 0)
|
||||
|
||||
reader, _, _, err = disabled.Variant(testVariantKeyOne)
|
||||
variant, err = disabled.Variant(testVariantKeyOne)
|
||||
|
||||
reader = variant.Content
|
||||
if reader != nil || !errors.Is(err, ErrNotFound) {
|
||||
t.Errorf("Variant() on a disabled cache = reader %v, error %v; "+
|
||||
"want no reader and %v", reader, err, ErrNotFound)
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
package imgcache
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"os"
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
)
|
||||
|
||||
// TestEvictNotCheckedRecently stores a source and a variant last checked with
|
||||
// upstream 31 days ago and a source and a variant checked 29 days ago.
|
||||
// EvictNotCheckedRecently must remove the rows and files of the first two and
|
||||
// keep the other two.
|
||||
func TestEvictNotCheckedRecently(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, SourceInfo{CheckedAt: checkedAt},
|
||||
bytes.NewReader([]byte("variant")))
|
||||
if err != nil {
|
||||
t.Fatalf("PutVariant(%s) error = %v", key, err)
|
||||
}
|
||||
}
|
||||
|
||||
err := cache.EvictNotCheckedRecently(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("EvictNotCheckedRecently() 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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -131,6 +131,56 @@ func (c *Cache) EvictToLimit(ctx context.Context) error {
|
||||
}
|
||||
}
|
||||
|
||||
// EvictNotCheckedRecently evicts each source and variant whose checked_at is
|
||||
// EvictionAfterLastCheck 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) EvictNotCheckedRecently(ctx context.Context) error {
|
||||
if c.config.MaxBytes == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
cutoff := time.Now().UTC().Add(-EvictionAfterLastCheck)
|
||||
|
||||
for {
|
||||
sources, err := queryStringColumn[imagerequest.SourceKey](ctx, c.db, `
|
||||
SELECT source_key FROM source_content WHERE checked_at < ? LIMIT ?
|
||||
`, "sources not checked recently", "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 not checked recently", "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 not checked with upstream recently",
|
||||
"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 not
|
||||
// checked for EvictionAfterLastCheck, 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.runEvictNotCheckedRecently(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.runEvictNotCheckedRecently(ctx)
|
||||
case <-c.evictionPressure:
|
||||
}
|
||||
|
||||
@@ -474,6 +527,21 @@ func (c *Cache) runCleanExpired(ctx context.Context) {
|
||||
}
|
||||
}
|
||||
|
||||
// runEvictNotCheckedRecently runs EvictNotCheckedRecently, logging a failure
|
||||
// instead of propagating it (the loop must keep running). It does nothing
|
||||
// once ctx is cancelled.
|
||||
func (c *Cache) runEvictNotCheckedRecently(ctx context.Context) {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
|
||||
err := c.EvictNotCheckedRecently(ctx)
|
||||
if err != nil {
|
||||
c.log.Warn("evicting cache content not checked recently 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 +677,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)
|
||||
}
|
||||
|
||||
@@ -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,8 @@ func storeEvictionTestVariant(
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
err := cache.PutVariant(t.Context(), key, bytes.NewReader(content))
|
||||
err := cache.PutVariant(t.Context(), key, SourceInfo{CheckedAt: time.Now()},
|
||||
bytes.NewReader(content))
|
||||
if err != nil {
|
||||
t.Fatalf("PutVariant(%s) failed: %v", key, err)
|
||||
}
|
||||
@@ -432,13 +433,14 @@ func assertDisabledCacheWritesAreNoOps(
|
||||
ctx := t.Context()
|
||||
|
||||
err := cache.PutVariant(
|
||||
ctx, imagerequest.CacheKey(req), bytes.NewReader([]byte("data")),
|
||||
ctx, imagerequest.CacheKey(req), SourceInfo{}, 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)
|
||||
}
|
||||
|
||||
@@ -3,8 +3,10 @@ package imgcache
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Storage constants.
|
||||
@@ -54,26 +56,26 @@ func (s *FileStore) Write(key string, data []byte) error {
|
||||
return writeFileSynced(path, data)
|
||||
}
|
||||
|
||||
// Open opens the file for key and returns it with its size in bytes. A
|
||||
// missing file is ErrNotFound.
|
||||
func (s *FileStore) Open(key string) (*os.File, int64, error) {
|
||||
// Open opens the file for key and returns it with its information, which
|
||||
// gives its size and its modification time. A missing file is ErrNotFound.
|
||||
func (s *FileStore) Open(key string) (*os.File, fs.FileInfo, error) {
|
||||
f, err := os.Open(s.keyToPath(key))
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return nil, 0, ErrNotFound
|
||||
return nil, nil, ErrNotFound
|
||||
}
|
||||
|
||||
return nil, 0, fmt.Errorf("failed to open content: %w", err)
|
||||
return nil, nil, fmt.Errorf("failed to open content: %w", err)
|
||||
}
|
||||
|
||||
stat, err := f.Stat()
|
||||
info, err := f.Stat()
|
||||
if err != nil {
|
||||
_ = f.Close()
|
||||
|
||||
return nil, 0, fmt.Errorf("failed to stat content: %w", err)
|
||||
return nil, nil, fmt.Errorf("failed to stat content: %w", err)
|
||||
}
|
||||
|
||||
return f, stat.Size(), nil
|
||||
return f, info, nil
|
||||
}
|
||||
|
||||
// Size returns the size in bytes of the file for key. A missing file is
|
||||
@@ -91,6 +93,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))
|
||||
|
||||
@@ -38,15 +38,15 @@ func TestFileStore_WriteAndOpen(t *testing.T) {
|
||||
}
|
||||
|
||||
// Open and verify size and content
|
||||
r, size, err := storage.Open(key)
|
||||
r, info, err := storage.Open(key)
|
||||
if err != nil {
|
||||
t.Fatalf("Open() error = %v", err)
|
||||
}
|
||||
|
||||
defer func() { _ = r.Close() }()
|
||||
|
||||
if size != int64(len(content)) {
|
||||
t.Errorf("Open() size = %d, want %d", size, len(content))
|
||||
if info.Size() != int64(len(content)) {
|
||||
t.Errorf("Open() size = %d, want %d", info.Size(), len(content))
|
||||
}
|
||||
|
||||
loaded, err := io.ReadAll(r)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user