🐛(backend) make start-recording atomic and fault-tolerant

Wrap Recording and RecordingAccess creation in a single transaction so a
partial failure does not leave orphan rows, and return 409 instead of 500
when a recording is already in progress for the room.

When the worker fails to start, transition the Recording to
FAILED_TO_START so the unique partial constraint on (room, status) no
longer blocks future recording attempts on the same room.

src/backend/core/tests/rooms/test_api_rooms_start_recording.py

@@ -140,16 +140,18 @@ def test_start_recording_worker_error(
 
     mock_worker_service_factory.assert_called_once_with(mode="screen_recording")
 
-    assert response.status_code == 500
+    assert response.status_code == 502
     assert response.json() == {
         "error": f"Recording failed to start for room {room.slug}"
     }
 
-    # Recording object should be created even if worker fails
+    # Recording object should be created even if worker fails, and moved out
+    # of the unique-constraint window so the room is not locked.
     assert Recording.objects.count() == 1
     recording = Recording.objects.first()
     assert recording.room == room
     assert recording.mode == "screen_recording"
+    assert recording.status == "failed_to_start"
 
     # Verify recording access details
     assert recording.accesses.count() == 1
@@ -158,6 +160,72 @@ def test_start_recording_worker_error(
     assert access.role == "owner"
 
 
+@pytest.mark.parametrize(
+    "status",
+    ["active", "initiated"],
+)
+def test_start_recording_conflict_when_already_in_progress(
+    status, mock_worker_service_factory, mock_worker_manager, settings
+):
+    """Should return 409 when a second start is attempted while a recording is already active."""
+    settings.RECORDING_ENABLE = True
+
+    room = RoomFactory()
+    user = UserFactory()
+    room.accesses.create(user=user, role="owner")
+
+    # Pre-existing active recording for the same room.
+    Recording.objects.create(room=room, mode="screen_recording", status="active")
+
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(
+        f"/api/v1.0/rooms/{room.id}/start-recording/",
+        {"mode": "screen_recording"},
+    )
+
+    assert response.status_code == 409
+    assert response.json() == {
+        "error": f"A recording is already in progress for room {room.slug}"
+    }
+    # No new recording row, no access row leaked from the rolled-back transaction.
+    assert Recording.objects.count() == 1
+    assert Recording.objects.first().accesses.count() == 0
+    mock_worker_manager.start.assert_not_called()
+
+
+def test_start_recording_after_worker_failure_unblocks_room(
+    mock_worker_service_factory, mock_worker_manager, settings
+):
+    """Should allow a new recording when the previous recording failed."""
+    settings.RECORDING_ENABLE = True
+
+    room = RoomFactory()
+    user = UserFactory()
+    room.accesses.create(user=user, role="owner")
+
+    mock_worker_manager.start = mock.Mock(
+        side_effect=[RecordingStartError("boom"), None]
+    )
+
+    client = APIClient()
+    client.force_login(user)
+
+    first = client.post(
+        f"/api/v1.0/rooms/{room.id}/start-recording/",
+        {"mode": "screen_recording"},
+    )
+    assert first.status_code == 502
+
+    second = client.post(
+        f"/api/v1.0/rooms/{room.id}/start-recording/",
+        {"mode": "screen_recording"},
+    )
+    assert second.status_code == 201
+    assert Recording.objects.count() == 2
+
+
 def test_start_recording_success(
     mock_worker_service_factory, mock_worker_manager, settings
 ):