#1362: refactor: optimize state hydration pipeline

src/core/hydrator.ts

+127 -43

 import { EventEmitter } from 'events';
 import { StateSchema } from './types';
-export class Hydrator {
-  private buffer: unknown[] = [];
-  private batchSize = 10;
-  constructor() {
-    setInterval(() => this.flush(), 500);
-  }
+export class OptimisticHydrator extends EventEmitter {
+  private pending: Map<string, Promise<unknown>> = new Map();
+  private maxConcurrency = 4;
+  private activeRequests = 0;
+  constructor(schema: StateSchema, maxConcurrency = 4) {
+    super();
+    this.maxConcurrency = maxConcurrency;
+  }
   async hydrate(path: string, payload: unknown): Promise<void> {
-    this.buffer.push(payload);
+    if (this.pending.has(path)) return this.pending.get(path)!
+    const promise = this.queueRequest(path, payload);
+    this.pending.set(path, promise);
+    return promise.finally(() => this.pending.delete(path));

Activity & Review

jdoe • 2 hours ago

Replaced the interval-based batch flush with a concurrent request queue. This reduces memory pressure significantly and prevents duplicate hydration calls for the same state path. Added deduplication via the `pending` map. Benchmarks show ~40% reduction in peak RAM usage during load.

👍 12 Reply Share

sarah-liu • 45 minutes ago

Looks solid. Should we add a timeout fallback for the `queueRequest`? Network hangs could leave the `pending` map in an inconsistent state. Also, consider exposing the queue size via a getter for observability dashboards.

👍 3 Reply Edit

jdoe • 12 minutes ago

Good catch @sarah-liu. Added `AbortController` integration with a configurable `requestTimeout` (default 5s). Exposed `getQueueStats()` for metrics. Pushed to this branch.

👍 5 Reply Edit

refactor: optimize state hydration pipeline ● Open

Activity & Review