Remove idle host semaphores and delete .meta with its variant (closes #87)

Each upstream host's semaphore now counts the fetches holding or waiting
for one of its slots, and is removed from hostSems when the last of them
gives its slot back or stops waiting, so a long-running pixad no longer
keeps one semaphore per host it ever fetched from. The semLen test helper
reads hostSems directly, as getHostSemaphore now counts its caller.

VariantStorage.Delete removes the variant's .meta file too, a missing one
not being an error; DeleteWithMeta, which eviction called for that, is
gone.

Model: opus-5-5

TODO.md

@@ -36,17 +36,6 @@ P2: security: referer blacklist
   semaphore each until restart; `VariantStorage.Delete` removes the variant's
   `.meta` file along with it, a missing `.meta` file not being an error, and
   `DeleteWithMeta`, which eviction called for that, is gone.
-- 2026-10-03 shutdown sets the exit code and waits for image processing
-  (closes #86): fx alone handles SIGINT and SIGTERM, and the server's own
-  signal handler is gone; fx's `Run` in `cmd/pixad` exits with the shutdown's
-  code: 0 for a signal, 1 when the HTTP server cannot listen or the app fails
-  to start or to stop; the server's stop hook, which fx waits for, stops the
-  HTTP server, waits for the images still being processed, both within 5
-  seconds, then flushes Sentry; images still being processed after that are
-  logged with their count and make the exit code 1; a Sentry DSN that cannot be
-  used fails startup, so the stop hooks of what had already started run,
-  instead of exiting the process from a goroutine; the eviction loop is left to
-  #102.
 - 2026-10-03 every `script/cibuild` and `script/docker` run executes the checks
   (closes #101): the `Dockerfile` declares `CHECK_EPOCH` above `make fmt-check`
   and `make lint` in the lint stage and above `make test` in the build stage,

cmd/pixad/main.go

@@ -4,8 +4,6 @@ package main
 import (
 	"fmt"
 	"os"
-	"os/signal"
-	"syscall"
 
 	"github.com/spf13/cobra"
 	"go.uber.org/fx"
@@ -47,9 +45,6 @@ func run(_ *cobra.Command, _ []string) {
 		_ = os.Setenv("PIXA_CONFIG_PATH", configPath)
 	}
 
-	// A write to a closed stdout or stderr must not end the process.
-	signal.Ignore(syscall.SIGPIPE)
-
 	fx.New(
 		fx.Provide(
 			config.New,

internal/handlers/handlers.go

@@ -81,12 +81,6 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
 	return s, nil
 }
 
-// WaitForProcessing waits until no image is being processed, or until ctx
-// ends, and returns how many images were still being processed then.
-func (s *Handlers) WaitForProcessing(ctx context.Context) int {
-	return s.imgSvc.WaitForProcessing(ctx)
-}
-
 // initImageService initializes the image cache and service.
 func (s *Handlers) initImageService() error {
 	// Create the cache. cache_max_bytes: 0 disables the disk cache

internal/httpfetcher/max_connections_internal_test.go

@@ -153,11 +153,9 @@ func TestFetchRemovesIdleHostSemaphores(t *testing.T) {
 }
 
 // TestFetchRemovesHostSemaphoreWhenNoConnection checks that a fetch that
-// ends without a connection leaves no semaphore behind: when it is refused
-// after waiting for a connection shared by all hosts, when its context ends
-// while it waits for its host's slot, and when its context ends while it
-// waits for a connection shared by all hosts, long before the 10 second
-// wait timeout.
+// ends without a connection leaves no semaphore behind, both when it is
+// refused after waiting for a connection shared by all hosts and when its
+// context ends while it waits for its host's slot.
 func TestFetchRemovesHostSemaphoreWhenNoConnection(t *testing.T) {
 	t.Parallel()
 
@@ -190,18 +188,6 @@ func TestFetchRemovesHostSemaphoreWhenNoConnection(t *testing.T) {
 			"want context.DeadlineExceeded", err)
 	}
 
-	// Back to the 10 second wait, so the next fetch's context ends first.
-	f.connectionWaitTimeout = ConnectionWaitTimeout
-
-	ctx, cancel = context.WithTimeout(t.Context(), 100*time.Millisecond)
-	defer cancel()
-
-	_, err = f.Fetch(ctx, imageURLOnPort(83))
-	if !errors.Is(err, context.DeadlineExceeded) {
-		t.Fatalf("Fetch() from a host with nothing open: error = %v, "+
-			"want context.DeadlineExceeded", err)
-	}
-
 	err = open.Content.Close()
 	if err != nil {
 		t.Fatalf("close first body: %v", err)

internal/imageprocessor/imageprocessor.go

@@ -331,31 +331,6 @@ func FormatToMIME(format Format) string {
 	}
 }
 
-// WaitForProcessing waits until no image is being processed, or until ctx
-// ends, and returns how many images were still being processed then. It
-// waits by taking each slot in processingSemaphore as it frees up until it
-// holds them all, or until ctx ends, then gives back the slots it took.
-func (p *ImageProcessor) WaitForProcessing(ctx context.Context) int {
-	taken := 0
-
-	defer func() {
-		for range taken {
-			<-p.processingSemaphore
-		}
-	}()
-
-	for taken < cap(p.processingSemaphore) {
-		select {
-		case p.processingSemaphore <- struct{}{}:
-			taken++
-		case <-ctx.Done():
-			return len(p.processingSemaphore) - taken
-		}
-	}
-
-	return 0
-}
-
 // acquireSlot takes a slot in processingSemaphore, waiting at most
 // processingWaitTimeout for one to free up, and returns the func that gives
 // it back. A free slot is taken even when ctx has ended; only the wait for

internal/imageprocessor/max_concurrent_processing_internal_test.go

@@ -300,69 +300,3 @@ func TestProcessReleasesSlotOnError(t *testing.T) {
 		})
 	}
 }
-
-// TestWaitForProcessing holds a processing slot with a Process call that
-// cannot finish reading its input. WaitForProcessing must report that image
-// when its context ends first, wait for it otherwise, return 0 once it has
-// finished, and give back the slots it took while waiting.
-func TestWaitForProcessing(t *testing.T) {
-	t.Parallel()
-
-	proc := New(Params{MaxConcurrentProcessing: 2})
-
-	gate := make(chan struct{})
-	entered := make(chan struct{}, 1)
-	results := make(chan error, 1)
-
-	openGate := sync.OnceFunc(func() { close(gate) })
-	t.Cleanup(openGate)
-
-	processInBackground(proc, &gatedReader{
-		data: bytes.NewReader(createTestJPEG(t, 10, 10)), gate: gate,
-		entered: entered, counter: &readingCounter{},
-	}, results)
-	waitForEntries(t, entered, 1)
-
-	ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
-	defer cancel()
-
-	stillProcessing := proc.WaitForProcessing(ctx)
-	t.Logf("WaitForProcessing() after its context ended: %d", stillProcessing)
-
-	if stillProcessing != 1 {
-		t.Errorf("WaitForProcessing() after its context ended = %d, want 1",
-			stillProcessing)
-	}
-
-	waited := make(chan int, 1)
-
-	go func() { waited <- proc.WaitForProcessing(t.Context()) }()
-
-	select {
-	case got := <-waited:
-		t.Fatalf("WaitForProcessing() = %d while an image was being processed",
-			got)
-	case <-time.After(100 * time.Millisecond):
-	}
-
-	openGate()
-
-	err := <-results
-	if err != nil {
-		t.Errorf("Process() error = %v, want nil", err)
-	}
-
-	select {
-	case got := <-waited:
-		if got != 0 {
-			t.Errorf("WaitForProcessing() once processing finished = %d, want 0",
-				got)
-		}
-	case <-time.After(5 * time.Second):
-		t.Fatal("WaitForProcessing() did not return once processing finished")
-	}
-
-	if held := len(proc.processingSemaphore); held != 0 {
-		t.Errorf("%d slots still held after WaitForProcessing() returned", held)
-	}
-}

internal/imgcache/service.go

@@ -199,12 +199,6 @@ func (s *Service) Stats(ctx context.Context) (*CacheStats, error) {
 	return s.cache.Stats(ctx)
 }
 
-// WaitForProcessing waits until no image is being processed, or until ctx
-// ends, and returns how many images were still being processed then.
-func (s *Service) WaitForProcessing(ctx context.Context) int {
-	return s.processor.WaitForProcessing(ctx)
-}
-
 // ValidateRequest validates the request signature if required.
 func (s *Service) ValidateRequest(req *ImageRequest) error {
 	// Check if host is allowed (no signature required)

internal/server/http.go

@@ -5,8 +5,6 @@ import (
 	"fmt"
 	"net/http"
 	"time"
-
-	"go.uber.org/fx"
 )
 
 // HTTP server configuration constants.
@@ -38,19 +36,19 @@ func (s *Server) newHTTPServer() *http.Server {
 	}
 }
 
-// serveUntilShutdown serves on s.httpServer until it is shut down. When it
-// stops for any other reason, such as its port being in use, it asks fx to
-// shut down with exit code 1.
 func (s *Server) serveUntilShutdown() {
+	s.httpServer = s.newHTTPServer()
+
+	s.SetupRoutes()
+
 	s.log.Info("http begin listen", "listenaddr", s.httpServer.Addr)
 
 	err := s.httpServer.ListenAndServe()
 	if err != nil && !errors.Is(err, http.ErrServerClosed) {
 		s.log.Error("listen error", "error", err)
 
-		err = s.shutdowner.Shutdown(fx.ExitCode(1))
-		if err != nil {
-			s.log.Error("shutdown request failed", "error", err)
+		if s.cancelFunc != nil {
+			s.cancelFunc()
 		}
 	}
 }

internal/server/server.go

@@ -3,10 +3,12 @@ package server
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"log/slog"
 	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
 	"time"
 
 	"github.com/getsentry/sentry-go"
@@ -25,10 +27,6 @@ const (
 	SentryFlushTimeout = 2 * time.Second
 )
 
-// errStillProcessing is returned by the server's stop hook when images are
-// still being processed once ShutdownTimeout has passed.
-var errStillProcessing = errors.New("images still being processed at shutdown")
-
 // Params defines dependencies for Server.
 type Params struct {
 	fx.In
@@ -38,7 +36,6 @@ type Params struct {
 	Config     *config.Config
 	Middleware *middleware.Middleware
 	Handlers   *handlers.Handlers
-	Shutdowner fx.Shutdowner
 }
 
 // Server is the main HTTP server.
@@ -48,58 +45,59 @@ type Server struct {
 	globals       *globals.Globals
 	mw            *middleware.Middleware
 	h             *handlers.Handlers
-	shutdowner    fx.Shutdowner
 	startupTime   time.Time
+	exitCode      int
 	sentryEnabled bool
+	cancelFunc    context.CancelFunc
 	httpServer    *http.Server
 	router        *chi.Mux
 }
 
-// New creates a new Server instance. Its start hook starts Sentry and the
-// HTTP server; its stop hook, which fx runs on SIGINT, SIGTERM or a
-// shutdown request, shuts them down.
+// New creates a new Server instance.
 func New(lc fx.Lifecycle, params Params) (*Server, error) {
 	s := &Server{
-		log:        params.Logger.Get(),
-		config:     params.Config,
-		globals:    params.Globals,
-		mw:         params.Middleware,
-		h:          params.Handlers,
-		shutdowner: params.Shutdowner,
+		log:     params.Logger.Get(),
+		config:  params.Config,
+		globals: params.Globals,
+		mw:      params.Middleware,
+		h:       params.Handlers,
 	}
 
 	lc.Append(fx.Hook{
-		OnStart: func(_ context.Context) error {
+		OnStart: func(ctx context.Context) error {
 			s.startupTime = time.Now()
+			go s.Run(context.WithoutCancel(ctx))
 
-			err := s.enableSentry()
-			if err != nil {
-				return err
+			return nil
+		},
+		OnStop: func(_ context.Context) error {
+			if s.cancelFunc != nil {
+				s.cancelFunc()
 			}
-
-			s.SetupRoutes()
-			s.httpServer = s.newHTTPServer()
-
-			go s.serveUntilShutdown()
 
 			return nil
 		},
-		OnStop: s.cleanShutdown,
 	})
 
 	return s, nil
 }
 
+// Run starts the server.
+func (s *Server) Run(ctx context.Context) {
+	s.enableSentry()
+	s.serve(ctx)
+}
+
 // MaintenanceMode returns whether maintenance mode is enabled.
 func (s *Server) MaintenanceMode() bool {
 	return s.config.MaintenanceMode
 }
 
-func (s *Server) enableSentry() error {
+func (s *Server) enableSentry() {
 	s.sentryEnabled = false
 
 	if s.config.SentryDSN == "" {
-		return nil
+		return
 	}
 
 	err := sentry.Init(sentry.ClientOptions{
@@ -107,42 +105,55 @@ func (s *Server) enableSentry() error {
 		Release: fmt.Sprintf("%s-%s", s.globals.Appname, s.globals.Version),
 	})
 	if err != nil {
-		return fmt.Errorf("sentry init failure: %w", err)
+		s.log.Error("sentry init failure", "error", err)
+		os.Exit(1)
 	}
 
 	s.log.Info("sentry error reporting activated")
 	s.sentryEnabled = true
-
-	return nil
 }
 
-// cleanShutdown stops the HTTP server, waits for the images still being
-// processed, then flushes Sentry. The first two share ShutdownTimeout. It
-// returns errStillProcessing when images are still being processed after
-// that, as their work is abandoned.
-func (s *Server) cleanShutdown(ctx context.Context) error {
-	s.log.Info("shutting down")
+func (s *Server) serve(ctx context.Context) int {
+	ctx, cancelFunc := context.WithCancel(ctx)
+	s.cancelFunc = cancelFunc
 
-	ctxShutdown, shutdownCancel := context.WithTimeout(ctx, ShutdownTimeout)
+	go func() {
+		c := make(chan os.Signal, 1)
+
+		signal.Ignore(syscall.SIGPIPE)
+		signal.Notify(c, os.Interrupt, syscall.SIGTERM)
+
+		sig := <-c
+		s.log.Info("signal received", "signal", sig)
+
+		if s.cancelFunc != nil {
+			s.cancelFunc()
+		}
+	}()
+
+	go s.serveUntilShutdown()
+
+	<-ctx.Done()
+	s.cleanShutdown(ctx)
+
+	return s.exitCode
+}
+
+func (s *Server) cleanShutdown(ctx context.Context) {
+	s.exitCode = 0
+
+	ctxShutdown, shutdownCancel := context.WithTimeout(
+		context.WithoutCancel(ctx), ShutdownTimeout)
 	defer shutdownCancel()
 
-	err := s.httpServer.Shutdown(ctxShutdown)
-	if err != nil {
-		s.log.Error("server clean shutdown failed", "error", err)
+	if s.httpServer != nil {
+		err := s.httpServer.Shutdown(ctxShutdown)
+		if err != nil {
+			s.log.Error("server clean shutdown failed", "error", err)
+		}
 	}
 
-	stillProcessing := s.h.WaitForProcessing(ctxShutdown)
-
 	if s.sentryEnabled {
 		sentry.Flush(SentryFlushTimeout)
 	}
-
-	if stillProcessing > 0 {
-		s.log.Error("images still being processed at shutdown",
-			"count", stillProcessing)
-
-		return errStillProcessing
-	}
-
-	return nil
 }

internal/server/shutdown_internal_test.go

@@ -1,96 +0,0 @@
-package server
-
-import (
-	"log/slog"
-	"net"
-	"testing"
-	"time"
-
-	"go.uber.org/fx"
-	"go.uber.org/fx/fxtest"
-
-	"sneak.berlin/go/pixa/internal/config"
-	"sneak.berlin/go/pixa/internal/globals"
-	"sneak.berlin/go/pixa/internal/logger"
-)
-
-// shutdownRecorder is an fx.Shutdowner that sends the options of each
-// shutdown request on requests.
-type shutdownRecorder struct {
-	requests chan []fx.ShutdownOption
-}
-
-func (r shutdownRecorder) Shutdown(opts ...fx.ShutdownOption) error {
-	r.requests <- opts
-
-	return nil
-}
-
-// TestSentryInitFailureFailsStartup checks that a Sentry DSN that cannot be
-// used makes the server's start hook fail, so fx stops what has already
-// started, instead of the process exiting from a goroutine.
-func TestSentryInitFailureFailsStartup(t *testing.T) {
-	t.Parallel()
-
-	lc := fxtest.NewLifecycle(t)
-
-	log, err := logger.New(lc, logger.Params{Globals: &globals.Globals{}})
-	if err != nil {
-		t.Fatalf("logger.New() error = %v", err)
-	}
-
-	_, err = New(lc, Params{
-		Logger:  log,
-		Globals: &globals.Globals{Appname: "pixad"},
-		Config:  &config.Config{SentryDSN: "not-a-dsn"},
-	})
-	if err != nil {
-		t.Fatalf("New() error = %v", err)
-	}
-
-	err = lc.Start(t.Context())
-	t.Logf("Start() error = %v", err)
-
-	if err == nil {
-		t.Fatal("Start() error = nil, want the Sentry initialization error")
-	}
-}
-
-// TestListenErrorRequestsShutdownWithExitCode1 occupies the server's port
-// and checks that the listen error asks fx to shut down with exit code 1.
-func TestListenErrorRequestsShutdownWithExitCode1(t *testing.T) {
-	t.Parallel()
-
-	busy, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", ":0")
-	if err != nil {
-		t.Fatalf("Listen() error = %v", err)
-	}
-
-	t.Cleanup(func() { _ = busy.Close() })
-
-	addr, ok := busy.Addr().(*net.TCPAddr)
-	if !ok {
-		t.Fatalf("listener address %v is not a TCP address", busy.Addr())
-	}
-
-	requests := make(chan []fx.ShutdownOption, 1)
-	s := &Server{
-		log:        slog.New(slog.DiscardHandler),
-		config:     &config.Config{Port: addr.Port},
-		shutdowner: shutdownRecorder{requests: requests},
-	}
-	s.httpServer = s.newHTTPServer()
-
-	go s.serveUntilShutdown()
-
-	select {
-	case opts := <-requests:
-		t.Logf("shutdown options = %v", opts)
-
-		if len(opts) != 1 || opts[0] != fx.ExitCode(1) {
-			t.Errorf("shutdown options = %v, want [fx.ExitCode(1)]", opts)
-		}
-	case <-time.After(5 * time.Second):
-		t.Fatal("no shutdown was requested after the listen error")
-	}
-}