River

River jobs transition through several states during their lifecycle. Understanding these states helps in debugging job flows and managing job lifecycles (e.g., retries, cancellations, and manual interventions).

Job States

Initial/Waiting States:

• Available: The job is ready to be fetched and run.
• Scheduled: The job is waiting for a specific time to become available.
• Pending: The job is waiting for preconditions to be met or for a future schedule.

Intermediate States:

• Running: The job is currently being executed by a worker.
• Retryable: The job encountered an error and is waiting for its next retry attempt.

Final States:

• Completed: The job finished successfully.
• Cancelled: The job was explicitly cancelled.
• Discarded: The job failed too many times or was explicitly discarded.

Key Transitions

• Success/Failure: A Running job moves to Completed on success, or Retryable on error. If it exceeds error limits, it moves to Discarded.
• Retries: Jobs in Retryable, Scheduled, Pending, Available, Running, Cancelled, or Discarded states can all be moved back to Available via a manual retry.
• Cancellations: Jobs in Available, Running, Scheduled, Pending, or Retryable states can be moved to Cancelled via a manual cancel.
• Rescuing: A Running job can be rescued into either Retryable or Discarded states.

River jobs are defined using a pair of structs: one for arguments (JobArgs) and one for the processing logic (Worker).

1. JobArgs: Must implement the JobArgs interface by providing a Kind() string method. Use json tags to define how the arguments are serialized to the database.
2. Worker: Implements the Work method. It is recommended to embed river.WorkerDefaults[T] (where T is your args type) to simplify implementation.

type SortArgs struct {
    // Strings is a slice of strings to sort.
    Strings []string `json:"strings"`
}

func (SortArgs) Kind() string { return "sort" }

type SortWorker struct {
    // An embedded WorkerDefaults sets up default methods to fulfill the rest of
    // the Worker interface:
    river.WorkerDefaults[SortArgs]
}

func (w *SortWorker) Work(ctx context.Context, job *river.Job[SortArgs]) error {
    sort.Strings(job.Args.Strings)
    fmt.Printf("Sorted strings: %+v\n", job.Args.Strings)
    return nil
}

If you need to rename a job's Kind, you can implement the JobArgsWithKindAliases interface to prevent existing jobs in the database from being orphaned. This is a three-step process:

1. Initial State: The struct implements Kind() returning the old_name.
2. Transition State: Update Kind() to return the new_name and implement KindAliases() to return []string{"old_name"}. This allows workers to process both old and new job types.
3. Final State: Once all jobs with the old name have been processed (including retries), remove the KindAliases() method so the worker only responds to the new_name.

// Step 1: Original
type jobArgsBeingRenamed struct{}
func (a jobArgsBeingRenamed) Kind() string { return "old_name" }

// Step 2: Transition
type jobArgsBeingRenamed struct{}
func (a jobArgsBeingRenamed) Kind() string          { return "new_name" }
func (a jobArgsBeingRenamed) KindAliases() []string { return []string{"old_name"} }

// Step 3: Final
type jobArgsBeingRenamed struct{}
func (a jobArgsBeingRenamed) Kind() string { return "new_name" }