Differential D11419 Diff 27435 src/infrastructure/daemon/workers/PhabricatorTriggerDaemon.php

Changeset View

Standalone View

src/infrastructure/daemon/workers/PhabricatorTriggerDaemon.php

This file was added.

				<?php

				/**
				* Schedule and execute event triggers, which run code at specific times.
				*/
				final class PhabricatorTriggerDaemon
				extends PhabricatorDaemon {

				const COUNTER_VERSION = 'trigger.version';
				const COUNTER_CURSOR = 'trigger.cursor';

				protected function run() {

				// The trigger daemon is a low-level infrastructure daemon which schedules
				// and executes chronological events. Examples include a subscription which
				// generates a bill on the 12th of every month, or a reminder email 15
				// minutes before a meeting.

				// Only one trigger daemon can run at a time, and very little work should
				// happen in the daemon process. In general, triggered events should
				// just schedule a task into the normal daemon worker queue and then
				// return. This allows the real work to take longer to execute without
				// disrupting other triggers.

				// The trigger mechanism guarantees that events will execute exactly once,
				// but does not guarantee that they will execute at precisely the specified
				// time. Under normal circumstances, they should execute within a minute or
				// so of the desired time, so this mechanism can be used for things like
				// meeting reminders.

				// If the trigger queue backs up (for example, because it is overwhelmed by
				// trigger updates, doesn't run for a while, or a trigger action is written
				// inefficiently) or the daemon queue backs up (usually for similar
				// reasons), events may execute an arbitrarily long time after they were
				// scheduled to execute. In some cases (like billing a subscription) this
				// may be desirable; in other cases (like sending a meeting reminder) the
				// action may want to check the current time and see if the event is still
				// relevant.

				// The trigger daemon works in two phases:
				//
				// 1. A scheduling phase processes recently updated triggers and
				// schedules them for future execution. For example, this phase would
				// see that a meeting trigger had been changed recently, determine
				// when the reminder for it should execute, and then schedule the
				// action to execute at that future date.
				// 2. An execution phase runs the actions for any scheduled events which
				// are due to execute.
				//
				// The major goal of this design is to deliver on the guarantee that events
				// will execute exactly once. It prevents race conditions in scheduling
				// and execution by ensuring there is only one writer for either of these
				// phases. Without this separation of responsibilities, web processes
				// trying to reschedule events after an update could race with other web
				// processes or the daemon.

				do {
				$lock = PhabricatorGlobalLock::newLock('trigger');

				try {
				$lock->lock(5);
				} catch (PhutilLockException $ex) {
				throw new PhutilProxyException(
				pht(
				'Another process is holding the trigger lock. Usually, this '.
				'means another copy of the trigger daemon is running elsewhere. '.
				'Multiple processes are not permitted to update triggers '.
				'simultaneously.'),
				$ex);
				}

				// Run the scheduling phase. This finds updated triggers which we have
				// not scheduled yet and schedules them.
				$last_version = $this->loadCurrentCursor();
				$head_version = $this->loadCurrentVersion();

				// The cursor points at the next record to process, so we can only skip
				// this step if we're ahead of the version number.
				if ($last_version <= $head_version) {
				$this->scheduleTriggers($last_version);
				}

				// Run the execution phase. This finds events which are due to execute
				// and runs them.
				$this->executeTriggers();

				$lock->unlock();

				$this->sleep($this->getSleepDuration());
				} while (!$this->shouldExit());
				}


				/**
				* Process all of the triggers which have been updated since the last time
				* the daemon ran, scheduling them into the event table.
				*
				* @param int Cursor for the next version update to process.
				* @return void
				epriestleyAuthorUnsubmitted Not Done Inline Actions I'll type some more text here. epriestley: I'll type some more text here.
				*/
				private function scheduleTriggers($cursor) {
				$limit = 100;

				$query = id(new PhabricatorWorkerTriggerQuery())
				->withVersionBetween($cursor, null)
				->setOrder(PhabricatorWorkerTriggerQuery::ORDER_VERSION)
				->needEvents(true)
				->setLimit($limit);
				while (true) {
				$triggers = $query->execute();

				foreach ($triggers as $trigger) {
				$event = $trigger->getEvent();
				if ($event) {
				$last_epoch = $event->getLastEventEpoch();
				} else {
				$last_epoch = null;
				}

				$next_epoch = $trigger->getNextEventEpoch(
				$last_epoch,
				$is_reschedule = false);

				$new_event = PhabricatorWorkerTriggerEvent::initializeNewEvent($trigger)
				->setLastEventEpoch($last_epoch)
				->setNextEventEpoch($next_epoch);

				$new_event->openTransaction();
				if ($event) {
				$event->delete();
				}

				// Always save the new event. Note that we save it even if the next
				// epoch is `null`, indicating that it will never fire, because we
				// would lose the last epoch information if we delete it.
				//
				// In particular, some events may want to execute exactly once.
				// Retaining the last epoch allows them to do this, even if the
				// trigger is updated.
				$new_event->save();

				// Move the cursor forward to make sure we don't reprocess this
				// trigger until it is updated again.
				$this->updateCursor($trigger->getTriggerVersion() + 1);
				$new_event->saveTransaction();
				}

				// If we saw fewer than a full page of updated triggers, we're caught
				// up, so we can move on to the execution phase.
				if (count($triggers) < $limit) {
				break;
				}

				// Otherwise, skip past the stuff we just processed and grab another
				// page of updated triggers.
				$min = last($triggers)->getTriggerVersion() + 1;
				$query->withVersionBetween($min, null);

				$this->stillWorking();
				}
				}


				/**
				* Run scheduled event triggers which are due for execution.
				*
				* @return void
				*/
				private function executeTriggers() {

				// We run only a limited number of triggers before ending the execution
				// phase. If we ran until exhaustion, we could end up executing very
				// out-of-date triggers if there was a long backlog: trigger changes
				// during this phase are not reflected in the event table until we run
				// another scheduling phase.

				// If we exit this phase with triggers still ready to execute we'll
				// jump back into the scheduling phase immediately, so this just makes
				// sure we don't spend an unreasonably long amount of time without
				// processing trigger updates and doing rescheduling.

				$limit = 100;
				$now = PhabricatorTime::getNow();

				$triggers = id(new PhabricatorWorkerTriggerQuery())
				->setOrder(PhabricatorWorkerTriggerQuery::ORDER_EXECUTION)
				->withNextEventBetween(null, $now)
				->needEvents(true)
				->setLimit($limit)
				->execute();
				foreach ($triggers as $trigger) {
				$event = $trigger->getEvent();

				// Execute the trigger action.
				$trigger->executeTrigger(
				$event->getLastEventEpoch(),
				$event->getNextEventEpoch());

				// Now that we've executed the trigger, the current trigger epoch is
				// going to become the last epoch.
				$last_epoch = $event->getNextEventEpoch();

				// If this is a recurring trigger, give it an opportunity to reschedule.
				$reschedule_epoch = $trigger->getNextEventEpoch(
				epriestleyAuthorUnsubmitted Not Done Inline Actions This seems ungrammatical. epriestley: This seems ungrammatical.
				btrahanUnsubmitted Not Done Inline Actions Maybe "Don't reschedule events unless the reschedule epoch is in the future." and maybe rename $next_epoch to $reschedule_epoch ? btrahan: Maybe "Don't reschedule events unless the reschedule epoch is in the future." and maybe rename…
				$last_epoch,
				$is_reschedule = true);

				// Don't reschedule events unless the next occurrence is in the future.
				if (($reschedule_epoch !== null) &&
				($last_epoch !== null) &&
				($reschedule_epoch <= $last_epoch)) {
				throw new Exception(
				pht(
				'Trigger is attempting to perform a routine reschedule where '.
				'the next event (at %s) does not occur after the previous event '.
				'(at %s). Routine reschedules must strictly move event triggers '.
				'forward through time to avoid executing a trigger an infinite '.
				'number of times instantaneously.',
				$reschedule_epoch,
				epriestleyAuthorUnsubmitted Not Done Inline Actions This should be `$last_epoch`. epriestley: This should be `$last_epoch`.
				$last_epoch));
				}

				$new_event = PhabricatorWorkerTriggerEvent::initializeNewEvent($trigger)
				->setLastEventEpoch($last_epoch)
				->setNextEventEpoch($reschedule_epoch);

				$event->openTransaction();
				// Remove the event we just processed.
				$event->delete();
				epriestleyAuthorUnsubmitted Not Done Inline Actions I need to always save this so we retain the last execution time in case triggers are later updated and the clock rule depends on the previous execution epoch (the scheduling phase does this correctly). epriestley: I need to always save this so we retain the last execution time in case triggers are later…

				// See note in the scheduling phase about this; we save the new event
				// even if the next epoch is `null`.
				$new_event->save();
				$event->saveTransaction();
				}
				}


				/**
				* Get the number of seconds to sleep for before starting the next scheduling
				* phase.
				*
				* If no events are scheduled soon, we'll sleep for 60 seconds. Otherwise,
				* we'll sleep until the next scheduled event.
				*
				* @return int Number of seconds to sleep for.
				*/
				private function getSleepDuration() {
				$sleep = 60;

				$next_triggers = id(new PhabricatorWorkerTriggerQuery())
				->setOrder(PhabricatorWorkerTriggerQuery::ORDER_EXECUTION)
				->setLimit(1)
				->needEvents(true)
				->execute();
				if ($next_triggers) {
				$next_trigger = head($next_triggers);
				$next_epoch = $next_trigger->getEvent()->getNextEventEpoch();
				$until = max(0, $next_epoch - PhabricatorTime::getNow());
				$sleep = min($sleep, $until);
				}

				return $sleep;
				}


				/* -( Counters )----------------------------------------------------------- */


				private function loadCurrentCursor() {
				return $this->loadCurrentCounter(self::COUNTER_CURSOR);
				}

				private function loadCurrentVersion() {
				return $this->loadCurrentCounter(self::COUNTER_VERSION);
				}

				private function updateCursor($value) {
				LiskDAO::overwriteCounterValue(
				id(new PhabricatorWorkerTrigger())->establishConnection('w'),
				self::COUNTER_CURSOR,
				$value);
				}

				private function loadCurrentCounter($counter_name) {
				return (int)LiskDAO::loadCurrentCounterValue(
				id(new PhabricatorWorkerTrigger())->establishConnection('w'),
				$counter_name);
				}
				}