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").
@@ -749,7 +775,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,
 
@@ -862,13 +888,22 @@ module.exports = {
                             result && result._variables &&
                                 (payload.context._variables = new sdk.VariableScope(result._variables));
 
-                            // persist the pm.variables for the next request
-                            result && result._variables &&
+                            // persist the pm.variables for the next request.
+                            // In parallelized (per-VU) runs the run-global scope must stay the
+                            // pristine baseline that new/recycled VUs clone from (via
+                            // Partition#resetVariables -> _cloneVariables(state._variables));
+                            // the live per-VU scope is persisted through updatePartitionVariables
+                            // below. Writing each VU's mutations here would leak them across VU
+                            // reuse and race between concurrent VUs, so skip it in parallel mode.
+                            result && result._variables && !this.areIterationsParallelized &&
                                 (this.state._variables = new sdk.VariableScope(result._variables));
 
                             if (this.areIterationsParallelized) {
-                                // persist the pm.variables for the next request in the current partition
-                                this.partitionManager.updatePartitionVariables(payload.coords.partitionIndex, result);
+                                // persist the pm.variables for the next request in the current partition.
+                                // Pass the generation this execution was queued under so a late
+                                // completion from a reused slot's prior occupant is dropped.
+                                this.partitionManager.updatePartitionVariables(payload.coords.partitionIndex, result,
+                                    payload.coords.partitionGeneration);
                             }
 
                             // persist the mutated request in payload context,
@@ -918,3 +953,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,32 +99,70 @@ 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();
             }
 
             this.queueDelay(function () {
                 this.queue('item', {
                     item: item,
-                    coords: coords,
-                    data: getIterationData(this.state.data, coords.iteration + partition.startIndex),
+                    // Stamp the partition's current generation so a late completion
+                    // from a prior occupant of a reused slot is dropped by
+                    // PartitionManager#updatePartitionVariables (custom mode only).
+                    coords: this.isCustomParallelIterations ?
+                        { ...coords, partitionGeneration: partition.generation } :
+                        coords,
+                    // 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();
 
-            if (err) {
-                completed = true;
+                    return complete(null);
+                }
 
-                return callback(err);
-            }
+                if (err) {
+                    completed = true;
 
-            remainingPools--;
-            if (remainingPools === 0) {
-                completed = true;
-                this.runInstance.host && this.runInstance.host.dispose();
+                    return complete(err);
+                }
 
-                return callback(null);
-            }
-        };
+                remainingPools--;
+                if (remainingPools === 0) {
+                    completed = true;
+                    this.runInstance.host && this.runInstance.host.dispose();
+
+                    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,35 @@ 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;
+                // New generation for the recycled slot — see Partition#generation.
+                // A prior occupant's in-flight completion carries the old
+                // generation and is dropped by updatePartitionVariables.
+                partition.generation++;
+                partition.loopIteration = 0;
+                partition.resetVariables();
                 partition.resetCookieJar();
             }
         }
@@ -310,8 +363,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);
         }
     }
 
@@ -322,10 +379,32 @@ class PartitionManager {
      *
      * @param {Number} partitionIndex - The index of the partition to update
      * @param {Object} result - The variables to update
+     * @param {Number} [generation] - The partition generation this execution was queued under;
+     *   a mismatch means the write is from a prior occupant of a reused slot and is dropped
      */
-    updatePartitionVariables (partitionIndex, result) {
-        if (this.partitions[partitionIndex] && result && result._variables) {
-            this.partitions[partitionIndex].variables._variables = new sdk.VariableScope(result._variables);
+    updatePartitionVariables (partitionIndex, result, generation) {
+        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;
+        }
+
+        // Generation guard: an in-flight completion from a PREVIOUS occupant of
+        // this (reused) slot can arrive after `stopped` was cleared for the new
+        // VU. Its execution carries the generation it was queued under; if that
+        // no longer matches the slot's current generation, drop it so it can't
+        // overwrite the new occupant's scope.
+        if (generation !== undefined && generation !== partition.generation) {
+            return;
+        }
+
+        if (result && result._variables) {
+            partition.variables._variables = new sdk.VariableScope(result._variables);
         }
     }
 }