Method execute

internal/jobexecutor/job_executor.go:185–269

Executes the job, handling a panic if necessary (and various other error conditions). The named return value is so that we can still return a value in case of a panic. nolint:nonamedreturns

(ctx context.Context)

//
//nolint:nonamedreturns
func (e *JobExecutor) execute(ctx context.Context) (res *jobExecutorResult) {
	metadataUpdates := make(map[string]any)
	ctx = context.WithValue(ctx, ContextKeyMetadataUpdates, metadataUpdates)

	defer func() {
		if recovery := recover(); recovery != nil {
			e.Logger.ErrorContext(ctx, e.Name+": panic recovery; possible bug with Worker",
				slog.Int64("job_id", e.JobRow.ID),
				slog.String("kind", e.JobRow.Kind),
				slog.String("panic_val", fmt.Sprintf("%v", recovery)),
			)

			res = &jobExecutorResult{
				MetadataUpdates: metadataUpdates,
				// Skip the first 4 frames which are:
				//
				// 1. The `runtime.Callers` function.
				// 2. The `captureStackTraceSkipFrames` function.
				// 3. The current recovery defer function.
				// 4. The `JobExecutor.execute` method working the job.
				PanicTrace: captureStackTraceSkipFrames(4),
				PanicVal:   recovery,
			}
		}
		e.stats.RunDuration = e.Time.Now().Sub(e.start)
	}()

	if e.WorkUnit == nil {
		e.Logger.ErrorContext(ctx, e.Name+": Unhandled job kind",
			slog.String("kind", e.JobRow.Kind),
			slog.Int64("job_id", e.JobRow.ID),
		)
		return &jobExecutorResult{Err: &rivertype.UnknownJobKindError{Kind: e.JobRow.Kind}, MetadataUpdates: metadataUpdates}
	}

	doInner := execution.Func(func(ctx context.Context) error {
		{
			for _, hook := range append(
				e.HookLookupGlobal.ByHookKind(hooklookup.HookKindWorkBegin),
				e.WorkUnit.HookLookup(e.HookLookupByJob).ByHookKind(hooklookup.HookKindWorkBegin)...,
			) {
				if err := hook.(rivertype.HookWorkBegin).WorkBegin(ctx, e.JobRow); err != nil { //nolint:forcetypeassert
					return err
				}
			}
		}

		if err := e.WorkUnit.UnmarshalJob(); err != nil {
			return err
		}

		jobTimeout := cmp.Or(e.WorkUnit.Timeout(), e.ClientJobTimeout)

		if jobTimeout > 0 {
			var timeoutCancel context.CancelFunc
			ctx, timeoutCancel = context.WithTimeout(ctx, jobTimeout)
			defer timeoutCancel()