Clone workers in streams

[neekolas]

Add cloned worker support for background streams and make StreamsChaos.start async in streams.ts

Introduce StreamsConfig with cloned flag, update StreamsChaos to optionally clone() WorkerClient instances before starting streams, and change RuntimeConfig to carry backgroundStreams config instead of a boolean.

📍Where to Start

Start with StreamsChaos construction and start flow in streams.ts.

---

📊 Macroscope summarized 3d501d6. 7 files reviewed, 14 issues evaluated, 13 issues filtered, 0 comments posted

🗂️ Filtered Issues

chaos/db.ts — 0 comments posted, 2 evaluated, 2 filtered

• line 28: DbChaos.start does not guard against being called multiple times. Each call sets this.interval = setInterval(...) without clearing any existing interval, resulting in multiple intervals running concurrently and repeatedly applying locks. This can cause overlapping lock attempts and unexpected load. Add a guard (e.g., clear existing interval or throw if already started) before setting a new one. [ Low confidence ]
• line 30: impactedWorkerPercentage is used without validation in DbChaos.start. If it is NaN, negative, or >100, the condition Math.random() * 100 > impactedWorkerPercentage misbehaves (e.g., any comparison with NaN is false, locking all workers every interval). Validate the value is a number between 0 and 100 and handle invalid input (e.g., clamp or skip). [ Low confidence ]

chaos/streams.ts — 0 comments posted, 1 evaluated, 1 filtered

• line 33: StreamsChaos.stop only calls stopStreams() on stored workers but never terminates cloned WorkerClient instances created in start when config.cloned is true. This leaves worker threads/process resources alive, causing leaks and lingering background activity. Ensure clones are terminated (e.g., await worker.terminate()) and cleanly disposed. [ Low confidence ]

forks/cli.ts — 0 comments posted, 1 evaluated, 1 filtered

• line 146: runForkDetection misinterprets per-run fork detection by using the total count of files in logs/cleaned for each run (getForkCount()), then accumulating that total into stats.forksDetected on every iteration. This causes overcounting across runs and per-run output to show cumulative totals rather than forks detected in that run. Specifically: forkCount is the cumulative number of cleaned logs across all runs, but the code prints Run i: ... ${forkCount} fork(s) detected and does stats.forksDetected += forkCount. You should compute forks detected for the current run only (e.g., diff the count before/after cleaning, or base on filenames associated with the current run), and increment stats by that delta. [ Out of scope ]

helpers/versions.ts — 0 comments posted, 2 evaluated, 2 filtered

• line 232: regressionClient matches the provided nodeBindings string against VersionList.find((v) => v.nodeBindings === nodeBindings). Callers pass sdk (e.g., "4.3.0") into regressionClient as the nodeBindings argument, which will not match VersionList.nodeBindings values (e.g., "1.7.0"). This causes a runtime error: SDK version ${nodeBindings} not found in VersionList. Ensure the lookup uses the correct field (nodeSDK vs nodeBindings) or pass the correct value from the caller. [ Out of scope ]
• line 274: Config parity bug: on the retry path after a DB open error, ClientClass.create is invoked without historySyncUrl and disableDeviceSync, while the initial attempt includes these options. This yields inconsistent client behavior depending on whether the first attempt fails. Preserve identical options across both paths. [ Low confidence ]

workers/main.ts — 0 comments posted, 4 evaluated, 4 filtered

• line 273: getStats() may dereference this.client.debugInformation when it’s undefined. apiStatistics() is guarded by ?., but this.client.debugInformation.clearAllStatistics() is not, causing a possible runtime exception. Guard both calls or null-check debugInformation first. [ Out of scope ]
• line 315: startSync launches an infinite while (true) loop without any cancellation mechanism, independent of stopStreams() or terminate(). Multiple invocations spawn multiple perpetual loops. This can cause runaway background activity and hinder clean shutdown. Provide an abort/cancel control or tie the loop to a tracked flag that’s cleared on terminate()/stopStreams(). [ Out of scope ]
• line 339: startStream allows starting the same stream type multiple times. When the stream is already active, it does not return, creating another stream and overwriting the streamControllers entry for that type. Older streams cannot be aborted via endStream, leading to orphaned streams and potential leaks. Restore the early return when a stream is already active or manage multiple instances safely. [ Out of scope ]
• line 682: Unhandled rejection: In initMessageStream and initConversationStream, errors inside the async stream loop are re-thrown from within a void IIFE. This results in unhandled promise rejections, breaking the loop and potentially destabilizing the process without a controlled shutdown or retry. Handle errors by logging and gracefully breaking the loop or implementing a bounded retry with backoff, and ensure stream cleanup on all paths. [ Out of scope ]

workers/manager.ts — 0 comments posted, 3 evaluated, 3 filtered

• line 540: getWorkers no longer uses the mapped apiUrl values when the input is a Record<string, string>. Previously, the record’s value was forwarded to manager.createWorker(descriptor, sdkVersions[0], apiUrl). The current code ignores apiUrl and calls manager.createWorker(descriptor, randomSdk) for each descriptor, breaking the prior external contract and causing workers to be initialized without the intended API URL routing. [ Low confidence ]
• line 627: The function getNextFolderName does not actually ensure the folder name is available. When .data exists, it returns a random 32-character alphanumeric string without checking fs.readdirSync or similar to confirm non-existence. This can collide with an existing folder and cause downstream failures when trying to create/use the directory. Given the comment "Helper function to get the next available folder name", this is a behavioral bug: availability is not verified. [ Low confidence ]
• line 628: Contract/semantics change: when .data exists, getNextFolderName now returns a 32-character alphanumeric string (including digits) instead of a deterministic single lowercase letter based on existing folders. Any callers expecting a short, [a-z] name or 'next' sequential behavior may break (validation, UI, path length constraints, or ordering). This introduces a runtime incompatibility risk with existing consumers. [ Low confidence ]

[neekolas]

Warning

This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack on Graphite.
Learn more

• Clone workers in streams #1585 👈 (View in Graphite)
• Add network chaos to fork tests #1570 : 1 other dependent PR (#1574 )
• main

This stack of pull requests is managed by Graphite. Learn more about stacking.

[macroscopeapp]

Promise.all here can leak clones when one clone() rejects: successful clones aren’t tracked or cleaned up. Consider cloning sequentially with try/catch, cleaning up any successful clones before rethrowing.

-      allWorkers = await Promise.all(allWorkers.map((w) => w.clone()));
-    }
+      const clones: WorkerClient[] = [];
+      try {
+        for (const w of allWorkers) {
+          const c = await w.clone();
+          clones.push(c);
+        }
+        allWorkers = clones;
+      } catch (err) {
+        for (const c of clones) {
+          try {
+            c.stopStreams();
+          } catch {}
+        }
+        throw err;
+      }
+    }

🚀 Reply to ask Macroscope to explain or update this suggestion.

👍 Helpful? React to give us feedback.