strk/friendica
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Merge pull request #9620 from annando/worker-command

Browse source 
The worker now has got a "command" field
This commit is contained in:
Hypolite Petovan 2020-12-04 08:36:35 -05:00 committed by GitHub
parent 96a927f6e1 8e9a9b2049
commit 741bc98174
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 245 additions and 155 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
database.sql
View file

@ -1,6 +1,6 @@
-- ------------------------------------------ -- ------------------------------------------
-- Friendica 2020.12-dev (Red Hot Poker) -- Friendica 2020.12-dev (Red Hot Poker)
-- DB_UPDATE_VERSION 1382 -- DB_UPDATE_VERSION 1383
-- ------------------------------------------ -- ------------------------------------------
@ -26,11 +26,13 @@ CREATE TABLE IF NOT EXISTS `gserver` (
`detection-method` tinyint unsigned COMMENT 'Method that had been used to detect that server', `detection-method` tinyint unsigned COMMENT 'Method that had been used to detect that server',
`created` datetime NOT NULL DEFAULT '0001-01-01 00:00:00' COMMENT '', `created` datetime NOT NULL DEFAULT '0001-01-01 00:00:00' COMMENT '',
`last_poco_query` datetime DEFAULT '0001-01-01 00:00:00' COMMENT '', `last_poco_query` datetime DEFAULT '0001-01-01 00:00:00' COMMENT '',
`last_contact` datetime DEFAULT '0001-01-01 00:00:00' COMMENT '', `last_contact` datetime DEFAULT '0001-01-01 00:00:00' COMMENT 'Last successful connection request',
`last_failure` datetime DEFAULT '0001-01-01 00:00:00' COMMENT '', `last_failure` datetime DEFAULT '0001-01-01 00:00:00' COMMENT 'Last failed connection request',
`failed` boolean COMMENT 'Connection failed', `failed` boolean COMMENT 'Connection failed',
`next_contact` datetime DEFAULT '0001-01-01 00:00:00' COMMENT 'Next connection request',
PRIMARY KEY(`id`), PRIMARY KEY(`id`),
UNIQUE INDEX `nurl` (`nurl`(190)) UNIQUE INDEX `nurl` (`nurl`(190)),
INDEX `next_contact` (`next_contact`)
) DEFAULT COLLATE utf8mb4_general_ci COMMENT='Global servers'; ) DEFAULT COLLATE utf8mb4_general_ci COMMENT='Global servers';
-- --
@ -1464,7 +1466,8 @@ CREATE TABLE IF NOT EXISTS `worker-ipc` (
-- --
CREATE TABLE IF NOT EXISTS `workerqueue` ( CREATE TABLE IF NOT EXISTS `workerqueue` (
`id` int unsigned NOT NULL auto_increment COMMENT 'Auto incremented worker task id', `id` int unsigned NOT NULL auto_increment COMMENT 'Auto incremented worker task id',
`parameter` mediumtext COMMENT 'Task command', `command` varchar(100) COMMENT 'Task command',
`parameter` mediumtext COMMENT 'Task parameter',
`priority` tinyint unsigned NOT NULL DEFAULT 0 COMMENT 'Task priority', `priority` tinyint unsigned NOT NULL DEFAULT 0 COMMENT 'Task priority',
`created` datetime NOT NULL DEFAULT '0001-01-01 00:00:00' COMMENT 'Creation date', `created` datetime NOT NULL DEFAULT '0001-01-01 00:00:00' COMMENT 'Creation date',
`pid` int unsigned NOT NULL DEFAULT 0 COMMENT 'Process id of the worker', `pid` int unsigned NOT NULL DEFAULT 0 COMMENT 'Process id of the worker',
@ -1473,7 +1476,8 @@ CREATE TABLE IF NOT EXISTS `workerqueue` (
`retrial` tinyint NOT NULL DEFAULT 0 COMMENT 'Retrial counter', `retrial` tinyint NOT NULL DEFAULT 0 COMMENT 'Retrial counter',
`done` boolean NOT NULL DEFAULT '0' COMMENT 'Marked 1 when the task was done - will be deleted later', `done` boolean NOT NULL DEFAULT '0' COMMENT 'Marked 1 when the task was done - will be deleted later',
PRIMARY KEY(`id`), PRIMARY KEY(`id`),
INDEX `done_parameter` (`done`,`parameter`(64)), INDEX `command` (`command`),
INDEX `done_command_parameter` (`done`,`command`,`parameter`(64)),
INDEX `done_executed` (`done`,`executed`), INDEX `done_executed` (`done`,`executed`),
INDEX `done_priority_retrial_created` (`done`,`priority`,`retrial`,`created`), INDEX `done_priority_retrial_created` (`done`,`priority`,`retrial`,`created`),
INDEX `done_priority_next_try` (`done`,`priority`,`next_try`), INDEX `done_priority_next_try` (`done`,`priority`,`next_try`),

51
src/Core/Worker.php
View file

@ -300,7 +300,11 @@ class Worker
return false; return false;
} }
$argv = json_decode($queue["parameter"], true); $argv = json_decode($queue['parameter'], true);
if (!empty($queue['command'])) {
array_unshift($argv, $queue['command']);
}
if (empty($argv)) { if (empty($argv)) {
Logger::warning('Parameter is empty', ['queue' => $queue]); Logger::warning('Parameter is empty', ['queue' => $queue]);
return false; return false;
@ -576,7 +580,7 @@ class Worker
$stamp = (float)microtime(true); $stamp = (float)microtime(true);
$entries = DBA::select( $entries = DBA::select(
'workerqueue', 'workerqueue',
['id', 'pid', 'executed', 'priority', 'parameter'], ['id', 'pid', 'executed', 'priority', 'command', 'parameter'],
['NOT `done` AND `pid` != 0'], ['NOT `done` AND `pid` != 0'],
['order' => ['priority', 'retrial', 'created']] ['order' => ['priority', 'retrial', 'created']]
); );
@ -603,17 +607,21 @@ class Worker
$max_duration_defaults = [PRIORITY_CRITICAL => 720, PRIORITY_HIGH => 10, PRIORITY_MEDIUM => 60, PRIORITY_LOW => 180, PRIORITY_NEGLIGIBLE => 720]; $max_duration_defaults = [PRIORITY_CRITICAL => 720, PRIORITY_HIGH => 10, PRIORITY_MEDIUM => 60, PRIORITY_LOW => 180, PRIORITY_NEGLIGIBLE => 720];
$max_duration = $max_duration_defaults[$entry["priority"]]; $max_duration = $max_duration_defaults[$entry["priority"]];
$argv = json_decode($entry["parameter"], true); $argv = json_decode($entry['parameter'], true);
if (empty($argv)) { if (!empty($entry['command'])) {
$command = $entry['command'];
} elseif (!empty($argv)) {
$command = array_shift($argv);
} else {
return; return;
} }
$argv[0] = basename($argv[0]); $command = basename($command);
// How long is the process already running? // How long is the process already running?
$duration = (time() - strtotime($entry["executed"])) / 60; $duration = (time() - strtotime($entry["executed"])) / 60;
if ($duration > $max_duration) { if ($duration > $max_duration) {
Logger::notice("Worker process ".$entry["pid"]." (".substr(json_encode($argv), 0, 50).") took more than ".$max_duration." minutes. It will be killed now."); Logger::notice('Worker process took too much time - killed', ['duration' => number_format($duration, 3), 'max' => $max_duration, 'id' => $entry["id"], 'pid' => $entry["pid"], 'command' => $command]);
posix_kill($entry["pid"], SIGTERM); posix_kill($entry["pid"], SIGTERM);
// We killed the stale process. // We killed the stale process.
@ -636,7 +644,7 @@ class Worker
self::$db_duration += (microtime(true) - $stamp); self::$db_duration += (microtime(true) - $stamp);
self::$db_duration_write += (microtime(true) - $stamp); self::$db_duration_write += (microtime(true) - $stamp);
} else { } else {
Logger::info('Process runtime is okay', ['pid' => $entry["pid"], 'duration' => $duration, 'max' => $max_duration, 'command' => substr(json_encode($argv), 0, 50)]); Logger::info('Process runtime is okay', ['duration' => number_format($duration, 3), 'max' => $max_duration, 'id' => $entry["id"], 'pid' => $entry["pid"], 'command' => $command]);
} }
} }
} }
@ -848,12 +856,17 @@ class Worker
$ids = []; $ids = [];
$stamp = (float)microtime(true); $stamp = (float)microtime(true);
$condition = ["`priority` = ? AND `pid` = 0 AND NOT `done` AND `next_try` < ?", $priority, DateTimeFormat::utcNow()]; $condition = ["`priority` = ? AND `pid` = 0 AND NOT `done` AND `next_try` < ?", $priority, DateTimeFormat::utcNow()];
$tasks = DBA::select('workerqueue', ['id', 'parameter'], $condition, ['limit' => $limit, 'order' => ['retrial', 'created']]); $tasks = DBA::select('workerqueue', ['id', 'command', 'parameter'], $condition, ['limit' => $limit, 'order' => ['retrial', 'created']]);
self::$db_duration += (microtime(true) - $stamp); self::$db_duration += (microtime(true) - $stamp);
while ($task = DBA::fetch($tasks)) { while ($task = DBA::fetch($tasks)) {
$ids[] = $task['id']; $ids[] = $task['id'];
// Only continue that loop while we are storing commands that can be processed quickly // Only continue that loop while we are storing commands that can be processed quickly
$command = json_decode($task['parameter'])[0]; if (!empty($task['command'])) {
$command = $task['command'];
} else {
$command = json_decode($task['parameter'])[0];
}
if (!in_array($command, self::FAST_COMMANDS)) { if (!in_array($command, self::FAST_COMMANDS)) {
break; break;
} }
@ -968,13 +981,17 @@ class Worker
if ($limit > 0) { if ($limit > 0) {
$stamp = (float)microtime(true); $stamp = (float)microtime(true);
$condition = ["`pid` = 0 AND NOT `done` AND `next_try` < ?", DateTimeFormat::utcNow()]; $condition = ["`pid` = 0 AND NOT `done` AND `next_try` < ?", DateTimeFormat::utcNow()];
$tasks = DBA::select('workerqueue', ['id', 'parameter'], $condition, ['limit' => $limit, 'order' => ['priority', 'retrial', 'created']]); $tasks = DBA::select('workerqueue', ['id', 'command', 'parameter'], $condition, ['limit' => $limit, 'order' => ['priority', 'retrial', 'created']]);
self::$db_duration += (microtime(true) - $stamp); self::$db_duration += (microtime(true) - $stamp);
while ($task = DBA::fetch($tasks)) { while ($task = DBA::fetch($tasks)) {
$ids[] = $task['id']; $ids[] = $task['id'];
// Only continue that loop while we are storing commands that can be processed quickly // Only continue that loop while we are storing commands that can be processed quickly
$command = json_decode($task['parameter'])[0]; if (!empty($task['command'])) {
$command = $task['command'];
} else {
$command = json_decode($task['parameter'])[0];
}
if (!in_array($command, self::FAST_COMMANDS)) { if (!in_array($command, self::FAST_COMMANDS)) {
break; break;
} }
@ -1242,8 +1259,9 @@ class Worker
} }
} }
$command = array_shift($args);
$parameters = json_encode($args); $parameters = json_encode($args);
$found = DBA::exists('workerqueue', ['parameter' => $parameters, 'done' => false]); $found = DBA::exists('workerqueue', ['command' => $command, 'parameter' => $parameters, 'done' => false]);
$added = false; $added = false;
// Quit if there was a database error - a precaution for the update process to 3.5.3 // Quit if there was a database error - a precaution for the update process to 3.5.3
@ -1252,13 +1270,13 @@ class Worker
} }
if (!$found) { if (!$found) {
$added = DBA::insert('workerqueue', ['parameter' => $parameters, 'created' => $created, $added = DBA::insert('workerqueue', ['command' => $command, 'parameter' => $parameters, 'created' => $created,
'priority' => $priority, 'next_try' => $delayed]); 'priority' => $priority, 'next_try' => $delayed]);
if (!$added) { if (!$added) {
return false; return false;
} }
} elseif ($force_priority) { } elseif ($force_priority) {
DBA::update('workerqueue', ['priority' => $priority], ['parameter' => $parameters, 'done' => false, 'pid' => 0]); DBA::update('workerqueue', ['priority' => $priority], ['command' => $command, 'parameter' => $parameters, 'done' => false, 'pid' => 0]);
} }
// Set the IPC flag to ensure an immediate process execution via daemon // Set the IPC flag to ensure an immediate process execution via daemon
@ -1297,6 +1315,11 @@ class Worker
return $added; return $added;
} }
public static function countWorkersByCommand(string $command)
{
return DBA::count('workerqueue', ['done' => false, 'pid' => 0, 'command' => $command]);
}
/** /**
* Returns the next retrial level for worker jobs. * Returns the next retrial level for worker jobs.
* This function will skip levels when jobs are older. * This function will skip levels when jobs are older.

40
src/Model/Contact.php
View file

@ -1722,31 +1722,38 @@ class Contact
} }
// Search for duplicated contacts and get rid of them // Search for duplicated contacts and get rid of them
if (self::removeDuplicates(Strings::normaliseLink($url), $uid) || ($uid != 0)) { if (self::removeDuplicates(Strings::normaliseLink($url), $uid)) {
return; return;
} }
// Archive or unarchive the contact. We only need to do this for the public contact. // Archive or unarchive the contact.
// The archive/unarchive function will update the personal contacts by themselves.
$contact = DBA::selectFirst('contact', [], ['id' => $id]); $contact = DBA::selectFirst('contact', [], ['id' => $id]);
if (!DBA::isResult($contact)) { if (!DBA::isResult($contact)) {
Logger::info('Couldn\'t select contact for archival.', ['id' => $id]); Logger::info('Couldn\'t select contact for archival.', ['id' => $id]);
return; return;
} }
if (!empty($fields['success_update'])) { if (isset($fields['failed'])) {
self::unmarkForArchival($contact); if ($fields['failed']) {
} elseif (!empty($fields['failure_update'])) { self::markForArchival($contact);
self::markForArchival($contact); } else {
self::unmarkForArchival($contact);
}
} }
$condition = ['self' => false, 'nurl' => Strings::normaliseLink($url), 'network' => Protocol::FEDERATED]; if ($contact['uid'] != 0) {
return;
}
// These contacts are sharing with us, we don't poll them. // Update contact data for all users
// This means that we don't set the update fields in "OnePoll.php". $condition = ['self' => false, 'nurl' => Strings::normaliseLink($url)];
$condition['rel'] = self::SHARING;
$condition['network'] = [Protocol::DFRN, Protocol::DIASPORA, Protocol::ACTIVITYPUB];
DBA::update('contact', $fields, $condition); DBA::update('contact', $fields, $condition);
// We mustn't set the update fields for OStatus contacts since they are updated in OnePoll
$condition['network'] = Protocol::OSTATUS;
// If the contact failed, propagate the update fields to all contacts // If the contact failed, propagate the update fields to all contacts
if (empty($fields['failed'])) { if (empty($fields['failed'])) {
unset($fields['last-update']); unset($fields['last-update']);
@ -1758,8 +1765,6 @@ class Contact
return; return;
} }
// We are polling these contacts, so we mustn't set the update fields here.
$condition['rel'] = [self::FOLLOWER, self::FRIEND];
DBA::update('contact', $fields, $condition); DBA::update('contact', $fields, $condition);
} }
@ -1957,7 +1962,7 @@ class Contact
$ret['name-date'] = $updated; $ret['name-date'] = $updated;
} }
if ($uid == 0) { if (($uid == 0) || in_array($ret['network'], [Protocol::DFRN, Protocol::DIASPORA, Protocol::ACTIVITYPUB])) {
$ret['last-update'] = $updated; $ret['last-update'] = $updated;
$ret['success_update'] = $updated; $ret['success_update'] = $updated;
} }
@ -2224,8 +2229,11 @@ class Contact
self::updateAvatar($contact_id, $ret['photo']); self::updateAvatar($contact_id, $ret['photo']);
// pull feed and consume it, which should subscribe to the hub. // pull feed and consume it, which should subscribe to the hub.
if ($contact['network'] == Protocol::OSTATUS) {
Worker::add(PRIORITY_HIGH, "OnePoll", $contact_id, "force"); Worker::add(PRIORITY_HIGH, 'OnePoll', $contact_id, 'force');
} else {
Worker::add(PRIORITY_HIGH, 'UpdateContact', $contact_id);
}
$owner = User::getOwnerDataById($user['uid']); $owner = User::getOwnerDataById($user['uid']);

123
src/Model/GServer.php
View file

@ -70,6 +70,22 @@ class GServer
const DETECT_NODEINFO_1 = 101; const DETECT_NODEINFO_1 = 101;
const DETECT_NODEINFO_2 = 102; const DETECT_NODEINFO_2 = 102;
/**
* Check for the existance of a server and adds it in the background if not existant
*
* @param string $url
* @param boolean $only_nodeinfo
* @return void
*/
public static function add(string $url, bool $only_nodeinfo = false)
{
if (self::getID($url, false)) {
return;
}
Worker::add(PRIORITY_LOW, 'UpdateGServer', $url, $only_nodeinfo);
}
/** /**
* Get the ID for the given server URL * Get the ID for the given server URL
* *
@ -124,59 +140,53 @@ class GServer
return self::check($server, $network, $force); return self::check($server, $network, $force);
} }
/** public static function getNextUpdateDate(bool $success, string $created = '', string $last_contact = '')
* Decides if a server needs to be updated, based upon several date fields
*
* @param date $created Creation date of that server entry
* @param date $updated When had the server entry be updated
* @param date $last_failure Last failure when contacting that server
* @param date $last_contact Last time the server had been contacted
*
* @return boolean Does the server record needs an update?
*/
public static function updateNeeded($created, $updated, $last_failure, $last_contact)
{ {
// On successful contact process check again next week
if ($success) {
return DateTimeFormat::utc('now +7 day');
}
$now = strtotime(DateTimeFormat::utcNow()); $now = strtotime(DateTimeFormat::utcNow());
if ($updated > $last_contact) { if ($created > $last_contact) {
$contact_time = strtotime($updated); $contact_time = strtotime($created);
} else { } else {
$contact_time = strtotime($last_contact); $contact_time = strtotime($last_contact);
} }
$failure_time = strtotime($last_failure); // If the last contact was less than 6 hours before then try again in 6 hours
$created_time = strtotime($created); if (($now - $contact_time) < (60 * 60 * 6)) {
return DateTimeFormat::utc('now +6 hour');
// If there is no "created" time then use the current time
if ($created_time <= 0) {
$created_time = $now;
} }
// If the last contact was less than 24 hours then don't update // If the last contact was less than 12 hours before then try again in 12 hours
if (($now - $contact_time) < (60 * 60 * 12)) {
return DateTimeFormat::utc('now +12 hour');
}
// If the last contact was less than 24 hours before then try tomorrow again
if (($now - $contact_time) < (60 * 60 * 24)) { if (($now - $contact_time) < (60 * 60 * 24)) {
return false; return DateTimeFormat::utc('now +1 day');
} }
// If the last failure was less than 24 hours then don't update // If the last contact was less than a week before then try again in a week
if (($now - $failure_time) < (60 * 60 * 24)) { if (($now - $contact_time) < (60 * 60 * 24 * 7)) {
return false; return DateTimeFormat::utc('now +1 week');
} }
// If the last contact was less than a week ago and the last failure is older than a week then don't update // If the last contact was less than two weeks before then try again in two week
//if ((($now - $contact_time) < (60 * 60 * 24 * 7)) && ($contact_time > $failure_time)) if (($now - $contact_time) < (60 * 60 * 24 * 14)) {
// return false; return DateTimeFormat::utc('now +2 week');
// If the last contact time was more than a week ago and the contact was created more than a week ago, then only try once a week
if ((($now - $contact_time) > (60 * 60 * 24 * 7)) && (($now - $created_time) > (60 * 60 * 24 * 7)) && (($now - $failure_time) < (60 * 60 * 24 * 7))) {
return false;
} }
// If the last contact time was more than a month ago and the contact was created more than a month ago, then only try once a month // If the last contact was less than a month before then try again in a month
if ((($now - $contact_time) > (60 * 60 * 24 * 30)) && (($now - $created_time) > (60 * 60 * 24 * 30)) && (($now - $failure_time) < (60 * 60 * 24 * 30))) { if (($now - $contact_time) < (60 * 60 * 24 * 30)) {
return false; return DateTimeFormat::utc('now +1 month');
} }
return true; // The system hadn't been successul contacted for more than a month, so try again in three months
return DateTimeFormat::utc('now +3 month');
} }
/** /**
@ -192,7 +202,6 @@ class GServer
public static function check(string $server_url, string $network = '', bool $force = false, bool $only_nodeinfo = false) public static function check(string $server_url, string $network = '', bool $force = false, bool $only_nodeinfo = false)
{ {
$server_url = self::cleanURL($server_url); $server_url = self::cleanURL($server_url);
if ($server_url == '') { if ($server_url == '') {
return false; return false;
} }
@ -205,24 +214,11 @@ class GServer
DBA::update('gserver', $fields, $condition); DBA::update('gserver', $fields, $condition);
} }
$last_contact = $gserver['last_contact']; if (!$force && (strtotime($gserver['next_contact']) > time())) {
$last_failure = $gserver['last_failure'];
// See discussion under https://forum.friendi.ca/display/0b6b25a8135aabc37a5a0f5684081633
// It can happen that a zero date is in the database, but storing it again is forbidden.
if ($last_contact < DBA::NULL_DATETIME) {
$last_contact = DBA::NULL_DATETIME;
}
if ($last_failure < DBA::NULL_DATETIME) {
$last_failure = DBA::NULL_DATETIME;
}
if (!$force && !self::updateNeeded($gserver['created'], '', $last_failure, $last_contact)) {
Logger::info('No update needed', ['server' => $server_url]); Logger::info('No update needed', ['server' => $server_url]);
return ($last_contact >= $last_failure); return (!$gserver['failed']);
} }
Logger::info('Server is outdated. Start discovery.', ['Server' => $server_url, 'Force' => $force, 'Created' => $gserver['created'], 'Failure' => $last_failure, 'Contact' => $last_contact]); Logger::info('Server is outdated. Start discovery.', ['Server' => $server_url, 'Force' => $force]);
} else { } else {
Logger::info('Server is unknown. Start discovery.', ['Server' => $server_url]); Logger::info('Server is unknown. Start discovery.', ['Server' => $server_url]);
} }
@ -235,10 +231,13 @@ class GServer
* *
* @param string $url * @param string $url
*/ */
private static function setFailure(string $url) public static function setFailure(string $url)
{ {
if (DBA::exists('gserver', ['nurl' => Strings::normaliseLink($url)])) { $gserver = DBA::selectFirst('gserver', [], ['nurl' => Strings::normaliseLink($url)]);
DBA::update('gserver', ['failed' => true, 'last_failure' => DateTimeFormat::utcNow(), 'detection-method' => null], if (DBA::isResult($gserver)) {
$next_update = self::getNextUpdateDate(false, $gserver['created'], $gserver['last_contact']);
DBA::update('gserver', ['failed' => true, 'last_failure' => DateTimeFormat::utcNow(),
'next_contact' => $next_update, 'detection-method' => null],
['nurl' => Strings::normaliseLink($url)]); ['nurl' => Strings::normaliseLink($url)]);
Logger::info('Set failed status for existing server', ['url' => $url]); Logger::info('Set failed status for existing server', ['url' => $url]);
return; return;
@ -306,6 +305,7 @@ class GServer
// If the URL missmatches, then we mark the old entry as failure // If the URL missmatches, then we mark the old entry as failure
if ($url != $original_url) { if ($url != $original_url) {
/// @todo What to do with "next_contact" here?
DBA::update('gserver', ['failed' => true, 'last_failure' => DateTimeFormat::utcNow()], DBA::update('gserver', ['failed' => true, 'last_failure' => DateTimeFormat::utcNow()],
['nurl' => Strings::normaliseLink($original_url)]); ['nurl' => Strings::normaliseLink($original_url)]);
} }
@ -452,6 +452,8 @@ class GServer
$serverdata = self::detectNetworkViaContacts($url, $serverdata); $serverdata = self::detectNetworkViaContacts($url, $serverdata);
} }
$serverdata['next_contact'] = self::getNextUpdateDate(true);
$serverdata['last_contact'] = DateTimeFormat::utcNow(); $serverdata['last_contact'] = DateTimeFormat::utcNow();
$serverdata['failed'] = false; $serverdata['failed'] = false;
@ -1593,13 +1595,6 @@ class GServer
); );
while ($gserver = DBA::fetch($gservers)) { while ($gserver = DBA::fetch($gservers)) {
if (!GServer::check($gserver['url'], $gserver['network'])) {
// The server is not reachable? Okay, then we will try it later
$fields = ['last_poco_query' => DateTimeFormat::utcNow()];
DBA::update('gserver', $fields, ['nurl' => $gserver['nurl']]);
continue;
}
Logger::info('Update peer list', ['server' => $gserver['url'], 'id' => $gserver['id']]); Logger::info('Update peer list', ['server' => $gserver['url'], 'id' => $gserver['id']]);
Worker::add(PRIORITY_LOW, 'UpdateServerPeers', $gserver['url']); Worker::add(PRIORITY_LOW, 'UpdateServerPeers', $gserver['url']);
@ -1642,7 +1637,7 @@ class GServer
if (!empty($data['data']['nodes'])) { if (!empty($data['data']['nodes'])) {
foreach ($data['data']['nodes'] as $server) { foreach ($data['data']['nodes'] as $server) {
// Using "only_nodeinfo" since servers that are listed on that page should always have it. // Using "only_nodeinfo" since servers that are listed on that page should always have it.
Worker::add(PRIORITY_LOW, 'UpdateGServer', 'https://' . $server['host'], true); self::add('https://' . $server['host'], true);
} }
} }
} }
@ -1661,7 +1656,7 @@ class GServer
foreach ($servers['instances'] as $server) { foreach ($servers['instances'] as $server) {
$url = (is_null($server['https_score']) ? 'http' : 'https') . '://' . $server['name']; $url = (is_null($server['https_score']) ? 'http' : 'https') . '://' . $server['name'];
Worker::add(PRIORITY_LOW, 'UpdateGServer', $url); self::add($url);
} }
} }
} }

4
src/Module/Contact.php
View file

@ -181,9 +181,11 @@ class Contact extends BaseModule
if ($result['success']) { if ($result['success']) {
DBA::update('contact', ['subhub' => 1], ['id' => $contact_id]); DBA::update('contact', ['subhub' => 1], ['id' => $contact_id]);
} }
} else {
// pull feed and consume it, which should subscribe to the hub. // pull feed and consume it, which should subscribe to the hub.
Worker::add(PRIORITY_HIGH, 'OnePoll', $contact_id, 'force'); Worker::add(PRIORITY_HIGH, 'OnePoll', $contact_id, 'force');
} else {
Worker::add(PRIORITY_HIGH, 'UpdateContact', $contact_id);
} }
} }

7
src/Worker/Cron.php
View file

@ -61,7 +61,10 @@ class Cron
Worker::add(PRIORITY_MEDIUM, 'PollContacts'); Worker::add(PRIORITY_MEDIUM, 'PollContacts');
// Update contact information // Update contact information
Worker::add(PRIORITY_LOW, 'UpdatePublicContacts'); Worker::add(PRIORITY_LOW, 'UpdateContacts');
// Update server information
Worker::add(PRIORITY_LOW, 'UpdateGServers');
// run the process to update server directories in the background // run the process to update server directories in the background
Worker::add(PRIORITY_LOW, 'UpdateServerDirectories'); Worker::add(PRIORITY_LOW, 'UpdateServerDirectories');
@ -103,8 +106,6 @@ class Cron
// update nodeinfo data // update nodeinfo data
Worker::add(PRIORITY_LOW, 'NodeInfo'); Worker::add(PRIORITY_LOW, 'NodeInfo');
Worker::add(PRIORITY_LOW, 'UpdateGServers');
// Repair entries in the database // Repair entries in the database
Worker::add(PRIORITY_LOW, 'RepairDatabase'); Worker::add(PRIORITY_LOW, 'RepairDatabase');

2
src/Worker/OnePoll.php
View file

@ -56,7 +56,7 @@ class OnePoll
// We never probe mail contacts since their probing demands a mail from the contact in the inbox. // We never probe mail contacts since their probing demands a mail from the contact in the inbox.
// We don't probe feed accounts by default since they are polled in a higher frequency, but forced probes are okay. // We don't probe feed accounts by default since they are polled in a higher frequency, but forced probes are okay.
if (!in_array($contact['network'], [Protocol::MAIL, Protocol::FEED]) || ($force && ($contact['network'] == Protocol::FEED))) { if ($force && ($contact['network'] == Protocol::FEED)) {
$success = Contact::updateFromProbe($contact_id); $success = Contact::updateFromProbe($contact_id);
} else { } else {
$success = true; $success = true;

3
src/Worker/PollContacts.php
View file

@ -41,8 +41,7 @@ class PollContacts
$abandon_days = 0; $abandon_days = 0;
} }
$condition = ['network' => [Protocol::DFRN, Protocol::ACTIVITYPUB, Protocol::OSTATUS, Protocol::FEED, $condition = ['network' => [Protocol::FEED, Protocol::MAIL, Protocol::OSTATUS], 'self' => false, 'blocked' => false];
Protocol::MAIL, Protocol::ZOT, Protocol::PHANTOM], 'self' => false, 'blocked' => false];
if (!empty($abandon_days)) { if (!empty($abandon_days)) {
$condition = DBA::mergeConditions($condition, $condition = DBA::mergeConditions($condition,

59
src/Worker/UpdatePublicContacts.php → src/Worker/UpdateContacts.php
View file

@ -29,41 +29,56 @@ use Friendica\DI;
use Friendica\Util\DateTimeFormat; use Friendica\Util\DateTimeFormat;
/** /**
* Update public contacts * Update federated contacts
*/ */
class UpdatePublicContacts class UpdateContacts
{ {
public static function execute() public static function execute()
{ {
$count = 0; $base_condition = ['network' => array_merge(Protocol::FEDERATED, [Protocol::ZOT, Protocol::PHANTOM]), 'self' => false];
$ids = [];
$base_condition = ['network' => Protocol::FEDERATED, 'uid' => 0, 'self' => false]; $update_limit = DI::config()->get('system', 'contact_update_limit');
if (empty($update_limit)) {
return;
}
$updating = Worker::countWorkersByCommand('UpdateContact');
$limit = $update_limit - $updating;
if ($limit <= 0) {
Logger::info('The number of currently running jobs exceed the limit');
return;
}
// Add every contact our system interacted with and hadn't been updated for a week if unarchived
// or for a month if archived.
$condition = DBA::mergeConditions($base_condition, ["(`id` IN (SELECT `author-id` FROM `item`) OR
`id` IN (SELECT `owner-id` FROM `item`) OR `id` IN (SELECT `causer-id` FROM `item`) OR
`id` IN (SELECT `cid` FROM `post-tag`) OR `id` IN (SELECT `cid` FROM `user-contact`) OR `uid` != ?) AND
(`last-update` < ? OR (NOT `archive` AND `last-update` < ?))",
0, DateTimeFormat::utc('now - 1 month'), DateTimeFormat::utc('now - 1 week')]);
Logger::info('Updatable interacting federated contacts', ['count' => DBA::count('contact', $condition)]);
$ids = self::getContactsToUpdate($condition, [], $limit);
Logger::info('Fetched interacting federated contacts', ['count' => count($ids)]);
if (!DI::config()->get('system', 'update_active_contacts')) { if (!DI::config()->get('system', 'update_active_contacts')) {
// Add every contact (mostly failed ones) that hadn't been updated for six months // Add every contact (mostly failed ones) that hadn't been updated for six months
// and every non failed contact that hadn't been updated for a month
$condition = DBA::mergeConditions($base_condition, $condition = DBA::mergeConditions($base_condition,
["`last-update` < ?", DateTimeFormat::utc('now - 6 month')]); ["(`last-update` < ? OR (NOT `archive` AND `last-update` < ?))",
$ids = self::getContactsToUpdate($condition, $ids); DateTimeFormat::utc('now - 6 month'), DateTimeFormat::utc('now - 1 month')]);
Logger::info('Updatable federated contacts', ['count' => DBA::count('contact', $condition)]);
// Add every non failed contact that hadn't been updated for a month $previous = count($ids);
$condition = DBA::mergeConditions($base_condition, $ids = self::getContactsToUpdate($condition, $ids, $limit - $previous);
["NOT `failed` AND `last-update` < ?", DateTimeFormat::utc('now - 1 month')]); Logger::info('Fetched federated contacts', ['count' => count($ids) - $previous]);
$ids = self::getContactsToUpdate($condition, $ids);
} }
// Add every contact our system interacted with and hadn't been updated for a week $count = 0;
$condition = DBA::mergeConditions($base_condition, ["(`id` IN (SELECT `author-id` FROM `item`) OR
`id` IN (SELECT `owner-id` FROM `item`) OR `id` IN (SELECT `causer-id` FROM `item`) OR
`id` IN (SELECT `cid` FROM `post-tag`) OR `id` IN (SELECT `cid` FROM `user-contact`)) AND
`last-update` < ?", DateTimeFormat::utc('now - 1 week')]);
$ids = self::getContactsToUpdate($condition, $ids);
foreach ($ids as $id) { foreach ($ids as $id) {
Worker::add(PRIORITY_LOW, "UpdateContact", $id); Worker::add(PRIORITY_LOW, "UpdateContact", $id);
++$count; ++$count;
} }
Logger::info('Initiated update for public contacts', ['count' => $count]); Logger::info('Initiated update for federated contacts', ['count' => $count]);
} }
/** /**
@ -73,9 +88,9 @@ class UpdatePublicContacts
* @param array $ids * @param array $ids
* @return array contact ids * @return array contact ids
*/ */
private static function getContactsToUpdate(array $condition, array $ids = []) private static function getContactsToUpdate(array $condition, array $ids = [], int $limit)
{ {
$contacts = DBA::select('contact', ['id'], $condition, ['limit' => 100, 'order' => ['last-update']]); $contacts = DBA::select('contact', ['id'], $condition, ['limit' => $limit, 'order' => ['last-update']]);
while ($contact = DBA::fetch($contacts)) { while ($contact = DBA::fetch($contacts)) {
$ids[] = $contact['id']; $ids[] = $contact['id'];
} }

21
src/Worker/UpdateGServer.php
View file

@ -22,6 +22,7 @@
namespace Friendica\Worker; namespace Friendica\Worker;
use Friendica\Core\Logger; use Friendica\Core\Logger;
use Friendica\Database\DBA;
use Friendica\Model\GServer; use Friendica\Model\GServer;
use Friendica\Util\Strings; use Friendica\Util\Strings;
@ -38,12 +39,24 @@ class UpdateGServer
return; return;
} }
$server_url = filter_var($server_url, FILTER_SANITIZE_URL); $filtered = filter_var($server_url, FILTER_SANITIZE_URL);
if (substr(Strings::normaliseLink($server_url), 0, 7) != 'http://') { if (substr(Strings::normaliseLink($filtered), 0, 7) != 'http://') {
GServer::setFailure($server_url);
return; return;
} }
$ret = GServer::check($server_url, '', false, $only_nodeinfo); if (($filtered != $server_url) && DBA::exists('gserver', ['nurl' => Strings::normaliseLink($server_url)])) {
Logger::info('Updated gserver', ['url' => $server_url, 'result' => $ret]); GServer::setFailure($server_url);
return;
}
$cleaned = GServer::cleanURL($server_url);
if (($cleaned != $server_url) && DBA::exists('gserver', ['nurl' => Strings::normaliseLink($server_url)])) {
GServer::setFailure($server_url);
return;
}
$ret = GServer::check($filtered, '', true, $only_nodeinfo);
Logger::info('Updated gserver', ['url' => $filtered, 'result' => $ret]);
} }
} }

44
src/Worker/UpdateGServers.php
View file

@ -24,34 +24,52 @@ namespace Friendica\Worker;
use Friendica\Core\Logger; use Friendica\Core\Logger;
use Friendica\Core\Worker; use Friendica\Core\Worker;
use Friendica\Database\DBA; use Friendica\Database\DBA;
use Friendica\Model\GServer; use Friendica\DI;
use Friendica\Util\Strings;
class UpdateGServers class UpdateGServers
{ {
/** /**
* Updates the first 250 servers * Updates a defined number of servers
*/ */
public static function execute() public static function execute()
{ {
$gservers = DBA::p("SELECT `url`, `created`, `last_failure`, `last_contact` FROM `gserver` ORDER BY rand()"); $update_limit = DI::config()->get('system', 'gserver_update_limit');
if (empty($update_limit)) {
return;
}
$updating = Worker::countWorkersByCommand('UpdateGServer');
$limit = $update_limit - $updating;
if ($limit <= 0) {
Logger::info('The number of currently running jobs exceed the limit');
return;
}
$total = DBA::count('gserver');
$condition = ["`next_contact` < UTC_TIMESTAMP() AND (`nurl` != ? OR `url` != ?)", '', ''];
$outdated = DBA::count('gserver', $condition);
Logger::info('Server status', ['total' => $total, 'outdated' => $outdated, 'updating' => $limit]);
$gservers = DBA::select('gserver', ['url', 'nurl'], $condition, ['limit' => $limit]);
if (!DBA::isResult($gservers)) { if (!DBA::isResult($gservers)) {
return; return;
} }
$updated = 0; $count = 0;
while ($gserver = DBA::fetch($gservers)) { while ($gserver = DBA::fetch($gservers)) {
if (!GServer::updateNeeded($gserver['created'], '', $gserver['last_failure'], $gserver['last_contact'])) { // Sometimes the "nurl" and "url" doesn't seem to fit, see https://forum.friendi.ca/display/ec054ce7-155f-c94d-6159-f50372664245
continue; // There are duplicated "url" but not "nurl". So we check both addresses instead of just overwriting them,
// since that would mean loosing data.
if (!empty($gserver['url'])) {
Worker::add(PRIORITY_LOW, 'UpdateGServer', $gserver['url']);
} }
Logger::info('Update server status', ['server' => $gserver['url']]); if (!empty($gserver['nurl']) && ($gserver['nurl'] != Strings::normaliseLink($gserver['url']))) {
Worker::add(PRIORITY_LOW, 'UpdateGServer', $gserver['nurl']);
Worker::add(PRIORITY_LOW, 'UpdateGServer', $gserver['url']);
if (++$updated > 250) {
return;
} }
$count++;
} }
DBA::close($gservers); DBA::close($gservers);
Logger::info('Updated servers', ['count' => $count]);
} }
} }

4
src/Worker/UpdateServerPeers.php
View file

@ -22,9 +22,9 @@
namespace Friendica\Worker; namespace Friendica\Worker;
use Friendica\Core\Logger; use Friendica\Core\Logger;
use Friendica\Core\Worker;
use Friendica\Database\DBA; use Friendica\Database\DBA;
use Friendica\DI; use Friendica\DI;
use Friendica\Model\GServer;
use Friendica\Util\Strings; use Friendica\Util\Strings;
class UpdateServerPeers class UpdateServerPeers
@ -58,7 +58,7 @@ class UpdateServerPeers
continue; continue;
} }
// This endpoint doesn't offer the schema. So we assume that it is HTTPS. // This endpoint doesn't offer the schema. So we assume that it is HTTPS.
Worker::add(PRIORITY_LOW, 'UpdateGServer', 'https://' . $peer); GServer::add('https://' . $peer);
++$added; ++$added;
} }
Logger::info('Server peer update ended', ['total' => $total, 'added' => $added, 'url' => $url]); Logger::info('Server peer update ended', ['total' => $total, 'added' => $added, 'url' => $url]);

14
static/dbstructure.config.php
View file

@ -55,7 +55,7 @@
use Friendica\Database\DBA; use Friendica\Database\DBA;
if (!defined('DB_UPDATE_VERSION')) { if (!defined('DB_UPDATE_VERSION')) {
define('DB_UPDATE_VERSION', 1382); define('DB_UPDATE_VERSION', 1383);
} }
return [ return [
@ -81,13 +81,15 @@ return [
"detection-method" => ["type" => "tinyint unsigned", "comment" => "Method that had been used to detect that server"], "detection-method" => ["type" => "tinyint unsigned", "comment" => "Method that had been used to detect that server"],
"created" => ["type" => "datetime", "not null" => "1", "default" => DBA::NULL_DATETIME, "comment" => ""], "created" => ["type" => "datetime", "not null" => "1", "default" => DBA::NULL_DATETIME, "comment" => ""],
"last_poco_query" => ["type" => "datetime", "default" => DBA::NULL_DATETIME, "comment" => ""], "last_poco_query" => ["type" => "datetime", "default" => DBA::NULL_DATETIME, "comment" => ""],
"last_contact" => ["type" => "datetime", "default" => DBA::NULL_DATETIME, "comment" => ""], "last_contact" => ["type" => "datetime", "default" => DBA::NULL_DATETIME, "comment" => "Last successful connection request"],
"last_failure" => ["type" => "datetime", "default" => DBA::NULL_DATETIME, "comment" => ""], "last_failure" => ["type" => "datetime", "default" => DBA::NULL_DATETIME, "comment" => "Last failed connection request"],
"failed" => ["type" => "boolean", "comment" => "Connection failed"], "failed" => ["type" => "boolean", "comment" => "Connection failed"],
"next_contact" => ["type" => "datetime", "default" => DBA::NULL_DATETIME, "comment" => "Next connection request"],
], ],
"indexes" => [ "indexes" => [
"PRIMARY" => ["id"], "PRIMARY" => ["id"],
"nurl" => ["UNIQUE", "nurl(190)"], "nurl" => ["UNIQUE", "nurl(190)"],
"next_contact" => ["next_contact"],
] ]
], ],
"user" => [ "user" => [
@ -1509,7 +1511,8 @@ return [
"comment" => "Background tasks queue entries", "comment" => "Background tasks queue entries",
"fields" => [ "fields" => [
"id" => ["type" => "int unsigned", "not null" => "1", "extra" => "auto_increment", "primary" => "1", "comment" => "Auto incremented worker task id"], "id" => ["type" => "int unsigned", "not null" => "1", "extra" => "auto_increment", "primary" => "1", "comment" => "Auto incremented worker task id"],
"parameter" => ["type" => "mediumtext", "comment" => "Task command"], "command" => ["type" => "varchar(100)", "comment" => "Task command"],
"parameter" => ["type" => "mediumtext", "comment" => "Task parameter"],
"priority" => ["type" => "tinyint unsigned", "not null" => "1", "default" => "0", "comment" => "Task priority"], "priority" => ["type" => "tinyint unsigned", "not null" => "1", "default" => "0", "comment" => "Task priority"],
"created" => ["type" => "datetime", "not null" => "1", "default" => DBA::NULL_DATETIME, "comment" => "Creation date"], "created" => ["type" => "datetime", "not null" => "1", "default" => DBA::NULL_DATETIME, "comment" => "Creation date"],
"pid" => ["type" => "int unsigned", "not null" => "1", "default" => "0", "comment" => "Process id of the worker"], "pid" => ["type" => "int unsigned", "not null" => "1", "default" => "0", "comment" => "Process id of the worker"],
@ -1520,7 +1523,8 @@ return [
], ],
"indexes" => [ "indexes" => [
"PRIMARY" => ["id"], "PRIMARY" => ["id"],
"done_parameter" => ["done", "parameter(64)"], "command" => ["command"],
"done_command_parameter" => ["done", "command", "parameter(64)"],
"done_executed" => ["done", "executed"], "done_executed" => ["done", "executed"],
"done_priority_retrial_created" => ["done", "priority", "retrial", "created"], "done_priority_retrial_created" => ["done", "priority", "retrial", "created"],
"done_priority_next_try" => ["done", "priority", "next_try"], "done_priority_next_try" => ["done", "priority", "next_try"],

8
static/defaults.config.php
View file

@ -131,6 +131,10 @@ return [
// Don't display sharing accounts on the global community // Don't display sharing accounts on the global community
'community_no_sharer' => false, 'community_no_sharer' => false,
// contact_update_limit (Integer)
// How much contacts should be checked at a time?
'contact_update_limit' => 100,
// cron_interval (Integer) // cron_interval (Integer)
// Minimal period in minutes between two calls of the "Cron" worker job. // Minimal period in minutes between two calls of the "Cron" worker job.
'cron_interval' => 5, 'cron_interval' => 5,
@ -254,6 +258,10 @@ return [
// This can alternatively be set on a per account basis in the pconfig table. // This can alternatively be set on a per account basis in the pconfig table.
'groupedit_image_limit' => 400, 'groupedit_image_limit' => 400,
// gserver_update_limit (Integer)
// How much servers should be checked at a time?
'gserver_update_limit' => 100,
// hsts (Boolean) // hsts (Boolean)
// Enables the sending of HTTP Strict Transport Security headers. // Enables the sending of HTTP Strict Transport Security headers.
'hsts' => false, 'hsts' => false,