add logging for upload queue processing (WP-1014)

inc/Smartling/Base/SmartlingEntityAbstract.php

@@ -14,6 +14,7 @@ abstract class SmartlingEntityAbstract implements SmartlingTableDefinitionInterf
 
     public const DB_TYPE_U_BIGINT = 'INT(20) UNSIGNED NOT NULL'; // BIGINT alias of INT(20)
     public const DB_TYPE_DATETIME = 'DATETIME NOT NULL DEFAULT \'0000-00-00 00:00:00\'';
+    public const DB_TYPE_DATETIME_NULL = 'DATETIME NULL DEFAULT NULL';
     public const DB_TYPE_STRING_STANDARD = 'VARCHAR(255) NOT NULL';
     public const DB_TYPE_STRING_64 = 'VARCHAR(64) NOT NULL';
     public const DB_TYPE_STRING_SMALL = 'VARCHAR(16) NOT NULL';

inc/Smartling/DbAl/Migrations/Migration260825.php

@@ -0,0 +1,50 @@
+<?php
+
+namespace Smartling\DbAl\Migrations;
+
+use Smartling\Base\SmartlingEntityAbstract;
+use Smartling\DbAl\DB;
+use Smartling\Models\UploadQueueEntity;
+
+/**
+ * Adds claim tracking to the upload queue.
+ *
+ * Queue rows used to be deleted the moment they were handed to the upload job, so a
+ * fatal error, timeout or out of memory during the upload that followed destroyed the
+ * queued work without a trace. Rows are now claimed instead, and only removed once the
+ * upload has been accounted for.
+ */
+class Migration260825 implements SmartlingDbMigrationInterface
+{
+    public function getVersion(): int
+    {
+        return 260825;
+    }
+
+    public function getQueries($tablePrefix = 'wp_'): array
+    {
+        $db = new DB();
+        $tableName = $db->completeTableName(UploadQueueEntity::getTableName());
+
+        // Migration240315 (re)creates this table with `CREATE TABLE IF NOT EXISTS` from the
+        // current, evolving UploadQueueEntity::getFieldDefinitions(). A site upgrading from a
+        // schema version older than 240315 runs that migration first, and it already creates
+        // the table with the columns below, so blindly adding them here would fail with a
+        // duplicate column error. Only add what isn't already there.
+        $existingColumns = $db->getColumnArray("SHOW COLUMNS FROM `$tableName`");
+
+        $columns = [
+            UploadQueueEntity::FIELD_CLAIMED => SmartlingEntityAbstract::DB_TYPE_DATETIME_NULL,
+            UploadQueueEntity::FIELD_ATTEMPTS => SmartlingEntityAbstract::DB_TYPE_U_BIGINT . ' ' . SmartlingEntityAbstract::DB_TYPE_DEFAULT_ZERO,
+        ];
+
+        $queries = [];
+        foreach ($columns as $column => $definition) {
+            if (!in_array($column, $existingColumns, true)) {
+                $queries[] = sprintf('ALTER TABLE `%s` ADD COLUMN `%s` %s', $tableName, $column, $definition);
+            }
+        }
+
+        return $queries;
+    }
+}

inc/Smartling/DbAl/UploadQueueManager.php

@@ -23,6 +23,11 @@
 
 class UploadQueueManager {
     use LoggerSafeTrait;
+
+    public const MAX_ATTEMPTS = 3;
+
+    public const STALE_CLAIM_SECONDS = 900;
+
     private string $tableName;
     public function __construct(
         private ApiWrapperInterface $api,
@@ -42,13 +47,27 @@ public function count(): int
 
     public function dequeue(int $blogId): ?UploadQueueItem
     {
-        // Get queue items with first submission having its source blog id = $blogId.
-        // It's impossible to create a single queue item with submissions from multiple source blog ids,
+        // Get queue items with the first submission having its source blog id = $blogId.
+        // It should be impossible to create a single queue item with submissions from multiple source blog ids,
         // so only checking one is enough.
+        $staleClaimCondition = new ConditionBlock(ConditionBuilder::CONDITION_BLOCK_LEVEL_OPERATOR_OR);
+        $staleClaimCondition->addCondition(new Condition(
+            ConditionBuilder::CONDITION_IS_NULL,
+            'q.' . UploadQueueEntity::FIELD_CLAIMED,
+            [],
+            false,
+        ));
+        $staleClaimCondition->addCondition(new Condition(
+            ConditionBuilder::CONDITION_SIGN_LESS,
+            'q.' . UploadQueueEntity::FIELD_CLAIMED,
+            $this->getStaleClaimThreshold(),
+            false,
+        ));
+
         $query = sprintf(<<<'SQL'
-select q.%1$s, q.%2$s, q.%3$s from %7$s q left join %8$s s
+select q.%1$s, q.%2$s, q.%3$s, q.%9$s, q.%10$s from %7$s q left join %8$s s
     on if(locate(',', q.%2$s), left(%2$s, locate(',', %2$s) - 1), %2$s) = s.%4$s
-    where s.%5$s = %6$d
+    where s.%5$s = %6$d and %11$s
 SQL,
             UploadQueueEntity::FIELD_ID,
             UploadQueueEntity::FIELD_SUBMISSION_IDS,
@@ -58,32 +77,122 @@ public function dequeue(int $blogId): ?UploadQueueItem
             $blogId,
             $this->db->completeTableName(UploadQueueEntity::getTableName()),
             $this->db->completeTableName(SubmissionEntity::getTableName()),
+            UploadQueueEntity::FIELD_CLAIMED,
+            UploadQueueEntity::FIELD_ATTEMPTS,
+            $staleClaimCondition,
         );
         while (($row = $this->db->getRowArray($query)) !== null) {
-            $this->delete($row[UploadQueueEntity::FIELD_ID]);
+            $queueId = (int)$row[UploadQueueEntity::FIELD_ID];
+            $attempts = (int)($row[UploadQueueEntity::FIELD_ATTEMPTS] ?? 0);
             $locales = new IntStringPairCollection();
-            $submissions = [];
+            $existingSubmissions = [];
+            $unprocessable = false;
             foreach (IntegerIterator::fromString($row[UploadQueueEntity::FIELD_SUBMISSION_IDS]) as $submissionId) {
-                $submission = $this->submissionManager->getEntityById($submissionId);
-                if ($submission === null) {
-                    continue 2;
+                try {
+                    $submission = $this->submissionManager->getEntityById($submissionId);
+                    if ($submission === null) {
+                        $this->getLogger()->warning("Discarding upload queue item id=$queueId: submissionId=$submissionId no longer exists");
+                        $unprocessable = true;
+                        continue;
+                    }
+
+                    $existingSubmissions[] = $submission;
+
+                    $locale = $this->getSmartlingLocale($submission);
+                    if ($locale === null) {
+                        $this->getLogger()->warning("Discarding upload queue item id=$queueId: unable to resolve target locale for submissionId=$submissionId, targetBlogId={$submission->getTargetBlogId()}");
+                        $unprocessable = true;
+                        continue;
+                    }
+
+                    $locales = $locales->add([new IntStringPair($submission->getId(), $locale)]);
+                } catch (\Throwable $e) {
+                    $this->getLogger()->warning("Discarding upload queue item id=$queueId: failed to resolve submissionId=$submissionId: {$e->getMessage()}");
+                    $unprocessable = true;
                 }
+            }
 
-                $locale = $this->getSmartlingLocale($submission);
-                if ($locale === null) {
-                    continue 2;
+            if ($unprocessable) {
+                if (!$this->discardQueueItem(
+                    $queueId,
+                    $existingSubmissions,
+                    'Upload queue item discarded: unable to resolve one or more submissions grouped with this item, see log for details.',
+                )) {
+                    return null;
                 }
+                continue;
+            }
 
-                $locales = $locales->add([new IntStringPair($submission->getId(), $locale)]);
-                $submissions[] = $submission;
+            if ($attempts >= self::MAX_ATTEMPTS) {
+                $message = sprintf(
+                    'Upload abandoned after %d attempts. The upload process most likely terminated unexpectedly (fatal error, timeout or out of memory) while handling this content.',
+                    $attempts,
+                );
+                $this->getLogger()->error("Failing upload queue item id=$queueId: $message");
+                if (!$this->discardQueueItem($queueId, $existingSubmissions, $message)) {
+                    return null;
+                }
+                continue;
             }
 
-            return new UploadQueueItem($submissions, $row[UploadQueueEntity::FIELD_BATCH_UID], $locales);
+            if (!$this->claim($queueId, $attempts)) {
+                $this->getLogger()->error("Failed to claim upload queue item id=$queueId");
+                return null;
+            }
+
+            return new UploadQueueItem($existingSubmissions, $row[UploadQueueEntity::FIELD_BATCH_UID], $locales, $queueId);
         }
 
         return null;
     }
 
+    /**
+     * @param SubmissionEntity[] $submissions
+     * @return bool Whether the row was actually removed from the queue.
+     */
+    private function discardQueueItem(int $queueId, array $submissions, string $errorMessage): bool
+    {
+        foreach ($submissions as $submission) {
+            $this->submissionManager->setErrorMessage($submission, $errorMessage);
+        }
+        if (!$this->delete($queueId)) {
+            $this->getLogger()->error("Failed to delete upload queue item id=$queueId after discarding it");
+            return false;
+        }
+
+        return true;
+    }
+
+    private function getStaleClaimThreshold(): string
+    {
+        return DateTimeHelper::dateTimeToString(
+            (new \DateTime('now', DateTimeHelper::getDefaultTimezone()))
+                ->modify('-' . self::STALE_CLAIM_SECONDS . ' seconds')
+        );
+    }
+
+    public function complete(UploadQueueItem $item): void
+    {
+        if (!$this->delete($item->getId())) {
+            $this->getLogger()->error("Failed to delete completed upload queue item id={$item->getId()}");
+        }
+    }
+
+    /**
+     * @return bool Whether the row was actually claimed.
+     */
+    private function claim(int $id, int $attempts): bool
+    {
+        return $this->db->query(QueryBuilder::buildUpdateQuery(
+            $this->tableName,
+            [
+                UploadQueueEntity::FIELD_CLAIMED => DateTimeHelper::nowAsString(),
+                UploadQueueEntity::FIELD_ATTEMPTS => $attempts + 1,
+            ],
+            $this->idCondition($id),
+        )) !== false;
+    }
+
     public function enqueue(IntegerIterator $submissionIds, string $batchUid): void
     {
         $this->db->withTransaction(function () use ($batchUid, $submissionIds) {
@@ -167,12 +276,20 @@ private function getSmartlingLocale(SubmissionEntity $submission): ?string
         return null;
     }
 
-    private function delete(int $id): void
+    /**
+     * @return bool Whether the row was actually removed.
+     */
+    private function delete(int $id): bool
+    {
+        return $this->db->query(QueryBuilder::buildDeleteQuery($this->tableName, $this->idCondition($id))) !== false;
+    }
+
+    private function idCondition(int $id): ConditionBlock
     {
         $block = new ConditionBlock(ConditionBuilder::CONDITION_BLOCK_LEVEL_OPERATOR_AND);
         $block->addCondition(new Condition(ConditionBuilder::CONDITION_SIGN_EQ, UploadQueueEntity::FIELD_ID, $id));
 
-        $this->db->query(QueryBuilder::buildDeleteQuery($this->tableName, $block));
+        return $block;
     }
 
 }

inc/Smartling/DebugTrait.php

@@ -88,28 +88,71 @@ public static function BacktracePrint()
     }
 
     /**
-     * Last chance to know what had happened if Wordpress is down.
+     * Error types that actually terminate the request. Anything else - notices,
+     * warnings, and in particular deprecations - is left alone: error_get_last()
+     * returns the last error of *any* severity, so treating non-fatal types as
+     * fatal reports an emergency on every otherwise healthy request and buries
+     * the real crashes.
      */
-    public function shutdownHandler()
+    private static function fatalErrorTypes(): int
     {
-        $logger = Bootstrap::getLogger();
+        return E_ERROR
+            | E_PARSE
+            | E_CORE_ERROR
+            | E_COMPILE_ERROR
+            | E_USER_ERROR
+            | E_RECOVERABLE_ERROR;
+    }
 
-        $skipLogging = E_NOTICE | E_WARNING | E_USER_NOTICE | E_USER_WARNING | E_STRICT | E_DEPRECATED;
+    private static function errorTypeNames(): array
+    {
+        return [
+            E_ERROR => 'E_ERROR',
+            E_WARNING => 'E_WARNING',
+            E_PARSE => 'E_PARSE',
+            E_NOTICE => 'E_NOTICE',
+            E_CORE_ERROR => 'E_CORE_ERROR',
+            E_CORE_WARNING => 'E_CORE_WARNING',
+            E_COMPILE_ERROR => 'E_COMPILE_ERROR',
+            E_COMPILE_WARNING => 'E_COMPILE_WARNING',
+            E_USER_ERROR => 'E_USER_ERROR',
+            E_USER_WARNING => 'E_USER_WARNING',
+            E_USER_NOTICE => 'E_USER_NOTICE',
+            E_RECOVERABLE_ERROR => 'E_RECOVERABLE_ERROR',
+            E_DEPRECATED => 'E_DEPRECATED',
+            E_USER_DEPRECATED => 'E_USER_DEPRECATED',
+        ];
+    }
 
-        $loggingPattern = E_ALL ^ $skipLogging;
+    public static function isFatalError(?int $errorType): bool
+    {
+        return $errorType !== null && ($errorType & self::fatalErrorTypes()) !== 0;
+    }
 
-        $data = error_get_last();
+    public static function getErrorTypeName(int $errorType): string
+    {
+        return self::errorTypeNames()[$errorType] ?? "UNKNOWN($errorType)";
+    }
 
-        /**
-         * @var int $errorType
-         */
-        $errorType = &$data['type'];
+    /**
+     * Last chance to know what had happened if Wordpress is down.
+     */
+    public function shutdownHandler()
+    {
+        $data = error_get_last();
 
-        if ($errorType & $loggingPattern) {
-            $message = "An Error (0x{$data['type']}) occurred and Wordpress is down.\n";
-            $message .= "Message: '{$data['message']}'\n";
-            $message .= "Location: '{$data['file']}:{$data['line']}'\n";
-            $logger->emergency($message);
+        if (!self::isFatalError($data['type'] ?? null)) {
+            return;
         }
+
+        $message = sprintf(
+            "A fatal error (%s) occurred and Wordpress is down.\nMessage: '%s'\nLocation: '%s:%s'\n",
+            self::getErrorTypeName($data['type']),
+            $data['message'],
+            $data['file'],
+            $data['line'],
+        );
+
+        Bootstrap::getLogger()->emergency($message);
     }
 }

inc/Smartling/Helpers/WordpressFunctionProxyHelper.php

@@ -114,6 +114,11 @@ public function apply_filters()
         return apply_filters(...func_get_args());
     }
 
+    public function do_action()
+    {
+        return do_action(...func_get_args());
+    }
+
     public function delete_post_meta()
     {
         return delete_post_meta(...func_get_args());

inc/Smartling/Jobs/JobAbstract.php

@@ -173,6 +173,7 @@ public function runCronJob(string $source = ''): void
         } catch (\RuntimeException $e) {
             if ($e->getMessage() === self::THROTTLED_MESSAGE) {
                 $message = self::THROTTLED_MESSAGE;
+                $this->getLogger()->debug("Skipping {$this->getJobHookName()} run: throttled");
             } else {
                 throw $e;
             }