We are now setting the corresponding worker id

database.sql

@@ -739,10 +739,13 @@ CREATE TABLE IF NOT EXISTS `inbox-entry` (
 	`activity` mediumtext COMMENT 'The JSON activity',
 	`signer` varchar(255) COMMENT '',
 	`push` boolean NOT NULL DEFAULT '0' COMMENT '',
+	`wid` int unsigned COMMENT 'Workerqueue id',
 	 PRIMARY KEY(`id`),
 	 UNIQUE INDEX `activity-id` (`activity-id`),
 	 INDEX `object-id` (`object-id`),
-	 INDEX `received` (`received`)
+	 INDEX `received` (`received`),
+	 INDEX `wid` (`wid`),
+	FOREIGN KEY (`wid`) REFERENCES `workerqueue` (`id`) ON UPDATE RESTRICT ON DELETE CASCADE
 ) DEFAULT COLLATE utf8mb4_general_ci COMMENT='Incoming activity';
 
 --

doc/database/db_inbox-entry.md

@@ -19,6 +19,7 @@ Fields
 | activity           | The JSON activity                    | mediumtext     | YES  |     | NULL    |                |
 | signer             |                                      | varchar(255)   | YES  |     | NULL    |                |
 | push               |                                      | boolean        | NO   |     | 0       |                |
+| wid                | Workerqueue id                       | int unsigned   | YES  |     | NULL    |                |
 
 Indexes
 ------------
@@ -29,6 +30,13 @@ Indexes
 | activity-id | UNIQUE, activity-id |
 | object-id   | object-id           |
 | received    | received            |
+| wid         | wid                 |
 
+Foreign Keys
+------------
+
+| Field | Target Table | Target Field |
+|-------|--------------|--------------|
+| wid | [workerqueue](help/database/db_workerqueue) | id |
 
 Return to [database documentation](help/database)

src/Protocol/ActivityPub/Processor.php

@@ -294,7 +294,8 @@ 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;
-				Worker::add(PRIORITY_HIGH, 'FetchMissingActivity', $activity['reply-to-id'], $activity, '', Receiver::COMPLETION_AUTO);
+				$wid = Worker::add(PRIORITY_HIGH, 'FetchMissingActivity', $activity['reply-to-id'], $activity, '', Receiver::COMPLETION_AUTO);
+				Queue::setWorkerId($activity, $wid);
 				return [];
 			}
 		}

src/Protocol/ActivityPub/Queue.php

@@ -89,6 +89,21 @@ class Queue
 		DBA::delete('inbox-entry', ['id' => $activity['entry-id']]);
 	}
 
+	/**
+	 * Set the worker id for the queue entry
+	 *
+	 * @param array $activity
+	 * @param int   $wid
+	 * @return void
+	 */
+	public static function setWorkerId(array $activity, int $wid)
+	{
+		if (empty($activity['entry-id']) || empty($wid)) {
+			return;
+		}
+		DBA::update('inbox-entry', ['wid' => $wid], ['id' => $activity['entry-id']]);
+	}
+
 	/**
 	 * Process the activity with the given id
 	 *
@@ -129,7 +144,7 @@ class Queue
 	}
 
 	/**
-	 * Process all activities
+	 * Clear old activities
 	 *
 	 * @return void
 	 */

static/dbstructure.config.php

@@ -798,12 +798,13 @@ return [
 			"activity" => ["type" => "mediumtext", "comment" => "The JSON activity"],
 			"signer" => ["type" => "varchar(255)", "comment" => ""],
 			"push" => ["type" => "boolean", "not null" => "1", "default" => "0", "comment" => ""],
-		],
+			"wid" => ["type" => "int unsigned", "foreign" => ["workerqueue" => "id"], "comment" => "Workerqueue id"],		],
 		"indexes" => [
 			"PRIMARY" => ["id"],
 			"activity-id" => ["UNIQUE", "activity-id"],
 			"object-id" => ["object-id"],
 			"received" => ["received"],
+			"wid" => ["wid"],
 		]
 	],
 	"inbox-entry-receiver" => [