Space reconciliation passes by ten times the last pass's time (closes #250)

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.
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.

Model: opus-5-5

README.md

@@ -199,7 +199,9 @@ 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
 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
-or converts the image again.
+or converts the image again. Looking for those files reads the database once for
+every cached file, so after a look that took longer than 30 seconds, pixa waits
+ten times as long as it took before looking again.
 
 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
@@ -694,7 +696,10 @@ Key settings in more detail:
   and generator pages. Default: empty
 - `blocked_networks` — list of CIDR ranges to refuse for SSRF protection, added
   to the always-enforced built-in ranges (loopback, private, link-local, CGNAT,
-  benchmark, NAT64, and the like); an invalid CIDR aborts startup
+  benchmark, NAT64, and the like); an invalid CIDR aborts startup. An IPv6
+  address in the IPv4-mapped form (`::ffff:a.b.c.d`) is matched only as the IPv4
+  address it holds, so an IPv6 range such as `::/0` refuses neither it nor a
+  plain IPv4 address
 - `trusted_proxies` — list of CIDR ranges of the reverse proxies in front of
   pixa. `X-Forwarded-For` is believed only when the direct peer falls inside one
   of these ranges; the logged and login-recorded client address is then the

TODO.md

@@ -30,6 +30,20 @@ P2: security: per-origin rate limiting
 
 # Completed Steps
 
+- 2026-10-18 the next reconciliation pass waits 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-16 P1 security bug: the check that keeps pixa from fetching internal
+  addresses refuses the local-use NAT64 range (`64:ff9b:1::/48`, RFC 8215) as it
+  refuses `64:ff9b::/96`, and refuses an IPv6 address in the IPv4-compatible
+  form (`::/96`) when `blocked_networks` contains either the address as written
+  or the IPv4 address it holds (closes #299). A source host such as
+  `[64:ff9b:1::a00:1]` was fetched, which on a network that runs NAT64 on that
+  range reaches the IPv4 address it holds, here `10.0.0.1`; and with
+  `203.0.113.0/24` in `blocked_networks`, `[::ffff:203.0.113.5]` was refused but
+  `[::203.0.113.5]` was fetched. Both now answer 403. The mapped form
+  (`::ffff:a.b.c.d`) is still matched only as the IPv4 address it holds: Go
+  keeps a plain IPv4 address in that form, and an IPv6 entry such as `::/0` must
+  not refuse every IPv4 address.
 - 2026-10-16 P1 security bug: the check that keeps pixa from fetching internal
   addresses judges an IPv6 address in the IPv4-compatible form (`::/96`), such
   as `::127.0.0.1` or `::10.0.0.1`, by the IPv4 address in its last four bytes

internal/httpfetcher/httpfetcher.go

@@ -59,8 +59,9 @@ const (
 // builtinBlockedPrefixes are internal or special-use ranges that Go's
 // net.IP predicates (IsPrivate, IsLinkLocalUnicast, and the like) do not
 // already cover. They are always blocked, in addition to any
-// operator-supplied networks. IPv4-mapped IPv6 addresses are unmapped
-// before matching, so these IPv4 ranges are caught in both forms.
+// operator-supplied networks. IPv4-mapped and IPv4-compatible IPv6
+// addresses are read as the IPv4 address they hold before matching, so
+// these IPv4 ranges are caught in all three forms.
 //
 //nolint:gochecknoglobals // immutable built-in blocklist
 var builtinBlockedPrefixes = []netip.Prefix{
@@ -68,6 +69,7 @@ var builtinBlockedPrefixes = []netip.Prefix{
 	netip.MustParsePrefix("192.0.0.0/24"),   // RFC 6890 IETF protocol assignments
 	netip.MustParsePrefix("198.18.0.0/15"),  // RFC 2544 benchmarking range
 	netip.MustParsePrefix("64:ff9b::/96"),   // RFC 6052 NAT64 (maps onto IPv4)
+	netip.MustParsePrefix("64:ff9b:1::/48"), // RFC 8215 local-use NAT64 (maps onto IPv4)
 }
 
 // Fetcher errors.
@@ -543,12 +545,8 @@ func isPrivateIP(ip net.IP) bool {
 		return true
 	}
 
-	// An IPv4-compatible IPv6 address (in ::/96, such as ::127.0.0.1) holds
-	// an IPv4 address in its last four bytes and is judged as that address
-	if addr, ok := netip.AddrFromSlice(ip); ok &&
-		netip.MustParsePrefix("::/96").Contains(addr) {
-		return isPrivateIP(ip[12:])
-	}
+	// An IPv4-compatible IPv6 address is judged as the IPv4 address it holds
+	ip = ipv4CompatibleToIPv4(ip)
 
 	// Check for loopback
 	if ip.IsLoopback() {
@@ -601,7 +599,12 @@ func isPrivateIP(ip net.IP) bool {
 }
 
 // isBlockedIP reports whether ip is refused, either by the built-in
-// internal-range check or by one of the operator-supplied prefixes.
+// internal-range check or by one of the operator-supplied prefixes. An
+// IPv4-compatible address (in ::/96) is refused when a prefix contains the
+// address as written or the IPv4 address it holds. An IPv4-mapped address (in
+// ::ffff:0:0/96) is matched only as the IPv4 address it holds: Go keeps a
+// plain IPv4 address in that form too, and an IPv6 prefix such as ::/0 must
+// not refuse every IPv4 address.
 func isBlockedIP(ip net.IP, blocked []netip.Prefix) bool {
 	if isPrivateIP(ip) {
 		return true
@@ -614,11 +617,28 @@ func isBlockedIP(ip net.IP, blocked []netip.Prefix) bool {
 
 	addr = addr.Unmap()
 
+	// ip read as IPv4 where it is or holds an IPv4 address, otherwise the
+	// zero netip.Addr, which no prefix contains. Matching asIPv4 as well as
+	// addr refuses more only for an IPv4-compatible address.
+	asIPv4, _ := netip.AddrFromSlice(ipv4CompatibleToIPv4(ip).To4())
+
 	return slices.ContainsFunc(blocked, func(prefix netip.Prefix) bool {
-		return prefix.Contains(addr)
+		return prefix.Contains(addr) || prefix.Contains(asIPv4)
 	})
 }
 
+// ipv4CompatibleToIPv4 returns the IPv4 address that an IPv4-compatible IPv6
+// address (in ::/96, such as ::127.0.0.1) holds in its last four bytes, and
+// any other address unchanged.
+func ipv4CompatibleToIPv4(ip net.IP) net.IP {
+	addr, ok := netip.AddrFromSlice(ip)
+	if ok && netip.MustParsePrefix("::/96").Contains(addr) {
+		return ip[12:]
+	}
+
+	return ip
+}
+
 // ssrfSafeDialer validates IP addresses against the built-in blocked ranges
 // before connecting. New wraps dialSSRFSafe with the operator-supplied
 // blocklist; this entry point enforces the built-in ranges alone.

internal/imgcache/eviction.go

@@ -18,6 +18,14 @@ import (
 // write-pressure wakeups triggered by stores.
 const DefaultEvictionInterval = 5 * time.Minute
 
+// reconciliationWaitFactor is how many times as long as a
+// reconciliation pass took the eviction loop waits, 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
 // and source blobs) one eviction pass fetches from the database.
 const evictionBatchSize = 100
@@ -438,10 +446,12 @@ func (c *Cache) notifyWritePressure() {
 }
 
 // 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
-// periodic tick thereafter, and evicts to the configured limit on the
-// given periodic interval and on write-pressure notifications. On a
+// periodic tick thereafter, reconciles the database accounting with the
+// cache directories at startup and again on the first periodic tick
+// that comes 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
 // 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
@@ -482,7 +492,7 @@ func (c *Cache) StopEviction(ctx context.Context) error {
 func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
 	defer close(c.evictionDone)
 
-	c.runReconciliationPass(ctx)
+	nextReconciliation := c.runReconciliationPass(ctx)
 	c.runCleanExpired(ctx)
 	c.runEvictionPass(ctx)
 
@@ -494,16 +504,20 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			// Reconciliation walks the cache directories, so it only
-			// runs on the periodic ticker rather than on every
-			// write-pressure wakeup, keeping it off the per-store hot
-			// path. Reusing the eviction interval itself (rather than a
-			// separate, longer one) is a deliberate choice: it is the
-			// simplest option that still bounds how long a store's
-			// best-effort accounting insert can stay silently
-			// unaccounted for to one interval, on a process that is
-			// already running this loop regardless.
-			c.runReconciliationPass(ctx)
+			// Reconciliation walks the cache directories and reads a
+			// row for every file, so it runs only on the periodic
+			// ticker rather than on every write-pressure wakeup, and
+			// not before nextReconciliation, which is ten times as long
+			// after the last pass ended as that pass took. Passes then
+			// never run back to back, however long one takes. A file
+			// whose best-effort accounting insert failed in a store
+			// stays unaccounted for until the next pass, for about the
+			// longer of one eviction interval and ten times the last
+			// pass's time.
+			if !time.Now().Before(nextReconciliation) {
+				nextReconciliation = c.runReconciliationPass(ctx)
+			}
+
 			c.runCleanExpired(ctx)
 		case <-c.evictionPressure:
 		}
@@ -527,17 +541,25 @@ func (c *Cache) runEvictionPass(ctx context.Context) {
 }
 
 // runReconciliationPass runs one reconciliation pass, logging failures
-// instead of propagating them (the loop must keep running). It does
-// nothing once ctx is cancelled.
-func (c *Cache) runReconciliationPass(ctx context.Context) {
+// instead of propagating them (the loop must keep running), and returns
+// the earliest time the next pass may start: reconciliationWaitFactor
+// 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 {
-		return
+		return time.Now()
 	}
 
+	started := time.Now()
+
 	err := c.reconcileAccounting(ctx)
 	if err != nil {
 		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
@@ -556,7 +578,8 @@ func (c *Cache) runCleanExpired(ctx context.Context) {
 
 // reconcileAccounting synchronizes the database size accounting with
 // the actual contents of the cache directories. It runs at startup and
-// again on every periodic eviction tick thereafter, off the request
+// again on later periodic eviction ticks, each no sooner than 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
 // (or whose accounting insert failed, e.g. StoreVariant's best-effort
 // 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
 // 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
-// accumulate unaccounted for on a long-running process to one eviction
-// interval. Once ctx is cancelled, it stops at the next file or row and
-// returns ctx's error.
+// accumulate unaccounted for on a long-running process to about the
+// longer of one eviction interval and ten times the last pass's time.
+// 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 {
 		return nil

internal/imgcache/eviction_internal_test.go

@@ -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
 // only connection, so the startup reconciliation pass waits for it, and
 // checks that StopEviction stops that pass instead of waiting for the

internal/server/image_proxy_internal_test.go

@@ -14,10 +14,12 @@ import (
 	"net"
 	"net/http"
 	"net/http/httptest"
+	"net/netip"
 	"os"
 	"path/filepath"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"go.uber.org/fx"
 	"go.uber.org/fx/fxtest"
@@ -183,6 +185,57 @@ func startImageProxy(
 	return s, db.DB(), stateDir
 }
 
+// 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.
+func startImageProxyWithBlockedNetworks(
+	t *testing.T, blockedNetworks []netip.Prefix,
+) *Server {
+	t.Helper()
+
+	stateDir := t.TempDir()
+	cfg := &config.Config{
+		SigningKey:      testSigningKey,
+		StateDir:        stateDir,
+		DBURL:           "file:" + filepath.Join(stateDir, "state.sqlite3"),
+		BlockedNetworks: blockedNetworks,
+		// A fetch that gets past the checks and finds nothing at the address
+		// 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,
+	}
+
+	var (
+		h  *handlers.Handlers
+		mw *middleware.Middleware
+	)
+
+	app := fxtest.New(t,
+		fx.Supply(cfg),
+		fx.Provide(
+			globals.New,
+			logger.New,
+			database.New,
+			healthcheck.New,
+			handlers.New,
+			middleware.New,
+		),
+		fx.Populate(&h, &mw),
+	)
+	app.RequireStart()
+	t.Cleanup(app.RequireStop)
+
+	s := &Server{config: cfg, mw: mw, h: h}
+	s.SetupRoutes()
+
+	return s
+}
+
 // getImage sends a GET for target to s and fails unless it answers 200.
 func getImage(t *testing.T, s *Server, target string) *httptest.ResponseRecorder {
 	t.Helper()

internal/server/source_url_internal_test.go

@@ -3,6 +3,7 @@ package server
 import (
 	"net/http"
 	"net/http/httptest"
+	"net/netip"
 	"net/url"
 	"regexp"
 	"slices"
@@ -586,3 +587,79 @@ func TestImageRoutesRefuseIPv6SourceHostHoldingInternalIPv4Address(t *testing.T)
 		t.Errorf("upstream received %q, want no request", got)
 	}
 }
+
+// TestImageRoutesCheckIPv4AddressInIPv6SourceHost requests ipv6SourcePath on
+// source hosts that hold an IPv4 address, through a /v1/image/ URL with a
+// signature and through an encrypted URL, with the fetcher pixad uses.
+// [64:ff9b:1::a00:1], in the local-use NAT64 range (64:ff9b:1::/48, RFC 8215),
+// answers 403. [::203.0.113.5] holds the public address 203.0.113.5 in the
+// IPv4-compatible form (::/96) and answers 403 when blocked_networks lists
+// 203.0.113.0/24 or ::/96. [::ffff:203.0.113.5] holds it in the IPv4-mapped
+// form (::ffff:0:0/96) and answers 403 when blocked_networks lists
+// 203.0.113.0/24. Go keeps the plain address 203.0.113.5 in the mapped form
+// too, and an IPv6 range must not refuse it: with ::/0 listed it is not
+// refused. Neither is [::203.0.113.5] with blocked_networks empty. Where a
+// host is not refused, the fetcher tries to connect to it, finds nothing
+// there, and the request fails with another status.
+func TestImageRoutesCheckIPv4AddressInIPv6SourceHost(t *testing.T) {
+	t.Parallel()
+
+	const (
+		ipv4CompatibleHost = "[::203.0.113.5]"
+		ipv4MappedHost     = "[::ffff:203.0.113.5]"
+	)
+
+	tests := []struct {
+		name            string
+		sourceHost      string
+		blockedNetworks []netip.Prefix
+		wantForbidden   bool
+	}{
+		{"local-use NAT64 form of 10.0.0.1", "[64:ff9b:1::a00:1]", nil, true},
+		{"IPv4-compatible form of an address in blocked_networks",
+			ipv4CompatibleHost,
+			[]netip.Prefix{netip.MustParsePrefix("203.0.113.0/24")}, true},
+		{"IPv4-compatible form with ::/96 in blocked_networks",
+			ipv4CompatibleHost,
+			[]netip.Prefix{netip.MustParsePrefix("::/96")}, true},
+		{"IPv4-mapped form of an address in blocked_networks",
+			ipv4MappedHost,
+			[]netip.Prefix{netip.MustParsePrefix("203.0.113.0/24")}, true},
+		{"plain IPv4 address with ::/0 in blocked_networks",
+			"203.0.113.5",
+			[]netip.Prefix{netip.MustParsePrefix("::/0")}, false},
+		{"IPv4-compatible form of an address with blocked_networks empty",
+			ipv4CompatibleHost, nil, false},
+	}
+
+	for _, tt := range tests {
+		imageURLs := map[string]string{
+			"image URL with a signature": signedImageURL(t, tt.sourceHost),
+			"encrypted URL":              encryptedImageURL(t, tt.sourceHost),
+		}
+
+		for route, imageURL := range imageURLs {
+			t.Run(tt.name+" "+route, func(t *testing.T) {
+				t.Parallel()
+
+				s := startImageProxyWithBlockedNetworks(t, tt.blockedNetworks)
+
+				rec := httptest.NewRecorder()
+				s.ServeHTTP(rec, httptest.NewRequestWithContext(
+					t.Context(), http.MethodGet, imageURL, nil))
+				t.Logf("GET %s: %d", imageURL, rec.Code)
+
+				forbidden := rec.Code == http.StatusForbidden
+				if tt.wantForbidden && !forbidden {
+					t.Errorf("status = %d, want %d; body %s",
+						rec.Code, http.StatusForbidden, rec.Body.String())
+				}
+
+				if !tt.wantForbidden && forbidden {
+					t.Errorf("status = %d, want any other status; body %s",
+						rec.Code, rec.Body.String())
+				}
+			})
+		}
+	}
+}