refactor: improve worker logging with structured formats and consistent messaging (#532)

# Improved Worker Logging with Structured Context

This PR enhances the logging system for edge workers by implementing structured, context-aware logging throughout the worker lifecycle. The changes:

- Replace generic log messages with context-specific logging methods
- Add structured logging for task execution with detailed context (flow slug, step slug, message ID, run ID, retry information)
- Implement a startup banner with worker identity and flow compilation status
- Add consistent shutdown/deprecation logging to prevent duplicate messages
- Track compilation status during worker startup and include it in logs
- Pass worker identity information to task executors for better log context
- Improve task execution logging with standardized start/complete/fail messages

These changes make logs more consistent and informative, making it easier to track worker and task execution through the system.

Author: jumski

pkgs/edge-worker/src/core/BatchProcessor.ts

@@ -18,15 +18,15 @@ export class BatchProcessor<TMessage extends IMessage> {
   }
 
   async processBatch() {
-    this.logger.verbose('Polling for new batch of messages...');
+    this.logger.polling();
     const messageRecords = await this.poller.poll();
 
     if (this.signal.aborted) {
       this.logger.info('Discarding messageRecords because worker is stopping');
       return;
     }
 
-    this.logger.verbose(`Starting ${messageRecords.length} tasks`);
+    this.logger.taskCount(messageRecords.length);
 
     const startPromises = messageRecords.map((message) =>
       this.executionController.start(message)

pkgs/edge-worker/src/core/Worker.ts

@@ -10,6 +10,7 @@ export class Worker {
   private batchProcessor: IBatchProcessor;
   private sql: postgres.Sql;
   private mainLoopPromise: Promise<void> | undefined;
+  private deprecationLogged = false;
 
   constructor(
     batchProcessor: IBatchProcessor,
@@ -45,7 +46,7 @@ export class Worker {
 
         // Check if deprecated after heartbeat
         if (!this.isMainLoopActive) {
-          this.logger.info('Worker deprecated, exiting main loop');
+          this.logDeprecation();
           break;
         }
 
@@ -71,7 +72,8 @@ export class Worker {
     this.lifecycle.transitionToStopping();
 
     try {
-      this.logger.info('-> Stopped accepting new messages');
+      // Signal deprecation (which includes "Stopped accepting new messages")
+      this.logDeprecation();
       this.abortController.abort();
 
       try {
@@ -84,15 +86,17 @@ export class Worker {
         throw error;
       }
 
-      this.logger.debug('-> Waiting for pending tasks to complete...');
+      // Signal waiting for pending tasks
+      this.logger.shutdown('waiting');
       await this.batchProcessor.awaitCompletion();
-      this.logger.debug('-> Pending tasks completed!');
 
       this.lifecycle.acknowledgeStop();
 
       this.logger.debug('-> Closing SQL connection...');
       await this.sql.end();
-      this.logger.debug('-> SQL connection closed!');
+
+      // Signal graceful stop complete
+      this.logger.shutdown('stopped');
     } catch (error) {
       this.logger.debug(`Error during worker stop: ${error}`);
       throw error;
@@ -103,6 +107,17 @@ export class Worker {
     return this.lifecycle.edgeFunctionName;
   }
 
+  /**
+   * Log deprecation message only once (prevents duplicate logs when deprecation
+   * is detected in heartbeat and then stop() is called)
+   */
+  private logDeprecation(): void {
+    if (!this.deprecationLogged) {
+      this.logger.shutdown('deprecating');
+      this.deprecationLogged = true;
+    }
+  }
+
   /**
    * Returns true if worker state is Running and worker was not stopped
    */

pkgs/edge-worker/src/flow/FlowWorkerLifecycle.ts

@@ -1,6 +1,6 @@
 import type { Queries } from '../core/Queries.js';
 import type { ILifecycle, WorkerBootstrap, WorkerRow } from '../core/types.js';
-import type { Logger } from '../platform/types.js';
+import type { Logger, StartupContext } from '../platform/types.js';
 import { States, WorkerState } from '../core/WorkerState.js';
 import type { AnyFlow } from '@pgflow/dsl';
 import { extractFlowShape } from '@pgflow/dsl';
@@ -11,6 +11,11 @@ export interface FlowLifecycleConfig {
   ensureCompiledOnStartup?: boolean;
 }
 
+/**
+ * Compilation status returned by ensureFlowCompiled
+ */
+type CompilationStatus = 'compiled' | 'verified' | 'recompiled' | 'mismatch';
+
 /**
  * A specialized WorkerLifecycle for Flow-based workers that is aware of the Flow's step types
  */
@@ -22,6 +27,7 @@ export class FlowWorkerLifecycle<TFlow extends AnyFlow> implements ILifecycle {
   private flow: TFlow;
   // TODO: Temporary field for supplier pattern until we refactor initialization
   private _workerId?: string;
+  private _edgeFunctionName?: string;
   private heartbeatInterval: number;
   private lastHeartbeat = 0;
   private ensureCompiledOnStartup: boolean;
@@ -38,19 +44,24 @@ export class FlowWorkerLifecycle<TFlow extends AnyFlow> implements ILifecycle {
   async acknowledgeStart(workerBootstrap: WorkerBootstrap): Promise<void> {
     this.workerState.transitionTo(States.Starting);
 
-    // Store workerId for supplier pattern
+    // Store workerId and edgeFunctionName for supplier pattern
     this._workerId = workerBootstrap.workerId;
+    this._edgeFunctionName = workerBootstrap.edgeFunctionName;
 
     // Register this edge function for monitoring by ensure_workers() cron.
     await this.queries.trackWorkerFunction(workerBootstrap.edgeFunctionName);
 
     // Compile/verify flow as part of Starting (before registering worker)
+    let compilationStatus: CompilationStatus = 'verified';
     if (this.ensureCompiledOnStartup) {
-      await this.ensureFlowCompiled();
+      compilationStatus = await this.ensureFlowCompiled();
     } else {
       this.logger.info(`Skipping compilation check for flow '${this.flow.slug}' (ensureCompiledOnStartup=false)`);
     }
 
+    // Log startup banner with compilation status
+    this.logStartupBanner(compilationStatus);
+
     // Only register worker after successful compilation
     this.workerRow = await this.queries.onWorkerStarted({
       queueName: this.queueName,
@@ -60,9 +71,7 @@ export class FlowWorkerLifecycle<TFlow extends AnyFlow> implements ILifecycle {
     this.workerState.transitionTo(States.Running);
   }
 
-  private async ensureFlowCompiled(): Promise<void> {
-    this.logger.info(`Compiling flow '${this.flow.slug}'...`);
-
+  private async ensureFlowCompiled(): Promise<CompilationStatus> {
     const shape = extractFlowShape(this.flow);
 
     const result = await this.queries.ensureFlowCompiled(
@@ -74,7 +83,26 @@ export class FlowWorkerLifecycle<TFlow extends AnyFlow> implements ILifecycle {
       throw new FlowShapeMismatchError(this.flow.slug, result.differences);
     }
 
-    this.logger.info(`Flow '${this.flow.slug}' compilation complete: ${result.status}`);
+    return result.status;
+  }
+
+  /**
+   * Log the startup banner with worker and flow information
+   */
+  private logStartupBanner(compilationStatus: CompilationStatus): void {
+    const startupContext: StartupContext = {
+      workerName: this._edgeFunctionName ?? 'unknown',
+      workerId: this._workerId ?? 'unknown',
+      queueName: this.queueName,
+      flows: [
+        {
+          flowSlug: this.flow.slug,
+          compilationStatus,
+        },
+      ],
+    };
+
+    this.logger.startupBanner(startupContext);
   }
 
   acknowledgeStop() {
@@ -95,7 +123,7 @@ export class FlowWorkerLifecycle<TFlow extends AnyFlow> implements ILifecycle {
   }
 
   get edgeFunctionName() {
-    return this.workerRow?.function_name;
+    return this._edgeFunctionName ?? this.workerRow?.function_name;
   }
 
   get queueName() {

pkgs/edge-worker/src/flow/StepTaskExecutor.ts

@@ -1,7 +1,7 @@
 import type { AnyFlow } from '@pgflow/dsl';
 import type { IPgflowClient } from './types.js';
 import type { IExecutor } from '../core/types.js';
-import type { Logger } from '../platform/types.js';
+import type { Logger, TaskLogContext } from '../platform/types.js';
 import type { StepTaskHandlerContext } from '../core/context.js';
 
 class AbortError extends Error {
@@ -11,6 +11,15 @@ class AbortError extends Error {
   }
 }
 
+/**
+ * Configuration for worker identity in logging
+ */
+export interface WorkerIdentity {
+  workerId: string;
+  workerName: string;
+  queueName: string;
+}
+
 /**
  * An executor that processes step tasks using an IPgflowClient
  * with strong typing for the flow's step handlers
@@ -23,7 +32,8 @@ export class StepTaskExecutor<TFlow extends AnyFlow, TContext extends StepTaskHa
     private readonly adapter: IPgflowClient<TFlow>,
     private readonly signal: AbortSignal,
     logger: Logger,
-    private readonly context: TContext
+    private readonly context: TContext,
+    private readonly workerIdentity: WorkerIdentity
   ) {
     this.logger = logger;
   }
@@ -41,7 +51,34 @@ export class StepTaskExecutor<TFlow extends AnyFlow, TContext extends StepTaskHa
     return this.stepTask.msg_id;
   }
 
+  /**
+   * Build TaskLogContext for structured logging
+   */
+  private buildLogContext(): TaskLogContext {
+    // Note: read_ct includes the initial read (1-indexed), so read_ct=1 is the first attempt,
+    // read_ct=2 is the first retry, etc.
+    const retryAttempt = this.rawMessage.read_ct;
+    const stepDef = this.flow.getStepDefinition(this.stepTask.step_slug);
+    // maxAttempts includes the initial attempt, so maxRetries = maxAttempts - 1
+    const maxRetries = stepDef?.options?.maxAttempts ? stepDef.options.maxAttempts - 1 : undefined;
+
+    return {
+      flowSlug: this.stepTask.flow_slug,
+      stepSlug: this.stepTask.step_slug,
+      // Convert to string for logging (msg_id is number from PostgreSQL bigint)
+      msgId: String(this.stepTask.msg_id),
+      runId: this.stepTask.run_id,
+      workerId: this.workerIdentity.workerId,
+      workerName: this.workerIdentity.workerName,
+      queueName: this.workerIdentity.queueName,
+      retryAttempt,
+      maxRetries,
+    };
+  }
+
   async execute(): Promise<void> {
+    const logContext = this.buildLogContext();
+
     try {
       if (this.signal.aborted) {
         throw new AbortError();
@@ -51,9 +88,9 @@ export class StepTaskExecutor<TFlow extends AnyFlow, TContext extends StepTaskHa
       this.signal.throwIfAborted();
 
       const stepSlug = this.stepTask.step_slug;
-      this.logger.debug(
-        `Executing step task ${this.msgId} for step ${stepSlug}`
-      );
+
+      // Log task started at debug level
+      this.logger.taskStarted(logContext);
 
       // Get the step handler from the flow with proper typing
       const stepDef = this.flow.getStepDefinition(stepSlug);
@@ -71,14 +108,11 @@ export class StepTaskExecutor<TFlow extends AnyFlow, TContext extends StepTaskHa
 
       const durationMs = Date.now() - startTime;
 
-      this.logger.verbose(
-        `step task ${this.msgId} completed in ${durationMs}ms`
-      );
+      // Log task completed at verbose level
+      this.logger.taskCompleted(logContext, durationMs);
       await this.adapter.completeTask(this.stepTask, result);
-
-      this.logger.debug(`step task ${this.msgId} marked as complete`);
     } catch (error) {
-      await this.handleExecutionError(error);
+      await this.handleExecutionError(error, logContext);
     }
   }
 
@@ -90,15 +124,15 @@ export class StepTaskExecutor<TFlow extends AnyFlow, TContext extends StepTaskHa
    *
    * Otherwise, it marks the task as failed.
    */
-  private async handleExecutionError(error: unknown) {
+  private async handleExecutionError(error: unknown, logContext: TaskLogContext) {
     if (error instanceof Error && error.name === 'AbortError') {
       this.logger.debug(`Aborted execution for step task ${this.msgId}`);
       // Do not mark as failed - the worker was aborted and stopping,
       // the task will be picked up by another worker later
     } else {
-      this.logger.verbose(
-        `step task ${this.msgId} failed with error: ${error}`
-      );
+      // Log task failed at verbose level using structured method
+      const errorObj = error instanceof Error ? error : new Error(String(error));
+      this.logger.taskFailed(logContext, errorObj);
       await this.adapter.failTask(this.stepTask, error);
     }
   }

pkgs/edge-worker/src/flow/createFlowWorker.ts

@@ -1,7 +1,7 @@
 import type { AnyFlow, FlowContext } from '@pgflow/dsl';
 import { ExecutionController } from '../core/ExecutionController.js';
 import { StepTaskPoller, type StepTaskPollerConfig } from './StepTaskPoller.js';
-import { StepTaskExecutor } from './StepTaskExecutor.js';
+import { StepTaskExecutor, type WorkerIdentity } from './StepTaskExecutor.js';
 import { PgflowSqlClient } from '@pgflow/core';
 import { Queries } from '../core/Queries.js';
 import type { IExecutor } from '../core/types.js';
@@ -116,6 +116,8 @@ export function createFlowWorker<TFlow extends AnyFlow, TResources extends Recor
   );
 
   // Create executor factory with proper typing
+  // Note: This factory is only called during task execution (after acknowledgeStart completes),
+  // so lifecycle.workerId and lifecycle.edgeFunctionName are guaranteed to be set.
   const executorFactory = (
     taskWithMessage: StepTaskWithMessage<TFlow>,
     signal: AbortSignal
@@ -135,14 +137,23 @@ export function createFlowWorker<TFlow extends AnyFlow, TResources extends Recor
       ...platformAdapter.platformResources
     };
 
+    // Build worker identity for structured logging
+    // Safe to access here because factory is only called after acknowledgeStart()
+    const workerIdentity: WorkerIdentity = {
+      workerId: lifecycle.workerId,
+      workerName: lifecycle.edgeFunctionName ?? 'unknown',
+      queueName: queueName,
+    };
+
     // Type assertion: FlowContext & TResources is compatible with StepTaskHandlerContext<TFlow>
     // at runtime, but TypeScript needs help due to generic type variance
     return new StepTaskExecutor<TFlow>(
       flow,
       pgflowAdapter,
       signal,
       createLogger('StepTaskExecutor'),
-      context as StepTaskHandlerContext<TFlow>
+      context as StepTaskHandlerContext<TFlow>,
+      workerIdentity
     );
   };
 

pkgs/edge-worker/src/platform/SupabasePlatformAdapter.ts

@@ -223,6 +223,7 @@ export class SupabasePlatformAdapter implements PlatformAdapter<SupabaseResource
         const workerId = this.validatedEnv.SB_EXECUTION_ID;
 
         this.loggingFactory.setWorkerId(workerId);
+        this.loggingFactory.setWorkerName(this.edgeFunctionName);
 
         // Create the worker using the factory function and the logger
         this.worker = createWorkerFn(this.loggingFactory.createLogger);