src/Services/QueuedJobService.php
public function runJob($jobId)
    {
        // first retrieve the descriptor
        /** @var QueuedJobDescriptor $jobDescriptor */
        $jobDescriptor = DataObject::get_by_id(
            QueuedJobDescriptor::class,
            (int)$jobId
        );
        if (!$jobDescriptor) {
            throw new Exception("$jobId is invalid");
        }

        // now lets see whether we have a current user to run as. Typically, if the job is executing via the CLI,
        // we want it to actually execute as the RunAs user - however, if running via the web (which is rare...), we
        // want to ensure that the current user has admin privileges before switching. Otherwise, we just run it
        // as the currently logged in user and hope for the best

        // We need to use $_SESSION directly because SS ties the session to a controller that no longer exists at
        // this point of execution in some circumstances
        $originalUserID = isset($_SESSION['loggedInAs']) ? $_SESSION['loggedInAs'] : 0;
        /** @var Member|null $originalUser */
        $originalUser = $originalUserID
            ? DataObject::get_by_id(Member::class, $originalUserID)
            : null;
        $runAsUser = null;

        // If the Job has requested that we run it as a particular user, then we should try and do that.
        if ($jobDescriptor->RunAs() !== null) {
            $runAsUser = $this->setRunAsUser($jobDescriptor->RunAs(), $originalUser);
        }

        // set up a custom error handler for this processing
        $errorHandler = new JobErrorHandler();

        $job = null;

        $broken = false;

        // Push a config context onto the stack for the duration of this job run.
        Config::nest();

        if ($this->grabMutex($jobDescriptor)) {
            try {
                $job = $this->initialiseJob($jobDescriptor);

                // get the job ready to begin.
                if (!$jobDescriptor->JobStarted) {
                    $jobDescriptor->JobStarted = DBDatetime::now()->Rfc2822();
                } else {
                    $jobDescriptor->JobRestarted = DBDatetime::now()->Rfc2822();
                }

                // Only write to job as "Running" if 'isComplete' was NOT set to true
                // during setup() or prepareForRestart()
                if (!$job->jobFinished()) {
                    $jobDescriptor->JobStatus = QueuedJob::STATUS_RUN;
                    $jobDescriptor->write();
                }

                $lastStepProcessed = 0;
                // have we stalled at all?
                $stallCount = 0;

                if (class_exists(Subsite::class) && !empty($job->SubsiteID)) {
                    Subsite::changeSubsite($job->SubsiteID);

                    // lets set the base URL as far as Director is concerned so that our URLs are correct
                    /** @var Subsite $subsite */
                    $subsite = DataObject::get_by_id(Subsite::class, $job->SubsiteID);
                    if ($subsite && $subsite->exists()) {
                        $domain = $subsite->domain();
                        $base = rtrim(Director::protocol() . $domain, '/') . '/';

                        Config::modify()->set(Director::class, 'alternate_base_url', $base);
                    }
                }

                // while not finished
                while (!$job->jobFinished() && !$broken) {
                    // see that we haven't been set to 'paused' or otherwise by another process
                    /** @var QueuedJobDescriptor $jobDescriptor */
                    $jobDescriptor = DataObject::get_by_id(
                        QueuedJobDescriptor::class,
                        (int)$jobId
                    );
                    if (!$jobDescriptor || !$jobDescriptor->exists()) {
                        $broken = true;
                        $this->getLogger()->error(
                            print_r(
                                [
                                    'errno' => 0,
                                    'errstr' => 'Job descriptor ' . $jobId . ' could not be found',
                                    'errfile' => __FILE__,
                                    'errline' => __LINE__,
                                    'errcontext' => [],
                                ],
                                true
                            ),
                            [
                                'file' => __FILE__,
                                'line' => __LINE__,
                            ]
                        );
                        break;
                    }
                    if ($jobDescriptor->JobStatus != QueuedJob::STATUS_RUN) {
                        // we've been paused by something, so we'll just exit
                        $job->addMessage(_t(
                            __CLASS__ . '.JOB_PAUSED',
                            'Job paused at {time}',
                            ['time' => DBDatetime::now()->Rfc2822()]
                        ));
                        $broken = true;
                    }

                    if (!$broken) {
                        // Collect output as job messages as well as sending it to the screen
                        $obLogger = function ($buffer, $phase) use ($job, $jobDescriptor) {
                            $job->addMessage($buffer);
                            if ($jobDescriptor) {
                                $this->copyJobToDescriptor($job, $jobDescriptor);
                                $jobDescriptor->write();
                            }
                            return $buffer;
                        };
                        ob_start($obLogger, 256);

                        try {
                            $job->process();
                        } catch (Exception $e) {
                            // okay, we'll just catch this exception for now
                            $job->addMessage(
                                _t(
                                    __CLASS__ . '.JOB_EXCEPT',
                                    'Job caused exception {message} in {file} at line {line}',
                                    [
                                        'message' => $e->getMessage(),
                                        'file' => $e->getFile(),
                                        'line' => $e->getLine(),
                                    ]
                                )
                            );
                            $this->getLogger()->error(
                                $e->getMessage(),
                                [
                                    'exception' => $e,
                                ]
                            );
                            $jobDescriptor->JobStatus =  QueuedJob::STATUS_BROKEN;
                            $this->extend('updateJobDescriptorAndJobOnException', $jobDescriptor, $job, $e);
                        }

                        ob_end_flush();

                        // now check the job state
                        $data = $job->getJobData();
                        if ($data->currentStep == $lastStepProcessed) {
                            $stallCount++;
                        }

                        if ($stallCount > static::config()->get('stall_threshold')) {
                            $broken = true;
                            $job->addMessage(
                                _t(
                                    __CLASS__ . '.JOB_STALLED',
                                    'Job stalled after {attempts} attempts - please check',
                                    ['attempts' => $stallCount]
                                )
                            );
                            $jobDescriptor->JobStatus = QueuedJob::STATUS_BROKEN;
                        }

                        // now we'll be good and check our memory usage. If it is too high, we'll set the job to
                        // a 'Waiting' state, and let the next processing run pick up the job.
                        if ($this->isMemoryTooHigh()) {
                            $job->addMessage(
                                _t(
                                    __CLASS__ . '.MEMORY_RELEASE',
                                    'Job releasing memory and waiting ({used} used)',
                                    ['used' => $this->humanReadable($this->getMemoryUsage())]
                                )
                            );
                            if ($jobDescriptor->JobStatus != QueuedJob::STATUS_BROKEN) {
                                $jobDescriptor->JobStatus = QueuedJob::STATUS_WAIT;
                            }
                            $broken = true;
                        }

                        // Also check if we are running too long
                        if ($this->hasPassedTimeLimit()) {
                            $job->addMessage(_t(
                                __CLASS__ . '.TIME_LIMIT',
                                'Queue has passed time limit and will restart before continuing'
                            ));
                            if ($jobDescriptor->JobStatus != QueuedJob::STATUS_BROKEN) {
                                $jobDescriptor->JobStatus = QueuedJob::STATUS_WAIT;
                            }
                            $broken = true;
                        }
                    }

                    if ($jobDescriptor) {
                        $this->copyJobToDescriptor($job, $jobDescriptor);
                        $jobDescriptor->write();
                    } else {
                        $this->getLogger()->error(
                            print_r(
                                [
                                    'errno' => 0,
                                    'errstr' => 'Job descriptor has been set to null',
                                    'errfile' => __FILE__,
                                    'errline' => __LINE__,
                                    'errcontext' => [],
                                ],
                                true
                            ),
                            [
                                'file' => __FILE__,
                                'line' => __LINE__,
                            ]
                        );
                        $broken = true;
                    }
                }

                // a last final save. The job is complete by now
                if ($jobDescriptor) {
                    $jobDescriptor->write();
                }

                if ($job->jobFinished()) {
                    /** @var AbstractQueuedJob|QueuedJob $job */
                    $job->afterComplete();
                    $jobDescriptor->cleanupJob();
                }
            } catch (Exception $e) {
                // okay, we'll just catch this exception for now
                $this->getLogger()->error(
                    $e->getMessage(),
                    [
                        'exception' => $e,
                    ]
                );
                $jobDescriptor->JobStatus =  QueuedJob::STATUS_BROKEN;
                $this->extend('updateJobDescriptorAndJobOnException', $jobDescriptor, $job, $e);
                $jobDescriptor->write();
                $broken = true;
            }
        }

        $errorHandler->clear();

        Config::unnest();

        $this->unsetRunAsUser($runAsUser, $originalUser);

        return !$broken;
    }
public function runJob($jobId)
    {
        // first retrieve the descriptor
        /** @var QueuedJobDescriptor $jobDescriptor */
        $jobDescriptor = DataObject::get_by_id(
            QueuedJobDescriptor::class,
            (int)$jobId
        );
        if (!$jobDescriptor) {
            throw new Exception("$jobId is invalid");
        }

        // now lets see whether we have a current user to run as. Typically, if the job is executing via the CLI,
        // we want it to actually execute as the RunAs user - however, if running via the web (which is rare...), we
        // want to ensure that the current user has admin privileges before switching. Otherwise, we just run it
        // as the currently logged in user and hope for the best

        // We need to use $_SESSION directly because SS ties the session to a controller that no longer exists at
        // this point of execution in some circumstances
        $originalUserID = isset($_SESSION['loggedInAs']) ? $_SESSION['loggedInAs'] : 0;
        /** @var Member|null $originalUser */
        $originalUser = $originalUserID
            ? DataObject::get_by_id(Member::class, $originalUserID)
            : null;
        $runAsUser = null;

        // If the Job has requested that we run it as a particular user, then we should try and do that.
        if ($jobDescriptor->RunAs() !== null) {
            $runAsUser = $this->setRunAsUser($jobDescriptor->RunAs(), $originalUser);
        }

        // set up a custom error handler for this processing
        $errorHandler = new JobErrorHandler();

        $job = null;

        $broken = false;

        // Push a config context onto the stack for the duration of this job run.
        Config::nest();

        if ($this->grabMutex($jobDescriptor)) {
            try {
                $job = $this->initialiseJob($jobDescriptor);

                // get the job ready to begin.
                if (!$jobDescriptor->JobStarted) {
                    $jobDescriptor->JobStarted = DBDatetime::now()->Rfc2822();
                } else {
                    $jobDescriptor->JobRestarted = DBDatetime::now()->Rfc2822();
                }

                // Only write to job as "Running" if 'isComplete' was NOT set to true
                // during setup() or prepareForRestart()
                if (!$job->jobFinished()) {
                    $jobDescriptor->JobStatus = QueuedJob::STATUS_RUN;
                    $jobDescriptor->write();
                }

                $lastStepProcessed = 0;
                // have we stalled at all?
                $stallCount = 0;

                if (class_exists(Subsite::class) && !empty($job->SubsiteID)) {
                    Subsite::changeSubsite($job->SubsiteID);

                    // lets set the base URL as far as Director is concerned so that our URLs are correct
                    /** @var Subsite $subsite */
                    $subsite = DataObject::get_by_id(Subsite::class, $job->SubsiteID);
                    if ($subsite && $subsite->exists()) {
                        $domain = $subsite->domain();
                        $base = rtrim(Director::protocol() . $domain, '/') . '/';

                        Config::modify()->set(Director::class, 'alternate_base_url', $base);
                    }
                }

                // while not finished
                while (!$job->jobFinished() && !$broken) {
                    // see that we haven't been set to 'paused' or otherwise by another process
                    /** @var QueuedJobDescriptor $jobDescriptor */
                    $jobDescriptor = DataObject::get_by_id(
                        QueuedJobDescriptor::class,
                        (int)$jobId
                    );
                    if (!$jobDescriptor || !$jobDescriptor->exists()) {
                        $broken = true;
                        $this->getLogger()->error(
                            print_r(
                                [
                                    'errno' => 0,
                                    'errstr' => 'Job descriptor ' . $jobId . ' could not be found',
                                    'errfile' => __FILE__,
                                    'errline' => __LINE__,
                                    'errcontext' => [],
                                ],
                                true
                            ),
                            [
                                'file' => __FILE__,
                                'line' => __LINE__,
                            ]
                        );
                        break;
                    }
                    if ($jobDescriptor->JobStatus != QueuedJob::STATUS_RUN) {
                        // we've been paused by something, so we'll just exit
                        $job->addMessage(_t(
                            __CLASS__ . '.JOB_PAUSED',
                            'Job paused at {time}',
                            ['time' => DBDatetime::now()->Rfc2822()]
                        ));
                        $broken = true;
                    }

                    if (!$broken) {
                        // Collect output as job messages as well as sending it to the screen
                        $obLogger = function ($buffer, $phase) use ($job, $jobDescriptor) {
                            $job->addMessage($buffer);
                            if ($jobDescriptor) {
                                $this->copyJobToDescriptor($job, $jobDescriptor);
                                $jobDescriptor->write();
                            }
                            return $buffer;
                        };
                        ob_start($obLogger, 256);

                        try {
                            $job->process();
                        } catch (Exception $e) {
                            // okay, we'll just catch this exception for now
                            $job->addMessage(
                                _t(
                                    __CLASS__ . '.JOB_EXCEPT',
                                    'Job caused exception {message} in {file} at line {line}',
                                    [
                                        'message' => $e->getMessage(),
                                        'file' => $e->getFile(),
                                        'line' => $e->getLine(),
                                    ]
                                )
                            );
                            $this->getLogger()->error(
                                $e->getMessage(),
                                [
                                    'exception' => $e,
                                ]
                            );
                            $jobDescriptor->JobStatus =  QueuedJob::STATUS_BROKEN;
                            $this->extend('updateJobDescriptorAndJobOnException', $jobDescriptor, $job, $e);
                        }

                        ob_end_flush();

                        // now check the job state
                        $data = $job->getJobData();
                        if ($data->currentStep == $lastStepProcessed) {
                            $stallCount++;
                        }

                        if ($stallCount > static::config()->get('stall_threshold')) {
                            $broken = true;
                            $job->addMessage(
                                _t(
                                    __CLASS__ . '.JOB_STALLED',
                                    'Job stalled after {attempts} attempts - please check',
                                    ['attempts' => $stallCount]
                                )
                            );
                            $jobDescriptor->JobStatus = QueuedJob::STATUS_BROKEN;
                        }

                        // now we'll be good and check our memory usage. If it is too high, we'll set the job to
                        // a 'Waiting' state, and let the next processing run pick up the job.
                        if ($this->isMemoryTooHigh()) {
                            $job->addMessage(
                                _t(
                                    __CLASS__ . '.MEMORY_RELEASE',
                                    'Job releasing memory and waiting ({used} used)',
                                    ['used' => $this->humanReadable($this->getMemoryUsage())]
                                )
                            );
                            if ($jobDescriptor->JobStatus != QueuedJob::STATUS_BROKEN) {
                                $jobDescriptor->JobStatus = QueuedJob::STATUS_WAIT;
                            }
                            $broken = true;
                        }

                        // Also check if we are running too long
                        if ($this->hasPassedTimeLimit()) {
                            $job->addMessage(_t(
                                __CLASS__ . '.TIME_LIMIT',
                                'Queue has passed time limit and will restart before continuing'
                            ));
                            if ($jobDescriptor->JobStatus != QueuedJob::STATUS_BROKEN) {
                                $jobDescriptor->JobStatus = QueuedJob::STATUS_WAIT;
                            }
                            $broken = true;
                        }
                    }

                    if ($jobDescriptor) {
                        $this->copyJobToDescriptor($job, $jobDescriptor);
                        $jobDescriptor->write();
                    } else {
                        $this->getLogger()->error(
                            print_r(
                                [
                                    'errno' => 0,
                                    'errstr' => 'Job descriptor has been set to null',
                                    'errfile' => __FILE__,
                                    'errline' => __LINE__,
                                    'errcontext' => [],
                                ],
                                true
                            ),
                            [
                                'file' => __FILE__,
                                'line' => __LINE__,
                            ]
                        );
                        $broken = true;
                    }
                }

                // a last final save. The job is complete by now
                if ($jobDescriptor) {
                    $jobDescriptor->write();
                }

                if ($job->jobFinished()) {
                    /** @var AbstractQueuedJob|QueuedJob $job */
                    $job->afterComplete();
                    $jobDescriptor->cleanupJob();
                }
            } catch (Exception $e) {
                // okay, we'll just catch this exception for now
                $this->getLogger()->error(
                    $e->getMessage(),
                    [
                        'exception' => $e,
                    ]
                );
                $jobDescriptor->JobStatus =  QueuedJob::STATUS_BROKEN;
                $this->extend('updateJobDescriptorAndJobOnException', $jobDescriptor, $job, $e);
                $jobDescriptor->write();
                $broken = true;
            }
        }

        $errorHandler->clear();

        Config::unnest();

        $this->unsetRunAsUser($runAsUser, $originalUser);

        return !$broken;
    }
Start the actual execution of a job.
The assumption is the jobID refers to a {@link QueuedJobDescriptor} that is status set as "Queued".

This method will continue executing until the job says it's completed

@param int $jobId
The ID of the job to start executing

@return boolean
@throws Exception