diff --git a/.github/workflows/js.yaml b/.github/workflows/js.yaml
index dab55a710..4c7ff727b 100644
--- a/.github/workflows/js.yaml
+++ b/.github/workflows/js.yaml
@@ -245,6 +245,14 @@ jobs:
         with:
           node-version: ${{ matrix.node-version }}
 
+      # The resumable integration scenarios read Go-generated fixtures.
+      - uses: actions/setup-go@v7
+        with:
+          go-version-file: go.work
+
+      - run: make generate/fixtures
+        working-directory: .
+
       - run: pnpm run build:all
 
       - run: node cli/dist/bin.js migrate-up --database-url "$TEST_DATABASE_URL"
diff --git a/CHANGELOG.md b/CHANGELOG.md
index b347dea0d..d751f91c2 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
 
+- 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.
+
 - Fixed `AddWorkerSafely` leaving a worker's primary kind and earlier aliases registered when a later kind alias conflicts with an existing registration. Failed registrations now leave the worker registry unchanged. [PR #1440](https://github.com/riverqueue/river/pull/1440).
 - Fixed the notifier retrying without backoff and flooding logs when the context passed to `Client.Start` reaches its deadline. [PR #1495](https://github.com/riverqueue/river/pull/1495).
 
diff --git a/Makefile b/Makefile
index 4c9f13ab6..bdfddcdd3 100644
--- a/Makefile
+++ b/Makefile
@@ -219,6 +219,7 @@ test/js/conformance: generate/fixtures
 .PHONY: test/js/integration
 test/js/integration: ## Run JavaScript integration tests against Postgres
 test/js/integration: build/js
+test/js/integration: generate/fixtures
 	pnpm -C js run test:integration
 
 .PHONY: test/rust
@@ -238,7 +239,7 @@ test/rust: generate/fixtures
 .PHONY: test/rust/conformance
 test/rust/conformance: ## Run Rust tests that check Go-generated conformance fixtures
 test/rust/conformance: generate/fixtures
-	cd rust && cargo test -p riverqueue --features chrono-tz --lib --test protocol_fixtures --locked
+	cd rust && cargo test -p riverqueue --features chrono-tz,sqlite --lib --test protocol_fixtures --test resumable_conformance --locked
 
 .PHONY: test/rust/postgres
 test/rust/postgres: ## Run all Rust tests, including Postgres integration tests (requires RIVER_RUST_DATABASE_URL)
diff --git a/conformance/cmd/generatefixtures/main.go b/conformance/cmd/generatefixtures/main.go
index a4125a432..aa568eaed 100644
--- a/conformance/cmd/generatefixtures/main.go
+++ b/conformance/cmd/generatefixtures/main.go
@@ -49,12 +49,18 @@ func run() error {
 		return err
 	}
 
+	resumable, err := makeResumableCases()
+	if err != nil {
+		return err
+	}
+
 	for _, fixture := range []struct {
 		name  string
 		value any
 	}{
 		{"cron_schedules.json", cronSchedules},
 		{"protocol_values.json", protocolValues},
+		{"resumable.json", resumable},
 		{"snooze_counters.json", makeSnoozeCounters()},
 		{"unique_keys.json", uniqueKeys},
 	} {
diff --git a/conformance/cmd/generatefixtures/resumable.go b/conformance/cmd/generatefixtures/resumable.go
new file mode 100644
index 000000000..b6e3fedd1
--- /dev/null
+++ b/conformance/cmd/generatefixtures/resumable.go
@@ -0,0 +1,112 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"maps"
+
+	"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"
+)
+
+// fixtureCheckpointDriver only supplies the transactional update boundary. The
+// staged row is committed or discarded by each scenario, like a caller's tx.
+type fixtureCheckpointDriver struct {
+	riverdriver.Driver[riverdriver.ExecutorTx]
+}
+
+func (*fixtureCheckpointDriver) UnwrapExecutor(tx riverdriver.ExecutorTx) riverdriver.ExecutorTx {
+	return tx
+}
+
+type fixtureCheckpointExecutor struct {
+	riverdriver.ExecutorTx
+
+	fail bool
+	row  *rivertype.JobRow
+}
+
+func (e *fixtureCheckpointExecutor) JobUpdate(_ context.Context, params *riverdriver.JobUpdateParams) (*rivertype.JobRow, error) {
+	if e.fail {
+		return nil, errors.New("checkpoint write failed")
+	}
+	e.row.Metadata = params.Metadata
+	return e.row, nil
+}
+
+type fixtureResumableArgs struct{}
+
+func (fixtureResumableArgs) Kind() string { return "resumable_fixture" }
+
+type resumableCase struct {
+	ExpectedMetadata map[string]any `json:"expected_metadata"`
+	Name             string         `json:"name"`
+}
+
+type resumableCases struct {
+	Cases   []resumableCase `json:"cases"`
+	Comment string          `json:"$comment"`
+}
+
+func makeResumableCases() (resumableCases, error) {
+	fixture := resumableCases{Comment: generatedComment("River resumable steps and transactional checkpoints")}
+	for _, name := range []string{
+		"checkpoint_commit_error", "checkpoint_rollback_error", "checkpoint_then_cursor_error", "checkpoint_write_failure",
+		"first_cursor_error", "step_snooze",
+	} {
+		updates := make(map[string]any)
+		ctx := context.WithValue(context.Background(), jobexecutor.ContextKeyMetadataUpdates, updates)
+		client, err := river.NewClient(&fixtureCheckpointDriver{}, &river.Config{})
+		if err != nil {
+			return fixture, err
+		}
+		ctx = context.WithValue(ctx, rivercommon.ContextKeyClient{}, client)
+		row := &rivertype.JobRow{ID: 1, EncodedArgs: []byte(`{}`), Metadata: []byte(`{}`), State: rivertype.JobStateRunning}
+		staged := *row
+		exec := &fixtureCheckpointExecutor{fail: name == "checkpoint_write_failure", row: &staged}
+		attemptErr := (&riverplugin.ResumableMiddleware{}).Work(ctx, row, func(ctx context.Context) error {
+			if name != "first_cursor_error" {
+				river.ResumableStep(ctx, "first", nil, func(context.Context) error { return nil })
+			}
+			if name == "step_snooze" {
+				return river.JobSnooze(60_000_000_000)
+			}
+			river.ResumableStepCursor(ctx, "loop", nil, func(ctx context.Context, _ int) error {
+				if name == "first_cursor_error" {
+					if err := river.ResumableSetCursor(ctx, 7); err != nil {
+						return err
+					}
+				} else {
+					if _, err := river.ResumableSetStepCursorTx[*fixtureCheckpointDriver](ctx, riverdriver.ExecutorTx(exec), &river.Job[fixtureResumableArgs]{JobRow: row}, 1); err != nil {
+						return err
+					}
+				}
+				if name == "checkpoint_then_cursor_error" {
+					if err := river.ResumableSetCursor(ctx, 2); err != nil {
+						return err
+					}
+				}
+				return errors.New("retry")
+			})
+			return nil
+		})
+		if attemptErr == nil {
+			return fixture, fmt.Errorf("resumable scenario %s unexpectedly succeeded", name)
+		}
+		metadata := make(map[string]any)
+		if name == "checkpoint_commit_error" || name == "checkpoint_then_cursor_error" {
+			if err := json.Unmarshal(staged.Metadata, &metadata); err != nil {
+				return fixture, err
+			}
+		}
+		maps.Copy(metadata, updates)
+		fixture.Cases = append(fixture.Cases, resumableCase{ExpectedMetadata: metadata, Name: name})
+	}
+	return fixture, nil
+}
diff --git a/conformance/cmd/generatefixtures/resumable_test.go b/conformance/cmd/generatefixtures/resumable_test.go
new file mode 100644
index 000000000..263c2f212
--- /dev/null
+++ b/conformance/cmd/generatefixtures/resumable_test.go
@@ -0,0 +1,33 @@
+package main
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestMakeResumableCases(t *testing.T) {
+	t.Parallel()
+
+	fixture, err := makeResumableCases()
+	require.NoError(t, err)
+	want := map[string]string{
+		"checkpoint_commit_error":      `{"river:resumable_step":"loop","river:resumable_cursor":{"loop":1}}`,
+		"checkpoint_then_cursor_error": `{"river:resumable_step":"loop","river:resumable_cursor":{"loop":2}}`,
+		"checkpoint_rollback_error":    `{}`,
+		"checkpoint_write_failure":     `{"river:resumable_step":"first"}`,
+		"first_cursor_error":           `{"river:resumable_cursor":{"loop":7}}`,
+		"step_snooze":                  `{"river:resumable_step":"first"}`,
+	}
+	require.Len(t, fixture.Cases, len(want))
+	for _, testCase := range fixture.Cases {
+		t.Run(testCase.Name, func(t *testing.T) {
+			t.Parallel()
+
+			metadata, err := json.Marshal(testCase.ExpectedMetadata)
+			require.NoError(t, err)
+			require.JSONEq(t, want[testCase.Name], string(metadata))
+		})
+	}
+}
diff --git a/conformance/go.mod b/conformance/go.mod
index be1ef7519..f8526a3c8 100644
--- a/conformance/go.mod
+++ b/conformance/go.mod
@@ -13,9 +13,12 @@ require (
 	github.com/riverqueue/river/rivershared v0.49.0
 	github.com/riverqueue/river/rivertype v0.49.0
 	github.com/robfig/cron/v3 v3.0.1
+	github.com/stretchr/testify v1.12.1
 	golang.org/x/mod v0.42.0
 )
 
+require go.yaml.in/yaml/v3 v3.0.5 // indirect
+
 require (
 	github.com/jackc/puddle/v2 v2.2.3 // indirect
 	github.com/tidwall/gjson v1.20.0 // indirect
diff --git a/conformance/go.sum b/conformance/go.sum
index 9ffcfbcb4..9664bdf18 100644
--- a/conformance/go.sum
+++ b/conformance/go.sum
@@ -1,3 +1,5 @@
+github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438 h1:Dj0L5fhJ9F82ZJyVOmBx6msDp/kfd1t9GRfny/mfJA0=
+github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
 github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
diff --git a/internal/riverplugin/plugin.go b/internal/riverplugin/plugin.go
index 9cce6d391..ea50a0db9 100644
--- a/internal/riverplugin/plugin.go
+++ b/internal/riverplugin/plugin.go
@@ -44,12 +44,10 @@ func (*ResumableMiddleware) Work(ctx context.Context, job *rivertype.JobRow, doI
 		state.ResumeMatched = false
 	}
 
-	hadCursors := false
 	if cursorJSON := gjson.GetBytes(job.Metadata, rivercommon.MetadataKeyResumableCursor); cursorJSON.Exists() && cursorJSON.Type == gjson.JSON {
 		if err := json.Unmarshal([]byte(cursorJSON.Raw), &state.Cursors); err != nil {
 			return fmt.Errorf("river: unmarshal resumable cursors: %w", err)
 		}
-		hadCursors = len(state.Cursors) > 0
 	}
 
 	ctx = context.WithValue(ctx, ResumableContextKey{}, state)
@@ -64,16 +62,18 @@ func (*ResumableMiddleware) Work(ctx context.Context, job *rivertype.JobRow, doI
 		}
 	}
 
-	if err != nil && state.CompletedStep != "" {
-		if len(state.Cursors) > 0 {
+	if err != nil {
+		if state.CursorsDirty && len(state.Cursors) > 0 {
 			metadataUpdates[rivercommon.MetadataKeyResumableCursor] = state.Cursors
-		} else if hadCursors {
+		} else if state.CursorsDirty {
 			// All cursors were consumed (their steps completed). Write null
 			// to clear the stale cursor metadata, since the metadata merge
 			// is additive and wouldn't remove the old key on its own.
 			metadataUpdates[rivercommon.MetadataKeyResumableCursor] = nil
 		}
-		metadataUpdates[rivercommon.MetadataKeyResumableStep] = state.CompletedStep
+		if state.CompletedStep != "" {
+			metadataUpdates[rivercommon.MetadataKeyResumableStep] = state.CompletedStep
+		}
 	}
 
 	return err
@@ -85,6 +85,7 @@ type ResumableState struct {
 	AllStepNames  map[string]struct{}
 	CompletedStep string
 	Cursors       map[string]json.RawMessage
+	CursorsDirty  bool
 	Err           error
 	ResumeMatched bool
 	ResumeStep    string
diff --git a/java/river/src/main/java/com/riverqueue/WorkContext.java b/java/river/src/main/java/com/riverqueue/WorkContext.java
index 3a7be3cac..fcdd51681 100644
--- a/java/river/src/main/java/com/riverqueue/WorkContext.java
+++ b/java/river/src/main/java/com/riverqueue/WorkContext.java
@@ -149,8 +149,8 @@ synchronized Exception finish(Exception failure) {
     if (stepFailure != null) failure = stepFailure;
     if (failure == null && !resumeMatched)
       failure = new IllegalArgumentException("Saved resumable step was not found");
-    if (failure != null && completedStep != null) {
-      metadata(Protocol.METADATA_RESUMABLE_STEP, completedStep);
+    if (failure != null) {
+      if (completedStep != null) metadata(Protocol.METADATA_RESUMABLE_STEP, completedStep);
       if (!cursors.isEmpty() || job.metadata().has(Protocol.METADATA_RESUMABLE_CURSOR))
         metadata(Protocol.METADATA_RESUMABLE_CURSOR, cursors.isEmpty() ? null : cursors);
     }
diff --git a/java/river/src/test/java/com/riverqueue/ProtocolTest.java b/java/river/src/test/java/com/riverqueue/ProtocolTest.java
index eb0424058..75ccd15cd 100644
--- a/java/river/src/test/java/com/riverqueue/ProtocolTest.java
+++ b/java/river/src/test/java/com/riverqueue/ProtocolTest.java
@@ -300,6 +300,65 @@ public double nextDouble() {
     return tests.stream();
   }
 
+  @TestFactory
+  Stream<DynamicTest> resumableConformance() throws Exception {
+    var tests = new ArrayList<DynamicTest>();
+    for (var fixture : Conformance.fixture("resumable.json").required("cases")) {
+      String name = fixture.required("name").asString();
+      // Java intentionally has no transactional checkpoint helper.
+      if (name.startsWith("checkpoint_")) continue;
+      tests.add(
+          DynamicTest.dynamicTest(
+              name,
+              () -> {
+                var context = context(42);
+                Exception failure =
+                    assertThrows(
+                        Exception.class,
+                        () -> {
+                          if (name.equals("step_snooze")) {
+                            context.step("first", () -> {});
+                            context.snooze(java.time.Duration.ofSeconds(60));
+                          } else {
+                            context.stepWithCursor(
+                                "loop",
+                                cursor -> {
+                                  context.cursor(7);
+                                  throw new Exception("retry");
+                                });
+                          }
+                        });
+                context.finish(failure);
+                assertEquals(fixture.required("expected_metadata"), context.metadataUpdates());
+              }));
+    }
+    return tests.stream();
+  }
+
+  @Test
+  void resumableFirstStepCursor() throws Exception {
+    var context = context(42, Json.object());
+    var failure = new Exception("retry");
+    assertThrows(
+        Exception.class,
+        () ->
+            context.stepWithCursor(
+                "loop",
+                cursor -> {
+                  context.cursor(Map.of("n", 7));
+                  throw failure;
+                }));
+    assertSame(failure, context.finish(failure));
+    assertEquals(
+        Json.tree(Map.of(Protocol.METADATA_RESUMABLE_CURSOR, Map.of("loop", Map.of("n", 7)))),
+        context.metadataUpdates());
+    var retry = context(42, context.metadataUpdates());
+    retry.stepWithCursor("loop", cursor -> assertEquals(Json.tree(Map.of("n", 7)), cursor));
+    retry.finish(new Exception("later failure"));
+    assertEquals("loop", retry.metadataUpdates().path(Protocol.METADATA_RESUMABLE_STEP).asString());
+    assertTrue(retry.metadataUpdates().path(Protocol.METADATA_RESUMABLE_CURSOR).isNull());
+  }
+
   @Test
   void resumableMetadata() throws Exception {
     var keys = Conformance.fixture("protocol_values.json").required("metadata_keys");
diff --git a/java/river/src/test/java/com/riverqueue/WorkersTest.java b/java/river/src/test/java/com/riverqueue/WorkersTest.java
index e4fe2d80b..a8015abc5 100644
--- a/java/river/src/test/java/com/riverqueue/WorkersTest.java
+++ b/java/river/src/test/java/com/riverqueue/WorkersTest.java
@@ -403,6 +403,44 @@ void oversizedDurationsFailDuringConfiguration(String option) {
         });
   }
 
+  @ParameterizedTest
+  @ValueSource(strings = {"error", "shutdown", "snooze"})
+  void firstStepCursorSurvivesIncompleteAttempt(String outcome) throws Exception {
+    var client = client();
+    var entered = new CountDownLatch(1);
+    var finished = new CompletableFuture<Workers.Event>();
+    try (var workers =
+            workers(client)
+                .add(
+                    TYPE,
+                    context -> {
+                      context.stepWithCursor(
+                          "loop",
+                          cursor -> {
+                            context.cursor(7);
+                            entered.countDown();
+                            if (outcome.equals("snooze")) context.snooze(Duration.ofSeconds(60));
+                            if (outcome.equals("shutdown")) {
+                              context.awaitCancellation();
+                              context.checkCancelled();
+                            }
+                            throw new Exception("retry");
+                          });
+                    })
+                .start();
+        var subscription = workers.subscribe(finished::complete)) {
+      var job = client.insert(TYPE, "cursor", InsertOptions.builder().maxAttempts(1).build()).job();
+      assertTrue(entered.await(5, TimeUnit.SECONDS));
+      if (outcome.equals("shutdown")) workers.stopAndCancel();
+      var event = finished.get(5, TimeUnit.SECONDS);
+      assertEquals(job.id(), event.job().id());
+      assertFalse(event.job().metadata().has(Protocol.METADATA_RESUMABLE_STEP));
+      assertEquals(
+          Json.tree(java.util.Map.of("loop", 7)),
+          event.job().metadata().path(Protocol.METADATA_RESUMABLE_CURSOR));
+    }
+  }
+
   @Test
   void oversizedSnoozeFailsTheAttemptInsteadOfStrandingIt() throws Exception {
     var client = client();
diff --git a/js/CHANGELOG.md b/js/CHANGELOG.md
index 9d8e2d797..e7ae79917 100644
--- a/js/CHANGELOG.md
+++ b/js/CHANGELOG.md
@@ -19,6 +19,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- 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.
+
 - Worker threads respect cancellation received while loading a handler module, before invoking its handler.
 - Like River for Go, a leader whose renewal finds its term gone, because it expired or another process replaced it, gives up leadership at once instead of running maintenance until its local deadline passes. [PR #1504](https://github.com/riverqueue/river/pull/1504).
 - `updateQueue` raising a full queue's `maxWorkers` starts jobs at once instead of waiting for a running job to finish. [PR #1504](https://github.com/riverqueue/river/pull/1504).
diff --git a/js/docs/resumable-jobs.md b/js/docs/resumable-jobs.md
index 9948cd0e8..cccce72d1 100644
--- a/js/docs/resumable-jobs.md
+++ b/js/docs/resumable-jobs.md
@@ -2,7 +2,7 @@
 
 A job that does several expensive steps can skip the steps an earlier attempt
 already finished. Wrap each step in `ctx.resumable.step`; when an attempt
-fails, River records the last completed step in the job's metadata, and the
+fails or snoozes, River records the last completed step in the job's metadata, and the
 next attempt skips every step up to and including it:
 
 <!-- ts-setup
@@ -66,7 +66,13 @@ const workers = new Workers().add(exportRows, async ({ resumable }) => {
 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
-transaction.
+transaction. Once that write succeeds, the transaction owns the checkpoint:
+River does not replay it in the attempt's deferred metadata. If the transaction
+rolls back, earlier progress that was not already persisted may run again.
+A failed checkpoint write leaves the previous deferred progress intact.
+
+A cursor can be saved even when no step has completed. In that case a retry
+starts at the beginning and supplies the saved cursor at its named step.
 
 ## Across languages
 
diff --git a/js/driver/pg/src/resumable.integration.test.ts b/js/driver/pg/src/resumable.integration.test.ts
new file mode 100644
index 000000000..7909b2611
--- /dev/null
+++ b/js/driver/pg/src/resumable.integration.test.ts
@@ -0,0 +1,115 @@
+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,
+  snooze,
+  Workers,
+  type JsonObject,
+} from "riverqueue";
+
+import { PgDriver } from "./driver.js";
+
+const fixtures = JSON.parse(
+  await readFile(
+    new URL("../../../../conformance/testdata/resumable.json", import.meta.url),
+    "utf8"
+  )
+) as {
+  cases: { name: string; expected_metadata: JsonObject }[];
+};
+
+describe("Go resumable conformance", () => {
+  const schema = `js_resumable_${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(fixtures.cases)("$name", async (fixture) => {
+    const definition = defineJob({ kind: fixture.name });
+    const queue = fixture.name;
+    const client = new Client(new PgDriver(pool, { schema }), {
+      queues: { [queue]: { maxWorkers: 1 } },
+      workers: new Workers().add(definition, async ({ resumable }) => {
+        if (!fixture.name.startsWith("first_cursor")) {
+          await resumable.step("first", () => undefined);
+        }
+        if (fixture.name === "step_snooze") return snooze({ seconds: 60 });
+        await resumable.stepWithCursor("loop", async () => {
+          if (fixture.name.startsWith("first_cursor")) {
+            resumable.setCursor(7);
+          } else {
+            const tx = await pool.connect();
+            try {
+              await tx.query("BEGIN");
+              if (fixture.name === "checkpoint_write_failure") {
+                await tx.query("SELECT 1 / 0").catch(() => undefined);
+              }
+              await resumable.checkpoint({ cursor: 1, tx });
+              if (
+                fixture.name === "checkpoint_commit_error" ||
+                fixture.name === "checkpoint_then_cursor_error"
+              )
+                await tx.query("COMMIT");
+            } finally {
+              await tx.query("ROLLBACK");
+              tx.release();
+            }
+          }
+          if (fixture.name === "checkpoint_then_cursor_error")
+            resumable.setCursor(2);
+          throw new Error("retry");
+        });
+        return snooze({ seconds: 60 });
+      }),
+    });
+    const inserted = await client.insert(
+      definition,
+      {},
+      { maxAttempts: 1, queue }
+    );
+    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(fixture.name.endsWith("snooze") ? "scheduled" : "discarded");
+      const metadata = Object.fromEntries(
+        Object.entries(row?.metadata ?? {}).filter(([key]) =>
+          key.startsWith("river:resumable_")
+        )
+      );
+      expect(metadata).toEqual(fixture.expected_metadata);
+    } finally {
+      await runtime.stop();
+    }
+  });
+});
diff --git a/js/driver/sqlite/src/coordination.test.ts b/js/driver/sqlite/src/coordination.test.ts
index ddecd8091..bf0e0f803 100644
--- a/js/driver/sqlite/src/coordination.test.ts
+++ b/js/driver/sqlite/src/coordination.test.ts
@@ -328,18 +328,17 @@ describe("SqliteDriver connection coordination", () => {
     });
   });
 
-  test("keeps River's background work waiting asynchronously during a long application transaction", async () => {
-    const { database, driver } = await fileSetup();
+  test("keeps River's background work waiting asynchronously during an application transaction", async () => {
+    const busy = busyWaits();
+    const { database, driver } = await fileSetup({
+      [SQLITE_DRIVER_TEST_HOOKS]: { sleep: busy.sleep },
+    } as SqliteDriverOptions);
     await driver.jobInsert({ args: {}, kind: "background_waits" });
-    let maxLagMs = 0;
-    let last = performance.now();
-    const interval = setInterval(() => {
-      const now = performance.now();
-      maxLagMs = Math.max(maxLagMs, now - last);
-      last = now;
-    }, 1);
-    onTestFinished(() => clearInterval(interval));
 
+    let releaseTransaction!: () => void;
+    const released = new Promise<void>((resolve) => {
+      releaseTransaction = resolve;
+    });
     let started!: () => void;
     const open = new Promise<void>((resolve) => {
       started = resolve;
@@ -347,23 +346,29 @@ describe("SqliteDriver connection coordination", () => {
     const pending = transaction(database, async (tx) => {
       await driver.jobInsert({ args: {}, kind: "application_tx" }, { tx });
       started();
-      await new Promise((resolve) => setTimeout(resolve, 300));
+      await released;
     });
     await open;
-    const claim = driver.jobClaim(claimParams());
-    await pending;
+    let claimed = false;
+    const claim = driver.jobClaim(claimParams()).then((result) => {
+      claimed = true;
+      return result;
+    });
+    try {
+      // Observe an actual asynchronous busy retry while the application still
+      // holds the transaction, without depending on elapsed time under CI load.
+      await expect.poll(() => busy.count).toBeGreaterThan(0);
+      expect(claimed).toBe(false);
+    } finally {
+      releaseTransaction();
+      await pending;
+    }
     expect((await claim).jobs).toHaveLength(2);
-
-    // River's connection never lets SQLite wait for a lock synchronously; it
-    // retries between turns of the event loop. Blocking instead would have
-    // frozen timers for the whole 300 ms transaction (or deadlocked it), so
-    // a bound well below that is generous even on a loaded machine.
     expect(
       await driver.execute("busy_timeout", {}, (db) =>
         db.prepare("PRAGMA busy_timeout").get()
       )
     ).toEqual({ timeout: 0 });
-    expect(maxLagMs).toBeLessThan(250);
   });
 
   test("stops a River call waiting behind the transaction it runs inside", async () => {
diff --git a/js/etc/riverqueue.api.md b/js/etc/riverqueue.api.md
index 8418e3a83..6c65531c2 100644
--- a/js/etc/riverqueue.api.md
+++ b/js/etc/riverqueue.api.md
@@ -2375,7 +2375,7 @@ export declare class Resumable {
 ```ts
 /**
  * Save resumable progress now, atomically with `tx`, instead of when the
- * attempt ends.
+ * attempt ends. If it rolls back, earlier unpersisted progress may be replayed.
  */
 export interface ResumableCheckpointOptions<Transaction = unknown> {
   readonly cursor?: JsonValue;
diff --git a/js/src/conformance.test.ts b/js/src/conformance.test.ts
index 51c226f61..7a8186b57 100644
--- a/js/src/conformance.test.ts
+++ b/js/src/conformance.test.ts
@@ -96,6 +96,47 @@ const protocol = (await readFixture("protocol_values")) as unknown as {
   }[];
 };
 
+const resumableCases = (await readFixture("resumable")) as unknown as {
+  readonly cases: readonly { name: string; expected_metadata: JsonObject }[];
+};
+
+describe("Go resumable fixtures", () => {
+  it.each(resumableCases.cases)("$name", async (fixture) => {
+    let persisted: JsonObject = {};
+    const client = {
+      jobs: {
+        update: (_id: bigint, update: { metadata: JsonObject }) => {
+          if (fixture.name === "checkpoint_write_failure")
+            throw new Error("write failed");
+          if (
+            fixture.name === "checkpoint_commit_error" ||
+            fixture.name === "checkpoint_then_cursor_error"
+          )
+            persisted = update.metadata;
+          return Promise.resolve(job());
+        },
+      },
+    } as unknown as Client;
+    const resumable = new Resumable(client, job());
+    if (fixture.name !== "first_cursor_error")
+      await resumable.step("first", () => undefined);
+    if (fixture.name !== "step_snooze") {
+      await resumable
+        .stepWithCursor("loop", async () => {
+          if (fixture.name === "first_cursor_error") resumable.setCursor(7);
+          else await resumable.checkpoint({ cursor: 1, tx: {} });
+          if (fixture.name === "checkpoint_then_cursor_error")
+            resumable.setCursor(2);
+          throw new Error("retry");
+        })
+        .catch(() => undefined);
+    }
+    expect({ ...persisted, ...resumable.finish(true).metadata }).toEqual(
+      fixture.expected_metadata
+    );
+  });
+});
+
 describe("Go unique-key fixtures", () => {
   it.each(uniqueKeys.cases)("$name", (fixture) => {
     const now = Temporal.Instant.from(fixture.now);
diff --git a/js/src/resumable.test.ts b/js/src/resumable.test.ts
index efbc21c83..432dee75d 100644
--- a/js/src/resumable.test.ts
+++ b/js/src/resumable.test.ts
@@ -22,6 +22,53 @@ describe("Resumable", () => {
     }
   );
 
+  it("persists and resumes a first-step cursor without a completed step", async () => {
+    const resumable = new Resumable(client(), job({}));
+    await expect(
+      resumable.stepWithCursor("loop", () => {
+        resumable.setCursor({ n: 7 });
+        throw new Error("retry");
+      })
+    ).rejects.toThrow("retry");
+    const metadata = resumable.finish(true).metadata;
+    expect(metadata).toEqual({ "river:resumable_cursor": { loop: { n: 7 } } });
+    const retry = new Resumable(client(), job(metadata));
+    await retry.stepWithCursor("loop", (cursor) => {
+      expect(cursor).toEqual({ n: 7 });
+    });
+    expect(retry.finish(true).metadata).toEqual({
+      "river:resumable_step": "loop",
+      "river:resumable_cursor": null,
+    });
+  });
+
+  it("leaves deferred progress unchanged when a checkpoint write fails", async () => {
+    const update = vi.fn().mockRejectedValue(new Error("write failed"));
+    const resumable = new Resumable(client(update), job({}));
+    await resumable.step("first", () => undefined);
+    await expect(
+      resumable.stepWithCursor("loop", async () => {
+        await resumable.checkpoint({ cursor: { n: 1 }, tx: {} });
+      })
+    ).rejects.toThrow("write failed");
+    expect(resumable.finish(true).metadata).toEqual({
+      "river:resumable_step": "first",
+    });
+  });
+
+  it("does not replay a rolled-back checkpoint on failure", async () => {
+    const update = vi.fn().mockResolvedValue(job({}));
+    const resumable = new Resumable(client(update), job({}));
+    await resumable.step("first", () => undefined);
+    await expect(
+      resumable.stepWithCursor("loop", async () => {
+        await resumable.checkpoint({ cursor: 1, tx: {} });
+        throw new Error("rollback");
+      })
+    ).rejects.toThrow("rollback");
+    expect(resumable.finish(true).metadata).toEqual({});
+  });
+
   it("treats an empty persisted step as no checkpoint", async () => {
     const resumable = new Resumable(
       client(),
@@ -131,10 +178,7 @@ describe("Resumable", () => {
       },
       { tx }
     );
-    expect(resumable.finish(true).metadata).toEqual({
-      "river:resumable_cursor": { process: { offset: 3 } },
-      "river:resumable_step": "process",
-    });
+    expect(resumable.finish(true).metadata).toEqual({});
   });
 });
 
diff --git a/js/src/resumable.ts b/js/src/resumable.ts
index df0bebae1..d20182a2b 100644
--- a/js/src/resumable.ts
+++ b/js/src/resumable.ts
@@ -9,7 +9,7 @@ const STEP_KEY = "river:resumable_step";
 
 /**
  * Save resumable progress now, atomically with `tx`, instead of when the
- * attempt ends.
+ * attempt ends. If it rolls back, earlier unpersisted progress may be replayed.
  */
 export interface ResumableCheckpointOptions<Transaction = unknown> {
   readonly cursor?: JsonValue;
@@ -26,10 +26,10 @@ export class Resumable {
   readonly #allNames = new Set<string>();
   readonly #client: Client;
   readonly #cursors = new Map<string, JsonValue>();
-  readonly #hadCursors: boolean;
   readonly #job: JobRow;
   readonly #resumeStep: string | null;
   #completedStep: string | null = null;
+  #cursorsDirty = false;
   #failure: Error | null = null;
   #resumeMatched: boolean;
   #stepName: string | null = null;
@@ -61,7 +61,6 @@ export class Resumable {
         if (cursor !== undefined) this.#cursors.set(name, cursor);
       }
     }
-    this.#hadCursors = this.#cursors.size > 0;
   }
 
   /**
@@ -95,7 +94,7 @@ export class Resumable {
     try {
       await callback(this.#cursors.get(name) ?? null);
       this.#completedStep = name;
-      this.#cursors.delete(name);
+      if (this.#cursors.delete(name)) this.#cursorsDirty = true;
     } catch (cause: unknown) {
       this.#failure = stepError(name, cause);
       throw cause;
@@ -112,6 +111,7 @@ export class Resumable {
       );
     }
     this.#cursors.set(this.#stepName, toJsonValue(cursor));
+    this.#cursorsDirty = true;
   }
 
   /** Persist the current step and optional cursor in a caller transaction. */
@@ -123,18 +123,30 @@ export class Resumable {
         "resumable checkpoint can only be set inside a resumable step"
       );
     }
-    this.#completedStep = this.#stepName;
+    const cursors = new Map(this.#cursors);
     if (options.cursor !== undefined) {
-      this.#cursors.set(this.#stepName, toJsonValue(options.cursor));
+      cursors.set(this.#stepName, toJsonValue(options.cursor));
     }
     const updated = await this.#client.jobs.update(
       this.#job.id,
-      { metadata: this.#checkpointMetadata() },
+      {
+        metadata: {
+          [STEP_KEY]: this.#stepName,
+          [CURSOR_KEY]:
+            cursors.size > 0 ? toJsonObject(Object.fromEntries(cursors)) : null,
+        },
+      },
       { tx: options.tx }
     );
     if (updated === null) {
       throw new LifecycleError(`running job ${this.#job.id} no longer exists`);
     }
+    // The caller transaction owns this checkpoint, including its rollback.
+    // Only subsequent progress belongs in deferred attempt metadata.
+    this.#cursors.clear();
+    for (const [name, cursor] of cursors) this.#cursors.set(name, cursor);
+    this.#cursorsDirty = false;
+    this.#completedStep = null;
     return updated;
   }
 
@@ -150,7 +162,7 @@ export class Resumable {
     }
     return {
       error: this.#failure,
-      metadata: this.#completedStep === null ? {} : this.#checkpointMetadata(),
+      metadata: this.#checkpointMetadata(),
     };
   }
 
@@ -177,15 +189,13 @@ export class Resumable {
   }
 
   #checkpointMetadata(): JsonObject {
-    const metadata: JsonObject = {
-      [STEP_KEY]: this.#completedStep ?? this.#stepName ?? "",
-    };
-    if (this.#cursors.size > 0) {
-      metadata[CURSOR_KEY] = toJsonObject(
-        Object.fromEntries(this.#cursors.entries())
-      );
-    } else if (this.#hadCursors) {
-      metadata[CURSOR_KEY] = null;
+    const metadata: JsonObject = {};
+    if (this.#completedStep !== null) metadata[STEP_KEY] = this.#completedStep;
+    if (this.#cursorsDirty) {
+      metadata[CURSOR_KEY] =
+        this.#cursors.size > 0
+          ? toJsonObject(Object.fromEntries(this.#cursors))
+          : null;
     }
     return metadata;
   }
diff --git a/js/src/runtime/attempt-runner.ts b/js/src/runtime/attempt-runner.ts
index cca2ea5a5..7f2a36c6e 100644
--- a/js/src/runtime/attempt-runner.ts
+++ b/js/src/runtime/attempt-runner.ts
@@ -691,7 +691,8 @@ export class AttemptRunner {
     }
     if (decodedContext !== undefined) {
       const resumable = decodedContext.resumable.finish(
-        result.status !== "succeeded"
+        result.status !== "succeeded" ||
+          (result.outcome !== undefined && result.outcome.type !== "complete")
       );
       if (resumable.error !== null && result.status !== "failed") {
         result = { error: resumable.error, status: "failed" };
diff --git a/js/test/src/index.test.ts b/js/test/src/index.test.ts
index 9cad3273d..746e5f2a8 100644
--- a/js/test/src/index.test.ts
+++ b/js/test/src/index.test.ts
@@ -71,6 +71,15 @@ describe("testJob", () => {
 });
 
 describe("workOnce", () => {
+  test("preserves completed steps when the worker snoozes", async () => {
+    const job = await testJob(emailJob, { message: "snooze" });
+    const result = await workOnce(job, async ({ resumable }) => {
+      await resumable.step("first", () => undefined);
+      return snooze({ seconds: 60 });
+    });
+    expect(result.metadata).toEqual({ "river:resumable_step": "first" });
+  });
+
   test("persists metadata and suppressed resumable errors like the runtime", async () => {
     const job = await testJob(emailJob, { message: "resume" });
     let firstRuns = 0;
diff --git a/js/test/src/index.ts b/js/test/src/index.ts
index 8d6f9c941..bdc947ec1 100644
--- a/js/test/src/index.ts
+++ b/js/test/src/index.ts
@@ -487,7 +487,10 @@ export async function workOnce<
     if (outcome?.type === "complete" && outcome.output !== undefined) {
       context.recordOutput(outcome.output);
     }
-    const finished = finishResumable(context.resumable, false);
+    const finished = finishResumable(
+      context.resumable,
+      outcome !== undefined && outcome.type !== "complete"
+    );
     Object.assign(metadata, finished.metadata);
     if (finished.error !== null) throw finished.error;
     return {
diff --git a/resumable.go b/resumable.go
index 272b3d9c1..e9631702c 100644
--- a/resumable.go
+++ b/resumable.go
@@ -15,7 +15,8 @@ var (
 )
 
 // ResumableSetCursor records a cursor for the current resumable cursor step.
-// The cursor is stored only if the job attempt ends in an error, allowing a
+// The cursor is stored if the job attempt ends in an error, snooze, or
+// cooperative cancellation, allowing a
 // later retry to resume the same step from the recorded position.
 //
 // Alternatively, ResumableSetStepCursorTx is available to persist a step and
@@ -36,6 +37,7 @@ func ResumableSetCursor[TCursor any](ctx context.Context, cursor TCursor) error
 		state.Cursors = make(map[string]json.RawMessage)
 	}
 	state.Cursors[state.StepName] = json.RawMessage(cursorBytes)
+	state.CursorsDirty = true
 	return nil
 }
 
@@ -85,7 +87,8 @@ func ResumableStep(ctx context.Context, name string, opts *StepOpts, stepFunc fu
 
 // ResumableStepCursor runs a resumable step that also receives a persisted
 // cursor value from an earlier failed attempt, if one was recorded with
-// ResumableSetCursor.
+// ResumableSetCursor. If no step has completed yet, the retry starts at the
+// 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.
 //
@@ -147,7 +150,10 @@ func ResumableStepCursor[TCursor any](ctx context.Context, name string, opts *St
 	}
 
 	state.CompletedStep = name
-	delete(state.Cursors, name)
+	if _, ok := state.Cursors[name]; ok {
+		delete(state.Cursors, name)
+		state.CursorsDirty = true
+	}
 }
 
 func mustResumableState(ctx context.Context) *riverplugin.ResumableState {
diff --git a/resumable_step_tx.go b/resumable_step_tx.go
index fddb8c63a..40381d4d8 100644
--- a/resumable_step_tx.go
+++ b/resumable_step_tx.go
@@ -4,9 +4,9 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"maps"
 
 	"github.com/riverqueue/river/internal/execution"
-	"github.com/riverqueue/river/internal/jobexecutor"
 	"github.com/riverqueue/river/internal/rivercommon"
 	"github.com/riverqueue/river/riverdriver"
 	"github.com/riverqueue/river/rivertype"
@@ -14,7 +14,7 @@ import (
 
 // ResumableSetStepTx immediately persists the current resumable step as
 // part of transaction tx. If tx is rolled back, the step update will be as
-// well.
+// well. Earlier progress not already persisted may be replayed after rollback.
 //
 // Normally, a resumable job's step progress is recorded after it runs along
 // with its result status. This is normally sufficient, but because it happens
@@ -31,7 +31,8 @@ func ResumableSetStepTx[TDriver riverdriver.Driver[TTx], TTx any, TArgs JobArgs]
 
 // ResumableSetStepCursorTx immediately persists the current resumable step and
 // cursor as part of transaction tx. If tx is rolled back, the step and cursor
-// update will be as well.
+// update will be as well. Earlier progress not already persisted may be
+// replayed after rollback.
 //
 // Normally, a resumable job's step progress is recorded after it runs along
 // with its result status. This is normally sufficient, but because it happens
@@ -75,23 +76,16 @@ func resumableSetStepTx[TTx any, TArgs JobArgs](ctx context.Context, tx TTx, job
 		rivercommon.MetadataKeyResumableStep: step,
 	}
 
-	state.CompletedStep = step
-	if cursor != nil {
-		if state.Cursors == nil {
-			state.Cursors = make(map[string]json.RawMessage)
-		}
-		state.Cursors[step] = cursor
+	cursors := maps.Clone(state.Cursors)
+	if cursors == nil {
+		cursors = make(map[string]json.RawMessage)
 	}
-	if len(state.Cursors) > 0 {
-		metadataUpdates[rivercommon.MetadataKeyResumableCursor] = state.Cursors
+	if cursor != nil {
+		cursors[step] = cursor
 	}
-
-	workMetadataUpdates, hasWorkMetadataUpdates := jobexecutor.MetadataUpdatesFromWorkContext(ctx)
-	if hasWorkMetadataUpdates {
-		workMetadataUpdates[rivercommon.MetadataKeyResumableStep] = step
-		if resumableCursorMetadata, ok := metadataUpdates[rivercommon.MetadataKeyResumableCursor]; ok {
-			workMetadataUpdates[rivercommon.MetadataKeyResumableCursor] = resumableCursorMetadata
-		}
+	metadataUpdates[rivercommon.MetadataKeyResumableCursor] = nil
+	if len(cursors) > 0 {
+		metadataUpdates[rivercommon.MetadataKeyResumableCursor] = cursors
 	}
 
 	metadataUpdatesBytes, err := json.Marshal(metadataUpdates)
@@ -115,6 +109,13 @@ func resumableSetStepTx[TTx any, TArgs JobArgs](ctx context.Context, tx TTx, job
 		return nil, err
 	}
 
+	// The database owns this checkpoint, including rollback of the caller's
+	// transaction. Only progress recorded after this write belongs in deferred
+	// attempt metadata; replaying it could persist rolled-back application work.
+	state.CompletedStep = ""
+	state.Cursors = cursors
+	state.CursorsDirty = false
+
 	result := &Job[TArgs]{JobRow: updatedJob}
 	if err := json.Unmarshal(result.EncodedArgs, &result.Args); err != nil {
 		return nil, err
diff --git a/resumable_step_tx_test.go b/resumable_step_tx_test.go
index da1c3280b..881577212 100644
--- a/resumable_step_tx_test.go
+++ b/resumable_step_tx_test.go
@@ -3,7 +3,9 @@ package river
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"testing"
+	"time"
 
 	"github.com/jackc/pgx/v5"
 	"github.com/stretchr/testify/require"
@@ -17,10 +19,89 @@ import (
 	"github.com/riverqueue/river/riverdriver/riverpgxv5"
 	"github.com/riverqueue/river/rivershared/riversharedtest"
 	"github.com/riverqueue/river/rivershared/testfactory"
+	"github.com/riverqueue/river/rivershared/testsignal"
 	"github.com/riverqueue/river/rivershared/util/testutil"
 	"github.com/riverqueue/river/rivertype"
 )
 
+func TestResumableAttemptMetadata(t *testing.T) {
+	t.Parallel()
+
+	type args struct {
+		testutil.JobArgsReflectKind[args]
+	}
+	for _, outcome := range []string{"Error", "Rollback", "Shutdown", "Snooze"} {
+		t.Run(outcome, func(t *testing.T) {
+			t.Parallel()
+
+			ctx := context.Background()
+			pool := riversharedtest.DBPool(ctx, t)
+			schema := riverdbtest.TestSchema(ctx, t, riverpgxv5.New(pool), nil)
+			config := newTestConfig(t, schema)
+			var entered testsignal.TestSignal[struct{}]
+			entered.Init(t)
+			AddWorker(config.Workers, WorkFunc(func(ctx context.Context, job *Job[args]) error {
+				if outcome == "Rollback" {
+					ResumableStep(ctx, "first", nil, func(context.Context) error { return nil })
+				}
+				ResumableStepCursor(ctx, "loop", nil, func(ctx context.Context, _ int) error {
+					if outcome == "Rollback" {
+						return pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error {
+							_, err := ResumableSetStepCursorTx[*riverpgxv5.Driver](ctx, tx, job, 1)
+							if err != nil {
+								return err
+							}
+							return errors.New("rollback")
+						})
+					}
+					if err := ResumableSetCursor(ctx, 7); err != nil {
+						return err
+					}
+					switch outcome {
+					case "Shutdown":
+						entered.Signal(struct{}{})
+						<-ctx.Done()
+						return ctx.Err()
+					case "Snooze":
+						return JobSnooze(time.Minute)
+					default:
+						return errors.New("retry")
+					}
+				})
+				return nil
+			}))
+			client := newTestClient(t, pool, config)
+			events := subscribe(t, client)
+			startClient(ctx, t, client)
+			inserted, err := client.Insert(ctx, args{}, &InsertOpts{MaxAttempts: 1})
+			require.NoError(t, err)
+			if outcome == "Shutdown" {
+				entered.WaitOrTimeout()
+				require.NoError(t, client.StopAndCancel(ctx))
+			}
+			event := riversharedtest.WaitOrTimeout(t, events)
+			require.Equal(t, inserted.Job.ID, event.Job.ID)
+			if outcome == "Rollback" {
+				require.Equal(t, EventKindJobFailed, event.Kind)
+				require.JSONEq(t, `{}`, string(event.Job.Metadata))
+			} else {
+				expectedKind := EventKindJobFailed
+				if outcome == "Shutdown" {
+					expectedKind = EventKindJobInterrupted
+				}
+				if outcome == "Snooze" {
+					expectedKind = EventKindJobSnoozed
+				}
+				require.Equal(t, expectedKind, event.Kind)
+				var metadata map[string]any
+				require.NoError(t, json.Unmarshal(event.Job.Metadata, &metadata))
+				require.NotContains(t, metadata, rivercommon.MetadataKeyResumableStep)
+				require.Equal(t, map[string]any{"loop": float64(7)}, metadata[rivercommon.MetadataKeyResumableCursor])
+			}
+		})
+	}
+}
+
 func TestResumableSetStepTx(t *testing.T) {
 	t.Parallel()
 
@@ -58,6 +139,50 @@ func TestResumableSetStepTx(t *testing.T) {
 		}
 	}
 
+	for _, mode := range []string{"Commit", "Rollback", "WriteFailure"} {
+		t.Run("Checkpoint"+mode+"ThenError", func(t *testing.T) {
+			t.Parallel()
+
+			ctx, bundle := setup(ctx, 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 {
+				ResumableStep(ctx, "first", nil, func(context.Context) error { return nil })
+				ResumableStepCursor(ctx, "loop", nil, func(ctx context.Context, _ int) error {
+					err := pgx.BeginFunc(ctx, bundle.tx, func(tx pgx.Tx) error {
+						checkpointJob := &Job[JobArgs]{JobRow: job}
+						if mode == "WriteFailure" {
+							checkpointJob.ID = -1
+						}
+						_, err := ResumableSetStepCursorTx[*riverpgxv5.Driver](ctx, tx, checkpointJob, 1)
+						if mode == "WriteFailure" {
+							require.ErrorIs(t, err, rivertype.ErrNotFound)
+							return err
+						}
+						require.NoError(t, err)
+						if mode == "Rollback" {
+							return errors.New("rollback")
+						}
+						return nil
+					})
+					if err != nil {
+						return err
+					}
+					return errors.New("after commit")
+				})
+				return errors.New("attempt failed")
+			})
+			require.Error(t, err)
+			updates, ok := jobexecutor.MetadataUpdatesFromWorkContext(ctx)
+			require.True(t, ok)
+			if mode == "WriteFailure" {
+				require.Equal(t, "first", updates[rivercommon.MetadataKeyResumableStep])
+			} else {
+				require.NotContains(t, updates, rivercommon.MetadataKeyResumableStep)
+			}
+			require.NotContains(t, updates, rivercommon.MetadataKeyResumableCursor)
+		})
+	}
+
 	t.Run("SetsStep", func(t *testing.T) {
 		t.Parallel()
 
@@ -106,7 +231,7 @@ func TestResumableSetStepTx(t *testing.T) {
 
 		metadataUpdates, ok := jobexecutor.MetadataUpdatesFromWorkContext(ctx)
 		require.True(t, ok)
-		require.Equal(t, "step2", metadataUpdates[rivercommon.MetadataKeyResumableStep])
+		require.Empty(t, metadataUpdates)
 	})
 
 	t.Run("ErrorIfNotRunning", func(t *testing.T) {
diff --git a/resumable_test.go b/resumable_test.go
index ce59bf9b4..8ee55cd34 100644
--- a/resumable_test.go
+++ b/resumable_test.go
@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -201,6 +202,54 @@ func TestResumableStepCursor(t *testing.T) {
 		require.Equal(t, []string{"step"}, ran)
 	})
 
+	for _, outcome := range []string{"Cancel", "Error", "Shutdown", "Snooze"} {
+		t.Run("FirstStepCursorOn"+outcome, func(t *testing.T) {
+			t.Parallel()
+
+			ctx, updates, job := setup(t, `{}`)
+			ctx, cancel := context.WithCancel(ctx)
+			defer cancel()
+
+			err := (&riverplugin.ResumableMiddleware{}).Work(ctx, job, func(ctx context.Context) error {
+				ResumableStepCursor(ctx, "loop", nil, func(ctx context.Context, cursor resumableCursor) error {
+					require.Zero(t, cursor.ID)
+					require.NoError(t, ResumableSetCursor(ctx, resumableCursor{ID: 7}))
+					switch outcome {
+					case "Cancel":
+						return JobCancel(errors.New("cancel"))
+					case "Shutdown":
+						cancel()
+						return ctx.Err()
+					case "Snooze":
+						return JobSnooze(time.Minute)
+					default:
+						return errors.New("retry")
+					}
+				})
+				return nil
+			})
+			require.Error(t, err)
+			encoded, err := json.Marshal(updates)
+			require.NoError(t, err)
+			require.JSONEq(t, `{"river:resumable_cursor":{"loop":{"id":7}}}`, string(encoded))
+
+			// Without a completed step, start from the beginning and supply the
+			// saved cursor when its named step is reached.
+			ctx, updates, job = setup(t, string(encoded))
+			err = (&riverplugin.ResumableMiddleware{}).Work(ctx, job, func(ctx context.Context) error {
+				ResumableStepCursor(ctx, "loop", nil, func(ctx context.Context, cursor resumableCursor) error {
+					require.Equal(t, 7, cursor.ID)
+					return nil
+				})
+				return errors.New("later failure")
+			})
+			require.EqualError(t, err, "later failure")
+			require.Equal(t, "loop", updates[rivercommon.MetadataKeyResumableStep])
+			require.Contains(t, updates, rivercommon.MetadataKeyResumableCursor)
+			require.Nil(t, updates[rivercommon.MetadataKeyResumableCursor])
+		})
+	}
+
 	t.Run("ResumesCursor", func(t *testing.T) {
 		t.Parallel()
 
diff --git a/ruby/spec/conformance_spec.rb b/ruby/spec/conformance_spec.rb
index 918c1b967..5fecf4d66 100644
--- a/ruby/spec/conformance_spec.rb
+++ b/ruby/spec/conformance_spec.rb
@@ -268,6 +268,46 @@ class ConformanceRecord < ActiveRecord::Base
         end
       end
 
+      RiverConformanceFixtures.load("resumable").fetch("cases").each do |fixture|
+        it "matches Go's resumable #{fixture.fetch("name")}" do
+          driver = @driver
+          name = fixture.fetch("name")
+          worker = Object.new
+          worker.define_singleton_method(:work) do |job|
+            job.resumable_step("first") {} unless name == "first_cursor_error"
+            raise River.job_snooze(60) if name == "step_snooze"
+            job.resumable_step_cursor("loop") do
+              if name == "first_cursor_error"
+                job.resumable_set_cursor(7)
+              else
+                driver.transaction do
+                  if name == "checkpoint_write_failure"
+                    # Exercise the write failure boundary without disrupting
+                    # the result update on the same database connection.
+                    original = driver.method(:job_metadata_merge)
+                    begin
+                      driver.define_singleton_method(:job_metadata_merge) { |*_args| raise "write failed" }
+                      job.resumable_checkpoint(cursor: 1)
+                    ensure
+                      driver.define_singleton_method(:job_metadata_merge, original)
+                    end
+                  else
+                    job.resumable_checkpoint(cursor: 1)
+                    raise "rollback" if name == "checkpoint_rollback_error"
+                  end
+                end
+              end
+              job.resumable_set_cursor(2) if name == "checkpoint_then_cursor_error"
+              raise "retry"
+            end
+          end
+          client = River::Client.new(driver, config: River::Config.new(workers: River::Workers.new.add(:resumable, worker)))
+          row = client.insert(River::JobArgsHash.new(:resumable, {}), max_attempts: 1).job
+          result = River::Testing.perform_job(client, row.id)
+          expect(result.job.metadata.select { |key, _| key.start_with?("river:resumable_") }).to eq(fixture.fetch("expected_metadata"))
+        end
+      end
+
       it "resumes from and persists Go's step and cursor metadata" do
         keys = protocol.fetch("metadata_keys")
         calls = []
diff --git a/rust/CHANGELOG.md b/rust/CHANGELOG.md
index 161da85e0..9909c7bd4 100644
--- a/rust/CHANGELOG.md
+++ b/rust/CHANGELOG.md
@@ -18,6 +18,10 @@ Changes to River for Go are recorded in the [repository changelog](../CHANGELOG.
 
 - **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
+
+- 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.
+
 ## [0.3.0] - 2026-10-07
 
 ### Changed
diff --git a/rust/riverqueue-test/src/lib.rs b/rust/riverqueue-test/src/lib.rs
index e86199131..febc55bb4 100644
--- a/rust/riverqueue-test/src/lib.rs
+++ b/rust/riverqueue-test/src/lib.rs
@@ -218,8 +218,10 @@ where
             .map_err(TestWorkError::Worker),
         Err(error) => Err(TestWorkError::Resumable(error)),
     };
-    if let Some(error) = __private::work_context_resumable_finish(&context, result.is_err())
-        && result.is_ok()
+    if let Some(error) = __private::work_context_resumable_finish(
+        &context,
+        !matches!(result, Ok(WorkOutcome::Complete)),
+    ) && result.is_ok()
     {
         result = Err(TestWorkError::Resumable(error));
     }
@@ -244,6 +246,37 @@ mod tests {
 
     use super::*;
 
+    struct ResumableSnoozeWorker;
+
+    impl Worker<TestArgs> for ResumableSnoozeWorker {
+        type Error = Error;
+
+        async fn work(
+            &self,
+            context: WorkContext,
+            _job: Job<TestArgs>,
+        ) -> Result<WorkOutcome, Self::Error> {
+            context
+                .resumable_step("first", || async { Ok::<_, Error>(()) })
+                .await?;
+            Ok(WorkOutcome::Snooze(Duration::from_secs(60)))
+        }
+    }
+
+    #[tokio::test]
+    async fn resumable_progress_survives_snooze() {
+        let job = TestJobBuilder::new(TestArgs {
+            message: "snooze".to_owned(),
+        })
+        .build()
+        .unwrap();
+        let attempt = work_once(&ResumableSnoozeWorker, job).await;
+        assert_eq!(
+            serde_json::Value::Object(attempt.metadata_updates),
+            serde_json::json!({"river:resumable_step": "first"})
+        );
+    }
+
     struct ResumableWorker {
         calls: AtomicUsize,
     }
diff --git a/rust/riverqueue/src/client/executor.rs b/rust/riverqueue/src/client/executor.rs
index efe2be2ba..1e71f8669 100644
--- a/rust/riverqueue/src/client/executor.rs
+++ b/rust/riverqueue/src/client/executor.rs
@@ -114,7 +114,7 @@ pub(super) async fn execute_job(
 
         let run_duration = work_started.elapsed();
         let mut result = result;
-        if let Some(resumable_failure) = context.resumable_finish(result.is_err())
+        if let Some(resumable_failure) = context.resumable_finish(!matches!(result, Ok(WorkOutcome::Complete)))
             && result.is_ok()
         {
             result = Err(WorkerFailure {
diff --git a/rust/riverqueue/src/worker.rs b/rust/riverqueue/src/worker.rs
index 1b0c2b486..38bf54f14 100644
--- a/rust/riverqueue/src/worker.rs
+++ b/rust/riverqueue/src/worker.rs
@@ -247,7 +247,9 @@ impl WorkContext {
         match result {
             Ok(()) => {
                 state.completed_step = Some(name.to_owned());
-                state.cursors.remove(name);
+                if state.cursors.remove(name).is_some() {
+                    state.cursors_dirty = true;
+                }
                 Ok(())
             }
             Err(error) => Err(state.fail_step(name, error.into())),
@@ -270,10 +272,13 @@ impl WorkContext {
             )
         })?;
         state.cursors.insert(step_name, cursor);
+        state.cursors_dirty = true;
         Ok(())
     }
 
     /// Persists the current resumable step in a caller-managed transaction.
+    /// After a successful write the transaction owns that progress. Rolling it
+    /// back may replay earlier progress that was not already persisted.
     ///
     /// # Errors
     ///
@@ -294,7 +299,7 @@ impl WorkContext {
     }
 
     /// Persists the current resumable step and cursor in a caller-managed
-    /// transaction.
+    /// transaction. Rolling it back may replay earlier unpersisted progress.
     ///
     /// # Errors
     ///
@@ -329,48 +334,57 @@ impl WorkContext {
     {
         let metadata = self.resumable_checkpoint(cursor)?;
         let (client, job_id) = self.current_job()?;
-        client
+        let updated = client
             .job_update_tx(
                 connection,
                 job_id,
                 JobUpdateParams {
-                    metadata,
+                    metadata: metadata.clone(),
                     output: None,
                 },
             )
-            .await
+            .await?;
+        // The caller transaction owns the checkpoint, including rollback.
+        // Only subsequent progress belongs in deferred attempt metadata.
+        let mut state = self.lock_resumable();
+        state.cursors = metadata[crate::METADATA_KEY_RESUMABLE_CURSOR]
+            .as_object()
+            .cloned()
+            .unwrap_or_default();
+        state.cursors_dirty = false;
+        state.completed_step = None;
+        Ok(updated)
     }
 
-    /// Marks the current step complete, records `cursor` for it, and returns
-    /// the checkpoint metadata to persist.
+    /// Returns checkpoint metadata without changing deferred progress.
     fn resumable_checkpoint<T: Serialize>(
         &self,
         cursor: Option<&T>,
     ) -> Result<Map<String, Value>, Error> {
-        let mut state = self.lock_resumable();
+        let state = self.lock_resumable();
         let step_name = state.step_name.clone().ok_or_else(|| {
             Error::runtime_context(
                 "worker context",
                 "resumable checkpoint must be set inside a resumable step".to_owned(),
             )
         })?;
-        state.completed_step = Some(step_name.clone());
+        let mut cursors = state.cursors.clone();
         if let Some(cursor) = cursor {
-            state
-                .cursors
-                .insert(step_name.clone(), serde_json::to_value(cursor)?);
+            cursors.insert(step_name.clone(), serde_json::to_value(cursor)?);
         }
         let mut metadata = Map::new();
         metadata.insert(
             crate::METADATA_KEY_RESUMABLE_STEP.to_owned(),
             step_name.into(),
         );
-        if !state.cursors.is_empty() {
-            metadata.insert(
-                crate::METADATA_KEY_RESUMABLE_CURSOR.to_owned(),
-                Value::Object(state.cursors.clone()),
-            );
-        }
+        metadata.insert(
+            crate::METADATA_KEY_RESUMABLE_CURSOR.to_owned(),
+            if cursors.is_empty() {
+                Value::Null
+            } else {
+                Value::Object(cursors)
+            },
+        );
         Ok(metadata)
     }
 
@@ -488,22 +502,22 @@ impl WorkContext {
                 )))
             })
         });
-        if (worker_failed || failure.is_some())
-            && let Some(completed_step) = &state.completed_step
-        {
+        if worker_failed || failure.is_some() {
             let mut updates = self.lock_metadata();
-            updates.insert(
-                crate::METADATA_KEY_RESUMABLE_STEP.to_owned(),
-                completed_step.clone().into(),
-            );
-            if state.cursors.is_empty() {
-                if state.had_cursors {
-                    updates.insert(crate::METADATA_KEY_RESUMABLE_CURSOR.to_owned(), Value::Null);
-                }
-            } else {
+            if let Some(completed_step) = &state.completed_step {
+                updates.insert(
+                    crate::METADATA_KEY_RESUMABLE_STEP.to_owned(),
+                    completed_step.clone().into(),
+                );
+            }
+            if state.cursors_dirty {
                 updates.insert(
                     crate::METADATA_KEY_RESUMABLE_CURSOR.to_owned(),
-                    Value::Object(state.cursors.clone()),
+                    if state.cursors.is_empty() {
+                        Value::Null
+                    } else {
+                        Value::Object(state.cursors.clone())
+                    },
                 );
             }
         }
@@ -527,8 +541,8 @@ struct ResumableState {
     all_step_names: HashSet<String>,
     completed_step: Option<String>,
     cursors: Map<String, Value>,
+    cursors_dirty: bool,
     failure: Option<WorkError>,
-    had_cursors: bool,
     resume_matched: bool,
     resume_step: Option<String>,
     step_name: Option<String>,
@@ -540,8 +554,8 @@ impl Default for ResumableState {
             all_step_names: HashSet::new(),
             completed_step: None,
             cursors: Map::new(),
+            cursors_dirty: false,
             failure: None,
-            had_cursors: false,
             resume_matched: true,
             resume_step: None,
             step_name: None,
@@ -561,7 +575,6 @@ impl ResumableState {
         match metadata.get_raw(crate::METADATA_KEY_RESUMABLE_CURSOR) {
             Some(raw) if raw.get().starts_with('{') => {
                 if let Ok(cursors) = serde_json::from_str::<Map<String, Value>>(raw.get()) {
-                    state.had_cursors = !cursors.is_empty();
                     state.cursors = cursors;
                 } else {
                     state.failure = Some(WorkError::new(Box::new(Error::invalid_job(
@@ -1178,6 +1191,41 @@ mod tests {
             .unwrap();
     }
 
+    #[tokio::test]
+    async fn resumable_first_step_cursor_survives_failure() {
+        let row = job_row(FunctionJobArgs::KIND, false);
+        let context = WorkContext::for_test_job(&row);
+        context
+            .resumable_step_with_cursor("loop", |_: i64| async {
+                context.resumable_set_cursor(&7)?;
+                Err::<(), Error>(Error::invalid_job("retry"))
+            })
+            .await
+            .unwrap_err();
+        context.resumable_finish(true);
+        assert_eq!(
+            Value::Object(context.metadata_updates()),
+            json!({"river:resumable_cursor": {"loop": 7}})
+        );
+        let mut retry_row = row;
+        retry_row.metadata = context.metadata_updates().into();
+        let retry = WorkContext::for_test_job(&retry_row);
+        retry
+            .resumable_step_with_cursor("loop", |cursor: i64| async move {
+                assert_eq!(cursor, 7);
+                Ok::<_, Error>(())
+            })
+            .await
+            .unwrap();
+        retry.resumable_finish(true);
+        assert_eq!(
+            Value::Object(retry.metadata_updates()),
+            json!({
+                "river:resumable_step": "loop", "river:resumable_cursor": null
+            })
+        );
+    }
+
     #[tokio::test]
     async fn resumable_nested_steps_restore_parent_and_error_sources() {
         let context = WorkContext::new(CancellationToken::new());
@@ -1433,6 +1481,22 @@ mod tests {
         assert!(error.get_ref().downcast_ref::<FunctionError>().is_some());
     }
 
+    #[test]
+    fn resumable_preparing_checkpoint_does_not_advance_deferred_progress() {
+        let context = WorkContext::new(CancellationToken::new());
+        {
+            let mut state = context.lock_resumable();
+            state.completed_step = Some("first".to_owned());
+            state.step_name = Some("loop".to_owned());
+        }
+        context.resumable_checkpoint(Some(&1)).unwrap();
+        context.resumable_finish(true);
+        assert_eq!(
+            Value::Object(context.metadata_updates()),
+            json!({"river:resumable_step": "first"})
+        );
+    }
+
     #[tokio::test]
     async fn resumable_steps_accept_anyhow_errors() {
         let context = WorkContext::new(CancellationToken::new());
diff --git a/rust/riverqueue/tests/postgres.rs b/rust/riverqueue/tests/postgres.rs
index 17b3c4054..5f5a6c7aa 100644
--- a/rust/riverqueue/tests/postgres.rs
+++ b/rust/riverqueue/tests/postgres.rs
@@ -1831,10 +1831,12 @@ async fn resumable_cursor_and_transactional_checkpoints() {
             .as_deref(),
         Some("tx_step")
     );
-    assert!(
-        !committed_step
+    assert_eq!(
+        committed_step
             .metadata
-            .contains_key("river:resumable_cursor")
+            .get::<serde_json::Value>("river:resumable_cursor")
+            .unwrap(),
+        Some(serde_json::Value::Null)
     );
 
     for mode in ["rollback_cursor", "rollback_step"] {
diff --git a/rust/riverqueue/tests/resumable_conformance.rs b/rust/riverqueue/tests/resumable_conformance.rs
new file mode 100644
index 000000000..819114420
--- /dev/null
+++ b/rust/riverqueue/tests/resumable_conformance.rs
@@ -0,0 +1,131 @@
+//! Shared resumable semantics, exercised through a real worker and caller tx.
+#![cfg(feature = "sqlite")]
+
+use std::{collections::BTreeMap, path::Path, time::Duration};
+
+use riverqueue::{
+    BoxError, Client, EventKind, InsertOpts, Job, JobArgs, QueueConfig, WorkContext, WorkOutcome,
+    Workers,
+};
+use riverqueue_migrate::SqliteMigrator;
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use sqlx::sqlite::SqlitePoolOptions;
+
+#[derive(Clone, Debug, Deserialize, JobArgs, Serialize)]
+#[river(kind = "resumable_conformance")]
+struct Args {
+    scenario: String,
+}
+
+#[tokio::test]
+async fn resumable_conformance() {
+    let fixture: Value = serde_json::from_str(
+        &std::fs::read_to_string(
+            Path::new(env!("CARGO_MANIFEST_DIR")).join("../../conformance/testdata/resumable.json"),
+        )
+        .unwrap(),
+    )
+    .unwrap();
+    let pool = SqlitePoolOptions::new()
+        .max_connections(1)
+        .connect("sqlite::memory:")
+        .await
+        .unwrap();
+    SqliteMigrator::new(pool.clone())
+        .migrate_up()
+        .await
+        .unwrap();
+    sqlx::query("CREATE TRIGGER reject_checkpoint BEFORE UPDATE OF metadata ON river_job WHEN json_extract(NEW.args, '$.scenario') = 'checkpoint_write_failure' AND json_extract(NEW.metadata, '$.\"river:resumable_step\"') = 'loop' BEGIN SELECT RAISE(ABORT, 'checkpoint write failed'); END").execute(&pool).await.unwrap();
+    let worker_pool = pool.clone();
+    let mut workers = Workers::new();
+    workers
+        .add_fn(move |context: WorkContext, job: Job<Args>| {
+            run_scenario(context, job, worker_pool.clone())
+        })
+        .unwrap();
+    let client = Client::builder(pool.clone())
+        .workers(workers)
+        .without_notifications()
+        .queue(
+            "default",
+            QueueConfig::new(1)
+                .with_fetch_cooldown(Duration::from_millis(1))
+                .with_fetch_poll_interval(Duration::from_millis(10)),
+        )
+        .build()
+        .unwrap();
+    let mut events = client
+        .subscribe(&[EventKind::JobFailed, EventKind::JobSnoozed])
+        .unwrap();
+    let mut expected = BTreeMap::new();
+    for case in fixture["cases"].as_array().unwrap() {
+        let inserted = client
+            .insert(Args {
+                scenario: case["name"].as_str().unwrap().to_owned(),
+            })
+            .opts(InsertOpts::default().with_max_attempts(1))
+            .await
+            .unwrap();
+        expected.insert(inserted.job.row.id, case["expected_metadata"].clone());
+    }
+    let mut run = client.start().unwrap();
+    for _ in 0..expected.len() {
+        let event = tokio::time::timeout(Duration::from_secs(10), events.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        let row = &event.as_job().unwrap().job;
+        let metadata: Value = serde_json::to_value(&row.metadata).unwrap();
+        let actual: serde_json::Map<String, Value> = metadata
+            .as_object()
+            .unwrap()
+            .iter()
+            .filter(|(key, _)| key.starts_with("river:resumable_"))
+            .map(|(k, v)| (k.clone(), v.clone()))
+            .collect();
+        assert_eq!(
+            Value::Object(actual),
+            expected.remove(&row.id).unwrap(),
+            "job {}",
+            row.id
+        );
+    }
+    run.stop().await.unwrap();
+}
+
+async fn run_scenario(
+    context: WorkContext,
+    job: Job<Args>,
+    pool: sqlx::SqlitePool,
+) -> Result<WorkOutcome, BoxError> {
+    let name = job.args.scenario;
+    if name != "first_cursor_error" {
+        context
+            .resumable_step("first", || async { Ok::<_, BoxError>(()) })
+            .await?;
+    }
+    if name == "step_snooze" {
+        return Ok::<_, BoxError>(WorkOutcome::Snooze(Duration::from_secs(60)));
+    }
+    context
+        .resumable_step_with_cursor("loop", |_: i64| async {
+            if name == "first_cursor_error" {
+                context.resumable_set_cursor(&7)?;
+            } else {
+                let mut tx = riverqueue::database::begin_sqlite_write(&pool).await?;
+                context.resumable_set_step_cursor_tx(&mut tx, &1).await?;
+                if name == "checkpoint_commit_error" || name == "checkpoint_then_cursor_error" {
+                    tx.commit().await?;
+                } else {
+                    tx.rollback().await?;
+                }
+            }
+            if name == "checkpoint_then_cursor_error" {
+                context.resumable_set_cursor(&2)?;
+            }
+            Err::<(), BoxError>("retry".into())
+        })
+        .await?;
+    Ok(WorkOutcome::Complete)
+}