Structure debug logs and retry workflow registration up to 4 times

apps/wss/src/server.ts

@@ -8,6 +8,8 @@ import {
 import {
   CommandCatalog,
   InternalApiClient,
+  InternalResponseError,
+  RegisterWorkflowResponse,
   TriggerCatalog,
   triggerCatalog,
   ZodPublisher,
@@ -22,6 +24,9 @@ import { env } from "./env";
 import { pulsarClient } from "./pulsarClient";
 import { WorkflowRunController } from "./runController";
 
+const MAX_RETRY_ATTEMPTS = 4;
+const RETRY_DELAY_MS = 888;
+
 export class TriggerServer {
   #connection?: TriggerServerConnection;
   #serverRPC?: ZodRPC<typeof HostRPCSchema, typeof ServerRPCSchema>;
@@ -423,17 +428,7 @@ export class TriggerServer {
 
     try {
       // register the workflow with the platform
-      const response = await this.#apiClient.registerWorkflow({
-        id: data.workflowId,
-        name: data.workflowName,
-        trigger: data.trigger,
-        package: {
-          name: data.packageName,
-          version: data.packageVersion,
-        },
-        triggerTTL: data.triggerTTL,
-        metadata: data.metadata ? JSON.stringify(data.metadata) : undefined,
-      });
+      const response = await this.#registerWorkflow(data);
 
       this.#workflowId = response.workflow.id;
 
@@ -448,20 +443,12 @@ export class TriggerServer {
           subscriptionType: "Shared",
           subscriptionInitialPosition: "Latest",
         },
+        filter: {
+          "x-workflow-id": this.#workflowId,
+          "x-api-key": this.#apiKey,
+        },
         handlers: {
           TRIGGER_WORKFLOW: async (id, data, properties, messageAttributes) => {
-            // If the API keys don't match, then we should ignore it
-            // This ensures the workflow is triggered for the correct environment
-            if (properties["x-api-key"] !== this.#apiKey) {
-              return true;
-            }
-
-            // If the workflow id is not the same as the workflow id
-            // that we are listening for, then we should ignore it
-            if (properties["x-workflow-id"] !== this.#workflowId) {
-              return true;
-            }
-
             if (!this.#serverRPC) {
               throw new Error(
                 "Cannot trigger workflow without an RPC connection"
@@ -582,6 +569,41 @@ export class TriggerServer {
     }
   }
 
+  async #registerWorkflow(
+    data: z.infer<(typeof ServerRPCSchema)["INITIALIZE_HOST_V2"]["request"]>,
+    attempt: number = 0
+  ): Promise<RegisterWorkflowResponse> {
+    try {
+      return await this.#apiClient.registerWorkflow({
+        id: data.workflowId,
+        name: data.workflowName,
+        trigger: data.trigger,
+        package: {
+          name: data.packageName,
+          version: data.packageVersion,
+        },
+        triggerTTL: data.triggerTTL,
+        metadata: data.metadata ? JSON.stringify(data.metadata) : undefined,
+      });
+    } catch (error) {
+      if (error instanceof InternalResponseError && error.retryable) {
+        // If error.status is a retryable status code, and we haven't exceeded the max number of attempts, retry
+        if (attempt < MAX_RETRY_ATTEMPTS) {
+          this.#logger.debug(
+            `Failed to register workflow, retrying in ${RETRY_DELAY_MS}ms...`,
+            error
+          );
+
+          await sleep(RETRY_DELAY_MS);
+
+          return this.#registerWorkflow(data, attempt + 1);
+        }
+      }
+
+      throw error;
+    }
+  }
+
   async #closePubSub() {
     this.#logger.debug(
       "Closing run controllers...",

packages/internal-bridge/src/logger.ts

@@ -71,13 +71,18 @@ export class Logger {
     );
   }
 
-  debug(...args: any[]) {
+  debug(message: string, ...args: any[]) {
     if (this.#level < 5) return;
 
-    console.debug(
-      `[${formattedDateTime()}] ${this.#formatName()} `,
-      ...[...args, ...this.#formatTags()]
-    );
+    const structuredLog = {
+      timestamp: formattedDateTime(),
+      name: this.#name,
+      tags: this.#tags,
+      message,
+      args: structureArgs(args),
+    };
+
+    console.debug(JSON.stringify(structuredLog));
   }
 
   #formatName() {
@@ -114,3 +119,16 @@ function formattedDateTime() {
 
   return `${formattedHours}:${formattedMinutes}:${formattedSeconds}.${formattedMilliseconds}`;
 }
+
+// If args is has a single item that is an object, return that object
+function structureArgs(args: any[]) {
+  if (args.length === 0) {
+    return;
+  }
+
+  if (args.length === 1 && typeof args[0] === "object") {
+    return args[0];
+  }
+
+  return args;
+}

packages/internal-platform/src/api/client.ts

@@ -3,6 +3,24 @@ import { UpdateWorkflowRun, RegisteredWorkflow } from "../schemas";
 import fetch from "node-fetch";
 import { Logger } from "../logger";
 
+const RETRYABLE_STATUS_CODES = [500, 502, 503, 504];
+
+// Response error with status code
+export class InternalResponseError extends Error {
+  retryable: boolean;
+
+  constructor(retryable: boolean, message: string) {
+    super(message);
+
+    this.retryable = retryable;
+  }
+}
+
+// export type of InternalApiClient.registerWorkflow
+export type RegisterWorkflowResponse = Awaited<
+  ReturnType<InstanceType<typeof InternalApiClient>["registerWorkflow"]>
+>;
+
 export class InternalApiClient {
   #apiKey: string;
   #baseUrl: string;
@@ -104,8 +122,9 @@ export class InternalApiClient {
       throw new Error(body.error);
     }
 
-    throw new Error(
-      `[${response.status}] Something went wrong: ${response.statusText}`
+    throw new InternalResponseError(
+      RETRYABLE_STATUS_CODES.includes(response.status),
+      response.statusText
     );
   }
 

packages/internal-platform/src/logger.ts

@@ -39,10 +39,17 @@ export class Logger {
     console.info(`[${formattedDateTime()}] [${this.#name}] `, ...args);
   }
 
-  debug(...args: any[]) {
+  debug(message: string, ...args: any[]) {
     if (this.#level < 4) return;
 
-    console.debug(`[${formattedDateTime()}] [${this.#name}] `, ...args);
+    const structuredLog = {
+      timestamp: formattedDateTime(),
+      name: this.#name,
+      message,
+      args: structureArgs(args),
+    };
+
+    console.debug(JSON.stringify(structuredLog));
   }
 }
 
@@ -67,3 +74,16 @@ function formattedDateTime() {
 
   return `${formattedHours}:${formattedMinutes}:${formattedSeconds}.${formattedMilliseconds}`;
 }
+
+// If args is has a single item that is an object, return that object
+function structureArgs(args: any[]) {
+  if (args.length === 0) {
+    return;
+  }
+
+  if (args.length === 1 && typeof args[0] === "object") {
+    return args[0];
+  }
+
+  return args;
+}