feat: Control async context tracking with timeout

Adds the ability to control async tracking via the
async_tracking_timeout field in the configuration file and
the APPMAP_ASYNC_TRACKING_TIMEOUT environment variable. The
environment variable takes precedence over the config file
value. If neither is set, a default timeout of 3000
milliseconds is applied. Setting the value to 0 disables
async tracking entirely.

src/Recording.ts

@@ -14,7 +14,7 @@ import { getDefaultMetadata } from "./metadata";
 import { parameter } from "./parameter";
 import type { FunctionInfo } from "./registry";
 import compactObject from "./util/compactObject";
-import { getTime } from "./util/getTime";
+import { getTime, getTimeInMilliseconds } from "./util/getTime";
 import { warn } from "./message";
 
 export default class Recording {
@@ -243,26 +243,42 @@ export default class Recording {
       warn("event emitted while recording not running");
       return;
     }
-    if (Recording.buffering) {
+    // If the current buffer is alive more than allowed pass its events
+    // to the stream and clear it recursively.
+    if (Recording.buffering && !Recording.buffer.disposed && config().asyncTrackingTimeout != 0) {
+      const elapsed = getTimeInMilliseconds() - Recording.buffer.createdAt;
+      if (elapsed >= config().asyncTrackingTimeout)
+        Recording.passEventsAndClearBuffer(Recording.buffer);
+    }
+
+    if (Recording.buffering && !Recording.buffer.disposed && config().asyncTrackingTimeout != 0) {
       this.bufferedEvents.set(event.id, event);
-      Recording.buffer.push({ event, owner: new WeakRef(this) });
+      Recording.buffer.items.push({ event, owner: new WeakRef(this) });
     } else this.stream.push(event);
   }
 
-  private static rootBuffer: EventBuffer = [];
+  private static _rootBuffer: EventBuffer | undefined;
+  private static get rootBuffer(): EventBuffer {
+    if (Recording._rootBuffer == undefined)
+      Recording._rootBuffer = { items: [], disposed: false, createdAt: getTimeInMilliseconds() };
+
+    return Recording._rootBuffer;
+  }
+
   private static asyncStorage = new AsyncLocalStorage<EventBuffer>();
 
   private static get buffering(): boolean {
-    return Recording.rootBuffer.length > 0;
+    return Recording.rootBuffer.items.length > 0;
   }
 
   private static get buffer(): EventBuffer {
     return Recording.asyncStorage.getStore() ?? Recording.rootBuffer;
   }
 
   public static fork<T>(fun: () => T): T {
-    const forked: EventBuffer = [];
-    Recording.buffer.push(forked);
+    const forked: EventBuffer = { items: [], disposed: false, createdAt: getTimeInMilliseconds() };
+    Recording.buffer.items.push(forked);
+
     return Recording.asyncStorage.run(forked, fun);
   }
 
@@ -275,34 +291,57 @@ export default class Recording {
     return Recording.asyncStorage.getStore();
   }
 
+  private static passEventsAndClearBuffer(buffer: EventBuffer) {
+    for (const event of buffer.items) {
+      if (isEventBuffer(event)) this.passEventsAndClearBuffer(event);
+      else {
+        const recording = event?.owner.deref();
+        if (event && recording) {
+          recording.stream.push(event.event);
+          recording.bufferedEvents.delete(event.event.id);
+        }
+      }
+    }
+    buffer.disposed = true;
+    buffer.items = [];
+  }
+
   readAppMap(): AppMap.AppMap {
     assert(!this.running);
     return JSON.parse(readFileSync(this.path, "utf8")) as AppMap.AppMap;
   }
 
   private passEvents(stream: AppMapStream, buffer: EventBuffer) {
-    for (const event of buffer) {
-      if (Array.isArray(event)) this.passEvents(stream, event);
+    for (const event of buffer.items) {
+      if (isEventBuffer(event)) this.passEvents(stream, event);
       else if (event?.owner.deref() == this) stream.push(event.event);
     }
   }
 
   private disposeBufferedEvents(buffer: EventBuffer) {
-    for (let i = 0; i < buffer.length; i++) {
-      const event = buffer[i];
-      if (Array.isArray(event)) this.disposeBufferedEvents(event);
-      else if (event?.owner.deref() == this) buffer[i] = null;
+    for (let i = 0; i < buffer.items.length; i++) {
+      const event = buffer.items[i];
+      if (isEventBuffer(event)) this.disposeBufferedEvents(event);
+      else if (event?.owner.deref() == this) buffer.items[i] = null;
     }
   }
 }
 
+function isEventBuffer(obj: EventOrBuffer): obj is EventBuffer {
+  return obj != null && "items" in obj;
+}
+
 interface EventWithOwner {
   owner: WeakRef<Recording>;
   event: AppMap.Event;
 }
 
-type EventOrBuffer = EventWithOwner | null | EventOrBuffer[];
-export type EventBuffer = EventOrBuffer[];
+type EventOrBuffer = EventWithOwner | null | EventBuffer;
+export interface EventBuffer {
+  items: EventOrBuffer[];
+  disposed: boolean;
+  createdAt: number;
+}
 
 export const writtenAppMaps: string[] = [];
 

src/config.ts

@@ -15,6 +15,9 @@ import tryOr from "./util/tryOr";
 const responseBodyMaxLengthDefault = 10000;
 const kResponseBodyMaxLengthEnvar = "APPMAP_RESPONSE_BODY_MAX_LENGTH";
 
+const asyncTrackingTimeoutDefault = 3000;
+const kAsyncTrackingTimeoutEnvar = "APPMAP_ASYNC_TRACKING_TIMEOUT";
+
 export class Config {
   public readonly relativeAppmapDir: string;
   public readonly appName: string;
@@ -24,6 +27,8 @@ export class Config {
   public readonly packages: PackageMatcher;
   public readonly responseBodyMaxLength: number;
   public readonly language: string;
+  // If it's 0 then no async tracking.
+  public readonly asyncTrackingTimeout: number;
 
   private readonly document?: Document;
   private migrationPending = false;
@@ -59,16 +64,26 @@ export class Config {
       ],
     );
 
-    let envResponseBodyMaxLength: number | undefined;
-    if (process.env[kResponseBodyMaxLengthEnvar] != undefined) {
-      const value = parseInt(process.env[kResponseBodyMaxLengthEnvar]);
-      envResponseBodyMaxLength = value >= 0 ? value : undefined;
-      if (envResponseBodyMaxLength == undefined)
-        warn(`Environment variable ${kResponseBodyMaxLengthEnvar} must be a non-negative integer.`);
+    function getNonNegativeIntegerEnvVarValue(enVarName: string) {
+      let result: number | undefined;
+      if (process.env[enVarName] != undefined) {
+        const value = parseInt(process.env[enVarName]!);
+        result = value >= 0 ? value : undefined;
+        if (result == undefined)
+          warn(`Environment variable ${enVarName} must be a non-negative integer.`);
+      }
+      return result;
     }
 
     this.responseBodyMaxLength =
-      envResponseBodyMaxLength ?? config?.response_body_max_length ?? responseBodyMaxLengthDefault;
+      getNonNegativeIntegerEnvVarValue(kResponseBodyMaxLengthEnvar) ??
+      config?.response_body_max_length ??
+      responseBodyMaxLengthDefault;
+
+    this.asyncTrackingTimeout =
+      getNonNegativeIntegerEnvVarValue(kAsyncTrackingTimeoutEnvar) ??
+      config?.async_tracking_timeout ??
+      asyncTrackingTimeoutDefault;
   }
 
   private absoluteAppmapDir?: string;
@@ -151,6 +166,7 @@ interface ConfigFile {
   packages?: Package[];
   response_body_max_length?: number;
   language?: string;
+  async_tracking_timeout?: number;
 }
 
 // Maintaining the YAML document is important to preserve existing comments and formatting
@@ -188,6 +204,10 @@ function readConfigFile(document: Document): ConfigFile {
     const value = parseInt(String(config.response_body_max_length));
     result.response_body_max_length = value >= 0 ? value : undefined;
   }
+  if ("async_tracking_timeout" in config) {
+    const value = parseInt(String(config.async_tracking_timeout));
+    result.async_tracking_timeout = value >= 0 ? value : undefined;
+  }
 
   return result;
 }

src/util/getTime.ts

@@ -4,3 +4,7 @@ export function getTime(): number {
   const [sec, nano] = hrtime();
   return sec + nano / 1000000000;
 }
+
+export function getTimeInMilliseconds(): number {
+  return Number(hrtime.bigint() / 1_000_000n);
+}