Method watchStuck

internal/jobexecutor/job_executor.go:277–321

Watches for jobs that may have become stuck. i.e. They've run longer than their job timeout (plus a small margin) and don't appear to be responding to context cancellation (unfortunately, quite an easy error to make in Go). Producers use stuck-job notifications for periodic stats and optional user

(ctx context.Context, jobTimeout time.Duration)

// Producers use stuck-job notifications for periodic stats and optional user
// handlers.
func (e *JobExecutor) watchStuck(ctx context.Context, jobTimeout time.Duration) context.CancelFunc {
	// We add a WithoutCancel here so that this inner goroutine becomes
	// immune to all context cancellations _except_ the one where it's
	// cancelled because we leave JobExecutor.execute.
	//
	// This shadows the context outside the e.ClientJobTimeout > 0 check.
	ctx, cancel := context.WithCancel(context.WithoutCancel(ctx))

	go func() {
		const stuckThresholdDefault = 5 * time.Second

		select {
		case <-ctx.Done():
			// context cancelled as we leave JobExecutor.execute

		case <-time.After(jobTimeout + cmp.Or(e.StuckThresholdOverride, stuckThresholdDefault)):
			e.ProducerCallbacks.Stuck(ctx, e.JobRow)

			e.Logger.WarnContext(ctx, e.Name+": Job appears to be stuck",
				slog.Int64("job_id", e.JobRow.ID),
				slog.String("kind", e.JobRow.Kind),
				slog.Duration("timeout", e.ClientJobTimeout),
			)

			// context cancelled as we leave JobExecutor.execute
			<-ctx.Done()

			// In case the executor ever becomes unstuck, inform the
			// producer. However, if we got all the way here there's a good
			// chance this will never happen (the worker is really stuck and
			// will never return).
			defer e.ProducerCallbacks.Unstuck()

			defer func() {
				e.Logger.InfoContext(ctx, e.Name+": Job became unstuck",
					slog.Duration("duration", time.Since(e.start)),
					slog.Int64("job_id", e.JobRow.ID),
					slog.String("kind", e.JobRow.Kind),
				)
			}()
		}
	}()

	return cancel
}

func (e *JobExecutor) invokeErrorHandler(ctx context.Context, res *jobExecutorResult) bool {
	invokeAndHandlePanic := func(funcName string, errorHandler func() *ErrorHandlerResult) *ErrorHandlerResult {