perf(webpush): Improve performance of webpush notifications

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);
 	}
 
 	/**

lib/DeferredFlusher.php

@@ -0,0 +1,124 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+
+namespace OCA\Notifications;
+
+use OCA\Notifications\Vendor\GuzzleHttp\Promise\PromiseInterface;
+use OCP\Http\Client\IPromise;
+use Psr\Log\LoggerInterface;
+
+/**
+ * Collects the pending push notification requests and decides when to wait
+ * for them.
+ *
+ * On the command line the requests are sent right away, so the output of e.g.
+ * `occ notification:test-push` stays in order and a cron run does not pile up
+ * the push traffic of all jobs at the very end.
+ *
+ * For web requests the sending is postponed to the shutdown of the request.
+ * When the SAPI supports it, the response is sent to the client before the
+ * push notifications are sent, so the user does not wait for the push
+ * endpoints to answer.
+ */
+class DeferredFlusher {
+	/**
+	 * Guzzle promises of the web push requests and IPromises of the requests
+	 * to the push proxies. The two HTTP stacks are separate, so they can not
+	 * run at the same time, but both have a wait() method.
+	 *
+	 * @var list<PromiseInterface|IPromise>
+	 */
+	protected array $promises = [];
+
+	protected bool $isShutdownFunctionRegistered = false;
+
+	public function __construct(
+		protected LoggerInterface $logger,
+		protected bool $isCLI,
+	) {
+	}
+
+	/**
+	 * @param list<PromiseInterface|IPromise> $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);
+	}
+}

lib/PooledWebPush.php

@@ -0,0 +1,108 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+
+namespace OCA\Notifications;
+
+use OCA\Notifications\Vendor\GuzzleHttp\Pool;
+use OCA\Notifications\Vendor\GuzzleHttp\Promise\PromiseInterface;
+use OCA\Notifications\Vendor\Minishlink\WebPush\MessageSentReport;
+use OCA\Notifications\Vendor\Minishlink\WebPush\WebPush;
+use OCA\Notifications\Vendor\Psr\Http\Message\RequestInterface;
+use OCA\Notifications\Vendor\Psr\Http\Message\ResponseInterface;
+
+/**
+ * Extends the Mozart prefixed copy of minishlink/web-push (v10.1.0) to send
+ * the queued notifications without blocking on the result.
+ *
+ * This depends on WebPush::$notifications, ::$client, ::$defaultOptions,
+ * ::$vapidHeaders, ::prepare() and ::createRejectedReport() staying protected,
+ * so it has to be re-checked whenever the library is bumped.
+ */
+class PooledWebPush extends WebPush {
+	/**
+	 * Same as flushPooled(), but returns the promises of the pools instead of
+	 * waiting for them, and prepares (encrypts) each request only when the
+	 * pool is ready to send it, so the encryption of the next payload happens
+	 * while the previous requests are in flight.
+	 *
+	 * @param callable(MessageSentReport): void $callback Callback for each notification
+	 * @param null|int $batchSize Defaults to the value defined in defaultOptions during instantiation
+	 * @param null|int $requestConcurrency Defaults to the value defined in defaultOptions during instantiation
+	 * @return list<PromiseInterface> 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<int, RequestInterface> $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 = [];
+	}
+}

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->sendNotificationsToProxies();
+			$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<IPromise> 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'), '/');
@@ -734,7 +759,7 @@ function (IResponse $response) use ($proxyServer, $postStartTime): void {
 							$this->printInfo('<comment>Request to push proxy [' . $proxyServer . '] took ' . (string)round(microtime(true) - $postStartTime, 2) . 's</comment>');
 							$this->handleProxyResponse($proxyServer, $response->getStatusCode(), (string)$response->getBody());
 						},
-						function (\Exception $e) use ($proxyServer): void {
+						function (\Throwable $e) use ($proxyServer): void {
 							$this->handleProxyException($proxyServer, $e);
 						},
 					);
@@ -744,18 +769,13 @@ function (\Exception $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;
 	}
 
 	/**
 	 * 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();