Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
dc1a78324a |
+3
-22
@@ -59,28 +59,9 @@ RUN --mount=type=cache,target=/root/.cache/go-build \
|
||||
{ echo "--- Rerunning with -v for details ---"; \
|
||||
go test -count=1 -timeout 90s -race -parallel 4 -v ./...; exit 1; }
|
||||
|
||||
# Benchmark phase. script/bench builds it alone; no other stage copies from
|
||||
# it, so script/check and the image build never run it. It is set up as the
|
||||
# test phase is, and runs the benchmarks only, without -race.
|
||||
# golang:1.25.4-trixie, 2026-10-15
|
||||
FROM golang@sha256:a02d35efc036053fdf0da8c15919276bf777a80cbfda6a35c5e9f087e652adfc AS bench
|
||||
|
||||
RUN apt-get update \
|
||||
&& apt-get install -y --no-install-recommends libvips-dev libheif-plugin-aomenc \
|
||||
&& rm -rf /var/lib/apt/lists/*
|
||||
|
||||
WORKDIR /src
|
||||
COPY go.mod go.sum ./
|
||||
RUN go mod download
|
||||
ARG CACHEBUST
|
||||
COPY . .
|
||||
|
||||
RUN --mount=type=cache,target=/root/.cache/go-build \
|
||||
go test -count=1 -run '^$' -bench . ./internal/handlers
|
||||
|
||||
# Build stage. Nothing is wanted from the lint and test phases above: these
|
||||
# copies make BuildKit build them first, so nothing after them runs unless
|
||||
# lint and test passed.
|
||||
# Build stage. Nothing is wanted from the two phases above: these copies
|
||||
# make BuildKit build them first, so nothing after them runs unless lint
|
||||
# and test passed.
|
||||
# golang:1.25.4-alpine3.22, 2026-02-25; the runtime stage uses Alpine 3.22 too
|
||||
FROM golang:1.25.4-alpine3.22@sha256:d3f0cf7723f3429e3f9ed846243970b20a2de7bae6a5b66fc5914e228d831bbb AS builder
|
||||
|
||||
|
||||
@@ -65,7 +65,8 @@ loadtest:
|
||||
@script/loadtest
|
||||
|
||||
# Serve cache hits in parallel through the image handler and report hits per
|
||||
# second (needs Docker; a benchmark, not part of check)
|
||||
# second (needs Go on the host after `script/bootstrap --cgo`, not Docker; a
|
||||
# benchmark, not part of check)
|
||||
bench:
|
||||
@script/bench
|
||||
|
||||
|
||||
@@ -91,9 +91,9 @@ which answers 200 whenever pixa is running, in maintenance mode too (see
|
||||
On SIGTERM or SIGINT pixa stops accepting connections, gives the requests in
|
||||
progress and the images being processed 5 seconds to finish, then writes to the
|
||||
database the counts of cache hits, misses, fetches and conversions and the times
|
||||
cached images were last served that are not written yet (see Storage), and
|
||||
exits: with 0, or with 1 when images were still being processed after those 5
|
||||
seconds, some of those counts could not be written, or another part of pixa
|
||||
cached images were last used that are not written yet (see Storage), and exits:
|
||||
with 0, or with 1 when images were still being processed after those 5 seconds,
|
||||
some of those counts or times could not be written, or another part of pixa
|
||||
failed to stop. A request not finished by then is cut off. `docker stop` waits
|
||||
10 seconds before it kills the container.
|
||||
|
||||
@@ -211,11 +211,12 @@ bytes, which are those of the format pixa wrote it in; a file whose first bytes
|
||||
are not those of a format pixa writes is not served, and the image is converted
|
||||
again. The time the image was last served, which eviction orders by, and the hit
|
||||
count are handed to a background writer without waiting for it, as are the
|
||||
counts of misses, fetches and conversions. Once a second that writer writes the
|
||||
waiting times in one transaction and the counts in one statement. While 10,000
|
||||
times wait to be written, the time of a further hit is dropped, and that image
|
||||
only looks less recently used to eviction. `script/bench` measures how many hits
|
||||
pixa serves per second (see Load Test).
|
||||
counts of misses, fetches and conversions, and the time a miss last used a
|
||||
cached source image. Once a second that writer writes the waiting times in one
|
||||
transaction and the counts in one statement. While 10,000 times of transformed
|
||||
images, or 10,000 of source images, wait to be written, a further one is
|
||||
dropped, and that image only looks less recently used to eviction.
|
||||
`script/bench` measures how many hits pixa serves per second (see Load Test).
|
||||
|
||||
### Routes
|
||||
|
||||
@@ -854,8 +855,8 @@ provide:
|
||||
- `script/loadtest` — measure pixad's throughput, latency and peak memory; a
|
||||
benchmark, not part of `script/check` (see Load Test)
|
||||
- `script/bench` — run the Go benchmark that serves cache hits in parallel
|
||||
through the image handler: build the `bench` phase of the `Dockerfile`; a
|
||||
benchmark, not part of `script/check` (see Load Test)
|
||||
through the image handler, with Go on the host after `script/bootstrap --cgo`;
|
||||
a benchmark, not part of `script/check` (see Load Test)
|
||||
- `script/cibuild` — CI entrypoint: run `script/bootstrap` (without `--cgo`),
|
||||
then `script/check`, then build the image as `script/docker` does
|
||||
- `script/precommit` — pre-commit checks (`go mod tidy` guard, then
|
||||
@@ -870,7 +871,7 @@ instead of coming from the build cache. Above the line are the steps that
|
||||
install system packages and download the Go modules; Docker keeps their layers
|
||||
until the base image, the step itself, `go.mod` or `go.sum` changes. Go's build
|
||||
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 every `go test` run passes
|
||||
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
|
||||
golangci-lint's cache, which reuses a package's findings only when its files,
|
||||
those of the packages it imports, `go.mod`, the linters and their settings in
|
||||
@@ -943,8 +944,9 @@ the machine has CPUs, and reports the hits served per second as `hits/s`. Each
|
||||
run starts with an empty cache and an empty database, a SQLite file of its own,
|
||||
and puts the image in the cache before it measures. The benchmark fails if
|
||||
serving the hits ran any statement on the database. Like the load test it is not
|
||||
part of `script/check`; it needs Docker, as it runs in the `bench` phase of the
|
||||
`Dockerfile`, and prints the result with the build's output.
|
||||
part of `script/check`. It runs Go on the host, as the load test does, and
|
||||
compiles pixa there, so it needs the C compiler and libraries that
|
||||
`script/bootstrap --cgo` installs, but not Docker.
|
||||
|
||||
## TODO
|
||||
|
||||
|
||||
@@ -36,15 +36,17 @@ P2: security: per-origin rate limiting
|
||||
file whose first bytes are not those of an output format, SVG included, is a
|
||||
miss, and the variant is made again. A hit hands its access time to a channel
|
||||
that holds up to 10,000 and drops the rest, and its hit count to the pending
|
||||
counts, without waiting; every count is now a pending count. The goroutine
|
||||
`StartPendingCountWrites` starts writes, once a second, the access times in
|
||||
one transaction and the counts in one `UPDATE`; `StopPendingCountWrites`
|
||||
writes what is left once at shutdown. `checkNegativeCache` no longer deletes
|
||||
an expired row; `CleanExpired` in the eviction goroutine does. The `.meta`
|
||||
files, `VariantMeta`, `LoadWithMeta`, `variantContentTypeFromSidecar`,
|
||||
`metaCache`, `Lookup` and `golang-lru` are gone. `make bench` (`script/bench`)
|
||||
runs `BenchmarkHandleImage_CacheHits` in the new `bench` phase of the
|
||||
`Dockerfile`, outside `make check`.
|
||||
counts, without waiting; every count is now a pending count. A miss that finds
|
||||
a cached source hands that source's access time to a second such channel. The
|
||||
goroutine `StartPendingCountWrites` starts writes, once a second, the access
|
||||
times in one transaction and the counts in one `UPDATE`;
|
||||
`StopPendingCountWrites` writes what is left once at shutdown and logs
|
||||
separately the access times and the counts it could not write.
|
||||
`checkNegativeCache` no longer deletes an expired row; `CleanExpired` in the
|
||||
eviction goroutine does. The `.meta` files, `VariantMeta`, `LoadWithMeta`,
|
||||
`variantContentTypeFromSidecar`, `metaCache`, `Lookup` and `golang-lru` are
|
||||
gone. `make bench` (`script/bench`) runs `BenchmarkHandleImage_CacheHits` with
|
||||
Go on the host after `script/bootstrap --cgo`, outside `make check`.
|
||||
- 2026-10-21 a source is stored the way a variant is (closes #308): one
|
||||
`source_content` row and one file per source URL, the file named by the
|
||||
source's `SourceKey`, the SHA-256 of its host, path and query made as the
|
||||
|
||||
+35
-33
@@ -23,8 +23,9 @@ var (
|
||||
)
|
||||
|
||||
// maxPendingAccessTimes is how many access times of variants served from
|
||||
// the cache may wait to be written, at about 80 bytes each. The access time
|
||||
// of a hit served while that many wait is dropped.
|
||||
// the cache, and how many of cached sources a miss used, may each wait to be
|
||||
// written, at about 80 bytes each. An access time handed over while that
|
||||
// many of its kind wait is dropped.
|
||||
const maxPendingAccessTimes = 10000
|
||||
|
||||
// cleanExpiredBatchSize is the most expired negative cache rows one
|
||||
@@ -84,10 +85,11 @@ type Cache struct {
|
||||
// The pending counts: hits, misses, upstream fetches and transforms
|
||||
// that requests counted, kept here until they are written by the
|
||||
// goroutine that StartPendingCountWrites starts, so that no request
|
||||
// waits for the database to count it. pendingAccessTimes holds the
|
||||
// variants served from the cache whose access time is still to be
|
||||
// written by that goroutine. As for eviction, the channels are created
|
||||
// in NewCache, so that requests can hand over access times before that
|
||||
// waits for the database to count it. pendingVariantAccessTimes holds
|
||||
// the variants served from the cache, and pendingSourceAccessTimes the
|
||||
// cached sources a miss used, whose access time is still to be written
|
||||
// by that goroutine. As for eviction, the channels are created in
|
||||
// NewCache, so that requests can hand over access times before that
|
||||
// goroutine starts; pendingCountsDone is closed when it returns.
|
||||
// pendingCountsCancel, set by StartPendingCountWrites, cancels its
|
||||
// context, which tells it to finish.
|
||||
@@ -96,7 +98,8 @@ type Cache struct {
|
||||
pendingUpstreamFetches atomic.Int64
|
||||
pendingUpstreamFetchBytes atomic.Int64
|
||||
pendingTransforms atomic.Int64
|
||||
pendingAccessTimes chan imagerequest.VariantKey
|
||||
pendingVariantAccessTimes chan imagerequest.VariantKey
|
||||
pendingSourceAccessTimes chan imagerequest.SourceKey
|
||||
pendingCountsDone chan struct{}
|
||||
pendingCountsCancel context.CancelFunc
|
||||
|
||||
@@ -139,8 +142,9 @@ func newCache(
|
||||
evictionPressure: make(chan struct{}, 1),
|
||||
evictionDone: make(chan struct{}),
|
||||
|
||||
pendingAccessTimes: make(chan imagerequest.VariantKey, maxPendingAccessTimes),
|
||||
pendingCountsDone: make(chan struct{}),
|
||||
pendingVariantAccessTimes: make(chan imagerequest.VariantKey, maxPendingAccessTimes),
|
||||
pendingSourceAccessTimes: make(chan imagerequest.SourceKey, maxPendingAccessTimes),
|
||||
pendingCountsDone: make(chan struct{}),
|
||||
|
||||
reconciliationPageSize: defaultReconciliationPageSize,
|
||||
}
|
||||
@@ -290,9 +294,9 @@ func (c *Cache) StoreVariant(
|
||||
|
||||
// LookupSource checks if we have cached source content for a request.
|
||||
// Returns the source key and content type if found, or empty values
|
||||
// if not; a row whose file is missing is not found. Hits touch the
|
||||
// source's LRU timestamp; a disabled cache always reports no cached
|
||||
// source.
|
||||
// if not; a row whose file is missing is not found. A source found has
|
||||
// its access time handed to the background writer without waiting (see
|
||||
// touchSourceContent); a disabled cache always reports no cached source.
|
||||
func (c *Cache) LookupSource(
|
||||
ctx context.Context, req *imagerequest.Request,
|
||||
) (imagerequest.SourceKey, string, error) {
|
||||
@@ -321,7 +325,7 @@ func (c *Cache) LookupSource(
|
||||
return "", "", nil
|
||||
}
|
||||
|
||||
c.touchSourceContent(ctx, key)
|
||||
c.touchSourceContent(key)
|
||||
|
||||
return key, contentType, nil
|
||||
}
|
||||
@@ -425,19 +429,19 @@ func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) {
|
||||
// methods, it only adds to the pending counts, which the goroutine
|
||||
// StartPendingCountWrites starts writes to the database, so it never waits
|
||||
// for the database or fails.
|
||||
func (c *Cache) IncrementStats(ctx context.Context, hit bool, fetchBytes int64) {
|
||||
func (c *Cache) IncrementStats(hit bool, fetchBytes int64) {
|
||||
if hit {
|
||||
c.pendingHits.Add(1)
|
||||
} else {
|
||||
c.pendingMisses.Add(1)
|
||||
}
|
||||
|
||||
c.IncrementUpstreamFetch(ctx, fetchBytes)
|
||||
c.IncrementUpstreamFetch(fetchBytes)
|
||||
}
|
||||
|
||||
// IncrementUpstreamFetch counts one upstream fetch that read fetchBytes bytes.
|
||||
// A fetch that read no bytes is not counted. The context is not used.
|
||||
func (c *Cache) IncrementUpstreamFetch(_ context.Context, fetchBytes int64) {
|
||||
// A fetch that read no bytes is not counted.
|
||||
func (c *Cache) IncrementUpstreamFetch(fetchBytes int64) {
|
||||
if fetchBytes <= 0 {
|
||||
return
|
||||
}
|
||||
@@ -447,8 +451,7 @@ func (c *Cache) IncrementUpstreamFetch(_ context.Context, fetchBytes int64) {
|
||||
}
|
||||
|
||||
// IncrementTransformCount counts one image transcoded by the image processor.
|
||||
// The context is not used.
|
||||
func (c *Cache) IncrementTransformCount(_ context.Context) {
|
||||
func (c *Cache) IncrementTransformCount() {
|
||||
c.pendingTransforms.Add(1)
|
||||
}
|
||||
|
||||
@@ -473,26 +476,25 @@ func (c *Cache) readHitAndMissCounts(ctx context.Context, stats *CacheStats) err
|
||||
|
||||
// touchVariant hands the access time of a variant served from the cache to
|
||||
// the goroutine StartPendingCountWrites starts, which writes it in its next
|
||||
// batch. It never waits: while maxPendingAccessTimes access times wait to be
|
||||
// written, this one is dropped, and the variant only looks colder to
|
||||
// eviction.
|
||||
// batch. It never waits: while maxPendingAccessTimes access times of
|
||||
// variants wait to be written, this one is dropped, and the variant only
|
||||
// looks colder to eviction.
|
||||
func (c *Cache) touchVariant(cacheKey imagerequest.VariantKey) {
|
||||
select {
|
||||
case c.pendingAccessTimes <- cacheKey:
|
||||
case c.pendingVariantAccessTimes <- cacheKey:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// touchSourceContent updates the LRU timestamp of a cached source,
|
||||
// best-effort: a failed touch only makes the source look colder.
|
||||
func (c *Cache) touchSourceContent(ctx context.Context, key imagerequest.SourceKey) {
|
||||
_, err := c.db.ExecContext(ctx, `
|
||||
UPDATE source_content SET last_accessed_at = CURRENT_TIMESTAMP
|
||||
WHERE source_key = ?
|
||||
`, string(key))
|
||||
if err != nil {
|
||||
c.log.Debug("failed to touch source content LRU timestamp",
|
||||
"source_key", key, "error", err)
|
||||
// touchSourceContent hands the access time of a cached source a miss used to
|
||||
// the goroutine StartPendingCountWrites starts, which writes it in its next
|
||||
// batch. It never waits: while maxPendingAccessTimes access times of sources
|
||||
// wait to be written, this one is dropped, and the source only looks colder
|
||||
// to eviction.
|
||||
func (c *Cache) touchSourceContent(key imagerequest.SourceKey) {
|
||||
select {
|
||||
case c.pendingSourceAccessTimes <- key:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -218,10 +218,11 @@ func TestService_Get_HitWaitsForNoDatabaseWrite(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestTouchVariant_DropsAccessTimesPastTheLimit hands one more access time
|
||||
// than may wait to be written. touchVariant must return each time without
|
||||
// waiting, and the last one must be dropped.
|
||||
func TestTouchVariant_DropsAccessTimesPastTheLimit(t *testing.T) {
|
||||
// TestTouchVariantAndTouchSourceContent_DropAccessTimesPastTheLimit hands one
|
||||
// more access time of a variant, and one more of a source, than may wait to
|
||||
// be written. touchVariant and touchSourceContent must return each time
|
||||
// without waiting, and the last of each must be dropped.
|
||||
func TestTouchVariantAndTouchSourceContent_DropAccessTimesPastTheLimit(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
@@ -233,17 +234,24 @@ func TestTouchVariant_DropsAccessTimesPastTheLimit(t *testing.T) {
|
||||
|
||||
for range maxPendingAccessTimes + 1 {
|
||||
cache.touchVariant(testVariantKeyOne)
|
||||
cache.touchSourceContent("aabbccdd3001")
|
||||
}
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(pendingCountsWait):
|
||||
t.Fatal("touchVariant waited once the access times reached the limit")
|
||||
t.Fatal("touchVariant or touchSourceContent waited once the access " +
|
||||
"times reached the limit")
|
||||
}
|
||||
|
||||
if waiting := len(cache.pendingAccessTimes); waiting != maxPendingAccessTimes {
|
||||
t.Errorf("%d access times wait to be written, want %d",
|
||||
if waiting := len(cache.pendingVariantAccessTimes); waiting != maxPendingAccessTimes {
|
||||
t.Errorf("%d access times of variants wait to be written, want %d",
|
||||
waiting, maxPendingAccessTimes)
|
||||
}
|
||||
|
||||
if waiting := len(cache.pendingSourceAccessTimes); waiting != maxPendingAccessTimes {
|
||||
t.Errorf("%d access times of sources wait to be written, want %d",
|
||||
waiting, maxPendingAccessTimes)
|
||||
}
|
||||
}
|
||||
@@ -285,7 +293,7 @@ func TestStopPendingCountWrites_WritesPendingAccessTimes(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
if waiting := len(cache.pendingAccessTimes); waiting != 0 {
|
||||
if waiting := len(cache.pendingVariantAccessTimes); waiting != 0 {
|
||||
t.Errorf("%d access times still wait to be written, want 0", waiting)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -104,9 +104,9 @@ func TestCache_Stats(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
// Increment some stats
|
||||
cache.IncrementStats(ctx, true, 0)
|
||||
cache.IncrementStats(ctx, true, 0)
|
||||
cache.IncrementStats(ctx, false, 1024)
|
||||
cache.IncrementStats(true, 0)
|
||||
cache.IncrementStats(true, 0)
|
||||
cache.IncrementStats(false, 1024)
|
||||
|
||||
stats, err := cache.Stats(ctx)
|
||||
if err != nil {
|
||||
|
||||
@@ -512,16 +512,12 @@ func (c *Cache) reconcileAccounting(ctx context.Context) error {
|
||||
return c.reconcileUsageTotal(ctx)
|
||||
}
|
||||
|
||||
// reconcileFiles walks baseDir, one of the cache's storage directories,
|
||||
// sweeping stale temp files and passing each other file, with its name, to
|
||||
// reconcileFile. Once ctx is cancelled it stops at the next file and
|
||||
// returns ctx's error.
|
||||
func (c *Cache) reconcileFiles(
|
||||
ctx context.Context, baseDir string,
|
||||
reconcileFile func(path string, entry fs.DirEntry, name string) error,
|
||||
) error {
|
||||
// reconcileVariantFiles walks the variant storage directory, checking
|
||||
// each variant file against its accounting row (reconcileVariantFile)
|
||||
// and sweeping stale temp files.
|
||||
func (c *Cache) reconcileVariantFiles(ctx context.Context) error {
|
||||
return filepath.WalkDir(
|
||||
baseDir,
|
||||
c.variants.baseDir,
|
||||
func(path string, entry fs.DirEntry, err error) error {
|
||||
if ctx.Err() != nil {
|
||||
return ctx.Err()
|
||||
@@ -539,21 +535,11 @@ func (c *Cache) reconcileFiles(
|
||||
return nil
|
||||
}
|
||||
|
||||
return reconcileFile(path, entry, name)
|
||||
return c.reconcileVariantFile(ctx, path, entry, imagerequest.VariantKey(name))
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
// reconcileVariantFiles walks the variant storage directory, checking
|
||||
// each variant file against its accounting row (reconcileVariantFile)
|
||||
// and sweeping stale temp files.
|
||||
func (c *Cache) reconcileVariantFiles(ctx context.Context) error {
|
||||
return c.reconcileFiles(ctx, c.variants.baseDir,
|
||||
func(path string, entry fs.DirEntry, name string) error {
|
||||
return c.reconcileVariantFile(ctx, path, entry, imagerequest.VariantKey(name))
|
||||
})
|
||||
}
|
||||
|
||||
// reconcileVariantFile checks one variant file against its accounting
|
||||
// row. A file without a row is adopted into the accounting, unless it is
|
||||
// empty. A file that is empty, or not the size its row records, as a
|
||||
@@ -710,10 +696,28 @@ func queryStringColumn[T ~string](
|
||||
// each blob file against its source_content row (reconcileSourceFile)
|
||||
// and sweeping stale temp files.
|
||||
func (c *Cache) reconcileSourceFiles(ctx context.Context) error {
|
||||
return c.reconcileFiles(ctx, c.srcContent.baseDir,
|
||||
func(path string, entry fs.DirEntry, name string) error {
|
||||
return filepath.WalkDir(
|
||||
c.srcContent.baseDir,
|
||||
func(path string, entry fs.DirEntry, err error) error {
|
||||
if ctx.Err() != nil {
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
if err != nil || entry.IsDir() {
|
||||
return err
|
||||
}
|
||||
|
||||
name := entry.Name()
|
||||
|
||||
if strings.HasPrefix(name, tempFilePrefix) {
|
||||
c.sweepStaleTempFile(path, entry)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
return c.reconcileSourceFile(ctx, path, entry, imagerequest.SourceKey(name))
|
||||
})
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
// reconcileSourceFile checks one source blob file against its
|
||||
|
||||
@@ -39,10 +39,10 @@ func (c *Cache) StartPendingCountWrites() {
|
||||
// started, if it did, to finish, and waits for its write under way to end
|
||||
// and for it to return. It then writes the pending access times and counts
|
||||
// left, once, so that they are written before the database closes. It waits
|
||||
// for all of this at most until ctx ends, then logs the counts not written
|
||||
// and returns an error.
|
||||
// for all of this at most until ctx ends. It logs what was not written,
|
||||
// access times or counts, and returns an error.
|
||||
func (c *Cache) StopPendingCountWrites(ctx context.Context) error {
|
||||
var err error
|
||||
var accessTimesErr, countsErr error
|
||||
|
||||
if c.pendingCountsCancel != nil {
|
||||
c.pendingCountsCancel()
|
||||
@@ -50,32 +50,37 @@ func (c *Cache) StopPendingCountWrites(ctx context.Context) error {
|
||||
select {
|
||||
case <-c.pendingCountsDone:
|
||||
case <-ctx.Done():
|
||||
// The write given up on here may not reach the database:
|
||||
// database/sql closes without waiting for a query, and a
|
||||
// write still waiting for the connection fails when the
|
||||
// database closes.
|
||||
err = fmt.Errorf("pending counts still being written: %w", ctx.Err())
|
||||
// The write given up on here, of access times or of counts,
|
||||
// may not reach the database: database/sql closes without
|
||||
// waiting for a query, and a write still waiting for the
|
||||
// connection fails when the database closes. Neither is
|
||||
// written here.
|
||||
accessTimesErr = fmt.Errorf("access times still being written: %w", ctx.Err())
|
||||
countsErr = fmt.Errorf("pending counts still being written: %w", ctx.Err())
|
||||
}
|
||||
}
|
||||
|
||||
if err == nil {
|
||||
err = errors.Join(c.writePendingAccessTimes(ctx), c.writePendingCounts(ctx))
|
||||
if accessTimesErr == nil && countsErr == nil {
|
||||
accessTimesErr = c.writePendingAccessTimes(ctx)
|
||||
countsErr = c.writePendingCounts(ctx)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
if accessTimesErr != nil {
|
||||
c.log.Error("access times not written at shutdown", "error", accessTimesErr)
|
||||
}
|
||||
|
||||
if countsErr != nil {
|
||||
c.log.Error("counts not written at shutdown",
|
||||
"hits", c.pendingHits.Load(),
|
||||
"misses", c.pendingMisses.Load(),
|
||||
"upstream_fetches", c.pendingUpstreamFetches.Load(),
|
||||
"upstream_fetch_bytes", c.pendingUpstreamFetchBytes.Load(),
|
||||
"transforms", c.pendingTransforms.Load(),
|
||||
"error", err,
|
||||
"error", countsErr,
|
||||
)
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
return errors.Join(accessTimesErr, countsErr)
|
||||
}
|
||||
|
||||
// pendingCountWriteLoop is the body of the goroutine
|
||||
@@ -115,43 +120,61 @@ func (c *Cache) pendingCountWriteLoop(ctx context.Context) {
|
||||
}
|
||||
|
||||
// writePendingAccessTimes takes out the variants waiting in
|
||||
// pendingAccessTimes and sets their last_accessed_at to now in one
|
||||
// transaction, once for each variant however often it was served. Nothing
|
||||
// else takes from pendingAccessTimes while it runs: only the goroutine
|
||||
// pendingVariantAccessTimes and the sources waiting in
|
||||
// pendingSourceAccessTimes, and sets their last_accessed_at to now in one
|
||||
// transaction, once for each however often it was used. Nothing else takes
|
||||
// from those channels while it runs: only the goroutine
|
||||
// StartPendingCountWrites starts calls it, or StopPendingCountWrites once
|
||||
// that goroutine has returned.
|
||||
func (c *Cache) writePendingAccessTimes(ctx context.Context) error {
|
||||
served := make(map[imagerequest.VariantKey]struct{})
|
||||
variants := make(map[imagerequest.VariantKey]struct{})
|
||||
|
||||
for range len(c.pendingAccessTimes) {
|
||||
served[<-c.pendingAccessTimes] = struct{}{}
|
||||
for range len(c.pendingVariantAccessTimes) {
|
||||
variants[<-c.pendingVariantAccessTimes] = struct{}{}
|
||||
}
|
||||
|
||||
if len(served) == 0 {
|
||||
sources := make(map[imagerequest.SourceKey]struct{})
|
||||
|
||||
for range len(c.pendingSourceAccessTimes) {
|
||||
sources[<-c.pendingSourceAccessTimes] = struct{}{}
|
||||
}
|
||||
|
||||
accessTimes := len(variants) + len(sources)
|
||||
if accessTimes == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
tx, err := c.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write %d access times: %w", len(served), err)
|
||||
return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
|
||||
}
|
||||
|
||||
// Does nothing once the transaction is committed
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
|
||||
for cacheKey := range served {
|
||||
for cacheKey := range variants {
|
||||
_, err = tx.ExecContext(ctx, `
|
||||
UPDATE variant_content SET last_accessed_at = CURRENT_TIMESTAMP
|
||||
WHERE cache_key = ?
|
||||
`, string(cacheKey))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write %d access times: %w", len(served), err)
|
||||
return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
|
||||
}
|
||||
}
|
||||
|
||||
for sourceKey := range sources {
|
||||
_, err = tx.ExecContext(ctx, `
|
||||
UPDATE source_content SET last_accessed_at = CURRENT_TIMESTAMP
|
||||
WHERE source_key = ?
|
||||
`, string(sourceKey))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
|
||||
}
|
||||
}
|
||||
|
||||
err = tx.Commit()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write %d access times: %w", len(served), err)
|
||||
return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
|
||||
@@ -1,9 +1,12 @@
|
||||
package imgcache
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -169,14 +172,10 @@ func TestIncrementStats_ManyCountsPastTheirDeadlineLeaveOneWriteWaiting(t *testi
|
||||
|
||||
waitsBefore := cache.db.Stats().WaitCount
|
||||
|
||||
// Writes given a context past its deadline fail at once.
|
||||
ended, cancel := context.WithDeadline(t.Context(), time.Now())
|
||||
defer cancel()
|
||||
|
||||
const misses = 20
|
||||
|
||||
for range misses {
|
||||
cache.IncrementStats(ended, false, 0)
|
||||
cache.IncrementStats(false, 0)
|
||||
}
|
||||
|
||||
// Once one write waits for the connection, any others start waiting too.
|
||||
@@ -209,13 +208,9 @@ func TestStats_IncludesPendingCounts(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Writes given a context past its deadline fail at once.
|
||||
ended, cancel := context.WithDeadline(t.Context(), time.Now())
|
||||
defer cancel()
|
||||
|
||||
cache.IncrementStats(ended, true, 0)
|
||||
cache.IncrementStats(ended, true, 0)
|
||||
cache.IncrementStats(ended, false, 0)
|
||||
cache.IncrementStats(true, 0)
|
||||
cache.IncrementStats(true, 0)
|
||||
cache.IncrementStats(false, 0)
|
||||
|
||||
stats, err := cache.Stats(t.Context())
|
||||
if err != nil {
|
||||
@@ -251,19 +246,15 @@ func TestStopPendingCountWrites_WritesPendingCounts(t *testing.T) {
|
||||
|
||||
waitsBefore := cache.db.Stats().WaitCount
|
||||
|
||||
// Writes given a context past its deadline fail at once.
|
||||
ended, cancel := context.WithDeadline(t.Context(), time.Now())
|
||||
defer cancel()
|
||||
|
||||
cache.IncrementStats(ended, true, 0)
|
||||
cache.IncrementStats(true, 0)
|
||||
|
||||
// The goroutine's write waits for the connection.
|
||||
waitForConnectionWaits(t, cache.db, waitsBefore+1)
|
||||
|
||||
// Counted after that write read the pending counts
|
||||
cache.IncrementStats(ended, false, 0)
|
||||
cache.IncrementUpstreamFetch(ended, 1024)
|
||||
cache.IncrementTransformCount(ended)
|
||||
cache.IncrementStats(false, 0)
|
||||
cache.IncrementUpstreamFetch(1024)
|
||||
cache.IncrementTransformCount()
|
||||
|
||||
// The goroutine is told to finish here as well as by the stop, so that it
|
||||
// is told before its write under way ends, however late the stop runs.
|
||||
@@ -324,11 +315,7 @@ func TestStopPendingCountWrites_ReturnsWhenItsContextEnds(t *testing.T) {
|
||||
|
||||
waitsBefore := cache.db.Stats().WaitCount
|
||||
|
||||
// Writes given a context past its deadline fail at once.
|
||||
ended, cancel := context.WithDeadline(t.Context(), time.Now())
|
||||
defer cancel()
|
||||
|
||||
cache.IncrementStats(ended, true, 0)
|
||||
cache.IncrementStats(true, 0)
|
||||
|
||||
// The goroutine's write waits for the connection.
|
||||
waitForConnectionWaits(t, cache.db, waitsBefore+1)
|
||||
@@ -356,3 +343,88 @@ func TestStopPendingCountWrites_ReturnsWhenItsContextEnds(t *testing.T) {
|
||||
|
||||
waitForCounters(t, cache, cacheStatsCounters{hitCount: 1})
|
||||
}
|
||||
|
||||
// TestLookupSource_HandsAccessTimeToTheBackgroundWriter finds a cached source
|
||||
// whose access time is an hour old. LookupSource must not write the access
|
||||
// time itself but hand it over to be written, and StopPendingCountWrites must
|
||||
// then write it.
|
||||
func TestLookupSource_HandsAccessTimeToTheBackgroundWriter(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
key := storeEvictionTestSource(t, cache, "example.com", "/photo.jpg",
|
||||
[]byte("source bytes"))
|
||||
anHourAgo := time.Now().Add(-time.Hour)
|
||||
setSourceLastAccessed(t, cache, key, anHourAgo)
|
||||
|
||||
touchedQuery := `
|
||||
SELECT COUNT(*) FROM source_content
|
||||
WHERE source_key = ? AND last_accessed_at > ?
|
||||
`
|
||||
anHourAgoInDatabase := anHourAgo.UTC().Format(sqliteTimestampFormat)
|
||||
|
||||
found, _, err := cache.LookupSource(t.Context(), &imagerequest.Request{
|
||||
Source: imagerequest.Source{Host: "example.com", Path: "/photo.jpg"},
|
||||
})
|
||||
if err != nil || found != key {
|
||||
t.Fatalf("LookupSource() = %q, %v; want %q", found, err, key)
|
||||
}
|
||||
|
||||
if countRows(t, cache, touchedQuery, string(key), anHourAgoInDatabase) != 0 {
|
||||
t.Error("LookupSource wrote the source's access time itself")
|
||||
}
|
||||
|
||||
if waiting := len(cache.pendingSourceAccessTimes); waiting != 1 {
|
||||
t.Errorf("%d access times of sources wait to be written, want 1", waiting)
|
||||
}
|
||||
|
||||
err = cache.StopPendingCountWrites(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("StopPendingCountWrites() error = %v", err)
|
||||
}
|
||||
|
||||
if countRows(t, cache, touchedQuery, string(key), anHourAgoInDatabase) != 1 {
|
||||
t.Error("the source's access time was not written")
|
||||
}
|
||||
}
|
||||
|
||||
// TestStopPendingCountWrites_LogsWhatWasNotWritten makes the write of the
|
||||
// access times fail at shutdown, by dropping the table they are written to,
|
||||
// while a hit count is written. The stop must return an error and log that
|
||||
// access times were not written, and must not log the counts as not written.
|
||||
func TestStopPendingCountWrites_LogsWhatWasNotWritten(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
|
||||
var logs bytes.Buffer
|
||||
|
||||
cache.log = slog.New(slog.NewJSONHandler(&logs, nil))
|
||||
|
||||
cache.touchVariant(testVariantKeyOne)
|
||||
cache.IncrementStats(true, 0)
|
||||
|
||||
_, err := cache.db.ExecContext(t.Context(), `DROP TABLE variant_content`)
|
||||
if err != nil {
|
||||
t.Fatalf("dropping variant_content: %v", err)
|
||||
}
|
||||
|
||||
err = cache.StopPendingCountWrites(t.Context())
|
||||
t.Logf("StopPendingCountWrites() error = %v; logged:\n%s", err, logs.String())
|
||||
|
||||
if err == nil {
|
||||
t.Error("StopPendingCountWrites() error = nil, want the access times' error")
|
||||
}
|
||||
|
||||
if !strings.Contains(logs.String(), "access times not written at shutdown") {
|
||||
t.Error("the access times not written were not logged")
|
||||
}
|
||||
|
||||
if strings.Contains(logs.String(), "counts not written at shutdown") {
|
||||
t.Error("the counts, which were written, were logged as not written")
|
||||
}
|
||||
|
||||
if got := readCacheStatsCounters(t, cache); got.hitCount != 1 {
|
||||
t.Errorf("hit count in the database = %d, want 1", got.hitCount)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -157,7 +157,7 @@ func (s *Service) Get(
|
||||
reader, size, contentType, err := s.cache.GetVariant(cacheKey)
|
||||
if err == nil {
|
||||
s.cache.touchVariant(cacheKey)
|
||||
s.cache.IncrementStats(ctx, true, 0)
|
||||
s.cache.IncrementStats(true, 0)
|
||||
|
||||
return &ImageResponse{
|
||||
Content: reader,
|
||||
@@ -178,7 +178,7 @@ func (s *Service) Get(
|
||||
// failed or the request context has ended meanwhile
|
||||
response, err := s.processOrWait(ctx, req)
|
||||
|
||||
s.cache.IncrementStats(ctx, false, 0)
|
||||
s.cache.IncrementStats(false, 0)
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -475,7 +475,7 @@ func (s *Service) fetchAndProcess(
|
||||
fetchBytes := int64(len(sourceData))
|
||||
|
||||
// Counted also when the request context has ended meanwhile
|
||||
s.cache.IncrementUpstreamFetch(ctx, fetchBytes)
|
||||
s.cache.IncrementUpstreamFetch(fetchBytes)
|
||||
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read upstream response: %w", err)
|
||||
@@ -576,7 +576,7 @@ func (s *Service) processAndStore(
|
||||
processDuration := time.Since(processStart)
|
||||
|
||||
// Counted also when the request context has ended meanwhile
|
||||
s.cache.IncrementTransformCount(ctx)
|
||||
s.cache.IncrementTransformCount()
|
||||
|
||||
// Read processed content
|
||||
processedData, err := io.ReadAll(processResult.Content)
|
||||
|
||||
+5
-8
@@ -2,10 +2,10 @@
|
||||
# script/bench: run the Go benchmarks of internal/handlers, which serve
|
||||
# cache hits in parallel through the image handler and report the hits
|
||||
# served per second. A benchmark, not a check: script/check does not run
|
||||
# it. It builds the bench phase of the Dockerfile alone, as script/test
|
||||
# builds the test phase, with plain progress output so that the
|
||||
# benchmark's results are shown. A new CACHEBUST value reruns the
|
||||
# benchmark on every build. --output type=cacheonly writes no image.
|
||||
# it. It runs Go on the host, as script/loadtest does, and compiles pixa
|
||||
# there with CGO, so it needs the C compiler and libraries that
|
||||
# `script/bootstrap --cgo` installs. -run '^$' runs no test, only the
|
||||
# benchmarks.
|
||||
set -eu
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd -P)"
|
||||
@@ -13,10 +13,7 @@ ROOT="$(cd "$SCRIPT_DIR/.." && pwd -P)"
|
||||
|
||||
main() {
|
||||
cd "$ROOT"
|
||||
docker build --build-arg CACHEBUST="$(date +%s)-$$" \
|
||||
--progress=plain \
|
||||
--target bench \
|
||||
--output type=cacheonly .
|
||||
CGO_ENABLED=1 go test -run '^$' -bench . ./internal/handlers
|
||||
}
|
||||
|
||||
main "$@"
|
||||
|
||||
Reference in New Issue
Block a user