taler-typescript-core

commit e5f13e4ec1f45ea4b0751d45f6c6019c7c0623db
parent df3261ea71a81fa1d7c7e2d2cc64eb46661ecc58
Author: Florian Dold <florian@dold.me>
Date:   Fri, 17 Jul 2026 19:38:13 +0200

wallet-core: fix race in shepherd

Diffstat:
M
packages/taler-wallet-core/src/shepherd.ts
 | 
43
+++++++++++++++++++++++++++++++++++++++++++

1 file changed, 43 insertions(+), 0 deletions(-)
diff --git a/packages/taler-wallet-core/src/shepherd.ts b/packages/taler-wallet-core/src/shepherd.ts
@@ -141,6 +141,15 @@ interface ShepherdInfo {
   taskState: TaskState;
 
   /**
+   * 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.
+   */
+  rerunRequested: boolean;
+
+  /**
    * Unique numeric identifier for the shepherd.
    */
   shepId: number;
@@ -324,6 +333,10 @@ export class TaskSchedulerImpl implements TaskScheduler {
       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:
@@ -347,6 +360,7 @@ export class TaskSchedulerImpl implements TaskScheduler {
       cts: CancellationToken.create(),
       taskState: TaskState.Running,
       interruptWait: () => undefined,
+      rerunRequested: false,
       shepId: this.shepCounter++,
     };
     logger.trace(
@@ -363,10 +377,27 @@ export class TaskSchedulerImpl implements TaskScheduler {
     } finally {
       logger.trace(`Done shepherding ${taskId}`);
       newShep.cts.cancel();
+      let deleted = false;
       if (this.sheps.get(taskId)?.shepId === newShep.shepId) {
         this.sheps.delete(taskId);
+        deleted = true;
       }
       this.iterCond.trigger();
+      // Close the residual race: a wakeup may have arrived after the loop
+      // decided to terminate but before we removed the shepherd. Only restart
+      // when this shepherd terminated on its own (not via stop/cancel), so we
+      // don't resurrect a task that was deliberately stopped.
+      if (
+        deleted &&
+        newShep.rerunRequested &&
+        newShep.taskState !== TaskState.StopRequested &&
+        !this.ws.stopped
+      ) {
+        logger.trace(
+          `rerun requested for ${taskId} after shepherd finished, restarting`,
+        );
+        this.startShepherdTask(taskId);
+      }
     }
   }
 
@@ -466,6 +497,9 @@ export class TaskSchedulerImpl implements TaskScheduler {
         info.cts.token,
       );
       const startTime = AbsoluteTime.now();
+      // Reset before running the handler so that any wakeup arriving while the
+      // handler (and its result storage) runs is captured and honored below.
+      info.rerunRequested = false;
       logger.trace(`Shepherd for ${taskId} will call handler`);
       let res: TaskRunResult;
       try {
@@ -537,6 +571,15 @@ export class TaskSchedulerImpl implements TaskScheduler {
         }
         case TaskRunResultType.Finished:
           await storePendingTaskFinished(this.ws, taskId);
+          if (info.rerunRequested) {
+            // A wakeup arrived while we were finishing.
+            // Re-run instead of terminating, otherwise the
+            // task would be stuck until the next unrelated wakeup.
+            logger.trace(
+              `task ${taskId} finished but a rerun was requested, running again`,
+            );
+            break;
+          }
           return;
         case TaskRunResultType.LongpollReturnedPending: {
           await storeTaskProgress(this.ws, taskId);