diff --git a/CHANGELOG.md b/CHANGELOG.md index 19a52fa38..0755bb469 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/conformance/cmd/generatefixtures/main.go b/conformance/cmd/generatefixtures/main.go index aa568eaed..9b26b9471 100644 --- a/conformance/cmd/generatefixtures/main.go +++ b/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}, } { diff --git a/conformance/cmd/generatefixtures/resumable.go b/conformance/cmd/generatefixtures/resumable.go index b6e3fedd1..ab62b41fc 100644 --- a/conformance/cmd/generatefixtures/resumable.go +++ b/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")} +} diff --git a/conformance/cmd/generatefixtures/resumable_test.go b/conformance/cmd/generatefixtures/resumable_test.go index 263c2f212..d4308a286 100644 --- a/conformance/cmd/generatefixtures/resumable_test.go +++ b/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) + } + }) + } +} diff --git a/internal/riverplugin/plugin.go b/internal/riverplugin/plugin.go index ea50a0db9..f224d8379 100644 --- a/internal/riverplugin/plugin.go +++ b/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. diff --git a/java/CHANGELOG.md b/java/CHANGELOG.md index 812c47f45..d3f3fd72a 100644 --- a/java/CHANGELOG.md +++ b/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). diff --git a/java/DIFFERENCES.md b/java/DIFFERENCES.md index a18a9ebee..da02bb4a9 100644 --- a/java/DIFFERENCES.md +++ b/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. diff --git a/java/river/src/main/java/com/riverqueue/WorkContext.java b/java/river/src/main/java/com/riverqueue/WorkContext.java index c6e19b80e..f46c527f2 100644 --- a/java/river/src/main/java/com/riverqueue/WorkContext.java +++ b/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 { private final Client client; private Workers runtime; final java.util.Map peers; + private final SortedSet pendingCursorSteps = new TreeSet<>(); private ObjectNode previousCursors; private String previousStep; private final Set steps = new HashSet<>(); @@ -52,6 +55,8 @@ private WorkContext(Client client, Job 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 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); diff --git a/java/river/src/test/java/com/riverqueue/ProtocolTest.java b/java/river/src/test/java/com/riverqueue/ProtocolTest.java index 442a28d03..6d6f015d4 100644 --- a/java/river/src/test/java/com/riverqueue/ProtocolTest.java +++ b/java/river/src/test/java/com/riverqueue/ProtocolTest.java @@ -300,6 +300,30 @@ public double nextDouble() { return tests.stream(); } + @Test + void nestedStepsAreRejectedBeforeRecordingProgress() { + for (boolean parentCursor : new boolean[] {false, true}) { + for (boolean childCursor : new boolean[] {false, true}) { + var work = context(42); + WorkContext.Step parent = + () -> { + if (childCursor) work.stepWithCursor("child", ignored -> fail("child must not run")); + else work.step("child", () -> fail("child must not run")); + }; + var failure = + assertThrows( + IllegalStateException.class, + () -> { + if (parentCursor) work.stepWithCursor("parent", ignored -> parent.run()); + else work.step("parent", parent); + }); + assertEquals("Resumable steps cannot be nested", failure.getMessage()); + assertSame(failure, work.finish(failure)); + assertTrue(work.metadataUpdates().isEmpty()); + } + } + } + @TestFactory Stream resumableConformance() throws Exception { var tests = new ArrayList(); @@ -335,6 +359,75 @@ Stream resumableConformance() throws Exception { return tests.stream(); } + @TestFactory + Stream resumableCursorOnlyConformance() throws Exception { + var tests = new ArrayList(); + for (var fixture : Conformance.fixture("resumable_cursor_only.json").required("cases")) { + tests.add( + DynamicTest.dynamicTest( + fixture.required("name").asString(), + () -> { + var metadata = fixture.required("metadata"); + var work = context(42, metadata); + var runs = new ArrayList(); + Exception failure = null; + try { + for (var action : fixture.required("actions")) { + String name = action.required("name").asString(); + String kind = action.required("kind").asString(); + WorkContext.Step callback = + () -> { + runs.add(name); + if (kind.equals("checkpoint")) work.checkpoint(null); + if (action.required("set_cursor").asBoolean()) work.cursor(7); + if (action.required("fail").asBoolean()) + throw new Exception("step interrupted"); + }; + if (kind.equals("cursor")) + work.stepWithCursor( + name, + cursor -> { + var expected = + metadata.path(Protocol.METADATA_RESUMABLE_CURSOR).path(name); + assertEquals( + expected.isMissingNode() ? Json.tree(null) : expected, + cursor.isMissingNode() ? Json.tree(null) : cursor); + callback.run(); + }); + else work.step(name, callback); + } + if (fixture.required("worker_error").asBoolean()) + throw new Exception("worker interrupted"); + } catch (Exception error) { + failure = error; + } + failure = work.finish(failure); + String expectedError = fixture.required("expected_error").asString(); + if (expectedError.isEmpty()) assertNull(failure); + else { + assertNotNull(failure); + assertTrue(failure.getMessage().contains(expectedError), failure.toString()); + } + assertEquals(fixture.required("expected_runs"), Json.tree(runs)); + assertEquals(fixture.required("expected_updates"), work.metadataUpdates()); + if (!expectedError.isEmpty() && work.metadataUpdates().isEmpty()) { + var retry = context(42, metadata); + for (String name : + metadata.required(Protocol.METADATA_RESUMABLE_CURSOR).propertyNames()) { + retry.stepWithCursor( + name, + cursor -> + assertEquals( + metadata.path(Protocol.METADATA_RESUMABLE_CURSOR).path(name), + cursor)); + } + assertNull(retry.finish(null)); + } + })); + } + return tests.stream(); + } + @Test void resumableFirstStepCursor() throws Exception { var context = context(42, Json.object()); diff --git a/java/river/src/test/java/com/riverqueue/ResumableTest.java b/java/river/src/test/java/com/riverqueue/ResumableTest.java index 08349069a..255b7953e 100644 --- a/java/river/src/test/java/com/riverqueue/ResumableTest.java +++ b/java/river/src/test/java/com/riverqueue/ResumableTest.java @@ -21,6 +21,53 @@ class ResumableTest { @RegisterExtension final TestDatabase databases = new TestDatabase(); @TempDir Path directory; + @Test + @SuppressWarnings("try") // Resources intentionally scope background activity. + void cursorOnlyStepRemainsRecoverableAfterItIsOmitted() throws Exception { + var client = new Client(databases.open(directory.resolve("cursor-only.db"))); + var events = new LinkedBlockingQueue(); + var received = new LinkedBlockingQueue(); + try (var workers = + client + .workers() + .queue("default", 1) + .pollOnly(true) + .leadership(false) + .retryPolicy(job -> Duration.ofDays(1)) + .stopTimeout(Duration.ofSeconds(5)) + .add( + TYPE, + context -> { + if (context.job().attempt() > 1) + context.stepWithCursor( + "loop", + cursor -> { + received.add(cursor.asInt()); + context.transaction(connection -> context.checkpoint(connection, 1)); + }); + context.step("later", () -> {}); + }) + .start(); + var subscription = workers.subscribe(events::add)) { + var metadata = Map.of(Protocol.METADATA_RESUMABLE_CURSOR, Map.of("loop", 0)); + var inserted = + client.insert(TYPE, "work", InsertOptions.builder().metadata(metadata).build()).job(); + var failed = events.poll(5, TimeUnit.SECONDS); + assertNotNull(failed); + assertEquals(Workers.EventKind.JOB_FAILED, failed.kind()); + var savedMetadata = client.get(inserted.id()).metadata(); + assertEquals( + Json.tree(metadata).path(Protocol.METADATA_RESUMABLE_CURSOR), + savedMetadata.path(Protocol.METADATA_RESUMABLE_CURSOR)); + assertFalse(savedMetadata.has(Protocol.METADATA_RESUMABLE_STEP)); + client.retry(inserted.id()); + var completed = events.poll(5, TimeUnit.SECONDS); + assertNotNull(completed); + assertEquals(Workers.EventKind.JOB_COMPLETED, completed.kind()); + assertEquals(java.util.List.of(0), java.util.List.copyOf(received)); + } + } + @ParameterizedTest @CsvSource({ "false,false,false", diff --git a/js/CHANGELOG.md b/js/CHANGELOG.md index 64300a9ad..c1a878de1 100644 --- a/js/CHANGELOG.md +++ b/js/CHANGELOG.md @@ -15,12 +15,15 @@ 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). - Like River for Go, `insertMany` rejects an empty batch with a `ValidationError` instead of returning an empty result. [PR #1504](https://github.com/riverqueue/river/pull/1504). - Like River for Go, the client no longer caps `maxAttempts` at 32767: SQLite stores wider attempt counts, and the Postgres and Prisma drivers clamp `maxAttempts` to their 16-bit column on insert instead of failing. [PR #1504](https://github.com/riverqueue/river/pull/1504). - Like River for Go's `ResumableStep`, a failed resumable step records the step's own error as the attempt's error instead of wrapping it in a `LifecycleError`. [PR #1504](https://github.com/riverqueue/river/pull/1504). ### 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). - Preserve first-step resumable cursors on failure and step progress on snooze. Transactional checkpoints no longer leak into deferred updates after rollback or a failed write; rolling back a successful checkpoint write may replay earlier unpersisted steps. diff --git a/js/docs/deployment.md b/js/docs/deployment.md index 76975ffa9..062cefe7e 100644 --- a/js/docs/deployment.md +++ b/js/docs/deployment.md @@ -16,8 +16,8 @@ order or sorted map keys. Select scalar leaf paths (for example, `byArgs: of identity. Missing selected fields and explicit `null` are distinct. Resumable step names and cursor shapes must also match between implementations. -Keep names stable and await steps sequentially; nested steps are supported, -but concurrent steps do not define a checkpoint order. Catching a step error +Keep names stable and await steps sequentially. Steps cannot be nested inside +other step callbacks. Catching a step error does not make the attempt successful: River retains the failure and checkpoint. The test helper returns updated metadata so a subsequent test attempt can resume with the same persisted state as a real worker. diff --git a/js/docs/resumable-jobs.md b/js/docs/resumable-jobs.md index 5ccc81b55..bd22289a6 100644 --- a/js/docs/resumable-jobs.md +++ b/js/docs/resumable-jobs.md @@ -28,8 +28,8 @@ const workers = new Workers().add( ``` If `storage` fails, the retry skips `billing` and runs `storage` again. Steps -must be awaited one at a time (nested steps are fine, concurrent ones are -not), and names must stay stable across deploys because they are persisted. +must be awaited one at a time and cannot be nested inside other step callbacks. +Names must stay stable across deploys because they are persisted. Catching a step's error doesn't make the attempt succeed: River still fails the attempt and keeps the checkpoint. @@ -63,6 +63,14 @@ const workers = new Workers().add(exportRows, async ({ resumable }) => { }); ``` +A saved cursor without a step marker keeps the existing replay behavior: the +retry starts from the beginning and supplies the cursor at its named step. +That step must still be reached through `stepWithCursor()`. If it is omitted, +renamed, or replaced by `step()`, the attempt fails and retains its original +checkpoint so restoring the step can recover. Transactional checkpoints in +other steps must wait until the saved cursor step is reached. Existing step +markers keep their current behavior, including legacy stale cursor entries. + The cursor is any JSON value and is saved when the attempt ends. To save progress immediately, and atomically with the step's own writes, call `resumable.checkpoint({ cursor, tx })` inside the step with your database diff --git a/js/driver/pg/src/resumable-nested.integration.test.ts b/js/driver/pg/src/resumable-nested.integration.test.ts new file mode 100644 index 000000000..140107496 --- /dev/null +++ b/js/driver/pg/src/resumable-nested.integration.test.ts @@ -0,0 +1,100 @@ +import { readdir, readFile } from "node:fs/promises"; +import { fileURLToPath } from "node:url"; + +import pg from "pg"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { Client, defineJob, Workers, type JsonObject } from "riverqueue"; + +import { PgDriver } from "./driver.js"; + +describe("nested transactional cursor checkpoints", () => { + const schema = `js_nested_${Math.random().toString(36).slice(2, 10)}`; + const pool = new pg.Pool({ + connectionString: + process.env.TEST_DATABASE_URL ?? + "postgres://localhost:5432/river_test?sslmode=disable", + }); + beforeAll(async () => { + await pool.query(`CREATE SCHEMA "${schema}"`); + const directory = fileURLToPath( + new URL("../../../migrate/migrations/postgres/main/", import.meta.url) + ); + for (const file of (await readdir(directory)) + .filter((name) => name.endsWith(".up.sql")) + .sort()) { + const sql = await readFile(`${directory}/${file}`, "utf8"); + await pool.query( + sql.replaceAll("/* TEMPLATE: schema */", `"${schema}".`) + ); + } + }); + afterAll(async () => { + await pool.query(`DROP SCHEMA "${schema}" CASCADE`); + await pool.end(); + }); + + it.each([ + { commit: true, name: "committed" }, + { commit: false, name: "rolled_back" }, + ])( + "preserves only durable parent progress after the parent transaction is $name", + async ({ commit, name }) => { + const definition = defineJob({ kind: name }); + let afterTransaction: JsonObject | undefined; + const client = new Client(new PgDriver(pool, { schema }), { + queues: { [name]: { maxWorkers: 1 } }, + workers: new Workers().add(definition, async ({ job, resumable }) => { + await resumable.stepWithCursor("parent", async () => { + const tx = await pool.connect(); + try { + await tx.query("BEGIN"); + await resumable.checkpoint({ cursor: 1, tx }); + await resumable.stepWithCursor("child", () => { + throw new Error("nested child must not run"); + }); + } finally { + await tx.query(commit ? "COMMIT" : "ROLLBACK"); + tx.release(); + afterTransaction = (await client.jobs.get(job.id))?.metadata; + } + }); + }), + }); + const inserted = await client.insert( + definition, + {}, + { maxAttempts: 1, queue: name } + ); + const runtime = await client.start(); + try { + let row = await client.jobs.get(inserted.job.id); + await expect + .poll(async () => { + row = await client.jobs.get(inserted.job.id); + return row?.state; + }) + .toBe("discarded"); + expect(row?.errors.at(-1)?.error).toBe( + "resumable steps cannot be nested" + ); + if (!commit) + expect(afterTransaction?.["river:resumable_cursor"]).toBeUndefined(); + const metadata = Object.fromEntries( + Object.entries(row?.metadata ?? {}).filter(([key]) => + key.startsWith("river:resumable_") + ) + ); + expect(metadata).toEqual( + commit + ? { + "river:resumable_step": "parent", + "river:resumable_cursor": { parent: 1 }, + } + : {} + ); + } finally { + await runtime.stop(); + } + } + ); +}); diff --git a/js/driver/pg/src/resumable.integration.test.ts b/js/driver/pg/src/resumable.integration.test.ts index 7909b2611..b84214544 100644 --- a/js/driver/pg/src/resumable.integration.test.ts +++ b/js/driver/pg/src/resumable.integration.test.ts @@ -50,6 +50,64 @@ describe("Go resumable conformance", () => { await pool.end(); }); + it("keeps an omitted cursor-only step recoverable", async () => { + const driver = new PgDriver(pool, { schema }); + const definition = defineJob({ kind: "cursor_only_recovery" }); + const queue = "cursor_only_recovery"; + const received: unknown[] = []; + let restored = false; + const client = new Client(driver, { + completionBatchSize: 1, + leaderElectionDisabled: true, + queues: { + [queue]: { + maxWorkers: 1, + fetchCooldown: { milliseconds: 1 }, + pollInterval: { milliseconds: 5 }, + }, + }, + retryPolicy: (_job, now) => now.add({ hours: 1 }), + workers: new Workers().add(definition, async ({ resumable }) => { + if (restored) + await resumable.stepWithCursor("loop", async (cursor) => { + received.push(cursor); + const tx = await pool.connect(); + try { + await tx.query("BEGIN"); + await resumable.checkpoint({ cursor: 1, tx }); + await tx.query("COMMIT"); + } finally { + await tx.query("ROLLBACK"); + tx.release(); + } + }); + await resumable.step("later", () => undefined); + }), + }); + const metadata = { "river:resumable_cursor": { loop: 0 } }; + const inserted = await client.insert(definition, {}, { metadata, queue }); + const runtime = await client.start(); + try { + await expect + .poll(async () => (await client.jobs.get(inserted.job.id))?.state) + .toBe("retryable"); + const failed = await client.jobs.get(inserted.job.id); + expect(failed?.metadata).toMatchObject(metadata); + expect(failed?.metadata["river:resumable_step"]).toBeUndefined(); + expect(failed?.errors.at(-1)?.error).toContain( + 'resumable cursor step "loop" not found' + ); + restored = true; + await client.jobs.retry(inserted.job.id); + await expect + .poll(async () => (await client.jobs.get(inserted.job.id))?.state) + .toBe("completed"); + expect(received).toEqual([0]); + } finally { + await runtime.stop(); + } + }); + it.each(fixtures.cases)("$name", async (fixture) => { const definition = defineJob({ kind: fixture.name }); const queue = fixture.name; diff --git a/js/driver/sqlite/src/runtime.test.ts b/js/driver/sqlite/src/runtime.test.ts index afa7d3315..1e8a444e8 100644 --- a/js/driver/sqlite/src/runtime.test.ts +++ b/js/driver/sqlite/src/runtime.test.ts @@ -331,6 +331,62 @@ describe("SqliteDriver runtime", () => { } }); + test("keeps an omitted cursor-only step recoverable", async () => { + const { database, driver } = await setup(); + const definition = defineJob({ kind: "cursor_only_recovery" }); + const queue = "cursor_only_recovery"; + const received: unknown[] = []; + let restored = false; + const client = new Client(driver, { + completionBatchSize: 1, + leaderElectionDisabled: true, + queues: { + [queue]: { + maxWorkers: 1, + fetchCooldown: { milliseconds: 1 }, + pollInterval: { milliseconds: 5 }, + }, + }, + retryPolicy: (_job, now) => now.add({ hours: 1 }), + workers: new Workers().add(definition, async ({ resumable }) => { + if (restored) + await resumable.stepWithCursor("loop", async (cursor) => { + received.push(cursor); + database.exec("BEGIN IMMEDIATE"); + try { + await resumable.checkpoint({ cursor: 1, tx: database }); + database.exec("COMMIT"); + } finally { + if (database.isTransaction) database.exec("ROLLBACK"); + } + }); + await resumable.step("later", () => undefined); + }), + }); + const metadata = { "river:resumable_cursor": { loop: 0 } }; + const inserted = await client.insert(definition, {}, { metadata, queue }); + const runtime = await client.start(); + try { + await expect + .poll(async () => (await client.jobs.get(inserted.job.id))?.state) + .toBe("retryable"); + const failed = await client.jobs.get(inserted.job.id); + expect(failed?.metadata).toMatchObject(metadata); + expect(failed?.metadata["river:resumable_step"]).toBeUndefined(); + expect(failed?.errors.at(-1)?.error).toContain( + 'resumable cursor step "loop" not found' + ); + restored = true; + await client.jobs.retry(inserted.job.id); + await expect + .poll(async () => (await client.jobs.get(inserted.job.id))?.state) + .toBe("completed"); + expect(received).toEqual([0]); + } finally { + await runtime.stop(); + } + }); + test("persists resumable checkpoints on failed SQLite attempts", async () => { const { driver } = await setup(); const definition = defineJob({ kind: "sqlite_runtime_resumable_retry" }); @@ -368,6 +424,62 @@ describe("SqliteDriver runtime", () => { } }); + test.each([false, true])( + "rejects nested steps after a parent checkpoint (commit=%s)", + async (commit) => { + const { database, driver } = await setup(); + const definition = defineJob({ kind: "sqlite_nested_resumable" }); + const client = new Client(driver, { + completionBatchSize: 1, + leaderElectionDisabled: true, + queues: { + default: { + fetchCooldown: { milliseconds: 1 }, + maxWorkers: 1, + pollInterval: { milliseconds: 5 }, + }, + }, + workers: new Workers().add(definition, async ({ resumable }) => { + await resumable.stepWithCursor("parent", async () => { + database.exec("BEGIN IMMEDIATE"); + try { + await resumable.checkpoint({ cursor: 1, tx: database }); + await resumable.stepWithCursor("child", () => { + throw new Error("nested child must not run"); + }); + } finally { + database.exec(commit ? "COMMIT" : "ROLLBACK"); + } + }); + }), + }); + const inserted = await client.insert(definition, {}, { maxAttempts: 1 }); + const run = await client.start(); + try { + const row = await waitForJob(client, inserted.job.id, "discarded"); + expect(row.errors.at(-1)?.error).toBe( + "resumable steps cannot be nested" + ); + expect( + Object.fromEntries( + Object.entries(row.metadata).filter(([key]) => + key.startsWith("river:resumable_") + ) + ) + ).toEqual( + commit + ? { + "river:resumable_step": "parent", + "river:resumable_cursor": { parent: 1 }, + } + : {} + ); + } finally { + await run.stop(); + } + } + ); + test("drops a finished cursor step's cursor on later SQLite attempts", async () => { const { driver } = await setup(); const definition = defineJob({ kind: "sqlite_runtime_resumable_cursor" }); diff --git a/js/etc/riverqueue.api.md b/js/etc/riverqueue.api.md index aeddab899..92f077049 100644 --- a/js/etc/riverqueue.api.md +++ b/js/etc/riverqueue.api.md @@ -2353,10 +2353,14 @@ export type ReindexerSchedule = (after: Temporal.Instant) => Temporal.Instant; export declare class Resumable { /** * Run a named step unless an earlier failed attempt completed it. - * Await steps sequentially; nested steps are supported, concurrent steps are not. + * Await steps sequentially. Steps cannot be nested inside other step callbacks. */ step(name: string, callback: () => PromiseLike | void): Promise; - /** Run a named cursor step with the last JSON cursor, or null initially. */ + /** + * Run a named cursor step with the last JSON cursor, or null initially. + * Steps cannot be nested inside other step callbacks. A saved cursor-only + * step must be reached before the attempt completes or another step checkpoints. + */ stepWithCursor( name: string, callback: (cursor: JsonValue | null) => PromiseLike | void diff --git a/js/src/conformance.test.ts b/js/src/conformance.test.ts index 7a8186b57..da49aa507 100644 --- a/js/src/conformance.test.ts +++ b/js/src/conformance.test.ts @@ -137,6 +137,88 @@ describe("Go resumable fixtures", () => { }); }); +const cursorOnlyCases = (await readFixture( + "resumable_cursor_only" +)) as unknown as { + readonly cases: readonly { + readonly actions: readonly { + readonly fail: boolean; + readonly kind: string; + readonly name: string; + readonly set_cursor: boolean; + }[]; + readonly expected_error: string; + readonly expected_runs: readonly string[]; + readonly expected_updates: JsonObject; + readonly metadata: JsonObject; + readonly name: string; + readonly worker_error: boolean; + }[]; +}; + +describe("Go cursor-only checkpoint fixtures", () => { + it.each(cursorOnlyCases.cases)("$name", async (fixture) => { + const update = vi.fn(); + const client = { jobs: { update } } as unknown as Client; + const resumable = new Resumable(client, { + ...job(), + metadata: fixture.metadata, + }); + const runs: string[] = []; + let failure: unknown; + try { + for (const action of fixture.actions) { + const callback = async () => { + runs.push(action.name); + if (action.kind === "checkpoint") + await resumable.checkpoint({ tx: {} }); + if (action.set_cursor) resumable.setCursor(7); + if (action.fail) throw new Error("step interrupted"); + }; + if (action.kind === "cursor") { + await resumable.stepWithCursor(action.name, async (cursor) => { + expect(cursor).toEqual( + (fixture.metadata["river:resumable_cursor"] as JsonObject)[ + action.name + ] ?? null + ); + await callback(); + }); + } else await resumable.step(action.name, callback); + } + if (fixture.worker_error) throw new Error("worker interrupted"); + } catch (error: unknown) { + failure = error; + } + const finished = resumable.finish(failure !== undefined); + const error = failure ?? finished.error; + if (fixture.expected_error.length > 0) { + expect(error).toBeInstanceOf(Error); + expect((error as Error).message).toContain(fixture.expected_error); + } else expect(error).toBeNull(); + expect(finished.metadata).toEqual(fixture.expected_updates); + expect(runs).toEqual(fixture.expected_runs); + expect(update).not.toHaveBeenCalled(); + if ( + fixture.expected_error.length > 0 && + Object.keys(fixture.expected_updates).length === 0 + ) { + const retry = new Resumable(client, { + ...job(), + metadata: { ...fixture.metadata, ...finished.metadata }, + }); + for (const [name, expected] of Object.entries( + fixture.metadata["river:resumable_cursor"] as JsonObject + )) { + await retry.stepWithCursor(name, (cursor) => { + expect(cursor).toEqual(expected); + }); + } + expect(retry.finish(false).error).toBeNull(); + } + }); +}); + describe("Go unique-key fixtures", () => { it.each(uniqueKeys.cases)("$name", (fixture) => { const now = Temporal.Instant.from(fixture.now); diff --git a/js/src/resumable-nested.test.ts b/js/src/resumable-nested.test.ts new file mode 100644 index 000000000..da377c71d --- /dev/null +++ b/js/src/resumable-nested.test.ts @@ -0,0 +1,87 @@ +import { describe, expect, it, vi } from "vitest"; + +import type { Client } from "./client.js"; +import type { JsonObject } from "./json.js"; +import type { JobRow } from "./job.js"; +import { Resumable } from "./resumable.js"; + +describe("nested resumable steps", () => { + it.each([ + { parent: "step", child: "step" }, + { parent: "step", child: "stepWithCursor" }, + { parent: "stepWithCursor", child: "step" }, + { parent: "stepWithCursor", child: "stepWithCursor" }, + ] as const)( + "rejects $child inside $parent before recording progress", + async ({ parent, child }) => { + const resumable = new Resumable({} as Client, job({})); + const childCallback = vi.fn(); + await expect( + resumable[parent]("parent", () => + resumable[child]("child", childCallback) + ) + ).rejects.toThrow("resumable steps cannot be nested"); + expect(childCallback).not.toHaveBeenCalled(); + expect(resumable.finish(false).error?.message).toBe( + "resumable steps cannot be nested" + ); + expect(resumable.finish(false).metadata).toEqual({}); + expect(() => resumable.setCursor(1)).toThrow("inside stepWithCursor"); + } + ); + + it.each(["step", "stepWithCursor"] as const)( + "does not complete a %s parent that catches the nesting error", + async (parent) => { + const resumable = new Resumable({} as Client, job({})); + const later = vi.fn(); + await resumable.step("before", () => undefined); + await resumable[parent]("parent", async () => { + if (parent === "stepWithCursor") resumable.setCursor(1); + await expect( + resumable.step("child", () => { + throw new Error("must not run"); + }) + ).rejects.toThrow("resumable steps cannot be nested"); + }); + await expect(resumable.step("after", later)).rejects.toThrow( + "resumable steps cannot be nested" + ); + expect(later).not.toHaveBeenCalled(); + const finished = resumable.finish(false); + expect(finished.error?.message).toBe("resumable steps cannot be nested"); + expect(finished.metadata).toEqual( + parent === "stepWithCursor" + ? { + "river:resumable_step": "before", + "river:resumable_cursor": { parent: 1 }, + } + : { "river:resumable_step": "before" } + ); + } + ); +}); + +function job(metadata: JsonObject): JobRow { + const now = Temporal.Instant.from("2026-10-24T12:00:00Z"); + return { + args: {}, + attempt: 1, + attemptedAt: now, + attemptedBy: ["test"], + createdAt: now, + errors: [], + finalizedAt: null, + id: 1n, + kind: "nested", + maxAttempts: 25, + metadata, + priority: 1, + queue: "default", + scheduledAt: now, + state: "running", + tags: [], + uniqueKey: null, + uniqueStates: [], + }; +} diff --git a/js/src/resumable.test.ts b/js/src/resumable.test.ts index 432dee75d..85a86c9f2 100644 --- a/js/src/resumable.test.ts +++ b/js/src/resumable.test.ts @@ -80,21 +80,6 @@ describe("Resumable", () => { expect(resumable.finish(false).error).toBeNull(); }); - it("restores the enclosing step after a nested step", async () => { - const resumable = new Resumable(client(), job({})); - await resumable - .stepWithCursor("outer", async () => { - await resumable.step("inner", () => undefined); - resumable.setCursor({ offset: 7 }); - throw new Error("retry outer"); - }) - .catch(() => undefined); - expect(resumable.finish(false).metadata).toEqual({ - "river:resumable_step": "inner", - "river:resumable_cursor": { outer: { offset: 7 } }, - }); - }); - it("retains progress and the step's own error when it is caught", async () => { const resumable = new Resumable(client(), job({})); const cause = new Error("service unavailable"); diff --git a/js/src/resumable.ts b/js/src/resumable.ts index aeb57d899..a795dfc32 100644 --- a/js/src/resumable.ts +++ b/js/src/resumable.ts @@ -31,6 +31,7 @@ export class Resumable { readonly #client: Client; readonly #cursors = new Map(); readonly #job: JobRow; + readonly #pendingCursorSteps = new Set(); readonly #resumeStep: string | null; #completedStep: string | null = null; #cursorsDirty = false; @@ -65,45 +66,56 @@ export class Resumable { if (cursor !== undefined) this.#cursors.set(name, cursor); } } + // A step marker may accompany stale cursors written by older clients. + // Require only targets from cursor-only checkpoints. + if (this.#resumeStep === null) { + for (const name of this.#cursors.keys()) + this.#pendingCursorSteps.add(name); + } } /** * Run a named step unless an earlier failed attempt completed it. - * Await steps sequentially; nested steps are supported, concurrent steps are not. + * Await steps sequentially. Steps cannot be nested inside other step callbacks. */ async step( name: string, callback: () => PromiseLike | void ): Promise { - const previousStepName = this.#stepName; if (!this.#begin(name, false)) return; try { await callback(); + if (this.#failure !== null) return; this.#completedStep = name; } catch (cause: unknown) { this.#failure = stepError(name, cause); throw cause; } finally { - this.#stepName = previousStepName; + this.#stepName = null; } } - /** Run a named cursor step with the last JSON cursor, or null initially. */ + /** + * Run a named cursor step with the last JSON cursor, or null initially. + * Steps cannot be nested inside other step callbacks. A saved cursor-only + * step must be reached before the attempt completes or another step checkpoints. + */ async stepWithCursor( name: string, callback: (cursor: JsonValue | null) => PromiseLike | void ): Promise { - const previousStepName = this.#stepName; if (!this.#begin(name, true)) return; try { + this.#pendingCursorSteps.delete(name); await callback(this.#cursors.get(name) ?? null); + if (this.#failure !== null) return; this.#completedStep = name; if (this.#cursors.delete(name)) this.#cursorsDirty = true; } catch (cause: unknown) { this.#failure = stepError(name, cause); throw cause; } finally { - this.#stepName = previousStepName; + this.#stepName = null; } } @@ -133,6 +145,12 @@ export class Resumable { "resumable checkpoint can only be set inside a resumable step" ); } + const pending = this.#pendingCursorStep(); + if (pending !== undefined) { + throw new LifecycleError( + `resumable checkpoint cannot precede saved cursor step ${JSON.stringify(pending)}` + ); + } const cursors = new Map(this.#cursors); if (options.cursor !== undefined) { cursors.set(this.#stepName, toJsonValue(options.cursor)); @@ -167,6 +185,17 @@ export class Resumable { `resumable step ${JSON.stringify(this.#resumeStep)} not found in worker` ); } + const pending = this.#pendingCursorStep(); + if (pending !== undefined) { + if (!workerFailed && this.#failure === null) { + this.#failure = new LifecycleError( + `resumable cursor step ${JSON.stringify(pending)} not found in worker` + ); + } + // Preserve the original cursor-only checkpoint so restoring the missing + // step can recover; later steps must not advance its skip boundary. + return { error: this.#failure, metadata: {} }; + } if (!workerFailed && this.#failure === null) { return { error: null, metadata: {} }; } @@ -181,6 +210,16 @@ export class Resumable { throw new ValidationError("resumable step name is empty"); } if (this.#failure !== null) throw this.#failure; + if (this.#stepName !== null) { + this.#failure = new LifecycleError("resumable steps cannot be nested"); + throw this.#failure; + } + if (!cursorStep && this.#pendingCursorSteps.has(name)) { + this.#failure = new LifecycleError( + `resumable cursor step ${JSON.stringify(name)} requires a cursor step` + ); + throw this.#failure; + } if (this.#allNames.has(name)) { this.#failure = new ValidationError( `duplicate resumable step name ${JSON.stringify(name)}` @@ -209,6 +248,10 @@ export class Resumable { } return metadata; } + + #pendingCursorStep(): string | undefined { + return [...this.#pendingCursorSteps].sort()[0]; + } } /** Construct an attempt-scoped resumable coordinator for first-party tooling. */ diff --git a/resumable.go b/resumable.go index e9631702c..74db57e2c 100644 --- a/resumable.go +++ b/resumable.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "slices" "github.com/riverqueue/river/internal/riverplugin" ) @@ -49,6 +50,7 @@ type StepOpts struct{} // an earlier attempt already completed it successfully. // Step names must be unique across all ResumableStep and ResumableStepCursor // calls in the same Worker execution. +// Steps cannot be nested inside other step callbacks. // // After a step returns an error, no subsequent steps will be run and the // overall job will be marked as failed with that error. Be careful to put all @@ -64,6 +66,10 @@ func ResumableStep(ctx context.Context, name string, opts *StepOpts, stepFunc fu if !registerResumableStepName(state, name) { return } + if slices.Contains(state.PendingCursorSteps, name) { + state.Err = fmt.Errorf("river: resumable cursor step %q requires a cursor step", name) + return + } if !state.ResumeMatched { if name == state.ResumeStep { @@ -73,14 +79,20 @@ func ResumableStep(ctx context.Context, name string, opts *StepOpts, stepFunc fu return } - previousStepName := state.StepName + state.StepActive = true state.StepName = name - defer func() { state.StepName = previousStepName }() + defer func() { + state.StepActive = false + state.StepName = "" + }() if err := stepFunc(ctx); err != nil { state.Err = err return } + if state.Err != nil { + return + } state.CompletedStep = name } @@ -89,8 +101,11 @@ func ResumableStep(ctx context.Context, name string, opts *StepOpts, stepFunc fu // cursor value from an earlier failed attempt, if one was recorded with // ResumableSetCursor. If no step has completed yet, the retry starts at the // beginning and supplies the saved cursor when it reaches its named step. +// A cursor-only checkpoint must be reached through ResumableStepCursor before +// the attempt can complete or another step can persist a transactional checkpoint. // Step names must be unique across all ResumableStep and ResumableStepCursor // calls in the same Worker execution. +// Steps cannot be nested inside other step callbacks. // // The cursor type T is user-specified. It may be a primitive value like an // integer ID, or a more complex type like a struct with multiple fields. It's @@ -140,14 +155,22 @@ func ResumableStepCursor[TCursor any](ctx context.Context, name string, opts *St } } - previousStepName := state.StepName + state.StepActive = true state.StepName = name - defer func() { state.StepName = previousStepName }() + defer func() { + state.StepActive = false + state.StepName = "" + }() + + state.PendingCursorSteps = slices.DeleteFunc(state.PendingCursorSteps, func(pending string) bool { return pending == name }) if err := stepFunc(ctx, cursor); err != nil { state.Err = err return } + if state.Err != nil { + return + } state.CompletedStep = name if _, ok := state.Cursors[name]; ok { @@ -166,6 +189,11 @@ func mustResumableState(ctx context.Context) *riverplugin.ResumableState { } func registerResumableStepName(state *riverplugin.ResumableState, name string) bool { + if state.StepActive { + state.Err = errors.New("river: resumable steps cannot be nested") + return false + } + if _, ok := state.AllStepNames[name]; ok { state.Err = fmt.Errorf("river: duplicate resumable step name %q", name) return false diff --git a/resumable_step_tx.go b/resumable_step_tx.go index c10d7b176..65602db45 100644 --- a/resumable_step_tx.go +++ b/resumable_step_tx.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "maps" "github.com/riverqueue/river/internal/execution" @@ -25,6 +26,8 @@ import ( // // Must be called from within a ResumableStep or ResumableStepCursor callback. // The current step name to persist is read from context. +// On a cursor-only retry, checkpointing another step returns an error until +// the saved cursor step is reached. // // Commit tx before the step callback returns successfully, and return any // transaction or commit error from that callback. A successful callback marks @@ -47,6 +50,8 @@ func ResumableSetStepTx[TDriver riverdriver.Driver[TTx], TTx any, TArgs JobArgs] // // Must be called from within a ResumableStepCursor callback. The current step // name to persist is read from context. +// On a cursor-only retry, checkpointing another step returns an error until +// the saved cursor step is reached. // // Commit tx before the step callback returns successfully, and return any // transaction or commit error from that callback. A successful callback marks @@ -73,6 +78,10 @@ func resumableSetStepTx[TTx any, TArgs JobArgs](ctx context.Context, tx TTx, job return nil, errors.New("not inside a resumable step; must be called from within ResumableStep or ResumableStepCursor") } + if len(state.PendingCursorSteps) > 0 { + return nil, fmt.Errorf("river: resumable checkpoint cannot precede saved cursor step %q", state.PendingCursorSteps[0]) + } + step := state.StepName client := ClientFromContext[TTx](ctx) diff --git a/resumable_step_tx_test.go b/resumable_step_tx_test.go index 881577212..72d8ee1af 100644 --- a/resumable_step_tx_test.go +++ b/resumable_step_tx_test.go @@ -102,6 +102,122 @@ func TestResumableAttemptMetadata(t *testing.T) { } } +func TestResumableCursorOnlyRecovery(t *testing.T) { + t.Parallel() + + type args struct { + testutil.JobArgsReflectKind[args] + } + + ctx := context.Background() + tx := riverdbtest.TestTxPgx(ctx, t) + exec := riverpgxv5.New(nil).UnwrapExecutor(tx) + client, err := NewClient(riverpgxv5.New(nil), &Config{}) + require.NoError(t, err) + ctx = context.WithValue(ctx, rivercommon.ContextKeyClient{}, client) + metadata := []byte(`{"river:resumable_cursor":{"loop":0}}`) + row := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Metadata: metadata, State: new(rivertype.JobStateRunning)}) + updates := make(map[string]any) + ctx = context.WithValue(ctx, jobexecutor.ContextKeyMetadataUpdates, updates) + + err = (&riverplugin.ResumableMiddleware{}).Work(ctx, row, func(ctx context.Context) error { + ResumableStep(ctx, "later", nil, func(context.Context) error { return nil }) + return nil + }) + require.ErrorContains(t, err, `resumable cursor step "loop" not found`) + encodedUpdates, err := json.Marshal(updates) + require.NoError(t, err) + saved, err := exec.JobUpdate(ctx, &riverdriver.JobUpdateParams{ID: row.ID, Metadata: encodedUpdates, MetadataDoMerge: true}) + require.NoError(t, err) + require.JSONEq(t, string(metadata), string(saved.Metadata)) + + err = (&riverplugin.ResumableMiddleware{}).Work(ctx, saved, func(ctx context.Context) error { + ResumableStepCursor(ctx, "loop", nil, func(ctx context.Context, cursor int) error { + require.Zero(t, cursor) + _, err := ResumableSetStepCursorTx[*riverpgxv5.Driver](ctx, tx, &Job[args]{JobRow: saved}, 1) + return err + }) + return nil + }) + require.NoError(t, err) +} + +func TestResumableNestedStepCheckpointTx(t *testing.T) { + t.Parallel() + + type args struct { + testutil.JobArgsReflectKind[args] + } + type testBundle struct { + exec riverdriver.Executor + tx pgx.Tx + updates map[string]any + } + setup := func(t *testing.T) (*testBundle, context.Context) { + t.Helper() + + ctx := context.Background() + tx := riverdbtest.TestTxPgx(ctx, t) + client, err := NewClient(riverpgxv5.New(nil), &Config{Logger: riversharedtest.Logger(t)}) + require.NoError(t, err) + updates := make(map[string]any) + ctx = context.WithValue(ctx, rivercommon.ContextKeyClient{}, client) + ctx = context.WithValue(ctx, jobexecutor.ContextKeyMetadataUpdates, updates) + return &testBundle{exec: riverpgxv5.New(nil).UnwrapExecutor(tx), tx: tx, updates: updates}, ctx + } + + for _, commit := range []bool{true, false} { + name := "RolledBackParentCursorIsNotReplayed" + if commit { + name = "CommittedParentCursorIsPreserved" + } + t.Run(name, func(t *testing.T) { + t.Parallel() + + bundle, ctx := setup(t) + job := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: new(rivertype.JobStateRunning)}) + err := (&riverplugin.ResumableMiddleware{}).Work(ctx, job, func(ctx context.Context) error { + ResumableStepCursor(ctx, "parent", nil, func(ctx context.Context, _ int) error { + tx, err := bundle.tx.Begin(ctx) + require.NoError(t, err) + defer tx.Rollback(ctx) + _, err = ResumableSetStepCursorTx[*riverpgxv5.Driver](ctx, tx, &Job[args]{JobRow: job}, 1) + require.NoError(t, err) + ResumableStepCursor(ctx, "child", nil, func(ctx context.Context, _ int) error { + t.Fatal("nested child must not run") + return nil + }) + if commit { + require.NoError(t, tx.Commit(ctx)) + } else { + require.NoError(t, tx.Rollback(ctx)) + } + return nil + }) + return nil + }) + require.EqualError(t, err, "river: resumable steps cannot be nested") + + beforeResult, err := bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID}) + require.NoError(t, err) + if !commit { + require.JSONEq(t, `{}`, string(beforeResult.Metadata)) + } + updates, err := json.Marshal(bundle.updates) + require.NoError(t, err) + // Rejecting the child must preserve a committed parent checkpoint + // without replaying one that rolled back. + result, err := bundle.exec.JobUpdate(ctx, &riverdriver.JobUpdateParams{ID: job.ID, Metadata: updates, MetadataDoMerge: true}) + require.NoError(t, err) + if commit { + require.JSONEq(t, `{"river:resumable_step":"parent","river:resumable_cursor":{"parent":1}}`, string(result.Metadata)) + } else { + require.JSONEq(t, `{}`, string(result.Metadata)) + } + }) + } +} + func TestResumableSetStepTx(t *testing.T) { t.Parallel() diff --git a/resumable_test.go b/resumable_test.go index 8ee55cd34..a59542faf 100644 --- a/resumable_test.go +++ b/resumable_test.go @@ -15,6 +15,73 @@ import ( "github.com/riverqueue/river/rivertype" ) +func TestResumableNestedStep(t *testing.T) { + t.Parallel() + + for _, testCase := range []struct { + name string + childCursor bool + parentCursor bool + parentName string + }{ + {name: "CursorInCursor", childCursor: true, parentCursor: true, parentName: "parent"}, + {name: "CursorInPlain", childCursor: true, parentName: "parent"}, + {name: "EmptyParentName"}, + {name: "PlainInCursor", parentCursor: true, parentName: "parent"}, + {name: "PlainInPlain", parentName: "parent"}, + } { + t.Run(testCase.name, func(t *testing.T) { + t.Parallel() + + updates := make(map[string]any) + ctx := context.WithValue(context.Background(), jobexecutor.ContextKeyMetadataUpdates, updates) + job := &rivertype.JobRow{Metadata: []byte(`{}`)} + + err := (&riverplugin.ResumableMiddleware{}).Work(ctx, job, func(ctx context.Context) error { + ResumableStep(ctx, "before", nil, func(context.Context) error { return nil }) + parentFunc := func(ctx context.Context) { + if testCase.childCursor { + ResumableStepCursor(ctx, "child", nil, func(context.Context, int) error { + t.Fatal("nested child must not run") + return nil + }) + } else { + ResumableStep(ctx, "child", nil, func(context.Context) error { + t.Fatal("nested child must not run") + return nil + }) + } + } + if testCase.parentCursor { + ResumableStepCursor(ctx, testCase.parentName, nil, func(ctx context.Context, _ int) error { + require.NoError(t, ResumableSetCursor(ctx, 1)) + parentFunc(ctx) + return nil + }) + } else { + ResumableStep(ctx, testCase.parentName, nil, func(ctx context.Context) error { + parentFunc(ctx) + return nil + }) + } + ResumableStep(ctx, "after", nil, func(context.Context) error { + t.Fatal("steps after the nesting error must not run") + return nil + }) + return nil + }) + require.EqualError(t, err, "river: resumable steps cannot be nested") + metadata, err := json.Marshal(updates) + require.NoError(t, err) + if testCase.parentCursor { + require.JSONEq(t, `{"river:resumable_step":"before","river:resumable_cursor":{"parent":1}}`, string(metadata)) + } else { + require.JSONEq(t, `{"river:resumable_step":"before"}`, string(metadata)) + } + }) + } +} + func TestResumableStep(t *testing.T) { t.Parallel() diff --git a/ruby/CHANGELOG.md b/ruby/CHANGELOG.md index de46723b2..375c49ab8 100644 --- a/ruby/CHANGELOG.md +++ b/ruby/CHANGELOG.md @@ -20,6 +20,8 @@ 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). + - Support large SQLite metadata updates without reaching the expression-depth limit, preserving top-level replacement and JSON null values. [PR #1533](https://github.com/riverqueue/river/pull/1533). - A `River::Config` without an explicit `id:` generates its default client ID on first use in each process, so clients built before a fork (as with preloading Puma, Unicorn, or Pitchfork servers) no longer share one ID across forked workers. Explicit IDs are unchanged, and `Config#with` copies share the original's ID. [PR #1498](https://github.com/riverqueue/river/pull/1498). diff --git a/ruby/lib/resumable.rb b/ruby/lib/resumable.rb index efb606fc5..41053cca5 100644 --- a/ruby/lib/resumable.rb +++ b/ruby/lib/resumable.rb @@ -11,6 +11,7 @@ class ResumableState attr_accessor :completed_step attr_reader :cursors attr_accessor :cursors_dirty + attr_reader :pending_cursor_steps attr_accessor :resume_matched attr_reader :resume_step attr_accessor :step_name @@ -24,6 +25,8 @@ def initialize(metadata) @step_name = nil @resume_matched = @resume_step.to_s.empty? + # Step markers may accompany stale cursors from older clients. + @pending_cursor_steps = @resume_matched ? @cursors.keys.sort : [] end def register(name) @@ -37,6 +40,9 @@ class Job # Internal runtime boundary: attach progress only to attempts that did not # complete, matching River Go's persisted metadata format. def __capture_resumable_metadata! + # Keep missing cursor steps reachable when the worker restores them. + return unless @resumable_state.pending_cursor_steps.empty? + if @resumable_state.cursors_dirty @metadata_updates[RESUMABLE_CURSOR_METADATA_KEY] = @resumable_state.cursors.empty? ? nil : @resumable_state.cursors.dup end @@ -49,6 +55,9 @@ def __capture_resumable_metadata! # Internal runtime boundary: validate that the recorded resume point still # exists in the worker after it has returned. def __finish_resumable_work! + if (pending = @resumable_state.pending_cursor_steps.first) + raise Error, "resumable cursor step #{pending.inspect} not found in worker" + end if @resumable_state.resume_step && !@resumable_state.resume_matched raise Error, "resumable step #{@resumable_state.resume_step.inspect} not found in worker" end @@ -56,7 +65,8 @@ def __finish_resumable_work! # Immediately checkpoints the current step and cursor progress. Wrap the # call in Driver#transaction alongside application writes when an atomic - # checkpoint is needed. + # checkpoint is needed. When resuming cursor-only progress, reach its saved + # cursor step before checkpointing another step. # # Commit the transaction before the step block returns # successfully, and let transaction and commit errors propagate out of it. @@ -67,6 +77,10 @@ def resumable_checkpoint(cursor: RESUMABLE_CURSOR_UNSET) raise Error, "resumable step can only be persisted inside a resumable step" unless step_name raise Error, "job must be running" unless row.state == JOB_STATE_RUNNING + if (pending = @resumable_state.pending_cursor_steps.first) + raise Error, "resumable checkpoint cannot precede saved cursor step #{pending.inspect}" + end + cursors = @resumable_state.cursors.dup cursors[step_name] = JSON.parse(JSON.generate(cursor)) unless cursor.equal?(RESUMABLE_CURSOR_UNSET) updates = { @@ -122,6 +136,9 @@ def resumable_step_cursor(name, default: nil, &block) private def run_resumable_step(name, cursor:, default:) raise Error, "resumable steps cannot be nested" if @resumable_state.step_name raise ArgumentError, "resumable step name must be non-empty" if name.empty? + if !cursor && @resumable_state.pending_cursor_steps.include?(name) + raise Error, "resumable cursor step #{name.inspect} requires a cursor step" + end @resumable_state.register(name) unless @resumable_state.resume_matched @@ -137,6 +154,7 @@ def resumable_step_cursor(name, default: nil, &block) @resumable_state.step_name = name begin value = (cursor && @resumable_state.cursors.key?(name)) ? @resumable_state.cursors[name] : default + @resumable_state.pending_cursor_steps.delete(name) if cursor result = cursor ? yield(value) : yield @resumable_state.completed_step = name if cursor && @resumable_state.cursors.key?(name) diff --git a/ruby/sig/runtime.rbs b/ruby/sig/runtime.rbs index ab94b3a7e..0610ab634 100644 --- a/ruby/sig/runtime.rbs +++ b/ruby/sig/runtime.rbs @@ -137,6 +137,7 @@ module River attr_accessor completed_step: String? attr_reader cursors: Hash[String, untyped] attr_accessor cursors_dirty: bool + attr_reader pending_cursor_steps: Array[String] attr_accessor resume_matched: bool attr_reader resume_step: String? attr_accessor step_name: String? diff --git a/ruby/spec/client_driver_shared_examples.rb b/ruby/spec/client_driver_shared_examples.rb index af7c01987..674e09537 100644 --- a/ruby/spec/client_driver_shared_examples.rb +++ b/ruby/spec/client_driver_shared_examples.rb @@ -241,6 +241,28 @@ it_behaves_like "cancellation while draining" it_behaves_like "externally claimed job finalization" it_behaves_like "transactional job completion" + it "keeps an omitted cursor-only step recoverable" do + received = [] + worker.define_method(:work) do |job| + if job.attempt > 1 + job.resumable_step_cursor(:loop) do |cursor| + received << cursor + job.client.driver.transaction { job.resumable_checkpoint(cursor: 1) } + end + end + job.resumable_step(:later) {} + end + metadata = {River::RESUMABLE_CURSOR_METADATA_KEY => {"loop" => 0}} + row = client.insert(River::JobArgsHash.new("driver_e2e", {}), insert_opts: River::InsertOpts.new(metadata: metadata, queue: "driver_e2e")).job + failed = River::Testing.perform_job(client, row.id) + expect(failed.outcome).to eq(:retried) + expect(failed.job.metadata).to include(metadata) + expect(failed.job.metadata).not_to have_key(River::RESUMABLE_STEP_METADATA_KEY) + client.job_retry row.id + expect(River::Testing.perform_job(client, row.id).outcome).to eq(:completed) + expect(received).to eq([0]) + end + [false, true].product([false, true]).each do |with_cursor, rollback| it "resumes after a #{rollback ? "rolled-back" : "committed"} #{with_cursor ? "cursor" : "step"} checkpoint" do received = [] diff --git a/ruby/spec/resumable_spec.rb b/ruby/spec/resumable_spec.rb index 0cd982cfe..269dc5ef3 100644 --- a/ruby/spec/resumable_spec.rb +++ b/ruby/spec/resumable_spec.rb @@ -51,6 +51,54 @@ def build_job(row: build_row, driver: Object.new) River::Job.new(Struct.new(:driver).new(driver), row) end + fixtures = JSON.parse(File.read(File.expand_path("../../conformance/testdata/resumable_cursor_only.json", __dir__))) + fixtures.fetch("cases").each do |fixture| + it "matches cursor-only checkpoint fixture #{fixture.fetch("name")}" do + metadata = fixture.fetch("metadata") + job = build_job(row: build_row(metadata: metadata)) + runs = [] + failure = nil + begin + fixture.fetch("actions").each do |action| + name = action.fetch("name") + callback = proc do + runs << name + job.resumable_checkpoint if action.fetch("kind") == "checkpoint" + job.resumable_set_cursor(7) if action.fetch("set_cursor") + raise "step interrupted" if action.fetch("fail") + end + if action.fetch("kind") == "cursor" + job.resumable_step_cursor(name) do |cursor| + expect(cursor).to eq(metadata.fetch(River::RESUMABLE_CURSOR_METADATA_KEY)[name]) + callback.call + end + else + job.resumable_step(name, &callback) + end + end + raise "worker interrupted" if fixture.fetch("worker_error") + job.__finish_resumable_work! + rescue => error + failure = error + end + if fixture.fetch("expected_error").empty? + expect(failure).to be_nil + else + expect(failure.message).to include(fixture.fetch("expected_error")) + end + job.__capture_resumable_metadata! if failure + expect(runs).to eq(fixture.fetch("expected_runs")) + expect(job.metadata_updates).to eq(fixture.fetch("expected_updates")) + if failure && job.metadata_updates.empty? + retry_job = build_job(row: build_row(metadata: metadata)) + metadata.fetch(River::RESUMABLE_CURSOR_METADATA_KEY).each do |name, expected| + retry_job.resumable_step_cursor(name) { |cursor| expect(cursor).to eq(expected) } + end + expect { retry_job.__finish_resumable_work! }.not_to raise_error + end + end + end + it "runs named steps in order and returns their values" do job = build_job calls = [] @@ -201,11 +249,11 @@ def build_job(row: build_row, driver: Object.new) end it "rejects nested steps without recording an unreachable resume point" do - [:resumable_step, :resumable_step_cursor].each do |method| + [:resumable_step, :resumable_step_cursor].product([:resumable_step, :resumable_step_cursor]).each do |parent, child| job = build_job expect do - job.resumable_step :outer do - job.public_send(method, :inner) { raise "must not run" } + job.public_send(parent, :outer) do + job.public_send(child, :inner) { raise "must not run" } end end.to raise_error(River::Error, /cannot be nested/) job.__capture_resumable_metadata! diff --git a/rust/CHANGELOG.md b/rust/CHANGELOG.md index a83ef5c2d..38f1d573c 100644 --- a/rust/CHANGELOG.md +++ b/rust/CHANGELOG.md @@ -16,10 +16,13 @@ Changes to River for Go are recorded in the [repository changelog](../CHANGELOG. ### 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). - **Breaking:** Attempt counts are now `i32` instead of `i16`, like River for Go's `int`: `JobRow::attempt` and `max_attempts`, `AttemptError::attempt` and `AttemptError::new`, `InsertOpts::max_attempts` and `with_max_attempts`, `InsertParams::max_attempts`, `ClientBuilder::default_max_attempts`, `MAX_ATTEMPTS_DEFAULT`, and `riverqueue-test`'s builders. On SQLite, which stores native integers, a job with more than 32,767 maximum attempts inserts and reads back unchanged, as in Go. On Postgres, whose columns are 16-bit, an insert clamps `max_attempts` to 32,767 like Go's drivers instead of failing. [PR #1504](https://github.com/riverqueue/river/pull/1504). ### 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). - Preserve first-step resumable cursors on failure and step progress on snooze. Transactional checkpoints no longer leak into deferred updates after rollback or a failed write; rolling back a successful checkpoint write may replay earlier unpersisted steps. diff --git a/rust/riverqueue/src/worker.rs b/rust/riverqueue/src/worker.rs index 6d81032ab..a83de48aa 100644 --- a/rust/riverqueue/src/worker.rs +++ b/rust/riverqueue/src/worker.rs @@ -1,7 +1,7 @@ //! Typed worker interfaces and registration. use std::{ - collections::{HashMap, HashSet}, + collections::{BTreeSet, HashMap, HashSet}, error::Error as StdError, future::Future, sync::{Arc, Mutex, MutexGuard, PoisonError}, @@ -159,15 +159,16 @@ impl WorkContext { /// Runs a named resumable step, skipping work completed by an earlier /// failed attempt. /// - /// Await steps sequentially. Nested steps are supported, but concurrent - /// steps do not define a checkpoint order. A step may fail with any error - /// convertible into [`BoxError`], including `anyhow::Error`; it is + /// Await steps sequentially. Steps cannot be nested inside other step + /// callbacks. A step may fail with any error convertible into [`BoxError`], + /// including `anyhow::Error`; it is /// returned as the source of [`Error::ResumableStep`]. /// /// # Errors /// /// Returns [`Error::ResumableStep`] with the step's error, and a runtime - /// error when the context doesn't belong to a job being worked. + /// error when steps are nested or the context doesn't belong to a job + /// being worked. /// /// # Cancel safety /// @@ -181,12 +182,19 @@ impl WorkContext { Fut: Future>, { let guard = match self.begin_resumable_step(name, false)? { - StepAction::Run(previous) => RunningStep::new(self, name, previous), + StepAction::Run => RunningStep::new(self, name), StepAction::Skip => return Ok(()), }; let result = step().await; let mut state = guard.finish(); + if let Some(failure) = &state.failure { + return Err(Error::runtime_source( + "worker context", + failure.to_string(), + failure.clone(), + )); + } match result { Ok(()) => { state.completed_step = Some(name.to_owned()); @@ -222,7 +230,7 @@ impl WorkContext { T: Default + DeserializeOwned, { let guard = match self.begin_resumable_step(name, true)? { - StepAction::Run(previous) => RunningStep::new(self, name, previous), + StepAction::Run => RunningStep::new(self, name), StepAction::Skip => return Ok(()), }; @@ -242,8 +250,16 @@ impl WorkContext { return Err(state.fail_step(name, Box::new(error))); } }; + self.lock_resumable().pending_cursor_steps.remove(name); let result = step(cursor).await; let mut state = guard.finish(); + if let Some(failure) = &state.failure { + return Err(Error::runtime_source( + "worker context", + failure.to_string(), + failure.clone(), + )); + } match result { Ok(()) => { state.completed_step = Some(name.to_owned()); @@ -367,6 +383,7 @@ impl WorkContext { } /// Returns checkpoint metadata without changing deferred progress. + /// A cursor-only target must be reached before another step checkpoints. fn resumable_checkpoint( &self, cursor: Option<&T>, @@ -378,6 +395,12 @@ impl WorkContext { "resumable checkpoint must be set inside a resumable step".to_owned(), ) })?; + if let Some(pending) = state.pending_cursor_steps.first() { + return Err(Error::runtime_context( + "worker context", + format!("resumable checkpoint cannot precede saved cursor step {pending:?}"), + )); + } let mut cursors = state.cursors.clone(); if let Some(cursor) = cursor { cursors.insert(step_name.clone(), serde_json::to_value(cursor)?); @@ -413,6 +436,22 @@ impl WorkContext { failure.clone(), )); } + if state.step_name.is_some() { + let message = "resumable steps cannot be nested".to_owned(); + state.failure = Some(WorkError::new(Box::new(Error::runtime_context( + "worker context", + message.clone(), + )))); + return Err(Error::runtime_context("worker context", message)); + } + if !cursor_step && state.pending_cursor_steps.contains(name) { + let message = format!("resumable cursor step {name:?} requires a cursor step"); + state.failure = Some(WorkError::new(Box::new(Error::runtime_context( + "worker context", + message.clone(), + )))); + return Err(Error::runtime_context("worker context", message)); + } if !state.all_step_names.insert(name.to_owned()) { let message = format!("duplicate resumable step name {name:?}"); state.failure = Some(WorkError::new(Box::new(Error::runtime_context( @@ -432,8 +471,8 @@ impl WorkContext { return Ok(StepAction::Skip); } } - let previous = state.step_name.replace(name.to_owned()); - Ok(StepAction::Run(previous)) + state.step_name = Some(name.to_owned()); + Ok(StepAction::Run) } fn current_job(&self) -> Result<(&Client, i64), Error> { @@ -512,6 +551,21 @@ impl WorkContext { ))) }) }); + let failure = failure.or_else(|| { + (!worker_failed) + .then(|| state.pending_cursor_steps.first()) + .flatten() + .map(|pending| { + WorkError::new(Box::new(Error::runtime_context( + "worker context", + format!("resumable cursor step {pending:?} not found in worker"), + ))) + }) + }); + // Unmatched cursor-only steps must stay reachable on the next attempt. + if !state.pending_cursor_steps.is_empty() { + return failure; + } if worker_failed || failure.is_some() { let mut updates = self.lock_metadata(); if let Some(completed_step) = &state.completed_step { @@ -553,6 +607,7 @@ struct ResumableState { cursors: Map, cursors_dirty: bool, failure: Option, + pending_cursor_steps: BTreeSet, resume_matched: bool, resume_step: Option, step_name: Option, @@ -566,6 +621,7 @@ impl Default for ResumableState { cursors: Map::new(), cursors_dirty: false, failure: None, + pending_cursor_steps: BTreeSet::new(), resume_matched: true, resume_step: None, step_name: None, @@ -599,6 +655,10 @@ impl ResumableState { } _ => {} } + // Step markers may accompany stale cursors from older clients. + if state.resume_step.is_none() { + state.pending_cursor_steps = state.cursors.keys().cloned().collect(); + } state } @@ -616,36 +676,34 @@ impl ResumableState { } enum StepAction { - Run(Option), + Run, Skip, } -/// A resumable step that has started. Finishing it restores the enclosing -/// step. Dropping it unfinished, when the step's future is dropped, also +/// A resumable step that has started. Finishing it clears the active step. +/// Dropping it unfinished, when the step's future is dropped, also /// forgets that the step started, so the step can run again in this attempt /// and no cursor can be set for it outside the step. struct RunningStep<'a> { context: &'a WorkContext, finished: bool, name: &'a str, - previous: Option, } impl<'a> RunningStep<'a> { - const fn new(context: &'a WorkContext, name: &'a str, previous: Option) -> Self { + const fn new(context: &'a WorkContext, name: &'a str) -> Self { Self { context, finished: false, name, - previous, } } - /// Restores the enclosing step and returns the locked state. + /// Clears the active step and returns the locked state. fn finish(mut self) -> MutexGuard<'a, ResumableState> { self.finished = true; let mut state = self.context.lock_resumable(); - state.step_name = self.previous.take(); + state.step_name = None; state } } @@ -654,7 +712,7 @@ impl Drop for RunningStep<'_> { fn drop(&mut self) { if !self.finished { let mut state = self.context.lock_resumable(); - state.step_name = self.previous.take(); + state.step_name = None; state.all_step_names.remove(self.name); } } @@ -1123,6 +1181,128 @@ mod tests { } } + #[tokio::test] + async fn resumable_cursor_only_checkpoint_in_target() { + let mut row = job_row(FunctionJobArgs::KIND, false); + row.metadata = json!({"river:resumable_cursor": {"loop": 0}}) + .as_object() + .unwrap() + .clone() + .into(); + let context = WorkContext::for_test_job(&row); + context + .resumable_step_with_cursor("loop", |cursor: Value| { + assert_eq!(cursor, json!(0)); + assert_eq!( + Value::Object(context.resumable_checkpoint(Some(&7)).unwrap()), + json!({"river:resumable_step": "loop", "river:resumable_cursor": {"loop": 7}}) + ); + std::future::ready(Ok::<(), Error>(())) + }) + .await + .unwrap(); + assert!(context.resumable_finish(false).is_none()); + } + + #[tokio::test] + async fn resumable_cursor_only_conformance() { + let fixture: Value = serde_json::from_str( + &std::fs::read_to_string( + std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../../conformance/testdata/resumable_cursor_only.json"), + ) + .unwrap(), + ) + .unwrap(); + for case in fixture["cases"].as_array().unwrap() { + let mut row = job_row(FunctionJobArgs::KIND, false); + row.metadata = case["metadata"].as_object().unwrap().clone().into(); + let context = WorkContext::for_test_job(&row); + let mut runs = Vec::new(); + let mut failure = None; + for action in case["actions"].as_array().unwrap() { + let name = action["name"].as_str().unwrap(); + let kind = action["kind"].as_str().unwrap(); + let mut callback = || { + runs.push(name); + if kind == "checkpoint" { + context.resumable_checkpoint::(None)?; + } + if action["set_cursor"].as_bool().unwrap() { + context.resumable_set_cursor(&7)?; + } + if action["fail"].as_bool().unwrap() { + return Err(Error::invalid_job("step interrupted")); + } + Ok::<(), Error>(()) + }; + let result = if kind == "cursor" { + context + .resumable_step_with_cursor(name, |cursor: Value| { + assert_eq!( + cursor, + case["metadata"][crate::METADATA_KEY_RESUMABLE_CURSOR][name] + ); + std::future::ready(callback()) + }) + .await + } else { + context + .resumable_step(name, || std::future::ready(callback())) + .await + }; + if let Err(error) = result { + failure = Some(error); + break; + } + } + if failure.is_none() && case["worker_error"].as_bool().unwrap() { + failure = Some(Error::invalid_job("worker interrupted")); + } + let finished = context.resumable_finish(failure.is_some()); + let actual_error = failure + .map(|error| crate::error::Chain(&error).to_string()) + .or_else(|| finished.map(|error| crate::error::Chain(&error).to_string())); + let expected_error = case["expected_error"].as_str().unwrap(); + if expected_error.is_empty() { + assert!(actual_error.is_none(), "{case:?}: {actual_error:?}"); + } else { + assert!( + actual_error.as_ref().unwrap().contains(expected_error), + "{case:?}: {actual_error:?}" + ); + } + assert_eq!( + serde_json::to_value(runs).unwrap(), + case["expected_runs"], + "{}", + case["name"] + ); + assert_eq!( + Value::Object(context.metadata_updates()), + case["expected_updates"], + "{}", + case["name"] + ); + if !expected_error.is_empty() && context.metadata_updates().is_empty() { + let retry = WorkContext::for_test_job(&row); + for (name, expected) in case["metadata"][crate::METADATA_KEY_RESUMABLE_CURSOR] + .as_object() + .unwrap() + { + retry + .resumable_step_with_cursor(name, |cursor: Value| { + assert_eq!(&cursor, expected); + std::future::ready(Ok::<(), Error>(())) + }) + .await + .unwrap(); + } + assert!(retry.resumable_finish(false).is_none()); + } + } + } + #[tokio::test] async fn resumable_cursor_decode_failure_is_sticky() { let mut row = job_row(FunctionJobArgs::KIND, false); @@ -1201,6 +1381,66 @@ mod tests { .unwrap(); } + #[tokio::test] + async fn resumable_caught_nested_step_error_does_not_complete_parent() { + for cursor in [false, true] { + let context = WorkContext::new(CancellationToken::new()); + context + .resumable_step("before", || async { Ok::<(), Error>(()) }) + .await + .unwrap(); + let parent = || async { + if cursor { + context.resumable_set_cursor(&1)?; + } + let error = context + .resumable_step("child", || async { + Err::<(), Error>(Error::invalid_job("nested child must not run")) + }) + .await + .unwrap_err(); + assert!( + error + .to_string() + .contains("resumable steps cannot be nested") + ); + Ok::<(), Error>(()) + }; + let result = if cursor { + context + .resumable_step_with_cursor("parent", |_: i64| parent()) + .await + } else { + context.resumable_step("parent", parent).await + }; + assert!( + result + .unwrap_err() + .to_string() + .contains("resumable steps cannot be nested") + ); + assert!( + context + .resumable_step("after", || async { + Err::<(), Error>(Error::invalid_job("later step must not run")) + }) + .await + .unwrap_err() + .to_string() + .contains("resumable steps cannot be nested") + ); + assert!(context.resumable_finish(false).is_some()); + assert_eq!( + Value::Object(context.metadata_updates()), + if cursor { + json!({"river:resumable_step": "before", "river:resumable_cursor": {"parent": 1}}) + } else { + json!({"river:resumable_step": "before"}) + } + ); + } + } + #[tokio::test] async fn resumable_first_step_cursor_survives_failure() { let row = job_row(FunctionJobArgs::KIND, false); @@ -1237,13 +1477,56 @@ mod tests { } #[tokio::test] - async fn resumable_nested_steps_restore_parent_and_error_sources() { + async fn resumable_nested_steps_are_rejected() { + for parent_cursor in [false, true] { + for child_cursor in [false, true] { + let context = WorkContext::new(CancellationToken::new()); + let parent = || async { + if child_cursor { + context + .resumable_step_with_cursor("child", |_: i64| async { + Err::<(), Error>(Error::invalid_job("nested child must not run")) + }) + .await + } else { + context + .resumable_step("child", || async { + Err::<(), Error>(Error::invalid_job("nested child must not run")) + }) + .await + } + }; + let result = if parent_cursor { + context + .resumable_step_with_cursor("parent", |_: i64| parent()) + .await + } else { + context.resumable_step("parent", parent).await + }; + assert!( + result + .unwrap_err() + .to_string() + .contains("resumable steps cannot be nested") + ); + assert!( + context + .resumable_finish(false) + .unwrap() + .to_string() + .contains("resumable steps cannot be nested") + ); + assert!(context.metadata_updates().is_empty()); + assert!(context.resumable_set_cursor(&1).is_err()); + } + } + } + + #[tokio::test] + async fn resumable_steps_preserve_error_sources() { let context = WorkContext::new(CancellationToken::new()); let error = context .resumable_step_with_cursor("outer", |_: i64| async { - context - .resumable_step("inner", || async { Ok::<_, Error>(()) }) - .await?; context.resumable_set_cursor(&7)?; Err::<(), _>(Error::ResumableStep { name: "source".to_owned(), @@ -1264,7 +1547,6 @@ mod tests { assert_eq!( context.metadata_updates(), json!({ - "river:resumable_step": "inner", "river:resumable_cursor": {"outer": 7} }) .as_object() diff --git a/rust/riverqueue/tests/resumable_nested.rs b/rust/riverqueue/tests/resumable_nested.rs new file mode 100644 index 000000000..156694515 --- /dev/null +++ b/rust/riverqueue/tests/resumable_nested.rs @@ -0,0 +1,238 @@ +//! Rejected nested steps cannot replay rolled-back transactional checkpoints. +#![cfg(any(all(feature = "postgres", river_postgres_tests), feature = "sqlite"))] + +#[cfg(all(feature = "postgres", river_postgres_tests))] +mod support; + +use riverqueue::{__private, BoxError, Client, JobArgs, JobRow, JobUpdateParams, WorkContext}; +#[cfg(feature = "sqlite")] +use riverqueue_migrate::SqliteMigrator; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; +#[cfg(feature = "sqlite")] +use sqlx::{SqlitePool, sqlite::SqlitePoolOptions}; + +#[derive(Clone, Debug, Deserialize, JobArgs, Serialize)] +#[river(kind = "nested_resumable")] +struct Args {} + +#[cfg(feature = "sqlite")] +#[tokio::test] +async fn cursor_only_sqlite_step_remains_recoverable() { + let (_pool, client, _, _) = setup().await; + assert_cursor_only_recovery(&client).await; +} + +#[cfg(all(feature = "postgres", river_postgres_tests))] +#[tokio::test] +async fn cursor_only_postgres_step_remains_recoverable() { + let database = support::PostgresSchema::new("cursor_only").await; + let client = Client::builder( + riverqueue::database::PostgresDatabase::new(database.pool.clone()) + .with_schema(database.schema.clone()), + ) + .build() + .unwrap(); + assert_cursor_only_recovery(&client).await; + database.cleanup().await; +} + +async fn assert_cursor_only_recovery(client: &Client) { + let inserted = client.insert(Args {}).await.unwrap(); + let metadata = json!({"river:resumable_cursor": {"loop": 0}}); + client + .jobs() + .update( + inserted.job.row.id, + JobUpdateParams::default().metadata(metadata.as_object().unwrap().clone()), + ) + .await + .unwrap(); + let (row, context) = __private::claim_job_for_test(client, inserted.job.row.id) + .await + .unwrap(); + context + .resumable_step("later", || async { Ok::<(), BoxError>(()) }) + .await + .unwrap(); + assert!( + __private::work_context_resumable_finish(&context, false) + .unwrap() + .to_string() + .contains("resumable cursor step \"loop\" not found") + ); + let saved = client + .jobs() + .update( + row.id, + JobUpdateParams::default().metadata(__private::work_context_metadata_updates(&context)), + ) + .await + .unwrap(); + let saved_metadata = serde_json::to_value(&saved.metadata).unwrap(); + assert_eq!( + saved_metadata["river:resumable_cursor"], + metadata["river:resumable_cursor"] + ); + assert!(saved_metadata.get("river:resumable_step").is_none()); + let retry = __private::work_context_for_job(&saved); + retry + .resumable_step_with_cursor("loop", |cursor: Value| async move { + assert_eq!(cursor, json!(0)); + Ok::<(), BoxError>(()) + }) + .await + .unwrap(); + assert!(__private::work_context_resumable_finish(&retry, false).is_none()); +} + +#[cfg(feature = "sqlite")] +#[tokio::test] +async fn committed_parent_cursor_is_preserved() { + assert_parent_checkpoint(true).await; +} + +#[cfg(all(feature = "postgres", river_postgres_tests))] +#[tokio::test] +async fn postgres_parent_checkpoints_survive_rejected_nesting() { + for commit in [false, true] { + let database = support::PostgresSchema::new("nested_checkpoint").await; + let pool = database.pool.clone(); + let client = Client::builder( + riverqueue::database::PostgresDatabase::new(pool.clone()) + .with_schema(database.schema.clone()), + ) + .build() + .unwrap(); + let inserted = client.insert(Args {}).await.unwrap(); + let (row, context) = __private::claim_job_for_test(&client, inserted.job.row.id) + .await + .unwrap(); + let result = context + .resumable_step_with_cursor("parent", |_: i64| async { + let mut tx = pool.begin().await?; + context.resumable_set_step_cursor_tx(&mut tx, &1).await?; + let child_result = context + .resumable_step_with_cursor("child", |_: i64| async { + Err::<(), BoxError>("nested child must not run".into()) + }) + .await; + if commit { + tx.commit().await?; + } else { + tx.rollback().await?; + } + child_result?; + Ok::<(), BoxError>(()) + }) + .await; + assert!( + result + .unwrap_err() + .to_string() + .contains("resumable steps cannot be nested") + ); + assert_parent_metadata(&client, &row, &context, commit).await; + database.cleanup().await; + } +} + +#[cfg(feature = "sqlite")] +#[tokio::test] +async fn rolled_back_parent_cursor_is_not_replayed() { + assert_parent_checkpoint(false).await; +} + +#[cfg(feature = "sqlite")] +async fn assert_parent_checkpoint(commit: bool) { + let (pool, client, row, context) = setup().await; + let result = context + .resumable_step_with_cursor("parent", |_: i64| async { + let mut tx = riverqueue::database::begin_sqlite_write(&pool).await?; + context.resumable_set_step_cursor_tx(&mut tx, &1).await?; + let child_result = context + .resumable_step_with_cursor("child", |_: i64| async { + Err::<(), BoxError>("nested child must not run".into()) + }) + .await; + if commit { + tx.commit().await?; + } else { + tx.rollback().await?; + } + child_result?; + Ok::<(), BoxError>(()) + }) + .await; + assert!( + result + .unwrap_err() + .to_string() + .contains("resumable steps cannot be nested") + ); + assert_parent_metadata(&client, &row, &context, commit).await; +} + +async fn assert_parent_metadata( + client: &Client, + row: &JobRow, + context: &WorkContext, + commit: bool, +) { + __private::work_context_resumable_finish(context, true); + if !commit { + let before_result = client.jobs().get(row.id).await.unwrap(); + assert!( + !before_result + .metadata + .contains_key("river:resumable_cursor") + ); + } + // Rejecting the child preserves committed parent progress without + // replaying a parent checkpoint that rolled back. + let result = client + .jobs() + .update( + row.id, + JobUpdateParams::default().metadata(__private::work_context_metadata_updates(context)), + ) + .await + .unwrap(); + let metadata: Value = serde_json::to_value(&result.metadata).unwrap(); + let actual = Value::Object( + metadata + .as_object() + .unwrap() + .iter() + .filter(|(key, _)| key.starts_with("river:resumable_")) + .map(|(key, value)| (key.clone(), value.clone())) + .collect(), + ); + assert_eq!( + actual, + if commit { + json!({"river:resumable_step": "parent", "river:resumable_cursor": {"parent": 1}}) + } else { + json!({}) + } + ); +} + +#[cfg(feature = "sqlite")] +async fn setup() -> (SqlitePool, Client, JobRow, WorkContext) { + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .unwrap(); + SqliteMigrator::new(pool.clone()) + .migrate_up() + .await + .unwrap(); + let client = Client::builder(pool.clone()).build().unwrap(); + let inserted = client.insert(Args {}).await.unwrap(); + let (row, context) = __private::claim_job_for_test(&client, inserted.job.row.id) + .await + .unwrap(); + (pool, client, row, context) +}