feat(nodes): add loopNode - iterate over arrays with rate limiting by FelixNg1022 · Pull Request #66 · wespreadjam/jam-nodes

FelixNg1022 · 2026-04-03T22:21:41Z

Summary

Implement loopNode primitive for iterating over arrays with rate limiting and concurrency control
Sequential iteration by default (concurrency: 1)
Optional parallel execution with configurable concurrency limit
Rate limiting via delayMs between iterations/batches
Per-item error handling with continueOnError option
Full Zod schemas for input/output validation
Unit tests for metadata, schema validation, and execution paths

Closes #9

Input/Output Schema

Input:

{
  items: unknown[];        // Array to iterate over
  concurrency?: number;    // Max parallel executions (default: 1)
  delayMs?: number;        // Delay between iterations in ms (default: 0)
  continueOnError?: boolean; // Continue on failure (default: false)
}

Output:

{
  results: unknown[];      // Results from each iteration
  errors?: { index: number; error: string }[]; // Errors if any
}

Test plan

pnpm test passes — new unit tests for loopNode
Schema validation: rejects invalid concurrency (0), negative delay
Empty array returns empty results
Sequential processing preserves order
Concurrent processing with batching preserves order

Implements a loop primitive node with: - Sequential iteration by default - Optional parallel execution with configurable concurrency limit - Rate limiting via delayMs between iterations/batches - Per-item error handling with continueOnError option - Zod schemas for input/output validation - Unit tests for metadata, schema validation, and execution Closes wespreadjam#9

Copilot

Pull request overview

This PR introduces a new loopNode under packages/nodes intended to support iterating over arrays with configurable concurrency and delay, and wires it into the public exports and built-in node list.

Changes:

Added a new loopNode implementation with Zod input/output schemas.
Exported the node + schemas/types from the logic index and package root, and registered it in builtInNodes.
Added initial Vitest unit tests for metadata and basic validation/execution paths.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 7 comments.

File	Description
packages/nodes/src/logic/loop.ts	Adds the new `loopNode` implementation and schemas.
packages/nodes/src/logic/index.ts	Re-exports `loopNode` and its schemas/types from the logic barrel.
packages/nodes/src/logic/tests/loop.test.ts	Adds unit tests for loop node metadata and basic execution.
packages/nodes/src/index.ts	Re-exports loop node artifacts from the package root and adds it to `builtInNodes`.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-04-03T22:27:18Z

+  continueOnError: boolean
+): Promise<{ results: unknown[]; errors: LoopError[] }> {
+  const results: unknown[] = [];
+  const errors: LoopError[] = [];
+
+  for (let i = 0; i < items.length; i++) {
+    try {
+      results.push(items[i]);
+    } catch (err) {
+      const errorMsg = err instanceof Error ? err.message : String(err);
+      errors.push({ index: i, error: errorMsg });
+      if (!continueOnError) {
+        break;
+      }
+      results.push(null);
+    }


processSequential currently just pushes the original items[i] into results. This operation cannot throw for normal arrays, so the try/catch, errors collection, and continueOnError behavior are effectively dead code. If this node is meant to execute per-item work (e.g., invoking child node execution), that processing callback needs to happen here so failures can be captured; otherwise the node should be simplified to remove the misleading error-handling paths.

Suggested change

continueOnError: boolean

): Promise<{ results: unknown[]; errors: LoopError[] }> {

const results: unknown[] = [];

const errors: LoopError[] = [];

for (let i = 0; i < items.length; i++) {

try {

results.push(items[i]);

} catch (err) {

const errorMsg = err instanceof Error ? err.message : String(err);

errors.push({ index: i, error: errorMsg });

if (!continueOnError) {

break;

}

results.push(null);

}

_continueOnError: boolean

): Promise<{ results: unknown[]; errors: LoopError[] }> {

const results: unknown[] = [];

const errors: LoopError[] = [];

for (let i = 0; i < items.length; i++) {

results.push(items[i]);

Copilot · 2026-04-03T22:27:18Z

+  let stopped = false;
+
+  // Process in batches of `concurrency`
+  for (let batchStart = 0; batchStart < items.length; batchStart += concurrency) {
+    if (stopped) break;
+
+    const batchEnd = Math.min(batchStart + concurrency, items.length);
+    const batch = items.slice(batchStart, batchEnd);
+
+    const batchPromises = batch.map(async (item, batchIndex) => {
+      const globalIndex = batchStart + batchIndex;
+      try {
+        results[globalIndex] = item;
+      } catch (err) {
+        const errorMsg = err instanceof Error ? err.message : String(err);
+        errors.push({ index: globalIndex, error: errorMsg });
+        if (!continueOnError) {
+          stopped = true;
+        }
+      }
+    });
+
+    await Promise.all(batchPromises);
+
+    if (delayMs > 0 && batchEnd < items.length && !stopped) {


processConcurrent assigns results[globalIndex] = item inside a try/catch, but that assignment (and reading item) will not throw in typical usage. As a result, errors will never populate and continueOnError/stopped won't have any real effect. If per-item execution is intended, introduce an actual per-item processing step that can fail and be captured; otherwise remove this error-handling logic to avoid giving a false sense of resilience.

Suggested change

let stopped = false;

// Process in batches of `concurrency`

for (let batchStart = 0; batchStart < items.length; batchStart += concurrency) {

if (stopped) break;

const batchEnd = Math.min(batchStart + concurrency, items.length);

const batch = items.slice(batchStart, batchEnd);

const batchPromises = batch.map(async (item, batchIndex) => {

const globalIndex = batchStart + batchIndex;

try {

results[globalIndex] = item;

} catch (err) {

const errorMsg = err instanceof Error ? err.message : String(err);

errors.push({ index: globalIndex, error: errorMsg });

if (!continueOnError) {

stopped = true;

}

}

});

await Promise.all(batchPromises);

if (delayMs > 0 && batchEnd < items.length && !stopped) {

void continueOnError;

// Process in batches of `concurrency`

for (let batchStart = 0; batchStart < items.length; batchStart += concurrency) {

const batchEnd = Math.min(batchStart + concurrency, items.length);

const batch = items.slice(batchStart, batchEnd);

const batchPromises = batch.map(async (item, batchIndex) => {

const globalIndex = batchStart + batchIndex;

results[globalIndex] = item;

});

await Promise.all(batchPromises);

if (delayMs > 0 && batchEnd < items.length) {

Copilot · 2026-04-03T22:27:19Z

+/**
+ * Loop node - iterate over an array of items with rate limiting and concurrency control.
+ *
+ * Processes each item in the array, supporting both sequential and parallel execution.
+ * In a workflow context, the execution engine handles running child nodes for each item;
+ * this node manages the iteration, concurrency, delay, and error tracking.
+ *


The docstring/PR description says the loop node manages iteration while the execution engine runs child nodes for each item, but this implementation never invokes any child execution and simply echoes items back as results. This makes concurrency/rate limiting largely meaningless and doesn't meet the stated behavior of “executing child nodes for each item”. Either integrate with the runtime’s child-node execution mechanism (so results reflect per-item execution outputs) or update the node/PR description to reflect that this is currently a pass-through iterator.

Copilot · 2026-04-03T22:27:19Z

+          success: true,
+          output: {
+            results: [],
+            errors: [],


In the empty-items fast path, errors is returned as an empty array even though the schema marks it as optional and the non-empty path omits it when there are no errors. For consistency (and to match the documented output shape), return errors: undefined (or omit the property) when there are no errors.

Suggested change

errors: [],

errors: undefined,

Copilot · 2026-04-03T22:27:19Z

+  it('should process items sequentially', async () => {
+    const mockContext = {
+      userId: 'test',
+      workflowExecutionId: 'test',
+      credentials: {},
+      variables: {},
+      interpolate: (s: string) => s,
+      evaluateJsonPath: (s: string) => s,
+    };
+
+    const result = await loopNode.executor(
+      { items: ['a', 'b', 'c'], concurrency: 1, delayMs: 0, continueOnError: false },
+      mockContext as any
+    );
+
+    expect(result.success).toBe(true);
+    expect(result.output?.results).toEqual(['a', 'b', 'c']);
+  });
+
+  it('should process items concurrently', async () => {
+    const mockContext = {
+      userId: 'test',
+      workflowExecutionId: 'test',
+      credentials: {},
+      variables: {},
+      interpolate: (s: string) => s,
+      evaluateJsonPath: (s: string) => s,
+    };
+
+    const result = await loopNode.executor(
+      { items: [1, 2, 3, 4, 5], concurrency: 3, delayMs: 0, continueOnError: false },
+      mockContext as any
+    );
+
+    expect(result.success).toBe(true);
+    expect(result.output?.results).toEqual([1, 2, 3, 4, 5]);
+  });


These tests cover metadata and basic schema defaults, but they don’t exercise key behavior introduced by this node: delayMs timing, continueOnError/error accumulation, or that concurrency batching preserves order when work is actually performed. Add unit tests that (1) simulate a per-item failure and assert errors + results placeholder behavior, and (2) use fake timers to assert delays are applied between iterations/batches.

Copilot · 2026-04-03T22:27:19Z

@@ -0,0 +1,99 @@
+import { describe, it, expect } from 'vitest';
+import { loopNode, LoopInputSchema } from '../loop';


Test imports should follow the project’s ESM convention of using .js extensions (see webhook-trigger.test.ts). import { loopNode, LoopInputSchema } from '../loop'; is inconsistent and can break under strict ESM resolution. Update to import from ../loop.js.

Suggested change

import { loopNode, LoopInputSchema } from '../loop';

import { loopNode, LoopInputSchema } from '../loop.js';

Copilot · 2026-04-03T22:27:20Z

+    const mockContext = {
+      userId: 'test',
+      workflowExecutionId: 'test',
+      credentials: {},
+      variables: {},
+      interpolate: (s: string) => s,
+      evaluateJsonPath: (s: string) => s,
+    };


The mocked execution context here doesn’t match NodeExecutionContext (it provides interpolate/evaluateJsonPath but omits resolveNestedPath) and then is cast to any. This weakens the tests because they won’t fail if loopNode later starts relying on resolveNestedPath like other logic nodes do. Prefer a minimal, correctly-shaped context (include resolveNestedPath) and avoid as any where possible.

Copilot AI review requested due to automatic review settings April 3, 2026 22:21

Copilot started reviewing on behalf of FelixNg1022 April 3, 2026 22:22 View session

Copilot AI reviewed Apr 3, 2026

View reviewed changes

FelixNg1022 mentioned this pull request Apr 4, 2026

fix: docs-mcp corrections and minor code cleanup #67

Open

4 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(nodes): add loopNode - iterate over arrays with rate limiting#66

feat(nodes): add loopNode - iterate over arrays with rate limiting#66
FelixNg1022 wants to merge 1 commit into
wespreadjam:mainfrom
FelixNg1022:feat/loop-node

FelixNg1022 commented Apr 3, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Apr 3, 2026

Uh oh!

Copilot AI Apr 3, 2026

Uh oh!

Copilot AI Apr 3, 2026

Uh oh!

Copilot AI Apr 3, 2026

Uh oh!

Copilot AI Apr 3, 2026

Uh oh!

Copilot AI Apr 3, 2026

Uh oh!

Copilot AI Apr 3, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

		@@ -0,0 +1,99 @@
		import { describe, it, expect } from 'vitest';
		import { loopNode, LoopInputSchema } from '../loop';

Conversation

FelixNg1022 commented Apr 3, 2026

Summary

Input/Output Schema

Test plan

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants