[Http Streaming] Fix error streaming response (#30760)

GitOrigin-RevId: 0b53282c14f0defa0ba80ab0d77f9e83f9bcddb1

Author: jordanhunt22

crates/isolate/src/environment/helpers/validation.rs

@@ -32,13 +32,7 @@ use database::{
 use errors::ErrorMetadata;
 use keybroker::Identity;
 use model::{
-    backend_state::{
-        types::BackendState,
-        BackendStateModel,
-        DISABLED_ERROR_MESSAGE,
-        PAUSED_ERROR_MESSAGE,
-        SUSPENDED_ERROR_MESSAGE,
-    },
+    backend_state::BackendStateModel,
     components::ComponentsModel,
     modules::{
         function_validators::ReturnsValidator,
@@ -310,23 +304,13 @@ impl ValidatedPathAndArgs {
             };
             return Ok(result);
         }
-        let mut backend_state_model = BackendStateModel::new(tx);
-        let backend_state = backend_state_model.get_backend_state().await?;
-        match backend_state {
-            BackendState::Running => {},
-            BackendState::Paused => {
-                return Ok(Err(JsError::from_message(PAUSED_ERROR_MESSAGE.to_string())));
-            },
-            BackendState::Disabled => {
-                return Ok(Err(JsError::from_message(
-                    DISABLED_ERROR_MESSAGE.to_string(),
-                )));
-            },
-            BackendState::Suspended => {
-                return Ok(Err(JsError::from_message(
-                    SUSPENDED_ERROR_MESSAGE.to_string(),
-                )));
+
+        match BackendStateModel::new(tx).fail_while_not_running().await {
+            Ok(Ok(())) => {},
+            Ok(Err(e)) => {
+                return Ok(Err(e));
             },
+            Err(e) => return Err(e),
         }
 
         let path = match public_path.clone() {
@@ -565,8 +549,7 @@ impl TryFrom<ValidatedPathAndArgs> for pb::common::ValidatedPathAndArgs {
 ///
 /// This should only be constructed via `ValidatedHttpRoute::try_from` to use
 /// the type system to enforce that validation is never skipped.
-#[derive(Clone, Eq, PartialEq)]
-#[cfg_attr(any(test, feature = "testing"), derive(Debug))]
+#[derive(Clone, Eq, PartialEq, Debug)]
 pub struct ValidatedHttpPath {
     path: ResolvedComponentFunctionPath,
     npm_version: Option<Version>,
@@ -600,18 +583,24 @@ impl ValidatedHttpPath {
         tx: &mut Transaction<RT>,
         udf_path: sync_types::CanonicalizedUdfPath,
         npm_version: Option<Version>,
-    ) -> anyhow::Result<Self> {
+    ) -> anyhow::Result<Result<Self, JsError>> {
         if !udf_path.is_system() {
-            BackendStateModel::new(tx).fail_while_not_running().await?;
+            match BackendStateModel::new(tx).fail_while_not_running().await {
+                Ok(Ok(())) => {},
+                Ok(Err(e)) => {
+                    return Ok(Err(e));
+                },
+                Err(e) => return Err(e),
+            }
         }
-        Ok(Self {
+        Ok(Ok(Self {
             path: ResolvedComponentFunctionPath {
                 component: ComponentId::test_user(),
                 udf_path,
                 component_path: Some(ComponentPath::test_user()),
             },
             npm_version,
-        })
+        }))
     }
 
     pub async fn new<RT: Runtime>(
@@ -625,7 +614,13 @@ impl ValidatedHttpPath {
             path.udf_path,
         );
         if !path.udf_path.is_system() {
-            BackendStateModel::new(tx).fail_while_not_running().await?;
+            match BackendStateModel::new(tx).fail_while_not_running().await {
+                Ok(Ok(())) => {},
+                Ok(Err(e)) => {
+                    return Ok(Err(e));
+                },
+                Err(e) => return Err(e),
+            }
         }
         let (_, component) =
             BootstrapComponentsModel::new(tx).must_component_path_to_ids(&path.component)?;

crates/isolate/src/test_helpers.rs

@@ -173,7 +173,6 @@ use crate::{
     validate_schedule_args,
     ActionCallbacks,
     BackendIsolateWorker,
-    HttpActionOutcome,
     HttpActionResponse,
     HttpActionResult,
     IsolateClient,
@@ -915,7 +914,7 @@ impl<RT: Runtime, P: Persistence + Clone> UdfTest<RT, P> {
                 HttpActionResponsePart::Head(head) => response_head = Some(head),
             }
         }
-        let response = match outcome.result {
+        let response = match outcome {
             HttpActionResult::Error(e) => Err(e),
             HttpActionResult::Streamed => {
                 let response_head = response_head.unwrap();
@@ -935,17 +934,22 @@ impl<RT: Runtime, P: Persistence + Clone> UdfTest<RT, P> {
         http_request: HttpActionRequest,
         identity: Identity,
         http_response_streamer: HttpActionResponseStreamer,
-    ) -> anyhow::Result<(HttpActionOutcome, LogLines)> {
+    ) -> anyhow::Result<(HttpActionResult, LogLines)> {
         let app = Arc::new(self.clone());
         let mut tx = self.database.begin(identity.clone()).await?;
         let path: UdfPath = udf_path.parse()?;
+        let validated_path =
+            match ValidatedHttpPath::new_for_tests(&mut tx, path.canonicalize(), None).await? {
+                Ok(validated_path) => validated_path,
+                Err(e) => return Ok((HttpActionResult::Error(e), vec![].into())),
+            };
 
         let fetch_client = Arc::new(ProxiedFetchClient::new(None, DEV_INSTANCE_NAME.to_owned()));
         let (log_line_sender, mut log_line_receiver) = mpsc::unbounded_channel();
         let outcome = self
             .isolate
             .execute_http_action(
-                ValidatedHttpPath::new_for_tests(&mut tx, path.canonicalize(), None).await?,
+                validated_path,
                 RoutedHttpPath(http_request.head.url.path().to_string()),
                 http_request,
                 identity,
@@ -961,7 +965,7 @@ impl<RT: Runtime, P: Persistence + Clone> UdfTest<RT, P> {
         while let Some(log_line) = log_line_receiver.recv().await {
             log_lines.push(log_line);
         }
-        Ok((outcome, log_lines.into()))
+        Ok((outcome.result, log_lines.into()))
     }
 
     pub async fn action(&self, udf_path: &str, args: ConvexObject) -> anyhow::Result<ConvexValue> {

crates/isolate/src/tests/backend_state.rs

@@ -3,7 +3,6 @@ use common::{
     testing::assert_contains,
 };
 use database::Database;
-use errors::ErrorMetadata;
 use keybroker::Identity;
 use model::backend_state::{
     types::BackendState,
@@ -23,6 +22,7 @@ use crate::{
         http_post_request,
     },
     HttpActionResponseStreamer,
+    HttpActionResult,
 };
 
 #[convex_macro::test_runtime]
@@ -42,7 +42,7 @@ async fn test_action_while_paused(rt: TestRuntime) -> anyhow::Result<()> {
 
 #[convex_macro::test_runtime]
 async fn test_http_action_while_paused(rt: TestRuntime) -> anyhow::Result<()> {
-    test_http_action_helper(rt, BackendState::Paused).await
+    test_http_action_helper(rt, BackendState::Paused, PAUSED_ERROR_MESSAGE).await
 }
 
 #[convex_macro::test_runtime]
@@ -62,7 +62,7 @@ async fn test_action_while_disabled(rt: TestRuntime) -> anyhow::Result<()> {
 
 #[convex_macro::test_runtime]
 async fn test_http_action_while_disabled(rt: TestRuntime) -> anyhow::Result<()> {
-    test_http_action_helper(rt, BackendState::Disabled).await
+    test_http_action_helper(rt, BackendState::Disabled, DISABLED_ERROR_MESSAGE).await
 }
 
 #[convex_macro::test_runtime]
@@ -82,7 +82,7 @@ async fn test_action_while_suspended(rt: TestRuntime) -> anyhow::Result<()> {
 
 #[convex_macro::test_runtime]
 async fn test_http_action_while_suspended(rt: TestRuntime) -> anyhow::Result<()> {
-    test_http_action_helper(rt, BackendState::Suspended).await
+    test_http_action_helper(rt, BackendState::Suspended, SUSPENDED_ERROR_MESSAGE).await
 }
 
 async fn test_query_helper(
@@ -128,32 +128,25 @@ async fn test_action_helper(
 async fn test_http_action_helper(
     rt: TestRuntime,
     backend_state: BackendState,
+    error_message: &str,
 ) -> anyhow::Result<()> {
     let t = http_action_udf_test(rt).await?;
     toggle_backend_state(&t.database, backend_state.clone()).await?;
     let (http_response_sender, _http_response_receiver) = mpsc::unbounded_channel();
-    let error = t
+    let result = t
         .raw_http_action(
             "http_action",
             http_post_request("basic", "hi".as_bytes().to_vec()),
             Identity::system(),
             HttpActionResponseStreamer::new(http_response_sender),
         )
-        .await
-        .unwrap_err();
-    assert_error(error, backend_state);
-    Ok(())
-}
-
-fn assert_error(error: anyhow::Error, backend_state: BackendState) {
-    let error_message = match backend_state {
-        BackendState::Paused => "NoRunWhilePaused",
-        BackendState::Disabled => "NoRunWhileDisabled",
-        BackendState::Suspended => "NoRunWhileSuspended",
-        BackendState::Running => return,
+        .await?;
+    let (HttpActionResult::Error(e), _) = result else {
+        anyhow::bail!("Expected error, got {:?}", result);
     };
-    let error_metadata = error.downcast_ref::<ErrorMetadata>().unwrap();
-    assert_eq!(error_metadata.short_msg, error_message);
+    assert_contains(&e.message, error_message);
+
+    Ok(())
 }
 
 async fn toggle_backend_state<RT: Runtime>(

crates/isolate/src/tests/http_action.rs

@@ -198,19 +198,20 @@ async fn test_http_echo(rt: TestRuntime) -> anyhow::Result<()> {
 
 #[convex_macro::test_runtime]
 async fn test_http_scheduler(rt: TestRuntime) -> anyhow::Result<()> {
-    let t = http_action_udf_test(rt).await?;
+    let t = http_action_udf_test(rt.clone()).await?;
 
     let (http_response_sender, _http_response_receiver) = mpsc::unbounded_channel();
-    let (outcome, _log_lines) = t
+    let (result, _) = t
         .raw_http_action(
             "http_action",
             http_request("schedule"),
             Identity::system(),
             HttpActionResponseStreamer::new(http_response_sender),
         )
         .await?;
+    let http_finish_ts = rt.clone().unix_timestamp();
 
-    assert_matches!(outcome.result, HttpActionResult::Streamed);
+    assert_matches!(result, HttpActionResult::Streamed);
 
     let result = t.query("scheduler:getScheduledJobs", assert_obj!()).await?;
     must_let!(let ConvexValue::Array(scheduled_jobs) = result);
@@ -222,7 +223,7 @@ async fn test_http_scheduler(rt: TestRuntime) -> anyhow::Result<()> {
 
     // End time of the HTTP action + 2 seconds, which should be a little after when
     // the job was scheduled for
-    let expected_ts = (outcome.unix_timestamp + Duration::from_secs(2)).as_secs_f64() * 1000.0;
+    let expected_ts = (http_finish_ts + Duration::from_secs(2)).as_secs_f64() * 1000.0;
     assert!((job.scheduled_time - expected_ts).abs() < 500.0);
     Ok(())
 }

crates/model/src/backend_state/mod.rs

@@ -5,6 +5,7 @@ use common::{
         ParsedDocument,
         ResolvedDocument,
     },
+    errors::JsError,
     query::{
         Order,
         Query,
@@ -108,24 +109,29 @@ impl<'a, RT: Runtime> BackendStateModel<'a, RT> {
         Ok(backend_state)
     }
 
-    pub async fn fail_while_not_running(&mut self) -> anyhow::Result<()> {
+    /// Fails with an error if the backend is not running. We have to return a
+    /// result of a result of () and a JSError because we use them to
+    /// differentiate between system and user errors.
+    pub async fn fail_while_not_running(&mut self) -> anyhow::Result<Result<(), JsError>> {
         let backend_state = self.get_backend_state().await?;
         match backend_state {
             BackendState::Running => {},
-            BackendState::Paused => anyhow::bail!(ErrorMetadata::bad_request(
-                "NoRunWhilePaused",
-                PAUSED_ERROR_MESSAGE,
-            )),
-            BackendState::Disabled => anyhow::bail!(ErrorMetadata::bad_request(
-                "NoRunWhileDisabled",
-                DISABLED_ERROR_MESSAGE,
-            )),
-            BackendState::Suspended => anyhow::bail!(ErrorMetadata::bad_request(
-                "NoRunWhileSuspended",
-                SUSPENDED_ERROR_MESSAGE,
-            )),
-        };
-        Ok(())
+            BackendState::Paused => {
+                return Ok(Err(JsError::from_message(PAUSED_ERROR_MESSAGE.to_string())));
+            },
+            BackendState::Disabled => {
+                return Ok(Err(JsError::from_message(
+                    DISABLED_ERROR_MESSAGE.to_string(),
+                )));
+            },
+            BackendState::Suspended => {
+                return Ok(Err(JsError::from_message(
+                    SUSPENDED_ERROR_MESSAGE.to_string(),
+                )));
+            },
+        }
+
+        Ok(Ok(()))
     }
 
     pub async fn toggle_backend_state(&mut self, new_state: BackendState) -> anyhow::Result<()> {