From 1c6a8eb0927582cb48a3a51c352466d289e00b6e Mon Sep 17 00:00:00 2001 From: Joas Schilling Date: Tue, 28 Jul 2026 09:03:22 +0000 Subject: [PATCH 1/4] fix(push): Flush queued web push deletes right away webPushDeleteToDevice() queued the delete payloads on the web push client but then flushed the push proxies instead, so the queued web push deletes leaked into an unrelated later flush. Also widen the proxy rejection handler to \Throwable: Guzzle rejection reasons are not guaranteed to be \Exception, and a non-Exception reason would raise a TypeError inside the promise handler. Assisted-by: ClaudeCode:claude-opus-5 Signed-off-by: Joas Schilling --- lib/Push.php | 6 +++--- tests/Unit/PushTest.php | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 37 insertions(+), 3 deletions(-) diff --git a/lib/Push.php b/lib/Push.php index 2055bf0eb..1e4fc29d0 100644 --- a/lib/Push.php +++ b/lib/Push.php @@ -604,7 +604,7 @@ public function webPushDeleteToDevice(string $userId, IUser $user, array $device } if (!$this->deferPayloads) { - $this->sendNotificationsToProxies(); + $this->wpClient->flush(fn ($r) => $this->webPushCallback($r)); } } @@ -734,7 +734,7 @@ function (IResponse $response) use ($proxyServer, $postStartTime): void { $this->printInfo('Request to push proxy [' . $proxyServer . '] took ' . (string)round(microtime(true) - $postStartTime, 2) . 's'); $this->handleProxyResponse($proxyServer, $response->getStatusCode(), (string)$response->getBody()); }, - function (\Exception $e) use ($proxyServer): void { + function (\Throwable $e) use ($proxyServer): void { $this->handleProxyException($proxyServer, $e); }, ); @@ -755,7 +755,7 @@ function (\Exception $e) use ($proxyServer): void { /** * Handle a request to a push proxy that did not yield a response */ - protected function handleProxyException(string $proxyServer, \Exception $e): void { + protected function handleProxyException(string $proxyServer, \Throwable $e): void { if ($e instanceof ClientException) { // Server responded with 4xx (400 Bad Request mostlikely) $response = $e->getResponse(); diff --git a/tests/Unit/PushTest.php b/tests/Unit/PushTest.php index f3f082d31..9f0ee6960 100644 --- a/tests/Unit/PushTest.php +++ b/tests/Unit/PushTest.php @@ -1233,6 +1233,40 @@ public function testWebPushToDeviceSending(bool $isRateLimited): void { $push->pushToDevice(207787, $notification); } + public function testWebPushDeleteToDeviceFlushesWebPush(): void { + $push = $this->getPush(); + + /** @var IUser&MockObject $user */ + $user = $this->createStub(IUser::class); + + $this->cache->method('get') + ->willReturn(false); + + $this->wpClient->expects($this->once()) + ->method('enqueue') + ->with('endpoint1', self::EX_UA_PUBLIC, self::EX_AUTH); + + // The queued deletes have to be sent right away, they must not leak + // into an unrelated later flush + $this->wpClient->expects($this->once()) + ->method('flush'); + + // Nothing was queued for the push proxies + $this->clientService->expects($this->never()) + ->method('newClient'); + + $push->webPushDeleteToDevice('valid', $user, [ + [ + 'activated' => true, + 'endpoint' => 'endpoint1', + 'ua_public' => self::EX_UA_PUBLIC, + 'auth' => self::EX_AUTH, + 'token' => 16, + 'app_types' => 'all', + ], + ], true, null); + } + public static function dataFilterWebPushDeviceList(): array { return [ [false, 'all', 'myApp', false], From 6c699c853ffc1094539a9a534f66e600fb35bb9a Mon Sep 17 00:00:00 2001 From: Joas Schilling Date: Tue, 28 Jul 2026 09:04:36 +0000 Subject: [PATCH 2/4] perf(push): Configure explicit web push client options Sending a web push notification blocks the process it runs in, so the library default of a 30 seconds timeout per request is far too long. Also pass the batch size and request concurrency explicitly instead of relying on the library defaults, and hand the library a PSR-3 logger so it stops falling back to trigger_error(). Assisted-by: ClaudeCode:claude-opus-5 Signed-off-by: Joas Schilling --- lib/Controller/WebPushController.php | 2 +- lib/WebPush/LoggerAdapter.php | 31 ++++++++++++++++++++++++++++ lib/WebPushClient.php | 30 ++++++++++++++++++++++++++- tests/Unit/WebPushClientTest.php | 9 +++++--- 4 files changed, 67 insertions(+), 5 deletions(-) create mode 100644 lib/WebPush/LoggerAdapter.php diff --git a/lib/Controller/WebPushController.php b/lib/Controller/WebPushController.php index c4b6f909e..c1b293479 100644 --- a/lib/Controller/WebPushController.php +++ b/lib/Controller/WebPushController.php @@ -233,7 +233,7 @@ public function removeWP(): DataResponse { } protected function getWPClient(): WebPushClient { - return new WebPushClient($this->appConfig); + return new WebPushClient($this->appConfig, $this->logger); } /** diff --git a/lib/WebPush/LoggerAdapter.php b/lib/WebPush/LoggerAdapter.php new file mode 100644 index 000000000..4e405cf42 --- /dev/null +++ b/lib/WebPush/LoggerAdapter.php @@ -0,0 +1,31 @@ +logger->log($level, $message, $context + ['app' => 'notifications']); + } +} diff --git a/lib/WebPushClient.php b/lib/WebPushClient.php index a0774592d..d084b7630 100644 --- a/lib/WebPushClient.php +++ b/lib/WebPushClient.php @@ -14,15 +14,35 @@ use OCA\Notifications\Vendor\Minishlink\WebPush\Utils; use OCA\Notifications\Vendor\Minishlink\WebPush\VAPID; use OCA\Notifications\Vendor\Minishlink\WebPush\WebPush; +use OCA\Notifications\WebPush\LoggerAdapter; use OCP\AppFramework\Services\IAppConfig; +use Psr\Log\LoggerInterface; class WebPushClient { + /** + * Number of notifications that are sent as one pooled batch + */ + public const BATCH_SIZE = 1000; + + /** + * Number of requests that are in flight at the same time within a batch + */ + public const REQUEST_CONCURRENCY = 100; + + /** + * Timeout in seconds of a single request to a push endpoint. + * The library default of 30 seconds is way too long, as the sending + * blocks the process it runs in. + */ + public const REQUEST_TIMEOUT = 10; + private WebPush $client; /** @psalm-var array{publicKey: string, privateKey: string, subject: string} */ private array $vapid; public function __construct( protected IAppConfig $appConfig, + protected LoggerInterface $logger, ) { $this->vapid = $this->getVapid(); } @@ -55,7 +75,15 @@ private function getClient(): WebPush { if (isset($this->client)) { return $this->client; } - $this->client = new WebPush(auth: ['VAPID' => $this->vapid]); + $this->client = new WebPush( + auth: ['VAPID' => $this->vapid], + defaultOptions: [ + 'batchSize' => self::BATCH_SIZE, + 'requestConcurrency' => self::REQUEST_CONCURRENCY, + ], + timeout: self::REQUEST_TIMEOUT, + logger: new LoggerAdapter($this->logger), + ); $this->client->setReuseVAPIDHeaders(true); return $this->client; } diff --git a/tests/Unit/WebPushClientTest.php b/tests/Unit/WebPushClientTest.php index 9b3c235cc..b6a2c76ae 100644 --- a/tests/Unit/WebPushClientTest.php +++ b/tests/Unit/WebPushClientTest.php @@ -12,14 +12,17 @@ use OCA\Notifications\WebPushClient; use OCP\AppFramework\Services\IAppConfig; use PHPUnit\Framework\MockObject\MockObject; +use Psr\Log\LoggerInterface; use Test\TestCase; class WebPushClientTest extends TestCase { protected IAppConfig&MockObject $appConfig; + protected LoggerInterface&MockObject $logger; protected function setUp(): void { parent::setUp(); $this->appConfig = $this->createMock(IAppConfig::class); + $this->logger = $this->createMock(LoggerInterface::class); } public function testConstructSucceedsWhenVapidKeysAreStored(): void { @@ -31,7 +34,7 @@ public function testConstructSucceedsWhenVapidKeysAreStored(): void { $this->appConfig->expects($this->never())->method('setAppValueString'); - $client = new WebPushClient($this->appConfig); + $client = new WebPushClient($this->appConfig, $this->logger); $this->assertInstanceOf(WebPushClient::class, $client); } @@ -49,7 +52,7 @@ public function testConstructRegeneratesVapidKeysWhenDecryptionFails(): void { )); // Must not throw — corrupted keys should be transparently regenerated - $client = new WebPushClient($this->appConfig); + $client = new WebPushClient($this->appConfig, $this->logger); $this->assertInstanceOf(WebPushClient::class, $client); } @@ -67,7 +70,7 @@ public function testConstructRegeneratesVapidKeysWhenMissing(): void { $this->equalTo('webpush_vapid_privkey'), )); - $client = new WebPushClient($this->appConfig); + $client = new WebPushClient($this->appConfig, $this->logger); $this->assertInstanceOf(WebPushClient::class, $client); } } From e4728c743b7f41a8097a5a0065e36c38667b2a2c Mon Sep 17 00:00:00 2001 From: Joas Schilling Date: Tue, 28 Jul 2026 09:06:38 +0000 Subject: [PATCH 3/4] perf(push): Prepare web push requests lazily Encrypting the payload for every subscription happened for the whole batch before the first request was sent, so the CPU time was spent inside the blocking window instead of overlapping the network I/O. PooledWebPush feeds the pool a generator instead, so the payload of the next subscription is encrypted while the previous requests are in flight, and it returns the promises of the pools instead of waiting for them. WebPushClient::flush() waits for them right away, so all existing call sites keep behaving as before. As a side effect an encryption error of a single subscription does not abort the requests of all the other subscriptions of the batch anymore. Assisted-by: ClaudeCode:claude-opus-5 Signed-off-by: Joas Schilling --- lib/PooledWebPush.php | 108 +++++++++++++++++++++++++++++++ lib/WebPushClient.php | 35 ++++++++-- tests/Unit/WebPushClientTest.php | 26 ++++++++ 3 files changed, 162 insertions(+), 7 deletions(-) create mode 100644 lib/PooledWebPush.php diff --git a/lib/PooledWebPush.php b/lib/PooledWebPush.php new file mode 100644 index 000000000..02e9208b5 --- /dev/null +++ b/lib/PooledWebPush.php @@ -0,0 +1,108 @@ + One promise per batch, the caller has to wait for them + */ + public function flushPooledAsync(callable $callback, ?int $batchSize = null, ?int $requestConcurrency = null): array { + if (empty($this->notifications)) { + return []; + } + + $batchSize ??= $this->defaultOptions['batchSize']; + $requestConcurrency ??= $this->defaultOptions['requestConcurrency']; + + $batches = array_chunk($this->notifications, $batchSize); + // Reset the queue + $this->notifications = []; + + $promises = []; + foreach ($batches as $batch) { + /** + * Filled while the generator is consumed. The fulfilled callback + * needs the request back and can not index into a prepared array + * anymore, as the requests are only created on demand now. + * @var array $prepared + */ + $prepared = []; + + $requests = function () use ($batch, &$prepared): \Generator { + foreach ($batch as $index => $notification) { + try { + $request = $this->prepare([$notification])[0] ?? null; + } catch (\Throwable $e) { + // A single subscription that can not be encrypted must + // not abort the requests of all the other subscriptions + $this->logger?->error('Failed to prepare push notification: ' . $e->getMessage(), ['exception' => $e]); + continue; + } + + if ($request === null) { + continue; + } + + $prepared[$index] = $request; + yield $index => $request; + } + }; + + $pool = new Pool($this->client, $requests(), [ + 'concurrency' => $requestConcurrency, + 'fulfilled' => function (ResponseInterface $response, int $index) use ($callback, &$prepared): void { + $callback(new MessageSentReport($prepared[$index], $response)); + }, + 'rejected' => function ($reason) use ($callback): void { + $callback($this->createRejectedReport($reason)); + }, + ]); + + $promises[] = $pool->promise(); + } + + return $promises; + } + + /** + * Reset the cache of reused VAPID headers. + * + * flushPooled() does this after waiting for the requests. As + * flushPooledAsync() does not wait, the cache is instead reset before the + * requests of a flush are prepared, which gives the same result: the + * headers are only reused within one flush. + */ + public function clearVAPIDHeaderCache(): void { + $this->vapidHeaders = []; + } +} diff --git a/lib/WebPushClient.php b/lib/WebPushClient.php index d084b7630..64cda8834 100644 --- a/lib/WebPushClient.php +++ b/lib/WebPushClient.php @@ -10,10 +10,10 @@ namespace OCA\Notifications; use OCA\Notifications\Vendor\Base64Url\Base64Url; +use OCA\Notifications\Vendor\GuzzleHttp\Promise\PromiseInterface; use OCA\Notifications\Vendor\Minishlink\WebPush\Subscription; use OCA\Notifications\Vendor\Minishlink\WebPush\Utils; use OCA\Notifications\Vendor\Minishlink\WebPush\VAPID; -use OCA\Notifications\Vendor\Minishlink\WebPush\WebPush; use OCA\Notifications\WebPush\LoggerAdapter; use OCP\AppFramework\Services\IAppConfig; use Psr\Log\LoggerInterface; @@ -36,7 +36,7 @@ class WebPushClient { */ public const REQUEST_TIMEOUT = 10; - private WebPush $client; + private PooledWebPush $client; /** @psalm-var array{publicKey: string, privateKey: string, subject: string} */ private array $vapid; @@ -71,11 +71,11 @@ public static function isValidAuth(string $auth): bool { return strlen($a) === 16; } - private function getClient(): WebPush { + private function getClient(): PooledWebPush { if (isset($this->client)) { return $this->client; } - $this->client = new WebPush( + $this->client = new PooledWebPush( auth: ['VAPID' => $this->vapid], defaultOptions: [ 'batchSize' => self::BATCH_SIZE, @@ -145,7 +145,7 @@ public function notify(string $endpoint, string $uaPublicKey, string $auth, stri // as the registration isn't activated $callback = function ($r): void { }; - $c->flushPooled($callback); + $this->flush($callback); } /** @@ -164,11 +164,32 @@ public function enqueue(string $endpoint, string $uaPublicKey, string $auth, str } /** + * Start sending the queued notifications, the caller has to wait for the + * returned promises to finish the sending. + * * @param callable $callback * @psalm-param $callback callable(MessageSentReport): void + * @return list */ - public function flush(callable $callback): void { + public function flushAsync(callable $callback): array { $c = $this->getClient(); - $c->flushPooled($callback); + // The headers are only reused within one flush + $c->clearVAPIDHeaderCache(); + return $c->flushPooledAsync($callback); + } + + /** + * Send the queued notifications - blocking + * + * @param callable $callback + * @psalm-param $callback callable(MessageSentReport): void + */ + public function flush(callable $callback): void { + foreach ($this->flushAsync($callback) as $promise) { + // Rejections are already reported to the callback, so the result + // must not be unwrapped as that would rethrow the exception and + // skip the remaining batches. + $promise->wait(false); + } } } diff --git a/tests/Unit/WebPushClientTest.php b/tests/Unit/WebPushClientTest.php index b6a2c76ae..a55a6c780 100644 --- a/tests/Unit/WebPushClientTest.php +++ b/tests/Unit/WebPushClientTest.php @@ -73,4 +73,30 @@ public function testConstructRegeneratesVapidKeysWhenMissing(): void { $client = new WebPushClient($this->appConfig, $this->logger); $this->assertInstanceOf(WebPushClient::class, $client); } + + public function testFlushAsyncWithoutQueuedNotifications(): void { + $client = new WebPushClient($this->appConfig, $this->logger); + + $this->assertSame([], $client->flushAsync(static function (): void { + })); + } + + public function testFlushAsyncSkipsNotificationsThatCanNotBePrepared(): void { + $client = new WebPushClient($this->appConfig, $this->logger); + + // Not a valid public key, so the encryption fails while the pool + // prepares the request + $client->enqueue('https://example.tld/endpoint', str_repeat('A', 87), str_repeat('B', 22), '{"nid":1}'); + + // The library logs through the adapter, which only implements log() + $this->logger->expects($this->once()) + ->method('log') + ->with('error', $this->stringContains('Failed to prepare push notification'), $this->anything()); + + // One batch, but the request of the broken subscription is skipped, + // so nothing is sent and waiting for the promise is not needed + $promises = $client->flushAsync(static function (): void { + }); + $this->assertCount(1, $promises); + } } From 62f21d820b15e934c1c5b1c5380e541aa673b582 Mon Sep 17 00:00:00 2001 From: Joas Schilling Date: Tue, 28 Jul 2026 09:11:50 +0000 Subject: [PATCH 4/4] perf(push): Send push notifications after the response Push notifications were sent inline in whatever process triggered them, so a web request could not finish before every push endpoint and push proxy had answered. The requests are now started, but the waiting is handed to the new DeferredFlusher: on the command line it waits right away, so the output of `occ notification:test-push` stays in order and a cron run does not pile up the push traffic at the end. For web requests the waiting is postponed to the shutdown of the request, where the response is sent to the client first, when the SAPI supports it (PHP-FPM and LiteSpeed, but not mod_php). The callbacks that delete expired subscriptions and unknown devices run at shutdown as well. The shutdown functions of the server neither close the database connection nor the session, and PHP runs them in registration order, so the database and the logger are still available. Assisted-by: ClaudeCode:claude-opus-5 Signed-off-by: Joas Schilling --- lib/DeferredFlusher.php | 124 +++++++++++++++++++++++++++++ lib/Push.php | 50 ++++++++---- lib/WebPush/LoggerAdapter.php | 2 +- tests/Unit/DeferredFlusherTest.php | 119 +++++++++++++++++++++++++++ tests/Unit/PushTest.php | 101 ++++++++++++++++++++++- 5 files changed, 378 insertions(+), 18 deletions(-) create mode 100644 lib/DeferredFlusher.php create mode 100644 tests/Unit/DeferredFlusherTest.php diff --git a/lib/DeferredFlusher.php b/lib/DeferredFlusher.php new file mode 100644 index 000000000..9d71c4b1b --- /dev/null +++ b/lib/DeferredFlusher.php @@ -0,0 +1,124 @@ + + */ + protected array $promises = []; + + protected bool $isShutdownFunctionRegistered = false; + + public function __construct( + protected LoggerInterface $logger, + protected bool $isCLI, + ) { + } + + /** + * @param list $promises + */ + public function add(array $promises): void { + foreach ($promises as $promise) { + $this->promises[] = $promise; + } + } + + /** + * Make sure the added promises are waited for, either right away on the + * command line, or at the end of the request otherwise. + */ + public function schedule(): void { + if ($this->isCLI) { + $this->flushNow(); + return; + } + + if ($this->isShutdownFunctionRegistered) { + return; + } + $this->isShutdownFunctionRegistered = true; + $this->registerShutdownFunction($this->onShutdown(...)); + } + + /** + * Wait for all pending requests to finish + */ + public function flushNow(): void { + $promises = $this->promises; + $this->promises = []; + + foreach ($promises as $promise) { + try { + // Rejections are already handled by the callbacks of the + // requests, so the result must not be unwrapped as that would + // rethrow the exception and skip the remaining requests. + $promise->wait(false); + } catch (\Throwable $e) { + $this->logger->error('Error while sending push notifications: ' . $e->getMessage(), ['exception' => $e]); + } + } + } + + protected function onShutdown(): void { + $this->finishRequest(); + + // Flushing the response does not reset the execution timer, so a slow + // push endpoint could kill the process in the middle of the sending. + @set_time_limit(0); + + try { + $this->flushNow(); + } catch (\Throwable $e) { + // An exception escaping a shutdown function is close to + // impossible to attribute, so nothing may leave this method + $this->logger->error('Error while sending deferred push notifications: ' . $e->getMessage(), ['exception' => $e]); + } + } + + /** + * Send the response to the client, so it does not have to wait for the + * push notifications. Only PHP-FPM and LiteSpeed can do this, with e.g. + * mod_php the connection stays open until the sending is done. + */ + protected function finishRequest(): void { + if (function_exists('fastcgi_finish_request')) { + fastcgi_finish_request(); + } elseif (function_exists('litespeed_finish_request')) { + litespeed_finish_request(); + } + } + + protected function registerShutdownFunction(callable $callback): void { + register_shutdown_function($callback); + } +} diff --git a/lib/Push.php b/lib/Push.php index 1e4fc29d0..8718d971e 100644 --- a/lib/Push.php +++ b/lib/Push.php @@ -114,6 +114,7 @@ public function __construct( protected ITimeFactory $timeFactory, protected ISecureRandom $random, protected LoggerInterface $log, + protected DeferredFlusher $flusher, ) { $this->cache = $cacheFactory->createDistributed('pushtokens'); } @@ -195,8 +196,27 @@ public function flushPayloads(): void { } $this->deferPayloads = false; - $this->wpClient->flush(fn ($r) => $this->webPushCallback($r)); - $this->sendNotificationsToProxies(); + $this->flusher->add($this->wpClient->flushAsync(fn ($r) => $this->webPushCallback($r))); + $this->flusher->add($this->startNotificationsToProxies()); + $this->flusher->schedule(); + } + + /** + * Start sending the queued web push notifications and hand the pending + * requests over to the flusher + */ + protected function flushWebPushPayloads(): void { + $this->flusher->add($this->wpClient->flushAsync(fn ($r) => $this->webPushCallback($r))); + $this->flusher->schedule(); + } + + /** + * Start sending the queued push proxy notifications and hand the pending + * requests over to the flusher + */ + protected function flushProxyPayloads(): void { + $this->flusher->add($this->startNotificationsToProxies()); + $this->flusher->schedule(); } /** @@ -431,7 +451,7 @@ public function webPushToDevice(int $id, IUser $user, array $devices, INotificat $this->printInfo(''); if (!$this->deferPayloads) { - $this->wpClient->flush(fn ($r) => $this->webPushCallback($r)); + $this->flushWebPushPayloads(); } } @@ -479,7 +499,7 @@ public function proxyPushToDevice(int $id, IUser $user, array $devices, INotific $this->printInfo(''); if (!$this->deferPayloads) { - $this->sendNotificationsToProxies(); + $this->flushProxyPayloads(); } } @@ -604,7 +624,7 @@ public function webPushDeleteToDevice(string $userId, IUser $user, array $device } if (!$this->deferPayloads) { - $this->wpClient->flush(fn ($r) => $this->webPushCallback($r)); + $this->flushWebPushPayloads(); } } @@ -668,7 +688,7 @@ public function proxyPushDeleteToDevice(string $userId, IUser $user, array $devi } if (!$this->deferPayloads) { - $this->sendNotificationsToProxies(); + $this->flushProxyPayloads(); } } @@ -684,11 +704,16 @@ protected function webPushCallback(MessageSentReport $report): void { } } - protected function sendNotificationsToProxies(): void { + /** + * Start sending the queued notifications to the push proxies + * + * @return list The caller has to wait for the requests to finish + */ + protected function startNotificationsToProxies(): array { $pushNotifications = $this->payloadsToSend; $this->payloadsToSend = []; if (empty($pushNotifications)) { - return; + return []; } if (!$this->notificationManager->isFairUseOfFreePushService()) { @@ -697,7 +722,7 @@ protected function sendNotificationsToProxies(): void { * users overload our infrastructure. For this reason we have to rate-limit the * use of push notifications. If you need this feature, consider using Nextcloud Enterprise. */ - return; + return []; } $subscriptionAwareServer = rtrim($this->appConfig->getAppValueString('subscription_aware_server', 'https://push-notifications.nextcloud.com'), '/'); @@ -744,12 +769,7 @@ function (\Throwable $e) use ($proxyServer): void { } } - foreach ($promises as $promise) { - // Rejections are already taken care of by the handler above, so the - // result must not be unwrapped as that would rethrow the exception - // and skip the remaining proxy servers. - $promise->wait(false); - } + return $promises; } /** diff --git a/lib/WebPush/LoggerAdapter.php b/lib/WebPush/LoggerAdapter.php index 4e405cf42..f3fb23c48 100644 --- a/lib/WebPush/LoggerAdapter.php +++ b/lib/WebPush/LoggerAdapter.php @@ -20,7 +20,7 @@ */ class LoggerAdapter extends AbstractLogger { public function __construct( - private LoggerInterface $logger, + private readonly LoggerInterface $logger, ) { } diff --git a/tests/Unit/DeferredFlusherTest.php b/tests/Unit/DeferredFlusherTest.php new file mode 100644 index 000000000..11cf1cc73 --- /dev/null +++ b/tests/Unit/DeferredFlusherTest.php @@ -0,0 +1,119 @@ +registeredShutdownFunctions++; + $this->shutdownFunction = $callback(...); + } +} + +class DeferredFlusherTest extends TestCase { + protected LoggerInterface&MockObject $logger; + + protected function setUp(): void { + parent::setUp(); + $this->logger = $this->createMock(LoggerInterface::class); + } + + /** + * @param int $waited Counts how often the promise was waited for + */ + protected function createPromise(int &$waited, ?\Throwable $throw = null): IPromise&MockObject { + $promise = $this->createMock(IPromise::class); + $promise->method('wait') + ->willReturnCallback(function (bool $unwrap = true) use (&$waited, $throw): mixed { + $waited++; + // The rejections are handled by the callbacks of the requests + $this->assertFalse($unwrap, 'The result must not be unwrapped'); + if ($throw !== null) { + throw $throw; + } + return null; + }); + + return $promise; + } + + public function testCommandLineSendsRightAway(): void { + $flusher = new TestingDeferredFlusher($this->logger, true); + + $waited = 0; + $flusher->add([$this->createPromise($waited)]); + $flusher->schedule(); + + $this->assertSame(1, $waited); + $this->assertSame(0, $flusher->registeredShutdownFunctions); + } + + public function testWebRequestSendsOnShutdown(): void { + $flusher = new TestingDeferredFlusher($this->logger, false); + + $waited = 0; + $flusher->add([$this->createPromise($waited)]); + $flusher->schedule(); + $flusher->add([$this->createPromise($waited)]); + // Scheduling again must not register a second shutdown function + $flusher->schedule(); + + $this->assertSame(0, $waited, 'The request must not wait for the push notifications'); + $this->assertSame(1, $flusher->registeredShutdownFunctions); + + ($flusher->shutdownFunction)(); + + $this->assertSame(2, $waited); + } + + public function testPromisesAreOnlyWaitedForOnce(): void { + $flusher = new TestingDeferredFlusher($this->logger, false); + + $waited = 0; + $flusher->add([$this->createPromise($waited)]); + $flusher->schedule(); + + ($flusher->shutdownFunction)(); + ($flusher->shutdownFunction)(); + + $this->assertSame(1, $waited); + } + + public function testFailingRequestDoesNotSkipTheOthers(): void { + $flusher = new TestingDeferredFlusher($this->logger, true); + + $this->logger->expects($this->once()) + ->method('error'); + + $failing = 0; + $succeeding = 0; + $flusher->add([ + $this->createPromise($failing, new \RuntimeException('Nope')), + $this->createPromise($succeeding), + ]); + $flusher->schedule(); + + $this->assertSame(1, $failing); + $this->assertSame(1, $succeeding); + } +} diff --git a/tests/Unit/PushTest.php b/tests/Unit/PushTest.php index 9f0ee6960..b3f7ffdcb 100644 --- a/tests/Unit/PushTest.php +++ b/tests/Unit/PushTest.php @@ -15,8 +15,13 @@ use OC\Authentication\Token\PublicKeyToken; use OC\Security\IdentityProof\Key; use OC\Security\IdentityProof\Manager; +use OCA\Notifications\DeferredFlusher; use OCA\Notifications\Exceptions\InvalidDeviceTokenException; use OCA\Notifications\Push; +use OCA\Notifications\Vendor\GuzzleHttp\Promise\PromiseInterface; +use OCA\Notifications\Vendor\GuzzleHttp\Psr7\Request; +use OCA\Notifications\Vendor\GuzzleHttp\Psr7\Response; +use OCA\Notifications\Vendor\Minishlink\WebPush\MessageSentReport; use OCA\Notifications\WebPushClient; use OCP\AppFramework\Http; use OCP\AppFramework\Services\IAppConfig; @@ -66,6 +71,7 @@ class PushTest extends TestCase { protected ISecureRandom&MockObject $random; protected LoggerInterface&MockObject $logger; protected IUserManager&MockObject $userManager; + protected DeferredFlusher&MockObject $flusher; public const EX_UA_PUBLIC = 'BCVxsr7N_eNgVRqvHtD0zTZsEc6-VV-JvLexhqUzORcx aOzi6-AYWXvTBHm4bjyPjs7Vd8pZGH6SRpkNtoIAiw4'; public const EX_AUTH = 'BTBZMqHH6r4Tts7J_aSIgg'; @@ -90,6 +96,7 @@ protected function setUp(): void { $this->random = $this->createMock(ISecureRandom::class); $this->logger = $this->createMock(LoggerInterface::class); $this->userManager = $this->createMock(IUserManager::class); + $this->flusher = $this->createMock(DeferredFlusher::class); $this->cacheFactory->method('createDistributed') ->with('pushtokens') @@ -123,6 +130,7 @@ protected function getPush(array $methods = []): Push|MockObject { $this->timeFactory, $this->random, $this->logger, + $this->flusher, ]) ->onlyMethods($methods) ->getMock(); @@ -145,6 +153,7 @@ protected function getPush(array $methods = []): Push|MockObject { $this->timeFactory, $this->random, $this->logger, + $this->flusher, ); } @@ -1223,7 +1232,7 @@ public function testWebPushToDeviceSending(bool $isRateLimited): void { } $this->wpClient->expects($this->once()) - ->method('flush'); + ->method('flushAsync'); $this->config->expects($this->once()) ->method('getSystemValueBool') @@ -1249,7 +1258,7 @@ public function testWebPushDeleteToDeviceFlushesWebPush(): void { // The queued deletes have to be sent right away, they must not leak // into an unrelated later flush $this->wpClient->expects($this->once()) - ->method('flush'); + ->method('flushAsync'); // Nothing was queued for the push proxies $this->clientService->expects($this->never()) @@ -1267,6 +1276,94 @@ public function testWebPushDeleteToDeviceFlushesWebPush(): void { ], true, null); } + public function testFlushPayloadsDefersTheSending(): void { + $push = $this->getPush(['startNotificationsToProxies']); + + $webPushPromise = $this->createMock(PromiseInterface::class); + $proxyPromise = $this->createMock(IPromise::class); + + $this->wpClient->expects($this->once()) + ->method('flushAsync') + ->willReturn([$webPushPromise]); + + $push->expects($this->once()) + ->method('startNotificationsToProxies') + ->willReturn([$proxyPromise]); + + $added = []; + $this->flusher->expects($this->exactly(2)) + ->method('add') + ->willReturnCallback(static function (array $promises) use (&$added): void { + $added[] = $promises; + }); + $this->flusher->expects($this->once()) + ->method('schedule'); + + // The requests must not be waited for while the payloads are flushed + $webPushPromise->expects($this->never()) + ->method('wait'); + $proxyPromise->expects($this->never()) + ->method('wait'); + + $push->deferPayloads(); + $push->flushPayloads(); + + $this->assertSame([[$webPushPromise], [$proxyPromise]], $added); + } + + /** + * Capture the callback the deferred web push requests report to + */ + protected function getWebPushCallback(Push $push): callable { + $callback = null; + $this->wpClient->method('flushAsync') + ->willReturnCallback(static function (callable $c) use (&$callback): array { + $callback = $c; + return []; + }); + + $push->deferPayloads(); + $push->flushPayloads(); + + $this->assertIsCallable($callback); + return $callback; + } + + public function testWebPushCallbackDeletesExpiredSubscriptionWhenDeferred(): void { + $push = $this->getPush(['deleteWebPushTokenByEndpoint']); + + $push->expects($this->once()) + ->method('deleteWebPushTokenByEndpoint') + ->with('https://push.example.tld/endpoint'); + + $callback = $this->getWebPushCallback($push); + + // Happens when the flusher waits for the requests, e.g. at the end of + // the request + $callback(new MessageSentReport( + new Request('POST', 'https://push.example.tld/endpoint'), + new Response(410), + )); + } + + public function testWebPushCallbackHandlesRateLimitWhenDeferred(): void { + $push = $this->getPush(['deleteWebPushTokenByEndpoint']); + + $push->expects($this->never()) + ->method('deleteWebPushTokenByEndpoint'); + + $this->cache->expects($this->once()) + ->method('set') + ->with('wp.https://push.example.tld/endpoint', true, 120); + + $callback = $this->getWebPushCallback($push); + + $callback(new MessageSentReport( + new Request('POST', 'https://push.example.tld/endpoint'), + new Response(429, ['Retry-After' => '120']), + )); + } + public static function dataFilterWebPushDeviceList(): array { return [ [false, 'all', 'myApp', false],