A parallel scope's step asks and records like the root executor's

source/FluxBase-Core-Tests/FBParallelExecutorTest.class.st

@@ -392,3 +392,63 @@ FBParallelExecutorTest >> testTheScopeJoinIsFoundPastAnInnerLoop [
 	self assert: instance state equals: #done.
 	self assert: seen asArray equals: { { #( 2 4 ). nil } }
 ]
+
+{ #category : 'tests' }
+FBParallelExecutorTest >> testEveryIterationRecordsTheTransitionItFollows [
+	"a transition followed inside a parallel scope is recorded like one followed by the
+	root executor - once per iteration that follows it, not for an iteration its
+	condition holds back. The scope's log used to have no transitions at all"
+	| seen instance body followed |
+	seen := OrderedCollection new.
+	instance := self loopOver: #( 1 2 3 ) bodyWhen: [ :x | x < 3 ] into: seen.
+	body := instance flow nodes detect: [ :each | each name = 'body' ].
+	followed := instance events select: [ :each |
+		(each isKindOf: FBTransitionFollowed) and: [ each transition target == body ] ].
+	self assert: followed size equals: 2.
+	self assert: (followed collect: [ :each | each value ]) asArray equals: #( 1 2 ).
+	self assert: instance state equals: #done
+]
+
+{ #category : 'tests' }
+FBParallelExecutorTest >> testTheInstanceIsAskedBeforeEachIterationsTransition [
+	"FBFlowInstance>>aboutToTransition: is asked once per iteration that follows the
+	body's transition - what ApptiveBase counts its rate limits with, and what a loop
+	used to bypass"
+	| start fork body instance flow |
+	start := FBStartEvent new.
+	fork := start addOutgoingNode: (FBParallelNode named: 'fork').
+	body := (FBPluggableTask named: 'body') block: [ :act | act value: act argument ].
+	fork --> (body when: [ :x | x < 3 ]).
+	(body addOutgoingNode: (FBParallelJoinNode named: 'scope join')) addOutgoingNode: FBEndEvent new.
+	flow := FBFlow new start: start.
+	instance := FBProbingFlowInstance new flow: flow.
+	instance executeWith: #( 1 2 3 ).
+	self assert: instance state equals: #done.
+	self assert: (instance asked count: [ :each | each target == body ]) equals: 2
+]
+
+{ #category : 'tests' }
+FBParallelExecutorTest >> testARefusedTransitionLeavesTheIterationsWorkQueued [
+	"the hook refuses the second iteration's transition: the error comes out of the
+	step, the work item stays queued as it does in the root executor, and nothing of
+	that iteration ran"
+	| start fork body instance flow ran |
+	ran := OrderedCollection new.
+	start := FBStartEvent new.
+	fork := start addOutgoingNode: (FBParallelNode named: 'fork').
+	body := (FBPluggableTask named: 'body') block: [ :act | ran add: act argument. act value: act argument ].
+	fork --> body.
+	(body addOutgoingNode: (FBParallelJoinNode named: 'scope join')) addOutgoingNode: FBEndEvent new.
+	flow := FBFlow new start: start.
+	instance := FBProbingFlowInstance new flow: flow.
+	instance refuseWith: [ :transition |
+		(transition target == body and: [ instance executor currentIteration = 2 ])
+			ifTrue: [ Error signal: 'refused' ] ].
+	instance initialValue: #( 1 2 3 ).
+	self should: [ instance execute ] raise: Error.
+	self assert: ran asArray equals: #( 1 ).
+	self assert: instance executor pending size equals: 1.
+	self assert: instance executor pending first transition target identicalTo: body.
+	self deny: (instance events anySatisfy: [ :each |
+		(each isKindOf: FBTransitionFollowed) and: [ each transition target == body and: [ each value = 2 ] ] ])
+]

source/FluxBase-Core-Tests/FBProbingFlowInstance.class.st

@@ -0,0 +1,32 @@
+"
+A flow instance for tests that records what the engine asks it before a transition
+(FBFlowInstance>>aboutToTransition:) and can refuse one, the way ApptiveBase's instance
+refuses a transition whose rate limit is used up.
+"
+Class {
+	#name : 'FBProbingFlowInstance',
+	#superclass : 'FBFlowInstance',
+	#instVars : [
+		'asked',
+		'refusal'
+	],
+	#category : 'FluxBase-Core-Tests',
+	#package : 'FluxBase-Core-Tests'
+}
+
+{ #category : 'hooks' }
+FBProbingFlowInstance >> aboutToTransition: aTransition [
+	self asked add: aTransition.
+	refusal ifNotNil: [ :block | block value: aTransition ]
+]
+
+{ #category : 'accessing' }
+FBProbingFlowInstance >> asked [
+	^ asked ifNil: [ asked := OrderedCollection new ]
+]
+
+{ #category : 'accessing' }
+FBProbingFlowInstance >> refuseWith: aBlock [
+	"aBlock gets every transition asked about and may raise"
+	refusal := aBlock
+]

source/FluxBase-Core/FBParallelExecutor.class.st

@@ -390,16 +390,24 @@ FBParallelExecutor >> shouldFollowTransition: transition with: value [
 
 { #category : 'services' }
 FBParallelExecutor >> step [
-	| workItem transition lastValue |
-	workItem := pending removeFirst.
+	| workItem transition lastValue followed |
+	workItem := pending first.
 	transition := workItem transition.
 	"the iteration this step belongs to, read off the work item's token: a body seed
 	(no token yet, see #scopeToken) is the newest iteration's, everything else carries
 	the token of the iteration that queued it"
 	current := workItem token ifNil: [ index ] ifNotNil: [ :token | self iterationOf: token ].
 	lastValue := transition source lastValueIn: self.
-
-	(self shouldFollowTransition: transition with: lastValue) ifTrue: [
+	followed := self shouldFollowTransition: transition with: lastValue.
+	"the same two things FBFlowExecutor>>step does around a transition: the instance is
+	asked before it is taken - ApptiveBase counts its rate limits there, and a refusal
+	raised from the hook leaves the work item queued, as it does in the root executor -
+	and the transition is recorded. Only for a transition actually followed: inside a
+	scope the branch condition is decided here, not at queueing time"
+	followed ifTrue: [ flowInstance aboutToTransition: transition ].
+	pending removeFirst.
+	followed ifTrue: [
+		flowInstance addEvent: (FBTransitionFollowed new transition: transition; value: lastValue).
 		self
 			activateTransition: transition
 			with: lastValue