mirror of
https://github.com/karatektus/pl_mail.git
synced 2026-10-27 02:45:32 +00:00
Learn Gmail's quota from Google's refusal instead of assuming 15,000
Google lowered the per-user limit to 6,000 units a minute for projects created since May 2026 and raised a messages.get from 5 units to 20. The pacer had 15,000 and 5 written into it, so what it took for 9,000 units a minute was 36,000: most of what an import asked for was refused, the log filled with retries, and a refused listing ended the import's chain at CRITICAL (#90). the limit starts at 15,000 and is read out of the first 403, which names the limit and its value; forgotten after a day so a raised one is found again. Nothing to set. the cost 20 a message, and fifty messages a batch shared spend, refusal and limit are in a database-backed cache pool, so worker-backlog, worker-ingest and the web container draw on one allowance a refusal is "not now", not an error: a batch, an import page or a sync that meets one is put back until the account's minute is over, and a refused sync still restarts an import that has gone quiet The import line in the background-work indicator shows the limit, roughly what is in use of it, how long the rest takes, and says so when Google has paused the account. No migration: the pool is a key prefix in cache_items. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Paul LütznerandClaude Opus 5.5 committed
1 parent
5debbdbc01
commit
b44283cb1c
23 files changed
+1133
-122
No files matched your search
@@ -8,6 +8,18 @@ The published image tags: `latest` follows the most recent release below,
|
||||
|
||||
## Unreleased
|
||||
|
||||
**A Gmail import no longer runs into Google's quota and stays there.** Google lowered the
|
||||
per-user limit to 6,000 units a minute for projects created since May 2026 and raised the cost
|
||||
of fetching a message from 5 units to 20; plMail paced imports for 15,000 at the old cost, so on
|
||||
a new project most of what an import asked for was refused, the log filled with rate-limit
|
||||
warnings and an import could stop without resuming. plMail now reads the limit out of Google's
|
||||
first refusal and keeps to it — nothing to configure — counts a message at its real cost,
|
||||
fetches fifty at a time, and shares one allowance between its workers. A refusal is no longer an
|
||||
error: the batch, import page or sync that met it waits out the minute and carries on. The
|
||||
import line in the background-work indicator now shows the quota, roughly how much of it is in
|
||||
use, how long the rest of the import takes, and says so when Google has paused it. No
|
||||
migration. [#90](https://github.com/karatektus/pl_mail/issues/90), reported by @karamanliev.
|
||||
|
||||
**The tab icon is the one you picked, in Firefox too.** Settings → Appearance → Logo changed the
|
||||
icon in the app but not in the tab, which kept showing the wax seal. Every page also linked a
|
||||
fixed 192px wax-seal PNG as a second tab icon, and Firefox prefers that to the SVG for a tab. The
|
||||
|
||||
@@ -47,6 +47,16 @@ framework:
|
||||
provider: doctrine.dbal.default_connection
|
||||
default_lifetime: 3600
|
||||
|
||||
# What each Gmail account has spent of its quota this minute, when
|
||||
# Google last refused it, and the limit a refusal named. Shared for
|
||||
# the reason the signal above is: worker-backlog, worker-ingest and
|
||||
# the web container all call Gmail for the same account, and a
|
||||
# refusal one of them met has to hold the others back. No default
|
||||
# lifetime: GmailQuotaPacer sets each entry's own.
|
||||
app.gmail_quota:
|
||||
adapter: cache.adapter.doctrine_dbal
|
||||
provider: doctrine.dbal.default_connection
|
||||
|
||||
# A Stop pressed on the summary card (see ThreadSummaryStop).
|
||||
#
|
||||
# Shared for the same reason again: the Stop is a request of its
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
<!-- translated-from: internals/mail-ingest.md sha1:dcb2b781f03c1702974cc382857086d18024e21c -->
|
||||
<!-- translated-from: internals/mail-ingest.md sha1:7329297e0b7553b6e0fb3e23948afdcff155a609 -->
|
||||
# Mail-Ingest
|
||||
|
||||
Vom Anbieter in die Datenbank: wie aus Bytes eine Zeile `Message` wird, was mit ihr geschieht,
|
||||
@@ -93,6 +93,15 @@ die Vorgeschichte gehört `App\Service\Mail\MailImporter`, auf der Warteschlange
|
||||
- **Gmail fragt zuerst namentlich nach dem Posteingang,** dann blättert es die Auflistung des
|
||||
ganzen Postfachs Anfrage für Anfrage durch; das Seiten-Token liegt zwischen den Jobs auf dem
|
||||
Konto.
|
||||
- **Gmails Kontingent wird gelernt, nicht eingestellt.** Google erlaubt einem Nutzer eine Zahl
|
||||
von Kontingent-Einheiten pro Minute und sagt erst, wie viele, wenn es eine Anfrage ablehnt.
|
||||
`GmailQuotaPacer` beginnt bei 15.000, liest das Limit aus der ersten Ablehnung (ein seit Mai
|
||||
2026 angelegtes Projekt hat 6.000) und taktet von da an jedes Konto auf vier Fünftel davon;
|
||||
ein gelerntes Limit wird nach einem Tag vergessen, damit ein angehobenes wiedergefunden wird.
|
||||
Was ein Konto verbraucht hat und wann es zuletzt abgelehnt wurde, liegt in einem Cache-Pool
|
||||
in der Datenbank, sodass alle Worker aus einem Kontingent schöpfen. Eine Ablehnung ist auf
|
||||
diesem Weg nirgends ein Fehler: Ein Stapel, eine Import-Seite oder ein Sync, der auf eine
|
||||
trifft, wird zurückgestellt und läuft, wenn die Minute des Kontos vorbei ist.
|
||||
- **Microsoft liest die erste Delta-Runde jedes Ordners seitenweise.** Ein Ordner ohne
|
||||
Delta-Link wird auf dem Konto eingeplant (`Account::$graphImport`), statt vom Abgleich
|
||||
aufgezählt zu werden, und `GraphApiSyncer::importPage()` geht diese Runde in Schritten von
|
||||
|
||||
@@ -82,6 +82,14 @@ The two are separate now. A sync arranges for new mail to be seen and stops; the
|
||||
rule.
|
||||
- **Gmail asks for the Inbox by name first,** then pages the whole-mailbox listing one request
|
||||
at a time, with the page token kept on the account between jobs.
|
||||
- **Gmail's quota is learnt, not configured.** Google allows one user a number of quota units
|
||||
a minute and does not say how many until it refuses a request. `GmailQuotaPacer` starts from
|
||||
15,000, reads the limit out of the first refusal (a project created since May 2026 has
|
||||
6,000) and paces every account to four fifths of it from then on; a learnt limit is forgotten
|
||||
after a day so a raised one is found again. What an account has spent and when it was last
|
||||
refused are in a database-backed cache pool, so every worker draws on one allowance. A
|
||||
refusal is not an error anywhere on this path: a batch, an import page or a sync that meets
|
||||
one is put back to run when the account's minute is over.
|
||||
- **Microsoft reads each folder's first delta round a page at a time.** A folder without a
|
||||
delta link is planned on the account (`Account::$graphImport`) instead of being enumerated
|
||||
by the sync, and `GraphApiSyncer::importPage()` walks that round fifty messages at a call,
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace App\Domain\DTO\Mail;
|
||||
|
||||
/**
|
||||
* Where a Gmail import stands against Google's quota, for the line that shows
|
||||
* the import.
|
||||
*
|
||||
* An import of a large mailbox is as fast as the quota lets it be and no
|
||||
* faster, so "why is this taking so long" and "is it stuck" are both
|
||||
* questions about the quota. Before this the line had one word for all of it
|
||||
* — "waiting" — whether the import was between two listings, held for a
|
||||
* minute by Google, or had stopped (#90).
|
||||
*
|
||||
* Every figure here is this application's own reckoning. Google reports
|
||||
* neither what is left of a quota nor, until it refuses a request, what the
|
||||
* quota is. See GmailQuotaPacer.
|
||||
*/
|
||||
final readonly class GmailQuotaStatus
|
||||
{
|
||||
/**
|
||||
* @param int $limit units a minute the import is being paced to
|
||||
* @param bool $learnt whether Google named that limit, or it is
|
||||
* the one assumed until it does
|
||||
* @param int $used roughly what the bulk fetching spent in
|
||||
* the minute just gone
|
||||
* @param int $pausedSeconds how long Google's last refusal still
|
||||
* holds the account; 0 when it does not
|
||||
* @param int|null $minutesLeft how long the rest of the import takes at
|
||||
* the paced rate, or null when there is no
|
||||
* total to reckon it from
|
||||
*/
|
||||
public function __construct(
|
||||
public int $limit,
|
||||
public bool $learnt,
|
||||
public int $used,
|
||||
public int $pausedSeconds,
|
||||
public ?int $minutesLeft,
|
||||
) {}
|
||||
|
||||
public function isPaused(): bool
|
||||
{
|
||||
return $this->pausedSeconds > 0;
|
||||
}
|
||||
}
|
||||
@@ -31,6 +31,8 @@ final readonly class ImportLine
|
||||
* @param bool $waiting nothing has moved for a while: between two
|
||||
* listings, or because something is wrong —
|
||||
* the account's own health says which
|
||||
* @param GmailQuotaStatus|null $quota what Google's quota allows this
|
||||
* import, for a Gmail account
|
||||
*/
|
||||
public function __construct(
|
||||
public Account $account,
|
||||
@@ -39,6 +41,7 @@ final readonly class ImportLine
|
||||
public ?Mailbox $folder,
|
||||
public bool $waiting,
|
||||
public ?Label $label = null,
|
||||
public ?GmailQuotaStatus $quota = null,
|
||||
) {}
|
||||
|
||||
/** Whole percent, floored and never past 100; 0 until there is a total. */
|
||||
|
||||
@@ -4,6 +4,7 @@ declare(strict_types=1);
|
||||
|
||||
namespace App\Infrastructure\Messaging\Handler;
|
||||
|
||||
use App\Domain\Exception\GmailThrottledException;
|
||||
use App\Domain\Exception\OAuthGrantRevokedException;
|
||||
use App\Domain\Interface\AccountSyncerInterface;
|
||||
use App\Entity\Mail\Account;
|
||||
@@ -17,6 +18,8 @@ use Doctrine\ORM\EntityManagerInterface;
|
||||
use Psr\Log\LoggerInterface;
|
||||
use Symfony\Component\DependencyInjection\Attribute\AutowireIterator;
|
||||
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
|
||||
use Symfony\Component\Messenger\MessageBusInterface;
|
||||
use Symfony\Component\Messenger\Stamp\DelayStamp;
|
||||
use Throwable;
|
||||
|
||||
#[AsMessageHandler]
|
||||
@@ -35,6 +38,7 @@ final readonly class SyncAccountMessageHandler
|
||||
private iterable $syncers,
|
||||
private SyncOrigin $origin,
|
||||
private MailImporter $importer,
|
||||
private MessageBusInterface $bus,
|
||||
) {}
|
||||
|
||||
public function __invoke(SyncAccountMessage $message): void
|
||||
@@ -120,6 +124,42 @@ final readonly class SyncAccountMessageHandler
|
||||
'accountId' => $message->accountId,
|
||||
]);
|
||||
|
||||
return;
|
||||
} catch (GmailThrottledException $throttled) {
|
||||
// NOT A FAILURE, and so neither recorded on the account nor
|
||||
// rethrown.
|
||||
//
|
||||
// Google has refused this account on quota, which while a mailbox
|
||||
// is being imported is expected — it is how the import learns
|
||||
// what the quota is (see GmailQuotaPacer). Rethrown, it was a
|
||||
// warning with two stack traces on each of five retries a minute
|
||||
// apart, and then the sync was dropped: twenty of those an hour
|
||||
// on an install that was working as designed (#90).
|
||||
//
|
||||
// The same sync is asked for again once the refusal has cleared,
|
||||
// carrying the moment it was first asked for, so a sync that got
|
||||
// through in the meantime makes it redundant like any other.
|
||||
$seconds = max(1, $throttled->getRetryAfterSeconds() ?? 60);
|
||||
|
||||
$this->bus->dispatch(
|
||||
new SyncAccountMessage($message->accountId, $message->trigger ?? null, $message->requestedAt ?? null),
|
||||
[new DelayStamp(1000 * $seconds)],
|
||||
);
|
||||
|
||||
$this->logger->info('Sync put back: waiting for Gmail quota', [
|
||||
'accountId' => $message->accountId,
|
||||
'delaySeconds' => $seconds,
|
||||
]);
|
||||
|
||||
// And the import is still looked after. This used to be reached
|
||||
// only by a sync that went through, so an import whose chain had
|
||||
// stopped stayed stopped for as long as the syncs were refused.
|
||||
$account = $this->accountRepository->find($message->accountId);
|
||||
|
||||
if (null !== $account) {
|
||||
$this->importer->ensureRunning($account);
|
||||
}
|
||||
|
||||
return;
|
||||
} catch (Throwable $failure) {
|
||||
// Recorded and RETHROWN. The rethrow is what keeps Messenger's
|
||||
|
||||
@@ -17,6 +17,7 @@ use App\Repository\Mail\MessageRepository;
|
||||
use App\Service\Gmail\GmailAddressFilter;
|
||||
use App\Service\Gmail\GmailApiSyncer;
|
||||
use App\Service\Gmail\GmailMessageBuilder;
|
||||
use App\Service\Gmail\GmailQuotaPacer;
|
||||
use App\Service\HarvestContactsService;
|
||||
use App\Service\Label\ThreadLabelSynchronizer;
|
||||
use App\Service\Imap\MessageThreader;
|
||||
@@ -36,7 +37,11 @@ use Symfony\Component\Messenger\Stamp\TransportNamesStamp;
|
||||
#[AsMessageHandler]
|
||||
final readonly class SyncGmailMessageBatchHandler
|
||||
{
|
||||
/** Delay before re-fetching sub-requests that hit the rate limit. */
|
||||
/**
|
||||
* Delay before re-fetching sub-requests that failed for a reason other
|
||||
* than quota — a 5xx, a part missing from the response. One refused on
|
||||
* quota waits as long as the pacer says the account is being left alone.
|
||||
*/
|
||||
private const int RETRY_DELAY_MS = 30000;
|
||||
|
||||
public function __construct(
|
||||
@@ -57,6 +62,7 @@ final readonly class SyncGmailMessageBatchHandler
|
||||
private MessageCategorizer $categorizer,
|
||||
private MessageThreader $messageThreader,
|
||||
private SyncOrigin $origin,
|
||||
private GmailQuotaPacer $pacer,
|
||||
) {}
|
||||
|
||||
public function __invoke(SyncGmailMessageBatchMessage $message): void
|
||||
@@ -104,6 +110,17 @@ final readonly class SyncGmailMessageBatchHandler
|
||||
return;
|
||||
}
|
||||
|
||||
// Google has just refused this account, here or in another process.
|
||||
// Asking now is a request that will be refused too and counted
|
||||
// against the minute all the same, so the batch is put back whole to
|
||||
// come round when the refusal has cleared. Not an error and not a
|
||||
// retry: nothing went wrong, and there is no ladder to climb.
|
||||
if ($this->pacer->waitSeconds($account) > 0) {
|
||||
$this->putBack($account, $toFetch);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
$fetch = $this->apiClient->getMessages($account, $toFetch);
|
||||
$payloads = $fetch['payloads'];
|
||||
|
||||
@@ -124,21 +141,7 @@ final readonly class SyncGmailMessageBatchHandler
|
||||
'count' => count($fetch['retryable']),
|
||||
]);
|
||||
|
||||
// Back onto the queue this kind of work belongs on. Left to the
|
||||
// routing table a retry lands on the live queue, and while an
|
||||
// account is importing that is its history jumping in front of
|
||||
// the mail arriving now — a few ids at a time, on every batch
|
||||
// Google throttled, which during an import is most of them.
|
||||
$stamps = [new DelayStamp(self::RETRY_DELAY_MS)];
|
||||
|
||||
if (true === $account->needsBackfill()) {
|
||||
$stamps[] = new TransportNamesStamp([GmailApiSyncer::IMPORT_QUEUE]);
|
||||
}
|
||||
|
||||
$this->bus->dispatch(
|
||||
new SyncGmailMessageBatchMessage($account->id, $fetch['retryable'], $this->origin->current()),
|
||||
$stamps,
|
||||
);
|
||||
$this->putBack($account, $fetch['retryable']);
|
||||
}
|
||||
|
||||
/** @var list<array{message: Message, account: Account}> $built */
|
||||
@@ -276,6 +279,45 @@ final readonly class SyncGmailMessageBatchHandler
|
||||
|
||||
// ── Private ───────────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Queue these ids to be fetched later: after the account's quota refusal
|
||||
* has cleared, if it is under one, and after a short pause otherwise.
|
||||
*
|
||||
* A few seconds are added to a quota wait, and not the same few for every
|
||||
* batch. An import has dozens of batches queued behind the one that was
|
||||
* refused, each of which is put back in turn; all given the same delay
|
||||
* they would all come due in the same second.
|
||||
*
|
||||
* Back onto the queue this kind of work belongs on. Left to the routing
|
||||
* table it lands on the live queue, and while an account is importing
|
||||
* that is its history jumping in front of the mail arriving now.
|
||||
*
|
||||
* @param list<string> $gmailIds
|
||||
*/
|
||||
private function putBack(Account $account, array $gmailIds): void
|
||||
{
|
||||
$wait = $this->pacer->waitSeconds($account);
|
||||
$delay = $wait > 0 ? 1000 * ($wait + random_int(1, 15)) : self::RETRY_DELAY_MS;
|
||||
|
||||
$stamps = [new DelayStamp($delay)];
|
||||
|
||||
if (true === $account->needsBackfill()) {
|
||||
$stamps[] = new TransportNamesStamp([GmailApiSyncer::IMPORT_QUEUE]);
|
||||
}
|
||||
|
||||
$this->bus->dispatch(
|
||||
new SyncGmailMessageBatchMessage((int) $account->id, $gmailIds, $this->origin->current()),
|
||||
$stamps,
|
||||
);
|
||||
|
||||
$this->logger->info('SyncGmailMessageBatch: put back to be fetched later', [
|
||||
'accountId' => $account->id,
|
||||
'count' => count($gmailIds),
|
||||
'delaySeconds' => intdiv($delay, 1000),
|
||||
'quota' => $wait > 0,
|
||||
]);
|
||||
}
|
||||
|
||||
/**
|
||||
* Merge the Gmail copy's knowledge onto an existing row: gmailId (covers it
|
||||
* under the user-scoped dedup from now on), gmailLabelIds, the inbox
|
||||
|
||||
@@ -33,8 +33,13 @@ final class GmailApiSyncer
|
||||
/** Message-list page size (API max is 500). */
|
||||
private const int PAGE_SIZE = 500;
|
||||
|
||||
/** Gmail message IDs per fan-out batch. */
|
||||
private const int BATCH_SIZE = 100;
|
||||
/**
|
||||
* Gmail message IDs per fan-out batch. Fifty is the most Google asks to
|
||||
* be sent in one batch request, and at twenty units a message it is a
|
||||
* thousand units — a sixth of the minute on a project with the lowest
|
||||
* limit, where a hundred was a third of it in one request.
|
||||
*/
|
||||
private const int BATCH_SIZE = 50;
|
||||
|
||||
/** How many of the Inbox's newest messages an import asks for by name, first. */
|
||||
private const int INBOX_SLICE = 200;
|
||||
|
||||
@@ -5,77 +5,139 @@ declare(strict_types=1);
|
||||
namespace App\Service\Gmail;
|
||||
|
||||
use App\Entity\Mail\Account;
|
||||
use Psr\Cache\CacheItemPoolInterface;
|
||||
use Symfony\Component\Cache\Adapter\ArrayAdapter;
|
||||
use Symfony\Component\Clock\ClockInterface;
|
||||
use Symfony\Component\DependencyInjection\Attribute\Autowire;
|
||||
|
||||
/**
|
||||
* Spaces out the bulk Gmail calls so an import stays inside the account's
|
||||
* quota instead of finding its edge by being refused.
|
||||
* quota, and finds out what that quota is by being told.
|
||||
*
|
||||
* Google allows one Gmail user 15,000 quota units a minute, across everything
|
||||
* done in that user's name. Fetching a message costs 5, so a batch of a
|
||||
* hundred is 500 and thirty batches are the whole minute. Nothing used to
|
||||
* count: the worker fetched one batch after another as fast as Google
|
||||
* answered. While the same worker also did the follow-up work for every
|
||||
* message, that was slow enough to stay under the limit by accident. Once the
|
||||
* ingest queue held nothing but fetching, an import ran into it — and
|
||||
* the call that reported it was not the import. A batch that is partly refused
|
||||
* re-queues the refused ids at `info` and carries on; what failed out loud was
|
||||
* the next "see what is new" sync of the same account, on labels.list, a call
|
||||
* that costs one unit. New mail was held up by old mail being fetched too
|
||||
* fast.
|
||||
* Google allows one Gmail user a number of quota units a minute, across
|
||||
* everything done in that user's name. Nothing used to count: the worker
|
||||
* fetched one batch after another as fast as Google answered, an import ran
|
||||
* into the limit — and the call that reported it was not the import but the
|
||||
* next "see what is new" sync of the same account, on labels.list, a call that
|
||||
* costs one unit. New mail was held up by old mail being fetched too fast.
|
||||
*
|
||||
* So this keeps a rate rather than reacting to a refusal. Each account has a
|
||||
* steady allowance and a small reserve on top of it for a burst; a caller says
|
||||
* what it is about to spend, and is made to wait first if that would run ahead
|
||||
* of the allowance. The wait is a few seconds at most for one batch, which is
|
||||
* why it is a sleep in the handler and not a delayed redelivery: the queue
|
||||
* behind it is no worse off than behind any other batch, and nothing has to
|
||||
* be re-sent.
|
||||
* So this keeps a rate. Each account has a steady allowance and a small
|
||||
* reserve on top of it for a burst; a caller says what it is about to spend,
|
||||
* and is made to wait first if that would run ahead of the allowance. The wait
|
||||
* is a few seconds at most for one batch, which is why it is a sleep in the
|
||||
* handler and not a delayed redelivery.
|
||||
*
|
||||
* WELL UNDER the limit on purpose — the worst minute here is 12,000 units of
|
||||
* the 15,000. The rest belongs to what is not counted: the person using the
|
||||
* mailbox (a send is 100, a label change on a selection 50), push-driven
|
||||
* syncs, and whatever else is signed in to the same Google account.
|
||||
* ## The limit is learnt, not configured
|
||||
*
|
||||
* IN MEMORY, in this process. That is enough because the bulk fetching is one
|
||||
* worker's job — worker-backlog, which runs an import's listing and its
|
||||
* batches; worker-ingest also talks to Gmail, but for the handful of calls a
|
||||
* "see what is new" sync makes, and those are what the headroom above is left
|
||||
* for. It survives from one message to the next because a worker only resets
|
||||
* services that ask to be. A restart forgets what was spent and
|
||||
* starts with a full reserve, which can overshoot for at most the reserve.
|
||||
* Shared storage would close that, at the price of a database round trip per
|
||||
* batch to guard against something GmailThrottledException already handles.
|
||||
* The first version of this had the limit written into it: 15,000 units a
|
||||
* minute, with a message costing 5. Both were Google's numbers and both
|
||||
* stopped being true on 1 May 2026 — a project created since then gets 6,000,
|
||||
* and a message costs 20 for everybody. The pacer went on allowing what it
|
||||
* believed was 9,000 units a minute and was really 36,000, and an import on a
|
||||
* new project was refused for most of what it asked (#90).
|
||||
*
|
||||
* The limit cannot be asked for: it belongs to the Google Cloud project, and
|
||||
* reading it takes a Cloud permission no mailbox grant carries. A setting
|
||||
* would be a number every self-hoster has to look up in a console and keep in
|
||||
* step with Google. But a refusal says it — the 403 names the limit it was
|
||||
* refused under and its value — so the pacer starts from the highest limit
|
||||
* there is, is refused once on a project that has less, and keeps to what it
|
||||
* was told from then on. A refusal is therefore not a fault here. It is how
|
||||
* the number arrives, and everything that calls Gmail in bulk treats it as
|
||||
* "not now" rather than as an error: see refused() and waitSeconds().
|
||||
*
|
||||
* FORGOTTEN AFTER A DAY, on purpose. A limit only ever learnt downwards would
|
||||
* hold an install to 6,000 for ever after Google or its administrator raised
|
||||
* it. Starting from the top again costs one refused batch a day on a project
|
||||
* that has less, and that batch is put back like any other.
|
||||
*
|
||||
* WELL UNDER the limit — the worst minute here is four fifths of it. The rest
|
||||
* belongs to what is not counted: the person using the mailbox (a send is
|
||||
* 100, a label change on a selection 50), push-driven syncs, and whatever
|
||||
* else is signed in to the same Google account.
|
||||
*
|
||||
* ## Shared between processes
|
||||
*
|
||||
* What an account has spent and what the project's limit is are kept in a
|
||||
* cache pool the database backs, not in this object. It was in memory once,
|
||||
* on the reasoning that bulk fetching is one worker's job. It is two workers'
|
||||
* — worker-backlog fetches an import's batches, worker-ingest fetches the
|
||||
* mail arriving meanwhile — and each kept its own count of the same quota.
|
||||
* More to the point, a refusal one worker met has to stop the other, and the
|
||||
* web request that is about to ask as well. One read and one write per batch
|
||||
* is the price, against a batch that is fifty requests to Google.
|
||||
*
|
||||
* Two workers spending in the same instant can each read the other's stale
|
||||
* figure and both go ahead. That overshoots by one batch, inside the fifth
|
||||
* left spare, and is not worth a lock.
|
||||
*/
|
||||
final class GmailQuotaPacer
|
||||
{
|
||||
/** What a messages.get costs, and so what each id in a batch costs. */
|
||||
public const int UNITS_PER_MESSAGE = 5;
|
||||
public const int UNITS_PER_MESSAGE = 20;
|
||||
|
||||
/** What one page of messages.list costs. */
|
||||
public const int UNITS_PER_LIST_PAGE = 5;
|
||||
|
||||
/** The steady allowance: 9,000 units a minute. */
|
||||
private const float UNITS_PER_SECOND = 150.0;
|
||||
/**
|
||||
* Units a minute for one user, before Google has said otherwise: what a
|
||||
* project that used the API before May 2026 was left with, and the most
|
||||
* any project has without asking for more.
|
||||
*/
|
||||
public const int HIGHEST_LIMIT = 15_000;
|
||||
|
||||
/** How far ahead of the allowance a burst may run: 3,000 units. */
|
||||
/**
|
||||
* What a project created since May 2026 has. Fallen back to when a
|
||||
* refusal names the per-user limit and leaves out its value.
|
||||
*/
|
||||
public const int CURRENT_DEFAULT_LIMIT = 6_000;
|
||||
|
||||
/** The share of the limit that is the steady allowance. */
|
||||
private const float STEADY_SHARE = 0.6;
|
||||
|
||||
/** How far ahead of the allowance a burst may run: a fifth of the limit. */
|
||||
private const float RESERVE_SECONDS = 20.0;
|
||||
|
||||
/**
|
||||
* Per account id, the moment its spending so far is paid off at the
|
||||
* steady rate. In the past for an account that has been idle.
|
||||
*
|
||||
* @var array<int, float>
|
||||
* How long an account is left alone after a refusal that came with no
|
||||
* Retry-After, which is the norm. The limit is counted over a minute, so
|
||||
* a minute later the whole of it is back.
|
||||
*/
|
||||
private array $paidOffAt = [];
|
||||
private const int COOLDOWN_SECONDS = 60;
|
||||
|
||||
/** The longest a Retry-After is believed for. */
|
||||
private const int LONGEST_COOLDOWN_SECONDS = 900;
|
||||
|
||||
/** How long a learnt limit is kept before the pacer starts from the top again. */
|
||||
private const int LIMIT_LIFETIME_SECONDS = 86_400;
|
||||
|
||||
/**
|
||||
* The width of one slot of the record of what was spent when. Six of them
|
||||
* are the minute Google counts over; see usedThisMinute().
|
||||
*/
|
||||
private const int SLOT_SECONDS = 10;
|
||||
|
||||
private const string LIMIT_KEY = 'gmail_quota.limit';
|
||||
|
||||
private readonly CacheItemPoolInterface $state;
|
||||
|
||||
public function __construct(
|
||||
private readonly ClockInterface $clock,
|
||||
// Null in the tests that build this by hand and have one process to
|
||||
// share with: an array is all the sharing they need.
|
||||
#[Autowire(service: 'app.gmail_quota')]
|
||||
?CacheItemPoolInterface $state = null,
|
||||
) {
|
||||
$this->state = $state ?? new ArrayAdapter();
|
||||
}
|
||||
|
||||
/**
|
||||
* Announce a spend, and wait here until the account can afford it.
|
||||
*
|
||||
* Only ever a short wait. An account Google has just refused is not
|
||||
* waited on here — that is a minute, and a worker asleep for a minute is
|
||||
* every other account's mail not being fetched. The caller asks
|
||||
* waitSeconds() first and puts its job back instead.
|
||||
*/
|
||||
public function spend(Account $account, int $units): void
|
||||
{
|
||||
@@ -83,14 +145,23 @@ final class GmailQuotaPacer
|
||||
return;
|
||||
}
|
||||
|
||||
$key = (int) $account->id;
|
||||
$now = (float) $this->clock->now()->format('U.u');
|
||||
$now = $this->now();
|
||||
$item = $this->state->getItem($this->accountKey($account));
|
||||
$state = $this->read($item->get());
|
||||
|
||||
$paidOffAt = max($this->paidOffAt[$key] ?? 0.0, $now) + $units / self::UNITS_PER_SECOND;
|
||||
$paidOffAt = max($state['paidOffAt'], $now) + $units / $this->unitsPerSecond();
|
||||
|
||||
// Recorded before the wait, not after: what was spent is spent whether
|
||||
// or not the sleep is interrupted by the worker being stopped.
|
||||
$this->paidOffAt[$key] = $paidOffAt;
|
||||
$state['paidOffAt'] = $paidOffAt;
|
||||
|
||||
// Against the moment the call will actually be made, which is after
|
||||
// the wait when there is one.
|
||||
$slot = $this->slot(max($now, $paidOffAt - self::RESERVE_SECONDS));
|
||||
$state['spent'][$slot] = ($state['spent'][$slot] ?? 0) + $units;
|
||||
$state['spent'] = $this->sinceAMinuteAgo($state['spent'], $now);
|
||||
|
||||
$this->state->save($item->set($state)->expiresAfter(self::LIMIT_LIFETIME_SECONDS));
|
||||
|
||||
$ahead = $paidOffAt - $now - self::RESERVE_SECONDS;
|
||||
|
||||
@@ -98,4 +169,163 @@ final class GmailQuotaPacer
|
||||
$this->clock->sleep($ahead);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Google refused a call for this account on quota.
|
||||
*
|
||||
* Two things follow. The account is left alone until the refusal has had
|
||||
* time to clear, by every process — see waitSeconds(). And if the refusal
|
||||
* was under the per-user limit, the limit it named becomes the one every
|
||||
* account is paced to, since it is the project's and not the mailbox's.
|
||||
*
|
||||
* @param string $limitName Google's name for the limit, e.g. "…PerMinutePerUser"; '' when the body had none
|
||||
* @param int|null $limitValue units a minute, when the body said
|
||||
* @param int|null $retryAfterSeconds the Retry-After header, when there was one
|
||||
*/
|
||||
public function refused(Account $account, string $limitName = '', ?int $limitValue = null, ?int $retryAfterSeconds = null): void
|
||||
{
|
||||
$now = $this->now();
|
||||
$cooldown = min(self::LONGEST_COOLDOWN_SECONDS, max(1, $retryAfterSeconds ?? self::COOLDOWN_SECONDS));
|
||||
|
||||
$item = $this->state->getItem($this->accountKey($account));
|
||||
$state = $this->read($item->get());
|
||||
|
||||
$state['refusedUntil'] = max($state['refusedUntil'], $now + $cooldown);
|
||||
// The reserve is gone as well: whatever was thought to be left of this
|
||||
// minute, Google has just said there is none.
|
||||
$state['paidOffAt'] = max($state['paidOffAt'], $now + self::RESERVE_SECONDS);
|
||||
|
||||
$this->state->save($item->set($state)->expiresAfter(self::LIMIT_LIFETIME_SECONDS));
|
||||
|
||||
// The per-project limit is two hundred times this one and says nothing
|
||||
// about how fast a single mailbox may be read.
|
||||
if (false === str_contains(strtolower($limitName), 'peruser')) {
|
||||
return;
|
||||
}
|
||||
|
||||
$current = $this->limit();
|
||||
$learnt = $limitValue ?? self::CURRENT_DEFAULT_LIMIT;
|
||||
|
||||
// Downwards only. A refusal under a limit higher than the one being
|
||||
// kept to is another client spending the same user's quota, not news
|
||||
// about the limit.
|
||||
if ($learnt > 0 && $learnt < $current) {
|
||||
$limit = $this->state->getItem(self::LIMIT_KEY);
|
||||
$this->state->save($limit->set($learnt)->expiresAfter(self::LIMIT_LIFETIME_SECONDS));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* How long to leave this account alone, in whole seconds; 0 when it may
|
||||
* be asked now.
|
||||
*
|
||||
* What a job asks before it calls Gmail in bulk, and what it delays itself
|
||||
* by when the answer is not 0.
|
||||
*/
|
||||
public function waitSeconds(Account $account): int
|
||||
{
|
||||
$state = $this->read($this->state->getItem($this->accountKey($account))->get());
|
||||
|
||||
return max(0, (int) ceil($state['refusedUntil'] - $this->now()));
|
||||
}
|
||||
|
||||
/**
|
||||
* Roughly what this account has spent in the minute just gone, in units.
|
||||
*
|
||||
* AN ESTIMATE, and shown as one. Google does not report what is left of a
|
||||
* quota, so this is this application's own count of what it announced to
|
||||
* spend() — the bulk fetching, which is nearly all of an import and none
|
||||
* of what the person or another client did in the same mailbox. It is
|
||||
* also counted in ten-second slots, so it runs up to ten seconds stale.
|
||||
*/
|
||||
public function usedThisMinute(Account $account): int
|
||||
{
|
||||
$now = $this->now();
|
||||
$state = $this->read($this->state->getItem($this->accountKey($account))->get());
|
||||
|
||||
return array_sum(array_filter(
|
||||
$this->sinceAMinuteAgo($state['spent'], $now),
|
||||
// A spend that was made to wait is recorded a little ahead, and
|
||||
// counts once its moment has come.
|
||||
static fn (int $slot): bool => $slot <= $now,
|
||||
ARRAY_FILTER_USE_KEY,
|
||||
));
|
||||
}
|
||||
|
||||
/** Whether the limit is one Google named, rather than the one assumed until it does. */
|
||||
public function isLimitLearnt(): bool
|
||||
{
|
||||
$limit = $this->state->getItem(self::LIMIT_KEY)->get();
|
||||
|
||||
return true === is_int($limit) && $limit > 0;
|
||||
}
|
||||
|
||||
/** How many messages a minute an import fetches at the steady allowance. */
|
||||
public function messagesPerMinute(): int
|
||||
{
|
||||
return max(1, (int) floor($this->unitsPerSecond() * 60.0 / self::UNITS_PER_MESSAGE));
|
||||
}
|
||||
|
||||
/** Units a minute one user is being held to: Google's last word, or the highest there is. */
|
||||
public function limit(): int
|
||||
{
|
||||
$limit = $this->state->getItem(self::LIMIT_KEY)->get();
|
||||
|
||||
return true === is_int($limit) && $limit > 0 ? $limit : self::HIGHEST_LIMIT;
|
||||
}
|
||||
|
||||
private function unitsPerSecond(): float
|
||||
{
|
||||
return $this->limit() * self::STEADY_SHARE / 60.0;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return array{paidOffAt: float, refusedUntil: float, spent: array<int, int>}
|
||||
* the moment the account's spending so far is paid off at the
|
||||
* steady rate, and the moment a refusal stops applying — both in
|
||||
* the past for an account that has been idle — and what was spent
|
||||
* in each recent slot, keyed by the second the slot began
|
||||
*/
|
||||
private function read(mixed $stored): array
|
||||
{
|
||||
$spent = true === is_array($stored) && true === is_array($stored['spent'] ?? null) ? $stored['spent'] : [];
|
||||
|
||||
return [
|
||||
'paidOffAt' => (float) (true === is_array($stored) ? ($stored['paidOffAt'] ?? 0.0) : 0.0),
|
||||
'refusedUntil' => (float) (true === is_array($stored) ? ($stored['refusedUntil'] ?? 0.0) : 0.0),
|
||||
'spent' => array_map(intval(...), $spent),
|
||||
];
|
||||
}
|
||||
|
||||
private function slot(float $at): int
|
||||
{
|
||||
return intdiv((int) $at, self::SLOT_SECONDS) * self::SLOT_SECONDS;
|
||||
}
|
||||
|
||||
/**
|
||||
* The slots a minute ending now has not yet left behind, those still to
|
||||
* come included.
|
||||
*
|
||||
* @param array<int, int> $spent
|
||||
*
|
||||
* @return array<int, int>
|
||||
*/
|
||||
private function sinceAMinuteAgo(array $spent, float $now): array
|
||||
{
|
||||
return array_filter(
|
||||
$spent,
|
||||
static fn (int $slot): bool => $slot > $now - 60.0,
|
||||
ARRAY_FILTER_USE_KEY,
|
||||
);
|
||||
}
|
||||
|
||||
private function accountKey(Account $account): string
|
||||
{
|
||||
return 'gmail_quota.account.' . (int) $account->id;
|
||||
}
|
||||
|
||||
private function now(): float
|
||||
{
|
||||
return (float) $this->clock->now()->format('U.u');
|
||||
}
|
||||
}
|
||||
@@ -124,7 +124,7 @@ final class GmailApiClient
|
||||
'query' => $params,
|
||||
]);
|
||||
|
||||
$body = $this->decode($response, 'messages.list');
|
||||
$body = $this->decode($account, $response, 'messages.list');
|
||||
$next = $body['nextPageToken'] ?? null;
|
||||
|
||||
return [
|
||||
@@ -136,7 +136,7 @@ final class GmailApiClient
|
||||
/**
|
||||
* Fetch multiple messages (format=full) using the Gmail Batch API.
|
||||
*
|
||||
* Packs up to 100 individual messages.get sub-requests into a single
|
||||
* Packs up to 50 individual messages.get sub-requests into a single
|
||||
* multipart/mixed HTTP POST. This avoids hammering the per-user per-second
|
||||
* quota with individual concurrent requests and dramatically reduces
|
||||
* round-trips for large initial syncs.
|
||||
@@ -151,7 +151,12 @@ final class GmailApiClient
|
||||
* empty would let the Messenger message ack and silently drop every id
|
||||
* in the batch.
|
||||
*
|
||||
* @param list<string> $messageIds Maximum 100 per call (enforced by caller via BATCH_SIZE)
|
||||
* EXCEPT A QUOTA REFUSAL, which is not a failure. Whether Google refuses
|
||||
* the whole request or some of its parts, the answer is the same: those
|
||||
* ids come back as `retryable`, and the pacer has been told, so the
|
||||
* caller can ask it how long to wait before trying them again.
|
||||
*
|
||||
* @param list<string> $messageIds Maximum 50 per call (enforced by caller via BATCH_SIZE)
|
||||
* @return array{payloads: array<string,array<string,mixed>>, retryable: list<string>, gone: list<string>}
|
||||
*/
|
||||
public function getMessages(Account $account, array $messageIds): array
|
||||
@@ -181,7 +186,17 @@ final class GmailApiClient
|
||||
]);
|
||||
|
||||
try {
|
||||
// Classified first, so the whole request being refused on quota
|
||||
// is recognised as that and not as "the batch request failed".
|
||||
$this->assertSuccess($account, $response, 'messages.get(batch)');
|
||||
|
||||
$rawBody = $response->getContent();
|
||||
} catch (GmailThrottledException) {
|
||||
return [
|
||||
'payloads' => [],
|
||||
'retryable' => array_values($messageIds),
|
||||
'gone' => [],
|
||||
];
|
||||
} catch (HttpException $e) {
|
||||
throw new \RuntimeException(
|
||||
'Gmail batch request failed: ' . $e->getMessage(),
|
||||
@@ -191,6 +206,11 @@ final class GmailApiClient
|
||||
}
|
||||
|
||||
$parsed = $this->parseBatchResponse($rawBody);
|
||||
|
||||
if (null !== $parsed['refusal']) {
|
||||
$this->pacer->refused($account, $parsed['refusal']['limitName'], $parsed['refusal']['limitValue']);
|
||||
}
|
||||
|
||||
$payloads = $parsed['payloads'];
|
||||
$retryable = [];
|
||||
$gone = [];
|
||||
@@ -242,7 +262,7 @@ final class GmailApiClient
|
||||
],
|
||||
);
|
||||
|
||||
return $this->decode($response, 'messages.get');
|
||||
return $this->decode($account, $response, 'messages.get');
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -265,7 +285,7 @@ final class GmailApiClient
|
||||
],
|
||||
);
|
||||
|
||||
$body = $this->decode($response, 'messages.get(raw)');
|
||||
$body = $this->decode($account, $response, 'messages.get(raw)');
|
||||
$raw = (string) ($body['raw'] ?? '');
|
||||
|
||||
if ('' === $raw) {
|
||||
@@ -303,7 +323,7 @@ final class GmailApiClient
|
||||
'query' => $query,
|
||||
]);
|
||||
|
||||
$body = $this->decode($response, 'history.list');
|
||||
$body = $this->decode($account, $response, 'history.list');
|
||||
$page = $body['nextPageToken'] ?? null;
|
||||
|
||||
if (true === isset($body['historyId'])) {
|
||||
@@ -339,7 +359,7 @@ final class GmailApiClient
|
||||
],
|
||||
]);
|
||||
|
||||
return $this->decode($response, 'watch');
|
||||
return $this->decode($account, $response, 'watch');
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -367,7 +387,7 @@ final class GmailApiClient
|
||||
'auth_bearer' => $token,
|
||||
]);
|
||||
|
||||
return $this->decode($response, 'getProfile');
|
||||
return $this->decode($account, $response, 'getProfile');
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -430,7 +450,7 @@ final class GmailApiClient
|
||||
['auth_bearer' => $token],
|
||||
);
|
||||
|
||||
$body = $this->decode($response, 'attachments.get');
|
||||
$body = $this->decode($account, $response, 'attachments.get');
|
||||
$data = (string) ($body['data'] ?? '');
|
||||
|
||||
return base64_decode(strtr($data, '-_', '+/'));
|
||||
@@ -477,7 +497,10 @@ final class GmailApiClient
|
||||
* by a JSON body. 200 parts are decoded into payloads; every part's
|
||||
* status is recorded so the caller can classify failures per id.
|
||||
*
|
||||
* @return array{payloads: array<string,array<string,mixed>>, statuses: array<string,int>}
|
||||
* `refusal` is the first part Google refused on quota, read the way a
|
||||
* refused response of its own would be; null when no part was.
|
||||
*
|
||||
* @return array{payloads: array<string,array<string,mixed>>, statuses: array<string,int>, refusal: ?array{limitName: string, limitValue: ?int}}
|
||||
*/
|
||||
private function parseBatchResponse(string $rawBody): array
|
||||
{
|
||||
@@ -490,12 +513,14 @@ final class GmailApiClient
|
||||
return [
|
||||
'payloads' => [],
|
||||
'statuses' => [],
|
||||
'refusal' => null,
|
||||
];
|
||||
}
|
||||
|
||||
$boundary = $m[1];
|
||||
$payloads = [];
|
||||
$statuses = [];
|
||||
$refusal = null;
|
||||
|
||||
// Split on the boundary lines, drop the preamble and epilogue.
|
||||
$parts = preg_split('/\r?\n--' . preg_quote($boundary, '/') . '(?:--)?(?:\r?\n|$)/', $rawBody);
|
||||
@@ -504,6 +529,7 @@ final class GmailApiClient
|
||||
return [
|
||||
'payloads' => [],
|
||||
'statuses' => [],
|
||||
'refusal' => null,
|
||||
];
|
||||
}
|
||||
|
||||
@@ -547,11 +573,20 @@ final class GmailApiClient
|
||||
if (null !== $contentId && null !== $status) {
|
||||
$statuses[$contentId] = $status;
|
||||
}
|
||||
|
||||
if (null === $refusal && true === in_array($status, [403, 429], true)) {
|
||||
$failure = $this->describeErrorBody($json);
|
||||
|
||||
if (429 === $status || true === in_array($failure['reason'], self::TRANSIENT_REASONS, true)) {
|
||||
$refusal = ['limitName' => $failure['limitName'], 'limitValue' => $failure['limitValue']];
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return [
|
||||
'payloads' => $payloads,
|
||||
'statuses' => $statuses,
|
||||
'refusal' => $refusal,
|
||||
];
|
||||
}
|
||||
// ── labels ────────────────────────────────────────────────────────────────
|
||||
@@ -567,7 +602,7 @@ final class GmailApiClient
|
||||
'auth_bearer' => $token,
|
||||
]);
|
||||
|
||||
$body = $this->decode($response, 'labels.list');
|
||||
$body = $this->decode($account, $response, 'labels.list');
|
||||
|
||||
return $body['labels'] ?? [];
|
||||
}
|
||||
@@ -604,7 +639,7 @@ final class GmailApiClient
|
||||
'json' => $payload,
|
||||
]);
|
||||
|
||||
return $this->decode($response, 'labels.create');
|
||||
return $this->decode($account, $response, 'labels.create');
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -622,7 +657,7 @@ final class GmailApiClient
|
||||
'json' => ['name' => $name],
|
||||
]);
|
||||
|
||||
return $this->decode($response, 'labels.patch');
|
||||
return $this->decode($account, $response, 'labels.patch');
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -646,7 +681,7 @@ final class GmailApiClient
|
||||
return;
|
||||
}
|
||||
|
||||
$this->assertSuccess($response, 'labels.delete');
|
||||
$this->assertSuccess($account, $response, 'labels.delete');
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -702,7 +737,7 @@ final class GmailApiClient
|
||||
'json' => $payload,
|
||||
]);
|
||||
|
||||
$this->assertSuccess($response, 'messages.batchModify');
|
||||
$this->assertSuccess($account, $response, 'messages.batchModify');
|
||||
}
|
||||
}
|
||||
|
||||
@@ -743,7 +778,7 @@ final class GmailApiClient
|
||||
'json' => ['ids' => $chunk],
|
||||
]);
|
||||
|
||||
$this->assertSuccess($response, 'messages.batchDelete');
|
||||
$this->assertSuccess($account, $response, 'messages.batchDelete');
|
||||
}
|
||||
}
|
||||
|
||||
@@ -760,9 +795,9 @@ final class GmailApiClient
|
||||
*
|
||||
* @return array<string,mixed>
|
||||
*/
|
||||
private function decode(ResponseInterface $response, string $operation): array
|
||||
private function decode(Account $account, ResponseInterface $response, string $operation): array
|
||||
{
|
||||
$this->assertSuccess($response, $operation);
|
||||
$this->assertSuccess($account, $response, $operation);
|
||||
|
||||
// throw=false so this stays the only place that decides what a failure
|
||||
// means. Left at true, a status assertSuccess() deliberately let past
|
||||
@@ -783,8 +818,12 @@ final class GmailApiClient
|
||||
* did before. The status stays in the message text as well as on the
|
||||
* exception because it is the only part of a failure that always survives
|
||||
* into a log line.
|
||||
*
|
||||
* A quota rejection is also reported to the pacer before it is thrown,
|
||||
* which is what makes it known to every other job for this account and
|
||||
* not only to the one that met it. See GmailQuotaPacer::refused().
|
||||
*/
|
||||
private function assertSuccess(ResponseInterface $response, string $operation): void
|
||||
private function assertSuccess(Account $account, ResponseInterface $response, string $operation): void
|
||||
{
|
||||
$status = $response->getStatusCode();
|
||||
|
||||
@@ -806,11 +845,18 @@ final class GmailApiClient
|
||||
// 429 is in here for completeness only — Gmail answers 403 for quota,
|
||||
// which is the entire reason the reason has to be read.
|
||||
if (429 === $status || (403 === $status && true === in_array($reason, self::TRANSIENT_REASONS, true))) {
|
||||
$retryAfter = $this->retryAfterSeconds($response);
|
||||
|
||||
$this->pacer->refused($account, $failure['limitName'], $failure['limitValue'], $retryAfter);
|
||||
|
||||
// The pacer's answer rather than the header's: it is the header
|
||||
// when there was one, and otherwise how long the account is being
|
||||
// left alone, which is when a retry is worth making.
|
||||
throw new GmailThrottledException(
|
||||
$message,
|
||||
$status,
|
||||
$reason,
|
||||
$this->retryAfterSeconds($response),
|
||||
max(1, $this->pacer->waitSeconds($account)),
|
||||
);
|
||||
}
|
||||
|
||||
@@ -842,7 +888,7 @@ final class GmailApiClient
|
||||
* all of which would otherwise turn a clean 403 into a JsonException that
|
||||
* says nothing about what went wrong.
|
||||
*
|
||||
* @return array{reason: string, message: string}
|
||||
* @return array{reason: string, message: string, limitName: string, limitValue: ?int}
|
||||
*/
|
||||
private function describeFailure(ResponseInterface $response): array
|
||||
{
|
||||
@@ -851,9 +897,35 @@ final class GmailApiClient
|
||||
} catch (HttpException) {
|
||||
// The body never arrived — a truncated or reset stream. The status
|
||||
// is still worth reporting on its own.
|
||||
return ['reason' => '', 'message' => ''];
|
||||
return ['reason' => '', 'message' => '', 'limitName' => '', 'limitValue' => null];
|
||||
}
|
||||
|
||||
return $this->describeErrorBody($raw);
|
||||
}
|
||||
|
||||
/**
|
||||
* The same reading of an error body, for the two places one turns up: a
|
||||
* response of its own, and one part of a batch response.
|
||||
*
|
||||
* ## The limit, when the refusal names it
|
||||
*
|
||||
* A quota rejection carries, beside the reason, an ErrorInfo detail whose
|
||||
* metadata says which limit was hit and what it is:
|
||||
*
|
||||
* "details": [{"@type": "type.googleapis.com/google.rpc.ErrorInfo",
|
||||
* "reason": "RATE_LIMIT_EXCEEDED",
|
||||
* "metadata": {"quota_limit": "…PerMinutePerUser",
|
||||
* "quota_limit_value": "6000", …}}]
|
||||
*
|
||||
* That number is the only place this application can learn its own quota
|
||||
* from — see GmailQuotaPacer. A body without the detail still names the
|
||||
* limit in its sentence ("… and limit 'Units per minute per user' of
|
||||
* service …"), which is enough to know WHICH limit and not what it is.
|
||||
*
|
||||
* @return array{reason: string, message: string, limitName: string, limitValue: ?int}
|
||||
*/
|
||||
private function describeErrorBody(string $raw): array
|
||||
{
|
||||
$decoded = json_decode($raw, true);
|
||||
$error = true === is_array($decoded) ? ($decoded['error'] ?? null) : null;
|
||||
|
||||
@@ -861,7 +933,7 @@ final class GmailApiClient
|
||||
// Not a Google error envelope — an empty body, a truncated one, or
|
||||
// an HTML page from a proxy in front of the API. A short excerpt
|
||||
// beats discarding it: the proxy page at least names the proxy.
|
||||
return ['reason' => '', 'message' => trim(substr($raw, 0, 200))];
|
||||
return ['reason' => '', 'message' => trim(substr($raw, 0, 200)), 'limitName' => '', 'limitValue' => null];
|
||||
}
|
||||
|
||||
$errors = $error['errors'] ?? null;
|
||||
@@ -872,10 +944,38 @@ final class GmailApiClient
|
||||
// "PERMISSION_DENIED") is what newer Google endpoints send instead.
|
||||
$reason ??= $error['status'] ?? null;
|
||||
$detail = $error['message'] ?? null;
|
||||
$message = true === is_scalar($detail) ? (string) $detail : '';
|
||||
|
||||
$limitName = '';
|
||||
$limitValue = null;
|
||||
|
||||
foreach (true === is_array($error['details'] ?? null) ? $error['details'] : [] as $entry) {
|
||||
$metadata = true === is_array($entry) ? ($entry['metadata'] ?? null) : null;
|
||||
|
||||
if (false === is_array($metadata)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
if (true === is_scalar($metadata['quota_limit'] ?? null)) {
|
||||
$limitName = (string) $metadata['quota_limit'];
|
||||
}
|
||||
|
||||
if (true === is_numeric($metadata['quota_limit_value'] ?? null)) {
|
||||
$limitValue = (int) $metadata['quota_limit_value'];
|
||||
}
|
||||
}
|
||||
|
||||
if ('' === $limitName && 1 === preg_match("/limit '([^']+)'/", $message, $named)) {
|
||||
// "Units per minute per user" → "unitsperminuteperuser", so both
|
||||
// spellings answer the one question asked of them: per user or not.
|
||||
$limitName = str_replace(' ', '', $named[1]);
|
||||
}
|
||||
|
||||
return [
|
||||
'reason' => true === is_scalar($reason) ? (string) $reason : '',
|
||||
'message' => true === is_scalar($detail) ? (string) $detail : '',
|
||||
'reason' => true === is_scalar($reason) ? (string) $reason : '',
|
||||
'message' => $message,
|
||||
'limitName' => $limitName,
|
||||
'limitValue' => $limitValue,
|
||||
];
|
||||
}
|
||||
|
||||
|
||||
@@ -4,12 +4,14 @@ declare(strict_types=1);
|
||||
|
||||
namespace App\Service\Mail;
|
||||
|
||||
use App\Domain\DTO\Mail\GmailQuotaStatus;
|
||||
use App\Domain\DTO\Mail\ImportLine;
|
||||
use App\Entity\Mail\Account;
|
||||
use App\Entity\User\User;
|
||||
use App\Repository\Mail\AccountRepository;
|
||||
use App\Repository\Mail\MailboxRepository;
|
||||
use App\Repository\Mail\MessageRepository;
|
||||
use App\Service\Gmail\GmailQuotaPacer;
|
||||
use App\Service\Graph\GraphFolderResolver;
|
||||
use DateTimeImmutable;
|
||||
|
||||
@@ -45,6 +47,7 @@ final readonly class ImportProgress
|
||||
private MailboxRepository $mailboxes,
|
||||
private MessageRepository $messages,
|
||||
private GraphFolderResolver $folders,
|
||||
private GmailQuotaPacer $pacer,
|
||||
) {}
|
||||
|
||||
/** @return list<ImportLine> one per importing account, in the accounts' own order */
|
||||
@@ -80,12 +83,15 @@ final readonly class ImportProgress
|
||||
|
||||
if (true === $account->isGmail()) {
|
||||
if (true === $account->needsBackfill()) {
|
||||
$done = $this->messages->countForAccount($account);
|
||||
|
||||
$lines[] = new ImportLine(
|
||||
$account,
|
||||
$this->messages->countForAccount($account),
|
||||
$done,
|
||||
$account->importTotal,
|
||||
null,
|
||||
$this->isWaiting($account),
|
||||
quota: $this->quota($account, $done),
|
||||
);
|
||||
}
|
||||
|
||||
@@ -110,6 +116,28 @@ final readonly class ImportProgress
|
||||
return $lines;
|
||||
}
|
||||
|
||||
/**
|
||||
* What the quota allows this import, and how long the rest takes at that.
|
||||
*
|
||||
* The time is the messages still to come over the rate the pacer holds an
|
||||
* import to. It runs long towards the end: the total is Google's count of
|
||||
* the mailbox, which includes what is never stored — a draft, a chat —
|
||||
* so the last of "what is left" is mail that is not coming.
|
||||
*/
|
||||
private function quota(Account $account, int $done): GmailQuotaStatus
|
||||
{
|
||||
$total = $account->importTotal;
|
||||
$left = null === $total ? null : max(0, $total - $done);
|
||||
|
||||
return new GmailQuotaStatus(
|
||||
$this->pacer->limit(),
|
||||
$this->pacer->isLimitLearnt(),
|
||||
$this->pacer->usedThisMinute($account),
|
||||
$this->pacer->waitSeconds($account),
|
||||
null === $left || 0 === $left ? null : (int) ceil($left / $this->pacer->messagesPerMinute()),
|
||||
);
|
||||
}
|
||||
|
||||
private function isWaiting(Account $account): bool
|
||||
{
|
||||
$beatAt = $account->importBeatAt;
|
||||
|
||||
@@ -5,6 +5,7 @@ declare(strict_types=1);
|
||||
namespace App\Service\Mail;
|
||||
|
||||
use App\Domain\Enum\Mail\SyncTrigger;
|
||||
use App\Domain\Exception\GmailThrottledException;
|
||||
use App\Domain\Helper\ImapConnectionFactory;
|
||||
use App\Entity\Mail\Account;
|
||||
use App\Infrastructure\Messaging\Message\ImportMailMessage;
|
||||
@@ -12,6 +13,7 @@ use App\Repository\Mail\AccountRepository;
|
||||
use App\Repository\Mail\MailboxRepository;
|
||||
use App\Service\Gmail\GmailApiSyncer;
|
||||
use App\Service\Gmail\GmailBackfillStep;
|
||||
use App\Service\Gmail\GmailQuotaPacer;
|
||||
use App\Service\Graph\GraphApiSyncer;
|
||||
use App\Service\Graph\GraphImportStep;
|
||||
use App\Service\Imap\ImapImportOutcome;
|
||||
@@ -101,6 +103,7 @@ final readonly class MailImporter
|
||||
private MessageBusInterface $bus,
|
||||
private EntityManagerInterface $em,
|
||||
private LoggerInterface $logger,
|
||||
private GmailQuotaPacer $pacer,
|
||||
) {}
|
||||
|
||||
/** Whether this account still has history that this class would fetch. */
|
||||
@@ -162,17 +165,40 @@ final readonly class MailImporter
|
||||
$this->beat($account);
|
||||
$this->heartbeats->beatWhileBusy();
|
||||
|
||||
// Google is refusing this account just now — this page was, or a
|
||||
// batch on another worker. The page is asked for again when that has
|
||||
// cleared, which keeps the chain: it used to be retried five times a
|
||||
// minute apart and then given up on at CRITICAL, and an import whose
|
||||
// chain had gone that way waited for a sync to notice (#90).
|
||||
if (true === $account->isGmail() && $this->pacer->waitSeconds($account) > 0) {
|
||||
$this->askAgainAfterQuota($account);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
// Imported mail is marked as such, like mail a poll brought in: it is
|
||||
// not something a person was just sent.
|
||||
$wait = false;
|
||||
|
||||
$this->origin->during(SyncTrigger::Poll, function () use ($account, &$wait): void {
|
||||
$wait = match (true) {
|
||||
$account->isGmail() => $this->gmailPage($account),
|
||||
$account->isMicrosoft() => GraphImportStep::Waiting === $this->graph->importPage($account),
|
||||
default => $this->imapPage($account),
|
||||
};
|
||||
});
|
||||
try {
|
||||
$this->origin->during(SyncTrigger::Poll, function () use ($account, &$wait): void {
|
||||
$wait = match (true) {
|
||||
$account->isGmail() => $this->gmailPage($account),
|
||||
$account->isMicrosoft() => GraphImportStep::Waiting === $this->graph->importPage($account),
|
||||
default => $this->imapPage($account),
|
||||
};
|
||||
});
|
||||
} catch (GmailThrottledException) {
|
||||
// Where the listing has got to is on the account and only moves
|
||||
// when a page goes through, so the same page is asked for again.
|
||||
$account = $this->accounts->find($accountId);
|
||||
|
||||
if (null !== $account) {
|
||||
$this->askAgainAfterQuota($account);
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
// A page clears the entity manager; this is the account as it is now.
|
||||
$account = $this->accounts->find($accountId);
|
||||
@@ -203,6 +229,25 @@ final readonly class MailImporter
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* The next page, delayed until Google will answer for this account again.
|
||||
*/
|
||||
private function askAgainAfterQuota(Account $account): void
|
||||
{
|
||||
$seconds = max(1, $this->pacer->waitSeconds($account));
|
||||
|
||||
$this->bus->dispatch(
|
||||
new ImportMailMessage((int) $account->id),
|
||||
[new DelayStamp(1000 * $seconds)],
|
||||
);
|
||||
|
||||
$this->logger->info('Import: waiting for Gmail quota', [
|
||||
'accountId' => $account->id,
|
||||
'delaySeconds' => $seconds,
|
||||
'limit' => $this->pacer->limit(),
|
||||
]);
|
||||
}
|
||||
|
||||
/** @return bool whether asking again at once would only repeat this */
|
||||
private function gmailPage(Account $account): bool
|
||||
{
|
||||
|
||||
@@ -139,7 +139,12 @@ the record and a missed nudge costs one stale render, never a wrong answer.
|
||||
bar that has stopped says nothing by itself;
|
||||
"waiting" at least says it is known to have. #}
|
||||
<p class="mt-1 text-[11px] text-ink-faint truncate">
|
||||
{% if line.waiting %}
|
||||
{# Held by Google for a minute is not
|
||||
"waiting": it is known why, and when it
|
||||
ends. #}
|
||||
{% if line.quota and line.quota.isPaused() %}
|
||||
{{ 'jobs.import.quota_paused'|trans({ '%seconds%': line.quota.pausedSeconds }) }}
|
||||
{% elseif line.waiting %}
|
||||
{{ 'jobs.import.waiting'|trans }}
|
||||
{% elseif line.label %}
|
||||
{{ 'jobs.import.folder'|trans({ '%folder%': line.label|label_name }) }}
|
||||
@@ -149,6 +154,35 @@ the record and a missed nudge costs one stale render, never a wrong answer.
|
||||
{{ 'jobs.import.newest_first'|trans }}
|
||||
{% endif %}
|
||||
</p>
|
||||
|
||||
{# A Gmail import is as fast as Google's quota
|
||||
lets it be, so the quota is what answers "why
|
||||
so slow" and "how much longer". All of it is
|
||||
plMail's own reckoning — Google reports none
|
||||
of it — hence "about". #}
|
||||
{% if line.quota %}
|
||||
{# Not truncated like its neighbours: the note that
|
||||
the limit is only assumed is the end of the
|
||||
sentence, and the part that was cut off. #}
|
||||
<p class="mt-0.5 text-[11px] text-ink-faint" data-import-quota>
|
||||
{{ (line.quota.learnt ? 'jobs.import.quota' : 'jobs.import.quota_assumed')|trans({
|
||||
'%used%': line.quota.used|number_format,
|
||||
'%limit%': line.quota.limit|number_format,
|
||||
}) }}
|
||||
</p>
|
||||
{% if line.quota.minutesLeft is not null and (not line.waiting or line.quota.isPaused()) %}
|
||||
<p class="mt-0.5 text-[11px] text-ink-faint truncate" data-import-remaining>
|
||||
{% if line.quota.minutesLeft >= 60 %}
|
||||
{{ 'jobs.import.remaining_hours'|trans({
|
||||
'%hours%': line.quota.minutesLeft // 60,
|
||||
'%minutes%': line.quota.minutesLeft % 60,
|
||||
}) }}
|
||||
{% else %}
|
||||
{{ 'jobs.import.remaining_minutes'|trans({ '%minutes%': line.quota.minutesLeft }) }}
|
||||
{% endif %}
|
||||
</p>
|
||||
{% endif %}
|
||||
{% endif %}
|
||||
</li>
|
||||
{% endfor %}
|
||||
{% for job in jobs %}
|
||||
|
||||
@@ -188,6 +188,7 @@ final class GmailInboundCategoryTest extends KernelTestCase
|
||||
$container->get('App\Service\Mail\MessageCategorizer'),
|
||||
$container->get('App\Service\Imap\MessageThreader'),
|
||||
new \App\Service\Mail\SyncOrigin(),
|
||||
new \App\Service\Gmail\GmailQuotaPacer(new \Symfony\Component\Clock\MockClock()),
|
||||
);
|
||||
|
||||
$handler(new SyncGmailMessageBatchMessage((int) $this->account->id, [self::GMAIL_ID]));
|
||||
|
||||
@@ -270,6 +270,7 @@ final class GmailInboundReadStateTest extends KernelTestCase
|
||||
$container->get('App\Service\Mail\MessageCategorizer'),
|
||||
$container->get('App\Service\Imap\MessageThreader'),
|
||||
new \App\Service\Mail\SyncOrigin(),
|
||||
new \App\Service\Gmail\GmailQuotaPacer(new \Symfony\Component\Clock\MockClock()),
|
||||
);
|
||||
|
||||
$handler(new SyncGmailMessageBatchMessage((int) $this->account->id, [self::GMAIL_ID]));
|
||||
|
||||
@@ -8,20 +8,22 @@ use App\Entity\Mail\Account;
|
||||
use App\Service\Gmail\GmailQuotaPacer;
|
||||
use PHPUnit\Framework\TestCase;
|
||||
use ReflectionProperty;
|
||||
use Symfony\Component\Cache\Adapter\ArrayAdapter;
|
||||
use Symfony\Component\Clock\MockClock;
|
||||
|
||||
/**
|
||||
* The pacer exists to keep an import under Google's 15,000 units a minute, so
|
||||
* the test that matters is the one that counts a minute. The rest pin down who
|
||||
* is made to wait and who is not.
|
||||
* The pacer exists to keep an import under Google's per-user limit, so the
|
||||
* tests that matter count a minute — at the limit it starts from, and at the
|
||||
* one a refusal teaches it (#90). The rest pin down who is made to wait and
|
||||
* who is not.
|
||||
*
|
||||
* MockClock::sleep() moves the clock instead of waiting, which is what makes
|
||||
* "how long was the caller held" readable as a difference of two times.
|
||||
*/
|
||||
final class GmailQuotaPacerTest extends TestCase
|
||||
{
|
||||
/** A full batch: a hundred messages.get at five units. */
|
||||
private const int BATCH = 500;
|
||||
/** A full batch: fifty messages.get at twenty units. */
|
||||
private const int BATCH = 1000;
|
||||
|
||||
public function testAFewBatchesInARowAreNotHeld(): void
|
||||
{
|
||||
@@ -30,7 +32,7 @@ final class GmailQuotaPacerTest extends TestCase
|
||||
$account = $this->account(1);
|
||||
$start = $clock->now();
|
||||
|
||||
for ($i = 0; $i < 6; $i++) {
|
||||
for ($i = 0; $i < 3; $i++) {
|
||||
$pacer->spend($account, self::BATCH);
|
||||
}
|
||||
|
||||
@@ -38,30 +40,188 @@ final class GmailQuotaPacerTest extends TestCase
|
||||
}
|
||||
|
||||
public function testAnImportNeverSpendsMoreThanTheLimitInAnyMinute(): void
|
||||
{
|
||||
$clock = new MockClock('2026-10-08 14:30:00');
|
||||
$pacer = new GmailQuotaPacer($clock);
|
||||
|
||||
$worst = $this->worstMinute($pacer, $clock, $this->account(1));
|
||||
|
||||
self::assertLessThanOrEqual(13_000, $worst);
|
||||
// And not so cautious that an import crawls: the steady rate is used.
|
||||
self::assertGreaterThanOrEqual(9_000, $worst);
|
||||
}
|
||||
|
||||
/**
|
||||
* The whole of #90: a project with 6,000 units a minute, and a pacer that
|
||||
* allowed 9,000 because 15,000 was written into it.
|
||||
*/
|
||||
public function testARefusalUnderALowerLimitIsWhatTheImportIsPacedToFromThenOn(): void
|
||||
{
|
||||
$clock = new MockClock('2026-10-08 14:30:00');
|
||||
$pacer = new GmailQuotaPacer($clock);
|
||||
$account = $this->account(1);
|
||||
|
||||
/** @var list<float> $spentAt */
|
||||
$spentAt = [];
|
||||
self::assertSame(15_000, $pacer->limit());
|
||||
|
||||
// Ten minutes of a worker that fetches as fast as it is allowed to.
|
||||
for ($i = 0; $i < 200; $i++) {
|
||||
$pacer->refused($account, 'defaultPerMinutePerUser', 6000);
|
||||
|
||||
self::assertSame(6000, $pacer->limit());
|
||||
|
||||
$clock->sleep(120);
|
||||
|
||||
$worst = $this->worstMinute($pacer, $clock, $account);
|
||||
|
||||
self::assertLessThanOrEqual(6000, $worst);
|
||||
self::assertGreaterThanOrEqual(3600, $worst);
|
||||
}
|
||||
|
||||
public function testARefusedAccountIsLeftAloneForAMinuteAndThenIsNot(): void
|
||||
{
|
||||
$clock = new MockClock('2026-10-08 14:30:00');
|
||||
$pacer = new GmailQuotaPacer($clock);
|
||||
$account = $this->account(1);
|
||||
|
||||
self::assertSame(0, $pacer->waitSeconds($account));
|
||||
|
||||
$pacer->refused($account, 'defaultPerMinutePerUser', 6000);
|
||||
|
||||
self::assertSame(60, $pacer->waitSeconds($account));
|
||||
self::assertSame(0, $pacer->waitSeconds($this->account(2)), 'a refusal is one mailbox\'s, the limit is everybody\'s');
|
||||
|
||||
$clock->sleep(61);
|
||||
|
||||
self::assertSame(0, $pacer->waitSeconds($account));
|
||||
}
|
||||
|
||||
public function testGooglesOwnRetryAfterIsBelievedOverTheMinute(): void
|
||||
{
|
||||
$pacer = new GmailQuotaPacer(new MockClock('2026-10-08 14:30:00'));
|
||||
$account = $this->account(1);
|
||||
|
||||
$pacer->refused($account, 'defaultPerMinutePerUser', 6000, 200);
|
||||
|
||||
self::assertSame(200, $pacer->waitSeconds($account));
|
||||
}
|
||||
|
||||
/**
|
||||
* The per-project limit is 1,200,000. Taken for the per-user one it would
|
||||
* teach nothing; a refusal under it still means "not now".
|
||||
*/
|
||||
public function testARefusalUnderTheProjectsLimitHoldsTheAccountAndTeachesNothing(): void
|
||||
{
|
||||
$pacer = new GmailQuotaPacer(new MockClock('2026-10-08 14:30:00'));
|
||||
$account = $this->account(1);
|
||||
|
||||
$pacer->refused($account, 'defaultPerMinutePerProject', 1_200_000);
|
||||
|
||||
self::assertSame(15_000, $pacer->limit());
|
||||
self::assertSame(60, $pacer->waitSeconds($account));
|
||||
}
|
||||
|
||||
/** The sentence names the limit even when the detail with its value is missing. */
|
||||
public function testAPerUserRefusalThatGivesNoValueFallsBackToTodaysDefault(): void
|
||||
{
|
||||
$pacer = new GmailQuotaPacer(new MockClock('2026-10-08 14:30:00'));
|
||||
|
||||
$pacer->refused($this->account(1), 'Unitsperminuteperuser');
|
||||
|
||||
self::assertSame(6000, $pacer->limit());
|
||||
}
|
||||
|
||||
/** Another client spending the same user's quota is not news about the limit. */
|
||||
public function testALearntLimitIsNeverRaisedByALaterRefusal(): void
|
||||
{
|
||||
$pacer = new GmailQuotaPacer(new MockClock('2026-10-08 14:30:00'));
|
||||
$account = $this->account(1);
|
||||
|
||||
$pacer->refused($account, 'defaultPerMinutePerUser', 6000);
|
||||
$pacer->refused($account, 'defaultPerMinutePerUser', 15_000);
|
||||
$pacer->refused($account, 'Unitsperminuteperuser');
|
||||
|
||||
self::assertSame(6000, $pacer->limit());
|
||||
}
|
||||
|
||||
/**
|
||||
* worker-backlog fetches an import's batches while worker-ingest fetches
|
||||
* the mail arriving meanwhile, each with a pacer of its own. What one
|
||||
* spent and what one was refused has to count for the other.
|
||||
*/
|
||||
public function testTwoProcessesShareOneAllowanceAndOneRefusal(): void
|
||||
{
|
||||
$clock = new MockClock('2026-10-08 14:30:00');
|
||||
$shared = new ArrayAdapter();
|
||||
$backlog = new GmailQuotaPacer($clock, $shared);
|
||||
$ingest = new GmailQuotaPacer($clock, $shared);
|
||||
$account = $this->account(1);
|
||||
$start = $clock->now();
|
||||
|
||||
for ($i = 0; $i < 3; $i++) {
|
||||
$backlog->spend($account, self::BATCH);
|
||||
}
|
||||
|
||||
self::assertEquals($start, $clock->now());
|
||||
|
||||
$ingest->spend($account, self::BATCH);
|
||||
|
||||
self::assertGreaterThan($start, $clock->now(), 'the reserve the other process used is used');
|
||||
|
||||
$backlog->refused($account, 'defaultPerMinutePerUser', 6000);
|
||||
|
||||
self::assertSame(6000, $ingest->limit());
|
||||
self::assertSame(60, $ingest->waitSeconds($account));
|
||||
}
|
||||
|
||||
/**
|
||||
* What the import line shows as "about N of the limit this minute". Its
|
||||
* own count, since Google reports none — and a minute's worth, so it falls
|
||||
* back as the minute passes.
|
||||
*/
|
||||
public function testWhatWasSpentInTheLastMinuteIsKnownAndThenForgotten(): void
|
||||
{
|
||||
$clock = new MockClock('2026-10-08 14:30:00');
|
||||
$pacer = new GmailQuotaPacer($clock);
|
||||
$account = $this->account(1);
|
||||
|
||||
self::assertSame(0, $pacer->usedThisMinute($account));
|
||||
|
||||
$pacer->spend($account, self::BATCH);
|
||||
$pacer->spend($account, self::BATCH);
|
||||
|
||||
self::assertSame(2000, $pacer->usedThisMinute($account));
|
||||
self::assertSame(0, $pacer->usedThisMinute($this->account(2)));
|
||||
|
||||
$clock->sleep(30);
|
||||
$pacer->spend($account, self::BATCH);
|
||||
|
||||
self::assertSame(3000, $pacer->usedThisMinute($account));
|
||||
|
||||
$clock->sleep(45);
|
||||
|
||||
self::assertSame(1000, $pacer->usedThisMinute($account));
|
||||
|
||||
$clock->sleep(60);
|
||||
|
||||
self::assertSame(0, $pacer->usedThisMinute($account));
|
||||
}
|
||||
|
||||
/** A worker flat out is shown spending what it is allowed, not more. */
|
||||
public function testAnImportFlatOutShowsItsSteadyAllowanceAsUsed(): void
|
||||
{
|
||||
$clock = new MockClock('2026-10-08 14:30:00');
|
||||
$pacer = new GmailQuotaPacer($clock);
|
||||
$account = $this->account(1);
|
||||
|
||||
$pacer->refused($account, 'defaultPerMinutePerUser', 6000);
|
||||
$clock->sleep(120);
|
||||
|
||||
for ($i = 0; $i < 40; $i++) {
|
||||
$pacer->spend($account, self::BATCH);
|
||||
$spentAt[] = (float) $clock->now()->format('U.u');
|
||||
}
|
||||
|
||||
$worst = 0;
|
||||
|
||||
foreach ($spentAt as $from) {
|
||||
$inWindow = count(array_filter($spentAt, static fn (float $at): bool => $at >= $from && $at < $from + 60.0));
|
||||
$worst = max($worst, $inWindow * self::BATCH);
|
||||
}
|
||||
|
||||
self::assertLessThanOrEqual(12_500, $worst);
|
||||
// And not so cautious that an import crawls: the steady rate is used.
|
||||
self::assertGreaterThanOrEqual(9_000, $worst);
|
||||
self::assertLessThanOrEqual(6000, $pacer->usedThisMinute($account));
|
||||
self::assertGreaterThanOrEqual(3000, $pacer->usedThisMinute($account));
|
||||
self::assertTrue($pacer->isLimitLearnt());
|
||||
self::assertSame(180, $pacer->messagesPerMinute());
|
||||
}
|
||||
|
||||
public function testOneAccountImportingDoesNotHoldAnother(): void
|
||||
@@ -92,13 +252,37 @@ final class GmailQuotaPacerTest extends TestCase
|
||||
$clock->sleep(120);
|
||||
$before = $clock->now();
|
||||
|
||||
for ($i = 0; $i < 6; $i++) {
|
||||
for ($i = 0; $i < 3; $i++) {
|
||||
$pacer->spend($account, self::BATCH);
|
||||
}
|
||||
|
||||
self::assertEquals($before, $clock->now());
|
||||
}
|
||||
|
||||
/**
|
||||
* Ten minutes of a worker that fetches as fast as it is allowed to, and
|
||||
* the most it spent in any sixty seconds of them.
|
||||
*/
|
||||
private function worstMinute(GmailQuotaPacer $pacer, MockClock $clock, Account $account): int
|
||||
{
|
||||
/** @var list<float> $spentAt */
|
||||
$spentAt = [];
|
||||
|
||||
for ($i = 0; $i < 100; $i++) {
|
||||
$pacer->spend($account, self::BATCH);
|
||||
$spentAt[] = (float) $clock->now()->format('U.u');
|
||||
}
|
||||
|
||||
$worst = 0;
|
||||
|
||||
foreach ($spentAt as $from) {
|
||||
$inWindow = count(array_filter($spentAt, static fn (float $at): bool => $at >= $from && $at < $from + 60.0));
|
||||
$worst = max($worst, $inWindow * self::BATCH);
|
||||
}
|
||||
|
||||
return $worst;
|
||||
}
|
||||
|
||||
private function account(int $id): Account
|
||||
{
|
||||
$account = new Account();
|
||||
|
||||
@@ -9,9 +9,11 @@ use App\Domain\Exception\GmailPermanentException;
|
||||
use App\Domain\Exception\GmailThrottledException;
|
||||
use App\Entity\Mail\Account;
|
||||
use App\Entity\User\User;
|
||||
use App\Service\Gmail\GmailQuotaPacer;
|
||||
use App\Service\Mail\GmailApiClient;
|
||||
use App\Service\OAuth\OAuthTokenManager;
|
||||
use PHPUnit\Framework\TestCase;
|
||||
use Symfony\Component\Clock\MockClock;
|
||||
use Symfony\Component\HttpClient\MockHttpClient;
|
||||
use Symfony\Component\HttpClient\Response\MockResponse;
|
||||
use Symfony\Component\Messenger\Exception\RecoverableExceptionInterface;
|
||||
@@ -112,7 +114,9 @@ final class GmailApiClientFailureTest extends TestCase
|
||||
$client->listLabels($this->account());
|
||||
self::fail('a 403 must not be swallowed');
|
||||
} catch (GmailThrottledException $e) {
|
||||
self::assertNull($e->getRetryAfterSeconds());
|
||||
// The minute the account is left alone for when Google names no
|
||||
// time of its own — not the date read as a number.
|
||||
self::assertSame(60, $e->getRetryAfterSeconds());
|
||||
self::assertSame(60000, $e->getRetryDelay());
|
||||
}
|
||||
}
|
||||
@@ -377,8 +381,122 @@ final class GmailApiClientFailureTest extends TestCase
|
||||
$this->expectNotToPerformAssertions();
|
||||
}
|
||||
|
||||
// ── What a refusal teaches (#90) ─────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* The limit is not configured anywhere: the refusal names it, and that is
|
||||
* where the pacer has it from.
|
||||
*/
|
||||
public function testARefusalTellsThePacerTheLimitItWasRefusedUnder(): void
|
||||
{
|
||||
$pacer = new GmailQuotaPacer(new MockClock());
|
||||
$client = $this->clientWith($pacer, new MockResponse(json_encode($this->perUserQuotaBody(6000), JSON_THROW_ON_ERROR), ['http_code' => 403]));
|
||||
|
||||
self::assertSame(GmailQuotaPacer::HIGHEST_LIMIT, $pacer->limit());
|
||||
|
||||
try {
|
||||
$client->listLabels($this->account());
|
||||
self::fail('a 403 must not be swallowed');
|
||||
} catch (GmailThrottledException) {
|
||||
}
|
||||
|
||||
self::assertSame(6000, $pacer->limit());
|
||||
self::assertSame(60, $pacer->waitSeconds($this->account()));
|
||||
}
|
||||
|
||||
/**
|
||||
* Google refusing some of a batch is the ordinary way an import meets its
|
||||
* quota. Those ids are handed back to be asked for again and the pacer is
|
||||
* told; nothing is thrown.
|
||||
*/
|
||||
public function testARefusedPartOfABatchIsHandedBackAndReportedToThePacer(): void
|
||||
{
|
||||
$refusal = json_encode($this->perUserQuotaBody(6000), JSON_THROW_ON_ERROR);
|
||||
|
||||
$body = implode("\r\n", [
|
||||
'--batch_x',
|
||||
'Content-Type: application/http',
|
||||
'Content-ID: <response-m1>',
|
||||
'',
|
||||
'HTTP/1.1 200 OK',
|
||||
'Content-Type: application/json',
|
||||
'',
|
||||
'{"id": "m1", "labelIds": ["INBOX"]}',
|
||||
'--batch_x',
|
||||
'Content-Type: application/http',
|
||||
'Content-ID: <response-m2>',
|
||||
'',
|
||||
'HTTP/1.1 403 Forbidden',
|
||||
'Content-Type: application/json',
|
||||
'',
|
||||
$refusal,
|
||||
'--batch_x--',
|
||||
]);
|
||||
|
||||
$pacer = new GmailQuotaPacer(new MockClock());
|
||||
$client = $this->clientWith($pacer, new MockResponse($body, ['http_code' => 200]));
|
||||
|
||||
$fetched = $client->getMessages($this->account(), ['m1', 'm2']);
|
||||
|
||||
self::assertSame(['m1'], array_keys($fetched['payloads']));
|
||||
self::assertSame(['m2'], $fetched['retryable']);
|
||||
self::assertSame([], $fetched['gone']);
|
||||
self::assertSame(6000, $pacer->limit());
|
||||
self::assertGreaterThan(0, $pacer->waitSeconds($this->account()));
|
||||
}
|
||||
|
||||
/** The whole request refused is the same answer as every part refused. */
|
||||
public function testAWhollyRefusedBatchIsHandedBackRatherThanThrown(): void
|
||||
{
|
||||
$pacer = new GmailQuotaPacer(new MockClock());
|
||||
$client = $this->clientWith($pacer, new MockResponse(json_encode($this->quotaBody('rateLimitExceeded'), JSON_THROW_ON_ERROR), ['http_code' => 403]));
|
||||
|
||||
$fetched = $client->getMessages($this->account(), ['m1', 'm2']);
|
||||
|
||||
self::assertSame([], $fetched['payloads']);
|
||||
self::assertSame(['m1', 'm2'], $fetched['retryable']);
|
||||
self::assertGreaterThan(0, $pacer->waitSeconds($this->account()));
|
||||
}
|
||||
|
||||
// ── Fixture ──────────────────────────────────────────────────────────────
|
||||
|
||||
private function clientWith(GmailQuotaPacer $pacer, MockResponse $response): GmailApiClient
|
||||
{
|
||||
$tokenManager = $this->createStub(OAuthTokenManager::class);
|
||||
$tokenManager->method('getValidAccessToken')->willReturn('test-token');
|
||||
|
||||
return new GmailApiClient(new MockHttpClient($response), $tokenManager, $pacer);
|
||||
}
|
||||
|
||||
/**
|
||||
* A per-user quota refusal as Google sends it, with the ErrorInfo detail
|
||||
* that names the limit and its value.
|
||||
*
|
||||
* @return array<string,mixed>
|
||||
*/
|
||||
private function perUserQuotaBody(int $limit): array
|
||||
{
|
||||
return [
|
||||
'error' => [
|
||||
'code' => 403,
|
||||
'errors' => [['reason' => 'rateLimitExceeded', 'message' => 'Quota exceeded']],
|
||||
'message' => "Quota exceeded for quota metric 'Total Query Cost' and limit 'Units per minute per user' of service 'gmail.googleapis.com'.",
|
||||
'status' => 'RESOURCE_EXHAUSTED',
|
||||
'details' => [[
|
||||
'@type' => 'type.googleapis.com/google.rpc.ErrorInfo',
|
||||
'reason' => 'RATE_LIMIT_EXCEEDED',
|
||||
'domain' => 'googleapis.com',
|
||||
'metadata' => [
|
||||
'quota_limit' => 'defaultPerMinutePerUser',
|
||||
'quota_limit_value' => (string) $limit,
|
||||
'quota_metric' => 'gmail.googleapis.com/default',
|
||||
'service' => 'gmail.googleapis.com',
|
||||
],
|
||||
]],
|
||||
],
|
||||
];
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array<string,mixed>|null $json decoded body, or null to send $raw verbatim
|
||||
* @param array<string,string> $headers
|
||||
|
||||
@@ -9,6 +9,7 @@ use App\Domain\Enum\Mail\MailboxSpecialUse;
|
||||
use App\Entity\Label\Label;
|
||||
use App\Entity\Label\LabelBinding;
|
||||
use App\Entity\Mail\Mailbox;
|
||||
use App\Service\Gmail\GmailQuotaPacer;
|
||||
use App\Service\Mail\ImportProgress;
|
||||
use App\Tests\Support\Mail\SeedsMarkerFixtures;
|
||||
use DateTimeImmutable;
|
||||
@@ -174,6 +175,54 @@ final class ImportProgressTest extends KernelTestCase
|
||||
self::assertStringContainsString('INBOX', $html);
|
||||
}
|
||||
|
||||
/**
|
||||
* A Gmail import is as fast as Google's quota lets it be, so its line says
|
||||
* what the quota is, roughly what is being used of it, and how long the
|
||||
* rest takes at that — and "held by Google for a minute" is said as that
|
||||
* and not as the "waiting" that could mean anything (#90).
|
||||
*/
|
||||
public function testAGmailImportShowsItsQuotaAndHowLongIsLeft(): void
|
||||
{
|
||||
$this->account->authType = AuthType::OAuth2->value;
|
||||
$this->account->oauthProvider = 'google';
|
||||
$this->account->importTotal = 9000;
|
||||
$this->account->importBeatAt = new DateTimeImmutable();
|
||||
$this->em->flush();
|
||||
|
||||
$pacer = static::getContainer()->get(GmailQuotaPacer::class);
|
||||
$pacer->spend($this->account, 2000);
|
||||
|
||||
$line = $this->progress->forUser($this->user)[0];
|
||||
|
||||
self::assertNotNull($line->quota);
|
||||
self::assertSame(15_000, $line->quota->limit);
|
||||
self::assertFalse($line->quota->learnt);
|
||||
self::assertSame(2000, $line->quota->used);
|
||||
self::assertFalse($line->quota->isPaused());
|
||||
// 9,000 messages at 450 a minute, which is three fifths of 15,000
|
||||
// units at twenty a message.
|
||||
self::assertSame(20, $line->quota->minutesLeft);
|
||||
|
||||
$pacer->refused($this->account, 'defaultPerMinutePerUser', 6000);
|
||||
|
||||
$line = $this->progress->forUser($this->user)[0];
|
||||
|
||||
self::assertSame(6000, $line->quota->limit);
|
||||
self::assertTrue($line->quota->learnt);
|
||||
self::assertTrue($line->quota->isPaused());
|
||||
self::assertSame(50, $line->quota->minutesLeft);
|
||||
|
||||
$html = static::getContainer()->get(Environment::class)->render('mail/_jobs_indicator.html.twig', [
|
||||
'jobs' => [],
|
||||
'imports' => [$line],
|
||||
]);
|
||||
|
||||
self::assertStringContainsString('data-import-quota', $html);
|
||||
self::assertStringContainsString('6,000', $html);
|
||||
self::assertStringContainsString('About 50 min left', $html);
|
||||
self::assertStringContainsString('Paused by Gmail', $html);
|
||||
}
|
||||
|
||||
/** Nothing running is no button at all, as it was before imports were shown. */
|
||||
public function testWithNothingRunningTheIndicatorIsEmpty(): void
|
||||
{
|
||||
|
||||
@@ -10,6 +10,7 @@ use App\Entity\Mail\Mailbox;
|
||||
use App\Infrastructure\Messaging\Message\ImportMailMessage;
|
||||
use App\Infrastructure\Messaging\Message\SyncAccountMessage;
|
||||
use App\Repository\Mail\MailboxRepository;
|
||||
use App\Service\Gmail\GmailQuotaPacer;
|
||||
use App\Service\Mail\InitialImportState;
|
||||
use App\Service\Mail\MailImporter;
|
||||
use App\Tests\Support\Mail\SeedsMarkerFixtures;
|
||||
@@ -17,6 +18,7 @@ use DateTimeImmutable;
|
||||
use Doctrine\DBAL\Connection;
|
||||
use Doctrine\ORM\EntityManagerInterface;
|
||||
use Symfony\Bundle\FrameworkBundle\Test\KernelTestCase;
|
||||
use Symfony\Component\Messenger\Stamp\DelayStamp;
|
||||
use Symfony\Component\Messenger\Transport\InMemory\InMemoryTransport;
|
||||
|
||||
/**
|
||||
@@ -74,6 +76,34 @@ final class MailImporterTest extends KernelTestCase
|
||||
self::assertSame([], $this->queue('ingest')->getSent(), 'history is never put on the queue new mail arrives on');
|
||||
}
|
||||
|
||||
/**
|
||||
* Google refusing a Gmail account is not the end of its import. The page
|
||||
* is asked for again when the refusal has cleared, and nothing is thrown:
|
||||
* it used to be retried, given up on at CRITICAL, and left for a later
|
||||
* sync to restart (#90).
|
||||
*/
|
||||
public function testAnImportGoogleIsRefusingAsksAgainLaterInsteadOfFailing(): void
|
||||
{
|
||||
$this->account->authType = AuthType::OAuth2->value;
|
||||
$this->account->oauthProvider = 'google';
|
||||
$this->em->flush();
|
||||
|
||||
static::getContainer()->get(GmailQuotaPacer::class)->refused($this->account, 'defaultPerMinutePerUser', 6000);
|
||||
|
||||
$this->importer->runPage((int) $this->account->id);
|
||||
|
||||
$sent = $this->queue('ingest_backlog')->getSent();
|
||||
|
||||
self::assertCount(1, $sent);
|
||||
self::assertInstanceOf(ImportMailMessage::class, $sent[0]->getMessage());
|
||||
|
||||
$delay = $sent[0]->last(DelayStamp::class);
|
||||
|
||||
self::assertNotNull($delay);
|
||||
self::assertGreaterThan(0, $delay->getDelay());
|
||||
self::assertLessThanOrEqual(60_000, $delay->getDelay());
|
||||
}
|
||||
|
||||
/**
|
||||
* Every sync of the account calls ensureRunning(). While the import is
|
||||
* moving, that must not put a second chain beside the first.
|
||||
|
||||
@@ -2187,6 +2187,11 @@ jobs:
|
||||
folder: "Gerade: %folder%"
|
||||
heading: "%account% wird importiert"
|
||||
newest_first: "Neueste Mail zuerst"
|
||||
quota: "Gmail-Kontingent: etwa %used% von %limit% Einheiten in dieser Minute"
|
||||
quota_assumed: "Gmail-Kontingent: etwa %used% von %limit% Einheiten in dieser Minute (Limit angenommen, bis Google es nennt)"
|
||||
quota_paused: "Pausiert durch Gmails Kontingent · weiter in %seconds% s"
|
||||
remaining_hours: "Noch etwa %hours% Std. %minutes% Min."
|
||||
remaining_minutes: "Noch etwa %minutes% Min."
|
||||
waiting: "Wartet darauf, weiterzumachen"
|
||||
started: '%action% gestartet — das läuft im Hintergrund.'
|
||||
failure:
|
||||
|
||||
@@ -2186,6 +2186,11 @@ jobs:
|
||||
folder: "Now: %folder%"
|
||||
heading: "Importing %account%"
|
||||
newest_first: "Newest mail first"
|
||||
quota: "Gmail quota: about %used% of %limit% units this minute"
|
||||
quota_assumed: "Gmail quota: about %used% of %limit% units this minute (limit assumed until Google names it)"
|
||||
quota_paused: "Paused by Gmail's quota · continues in %seconds% s"
|
||||
remaining_hours: "About %hours% h %minutes% min left"
|
||||
remaining_minutes: "About %minutes% min left"
|
||||
waiting: "Waiting to continue"
|
||||
started: 'Started %action% — this runs in the background.'
|
||||
failure:
|
||||
|
||||
@@ -2066,6 +2066,11 @@ jobs:
|
||||
folder: "Now: %folder%"
|
||||
heading: "Haulin' %account% aboard"
|
||||
newest_first: "Freshest post first"
|
||||
quota: "Gmail's ration: about %used% o' %limit% units this minute"
|
||||
quota_assumed: "Gmail's ration: about %used% o' %limit% units this minute (a guess, till Google names the limit)"
|
||||
quota_paused: "Becalmed by Gmail's ration · under way in %seconds% s"
|
||||
remaining_hours: "About %hours% h %minutes% min o' haulin' left"
|
||||
remaining_minutes: "About %minutes% min o' haulin' left"
|
||||
waiting: "Waitin' fer the tide"
|
||||
started: 'Started %action% — she runs below decks.'
|
||||
failure:
|
||||
|
||||
Reference in new issue
Block a user