feat(parallel): per-VU iteration data for customParallelIterations

lib/runner/extensions/event.command.js

@@ -49,7 +49,33 @@ var _ = require('lodash'),
     getCookieDomain, // fn
     postProcessContext, // fn
     sanitizeFiles, // fn
-    maskForbiddenSecrets; // fn
+    maskForbiddenSecrets, // fn
+    applySandboxCursorSentinel; // fn
+
+/**
+ * Replaces cursor.cycles with -1 when in customParallelIterations mode.
+ *
+ * Why -1 over the wire: the runtime sends cursor.cycles = Number.MAX_SAFE_INTEGER
+ * internally (so cursor.seek's bounds check accepts arbitrary iteration
+ * values), but exposing that to scripts would leak as
+ * pm.info.iterationCount === 9007199254740991. We use -1 as the wire
+ * sentinel because it survives JSON, structured-clone, and v8 IPC
+ * encodings intact — Infinity would be clobbered to null by
+ * JSON.stringify. The sandbox renders -1 as Infinity in pm.info.iterationCount
+ * (postman-sandbox/lib/sandbox/pmapi.js).
+ *
+ * @param {Object} scriptCursor - cursor object about to be passed to host.execute
+ * @param {Boolean} isCustomParallelIterations - the Run-level gate
+ * @returns {Object} either a new object with cycles=-1, or the original
+ *   reference if no transform applies
+ */
+applySandboxCursorSentinel = function (scriptCursor, isCustomParallelIterations) {
+    if (!isCustomParallelIterations) {
+        return scriptCursor;
+    }
+
+    return { ...scriptCursor, cycles: -1 };
+};
 
 /**
  * Clones variable scopes and masks variables in forbiddenSecretKeys (Set of "scopeName:variableKey").
@@ -748,7 +774,7 @@ module.exports = {
 
                             // @todo: Expose this as a property in Collection SDK's Script
                             timeout: payload.scriptTimeout,
-                            cursor: scriptCursor,
+                            cursor: applySandboxCursorSentinel(scriptCursor, this.isCustomParallelIterations),
                             context: contextToUse,
                             resolvedPackages: resolvedPackages,
 
@@ -917,3 +943,7 @@ module.exports = {
         }
     }
 };
+
+// Exposed for unit tests — see test/unit/custom-parallel-iterations.test.js.
+// Not part of the public runtime API.
+module.exports._applySandboxCursorSentinel = applySandboxCursorSentinel;

lib/runner/extensions/parallel.command.js

@@ -99,23 +99,50 @@ module.exports = {
             // since we will never reach coords.eof for some partitions because each cursor
             // contains cycles for the entire run, we are breaking off early here.
             // this has been done to keep the contract of a cursor intact.
-            // cycles is defined as "number of iterations in the run"
-            if (coords.iteration === partition.startIndex + coords.partitionCycles) {
+            // cycles is defined as "number of iterations in the run".
+            //
+            // Gated on !isCustomParallelIterations: under custom mode the partition's
+            // cursor.iteration is the VU's per-loop counter (always equal to
+            // startIndex+partitionCycles on loop 2+ given startIndex=0, partitionCycles=1),
+            // so this guard would short-circuit every loop after the first and skip all
+            // items. The host (perftest) drives loop boundaries, not the runtime.
+            if (!this.isCustomParallelIterations &&
+                coords.iteration === partition.startIndex + coords.partitionCycles) {
                 this.triggers.iteration(null, payload.coords);
 
                 return next();
             }
 
             if (coords.cr) {
-                delay = _.get(this.options, 'delay.iteration', 0);
-
                 this.triggers.iteration(null, payload.coords);
+
+                // Custom mode: the host (perftest) is the loop driver. After
+                // firing the iteration trigger so the host can decide what to
+                // do next (continue, addUsers, removeUsers, abort), return
+                // and let the host queue the next iteration explicitly. Do
+                // NOT fire beforeIteration here — startParallelIteration's
+                // own parallel-command invocation carries start:true and
+                // fires it.
+                if (this.isCustomParallelIterations) {
+                    return next();
+                }
+
                 this.triggers.beforeIteration(null, coords);
+                delay = _.get(this.options, 'delay.iteration', 0);
             }
 
 
             if (coords.eof) {
-                this.triggers.iteration(null, coords);
+                // Change 9: in customParallelIterations mode the iteration
+                // trigger carries "loop just completed" coords
+                // (payload.coords, pre-rollover) so the host can attribute
+                // the just-finished work to the correct iteration N. In
+                // runtime-managed mode (Newman/Postman desktop) the eof
+                // site has historically fired with the post-rollover
+                // snapshot — preserve that to avoid silent regressions in
+                // downstream subscribers we can't audit from here.
+                this.triggers.iteration(null,
+                    this.isCustomParallelIterations ? payload.coords : coords);
 
                 return next();
             }
@@ -124,7 +151,13 @@ module.exports = {
                 this.queue('item', {
                     item: item,
                     coords: coords,
-                    data: getIterationData(this.state.data, coords.iteration + partition.startIndex),
+                    // customParallelIterations: the host injects each VU's data row
+                    // via startParallelIteration -> partition.iterationData. Surface it
+                    // as `data` so the sandbox exposes it as pm.iterationData (and layers
+                    // it into _variables, so {{key}} / pm.variables.get() keep working).
+                    data: this.isCustomParallelIterations ?
+                        partition.iterationData :
+                        getIterationData(this.state.data, coords.iteration + partition.startIndex),
                     environment: partition.variables.environment,
                     globals: partition.variables.globals,
                     vaultSecrets: this.state.vaultSecrets,

lib/runner/partition-manager.js

@@ -10,6 +10,7 @@ class PartitionManager {
     spawn () {
         this.partitions = [];
         this.stopActionTriggered = false;
+        this.completionCallback = null;
 
         // we need at least one pool to start with.
         // this is the pool that will be used to process the control instruction
@@ -128,40 +129,47 @@ class PartitionManager {
     }
 
     process (callback) {
-        if (this.runInstance.aborted) {
-            return callback();
-        }
+        this.completionCallback = callback;
 
         let remainingPools = this.partitions.length,
             completed = false;
 
-        const poolFinished = (err) => {
-            if (completed) {
-                return;
-            }
+        const complete = (...args) => {
+                this.completionCallback = null;
 
-            // If run has been aborted, complete immediately
-            if (this.runInstance.aborted) {
-                completed = true;
-                this.runInstance.host && this.runInstance.host.dispose();
+                return callback(...args);
+            },
+            poolFinished = (err) => {
+                if (completed) {
+                    return;
+                }
 
-                return callback(null);
-            }
+                // If run has been aborted, complete immediately
+                if (this.runInstance.aborted) {
+                    completed = true;
+                    this.runInstance.host && this.runInstance.host.dispose();
+
+                    return complete(null);
+                }
 
-            if (err) {
-                completed = true;
+                if (err) {
+                    completed = true;
 
-                return callback(err);
-            }
+                    return complete(err);
+                }
 
-            remainingPools--;
-            if (remainingPools === 0) {
-                completed = true;
-                this.runInstance.host && this.runInstance.host.dispose();
+                remainingPools--;
+                if (remainingPools === 0) {
+                    completed = true;
+                    this.runInstance.host && this.runInstance.host.dispose();
 
-                return callback(null);
-            }
-        };
+                    return complete(null);
+                }
+            };
+
+        if (this.runInstance.aborted) {
+            return complete();
+        }
 
         // First check if priority partition has items
         if (this.priorityPartition && this.priorityPartition.hasInstructions()) {
@@ -171,7 +179,7 @@ class PartitionManager {
                 this.priorityLock = false;
 
                 if (err) {
-                    return callback(err, this.state.cursor.current());
+                    return complete(err, this.state.cursor.current());
                 }
                 // if custom parallel iterations is true, then do not process other partitions
                 if (!this.options.customParallelIterations) {
@@ -262,8 +270,31 @@ class PartitionManager {
         // Always use iteration 0 since we only have 1 iteration of data.
         // and start from the 0th request position.
         partition.cursor.seek(0, 0);
+
+        // Clear the stop flag before queueing new work — see
+        // updatePartitionVariables for the late-write race this guards.
+        partition.stopped = false;
+
+        // customParallelIterations: bump cursor.iteration to reflect the
+        // VU's current loop count so pm.info.iteration is meaningful.
+        // Direct assignment bypasses Cursor.seek's bounds check (safe
+        // because Partition._createCursor raised cycles to MAX_SAFE_INTEGER
+        // for this mode). Post-increment encodes the 0-indexed contract.
+        if (this.runInstance.isCustomParallelIterations) {
+            partition.cursor.iteration = partition.loopIteration++;
+        }
+
         if (localVariables) {
+            // Seed this VU's per-VU variable scope with the payload so {{key}} and
+            // pm.variables.get() resolve (per-partition markers, cookie isolation, etc.).
             partition.variables._variables = localVariables;
+
+            // customParallelIterations: the same payload IS this VU's iteration-data
+            // row. Also route it to `iterationData` so parallel.command sources the
+            // item payload's `data` from it and the sandbox exposes pm.iterationData.
+            if (this.runInstance.isCustomParallelIterations) {
+                partition.iterationData = localVariables;
+            }
         }
 
         this.runInstance.queue('parallel', {
@@ -286,13 +317,31 @@ class PartitionManager {
         if (partition) {
             partition.clearPool();
 
+            // Full-fresh-on-reuse contract (agent/research/07#D3): in
+            // custom mode, a partition recycled via removeUsers/addUsers
+            // must behave as a brand-new VU. Reset the loop counter and
+            // re-clone variables from run-level state.
+            //
+            // partition.stopped=true also guards updatePartitionVariables
+            // against late writes from a script that was mid-execution
+            // when the stop fired — those writes would otherwise land in
+            // the freshly-reset scope and leak the dead VU's mutations
+            // into the next VU.
+            //
+            // Note: loopIteration is NOT reset on abort. perftest's
+            // documented lifecycle constructs one Run per session; no
+            // Run.start retry pattern. If that changes, a parallel reset
+            // hook on the abort path is the one-line fix.
+            //
             // Per-partition cookie jar (requester.perPartitionCookieJar):
-            // in customParallelIterations mode a stopped partition index
-            // is reused by the next parallel iteration — the previous
-            // occupant's session cookies must not leak into it. In-flight
-            // writers hold a reference to the old jar object, so this swap is race-safe
-            // — see Partition#resetCookieJar.
-            if (this.runInstance.options && this.runInstance.options.customParallelIterations) {
+            // the reused partition index must not inherit the previous
+            // occupant's session cookies either. In-flight writers hold a
+            // reference to the old jar object, so this swap is race-safe —
+            // see Partition#resetCookieJar.
+            if (this.runInstance.isCustomParallelIterations) {
+                partition.stopped = true;
+                partition.loopIteration = 0;
+                partition.resetVariables();
                 partition.resetCookieJar();
             }
         }
@@ -310,8 +359,12 @@ class PartitionManager {
         }
 
         if (this.options && this.options.customParallelIterations && this.runInstance.triggers) {
+            const callback = this.completionCallback || this.runInstance.triggers;
+
             this.stopActionTriggered = true;
-            this.runInstance.triggers(null);
+            this.completionCallback = null;
+
+            return callback(null);
         }
     }
 
@@ -324,8 +377,19 @@ class PartitionManager {
      * @param {Object} result - The variables to update
      */
     updatePartitionVariables (partitionIndex, result) {
-        if (this.partitions[partitionIndex] && result && result._variables) {
-            this.partitions[partitionIndex].variables._variables = new sdk.VariableScope(result._variables);
+        const partition = this.partitions[partitionIndex];
+
+        // Drop late writes from a dead VU. A script that was in flight
+        // at stopSinglePartition time finishes after the variables
+        // re-clone — without this guard its write would land in the
+        // fresh scope and leak the dead VU's pm.variables mutations.
+        // Cleared by runSinglePartition before queueing new work.
+        if (!partition || partition.stopped) {
+            return;
+        }
+
+        if (result && result._variables) {
+            partition.variables._variables = new sdk.VariableScope(result._variables);
         }
     }
 }

lib/runner/partition.js

@@ -38,6 +38,17 @@ class Partition {
         // `variables`, the jar is not cloned from run-level state — a
         // fresh partition is born with an empty jar.
         this.cookieJar = null;
+
+        // Per-VU loop counter for customParallelIterations mode. Survives
+        // across runSinglePartition calls and feeds cursor.iteration.
+        // Reset by stopSinglePartition under the "full fresh on reuse"
+        // contract. See agent/research/07-decisions.html D3/D8.
+        this.loopIteration = 0;
+        // Set by stopSinglePartition; consulted by updatePartitionVariables
+        // to drop late writes from a script that was in flight at stop time
+        // (otherwise the dead VU's mutations leak into the next VU's fresh
+        // scope). Cleared in runSinglePartition before queueing new work.
+        this.stopped = false;
     }
 
     /**
@@ -95,6 +106,16 @@ class Partition {
         this.cookieJar = null;
     }
 
+    /**
+     * Re-clones variables from run-level state. Used by stopSinglePartition
+     * to enforce the "full fresh on reuse" contract — when a partition is
+     * recycled by removeUsers/addUsers, the next loop must see a clean
+     * variables scope rather than inheriting the dead VU's mutations.
+     */
+    resetVariables () {
+        this.variables = this._cloneVariables();
+    }
+
     /**
      * Creates a cursor for this partition
      *
@@ -107,7 +128,14 @@ class Partition {
     _createCursor (startIteration, partitionSize, partitionIndex) {
         return Cursor.box({
             length: _.get(this.runInstance, 'state.items.length', 0),
-            cycles: _.get(this.runInstance, 'options.iterationCount', 0),
+            // In customParallelIterations mode the loop count is unbounded
+            // per VU (driven by perftest's duration, not iterationCount).
+            // Raising cycles to MAX_SAFE_INTEGER lets us assign incremented
+            // iteration values via direct write or cursor.seek without
+            // tripping the "seeking out of bounds" guard in cursor.js.
+            cycles: this.runInstance.isCustomParallelIterations ?
+                Number.MAX_SAFE_INTEGER :
+                _.get(this.runInstance, 'options.iterationCount', 0),
             partitionCycles: partitionSize,
             partitionIndex: partitionIndex,
             iteration: startIteration,

lib/runner/run.js

@@ -68,6 +68,13 @@ Run = function PostmanCollectionRun (state, options) { // eslint-disable-line fu
     this._perPartitionCookieJarIgnored =
         Boolean(_.get(this.options, 'requester.perPartitionCookieJar')) &&
         Boolean(_.get(this.options, 'requester.cookieJar'));
+
+    // Single source of truth for the customParallelIterations gate.
+    // Every mode-conditional read in this file and its callers MUST go
+    // through this property — no `this.options?.customParallelIterations`
+    // sprinkled across files. If `options` is missing, the Boolean cast
+    // yields false; if the option is set to a truthy value, true.
+    this.isCustomParallelIterations = Boolean(this.options.customParallelIterations);
 };
 
 _.assign(Run.prototype, {