diff --git a/crates/sdk-axum/src/api_tests.rs b/crates/sdk-axum/src/api_tests.rs index 27fdd25..c42f0c3 100644 --- a/crates/sdk-axum/src/api_tests.rs +++ b/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>; - 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") diff --git a/crates/sdk-postgres/migrations/20260928000000_scope_data_flows_by_participant_context.sql b/crates/sdk-postgres/migrations/20260928000000_scope_data_flows_by_participant_context.sql new file mode 100644 index 0000000..6f8d8c3 --- /dev/null +++ b/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); diff --git a/crates/sdk-postgres/src/data_flow/repo.rs b/crates/sdk-postgres/src/data_flow/repo.rs index a48dcae..c582bc4 100644 --- a/crates/sdk-postgres/src/data_flow/repo.rs +++ b/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> { 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)))? diff --git a/crates/sdk/src/core/db/data_flow.rs b/crates/sdk/src/core/db/data_flow.rs index d12240d..6adc8d7 100644 --- a/crates/sdk/src/core/db/data_flow.rs +++ b/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>; + /// 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<()>; } diff --git a/crates/sdk/src/core/db/data_flow/memory.rs b/crates/sdk/src/core/db/data_flow/memory.rs index 08cf944..1d4817a 100644 --- a/crates/sdk/src/core/db/data_flow/memory.rs +++ b/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); +/// 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> { - 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)) } } diff --git a/crates/sdk/src/core/db/test_suite.rs b/crates/sdk/src/core/db/test_suite.rs index bf47b5c..bf9eea5 100644 --- a/crates/sdk/src/core/db/test_suite.rs +++ b/crates/sdk/src/core/db/test_suite.rs @@ -30,6 +30,12 @@ pub trait Tester { fn begin(&self) -> impl Future; } +/// Participant context that owns the flows built by [`create_data_flow`]. +pub const PARTICIPANT_CONTEXT_ID: &str = "participant_id"; + +/// A second participant context, used to check isolation between contexts. +pub const OTHER_PARTICIPANT_CONTEXT_ID: &str = "other_participant_context_id"; + pub fn create_data_flow(id: &str) -> DataFlow { DataFlow::builder() .id(id.to_string()) @@ -40,7 +46,7 @@ pub fn create_data_flow(id: &str) -> DataFlow { .endpoint("endpoint") .build(), ) - .participant_context_id("participant_id") + .participant_context_id(PARTICIPANT_CONTEXT_ID) .counter_party_id("counter_party_id") .state(DataFlowState::Started) .labels(vec!["label1".to_string(), "label2".to_string()]) @@ -78,7 +84,11 @@ where store.create(&mut tx, &transfer).await.unwrap(); - let saved = store.fetch_by_id(&mut tx, &id).await.unwrap().unwrap(); + let saved = store + .fetch_by_id(&mut tx, PARTICIPANT_CONTEXT_ID, &id) + .await + .unwrap() + .unwrap(); assert_eq!(saved, transfer); } @@ -110,13 +120,20 @@ where store.create(&mut tx, &transfer).await.unwrap(); - let saved = store.fetch_by_id(&mut tx, &id).await.unwrap().unwrap(); + let saved = store + .fetch_by_id(&mut tx, PARTICIPANT_CONTEXT_ID, &id) + .await + .unwrap() + .unwrap(); assert_eq!(saved, transfer); tx.rollback().await.expect("Failed to rollback transaction"); let mut tx = tester.begin().await; - let result = store.fetch_by_id(&mut tx, &id).await.unwrap(); + let result = store + .fetch_by_id(&mut tx, PARTICIPANT_CONTEXT_ID, &id) + .await + .unwrap(); assert!(result.is_none()); } @@ -134,12 +151,21 @@ where store.create(&mut tx, &transfer).await.unwrap(); store.create(&mut tx, &transfer_2).await.unwrap(); - store.delete(&mut tx, &transfer_2.id).await.unwrap(); + store + .delete(&mut tx, PARTICIPANT_CONTEXT_ID, &transfer_2.id) + .await + .unwrap(); - let result = store.fetch_by_id(&mut tx, &transfer_2.id).await.unwrap(); + let result = store + .fetch_by_id(&mut tx, PARTICIPANT_CONTEXT_ID, &transfer_2.id) + .await + .unwrap(); assert!(result.is_none()); - let result = store.fetch_by_id(&mut tx, &transfer.id).await.unwrap(); + let result = store + .fetch_by_id(&mut tx, PARTICIPANT_CONTEXT_ID, &transfer.id) + .await + .unwrap(); assert!(result.is_some()); } @@ -152,7 +178,7 @@ where let id = Uuid::new_v4().to_string(); - let result = store.delete(&mut tx, &id).await; + let result = store.delete(&mut tx, PARTICIPANT_CONTEXT_ID, &id).await; assert!(matches!(result, Err(DbError::NotFound(..)))); } @@ -174,7 +200,7 @@ where store.update(&mut tx, &updated).await.unwrap(); let updated = store - .fetch_by_id(&mut tx, &transfer.id) + .fetch_by_id(&mut tx, PARTICIPANT_CONTEXT_ID, &transfer.id) .await .unwrap() .unwrap(); @@ -182,7 +208,7 @@ where assert_eq!(updated.state, DataFlowState::Suspended); let result = store - .fetch_by_id(&mut tx, &transfer.id) + .fetch_by_id(&mut tx, PARTICIPANT_CONTEXT_ID, &transfer.id) .await .unwrap() .unwrap(); @@ -203,6 +229,73 @@ where assert!(matches!(result, Err(DbError::NotFound(..)))); } +/// A flow owned by one participant context is invisible to, and cannot be updated or deleted by, +/// another participant context that knows its id. +pub async fn other_context_isolated(tester: impl Tester) +where + T: DataFlowRepo, +{ + let store = tester.store(); + let mut tx = tester.begin().await; + + let transfer = create_data_flow(&Uuid::new_v4().to_string()); + store.create(&mut tx, &transfer).await.unwrap(); + + let result = store + .fetch_by_id(&mut tx, OTHER_PARTICIPANT_CONTEXT_ID, &transfer.id) + .await + .unwrap(); + assert!(result.is_none()); + + let mut hijacked = transfer.clone(); + hijacked.participant_context_id = OTHER_PARTICIPANT_CONTEXT_ID.to_string(); + hijacked.state = DataFlowState::Terminated; + let result = store.update(&mut tx, &hijacked).await; + assert!(matches!(result, Err(DbError::NotFound(..)))); + + let result = store + .delete(&mut tx, OTHER_PARTICIPANT_CONTEXT_ID, &transfer.id) + .await; + assert!(matches!(result, Err(DbError::NotFound(..)))); + + let saved = store + .fetch_by_id(&mut tx, PARTICIPANT_CONTEXT_ID, &transfer.id) + .await + .unwrap() + .unwrap(); + assert_eq!(saved, transfer); +} + +/// Two participant contexts can each own a flow with the same id without colliding. +pub async fn same_id_in_other_context(tester: impl Tester) +where + T: DataFlowRepo, +{ + let store = tester.store(); + let mut tx = tester.begin().await; + + let id = Uuid::new_v4().to_string(); + let transfer = create_data_flow(&id); + let mut other = create_data_flow(&id); + other.participant_context_id = OTHER_PARTICIPANT_CONTEXT_ID.to_string(); + other.state = DataFlowState::Suspended; + + store.create(&mut tx, &transfer).await.unwrap(); + store.create(&mut tx, &other).await.unwrap(); + + store + .delete(&mut tx, OTHER_PARTICIPANT_CONTEXT_ID, &id) + .await + .unwrap(); + + let saved = store + .fetch_by_id(&mut tx, PARTICIPANT_CONTEXT_ID, &id) + .await + .unwrap() + .unwrap(); + assert_eq!(saved, transfer); +} + #[macro_export] macro_rules! generate_data_flow_store_tests { ($tester:ident) => { @@ -227,6 +320,14 @@ macro_rules! generate_data_flow_store_tests { data_flow_update_not_found, $crate::core::db::test_suite::update_not_found ); + test!( + data_flow_other_context_isolated, + $crate::core::db::test_suite::other_context_isolated + ); + test!( + data_flow_same_id_in_other_context, + $crate::core::db::test_suite::same_id_in_other_context + ); }; } diff --git a/crates/sdk/src/error.rs b/crates/sdk/src/error.rs index a8879c6..601ac9a 100644 --- a/crates/sdk/src/error.rs +++ b/crates/sdk/src/error.rs @@ -29,4 +29,6 @@ pub enum SdkError { Notification(#[from] reqwest::Error), #[error("Notification rejected with status {status}: {body}")] NotificationStatus { status: u16, body: String }, + #[error("Invalid control plane callback URL: {0}")] + InvalidCallbackUrl(String), } diff --git a/crates/sdk/src/sdk/internal.rs b/crates/sdk/src/sdk/internal.rs index 0d1ed56..a5d96b4 100644 --- a/crates/sdk/src/sdk/internal.rs +++ b/crates/sdk/src/sdk/internal.rs @@ -146,14 +146,14 @@ where pub async fn terminate( &self, - _ctx: &str, + ctx: &str, flow_id: &str, reason: Option, ) -> SdkResult<()> { let mut tx = self.ctx.begin().await?; let mut flow = self .repo - .fetch_by_id(&mut tx, flow_id) + .fetch_by_id(&mut tx, ctx, flow_id) .await? .ok_or_else(|| DbError::NotFound(flow_id.to_string()))?; @@ -168,14 +168,14 @@ where pub async fn started( &self, - _ctx: &str, + ctx: &str, flow_id: &str, msg: DataFlowStartedNotificationMessage, ) -> SdkResult<()> { let mut tx = self.ctx.begin().await?; let mut flow = self .repo - .fetch_by_id(&mut tx, flow_id) + .fetch_by_id(&mut tx, ctx, flow_id) .await? .ok_or_else(|| DbError::NotFound(flow_id.to_string()))?; @@ -190,11 +190,11 @@ where Ok(()) } - pub async fn completed(&self, _ctx: &str, flow_id: &str) -> SdkResult<()> { + pub async fn completed(&self, ctx: &str, flow_id: &str) -> SdkResult<()> { let mut tx = self.ctx.begin().await?; let mut flow = self .repo - .fetch_by_id(&mut tx, flow_id) + .fetch_by_id(&mut tx, ctx, flow_id) .await? .ok_or_else(|| DbError::NotFound(flow_id.to_string()))?; @@ -208,17 +208,12 @@ where Ok(()) } - pub async fn suspend( - &self, - _ctx: &str, - flow_id: &str, - reason: Option, - ) -> SdkResult<()> { + pub async fn suspend(&self, ctx: &str, flow_id: &str, reason: Option) -> SdkResult<()> { let mut tx = self.ctx.begin().await?; let mut flow = self .repo - .fetch_by_id(&mut tx, flow_id) + .fetch_by_id(&mut tx, ctx, flow_id) .await? .ok_or_else(|| DbError::NotFound(flow_id.to_string()))?; @@ -234,7 +229,7 @@ where pub async fn resume( &self, - _ctx: &str, + ctx: &str, flow_id: &str, msg: DataFlowResumeMessage, ) -> SdkResult { @@ -242,7 +237,7 @@ where let mut flow = self .repo - .fetch_by_id(&mut tx, flow_id) + .fetch_by_id(&mut tx, ctx, flow_id) .await? .ok_or_else(|| DbError::NotFound(flow_id.to_string()))?; @@ -268,14 +263,14 @@ where } pub async fn status( &self, - _ctx: &str, + ctx: &str, flow_id: &str, ) -> SdkResult { let mut tx = self.ctx.begin().await?; let flow = self .repo - .fetch_by_id(&mut tx, flow_id) + .fetch_by_id(&mut tx, ctx, flow_id) .await? .ok_or_else(|| DbError::NotFound(flow_id.to_string()))?; @@ -356,7 +351,7 @@ where async fn send_callback( &self, - _ctx: &str, + ctx: &str, flow_id: &str, operation: &str, data_address: Option, @@ -369,7 +364,7 @@ where let mut tx = self.ctx.begin().await?; let mut flow = self - .fetch_by_id(&mut tx, flow_id) + .fetch_by_id(&mut tx, ctx, flow_id) .await? .ok_or_else(|| DbError::NotFound(flow_id.to_string()))?; @@ -388,14 +383,9 @@ where .maybe_error(error) .build(); - let url = format!( - "{}/transfers/{}/dataflow/{}", - control_plane.url.trim_end_matches('/'), - flow.id, - operation - ); + let url = callback_url(&control_plane.url, &flow.id, operation)?; - let resp = self.client.post(&url).json(&msg).send().await?; + let resp = self.client.post(url).json(&msg).send().await?; if !resp.status().is_success() { let status = resp.status().as_u16(); let body = resp.text().await.unwrap_or_default(); @@ -408,15 +398,59 @@ where Ok(()) } + /// Fetches the flow `flow_id` owned by participant context `ctx`; a flow owned by another + /// participant context is reported as absent. pub async fn fetch_by_id( &self, tx: &mut C::Transaction, + ctx: &str, flow_id: &str, ) -> SdkResult> { - self.repo.fetch_by_id(tx, flow_id).await.map(Ok)? + self.repo.fetch_by_id(tx, ctx, flow_id).await.map(Ok)? } pub fn ctx(&self) -> &C { &self.ctx } } + +/// Builds `{base}/transfers/{flow_id}/dataflow/{operation}`, percent-encoding `flow_id` as a single +/// path segment. The flow id is supplied by the caller of `start`/`prepare`, so it must not be able +/// to add path segments, a query or a fragment to the control plane URL. +fn callback_url(base: &str, flow_id: &str, operation: &str) -> SdkResult { + let mut url = + reqwest::Url::parse(base).map_err(|e| SdkError::InvalidCallbackUrl(e.to_string()))?; + url.path_segments_mut() + .map_err(|_| SdkError::InvalidCallbackUrl(format!("{} cannot be a base URL", base)))? + .pop_if_empty() + .extend(["transfers", flow_id, "dataflow", operation]); + Ok(url) +} + +#[cfg(test)] +mod tests { + use super::callback_url; + + #[test] + fn callback_url_appends_segments() { + let url = callback_url("http://cp.example/api/", "flow-1", "started").unwrap(); + assert_eq!( + url.as_str(), + "http://cp.example/api/transfers/flow-1/dataflow/started" + ); + } + + #[test] + fn callback_url_encodes_flow_id_as_one_segment() { + let url = callback_url("http://cp.example/api", "../../admin?x=1#f", "started").unwrap(); + assert_eq!( + url.as_str(), + "http://cp.example/api/transfers/..%2F..%2Fadmin%3Fx=1%23f/dataflow/started" + ); + } + + #[test] + fn callback_url_rejects_invalid_base() { + assert!(callback_url("not a url", "flow-1", "started").is_err()); + } +} diff --git a/crates/sdk/src/sdk_test.rs b/crates/sdk/src/sdk_test.rs index b4158b1..4b58b55 100644 --- a/crates/sdk/src/sdk_test.rs +++ b/crates/sdk/src/sdk_test.rs @@ -423,7 +423,7 @@ mod terminate { }); repo.expect_fetch_by_id() - .returning(|_, _| Box::pin(future::ready(Ok(Some(flow()))))); + .returning(|_, _, _| Box::pin(future::ready(Ok(Some(flow()))))); repo.expect_update() .returning(|_, _| Box::pin(future::ready(Ok(())))); @@ -460,7 +460,7 @@ mod terminate { }); repo.expect_fetch_by_id() - .returning(|_, _| Box::pin(future::ready(Ok(None)))); + .returning(|_, _, _| Box::pin(future::ready(Ok(None)))); handler .expect_can_handle() @@ -505,7 +505,7 @@ mod suspend { }); repo.expect_fetch_by_id() - .returning(|_, _| Box::pin(future::ready(Ok(Some(flow()))))); + .returning(|_, _, _| Box::pin(future::ready(Ok(Some(flow()))))); repo.expect_update() .returning(|_, _| Box::pin(future::ready(Ok(())))); @@ -542,7 +542,7 @@ mod suspend { }); repo.expect_fetch_by_id() - .returning(|_, _| Box::pin(future::ready(Ok(None)))); + .returning(|_, _, _| Box::pin(future::ready(Ok(None)))); handler .expect_can_handle() @@ -594,7 +594,7 @@ mod started { }); repo.expect_fetch_by_id() - .returning(|_, _| Box::pin(future::ready(Ok(Some(flow()))))); + .returning(|_, _, _| Box::pin(future::ready(Ok(Some(flow()))))); repo.expect_update() .returning(|_, _| Box::pin(future::ready(Ok(())))); @@ -629,7 +629,7 @@ mod started { }); repo.expect_fetch_by_id() - .returning(|_, _| Box::pin(future::ready(Ok(None)))); + .returning(|_, _, _| Box::pin(future::ready(Ok(None)))); handler.expect_on_started().times(0); @@ -672,7 +672,7 @@ mod completed { }); repo.expect_fetch_by_id() - .returning(|_, _| Box::pin(future::ready(Ok(Some(flow()))))); + .returning(|_, _, _| Box::pin(future::ready(Ok(Some(flow()))))); repo.expect_update() .returning(|_, _| Box::pin(future::ready(Ok(())))); @@ -705,7 +705,7 @@ mod completed { }); repo.expect_fetch_by_id() - .returning(|_, _| Box::pin(future::ready(Ok(None)))); + .returning(|_, _, _| Box::pin(future::ready(Ok(None)))); handler.expect_on_completed().times(0); @@ -756,7 +756,7 @@ mod resume { Box::pin(future::ready(Ok(tx))) }); - repo.expect_fetch_by_id().returning(|_, _| { + repo.expect_fetch_by_id().returning(|_, _, _| { let mut f = flow(); f.state = DataFlowState::Suspended; Box::pin(future::ready(Ok(Some(f)))) @@ -799,7 +799,7 @@ mod resume { }); repo.expect_fetch_by_id() - .returning(|_, _| Box::pin(future::ready(Ok(None)))); + .returning(|_, _, _| Box::pin(future::ready(Ok(None)))); handler.expect_on_resume().times(0); @@ -840,7 +840,7 @@ mod status { }); repo.expect_fetch_by_id() - .returning(|_, _| Box::pin(future::ready(Ok(Some(flow()))))); + .returning(|_, _, _| Box::pin(future::ready(Ok(Some(flow()))))); let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) @@ -867,7 +867,7 @@ mod status { }); repo.expect_fetch_by_id() - .returning(|_, _| Box::pin(future::ready(Ok(None)))); + .returning(|_, _, _| Box::pin(future::ready(Ok(None)))); let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) @@ -932,7 +932,7 @@ mod notify { let control_plane_id = f.control_plane_id.clone(); repo.expect_fetch_by_id() - .returning(move |_, _| Box::pin(future::ready(Ok(Some(f.clone()))))); + .returning(move |_, _, _| Box::pin(future::ready(Ok(Some(f.clone()))))); repo.expect_update() .returning(|_, _| Box::pin(future::ready(Ok(())))); @@ -1061,7 +1061,7 @@ mod notify { }); repo.expect_fetch_by_id() - .returning(|_, _| Box::pin(future::ready(Ok(None)))); + .returning(|_, _, _| Box::pin(future::ready(Ok(None)))); let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) @@ -1079,6 +1079,98 @@ mod notify { } } +/// Operations on an existing flow are scoped to the calling participant context: a caller that +/// knows another participant context's flow id must get `NotFound`, and the handler must not run. +mod isolation { + use std::future; + + use crate::{ + core::{ + db::{control_plane::MockControlPlaneRepo, tx::MockTransaction}, + error::DbError, + model::messages::{DataFlowResumeMessage, DataFlowStartedNotificationMessage}, + }, + error::SdkError, + sdk::DataPlaneSdk, + sdk_test::{context, flow}, + }; + + const OWNER: &str = "participant-1"; + const INTRUDER: &str = "participant-2"; + + /// The repo only returns the flow for its owning participant context. The handler has no + /// expectations, so any handler callback made for the intruder fails the test. + fn sdk() -> DataPlaneSdk { + let (mut ctx, mut repo, handler) = context(); + + ctx.expect_begin().returning(|| { + let mut tx = MockTransaction::new(); + tx.expect_commit() + .returning(|| Box::pin(future::ready(Ok(())))); + tx.expect_rollback() + .returning(|| Box::pin(future::ready(Ok(())))); + Box::pin(future::ready(Ok(tx))) + }); + + repo.expect_fetch_by_id().returning(|_, pc, _| { + let found = (pc == OWNER).then(flow); + Box::pin(future::ready(Ok(found))) + }); + + DataPlaneSdk::builder(ctx) + .with_repo(repo) + .with_control_plane_repo(MockControlPlaneRepo::new()) + .with_handler(handler) + .build() + .unwrap() + } + + fn assert_not_found(result: Result) { + assert!( + matches!(result, Err(SdkError::Repo(DbError::NotFound(..)))), + "expected NotFound, got {:?}", + result + ); + } + + #[tokio::test] + async fn status_of_other_context_flow_is_not_found() { + assert_not_found(sdk().status(INTRUDER, "flow-id").await); + } + + #[tokio::test] + async fn terminate_of_other_context_flow_is_not_found() { + assert_not_found(sdk().terminate(INTRUDER, "flow-id", None).await); + } + + #[tokio::test] + async fn suspend_of_other_context_flow_is_not_found() { + assert_not_found(sdk().suspend(INTRUDER, "flow-id", None).await); + } + + #[tokio::test] + async fn completed_of_other_context_flow_is_not_found() { + assert_not_found(sdk().completed(INTRUDER, "flow-id").await); + } + + #[tokio::test] + async fn started_of_other_context_flow_is_not_found() { + let msg = DataFlowStartedNotificationMessage::builder().build(); + assert_not_found(sdk().started(INTRUDER, "flow-id", msg).await); + } + + #[tokio::test] + async fn resume_of_other_context_flow_is_not_found() { + let msg = DataFlowResumeMessage::builder().build(); + assert_not_found(sdk().resume(INTRUDER, "flow-id", msg).await); + } + + #[tokio::test] + async fn status_of_own_flow_is_found() { + assert!(sdk().status(OWNER, "flow-id").await.is_ok()); + } +} + fn start_message() -> DataFlowStartMessage { DataFlowStartMessage::builder() .data_flow_id("process-id")