fix(api): honour the client-supplied flow id on POST /api/flows

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 })))

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<bool> {
+        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());
         }

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);

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"
+    );
+}

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": {