Merge pull request #1 from nswdpc/fix-queuedjob

Queued job fixes and enhancements
nswdpc · Dec 2, 2020 · 79d137c · 79d137c
2 parents 4d02912 + 4f727d6
commit 79d137c
Show file tree

Hide file tree

Showing 4 changed files with 232 additions and 46 deletions.
diff --git a/README.md b/README.md
@@ -184,11 +184,25 @@ This is a queued job that can be used to send emails depending on the ```send_vi
 + 'when-attachments' - only when attachments are present, or
 + 'no' - never (in which case messages will never send via a Queued Job)
 
-Relevant messages are handed off to the queued job, which is configured to send after one minute. Once delivered, the message parameters are cleared to reduce space used by large messages.
+Messages are handed off to this queued job, which is configured to send after one minute. Once delivered, the message parameters are cleared to reduce space used by large messages.
+
+This job is marked as 'broken' immediately upon an API or other general error.
 
 ### TruncateJob
 
-Use this job to clear out older MailgunEvent records.
+Use this job to clear out older MailgunEvent webhook records.
+
+### RequeueJob
+
+Use this job to kick broken SendJob instances, which happen from time-to-time due to API or connectivity issues.
+
+This job will:
+1. Take all job descriptor records for SendJob that are Broken
+1. Reset their status, processing counts and worker value to default initial values
+1. Set them to start after a minute
+1. Save the record
+
+On the next queue run, these jobs will attempt to send again.
 
 ## Manual Resubmission
 

diff --git a/src/Exceptions/JobProcessingException.php b/src/Exceptions/JobProcessingException.php
@@ -0,0 +1,9 @@
+<?php
+namespace NSWDPC\Messaging\Mailgun;
+
+/**
+ * This exception is thrown when an error occurs in a job
+ */
+class JobProcessingException extends \Exception
+{
+}
diff --git a/src/Jobs/RequeueJob.php b/src/Jobs/RequeueJob.php
@@ -0,0 +1,90 @@
+<?php
+namespace NSWDPC\Messaging\Mailgun;
+
+use Symbiote\QueuedJobs\Services\AbstractQueuedJob;
+use Symbiote\QueuedJobs\Services\QueuedJob;
+use SilverStripe\Core\Config\Config;
+use Symbiote\QueuedJobs\Services\QueuedJobService;
+use Symbiote\QueuedJobs\DataObjects\QueuedJobDescriptor;
+
+/**
+ * @author James Ellis <james.ellis@dpc.nsw.gov.au>
+ * Job used to requeue SendJob descriptors marked broken
+ * This is helpful if you have high mail traffic and have many broken {@link SendJob} records sitting in the queue
+ */
+class RequeueJob extends AbstractQueuedJob
+{
+
+    public function getTitle()
+    {
+        return _t(
+            __CLASS__ . ".JOB_TITLE",
+            "Re-queue failed attempts to send messages via the Mailgun API"
+        );
+    }
+
+    /**
+     * Attempt to send the message via the Mailgun API
+     */
+    public function process()
+    {
+
+        $descriptors = QueuedJobDescriptor::get()
+            ->filter([
+                'JobStatus' => QueuedJob::STATUS_BROKEN,
+                'Implementation' => SendJob::class
+            ]);
+        $count = $descriptors->count();
+        $kick = $skip = 0;
+        if($count > 0) {
+            $this->totalSteps = $count;
+            foreach($descriptors as $descriptor) {
+
+                $data = @unserialize($descriptor->SavedJobData);
+                if(empty($data->parameters)) {
+                    // parameters cleared so pointless re-queuing
+                    $skip++;
+                    continue;
+                }
+
+                // recreate this job as new
+                $next = new \Datetime();
+                $next->modify('+1 minute');
+
+                $descriptor->StartAfter = $next->format('Y-m-d H:i:s');
+                $descriptor->JobStatus = QueuedJob::STATUS_NEW;
+                $descriptor->StepsProcessed = 0;
+                $descriptor->LastProcessedCount = -1;
+                $descriptor->Worker = null;// clear otherwise job is considered locked
+                $descriptor->write();
+
+                $kick++;
+                $this->currentStep++;
+            }
+
+            $this->addMessage(
+                _t(
+                    __CLASS__ . '.JOB_STATUS',
+                    "Marked {kick}, ignored {skip} broken SendJob descriptors as new",
+                    [
+                        'kick' => $kick,
+                        'skip' => $skip
+                    ]
+                ),
+                "info"
+            );
+
+        } else {
+            $this->addMessage(
+                _t(
+                    __CLASS__ . '.JOB_STATUS_NO_JOBS',
+                    "No jobs can be re-queued"
+                ),
+                "info"
+            );
+        }
+
+        $this->isComplete = true;
+
+    }
+}
diff --git a/src/Jobs/SendJob.php b/src/Jobs/SendJob.php
@@ -7,97 +7,170 @@
 use Symbiote\QueuedJobs\Services\QueuedJob;
 use SilverStripe\Core\Config\Config;
 use Symbiote\QueuedJobs\Services\QueuedJobService;
-use DateTime;
-use DateTimeZone;
-use Exception;
 
 /**
  * @author James Ellis <james.ellis@dpc.nsw.gov.au>
  * Queued Job for sending messages to the Mailgun API
  */
 class SendJob extends AbstractQueuedJob
 {
+
+    /**
+     * Total steps for this job
+     * @var int
+     */
     protected $totalSteps = 1;
 
+    /**
+     * @var \NSWDPC\Messaging\Mailgun\Connector\Message
+     */
+    protected $connector;
+
+    /**
+     * Job type
+     */
     public function getJobType()
     {
-        $this->totalSteps = 1;
         return QueuedJob::QUEUED;
     }
 
     public function getTitle()
     {
-        $to = $this->parameters['to'] ?? 'to not set';
-        $subject = $this->parameters['subject'] ?? 'subject not set';
-        $from = $this->parameters['from'] ?? 'from not set';
-        $testmode = $this->parameters['o:testmode'] ?? 'no';
-        return "Email via Mailgun To: '{$to}' From: '{$from}' Subject: '{$subject}' TestMode: '{$testmode}'";
+        $parameters = $this->parameters;
+        $to = $parameters['to'] ?? 'to not set';
+        $subject = $parameters['subject'] ?? 'subject not set';
+        $from = $parameters['from'] ?? 'from not set';
+        $testmode = $parameters['o:testmode'] ?? 'no';
+        return _t(
+            __CLASS__ . ".JOB_TITLE",
+            "Email via Mailgun To: '{to}' From: '{from}' Subject: '{subject}' Test mode: '{testmode}'",
+            [
+                'to' => $to,
+                'from' => $from,
+                'subject' => $subject,
+                'testmode' => $testmode
+            ]
+        );
     }
 
+    /**
+     * The job signature is a combination of the API domain and the job parameters
+     */
     public function getSignature()
     {
         return md5($this->domain . ":" . serialize($this->parameters));
     }
 
+    /**
+     * Create the job
+     * @param string domain DEPRECATED
+     * @param array parameters for Mailgun API
+     */
     public function __construct($domain = "", $parameters = [])
     {
-        if (!$domain) {
-            return;
-        }
-        if (empty($parameters)) {
-            return;
+        $this->connector = new MessageConnector;
+        $this->domain = $this->connector->getApiDomain();
+        if(!empty($parameters)) {
+            $this->parameters = $parameters;
         }
-        $this->domain = $domain;
-        $this->parameters = $parameters;
     }
 
     /**
-     * polls for 'failed' events in the last day and tries to resubmit them
+     * Attempt to send the message via the Mailgun API
      */
     public function process()
     {
 
-        if ($this->isComplete) {
-            return;
-        }
+        try {
 
-        $this->currentStep += 1;
+            if ($this->isComplete) {
+                // the job has already been marked complete
+                return;
+            }
 
-        $connector = new MessageConnector;
-        $client = $connector->getClient();
+            $this->currentStep++;
 
-        $domain = $this->domain;
-        $parameters = $this->parameters;
+            $client = $this->connector->getClient();
+            $domain = $this->connector->getApiDomain();
 
-        if (!$domain || empty($parameters)) {
-            $msg = "SendJob is missing either the domain or parameters properties";
-            $this->messages[] = $msg;
-            throw new Exception($msg);
-        }
+            if (!$domain) {
+                $msg = _t(
+                    __CLASS__ . ".MISSING_API_DOMAIN",
+                    "Mailgun configuration is missing the Mailgun API domain value"
+                );
+                throw new JobProcessingException($msg);
+            }
+
+            $parameters = $this->parameters;
+            if(empty($parameters)) {
+                $msg = _t(
+                    __CLASS__ . ".EMPTY_PARAMS",
+                    "Mailgun SendJob was called with empty parameters"
+                );
+                throw new JobProcessingException($msg);
+            }
 
-        $msg = "Unknown error";
-        try {
             // if required, apply the default recipient
-            $connector->applyDefaultRecipient($parameters);
+            $this->connector->applyDefaultRecipient($parameters);
             // decode all attachments
-            $connector->decodeAttachments($parameters);
+            $this->connector->decodeAttachments($parameters);
             // send directly via the API client
             $response = $client->messages()->send($domain, $parameters);
             $message_id = "";
+
             if ($response && ($response instanceof SendResponse) && ($message_id = $response->getId())) {
-                $message_id = $connector::cleanMessageId($message_id);
-                $this->parameters = [];//remove all params
-                $msg = "OK {$message_id}";
-                $this->messages[] = $msg;
+                $message_id = MessageConnector::cleanMessageId($message_id);
+                $this->parameters = [];//remove all params once message is Accepted by Mailgun
+                $this->addMessage("OK {$message_id}", "INFO");
                 $this->isComplete = true;
                 return;
             }
-            throw new Exception("SendJob invalid response or no message.id returned");
-        } catch (Exception $e) {
-            // API level errors caught here
-            $msg = $e->getMessage();
+
+            // handle if the response is invalid
+            throw new JobProcessingException(
+                $this->addMessage(
+                    _t(
+                        __CLASS__ . ".SEND_INVALID_RESPONSE_FROM_MAILGUN",
+                        "SendJob invalid response or no message.id returned"
+                    )
+                )
+            );
+
+        } catch (JobProcessingException $e) {
+            $this->addMessage(
+                _t(
+                    __CLASS__ . ".SEND_EXCEPTON",
+                    "Mailgun send processing exception: {error}",
+                    [
+                        "error" => $e->getMessage()
+                    ]
+                ),
+                "ERROR"
+            );
+        } catch (\Exception $e) {
+            $this->addMessage(
+                _t(
+                    __CLASS__ . ".GENERAL_EXCEPTON",
+                    "Mailgun send general exception: {error}",
+                    [
+                        "error" => $e->getMessage()
+                    ]
+                ),
+                "ERROR"
+            );
         }
-        $this->messages[] = $msg;
-        throw new Exception($msg);
+
+        /**
+         * Mark the job as broken. This avoids repeated requests to the  API
+         * for the same send attempt and possibly cause quota issues.
+         * Semd attempts that arrive here need to be manually re-queued
+         */
+        throw new \Exception(
+            _t(
+                __CLASS__ . ".MAILGUN_SEND_FAILED",
+                "Mailgun send failed. Check status.mailgun.com or connectivity?"
+            )
+        );
+
     }
 }