<?php
/**
* SeekQuarry/Yioop --
* Open Source Pure PHP Search Engine, Crawler, and Indexer
*
* Copyright (C) 2009 - 2026 Chris Pollett chris@pollett.org
*
* LICENSE:
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*
* END LICENSE
*
* @author Chris Pollett chris@pollett.org
* @license https://www.gnu.org/licenses/ GPL3
* @link https://www.seekquarry.com/
* @copyright 2009 - 2026
* @filesource
*/
namespace seekquarry\yioop\library\media_jobs;
use seekquarry\yioop\configs as C;
use seekquarry\yioop\library as L;
use seekquarry\yioop\library\mail as ML;
use seekquarry\yioop\library\mail\ImapClient;
use seekquarry\yioop\library\mail\ImapMailBackend;
use seekquarry\yioop\library\mail\MailSiteFactory;
use seekquarry\yioop\library\mail\MailSiteMailBackend;
use seekquarry\yioop\models\MailAccountModel;
use seekquarry\yioop\models\MailCloneModel;
/**
* MediaJob that clones an external IMAP account into the user's
* local MailSite mailbox. Driven by MAIL_CLONE_JOB rows the user
* submits through the Edit Account UI; the job picks up
* pending/running rows on each MediaUpdater tick, walks the
* source IMAP server folder by folder, message by message, and
* appends each unseen message to the destination user's MailSite
* storage via MailSiteMailBackend::appendMessage. Per-message
* deduplication uses the (UIDVALIDITY, UID) pair stored in
* MAIL_CLONE_SEEN so a tick crash, deliberate cancel, or
* intentional re-run is idempotent. Tick work is bounded by
* MAX_TICK_SECONDS so MediaUpdater can interleave other jobs;
* unfinished work resumes from CURRENT_FOLDER, CURRENT_UID, and
* (for very large messages) CURRENT_UID_PARTIAL_OFFSET on the
* next tick.
*
* Runs in nondistributed mode only (name_server_does_client_tasks_only
* = true) for the same reason PodcastDownloadJob does: the work
* is IO against the local DB and an upstream IMAP server, neither
* of which benefits from sharding across MediaUpdater clients.
*
* Large-message ratchet: messages above LARGE_MESSAGE_THRESHOLD_BYTES
* (2 MB) are fetched in PARTIAL_FETCH_CHUNK_BYTES (1 MB) chunks
* via BODY.PEEK[<offset.size>], with each chunk appended to a
* staging file under WORK_DIRECTORY/clone-staging/<job_id>/<uid>.part.
* When the running offset equals the announced RFC822.SIZE the
* staging file is read in full and handed to appendMessage, then
* deleted. A mid-ratchet crash leaves the staging file behind;
* the next tick reads its current size from disk and resumes
* from that byte. This keeps any single FETCH inside the
* MAX_TICK_SECONDS budget even for multi-megabyte messages.
*
* Wipe semantics: when MODE='wipe' on a row transitioning from
* pending to running, the destination user's MailSite mailbox is
* fully purged (every folder, every message) before any clone
* append happens. The purge is scoped strictly to the
* DESTINATION_USER named in the job row; no other user's mail
* is affected.
*
* @author Chris Pollett
*/
class MailCloneJob extends MediaJob
{
/**
* Maximum wall-clock seconds a single nondistributedTasks
* invocation may spend before yielding back to the
* MediaUpdater. Sized so other media jobs (podcasts, group
* feeds) keep getting their turn even when a clone has
* thousands of messages to process.
*/
const MAX_TICK_SECONDS = 60;
/**
* Messages whose RFC822.SIZE is at or above this threshold
* (2 MB) are fetched in partial chunks rather than in one
* FETCH so any single tick stays bounded. Below the
* threshold messages are fetched whole, which is the
* common case (typical inbox averages 20-100 KB / message).
*/
const LARGE_MESSAGE_THRESHOLD_BYTES = 2097152;
/**
* Chunk size (1 MB) for the partial-fetch ratchet on large
* messages. Each tick reads at most one chunk per active
* large message and appends it to the staging file. Sized
* to fit inside the 60-second tick budget even on slow
* upstream IMAP servers and to keep the per-tick memory
* footprint predictable.
*/
const PARTIAL_FETCH_CHUNK_BYTES = 1048576;
/**
* How many UIDs to FETCH per pipelined IMAP round-trip.
* The whole batch is read into memory at once (each
* message's BODY.PEEK[] arrives as a literal in the same
* response), so the upper bound is set to keep average
* batch payload under a few tens of MB: 20 messages *
* <2MB-large-threshold = <40 MB worst case for a
* full-of-edge-case batch, well under any reasonable PHP
* memory limit. With 350ms-per-RTT, 20-per-batch reduces
* latency cost from 350ms-per-message to 17.5ms-per-
* message; further increases are dominated by body-bytes
* transit time, not RTT.
*/
const FETCH_BATCH_SIZE = 20;
/**
* Minimum seconds between checkPrerequisites returning true.
* Without this throttle a no-work tick still pays the cost
* of MediaUpdater dispatch + DB connection; with it, the
* job is effectively idle when no clone is pending.
*/
const TICK_INTERVAL_SECONDS = 5;
/**
* Database handle, opened once in init() and held for the
* job lifetime so per-tick reconnects are avoided.
* @var object
*/
public $db;
/**
* MailCloneModel instance, scoped to $this->db.
* @var MailCloneModel
*/
public $clone_model;
/**
* MailAccountModel instance, used to load source-account
* rows with their decrypted passwords for IMAP login.
* @var MailAccountModel
*/
public $account_model;
/**
* Wall-clock time of last successful tick; used by
* checkPrerequisites to enforce TICK_INTERVAL_SECONDS.
* @var int
*/
public $last_tick_time;
/**
* Wall-clock deadline (microtime) the current tick must
* yield by. Recomputed at the start of each
* nondistributedTasks invocation; checked between message
* fetches to honor MAX_TICK_SECONDS.
* @var float
*/
public $tick_deadline;
/**
* Cached MailSiteMailBackend for the current job's
* destination user. Built once per advanceJob call and
* reused across folder/message iterations, instead of
* rebuilding MailSiteFactory + FileMailStorage on every
* appended message. Cleared at the start of each
* advanceJob.
* @var MailSiteMailBackend|null
*/
public $destination_backend;
/**
* Number of messages imported in the current
* advanceFolder iteration that have not yet been flushed
* to MAIL_CLONE_JOB.IMPORTED via updateProgress. Batched
* to PROGRESS_FLUSH_EVERY messages so we do one DB UPDATE
* per N appends instead of one per append. Status-banner
* polling is 5-second granular so the user does not see
* the batching delay.
* @var int
*/
public $pending_imported_delta = 0;
/**
* Number of messages skipped in the current advanceFolder
* iteration -- because an identical message (same Message-ID)
* already exists in the destination folder -- that have not
* yet been flushed to MAIL_CLONE_JOB.SKIPPED. Batched the
* same way as pending_imported_delta so duplicate-heavy
* re-clones do one DB UPDATE per N skips rather than one per
* skip.
* @var int
*/
public $pending_skipped_delta = 0;
/**
* Set of Message-IDs already present in the destination
* folder named by destination_message_id_folder, used to
* skip re-importing a message that is already in the local
* mailbox (e.g. from a prior clone). Keys are normalized
* Message-IDs, values are true. Built lazily once per folder
* per run and reused for every message in that folder.
* @var array
*/
public $destination_message_ids = [];
/**
* Destination folder name that destination_message_ids was
* built for, or null when no set has been built yet. When
* the clone moves to a different folder the set is rebuilt
* for that folder.
* @var string|null
*/
public $destination_message_id_folder = null;
/**
* Number of messages flushed since the last log line.
* Used by advanceFolder to emit a one-line summary every
* PROGRESS_LOG_EVERY messages: msgs/sec and IMAP-fetch
* mean latency. The log line is the primary on-server
* instrumentation for diagnosing clone-speed problems.
* @var int
*/
public $messages_since_log = 0;
/**
* Sum (in microseconds) of round-trip times for the IMAP
* FETCHes that produced messages_since_log. Reset to 0 on
* every log emit. Mean of (fetch_us / messages_since_log)
* is the WAN-latency component; subtracting that from the
* overall msg/sec interval reveals local-DB-and-disk cost.
* @var float
*/
public $fetch_us_since_log = 0.0;
/**
* Wall clock at the start of the current messages_since_log
* window, used to compute msgs/sec at log-emit time.
* @var float
*/
public $log_window_started = 0.0;
/**
* How many messages to import before flushing IMPORTED to
* MAIL_CLONE_JOB via updateProgress. Higher batches save
* DB writes; lower batches keep the status banner more
* current. 10 means at most a 10-message lag visible to
* the polling banner.
*/
const PROGRESS_FLUSH_EVERY = 10;
/**
* How many messages between log lines that show msg/sec
* and mean IMAP-fetch latency. Tuned so a small mailbox
* (~50 messages) emits a couple of lines and a big one
* (10k+ messages) does not flood the log.
*/
const PROGRESS_LOG_EVERY = 100;
/**
* Initializes the job. Sets nondistributed-only flags (the
* job's work is local DB plus a single upstream IMAP server
* per row, neither of which shards across MediaUpdater
* clients), opens the configured DB, and instantiates the
* two models we drive. last_tick_time is set to 0 so the
* first MediaUpdater pass after a (re)start runs immediately
* rather than waiting TICK_INTERVAL_SECONDS.
*/
public function init()
{
$this->name_server_does_client_tasks = false;
$this->name_server_does_client_tasks_only = true;
$this->last_tick_time = 0;
/* Let the Model base constructor wire BOTH the public
and the private DB connections (the private one is
where MAIL_SECRET -- the libsodium master key for
credential decryption -- lives). Sharing a single
pre-built public-DB handle across both models would
save one connection but skips the private-DB setup
entirely and breaks CredentialCipher::decrypt at
runtime; the cost of one extra connect per tick is
negligible against the 60-second tick budget. */
$this->clone_model = new MailCloneModel(C\p('DB_NAME'), true);
$this->account_model = new MailAccountModel(C\p('DB_NAME'),
true);
$this->db = $this->clone_model->db;
}
/**
* Throttles to one tick per TICK_INTERVAL_SECONDS and skips
* empty-queue ticks fast. Returning false here is the
* common case (no clone jobs in flight), so the cost of
* this check is the hot path.
*
* @return bool whether the tick body should run
*/
public function checkPrerequisites()
{
$now = time();
if ($now - $this->last_tick_time <
self::TICK_INTERVAL_SECONDS) {
return false;
}
$claimable = $this->clone_model->listClaimable();
if (empty($claimable)) {
$this->last_tick_time = $now;
return false;
}
$this->last_tick_time = $now;
L\crawlLog("MailCloneJob: " . count($claimable) .
" clone job(s) to advance");
return true;
}
/**
* Per-tick work loop. Iterates the claimable jobs in
* STARTED_AT order so older submissions drain before newer
* ones, advancing each as far as the per-tick budget
* allows. Each job's work is wrapped in try/catch so a
* fatal error on one user's clone does not stall every
* other user's. Catches at the outer boundary record a
* job-level error row and mark the job failed for UI surface.
*/
public function nondistributedTasks()
{
$this->tick_deadline = microtime(true) +
self::MAX_TICK_SECONDS;
$this->clone_model->purgeOldSeen();
$jobs = $this->clone_model->listClaimable();
/* Drop staging directories belonging to jobs that are
no longer live (done, cancelled, failed, or deleted).
The live set is exactly the claimable jobs, since
listClaimable returns pending + running rows. */
$live_job_ids = [];
foreach ($jobs as $job) {
$live_job_ids[(int) $job['ID']] = true;
}
$this->sweepOrphanStaging($live_job_ids);
foreach ($jobs as $job) {
if (microtime(true) >= $this->tick_deadline) {
L\crawlLog("MailCloneJob: tick deadline " .
"reached, yielding");
return;
}
try {
$this->advanceJob($job);
} catch (\Exception $exception) {
if ($exception instanceof ML\MailBackendException &&
$exception->isTransient()) {
/* A transient connection-level failure (socket
dropped mid-response, server hung up, timed
out). The clone is idempotent -- progress is
checkpointed per folder/UID and already-seen
UIDs are skipped via MAIL_CLONE_SEEN -- so
leave STATUS at 'running', record the blip so
it is visible, and end the tick. listClaimable
re-claims running jobs, so the next
MediaUpdater pass resumes where this left off
rather than abandoning thousands of already-
imported messages. Staging is left in place so
an in-progress large-message ratchet can
continue. */
L\crawlLog("MailCloneJob: job " . $job['ID'] .
" transient connection error; will resume " .
"next tick: " . $exception->getMessage());
$this->clone_model->recordError(
(int) $job['ID'], '', 0, 0,
'transient_connection',
$exception->getMessage());
return;
}
L\crawlLog("MailCloneJob: job " . $job['ID'] .
" failed: " . $exception->getMessage());
$this->clone_model->recordError(
(int) $job['ID'], '', 0, 0, 'job_exception',
$exception->getMessage());
$this->clone_model->updateProgress($job['ID'], [
'FAILED' => $this->clone_model
->unresolvedErrorCount(
(int) $job['ID']),
]);
$this->clone_model->setStatus($job['ID'],
'failed');
$this->purgeStagingForJob((int) $job['ID']);
}
}
}
/**
* Advances one clone job by as much work as fits in the
* remaining tick budget. The job row passed in is a
* snapshot; before each significant phase (wipe, folder
* loop entry) we re-fetch to honor any STATUS='cancelled'
* transition the user may have requested through the UI
* since this tick started.
*
* @param array $job a row from listClaimable
*/
protected function advanceJob($job)
{
$job_id = (int) $job['ID'];
$current = $this->clone_model->getJob($job_id,
(int) $job['USER_ID']);
if (!$current || $current['STATUS'] === 'cancelled') {
/* A cancelled job will not advance again, so drop any
partial large-message downloads it left staged. */
$this->purgeStagingForJob($job_id);
return;
}
$account = $this->account_model->getAccountWithPassword(
(int) $job['SRC_ACCOUNT'], (int) $job['USER_ID']);
if (!$account) {
throw new \Exception(
"source account " . $job['SRC_ACCOUNT'] .
" missing");
}
$backend = new ImapMailBackend($account,
new MailCloneLogger());
/* Best-effort second IMAP connection for read-ahead
pipelining: while the main connection is draining
one FETCH response, this shadow connection holds
the next batch's FETCH in flight, so the upstream
server is preparing batch N+1 while we are still
receiving batch N. Doubles wall-clock throughput
on links where the server-side per-batch latency
is the dominant cost.
Best-effort because some IMAP servers cap per-user
connection counts; on a rejected second LOGIN we
swallow the exception and run with a single
connection (degraded but correct). */
$prefetch_backend = null;
try {
$prefetch_backend = new ImapMailBackend($account,
new MailCloneLogger());
/* Force-open the connection now so we either
succeed (and use it) or fail (and discard it)
before any FETCH work starts. The first call
on a fresh backend opens the socket and runs
LOGIN. */
$prefetch_backend->imapClient();
} catch (\Exception $exception) {
L\crawlLog("MailCloneJob: prefetch connection " .
"unavailable (" . $exception->getMessage() .
"); falling back to single-connection mode");
$prefetch_backend = null;
}
/* Build the MailSite destination backend once per
advanceJob call instead of per-message: each
new MailSiteMailBackend pays a MailSiteFactory::build
plus a fresh FileMailStorage open, which on a deep
clone (thousands of messages) adds up. Dry-run jobs
skip the build entirely since deliverToDestination
returns early. */
$this->destination_backend = empty($current['DRY_RUN']) ?
new MailSiteMailBackend(
$current['DESTINATION_USER']) : null;
$this->pending_imported_delta = 0;
$this->pending_skipped_delta = 0;
$this->destination_message_ids = [];
$this->destination_message_id_folder = null;
$this->messages_since_log = 0;
$this->fetch_us_since_log = 0.0;
$this->log_window_started = microtime(true);
try {
if ($current['STATUS'] === 'pending') {
if ($current['MODE'] === 'wipe') {
$this->wipeDestination(
$current['DESTINATION_USER']);
}
$this->clone_model->setStatus($job_id,
'running');
$current['STATUS'] = 'running';
}
$folders = $backend->listFolders();
$done = json_decode($current['FOLDERS_DONE'], true);
if (!is_array($done)) {
$done = [];
}
foreach ($folders as $folder_row) {
if (microtime(true) >= $this->tick_deadline) {
return;
}
$folder_name = $folder_row['NAME'];
if (in_array($folder_name, $done, true)) {
continue;
}
$resume_uid = ($current['CURRENT_FOLDER'] ===
$folder_name) ?
(int) $current['CURRENT_UID'] : 0;
$this->advanceFolder($job_id, $current, $backend,
$prefetch_backend, $folder_name, $resume_uid);
$current = $this->clone_model->getJob($job_id,
(int) $job['USER_ID']);
if (!$current ||
$current['STATUS'] === 'cancelled') {
if ($current &&
$current['STATUS'] === 'cancelled') {
$this->purgeStagingForJob($job_id);
}
return;
}
if (microtime(true) >= $this->tick_deadline) {
return;
}
$done[] = $folder_name;
$this->clone_model->updateProgress($job_id, [
'FOLDERS_DONE' => json_encode($done),
'CURRENT_FOLDER' => '',
'CURRENT_UID' => 0,
'CURRENT_UID_PARTIAL_OFFSET' => 0,
'CURRENT_UID_PARTIAL_SIZE' => 0,
]);
}
$this->clone_model->setStatus($job_id, 'done');
$this->purgeStagingForJob($job_id);
} finally {
/* Flush any pending IMPORTED counter delta so the
status banner sees the final count even when the
tick ends mid-batch. */
$this->flushImportedDelta($job_id);
$backend->close();
if ($prefetch_backend !== null) {
$prefetch_backend->close();
}
$this->destination_backend = null;
}
}
/**
* Walks one source folder from $resume_uid up to the
* current UIDNEXT, fetching and appending unseen messages
* one by one. SELECT is issued unconditionally so a fresh
* UIDVALIDITY value is captured for the dedup key even on
* a resumed folder. Honors the tick deadline between
* messages and between batches.
*
* @param int $job_id MAIL_CLONE_JOB.ID being advanced
* @param array $job_row current job row (FOLDERS_DONE etc)
* @param ImapMailBackend $backend open IMAP backend
* @param string $folder_name source folder to traverse
* @param int $resume_uid skip UIDs at or below this value
* (resume cursor from a prior tick)
*/
protected function advanceFolder($job_id, $job_row, $backend,
$prefetch_backend, $folder_name, $resume_uid)
{
$client = $backend->imapClient();
$select = $client->send('SELECT "' .
ImapMailBackend::imapQuote($folder_name) . '"');
if ($select['status'] !== 'OK') {
throw new \Exception(
"SELECT $folder_name failed: " .
$select['detail']);
}
/* Register the folder with the backend so its reconnect
handler can re-SELECT it if the socket drops during a
later raw FETCH (the large-message ratchet issues
folder-scoped UID FETCH directly on the client without
going back through selectFolder). */
$backend->noteSelectedFolder($folder_name);
$uidvalidity = $this->parseUidValidity(
$select['untagged']);
/* Mirror the SELECT on the prefetch connection so it
is also positioned in the same folder before we
start pipelining UID FETCH on it. A SELECT failure
on the prefetch path is non-fatal: we drop the
prefetch and run single-connection on this folder. */
if ($prefetch_backend !== null) {
try {
$prefetch_client =
$prefetch_backend->imapClient();
$prefetch_select = $prefetch_client->send(
'SELECT "' .
ImapMailBackend::imapQuote($folder_name) .
'"');
if ($prefetch_select['status'] !== 'OK') {
L\crawlLog("MailCloneJob: prefetch SELECT " .
"$folder_name failed (" .
$prefetch_select['detail'] . "); " .
"falling back to single connection " .
"for this folder");
$prefetch_backend = null;
} else {
$prefetch_backend->noteSelectedFolder(
$folder_name);
}
} catch (\Exception $exception) {
L\crawlLog("MailCloneJob: prefetch SELECT " .
"exception (" . $exception->getMessage() .
"); falling back to single connection for " .
"this folder");
$prefetch_backend = null;
}
}
$this->clone_model->updateProgress($job_id, [
'CURRENT_FOLDER' => $folder_name,
]);
if ((int) $job_row['CURRENT_UID_PARTIAL_OFFSET'] > 0 &&
$job_row['CURRENT_FOLDER'] === $folder_name) {
$partial_uid = (int) $job_row['CURRENT_UID'];
$finished = $this->ratchetLargeMessage($job_id,
$job_row, $backend, $folder_name, $uidvalidity,
$partial_uid);
if (!$finished) {
return;
}
$resume_uid = $partial_uid;
}
$start = $resume_uid + 1;
$search = $client->send("UID SEARCH UID $start:*");
if ($search['status'] !== 'OK') {
throw new \Exception(
"UID SEARCH failed: " . $search['detail']);
}
$uids = $this->parseSearchUids($search['untagged']);
sort($uids);
$uids = array_values(array_filter($uids,
fn ($uid) => $uid >= $start));
/* Newest-N-per-folder cap: when FOLDER_CAP is set, keep
only the highest (newest) cap UIDs in this folder.
IMAP UIDs ascend with arrival, so the cap is a fixed
floor derived from the folder's current maximum UID --
which makes it resume-safe across ticks regardless of
how the walk is chunked, rather than slicing the
per-tick window (which would drift as the walk
advances). A folder with fewer than cap messages is
unaffected. */
$folder_cap = (int) ($job_row['FOLDER_CAP'] ?? 0);
if ($folder_cap > 0 && count($uids) > 0) {
$max_uid = max($uids);
$floor = $max_uid - $folder_cap + 1;
if ($floor > $start) {
$uids = array_values(array_filter($uids,
fn ($uid) => $uid >= $floor));
}
}
$batches = array_chunk($uids, self::FETCH_BATCH_SIZE);
if ($prefetch_backend !== null) {
$this->advanceFolderPipelined($job_id, $backend,
$prefetch_backend, $folder_name, $uidvalidity,
$batches);
} else {
foreach ($batches as $batch) {
if (microtime(true) >= $this->tick_deadline) {
return;
}
$this->advanceBatch($job_id, $backend,
$folder_name, $uidvalidity, $batch);
}
}
}
/**
* Pipelined batch loop using two IMAP connections. While
* the main client is reading batch N's response and the
* caller is writing batch N's bodies to disk + committing
* the dedup transaction, the upstream server is already
* working on batch N+1 because we issued its FETCH on the
* prefetch connection before draining batch N. Net effect
* is that wall-clock time per batch becomes
* max(server-fetch-time, local-disk+DB-time) instead of
* their sum.
*
* Connections alternate ownership of the "current" and
* "next" FETCH; on the first iteration the prefetch
* connection holds the only in-flight FETCH and the main
* connection idles. On the last iteration the prefetch
* connection has nothing queued.
*
* Falls back to single-connection mode if any FETCH issue
* fails partway through; that branch surfaces the failure
* as a per-batch processing error via advanceBatch.
*
* @param int $job_id MAIL_CLONE_JOB.ID being advanced
* @param object $backend main ImapMailBackend (selected)
* @param object $prefetch_backend shadow ImapMailBackend
* (selected) used for read-ahead
* @param string $folder_name source folder
* @param int $uidvalidity source folder's UIDVALIDITY
* @param array $batches list of UID arrays, one per batch
*/
protected function advanceFolderPipelined($job_id,
$backend, $prefetch_backend, $folder_name,
$uidvalidity, $batches)
{
$count = count($batches);
if ($count === 0) {
return;
}
/* Track the two backends, not just their cached clients,
because the ratchet path inside processFetchedBatch
needs a backend whose IMAP socket has no in-flight
command pending. Issuing a synchronous FETCH on a
client that already has an in-flight FETCH would
interleave responses and the demultiplex would
desynchronize, producing the "No mailbox selected"
errors seen on large-message ratchet hops.
We track which backend is "pending" (has an in-flight
UID FETCH whose response we have not yet collected)
and which is "idle" (no in-flight command). On each
iteration we pre-fire the next FETCH on the idle
backend, drain the pending one, and then process the
drained response using the now-pending backend's
OTHER end -- the just-drained backend, which is once
again idle and safe for synchronous follow-ups like
ratchetLargeMessage. */
$pending_backend = $prefetch_backend;
$idle_backend = $backend;
$first_filtered = $this->filterUnseenForBatch($job_id,
$uidvalidity, $batches[0]);
$pending_tag = null;
$pending_fetch_started = 0;
if (!empty($first_filtered)) {
$pending_tag = $pending_backend->imapClient()
->issueCommand("UID FETCH " .
implode(',', $first_filtered) .
" (UID RFC822.SIZE FLAGS INTERNALDATE " .
"BODY.PEEK[])");
$pending_fetch_started = microtime(true);
if ($pending_tag === null) {
$this->advanceBatchSingleConnection($job_id,
$backend, $folder_name, $uidvalidity,
$batches);
return;
}
}
$pending_filtered = $first_filtered;
for ($i = 0; $i < $count; $i++) {
if (microtime(true) >= $this->tick_deadline) {
return;
}
/* Pre-fire the next batch on the idle backend (if
any) before reading the pending response, so the
upstream server is already preparing the next
result while we drain this one. */
$next_tag = null;
$next_filtered = [];
$next_fetch_started = 0;
if ($i + 1 < $count) {
$next_filtered = $this->filterUnseenForBatch(
$job_id, $uidvalidity, $batches[$i + 1]);
if (!empty($next_filtered)) {
$next_tag = $idle_backend->imapClient()
->issueCommand("UID FETCH " .
implode(',', $next_filtered) .
" (UID RFC822.SIZE FLAGS INTERNALDATE" .
" BODY.PEEK[])");
$next_fetch_started = microtime(true);
}
}
/* Drain the pending response (if any). After the
drain, the pending backend's socket is idle; we
hand it to processFetchedBatch as the
safe-for-synchronous-use backend so the ratchet
path can issue UID FETCH on it without colliding
with the next-batch FETCH that is now in flight
on the OTHER backend. */
if ($pending_tag !== null) {
/* collectResponse can throw if the source server
drops the connection mid-response on a corrupt
message. Rather than let that unwind the whole
job (which would resume into the same poison
batch and drop again), fall back to the
single-connection advanceBatch for just this
batch: that path fetches each UID on its own and
records-and-skips the specific bad one, so the
pipeline makes forward progress. The prefetch
FETCH already issued on the other backend for the
next batch is abandoned harmlessly -- its UIDs
are re-filtered against MAIL_CLONE_SEEN and
re-fetched on the following iteration. */
try {
$response = $pending_backend->imapClient()
->collectResponse($pending_tag);
$fetch_elapsed_us = (microtime(true) -
$pending_fetch_started) * 1000000;
$this->processFetchedBatch($job_id,
$pending_backend, $folder_name,
$uidvalidity, $pending_filtered,
$response, $fetch_elapsed_us);
} catch (\Exception $drain_exception) {
L\crawlLog("MailCloneJob: pipelined drain " .
"failed (" .
$drain_exception->getMessage() .
"); falling back to per-UID fetch for " .
"this batch.");
$this->advanceBatch($job_id, $pending_backend,
$folder_name, $uidvalidity,
$pending_filtered);
}
}
/* Swap roles: the backend we just drained becomes
the idle one available for the next-next pre-fire;
the backend that now holds the just-issued FETCH
becomes pending. */
$temp = $pending_backend;
$pending_backend = $idle_backend;
$idle_backend = $temp;
$pending_tag = $next_tag;
$pending_filtered = $next_filtered;
$pending_fetch_started = $next_fetch_started;
if ($this->messages_since_log >=
self::PROGRESS_LOG_EVERY) {
$this->logProgressWindow($job_id, $folder_name);
}
}
}
/**
* Single-connection fallback used when an issueCommand on
* the prefetch path fails partway through. Walks the rest
* of the batches via the existing advanceBatch path.
*
* @param int $job_id job being advanced
* @param object $backend main backend (selected)
* @param string $folder_name source folder
* @param int $uidvalidity source folder's UIDVALIDITY
* @param array $batches batches to process
*/
protected function advanceBatchSingleConnection($job_id,
$backend, $folder_name, $uidvalidity, $batches)
{
foreach ($batches as $batch) {
if (microtime(true) >= $this->tick_deadline) {
return;
}
$this->advanceBatch($job_id, $backend, $folder_name,
$uidvalidity, $batch);
}
}
/**
* Pre-filters a batch's UIDs against MAIL_CLONE_SEEN,
* ratcheting CURRENT_UID past any already-imported UIDs
* so a tick-resume skips them. Returns the list of UIDs
* that still need fetching, in input order.
*
* Used by the pipelined batch loop to decide what UIDs
* to include in the next UID FETCH command. An empty
* return means the batch is fully cached and the caller
* skips issuing a FETCH for it.
*
* @param int $job_id job whose dedup state to consult
* @param int $uidvalidity source folder's UIDVALIDITY
* @param array $uids candidate UIDs from this batch
* @return array UIDs that need fetching
*/
protected function filterUnseenForBatch($job_id,
$uidvalidity, $uids)
{
$needed = [];
foreach ($uids as $uid) {
if ($this->clone_model->isSeen($job_id,
$uidvalidity, $uid)) {
$this->clone_model->updateProgress($job_id, [
'CURRENT_UID' => $uid,
]);
continue;
}
$needed[] = $uid;
}
return $needed;
}
/**
* Processes a pipelined FETCH response: demultiplexes
* per-UID, dispatches large messages to the ratchet,
* appends normal messages to the destination backend,
* and commits the per-batch DB state via flushBatch.
*
* Mirrors the steady-state body of advanceBatch but
* takes the response as a parameter instead of issuing
* the FETCH itself. The pipelined caller has already
* paid the IMAP wait and already started the NEXT
* batch's fetch on the other connection.
*
* @param int $job_id job being advanced
* @param object $backend main backend (for large-message
* ratchet which needs its own FETCH)
* @param string $folder_name source folder
* @param int $uidvalidity source folder's UIDVALIDITY
* @param array $needed list of UIDs this batch asked for
* @param array $response collectResponse() result
* @param float $fetch_elapsed_us batch wall-clock fetch
* duration in microseconds (for amortized per-msg
* latency reporting)
*/
protected function processFetchedBatch($job_id, $backend,
$folder_name, $uidvalidity, $needed, $response,
$fetch_elapsed_us)
{
if ($response['status'] !== 'OK') {
/* Batch-level failure on the pipelined path: record
a per-UID error against every UID we asked for so
the status page can surface which specific
messages were affected. Do not advance the dedup
state for any of them; the next tick's pipelined
retry will re-fetch them, and if it succeeds
flushBatch will mark these errors resolved
(via='retry_succeeded'). */
$message = "pipelined UID FETCH failed: " .
$response['detail'];
foreach ($needed as $uid) {
$this->clone_model->recordError($job_id,
$folder_name, $uidvalidity, $uid,
'pipelined_fetch_eof', $message);
}
/* Refresh the FAILED counter so the status page
sees the new errors right away. */
$this->clone_model->updateProgress($job_id, [
'FAILED' => $this->clone_model
->unresolvedErrorCount($job_id),
]);
return;
}
$by_uid = $this->parseFetchedBatch(
$response['untagged'], $response['literals']);
$per_msg_fetch_us = (count($by_uid) > 0) ?
($fetch_elapsed_us / count($by_uid)) : 0;
$seen_tuples = [];
$last_uid = 0;
$batch_imported = 0;
foreach ($needed as $uid) {
if (microtime(true) >= $this->tick_deadline) {
break;
}
$last_uid = $uid;
if (!isset($by_uid[$uid])) {
$seen_tuples[] = [$uidvalidity, $uid];
continue;
}
$entry = $by_uid[$uid];
$size = $entry['size'];
if ($size >= self::LARGE_MESSAGE_THRESHOLD_BYTES) {
$this->flushBatch($job_id, $seen_tuples,
$batch_imported, $last_uid);
$seen_tuples = [];
$batch_imported = 0;
$this->clone_model->updateProgress($job_id, [
'CURRENT_UID' => $uid,
'CURRENT_UID_PARTIAL_OFFSET' => 0,
'CURRENT_UID_PARTIAL_SIZE' => $size,
]);
$job_row = $this->clone_model->getJob($job_id,
$this->jobUserId($job_id));
$this->ratchetLargeMessage($job_id, $job_row,
$backend, $folder_name, $uidvalidity,
$uid);
continue;
}
$bytes = $entry['body'];
if ($bytes === null || $bytes === '') {
$this->clone_model->recordError($job_id,
$folder_name, $uidvalidity, $uid,
'empty_body',
"UID $uid: server returned empty body");
$seen_tuples[] = [$uidvalidity, $uid];
continue;
}
$delivered = $this->deliverToDestinationCached(
$folder_name, $bytes, $entry['flags'],
$entry['internal_date']);
$seen_tuples[] = [$uidvalidity, $uid];
if ($delivered) {
$batch_imported++;
}
$this->messages_since_log++;
$this->fetch_us_since_log += $per_msg_fetch_us;
}
$this->flushBatch($job_id, $seen_tuples,
$batch_imported, $last_uid);
}
/**
* Fetches a batch of normal-sized UIDs in one round-trip
* via FETCH (UID RFC822.SIZE FLAGS INTERNALDATE BODY.PEEK[]).
* Routes any UID whose size meets the large-message
* threshold to the per-UID ratchet path so the batch
* fetch stays bounded.
*
* @param int $job_id job being advanced
* @param ImapMailBackend $backend open backend
* @param string $folder_name source folder
* @param int $uidvalidity source folder's UIDVALIDITY
* @param array $uids list of source UIDs in this batch
*/
protected function advanceBatch($job_id, $backend,
$folder_name, $uidvalidity, $uids)
{
if (empty($uids)) {
return;
}
/* Filter out already-seen UIDs in a single pass: those
cost zero IMAP latency, just a DB lookup. Each one
still ratchets the resume cursor so a tick-resume
skips past them next time. The pre-filter shrinks the
pipelined FETCH set to only the UIDs we actually need
to read from the server. */
$needed = [];
foreach ($uids as $uid) {
if ($this->clone_model->isSeen($job_id,
$uidvalidity, $uid)) {
$this->clone_model->updateProgress($job_id, [
'CURRENT_UID' => $uid,
]);
continue;
}
$needed[] = $uid;
}
if (empty($needed)) {
return;
}
if (microtime(true) >= $this->tick_deadline) {
return;
}
$client = $backend->imapClient();
/* One pipelined IMAP round-trip for the whole batch
instead of one per message. The dominant cost on a
remote IMAP server is round-trip latency
(~350ms-per-RTT typical, measured via the mean IMAP
fetch log); pipelining batch_size messages into one
FETCH means we pay one RTT plus body-bytes transit
time instead of batch_size full RTTs. The server
emits one untagged FETCH response per UID, in order;
ImapClient::send aggregates them in $response['untagged']
and matches each literal to its untagged-line index. */
$uid_set = implode(',', $needed);
$fetch_started = microtime(true);
$combined = $client->send("UID FETCH $uid_set " .
"(UID RFC822.SIZE FLAGS INTERNALDATE BODY.PEEK[])");
$fetch_elapsed_us = (microtime(true) - $fetch_started)
* 1000000;
if ($combined['status'] !== 'OK') {
/* Batch-level failure: fall back to per-UID fetch so
a single corrupt message does not block the whole
folder. The per-UID path through advanceMessage
records the specific bad UID in MAIL_CLONE_ERROR.
advanceMessage can also *throw* rather than return a
non-OK status when the source server drops the
connection mid-response on a corrupt message (the
classic "FETCH failed, server hung up" case): catch
that here so one poison UID cannot stall the job
forever. A transient connection-level throw means
this UID is what broke the link, so record it, mark
it seen (the resumed tick skips past it via
MAIL_CLONE_SEEN), then re-throw to let the job yield
and reconnect cleanly. Any other per-message throw is
recorded and skipped, and the loop moves on to the
next UID. */
foreach ($needed as $uid) {
if (microtime(true) >= $this->tick_deadline) {
return;
}
try {
$this->advanceMessage($job_id, $backend,
$folder_name, $uidvalidity, $uid);
} catch (\Exception $message_exception) {
$this->clone_model->recordError($job_id,
$folder_name, $uidvalidity, $uid,
'message_skipped',
"UID $uid skipped after error: " .
$message_exception->getMessage());
$this->clone_model->markSeen($job_id,
$uidvalidity, $uid);
$this->clone_model->updateProgress($job_id, [
'FAILED' => $this->clone_model
->unresolvedErrorCount($job_id),
'CURRENT_UID' => $uid,
]);
if ($message_exception instanceof
ML\MailBackendException &&
$message_exception->isTransient()) {
throw $message_exception;
}
}
}
return;
}
$by_uid = $this->parseFetchedBatch(
$combined['untagged'], $combined['literals']);
/* Spread the batch-RTT cost evenly across the messages
we successfully fetched, so the per-message latency
reported by logProgressWindow reflects the
amortized cost. Pre-divide once. */
$per_msg_fetch_us = (count($by_uid) > 0) ?
($fetch_elapsed_us / count($by_uid)) : 0;
/* Collect MAIL_CLONE_SEEN inserts so the entire batch
flushes in a single transaction at the end (one
fsync on default-mode SQLite, one round-trip on
MySQL/Postgres). The previous per-message markSeen
was the largest single per-message local cost. */
$seen_tuples = [];
$last_uid = 0;
$batch_imported = 0;
foreach ($needed as $uid) {
if (microtime(true) >= $this->tick_deadline) {
break;
}
$last_uid = $uid;
if (!isset($by_uid[$uid])) {
/* Server omitted this UID from the response;
could be a server-side EXPUNGE between SEARCH
and FETCH. Treat as seen-and-skipped so we
don't retry forever. */
$seen_tuples[] = [$uidvalidity, $uid];
continue;
}
$entry = $by_uid[$uid];
$size = $entry['size'];
if ($size >= self::LARGE_MESSAGE_THRESHOLD_BYTES) {
/* Discard the batch-fetched body for this UID
and re-fetch in chunks through the ratchet
path: the combined fetch is wasteful for large
messages but the batched savings on the many
small messages make it a clear net win, and
large messages are the minority case. Flush
the accumulated batch state first so the
ratchet path observes up-to-date counters. */
$this->flushBatch($job_id, $seen_tuples,
$batch_imported, $last_uid);
$seen_tuples = [];
$batch_imported = 0;
$this->clone_model->updateProgress($job_id, [
'CURRENT_UID' => $uid,
'CURRENT_UID_PARTIAL_OFFSET' => 0,
'CURRENT_UID_PARTIAL_SIZE' => $size,
]);
$job_row = $this->clone_model->getJob($job_id,
$this->jobUserId($job_id));
$this->ratchetLargeMessage($job_id, $job_row,
$backend, $folder_name, $uidvalidity,
$uid);
continue;
}
$bytes = $entry['body'];
if ($bytes === null || $bytes === '') {
$this->clone_model->recordError($job_id,
$folder_name, $uidvalidity, $uid,
'empty_body',
"UID $uid: server returned empty body");
$seen_tuples[] = [$uidvalidity, $uid];
continue;
}
$delivered = $this->deliverToDestinationCached(
$folder_name, $bytes, $entry['flags'],
$entry['internal_date']);
$seen_tuples[] = [$uidvalidity, $uid];
if ($delivered) {
$batch_imported++;
}
$this->messages_since_log++;
$this->fetch_us_since_log += $per_msg_fetch_us;
}
$this->flushBatch($job_id, $seen_tuples,
$batch_imported, $last_uid);
if ($this->messages_since_log >=
self::PROGRESS_LOG_EVERY) {
$this->logProgressWindow($job_id, $folder_name);
}
}
/**
* Flushes the per-batch accumulated state to the DB in as
* few statements as possible: one transactional bulk
* insert into MAIL_CLONE_SEEN, then one UPDATE on
* MAIL_CLONE_JOB that bumps IMPORTED / FAILED / CURRENT_UID
* together. Empty input is a no-op.
*
* The IMPORTED counter is the sum of the existing row
* value plus the per-batch delta (read once via getJob,
* then computed in PHP); the same applies to FAILED. This
* means the whole batch's persistence cost is two DB
* round-trips regardless of batch size, instead of the
* previous N markSeen calls + N/10 updateProgress calls.
*
* @param int $job_id job whose row to update
* @param array $seen_tuples list of [uidvalidity, uid] pairs
* to insert into MAIL_CLONE_SEEN
* @param int $imported_delta count to add to IMPORTED
* @param int $last_uid highest UID processed this batch; 0
* means leave CURRENT_UID alone
*/
protected function flushBatch($job_id, $seen_tuples,
$imported_delta, $last_uid)
{
if (empty($seen_tuples) && $imported_delta === 0 &&
$last_uid === 0) {
return;
}
try {
$this->clone_model->markSeenBulk($job_id,
$seen_tuples);
} catch (\Exception $exception) {
/* Bulk insert failed: fall back to per-row inserts
so a single bad row does not abandon the whole
batch's dedup state. Each per-row insert is
idempotent (insertIgnore) so retries here are
safe. */
L\crawlLog("MailCloneJob: markSeenBulk failed (" .
$exception->getMessage() . "); falling back " .
"to per-row inserts");
foreach ($seen_tuples as $tuple) {
$this->clone_model->markSeen($job_id,
(int) $tuple[0], (int) $tuple[1]);
}
}
/* Any previously-recorded errors against UIDs that just
succeeded get marked resolved so the status page can
show the rescue and stop counting them as Failed.
Group by uidvalidity in case the batch spans folders
(in practice it never does, but the grouping is cheap
and keeps the contract honest). */
$by_uidvalidity = [];
foreach ($seen_tuples as $tuple) {
$by_uidvalidity[(int) $tuple[0]][] = (int) $tuple[1];
}
foreach ($by_uidvalidity as $uidvalidity => $uids) {
$this->clone_model->markErrorsResolved($job_id,
$uidvalidity, $uids, 'retry_succeeded');
}
$updates = [];
if ($last_uid > 0) {
$updates['CURRENT_UID'] = $last_uid;
}
if ($imported_delta > 0 ||
$this->pending_skipped_delta > 0) {
$row = $this->clone_model->getJob($job_id,
$this->jobUserId($job_id));
if ($imported_delta > 0) {
$updates['IMPORTED'] =
((int) ($row['IMPORTED'] ?? 0)) +
$imported_delta;
}
if ($this->pending_skipped_delta > 0) {
$updates['SKIPPED'] =
((int) ($row['SKIPPED'] ?? 0)) +
$this->pending_skipped_delta;
$this->pending_skipped_delta = 0;
}
}
/* FAILED is recomputed from the authoritative source
(MAIL_CLONE_ERROR.RESOLVED_AT = 0) rather than
accumulated from a per-batch delta: previously-failed
UIDs that get rescued this batch must subtract from
the count, and the table query gets that for free
where a running delta would not. */
$updates['FAILED'] =
$this->clone_model->unresolvedErrorCount($job_id);
if (!empty($updates)) {
$this->clone_model->updateProgress($job_id,
$updates);
}
}
/**
* Demultiplexes a pipelined UID FETCH response into a
* per-UID array. Each untagged FETCH line starts with
* "* N FETCH (UID U ...)"; the asked-for UID is extracted
* from that inner UID token, and the matching literal (if
* the FETCH included BODY[]) lives at the same index in
* the literals array. Returns
* [uid => ['size' => int, 'flags' => array,
* 'internal_date' => int, 'body' => string|null]].
*
* @param array $untagged untagged response lines
* @param array $literals parallel literal payloads keyed
* by untagged-line index
* @return array per-UID demultiplexed entries
*/
protected function parseFetchedBatch($untagged, $literals)
{
$result = [];
foreach ($untagged as $idx => $line) {
if (!preg_match('/^\* \d+ FETCH /', $line)) {
continue;
}
if (!preg_match('/\bUID\s+(\d+)/i', $line,
$match)) {
continue;
}
$uid = (int) $match[1];
$entry = $this->parseFetchMeta([$line], $uid);
$entry['body'] = $literals[$idx] ?? null;
$result[$uid] = $entry;
}
return $result;
}
/**
* Fetches and appends one source UID. For messages at or
* above LARGE_MESSAGE_THRESHOLD_BYTES, delegates to the
* partial-fetch ratchet which can span multiple ticks;
* otherwise issues one FETCH for the whole message.
*
* @param int $job_id job being advanced
* @param ImapMailBackend $backend open backend
* @param string $folder_name source folder
* @param int $uidvalidity source folder's UIDVALIDITY
* @param int $uid source message UID
*/
protected function advanceMessage($job_id, $backend,
$folder_name, $uidvalidity, $uid)
{
$client = $backend->imapClient();
$fetch_started = microtime(true);
/* One combined FETCH for size + flags + INTERNALDATE +
body. This was previously two round-trips (a small
meta FETCH followed by a BODY.PEEK[] FETCH); merging
halves the per-message IMAP latency, which is the
dominant cost on a WAN connection to the source
server. Large messages (>= LARGE_MESSAGE_THRESHOLD_BYTES)
still go through the ratchet branch: we throw away
the combined-fetch body in that case and re-fetch in
chunks. The waste only happens once per large
message and large messages are the minority case. */
$combined = $client->send("UID FETCH $uid " .
"(RFC822.SIZE FLAGS INTERNALDATE BODY.PEEK[])");
$fetch_elapsed_us = (microtime(true) - $fetch_started)
* 1000000;
$this->fetch_us_since_log += $fetch_elapsed_us;
if ($combined['status'] !== 'OK') {
$this->clone_model->recordError($job_id,
$folder_name, $uidvalidity, $uid,
'combined_fetch_failed',
"UID $uid combined fetch failed: " .
$combined['detail']);
$this->clone_model->updateProgress($job_id, [
'FAILED' => $this->clone_model
->unresolvedErrorCount($job_id),
]);
$this->clone_model->markSeen($job_id, $uidvalidity,
$uid);
return;
}
$parsed_meta = $this->parseFetchMeta(
$combined['untagged'], $uid);
$size = $parsed_meta['size'];
if ($size >= self::LARGE_MESSAGE_THRESHOLD_BYTES) {
/* Flush the pending IMPORTED counter before
dropping into the ratchet path: ratchetLargeMessage
re-reads the job row from the DB and we want it
to see the up-to-date count. */
$this->flushImportedDelta($job_id);
$this->clone_model->updateProgress($job_id, [
'CURRENT_UID' => $uid,
'CURRENT_UID_PARTIAL_OFFSET' => 0,
'CURRENT_UID_PARTIAL_SIZE' => $size,
]);
$job_row = $this->clone_model->getJob($job_id,
$this->jobUserId($job_id));
$this->ratchetLargeMessage($job_id, $job_row,
$backend, $folder_name, $uidvalidity, $uid);
return;
}
$bytes = $this->parseFetchBody($combined['untagged'],
$combined['literals'], $uid);
if ($bytes === null) {
$this->clone_model->recordError($job_id,
$folder_name, $uidvalidity, $uid, 'empty_body',
"UID $uid: empty body parse");
$this->clone_model->updateProgress($job_id, [
'FAILED' => $this->clone_model
->unresolvedErrorCount($job_id),
'CURRENT_UID' => $uid,
]);
$this->clone_model->markSeen($job_id, $uidvalidity,
$uid);
return;
}
/* deliverToDestinationCached uses the per-tick
destination_backend cached in advanceJob; this saves
one MailSiteFactory::build + FileMailStorage open per
appended message versus rebuilding the backend each
call. */
$delivered = $this->deliverToDestinationCached(
$folder_name, $bytes,
$parsed_meta['flags'], $parsed_meta['internal_date']);
$this->clone_model->markSeen($job_id, $uidvalidity,
$uid);
if ($delivered) {
$this->pending_imported_delta++;
}
$this->messages_since_log++;
if ($this->pending_imported_delta >=
self::PROGRESS_FLUSH_EVERY) {
$this->flushImportedDelta($job_id);
$this->clone_model->updateProgress($job_id, [
'CURRENT_UID' => $uid,
]);
}
if ($this->messages_since_log >=
self::PROGRESS_LOG_EVERY) {
$this->logProgressWindow($job_id, $folder_name);
}
}
/**
* Returns the staging directory path for a job's large-message
* ratchet files. Centralizes the WORK_DIRECTORY/clone-staging/
* <job_id> layout so the ratchet and the purge code agree on
* one location.
*
* @param int $job_id job whose staging directory to name
* @return string absolute path (not guaranteed to exist)
*/
protected function stagingDir($job_id)
{
return C\WORK_DIRECTORY . "/clone-staging/" .
(int) $job_id;
}
/**
* Removes a job's staging directory and any partial-message
* (.part) files it holds. Called when a job reaches a
* terminal state (done, cancelled, failed): the partial
* large-message downloads in there are only useful for
* resuming an in-flight ratchet, so once the job will not
* advance again they are pure detritus. Deleting them is
* always safe even if the job is later resurrected, because
* the ratchet treats a missing .part file as offset 0 and
* simply re-downloads from the start.
*
* Best-effort: a file that cannot be unlinked (permissions,
* races) is skipped rather than throwing, since cleanup must
* never abort a status transition. Unexpected entries (not
* .part) are left alone and the directory removal is then
* skipped so nothing outside this job's own files is touched.
*
* @param int $job_id job whose staging directory to remove
*/
protected function purgeStagingForJob($job_id)
{
$staging_dir = $this->stagingDir($job_id);
if (!is_dir($staging_dir)) {
return;
}
$entries = @scandir($staging_dir);
if ($entries === false) {
return;
}
$unexpected = false;
foreach ($entries as $entry) {
if ($entry === '.' || $entry === '..') {
continue;
}
if (substr($entry, -5) === '.part') {
@unlink($staging_dir . "/" . $entry);
} else {
$unexpected = true;
}
}
if (!$unexpected) {
@rmdir($staging_dir);
}
}
/**
* Sweeps the clone-staging root for directories whose job is
* gone or already terminal, removing their leftover .part
* files. This catches detritus from before terminal-state
* cleanup existed, and from any crash that skipped the
* per-job purge. Runs once per tick at job-list build time;
* the scan is over a small directory (one entry per job that
* ever ratcheted a large message) so the cost is negligible.
*
* A directory is purged when its job id no longer maps to a
* live ('pending' or 'running') job. Live jobs are passed in
* as a set of ids so the sweep needs no extra DB round-trip
* per directory.
*
* @param array $live_job_ids set of job ids (as array keys)
* that are still pending or running and must be kept
*/
protected function sweepOrphanStaging($live_job_ids)
{
$root = C\WORK_DIRECTORY . "/clone-staging";
if (!is_dir($root)) {
return;
}
$entries = @scandir($root);
if ($entries === false) {
return;
}
foreach ($entries as $entry) {
if ($entry === '.' || $entry === '..') {
continue;
}
if (!ctype_digit($entry)) {
continue;
}
if (!isset($live_job_ids[(int) $entry])) {
$this->purgeStagingForJob((int) $entry);
}
}
}
/**
* Drives the partial-fetch ratchet on one large message.
* Reads existing staging-file size as the starting offset
* (so a tick-crash mid-ratchet resumes from disk), pulls
* one PARTIAL_FETCH_CHUNK_BYTES chunk via BODY.PEEK[<off.n>]
* per call until the offset equals the announced size,
* then reads the staging file, calls appendMessage, and
* cleans up. Returns whether the message is fully imported
* by the end of this call; false means the caller should
* yield (more chunks needed on a future tick) without
* advancing past this UID.
*
* @param int $job_id job being advanced
* @param array $job_row current job row (for sizes/offsets)
* @param ImapMailBackend $backend open backend
* @param string $folder_name source folder
* @param int $uidvalidity source folder's UIDVALIDITY
* @param int $uid source message UID
* @return bool whether this UID is now fully imported
*/
protected function ratchetLargeMessage($job_id, $job_row,
$backend, $folder_name, $uidvalidity, $uid)
{
$staging_dir = $this->stagingDir($job_id);
if (!is_dir($staging_dir)) {
@mkdir($staging_dir, 0700, true);
}
$staging_path = $staging_dir . "/" . $uid . ".part";
$offset = file_exists($staging_path) ?
filesize($staging_path) : 0;
$size = (int) $job_row['CURRENT_UID_PARTIAL_SIZE'];
if ($size <= 0) {
$client = $backend->imapClient();
$meta = $client->send("UID FETCH $uid RFC822.SIZE");
if ($meta['status'] !== 'OK') {
throw new \Exception(
"UID $uid RFC822.SIZE re-fetch failed: " .
$meta['detail']);
}
$parsed = $this->parseFetchMeta($meta['untagged'],
$uid);
$size = $parsed['size'];
}
$client = $backend->imapClient();
while ($offset < $size) {
if (microtime(true) >= $this->tick_deadline) {
$this->clone_model->updateProgress($job_id, [
'CURRENT_UID' => $uid,
'CURRENT_UID_PARTIAL_OFFSET' => $offset,
'CURRENT_UID_PARTIAL_SIZE' => $size,
]);
return false;
}
$want = min(self::PARTIAL_FETCH_CHUNK_BYTES,
$size - $offset);
$fetch = $client->send(
"UID FETCH $uid BODY.PEEK[]<$offset.$want>");
if ($fetch['status'] !== 'OK') {
throw new \Exception("UID $uid partial " .
"fetch failed at offset $offset: " .
$fetch['detail']);
}
$chunk = $this->parseFetchBody($fetch['untagged'],
$fetch['literals'], $uid);
if ($chunk === null || $chunk === '') {
throw new \Exception("UID $uid partial " .
"fetch returned empty chunk at offset " .
$offset);
}
$handle = fopen($staging_path, 'a');
if ($handle === false) {
throw new \Exception(
"could not open staging $staging_path");
}
fwrite($handle, $chunk);
fclose($handle);
$offset += strlen($chunk);
}
$bytes = file_get_contents($staging_path);
if ($bytes === false) {
throw new \Exception(
"could not read staging $staging_path");
}
$meta_command = $client->send(
"UID FETCH $uid (FLAGS INTERNALDATE)");
$flags = [];
$internal_date = 0;
if ($meta_command['status'] === 'OK') {
$parsed = $this->parseFetchMeta(
$meta_command['untagged'], $uid);
$flags = $parsed['flags'];
$internal_date = $parsed['internal_date'];
}
$current = $this->clone_model->getJob($job_id,
$this->jobUserId($job_id));
/* Apply the same skip-if-present dedup the small-message
path uses: a large message already in the destination
folder (e.g. from a prior clone) is counted as skipped
rather than appended again. Dry-run jobs have no
destination to check, so destinationHasMessage returns
false and they always "deliver" (a no-op) to keep
dry-run counts faithful. */
if ($this->destinationHasMessage($folder_name, $bytes)) {
@unlink($staging_path);
$this->clone_model->markSeen($job_id, $uidvalidity,
$uid);
$this->clone_model->updateProgress($job_id, [
'CURRENT_UID' => $uid,
'CURRENT_UID_PARTIAL_OFFSET' => 0,
'CURRENT_UID_PARTIAL_SIZE' => 0,
'SKIPPED' => $this->incrementSkipped($job_id),
]);
return true;
}
$this->deliverToDestination($current, $folder_name,
$bytes, $flags, $internal_date);
$message_id = $this->parseMessageId($bytes);
if ($message_id !== '' &&
$this->destination_message_id_folder ===
$folder_name) {
$this->destination_message_ids[$message_id] = true;
}
@unlink($staging_path);
$this->clone_model->markSeen($job_id, $uidvalidity,
$uid);
$this->clone_model->updateProgress($job_id, [
'CURRENT_UID' => $uid,
'CURRENT_UID_PARTIAL_OFFSET' => 0,
'CURRENT_UID_PARTIAL_SIZE' => 0,
'IMPORTED' => $this->incrementImported($job_id),
]);
return true;
}
/**
* Appends one message to the destination user's MailSite
* mailbox, preserving original flags and INTERNALDATE so
* the local copy matches the source view. Skipped silently
* when DRY_RUN is set on the job (the job still advances
* its cursor and dedup state so a follow-up real run
* starts from where the dry-run left off).
*
* @param array $job_row current job row
* @param string $folder_name destination folder (same name
* as source; MailSite creates on first append if
* missing)
* @param string $bytes raw RFC 5322 message bytes
* @param array $flags IMAP flag list (e.g. ['\\Seen'])
* @param int $internal_date source INTERNALDATE as unix ts
*/
protected function deliverToDestination($job_row,
$folder_name, $bytes, $flags, $internal_date)
{
if (!empty($job_row['DRY_RUN'])) {
return;
}
$backend = new MailSiteMailBackend(
$job_row['DESTINATION_USER']);
$backend->appendMessageWithInternalDate($folder_name,
$bytes, $flags, $internal_date);
}
/**
* Hot-path delivery that reuses the per-tick
* destination_backend cached in advanceJob, saving one
* MailSiteFactory::build and one FileMailStorage open per
* message versus the old per-message-rebuilt path. Dry-run
* jobs leave destination_backend null in advanceJob, so
* this is a no-op for them.
*
* @param string $folder_name destination folder
* @param string $bytes raw RFC 5322 message
* @param array $flags IMAP flag list
* @param int $internal_date source INTERNALDATE as unix ts
*/
protected function deliverToDestinationCached($folder_name,
$bytes, $flags, $internal_date)
{
if ($this->destination_backend === null) {
/* Dry-run: nothing is written, so there is no
destination to deduplicate against. Report as
delivered so dry-run counts mirror a real run. */
return true;
}
if ($this->destinationHasMessage($folder_name, $bytes)) {
$this->pending_skipped_delta++;
return false;
}
$this->destination_backend->appendMessageWithInternalDate(
$folder_name, $bytes, $flags, $internal_date);
/* Track the just-appended Message-ID so a duplicate later
in the same run (e.g. the source listing the same UID
twice, or two copies with one Message-ID) is also
skipped without a second destination scan. */
$message_id = $this->parseMessageId($bytes);
if ($message_id !== '') {
$this->destination_message_ids[$message_id] = true;
}
return true;
}
/**
* Whether the destination folder already holds a message
* with the same Message-ID as the supplied raw source
* message. Builds the set of destination Message-IDs once
* per folder per run (header-only reads, cached on the job
* instance) and reuses it for every message in that folder;
* moving to a new folder rebuilds the set. A source message
* with no parseable Message-ID is treated as not-present so
* it is always delivered (we cannot prove it is a
* duplicate).
*
* @param string $folder_name destination folder to check
* @param string $bytes raw RFC 5322 source message
* @return bool true when an identical message already exists
*/
protected function destinationHasMessage($folder_name,
$bytes)
{
if ($this->destination_backend === null) {
return false;
}
if ($this->destination_message_id_folder !==
$folder_name) {
$this->buildDestinationMessageIds($folder_name);
}
$message_id = $this->parseMessageId($bytes);
if ($message_id === '') {
return false;
}
return isset(
$this->destination_message_ids[$message_id]);
}
/**
* Builds destination_message_ids for a folder: enumerates
* the destination folder's messages and records each one's
* Message-ID. Reads headers only (not bodies) and runs once
* per folder per run. A folder the destination does not yet
* have simply yields an empty set.
*
* @param string $folder_name destination folder to index
* @return void
*/
protected function buildDestinationMessageIds($folder_name)
{
$this->destination_message_ids = [];
$this->destination_message_id_folder = $folder_name;
try {
$records = $this->destination_backend
->listMessageHeaders($folder_name);
} catch (\Exception $exception) {
L\crawlLog("MailCloneJob: could not index " .
"destination folder for dedup (" .
$exception->getMessage() . "); proceeding " .
"without skip-if-present for this folder.");
return;
}
foreach ($records as $header_bytes) {
$message_id = $this->parseMessageId($header_bytes);
if ($message_id !== '') {
$this->destination_message_ids[$message_id] =
true;
}
}
}
/**
* Extracts and normalizes the Message-ID from a raw message
* or header block. Returns the lowercased token inside the
* angle brackets (or the trimmed lowercased value when no
* brackets are present), or the empty string when there is
* no Message-ID header. Lowercasing makes the comparison
* case-insensitive on the domain part, which RFC 5322 treats
* as case-insensitive; the local part is technically
* case-sensitive but mail systems overwhelmingly treat the
* whole token case-insensitively and an over-match here only
* skips a message the user already has.
*
* @param string $bytes raw message or header bytes
* @return string normalized Message-ID or empty string
*/
protected function parseMessageId($bytes)
{
if ($bytes === null || $bytes === '') {
return '';
}
$matched = [];
if (!preg_match('/^message-id:\s*(.+)$/im', $bytes,
$matched)) {
return '';
}
$value = trim($matched[1]);
$bracketed = [];
if (preg_match('/<([^>]+)>/', $value, $bracketed)) {
$value = $bracketed[1];
}
return strtolower(trim($value));
}
/**
* Flushes any queued IMPORTED and SKIPPED counter deltas to
* MAIL_CLONE_JOB in one UPDATE. Called every
* PROGRESS_FLUSH_EVERY messages from the per-UID path and
* once more at advanceJob's finally so the final counts are
* visible even when a tick ends mid-batch. SKIPPED
* accumulates duplicates skipped because the message already
* existed in the destination folder.
*
* @param int $job_id job whose counters to flush
*/
protected function flushImportedDelta($job_id)
{
if ($this->pending_imported_delta <= 0 &&
$this->pending_skipped_delta <= 0) {
return;
}
$row = $this->clone_model->getJob($job_id,
$this->jobUserId($job_id));
$updates = [];
if ($this->pending_imported_delta > 0) {
$updates['IMPORTED'] =
((int) ($row['IMPORTED'] ?? 0)) +
$this->pending_imported_delta;
$this->pending_imported_delta = 0;
}
if ($this->pending_skipped_delta > 0) {
$updates['SKIPPED'] =
((int) ($row['SKIPPED'] ?? 0)) +
$this->pending_skipped_delta;
$this->pending_skipped_delta = 0;
}
$this->clone_model->updateProgress($job_id, $updates);
}
/**
* Emits one log line summarizing the last
* PROGRESS_LOG_EVERY messages: msg/sec throughput overall
* and mean IMAP-fetch latency in milliseconds. This is the
* primary on-server instrumentation for diagnosing slow
* clones: if msg/sec is low and mean fetch is high, the
* bottleneck is the upstream IMAP server; if msg/sec is
* low and mean fetch is low, the bottleneck is local I/O
* (DB writes or FileMailStorage). Resets the counters so
* the next window measures fresh.
*
* @param int $job_id job being logged
* @param string $folder_name current folder
*/
protected function logProgressWindow($job_id, $folder_name)
{
$elapsed = microtime(true) - $this->log_window_started;
$msgs_per_sec = ($elapsed > 0) ?
($this->messages_since_log / $elapsed) : 0;
$mean_fetch_ms = ($this->messages_since_log > 0) ?
($this->fetch_us_since_log /
$this->messages_since_log / 1000) : 0;
L\crawlLog(sprintf("MailCloneJob: job %d folder %s " .
"%d msgs in %.2fs (%.1f msg/s, mean IMAP fetch " .
"%.0fms)", $job_id, $folder_name,
$this->messages_since_log, $elapsed,
$msgs_per_sec, $mean_fetch_ms));
$this->messages_since_log = 0;
$this->fetch_us_since_log = 0.0;
$this->log_window_started = microtime(true);
}
/**
* Wipes every folder under the destination MailSite user.
* Used at the pending-to-running transition when MODE='wipe'.
* Scope is strictly the user named in DESTINATION_USER; no
* other user's mail is touched. Folders are listed and
* purged one at a time (idempotent if interrupted; a partial
* wipe is fine to resume).
*
* @param string $destination_user MailSite local-part to wipe
*/
protected function wipeDestination($destination_user)
{
$backend = new MailSiteMailBackend($destination_user);
$folders = $backend->listFolders();
foreach ($folders as $folder_row) {
$folder_name = $folder_row['NAME'];
$backend->purgeFolder($folder_name);
}
}
/**
* Pulls UIDVALIDITY out of a SELECT response. Per RFC 3501
* the value appears in an untagged OK [UIDVALIDITY N] line.
* Returns 0 when not found, which a caller should treat as
* a server-quirk fallback (the dedup key still works but
* loses its UIDVALIDITY-rotation safety net).
*
* @param array $untagged untagged response lines
* @return int the UIDVALIDITY value, or 0 when missing
*/
protected function parseUidValidity($untagged)
{
foreach ($untagged as $line) {
if (preg_match('/\[UIDVALIDITY\s+(\d+)\]/i',
$line, $match)) {
return (int) $match[1];
}
}
return 0;
}
/**
* Parses UID SEARCH untagged output into a sorted, deduped
* list of integer UIDs. Mirrors the ImapMailBackend helper
* of the same name; reproduced here because that one is
* protected.
*
* @param array $untagged untagged response lines
* @return array list of ints
*/
protected function parseSearchUids($untagged)
{
$uids = [];
foreach ($untagged as $line) {
if (preg_match('/^\* SEARCH(.*)$/', $line, $match)) {
$parts = preg_split('/\s+/',
trim($match[1]));
foreach ($parts as $part) {
if (ctype_digit($part)) {
$uids[] = (int) $part;
}
}
}
}
return $uids;
}
/**
* Extracts RFC822.SIZE / FLAGS / INTERNALDATE values from a
* FETCH response keyed by UID. Returned shape:
* ['size' => int, 'flags' => array, 'internal_date' => int].
* Defaults: size 0, empty flags, internal_date 0 (current
* time-equivalent at append; the storage layer treats 0 as
* "use now").
*
* @param array $untagged untagged response lines from the
* meta FETCH
* @param int $uid the UID we asked for
* @return array meta values
*/
protected function parseFetchMeta($untagged, $uid)
{
$size = 0;
$flags = [];
$internal_date = 0;
foreach ($untagged as $line) {
if (preg_match('/RFC822\.SIZE\s+(\d+)/i', $line,
$match)) {
$size = (int) $match[1];
}
if (preg_match('/FLAGS\s+\(([^)]*)\)/i', $line,
$match)) {
$raw_flags = trim($match[1]);
if ($raw_flags !== '') {
$flags = preg_split('/\s+/', $raw_flags);
}
}
if (preg_match('/INTERNALDATE\s+"([^"]+)"/i',
$line, $match)) {
$parsed = strtotime($match[1]);
if ($parsed !== false) {
$internal_date = $parsed;
}
}
}
return ['size' => $size, 'flags' => $flags,
'internal_date' => $internal_date];
}
/**
* Extracts the raw body bytes from a BODY.PEEK[] FETCH
* response. The wire format puts the bytes in a literal
* {N} immediately following the FETCH header line; the
* ImapClient send() machinery has already captured them
* into the parallel 'literals' array keyed by line index.
* Returns null when no literal was found (shouldn't
* happen on a well-formed server response but the caller
* treats null as a transient failure).
*
* @param array $untagged untagged response lines
* @param array $literals parallel array of literal payloads
* keyed by the untagged-line index
* @param int $uid UID this fetch was for (logging only)
* @return string|null body bytes, or null on parse failure
*/
protected function parseFetchBody($untagged, $literals,
$uid)
{
foreach ($untagged as $idx => $line) {
if (preg_match('/^\* \d+ FETCH /', $line) &&
isset($literals[$idx])) {
return $literals[$idx];
}
}
return null;
}
/**
* Reads the IMPORTED counter, adds one, returns the new
* value. Separate helper so updateProgress callers can
* pass an absolute new value (the allow-list filter in
* updateProgress does not support relative arithmetic).
*
* @param int $job_id job whose counter to bump
* @return int the new IMPORTED value
*/
protected function incrementImported($job_id)
{
$row = $this->clone_model->getJob($job_id,
$this->jobUserId($job_id));
return ((int) ($row['IMPORTED'] ?? 0)) + 1;
}
/**
* Reads the SKIPPED counter, adds one, returns the new
* value. Mirror of incrementImported for the large-message
* ratchet path, which writes an absolute SKIPPED value
* through updateProgress when a large message is already
* present in the destination folder.
*
* @param int $job_id job whose counter to bump
* @return int the new SKIPPED value
*/
protected function incrementSkipped($job_id)
{
$row = $this->clone_model->getJob($job_id,
$this->jobUserId($job_id));
return ((int) ($row['SKIPPED'] ?? 0)) + 1;
}
/**
* Looks up the USER_ID for a job, used by counter-bump
* helpers that need the user-scope predicate on getJob.
* Cached per call (the row is short and the model lookup
* is one query) rather than threaded through every helper
* signature.
*
* @param int $job_id job whose owner to find
* @return int USER_ID or 0 when missing
*/
protected function jobUserId($job_id)
{
$sql = "SELECT USER_ID FROM MAIL_CLONE_JOB " .
"WHERE ID = ? " . $this->db->limitOffset(1);
$result = $this->db->execute($sql, [$job_id]);
if ($result === false) {
return 0;
}
$row = $this->db->fetchArray($result);
return (int) ($row['USER_ID'] ?? 0);
}
}
/**
* Tiny duck-type shim that satisfies the userMailLog interface
* ImapMailBackend expects from a SocialComponent. The backend
* calls $this->component->userMailLog($tag, $fields) from
* roughly a dozen call sites; outside an HTTP request there is
* no SocialComponent to hand in, but the log lines are useful
* for diagnostics. This shim writes the same one-line-per-event
* format SmtpClient::appendLog produces, so admin-tailing
* mail.log sees a continuous stream regardless of whether the
* IMAP traffic originated from a user request or from
* MailCloneJob.
*
* @author Chris Pollett
*/
class MailCloneLogger
{
/**
* Max characters per individual log field value before
* truncation. Matches the limit SocialComponent::userMailLog
* uses so the two log streams look the same in mail.log.
*/
const FIELD_MAX_LEN = 200;
/**
* Appends one line to mail.log in the same shape that
* SocialComponent::userMailLog produces, so tailers cannot
* tell which originator wrote the line.
*
* @param string $tag short kebab-case event name, e.g.
* 'imap-login-ok'
* @param array $fields associative array of context fields
* stringified into key=value pairs
*/
public function userMailLog($tag, $fields = [])
{
$parts = [];
foreach ($fields as $key => $value) {
$clean = str_replace(["\r", "\n", "\t"], " ",
(string) $value);
if (strlen($clean) > self::FIELD_MAX_LEN) {
$clean = substr($clean, 0,
self::FIELD_MAX_LEN) . "…";
}
$parts[] = $key . "=" . $clean;
}
$line = "[" . date(DATE_RFC822) . "] " . $tag;
if (!empty($parts)) {
$line .= ": " . implode(" ", $parts);
}
$line .= "\n";
\seekquarry\yioop\library\mail\SmtpClient::appendLog($line);
}
}
(c) 2026 Seekquarry - Open Source Search Engine Software