Merge pull request #11503 from annando/bulk-delivery

Fix workerqueue entries with wrong priority

src/Core/Worker.php

@@ -1322,7 +1322,7 @@ class Worker
 		$found = DBA::exists('workerqueue', ['command' => $command, 'parameter' => $parameters, 'done' => false]);
 		$added = 0;
 
-		if (!in_array($priority, PRIORITIES)) {
+		if (!is_int($priority) || !in_array($priority, PRIORITIES)) {
 			Logger::warning('Invalid priority', ['priority' => $priority, 'command' => $command, 'callstack' => System::callstack(20)]);
 			$priority = PRIORITY_MEDIUM;
 		}

src/Model/Item.php

@@ -725,7 +725,7 @@ class Item
 		return GRAVITY_UNKNOWN;   // Should not happen
 	}
 
-	public static function insert(array $item, bool $notify = false, bool $post_local = true)
+	public static function insert(array $item, int $notify = 0, bool $post_local = true)
 	{
 		$orig_item = $item;
 
@@ -739,7 +739,7 @@ class Item
 			$item['protocol'] = Conversation::PARCEL_DIRECT;
 			$item['direction'] = Conversation::PUSH;
 
-			if (in_array($notify, PRIORITIES)) {
+			if (is_int($notify) && in_array($notify, PRIORITIES)) {
 				$priority = $notify;
 			}
 		} else {

src/Util/HTTPSignature.php

@@ -29,6 +29,7 @@ use Friendica\Model\APContact;
 use Friendica\Model\Contact;
 use Friendica\Model\ItemURI;
 use Friendica\Model\User;
+use Friendica\Network\HTTPClient\Capability\ICanHandleHttpResponses;
 use Friendica\Network\HTTPClient\Client\HttpClientAccept;
 use Friendica\Network\HTTPClient\Client\HttpClientOptions;
 
@@ -264,21 +265,21 @@ class HTTPSignature
 	 */
 
 	/**
-	 * Transmit given data to a target for a user
+	 * Post given data to a target for a user, returns the result class
 	 *
 	 * @param array   $data   Data that is about to be send
 	 * @param string  $target The URL of the inbox
 	 * @param integer $uid    User id of the sender
 	 *
-	 * @return boolean Was the transmission successful?
+	 * @return ICanHandleHttpResponses
 	 * @throws \Friendica\Network\HTTPException\InternalServerErrorException
 	 */
-	public static function transmit($data, $target, $uid)
+	public static function post(array $data, string $target, int $uid): ICanHandleHttpResponses
 	{
 		$owner = User::getOwnerDataById($uid);
 
 		if (!$owner) {
-			return;
+			return null;
 		}
 
 		$content = json_encode($data, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
@@ -305,16 +306,32 @@ class HTTPSignature
 
 		$headers['Content-Type'] = 'application/activity+json';
 
-		$postResult = DI::httpClient()->post($target, $content, $headers);
+		$postResult = DI::httpClient()->post($target, $content, $headers, DI::config()->get('system', 'curl_timeout'));
 		$return_code = $postResult->getReturnCode();
 
 		Logger::info('Transmit to ' . $target . ' returned ' . $return_code);
 
-		$success = ($return_code >= 200) && ($return_code <= 299);
+		self::setInboxStatus($target, ($return_code >= 200) && ($return_code <= 299));
 
-		self::setInboxStatus($target, $success);
+		return $postResult;
+	}
 
-		return $success;
+	/**
+	 * Transmit given data to a target for a user
+	 *
+	 * @param array   $data   Data that is about to be send
+	 * @param string  $target The URL of the inbox
+	 * @param integer $uid    User id of the sender
+	 *
+	 * @return boolean Was the transmission successful?
+	 * @throws \Friendica\Network\HTTPException\InternalServerErrorException
+	 */
+	public static function transmit(array $data, string $target, int $uid): bool
+	{
+		$postResult = self::post($data, $target, $uid);
+		$return_code = $postResult->getReturnCode();
+
+		return ($return_code >= 200) && ($return_code <= 299);
 	}
 
 	/**

src/Worker/APDelivery.php

@@ -23,6 +23,7 @@ namespace Friendica\Worker;
 
 use Friendica\Core\Logger;
 use Friendica\Core\Worker;
+use Friendica\DI;
 use Friendica\Model\Contact;
 use Friendica\Model\GServer;
 use Friendica\Model\Post;
@@ -65,14 +66,15 @@ class APDelivery
 			return;
 		}
 
-		Logger::info('Invoked', ['cmd' => $cmd, 'inbox' => $inbox, 'id' => $item_id, 'uri-id' => $uri_id, 'uid' => $uid]);
+		Logger::debug('Invoked', ['cmd' => $cmd, 'inbox' => $inbox, 'id' => $item_id, 'uri-id' => $uri_id, 'uid' => $uid]);
 
 		if (empty($uri_id)) {
-			$result = self::deliver($inbox);
+			$result  = self::deliver($inbox);
 			$success = $result['success'];
 			$uri_ids = $result['uri_ids'];
 		} else {
-			$success = self::deliverToInbox($cmd, $item_id, $inbox, $uid, $receivers, $uri_id);
+			$result  = self::deliverToInbox($cmd, $item_id, $inbox, $uid, $receivers, $uri_id);
+			$success = $result['success'];
 			$uri_ids = [$uri_id];
 		}
 
@@ -84,27 +86,39 @@ class APDelivery
 		}
 	}
 
-	private static function deliver(string $inbox)
+	private static function deliver(string $inbox):array
 	{
-		$uri_ids = [];
+		$uri_ids    = [];
-		$posts   = Post\Delivery::selectForInbox($inbox);
+		$posts      = Post\Delivery::selectForInbox($inbox);
+		$serverfail = false;
 
 		foreach ($posts as $post) {
-			if (!self::deliverToInbox($post['command'], 0, $inbox, $post['uid'], $post['receivers'], $post['uri-id'])) {
+			if (!$serverfail) {
+				$result = self::deliverToInbox($post['command'], 0, $inbox, $post['uid'], $post['receivers'], $post['uri-id']);
+
+				if ($result['serverfailure']) {
+					// In a timeout situation we assume that every delivery to that inbox will time out.
+					// So we set the flag and try all deliveries at a later time.
+					Logger::info('Inbox delivery has a server failure', ['inbox' => $inbox]);
+					$serverfail = true;
+				}
+			}
+
+			if ($serverfail || !$result['success']) {
 				$uri_ids[] = $post['uri-id'];
 			}
 		}
 
-		Logger::debug('Inbox delivery done', ['inbox' => $inbox, 'posts' => count($posts), 'failed' => count($uri_ids)]);
+		Logger::debug('Inbox delivery done', ['inbox' => $inbox, 'posts' => count($posts), 'failed' => count($uri_ids), 'serverfailure' => $serverfail]);
 		return ['success' => empty($uri_ids), 'uri_ids' => $uri_ids];
 	}
 
-	private static function deliverToInbox(string $cmd, int $item_id, string $inbox, int $uid, array $receivers, int $uri_id)
+	private static function deliverToInbox(string $cmd, int $item_id, string $inbox, int $uid, array $receivers, int $uri_id): array
 	{
 		if (empty($item_id) && !empty($uri_id) && !empty($uid)) {
 			$item = Post::selectFirst(['id', 'parent', 'origin'], ['uri-id' => $uri_id, 'uid' => [$uid, 0]], ['order' => ['uid' => true]]);
 			if (empty($item['id'])) {
-				Logger::debug('Item not found, removing delivery', ['uri-id' => $uri_id, 'uid' => $uid, 'cmd' => $cmd, 'inbox' => $inbox]);
+				Logger::notice('Item not found, removing delivery', ['uri-id' => $uri_id, 'uid' => $uid, 'cmd' => $cmd, 'inbox' => $inbox]);
 				Post\Delivery::remove($uri_id, $inbox);
 				return true;
 			} else {
@@ -112,7 +126,8 @@ class APDelivery
 			}
 		}
 
-		$success = true;
+		$success    = true;
+		$serverfail = false;
 
 		if ($cmd == Delivery::MAIL) {
 			$data = ActivityPub\Transmitter::createActivityFromMail($item_id);
@@ -132,7 +147,27 @@ class APDelivery
 		} else {
 			$data = ActivityPub\Transmitter::createCachedActivityFromItem($item_id);
 			if (!empty($data)) {
-				$success = HTTPSignature::transmit($data, $inbox, $uid);
+				$timestamp  = microtime(true);
+				$response   = HTTPSignature::post($data, $inbox, $uid);
+				$runtime    = microtime(true) - $timestamp;
+				$success    = $response->isSuccess();
+				$serverfail = $response->isTimeout();
+				if (!$success) {
+					if (!$serverfail && ($response->getReturnCode() >= 500) && ($response->getReturnCode() <= 599)) {
+						$serverfail = true;
+					}
+
+					$xrd_timeout = DI::config()->get('system', 'xrd_timeout');
+					if (!$serverfail && $xrd_timeout && ($runtime > $xrd_timeout)) {
+						$serverfail = true;
+					}
+					$curl_timeout = DI::config()->get('system', 'curl_timeout');
+					if (!$serverfail && $curl_timeout && ($runtime > $curl_timeout)) {
+						$serverfail = true;
+					}
+
+					Logger::info('Delivery failed', ['retcode' => $response->getReturnCode(), 'serverfailure' => $serverfail, 'runtime' => round($runtime, 3), 'uri-id' => $uri_id, 'uid' => $uid, 'item_id' => $item_id, 'cmd' => $cmd, 'inbox' => $inbox]);
+				}
 				if ($uri_id) {
 					if ($success) {
 						Post\Delivery::remove($uri_id, $inbox);
@@ -145,13 +180,13 @@ class APDelivery
 
 		self::setSuccess($receivers, $success);
 
-		Logger::info('Delivered', ['uri-id' => $uri_id, 'uid' => $uid, 'item_id' => $item_id, 'cmd' => $cmd, 'inbox' => $inbox, 'success' => $success]);
+		Logger::debug('Delivered', ['uri-id' => $uri_id, 'uid' => $uid, 'item_id' => $item_id, 'cmd' => $cmd, 'inbox' => $inbox, 'success' => $success]);
 
 		if ($success && in_array($cmd, [Delivery::POST])) {
 			Post\DeliveryData::incrementQueueDone($uri_id, Post\DeliveryData::ACTIVITYPUB);
 		}
 
-		return $success;
+		return ['success' => $success, 'serverfailure' => $serverfail];
 	}
 
 	private static function setSuccess(array $receivers, bool $success)