1 Commits
Author SHA1 Message Date
sneak bd700650e0 Run at most 4 tests of a package at once in the test phase (closes #224)
check / check (push) Canceled after 0s
On a busy host, TestService_Get_ReturnsByItsDeadline failed because its
request returned over a second after its 200 ms deadline. The request
does not ignore its deadline: with as many tests running at once in the
package's test process as the host has CPUs, it waited to be scheduled,
or for SQLite calls queued on a lock inside the SQLite library that
every database in the process shares. The test phase now passes
-parallel 4 to go test. -p is unchanged: capping it also caps
compiling, which made the step much slower.

Model: opus-5-5
2026-10-08 12:20:17 +00:00
7 changed files with 28 additions and 228 deletions
+4 -3
View File
@@ -33,9 +33,10 @@ COPY . .
# Without -v first; on a failure, again with -v for the details, and # Without -v first; on a failure, again with -v for the details, and
# the step fails even if the second run passes. -parallel 4: by default # the step fails even if the second run passes. -parallel 4: by default
# a package runs as many of its tests at once as the host has CPUs, and # a package runs as many of its tests at once as the host has CPUs. On
# on a busy host they then wait so long to be scheduled that a test # a busy host they then wait to be scheduled, and their SQLite calls
# that times a request can see it return a second late. # queue on one lock that every database in the test process shares, so
# a test that times a request can see it return a second late.
RUN go test -count=1 -timeout 90s -race -parallel 4 -cover ./... || \ RUN go test -count=1 -timeout 90s -race -parallel 4 -cover ./... || \
{ echo "--- Rerunning with -v for details ---"; \ { echo "--- Rerunning with -v for details ---"; \
go test -count=1 -timeout 90s -race -parallel 4 -v ./...; exit 1; } go test -count=1 -timeout 90s -race -parallel 4 -v ./...; exit 1; }
+4 -6
View File
@@ -83,12 +83,10 @@ which answers 200 whenever pixa is running, in maintenance mode too (see
`maintenance_mode`). `maintenance_mode`).
On SIGTERM or SIGINT pixa stops accepting connections, gives the requests in On SIGTERM or SIGINT pixa stops accepting connections, gives the requests in
progress, the images being processed and the counts of cache hits, misses, progress and the images being processed 5 seconds to finish, and exits: with 0,
fetches and conversions being written to the database 5 seconds to finish, and or with 1 when images were still being processed after those 5 seconds or
exits: with 0, or with 1 when images were still being processed or counts still another part of pixa failed to stop. A request not finished by then is cut off.
being written after those 5 seconds, or another part of pixa failed to stop. A `docker stop` waits 10 seconds before it kills the container.
request not finished by then is cut off. `docker stop` waits 10 seconds before
it kills the container.
Outside Docker, pixa needs libvips (the image has 8.16) and libheif to run, as Outside Docker, pixa needs libvips (the image has 8.16) and libheif to run, as
it uses libvips through CGO. pixad does not start unless libvips has its JPEG XL it uses libvips through CGO. pixad does not start unless libvips has its JPEG XL
+8 -15
View File
@@ -30,21 +30,14 @@ P2: security: per-IP rate limiting on the image routes
# Completed Steps # Completed Steps
- 2026-10-08 a request past its deadline no longer waits for a database write - 2026-10-08 the test phase of the `Dockerfile` runs at most 4 tests of a
(closes #224). On a busy host, `TestService_Get_ReturnsByItsDeadline` failed package at once (`go test -parallel 4`, closes #224), where it ran as many as
because its request returned over a second after its deadline: it was still the host has CPUs. On a busy host, `TestService_Get_ReturnsByItsDeadline`
waiting for its miss count to be written to the database, a write with no failed because its request returned over a second after its deadline: the
deadline, which in pixad waits for the one database connection that every request was waiting to be scheduled, or for SQLite calls queued on a lock
request shares. The hit, miss, upstream fetch and transform counts are now inside the SQLite library that every database in the test process shares.
each written in a goroutine of their own, which the request waits for only `-p`, how many packages are tested at once, is unchanged, as capping it also
until its deadline; a count not written by then is written after the request slows compiling.
returns. On shutdown pixad waits for those writes within the same 5 seconds as
for the images being processed, and exits with 1 when some are unfinished. The
test phase of the `Dockerfile` also runs at most 4 tests of a package at once
(`go test -parallel 4`), where it ran as many as the host has CPUs: on a busy
host they then wait so long to be scheduled that a test that times a request
can fail. `-p`, how many packages are tested at once, is unchanged, as capping
it also slows compiling.
- 2026-10-08 every output format is saved with settings pixa sets on purpose - 2026-10-08 every output format is saved with settings pixa sets on purpose
(closes #232): each format has its own govips export, as JPEG XL does, in (closes #232): each format has its own govips export, as JPEG XL does, in
place of govips' generic `Export`, which sent libvips a zero for some settings place of govips' generic `Export`, which sent libvips a zero for some settings
-7
View File
@@ -93,13 +93,6 @@ func (s *Handlers) WaitForProcessing(ctx context.Context) int {
return s.imgSvc.WaitForProcessing(ctx) return s.imgSvc.WaitForProcessing(ctx)
} }
// WaitForCountWrites waits until every count a request has started writing
// to the database is written, or until ctx ends, and reports whether they
// all were.
func (s *Handlers) WaitForCountWrites(ctx context.Context) bool {
return s.imgSvc.WaitForCountWrites(ctx)
}
// newCacheConfig builds the image cache's configuration from cfg. // newCacheConfig builds the image cache's configuration from cfg.
// cache_max_bytes: 0 disables the disk cache entirely; any other value // cache_max_bytes: 0 disables the disk cache entirely; any other value
// is the eviction limit in bytes; when it is omitted, the cache works // is the eviction limit in bytes; when it is omitted, the cache works
@@ -1,116 +0,0 @@
package imgcache
import (
"context"
"errors"
"testing"
"time"
)
// holdDatabase takes the one connection of the test service's database, so
// that every other query waits for it, and returns the func that frees it.
func holdDatabase(t *testing.T, svc *Service) func() {
t.Helper()
conn, err := svc.cache.db.Conn(t.Context())
if err != nil {
t.Fatalf("failed to take the database connection: %v", err)
}
release := func() { _ = conn.Close() }
t.Cleanup(release)
return release
}
// TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy holds the
// database's one connection while a request whose fetch is held reaches its
// deadline. The request must still return by its deadline with the
// deadline's error, and its miss must be counted once the connection is free.
func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) {
t.Parallel()
svc, fixtures, fetcher := setupHeldFetchService(t)
const timeout = 200 * time.Millisecond
ctx, cancel := context.WithTimeout(t.Context(), timeout)
defer cancel()
results := startGet(ctx, svc, photoVariant(fixtures, 85, FitCover))
// The request has made its database reads by the time it fetches.
<-fetcher.started
releaseDatabase := holdDatabase(t, svc)
select {
case got := <-results:
deadline, _ := ctx.Deadline()
t.Logf("Get() returned %v after its deadline, error = %v",
time.Since(deadline), got.err)
if !errors.Is(got.err, context.DeadlineExceeded) {
t.Errorf("Get() error = %v, want %v", got.err, context.DeadlineExceeded)
}
case <-time.After(timeout + time.Second):
t.Fatal("request did not return by its deadline while the database was busy")
}
releaseDatabase()
waitCtx, cancelWait := context.WithTimeout(t.Context(), 5*time.Second)
defer cancelWait()
if !svc.WaitForCountWrites(waitCtx) {
t.Fatal("the miss was not counted once the database was free")
}
want := cacheStatsCounters{missCount: 1}
if got := readCacheStatsCounters(t, svc.cache); got != want {
t.Errorf("counters = %+v, want %+v", got, want)
}
}
// TestService_WaitForCountWrites holds the database's one connection while a
// request past its deadline counts a miss. WaitForCountWrites must report the
// count unwritten when its context ends, and written once the connection is
// free.
func TestService_WaitForCountWrites(t *testing.T) {
t.Parallel()
svc, _, _ := setupHeldFetchService(t)
releaseDatabase := holdDatabase(t, svc)
ended, cancel := context.WithDeadline(t.Context(), time.Now())
defer cancel()
svc.writeCount(ended, func(writeCtx context.Context) {
svc.cache.IncrementStats(writeCtx, false, 0)
})
shortCtx, cancelShort := context.WithTimeout(t.Context(), 100*time.Millisecond)
defer cancelShort()
written := svc.WaitForCountWrites(shortCtx)
t.Logf("WaitForCountWrites() while the database was busy = %t", written)
if written {
t.Fatal("WaitForCountWrites() = true while the database was busy")
}
releaseDatabase()
waitCtx, cancelWait := context.WithTimeout(t.Context(), 5*time.Second)
defer cancelWait()
if !svc.WaitForCountWrites(waitCtx) {
t.Fatal("WaitForCountWrites() = false once the database was free")
}
want := cacheStatsCounters{missCount: 1}
if got := readCacheStatsCounters(t, svc.cache); got != want {
t.Errorf("counters = %+v, want %+v", got, want)
}
}
+7 -61
View File
@@ -9,7 +9,6 @@ import (
"log/slog" "log/slog"
"net/url" "net/url"
"runtime/debug" "runtime/debug"
"sync"
"time" "time"
"github.com/dustin/go-humanize" "github.com/dustin/go-humanize"
@@ -37,9 +36,6 @@ type Service struct {
// variantsInProgress lets the requests that miss the same variant at the // variantsInProgress lets the requests that miss the same variant at the
// same time share one fetch and one transcode. // same time share one fetch and one transcode.
variantsInProgress singleflight.Group variantsInProgress singleflight.Group
// countWrites holds the count writes writeCount has started, for
// WaitForCountWrites.
countWrites sync.WaitGroup
} }
// ServiceConfig holds configuration for the image service. // ServiceConfig holds configuration for the image service.
@@ -165,9 +161,8 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e
s.log.Error("failed to get cached variant", "key", result.CacheKey, "error", err) s.log.Error("failed to get cached variant", "key", result.CacheKey, "error", err)
// Fall through to re-process // Fall through to re-process
} else { } else {
s.writeCount(ctx, func(writeCtx context.Context) { // Counted also when the request context has ended meanwhile
s.cache.IncrementStats(writeCtx, true, 0) s.cache.IncrementStats(context.WithoutCancel(ctx), true, 0)
})
return &ImageResponse{ return &ImageResponse{
Content: reader, Content: reader,
@@ -184,9 +179,7 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e
// failed or the request context has ended meanwhile // failed or the request context has ended meanwhile
response, err := s.processOrWait(ctx, req) response, err := s.processOrWait(ctx, req)
s.writeCount(ctx, func(writeCtx context.Context) { s.cache.IncrementStats(context.WithoutCancel(ctx), false, 0)
s.cache.IncrementStats(writeCtx, false, 0)
})
if err != nil { if err != nil {
return nil, err return nil, err
@@ -215,53 +208,6 @@ func (s *Service) WaitForProcessing(ctx context.Context) int {
return s.processor.WaitForProcessing(ctx) return s.processor.WaitForProcessing(ctx)
} }
// WaitForCountWrites waits until every count a request has started writing
// to the database is written, or until ctx ends, and reports whether they
// all were.
func (s *Service) WaitForCountWrites(ctx context.Context) bool {
written := make(chan struct{})
go func() {
s.countWrites.Wait()
close(written)
}()
select {
case <-written:
return true
case <-ctx.Done():
return false
}
}
// writeCount runs write, which adds to a count in the database, in a
// goroutine of its own, and waits for it, though not past ctx's deadline.
// The write gets ctx without its cancellation or deadline, so an ended
// request is still counted. A request past its deadline thus does not wait
// for the database connection, which every request shares; its count is
// written after it returns.
func (s *Service) writeCount(ctx context.Context, write func(context.Context)) {
written := make(chan struct{})
s.countWrites.Go(func() {
defer close(written)
write(context.WithoutCancel(ctx))
})
deadline, hasDeadline := ctx.Deadline()
if !hasDeadline {
<-written
return
}
select {
case <-written:
case <-time.After(time.Until(deadline)):
}
}
// ValidateRequest validates the request signature if required. // ValidateRequest validates the request signature if required.
func (s *Service) ValidateRequest(req *ImageRequest) error { func (s *Service) ValidateRequest(req *ImageRequest) error {
// Check if host is allowed (no signature required) // Check if host is allowed (no signature required)
@@ -506,9 +452,8 @@ func (s *Service) fetchAndProcess(
sourceData, err := io.ReadAll(fetchResult.Content) sourceData, err := io.ReadAll(fetchResult.Content)
fetchBytes := int64(len(sourceData)) fetchBytes := int64(len(sourceData))
s.writeCount(ctx, func(writeCtx context.Context) { // Counted also when the request context has ended meanwhile
s.cache.IncrementUpstreamFetch(writeCtx, fetchBytes) s.cache.IncrementUpstreamFetch(context.WithoutCancel(ctx), fetchBytes)
})
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to read upstream response: %w", err) return nil, fmt.Errorf("failed to read upstream response: %w", err)
@@ -589,7 +534,8 @@ func (s *Service) processAndStore(
processDuration := time.Since(processStart) processDuration := time.Since(processStart)
s.writeCount(ctx, s.cache.IncrementTransformCount) // Counted also when the request context has ended meanwhile
s.cache.IncrementTransformCount(context.WithoutCancel(ctx))
// Read processed content // Read processed content
processedData, err := io.ReadAll(processResult.Content) processedData, err := io.ReadAll(processResult.Content)
+5 -20
View File
@@ -29,10 +29,6 @@ const (
// still being processed once ShutdownTimeout has passed. // still being processed once ShutdownTimeout has passed.
var errStillProcessing = errors.New("images still being processed at shutdown") var errStillProcessing = errors.New("images still being processed at shutdown")
// errStillWritingCounts is returned by the server's stop hook when counts
// are still being written to the database once ShutdownTimeout has passed.
var errStillWritingCounts = errors.New("counts still being written at shutdown")
// Params defines dependencies for Server. // Params defines dependencies for Server.
type Params struct { type Params struct {
fx.In fx.In
@@ -121,11 +117,9 @@ func (s *Server) enableSentry() error {
} }
// cleanShutdown stops the HTTP server, waits for the images still being // cleanShutdown stops the HTTP server, waits for the images still being
// processed and then for the counts still being written to the database, // processed, then flushes Sentry. The first two share ShutdownTimeout. It
// which closes after this hook, then flushes Sentry. The first three share // returns errStillProcessing when images are still being processed after
// ShutdownTimeout. It returns errStillProcessing and errStillWritingCounts // that, as their work is abandoned.
// for the images and counts still unfinished after that, as their work is
// abandoned.
func (s *Server) cleanShutdown(ctx context.Context) error { func (s *Server) cleanShutdown(ctx context.Context) error {
s.log.Info("shutting down") s.log.Info("shutting down")
@@ -138,26 +132,17 @@ func (s *Server) cleanShutdown(ctx context.Context) error {
} }
stillProcessing := s.h.WaitForProcessing(ctxShutdown) stillProcessing := s.h.WaitForProcessing(ctxShutdown)
countsWritten := s.h.WaitForCountWrites(ctxShutdown)
if s.sentryEnabled { if s.sentryEnabled {
sentry.Flush(SentryFlushTimeout) sentry.Flush(SentryFlushTimeout)
} }
var unfinished []error
if stillProcessing > 0 { if stillProcessing > 0 {
s.log.Error("images still being processed at shutdown", s.log.Error("images still being processed at shutdown",
"count", stillProcessing) "count", stillProcessing)
unfinished = append(unfinished, errStillProcessing) return errStillProcessing
} }
if !countsWritten { return nil
s.log.Error("counts still being written at shutdown")
unfinished = append(unfinished, errStillWritingCounts)
}
return errors.Join(unfinished...)
} }