Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7a348655ee | ||
|
|
3a7af08838 | ||
|
|
3392cb0286 | ||
|
|
0233001110 | ||
|
|
9a898db10a |
+17
-16
@@ -3,14 +3,13 @@
|
||||
# golangci/golangci-lint:v2.12.2-alpine, 2026-08-07
|
||||
FROM golangci/golangci-lint:v2.12.2-alpine@sha256:91b27804074a0bacea298707f016911e60cf0cdbc6c7bf5ccacb5f0606d18d60 AS lint
|
||||
|
||||
RUN apk add --no-cache make build-base vips-dev libheif-dev pkgconfig
|
||||
|
||||
WORKDIR /src
|
||||
|
||||
# script/bootstrap installs the build dependencies and downloads the Go
|
||||
# modules. Only script/, go.mod and go.sum are copied first, so this
|
||||
# layer is reused until one of them changes.
|
||||
COPY script/ ./script/
|
||||
# Copy go mod files first for better layer caching
|
||||
COPY go.mod go.sum ./
|
||||
RUN script/bootstrap
|
||||
RUN go mod download
|
||||
|
||||
# Copy source code
|
||||
COPY . .
|
||||
@@ -29,12 +28,20 @@ FROM golang:1.25.4-alpine@sha256:d3f0cf7723f3429e3f9ed846243970b20a2de7bae6a5b66
|
||||
# Depend on lint stage passing
|
||||
COPY --from=lint /src/go.sum /dev/null
|
||||
|
||||
ARG VERSION=dev
|
||||
|
||||
# Install build dependencies for CGO image libraries
|
||||
RUN apk add --no-cache \
|
||||
build-base \
|
||||
vips-dev \
|
||||
libheif-dev \
|
||||
pkgconfig
|
||||
|
||||
WORKDIR /src
|
||||
|
||||
# Build dependencies and Go modules, as in the lint stage
|
||||
COPY script/ ./script/
|
||||
# Copy go mod files first for better layer caching
|
||||
COPY go.mod go.sum ./
|
||||
RUN script/bootstrap
|
||||
RUN GOTOOLCHAIN=auto go mod download
|
||||
|
||||
# Copy source code
|
||||
COPY . .
|
||||
@@ -42,14 +49,8 @@ COPY . .
|
||||
# Run tests
|
||||
RUN make test
|
||||
|
||||
# VERSION is declared here, not earlier: a new value reruns only the
|
||||
# build, not script/bootstrap or the tests. CGO stays enabled for
|
||||
# govips; -trimpath keeps build paths out of the binary, and -s -w
|
||||
# leave out the symbol table and debug information.
|
||||
ARG VERSION=dev
|
||||
RUN CGO_ENABLED=1 GOTOOLCHAIN=auto go build -trimpath \
|
||||
-ldflags "-s -w -X main.Version=${VERSION}" \
|
||||
-o /pixad ./cmd/pixad
|
||||
# Build with CGO enabled
|
||||
RUN CGO_ENABLED=1 GOTOOLCHAIN=auto go build -ldflags "-X main.Version=${VERSION}" -o /pixad ./cmd/pixad
|
||||
|
||||
# Runtime stage
|
||||
# alpine:3.21, 2026-02-25
|
||||
|
||||
+7
-7
@@ -6,16 +6,16 @@
|
||||
# golangci/golangci-lint:v2.12.2-alpine, 2026-08-07
|
||||
FROM golangci/golangci-lint:v2.12.2-alpine@sha256:91b27804074a0bacea298707f016911e60cf0cdbc6c7bf5ccacb5f0606d18d60
|
||||
|
||||
# pixa is CGO/libvips: the type-aware linters compile every package, so
|
||||
# this image needs the same C libraries the build does.
|
||||
RUN apk add --no-cache build-base vips-dev libheif-dev pkgconfig
|
||||
|
||||
WORKDIR /src
|
||||
|
||||
# pixa is CGO/libvips: the type-aware linters compile every package, so
|
||||
# this image needs the same C libraries the build does. script/bootstrap
|
||||
# installs them and downloads the Go modules. Only script/, go.mod and
|
||||
# go.sum are copied first; they settle this layer's result, so it may
|
||||
# safely be reused between runs.
|
||||
COPY script/ ./script/
|
||||
# Modules first for layer caching; go.mod/go.sum settle this layer's
|
||||
# result, so it may safely be reused between runs.
|
||||
COPY go.mod go.sum ./
|
||||
RUN script/bootstrap
|
||||
RUN go mod download
|
||||
|
||||
COPY . .
|
||||
|
||||
|
||||
@@ -40,8 +40,9 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for pixa needs:
|
||||
|
||||
- **Port:** pixa listens on container port `8080`.
|
||||
- **Volume:** container path `/var/lib/pixa`, where pixa keeps its
|
||||
database and cache. Creating the host directory when it is missing is
|
||||
upaas's job, tracked in https://git.eeqj.de/sneak/upaas/issues/235.
|
||||
database and cache. upaas bind-mounts the host path it is given and
|
||||
does not create it, so the host directory must exist before the first
|
||||
deploy.
|
||||
- **Environment variables:**
|
||||
- `PIXA_SIGNING_KEY` (required): secret for signed and encrypted URLs
|
||||
and login, 32+ characters, for example from
|
||||
@@ -57,6 +58,10 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for pixa needs:
|
||||
`healthy`. The probe uses the port from `PORT` (default `8080`), so a
|
||||
port changed only in a mounted config file is not seen by it: change
|
||||
the port with `PORT`.
|
||||
- **First run:** create the host directory, owned by root or by uid
|
||||
`65532` and gid `65532`. The server runs as the container's `pixad`
|
||||
user, which has that uid and gid, and the container gives the
|
||||
directory to `pixad` when it starts.
|
||||
|
||||
## Rationale
|
||||
|
||||
@@ -81,10 +86,7 @@ prevent abuse, and allowlisted source hosts for open access.
|
||||
|
||||
Multiple source paths may reference the same content blob; the
|
||||
database tracks references rather than using filesystem refcounting.
|
||||
Toward a target of 1-5k r/s, pixa keeps in memory the content types of
|
||||
the 10,000 transformed images most recently cached or served, so a
|
||||
cache hit on one of them reads only the image file from disk and not
|
||||
the metadata file stored beside it.
|
||||
In-process caching of request-to-output mappings targets 1-5k r/s.
|
||||
|
||||
### Routes
|
||||
|
||||
@@ -107,13 +109,6 @@ 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.
|
||||
|
||||
When several requests for the same image, size, format, quality and fit miss
|
||||
the cache at once, they share one upstream fetch (or one read of the cached
|
||||
source) and one transcode: the first request does the work, and the others wait
|
||||
for its image or its error, holding no upstream connection or processing slot
|
||||
of their own. A waiting request stops waiting when its own client goes away;
|
||||
the work goes on for the others.
|
||||
|
||||
The login form (`POST /`) is limited to 5 attempts per minute per client
|
||||
address, counting an IPv6 client by its /64; an attempt over the limit is
|
||||
refused with 429 and a `Retry-After` header. Behind a reverse proxy the client
|
||||
@@ -250,7 +245,7 @@ variables set by the file's `env:` section are checked the same way.
|
||||
| `PIXA_METRICS_PASSWORD` | `metrics.password` | Password for `/metrics`; set together with the username |
|
||||
| `PIXA_SENTRY_DSN` | `sentry_dsn` | Sentry DSN for error reporting; empty disables it |
|
||||
| `PIXA_DEBUG` | `debug` | Debug logging and plain-HTTP local development; default `false` |
|
||||
| `PIXA_MAINTENANCE_MODE` | `maintenance_mode` | Answer image requests with 503; the health check stays 200; default `false` |
|
||||
| `PIXA_MAINTENANCE_MODE` | `maintenance_mode` | Maintenance flag reported by the health check; default `false` |
|
||||
|
||||
Key settings in more detail:
|
||||
|
||||
@@ -293,9 +288,8 @@ Key settings in more detail:
|
||||
bytes; default `52428800` (50 MiB). It also limits the image data pixa
|
||||
decodes
|
||||
- `downstream_timeout` — time allowed for answering one client request, as a
|
||||
duration; default `60s`. The upstream fetch counts toward it, and so do the
|
||||
waits for an upstream connection and for a processing slot (up to 10 seconds
|
||||
each), so keep it longer than `upstream_fetch_timeout` plus 20 seconds
|
||||
duration; default `60s`. The upstream fetch counts toward it, so keep it
|
||||
longer than `upstream_fetch_timeout`
|
||||
- `signing_key` — HMAC secret for URL signatures
|
||||
- `cache_max_bytes` — disk cache size limit in bytes; `0` disables the
|
||||
disk cache entirely; omitted defaults to 75% of the free space on
|
||||
@@ -304,20 +298,12 @@ Key settings in more detail:
|
||||
hosts together, on top of `upstream_connections_per_host`; default `64`. A
|
||||
fetch holds its connection until its image has been processed. A fetch that
|
||||
finds all of them in use waits up to 10 seconds for one to free up; if none
|
||||
does, and `downstream_timeout` has not ended first, the request is answered
|
||||
503 with the error `server busy, try again later`
|
||||
does, the request is answered 503 with the error
|
||||
`server busy, try again later`
|
||||
- `max_concurrent_processing` — the most images decoded and encoded at once;
|
||||
default the number of CPUs pixa can use (`GOMAXPROCS`), which follows a
|
||||
container's CPU limit. A request that finds all of them in use waits up to 10
|
||||
seconds for one to free up; if none does, and `downstream_timeout` has not
|
||||
ended first, it is answered 503 the same way
|
||||
- `maintenance_mode` — while `true`, the image routes (`/v1/image/` and
|
||||
`/v1/e/`) answer every request with 503, a `Retry-After` header and a JSON
|
||||
error body. The health check (`/.well-known/healthcheck.json`) still answers
|
||||
200 and reports `"maintenance_mode": true`. It stays 200 because the image's
|
||||
Docker `HEALTHCHECK` requests it: a 503 there would make the container
|
||||
unhealthy, and upaas marks a deploy failed when its container is unhealthy.
|
||||
The login and URL generator pages and `/metrics` keep working
|
||||
seconds for one to free up; if none does, it is answered 503 the same way
|
||||
|
||||
See `config.example.yml` for all options with defaults.
|
||||
|
||||
|
||||
@@ -29,43 +29,6 @@ P2: security: referer blacklist
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- 2026-09-29 share concurrent misses (closes #65): requests that miss the same
|
||||
variant at once (the same cache key, so quality and fit included) share one
|
||||
upstream fetch or cached source read and one transcode through
|
||||
`golang.org/x/sync/singleflight`; the first request's processing runs with a
|
||||
context that does not end with its own, and the others wait for its image or
|
||||
error holding no upstream connection or processing slot, and stop waiting
|
||||
when their own context ends; the request doing the processing waits for it
|
||||
even then, as before; a request whose context has already ended starts
|
||||
nothing; each request counts one miss, and the processing counts its fetch
|
||||
and transcode once; a panic while processing becomes an error for every
|
||||
waiting request instead of stopping pixad; documented in `README.md`.
|
||||
- 2026-09-29 the container makes `/var/lib/pixa` usable by itself (closes
|
||||
#159): `deploy/docker-entrypoint.sh` creates the directory if it is missing,
|
||||
gives the directory and everything in it to `pixad` when the directory or one
|
||||
of its top-level entries belongs to another user or group, sets its mode to
|
||||
`750`, then runs the server as `pixad`; data left by an earlier run under
|
||||
another uid is taken over this way; "Running under upaas" in `README.md` no
|
||||
longer tells the operator to create or chown the host directory.
|
||||
- 2026-09-29 variant content types kept in memory (closes #70):
|
||||
`Cache.metaCache` holds the content types of up to 10,000 variants in an LRU
|
||||
(`github.com/hashicorp/golang-lru/v2`), filled by `StoreVariant` and by
|
||||
`GetVariant` after it reads a `.meta` file, where a type `StoreVariant` added
|
||||
meanwhile is kept over the one read, and never with the
|
||||
`application/octet-stream` served for a variant without one; for a variant it
|
||||
holds, `GetVariant` skips the `.meta` read, still opening the variant file and
|
||||
taking the size from it; eviction removes the entry before deleting the files,
|
||||
and `GetVariant` removes it when the file will not open; the cap is a
|
||||
constant, not a setting; the unused `variantMeta` type is gone; `README.md`
|
||||
describes it.
|
||||
- 2026-09-29 maintenance mode refuses image requests (closes #71): while
|
||||
`maintenance_mode` is on, `/v1/image/` and `/v1/e/` answer 503 with a
|
||||
`Retry-After` header and the JSON error body, from one middleware in
|
||||
`internal/server/routes.go`; the health check stays 200 and reports
|
||||
`maintenance_mode`, as the image's Docker `HEALTHCHECK` requests it and upaas
|
||||
marks a deploy failed when its container is unhealthy; the login and URL
|
||||
generator pages and `/metrics` keep working; documented in `README.md` and
|
||||
`config.example.yml`.
|
||||
- 2026-09-29 bound concurrent image processing and upstream fetches (closes
|
||||
#64): `max_concurrent_processing` (default the number of CPUs pixa can use)
|
||||
limits the images decoded and encoded at once, and `upstream_connections`
|
||||
@@ -76,14 +39,6 @@ P2: security: referer blacklist
|
||||
10 seconds for a free one, then gets 503 `server busy, try again later`;
|
||||
libvips runs one worker thread per image with its operation cache off;
|
||||
documented in `README.md` and `config.example.yml`.
|
||||
- 2026-09-29 Dockerfiles install through `script/bootstrap` (closes #95): the
|
||||
`Dockerfile` lint and build stages and `Dockerfile.lint` copy `script/`,
|
||||
`go.mod` and `go.sum`, then run `script/bootstrap` in place of their own
|
||||
`apk add` lines, so the build dependencies are listed in one place;
|
||||
`script/bootstrap` now also installs a C compiler when `gcc` is missing; the
|
||||
build uses `-trimpath` and `-s -w` and keeps `CGO_ENABLED=1` for govips;
|
||||
`ARG VERSION` sits just above the build, so a new version reruns neither
|
||||
`script/bootstrap` nor the tests.
|
||||
- 2026-09-29 migrations at the path `REPO_POLICIES.md` sets (closes #96): the
|
||||
migration files moved, contents unchanged, from `internal/database/schema/`
|
||||
to `internal/db/migrations/` as `000_migration.sql` and `001_schema.sql`; the
|
||||
|
||||
+4
-13
@@ -16,12 +16,6 @@
|
||||
# Server settings
|
||||
port: 8080
|
||||
debug: false
|
||||
|
||||
# While true, the image routes (/v1/image/ and /v1/e/) answer every request
|
||||
# with 503 and a Retry-After header. The health check keeps answering 200 and
|
||||
# reports maintenance_mode as true. It stays 200 because the image's Docker
|
||||
# HEALTHCHECK requests it: a 503 there would make the container unhealthy, and
|
||||
# upaas marks a deploy failed when its container is unhealthy.
|
||||
maintenance_mode: false
|
||||
|
||||
# Data directory for SQLite database and cache files
|
||||
@@ -80,14 +74,13 @@ upstream_connections_per_host: 20
|
||||
# Maximum concurrent connections to all upstream hosts together, on top of
|
||||
# the per-host limit (default: 64). A fetch holds its connection until its
|
||||
# image has been processed. A fetch that finds none free waits up to 10
|
||||
# seconds for one, and if none frees up the request is answered 503, unless
|
||||
# downstream_timeout has ended first.
|
||||
# seconds for one, and if none frees up the request is answered 503.
|
||||
upstream_connections: 64
|
||||
|
||||
# Maximum number of images decoded and encoded at once (default: the
|
||||
# number of CPUs pixa can use, which follows a container's CPU limit). A
|
||||
# request that finds none free waits up to 10 seconds for one, and if none
|
||||
# frees up it is answered 503, unless downstream_timeout has ended first.
|
||||
# frees up it is answered 503.
|
||||
# max_concurrent_processing: 4
|
||||
|
||||
# Time allowed for one fetch from an upstream host (default: 30s)
|
||||
@@ -97,10 +90,8 @@ upstream_fetch_timeout: 30s
|
||||
# (1 GiB) (default: 52428800, 50 MiB)
|
||||
upstream_max_response_size: 52428800
|
||||
|
||||
# Time allowed for answering one client request (default: 60s). The
|
||||
# upstream fetch counts toward it, and so do the waits for an upstream
|
||||
# connection and for a processing slot (up to 10 seconds each), so keep it
|
||||
# longer than upstream_fetch_timeout plus 20 seconds.
|
||||
# Time allowed for answering one client request, the upstream fetch
|
||||
# included, so keep it longer than upstream_fetch_timeout (default: 60s)
|
||||
downstream_timeout: 60s
|
||||
|
||||
# The origin a browser lets read pixa's responses, sent as the CORS
|
||||
|
||||
@@ -1,22 +1,14 @@
|
||||
#!/bin/sh
|
||||
# deploy/docker-entrypoint.sh: the Docker image's ENTRYPOINT. It runs as
|
||||
# root only to make /var/lib/pixa usable by pixad: a host directory
|
||||
# bind-mounted there keeps its host owner, often root, and data from an
|
||||
# earlier run may belong to another uid. The server itself always runs
|
||||
# as pixad.
|
||||
# root only to give /var/lib/pixa to pixad: a host directory
|
||||
# bind-mounted there keeps its host owner, often root, and pixad could
|
||||
# not write to it. The server itself always runs as pixad.
|
||||
set -eu
|
||||
|
||||
main() {
|
||||
mkdir -p /var/lib/pixa
|
||||
# Only the directory and its top-level entries are checked, so a
|
||||
# normal start does not walk the cache. -depth gives each directory
|
||||
# to pixad after its contents, so a start stopped part way leaves
|
||||
# something at the top for the next start to find; -h changes a
|
||||
# symlink itself, never the file it points to.
|
||||
if [ -n "$(find /var/lib/pixa -maxdepth 1 \( ! -user pixad -o ! -group pixad \))" ]; then
|
||||
find /var/lib/pixa -depth -exec chown -h pixad:pixad {} +
|
||||
if [ "$(stat -c %U /var/lib/pixa)" != pixad ]; then
|
||||
chown pixad:pixad /var/lib/pixa
|
||||
fi
|
||||
chmod 750 /var/lib/pixa
|
||||
exec su-exec pixad /usr/local/bin/pixad "$@"
|
||||
}
|
||||
|
||||
|
||||
@@ -14,13 +14,11 @@ require (
|
||||
github.com/go-chi/httprate v0.16.0
|
||||
github.com/gorilla/csrf v1.7.3
|
||||
github.com/gorilla/securecookie v1.1.2
|
||||
github.com/hashicorp/golang-lru/v2 v2.0.7
|
||||
github.com/prometheus/client_golang v1.23.2
|
||||
github.com/slok/go-http-metrics v0.13.0
|
||||
github.com/spf13/cobra v1.10.2
|
||||
go.uber.org/fx v1.24.0
|
||||
golang.org/x/crypto v0.41.0
|
||||
golang.org/x/sync v0.19.0
|
||||
modernc.org/sqlite v1.42.2
|
||||
)
|
||||
|
||||
@@ -135,6 +133,7 @@ require (
|
||||
golang.org/x/image v0.34.0 // indirect
|
||||
golang.org/x/net v0.43.0 // indirect
|
||||
golang.org/x/oauth2 v0.30.0 // indirect
|
||||
golang.org/x/sync v0.19.0 // indirect
|
||||
golang.org/x/sys v0.36.0 // indirect
|
||||
golang.org/x/term v0.34.0 // indirect
|
||||
golang.org/x/text v0.32.0 // indirect
|
||||
|
||||
@@ -228,8 +228,6 @@ github.com/hashicorp/go-version v1.2.1/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09
|
||||
github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
|
||||
github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
|
||||
github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
|
||||
github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
|
||||
github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
|
||||
github.com/hashicorp/hcl v1.0.1-vault-7 h1:ag5OxFVy3QYTFTJODRzTKVZ6xvdfLLCA1cy/Y6xGI0I=
|
||||
github.com/hashicorp/hcl v1.0.1-vault-7/go.mod h1:XYhtn6ijBSAj6n4YqAaf7RBPS4I06AItNorpy+MoQNM=
|
||||
github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64=
|
||||
|
||||
+25
-83
@@ -14,7 +14,6 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
lru "github.com/hashicorp/golang-lru/v2"
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
)
|
||||
|
||||
@@ -27,10 +26,6 @@ var (
|
||||
// HTTP status code for successful fetch.
|
||||
const httpStatusOK = 200
|
||||
|
||||
// metaCacheSize is how many variants' content types metaCache holds. A
|
||||
// variant not among them is served as before, reading its .meta file.
|
||||
const metaCacheSize = 10000
|
||||
|
||||
// CacheConfig holds cache configuration.
|
||||
type CacheConfig struct {
|
||||
StateDir string
|
||||
@@ -54,6 +49,12 @@ type CacheConfig struct {
|
||||
Logger *slog.Logger
|
||||
}
|
||||
|
||||
// variantMeta stores content type for fast cache hits without reading .meta file.
|
||||
type variantMeta struct {
|
||||
ContentType string
|
||||
Size int64
|
||||
}
|
||||
|
||||
// Cache implements the caching layer for the image proxy.
|
||||
type Cache struct {
|
||||
db *sql.DB
|
||||
@@ -75,10 +76,9 @@ type Cache struct {
|
||||
evictionStarted bool
|
||||
evictionStopOnce sync.Once
|
||||
|
||||
// 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.
|
||||
metaCache *lru.Cache[VariantKey, string]
|
||||
// In-memory cache of variant metadata (content type, size) to avoid
|
||||
// reading .meta files
|
||||
metaCache map[VariantKey]variantMeta
|
||||
|
||||
// contentLocks serializes StoreSource and evictSourceBlob per
|
||||
// content hash, closing the race window between an eviction's row
|
||||
@@ -101,11 +101,6 @@ func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) {
|
||||
log = slog.Default()
|
||||
}
|
||||
|
||||
metaCache, err := lru.New[VariantKey, string](metaCacheSize)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create variant content type cache: %w", err)
|
||||
}
|
||||
|
||||
c := &Cache{
|
||||
db: db,
|
||||
config: config,
|
||||
@@ -114,7 +109,7 @@ func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) {
|
||||
evictionPressure: make(chan struct{}, 1),
|
||||
evictionStop: make(chan struct{}),
|
||||
evictionDone: make(chan struct{}),
|
||||
metaCache: metaCache,
|
||||
metaCache: make(map[VariantKey]variantMeta),
|
||||
contentLocks: newContentLock(),
|
||||
}
|
||||
|
||||
@@ -182,30 +177,13 @@ func (c *Cache) Lookup(ctx context.Context, req *ImageRequest) (*LookupResult, e
|
||||
}, nil
|
||||
}
|
||||
|
||||
// GetVariant returns a reader, size, and content type for a cached
|
||||
// variant. The content type comes from metaCache, or else from the
|
||||
// variant's .meta file and is then kept in metaCache. A variant with
|
||||
// no .meta file is served as application/octet-stream, which is not
|
||||
// kept.
|
||||
// GetVariant returns a reader, size, and content type for a cached variant.
|
||||
func (c *Cache) GetVariant(cacheKey VariantKey) (io.ReadCloser, int64, string, error) {
|
||||
if c.disabled {
|
||||
return nil, 0, "", ErrNotFound
|
||||
}
|
||||
|
||||
contentType, known := c.metaCache.Get(cacheKey)
|
||||
if !known {
|
||||
return c.loadVariantWithMeta(cacheKey)
|
||||
}
|
||||
|
||||
reader, size, err := c.variants.LoadWithSize(cacheKey)
|
||||
if err != nil {
|
||||
// The file is gone, e.g. deleted outside pixa
|
||||
c.metaCache.Remove(cacheKey)
|
||||
|
||||
return nil, 0, "", err
|
||||
}
|
||||
|
||||
return reader, size, contentType, nil
|
||||
return c.variants.LoadWithMeta(cacheKey)
|
||||
}
|
||||
|
||||
// StoreSource stores fetched source content and metadata. On a
|
||||
@@ -308,8 +286,6 @@ func (c *Cache) StoreVariant(
|
||||
return err
|
||||
}
|
||||
|
||||
c.metaCache.Add(cacheKey, contentType)
|
||||
|
||||
_, err = c.db.ExecContext(ctx, `
|
||||
INSERT INTO variant_content (cache_key, size_bytes, content_type)
|
||||
VALUES (?, ?, ?)
|
||||
@@ -471,8 +447,7 @@ func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) {
|
||||
return &stats, nil
|
||||
}
|
||||
|
||||
// IncrementStats counts a cache hit or miss, and an upstream fetch that read
|
||||
// fetchBytes bytes, as IncrementUpstreamFetch does.
|
||||
// IncrementStats increments cache statistics.
|
||||
func (c *Cache) IncrementStats(ctx context.Context, hit bool, fetchBytes int64) {
|
||||
var err error
|
||||
|
||||
@@ -496,26 +471,18 @@ func (c *Cache) IncrementStats(ctx context.Context, hit bool, fetchBytes int64)
|
||||
c.log.Warn("failed to count cache hit or miss", "hit", hit, "error", err)
|
||||
}
|
||||
|
||||
c.IncrementUpstreamFetch(ctx, fetchBytes)
|
||||
}
|
||||
|
||||
// IncrementUpstreamFetch counts one upstream fetch that read fetchBytes bytes.
|
||||
// A fetch that read no bytes is not counted.
|
||||
func (c *Cache) IncrementUpstreamFetch(ctx context.Context, fetchBytes int64) {
|
||||
if fetchBytes <= 0 {
|
||||
return
|
||||
}
|
||||
|
||||
_, err := c.db.ExecContext(ctx, `
|
||||
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 {
|
||||
c.log.Warn("failed to count upstream fetch",
|
||||
"fetch_bytes", fetchBytes, "error", err)
|
||||
if fetchBytes > 0 {
|
||||
_, err = c.db.ExecContext(ctx, `
|
||||
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 {
|
||||
c.log.Warn("failed to count upstream fetch",
|
||||
"fetch_bytes", fetchBytes, "error", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -532,31 +499,6 @@ func (c *Cache) IncrementTransformCount(ctx context.Context) {
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
// meanwhile, as the store's is newer. A read that finds no .meta file,
|
||||
// as one can between a store's writing of the variant file and of its
|
||||
// .meta file, serves application/octet-stream and keeps nothing, so
|
||||
// metaCache only ever holds a type read from a .meta file or passed to
|
||||
// StoreVariant.
|
||||
func (c *Cache) loadVariantWithMeta(
|
||||
cacheKey VariantKey,
|
||||
) (io.ReadCloser, int64, string, error) {
|
||||
reader, size, contentType, err := c.variants.LoadWithMeta(cacheKey)
|
||||
if err != nil {
|
||||
return nil, 0, "", err
|
||||
}
|
||||
|
||||
if contentType == "" {
|
||||
return reader, size, fallbackContentType, nil
|
||||
}
|
||||
|
||||
c.metaCache.ContainsOrAdd(cacheKey, contentType)
|
||||
|
||||
return reader, size, contentType, nil
|
||||
}
|
||||
|
||||
// writeMetadataSidecar writes the JSON metadata sidecar of a stored source.
|
||||
// A failure is logged and is otherwise non-fatal; the metadata is in the
|
||||
// database.
|
||||
|
||||
@@ -1,444 +0,0 @@
|
||||
package imgcache
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"image/jpeg"
|
||||
"io"
|
||||
"io/fs"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
"sneak.berlin/go/pixa/internal/magic"
|
||||
)
|
||||
|
||||
// arrivalWait is how long a test gives requests it has started to reach the
|
||||
// point where they wait for a held fetch.
|
||||
const arrivalWait = 100 * time.Millisecond
|
||||
|
||||
// heldFetcher counts the fetches it is asked for and holds each one until
|
||||
// releaseFetches is called, so that a test can have requests arrive while a
|
||||
// fetch is in progress. started receives once for every fetch.
|
||||
type heldFetcher struct {
|
||||
upstream httpfetcher.Fetcher
|
||||
fetches atomic.Int32
|
||||
started chan struct{}
|
||||
release chan struct{}
|
||||
releaseOnce sync.Once
|
||||
}
|
||||
|
||||
func (f *heldFetcher) Fetch(
|
||||
ctx context.Context, url string,
|
||||
) (*httpfetcher.FetchResult, error) {
|
||||
f.fetches.Add(1)
|
||||
|
||||
f.started <- struct{}{}
|
||||
|
||||
select {
|
||||
case <-f.release:
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
|
||||
return f.upstream.Fetch(ctx, url)
|
||||
}
|
||||
|
||||
// releaseFetches lets every held fetch, and every later one, go on.
|
||||
func (f *heldFetcher) releaseFetches() {
|
||||
f.releaseOnce.Do(func() { close(f.release) })
|
||||
}
|
||||
|
||||
// setupHeldFetchService returns a test service whose fetches go through a
|
||||
// heldFetcher. Its database is limited to one connection: each connection to
|
||||
// an in-memory SQLite database opens a new, empty one, so requests running at
|
||||
// once must share the connection that holds the schema.
|
||||
func setupHeldFetchService(t *testing.T) (*Service, *TestFixtures, *heldFetcher) {
|
||||
t.Helper()
|
||||
|
||||
svc, fixtures := SetupTestService(t)
|
||||
svc.cache.db.SetMaxOpenConns(1)
|
||||
|
||||
fetcher := &heldFetcher{
|
||||
upstream: svc.fetcher,
|
||||
started: make(chan struct{}, 100),
|
||||
release: make(chan struct{}),
|
||||
}
|
||||
svc.fetcher = fetcher
|
||||
|
||||
t.Cleanup(fetcher.releaseFetches)
|
||||
|
||||
return svc, fixtures, fetcher
|
||||
}
|
||||
|
||||
// photoVariant asks for the test photo, 100x100, at 50x25 as a JPEG of the
|
||||
// given quality and fit mode. Each call returns a new request, as Get writes
|
||||
// to the request it is given.
|
||||
func photoVariant(fixtures *TestFixtures, quality int, fit FitMode) *ImageRequest {
|
||||
return &ImageRequest{
|
||||
SourceHost: fixtures.GoodHost,
|
||||
SourcePath: testPathPhoto,
|
||||
Size: Size{Width: 50, Height: 25},
|
||||
Format: FormatJPEG,
|
||||
Quality: quality,
|
||||
FitMode: fit,
|
||||
}
|
||||
}
|
||||
|
||||
// getResult is what one Get call returned, with the image read out.
|
||||
type getResult struct {
|
||||
image []byte
|
||||
err error
|
||||
}
|
||||
|
||||
// startGet calls Get in a goroutine of its own and delivers what it returned
|
||||
// on the channel.
|
||||
func startGet(
|
||||
ctx context.Context, svc *Service, req *ImageRequest,
|
||||
) <-chan getResult {
|
||||
results := make(chan getResult, 1)
|
||||
|
||||
go func() {
|
||||
resp, err := svc.Get(ctx, req)
|
||||
if err != nil {
|
||||
results <- getResult{err: err}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
defer func() { _ = resp.Content.Close() }()
|
||||
|
||||
image, err := io.ReadAll(resp.Content)
|
||||
results <- getResult{image: image, err: err}
|
||||
}()
|
||||
|
||||
return results
|
||||
}
|
||||
|
||||
// jpegSize returns the width and height of the JPEG image in data, or 0 and 0
|
||||
// if data is not one.
|
||||
func jpegSize(data []byte) (int, int) {
|
||||
config, err := jpeg.DecodeConfig(bytes.NewReader(data))
|
||||
if err != nil {
|
||||
return 0, 0
|
||||
}
|
||||
|
||||
return config.Width, config.Height
|
||||
}
|
||||
|
||||
// TestService_Get_ConcurrentMissesShareOneFetch starts several requests for
|
||||
// one uncached variant while the first one's fetch is held. Between them they
|
||||
// must fetch the source once and transcode it once, every one must be answered
|
||||
// with the same 50x25 JPEG, and each must count one miss.
|
||||
func TestService_Get_ConcurrentMissesShareOneFetch(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
svc, fixtures, fetcher := setupHeldFetchService(t)
|
||||
|
||||
const requests = 8
|
||||
|
||||
pending := make([]<-chan getResult, 0, requests)
|
||||
for range requests {
|
||||
pending = append(pending,
|
||||
startGet(t.Context(), svc, photoVariant(fixtures, 85, FitCover)))
|
||||
}
|
||||
|
||||
<-fetcher.started
|
||||
time.Sleep(arrivalWait)
|
||||
fetcher.releaseFetches()
|
||||
|
||||
var first []byte
|
||||
|
||||
for i, results := range pending {
|
||||
got := <-results
|
||||
if got.err != nil {
|
||||
t.Fatalf("request %d: Get() error = %v", i, got.err)
|
||||
}
|
||||
|
||||
if width, height := jpegSize(got.image); width != 50 || height != 25 {
|
||||
t.Errorf("request %d: image is %dx%d, want a 50x25 JPEG", i, width, height)
|
||||
}
|
||||
|
||||
if first == nil {
|
||||
first = got.image
|
||||
} else if !bytes.Equal(got.image, first) {
|
||||
t.Errorf("request %d: image differs from request 0's", i)
|
||||
}
|
||||
}
|
||||
|
||||
if fetches := fetcher.fetches.Load(); fetches != 1 {
|
||||
t.Errorf("%d requests made %d upstream fetches, want 1", requests, fetches)
|
||||
}
|
||||
|
||||
// NewTestFS builds the same files the test service's fetcher serves.
|
||||
testFS, _ := NewTestFS(t)
|
||||
|
||||
photo, err := fs.ReadFile(testFS, fixtures.GoodHostJPEG)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
want := cacheStatsCounters{0, requests, 1, int64(len(photo)), 1}
|
||||
|
||||
if got := readCacheStatsCounters(t, svc.cache); got != want {
|
||||
t.Errorf("counters = %+v, want %+v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// TestService_Get_ConcurrentVariantsStayApart requests three variants of the
|
||||
// test photo at once that differ only in quality or fit. Each must be made by
|
||||
// a fetch and a transcode of its own, and each answer must be its own variant.
|
||||
func TestService_Get_ConcurrentVariantsStayApart(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
svc, fixtures, fetcher := setupHeldFetchService(t)
|
||||
|
||||
cover := startGet(t.Context(), svc, photoVariant(fixtures, 85, FitCover))
|
||||
lowQuality := startGet(t.Context(), svc, photoVariant(fixtures, 40, FitCover))
|
||||
contain := startGet(t.Context(), svc, photoVariant(fixtures, 85, FitContain))
|
||||
|
||||
for range 3 {
|
||||
select {
|
||||
case <-fetcher.started:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("fewer fetches started than variants requested: " +
|
||||
"variants differing in quality or fit were merged")
|
||||
}
|
||||
}
|
||||
|
||||
fetcher.releaseFetches()
|
||||
|
||||
images := make(map[string][]byte)
|
||||
|
||||
for _, variant := range []struct {
|
||||
name string
|
||||
results <-chan getResult
|
||||
width, height int
|
||||
}{
|
||||
{"q=85 fit=cover", cover, 50, 25},
|
||||
{"q=40 fit=cover", lowQuality, 50, 25},
|
||||
{"q=85 fit=contain", contain, 25, 25},
|
||||
} {
|
||||
got := <-variant.results
|
||||
if got.err != nil {
|
||||
t.Fatalf("%s: Get() error = %v", variant.name, got.err)
|
||||
}
|
||||
|
||||
width, height := jpegSize(got.image)
|
||||
if width != variant.width || height != variant.height {
|
||||
t.Errorf("%s: image is %dx%d, want a %dx%d JPEG", variant.name,
|
||||
width, height, variant.width, variant.height)
|
||||
}
|
||||
|
||||
images[variant.name] = got.image
|
||||
}
|
||||
|
||||
if bytes.Equal(images["q=85 fit=cover"], images["q=40 fit=cover"]) {
|
||||
t.Error("q=40 was answered with the q=85 image")
|
||||
}
|
||||
}
|
||||
|
||||
// TestService_Get_WaiterStopsWhenItsContextEnds has a second request for a
|
||||
// variant join the first one's held fetch, then ends the second request's
|
||||
// context. The second request must return at once with the context's error,
|
||||
// while the fetch is still held, and the first must still be answered.
|
||||
func TestService_Get_WaiterStopsWhenItsContextEnds(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
svc, fixtures, fetcher := setupHeldFetchService(t)
|
||||
|
||||
first := startGet(t.Context(), svc, photoVariant(fixtures, 85, FitCover))
|
||||
|
||||
<-fetcher.started
|
||||
|
||||
waiterCtx, cancelWaiter := context.WithCancel(t.Context())
|
||||
waiter := startGet(waiterCtx, svc, photoVariant(fixtures, 85, FitCover))
|
||||
|
||||
time.Sleep(arrivalWait)
|
||||
cancelWaiter()
|
||||
|
||||
select {
|
||||
case got := <-waiter:
|
||||
if !errors.Is(got.err, context.Canceled) {
|
||||
t.Errorf("waiting request: Get() error = %v, want %v",
|
||||
got.err, context.Canceled)
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("waiting request did not return when its context ended")
|
||||
}
|
||||
|
||||
fetcher.releaseFetches()
|
||||
|
||||
got := <-first
|
||||
if got.err != nil {
|
||||
t.Fatalf("first request: Get() error = %v", got.err)
|
||||
}
|
||||
|
||||
if width, height := jpegSize(got.image); width != 50 || height != 25 {
|
||||
t.Errorf("first request: image is %dx%d, want a 50x25 JPEG", width, height)
|
||||
}
|
||||
|
||||
if fetches := fetcher.fetches.Load(); fetches != 1 {
|
||||
t.Errorf("upstream fetches = %d, want 1", fetches)
|
||||
}
|
||||
}
|
||||
|
||||
// TestService_Get_FirstRequestLeavingKeepsTheWork ends the context of the
|
||||
// request whose fetch is held, after a second request has joined it. The fetch
|
||||
// and transcode must go on and answer the second request. The first request
|
||||
// waits for its own work, as every request did before misses were shared, and
|
||||
// is answered too.
|
||||
func TestService_Get_FirstRequestLeavingKeepsTheWork(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
svc, fixtures, fetcher := setupHeldFetchService(t)
|
||||
|
||||
firstCtx, cancelFirst := context.WithCancel(t.Context())
|
||||
first := startGet(firstCtx, svc, photoVariant(fixtures, 85, FitCover))
|
||||
|
||||
<-fetcher.started
|
||||
|
||||
second := startGet(t.Context(), svc, photoVariant(fixtures, 85, FitCover))
|
||||
|
||||
time.Sleep(arrivalWait)
|
||||
cancelFirst()
|
||||
fetcher.releaseFetches()
|
||||
|
||||
for name, results := range map[string]<-chan getResult{
|
||||
"first request": first, "second request": second,
|
||||
} {
|
||||
got := <-results
|
||||
if got.err != nil {
|
||||
t.Fatalf("%s: Get() error = %v", name, got.err)
|
||||
}
|
||||
|
||||
if width, height := jpegSize(got.image); width != 50 || height != 25 {
|
||||
t.Errorf("%s: image is %dx%d, want a 50x25 JPEG", name, width, height)
|
||||
}
|
||||
}
|
||||
|
||||
if fetches := fetcher.fetches.Load(); fetches != 1 {
|
||||
t.Errorf("upstream fetches = %d, want 1", fetches)
|
||||
}
|
||||
}
|
||||
|
||||
// TestService_Get_ConcurrentMissesShareAFailure has several requests for an
|
||||
// image that cannot be served arrive while its fetch is held: the one fetch
|
||||
// answers all of them with its error. The request after them is answered from
|
||||
// the negative cache when the failure is kept there, and fetches again when it
|
||||
// is not.
|
||||
func TestService_Get_ConcurrentMissesShareAFailure(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
path string
|
||||
wantErr error // what every request at once gets
|
||||
wantNextErr error // what the request after them gets
|
||||
wantFetches int32 // fetches once the request after them is answered
|
||||
}{
|
||||
{"upstream answers 404, kept in the negative cache",
|
||||
"/images/missing.jpg", httpfetcher.ErrUpstreamError,
|
||||
ErrNegativeCached, 1},
|
||||
{"source fails the magic byte check, not kept",
|
||||
"/images/text.png", magic.ErrUnknownFormat, magic.ErrUnknownFormat, 2},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
svc, fixtures, fetcher := setupHeldFetchService(t)
|
||||
|
||||
request := func() *ImageRequest {
|
||||
req := photoVariant(fixtures, 85, FitCover)
|
||||
req.SourcePath = tc.path
|
||||
|
||||
return req
|
||||
}
|
||||
|
||||
pending := make([]<-chan getResult, 0, 4)
|
||||
for range 4 {
|
||||
pending = append(pending, startGet(t.Context(), svc, request()))
|
||||
}
|
||||
|
||||
<-fetcher.started
|
||||
time.Sleep(arrivalWait)
|
||||
fetcher.releaseFetches()
|
||||
|
||||
for i, results := range pending {
|
||||
if got := <-results; !errors.Is(got.err, tc.wantErr) {
|
||||
t.Errorf("request %d: Get() error = %v, want %v", i, got.err, tc.wantErr)
|
||||
}
|
||||
}
|
||||
|
||||
_, err := svc.Get(t.Context(), request())
|
||||
if !errors.Is(err, tc.wantNextErr) {
|
||||
t.Errorf("next request: Get() error = %v, want %v", err, tc.wantNextErr)
|
||||
}
|
||||
|
||||
if fetches := fetcher.fetches.Load(); fetches != tc.wantFetches {
|
||||
t.Errorf("upstream fetches = %d, want %d", fetches, tc.wantFetches)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestService_Get_EndedRequestFetchesNothing checks that a request whose
|
||||
// context has already ended when it misses the cache starts no fetch.
|
||||
func TestService_Get_EndedRequestFetchesNothing(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
svc, fixtures, fetcher := setupHeldFetchService(t)
|
||||
|
||||
ctx, cancel := context.WithCancel(t.Context())
|
||||
cancel()
|
||||
|
||||
_, err := svc.Get(ctx, photoVariant(fixtures, 85, FitCover))
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Errorf("Get() error = %v, want %v", err, context.Canceled)
|
||||
}
|
||||
|
||||
if fetches := fetcher.fetches.Load(); fetches != 0 {
|
||||
t.Errorf("upstream fetches = %d, want 0", fetches)
|
||||
}
|
||||
}
|
||||
|
||||
// panickingFetcher panics on every fetch.
|
||||
type panickingFetcher struct{}
|
||||
|
||||
func (panickingFetcher) Fetch(
|
||||
context.Context, string,
|
||||
) (*httpfetcher.FetchResult, error) {
|
||||
panic("upstream fetcher panicked")
|
||||
}
|
||||
|
||||
// TestService_Get_PanicBecomesAnError checks that a panic while a variant is
|
||||
// being made reaches its request as an error naming the panic, instead of
|
||||
// being raised again.
|
||||
func TestService_Get_PanicBecomesAnError(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
svc, fixtures := SetupTestService(t)
|
||||
svc.fetcher = panickingFetcher{}
|
||||
|
||||
var err error
|
||||
|
||||
func() {
|
||||
defer func() {
|
||||
if recovered := recover(); recovered != nil {
|
||||
t.Fatalf("Get() panicked: %v", recovered)
|
||||
}
|
||||
}()
|
||||
|
||||
_, err = svc.Get(t.Context(), photoVariant(fixtures, 85, FitCover))
|
||||
}()
|
||||
|
||||
if err == nil || !strings.Contains(err.Error(), "upstream fetcher panicked") {
|
||||
t.Errorf("Get() error = %v, want one naming the panic", err)
|
||||
}
|
||||
}
|
||||
@@ -39,8 +39,8 @@ const tempFilePrefix = ".tmp-"
|
||||
// to each variant file.
|
||||
const variantMetaSuffix = ".meta"
|
||||
|
||||
// fallbackContentType is the content type given to a variant file that
|
||||
// has no readable .meta sidecar, when it is served or reconciled.
|
||||
// fallbackContentType is recorded when a reconciled variant file has
|
||||
// no readable .meta sidecar.
|
||||
const fallbackContentType = "application/octet-stream"
|
||||
|
||||
// UsageBytes returns the total number of bytes of cache content
|
||||
@@ -271,9 +271,7 @@ func (c *Cache) sourceCandidates(ctx context.Context) ([]evictionCandidate, erro
|
||||
|
||||
// evictVariant removes one variant: accounting row first, then the
|
||||
// content and .meta files, so the database never references a deleted
|
||||
// file. The metaCache entry goes before the files; a GetVariant that
|
||||
// read them just before may put it back, and the next GetVariant then
|
||||
// fails to open the file and removes it again.
|
||||
// file.
|
||||
func (c *Cache) evictVariant(ctx context.Context, cacheKey VariantKey) error {
|
||||
_, err := c.db.ExecContext(ctx,
|
||||
`DELETE FROM variant_content WHERE cache_key = ?`, string(cacheKey))
|
||||
@@ -281,8 +279,6 @@ func (c *Cache) evictVariant(ctx context.Context, cacheKey VariantKey) error {
|
||||
return fmt.Errorf("failed to delete variant accounting row: %w", err)
|
||||
}
|
||||
|
||||
c.metaCache.Remove(cacheKey)
|
||||
|
||||
err = c.variants.DeleteWithMeta(cacheKey)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
@@ -1,349 +0,0 @@
|
||||
package imgcache
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// webpRequest returns a request for a 100x100 WebP variant of path.
|
||||
func webpRequest(path string) *ImageRequest {
|
||||
return &ImageRequest{
|
||||
SourceHost: testHostCDN,
|
||||
SourcePath: path,
|
||||
Size: Size{Width: 100, Height: 100},
|
||||
Format: FormatWebP,
|
||||
Quality: 85,
|
||||
FitMode: FitCover,
|
||||
}
|
||||
}
|
||||
|
||||
// assertVariantServed checks that GetVariant serves key with the given
|
||||
// content and the image/webp content type storeEvictionTestVariant stores.
|
||||
func assertVariantServed(t *testing.T, cache *Cache, key VariantKey, content []byte) {
|
||||
t.Helper()
|
||||
|
||||
reader, size, contentType, err := cache.GetVariant(key)
|
||||
if err != nil {
|
||||
t.Fatalf("GetVariant(%s) error = %v", key, err)
|
||||
}
|
||||
|
||||
defer func() { _ = reader.Close() }()
|
||||
|
||||
got, err := io.ReadAll(reader)
|
||||
if err != nil {
|
||||
t.Fatalf("reading variant %s: %v", key, err)
|
||||
}
|
||||
|
||||
if !bytes.Equal(got, content) {
|
||||
t.Errorf("GetVariant(%s) content = %q, want %q", key, got, content)
|
||||
}
|
||||
|
||||
if size != int64(len(content)) {
|
||||
t.Errorf("GetVariant(%s) size = %d, want %d", key, size, len(content))
|
||||
}
|
||||
|
||||
if contentType != testContentTypeWebP {
|
||||
t.Errorf("GetVariant(%s) content type = %q, want %q",
|
||||
key, contentType, testContentTypeWebP)
|
||||
}
|
||||
}
|
||||
|
||||
// assertVariantNotFound checks that GetVariant refuses key with
|
||||
// ErrNotFound.
|
||||
func assertVariantNotFound(t *testing.T, cache *Cache, key VariantKey) {
|
||||
t.Helper()
|
||||
|
||||
reader, _, _, err := cache.GetVariant(key)
|
||||
if err == nil {
|
||||
_ = reader.Close()
|
||||
}
|
||||
|
||||
if !errors.Is(err, ErrNotFound) {
|
||||
t.Errorf("GetVariant(%s) error = %v, want ErrNotFound", key, err)
|
||||
}
|
||||
}
|
||||
|
||||
// assertLookupMisses checks that Lookup reports request as a miss.
|
||||
func assertLookupMisses(t *testing.T, cache *Cache, request *ImageRequest) {
|
||||
t.Helper()
|
||||
|
||||
lookup, err := cache.Lookup(t.Context(), request)
|
||||
if err != nil {
|
||||
t.Fatalf("Lookup(%s) error = %v", request.SourcePath, err)
|
||||
}
|
||||
|
||||
if lookup.Hit {
|
||||
t.Errorf("Lookup(%s) is a hit, want a miss", request.SourcePath)
|
||||
}
|
||||
}
|
||||
|
||||
// renameFile renames the file at from to to.
|
||||
func renameFile(t *testing.T, from, to string) {
|
||||
t.Helper()
|
||||
|
||||
err := os.Rename(from, to)
|
||||
if err != nil {
|
||||
t.Fatalf("renaming %s: %v", from, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSecondHitDoesNotReadMetaFile checks that once a variant has been
|
||||
// stored or read, a hit takes its content type from memory: with the
|
||||
// .meta file deleted, GetVariant must still return the stored content
|
||||
// type rather than the application/octet-stream it uses without one.
|
||||
func TestSecondHitDoesNotReadMetaFile(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<20)
|
||||
content := []byte("webp variant bytes")
|
||||
|
||||
storeEvictionTestVariant(t, cache, testVariantKeyOne, content)
|
||||
|
||||
// A second Cache on the same state directory starts with nothing in
|
||||
// memory, as pixad does after a restart, so its first read uses the
|
||||
// .meta file.
|
||||
restarted, err := NewCache(cache.db, cache.config)
|
||||
if err != nil {
|
||||
t.Fatalf("NewCache() error = %v", err)
|
||||
}
|
||||
|
||||
assertVariantServed(t, restarted, testVariantKeyOne, content)
|
||||
|
||||
err = os.Remove(cache.variants.keyToPath(testVariantKeyOne) + ".meta")
|
||||
if err != nil {
|
||||
t.Fatalf("removing .meta file: %v", err)
|
||||
}
|
||||
|
||||
assertVariantServed(t, cache, testVariantKeyOne, content)
|
||||
assertVariantServed(t, restarted, testVariantKeyOne, content)
|
||||
}
|
||||
|
||||
// TestReadDuringStoreKeepsStoredContentType checks that a GetVariant
|
||||
// which began before StoreVariant finished cannot replace the content
|
||||
// type the store kept in memory. Such a read can find the variant file
|
||||
// but not yet its .meta file, and so gets application/octet-stream. The
|
||||
// test deletes the .meta file after the store, then runs the part of
|
||||
// GetVariant that comes after its check of memory.
|
||||
func TestReadDuringStoreKeepsStoredContentType(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<20)
|
||||
content := []byte("webp variant bytes")
|
||||
|
||||
storeEvictionTestVariant(t, cache, testVariantKeyOne, content)
|
||||
|
||||
err := os.Remove(cache.variants.keyToPath(testVariantKeyOne) + ".meta")
|
||||
if err != nil {
|
||||
t.Fatalf("removing .meta file: %v", err)
|
||||
}
|
||||
|
||||
reader, _, contentType, err := cache.loadVariantWithMeta(testVariantKeyOne)
|
||||
if err != nil {
|
||||
t.Fatalf("loadVariantWithMeta(%s) error = %v", testVariantKeyOne, err)
|
||||
}
|
||||
|
||||
_ = reader.Close()
|
||||
|
||||
t.Logf("the read without a .meta file got content type %q", contentType)
|
||||
|
||||
assertVariantServed(t, cache, testVariantKeyOne, content)
|
||||
}
|
||||
|
||||
// TestReadOfOlderMetaFileKeepsStoredContentType checks that a read
|
||||
// which got its content type from a .meta file that StoreVariant had
|
||||
// not yet rewritten cannot replace the type the store kept in memory.
|
||||
// The test writes such a .meta file, with a different content type,
|
||||
// after the store, then runs the part of GetVariant that comes after
|
||||
// its check of memory.
|
||||
func TestReadOfOlderMetaFileKeepsStoredContentType(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<20)
|
||||
content := []byte("webp variant bytes")
|
||||
|
||||
storeEvictionTestVariant(t, cache, testVariantKeyOne, content)
|
||||
|
||||
olderMeta, err := json.Marshal(VariantMeta{
|
||||
ContentType: testContentTypeJPEG,
|
||||
Size: int64(len(content)),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("encoding .meta file: %v", err)
|
||||
}
|
||||
|
||||
metaPath := cache.variants.keyToPath(testVariantKeyOne) + ".meta"
|
||||
|
||||
err = os.WriteFile(metaPath, olderMeta, StorageFilePerm)
|
||||
if err != nil {
|
||||
t.Fatalf("writing .meta file: %v", err)
|
||||
}
|
||||
|
||||
reader, _, contentType, err := cache.loadVariantWithMeta(testVariantKeyOne)
|
||||
if err != nil {
|
||||
t.Fatalf("loadVariantWithMeta(%s) error = %v", testVariantKeyOne, err)
|
||||
}
|
||||
|
||||
_ = reader.Close()
|
||||
|
||||
if contentType != testContentTypeJPEG {
|
||||
t.Fatalf("loadVariantWithMeta(%s) content type = %q, want %q from the .meta file",
|
||||
testVariantKeyOne, contentType, testContentTypeJPEG)
|
||||
}
|
||||
|
||||
kept, _ := cache.metaCache.Get(testVariantKeyOne)
|
||||
if kept != testContentTypeWebP {
|
||||
t.Errorf("content type in memory = %q, want the stored %q",
|
||||
kept, testContentTypeWebP)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFailedReadDuringStoreKeepsStoredContentType checks that a read
|
||||
// which found no .meta file cannot leave application/octet-stream in
|
||||
// memory, even when another read has removed the content type
|
||||
// StoreVariant kept there. In this order: the store; a read that finds
|
||||
// the variant in memory but cannot open its file, and so removes it
|
||||
// from memory; a read that opened the variant file before the store
|
||||
// wrote its .meta file. Later hits must get the stored content type.
|
||||
func TestFailedReadDuringStoreKeepsStoredContentType(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<20)
|
||||
content := []byte("webp variant bytes")
|
||||
variantPath := cache.variants.keyToPath(testVariantKeyOne)
|
||||
metaPath := variantPath + ".meta"
|
||||
|
||||
storeEvictionTestVariant(t, cache, testVariantKeyOne, content)
|
||||
|
||||
renameFile(t, variantPath, variantPath+".hidden")
|
||||
assertVariantNotFound(t, cache, testVariantKeyOne)
|
||||
renameFile(t, variantPath+".hidden", variantPath)
|
||||
|
||||
renameFile(t, metaPath, metaPath+".hidden")
|
||||
|
||||
reader, _, contentType, err := cache.GetVariant(testVariantKeyOne)
|
||||
if err != nil {
|
||||
t.Fatalf("GetVariant(%s) error = %v", testVariantKeyOne, err)
|
||||
}
|
||||
|
||||
_ = reader.Close()
|
||||
|
||||
t.Logf("the read without a .meta file got content type %q", contentType)
|
||||
|
||||
renameFile(t, metaPath+".hidden", metaPath)
|
||||
|
||||
assertVariantServed(t, cache, testVariantKeyOne, content)
|
||||
}
|
||||
|
||||
// TestEvictedVariantIsNotServed checks that a variant the evictor
|
||||
// removed is a miss and cannot be read, although it had been stored
|
||||
// and served before.
|
||||
func TestEvictedVariantIsNotServed(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1500)
|
||||
|
||||
oldRequest := webpRequest("/old.jpg")
|
||||
newRequest := webpRequest("/new.jpg")
|
||||
oldKey := CacheKey(oldRequest)
|
||||
newKey := CacheKey(newRequest)
|
||||
oldContent := bytes.Repeat([]byte{0x01}, 1000)
|
||||
newContent := bytes.Repeat([]byte{0x02}, 1000)
|
||||
|
||||
storeEvictionTestVariant(t, cache, oldKey, oldContent)
|
||||
storeEvictionTestVariant(t, cache, newKey, newContent)
|
||||
assertVariantServed(t, cache, oldKey, oldContent)
|
||||
setVariantLastAccessed(t, cache, oldKey, time.Now().Add(-time.Hour))
|
||||
|
||||
err := cache.EvictToLimit(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("EvictToLimit() error = %v", err)
|
||||
}
|
||||
|
||||
assertLookupMisses(t, cache, oldRequest)
|
||||
assertVariantNotFound(t, cache, oldKey)
|
||||
assertVariantServed(t, cache, newKey, newContent)
|
||||
}
|
||||
|
||||
// TestVariantDeletedFromDiskIsNotServed checks that a variant whose
|
||||
// file was deleted by something other than the evictor cannot be read,
|
||||
// and is a miss afterwards.
|
||||
func TestVariantDeletedFromDiskIsNotServed(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<20)
|
||||
|
||||
request := webpRequest("/deleted.jpg")
|
||||
key := CacheKey(request)
|
||||
content := []byte("webp variant bytes")
|
||||
|
||||
storeEvictionTestVariant(t, cache, key, content)
|
||||
assertVariantServed(t, cache, key, content)
|
||||
|
||||
err := os.Remove(cache.variants.keyToPath(key))
|
||||
if err != nil {
|
||||
t.Fatalf("removing variant file: %v", err)
|
||||
}
|
||||
|
||||
assertVariantNotFound(t, cache, key)
|
||||
assertLookupMisses(t, cache, request)
|
||||
}
|
||||
|
||||
// TestConcurrentVariantStoreReadAndEvict stores, reads and evicts
|
||||
// variants from several goroutines at once, for the race detector.
|
||||
func TestConcurrentVariantStoreReadAndEvict(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<20)
|
||||
ctx := t.Context()
|
||||
|
||||
var wg sync.WaitGroup
|
||||
|
||||
for goroutine := range 8 {
|
||||
wg.Go(func() {
|
||||
key := VariantKey(fmt.Sprintf("aabbccdd01%02d", goroutine))
|
||||
content := []byte(key)
|
||||
|
||||
for range 20 {
|
||||
err := cache.StoreVariant(
|
||||
ctx, key, bytes.NewReader(content), testContentTypeWebP)
|
||||
if err != nil {
|
||||
t.Errorf("StoreVariant(%s) error = %v", key, err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
reader, _, contentType, err := cache.GetVariant(key)
|
||||
if err != nil {
|
||||
t.Errorf("GetVariant(%s) error = %v", key, err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
_ = reader.Close()
|
||||
|
||||
if contentType != testContentTypeWebP {
|
||||
t.Errorf("GetVariant(%s) content type = %q, want %q",
|
||||
key, contentType, testContentTypeWebP)
|
||||
}
|
||||
|
||||
err = cache.evictVariant(ctx, key)
|
||||
if err != nil {
|
||||
t.Errorf("evictVariant(%s) error = %v", key, err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
assertVariantNotFound(t, cache, key)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
}
|
||||
+27
-111
@@ -8,11 +8,9 @@ import (
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/url"
|
||||
"runtime/debug"
|
||||
"time"
|
||||
|
||||
"github.com/dustin/go-humanize"
|
||||
"golang.org/x/sync/singleflight"
|
||||
"sneak.berlin/go/pixa/internal/allowlist"
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
"sneak.berlin/go/pixa/internal/imageprocessor"
|
||||
@@ -31,9 +29,6 @@ type Service struct {
|
||||
log *slog.Logger
|
||||
allowHTTP bool
|
||||
maxResponseSize int64
|
||||
// variantsInProgress lets the requests that miss the same variant at the
|
||||
// same time share one fetch and one transcode.
|
||||
variantsInProgress singleflight.Group
|
||||
}
|
||||
|
||||
// ServiceConfig holds configuration for the image service.
|
||||
@@ -165,12 +160,14 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e
|
||||
}
|
||||
}
|
||||
|
||||
// Cache miss - get the variant, processed once for all the requests that
|
||||
// miss it at the same time, then count this request's miss, also when it
|
||||
// failed or the request context has ended meanwhile
|
||||
response, err := s.processOrWait(ctx, req)
|
||||
// Cache miss - process the cached source or fetch it, then count the
|
||||
// miss with the bytes it fetched from upstream, also when it failed or
|
||||
// the request context has ended meanwhile
|
||||
cacheKey := CacheKey(req)
|
||||
|
||||
s.cache.IncrementStats(context.WithoutCancel(ctx), false, 0)
|
||||
response, fetchedBytes, err := s.processFromSourceOrFetch(ctx, req, cacheKey)
|
||||
|
||||
s.cache.IncrementStats(context.WithoutCancel(ctx), false, fetchedBytes)
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -244,83 +241,6 @@ func (s *Service) GenerateSignedURL(
|
||||
baseURL, path, sig, exp, req.Quality, req.FitMode), nil
|
||||
}
|
||||
|
||||
// errPanicked is returned when processing a variant panicked.
|
||||
var errPanicked = errors.New("panic while processing image")
|
||||
|
||||
// processOrWait returns the variant req asks for. The first of the requests
|
||||
// that miss a variant at the same time processes it, and singleflight hands
|
||||
// its result, or its error, to the others: they fetch nothing, read no source
|
||||
// and take no upstream connection or processing slot. The processing runs with
|
||||
// a context that does not end with the first request's, so the others are
|
||||
// still served if that client goes away; the fetch timeout and the waits for a
|
||||
// connection and a processing slot still bound it.
|
||||
func (s *Service) processOrWait(
|
||||
ctx context.Context, req *ImageRequest,
|
||||
) (*ImageResponse, error) {
|
||||
// A request that has already ended starts no processing
|
||||
if ctx.Err() != nil {
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
|
||||
cacheKey := CacheKey(req)
|
||||
|
||||
// Closed when this request's own function runs, which singleflight does
|
||||
// only when no other request is processing the variant
|
||||
processing := make(chan struct{})
|
||||
|
||||
results := s.variantsInProgress.DoChan(string(cacheKey),
|
||||
func() (_ any, err error) {
|
||||
close(processing)
|
||||
|
||||
// singleflight would raise a panic again in a goroutine of its
|
||||
// own, where no handler recovers it, and stop pixad
|
||||
defer func() {
|
||||
recovered := recover()
|
||||
if recovered != nil {
|
||||
s.log.Error("panic while processing image",
|
||||
"host", req.SourceHost, "path", req.SourcePath,
|
||||
"panic", recovered, "stack", string(debug.Stack()))
|
||||
|
||||
err = fmt.Errorf("%w: %v", errPanicked, recovered)
|
||||
}
|
||||
}()
|
||||
|
||||
return s.processFromSourceOrFetch(
|
||||
context.WithoutCancel(ctx), req, cacheKey)
|
||||
})
|
||||
|
||||
var result singleflight.Result
|
||||
|
||||
select {
|
||||
case result = <-results:
|
||||
case <-ctx.Done():
|
||||
select {
|
||||
case <-processing:
|
||||
// This request is processing the variant: it waits for the
|
||||
// result, as every request did before misses were shared
|
||||
result = <-results
|
||||
default:
|
||||
// Another request is processing the variant, or this request's
|
||||
// function has not started yet; the processing goes on without it
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
if result.Err != nil {
|
||||
return nil, result.Err
|
||||
}
|
||||
|
||||
variant, _ := result.Val.(*processedVariant)
|
||||
|
||||
return &ImageResponse{
|
||||
Content: io.NopCloser(bytes.NewReader(variant.data)),
|
||||
ContentLength: int64(len(variant.data)),
|
||||
ContentType: variant.contentType,
|
||||
FetchedBytes: variant.fetchedBytes,
|
||||
ETag: formatETag(cacheKey),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// 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.
|
||||
@@ -355,12 +275,13 @@ func (s *Service) loadCachedSource(
|
||||
}
|
||||
|
||||
// processFromSourceOrFetch processes an image, using cached source content
|
||||
// if available.
|
||||
// if available. It also returns the number of bytes fetched from upstream,
|
||||
// as fetchAndProcess does, or 0 when the cached source was used.
|
||||
func (s *Service) processFromSourceOrFetch(
|
||||
ctx context.Context,
|
||||
req *ImageRequest,
|
||||
cacheKey VariantKey,
|
||||
) (*processedVariant, error) {
|
||||
) (*ImageResponse, int64, error) {
|
||||
// Check if we have cached source content
|
||||
contentHash, _, err := s.cache.LookupSource(ctx, req)
|
||||
if err != nil {
|
||||
@@ -387,17 +308,19 @@ func (s *Service) 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 s.processAndStore(ctx, req, cacheKey, source, sourceSize)
|
||||
resp, err := s.processAndStore(ctx, req, cacheKey, source, sourceSize)
|
||||
|
||||
return resp, 0, err
|
||||
}
|
||||
|
||||
// fetchAndProcess fetches from upstream, processes, and caches the result.
|
||||
// It counts the fetch with the bytes read from upstream, including when
|
||||
// It also returns the number of bytes read from upstream, including when
|
||||
// reading the response or a later step fails.
|
||||
func (s *Service) fetchAndProcess(
|
||||
ctx context.Context,
|
||||
req *ImageRequest,
|
||||
cacheKey VariantKey,
|
||||
) (*processedVariant, error) {
|
||||
) (*ImageResponse, int64, error) {
|
||||
// Fetch from upstream
|
||||
sourceURL := req.SourceURL()
|
||||
|
||||
@@ -416,7 +339,7 @@ func (s *Service) fetchAndProcess(
|
||||
}
|
||||
}
|
||||
|
||||
return nil, fmt.Errorf("upstream fetch failed: %w", err)
|
||||
return nil, 0, fmt.Errorf("upstream fetch failed: %w", err)
|
||||
}
|
||||
|
||||
// Closing the body frees the upstream connection. It is closed only
|
||||
@@ -429,11 +352,8 @@ func (s *Service) fetchAndProcess(
|
||||
sourceData, err := io.ReadAll(fetchResult.Content)
|
||||
fetchBytes := int64(len(sourceData))
|
||||
|
||||
// Counted also when the request context has ended meanwhile
|
||||
s.cache.IncrementUpstreamFetch(context.WithoutCancel(ctx), fetchBytes)
|
||||
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read upstream response: %w", err)
|
||||
return nil, fetchBytes, fmt.Errorf("failed to read upstream response: %w", err)
|
||||
}
|
||||
|
||||
// Calculate download bitrate
|
||||
@@ -461,7 +381,7 @@ func (s *Service) fetchAndProcess(
|
||||
// Validate magic bytes match content type
|
||||
err = magic.ValidateMagicBytes(sourceData, fetchResult.ContentType)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("content validation failed: %w", err)
|
||||
return nil, fetchBytes, fmt.Errorf("content validation failed: %w", err)
|
||||
}
|
||||
|
||||
// Store source content
|
||||
@@ -471,17 +391,11 @@ func (s *Service) fetchAndProcess(
|
||||
// Continue even if caching fails
|
||||
}
|
||||
|
||||
return s.processAndStore(
|
||||
resp, err := s.processAndStore(
|
||||
ctx, req, cacheKey, bytes.NewReader(sourceData), fetchBytes,
|
||||
)
|
||||
}
|
||||
|
||||
// processedVariant is a variant as processAndStore made it. Each request that
|
||||
// shared its processing serves it through a reader of its own.
|
||||
type processedVariant struct {
|
||||
data []byte
|
||||
contentType string
|
||||
fetchedBytes int64
|
||||
return resp, fetchBytes, err
|
||||
}
|
||||
|
||||
// processAndStore processes the image read from source and stores the
|
||||
@@ -492,7 +406,7 @@ func (s *Service) processAndStore(
|
||||
cacheKey VariantKey,
|
||||
source io.Reader,
|
||||
fetchBytes int64,
|
||||
) (*processedVariant, error) {
|
||||
) (*ImageResponse, error) {
|
||||
// Process the image
|
||||
processStart := time.Now()
|
||||
|
||||
@@ -556,10 +470,12 @@ func (s *Service) processAndStore(
|
||||
// Continue even if caching fails
|
||||
}
|
||||
|
||||
return &processedVariant{
|
||||
data: processedData,
|
||||
contentType: processResult.ContentType,
|
||||
fetchedBytes: fetchBytes,
|
||||
return &ImageResponse{
|
||||
Content: io.NopCloser(bytes.NewReader(processedData)),
|
||||
ContentLength: outputSize,
|
||||
ContentType: processResult.ContentType,
|
||||
FetchedBytes: fetchBytes,
|
||||
ETag: formatETag(cacheKey),
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -506,44 +506,32 @@ func (s *VariantStorage) Load(key VariantKey) (io.ReadCloser, error) {
|
||||
return f, nil
|
||||
}
|
||||
|
||||
// LoadWithSize returns a reader and file size for the content at the
|
||||
// given key.
|
||||
func (s *VariantStorage) LoadWithSize(key VariantKey) (io.ReadCloser, int64, error) {
|
||||
// LoadWithMeta returns a reader, size, and content type for the content at
|
||||
// the given key.
|
||||
func (s *VariantStorage) LoadWithMeta(
|
||||
key VariantKey,
|
||||
) (io.ReadCloser, int64, string, error) {
|
||||
path := s.keyToPath(key)
|
||||
metaPath := path + ".meta"
|
||||
|
||||
f, err := os.Open(path) //nolint:gosec // path derived from cache key
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return nil, 0, ErrNotFound
|
||||
return nil, 0, "", ErrNotFound
|
||||
}
|
||||
|
||||
return nil, 0, fmt.Errorf("failed to open content: %w", err)
|
||||
return nil, 0, "", fmt.Errorf("failed to open content: %w", err)
|
||||
}
|
||||
|
||||
stat, err := f.Stat()
|
||||
if err != nil {
|
||||
_ = f.Close()
|
||||
|
||||
return nil, 0, fmt.Errorf("failed to stat content: %w", err)
|
||||
return nil, 0, "", fmt.Errorf("failed to stat content: %w", err)
|
||||
}
|
||||
|
||||
return f, stat.Size(), nil
|
||||
}
|
||||
|
||||
// LoadWithMeta returns a reader, size, and content type for the content at
|
||||
// the given key. The content type is read from the .meta file, and is
|
||||
// empty when that file is missing or unreadable.
|
||||
func (s *VariantStorage) LoadWithMeta(
|
||||
key VariantKey,
|
||||
) (io.ReadCloser, int64, string, error) {
|
||||
f, size, err := s.LoadWithSize(key)
|
||||
if err != nil {
|
||||
return nil, 0, "", err
|
||||
}
|
||||
|
||||
var contentType string
|
||||
|
||||
metaPath := s.keyToPath(key) + ".meta"
|
||||
// Load metadata for content type
|
||||
contentType := "application/octet-stream" // fallback
|
||||
|
||||
metaData, err := os.ReadFile(metaPath) //nolint:gosec // path derived from cache key
|
||||
if err == nil {
|
||||
@@ -553,7 +541,7 @@ func (s *VariantStorage) LoadWithMeta(
|
||||
}
|
||||
}
|
||||
|
||||
return f, size, contentType, nil
|
||||
return f, stat.Size(), contentType, nil
|
||||
}
|
||||
|
||||
// Exists checks if content exists at the given key.
|
||||
|
||||
@@ -24,7 +24,6 @@ const (
|
||||
testHostExample = "example.com"
|
||||
testPathCat = "/photos/cat.jpg"
|
||||
testContentTypeJPEG = "image/jpeg"
|
||||
testContentTypeWebP = "image/webp"
|
||||
testHeaderContentType = "Content-Type"
|
||||
)
|
||||
|
||||
|
||||
@@ -18,7 +18,6 @@ import (
|
||||
"sneak.berlin/go/pixa/internal/database"
|
||||
"sneak.berlin/go/pixa/internal/globals"
|
||||
"sneak.berlin/go/pixa/internal/handlers"
|
||||
"sneak.berlin/go/pixa/internal/healthcheck"
|
||||
"sneak.berlin/go/pixa/internal/logger"
|
||||
"sneak.berlin/go/pixa/internal/middleware"
|
||||
)
|
||||
@@ -72,15 +71,8 @@ func newTestServer(t *testing.T) *Server {
|
||||
t.Fatalf("database.New() error = %v", err)
|
||||
}
|
||||
|
||||
hc, err := healthcheck.New(lc, healthcheck.Params{
|
||||
Globals: &globals.Globals{}, Config: cfg, Logger: log, Database: db,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("healthcheck.New() error = %v", err)
|
||||
}
|
||||
|
||||
h, err := handlers.New(lc, handlers.Params{
|
||||
Logger: log, Healthcheck: hc, Database: db, Config: cfg,
|
||||
Logger: log, Database: db, Config: cfg,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("handlers.New() error = %v", err)
|
||||
|
||||
@@ -1,169 +0,0 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strconv"
|
||||
"testing"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/healthcheck"
|
||||
)
|
||||
|
||||
// unsignedImagePath is an image URL that carries no signature.
|
||||
const unsignedImagePath = "/v1/image/cdn.example.com/cat.jpg/100x100.jpeg"
|
||||
|
||||
// TestMaintenanceModeRefusesImageRequests verifies that while maintenance
|
||||
// mode is on, both image routes answer 503 Service Unavailable with a
|
||||
// Retry-After header and the JSON error body the image handlers send.
|
||||
func TestMaintenanceModeRefusesImageRequests(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newTestServer(t)
|
||||
s.config.MaintenanceMode = true
|
||||
|
||||
requests := []struct {
|
||||
method string
|
||||
path string
|
||||
}{
|
||||
{http.MethodGet, unsignedImagePath},
|
||||
{http.MethodHead, unsignedImagePath},
|
||||
{http.MethodGet, "/v1/e/token/cat.jpg"},
|
||||
}
|
||||
|
||||
for _, tc := range requests {
|
||||
t.Run(tc.method+" "+tc.path, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
s.ServeHTTP(rec, httptest.NewRequestWithContext(
|
||||
t.Context(), tc.method, tc.path, nil))
|
||||
t.Logf("status %d, body %s", rec.Code, rec.Body.String())
|
||||
|
||||
if rec.Code != http.StatusServiceUnavailable {
|
||||
t.Fatalf("status = %d, want %d",
|
||||
rec.Code, http.StatusServiceUnavailable)
|
||||
}
|
||||
|
||||
retryAfter := rec.Header().Get("Retry-After")
|
||||
|
||||
seconds, err := strconv.Atoi(retryAfter)
|
||||
if err != nil || seconds <= 0 {
|
||||
t.Errorf("Retry-After = %q, want a positive number of seconds",
|
||||
retryAfter)
|
||||
}
|
||||
|
||||
// A HEAD response carries no body.
|
||||
if tc.method == http.MethodHead {
|
||||
return
|
||||
}
|
||||
|
||||
var body struct {
|
||||
Error string `json:"error"`
|
||||
Status int `json:"status"`
|
||||
Timestamp string `json:"timestamp"`
|
||||
}
|
||||
|
||||
err = json.NewDecoder(rec.Body).Decode(&body)
|
||||
if err != nil {
|
||||
t.Fatalf("body is not JSON: %v", err)
|
||||
}
|
||||
|
||||
if body.Error == "" || body.Status != http.StatusServiceUnavailable ||
|
||||
body.Timestamp == "" {
|
||||
t.Errorf("body = %+v, want an error, status %d and a timestamp",
|
||||
body, http.StatusServiceUnavailable)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestImageRequestsServedWithoutMaintenanceMode verifies that while
|
||||
// maintenance mode is off, image requests reach the image handlers instead
|
||||
// of the 503. The handlers refuse an unsigned image URL with 401 and a token
|
||||
// they cannot decrypt with 400, so either status shows a request got through.
|
||||
func TestImageRequestsServedWithoutMaintenanceMode(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newTestServer(t)
|
||||
s.config.MaintenanceMode = false
|
||||
|
||||
requests := []struct {
|
||||
method string
|
||||
path string
|
||||
want int
|
||||
}{
|
||||
{http.MethodGet, unsignedImagePath, http.StatusUnauthorized},
|
||||
{http.MethodHead, unsignedImagePath, http.StatusUnauthorized},
|
||||
{http.MethodGet, "/v1/e/token/cat.jpg", http.StatusBadRequest},
|
||||
}
|
||||
|
||||
for _, tc := range requests {
|
||||
t.Run(tc.method+" "+tc.path, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
s.ServeHTTP(rec, httptest.NewRequestWithContext(
|
||||
t.Context(), tc.method, tc.path, nil))
|
||||
t.Logf("status %d, body %s", rec.Code, rec.Body.String())
|
||||
|
||||
if rec.Code != tc.want {
|
||||
t.Errorf("status = %d, want %d from the image handler",
|
||||
rec.Code, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestMaintenanceModeKeepsOtherRoutes verifies that while maintenance mode
|
||||
// is on, the health check still answers 200 and reports it, and the login
|
||||
// page and /metrics still answer 200. The image's Docker HEALTHCHECK
|
||||
// requests the health check: a 503 there would make the container
|
||||
// unhealthy, and upaas marks a deploy failed when its container is
|
||||
// unhealthy.
|
||||
func TestMaintenanceModeKeepsOtherRoutes(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newTestServer(t)
|
||||
s.config.MaintenanceMode = true
|
||||
|
||||
// /metrics is routed only when its username is set.
|
||||
s.config.MetricsUsername = "metrics"
|
||||
s.config.MetricsPassword = "metrics-password"
|
||||
s.SetupRoutes()
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
s.ServeHTTP(rec, httptest.NewRequestWithContext(t.Context(),
|
||||
http.MethodGet, "/.well-known/healthcheck.json", nil))
|
||||
t.Logf("health check status %d, body %s", rec.Code, rec.Body.String())
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("health check status = %d, want %d", rec.Code, http.StatusOK)
|
||||
}
|
||||
|
||||
var health healthcheck.Response
|
||||
|
||||
err := json.NewDecoder(rec.Body).Decode(&health)
|
||||
if err != nil || !health.Maintenance {
|
||||
t.Errorf("health check maintenance_mode = %v (error %v), want true",
|
||||
health.Maintenance, err)
|
||||
}
|
||||
|
||||
rec = httptest.NewRecorder()
|
||||
s.ServeHTTP(rec, clientRequest(t, http.MethodGet, nil, firstClient, ""))
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Errorf("login page status = %d, want %d", rec.Code, http.StatusOK)
|
||||
}
|
||||
|
||||
req := httptest.NewRequestWithContext(t.Context(),
|
||||
http.MethodGet, "/metrics", nil)
|
||||
req.SetBasicAuth(s.config.MetricsUsername, s.config.MetricsPassword)
|
||||
|
||||
rec = httptest.NewRecorder()
|
||||
s.ServeHTTP(rec, req)
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Errorf("/metrics status = %d, want %d", rec.Code, http.StatusOK)
|
||||
}
|
||||
}
|
||||
@@ -1,9 +1,7 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
sentryhttp "github.com/getsentry/sentry-go/http"
|
||||
@@ -19,10 +17,6 @@ import (
|
||||
// make per minute; the next is refused with 429 Too Many Requests.
|
||||
const LoginAttemptsPerMinute = 5
|
||||
|
||||
// MaintenanceRetryAfterSeconds is the Retry-After, in seconds, sent with
|
||||
// the 503 that the image routes answer while maintenance mode is on.
|
||||
const MaintenanceRetryAfterSeconds = 300
|
||||
|
||||
// SetupRoutes configures all HTTP routes.
|
||||
func (s *Server) SetupRoutes() {
|
||||
s.router = chi.NewRouter()
|
||||
@@ -74,23 +68,15 @@ func (s *Server) SetupRoutes() {
|
||||
|
||||
s.router.Get("/logout", s.h.HandleLogout())
|
||||
|
||||
// Image routes, refused while maintenance mode is on. Only these: the
|
||||
// image's Docker HEALTHCHECK requests the health check, a 503 there
|
||||
// would make the container unhealthy, and upaas marks a deploy failed
|
||||
// when its container is unhealthy.
|
||||
s.router.Group(func(r chi.Router) {
|
||||
r.Use(s.refuseDuringMaintenance)
|
||||
// Main image proxy route
|
||||
// /v1/image/<host>/<path>/<width>x<height>.<format>
|
||||
s.router.Get("/v1/image/*", s.h.HandleImage())
|
||||
s.router.Head("/v1/image/*", s.h.HandleImage())
|
||||
|
||||
// Main image proxy route
|
||||
// /v1/image/<host>/<path>/<width>x<height>.<format>
|
||||
r.Get("/v1/image/*", s.h.HandleImage())
|
||||
r.Head("/v1/image/*", s.h.HandleImage())
|
||||
|
||||
// Encrypted image URL route
|
||||
// The trailing filename (e.g., /img.jpg) is ignored but helps
|
||||
// browsers with content type
|
||||
r.Get("/v1/e/{token}/*", s.h.HandleImageEnc())
|
||||
})
|
||||
// Encrypted image URL route
|
||||
// The trailing filename (e.g., /img.jpg) is ignored but helps
|
||||
// browsers with content type
|
||||
s.router.Get("/v1/e/{token}/*", s.h.HandleImageEnc())
|
||||
|
||||
// Metrics endpoint with auth
|
||||
if s.config.MetricsUsername != "" {
|
||||
@@ -100,30 +86,3 @@ func (s *Server) SetupRoutes() {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// refuseDuringMaintenance answers a request with 503 Service Unavailable,
|
||||
// a Retry-After header and a JSON error body while maintenance mode is on,
|
||||
// and passes it on otherwise. The body has the fields of the JSON errors
|
||||
// the image handlers send.
|
||||
func (s *Server) refuseDuringMaintenance(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if !s.MaintenanceMode() {
|
||||
next.ServeHTTP(w, r)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Retry-After", strconv.Itoa(MaintenanceRetryAfterSeconds))
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
|
||||
err := json.NewEncoder(w).Encode(map[string]any{
|
||||
"error": "down for maintenance, try again later",
|
||||
"status": http.StatusServiceUnavailable,
|
||||
"timestamp": time.Now().UTC().Format(time.RFC3339),
|
||||
})
|
||||
if err != nil {
|
||||
s.log.Error("json encode error", "error", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
+2
-9
@@ -5,10 +5,8 @@
|
||||
# or apk (detected in that order); assumes NOTHING is present (not git,
|
||||
# make, or go). The linter is never installed on the host: golangci-lint
|
||||
# runs only inside a container, Dockerfile.lint or the Dockerfile lint
|
||||
# stage (see script/lint). A C compiler and the CGO image libraries
|
||||
# (pkg-config, vips, libheif) are installed for the govips bindings.
|
||||
# Both Dockerfiles run this script too, so their build dependencies are
|
||||
# the ones listed here.
|
||||
# stage (see script/lint). CGO image libraries (pkg-config, vips,
|
||||
# libheif) are installed for the govips bindings.
|
||||
set -eu
|
||||
|
||||
ROOT="$(cd "$(dirname "$0")/.." && pwd -P)"
|
||||
@@ -55,11 +53,6 @@ missing() {
|
||||
|
||||
# CGO dependencies for govips (image processing)
|
||||
ensure_cgo_deps() {
|
||||
# cgo compiles with gcc on Linux; build-base and build-essential
|
||||
# also bring the C library headers.
|
||||
if missing gcc; then
|
||||
pkg_install gcc build-essential gcc build-base
|
||||
fi
|
||||
if missing pkg-config; then
|
||||
pkg_install pkg-config pkg-config pkg-config pkgconfig
|
||||
fi
|
||||
|
||||
Reference in New Issue
Block a user