Fix recurring nudge persistence and add failover simulations

CHANGELOG.md

@@ -2,6 +2,46 @@
 
 ## Unreleased
 
+- Fixed Postgres self-owned writes being silently rolled back when the host
+  `DataSource` hands out connections with `autoCommit=false` (issue #111).
+  Queue pauses, execution and node heartbeats, maintenance leases, retention,
+  mutexes, recurring definitions/state, and the expired-dedup fallback now use
+  explicit Threadmill-owned transactions, restoring the connection's prior
+  auto-commit mode after commit or rollback. `saveAtomic`, `softDelete`,
+  `claimReady`, and `replaceJob` now use that same boundary instead of forcing
+  auto-commit on return. Migration history bootstrap and destructive schema
+  reset also commit explicitly. The expired-dedup fallback now keeps its
+  `SELECT … FOR UPDATE` lock through the delete, closing a lost-update window
+  that also existed on auto-commit pools.
+- The recurring materializer now repairs a timing-fingerprint mismatch found
+  during its under-mutex definition reload (issue #112). The current task
+  definition wins: Threadmill preserves run bookkeeping and pending nudge
+  demand, recomputes timing forward from the tick, and does not fire or
+  `CATCH_UP` the obsolete trigger's overdue backlog. A listed fingerprint
+  mismatch now triggers that repair even when the stale `nextRunAt` is still in
+  the future, so a faster replacement schedule cannot remain dormant until the
+  obsolete fire time. Legacy null fingerprints are adopted without dropping or
+  moving an existing fire, and detected crashed edits emit an operator-visible
+  warning.
+- Added fixed process-separated nudge simulations for Postgres and Redis
+  (issue #114). They hard-kill a maintenance leader after an accepted nudge
+  and prove the standby serves it, then hard-kill a producer after its durable
+  work write but before its nudge and prove the regular recurring backstop
+  drains the row. Cross-process JSON-lines traces record process ids,
+  leadership, trigger origins, and the verified event ordering. Ready markers
+  are atomically published and trace writes drain their buffers fully; the
+  fixed real-backend simulation is part of `productionCheck`. The failover
+  phase uses a one-minute backstop and five-minute leader poll, then edits the
+  task to an eight-second backstop for the producer-crash phase, eliminating
+  process-start timing races. Simulation work uses its own datastore namespace
+  and long-lived per-process connections.
+- Recorded the issue #113 decision not to add per-task lifecycle generations
+  to the `JobStore` SPI. Issue #114 proves hard-kill handoff and backstop
+  recovery, while the remaining revision-reuse ABA additionally requires an
+  old materializer to resume after outliving its mutex lease and a same-name
+  delete/re-registration. A persistent generation would not replace the
+  existing lifecycle orderings and does not justify its cross-backend storage,
+  migration, and SPI cost without evidence that sequence occurs.
 - Recurring tasks can declare claim-time exclusivity (issue #110). A
   `CronTask` gains an `exclusive` flag, surfaced as
   `@Recurring(exclusive = true)` and as an `exclusive` parameter on
@@ -137,7 +177,8 @@
   trigger timing its `next_run_at` was computed from, written atomically with
   the state row (additive Postgres migration
   `V4__cron_state_timing_fingerprint.sql`; a Redis hash field; legacy rows
-  read as null and simply recompute once). The unchanged-schedule decision
+  read as null and adopt the current fingerprint without moving their recorded
+  timing). The unchanged-schedule decision
   reads this fingerprint rather than comparing stored task definitions, so a
   crash between the separate task-definition and state writes can never pair
   a new trigger with old timing undetectably — the retry detects the

build.gradle.kts

@@ -228,6 +228,9 @@ tasks.register("productionCheck") {
         // The correctness simulation is the gate that caught the C1
         // in-memory concurrency bypass — a release candidate must run it.
         ":threadmill-simulation:simulate",
+        // The process-separated nudge simulation pins maintenance-leader
+        // hard-kill handoff and the documented producer crash window.
+        ":threadmill-simulation:simulateNudge",
     )
     dependsOn(":threadmill-example:run")
 }
@@ -240,6 +243,9 @@ tasks
 
 subprojects {
     tasks
-        .matching { it.name in setOf("jar", "test", "javadoc", "soak", "run") }
+        .matching {
+            it.name in setOf("jar", "test", "javadoc", "soak", "run") ||
+                it.name.startsWith("simulate")
+        }
         .configureEach { mustRunAfter(cleanTask) }
 }

docs/transactions.md

@@ -258,9 +258,10 @@ closing the crash window between the caller's commit and the nudge write, and
 that window is an explicit non-goal: the backstop schedule bounds the
 worst-case latency, and a lost nudge costs one schedule period, never a run.
 
-On the hot path the write is a microseconds-held autocommit update, and an
-in-JVM per-task coalescer additionally bounds the store write rate to about
-one round trip regardless of producer rate.
+On the hot path the write is a short Threadmill-owned transaction that commits
+independently of the pool's `autoCommit` default. An in-JVM per-task coalescer
+additionally bounds the store write rate to about one round trip regardless of
+producer rate.
 
 ## Connection-pool sharing
 

threadmill-core/src/main/java/com/hemju/threadmill/core/schedule/CronTaskScheduleState.java

@@ -25,7 +25,8 @@
  *                          or {@code null} for rows written before the
  *                          fingerprint existed. Written atomically with
  *                          {@code nextRunAt}, it lets {@code Scheduler.upsertCron}
- *                          decide preserve-vs-recompute from this record alone —
+ *                          and {@link RecurringMaterializer} decide
+ *                          preserve-vs-recompute from this record alone —
  *                          a crash between the separate task and state writes can
  *                          never pair a new trigger with old timing undetectably
  * @param nudgeRequestedAt  when an on-demand materialization ("nudge") was most
@@ -83,10 +84,11 @@ public CronTaskScheduleState(
     }
 
     /**
-     * Convenience constructor with no timing fingerprint. A null fingerprint
-     * is always safe — the next re-registration recomputes the schedule
-     * instead of preserving it — but production writers that recompute
-     * {@code nextRunAt} from a task should stamp
+     * Convenience constructor with no timing fingerprint. A materializer tick
+     * adopts the current task's fingerprint without moving a non-null
+     * {@code nextRunAt}; if timing is also null, it initializes the next fire
+     * from that tick. Production writers that compute {@code nextRunAt} from a
+     * task should stamp
      * {@link #timingFingerprintOf(CronTask)} so unchanged re-registrations
      * can preserve overdue state.
      */

threadmill-core/src/main/java/com/hemju/threadmill/core/schedule/RecurringMaterializer.java

@@ -35,6 +35,13 @@
  * with carry-over. That cap — not re-registration — is the catch-up-storm
  * defense.
  *
+ * <p>If the under-mutex definition reload finds that the schedule state's
+ * timing fingerprint belongs to a different definition, the definition wins:
+ * timing is recomputed forward from the current tick and the stale schedule
+ * produces no firing. This completes a crashed timing edit without running a
+ * trigger the user already replaced. {@code CATCH_UP} resumes normally from
+ * the repaired timing; it never catches up the obsolete trigger's backlog.
+ *
  * <p>If a previously-materialised instance is still un-terminal, no new
  * instance is created until that one finishes. This guard prevents
  * pile-up under long-running recurring work.
@@ -134,24 +141,72 @@ private void tickOneLocked(CronTask listed, Instant now) {
         var state = stateOpt.get();
         Long nudge = state.nudgeRequestedAt() == null ? null : state.nudgeRevision();
         boolean due = state.nextRunAt() != null && !state.nextRunAt().isAfter(now);
-        if (!due && nudge == null) return;
+        boolean listedFingerprintMismatch =
+                !CronTaskScheduleState.timingFingerprintOf(listed).equals(state.timingFingerprint());
+        if (!due && nudge == null && !listedFingerprintMismatch) return;
 
         // About to act — reload the definition now that we hold the task
         // mutex. The listed object was snapshotted by tick() BEFORE the
         // mutex: a re-registration, edit, or disable can commit in between,
         // and materializing from the stale object would insert the old
         // handler/payload (and, for a nudge, consume a request that was made
         // against the new definition). The reload is deliberately done only
-        // when a materialization is imminent, so idle ticks stay at one
-        // state read per task.
+        // when a materialization is imminent or the listed definition already
+        // proves the timing state is stale, so ordinary idle ticks stay at one
+        // state read per task while future stale schedules self-heal promptly.
         CronTask task = store.findCronTask(listed.name()).orElse(null);
         if (task == null || !task.enabled()) return;
 
+        String fingerprint = CronTaskScheduleState.timingFingerprintOf(task);
+        boolean timingStateChanged = false;
+        if (!fingerprint.equals(state.timingFingerprint())) {
+            String previousFingerprint = state.timingFingerprint();
+            boolean legacyTiming = previousFingerprint == null && state.nextRunAt() != null;
+            // A non-null mismatch is the crash signature for a timing edit
+            // that wrote the definition before its schedule state. Finish it
+            // by scheduling forward from this tick: firing the stale timing
+            // would run a trigger the user already replaced. A legacy null
+            // fingerprint does not prove an edit, so adopt the fingerprint
+            // without dropping or moving an already-recorded firing.
+            Instant next = legacyTiming ? state.nextRunAt() : task.trigger().nextAfter(now, task.zone());
+            if (!legacyTiming) due = false;
+            state = new CronTaskScheduleState(
+                    task.name(),
+                    state.lastRunAt(),
+                    state.lastRunJobId(),
+                    next,
+                    state.inFlightJobId(),
+                    fingerprint,
+                    // These cells are carried in the in-memory record only;
+                    // upsertCronTaskState deliberately never writes them.
+                    state.nudgeRequestedAt(),
+                    state.nudgeRevision());
+            timingStateChanged = true;
+            if (previousFingerprint != null) {
+                LOG.warn(
+                        "Repairing stale recurring timing for task {} from fingerprint {} to {}; next run at {}",
+                        task.name(),
+                        previousFingerprint,
+                        fingerprint,
+                        next);
+            }
+            if (nudge == null && !due) {
+                store.upsertCronTaskState(state);
+                return;
+            }
+        }
+
+        // The listed definition may simply be stale relative to an
+        // authoritative definition and state that already agree. In that
+        // case no scheduled firing or nudge is owed this tick.
+        if (!due && nudge == null) return;
+
         // Pile-up guard: an in-flight instance that is still going to run
         // blocks the next materialization.
         if (state.inFlightJobId() != null) {
             Job inFlight = store.findById(JobId.of(state.inFlightJobId())).orElse(null);
             if (inFlight != null && blocksNextMaterialization(inFlight, now)) {
+                if (timingStateChanged) store.upsertCronTaskState(state);
                 // Still running — leave the next_run_at where it is so we revisit on the next tick.
                 return;
             }
@@ -166,7 +221,7 @@ private void tickOneLocked(CronTask listed, Instant now) {
             // represents no schedule tick), only the nudge origin marker.
             JobId id = materializeNudge(task);
             store.upsertCronTaskState(new CronTaskScheduleState(
-                    task.name(), now, id.asUuid(), state.nextRunAt(), id.asUuid(), state.timingFingerprint()));
+                    task.name(), now, id.asUuid(), state.nextRunAt(), id.asUuid(), fingerprint));
             // Clear AFTER materializing (a crash between the two costs one
             // extra run, never a lost one — see the failure-semantics note in
             // the class Javadoc), and only the observed revision — a nudge
@@ -176,7 +231,6 @@ private void tickOneLocked(CronTask listed, Instant now) {
             return;
         }
 
-        String fingerprint = CronTaskScheduleState.timingFingerprintOf(task);
         if (task.missedRunPolicy() == CronTask.MissedRunPolicy.CATCH_UP) {
             // Materialize every fire from nextRunAt up to and including now,
             // capped per tick so an unbounded backlog cannot occupy the

threadmill-dashboard-api/src/test/java/com/hemju/threadmill/dashboard/api/DashboardApiServiceTest.java

@@ -82,8 +82,8 @@ void manualTriggerDoesNotStealThePileUpGuardFromARunningScheduledInstance() {
 
         // A scheduled instance materialized earlier is still PROCESSING.
         var materializer = new RecurringMaterializer(store);
-        store.upsertCronTaskState(
-                CronTaskScheduleState.initial("report", Instant.now().minusSeconds(60)));
+        store.upsertCronTaskState(CronTaskScheduleState.initial(
+                "report", Instant.now().minusSeconds(60), CronTaskScheduleState.timingFingerprintOf(task)));
         materializer.tick(Instant.now());
         var scheduledInstance =
                 store.claimReady(NodeId.newId(), "default", 1, Instant.now()).getFirst();
@@ -107,7 +107,8 @@ void manualTriggerDoesNotStealThePileUpGuardFromARunningScheduledInstance() {
                 afterTrigger.lastRunAt(),
                 afterTrigger.lastRunJobId(),
                 Instant.now().minusSeconds(1),
-                afterTrigger.inFlightJobId()));
+                afterTrigger.inFlightJobId(),
+                afterTrigger.timingFingerprint()));
         materializer.tick(Instant.now());
         assertThat(store.findByHandlerSignature("com.example.ReportHandler", 10))
                 .hasSize(2); // scheduled instance + manual trigger, nothing else

threadmill-simulation/README.md

@@ -145,6 +145,46 @@ attempts, and periodic store snapshots. It is a production-behaviour trace
 generator, not a replacement for the contract tests. The task exits non-zero
 when the queue does not drain within `--drain-timeout`.
 
+## Process-Separated Nudge Simulation
+
+This fixed simulation starts real Postgres and Redis datastores with
+Testcontainers, then runs the maintenance leader, standby, and producers in
+separate JVMs. Run both backends:
+
+```bash
+./gradlew :threadmill-simulation:simulateNudge
+```
+
+The combined real-backend task is also part of the root `productionCheck`
+release-candidate gate so the crash paths cannot silently rot.
+
+Or select one:
+
+```bash
+./gradlew :threadmill-simulation:simulateNudgePostgres
+./gradlew :threadmill-simulation:simulateNudgeRedis
+```
+
+The leader first registers an exclusive one-minute recurring outbox pump, with
+its own maintenance poll set to five minutes, so process startup and lease
+handoff cannot race the backstop or the leader's next tick. After failover is
+proved, the supervisor edits the backstop to eight seconds for the producer
+crash phase. It verifies two crash windows:
+
+- A producer commits durable work and an accepted nudge. The supervisor
+  hard-kills the maintenance leader before its next materializer tick; the
+  standby must acquire the expired maintenance lease and run the pump with
+  `origin=nudge`.
+- A second producer commits durable work, stops immediately before the nudge,
+  and is hard-killed. No nudge may appear for that row; the regular recurring
+  fire must run with `origin=schedule` and drain it.
+
+The verifier reads a cross-process JSON-lines trace and proves the event
+ordering, distinct process ids, standby ownership, trigger origins, and final
+drain. Artifacts live under
+`build/simulation/nudge-cross-node-<timestamp>-<backend>/`, including
+`trace.jsonl` and one output log per child JVM.
+
 ## Why a separate module
 
 `threadmill-soak` is about sustained load and operational performance

threadmill-simulation/build.gradle.kts

@@ -24,6 +24,7 @@ dependencies {
 // Per-backend simulation tasks. The base `simulate` runs all three sequentially.
 val simulationMainClass = "com.hemju.threadmill.simulation.SimulationMain"
 val workerChurnMainClass = "com.hemju.threadmill.simulation.workerchurn.WorkerChurnSimulatorMain"
+val nudgeSimulationMainClass = "com.hemju.threadmill.simulation.nudge.NudgeSimulationMain"
 
 tasks.register<JavaExec>("simulateMemory") {
     group = "verification"
@@ -71,3 +72,27 @@ tasks.register<JavaExec>("simulateWorkerChurnRedis") {
     mainClass.set(workerChurnMainClass)
     args = listOf("--backend", "redis")
 }
+
+tasks.register<JavaExec>("simulateNudgePostgres") {
+    group = "verification"
+    description =
+        "Run the process-separated leader-kill and producer-kill nudge simulation against PostgreSQL."
+    classpath = sourceSets["main"].runtimeClasspath
+    mainClass.set(nudgeSimulationMainClass)
+    args = listOf("--backend", "postgres")
+}
+
+tasks.register<JavaExec>("simulateNudgeRedis") {
+    group = "verification"
+    description =
+        "Run the process-separated leader-kill and producer-kill nudge simulation against Redis."
+    classpath = sourceSets["main"].runtimeClasspath
+    mainClass.set(nudgeSimulationMainClass)
+    args = listOf("--backend", "redis")
+}
+
+tasks.register("simulateNudge") {
+    group = "verification"
+    description = "Run the process-separated nudge simulation against PostgreSQL and Redis."
+    dependsOn("simulateNudgePostgres", "simulateNudgeRedis")
+}

threadmill-simulation/src/main/java/com/hemju/threadmill/simulation/nudge/NudgeSimulationHandler.java

@@ -0,0 +1,53 @@
+package com.hemju.threadmill.simulation.nudge;
+
+import java.nio.file.Path;
+import java.util.Map;
+
+import com.hemju.threadmill.core.handler.JobExecutionContext;
+import com.hemju.threadmill.core.handler.JobHandler;
+
+/** Recurring outbox-pump handler used by the process-separated nudge simulation. */
+public final class NudgeSimulationHandler implements JobHandler<NudgeSimulationPayload> {
+
+    @Override
+    public void run(NudgeSimulationPayload payload, JobExecutionContext context) {
+        var trace = Path.of(payload.traceFile);
+        var pid = ProcessHandle.current().pid();
+        var jobId = context.jobId().toString();
+        var origin = context.cronOrigin().orElse("unknown");
+        NudgeSimulationTrace.append(
+                trace,
+                "pump-run-start",
+                Map.of(
+                        "runId", payload.runId,
+                        "pid", pid,
+                        "nodeId", context.nodeId().toString(),
+                        "jobId", jobId,
+                        "origin", origin));
+
+        try (var workStore = NudgeSimulationStores.openProcessWorkStore(payload.runId)) {
+            for (var sequence : workStore.drain()) {
+                NudgeSimulationTrace.append(
+                        trace,
+                        "work-drained",
+                        Map.of(
+                                "runId", payload.runId,
+                                "sequence", sequence,
+                                "pid", pid,
+                                "nodeId", context.nodeId().toString(),
+                                "jobId", jobId,
+                                "origin", origin));
+            }
+        }
+
+        NudgeSimulationTrace.append(
+                trace,
+                "pump-run-finish",
+                Map.of(
+                        "runId", payload.runId,
+                        "pid", pid,
+                        "nodeId", context.nodeId().toString(),
+                        "jobId", jobId,
+                        "origin", origin));
+    }
+}