Recorded output

River provides a river.RecordOutput function that can be called within a worker’s Work function to record output for the job:

type MyWorkerOutput struct {
    CreatedResourceID string `json:"created_resource_id"`
}

// ...

func (w *MyWorker) Work(ctx context.Context, job *river.Job[MyJobArgs]) error {
    // ... other job execution logic ...
    // Now that the job has finished, record its output as a final step:
    output := MyWorkerOutput{CreatedResourceID: "123"}
    if err := river.RecordOutput(ctx, job, output); err != nil {
        return err
    }
    return nil
}

Assign a JSON-encodable value to job.output inside the worker. Here, the output records the ID of a created resource:

def work(job)
  job.output = {created_resource_id: "123"}
end

Call record_output on the work context with the value to store. This example records a JSON object containing a resource ID:

context.record_output(riverqueue::serde_json::json!({"created_resource_id": "123"}))?;

Use recordOutput from the work context to attach a JSON value to the job’s output:

workers.add(myJob, async ({ recordOutput }) => {
  recordOutput({ created_resource_id: "123" });
});

Alternatively, return complete({ output: value }) when completing successfully.

This feature is also compatible with transactional job completion using JobCompleteTx, however there is an important caveat. While a job’s output can be recorded at any time during execution, it must be recorded before calling JobCompleteTx in order for the output to be recorded as part of the same transaction.

Calling RecordOutput after JobCompleteTx has been called will still work, but there will be a window of time where the output is not yet recorded in the database on the completed job before it is later added asynchronously.

Ruby does not yet support transactional job completion alongside application writes. Output is persisted with the worker’s final state transition.

Call context.record_output(value) before context.job_complete_tx(&mut tx).await? to include the output in the same transaction. See transactional completion.

Call recordOutput(value) before await completeTx(tx), or pass { output: value } as its second argument, to persist output in the completion transaction. See transactional completion.

Once saved on the job row, the output can be read from a loaded river.JobRow using the Output method. The output is a []byte of the JSON-encoded output value provided to RecordOutput, so it must be unmarshalled to the appropriate type before use:

output := job.Output()
if output == nil {
    // handle error
}

var myOutput MyWorkerOutput
if err := json.Unmarshal(output, &myOutput); err != nil {
    // handle error
}

Fetch the job and read the "output" entry from its metadata:

job = client.job_get(job_id)
output = job.metadata["output"]

Use job.output() to access the stored JSON when output is present, then deserialize it into the value or type you need:

if let Some(output) = job.output() {
    let value: riverqueue::serde_json::Value = riverqueue::serde_json::from_str(output.get())?;
    println!("{value}");
}

Fetch the job through client.jobs and read metadata.output from the returned row:

const job = await client.jobs.get(jobId);
const output = job?.metadata.output;