Bound the suggestion batch's end write like a job's final write
runBatch recorded the batch's end with an uncancellable context and no deadline, before its handler returned, so a writer connection that stayed busy could hold the worker, and shutdown, open indefinitely. jobs.FinalWriteContext now gives every final write one bounded attempt context (FinalWriteTimeout, 5 seconds, not cancelled with ctx); the runner's record and finishBatch both use it. On timeout at shutdown the handler returns, the job and batch stay running, the next Start fails the job as interrupted and the next StartBatch fails the orphaned batch. TestBlockedBatchEndDoesNotHoldShutdown holds the writer while a batch reaches finishBatch and asserts the worker stops within the bound and the batch is reconciled after restart. TestBlockedFinalWriteDoesNotHoldShutdown no longer sets a runner field, so it compiles against earlier heads. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -51,7 +51,9 @@ func (r batchResult) PublicJSON() ([]byte, error) { return json.Marshal(r) }
|
||||
// instead. Each round's grounded Side quests are stored before the next
|
||||
// round starts, so the deck fills as the batch goes. The batch ends done
|
||||
// or failed; Side quests stored before a failure stay in the deck. When
|
||||
// recording the batch's end fails, so does the job.
|
||||
// recording the batch's end fails, so does the job; when it times out at
|
||||
// shutdown, the job is left running and the batch with it, for the next
|
||||
// start to fail.
|
||||
func (s *Service) runBatch(ctx context.Context, p BatchJob, publish func(batchResult) error) (batchResult, error) {
|
||||
n, err := db.New(s.st.Writer).StartSuggestionBatch(ctx, db.StartSuggestionBatchParams{
|
||||
UpdatedAt: store.Now().String(),
|
||||
@@ -64,14 +66,13 @@ func (s *Service) runBatch(ctx context.Context, p BatchJob, publish func(batchRe
|
||||
return batchResult{}, fmt.Errorf("suggest: batch %s is not pending", p.BatchID)
|
||||
}
|
||||
res, err := s.fill(ctx, p.BatchID, publish)
|
||||
write := context.WithoutCancel(ctx)
|
||||
if err == nil {
|
||||
if err := s.finishBatch(write, p.BatchID, BatchDone); err != nil {
|
||||
if err := s.finishBatch(ctx, p.BatchID, BatchDone); err != nil {
|
||||
return batchResult{}, err
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
finishErr := s.finishBatch(write, p.BatchID, BatchFailed)
|
||||
finishErr := s.finishBatch(ctx, p.BatchID, BatchFailed)
|
||||
if ctx.Err() != nil {
|
||||
return batchResult{}, errors.Join(ctx.Err(), finishErr)
|
||||
}
|
||||
|
||||
@@ -217,11 +217,16 @@ func (s *Service) startBatch(ctx context.Context, tx *sql.Tx, trigger string) (B
|
||||
return Batch{ID: row.ID, JobID: jobID, Trigger: row.Trigger, Status: row.Status, Created: true}, nil
|
||||
}
|
||||
|
||||
// finishBatch records running batch id as status. Its job's handler
|
||||
// returns any error, so the job never ends succeeded with the batch still
|
||||
// running; the runner keeps only a code, so the error is logged too.
|
||||
// finishBatch records running batch id as status, with
|
||||
// jobs.FinalWriteContext: the write is made even when ctx has been
|
||||
// cancelled, but a writer that stays busy cannot hold its worker past
|
||||
// jobs.FinalWriteTimeout. Its job's handler returns any error, so the job
|
||||
// never ends succeeded with the batch still running; the runner keeps only
|
||||
// a code, so the error is logged too.
|
||||
func (s *Service) finishBatch(ctx context.Context, id, status string) error {
|
||||
n, err := db.New(s.st.Writer).FinishSuggestionBatch(ctx, db.FinishSuggestionBatchParams{
|
||||
write, cancel := jobs.FinalWriteContext(ctx)
|
||||
defer cancel()
|
||||
n, err := db.New(s.st.Writer).FinishSuggestionBatch(write, db.FinishSuggestionBatchParams{
|
||||
Status: status,
|
||||
UpdatedAt: store.Now().String(),
|
||||
ID: id,
|
||||
|
||||
@@ -1029,6 +1029,113 @@ func TestBatchRecoversWhenRecordingItsEndAndItsJobsEndFail(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestBlockedBatchEndDoesNotHoldShutdown(t *testing.T) {
|
||||
e := newEnv(t)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
if err := e.runner.Start(ctx); err != nil {
|
||||
cancel()
|
||||
t.Fatal(err)
|
||||
}
|
||||
stopped := make(chan struct{})
|
||||
t.Cleanup(func() {
|
||||
cancel()
|
||||
<-stopped
|
||||
})
|
||||
e.search.Gate = make(chan struct{})
|
||||
e.search.Err = errors.New("search down")
|
||||
e.reply(sideQuest("Chili Verde", "Your Beef Chili gone green.", "Beef Chili"))
|
||||
|
||||
b, err := e.svc.StartBatch(t.Context(), suggest.TriggerManual)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
deadline := time.Now().Add(10 * time.Second)
|
||||
for len(e.search.Queries()) == 0 {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("batch never searched")
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
|
||||
// Hold the only writer connection, so recording the batch's end
|
||||
// waits for it once the search fails.
|
||||
conn, err := e.st.Writer.Conn(t.Context())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waits := e.st.Writer.Stats().WaitCount
|
||||
close(e.search.Gate)
|
||||
for e.st.Writer.Stats().WaitCount == waits {
|
||||
if time.Now().After(deadline) {
|
||||
conn.Close()
|
||||
t.Fatal("batch end never waited for the writer")
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
|
||||
go func() {
|
||||
e.runner.Wait()
|
||||
close(stopped)
|
||||
}()
|
||||
cancel()
|
||||
// Each final write's attempt ends after jobs.FinalWriteTimeout; well
|
||||
// past that, the worker is waiting for the writer with no bound.
|
||||
select {
|
||||
case <-stopped:
|
||||
case <-time.After(jobs.FinalWriteTimeout + 10*time.Second):
|
||||
conn.Close()
|
||||
<-stopped
|
||||
t.Fatal("worker waited for a held writer through shutdown")
|
||||
}
|
||||
if err := conn.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := e.batchStatus(t, b.ID); got != suggest.BatchRunning {
|
||||
t.Fatalf("after shutdown batch status = %s, want left running", got)
|
||||
}
|
||||
|
||||
// The next process fails the job as interrupted, and the batch no
|
||||
// longer blocks the next one.
|
||||
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
next, err := jobs.NewRunner(e.st, logger, 1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
refuse := func(context.Context, string, skills.SideQuestsRequest) (skills.SideQuests, error) {
|
||||
return skills.SideQuests{}, errors.New("not wanted")
|
||||
}
|
||||
svc, err := suggest.NewService(e.st, e.grounder, refuse, next, logger)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
nextCtx, nextCancel := context.WithCancel(context.Background())
|
||||
if err := next.Start(nextCtx); err != nil {
|
||||
nextCancel()
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
nextCancel()
|
||||
next.Wait()
|
||||
})
|
||||
job, err := next.Get(t.Context(), b.JobID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if job.Status != jobs.StatusFailed || job.Error == nil || *job.Error != jobs.CodeInterrupted {
|
||||
t.Fatalf("interrupted job = %+v, want failed with %q", job, jobs.CodeInterrupted)
|
||||
}
|
||||
again, err := svc.StartBatch(t.Context(), suggest.TriggerManual)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !again.Created || again.ID == b.ID {
|
||||
t.Errorf("next batch = %+v, want a new one", again)
|
||||
}
|
||||
if got := e.batchStatus(t, b.ID); got != suggest.BatchFailed {
|
||||
t.Errorf("interrupted batch status = %s, want failed", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStartBatchRefusesUnknownTrigger(t *testing.T) {
|
||||
e := newEnv(t)
|
||||
if _, err := e.svc.StartBatch(t.Context(), "whenever"); err == nil {
|
||||
|
||||
Reference in New Issue
Block a user