3 Commits
Author SHA1 Message Date
clawbot ca512fdb4f Remove the unused cache lifetime setting (closes #69)
check / check (push) Canceled after 0s
CacheConfig.CacheTTL was set to DefaultCacheTTL (24 hours) and never
read, so cached images never expired. Remove the field, the constant
and the line that set it. Cached source and transformed images stay
until eviction removes them to keep the cache within cache_max_bytes.

A new test caches a source and a transformed image, sets every time
pixa stores for them two days back (database rows, metadata files and
file times), and checks that both are still served from the cache
without a new upstream fetch.

Existing tests lose only the CacheTTL line of their cache config.

Model: opus-5-5
2026-10-13 07:58:55 +02:00
clawbot 52f69d3082 Refuse SVG sources and never save an unknown input as JPEG (closes #68)
check / check (push) Canceled after 0s
The upstream fetch no longer accepts image/svg+xml, and both image
routes answer 415, not 500, for any upstream Content-Type outside the
source formats. Before decoding, the image processor names the input's
format from govips' DetermineImageType and refuses, with 415, input
whose first bytes libvips reads as anything but the six source formats,
so an SVG passing the first-bytes check for AVIF never reaches librsvg.
orig of such an input is refused instead of being saved as JPEG.
README.md drops SVG from the source formats.

SVG detection stays in internal/magic, unused on the request path, for
#168. An AVIF image sequence
(brand avis) answers 415.

Model: opus-5-5
2026-10-13 07:13:17 +02:00
clawbot 162a2a33ef Keep counts not written by the request's deadline in memory (closes #224)
check / check (push) Successful in 32m7s
TestService_Get_ReturnsByItsDeadline failed on a busy host because its
request, past its deadline, still waited to write its miss count, a write
with no deadline; in pixad that write waits for the one database
connection every request shares. Each count write now keeps the request's
deadline but not its cancellation. A count not written by then is kept in
memory, where Stats includes it, and one goroutine of the cache writes
those counts one UPDATE at a time, and once more at shutdown before the
database closes. The test phase also runs go test with -parallel 4: on a
busy host, as many tests at once as there are CPUs wait so long to be
scheduled that a timed request can fail.

Model: opus-5-5
2026-10-13 03:07:23 +02:00
28 changed files with 1110 additions and 78 deletions
+6 -3
View File
@@ -32,10 +32,13 @@ RUN script/bootstrap --cgo
COPY . .
# Without -v first; on a failure, again with -v for the details, and
# the step fails even if the second run passes.
RUN go test -count=1 -timeout 90s -race -cover ./... || \
# the step fails even if the second run passes. -parallel 4: by default
# a package runs as many of its tests at once as the host has CPUs, and
# on a busy host they then wait so long to be scheduled that a test
# that times a request can see it return a second late.
RUN go test -count=1 -timeout 90s -race -parallel 4 -cover ./... || \
{ echo "--- Rerunning with -v for details ---"; \
go test -count=1 -timeout 90s -race -v ./...; exit 1; }
go test -count=1 -timeout 90s -race -parallel 4 -v ./...; exit 1; }
# Build stage. Nothing is wanted from the two phases above: these copies
# make BuildKit build them first, so this stage runs only when lint and
+18 -11
View File
@@ -84,10 +84,12 @@ which answers 200 whenever pixa is running, in maintenance mode too (see
`maintenance_mode`).
On SIGTERM or SIGINT pixa stops accepting connections, gives the requests in
progress and the images being processed 5 seconds to finish, and exits: with 0,
or with 1 when images were still being processed after those 5 seconds or
another part of pixa failed to stop. A request not finished by then is cut off.
`docker stop` waits 10 seconds before it kills the container.
progress and the images being processed 5 seconds to finish, then writes to the
database the counts of cache hits, misses, fetches and conversions that requests
could not write by their deadline, and exits: with 0, or with 1 when images were
still being processed after those 5 seconds, some of those counts could not be
written, or another part of pixa failed to stop. A request not finished by then
is cut off. `docker stop` waits 10 seconds before it kills the container.
Outside Docker, pixa needs libvips (the image has 8.16) and libheif to run, as
it uses libvips through CGO. pixad does not start unless libvips has its JPEG XL
@@ -188,9 +190,12 @@ path under `/v1/` answers 200, in maintenance mode too.
the cache and the upstream fetch; 403 when the upstream host, or a host it
redirects to, is `localhost`, ends in `.localhost` or `.local`, or has an
address in a blocked network (see `blocked_networks`); 429 over
`image_requests_per_minute` (below); 502 when the upstream answered with an
error status, and for 5 minutes after that for the same source URL; 503 when
pixa is busy or in maintenance mode; 500 for any other failure.
`image_requests_per_minute` (below); 415 when the upstream answers with a
`Content-Type` that is not one of the source formats (below), or with an image
whose first bytes libvips reads as another format; 502 when the upstream
answered with an error status, and for 5 minutes after that for the same
source URL; 503 when pixa is busy or in maintenance mode; 500 for any other
failure.
- `GET` or `HEAD` `/v1/e/<token>/<name>` — an image through an encrypted URL
(see Encrypted URLs). Needs: nothing but the URL. Answers: 200; 304 when
`If-None-Match` matches the image's `ETag`; 400 for a token that does not
@@ -198,7 +203,7 @@ path under `/v1/` answers 200, in maintenance mode too.
expired; 504 when the upstream has not sent its response headers within
`upstream_fetch_timeout`, but 500 when that time runs out while the image
itself is still arriving; 400 for an `Accept` header that is not valid, and
406, 403, 429, 502, 503 and 500, as for `/v1/image/`.
406, 403, 415, 429, 502, 503 and 500, as for `/v1/image/`.
- `GET /robots.txt` — asks every crawler to stay away (`Disallow: /`). Needs:
nothing. Answers: 200.
- `GET /.well-known/healthcheck.json` — JSON with `status` (`ok`), `now`,
@@ -258,9 +263,11 @@ once, is refused with 400.
both optional (values under Signature Specification). Both are part of what is
cached, so each value of either is a separate cached image.
The source image may be JPEG, PNG, GIF, WebP, AVIF, JPEG XL or SVG. The
upstream's `Content-Type` must name its format, and the image's first bytes must
match it.
The source image may be JPEG, PNG, GIF, WebP, AVIF or JPEG XL. The upstream's
`Content-Type` must name its format, and the image's first bytes must match it.
An upstream answering with any other `Content-Type`, SVG included, is refused
with 415, and so is an image whose first bytes libvips reads as another format,
such as an SVG whose first bytes pass the check for AVIF.
With the format `auto`, pixa chooses the format for each request from its
`Accept` header, in this order:
+34
View File
@@ -30,6 +30,27 @@ P2: security: per-origin rate limiting
# Completed Steps
- 2026-10-13 the unused cache lifetime is removed (closes #69):
`CacheConfig.CacheTTL`, which pixad set to `DefaultCacheTTL` (24 hours), was
never read, so cached images never expired. The field and the constant are
removed. Cached source and transformed images stay until eviction removes them
to keep the cache within `cache_max_bytes`.
- 2026-10-13 SVG sources are refused, and `orig` is never saved as JPEG in place
of the source's own format (closes #68): the upstream fetch no longer accepts
`image/svg+xml`, which libvips would render. Both image routes answer 415 for
an upstream `Content-Type` that is not one of the source formats, SVG
included, where they answered 500. Before decoding, the image processor
refuses with `ErrUnsupportedInputFormat`, also answered 415, any input that
govips' `DetermineImageType`, which picks the libvips loader from the first
bytes, does not name as a source format. That catches an SVG whose first bytes
pass the first-bytes check for another type, as an XML comment can make them
do for AVIF; an SVG whose first bytes do not pass that check answers 500, as
before, and so does a damaged image in a source format, whose first bytes
`DetermineImageType` names. `orig` is always the source's own format, where a
source the processor could not name was saved as JPEG. `README.md` no longer
lists SVG. The SVG detection in `internal/magic` stays: the request path does
not call it, and it goes with the unused code of
https://git.eeqj.de/sneak/pixa/issues/168.
- 2026-10-12 per-client limit on the image routes (closes #236): the new setting
`image_requests_per_minute` (`PIXA_IMAGE_REQUESTS_PER_MINUTE`), default 600,
is how many `GET` and `HEAD` requests one client may send to `/v1/image/` and
@@ -38,6 +59,19 @@ P2: security: per-origin rate limiting
clients are told apart the same way, and runs after the CORS middleware, so a
preflight is not counted. `0` turns the limit off; a negative or non-integer
value aborts startup.
- 2026-10-08 a request past its deadline no longer waits for the database to
count it (closes #224). On a busy host, `TestService_Get_ReturnsByItsDeadline`
failed because its request, past its deadline, still waited for its miss count
to be written, a write with no deadline, which in pixad waits for the one
database connection every request shares. Each count write now keeps the
request's deadline but not its cancellation. A count not written by then is
kept in memory, where `Stats` sees it, and one goroutine of the cache writes
those counts, one write at a time, and once more at shutdown before the
database closes. The test phase of the `Dockerfile` also runs at most 4 tests
of a package at once (`go test -parallel 4`), where it ran as many as the host
has CPUs: on a busy host they then wait so long to be scheduled that a test
that times a request can fail. `-p`, how many packages are tested at once, is
unchanged, as capping it also slows compiling.
- 2026-10-08 every output format is saved with settings pixa sets on purpose
(closes #232): each format has its own govips export, as JPEG XL does, in
place of govips' generic `Export`, which sent libvips a zero for some settings
+11 -3
View File
@@ -4,6 +4,7 @@ package handlers
import (
"context"
"encoding/json"
"errors"
"log/slog"
"net/http"
"time"
@@ -71,16 +72,21 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
}
lc.Append(fx.Hook{
//nolint:contextcheck // the eviction loop outlives OnStart; OnStop cancels it
//nolint:contextcheck // the cache's goroutines outlive OnStart; OnStop stops them
OnStart: func(_ context.Context) error {
return s.initImageService()
},
// The pending counts are written here, before the database's stop
// hook closes it.
OnStop: func(ctx context.Context) error {
if s.imgCache == nil {
return nil
}
return s.imgCache.StopEviction(ctx)
return errors.Join(
s.imgCache.StopEviction(ctx),
s.imgCache.StopPendingCountWrites(ctx),
)
},
})
@@ -100,7 +106,6 @@ func (s *Handlers) WaitForProcessing(ctx context.Context) int {
func newCacheConfig(cfg *config.Config, log *slog.Logger) imgcache.CacheConfig {
return imgcache.CacheConfig{
StateDir: cfg.StateDir,
CacheTTL: imgcache.DefaultCacheTTL,
NegativeTTL: imgcache.DefaultNegativeTTL,
MaxBytes: cfg.CacheMaxBytes,
UseDefaultMaxBytes: !cfg.CacheMaxBytesExplicit,
@@ -122,6 +127,9 @@ func (s *Handlers) initImageService() error {
// write-pressure passes. No-op when the disk cache is disabled.
cache.StartEviction(imgcache.DefaultEvictionInterval)
// Writes the counts requests could not write by their deadline
cache.StartPendingCountWrites()
// Create the fetcher config
fetcherCfg := httpfetcher.DefaultConfig()
fetcherCfg.AllowHTTP = s.config.AllowHTTP
@@ -47,7 +47,6 @@ func setupTestHandler(t *testing.T) *testFixtures {
cache, err := imgcache.NewCache(db, imgcache.CacheConfig{
StateDir: tmpDir,
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute,
})
if err != nil {
+8
View File
@@ -247,6 +247,14 @@ func (s *Handlers) respondImageError(
return
}
if errors.Is(err, httpfetcher.ErrInvalidContentType) ||
errors.Is(err, imageprocessor.ErrUnsupportedInputFormat) {
s.respondError(w, "unsupported source format",
http.StatusUnsupportedMediaType)
return
}
s.respondError(w, "internal error", http.StatusInternalServerError)
}
@@ -31,7 +31,6 @@ func newSignedHostServer(
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
StateDir: t.TempDir(),
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute,
})
if err != nil {
@@ -29,7 +29,6 @@ func newImageRoute(t *testing.T, fetcher httpfetcher.Fetcher) http.Handler {
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
StateDir: t.TempDir(),
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute,
})
if err != nil {
+4
View File
@@ -161,6 +161,10 @@ func (s *Handlers) handleImageError(w http.ResponseWriter, err error) {
errors.Is(err, imageprocessor.ErrTooManyImages):
s.respondError(w, "server busy, try again later",
http.StatusServiceUnavailable)
case errors.Is(err, httpfetcher.ErrInvalidContentType),
errors.Is(err, imageprocessor.ErrUnsupportedInputFormat):
s.respondError(w, "unsupported source format",
http.StatusUnsupportedMediaType)
default:
s.log.Error("image request failed", "error", err)
s.respondError(w, "internal error", http.StatusInternalServerError)
@@ -49,7 +49,6 @@ func newRefererRoutes(t *testing.T) (http.Handler, *Handlers, *countingFetcher)
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
StateDir: t.TempDir(),
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute,
})
if err != nil {
@@ -0,0 +1,54 @@
package handlers
import (
"fmt"
"log/slog"
"net/http"
"net/http/httptest"
"strings"
"testing"
"sneak.berlin/go/pixa/internal/httpfetcher"
"sneak.berlin/go/pixa/internal/imageprocessor"
"sneak.berlin/go/pixa/internal/imgcache"
)
// TestUnsupportedSourceFormatAnswers415 checks that both image routes answer
// 415 with a clear error when the upstream answers with a content type pixa
// does not accept, such as SVG, and when orig is asked of a source in a
// format the processor cannot name, wrapped as the service wraps them.
func TestUnsupportedSourceFormatAnswers415(t *testing.T) {
t.Parallel()
h := &Handlers{log: slog.New(slog.DiscardHandler)}
req := &imgcache.ImageRequest{SourceHost: "img.example.com", SourcePath: "/a.svg"}
for _, err := range []error{
fmt.Errorf("upstream fetch failed: %w: image/svg+xml",
httpfetcher.ErrInvalidContentType),
fmt.Errorf("image processing failed: %w: unknown",
imageprocessor.ErrUnsupportedInputFormat),
} {
plain := httptest.NewRecorder()
h.respondImageError(plain, req, err)
encrypted := httptest.NewRecorder()
h.handleImageError(encrypted, err)
for route, rec := range map[string]*httptest.ResponseRecorder{
"HandleImage": plain, "HandleImageEnc": encrypted,
} {
t.Logf("%s for %v: %d %s", route, err, rec.Code, rec.Body.String())
if rec.Code != http.StatusUnsupportedMediaType {
t.Errorf("%s for %v: status = %d, want %d",
route, err, rec.Code, http.StatusUnsupportedMediaType)
}
if !strings.Contains(rec.Body.String(), "unsupported source format") {
t.Errorf("%s for %v: body = %q, want the unsupported source format error",
route, err, rec.Body.String())
}
}
}
}
-1
View File
@@ -158,7 +158,6 @@ func DefaultConfig() *Config {
contentTypeWebP,
contentTypeAVIF,
contentTypeJXL,
contentTypeSVG,
},
AllowHTTP: false,
MaxConnectionsPerHost: DefaultMaxConnectionsPerHost,
+31 -17
View File
@@ -137,6 +137,12 @@ var ErrInputDataTooLarge = errors.New("input data exceeds maximum allowed size")
// not supported.
var ErrUnsupportedOutputFormat = errors.New("unsupported output format")
// ErrUnsupportedInputFormat is returned, before the input is decoded, when
// libvips reads the input's first bytes as another format than the source
// formats, or as none: for example an SVG whose first bytes pass the
// first-bytes check for AVIF.
var ErrUnsupportedInputFormat = errors.New("unsupported input format")
// ErrTooManyImages is returned when MaxConcurrentProcessing images are being
// processed and none finishes within ProcessingWaitTimeout.
var ErrTooManyImages = errors.New("too many images being processed at once")
@@ -219,6 +225,17 @@ func (p *ImageProcessor) Process(
return nil, ErrInputDataTooLarge
}
// govips picks the libvips loader from the input's first bytes with
// DetermineImageType. Input whose first bytes it reads as no source
// format, such as SVG, which librsvg would render, is refused before
// libvips reads it.
inputFormat := p.detectFormat(vips.DetermineImageType(data))
sourceFormat, err := p.formatFromString(inputFormat)
if err != nil {
return nil, err
}
// Decode image
img, err := vips.NewImageFromBuffer(data)
if err != nil {
@@ -237,9 +254,6 @@ func (p *ImageProcessor) Process(
origWidth := img.Width()
origHeight := img.Height()
// Detect input format
inputFormat := p.detectFormat(img)
// Validate input dimensions to prevent DoS via decompression bombs
if origWidth > MaxInputDimension || origHeight > MaxInputDimension {
return nil, ErrInputTooLarge
@@ -259,7 +273,7 @@ func (p *ImageProcessor) Process(
// orig is the source's own format; encode refuses an empty format
outputFormat := req.Format
if outputFormat == FormatOriginal {
outputFormat = p.formatFromString(inputFormat)
outputFormat = sourceFormat
}
// Encode to target format
@@ -403,10 +417,9 @@ func (p *ImageProcessor) acquireSlot(ctx context.Context) (func(), error) {
}
}
// detectFormat returns the format string from a vips image.
func (p *ImageProcessor) detectFormat(img *vips.ImageRef) string {
format := img.Format()
switch format {
// detectFormat returns the format string for a govips image type.
func (p *ImageProcessor) detectFormat(imageType vips.ImageType) string {
switch imageType {
case vips.ImageTypeJPEG:
return "jpeg"
case vips.ImageTypePNG:
@@ -687,22 +700,23 @@ func jxlDistance(quality int) float64 {
return 53.0/3000.0*q*q - 23.0/20.0*q + 25.0
}
// formatFromString converts a format string to Format.
func (p *ImageProcessor) formatFromString(format string) Format {
// formatFromString converts a format string from detectFormat to Format. A
// format detectFormat could not name is ErrUnsupportedInputFormat.
func (p *ImageProcessor) formatFromString(format string) (Format, error) {
switch format {
case "jpeg":
return FormatJPEG
return FormatJPEG, nil
case "png":
return FormatPNG
return FormatPNG, nil
case "gif":
return FormatGIF
return FormatGIF, nil
case "webp":
return FormatWebP
return FormatWebP, nil
case string(FormatAVIF):
return FormatAVIF
return FormatAVIF, nil
case string(FormatJXL):
return FormatJXL
return FormatJXL, nil
default:
return FormatJPEG
return "", fmt.Errorf("%w: %s", ErrUnsupportedInputFormat, format)
}
}
@@ -0,0 +1,39 @@
package imageprocessor
import (
"bytes"
"errors"
"testing"
"github.com/davidbyttow/govips/v2/vips"
)
// TestImageProcessor_OrigOfUnnamedFormatRefused verifies that orig of a TIFF
// input, which is not a source format, is refused with
// ErrUnsupportedInputFormat before it is decoded, and not saved as JPEG.
func TestImageProcessor_OrigOfUnnamedFormatRefused(t *testing.T) {
t.Parallel()
img, err := vips.NewImageFromBuffer(createTestPNG(t, 20, 20))
if err != nil {
t.Fatalf("failed to decode test PNG: %v", err)
}
t.Cleanup(img.Close)
tiff, _, err := img.ExportTiff(vips.NewTiffExportParams())
if err != nil {
t.Fatalf("failed to save test TIFF: %v", err)
}
result, err := New(Params{}).Process(
t.Context(), bytes.NewReader(tiff), &Request{Format: FormatOriginal},
)
if result != nil {
t.Logf("Process() returned %s", result.ContentType)
}
if !errors.Is(err, ErrUnsupportedInputFormat) {
t.Errorf("Process() error = %v, want %v", err, ErrUnsupportedInputFormat)
}
}
@@ -0,0 +1,46 @@
package imageprocessor
import (
"bytes"
"io"
"testing"
)
// TestImageProcessor_LoadsEachSourceFormat verifies that an image in each of
// the six source formats, saved by the processor itself, is loaded and saved
// again in its own format (orig).
func TestImageProcessor_LoadsEachSourceFormat(t *testing.T) {
t.Parallel()
proc := New(Params{})
for _, format := range []Format{
FormatJPEG, FormatPNG, FormatGIF, FormatWebP, FormatAVIF, FormatJXL,
} {
saved, err := proc.Process(t.Context(),
bytes.NewReader(createTestPNG(t, 20, 20)), &Request{Format: format})
if err != nil {
t.Fatalf("saving the test image as %s: %v", format, err)
}
source, err := io.ReadAll(saved.Content)
if err != nil {
t.Fatalf("reading the %s image: %v", format, err)
}
result, err := proc.Process(t.Context(), bytes.NewReader(source),
&Request{Format: FormatOriginal})
if err != nil {
t.Errorf("orig of the %s image: %v", format, err)
continue
}
t.Logf("orig of the %s image: %s", format, result.ContentType)
if result.ContentType != FormatToMIME(format) {
t.Errorf("orig of the %s image is %s, want %s",
format, result.ContentType, FormatToMIME(format))
}
}
}
+83 -15
View File
@@ -11,6 +11,8 @@ import (
"io"
"log/slog"
"path/filepath"
"sync"
"sync/atomic"
"time"
lru "github.com/hashicorp/golang-lru/v2"
@@ -33,7 +35,6 @@ const metaCacheSize = 10000
// CacheConfig holds cache configuration.
type CacheConfig struct {
StateDir string
CacheTTL time.Duration
NegativeTTL time.Duration
// MaxBytes is the disk cache size limit in bytes that eviction
@@ -79,6 +80,27 @@ type Cache struct {
evictionDone chan struct{}
evictionCancel context.CancelFunc
// The pending counts: hits, misses, upstream fetches and transforms
// whose write to the database missed its request's deadline, kept
// here until they are written by the goroutine that
// StartPendingCountWrites starts. As for eviction, the channels are
// created in NewCache: pendingCountsAdded wakes that goroutine, and
// pendingCountsDone is closed when it returns. pendingCountsCancel,
// set by StartPendingCountWrites, cancels its context, which tells it
// to finish.
pendingHits atomic.Int64
pendingMisses atomic.Int64
pendingUpstreamFetches atomic.Int64
pendingUpstreamFetchBytes atomic.Int64
pendingTransforms atomic.Int64
pendingCountsAdded chan struct{}
pendingCountsDone chan struct{}
pendingCountsCancel context.CancelFunc
// Held through each write of the pending counts, and by
// readHitAndMissCounts to read them
pendingCountsWriteMutex sync.Mutex
// metaCache holds the content types of the variants most recently
// stored or served, so a hit does not read the variant's .meta file.
// It never stands in for the variant file, which is always opened.
@@ -139,6 +161,9 @@ func newCache(
metaCache: metaCache,
contentLocks: newContentLock(),
pendingCountsAdded: make(chan struct{}, 1),
pendingCountsDone: make(chan struct{}),
reconciliationPageSize: defaultReconciliationPageSize,
}
@@ -474,12 +499,7 @@ func (c *Cache) CleanExpired(ctx context.Context) error {
func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) {
var stats CacheStats
// Fetch hit/miss counts from the stats table
err := c.db.QueryRowContext(ctx, `
SELECT hit_count, miss_count
FROM cache_stats WHERE id = 1
`).Scan(&stats.HitCount, &stats.MissCount)
err := c.readHitAndMissCounts(ctx, &stats)
if err != nil && !errors.Is(err, sql.ErrNoRows) {
return nil, fmt.Errorf("failed to get cache stats: %w", err)
}
@@ -511,19 +531,30 @@ func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) {
}
// IncrementStats counts a cache hit or miss, and an upstream fetch that read
// fetchBytes bytes, as IncrementUpstreamFetch does.
// fetchBytes bytes, as IncrementUpstreamFetch does. Like the other Increment
// methods, it writes to the database with ctx's deadline but not its
// cancellation, so an ended request is still counted but does not wait for
// the database past its deadline; a count not written by then becomes a
// pending count (see StartPendingCountWrites).
func (c *Cache) IncrementStats(ctx context.Context, hit bool, fetchBytes int64) {
countCtx, cancel := withoutCancelKeepingDeadline(ctx)
defer cancel()
var err error
pendingCount := &c.pendingMisses
if hit {
_, err = c.db.ExecContext(ctx, `
pendingCount = &c.pendingHits
_, err = c.db.ExecContext(countCtx, `
UPDATE cache_stats
SET hit_count = hit_count + 1,
last_updated_at = CURRENT_TIMESTAMP
WHERE id = 1
`)
} else {
_, err = c.db.ExecContext(ctx, `
_, err = c.db.ExecContext(countCtx, `
UPDATE cache_stats
SET miss_count = miss_count + 1,
last_updated_at = CURRENT_TIMESTAMP
@@ -531,7 +562,10 @@ func (c *Cache) IncrementStats(ctx context.Context, hit bool, fetchBytes int64)
`)
}
if err != nil {
switch {
case errors.Is(err, context.DeadlineExceeded):
c.addPendingCount(pendingCount, 1)
case err != nil:
c.log.Warn("failed to count cache hit or miss", "hit", hit, "error", err)
}
@@ -545,14 +579,22 @@ func (c *Cache) IncrementUpstreamFetch(ctx context.Context, fetchBytes int64) {
return
}
_, err := c.db.ExecContext(ctx, `
countCtx, cancel := withoutCancelKeepingDeadline(ctx)
defer cancel()
_, err := c.db.ExecContext(countCtx, `
UPDATE cache_stats
SET upstream_fetch_count = upstream_fetch_count + 1,
upstream_fetch_bytes = upstream_fetch_bytes + ?,
last_updated_at = CURRENT_TIMESTAMP
WHERE id = 1
`, fetchBytes)
if err != nil {
switch {
case errors.Is(err, context.DeadlineExceeded):
c.addPendingCount(&c.pendingUpstreamFetches, 1)
c.addPendingCount(&c.pendingUpstreamFetchBytes, fetchBytes)
case err != nil:
c.log.Warn("failed to count upstream fetch",
"fetch_bytes", fetchBytes, "error", err)
}
@@ -560,17 +602,43 @@ func (c *Cache) IncrementUpstreamFetch(ctx context.Context, fetchBytes int64) {
// IncrementTransformCount counts one image transcoded by the image processor.
func (c *Cache) IncrementTransformCount(ctx context.Context) {
_, err := c.db.ExecContext(ctx, `
countCtx, cancel := withoutCancelKeepingDeadline(ctx)
defer cancel()
_, err := c.db.ExecContext(countCtx, `
UPDATE cache_stats
SET transform_count = transform_count + 1,
last_updated_at = CURRENT_TIMESTAMP
WHERE id = 1
`)
if err != nil {
switch {
case errors.Is(err, context.DeadlineExceeded):
c.addPendingCount(&c.pendingTransforms, 1)
case err != nil:
c.log.Warn("failed to count transform", "error", err)
}
}
// readHitAndMissCounts sets the hit and miss counts of stats to those of the
// cache_stats row plus the pending hits and misses (see
// StartPendingCountWrites). It holds pendingCountsWriteMutex, so that no
// write of the pending counts falls between the two reads.
func (c *Cache) readHitAndMissCounts(ctx context.Context, stats *CacheStats) error {
c.pendingCountsWriteMutex.Lock()
defer c.pendingCountsWriteMutex.Unlock()
err := c.db.QueryRowContext(ctx, `
SELECT hit_count, miss_count
FROM cache_stats WHERE id = 1
`).Scan(&stats.HitCount, &stats.MissCount)
stats.HitCount += c.pendingHits.Load()
stats.MissCount += c.pendingMisses.Load()
return err
}
// loadVariantWithMeta is GetVariant for a variant metaCache does not
// hold: it reads the content type from the variant's .meta file and
// keeps it in metaCache, unless a StoreVariant has put one there
-2
View File
@@ -102,7 +102,6 @@ func setupTestCache(t *testing.T) *Cache {
cache, err := NewCache(db, CacheConfig{
StateDir: tmpDir,
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute,
})
if err != nil {
@@ -241,7 +240,6 @@ func TestCache_NegativeCacheExpiry(t *testing.T) {
// Very short negative TTL for testing
cache, err := NewCache(db, CacheConfig{
StateDir: tmpDir,
CacheTTL: time.Hour,
NegativeTTL: 1 * time.Millisecond,
})
if err != nil {
@@ -70,7 +70,6 @@ func newEvictionTestCache(t *testing.T, maxBytes int64) (*Cache, string) {
// backwards compatibility with existing fixtures.
cache, err := NewCache(db, CacheConfig{
StateDir: tmpDir,
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute,
MaxBytes: maxBytes,
DisableDiskCache: maxBytes == 0,
-1
View File
@@ -4,6 +4,5 @@ import "time"
// CacheConfig defaults.
const (
DefaultCacheTTL = 24 * time.Hour
DefaultNegativeTTL = 5 * time.Minute
)
@@ -15,7 +15,6 @@ func TestNegativeCache_StoreAndCheck(t *testing.T) {
cache, err := NewCache(db, CacheConfig{
StateDir: dir,
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute,
})
if err != nil {
@@ -63,7 +62,6 @@ func TestNegativeCache_Expired(t *testing.T) {
cache, err := NewCache(db, CacheConfig{
StateDir: dir,
CacheTTL: time.Hour,
NegativeTTL: 1 * time.Millisecond, // very short TTL
})
if err != nil {
+149
View File
@@ -0,0 +1,149 @@
package imgcache
import (
"context"
"fmt"
"sync/atomic"
)
// StartPendingCountWrites starts the goroutine that writes the pending
// counts to the database whenever there are some, one UPDATE at a time:
// however many requests pass their deadline before their counts are
// written, at most this one write waits for the database. It is a no-op
// when already started. The goroutine outlives the caller, so it runs with
// its own context, which StopPendingCountWrites cancels to tell it to
// finish.
func (c *Cache) StartPendingCountWrites() {
if c.pendingCountsCancel != nil {
return
}
ctx, cancel := context.WithCancel(context.Background())
c.pendingCountsCancel = cancel
go func() {
c.pendingCountWriteLoop(ctx)
}()
}
// StopPendingCountWrites tells the goroutine StartPendingCountWrites
// started, if it did, to finish, and waits for its write under way to end
// and for it to return. It then writes the pending counts left, so that
// they are written before the database closes. It waits for all of this at
// most until ctx ends, then logs the counts not written and returns an
// error.
func (c *Cache) StopPendingCountWrites(ctx context.Context) error {
var err error
if c.pendingCountsCancel != nil {
c.pendingCountsCancel()
select {
case <-c.pendingCountsDone:
case <-ctx.Done():
// The write given up on here may not reach the database:
// database/sql closes without waiting for a query, and a
// write still waiting for the connection fails when the
// database closes.
err = fmt.Errorf("pending counts still being written: %w", ctx.Err())
}
}
if err == nil {
err = c.writePendingCounts(ctx)
}
if err != nil {
c.log.Error("counts not written at shutdown",
"hits", c.pendingHits.Load(),
"misses", c.pendingMisses.Load(),
"upstream_fetches", c.pendingUpstreamFetches.Load(),
"upstream_fetch_bytes", c.pendingUpstreamFetchBytes.Load(),
"transforms", c.pendingTransforms.Load(),
"error", err,
)
return err
}
return nil
}
// pendingCountWriteLoop is the body of the goroutine
// StartPendingCountWrites starts. Once ctx is cancelled it starts no write
// and returns, but does not cut off a write under way then. A write that
// fails leaves the counts pending, for the next write.
func (c *Cache) pendingCountWriteLoop(ctx context.Context) {
defer close(c.pendingCountsDone)
for {
select {
case <-ctx.Done():
return
case <-c.pendingCountsAdded:
}
// select picks at random when ctx is cancelled and a wakeup
// waits too.
if ctx.Err() != nil {
return
}
err := c.writePendingCounts(context.WithoutCancel(ctx))
if err != nil {
c.log.Warn("failed to write pending counts", "error", err)
}
}
}
// addPendingCount adds n to pendingCount, one of the pending counts, and
// wakes the goroutine that writes them. pendingCountsAdded has capacity one,
// so a wakeup already waiting is enough.
func (c *Cache) addPendingCount(pendingCount *atomic.Int64, n int64) {
pendingCount.Add(n)
select {
case c.pendingCountsAdded <- struct{}{}:
default:
}
}
// writePendingCounts adds the pending counts to the cache_stats row in one
// UPDATE, then takes what it wrote out of them, leaving any added
// meanwhile. When the UPDATE fails, they all stay pending.
func (c *Cache) writePendingCounts(ctx context.Context) error {
c.pendingCountsWriteMutex.Lock()
defer c.pendingCountsWriteMutex.Unlock()
hits := c.pendingHits.Load()
misses := c.pendingMisses.Load()
upstreamFetches := c.pendingUpstreamFetches.Load()
upstreamFetchBytes := c.pendingUpstreamFetchBytes.Load()
transforms := c.pendingTransforms.Load()
if hits+misses+upstreamFetches+upstreamFetchBytes+transforms == 0 {
return nil
}
_, err := c.db.ExecContext(ctx, `
UPDATE cache_stats
SET hit_count = hit_count + ?,
miss_count = miss_count + ?,
upstream_fetch_count = upstream_fetch_count + ?,
upstream_fetch_bytes = upstream_fetch_bytes + ?,
transform_count = transform_count + ?,
last_updated_at = CURRENT_TIMESTAMP
WHERE id = 1
`, hits, misses, upstreamFetches, upstreamFetchBytes, transforms)
if err != nil {
return fmt.Errorf("failed to write pending counts: %w", err)
}
c.pendingHits.Add(-hits)
c.pendingMisses.Add(-misses)
c.pendingUpstreamFetches.Add(-upstreamFetches)
c.pendingUpstreamFetchBytes.Add(-upstreamFetchBytes)
c.pendingTransforms.Add(-transforms)
return nil
}
@@ -0,0 +1,327 @@
package imgcache
import (
"context"
"database/sql"
"errors"
"testing"
"time"
)
// pendingCountsWait is how long a test waits for the database's connection,
// for a count to reach the database, for a write to start waiting for the
// connection, for a request to return past its deadline, or for
// StopPendingCountWrites to return.
const pendingCountsWait = 5 * time.Second
// otherWritesWait is how long a test gives other writes, once one write waits
// for the database's connection, to start waiting for it too.
const otherWritesWait = 100 * time.Millisecond
// holdDatabase takes the one connection of cache's database, so that every
// other query waits for it, and returns the func that frees it.
func holdDatabase(t *testing.T, cache *Cache) func() {
t.Helper()
ctx, cancel := context.WithTimeout(t.Context(), pendingCountsWait)
defer cancel()
conn, err := cache.db.Conn(ctx)
if err != nil {
t.Fatalf("failed to take the database connection: %v", err)
}
return func() { _ = conn.Close() }
}
// waitForConnectionWaits waits until db.Stats().WaitCount, the number of
// times a caller has waited for a connection, reaches waits.
func waitForConnectionWaits(t *testing.T, db *sql.DB, waits int64) {
t.Helper()
deadline := time.Now().Add(pendingCountsWait)
for db.Stats().WaitCount < waits {
if time.Now().After(deadline) {
t.Fatalf("callers waited for the database connection %d times, want %d",
db.Stats().WaitCount, waits)
}
time.Sleep(10 * time.Millisecond)
}
}
// waitForCounters waits until the cache_stats row holds want.
func waitForCounters(t *testing.T, cache *Cache, want cacheStatsCounters) {
t.Helper()
deadline := time.Now().Add(pendingCountsWait)
for {
got := readCacheStatsCounters(t, cache)
if got == want {
return
}
if time.Now().After(deadline) {
t.Fatalf("counters = %+v, want %+v", got, want)
}
time.Sleep(10 * time.Millisecond)
}
}
// TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy holds the
// database's one connection while a request whose fetch is held reaches its
// deadline. The request must return with the deadline's error while the
// connection is still held, as it waits for no database write past its
// deadline, and its miss must reach the database once the connection is free.
func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) {
t.Parallel()
svc, fixtures, fetcher := setupHeldFetchService(t)
svc.cache.StartPendingCountWrites()
defer func() { _ = svc.cache.StopPendingCountWrites(t.Context()) }()
// Room for the request to reach its fetch on a busy host
const timeout = 2 * time.Second
ctx, cancel := context.WithTimeout(t.Context(), timeout)
defer cancel()
deadline, _ := ctx.Deadline()
results := startGet(ctx, svc, photoVariant(fixtures, 85, FitCover))
// The request has made its database reads by the time it fetches.
select {
case <-fetcher.started:
case <-time.After(timeout):
t.Fatal("request did not reach its fetch by its deadline")
}
releaseDatabase := holdDatabase(t, svc.cache)
defer releaseDatabase()
select {
case got := <-results:
t.Logf("Get() returned %v after its deadline, error = %v",
time.Since(deadline), got.err)
if !errors.Is(got.err, context.DeadlineExceeded) {
t.Errorf("Get() error = %v, want %v", got.err, context.DeadlineExceeded)
}
case <-time.After(time.Until(deadline) + pendingCountsWait):
t.Fatal("request did not return while the database was busy")
}
releaseDatabase()
waitForCounters(t, svc.cache, cacheStatsCounters{missCount: 1})
}
// TestIncrementStats_ManyCountsPastTheirDeadlineLeaveOneWriteWaiting holds the
// database's one connection while many misses are counted past their
// deadline. At most one write may then wait for the connection, and every
// miss must reach the database once it is free.
func TestIncrementStats_ManyCountsPastTheirDeadlineLeaveOneWriteWaiting(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
cache.StartPendingCountWrites()
defer func() { _ = cache.StopPendingCountWrites(t.Context()) }()
releaseDatabase := holdDatabase(t, cache)
defer releaseDatabase()
waitsBefore := cache.db.Stats().WaitCount
// Writes given a context past its deadline fail at once.
ended, cancel := context.WithDeadline(t.Context(), time.Now())
defer cancel()
const misses = 20
for range misses {
cache.IncrementStats(ended, false, 0)
}
// Once one write waits for the connection, any others start waiting too.
waitForConnectionWaits(t, cache.db, waitsBefore+1)
time.Sleep(otherWritesWait)
waits := cache.db.Stats().WaitCount - waitsBefore
t.Logf("%d writes waited for the database", waits)
if waits > 1 {
t.Errorf("%d writes waited for the database, want at most 1", waits)
}
releaseDatabase()
waitForCounters(t, cache, cacheStatsCounters{missCount: misses})
}
// TestStats_IncludesPendingCounts counts hits and misses whose writes miss
// their deadline, and checks that Stats adds them to the database's counts
// before they are written.
func TestStats_IncludesPendingCounts(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
_, err := cache.db.ExecContext(t.Context(),
`UPDATE cache_stats SET hit_count = 75, miss_count = 25 WHERE id = 1`)
if err != nil {
t.Fatal(err)
}
// Writes given a context past its deadline fail at once.
ended, cancel := context.WithDeadline(t.Context(), time.Now())
defer cancel()
cache.IncrementStats(ended, true, 0)
cache.IncrementStats(ended, true, 0)
cache.IncrementStats(ended, false, 0)
stats, err := cache.Stats(t.Context())
if err != nil {
t.Fatalf("Stats() error = %v", err)
}
if stats.HitCount != 77 || stats.MissCount != 26 {
t.Errorf("HitCount = %d, MissCount = %d, want 77 and 26",
stats.HitCount, stats.MissCount)
}
want := cacheStatsCounters{hitCount: 75, missCount: 25}
if got := readCacheStatsCounters(t, cache); got != want {
t.Errorf("counters in the database = %+v, want %+v", got, want)
}
}
// TestStopPendingCountWrites_WritesPendingCounts holds the database's one
// connection while the goroutine that writes the pending counts waits for it,
// counts more, then stops that goroutine. The stop must leave the write under
// way to finish rather than cut it off, and once the connection is free it
// must write the counts that write left.
func TestStopPendingCountWrites_WritesPendingCounts(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
cache.StartPendingCountWrites()
defer func() { _ = cache.StopPendingCountWrites(t.Context()) }()
releaseDatabase := holdDatabase(t, cache)
defer releaseDatabase()
waitsBefore := cache.db.Stats().WaitCount
// Writes given a context past its deadline fail at once.
ended, cancel := context.WithDeadline(t.Context(), time.Now())
defer cancel()
cache.IncrementStats(ended, true, 0)
// The goroutine's write waits for the connection.
waitForConnectionWaits(t, cache.db, waitsBefore+1)
// Counted after that write read the pending counts
cache.IncrementStats(ended, false, 0)
cache.IncrementUpstreamFetch(ended, 1024)
cache.IncrementTransformCount(ended)
// The goroutine is told to finish here as well as by the stop, so that it
// is told before its write under way ends, however late the stop runs.
// The counts above then reach the database only through the stop's own
// write.
cache.pendingCountsCancel()
stopped := make(chan error, 1)
go func() {
stopped <- cache.StopPendingCountWrites(t.Context())
}()
// Had the stop cut off the write under way, its own write would wait for
// the connection too.
time.Sleep(otherWritesWait)
waits := cache.db.Stats().WaitCount - waitsBefore
t.Logf("%d writes waited for the database", waits)
if waits > 1 {
t.Errorf("%d writes waited for the database, want 1: "+
"the stop cut off the write under way", waits)
}
releaseDatabase()
select {
case err := <-stopped:
if err != nil {
t.Fatalf("StopPendingCountWrites() error = %v", err)
}
case <-time.After(pendingCountsWait):
t.Fatal("StopPendingCountWrites() did not return once the database was free")
}
want := cacheStatsCounters{1, 1, 1, 1024, 1}
if got := readCacheStatsCounters(t, cache); got != want {
t.Errorf("counters = %+v, want %+v", got, want)
}
}
// TestStopPendingCountWrites_ReturnsWhenItsContextEnds holds the database's
// one connection while the goroutine that writes the pending counts waits for
// it, then stops that goroutine with a context that ends first. The stop must
// return the context's error, and the write under way must still reach the
// database once the connection is free.
func TestStopPendingCountWrites_ReturnsWhenItsContextEnds(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
cache.StartPendingCountWrites()
defer func() { _ = cache.StopPendingCountWrites(t.Context()) }()
releaseDatabase := holdDatabase(t, cache)
defer releaseDatabase()
waitsBefore := cache.db.Stats().WaitCount
// Writes given a context past its deadline fail at once.
ended, cancel := context.WithDeadline(t.Context(), time.Now())
defer cancel()
cache.IncrementStats(ended, true, 0)
// The goroutine's write waits for the connection.
waitForConnectionWaits(t, cache.db, waitsBefore+1)
stopCtx, cancelStop := context.WithCancel(t.Context())
cancelStop()
stopped := make(chan error, 1)
go func() {
stopped <- cache.StopPendingCountWrites(stopCtx)
}()
select {
case err := <-stopped:
if !errors.Is(err, context.Canceled) {
t.Errorf("StopPendingCountWrites() error = %v, want %v",
err, context.Canceled)
}
case <-time.After(pendingCountsWait):
t.Fatal("StopPendingCountWrites() did not return once its context ended")
}
releaseDatabase()
waitForCounters(t, cache, cacheStatsCounters{hitCount: 1})
}
+20 -12
View File
@@ -162,7 +162,7 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e
// Fall through to re-process
} else {
// Counted also when the request context has ended meanwhile
s.cache.IncrementStats(context.WithoutCancel(ctx), true, 0)
s.cache.IncrementStats(ctx, true, 0)
return &ImageResponse{
Content: reader,
@@ -179,7 +179,7 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e
// failed or the request context has ended meanwhile
response, err := s.processOrWait(ctx, req)
s.cache.IncrementStats(context.WithoutCancel(ctx), false, 0)
s.cache.IncrementStats(ctx, false, 0)
if err != nil {
return nil, err
@@ -300,14 +300,8 @@ func (s *Service) processOrWait(
}
}()
processingCtx := context.WithoutCancel(ctx)
if deadline, ok := ctx.Deadline(); ok {
var cancel context.CancelFunc
processingCtx, cancel = context.WithDeadline(processingCtx, deadline)
defer cancel()
}
processingCtx, cancel := withoutCancelKeepingDeadline(ctx)
defer cancel()
return s.processFromSourceOrFetch(processingCtx, req, cacheKey)
})
@@ -344,6 +338,20 @@ func (s *Service) processOrWait(
}, nil
}
// withoutCancelKeepingDeadline returns ctx without its cancellation but with
// its deadline, if it has one, and the func that releases the returned
// context.
func withoutCancelKeepingDeadline(
ctx context.Context,
) (context.Context, context.CancelFunc) {
deadline, hasDeadline := ctx.Deadline()
if !hasDeadline {
return context.WithoutCancel(ctx), func() {}
}
return context.WithDeadline(context.WithoutCancel(ctx), deadline)
}
// loadCachedSource opens source content from cache, without reading it, and
// returns it with its size; nil if the cached data is unavailable, empty or
// exceeds maxResponseSize.
@@ -453,7 +461,7 @@ func (s *Service) fetchAndProcess(
fetchBytes := int64(len(sourceData))
// Counted also when the request context has ended meanwhile
s.cache.IncrementUpstreamFetch(context.WithoutCancel(ctx), fetchBytes)
s.cache.IncrementUpstreamFetch(ctx, fetchBytes)
if err != nil {
return nil, fmt.Errorf("failed to read upstream response: %w", err)
@@ -535,7 +543,7 @@ func (s *Service) processAndStore(
processDuration := time.Since(processStart)
// Counted also when the request context has ended meanwhile
s.cache.IncrementTransformCount(context.WithoutCancel(ctx))
s.cache.IncrementTransformCount(ctx)
// Read processed content
processedData, err := io.ReadAll(processResult.Content)
+168
View File
@@ -2,7 +2,9 @@ package imgcache
import (
"context"
"encoding/json"
"io"
"os"
"testing"
"time"
@@ -430,6 +432,172 @@ func TestService_Get_Caching(t *testing.T) {
}
}
// TestService_Get_ServesSourceAndVariantCachedTwoDaysAgo caches a source and
// a variant of it and sets back two days every time pixa stored for them.
// For the source: fetched_at in source_metadata, fetched_at and
// last_accessed_at in source_content, the fetched_at in its metadata file,
// and the access and modification times of its file and of its metadata
// file. For the variant: created_at and last_accessed_at in variant_content,
// the created_at in its .meta file, and the access and modification times of
// its file and of its .meta 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. The source's stored response headers are empty,
// as the mock fetcher sends none. expires_at, etag and last_modified in
// source_metadata and in the metadata file are never written, nor are the
// request_cache and output_content tables.
func TestService_Get_ServesSourceAndVariantCachedTwoDaysAgo(t *testing.T) {
t.Parallel()
svc, fixtures := SetupTestService(t)
variant := photoVariant(fixtures, 85, 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.cache, variant, twoDaysAgo)
backdateVariant(t, svc.cache, 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, FitCover))
if err != nil {
t.Fatalf("Get() of a new variant error = %v", err)
}
_ = resp.Content.Close()
fetches := readCacheStatsCounters(t, svc.cache).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 TestService_Get_ServesSourceAndVariantCachedTwoDaysAgo.
func backdateSource(
t *testing.T, cache *Cache, req *ImageRequest, when time.Time,
) {
t.Helper()
pathHash := HashPath(req.SourcePath + "?" + req.SourceQuery)
meta, err := cache.srcMetadata.Load(req.SourceHost, pathHash)
if err != nil {
t.Fatalf("reading the source's metadata file: %v", err)
}
meta.FetchedAt = when.Unix()
err = cache.srcMetadata.Store(req.SourceHost, pathHash, meta)
if err != nil {
t.Fatalf("writing the source's metadata file: %v", err)
}
dbTime := when.UTC().Format(sqliteTimestampFormat)
updateOneRow(t, cache, `
UPDATE source_metadata SET fetched_at = ?
WHERE source_host = ? AND source_path = ? AND source_query = ?
`, dbTime, req.SourceHost, req.SourcePath, req.SourceQuery)
updateOneRow(t, cache,
`UPDATE source_content SET fetched_at = ? WHERE content_hash = ?`,
dbTime, meta.ContentHash)
setSourceLastAccessed(t, cache, ContentHash(meta.ContentHash), when)
for _, path := range []string{
cache.srcContent.hashToPath(ContentHash(meta.ContentHash)),
cache.srcMetadata.metaPath(req.SourceHost, pathHash),
} {
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
// TestService_Get_ServesSourceAndVariantCachedTwoDaysAgo.
func backdateVariant(
t *testing.T, cache *Cache, req *ImageRequest, when time.Time,
) {
t.Helper()
key := CacheKey(req)
content, size, contentType, err := cache.variants.LoadWithMeta(key)
if err != nil || contentType == "" {
t.Fatalf("reading the variant's .meta file: content type %q, error %v",
contentType, err)
}
_ = content.Close()
meta, err := json.Marshal(VariantMeta{
ContentType: contentType,
Size: size,
CreatedAt: when.Unix(),
})
if err != nil {
t.Fatalf("encoding the variant's .meta file: %v", err)
}
path := cache.variants.keyToPath(key)
err = os.WriteFile(path+variantMetaSuffix, meta, StorageFilePerm)
if err != nil {
t.Fatalf("writing the variant's .meta file: %v", err)
}
updateOneRow(t, cache,
`UPDATE variant_content SET created_at = ? WHERE cache_key = ?`,
when.UTC().Format(sqliteTimestampFormat), string(key))
setVariantLastAccessed(t, cache, key, when)
for _, file := range []string{path, path + variantMetaSuffix} {
err = os.Chtimes(file, when, when)
if err != nil {
t.Fatalf("setting back the times of %s: %v", file, err)
}
}
}
// updateOneRow runs the UPDATE statement query and fails the test unless it
// changed exactly one row.
func updateOneRow(t *testing.T, cache *Cache, query string, args ...any) {
t.Helper()
result, err := cache.db.ExecContext(t.Context(), query, args...)
if err != nil {
t.Fatalf("%s: %v", query, err)
}
rows, err := result.RowsAffected()
if err != nil || rows != 1 {
t.Fatalf("%s: rows = %d, error = %v, want 1 row", query, rows, err)
}
}
func TestService_Get_DifferentSizes(t *testing.T) {
t.Parallel()
-5
View File
@@ -43,7 +43,6 @@ func TestStats_HitRateIsRatio(t *testing.T) {
cache, err := NewCache(db, CacheConfig{
StateDir: dir,
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute,
})
if err != nil {
@@ -91,7 +90,6 @@ func TestStats_ZeroCounts(t *testing.T) {
cache, err := NewCache(db, CacheConfig{
StateDir: dir,
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute,
})
if err != nil {
@@ -120,7 +118,6 @@ func TestStats_LogsFailedCountQueries(t *testing.T) {
cache, err := NewCache(db, CacheConfig{
StateDir: t.TempDir(),
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute,
Logger: slog.New(slog.NewJSONHandler(&logBuf, nil)),
})
@@ -162,7 +159,6 @@ func TestIncrementStats_LogsFailedUpdates(t *testing.T) {
cache, err := NewCache(db, CacheConfig{
StateDir: t.TempDir(),
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute,
Logger: slog.New(slog.NewJSONHandler(&logBuf, nil)),
})
@@ -232,7 +228,6 @@ func TestStats_DisabledCacheReportsNoItems(t *testing.T) {
disabled, err := NewCache(enabled.db, CacheConfig{
StateDir: t.TempDir(),
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute,
DisableDiskCache: true,
})
@@ -185,7 +185,6 @@ func SetupTestService(
cache, err := NewCache(db, CacheConfig{
StateDir: tmpDir,
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute,
})
if err != nil {
@@ -0,0 +1,71 @@
package server
import (
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"sneak.berlin/go/pixa/internal/encurl"
"sneak.berlin/go/pixa/internal/imgcache"
)
// svgServedAsAVIF is an SVG whose first bytes pass the first-bytes check for
// AVIF: its XML comment puts "ftypavis" at bytes 4 to 11, where an AVIF file
// has its ftyp box and brand. libvips would load it as SVG.
const svgServedAsAVIF = `<!--ftypavis--><svg xmlns="http://www.w3.org/2000/svg" ` +
`width="64" height="48"><rect width="64" height="48"/></svg>`
// TestImageProxyRefusesSVGServedAsAVIF requests svgServedAsAVIF, which the
// upstream answers as image/avif, through both image routes, resized as PNG
// and in its own format (orig). Each request is answered 415, as for an SVG
// source, never with an image.
func TestImageProxyRefusesSVGServedAsAVIF(t *testing.T) {
t.Parallel()
upstream := httptest.NewServer(http.HandlerFunc(
func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "image/avif")
_, _ = io.WriteString(w, svgServedAsAVIF)
}))
t.Cleanup(upstream.Close)
s, _, _ := startImageProxy(t, upstream)
generator, err := encurl.NewGenerator(testSigningKey)
if err != nil {
t.Fatalf("encurl.NewGenerator: %v", err)
}
encryptedURL := func(format imgcache.ImageFormat) string {
token, err := generator.Generate(&encurl.Payload{
SourceHost: upstreamHost, SourcePath: "/drawing.avif",
Width: 32, Height: 24, Format: format,
})
if err != nil {
t.Fatalf("encurl Generate: %v", err)
}
return "/v1/e/" + token + "/drawing." + string(format)
}
for _, target := range []string{
"/v1/image/" + upstreamHost + "/drawing.avif/32x24.png",
"/v1/image/" + upstreamHost + "/drawing.avif/64x48.orig",
encryptedURL(imgcache.FormatPNG),
encryptedURL(imgcache.FormatOriginal),
} {
rec := httptest.NewRecorder()
s.ServeHTTP(rec, httptest.NewRequestWithContext(
t.Context(), http.MethodGet, target, nil))
t.Logf("GET %s: %d, Content-Type %s, body %q",
target, rec.Code, rec.Header().Get("Content-Type"), rec.Body.String())
if rec.Code != http.StatusUnsupportedMediaType ||
!strings.Contains(rec.Body.String(), "unsupported source format") {
t.Errorf("GET %s: %d %q, want %d unsupported source format",
target, rec.Code, rec.Body.String(), http.StatusUnsupportedMediaType)
}
}
}
+41
View File
@@ -0,0 +1,41 @@
package server
import (
"io"
"net/http"
"net/http/httptest"
"testing"
)
// TestImageProxyRefusesSVG requests an image the upstream answers as SVG,
// in its own format (orig) and resized as PNG, through pixa's router,
// upstream fetcher and image processor. pixa does not accept SVG as a
// source, so each request is answered 415, never with an image.
func TestImageProxyRefusesSVG(t *testing.T) {
t.Parallel()
upstream := httptest.NewServer(http.HandlerFunc(
func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "image/svg+xml")
_, _ = io.WriteString(w, `<svg xmlns="http://www.w3.org/2000/svg" `+
`width="64" height="48"><rect width="64" height="48"/></svg>`)
}))
t.Cleanup(upstream.Close)
s, _, _ := startImageProxy(t, upstream)
for _, sizeFormat := range []string{"64x48.orig", "32x24.png"} {
target := "/v1/image/" + upstreamHost + "/drawing.svg/" + sizeFormat
rec := httptest.NewRecorder()
s.ServeHTTP(rec, httptest.NewRequestWithContext(
t.Context(), http.MethodGet, target, nil))
t.Logf("GET %s: %d, Content-Type %s, body %s",
target, rec.Code, rec.Header().Get("Content-Type"), rec.Body.String())
if rec.Code != http.StatusUnsupportedMediaType {
t.Errorf("GET %s status = %d, want %d",
target, rec.Code, http.StatusUnsupportedMediaType)
}
}
}