16290302677

Committed 15 Jul 2025 10:04AM UTC coverage: 96.831% (+0.7%) from 96.15%

Build # 16290302677

Build Type

push

github

Committed by

yiming-liao

Commit Message

fix(core/handler-manager): prevent duplicate handler IDs and expose internal maps

Avoids memory leaks by ensuring handler IDs are unique.
Exported loggerMap and coreMap to support internal testing.

Run Details

443 of 492 branches covered (90.04%)

Branch coverage included in aggregate %.

13 of 15 new or added lines in 3 files covered. (86.67%)

22 existing lines in 5 files now uncovered.

3987 of 4083 relevant lines covered (97.65%)

4.82 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.71

/src/core/handler-manager/handler-manager.ts

import type {
  CancelFlushStrategy,
  OnErrorCallback,
  HandlerManagerConfig,
} from "@/core/handler-manager/handler-manager-config-types";
import type { AddHandlerPosition, Handler } from "@/core/handler-manager/types";
import type { RawPayload } from "@/shared/types/log-payload";
import { DEFAULT_FLUSH_TIMEOUT } from "@/core/handler-manager/constants";
import { executeHandler } from "@/core/handler-manager/utils/execute-handler";
import { flushTasksWithTimeout } from "@/core/handler-manager/utils/flush-tasks-with-timeout";
import { initHandler } from "@/core/handler-manager/utils/init-handler";
import { isHandlerClass } from "@/core/handler-manager/utils/is-handler-class";
import { setupFlushStrategy } from "@/core/handler-manager/utils/setup-flush-strategy";
import { internalLog } from "@/internal";

export class HandlerManager {
  /** List of registered handlers with unique IDs */
  private handlers: { id: string; handler: Handler }[] = [];
  /** Pending async handler tasks mapped to their IDs */
  private pendingTasks = new Map<Promise<void>, string>();

  /** Optional callback when a handler throws an error */
  private onError?: OnErrorCallback;
  /** Timeout duration (ms) for flushing pending tasks */
  private flushTimeout: number | undefined;
  /** Cancel function returned by the active flush strategy */
  private cancelFlushStrategy?: CancelFlushStrategy;

  constructor(private config: HandlerManagerConfig = {}) {
    this.setConfig(config);
  }

  /** Get current configuration */
  public getConfig(): HandlerManagerConfig {
    return this.config;
  }

  /** Updates configuration and resets flush strategy */
  public setConfig(config: HandlerManagerConfig): void {
    this.config = { ...this.config, ...config };
    this.onError = this.config.onError;
    this.flushTimeout = this.config.flushTimeout;
    // Cancel previous flush strategy if any
    if (this.cancelFlushStrategy) {
      this.cancelFlushStrategy();
      this.cancelFlushStrategy = undefined;
    }
    // Setup new flush strategy
    if (this.config.flushStrategy) {
      const cancel = setupFlushStrategy({
        flushStrategy: this.config.flushStrategy,
        flush: this.flush.bind(this),
        onError: this.onError,
      });
      if (cancel) {
        this.cancelFlushStrategy = cancel;
      }
    }
  }

  /** Returns all registered handlers as a shallow copy */
  public getHandlers(): readonly { id: string; handler: Handler }[] {
    return [...this.handlers];
  }

  /** Get a handler by id */
  public getHandler(id: string): { id: string; handler: Handler } | undefined {
    return this.handlers.find((h) => h.id === id);
  }

  /** Registers a new handler (inserted at start or end) */
  public addHandler(
    handler: Handler,
    id: string,
    position: AddHandlerPosition = "end",
  ): boolean {
    if (this.handlers.some((h) => h.id === id)) {
      return false;
    }

    // Insert handler at the start or end of the list
    if (position === "start") {
      this.handlers.unshift({ id, handler });
    } else {
      this.handlers.push({ id, handler });
    }
    // Optional lifecycle init
    initHandler({ handler, id, onError: this.onError });
    // Return id
    return true;
  }

  /** Executes all handlers with the given log payload */
  public runHandlers(rawPayload: RawPayload): void {
    for (const { id, handler } of this.handlers) {
      const task = executeHandler({
        handler,
        id,
        rawPayload,
        onError: this.onError,
      });
      this.pendingTasks.set(task, id);
      task.finally(() => this.pendingTasks.delete(task));
    }
  }

  /** Remove a handler by its ID */
  public removeHandler(id: string): boolean {
    const index = this.handlers.findIndex((item) => item.id === id);
    if (index === -1) {
      return false;
    }
    this.handlers.splice(index, 1);
    return true;
  }

  /** Waits for all pending handler tasks to complete, with optional timeout */
  public async flush(timeout?: number): Promise<void> {
    timeout = timeout ?? this.flushTimeout ?? DEFAULT_FLUSH_TIMEOUT;
    if (this.pendingTasks.size === 0) {
      return;
    }
    try {
      await flushTasksWithTimeout(this.pendingTasks, timeout);
    } catch (error) {
      const handlerId = (error as Error & { handlerId: string }).handlerId;
      internalLog({
        type: "error",
        tag: "HandlerManager.flush",
        message: `Error occurred while flushing pending handler tasks (handlerId: ${handlerId}, timeout: ${timeout}ms)`,
      });
      this.onError?.({ error, id: handlerId, isFlushError: true });
    } finally {
      this.pendingTasks.clear();
    }
  }

  /** Dispose handlers, flush strategy, and clear resources */
  public async dispose(): Promise<void> {
    this.cancelFlushStrategy?.();
    this.cancelFlushStrategy = undefined;
    await Promise.all(
      this.handlers.map(async ({ id, handler }) => {
        if (isHandlerClass(handler)) {
          try {
            await handler.dispose?.();
          } catch (error) {
            this.onError?.({ error, id });
          }
        }
      }),
    );
    this.handlers = [];
    this.pendingTasks.clear();
    this.onError = undefined;
  }
}

1	import type {	1✔
2	CancelFlushStrategy,	1✔
3	OnErrorCallback,	1✔
4	HandlerManagerConfig,	1✔
5	} from "@/core/handler-manager/handler-manager-config-types";	1✔
6	import type { AddHandlerPosition, Handler } from "@/core/handler-manager/types";	1✔
7	import type { RawPayload } from "@/shared/types/log-payload";	1✔
8	import { DEFAULT_FLUSH_TIMEOUT } from "@/core/handler-manager/constants";	1✔
9	import { executeHandler } from "@/core/handler-manager/utils/execute-handler";	1✔
10	import { flushTasksWithTimeout } from "@/core/handler-manager/utils/flush-tasks-with-timeout";	1✔
11	import { initHandler } from "@/core/handler-manager/utils/init-handler";	1✔
12	import { isHandlerClass } from "@/core/handler-manager/utils/is-handler-class";	1✔
13	import { setupFlushStrategy } from "@/core/handler-manager/utils/setup-flush-strategy";	1✔
14	import { internalLog } from "@/internal";	1✔
15		1✔
16	export class HandlerManager {	1✔
17	/** List of registered handlers with unique IDs */	1✔
18	private handlers: { id: string; handler: Handler }[] = [];	1✔
19	/** Pending async handler tasks mapped to their IDs */	1✔
20	private pendingTasks = new Map<Promise<void>, string>();	1✔
21		1✔
22	/** Optional callback when a handler throws an error */	1✔
23	private onError?: OnErrorCallback;	1✔
24	/** Timeout duration (ms) for flushing pending tasks */	1✔
25	private flushTimeout: number \| undefined;	1✔
26	/** Cancel function returned by the active flush strategy */	1✔
27	private cancelFlushStrategy?: CancelFlushStrategy;	1✔
28		1✔
29	constructor(private config: HandlerManagerConfig = {}) {	1✔
30	this.setConfig(config);	24✔
31	}	24✔
32		1✔
33	/** Get current configuration */	1✔
34	public getConfig(): HandlerManagerConfig {	1✔
35	return this.config;	×
36	}	×
37		1✔
38	/** Updates configuration and resets flush strategy */	1✔
39	public setConfig(config: HandlerManagerConfig): void {	1✔
40	this.config = { ...this.config, ...config };	25✔
41	this.onError = this.config.onError;	25✔
42	this.flushTimeout = this.config.flushTimeout;	25✔
43	// Cancel previous flush strategy if any	25✔
44	if (this.cancelFlushStrategy) {	25✔
45	this.cancelFlushStrategy();	1✔
46	this.cancelFlushStrategy = undefined;	1✔
47	}	1✔
48	// Setup new flush strategy	25✔
49	if (this.config.flushStrategy) {	25✔
50	const cancel = setupFlushStrategy({	2✔
51	flushStrategy: this.config.flushStrategy,	2✔
52	flush: this.flush.bind(this),	2✔
53	onError: this.onError,	2✔
54	});	2✔
55	if (cancel) {	2✔
56	this.cancelFlushStrategy = cancel;	2✔
57	}	2✔
58	}	2✔
59	}	25✔
60		1✔
61	/** Returns all registered handlers as a shallow copy */	1✔
62	public getHandlers(): readonly { id: string; handler: Handler }[] {	1✔
63	return [...this.handlers];	4✔
64	}	4✔
65		1✔
66	/** Get a handler by id */	1✔
67	public getHandler(id: string): { id: string; handler: Handler } \| undefined {	1✔
68	return this.handlers.find((h) => h.id === id);	3✔
69	}	3✔
70		1✔
71	/** Registers a new handler (inserted at start or end) */	1✔
72	public addHandler(	1✔
73	handler: Handler,	9✔
74	id: string,	9✔
75	position: AddHandlerPosition = "end",	9✔
76	): boolean {	9✔
77	if (this.handlers.some((h) => h.id === id)) {	9!
NEW 78	return false;	×
NEW 79	}	×
80		9✔
81	// Insert handler at the start or end of the list	9✔
82	if (position === "start") {	9✔
83	this.handlers.unshift({ id, handler });	1✔
84	} else {	9✔
85	this.handlers.push({ id, handler });	8✔
86	}	8✔
87	// Optional lifecycle init	9✔
88	initHandler({ handler, id, onError: this.onError });	9✔
89	// Return id	9✔
90	return true;	9✔
91	}	9✔
92		1✔
93	/** Executes all handlers with the given log payload */	1✔
94	public runHandlers(rawPayload: RawPayload): void {	1✔
95	for (const { id, handler } of this.handlers) {	2✔
96	const task = executeHandler({	2✔
97	handler,	2✔
98	id,	2✔
99	rawPayload,	2✔
100	onError: this.onError,	2✔
101	});	2✔
102	this.pendingTasks.set(task, id);	2✔
103	task.finally(() => this.pendingTasks.delete(task));	2✔
104	}	2✔
105	}	2✔
106		1✔
107	/** Remove a handler by its ID */	1✔
108	public removeHandler(id: string): boolean {	1✔
109	const index = this.handlers.findIndex((item) => item.id === id);	2✔
110	if (index === -1) {	2✔
111	return false;	1✔
112	}	1✔
113	this.handlers.splice(index, 1);	1✔
114	return true;	1✔
115	}	1✔
116		1✔
117	/** Waits for all pending handler tasks to complete, with optional timeout */	1✔
118	public async flush(timeout?: number): Promise<void> {	1✔
119	timeout = timeout ?? this.flushTimeout ?? DEFAULT_FLUSH_TIMEOUT;	2✔
120	if (this.pendingTasks.size === 0) {	2✔
121	return;	1✔
122	}	1✔
123	try {	1✔
124	await flushTasksWithTimeout(this.pendingTasks, timeout);	1✔
125	} catch (error) {	2!
126	const handlerId = (error as Error & { handlerId: string }).handlerId;	×
127	internalLog({	×
128	type: "error",	×
129	tag: "HandlerManager.flush",	×
130	message: `Error occurred while flushing pending handler tasks (handlerId: ${handlerId}, timeout: ${timeout}ms)`,	×
131	});	×
132	this.onError?.({ error, id: handlerId, isFlushError: true });	×
133	} finally {	2✔
134	this.pendingTasks.clear();	1✔
135	}	1✔
136	}	2✔
137		1✔
138	/** Dispose handlers, flush strategy, and clear resources */	1✔
139	public async dispose(): Promise<void> {	1✔
140	this.cancelFlushStrategy?.();	2!
141	this.cancelFlushStrategy = undefined;	2✔
142	await Promise.all(	2✔
143	this.handlers.map(async ({ id, handler }) => {	2✔
144	if (isHandlerClass(handler)) {	2✔
145	try {	1✔
146	await handler.dispose?.();	1✔
147	} catch (error) {	1!
148	this.onError?.({ error, id });	×
149	}	×
150	}	1✔
151	}),	2✔
152	);	2✔
153	this.handlers = [];	2✔
154	this.pendingTasks.clear();	2✔
155	this.onError = undefined;	2✔
156	}	2✔
157	}	1✔

yiming-liao / logry / 16290302677

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous