lib

commit 7987c844ccc965cb77b5c40b69edde83791646e5
parent 54f8da6489e37fac37abc21837364cb6eb8f648c
Author: triesap <tyson@radroots.org>
Date:   Tue, 28 Jul 2026 16:59:06 +0000

event_store: close projection cursor scalars

- validate reset residue directly
- bind the governed feed version
- reject stationary cursor row drift
- cover every scalar outcome

Diffstat:
M
contracts/event_store_production_sources.toml
 | 
2
+-
M
crates/event_store/src/store/food_availability_projection_v1.rs
 | 
85
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++----------------

2 files changed, 69 insertions(+), 18 deletions(-)
diff --git a/contracts/event_store_production_sources.toml b/contracts/event_store_production_sources.toml
@@ -79,7 +79,7 @@ sha256 = "93129c02bf55858aa52e538ef50bf920d35719696183d339443e5c4060d2d4a3"
 
 [[sources]]
 path = "crates/event_store/src/store/food_availability_projection_v1.rs"
-sha256 = "50bfde11cd2d467410e767415ef06f047d5b3d434d53a71d0a9a83b89d9a8cae"
+sha256 = "2a41b873a628b58969638a62fb915c8b8025875adc973679ae971480e6d24c37"
 
 [[sources]]
 path = "crates/event_store/src/store/post_core_extension_capabilities.rs"
diff --git a/crates/event_store/src/store/food_availability_projection_v1.rs b/crates/event_store/src/store/food_availability_projection_v1.rs
@@ -169,12 +169,7 @@ pub(crate) async fn reset_and_replay_food_availability_from_raw_v1(
     )
     .fetch_one(&mut *connection)
     .await?;
-    if residual_count != 0 {
-        return Err(RadrootsEventStoreError::RawSourceRebuildStateDrift {
-            kind: RadrootsEventStoreRawSourceRebuildDriftV1::DerivedProductStateAuthority,
-            detail: format!("FoodAvailability reset left {residual_count} stale derived row(s)"),
-        });
-    }
+    validate_projection_reset_residual_count(residual_count)?;
     apply_pending_food_availability_transitions_v1(connection).await
 }
 
@@ -222,11 +217,7 @@ async fn ensure_projection_cursor(
     )?;
     let floor: i64 = source.try_get("transition_floor_seq")?;
     let feed_version: i64 = source.try_get("addressable_feed_version")?;
-    if feed_version != i64::from(RADROOTS_ADDRESSABLE_TRANSITION_FEED_VERSION_V1) {
-        return Err(projection_drift(format!(
-            "addressable feed version is {feed_version}"
-        )));
-    }
+    validate_projection_feed_version(feed_version)?;
 
     let existing = sqlx::query(
         "SELECT source_generation, feed_version, projection_version, scope_fingerprint, hook_manifest_sha256, last_transition_seq, projected_row_count FROM radroots_event_store_food_availability_cursor WHERE singleton = 1",
@@ -333,12 +324,11 @@ async fn advance_projection_cursor(
         .checked_add(projected_row_delta)
         .ok_or_else(|| projection_drift("projection row count overflowed"))?;
     validate_projected_row_count(next_projected_row_count)?;
-    if next.last_transition_seq() == expected.feed_cursor.last_transition_seq() {
-        if projected_row_delta != 0 {
-            return Err(projection_drift(
-                "projection row count changed without a feed transition",
-            ));
-        }
+    if projection_cursor_is_stationary(
+        expected.feed_cursor.last_transition_seq(),
+        next.last_transition_seq(),
+        projected_row_delta,
+    )? {
         return Ok(expected.clone());
     }
     let updated = sqlx::query(
@@ -959,6 +949,43 @@ fn require_single_projection_row(
     Ok(())
 }
 
+fn validate_projection_reset_residual_count(
+    residual_count: i64,
+) -> Result<(), RadrootsEventStoreError> {
+    if residual_count != 0 {
+        return Err(RadrootsEventStoreError::RawSourceRebuildStateDrift {
+            kind: RadrootsEventStoreRawSourceRebuildDriftV1::DerivedProductStateAuthority,
+            detail: format!("FoodAvailability reset left {residual_count} stale derived row(s)"),
+        });
+    }
+    Ok(())
+}
+
+fn validate_projection_feed_version(feed_version: i64) -> Result<(), RadrootsEventStoreError> {
+    if feed_version != i64::from(RADROOTS_ADDRESSABLE_TRANSITION_FEED_VERSION_V1) {
+        return Err(projection_drift(format!(
+            "addressable feed version is {feed_version}"
+        )));
+    }
+    Ok(())
+}
+
+fn projection_cursor_is_stationary(
+    current_transition_seq: i64,
+    next_transition_seq: i64,
+    projected_row_delta: i64,
+) -> Result<bool, RadrootsEventStoreError> {
+    if next_transition_seq != current_transition_seq {
+        return Ok(false);
+    }
+    if projected_row_delta != 0 {
+        return Err(projection_drift(
+            "projection row count changed without a feed transition",
+        ));
+    }
+    Ok(true)
+}
+
 fn validate_projected_row_count(value: i64) -> Result<(), RadrootsEventStoreError> {
     if value < 0 {
         return Err(projection_drift(format!(
@@ -1184,5 +1211,29 @@ mod tests {
             Err(RadrootsEventStoreError::FoodAvailabilityProjectionDrift { reason })
                 if reason == "fixture mutation failed"
         ));
+        validate_projection_reset_residual_count(0).expect("empty reset state");
+        assert!(matches!(
+            validate_projection_reset_residual_count(2),
+            Err(RadrootsEventStoreError::RawSourceRebuildStateDrift {
+                kind: RadrootsEventStoreRawSourceRebuildDriftV1::DerivedProductStateAuthority,
+                detail,
+            }) if detail == "FoodAvailability reset left 2 stale derived row(s)"
+        ));
+        validate_projection_feed_version(i64::from(
+            RADROOTS_ADDRESSABLE_TRANSITION_FEED_VERSION_V1,
+        ))
+        .expect("governed feed version");
+        assert!(matches!(
+            validate_projection_feed_version(2),
+            Err(RadrootsEventStoreError::FoodAvailabilityProjectionDrift { reason })
+                if reason == "addressable feed version is 2"
+        ));
+        assert!(projection_cursor_is_stationary(7, 7, 0).expect("stationary projection cursor"));
+        assert!(!projection_cursor_is_stationary(7, 8, 1).expect("advancing projection cursor"));
+        assert!(matches!(
+            projection_cursor_is_stationary(7, 7, 1),
+            Err(RadrootsEventStoreError::FoodAvailabilityProjectionDrift { reason })
+                if reason == "projection row count changed without a feed transition"
+        ));
     }
 }