Fix worker logs not getting flushed in Core log level set to ERROR

packages/test/src/helpers.ts

@@ -141,6 +141,7 @@ export const bundlerOptions = {
     'timers',
     'timers/promises',
     require.resolve('./activities'),
+    require.resolve('./mock-native-worker'),
   ],
 };
 

packages/test/src/test-runtime.ts

@@ -2,14 +2,15 @@
  * Test the lifecycle of the Runtime singleton.
  * Tests run serially because Runtime is a singleton.
  */
-import test from 'ava';
 import { v4 as uuid4 } from 'uuid';
 import asyncRetry from 'async-retry';
-import { Runtime, DefaultLogger, LogEntry, makeTelemetryFilterString } from '@temporalio/worker';
+import { Runtime, DefaultLogger, LogEntry, makeTelemetryFilterString, NativeConnection } from '@temporalio/worker';
 import { Client, WorkflowClient } from '@temporalio/client';
+import * as wf from '@temporalio/workflow';
 import { defaultOptions } from './mock-native-worker';
 import * as workflows from './workflows';
-import { RUN_INTEGRATION_TESTS, Worker } from './helpers';
+import { RUN_INTEGRATION_TESTS, Worker, test } from './helpers';
+import { createTestWorkflowBundle } from './helpers-integration';
 
 if (RUN_INTEGRATION_TESTS) {
   test.serial('Runtime can be created and disposed', async (t) => {
@@ -113,4 +114,53 @@ if (RUN_INTEGRATION_TESTS) {
       await Runtime.instance().shutdown();
     }
   });
+
+  test.serial(`NativeLogCollector: Buffered logs are periodically flushed even if Core isn't flushing`, async (t) => {
+    const logEntries: LogEntry[] = [];
+
+    const runtime = Runtime.install({
+      logger: new DefaultLogger('DEBUG', (entry) => logEntries.push(entry)),
+      telemetryOptions: {
+        // Sets native logger to ERROR level, so that it never flushes
+        logging: { forward: {}, filter: { core: 'ERROR', other: 'ERROR' } },
+      },
+    });
+    const bufferedLogger = runtime.logger;
+
+    // Hold on to a connection to prevent implicit shutdown of the runtime before we print 'final log'
+    const connection = await NativeConnection.connect();
+
+    try {
+      const taskQueue = `runtime-native-log-collector-preriodically-flushed-${uuid4()}`;
+      const worker = await Worker.create({
+        ...defaultOptions,
+        connection,
+        taskQueue,
+        workflowBundle: await createTestWorkflowBundle({ workflowsPath: __filename }),
+      });
+
+      await worker.runUntil(async () => {
+        await new Client().workflow.execute(log5Times, { taskQueue, workflowId: uuid4() });
+      });
+      t.true(logEntries.some((x) => x.message.startsWith('workflow log ')));
+
+      // This one will get buffered
+      bufferedLogger.info('final log');
+      t.false(logEntries.some((x) => x.message.startsWith('final log')));
+    } finally {
+      await connection.close();
+      await runtime.shutdown();
+    }
+
+    // Assert all log messages have been flushed
+    t.is(logEntries.filter((x) => x.message.startsWith('workflow log ')).length, 5);
+    t.is(logEntries.filter((x) => x.message.startsWith('final log')).length, 1);
+  });
+}
+
+export async function log5Times(): Promise<void> {
+  for (let i = 0; i < 5; i++) {
+    wf.log.info(`workflow log ${i}`);
+    await wf.sleep(1);
+  }
 }

packages/worker/package.json

@@ -23,7 +23,7 @@
     "@temporalio/proto": "file:../proto",
     "@temporalio/workflow": "file:../workflow",
     "abort-controller": "^3.0.0",
-    "heap-js": "^2.3.0",
+    "heap-js": "^2.6.0",
     "memfs": "^4.6.0",
     "nexus-rpc": "^0.0.1",
     "proto3-json-serializer": "^2.0.0",

packages/worker/src/logger.ts

@@ -118,6 +118,7 @@ export function hasColorSupport(logger: Logger): boolean {
 
 export interface FlushableLogger extends Logger {
   flush(): void;
+  close?(): void;
 }
 
 export function isFlushableLogger(logger: Logger): logger is FlushableLogger {

packages/worker/src/runtime-logger.ts

@@ -1,7 +1,7 @@
 import { Heap } from 'heap-js';
 import { SdkComponent } from '@temporalio/common';
 import { native } from '@temporalio/core-bridge';
-import { DefaultLogger, LogEntry, Logger, LogTimestamp } from './logger';
+import { DefaultLogger, FlushableLogger, LogEntry, Logger, LogTimestamp } from './logger';
 
 /**
  * A log collector that accepts log entries either through the TS `Logger` interface (e.g. used by
@@ -25,10 +25,60 @@ export class NativeLogCollector {
 
   protected buffer = new Heap<LogEntry>((a, b) => Number(a.timestampNanos - b.timestampNanos));
 
+  /**
+   * A timer that periodically flushes the buffer to the downstream logger.
+   */
+  protected flushIntervalTimer: NodeJS.Timeout;
+
+  /**
+   * The minimum time an entry should be buffered before getting flushed.
+   *
+   * Increasing this value allows the buffer to do a better job of correctly reordering messages
+   * emitted from different sources (notably from Workflow executions through Sinks, and from Core)
+   * based on their absolute timestamps, but also increases latency of logs.
+   *
+   * The minimum buffer time requirement only applies as long as the buffer is not full. Once the
+   * buffer reaches its maximum size, older messages are unconditionally flushed, to prevent
+   * unbounded growth of the buffer.
+   *
+   * TODO(JWH): Is 100ms a reasonable compromise? That might seem a little high on latency, but to
+   *            be useful, that value needs to exceed the time it typically takes to process
+   *            Workflow Activations, let's say above the expected P90, but that's highly variable
+   *            across our user base, and we don't really have field data anyway.
+   *            We can revisit depending on user feedback.
+   */
+  protected readonly minBufferTimeMs = 100;
+
+  /**
+   * Interval between flush passes checking for expired messages.
+   *
+   * This really is redundant, since Core itself is expected to flush its buffer every 10 ms, and
+   * we're checking for expired messages when it does. However, Core will only flush if it has
+   * accumulated at least one message; when Core's log level is set to WARN or higher, it may be
+   * many seconds, and even minutes, between Core's log messages, resulting in very rare flush
+   * from that end, which cause considerable delay on flushing log messages from other sources.
+   */
+  protected readonly flushPassIntervalMs = 100;
+
+  /**
+   * The maximum number of log messages to buffer before flushing.
+   *
+   * When the buffer reaches this limit, older messages are unconditionally flushed (i.e. without
+   * regard to the minimum buffer time requirement), to prevent unbounded growth of the buffer.
+   */
+  protected readonly maxBufferSize = 2000;
+
   constructor(downstream: Logger) {
-    this.logger = new DefaultLogger('TRACE', (entry) => this.buffer.add(entry));
+    this.logger = new DefaultLogger('TRACE', this.appendOne.bind(this));
+    (this.logger as FlushableLogger).flush = this.flush.bind(this);
+    (this.logger as FlushableLogger).close = this.close.bind(this);
+
     this.downstream = downstream;
     this.receive = this.receive.bind(this);
+
+    // Flush the buffer every so often.
+    // Unref'ed so that it doesn't prevent the process from exiting.
+    this.flushIntervalTimer = setInterval(this.flushExpired.bind(this), this.flushPassIntervalMs).unref();
   }
 
   /**
@@ -44,14 +94,20 @@ export class NativeLogCollector {
           this.buffer.add(log);
         }
       }
-      this.flush();
+      this.flushUnconditionally();
+      this.flushExpired();
     } catch (_e) {
       // We're not allowed to throw from here, and conversion errors have already been handled in
       // convertFromNativeLogEntry(), so an error at this point almost certainly indicates a problem
       // with the downstream logger. Just swallow it, there's really nothing else we can do.
     }
   }
 
+  private appendOne(entry: LogEntry): void {
+    this.buffer.add(entry);
+    this.flushUnconditionally();
+  }
+
   private convertFromNativeLogEntry(entry: native.JsonString<native.LogEntry>): LogEntry | undefined {
     try {
       const log = JSON.parse(entry) as native.LogEntry;
@@ -78,15 +134,49 @@ export class NativeLogCollector {
   }
 
   /**
-   * Flush all buffered logs into the logger supplied to the constructor/
+   * Flush messages that have exceeded their required minimal buffering time.
    */
-  flush(): void {
-    for (const entry of this.buffer) {
+  private flushExpired(): void {
+    const threadholdTimeNanos = BigInt(Date.now() - this.minBufferTimeMs) * 1_000_000n;
+    for (;;) {
+      const entry = this.buffer.peek();
+      if (!entry || entry.timestampNanos > threadholdTimeNanos) break;
+      this.buffer.pop();
+
       this.downstream.log(entry.level, entry.message, {
         [LogTimestamp]: entry.timestampNanos,
         ...entry.meta,
       });
     }
-    this.buffer.clear();
+  }
+
+  /**
+   * Flush messages without regard to the time threshold, up to a given number of messages.
+   *
+   * If no limit is provided, flushes messages in excess to the maximum buffer size.
+   */
+  private flushUnconditionally(maxFlushCount?: number): void {
+    if (maxFlushCount === undefined) {
+      maxFlushCount = this.buffer.size() - this.maxBufferSize;
+    }
+
+    while (maxFlushCount-- > 0) {
+      const entry = this.buffer.pop();
+      if (!entry) break;
+
+      this.downstream.log(entry.level, entry.message, {
+        [LogTimestamp]: entry.timestampNanos,
+        ...entry.meta,
+      });
+    }
+  }
+
+  public flush(): void {
+    this.flushUnconditionally(Number.MAX_SAFE_INTEGER);
+  }
+
+  public close(): void {
+    this.flush();
+    clearInterval(this.flushIntervalTimer);
   }
 }

packages/worker/src/runtime-options.ts

@@ -69,7 +69,7 @@ export interface TelemetryOptions {
    * ### Log Forwarding
    *
    * By default, logs emitted by the native side of the SDK are printed directly to the console,
-   * _independently of `RuntimeOptions.logger`_. To enable forwarding of those logs messages to the
+   * _independently of `RuntimeOptions.logger`_. To enable forwarding of those log messages to the
    * TS side logger, add the `forward` property to the `logging` object.
    *
    * For example:
@@ -86,10 +86,14 @@ export interface TelemetryOptions {
    * });
    * ```
    *
-   * Note that forwarded log messages are internally throttled/buffered for a few milliseconds to
-   * reduce overhead incurred by Rust-to-JS calls. In rare cases, this may result in log messages
-   * appearing out of order by a few milliseconds. Users are discouraged from using log forwarding
-   * with verboseness sets to `DEBUG` or `TRACE`.
+   * Note that when log forwarding is enabled, all log messages sent to the runtime logger are
+   * internally buffered for 100 ms, to allow global sorting of messages from different sources
+   * based on their absolute timestamps. This helps reduce incoherencies in the order of messages,
+   * notably those emitted through the Workflow Logging API vs those emitted through Core.
+   *
+   * However, in some situations, log messages may still appear out of order, e.g. when a Workflow
+   * Activation takes longer than 100ms to complete or when log flow exceeds the buffer's capacity
+   * (2000 messages).
    */
   logging?: LogExporterConfig;
 

packages/worker/src/runtime.ts

@@ -280,7 +280,9 @@ export class Runtime {
       this.teardownShutdownHook();
       // FIXME(JWH): I think we no longer need this, but will have to thoroughly validate.
       native.runtimeShutdown(this.native);
-      this.flushLogs();
+      if (isFlushableLogger(this.logger)) {
+        this.logger.close?.();
+      }
     } finally {
       delete (this as any).native;
     }