FOUR-32636: Engine performance improvement for sequential script/service tasks

ProcessMaker/BpmnEngine.php

@@ -20,6 +20,18 @@ class BpmnEngine implements EngineInterface
 {
     use EngineTrait;
 
+    public const INACTIVE_TOKEN_STATUSES = ['CLOSED', 'TRIGGERED', 'COMPLETED'];
+
+    /**
+     * @var array|null
+     */
+    private $nextInlineJob = null;
+
+    /**
+     * @var bool
+     */
+    private $inlineTaskExecutionEnabled = false;
+
     /**
      * @var RepositoryFactoryInterface
      */
@@ -164,4 +176,27 @@ public function registerStartTimerEvents(ProcessInterface $process)
         }
         $this->getJobManager()->disableRegisterStartEvents();
     }
+
+    public function isInlineTaskExecutionEnabled(): bool
+    {
+        return $this->inlineTaskExecutionEnabled;
+    }
+
+    public function setInlineTaskExecutionEnabled(bool $enabled): void
+    {
+        $this->inlineTaskExecutionEnabled = $enabled;
+    }
+
+    public function scheduleInlineJob(array $job): void
+    {
+        $this->nextInlineJob = $job;
+    }
+
+    public function pullInlineJob(): ?array
+    {
+        $job = $this->nextInlineJob;
+        $this->nextInlineJob = null;
+
+        return $job;
+    }
 }

ProcessMaker/Jobs/BpmnAction.php

@@ -18,6 +18,7 @@
 use ProcessMaker\Models\Process as Definitions;
 use ProcessMaker\Models\ProcessRequest;
 use ProcessMaker\Models\ProcessRequestLock;
+use ProcessMaker\Models\ProcessRequestToken;
 use Throwable;
 
 abstract class BpmnAction implements ShouldQueue
@@ -47,6 +48,17 @@ abstract class BpmnAction implements ShouldQueue
 
     protected $processId;
 
+    /**
+     * Context loaded at the beginning of the job. It can be reused after an
+     * external action when the persisted execution state has not changed.
+     *
+     * @var array|null
+     */
+    private $loadedContext;
+
+    /** @var int|null */
+    private $loadedExecutionRevision;
+
     /**
      * @var ProcessRequestLock
      */
@@ -60,8 +72,11 @@ abstract class BpmnAction implements ShouldQueue
     public function handle()
     {
         $response = null;
+        $currentAction = $this;
         try {
-            extract($this->loadContext());
+            $this->loadedContext = $this->loadContext();
+            $this->loadedExecutionRevision = $this->loadedContext['instance']?->execution_revision;
+            extract($this->loadedContext);
             $this->engine = $engine;
             $this->instance = $instance;
 
@@ -70,6 +85,21 @@ public function handle()
 
             // Run engine to the next state
             $this->engine->runToNextState();
+
+            while ($inlineJob = $currentAction->engine->pullInlineJob()) {
+                $context = $currentAction->loadedContext;
+                $context['token'] = $inlineJob['context']['token'];
+                $context['instance'] = $inlineJob['context']['instance'];
+                $context['element'] = $inlineJob['context']['element'];
+                $currentAction->loadedContext = $context;
+                $currentAction->loadedExecutionRevision = $context['instance']->execution_revision;
+                $currentAction->transferInternalContext($inlineJob['job']);
+                $currentAction = $inlineJob['job'];
+
+                $response = App::call([$currentAction, 'action'], $context);
+                $currentAction->engine->runToNextState();
+            }
+
             // call to redirect after all events are completed
             // (e.g. completed, assigned, process completed, etc)
             // excluding system process (non_persistent_process)
@@ -82,27 +112,38 @@ public function handle()
         } catch (Throwable $exception) {
             Log::error($exception->getMessage());
             // Change the Request to error status
-            $request = !$this->instance && $this instanceof StartEvent ? $response : $this->instance;
+            $request = !$currentAction->instance && $currentAction instanceof StartEvent ? $response : $currentAction->instance;
             if ($request) {
-                $request->logError($exception, $element);
+                $request->logError($exception, $context['element'] ?? $element ?? null);
             }
         } finally {
-            $this->unlock();
+            $currentAction->unlock();
         }
 
         return $response;
     }
 
+    public function transferInternalContext(BpmnAction $action): void
+    {
+        $action->engine = $this->engine;
+        $action->instance = $this->instance;
+        $action->loadedContext = $this->loadedContext;
+        $action->loadedExecutionRevision = $this->loadedExecutionRevision;
+        $action->lock = $this->lock;
+        $action->disableGlobalEvents = $this->disableGlobalEvents;
+        $this->lock = null;
+    }
+
     /**
      * Load the context for the action
      *
      * @return array
      */
-    private function loadContext()
+    private function loadContext(?ProcessRequest $lockedInstance = null)
     {
         // Load the process definition
         if (isset($this->instanceId)) {
-            $instance = $this->lockInstance($this->instanceId);
+            $instance = $lockedInstance ?: $this->lockInstance($this->instanceId);
             $processModel = $instance->process;
             $definitions = ($instance->processVersion ?? $instance->process)->getDefinitions(true);
             $engine = app(BpmnEngine::class, ['definitions' => $definitions, 'globalEvents' => !$this->disableGlobalEvents]);
@@ -114,6 +155,8 @@ private function loadContext()
             $instance = null;
         }
 
+        $engine->setInlineTaskExecutionEnabled($this->allowsInlineTaskExecution());
+
         // Load the instances of the process and its collaborators
         if ($instance && $instance->collaboration) {
             $activeRequests = $instance->collaboration->requests()->where('status', 'ACTIVE')->get();
@@ -152,30 +195,73 @@ private function loadContext()
         return compact('definitions', 'instance', 'token', 'process', 'element', 'data', 'processModel', 'engine');
     }
 
+    protected function allowsInlineTaskExecution(): bool
+    {
+        return $this instanceof RunScriptTask || $this instanceof RunServiceTask;
+    }
+
     /**
      * This method execute a callback with the context updated
      *
      * @return array
      */
     public function withUpdatedContext(callable $callable)
     {
-        $context = $this->loadContext();
+        $lockedInstance = $this->lockInstance($this->instanceId, true);
+        $contextReused = $this->canReuseLoadedContext($lockedInstance);
+        if ($contextReused) {
+            $context = $this->loadedContext;
+        } else {
+            $context = $this->loadContext(ProcessRequest::findOrFail($this->instanceId));
+        }
+
+        $this->loadedContext = $context;
+        $this->loadedExecutionRevision = $context['instance']?->execution_revision;
 
         return App::call($callable, $context);
     }
 
+    /**
+     * Determine whether the in-memory engine still represents the persisted
+     * request. This optimization is intentionally limited to linear states.
+     * true: can reuse the loaded context
+     * false: cannot reuse the loaded context
+     * null: cannot determine if the context can be reused
+     */
+    private function canReuseLoadedContext(ProcessRequest $lockedInstance): bool
+    {
+        $activeTokenIds = [];
+        if ((int) $lockedInstance->execution_revision === (int) $this->loadedExecutionRevision) {
+            $activeTokenIds = ProcessRequestToken::query()
+                ->where('process_request_id', $this->instanceId)
+                ->whereNotIn('status', BpmnEngine::INACTIVE_TOKEN_STATUSES)
+                ->limit(2)
+                ->pluck('id')
+                ->all();
+        }
+
+        $fallbackReason = app(BpmnContextReuseGuard::class)->fallbackReason(
+            $this->loadedContext['instance'] ?? null,
+            $lockedInstance,
+            $this->loadedExecutionRevision,
+            $activeTokenIds
+        );
+
+        return $fallbackReason === null;
+    }
+
     /**
      * Lock the instance and its collaborators
      *
      * @param int $instanceId
      *
      * @return ProcessRequest
      */
-    private function lockInstance($instanceId)
+    private function lockInstance($instanceId, bool $lightweight = false)
     {
         try {
             // First attempt to find the instance with retry logic for race conditions
-            $instance = $this->findInstanceWithRetry($instanceId);
+            $instance = $this->findInstanceWithRetry($instanceId, $lightweight);
 
             if (config('queue.default') === 'sync') {
                 return $instance;
@@ -194,13 +280,13 @@ private function lockInstance($instanceId)
             for ($tries = 0; $tries < $maxRetries; $tries++) {
                 $currentLock = $this->currentLock($ids);
                 if (!$currentLock) {
-                    if (ProcessRequest::find($instanceId)) {
+                    if (ProcessRequest::whereKey($instanceId)->exists()) {
                         $lock = $this->requestLock($ids);
                     } else {
                         throw new Exception('Unable to lock instance #' . $this->instanceId . ': Request does not exists');
                     }
                 } elseif ($lock->id == $currentLock->id) {
-                    $instance = ProcessRequest::findOrFail($instanceId);
+                    $instance = $this->findInstance($instanceId, $lightweight);
                     $this->activateLock($lock);
 
                     return $instance;
@@ -221,7 +307,7 @@ private function lockInstance($instanceId)
      * @return ProcessRequest
      * @throws Exception
      */
-    private function findInstanceWithRetry($instanceId)
+    private function findInstanceWithRetry($instanceId, bool $lightweight = false)
     {
         $maxRetries = config('app.bpmn_actions_find_retries', 5);
         $retryDelay = config('app.bpmn_actions_find_retry_delay', 50); // milliseconds
@@ -231,7 +317,7 @@ private function findInstanceWithRetry($instanceId)
 
         for ($attempt = 0; $attempt < $totalAttempts; $attempt++) {
             try {
-                $instance = ProcessRequest::findOrFail($instanceId);
+                $instance = $this->findInstance($instanceId, $lightweight);
 
                 return $instance;
             } catch (ModelNotFoundException $e) {
@@ -251,6 +337,23 @@ private function findInstanceWithRetry($instanceId)
         throw new ModelNotFoundException("ProcessRequest #{$instanceId} not found after {$totalAttempts} attempts");
     }
 
+    /**
+     * Load only lock and revision metadata when validating the fast path.
+     */
+    private function findInstance($instanceId, bool $lightweight): ProcessRequest
+    {
+        $query = ProcessRequest::query();
+        if ($lightweight) {
+            $query->select([
+                'id',
+                'process_collaboration_id',
+                'execution_revision',
+            ]);
+        }
+
+        return $query->findOrFail($instanceId);
+    }
+
     /**
      * Request a lock for the instance
      * @param array $ids
@@ -349,6 +452,8 @@ public function __destruct()
         $this->instance = null;
         $this->engine = null;
         $this->lock = null;
+        $this->loadedContext = null;
+        $this->loadedExecutionRevision = null;
         gc_collect_cycles();
     }
 }

ProcessMaker/Jobs/BpmnContextReuseGuard.php

@@ -0,0 +1,64 @@
+<?php
+
+namespace ProcessMaker\Jobs;
+
+use ProcessMaker\BpmnEngine;
+use ProcessMaker\Models\ProcessRequest;
+
+class BpmnContextReuseGuard
+{
+    /**
+     * Return null when reuse is safe, otherwise return the fallback reason.
+     */
+    public function fallbackReason(
+        ?ProcessRequest $loadedInstance,
+        ProcessRequest $persistedInstance,
+        ?int $loadedRevision,
+        array $persistedActiveTokenIds
+    ): ?string {
+        if (!$loadedInstance) {
+            return 'context_not_loaded';
+        }
+
+        if (!$this->isLinearExecution($loadedInstance)) {
+            return 'execution_not_linear';
+        }
+
+        if ((int) $persistedInstance->execution_revision !== (int) $loadedRevision) {
+            return 'execution_revision_changed';
+        }
+
+        $loadedTokenIds = collect($loadedInstance->getTokens())
+            ->filter(fn ($token) => !in_array($token->getStatus(), BpmnEngine::INACTIVE_TOKEN_STATUSES, true))
+            ->map(fn ($token) => (string) $token->getId())
+            ->values()
+            ->all();
+
+        if (count($persistedActiveTokenIds) !== 1) {
+            return 'persisted_execution_not_linear';
+        }
+
+        if ((string) $persistedActiveTokenIds[0] !== $loadedTokenIds[0]) {
+            return 'active_token_changed';
+        }
+
+        return null;
+    }
+
+    /**
+     * A collaboration, multi-instance token, or multiple active tokens must
+     * use a freshly hydrated engine.
+     */
+    private function isLinearExecution(ProcessRequest $instance): bool
+    {
+        if ($instance->getRawOriginal('process_collaboration_id')) {
+            return false;
+        }
+
+        $tokens = collect($instance->getTokens())
+            ->filter(fn ($token) => !in_array($token->getStatus(), BpmnEngine::INACTIVE_TOKEN_STATUSES, true))
+            ->values();
+
+        return $tokens->count() === 1 && !$tokens->first()->isMultiInstance();
+    }
+}

ProcessMaker/Models/ProcessRequest.php

@@ -155,6 +155,7 @@ class ProcessRequest extends ProcessMakerModel implements ExecutionInstanceInter
         'do_not_sanitize' => 'array',
         'signal_events' => 'array',
         'locked_at' => 'datetime:c',
+        'execution_revision' => 'integer',
     ];
 
     /**