simplify routes (#1650)

diptanu · web-flow · commit 8b1651323fb5 · 2025-07-29T16:50:01.000-07:00
* simplify routes

* add comment

* add request_id to v1 invoke

* fix request_id

* fix build

* returning no content if there is no output

* update tensorlake

* update deps
diff --git a/indexify/poetry.lock b/indexify/poetry.lock
diff --git a/indexify/pyproject.toml b/indexify/pyproject.toml
@@ -29,7 +29,7 @@ psutil = "^7.0.0"
 boto3 = "^1.39.15"
 # Adds function-executor binary, utils lib, sdk used in indexify-cli commands.
 # We need to specify the tensorlake version exactly because pip install doesn't respect poetry.lock files.
-tensorlake = "0.2.33"
+tensorlake = "0.2.35"
 # Uncomment the next line to use local tensorlake package (only for development!)
 # tensorlake = { path = "../tensorlake", develop = true }
 # grpcio is provided by tensorlake
diff --git a/server/src/http_objects_v1.rs b/server/src/http_objects_v1.rs
@@ -122,7 +122,7 @@ impl From<GraphInvocationCtx> for ShallowGraphRequest {
             id: ctx.invocation_id.to_string(),
             created_at: ctx.created_at,
             status: if ctx.completed {
-                RequestStatus::Finalized
+                RequestStatus::Complete
             } else if ctx.outstanding_tasks > 0 {
                 RequestStatus::Running
             } else {
@@ -170,13 +170,15 @@ pub struct Tasks {
 }
 
 #[derive(Debug, Serialize, Deserialize, ToSchema, Clone)]
+#[serde(rename_all = "lowercase")]
 pub enum RequestStatus {
     Pending,
     Running,
-    Finalized,
+    Complete,
 }
 
 #[derive(Debug, Serialize, Deserialize, ToSchema, Clone)]
+#[serde(rename_all = "lowercase")]
 pub enum RequestOutcome {
     Undefined,
     Success,
@@ -194,6 +196,7 @@ impl From<GraphInvocationOutcome> for RequestOutcome {
 }
 
 #[derive(Debug, Serialize, Deserialize, ToSchema, Clone)]
+#[serde(rename_all = "lowercase")]
 pub enum RequestFailureReason {
     Unknown,
     InternalError,
@@ -257,7 +260,7 @@ impl Request {
             );
         }
         let status = if ctx.completed {
-            RequestStatus::Finalized
+            RequestStatus::Complete
         } else if ctx.outstanding_tasks > 0 {
             RequestStatus::Running
         } else {
diff --git a/server/src/routes/download.rs b/server/src/routes/download.rs
@@ -2,14 +2,14 @@ use anyhow::anyhow;
 use axum::{
     body::Body,
     extract::{Path, State},
-    response::Response,
+    response::{IntoResponse, NoContent, Response},
 };
 use futures::TryStreamExt;
 
 use super::routes_state::RouteState;
 use crate::{
     blob_store::BlobStorage,
-    data_model::GraphInvocationError,
+    data_model::{DataPayload, GraphInvocationError},
     http_objects::{IndexifyAPIError, RequestError},
 };
 
@@ -120,7 +120,7 @@ pub async fn download_fn_output_payload(
         .map_err(|e| IndexifyAPIError::internal_error_str(&e.to_string()))
 }
 
-/// Get function output
+/// Get function output by index
 #[utoipa::path(
     get,
     path = "v1/namespaces/{namespace}/compute-graphs/{compute_graph}/requests/{request_id}/fn/{fn_name}/outputs/{id}/index/{index}",
@@ -159,14 +159,60 @@ pub async fn v1_download_fn_output_payload(
     let encoding = output.encoding.clone();
 
     let payload = &output.payloads[index];
-    let storage_reader = state
-        .blob_storage
-        .get_blob_store(&namespace)
+    let blob_storage = state.blob_storage.get_blob_store(&namespace);
+    stream_data_payload(&payload, &blob_storage, &encoding).await
+}
+
+/// Get function output
+#[utoipa::path(
+    get,
+    path = "v1/namespaces/{namespace}/compute-graphs/{compute_graph}/requests/{request_id}/output/{fn_name}",
+    tag = "retrieve",
+    responses(
+        (status = 200, description = "function output"),
+        (status = INTERNAL_SERVER_ERROR, description = "internal server error")
+    ),
+)]
+pub async fn v1_download_fn_output_payload_simple(
+    Path((namespace, compute_graph, invocation_id, fn_name)): Path<(
+        String,
+        String,
+        String,
+        String,
+    )>,
+    State(state): State<RouteState>,
+) -> Result<Response<Body>, IndexifyAPIError> {
+    let output = state
+        .indexify_state
+        .reader()
+        .fn_output_payload_first(&namespace, &compute_graph, &invocation_id, &fn_name)
+        .map_err(|e| {
+            IndexifyAPIError::internal_error(anyhow!(
+                "failed to download invocation payload: {}",
+                e
+            ))
+        })?;
+    let Some(output) = output else {
+        return Ok(NoContent.into_response());
+    };
+
+    let encoding = output.encoding.clone();
+
+    let payload = &output.payloads[0];
+    let blob_storage = state.blob_storage.get_blob_store(&namespace);
+    stream_data_payload(&payload, &blob_storage, &encoding).await
+}
+
+async fn stream_data_payload(
+    payload: &DataPayload,
+    blob_storage: &BlobStorage,
+    encoding: &str,
+) -> Result<Response<Body>, IndexifyAPIError> {
+    let storage_reader = blob_storage
         .get(&payload.path)
         .await
         .map_err(IndexifyAPIError::internal_error)?;
 
-    // Check if the content type is JSON
     if encoding == "application/json" {
         let json_bytes = storage_reader
             .map_ok(|chunk| chunk.to_vec())
diff --git a/server/src/routes/invoke.rs b/server/src/routes/invoke.rs
@@ -9,7 +9,7 @@ use axum::{
     Json,
 };
 use futures::{Stream, StreamExt};
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 use tokio::sync::broadcast::{error::RecvError, Receiver};
 use tracing::{error, info, warn};
 use uuid::Uuid;
@@ -134,6 +134,12 @@ async fn create_invocation_progress_stream(
     }
 }
 
+#[derive(Serialize)]
+struct RequestIdV1 {
+    // FIXME: Remove this once we migrate clients off this.
+    id: String,
+    request_id: String,
+}
 /// Make a request to a workflow
 #[utoipa::path(
     post,
@@ -244,8 +250,9 @@ pub async fn invoke_with_object_v1(
         })?;
 
     if accept_header.contains("application/json") {
-        return Ok(Json(RequestId {
-            id: graph_invocation_ctx.invocation_id,
+        return Ok(Json(RequestIdV1 {
+            id: graph_invocation_ctx.invocation_id.clone(),
+            request_id: graph_invocation_ctx.invocation_id.clone(),
         })
         .into_response());
     }
diff --git a/server/src/routes_v1.rs b/server/src/routes_v1.rs
@@ -46,7 +46,7 @@ use crate::{
     http_objects_v1::{self, GraphRequests},
     routes::{
         compute_graphs::{self, create_or_update_compute_graph_v1},
-        download::{self, v1_download_fn_output_payload},
+        download::{self, v1_download_fn_output_payload, v1_download_fn_output_payload_simple},
         invoke::{self, progress_stream},
         routes_state::RouteState,
     },
@@ -163,10 +163,19 @@ fn v1_namespace_routes(route_state: RouteState) -> Router {
             "/compute-graphs/{compute_graph}/requests/{request_id}/tasks",
             get(list_tasks).with_state(route_state.clone()),
         )
+        // FIXME: remove this route once we migrate tensorlake sdk to this
         .route(
             "/compute-graphs/{compute_graph}/requests/{request_id}/fn/{fn_name}/outputs/{id}/index/{index}",
             get(v1_download_fn_output_payload).with_state(route_state.clone()),
         )
+        .route(
+            "/compute-graphs/{compute_graph}/requests/{request_id}/output/{fn_name}/id/{id}/index/{index}",
+            get(v1_download_fn_output_payload).with_state(route_state.clone()),
+        )
+        .route(
+            "/compute-graphs/{compute_graph}/requests/{request_id}/output/{fn_name}",
+            get(v1_download_fn_output_payload_simple).with_state(route_state.clone()),
+        )
         .layer(middleware::from_fn(move |rpp, r, n| {
             namespace_middleware(route_state.clone(), rpp, r, n)
         }))
diff --git a/server/src/state_store/invocation_events.rs b/server/src/state_store/invocation_events.rs
@@ -95,6 +95,7 @@ pub struct TaskAssigned {
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(rename_all = "lowercase")]
 pub enum TaskOutcomeSummary {
     Unknown,
     Success,
diff --git a/server/src/state_store/scanner.rs b/server/src/state_store/scanner.rs
@@ -714,6 +714,30 @@ impl StateReader {
             None => Ok(None),
         }
     }
+
+    pub fn fn_output_payload_first(
+        &self,
+        namespace: &str,
+        compute_graph: &str,
+        invocation_id: &str,
+        compute_fn: &str,
+    ) -> Result<Option<NodeOutput>> {
+        let kvs = &[KeyValue::new("op", "fn_output_payload_first")];
+        let _timer = Timer::start_with_labels(&self.metrics.state_read, kvs);
+
+        let key = format!(
+            "{}|{}|{}|{}",
+            namespace, compute_graph, invocation_id, compute_fn
+        );
+
+        let (node_outputs, _) = self.get_rows_from_cf_with_limits::<NodeOutput>(
+            key.as_bytes(),
+            None,
+            IndexifyObjectsColumns::FnOutputs,
+            Some(1),
+        )?;
+        Ok(node_outputs.first().cloned())
+    }
 }
 
 #[cfg(test)]
diff --git a/tensorlake b/tensorlake
@@ -1 +1 @@
-Subproject commit d0e97eaccf2a75e012d5ccf816adc48239636f18
+Subproject commit 457bb03d2f6e1a5bb506966c0a48c5c09b97f851

Original file line number	Diff line number	Diff line change
`@@ -95,6 +95,7 @@ pub struct TaskAssigned {`
`95`	`95`	`}`
`96`	`96`
`97`	`97`	`#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]`
	`98`	`+#[serde(rename_all = "lowercase")]`
`98`	`99`	`pub enum TaskOutcomeSummary {`
`99`	`100`	`Unknown,`
`100`	`101`	`Success,`