From 4e4399d974b172604fb2c3078d856740ee1c2dfc Mon Sep 17 00:00:00 2001 From: Brandur Date: Thu, 23 Jul 2026 13:02:00 -0500 Subject: [PATCH] Improve error handling around job retry with conflict Fixes #622, which indicates how when retrying a job and getting a unique conflict, River UI produces an internal server error instead of a proper one that's raised to the user. --- CHANGELOG.md | 1 + go.mod | 2 +- handler_api_endpoint.go | 32 ++++++++++++++++ handler_api_endpoint_test.go | 34 +++++++++++++++++ src/hooks/use-retry-jobs.test.tsx | 63 +++++++++++++++++++++++++++++++ src/hooks/use-retry-jobs.ts | 30 +++++++++++++++ src/routes/jobs/$jobId.tsx | 24 +++--------- src/routes/jobs/index.tsx | 14 ++----- 8 files changed, 171 insertions(+), 29 deletions(-) create mode 100644 src/hooks/use-retry-jobs.test.tsx create mode 100644 src/hooks/use-retry-jobs.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 0618d8e2..5b860076 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Job retry: keep the current page open and show an actionable error when a retry conflicts with another active unique job. [PR #623](https://github.com/riverqueue/riverui/pull/623). - Global live update pause: disable automatic query refreshes on browser focus and reconnect, preventing paused workflow detail pages from re-fetching wait data outside the configured refresh interval. [PR #584](https://github.com/riverqueue/riverui/pull/584). - Job detail: preserve line breaks in attempt logs while keeping structured JSON viewer content outside preformatted code markup. [PR #592](https://github.com/riverqueue/riverui/pull/592). diff --git a/go.mod b/go.mod index ebb7a6e1..6fdeb45a 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.25.0 toolchain go1.25.7 require ( + github.com/jackc/pgerrcode v0.0.0-20250907135507-afb5586c32a6 github.com/jackc/pgx/v5 v5.10.0 github.com/riverqueue/apiframe v0.0.0-20251229202423-2b52ce1c482e github.com/riverqueue/river v0.40.0 @@ -24,7 +25,6 @@ require ( github.com/go-playground/universal-translator v0.18.1 // indirect github.com/go-playground/validator/v10 v10.30.1 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/jackc/pgerrcode v0.0.0-20250907135507-afb5586c32a6 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect diff --git a/handler_api_endpoint.go b/handler_api_endpoint.go index 22cd326f..2da5d7eb 100644 --- a/handler_api_endpoint.go +++ b/handler_api_endpoint.go @@ -10,6 +10,9 @@ import ( "strconv" "time" + "github.com/jackc/pgerrcode" + "github.com/jackc/pgx/v5/pgconn" + "github.com/riverqueue/apiframe/apiendpoint" "github.com/riverqueue/apiframe/apierror" "github.com/riverqueue/apiframe/apitype" @@ -542,6 +545,9 @@ func (a *jobRetryEndpoint[TTx]) Execute(ctx context.Context, req *jobRetryReques if errors.Is(err, river.ErrNotFound) { return nil, NewNotFoundJob(jobID) } + if isJobRetryUniqueConflict(err) { + return nil, newJobRetryUniqueConflictError(err) + } return nil, err } } @@ -550,6 +556,32 @@ func (a *jobRetryEndpoint[TTx]) Execute(ctx context.Context, req *jobRetryReques }) } +const ( + jobRetryUniqueConstraint = "river_job_unique_idx" + jobRetryUniqueMessage = "Job can't be retried because another active job has the same unique properties. Wait for it to finish or delete it before retrying." +) + +type jobRetryUniqueConflictError struct { + apierror.APIError +} + +func newJobRetryUniqueConflictError(internalErr error) *jobRetryUniqueConflictError { + return &jobRetryUniqueConflictError{ + APIError: apierror.APIError{ + InternalError: internalErr, + Message: jobRetryUniqueMessage, + StatusCode: http.StatusConflict, + }, + } +} + +func isJobRetryUniqueConflict(err error) bool { + var pgErr *pgconn.PgError + return errors.As(err, &pgErr) && + pgErr.Code == pgerrcode.UniqueViolation && + pgErr.ConstraintName == jobRetryUniqueConstraint +} + // // queueGetEndpoint // diff --git a/handler_api_endpoint_test.go b/handler_api_endpoint_test.go index b6bab195..6645a8e9 100644 --- a/handler_api_endpoint_test.go +++ b/handler_api_endpoint_test.go @@ -3,6 +3,7 @@ package riverui import ( "context" "log/slog" + "net/http" "testing" "time" @@ -19,6 +20,7 @@ import ( "github.com/riverqueue/river/riverdriver/riverpgxv5" "github.com/riverqueue/river/rivershared/riversharedtest" "github.com/riverqueue/river/rivershared/startstop" + "github.com/riverqueue/river/rivershared/uniquestates" "github.com/riverqueue/river/rivershared/util/ptrutil" "github.com/riverqueue/river/rivertype" @@ -646,6 +648,38 @@ func TestAPIHandlerJobRetry(t *testing.T) { _, err := apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &jobRetryRequest{JobIDs: []int64String{123}}) uicommontest.RequireAPIError(t, NewNotFoundJob(123), err) }) + + t.Run("UniqueConflict", func(t *testing.T) { + t.Parallel() + + endpoint, bundle := setupEndpoint(ctx, t, newJobRetryEndpoint) + uniqueKey := []byte("job-retry-unique-conflict") + uniqueStates := uniquestates.UniqueStatesToBitmask([]rivertype.JobState{rivertype.JobStateAvailable}) + + discardedParams := testfactory.Job_Build(t, &testfactory.JobOpts{ + FinalizedAt: ptrutil.Ptr(time.Now()), + State: ptrutil.Ptr(rivertype.JobStateDiscarded), + }) + discardedParams.UniqueKey = uniqueKey + discardedParams.UniqueStates = uniqueStates + discardedJob, err := bundle.exec.JobInsertFull(ctx, discardedParams) + require.NoError(t, err) + + activeParams := testfactory.Job_Build(t, nil) + activeParams.UniqueKey = uniqueKey + activeParams.UniqueStates = uniqueStates + _, err = bundle.exec.JobInsertFull(ctx, activeParams) + require.NoError(t, err) + + _, err = apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &jobRetryRequest{JobIDs: []int64String{int64String(discardedJob.ID)}}) + require.Error(t, err) + + var apiErr *jobRetryUniqueConflictError + require.ErrorAs(t, err, &apiErr) + require.Equal(t, http.StatusConflict, apiErr.StatusCode) + require.Equal(t, jobRetryUniqueMessage, apiErr.Message) + require.Error(t, apiErr.InternalError) + }) } func TestAPIHandlerQueueGet(t *testing.T) { diff --git a/src/hooks/use-retry-jobs.test.tsx b/src/hooks/use-retry-jobs.test.tsx new file mode 100644 index 00000000..cf4f959f --- /dev/null +++ b/src/hooks/use-retry-jobs.test.tsx @@ -0,0 +1,63 @@ +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { act, renderHook, waitFor } from "@testing-library/react"; +import { type ReactNode } from "react"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { useRetryJobs } from "./use-retry-jobs"; + +const { mockRetryJobs, mockToastError, mockToastSuccess } = vi.hoisted(() => ({ + mockRetryJobs: vi.fn(), + mockToastError: vi.fn(), + mockToastSuccess: vi.fn(), +})); + +vi.mock("@services/jobs", () => ({ + retryJobs: mockRetryJobs, +})); + +vi.mock("@services/toast", () => ({ + toastError: mockToastError, + toastSuccess: mockToastSuccess, +})); + +describe("useRetryJobs", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("keeps retry errors in the mutation and shows the API message", async () => { + const error = new Error( + "Job can't be retried because another active job has the same unique properties. Wait for it to finish or delete it before retrying.", + ); + mockRetryJobs.mockRejectedValue(error); + const onSuccess = vi.fn(); + const { result } = renderHook( + () => + useRetryJobs({ + onSuccess, + successMessage: "Job enqueued for retry", + }), + { wrapper: createWrapper() }, + ); + + act(() => result.current.mutate([123n])); + + await waitFor(() => expect(result.current.isError).toBe(true)); + expect(mockToastError).toHaveBeenCalledWith({ + message: "Job retry failed", + subtext: error.message, + }); + expect(mockToastSuccess).not.toHaveBeenCalled(); + expect(onSuccess).not.toHaveBeenCalled(); + }); +}); + +const createWrapper = () => { + const queryClient = new QueryClient({ + defaultOptions: { mutations: { retry: false, throwOnError: true } }, + }); + + return ({ children }: { children: ReactNode }) => ( + {children} + ); +}; diff --git a/src/hooks/use-retry-jobs.ts b/src/hooks/use-retry-jobs.ts new file mode 100644 index 00000000..65428b1b --- /dev/null +++ b/src/hooks/use-retry-jobs.ts @@ -0,0 +1,30 @@ +import { retryJobs } from "@services/jobs"; +import { toastError, toastSuccess } from "@services/toast"; +import { useMutation } from "@tanstack/react-query"; + +type UseRetryJobsOptions = { + onSuccess: () => unknown; + successMessage: string; +}; + +export const useRetryJobs = ({ + onSuccess, + successMessage, +}: UseRetryJobsOptions) => + useMutation({ + mutationFn: async (jobIDs, context) => retryJobs({ ids: jobIDs }, context), + onError: (error) => { + toastError({ + message: "Job retry failed", + subtext: error.message, + }); + }, + onSuccess: () => { + toastSuccess({ + duration: 2000, + message: successMessage, + }); + return onSuccess(); + }, + throwOnError: false, + }); diff --git a/src/routes/jobs/$jobId.tsx b/src/routes/jobs/$jobId.tsx index b3b65979..f3a8415f 100644 --- a/src/routes/jobs/$jobId.tsx +++ b/src/routes/jobs/$jobId.tsx @@ -2,14 +2,9 @@ import JobDetail from "@components/JobDetail"; import JobNotFound from "@components/JobNotFound"; import { useRefreshSetting } from "@contexts/RefreshSettings.hook"; import { refreshQueryOptions } from "@contexts/RefreshSettings.query"; -import { - cancelJobs, - deleteJobs, - getJob, - getJobKey, - retryJobs, -} from "@services/jobs"; -import { toastError, toastSuccess } from "@services/toast"; +import { useRetryJobs } from "@hooks/use-retry-jobs"; +import { cancelJobs, deleteJobs, getJob, getJobKey } from "@services/jobs"; +import { toastError } from "@services/toast"; import { JobState } from "@services/types"; import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { @@ -99,20 +94,13 @@ function JobComponent() { }, }); - const retryMutation = useMutation({ - mutationFn: async (_variables, context) => - retryJobs({ ids: [jobId] }, context), - throwOnError: true, + const retryMutation = useRetryJobs({ onSuccess: () => { - toastSuccess({ - message: "Job enqueued for retry", - duration: 2000, - }); - return queryClient.invalidateQueries({ queryKey: queryOptionsWithRefresh.queryKey, }); }, + successMessage: "Job enqueued for retry", }); if (jobQuery.isLoading || !jobQuery.data) { @@ -126,7 +114,7 @@ function JobComponent() { cancel={cancelMutation.mutate} deleteFn={deleteMutation.mutate} job={job} - retry={retryMutation.mutate} + retry={() => retryMutation.mutate([jobId])} /> ); } diff --git a/src/routes/jobs/index.tsx b/src/routes/jobs/index.tsx index e7426da6..903ae4ef 100644 --- a/src/routes/jobs/index.tsx +++ b/src/routes/jobs/index.tsx @@ -5,6 +5,7 @@ import { type RefreshQueryOptions, refreshQueryOptions, } from "@contexts/RefreshSettings.query"; +import { useRetryJobs } from "@hooks/use-retry-jobs"; import { defaultValues, jobSearchSchema } from "@routes/jobs/index.schema"; import { cancelJobs, @@ -13,10 +14,9 @@ import { listJobs, ListJobsKey, listJobsKey, - retryJobs, } from "@services/jobs"; import { countsByState, countsByStateKey } from "@services/states"; -import { toastError, toastSuccess } from "@services/toast"; +import { toastError } from "@services/toast"; import { JobState } from "@services/types"; import { PlaceholderDataFunction, @@ -291,20 +291,14 @@ function JobsIndexComponent() { }, }); - const retryMutation = useMutation({ - mutationFn: async (jobIDs: bigint[], context) => - retryJobs({ ids: jobIDs }, context), - throwOnError: true, + const retryMutation = useRetryJobs({ onSuccess: () => { - toastSuccess({ - message: "Jobs enqueued for retry", - duration: 2000, - }); queryClient.invalidateQueries({ queryKey: listJobsKey({ limit, state }), }); queryClient.invalidateQueries({ queryKey: countsByStateKey() }); }, + successMessage: "Jobs enqueued for retry", }); return (