Compare commits
2 Commits
13e9f2c072
...
next
| Author | SHA1 | Date | |
|---|---|---|---|
| 2d805125ee | |||
| 63fbc98e63 |
@@ -115,6 +115,9 @@ Configured via YAML file (`--config`). Key settings:
|
||||
- `upstream_max_response_size` — max origin response size
|
||||
- `downstream_timeout` — client response timeout
|
||||
- `signing_key` — HMAC secret for URL signatures
|
||||
- `cache_max_bytes` — disk cache size limit in bytes; `0` disables the
|
||||
disk cache entirely; omitted defaults to 75% of the free space on
|
||||
the filesystem containing `<state_dir>/cache/` (minimum 500 MiB)
|
||||
|
||||
See `config.example.yml` for all options with defaults.
|
||||
|
||||
|
||||
51
TODO.md
51
TODO.md
@@ -12,15 +12,14 @@
|
||||
|
||||
pre-1.0. No git tags exist. Recent work extracted the internal/magic,
|
||||
internal/allowlist, internal/httpfetcher, and internal/signature
|
||||
packages. The gosec findings from the 2026-07-06 survey are resolved:
|
||||
the last two open findings (G124, session cookie attributes in
|
||||
internal/session) are fixed as of this change, so `make check` is green
|
||||
on main.
|
||||
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
|
||||
with LRU eviction (`cache_max_bytes`), closing the unbounded disk
|
||||
growth DoS vector.
|
||||
|
||||
# Next Step
|
||||
|
||||
P0: implement cache size management and eviction so the disk cannot
|
||||
fill up
|
||||
P1: implement blocked networks configuration to extend SSRF protection
|
||||
|
||||
# Completed Steps
|
||||
|
||||
@@ -28,12 +27,40 @@ fill up
|
||||
`.golangci.yml` (v2 schema, `default: all` minus six disabled
|
||||
linters, `lll` 88, tests included): bumped the pinned
|
||||
`golangci/golangci-lint:v2.12.2-alpine` image in `Dockerfile` and the
|
||||
release-archive sha256 pins in `script/bootstrap`; fixed all 747
|
||||
findings the stricter config surfaced (notably `paralleltest`,
|
||||
`wsl_v5`, `goconst`, `lll`, `noinlineerr`, `err113`, `errcheck`,
|
||||
`testpackage` — white-box test files renamed to
|
||||
`*_internal_test.go`); three `//nolint:tagliatelle` directives keep
|
||||
release-archive sha256 pins in `script/bootstrap`; fixed the findings
|
||||
the stricter config surfaced (notably `paralleltest`, `wsl_v5`,
|
||||
`goconst`, `lll`, `noinlineerr`, `err113`, `errcheck`, `testpackage` —
|
||||
white-box test files renamed to `*_internal_test.go`), including #55's
|
||||
code absorbed after it merged, iterating the pinned linter to
|
||||
`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
|
||||
- 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
|
||||
@@ -89,8 +116,6 @@ fill up
|
||||
|
||||
# Future Steps
|
||||
|
||||
- P1: implement blocked networks configuration to extend SSRF
|
||||
protection
|
||||
- P1: rate limit global concurrent upstream fetches to prevent
|
||||
resource exhaustion
|
||||
- P1: strip EXIF and other metadata from processed images (privacy)
|
||||
|
||||
@@ -28,6 +28,13 @@ allow_http: false
|
||||
# Maximum concurrent connections per upstream host (default: 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_dsn: ""
|
||||
|
||||
|
||||
305
internal/config/cache_max_bytes_internal_test.go
Normal file
305
internal/config/cache_max_bytes_internal_test.go
Normal file
@@ -0,0 +1,305 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
116
internal/config/cachesize.go
Normal file
116
internal/config/cachesize.go
Normal file
@@ -0,0 +1,116 @@
|
||||
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
|
||||
}
|
||||
@@ -41,6 +41,7 @@ const (
|
||||
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
|
||||
@@ -61,7 +62,11 @@ var (
|
||||
errTooFewConnections = errors.New("must be at least 1")
|
||||
errValueTooShort = errors.New("value too short")
|
||||
errMustBeSetTogether = errors.New("must be set together")
|
||||
errValueNull = errors.New(
|
||||
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")
|
||||
@@ -94,6 +99,19 @@ type Config struct {
|
||||
AllowlistHosts []string // Hosts that don't require signatures
|
||||
AllowHTTP bool // Allow non-TLS upstream (testing only)
|
||||
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.
|
||||
@@ -120,6 +138,11 @@ func New(_ fx.Lifecycle, params Params) (*Config, error) {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
err = c.resolveCacheMaxBytes(log, defaultFreeSpaceProbe)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if c.Debug {
|
||||
params.Logger.EnableDebugLogging()
|
||||
}
|
||||
@@ -160,6 +183,16 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
|
||||
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
|
||||
@@ -266,7 +299,7 @@ func isKnownConfigKey(key string) bool {
|
||||
switch key {
|
||||
case keyDebug, keyMaintenanceMode, keyPort, keyStateDir, keySentryDSN,
|
||||
keyDBURL, keyMetrics, keySigningKey, keyAllowlistHosts, keyAllowHTTP,
|
||||
keyUpstreamConnectionsPerHost, "env":
|
||||
keyUpstreamConnectionsPerHost, keyCacheMaxBytes, "env":
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -299,7 +332,7 @@ func (c *Config) ensureStateDirWritable() error {
|
||||
keyStateDir, probePath, err)
|
||||
}
|
||||
|
||||
err = os.Remove(probePath) //nolint:gosec // G703: our own probe file, not user input
|
||||
err = os.Remove(probePath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("config key %q: cannot remove probe file %q: %w",
|
||||
keyStateDir, probePath, err)
|
||||
@@ -339,6 +372,13 @@ func (c *Config) validate() error {
|
||||
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 {
|
||||
@@ -411,7 +451,6 @@ func loadConfigFile(log *slog.Logger, appName string) (*smartconfig.Config, erro
|
||||
for _, path := range configPaths {
|
||||
cleanPath := filepath.Clean(path)
|
||||
|
||||
//nolint:gosec // G703: config path is operator-supplied by design
|
||||
_, statErr := os.Stat(cleanPath)
|
||||
if statErr == nil {
|
||||
// A config file that exists but does not parse is a fatal
|
||||
@@ -464,6 +503,19 @@ func (l *strictLoader) intVal(key string, defaultVal int) int {
|
||||
return val
|
||||
}
|
||||
|
||||
func (l *strictLoader) int64Val(key string, defaultVal int64) int64 {
|
||||
if l.err != nil {
|
||||
return 0
|
||||
}
|
||||
|
||||
val, err := getInt64(l.sc, key, defaultVal)
|
||||
if err != nil {
|
||||
l.err = err
|
||||
}
|
||||
|
||||
return val
|
||||
}
|
||||
|
||||
func (l *strictLoader) boolVal(key string, defaultVal bool) bool {
|
||||
if l.err != nil {
|
||||
return false
|
||||
@@ -546,6 +598,57 @@ func getInt(sc *smartconfig.Config, key string, defaultVal int) (int, error) {
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
|
||||
@@ -3,12 +3,17 @@
|
||||
|
||||
-- Source content blobs
|
||||
-- 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 (
|
||||
content_hash TEXT PRIMARY KEY,
|
||||
content_type TEXT 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
|
||||
-- JSON stored at: cache/src-metadata/<hostname>/<path_hash>.json
|
||||
@@ -34,6 +39,22 @@ 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_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
|
||||
-- Files stored at: cache/dst-content/<ab>/<cd>/<sha256>
|
||||
CREATE TABLE IF NOT EXISTS output_content (
|
||||
|
||||
@@ -51,9 +51,24 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
|
||||
}
|
||||
|
||||
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 {
|
||||
return s.initImageService()
|
||||
},
|
||||
OnStop: func(_ context.Context) error {
|
||||
if s.imgCache != nil {
|
||||
s.imgCache.StopEviction()
|
||||
}
|
||||
|
||||
return nil
|
||||
},
|
||||
})
|
||||
|
||||
return s, nil
|
||||
@@ -61,11 +76,15 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
|
||||
|
||||
// initImageService initializes the image cache and service.
|
||||
func (s *Handlers) initImageService() error {
|
||||
// Create the cache
|
||||
// Create the cache. cache_max_bytes: 0 disables the disk cache
|
||||
// entirely; any other value is the eviction limit in bytes.
|
||||
cache, err := imgcache.NewCache(s.db.DB(), imgcache.CacheConfig{
|
||||
StateDir: s.config.StateDir,
|
||||
CacheTTL: imgcache.DefaultCacheTTL,
|
||||
NegativeTTL: imgcache.DefaultNegativeTTL,
|
||||
StateDir: s.config.StateDir,
|
||||
CacheTTL: imgcache.DefaultCacheTTL,
|
||||
NegativeTTL: imgcache.DefaultNegativeTTL,
|
||||
MaxBytes: s.config.CacheMaxBytes,
|
||||
DisableDiskCache: s.config.CacheMaxBytes == 0,
|
||||
Logger: s.log,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -73,6 +92,10 @@ func (s *Handlers) initImageService() error {
|
||||
|
||||
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
|
||||
fetcherCfg := httpfetcher.DefaultConfig()
|
||||
fetcherCfg.AllowHTTP = s.config.AllowHTTP
|
||||
|
||||
@@ -233,7 +233,6 @@ func (f *HTTPFetcher) Fetch(ctx context.Context, url string) (*FetchResult, erro
|
||||
|
||||
startTime := time.Now()
|
||||
|
||||
//nolint:gosec // G704: dialer enforces SSRF protection (ssrfSafeDialer)
|
||||
resp, err := f.client.Do(req)
|
||||
|
||||
fetchDuration := time.Since(startTime)
|
||||
|
||||
@@ -2,12 +2,16 @@ package imgcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
@@ -27,6 +31,22 @@ type CacheConfig struct {
|
||||
StateDir string
|
||||
CacheTTL 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.
|
||||
@@ -42,14 +62,61 @@ type Cache struct {
|
||||
variants *VariantStorage // processed variants by cache key
|
||||
srcMetadata *MetadataStorage // source metadata by host/path
|
||||
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
|
||||
// reading .meta files
|
||||
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.
|
||||
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(
|
||||
filepath.Join(config.StateDir, "cache", "sources"),
|
||||
)
|
||||
@@ -71,14 +138,11 @@ func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) {
|
||||
return nil, fmt.Errorf("failed to create source metadata storage: %w", err)
|
||||
}
|
||||
|
||||
return &Cache{
|
||||
db: db,
|
||||
srcContent: srcContent,
|
||||
variants: variants,
|
||||
srcMetadata: srcMetadata,
|
||||
config: config,
|
||||
metaCache: make(map[VariantKey]variantMeta),
|
||||
}, nil
|
||||
c.srcContent = srcContent
|
||||
c.variants = variants
|
||||
c.srcMetadata = srcMetadata
|
||||
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// LookupResult contains the result of a cache lookup.
|
||||
@@ -90,12 +154,15 @@ type LookupResult struct {
|
||||
CacheStatus CacheStatus
|
||||
}
|
||||
|
||||
// Lookup checks if a processed variant exists on disk (no DB access for hits).
|
||||
func (c *Cache) Lookup(_ context.Context, req *ImageRequest) (*LookupResult, error) {
|
||||
// Lookup checks if a processed variant exists on disk. Hits touch the
|
||||
// variant's LRU timestamp; a disabled cache always misses.
|
||||
func (c *Cache) Lookup(ctx context.Context, req *ImageRequest) (*LookupResult, error) {
|
||||
cacheKey := CacheKey(req)
|
||||
|
||||
// Check variant storage directly - no DB needed for cache hits
|
||||
if c.variants.Exists(cacheKey) {
|
||||
if !c.disabled && c.variants.Exists(cacheKey) {
|
||||
c.touchVariant(ctx, cacheKey)
|
||||
|
||||
return &LookupResult{
|
||||
Hit: true,
|
||||
CacheKey: cacheKey,
|
||||
@@ -112,18 +179,52 @@ func (c *Cache) Lookup(_ context.Context, req *ImageRequest) (*LookupResult, err
|
||||
|
||||
// GetVariant returns a reader, size, and content type for a cached variant.
|
||||
func (c *Cache) GetVariant(cacheKey VariantKey) (io.ReadCloser, int64, string, error) {
|
||||
if c.disabled {
|
||||
return nil, 0, "", ErrNotFound
|
||||
}
|
||||
|
||||
return c.variants.LoadWithMeta(cacheKey)
|
||||
}
|
||||
|
||||
// StoreSource stores fetched source content and metadata.
|
||||
// StoreSource stores fetched source content and metadata. On a
|
||||
// disabled cache it is a no-op returning an empty hash.
|
||||
func (c *Cache) StoreSource(
|
||||
ctx context.Context,
|
||||
req *ImageRequest,
|
||||
content io.Reader,
|
||||
result *httpfetcher.FetchResult,
|
||||
) (ContentHash, error) {
|
||||
// Store content
|
||||
contentHash, size, err := c.srcContent.Store(content)
|
||||
if c.disabled {
|
||||
return "", nil
|
||||
}
|
||||
|
||||
// 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 {
|
||||
return "", fmt.Errorf("failed to store source content: %w", err)
|
||||
}
|
||||
@@ -180,23 +281,56 @@ func (c *Cache) StoreSource(
|
||||
// A failure here is non-fatal; the metadata is in the database.
|
||||
_ = c.srcMetadata.Store(req.SourceHost, pathHash, meta)
|
||||
|
||||
c.notifyWritePressure()
|
||||
|
||||
return contentHash, nil
|
||||
}
|
||||
|
||||
// StoreVariant stores a processed variant by its cache key.
|
||||
// StoreVariant stores a processed variant by its cache key and records
|
||||
// it in the size accounting. On a disabled cache it is a no-op. The
|
||||
// accounting insert is best-effort (the startup reconciliation pass
|
||||
// adopts any variant file that misses its accounting row).
|
||||
func (c *Cache) StoreVariant(
|
||||
cacheKey VariantKey, content io.Reader, contentType string,
|
||||
ctx context.Context, cacheKey VariantKey, content io.Reader, contentType string,
|
||||
) error {
|
||||
_, err := c.variants.Store(cacheKey, content, contentType)
|
||||
if c.disabled {
|
||||
return nil
|
||||
}
|
||||
|
||||
return err
|
||||
size, err := c.variants.Store(cacheKey, content, contentType)
|
||||
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.
|
||||
// Returns the content hash and content type if found, or empty values if not.
|
||||
// Returns the content hash and content type if found, or empty values
|
||||
// if not. Hits touch the blob's LRU timestamp; a disabled cache always
|
||||
// reports no cached source.
|
||||
func (c *Cache) LookupSource(
|
||||
ctx context.Context, req *ImageRequest,
|
||||
) (ContentHash, string, error) {
|
||||
if c.disabled {
|
||||
return "", "", nil
|
||||
}
|
||||
|
||||
var hashStr, contentType string
|
||||
|
||||
err := c.db.QueryRowContext(ctx, `
|
||||
@@ -219,6 +353,8 @@ func (c *Cache) LookupSource(
|
||||
return "", "", nil
|
||||
}
|
||||
|
||||
c.touchSourceContent(ctx, contentHash)
|
||||
|
||||
return contentHash, contentType, nil
|
||||
}
|
||||
|
||||
@@ -265,6 +401,10 @@ func (c *Cache) GetSourceMetadataID(
|
||||
|
||||
// GetSourceContent returns a reader for cached source content by its hash.
|
||||
func (c *Cache) GetSourceContent(contentHash ContentHash) (io.ReadCloser, error) {
|
||||
if c.disabled {
|
||||
return nil, ErrNotFound
|
||||
}
|
||||
|
||||
return c.srcContent.Load(contentHash)
|
||||
}
|
||||
|
||||
@@ -340,6 +480,32 @@ 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.
|
||||
func (c *Cache) checkNegativeCache(
|
||||
ctx context.Context, req *ImageRequest,
|
||||
|
||||
@@ -160,7 +160,7 @@ func TestCache_StoreAndLookup(t *testing.T) {
|
||||
sourceContent := []byte("fake jpeg data")
|
||||
fetchResult := &httpfetcher.FetchResult{
|
||||
ContentType: testContentTypeJPEG,
|
||||
Headers: map[string][]string{"Content-Type": {testContentTypeJPEG}},
|
||||
Headers: map[string][]string{testHeaderContentType: {testContentTypeJPEG}},
|
||||
}
|
||||
|
||||
contentHash, err := cache.StoreSource(
|
||||
@@ -177,7 +177,8 @@ func TestCache_StoreAndLookup(t *testing.T) {
|
||||
cacheKey := CacheKey(req)
|
||||
outputContent := []byte("fake webp data")
|
||||
|
||||
err = cache.StoreVariant(cacheKey, bytes.NewReader(outputContent), "image/webp")
|
||||
err = cache.StoreVariant(
|
||||
t.Context(), cacheKey, bytes.NewReader(outputContent), "image/webp")
|
||||
if err != nil {
|
||||
t.Fatalf("StoreVariant() error = %v", err)
|
||||
}
|
||||
@@ -295,7 +296,8 @@ func TestCache_VariantLookup(t *testing.T) {
|
||||
cacheKey := CacheKey(req)
|
||||
outputContent := []byte("output data")
|
||||
|
||||
err := cache.StoreVariant(cacheKey, bytes.NewReader(outputContent), "image/webp")
|
||||
err := cache.StoreVariant(
|
||||
t.Context(), cacheKey, bytes.NewReader(outputContent), "image/webp")
|
||||
if err != nil {
|
||||
t.Fatalf("StoreVariant() error = %v", err)
|
||||
}
|
||||
@@ -344,7 +346,8 @@ func TestCache_GetVariant_ReturnsContentType(t *testing.T) {
|
||||
cacheKey := CacheKey(req)
|
||||
outputContent := []byte("output webp data")
|
||||
|
||||
err := cache.StoreVariant(cacheKey, bytes.NewReader(outputContent), "image/webp")
|
||||
err := cache.StoreVariant(
|
||||
t.Context(), cacheKey, bytes.NewReader(outputContent), "image/webp")
|
||||
if err != nil {
|
||||
t.Fatalf("StoreVariant() error = %v", err)
|
||||
}
|
||||
@@ -395,7 +398,8 @@ func TestCache_GetVariant(t *testing.T) {
|
||||
cacheKey := CacheKey(req)
|
||||
outputContent := []byte("the actual output content")
|
||||
|
||||
err := cache.StoreVariant(cacheKey, bytes.NewReader(outputContent), "image/webp")
|
||||
err := cache.StoreVariant(
|
||||
t.Context(), cacheKey, bytes.NewReader(outputContent), "image/webp")
|
||||
if err != nil {
|
||||
t.Fatalf("StoreVariant() error = %v", err)
|
||||
}
|
||||
|
||||
64
internal/imgcache/contentlock.go
Normal file
64
internal/imgcache/contentlock.go
Normal file
@@ -0,0 +1,64 @@
|
||||
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()
|
||||
}
|
||||
}
|
||||
142
internal/imgcache/contentlock_internal_test.go
Normal file
142
internal/imgcache/contentlock_internal_test.go
Normal file
@@ -0,0 +1,142 @@
|
||||
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")
|
||||
}
|
||||
}
|
||||
803
internal/imgcache/eviction.go
Normal file
803
internal/imgcache/eviction.go
Normal file
@@ -0,0 +1,803 @@
|
||||
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)
|
||||
}
|
||||
985
internal/imgcache/eviction_internal_test.go
Normal file
985
internal/imgcache/eviction_internal_test.go
Normal file
@@ -0,0 +1,985 @@
|
||||
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
|
||||
}
|
||||
@@ -425,7 +425,7 @@ func (s *Service) processAndStore(
|
||||
|
||||
// Store variant to cache
|
||||
err = s.cache.StoreVariant(
|
||||
cacheKey, bytes.NewReader(processedData), processResult.ContentType,
|
||||
ctx, cacheKey, bytes.NewReader(processedData), processResult.ContentType,
|
||||
)
|
||||
if err != nil {
|
||||
s.log.Warn("failed to store variant", "error", err)
|
||||
|
||||
@@ -66,58 +66,31 @@ func (s *ContentStorage) Store(r io.Reader) (ContentHash, int64, error) {
|
||||
hash := ContentHash(hex.EncodeToString(h[:]))
|
||||
size := int64(len(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)
|
||||
err = s.writeIfAbsent(hash, data)
|
||||
if err != nil {
|
||||
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) //nolint:gosec // G703: our own temp file, not user input
|
||||
|
||||
return "", 0, fmt.Errorf("failed to write content: %w", err)
|
||||
}
|
||||
|
||||
err = tmpFile.Close()
|
||||
if err != nil {
|
||||
_ = os.Remove(tmpPath) //nolint:gosec // G703: our own temp file, not user input
|
||||
|
||||
return "", 0, fmt.Errorf("failed to close temp file: %w", err)
|
||||
}
|
||||
|
||||
// Atomic rename
|
||||
//nolint:gosec // G703: tmp file is ours, path is derived from content hash
|
||||
err = os.Rename(filepath.Clean(tmpPath), filepath.Clean(path))
|
||||
if err != nil {
|
||||
_ = os.Remove(tmpPath) //nolint:gosec // G703: our own temp file, not user input
|
||||
|
||||
return "", 0, fmt.Errorf("failed to rename temp file: %w", err)
|
||||
return "", 0, err
|
||||
}
|
||||
|
||||
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.
|
||||
func (s *ContentStorage) Load(hash ContentHash) (io.ReadCloser, error) {
|
||||
path := s.hashToPath(hash)
|
||||
@@ -177,6 +150,67 @@ func (s *ContentStorage) Exists(hash ContentHash) bool {
|
||||
return err == nil
|
||||
}
|
||||
|
||||
// writeIfAbsent writes data to the path derived from hash, unless
|
||||
// content already exists there, via a temp-file-plus-rename so
|
||||
// concurrent readers never observe a partial file.
|
||||
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>
|
||||
func (s *ContentStorage) hashToPath(hash ContentHash) string {
|
||||
h := string(hash)
|
||||
@@ -254,23 +288,22 @@ func (s *MetadataStorage) Store(
|
||||
_, err = tmpFile.Write(data)
|
||||
if err != nil {
|
||||
_ = tmpFile.Close()
|
||||
_ = os.Remove(tmpPath) //nolint:gosec // G703: our own temp file, not user input
|
||||
_ = os.Remove(tmpPath)
|
||||
|
||||
return fmt.Errorf("failed to write metadata: %w", err)
|
||||
}
|
||||
|
||||
err = tmpFile.Close()
|
||||
if err != nil {
|
||||
_ = os.Remove(tmpPath) //nolint:gosec // G703: our own temp file, not user input
|
||||
_ = os.Remove(tmpPath)
|
||||
|
||||
return fmt.Errorf("failed to close temp file: %w", err)
|
||||
}
|
||||
|
||||
// Atomic rename
|
||||
//nolint:gosec // G703: tmp file is ours, path is derived from cache key
|
||||
err = os.Rename(filepath.Clean(tmpPath), filepath.Clean(path))
|
||||
if err != nil {
|
||||
_ = os.Remove(tmpPath) //nolint:gosec // G703: our own temp file, not user input
|
||||
_ = os.Remove(tmpPath)
|
||||
|
||||
return fmt.Errorf("failed to rename temp file: %w", err)
|
||||
}
|
||||
@@ -412,23 +445,22 @@ func (s *VariantStorage) Store(
|
||||
_, err = tmpFile.Write(data)
|
||||
if err != nil {
|
||||
_ = tmpFile.Close()
|
||||
_ = os.Remove(tmpPath) //nolint:gosec // G703: our own temp file, not user input
|
||||
_ = os.Remove(tmpPath)
|
||||
|
||||
return 0, fmt.Errorf("failed to write content: %w", err)
|
||||
}
|
||||
|
||||
err = tmpFile.Close()
|
||||
if err != nil {
|
||||
_ = os.Remove(tmpPath) //nolint:gosec // G703: our own temp file, not user input
|
||||
_ = os.Remove(tmpPath)
|
||||
|
||||
return 0, fmt.Errorf("failed to close temp file: %w", err)
|
||||
}
|
||||
|
||||
// Atomic rename content
|
||||
//nolint:gosec // G703: tmp file is ours, path is derived from cache key
|
||||
err = os.Rename(filepath.Clean(tmpPath), filepath.Clean(path))
|
||||
if err != nil {
|
||||
_ = os.Remove(tmpPath) //nolint:gosec // G703: our own temp file, not user input
|
||||
_ = os.Remove(tmpPath)
|
||||
|
||||
return 0, fmt.Errorf("failed to rename temp file: %w", err)
|
||||
}
|
||||
@@ -525,6 +557,24 @@ func (s *VariantStorage) Delete(key VariantKey) error {
|
||||
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>
|
||||
func (s *VariantStorage) keyToPath(key VariantKey) string {
|
||||
k := string(key)
|
||||
|
||||
@@ -20,10 +20,11 @@ import (
|
||||
|
||||
// Shared test data literals, extracted as constants for goconst.
|
||||
const (
|
||||
testHostCDN = "cdn.example.com"
|
||||
testHostExample = "example.com"
|
||||
testPathCat = "/photos/cat.jpg"
|
||||
testContentTypeJPEG = "image/jpeg"
|
||||
testHostCDN = "cdn.example.com"
|
||||
testHostExample = "example.com"
|
||||
testPathCat = "/photos/cat.jpg"
|
||||
testContentTypeJPEG = "image/jpeg"
|
||||
testHeaderContentType = "Content-Type"
|
||||
)
|
||||
|
||||
// TestFixtures contains paths to test files in the mock filesystem.
|
||||
|
||||
@@ -111,7 +111,7 @@ func TestSigner_GoldenVectors(t *testing.T) {
|
||||
|
||||
gotPath, _, _ := signer.GenerateSignedURL(&urlReq, time.Hour)
|
||||
if gotPath != tt.wantSignedPath {
|
||||
t.Errorf("GenerateSignedURL() path = %q, want %q (layout changed?)",
|
||||
t.Errorf("GenerateSignedURL() path = %q, want %q (signed URL layout changed?)",
|
||||
gotPath, tt.wantSignedPath)
|
||||
}
|
||||
})
|
||||
|
||||
@@ -17,7 +17,7 @@ run_with_cgo_deps() {
|
||||
main() {
|
||||
cd "$ROOT"
|
||||
echo "Running tests..."
|
||||
run_with_cgo_deps "CGO_ENABLED=1 go test -timeout 30s -v ./..."
|
||||
run_with_cgo_deps "CGO_ENABLED=1 go test -timeout 30s -race -v ./..."
|
||||
}
|
||||
|
||||
main "$@"
|
||||
|
||||
Reference in New Issue
Block a user