[Imapsync] switch from order_id to prio

This commit is contained in:
FreddleSpl0it
2026-07-22 13:08:15 +02:00
parent 52ca8f95c2
commit 7e63061b30
11 changed files with 71 additions and 193 deletions
+8 -33
View File
@@ -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();
+6 -66
View File
@@ -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:
+31 -53
View File
@@ -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)) {
+3 -8
View File
@@ -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.
+2 -2
View File
@@ -2231,8 +2231,8 @@ jQuery(function($){
defaultContent: ''
},
{
title: lang.syncjobs.order_id,
data: 'order_id',
title: lang.syncjobs.prio,
data: 'prio',
responsivePriority: 3,
defaultContent: ''
},
+2 -2
View File
@@ -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
},
+2 -2
View File
@@ -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)));
+3 -4
View File
@@ -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",
+3 -4
View File
@@ -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",
+9 -17
View File
@@ -48,6 +48,15 @@
<small class="text-muted">1-43800</small>
</div>
</div>
{% if mailcow_cc_role == 'admin' %}
<div class="row mb-2">
<label class="control-label col-sm-2" for="prio">{{ lang.syncjobs.prio }}</label>
<div class="col-sm-10">
<input type="number" class="form-control" name="prio" id="prio" min="0" value="{{ result.prio }}">
<small class="text-muted">{{ lang.syncjobs.prio_hint }}</small>
</div>
</div>
{% endif %}
<div class="row mb-2">
<label class="control-label col-sm-2" for="subfolder2">{{ lang.syncjobs.subfolder2|raw }}</label>
<div class="col-sm-10">
@@ -160,23 +169,6 @@
</div>
</div>
</form>
{% if mailcow_cc_role == 'admin' %}
<hr>
<form class="form-horizontal" data-id="editsyncjob_order" role="form" method="post">
<div class="row mb-2">
<label class="control-label col-sm-2" for="order_id">{{ lang.syncjobs.order_id }}</label>
<div class="col-sm-10">
<input type="number" class="form-control" name="order_id" id="order_id" min="1" value="{{ result.order_id }}">
<small class="text-muted">{{ lang.syncjobs.order_id_hint }}</small>
</div>
</div>
<div class="row mb-2">
<div class="offset-sm-2 col-sm-10">
<button class="btn btn-xs-lg d-block d-sm-inline btn-primary" data-action="edit_selected" data-id="editsyncjob_order" data-item="{{ result.id }}" data-api-url='edit/syncjob/order' data-api-attr='{}' href="#">{{ lang.syncjobs.set_position }}</button>
</div>
</div>
</form>
{% endif %}
{% else %}
{{ parent() }}
{% endif %}
+2 -2
View File
@@ -41,7 +41,7 @@
<a class="btn btn-sm btn-xs-half btn-secondary" id="toggle_multi_select_all" data-id="syncjob" href="#"><i class="bi bi-check-all"></i> {{ lang.mailbox.toggle_all }}</a>
<a class="btn btn-sm btn-xs-half btn-secondary dropdown-toggle" data-bs-toggle="dropdown" href="#">{{ lang.mailbox.quick_actions }}</a>
<ul class="dropdown-menu">
<li><a class="dropdown-item" data-action="edit_selected" data-id="syncjob" data-api-url='edit/syncjob' data-api-attr='{"last_run":"","success":""}' href="#">{{ lang.mailbox.last_run_reset }}</a></li>
<li><a class="dropdown-item" data-action="edit_selected" data-id="syncjob" data-api-url='edit/syncjob/run_next' data-api-attr='{}' href="#">{{ lang.mailbox.last_run_reset }}</a></li>
<li><hr class="dropdown-divider"></li>
<li><a class="dropdown-item" data-action="edit_selected" data-id="syncjob" data-api-url='edit/syncjob' data-api-attr='{"active":"1"}' href="#">{{ lang.mailbox.activate }}</a></li>
<li><a class="dropdown-item" data-action="edit_selected" data-id="syncjob" data-api-url='edit/syncjob' data-api-attr='{"active":"0"}' href="#">{{ lang.mailbox.deactivate }}</a></li>
@@ -60,7 +60,7 @@
<a class="btn btn-sm btn-xs-lg btn-xs-half btn-secondary" id="toggle_multi_select_all" data-id="syncjob" href="#"><i class="bi bi-check-all"></i> {{ lang.mailbox.toggle_all }}</a>
<a class="btn btn-sm btn-xs-lg btn-xs-half btn-secondary dropdown-toggle" data-bs-toggle="dropdown" href="#">{{ lang.mailbox.quick_actions }}</a>
<ul class="dropdown-menu">
<li><a class="dropdown-item" data-action="edit_selected" data-id="syncjob" data-api-url='edit/syncjob' data-api-attr='{"last_run":"","success":""}' href="#">{{ lang.mailbox.last_run_reset }}</a></li>
<li><a class="dropdown-item" data-action="edit_selected" data-id="syncjob" data-api-url='edit/syncjob/run_next' data-api-attr='{}' href="#">{{ lang.mailbox.last_run_reset }}</a></li>
<li><hr class="dropdown-divider"></li>
<li><a class="dropdown-item" data-action="edit_selected" data-id="syncjob" data-api-url='edit/syncjob' data-api-attr='{"active":"1"}' href="#">{{ lang.mailbox.activate }}</a></li>
<li><a class="dropdown-item" data-action="edit_selected" data-id="syncjob" data-api-url='edit/syncjob' data-api-attr='{"active":"0"}' href="#">{{ lang.mailbox.deactivate }}</a></li>