Allow TransactionEditor to move publishing work to the daemons

Summary:
Ref T6367. This is similar to D11329, but not quite as ambitious.

Allow Editors to implement `supportsWorkers()` and move their publishing work into a daemon. So far, only Paste supports this.

Most of the complexity here is saving and restoring state across the barrier between the web process and the worker process, but I think this is ~90% of it and then we'll pick up a couple of random things in applications.

I'm primarily trying to keep this as gradual as possible.

Test Plan:
  - Published transactions with and without daemon support.
  - Looked at mail, feed.

Reviewers: btrahan

Reviewed By: btrahan

Subscribers: fabe, epriestley

Maniphest Tasks: T6367

Differential Revision: https://secure.phabricator.com/D13107

src/__phutil_library_map__.php

@@ -1334,6 +1334,7 @@ phutil_register_library_map(array(
     'PhabricatorApplicationTransactionInterface' => 'applications/transactions/interface/PhabricatorApplicationTransactionInterface.php',
     'PhabricatorApplicationTransactionNoEffectException' => 'applications/transactions/exception/PhabricatorApplicationTransactionNoEffectException.php',
     'PhabricatorApplicationTransactionNoEffectResponse' => 'applications/transactions/response/PhabricatorApplicationTransactionNoEffectResponse.php',
+    'PhabricatorApplicationTransactionPublishWorker' => 'applications/transactions/worker/PhabricatorApplicationTransactionPublishWorker.php',
     'PhabricatorApplicationTransactionQuery' => 'applications/transactions/query/PhabricatorApplicationTransactionQuery.php',
     'PhabricatorApplicationTransactionReplyHandler' => 'applications/transactions/replyhandler/PhabricatorApplicationTransactionReplyHandler.php',
     'PhabricatorApplicationTransactionResponse' => 'applications/transactions/response/PhabricatorApplicationTransactionResponse.php',
@@ -4678,6 +4679,7 @@ phutil_register_library_map(array(
     'PhabricatorApplicationTransactionFeedStory' => 'PhabricatorFeedStory',
     'PhabricatorApplicationTransactionNoEffectException' => 'Exception',
     'PhabricatorApplicationTransactionNoEffectResponse' => 'AphrontProxyResponse',
+    'PhabricatorApplicationTransactionPublishWorker' => 'PhabricatorWorker',
     'PhabricatorApplicationTransactionQuery' => 'PhabricatorCursorPagedPolicyAwareQuery',
     'PhabricatorApplicationTransactionReplyHandler' => 'PhabricatorMailReplyHandler',
     'PhabricatorApplicationTransactionResponse' => 'AphrontProxyResponse',

src/applications/paste/editor/PhabricatorPasteEditor.php

@@ -3,8 +3,6 @@
 final class PhabricatorPasteEditor
   extends PhabricatorApplicationTransactionEditor {
 
-  private $pasteFile;
-
   public function getEditorApplicationClass() {
     return 'PhabricatorPasteApplication';
   }
@@ -134,7 +132,7 @@ final class PhabricatorPasteEditor
   protected function getMailTo(PhabricatorLiskDAO $object) {
     return array(
       $object->getAuthorPHID(),
-      $this->requireActor()->getPHID(),
+      $this->getActingAsPHID(),
     );
   }
 
@@ -186,4 +184,8 @@ final class PhabricatorPasteEditor
     return false;
   }
 
+  protected function supportsWorkers() {
+    return true;
+  }
+
 }

src/applications/transactions/editor/PhabricatorApplicationTransactionEditor.php

@@ -5,6 +5,7 @@
  * @task feed Publishing Feed Stories
  * @task search Search Index
  * @task files Integration with Files
+ * @task workers Managing Workers
  */
 abstract class PhabricatorApplicationTransactionEditor
   extends PhabricatorEditor {
@@ -30,6 +31,9 @@ abstract class PhabricatorApplicationTransactionEditor
   private $actingAsPHID;
   private $disableEmail;
 
+  private $heraldEmailPHIDs = array();
+  private $heraldForcedEmailPHIDs = array();
+  private $heraldHeader;
 
   /**
    * Get the class name for the application this editor is a part of.
@@ -861,12 +865,72 @@ abstract class PhabricatorApplicationTransactionEditor
           $object,
           $herald_xactions);
 
+        $adapter = $this->getHeraldAdapter();
+        $this->heraldEmailPHIDs = $adapter->getEmailPHIDs();
+        $this->heraldForcedEmailPHIDs = $adapter->getForcedEmailPHIDs();
+
         // Merge the new transactions into the transaction list: we want to
         // send email and publish feed stories about them, too.
         $xactions = array_merge($xactions, $herald_xactions);
       }
     }
 
+    $this->didApplyTransactions($xactions);
+
+    if ($object instanceof PhabricatorCustomFieldInterface) {
+      // Maybe this makes more sense to move into the search index itself? For
+      // now I'm putting it here since I think we might end up with things that
+      // need it to be up to date once the next page loads, but if we don't go
+      // there we we could move it into search once search moves to the daemons.
+
+      // It now happens in the search indexer as well, but the search indexer is
+      // always daemonized, so the logic above still potentially holds. We could
+      // possibly get rid of this. The major motivation for putting it in the
+      // indexer was to enable reindexing to work.
+
+      $fields = PhabricatorCustomField::getObjectFields(
+        $object,
+        PhabricatorCustomField::ROLE_APPLICATIONSEARCH);
+      $fields->readFieldsFromStorage($object);
+      $fields->rebuildIndexes($object);
+    }
+
+    $herald_xscript = $this->getHeraldTranscript();
+    if ($herald_xscript) {
+      $herald_header = $herald_xscript->getXHeraldRulesHeader();
+      $herald_header = HeraldTranscript::saveXHeraldRulesHeader(
+        $object->getPHID(),
+        $herald_header);
+    } else {
+      $herald_header = HeraldTranscript::loadXHeraldRulesHeader(
+        $object->getPHID());
+    }
+    $this->heraldHeader = $herald_header;
+
+    if ($this->supportsWorkers()) {
+      PhabricatorWorker::scheduleTask(
+        'PhabricatorApplicationTransactionPublishWorker',
+        array(
+          'objectPHID' => $object->getPHID(),
+          'actorPHID' => $this->getActingAsPHID(),
+          'xactionPHIDs' => mpull($xactions, 'getPHID'),
+          'state' => $this->getWorkerState(),
+        ),
+        array(
+          'objectPHID' => $object->getPHID(),
+          'priority' => PhabricatorWorker::PRIORITY_ALERTS,
+        ));
+    } else {
+      $this->publishTransactions($object, $xactions);
+    }
+
+    return $xactions;
+  }
+
+  public function publishTransactions(
+    PhabricatorLiskDAO $object,
+    array $xactions) {
+
     // Before sending mail or publishing feed stories, reload the object
     // subscribers to pick up changes caused by Herald (or by other side effects
     // in various transaction phases).
@@ -901,26 +965,6 @@ abstract class PhabricatorApplicationTransactionEditor
         $mailed);
     }
 
-    $this->didApplyTransactions($xactions);
-
-    if ($object instanceof PhabricatorCustomFieldInterface) {
-      // Maybe this makes more sense to move into the search index itself? For
-      // now I'm putting it here since I think we might end up with things that
-      // need it to be up to date once the next page loads, but if we don't go
-      // there we we could move it into search once search moves to the daemons.
-
-      // It now happens in the search indexer as well, but the search indexer is
-      // always daemonized, so the logic above still potentially holds. We could
-      // possibly get rid of this. The major motivation for putting it in the
-      // indexer was to enable reindexing to work.
-
-      $fields = PhabricatorCustomField::getObjectFields(
-        $object,
-        PhabricatorCustomField::ROLE_APPLICATIONSEARCH);
-      $fields->readFieldsFromStorage($object);
-      $fields->rebuildIndexes($object);
-    }
-
     return $xactions;
   }
 
@@ -2043,11 +2087,8 @@ abstract class PhabricatorApplicationTransactionEditor
     $email_to = $this->getMailTo($object);
     $email_cc = $this->getMailCC($object);
 
-    $adapter = $this->getHeraldAdapter();
-    if ($adapter) {
-      $email_cc = array_merge($email_cc, $adapter->getEmailPHIDs());
-      $email_force = $adapter->getForcedEmailPHIDs();
-    }
+    $email_cc = array_merge($email_cc, $this->heraldEmailPHIDs);
+    $email_force = $this->heraldForcedEmailPHIDs;
 
     $phids = array_merge($email_to, $email_cc);
     $handles = id(new PhabricatorHandleQuery())
@@ -2087,19 +2128,8 @@ abstract class PhabricatorApplicationTransactionEditor
       $template->addAttachment($attachment);
     }
 
-    $herald_xscript = $this->getHeraldTranscript();
-    if ($herald_xscript) {
-      $herald_header = $herald_xscript->getXHeraldRulesHeader();
-      $herald_header = HeraldTranscript::saveXHeraldRulesHeader(
-        $object->getPHID(),
-        $herald_header);
-    } else {
-      $herald_header = HeraldTranscript::loadXHeraldRulesHeader(
-        $object->getPHID());
-    }
-
-    if ($herald_header) {
-      $template->addHeader('X-Herald-Rules', $herald_header);
+    if ($this->heraldHeader) {
+      $template->addHeader('X-Herald-Rules', $this->heraldHeader);
     }
 
     if ($object instanceof PhabricatorProjectInterface) {
@@ -2765,4 +2795,112 @@ abstract class PhabricatorApplicationTransactionEditor
     }
   }
 
+
+/* -(  Workers  )------------------------------------------------------------ */
+
+
+  /**
+   * @task workers
+   */
+  protected function supportsWorkers() {
+    // TODO: Remove this method once everything supports workers.
+    return false;
+  }
+
+
+  /**
+   * Convert the editor state to a serializable dictionary which can be passed
+   * to a worker.
+   *
+   * This data will be loaded with @{method:loadWorkerState} in the worker.
+   *
+   * @return dict<string, wild> Serializable editor state.
+   * @task workers
+   */
+  final private function getWorkerState() {
+    $state = array();
+    foreach ($this->getAutomaticStateProperties() as $property) {
+      $state[$property] = $this->$property;
+    }
+
+    $state += array(
+      'excludeMailRecipientPHIDs' => $this->getExcludeMailRecipientPHIDs(),
+    );
+
+    return $state + array(
+      'custom' => $this->getCustomWorkerState(),
+    );
+  }
+
+
+  /**
+   * Hook; return custom properties which need to be passed to workers.
+   *
+   * @return dict<string, wild> Custom properties.
+   * @task workers
+   */
+  protected function getCustomWorkerState() {
+    return array();
+  }
+
+
+  /**
+   * Load editor state using a dictionary emitted by @{method:getWorkerState}.
+   *
+   * This method is used to load state when running worker operations.
+   *
+   * @param dict<string, wild> Editor state, from @{method:getWorkerState}.
+   * @return this
+   * @task workers
+   */
+  final public function loadWorkerState(array $state) {
+    foreach ($this->getAutomaticStateProperties() as $property) {
+      $this->$property = idx($state, $property);
+    }
+
+    $exclude = idx($state, 'excludeMailRecipientPHIDs', array());
+    $this->setExcludeMailRecipientPHIDs($exclude);
+
+    $custom = idx($state, 'custom', array());
+    $this->loadCustomWorkerState($custom);
+
+    return $this;
+  }
+
+
+  /**
+   * Hook; set custom properties on the editor from data emitted by
+   * @{method:getCustomWorkerState}.
+   *
+   * @param dict<string, wild> Custom state,
+   *   from @{method:getCustomWorkerState}.
+   * @return this
+   * @task workers
+   */
+  protected function loadCustomWorkerState(array $state) {
+    return $this;
+  }
+
+
+  /**
+   * Get a list of object properties which should be automatically sent to
+   * workers in the state data.
+   *
+   * These properties will be automatically stored and loaded by the editor in
+   * the worker.
+   *
+   * @return list<string> List of properties.
+   * @task workers
+   */
+  private function getAutomaticStateProperties() {
+    return array(
+      'parentMessageID',
+      'disableEmail',
+      'isNewObject',
+      'heraldEmailPHIDs',
+      'heraldForcedEmailPHIDs',
+      'heraldHeader',
+    );
+  }
+
 }

src/applications/transactions/worker/PhabricatorApplicationTransactionPublishWorker.php

@@ -0,0 +1,134 @@
+<?php
+
+/**
+ * Performs backgroundable work after applying transactions.
+ *
+ * This class handles email, notifications, feed stories, search indexes, and
+ * other similar backgroundable work.
+ */
+final class PhabricatorApplicationTransactionPublishWorker
+  extends PhabricatorWorker {
+
+
+  /**
+   * Publish information about a set of transactions.
+   */
+  protected function doWork() {
+    $object = $this->loadObject();
+    $editor = $this->buildEditor($object);
+    $xactions = $this->loadTransactions($object);
+
+    $editor->publishTransactions($object, $xactions);
+  }
+
+
+  /**
+   * Load the object the transactions affect.
+   */
+  private function loadObject() {
+    $data = $this->getTaskData();
+    $viewer = PhabricatorUser::getOmnipotentUser();
+
+    $phid = idx($data, 'objectPHID');
+    if (!$phid) {
+      throw new PhabricatorWorkerPermanentFailureException(
+        pht('Task has no object PHID!'));
+    }
+
+    $object = id(new PhabricatorObjectQuery())
+      ->setViewer($viewer)
+      ->withPHIDs(array($phid))
+      ->executeOne();
+    if (!$object) {
+      throw new PhabricatorWorkerPermanentFailureException(
+        pht(
+          'Unable to load object with PHID "%s"!',
+          $phid));
+    }
+
+    return $object;
+  }
+
+
+  /**
+   * Build and configure an Editor to publish these transactions.
+   */
+  private function buildEditor(
+    PhabricatorApplicationTransactionInterface $object) {
+    $data = $this->getTaskData();
+
+    $daemon_source = PhabricatorContentSource::newForSource(
+      PhabricatorContentSource::SOURCE_DAEMON,
+      array());
+
+    $viewer = PhabricatorUser::getOmnipotentUser();
+    $editor = $object->getApplicationTransactionEditor()
+      ->setActor($viewer)
+      ->setContentSource($daemon_source)
+      ->setActingAsPHID(idx($data, 'actorPHID'))
+      ->loadWorkerState(idx($data, 'state', array()));
+
+    return $editor;
+  }
+
+
+  /**
+   * Load the transactions to be published.
+   */
+  private function loadTransactions(
+    PhabricatorApplicationTransactionInterface $object) {
+    $data = $this->getTaskData();
+
+    $xaction_phids = idx($data, 'xactionPHIDs');
+    if (!$xaction_phids) {
+      throw new PhabricatorWorkerPermanentFailureException(
+        pht('Task has no transaction PHIDs!'));
+    }
+
+    $viewer = PhabricatorUser::getOmnipotentUser();
+
+    $type = phid_get_subtype(head($xaction_phids));
+    $xactions = $this->buildTransactionQuery($type)
+      ->setViewer($viewer)
+      ->withPHIDs($xaction_phids)
+      ->needComments(true)
+      ->execute();
+    $xactions = mpull($xactions, null, 'getPHID');
+
+    $missing = array_diff($xaction_phids, array_keys($xactions));
+    if ($missing) {
+      throw new PhabricatorWorkerPermanentFailureException(
+        pht(
+          'Unable to load transactions: %s.',
+          implode(', ', $missing)));
+    }
+
+    return array_select_keys($xactions, $xaction_phids);
+  }
+
+
+  /**
+   * Build a new transaction query of the appropriate class so we can load
+   * the transactions.
+   */
+  private function buildTransactionQuery($type) {
+    $queries = id(new PhutilSymbolLoader())
+      ->setAncestorClass('PhabricatorApplicationTransactionQuery')
+      ->loadObjects();
+
+    foreach ($queries as $query) {
+      $query_type = $query
+        ->getTemplateApplicationTransaction()
+        ->getApplicationTransactionType();
+      if ($query_type == $type) {
+        return $query;
+      }
+    }
+
+    throw new PhabricatorWorkerPermanentFailureException(
+      pht(
+        'Unable to load query for transaction type "%s"!',
+        $type));
+  }
+
+}