Function ResumableStepCursor

resumable.go:107–151

ResumableStepCursor runs a resumable step that also receives a persisted cursor value from an earlier failed attempt, if one was recorded with ResumableSetCursor. Step names must be unique across all ResumableStep and ResumableStepCursor calls in the same Worker execution. The cursor type T is user

(ctx context.Context, name string, opts *StepOpts, stepFunc func(ctx context.Context, cursor TCursor) error)

//
// opts may be nil.
func ResumableStepCursor[TCursor any](ctx context.Context, name string, opts *StepOpts, stepFunc func(ctx context.Context, cursor TCursor) error) {
	state := mustResumableState(ctx)
	if state.Err != nil {
		return
	}
	if !registerResumableStepName(state, name) {
		return
	}

	if !state.ResumeMatched {
		if name == state.ResumeStep {
			state.CompletedStep = name
			state.ResumeMatched = true

			// If cursor data exists for this step, it was only partially
			// completed on the previous attempt. Fall through to re-execute
			// it with the cursor rather than skipping it.
			if _, hasCursor := state.Cursors[name]; !hasCursor {
				return
			}
		} else {
			return
		}
	}

	var cursor TCursor
	if cursorBytes, ok := state.Cursors[name]; ok && len(cursorBytes) > 0 {
		if err := json.Unmarshal(cursorBytes, &cursor); err != nil {
			state.Err = fmt.Errorf("river: unmarshal resumable cursor for step %q: %w", name, err)
			return
		}
	}

	previousStepName := state.StepName
	state.StepName = name
	defer func() { state.StepName = previousStepName }()

	if err := stepFunc(ctx, cursor); err != nil {
		state.Err = err
		return
	}

	state.CompletedStep = name
	delete(state.Cursors, name)
}

func mustResumableState(ctx context.Context) *rivermiddleware.ResumableState {
	state, ok := resumableStateFromContext(ctx)