1 Commits
Author SHA1 Message Date
sneak df7fff0ce1 Stop waiting for count writes after a request's deadline (closes #224)
check / check (push) Canceled after 0s
TestService_Get_ReturnsByItsDeadline failed on a busy host because its
request, past its deadline, still waited for its miss count to be
written, a database write with no deadline; in pixad that write waits
for the one database connection every request shares. The hit, miss,
upstream fetch and transform counts are now each written in a goroutine
of their own, which the request waits for only until its deadline; a
count not written by then is written after it returns. Shutdown waits
for those writes within ShutdownTimeout and reports any left unfinished.
The test phase also runs go test with -parallel 4: on a busy host, as
many tests at once as there are CPUs wait so long to be scheduled that
a timed request can fail.

Model: opus-5-5
2026-10-08 13:29:06 +00:00
2 changed files with 36 additions and 40 deletions
@@ -86,8 +86,8 @@ func TestService_WaitForCountWrites(t *testing.T) {
ended, cancel := context.WithDeadline(t.Context(), time.Now())
defer cancel()
svc.writeCount(ended, func(ctx context.Context) {
svc.cache.IncrementStats(ctx, false, 0)
svc.writeCount(ended, func(writeCtx context.Context) {
svc.cache.IncrementStats(writeCtx, false, 0)
})
shortCtx, cancelShort := context.WithTimeout(t.Context(), 100*time.Millisecond)
+34 -38
View File
@@ -165,8 +165,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)
// Fall through to re-process
} else {
s.writeCount(ctx, func(ctx context.Context) {
s.cache.IncrementStats(ctx, true, 0)
s.writeCount(ctx, func(writeCtx context.Context) {
s.cache.IncrementStats(writeCtx, true, 0)
})
return &ImageResponse{
@@ -184,8 +184,8 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e
// failed or the request context has ended meanwhile
response, err := s.processOrWait(ctx, req)
s.writeCount(ctx, func(ctx context.Context) {
s.cache.IncrementStats(ctx, false, 0)
s.writeCount(ctx, func(writeCtx context.Context) {
s.cache.IncrementStats(writeCtx, false, 0)
})
if err != nil {
@@ -234,6 +234,34 @@ func (s *Service) WaitForCountWrites(ctx context.Context) bool {
}
}
// 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.
func (s *Service) ValidateRequest(req *ImageRequest) error {
// Check if host is allowed (no signature required)
@@ -280,38 +308,6 @@ func (s *Service) GenerateSignedURL(
baseURL, path, sig, exp, req.Quality, req.FitMode), nil
}
// writeCount runs write, which adds to a count in the database, in a
// goroutine of its own. The write gets ctx without its cancellation or
// deadline, so an ended request is still counted. writeCount waits for it
// until ctx's deadline, even if ctx is cancelled first, so a request
// returns with its count written unless its deadline has passed; past
// that, it does not wait for the database connection, which every request
// shares.
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
}
deadlineTimer := time.NewTimer(time.Until(deadline))
defer deadlineTimer.Stop()
select {
case <-written:
case <-deadlineTimer.C:
}
}
// errPanicked is returned when processing a variant panicked.
var errPanicked = errors.New("panic while processing image")
@@ -510,8 +506,8 @@ func (s *Service) fetchAndProcess(
sourceData, err := io.ReadAll(fetchResult.Content)
fetchBytes := int64(len(sourceData))
s.writeCount(ctx, func(ctx context.Context) {
s.cache.IncrementUpstreamFetch(ctx, fetchBytes)
s.writeCount(ctx, func(writeCtx context.Context) {
s.cache.IncrementUpstreamFetch(writeCtx, fetchBytes)
})
if err != nil {