feat: hardening sdk on participant_context

crates/sdk-axum/src/api_tests.rs

@@ -68,10 +68,11 @@ mock! {
         async fn fetch_by_id(
             &self,
             tx: &mut MockTx,
+            participant_context_id: &str,
             flow_id: &str,
         ) -> DbResult<Option<DataFlow>>;
 
-        async fn delete(&self, tx: &mut MockTx, flow_id: &str) -> DbResult<()>;
+        async fn delete(&self, tx: &mut MockTx, participant_context_id: &str, flow_id: &str) -> DbResult<()>;
 
     }
 }
@@ -300,7 +301,7 @@ mod terminate {
             Ok(tx)
         });
 
-        repo.expect_fetch_by_id().returning(|_, _| {
+        repo.expect_fetch_by_id().returning(|_, _, _| {
             Ok(Some(
                 DataFlow::builder()
                     .id("example-flow-id")
@@ -381,7 +382,7 @@ mod suspend {
             Ok(tx)
         });
 
-        repo.expect_fetch_by_id().returning(|_, _| {
+        repo.expect_fetch_by_id().returning(|_, _, _| {
             Ok(Some(
                 DataFlow::builder()
                     .id("example-flow-id")
@@ -538,7 +539,7 @@ mod started {
             Ok(tx)
         });
 
-        repo.expect_fetch_by_id().returning(|_, _| {
+        repo.expect_fetch_by_id().returning(|_, _, _| {
             Ok(Some(
                 DataFlow::builder()
                     .id("example-flow-id")
@@ -608,7 +609,7 @@ mod completed {
             Ok(tx)
         });
 
-        repo.expect_fetch_by_id().returning(|_, _| {
+        repo.expect_fetch_by_id().returning(|_, _, _| {
             Ok(Some(
                 DataFlow::builder()
                     .id("example-flow-id")
@@ -676,7 +677,7 @@ mod flow_status {
             Ok(tx)
         });
 
-        repo.expect_fetch_by_id().returning(|_, _| {
+        repo.expect_fetch_by_id().returning(|_, _, _| {
             Ok(Some(
                 DataFlow::builder()
                     .id("example-flow-id")
@@ -752,7 +753,7 @@ mod resume {
             Ok(tx)
         });
 
-        repo.expect_fetch_by_id().returning(|_, _| {
+        repo.expect_fetch_by_id().returning(|_, _, _| {
             Ok(Some(
                 DataFlow::builder()
                     .id("example-flow-id")

crates/sdk-postgres/migrations/20260928000000_scope_data_flows_by_participant_context.sql

@@ -0,0 +1,4 @@
+-- Flow ids are unique per participant context, not globally. Keying the table on the pair keeps
+-- one participant context from colliding with, or probing for, another context's flow ids.
+ALTER TABLE data_flows DROP CONSTRAINT IF EXISTS data_flows_pkey;
+ALTER TABLE data_flows ADD PRIMARY KEY (participant_context_id, id);

crates/sdk-postgres/src/data_flow/repo.rs

@@ -69,27 +69,35 @@ impl DataFlowRepo for PgDataFlowRepo {
     async fn fetch_by_id(
         &self,
         tx: &mut Self::Transaction,
+        participant_context_id: &str,
         flow_id: &str,
     ) -> DbResult<Option<DataFlow>> {
         Ok(sqlx::query_as::<_, DbDataFlow>(
             r#"
-            SELECT * FROM data_flows where id = $1
+            SELECT * FROM data_flows WHERE id = $1 AND participant_context_id = $2
             "#,
         )
         .bind(flow_id)
+        .bind(participant_context_id)
         .fetch_optional(&mut *tx.0)
         .await
         .map_err(|err| DbError::Generic(Box::new(err)))?
         .map(|flow| flow.into()))
     }
 
-    async fn delete(&self, tx: &mut Self::Transaction, flow_id: &str) -> DbResult<()> {
+    async fn delete(
+        &self,
+        tx: &mut Self::Transaction,
+        participant_context_id: &str,
+        flow_id: &str,
+    ) -> DbResult<()> {
         let rows = sqlx::query(
             r#"
-            DELETE FROM data_flows where id = $1
+            DELETE FROM data_flows WHERE id = $1 AND participant_context_id = $2
             "#,
         )
         .bind(flow_id)
+        .bind(participant_context_id)
         .execute(&mut *tx.0)
         .await
         .map_err(|err| DbError::Generic(Box::new(err)))?
@@ -109,11 +117,12 @@ impl DataFlowRepo for PgDataFlowRepo {
         let rows = sqlx::query(
             r#"
             UPDATE data_flows SET state=$1
-            WHERE id = $2
+            WHERE id = $2 AND participant_context_id = $3
             "#,
         )
         .bind(DataFlowState::from(flow.state.clone()))
         .bind(&flow.id)
+        .bind(&flow.participant_context_id)
         .execute(&mut *tx.0)
         .await
         .map_err(|err| DbError::Generic(Box::new(err)))?

crates/sdk/src/core/db/data_flow.rs

@@ -21,15 +21,27 @@ use crate::core::db::tx::MockTransaction;
 pub trait DataFlowRepo: Send + Sync {
     type Transaction;
 
+    /// Creates a flow. Flow ids are unique per participant context, not globally: two participant
+    /// contexts may each own a flow with the same id.
     async fn create(&self, tx: &mut Self::Transaction, flow: &DataFlow) -> DbResult<()>;
 
+    /// Fetches the flow `flow_id` owned by `participant_context_id`. A flow owned by another
+    /// participant context is reported as absent (`Ok(None)`), never returned.
     async fn fetch_by_id(
         &self,
         tx: &mut Self::Transaction,
+        participant_context_id: &str,
         flow_id: &str,
     ) -> DbResult<Option<DataFlow>>;
 
+    /// Updates the flow identified by (`flow.participant_context_id`, `flow.id`).
     async fn update(&self, tx: &mut Self::Transaction, flow: &DataFlow) -> DbResult<()>;
 
-    async fn delete(&self, tx: &mut Self::Transaction, flow_id: &str) -> DbResult<()>;
+    /// Deletes the flow `flow_id` owned by `participant_context_id`.
+    async fn delete(
+        &self,
+        tx: &mut Self::Transaction,
+        participant_context_id: &str,
+        flow_id: &str,
+    ) -> DbResult<()>;
 }

crates/sdk/src/core/db/data_flow/memory.rs

@@ -1,35 +1,78 @@
 use crate::core::{
     db::memory::{MemoryRepo, MemoryTransaction},
-    error::DbResult,
+    error::{DbError, DbResult},
     model::data_flow::DataFlow,
 };
 
 use super::DataFlowRepo;
 
+/// In-memory [`DataFlowRepo`]. Entries are keyed by (participant context id, flow id), so flows
+/// of different participant contexts never collide or see each other.
 #[derive(Default, Clone)]
 pub struct MemoryDataFlowRepo(MemoryRepo<DataFlow>);
 
+/// Builds the storage key for a flow. The participant context id is length-prefixed so the
+/// encoding is unambiguous whatever characters either id contains.
+fn key(participant_context_id: &str, flow_id: &str) -> String {
+    format!(
+        "{}:{}{}",
+        participant_context_id.len(),
+        participant_context_id,
+        flow_id
+    )
+}
+
+/// Rewrites errors from the generic store so they name the flow id rather than the internal key.
+fn map_err(err: DbError, flow_id: &str) -> DbError {
+    match err {
+        DbError::AlreadyExists(_) => {
+            DbError::AlreadyExists(format!("Data flow with id {} already exists", flow_id))
+        }
+        DbError::NotFound(_) => {
+            DbError::NotFound(format!("Data flow with id {} not found", flow_id))
+        }
+        other => other,
+    }
+}
+
 #[async_trait::async_trait]
 impl DataFlowRepo for MemoryDataFlowRepo {
     type Transaction = MemoryTransaction;
     async fn create(&self, _tx: &mut Self::Transaction, flow: &DataFlow) -> DbResult<()> {
-        self.0.create(&flow.id, flow).await
+        self.0
+            .create(&key(&flow.participant_context_id, &flow.id), flow)
+            .await
+            .map_err(|e| map_err(e, &flow.id))
     }
 
     async fn fetch_by_id(
         &self,
         _tx: &mut Self::Transaction,
+        participant_context_id: &str,
         flow_id: &str,
     ) -> DbResult<Option<DataFlow>> {
-        self.0.fetch_by_id(flow_id).await
+        self.0
+            .fetch_by_id(&key(participant_context_id, flow_id))
+            .await
     }
 
     async fn update(&self, _tx: &mut Self::Transaction, flow: &DataFlow) -> DbResult<()> {
-        self.0.update(&flow.id, flow).await
+        self.0
+            .update(&key(&flow.participant_context_id, &flow.id), flow)
+            .await
+            .map_err(|e| map_err(e, &flow.id))
     }
 
-    async fn delete(&self, _tx: &mut Self::Transaction, flow_id: &str) -> DbResult<()> {
-        self.0.delete(flow_id).await
+    async fn delete(
+        &self,
+        _tx: &mut Self::Transaction,
+        participant_context_id: &str,
+        flow_id: &str,
+    ) -> DbResult<()> {
+        self.0
+            .delete(&key(participant_context_id, flow_id))
+            .await
+            .map_err(|e| map_err(e, flow_id))
     }
 }