Method Work

example_resumable_set_step_tx_test.go:34–63

(ctx context.Context, job *river.Job[ResumableStepTxArgs])

}

func (w *ResumableStepTxWorker) Work(ctx context.Context, job *river.Job[ResumableStepTxArgs]) error {
	const durableStep = "durable_step"

	river.ResumableStep(ctx, durableStep, nil, func(ctx context.Context) error {
		tx, err := w.dbPool.Begin(ctx)
		if err != nil {
			return err
		}
		defer tx.Rollback(ctx)

		// Perform some kind database work in a transaction.
		var result int
		if err := tx.QueryRow(ctx, "SELECT 1").Scan(&result); err != nil {
			return err
		}

		// Then, record the step as completed in the same transaction.
		if _, err := river.ResumableSetStepTx[*riverpgxv5.Driver](ctx, tx, job); err != nil {
			return err
		}

		if err := tx.Commit(ctx); err != nil {
			return err
		}

		return errors.New("simulated failure after persisting step")
	})

	return nil
}

// Example_resumableSetStepTx demonstrates how to transactionally persist a
// resumable step so it survives a failed attempt.