Method recordStart

backend/src/node-io/node-io.repository.ts:34–86

* Record node execution start (inputs captured)

(data: {
    runId: string;
    nodeRef: string;
    workflowId?: string;
    organizationId?: string | null;
    componentId: string;
    inputs?: Record<string, unknown>;
    inputsSpilled?: boolean;
    inputsStorageRef?: string | null;
    inputsSize?: number;
  })

   * Record node execution start (inputs captured)
   */
  async recordStart(data: {
    runId: string;
    nodeRef: string;
    workflowId?: string;
    organizationId?: string | null;
    componentId: string;
    inputs?: Record<string, unknown>;
    inputsSpilled?: boolean;
    inputsStorageRef?: string | null;
    inputsSize?: number;
  }): Promise<void> {
    const inputsJson = data.inputs ? JSON.stringify(data.inputs) : null;
    const computedInputsSize = inputsJson ? Buffer.byteLength(inputsJson, 'utf8') : 0;

    // Favor provided spilled info from worker, fallback to local calculation
    const inputsSize = data.inputsSize ?? computedInputsSize;
    const inputsSpilled = data.inputsSpilled ?? inputsSize > KAFKA_SPILL_THRESHOLD_BYTES;
    // Use the storage ref provided by worker (UUID), or generate a path-based fallback
    const inputsStorageRef = data.inputsStorageRef ?? null;

    const insert: NodeIOInsert = {
      runId: data.runId,
      nodeRef: data.nodeRef,
      workflowId: data.workflowId ?? null,
      organizationId: data.organizationId ?? null,
      componentId: data.componentId,
      inputs: inputsSpilled
        ? createSpilledMarker(inputsStorageRef ?? 'unknown', inputsSize)
        : data.inputs,
      inputsSize,
      inputsSpilled,
      inputsStorageRef,
      startedAt: new Date(),
      status: 'running',
    };

    await this.db
      .insert(nodeIOTable)
      .values(insert)
      .onConflictDoUpdate({
        target: [nodeIOTable.runId, nodeIOTable.nodeRef],
        set: {
          ...insert,
          // Only update status to 'running' if it's not already in a terminal state
          status: sql`CASE 
          WHEN ${nodeIOTable.status} IN ('completed', 'failed', 'skipped') 
          THEN ${nodeIOTable.status} 
          ELSE ${insert.status} 
        END`,
          updatedAt: new Date(),
        },
      });
  }

  /**
   * Update node execution with outputs (completion)