feat: record RUM workflow execution vitals

src/platform/telemetry/TelemetryRegistry.ts

@@ -7,9 +7,8 @@ import type {
   BeginCheckoutMetadata,
   DefaultViewSetMetadata,
   EnterLinearMetadata,
-  ShareFlowMetadata,
-  ShareLinkOpenedMetadata,
   ExecutionErrorMetadata,
+  ExecutionOutcomeMetadata,
   ExecutionSuccessMetadata,
   HelpCenterClosedMetadata,
   HelpCenterOpenedMetadata,
@@ -22,6 +21,8 @@ import type {
   PageVisibilityMetadata,
   ResubscribeClickMetadata,
   RunButtonProperties,
+  ShareFlowMetadata,
+  ShareLinkOpenedMetadata,
   SettingChangedMetadata,
   SharedWorkflowRunMetadata,
   ShellLayoutMetadata,
@@ -268,6 +269,10 @@ export class TelemetryRegistry implements TelemetryDispatcher {
     this.dispatch((provider) => provider.trackWorkflowExecution?.())
   }
 
+  trackExecutionOutcome(metadata: ExecutionOutcomeMetadata): void {
+    this.dispatch((provider) => provider.trackExecutionOutcome?.(metadata))
+  }
+
   trackExecutionError(metadata: ExecutionErrorMetadata): void {
     this.dispatch((provider) => provider.trackExecutionError?.(metadata))
   }

src/platform/telemetry/initTelemetry.ts

@@ -28,7 +28,8 @@ export async function initTelemetry(): Promise<void> {
       { PostHogTelemetryProvider },
       { ClickHouseTelemetryProvider },
       { SyftTelemetryProvider },
-      { CustomerIoTelemetryProvider }
+      { CustomerIoTelemetryProvider },
+      { DatadogRumTelemetryProvider }
     ] = await Promise.all([
       import('./TelemetryRegistry'),
       import('./providers/cloud/MixpanelTelemetryProvider'),
@@ -37,7 +38,8 @@ export async function initTelemetry(): Promise<void> {
       import('./providers/cloud/PostHogTelemetryProvider'),
       import('./providers/cloud/ClickHouseTelemetryProvider'),
       import('./providers/cloud/SyftTelemetryProvider'),
-      import('./providers/cloud/CustomerIoTelemetryProvider')
+      import('./providers/cloud/CustomerIoTelemetryProvider'),
+      import('./providers/cloud/DatadogRumTelemetryProvider')
     ])
 
     const registry = new TelemetryRegistry()
@@ -48,6 +50,7 @@ export async function initTelemetry(): Promise<void> {
     registry.registerProvider(new ClickHouseTelemetryProvider())
     registry.registerProvider(new SyftTelemetryProvider())
     registry.registerProvider(new CustomerIoTelemetryProvider())
+    registry.registerProvider(new DatadogRumTelemetryProvider())
 
     setTelemetryRegistry(registry)
   })()

src/platform/telemetry/providers/cloud/DatadogRumTelemetryProvider.test.ts

@@ -0,0 +1,48 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+
+import { DatadogRumTelemetryProvider } from './DatadogRumTelemetryProvider'
+
+const addDurationVital = vi.fn()
+
+function installDatadogRum(): void {
+  Object.defineProperty(window, 'DD_RUM', {
+    configurable: true,
+    value: { addDurationVital }
+  })
+}
+
+afterEach(() => {
+  vi.restoreAllMocks()
+  vi.clearAllMocks()
+  Reflect.deleteProperty(window, 'DD_RUM')
+})
+
+describe('DatadogRumTelemetryProvider', () => {
+  it('records a workflow vital with its duration and outcome', () => {
+    installDatadogRum()
+    vi.spyOn(performance, 'now').mockReturnValue(142)
+
+    new DatadogRumTelemetryProvider().trackExecutionOutcome({
+      startTime: 42,
+      outcome: 'success'
+    })
+
+    expect(addDurationVital).toHaveBeenCalledWith('workflow_execution', {
+      startTime: performance.timeOrigin + 42,
+      duration: 100,
+      context: {
+        outcome: 'success',
+        product: 'cloud_generation'
+      }
+    })
+  })
+
+  it('does nothing when Datadog RUM is unavailable', () => {
+    expect(() =>
+      new DatadogRumTelemetryProvider().trackExecutionOutcome({
+        startTime: 42,
+        outcome: 'failure'
+      })
+    ).not.toThrow()
+  })
+})

src/platform/telemetry/providers/cloud/DatadogRumTelemetryProvider.ts

@@ -0,0 +1,35 @@
+import type { ExecutionOutcomeMetadata, TelemetryProvider } from '../../types'
+
+interface DatadogRumDurationVitalOptions {
+  startTime: number
+  duration: number
+  context?: Record<string, unknown>
+}
+
+interface DatadogRumClient {
+  addDurationVital(name: string, options: DatadogRumDurationVitalOptions): void
+}
+
+interface WindowWithDatadogRum extends Window {
+  DD_RUM?: DatadogRumClient
+}
+
+function getDatadogRum(): DatadogRumClient | undefined {
+  return (window as WindowWithDatadogRum).DD_RUM
+}
+
+export class DatadogRumTelemetryProvider implements TelemetryProvider {
+  trackExecutionOutcome({
+    startTime,
+    outcome
+  }: ExecutionOutcomeMetadata): void {
+    getDatadogRum()?.addDurationVital('workflow_execution', {
+      startTime: performance.timeOrigin + startTime,
+      duration: performance.now() - startTime,
+      context: {
+        outcome,
+        product: 'cloud_generation'
+      }
+    })
+  }
+}

src/platform/telemetry/types.ts

@@ -156,6 +156,11 @@ export interface ExecutionErrorMetadata {
   error?: string
 }
 
+export interface ExecutionOutcomeMetadata {
+  startTime: number
+  outcome: 'success' | 'failure'
+}
+
 /**
  * Execution success metadata
  */
@@ -632,6 +637,7 @@ export interface TelemetryProvider {
 
   // Workflow execution events
   trackWorkflowExecution?(): void
+  trackExecutionOutcome?(metadata: ExecutionOutcomeMetadata): void
   trackExecutionError?(metadata: ExecutionErrorMetadata): void
   trackExecutionSuccess?(metadata: ExecutionSuccessMetadata): void
   trackSharedWorkflowRun?(metadata: SharedWorkflowRunMetadata): void

src/scripts/app.ts

@@ -1662,6 +1662,7 @@ export class ComfyApp {
           // user switches tabs while the request is in flight.
           const queuedWorkflow = useWorkspaceStore().workflow
             .activeWorkflow as ComfyWorkflow
+          const startTime = performance.now()
           const p = await this.graphToPrompt(this.rootGraph)
           const queuedNodes = collectAllNodes(this.rootGraph)
           try {
@@ -1685,6 +1686,7 @@ export class ComfyApp {
                   id: res.prompt_id,
                   nodes: Object.keys(p.output),
                   promptOutput: p.output,
+                  startTime,
                   workflow: queuedWorkflow
                 })
               }

src/services/extensionService.ts

@@ -42,6 +42,11 @@ export const useExtensionService = () => {
     await Promise.all(
       extensions
         .filter((extension) => !extension.includes('extensions/core'))
+        .filter(
+          (extension) =>
+            __DISTRIBUTION__ !== 'cloud' ||
+            extension !== '/extensions/cloud/rum.js'
+        )
         .map(async (ext) => {
           try {
             await import(/* @vite-ignore */ api.fileURL(ext))

src/stores/executionStore.test.ts

@@ -21,6 +21,7 @@ const {
   mockOpenWorkflows,
   mockShowTextPreview,
   mockTrackExecutionError,
+  mockTrackExecutionOutcome,
   mockTrackExecutionSuccess,
   mockTrackSharedWorkflowRun
 } = await vi.hoisted(async () => {
@@ -33,6 +34,7 @@ const {
     mockOpenWorkflows: shallowRef<{ path: string }[]>([]),
     mockShowTextPreview: vi.fn(),
     mockTrackExecutionError: vi.fn(),
+    mockTrackExecutionOutcome: vi.fn(),
     mockTrackExecutionSuccess: vi.fn(),
     mockTrackSharedWorkflowRun: vi.fn()
   }
@@ -92,6 +94,7 @@ vi.mock('@/platform/distribution/types', async () => ({
 vi.mock('@/platform/telemetry', () => ({
   useTelemetry: () => ({
     trackExecutionError: mockTrackExecutionError,
+    trackExecutionOutcome: mockTrackExecutionOutcome,
     trackExecutionSuccess: mockTrackExecutionSuccess,
     trackSharedWorkflowRun: mockTrackSharedWorkflowRun
   })
@@ -614,6 +617,7 @@ describe('useExecutionStore - workflowStatus', () => {
       nodes: ['1'],
       id: jobId,
       promptOutput: { '1': createPromptNode('Node', 'TestNode') },
+      startTime: 42,
       workflow
     })
   }
@@ -631,6 +635,7 @@ describe('useExecutionStore - workflowStatus', () => {
     callStoreJob('job-1', workflowA)
     fireExecutionStart('job-1')
 
+    expect(mockTrackExecutionOutcome).not.toHaveBeenCalled()
     expect(store.getWorkflowStatus(workflowA)).toBe('running')
   })
 
@@ -648,6 +653,11 @@ describe('useExecutionStore - workflowStatus', () => {
     fireExecutionSuccess('job-1')
 
     callStoreJob('job-1', workflowA)
+    expect(mockTrackExecutionOutcome).toHaveBeenCalledOnce()
+    expect(mockTrackExecutionOutcome).toHaveBeenCalledWith({
+      startTime: 42,
+      outcome: 'success'
+    })
     expect(store.getWorkflowStatus(workflowA)).toBe('completed')
   })
 
@@ -656,6 +666,10 @@ describe('useExecutionStore - workflowStatus', () => {
     fireExecutionError('job-1')
 
     callStoreJob('job-1', workflowA)
+    expect(mockTrackExecutionOutcome).toHaveBeenCalledWith({
+      startTime: 42,
+      outcome: 'failure'
+    })
     expect(store.getWorkflowStatus(workflowA)).toBe('failed')
   })
 
@@ -672,6 +686,10 @@ describe('useExecutionStore - workflowStatus', () => {
     fireExecutionStart('job-1')
     fireExecutionSuccess('job-1')
 
+    expect(mockTrackExecutionOutcome).toHaveBeenCalledWith({
+      startTime: 42,
+      outcome: 'success'
+    })
     expect(store.getWorkflowStatus(workflowA)).toBe('completed')
   })
 

src/stores/executionStore.ts

@@ -51,6 +51,7 @@ interface QueuedJob {
    * value is a boolean indicating if the node has been executed.
    */
   nodes: Record<string, boolean>
+  startTime?: number
   /**
    * The workflow that is queued to be executed
    */
@@ -177,6 +178,19 @@ export const useExecutionStore = defineStore('execution', () => {
     mutateStatus((m) => m.set(workflow, status))
   }
 
+  function trackExecutionOutcome(
+    jobId: string,
+    status: WorkflowExecutionStatus
+  ) {
+    if (status === 'running') return
+    const startTime = queuedJobs.value[jobId]?.startTime
+    if (startTime === undefined) return
+    useTelemetry()?.trackExecutionOutcome({
+      startTime,
+      outcome: status === 'completed' ? 'success' : 'failure'
+    })
+  }
+
   function setWorkflowStatus(jobId: string, status: WorkflowExecutionStatus) {
     const workflow = jobIdToWorkflow.get(jobId)
     if (!workflow) {
@@ -184,6 +198,7 @@ export const useExecutionStore = defineStore('execution', () => {
       return
     }
     applyWorkflowStatus(workflow, status)
+    trackExecutionOutcome(jobId, status)
   }
 
   function clearWorkflowStatus(workflow: ComfyWorkflow) {
@@ -718,11 +733,13 @@ export const useExecutionStore = defineStore('execution', () => {
     nodes,
     id,
     promptOutput,
+    startTime,
     workflow
   }: {
     nodes: string[]
     id: JobId
     promptOutput: ComfyApiWorkflow
+    startTime?: number
     workflow: ComfyWorkflow
   }) {
     queuedJobs.value[id] ??= { nodes: {} }
@@ -735,6 +752,7 @@ export const useExecutionStore = defineStore('execution', () => {
       ...queuedJob.nodes
     }
     queuedJob.nodeLookup = buildExecutionNodeLookup(promptOutput)
+    queuedJob.startTime = startTime
     queuedJob.workflow = workflow
     if (workflow) jobIdToWorkflow.set(String(id), workflow)
     queuedJob.shareId = workflow?.shareId
@@ -761,6 +779,7 @@ export const useExecutionStore = defineStore('execution', () => {
     // Don't let a stale 'running' overwrite a terminal status already set.
     if (pending === 'running' && workflowStatus.value.has(workflow)) return
     applyWorkflowStatus(workflow, pending)
+    trackExecutionOutcome(jobId, pending)
   }
 
   // ~0.65 MB at capacity (32 char GUID key + 50 char path value)

vite.config.mts

@@ -334,6 +334,27 @@ export default defineConfig({
     tailwindcss(),
     typegpuPlugin({}),
     comfyAPIPlugin(IS_DEV),
+    {
+      name: 'inject-cloud-rum',
+      apply: 'build',
+      transformIndexHtml(html) {
+        if (DISTRIBUTION !== 'cloud') return html
+
+        return {
+          html,
+          tags: [
+            {
+              tag: 'script',
+              attrs: {
+                type: 'module',
+                src: '/extensions/cloud/rum.js'
+              },
+              injectTo: 'head-prepend'
+            }
+          ]
+        }
+      }
+    },
     // Exclude proprietary ABCROM fonts from non-cloud builds
     {
       name: 'exclude-proprietary-fonts',