Build the image pipeline in pixad's fx graph (closes #313)
check / check (push) Waiting to run

The new package internal/providers builds the cache, the fetcher, the
processor, the image service, the session manager and the encrypted URL
generator from the config. pixad's fx graph provides them with
providers.Constructors(), and handlers.New receives them built. The
server tests build the same graph, with only the fetcher's dial function
replaced. database.New opens the database, so these parts can read it
while fx builds them; the cache's stop hook still runs before the
database's.

Each setting's default is declared in internal/config, and
internal/providers works out the default cache_max_bytes. The cache's
MaxBytes is the one value for the disk cache limit, and 0 turns the disk
cache off. The fetcher and the processor take explicit values.

Model: opus-5-5
This commit is contained in:
2026-10-26 11:16:09 +00:00
parent ac2bae0489
commit 6d9631a8fc
54 changed files with 1181 additions and 1102 deletions
+3 -1
View File
@@ -812,7 +812,9 @@ See `configs/config.example.yml` for all options with defaults.
### Architecture ### Architecture
- **Dependency injection**: Uber fx - **Dependency injection**: Uber fx. `internal/app` builds every part of pixad
from the config in one fx graph. Each setting's default is declared once, in
`internal/config`.
- **HTTP router**: go-chi - **HTTP router**: go-chi
- **Image processing**: govips (CGO wrapper for libvips) - **Image processing**: govips (CGO wrapper for libvips)
- **Database**: SQLite via modernc.org/sqlite - **Database**: SQLite via modernc.org/sqlite
+24
View File
@@ -30,6 +30,30 @@ P2: security: per-origin rate limiting
# Completed Steps # Completed Steps
- 2026-10-26 pixad's fx graph builds the image pipeline (closes #313).
`database.New` now opens the database, so the parts below can read it while
they are built; its stop hook still runs after the cache's. The cache's
`MaxBytes` is the one value for the disk cache limit, and 0 turns the disk
cache off. For an omitted `cache_max_bytes`, `internal/providers` works out
`config.DefaultCacheMaxBytes` before it builds the cache. `internal/app` and
the server tests build the parts below with `providers.Constructors()`; the
image proxy test replaces only the fetcher's dial function.
- Built by the providers in `internal/providers`, and given to
`handlers.New`:
- the cache
- the fetcher
- the processor
- the image service
- the session manager
- the encrypted URL generator
- Removed:
- `initImageService` and `handlers.Params.Fetcher`
- `CacheConfig.UseDefaultMaxBytes`, `CacheConfig.DisableDiskCache` and
`Cache.disabled`
- `httpfetcher.DefaultConfig()` and the fetcher's copies of setting
defaults
- `imageprocessor.DefaultMaxInputBytes`, and the processor's defaults
for a zero `MaxInputBytes` or `MaxConcurrentProcessing`
- 2026-10-26 the code nothing in production called is gone (closes #312). The - 2026-10-26 the code nothing in production called is gone (closes #312). The
handlers tests sign their image URLs with `generateSignedURL`, a test helper handlers tests sign their image URLs with `generateSignedURL`, a test helper
that writes them as the removed generator did. Removed: that writes them as the removed generator did. Removed:
+2
View File
@@ -17,6 +17,7 @@ import (
"sneak.berlin/go/pixa/internal/healthcheck" "sneak.berlin/go/pixa/internal/healthcheck"
"sneak.berlin/go/pixa/internal/logger" "sneak.berlin/go/pixa/internal/logger"
"sneak.berlin/go/pixa/internal/middleware" "sneak.berlin/go/pixa/internal/middleware"
"sneak.berlin/go/pixa/internal/providers"
"sneak.berlin/go/pixa/internal/server" "sneak.berlin/go/pixa/internal/server"
) )
@@ -65,6 +66,7 @@ func run(_ *cobra.Command, _ []string) {
// The registry /metrics serves // The registry /metrics serves
func() prometheus.Registerer { return prometheus.DefaultRegisterer }, func() prometheus.Registerer { return prometheus.DefaultRegisterer },
), ),
fx.Provide(providers.Constructors()...),
fx.Invoke( fx.Invoke(
func(log *logger.Logger) { log.Identify() }, func(log *logger.Logger) { log.Identify() },
func(*server.Server) {}, func(*server.Server) {},
@@ -135,10 +135,38 @@ func TestCacheMaxBytesInvalidValuesAbortStartup(t *testing.T) {
} }
} }
// TestDefaultCacheMaxBytesAppliesFloor verifies that when 75% of the
// free space plus what the cache holds is below 500 MiB, the default is
// floored at DefaultCacheMaxBytesFloor.
func TestDefaultCacheMaxBytesAppliesFloor(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()
got := DefaultCacheMaxBytes(tc.freeBytes, 0)
if got != DefaultCacheMaxBytesFloor {
t.Errorf("DefaultCacheMaxBytes = %d, want floor %d",
got, DefaultCacheMaxBytesFloor)
}
})
}
}
// TestCacheMaxBytesExplicitIsRecorded verifies that an omitted // TestCacheMaxBytesExplicitIsRecorded verifies that an omitted
// cache_max_bytes is recorded as not explicit, so the cache works out // cache_max_bytes is recorded as not explicit, so pixad works out the
// the default when it opens, and that an explicit zero is recorded as // default when it builds the cache, and that an explicit zero is recorded
// explicit, so it disables the disk cache instead. // as explicit, so it disables the disk cache instead.
func TestCacheMaxBytesExplicitIsRecorded(t *testing.T) { func TestCacheMaxBytesExplicitIsRecorded(t *testing.T) {
t.Parallel() t.Parallel()
+39 -12
View File
@@ -45,6 +45,37 @@ func DefaultMaxConcurrentProcessing() int {
return max(runtime.GOMAXPROCS(0)/2, 1) //nolint:mnd // half the CPUs return max(runtime.GOMAXPROCS(0)/2, 1) //nolint:mnd // half the CPUs
} }
// DefaultCacheMaxBytesFloor is the least DefaultCacheMaxBytes returns:
// 500 MiB. It applies only to the default, never to an explicit
// cache_max_bytes.
const DefaultCacheMaxBytesFloor int64 = 500 << 20
// freeSpaceFractionNumerator and freeSpaceFractionDenominator express
// the 75% share of DefaultCacheMaxBytes as integer arithmetic (dividing
// before multiplying avoids overflow).
const (
freeSpaceFractionNumerator uint64 = 3
freeSpaceFractionDenominator uint64 = 4
)
// DefaultCacheMaxBytes returns the disk cache limit when cache_max_bytes
// is omitted: 75% of the sum of freeBytes, the space free on the
// filesystem holding the cache, and heldBytes, the bytes the cache
// already holds, and at least DefaultCacheMaxBytesFloor. Counting what
// the cache holds keeps the limit from shrinking as the cache fills.
// pixad works it out when it builds the cache, as both numbers are known
// only then.
func DefaultCacheMaxBytes(freeBytes uint64, heldBytes int64) int64 {
// Both terms are at most math.MaxInt64, so the sum cannot overflow.
//nolint:gosec // G115: heldBytes is the cache's total, never negative
spaceBytes := min(freeBytes, math.MaxInt64) + uint64(heldBytes)
computed := spaceBytes / freeSpaceFractionDenominator * freeSpaceFractionNumerator
computed = min(computed, math.MaxInt64)
return max(int64(computed), DefaultCacheMaxBytesFloor)
}
// Configuration key names. // Configuration key names.
const ( const (
keyDebug = "debug" keyDebug = "debug"
@@ -190,18 +221,14 @@ type Config struct {
// address, and an explicit list replaces the default. // address, and an explicit list replaces the default.
TrustedProxies []netip.Prefix TrustedProxies []netip.Prefix
// CacheMaxBytes is the disk cache size limit in bytes. Only an // CacheMaxBytes is the disk cache size limit in bytes, as set; 0
// explicit zero (CacheMaxBytesExplicit true) disables the disk // turns the disk cache off.
// cache. Zero with CacheMaxBytesExplicit false means
// cache_max_bytes was omitted, and the cache works out the default
// limit when it opens.
CacheMaxBytes int64 CacheMaxBytes int64
// CacheMaxBytesExplicit records whether cache_max_bytes was // CacheMaxBytesExplicit records whether cache_max_bytes was
// explicitly set, in the environment or the configuration file. // explicitly set, in the environment or the configuration file.
// Explicit values are used exactly as given; for an omitted key the // Explicit values are used exactly as given. For an omitted key
// cache works out the default limit when it opens (see // pixad works out DefaultCacheMaxBytes when it builds the cache.
// imgcache.CacheConfig.UseDefaultMaxBytes).
CacheMaxBytesExplicit bool CacheMaxBytesExplicit bool
} }
@@ -239,8 +266,8 @@ func New(_ fx.Lifecycle, params Params) (*Config, error) {
return nil, err return nil, err
} }
// An omitted cache_max_bytes is worked out and logged when the // An omitted cache_max_bytes is worked out and logged when pixad
// cache opens. // builds the cache.
if c.CacheMaxBytesExplicit { if c.CacheMaxBytesExplicit {
log.Info("effective cache size limit", log.Info("effective cache size limit",
"cache_max_bytes", c.CacheMaxBytes, "cache_max_bytes", c.CacheMaxBytes,
@@ -325,8 +352,8 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
} }
// The default for an omitted cache_max_bytes is worked out when // The default for an omitted cache_max_bytes is worked out when
// the cache opens; here we only record whether the operator set // pixad builds the cache; here we only record whether the operator
// the key explicitly. // set the key explicitly.
_, c.CacheMaxBytesExplicit = lookupValue(sc, keyCacheMaxBytes) _, c.CacheMaxBytesExplicit = lookupValue(sc, keyCacheMaxBytes)
// Build DBURL from StateDir if not explicitly set. The derived URL // Build DBURL from StateDir if not explicitly set. The derived URL
+11 -11
View File
@@ -82,7 +82,11 @@ func ParseMigrationVersion(filename string) (int, error) {
return version, nil return version, nil
} }
// New creates a new Database instance. // New opens the database and applies the migrations, so that the parts
// fx builds after it can read the database while they are built. Its
// stop hook closes the database. fx runs stop hooks in the reverse of the
// order it built their parts in, so the stop hooks of those parts run
// before this one.
func New(lc fx.Lifecycle, params Params) (*Database, error) { func New(lc fx.Lifecycle, params Params) (*Database, error) {
s := &Database{ s := &Database{
log: params.Logger.Get(), log: params.Logger.Get(),
@@ -91,20 +95,16 @@ func New(lc fx.Lifecycle, params Params) (*Database, error) {
s.log.Info("Database instantiated") s.log.Info("Database instantiated")
lc.Append(fx.Hook{ err := s.connect(context.Background())
OnStart: func(ctx context.Context) error { if err != nil {
s.log.Info("Database OnStart Hook") return nil, err
}
return s.connect(ctx) lc.Append(fx.Hook{
},
OnStop: func(_ context.Context) error { OnStop: func(_ context.Context) error {
s.log.Info("Database OnStop Hook") s.log.Info("Database OnStop Hook")
if s.db != nil { return s.db.Close()
return s.db.Close()
}
return nil
}, },
}) })
+10 -3
View File
@@ -21,6 +21,7 @@ import (
"sneak.berlin/go/pixa/internal/database" "sneak.berlin/go/pixa/internal/database"
"sneak.berlin/go/pixa/internal/encurl" "sneak.berlin/go/pixa/internal/encurl"
"sneak.berlin/go/pixa/internal/imageprocessor"
"sneak.berlin/go/pixa/internal/imagerequest" "sneak.berlin/go/pixa/internal/imagerequest"
"sneak.berlin/go/pixa/internal/imgcache" "sneak.berlin/go/pixa/internal/imgcache"
) )
@@ -121,6 +122,7 @@ func newCountingDatabaseServer(
cache, err := imgcache.NewCache(db, imgcache.CacheConfig{ cache, err := imgcache.NewCache(db, imgcache.CacheConfig{
StateDir: tb.TempDir(), StateDir: tb.TempDir(),
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
MaxBytes: 1 << 30,
}, prometheus.NewRegistry()) }, prometheus.NewRegistry())
if err != nil { if err != nil {
tb.Fatalf("imgcache.NewCache() error = %v", err) tb.Fatalf("imgcache.NewCache() error = %v", err)
@@ -131,9 +133,14 @@ func newCountingDatabaseServer(
Fetcher: newMockFetcher(fstest.MapFS{ Fetcher: newMockFetcher(fstest.MapFS{
countingHost + photoPath: &fstest.MapFile{Data: photo.Bytes()}, countingHost + photoPath: &fstest.MapFile{Data: photo.Bytes()},
}), }),
SigningKey: testSigningKey, Processor: imageprocessor.New(imageprocessor.Params{
Allowlist: []string{countingHost}, MaxInputBytes: 50 << 20,
Logger: slog.New(slog.DiscardHandler), MaxConcurrentProcessing: 2,
}),
SigningKey: testSigningKey,
Allowlist: []string{countingHost},
MaxResponseSize: 50 << 20,
Logger: slog.New(slog.DiscardHandler),
}) })
if err != nil { if err != nil {
tb.Fatalf("imgcache.NewService() error = %v", err) tb.Fatalf("imgcache.NewService() error = %v", err)
@@ -1,151 +0,0 @@
package handlers
import (
"os"
"path/filepath"
"testing"
"github.com/prometheus/client_golang/prometheus"
"go.uber.org/fx/fxtest"
"sneak.berlin/go/pixa/internal/config"
"sneak.berlin/go/pixa/internal/database"
"sneak.berlin/go/pixa/internal/globals"
"sneak.berlin/go/pixa/internal/logger"
)
// TestNewCacheConfigFromCacheMaxBytes checks the cache configuration
// built from cache_max_bytes: omitted, the cache works out the default
// limit; 0 turns the disk cache off; a positive value is the limit,
// unchanged.
func TestNewCacheConfigFromCacheMaxBytes(t *testing.T) {
t.Parallel()
const oneGiB = 1 << 30
cases := []struct {
name string
cacheMaxBytes int64
cacheMaxBytesExplicit bool
wantMaxBytes int64
wantUseDefaultMaxBytes bool
wantDisableDiskCache bool
}{
{
name: "cache_max_bytes omitted",
cacheMaxBytes: 0,
cacheMaxBytesExplicit: false,
wantMaxBytes: 0,
wantUseDefaultMaxBytes: true,
wantDisableDiskCache: false,
},
{
name: "cache_max_bytes: 0",
cacheMaxBytes: 0,
cacheMaxBytesExplicit: true,
wantMaxBytes: 0,
wantUseDefaultMaxBytes: false,
wantDisableDiskCache: true,
},
{
name: "cache_max_bytes: 1 GiB",
cacheMaxBytes: oneGiB,
cacheMaxBytesExplicit: true,
wantMaxBytes: oneGiB,
wantUseDefaultMaxBytes: false,
wantDisableDiskCache: false,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
cfg := &config.Config{
CacheMaxBytes: tc.cacheMaxBytes,
CacheMaxBytesExplicit: tc.cacheMaxBytesExplicit,
}
got := newCacheConfig(cfg, nil)
t.Logf("MaxBytes = %d, UseDefaultMaxBytes = %v, DisableDiskCache = %v",
got.MaxBytes, got.UseDefaultMaxBytes, got.DisableDiskCache)
if got.MaxBytes != tc.wantMaxBytes {
t.Errorf("MaxBytes = %d, want %d", got.MaxBytes, tc.wantMaxBytes)
}
if got.UseDefaultMaxBytes != tc.wantUseDefaultMaxBytes {
t.Errorf("UseDefaultMaxBytes = %v, want %v",
got.UseDefaultMaxBytes, tc.wantUseDefaultMaxBytes)
}
if got.DisableDiskCache != tc.wantDisableDiskCache {
t.Errorf("DisableDiskCache = %v, want %v",
got.DisableDiskCache, tc.wantDisableDiskCache)
}
})
}
}
// TestDiskCacheOffOnlyForExplicitZeroCacheMaxBytes starts the handlers
// once with cache_max_bytes omitted and once with cache_max_bytes: 0,
// and checks by whether the cache directories were created that the
// disk cache is on in the first case and off in the second.
func TestDiskCacheOffOnlyForExplicitZeroCacheMaxBytes(t *testing.T) {
t.Parallel()
cases := []struct {
name string
cacheMaxBytesExplicit bool
wantDiskCache bool
}{
{name: "cache_max_bytes omitted", cacheMaxBytesExplicit: false, wantDiskCache: true},
{name: "cache_max_bytes: 0", cacheMaxBytesExplicit: true, wantDiskCache: false},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
stateDir := t.TempDir()
cfg := &config.Config{
SigningKey: testSigningKey,
StateDir: stateDir,
DBURL: "file:" + filepath.Join(stateDir, "state.sqlite3"),
CacheMaxBytes: 0,
CacheMaxBytesExplicit: tc.cacheMaxBytesExplicit,
}
lc := fxtest.NewLifecycle(t)
log, err := logger.New(lc, logger.Params{Globals: &globals.Globals{}})
if err != nil {
t.Fatalf("logger.New() error = %v", err)
}
db, err := database.New(lc, database.Params{Logger: log, Config: cfg})
if err != nil {
t.Fatalf("database.New() error = %v", err)
}
_, err = New(lc, Params{
Logger: log, Database: db, Config: cfg,
MetricsRegistry: prometheus.NewRegistry(),
})
if err != nil {
t.Fatalf("New() error = %v", err)
}
lc.RequireStart()
t.Cleanup(lc.RequireStop)
_, err = os.Stat(filepath.Join(stateDir, "cache", "variants"))
gotDiskCache := err == nil
if gotDiskCache != tc.wantDiskCache {
t.Errorf("cache directories created = %v, want %v",
gotDiskCache, tc.wantDiskCache)
}
})
}
}
@@ -1,63 +0,0 @@
package handlers
import (
"net/http"
"net/netip"
"path/filepath"
"testing"
"time"
"github.com/go-chi/chi/v5"
"github.com/prometheus/client_golang/prometheus"
"go.uber.org/fx"
"go.uber.org/fx/fxtest"
"sneak.berlin/go/pixa/internal/config"
"sneak.berlin/go/pixa/internal/database"
"sneak.berlin/go/pixa/internal/globals"
"sneak.berlin/go/pixa/internal/healthcheck"
"sneak.berlin/go/pixa/internal/logger"
)
// TestHandlersBuildTheirOwnFetcherWhenNoneIsProvided builds the handlers as
// pixad does, in an fx app that provides no fetcher, and requests an image
// from 192.0.2.10, which is on the allowlist and in blocked_networks. The URL
// check accepts that address; only the dialer that refuses internal
// addresses checks blocked_networks, so the answer is 403 only if the
// fetcher the handlers build from the config connects with that dialer. Any
// other dialer would try to connect until the upstream fetch timeout, which
// is short so that the test then fails quickly.
func TestHandlersBuildTheirOwnFetcherWhenNoneIsProvided(t *testing.T) {
t.Parallel()
const host = "192.0.2.10"
stateDir := t.TempDir()
cfg := &config.Config{
SigningKey: testSigningKey,
StateDir: stateDir,
DBURL: "file:" + filepath.Join(stateDir, "state.sqlite3"),
AllowlistHosts: []string{host},
BlockedNetworks: []netip.Prefix{netip.MustParsePrefix("192.0.2.0/24")},
UpstreamFetchTimeout: 2 * time.Second,
// With no connection slots, the fetch would fail before dialing.
UpstreamConnections: config.DefaultUpstreamConnections,
}
var h *Handlers
app := fxtest.New(t,
fx.Supply(cfg),
fx.Provide(globals.New, logger.New, database.New, healthcheck.New, New,
func() prometheus.Registerer { return prometheus.NewRegistry() }),
fx.Populate(&h),
)
app.RequireStart()
t.Cleanup(app.RequireStop)
r := chi.NewRouter()
r.Get("/v1/image/*", h.HandleImage())
rec := sendGet(t, r, photoURL(host))
checkErrorBody(t, rec, http.StatusForbidden, "forbidden")
}
+12 -5
View File
@@ -15,6 +15,7 @@ import (
"github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus"
"sneak.berlin/go/pixa/internal/encurl" "sneak.berlin/go/pixa/internal/encurl"
"sneak.berlin/go/pixa/internal/httpfetcher" "sneak.berlin/go/pixa/internal/httpfetcher"
"sneak.berlin/go/pixa/internal/imageprocessor"
"sneak.berlin/go/pixa/internal/imagerequest" "sneak.berlin/go/pixa/internal/imagerequest"
"sneak.berlin/go/pixa/internal/imgcache" "sneak.berlin/go/pixa/internal/imgcache"
) )
@@ -36,17 +37,23 @@ func newRoutesFetchingThrough(
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{ cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
StateDir: t.TempDir(), StateDir: t.TempDir(),
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
MaxBytes: 1 << 30,
}, prometheus.NewRegistry()) }, prometheus.NewRegistry())
if err != nil { if err != nil {
t.Fatalf("imgcache.NewCache() error = %v", err) t.Fatalf("imgcache.NewCache() error = %v", err)
} }
svc, err := imgcache.NewService(&imgcache.ServiceConfig{ svc, err := imgcache.NewService(&imgcache.ServiceConfig{
Cache: cache, Cache: cache,
Fetcher: fetcher, Fetcher: fetcher,
SigningKey: testSigningKey, Processor: imageprocessor.New(imageprocessor.Params{
Allowlist: []string{allowlistedHost}, MaxInputBytes: 50 << 20,
Logger: log, MaxConcurrentProcessing: 2,
}),
SigningKey: testSigningKey,
Allowlist: []string{allowlistedHost},
MaxResponseSize: 50 << 20,
Logger: log,
}) })
if err != nil { if err != nil {
t.Fatalf("imgcache.NewService() error = %v", err) t.Fatalf("imgcache.NewService() error = %v", err)
+9 -138
View File
@@ -4,19 +4,15 @@ package handlers
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"errors"
"log/slog" "log/slog"
"net/http" "net/http"
"time" "time"
"github.com/prometheus/client_golang/prometheus"
"go.uber.org/fx" "go.uber.org/fx"
"sneak.berlin/go/pixa/internal/allowlist" "sneak.berlin/go/pixa/internal/allowlist"
"sneak.berlin/go/pixa/internal/config" "sneak.berlin/go/pixa/internal/config"
"sneak.berlin/go/pixa/internal/database"
"sneak.berlin/go/pixa/internal/encurl" "sneak.berlin/go/pixa/internal/encurl"
"sneak.berlin/go/pixa/internal/healthcheck" "sneak.berlin/go/pixa/internal/healthcheck"
"sneak.berlin/go/pixa/internal/httpfetcher"
"sneak.berlin/go/pixa/internal/imgcache" "sneak.berlin/go/pixa/internal/imgcache"
"sneak.berlin/go/pixa/internal/logger" "sneak.berlin/go/pixa/internal/logger"
"sneak.berlin/go/pixa/internal/session" "sneak.berlin/go/pixa/internal/session"
@@ -28,29 +24,19 @@ type Params struct {
Logger *logger.Logger Logger *logger.Logger
Healthcheck *healthcheck.Healthcheck Healthcheck *healthcheck.Healthcheck
Database *database.Database
Config *config.Config Config *config.Config
// MetricsRegistry is the registry the image cache registers its ImageService *imgcache.Service
// counters on. pixad provides Prometheus' default registry, the one SessionManager *session.Manager
// /metrics serves. EncryptedURLGenerator *encurl.Generator
MetricsRegistry prometheus.Registerer
// Fetcher, when provided, fetches upstream images in place of the
// fetcher the handlers build from the config. Only tests provide one;
// pixad does not.
Fetcher httpfetcher.Fetcher `optional:"true"`
} }
// Handlers provides HTTP request handlers. // Handlers provides HTTP request handlers.
type Handlers struct { type Handlers struct {
log *slog.Logger log *slog.Logger
hc *healthcheck.Healthcheck hc *healthcheck.Healthcheck
db *database.Database
config *config.Config config *config.Config
fetcher httpfetcher.Fetcher
imgSvc *imgcache.Service imgSvc *imgcache.Service
imgCache *imgcache.Cache
sessMgr *session.Manager sessMgr *session.Manager
encGen *encurl.Generator encGen *encurl.Generator
csrfProtect func(http.Handler) http.Handler csrfProtect func(http.Handler) http.Handler
@@ -58,50 +44,25 @@ type Handlers struct {
// refererBlocklist matches the hosts of referer_blocklist; its IsAllowed // refererBlocklist matches the hosts of referer_blocklist; its IsAllowed
// reports whether a URL's host is on that list. // reports whether a URL's host is on that list.
refererBlocklist *allowlist.HostAllowList refererBlocklist *allowlist.HostAllowList
// metricsRegistry is the registry the image cache registers its
// counters on.
metricsRegistry prometheus.Registerer
} }
// New creates a new Handlers instance. // New creates a new Handlers instance.
func New(lc fx.Lifecycle, params Params) (*Handlers, error) { func New(_ fx.Lifecycle, params Params) (*Handlers, error) {
csrfProtect, err := newCSRFProtect(params.Config.SigningKey, params.Config.Debug) csrfProtect, err := newCSRFProtect(params.Config.SigningKey, params.Config.Debug)
if err != nil { if err != nil {
return nil, err return nil, err
} }
s := &Handlers{ return &Handlers{
log: params.Logger.Get(), log: params.Logger.Get(),
hc: params.Healthcheck, hc: params.Healthcheck,
db: params.Database,
config: params.Config, config: params.Config,
fetcher: params.Fetcher, imgSvc: params.ImageService,
sessMgr: params.SessionManager,
encGen: params.EncryptedURLGenerator,
csrfProtect: csrfProtect, csrfProtect: csrfProtect,
refererBlocklist: allowlist.New(params.Config.RefererBlocklist), refererBlocklist: allowlist.New(params.Config.RefererBlocklist),
metricsRegistry: params.MetricsRegistry, }, nil
}
lc.Append(fx.Hook{
//nolint:contextcheck // the cache's goroutines outlive OnStart; OnStop stops them
OnStart: func(_ context.Context) error {
return s.initImageService()
},
// The pending access times are written here, before the database's
// stop hook closes it.
OnStop: func(ctx context.Context) error {
if s.imgCache == nil {
return nil
}
return errors.Join(
s.imgCache.StopEviction(ctx),
s.imgCache.StopPendingAccessTimeWrites(ctx),
)
},
})
return s, nil
} }
// WaitForProcessing waits until no image is being processed, or until ctx // WaitForProcessing waits until no image is being processed, or until ctx
@@ -110,96 +71,6 @@ func (s *Handlers) WaitForProcessing(ctx context.Context) int {
return s.imgSvc.WaitForProcessing(ctx) return s.imgSvc.WaitForProcessing(ctx)
} }
// newCacheConfig builds the image cache's configuration from cfg.
// cache_max_bytes: 0 disables the disk cache entirely; any other value
// is the eviction limit in bytes; when it is omitted, the cache works
// out the default limit itself.
func newCacheConfig(cfg *config.Config, log *slog.Logger) imgcache.CacheConfig {
return imgcache.CacheConfig{
StateDir: cfg.StateDir,
NegativeTTL: imgcache.DefaultNegativeTTL,
MaxBytes: cfg.CacheMaxBytes,
UseDefaultMaxBytes: !cfg.CacheMaxBytesExplicit,
DisableDiskCache: cfg.CacheMaxBytesExplicit && cfg.CacheMaxBytes == 0,
Logger: log,
}
}
// initImageService initializes the image cache and service.
func (s *Handlers) initImageService() error {
cache, err := imgcache.NewCache(
s.db.DB(), newCacheConfig(s.config, s.log), s.metricsRegistry)
if err != nil {
return err
}
s.imgCache = cache
// Background eviction: startup reconciliation, then periodic and
// write-pressure passes, and the removal of expired negative cache
// rows, which is all it does when the disk cache is disabled.
cache.StartEviction(imgcache.DefaultEvictionInterval)
// Writes the access times that requests hand over, so that no request
// waits for the database to record them
cache.StartPendingAccessTimeWrites()
// Create the fetcher config
fetcherCfg := httpfetcher.DefaultConfig()
fetcherCfg.AllowHTTP = s.config.AllowHTTP
fetcherCfg.Timeout = s.config.UpstreamFetchTimeout
fetcherCfg.MaxResponseSize = s.config.UpstreamMaxResponseSize
if s.config.UpstreamConnectionsPerHost > 0 {
fetcherCfg.MaxConnectionsPerHost = s.config.UpstreamConnectionsPerHost
}
fetcherCfg.MaxConnections = s.config.UpstreamConnections
fetcherCfg.BlockedNetworks = s.config.BlockedNetworks
// Create the service. With no fetcher provided, it builds its own from
// fetcherCfg.
svc, err := imgcache.NewService(&imgcache.ServiceConfig{
Cache: cache,
FetcherConfig: fetcherCfg,
Fetcher: s.fetcher,
SigningKey: s.config.SigningKey,
Allowlist: s.config.AllowlistHosts,
MaxConcurrentProcessing: s.config.MaxConcurrentProcessing,
Logger: s.log,
})
if err != nil {
return err
}
s.imgSvc = svc
s.log.Info("image service initialized")
// Initialize session manager (signing key is validated at config load
// time). Session cookies are always Secure/HttpOnly/SameSite=Strict.
// The random value that each logout replaces is saved in the database,
// so it outlasts a restart.
sessMgr, err := session.NewManagerWithDatabase(
context.Background(), s.config.SigningKey, s.db.DB())
if err != nil {
return err
}
s.sessMgr = sessMgr
// Initialize encrypted URL generator
encGen, err := encurl.NewGenerator(s.config.SigningKey)
if err != nil {
return err
}
s.encGen = encGen
s.log.Info("session manager and URL generator initialized")
return nil
}
func (s *Handlers) respondJSON(w http.ResponseWriter, data any, status int) { func (s *Handlers) respondJSON(w http.ResponseWriter, data any, status int) {
w.Header().Set("Content-Type", "application/json") w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status) w.WriteHeader(status)
+11 -4
View File
@@ -21,6 +21,7 @@ import (
"github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus"
"sneak.berlin/go/pixa/internal/database" "sneak.berlin/go/pixa/internal/database"
"sneak.berlin/go/pixa/internal/httpfetcher" "sneak.berlin/go/pixa/internal/httpfetcher"
"sneak.berlin/go/pixa/internal/imageprocessor"
"sneak.berlin/go/pixa/internal/imgcache" "sneak.berlin/go/pixa/internal/imgcache"
) )
@@ -48,16 +49,22 @@ func setupTestHandler(t *testing.T) *testFixtures {
cache, err := imgcache.NewCache(db, imgcache.CacheConfig{ cache, err := imgcache.NewCache(db, imgcache.CacheConfig{
StateDir: tmpDir, StateDir: tmpDir,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
MaxBytes: 1 << 30,
}, prometheus.NewRegistry()) }, prometheus.NewRegistry())
if err != nil { if err != nil {
t.Fatalf("failed to create cache: %v", err) t.Fatalf("failed to create cache: %v", err)
} }
svc, err := imgcache.NewService(&imgcache.ServiceConfig{ svc, err := imgcache.NewService(&imgcache.ServiceConfig{
Cache: cache, Cache: cache,
Fetcher: newMockFetcher(mockFS), Fetcher: newMockFetcher(mockFS),
SigningKey: "test-signing-key-must-be-32-chars", Processor: imageprocessor.New(imageprocessor.Params{
Allowlist: []string{goodHost}, MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 2,
}),
SigningKey: "test-signing-key-must-be-32-chars",
Allowlist: []string{goodHost},
MaxResponseSize: 50 << 20,
}) })
if err != nil { if err != nil {
t.Fatalf("failed to create service: %v", err) t.Fatalf("failed to create service: %v", err)
@@ -15,6 +15,7 @@ import (
"github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus"
"sneak.berlin/go/pixa/internal/encurl" "sneak.berlin/go/pixa/internal/encurl"
"sneak.berlin/go/pixa/internal/imageprocessor"
"sneak.berlin/go/pixa/internal/imagerequest" "sneak.berlin/go/pixa/internal/imagerequest"
"sneak.berlin/go/pixa/internal/imgcache" "sneak.berlin/go/pixa/internal/imgcache"
) )
@@ -34,6 +35,7 @@ func newSignedHostServer(
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{ cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
StateDir: t.TempDir(), StateDir: t.TempDir(),
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
MaxBytes: 1 << 30,
}, prometheus.NewRegistry()) }, prometheus.NewRegistry())
if err != nil { if err != nil {
t.Fatalf("imgcache.NewCache() error = %v", err) t.Fatalf("imgcache.NewCache() error = %v", err)
@@ -46,8 +48,13 @@ func newSignedHostServer(
Fetcher: newMockFetcher(fstest.MapFS{ Fetcher: newMockFetcher(fstest.MapFS{
signedHost + photoPath: &fstest.MapFile{Data: jpegData}, signedHost + photoPath: &fstest.MapFile{Data: jpegData},
}), }),
SigningKey: testSigningKey, Processor: imageprocessor.New(imageprocessor.Params{
Logger: log, MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 2,
}),
SigningKey: testSigningKey,
MaxResponseSize: 50 << 20,
Logger: log,
}) })
if err != nil { if err != nil {
t.Fatalf("imgcache.NewService() error = %v", err) t.Fatalf("imgcache.NewService() error = %v", err)
@@ -15,6 +15,7 @@ import (
"github.com/go-chi/chi/v5" "github.com/go-chi/chi/v5"
"github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus"
"sneak.berlin/go/pixa/internal/httpfetcher" "sneak.berlin/go/pixa/internal/httpfetcher"
"sneak.berlin/go/pixa/internal/imageprocessor"
"sneak.berlin/go/pixa/internal/imagerequest" "sneak.berlin/go/pixa/internal/imagerequest"
"sneak.berlin/go/pixa/internal/imgcache" "sneak.berlin/go/pixa/internal/imgcache"
"sneak.berlin/go/pixa/internal/signature" "sneak.berlin/go/pixa/internal/signature"
@@ -32,16 +33,22 @@ func newImageRoute(t *testing.T, fetcher httpfetcher.Fetcher) http.Handler {
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{ cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
StateDir: t.TempDir(), StateDir: t.TempDir(),
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
MaxBytes: 1 << 30,
}, prometheus.NewRegistry()) }, prometheus.NewRegistry())
if err != nil { if err != nil {
t.Fatalf("failed to create cache: %v", err) t.Fatalf("failed to create cache: %v", err)
} }
svc, err := imgcache.NewService(&imgcache.ServiceConfig{ svc, err := imgcache.NewService(&imgcache.ServiceConfig{
Cache: cache, Cache: cache,
Fetcher: fetcher, Fetcher: fetcher,
SigningKey: testSigningKey, Processor: imageprocessor.New(imageprocessor.Params{
Allowlist: []string{allowlistedHost}, MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 2,
}),
SigningKey: testSigningKey,
Allowlist: []string{allowlistedHost},
MaxResponseSize: 50 << 20,
}) })
if err != nil { if err != nil {
t.Fatalf("failed to create service: %v", err) t.Fatalf("failed to create service: %v", err)
@@ -169,7 +176,12 @@ func TestHandleImage_ErrorAnswers(t *testing.T) {
photos := newPhotoFetcher(t, photos := newPhotoFetcher(t,
signedHost, parentHost, siblingHost, subdomainHost, appendedHost) signedHost, parentHost, siblingHost, subdomainHost, appendedHost)
// The real fetcher refuses localhost before any lookup or connection. // The real fetcher refuses localhost before any lookup or connection.
realFetcher := httpfetcher.New(httpfetcher.DefaultConfig()) realFetcher := httpfetcher.New(&httpfetcher.Config{
Timeout: 30 * time.Second,
MaxResponseSize: 50 << 20,
MaxConnectionsPerHost: 20,
MaxConnections: 64,
})
exp := time.Now().Add(time.Hour) exp := time.Now().Add(time.Hour)
expired := time.Now().Add(-time.Hour) expired := time.Now().Add(-time.Hour)
@@ -14,6 +14,7 @@ import (
"sneak.berlin/go/pixa/internal/allowlist" "sneak.berlin/go/pixa/internal/allowlist"
"sneak.berlin/go/pixa/internal/encurl" "sneak.berlin/go/pixa/internal/encurl"
"sneak.berlin/go/pixa/internal/httpfetcher" "sneak.berlin/go/pixa/internal/httpfetcher"
"sneak.berlin/go/pixa/internal/imageprocessor"
"sneak.berlin/go/pixa/internal/imgcache" "sneak.berlin/go/pixa/internal/imgcache"
) )
@@ -51,16 +52,22 @@ func newRefererRoutes(t *testing.T) (http.Handler, *Handlers, *countingFetcher)
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{ cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
StateDir: t.TempDir(), StateDir: t.TempDir(),
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
MaxBytes: 1 << 30,
}, prometheus.NewRegistry()) }, prometheus.NewRegistry())
if err != nil { if err != nil {
t.Fatalf("imgcache.NewCache() error = %v", err) t.Fatalf("imgcache.NewCache() error = %v", err)
} }
svc, err := imgcache.NewService(&imgcache.ServiceConfig{ svc, err := imgcache.NewService(&imgcache.ServiceConfig{
Cache: cache, Cache: cache,
Fetcher: fetcher, Fetcher: fetcher,
SigningKey: testSigningKey, Processor: imageprocessor.New(imageprocessor.Params{
Allowlist: []string{allowlistedHost}, MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 2,
}),
SigningKey: testSigningKey,
Allowlist: []string{allowlistedHost},
MaxResponseSize: 50 << 20,
}) })
if err != nil { if err != nil {
t.Fatalf("imgcache.NewService() error = %v", err) t.Fatalf("imgcache.NewService() error = %v", err)
@@ -7,6 +7,7 @@ import (
"net/http" "net/http"
"net/netip" "net/netip"
"testing" "testing"
"time"
) )
// TestIsPrivateIPBlocksSpecialRanges covers the internal and special-use // TestIsPrivateIPBlocksSpecialRanges covers the internal and special-use
@@ -70,12 +71,15 @@ func transportOf(t *testing.T, f *HTTPFetcher) *http.Transport {
func TestDialerEnforcesBlockedNetworks(t *testing.T) { func TestDialerEnforcesBlockedNetworks(t *testing.T) {
t.Parallel() t.Parallel()
cfg := DefaultConfig() transport := transportOf(t, New(&Config{
// TEST-NET-2 (198.51.100.0/24) is public to the built-in check, so Timeout: 30 * time.Second,
// blocking it can only come from the operator-supplied list. MaxResponseSize: 50 << 20,
cfg.BlockedNetworks = []netip.Prefix{netip.MustParsePrefix("198.51.100.0/24")} MaxConnectionsPerHost: 20,
MaxConnections: 64,
transport := transportOf(t, New(cfg)) // TEST-NET-2 (198.51.100.0/24) is public to the built-in check, so
// blocking it can only come from the operator-supplied list.
BlockedNetworks: []netip.Prefix{netip.MustParsePrefix("198.51.100.0/24")},
}))
blocked := []string{ blocked := []string{
"198.51.100.5:80", // operator-supplied range "198.51.100.5:80", // operator-supplied range
@@ -4,6 +4,7 @@ import (
"errors" "errors"
"net" "net"
"testing" "testing"
"time"
) )
// TestNewUsesCheckedDialerWithoutDialContext checks that a fetcher built // TestNewUsesCheckedDialerWithoutDialContext checks that a fetcher built
@@ -13,7 +14,12 @@ func TestNewUsesCheckedDialerWithoutDialContext(t *testing.T) {
t.Parallel() t.Parallel()
srv := startUpstream(t) srv := startUpstream(t)
transport := transportOf(t, New(DefaultConfig())) transport := transportOf(t, New(&Config{
Timeout: 30 * time.Second,
MaxResponseSize: 50 << 20,
MaxConnectionsPerHost: 20,
MaxConnections: 64,
}))
addr := srv.Listener.Addr().String() addr := srv.Listener.Addr().String()
@@ -32,10 +38,14 @@ func TestDialContextReplacesOnlyTheDialer(t *testing.T) {
srv := startUpstream(t) srv := startUpstream(t)
dialer := &recordingDialer{target: srv.Listener.Addr().String()} dialer := &recordingDialer{target: srv.Listener.Addr().String()}
cfg := DefaultConfig() f := New(&Config{
cfg.AllowHTTP = true Timeout: 30 * time.Second,
cfg.DialContext = dialer.dialContext MaxResponseSize: 50 << 20,
f := New(cfg) AllowHTTP: true,
MaxConnectionsPerHost: 20,
MaxConnections: 64,
DialContext: dialer.dialContext,
})
if body := fetchBody(t, f, "/image"); body != imagePayload { if body := fetchBody(t, f, "/image"); body != imagePayload {
t.Errorf("body = %q, want %q", body, imagePayload) t.Errorf("body = %q, want %q", body, imagePayload)
+30 -9
View File
@@ -120,7 +120,12 @@ func newServerFetcher(
t.Helper() t.Helper()
if cfg == nil { if cfg == nil {
cfg = DefaultConfig() cfg = &Config{
Timeout: 30 * time.Second,
MaxResponseSize: 50 << 20,
MaxConnectionsPerHost: 20,
MaxConnections: 64,
}
} }
cfg.AllowHTTP = true cfg.AllowHTTP = true
@@ -280,8 +285,12 @@ func TestFetchMaxResponseSizeEnforced(t *testing.T) {
srv := startUpstream(t) srv := startUpstream(t)
cfg := DefaultConfig() cfg := &Config{
cfg.MaxResponseSize = 8 Timeout: 30 * time.Second,
MaxResponseSize: 8,
MaxConnectionsPerHost: 20,
MaxConnections: 64,
}
f, _ := newServerFetcher(t, srv, cfg) f, _ := newServerFetcher(t, srv, cfg)
@@ -310,8 +319,12 @@ func fetchWithMaxResponseSize(
srv := httptest.NewServer(handler) srv := httptest.NewServer(handler)
t.Cleanup(srv.Close) t.Cleanup(srv.Close)
cfg := DefaultConfig() cfg := &Config{
cfg.MaxResponseSize = maxResponseSize Timeout: 30 * time.Second,
MaxResponseSize: maxResponseSize,
MaxConnectionsPerHost: 20,
MaxConnections: 64,
}
f, _ := newServerFetcher(t, srv, cfg) f, _ := newServerFetcher(t, srv, cfg)
@@ -436,8 +449,12 @@ func TestFetchSemaphoreReleasedOnError(t *testing.T) {
srv := startUpstream(t) srv := startUpstream(t)
cfg := DefaultConfig() cfg := &Config{
cfg.MaxConnectionsPerHost = 1 Timeout: 30 * time.Second,
MaxResponseSize: 50 << 20,
MaxConnectionsPerHost: 1,
MaxConnections: 64,
}
f, _ := newServerFetcher(t, srv, cfg) f, _ := newServerFetcher(t, srv, cfg)
@@ -466,8 +483,12 @@ func assertSlotReleasedByClose(
srv := startUpstream(t) srv := startUpstream(t)
cfg := DefaultConfig() cfg := &Config{
cfg.MaxConnectionsPerHost = 1 Timeout: 30 * time.Second,
MaxResponseSize: 50 << 20,
MaxConnectionsPerHost: 1,
MaxConnections: 64,
}
f, _ := newServerFetcher(t, srv, cfg) f, _ := newServerFetcher(t, srv, cfg)
+25 -40
View File
@@ -24,16 +24,15 @@ import (
// Fetcher configuration constants. // Fetcher configuration constants.
const ( const (
DefaultFetchTimeout = 30 * time.Second DefaultTLSTimeout = 10 * time.Second
DefaultMaxResponseSize = 50 << 20 // 50MB DefaultMaxIdleConns = 100
DefaultTLSTimeout = 10 * time.Second DefaultIdleConnTimeout = 90 * time.Second
DefaultMaxIdleConns = 100 DefaultMaxRedirects = 10
DefaultIdleConnTimeout = 90 * time.Second
DefaultMaxRedirects = 10
DefaultMaxConnectionsPerHost = 20
DefaultMaxConnections = 64
) )
// userAgent is the User-Agent header sent to upstream servers.
const userAgent = "pixa/1.0"
// ConnectionWaitTimeout is how long Fetch waits for a free connection when // ConnectionWaitTimeout is how long Fetch waits for a free connection when
// MaxConnections fetches are already in progress. // MaxConnections fetches are already in progress.
const ConnectionWaitTimeout = 10 * time.Second const ConnectionWaitTimeout = 10 * time.Second
@@ -120,16 +119,26 @@ type FetchResult struct {
TLSCipherSuite string TLSCipherSuite string
} }
// Config holds configuration for the upstream fetcher. // allowedContentTypes returns the content types an upstream response may
// have, which Fetch also sends as its Accept header.
func allowedContentTypes() []string {
return []string{
contentTypeJPEG,
contentTypePNG,
contentTypeGIF,
contentTypeWebP,
contentTypeAVIF,
contentTypeJXL,
}
}
// Config holds configuration for the upstream fetcher. Every field but
// AllowHTTP, BlockedNetworks and DialContext must be set.
type Config struct { type Config struct {
// Timeout for upstream requests. // Timeout for upstream requests.
Timeout time.Duration Timeout time.Duration
// MaxResponseSize is the maximum allowed response body size. // MaxResponseSize is the maximum allowed response body size.
MaxResponseSize int64 MaxResponseSize int64
// UserAgent to send to upstream servers.
UserAgent string
// AllowedContentTypes is an allow list of MIME types to accept.
AllowedContentTypes []string
// AllowHTTP allows non-TLS connections (for testing only). // AllowHTTP allows non-TLS connections (for testing only).
AllowHTTP bool AllowHTTP bool
// MaxConnectionsPerHost limits concurrent connections to each upstream host. // MaxConnectionsPerHost limits concurrent connections to each upstream host.
@@ -147,26 +156,6 @@ type Config struct {
DialContext func(ctx context.Context, network, addr string) (net.Conn, error) DialContext func(ctx context.Context, network, addr string) (net.Conn, error)
} }
// DefaultConfig returns a Config with sensible defaults.
func DefaultConfig() *Config {
return &Config{
Timeout: DefaultFetchTimeout,
MaxResponseSize: DefaultMaxResponseSize,
UserAgent: "pixa/1.0",
AllowedContentTypes: []string{
contentTypeJPEG,
contentTypePNG,
contentTypeGIF,
contentTypeWebP,
contentTypeAVIF,
contentTypeJXL,
},
AllowHTTP: false,
MaxConnectionsPerHost: DefaultMaxConnectionsPerHost,
MaxConnections: DefaultMaxConnections,
}
}
// HTTPFetcher implements Fetcher with SSRF protection and connection limits // HTTPFetcher implements Fetcher with SSRF protection and connection limits
// per host and for all hosts together. // per host and for all hosts together.
type HTTPFetcher struct { type HTTPFetcher struct {
@@ -194,10 +183,6 @@ type hostSemaphore struct {
// New creates a new HTTPFetcher with SSRF protection. // New creates a new HTTPFetcher with SSRF protection.
func New(config *Config) *HTTPFetcher { func New(config *Config) *HTTPFetcher {
if config == nil {
config = DefaultConfig()
}
// Unless config.DialContext replaces it, the transport connects with // Unless config.DialContext replaces it, the transport connects with
// the SSRF-safe dialer, which re-resolves and re-checks at connect time // the SSRF-safe dialer, which re-resolves and re-checks at connect time
// (closing the DNS-rebinding window) against both the built-in ranges // (closing the DNS-rebinding window) against both the built-in ranges
@@ -280,8 +265,8 @@ func (f *HTTPFetcher) Fetch(ctx context.Context, url string) (*FetchResult, erro
Header: make(http.Header), Header: make(http.Header),
} }
req.Header.Set("User-Agent", f.config.UserAgent) req.Header.Set("User-Agent", userAgent)
req.Header.Set("Accept", strings.Join(f.config.AllowedContentTypes, ", ")) req.Header.Set("Accept", strings.Join(allowedContentTypes(), ", "))
// The ID of the request this fetch serves, so the fetch can be found in // The ID of the request this fetch serves, so the fetch can be found in
// the upstream host's logs // the upstream host's logs
@@ -460,7 +445,7 @@ func (f *HTTPFetcher) isAllowedContentType(contentType string) bool {
// Extract the MIME type without parameters // Extract the MIME type without parameters
mediaType := strings.TrimSpace(strings.Split(contentType, ";")[0]) mediaType := strings.TrimSpace(strings.Split(contentType, ";")[0])
for _, allowed := range f.config.AllowedContentTypes { for _, allowed := range allowedContentTypes() {
if strings.EqualFold(mediaType, allowed) { if strings.EqualFold(mediaType, allowed) {
return true return true
} }
@@ -7,60 +7,21 @@ import (
"net" "net"
"testing" "testing"
"testing/fstest" "testing/fstest"
"time"
) )
// testHost is the hostname used by mock fetch tests. // testHost is the hostname used by mock fetch tests.
const testHost = "example.com" const testHost = "example.com"
func TestDefaultConfig(t *testing.T) {
t.Parallel()
cfg := DefaultConfig()
if cfg.Timeout != DefaultFetchTimeout {
t.Errorf("Timeout = %v, want %v", cfg.Timeout, DefaultFetchTimeout)
}
if cfg.MaxResponseSize != DefaultMaxResponseSize {
t.Errorf("MaxResponseSize = %d, want %d", cfg.MaxResponseSize, DefaultMaxResponseSize)
}
if cfg.MaxConnectionsPerHost != DefaultMaxConnectionsPerHost {
t.Errorf("MaxConnectionsPerHost = %d, want %d",
cfg.MaxConnectionsPerHost, DefaultMaxConnectionsPerHost)
}
if cfg.AllowHTTP {
t.Error("AllowHTTP should default to false")
}
if len(cfg.AllowedContentTypes) == 0 {
t.Error("AllowedContentTypes should not be empty")
}
}
func TestNewWithNilConfigUsesDefaults(t *testing.T) {
t.Parallel()
f := New(nil)
if f == nil {
t.Fatal("New(nil) returned nil")
}
if f.config == nil {
t.Fatal("config should be populated from DefaultConfig")
}
if f.config.Timeout != DefaultFetchTimeout {
t.Errorf("Timeout = %v, want %v", f.config.Timeout, DefaultFetchTimeout)
}
}
func TestIsAllowedContentType(t *testing.T) { func TestIsAllowedContentType(t *testing.T) {
t.Parallel() t.Parallel()
f := New(DefaultConfig()) f := New(&Config{
Timeout: 30 * time.Second,
MaxResponseSize: 50 << 20,
MaxConnectionsPerHost: 20,
MaxConnections: 64,
})
tests := []struct { tests := []struct {
contentType string contentType string
+14 -4
View File
@@ -1,16 +1,26 @@
package httpfetcher package httpfetcher
import "testing" import (
"testing"
"time"
)
// TestJPEGXLContentType verifies that the fetcher accepts an upstream answer // TestJPEGXLContentType verifies that the fetcher accepts an upstream answer
// of type image/jxl by default, and that the mock fetcher serves a .jxl file // of type image/jxl, and that the mock fetcher serves a .jxl file as
// as image/jxl. // image/jxl.
func TestJPEGXLContentType(t *testing.T) { func TestJPEGXLContentType(t *testing.T) {
t.Parallel() t.Parallel()
const jxlType = "image/jxl" const jxlType = "image/jxl"
if !New(DefaultConfig()).isAllowedContentType(jxlType) { f := New(&Config{
Timeout: 30 * time.Second,
MaxResponseSize: 50 << 20,
MaxConnectionsPerHost: 20,
MaxConnections: 64,
})
if !f.isAllowedContentType(jxlType) {
t.Errorf("isAllowedContentType(%q) = false, want true", jxlType) t.Errorf("isAllowedContentType(%q) = false, want true", jxlType)
} }
@@ -18,14 +18,6 @@ func imageURLOnPort(port int) string {
"/image" "/image"
} }
func TestDefaultConfigMaxConnections(t *testing.T) {
t.Parallel()
if got := DefaultConfig().MaxConnections; got != DefaultMaxConnections {
t.Errorf("MaxConnections = %d, want %d", got, DefaultMaxConnections)
}
}
// TestFetchLimitsConnectionsToAllHostsTogether checks that MaxConnections // TestFetchLimitsConnectionsToAllHostsTogether checks that MaxConnections
// counts the fetches to every host together, apart from the per-host // counts the fetches to every host together, apart from the per-host
// limit: with MaxConnections at 2 and two responses open from two hosts, a // limit: with MaxConnections at 2 and two responses open from two hosts, a
@@ -37,8 +29,12 @@ func TestFetchLimitsConnectionsToAllHostsTogether(t *testing.T) {
srv := startUpstream(t) srv := startUpstream(t)
cfg := DefaultConfig() cfg := &Config{
cfg.MaxConnections = 2 Timeout: 30 * time.Second,
MaxResponseSize: 50 << 20,
MaxConnectionsPerHost: 20,
MaxConnections: 2,
}
f, _ := newServerFetcher(t, srv, cfg) f, _ := newServerFetcher(t, srv, cfg)
f.connectionWaitTimeout = 100 * time.Millisecond f.connectionWaitTimeout = 100 * time.Millisecond
@@ -94,8 +90,12 @@ func TestFetchFreesHostSlotWhenContextEndsWaitingForConnection(t *testing.T) {
srv := startUpstream(t) srv := startUpstream(t)
cfg := DefaultConfig() cfg := &Config{
cfg.MaxConnections = 1 Timeout: 30 * time.Second,
MaxResponseSize: 50 << 20,
MaxConnectionsPerHost: 20,
MaxConnections: 1,
}
f, _ := newServerFetcher(t, srv, cfg) f, _ := newServerFetcher(t, srv, cfg)
@@ -163,9 +163,12 @@ func TestFetchRemovesHostSemaphoreWhenNoConnection(t *testing.T) {
srv := startUpstream(t) srv := startUpstream(t)
cfg := DefaultConfig() cfg := &Config{
cfg.MaxConnections = 1 Timeout: 30 * time.Second,
cfg.MaxConnectionsPerHost = 1 MaxResponseSize: 50 << 20,
MaxConnectionsPerHost: 1,
MaxConnections: 1,
}
f, _ := newServerFetcher(t, srv, cfg) f, _ := newServerFetcher(t, srv, cfg)
f.connectionWaitTimeout = 100 * time.Millisecond f.connectionWaitTimeout = 100 * time.Millisecond
@@ -236,8 +239,12 @@ func TestFetchReleasesConnectionOnError(t *testing.T) {
srv := startUpstream(t) srv := startUpstream(t)
cfg := DefaultConfig() cfg := &Config{
cfg.MaxConnections = 1 Timeout: 30 * time.Second,
MaxResponseSize: 50 << 20,
MaxConnectionsPerHost: 20,
MaxConnections: 1,
}
f, _ := newServerFetcher(t, srv, cfg) f, _ := newServerFetcher(t, srv, cfg)
f.connectionWaitTimeout = 100 * time.Millisecond f.connectionWaitTimeout = 100 * time.Millisecond
@@ -69,7 +69,10 @@ func TestImageProcessor_RefusesAVIFAboveMaxPixels(t *testing.T) {
t.Run(tt.name, func(t *testing.T) { t.Run(tt.name, func(t *testing.T) {
t.Parallel() t.Parallel()
_, err := New(Params{}).Process(t.Context(), bytes.NewReader(tt.input), _, err := New(Params{
MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 1,
}).Process(t.Context(), bytes.NewReader(tt.input),
&imagerequest.Transform{ &imagerequest.Transform{
Size: imagerequest.Size{Width: tt.width, Height: tt.height}, Size: imagerequest.Size{Width: tt.width, Height: tt.height},
Format: tt.format, Format: tt.format,
@@ -38,7 +38,10 @@ func TestImageProcessor_RefusesComputedSideAboveMaximum(t *testing.T) {
input := createTestJPEG(t, tt.inputWidth, tt.inputHeight) input := createTestJPEG(t, tt.inputWidth, tt.inputHeight)
result, err := New(Params{}).Process(t.Context(), result, err := New(Params{
MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 1,
}).Process(t.Context(),
bytes.NewReader(input), bytes.NewReader(input),
&imagerequest.Transform{ &imagerequest.Transform{
Size: tt.size, Size: tt.size,
@@ -104,7 +107,10 @@ func TestImageProcessor_ComputedSideIsAtLeastOne(t *testing.T) {
input := createTestJPEG(t, tt.inputWidth, tt.inputHeight) input := createTestJPEG(t, tt.inputWidth, tt.inputHeight)
result, err := New(Params{}).Process(t.Context(), result, err := New(Params{
MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 1,
}).Process(t.Context(),
bytes.NewReader(input), bytes.NewReader(input),
&imagerequest.Transform{ &imagerequest.Transform{
Size: tt.size, Size: tt.size,
@@ -84,7 +84,7 @@ func checkProcessStopsAtDeadline(
) { ) {
t.Helper() t.Helper()
proc := New(Params{MaxConcurrentProcessing: 1}) proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1})
result, err := proc.Process(ctx, input, smallJPEGTransform()) result, err := proc.Process(ctx, input, smallJPEGTransform())
if result != nil { if result != nil {
@@ -14,7 +14,10 @@ import (
func TestImageProcessor_EmptyFormatRefused(t *testing.T) { func TestImageProcessor_EmptyFormatRefused(t *testing.T) {
t.Parallel() t.Parallel()
_, err := New(Params{}).Process( _, err := New(Params{
MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 1,
}).Process(
t.Context(), bytes.NewReader(createTestJPEG(t, 20, 20)), t.Context(), bytes.NewReader(createTestJPEG(t, 20, 20)),
&imagerequest.Transform{}, &imagerequest.Transform{},
) )
@@ -18,7 +18,10 @@ func processedSize(
) int64 { ) int64 {
t.Helper() t.Helper()
result, err := New(Params{}).Process(t.Context(), bytes.NewReader(input), transform) result, err := New(Params{
MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 1,
}).Process(t.Context(), bytes.NewReader(input), transform)
if err != nil { if err != nil {
t.Fatalf("Process() error = %v", err) t.Fatalf("Process() error = %v", err)
} }
@@ -36,7 +36,10 @@ func TestImageProcessor_FitCoverServesThinSource(t *testing.T) {
input := createTestJPEG(t, tt.inputWidth, tt.inputHeight) input := createTestJPEG(t, tt.inputWidth, tt.inputHeight)
result, err := New(Params{}).Process(t.Context(), result, err := New(Params{
MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 1,
}).Process(t.Context(),
bytes.NewReader(input), bytes.NewReader(input),
&imagerequest.Transform{ &imagerequest.Transform{
Size: tt.size, Size: tt.size,
@@ -106,7 +109,10 @@ func TestImageProcessor_FitCoverCropsTheCentre(t *testing.T) {
input := createBandsPNG(t, tt.inputWidth, tt.inputHeight) input := createBandsPNG(t, tt.inputWidth, tt.inputHeight)
result, err := New(Params{}).Process(t.Context(), result, err := New(Params{
MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 1,
}).Process(t.Context(),
bytes.NewReader(input), bytes.NewReader(input),
&imagerequest.Transform{ &imagerequest.Transform{
Size: imagerequest.Size{Width: 50, Height: 50}, Size: imagerequest.Size{Width: 50, Height: 50},
@@ -230,7 +236,10 @@ func TestImageProcessor_FitCoverKeepsProportions(t *testing.T) {
input := createSquarePNG(t, tt.side, tt.squareSide) input := createSquarePNG(t, tt.side, tt.squareSide)
result, err := New(Params{}).Process(t.Context(), result, err := New(Params{
MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 1,
}).Process(t.Context(),
bytes.NewReader(input), bytes.NewReader(input),
&imagerequest.Transform{ &imagerequest.Transform{
Size: tt.size, Size: tt.size,
@@ -41,7 +41,10 @@ func TestImageProcessor_RefusesFitOutsideAboveMaximum(t *testing.T) {
input := createTestJPEG(t, tt.inputWidth, tt.inputHeight) input := createTestJPEG(t, tt.inputWidth, tt.inputHeight)
result, err := New(Params{}).Process(t.Context(), result, err := New(Params{
MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 1,
}).Process(t.Context(),
bytes.NewReader(input), bytes.NewReader(input),
&imagerequest.Transform{ &imagerequest.Transform{
Size: tt.size, Size: tt.size,
@@ -95,7 +98,10 @@ func TestImageProcessor_FitOutsideSideIsAtLeastOne(t *testing.T) {
input := createTestJPEG(t, tt.inputWidth, tt.inputHeight) input := createTestJPEG(t, tt.inputWidth, tt.inputHeight)
result, err := New(Params{}).Process(t.Context(), result, err := New(Params{
MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 1,
}).Process(t.Context(),
bytes.NewReader(input), bytes.NewReader(input),
&imagerequest.Transform{ &imagerequest.Transform{
Size: tt.size, Size: tt.size,
+4 -22
View File
@@ -11,7 +11,6 @@ import (
"time" "time"
"github.com/davidbyttow/govips/v2/vips" "github.com/davidbyttow/govips/v2/vips"
"sneak.berlin/go/pixa/internal/config"
"sneak.berlin/go/pixa/internal/imagerequest" "sneak.berlin/go/pixa/internal/imagerequest"
) )
@@ -79,10 +78,6 @@ type Result struct {
// Images larger than this are rejected to prevent DoS via decompression bombs. // Images larger than this are rejected to prevent DoS via decompression bombs.
const MaxInputDimension = 8192 const MaxInputDimension = 8192
// DefaultMaxInputBytes is the default maximum input size in bytes (50 MiB).
// This matches the default upstream fetcher limit.
const DefaultMaxInputBytes = 50 << 20
// ErrInputTooLarge is returned when input image dimensions exceed MaxInputDimension. // ErrInputTooLarge is returned when input image dimensions exceed MaxInputDimension.
var ErrInputTooLarge = errors.New("input image dimensions exceed maximum") var ErrInputTooLarge = errors.New("input image dimensions exceed maximum")
@@ -142,35 +137,22 @@ type ImageProcessor struct {
processingWaitTimeout time.Duration processingWaitTimeout time.Duration
} }
// Params holds configuration for creating an ImageProcessor. // Params holds configuration for creating an ImageProcessor. Both fields
// Zero values use sensible defaults (MaxInputBytes defaults to DefaultMaxInputBytes). // must be at least 1.
type Params struct { type Params struct {
// MaxInputBytes is the maximum allowed input size in bytes. // MaxInputBytes is the maximum allowed input size in bytes.
// If <= 0, DefaultMaxInputBytes is used.
MaxInputBytes int64 MaxInputBytes int64
// MaxConcurrentProcessing is the most images processed at once. // MaxConcurrentProcessing is the most images processed at once.
// If <= 0, config.DefaultMaxConcurrentProcessing() is used.
MaxConcurrentProcessing int MaxConcurrentProcessing int
} }
// New creates a new image processor with the given parameters. // New creates a new image processor with the given parameters.
// A zero-value Params{} uses sensible defaults.
func New(params Params) *ImageProcessor { func New(params Params) *ImageProcessor {
initVips() initVips()
maxInputBytes := params.MaxInputBytes
if maxInputBytes <= 0 {
maxInputBytes = DefaultMaxInputBytes
}
maxConcurrentProcessing := params.MaxConcurrentProcessing
if maxConcurrentProcessing <= 0 {
maxConcurrentProcessing = config.DefaultMaxConcurrentProcessing()
}
return &ImageProcessor{ return &ImageProcessor{
maxInputBytes: maxInputBytes, maxInputBytes: params.MaxInputBytes,
processingSemaphore: make(chan struct{}, maxConcurrentProcessing), processingSemaphore: make(chan struct{}, params.MaxConcurrentProcessing),
processingWaitTimeout: ProcessingWaitTimeout, processingWaitTimeout: ProcessingWaitTimeout,
} }
} }
@@ -119,7 +119,7 @@ func detectMIME(data []byte) string {
func TestImageProcessor_ResizeJPEG(t *testing.T) { func TestImageProcessor_ResizeJPEG(t *testing.T) {
t.Parallel() t.Parallel()
proc := New(Params{}) proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1})
ctx := context.Background() ctx := context.Background()
input := createTestJPEG(t, 800, 600) input := createTestJPEG(t, 800, 600)
@@ -165,7 +165,7 @@ func TestImageProcessor_ResizeJPEG(t *testing.T) {
func TestImageProcessor_ConvertToPNG(t *testing.T) { func TestImageProcessor_ConvertToPNG(t *testing.T) {
t.Parallel() t.Parallel()
proc := New(Params{}) proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1})
ctx := context.Background() ctx := context.Background()
input := createTestJPEG(t, 200, 150) input := createTestJPEG(t, 200, 150)
@@ -201,7 +201,7 @@ func processAndCheckSize(
) { ) {
t.Helper() t.Helper()
proc := New(Params{}) proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1})
ctx := context.Background() ctx := context.Background()
input := createTestJPEG(t, inputW, inputH) input := createTestJPEG(t, inputW, inputH)
@@ -239,7 +239,7 @@ func TestImageProcessor_OriginalSize(t *testing.T) {
func TestImageProcessor_FitContain(t *testing.T) { func TestImageProcessor_FitContain(t *testing.T) {
t.Parallel() t.Parallel()
proc := New(Params{}) proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1})
ctx := context.Background() ctx := context.Background()
// 800x400 image (2:1 aspect) into 400x400 box with contain // 800x400 image (2:1 aspect) into 400x400 box with contain
@@ -285,7 +285,7 @@ func TestImageProcessor_ProportionalScale_HeightOnly(t *testing.T) {
func TestImageProcessor_ProcessPNG(t *testing.T) { func TestImageProcessor_ProcessPNG(t *testing.T) {
t.Parallel() t.Parallel()
proc := New(Params{}) proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1})
ctx := context.Background() ctx := context.Background()
input := createTestPNG(t, 400, 300) input := createTestPNG(t, 400, 300)
@@ -330,7 +330,7 @@ func TestImageProcessor_RejectsOversizedInput(t *testing.T) {
t.Run(tt.name, func(t *testing.T) { t.Run(tt.name, func(t *testing.T) {
t.Parallel() t.Parallel()
proc := New(Params{}) proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1})
ctx := context.Background() ctx := context.Background()
input := createTestJPEG(t, tt.width, tt.height) input := createTestJPEG(t, tt.width, tt.height)
@@ -356,7 +356,7 @@ func TestImageProcessor_RejectsOversizedInput(t *testing.T) {
func TestImageProcessor_AcceptsMaxDimensionInput(t *testing.T) { func TestImageProcessor_AcceptsMaxDimensionInput(t *testing.T) {
t.Parallel() t.Parallel()
proc := New(Params{}) proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1})
ctx := context.Background() ctx := context.Background()
// Create an image at exactly MaxInputDimension - should be accepted // Create an image at exactly MaxInputDimension - should be accepted
@@ -387,7 +387,7 @@ func encodeAndCheck(
) { ) {
t.Helper() t.Helper()
proc := New(Params{}) proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1})
ctx := context.Background() ctx := context.Background()
input := createTestJPEG(t, 200, 150) input := createTestJPEG(t, 200, 150)
@@ -436,7 +436,7 @@ func TestImageProcessor_EncodeWebP(t *testing.T) {
func TestImageProcessor_DecodeAVIF(t *testing.T) { func TestImageProcessor_DecodeAVIF(t *testing.T) {
t.Parallel() t.Parallel()
proc := New(Params{}) proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1})
ctx := context.Background() ctx := context.Background()
// Load test AVIF file // Load test AVIF file
@@ -478,7 +478,7 @@ func TestImageProcessor_RejectsOversizedInputData(t *testing.T) {
// Create a processor with a very small byte limit // Create a processor with a very small byte limit
const limit = 1024 const limit = 1024
proc := New(Params{MaxInputBytes: limit}) proc := New(Params{MaxInputBytes: limit, MaxConcurrentProcessing: 1})
ctx := context.Background() ctx := context.Background()
// Create a valid JPEG that exceeds the byte limit // Create a valid JPEG that exceeds the byte limit
@@ -511,7 +511,7 @@ func TestImageProcessor_AcceptsInputWithinLimit(t *testing.T) {
input := createTestJPEG(t, 10, 10) input := createTestJPEG(t, 10, 10)
limit := int64(len(input)) * 10 // 10× headroom limit := int64(len(input)) * 10 // 10× headroom
proc := New(Params{MaxInputBytes: limit}) proc := New(Params{MaxInputBytes: limit, MaxConcurrentProcessing: 1})
ctx := context.Background() ctx := context.Background()
transform := &imagerequest.Transform{ transform := &imagerequest.Transform{
@@ -529,22 +529,6 @@ func TestImageProcessor_AcceptsInputWithinLimit(t *testing.T) {
defer func() { _ = result.Content.Close() }() defer func() { _ = result.Content.Close() }()
} }
func TestImageProcessor_DefaultMaxInputBytes(t *testing.T) {
t.Parallel()
// Passing 0 should use the default
proc := New(Params{})
if proc.maxInputBytes != DefaultMaxInputBytes {
t.Errorf("maxInputBytes = %d, want %d", proc.maxInputBytes, DefaultMaxInputBytes)
}
// Passing negative should also use the default
proc = New(Params{MaxInputBytes: -1})
if proc.maxInputBytes != DefaultMaxInputBytes {
t.Errorf("maxInputBytes = %d, want %d", proc.maxInputBytes, DefaultMaxInputBytes)
}
}
func TestImageProcessor_EncodeAVIF(t *testing.T) { func TestImageProcessor_EncodeAVIF(t *testing.T) {
t.Parallel() t.Parallel()
@@ -558,7 +542,10 @@ func processAndDecode(
) *vips.ImageRef { ) *vips.ImageRef {
t.Helper() t.Helper()
result, err := New(Params{}).Process( result, err := New(Params{
MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 1,
}).Process(
context.Background(), bytes.NewReader(input), transform, context.Background(), bytes.NewReader(input), transform,
) )
if err != nil { if err != nil {
@@ -21,7 +21,10 @@ func TestImageProcessor_EncodeJPEGXL(t *testing.T) {
Format: imagerequest.FormatJXL, Format: imagerequest.FormatJXL,
} }
result, err := New(Params{}).Process( result, err := New(Params{
MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 1,
}).Process(
t.Context(), bytes.NewReader(createTestJPEG(t, 200, 150)), transform, t.Context(), bytes.NewReader(createTestJPEG(t, 200, 150)), transform,
) )
if err != nil { if err != nil {
@@ -64,7 +67,10 @@ func TestImageProcessor_JPEGXLQuality(t *testing.T) {
outputBytes := make(map[int]int64) outputBytes := make(map[int]int64)
for _, quality := range []int{30, 90} { for _, quality := range []int{30, 90} {
result, err := New(Params{}).Process(t.Context(), result, err := New(Params{
MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 1,
}).Process(t.Context(),
bytes.NewReader(input), bytes.NewReader(input),
&imagerequest.Transform{Format: imagerequest.FormatJXL, Quality: quality}) &imagerequest.Transform{Format: imagerequest.FormatJXL, Quality: quality})
if err != nil { if err != nil {
@@ -5,7 +5,6 @@ import (
"context" "context"
"errors" "errors"
"io" "io"
"runtime"
"strings" "strings"
"sync" "sync"
"testing" "testing"
@@ -117,47 +116,15 @@ func waitForEntries(t *testing.T, entered <-chan struct{}, count int) {
} }
} }
func TestNewDefaultsMaxConcurrentProcessingToHalfTheCPUs(t *testing.T) { func TestNewMakesMaxConcurrentProcessingSlots(t *testing.T) {
t.Parallel() t.Parallel()
for _, limit := range []int{0, -1} { proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 3})
proc := New(Params{MaxConcurrentProcessing: limit})
if got := cap(proc.processingSemaphore); got != max(runtime.GOMAXPROCS(0)/2, 1) {
t.Errorf("MaxConcurrentProcessing %d: %d slots, want %d, half the CPUs",
limit, got, max(runtime.GOMAXPROCS(0)/2, 1))
}
}
proc := New(Params{MaxConcurrentProcessing: 3})
if got := cap(proc.processingSemaphore); got != 3 { if got := cap(proc.processingSemaphore); got != 3 {
t.Errorf("MaxConcurrentProcessing 3: %d slots, want 3", got) t.Errorf("MaxConcurrentProcessing 3: %d slots, want 3", got)
} }
} }
// TestNewDefaultIsHalfTheCPUsAndAtLeastOne checks that New, given no
// MaxConcurrentProcessing, makes one slot for every two CPUs Go uses,
// rounded down, and one slot when Go uses one CPU.
//
//nolint:paralleltest // it sets GOMAXPROCS, which other tests read
func TestNewDefaultIsHalfTheCPUsAndAtLeastOne(t *testing.T) {
cpusBefore := runtime.GOMAXPROCS(0)
t.Cleanup(func() { runtime.GOMAXPROCS(cpusBefore) })
for _, tc := range []struct{ cpus, want int }{
{1, 1}, {2, 1}, {3, 1}, {8, 4},
} {
runtime.GOMAXPROCS(tc.cpus)
got := cap(New(Params{}).processingSemaphore)
t.Logf("GOMAXPROCS %d: %d slots", tc.cpus, got)
if got != tc.want {
t.Errorf("GOMAXPROCS %d: %d slots, want %d", tc.cpus, got, tc.want)
}
}
}
// TestProcessNeverExceedsMaxConcurrentProcessing starts more Process calls // TestProcessNeverExceedsMaxConcurrentProcessing starts more Process calls
// than MaxConcurrentProcessing allows and holds the first ones inside // than MaxConcurrentProcessing allows and holds the first ones inside
// Process until the test lets them go. No more than the limit may be // Process until the test lets them go. No more than the limit may be
@@ -171,7 +138,7 @@ func TestProcessNeverExceedsMaxConcurrentProcessing(t *testing.T) {
calls = 6 calls = 6
) )
proc := New(Params{MaxConcurrentProcessing: limit}) proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: limit})
input := createTestJPEG(t, 50, 50) input := createTestJPEG(t, 50, 50)
counter := &readingCounter{} counter := &readingCounter{}
@@ -218,7 +185,7 @@ func TestProcessNeverExceedsMaxConcurrentProcessing(t *testing.T) {
func TestProcessWaitsThenFailsWhenNoSlotFrees(t *testing.T) { func TestProcessWaitsThenFailsWhenNoSlotFrees(t *testing.T) {
t.Parallel() t.Parallel()
proc := New(Params{MaxConcurrentProcessing: 1}) proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1})
proc.processingWaitTimeout = 100 * time.Millisecond proc.processingWaitTimeout = 100 * time.Millisecond
input := createTestJPEG(t, 10, 10) input := createTestJPEG(t, 10, 10)
@@ -334,7 +301,7 @@ func TestProcessReleasesSlotOnError(t *testing.T) {
func TestWaitForProcessing(t *testing.T) { func TestWaitForProcessing(t *testing.T) {
t.Parallel() t.Parallel()
proc := New(Params{MaxConcurrentProcessing: 2}) proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 2})
gate := make(chan struct{}) gate := make(chan struct{})
entered := make(chan struct{}, 1) entered := make(chan struct{}, 1)
@@ -27,7 +27,10 @@ func TestImageProcessor_OrigOfUnnamedFormatRefused(t *testing.T) {
t.Fatalf("failed to save test TIFF: %v", err) t.Fatalf("failed to save test TIFF: %v", err)
} }
result, err := New(Params{}).Process( result, err := New(Params{
MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 1,
}).Process(
t.Context(), bytes.NewReader(tiff), t.Context(), bytes.NewReader(tiff),
&imagerequest.Transform{Format: imagerequest.FormatOriginal}, &imagerequest.Transform{Format: imagerequest.FormatOriginal},
) )
@@ -14,7 +14,7 @@ import (
func TestImageProcessor_LoadsEachSourceFormat(t *testing.T) { func TestImageProcessor_LoadsEachSourceFormat(t *testing.T) {
t.Parallel() t.Parallel()
proc := New(Params{}) proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1})
for _, format := range []imagerequest.Format{ for _, format := range []imagerequest.Format{
imagerequest.FormatJPEG, imagerequest.FormatPNG, imagerequest.FormatGIF, imagerequest.FormatJPEG, imagerequest.FormatPNG, imagerequest.FormatGIF,
+12 -53
View File
@@ -32,23 +32,12 @@ type CacheConfig struct {
NegativeTTL time.Duration NegativeTTL time.Duration
// MaxBytes is the disk cache size limit in bytes that eviction // MaxBytes is the disk cache size limit in bytes that eviction
// enforces. Zero means no limit is enforced (no eviction). // enforces. Zero turns the disk cache off entirely: no cache
// directories are created, lookups always miss, stores are no-ops,
// and the eviction goroutine only removes the expired negative cache
// rows.
MaxBytes int64 MaxBytes int64
// UseDefaultMaxBytes makes NewCache replace MaxBytes with the
// default limit: 75% of the sum of the space free on the filesystem
// holding the cache and the bytes the cache already holds, at least
// DefaultCacheMaxBytesFloor. The config layer sets this when the
// operator omits cache_max_bytes.
UseDefaultMaxBytes bool
// DisableDiskCache turns the disk cache off entirely: no cache
// directories are created, lookups always miss, stores are
// no-ops, and the eviction goroutine only removes the expired
// negative cache rows. 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 receives accounting and eviction log output. A nil
// Logger means slog.Default(). // Logger means slog.Default().
Logger *slog.Logger Logger *slog.Logger
@@ -62,10 +51,6 @@ type Cache struct {
config CacheConfig config CacheConfig
log *slog.Logger 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 // Eviction machinery. The channels are created in NewCache so
// stores can signal write pressure without racing StartEviction. // stores can signal write pressure without racing StartEviction.
// evictionCancel, set by StartEviction, cancels the eviction // evictionCancel, set by StartEviction, cancels the eviction
@@ -97,7 +82,7 @@ type Cache struct {
transforms prometheus.Counter transforms prometheus.Counter
// reconciliationPageSize is the most rows one read of a content table // reconciliationPageSize is the most rows one read of a content table
// returns in the reconciliation pass. newCache sets it to // returns in the reconciliation pass. NewCache sets it to
// defaultReconciliationPageSize; tests set it smaller. // defaultReconciliationPageSize; tests set it smaller.
reconciliationPageSize int reconciliationPageSize int
@@ -110,20 +95,9 @@ type Cache struct {
// NewCache creates a new cache instance and registers its counters on // NewCache creates a new cache instance and registers its counters on
// metricsRegistry. pixad passes Prometheus' default registry, the one // metricsRegistry. pixad passes Prometheus' default registry, the one
// /metrics serves. // /metrics serves. It makes no database call.
func NewCache( func NewCache(
db *sql.DB, config CacheConfig, metricsRegistry prometheus.Registerer, db *sql.DB, config CacheConfig, metricsRegistry prometheus.Registerer,
) (*Cache, error) {
return newCache(db, config, metricsRegistry, defaultFreeSpaceProbe)
}
// newCache is NewCache with the free-space probe passed in, so tests
// can fake the free space the default limit is worked out from.
func newCache(
db *sql.DB,
config CacheConfig,
metricsRegistry prometheus.Registerer,
probe FreeSpaceProbeFunc,
) (*Cache, error) { ) (*Cache, error) {
log := config.Logger log := config.Logger
if log == nil { if log == nil {
@@ -134,7 +108,6 @@ func newCache(
db: db, db: db,
config: config, config: config,
log: log, log: log,
disabled: config.DisableDiskCache,
evictionPressure: make(chan struct{}, 1), evictionPressure: make(chan struct{}, 1),
evictionDone: make(chan struct{}), evictionDone: make(chan struct{}),
@@ -150,7 +123,7 @@ func newCache(
return nil, err return nil, err
} }
if c.disabled { if config.MaxBytes == 0 {
return c, nil return c, nil
} }
@@ -171,20 +144,6 @@ func newCache(
c.srcContent = srcContent c.srcContent = srcContent
c.variants = variants c.variants = variants
if config.UseDefaultMaxBytes {
limit, err := c.computeDefaultMaxBytes(context.Background(), probe)
if err != nil {
return nil, err
}
c.config.MaxBytes = limit
log.Info("computed default cache size limit from free space and cache contents",
"cache_max_bytes", limit,
"cache_dir", filepath.Join(config.StateDir, "cache"),
)
}
return c, nil return c, nil
} }
@@ -196,7 +155,7 @@ func newCache(
func (c *Cache) GetVariant( func (c *Cache) GetVariant(
cacheKey imagerequest.VariantKey, cacheKey imagerequest.VariantKey,
) (io.ReadCloser, int64, string, error) { ) (io.ReadCloser, int64, string, error) {
if c.disabled { if c.config.MaxBytes == 0 {
return nil, 0, "", ErrNotFound return nil, 0, "", ErrNotFound
} }
@@ -214,7 +173,7 @@ func (c *Cache) StoreSource(
req *imagerequest.Request, req *imagerequest.Request,
content io.Reader, content io.Reader,
) (imagerequest.SourceKey, error) { ) (imagerequest.SourceKey, error) {
if c.disabled { if c.config.MaxBytes == 0 {
return "", nil return "", nil
} }
@@ -259,7 +218,7 @@ func (c *Cache) StoreSource(
func (c *Cache) StoreVariant( func (c *Cache) StoreVariant(
ctx context.Context, cacheKey imagerequest.VariantKey, content io.Reader, ctx context.Context, cacheKey imagerequest.VariantKey, content io.Reader,
) error { ) error {
if c.disabled { if c.config.MaxBytes == 0 {
return nil return nil
} }
@@ -293,7 +252,7 @@ func (c *Cache) StoreVariant(
func (c *Cache) LookupSource( func (c *Cache) LookupSource(
ctx context.Context, req *imagerequest.Request, ctx context.Context, req *imagerequest.Request,
) (imagerequest.SourceKey, error) { ) (imagerequest.SourceKey, error) {
if c.disabled { if c.config.MaxBytes == 0 {
return "", nil return "", nil
} }
@@ -345,7 +304,7 @@ func (c *Cache) StoreNegative(ctx context.Context, req *imagerequest.Request) er
func (c *Cache) GetSourceContent( func (c *Cache) GetSourceContent(
key imagerequest.SourceKey, key imagerequest.SourceKey,
) (io.ReadCloser, int64, error) { ) (io.ReadCloser, int64, error) {
if c.disabled { if c.config.MaxBytes == 0 {
return nil, 0, ErrNotFound return nil, 0, ErrNotFound
} }
+4 -1
View File
@@ -22,7 +22,10 @@ func TestGetVariant_ContentTypeFromFirstBytes(t *testing.T) {
t.Parallel() t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30) cache, _ := newEvictionTestCache(t, 1<<30)
processor := imageprocessor.New(imageprocessor.Params{}) processor := imageprocessor.New(imageprocessor.Params{
MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 1,
})
source := generateTestJPEG(t, 40, 30, color.RGBA{200, 40, 40, 255}) source := generateTestJPEG(t, 40, 30, color.RGBA{200, 40, 40, 255})
for _, format := range []imagerequest.Format{ for _, format := range []imagerequest.Format{
+2
View File
@@ -53,6 +53,7 @@ func TestCache_CleanExpired(t *testing.T) {
cache, _ := NewCache(db, CacheConfig{ cache, _ := NewCache(db, CacheConfig{
StateDir: tmpDir, StateDir: tmpDir,
NegativeTTL: 1 * time.Millisecond, NegativeTTL: 1 * time.Millisecond,
MaxBytes: 1 << 30,
}, prometheus.NewRegistry()) }, prometheus.NewRegistry())
ctx := context.Background() ctx := context.Background()
@@ -104,6 +105,7 @@ func TestCache_StorageDirectoriesCreated(t *testing.T) {
_, err := NewCache(db, CacheConfig{ _, err := NewCache(db, CacheConfig{
StateDir: tmpDir, StateDir: tmpDir,
MaxBytes: 1 << 30,
}, prometheus.NewRegistry()) }, prometheus.NewRegistry())
if err != nil { if err != nil {
t.Fatalf("NewCache() error = %v", err) t.Fatalf("NewCache() error = %v", err)
-85
View File
@@ -1,85 +0,0 @@
package imgcache
import (
"context"
"errors"
"fmt"
"math"
"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
// freeSpaceFractionNumerator and freeSpaceFractionDenominator express
// the 75% share used for the computed default limit as integer
// arithmetic (dividing before multiplying avoids overflow).
const (
freeSpaceFractionNumerator uint64 = 3
freeSpaceFractionDenominator uint64 = 4
)
var errNegativeBlockSize = errors.New("statfs reported negative block size")
// 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
}
// computeDefaultMaxBytes returns the default cache size limit: 75% of
// the sum of the free bytes probe reports for <state_dir>/cache/ and
// the bytes the cache already holds, with a floor of
// DefaultCacheMaxBytesFloor. Counting what the cache holds keeps the
// limit from shrinking as the cache fills.
func (c *Cache) computeDefaultMaxBytes(
ctx context.Context, probe FreeSpaceProbeFunc,
) (int64, error) {
cacheDir := filepath.Join(c.config.StateDir, "cache")
freeBytes, err := probe(cacheDir)
if err != nil {
return 0, fmt.Errorf(
"default cache_max_bytes: cannot determine free space for %q: %w",
cacheDir, err)
}
usedBytes, err := c.UsageBytes(ctx)
if err != nil {
return 0, err
}
// Both terms are at most math.MaxInt64, so the sum cannot overflow.
//nolint:gosec // G115: UsageBytes returns the total cache usage, never negative
spaceBytes := min(freeBytes, math.MaxInt64) + uint64(usedBytes)
computed := spaceBytes / freeSpaceFractionDenominator * freeSpaceFractionNumerator
computed = min(computed, math.MaxInt64)
limit := int64(computed)
limit = max(limit, DefaultCacheMaxBytesFloor)
return limit, nil
}
@@ -1,189 +0,0 @@
package imgcache
import (
"errors"
"os"
"path/filepath"
"strings"
"testing"
"github.com/prometheus/client_golang/prometheus"
)
// Static errors returned by the stub free-space probes below.
var (
errTestStatfsFailed = errors.New("statfs failed")
errTestProbeNotExpected = errors.New("probe must not be called")
)
// TestComputeDefaultMaxBytesCountsWhatTheCacheHolds verifies that the
// default limit is 75% of the free space plus what the cache already
// holds, so a cache filled to its limit keeps that limit across a
// restart instead of shrinking to 75% of the space left free.
func TestComputeDefaultMaxBytesCountsWhatTheCacheHolds(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
// Empty cache, 4 GiB free -> 3 GiB default.
got, err := cache.computeDefaultMaxBytes(t.Context(),
func(string) (uint64, error) { return 4294967296, nil })
if err != nil {
t.Fatalf("computeDefaultMaxBytes returned error: %v", err)
}
t.Logf("default for an empty cache with 4 GiB free: %d", got)
if got != 3221225472 {
t.Errorf("default for an empty cache = %d, want 3221225472 (75%% of 4 GiB)",
got)
}
// The cache now holds those 3 GiB, which leaves 1 GiB free.
_, err = cache.db.ExecContext(t.Context(),
`INSERT INTO variant_content (cache_key, size_bytes) VALUES (?, ?)`,
string(testVariantKeyOne), 3221225472,
)
if err != nil {
t.Fatalf("failed to insert variant accounting row: %v", err)
}
got, err = cache.computeDefaultMaxBytes(t.Context(),
func(string) (uint64, error) { return 1073741824, nil })
if err != nil {
t.Fatalf("computeDefaultMaxBytes returned error: %v", err)
}
t.Logf("default for a cache holding 3 GiB with 1 GiB free: %d", got)
if got != 3221225472 {
t.Errorf("default for a cache holding 3 GiB with 1 GiB free = %d, "+
"want 3221225472 (75%% of 1 GiB + 3 GiB)", got)
}
}
// TestComputeDefaultMaxBytesAppliesFloor verifies that when 75% of the
// free space plus what the cache holds is below 500 MiB, the default
// is floored at DefaultCacheMaxBytesFloor.
func TestComputeDefaultMaxBytesAppliesFloor(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()
cache, _ := newEvictionTestCache(t, 1<<30)
got, err := cache.computeDefaultMaxBytes(t.Context(),
func(string) (uint64, error) { return tc.freeBytes, nil })
if err != nil {
t.Fatalf("computeDefaultMaxBytes returned error: %v", err)
}
if got != DefaultCacheMaxBytesFloor {
t.Errorf("computeDefaultMaxBytes = %d, want floor %d",
got, DefaultCacheMaxBytesFloor)
}
})
}
}
// TestComputeDefaultMaxBytesPropagatesProbeError verifies that a
// failing free-space probe produces an error naming cache_max_bytes,
// instead of a silently wrong default.
func TestComputeDefaultMaxBytesPropagatesProbeError(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
_, err := cache.computeDefaultMaxBytes(t.Context(),
func(string) (uint64, error) { return 0, errTestStatfsFailed })
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(), "cache_max_bytes") {
t.Errorf("error %q does not name cache_max_bytes", err.Error())
}
}
// TestNewCacheComputesDefaultMaxBytesWhenAsked verifies that with
// UseDefaultMaxBytes set, the cache's limit becomes the computed
// default, and that the probe is pointed at <state_dir>/cache/, which
// must be created first so statfs measures the right filesystem.
func TestNewCacheComputesDefaultMaxBytesWhenAsked(t *testing.T) {
t.Parallel()
stateDir := t.TempDir()
wantCacheDir := filepath.Join(stateDir, "cache")
var probedPath string
// 4 GiB free -> 3 GiB default.
probe := func(path string) (uint64, error) {
probedPath = path
info, err := os.Stat(path)
if err != nil || !info.IsDir() {
t.Errorf("cache directory %q was not created before probing: info=%v err=%v",
path, info, err)
}
return 4294967296, nil
}
cache, err := newCache(evictionTestDB(t), CacheConfig{
StateDir: stateDir,
UseDefaultMaxBytes: true,
}, prometheus.NewRegistry(), probe)
if err != nil {
t.Fatalf("newCache returned error: %v", err)
}
if cache.config.MaxBytes != 3221225472 {
t.Errorf("MaxBytes = %d, want computed default 3221225472",
cache.config.MaxBytes)
}
if probedPath != wantCacheDir {
t.Errorf("free space probed at %q, want cache directory %q",
probedPath, wantCacheDir)
}
}
// TestNewCacheKeepsExplicitMaxBytes verifies that without
// UseDefaultMaxBytes the cache keeps MaxBytes exactly as given and
// never consults the free-space probe.
func TestNewCacheKeepsExplicitMaxBytes(t *testing.T) {
t.Parallel()
probe := func(string) (uint64, error) {
t.Error("free-space probe must not be consulted for explicit values")
return 0, errTestProbeNotExpected
}
cache, err := newCache(evictionTestDB(t), CacheConfig{
StateDir: t.TempDir(),
MaxBytes: 1024,
}, prometheus.NewRegistry(), probe)
if err != nil {
t.Fatalf("newCache returned error: %v", err)
}
if cache.config.MaxBytes != 1024 {
t.Errorf("MaxBytes = %d, want explicit 1024 (no floor, no recompute)",
cache.config.MaxBytes)
}
}
@@ -455,7 +455,7 @@ func TestService_Get_WaiterGetsProcessingTimedOut(t *testing.T) {
svc, fixtures, fetcher := setupHeldFetchService(t) svc, fixtures, fetcher := setupHeldFetchService(t)
svc.processor = imageprocessor.New( svc.processor = imageprocessor.New(
imageprocessor.Params{MaxConcurrentProcessing: 1}, imageprocessor.Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1},
) )
holdProcessingSlot(t, svc.processor) holdProcessingSlot(t, svc.processor)
+6 -3
View File
@@ -36,7 +36,8 @@ func TestNewCache_RegistersItsCountersOnItsRegistry(t *testing.T) {
registry := prometheus.NewRegistry() registry := prometheus.NewRegistry()
_, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()}, registry) _, err := NewCache(evictionTestDB(t),
CacheConfig{StateDir: t.TempDir(), MaxBytes: 1 << 30}, registry)
if err != nil { if err != nil {
t.Fatalf("NewCache() error = %v", err) t.Fatalf("NewCache() error = %v", err)
} }
@@ -68,12 +69,14 @@ func TestNewCache_FailsWhenItsCountersCannotBeRegistered(t *testing.T) {
registry := prometheus.NewRegistry() registry := prometheus.NewRegistry()
_, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()}, registry) _, err := NewCache(evictionTestDB(t),
CacheConfig{StateDir: t.TempDir(), MaxBytes: 1 << 30}, registry)
if err != nil { if err != nil {
t.Fatalf("NewCache() of the first cache error = %v", err) t.Fatalf("NewCache() of the first cache error = %v", err)
} }
_, err = NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()}, registry) _, err = NewCache(evictionTestDB(t),
CacheConfig{StateDir: t.TempDir(), MaxBytes: 1 << 30}, registry)
t.Logf("NewCache() of the second cache error = %v", err) t.Logf("NewCache() of the second cache error = %v", err)
var alreadyRegistered prometheus.AlreadyRegisteredError var alreadyRegistered prometheus.AlreadyRegisteredError
+17 -10
View File
@@ -52,18 +52,25 @@ const sqliteTimestampLayout = "2006-01-02 15:04:05"
const tempFilePrefix = ".tmp-" const tempFilePrefix = ".tmp-"
// UsageBytes returns the total number of bytes of cache content // UsageBytes returns the total number of bytes of cache content
// tracked in the database (source content blobs plus processed // tracked in the database, as the package function UsageBytes does, or 0
// variants). It reads the total the database keeps up to date as rows // when the disk cache is off.
// are added and removed, so it neither scans the cache directories nor
// sums the tables.
func (c *Cache) UsageBytes(ctx context.Context) (int64, error) { func (c *Cache) UsageBytes(ctx context.Context) (int64, error) {
if c.disabled { if c.config.MaxBytes == 0 {
return 0, nil return 0, nil
} }
return UsageBytes(ctx, c.db)
}
// UsageBytes returns the total number of bytes of cache content tracked
// in db (source content blobs plus processed variants). It reads the
// total the database keeps up to date as rows are added and removed, so
// it neither scans the cache directories nor sums the tables. pixad reads
// it before it builds the cache, to work out the default cache_max_bytes.
func UsageBytes(ctx context.Context, db *sql.DB) (int64, error) {
var total int64 var total int64
err := c.db.QueryRowContext(ctx, err := db.QueryRowContext(ctx,
`SELECT total_size_bytes FROM cache_usage WHERE id = 1`, `SELECT total_size_bytes FROM cache_usage WHERE id = 1`,
).Scan(&total) ).Scan(&total)
if err != nil { if err != nil {
@@ -86,9 +93,9 @@ type evictionCandidate struct {
// EvictToLimit evicts least-recently-used cache entries until total // EvictToLimit evicts least-recently-used cache entries until total
// tracked usage is at or below the configured MaxBytes limit. It is a // 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. // no-op when the cache is disabled.
func (c *Cache) EvictToLimit(ctx context.Context) error { func (c *Cache) EvictToLimit(ctx context.Context) error {
if c.disabled || c.config.MaxBytes <= 0 { if c.config.MaxBytes == 0 {
return nil return nil
} }
@@ -326,7 +333,7 @@ func (c *Cache) evictSourceBlob(ctx context.Context, key imagerequest.SourceKey)
// storing request. The notification channel has capacity one and drops // storing request. The notification channel has capacity one and drops
// when a wakeup is already pending. // when a wakeup is already pending.
func (c *Cache) notifyWritePressure() { func (c *Cache) notifyWritePressure() {
if c.disabled || c.config.MaxBytes <= 0 { if c.config.MaxBytes == 0 {
return return
} }
@@ -485,7 +492,7 @@ func (c *Cache) runCleanExpired(ctx context.Context) {
// Once ctx is cancelled, it stops at the next file or row and returns // Once ctx is cancelled, it stops at the next file or row and returns
// ctx's error. // ctx's error.
func (c *Cache) reconcileAccounting(ctx context.Context) error { func (c *Cache) reconcileAccounting(ctx context.Context) error {
if c.disabled { if c.config.MaxBytes == 0 {
return nil return nil
} }
+4 -8
View File
@@ -66,15 +66,11 @@ func newEvictionTestCache(t *testing.T, maxBytes int64) (*Cache, string) {
tmpDir := t.TempDir() tmpDir := t.TempDir()
db := evictionTestDB(t) db := evictionTestDB(t)
// maxBytes zero mirrors the production mapping of // maxBytes zero turns the disk cache off, as cache_max_bytes: 0 does.
// 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{ cache, err := NewCache(db, CacheConfig{
StateDir: tmpDir, StateDir: tmpDir,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
MaxBytes: maxBytes, MaxBytes: maxBytes,
DisableDiskCache: maxBytes == 0,
}, prometheus.NewRegistry()) }, prometheus.NewRegistry())
if err != nil { if err != nil {
t.Fatalf("failed to create cache: %v", err) t.Fatalf("failed to create cache: %v", err)
@@ -66,7 +66,7 @@ func TestService_Get_WaitsForSlotBeforeReadingCachedSource(t *testing.T) {
svc, fixtures := SetupTestService(t) svc, fixtures := SetupTestService(t)
svc.processor = imageprocessor.New( svc.processor = imageprocessor.New(
imageprocessor.Params{MaxConcurrentProcessing: 1}, imageprocessor.Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1},
) )
// A first request caches the photo as a source. // A first request caches the photo as a source.
@@ -20,6 +20,7 @@ func TestNegativeCache_StoreAndCheck(t *testing.T) {
cache, err := NewCache(db, CacheConfig{ cache, err := NewCache(db, CacheConfig{
StateDir: dir, StateDir: dir,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
MaxBytes: 1 << 30,
}, prometheus.NewRegistry()) }, prometheus.NewRegistry())
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
@@ -69,6 +70,7 @@ func TestNegativeCache_Expired(t *testing.T) {
cache, err := NewCache(db, CacheConfig{ cache, err := NewCache(db, CacheConfig{
StateDir: dir, StateDir: dir,
NegativeTTL: 1 * time.Millisecond, // very short TTL NegativeTTL: 1 * time.Millisecond, // very short TTL
MaxBytes: 1 << 30,
}, prometheus.NewRegistry()) }, prometheus.NewRegistry())
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
@@ -177,11 +179,11 @@ func TestEvictionLoopRemovesExpiredNegativeCacheRows(t *testing.T) {
t.Parallel() t.Parallel()
tests := []struct { tests := []struct {
name string name string
disableDiskCache bool maxBytes int64
}{ }{
{name: "disk cache enabled", disableDiskCache: false}, {name: "disk cache enabled", maxBytes: 1 << 30},
{name: "disk cache disabled", disableDiskCache: true}, {name: "disk cache disabled", maxBytes: 0},
} }
for _, tt := range tests { for _, tt := range tests {
@@ -189,9 +191,9 @@ func TestEvictionLoopRemovesExpiredNegativeCacheRows(t *testing.T) {
t.Parallel() t.Parallel()
cache, err := NewCache(evictionTestDB(t), CacheConfig{ cache, err := NewCache(evictionTestDB(t), CacheConfig{
StateDir: t.TempDir(), StateDir: t.TempDir(),
NegativeTTL: time.Second, NegativeTTL: time.Second,
DisableDiskCache: tt.disableDiskCache, MaxBytes: tt.maxBytes,
}, prometheus.NewRegistry()) }, prometheus.NewRegistry())
if err != nil { if err != nil {
t.Fatalf("failed to create cache: %v", err) t.Fatalf("failed to create cache: %v", err)
@@ -266,7 +268,8 @@ func negativeCacheRowCount(tb testing.TB, cache *Cache, sourcePath string) int {
func TestCleanExpiredDeletesInBatches(t *testing.T) { func TestCleanExpiredDeletesInBatches(t *testing.T) {
t.Parallel() t.Parallel()
cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()}, cache, err := NewCache(evictionTestDB(t),
CacheConfig{StateDir: t.TempDir(), MaxBytes: 1 << 30},
prometheus.NewRegistry()) prometheus.NewRegistry())
if err != nil { if err != nil {
t.Fatalf("failed to create cache: %v", err) t.Fatalf("failed to create cache: %v", err)
@@ -321,7 +324,8 @@ func TestCleanExpiredDeletesInBatches(t *testing.T) {
func TestCleanExpiredRemovesMoreThanOneBatch(t *testing.T) { func TestCleanExpiredRemovesMoreThanOneBatch(t *testing.T) {
t.Parallel() t.Parallel()
cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()}, cache, err := NewCache(evictionTestDB(t),
CacheConfig{StateDir: t.TempDir(), MaxBytes: 1 << 30},
prometheus.NewRegistry()) prometheus.NewRegistry())
if err != nil { if err != nil {
t.Fatalf("failed to create cache: %v", err) t.Fatalf("failed to create cache: %v", err)
@@ -352,7 +356,8 @@ func TestCleanExpiredRemovesMoreThanOneBatch(t *testing.T) {
func TestCleanExpiredStopsOnceContextEnds(t *testing.T) { func TestCleanExpiredStopsOnceContextEnds(t *testing.T) {
t.Parallel() t.Parallel()
cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()}, cache, err := NewCache(evictionTestDB(t),
CacheConfig{StateDir: t.TempDir(), MaxBytes: 1 << 30},
prometheus.NewRegistry()) prometheus.NewRegistry())
if err != nil { if err != nil {
t.Fatalf("failed to create cache: %v", err) t.Fatalf("failed to create cache: %v", err)
+14 -39
View File
@@ -43,18 +43,20 @@ type Service struct {
type ServiceConfig struct { type ServiceConfig struct {
// Cache is the cache instance // Cache is the cache instance
Cache *Cache Cache *Cache
// FetcherConfig configures the upstream fetcher built when Fetcher is // Fetcher fetches source images from upstream hosts
// not set. Its AllowHTTP and MaxResponseSize are used either way.
FetcherConfig *httpfetcher.Config
// Fetcher is an optional custom fetcher (for testing)
Fetcher httpfetcher.Fetcher Fetcher httpfetcher.Fetcher
// Processor converts and resizes the images
Processor *imageprocessor.ImageProcessor
// SigningKey is the HMAC signing key (empty disables signing) // SigningKey is the HMAC signing key (empty disables signing)
SigningKey string SigningKey string
// Allowlist is the list of hosts that don't require signatures // Allowlist is the list of hosts that don't require signatures
Allowlist []string Allowlist []string
// MaxConcurrentProcessing is the most images processed at once; zero // AllowHTTP makes the service fetch sources over http instead of
// uses the image processor's default, half the CPUs // https (testing only), so the fetcher must allow http too
MaxConcurrentProcessing int AllowHTTP bool
// MaxResponseSize is the largest source accepted, in bytes; a cached
// source larger than this is fetched again
MaxResponseSize int64
// Logger for logging // Logger for logging
Logger *slog.Logger Logger *slog.Logger
} }
@@ -75,32 +77,11 @@ func NewService(cfg *ServiceConfig) (*Service, error) {
return nil, errSigningKeyRequired return nil, errSigningKeyRequired
} }
// Resolve fetcher config for defaults
fetcherCfg := cfg.FetcherConfig
if fetcherCfg == nil {
fetcherCfg = httpfetcher.DefaultConfig()
}
// Use custom fetcher if provided, otherwise create HTTP fetcher
var fetcher httpfetcher.Fetcher
if cfg.Fetcher != nil {
fetcher = cfg.Fetcher
} else {
fetcher = httpfetcher.New(fetcherCfg)
}
signer := signature.New(cfg.SigningKey)
log := cfg.Logger log := cfg.Logger
if log == nil { if log == nil {
log = slog.Default() log = slog.Default()
} }
allowHTTP := false
if cfg.FetcherConfig != nil {
allowHTTP = cfg.FetcherConfig.AllowHTTP
}
// JPEG XL is the default output format, so pixad does not start // JPEG XL is the default output format, so pixad does not start
// without it. // without it.
err := imageprocessor.CheckJPEGXLSupport() err := imageprocessor.CheckJPEGXLSupport()
@@ -108,21 +89,15 @@ func NewService(cfg *ServiceConfig) (*Service, error) {
return nil, err return nil, err
} }
maxResponseSize := fetcherCfg.MaxResponseSize
processor := imageprocessor.New(imageprocessor.Params{
MaxInputBytes: maxResponseSize,
MaxConcurrentProcessing: cfg.MaxConcurrentProcessing,
})
return &Service{ return &Service{
cache: cfg.Cache, cache: cfg.Cache,
fetcher: fetcher, fetcher: cfg.Fetcher,
processor: processor, processor: cfg.Processor,
signer: signer, signer: signature.New(cfg.SigningKey),
allowlist: allowlist.New(cfg.Allowlist), allowlist: allowlist.New(cfg.Allowlist),
log: log, log: log,
allowHTTP: allowHTTP, allowHTTP: cfg.AllowHTTP,
maxResponseSize: maxResponseSize, maxResponseSize: cfg.MaxResponseSize,
}, nil }, nil
} }
+11 -4
View File
@@ -17,6 +17,7 @@ import (
"github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus"
"sneak.berlin/go/pixa/internal/database" "sneak.berlin/go/pixa/internal/database"
"sneak.berlin/go/pixa/internal/httpfetcher" "sneak.berlin/go/pixa/internal/httpfetcher"
"sneak.berlin/go/pixa/internal/imageprocessor"
) )
// Shared test data literals, extracted as constants for goconst. // Shared test data literals, extracted as constants for goconst.
@@ -185,16 +186,22 @@ func SetupTestService(
cache, err := NewCache(db, CacheConfig{ cache, err := NewCache(db, CacheConfig{
StateDir: tmpDir, StateDir: tmpDir,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
MaxBytes: 1 << 30,
}, prometheus.NewRegistry()) }, prometheus.NewRegistry())
if err != nil { if err != nil {
t.Fatalf("failed to create cache: %v", err) t.Fatalf("failed to create cache: %v", err)
} }
svc, err := NewService(&ServiceConfig{ svc, err := NewService(&ServiceConfig{
Cache: cache, Cache: cache,
Fetcher: httpfetcher.NewMock(mockFS), Fetcher: httpfetcher.NewMock(mockFS),
SigningKey: cfg.signingKey, Processor: imageprocessor.New(imageprocessor.Params{
Allowlist: cfg.allowlist, MaxInputBytes: 50 << 20,
MaxConcurrentProcessing: 2,
}),
SigningKey: cfg.signingKey,
Allowlist: cfg.allowlist,
MaxResponseSize: 50 << 20,
}) })
if err != nil { if err != nil {
t.Fatalf("failed to create service: %v", err) t.Fatalf("failed to create service: %v", err)
@@ -0,0 +1,315 @@
package providers
import (
"context"
"database/sql"
"errors"
"log/slog"
"os"
"path/filepath"
"strings"
"testing"
"github.com/prometheus/client_golang/prometheus"
"go.uber.org/fx/fxtest"
"sneak.berlin/go/pixa/internal/config"
"sneak.berlin/go/pixa/internal/database"
"sneak.berlin/go/pixa/internal/globals"
"sneak.berlin/go/pixa/internal/logger"
)
// testSigningKey is a throwaway signing key.
const testSigningKey = "test-signing-key-0123456789abcdef"
// Static errors returned by the stub free-space probes below.
var (
errTestStatfsFailed = errors.New("statfs failed")
errTestProbeNotExpected = errors.New("probe must not be called")
)
// newTestDB opens an in-memory database with pixa's schema, on one
// connection, as pixad's database has.
func newTestDB(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
}
// TestCacheMaxBytesFromConfig checks the disk cache limit worked out from
// cache_max_bytes: omitted, it is the default from the free space; 0 turns
// the disk cache off; a positive value is the limit, unchanged.
func TestCacheMaxBytesFromConfig(t *testing.T) {
t.Parallel()
const oneGiB = 1 << 30
cases := []struct {
name string
cacheMaxBytes int64
cacheMaxBytesExplicit bool
wantMaxBytes int64
}{
{
name: "cache_max_bytes omitted",
cacheMaxBytes: 0,
cacheMaxBytesExplicit: false,
// 75% of the 4 GiB the probe reports free
wantMaxBytes: 3221225472,
},
{
name: "cache_max_bytes: 0",
cacheMaxBytes: 0,
cacheMaxBytesExplicit: true,
wantMaxBytes: 0,
},
{
name: "cache_max_bytes: 1 GiB",
cacheMaxBytes: oneGiB,
cacheMaxBytesExplicit: true,
wantMaxBytes: oneGiB,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
cfg := &config.Config{
StateDir: t.TempDir(),
CacheMaxBytes: tc.cacheMaxBytes,
CacheMaxBytesExplicit: tc.cacheMaxBytesExplicit,
}
got, err := cacheMaxBytes(t.Context(), cfg, newTestDB(t),
slog.New(slog.DiscardHandler),
func(string) (uint64, error) { return 4294967296, nil })
if err != nil {
t.Fatalf("cacheMaxBytes() error = %v", err)
}
t.Logf("MaxBytes = %d", got)
if got != tc.wantMaxBytes {
t.Errorf("MaxBytes = %d, want %d", got, tc.wantMaxBytes)
}
})
}
}
// TestDiskCacheOffOnlyForExplicitZeroCacheMaxBytes builds and starts the
// cache once with cache_max_bytes omitted and once with cache_max_bytes:
// 0, and checks by whether the cache directories were created that the
// disk cache is on in the first case and off in the second.
func TestDiskCacheOffOnlyForExplicitZeroCacheMaxBytes(t *testing.T) {
t.Parallel()
cases := []struct {
name string
cacheMaxBytesExplicit bool
wantDiskCache bool
}{
{name: "cache_max_bytes omitted", cacheMaxBytesExplicit: false, wantDiskCache: true},
{name: "cache_max_bytes: 0", cacheMaxBytesExplicit: true, wantDiskCache: false},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
stateDir := t.TempDir()
cfg := &config.Config{
SigningKey: testSigningKey,
StateDir: stateDir,
DBURL: "file:" + filepath.Join(stateDir, "state.sqlite3"),
CacheMaxBytes: 0,
CacheMaxBytesExplicit: tc.cacheMaxBytesExplicit,
}
lc := fxtest.NewLifecycle(t)
log, err := logger.New(lc, logger.Params{Globals: &globals.Globals{}})
if err != nil {
t.Fatalf("logger.New() error = %v", err)
}
db, err := database.New(lc, database.Params{Logger: log, Config: cfg})
if err != nil {
t.Fatalf("database.New() error = %v", err)
}
_, err = newCache(lc, cacheParams{
Config: cfg,
Database: db,
Logger: log,
MetricsRegistry: prometheus.NewRegistry(),
})
if err != nil {
t.Fatalf("newCache() error = %v", err)
}
lc.RequireStart()
t.Cleanup(lc.RequireStop)
_, err = os.Stat(filepath.Join(stateDir, "cache", "variants"))
gotDiskCache := err == nil
if gotDiskCache != tc.wantDiskCache {
t.Errorf("cache directories created = %v, want %v",
gotDiskCache, tc.wantDiskCache)
}
})
}
}
// TestDefaultCacheMaxBytesCountsWhatTheCacheHolds verifies that the
// default limit is 75% of the free space plus what the cache already
// holds, so a cache filled to its limit keeps that limit across a
// restart instead of shrinking to 75% of the space left free.
func TestDefaultCacheMaxBytesCountsWhatTheCacheHolds(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db := newTestDB(t)
log := slog.New(slog.DiscardHandler)
// Empty cache, 4 GiB free -> 3 GiB default.
got, err := cacheMaxBytes(t.Context(), cfg, db, log,
func(string) (uint64, error) { return 4294967296, nil })
if err != nil {
t.Fatalf("cacheMaxBytes returned error: %v", err)
}
t.Logf("default for an empty cache with 4 GiB free: %d", got)
if got != 3221225472 {
t.Errorf("default for an empty cache = %d, want 3221225472 (75%% of 4 GiB)",
got)
}
// The cache now holds those 3 GiB, which leaves 1 GiB free.
_, err = db.ExecContext(t.Context(),
`INSERT INTO variant_content (cache_key, size_bytes) VALUES (?, ?)`,
"aabbccdd0001", 3221225472,
)
if err != nil {
t.Fatalf("failed to insert variant accounting row: %v", err)
}
got, err = cacheMaxBytes(t.Context(), cfg, db, log,
func(string) (uint64, error) { return 1073741824, nil })
if err != nil {
t.Fatalf("cacheMaxBytes returned error: %v", err)
}
t.Logf("default for a cache holding 3 GiB with 1 GiB free: %d", got)
if got != 3221225472 {
t.Errorf("default for a cache holding 3 GiB with 1 GiB free = %d, "+
"want 3221225472 (75%% of 1 GiB + 3 GiB)", got)
}
}
// TestDefaultCacheMaxBytesPropagatesProbeError verifies that a failing
// free-space probe produces an error naming cache_max_bytes, instead of
// a silently wrong default.
func TestDefaultCacheMaxBytesPropagatesProbeError(t *testing.T) {
t.Parallel()
_, err := cacheMaxBytes(t.Context(), &config.Config{StateDir: t.TempDir()},
newTestDB(t), slog.New(slog.DiscardHandler),
func(string) (uint64, error) { return 0, errTestStatfsFailed })
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(), "cache_max_bytes") {
t.Errorf("error %q does not name cache_max_bytes", err.Error())
}
}
// TestDefaultCacheMaxBytesProbesTheCacheDirectory verifies that with
// cache_max_bytes omitted the limit is the computed default, and that the
// probe is pointed at <state_dir>/cache/, which must be created first so
// statfs measures the right filesystem.
func TestDefaultCacheMaxBytesProbesTheCacheDirectory(t *testing.T) {
t.Parallel()
stateDir := t.TempDir()
wantCacheDir := filepath.Join(stateDir, "cache")
var probedPath string
// 4 GiB free -> 3 GiB default.
probe := func(path string) (uint64, error) {
probedPath = path
info, err := os.Stat(path)
if err != nil || !info.IsDir() {
t.Errorf("cache directory %q was not created before probing: info=%v err=%v",
path, info, err)
}
return 4294967296, nil
}
got, err := cacheMaxBytes(t.Context(), &config.Config{StateDir: stateDir},
newTestDB(t), slog.New(slog.DiscardHandler), probe)
if err != nil {
t.Fatalf("cacheMaxBytes returned error: %v", err)
}
if got != 3221225472 {
t.Errorf("MaxBytes = %d, want computed default 3221225472", got)
}
if probedPath != wantCacheDir {
t.Errorf("free space probed at %q, want cache directory %q",
probedPath, wantCacheDir)
}
}
// TestExplicitCacheMaxBytesIsKept verifies that an explicit
// cache_max_bytes is kept exactly as given and the free-space probe is
// never consulted.
func TestExplicitCacheMaxBytesIsKept(t *testing.T) {
t.Parallel()
probe := func(string) (uint64, error) {
t.Error("free-space probe must not be consulted for explicit values")
return 0, errTestProbeNotExpected
}
cfg := &config.Config{
StateDir: t.TempDir(),
CacheMaxBytes: 1024,
CacheMaxBytesExplicit: true,
}
got, err := cacheMaxBytes(t.Context(), cfg, newTestDB(t),
slog.New(slog.DiscardHandler), probe)
if err != nil {
t.Fatalf("cacheMaxBytes returned error: %v", err)
}
if got != 1024 {
t.Errorf("MaxBytes = %d, want explicit 1024 (no floor, no recompute)", got)
}
}
+103
View File
@@ -0,0 +1,103 @@
package providers
import (
"encoding/json"
"net/http"
"net/http/httptest"
"net/netip"
"path/filepath"
"testing"
"time"
"github.com/go-chi/chi/v5"
"github.com/prometheus/client_golang/prometheus"
"go.uber.org/fx"
"go.uber.org/fx/fxtest"
"sneak.berlin/go/pixa/internal/config"
"sneak.berlin/go/pixa/internal/database"
"sneak.berlin/go/pixa/internal/globals"
"sneak.berlin/go/pixa/internal/handlers"
"sneak.berlin/go/pixa/internal/healthcheck"
"sneak.berlin/go/pixa/internal/logger"
)
// TestFetcherFromConfigRefusesBlockedNetworks builds the handlers as pixad
// does, in an fx app with the providers of pixad's fx app, and requests an
// image from 192.0.2.10, which is on the allowlist and in blocked_networks.
// The URL check accepts that address; only the dialer that refuses internal
// addresses checks blocked_networks, so the answer is 403 only if the
// fetcher this package builds from the config connects with that dialer.
// Any other dialer would try to connect until the upstream fetch timeout,
// which is short so that the test then fails quickly.
func TestFetcherFromConfigRefusesBlockedNetworks(t *testing.T) {
t.Parallel()
const host = "192.0.2.10"
stateDir := t.TempDir()
cfg := &config.Config{
SigningKey: testSigningKey,
StateDir: stateDir,
DBURL: "file:" + filepath.Join(stateDir, "state.sqlite3"),
AllowlistHosts: []string{host},
BlockedNetworks: []netip.Prefix{netip.MustParsePrefix("192.0.2.0/24")},
UpstreamFetchTimeout: 2 * time.Second,
// With no connection slots, the fetch would fail before dialing.
UpstreamConnectionsPerHost: config.DefaultUpstreamConnectionsPerHost,
UpstreamConnections: config.DefaultUpstreamConnections,
UpstreamMaxResponseSize: config.DefaultUpstreamMaxResponseSize,
MaxConcurrentProcessing: 1,
}
var h *handlers.Handlers
app := fxtest.New(t,
fx.Supply(cfg),
fx.Provide(globals.New, logger.New, database.New, healthcheck.New,
handlers.New,
func() prometheus.Registerer { return prometheus.NewRegistry() }),
fx.Provide(Constructors()...),
fx.Populate(&h),
)
app.RequireStart()
t.Cleanup(app.RequireStop)
r := chi.NewRouter()
r.Get("/v1/image/*", h.HandleImage())
target := "/v1/image/" + host + "/images/photo.jpg/50x50.jpeg"
rec := httptest.NewRecorder()
r.ServeHTTP(rec, httptest.NewRequestWithContext(
t.Context(), http.MethodGet, target, nil))
t.Logf("GET %s: %d %s", target, rec.Code, rec.Body)
if rec.Code != http.StatusForbidden {
t.Errorf("status = %d, want %d", rec.Code, http.StatusForbidden)
}
if ct := rec.Header().Get("Content-Type"); ct != "application/json" {
t.Errorf("Content-Type = %q, want application/json", ct)
}
var body struct {
Error string `json:"error"`
Status int `json:"status"`
Timestamp string `json:"timestamp"`
}
err := json.NewDecoder(rec.Body).Decode(&body)
if err != nil {
t.Fatalf("decoding response body: %v", err)
}
if body.Error != "forbidden" || body.Status != http.StatusForbidden {
t.Errorf("body error = %q, status = %d; want %q, %d",
body.Error, body.Status, "forbidden", http.StatusForbidden)
}
_, err = time.Parse(time.RFC3339, body.Timestamp)
if err != nil {
t.Errorf("body timestamp %q is not RFC 3339: %v", body.Timestamp, err)
}
}
+256
View File
@@ -0,0 +1,256 @@
// Package providers builds from the config the parts pixad serves images
// and logins with, and provides them to the fx graph:
// - the image cache, with its default limit
// - the fetcher
// - the processor
// - the image service
// - the session manager
// - the encrypted URL generator
package providers
import (
"context"
"database/sql"
"errors"
"fmt"
"log/slog"
"os"
"path/filepath"
"syscall"
"github.com/prometheus/client_golang/prometheus"
"go.uber.org/fx"
"sneak.berlin/go/pixa/internal/config"
"sneak.berlin/go/pixa/internal/database"
"sneak.berlin/go/pixa/internal/encurl"
"sneak.berlin/go/pixa/internal/httpfetcher"
"sneak.berlin/go/pixa/internal/imageprocessor"
"sneak.berlin/go/pixa/internal/imgcache"
"sneak.berlin/go/pixa/internal/logger"
"sneak.berlin/go/pixa/internal/session"
)
// Constructors returns the constructors of the parts this package builds,
// for fx.Provide. The fx graph must also provide the prometheus.Registerer
// the cache registers its counters on.
func Constructors() []any {
return []any{
newCache,
newFetcherConfig,
httpfetcher.New,
newProcessor,
newImageService,
newSessionManager,
newEncryptedURLGenerator,
}
}
// cacheParams defines dependencies for newCache.
type cacheParams struct {
fx.In
Config *config.Config
Database *database.Database
Logger *logger.Logger
MetricsRegistry prometheus.Registerer
}
// newCache builds the image cache with the disk cache limit cacheMaxBytes
// returns. Its start hook starts the cache's background goroutines. Its
// stop hook stops them, which writes the pending access times. The
// database is open for both: it opens before the cache is built and
// closes after this stop hook.
func newCache(lc fx.Lifecycle, params cacheParams) (*imgcache.Cache, error) {
cfg := params.Config
db := params.Database.DB()
maxBytes, err := cacheMaxBytes(
context.Background(), cfg, db, params.Logger.Get(), defaultFreeSpaceProbe)
if err != nil {
return nil, err
}
cache, err := imgcache.NewCache(db, imgcache.CacheConfig{
StateDir: cfg.StateDir,
NegativeTTL: imgcache.DefaultNegativeTTL,
MaxBytes: maxBytes,
Logger: params.Logger.Get(),
}, params.MetricsRegistry)
if err != nil {
return nil, err
}
// The start hook takes no context: the goroutines it starts outlive it,
// and the stop hook stops them.
lc.Append(fx.StartStopHook(
func() {
// Background eviction: startup reconciliation, then periodic
// and write-pressure passes, and the removal of expired
// negative cache rows, which is all it does when the disk
// cache is off.
cache.StartEviction(imgcache.DefaultEvictionInterval)
// Writes the access times that requests hand over, so that no
// request waits for the database to record them
cache.StartPendingAccessTimeWrites()
},
func(ctx context.Context) error {
return errors.Join(
cache.StopEviction(ctx),
cache.StopPendingAccessTimeWrites(ctx),
)
},
))
return cache, nil
}
// freeSpaceProbeFunc reports the number of free bytes available on the
// filesystem containing path. It is a function type so tests can pass a
// fake probe instead of depending on the host disk.
type freeSpaceProbeFunc func(path string) (uint64, error)
var errNegativeBlockSize = errors.New("statfs reported negative block size")
// 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
}
// cacheMaxBytes returns cache_max_bytes as set. When the key is omitted,
// it returns config.DefaultCacheMaxBytes of the free bytes probe reports
// for <state_dir>/cache/ and the bytes the cache holds, as db records
// them. It creates that directory first, so the probe measures the
// filesystem that holds the cache.
func cacheMaxBytes(
ctx context.Context,
cfg *config.Config,
db *sql.DB,
log *slog.Logger,
probe freeSpaceProbeFunc,
) (int64, error) {
if cfg.CacheMaxBytesExplicit {
return cfg.CacheMaxBytes, nil
}
cacheDir := filepath.Join(cfg.StateDir, "cache")
err := os.MkdirAll(cacheDir, imgcache.StorageDirPerm)
if err != nil {
return 0, fmt.Errorf("failed to create cache directory: %w", err)
}
freeBytes, err := probe(cacheDir)
if err != nil {
return 0, fmt.Errorf(
"default cache_max_bytes: cannot determine free space for %q: %w",
cacheDir, err)
}
heldBytes, err := imgcache.UsageBytes(ctx, db)
if err != nil {
return 0, err
}
limit := config.DefaultCacheMaxBytes(freeBytes, heldBytes)
log.Info("computed default cache size limit from free space and cache contents",
"cache_max_bytes", limit,
"cache_dir", cacheDir,
)
return limit, nil
}
// newFetcherConfig builds the upstream fetcher's settings from cfg, and
// httpfetcher.New builds the fetcher from them. The settings are a value
// of their own in the fx graph, so that a test can replace the dial
// function alone.
func newFetcherConfig(cfg *config.Config) *httpfetcher.Config {
return &httpfetcher.Config{
Timeout: cfg.UpstreamFetchTimeout,
MaxResponseSize: cfg.UpstreamMaxResponseSize,
AllowHTTP: cfg.AllowHTTP,
MaxConnectionsPerHost: cfg.UpstreamConnectionsPerHost,
MaxConnections: cfg.UpstreamConnections,
BlockedNetworks: cfg.BlockedNetworks,
}
}
// newProcessor builds the image processor from cfg. Its input limit is
// the largest upstream response the fetcher accepts.
func newProcessor(cfg *config.Config) *imageprocessor.ImageProcessor {
return imageprocessor.New(imageprocessor.Params{
MaxInputBytes: cfg.UpstreamMaxResponseSize,
MaxConcurrentProcessing: cfg.MaxConcurrentProcessing,
})
}
// imageServiceParams defines dependencies for newImageService.
type imageServiceParams struct {
fx.In
Config *config.Config
Cache *imgcache.Cache
Fetcher *httpfetcher.HTTPFetcher
Processor *imageprocessor.ImageProcessor
Logger *logger.Logger
}
// newImageService builds the image service from the cache, the fetcher
// and the processor, with the signing key, the allowlist and the upstream
// settings of the config.
func newImageService(
_ fx.Lifecycle, params imageServiceParams,
) (*imgcache.Service, error) {
cfg := params.Config
return imgcache.NewService(&imgcache.ServiceConfig{
Cache: params.Cache,
Fetcher: params.Fetcher,
Processor: params.Processor,
SigningKey: cfg.SigningKey,
Allowlist: cfg.AllowlistHosts,
AllowHTTP: cfg.AllowHTTP,
MaxResponseSize: cfg.UpstreamMaxResponseSize,
Logger: params.Logger.Get(),
})
}
// sessionManagerParams defines dependencies for newSessionManager.
type sessionManagerParams struct {
fx.In
Config *config.Config
Database *database.Database
}
// newSessionManager builds the session manager. Session cookies are
// always Secure, HttpOnly and SameSite=Strict. The random value that each
// logout replaces is saved in the database, so it outlasts a restart.
func newSessionManager(
_ fx.Lifecycle, params sessionManagerParams,
) (*session.Manager, error) {
return session.NewManagerWithDatabase(
context.Background(), params.Config.SigningKey, params.Database.DB())
}
// newEncryptedURLGenerator builds the encrypted URL generator, with a key
// derived from the signing key.
func newEncryptedURLGenerator(cfg *config.Config) (*encurl.Generator, error) {
return encurl.NewGenerator(cfg.SigningKey)
}
+24 -17
View File
@@ -32,6 +32,7 @@ import (
"sneak.berlin/go/pixa/internal/imagerequest" "sneak.berlin/go/pixa/internal/imagerequest"
"sneak.berlin/go/pixa/internal/logger" "sneak.berlin/go/pixa/internal/logger"
"sneak.berlin/go/pixa/internal/middleware" "sneak.berlin/go/pixa/internal/middleware"
"sneak.berlin/go/pixa/internal/providers"
) )
// upstreamHost is the upstream host of the image URLs below. It is a // upstreamHost is the upstream host of the image URLs below. It is a
@@ -138,22 +139,21 @@ func startImageProxy(
AllowHTTP: true, AllowHTTP: true,
// A limit of its own, so the cache does not size itself from the // A limit of its own, so the cache does not size itself from the
// host's free disk space. // host's free disk space.
CacheMaxBytes: 64 << 20, CacheMaxBytes: 64 << 20,
CacheMaxBytesExplicit: true, CacheMaxBytesExplicit: true,
UpstreamMaxResponseSize: config.DefaultUpstreamMaxResponseSize, UpstreamFetchTimeout: config.DefaultUpstreamFetchTimeout,
DownstreamTimeout: config.DefaultDownstreamTimeout, UpstreamMaxResponseSize: config.DefaultUpstreamMaxResponseSize,
UpstreamConnectionsPerHost: config.DefaultUpstreamConnectionsPerHost,
UpstreamConnections: config.DefaultUpstreamConnections,
MaxConcurrentProcessing: 2,
DownstreamTimeout: config.DefaultDownstreamTimeout,
} }
fetcherCfg := httpfetcher.DefaultConfig() dialUpstream := func(ctx context.Context, network, _ string) (net.Conn, error) {
fetcherCfg.AllowHTTP = true
fetcherCfg.DialContext = func(
ctx context.Context, network, _ string,
) (net.Conn, error) {
var dialer net.Dialer var dialer net.Dialer
return dialer.DialContext(ctx, network, upstream.Listener.Addr().String()) return dialer.DialContext(ctx, network, upstream.Listener.Addr().String())
} }
fetcher := httpfetcher.New(fetcherCfg)
var ( var (
h *handlers.Handlers h *handlers.Handlers
@@ -170,9 +170,14 @@ func startImageProxy(
healthcheck.New, healthcheck.New,
handlers.New, handlers.New,
middleware.New, middleware.New,
func() httpfetcher.Fetcher { return fetcher },
func() prometheus.Registerer { return prometheus.NewRegistry() }, func() prometheus.Registerer { return prometheus.NewRegistry() },
), ),
fx.Provide(providers.Constructors()...),
fx.Decorate(func(fetcherConfig *httpfetcher.Config) *httpfetcher.Config {
fetcherConfig.DialContext = dialUpstream
return fetcherConfig
}),
fx.Populate(&h, &mw, &db), fx.Populate(&h, &mw, &db),
) )
app.RequireStart() app.RequireStart()
@@ -188,10 +193,10 @@ func startImageProxy(
// startImageProxyWithBlockedNetworks starts the components pixad's fx app // startImageProxyWithBlockedNetworks starts the components pixad's fx app
// builds, from a config with a fresh state directory and blockedNetworks as // builds, from a config with a fresh state directory and blockedNetworks as
// blocked_networks. Unlike startImageProxy it provides no upstream fetcher, so // blocked_networks. Unlike startImageProxy it keeps pixad's dial function,
// the handlers build the one pixad uses, which checks every address it // which checks every address the fetcher connects to against the built-in
// connects to against the built-in blocked ranges and blocked_networks. It // blocked ranges and blocked_networks. It returns the server with its
// returns the server with its routes. // routes.
func startImageProxyWithBlockedNetworks( func startImageProxyWithBlockedNetworks(
t *testing.T, blockedNetworks []netip.Prefix, t *testing.T, blockedNetworks []netip.Prefix,
) *Server { ) *Server {
@@ -207,8 +212,9 @@ func startImageProxyWithBlockedNetworks(
// fails within this time at the latest. // fails within this time at the latest.
UpstreamFetchTimeout: 2 * time.Second, UpstreamFetchTimeout: 2 * time.Second,
// With no connection slots, the fetch would fail before connecting. // With no connection slots, the fetch would fail before connecting.
UpstreamConnections: config.DefaultUpstreamConnections, UpstreamConnectionsPerHost: config.DefaultUpstreamConnectionsPerHost,
DownstreamTimeout: config.DefaultDownstreamTimeout, UpstreamConnections: config.DefaultUpstreamConnections,
DownstreamTimeout: config.DefaultDownstreamTimeout,
} }
var ( var (
@@ -227,6 +233,7 @@ func startImageProxyWithBlockedNetworks(
middleware.New, middleware.New,
func() prometheus.Registerer { return prometheus.NewRegistry() }, func() prometheus.Registerer { return prometheus.NewRegistry() },
), ),
fx.Provide(providers.Constructors()...),
fx.Populate(&h, &mw), fx.Populate(&h, &mw),
) )
app.RequireStart() app.RequireStart()
@@ -12,16 +12,7 @@ import (
"strings" "strings"
"testing" "testing"
"github.com/prometheus/client_golang/prometheus"
"go.uber.org/fx/fxtest"
"sneak.berlin/go/pixa/internal/config" "sneak.berlin/go/pixa/internal/config"
"sneak.berlin/go/pixa/internal/database"
"sneak.berlin/go/pixa/internal/globals"
"sneak.berlin/go/pixa/internal/handlers"
"sneak.berlin/go/pixa/internal/healthcheck"
"sneak.berlin/go/pixa/internal/logger"
"sneak.berlin/go/pixa/internal/middleware"
) )
// testSigningKey is a throwaway signing key; submitting it logs in. // testSigningKey is a throwaway signing key; submitting it logs in.
@@ -61,45 +52,7 @@ func newTestServer(t *testing.T) *Server {
DownstreamTimeout: config.DefaultDownstreamTimeout, DownstreamTimeout: config.DefaultDownstreamTimeout,
} }
lc := fxtest.NewLifecycle(t) return startPixa(t, cfg)
log, err := logger.New(lc, logger.Params{Globals: &globals.Globals{}})
if err != nil {
t.Fatalf("logger.New() error = %v", err)
}
db, err := database.New(lc, database.Params{Logger: log, Config: cfg})
if err != nil {
t.Fatalf("database.New() error = %v", err)
}
hc, err := healthcheck.New(lc, healthcheck.Params{
Globals: &globals.Globals{}, Config: cfg, Logger: log, Database: db,
})
if err != nil {
t.Fatalf("healthcheck.New() error = %v", err)
}
h, err := handlers.New(lc, handlers.Params{
Logger: log, Healthcheck: hc, Database: db, Config: cfg,
MetricsRegistry: prometheus.NewRegistry(),
})
if err != nil {
t.Fatalf("handlers.New() error = %v", err)
}
mw, err := middleware.New(lc, middleware.Params{Logger: log, Config: cfg})
if err != nil {
t.Fatalf("middleware.New() error = %v", err)
}
lc.RequireStart()
t.Cleanup(lc.RequireStop)
s := &Server{config: cfg, mw: mw, h: h}
s.SetupRoutes()
return s
} }
// clientRequest builds a request for / arriving from remoteAddr, carrying // clientRequest builds a request for / arriving from remoteAddr, carrying
+3 -1
View File
@@ -20,6 +20,7 @@ import (
"sneak.berlin/go/pixa/internal/healthcheck" "sneak.berlin/go/pixa/internal/healthcheck"
"sneak.berlin/go/pixa/internal/logger" "sneak.berlin/go/pixa/internal/logger"
"sneak.berlin/go/pixa/internal/middleware" "sneak.berlin/go/pixa/internal/middleware"
"sneak.berlin/go/pixa/internal/providers"
"sneak.berlin/go/pixa/internal/session" "sneak.berlin/go/pixa/internal/session"
) )
@@ -29,7 +30,7 @@ var logoutFormPattern = regexp.MustCompile(
`(?s)action="/logout".*?name="gorilla\.csrf\.Token" value="([^"]+)"`) `(?s)action="/logout".*?name="gorilla\.csrf\.Token" value="([^"]+)"`)
// startPixa starts the components pixad's fx app builds, from cfg, and // startPixa starts the components pixad's fx app builds, from cfg, and
// returns the server with its routes, as newTestServer does. // returns the server with its routes.
func startPixa(t *testing.T, cfg *config.Config) *Server { func startPixa(t *testing.T, cfg *config.Config) *Server {
t.Helper() t.Helper()
@@ -49,6 +50,7 @@ func startPixa(t *testing.T, cfg *config.Config) *Server {
middleware.New, middleware.New,
func() prometheus.Registerer { return prometheus.NewRegistry() }, func() prometheus.Registerer { return prometheus.NewRegistry() },
), ),
fx.Provide(providers.Constructors()...),
fx.Populate(&h, &mw), fx.Populate(&h, &mw),
) )
app.RequireStart() app.RequireStart()