New table to control the fetching process

diff --git a/src/Protocol/ActivityPub/Fetch.php b/src/Protocol/ActivityPub/Fetch.php
new file mode 100644 (file)
index 0000000..198202b
--- /dev/null
+++ b/src/Protocol/ActivityPub/Fetch.php
@@ -0,0 +1,84 @@
+<?php
+/**
+ * @copyright Copyright (C) 2010-2022, the Friendica project
+ *
+ * @license GNU AGPL version 3 or any later version
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ *
+ */
+
+namespace Friendica\Protocol\ActivityPub;
+
+use Friendica\Core\Logger;
+use Friendica\Database\Database;
+use Friendica\Database\DBA;
+use Friendica\Util\DateTimeFormat;
+
+/**
+ * This class handles the fetching of posts
+ */
+class Fetch
+{
+       public static function add(string $url): int
+       {
+               DBA::insert('fetch-entry', ['url' => $url, 'created' => DateTimeFormat::utcNow()], Database::INSERT_IGNORE);
+
+               $fetch = DBA::selectFirst('fetch-entry', ['id'], ['url' => $url]);
+               Logger::debug('Added fetch entry', ['url' => $url, 'fetch' => $fetch]);
+               return $fetch['id'] ?? 0;
+       }
+
+       /**
+        * Set the worker id for the queue entry
+        *
+        * @param array $activity
+        * @param int   $wid
+        * @return void
+        */
+       public static function setWorkerId(string $url, int $wid)
+       {
+               if (empty($url) || empty($wid)) {
+                       return;
+               }
+
+               DBA::update('fetch-entry', ['wid' => $wid], ['url' => $url]);
+               Logger::debug('Worker id set', ['url' => $url, 'wid' => $wid]);
+       }
+
+       /**
+        * Check if there is an assigned worker task
+        *
+        * @param array $activity
+        * @return bool
+        */
+       public static function hasWorker(string $url): bool
+       {
+               $fetch = DBA::selectFirst('fetch-entry', ['id', 'wid'], ['url' => $url]);
+               if (empty($fetch['id'])) {
+                       Logger::debug('No entry found for url', ['url' => $url]);
+                       return false;
+               }
+
+               // We don't have a workerqueue id yet. So most likely is isn't assigned yet.
+               // To avoid the ramping up of another fetch request we simply claim that there is a waiting worker.
+               if (!empty($fetch['id']) && empty($fetch['wid'])) {
+                       Logger::debug('Entry without worker found for url', ['url' => $url]);
+                       return true;
+               }
+
+               return DBA::exists('workerqueue', ['id' => $fetch['wid'], 'done' => false]);
+       }
+
+}

diff --git a/src/Protocol/ActivityPub/Processor.php b/src/Protocol/ActivityPub/Processor.php
index 96308ca2cfdec3cc6c487ce20f96c4e5348cdd46..6d36786a22f41f726a611e7e2469ea7d8d894bcf 100644 (file)
--- a/src/Protocol/ActivityPub/Processor.php
+++ b/src/Protocol/ActivityPub/Processor.php
@@ -333,8 +333,14 @@ class Processor
                        if ($fetch_by_worker) {
                                Logger::notice('Fetching is done by worker.', ['parent' => $activity['reply-to-id'], 'recursion-depth' => $recursion_depth]);
                                $activity['recursion-depth'] = 0;
-                               $wid = Worker::add(PRIORITY_HIGH, 'FetchMissingActivity', $activity['reply-to-id'], $activity, '', Receiver::COMPLETION_AUTO);
-                               Queue::setWorkerId($activity, $wid);
+                               if (!Fetch::hasWorker($activity['reply-to-id'])) {
+                                       Fetch::add($activity['reply-to-id']);
+                                       $wid = Worker::add(PRIORITY_HIGH, 'FetchMissingActivity', $activity['reply-to-id'], $activity, '', Receiver::COMPLETION_AUTO);
+                                       Fetch::setWorkerId($activity['reply-to-id'], $wid);
+                                       Queue::setWorkerId($activity, $wid);
+                               } else {
+                                       Logger::debug('Activity is already in the fetching process', ['url' => $activity['reply-to-id']]);
+                               }
                                if (!empty($conversation)) {
                                        return [];
                                }

diff --git a/static/dbstructure.config.php b/static/dbstructure.config.php
index 79da57a2bbc4fa37b28c61f6891195a826cb29d6..f72ec0802025505709947fd76e0267c8a7a37718 100644 (file)
--- a/static/dbstructure.config.php
+++ b/static/dbstructure.config.php
@@ -55,7 +55,7 @@
 use Friendica\Database\DBA;
 
 if (!defined('DB_UPDATE_VERSION')) {
-       define('DB_UPDATE_VERSION', 1476);
+       define('DB_UPDATE_VERSION', 1477);
 }
 
 return [
@@ -692,6 +692,20 @@ return [
                        "uri-id" => ["UNIQUE", "uri-id"],
                ]
        ],
+       "fetch-entry" => [
+               "comment" => "",
+               "fields" => [
+                       "id" => ["type" => "int unsigned", "not null" => "1", "extra" => "auto_increment", "primary" => "1", "comment" => "sequential ID"],
+                       "url" => ["type" => "varbinary(255)", "comment" => "url that awaiting to be fetched"],
+                       "created" => ["type" => "datetime", "not null" => "1", "default" => DBA::NULL_DATETIME, "comment" => "Creation date of the fetch request"],
+                       "wid" => ["type" => "int unsigned", "foreign" => ["workerqueue" => "id"], "comment" => "Workerqueue id"],               ],
+               "indexes" => [
+                       "PRIMARY" => ["id"],
+                       "url" => ["UNIQUE", "url"],
+                       "created" => ["created"],
+                       "wid" => ["wid"],
+               ]
+       ],
        "fsuggest" => [
                "comment" => "friend suggestion stuff",
                "fields" => [