fix(webapp): optimize reconciliation to O(1) and add trailing-edge throttle

Author: bharathkumar39293

apps/webapp/app/presenters/v3/RunPresenter.server.ts

@@ -208,16 +208,41 @@ export class RunPresenter {
 
     //we need the start offset for each item, and the total duration of the entire tree
     const treeRootStartTimeMs = tree ? tree?.data.startTime.getTime() : 0;
+
+    const postgresRunDuration =
+      runData.isFinished && run.completedAt
+        ? millisecondsToNanoseconds(
+          run.completedAt.getTime() -
+          (run.rootTaskRun?.createdAt ?? run.createdAt).getTime()
+        )
+        : 0;
+
     let totalDuration = tree?.data.duration ?? 0;
     const events = tree
-      ? flattenTree(tree).map((n) => {
-        const offset = millisecondsToNanoseconds(
-          n.data.startTime.getTime() - treeRootStartTimeMs
-        );
+      ? flattenTree(tree).map((n, index) => {
+        const isRoot = index === 0;
+        const offset = millisecondsToNanoseconds(n.data.startTime.getTime() - treeRootStartTimeMs);
+
+        let nIsPartial = n.data.isPartial;
+        let nDuration = n.data.duration;
+        let nIsError = n.data.isError;
+
+        // NOTE: Clickhouse trace ingestion is eventually consistent.
+        // When a run is marked finished in Postgres, we reconcile the
+        // root span to reflect completion even if telemetry is still partial.
+        // This is a deliberate UI-layer tradeoff to prevent stale or "stuck"
+        // run states in the dashboard.
+        if (isRoot && runData.isFinished && nIsPartial) {
+          nIsPartial = false;
+          nDuration = Math.max(nDuration ?? 0, postgresRunDuration);
+          nIsError = isFailedRunStatus(runData.status);
+        }
+
         //only let non-debug events extend the total duration
         if (!n.data.isDebug) {
-          totalDuration = Math.max(totalDuration, offset + n.data.duration);
+          totalDuration = Math.max(totalDuration, offset + (nIsPartial ? 0 : nDuration));
         }
+
         return {
           ...n,
           data: {
@@ -228,23 +253,24 @@ export class RunPresenter {
               treeRootStartTimeMs
             ),
             //set partial nodes to null duration
-            duration: n.data.isPartial ? null : n.data.duration,
+            duration: nIsPartial ? null : nDuration,
+            isPartial: nIsPartial,
+            isError: nIsError,
             offset,
-            isRoot: n.id === traceSummary.rootSpan.id,
+            isRoot,
           },
         };
       })
       : [];
 
+    if (runData.isFinished) {
+      totalDuration = Math.max(totalDuration, postgresRunDuration);
+    }
+
     //total duration should be a minimum of 1ms
     totalDuration = Math.max(totalDuration, millisecondsToNanoseconds(1));
 
-    const reconciled = reconcileTraceWithRunLifecycle(
-      runData,
-      traceSummary.rootSpan.id,
-      events,
-      totalDuration
-    );
+    const reconciled = reconcileTraceWithRunLifecycle(runData, traceSummary.rootSpan.id, events, totalDuration);
 
     return {
       run: runData,
@@ -285,14 +311,17 @@ export function reconcileTraceWithRunLifecycle(
   totalDuration: number;
   rootSpanStatus: "executing" | "completed" | "failed";
 } {
-  const rootEvent = events.find((e) => e.id === rootSpanId);
-  const currentStatus: "executing" | "completed" | "failed" = rootEvent
-    ? rootEvent.data.isError
-      ? "failed"
-      : !rootEvent.data.isPartial
-        ? "completed"
-        : "executing"
-    : "executing";
+  const rootEvent = events[0];
+  const isActualRoot = rootEvent?.id === rootSpanId;
+
+  const currentStatus: "executing" | "completed" | "failed" =
+    isActualRoot && rootEvent
+      ? rootEvent.data.isError
+        ? "failed"
+        : !rootEvent.data.isPartial
+          ? "completed"
+          : "executing"
+      : "executing";
 
   if (!runData.isFinished) {
     return { events, totalDuration, rootSpanStatus: currentStatus };
@@ -307,23 +336,28 @@ export function reconcileTraceWithRunLifecycle(
 
   const updatedTotalDuration = Math.max(totalDuration, postgresRunDuration);
 
-  const updatedEvents = events.map((e) => {
-    if (e.id === rootSpanId && e.data.isPartial) {
-      return {
-        ...e,
-        data: {
-          ...e.data,
-          isPartial: false,
-          duration: Math.max(e.data.duration ?? 0, postgresRunDuration),
-          isError: isFailedRunStatus(runData.status),
-        },
-      };
-    }
-    return e;
-  });
+  // We only need to potentially update the root event (the first one) if it matches our ID
+  if (isActualRoot && rootEvent && rootEvent.data.isPartial) {
+    const updatedEvents = [...events];
+    updatedEvents[0] = {
+      ...rootEvent,
+      data: {
+        ...rootEvent.data,
+        isPartial: false,
+        duration: Math.max(rootEvent.data.duration ?? 0, postgresRunDuration),
+        isError: isFailedRunStatus(runData.status),
+      },
+    };
+
+    return {
+      events: updatedEvents,
+      totalDuration: updatedTotalDuration,
+      rootSpanStatus: isFailedRunStatus(runData.status) ? "failed" : "completed",
+    };
+  }
 
   return {
-    events: updatedEvents,
+    events,
     totalDuration: updatedTotalDuration,
     rootSpanStatus: isFailedRunStatus(runData.status) ? "failed" : "completed",
   };

apps/webapp/app/utils/throttle.ts

@@ -1,20 +1,28 @@
-//From: https://kettanaito.com/blog/debounce-vs-throttle
-
-/** A very simple throttle. Will execute the function at the end of each period and discard any other calls during that period. */
+/** A throttle that fires the first call immediately and ensures the last call during the duration is also fired. */
 export function throttle(
   func: (...args: any[]) => void,
   durationMs: number
 ): (...args: any[]) => void {
-  let isPrimedToFire = false;
-
-  return (...args: any[]) => {
-    if (!isPrimedToFire) {
-      isPrimedToFire = true;
+  let timeoutId: NodeJS.Timeout | null = null;
+  let nextArgs: any[] | null = null;
 
-      setTimeout(() => {
-        func(...args);
-        isPrimedToFire = false;
-      }, durationMs);
+  const wrapped = (...args: any[]) => {
+    if (timeoutId) {
+      nextArgs = args;
+      return;
     }
+
+    func(...args);
+
+    timeoutId = setTimeout(() => {
+      timeoutId = null;
+      if (nextArgs) {
+        const argsToUse = nextArgs;
+        nextArgs = null;
+        wrapped(...argsToUse);
+      }
+    }, durationMs);
   };
+
+  return wrapped;
 }