diff --git a/data/Dockerfiles/dovecot/imapsync_runner.pl b/data/Dockerfiles/dovecot/imapsync_runner.pl index 35610c93c..c384552ef 100644 --- a/data/Dockerfiles/dovecot/imapsync_runner.pl +++ b/data/Dockerfiles/dovecot/imapsync_runner.pl @@ -13,19 +13,6 @@ use sigtrap 'handler' => \&sig_handler, qw(INT TERM KILL QUIT); sub trim { my $s = shift; $s =~ s/^\s+|\s+$//g; return $s }; -# Move a finished job to the back of the queue, keeping order_id contiguous (1..N). -# Called only from the parent after reaping a child, so all rotations are serialized. -sub rotate_to_back { - my ($dbh, $jid) = @_; - return unless defined $jid; - my ($cnt) = $dbh->selectrow_array("SELECT COUNT(*) FROM imapsync"); - my ($old) = $dbh->selectrow_array("SELECT order_id FROM imapsync WHERE id = ?", undef, $jid); - return unless (defined $cnt && defined $old && $old < $cnt); - # close the gap left behind, then place this job last - $dbh->do("UPDATE imapsync SET order_id = order_id - 1 WHERE order_id > ? ORDER BY order_id ASC", undef, $old); - $dbh->do("UPDATE imapsync SET order_id = ? WHERE id = ?", undef, $cnt, $jid); -} - $run_dir="/tmp"; $dsn = 'DBI:mysql:database=' . $ENV{'DBNAME'} . ';mysql_socket=/var/run/mysqld/mysqld.sock'; $lock_file = $run_dir . "/imapsync_busy"; @@ -92,29 +79,24 @@ my $sth = $dbh->prepare("SELECT UNIX_TIMESTAMP(NOW()) - UNIX_TIMESTAMP(i.last_run) > i.mins_interval * 60 OR i.last_run IS NULL) - ORDER BY i.order_id ASC, i.id ASC"); + ORDER BY i.last_run ASC, i.prio DESC, i.id ASC + LIMIT $max_parallel"); $sth->execute(); my @rows = @{$sth->fetchall_arrayref()}; $sth->finish(); -# Fork pool: run up to $max_parallel imapsync children at once, in order_id order. +# Only up to $max_parallel due jobs are pulled per run (LIMIT above) my $active = 0; -my %pid_job; # child pid => syncjob id, so the parent can rotate it to the back when it finishes foreach my $row (@rows) { # Throttle: once max children are running, reap one before starting the next if ($active >= $max_parallel) { - my $done = wait(); - my $rc = $? >> 8; + wait(); $active-- if $active > 0; - if (defined $pid_job{$done}) { - rotate_to_back($dbh, $pid_job{$done}) if $rc == 0; # rotate only jobs that actually ran - delete $pid_job{$done}; - } } my $pid = fork(); next if (!defined $pid); - if ($pid != 0) { $pid_job{$pid} = @$row[0]; $active++; next; } # parent: remember pid->job, continue + if ($pid != 0) { $active++; next; } # parent: one more child running, continue # ---- CHILD ---- own DB connection so the parent's handle stays intact across forks $dbh->{InactiveDestroy} = 1; @@ -179,8 +161,7 @@ foreach my $row (@rows) { $dbh->disconnect(); unlink $passfile1->filename if defined $passfile1; unlink $passfile2->filename if defined $passfile2; - # exit 75 (EX_TEMPFAIL): did not actually sync -> parent keeps its queue position (no rotate) - POSIX::_exit(75); + POSIX::_exit(75); # EX_TEMPFAIL: did not actually sync (token not ready) } $tokenfile1 = File::Temp->new(TEMPLATE => $template); binmode($tokenfile1, ":utf8"); @@ -287,14 +268,8 @@ foreach my $row (@rows) { POSIX::_exit(0); # ---- end CHILD ---- } -# Parent: wait for all remaining children, rotating each finished job to the back -while ((my $done = wait()) != -1) { - my $rc = $? >> 8; - if (defined $pid_job{$done}) { - rotate_to_back($dbh, $pid_job{$done}) if $rc == 0; # rotate only jobs that actually ran - delete $pid_job{$done}; - } -} +# Parent: wait for all remaining children to finish +while (wait() != -1) { } $dbh->disconnect(); diff --git a/data/web/api/openapi.yaml b/data/web/api/openapi.yaml index f22ad86a7..04524afab 100644 --- a/data/web/api/openapi.yaml +++ b/data/web/api/openapi.yaml @@ -4089,6 +4089,12 @@ paths: type: string v: type: string + prio: + description: >- + Priority (admin only; ignored otherwise). The runner processes due + jobs least-recently-run first, priority breaks ties (0 = normal, + higher wins). + type: number delete1: description: Delete from source when completed type: boolean @@ -4356,72 +4362,6 @@ paths: items: type: string type: object - /api/v1/edit/syncjob/order: - post: - responses: - "401": - $ref: "#/components/responses/Unauthorized" - "200": - content: - application/json: - examples: - response: - value: - - log: - - syncjob - - edit - - job_order - msg: - - imapsync_order_updated - - "1" - type: success - schema: - properties: - log: - description: contains request object - items: {} - type: array - msg: - items: {} - type: array - type: - enum: - - success - - danger - - error - type: string - type: object - description: OK - headers: {} - tags: - - Sync jobs - description: >- - Move a sync job to a global queue position (admin only). Positions are - contiguous (1..N) across all sync jobs; setting a job to a position - shifts the others accordingly. The runner processes jobs in ascending - order, up to the configured `max_parallel` at a time, and rotates a job - to the back after it runs. - operationId: Set sync job queue position - summary: Set sync job queue position - requestBody: - content: - application/json: - schema: - example: - items: "3" - attr: - order_id: "1" - properties: - attr: - properties: - order_id: - description: target position (1..N) - type: number - type: object - items: - description: sync job id to move - type: string - type: object /api/v1/edit/imapsync_settings: post: responses: diff --git a/data/web/inc/functions.syncjob.inc.php b/data/web/inc/functions.syncjob.inc.php index cd6155a6c..9acdc4e3c 100644 --- a/data/web/inc/functions.syncjob.inc.php +++ b/data/web/inc/functions.syncjob.inc.php @@ -385,42 +385,27 @@ function imapsync_set_setting($name, $value) { return true; } -function imapsync_set_order($id, $pos) { - // Move syncjob $id to global queue position $pos (1..N), shifting only the rows between - // the old and new position (range-shift, not a full renumber). Admin-only, transactional. +function imapsync_run_next($ids) { + // clear last_run and bump prio. + // Admin-only. global $pdo; - if (($_SESSION['mailcow_cc_role'] ?? '') !== 'admin' || !is_numeric($id)) return false; - $pos = intval($pos); + if (($_SESSION['mailcow_cc_role'] ?? '') !== 'admin') return false; + if (!is_array($ids)) $ids = array($ids); + $sel = array(); + foreach ($ids as $i) { if (is_numeric($i)) $sel[] = (int)$i; } + $sel = array_values(array_unique($sel)); + if (empty($sel)) return false; try { - $pdo->beginTransaction(); - $cnt = intval($pdo->query("SELECT COUNT(*) FROM `imapsync`")->fetchColumn()); - if ($cnt === 0) { $pdo->commit(); return false; } - if ($pos < 1) $pos = 1; - if ($pos > $cnt) $pos = $cnt; - $stmt = $pdo->prepare("SELECT `order_id` FROM `imapsync` WHERE `id` = :id"); - $stmt->execute(array(':id' => $id)); - $old = $stmt->fetchColumn(); - if ($old === false) { $pdo->rollBack(); return false; } - $old = intval($old); - if ($old !== $pos) { - if ($pos < $old) { - // moving up: shift the [pos, old-1] band down (+1), highest first to avoid collisions - $pdo->prepare("UPDATE `imapsync` SET `order_id` = `order_id` + 1 - WHERE `order_id` >= :pos AND `order_id` < :old ORDER BY `order_id` DESC") - ->execute(array(':pos' => $pos, ':old' => $old)); - } else { - // moving down: shift the (old, pos] band up (-1), lowest first - $pdo->prepare("UPDATE `imapsync` SET `order_id` = `order_id` - 1 - WHERE `order_id` > :old AND `order_id` <= :pos ORDER BY `order_id` ASC") - ->execute(array(':old' => $old, ':pos' => $pos)); - } - $pdo->prepare("UPDATE `imapsync` SET `order_id` = :pos WHERE `id` = :id") - ->execute(array(':pos' => $pos, ':id' => $id)); - } - $pdo->commit(); - return true; + $in = implode(',', array_fill(0, count($sel), '?')); + // Assign each selected job a prio above the current global max, lowest-current first. + $pdo->exec("SET @m := (SELECT COALESCE(MAX(`prio`), 0) FROM `imapsync`)"); + $pdo->exec("SET @r := 0"); + $pdo->prepare("UPDATE `imapsync` SET `prio` = @m + (@r := @r + 1) + WHERE `id` IN ($in) ORDER BY `prio` ASC, `id` ASC")->execute($sel); + $pdo->prepare("UPDATE `imapsync` SET `last_run` = NULL, `success` = NULL + WHERE `id` IN ($in)")->execute($sel); + return count($sel); } catch (Exception $e) { - if ($pdo->inTransaction()) $pdo->rollBack(); return false; } } @@ -576,9 +561,8 @@ function syncjob($_action, $_type, $_data = null, $_extra = null) { ); return false; } - $stmt = $pdo->prepare("INSERT INTO `imapsync` (`user2`, `exclude`, `delete1`, `delete2`, `timeout1`, `timeout2`, `automap`, `skipcrossduplicates`, `maxbytespersecond`, `subscribeall`, `dry`, `maxage`, `subfolder2`, `source_id`, `user1`, `password1`, `mins_interval`, `delete2duplicates`, `custom_params`, `active`, `order_id`) - VALUES (:user2, :exclude, :delete1, :delete2, :timeout1, :timeout2, :automap, :skipcrossduplicates, :maxbytespersecond, :subscribeall, :dry, :maxage, :subfolder2, :source_id, :user1, :password1, :mins_interval, :delete2duplicates, :custom_params, :active, - (SELECT COALESCE(MAX(o.`order_id`), 0) + 1 FROM (SELECT `order_id` FROM `imapsync`) o))"); + $stmt = $pdo->prepare("INSERT INTO `imapsync` (`user2`, `exclude`, `delete1`, `delete2`, `timeout1`, `timeout2`, `automap`, `skipcrossduplicates`, `maxbytespersecond`, `subscribeall`, `dry`, `maxage`, `subfolder2`, `source_id`, `user1`, `password1`, `mins_interval`, `delete2duplicates`, `custom_params`, `active`) + VALUES (:user2, :exclude, :delete1, :delete2, :timeout1, :timeout2, :automap, :skipcrossduplicates, :maxbytespersecond, :subscribeall, :dry, :maxage, :subfolder2, :source_id, :user1, :password1, :mins_interval, :delete2duplicates, :custom_params, :active)"); $stmt->execute(array( ':user2' => $username, ':custom_params' => $custom_params, @@ -787,12 +771,10 @@ function syncjob($_action, $_type, $_data = null, $_extra = null) { break; case 'edit': switch ($_type) { - case 'job_order': - // Set a syncjob's global queue position (admin-only). $_data = {id, order_id} - $id = is_array($_data) ? ($_data['id'] ?? null) : null; - if (is_array($id)) $id = reset($id); - $pos = is_array($_data) ? ($_data['order_id'] ?? null) : null; - if (!is_numeric($id) || !is_numeric($pos) || imapsync_set_order($id, $pos) === false) { + case 'run_next': + $ids = is_array($_data) ? ($_data['id'] ?? null) : null; + $moved = imapsync_run_next($ids); + if ($moved === false) { $_SESSION['return'][] = array( 'type' => 'danger', 'log' => array(__FUNCTION__, $_action, $_type, $_data_log), @@ -803,7 +785,7 @@ function syncjob($_action, $_type, $_data = null, $_extra = null) { $_SESSION['return'][] = array( 'type' => 'success', 'log' => array(__FUNCTION__, $_action, $_type, $_data_log), - 'msg' => array('imapsync_order_updated', intval($pos)) + 'msg' => array('imapsync_run_next_done', intval($moved)) ); return true; break; @@ -849,6 +831,8 @@ function syncjob($_action, $_type, $_data = null, $_extra = null) { $maxbytespersecond = (isset($_data['maxbytespersecond']) && $_data['maxbytespersecond'] != "") ? intval($_data['maxbytespersecond']) : $is_now['maxbytespersecond']; $timeout1 = (isset($_data['timeout1']) && $_data['timeout1'] != "") ? intval($_data['timeout1']) : $is_now['timeout1']; $timeout2 = (isset($_data['timeout2']) && $_data['timeout2'] != "") ? intval($_data['timeout2']) : $is_now['timeout2']; + // prio is a global-queue concern: only admins may change it, others keep the stored value + $prio = (isset($_data['prio']) && is_numeric($_data['prio']) && ($_SESSION['mailcow_cc_role'] ?? '') === 'admin') ? intval($_data['prio']) : $is_now['prio']; } else { $_SESSION['return'][] = array( @@ -936,6 +920,7 @@ function syncjob($_action, $_type, $_data = null, $_extra = null) { `timeout2` = :timeout2, `subscribeall` = :subscribeall, `dry` = :dry, + `prio` = :prio, `active` = :active WHERE `id` = :id"); $stmt->execute(array( @@ -960,6 +945,7 @@ function syncjob($_action, $_type, $_data = null, $_extra = null) { ':timeout2' => $timeout2, ':subscribeall' => $subscribeall, ':dry' => $dry, + ':prio' => $prio, ':active' => $active, )); $_SESSION['return'][] = array( @@ -1162,17 +1148,9 @@ function syncjob($_action, $_type, $_data = null, $_extra = null) { ); continue; } - // Read the position first so we can close the gap in the global queue afterwards - $ord = $pdo->prepare("SELECT `order_id` FROM `imapsync` WHERE `id` = :id"); - $ord->execute(array(':id' => $id)); - $deleted_pos = $ord->fetchColumn(); + // prio is a sparse sort key (ties allowed), so no gap needs closing on delete. $stmt = $pdo->prepare("DELETE FROM `imapsync` WHERE `id`= :id"); $stmt->execute(array(':id' => $id)); - if ($deleted_pos !== false) { - $pdo->prepare("UPDATE `imapsync` SET `order_id` = `order_id` - 1 - WHERE `order_id` > :pos ORDER BY `order_id` ASC") - ->execute(array(':pos' => intval($deleted_pos))); - } $_SESSION['return'][] = array( 'type' => 'success', 'log' => array(__FUNCTION__, $_action, $_type, $_data_log, $_attr), @@ -1289,7 +1267,7 @@ function syncjob($_action, $_type, $_data = null, $_extra = null) { else { $_data = $_SESSION['mailcow_cc_username']; } - $stmt = $pdo->prepare("SELECT `id` FROM `imapsync` WHERE `user2` = :username ORDER BY `order_id`"); + $stmt = $pdo->prepare("SELECT `id` FROM `imapsync` WHERE `user2` = :username ORDER BY `last_run` ASC, `prio` DESC, `id` ASC"); $stmt->execute(array(':username' => $_data)); $rows = $stmt->fetchAll(PDO::FETCH_ASSOC); while($row = array_shift($rows)) { diff --git a/data/web/inc/init_db.inc.php b/data/web/inc/init_db.inc.php index acf5c47b5..778716d15 100644 --- a/data/web/inc/init_db.inc.php +++ b/data/web/inc/init_db.inc.php @@ -892,7 +892,7 @@ function init_db_schema() "subscribeall" => "TINYINT(1) NOT NULL DEFAULT '1'", "dry" => "TINYINT(1) NOT NULL DEFAULT '0'", "is_running" => "TINYINT(1) NOT NULL DEFAULT '0'", - "order_id" => "INT NOT NULL DEFAULT 0", + "prio" => "INT NOT NULL DEFAULT 0", "returned_text" => "LONGTEXT", "last_run" => "TIMESTAMP NULL DEFAULT NULL", "success" => "TINYINT(1) UNSIGNED DEFAULT NULL", @@ -907,7 +907,7 @@ function init_db_schema() ), "key" => array( "idx_source_id" => array("source_id"), - "idx_order_id" => array("order_id") + "idx_prio" => array("prio") ), "fkey" => array( "fk_imapsync_source" => array( @@ -1525,13 +1525,8 @@ function init_db_schema() // Migrate webauthn tfa $stmt = $pdo->query("ALTER TABLE `tfa` MODIFY COLUMN `authmech` ENUM('yubi_otp', 'u2f', 'hotp', 'totp', 'webauthn')"); - // Syncjobs: seed global settings + one-time sequential order_id for pre-existing jobs + // Syncjobs: seed global settings $pdo->query("INSERT IGNORE INTO `imapsync_settings` (`name`, `value`) VALUES ('max_parallel', '1')"); - if (intval($pdo->query("SELECT COUNT(*) FROM `imapsync`")->fetchColumn()) > 0 - && intval($pdo->query("SELECT MAX(`order_id`) FROM `imapsync`")->fetchColumn()) === 0) { - $pdo->query("SET @r := 0"); - $pdo->query("UPDATE `imapsync` SET `order_id` = (@r := @r + 1) ORDER BY `id`"); - } // Syncjobs: migrate custom_params from raw imapsync string to structured JSON pairs. // Old values could not contain spaces (they were rejected), so a whitespace split is safe. diff --git a/data/web/js/site/mailbox.js b/data/web/js/site/mailbox.js index 494d6beda..a8ce76e67 100644 --- a/data/web/js/site/mailbox.js +++ b/data/web/js/site/mailbox.js @@ -2231,8 +2231,8 @@ jQuery(function($){ defaultContent: '' }, { - title: lang.syncjobs.order_id, - data: 'order_id', + title: lang.syncjobs.prio, + data: 'prio', responsivePriority: 3, defaultContent: '' }, diff --git a/data/web/js/site/user.js b/data/web/js/site/user.js index abc0a134b..74d915346 100644 --- a/data/web/js/site/user.js +++ b/data/web/js/site/user.js @@ -349,8 +349,8 @@ jQuery(function($){ responsivePriority: 3 }, { - title: lang.syncjobs.order_id, - data: 'order_id', + title: lang.syncjobs.prio, + data: 'prio', defaultContent: '', responsivePriority: 3 }, diff --git a/data/web/json_api.php b/data/web/json_api.php index fdf2d13c3..5e924e5f7 100644 --- a/data/web/json_api.php +++ b/data/web/json_api.php @@ -2005,8 +2005,8 @@ if (isset($_GET['query'])) { break; case "syncjob": switch ($object) { - case "order": - process_edit_return(syncjob('edit', 'job_order', array_merge(array('id' => $items), $attr))); + case "run_next": + process_edit_return(syncjob('edit', 'run_next', array('id' => $items))); break; default: process_edit_return(syncjob('edit', 'job', array_merge(array('id' => $items), $attr))); diff --git a/data/web/lang/lang.de-de.json b/data/web/lang/lang.de-de.json index f9529bbdd..d479b123b 100644 --- a/data/web/lang/lang.de-de.json +++ b/data/web/lang/lang.de-de.json @@ -1197,7 +1197,7 @@ "imapsync_source_deleted": "Sync-Quelle %s wurde gelöscht", "imapsync_source_token_refreshed": "OAuth-Token für %s wurde erneuert", "max_parallel_saved": "Sync-Job-Einstellungen gespeichert", - "imapsync_order_updated": "Sync-Job auf Position %s verschoben" + "imapsync_run_next_done": "%s Sync-Job(s) an den Anfang der Warteschlange gesetzt" }, "tfa": { "authenticators": "Authentikatoren", @@ -1300,9 +1300,8 @@ "custom_param_value": "Wert", "custom_param_no_value": "kein Wert nötig", "custom_param_add": "Parameter hinzufügen", - "order_id": "Position", - "order_id_hint": "Kleinere Position läuft zuerst; Ändern sortiert die globale Warteschlange um.", - "set_position": "Position setzen" + "prio": "Priorität", + "prio_hint": "Fällige Jobs laufen am längsten-nicht-gelaufen zuerst; bei gleichem Stand entscheidet die höhere Priorität (0 = normal)." }, "user": { "action": "Aktion", diff --git a/data/web/lang/lang.en-gb.json b/data/web/lang/lang.en-gb.json index b002d6cba..67138a02f 100644 --- a/data/web/lang/lang.en-gb.json +++ b/data/web/lang/lang.en-gb.json @@ -1204,7 +1204,7 @@ "imapsync_source_deleted": "Sync source %s has been deleted", "imapsync_source_token_refreshed": "OAuth token for %s has been refreshed", "max_parallel_saved": "Sync job settings saved", - "imapsync_order_updated": "Sync job moved to position %s" + "imapsync_run_next_done": "%s sync job(s) moved to the front of the queue" }, "tfa": { "authenticators": "Authenticators", @@ -1307,9 +1307,8 @@ "custom_param_value": "Value", "custom_param_no_value": "no value needed", "custom_param_add": "Add parameter", - "order_id": "Position", - "order_id_hint": "Lower position runs first; setting it re-sorts the global queue.", - "set_position": "Set position" + "prio": "Priority", + "prio_hint": "Due jobs run least-recently-run first; on a tie the higher priority wins (0 = normal)." }, "user": { "action": "Action", diff --git a/data/web/templates/edit/syncjob.twig b/data/web/templates/edit/syncjob.twig index 3375a5596..a1f64379b 100644 --- a/data/web/templates/edit/syncjob.twig +++ b/data/web/templates/edit/syncjob.twig @@ -48,6 +48,15 @@ 1-43800 + {% if mailcow_cc_role == 'admin' %} +
+ +
+ + {{ lang.syncjobs.prio_hint }} +
+
+ {% endif %}
@@ -160,23 +169,6 @@
- {% if mailcow_cc_role == 'admin' %} -
-
-
- -
- - {{ lang.syncjobs.order_id_hint }} -
-
-
-
- -
-
-
- {% endif %} {% else %} {{ parent() }} {% endif %} diff --git a/data/web/templates/mailbox/tab-syncjobs.twig b/data/web/templates/mailbox/tab-syncjobs.twig index 257927b1f..b7cfbb5f8 100644 --- a/data/web/templates/mailbox/tab-syncjobs.twig +++ b/data/web/templates/mailbox/tab-syncjobs.twig @@ -41,7 +41,7 @@ {{ lang.mailbox.toggle_all }} {{ lang.mailbox.quick_actions }}