Skip to content

Commit

Permalink
Merge pull request #22812 from totten/master-queue-schema
Browse files Browse the repository at this point in the history 
(dev/core#1304) Queues - Define schema for runner, run_count, lease_time, et al
  • Loading branch information
@totten
totten authored Mar 1, 2022
2 parents 7058955 + f029797 commit 245c061
Show file tree
Hide file tree
Showing 8 changed files with 301 additions and 77 deletions.
34 changes: 20 additions & 14 deletions CRM/Queue/BAO/QueueItem.php
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,25 +25,31 @@ class CRM_Queue_BAO_QueueItem extends CRM_Queue_DAO_QueueItem {
/**
* Ensure that the required SQL table exists.
*
* The `civicrm_queue_item` table is a special requirement - without it, the upgrader cannot run.
* The upgrader will make a special request for `findCreateTable()` before computing upgrade-tasks.
*
* @return bool
* TRUE if table now exists
*/
public static function findCreateTable() {
$checkTableSql = "show tables like 'civicrm_queue_item'";
$foundName = CRM_Core_DAO::singleValueQuery($checkTableSql);
if ($foundName == 'civicrm_queue_item') {
return TRUE;
public static function findCreateTable(): bool {
if (!CRM_Core_DAO::checkTableExists('civicrm_queue_item')) {
// Table originated in 4.2. We no longer support direct upgrades from <=4.2. Don't bother trying to create table.
return FALSE;
}
else {
return static::updateTable();
}
}

// civicrm/sql/civicrm_queue_item.mysql
$fileName = dirname(__FILE__) . '/../../../sql/civicrm_queue_item.mysql';

$config = CRM_Core_Config::singleton();
CRM_Utils_File::sourceSQLFile($config->dsn, $fileName);

// Make sure it succeeded
$foundName = CRM_Core_DAO::singleValueQuery($checkTableSql);
return ($foundName == 'civicrm_queue_item');
/**
* Ensure that the `civicrm_queue_item` table is up-to-date.
*
* @return bool
*/
public static function updateTable(): bool {
CRM_Upgrade_Incremental_Base::addColumn(NULL, 'civicrm_queue_item', 'run_count',
"int NOT NULL DEFAULT 0 COMMENT 'Number of times execution has been attempted.'");
return TRUE;
}

}
133 changes: 119 additions & 14 deletions CRM/Queue/DAO/Queue.php
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
*
* Generated from xml/schema/CRM/Queue/Queue.xml
* DO NOT EDIT. Generated by CRM_Core_CodeGen
* (GenCodeChecksum:5a49f8a2765460d48e4266efa9d447f2)
* (GenCodeChecksum:3b50eca7549430727237a4b2e295df1f)
*/

/**
Expand Down Expand Up @@ -56,13 +56,49 @@ class CRM_Queue_DAO_Queue extends CRM_Core_DAO {
public $type;

/**
* Should the standard background attempt to autorun tasks in this queue?
* Name of the task runner
*
* @var bool|string|null
* (SQL type: tinyint)
* @var string
* (SQL type: varchar(64))
* Note that values will be retrieved from the database as a string.
*/
public $runner;

/**
* Maximum number of items in a batch.
*
* @var int|string
* (SQL type: int unsigned)
* Note that values will be retrieved from the database as a string.
*/
public $batch_limit;

/**
* When claiming an item (or batch of items) for work, how long should the item(s) be reserved. (Seconds)
*
* @var int|string
* (SQL type: int unsigned)
* Note that values will be retrieved from the database as a string.
*/
public $lease_time;

/**
* Number of permitted retries. Set to zero (0) to disable.
*
* @var int|string
* (SQL type: int)
* Note that values will be retrieved from the database as a string.
*/
public $is_autorun;
public $retry_limit;

/**
* Number of seconds to wait before retrying a failed execution.
*
* @var int|string
* (SQL type: int)
* Note that values will be retrieved from the database as a string.
*/
public $retry_interval;

/**
* Class constructor.
Expand Down Expand Up @@ -145,21 +181,90 @@ public static function &fields() {
],
'add' => '5.47',
],
'is_autorun' => [
'name' => 'is_autorun',
'type' => CRM_Utils_Type::T_BOOLEAN,
'title' => ts('Enable Autorun'),
'description' => ts('Should the standard background attempt to autorun tasks in this queue?'),
'where' => 'civicrm_queue.is_autorun',
'runner' => [
'name' => 'runner',
'type' => CRM_Utils_Type::T_STRING,
'title' => ts('Runner'),
'description' => ts('Name of the task runner'),
'required' => FALSE,
'maxlength' => 64,
'size' => CRM_Utils_Type::BIG,
'where' => 'civicrm_queue.runner',
'table_name' => 'civicrm_queue',
'entity' => 'Queue',
'bao' => 'CRM_Queue_BAO_Queue',
'localizable' => 0,
'html' => [
'type' => 'CheckBox',
'label' => ts("Auto Run"),
'type' => 'Text',
],
'add' => '5.48',
],
'batch_limit' => [
'name' => 'batch_limit',
'type' => CRM_Utils_Type::T_INT,
'title' => ts('Batch Limit'),
'description' => ts('Maximum number of items in a batch.'),
'required' => TRUE,
'where' => 'civicrm_queue.batch_limit',
'default' => '1',
'table_name' => 'civicrm_queue',
'entity' => 'Queue',
'bao' => 'CRM_Queue_BAO_Queue',
'localizable' => 0,
'html' => [
'type' => 'Text',
],
'add' => '5.48',
],
'lease_time' => [
'name' => 'lease_time',
'type' => CRM_Utils_Type::T_INT,
'title' => ts('Lease Time'),
'description' => ts('When claiming an item (or batch of items) for work, how long should the item(s) be reserved. (Seconds)'),
'required' => TRUE,
'where' => 'civicrm_queue.lease_time',
'default' => '3600',
'table_name' => 'civicrm_queue',
'entity' => 'Queue',
'bao' => 'CRM_Queue_BAO_Queue',
'localizable' => 0,
'html' => [
'type' => 'Text',
],
'add' => '5.48',
],
'retry_limit' => [
'name' => 'retry_limit',
'type' => CRM_Utils_Type::T_INT,
'title' => ts('Retry Limit'),
'description' => ts('Number of permitted retries. Set to zero (0) to disable.'),
'required' => TRUE,
'where' => 'civicrm_queue.retry_limit',
'default' => '0',
'table_name' => 'civicrm_queue',
'entity' => 'Queue',
'bao' => 'CRM_Queue_BAO_Queue',
'localizable' => 0,
'html' => [
'type' => 'Text',
],
'add' => '5.48',
],
'retry_interval' => [
'name' => 'retry_interval',
'type' => CRM_Utils_Type::T_INT,
'title' => ts('Retry Interval'),
'description' => ts('Number of seconds to wait before retrying a failed execution.'),
'required' => FALSE,
'where' => 'civicrm_queue.retry_interval',
'table_name' => 'civicrm_queue',
'entity' => 'Queue',
'bao' => 'CRM_Queue_BAO_Queue',
'localizable' => 0,
'html' => [
'type' => 'Text',
],
'add' => NULL,
'add' => '5.48',
],
];
CRM_Core_DAO_AllCoreTables::invoke(__CLASS__, 'fields_callback', Civi::$statics[__CLASS__]['fields']);
Expand Down
28 changes: 27 additions & 1 deletion CRM/Queue/DAO/QueueItem.php
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
*
* Generated from xml/schema/CRM/Queue/QueueItem.xml
* DO NOT EDIT. Generated by CRM_Core_CodeGen
* (GenCodeChecksum:36871610524adb64bc8aa3bb24b295f4)
* (GenCodeChecksum:f5163d86b425127deb25d105976212bf)
*/

/**
Expand Down Expand Up @@ -71,6 +71,15 @@ class CRM_Queue_DAO_QueueItem extends CRM_Core_DAO {
*/
public $release_time;

/**
* Number of times execution has been attempted.
*
* @var int|string
* (SQL type: int)
* Note that values will be retrieved from the database as a string.
*/
public $run_count;

/**
* Serialized queue data
*
Expand Down Expand Up @@ -188,6 +197,23 @@ public static function &fields() {
],
'add' => NULL,
],
'run_count' => [
'name' => 'run_count',
'type' => CRM_Utils_Type::T_INT,
'title' => ts('Run Count'),
'description' => ts('Number of times execution has been attempted.'),
'required' => TRUE,
'where' => 'civicrm_queue_item.run_count',
'default' => '0',
'table_name' => 'civicrm_queue_item',
'entity' => 'QueueItem',
'bao' => 'CRM_Queue_BAO_QueueItem',
'localizable' => 0,
'html' => [
'type' => 'Text',
],
'add' => '5.48',
],
'data' => [
'name' => 'data',
'type' => CRM_Utils_Type::T_LONGTEXT,
Expand Down
42 changes: 32 additions & 10 deletions CRM/Queue/Service.php
View file
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,14 @@ class CRM_Queue_Service {

protected static $_singleton;

/**
* List of fields which are shared by `$queueSpec` and `civicrm_queue`.
*
* @var string[]
* @readonly
*/
private static $commonFields = ['name', 'type', 'runner', 'batch_limit', 'lease_time', 'retry_limit', 'retry_interval'];

/**
* FIXME: Singleton pattern should be removed when dependency-injection
* becomes available.
Expand Down Expand Up @@ -80,9 +88,12 @@ public function __construct() {
* flushed; default to TRUE
* - (additional keys depending on the queue provider).
* - is_persistent: bool, optional; if true, then this queue is loaded from `civicrm_queue` list
* - is_autorun: bool, optional; if true, then this queue will be auto-scanned
* by background task-runners
*
* - runner: string, optional; if given, then items in this queue can run
* automatically via `hook_civicrm_queueRun_{$runner}`
* - batch_limit: int, Maximum number of items in a batch.
* - lease_time: int, When claiming an item (or batch of items) for work, how long should the item(s) be reserved. (Seconds)
* - retry_limit: int, Number of permitted retries. Set to zero (0) to disable.
* - retry_interval: int, Number of seconds to wait before retrying a failed execution.
* @return CRM_Queue_Queue
*/
public function create($queueSpec) {
Expand Down Expand Up @@ -121,22 +132,33 @@ public function create($queueSpec) {
* @throws \CRM_Core_Exception
*/
protected function findCreateQueueSpec(array $queueSpec): array {
$storageFields = ['type', 'is_autorun'];
$dao = new CRM_Queue_DAO_Queue();
$dao->name = $queueSpec['name'];
if ($dao->find(TRUE)) {
return array_merge($queueSpec, CRM_Utils_Array::subset($dao->toArray(), $storageFields));
$loaded = $this->findQueueSpec($queueSpec);
if ($loaded !== NULL) {
return $loaded;
}

if (empty($queueSpec['type'])) {
throw new \CRM_Core_Exception(sprintf('Failed to find or create persistent queue "%s". Missing field "%s".',
$queueSpec['name'], 'type'));
}
$queueSpec = array_merge(['is_autorun' => FALSE], $queueSpec);

$dao = new CRM_Queue_DAO_Queue();
$dao->name = $queueSpec['name'];
$dao->copyValues($queueSpec);
$dao->insert();

return $queueSpec;
return $this->findQueueSpec($queueSpec);
}

protected function findQueueSpec(array $queueSpec): ?array {
$dao = new CRM_Queue_DAO_Queue();
$dao->name = $queueSpec['name'];
if ($dao->find(TRUE)) {
return array_merge($queueSpec, CRM_Utils_Array::subset($dao->toArray(), static::$commonFields));
}
else {
return NULL;
}
}

/**
Expand Down
30 changes: 30 additions & 0 deletions CRM/Upgrade/Incremental/php/FiveFortyEight.php
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,36 @@ class CRM_Upgrade_Incremental_php_FiveFortyEight extends CRM_Upgrade_Incremental
*/
public function upgrade_5_48_alpha1($rev): void {
$this->addTask(ts('Upgrade DB to %1: SQL', [1 => $rev]), 'runSql', $rev);
$this->addTask('Add "runner" to "civicrm_queue"', 'addColumn', 'civicrm_queue', 'runner',
"varchar(64) NULL COMMENT 'Name of the task runner'"
);
$this->addTask('Convert "is_autorun" to "runner"', 'convertAutorun');
$this->addTask('Drop "is_autorun" from "civicrm_queue"', 'dropColumn', 'civicrm_queue', 'is_autorun');
$this->addTask('Add "batch_limit" to "civicrm_queue"', 'addColumn', 'civicrm_queue', 'batch_limit',
"int unsigned NOT NULL DEFAULT 1 COMMENT 'Maximum number of items in a batch.'"
);
$this->addTask('Add "lease_time" to "civicrm_queue"', 'addColumn', 'civicrm_queue', 'lease_time',
"int unsigned NOT NULL DEFAULT 3600 COMMENT 'When claiming an item (or batch of items) for work, how long should the item(s) be reserved. (Seconds)'"
);
$this->addTask('Add "retry_limit" to "civicrm_queue"', 'addColumn', 'civicrm_queue', 'retry_limit',
"int NOT NULL DEFAULT 0 COMMENT 'Number of permitted retries. Set to zero (0) to disable.'"
);
$this->addTask('Add "retry_interval" to "civicrm_queue"', 'addColumn', 'civicrm_queue', 'retry_interval',
"int NULL COMMENT 'Number of seconds to wait before retrying a failed execution.'"
);
}

/**
* The `is_autorun` column was introduced in 5.47, but we didn't finish adding the
* additional changes to use, so there shouldn't be any real usage. But just to be
* paranoid, we'll convert to 5.48's `runner`.
*
* @param \CRM_Queue_TaskContext $ctx
* @return bool
*/
public static function convertAutorun(CRM_Queue_TaskContext $ctx) {
CRM_Core_DAO::executeQuery('UPDATE civicrm_queue SET runner = "task" WHERE is_autorun = 1');
return TRUE;
}

}
30 changes: 0 additions & 30 deletions sql/civicrm_queue_item.mysql
View file

This file was deleted.

Loading

0 comments on commit 245c061

Please sign in to comment.