amadan
Say what is true of a stalled log sink: close does not wait, the pump cannot be interrupted
Astra's round 4 on oficina/slides #20, carried here with the same queue: a write blocked inside the logger cannot be interrupted from Go, so the pump's and close's comments state the limit (it can only finish or die with the process; shutdown does not wait for it), and TestMailQueueCloseIsBoundedWithAStalledLogger asserts, with the sink STILL stalled, that close returned in its bound, the worker has exited and at most the pump remains (relative to baselines, since another test's queue may still be shutting down); then it releases the sink and asserts the pump exits and the count is back at baseline.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2 files changed,
+46
−14
internal/sheets/mailqueue.go+10 −1internal/sheets/mailqueue_test.go+36 −13
CI passed — run details
CI log
amadan: .amadan/timeout: this job's budget is 15m0s (this runner allows up to 30m0s) === vet === CGO_ENABLED=0 go vet ./... CGO_ENABLED=0 go vet -tags browser ./internal/sheets/ === fmt === === build === CGO_ENABLED=0 go build ./... === migrations === go tool rastrillo migration check rastrillo migration check: models and migrations agree === go-test === CGO_ENABLED=0 go test -count=1 ./... ? amadan.net/oficina/sheets/cmd/sheets [no test files] ok amadan.net/oficina/sheets/internal/guardstest 0.083s ok amadan.net/oficina/sheets/internal/sheets 3.993s ok amadan.net/oficina/sheets/internal/sheets/calc 0.009s ok amadan.net/oficina/sheets/internal/sheets/format 0.002s ? amadan.net/oficina/sheets/internal/sheets/keymailtest [no test files] ok amadan.net/oficina/sheets/internal/sheets/live 0.053s ok amadan.net/oficina/sheets/internal/sheets/seal 0.002s ok amadan.net/oficina/sheets/internal/sheets/sealtest 0.002s ok amadan.net/oficina/sheets/internal/sheets/store 2.956s ok amadan.net/oficina/sheets/internal/sheets/xlsx 0.300s ? amadan.net/oficina/sheets/tools/gencorpus [no test files] === race === go test -race -count=1 ./internal/sheets/store/... ./internal/sheets/live/... ok amadan.net/oficina/sheets/internal/sheets/store 46.495s ok amadan.net/oficina/sheets/internal/sheets/live 1.061s
diff --git a/internal/sheets/mailqueue.go b/internal/sheets/mailqueue.go| index 27d01f6..e807aac 100644 |
| --- a/internal/sheets/mailqueue.go |
| +++ b/internal/sheets/mailqueue.go |
| @@ -168,6 +168,13 @@ func (q *mailQueue) logf(msg string, args ...any) { |
| // pump is the only goroutine that writes to the logger. It ends when |
| // the worker closes logs on its way out. |
| +// |
| +// Its limit, stated rather than engineered around: a write blocked |
| +// inside the logger (a stalled sink) cannot be interrupted from Go. It |
| +// can only finish, or die with the process. Closing logs does not reach |
| +// it mid-write, so a stalled sink leaves this one goroutine behind |
| +// until the sink moves. Nothing else waits on it: not Send, not the |
| +// worker, not close. |
| func (q *mailQueue) pump() { |
| for l := range q.logs { |
| if n := q.lostLogs.Swap(0); n > 0 { |
| @@ -245,7 +252,9 @@ func (q *mailQueue) wait() { |
| // close stops taking links and drains what is queued, within |
| // drainTimeout; then it cancels, abandons the rest, and stops the |
| -// worker (see the type's comment). |
| +// worker (see the type's comment). It never waits on the log pump: |
| +// shutdown deliberately does not wait for a write blocked inside the |
| +// logger, which can only finish or die with the process (see pump). |
| func (q *mailQueue) close() { |
| q.mu.Lock() |
| if q.closed { |
diff --git a/internal/sheets/mailqueue_test.go b/internal/sheets/mailqueue_test.go| index 9120e65..d506fbe 100644 |
| --- a/internal/sheets/mailqueue_test.go |
| +++ b/internal/sheets/mailqueue_test.go |
| @@ -195,14 +195,31 @@ func mailWorkers() int { |
| } |
| // TestMailQueueCloseIsBoundedWithAStalledLogger: close reports through |
| -// nothing that can block. With the log sink stalled — every write hangs |
| -// — and a relay that never answers but honours cancellation, close |
| -// still returns within its bound and the worker has exited. Once the |
| -// sink is released, the goroutine count is back to its baseline: the |
| -// only thing a stalled sink can hold is the log pump's one write, never |
| -// close or the worker (Astra, round 3, P2). |
| +// nothing that can block, and never waits on the log pump. With the log |
| +// sink stalled — every write hangs — and a relay that never answers but |
| +// honours cancellation, and WHILE THE SINK IS STILL STALLED: |
| +// |
| +// - close returns within its bound; |
| +// - the mail worker has exited; |
| +// - at most the one log pump remains, blocked in its write. |
| +// |
| +// A write blocked inside the logger cannot be interrupted from Go; it |
| +// can only finish or die with the process, and shutdown deliberately |
| +// does not wait for it. So only once the sink is released must the pump |
| +// exit and the goroutine count return to its baseline (Astra on #20, |
| +// rounds 3 and 4). The test fails if close waited on the pump, or if |
| +// the pump could hold the worker. |
| func TestMailQueueCloseIsBoundedWithAStalledLogger(t *testing.T) { |
| - baseline := runtime.NumGoroutine() |
| + waitFor := func(cond func() bool) bool { |
| + deadline := time.Now().Add(2 * time.Second) |
| + for !cond() && time.Now().Before(deadline) { |
| + time.Sleep(5 * time.Millisecond) |
| + } |
| + return cond() |
| + } |
| + // Baselines, not zero: another test's queue may still be shutting |
| + // down beside this one. |
| + baseline, workers := runtime.NumGoroutine(), mailWorkers() |
| release := make(chan struct{}) |
| var once sync.Once |
| free := func() { once.Do(func() { close(release) }) } |
| @@ -227,18 +244,24 @@ func TestMailQueueCloseIsBoundedWithAStalledLogger(t *testing.T) { |
| <-closed |
| t.Fatal("close hung on a stalled log sink") |
| } |
| + |
| + // The sink is still stalled. |
| select { |
| case <-q.done: |
| default: |
| t.Fatal("the worker was still running after close, held by the stalled log sink") |
| } |
| - free() |
| - deadline := time.Now().Add(2 * time.Second) |
| - for runtime.NumGoroutine() > baseline && time.Now().Before(deadline) { |
| - time.Sleep(5 * time.Millisecond) |
| + if !waitFor(func() bool { return runtime.NumGoroutine() <= baseline+1 }) { |
| + t.Fatalf("%d goroutines with the sink stalled, %d before: more than the one blocked log pump remains", runtime.NumGoroutine(), baseline) |
| } |
| - if n := runtime.NumGoroutine(); n > baseline { |
| - t.Fatalf("%d goroutines after close and release, %d before", n, baseline) |
| + if n := mailWorkers(); n > workers+1 { |
| + t.Fatalf("%d mail queue goroutines with the sink stalled, %d before: want at most the log pump more", n, workers) |
| + } |
| + |
| + // Released, the pump finishes its write and exits. |
| + free() |
| + if !waitFor(func() bool { return mailWorkers() <= workers && runtime.NumGoroutine() <= baseline }) { |
| + t.Fatalf("after release: %d mail queue goroutines, %d goroutines against %d before", mailWorkers(), runtime.NumGoroutine(), baseline) |
| } |
| } |