commit 246905ec79c4432095c07413a26e58b7588aafd6
parent d2e770e0ff271cc0477fa317b5572d726a51d6cb
Author: Florian Dold <dold@taler.net>
Date: Wed, 22 Jul 2026 17:10:16 +0200
wallet: deliver a wakeup whatever the shepherd is doing
Diffstat:
1 file changed, 37 insertions(+), 53 deletions(-)
diff --git a/packages/taler-wallet-core/src/shepherd.ts b/packages/taler-wallet-core/src/shepherd.ts
@@ -115,14 +115,6 @@ import { WalletDbTransaction } from "./dbtx.js";
const logger = new Logger("shepherd.ts");
-enum TaskState {
- StopRequested = 0,
- Running = 1,
- WaitingBackoff = 2,
- WaitingScheduled = 3,
- Done = 4,
-}
-
/**
* Info about one task being shepherded.
*/
@@ -134,16 +126,25 @@ interface ShepherdInfo {
*/
latch?: Promise<void>;
+ /**
+ * Cut the current wait short. Only installed while the loop is actually
+ * waiting; a no-op otherwise, so it is always safe to call.
+ */
interruptWait: () => void;
- taskState: TaskState;
+ /**
+ * Set when this shepherd has been asked to stop. It then no longer accepts
+ * wakeups and is replaced rather than reused.
+ */
+ stopRequested: boolean;
/**
- * Set when a wakeup (startShepherdTask) arrives while this shepherd is
- * already Running and can therefore not be delivered via interruptWait().
- * The shepherd loop re-runs the handler instead of terminating so that a
- * transition back into a working state right as the task finishes is not
- * lost.
+ * Set by a wakeup (startShepherdTask) for a shepherd that already exists.
+ *
+ * This is what makes a wakeup survive until the loop can act on it, whatever
+ * the loop happens to be doing at the time: the loop clears it before each
+ * handler run, skips waiting while it is set, and re-runs instead of
+ * terminating when the handler reports that it is finished.
*/
rerunRequested: boolean;
@@ -328,26 +329,16 @@ export class TaskSchedulerImpl implements TaskScheduler {
logger.trace(`Starting to shepherd task ${taskId}`);
const oldShep = this.sheps.get(taskId);
if (oldShep) {
- switch (oldShep.taskState) {
- case TaskState.WaitingBackoff:
- case TaskState.Running:
- // We can't interrupt a running/backing-off shepherd, so record that
- // another run is required. The loop (and the finally below) will
- // honor this so the wakeup is not silently dropped.
- oldShep.rerunRequested = true;
- logger.trace(`Already have a shepherd for ${taskId}`);
- return;
- case TaskState.WaitingScheduled:
- logger.trace(
- `Already have a shepherd (waiting) for ${taskId}, kicking shepherd`,
- );
- oldShep.interruptWait();
- return;
- case TaskState.StopRequested:
- case TaskState.Done:
- break;
- default:
- assertUnreachable(oldShep.taskState);
+ if (!oldShep.stopRequested) {
+ // Both of these, always. Recording the wakeup is what makes it
+ // survive; interrupting the wait only shortens a sleep that happens
+ // to be in progress and is a no-op otherwise. Deciding between them
+ // would mean knowing where the loop currently is, and getting that
+ // wrong drops the wakeup.
+ oldShep.rerunRequested = true;
+ oldShep.interruptWait();
+ logger.trace(`Already have a shepherd for ${taskId}`);
+ return;
}
logger.trace(
`Waiting for old task to complete the loop in cancel mode ${taskId}`,
@@ -356,7 +347,7 @@ export class TaskSchedulerImpl implements TaskScheduler {
}
const newShep: ShepherdInfo = {
cts: CancellationToken.create(),
- taskState: TaskState.Running,
+ stopRequested: false,
interruptWait: () => undefined,
rerunRequested: false,
shepId: this.shepCounter++,
@@ -388,7 +379,7 @@ export class TaskSchedulerImpl implements TaskScheduler {
if (
deleted &&
newShep.rerunRequested &&
- newShep.taskState !== TaskState.StopRequested &&
+ !newShep.stopRequested &&
!this.ws.stopped
) {
logger.trace(
@@ -404,18 +395,9 @@ export class TaskSchedulerImpl implements TaskScheduler {
const oldShep = this.sheps.get(taskId);
if (oldShep) {
logger.trace(`Cancelling old shepherd for ${taskId}`);
- switch (oldShep.taskState) {
- case TaskState.Running:
- case TaskState.WaitingBackoff:
- case TaskState.WaitingScheduled:
- oldShep.cts.cancel(`stopping task ${taskId}`);
- oldShep.taskState = TaskState.StopRequested;
- break;
- case TaskState.Done:
- case TaskState.StopRequested:
- break;
- default:
- assertUnreachable(oldShep.taskState);
+ if (!oldShep.stopRequested) {
+ oldShep.cts.cancel(`stopping task ${taskId}`);
+ oldShep.stopRequested = true;
}
this.iterCond.trigger();
}
@@ -444,6 +426,13 @@ export class TaskSchedulerImpl implements TaskScheduler {
info: ShepherdInfo,
delay: Duration,
): Promise<void> {
+ if (info.rerunRequested) {
+ // A wakeup arrived while the handler was running. Waiting it out would
+ // mean answering it only after the delay that the wakeup was meant to
+ // cut short.
+ logger.trace(`Not waiting on ${taskId}, a rerun is already requested`);
+ return;
+ }
// Limit how long we wait.
// The NodeJS runtime doesn't like very long timeouts.
delay = Duration.min(delay, Duration.fromSpec({ hours: 1 }));
@@ -537,7 +526,6 @@ export class TaskSchedulerImpl implements TaskScheduler {
const t = timestampAbsoluteFromDb(retryRecord.retryInfo.nextRetry);
const delay = AbsoluteTime.remaining(t);
logger.trace(`Stored error for ${taskId},`);
- info.taskState = TaskState.WaitingBackoff;
await this.wait(taskId, info, delay);
break;
}
@@ -546,7 +534,6 @@ export class TaskSchedulerImpl implements TaskScheduler {
const t = timestampAbsoluteFromDb(retryRecord.retryInfo.nextRetry);
const delay = AbsoluteTime.remaining(t);
logger.trace(`Waiting on ${taskId} (backoff)`);
- info.taskState = TaskState.WaitingBackoff;
await this.wait(taskId, info, delay);
break;
}
@@ -563,7 +550,6 @@ export class TaskSchedulerImpl implements TaskScheduler {
const t = timestampAbsoluteFromDb(retryRecord.retryInfo.nextRetry);
const delay = AbsoluteTime.remaining(t);
logger.trace(`Waiting on ${taskId} (scheduled later)`);
- info.taskState = TaskState.WaitingScheduled;
await this.wait(taskId, info, delay);
break;
}
@@ -588,7 +574,6 @@ export class TaskSchedulerImpl implements TaskScheduler {
logger.info(
`long-poller for ${taskId} returned unexpectedly early (${taskDuration.d_ms} ms), waiting 10 seconds`,
);
- info.taskState = TaskState.WaitingScheduled;
await this.wait(taskId, info, Duration.fromSpec({ seconds: 10 }));
} else {
logger.info(`task ${taskId} will long-poll again`);
@@ -599,7 +584,6 @@ export class TaskSchedulerImpl implements TaskScheduler {
await storePendingTaskPending(this.ws, taskId);
const delay = Duration.getForever();
logger.trace(`Not retrying task until network is restored.`);
- info.taskState = TaskState.WaitingScheduled;
await this.wait(taskId, info, delay);
break;
}
