1 Commits

Author SHA1 Message Date
23506df609 chore: update golangci-lint to v2.12.2 with canonical config
All checks were successful
check / check (push) Successful in 2m3s
Replace .golangci.yml with the canonical v2-schema config
(default: all minus six disabled linters, lll 88, tests included)
and bump every golangci-lint pin to v2.12.2:

- Dockerfile: golangci/golangci-lint:v2.12.2-alpine (hash-pinned)
- script/bootstrap: GOLANGCI_LINT_VERSION 2.12.2 with new
  linux-amd64/arm64 release-archive sha256 pins

Fix all 747 findings the stricter config surfaces, with no behavior
changes: t.Parallel() throughout the test suite, static sentinel
errors and errors.Is comparisons, checked error returns, context
propagation (contextcheck/noctx), 88-column wrapping, extracted
constants and helpers for goconst/dupl/funlen/cyclop, exhaustive
switch cases replicating existing defaults, and white-box test files
renamed to *_internal_test.go for testpackage. Three
nolint:tagliatelle directives preserve the existing snake_case JSON
wire and on-disk metadata formats.
2026-08-07 17:10:27 +00:00
24 changed files with 147 additions and 4001 deletions

View File

@@ -115,9 +115,6 @@ Configured via YAML file (`--config`). Key settings:
- `upstream_max_response_size` — max origin response size - `upstream_max_response_size` — max origin response size
- `downstream_timeout` — client response timeout - `downstream_timeout` — client response timeout
- `signing_key` — HMAC secret for URL signatures - `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
the filesystem containing `<state_dir>/cache/` (minimum 500 MiB)
See `config.example.yml` for all options with defaults. See `config.example.yml` for all options with defaults.

58
TODO.md
View File

@@ -12,14 +12,15 @@
pre-1.0. No git tags exist. Recent work extracted the internal/magic, pre-1.0. No git tags exist. Recent work extracted the internal/magic,
internal/allowlist, internal/httpfetcher, and internal/signature internal/allowlist, internal/httpfetcher, and internal/signature
packages. The gosec findings from the 2026-07-06 survey are resolved packages. The gosec findings from the 2026-07-06 survey are resolved:
and `make check` is green on main. The disk cache is now size-bounded the last two open findings (G124, session cookie attributes in
with LRU eviction (`cache_max_bytes`), closing the unbounded disk internal/session) are fixed as of this change, so `make check` is green
growth DoS vector. on main.
# Next Step # Next Step
P1: implement blocked networks configuration to extend SSRF protection P0: implement cache size management and eviction so the disk cannot
fill up
# Completed Steps # Completed Steps
@@ -27,46 +28,12 @@ P1: implement blocked networks configuration to extend SSRF protection
`.golangci.yml` (v2 schema, `default: all` minus six disabled `.golangci.yml` (v2 schema, `default: all` minus six disabled
linters, `lll` 88, tests included): bumped the pinned linters, `lll` 88, tests included): bumped the pinned
`golangci/golangci-lint:v2.12.2-alpine` image in `Dockerfile` and the `golangci/golangci-lint:v2.12.2-alpine` image in `Dockerfile` and the
release-archive sha256 pins in `script/bootstrap`; fixed the findings release-archive sha256 pins in `script/bootstrap`; fixed all 747
the stricter config surfaced (notably `paralleltest`, `wsl_v5`, findings the stricter config surfaced (notably `paralleltest`,
`goconst`, `lll`, `noinlineerr`, `err113`, `errcheck`, `testpackage` `wsl_v5`, `goconst`, `lll`, `noinlineerr`, `err113`, `errcheck`,
white-box test files renamed to `*_internal_test.go`), including #55's `testpackage` white-box test files renamed to
code absorbed after it merged, iterating the pinned linter to `*_internal_test.go`); three `//nolint:tagliatelle` directives keep
`0 issues.`; no single finding total is substantiable, since
golangci-lint's `uniq-by-line` reveals new findings on a line as
others there are fixed — the documented re-measurements were 81 after
the #53 merge and 149 after the #55 merge; three behavior changes, so
not a pure no-op: `Cache.StoreVariant` now takes a `context.Context`
(`noctx`), so a cancelled request skips its best-effort accounting
row; `MetadataStorage.Store`'s cleanup defer was dead on `main` and
leaked `.tmp-*.json` on failure, now fixed with explicit removals; and
the `signing_key` validation error text gained `value too short: `;
the eviction loop's uncancellable context is deferred to #102 under a
`//nolint:contextcheck`; three `//nolint:tagliatelle` directives keep
the snake_case JSON wire/disk formats unchanged; `make check` green the snake_case JSON wire/disk formats unchanged; `make check` green
- 2026-08-07 implement cache size management and eviction (closes
#51): new `cache_max_bytes` config key validated by the startup
framework (explicit values used exactly with no floor, `0` disables
the disk cache entirely, omitted defaults to max(75% of free space
on the filesystem containing `<state_dir>/cache/`, 500 MiB), logged
at startup); processed variants are now tracked in the database (a
new `variant_content` table and an LRU timestamp on `source_content`)
so total usage is two SUMs, never a directory scan on the hot path; a
background goroutine evicts globally least-recently-used entries
(variants and source blobs merged) to the limit, woken by a periodic
ticker and by write-pressure notifications from stores; a source
blob and ALL of its `source_metadata` references are deleted in one
transaction before the file is unlinked, so multi-referenced blobs
are never removed while referenced and rows never point at deleted
files; a startup and periodic reconciliation pass adopts untracked
variant files, drops rows for missing files, removes unreachable
source blobs, and sweeps stale temp files
- 2026-08-07 validate configuration on startup, fail fast on bad
config (closes #52): a config value that is set but unparseable or
invalid aborts startup naming the key and value (defaults apply only
to omitted keys), unknown config keys abort startup, a malformed
config file aborts instead of being skipped, and `state_dir` is
verified creatable and writable before the listener binds
- 2026-08-07 manual test pass of the auth and encrypted URL flows - 2026-08-07 manual test pass of the auth and encrypted URL flows
against a locally built and running `pixad` (built from `main` at against a locally built and running `pixad` (built from `main` at
`6573b9d`, port 18099, local throwaway config); all six checks `6573b9d`, port 18099, local throwaway config); all six checks
@@ -116,6 +83,9 @@ P1: implement blocked networks configuration to extend SSRF protection
# Future Steps # Future Steps
- P0: validate configuration on startup, fail fast on bad config
- P1: implement blocked networks configuration to extend SSRF
protection
- P1: rate limit global concurrent upstream fetches to prevent - P1: rate limit global concurrent upstream fetches to prevent
resource exhaustion resource exhaustion
- P1: strip EXIF and other metadata from processed images (privacy) - P1: strip EXIF and other metadata from processed images (privacy)

View File

@@ -9,7 +9,7 @@ maintenance_mode: false
state_dir: ./data state_dir: ./data
# Image proxy settings # Image proxy settings
# HMAC signing key for URL signatures (required, at least 32 characters) # HMAC signing key for URL signatures (leave empty to require allowlist for all requests)
# Generate with: openssl rand -base64 32 # Generate with: openssl rand -base64 32
signing_key: "CHANGE_ME_generate_with_openssl_rand_base64_32" signing_key: "CHANGE_ME_generate_with_openssl_rand_base64_32"
@@ -28,13 +28,6 @@ allow_http: false
# Maximum concurrent connections per upstream host (default: 20) # Maximum concurrent connections per upstream host (default: 20)
upstream_connections_per_host: 20 upstream_connections_per_host: 20
# Maximum disk cache size in bytes. Explicit values are used exactly as
# given; 0 disables the disk cache entirely (every request fetches and
# processes uncached). When omitted, the default is 75% of the free
# space on the filesystem containing <state_dir>/cache/ at startup,
# with a minimum of 500 MiB.
# cache_max_bytes: 10737418240
# Sentry error reporting (optional) # Sentry error reporting (optional)
sentry_dsn: "" sentry_dsn: ""

View File

@@ -1,305 +0,0 @@
package config
import (
"errors"
"log/slog"
"os"
"path/filepath"
"strings"
"testing"
)
// Static errors returned by the stub free-space probes below.
var (
errTestStatfsFailed = errors.New("statfs failed")
errTestProbeNotExpected = errors.New("probe must not be called")
)
// discardLogger returns a logger that swallows all output, for tests
// that exercise code paths which log.
func discardLogger() *slog.Logger {
return slog.New(slog.DiscardHandler)
}
// TestCacheMaxBytesExplicitValueUsedWithoutFloor verifies that an
// explicitly configured cache_max_bytes value is used exactly as
// given: the 500 MiB floor applies only to the computed default, never
// to explicit values.
func TestCacheMaxBytesExplicitValueUsedWithoutFloor(t *testing.T) {
t.Parallel()
yamlContent := "signing_key: " + validTestSigningKey + "\ncache_max_bytes: 1024\n"
c, err := configFromYAML(t, yamlContent)
if err != nil {
t.Fatalf("explicit cache_max_bytes must be accepted, got error: %v", err)
}
if c.CacheMaxBytes != 1024 {
t.Errorf("CacheMaxBytes = %d, want 1024 (no floor for explicit values)",
c.CacheMaxBytes)
}
}
// TestCacheMaxBytesZeroIsValidAndDisablesCache verifies that an
// explicit zero is a valid value (it disables the disk cache), not an
// error.
func TestCacheMaxBytesZeroIsValidAndDisablesCache(t *testing.T) {
t.Parallel()
yamlContent := "signing_key: " + validTestSigningKey + "\ncache_max_bytes: 0\n"
c, err := configFromYAML(t, yamlContent)
if err != nil {
t.Fatalf("cache_max_bytes: 0 must be accepted, got error: %v", err)
}
if c.CacheMaxBytes != 0 {
t.Errorf("CacheMaxBytes = %d, want 0", c.CacheMaxBytes)
}
}
// TestCacheMaxBytesLargeExplicitValueParses verifies that values above
// 32-bit range parse correctly (the field is an int64 byte count).
func TestCacheMaxBytesLargeExplicitValueParses(t *testing.T) {
t.Parallel()
yamlContent := "signing_key: " + validTestSigningKey +
"\ncache_max_bytes: 10737418240\n"
c, err := configFromYAML(t, yamlContent)
if err != nil {
t.Fatalf("large cache_max_bytes must be accepted, got error: %v", err)
}
if c.CacheMaxBytes != 10737418240 {
t.Errorf("CacheMaxBytes = %d, want 10737418240", c.CacheMaxBytes)
}
}
// TestCacheMaxBytesInvalidValuesAbortStartup verifies that a SET but
// invalid cache_max_bytes value aborts startup naming the key and the
// offending value, per the no-silent-fallback rule: defaults apply
// only to omitted keys.
func TestCacheMaxBytesInvalidValuesAbortStartup(t *testing.T) {
t.Parallel()
signingKeyLine := "signing_key: " + validTestSigningKey + "\n"
cases := []struct {
name string
yaml string
// wantErrSubstrings must all appear in the error message.
wantErrSubstrings []string
}{
{
name: "negative",
yaml: signingKeyLine + "cache_max_bytes: -1024\n",
wantErrSubstrings: []string{keyCacheMaxBytes, "-1024"},
},
{
name: "float",
yaml: signingKeyLine + "cache_max_bytes: 3.5\n",
wantErrSubstrings: []string{keyCacheMaxBytes, "3.5"},
},
{
name: "non-numeric string",
yaml: signingKeyLine + "cache_max_bytes: banana\n",
wantErrSubstrings: []string{keyCacheMaxBytes, "banana"},
},
{
name: "explicit null",
yaml: signingKeyLine + "cache_max_bytes: null\n",
wantErrSubstrings: []string{keyCacheMaxBytes, "null"},
},
{
name: "bare key no value",
yaml: signingKeyLine + "cache_max_bytes:\n",
wantErrSubstrings: []string{keyCacheMaxBytes, "null"},
},
{
name: "boolean",
yaml: signingKeyLine + "cache_max_bytes: true\n",
wantErrSubstrings: []string{keyCacheMaxBytes, "true"},
},
{
name: "list",
yaml: signingKeyLine + "cache_max_bytes:\n - 1\n",
wantErrSubstrings: []string{keyCacheMaxBytes},
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
c, err := configFromYAML(t, tc.yaml)
if err == nil {
t.Fatalf(
"config with %s cache_max_bytes must abort startup, got config: %+v",
tc.name, c)
}
t.Logf("got expected error: %v", err)
for _, want := range tc.wantErrSubstrings {
if !strings.Contains(err.Error(), want) {
t.Errorf("error %q does not mention %q", err.Error(), want)
}
}
})
}
}
// TestComputeDefaultCacheMaxBytesUses75PercentOfFreeSpace verifies the
// computed default is 75% of the probed free space when that exceeds
// the floor.
func TestComputeDefaultCacheMaxBytesUses75PercentOfFreeSpace(t *testing.T) {
t.Parallel()
// 4 GiB free -> 3 GiB default.
probe := func(string) (uint64, error) { return 4294967296, nil }
got, err := ComputeDefaultCacheMaxBytes(t.TempDir(), probe)
if err != nil {
t.Fatalf("ComputeDefaultCacheMaxBytes returned error: %v", err)
}
if got != 3221225472 {
t.Errorf("ComputeDefaultCacheMaxBytes = %d, want 3221225472 (75%% of 4 GiB)",
got)
}
}
// TestComputeDefaultCacheMaxBytesAppliesFloorToComputedDefault
// verifies that when 75% of free space is below 500 MiB, the computed
// default is floored at DefaultCacheMaxBytesFloor.
func TestComputeDefaultCacheMaxBytesAppliesFloorToComputedDefault(t *testing.T) {
t.Parallel()
cases := []struct {
name string
freeBytes uint64
}{
{name: "100 MiB free", freeBytes: 104857600},
{name: "zero free", freeBytes: 0},
{name: "just below floor threshold", freeBytes: 699050665},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
probe := func(string) (uint64, error) { return tc.freeBytes, nil }
got, err := ComputeDefaultCacheMaxBytes(t.TempDir(), probe)
if err != nil {
t.Fatalf("ComputeDefaultCacheMaxBytes returned error: %v", err)
}
if got != DefaultCacheMaxBytesFloor {
t.Errorf("ComputeDefaultCacheMaxBytes = %d, want floor %d",
got, DefaultCacheMaxBytesFloor)
}
})
}
}
// TestComputeDefaultCacheMaxBytesPropagatesProbeError verifies that a
// failing free-space probe produces an error naming the config key,
// instead of a silently wrong default.
func TestComputeDefaultCacheMaxBytesPropagatesProbeError(t *testing.T) {
t.Parallel()
probe := func(string) (uint64, error) { return 0, errTestStatfsFailed }
_, err := ComputeDefaultCacheMaxBytes(t.TempDir(), probe)
if err == nil {
t.Fatal("probe failure must produce an error, got nil")
}
t.Logf("got expected error: %v", err)
if !strings.Contains(err.Error(), keyCacheMaxBytes) {
t.Errorf("error %q does not name the config key cache_max_bytes", err.Error())
}
}
// TestResolveCacheMaxBytesComputesDefaultWhenOmitted verifies that an
// omitted cache_max_bytes key resolves to the computed default, that
// the probe is pointed at <state_dir>/cache/ (which must be created
// first so statfs measures the right filesystem), and that the result
// lands on the Config.
func TestResolveCacheMaxBytesComputesDefaultWhenOmitted(t *testing.T) {
t.Parallel()
c, err := configFromYAML(t, "signing_key: "+validTestSigningKey+"\n")
if err != nil {
t.Fatalf("minimal config should be valid, got error: %v", err)
}
c.StateDir = t.TempDir()
wantCacheDir := filepath.Join(c.StateDir, "cache")
var probedPath string
// 4 GiB free -> 3 GiB default.
probe := func(path string) (uint64, error) {
probedPath = path
return 4294967296, nil
}
err = c.resolveCacheMaxBytes(discardLogger(), probe)
if err != nil {
t.Fatalf("resolveCacheMaxBytes returned error: %v", err)
}
if c.CacheMaxBytes != 3221225472 {
t.Errorf("CacheMaxBytes = %d, want computed default 3221225472",
c.CacheMaxBytes)
}
if probedPath != wantCacheDir {
t.Errorf("free space probed at %q, want cache directory %q",
probedPath, wantCacheDir)
}
info, err := os.Stat(wantCacheDir)
if err != nil || !info.IsDir() {
t.Errorf("cache directory %q was not created before probing: info=%v err=%v",
wantCacheDir, info, err)
}
}
// TestResolveCacheMaxBytesDoesNotOverrideExplicitValue verifies that
// an explicitly configured value survives resolution untouched and
// that the free-space probe is never consulted for it.
func TestResolveCacheMaxBytesDoesNotOverrideExplicitValue(t *testing.T) {
t.Parallel()
yamlContent := "signing_key: " + validTestSigningKey + "\ncache_max_bytes: 1024\n"
c, err := configFromYAML(t, yamlContent)
if err != nil {
t.Fatalf("explicit cache_max_bytes must be accepted, got error: %v", err)
}
c.StateDir = t.TempDir()
probe := func(string) (uint64, error) {
t.Error("free-space probe must not be consulted for explicit values")
return 0, errTestProbeNotExpected
}
err = c.resolveCacheMaxBytes(discardLogger(), probe)
if err != nil {
t.Fatalf("resolveCacheMaxBytes returned error: %v", err)
}
if c.CacheMaxBytes != 1024 {
t.Errorf("CacheMaxBytes = %d, want explicit 1024 (no floor, no recompute)",
c.CacheMaxBytes)
}
}

View File

@@ -1,116 +0,0 @@
package config
import (
"fmt"
"log/slog"
"math"
"os"
"path/filepath"
"syscall"
)
// DefaultCacheMaxBytesFloor is the minimum computed default for the
// cache_max_bytes setting: 500 MiB. The floor applies only to the
// computed default (when the key is omitted from the configuration),
// never to explicitly configured values.
const DefaultCacheMaxBytesFloor int64 = 524288000
// cacheDirPerms is the permission mode for the cache directory created
// before probing free space, matching the state directory permissions.
const cacheDirPerms = 0o750
// freeSpaceFractionNumerator and freeSpaceFractionDenominator express
// the 75% share of free space used for the computed default limit as
// integer arithmetic (dividing before multiplying avoids overflow).
const (
freeSpaceFractionNumerator uint64 = 3
freeSpaceFractionDenominator uint64 = 4
)
// FreeSpaceProbeFunc reports the number of free bytes available on the
// filesystem containing path. It is a function type so tests can
// inject a fake probe instead of depending on the host disk.
type FreeSpaceProbeFunc func(path string) (uint64, error)
// defaultFreeSpaceProbe reports free filesystem bytes via statfs on
// the given path, as available to unprivileged processes.
func defaultFreeSpaceProbe(path string) (uint64, error) {
var stat syscall.Statfs_t
err := syscall.Statfs(path, &stat)
if err != nil {
return 0, err
}
if stat.Bsize < 0 {
return 0, fmt.Errorf("%w %d for %q", errNegativeBlockSize, stat.Bsize, path)
}
blockSize := uint64(stat.Bsize)
return stat.Bavail * blockSize, nil
}
// ComputeDefaultCacheMaxBytes returns the default cache size limit for
// the filesystem containing cacheDir: 75% of the free bytes reported
// by probe, with a floor of DefaultCacheMaxBytesFloor.
func ComputeDefaultCacheMaxBytes(
cacheDir string, probe FreeSpaceProbeFunc,
) (int64, error) {
freeBytes, err := probe(cacheDir)
if err != nil {
return 0, fmt.Errorf("config key %q: cannot determine free space for %q: %w",
"cache_max_bytes", cacheDir, err)
}
computed := freeBytes / freeSpaceFractionDenominator * freeSpaceFractionNumerator
computed = min(computed, math.MaxInt64)
// gosec cannot see that min() above bounds computed, so it reads
// this conversion as potentially overflowing. It cannot: computed is
// at most math.MaxInt64 on every path here.
//nolint:gosec // G115: clamped to MaxInt64 by min above
limit := int64(computed)
limit = max(limit, DefaultCacheMaxBytesFloor)
return limit, nil
}
// resolveCacheMaxBytes finalizes CacheMaxBytes after state_dir
// validation: an explicitly configured value is kept as-is (no floor
// applies), while an omitted key receives the computed default based
// on free space in <state_dir>/cache/. The cache directory is created
// first so statfs measures the filesystem that will actually hold the
// cache. The effective limit is logged either way.
func (c *Config) resolveCacheMaxBytes(
log *slog.Logger, probe FreeSpaceProbeFunc,
) error {
if !c.cacheMaxBytesExplicit {
cacheDir := filepath.Join(c.StateDir, "cache")
err := os.MkdirAll(cacheDir, cacheDirPerms)
if err != nil {
return fmt.Errorf("config key %q: cannot create cache directory %q: %w",
keyCacheMaxBytes, cacheDir, err)
}
limit, err := ComputeDefaultCacheMaxBytes(cacheDir, probe)
if err != nil {
return err
}
c.CacheMaxBytes = limit
log.Info("computed default cache size limit from free space",
"cache_max_bytes", limit,
"cache_dir", cacheDir,
)
}
log.Info("effective cache size limit",
"cache_max_bytes", c.CacheMaxBytes,
"cache_disabled", c.CacheMaxBytes == 0,
)
return nil
}

View File

@@ -5,12 +5,8 @@ import (
"errors" "errors"
"fmt" "fmt"
"log/slog" "log/slog"
"math"
"net/url"
"os" "os"
"path/filepath" "path/filepath"
"sort"
"strconv"
"strings" "strings"
"git.eeqj.de/sneak/smartconfig" "git.eeqj.de/sneak/smartconfig"
@@ -26,55 +22,6 @@ const (
DefaultUpstreamConnectionsPerHost = 20 DefaultUpstreamConnectionsPerHost = 20
) )
// Configuration key names.
const (
keyDebug = "debug"
keyMaintenanceMode = "maintenance_mode"
keyPort = "port"
keyStateDir = "state_dir"
keySentryDSN = "sentry_dsn"
keyDBURL = "db_url"
keyMetrics = "metrics"
keyMetricsUsername = "metrics.username"
keyMetricsPassword = "metrics.password"
keySigningKey = "signing_key"
keyAllowlistHosts = "allowlist_hosts"
keyAllowHTTP = "allow_http"
keyUpstreamConnectionsPerHost = "upstream_connections_per_host"
keyCacheMaxBytes = "cache_max_bytes"
)
// Static validation errors. Each use site attaches the offending key
// and value by wrapping these with fmt.Errorf and %w.
var (
errValueRequired = errors.New("a value is required")
errValueEmpty = errors.New("value must not be empty")
errUnknownConfigKeys = errors.New("unknown config keys")
errNotAString = errors.New("not a string")
errNotAnInteger = errors.New("not an integer")
errNotABoolean = errors.New("not a boolean")
errNotAStringList = errors.New("not a list of strings")
errNotAMetricsMap = errors.New("not a map of metrics settings")
errEmptyListEntry = errors.New("list contains an empty entry")
errEmptyEntry = errors.New("contains an empty entry")
errNotAValidURL = errors.New("not a valid URL")
errPortOutOfRange = errors.New("outside the valid port range")
errTooFewConnections = errors.New("must be at least 1")
errValueTooShort = errors.New("value too short")
errMustBeSetTogether = errors.New("must be set together")
errMustNotBeNegative = errors.New("must not be negative")
errOverflowsInt64 = errors.New("overflows a 64-bit integer")
errNegativeBlockSize = errors.New(
"statfs reported negative block size")
errValueNull = errors.New(
"value is null; omit the key entirely to use the default")
errValuesNull = errors.New(
"value is null; omit a key entirely to use its default")
errNotBareHostname = errors.New(
"must be a bare hostname without scheme, path, or whitespace")
errNoHostnameLabels = errors.New("contains no hostname labels")
)
// Params defines dependencies for Config. // Params defines dependencies for Config.
type Params struct { type Params struct {
fx.In fx.In
@@ -83,6 +30,12 @@ type Params struct {
Logger *logger.Logger Logger *logger.Logger
} }
// Static validation errors.
var (
errSigningKeyRequired = errors.New("signing_key is required")
errSigningKeyTooShort = errors.New("signing_key too short")
)
// Config holds application configuration values. // Config holds application configuration values.
type Config struct { type Config struct {
Debug bool Debug bool
@@ -99,19 +52,6 @@ type Config struct {
AllowlistHosts []string // Hosts that don't require signatures AllowlistHosts []string // Hosts that don't require signatures
AllowHTTP bool // Allow non-TLS upstream (testing only) AllowHTTP bool // Allow non-TLS upstream (testing only)
UpstreamConnectionsPerHost int // Max concurrent connections per upstream host UpstreamConnectionsPerHost int // Max concurrent connections per upstream host
// CacheMaxBytes is the disk cache size limit in bytes. Zero
// disables the disk cache entirely. When cache_max_bytes is
// omitted from the configuration, this holds the computed default
// (75% of free space on the filesystem containing
// <state_dir>/cache/, floored at DefaultCacheMaxBytesFloor).
CacheMaxBytes int64
// cacheMaxBytesExplicit records whether cache_max_bytes was
// explicitly set in the configuration file. Explicit values are
// used exactly as given; only an omitted key gets the computed
// default (and its floor) in resolveCacheMaxBytes.
cacheMaxBytesExplicit bool
} }
// New creates a new Config instance by loading configuration from file. // New creates a new Config instance by loading configuration from file.
@@ -128,94 +68,34 @@ func New(_ fx.Lifecycle, params Params) (*Config, error) {
log.Info("no config file found, using defaults") log.Info("no config file found, using defaults")
} }
c, err := newFromSmartConfig(sc) c := &Config{
if err != nil { Debug: getBool(sc, "debug", false),
return nil, err MaintenanceMode: getBool(sc, "maintenance_mode", false),
Port: getInt(sc, "port", DefaultPort),
StateDir: getString(sc, "state_dir", DefaultStateDir),
SentryDSN: getString(sc, "sentry_dsn", ""),
MetricsUsername: getString(sc, "metrics.username", ""),
MetricsPassword: getString(sc, "metrics.password", ""),
SigningKey: getString(sc, "signing_key", ""),
AllowlistHosts: getStringSlice(sc),
AllowHTTP: getBool(sc, "allow_http", false),
UpstreamConnectionsPerHost: getInt(
sc, "upstream_connections_per_host", DefaultUpstreamConnectionsPerHost,
),
} }
err = c.ensureStateDirWritable() // Build DBURL from StateDir if not explicitly set
if err != nil { c.DBURL = getString(sc, "db_url", "")
return nil, err if c.DBURL == "" {
} c.DBURL = fmt.Sprintf("file:%s/state.sqlite3?_journal_mode=WAL", c.StateDir)
err = c.resolveCacheMaxBytes(log, defaultFreeSpaceProbe)
if err != nil {
return nil, err
} }
if c.Debug { if c.Debug {
params.Logger.EnableDebugLogging() params.Logger.EnableDebugLogging()
} }
return c, nil // Validate required configuration
} err = c.validate()
// newFromSmartConfig constructs a Config from a loaded smartconfig
// instance and validates it. A nil sc means no config file was found,
// in which case every option takes its default value. A key that is
// present but unparseable or invalid is an error: defaults apply only
// to omitted keys, never to invalid explicit values.
func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
if sc != nil {
err := validateKnownKeys(sc)
if err != nil {
return nil, err
}
err = validateAllowlistHostsValue(sc)
if err != nil {
return nil, err
}
}
loader := &strictLoader{sc: sc}
c := &Config{
Debug: loader.boolVal(keyDebug, false),
MaintenanceMode: loader.boolVal(keyMaintenanceMode, false),
Port: loader.intVal(keyPort, DefaultPort),
StateDir: loader.stringVal(keyStateDir, DefaultStateDir),
SentryDSN: loader.stringVal(keySentryDSN, ""),
MetricsUsername: loader.stringVal(keyMetricsUsername, ""),
MetricsPassword: loader.stringVal(keyMetricsPassword, ""),
SigningKey: loader.stringVal(keySigningKey, ""),
AllowlistHosts: getStringSlice(sc),
AllowHTTP: loader.boolVal(keyAllowHTTP, false),
UpstreamConnectionsPerHost: loader.intVal(
keyUpstreamConnectionsPerHost, DefaultUpstreamConnectionsPerHost),
CacheMaxBytes: loader.int64Val(keyCacheMaxBytes, 0),
}
// The computed default for cache_max_bytes needs a validated
// state_dir, so it is resolved later (resolveCacheMaxBytes); here
// we only record whether the operator set the key explicitly.
if sc != nil {
if _, present := sc.Get(keyCacheMaxBytes); present {
c.cacheMaxBytesExplicit = true
}
}
// Build DBURL from StateDir if not explicitly set. The derived URL
// is a default: it applies only when db_url is omitted, never to an
// explicitly empty value.
c.DBURL = loader.stringVal(keyDBURL, "")
if c.DBURL == "" && loader.err == nil {
if sc != nil {
if _, present := sc.Get(keyDBURL); present {
return nil, fmt.Errorf(
"config key %q: %w; omit the key to derive it from state_dir",
keyDBURL, errValueEmpty)
}
}
c.DBURL = fmt.Sprintf("file:%s/state.sqlite3?_journal_mode=WAL", c.StateDir)
}
if loader.err != nil {
return nil, loader.err
}
err := c.validate()
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -223,201 +103,18 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
return c, nil return c, nil
} }
// validateKnownKeys rejects configuration files containing keys the // validate checks that all required configuration values are set.
// application does not understand, so typos fail at startup instead of
// being silently ignored, and rejects keys that are explicitly set to
// null: a null is a SET value, never an omission, so it must not
// silently take the default. The env section is permitted because
// smartconfig consumes it for environment variable injection.
func validateKnownKeys(sc *smartconfig.Config) error {
var unknown, nullKeys []string
for key, value := range sc.Data() {
if !isKnownConfigKey(key) {
unknown = append(unknown, key)
continue
}
if value == nil {
nullKeys = append(nullKeys, key)
continue
}
if key == keyMetrics {
metricsMap, ok := value.(map[string]any)
if !ok {
return fmt.Errorf("config key %q: value %v is %w",
keyMetrics, value, errNotAMetricsMap)
}
for subkey, subvalue := range metricsMap {
if subkey != "username" && subkey != "password" {
unknown = append(unknown, keyMetrics+"."+subkey)
continue
}
if subvalue == nil {
nullKeys = append(nullKeys, keyMetrics+"."+subkey)
}
}
}
}
if len(unknown) > 0 {
sort.Strings(unknown)
return fmt.Errorf("%w: %s", errUnknownConfigKeys, strings.Join(unknown, ", "))
}
if len(nullKeys) > 0 {
sort.Strings(nullKeys)
if len(nullKeys) == 1 {
return errNullConfigValue(nullKeys[0])
}
return fmt.Errorf("config keys %s: %w",
strings.Join(nullKeys, ", "), errValuesNull)
}
return nil
}
// errNullConfigValue reports a config key that is explicitly set to
// null (including the bare "key:" form and the "~" alias). Silently
// applying the default would mask a truncated or typo'd config entry.
func errNullConfigValue(key string) error {
return fmt.Errorf("config key %q: %w", key, errValueNull)
}
// isKnownConfigKey reports whether key is a permitted top-level
// configuration key.
func isKnownConfigKey(key string) bool {
switch key {
case keyDebug, keyMaintenanceMode, keyPort, keyStateDir, keySentryDSN,
keyDBURL, keyMetrics, keySigningKey, keyAllowlistHosts, keyAllowHTTP,
keyUpstreamConnectionsPerHost, keyCacheMaxBytes, "env":
return true
}
return false
}
// ensureStateDirWritable verifies at startup that StateDir can be
// created and written to, so a misconfigured path aborts startup
// instead of failing later at first use.
func (c *Config) ensureStateDirWritable() error {
const stateDirPerms = 0o750
err := os.MkdirAll(c.StateDir, stateDirPerms)
if err != nil {
return fmt.Errorf("config key %q: cannot create directory %q: %w",
keyStateDir, c.StateDir, err)
}
probe, err := os.CreateTemp(c.StateDir, ".startup-write-probe-*")
if err != nil {
return fmt.Errorf("config key %q: directory %q is not writable: %w",
keyStateDir, c.StateDir, err)
}
probePath := probe.Name()
err = probe.Close()
if err != nil {
return fmt.Errorf("config key %q: cannot close probe file %q: %w",
keyStateDir, probePath, err)
}
err = os.Remove(probePath)
if err != nil {
return fmt.Errorf("config key %q: cannot remove probe file %q: %w",
keyStateDir, probePath, err)
}
return nil
}
// validate checks that all required configuration values are set and
// that every value is within its valid range.
func (c *Config) validate() error { func (c *Config) validate() error {
// The signing key value is never echoed in error messages.
if c.SigningKey == "" { if c.SigningKey == "" {
return fmt.Errorf("config key %q: %w", keySigningKey, errValueRequired) return errSigningKeyRequired
} }
// Minimum key length for security (32 bytes = 256 bits) // Minimum key length for security (32 bytes = 256 bits)
const minKeyLength = 32 const minKeyLength = 32
if len(c.SigningKey) < minKeyLength { if len(c.SigningKey) < minKeyLength {
return fmt.Errorf("config key %q: %w: must be at least %d characters, got %d", return fmt.Errorf(
keySigningKey, errValueTooShort, minKeyLength, len(c.SigningKey)) "%w: must be at least %d characters", errSigningKeyTooShort, minKeyLength,
} )
const maxPort = 65535
if c.Port < 1 || c.Port > maxPort {
return fmt.Errorf("config key %q: value %d is %w 1-%d",
keyPort, c.Port, errPortOutOfRange, maxPort)
}
if c.UpstreamConnectionsPerHost < 1 {
return fmt.Errorf("config key %q: value %d %w",
keyUpstreamConnectionsPerHost, c.UpstreamConnectionsPerHost,
errTooFewConnections)
}
if c.StateDir == "" {
return fmt.Errorf("config key %q: %w", keyStateDir, errValueEmpty)
}
// Zero is valid (it disables the disk cache); only negative
// values are rejected. No floor applies to explicit values.
if c.CacheMaxBytes < 0 {
return fmt.Errorf("config key %q: value %d %w",
keyCacheMaxBytes, c.CacheMaxBytes, errMustNotBeNegative)
}
for _, host := range c.AllowlistHosts {
err := validateAllowlistHost(host)
if err != nil {
return err
}
}
if c.SentryDSN != "" {
parsed, err := url.Parse(c.SentryDSN)
if err != nil || parsed.Scheme == "" || parsed.Host == "" {
return fmt.Errorf("config key %q: value %q is %w",
keySentryDSN, c.SentryDSN, errNotAValidURL)
}
}
if (c.MetricsUsername == "") != (c.MetricsPassword == "") {
return fmt.Errorf("config keys %q and %q %w",
keyMetricsUsername, keyMetricsPassword, errMustBeSetTogether)
}
return nil
}
// validateAllowlistHost checks that an allowlist_hosts entry is a bare
// hostname, optionally with a leading dot for suffix matching. URLs,
// paths, and whitespace indicate a misconfigured entry. An entry with
// no hostname labels (such as ".") is rejected: the allowlist matcher
// treats a leading dot as a suffix pattern, so a bare "." would match
// any upstream host written in FQDN trailing-dot form and effectively
// disable URL signing.
func validateAllowlistHost(host string) error {
if strings.Contains(host, "://") || strings.ContainsAny(host, "/ \t") {
return fmt.Errorf("config key %q: entry %q %w",
keyAllowlistHosts, host, errNotBareHostname)
}
if strings.Trim(host, ".") == "" {
return fmt.Errorf("config key %q: entry %q %w",
keyAllowlistHosts, host, errNoHostnameLabels)
} }
return nil return nil
@@ -453,11 +150,11 @@ func loadConfigFile(log *slog.Logger, appName string) (*smartconfig.Config, erro
_, statErr := os.Stat(cleanPath) _, statErr := os.Stat(cleanPath)
if statErr == nil { if statErr == nil {
// A config file that exists but does not parse is a fatal
// startup error, never something to skip over.
sc, err := smartconfig.NewFromConfigPath(path) sc, err := smartconfig.NewFromConfigPath(path)
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to parse config file %s: %w", path, err) log.Warn("failed to parse config file", "path", path, "error", err)
continue
} }
log.Info("loaded config file", "path", path) log.Info("loaded config file", "path", path)
@@ -469,279 +166,51 @@ func loadConfigFile(log *slog.Logger, appName string) (*smartconfig.Config, erro
return nil, nil //nolint:nilnil // nil config is valid (use defaults) return nil, nil //nolint:nilnil // nil config is valid (use defaults)
} }
// strictLoader accumulates the first error encountered while reading func getString(sc *smartconfig.Config, key, defaultVal string) string {
// typed values out of a smartconfig instance, so Config construction if sc == nil {
// can stay a single struct literal. return defaultVal
type strictLoader struct {
sc *smartconfig.Config
err error
}
func (l *strictLoader) stringVal(key, defaultVal string) string {
if l.err != nil {
return ""
} }
val, err := getString(l.sc, key, defaultVal) val, err := sc.GetString(key)
if err != nil { if err != nil {
l.err = err return defaultVal
} }
return val return val
} }
func (l *strictLoader) intVal(key string, defaultVal int) int { func getInt(sc *smartconfig.Config, key string, defaultVal int) int {
if l.err != nil { if sc == nil {
return 0 return defaultVal
} }
val, err := getInt(l.sc, key, defaultVal) val, err := sc.GetInt(key)
if err != nil { if err != nil {
l.err = err return defaultVal
} }
return val return val
} }
func (l *strictLoader) int64Val(key string, defaultVal int64) int64 { func getBool(sc *smartconfig.Config, key string, defaultVal bool) bool {
if l.err != nil { if sc == nil {
return 0 return defaultVal
} }
val, err := getInt64(l.sc, key, defaultVal) val, err := sc.GetBool(key)
if err != nil { if err != nil {
l.err = err return defaultVal
} }
return val return val
} }
func (l *strictLoader) boolVal(key string, defaultVal bool) bool {
if l.err != nil {
return false
}
val, err := getBool(l.sc, key, defaultVal)
if err != nil {
l.err = err
}
return val
}
// getString returns the string value for key, or defaultVal if the key
// is omitted. A present value that is not a string, or is explicitly
// null, is an error.
func getString(sc *smartconfig.Config, key, defaultVal string) (string, error) {
if sc == nil {
return defaultVal, nil
}
raw, ok := sc.Get(key)
if !ok {
return defaultVal, nil
}
if raw == nil {
return "", errNullConfigValue(key)
}
str, ok := raw.(string)
if !ok {
return "", fmt.Errorf("config key %q: value %v (%T) is %w",
key, raw, raw, errNotAString)
}
return str, nil
}
// getInt returns the integer value for key, or defaultVal if the key is
// omitted. A present value that is not a whole number, or is explicitly
// null, is an error; fractional values are never truncated.
func getInt(sc *smartconfig.Config, key string, defaultVal int) (int, error) {
if sc == nil {
return defaultVal, nil
}
raw, ok := sc.Get(key)
if !ok {
return defaultVal, nil
}
if raw == nil {
return 0, errNullConfigValue(key)
}
switch val := raw.(type) {
case int:
return val, nil
case int64:
return int(val), nil
case float64:
if val != math.Trunc(val) {
return 0, fmt.Errorf("config key %q: value %v is %w",
key, val, errNotAnInteger)
}
return int(val), nil
case string:
parsed, err := strconv.Atoi(strings.TrimSpace(val))
if err != nil {
return 0, fmt.Errorf("config key %q: value %q is %w",
key, val, errNotAnInteger)
}
return parsed, nil
default:
return 0, fmt.Errorf("config key %q: value %v (%T) is %w",
key, raw, raw, errNotAnInteger)
}
}
// getInt64 returns the 64-bit integer value for key, or defaultVal if
// the key is omitted. A present value that is not a whole number, or
// is explicitly null, is an error; fractional values are never
// truncated and out-of-range values are never clamped.
func getInt64(sc *smartconfig.Config, key string, defaultVal int64) (int64, error) {
if sc == nil {
return defaultVal, nil
}
raw, ok := sc.Get(key)
if !ok {
return defaultVal, nil
}
if raw == nil {
return 0, errNullConfigValue(key)
}
switch val := raw.(type) {
case int:
return int64(val), nil
case int64:
return val, nil
case uint64:
if val > math.MaxInt64 {
return 0, fmt.Errorf("config key %q: value %d %w",
key, val, errOverflowsInt64)
}
return int64(val), nil
case float64:
if val != math.Trunc(val) {
return 0, fmt.Errorf("config key %q: value %v is %w",
key, val, errNotAnInteger)
}
return int64(val), nil
case string:
parsed, err := strconv.ParseInt(strings.TrimSpace(val), 10, 64)
if err != nil {
return 0, fmt.Errorf("config key %q: value %q is %w",
key, val, errNotAnInteger)
}
return parsed, nil
default:
return 0, fmt.Errorf("config key %q: value %v (%T) is %w",
key, raw, raw, errNotAnInteger)
}
}
// getBool returns the boolean value for key, or defaultVal if the key
// is omitted. A present value that is not a boolean (or a ParseBool-able
// string), or is explicitly null, is an error; numbers are not accepted
// as booleans.
func getBool(sc *smartconfig.Config, key string, defaultVal bool) (bool, error) {
if sc == nil {
return defaultVal, nil
}
raw, ok := sc.Get(key)
if !ok {
return defaultVal, nil
}
if raw == nil {
return false, errNullConfigValue(key)
}
switch val := raw.(type) {
case bool:
return val, nil
case string:
parsed, err := strconv.ParseBool(strings.TrimSpace(val))
if err != nil {
return false, fmt.Errorf("config key %q: value %q is %w",
key, val, errNotABoolean)
}
return parsed, nil
default:
return false, fmt.Errorf("config key %q: value %v (%T) is %w",
key, raw, raw, errNotABoolean)
}
}
// validateAllowlistHostsValue checks the raw shape of the
// allowlist_hosts value before the lenient extraction in getStringSlice
// runs: an explicitly null value, a value that is not a list of strings
// (or a comma-separated string), a non-string entry, or an empty entry
// is an error, never silently skipped.
func validateAllowlistHostsValue(sc *smartconfig.Config) error {
raw, ok := sc.Get(keyAllowlistHosts)
if !ok {
return nil
}
if raw == nil {
return errNullConfigValue(keyAllowlistHosts)
}
switch val := raw.(type) {
case []any:
for _, item := range val {
str, ok := item.(string)
if !ok {
return fmt.Errorf("config key %q: list entry %v (%T) is %w",
keyAllowlistHosts, item, item, errNotAString)
}
if strings.TrimSpace(str) == "" {
return fmt.Errorf("config key %q: %w",
keyAllowlistHosts, errEmptyListEntry)
}
}
case string:
if strings.TrimSpace(val) == "" {
return nil
}
for part := range strings.SplitSeq(val, ",") {
if strings.TrimSpace(part) == "" {
return fmt.Errorf("config key %q: value %q %w",
keyAllowlistHosts, val, errEmptyEntry)
}
}
default:
return fmt.Errorf("config key %q: value %v (%T) is %w",
keyAllowlistHosts, raw, raw, errNotAStringList)
}
return nil
}
// getStringSlice returns the allowlist_hosts list of strings, or nil if
// the key is omitted. It accepts a YAML list of strings or a
// comma-separated string (backwards compatibility). Malformed entries
// are rejected beforehand by validateAllowlistHostsValue.
func getStringSlice(sc *smartconfig.Config) []string { func getStringSlice(sc *smartconfig.Config) []string {
if sc == nil { if sc == nil {
return nil return nil
} }
val, ok := sc.Get(keyAllowlistHosts) val, ok := sc.Get("allowlist_hosts")
if !ok || val == nil { if !ok || val == nil {
return nil return nil
} }

View File

@@ -40,7 +40,7 @@ func checkAllowlistHosts(t *testing.T, configPath string) {
t.Errorf("expected 3 hosts, got %d: %v", len(hosts), hosts) t.Errorf("expected 3 hosts, got %d: %v", len(hosts), hosts)
} }
expected := []string{"static.sneak.cloud", "sneak.berlin", testHostS3} expected := []string{"static.sneak.cloud", "sneak.berlin", "s3.sneak.cloud"}
for i, want := range expected { for i, want := range expected {
if i >= len(hosts) { if i >= len(hosts) {
t.Errorf("missing host at index %d: want %q", i, want) t.Errorf("missing host at index %d: want %q", i, want)

View File

@@ -1,596 +0,0 @@
package config
import (
"log/slog"
"os"
"path/filepath"
"strings"
"testing"
"git.eeqj.de/sneak/smartconfig"
)
// validTestSigningKey is a 32-character signing key that satisfies the
// minimum length requirement in validate().
const validTestSigningKey = "0123456789abcdef0123456789abcdef"
// signingKeyLine is a valid signing_key config line used as the base of
// test config files.
const signingKeyLine = "signing_key: " + validTestSigningKey + "\n"
// testHostS3 is an allowlist host entry used across the config tests.
const testHostS3 = "s3.sneak.cloud"
// nullValueText is the substring that error messages about explicitly
// null config values must contain.
const nullValueText = "null"
// abortCase describes a config file that must abort startup with an
// error mentioning every string in wantErrSubstrings.
type abortCase struct {
name string
yaml string
// wantErrSubstrings must all appear in the error message.
wantErrSubstrings []string
}
// configFromYAML writes yamlContent to a temporary config file, loads it
// via smartconfig, and constructs a Config from it using the same code
// path the server uses at startup.
func configFromYAML(t *testing.T, yamlContent string) (*Config, error) {
t.Helper()
tmpDir := t.TempDir()
configPath := filepath.Join(tmpDir, "config.yml")
err := os.WriteFile(configPath, []byte(yamlContent), 0o600)
if err != nil {
t.Fatalf("failed to write test config: %v", err)
}
sc, err := smartconfig.NewFromConfigPath(configPath)
if err != nil {
t.Fatalf("failed to load test config: %v", err)
}
return newFromSmartConfig(sc)
}
func TestOmittedValuesUseDefaults(t *testing.T) {
t.Parallel()
c, err := configFromYAML(t, signingKeyLine)
if err != nil {
t.Fatalf("minimal config should be valid, got error: %v", err)
}
if c.Port != DefaultPort {
t.Errorf("Port = %d, want default %d", c.Port, DefaultPort)
}
if c.StateDir != DefaultStateDir {
t.Errorf("StateDir = %q, want default %q", c.StateDir, DefaultStateDir)
}
if c.UpstreamConnectionsPerHost != DefaultUpstreamConnectionsPerHost {
t.Errorf("UpstreamConnectionsPerHost = %d, want default %d",
c.UpstreamConnectionsPerHost, DefaultUpstreamConnectionsPerHost)
}
if c.Debug {
t.Error("Debug = true, want default false")
}
if c.MaintenanceMode {
t.Error("MaintenanceMode = true, want default false")
}
if c.AllowHTTP {
t.Error("AllowHTTP = true, want default false")
}
if len(c.AllowlistHosts) != 0 {
t.Errorf("AllowlistHosts = %v, want empty", c.AllowlistHosts)
}
wantDBURL := "file:" + DefaultStateDir + "/state.sqlite3?_journal_mode=WAL"
if c.DBURL != wantDBURL {
t.Errorf("DBURL = %q, want derived default %q", c.DBURL, wantDBURL)
}
}
func TestExplicitValidValuesAreUsed(t *testing.T) {
t.Parallel()
yamlContent := `
port: 9090
debug: true
maintenance_mode: true
state_dir: /tmp/pixa-test-state
db_url: "file:/tmp/pixa-test-state/other.sqlite3"
signing_key: ` + validTestSigningKey + `
allowlist_hosts:
- s3.sneak.cloud
- .example.com
allow_http: true
upstream_connections_per_host: 5
sentry_dsn: "https://abc123@sentry.example.com/42"
metrics:
username: metricsuser
password: metricspass
`
c, err := configFromYAML(t, yamlContent)
if err != nil {
t.Fatalf("valid config should load, got error: %v", err)
}
if c.Port != 9090 {
t.Errorf("Port = %d, want 9090", c.Port)
}
if !c.Debug || !c.MaintenanceMode || !c.AllowHTTP {
t.Errorf("bool fields = debug %v maintenance %v allow_http %v, want all true",
c.Debug, c.MaintenanceMode, c.AllowHTTP)
}
if c.StateDir != "/tmp/pixa-test-state" {
t.Errorf("StateDir = %q, want /tmp/pixa-test-state", c.StateDir)
}
if c.DBURL != "file:/tmp/pixa-test-state/other.sqlite3" {
t.Errorf("DBURL = %q, want explicit value", c.DBURL)
}
if len(c.AllowlistHosts) != 2 || c.AllowlistHosts[0] != testHostS3 ||
c.AllowlistHosts[1] != ".example.com" {
t.Errorf("AllowlistHosts = %v, want [s3.sneak.cloud .example.com]",
c.AllowlistHosts)
}
if c.UpstreamConnectionsPerHost != 5 {
t.Errorf("UpstreamConnectionsPerHost = %d, want 5", c.UpstreamConnectionsPerHost)
}
if c.SentryDSN != "https://abc123@sentry.example.com/42" {
t.Errorf("SentryDSN = %q, want explicit value", c.SentryDSN)
}
if c.MetricsUsername != "metricsuser" || c.MetricsPassword != "metricspass" {
t.Errorf("metrics = %q/%q, want metricsuser/metricspass",
c.MetricsUsername, c.MetricsPassword)
}
}
func TestCommaSeparatedAllowlistStillSupported(t *testing.T) {
t.Parallel()
yamlContent := signingKeyLine +
`allowlist_hosts: "s3.sneak.cloud, sneak.berlin"
`
c, err := configFromYAML(t, yamlContent)
if err != nil {
t.Fatalf("comma-separated allowlist should load, got error: %v", err)
}
if len(c.AllowlistHosts) != 2 || c.AllowlistHosts[0] != testHostS3 ||
c.AllowlistHosts[1] != "sneak.berlin" {
t.Errorf("AllowlistHosts = %v, want [s3.sneak.cloud sneak.berlin]",
c.AllowlistHosts)
}
}
// runAbortCases asserts that each case's config aborts startup with an
// error message mentioning every expected substring.
func runAbortCases(t *testing.T, cases []abortCase) {
t.Helper()
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
c, err := configFromYAML(t, tc.yaml)
if err == nil {
t.Fatalf("config with %s must abort startup, got config: %+v", tc.name, c)
}
t.Logf("got expected error: %v", err)
for _, want := range tc.wantErrSubstrings {
if !strings.Contains(err.Error(), want) {
t.Errorf("error %q does not mention %q", err.Error(), want)
}
}
})
}
}
// invalidScalarValueCases are configs where a scalar key is explicitly
// set to an unparseable or out-of-range value; each must abort startup
// naming the offending key, never silently fall back to the default.
func invalidScalarValueCases() []abortCase {
return []abortCase{
{
name: "port not a number",
yaml: signingKeyLine + "port: banana\n",
wantErrSubstrings: []string{keyPort, "banana"},
},
{
name: "port zero",
yaml: signingKeyLine + "port: 0\n",
wantErrSubstrings: []string{keyPort, "0"},
},
{
name: "port above 65535",
yaml: signingKeyLine + "port: 99999\n",
wantErrSubstrings: []string{keyPort, "99999"},
},
{
name: "port fractional",
yaml: signingKeyLine + "port: 8080.5\n",
wantErrSubstrings: []string{keyPort, "8080.5"},
},
{
name: "debug not a bool",
yaml: signingKeyLine + "debug: notabool\n",
wantErrSubstrings: []string{keyDebug, "notabool"},
},
{
name: "maintenance_mode not a bool",
yaml: signingKeyLine + "maintenance_mode: sometimes\n",
wantErrSubstrings: []string{keyMaintenanceMode, "sometimes"},
},
{
name: "allow_http numeric",
yaml: signingKeyLine + "allow_http: 2\n",
wantErrSubstrings: []string{keyAllowHTTP, "2"},
},
{
name: "upstream_connections_per_host zero",
yaml: signingKeyLine + "upstream_connections_per_host: 0\n",
wantErrSubstrings: []string{keyUpstreamConnectionsPerHost, "0"},
},
{
name: "upstream_connections_per_host negative",
yaml: signingKeyLine + "upstream_connections_per_host: -3\n",
wantErrSubstrings: []string{keyUpstreamConnectionsPerHost, "-3"},
},
{
name: "upstream_connections_per_host not a number",
yaml: signingKeyLine + "upstream_connections_per_host: many\n",
wantErrSubstrings: []string{keyUpstreamConnectionsPerHost, "many"},
},
}
}
// invalidHostAndCredentialCases are configs where allowlist_hosts,
// signing_key, state_dir, sentry_dsn, or metrics is explicitly set to
// an invalid value; each must abort startup naming the offending key.
func invalidHostAndCredentialCases() []abortCase {
return []abortCase{
{
name: "allowlist host with scheme",
yaml: signingKeyLine + "allowlist_hosts:\n - https://example.com\n",
wantErrSubstrings: []string{
keyAllowlistHosts, "https://example.com",
},
},
{
name: "allowlist host with path",
yaml: signingKeyLine + "allowlist_hosts:\n - example.com/images\n",
wantErrSubstrings: []string{
keyAllowlistHosts, "example.com/images",
},
},
{
name: "allowlist host with whitespace",
yaml: signingKeyLine + "allowlist_hosts:\n - \"exa mple.com\"\n",
wantErrSubstrings: []string{keyAllowlistHosts, "exa mple.com"},
},
{
name: "allowlist entry not a string",
yaml: signingKeyLine + "allowlist_hosts:\n - 123\n",
wantErrSubstrings: []string{keyAllowlistHosts, "123"},
},
{
name: "allowlist not a list",
yaml: signingKeyLine + "allowlist_hosts:\n key: value\n",
wantErrSubstrings: []string{keyAllowlistHosts},
},
{
name: "signing_key too short",
yaml: "signing_key: short\n",
wantErrSubstrings: []string{keySigningKey},
},
{
name: "signing_key missing",
yaml: "port: 8080\n",
wantErrSubstrings: []string{keySigningKey},
},
{
name: "state_dir explicitly empty",
yaml: signingKeyLine + "state_dir: \"\"\n",
wantErrSubstrings: []string{keyStateDir},
},
{
name: "sentry_dsn not a URL",
yaml: signingKeyLine + "sentry_dsn: \"not a url\"\n",
wantErrSubstrings: []string{keySentryDSN, "not a url"},
},
{
name: "metrics username without password",
yaml: signingKeyLine + "metrics:\n username: bob\n",
wantErrSubstrings: []string{keyMetrics},
},
{
name: "metrics password without username",
yaml: signingKeyLine + "metrics:\n password: hunter2\n",
wantErrSubstrings: []string{keyMetrics},
},
}
}
// TestSetButInvalidValueAbortsStartup verifies the no-silent-fallback
// rule: a key that is explicitly set to an unparseable or out-of-range
// value must produce a startup error naming the offending key, never
// silently fall back to the default.
func TestSetButInvalidValueAbortsStartup(t *testing.T) {
t.Parallel()
runAbortCases(t, append(
invalidScalarValueCases(), invalidHostAndCredentialCases()...))
}
// explicitNullValueCases are configs where a key is explicitly set to
// null (including the bare "key:" form and the "~" alias); each must
// abort startup naming the key.
func explicitNullValueCases() []abortCase {
return []abortCase{
{
name: "port explicit null",
yaml: signingKeyLine + "port: null\n",
wantErrSubstrings: []string{keyPort, nullValueText},
},
{
name: "port bare key no value",
yaml: signingKeyLine + "port:\n",
wantErrSubstrings: []string{keyPort, nullValueText},
},
{
name: "debug tilde null",
yaml: signingKeyLine + "debug: ~\n",
wantErrSubstrings: []string{keyDebug, nullValueText},
},
{
name: "maintenance_mode null",
yaml: signingKeyLine + "maintenance_mode: null\n",
wantErrSubstrings: []string{keyMaintenanceMode, nullValueText},
},
{
name: "allow_http null",
yaml: signingKeyLine + "allow_http: null\n",
wantErrSubstrings: []string{keyAllowHTTP, nullValueText},
},
{
name: "state_dir null",
yaml: signingKeyLine + "state_dir: null\n",
wantErrSubstrings: []string{keyStateDir, nullValueText},
},
{
name: "db_url null",
yaml: signingKeyLine + "db_url: null\n",
wantErrSubstrings: []string{keyDBURL, nullValueText},
},
{
name: "sentry_dsn null",
yaml: signingKeyLine + "sentry_dsn: null\n",
wantErrSubstrings: []string{keySentryDSN, nullValueText},
},
{
name: "upstream_connections_per_host null",
yaml: signingKeyLine + "upstream_connections_per_host: null\n",
wantErrSubstrings: []string{keyUpstreamConnectionsPerHost, nullValueText},
},
{
name: "allowlist_hosts null",
yaml: signingKeyLine + "allowlist_hosts: null\n",
wantErrSubstrings: []string{keyAllowlistHosts, nullValueText},
},
{
name: "signing_key null",
yaml: "signing_key: null\n",
wantErrSubstrings: []string{keySigningKey, nullValueText},
},
{
name: "metrics null",
yaml: signingKeyLine + "metrics: null\n",
wantErrSubstrings: []string{keyMetrics, nullValueText},
},
{
name: "metrics subkeys null",
yaml: signingKeyLine + "metrics:\n username: null\n password: null\n",
wantErrSubstrings: []string{
keyMetricsUsername, keyMetricsPassword, nullValueText,
},
},
}
}
// TestExplicitNullValueAbortsStartup verifies that a key explicitly
// set to null (including the bare "key:" form and the "~" alias) aborts
// startup naming the key. An explicit null is a SET value: it must
// never silently fall back to the default the way an omitted key does.
func TestExplicitNullValueAbortsStartup(t *testing.T) {
t.Parallel()
runAbortCases(t, explicitNullValueCases())
}
// TestExplicitlyEmptyDBURLAbortsStartup verifies that db_url set to an
// empty string aborts startup: the derived file:...state.sqlite3 URL is
// a default, and defaults apply only to omitted keys. This matches
// state_dir, where an explicitly empty value already aborts.
func TestExplicitlyEmptyDBURLAbortsStartup(t *testing.T) {
t.Parallel()
yamlContent := signingKeyLine + "db_url: \"\"\n"
c, err := configFromYAML(t, yamlContent)
if err == nil {
t.Fatalf("explicitly empty db_url must abort startup, got config: %+v", c)
}
t.Logf("got expected error: %v", err)
if !strings.Contains(err.Error(), keyDBURL) {
t.Errorf("error %q does not name the offending key db_url", err.Error())
}
}
// TestAllowlistHostsRejectsDotOnlyEntries verifies that entries with no
// hostname labels are rejected. The allowlist matcher treats a leading
// dot as a suffix pattern, so a bare "." entry would match any upstream
// host written in FQDN trailing-dot form (e.g. evil.com.) and
// effectively disable URL signing with a single character.
func TestAllowlistHostsRejectsDotOnlyEntries(t *testing.T) {
t.Parallel()
for _, entry := range []string{".", ".."} {
t.Run(entry, func(t *testing.T) {
t.Parallel()
yamlContent := signingKeyLine +
"allowlist_hosts:\n - \"" + entry + "\"\n"
c, err := configFromYAML(t, yamlContent)
if err == nil {
t.Fatalf("allowlist entry %q must abort startup, got config: %+v",
entry, c)
}
t.Logf("got expected error: %v", err)
if !strings.Contains(err.Error(), keyAllowlistHosts) {
t.Errorf("error %q does not name the offending key allowlist_hosts",
err.Error())
}
})
}
}
func TestUnknownTopLevelKeyAbortsStartup(t *testing.T) {
t.Parallel()
yamlContent := signingKeyLine + `whitelist_hosts:
- example.com
`
c, err := configFromYAML(t, yamlContent)
if err == nil {
t.Fatalf("config with unknown key must abort startup, got config: %+v", c)
}
t.Logf("got expected error: %v", err)
if !strings.Contains(err.Error(), "whitelist_hosts") {
t.Errorf("error %q does not name the unknown key whitelist_hosts", err.Error())
}
}
func TestUnknownMetricsSubkeyAbortsStartup(t *testing.T) {
t.Parallel()
yamlContent := signingKeyLine + `metrics:
username: bob
password: hunter2
port: 9100
`
c, err := configFromYAML(t, yamlContent)
if err == nil {
t.Fatalf("config with unknown metrics subkey must abort startup, got config: %+v", c)
}
t.Logf("got expected error: %v", err)
if !strings.Contains(err.Error(), "metrics.port") {
t.Errorf("error %q does not name the unknown key metrics.port", err.Error())
}
}
func TestEnvSectionIsPermitted(t *testing.T) {
t.Parallel()
yamlContent := signingKeyLine + `env:
PIXA_TEST_ENV_INJECTION: injected
`
_, err := configFromYAML(t, yamlContent)
if err != nil {
t.Fatalf(
"env section must be permitted (smartconfig consumes it), got error: %v",
err)
}
}
func TestMalformedConfigFileAbortsStartup(t *testing.T) {
tmpDir := t.TempDir()
configPath := filepath.Join(tmpDir, "config.yml")
err := os.WriteFile(configPath, []byte("port: [unclosed\n"), 0o600)
if err != nil {
t.Fatalf("failed to write malformed config: %v", err)
}
// loadConfigFile falls through to the relative config.yml candidate;
// the appname is chosen so no /etc or $HOME candidate can exist.
t.Setenv("PIXA_CONFIG_PATH", "")
t.Chdir(tmpDir)
log := slog.New(slog.DiscardHandler)
sc, err := loadConfigFile(log, "pixa-test-nonexistent-app")
if err == nil {
t.Fatalf("malformed config file must abort startup, got config: %v", sc)
}
t.Logf("got expected error: %v", err)
}
func TestEnsureStateDirCreatesDirectory(t *testing.T) {
t.Parallel()
stateDir := filepath.Join(t.TempDir(), "nested", "state")
c := &Config{StateDir: stateDir}
err := c.ensureStateDirWritable()
if err != nil {
t.Fatalf("creatable state_dir must validate, got error: %v", err)
}
info, err := os.Stat(stateDir)
if err != nil || !info.IsDir() {
t.Fatalf("state_dir was not created: info=%v err=%v", info, err)
}
}
func TestEnsureStateDirFailsOnUncreatablePath(t *testing.T) {
t.Parallel()
// A path below /dev/null can never be created, even when running
// as root (as in the Docker build).
c := &Config{StateDir: "/dev/null/pixa-state"}
err := c.ensureStateDirWritable()
if err == nil {
t.Fatal("uncreatable state_dir must abort startup, got nil error")
}
t.Logf("got expected error: %v", err)
if !strings.Contains(err.Error(), keyStateDir) {
t.Errorf("error %q does not name the offending key state_dir", err.Error())
}
}

View File

@@ -3,17 +3,12 @@
-- Source content blobs -- Source content blobs
-- Files stored at: cache/src-content/<ab>/<cd>/<sha256> -- Files stored at: cache/src-content/<ab>/<cd>/<sha256>
-- last_accessed_at is NULL until the first LRU touch; eviction falls
-- back to fetched_at for rows that have never been touched.
CREATE TABLE IF NOT EXISTS source_content ( CREATE TABLE IF NOT EXISTS source_content (
content_hash TEXT PRIMARY KEY, content_hash TEXT PRIMARY KEY,
content_type TEXT NOT NULL, content_type TEXT NOT NULL,
size_bytes INTEGER NOT NULL, size_bytes INTEGER NOT NULL,
fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP, fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP
last_accessed_at DATETIME
); );
CREATE INDEX IF NOT EXISTS idx_source_content_last_accessed
ON source_content(last_accessed_at);
-- Source URL metadata - maps URLs to content hashes -- Source URL metadata - maps URLs to content hashes
-- JSON stored at: cache/src-metadata/<hostname>/<path_hash>.json -- JSON stored at: cache/src-metadata/<hostname>/<path_hash>.json
@@ -39,22 +34,6 @@ CREATE INDEX IF NOT EXISTS idx_source_meta_path_hash ON source_metadata(path_has
CREATE INDEX IF NOT EXISTS idx_source_meta_expires ON source_metadata(expires_at); CREATE INDEX IF NOT EXISTS idx_source_meta_expires ON source_metadata(expires_at);
CREATE INDEX IF NOT EXISTS idx_source_meta_content_hash ON source_metadata(content_hash); CREATE INDEX IF NOT EXISTS idx_source_meta_content_hash ON source_metadata(content_hash);
-- Processed variant blobs
-- Files stored at: cache/variants/<ab>/<cd>/<cache_key> (plus a .meta
-- sidecar with the content type). Tracked here (like source content
-- blobs above) so total cache usage can be computed with a SUM query,
-- never a directory scan, and so LRU eviction has a timestamp to order
-- on.
CREATE TABLE IF NOT EXISTS variant_content (
cache_key TEXT PRIMARY KEY,
size_bytes INTEGER NOT NULL,
content_type TEXT NOT NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
last_accessed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_variant_content_last_accessed
ON variant_content(last_accessed_at);
-- Output/transformed content blobs -- Output/transformed content blobs
-- Files stored at: cache/dst-content/<ab>/<cd>/<sha256> -- Files stored at: cache/dst-content/<ab>/<cd>/<sha256>
CREATE TABLE IF NOT EXISTS output_content ( CREATE TABLE IF NOT EXISTS output_content (

View File

@@ -51,24 +51,9 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
} }
lc.Append(fx.Hook{ lc.Append(fx.Hook{
// The eviction goroutine must outlive OnStart, so it cannot
// inherit this hook's context. It makes its own instead, which
// leaves it uncancellable: an in-flight pass runs to completion
// during OnStop regardless of the shutdown deadline. Making the
// loop cancellable changes shutdown semantics and is tracked
// separately in issue #102, rather than being folded into the
// lint-conformance change that surfaced it.
//nolint:contextcheck // see issue #102
OnStart: func(_ context.Context) error { OnStart: func(_ context.Context) error {
return s.initImageService() return s.initImageService()
}, },
OnStop: func(_ context.Context) error {
if s.imgCache != nil {
s.imgCache.StopEviction()
}
return nil
},
}) })
return s, nil return s, nil
@@ -76,15 +61,11 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
// initImageService initializes the image cache and service. // initImageService initializes the image cache and service.
func (s *Handlers) initImageService() error { func (s *Handlers) initImageService() error {
// Create the cache. cache_max_bytes: 0 disables the disk cache // Create the cache
// entirely; any other value is the eviction limit in bytes.
cache, err := imgcache.NewCache(s.db.DB(), imgcache.CacheConfig{ cache, err := imgcache.NewCache(s.db.DB(), imgcache.CacheConfig{
StateDir: s.config.StateDir, StateDir: s.config.StateDir,
CacheTTL: imgcache.DefaultCacheTTL, CacheTTL: imgcache.DefaultCacheTTL,
NegativeTTL: imgcache.DefaultNegativeTTL, NegativeTTL: imgcache.DefaultNegativeTTL,
MaxBytes: s.config.CacheMaxBytes,
DisableDiskCache: s.config.CacheMaxBytes == 0,
Logger: s.log,
}) })
if err != nil { if err != nil {
return err return err
@@ -92,10 +73,6 @@ func (s *Handlers) initImageService() error {
s.imgCache = cache s.imgCache = cache
// Background eviction: startup reconciliation, then periodic and
// write-pressure passes. No-op when the disk cache is disabled.
cache.StartEviction(imgcache.DefaultEvictionInterval)
// Create the fetcher config // Create the fetcher config
fetcherCfg := httpfetcher.DefaultConfig() fetcherCfg := httpfetcher.DefaultConfig()
fetcherCfg.AllowHTTP = s.config.AllowHTTP fetcherCfg.AllowHTTP = s.config.AllowHTTP

View File

@@ -66,7 +66,6 @@ func (s *Handlers) HandleImageEnc() http.HandlerFunc {
return return
} }
defer func() { _ = resp.Content.Close() }() defer func() { _ = resp.Content.Close() }()
// Set response headers // Set response headers

View File

@@ -198,7 +198,6 @@ func (f *HTTPFetcher) Fetch(ctx context.Context, url string) (*FetchResult, erro
// If we fail before returning a result, release the slot // If we fail before returning a result, release the slot
success := false success := false
defer func() { defer func() {
if !success { if !success {
<-sem <-sem

View File

@@ -237,7 +237,6 @@ func TestMockFetcher_FetchesFile(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("Fetch() error = %v", err) t.Fatalf("Fetch() error = %v", err)
} }
defer func() { _ = result.Content.Close() }() defer func() { _ = result.Content.Close() }()
if result.ContentType != contentTypeJPEG { if result.ContentType != contentTypeJPEG {

View File

@@ -2,16 +2,12 @@ package imgcache
import ( import (
"context" "context"
"crypto/sha256"
"database/sql" "database/sql"
"encoding/hex"
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"io" "io"
"log/slog"
"path/filepath" "path/filepath"
"sync"
"time" "time"
"sneak.berlin/go/pixa/internal/httpfetcher" "sneak.berlin/go/pixa/internal/httpfetcher"
@@ -31,22 +27,6 @@ type CacheConfig struct {
StateDir string StateDir string
CacheTTL time.Duration CacheTTL time.Duration
NegativeTTL time.Duration NegativeTTL time.Duration
// MaxBytes is the disk cache size limit in bytes that eviction
// enforces. Zero means no limit is enforced (no eviction). The
// config layer supplies the computed default when the operator
// omits cache_max_bytes.
MaxBytes int64
// DisableDiskCache turns the disk cache off entirely: no cache
// directories are created, lookups always miss, stores are
// no-ops, and no eviction machinery runs. The config layer sets
// this when the operator configures cache_max_bytes: 0.
DisableDiskCache bool
// Logger receives accounting and eviction log output. A nil
// Logger means slog.Default().
Logger *slog.Logger
} }
// variantMeta stores content type for fast cache hits without reading .meta file. // variantMeta stores content type for fast cache hits without reading .meta file.
@@ -62,61 +42,14 @@ type Cache struct {
variants *VariantStorage // processed variants by cache key variants *VariantStorage // processed variants by cache key
srcMetadata *MetadataStorage // source metadata by host/path srcMetadata *MetadataStorage // source metadata by host/path
config CacheConfig config CacheConfig
log *slog.Logger
// disabled means the disk cache is turned off entirely: lookups
// always miss, stores are no-ops, and no eviction runs.
disabled bool
// Eviction machinery. The channels are created in NewCache so
// stores can signal write pressure without racing StartEviction.
evictionPressure chan struct{}
evictionStop chan struct{}
evictionDone chan struct{}
evictionStarted bool
evictionStopOnce sync.Once
// In-memory cache of variant metadata (content type, size) to avoid // In-memory cache of variant metadata (content type, size) to avoid
// reading .meta files // reading .meta files
metaCache map[VariantKey]variantMeta metaCache map[VariantKey]variantMeta
// contentLocks serializes StoreSource and evictSourceBlob per
// content hash, closing the race window between an eviction's row
// deletion and its file unlink against a concurrent store of
// identical content.
contentLocks *contentLock
// evictSourceBlobTestHook, when set, is invoked by evictSourceBlob
// after its row-deletion transaction commits and before the
// content file is unlinked. It exists solely so tests can
// deterministically pause inside that window to exercise
// concurrent stores against it; production code leaves it nil.
evictSourceBlobTestHook func(ContentHash)
} }
// NewCache creates a new cache instance. // NewCache creates a new cache instance.
func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) { func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) {
log := config.Logger
if log == nil {
log = slog.Default()
}
c := &Cache{
db: db,
config: config,
log: log,
disabled: config.DisableDiskCache,
evictionPressure: make(chan struct{}, 1),
evictionStop: make(chan struct{}),
evictionDone: make(chan struct{}),
metaCache: make(map[VariantKey]variantMeta),
contentLocks: newContentLock(),
}
if c.disabled {
return c, nil
}
srcContent, err := NewContentStorage( srcContent, err := NewContentStorage(
filepath.Join(config.StateDir, "cache", "sources"), filepath.Join(config.StateDir, "cache", "sources"),
) )
@@ -138,11 +71,14 @@ func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) {
return nil, fmt.Errorf("failed to create source metadata storage: %w", err) return nil, fmt.Errorf("failed to create source metadata storage: %w", err)
} }
c.srcContent = srcContent return &Cache{
c.variants = variants db: db,
c.srcMetadata = srcMetadata srcContent: srcContent,
variants: variants,
return c, nil srcMetadata: srcMetadata,
config: config,
metaCache: make(map[VariantKey]variantMeta),
}, nil
} }
// LookupResult contains the result of a cache lookup. // LookupResult contains the result of a cache lookup.
@@ -154,15 +90,12 @@ type LookupResult struct {
CacheStatus CacheStatus CacheStatus CacheStatus
} }
// Lookup checks if a processed variant exists on disk. Hits touch the // Lookup checks if a processed variant exists on disk (no DB access for hits).
// variant's LRU timestamp; a disabled cache always misses. func (c *Cache) Lookup(_ context.Context, req *ImageRequest) (*LookupResult, error) {
func (c *Cache) Lookup(ctx context.Context, req *ImageRequest) (*LookupResult, error) {
cacheKey := CacheKey(req) cacheKey := CacheKey(req)
// Check variant storage directly - no DB needed for cache hits // Check variant storage directly - no DB needed for cache hits
if !c.disabled && c.variants.Exists(cacheKey) { if c.variants.Exists(cacheKey) {
c.touchVariant(ctx, cacheKey)
return &LookupResult{ return &LookupResult{
Hit: true, Hit: true,
CacheKey: cacheKey, CacheKey: cacheKey,
@@ -179,52 +112,18 @@ func (c *Cache) Lookup(ctx context.Context, req *ImageRequest) (*LookupResult, e
// GetVariant returns a reader, size, and content type for a cached variant. // GetVariant returns a reader, size, and content type for a cached variant.
func (c *Cache) GetVariant(cacheKey VariantKey) (io.ReadCloser, int64, string, error) { func (c *Cache) GetVariant(cacheKey VariantKey) (io.ReadCloser, int64, string, error) {
if c.disabled {
return nil, 0, "", ErrNotFound
}
return c.variants.LoadWithMeta(cacheKey) return c.variants.LoadWithMeta(cacheKey)
} }
// StoreSource stores fetched source content and metadata. On a // StoreSource stores fetched source content and metadata.
// disabled cache it is a no-op returning an empty hash.
func (c *Cache) StoreSource( func (c *Cache) StoreSource(
ctx context.Context, ctx context.Context,
req *ImageRequest, req *ImageRequest,
content io.Reader, content io.Reader,
result *httpfetcher.FetchResult, result *httpfetcher.FetchResult,
) (ContentHash, error) { ) (ContentHash, error) {
if c.disabled { // Store content
return "", nil contentHash, size, err := c.srcContent.Store(content)
}
// Hash the content ourselves (rather than via srcContent.Store,
// which would hash internally) so the content hash is known before
// any file or database work happens: that lets the entire store be
// serialized, per hash, against a concurrent eviction of the same
// content below.
data, err := io.ReadAll(content)
if err != nil {
return "", fmt.Errorf("failed to read source content: %w", err)
}
sum := sha256.Sum256(data)
contentHash := ContentHash(hex.EncodeToString(sum[:]))
// Hold the content hash's lock for the whole store operation. A
// concurrent eviction of this exact hash (the real SHA-256 dedup
// case: a different source path whose bytes hash identically)
// deletes the accounting rows and unlinks the file inside the same
// lock, so the two can never interleave: either this store
// completes first (and a subsequent eviction removes it together
// with its rows and file, correctly), or eviction completes first
// (and this store finds the file already gone and recreates it
// fresh) — never a fresh row left pointing at a file eviction is
// mid-unlink on.
unlock := c.contentLocks.Lock(string(contentHash))
defer unlock()
size, err := c.srcContent.StoreHashed(contentHash, data)
if err != nil { if err != nil {
return "", fmt.Errorf("failed to store source content: %w", err) return "", fmt.Errorf("failed to store source content: %w", err)
} }
@@ -281,56 +180,23 @@ func (c *Cache) StoreSource(
// A failure here is non-fatal; the metadata is in the database. // A failure here is non-fatal; the metadata is in the database.
_ = c.srcMetadata.Store(req.SourceHost, pathHash, meta) _ = c.srcMetadata.Store(req.SourceHost, pathHash, meta)
c.notifyWritePressure()
return contentHash, nil return contentHash, nil
} }
// StoreVariant stores a processed variant by its cache key and records // StoreVariant stores a processed variant by its cache key.
// it in the size accounting. On a disabled cache it is a no-op. The
// accounting insert is best-effort (the startup reconciliation pass
// adopts any variant file that misses its accounting row).
func (c *Cache) StoreVariant( func (c *Cache) StoreVariant(
ctx context.Context, cacheKey VariantKey, content io.Reader, contentType string, cacheKey VariantKey, content io.Reader, contentType string,
) error { ) error {
if c.disabled { _, err := c.variants.Store(cacheKey, content, contentType)
return nil
}
size, err := c.variants.Store(cacheKey, content, contentType) return err
if err != nil {
return err
}
_, err = c.db.ExecContext(ctx, `
INSERT INTO variant_content (cache_key, size_bytes, content_type)
VALUES (?, ?, ?)
ON CONFLICT(cache_key) DO UPDATE SET
size_bytes = excluded.size_bytes,
content_type = excluded.content_type,
last_accessed_at = CURRENT_TIMESTAMP
`, string(cacheKey), size, contentType)
if err != nil {
c.log.Warn("failed to record variant in size accounting",
"cache_key", cacheKey, "error", err)
}
c.notifyWritePressure()
return nil
} }
// LookupSource checks if we have cached source content for a request. // LookupSource checks if we have cached source content for a request.
// Returns the content hash and content type if found, or empty values // Returns the content hash and content type if found, or empty values if not.
// if not. Hits touch the blob's LRU timestamp; a disabled cache always
// reports no cached source.
func (c *Cache) LookupSource( func (c *Cache) LookupSource(
ctx context.Context, req *ImageRequest, ctx context.Context, req *ImageRequest,
) (ContentHash, string, error) { ) (ContentHash, string, error) {
if c.disabled {
return "", "", nil
}
var hashStr, contentType string var hashStr, contentType string
err := c.db.QueryRowContext(ctx, ` err := c.db.QueryRowContext(ctx, `
@@ -353,8 +219,6 @@ func (c *Cache) LookupSource(
return "", "", nil return "", "", nil
} }
c.touchSourceContent(ctx, contentHash)
return contentHash, contentType, nil return contentHash, contentType, nil
} }
@@ -401,10 +265,6 @@ func (c *Cache) GetSourceMetadataID(
// GetSourceContent returns a reader for cached source content by its hash. // GetSourceContent returns a reader for cached source content by its hash.
func (c *Cache) GetSourceContent(contentHash ContentHash) (io.ReadCloser, error) { func (c *Cache) GetSourceContent(contentHash ContentHash) (io.ReadCloser, error) {
if c.disabled {
return nil, ErrNotFound
}
return c.srcContent.Load(contentHash) return c.srcContent.Load(contentHash)
} }
@@ -480,32 +340,6 @@ func (c *Cache) IncrementStats(ctx context.Context, hit bool, fetchBytes int64)
} }
} }
// touchVariant updates the LRU timestamp of a variant, best-effort:
// a failed touch only makes the entry look colder to eviction.
func (c *Cache) touchVariant(ctx context.Context, cacheKey VariantKey) {
_, err := c.db.ExecContext(ctx, `
UPDATE variant_content SET last_accessed_at = CURRENT_TIMESTAMP
WHERE cache_key = ?
`, string(cacheKey))
if err != nil {
c.log.Debug("failed to touch variant LRU timestamp",
"cache_key", cacheKey, "error", err)
}
}
// touchSourceContent updates the LRU timestamp of a source content
// blob, best-effort: a failed touch only makes the blob look colder.
func (c *Cache) touchSourceContent(ctx context.Context, contentHash ContentHash) {
_, err := c.db.ExecContext(ctx, `
UPDATE source_content SET last_accessed_at = CURRENT_TIMESTAMP
WHERE content_hash = ?
`, string(contentHash))
if err != nil {
c.log.Debug("failed to touch source content LRU timestamp",
"content_hash", contentHash, "error", err)
}
}
// checkNegativeCache checks if a request is in the negative cache. // checkNegativeCache checks if a request is in the negative cache.
func (c *Cache) checkNegativeCache( func (c *Cache) checkNegativeCache(
ctx context.Context, req *ImageRequest, ctx context.Context, req *ImageRequest,

View File

@@ -160,7 +160,7 @@ func TestCache_StoreAndLookup(t *testing.T) {
sourceContent := []byte("fake jpeg data") sourceContent := []byte("fake jpeg data")
fetchResult := &httpfetcher.FetchResult{ fetchResult := &httpfetcher.FetchResult{
ContentType: testContentTypeJPEG, ContentType: testContentTypeJPEG,
Headers: map[string][]string{testHeaderContentType: {testContentTypeJPEG}}, Headers: map[string][]string{"Content-Type": {testContentTypeJPEG}},
} }
contentHash, err := cache.StoreSource( contentHash, err := cache.StoreSource(
@@ -177,8 +177,7 @@ func TestCache_StoreAndLookup(t *testing.T) {
cacheKey := CacheKey(req) cacheKey := CacheKey(req)
outputContent := []byte("fake webp data") outputContent := []byte("fake webp data")
err = cache.StoreVariant( err = cache.StoreVariant(cacheKey, bytes.NewReader(outputContent), "image/webp")
t.Context(), cacheKey, bytes.NewReader(outputContent), "image/webp")
if err != nil { if err != nil {
t.Fatalf("StoreVariant() error = %v", err) t.Fatalf("StoreVariant() error = %v", err)
} }
@@ -296,8 +295,7 @@ func TestCache_VariantLookup(t *testing.T) {
cacheKey := CacheKey(req) cacheKey := CacheKey(req)
outputContent := []byte("output data") outputContent := []byte("output data")
err := cache.StoreVariant( err := cache.StoreVariant(cacheKey, bytes.NewReader(outputContent), "image/webp")
t.Context(), cacheKey, bytes.NewReader(outputContent), "image/webp")
if err != nil { if err != nil {
t.Fatalf("StoreVariant() error = %v", err) t.Fatalf("StoreVariant() error = %v", err)
} }
@@ -346,8 +344,7 @@ func TestCache_GetVariant_ReturnsContentType(t *testing.T) {
cacheKey := CacheKey(req) cacheKey := CacheKey(req)
outputContent := []byte("output webp data") outputContent := []byte("output webp data")
err := cache.StoreVariant( err := cache.StoreVariant(cacheKey, bytes.NewReader(outputContent), "image/webp")
t.Context(), cacheKey, bytes.NewReader(outputContent), "image/webp")
if err != nil { if err != nil {
t.Fatalf("StoreVariant() error = %v", err) t.Fatalf("StoreVariant() error = %v", err)
} }
@@ -398,8 +395,7 @@ func TestCache_GetVariant(t *testing.T) {
cacheKey := CacheKey(req) cacheKey := CacheKey(req)
outputContent := []byte("the actual output content") outputContent := []byte("the actual output content")
err := cache.StoreVariant( err := cache.StoreVariant(cacheKey, bytes.NewReader(outputContent), "image/webp")
t.Context(), cacheKey, bytes.NewReader(outputContent), "image/webp")
if err != nil { if err != nil {
t.Fatalf("StoreVariant() error = %v", err) t.Fatalf("StoreVariant() error = %v", err)
} }

View File

@@ -1,64 +0,0 @@
package imgcache
import "sync"
// contentLock provides per-key mutual exclusion for content-hash keyed
// operations. StoreSource and evictSourceBlob each hold a content
// hash's lock for the full duration of their file-plus-accounting-row
// work, so a store and an eviction racing on identical content bytes
// (the real SHA-256 content-addressed dedup case, not a contrived one)
// can never interleave: the unlink of an evicted blob's file can never
// race the creation of a fresh database row for a concurrently
// re-stored copy of the same content. Entries are removed once no
// goroutine holds or is waiting for them, so a long-running process
// does not accumulate memory proportional to the number of distinct
// content hashes it has ever seen.
type contentLock struct {
mu sync.Mutex
entries map[string]*contentLockEntry
}
// contentLockEntry is one key's exclusion lock plus a count of
// goroutines currently holding or waiting to acquire it, used to know
// when it is safe to remove the entry from the map.
type contentLockEntry struct {
mu sync.Mutex
count int
}
// newContentLock creates an empty contentLock.
func newContentLock() *contentLock {
return &contentLock{entries: make(map[string]*contentLockEntry)}
}
// Lock acquires exclusive access for key, blocking until it is
// available, and returns a function that releases it. The caller must
// invoke the returned function exactly once to release the lock.
func (c *contentLock) Lock(key string) func() {
c.mu.Lock()
entry, ok := c.entries[key]
if !ok {
entry = &contentLockEntry{}
c.entries[key] = entry
}
entry.count++
c.mu.Unlock()
entry.mu.Lock()
return func() {
entry.mu.Unlock()
c.mu.Lock()
entry.count--
if entry.count == 0 {
delete(c.entries, key)
}
c.mu.Unlock()
}
}

View File

@@ -1,142 +0,0 @@
package imgcache
import (
"sync"
"sync/atomic"
"testing"
"time"
)
// TestContentLockExcludesSameKey verifies that two goroutines locking
// the same key never run their critical sections concurrently.
func TestContentLockExcludesSameKey(t *testing.T) {
t.Parallel()
lock := newContentLock()
var (
active atomic.Int32
maxSeen int32
wg sync.WaitGroup
)
const goroutines = 20
wg.Add(goroutines)
for range goroutines {
go func() {
defer wg.Done()
unlock := lock.Lock("same-key")
defer unlock()
n := active.Add(1)
for {
seen := atomic.LoadInt32(&maxSeen)
if n <= seen || atomic.CompareAndSwapInt32(&maxSeen, seen, n) {
break
}
}
time.Sleep(time.Millisecond)
active.Add(-1)
}()
}
wg.Wait()
if maxSeen != 1 {
t.Errorf("max concurrent holders of the same key = %d, want 1", maxSeen)
}
}
// TestContentLockAllowsDifferentKeys verifies that locking distinct
// keys does not serialize unrelated work: all goroutines must be able
// to enter their critical sections at once, proven by every one of
// them reaching the rendezvous point before any is allowed to
// proceed.
func TestContentLockAllowsDifferentKeys(t *testing.T) {
t.Parallel()
lock := newContentLock()
const goroutines = 20
var (
wg sync.WaitGroup
inside atomic.Int32
reached = make(chan struct{}, goroutines)
)
wg.Add(goroutines)
release := make(chan struct{})
for i := range goroutines {
key := string(rune('a' + i))
go func() {
defer wg.Done()
unlock := lock.Lock(key)
defer unlock()
inside.Add(1)
reached <- struct{}{}
<-release
}()
}
// Every goroutine must reach the rendezvous point (i.e. acquire its
// own key's lock) without needing any other to release first. If
// keys were incorrectly serialized onto one underlying lock, only
// one would get here and this would time out.
for i := range goroutines {
select {
case <-reached:
case <-time.After(2 * time.Second):
t.Fatalf("only %d/%d goroutines locking distinct keys made progress; "+
"keys may be incorrectly serialized", i, goroutines)
}
}
if n := inside.Load(); n != goroutines {
t.Errorf("goroutines inside their critical section = %d, want %d",
n, goroutines)
}
close(release)
wg.Wait()
}
// TestContentLockRemovesEntryAfterUnlock verifies that the internal
// entries map does not grow without bound: once no goroutine holds or
// awaits a key, its entry is removed.
func TestContentLockRemovesEntryAfterUnlock(t *testing.T) {
t.Parallel()
lock := newContentLock()
unlock := lock.Lock("k")
lock.mu.Lock()
if _, ok := lock.entries["k"]; !ok {
lock.mu.Unlock()
t.Fatal("entry missing while lock is held")
}
lock.mu.Unlock()
unlock()
lock.mu.Lock()
defer lock.mu.Unlock()
if _, ok := lock.entries["k"]; ok {
t.Error("entry for key still present after the last holder unlocked")
}
}

View File

@@ -1,803 +0,0 @@
package imgcache
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"io/fs"
"os"
"path/filepath"
"strings"
"time"
)
// DefaultEvictionInterval is how often the background evictor checks
// cache usage against the configured limit, in addition to the
// write-pressure wakeups triggered by stores.
const DefaultEvictionInterval = 5 * time.Minute
// evictionBatchSize is how many LRU candidates of each class (variants
// and source blobs) one eviction pass fetches from the database.
const evictionBatchSize = 100
// staleTempFileAge is how old an orphaned temp file (left behind by a
// crashed write) must be before reconciliation removes it. Fresh temp
// files may still belong to an in-flight store.
const staleTempFileAge = time.Hour
// sqliteTimestampLayout matches SQLite's CURRENT_TIMESTAMP format, so
// timestamps written by reconciliation order correctly against ones
// written by the hot path.
const sqliteTimestampLayout = "2006-01-02 15:04:05"
// tempFilePrefix is the prefix os.CreateTemp uses for in-flight cache
// writes (".tmp-*" patterns in the storage layer).
const tempFilePrefix = ".tmp-"
// variantMetaSuffix is the sidecar suffix VariantStorage writes next
// to each variant file.
const variantMetaSuffix = ".meta"
// 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
// tracked in the database (source content blobs plus processed
// variants). It never scans the cache directories.
func (c *Cache) UsageBytes(ctx context.Context) (int64, error) {
if c.disabled {
return 0, nil
}
var total int64
err := c.db.QueryRowContext(ctx, `
SELECT (SELECT COALESCE(SUM(size_bytes), 0) FROM source_content)
+ (SELECT COALESCE(SUM(size_bytes), 0) FROM variant_content)
`).Scan(&total)
if err != nil {
return 0, fmt.Errorf("failed to compute cache usage: %w", err)
}
return total, nil
}
// evictionCandidate is one LRU eviction victim candidate: either a
// processed variant (isVariant true, identified by cacheKey) or a
// source content blob (identified by contentHash).
type evictionCandidate struct {
isVariant bool
cacheKey VariantKey
contentHash ContentHash
sizeBytes int64
lastAccessedAt string
}
// EvictToLimit evicts least-recently-used cache entries until total
// tracked usage is at or below the configured MaxBytes limit. It is a
// no-op when the cache is disabled or no limit is configured.
func (c *Cache) EvictToLimit(ctx context.Context) error {
if c.disabled || c.config.MaxBytes <= 0 {
return nil
}
for {
usage, err := c.UsageBytes(ctx)
if err != nil {
return err
}
if usage <= c.config.MaxBytes {
return nil
}
freed, err := c.evictBatch(ctx, usage-c.config.MaxBytes)
if err != nil {
return err
}
if freed == 0 {
c.log.Warn("cache eviction made no progress",
"usage_bytes", usage,
"cache_max_bytes", c.config.MaxBytes,
)
return nil
}
c.log.Info("evicted cache content",
"freed_bytes", freed,
"usage_bytes", usage-freed,
"cache_max_bytes", c.config.MaxBytes,
)
}
}
// evictBatch fetches one batch of LRU candidates across variants and
// source blobs and evicts them oldest-first until excessBytes are
// freed or the batch is exhausted. It returns the bytes freed.
func (c *Cache) evictBatch(ctx context.Context, excessBytes int64) (int64, error) {
candidates, err := c.evictionCandidates(ctx)
if err != nil {
return 0, err
}
var freed int64
for _, candidate := range candidates {
if freed >= excessBytes {
break
}
err := c.evictCandidate(ctx, candidate)
if err != nil {
c.log.Warn("failed to evict cache entry",
"cache_key", candidate.cacheKey,
"content_hash", candidate.contentHash,
"error", err,
)
continue
}
freed += candidate.sizeBytes
}
return freed, nil
}
// evictCandidate removes a single eviction victim.
func (c *Cache) evictCandidate(ctx context.Context, candidate evictionCandidate) error {
if candidate.isVariant {
return c.evictVariant(ctx, candidate.cacheKey)
}
return c.evictSourceBlob(ctx, candidate.contentHash)
}
// evictionCandidates returns up to evictionBatchSize variants and
// evictionBatchSize source blobs, merged into a single list ordered by
// last access time (oldest first).
func (c *Cache) evictionCandidates(ctx context.Context) ([]evictionCandidate, error) {
variants, err := c.variantCandidates(ctx)
if err != nil {
return nil, err
}
sources, err := c.sourceCandidates(ctx)
if err != nil {
return nil, err
}
// Merge the two lists, each already sorted oldest-first. SQLite
// CURRENT_TIMESTAMP strings compare correctly lexicographically.
merged := make([]evictionCandidate, 0, len(variants)+len(sources))
for len(variants) > 0 && len(sources) > 0 {
if variants[0].lastAccessedAt <= sources[0].lastAccessedAt {
merged = append(merged, variants[0])
variants = variants[1:]
} else {
merged = append(merged, sources[0])
sources = sources[1:]
}
}
merged = append(merged, variants...)
merged = append(merged, sources...)
return merged, nil
}
// variantCandidates returns the least recently used variants.
func (c *Cache) variantCandidates(ctx context.Context) ([]evictionCandidate, error) {
rows, err := c.db.QueryContext(ctx, `
SELECT cache_key, size_bytes, last_accessed_at
FROM variant_content
ORDER BY last_accessed_at ASC, cache_key ASC
LIMIT ?
`, evictionBatchSize)
if err != nil {
return nil, fmt.Errorf("failed to query variant eviction candidates: %w", err)
}
defer func() { _ = rows.Close() }()
var candidates []evictionCandidate
for rows.Next() {
candidate := evictionCandidate{isVariant: true}
var key string
err := rows.Scan(&key, &candidate.sizeBytes, &candidate.lastAccessedAt)
if err != nil {
return nil, fmt.Errorf("failed to scan variant candidate: %w", err)
}
candidate.cacheKey = VariantKey(key)
candidates = append(candidates, candidate)
}
err = rows.Err()
if err != nil {
return nil, fmt.Errorf("variant candidate iteration failed: %w", err)
}
return candidates, nil
}
// sourceCandidates returns the least recently used source blobs. Rows
// written before the LRU column existed fall back to fetched_at.
func (c *Cache) sourceCandidates(ctx context.Context) ([]evictionCandidate, error) {
rows, err := c.db.QueryContext(ctx, `
SELECT content_hash, size_bytes,
COALESCE(last_accessed_at, fetched_at, '1970-01-01 00:00:00') AS lru
FROM source_content
ORDER BY lru ASC, content_hash ASC
LIMIT ?
`, evictionBatchSize)
if err != nil {
return nil, fmt.Errorf("failed to query source eviction candidates: %w", err)
}
defer func() { _ = rows.Close() }()
var candidates []evictionCandidate
for rows.Next() {
var candidate evictionCandidate
var hash string
err := rows.Scan(&hash, &candidate.sizeBytes, &candidate.lastAccessedAt)
if err != nil {
return nil, fmt.Errorf("failed to scan source candidate: %w", err)
}
candidate.contentHash = ContentHash(hash)
candidates = append(candidates, candidate)
}
err = rows.Err()
if err != nil {
return nil, fmt.Errorf("source candidate iteration failed: %w", err)
}
return candidates, nil
}
// evictVariant removes one variant: accounting row first, then the
// content and .meta files, so the database never references a deleted
// 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))
if err != nil {
return fmt.Errorf("failed to delete variant accounting row: %w", err)
}
err = c.variants.DeleteWithMeta(cacheKey)
if err != nil {
return err
}
return nil
}
// sourceReference identifies one source_metadata row's JSON sidecar.
type sourceReference struct {
host string
pathHash PathHash
}
// evictSourceBlob removes one source content blob. All source_metadata
// rows referencing the blob are deleted together with its
// source_content row in a single transaction BEFORE the file is
// unlinked: a blob referenced by multiple source paths is only ever
// removed together with all of its references, and database rows never
// point at deleted files. The JSON metadata sidecars for the removed
// rows are deleted afterwards.
//
// The whole operation holds the content hash's lock (the same one
// StoreSource holds for its full store), so a concurrent store of
// identical content bytes can never observe the file gone but a row
// still present, or insert a fresh row between this transaction's
// commit and the file unlink below: it either runs entirely before
// this eviction starts, or is blocked until this eviction (row
// deletion and unlink together) has fully completed.
func (c *Cache) evictSourceBlob(ctx context.Context, contentHash ContentHash) error {
unlock := c.contentLocks.Lock(string(contentHash))
defer unlock()
references, err := c.sourceReferences(ctx, contentHash)
if err != nil {
return err
}
tx, err := c.db.BeginTx(ctx, nil)
if err != nil {
return fmt.Errorf("failed to begin eviction transaction: %w", err)
}
defer func() { _ = tx.Rollback() }()
_, err = tx.ExecContext(ctx,
`DELETE FROM source_metadata WHERE content_hash = ?`, string(contentHash))
if err != nil {
return fmt.Errorf("failed to delete source metadata rows: %w", err)
}
_, err = tx.ExecContext(ctx,
`DELETE FROM source_content WHERE content_hash = ?`, string(contentHash))
if err != nil {
return fmt.Errorf("failed to delete source content row: %w", err)
}
err = tx.Commit()
if err != nil {
return fmt.Errorf("failed to commit eviction transaction: %w", err)
}
if c.evictSourceBlobTestHook != nil {
c.evictSourceBlobTestHook(contentHash)
}
// Only after the rows are gone may the files be removed.
for _, reference := range references {
err := c.srcMetadata.Delete(reference.host, reference.pathHash)
if err != nil {
c.log.Warn("failed to delete metadata sidecar",
"host", reference.host, "path_hash", reference.pathHash, "error", err)
}
}
err = c.srcContent.Delete(contentHash)
if err != nil {
return err
}
return nil
}
// sourceReferences lists the metadata sidecar locations of every
// source_metadata row referencing the given blob.
func (c *Cache) sourceReferences(
ctx context.Context, contentHash ContentHash,
) ([]sourceReference, error) {
rows, err := c.db.QueryContext(ctx, `
SELECT source_host, path_hash FROM source_metadata WHERE content_hash = ?
`, string(contentHash))
if err != nil {
return nil, fmt.Errorf("failed to query source references: %w", err)
}
defer func() { _ = rows.Close() }()
var references []sourceReference
for rows.Next() {
var reference sourceReference
var pathHash string
err := rows.Scan(&reference.host, &pathHash)
if err != nil {
return nil, fmt.Errorf("failed to scan source reference: %w", err)
}
reference.pathHash = PathHash(pathHash)
references = append(references, reference)
}
err = rows.Err()
if err != nil {
return nil, fmt.Errorf("source reference iteration failed: %w", err)
}
return references, nil
}
// notifyWritePressure wakes the background evictor after a store, so
// eviction under write pressure happens promptly without blocking the
// storing request. The notification channel has capacity one and drops
// when a wakeup is already pending.
func (c *Cache) notifyWritePressure() {
if c.disabled || c.config.MaxBytes <= 0 {
return
}
select {
case c.evictionPressure <- struct{}{}:
default:
}
}
// StartEviction launches the background eviction goroutine, which
// reconciles the database accounting with the cache directories at
// startup and again on every periodic tick thereafter, and evicts to
// the configured limit on the given periodic interval and on
// write-pressure notifications. It is a no-op on a disabled cache or
// when already started.
func (c *Cache) StartEviction(interval time.Duration) {
if c.disabled || c.evictionStarted {
return
}
c.evictionStarted = true
go c.evictionLoop(interval)
}
// StopEviction stops the background eviction goroutine and waits for
// it to exit. It is safe to call when eviction was never started, and
// safe to call more than once.
func (c *Cache) StopEviction() {
if !c.evictionStarted {
return
}
c.evictionStopOnce.Do(func() {
close(c.evictionStop)
<-c.evictionDone
})
}
// evictionLoop is the body of the background eviction goroutine.
func (c *Cache) evictionLoop(interval time.Duration) {
defer close(c.evictionDone)
ctx := context.Background()
c.runReconciliationPass(ctx)
c.runEvictionPass(ctx)
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-c.evictionStop:
return
case <-ticker.C:
// Reconciliation walks the cache directories, so it only
// runs on the periodic ticker rather than on every
// write-pressure wakeup, keeping it off the per-store hot
// path. Reusing the eviction interval itself (rather than a
// separate, longer one) is a deliberate choice: it is the
// simplest option that still bounds how long a store's
// best-effort accounting insert can stay silently
// unaccounted for to one interval, on a process that is
// already running this loop regardless.
c.runReconciliationPass(ctx)
case <-c.evictionPressure:
}
c.runEvictionPass(ctx)
}
}
// runEvictionPass runs one eviction pass, logging failures instead of
// propagating them (the loop must keep running).
func (c *Cache) runEvictionPass(ctx context.Context) {
err := c.EvictToLimit(ctx)
if err != nil {
c.log.Warn("cache eviction pass failed", "error", err)
}
}
// runReconciliationPass runs one reconciliation pass, logging failures
// instead of propagating them (the loop must keep running).
func (c *Cache) runReconciliationPass(ctx context.Context) {
err := c.reconcileAccounting(ctx)
if err != nil {
c.log.Warn("cache accounting reconciliation failed", "error", err)
}
}
// reconcileAccounting synchronizes the database size accounting with
// the actual contents of the cache directories. It runs at startup and
// again on every periodic eviction tick thereafter, off the request
// hot path: it adopts variant files that predate the accounting table
// (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), and sweeps stale temp files
// left behind by crashed writes. Running it periodically, not just
// once, bounds how long such drift can accumulate unaccounted for on a
// long-running process to one eviction interval.
func (c *Cache) reconcileAccounting(ctx context.Context) error {
if c.disabled {
return nil
}
err := c.reconcileVariantFiles(ctx)
if err != nil {
return err
}
err = c.reconcileVariantRows(ctx)
if err != nil {
return err
}
err = c.reconcileSourceFiles(ctx)
if err != nil {
return err
}
err = c.reconcileSourceRows(ctx)
if err != nil {
return err
}
return nil
}
// reconcileVariantFiles walks the variant storage directory, adopting
// files without accounting rows and sweeping stale temp files.
func (c *Cache) reconcileVariantFiles(ctx context.Context) error {
return filepath.WalkDir(
c.variants.baseDir,
func(path string, entry fs.DirEntry, err error) error {
if err != nil || entry.IsDir() {
return err
}
name := entry.Name()
if strings.HasPrefix(name, tempFilePrefix) {
c.sweepStaleTempFile(path, entry)
return nil
}
if strings.HasSuffix(name, variantMetaSuffix) {
return nil
}
return c.adoptVariantFile(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(
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)
}
modTime := info.ModTime().UTC().Format(sqliteTimestampLayout)
contentType := c.variantContentTypeFromSidecar(path)
_, err = c.db.ExecContext(ctx, `
INSERT INTO variant_content
(cache_key, size_bytes, content_type, created_at, last_accessed_at)
VALUES (?, ?, ?, ?, ?)
`, string(cacheKey), info.Size(), contentType, modTime, modTime)
if err != nil {
return fmt.Errorf("failed to adopt variant file into accounting: %w", err)
}
c.log.Info("adopted untracked variant file into size accounting",
"cache_key", cacheKey, "size_bytes", info.Size())
return nil
}
// variantContentTypeFromSidecar reads the content type from a variant
// .meta sidecar, falling back to application/octet-stream.
func (c *Cache) variantContentTypeFromSidecar(variantPath string) string {
//nolint:gosec // path from cache walk
metaData, err := os.ReadFile(variantPath + variantMetaSuffix)
if err != nil {
return fallbackContentType
}
var meta VariantMeta
if json.Unmarshal(metaData, &meta) != nil || meta.ContentType == "" {
return fallbackContentType
}
return meta.ContentType
}
// reconcileVariantRows drops accounting rows whose variant files are
// missing, so the database never references deleted content.
func (c *Cache) reconcileVariantRows(ctx context.Context) error {
keys, err := c.allVariantKeys(ctx)
if err != nil {
return err
}
for _, key := range keys {
if c.variants.Exists(key) {
continue
}
_, err := c.db.ExecContext(ctx,
`DELETE FROM variant_content WHERE cache_key = ?`, string(key))
if err != nil {
return fmt.Errorf("failed to drop stale variant accounting row: %w", err)
}
c.log.Info("dropped accounting row for missing variant file", "cache_key", key)
}
return nil
}
// allVariantKeys returns every tracked variant cache key.
func (c *Cache) allVariantKeys(ctx context.Context) ([]VariantKey, error) {
return queryStringColumn[VariantKey](ctx, c.db,
`SELECT cache_key FROM variant_content`, "variant keys", "variant key")
}
// queryStringColumn runs a single-column query and returns the column
// values as T. plural names the set for the query and scan failure
// messages; singular names one row for the scan and iteration failure
// messages.
func queryStringColumn[T ~string](
ctx context.Context, db *sql.DB, query, plural, singular string,
) ([]T, error) {
rows, err := db.QueryContext(ctx, query)
if err != nil {
return nil, fmt.Errorf("failed to query %s: %w", plural, err)
}
defer func() { _ = rows.Close() }()
var values []T
for rows.Next() {
var value string
err := rows.Scan(&value)
if err != nil {
return nil, fmt.Errorf("failed to scan %s: %w", singular, err)
}
values = append(values, T(value))
}
err = rows.Err()
if err != nil {
return nil, fmt.Errorf("%s iteration failed: %w", singular, err)
}
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.
func (c *Cache) reconcileSourceFiles(ctx context.Context) error {
return filepath.WalkDir(
c.srcContent.baseDir,
func(path string, entry fs.DirEntry, err error) error {
if err != nil || entry.IsDir() {
return err
}
name := entry.Name()
if strings.HasPrefix(name, tempFilePrefix) {
c.sweepStaleTempFile(path, entry)
return nil
}
return c.removeUntrackedSourceFile(ctx, path, ContentHash(name))
},
)
}
// 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,
`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)
}
err = os.Remove(path)
if err != nil && !os.IsNotExist(err) {
return fmt.Errorf("failed to remove untracked source file: %w", err)
}
c.log.Info("removed untracked source content file", "content_hash", contentHash)
return nil
}
// reconcileSourceRows removes source_content rows (and their metadata
// references and sidecars) whose blob files are missing on disk.
func (c *Cache) reconcileSourceRows(ctx context.Context) error {
hashes, err := c.allSourceContentHashes(ctx)
if err != nil {
return err
}
for _, hash := range hashes {
if c.srcContent.Exists(hash) {
continue
}
// The blob file is already gone; evictSourceBlob removes the
// rows and sidecars and tolerates the missing file.
err := c.evictSourceBlob(ctx, hash)
if err != nil {
return err
}
c.log.Info("dropped rows for missing source content file", "content_hash", hash)
}
return nil
}
// allSourceContentHashes returns every tracked source content hash.
func (c *Cache) allSourceContentHashes(ctx context.Context) ([]ContentHash, error) {
return queryStringColumn[ContentHash](ctx, c.db,
`SELECT content_hash FROM source_content`,
"source content hashes", "content hash")
}
// sweepStaleTempFile removes a temp file left behind by a crashed
// write once it is old enough that no in-flight store can own it.
func (c *Cache) sweepStaleTempFile(path string, entry fs.DirEntry) {
info, err := entry.Info()
if err != nil {
return
}
if time.Since(info.ModTime()) < staleTempFileAge {
return
}
err = os.Remove(path)
if err != nil && !os.IsNotExist(err) {
c.log.Warn("failed to remove stale temp file", "path", path, "error", err)
return
}
c.log.Info("removed stale temp file", "path", path)
}

View File

@@ -1,985 +0,0 @@
package imgcache
import (
"bytes"
"context"
"database/sql"
"io/fs"
"os"
"path/filepath"
"testing"
"time"
_ "modernc.org/sqlite"
"sneak.berlin/go/pixa/internal/database"
"sneak.berlin/go/pixa/internal/httpfetcher"
)
// sqliteTimestampFormat matches the format SQLite's CURRENT_TIMESTAMP
// produces, so injected timestamps compare correctly against ones the
// implementation writes.
const sqliteTimestampFormat = "2006-01-02 15:04:05"
// testVariantKeyOne through testVariantKeyFour are the variant cache
// keys reused across the eviction tests, in the order the tests store
// them.
const (
testVariantKeyOne VariantKey = "aabbccdd0001"
testVariantKeyTwo VariantKey = "aabbccdd0002"
testVariantKeyThree VariantKey = "aabbccdd0003"
testVariantKeyFour VariantKey = "aabbccdd0004"
)
// evictionTestDB creates an in-memory SQLite database with the real
// production schema, limited to a single connection so the background
// eviction goroutine shares the same in-memory database as the test.
func evictionTestDB(t *testing.T) *sql.DB {
t.Helper()
db, err := sql.Open("sqlite", ":memory:")
if err != nil {
t.Fatalf("failed to open test db: %v", err)
}
db.SetMaxOpenConns(1)
err = database.ApplyMigrations(context.Background(), db, nil)
if err != nil {
t.Fatalf("failed to apply migrations: %v", err)
}
t.Cleanup(func() { _ = db.Close() })
return db
}
// newEvictionTestCache creates a Cache backed by a temp directory and
// an in-memory database, with the given size limit.
func newEvictionTestCache(t *testing.T, maxBytes int64) (*Cache, string) {
t.Helper()
tmpDir := t.TempDir()
db := evictionTestDB(t)
// maxBytes zero mirrors the production mapping of
// cache_max_bytes: 0 (handlers sets DisableDiskCache); at the
// CacheConfig layer itself a zero MaxBytes means "no limit" for
// backwards compatibility with existing fixtures.
cache, err := NewCache(db, CacheConfig{
StateDir: tmpDir,
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute,
MaxBytes: maxBytes,
DisableDiskCache: maxBytes == 0,
})
if err != nil {
t.Fatalf("failed to create cache: %v", err)
}
return cache, tmpDir
}
// storeEvictionTestSource stores content as a fetched source for
// host/path and returns the resulting content hash.
func storeEvictionTestSource(
t *testing.T, cache *Cache, host, path string, content []byte,
) ContentHash {
t.Helper()
req := &ImageRequest{
SourceHost: host,
SourcePath: path,
Format: FormatJPEG,
Quality: 85,
FitMode: FitCover,
}
result := &httpfetcher.FetchResult{
StatusCode: 200,
ContentType: testContentTypeJPEG,
ContentLength: int64(len(content)),
Headers: map[string][]string{
testHeaderContentType: {testContentTypeJPEG},
},
}
hash, err := cache.StoreSource(
context.Background(), req, bytes.NewReader(content), result,
)
if err != nil {
t.Fatalf("StoreSource(%s%s) failed: %v", host, path, err)
}
return hash
}
// storeEvictionTestVariant stores content as a processed variant under
// the given cache key.
func storeEvictionTestVariant(
t *testing.T, cache *Cache, key VariantKey, content []byte,
) {
t.Helper()
err := cache.StoreVariant(t.Context(), key, bytes.NewReader(content), "image/webp")
if err != nil {
t.Fatalf("StoreVariant(%s) failed: %v", key, err)
}
}
// setVariantLastAccessed backdates the last access time of a tracked
// variant, to make LRU ordering deterministic in tests.
func setVariantLastAccessed(
t *testing.T, cache *Cache, key VariantKey, when time.Time,
) {
t.Helper()
res, err := cache.db.ExecContext(t.Context(),
`UPDATE variant_content SET last_accessed_at = ? WHERE cache_key = ?`,
when.UTC().Format(sqliteTimestampFormat), string(key),
)
if err != nil {
t.Fatalf("failed to set variant last_accessed_at: %v", err)
}
affected, err := res.RowsAffected()
if err != nil {
t.Fatalf("failed to read affected rows: %v", err)
}
if affected != 1 {
t.Fatalf("variant %s has no accounting row (affected=%d); "+
"stores must track variants in the database", key, affected)
}
}
// setSourceLastAccessed backdates the last access time of a tracked
// source content blob.
func setSourceLastAccessed(
t *testing.T, cache *Cache, hash ContentHash, when time.Time,
) {
t.Helper()
res, err := cache.db.ExecContext(t.Context(),
`UPDATE source_content SET last_accessed_at = ? WHERE content_hash = ?`,
when.UTC().Format(sqliteTimestampFormat), string(hash),
)
if err != nil {
t.Fatalf("failed to set source last_accessed_at: %v", err)
}
affected, err := res.RowsAffected()
if err != nil {
t.Fatalf("failed to read affected rows: %v", err)
}
if affected != 1 {
t.Fatalf("source %s has no accounting row (affected=%d)", hash, affected)
}
}
// countRows returns the number of rows the given query yields.
func countRows(t *testing.T, cache *Cache, query string, args ...any) int {
t.Helper()
var n int
err := cache.db.QueryRowContext(t.Context(), query, args...).Scan(&n)
if err != nil {
t.Fatalf("count query %q failed: %v", query, err)
}
return n
}
// assertNoDanglingReferences verifies the core eviction invariant:
// every database row that references cache content on disk points at a
// file that actually exists.
func assertNoDanglingReferences(t *testing.T, cache *Cache) {
t.Helper()
rows, err := cache.db.QueryContext(t.Context(),
`SELECT content_hash FROM source_metadata
WHERE content_hash IS NOT NULL AND content_hash != ''`,
)
if err != nil {
t.Fatalf("failed to query source_metadata: %v", err)
}
defer func() { _ = rows.Close() }()
for rows.Next() {
var hash string
err := rows.Scan(&hash)
if err != nil {
t.Fatalf("failed to scan content_hash: %v", err)
}
if !cache.srcContent.Exists(ContentHash(hash)) {
t.Errorf("source_metadata references content %s but the file is missing",
hash)
}
}
err = rows.Err()
if err != nil {
t.Fatalf("source_metadata iteration failed: %v", err)
}
variantRows, err := cache.db.QueryContext(t.Context(),
`SELECT cache_key FROM variant_content`)
if err != nil {
t.Fatalf("failed to query variant_content: %v", err)
}
defer func() { _ = variantRows.Close() }()
for variantRows.Next() {
var key string
err := variantRows.Scan(&key)
if err != nil {
t.Fatalf("failed to scan cache_key: %v", err)
}
if !cache.variants.Exists(VariantKey(key)) {
t.Errorf("variant_content references key %s but the file is missing", key)
}
}
err = variantRows.Err()
if err != nil {
t.Fatalf("variant_content iteration failed: %v", err)
}
}
// waitForUsageAtOrBelow polls UsageBytes until it reaches limit or the
// timeout expires, returning the last observed usage.
func waitForUsageAtOrBelow(
t *testing.T, cache *Cache, limit int64, timeout time.Duration,
) int64 {
t.Helper()
deadline := time.Now().Add(timeout)
var usage int64
for time.Now().Before(deadline) {
var err error
usage, err = cache.UsageBytes(context.Background())
if err != nil {
t.Fatalf("UsageBytes failed: %v", err)
}
if usage <= limit {
return usage
}
time.Sleep(25 * time.Millisecond)
}
return usage
}
func TestUsageBytesAccountsSourceAndVariantBytes(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
storeEvictionTestSource(t, cache, "src.example.com", "/a.jpg",
bytes.Repeat([]byte{0xAA}, 1000))
storeEvictionTestSource(t, cache, "src.example.com", "/b.jpg",
bytes.Repeat([]byte{0xAB}, 2000))
storeEvictionTestVariant(t, cache, testVariantKeyOne,
bytes.Repeat([]byte{0xAC}, 500))
storeEvictionTestVariant(t, cache, testVariantKeyTwo,
bytes.Repeat([]byte{0xAD}, 250))
usage, err := cache.UsageBytes(context.Background())
if err != nil {
t.Fatalf("UsageBytes failed: %v", err)
}
if usage != 3750 {
t.Errorf("UsageBytes = %d, want 3750 (1000+2000+500+250)", usage)
}
}
func TestUsageBytesCountsMultiReferencedBlobOnce(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
content := bytes.Repeat([]byte{0xCC}, 1200)
hashOne := storeEvictionTestSource(t, cache,
"src.example.com", "/one.jpg", content)
hashTwo := storeEvictionTestSource(t, cache,
"src.example.com", "/two.jpg", content)
if hashOne != hashTwo {
t.Fatalf("identical content produced different hashes: %s vs %s",
hashOne, hashTwo)
}
usage, err := cache.UsageBytes(context.Background())
if err != nil {
t.Fatalf("UsageBytes failed: %v", err)
}
if usage != 1200 {
t.Errorf("UsageBytes = %d, want 1200 (deduplicated blob counted once)", usage)
}
}
func TestEvictToLimitEvictsLeastRecentlyUsedFirst(t *testing.T) {
t.Parallel()
const limit = 3000
cache, _ := newEvictionTestCache(t, limit)
now := time.Now()
keys := []VariantKey{
testVariantKeyOne, testVariantKeyTwo,
testVariantKeyThree, testVariantKeyFour,
}
fills := []byte{0x01, 0x02, 0x03, 0x04}
ages := []time.Duration{4 * time.Hour, 3 * time.Hour, 2 * time.Hour, 1 * time.Hour}
for i, key := range keys {
storeEvictionTestVariant(t, cache, key, bytes.Repeat([]byte{fills[i]}, 1000))
setVariantLastAccessed(t, cache, key, now.Add(-ages[i]))
}
err := cache.EvictToLimit(context.Background())
if err != nil {
t.Fatalf("EvictToLimit failed: %v", err)
}
usage, err := cache.UsageBytes(context.Background())
if err != nil {
t.Fatalf("UsageBytes failed: %v", err)
}
if usage > limit {
t.Errorf("usage after eviction = %d, want <= %d", usage, limit)
}
if cache.variants.Exists(keys[0]) {
t.Errorf("least recently used variant %s must be evicted", keys[0])
}
if n := countRows(t, cache,
`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`, string(keys[0]),
); n != 0 {
t.Errorf("evicted variant %s still has %d accounting rows", keys[0], n)
}
for _, key := range keys[1:] {
if !cache.variants.Exists(key) {
t.Errorf("more recently used variant %s must survive eviction", key)
}
}
assertNoDanglingReferences(t, cache)
}
func TestEvictionRemovesMultiReferencedBlobTogetherWithAllReferences(t *testing.T) {
t.Parallel()
const limit = 1000
cache, _ := newEvictionTestCache(t, limit)
now := time.Now()
// One 800-byte blob referenced by two source paths.
sharedContent := bytes.Repeat([]byte{0xDD}, 800)
sharedHash := storeEvictionTestSource(t, cache,
"src.example.com", "/a.jpg", sharedContent)
h := storeEvictionTestSource(t, cache,
"src.example.com", "/b.jpg", sharedContent)
if h != sharedHash {
t.Fatalf("identical content produced different hashes: %s vs %s",
h, sharedHash)
}
// A newer 600-byte blob referenced by one source path.
recentHash := storeEvictionTestSource(t, cache, "src.example.com", "/c.jpg",
bytes.Repeat([]byte{0xEE}, 600))
setSourceLastAccessed(t, cache, sharedHash, now.Add(-2*time.Hour))
setSourceLastAccessed(t, cache, recentHash, now.Add(-time.Minute))
err := cache.EvictToLimit(context.Background())
if err != nil {
t.Fatalf("EvictToLimit failed: %v", err)
}
usage, err := cache.UsageBytes(context.Background())
if err != nil {
t.Fatalf("UsageBytes failed: %v", err)
}
if usage > limit {
t.Errorf("usage after eviction = %d, want <= %d", usage, limit)
}
// The multi-referenced blob must be gone from disk, from
// source_content, and from BOTH source_metadata rows: references
// are removed together with the blob, never left dangling.
if cache.srcContent.Exists(sharedHash) {
t.Errorf("evicted blob %s still exists on disk", sharedHash)
}
if n := countRows(t, cache,
`SELECT COUNT(*) FROM source_content WHERE content_hash = ?`, string(sharedHash),
); n != 0 {
t.Errorf("evicted blob %s still has %d source_content rows", sharedHash, n)
}
if n := countRows(t, cache,
`SELECT COUNT(*) FROM source_metadata WHERE content_hash = ?`, string(sharedHash),
); n != 0 {
t.Errorf("evicted blob %s still has %d source_metadata references", sharedHash, n)
}
// The JSON metadata sidecars for both referencing paths must be
// removed along with the rows.
for _, path := range []string{"/a.jpg", "/b.jpg"} {
pathHash := HashPath(path + "?")
if cache.srcMetadata.Exists("src.example.com", pathHash) {
t.Errorf("metadata sidecar for %s must be removed with its row", path)
}
}
// The more recently used blob survives fully intact.
if !cache.srcContent.Exists(recentHash) {
t.Errorf("recently used blob %s must survive eviction", recentHash)
}
if n := countRows(t, cache,
`SELECT COUNT(*) FROM source_metadata WHERE content_hash = ?`, string(recentHash),
); n != 1 {
t.Errorf("recently used blob %s has %d source_metadata rows, want 1",
recentHash, n)
}
assertNoDanglingReferences(t, cache)
}
func TestEvictionKeepsEverythingWhenUnderLimit(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
content := bytes.Repeat([]byte{0xDF}, 800)
hash := storeEvictionTestSource(t, cache, "src.example.com", "/a.jpg", content)
h := storeEvictionTestSource(t, cache, "src.example.com", "/b.jpg", content)
if h != hash {
t.Fatalf("identical content produced different hashes: %s vs %s", h, hash)
}
storeEvictionTestVariant(t, cache, testVariantKeyOne,
bytes.Repeat([]byte{0xE0}, 500))
err := cache.EvictToLimit(context.Background())
if err != nil {
t.Fatalf("EvictToLimit failed: %v", err)
}
if !cache.srcContent.Exists(hash) {
t.Errorf("blob %s must not be evicted while usage is under the limit", hash)
}
if n := countRows(t, cache,
`SELECT COUNT(*) FROM source_metadata WHERE content_hash = ?`, string(hash),
); n != 2 {
t.Errorf("blob %s has %d source_metadata rows, want 2", hash, n)
}
if !cache.variants.Exists(testVariantKeyOne) {
t.Error("variant must not be evicted while usage is under the limit")
}
assertNoDanglingReferences(t, cache)
}
func TestZeroMaxBytesDisablesDiskCache(t *testing.T) {
t.Parallel()
cache, tmpDir := newEvictionTestCache(t, 0)
req := &ImageRequest{
SourceHost: "src.example.com",
SourcePath: "/a.jpg",
Format: FormatJPEG,
Quality: 85,
FitMode: FitCover,
}
assertDisabledCacheWritesAreNoOps(t, cache, req)
assertDisabledCacheReadsAlwaysMiss(t, cache, req)
assertDisabledCacheTracksNothing(t, cache)
assertDisabledCacheWroteNothingToDisk(t, tmpDir)
}
// assertDisabledCacheWritesAreNoOps verifies that stores against a
// disabled cache report success without recording anything.
func assertDisabledCacheWritesAreNoOps(
t *testing.T, cache *Cache, req *ImageRequest,
) {
t.Helper()
ctx := t.Context()
err := cache.StoreVariant(
ctx, CacheKey(req), bytes.NewReader([]byte("data")), "image/webp",
)
if err != nil {
t.Fatalf("StoreVariant on disabled cache must be a no-op, got error: %v", err)
}
result := &httpfetcher.FetchResult{
StatusCode: 200,
ContentType: testContentTypeJPEG,
ContentLength: 4,
Headers: map[string][]string{},
}
hash, err := cache.StoreSource(ctx, req, bytes.NewReader([]byte("data")), result)
if err != nil {
t.Fatalf("StoreSource on disabled cache must be a no-op, got error: %v", err)
}
if hash != "" {
t.Errorf("StoreSource on disabled cache returned hash %q, want empty", hash)
}
}
// assertDisabledCacheReadsAlwaysMiss verifies that lookups against a
// disabled cache never report a hit.
func assertDisabledCacheReadsAlwaysMiss(
t *testing.T, cache *Cache, req *ImageRequest,
) {
t.Helper()
ctx := t.Context()
lookup, err := cache.Lookup(ctx, req)
if err != nil {
t.Fatalf("Lookup on disabled cache failed: %v", err)
}
if lookup.Hit {
t.Error("Lookup on disabled cache must always miss")
}
srcHash, srcType, err := cache.LookupSource(ctx, req)
if err != nil {
t.Fatalf("LookupSource on disabled cache failed: %v", err)
}
if srcHash != "" || srcType != "" {
t.Errorf("LookupSource on disabled cache = (%q, %q), want empty",
srcHash, srcType)
}
}
// assertDisabledCacheTracksNothing verifies that a disabled cache
// records no usage and writes no accounting rows.
func assertDisabledCacheTracksNothing(t *testing.T, cache *Cache) {
t.Helper()
usage, err := cache.UsageBytes(t.Context())
if err != nil {
t.Fatalf("UsageBytes failed: %v", err)
}
if usage != 0 {
t.Errorf("UsageBytes on disabled cache = %d, want 0", usage)
}
if n := countRows(t, cache, `SELECT COUNT(*) FROM source_content`); n != 0 {
t.Errorf("disabled cache wrote %d source_content rows, want 0", n)
}
if n := countRows(t, cache, `SELECT COUNT(*) FROM source_metadata`); n != 0 {
t.Errorf("disabled cache wrote %d source_metadata rows, want 0", n)
}
}
// assertDisabledCacheWroteNothingToDisk verifies that a disabled cache
// creates neither the cache directory tree nor any file under stateDir.
func assertDisabledCacheWroteNothingToDisk(t *testing.T, stateDir string) {
t.Helper()
_, err := os.Stat(filepath.Join(stateDir, "cache"))
if !os.IsNotExist(err) {
t.Errorf("disabled cache must not create the cache directory tree "+
"(stat err=%v)", err)
}
var foundFiles []string
walkErr := filepath.WalkDir(stateDir,
func(path string, d fs.DirEntry, err error) error {
if err != nil {
return err
}
if !d.IsDir() {
foundFiles = append(foundFiles, path)
}
return nil
})
if walkErr != nil {
t.Fatalf("failed to walk state dir: %v", walkErr)
}
if len(foundFiles) != 0 {
t.Errorf("disabled cache wrote files to disk: %v", foundFiles)
}
}
func TestEvictionRunsUnderWritePressure(t *testing.T) {
t.Parallel()
const limit = 1500
cache, _ := newEvictionTestCache(t, limit)
// An interval far longer than the test ensures only write
// pressure can trigger eviction here.
cache.StartEviction(time.Hour)
defer cache.StopEviction()
keys := []VariantKey{
testVariantKeyOne, testVariantKeyTwo, testVariantKeyThree,
}
fills := []byte{0x11, 0x12, 0x13}
for i, key := range keys {
storeEvictionTestVariant(t, cache, key,
bytes.Repeat([]byte{fills[i]}, 1000))
}
usage := waitForUsageAtOrBelow(t, cache, limit, 5*time.Second)
if usage > limit {
t.Errorf("write pressure did not trigger eviction: usage = %d, want <= %d",
usage, limit)
}
assertNoDanglingReferences(t, cache)
}
func TestEvictionRunsOnPeriodicSchedule(t *testing.T) {
t.Parallel()
const limit = 1500
cache, _ := newEvictionTestCache(t, limit)
// Start the evictor while the cache is empty, then create tracked
// over-limit state WITHOUT going through the store methods, so no
// write-pressure notification fires and only the periodic ticker
// can trigger eviction.
cache.StartEviction(100 * time.Millisecond)
defer cache.StopEviction()
keys := []VariantKey{
testVariantKeyOne, testVariantKeyTwo, testVariantKeyThree,
}
fills := []byte{0x21, 0x22, 0x23}
for i, key := range keys {
content := bytes.Repeat([]byte{fills[i]}, 1000)
_, err := cache.variants.Store(key, bytes.NewReader(content), "image/webp")
if err != nil {
t.Fatalf("failed to store variant file: %v", err)
}
_, err = cache.db.ExecContext(t.Context(),
`INSERT INTO variant_content (cache_key, size_bytes, content_type)
VALUES (?, ?, ?)`,
string(key), len(content), "image/webp",
)
if err != nil {
t.Fatalf("failed to insert variant accounting row: %v", err)
}
}
usage := waitForUsageAtOrBelow(t, cache, limit, 5*time.Second)
if usage > limit {
t.Errorf("periodic schedule did not trigger eviction: usage = %d, want <= %d",
usage, limit)
}
assertNoDanglingReferences(t, cache)
}
func TestStartEvictionReconcilesAccountingWithDisk(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
// An untracked variant file on disk (e.g. written before this
// feature existed) must be adopted into the accounting.
untracked := bytes.Repeat([]byte{0x31}, 1000)
_, err := cache.variants.Store(
testVariantKeyOne, bytes.NewReader(untracked), "image/webp",
)
if err != nil {
t.Fatalf("failed to store untracked variant file: %v", err)
}
// An accounting row whose file is missing must be dropped.
_, err = cache.db.ExecContext(t.Context(),
`INSERT INTO variant_content (cache_key, size_bytes, content_type)
VALUES (?, ?, ?)`,
"deadbeef0001", 700, "image/webp",
)
if err != nil {
t.Fatalf("failed to insert stale variant accounting row: %v", err)
}
cache.StartEviction(time.Hour)
defer cache.StopEviction()
deadline := time.Now().Add(5 * time.Second)
var usage int64
for time.Now().Before(deadline) {
var err error
usage, err = cache.UsageBytes(context.Background())
if err != nil {
t.Fatalf("UsageBytes failed: %v", err)
}
if usage == 1000 {
break
}
time.Sleep(25 * time.Millisecond)
}
if usage != 1000 {
t.Errorf("usage after reconciliation = %d, want 1000 "+
"(untracked file adopted, stale row dropped)", usage)
}
if n := countRows(t, cache,
`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`,
string(testVariantKeyOne),
); n != 1 {
t.Errorf("untracked variant file was not adopted into accounting (rows=%d)", n)
}
if n := countRows(t, cache,
`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`, "deadbeef0001",
); n != 0 {
t.Errorf("stale accounting row without a file was not dropped (rows=%d)", n)
}
}
// TestPeriodicReconciliationAdoptsFileThatAppearsAfterStartup proves
// reconciliation is not a one-shot startup-only pass: it must also run
// on the periodic ticker, so a variant file that lands on disk with no
// accounting row well after startup (e.g. because StoreVariant's
// best-effort accounting insert failed under transient contention, or
// any other cause of an untracked file appearing during steady-state
// operation) is still adopted into accounting eventually, rather than
// staying invisible to UsageBytes/EvictToLimit until the next process
// restart.
func TestPeriodicReconciliationAdoptsFileThatAppearsAfterStartup(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
const interval = 100 * time.Millisecond
cache.StartEviction(interval)
defer cache.StopEviction()
// Let startup reconciliation run and settle on an empty cache
// before introducing the untracked file, so the adoption we assert
// below can only be the work of a later, periodic pass.
time.Sleep(3 * interval)
// Simulate a variant whose accounting insert failed after the
// process was already running and serving requests: the content
// file is written directly, bypassing StoreVariant's (and thus its
// accounting insert) entirely, exactly as would happen if that
// insert had failed and only the file write had succeeded.
untracked := bytes.Repeat([]byte{0x41}, 900)
_, err := cache.variants.Store(
"aabbccdd0099", bytes.NewReader(untracked), "image/webp",
)
if err != nil {
t.Fatalf("failed to store untracked variant file: %v", err)
}
deadline := time.Now().Add(5 * time.Second)
var usage int64
for time.Now().Before(deadline) {
var err error
usage, err = cache.UsageBytes(context.Background())
if err != nil {
t.Fatalf("UsageBytes failed: %v", err)
}
if usage == 900 {
break
}
time.Sleep(25 * time.Millisecond)
}
if usage != 900 {
t.Errorf("usage after periodic reconciliation = %d, want 900 "+
"(a file that appeared after startup reconciliation already ran must still "+
"be adopted by a later periodic pass)", usage)
}
if n := countRows(t, cache,
`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`, "aabbccdd0099",
); n != 1 {
t.Errorf("file that appeared after startup was not adopted by periodic "+
"reconciliation (rows=%d)", n)
}
}
// TestEvictSourceBlobExcludesConcurrentStoreOfIdenticalContent exercises
// the exact TOCTOU window between evictSourceBlob's row-deletion
// transaction commit and its content file unlink: a concurrent
// StoreSource for a different source path whose content hashes to the
// same value (real SHA-256 dedup, not a contrived case) must not be
// able to insert a fresh row referencing the file while eviction is
// mid-unlink, and must not lose its own store once eviction has fully
// released the content hash.
func TestEvictSourceBlobExcludesConcurrentStoreOfIdenticalContent(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
ctx := context.Background()
content := bytes.Repeat([]byte{0x55}, 400)
hash := storeEvictionTestSource(t, cache,
"race.example.com", "/first.jpg", content)
proceed := make(chan struct{})
storeAttempted := make(chan struct{})
cache.evictSourceBlobTestHook = func(gotHash ContentHash) {
if gotHash != hash {
t.Errorf("test hook invoked for hash %s, want %s", gotHash, hash)
}
close(storeAttempted)
<-proceed
}
evictDone := make(chan error, 1)
go func() {
evictDone <- cache.evictSourceBlob(ctx, hash)
}()
// Wait until eviction has committed its delete transaction and is
// paused (inside the test hook) immediately before unlinking the
// content file: exactly the window the review flagged.
<-storeAttempted
storeDone := storeIdenticalContentConcurrently(ctx, cache, content)
// The concurrent store must not be able to complete while eviction
// still holds the content hash (i.e. before the file is unlinked):
// if it could, it would insert a row referencing a file about to be
// removed out from under it.
select {
case err := <-storeDone:
t.Fatalf("StoreSource for identical content completed (err=%v) while eviction "+
"still held the content hash open between commit and unlink; the store and "+
"the evict of identical content are not mutually exclusive", err)
case <-time.After(200 * time.Millisecond):
// Expected: the store is blocked behind eviction's exclusion.
}
close(proceed)
err := <-evictDone
if err != nil {
t.Fatalf("evictSourceBlob failed: %v", err)
}
err = <-storeDone
if err != nil {
t.Fatalf("StoreSource failed: %v", err)
}
assertNoDanglingReferences(t, cache)
dupHash, _, err := cache.LookupSource(ctx, &ImageRequest{
SourceHost: "race.example.com",
SourcePath: "/dup.jpg",
})
if err != nil {
t.Fatalf("LookupSource failed: %v", err)
}
if dupHash == "" {
t.Fatal("re-stored blob was lost: the store legitimately ran after eviction " +
"released the content hash and must have recreated the file and row")
}
if !cache.srcContent.Exists(dupHash) {
t.Errorf("source_content/source_metadata references %s but its file is missing",
dupHash)
}
}
// storeIdenticalContentConcurrently starts a StoreSource for a second
// source path whose body hashes to the same content hash, returning the
// channel its error is delivered on.
func storeIdenticalContentConcurrently(
ctx context.Context, cache *Cache, content []byte,
) chan error {
storeDone := make(chan error, 1)
go func() {
req := &ImageRequest{
SourceHost: "race.example.com",
SourcePath: "/dup.jpg",
Format: FormatJPEG,
Quality: 85,
FitMode: FitCover,
}
result := &httpfetcher.FetchResult{
StatusCode: 200,
ContentType: testContentTypeJPEG,
ContentLength: int64(len(content)),
Headers: map[string][]string{
testHeaderContentType: {testContentTypeJPEG},
},
}
_, err := cache.StoreSource(ctx, req, bytes.NewReader(content), result)
storeDone <- err
}()
return storeDone
}

View File

@@ -315,7 +315,6 @@ func (s *Service) fetchAndProcess(
return nil, fmt.Errorf("upstream fetch failed: %w", err) return nil, fmt.Errorf("upstream fetch failed: %w", err)
} }
defer func() { _ = fetchResult.Content.Close() }() defer func() { _ = fetchResult.Content.Close() }()
// Read and validate the source content // Read and validate the source content
@@ -425,7 +424,7 @@ func (s *Service) processAndStore(
// Store variant to cache // Store variant to cache
err = s.cache.StoreVariant( err = s.cache.StoreVariant(
ctx, cacheKey, bytes.NewReader(processedData), processResult.ContentType, cacheKey, bytes.NewReader(processedData), processResult.ContentType,
) )
if err != nil { if err != nil {
s.log.Warn("failed to store variant", "error", err) s.log.Warn("failed to store variant", "error", err)

View File

@@ -66,31 +66,57 @@ func (s *ContentStorage) Store(r io.Reader) (ContentHash, int64, error) {
hash := ContentHash(hex.EncodeToString(h[:])) hash := ContentHash(hex.EncodeToString(h[:]))
size := int64(len(data)) size := int64(len(data))
err = s.writeIfAbsent(hash, data) // Build path: <basedir>/<ab>/<cd>/<hash>
path := s.hashToPath(hash)
// Check if already exists
_, err = os.Stat(path)
if err == nil {
return hash, size, nil
}
// Create directory structure
dir := filepath.Dir(path)
err = os.MkdirAll(dir, StorageDirPerm)
if err != nil { if err != nil {
return "", 0, err return "", 0, fmt.Errorf("failed to create directory: %w", err)
}
// Write 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
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)
} }
return hash, size, nil return hash, size, nil
} }
// StoreHashed writes pre-hashed content to storage at the path derived
// 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.
func (s *ContentStorage) StoreHashed(
hash ContentHash, data []byte,
) (int64, error) {
err := s.writeIfAbsent(hash, data)
if err != nil {
return 0, err
}
return int64(len(data)), nil
}
// Load returns a reader for the content with the given hash. // Load returns a reader for the content with the given hash.
func (s *ContentStorage) Load(hash ContentHash) (io.ReadCloser, error) { func (s *ContentStorage) Load(hash ContentHash) (io.ReadCloser, error) {
path := s.hashToPath(hash) path := s.hashToPath(hash)
@@ -150,67 +176,6 @@ func (s *ContentStorage) Exists(hash ContentHash) bool {
return err == nil 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.
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 {
return nil
}
// Create directory structure
dir := filepath.Dir(path)
err := os.MkdirAll(dir, StorageDirPerm)
if err != nil {
return fmt.Errorf("failed to create directory: %w", err)
}
// Write to temp file first, then rename for atomicity
tmpFile, err := os.CreateTemp(dir, ".tmp-*")
if err != nil {
return fmt.Errorf("failed to create temp file: %w", err)
}
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)
}
err = tmpFile.Close()
if err != nil {
_ = os.Remove(tmpPath)
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)
return fmt.Errorf("failed to rename temp file: %w", err)
}
return nil
}
// hashToPath converts a hash to a file path: <basedir>/<ab>/<cd>/<hash> // hashToPath converts a hash to a file path: <basedir>/<ab>/<cd>/<hash>
func (s *ContentStorage) hashToPath(hash ContentHash) string { func (s *ContentStorage) hashToPath(hash ContentHash) string {
h := string(hash) h := string(hash)
@@ -557,24 +522,6 @@ func (s *VariantStorage) Delete(key VariantKey) error {
return nil return nil
} }
// DeleteWithMeta removes the content at the given key together with
// its .meta sidecar file. A missing file is not an error.
func (s *VariantStorage) DeleteWithMeta(key VariantKey) error {
err := s.Delete(key)
if err != nil {
return err
}
metaPath := s.keyToPath(key) + ".meta"
err = os.Remove(metaPath)
if err != nil && !os.IsNotExist(err) {
return fmt.Errorf("failed to delete variant metadata: %w", err)
}
return nil
}
// keyToPath converts a key to a file path: <basedir>/<ab>/<cd>/<key> // keyToPath converts a key to a file path: <basedir>/<ab>/<cd>/<key>
func (s *VariantStorage) keyToPath(key VariantKey) string { func (s *VariantStorage) keyToPath(key VariantKey) string {
k := string(key) k := string(key)

View File

@@ -20,11 +20,10 @@ import (
// Shared test data literals, extracted as constants for goconst. // Shared test data literals, extracted as constants for goconst.
const ( const (
testHostCDN = "cdn.example.com" testHostCDN = "cdn.example.com"
testHostExample = "example.com" testHostExample = "example.com"
testPathCat = "/photos/cat.jpg" testPathCat = "/photos/cat.jpg"
testContentTypeJPEG = "image/jpeg" testContentTypeJPEG = "image/jpeg"
testHeaderContentType = "Content-Type"
) )
// TestFixtures contains paths to test files in the mock filesystem. // TestFixtures contains paths to test files in the mock filesystem.

View File

@@ -111,7 +111,7 @@ func TestSigner_GoldenVectors(t *testing.T) {
gotPath, _, _ := signer.GenerateSignedURL(&urlReq, time.Hour) gotPath, _, _ := signer.GenerateSignedURL(&urlReq, time.Hour)
if gotPath != tt.wantSignedPath { if gotPath != tt.wantSignedPath {
t.Errorf("GenerateSignedURL() path = %q, want %q (signed URL layout changed?)", t.Errorf("GenerateSignedURL() path = %q, want %q (layout changed?)",
gotPath, tt.wantSignedPath) gotPath, tt.wantSignedPath)
} }
}) })

View File

@@ -17,7 +17,7 @@ run_with_cgo_deps() {
main() { main() {
cd "$ROOT" cd "$ROOT"
echo "Running tests..." echo "Running tests..."
run_with_cgo_deps "CGO_ENABLED=1 go test -timeout 30s -race -v ./..." run_with_cgo_deps "CGO_ENABLED=1 go test -timeout 30s -v ./..."
} }
main "$@" main "$@"