3 Commits
Author SHA1 Message Date
clawbot a14c5a1ba0 Read a trusted_proxies entry in the IPv4-mapped form as its IPv4 range (closes #303)
check / check (push) Canceled after 0s
The client address is matched against trusted_proxies as the IPv4 address
an IPv4-mapped address holds, so an entry such as ::ffff:10.0.0.0/104
trusted nothing. parseCIDRList, which reads both trusted_proxies and
blocked_networks, now keeps such an entry as the IPv4 range it covers and
refuses one shorter than /96 naming the setting it reads.
parseBlockedNetworks, which did this for blocked_networks alone, is gone.

Model: opus-5-5
2026-10-19 03:02:23 +00:00
clawbot 2c0fc688f9 Test that a trusted_proxies entry in the IPv4-mapped form is read as its IPv4 range
These tests fail on next: ::ffff:10.0.0.0/104 is kept as written and so
never matches, and ::ffff:0:0/95 loads without an error. The fix follows
in the next commit.

Model: opus-5-5
2026-10-19 03:02:01 +00:00
clawbot 9eacfa6406 Space reconciliation passes by ten times the last pass's time (closes #250)
check / check (push) Canceled after 0s
The reconciliation pass reads a row and stats a file for every cached
file, and it ran on every eviction tick, so on a large cache a pass
longer than the tick was followed by the next one at once. It still runs
at startup and is considered on each tick, but it now starts no sooner
than ten times as long after the last pass ended as that pass took, so
it takes at most about a tenth of the time however large the cache.
Eviction and the removal of expired negative cache rows still run on
every tick. The comments that bound the drift, and README.md, give the
new bound. No existing test changed.

Model: opus-5-5
2026-10-19 05:00:48 +02:00
6 changed files with 215 additions and 65 deletions

No files matched your search

+21 -11
View File
@@ -199,7 +199,11 @@ into place, so a crash leaves no file under its own name with part of its data.
When pixa starts, and every five minutes after, it removes each cached source or When pixa starts, and every five minutes after, it removes each cached source or
transformed image whose file is empty, or not the size the database records for transformed image whose file is empty, or not the size the database records for
it, with a warning naming the file; the next request for it fetches the source it, with a warning naming the file; the next request for it fetches the source
or converts the image again. or converts the image again. Looking for those files reads the database once for
every cached file, so pixa skips each five-minute look that would start less
than ten times as long after the last look ended as that look took. After a look
that took one minute, the next one starts 10 to 15 minutes after that look
ended.
pixa's target is 1-5k r/s, which has not been measured at that rate (see Load pixa's target is 1-5k r/s, which has not been measured at that rate (see Load
Test). Toward it, pixa keeps in memory the content types of the 10,000 Test). Toward it, pixa keeps in memory the content types of the 10,000
@@ -709,16 +713,22 @@ Key settings in more detail:
omitted key defaults to the RFC 1918 private ranges (`10.0.0.0/8`, omitted key defaults to the RFC 1918 private ranges (`10.0.0.0/8`,
`172.16.0.0/12`, `192.168.0.0/16`), since pixa is deployed behind a proxy on a `172.16.0.0/12`, `192.168.0.0/16`), since pixa is deployed behind a proxy on a
private network; an explicitly empty list (`[]`) trusts no one, and an private network; an explicitly empty list (`[]`) trusts no one, and an
explicit list replaces the default. An invalid CIDR aborts startup. Set this explicit list replaces the default. An invalid CIDR aborts startup. An IPv6
to the address pixa sees for requests that come through your proxy, such as address in the IPv4-mapped form (`::ffff:a.b.c.d`) is matched only as the IPv4
`172.17.0.1/32`, when the defaults do not cover it, or to trust nothing else address it holds, so an IPv6 range such as `::/0` trusts neither it nor a
(see the login limit and `image_requests_per_minute` under Routes). For a plain IPv4 address. An entry written in that form is read as the IPv4 range it
proxy on the Docker host that connects to pixa over `127.0.0.1`, that address covers, so `::ffff:10.0.0.0/104` trusts what `10.0.0.0/8` trusts; such an
is the gateway of the container's Docker network (`172.17.0.1` on the default entry shorter than `/96` covers more than IPv4 addresses and aborts startup.
bridge), not the proxy's own address; a proxy that connects through another of Set this to the address pixa sees for requests that come through your proxy,
the host's addresses is seen with that address. To be sure which address it such as `172.17.0.1/32`, when the defaults do not cover it, or to trust
is, set this to `[]` (or `PIXA_TRUSTED_PROXIES` to empty), send a request nothing else (see the login limit and `image_requests_per_minute` under
through the proxy, and read `remoteIP` in pixa's request log line for it Routes). For a proxy on the Docker host that connects to pixa over
`127.0.0.1`, that address is the gateway of the container's Docker network
(`172.17.0.1` on the default bridge), not the proxy's own address; a proxy
that connects through another of the host's addresses is seen with that
address. To be sure which address it is, set this to `[]` (or
`PIXA_TRUSTED_PROXIES` to empty), send a request through the proxy, and read
`remoteIP` in pixa's request log line for it
- `upstream_fetch_timeout` — time allowed for one fetch from an upstream host, - `upstream_fetch_timeout` — time allowed for one fetch from an upstream host,
as a duration such as `30s` (the default) or `2m` as a duration such as `30s` (the default) or `2m`
- `upstream_max_response_size` — largest upstream response accepted, in bytes; - `upstream_max_response_size` — largest upstream response accepted, in bytes;
+7
View File
@@ -30,6 +30,13 @@ P2: security: per-origin rate limiting
# Completed Steps # Completed Steps
- 2026-10-19 a `trusted_proxies` entry in the IPv4-mapped form trusted nothing;
it is now read as the IPv4 range it covers by the same code that reads
`blocked_networks`, and such an entry shorter than `/96` aborts startup
(closes #303).
- 2026-10-18 the next reconciliation pass waits at least ten times as long as
the last one took, so passes on a large cache no longer run back to back
(closes #250).
- 2026-10-18 a `blocked_networks` entry in the IPv4-mapped form refused nothing; - 2026-10-18 a `blocked_networks` entry in the IPv4-mapped form refused nothing;
`::ffff:10.0.0.0/104` is now read as `10.0.0.0/8`, and such an entry shorter `::ffff:10.0.0.0/104` is now read as `10.0.0.0/8`, and such an entry shorter
than `/96` aborts startup (closes #301). than `/96` aborts startup (closes #301).
+20 -32
View File
@@ -272,7 +272,7 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
return nil, err return nil, err
} }
blockedNetworks, err := parseBlockedNetworks(sc) blockedNetworks, err := parseCIDRList(sc, keyBlockedNetworks)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -1249,6 +1249,13 @@ func defaultTrustedProxies() []netip.Prefix {
// wrong type, an empty entry, a non-string entry, or an unparseable CIDR // wrong type, an empty entry, a non-string entry, or an unparseable CIDR
// aborts startup naming the key and the offending value; the default // aborts startup naming the key and the offending value; the default
// (an empty list) applies only to an omitted key. // (an empty list) applies only to an omitted key.
//
// An entry written in the IPv4-mapped form (::ffff:a.b.c.d) is kept as the
// IPv4 range it covers: ::ffff:10.0.0.0/104 is kept as 10.0.0.0/8. Both
// blocked_networks and trusted_proxies are matched against an IPv4-mapped
// address as the IPv4 address it holds, so the entry as written would match
// nothing. Such an entry shorter than /96 covers more than the IPv4
// addresses and aborts startup.
func parseCIDRList(sc *smartconfig.Config, key string) ([]netip.Prefix, error) { func parseCIDRList(sc *smartconfig.Config, key string) ([]netip.Prefix, error) {
raw, ok := lookupValue(sc, key) raw, ok := lookupValue(sc, key)
if !ok { if !ok {
@@ -1264,6 +1271,10 @@ func parseCIDRList(sc *smartconfig.Config, key string) ([]netip.Prefix, error) {
return nil, err return nil, err
} }
// An IPv4-mapped address is the 96 bits of ::ffff:0:0/96 followed by
// the 32 bits of the IPv4 address it holds.
const mappedPrefixBits = 96
prefixes := make([]netip.Prefix, 0, len(entries)) prefixes := make([]netip.Prefix, 0, len(entries))
for _, entry := range entries { for _, entry := range entries {
@@ -1273,40 +1284,17 @@ func parseCIDRList(sc *smartconfig.Config, key string) ([]netip.Prefix, error) {
settingName(key), entry, errNotAValidCIDR) settingName(key), entry, errNotAValidCIDR)
} }
prefixes = append(prefixes, prefix) if prefix.Addr().Is4In6() && prefix.Bits() < mappedPrefixBits {
}
return prefixes, nil
}
// parseBlockedNetworks parses blocked_networks with parseCIDRList, and keeps
// an entry written in the IPv4-mapped form (::ffff:a.b.c.d) as the IPv4
// range it covers: ::ffff:10.0.0.0/104 is kept as 10.0.0.0/8. The fetcher
// matches an IPv4-mapped address as the IPv4 address it holds, so the entry
// as written would match nothing. Such an entry shorter than /96 covers more
// than the IPv4 addresses and aborts startup.
func parseBlockedNetworks(sc *smartconfig.Config) ([]netip.Prefix, error) {
prefixes, err := parseCIDRList(sc, keyBlockedNetworks)
if err != nil {
return nil, err
}
// An IPv4-mapped address is the 96 bits of ::ffff:0:0/96 followed by
// the 32 bits of the IPv4 address it holds.
const mappedPrefixBits = 96
for i, prefix := range prefixes {
if !prefix.Addr().Is4In6() {
continue
}
if prefix.Bits() < mappedPrefixBits {
return nil, fmt.Errorf("%s: value %q is %w", return nil, fmt.Errorf("%s: value %q is %w",
settingName(keyBlockedNetworks), prefix, errMappedTooShort) settingName(key), entry, errMappedTooShort)
} }
prefixes[i] = netip.PrefixFrom(prefix.Addr().Unmap(), if prefix.Addr().Is4In6() {
prefix.Bits()-mappedPrefixBits) prefix = netip.PrefixFrom(prefix.Addr().Unmap(),
prefix.Bits()-mappedPrefixBits)
}
prefixes = append(prefixes, prefix)
} }
return prefixes, nil return prefixes, nil
@@ -83,3 +83,41 @@ func TestTrustedProxiesInvalidAbortsStartup(t *testing.T) {
}, },
}) })
} }
// TestTrustedProxiesMappedEntryIsIPv4Range checks that an entry written in
// the IPv4-mapped form (::ffff:a.b.c.d) is kept as the IPv4 range it covers,
// since the client address is matched as the IPv4 address it holds. /96, the
// shortest such entry, covers every IPv4 address.
func TestTrustedProxiesMappedEntryIsIPv4Range(t *testing.T) {
t.Parallel()
c, err := configFromYAML(t, signingKeyLine+
`trusted_proxies: ["::ffff:10.0.0.0/104", "::ffff:0:0/96"]`+"\n")
if err != nil {
t.Fatalf("IPv4-mapped trusted_proxies entries should load: %v", err)
}
got := make([]string, len(c.TrustedProxies))
for i, p := range c.TrustedProxies {
got[i] = p.String()
}
if joined := strings.Join(got, ","); joined != "10.0.0.0/8,0.0.0.0/0" {
t.Errorf("TrustedProxies = %v, want 10.0.0.0/8 and 0.0.0.0/0", got)
}
}
// TestTrustedProxiesMappedEntryShorterThan96AbortsStartup checks that an
// entry in the IPv4-mapped form shorter than /96, which covers more than the
// IPv4 addresses, aborts startup with an error naming the key and the value.
func TestTrustedProxiesMappedEntryShorterThan96AbortsStartup(t *testing.T) {
t.Parallel()
runAbortCases(t, []abortCase{
{
name: "mapped-95",
yaml: signingKeyLine + `trusted_proxies: ["::ffff:0:0/95"]` + "\n",
wantErrSubstrings: []string{keyTrustedProxies, "::ffff:0:0/95"},
},
})
}
+46 -22
View File
@@ -18,6 +18,14 @@ import (
// write-pressure wakeups triggered by stores. // write-pressure wakeups triggered by stores.
const DefaultEvictionInterval = 5 * time.Minute const DefaultEvictionInterval = 5 * time.Minute
// reconciliationWaitFactor is how many times as long as a
// reconciliation pass took the eviction loop waits at least, after that
// pass ends, before it starts the next one. A pass reads a row and stats a
// file for every cached file, so its time grows with the cache; waiting
// ten times as long keeps reconciliation to at most a tenth of the time,
// however large the cache is.
const reconciliationWaitFactor = 10
// evictionBatchSize is how many LRU candidates of each class (variants // evictionBatchSize is how many LRU candidates of each class (variants
// and source blobs) one eviction pass fetches from the database. // and source blobs) one eviction pass fetches from the database.
const evictionBatchSize = 100 const evictionBatchSize = 100
@@ -438,10 +446,12 @@ func (c *Cache) notifyWritePressure() {
} }
// StartEviction launches the background eviction goroutine, which // StartEviction launches the background eviction goroutine, which
// reconciles the database accounting with the cache directories and
// removes the expired negative cache rows at startup and again on every // removes the expired negative cache rows at startup and again on every
// periodic tick thereafter, and evicts to the configured limit on the // periodic tick thereafter, reconciles the database accounting with the
// given periodic interval and on write-pressure notifications. On a // cache directories at startup and again on the first periodic tick
// that comes at least ten times as long after a reconciliation pass
// ended as that pass took, and evicts to the configured limit on the given
// periodic interval and on write-pressure notifications. On a
// disabled cache it only removes the expired negative cache rows, as // disabled cache it only removes the expired negative cache rows, as
// the negative cache is used either way. It is a no-op when already // the negative cache is used either way. It is a no-op when already
// started. The goroutine outlives the caller, so it runs with its own // started. The goroutine outlives the caller, so it runs with its own
@@ -482,7 +492,7 @@ func (c *Cache) StopEviction(ctx context.Context) error {
func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) { func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
defer close(c.evictionDone) defer close(c.evictionDone)
c.runReconciliationPass(ctx) nextReconciliation := c.runReconciliationPass(ctx)
c.runCleanExpired(ctx) c.runCleanExpired(ctx)
c.runEvictionPass(ctx) c.runEvictionPass(ctx)
@@ -494,16 +504,20 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
case <-ctx.Done(): case <-ctx.Done():
return return
case <-ticker.C: case <-ticker.C:
// Reconciliation walks the cache directories, so it only // Reconciliation walks the cache directories and reads a
// runs on the periodic ticker rather than on every // row for every file, so it runs only on the periodic
// write-pressure wakeup, keeping it off the per-store hot // ticker rather than on every write-pressure wakeup, and
// path. Reusing the eviction interval itself (rather than a // only on the first tick at or after nextReconciliation,
// separate, longer one) is a deliberate choice: it is the // which is ten times as long after the last pass ended as
// simplest option that still bounds how long a store's // that pass took. Passes then never run back to back,
// best-effort accounting insert can stay silently // however long one takes. A file whose best-effort
// unaccounted for to one interval, on a process that is // accounting insert failed in a store stays unaccounted for
// already running this loop regardless. // until the next pass, up to about one eviction interval
c.runReconciliationPass(ctx) // plus ten times the last pass's time.
if !time.Now().Before(nextReconciliation) {
nextReconciliation = c.runReconciliationPass(ctx)
}
c.runCleanExpired(ctx) c.runCleanExpired(ctx)
case <-c.evictionPressure: case <-c.evictionPressure:
} }
@@ -527,17 +541,25 @@ func (c *Cache) runEvictionPass(ctx context.Context) {
} }
// runReconciliationPass runs one reconciliation pass, logging failures // runReconciliationPass runs one reconciliation pass, logging failures
// instead of propagating them (the loop must keep running). It does // instead of propagating them (the loop must keep running), and returns
// nothing once ctx is cancelled. // the earliest time the next pass may start: reconciliationWaitFactor
func (c *Cache) runReconciliationPass(ctx context.Context) { // times as long after this pass ended as it took. It does nothing once
// ctx is cancelled.
func (c *Cache) runReconciliationPass(ctx context.Context) time.Time {
if ctx.Err() != nil { if ctx.Err() != nil {
return return time.Now()
} }
started := time.Now()
err := c.reconcileAccounting(ctx) err := c.reconcileAccounting(ctx)
if err != nil { if err != nil {
c.log.Warn("cache accounting reconciliation failed", "error", err) c.log.Warn("cache accounting reconciliation failed", "error", err)
} }
ended := time.Now()
return ended.Add(reconciliationWaitFactor * ended.Sub(started))
} }
// runCleanExpired runs CleanExpired, logging a failure instead of // runCleanExpired runs CleanExpired, logging a failure instead of
@@ -556,7 +578,8 @@ func (c *Cache) runCleanExpired(ctx context.Context) {
// reconcileAccounting synchronizes the database size accounting with // reconcileAccounting synchronizes the database size accounting with
// the actual contents of the cache directories. It runs at startup and // the actual contents of the cache directories. It runs at startup and
// again on every periodic eviction tick thereafter, off the request // then on the first periodic eviction tick that comes at least ten times
// as long after the last pass ended as that pass took, off the request
// hot path: it adopts variant files that predate the accounting table // hot path: it adopts variant files that predate the accounting table
// (or whose accounting insert failed, e.g. StoreVariant's best-effort // (or whose accounting insert failed, e.g. StoreVariant's best-effort
// insert under transient DB contention), drops accounting rows whose // insert under transient DB contention), drops accounting rows whose
@@ -566,9 +589,10 @@ func (c *Cache) runCleanExpired(ctx context.Context) {
// files left behind by crashed writes, and last checks the total cache // files left behind by crashed writes, and last checks the total cache
// usage against the tables. It reads the tables a page at a time. // usage against the tables. It reads the tables a page at a time.
// Running it periodically, not just once, bounds how long such drift can // Running it periodically, not just once, bounds how long such drift can
// accumulate unaccounted for on a long-running process to one eviction // accumulate unaccounted for on a long-running process to about one
// interval. Once ctx is cancelled, it stops at the next file or row and // eviction interval plus ten times the last pass's time.
// returns ctx's error. // 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 { func (c *Cache) reconcileAccounting(ctx context.Context) error {
if c.disabled { if c.disabled {
return nil return nil
@@ -918,6 +918,89 @@ func TestPeriodicReconciliationAdoptsFileThatAppearsAfterStartup(t *testing.T) {
} }
} }
// TestReconciliationPassesAreSpacedByTenTimesTheirTime makes the hook
// called after every read of the reconciliation pass sleep 10 ms, runs
// the eviction loop on a 10 ms interval, and records when each of those
// calls starts and ends. A pass starts before its first call and ends
// after its last, so the time from the first call's start to the last
// call's end is at most the pass's time. The next pass's first call must
// start at least ten times that long after the last call ended.
func TestReconciliationPassesAreSpacedByTenTimesTheirTime(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
// A pass made here, before the loop starts, counts the reads one
// pass makes on this empty cache, so the reads recorded below can be
// split into passes.
readsPerPass := 0
cache.reconciliationReadTestHook = func(int) { readsPerPass++ }
err := cache.reconcileAccounting(t.Context())
if err != nil {
t.Fatalf("reconcileAccounting failed: %v", err)
}
t.Logf("one reconciliation pass makes %d reads", readsPerPass)
if readsPerPass == 0 {
t.Fatal("a reconciliation pass made no read")
}
// The pass at startup and two after it.
const passes = 3
type read struct{ start, end time.Time }
// The channel holds every read the test looks at, so none of those
// is dropped; reads after them are, and never block the loop.
reads := make(chan read, passes*readsPerPass)
cache.reconciliationReadTestHook = func(int) {
start := time.Now()
time.Sleep(10 * time.Millisecond)
select {
case reads <- read{start: start, end: time.Now()}:
default:
}
}
cache.StartEviction(10 * time.Millisecond)
defer func() { _ = cache.StopEviction(t.Context()) }()
recorded := make([]read, 0, passes*readsPerPass)
timeout := time.After(15 * time.Second)
for len(recorded) < passes*readsPerPass {
select {
case r := <-reads:
recorded = append(recorded, r)
case <-timeout:
t.Fatalf("saw %d reconciliation reads in 15 s, want %d (%d passes)",
len(recorded), passes*readsPerPass, passes)
}
}
for pass := range passes - 1 {
first := recorded[pass*readsPerPass]
last := recorded[(pass+1)*readsPerPass-1]
next := recorded[(pass+1)*readsPerPass]
passTime := last.end.Sub(first.start)
wait := next.start.Sub(last.end)
t.Logf("pass %d took at least %v, and the next pass started %v after it",
pass+1, passTime, wait)
if wait < 10*passTime {
t.Errorf("pass %d took at least %v, and the next pass started %v "+
"after it, want at least %v", pass+1, passTime, wait, 10*passTime)
}
}
}
// TestStopEvictionInterruptsPassInProgress holds the test database's // TestStopEvictionInterruptsPassInProgress holds the test database's
// only connection, so the startup reconciliation pass waits for it, and // only connection, so the startup reconciliation pass waits for it, and
// checks that StopEviction stops that pass instead of waiting for the // checks that StopEviction stops that pass instead of waiting for the