diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7251807e..0d45c00d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -18,6 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 
 - Write timestamps in SQLite to always include three digits after the second like `.000`. Previously, they may have been truncated down to just `.0` in the case of trailing zeroes. [PR #1349](https://github.com/riverqueue/river/pull/1349)
+- Job completion events now reflect the job's persisted outcome when it differs from the transition requested by the worker. For example, a job completed with `JobCompleteTx` before its worker returns an error emits `job_completed`, and a remotely cancelled job whose worker errors or snoozes emits `job_cancelled`. [PR #1350](https://github.com/riverqueue/river/pull/1350)
 
 ## [0.43.0] - 2026-08-05
 
diff --git a/client_test.go b/client_test.go
index cda5eb6b..db6a35ae 100644
--- a/client_test.go
+++ b/client_test.go
@@ -40,6 +40,7 @@ import (
 	"github.com/riverqueue/river/rivershared/riversharedtest"
 	"github.com/riverqueue/river/rivershared/startstoptest"
 	"github.com/riverqueue/river/rivershared/testfactory"
+	"github.com/riverqueue/river/rivershared/testsignal"
 	"github.com/riverqueue/river/rivershared/util/dbutil"
 	"github.com/riverqueue/river/rivershared/util/ptrutil"
 	"github.com/riverqueue/river/rivershared/util/randutil"
@@ -49,6 +50,18 @@ import (
 	"github.com/riverqueue/river/rivertype"
 )
 
+type clientJobCancelTestSignals struct {
+	ContinueWork testsignal.TestSignal[struct{}]
+	JobStarted   testsignal.TestSignal[int64]
+}
+
+func (s *clientJobCancelTestSignals) Init(tb testing.TB) {
+	tb.Helper()
+
+	s.ContinueWork.Init(tb)
+	s.JobStarted.Init(tb)
+}
+
 type invalidKindArgs struct{}
 
 func (a invalidKindArgs) Kind() string { return "this kind is invalid" }
@@ -1210,6 +1223,69 @@ func Test_Client_Common(t *testing.T) {
 		require.WithinDuration(t, time.Now(), *jobAfterCancel.FinalizedAt, 2*time.Second)
 	}
 
+	cancelRunningJobBeforeWorkReturnsTestHelper := func(t *testing.T) {
+		t.Helper()
+
+		config, bundle := setupConfig(t)
+		config.RetryPolicy = &retrypolicytest.RetryPolicySlow{}
+
+		client, err := NewClient(NewDriverWithoutListenNotify(bundle.dbPool), config)
+		require.NoError(t, err)
+
+		var signals clientJobCancelTestSignals
+		signals.Init(t)
+
+		type JobArgs struct {
+			testutil.JobArgsReflectKind[JobArgs]
+
+			Result string `json:"result"`
+		}
+
+		AddWorker(client.config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
+			signals.JobStarted.Signal(job.ID)
+			<-signals.ContinueWork.WaitC()
+
+			switch job.Args.Result {
+			case "error":
+				return errors.New("retry later")
+			case "snooze":
+				return JobSnooze(time.Hour)
+			default:
+				return fmt.Errorf("unknown test result: %s", job.Args.Result)
+			}
+		}))
+
+		subscribeChan := subscribe(t, client)
+		startClient(ctx, t, client)
+		t.Cleanup(func() { signals.ContinueWork.Signal(struct{}{}) })
+
+		for _, result := range []string{"error", "snooze"} {
+			insertRes, err := client.Insert(ctx, &JobArgs{Result: result}, nil)
+			require.NoError(t, err)
+			require.Equal(t, insertRes.Job.ID, signals.JobStarted.WaitOrTimeout())
+
+			var jobAfterCancel *rivertype.JobRow
+			err = pgx.BeginFunc(ctx, bundle.dbPool, func(tx pgx.Tx) error {
+				var err error
+				jobAfterCancel, err = client.JobCancelTx(ctx, tx, insertRes.Job.ID)
+				return err
+			})
+			require.NoError(t, err)
+			require.Equal(t, rivertype.JobStateRunning, jobAfterCancel.State)
+
+			signals.ContinueWork.Signal(struct{}{})
+
+			event := riversharedtest.WaitOrTimeout(t, subscribeChan)
+			require.Equal(t, EventKindJobCancelled, event.Kind)
+			require.Equal(t, insertRes.Job.ID, event.Job.ID)
+			require.Equal(t, rivertype.JobStateCancelled, event.Job.State)
+
+			reloadedJob, err := client.JobGet(ctx, insertRes.Job.ID)
+			require.NoError(t, err)
+			require.Equal(t, rivertype.JobStateCancelled, reloadedJob.State)
+		}
+	}
+
 	t.Run("CancelRunningJob", func(t *testing.T) {
 		t.Parallel()
 
@@ -1218,6 +1294,12 @@ func Test_Client_Common(t *testing.T) {
 		})
 	})
 
+	t.Run("CancelRunningJobBeforeWorkReturns", func(t *testing.T) {
+		t.Parallel()
+
+		cancelRunningJobBeforeWorkReturnsTestHelper(t)
+	})
+
 	t.Run("CancelRunningJobWithLongPollInterval", func(t *testing.T) {
 		t.Parallel()
 
@@ -8077,7 +8159,7 @@ func Test_Client_JobCompletion(t *testing.T) {
 		require.NotNil(t, reloadedJob.FinalizedAt)
 	})
 
-	t.Run("JobThatIsCompletedManuallyIsNotTouchedByCompleter", func(t *testing.T) {
+	t.Run("JobThatIsCompletedManuallyBeforeReturningErrIsNotTouchedByCompleter", func(t *testing.T) {
 		t.Parallel()
 
 		client, bundle := setup(t, newTestConfig(t, ""))
@@ -8096,17 +8178,22 @@ func Test_Client_JobCompletion(t *testing.T) {
 			updatedJob, err = JobCompleteTx[*riverpgxv5.Driver](ctx, tx, job)
 			require.NoError(t, err)
 
-			return tx.Commit(ctx)
+			if err := tx.Commit(ctx); err != nil {
+				return err
+			}
+
+			return errors.New("error after committed completion")
 		}))
 
 		insertRes, err := client.Insert(ctx, JobArgs{}, nil)
 		require.NoError(t, err)
 
 		event := riversharedtest.WaitOrTimeout(t, bundle.subscribeChan)
+		require.Equal(t, EventKindJobCompleted, event.Kind)
 		require.Equal(t, insertRes.Job.ID, event.Job.ID)
 		require.Equal(t, rivertype.JobStateCompleted, event.Job.State)
-		require.Equal(t, rivertype.JobStateCompleted, updatedJob.State)
 		require.NotNil(t, updatedJob)
+		require.Equal(t, rivertype.JobStateCompleted, updatedJob.State)
 		require.NotNil(t, event.Job.FinalizedAt)
 		require.NotNil(t, updatedJob.FinalizedAt)
 
diff --git a/internal/jobcompleter/job_completer.go b/internal/jobcompleter/job_completer.go
index b796a8c6..ab820f9f 100644
--- a/internal/jobcompleter/job_completer.go
+++ b/internal/jobcompleter/job_completer.go
@@ -48,6 +48,48 @@ type CompleterJobUpdated struct {
 	Reason   riverdriver.JobSetStateReason
 }
 
+func newCompleterJobUpdated(job *rivertype.JobRow, stats *jobstats.JobStatistics, requestedReason riverdriver.JobSetStateReason) CompleterJobUpdated {
+	var reason riverdriver.JobSetStateReason
+	switch job.State {
+	case rivertype.JobStateAvailable:
+		// Available is ambiguous: failed jobs being retried immediately, short
+		// snoozes, and jobs interrupted by client shutdown all use it. Preserve
+		// the requested reason when it describes one of those transitions.
+		switch requestedReason {
+		case riverdriver.JobSetStateReasonFailed, riverdriver.JobSetStateReasonInterrupted, riverdriver.JobSetStateReasonSnoozed:
+			reason = requestedReason
+		case riverdriver.JobSetStateReasonCancelled, riverdriver.JobSetStateReasonCompleted:
+			reason = riverdriver.JobSetStateReasonFailed
+		default:
+			panic("completion subscriber received an unknown reason, river bug")
+		}
+	case rivertype.JobStateCancelled:
+		reason = riverdriver.JobSetStateReasonCancelled
+
+	case rivertype.JobStateCompleted:
+		reason = riverdriver.JobSetStateReasonCompleted
+
+	case rivertype.JobStateDiscarded, rivertype.JobStateRetryable:
+		reason = riverdriver.JobSetStateReasonFailed
+
+	case rivertype.JobStateScheduled:
+		reason = riverdriver.JobSetStateReasonSnoozed
+
+	case rivertype.JobStatePending, rivertype.JobStateRunning:
+		panic("completion subscriber received a job that wasn't finalized, river bug")
+
+	default:
+		// linter exhaustive rule prevents this from being reached.
+		panic("completion subscriber received a job with an unknown state, river bug")
+	}
+
+	return CompleterJobUpdated{
+		Job:      job,
+		JobStats: stats,
+		Reason:   reason,
+	}
+}
+
 type InlineCompleter struct {
 	baseservice.BaseService
 	startstop.BaseStartStop
@@ -101,11 +143,7 @@ func (c *InlineCompleter) JobSetStateIfRunning(ctx context.Context, stats *jobst
 	}
 
 	stats.CompleteDuration = c.Time.Now().Sub(start)
-	c.subscribeCh <- []CompleterJobUpdated{{
-		Job:      jobs[0],
-		JobStats: stats,
-		Reason:   params.Reason,
-	}}
+	c.subscribeCh <- []CompleterJobUpdated{newCompleterJobUpdated(jobs[0], stats, params.Reason)}
 
 	return nil
 }
@@ -216,11 +254,7 @@ func (c *AsyncCompleter) JobSetStateIfRunning(ctx context.Context, stats *jobsta
 		}
 
 		stats.CompleteDuration = c.Time.Now().Sub(start)
-		c.subscribeCh <- []CompleterJobUpdated{{
-			Job:      jobs[0],
-			JobStats: stats,
-			Reason:   params.Reason,
-		}}
+		c.subscribeCh <- []CompleterJobUpdated{newCompleterJobUpdated(jobs[0], stats, params.Reason)}
 
 		return nil
 	})
@@ -508,11 +542,7 @@ func (c *BatchCompleter) handleBatch(ctx context.Context) error {
 	for _, jobRow := range jobRows {
 		setState := setStateBatch[jobRow.ID]
 		setState.Stats.CompleteDuration = completeTime.Sub(setState.StartTime)
-		events = append(events, CompleterJobUpdated{
-			Job:      jobRow,
-			JobStats: setState.Stats,
-			Reason:   setState.Params.Reason,
-		})
+		events = append(events, newCompleterJobUpdated(jobRow, setState.Stats, setState.Params.Reason))
 	}
 
 	if len(events) > 0 {
diff --git a/internal/jobcompleter/job_completer_test.go b/internal/jobcompleter/job_completer_test.go
index 77203933..77d9e40e 100644
--- a/internal/jobcompleter/job_completer_test.go
+++ b/internal/jobcompleter/job_completer_test.go
@@ -72,6 +72,41 @@ func (m *partialExecutorTxMock) JobSetStateIfRunningMany(ctx context.Context, pa
 	return m.partial.JobSetStateIfRunningMany(ctx, params)
 }
 
+func TestNewCompleterJobUpdated(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		expectedReason  riverdriver.JobSetStateReason
+		name            string
+		requestedReason riverdriver.JobSetStateReason
+		state           rivertype.JobState
+	}{
+		{expectedReason: riverdriver.JobSetStateReasonFailed, name: "AvailableFailed", requestedReason: riverdriver.JobSetStateReasonFailed, state: rivertype.JobStateAvailable},
+		{expectedReason: riverdriver.JobSetStateReasonInterrupted, name: "AvailableInterrupted", requestedReason: riverdriver.JobSetStateReasonInterrupted, state: rivertype.JobStateAvailable},
+		{expectedReason: riverdriver.JobSetStateReasonFailed, name: "AvailableRequestedCompleted", requestedReason: riverdriver.JobSetStateReasonCompleted, state: rivertype.JobStateAvailable},
+		{expectedReason: riverdriver.JobSetStateReasonSnoozed, name: "AvailableSnoozed", requestedReason: riverdriver.JobSetStateReasonSnoozed, state: rivertype.JobStateAvailable},
+		{expectedReason: riverdriver.JobSetStateReasonCancelled, name: "Cancelled", requestedReason: riverdriver.JobSetStateReasonFailed, state: rivertype.JobStateCancelled},
+		{expectedReason: riverdriver.JobSetStateReasonCompleted, name: "Completed", requestedReason: riverdriver.JobSetStateReasonFailed, state: rivertype.JobStateCompleted},
+		{expectedReason: riverdriver.JobSetStateReasonFailed, name: "Discarded", requestedReason: riverdriver.JobSetStateReasonCompleted, state: rivertype.JobStateDiscarded},
+		{expectedReason: riverdriver.JobSetStateReasonFailed, name: "Retryable", requestedReason: riverdriver.JobSetStateReasonCompleted, state: rivertype.JobStateRetryable},
+		{expectedReason: riverdriver.JobSetStateReasonSnoozed, name: "Scheduled", requestedReason: riverdriver.JobSetStateReasonFailed, state: rivertype.JobStateScheduled},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+
+			job := &rivertype.JobRow{State: test.state}
+			stats := &jobstats.JobStatistics{}
+
+			update := newCompleterJobUpdated(job, stats, test.requestedReason)
+			require.Same(t, job, update.Job)
+			require.Same(t, stats, update.JobStats)
+			require.Equal(t, test.expectedReason, update.Reason)
+		})
+	}
+}
+
 func TestInlineJobCompleter_Complete(t *testing.T) {
 	t.Parallel()
 
diff --git a/job_complete_tx.go b/job_complete_tx.go
index 5c1a75eb..f6e6d00f 100644
--- a/job_complete_tx.go
+++ b/job_complete_tx.go
@@ -24,6 +24,11 @@ import (
 //		// handle error
 //	}
 //
+// After successfully committing the transaction, the worker should normally
+// return nil. If it returns an error instead, the committed completion takes
+// precedence. the job remains completed and a resulting subscribe event has
+// kind EventKindJobCompleted. The error has no effect.
+//
 // Returns the updated, completed job.
 func JobCompleteTx[TDriver riverdriver.Driver[TTx], TTx any, TArgs JobArgs](ctx context.Context, tx TTx, job *Job[TArgs]) (*Job[TArgs], error) {
 	if job.State != rivertype.JobStateRunning {
diff --git a/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go b/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go
index c72d277a..a197f7bb 100644
--- a/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go
+++ b/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go
@@ -1483,7 +1483,7 @@ updated AS (
     SET
         attempt = CASE
             WHEN river_job.state = 'running'
-                 AND NOT (job_input.state IN ('retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
+                 AND NOT (job_input.state IN ('available','retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
                  AND job_input.attempt_do_update
             THEN job_input.attempt
             ELSE river_job.attempt
@@ -1496,7 +1496,7 @@ updated AS (
         END,
         finalized_at = CASE
             WHEN river_job.state = 'running'
-                 AND (job_input.state IN ('retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
+                 AND (job_input.state IN ('available','retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
             THEN coalesce($13::timestamptz, now())
             WHEN river_job.state = 'running'
                  AND job_input.finalized_at_do_update
@@ -1510,14 +1510,14 @@ updated AS (
         END,
         scheduled_at = CASE
             WHEN river_job.state = 'running'
-                 AND NOT (job_input.state IN ('retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
+                 AND NOT (job_input.state IN ('available','retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
                  AND job_input.scheduled_at_do_update
             THEN job_input.scheduled_at
             ELSE river_job.scheduled_at
         END,
         state = CASE
             WHEN river_job.state = 'running'
-                 AND (job_input.state IN ('retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
+                 AND (job_input.state IN ('available','retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
             THEN 'cancelled'::/* TEMPLATE: schema */river_job_state
             WHEN river_job.state = 'running'
             THEN job_input.state
diff --git a/riverdriver/riverdrivertest/job_update.go b/riverdriver/riverdrivertest/job_update.go
index e354ae31..955f775a 100644
--- a/riverdriver/riverdrivertest/job_update.go
+++ b/riverdriver/riverdrivertest/job_update.go
@@ -788,43 +788,50 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
 		})
 
 		t.Run("SetsAJobWithCancelAttemptedAtToCancelled", func(t *testing.T) {
+			t.Parallel()
+
 			// If a job has cancel_attempted_at in its metadata, it means that the user
 			// tried to cancel the job with the Cancel API but that the job
 			// finished/errored before the producer received the cancel notification.
-			//
-			// In this case, we want to move the job to cancelled instead of retryable
-			// so that the job is not retried.
-			t.Parallel()
-
-			exec, _ := setup(ctx, t)
-
-			now := time.Now().UTC()
-
-			job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{
-				Metadata:    fmt.Appendf(nil, `{"cancel_attempted_at":"%s"}`, time.Now().UTC().Format(time.RFC3339)),
-				State:       ptrutil.Ptr(rivertype.JobStateRunning),
-				ScheduledAt: ptrutil.Ptr(now.Add(-10 * time.Second)),
-			})
-
-			jobsAfter, err := exec.JobSetStateIfRunningMany(ctx, setStateManyParams(riverdriver.JobSetStateErrorRetryable(job.ID, now, makeErrPayload(t, now), nil)))
-			require.NoError(t, err)
-			jobAfter := jobsAfter[0]
-			require.Equal(t, rivertype.JobStateCancelled, jobAfter.State)
-			require.NotNil(t, jobAfter.FinalizedAt)
-			// Loose assertion against FinalizedAt just to make sure it was set (it uses
-			// the database's now() instead of a passed-in time):
-			require.WithinDuration(t, time.Now().UTC(), *jobAfter.FinalizedAt, 2*time.Second)
-			// ScheduledAt should not be touched:
-			require.WithinDuration(t, job.ScheduledAt, jobAfter.ScheduledAt, time.Microsecond)
-
-			// Errors should still be appended to:
-			require.Len(t, jobAfter.Errors, 1)
-			require.Contains(t, jobAfter.Errors[0].Error, "fake error")
+			// In this case, move the job to cancelled so that it is not retried.
+			tests := []struct {
+				name         string
+				setStateFunc func(id int64, scheduledAt time.Time, errData []byte, metadataUpdates []byte) *riverdriver.JobSetStateIfRunningParams
+			}{
+				{name: "ImmediatelyAvailable", setStateFunc: riverdriver.JobSetStateErrorAvailable},
+				{name: "Retryable", setStateFunc: riverdriver.JobSetStateErrorRetryable},
+			}
 
-			jobUpdated, err := exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: ""})
-			require.NoError(t, err)
-			require.Equal(t, rivertype.JobStateCancelled, jobUpdated.State)
-			require.WithinDuration(t, job.ScheduledAt, jobAfter.ScheduledAt, time.Microsecond)
+			for _, test := range tests {
+				t.Run(test.name, func(t *testing.T) {
+					t.Parallel()
+
+					exec, _ := setup(ctx, t)
+
+					now := time.Now().UTC()
+					job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{
+						Metadata:    fmt.Appendf(nil, `{"cancel_attempted_at":"%s"}`, time.Now().UTC().Format(time.RFC3339)),
+						State:       ptrutil.Ptr(rivertype.JobStateRunning),
+						ScheduledAt: ptrutil.Ptr(now.Add(-10 * time.Second)),
+					})
+
+					jobsAfter, err := exec.JobSetStateIfRunningMany(ctx, setStateManyParams(test.setStateFunc(job.ID, now, makeErrPayload(t, now), nil)))
+					require.NoError(t, err)
+					jobAfter := jobsAfter[0]
+					require.Equal(t, rivertype.JobStateCancelled, jobAfter.State)
+					require.NotNil(t, jobAfter.FinalizedAt)
+					require.WithinDuration(t, time.Now().UTC(), *jobAfter.FinalizedAt, 2*time.Second)
+					require.WithinDuration(t, job.ScheduledAt, jobAfter.ScheduledAt, time.Microsecond)
+
+					require.Len(t, jobAfter.Errors, 1)
+					require.Contains(t, jobAfter.Errors[0].Error, "fake error")
+
+					jobUpdated, err := exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: ""})
+					require.NoError(t, err)
+					require.Equal(t, rivertype.JobStateCancelled, jobUpdated.State)
+					require.WithinDuration(t, job.ScheduledAt, jobAfter.ScheduledAt, time.Microsecond)
+				})
+			}
 		})
 	})
 
diff --git a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql
index 509e479b..40098ed0 100644
--- a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql
+++ b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql
@@ -640,7 +640,7 @@ updated AS (
     SET
         attempt = CASE
             WHEN river_job.state = 'running'
-                 AND NOT (job_input.state IN ('retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
+                 AND NOT (job_input.state IN ('available','retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
                  AND job_input.attempt_do_update
             THEN job_input.attempt
             ELSE river_job.attempt
@@ -653,7 +653,7 @@ updated AS (
         END,
         finalized_at = CASE
             WHEN river_job.state = 'running'
-                 AND (job_input.state IN ('retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
+                 AND (job_input.state IN ('available','retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
             THEN coalesce(sqlc.narg('now')::timestamptz, now())
             WHEN river_job.state = 'running'
                  AND job_input.finalized_at_do_update
@@ -667,14 +667,14 @@ updated AS (
         END,
         scheduled_at = CASE
             WHEN river_job.state = 'running'
-                 AND NOT (job_input.state IN ('retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
+                 AND NOT (job_input.state IN ('available','retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
                  AND job_input.scheduled_at_do_update
             THEN job_input.scheduled_at
             ELSE river_job.scheduled_at
         END,
         state = CASE
             WHEN river_job.state = 'running'
-                 AND (job_input.state IN ('retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
+                 AND (job_input.state IN ('available','retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
             THEN 'cancelled'::/* TEMPLATE: schema */river_job_state
             WHEN river_job.state = 'running'
             THEN job_input.state
diff --git a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go
index a361baac..7eb082c5 100644
--- a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go
+++ b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go
@@ -1447,7 +1447,7 @@ updated AS (
     SET
         attempt = CASE
             WHEN river_job.state = 'running'
-                 AND NOT (job_input.state IN ('retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
+                 AND NOT (job_input.state IN ('available','retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
                  AND job_input.attempt_do_update
             THEN job_input.attempt
             ELSE river_job.attempt
@@ -1460,7 +1460,7 @@ updated AS (
         END,
         finalized_at = CASE
             WHEN river_job.state = 'running'
-                 AND (job_input.state IN ('retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
+                 AND (job_input.state IN ('available','retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
             THEN coalesce($13::timestamptz, now())
             WHEN river_job.state = 'running'
                  AND job_input.finalized_at_do_update
@@ -1474,14 +1474,14 @@ updated AS (
         END,
         scheduled_at = CASE
             WHEN river_job.state = 'running'
-                 AND NOT (job_input.state IN ('retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
+                 AND NOT (job_input.state IN ('available','retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
                  AND job_input.scheduled_at_do_update
             THEN job_input.scheduled_at
             ELSE river_job.scheduled_at
         END,
         state = CASE
             WHEN river_job.state = 'running'
-                 AND (job_input.state IN ('retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
+                 AND (job_input.state IN ('available','retryable','scheduled') AND river_job.metadata ? 'cancel_attempted_at')
             THEN 'cancelled'::/* TEMPLATE: schema */river_job_state
             WHEN river_job.state = 'running'
             THEN job_input.state
diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql
index 94b433ca..3ba74a82 100644
--- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql
+++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql
@@ -594,16 +594,16 @@ RETURNING *;
 -- name: JobSetStateIfRunning :one
 UPDATE /* TEMPLATE: schema */river_job
 SET
-    -- should_cancel: (job_input.state IN ('retryable', 'scheduled') AND river_job.metadata ? 'cancel_attempted_at')
+    -- should_cancel: (job_input.state IN ('available', 'retryable', 'scheduled') AND river_job.metadata ? 'cancel_attempted_at')
     --
-    -- or inverted:   (cast(@state AS text) <> 'retryable' AND @state <> 'scheduled' OR NOT (metadata -> 'cancel_attempted_at'))
-    attempt      = CASE WHEN /* NOT should_cancel */(cast(@state AS text) <> 'retryable' AND @state <> 'scheduled' OR (metadata -> 'cancel_attempted_at') IS NULL) AND cast(@attempt_do_update AS boolean)
+    -- or inverted:   (cast(@state AS text) <> 'available' AND @state <> 'retryable' AND @state <> 'scheduled' OR NOT (metadata -> 'cancel_attempted_at'))
+    attempt      = CASE WHEN /* NOT should_cancel */(cast(@state AS text) <> 'available' AND @state <> 'retryable' AND @state <> 'scheduled' OR (metadata -> 'cancel_attempted_at') IS NULL) AND cast(@attempt_do_update AS boolean)
                         THEN @attempt
                         ELSE attempt END,
     errors       = CASE WHEN cast(@errors_do_update AS boolean)
                         THEN jsonb(json_insert(json(coalesce(errors, jsonb('[]'))), '$[#]', json(@error)))
                         ELSE errors END,
-    finalized_at = CASE WHEN /* should_cancel */((@state = 'retryable' OR @state = 'scheduled') AND (metadata -> 'cancel_attempted_at') iS NOT NULL)
+    finalized_at = CASE WHEN /* should_cancel */((@state = 'available' OR @state = 'retryable' OR @state = 'scheduled') AND (metadata -> 'cancel_attempted_at') IS NOT NULL)
                         THEN coalesce(cast(sqlc.narg('now') AS text), datetime('now', 'subsec'))
                         WHEN cast(@finalized_at_do_update AS boolean)
                         THEN @finalized_at
@@ -611,10 +611,10 @@ SET
     metadata     = CASE WHEN cast(@metadata_do_merge AS boolean)
                         THEN jsonb_patch(json(metadata), json(@metadata_updates))
                         ELSE metadata END,
-    scheduled_at = CASE WHEN /* NOT should_cancel */(cast(@state AS text) <> 'retryable' AND @state <> 'scheduled' OR (metadata -> 'cancel_attempted_at') IS NULL) AND cast(@scheduled_at_do_update AS boolean)
+    scheduled_at = CASE WHEN /* NOT should_cancel */(cast(@state AS text) <> 'available' AND @state <> 'retryable' AND @state <> 'scheduled' OR (metadata -> 'cancel_attempted_at') IS NULL) AND cast(@scheduled_at_do_update AS boolean)
                         THEN @scheduled_at
                         ELSE scheduled_at END,
-    state        = CASE WHEN /* should_cancel */((@state = 'retryable' OR @state = 'scheduled') AND (metadata -> 'cancel_attempted_at') IS NOT NULL)
+    state        = CASE WHEN /* should_cancel */((@state = 'available' OR @state = 'retryable' OR @state = 'scheduled') AND (metadata -> 'cancel_attempted_at') IS NOT NULL)
                         THEN 'cancelled'
                         ELSE @state END
 WHERE id = @id
diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go
index 5bd39b2c..1a8c2971 100644
--- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go
+++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go
@@ -1633,16 +1633,16 @@ func (q *Queries) JobSetMetadataIfNotRunning(ctx context.Context, db DBTX, arg *
 const jobSetStateIfRunning = `-- name: JobSetStateIfRunning :one
 UPDATE /* TEMPLATE: schema */river_job
 SET
-    -- should_cancel: (job_input.state IN ('retryable', 'scheduled') AND river_job.metadata ? 'cancel_attempted_at')
+    -- should_cancel: (job_input.state IN ('available', 'retryable', 'scheduled') AND river_job.metadata ? 'cancel_attempted_at')
     --
-    -- or inverted:   (cast(@state AS text) <> 'retryable' AND @state <> 'scheduled' OR NOT (metadata -> 'cancel_attempted_at'))
-    attempt      = CASE WHEN /* NOT should_cancel */(cast(?1 AS text) <> 'retryable' AND ?1 <> 'scheduled' OR (metadata -> 'cancel_attempted_at') IS NULL) AND cast(?2 AS boolean)
+    -- or inverted:   (cast(@state AS text) <> 'available' AND @state <> 'retryable' AND @state <> 'scheduled' OR NOT (metadata -> 'cancel_attempted_at'))
+    attempt      = CASE WHEN /* NOT should_cancel */(cast(?1 AS text) <> 'available' AND ?1 <> 'retryable' AND ?1 <> 'scheduled' OR (metadata -> 'cancel_attempted_at') IS NULL) AND cast(?2 AS boolean)
                         THEN ?3
                         ELSE attempt END,
     errors       = CASE WHEN cast(?4 AS boolean)
                         THEN jsonb(json_insert(json(coalesce(errors, jsonb('[]'))), '$[#]', json(?5)))
                         ELSE errors END,
-    finalized_at = CASE WHEN /* should_cancel */((?1 = 'retryable' OR ?1 = 'scheduled') AND (metadata -> 'cancel_attempted_at') iS NOT NULL)
+    finalized_at = CASE WHEN /* should_cancel */((?1 = 'available' OR ?1 = 'retryable' OR ?1 = 'scheduled') AND (metadata -> 'cancel_attempted_at') IS NOT NULL)
                         THEN coalesce(cast(?6 AS text), datetime('now', 'subsec'))
                         WHEN cast(?7 AS boolean)
                         THEN ?8
@@ -1650,10 +1650,10 @@ SET
     metadata     = CASE WHEN cast(?9 AS boolean)
                         THEN jsonb_patch(json(metadata), json(?10))
                         ELSE metadata END,
-    scheduled_at = CASE WHEN /* NOT should_cancel */(cast(?1 AS text) <> 'retryable' AND ?1 <> 'scheduled' OR (metadata -> 'cancel_attempted_at') IS NULL) AND cast(?11 AS boolean)
+    scheduled_at = CASE WHEN /* NOT should_cancel */(cast(?1 AS text) <> 'available' AND ?1 <> 'retryable' AND ?1 <> 'scheduled' OR (metadata -> 'cancel_attempted_at') IS NULL) AND cast(?11 AS boolean)
                         THEN ?12
                         ELSE scheduled_at END,
-    state        = CASE WHEN /* should_cancel */((?1 = 'retryable' OR ?1 = 'scheduled') AND (metadata -> 'cancel_attempted_at') IS NOT NULL)
+    state        = CASE WHEN /* should_cancel */((?1 = 'available' OR ?1 = 'retryable' OR ?1 = 'scheduled') AND (metadata -> 'cancel_attempted_at') IS NOT NULL)
                         THEN 'cancelled'
                         ELSE ?1 END
 WHERE id = ?13
diff --git a/rivertest/worker_test.go b/rivertest/worker_test.go
index 496efccb..5f87d108 100644
--- a/rivertest/worker_test.go
+++ b/rivertest/worker_test.go
@@ -589,6 +589,28 @@ func TestWorker_WorkJob(t *testing.T) {
 		require.Equal(t, rivertype.JobStateCompleted, updatedJob.State)
 	})
 
+	t.Run("JobCompleteTxWithInsertedJobRowThenError", func(t *testing.T) {
+		t.Parallel()
+
+		testWorker, bundle := setup(t)
+
+		args := testArgs{}
+		insertRes, err := bundle.client.InsertTx(ctx, bundle.tx, args, nil)
+		require.NoError(t, err)
+
+		bundle.workFunc = func(ctx context.Context, job *river.Job[testArgs]) error {
+			_, err := river.JobCompleteTx[*riverpgxv5.Driver](ctx, bundle.tx, job)
+			require.NoError(t, err)
+
+			return errors.New("error after completion")
+		}
+
+		res, err := testWorker.WorkJob(ctx, t, bundle.tx, insertRes.Job)
+		require.EqualError(t, err, "error after completion")
+		require.Equal(t, river.EventKindJobCompleted, res.EventKind)
+		require.Equal(t, rivertype.JobStateCompleted, res.Job.State)
+	})
+
 	t.Run("ErrorsWhenGivenAlreadyCompletedJob", func(t *testing.T) {
 		t.Parallel()