Wave-2 review gate: scheduler fire expires resolution fix (D-27, ADR-020 §2), pressure-test lock quality (D-28), lineage diff re-verified clean (task review-wave-2)

alkstore-sqlite/src/substrate/PROVENANCE.md

@@ -82,8 +82,10 @@ only, not committed — the wave-2 review's primary input), and every
 divergence it surfaced is either a register entry below (D-21..D-26,
 appended; D-12 annotated in place) or a fixed omission (the
 lineage-less placeholder test in `ops.rs`'s pressure suite, removed).
-No unregistered divergence remains. Cherry-picks: empty at scaffold,
-append-only forever.
+No unregistered divergence remains. The wave-2 review gate
+(`review-wave-2`) re-ran the full-coverage lineage diff (clean —
+everything register-addressable) and appended D-27/D-28 for its two
+fixes. Cherry-picks: empty at scaffold, append-only forever.
 
 | ID | Category | What | Where | Why | Lineage |
 |----|----------|------|-------|-----|---------|
@@ -113,6 +115,8 @@ append-only forever.
 | D-24 | hygiene | `real_to_i64` boundary error text re-owned ("honker:" prefix → "alkstore:"; the lineage's better-sqlite3-facing sentence elided per D-15) | ops.rs (`real_to_i64`) | D-19's re-owning pattern at the error-string layer | ours |
 | D-25 | hygiene | watcher thread name "honker-update-poll" → "alkstore-update-poll" | watcher.rs (spawn path) | D-19's re-owning pattern at the thread-name layer | ours |
 | D-26 | hygiene | `volume_serial_number as u64` → `u64::from(volume_serial_number)` in `VolumeIdentity` (clippy lint conformance, zero behavior change) | watcher.rs (`stat_identity` region) | family clippy `-D warnings` gate | ours |
+| D-27 | re-derivation | wave-2 review fix — `scheduler_tick`'s fire enqueue resolved the schedule row's relative `expires_s` into the absolute row `expires_at` (`now + s` at the fire transaction, one clock read) instead of passing the relative value through; regression test pins a fired job's `expires_at = fire-instant + s` and its claimability. The pre-fix pass-through would have produced rows expired in 1970 under any `expires_s` schedule | `queue_ops.rs` (`scheduler_tick`, fire-expiry resolution + test) | ADR-020 §2 (relative `expires` → resolved absolute row value, resolved at enqueue against the single clock) — D-12's landed state carried the lineage's `expires_s` field shape without carrying the lineage's enqueue-side resolution; wave-2 review (task review-wave-2) re-derivation spot-check found it | ours |
+| D-28 | hygiene | wave-2 review fix — the cross-mechanism pressure test's lock-churn branch exercised a per-producer lock name (literal `"p{producer}"` had no interpolation, so all producers contended on one name) with its result swallowed via `.unwrap_or(0)`; now interpolates, releases periodically, propagates errors | `ops.rs` (pressure test, lock branch) | test-quality only (no lineage counterpart, no library code touched); `fork-provenance-and-floor`'s ported-surface pressure intent | ours |
 
 ### Cherry-picks
 

alkstore-sqlite/src/substrate/ops.rs

@@ -567,13 +567,23 @@ mod pressure_tests {
                         )
                         .unwrap();
                     if seq % 13 == 0 {
-                        let _: i64 = conn
+                        let lock_name = format!("pressure-l-{producer}");
+                        let locked: i64 = conn
                             .query_row(
-                                "SELECT alkstore_lock_acquire('pressure-l', 'p{producer}', 30)",
-                                [],
+                                "SELECT alkstore_lock_acquire(?1, ?2, 30)",
+                                rusqlite::params![lock_name, format!("p{producer}")],
                                 |r| r.get(0),
                             )
-                            .unwrap_or(0);
+                            .unwrap();
+                        if locked == 1 && seq % 39 == 0 {
+                            let _: i64 = conn
+                                .query_row(
+                                    "SELECT alkstore_lock_release(?1, ?2)",
+                                    rusqlite::params![lock_name, format!("p{producer}")],
+                                    |r| r.get(0),
+                                )
+                                .unwrap();
+                        }
                     }
                     if seq % 11 == 0 {
                         std::thread::sleep(Duration::from_millis(1));

alkstore-sqlite/src/substrate/queue_ops.rs

@@ -373,17 +373,19 @@ pub(crate) fn scheduler_tick(conn: &Connection, now: i64) -> rusqlite::Result<St
             backoff_base_s,
             dead_letter_retention_s,
         };
+        let expires_at = expires_s.map(|s| now + s);
         let mut fires: i64 = 0;
         while next_fire_at <= now {
             if fires >= SCHEDULER_MAX_CATCHUP_FIRES {
                 next_fire_at = skip_forward(next_fire_at, now, interval);
                 break;
             }
-            let job_id = enqueue(conn, &queue, &payload, now, priority, expires_s, stamps)?;
+            let fire_at = next_fire_at;
+            let job_id = enqueue(conn, &queue, &payload, now, priority, expires_at, stamps)?;
             out.push(json!({
                 "name": name,
                 "queue": queue,
-                "fire_at": next_fire_at,
+                "fire_at": fire_at,
                 "job_id": job_id,
             }));
             fires += 1;
@@ -1895,6 +1897,52 @@ mod queue_tests {
         assert_eq!(jobs, 64, "the skipped boundaries never enqueue");
     }
 
+    #[test]
+    fn tick_resolves_the_fires_expires_sticker_to_an_absolute_row_value() {
+        let conn = db();
+        let now: i64 = now_unix(&conn).unwrap();
+        scheduler_register(
+            &conn,
+            "t",
+            "jobs",
+            "@every 60s",
+            "\"work\"",
+            fire(0, Some(30)),
+            stamps_cfg(300, 3),
+        )
+        .unwrap();
+        conn.execute(
+            "UPDATE __alkstore_scheduler_tasks SET next_fire_at = ?1 - 5 WHERE name = 't'",
+            [now],
+        )
+        .unwrap();
+        scheduler_tick(&conn, now).unwrap();
+
+        let expires_at: Option<i64> = conn
+            .query_row(
+                "SELECT expires_at FROM __alkstore_live WHERE queue = 'jobs' LIMIT 1",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        assert_eq!(
+            expires_at,
+            Some(now + 30),
+            "the fire's relative expiry resolves against the fire instant \
+             (ADR-020 §2), not passed through as a raw unix value"
+        );
+
+        let unexpired: i64 = conn
+            .query_row(
+                "SELECT count(*) FROM __alkstore_live
+                 WHERE queue = 'jobs' AND (expires_at IS NULL OR expires_at > unixepoch())",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        assert_eq!(unexpired, 1, "the fired job must still be claimable");
+    }
+
     #[test]
     fn tick_ignores_disabled_and_future_tasks() {
         let conn = db();