lib

commit 54f8da6489e37fac37abc21837364cb6eb8f648c
parent 920c971ce16f4b2e1364894c3a085c9ec1a40329
Author: triesap <tyson@radroots.org>
Date:   Tue, 28 Jul 2026 16:50:38 +0000

event_store: close projection row effects

- require one-row projection mutations
- preserve cursor and retraction diagnostics
- cover failed database effects directly
- update the governed production source hash

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

2 files changed, 43 insertions(+), 30 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 = "72d28049b8e28b48bb268318c458928e254369d23502b934bcc98bd38bbea390"
+sha256 = "50bfde11cd2d467410e767415ef06f047d5b3d434d53a71d0a9a83b89d9a8cae"
 
 [[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
@@ -247,11 +247,10 @@ async fn ensure_projection_cursor(
             )
             .execute(&mut *connection)
             .await?;
-            if deleted.rows_affected() != 1 {
-                return Err(projection_drift(
-                    "generation reset did not delete exactly one projection cursor",
-                ));
-            }
+            require_single_projection_row(
+                deleted.rows_affected(),
+                "generation reset did not delete exactly one projection cursor".to_owned(),
+            )?;
         }
     }
 
@@ -272,11 +271,10 @@ async fn ensure_projection_cursor(
         .bind(floor)
         .execute(&mut *connection)
         .await?;
-        if inserted.rows_affected() != 1 {
-            return Err(projection_drift(
-                "projection cursor initialization did not insert one row",
-            ));
-        }
+        require_single_projection_row(
+            inserted.rows_affected(),
+            "projection cursor initialization did not insert one row".to_owned(),
+        )?;
     }
 
     let row = sqlx::query(
@@ -353,13 +351,14 @@ async fn advance_projection_cursor(
     .bind(expected.projected_row_count)
     .execute(&mut *connection)
     .await?;
-    if updated.rows_affected() != 1 {
-        return Err(projection_drift(format!(
+    require_single_projection_row(
+        updated.rows_affected(),
+        format!(
             "projection cursor compare-and-swap expected sequence {} and row count {}",
             expected.feed_cursor.last_transition_seq(),
             expected.projected_row_count,
-        )));
-    }
+        ),
+    )?;
     Ok(FoodAvailabilityProjectionCursorState {
         feed_cursor: next,
         projected_row_count: next_projected_row_count,
@@ -394,11 +393,10 @@ async fn apply_transition(
             .bind(retracted.event_id().as_str())
             .execute(&mut *connection)
             .await?;
-            if deleted.rows_affected() != 1 {
-                return Err(projection_drift(
-                    "pending FoodAvailability retraction did not delete one row",
-                ));
-            }
+            require_single_projection_row(
+                deleted.rows_affected(),
+                "pending FoodAvailability retraction did not delete one row".to_owned(),
+            )?;
             projected_row_delta = -1;
         }
         (Some(existing), None) if visible_event_id == Some(existing) => {
@@ -504,11 +502,10 @@ async fn persist_projection(
     .bind(projection.source_transition_seq())
     .execute(&mut *connection)
     .await?;
-    if inserted.rows_affected() != 1 {
-        return Err(projection_drift(
-            "FoodAvailability projection insert did not affect one row",
-        ));
-    }
+    require_single_projection_row(
+        inserted.rows_affected(),
+        "FoodAvailability projection insert did not affect one row".to_owned(),
+    )?;
     for image in projection.images() {
         persist_image(connection, projection, image).await?;
     }
@@ -539,11 +536,10 @@ async fn persist_image(
     .bind(diagnostics_json)
     .execute(&mut *connection)
     .await?;
-    if inserted.rows_affected() != 1 {
-        return Err(projection_drift(
-            "FoodAvailability image insert did not affect one row",
-        ));
-    }
+    require_single_projection_row(
+        inserted.rows_affected(),
+        "FoodAvailability image insert did not affect one row".to_owned(),
+    )?;
     Ok(())
 }
 
@@ -953,6 +949,16 @@ fn projection_drift(reason: impl Into<String>) -> RadrootsEventStoreError {
     }
 }
 
+fn require_single_projection_row(
+    rows_affected: u64,
+    reason: String,
+) -> Result<(), RadrootsEventStoreError> {
+    if rows_affected != 1 {
+        return Err(projection_drift(reason));
+    }
+    Ok(())
+}
+
 fn validate_projected_row_count(value: i64) -> Result<(), RadrootsEventStoreError> {
     if value < 0 {
         return Err(projection_drift(format!(
@@ -1171,5 +1177,12 @@ mod tests {
                 value,
             }) if value == i64::MAX as u64 + 1
         ));
+        require_single_projection_row(1, "fixture mutation failed".to_owned())
+            .expect("one affected projection row");
+        assert!(matches!(
+            require_single_projection_row(0, "fixture mutation failed".to_owned()),
+            Err(RadrootsEventStoreError::FoodAvailabilityProjectionDrift { reason })
+                if reason == "fixture mutation failed"
+        ));
     }
 }