[brandur]

Fixes #1576.

When the enqueuer was still starting up, an immediate stop could cancel
its maintenance context while it was in staggered sleep. The sleep would
return immeditaely, but the startup continued and tried to insert
RunOnStart jobs which would error as it tried to start a transaction
on an already-cancelled context.

Fix is to do more cancelled context checking.

I had Codex go through and audit the other maintenance services and
ports for similar problem. It found a few more including some in JS and
Rust, which are also fixed. Java and Ruby were unaffected.

[brandur]

@codex review

[chatgpt-codex-connector]

Codex Review Summary

This comment shows the latest Codex review activity on this pull request.

Review	Status	Commit	Review trigger
📝 Code Review	✅ Completed 2026-11-04T05:25:15.941457Z	75be917	Manual request

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review" or "@codex security review".

Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings.

[chatgpt-codex-connector]

Codex Review: Didn't find any major issues. Nice work!

Reviewed commit: 75be917303

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[bgentry]

Nice fixes. The bot reviewer found a few additional minor issues and cleanup opportunities that might be worth incorporating, LGTM otherwise.

🤖 Approving. The fix is correct in Go, JS, and Rust, and I confirmed Java and Ruby are unaffected: both stop maintenance cooperatively (Java drains the executor with shutdown() + awaitTermination(), Ruby sets a stop flag and joins the thread), so no in-flight query is ever interrupted there. The three ports now share one rule: a cancelled service or term swallows any error shape, while a local query timeout is still logged.

Verified locally: new Go tests with -race -count=5, the full internal/maintenance package with -race against Postgres, golangci-lint, the new vitest case, and the three Rust unit tests. All pass.

Comments below are one remaining gap of the same bug class in Rust, one duplication cleanup, and a few test nits. None are blocking.

[bgentry]

🤖 This covers everything routed through cancellable_within, but the reindexer has two error sources that bypass it and can still report a shutdown race as a failure: reindexable_index_names (plain tokio::time::timeout) and MaintenanceTransaction::begin (transaction? and row? return Failed directly). run_periodically already guards its log site with context.cancel.is_cancelled(), so periodic services are fine, but the reindexer's two error arms in reindexer.rs do not have that guard.

Suggested fix, mirroring the Go approach of checking at the log site so every error source is covered at once:

--- a/rust/riverqueue/src/maintenance/reindexer.rs
+++ b/rust/riverqueue/src/maintenance/reindexer.rs
                         Ok(true) => info!(index_name, "River reindexer rebuilt an index"),
                         Ok(false) => {}
                         Err(MaintenanceError::Cancelled) => return,
                         Err(reindex_error) => {
+                            if context.cancel.is_cancelled() {
+                                return;
+                            }
                             error!(error = %crate::error::Chain(&reindex_error), index_name, "River reindexer failed");
                         }
                     }
                 }
             }
             Err(MaintenanceError::Cancelled) => return,
             Err(list_error) => {
+                if context.cancel.is_cancelled() {
+                    return;
+                }
                 error!(error = %crate::error::Chain(&list_error), "River reindexer could not list indexes");
             }

The cancellable_within change is still worth keeping for the "preserve a successful commit" case its new test covers.

[brandur]

🤖 Fixed in 448d6e1: both reindexer error branches now check the term's cancellation token before logging. Kept the cancellable_within change so successful results, including commits, remain preserved. make test/rust/postgres and make lint/rust pass.

[bgentry]

🤖 This six-line check is now pasted verbatim into seven Start methods (cleaner, rescuer, scheduler, enqueuer, queue cleaner, reindexer, SQLite notification cleaner), the scheduler picked up a new //nolint:dupl because of it, and only the enqueuer's copy has a test. Suggest moving the decision into StaggerStart so it is written and tested once:

--- a/rivershared/riversharedmaintenance/river_shared_maintenance.go
+++ b/rivershared/riversharedmaintenance/river_shared_maintenance.go
 // StaggerStart is called when queue maintainer services start. It jitters by
 // sleeping for a short random period so services don't all perform their first
-// run at exactly the same time.
-func (s *QueueMaintainerServiceBase) StaggerStart(ctx context.Context) {
-	if s.staggerStartupDisabled {
-		return
-	}
-
-	serviceutil.CancellableSleep(ctx, randutil.DurationBetween(0*time.Second, 1*time.Second))
+// run at exactly the same time. Returns false if the context was cancelled
+// before or during the sleep, in which case the caller should mark itself
+// stopped and return instead of starting its run loop.
+func (s *QueueMaintainerServiceBase) StaggerStart(ctx context.Context) bool {
+	if !s.staggerStartupDisabled {
+		serviceutil.CancellableSleep(ctx, randutil.DurationBetween(0*time.Second, 1*time.Second))
+	}
+	return ctx.Err() == nil
 }

Then each of the seven call sites becomes:

	if !s.StaggerStart(ctx) {
		stopped()
		return nil
	}

and the //nolint:dupl added to JobScheduler.Start in this PR can be dropped. Add a small test next to StaggerStart in riversharedmaintenance with two cases: an already-cancelled context returns false (with and without StaggerStartupDisable(true)), and a live context with stagger disabled returns true. TestPeriodicJobEnqueuer_ShutdownDuringStagger still covers the end-to-end path through one service.

[brandur]

🤖 Moved the cancellation check into a boolean-returning StaggerStart and updated all seven callers in 448d6e1. Added shared tests for live/cancelled contexts with stagger enabled and disabled, and kept the end-to-end stagger shutdown test.

Retained the scheduler's //nolint:dupl: lint still flags its loop as a duplicate of the SQLite notification cleaner after this cleanup. make test, make lint, and the affected shutdown/startup tests with -race -count=5 pass.

[bgentry]

🤖 nit: the four if serviceCtx.Err() == nil { log } guards are the same decision made four times. The other services make it once by having runOnce return an error and logging in the loop. Optional, but insertBatch could do the same, which also removes the serviceCtx aliasing:

func (s *PeriodicJobEnqueuer) insertBatch(ctx context.Context, insertParamsMany []*rivertype.JobInsertParams, periodicJobUpsertParams *riverpilot.PeriodicJobUpsertManyParams, periodicJobsToKeepAlive map[rivertype.PeriodicJobHandle]*PeriodicJob) error {
	if len(insertParamsMany) < 1 && len(periodicJobUpsertParams.Jobs) < 1 {
		return nil
	}

	ctx, cancel := context.WithTimeout(ctx, riversharedmaintenance.TimeoutDefault)
	defer cancel()

	tx, err := s.exec.Begin(ctx)
	if err != nil {
		return fmt.Errorf("error starting transaction: %w", err)
	}
	defer dbutil.RollbackWithoutCancel(ctx, tx)

	if len(insertParamsMany) > 0 {
		if _, err := s.Config.Insert(ctx, tx, insertParamsMany); err != nil {
			return fmt.Errorf("error inserting %d periodic jobs: %w", len(insertParamsMany), err)
		}
	}

	if len(periodicJobUpsertParams.Jobs) > 0 {
		if _, err = s.Config.Pilot.PeriodicJobUpsertMany(ctx, tx, periodicJobUpsertParams); err != nil {
			return fmt.Errorf("error upserting %d periodic job next run times: %w", len(periodicJobUpsertParams.Jobs), err)
		}
	}

	if err := tx.Commit(ctx); err != nil {
		return fmt.Errorf("error committing transaction: %w", err)
	}

	// ... keepAlive marking and test signals unchanged ...
	return nil
}

Both call sites (inside validateInsertRunOnStartAndScheduleNewlyAdded and in the timer branch of the run loop) then use the same shape as the other services:

			if err := s.insertBatch(ctx, insertParamsMany, periodicJobUpsertParams, periodicJobsToKeepAlive); err != nil {
				if ctx.Err() != nil {
					return
				}
				s.Logger.ErrorContext(ctx, s.Name+": Error inserting periodic jobs", "error", err.Error())
			}

The ctx.Err() check at the top of validateInsertRunOnStartAndScheduleNewlyAdded should stay since it also skips calling user constructors. TestPeriodicJobEnqueuer_TransactionErrorLogging would need its log assertions adjusted to the single message plus the wrapped cause.

[brandur]

🤖 Leaving this optional refactor out of this PR to keep the change focused. The existing transaction error tests cover cancellation, local timeouts, and database failures at each failure point.

[bgentry]

🤖 nit: this leader is built without an Elector, so if tryStart ever exhausts its retries the test dies with a nil pointer dereference in RequestResign rather than an assertion. It cannot happen today because synctest.Wait returns instead of advancing the fake clock, so the backoff sleep never completes before Stop cancels the context, but that is a non-obvious invariant to leave implicit. Suggest recording it:

Suggested change

	// Exercise synchronous startup failures through the leader's real retry loop.
	// Exercise synchronous startup failures through the leader's real retry loop.
	//
	// Config.Elector is intentionally nil: the test stops the service while
	// tryStart is in its first backoff sleep (synctest.Wait returns without
	// advancing the fake clock), so retries are never exhausted and
	// RequestResign is never reached. If that changes, give this a stub elector.

[brandur]

🤖 Documented in 448d6e1 that the elector is intentionally nil: synctest.Wait does not advance the clock, so the tests stop during the first backoff before retries can exhaust and request resignation.

[bgentry]

🤖 nit: returning both a non-empty row set and a non-nil error from the same call is a shape no real driver produces, and it is easy to misread as intentional. Return rows only on success:

Suggested change

	func (e *maintenanceShutdownExecutor) JobGetStuck(ctx context.Context, _ *riverdriver.JobGetStuckParams) ([]*rivertype.JobRow, error) {
	return []*rivertype.JobRow{{ID: 1, Kind: "unknown", Metadata: []byte(`{}`)}}, e.operationFunc(ctx, "JobGetStuck")
	}
	func (e *maintenanceShutdownExecutor) JobGetStuck(ctx context.Context, _ *riverdriver.JobGetStuckParams) ([]*rivertype.JobRow, error) {
	if err := e.operationFunc(ctx, "JobGetStuck"); err != nil {
	return nil, err
	}
	return []*rivertype.JobRow{{ID: 1, Kind: "unknown", Metadata: []byte(`{}`)}}, nil
	}

[brandur]

🤖 Fixed in 448d6e1. JobGetStuck now returns rows only on success; applied the same correction to IndexesExist.

[bgentry]

🤖 nit, optional: this matrix now covers every maintenance service except the scheduler, which keeps its own TestJobScheduler_ErrorLogging in job_scheduler_test.go with the same shape (write-timeout error, cancel vs. database error, assert on log output). Folding a JobSchedule operation into this matrix would need the fake executor to grow a Begin returning a fake transaction with JobSchedule, Commit, and Rollback, after which TestJobScheduler_ErrorLogging and jobSchedulerExecutorMock could be deleted. Fine to leave for a follow-up or skip entirely; mentioning it so the two tests do not drift apart.

[brandur]

🤖 Keeping the scheduler's dedicated error-logging test for this PR. Folding it into the matrix would require additional transaction mock plumbing, and the existing test already covers cancellation and genuine query failures.