Improve handling of unavailable logs in completed workflows

.release-please-manifest.json

@@ -7,9 +7,9 @@
   "+3": "0.0.0",
   "frontend/dashboard": "0.1.18",
   "+4": "0.0.0",
-  "frontend/relay-workflows-lib": "0.1.13",
+  "frontend/relay-workflows-lib": "0.1.14",
   "+5": "0.0.0",
-  "frontend/workflows-lib": "0.1.9",
+  "frontend/workflows-lib": "0.1.10",
   "frontend/workflows-lib-shared": "0.1.1",
   "+6": "0.0.0",
   "backend/telemetry": "0.1.2",

backend/graph-proxy/src/graphql/subscription.rs

@@ -1,5 +1,7 @@
 use crate::graphql::AuthGuard;
-use argo_workflows_openapi::IoArgoprojWorkflowV1alpha1WorkflowWatchEvent;
+use argo_workflows_openapi::{
+    APIResult, IoArgoprojWorkflowV1alpha1Workflow, IoArgoprojWorkflowV1alpha1WorkflowWatchEvent,
+};
 use async_graphql::{Context, SimpleObject, Subscription};
 use async_stream::stream;
 use eventsource_stream::Eventsource;
@@ -66,6 +68,52 @@ pub fn get_auth_token(ctx: &Context<'_>) -> anyhow::Result<String> {
         .ok_or_else(|| WorkflowParsingError::MissingAuthToken.into())
 }
 
+/// Checks whether the specified workflow has completed by querying
+/// the Argo Workflows API and inspecting the workflow phase.
+async fn is_workflow_completed(
+    server_url: &ArgoServerUrl,
+    auth_token: &str,
+    namespace: &str,
+    workflow_name: &str,
+) -> anyhow::Result<bool> {
+    let mut url = server_url.deref().clone();
+
+    url.path_segments_mut().expect("Invalid base URL").extend([
+        "api",
+        "v1",
+        "workflows",
+        namespace,
+        workflow_name,
+    ]);
+
+    let workflow = reqwest::Client::new()
+        .get(url)
+        .bearer_auth(auth_token)
+        .send()
+        .await?
+        .json::<APIResult<IoArgoprojWorkflowV1alpha1Workflow>>()
+        .await?
+        .into_result()?;
+
+    let Some(status) = workflow.status else {
+        return Ok(false);
+    };
+
+    let phase = status.phase.as_deref();
+
+    tracing::info!(
+        "WORKFLOW_STATUS_CHECK namespace={} workflow={} phase={:?}",
+        namespace,
+        workflow_name,
+        phase
+    );
+
+    Ok(matches!(
+        phase,
+        Some("Succeeded") | Some("Failed") | Some("Error")
+    ))
+}
+
 #[Subscription(guard = "AuthGuard")]
 impl WorkflowsSubscription {
     /// Subscribe to logs for a single pod of a workflow.
@@ -82,14 +130,15 @@ impl WorkflowsSubscription {
     ) -> anyhow::Result<impl Stream<Item = Result<LogEntry, String>>> {
         let auth_token = get_auth_token(ctx)?;
 
+        let server_url = ctx.data_unchecked::<ArgoServerUrl>().clone();
+
         if task_id.is_empty() || task_id == "__NO_TASK_SELECTED__" {
             return Err(anyhow::anyhow!(
                 "A valid task ID is required to retrieve task logs"
             ));
         }
 
-        let server_url = ctx.data_unchecked::<ArgoServerUrl>().deref().clone();
-        let mut url = server_url;
+        let mut url = server_url.deref().clone();
 
         let namespace = visit.to_string();
 
@@ -150,9 +199,41 @@ impl WorkflowsSubscription {
             }
         };
 
+        // A terminal workflow with no archived log must never fall back to
+        // the live Argo log endpoint. The pod may already be gone, and there
+        // is nothing left that can produce additional log data.
+        let completed_without_log = if initial_archive.is_none() {
+            match is_workflow_completed(&server_url, &auth_token, &namespace, &workflow_name).await
+            {
+                Ok(true) => {
+                    tracing::info!(
+                        "COMPLETED_WORKFLOW_WITHOUT_LOG task={} workflow={}",
+                        task_id,
+                        workflow_name
+                    );
+                    true
+                }
+
+                Ok(false) => false,
+
+                Err(err) => {
+                    tracing::warn!(
+                        "FAILED_TO_CHECK_WORKFLOW_STATUS task={} workflow={} error={}",
+                        task_id,
+                        workflow_name,
+                        err
+                    );
+                    false
+                }
+            }
+        } else {
+            false
+        };
+
         // Only contact the Argo live-log endpoint when the archived
-        // main.log is not already available in S3.
-        let live_response = if initial_archive.is_none() {
+        // main.log is not already available in S3 and the workflow is still
+        // running.
+        let live_response = if initial_archive.is_none() && !completed_without_log {
             tracing::info!(
                 "STARTING_LIVE_STREAM namespace={} workflow={} task={}",
                 namespace,
@@ -179,6 +260,11 @@ impl WorkflowsSubscription {
         let mut byte_stream = live_response.map(|response| response.bytes_stream());
 
         let log_stream = stream! {
+            if completed_without_log {
+                yield Err("Log not available".to_string());
+                return;
+            }
+
             if let Some(archive_response) = initial_archive {
                 let archive_bytes = match archive_response.body.collect().await {
                     Ok(bytes) => bytes,
@@ -404,7 +490,7 @@ impl WorkflowsSubscription {
                         }
 
                         Err(_err) => {
-                            yield Err("No logs available".to_string());
+                            yield Err("Log not available".to_string());
                             return;
                         }
                     }

frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx

@@ -34,8 +34,11 @@ interface TaskLogSubscriptionProps {
   setLogLines: Dispatch<SetStateAction<string[]>>;
   setTaskCompleted: Dispatch<SetStateAction<boolean>>;
   setSubscriptionError: Dispatch<SetStateAction<string | null>>;
+  setLogUnavailable: Dispatch<SetStateAction<boolean>>;
 }
 
+const LOG_NOT_AVAILABLE = "Log not available";
+
 const taskLogViewerSubscription = graphql`
   subscription TaskLogViewerSubscription(
     $visit: VisitInput!
@@ -56,6 +59,7 @@ const TaskLogSubscription: React.FC<TaskLogSubscriptionProps> = ({
   setLogLines,
   setTaskCompleted,
   setSubscriptionError,
+  setLogUnavailable,
 }) => {
   const subscriptionConfig = useMemo<
     GraphQLSubscriptionConfig<TaskLogViewerSubscription>
@@ -67,30 +71,49 @@ const TaskLogSubscription: React.FC<TaskLogSubscriptionProps> = ({
         workflowName,
         taskId,
       },
+
       onNext: (payload) => {
         const line = payload?.logs.content;
 
         if (line) {
           setLogLines((previousLines) => [...previousLines, line]);
         }
       },
+
       onError: (error) => {
         console.error("Log subscription error:", error);
 
         const message = error instanceof Error ? error.message : String(error);
 
-        if (
+        const logUnavailable =
+          message.includes(LOG_NOT_AVAILABLE) ||
           message.includes("NoSuchKey") ||
           message.includes("No logs") ||
-          message.includes("Failed to retrieve archived log artifact")
-        ) {
-          setSubscriptionError("No logs available");
-        } else {
-          setSubscriptionError("Unable to retrieve task logs");
+          message.includes("Failed to retrieve archived log artifact");
+
+        if (logUnavailable) {
+          /*
+           * This is a terminal state.
+           *
+           * Do not allow the subscription to remain mounted because Relay
+           * may otherwise reconnect it and repeatedly request the same
+           * unavailable log.
+           */
+          setSubscriptionError(LOG_NOT_AVAILABLE);
+          setLogUnavailable(true);
+          setTaskCompleted(true);
+
+          return;
         }
 
+        /*
+         * Preserve the existing behaviour for non-terminal errors.
+         * These can still happen while a workflow is active.
+         */
+        setSubscriptionError("Unable to retrieve task logs");
         setTaskCompleted(true);
       },
+
       onCompleted: () => {
         setTaskCompleted(true);
       },
@@ -102,6 +125,7 @@ const TaskLogSubscription: React.FC<TaskLogSubscriptionProps> = ({
       setLogLines,
       setTaskCompleted,
       setSubscriptionError,
+      setLogUnavailable,
     ],
   );
 
@@ -121,10 +145,20 @@ const TaskLogViewerContent: React.FC<TaskLogViewerProps> = ({
   const [subscriptionError, setSubscriptionError] = useState<string | null>(
     null,
   );
+
+  /*
+   * Once the backend tells us that no archived log exists, this prevents
+   * the Relay subscription from being mounted again.
+   */
+  const [logUnavailable, setLogUnavailable] = useState(false);
+
   const [expanded, setExpanded] = useState(Boolean(selectedTaskId));
 
   const containerRef = useRef<HTMLDivElement>(null);
 
+  /*
+   * Keep the log view scrolled to the newest line.
+   */
   useEffect(() => {
     if (containerRef.current) {
       containerRef.current.scrollTop = containerRef.current.scrollHeight;
@@ -133,7 +167,7 @@ const TaskLogViewerContent: React.FC<TaskLogViewerProps> = ({
 
   return (
     <>
-      {selectedTaskId && (
+      {selectedTaskId && !logUnavailable && (
         <TaskLogSubscription
           key={`${workflowName}-${selectedTaskId}`}
           visit={visit}
@@ -142,6 +176,7 @@ const TaskLogViewerContent: React.FC<TaskLogViewerProps> = ({
           setLogLines={setLogLines}
           setTaskCompleted={setTaskCompleted}
           setSubscriptionError={setSubscriptionError}
+          setLogUnavailable={setLogUnavailable}
         />
       )}
 
@@ -169,7 +204,7 @@ const TaskLogViewerContent: React.FC<TaskLogViewerProps> = ({
             Logs: {selectedTaskName ?? selectedTaskId ?? "No task selected"}
           </Typography>
 
-          {selectedTaskId && !taskCompleted && (
+          {selectedTaskId && !taskCompleted && !logUnavailable && (
             <CircularProgress
               size={14}
               sx={{
@@ -179,7 +214,7 @@ const TaskLogViewerContent: React.FC<TaskLogViewerProps> = ({
             />
           )}
 
-          {selectedTaskId && taskCompleted && (
+          {selectedTaskId && taskCompleted && !logUnavailable && (
             <Typography
               sx={{
                 color: "#ff3333",
@@ -208,7 +243,17 @@ const TaskLogViewerContent: React.FC<TaskLogViewerProps> = ({
               whiteSpace: "pre-wrap",
             }}
           >
-            {subscriptionError ? (
+            {logUnavailable ? (
+              <Typography
+                sx={{
+                  color: "#ff3333",
+                  fontFamily: "monospace",
+                  fontSize: "12px",
+                }}
+              >
+                {LOG_NOT_AVAILABLE}
+              </Typography>
+            ) : subscriptionError ? (
               <Typography
                 sx={{
                   color: "#ff3333",
@@ -251,7 +296,9 @@ export const TaskLogViewer: React.FC<TaskLogViewerProps> = ({
 }) => {
   return (
     <TaskLogViewerContent
-      key={`${workflowName}-${selectedTaskId ?? "none"}-${visit.proposalCode}-${String(visit.proposalNumber)}-${String(visit.number)}`}
+      key={`${workflowName}-${selectedTaskId ?? "none"}-${visit.proposalCode}-${String(
+        visit.proposalNumber,
+      )}-${String(visit.number)}`}
       visit={visit}
       workflowName={workflowName}
       selectedTaskId={selectedTaskId}

frontend/relay-workflows-lib/package.json

@@ -1,7 +1,7 @@
 {
   "name": "relay-workflows-lib",
   "private": true,
-  "version": "0.1.13",
+  "version": "0.1.14",
   "type": "module",
   "main": "lib/main.ts",
   "scripts": {

frontend/relay-workflows-lib/tests/components/BaseWorkflowRelay.test.tsx

@@ -68,10 +68,8 @@ describe("BaseWorkflowRelay", () => {
     expect(
       await screen.findByText("conditional-steps-first"),
     ).toBeInTheDocument();
-
     expect(screen.getByTitle("abc12345")).toBeInTheDocument();
   });
-
   it("should display flow box nodes when expanded", async () => {
     const accordionButton = await screen.findByRole("button", {
       name: /conditional-steps-first/i,