keith/friendica
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge pull request #12583 from annando/delivery-queue

Browse source 
Bulk delivery added for all protocols
This commit is contained in:
Hypolite Petovan 2022-12-31 11:36:54 -05:00 committed by GitHub
commit 319857edaa
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
13 changed files with 450 additions and 61 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
database.sql
View file

@ -1,6 +1,6 @@
-- ------------------------------------------
-- Friendica 2023.03-dev (Giant Rhubarb)
-- DB_UPDATE_VERSION 1505
-- DB_UPDATE_VERSION 1506
-- ------------------------------------------
@ -579,6 +579,27 @@ CREATE TABLE IF NOT EXISTS `delayed-post` (
FOREIGN KEY (`wid`) REFERENCES `workerqueue` (`id`) ON UPDATE RESTRICT ON DELETE CASCADE
) DEFAULT COLLATE utf8mb4_general_ci COMMENT='Posts that are about to be distributed at a later time';
--
-- TABLE delivery-queue
--
CREATE TABLE IF NOT EXISTS `delivery-queue` (
`gsid` int unsigned NOT NULL COMMENT 'Target server',
`uri-id` int unsigned NOT NULL COMMENT 'Delivered post',
`created` datetime COMMENT '',
`command` varbinary(32) COMMENT '',
`cid` int unsigned COMMENT 'Target contact',
`uid` mediumint unsigned COMMENT 'Delivering user',
`failed` tinyint DEFAULT 0 COMMENT 'Number of times the delivery has failed',
PRIMARY KEY(`uri-id`,`gsid`),
INDEX `gsid_created` (`gsid`,`created`),
INDEX `uid` (`uid`),
INDEX `cid` (`cid`),
FOREIGN KEY (`gsid`) REFERENCES `gserver` (`id`) ON UPDATE RESTRICT ON DELETE RESTRICT,
FOREIGN KEY (`uri-id`) REFERENCES `item-uri` (`id`) ON UPDATE RESTRICT ON DELETE CASCADE,
FOREIGN KEY (`cid`) REFERENCES `contact` (`id`) ON UPDATE RESTRICT ON DELETE CASCADE,
FOREIGN KEY (`uid`) REFERENCES `user` (`uid`) ON UPDATE RESTRICT ON DELETE CASCADE
) DEFAULT COLLATE utf8mb4_general_ci COMMENT='Delivery data for posts for the batch processing';
--
-- TABLE diaspora-contact
--

1
doc/database.md
View file

@ -23,6 +23,7 @@ Database Tables
| [contact-relation](help/database/db_contact-relation) | Contact relations |
| [conv](help/database/db_conv) | private messages |
| [delayed-post](help/database/db_delayed-post) | Posts that are about to be distributed at a later time |
| [delivery-queue](help/database/db_delivery-queue) | Delivery data for posts for the batch processing |
| [diaspora-contact](help/database/db_diaspora-contact) | Diaspora compatible contacts - used in the Diaspora implementation |
| [diaspora-interaction](help/database/db_diaspora-interaction) | Signed Diaspora Interaction |
| [endpoint](help/database/db_endpoint) | ActivityPub endpoints - used in the ActivityPub implementation |

39
doc/database/db_delivery-queue.md Normal file
View file

@ -0,0 +1,39 @@
Table delivery-queue
===========
Delivery data for posts for the batch processing
Fields
------
| Field | Description | Type | Null | Key | Default | Extra |
| ------- | --------------------------------------- | ------------------ | ---- | --- | ------- | ----- |
| gsid | Target server | int unsigned | NO | PRI | NULL | |
| uri-id | Delivered post | int unsigned | NO | PRI | NULL | |
| created | | datetime | YES | | NULL | |
| command | | varbinary(32) | YES | | NULL | |
| cid | Target contact | int unsigned | YES | | NULL | |
| uid | Delivering user | mediumint unsigned | YES | | NULL | |
| failed | Number of times the delivery has failed | tinyint | YES | | 0 | |
Indexes
------------
| Name | Fields |
| ------------ | ------------- |
| PRIMARY | uri-id, gsid |
| gsid_created | gsid, created |
| uid | uid |
| cid | cid |
Foreign Keys
------------
| Field | Target Table | Target Field |
|-------|--------------|--------------|
| gsid | [gserver](help/database/db_gserver) | id |
| uri-id | [item-uri](help/database/db_item-uri) | id |
| cid | [contact](help/database/db_contact) | id |
| uid | [user](help/database/db_user) | uid |
Return to [database documentation](help/database)

9
src/Core/Worker.php
View file

@ -1264,7 +1264,7 @@ class Worker
$command = array_shift($args);
$parameters = json_encode($args);
$found = DBA::exists('workerqueue', ['command' => $command, 'parameter' => $parameters, 'done' => false]);
$queue = DBA::selectFirst('workerqueue', ['id', 'priority'], ['command' => $command, 'parameter' => $parameters, 'done' => false]);
$added = 0;
if (!is_int($priority) || !in_array($priority, self::PRIORITIES)) {
@ -1277,14 +1277,17 @@ class Worker
return 0;
}
if (!$found) {
if (empty($queue)) {
if (!DBA::insert('workerqueue', ['command' => $command, 'parameter' => $parameters, 'created' => $created,
'priority' => $priority, 'next_try' => $delayed])) {
return 0;
}
$added = DBA::lastInsertId();
} elseif ($force_priority) {
DBA::update('workerqueue', ['priority' => $priority], ['command' => $command, 'parameter' => $parameters, 'done' => false, 'pid' => 0]);
$ret = DBA::update('workerqueue', ['priority' => $priority], ['command' => $command, 'parameter' => $parameters, 'done' => false, 'pid' => 0]);
if ($ret && ($priority != $queue['priority'])) {
$added = $queue['id'];
}
}
// Set the IPC flag to ensure an immediate process execution via daemon

46
src/Core/Worker/Cron.php
View file

@ -26,8 +26,10 @@ use Friendica\Core\Worker;
use Friendica\Database\DBA;
use Friendica\DI;
use Friendica\Model\Contact;
use Friendica\Model\GServer;
use Friendica\Model\Post;
use Friendica\Protocol\ActivityPub;
use Friendica\Protocol\Delivery;
use Friendica\Util\DateTimeFormat;
use Friendica\Util\Strings;
@ -58,7 +60,10 @@ class Cron
// Remove old entries from the workerqueue
self::cleanWorkerQueue();
// Directly deliver or requeue posts
// Directly deliver or requeue posts to ActivityPub systems
self::deliverAPPosts();
// Directly deliver or requeue posts to other systems
self::deliverPosts();
}
@ -157,7 +162,7 @@ class Cron
*
* This function is placed here as a safeguard. Even when the worker queue is completely blocked, messages will be delivered.
*/
private static function deliverPosts()
private static function deliverAPPosts()
{
$deliveries = DBA::p("SELECT `item-uri`.`uri` AS `inbox`, MAX(`failed`) AS `failed` FROM `post-delivery` INNER JOIN `item-uri` ON `item-uri`.`id` = `post-delivery`.`inbox-id` GROUP BY `inbox` ORDER BY RAND()");
while ($delivery = DBA::fetch($deliveries)) {
@ -181,7 +186,7 @@ class Cron
}
if (Worker::add(['priority' => $priority, 'force_priority' => true], 'APDelivery', '', 0, $delivery['inbox'], 0)) {
Logger::info('Missing APDelivery worker added for inbox', ['inbox' => $delivery['inbox'], 'failed' => $delivery['failed'], 'priority' => $priority]);
Logger::info('Priority for APDelivery worker adjusted', ['inbox' => $delivery['inbox'], 'failed' => $delivery['failed'], 'priority' => $priority]);
}
}
@ -193,6 +198,41 @@ class Cron
}
}
/**
* Directly deliver messages or requeue them.
*/
private static function deliverPosts()
{
$deliveries = DBA::p("SELECT `gsid`, MAX(`failed`) AS `failed` FROM `delivery-queue` GROUP BY `gsid` ORDER BY RAND()");
while ($delivery = DBA::fetch($deliveries)) {
if ($delivery['failed'] > 0) {
Logger::info('Removing failed deliveries', ['gsid' => $delivery['gsid'], 'failed' => $delivery['failed']]);
Delivery::removeFailedQueue($delivery['gsid']);
}
if (($delivery['failed'] < 3) || GServer::isReachableById($delivery['gsid'])) {
$priority = Worker::PRIORITY_HIGH;
} elseif ($delivery['failed'] < 6) {
$priority = Worker::PRIORITY_MEDIUM;
} elseif ($delivery['failed'] < 8) {
$priority = Worker::PRIORITY_LOW;
} else {
$priority = Worker::PRIORITY_NEGLIGIBLE;
}
if (Worker::add(['priority' => $priority, 'force_priority' => true], 'BulkDelivery', $delivery['gsid'])) {
Logger::info('Priority for BulkDelivery worker adjusted', ['gsid' => $delivery['gsid'], 'failed' => $delivery['failed'], 'priority' => $priority]);
}
}
// Optimizing this table only last seconds
if (DI::config()->get('system', 'optimize_tables')) {
Logger::info('Optimize start');
DBA::e("OPTIMIZE TABLE `delivery-queue`");
Logger::info('Optimize end');
}
}
/**
* Add missing "intro" records.
*

114
src/Model/GServer.php
View file

@ -164,6 +164,58 @@ class GServer
return DI::dba()->toArray($stmt);
}
/**
* Checks if the given server array is unreachable for a long time now
*
* @param integer $gsid
* @return boolean
*/
private static function isDefunct(array $gserver): bool
{
return ($gserver['failed'] || in_array($gserver['network'], Protocol::FEDERATED)) &&
($gserver['last_contact'] >= $gserver['created']) &&
($gserver['last_contact'] < $gserver['last_failure']) &&
($gserver['last_contact'] < DateTimeFormat::utc('now - 90 days'));
}
/**
* Checks if the given server id is unreachable for a long time now
*
* @param integer $gsid
* @return boolean
*/
public static function isDefunctById(int $gsid): bool
{
$gserver = DBA::selectFirst('gserver', ['url', 'next_contact', 'last_contact', 'last_failure', 'created', 'failed', 'network'], ['id' => $gsid]);
if (empty($gserver)) {
return false;
} else {
if (strtotime($gserver['next_contact']) < time()) {
Worker::add(Worker::PRIORITY_LOW, 'UpdateGServer', $gserver['url'], false);
}
return self::isDefunct($gserver);
}
}
/**
* Checks if the given server id is reachable
*
* @param integer $gsid
* @return boolean
*/
public static function isReachableById(int $gsid): bool
{
$gserver = DBA::selectFirst('gserver', ['url', 'next_contact', 'failed', 'network'], ['id' => $gsid]);
if (empty($gserver)) {
return true;
} else {
if (strtotime($gserver['next_contact']) < time()) {
Worker::add(Worker::PRIORITY_LOW, 'UpdateGServer', $gserver['url'], false);
}
return !$gserver['failed'] && in_array($gserver['network'], Protocol::FEDERATED);
}
}
/**
* Checks if the given server is reachable
*
@ -305,12 +357,48 @@ class GServer
return self::detect($server_url, $network, $only_nodeinfo);
}
/**
* Reset failed server status by gserver id
*
* @param int $gsid
*/
public static function setReachableById(int $gsid)
{
$gserver = DBA::selectFirst('gserver', ['url', 'failed', 'next_contact'], ['id' => $gsid]);
if (DBA::isResult($gserver) && $gserver['failed']) {
self::update(['failed' => false, 'last_contact' => DateTimeFormat::utcNow()], ['id' => $gsid]);
Logger::info('Reset failed status for server', ['url' => $gserver['url']]);
if (strtotime($gserver['next_contact']) < time()) {
Worker::add(Worker::PRIORITY_LOW, 'UpdateGServer', $gserver['url'], false);
}
}
}
/**
* Set failed server status by gserver id
*
* @param int $gsid
*/
public static function setFailureById(int $gsid)
{
$gserver = DBA::selectFirst('gserver', ['url', 'failed', 'next_contact'], ['id' => $gsid]);
if (DBA::isResult($gserver) && !$gserver['failed']) {
self::update(['failed' => true, 'last_failure' => DateTimeFormat::utcNow()], ['id' => $gsid]);
Logger::info('Set failed status for server', ['url' => $gserver['url']]);
if (strtotime($gserver['next_contact']) < time()) {
Worker::add(Worker::PRIORITY_LOW, 'UpdateGServer', $gserver['url'], false);
}
}
}
/**
* Set failed server status
*
* @param string $url
*/
public static function setFailure(string $url)
public static function setFailureByUrl(string $url)
{
$gserver = DBA::selectFirst('gserver', [], ['nurl' => Strings::normaliseLink($url)]);
if (DBA::isResult($gserver)) {
@ -319,6 +407,9 @@ class GServer
'next_contact' => $next_update, 'network' => Protocol::PHANTOM, 'detection-method' => null],
['nurl' => Strings::normaliseLink($url)]);
Logger::info('Set failed status for existing server', ['url' => $url]);
if (self::isDefunct($gserver)) {
Contact::update(['archive' => true], ['gsid' => $gserver['id']]);
}
return;
}
self::insert(['url' => $url, 'nurl' => Strings::normaliseLink($url),
@ -371,7 +462,7 @@ class GServer
// If the URL missmatches, then we mark the old entry as failure
if (!Strings::compareLink($url, $original_url)) {
self::setFailure($original_url);
self::setFailureByUrl($original_url);
if (!self::getID($url, true)) {
self::detect($url, $network, $only_nodeinfo);
}
@ -380,7 +471,7 @@ class GServer
$valid_url = Network::isUrlValid($url);
if (!$valid_url) {
self::setFailure($url);
self::setFailureByUrl($url);
return false;
} else {
$valid_url = rtrim($valid_url, '/');
@ -392,7 +483,7 @@ class GServer
if (((parse_url($url, PHP_URL_HOST) != parse_url($valid_url, PHP_URL_HOST)) && (parse_url($url, PHP_URL_PATH) == parse_url($valid_url, PHP_URL_PATH))) ||
(((parse_url($url, PHP_URL_HOST) != parse_url($valid_url, PHP_URL_HOST)) || (parse_url($url, PHP_URL_PATH) != parse_url($valid_url, PHP_URL_PATH))) && empty(parse_url($valid_url, PHP_URL_PATH)))) {
Logger::debug('Found redirect. Mark old entry as failure', ['old' => $url, 'new' => $valid_url]);
self::setFailure($url);
self::setFailureByUrl($url);
if (!self::getID($valid_url, true)) {
self::detect($valid_url, $network, $only_nodeinfo);
}
@ -406,7 +497,7 @@ class GServer
unset($parts['path']);
$valid_url = (string)Uri::fromParts($parts);
self::setFailure($url);
self::setFailureByUrl($url);
if (!self::getID($valid_url, true)) {
self::detect($valid_url, $network, $only_nodeinfo);
}
@ -426,7 +517,7 @@ class GServer
// When a nodeinfo is present, we don't need to dig further
$curlResult = DI::httpClient()->get($url . '/.well-known/x-nodeinfo2', HttpClientAccept::JSON);
if ($curlResult->isTimeout()) {
self::setFailure($url);
self::setFailureByUrl($url);
return false;
}
@ -438,7 +529,7 @@ class GServer
if ($only_nodeinfo && empty($serverdata)) {
Logger::info('Invalid nodeinfo in nodeinfo-mode, server is marked as failure', ['url' => $url]);
self::setFailure($url);
self::setFailureByUrl($url);
return false;
} elseif (empty($serverdata)) {
$serverdata = ['detection-method' => self::DETECT_MANUAL, 'network' => Protocol::PHANTOM, 'platform' => '', 'version' => '', 'site_name' => '', 'info' => ''];
@ -477,7 +568,7 @@ class GServer
}
if (!$curlResult->isSuccess() || empty($curlResult->getBody())) {
self::setFailure($url);
self::setFailureByUrl($url);
return false;
}
@ -546,7 +637,7 @@ class GServer
// Most servers aren't installed in a subdirectory, so we declare this entry as failed
if (($serverdata['network'] == Protocol::PHANTOM) && !empty(parse_url($url, PHP_URL_PATH)) && in_array($serverdata['detection-method'], [self::DETECT_MANUAL])) {
self::setFailure($url);
self::setFailureByUrl($url);
return false;
}
@ -562,6 +653,11 @@ class GServer
$serverdata = self::detectNetworkViaContacts($url, $serverdata);
}
if (($serverdata['network'] == Protocol::PHANTOM) && in_array($serverdata['detection-method'], [self::DETECT_MANUAL, self::DETECT_BODY])) {
self::setFailureByUrl($url);
return false;
}
// Detect the directory type
$serverdata['directory-type'] = self::DT_NONE;

8
src/Protocol/DFRN.php
View file

@ -1014,6 +1014,10 @@ class DFRN
$xml = $postResult->getBody();
$curl_stat = $postResult->getReturnCode();
if (!empty($contact['gsid']) && ($postResult->isTimeout() || empty($curl_stat))) {
GServer::setFailureById($contact['gsid']);
}
if (empty($curl_stat) || empty($xml)) {
Logger::notice('Empty answer from ' . $contact['id'] . ' - ' . $dest_url);
return -9; // timed out
@ -1035,6 +1039,10 @@ class DFRN
return -23;
}
if (!empty($contact['gsid'])) {
GServer::setReachableById($contact['gsid']);
}
if (!empty($res->message)) {
Logger::info('Transmit to ' . $dest_url . ' returned status '.$res->status.' - '.$res->message);
}

156
src/Protocol/Delivery.php
View file

@ -26,6 +26,7 @@ use Friendica\Contact\FriendSuggest\Exception\FriendSuggestNotFoundException;
use Friendica\Core\Logger;
use Friendica\Core\Protocol;
use Friendica\Core\Worker;
use Friendica\Database\Database;
use Friendica\Database\DBA;
use Friendica\DI;
use Friendica\Model\Contact;
@ -45,7 +46,16 @@ class Delivery
const REMOVAL = 'removeme';
const PROFILEUPDATE = 'profileupdate';
public static function deliver(string $cmd, int $post_uriid, int $contact_id, int $sender_uid = 0)
/**
* Deliver posts to other systems
*
* @param string $cmd
* @param integer $post_uriid
* @param integer $contact_id
* @param integer $sender_uid
* @return bool "false" on remote system error. "true" when delivery was successful or we shouldn't retry.
*/
public static function deliver(string $cmd, int $post_uriid, int $contact_id, int $sender_uid = 0): bool
{
Logger::info('Invoked', ['cmd' => $cmd, 'target' => $post_uriid, 'sender_uid' => $sender_uid, 'contact' => $contact_id]);
@ -57,7 +67,7 @@ class Delivery
if ($cmd == self::MAIL) {
$target_item = DBA::selectFirst('mail', [], ['id' => $post_uriid]);
if (!DBA::isResult($target_item)) {
return;
return true;
}
$uid = $target_item['uid'];
} elseif ($cmd == self::SUGGESTION) {
@ -65,7 +75,7 @@ class Delivery
$target_item = DI::fsuggest()->selectOneById($post_uriid)->toArray();
} catch (FriendSuggestNotFoundException $e) {
DI::logger()->info('Cannot find FriendSuggestion', ['id' => $post_uriid]);
return;
return true;
}
$uid = $target_item['uid'];
} elseif ($cmd == self::RELOCATION) {
@ -75,7 +85,7 @@ class Delivery
$item = Post::selectFirst(['id', 'parent'], ['uri-id' => $post_uriid, 'uid' => $sender_uid]);
if (!DBA::isResult($item) || empty($item['parent'])) {
Logger::warning('Post not found', ['uri-id' => $post_uriid, 'uid' => $sender_uid]);
return;
return true;
}
$target_id = intval($item['id']);
$parent_id = intval($item['parent']);
@ -101,13 +111,13 @@ class Delivery
if (empty($target_item)) {
Logger::warning("No target item data. Quitting here.", ['id' => $target_id]);
return;
return true;
}
if (empty($parent)) {
Logger::warning('Parent ' . $parent_id . ' for item ' . $target_id . "wasn't found. Quitting here.");
self::setFailedQueue($cmd, $target_item);
return;
return true;
}
if (!empty($target_item['contact-uid'])) {
@ -117,7 +127,7 @@ class Delivery
} else {
Logger::info('Only public users for item ' . $target_id);
self::setFailedQueue($cmd, $target_item);
return;
return true;
}
$condition = ['uri' => $target_item['thr-parent'], 'uid' => $target_item['uid']];
@ -131,7 +141,7 @@ class Delivery
if (!empty($contact_id) && Contact::isArchived($contact_id)) {
Logger::info('Contact is archived', ['id' => $contact_id, 'cmd' => $cmd, 'item' => $target_item['id']]);
self::setFailedQueue($cmd, $target_item);
return;
return true;
}
// avoid race condition with deleting entries
@ -185,7 +195,7 @@ class Delivery
$owner = User::getOwnerDataById($uid);
if (!DBA::isResult($owner)) {
self::setFailedQueue($cmd, $target_item);
return;
return true;
}
// We don't deliver our items to blocked, archived or pending contacts, and not to ourselves either
@ -194,12 +204,12 @@ class Delivery
);
if (!DBA::isResult($contact)) {
self::setFailedQueue($cmd, $target_item);
return;
return true;
}
if (Network::isUrlBlocked($contact['url'])) {
self::setFailedQueue($cmd, $target_item);
return;
return true;
}
$protocol = GServer::getProtocol($contact['gsid'] ?? 0);
@ -217,22 +227,23 @@ class Delivery
switch ($contact['network']) {
case Protocol::DFRN:
self::deliverDFRN($cmd, $contact, $owner, $items, $target_item, $public_message, $top_level, $followup, $protocol);
$success = self::deliverDFRN($cmd, $contact, $owner, $items, $target_item, $public_message, $top_level, $followup, $protocol);
break;
case Protocol::DIASPORA:
self::deliverDiaspora($cmd, $contact, $owner, $items, $target_item, $public_message, $top_level, $followup);
$success = self::deliverDiaspora($cmd, $contact, $owner, $items, $target_item, $public_message, $top_level, $followup);
break;
case Protocol::MAIL:
self::deliverMail($cmd, $contact, $owner, $target_item, $thr_parent);
$success = self::deliverMail($cmd, $contact, $owner, $target_item, $thr_parent);
break;
default:
$success = true;
break;
}
return;
return $success;
}
/**
@ -265,19 +276,18 @@ class Delivery
* @param boolean $followup Is it an answer to a remote post?
* @param int|null $server_protocol The protocol of the server
*
* @return void
* @return bool "false" on remote system error. "true" when delivery was successful or we shouldn't retry.
* @throws \Friendica\Network\HTTPException\InternalServerErrorException
* @throws \ImagickException
*/
private static function deliverDFRN(string $cmd, array $contact, array $owner, array $items, array $target_item, bool $public_message, bool $top_level, bool $followup, int $server_protocol = null)
private static function deliverDFRN(string $cmd, array $contact, array $owner, array $items, array $target_item, bool $public_message, bool $top_level, bool $followup, int $server_protocol = null): bool
{
$target_item_id = $target_item['guid'] ?? '' ?: $target_item['id'] ?? null;
// Transmit Diaspora reshares via Diaspora if the Friendica contact support Diaspora
if (Diaspora::getReshareDetails($target_item) && Diaspora::isSupportedByContactUrl($contact['addr'])) {
Logger::info('Reshare will be transmitted via Diaspora', ['url' => $contact['url'], 'guid' => $target_item_id]);
self::deliverDiaspora($cmd, $contact, $owner, $items, $target_item, $public_message, $top_level, $followup);
return;
return self::deliverDiaspora($cmd, $contact, $owner, $items, $target_item, $public_message, $top_level, $followup);
}
Logger::info('Deliver ' . ($target_item_id ?? 'relocation') . ' via DFRN to ' . ($contact['addr'] ?? '' ?: $contact['url']));
@ -333,17 +343,18 @@ class Delivery
Post\DeliveryData::incrementQueueDone($target_item['uri-id'], $protocol);
GServer::setProtocol($contact['gsid'] ?? 0, $protocol);
$success = true;
} else {
Post\DeliveryData::incrementQueueFailed($target_item['uri-id']);
$success = false;
}
}
return;
return $success;
}
if ((($deliver_status < 200) || ($deliver_status > 299)) && (empty($server_protocol) || ($server_protocol == Post\DeliveryData::LEGACY_DFRN))) {
// Transmit via Diaspora if not possible via Friendica
self::deliverDiaspora($cmd, $contact, $owner, $items, $target_item, $public_message, $top_level, $followup);
return;
return self::deliverDiaspora($cmd, $contact, $owner, $items, $target_item, $public_message, $top_level, $followup);
}
} else {
// DFRN payload over Diaspora transport layer
@ -361,6 +372,7 @@ class Delivery
if ($cmd == Delivery::POST) {
Post\DeliveryData::incrementQueueDone($target_item['uri-id'], $protocol);
}
$success = true;
} else {
// The message could not be delivered. We mark the contact as "dead"
Contact::markForArchival($contact);
@ -369,7 +381,9 @@ class Delivery
if (!Worker::defer() && $cmd == Delivery::POST) {
Post\DeliveryData::incrementQueueFailed($target_item['uri-id']);
}
$success = false;
}
return $success;
}
/**
@ -384,11 +398,11 @@ class Delivery
* @param boolean $top_level Is it a thread starter?
* @param boolean $followup Is it an answer to a remote post?
*
* @return void
* @return bool "false" on remote system error. "true" when delivery was successful or we shouldn't retry.
* @throws \Friendica\Network\HTTPException\InternalServerErrorException
* @throws \ImagickException
*/
private static function deliverDiaspora(string $cmd, array $contact, array $owner, array $items, array $target_item, bool $public_message, bool $top_level, bool $followup)
private static function deliverDiaspora(string $cmd, array $contact, array $owner, array $items, array $target_item, bool $public_message, bool $top_level, bool $followup): bool
{
// We don't treat Forum posts as "wall-to-wall" to be able to post them via Diaspora
$walltowall = $top_level && ($owner['id'] != $items[0]['contact-id']) & ($owner['account-type'] != User::ACCOUNT_TYPE_COMMUNITY);
@ -402,20 +416,20 @@ class Delivery
Logger::notice('Deliver via Diaspora', ['target' => $target_item['id'], 'guid' => $target_item['guid'], 'to' => $loc]);
if (!DI::config()->get('system', 'diaspora_enabled')) {
return;
return true;
}
if ($cmd == self::MAIL) {
Diaspora::sendMail($target_item, $owner, $contact);
return;
$deliver_status = Diaspora::sendMail($target_item, $owner, $contact);
return ($deliver_status >= 200) && ($deliver_status <= 299);
}
if ($cmd == self::SUGGESTION) {
return;
return true;
}
if (!$contact['pubkey'] && !$public_message) {
return;
return true;
}
if ($cmd == self::RELOCATION) {
@ -438,7 +452,7 @@ class Delivery
$deliver_status = Diaspora::sendStatus($target_item, $owner, $contact, $public_message);
} else {
Logger::warning('Unknown mode', ['command' => $cmd, 'target' => $loc]);
return;
return true;
}
if (($deliver_status >= 200) && ($deliver_status <= 299)) {
@ -450,6 +464,7 @@ class Delivery
if ($cmd == Delivery::POST) {
Post\DeliveryData::incrementQueueDone($target_item['uri-id'], Post\DeliveryData::DIASPORA);
}
$success = true;
} else {
// The message could not be delivered. We mark the contact as "dead"
Contact::markForArchival($contact);
@ -468,7 +483,9 @@ class Delivery
} elseif ($cmd == Delivery::POST) {
Post\DeliveryData::incrementQueueFailed($target_item['uri-id']);
}
$success = false;
}
return $success;
}
/**
@ -480,27 +497,27 @@ class Delivery
* @param array $target_item Item record of the content
* @param array $thr_parent Item record of the direct parent in the thread
*
* @return void
* @return bool "false" on remote system error. "true" when delivery was successful or we shouldn't retry.
* @throws \Friendica\Network\HTTPException\InternalServerErrorException
* @throws \ImagickException
*/
private static function deliverMail(string $cmd, array $contact, array $owner, array $target_item, array $thr_parent)
private static function deliverMail(string $cmd, array $contact, array $owner, array $target_item, array $thr_parent): bool
{
if (DI::config()->get('system', 'imap_disabled')) {
return;
return true;
}
$addr = $contact['addr'];
if (!strlen($addr)) {
return;
return true;
}
if ($cmd != self::POST) {
return;
return true;
}
if ($target_item['verb'] != Activity::POST) {
return;
return true;
}
if (!empty($thr_parent['object'])) {
@ -516,7 +533,7 @@ class Delivery
$local_user = DBA::selectFirst('user', [], ['uid' => $owner['uid']]);
if (!DBA::isResult($local_user)) {
return;
return true;
}
Logger::info('About to deliver via mail', ['guid' => $target_item['guid'], 'to' => $addr]);
@ -587,5 +604,70 @@ class Delivery
// Failed
Logger::warning('Delivery of mail has FAILED', ['to' => $addr, 'subject' => $subject, 'guid' => $target_item['guid']]);
}
return $success;
}
/**
* Add post for a server
*
* @param string $cmd
* @param integer $uri_id
* @param string $created
* @param integer $cid
* @param integer $gsid
* @param integer $uid
* @return bool
*/
public static function addQueue(string $cmd, int $uri_id, string $created, int $cid, int $gsid, int $uid): bool
{
$fields = ['uri-id' => $uri_id, 'uid' => $uid, 'cid' => $cid, 'gsid' => $gsid, 'created' => $created, 'command' => $cmd];
return DBA::insert('delivery-queue', $fields, Database::INSERT_IGNORE);
}
/**
* Remove post by a server after delivery
*
* @param integer $uri_id
* @param integer $gsid
* @return bool
*/
public static function removeQueue(int $uri_id, int $gsid): bool
{
return DBA::delete('delivery-queue', ['uri-id' => $uri_id, 'gsid' => $gsid]);
}
/**
* Remove failed posts for the given server
*
* @param integer $gsid
* @return bool
*/
public static function removeFailedQueue(int $gsid): bool
{
return DBA::delete('delivery-queue', ["`gsid` = ? AND `failed` >= ?", $gsid, DI::config()->get('system', 'worker_defer_limit')]);
}
/**
* Increment "failed" counter for the given server and post
*
* @param integer $uri_id
* @param integer $gsid
* @return bool
*/
public static function incrementFailedQueue(int $uri_id, int $gsid): bool
{
return DBA::e('UPDATE `delivery-queue` SET `failed` = `failed` + 1 WHERE `uri-id` = ? AND `gsid` = ?', $uri_id, $gsid);
}
/**
* Select queue entries for the given server
*
* @param integer $gsid
* @return array
*/
public static function selectQueueForServer(int $gsid): array
{
return DBA::selectToArray('delivery-queue', [], ["`gsid` = ? AND `failed` < ?", $gsid, DI::config()->get('system', 'worker_defer_limit')], ['order' => ['created']]);
}
}

6
src/Protocol/Diaspora.php
View file

@ -2955,6 +2955,12 @@ class Diaspora
return 200;
}
if (!empty($contact['gsid']) && (empty($return_code) || $postResult->isTimeout())) {
GServer::setFailureById($contact['gsid']);
} elseif (!empty($contact['gsid']) && ($return_code >= 200) && ($return_code <= 299)) {
GServer::setReachableById($contact['gsid']);
}
Logger::notice('transmit: ' . $logid . '-' . $guid . ' to ' . $dest_url . ' returns: ' . $return_code);
return $return_code ? $return_code : -1;

60
src/Worker/BulkDelivery.php Normal file
View file

@ -0,0 +1,60 @@
<?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\Worker;
use Friendica\Core\Logger;
use Friendica\Core\Worker;
use Friendica\Model\GServer;
use Friendica\Protocol\Delivery as ProtocolDelivery;
class BulkDelivery
{
public static function execute(int $gsid)
{
$server_failure = false;
$delivery_failure = false;
$posts = ProtocolDelivery::selectQueueForServer($gsid);
foreach ($posts as $post) {
if (!$server_failure && ProtocolDelivery::deliver($post['command'], $post['uri-id'], $post['cid'], $post['uid'])) {
ProtocolDelivery::removeQueue($post['uri-id'], $post['gsid']);
Logger::debug('Delivery successful', $post);
} else {
ProtocolDelivery::incrementFailedQueue($post['uri-id'], $post['gsid']);
$delivery_failure = true;
if (!$server_failure) {
$server_failure = !GServer::isReachableById($gsid);
}
Logger::debug('Delivery failed', ['server_failure' => $server_failure, 'post' => $post]);
}
}
if ($server_failure) {
Worker::defer();
}
if ($delivery_failure) {
ProtocolDelivery::removeFailedQueue($gsid);
}
}
}

21
src/Worker/Notifier.php
View file

@ -565,7 +565,15 @@ class Notifier
continue;
}
if (!GServer::reachable($contact)) {
if (empty($contact['gsid'])) {
$reachable = !GServer::reachable($contact);
} elseif (!DI::config()->get('system', 'bulk_delivery')) {
$reachable = !GServer::isReachableById($contact['gsid']);
} else {
$reachable = !GServer::isDefunctById($contact['gsid']);
}
if (!$reachable) {
Logger::info('Server is not reachable', ['id' => $post_uriid, 'uid' => $sender_uid, 'contact' => $contact]);
continue;
}
@ -582,9 +590,16 @@ class Notifier
$deliver_options = ['priority' => $a->getQueueValue('priority'), 'created' => $a->getQueueValue('created'), 'dont_fork' => true];
}
if (!empty($contact['gsid']) && DI::config()->get('system', 'bulk_delivery')) {
$delivery_queue_count++;
Delivery::addQueue($cmd, $post_uriid, $target_item['created'], $contact['id'], $contact['gsid'], $sender_uid);
Worker::add(['priority' => Worker::PRIORITY_HIGH, 'dont_fork' => true], 'BulkDelivery', $contact['gsid']);
} else {
if (Worker::add($deliver_options, 'Delivery', $cmd, $post_uriid, (int)$contact['id'], $sender_uid)) {
$delivery_queue_count++;
}
}
Worker::coolDown();
}
return $delivery_queue_count;
@ -834,7 +849,7 @@ class Notifier
if (DI::config()->get('system', 'bulk_delivery')) {
$delivery_queue_count++;
Post\Delivery::add($target_item['uri-id'], $uid, $inbox, $target_item['created'], $cmd, $receivers);
Worker::add(Worker::PRIORITY_HIGH, 'APDelivery', '', 0, $inbox, 0);
Worker::add([Worker::PRIORITY_HIGH, 'dont_fork' => true], 'APDelivery', '', 0, $inbox, 0);
} else {
if (Worker::add(['priority' => $priority, 'created' => $created, 'dont_fork' => true],
'APDelivery', $cmd, $target_item['id'], $inbox, $uid, $receivers, $target_item['uri-id'])) {
@ -851,7 +866,7 @@ class Notifier
if (DI::config()->get('system', 'bulk_delivery')) {
$delivery_queue_count++;
Post\Delivery::add($target_item['uri-id'], $uid, $inbox, $target_item['created'], $cmd, []);
Worker::add(Worker::PRIORITY_MEDIUM, 'APDelivery', '', 0, $inbox, 0);
Worker::add([Worker::PRIORITY_MEDIUM, 'dont_fork' => true], 'APDelivery', '', 0, $inbox, 0);
} else {
if (Worker::add(['priority' => $priority, 'dont_fork' => true], 'APDelivery', $cmd, $target_item['id'], $inbox, $uid, [], $target_item['uri-id'])) {
$delivery_queue_count++;

6
src/Worker/UpdateGServer.php
View file

@ -43,18 +43,18 @@ class UpdateGServer
$filtered = filter_var($server_url, FILTER_SANITIZE_URL);
if (substr(Strings::normaliseLink($filtered), 0, 7) != 'http://') {
GServer::setFailure($server_url);
GServer::setFailureByUrl($server_url);
return;
}
if (($filtered != $server_url) && DBA::exists('gserver', ['nurl' => Strings::normaliseLink($server_url)])) {
GServer::setFailure($server_url);
GServer::setFailureByUrl($server_url);
return;
}
$cleaned = GServer::cleanURL($server_url);
if (($cleaned != $server_url) && DBA::exists('gserver', ['nurl' => Strings::normaliseLink($server_url)])) {
GServer::setFailure($server_url);
GServer::setFailureByUrl($server_url);
return;
}

20
static/dbstructure.config.php
View file

@ -55,7 +55,7 @@
use Friendica\Database\DBA;
if (!defined('DB_UPDATE_VERSION')) {
define('DB_UPDATE_VERSION', 1505);
define('DB_UPDATE_VERSION', 1506);
}
return [
@ -638,6 +638,24 @@ return [
"wid" => ["wid"],
]
],
"delivery-queue" => [
"comment" => "Delivery data for posts for the batch processing",
"fields" => [
"gsid" => ["type" => "int unsigned", "not null" => "1", "primary" => "1", "foreign" => ["gserver" => "id", "on delete" => "restrict"], "comment" => "Target server"],
"uri-id" => ["type" => "int unsigned", "not null" => "1", "primary" => "1", "foreign" => ["item-uri" => "id"], "comment" => "Delivered post"],
"created" => ["type" => "datetime", "comment" => ""],
"command" => ["type" => "varbinary(32)", "comment" => ""],
"cid" => ["type" => "int unsigned", "foreign" => ["contact" => "id"], "comment" => "Target contact"],
"uid" => ["type" => "mediumint unsigned", "foreign" => ["user" => "uid"], "comment" => "Delivering user"],
"failed" => ["type" => "tinyint", "default" => 0, "comment" => "Number of times the delivery has failed"],
],
"indexes" => [
"PRIMARY" => ["uri-id", "gsid"],
"gsid_created" => ["gsid", "created"],
"uid" => ["uid"],
"cid" => ["cid"],
]
],
"diaspora-contact" => [
"comment" => "Diaspora compatible contacts - used in the Diaspora implementation",
"fields" => [