diff --git a/README.md b/README.md index 87bf02b..d2170b9 100644 --- a/README.md +++ b/README.md @@ -812,7 +812,9 @@ See `configs/config.example.yml` for all options with defaults. ### 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 - **Image processing**: govips (CGO wrapper for libvips) - **Database**: SQLite via modernc.org/sqlite diff --git a/TODO.md b/TODO.md index 8c2ff37..2f56525 100644 --- a/TODO.md +++ b/TODO.md @@ -30,6 +30,30 @@ P2: security: per-origin rate limiting # 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 handlers tests sign their image URLs with `generateSignedURL`, a test helper that writes them as the removed generator did. Removed: diff --git a/internal/app/app.go b/internal/app/app.go index c09f554..0d9daf9 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -17,6 +17,7 @@ import ( "sneak.berlin/go/pixa/internal/healthcheck" "sneak.berlin/go/pixa/internal/logger" "sneak.berlin/go/pixa/internal/middleware" + "sneak.berlin/go/pixa/internal/providers" "sneak.berlin/go/pixa/internal/server" ) @@ -65,6 +66,7 @@ func run(_ *cobra.Command, _ []string) { // The registry /metrics serves func() prometheus.Registerer { return prometheus.DefaultRegisterer }, ), + fx.Provide(providers.Constructors()...), fx.Invoke( func(log *logger.Logger) { log.Identify() }, func(*server.Server) {}, diff --git a/internal/config/cache_max_bytes_internal_test.go b/internal/config/cache_max_bytes_internal_test.go index 6af1554..3538530 100644 --- a/internal/config/cache_max_bytes_internal_test.go +++ b/internal/config/cache_max_bytes_internal_test.go @@ -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 -// cache_max_bytes is recorded as not explicit, so the cache works out -// the default when it opens, and that an explicit zero is recorded as -// explicit, so it disables the disk cache instead. +// cache_max_bytes is recorded as not explicit, so pixad works out the +// default when it builds the cache, and that an explicit zero is recorded +// as explicit, so it disables the disk cache instead. func TestCacheMaxBytesExplicitIsRecorded(t *testing.T) { t.Parallel() diff --git a/internal/config/config.go b/internal/config/config.go index eff51ce..3f484b7 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -45,6 +45,37 @@ func DefaultMaxConcurrentProcessing() int { 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. const ( keyDebug = "debug" @@ -190,18 +221,14 @@ type Config struct { // address, and an explicit list replaces the default. TrustedProxies []netip.Prefix - // CacheMaxBytes is the disk cache size limit in bytes. Only an - // explicit zero (CacheMaxBytesExplicit true) disables the disk - // cache. Zero with CacheMaxBytesExplicit false means - // cache_max_bytes was omitted, and the cache works out the default - // limit when it opens. + // CacheMaxBytes is the disk cache size limit in bytes, as set; 0 + // turns the disk cache off. CacheMaxBytes int64 // CacheMaxBytesExplicit records whether cache_max_bytes was // explicitly set, in the environment or the configuration file. - // Explicit values are used exactly as given; for an omitted key the - // cache works out the default limit when it opens (see - // imgcache.CacheConfig.UseDefaultMaxBytes). + // Explicit values are used exactly as given. For an omitted key + // pixad works out DefaultCacheMaxBytes when it builds the cache. CacheMaxBytesExplicit bool } @@ -239,8 +266,8 @@ func New(_ fx.Lifecycle, params Params) (*Config, error) { return nil, err } - // An omitted cache_max_bytes is worked out and logged when the - // cache opens. + // An omitted cache_max_bytes is worked out and logged when pixad + // builds the cache. if c.CacheMaxBytesExplicit { log.Info("effective cache size limit", "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 cache opens; here we only record whether the operator set - // the key explicitly. + // pixad builds the cache; here we only record whether the operator + // set the key explicitly. _, c.CacheMaxBytesExplicit = lookupValue(sc, keyCacheMaxBytes) // Build DBURL from StateDir if not explicitly set. The derived URL diff --git a/internal/database/database.go b/internal/database/database.go index 4eea04e..31b74e2 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -82,7 +82,11 @@ func ParseMigrationVersion(filename string) (int, error) { 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) { s := &Database{ log: params.Logger.Get(), @@ -91,20 +95,16 @@ func New(lc fx.Lifecycle, params Params) (*Database, error) { s.log.Info("Database instantiated") - lc.Append(fx.Hook{ - OnStart: func(ctx context.Context) error { - s.log.Info("Database OnStart Hook") + err := s.connect(context.Background()) + if err != nil { + return nil, err + } - return s.connect(ctx) - }, + lc.Append(fx.Hook{ OnStop: func(_ context.Context) error { s.log.Info("Database OnStop Hook") - if s.db != nil { - return s.db.Close() - } - - return nil + return s.db.Close() }, }) diff --git a/internal/handlers/cache_hit_internal_test.go b/internal/handlers/cache_hit_internal_test.go index 713395a..56d1dda 100644 --- a/internal/handlers/cache_hit_internal_test.go +++ b/internal/handlers/cache_hit_internal_test.go @@ -21,6 +21,7 @@ import ( "sneak.berlin/go/pixa/internal/database" "sneak.berlin/go/pixa/internal/encurl" + "sneak.berlin/go/pixa/internal/imageprocessor" "sneak.berlin/go/pixa/internal/imagerequest" "sneak.berlin/go/pixa/internal/imgcache" ) @@ -121,6 +122,7 @@ func newCountingDatabaseServer( cache, err := imgcache.NewCache(db, imgcache.CacheConfig{ StateDir: tb.TempDir(), NegativeTTL: 5 * time.Minute, + MaxBytes: 1 << 30, }, prometheus.NewRegistry()) if err != nil { tb.Fatalf("imgcache.NewCache() error = %v", err) @@ -131,9 +133,14 @@ func newCountingDatabaseServer( Fetcher: newMockFetcher(fstest.MapFS{ countingHost + photoPath: &fstest.MapFile{Data: photo.Bytes()}, }), - SigningKey: testSigningKey, - Allowlist: []string{countingHost}, - Logger: slog.New(slog.DiscardHandler), + Processor: imageprocessor.New(imageprocessor.Params{ + MaxInputBytes: 50 << 20, + MaxConcurrentProcessing: 2, + }), + SigningKey: testSigningKey, + Allowlist: []string{countingHost}, + MaxResponseSize: 50 << 20, + Logger: slog.New(slog.DiscardHandler), }) if err != nil { tb.Fatalf("imgcache.NewService() error = %v", err) diff --git a/internal/handlers/cache_max_bytes_internal_test.go b/internal/handlers/cache_max_bytes_internal_test.go deleted file mode 100644 index 1fec51d..0000000 --- a/internal/handlers/cache_max_bytes_internal_test.go +++ /dev/null @@ -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) - } - }) - } -} diff --git a/internal/handlers/fetcher_internal_test.go b/internal/handlers/fetcher_internal_test.go deleted file mode 100644 index e358f1b..0000000 --- a/internal/handlers/fetcher_internal_test.go +++ /dev/null @@ -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") -} diff --git a/internal/handlers/first_bytes_internal_test.go b/internal/handlers/first_bytes_internal_test.go index eeb80fe..4a31a7b 100644 --- a/internal/handlers/first_bytes_internal_test.go +++ b/internal/handlers/first_bytes_internal_test.go @@ -15,6 +15,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "sneak.berlin/go/pixa/internal/encurl" "sneak.berlin/go/pixa/internal/httpfetcher" + "sneak.berlin/go/pixa/internal/imageprocessor" "sneak.berlin/go/pixa/internal/imagerequest" "sneak.berlin/go/pixa/internal/imgcache" ) @@ -36,17 +37,23 @@ func newRoutesFetchingThrough( cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{ StateDir: t.TempDir(), NegativeTTL: 5 * time.Minute, + MaxBytes: 1 << 30, }, prometheus.NewRegistry()) if err != nil { t.Fatalf("imgcache.NewCache() error = %v", err) } svc, err := imgcache.NewService(&imgcache.ServiceConfig{ - Cache: cache, - Fetcher: fetcher, - SigningKey: testSigningKey, - Allowlist: []string{allowlistedHost}, - Logger: log, + Cache: cache, + Fetcher: fetcher, + Processor: imageprocessor.New(imageprocessor.Params{ + MaxInputBytes: 50 << 20, + MaxConcurrentProcessing: 2, + }), + SigningKey: testSigningKey, + Allowlist: []string{allowlistedHost}, + MaxResponseSize: 50 << 20, + Logger: log, }) if err != nil { t.Fatalf("imgcache.NewService() error = %v", err) diff --git a/internal/handlers/handlers.go b/internal/handlers/handlers.go index ba42481..555e9c6 100644 --- a/internal/handlers/handlers.go +++ b/internal/handlers/handlers.go @@ -4,19 +4,15 @@ package handlers import ( "context" "encoding/json" - "errors" "log/slog" "net/http" "time" - "github.com/prometheus/client_golang/prometheus" "go.uber.org/fx" "sneak.berlin/go/pixa/internal/allowlist" "sneak.berlin/go/pixa/internal/config" - "sneak.berlin/go/pixa/internal/database" "sneak.berlin/go/pixa/internal/encurl" "sneak.berlin/go/pixa/internal/healthcheck" - "sneak.berlin/go/pixa/internal/httpfetcher" "sneak.berlin/go/pixa/internal/imgcache" "sneak.berlin/go/pixa/internal/logger" "sneak.berlin/go/pixa/internal/session" @@ -28,29 +24,19 @@ type Params struct { Logger *logger.Logger Healthcheck *healthcheck.Healthcheck - Database *database.Database Config *config.Config - // MetricsRegistry is the registry the image cache registers its - // counters on. pixad provides Prometheus' default registry, the one - // /metrics serves. - 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"` + ImageService *imgcache.Service + SessionManager *session.Manager + EncryptedURLGenerator *encurl.Generator } // Handlers provides HTTP request handlers. type Handlers struct { log *slog.Logger hc *healthcheck.Healthcheck - db *database.Database config *config.Config - fetcher httpfetcher.Fetcher imgSvc *imgcache.Service - imgCache *imgcache.Cache sessMgr *session.Manager encGen *encurl.Generator csrfProtect func(http.Handler) http.Handler @@ -58,50 +44,25 @@ type Handlers struct { // refererBlocklist matches the hosts of referer_blocklist; its IsAllowed // reports whether a URL's host is on that list. refererBlocklist *allowlist.HostAllowList - - // metricsRegistry is the registry the image cache registers its - // counters on. - metricsRegistry prometheus.Registerer } // 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) if err != nil { return nil, err } - s := &Handlers{ + return &Handlers{ log: params.Logger.Get(), hc: params.Healthcheck, - db: params.Database, config: params.Config, - fetcher: params.Fetcher, + imgSvc: params.ImageService, + sessMgr: params.SessionManager, + encGen: params.EncryptedURLGenerator, csrfProtect: csrfProtect, refererBlocklist: allowlist.New(params.Config.RefererBlocklist), - metricsRegistry: params.MetricsRegistry, - } - - 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 + }, nil } // 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) } -// 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) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) diff --git a/internal/handlers/handlers_internal_test.go b/internal/handlers/handlers_internal_test.go index d175569..e463dda 100644 --- a/internal/handlers/handlers_internal_test.go +++ b/internal/handlers/handlers_internal_test.go @@ -21,6 +21,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "sneak.berlin/go/pixa/internal/database" "sneak.berlin/go/pixa/internal/httpfetcher" + "sneak.berlin/go/pixa/internal/imageprocessor" "sneak.berlin/go/pixa/internal/imgcache" ) @@ -48,16 +49,22 @@ func setupTestHandler(t *testing.T) *testFixtures { cache, err := imgcache.NewCache(db, imgcache.CacheConfig{ StateDir: tmpDir, NegativeTTL: 5 * time.Minute, + MaxBytes: 1 << 30, }, prometheus.NewRegistry()) if err != nil { t.Fatalf("failed to create cache: %v", err) } svc, err := imgcache.NewService(&imgcache.ServiceConfig{ - Cache: cache, - Fetcher: newMockFetcher(mockFS), - SigningKey: "test-signing-key-must-be-32-chars", - Allowlist: []string{goodHost}, + Cache: cache, + Fetcher: newMockFetcher(mockFS), + Processor: imageprocessor.New(imageprocessor.Params{ + MaxInputBytes: 50 << 20, + MaxConcurrentProcessing: 2, + }), + SigningKey: "test-signing-key-must-be-32-chars", + Allowlist: []string{goodHost}, + MaxResponseSize: 50 << 20, }) if err != nil { t.Fatalf("failed to create service: %v", err) diff --git a/internal/handlers/image_cache_control_internal_test.go b/internal/handlers/image_cache_control_internal_test.go index 79c4a38..b27465d 100644 --- a/internal/handlers/image_cache_control_internal_test.go +++ b/internal/handlers/image_cache_control_internal_test.go @@ -15,6 +15,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "sneak.berlin/go/pixa/internal/encurl" + "sneak.berlin/go/pixa/internal/imageprocessor" "sneak.berlin/go/pixa/internal/imagerequest" "sneak.berlin/go/pixa/internal/imgcache" ) @@ -34,6 +35,7 @@ func newSignedHostServer( cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{ StateDir: t.TempDir(), NegativeTTL: 5 * time.Minute, + MaxBytes: 1 << 30, }, prometheus.NewRegistry()) if err != nil { t.Fatalf("imgcache.NewCache() error = %v", err) @@ -46,8 +48,13 @@ func newSignedHostServer( Fetcher: newMockFetcher(fstest.MapFS{ signedHost + photoPath: &fstest.MapFile{Data: jpegData}, }), - SigningKey: testSigningKey, - Logger: log, + Processor: imageprocessor.New(imageprocessor.Params{ + MaxInputBytes: 50 << 20, + MaxConcurrentProcessing: 2, + }), + SigningKey: testSigningKey, + MaxResponseSize: 50 << 20, + Logger: log, }) if err != nil { t.Fatalf("imgcache.NewService() error = %v", err) diff --git a/internal/handlers/image_errors_internal_test.go b/internal/handlers/image_errors_internal_test.go index ef7130a..e4143ef 100644 --- a/internal/handlers/image_errors_internal_test.go +++ b/internal/handlers/image_errors_internal_test.go @@ -15,6 +15,7 @@ import ( "github.com/go-chi/chi/v5" "github.com/prometheus/client_golang/prometheus" "sneak.berlin/go/pixa/internal/httpfetcher" + "sneak.berlin/go/pixa/internal/imageprocessor" "sneak.berlin/go/pixa/internal/imagerequest" "sneak.berlin/go/pixa/internal/imgcache" "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{ StateDir: t.TempDir(), NegativeTTL: 5 * time.Minute, + MaxBytes: 1 << 30, }, prometheus.NewRegistry()) if err != nil { t.Fatalf("failed to create cache: %v", err) } svc, err := imgcache.NewService(&imgcache.ServiceConfig{ - Cache: cache, - Fetcher: fetcher, - SigningKey: testSigningKey, - Allowlist: []string{allowlistedHost}, + Cache: cache, + Fetcher: fetcher, + Processor: imageprocessor.New(imageprocessor.Params{ + MaxInputBytes: 50 << 20, + MaxConcurrentProcessing: 2, + }), + SigningKey: testSigningKey, + Allowlist: []string{allowlistedHost}, + MaxResponseSize: 50 << 20, }) if err != nil { t.Fatalf("failed to create service: %v", err) @@ -169,7 +176,12 @@ func TestHandleImage_ErrorAnswers(t *testing.T) { photos := newPhotoFetcher(t, signedHost, parentHost, siblingHost, subdomainHost, appendedHost) // 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) expired := time.Now().Add(-time.Hour) diff --git a/internal/handlers/referer_blocklist_internal_test.go b/internal/handlers/referer_blocklist_internal_test.go index 6cdfbe6..f3160a5 100644 --- a/internal/handlers/referer_blocklist_internal_test.go +++ b/internal/handlers/referer_blocklist_internal_test.go @@ -14,6 +14,7 @@ import ( "sneak.berlin/go/pixa/internal/allowlist" "sneak.berlin/go/pixa/internal/encurl" "sneak.berlin/go/pixa/internal/httpfetcher" + "sneak.berlin/go/pixa/internal/imageprocessor" "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{ StateDir: t.TempDir(), NegativeTTL: 5 * time.Minute, + MaxBytes: 1 << 30, }, prometheus.NewRegistry()) if err != nil { t.Fatalf("imgcache.NewCache() error = %v", err) } svc, err := imgcache.NewService(&imgcache.ServiceConfig{ - Cache: cache, - Fetcher: fetcher, - SigningKey: testSigningKey, - Allowlist: []string{allowlistedHost}, + Cache: cache, + Fetcher: fetcher, + Processor: imageprocessor.New(imageprocessor.Params{ + MaxInputBytes: 50 << 20, + MaxConcurrentProcessing: 2, + }), + SigningKey: testSigningKey, + Allowlist: []string{allowlistedHost}, + MaxResponseSize: 50 << 20, }) if err != nil { t.Fatalf("imgcache.NewService() error = %v", err) diff --git a/internal/httpfetcher/blocked_networks_internal_test.go b/internal/httpfetcher/blocked_networks_internal_test.go index dfbf8f5..d28d599 100644 --- a/internal/httpfetcher/blocked_networks_internal_test.go +++ b/internal/httpfetcher/blocked_networks_internal_test.go @@ -7,6 +7,7 @@ import ( "net/http" "net/netip" "testing" + "time" ) // 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) { t.Parallel() - cfg := DefaultConfig() - // 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. - cfg.BlockedNetworks = []netip.Prefix{netip.MustParsePrefix("198.51.100.0/24")} - - transport := transportOf(t, New(cfg)) + transport := transportOf(t, New(&Config{ + Timeout: 30 * time.Second, + MaxResponseSize: 50 << 20, + MaxConnectionsPerHost: 20, + MaxConnections: 64, + // 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{ "198.51.100.5:80", // operator-supplied range diff --git a/internal/httpfetcher/dial_context_internal_test.go b/internal/httpfetcher/dial_context_internal_test.go index 4553799..036bd78 100644 --- a/internal/httpfetcher/dial_context_internal_test.go +++ b/internal/httpfetcher/dial_context_internal_test.go @@ -4,6 +4,7 @@ import ( "errors" "net" "testing" + "time" ) // TestNewUsesCheckedDialerWithoutDialContext checks that a fetcher built @@ -13,7 +14,12 @@ func TestNewUsesCheckedDialerWithoutDialContext(t *testing.T) { t.Parallel() 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() @@ -32,10 +38,14 @@ func TestDialContextReplacesOnlyTheDialer(t *testing.T) { srv := startUpstream(t) dialer := &recordingDialer{target: srv.Listener.Addr().String()} - cfg := DefaultConfig() - cfg.AllowHTTP = true - cfg.DialContext = dialer.dialContext - f := New(cfg) + f := New(&Config{ + Timeout: 30 * time.Second, + MaxResponseSize: 50 << 20, + AllowHTTP: true, + MaxConnectionsPerHost: 20, + MaxConnections: 64, + DialContext: dialer.dialContext, + }) if body := fetchBody(t, f, "/image"); body != imagePayload { t.Errorf("body = %q, want %q", body, imagePayload) diff --git a/internal/httpfetcher/fetch_internal_test.go b/internal/httpfetcher/fetch_internal_test.go index b6523a8..f2f3e83 100644 --- a/internal/httpfetcher/fetch_internal_test.go +++ b/internal/httpfetcher/fetch_internal_test.go @@ -120,7 +120,12 @@ func newServerFetcher( t.Helper() if cfg == nil { - cfg = DefaultConfig() + cfg = &Config{ + Timeout: 30 * time.Second, + MaxResponseSize: 50 << 20, + MaxConnectionsPerHost: 20, + MaxConnections: 64, + } } cfg.AllowHTTP = true @@ -280,8 +285,12 @@ func TestFetchMaxResponseSizeEnforced(t *testing.T) { srv := startUpstream(t) - cfg := DefaultConfig() - cfg.MaxResponseSize = 8 + cfg := &Config{ + Timeout: 30 * time.Second, + MaxResponseSize: 8, + MaxConnectionsPerHost: 20, + MaxConnections: 64, + } f, _ := newServerFetcher(t, srv, cfg) @@ -310,8 +319,12 @@ func fetchWithMaxResponseSize( srv := httptest.NewServer(handler) t.Cleanup(srv.Close) - cfg := DefaultConfig() - cfg.MaxResponseSize = maxResponseSize + cfg := &Config{ + Timeout: 30 * time.Second, + MaxResponseSize: maxResponseSize, + MaxConnectionsPerHost: 20, + MaxConnections: 64, + } f, _ := newServerFetcher(t, srv, cfg) @@ -436,8 +449,12 @@ func TestFetchSemaphoreReleasedOnError(t *testing.T) { srv := startUpstream(t) - cfg := DefaultConfig() - cfg.MaxConnectionsPerHost = 1 + cfg := &Config{ + Timeout: 30 * time.Second, + MaxResponseSize: 50 << 20, + MaxConnectionsPerHost: 1, + MaxConnections: 64, + } f, _ := newServerFetcher(t, srv, cfg) @@ -466,8 +483,12 @@ func assertSlotReleasedByClose( srv := startUpstream(t) - cfg := DefaultConfig() - cfg.MaxConnectionsPerHost = 1 + cfg := &Config{ + Timeout: 30 * time.Second, + MaxResponseSize: 50 << 20, + MaxConnectionsPerHost: 1, + MaxConnections: 64, + } f, _ := newServerFetcher(t, srv, cfg) diff --git a/internal/httpfetcher/httpfetcher.go b/internal/httpfetcher/httpfetcher.go index 8988d55..c9d2eae 100644 --- a/internal/httpfetcher/httpfetcher.go +++ b/internal/httpfetcher/httpfetcher.go @@ -24,16 +24,15 @@ import ( // Fetcher configuration constants. const ( - DefaultFetchTimeout = 30 * time.Second - DefaultMaxResponseSize = 50 << 20 // 50MB - DefaultTLSTimeout = 10 * time.Second - DefaultMaxIdleConns = 100 - DefaultIdleConnTimeout = 90 * time.Second - DefaultMaxRedirects = 10 - DefaultMaxConnectionsPerHost = 20 - DefaultMaxConnections = 64 + DefaultTLSTimeout = 10 * time.Second + DefaultMaxIdleConns = 100 + DefaultIdleConnTimeout = 90 * time.Second + DefaultMaxRedirects = 10 ) +// 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 // MaxConnections fetches are already in progress. const ConnectionWaitTimeout = 10 * time.Second @@ -120,16 +119,26 @@ type FetchResult struct { 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 { // Timeout for upstream requests. Timeout time.Duration // MaxResponseSize is the maximum allowed response body size. 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 bool // 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) } -// 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 // per host and for all hosts together. type HTTPFetcher struct { @@ -194,10 +183,6 @@ type hostSemaphore struct { // New creates a new HTTPFetcher with SSRF protection. func New(config *Config) *HTTPFetcher { - if config == nil { - config = DefaultConfig() - } - // Unless config.DialContext replaces it, the transport connects with // the SSRF-safe dialer, which re-resolves and re-checks at connect time // (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), } - req.Header.Set("User-Agent", f.config.UserAgent) - req.Header.Set("Accept", strings.Join(f.config.AllowedContentTypes, ", ")) + req.Header.Set("User-Agent", userAgent) + req.Header.Set("Accept", strings.Join(allowedContentTypes(), ", ")) // The ID of the request this fetch serves, so the fetch can be found in // the upstream host's logs @@ -460,7 +445,7 @@ func (f *HTTPFetcher) isAllowedContentType(contentType string) bool { // Extract the MIME type without parameters mediaType := strings.TrimSpace(strings.Split(contentType, ";")[0]) - for _, allowed := range f.config.AllowedContentTypes { + for _, allowed := range allowedContentTypes() { if strings.EqualFold(mediaType, allowed) { return true } diff --git a/internal/httpfetcher/httpfetcher_internal_test.go b/internal/httpfetcher/httpfetcher_internal_test.go index c7a8970..7341d28 100644 --- a/internal/httpfetcher/httpfetcher_internal_test.go +++ b/internal/httpfetcher/httpfetcher_internal_test.go @@ -7,60 +7,21 @@ import ( "net" "testing" "testing/fstest" + "time" ) // testHost is the hostname used by mock fetch tests. 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) { t.Parallel() - f := New(DefaultConfig()) + f := New(&Config{ + Timeout: 30 * time.Second, + MaxResponseSize: 50 << 20, + MaxConnectionsPerHost: 20, + MaxConnections: 64, + }) tests := []struct { contentType string diff --git a/internal/httpfetcher/jpegxl_internal_test.go b/internal/httpfetcher/jpegxl_internal_test.go index 8be65ec..4df240b 100644 --- a/internal/httpfetcher/jpegxl_internal_test.go +++ b/internal/httpfetcher/jpegxl_internal_test.go @@ -1,16 +1,26 @@ package httpfetcher -import "testing" +import ( + "testing" + "time" +) // TestJPEGXLContentType verifies that the fetcher accepts an upstream answer -// of type image/jxl by default, and that the mock fetcher serves a .jxl file -// as image/jxl. +// of type image/jxl, and that the mock fetcher serves a .jxl file as +// image/jxl. func TestJPEGXLContentType(t *testing.T) { t.Parallel() 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) } diff --git a/internal/httpfetcher/max_connections_internal_test.go b/internal/httpfetcher/max_connections_internal_test.go index a776460..d414178 100644 --- a/internal/httpfetcher/max_connections_internal_test.go +++ b/internal/httpfetcher/max_connections_internal_test.go @@ -18,14 +18,6 @@ func imageURLOnPort(port int) string { "/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 // 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 @@ -37,8 +29,12 @@ func TestFetchLimitsConnectionsToAllHostsTogether(t *testing.T) { srv := startUpstream(t) - cfg := DefaultConfig() - cfg.MaxConnections = 2 + cfg := &Config{ + Timeout: 30 * time.Second, + MaxResponseSize: 50 << 20, + MaxConnectionsPerHost: 20, + MaxConnections: 2, + } f, _ := newServerFetcher(t, srv, cfg) f.connectionWaitTimeout = 100 * time.Millisecond @@ -94,8 +90,12 @@ func TestFetchFreesHostSlotWhenContextEndsWaitingForConnection(t *testing.T) { srv := startUpstream(t) - cfg := DefaultConfig() - cfg.MaxConnections = 1 + cfg := &Config{ + Timeout: 30 * time.Second, + MaxResponseSize: 50 << 20, + MaxConnectionsPerHost: 20, + MaxConnections: 1, + } f, _ := newServerFetcher(t, srv, cfg) @@ -163,9 +163,12 @@ func TestFetchRemovesHostSemaphoreWhenNoConnection(t *testing.T) { srv := startUpstream(t) - cfg := DefaultConfig() - cfg.MaxConnections = 1 - cfg.MaxConnectionsPerHost = 1 + cfg := &Config{ + Timeout: 30 * time.Second, + MaxResponseSize: 50 << 20, + MaxConnectionsPerHost: 1, + MaxConnections: 1, + } f, _ := newServerFetcher(t, srv, cfg) f.connectionWaitTimeout = 100 * time.Millisecond @@ -236,8 +239,12 @@ func TestFetchReleasesConnectionOnError(t *testing.T) { srv := startUpstream(t) - cfg := DefaultConfig() - cfg.MaxConnections = 1 + cfg := &Config{ + Timeout: 30 * time.Second, + MaxResponseSize: 50 << 20, + MaxConnectionsPerHost: 20, + MaxConnections: 1, + } f, _ := newServerFetcher(t, srv, cfg) f.connectionWaitTimeout = 100 * time.Millisecond diff --git a/internal/imageprocessor/avif_too_large_internal_test.go b/internal/imageprocessor/avif_too_large_internal_test.go index 1c1a576..15b184c 100644 --- a/internal/imageprocessor/avif_too_large_internal_test.go +++ b/internal/imageprocessor/avif_too_large_internal_test.go @@ -69,7 +69,10 @@ func TestImageProcessor_RefusesAVIFAboveMaxPixels(t *testing.T) { t.Run(tt.name, func(t *testing.T) { 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{ Size: imagerequest.Size{Width: tt.width, Height: tt.height}, Format: tt.format, diff --git a/internal/imageprocessor/computed_side_internal_test.go b/internal/imageprocessor/computed_side_internal_test.go index fbd49e6..6bfb23a 100644 --- a/internal/imageprocessor/computed_side_internal_test.go +++ b/internal/imageprocessor/computed_side_internal_test.go @@ -38,7 +38,10 @@ func TestImageProcessor_RefusesComputedSideAboveMaximum(t *testing.T) { 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), &imagerequest.Transform{ Size: tt.size, @@ -104,7 +107,10 @@ func TestImageProcessor_ComputedSideIsAtLeastOne(t *testing.T) { 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), &imagerequest.Transform{ Size: tt.size, diff --git a/internal/imageprocessor/deadline_internal_test.go b/internal/imageprocessor/deadline_internal_test.go index fb9b2ea..9f6d394 100644 --- a/internal/imageprocessor/deadline_internal_test.go +++ b/internal/imageprocessor/deadline_internal_test.go @@ -84,7 +84,7 @@ func checkProcessStopsAtDeadline( ) { t.Helper() - proc := New(Params{MaxConcurrentProcessing: 1}) + proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1}) result, err := proc.Process(ctx, input, smallJPEGTransform()) if result != nil { diff --git a/internal/imageprocessor/empty_format_internal_test.go b/internal/imageprocessor/empty_format_internal_test.go index 4feda08..48bba8e 100644 --- a/internal/imageprocessor/empty_format_internal_test.go +++ b/internal/imageprocessor/empty_format_internal_test.go @@ -14,7 +14,10 @@ import ( func TestImageProcessor_EmptyFormatRefused(t *testing.T) { t.Parallel() - _, err := New(Params{}).Process( + _, err := New(Params{ + MaxInputBytes: 50 << 20, + MaxConcurrentProcessing: 1, + }).Process( t.Context(), bytes.NewReader(createTestJPEG(t, 20, 20)), &imagerequest.Transform{}, ) diff --git a/internal/imageprocessor/encoder_settings_internal_test.go b/internal/imageprocessor/encoder_settings_internal_test.go index 8cf3388..dc93d76 100644 --- a/internal/imageprocessor/encoder_settings_internal_test.go +++ b/internal/imageprocessor/encoder_settings_internal_test.go @@ -18,7 +18,10 @@ func processedSize( ) int64 { 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 { t.Fatalf("Process() error = %v", err) } diff --git a/internal/imageprocessor/fit_cover_internal_test.go b/internal/imageprocessor/fit_cover_internal_test.go index 030074e..a12d155 100644 --- a/internal/imageprocessor/fit_cover_internal_test.go +++ b/internal/imageprocessor/fit_cover_internal_test.go @@ -36,7 +36,10 @@ func TestImageProcessor_FitCoverServesThinSource(t *testing.T) { 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), &imagerequest.Transform{ Size: tt.size, @@ -106,7 +109,10 @@ func TestImageProcessor_FitCoverCropsTheCentre(t *testing.T) { 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), &imagerequest.Transform{ Size: imagerequest.Size{Width: 50, Height: 50}, @@ -230,7 +236,10 @@ func TestImageProcessor_FitCoverKeepsProportions(t *testing.T) { 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), &imagerequest.Transform{ Size: tt.size, diff --git a/internal/imageprocessor/fit_outside_internal_test.go b/internal/imageprocessor/fit_outside_internal_test.go index 3d331a5..3c72d59 100644 --- a/internal/imageprocessor/fit_outside_internal_test.go +++ b/internal/imageprocessor/fit_outside_internal_test.go @@ -41,7 +41,10 @@ func TestImageProcessor_RefusesFitOutsideAboveMaximum(t *testing.T) { 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), &imagerequest.Transform{ Size: tt.size, @@ -95,7 +98,10 @@ func TestImageProcessor_FitOutsideSideIsAtLeastOne(t *testing.T) { 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), &imagerequest.Transform{ Size: tt.size, diff --git a/internal/imageprocessor/imageprocessor.go b/internal/imageprocessor/imageprocessor.go index c66148f..afe216e 100644 --- a/internal/imageprocessor/imageprocessor.go +++ b/internal/imageprocessor/imageprocessor.go @@ -11,7 +11,6 @@ import ( "time" "github.com/davidbyttow/govips/v2/vips" - "sneak.berlin/go/pixa/internal/config" "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. 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. var ErrInputTooLarge = errors.New("input image dimensions exceed maximum") @@ -142,35 +137,22 @@ type ImageProcessor struct { processingWaitTimeout time.Duration } -// Params holds configuration for creating an ImageProcessor. -// Zero values use sensible defaults (MaxInputBytes defaults to DefaultMaxInputBytes). +// Params holds configuration for creating an ImageProcessor. Both fields +// must be at least 1. type Params struct { // MaxInputBytes is the maximum allowed input size in bytes. - // If <= 0, DefaultMaxInputBytes is used. MaxInputBytes int64 // MaxConcurrentProcessing is the most images processed at once. - // If <= 0, config.DefaultMaxConcurrentProcessing() is used. MaxConcurrentProcessing int } // New creates a new image processor with the given parameters. -// A zero-value Params{} uses sensible defaults. func New(params Params) *ImageProcessor { initVips() - maxInputBytes := params.MaxInputBytes - if maxInputBytes <= 0 { - maxInputBytes = DefaultMaxInputBytes - } - - maxConcurrentProcessing := params.MaxConcurrentProcessing - if maxConcurrentProcessing <= 0 { - maxConcurrentProcessing = config.DefaultMaxConcurrentProcessing() - } - return &ImageProcessor{ - maxInputBytes: maxInputBytes, - processingSemaphore: make(chan struct{}, maxConcurrentProcessing), + maxInputBytes: params.MaxInputBytes, + processingSemaphore: make(chan struct{}, params.MaxConcurrentProcessing), processingWaitTimeout: ProcessingWaitTimeout, } } diff --git a/internal/imageprocessor/imageprocessor_internal_test.go b/internal/imageprocessor/imageprocessor_internal_test.go index 90c45b9..e9b3656 100644 --- a/internal/imageprocessor/imageprocessor_internal_test.go +++ b/internal/imageprocessor/imageprocessor_internal_test.go @@ -119,7 +119,7 @@ func detectMIME(data []byte) string { func TestImageProcessor_ResizeJPEG(t *testing.T) { t.Parallel() - proc := New(Params{}) + proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1}) ctx := context.Background() input := createTestJPEG(t, 800, 600) @@ -165,7 +165,7 @@ func TestImageProcessor_ResizeJPEG(t *testing.T) { func TestImageProcessor_ConvertToPNG(t *testing.T) { t.Parallel() - proc := New(Params{}) + proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1}) ctx := context.Background() input := createTestJPEG(t, 200, 150) @@ -201,7 +201,7 @@ func processAndCheckSize( ) { t.Helper() - proc := New(Params{}) + proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1}) ctx := context.Background() input := createTestJPEG(t, inputW, inputH) @@ -239,7 +239,7 @@ func TestImageProcessor_OriginalSize(t *testing.T) { func TestImageProcessor_FitContain(t *testing.T) { t.Parallel() - proc := New(Params{}) + proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1}) ctx := context.Background() // 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) { t.Parallel() - proc := New(Params{}) + proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1}) ctx := context.Background() input := createTestPNG(t, 400, 300) @@ -330,7 +330,7 @@ func TestImageProcessor_RejectsOversizedInput(t *testing.T) { t.Run(tt.name, func(t *testing.T) { t.Parallel() - proc := New(Params{}) + proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1}) ctx := context.Background() input := createTestJPEG(t, tt.width, tt.height) @@ -356,7 +356,7 @@ func TestImageProcessor_RejectsOversizedInput(t *testing.T) { func TestImageProcessor_AcceptsMaxDimensionInput(t *testing.T) { t.Parallel() - proc := New(Params{}) + proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1}) ctx := context.Background() // Create an image at exactly MaxInputDimension - should be accepted @@ -387,7 +387,7 @@ func encodeAndCheck( ) { t.Helper() - proc := New(Params{}) + proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1}) ctx := context.Background() input := createTestJPEG(t, 200, 150) @@ -436,7 +436,7 @@ func TestImageProcessor_EncodeWebP(t *testing.T) { func TestImageProcessor_DecodeAVIF(t *testing.T) { t.Parallel() - proc := New(Params{}) + proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1}) ctx := context.Background() // Load test AVIF file @@ -478,7 +478,7 @@ func TestImageProcessor_RejectsOversizedInputData(t *testing.T) { // Create a processor with a very small byte limit const limit = 1024 - proc := New(Params{MaxInputBytes: limit}) + proc := New(Params{MaxInputBytes: limit, MaxConcurrentProcessing: 1}) ctx := context.Background() // Create a valid JPEG that exceeds the byte limit @@ -511,7 +511,7 @@ func TestImageProcessor_AcceptsInputWithinLimit(t *testing.T) { input := createTestJPEG(t, 10, 10) limit := int64(len(input)) * 10 // 10× headroom - proc := New(Params{MaxInputBytes: limit}) + proc := New(Params{MaxInputBytes: limit, MaxConcurrentProcessing: 1}) ctx := context.Background() transform := &imagerequest.Transform{ @@ -529,22 +529,6 @@ func TestImageProcessor_AcceptsInputWithinLimit(t *testing.T) { 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) { t.Parallel() @@ -558,7 +542,10 @@ func processAndDecode( ) *vips.ImageRef { t.Helper() - result, err := New(Params{}).Process( + result, err := New(Params{ + MaxInputBytes: 50 << 20, + MaxConcurrentProcessing: 1, + }).Process( context.Background(), bytes.NewReader(input), transform, ) if err != nil { diff --git a/internal/imageprocessor/jpegxl_output_internal_test.go b/internal/imageprocessor/jpegxl_output_internal_test.go index 44b3f6a..29afb32 100644 --- a/internal/imageprocessor/jpegxl_output_internal_test.go +++ b/internal/imageprocessor/jpegxl_output_internal_test.go @@ -21,7 +21,10 @@ func TestImageProcessor_EncodeJPEGXL(t *testing.T) { 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, ) if err != nil { @@ -64,7 +67,10 @@ func TestImageProcessor_JPEGXLQuality(t *testing.T) { outputBytes := make(map[int]int64) 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), &imagerequest.Transform{Format: imagerequest.FormatJXL, Quality: quality}) if err != nil { diff --git a/internal/imageprocessor/max_concurrent_processing_internal_test.go b/internal/imageprocessor/max_concurrent_processing_internal_test.go index c3d428e..fff69d0 100644 --- a/internal/imageprocessor/max_concurrent_processing_internal_test.go +++ b/internal/imageprocessor/max_concurrent_processing_internal_test.go @@ -5,7 +5,6 @@ import ( "context" "errors" "io" - "runtime" "strings" "sync" "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() - for _, limit := range []int{0, -1} { - 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}) + proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 3}) if got := cap(proc.processingSemaphore); got != 3 { 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 // than MaxConcurrentProcessing allows and holds the first ones inside // 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 ) - proc := New(Params{MaxConcurrentProcessing: limit}) + proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: limit}) input := createTestJPEG(t, 50, 50) counter := &readingCounter{} @@ -218,7 +185,7 @@ func TestProcessNeverExceedsMaxConcurrentProcessing(t *testing.T) { func TestProcessWaitsThenFailsWhenNoSlotFrees(t *testing.T) { t.Parallel() - proc := New(Params{MaxConcurrentProcessing: 1}) + proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1}) proc.processingWaitTimeout = 100 * time.Millisecond input := createTestJPEG(t, 10, 10) @@ -334,7 +301,7 @@ func TestProcessReleasesSlotOnError(t *testing.T) { func TestWaitForProcessing(t *testing.T) { t.Parallel() - proc := New(Params{MaxConcurrentProcessing: 2}) + proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 2}) gate := make(chan struct{}) entered := make(chan struct{}, 1) diff --git a/internal/imageprocessor/orig_format_internal_test.go b/internal/imageprocessor/orig_format_internal_test.go index a0ad8dc..70b97ef 100644 --- a/internal/imageprocessor/orig_format_internal_test.go +++ b/internal/imageprocessor/orig_format_internal_test.go @@ -27,7 +27,10 @@ func TestImageProcessor_OrigOfUnnamedFormatRefused(t *testing.T) { 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), &imagerequest.Transform{Format: imagerequest.FormatOriginal}, ) diff --git a/internal/imageprocessor/source_formats_internal_test.go b/internal/imageprocessor/source_formats_internal_test.go index d724976..aaf3aec 100644 --- a/internal/imageprocessor/source_formats_internal_test.go +++ b/internal/imageprocessor/source_formats_internal_test.go @@ -14,7 +14,7 @@ import ( func TestImageProcessor_LoadsEachSourceFormat(t *testing.T) { t.Parallel() - proc := New(Params{}) + proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1}) for _, format := range []imagerequest.Format{ imagerequest.FormatJPEG, imagerequest.FormatPNG, imagerequest.FormatGIF, diff --git a/internal/imgcache/cache.go b/internal/imgcache/cache.go index 08ccd9a..e1b22a7 100644 --- a/internal/imgcache/cache.go +++ b/internal/imgcache/cache.go @@ -32,23 +32,12 @@ type CacheConfig struct { NegativeTTL time.Duration // 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 - // 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 means slog.Default(). Logger *slog.Logger @@ -62,10 +51,6 @@ type Cache struct { config CacheConfig log *slog.Logger - // disabled means the disk cache is turned off entirely: lookups - // always miss, stores are no-ops, and no eviction runs. - disabled bool - // Eviction machinery. The channels are created in NewCache so // stores can signal write pressure without racing StartEviction. // evictionCancel, set by StartEviction, cancels the eviction @@ -97,7 +82,7 @@ type Cache struct { transforms prometheus.Counter // 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. reconciliationPageSize int @@ -110,20 +95,9 @@ type Cache struct { // NewCache creates a new cache instance and registers its counters on // metricsRegistry. pixad passes Prometheus' default registry, the one -// /metrics serves. +// /metrics serves. It makes no database call. func NewCache( 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) { log := config.Logger if log == nil { @@ -134,7 +108,6 @@ func newCache( db: db, config: config, log: log, - disabled: config.DisableDiskCache, evictionPressure: make(chan struct{}, 1), evictionDone: make(chan struct{}), @@ -150,7 +123,7 @@ func newCache( return nil, err } - if c.disabled { + if config.MaxBytes == 0 { return c, nil } @@ -171,20 +144,6 @@ func newCache( c.srcContent = srcContent 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 } @@ -196,7 +155,7 @@ func newCache( func (c *Cache) GetVariant( cacheKey imagerequest.VariantKey, ) (io.ReadCloser, int64, string, error) { - if c.disabled { + if c.config.MaxBytes == 0 { return nil, 0, "", ErrNotFound } @@ -214,7 +173,7 @@ func (c *Cache) StoreSource( req *imagerequest.Request, content io.Reader, ) (imagerequest.SourceKey, error) { - if c.disabled { + if c.config.MaxBytes == 0 { return "", nil } @@ -259,7 +218,7 @@ func (c *Cache) StoreSource( func (c *Cache) StoreVariant( ctx context.Context, cacheKey imagerequest.VariantKey, content io.Reader, ) error { - if c.disabled { + if c.config.MaxBytes == 0 { return nil } @@ -293,7 +252,7 @@ func (c *Cache) StoreVariant( func (c *Cache) LookupSource( ctx context.Context, req *imagerequest.Request, ) (imagerequest.SourceKey, error) { - if c.disabled { + if c.config.MaxBytes == 0 { return "", nil } @@ -345,7 +304,7 @@ func (c *Cache) StoreNegative(ctx context.Context, req *imagerequest.Request) er func (c *Cache) GetSourceContent( key imagerequest.SourceKey, ) (io.ReadCloser, int64, error) { - if c.disabled { + if c.config.MaxBytes == 0 { return nil, 0, ErrNotFound } diff --git a/internal/imgcache/cache_hit_internal_test.go b/internal/imgcache/cache_hit_internal_test.go index b2d2301..d4a3c76 100644 --- a/internal/imgcache/cache_hit_internal_test.go +++ b/internal/imgcache/cache_hit_internal_test.go @@ -22,7 +22,10 @@ func TestGetVariant_ContentTypeFromFirstBytes(t *testing.T) { t.Parallel() 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}) for _, format := range []imagerequest.Format{ diff --git a/internal/imgcache/cache_internal_test.go b/internal/imgcache/cache_internal_test.go index 56cc5a8..b37af29 100644 --- a/internal/imgcache/cache_internal_test.go +++ b/internal/imgcache/cache_internal_test.go @@ -53,6 +53,7 @@ func TestCache_CleanExpired(t *testing.T) { cache, _ := NewCache(db, CacheConfig{ StateDir: tmpDir, NegativeTTL: 1 * time.Millisecond, + MaxBytes: 1 << 30, }, prometheus.NewRegistry()) ctx := context.Background() @@ -104,6 +105,7 @@ func TestCache_StorageDirectoriesCreated(t *testing.T) { _, err := NewCache(db, CacheConfig{ StateDir: tmpDir, + MaxBytes: 1 << 30, }, prometheus.NewRegistry()) if err != nil { t.Fatalf("NewCache() error = %v", err) diff --git a/internal/imgcache/cachesize.go b/internal/imgcache/cachesize.go deleted file mode 100644 index 231a907..0000000 --- a/internal/imgcache/cachesize.go +++ /dev/null @@ -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 /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 -} diff --git a/internal/imgcache/cachesize_internal_test.go b/internal/imgcache/cachesize_internal_test.go deleted file mode 100644 index e67f953..0000000 --- a/internal/imgcache/cachesize_internal_test.go +++ /dev/null @@ -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 /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) - } -} diff --git a/internal/imgcache/concurrent_misses_internal_test.go b/internal/imgcache/concurrent_misses_internal_test.go index 7fb3e0e..226e79e 100644 --- a/internal/imgcache/concurrent_misses_internal_test.go +++ b/internal/imgcache/concurrent_misses_internal_test.go @@ -455,7 +455,7 @@ func TestService_Get_WaiterGetsProcessingTimedOut(t *testing.T) { svc, fixtures, fetcher := setupHeldFetchService(t) svc.processor = imageprocessor.New( - imageprocessor.Params{MaxConcurrentProcessing: 1}, + imageprocessor.Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1}, ) holdProcessingSlot(t, svc.processor) diff --git a/internal/imgcache/counters_internal_test.go b/internal/imgcache/counters_internal_test.go index 7886cbf..cb11aa3 100644 --- a/internal/imgcache/counters_internal_test.go +++ b/internal/imgcache/counters_internal_test.go @@ -36,7 +36,8 @@ func TestNewCache_RegistersItsCountersOnItsRegistry(t *testing.T) { 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 { t.Fatalf("NewCache() error = %v", err) } @@ -68,12 +69,14 @@ func TestNewCache_FailsWhenItsCountersCannotBeRegistered(t *testing.T) { 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 { 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) var alreadyRegistered prometheus.AlreadyRegisteredError diff --git a/internal/imgcache/eviction.go b/internal/imgcache/eviction.go index 6f3cc9f..2b1feb0 100644 --- a/internal/imgcache/eviction.go +++ b/internal/imgcache/eviction.go @@ -52,18 +52,25 @@ const sqliteTimestampLayout = "2006-01-02 15:04:05" const tempFilePrefix = ".tmp-" // UsageBytes returns the total number of bytes of cache content -// tracked in the database (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. +// tracked in the database, as the package function UsageBytes does, or 0 +// when the disk cache is off. func (c *Cache) UsageBytes(ctx context.Context) (int64, error) { - if c.disabled { + if c.config.MaxBytes == 0 { 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 - err := c.db.QueryRowContext(ctx, + err := db.QueryRowContext(ctx, `SELECT total_size_bytes FROM cache_usage WHERE id = 1`, ).Scan(&total) if err != nil { @@ -86,9 +93,9 @@ type evictionCandidate struct { // EvictToLimit evicts least-recently-used cache entries until total // tracked usage is at or below the configured MaxBytes limit. It is a -// no-op when the cache is disabled or no limit is configured. +// no-op when the cache is disabled. func (c *Cache) EvictToLimit(ctx context.Context) error { - if c.disabled || c.config.MaxBytes <= 0 { + if c.config.MaxBytes == 0 { 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 // when a wakeup is already pending. func (c *Cache) notifyWritePressure() { - if c.disabled || c.config.MaxBytes <= 0 { + if c.config.MaxBytes == 0 { 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 // ctx's error. func (c *Cache) reconcileAccounting(ctx context.Context) error { - if c.disabled { + if c.config.MaxBytes == 0 { return nil } diff --git a/internal/imgcache/eviction_internal_test.go b/internal/imgcache/eviction_internal_test.go index 99ba09e..fec51d9 100644 --- a/internal/imgcache/eviction_internal_test.go +++ b/internal/imgcache/eviction_internal_test.go @@ -66,15 +66,11 @@ func newEvictionTestCache(t *testing.T, maxBytes int64) (*Cache, string) { tmpDir := t.TempDir() db := evictionTestDB(t) - // maxBytes zero mirrors the production mapping of - // cache_max_bytes: 0 (handlers sets DisableDiskCache); at the - // CacheConfig layer itself a zero MaxBytes means "no limit" for - // backwards compatibility with existing fixtures. + // maxBytes zero turns the disk cache off, as cache_max_bytes: 0 does. cache, err := NewCache(db, CacheConfig{ - StateDir: tmpDir, - NegativeTTL: 5 * time.Minute, - MaxBytes: maxBytes, - DisableDiskCache: maxBytes == 0, + StateDir: tmpDir, + NegativeTTL: 5 * time.Minute, + MaxBytes: maxBytes, }, prometheus.NewRegistry()) if err != nil { t.Fatalf("failed to create cache: %v", err) diff --git a/internal/imgcache/max_concurrent_processing_internal_test.go b/internal/imgcache/max_concurrent_processing_internal_test.go index 2ff8372..1de6011 100644 --- a/internal/imgcache/max_concurrent_processing_internal_test.go +++ b/internal/imgcache/max_concurrent_processing_internal_test.go @@ -66,7 +66,7 @@ func TestService_Get_WaitsForSlotBeforeReadingCachedSource(t *testing.T) { svc, fixtures := SetupTestService(t) svc.processor = imageprocessor.New( - imageprocessor.Params{MaxConcurrentProcessing: 1}, + imageprocessor.Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1}, ) // A first request caches the photo as a source. diff --git a/internal/imgcache/negative_cache_internal_test.go b/internal/imgcache/negative_cache_internal_test.go index 3aa7a10..709f090 100644 --- a/internal/imgcache/negative_cache_internal_test.go +++ b/internal/imgcache/negative_cache_internal_test.go @@ -20,6 +20,7 @@ func TestNegativeCache_StoreAndCheck(t *testing.T) { cache, err := NewCache(db, CacheConfig{ StateDir: dir, NegativeTTL: 5 * time.Minute, + MaxBytes: 1 << 30, }, prometheus.NewRegistry()) if err != nil { t.Fatal(err) @@ -69,6 +70,7 @@ func TestNegativeCache_Expired(t *testing.T) { cache, err := NewCache(db, CacheConfig{ StateDir: dir, NegativeTTL: 1 * time.Millisecond, // very short TTL + MaxBytes: 1 << 30, }, prometheus.NewRegistry()) if err != nil { t.Fatal(err) @@ -177,11 +179,11 @@ func TestEvictionLoopRemovesExpiredNegativeCacheRows(t *testing.T) { t.Parallel() tests := []struct { - name string - disableDiskCache bool + name string + maxBytes int64 }{ - {name: "disk cache enabled", disableDiskCache: false}, - {name: "disk cache disabled", disableDiskCache: true}, + {name: "disk cache enabled", maxBytes: 1 << 30}, + {name: "disk cache disabled", maxBytes: 0}, } for _, tt := range tests { @@ -189,9 +191,9 @@ func TestEvictionLoopRemovesExpiredNegativeCacheRows(t *testing.T) { t.Parallel() cache, err := NewCache(evictionTestDB(t), CacheConfig{ - StateDir: t.TempDir(), - NegativeTTL: time.Second, - DisableDiskCache: tt.disableDiskCache, + StateDir: t.TempDir(), + NegativeTTL: time.Second, + MaxBytes: tt.maxBytes, }, prometheus.NewRegistry()) if err != nil { 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) { 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()) if err != nil { t.Fatalf("failed to create cache: %v", err) @@ -321,7 +324,8 @@ func TestCleanExpiredDeletesInBatches(t *testing.T) { func TestCleanExpiredRemovesMoreThanOneBatch(t *testing.T) { 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()) if err != nil { t.Fatalf("failed to create cache: %v", err) @@ -352,7 +356,8 @@ func TestCleanExpiredRemovesMoreThanOneBatch(t *testing.T) { func TestCleanExpiredStopsOnceContextEnds(t *testing.T) { 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()) if err != nil { t.Fatalf("failed to create cache: %v", err) diff --git a/internal/imgcache/service.go b/internal/imgcache/service.go index 7d03c0a..ece9311 100644 --- a/internal/imgcache/service.go +++ b/internal/imgcache/service.go @@ -43,18 +43,20 @@ type Service struct { type ServiceConfig struct { // Cache is the cache instance Cache *Cache - // FetcherConfig configures the upstream fetcher built when Fetcher is - // not set. Its AllowHTTP and MaxResponseSize are used either way. - FetcherConfig *httpfetcher.Config - // Fetcher is an optional custom fetcher (for testing) + // Fetcher fetches source images from upstream hosts Fetcher httpfetcher.Fetcher + // Processor converts and resizes the images + Processor *imageprocessor.ImageProcessor // SigningKey is the HMAC signing key (empty disables signing) SigningKey string // Allowlist is the list of hosts that don't require signatures Allowlist []string - // MaxConcurrentProcessing is the most images processed at once; zero - // uses the image processor's default, half the CPUs - MaxConcurrentProcessing int + // AllowHTTP makes the service fetch sources over http instead of + // https (testing only), so the fetcher must allow http too + 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 *slog.Logger } @@ -75,32 +77,11 @@ func NewService(cfg *ServiceConfig) (*Service, error) { 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 if log == nil { 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 // without it. err := imageprocessor.CheckJPEGXLSupport() @@ -108,21 +89,15 @@ func NewService(cfg *ServiceConfig) (*Service, error) { return nil, err } - maxResponseSize := fetcherCfg.MaxResponseSize - processor := imageprocessor.New(imageprocessor.Params{ - MaxInputBytes: maxResponseSize, - MaxConcurrentProcessing: cfg.MaxConcurrentProcessing, - }) - return &Service{ cache: cfg.Cache, - fetcher: fetcher, - processor: processor, - signer: signer, + fetcher: cfg.Fetcher, + processor: cfg.Processor, + signer: signature.New(cfg.SigningKey), allowlist: allowlist.New(cfg.Allowlist), log: log, - allowHTTP: allowHTTP, - maxResponseSize: maxResponseSize, + allowHTTP: cfg.AllowHTTP, + maxResponseSize: cfg.MaxResponseSize, }, nil } diff --git a/internal/imgcache/testutil_internal_test.go b/internal/imgcache/testutil_internal_test.go index bd5ad8b..31c2c79 100644 --- a/internal/imgcache/testutil_internal_test.go +++ b/internal/imgcache/testutil_internal_test.go @@ -17,6 +17,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "sneak.berlin/go/pixa/internal/database" "sneak.berlin/go/pixa/internal/httpfetcher" + "sneak.berlin/go/pixa/internal/imageprocessor" ) // Shared test data literals, extracted as constants for goconst. @@ -185,16 +186,22 @@ func SetupTestService( cache, err := NewCache(db, CacheConfig{ StateDir: tmpDir, NegativeTTL: 5 * time.Minute, + MaxBytes: 1 << 30, }, prometheus.NewRegistry()) if err != nil { t.Fatalf("failed to create cache: %v", err) } svc, err := NewService(&ServiceConfig{ - Cache: cache, - Fetcher: httpfetcher.NewMock(mockFS), - SigningKey: cfg.signingKey, - Allowlist: cfg.allowlist, + Cache: cache, + Fetcher: httpfetcher.NewMock(mockFS), + Processor: imageprocessor.New(imageprocessor.Params{ + MaxInputBytes: 50 << 20, + MaxConcurrentProcessing: 2, + }), + SigningKey: cfg.signingKey, + Allowlist: cfg.allowlist, + MaxResponseSize: 50 << 20, }) if err != nil { t.Fatalf("failed to create service: %v", err) diff --git a/internal/providers/cache_max_bytes_internal_test.go b/internal/providers/cache_max_bytes_internal_test.go new file mode 100644 index 0000000..8b0fc03 --- /dev/null +++ b/internal/providers/cache_max_bytes_internal_test.go @@ -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 /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) + } +} diff --git a/internal/providers/fetcher_internal_test.go b/internal/providers/fetcher_internal_test.go new file mode 100644 index 0000000..75f2b02 --- /dev/null +++ b/internal/providers/fetcher_internal_test.go @@ -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) + } +} diff --git a/internal/providers/providers.go b/internal/providers/providers.go new file mode 100644 index 0000000..b098b4e --- /dev/null +++ b/internal/providers/providers.go @@ -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 /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) +} diff --git a/internal/server/image_proxy_internal_test.go b/internal/server/image_proxy_internal_test.go index ace59d5..86d598e 100644 --- a/internal/server/image_proxy_internal_test.go +++ b/internal/server/image_proxy_internal_test.go @@ -32,6 +32,7 @@ import ( "sneak.berlin/go/pixa/internal/imagerequest" "sneak.berlin/go/pixa/internal/logger" "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 @@ -138,22 +139,21 @@ func startImageProxy( AllowHTTP: true, // A limit of its own, so the cache does not size itself from the // host's free disk space. - CacheMaxBytes: 64 << 20, - CacheMaxBytesExplicit: true, - UpstreamMaxResponseSize: config.DefaultUpstreamMaxResponseSize, - DownstreamTimeout: config.DefaultDownstreamTimeout, + CacheMaxBytes: 64 << 20, + CacheMaxBytesExplicit: true, + UpstreamFetchTimeout: config.DefaultUpstreamFetchTimeout, + UpstreamMaxResponseSize: config.DefaultUpstreamMaxResponseSize, + UpstreamConnectionsPerHost: config.DefaultUpstreamConnectionsPerHost, + UpstreamConnections: config.DefaultUpstreamConnections, + MaxConcurrentProcessing: 2, + DownstreamTimeout: config.DefaultDownstreamTimeout, } - fetcherCfg := httpfetcher.DefaultConfig() - fetcherCfg.AllowHTTP = true - fetcherCfg.DialContext = func( - ctx context.Context, network, _ string, - ) (net.Conn, error) { + dialUpstream := func(ctx context.Context, network, _ string) (net.Conn, error) { var dialer net.Dialer return dialer.DialContext(ctx, network, upstream.Listener.Addr().String()) } - fetcher := httpfetcher.New(fetcherCfg) var ( h *handlers.Handlers @@ -170,9 +170,14 @@ func startImageProxy( healthcheck.New, handlers.New, middleware.New, - func() httpfetcher.Fetcher { return fetcher }, 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), ) app.RequireStart() @@ -188,10 +193,10 @@ func startImageProxy( // startImageProxyWithBlockedNetworks starts the components pixad's fx app // builds, from a config with a fresh state directory and blockedNetworks as -// blocked_networks. Unlike startImageProxy it provides no upstream fetcher, so -// the handlers build the one pixad uses, which checks every address it -// connects to against the built-in blocked ranges and blocked_networks. It -// returns the server with its routes. +// blocked_networks. Unlike startImageProxy it keeps pixad's dial function, +// which checks every address the fetcher connects to against the built-in +// blocked ranges and blocked_networks. It returns the server with its +// routes. func startImageProxyWithBlockedNetworks( t *testing.T, blockedNetworks []netip.Prefix, ) *Server { @@ -207,8 +212,9 @@ func startImageProxyWithBlockedNetworks( // fails within this time at the latest. UpstreamFetchTimeout: 2 * time.Second, // With no connection slots, the fetch would fail before connecting. - UpstreamConnections: config.DefaultUpstreamConnections, - DownstreamTimeout: config.DefaultDownstreamTimeout, + UpstreamConnectionsPerHost: config.DefaultUpstreamConnectionsPerHost, + UpstreamConnections: config.DefaultUpstreamConnections, + DownstreamTimeout: config.DefaultDownstreamTimeout, } var ( @@ -227,6 +233,7 @@ func startImageProxyWithBlockedNetworks( middleware.New, func() prometheus.Registerer { return prometheus.NewRegistry() }, ), + fx.Provide(providers.Constructors()...), fx.Populate(&h, &mw), ) app.RequireStart() diff --git a/internal/server/login_rate_limit_internal_test.go b/internal/server/login_rate_limit_internal_test.go index e55fb79..ebae939 100644 --- a/internal/server/login_rate_limit_internal_test.go +++ b/internal/server/login_rate_limit_internal_test.go @@ -12,16 +12,7 @@ import ( "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/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. @@ -61,45 +52,7 @@ func newTestServer(t *testing.T) *Server { DownstreamTimeout: config.DefaultDownstreamTimeout, } - 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) - } - - 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 + return startPixa(t, cfg) } // clientRequest builds a request for / arriving from remoteAddr, carrying diff --git a/internal/server/logout_internal_test.go b/internal/server/logout_internal_test.go index d518743..941268f 100644 --- a/internal/server/logout_internal_test.go +++ b/internal/server/logout_internal_test.go @@ -20,6 +20,7 @@ import ( "sneak.berlin/go/pixa/internal/healthcheck" "sneak.berlin/go/pixa/internal/logger" "sneak.berlin/go/pixa/internal/middleware" + "sneak.berlin/go/pixa/internal/providers" "sneak.berlin/go/pixa/internal/session" ) @@ -29,7 +30,7 @@ var logoutFormPattern = regexp.MustCompile( `(?s)action="/logout".*?name="gorilla\.csrf\.Token" value="([^"]+)"`) // 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 { t.Helper() @@ -49,6 +50,7 @@ func startPixa(t *testing.T, cfg *config.Config) *Server { middleware.New, func() prometheus.Registerer { return prometheus.NewRegistry() }, ), + fx.Provide(providers.Constructors()...), fx.Populate(&h, &mw), ) app.RequireStart()