Exit with the shutdown's code and wait for image processing (closes #86)
check / check (push) Successful in 9m56s
check / check (push) Successful in 9m56s
fx alone handles SIGINT and SIGTERM; the server's own handler, which only cancelled a context that fx's stop did not wait for, is gone. fx's Run exits with the shutdown's code: the one a shutdown request carries, 0 for a signal, 1 when the app fails to start or stop. A listen error asks fx to shut down with exit code 1. A Sentry DSN that cannot be used fails the server's start hook, so fx stops what had already started. The server's stop hook stops the HTTP server, then waits for the images still being processed, both within ShutdownTimeout; images still being processed after that are logged and fail the stop, so the exit code is 1. Model: opus-5-5
This commit was merged in pull request #169.
This commit is contained in:
@@ -331,6 +331,31 @@ 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
|
||||
|
||||
@@ -300,3 +300,69 @@ 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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user