Compare commits
4
Commits
87648b91d0
..
next
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
957ec85f97 | ||
|
|
930e43961e | ||
|
|
9eacfa6406 | ||
|
|
0ab15324f7 |
+4
-4
@@ -34,11 +34,11 @@ RUN --mount=type=cache,target=/root/.cache/go-build \
|
|||||||
FROM golang@sha256:a02d35efc036053fdf0da8c15919276bf777a80cbfda6a35c5e9f087e652adfc AS test
|
FROM golang@sha256:a02d35efc036053fdf0da8c15919276bf777a80cbfda6a35c5e9f087e652adfc AS test
|
||||||
|
|
||||||
# govips compiles against the libvips and libheif headers, and the tests
|
# govips compiles against the libvips and libheif headers, and the tests
|
||||||
# need libvips' JPEG XL support and libheif's AVIF encoder. Debian's
|
# need libvips' JPEG XL support; libvips-dev brings all three. The tests
|
||||||
# libvips-dev brings all of them, the encoder as a recommended package,
|
# also encode AVIF, which needs libheif's AVIF encoder,
|
||||||
# so recommended packages are installed here.
|
# libheif-plugin-aomenc: Debian only recommends it, so it is named here.
|
||||||
RUN apt-get update \
|
RUN apt-get update \
|
||||||
&& apt-get install -y libvips-dev \
|
&& apt-get install -y --no-install-recommends libvips-dev libheif-plugin-aomenc \
|
||||||
&& rm -rf /var/lib/apt/lists/*
|
&& rm -rf /var/lib/apt/lists/*
|
||||||
|
|
||||||
WORKDIR /src
|
WORKDIR /src
|
||||||
|
|||||||
@@ -102,8 +102,8 @@ it uses libvips through CGO. pixad does not start unless libvips has its JPEG XL
|
|||||||
support, which on Alpine is the `vips-jxl` package and which the nix and brew
|
support, which on Alpine is the `vips-jxl` package and which the nix and brew
|
||||||
packages of libvips include, as do the apt ones from Debian 12 and Ubuntu 24.04
|
packages of libvips include, as do the apt ones from Debian 12 and Ubuntu 24.04
|
||||||
on. Building pixa also needs the development files of libvips and libheif,
|
on. Building pixa also needs the development files of libvips and libheif,
|
||||||
`pkg-config` and a C compiler. `script/bootstrap --cgo` installs all of these,
|
`pkg-config` and a C compiler. `script/bootstrap --cgo` installs all of these;
|
||||||
as the `Dockerfile` does where it compiles pixa. Plain `script/bootstrap`, which
|
the `Dockerfile` installs its own packages. Plain `script/bootstrap`, which
|
||||||
`script/setup` and `script/cibuild` run, installs git, make and Go, and Node,
|
`script/setup` and `script/cibuild` run, installs git, make and Go, and Node,
|
||||||
Yarn and the prettier pinned in `yarn.lock` for formatting the markdown, but
|
Yarn and the prettier pinned in `yarn.lock` for formatting the markdown, but
|
||||||
none of the C libraries: the checks compile pixa in Docker. Docker itself must
|
none of the C libraries: the checks compile pixa in Docker. Docker itself must
|
||||||
@@ -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
|
||||||
@@ -697,7 +701,9 @@ Key settings in more detail:
|
|||||||
benchmark, NAT64, and the like); an invalid CIDR aborts startup. An IPv6
|
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 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
|
address it holds, so an IPv6 range such as `::/0` refuses neither it nor a
|
||||||
plain IPv4 address
|
plain IPv4 address. An entry written in that form is read as the IPv4 range it
|
||||||
|
covers, so `::ffff:10.0.0.0/104` refuses what `10.0.0.0/8` refuses; such an
|
||||||
|
entry shorter than `/96` covers more than IPv4 addresses and aborts startup
|
||||||
- `trusted_proxies` — list of CIDR ranges of the reverse proxies in front of
|
- `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
|
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
|
of these ranges; the logged and login-recorded client address is then the
|
||||||
@@ -707,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;
|
||||||
@@ -834,15 +846,16 @@ cache is kept from build to build: the steps that compile mount it, so a package
|
|||||||
whose inputs are unchanged is not compiled again, and both `go test` runs pass
|
whose inputs are unchanged is not compiled again, and both `go test` runs pass
|
||||||
`-count=1` so that no stored test result is replayed. The lint step also mounts
|
`-count=1` so that no stored test result is replayed. The lint step also mounts
|
||||||
golangci-lint's cache, which reuses a package's findings only when its files,
|
golangci-lint's cache, which reuses a package's findings only when its files,
|
||||||
those of the packages it imports, `go.mod`, `.golangci.yml` and the
|
those of the packages it imports, `go.mod`, the linters and their settings in
|
||||||
golangci-lint version are all unchanged. The `test` phase runs on the Debian Go
|
`.golangci.yml`, and the golangci-lint version are all unchanged. The `test`
|
||||||
image, with Debian's libvips and libheif; the image pixa ships is built and runs
|
phase runs on the Debian Go image, with Debian's libvips and libheif; the image
|
||||||
on Alpine. `script/check`, `script/cibuild`, `script/docker`, `script/lint`,
|
pixa ships is built and runs on Alpine. `script/check`, `script/cibuild`,
|
||||||
`script/test`, `script/setup` and `script/install-precommit` are the standard
|
`script/docker`, `script/lint`, `script/test`, `script/setup` and
|
||||||
copies from `sneak/prompts`, kept identical to them. `script/fmt` and
|
`script/install-precommit` are the standard copies from `sneak/prompts`, kept
|
||||||
`script/fmt-check` are the standard copies with pixa's `gofmt` step kept before
|
identical to them. `script/fmt` and `script/fmt-check` are the standard copies
|
||||||
prettier. prettier formats the markdown only, not the HTML templates, as it
|
with pixa's `gofmt` step kept before prettier. prettier formats the markdown
|
||||||
cannot parse a Go template action inside a tag.
|
only, not the HTML templates, as it cannot parse a Go template action inside a
|
||||||
|
tag.
|
||||||
|
|
||||||
## Load Test
|
## Load Test
|
||||||
|
|
||||||
|
|||||||
@@ -34,6 +34,16 @@ P2: security: per-origin rate limiting
|
|||||||
each `Dockerfile` stage installs its system packages and downloads the Go
|
each `Dockerfile` stage installs its system packages and downloads the Go
|
||||||
modules above `ARG CACHEBUST`, so check builds keep them, and the lint step
|
modules above `ARG CACHEBUST`, so check builds keep them, and the lint step
|
||||||
keeps golangci-lint's cache.
|
keeps golangci-lint's cache.
|
||||||
|
- 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;
|
||||||
|
`::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).
|
||||||
- 2026-10-16 P1 security bug: the check that keeps pixa from fetching internal
|
- 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
|
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
|
refuses `64:ff9b::/96`, and refuses an IPv6 address in the IPv4-compatible
|
||||||
|
|||||||
@@ -88,3 +88,56 @@ func TestBlockedNetworksInvalidAbortsStartup(t *testing.T) {
|
|||||||
},
|
},
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestBlockedNetworksMappedEntryIsIPv4Range 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 fetcher matches an IPv4-mapped address as the IPv4 address it
|
||||||
|
// holds. /96, the shortest such entry, covers every IPv4 address.
|
||||||
|
func TestBlockedNetworksMappedEntryIsIPv4Range(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
yamlContent := signingKeyLine + `blocked_networks:
|
||||||
|
- "::ffff:10.0.0.0/104"
|
||||||
|
- "::ffff:0:0/96"
|
||||||
|
`
|
||||||
|
|
||||||
|
c, err := configFromYAML(t, yamlContent)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("IPv4-mapped blocked_networks entries should load, got error: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
want := []string{"10.0.0.0/8", "0.0.0.0/0"}
|
||||||
|
if len(c.BlockedNetworks) != len(want) {
|
||||||
|
t.Fatalf("BlockedNetworks = %v, want %d entries", c.BlockedNetworks, len(want))
|
||||||
|
}
|
||||||
|
|
||||||
|
for i, w := range want {
|
||||||
|
if got := c.BlockedNetworks[i].String(); got != w {
|
||||||
|
t.Errorf("BlockedNetworks[%d] = %q, want %q", i, got, w)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestBlockedNetworksMappedEntryShorterThan96AbortsStartup 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 TestBlockedNetworksMappedEntryShorterThan96AbortsStartup(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
runAbortCases(t, []abortCase{
|
||||||
|
{
|
||||||
|
name: "mapped-shorter-than-96",
|
||||||
|
yaml: signingKeyLine + `blocked_networks:
|
||||||
|
- "::ffff:0.0.0.0/80"
|
||||||
|
`,
|
||||||
|
wantErrSubstrings: []string{keyBlockedNetworks, "::ffff:0.0.0.0/80"},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "mapped-95",
|
||||||
|
yaml: signingKeyLine + `blocked_networks:
|
||||||
|
- "::ffff:0:0/95"
|
||||||
|
`,
|
||||||
|
wantErrSubstrings: []string{keyBlockedNetworks},
|
||||||
|
},
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|||||||
@@ -91,6 +91,7 @@ var (
|
|||||||
errNotABoolean = errors.New("not a boolean")
|
errNotABoolean = errors.New("not a boolean")
|
||||||
errNotAStringList = errors.New("not a list of strings")
|
errNotAStringList = errors.New("not a list of strings")
|
||||||
errNotAValidCIDR = errors.New("not a valid CIDR network")
|
errNotAValidCIDR = errors.New("not a valid CIDR network")
|
||||||
|
errMappedTooShort = errors.New("in the IPv4-mapped form but shorter than /96")
|
||||||
errNotAMetricsMap = errors.New("not a map of metrics settings")
|
errNotAMetricsMap = errors.New("not a map of metrics settings")
|
||||||
errEmptyListEntry = errors.New("list contains an empty entry")
|
errEmptyListEntry = errors.New("list contains an empty entry")
|
||||||
errEmptyEntry = errors.New("contains an empty entry")
|
errEmptyEntry = errors.New("contains an empty entry")
|
||||||
@@ -1248,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 {
|
||||||
@@ -1263,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 {
|
||||||
@@ -1272,6 +1284,16 @@ func parseCIDRList(sc *smartconfig.Config, key string) ([]netip.Prefix, error) {
|
|||||||
settingName(key), entry, errNotAValidCIDR)
|
settingName(key), entry, errNotAValidCIDR)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if prefix.Addr().Is4In6() && prefix.Bits() < mappedPrefixBits {
|
||||||
|
return nil, fmt.Errorf("%s: value %q is %w",
|
||||||
|
settingName(key), entry, errMappedTooShort)
|
||||||
|
}
|
||||||
|
|
||||||
|
if prefix.Addr().Is4In6() {
|
||||||
|
prefix = netip.PrefixFrom(prefix.Addr().Unmap(),
|
||||||
|
prefix.Bits()-mappedPrefixBits)
|
||||||
|
}
|
||||||
|
|
||||||
prefixes = append(prefixes, prefix)
|
prefixes = append(prefixes, prefix)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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"},
|
||||||
|
},
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
Reference in New Issue
Block a user