commit dfce0e58c5fa57e2f865d345de6d9a0174241b33
parent 0e4807845fc06f7c705711974650d67ef30e7919
Author: Florian Dold <florian@dold.me>
Date: Wed, 15 Jul 2026 17:47:05 +0200
util: fix longpoll permit management issue
Diffstat:
3 files changed, 98 insertions(+), 86 deletions(-)
diff --git a/packages/taler-util/src/index.ts b/packages/taler-util/src/index.ts
@@ -39,7 +39,7 @@ export { sha256, HashSha256 } from "./sha256.js";
export * from "./libtool-version.js";
export * from "./logging.js";
-export * from "./longpool-queue.js";
+export * from "./longpoll-queue.js";
export {
crypto_sign_keyPair_fromSeed,
randomBytes,
diff --git a/packages/taler-util/src/longpoll-queue.ts b/packages/taler-util/src/longpoll-queue.ts
@@ -0,0 +1,97 @@
+/*
+ This file is part of GNU Taler
+ (C) 2025 Taler Systems S.A.
+
+ GNU Taler is free software; you can redistribute it and/or modify it under the
+ terms of the GNU General Public License as published by the Free Software
+ Foundation; either version 3, or (at your option) any later version.
+
+ GNU Taler is distributed in the hope that it will be useful, but WITHOUT ANY
+ WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+ A PARTICULAR PURPOSE. See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along with
+ GNU Taler; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+
+ SPDX-License-Identifier: AGPL3.0-or-later
+*/
+
+import { CancellationToken } from "./CancellationToken.js";
+import { Logger } from "./logging.js";
+import { openPromise } from "./promises.js";
+
+const logger = new Logger("longpoll-queue.ts");
+
+const PERMITS: number = 20;
+type LongpollRunFn<T> = (timeoutMs: number) => Promise<T>;
+
+export class LongpollQueue {
+ private idCounter: number = 0;
+ private queue: (() => void)[] = [];
+ private permits: number = PERMITS;
+
+ async run<T>(
+ url: URL,
+ cancellationToken: CancellationToken,
+ f: LongpollRunFn<T>,
+ ): Promise<T> {
+ const hostname = url.hostname;
+ const rid = this.idCounter++;
+
+ const triggerNextLongpoll = () => {
+ logger.trace(`cleaning up after long-poll ${rid} to ${hostname}`);
+ // Run pending task
+ const next = this.queue.shift();
+ if (next != null) {
+ next();
+ } else {
+ // Else release permit
+ this.permits++;
+ }
+ };
+
+ const doRunLongpoll: () => Promise<T> = async () => {
+ const numWaiting = this.queue.length;
+ const numConcurrent = PERMITS - this.permits;
+ logger.info(
+ `running long-poll ${rid} to ${hostname} with ${numWaiting} waiting and ${numConcurrent} running`,
+ );
+ try {
+ const timeoutMs = Math.round(Math.max(10000, 30000 / (numWaiting + 1)));
+ return await f(timeoutMs);
+ } finally {
+ triggerNextLongpoll();
+ }
+ };
+
+ if (this.permits > 0) {
+ this.permits--;
+ return doRunLongpoll();
+ } else {
+ logger.info(`long-poll ${rid} to ${hostname} queued`);
+ const promcap = openPromise<void>();
+ const entry = promcap.resolve;
+ this.queue.push(entry);
+ try {
+ await cancellationToken.racePromise(promcap.promise);
+ } catch (e) {
+ // The wait was interrupted (cancelled/errored) rather than woken
+ // normally. If our entry is still queued we were never handed a
+ // slot, so just remove ourselves without releasing a permit. If it
+ // was already dequeued (a departing task handed us a slot before the
+ // cancellation won the race), release that slot so it isn't leaked.
+ const idx = this.queue.indexOf(entry);
+ if (idx >= 0) {
+ this.queue.splice(idx, 1);
+ } else {
+ triggerNextLongpoll();
+ }
+ logger.info(`long-poll ${rid} to ${hostname} cancelled while queued`);
+ throw e;
+ }
+ // Woken normally: a departing task dequeued our entry and handed its
+ // slot to us. doRunLongpoll() is the sole releaser of that slot.
+ return doRunLongpoll();
+ }
+ }
+}
diff --git a/packages/taler-util/src/longpool-queue.ts b/packages/taler-util/src/longpool-queue.ts
@@ -1,85 +0,0 @@
-/*
- This file is part of GNU Taler
- (C) 2025 Taler Systems S.A.
-
- GNU Taler is free software; you can redistribute it and/or modify it under the
- terms of the GNU General Public License as published by the Free Software
- Foundation; either version 3, or (at your option) any later version.
-
- GNU Taler is distributed in the hope that it will be useful, but WITHOUT ANY
- WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
- A PARTICULAR PURPOSE. See the GNU General Public License for more details.
-
- You should have received a copy of the GNU General Public License along with
- GNU Taler; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
-
- SPDX-License-Identifier: AGPL3.0-or-later
-*/
-
-import { CancellationToken } from "./CancellationToken.js";
-import { Logger } from "./logging.js";
-import { openPromise } from "./promises.js";
-
-const logger = new Logger("longpoll-queue.ts");
-
-const PERMITS: number = 20;
-type LongpollRunFn<T> = (timeoutMs: number) => Promise<T>;
-
-export class LongpollQueue {
- private idCounter: number = 0;
- private queue: (() => void)[] = [];
- private permits: number = PERMITS;
-
- constructor() {}
-
- async run<T>(
- url: URL,
- cancellationToken: CancellationToken,
- f: LongpollRunFn<T>,
- ): Promise<T> {
- const hostname = url.hostname;
- const rid = this.idCounter++;
-
- const triggerNextLongpoll = () => {
- logger.trace(`cleaning up after long-poll ${rid} to ${hostname}`);
- // Run pending task
- const next = this.queue.shift();
- if (next != null) {
- next();
- } else {
- // Else release permit
- this.permits++;
- }
- };
-
- const doRunLongpoll: () => Promise<T> = async () => {
- const numWaiting = this.queue.length;
- const numConcurrent = PERMITS - this.permits;
- logger.info(
- `running long-poll ${rid} to ${hostname} with ${numWaiting} waiting and ${numConcurrent} running`,
- );
- try {
- const timeoutMs = Math.round(Math.max(10000, 30000 / (numWaiting + 1)));
- return await f(timeoutMs);
- } finally {
- triggerNextLongpoll();
- }
- };
-
- if (this.permits > 0) {
- this.permits--;
- return doRunLongpoll();
- } else {
- logger.info(`long-poll ${rid} to ${hostname} queued`);
- const promcap = openPromise<void>();
- this.queue.push(promcap.resolve);
- try {
- await cancellationToken.racePromise(promcap.promise);
- } finally {
- logger.info(`long-poll ${rid} to ${hostname} cancelled while queued`);
- triggerNextLongpoll();
- }
- return doRunLongpoll();
- }
- }
-}
