Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
36c9e1408f | ||
|
|
59063c3e26 | ||
|
|
36a4ce22c3 | ||
|
|
a1fc42b80e |
@@ -123,12 +123,8 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for pixa needs:
|
||||
comma-separated; default none. Every image on a listed host is served to
|
||||
anyone without a signature, so list only hosts you control, where nobody
|
||||
else can put images.
|
||||
- `PIXA_CACHE_MAX_BYTES`: disk cache limit in bytes; `0` disables it. Set
|
||||
it: the volume is a directory on the host's filesystem, shared with the
|
||||
host and every other app. The default is 75% of the sum of that
|
||||
filesystem's free space and what the cache holds, and at least 500 MiB, so
|
||||
the cache can grow to three quarters or more of the space they all have
|
||||
left, and to all of it when that is 500 MiB or less.
|
||||
- `PIXA_CACHE_MAX_BYTES`: disk cache limit in bytes; `0` disables it;
|
||||
default 75% of (free space + what the cache holds)
|
||||
- the rest are in the table under Configuration below
|
||||
- **Health check:** the image's `HEALTHCHECK` requests
|
||||
`/.well-known/healthcheck.json`. upaas reads the container's health 60 seconds
|
||||
@@ -164,6 +160,14 @@ name.
|
||||
Multiple source paths may reference the same content blob; the database tracks
|
||||
references rather than using filesystem refcounting.
|
||||
|
||||
Source and transformed image files, and the `.meta` file beside each transformed
|
||||
image, are written under a temporary name, synced to disk, and only then renamed
|
||||
into place, so a crash leaves no file under its own name with part of its data.
|
||||
When pixa starts, and every five minutes after, it removes each cached source or
|
||||
transformed image whose file is empty, or not the size the database records for
|
||||
it, with a warning naming the file; the next request for it fetches the source
|
||||
or converts the image again.
|
||||
|
||||
pixa's target is 1-5k r/s, which has not been measured at that rate (see Load
|
||||
Test). Toward it, 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
|
||||
@@ -423,11 +427,8 @@ HMAC-SHA256(secret, "host:path:query:width:height:format:expiration:quality:fit"
|
||||
Where:
|
||||
|
||||
- `host` — source origin hostname (e.g. `cdn.example.com`)
|
||||
- `path` — source path (e.g. `/photos/cat.jpg`), with each `%` written as `%25`
|
||||
and each `:` as `%3A`, so `/photos/a:b.jpg` is signed as `/photos/a%3Ab.jpg`
|
||||
- `query` — source query string, empty string if none, with `%` and `:` written
|
||||
as in `path`; neither then holds a `:`, so path `/a:b` with no query and path
|
||||
`/a` with query `b:` are signed differently
|
||||
- `path` — source path (e.g. `/photos/cat.jpg`)
|
||||
- `query` — source query string, empty string if none
|
||||
- `width` — requested width in pixels, `0` for original
|
||||
- `height` — requested height in pixels, `0` for original
|
||||
- `format` — output format, one of those listed under Routes, with `original`
|
||||
|
||||
@@ -30,21 +30,18 @@ P2: security: per-origin rate limiting
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- 2026-10-15 "Running under upaas" in `README.md` tells the operator to set
|
||||
`PIXA_CACHE_MAX_BYTES` (closes #251): it listed the variable as optional, but
|
||||
under upaas the volume is a directory on the host's filesystem. The default,
|
||||
75% of the sum of that filesystem's free space and what the cache holds, and
|
||||
at least 500 MiB, let one pixa take three quarters or more of the space the
|
||||
host and every other app had left, and all of it when that was 500 MiB or
|
||||
less. The section now says so. The default is unchanged.
|
||||
- 2026-10-15 a `:` in the source path or query no longer moves where the path
|
||||
ends and the query begins (closes #247): the signed data and the cache key
|
||||
joined host, path and query with `:` and no escaping, so a URL signed for
|
||||
`https://h/a:b` verified for `https://h/a?b:`; the two also shared a cache
|
||||
key, and whichever was fetched first was served for both. In both, the path
|
||||
and the query now have each `%` written as `%25` and each `:` as `%3A`; a path
|
||||
and query holding neither are signed and keyed as before. Signature
|
||||
Specification in `README.md` says so.
|
||||
- 2026-10-15 a cached file left empty or partial is no longer served, or fetched
|
||||
again on every request (closes #252). Source and transformed image files, and
|
||||
each transformed image's `.meta` file, are synced to disk before they are
|
||||
renamed into place, and their directory after; the `.meta` file is written
|
||||
before the image, and a failed directory sync removes the renamed file. The
|
||||
reconciliation pass, at startup and every five minutes, removes a cached
|
||||
source or transformed image whose file is empty, or not the size the database
|
||||
records for it, with its rows and a warning naming the file, so the next
|
||||
request fetches the source or converts the image again; requests compare no
|
||||
sizes. Storing a source over a file of another size replaces that file. A
|
||||
source that fails to decode is not removed for that alone: a damaged upstream
|
||||
image would then be fetched on every request.
|
||||
- 2026-10-15 the example config allowlists no host (closes #249): it allowlisted
|
||||
`github.com` and `user-images.githubusercontent.com`, where anyone can put an
|
||||
image, so a deployment set up as Getting Started says served those images to
|
||||
@@ -53,15 +50,6 @@ P2: security: per-origin rate limiting
|
||||
commented entry under `example.com`. Getting Started and "Running under upaas"
|
||||
in `README.md` say that every image on an allowlisted host is served without a
|
||||
signature, so an allowlisted host must be one the operator controls.
|
||||
- 2026-10-15 a chunked upstream body of exactly `upstream_max_response_size`
|
||||
bytes is accepted (closes #248): once the limit was used up, the fetcher's
|
||||
next read refused the body without looking for its end. Go's HTTP client
|
||||
returns the end of a body sent with a `Content-Length` together with its last
|
||||
bytes, but the end of a chunked body can come in a later read, so such a body
|
||||
was refused after it had been downloaded whole, and the image routes
|
||||
answered 500. Once the limit is used up, the fetcher now reads one more byte:
|
||||
if the body has ended it is accepted, and if there is more it is refused with
|
||||
`ErrResponseTooLarge`, as before.
|
||||
- 2026-10-13 `fit=outside` no longer makes an output above 8192 on a side
|
||||
(closes #257): it scales the image to cover the requested size, and the side
|
||||
scaled past that size was held to no maximum, so `1x8192?fit=outside` of a
|
||||
|
||||
@@ -9,7 +9,6 @@ import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
@@ -298,139 +297,6 @@ func TestFetchMaxResponseSizeEnforced(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// fetchWithMaxResponseSize serves handler as the fake upstream and fetches
|
||||
// its /image route with MaxResponseSize set to maxResponseSize.
|
||||
func fetchWithMaxResponseSize(
|
||||
t *testing.T,
|
||||
maxResponseSize int64,
|
||||
handler http.HandlerFunc,
|
||||
) *FetchResult {
|
||||
t.Helper()
|
||||
|
||||
srv := httptest.NewServer(handler)
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
cfg := DefaultConfig()
|
||||
cfg.MaxResponseSize = maxResponseSize
|
||||
|
||||
f, _ := newServerFetcher(t, srv, cfg)
|
||||
|
||||
res := fetchImage(t, f, "/image")
|
||||
t.Cleanup(func() { _ = res.Content.Close() })
|
||||
|
||||
return res
|
||||
}
|
||||
|
||||
// TestFetchContentLengthBodyAtMaxResponseSize checks that a body sent with a
|
||||
// Content-Length is read whole at exactly MaxResponseSize bytes and refused
|
||||
// at one byte more.
|
||||
func TestFetchContentLengthBodyAtMaxResponseSize(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const maxResponseSize = 8
|
||||
|
||||
for _, tc := range []struct {
|
||||
size int
|
||||
wantErr error
|
||||
}{
|
||||
{maxResponseSize, nil},
|
||||
{maxResponseSize + 1, ErrResponseTooLarge},
|
||||
} {
|
||||
t.Run(strconv.Itoa(tc.size)+" bytes", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
body := strings.Repeat("x", tc.size)
|
||||
|
||||
res := fetchWithMaxResponseSize(t, maxResponseSize,
|
||||
func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Content-Type", contentTypeJPEG)
|
||||
w.Header().Set("Content-Length", strconv.Itoa(tc.size))
|
||||
_, _ = io.WriteString(w, body)
|
||||
})
|
||||
|
||||
if res.ContentLength != int64(tc.size) {
|
||||
t.Fatalf("ContentLength = %d, want %d", res.ContentLength, tc.size)
|
||||
}
|
||||
|
||||
data, err := io.ReadAll(res.Content)
|
||||
t.Logf("body of %d bytes: read %d, error %v", tc.size, len(data), err)
|
||||
|
||||
if !errors.Is(err, tc.wantErr) {
|
||||
t.Fatalf("read error = %v, want %v", err, tc.wantErr)
|
||||
}
|
||||
|
||||
if err == nil && string(data) != body {
|
||||
t.Errorf("read %q, want %q", data, body)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestFetchChunkedBodyAtMaxResponseSize checks that a chunked body is read
|
||||
// whole at exactly MaxResponseSize bytes and refused at one byte more. The
|
||||
// upstream sends the end of the body only once the test has read
|
||||
// MaxResponseSize bytes, so the read after them comes before the end of the
|
||||
// body has arrived, as it can from a real upstream.
|
||||
func TestFetchChunkedBodyAtMaxResponseSize(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const maxResponseSize = 8
|
||||
|
||||
for _, tc := range []struct {
|
||||
size int
|
||||
wantErr error
|
||||
}{
|
||||
{maxResponseSize, nil},
|
||||
{maxResponseSize + 1, ErrResponseTooLarge},
|
||||
} {
|
||||
t.Run(strconv.Itoa(tc.size)+" bytes", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
body := strings.Repeat("x", tc.size)
|
||||
endBody := make(chan struct{})
|
||||
|
||||
res := fetchWithMaxResponseSize(t, maxResponseSize,
|
||||
func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", contentTypeJPEG)
|
||||
_, _ = io.WriteString(w, body)
|
||||
// With no Content-Length, flushing sends the body
|
||||
// chunked; its end is sent when the handler returns.
|
||||
_ = http.NewResponseController(w).Flush()
|
||||
|
||||
select {
|
||||
case <-endBody:
|
||||
case <-r.Context().Done():
|
||||
}
|
||||
})
|
||||
|
||||
if res.ContentLength != -1 {
|
||||
t.Fatalf("ContentLength = %d, want -1 (chunked)", res.ContentLength)
|
||||
}
|
||||
|
||||
head := make([]byte, maxResponseSize)
|
||||
|
||||
_, err := io.ReadFull(res.Content, head)
|
||||
if err != nil {
|
||||
t.Fatalf("read the first %d bytes: %v", maxResponseSize, err)
|
||||
}
|
||||
|
||||
close(endBody)
|
||||
|
||||
rest, err := io.ReadAll(res.Content)
|
||||
data := string(head) + string(rest)
|
||||
t.Logf("body of %d bytes: read %d, error %v", tc.size, len(data), err)
|
||||
|
||||
if !errors.Is(err, tc.wantErr) {
|
||||
t.Fatalf("read error = %v, want %v", err, tc.wantErr)
|
||||
}
|
||||
|
||||
if err == nil && data != body {
|
||||
t.Errorf("read %q, want %q", data, body)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestFetchSemaphoreReleasedOnError(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
@@ -660,33 +660,14 @@ func dialSSRFSafe(
|
||||
return nil, fmt.Errorf("%w to %s", errConnectFailed, host)
|
||||
}
|
||||
|
||||
// limitedReader wraps a reader and refuses a body longer than remaining
|
||||
// bytes with ErrResponseTooLarge. remaining goes below 0 once the body is
|
||||
// found to be longer.
|
||||
// limitedReader wraps a reader and limits the number of bytes read.
|
||||
type limitedReader struct {
|
||||
reader io.Reader
|
||||
remaining int64
|
||||
}
|
||||
|
||||
// Read reads at most the remaining bytes. Once they are read, it reads one
|
||||
// more byte to tell the end of the body from more data: at the end it
|
||||
// returns the wrapped reader's error, io.EOF; with more data it returns
|
||||
// ErrResponseTooLarge, as does every later Read.
|
||||
func (r *limitedReader) Read(p []byte) (int, error) {
|
||||
if r.remaining < 0 {
|
||||
return 0, ErrResponseTooLarge
|
||||
}
|
||||
|
||||
if r.remaining == 0 {
|
||||
var extra [1]byte
|
||||
|
||||
n, err := r.reader.Read(extra[:])
|
||||
r.remaining -= int64(n)
|
||||
|
||||
if n == 0 {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
if r.remaining <= 0 {
|
||||
return 0, ErrResponseTooLarge
|
||||
}
|
||||
|
||||
|
||||
@@ -643,10 +643,10 @@ func (c *Cache) readHitAndMissCounts(ctx context.Context, stats *CacheStats) err
|
||||
// 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.
|
||||
// as one can when the store's .meta write failed or an eviction removed
|
||||
// the file after the read opened the variant, 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) {
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"os"
|
||||
@@ -542,10 +543,11 @@ func (c *Cache) runReconciliationPass(ctx context.Context) {
|
||||
// (or whose accounting insert failed, e.g. StoreVariant's best-effort
|
||||
// insert under transient DB contention), drops accounting rows whose
|
||||
// files are missing, removes source blob files the database does not
|
||||
// know (and rows whose files are gone), sweeps stale temp files left
|
||||
// behind by crashed writes, and last checks the total cache usage
|
||||
// against the tables. It reads the tables a page at a time. Running it
|
||||
// periodically, not just once, bounds how long such drift can
|
||||
// know (and rows whose files are gone), removes source and variant files
|
||||
// that are empty or not the size their rows record, sweeps stale temp
|
||||
// files left behind by crashed writes, and last checks the total cache
|
||||
// usage against the tables. It reads the tables a page at a time.
|
||||
// Running it periodically, not just once, bounds how long such drift can
|
||||
// accumulate unaccounted for on a long-running process to one eviction
|
||||
// interval. Once ctx is cancelled, it stops at the next file or row and
|
||||
// returns ctx's error.
|
||||
@@ -577,8 +579,9 @@ func (c *Cache) reconcileAccounting(ctx context.Context) error {
|
||||
return c.reconcileUsageTotal(ctx)
|
||||
}
|
||||
|
||||
// reconcileVariantFiles walks the variant storage directory, adopting
|
||||
// files without accounting rows and sweeping stale temp files.
|
||||
// reconcileVariantFiles walks the variant storage directory, checking
|
||||
// each variant file against its accounting row (reconcileVariantFile)
|
||||
// and sweeping stale temp files.
|
||||
func (c *Cache) reconcileVariantFiles(ctx context.Context) error {
|
||||
return filepath.WalkDir(
|
||||
c.variants.baseDir,
|
||||
@@ -603,38 +606,55 @@ func (c *Cache) reconcileVariantFiles(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
return c.adoptVariantFile(ctx, path, entry, VariantKey(name))
|
||||
return c.reconcileVariantFile(ctx, path, entry, VariantKey(name))
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
// adoptVariantFile inserts an accounting row for a variant file that
|
||||
// has none, using the file's size and modification time.
|
||||
func (c *Cache) adoptVariantFile(
|
||||
// reconcileVariantFile checks one variant file against its accounting
|
||||
// row. A file without a row is adopted into the accounting, unless it is
|
||||
// empty. A file that is empty, or not the size its row records, as a
|
||||
// crash or a restored snapshot can leave it, is removed with its row and
|
||||
// a warning naming it, so the variant is converted again.
|
||||
func (c *Cache) reconcileVariantFile(
|
||||
ctx context.Context, path string, entry fs.DirEntry, cacheKey VariantKey,
|
||||
) error {
|
||||
var rowExists int
|
||||
|
||||
err := c.db.QueryRowContext(ctx,
|
||||
`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`, string(cacheKey),
|
||||
).Scan(&rowExists)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to check variant accounting row: %w", err)
|
||||
}
|
||||
|
||||
if rowExists > 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
info, err := entry.Info()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to stat variant file: %w", err)
|
||||
}
|
||||
|
||||
var recordedSize int64
|
||||
|
||||
err = c.db.QueryRowContext(ctx,
|
||||
`SELECT size_bytes FROM variant_content WHERE cache_key = ?`, string(cacheKey),
|
||||
).Scan(&recordedSize)
|
||||
|
||||
switch {
|
||||
case errors.Is(err, sql.ErrNoRows) && info.Size() > 0:
|
||||
return c.adoptVariantFile(ctx, path, info, cacheKey)
|
||||
case err != nil && !errors.Is(err, sql.ErrNoRows):
|
||||
return fmt.Errorf("failed to check variant accounting row: %w", err)
|
||||
case info.Size() > 0 && info.Size() == recordedSize:
|
||||
return nil
|
||||
}
|
||||
|
||||
// The file is empty, or not the size its row records
|
||||
c.log.Warn("removing damaged cached variant file", "path", path,
|
||||
"file_bytes", info.Size(), "recorded_bytes", recordedSize)
|
||||
|
||||
return c.evictVariant(ctx, cacheKey)
|
||||
}
|
||||
|
||||
// adoptVariantFile inserts an accounting row for a variant file that
|
||||
// has none, using the file's size and modification time.
|
||||
func (c *Cache) adoptVariantFile(
|
||||
ctx context.Context, path string, info fs.FileInfo, cacheKey VariantKey,
|
||||
) error {
|
||||
modTime := info.ModTime().UTC().Format(sqliteTimestampLayout)
|
||||
contentType := c.variantContentTypeFromSidecar(path)
|
||||
|
||||
_, err = c.db.ExecContext(ctx, `
|
||||
_, err := c.db.ExecContext(ctx, `
|
||||
INSERT INTO variant_content
|
||||
(cache_key, size_bytes, content_type, created_at, last_accessed_at)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
@@ -753,10 +773,9 @@ func queryStringColumn[T ~string](
|
||||
return values, nil
|
||||
}
|
||||
|
||||
// reconcileSourceFiles walks the source content directory, removing
|
||||
// blob files the database does not track (they are unreachable: source
|
||||
// lookups always go through source_metadata) and sweeping stale temp
|
||||
// files.
|
||||
// reconcileSourceFiles walks the source content directory, checking
|
||||
// each blob file against its source_content row (reconcileSourceFile)
|
||||
// and sweeping stale temp files.
|
||||
func (c *Cache) reconcileSourceFiles(ctx context.Context) error {
|
||||
return filepath.WalkDir(
|
||||
c.srcContent.baseDir,
|
||||
@@ -777,31 +796,55 @@ func (c *Cache) reconcileSourceFiles(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
return c.removeUntrackedSourceFile(ctx, path, ContentHash(name))
|
||||
return c.reconcileSourceFile(ctx, path, entry, ContentHash(name))
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
// reconcileSourceFile checks one source blob file against its
|
||||
// source_content row. A file without a row is removed: it is
|
||||
// unreachable, as source lookups always go through source_metadata. A
|
||||
// file that is empty, or not the size its row records, as a crash or a
|
||||
// restored snapshot can leave it, is removed with its rows and a warning
|
||||
// naming it, so the source is fetched again.
|
||||
func (c *Cache) reconcileSourceFile(
|
||||
ctx context.Context, path string, entry fs.DirEntry, contentHash ContentHash,
|
||||
) error {
|
||||
info, err := entry.Info()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to stat source content file: %w", err)
|
||||
}
|
||||
|
||||
var recordedSize int64
|
||||
|
||||
err = c.db.QueryRowContext(ctx,
|
||||
`SELECT size_bytes FROM source_content WHERE content_hash = ?`,
|
||||
string(contentHash),
|
||||
).Scan(&recordedSize)
|
||||
|
||||
switch {
|
||||
case errors.Is(err, sql.ErrNoRows):
|
||||
return c.removeUntrackedSourceFile(ctx, path, contentHash)
|
||||
case err != nil:
|
||||
return fmt.Errorf("failed to check source content row: %w", err)
|
||||
case info.Size() > 0 && info.Size() == recordedSize:
|
||||
return nil
|
||||
}
|
||||
|
||||
// The file is empty, or not the size its row records
|
||||
c.log.Warn("removing damaged cached source file", "path", path,
|
||||
"file_bytes", info.Size(), "recorded_bytes", recordedSize)
|
||||
|
||||
return c.evictSourceBlob(ctx, contentHash)
|
||||
}
|
||||
|
||||
// removeUntrackedSourceFile deletes a source blob file that has no
|
||||
// source_content row. Any source_metadata rows referencing the hash
|
||||
// are removed first so no row ever points at a deleted file.
|
||||
func (c *Cache) removeUntrackedSourceFile(
|
||||
ctx context.Context, path string, contentHash ContentHash,
|
||||
) error {
|
||||
var rowExists int
|
||||
|
||||
err := c.db.QueryRowContext(ctx,
|
||||
`SELECT COUNT(*) FROM source_content WHERE content_hash = ?`, string(contentHash),
|
||||
).Scan(&rowExists)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to check source content row: %w", err)
|
||||
}
|
||||
|
||||
if rowExists > 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
_, err = c.db.ExecContext(ctx,
|
||||
_, err := c.db.ExecContext(ctx,
|
||||
`DELETE FROM source_metadata WHERE content_hash = ?`, string(contentHash))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to delete metadata rows for untracked blob: %w", err)
|
||||
|
||||
@@ -0,0 +1,359 @@
|
||||
package imgcache
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"io"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// getImage calls Get for req and returns the image it answered with and
|
||||
// its cache status.
|
||||
func getImage(t *testing.T, svc *Service, req *ImageRequest) ([]byte, CacheStatus) {
|
||||
t.Helper()
|
||||
|
||||
resp, err := svc.Get(t.Context(), req)
|
||||
if err != nil {
|
||||
t.Fatalf("Get() of quality %d error = %v", req.Quality, err)
|
||||
}
|
||||
|
||||
defer func() { _ = resp.Content.Close() }()
|
||||
|
||||
body, err := io.ReadAll(resp.Content)
|
||||
if err != nil {
|
||||
t.Fatalf("reading the image of quality %d: %v", req.Quality, err)
|
||||
}
|
||||
|
||||
return body, resp.CacheStatus
|
||||
}
|
||||
|
||||
// fileSize returns the size of the file at path.
|
||||
func fileSize(t *testing.T, path string) int64 {
|
||||
t.Helper()
|
||||
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
t.Fatalf("stat %s: %v", path, err)
|
||||
}
|
||||
|
||||
return info.Size()
|
||||
}
|
||||
|
||||
// truncateFile cuts the file at path down to size bytes.
|
||||
func truncateFile(t *testing.T, path string, size int64) {
|
||||
t.Helper()
|
||||
|
||||
t.Logf("cutting %s from %d to %d bytes",
|
||||
filepath.Base(path), fileSize(t, path), size)
|
||||
|
||||
err := os.Truncate(path, size)
|
||||
if err != nil {
|
||||
t.Fatalf("truncating %s: %v", path, err)
|
||||
}
|
||||
}
|
||||
|
||||
// cachedSourcePath returns the path of the cached source file req uses.
|
||||
func cachedSourcePath(t *testing.T, svc *Service, req *ImageRequest) string {
|
||||
t.Helper()
|
||||
|
||||
contentHash, _, err := svc.cache.LookupSource(t.Context(), req)
|
||||
if err != nil || contentHash == "" {
|
||||
t.Fatalf("LookupSource() = %q, %v; want the cached source", contentHash, err)
|
||||
}
|
||||
|
||||
return svc.cache.srcContent.hashToPath(contentHash)
|
||||
}
|
||||
|
||||
// reconcileAsAtStartup runs the cache's reconciliation pass, as pixa does
|
||||
// when it starts, and returns what the pass logged.
|
||||
func reconcileAsAtStartup(t *testing.T, cache *Cache) string {
|
||||
t.Helper()
|
||||
|
||||
var logBuf bytes.Buffer
|
||||
|
||||
cache.log = slog.New(slog.NewJSONHandler(&logBuf, nil))
|
||||
|
||||
err := cache.reconcileAccounting(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("reconcileAccounting() error = %v", err)
|
||||
}
|
||||
|
||||
t.Logf("reconciliation log: %s", logBuf.String())
|
||||
|
||||
return logBuf.String()
|
||||
}
|
||||
|
||||
// assertWarningNames checks that logs hold a warning naming the file at
|
||||
// path.
|
||||
func assertWarningNames(t *testing.T, logs, path string) {
|
||||
t.Helper()
|
||||
|
||||
for line := range strings.Lines(logs) {
|
||||
if strings.Contains(line, `"level":"WARN"`) &&
|
||||
strings.Contains(line, `"path":"`+path+`"`) {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
t.Errorf("reconciliation logged no warning naming %s", path)
|
||||
}
|
||||
|
||||
// TestService_Get_FetchesAgainForEmptyCachedSource empties the cached source
|
||||
// file, as a crash or a restored snapshot could, and runs the reconciliation
|
||||
// pass, which must remove it with a warning naming it. It then asks for two
|
||||
// new sizes of the source: the first must fetch it again and cache it whole,
|
||||
// so that the second is made from the cache without a third fetch.
|
||||
func TestService_Get_FetchesAgainForEmptyCachedSource(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
svc, fixtures := SetupTestService(t)
|
||||
|
||||
getImage(t, svc, photoVariant(fixtures, 85, FitCover))
|
||||
|
||||
path := cachedSourcePath(t, svc, photoVariant(fixtures, 85, FitCover))
|
||||
wholeSize := fileSize(t, path)
|
||||
|
||||
truncateFile(t, path, 0)
|
||||
assertWarningNames(t, reconcileAsAtStartup(t, svc.cache), path)
|
||||
|
||||
getImage(t, svc, photoVariant(fixtures, 40, FitCover))
|
||||
getImage(t, svc, photoVariant(fixtures, 30, FitCover))
|
||||
|
||||
fetches := readCacheStatsCounters(t, svc.cache).upstreamFetchCount
|
||||
t.Logf("upstream fetches = %d", fetches)
|
||||
|
||||
if fetches != 2 {
|
||||
t.Errorf("upstream fetches = %d, want 2: the empty cached source "+
|
||||
"was kept and fetched again on every new size", fetches)
|
||||
}
|
||||
|
||||
size := fileSize(t, path)
|
||||
if size != wholeSize {
|
||||
t.Errorf("cached source is %d bytes, want %d", size, wholeSize)
|
||||
}
|
||||
}
|
||||
|
||||
// TestService_Get_FetchesAgainForPartialCachedSource cuts the cached source
|
||||
// file to half its size, as a crash or a restored snapshot could, and runs
|
||||
// the reconciliation pass, which must remove it with a warning naming it.
|
||||
// It then asks for two new sizes of the source: the first must fetch it
|
||||
// again rather than fail to decode the partial file, and the second is
|
||||
// made from the cache.
|
||||
func TestService_Get_FetchesAgainForPartialCachedSource(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
svc, fixtures := SetupTestService(t)
|
||||
|
||||
getImage(t, svc, photoVariant(fixtures, 85, FitCover))
|
||||
|
||||
path := cachedSourcePath(t, svc, photoVariant(fixtures, 85, FitCover))
|
||||
wholeSize := fileSize(t, path)
|
||||
|
||||
truncateFile(t, path, wholeSize/2)
|
||||
assertWarningNames(t, reconcileAsAtStartup(t, svc.cache), path)
|
||||
|
||||
getImage(t, svc, photoVariant(fixtures, 40, FitCover))
|
||||
getImage(t, svc, photoVariant(fixtures, 30, FitCover))
|
||||
|
||||
fetches := readCacheStatsCounters(t, svc.cache).upstreamFetchCount
|
||||
t.Logf("upstream fetches = %d", fetches)
|
||||
|
||||
if fetches != 2 {
|
||||
t.Errorf("upstream fetches = %d, want 2", fetches)
|
||||
}
|
||||
|
||||
size := fileSize(t, path)
|
||||
if size != wholeSize {
|
||||
t.Errorf("cached source is %d bytes, want %d", size, wholeSize)
|
||||
}
|
||||
}
|
||||
|
||||
// assertConvertedAgain caches a variant of the photo, cuts its file to size
|
||||
// bytes and runs the reconciliation pass, which must remove it with a
|
||||
// warning naming it. It then checks that Get converts the image again
|
||||
// instead of serving the file, and that the next Get serves the whole
|
||||
// variant from the cache.
|
||||
func assertConvertedAgain(t *testing.T, size int64) {
|
||||
t.Helper()
|
||||
|
||||
svc, fixtures := SetupTestService(t)
|
||||
req := photoVariant(fixtures, 85, FitCover)
|
||||
|
||||
want, _ := getImage(t, svc, req)
|
||||
|
||||
path := svc.cache.variants.keyToPath(CacheKey(req))
|
||||
|
||||
truncateFile(t, path, size)
|
||||
assertWarningNames(t, reconcileAsAtStartup(t, svc.cache), path)
|
||||
|
||||
got, status := getImage(t, svc, req)
|
||||
t.Logf("after the cut: %d bytes, %s", len(got), status)
|
||||
|
||||
if status != CacheMiss || !bytes.Equal(got, want) {
|
||||
t.Errorf("after the cut: Get() = %d bytes, %s; want the %d bytes "+
|
||||
"converted again, %s", len(got), status, len(want), CacheMiss)
|
||||
}
|
||||
|
||||
got, status = getImage(t, svc, req)
|
||||
t.Logf("next request: %d bytes, %s", len(got), status)
|
||||
|
||||
if status != CacheHit || !bytes.Equal(got, want) {
|
||||
t.Errorf("next request: Get() = %d bytes, %s; want the %d bytes, %s",
|
||||
len(got), status, len(want), CacheHit)
|
||||
}
|
||||
}
|
||||
|
||||
// TestService_Get_ConvertsAgainForEmptyCachedVariant empties a cached
|
||||
// variant file, as a crash or a restored snapshot could, and checks that it
|
||||
// is converted again rather than served empty.
|
||||
func TestService_Get_ConvertsAgainForEmptyCachedVariant(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
assertConvertedAgain(t, 0)
|
||||
}
|
||||
|
||||
// TestService_Get_ConvertsAgainForPartialCachedVariant cuts a cached
|
||||
// variant file to ten bytes, as a crash or a restored snapshot could, and
|
||||
// checks that it is converted again rather than served cut short.
|
||||
func TestService_Get_ConvertsAgainForPartialCachedVariant(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
assertConvertedAgain(t, 10)
|
||||
}
|
||||
|
||||
// TestReconciliationRemovesEmptyVariantFileWithoutRow checks that the
|
||||
// reconciliation pass removes an empty variant file that has no
|
||||
// accounting row, with a warning naming it, instead of adding a row for
|
||||
// it that would have it served empty.
|
||||
func TestReconciliationRemovesEmptyVariantFileWithoutRow(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
|
||||
_, err := cache.variants.Store(testVariantKeyOne, bytes.NewReader(nil),
|
||||
testContentTypeWebP)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to store an empty variant file: %v", err)
|
||||
}
|
||||
|
||||
logs := reconcileAsAtStartup(t, cache)
|
||||
|
||||
assertWarningNames(t, logs, cache.variants.keyToPath(testVariantKeyOne))
|
||||
|
||||
if cache.variants.Exists(testVariantKeyOne) {
|
||||
t.Error("the empty variant file is still on disk")
|
||||
}
|
||||
|
||||
rows := countRows(t, cache,
|
||||
`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`,
|
||||
string(testVariantKeyOne))
|
||||
if rows != 0 {
|
||||
t.Errorf("the empty variant file got %d accounting rows, want 0", rows)
|
||||
}
|
||||
}
|
||||
|
||||
// TestVariantStorage_StoreReplacesMetaFile stores a variant over one
|
||||
// already stored and checks that Store puts a new .meta file in place of
|
||||
// the old one, as it does the image, rather than writing into the old
|
||||
// file, which a crash could leave empty. The old .meta file is given a
|
||||
// second name before the store, so that it can be told apart from the new
|
||||
// one after it.
|
||||
func TestVariantStorage_StoreReplacesMetaFile(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
storage, key := storeTestVariant(t)
|
||||
metaPath := storage.keyToPath(key) + variantMetaSuffix
|
||||
oldMetaPath := metaPath + ".old"
|
||||
|
||||
err := os.Link(metaPath, oldMetaPath)
|
||||
if err != nil {
|
||||
t.Fatalf("giving the .meta file a second name: %v", err)
|
||||
}
|
||||
|
||||
_, err = storage.Store(key, bytes.NewReader([]byte("new variant data")),
|
||||
testContentTypeJPEG)
|
||||
if err != nil {
|
||||
t.Fatalf("Store() error = %v", err)
|
||||
}
|
||||
|
||||
oldMeta, err := os.Stat(oldMetaPath)
|
||||
if err != nil {
|
||||
t.Fatalf("stat of the old .meta file: %v", err)
|
||||
}
|
||||
|
||||
newMeta, err := os.Stat(metaPath)
|
||||
if err != nil {
|
||||
t.Fatalf("stat of the new .meta file: %v", err)
|
||||
}
|
||||
|
||||
if os.SameFile(oldMeta, newMeta) {
|
||||
t.Error("Store() wrote into the old .meta file instead of putting " +
|
||||
"a new one in its place")
|
||||
}
|
||||
|
||||
reader, _, contentType, err := storage.LoadWithMeta(key)
|
||||
if err != nil {
|
||||
t.Fatalf("LoadWithMeta() error = %v", err)
|
||||
}
|
||||
|
||||
_ = reader.Close()
|
||||
|
||||
t.Logf("content type after the store: %s", contentType)
|
||||
|
||||
if contentType != testContentTypeJPEG {
|
||||
t.Errorf("content type = %q, want %q", contentType, testContentTypeJPEG)
|
||||
}
|
||||
}
|
||||
|
||||
// TestContentStorage_StoreHashedReplacesFileOfAnotherSize checks that a
|
||||
// store over a file of another size at the content's path, such as a
|
||||
// partial file a crash left, replaces it with the content.
|
||||
func TestContentStorage_StoreHashedReplacesFileOfAnotherSize(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
storage, err := NewContentStorage(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatalf("NewContentStorage() error = %v", err)
|
||||
}
|
||||
|
||||
content := []byte("the whole source image")
|
||||
sum := sha256.Sum256(content)
|
||||
hash := ContentHash(hex.EncodeToString(sum[:]))
|
||||
path := storage.hashToPath(hash)
|
||||
|
||||
err = os.MkdirAll(filepath.Dir(path), StorageDirPerm)
|
||||
if err != nil {
|
||||
t.Fatalf("creating the content's directory: %v", err)
|
||||
}
|
||||
|
||||
err = os.WriteFile(path, content[:5], StorageFilePerm)
|
||||
if err != nil {
|
||||
t.Fatalf("writing a partial file: %v", err)
|
||||
}
|
||||
|
||||
_, err = storage.StoreHashed(hash, content)
|
||||
if err != nil {
|
||||
t.Fatalf("StoreHashed() error = %v", err)
|
||||
}
|
||||
|
||||
stored, err := storage.Load(hash)
|
||||
if err != nil {
|
||||
t.Fatalf("Load() error = %v", err)
|
||||
}
|
||||
|
||||
defer func() { _ = stored.Close() }()
|
||||
|
||||
got, err := io.ReadAll(stored)
|
||||
if err != nil {
|
||||
t.Fatalf("reading the stored content: %v", err)
|
||||
}
|
||||
|
||||
if !bytes.Equal(got, content) {
|
||||
t.Errorf("stored file = %q, want %q", got, content)
|
||||
}
|
||||
}
|
||||
@@ -10,7 +10,6 @@ import (
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -80,8 +79,8 @@ func (s *ContentStorage) Store(r io.Reader) (ContentHash, int64, error) {
|
||||
// from hash, without recomputing it. Callers that already know the
|
||||
// hash before writing (e.g. because they must hold a hash-keyed lock
|
||||
// across the whole store operation) use this instead of Store. Like
|
||||
// Store, it is idempotent: content already on disk at that path is
|
||||
// left untouched.
|
||||
// Store, it is idempotent: a file of the content's size already at that
|
||||
// path is left untouched, and a file of another size is replaced.
|
||||
func (s *ContentStorage) StoreHashed(
|
||||
hash ContentHash, data []byte,
|
||||
) (int64, error) {
|
||||
@@ -152,28 +151,35 @@ func (s *ContentStorage) Exists(hash ContentHash) bool {
|
||||
return err == nil
|
||||
}
|
||||
|
||||
// writeIfAbsent writes data to the path derived from hash, unless
|
||||
// content already exists there, via a temp-file-plus-rename so
|
||||
// concurrent readers never observe a partial file.
|
||||
// writeIfAbsent writes data with writeFileSynced to the path derived
|
||||
// from hash, unless a file of the data's size is already there.
|
||||
func (s *ContentStorage) writeIfAbsent(hash ContentHash, data []byte) error {
|
||||
// Build path: <basedir>/<ab>/<cd>/<hash>
|
||||
path := s.hashToPath(hash)
|
||||
|
||||
// Check if already exists
|
||||
_, statErr := os.Stat(path)
|
||||
if statErr == nil {
|
||||
// A file of another size there is damaged, and is replaced
|
||||
info, statErr := os.Stat(path)
|
||||
if statErr == nil && info.Size() == int64(len(data)) {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Create directory structure
|
||||
dir := filepath.Dir(path)
|
||||
|
||||
err := os.MkdirAll(dir, StorageDirPerm)
|
||||
err := os.MkdirAll(filepath.Dir(path), StorageDirPerm)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create directory: %w", err)
|
||||
}
|
||||
|
||||
// Write to temp file first, then rename for atomicity
|
||||
return writeFileSynced(path, data)
|
||||
}
|
||||
|
||||
// writeFileSynced writes data to path through a temp file in the same
|
||||
// directory: the temp file is synced to disk, renamed to path, and the
|
||||
// directory synced after, so that neither a concurrent reader nor a
|
||||
// crash finds a partial file under path. When the directory sync fails,
|
||||
// the file is removed from path, so that a failed write leaves no file
|
||||
// there.
|
||||
func writeFileSynced(path string, data []byte) error {
|
||||
dir := filepath.Dir(path)
|
||||
|
||||
tmpFile, err := os.CreateTemp(dir, ".tmp-*")
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create temp file: %w", err)
|
||||
@@ -181,18 +187,20 @@ func (s *ContentStorage) writeIfAbsent(hash ContentHash, data []byte) error {
|
||||
|
||||
tmpPath := tmpFile.Name()
|
||||
|
||||
// Each failure path below unlinks the temp file explicitly. This
|
||||
// replaces a deferred cleanup that read a named result, which the
|
||||
// canonical config does not permit; the set of paths that remove
|
||||
// tmpPath, and the order relative to Close, is unchanged. This
|
||||
// mirrors how MetadataStorage.Store and VariantStorage.Store below
|
||||
// already express the same cleanup.
|
||||
_, err = tmpFile.Write(data)
|
||||
if err != nil {
|
||||
_ = tmpFile.Close()
|
||||
_ = os.Remove(tmpPath)
|
||||
|
||||
return fmt.Errorf("failed to write content: %w", err)
|
||||
return fmt.Errorf("failed to write temp file: %w", err)
|
||||
}
|
||||
|
||||
err = tmpFile.Sync()
|
||||
if err != nil {
|
||||
_ = tmpFile.Close()
|
||||
_ = os.Remove(tmpPath)
|
||||
|
||||
return fmt.Errorf("failed to sync temp file: %w", err)
|
||||
}
|
||||
|
||||
err = tmpFile.Close()
|
||||
@@ -202,7 +210,6 @@ func (s *ContentStorage) writeIfAbsent(hash ContentHash, data []byte) error {
|
||||
return fmt.Errorf("failed to close temp file: %w", err)
|
||||
}
|
||||
|
||||
// Atomic rename
|
||||
err = os.Rename(filepath.Clean(tmpPath), filepath.Clean(path))
|
||||
if err != nil {
|
||||
_ = os.Remove(tmpPath)
|
||||
@@ -210,6 +217,31 @@ func (s *ContentStorage) writeIfAbsent(hash ContentHash, data []byte) error {
|
||||
return fmt.Errorf("failed to rename temp file: %w", err)
|
||||
}
|
||||
|
||||
err = syncDirectory(dir)
|
||||
if err != nil {
|
||||
_ = os.Remove(path)
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// syncDirectory syncs the directory dir to disk, so that a file just
|
||||
// renamed into it is still there after a crash.
|
||||
func syncDirectory(dir string) error {
|
||||
d, err := os.Open(filepath.Clean(dir))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to open directory: %w", err)
|
||||
}
|
||||
|
||||
err = d.Sync()
|
||||
_ = d.Close()
|
||||
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to sync directory: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -372,13 +404,11 @@ func HashPath(path string) PathHash {
|
||||
|
||||
// CacheKey generates a unique key for a request variant.
|
||||
// Format: sha256(host:path:query:width:height:format:quality:fit_mode)
|
||||
// The path and query go through escapePercentAndColon, so that a ":" in
|
||||
// either cannot be taken for the one between them.
|
||||
func CacheKey(req *ImageRequest) VariantKey {
|
||||
data := fmt.Sprintf("%s:%s:%s:%d:%d:%s:%d:%s",
|
||||
req.SourceHost,
|
||||
escapePercentAndColon(req.SourcePath),
|
||||
escapePercentAndColon(req.SourceQuery),
|
||||
req.SourcePath,
|
||||
req.SourceQuery,
|
||||
req.Size.Width,
|
||||
req.Size.Height,
|
||||
req.Format,
|
||||
@@ -390,14 +420,6 @@ func CacheKey(req *ImageRequest) VariantKey {
|
||||
return VariantKey(hex.EncodeToString(h[:]))
|
||||
}
|
||||
|
||||
// escapePercentAndColon returns s with each "%" written as "%25" and each
|
||||
// ":" as "%3A", so that the result holds no ":".
|
||||
func escapePercentAndColon(s string) string {
|
||||
s = strings.ReplaceAll(s, "%", "%25")
|
||||
|
||||
return strings.ReplaceAll(s, ":", "%3A")
|
||||
}
|
||||
|
||||
// VariantStorage handles key-based file storage for processed image variants.
|
||||
// Files are stored at: <basedir>/<ab>/<cd>/<cache_key>
|
||||
// Metadata is stored at: <basedir>/<ab>/<cd>/<cache_key>.meta
|
||||
@@ -427,7 +449,9 @@ func NewVariantStorage(baseDir string, log *slog.Logger) (*VariantStorage, error
|
||||
return &VariantStorage{baseDir: baseDir, log: log}, nil
|
||||
}
|
||||
|
||||
// Store writes content and metadata to storage at the given key.
|
||||
// Store writes content and metadata to storage at the given key, each
|
||||
// with writeFileSynced. The .meta file is written first, so that a crash
|
||||
// cannot leave the content in place without it.
|
||||
func (s *VariantStorage) Store(
|
||||
key VariantKey, r io.Reader, contentType string,
|
||||
) (int64, error) {
|
||||
@@ -440,46 +464,11 @@ func (s *VariantStorage) Store(
|
||||
path := s.keyToPath(key)
|
||||
metaPath := path + ".meta"
|
||||
|
||||
// Create directory structure
|
||||
dir := filepath.Dir(path)
|
||||
|
||||
err = os.MkdirAll(dir, StorageDirPerm)
|
||||
err = os.MkdirAll(filepath.Dir(path), StorageDirPerm)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("failed to create directory: %w", err)
|
||||
}
|
||||
|
||||
// Write content to temp file first, then rename for atomicity
|
||||
tmpFile, err := os.CreateTemp(dir, ".tmp-*")
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("failed to create temp file: %w", err)
|
||||
}
|
||||
|
||||
tmpPath := tmpFile.Name()
|
||||
|
||||
_, err = tmpFile.Write(data)
|
||||
if err != nil {
|
||||
_ = tmpFile.Close()
|
||||
_ = os.Remove(tmpPath)
|
||||
|
||||
return 0, fmt.Errorf("failed to write content: %w", err)
|
||||
}
|
||||
|
||||
err = tmpFile.Close()
|
||||
if err != nil {
|
||||
_ = os.Remove(tmpPath)
|
||||
|
||||
return 0, fmt.Errorf("failed to close temp file: %w", err)
|
||||
}
|
||||
|
||||
// Atomic rename content
|
||||
err = os.Rename(filepath.Clean(tmpPath), filepath.Clean(path))
|
||||
if err != nil {
|
||||
_ = os.Remove(tmpPath)
|
||||
|
||||
return 0, fmt.Errorf("failed to rename temp file: %w", err)
|
||||
}
|
||||
|
||||
// Write metadata
|
||||
meta := VariantMeta{
|
||||
ContentType: contentType,
|
||||
Size: size,
|
||||
@@ -491,13 +480,19 @@ func (s *VariantStorage) Store(
|
||||
return 0, fmt.Errorf("failed to marshal metadata: %w", err)
|
||||
}
|
||||
|
||||
// Metadata write failure is non-fatal; content is already stored.
|
||||
err = os.WriteFile(metaPath, metaData, StorageFilePerm)
|
||||
// Metadata write failure is non-fatal: the content is then served as
|
||||
// application/octet-stream.
|
||||
err = writeFileSynced(metaPath, metaData)
|
||||
if err != nil {
|
||||
s.log.Warn("failed to write variant metadata sidecar",
|
||||
"path", metaPath, "error", err)
|
||||
}
|
||||
|
||||
err = writeFileSynced(path, data)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
return size, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -407,62 +407,6 @@ func TestCacheKey(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestCacheKey_ColonOrPercentInPathOrQuery checks that two different source
|
||||
// URLs whose path or query holds ":" or "%" get different keys.
|
||||
func TestCacheKey_ColonOrPercentInPathOrQuery(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
path1, query1 string
|
||||
path2, query2 string
|
||||
}{
|
||||
{
|
||||
name: "colon moved from path to query",
|
||||
path1: "/a:b", query1: "",
|
||||
path2: "/a", query2: "b:",
|
||||
},
|
||||
{
|
||||
name: "colon in path and its percent-encoding",
|
||||
path1: "/a:b", query1: "",
|
||||
path2: "/a%3Ab", query2: "",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range cases {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
key1 := CacheKey(&ImageRequest{
|
||||
SourceHost: testHostCDN,
|
||||
SourcePath: tt.path1,
|
||||
SourceQuery: tt.query1,
|
||||
Size: Size{Width: 800, Height: 600},
|
||||
Format: FormatWebP,
|
||||
Quality: 85,
|
||||
FitMode: FitCover,
|
||||
})
|
||||
key2 := CacheKey(&ImageRequest{
|
||||
SourceHost: testHostCDN,
|
||||
SourcePath: tt.path2,
|
||||
SourceQuery: tt.query2,
|
||||
Size: Size{Width: 800, Height: 600},
|
||||
Format: FormatWebP,
|
||||
Quality: 85,
|
||||
FitMode: FitCover,
|
||||
})
|
||||
|
||||
t.Logf("CacheKey() of path %q query %q = %s", tt.path1, tt.query1, key1)
|
||||
t.Logf("CacheKey() of path %q query %q = %s", tt.path2, tt.query2, key2)
|
||||
|
||||
if key1 == key2 {
|
||||
t.Errorf("CacheKey() gave path %q query %q and path %q query %q "+
|
||||
"the same key", tt.path1, tt.query1, tt.path2, tt.query2)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestVariantStorage_StoreLogsFailedMetaWrite verifies that a .meta write
|
||||
// that fails is logged at warn and the store still succeeds.
|
||||
func TestVariantStorage_StoreLogsFailedMetaWrite(t *testing.T) {
|
||||
|
||||
@@ -1,113 +0,0 @@
|
||||
package signature_test
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/signature"
|
||||
)
|
||||
|
||||
// TestSigner_Verify_ColonCannotMovePathQueryBoundary checks that a URL
|
||||
// signed for path "/a:b" with no query does not verify for path "/a" with
|
||||
// query "b:", a different resource on the same host.
|
||||
func TestSigner_Verify_ColonCannotMovePathQueryBoundary(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
signer := signature.New(goldenSigningKey)
|
||||
|
||||
signed := &signature.Request{
|
||||
SourceHost: testHost,
|
||||
SourcePath: "/a:b",
|
||||
Width: 800,
|
||||
Height: 600,
|
||||
Format: testFormatWebP,
|
||||
Quality: 85,
|
||||
FitMode: testFitCover,
|
||||
Expires: time.Now().Add(1 * time.Hour),
|
||||
}
|
||||
signed.Signature = signer.Sign(signed)
|
||||
|
||||
replayed := *signed
|
||||
replayed.SourcePath = "/a"
|
||||
replayed.SourceQuery = "b:"
|
||||
|
||||
t.Logf("signature for path %q query %q: %s",
|
||||
signed.SourcePath, signed.SourceQuery, signed.Signature)
|
||||
t.Logf("signature for path %q query %q: %s",
|
||||
replayed.SourcePath, replayed.SourceQuery, signer.Sign(&replayed))
|
||||
|
||||
err := signer.Verify(&replayed)
|
||||
if !errors.Is(err, signature.ErrInvalid) {
|
||||
t.Errorf("Verify() of path %q query %q with the signature for "+
|
||||
"path %q = %v, want %v",
|
||||
replayed.SourcePath, replayed.SourceQuery, signed.SourcePath,
|
||||
err, signature.ErrInvalid)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSigner_EscapedGoldenVectors pins the signature of requests whose path
|
||||
// or query holds ":" or "%", which are signed with "%" written as "%25" and
|
||||
// ":" as "%3A". The expected values were computed once, outside pixa, from
|
||||
// the signed data in each comment.
|
||||
func TestSigner_EscapedGoldenVectors(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
signer := signature.New(goldenSigningKey)
|
||||
|
||||
vectors := []struct {
|
||||
name string
|
||||
path string
|
||||
query string
|
||||
wantSignature string
|
||||
}{
|
||||
{
|
||||
name: "colon in path",
|
||||
path: "/photos/a:b.jpg",
|
||||
// Signed data:
|
||||
// "cdn.example.com:/photos/a%3Ab.jpg::800:600:webp:1704067200:85:cover"
|
||||
wantSignature: "WnEpZ7bQp1Zvw8WDatlj2vu-4wYKWE3BFBizV8m__6k=",
|
||||
},
|
||||
{
|
||||
name: "colon in query",
|
||||
path: testPath,
|
||||
query: "t=1:2",
|
||||
// Signed data:
|
||||
// "cdn.example.com:/photos/cat.jpg:t=1%3A2:800:600:webp:1704067200:85:cover"
|
||||
wantSignature: "85kRoN8hhr4cCUkEUjNvUsvVYKjYYfWdeffx5TjJjqA=",
|
||||
},
|
||||
{
|
||||
name: "percent in path and query",
|
||||
path: "/photos/a%20b.jpg",
|
||||
query: "q=50%",
|
||||
// Signed data:
|
||||
// "cdn.example.com:/photos/a%2520b.jpg:q=50%25:800:600:webp:1704067200:85:cover"
|
||||
wantSignature: "MjDiiRdgXMrsjqPMsENtD4aG1hp1e-a6zy67vKTX4CQ=",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range vectors {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
req := &signature.Request{
|
||||
SourceHost: testHost,
|
||||
SourcePath: tt.path,
|
||||
SourceQuery: tt.query,
|
||||
Width: 800,
|
||||
Height: 600,
|
||||
Format: testFormatWebP,
|
||||
Quality: 85,
|
||||
FitMode: testFitCover,
|
||||
Expires: time.Unix(goldenExpiresUnix, 0),
|
||||
}
|
||||
|
||||
gotSignature := signer.Sign(req)
|
||||
t.Logf("Sign() of path %q query %q = %s", tt.path, tt.query, gotSignature)
|
||||
|
||||
if gotSignature != tt.wantSignature {
|
||||
t.Errorf("Sign() = %q, want %q", gotSignature, tt.wantSignature)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -10,7 +10,6 @@ import (
|
||||
"fmt"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -155,17 +154,15 @@ func (s *Signer) GenerateSignedURL(
|
||||
|
||||
// buildSignatureData creates the string to be signed.
|
||||
// Format: "host:path:query:width:height:format:expiration:quality:fit"
|
||||
// The path and query go through escapePercentAndColon, so that a ":" in
|
||||
// either cannot be taken for the one between them; every other component
|
||||
// is used verbatim. No normalization, suffix matching, or wildcard
|
||||
// expansion is performed. Quality and fit are the effective transform
|
||||
// values, so replaying a signed URL with a different quality or fit mode
|
||||
// fails verification.
|
||||
// All components are used verbatim (exact match). No normalization,
|
||||
// suffix matching, or wildcard expansion is performed. Quality and fit
|
||||
// are the effective transform values, so replaying a signed URL with a
|
||||
// different quality or fit mode fails verification.
|
||||
func (s *Signer) buildSignatureData(req *Request) string {
|
||||
return fmt.Sprintf("%s:%s:%s:%d:%d:%s:%d:%d:%s",
|
||||
req.SourceHost,
|
||||
escapePercentAndColon(req.SourcePath),
|
||||
escapePercentAndColon(req.SourceQuery),
|
||||
req.SourcePath,
|
||||
req.SourceQuery,
|
||||
req.Width,
|
||||
req.Height,
|
||||
req.Format,
|
||||
@@ -175,14 +172,6 @@ func (s *Signer) buildSignatureData(req *Request) string {
|
||||
)
|
||||
}
|
||||
|
||||
// escapePercentAndColon returns s with each "%" written as "%25" and each
|
||||
// ":" as "%3A", so that the result holds no ":".
|
||||
func escapePercentAndColon(s string) string {
|
||||
s = strings.ReplaceAll(s, "%", "%25")
|
||||
|
||||
return strings.ReplaceAll(s, ":", "%3A")
|
||||
}
|
||||
|
||||
// ParseParams extracts signature and expiration from query parameters.
|
||||
func ParseParams(sig, expStr string) (string, time.Time, error) {
|
||||
if expStr == "" {
|
||||
|
||||
Reference in New Issue
Block a user