Reject nested steps and missing cursor targets

CHANGELOG.md

@@ -9,10 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- **Breaking:** Resumable steps can no longer be nested. Calls fail before running child callbacks, preventing unreachable resume points and rolled-back parent checkpoints from being saved again. [PR #1534](https://github.com/riverqueue/river/pull/1534).
 - Job deletion through `Client.JobDelete`, `Client.JobDeleteMany`, their `Tx` variants, and the job cleaner now goes through optional pilot hooks, so a pilot can take part in deletes as it already can in cancels and retries. Behavior with the standard pilot is unchanged. [PR #1500](https://github.com/riverqueue/river/pull/1500).
 
 ### Fixed
 
+- Fail retries that omit a saved cursor-only step, while preserving their checkpoints for recovery. Jobs with an existing step marker retain their behavior, including legacy stale cursor entries. [PR #1534](https://github.com/riverqueue/river/pull/1534).
+
 - Fixed SQLite metadata updates recursively merging nested objects and deleting keys set to JSON null. Updates now replace whole top-level values, matching Postgres and preventing completed resumable cursor steps from running again on later retries. [PR #1533](https://github.com/riverqueue/river/pull/1533).
 
 - Fixed resumable transactional checkpoints being saved again after rollback or a failed write, and first-step cursors being lost on error, snooze, or cooperative shutdown. After a successful checkpoint write, the caller's transaction owns that progress; rollback may replay earlier unpersisted steps. The Rust, JavaScript, and Java runtimes now share the applicable behavior, including preserving progress on snooze.

conformance/cmd/generatefixtures/main.go

@@ -61,6 +61,7 @@ func run() error {
 		{"cron_schedules.json", cronSchedules},
 		{"protocol_values.json", protocolValues},
 		{"resumable.json", resumable},
+		{"resumable_cursor_only.json", makeResumableCursorOnlyCases()},
 		{"snooze_counters.json", makeSnoozeCounters()},
 		{"unique_keys.json", uniqueKeys},
 	} {

conformance/cmd/generatefixtures/resumable.go

@@ -54,6 +54,28 @@ type resumableCases struct {
 	Comment string          `json:"$comment"`
 }
 
+type resumableCursorOnlyAction struct {
+	Fail      bool   `json:"fail"`
+	Kind      string `json:"kind"`
+	Name      string `json:"name"`
+	SetCursor bool   `json:"set_cursor"`
+}
+
+type resumableCursorOnlyCase struct {
+	Actions         []resumableCursorOnlyAction `json:"actions"`
+	ExpectedError   string                      `json:"expected_error"`
+	ExpectedRuns    []string                    `json:"expected_runs"`
+	ExpectedUpdates map[string]any              `json:"expected_updates"`
+	Metadata        map[string]any              `json:"metadata"`
+	Name            string                      `json:"name"`
+	WorkerError     bool                        `json:"worker_error"`
+}
+
+type resumableCursorOnlyCases struct {
+	Cases   []resumableCursorOnlyCase `json:"cases"`
+	Comment string                    `json:"$comment"`
+}
+
 func makeResumableCases() (resumableCases, error) {
 	fixture := resumableCases{Comment: generatedComment("River resumable steps and transactional checkpoints")}
 	for _, name := range []string{
@@ -110,3 +132,42 @@ func makeResumableCases() (resumableCases, error) {
 	}
 	return fixture, nil
 }
+
+func makeResumableCursorOnlyCases() resumableCursorOnlyCases {
+	const (
+		cursorKey = "river:resumable_cursor"
+		stepKey   = "river:resumable_step"
+	)
+	cursorOnly := func(value any) map[string]any {
+		return map[string]any{cursorKey: map[string]any{"loop": value}}
+	}
+	cases := []resumableCursorOnlyCase{
+		{Actions: []resumableCursorOnlyAction{{Kind: "checkpoint", Name: "later"}}, ExpectedError: `resumable checkpoint cannot precede saved cursor step "loop"`, ExpectedRuns: []string{"later"}, Metadata: cursorOnly(0), Name: "CheckpointBeforeCursor"},
+		{Actions: []resumableCursorOnlyAction{{Fail: true, Kind: "plain", Name: "before"}}, ExpectedError: "step interrupted", ExpectedRuns: []string{"before"}, Metadata: cursorOnly(0), Name: "EarlierError"},
+		{Actions: []resumableCursorOnlyAction{{Kind: "cursor", Name: "loop"}}, ExpectedError: "worker interrupted", ExpectedRuns: []string{"loop"}, ExpectedUpdates: map[string]any{cursorKey: nil, stepKey: "loop"}, Metadata: map[string]any{cursorKey: map[string]any{"loop": nil}, stepKey: ""}, Name: "EmptyStepMarker", WorkerError: true},
+		{Actions: []resumableCursorOnlyAction{{Kind: "plain", Name: "done"}, {Kind: "plain", Name: "later"}}, ExpectedRuns: []string{"later"}, Metadata: map[string]any{cursorKey: map[string]any{"loop": 0}, stepKey: "done"}, Name: "LegacyStaleCursor"},
+		{Actions: []resumableCursorOnlyAction{{Kind: "plain", Name: "later"}}, ExpectedError: `resumable cursor step "loop" not found`, ExpectedRuns: []string{"later"}, Metadata: cursorOnly(0), Name: "MissingAfterLaterStep"},
+		{Actions: []resumableCursorOnlyAction{{Kind: "cursor", Name: "new", SetCursor: true}}, ExpectedError: `resumable cursor step "loop" not found`, ExpectedRuns: []string{"new"}, Metadata: cursorOnly(0), Name: "MissingAfterNewCursor"},
+		{ExpectedError: `resumable cursor step "loop" not found`, Metadata: cursorOnly(nil), Name: "MissingNull"},
+		{ExpectedError: `resumable cursor step "loop" not found`, Metadata: cursorOnly(0), Name: "MissingZero"},
+		{Actions: []resumableCursorOnlyAction{{Kind: "cursor", Name: "b"}, {Kind: "cursor", Name: "a"}}, ExpectedError: "worker interrupted", ExpectedRuns: []string{"b", "a"}, ExpectedUpdates: map[string]any{cursorKey: nil, stepKey: "a"}, Metadata: map[string]any{cursorKey: map[string]any{"a": 0, "b": nil}}, Name: "MultipleCursors", WorkerError: true},
+		{Actions: []resumableCursorOnlyAction{{Kind: "cursor", Name: "loop"}}, ExpectedError: "worker interrupted", ExpectedRuns: []string{"loop"}, ExpectedUpdates: map[string]any{cursorKey: nil, stepKey: "loop"}, Metadata: cursorOnly(nil), Name: "NullValueResumes", WorkerError: true},
+		{Actions: []resumableCursorOnlyAction{{Kind: "plain", Name: "loop"}}, ExpectedError: `resumable cursor step "loop" requires a cursor step`, Metadata: cursorOnly(0), Name: "PlainReplacement"},
+		{Actions: []resumableCursorOnlyAction{{Kind: "plain", Name: "before"}, {Kind: "cursor", Name: "loop"}}, ExpectedError: "worker interrupted", ExpectedRuns: []string{"before", "loop"}, ExpectedUpdates: map[string]any{cursorKey: nil, stepKey: "loop"}, Metadata: cursorOnly(0), Name: "ResumesAfterEarlierStep", WorkerError: true},
+		{Actions: []resumableCursorOnlyAction{{Kind: "plain", Name: "before"}, {Fail: true, Kind: "cursor", Name: "loop", SetCursor: true}}, ExpectedError: "step interrupted", ExpectedRuns: []string{"before", "loop"}, ExpectedUpdates: map[string]any{cursorKey: map[string]any{"loop": 7}, stepKey: "before"}, Metadata: cursorOnly(0), Name: "TargetError"},
+		{Actions: []resumableCursorOnlyAction{{Kind: "plain", Name: "later"}}, ExpectedError: "worker interrupted", ExpectedRuns: []string{"later"}, Metadata: cursorOnly(0), Name: "WorkerError", WorkerError: true},
+		{Actions: []resumableCursorOnlyAction{{Kind: "cursor", Name: "loop"}}, ExpectedError: "worker interrupted", ExpectedRuns: []string{"loop"}, ExpectedUpdates: map[string]any{cursorKey: nil, stepKey: "loop"}, Metadata: cursorOnly(0), Name: "ZeroValueResumes", WorkerError: true},
+	}
+	for i := range cases {
+		if cases[i].Actions == nil {
+			cases[i].Actions = []resumableCursorOnlyAction{}
+		}
+		if cases[i].ExpectedRuns == nil {
+			cases[i].ExpectedRuns = []string{}
+		}
+		if cases[i].ExpectedUpdates == nil {
+			cases[i].ExpectedUpdates = map[string]any{}
+		}
+	}
+	return resumableCursorOnlyCases{Cases: cases, Comment: generatedComment("River cursor-only checkpoint compatibility and validation")}
+}

conformance/cmd/generatefixtures/resumable_test.go

@@ -1,10 +1,21 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
+	"maps"
+	"slices"
 	"testing"
 
 	"github.com/stretchr/testify/require"
+
+	"github.com/riverqueue/river"
+	"github.com/riverqueue/river/internal/jobexecutor"
+	"github.com/riverqueue/river/internal/rivercommon"
+	"github.com/riverqueue/river/internal/riverplugin"
+	"github.com/riverqueue/river/riverdriver"
+	"github.com/riverqueue/river/rivertype"
 )
 
 func TestMakeResumableCases(t *testing.T) {
@@ -31,3 +42,88 @@ func TestMakeResumableCases(t *testing.T) {
 		})
 	}
 }
+
+func TestResumableCursorOnly(t *testing.T) {
+	t.Parallel()
+
+	for _, testCase := range makeResumableCursorOnlyCases().Cases {
+		t.Run(testCase.Name, func(t *testing.T) {
+			t.Parallel()
+
+			metadata, err := json.Marshal(testCase.Metadata)
+			require.NoError(t, err)
+			updates := make(map[string]any)
+			ctx := context.WithValue(context.Background(), jobexecutor.ContextKeyMetadataUpdates, updates)
+			client, err := river.NewClient(&fixtureCheckpointDriver{}, &river.Config{})
+			require.NoError(t, err)
+			ctx = context.WithValue(ctx, rivercommon.ContextKeyClient{}, client)
+			row := &rivertype.JobRow{ID: 1, EncodedArgs: []byte(`{}`), Metadata: metadata, State: rivertype.JobStateRunning}
+			staged := *row
+			exec := &fixtureCheckpointExecutor{row: &staged}
+			ran := []string{}
+
+			err = (&riverplugin.ResumableMiddleware{}).Work(ctx, row, func(ctx context.Context) error {
+				for _, action := range testCase.Actions {
+					actionFunc := func(ctx context.Context) error {
+						ran = append(ran, action.Name)
+						if action.Kind == "checkpoint" {
+							_, err := river.ResumableSetStepTx[*fixtureCheckpointDriver](ctx, riverdriver.ExecutorTx(exec), &river.Job[fixtureResumableArgs]{JobRow: row})
+							return err
+						}
+						if action.SetCursor {
+							require.NoError(t, river.ResumableSetCursor(ctx, 7))
+						}
+						if action.Fail {
+							return errors.New("step interrupted")
+						}
+						return nil
+					}
+					if action.Kind == "cursor" {
+						river.ResumableStepCursor(ctx, action.Name, nil, func(ctx context.Context, cursor any) error {
+							cursors, ok := testCase.Metadata[rivercommon.MetadataKeyResumableCursor].(map[string]any)
+							require.True(t, ok)
+							initial := cursors[action.Name]
+							actual, err := json.Marshal(cursor)
+							require.NoError(t, err)
+							expected, err := json.Marshal(initial)
+							require.NoError(t, err)
+							require.JSONEq(t, string(expected), string(actual))
+							return actionFunc(ctx)
+						})
+					} else {
+						river.ResumableStep(ctx, action.Name, nil, actionFunc)
+					}
+				}
+				if testCase.WorkerError {
+					return errors.New("worker interrupted")
+				}
+				return nil
+			})
+			if testCase.ExpectedError == "" {
+				require.NoError(t, err)
+			} else {
+				require.ErrorContains(t, err, testCase.ExpectedError)
+			}
+			require.Equal(t, testCase.ExpectedRuns, ran)
+			actual, err := json.Marshal(updates)
+			require.NoError(t, err)
+			expected, err := json.Marshal(testCase.ExpectedUpdates)
+			require.NoError(t, err)
+			require.JSONEq(t, string(expected), string(actual))
+			require.Equal(t, metadata, staged.Metadata, "a checkpoint before the target must not write")
+
+			if testCase.ExpectedError != "" && len(testCase.ExpectedUpdates) == 0 {
+				// Failed validation must leave a retry able to find the restored step.
+				err = (&riverplugin.ResumableMiddleware{}).Work(ctx, row, func(ctx context.Context) error {
+					cursors, ok := testCase.Metadata[rivercommon.MetadataKeyResumableCursor].(map[string]any)
+					require.True(t, ok)
+					for _, name := range slices.Sorted(maps.Keys(cursors)) {
+						river.ResumableStepCursor(ctx, name, nil, func(context.Context, any) error { return nil })
+					}
+					return nil
+				})
+				require.NoError(t, err)
+			}
+		})
+	}
+}

internal/riverplugin/plugin.go

@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"maps"
+	"slices"
 
 	"github.com/tidwall/gjson"
 
@@ -50,6 +52,14 @@ func (*ResumableMiddleware) Work(ctx context.Context, job *rivertype.JobRow, doI
 		}
 	}
 
+	// Cursor-only progress has no skip boundary. Track its targets separately
+	// so a missing step cannot turn partially processed work into a success.
+	// Older jobs with a step marker may contain stale cursors; preserve their
+	// existing behavior rather than interpreting every cursor as unfinished.
+	if state.ResumeStep == "" {
+		state.PendingCursorSteps = slices.Sorted(maps.Keys(state.Cursors))
+	}
+
 	ctx = context.WithValue(ctx, ResumableContextKey{}, state)
 
 	err := doInner(ctx)
@@ -59,10 +69,14 @@ func (*ResumableMiddleware) Work(ctx context.Context, job *rivertype.JobRow, doI
 			err = state.Err
 		case state.ResumeStep != "" && !state.ResumeMatched:
 			err = fmt.Errorf("river: resumable step %q not found in Worker", state.ResumeStep)
+		case len(state.PendingCursorSteps) > 0:
+			err = fmt.Errorf("river: resumable cursor step %q not found in Worker", state.PendingCursorSteps[0])
 		}
 	}
 
-	if err != nil {
+	// Advancing the skip boundary before finding the cursor could make that
+	// step unreachable even after the worker restores it on a later attempt.
+	if err != nil && len(state.PendingCursorSteps) == 0 {
 		if state.CursorsDirty && len(state.Cursors) > 0 {
 			metadataUpdates[rivercommon.MetadataKeyResumableCursor] = state.Cursors
 		} else if state.CursorsDirty {
@@ -87,9 +101,12 @@ type ResumableState struct {
 	Cursors       map[string]json.RawMessage
 	CursorsDirty  bool
 	Err           error
-	ResumeMatched bool
-	ResumeStep    string
-	StepName      string
+	// PendingCursorSteps contains unmatched targets from cursor-only metadata.
+	PendingCursorSteps []string
+	ResumeMatched      bool
+	ResumeStep         string
+	StepActive         bool
+	StepName           string
 }
 
 // ResumableContextKey is the context key for ResumableState.

java/CHANGELOG.md

@@ -16,4 +16,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fail retries that omit a saved cursor-only step, while preserving their checkpoints for recovery. Jobs with an existing step marker retain their behavior, including legacy stale cursor entries. [PR #1534](https://github.com/riverqueue/river/pull/1534).
+
 - Replace whole top-level metadata values on SQLite, including JSON null, matching Postgres. Completed resumable cursor steps no longer run again because of stale cursor entries. [PR #1533](https://github.com/riverqueue/river/pull/1533).

java/DIFFERENCES.md

@@ -39,9 +39,6 @@
   reproducing Go's nonterminating date loop.
 - SQL is kept in dialect-specific resource catalogs. The Go sqlc driver seam is
   not part of the Java API.
-- Resumable steps cannot be nested in Java. Each attempt has one active step and
-  checkpoint scope, so rejecting nesting avoids ambiguous resume order and
-  checkpoint ownership. Go currently permits nested steps.
 - Transactional checkpoints require the job to still be running with the same
   `attempt` and `attempted_at`. Go updates by job ID alone. Java rejects a stale
   attempt's checkpoint to prevent it from changing a replacement attempt.

java/river/src/main/java/com/riverqueue/WorkContext.java

@@ -4,6 +4,8 @@
 import java.time.Duration;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicReference;
 import tools.jackson.databind.JsonNode;
@@ -25,6 +27,7 @@ public final class WorkContext<A> {
   private final Client client;
   private Workers runtime;
   final java.util.Map<Long, Peer> peers;
+  private final SortedSet<String> pendingCursorSteps = new TreeSet<>();
   private ObjectNode previousCursors;
   private String previousStep;
   private final Set<String> steps = new HashSet<>();
@@ -52,6 +55,8 @@ private WorkContext(Client client, Job<A> job, boolean validate, CancellationSta
     if (validate && cursors.isArray())
       throw new IllegalArgumentException("Invalid resumable cursor metadata");
     this.cursors = cursors.isObject() ? (ObjectNode) cursors.deepCopy() : Json.object();
+    // Step markers may accompany stale cursors written by older clients.
+    if (resumeMatched) this.cursors.propertyNames().forEach(pendingCursorSteps::add);
   }
 
   public A args() {
@@ -197,6 +202,15 @@ synchronized Exception finish(Exception failure) {
     if (stepFailure != null) failure = stepFailure;
     if (failure == null && !resumeMatched)
       failure = new IllegalArgumentException("Saved resumable step was not found");
+    if (!pendingCursorSteps.isEmpty()) {
+      // Do not move the checkpoint past an unmatched cursor-only step.
+      return failure != null
+          ? failure
+          : new IllegalArgumentException(
+              "resumable cursor step "
+                  + Json.encode(pendingCursorSteps.first())
+                  + " not found in worker");
+    }
     if (failure != null && (completedStep != null || cursorsChanged))
       metadata.setAll(resumableMetadata(completedStep, cursors));
     if (failure == null || completedStep != null || cursorsChanged)
@@ -212,6 +226,10 @@ private Job<A> persistCheckpoint(Connection transaction, ObjectNode updates) {
     if (currentStep == null)
       throw new IllegalStateException("Checkpoint requires an active resumable step");
     if (job.state() != Job.State.RUNNING) throw new IllegalStateException("Job must be running");
+    if (!pendingCursorSteps.isEmpty())
+      throw new IllegalStateException(
+          "resumable checkpoint cannot precede saved cursor step "
+              + Json.encode(pendingCursorSteps.first()));
     // Only this step's progress belongs to the caller's transaction. Completion restores the
     // prior buffered progress if the checkpoint rolled back, without replacing a committed one.
     checkpointStep = currentStep;
@@ -312,6 +330,12 @@ public void cursor(Object value) {
   private void executeStep(String name, boolean withCursor, CursorStep action) throws Exception {
     if (stepFailure != null) throw stepFailure;
     if (currentStep != null) throw new IllegalStateException("Resumable steps cannot be nested");
+    if (!withCursor && pendingCursorSteps.contains(name)) {
+      stepFailure =
+          new IllegalArgumentException(
+              "resumable cursor step " + Json.encode(name) + " requires a cursor step");
+      throw stepFailure;
+    }
     if (name.isEmpty() || !steps.add(name)) {
       stepFailure = new IllegalArgumentException("duplicate resumable step or empty name: " + name);
       throw stepFailure;
@@ -327,6 +351,7 @@ private void executeStep(String name, boolean withCursor, CursorStep action) thr
     previousCursors = cursors.deepCopy();
     currentStep = name;
     try {
+      if (withCursor) pendingCursorSteps.remove(name);
       action.run(cursors.path(name));
       completedStep = name;
       cursors.remove(name);