From 6d9631a8fccd8a17524f1a5903887ed2111ffa7f Mon Sep 17 00:00:00 2001 From: sneak Date: Mon, 26 Oct 2026 10:31:44 +0000 Subject: [PATCH] Build the image pipeline in pixad's fx graph (closes #313) The new package internal/providers builds the cache, the fetcher, the processor, the image service, the session manager and the encrypted URL generator from the config. pixad's fx graph provides them with providers.Constructors(), and handlers.New receives them built. The server tests build the same graph, with only the fetcher's dial function replaced. database.New opens the database, so these parts can read it while fx builds them; the cache's stop hook still runs before the database's. Each setting's default is declared in internal/config, and internal/providers works out the default cache_max_bytes. The cache's MaxBytes is the one value for the disk cache limit, and 0 turns the disk cache off. The fetcher and the processor take explicit values. Model: opus-5-5 --- README.md | 4 +- TODO.md | 24 ++ internal/app/app.go | 2 + .../config/cache_max_bytes_internal_test.go | 34 +- internal/config/config.go | 51 ++- internal/database/database.go | 22 +- internal/handlers/cache_hit_internal_test.go | 13 +- .../handlers/cache_max_bytes_internal_test.go | 151 --------- internal/handlers/fetcher_internal_test.go | 63 ---- .../handlers/first_bytes_internal_test.go | 17 +- internal/handlers/handlers.go | 147 +------- internal/handlers/handlers_internal_test.go | 15 +- .../image_cache_control_internal_test.go | 11 +- .../handlers/image_errors_internal_test.go | 22 +- .../referer_blocklist_internal_test.go | 15 +- .../blocked_networks_internal_test.go | 16 +- .../httpfetcher/dial_context_internal_test.go | 20 +- internal/httpfetcher/fetch_internal_test.go | 39 ++- internal/httpfetcher/httpfetcher.go | 65 ++-- .../httpfetcher/httpfetcher_internal_test.go | 53 +-- internal/httpfetcher/jpegxl_internal_test.go | 18 +- .../max_connections_internal_test.go | 41 ++- .../avif_too_large_internal_test.go | 5 +- .../computed_side_internal_test.go | 10 +- .../imageprocessor/deadline_internal_test.go | 2 +- .../empty_format_internal_test.go | 5 +- .../encoder_settings_internal_test.go | 5 +- .../imageprocessor/fit_cover_internal_test.go | 15 +- .../fit_outside_internal_test.go | 10 +- internal/imageprocessor/imageprocessor.go | 26 +- .../imageprocessor_internal_test.go | 43 +-- .../jpegxl_output_internal_test.go | 10 +- ...max_concurrent_processing_internal_test.go | 43 +-- .../orig_format_internal_test.go | 5 +- .../source_formats_internal_test.go | 2 +- internal/imgcache/cache.go | 65 +--- internal/imgcache/cache_hit_internal_test.go | 5 +- internal/imgcache/cache_internal_test.go | 2 + internal/imgcache/cachesize.go | 85 ----- internal/imgcache/cachesize_internal_test.go | 189 ----------- .../concurrent_misses_internal_test.go | 2 +- internal/imgcache/counters_internal_test.go | 9 +- internal/imgcache/eviction.go | 27 +- internal/imgcache/eviction_internal_test.go | 12 +- ...max_concurrent_processing_internal_test.go | 2 +- .../imgcache/negative_cache_internal_test.go | 25 +- internal/imgcache/service.go | 53 +-- internal/imgcache/testutil_internal_test.go | 15 +- .../cache_max_bytes_internal_test.go | 315 ++++++++++++++++++ internal/providers/fetcher_internal_test.go | 103 ++++++ internal/providers/providers.go | 256 ++++++++++++++ internal/server/image_proxy_internal_test.go | 41 ++- .../server/login_rate_limit_internal_test.go | 49 +-- internal/server/logout_internal_test.go | 4 +- 54 files changed, 1181 insertions(+), 1102 deletions(-) delete mode 100644 internal/handlers/cache_max_bytes_internal_test.go delete mode 100644 internal/handlers/fetcher_internal_test.go delete mode 100644 internal/imgcache/cachesize.go delete mode 100644 internal/imgcache/cachesize_internal_test.go create mode 100644 internal/providers/cache_max_bytes_internal_test.go create mode 100644 internal/providers/fetcher_internal_test.go create mode 100644 internal/providers/providers.go 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()