fix: resolve performance issues #53-60

.github/workflows/ci.yml

@@ -8,6 +8,7 @@ on:
     tags:
       - "v*"
   pull_request:
+  workflow_dispatch:
 
 permissions:
   contents: read

main.ts

@@ -19,7 +19,7 @@ const PERSIST_ENGINE = CONFIG.persistEnabled
 PERSIST_ENGINE.dir(CONFIG.persistDir);
 
 // Set up the manager, which will handle our queues for us
-const MANAGER = new QueueManager(PERSIST_ENGINE, CONFIG.queueDepthLimit, CONFIG.queueCountLimit);
+const MANAGER = new QueueManager(PERSIST_ENGINE, CONFIG.queueDepthLimit, CONFIG.queueCountLimit, CONFIG.persistEnabled);
 
 // Load up any existing queue data, if we're persisting
 if (PERSIST_ENGINE instanceof Persistency.FileStore) {
@@ -49,6 +49,8 @@ async function shutdown(signal: string): Promise<void> {
         MANAGER.save();
     }
 
+    PERSIST_ENGINE.close();
+
     writeLog("Goodbye!");
 
     Deno.exit(0);

src/manager.ts

@@ -8,20 +8,70 @@ export class QueueNameTooLongError extends Error {
     }
 }
 
+/**
+ * FIFO queue with O(1) amortized enqueue and dequeue.
+ * Uses a head index instead of Array.shift() to avoid O(n) reindexing.
+ */
+class FIFOQueue<T> {
+    private items: T[] = [];
+    private head = 0;
+
+    push(item: T): void {
+        this.items.push(item);
+    }
+
+    shift(): T | undefined {
+        if (this.head >= this.items.length) return undefined;
+        const item = this.items[this.head];
+        this.items[this.head] = undefined as T; // help GC
+        this.head++;
+        // Compact when the consumed prefix exceeds the remaining items
+        if (this.head > 16 && this.head >= (this.items.length >> 1)) {
+            this.items = this.items.slice(this.head);
+            this.head = 0;
+        }
+        return item;
+    }
+
+    peek(): T | undefined {
+        return this.head < this.items.length ? this.items[this.head] : undefined;
+    }
+
+    get length(): number {
+        return this.items.length - this.head;
+    }
+
+    [Symbol.iterator](): Iterator<T> {
+        let index = this.head;
+        const items = this.items;
+        const end = items.length;
+        return {
+            next(): IteratorResult<T> {
+                if (index < end) {
+                    return { value: items[index++], done: false };
+                }
+                return { value: undefined as unknown as T, done: true };
+            },
+        };
+    }
+}
+
 export default class Manager<T = string> {
-    private queues: Map<string, Array<T>>;
+    private queues: Map<string, FIFOQueue<T>>;
     private store: QueueStore<T>;
     private queueDepthLimit: number;
     private queueCountLimit: number;
+    private persistEnabled: boolean;
 
-    constructor(store: QueueStore<T>, queueDepthLimit?: number, queueCountLimit?: number) {
+    constructor(store: QueueStore<T>, queueDepthLimit?: number, queueCountLimit?: number, persistEnabled?: boolean) {
         this.store = store;
-        this.queues = new Map;
+        this.queues = new Map();
         this.queueDepthLimit = queueDepthLimit ?? 10000;
         this.queueCountLimit = queueCountLimit ?? 1000;
+        this.persistEnabled = persistEnabled ?? true;
     }
 
-    private register(name: string, queue: Array<T>): Manager<T> {
+    private register(name: string, queue: FIFOQueue<T>): Manager<T> {
         this.queues.set(name, queue);
 
         return this;
@@ -52,13 +102,13 @@ export default class Manager<T = string> {
         return queue.length < this.queueDepthLimit;
     }
 
-    private find(name: string): Array<T> | undefined {
+    private find(name: string): FIFOQueue<T> | undefined {
         return this.queues.get(name);
     }
 
     public enqueue(name: string, payload: T): Manager<T> {
         this.validateName(name);
-        const queue = this.find(name) || [];
+        const queue = this.find(name) || new FIFOQueue<T>();
 
         if (this.registered(name) === false) {
             this.register(name, queue);
@@ -69,7 +119,9 @@ export default class Manager<T = string> {
         }
         queue.push(payload);
 
-        this.store.saveEvent(name, payload, true);
+        if (this.persistEnabled) {
+            this.store.saveEvent(name, payload, true);
+        }
 
         return this;
     }
@@ -88,7 +140,7 @@ export default class Manager<T = string> {
             this.queues.delete(name);
         }
 
-        if (payload !== undefined) {
+        if (payload !== undefined && this.persistEnabled) {
             this.store.saveEvent(name, payload, false);
         }
 
@@ -103,7 +155,7 @@ export default class Manager<T = string> {
             return undefined;
         }
 
-        return queue[0];
+        return queue.peek();
     }
 
     public length(name: string): number {
@@ -118,11 +170,13 @@ export default class Manager<T = string> {
 
     public save(): void {
         this.store.clear();
+        const events: QueueEvent<T>[] = [];
         for (const [name, queue] of this.queues) {
             for (const item of queue) {
-                this.store.saveEvent(name, item, true);
+                events.push({ queue: name, payload: item, enqueue: true, dequeue: false });
             }
         }
+        this.store.saveBatch(events);
     }
 
     public load(): void {
@@ -147,7 +201,7 @@ export default class Manager<T = string> {
         if (!existing && !this.canCreateQueue()) {
             return;
         }
-        const queue = existing || [];
+        const queue = existing || new FIFOQueue<T>();
         if (queue.length >= this.queueDepthLimit) {
             return;
         }
@@ -168,4 +222,4 @@ export default class Manager<T = string> {
             this.queues.delete(event.queue);
         }
     }
-}
+}

src/persist.ts

@@ -17,15 +17,31 @@ export function isQueueEvent<T>(value: unknown): value is QueueEvent<T> {
         event.enqueue !== event.dequeue;
 }
 
+function parseLine<T>(line: string): QueueEvent<T> | undefined {
+    try {
+        const event = JSON.parse(line);
+        if (isQueueEvent<T>(event)) {
+            return event;
+        }
+        return undefined;
+    } catch {
+        return undefined;
+    }
+}
+
 export interface QueueStore<T = string> {
     saveEvent(queueName: string, payload: T, isEnqueue: boolean): void;
+    saveBatch(events: Array<QueueEvent<T>>): void;
     loadState(): Array<QueueEvent<T>>;
     clear(): void;
     dir(dir: string): void;
+    close(): void;
 }
 
 export class FileStore<T = string> implements QueueStore<T> {
     private directory: string = '';
+    private writeHandle: Deno.FsFile | null = null;
+    private encoder = new TextEncoder();
 
     private get path(): string {
         return this.directory + "persist.dat";
@@ -38,33 +54,57 @@ export class FileStore<T = string> implements QueueStore<T> {
         Deno.mkdirSync(this.directory, { recursive: true });
     }
 
+    // Lazily open the write handle so that dir() with an invalid path
+    // doesn't throw until an actual I/O operation is attempted.
+    private ensureOpen(): void {
+        if (this.writeHandle === null) {
+            this.ensureDirectory();
+            this.writeHandle = Deno.openSync(this.path, { write: true, create: true, append: true });
+        }
+    }
+
     public saveEvent(queueName: string, payload: T, isEnqueue: boolean): void {
-        this.ensureDirectory();
+        this.ensureOpen();
         const line = JSON.stringify({
             queue: queueName,
             payload: payload,
             enqueue: isEnqueue,
             dequeue: !isEnqueue
         });
-        const file = Deno.openSync(this.path, { write: true, create: true, append: true });
-        file.lockSync(true);
+        this.writeHandle!.lockSync(true);
         try {
-            file.writeSync(new TextEncoder().encode(line + "\n"));
+            this.writeHandle!.writeSync(this.encoder.encode(line + "\n"));
+        } finally {
+            this.writeHandle!.unlockSync();
+        }
+    }
+
+    public saveBatch(events: Array<QueueEvent<T>>): void {
+        if (events.length === 0) return;
+        this.ensureOpen();
+        this.writeHandle!.lockSync(true);
+        try {
+            for (const event of events) {
+                const line = JSON.stringify({
+                    queue: event.queue,
+                    payload: event.payload,
+                    enqueue: event.enqueue,
+                    dequeue: event.dequeue
+                });
+                this.writeHandle!.writeSync(this.encoder.encode(line + "\n"));
+            }
         } finally {
-            file.unlockSync();
-            file.close();
+            this.writeHandle!.unlockSync();
         }
     }
 
     public clear(): void {
-        this.ensureDirectory();
-        const file = Deno.openSync(this.path, { write: true, create: true });
-        file.lockSync(true);
+        this.ensureOpen();
+        this.writeHandle!.lockSync(true);
         try {
-            file.truncateSync(0);
+            this.writeHandle!.truncateSync(0);
         } finally {
-            file.unlockSync();
-            file.close();
+            this.writeHandle!.unlockSync();
         }
     }
 
@@ -73,34 +113,39 @@ export class FileStore<T = string> implements QueueStore<T> {
             const file = Deno.openSync(this.path, { read: true });
             file.lockSync(false);
             try {
-                const chunks: Uint8Array[] = [];
+                // Stream-parse line by line to avoid 3x peak memory from split/filter/map
+                const events: QueueEvent<T>[] = [];
+                const decoder = new TextDecoder();
                 const chunk = new Uint8Array(4096);
-                let totalRead = 0;
+                let leftover = "";
                 while (true) {
                     const read = file.readSync(chunk);
-                    if (read === null || read <= 0) {
+                    if (!read) {
                         break;
                     }
-                    chunks.push(chunk.slice(0, read));
-                    totalRead += read;
+                    leftover += decoder.decode(chunk.subarray(0, read), { stream: true });
+                    let idx = leftover.indexOf("\n");
+                    while (idx >= 0) {
+                        const line = leftover.slice(0, idx);
+                        leftover = leftover.slice(idx + 1);
+                        if (line.length > 0) {
+                            const event = parseLine<T>(line);
+                            if (event) {
+                                events.push(event);
+                            }
+                        }
+                        idx = leftover.indexOf("\n");
+                    }
                 }
-                const buf = new Uint8Array(totalRead);
-                let offset = 0;
-                for (const c of chunks) {
-                    buf.set(c, offset);
-                    offset += c.length;
+                // Flush decoder and process any remaining line (no trailing newline)
+                leftover += decoder.decode();
+                if (leftover.length > 0) {
+                    const event = parseLine<T>(leftover);
+                    if (event) {
+                        events.push(event);
+                    }
                 }
-                const content = new TextDecoder().decode(buf);
-                return content.split("\n")
-                    .filter((line: string) => line.length > 0)
-                    .flatMap((line: string) => {
-                        try {
-                            const event = JSON.parse(line);
-                            return isQueueEvent<T>(event) ? [event] : [];
-                        } catch {
-                            return [];
-                        }
-                    });
+                return events;
             } finally {
                 file.unlockSync();
                 file.close();
@@ -114,8 +159,19 @@ export class FileStore<T = string> implements QueueStore<T> {
     }
 
     public dir(dir: string): void {
+        if (this.writeHandle !== null) {
+            this.writeHandle.close();
+            this.writeHandle = null;
+        }
         this.directory = dir.replace(/\/$/, '') + "/";
     }
+
+    public close(): void {
+        if (this.writeHandle !== null) {
+            this.writeHandle.close();
+            this.writeHandle = null;
+        }
+    }
 }
 
 export class MemoryStore<T = string> implements QueueStore<T> {
@@ -130,6 +186,12 @@ export class MemoryStore<T = string> implements QueueStore<T> {
         });
     }
 
+    public saveBatch(events: Array<QueueEvent<T>>): void {
+        for (const event of events) {
+            this.events.push(event);
+        }
+    }
+
     public clear(): void {
         this.events = [];
     }
@@ -139,4 +201,6 @@ export class MemoryStore<T = string> implements QueueStore<T> {
     }
 
     public dir(): void {}
-}
+
+    public close(): void {}
+}