From 2ba8a0dc92311ed0fae2e62495792bb6e3e12e7f Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Sat, 24 Oct 2026 14:58:05 -0500 Subject: [PATCH 1/2] reject nested resumable steps Nested resumable steps can persist unreachable resume points and replay rolled-back parent checkpoints when a child updates its cursor. Reject starting a step while another step callback is active in Go, JavaScript, and Rust, matching Java and Ruby. Preserve nesting errors so an unsuccessful parent cannot be marked complete or lose its cursor. Remove the enclosing-step state that nesting requires, and document the restriction in the public APIs and guides. Replace the draft reproductions with coverage for plain and cursor nesting, committed and rolled-back parent checkpoints, and caught nesting errors. Verify all parent and child step combinations across runtimes. --- CHANGELOG.md | 2 + internal/riverplugin/plugin.go | 1 + java/DIFFERENCES.md | 3 - .../java/com/riverqueue/ProtocolTest.java | 24 +++ js/CHANGELOG.md | 2 + js/docs/deployment.md | 4 +- js/docs/resumable-jobs.md | 4 +- .../src/resumable-nested.integration.test.ts | 100 +++++++++++ js/driver/sqlite/src/runtime.test.ts | 56 ++++++ js/etc/riverqueue.api.md | 7 +- js/src/resumable-nested.test.ts | 87 +++++++++ js/src/resumable.test.ts | 15 -- js/src/resumable.ts | 19 +- resumable.go | 27 ++- resumable_nested_test.go | 164 +++++++++++++++++ ruby/spec/resumable_spec.rb | 6 +- rust/CHANGELOG.md | 2 + rust/riverqueue/src/worker.rs | 167 ++++++++++++++--- rust/riverqueue/tests/resumable_nested.rs | 168 ++++++++++++++++++ 19 files changed, 799 insertions(+), 59 deletions(-) create mode 100644 js/driver/pg/src/resumable-nested.integration.test.ts create mode 100644 js/src/resumable-nested.test.ts create mode 100644 resumable_nested_test.go create mode 100644 rust/riverqueue/tests/resumable_nested.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 19a52fa38..1b60e3069 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Reject nested resumable steps 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). + - 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/internal/riverplugin/plugin.go b/internal/riverplugin/plugin.go index ea50a0db9..81772ac5c 100644 --- a/internal/riverplugin/plugin.go +++ b/internal/riverplugin/plugin.go @@ -89,6 +89,7 @@ type ResumableState struct { Err error ResumeMatched bool ResumeStep string + StepActive bool StepName string } 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/test/java/com/riverqueue/ProtocolTest.java b/java/river/src/test/java/com/riverqueue/ProtocolTest.java index 442a28d03..1a49c9b5e 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(); diff --git a/js/CHANGELOG.md b/js/CHANGELOG.md index 64300a9ad..bbe234a05 100644 --- a/js/CHANGELOG.md +++ b/js/CHANGELOG.md @@ -21,6 +21,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Reject nested resumable steps 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). + - 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..37d94af57 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. 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/sqlite/src/runtime.test.ts b/js/driver/sqlite/src/runtime.test.ts index afa7d3315..c9909a962 100644 --- a/js/driver/sqlite/src/runtime.test.ts +++ b/js/driver/sqlite/src/runtime.test.ts @@ -368,6 +368,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..e254f6233 100644 --- a/js/etc/riverqueue.api.md +++ b/js/etc/riverqueue.api.md @@ -2353,10 +2353,13 @@ 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. + */ stepWithCursor( name: string, callback: (cursor: JsonValue | null) => PromiseLike | void 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..27bbb1289 100644 --- a/js/src/resumable.ts +++ b/js/src/resumable.ts @@ -69,41 +69,44 @@ export 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. */ 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. + */ async stepWithCursor( name: string, callback: (cursor: JsonValue | null) => PromiseLike | void ): Promise { - const previousStepName = this.#stepName; if (!this.#begin(name, true)) return; try { 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; } } @@ -181,6 +184,10 @@ 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 (this.#allNames.has(name)) { this.#failure = new ValidationError( `duplicate resumable step name ${JSON.stringify(name)}` diff --git a/resumable.go b/resumable.go index e9631702c..48579a8ab 100644 --- a/resumable.go +++ b/resumable.go @@ -49,6 +49,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 @@ -73,14 +74,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 } @@ -91,6 +98,7 @@ func ResumableStep(ctx context.Context, name string, opts *StepOpts, stepFunc fu // beginning and supplies the saved cursor when it reaches its named step. // 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 +148,20 @@ 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 = "" + }() 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 +180,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_nested_test.go b/resumable_nested_test.go new file mode 100644 index 000000000..286464077 --- /dev/null +++ b/resumable_nested_test.go @@ -0,0 +1,164 @@ +package river + +import ( + "context" + "encoding/json" + "testing" + + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/require" + + "github.com/riverqueue/river/internal/jobexecutor" + "github.com/riverqueue/river/internal/rivercommon" + "github.com/riverqueue/river/internal/riverplugin" + "github.com/riverqueue/river/riverdbtest" + "github.com/riverqueue/river/riverdriver" + "github.com/riverqueue/river/riverdriver/riverpgxv5" + "github.com/riverqueue/river/rivershared/riversharedtest" + "github.com/riverqueue/river/rivershared/testfactory" + "github.com/riverqueue/river/rivershared/util/testutil" + "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 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)) + } + }) + } +} diff --git a/ruby/spec/resumable_spec.rb b/ruby/spec/resumable_spec.rb index 0cd982cfe..cc6d5af62 100644 --- a/ruby/spec/resumable_spec.rb +++ b/ruby/spec/resumable_spec.rb @@ -201,11 +201,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..d845db7d6 100644 --- a/rust/CHANGELOG.md +++ b/rust/CHANGELOG.md @@ -20,6 +20,8 @@ Changes to River for Go are recorded in the [repository changelog](../CHANGELOG. ### Fixed +- Reject nested resumable steps 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). + - 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..35b0524b2 100644 --- a/rust/riverqueue/src/worker.rs +++ b/rust/riverqueue/src/worker.rs @@ -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(()), }; @@ -244,6 +252,13 @@ impl WorkContext { }; 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()); @@ -413,6 +428,14 @@ 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 !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 +455,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> { @@ -616,36 +639,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 +675,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); } } @@ -1201,6 +1222,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 +1318,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 +1388,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..1a93d3773 --- /dev/null +++ b/rust/riverqueue/tests/resumable_nested.rs @@ -0,0 +1,168 @@ +//! 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 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) +} From 9d94fc6e8b0af6606866e752bb07326ff256d1b6 Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Sat, 24 Oct 2026 20:12:39 -0500 Subject: [PATCH 2/2] validate cursor-only resume targets A first-step cursor can be saved without a completed-step checkpoint. When a retry omits that step, the job can silently complete or persist a later checkpoint that makes the original cursor unreachable. Track cursor-only targets independently of the skip boundary across all five runtimes. Require each target to be reached through a cursor step, and preserve the original checkpoint while targets remain unmatched. Reject transactional checkpoints that would advance past those targets. Keep the existing metadata format and preserve behavior for jobs with a step marker, including legacy stale cursor entries. Add shared compatibility cases and database recovery coverage, including zero and null cursors, renamed steps, plain-step replacements, earlier failures, and checkpoints inside a restored target. Document the cursor-only restriction and the behavior change for nested steps. --- CHANGELOG.md | 3 +- conformance/cmd/generatefixtures/main.go | 1 + conformance/cmd/generatefixtures/resumable.go | 61 +++++++ .../cmd/generatefixtures/resumable_test.go | 96 ++++++++++ internal/riverplugin/plugin.go | 26 ++- java/CHANGELOG.md | 2 + .../main/java/com/riverqueue/WorkContext.java | 25 +++ .../java/com/riverqueue/ProtocolTest.java | 69 ++++++++ .../java/com/riverqueue/ResumableTest.java | 47 +++++ js/CHANGELOG.md | 3 +- js/docs/resumable-jobs.md | 8 + .../pg/src/resumable.integration.test.ts | 58 +++++++ js/driver/sqlite/src/runtime.test.ts | 56 ++++++ js/etc/riverqueue.api.md | 3 +- js/src/conformance.test.ts | 82 +++++++++ js/src/resumable.ts | 38 +++- resumable.go | 9 + resumable_nested_test.go | 164 ------------------ resumable_step_tx.go | 9 + resumable_step_tx_test.go | 116 +++++++++++++ resumable_test.go | 67 +++++++ ruby/CHANGELOG.md | 2 + ruby/lib/resumable.rb | 20 ++- ruby/sig/runtime.rbs | 1 + ruby/spec/client_driver_shared_examples.rb | 22 +++ ruby/spec/resumable_spec.rb | 48 +++++ rust/CHANGELOG.md | 3 +- rust/riverqueue/src/worker.rs | 161 ++++++++++++++++- rust/riverqueue/tests/resumable_nested.rs | 70 ++++++++ 29 files changed, 1094 insertions(+), 176 deletions(-) delete mode 100644 resumable_nested_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 1b60e3069..0755bb469 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,11 +9,12 @@ 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 -- Reject nested resumable steps 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). +- 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). 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 81772ac5c..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,10 +101,12 @@ type ResumableState struct { Cursors map[string]json.RawMessage CursorsDirty bool Err error - ResumeMatched bool - ResumeStep string - StepActive bool - 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/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 1a49c9b5e..6d6f015d4 100644 --- a/java/river/src/test/java/com/riverqueue/ProtocolTest.java +++ b/java/river/src/test/java/com/riverqueue/ProtocolTest.java @@ -359,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 bbe234a05..c1a878de1 100644 --- a/js/CHANGELOG.md +++ b/js/CHANGELOG.md @@ -15,13 +15,14 @@ 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 -- Reject nested resumable steps 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). +- 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/js/docs/resumable-jobs.md b/js/docs/resumable-jobs.md index 37d94af57..bd22289a6 100644 --- a/js/docs/resumable-jobs.md +++ b/js/docs/resumable-jobs.md @@ -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.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 c9909a962..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" }); diff --git a/js/etc/riverqueue.api.md b/js/etc/riverqueue.api.md index e254f6233..92f077049 100644 --- a/js/etc/riverqueue.api.md +++ b/js/etc/riverqueue.api.md @@ -2358,7 +2358,8 @@ export declare class Resumable { step(name: string, callback: () => PromiseLike | void): Promise; /** * Run a named cursor step with the last JSON cursor, or null initially. - * Steps cannot be nested inside other step callbacks. + * 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, 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.ts b/js/src/resumable.ts index 27bbb1289..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,6 +66,12 @@ 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); + } } /** @@ -90,7 +97,8 @@ export class Resumable { /** * Run a named cursor step with the last JSON cursor, or null initially. - * Steps cannot be nested inside other step callbacks. + * 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, @@ -98,6 +106,7 @@ export class Resumable { ): Promise { 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; @@ -136,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)); @@ -170,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: {} }; } @@ -188,6 +214,12 @@ export class Resumable { 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)}` @@ -216,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 48579a8ab..74db57e2c 100644 --- a/resumable.go +++ b/resumable.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "slices" "github.com/riverqueue/river/internal/riverplugin" ) @@ -65,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 { @@ -96,6 +101,8 @@ 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. @@ -155,6 +162,8 @@ func ResumableStepCursor[TCursor any](ctx context.Context, name string, opts *St 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 diff --git a/resumable_nested_test.go b/resumable_nested_test.go deleted file mode 100644 index 286464077..000000000 --- a/resumable_nested_test.go +++ /dev/null @@ -1,164 +0,0 @@ -package river - -import ( - "context" - "encoding/json" - "testing" - - "github.com/jackc/pgx/v5" - "github.com/stretchr/testify/require" - - "github.com/riverqueue/river/internal/jobexecutor" - "github.com/riverqueue/river/internal/rivercommon" - "github.com/riverqueue/river/internal/riverplugin" - "github.com/riverqueue/river/riverdbtest" - "github.com/riverqueue/river/riverdriver" - "github.com/riverqueue/river/riverdriver/riverpgxv5" - "github.com/riverqueue/river/rivershared/riversharedtest" - "github.com/riverqueue/river/rivershared/testfactory" - "github.com/riverqueue/river/rivershared/util/testutil" - "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 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)) - } - }) - } -} 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 cc6d5af62..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 = [] diff --git a/rust/CHANGELOG.md b/rust/CHANGELOG.md index d845db7d6..38f1d573c 100644 --- a/rust/CHANGELOG.md +++ b/rust/CHANGELOG.md @@ -16,11 +16,12 @@ 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 -- Reject nested resumable steps 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). +- 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/rust/riverqueue/src/worker.rs b/rust/riverqueue/src/worker.rs index 35b0524b2..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}, @@ -250,6 +250,7 @@ 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 { @@ -382,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>, @@ -393,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)?); @@ -436,6 +444,14 @@ impl WorkContext { )))); 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( @@ -535,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 { @@ -576,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, @@ -589,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, @@ -622,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 } @@ -1144,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); diff --git a/rust/riverqueue/tests/resumable_nested.rs b/rust/riverqueue/tests/resumable_nested.rs index 1a93d3773..156694515 100644 --- a/rust/riverqueue/tests/resumable_nested.rs +++ b/rust/riverqueue/tests/resumable_nested.rs @@ -16,6 +16,76 @@ use sqlx::{SqlitePool, sqlite::SqlitePoolOptions}; #[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() {