diff --git a/backend/src/api/flows.rs b/backend/src/api/flows.rs index b383361b..dd563de5 100644 --- a/backend/src/api/flows.rs +++ b/backend/src/api/flows.rs @@ -295,9 +295,16 @@ fn prepare_flow(flow: &mut Flow) { post, path = "/api/flows", tag = "flows", + description = "Creates a flow under the `id` supplied in the body, so a caller can \ + pre-generate an id and then start the flow by it. `id` is a required \ + field: to have the server assign one instead, send the nil uuid \ + (`00000000-0000-0000-0000-000000000000`) and read the assigned id from \ + the `flow.id` of the response. Reusing the id of an existing flow is a \ + 409; use `POST /api/flows/{id}` to update that flow instead.", request_body = Flow, responses( (status = 201, description = "Flow created", body = FlowResponse), + (status = 409, description = "A flow with the supplied id already exists", body = ErrorResponse), (status = 500, description = "Internal server error", body = ErrorResponse) ) )] @@ -318,8 +325,13 @@ pub async fn create_flow( info!("Received create flow request: name='{}'", flow.name); debug!("Create flow request body: {:?}", flow); - // Assign a new ID to avoid collisions with imported flows - flow.id = FlowId::new_v4(); + // Honour the client-supplied id. The schema requires it, so silently replacing + // it left callers unable to POST a flow and then start it by the id they chose. + // `id` cannot be omitted — it is required, so leaving it out is a 422 — which is + // why the nil uuid is the documented way to ask the server to assign one. + if flow.id.is_nil() { + flow.id = FlowId::new_v4(); + } // Clear runtime state flow.running = false; @@ -337,14 +349,30 @@ pub async fn create_flow( info!("Creating flow: {} ({})", flow.name, flow.id); - if let Err(e) = state.upsert_flow(flow.clone()).await { - return Err(( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ErrorResponse::with_details( - "Failed to save flow", - e.to_string(), - )), - )); + // Import and copy in the frontend already regenerate ids client-side + // (`regenerate_flow_ids`), so a clash here is a genuine one the caller needs to + // know about rather than something to paper over. The id is claimed inside the + // same write lock that checks it, so two concurrent creates supplying the same + // id cannot both pass and overwrite one another. + match state.insert_flow_if_absent(flow.clone()).await { + Ok(true) => {} + Ok(false) => { + return Err(( + StatusCode::CONFLICT, + Json(ErrorResponse::new( + "A flow with this id already exists; use POST /api/flows/{id} to update it", + )), + )); + } + Err(e) => { + return Err(( + StatusCode::INTERNAL_SERVER_ERROR, + Json(ErrorResponse::with_details( + "Failed to save flow", + e.to_string(), + )), + )); + } } Ok((StatusCode::CREATED, Json(FlowResponse { flow }))) diff --git a/backend/src/state.rs b/backend/src/state.rs index 7b2c79a5..27f854f1 100644 --- a/backend/src/state.rs +++ b/backend/src/state.rs @@ -647,28 +647,51 @@ impl AppState { /// Add or update a flow and persist to storage. pub async fn upsert_flow(&self, mut flow: Flow) -> anyhow::Result<()> { - let is_new = { - let flows = self.inner.flows.read().await; - !flows.contains_key(&flow.id) - }; - // Filter out transient (persist: false) block properties before this // flow definition reaches either the in-memory map or disk. Without // this, an explicit save from the frontend would re-engage transient // state (e.g. PFL/AFL solo) on the next restart. self.strip_transient_properties(&mut flow).await; - // Update in-memory state + // Update in-memory state. `insert` reports whether the id was already + // taken, so newness is decided under the same lock that writes it. + let is_new = { + let mut flows = self.inner.flows.write().await; + flows.insert(flow.id, flow.clone()).is_none() + }; + + self.persist_flow(&flow, is_new).await + } + + /// Insert a flow only if its id is not already taken. + /// + /// Returns `Ok(false)`, leaving the stored flow untouched, when a flow with + /// the same id already exists. The check and the insert happen under a + /// single write lock, so two concurrent creates supplying the same id + /// cannot both succeed and overwrite one another. + pub async fn insert_flow_if_absent(&self, mut flow: Flow) -> anyhow::Result { + self.strip_transient_properties(&mut flow).await; + { let mut flows = self.inner.flows.write().await; + if flows.contains_key(&flow.id) { + return Ok(false); + } flows.insert(flow.id, flow.clone()); } + self.persist_flow(&flow, true).await?; + Ok(true) + } + + /// Persist a flow that is already in the in-memory map, update its PTP + /// registration, and broadcast the created/updated event. + async fn persist_flow(&self, flow: &Flow, is_new: bool) -> anyhow::Result<()> { // Persist to storage (skip ephemeral flows) if flow.properties.ephemeral { // Remove any previously persisted copy so it doesn't reappear on restart let _ = self.inner.storage.delete_flow(&flow.id).await; - } else if let Err(e) = self.inner.storage.save_flow(&flow).await { + } else if let Err(e) = self.inner.storage.save_flow(flow).await { error!("Failed to save flow to storage: {}", e); return Err(e.into()); } diff --git a/backend/tests/api_tests.rs b/backend/tests/api_tests.rs index ccf65ecc..7500d634 100644 --- a/backend/tests/api_tests.rs +++ b/backend/tests/api_tests.rs @@ -87,9 +87,11 @@ async fn test_create_flow() { assert_eq!(response_json["flow"]["name"], "Test Flow"); - // The backend must assign a new ID (not reuse the one from the request) + // The backend must keep the id the caller supplied (see #672). It used to + // overwrite it, which made the required `id` field of the request schema + // meaningless and stopped callers starting a flow by an id they chose. let returned_id = response_json["flow"]["id"].as_str().unwrap(); - assert_ne!(returned_id, flow.id.to_string()); + assert_eq!(returned_id, flow.id.to_string()); // Runtime state must be cleared assert_eq!(response_json["flow"]["running"], false); diff --git a/backend/tests/flow_id_honoured_test.rs b/backend/tests/flow_id_honoured_test.rs new file mode 100644 index 00000000..afe9137b --- /dev/null +++ b/backend/tests/flow_id_honoured_test.rs @@ -0,0 +1,170 @@ +//! Regression tests for `POST /api/flows` discarding the client-supplied id. +//! +//! `create_flow` used to run `flow.id = FlowId::new_v4()` unconditionally, even +//! though `id` is a required field of the request schema. A caller that +//! pre-generated an id could not create a flow and then start it by that id — +//! it had to read the assigned id back out of the response first. +//! +//! These tests call `create_flow` directly, so reverting the fix in +//! `backend/src/api/flows.rs` turns `creates_flow_with_the_supplied_id` red. + +use axum::extract::State; +use axum::http::StatusCode; +use strom::api::flows::create_flow; +use strom::json_rejection::JsonBody; +use strom::state::AppState; +use strom::storage::JsonFileStorage; +use strom_types::Flow; +use tempfile::NamedTempFile; + +fn new_state() -> AppState { + let storage_file = NamedTempFile::new().unwrap(); + let blocks_file = NamedTempFile::new().unwrap(); + let storage = JsonFileStorage::new(storage_file.path()); + AppState::new( + storage, + blocks_file.path(), + std::env::temp_dir(), + vec![], + "all".to_string(), + vec![], + ) +} + +/// The id the caller sends is the id the flow gets, and the id it is stored +/// under. This is the assertion that fails if the unconditional overwrite +/// returns. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn creates_flow_with_the_supplied_id() { + gstreamer::init().unwrap(); + let state = new_state(); + + let mut flow = Flow::new("supplied-id"); + let chosen = Flow::new("scratch").id; // a fresh, known uuid + flow.id = chosen; + + let (status, body) = create_flow(State(state.clone()), JsonBody(flow)) + .await + .expect("create_flow should succeed"); + + assert_eq!(status, StatusCode::CREATED); + assert_eq!( + body.0.flow.id, chosen, + "the server must keep the id the caller supplied" + ); + assert!( + state.get_flow(&chosen).await.is_some(), + "the flow must be retrievable by the supplied id" + ); +} + +/// Reusing an existing id is a conflict, not a silent overwrite of the other +/// flow. Before this change the second create simply got a different id. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn rejects_a_duplicate_id_with_conflict() { + gstreamer::init().unwrap(); + let state = new_state(); + + let mut first = Flow::new("first"); + let shared = Flow::new("scratch").id; + first.id = shared; + let _first = create_flow(State(state.clone()), JsonBody(first)) + .await + .expect("first create should succeed"); + + let mut second = Flow::new("second"); + second.id = shared; + let err = create_flow(State(state.clone()), JsonBody(second)) + .await + .expect_err("a duplicate id must be rejected"); + + assert_eq!(err.0, StatusCode::CONFLICT); + + let stored = state + .get_flow(&shared) + .await + .expect("the original flow must still exist"); + assert_eq!( + stored.name, "first", + "the conflicting create must not have overwritten the original flow" + ); +} + +/// A nil uuid means "no id supplied" — the server assigns one rather than +/// storing a flow keyed on all-zeros. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn assigns_an_id_when_the_caller_sends_nil() { + gstreamer::init().unwrap(); + let state = new_state(); + + let mut flow = Flow::new("nil-id"); + flow.id = Default::default(); // uuid nil + assert!( + flow.id.is_nil(), + "precondition: the request carries a nil id" + ); + + let (status, body) = create_flow(State(state.clone()), JsonBody(flow)) + .await + .expect("create_flow should succeed"); + + assert_eq!(status, StatusCode::CREATED); + assert!( + !body.0.flow.id.is_nil(), + "a nil id must be replaced with a generated one" + ); + assert!(state.get_flow(&body.0.flow.id).await.is_some()); +} + +/// Concurrent creates that supply the same id must produce exactly one flow. +/// +/// The conflict check used to be a `get_flow` followed by a separate +/// `upsert_flow`, so two creates could both pass the check and the second would +/// overwrite the first. The id is now claimed inside the same write lock that +/// checks it. This is a probabilistic guard rather than a strict one — the old +/// code only lost the race when the tasks actually interleaved — but with this +/// many concurrent creates it fails reliably against the pre-fix version. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_creates_with_the_same_id_yield_one_flow() { + gstreamer::init().unwrap(); + let state = new_state(); + + let shared = Flow::new("scratch").id; + let attempts = 16; + + let mut handles = Vec::with_capacity(attempts); + for i in 0..attempts { + let state = state.clone(); + handles.push(tokio::spawn(async move { + let mut flow = Flow::new(format!("racer-{i}")); + flow.id = shared; + create_flow(State(state), JsonBody(flow)).await + })); + } + + let mut created = 0; + let mut conflicts = 0; + for handle in handles { + match handle.await.expect("task should not panic") { + Ok((status, _)) => { + assert_eq!(status, StatusCode::CREATED); + created += 1; + } + Err((status, _)) => { + assert_eq!( + status, + StatusCode::CONFLICT, + "a losing create must report a conflict, not a server error" + ); + conflicts += 1; + } + } + } + + assert_eq!(created, 1, "exactly one create may win the id"); + assert_eq!(conflicts, attempts - 1); + assert!( + state.get_flow(&shared).await.is_some(), + "the winning flow must be stored under the shared id" + ); +} diff --git a/openapi.json b/openapi.json index 68ab939f..1b62633b 100644 --- a/openapi.json +++ b/openapi.json @@ -717,6 +717,7 @@ "flows" ], "summary": "Create a new flow.", + "description": "Creates a flow under the `id` supplied in the body, so a caller can pre-generate an id and then start the flow by it. `id` is a required field: to have the server assign one instead, send the nil uuid (`00000000-0000-0000-0000-000000000000`) and read the assigned id from the `flow.id` of the response. Reusing the id of an existing flow is a 409; use `POST /api/flows/{id}` to update that flow instead.", "operationId": "create_flow", "requestBody": { "content": { @@ -739,6 +740,16 @@ } } }, + "409": { + "description": "A flow with the supplied id already exists", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, "500": { "description": "Internal server error", "content": {