Remove beforeThreadsWakeUp callback and condition watcher mechanism

With RepeatableWorkflowThread, condition evaluation now happens inside
workflow threads themselves, making the beforeThreadsWakeUp callback
and condition watcher mechanism obsolete.

- Remove beforeThreadsWakeUp parameter from DeterministicRunner
- Remove conditionWatchers, registerConditionWatcher, evaluateConditionWatchers
  from SyncWorkflowContext
- Remove ConditionWatcher class
- Remove getBeforeThreadsWakeUpCallback method
- Remove related tests

Author: mfateev

temporal-sdk/src/main/java/io/temporal/internal/sync/DeterministicRunner.java

@@ -30,7 +30,7 @@ static DeterministicRunner newRunner(
       SyncWorkflowContext workflowContext,
       Runnable root,
       WorkflowExecutorCache cache) {
-    return new DeterministicRunnerImpl(workflowThreadExecutor, workflowContext, root, cache, null);
+    return new DeterministicRunnerImpl(workflowThreadExecutor, workflowContext, root, cache);
   }
 
   /**
@@ -45,29 +45,7 @@ static DeterministicRunner newRunner(
       WorkflowThreadExecutor workflowThreadExecutor,
       SyncWorkflowContext workflowContext,
       Runnable root) {
-    return new DeterministicRunnerImpl(workflowThreadExecutor, workflowContext, root, null, null);
-  }
-
-  /**
-   * Create new instance of DeterministicRunner with a callback invoked before threads wake up.
-   *
-   * @param workflowThreadExecutor executor for workflow thread Runnables
-   * @param workflowContext workflow context to use
-   * @param root function that root thread of the runner executes.
-   * @param cache WorkflowExecutorCache used cache inflight workflows
-   * @param beforeThreadsWakeUp callback invoked once per loop iteration before threads run. Returns
-   *     true if progress was made (e.g., a condition watcher fired), which causes the loop to
-   *     continue even if all threads are blocked. Returns false if no progress was made.
-   * @return instance of the DeterministicRunner.
-   */
-  static DeterministicRunner newRunner(
-      WorkflowThreadExecutor workflowThreadExecutor,
-      SyncWorkflowContext workflowContext,
-      Runnable root,
-      WorkflowExecutorCache cache,
-      @Nullable Supplier<Boolean> beforeThreadsWakeUp) {
-    return new DeterministicRunnerImpl(
-        workflowThreadExecutor, workflowContext, root, cache, beforeThreadsWakeUp);
+    return new DeterministicRunnerImpl(workflowThreadExecutor, workflowContext, root, null);
   }
 
   /**

temporal-sdk/src/main/java/io/temporal/internal/sync/DeterministicRunnerImpl.java

@@ -69,9 +69,6 @@ class DeterministicRunnerImpl implements DeterministicRunner {
   // always accessed under the runner lock
   private final List<NamedRunnable> toExecuteInWorkflowThread = new ArrayList<>();
 
-  // Callback invoked before threads wake up in each event loop iteration
-  @Nullable private final Supplier<Boolean> beforeThreadsWakeUp;
-
   // Access to workflowThreadsToAdd, callbackThreadsToAdd, addedThreads doesn't have to be
   // synchronized.
   // Inside DeterministicRunner the access to these variables is under the runner lock.
@@ -147,30 +144,20 @@ static void setCurrentThreadInternal(WorkflowThread coroutine) {
       WorkflowThreadExecutor workflowThreadExecutor,
       @Nonnull SyncWorkflowContext workflowContext,
       Runnable root) {
-    this(workflowThreadExecutor, workflowContext, root, null, null);
+    this(workflowThreadExecutor, workflowContext, root, null);
   }
 
   DeterministicRunnerImpl(
       WorkflowThreadExecutor workflowThreadExecutor,
       @Nonnull SyncWorkflowContext workflowContext,
       Runnable root,
       WorkflowExecutorCache cache) {
-    this(workflowThreadExecutor, workflowContext, root, cache, null);
-  }
-
-  DeterministicRunnerImpl(
-      WorkflowThreadExecutor workflowThreadExecutor,
-      @Nonnull SyncWorkflowContext workflowContext,
-      Runnable root,
-      WorkflowExecutorCache cache,
-      @Nullable Supplier<Boolean> beforeThreadsWakeUp) {
     this.workflowThreadExecutor = workflowThreadExecutor;
     this.workflowContext = Preconditions.checkNotNull(workflowContext, "workflowContext");
     // TODO this should be refactored, publishing of this in an constructor into external objects is
     // a bad practice
     this.workflowContext.setRunner(this);
     this.cache = cache;
-    this.beforeThreadsWakeUp = beforeThreadsWakeUp;
     boolean deterministicCancellationScopeOrder =
         workflowContext
             .getReplayContext()
@@ -222,14 +209,7 @@ public void runUntilAllBlocked(long deadlockDetectionTimeout) {
         }
         toExecuteInWorkflowThread.clear();
 
-        // Invoke beforeThreadsWakeUp callback BEFORE running threads.
-        // This allows the callback to evaluate conditions and complete promises,
-        // ensuring threads see updated state when they wake up.
-        if (beforeThreadsWakeUp != null) {
-          progress = beforeThreadsWakeUp.get();
-        } else {
-          progress = false;
-        }
+        progress = false;
 
         Iterator<WorkflowThread> ci = threads.iterator();
         while (ci.hasNext()) {

temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflow.java

@@ -124,8 +124,7 @@ public void start(HistoryEvent event, ReplayWorkflowContext context) {
                           context.getWorkflowExecution()))
                   .start();
             },
-            cache,
-            workflowContext.getBeforeThreadsWakeUpCallback());
+            cache);
   }
 
   @Override

temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java

@@ -54,7 +54,6 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.util.*;
-import java.util.Iterator;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiPredicate;
@@ -105,9 +104,6 @@ final class SyncWorkflowContext implements WorkflowContext, WorkflowOutboundCall
   private final WorkflowThreadLocal<UpdateInfo> currentUpdateInfo = new WorkflowThreadLocal<>();
   @Nullable private String currentDetails;
 
-  // Condition watchers for async await functionality
-  private final List<ConditionWatcher> conditionWatchers = new ArrayList<>();
-
   public SyncWorkflowContext(
       @Nonnull String namespace,
       @Nonnull WorkflowExecution workflowExecution,
@@ -1675,93 +1671,4 @@ public Failure getFailure() {
       return failure;
     }
   }
-
-  /**
-   * Returns a callback to be used by DeterministicRunner before threads wake up. This callback
-   * evaluates condition watchers and completes promises as needed.
-   */
-  public Supplier<Boolean> getBeforeThreadsWakeUpCallback() {
-    return this::evaluateConditionWatchers;
-  }
-
-  /**
-   * Registers a condition watcher for async await functionality. The condition is evaluated
-   * immediately (inline) before registering. If already satisfied, returns immediately without
-   * registering a watcher. Otherwise, the condition is re-evaluated at the end of each event loop
-   * iteration until it returns true.
-   *
-   * <p>IMPORTANT: The condition must never throw exceptions. If it does, the workflow task will
-   * fail. Callers should handle exceptions within the condition supplier and complete their promise
-   * appropriately before returning true.
-   *
-   * @param condition Supplier that returns true when the wait is complete (caller handles promise
-   *     completion in the supplier body). Evaluated in read-only mode. Must not throw exceptions.
-   * @return A Runnable that cancels the watcher when invoked (no-op if condition was already
-   *     satisfied).
-   */
-  Runnable registerConditionWatcher(Supplier<Boolean> condition) {
-    // Evaluate condition inline - if already satisfied, no need to register
-    setReadOnly(true);
-    try {
-      if (condition.get()) {
-        return () -> {};
-      }
-    } finally {
-      setReadOnly(false);
-    }
-
-    ConditionWatcher watcher = new ConditionWatcher(condition);
-    conditionWatchers.add(watcher);
-    return () -> watcher.canceled = true;
-  }
-
-  /**
-   * Evaluates all condition watchers and removes those that return true. Watchers that are
-   * satisfied are removed from the list.
-   *
-   * <p>Note: If a condition throws an exception, it will propagate and fail the workflow task.
-   * Callers should handle exceptions within their condition supplier.
-   *
-   * @return true if any condition was satisfied (indicating progress was made)
-   */
-  private boolean evaluateConditionWatchers() {
-    boolean anyMatched = false;
-    Iterator<ConditionWatcher> it = conditionWatchers.iterator();
-    while (it.hasNext()) {
-      ConditionWatcher watcher = it.next();
-      if (watcher.canceled) {
-        it.remove();
-        continue;
-      }
-
-      // We must set read-only mode here because the condition is evaluated from the runner
-      // thread, not a workflow thread.
-      setReadOnly(true);
-      boolean satisfied;
-      try {
-        satisfied = watcher.condition.get();
-      } finally {
-        setReadOnly(false);
-      }
-
-      if (satisfied) {
-        it.remove();
-        anyMatched = true;
-      }
-    }
-    return anyMatched;
-  }
-
-  /**
-   * Holds a condition for async await functionality. The condition is evaluated at the end of each
-   * event loop iteration and must handle promise completion in its body before returning true.
-   */
-  private static class ConditionWatcher {
-    final Supplier<Boolean> condition;
-    boolean canceled;
-
-    ConditionWatcher(Supplier<Boolean> condition) {
-      this.condition = condition;
-    }
-  }
 }

temporal-sdk/src/test/java/io/temporal/internal/sync/DeterministicRunnerTest.java

@@ -963,80 +963,4 @@ public void testSupplierCalledMultipleWithoutCaching() {
             });
     d.runUntilAllBlocked(DeterministicRunner.DEFAULT_DEADLOCK_DETECTION_TIMEOUT_MS);
   }
-
-  /**
-   * Test that beforeThreadsWakeUp callback is invoked BEFORE threads run. The callback sets a value
-   * that the thread reads, proving the callback ran first.
-   */
-  @Test
-  public void testBeforeThreadsWakeUpCallbackInvokedBeforeThreads() {
-    AtomicBoolean valueSetByCallback = new AtomicBoolean(false);
-    AtomicBoolean threadSawValue = new AtomicBoolean(false);
-
-    DeterministicRunnerImpl d =
-        new DeterministicRunnerImpl(
-            threadPool::submit,
-            DummySyncWorkflowContext.newDummySyncWorkflowContext(),
-            () -> {
-              // Thread checks if callback set the value
-              threadSawValue.set(valueSetByCallback.get());
-              status = "done";
-            },
-            null,
-            () -> {
-              // Callback sets value before threads run
-              valueSetByCallback.set(true);
-              return false;
-            });
-
-    d.runUntilAllBlocked(DeterministicRunner.DEFAULT_DEADLOCK_DETECTION_TIMEOUT_MS);
-    assertTrue(d.isDone());
-    assertTrue("Callback should set value before thread runs", valueSetByCallback.get());
-    assertTrue("Thread should see value set by callback", threadSawValue.get());
-  }
-
-  /**
-   * Test that when beforeThreadsWakeUp returns true (progress made), the loop continues and threads
-   * run again. The callback can return true multiple times when notifying multiple conditions.
-   */
-  @Test
-  public void testBeforeThreadsWakeUpProgressContinuesLoop() {
-    AtomicBoolean shouldUnblock1 = new AtomicBoolean(false);
-    AtomicBoolean shouldUnblock2 = new AtomicBoolean(false);
-    AtomicInteger trueCount = new AtomicInteger(0);
-
-    DeterministicRunnerImpl d =
-        new DeterministicRunnerImpl(
-            threadPool::submit,
-            DummySyncWorkflowContext.newDummySyncWorkflowContext(),
-            () -> {
-              status = "waiting1";
-              WorkflowThread.await("wait1", shouldUnblock1::get);
-              status = "waiting2";
-              WorkflowThread.await("wait2", shouldUnblock2::get);
-              status = "done";
-            },
-            null,
-            () -> {
-              // Callback can return true multiple times - once for each condition it unblocks
-              if (status.equals("waiting1") && !shouldUnblock1.get()) {
-                shouldUnblock1.set(true);
-                trueCount.incrementAndGet();
-                return true;
-              }
-              if (status.equals("waiting2") && !shouldUnblock2.get()) {
-                shouldUnblock2.set(true);
-                trueCount.incrementAndGet();
-                return true;
-              }
-              return false;
-            });
-
-    // Single runUntilAllBlocked: callback returns true twice (once per condition),
-    // thread advances through both waits to completion
-    d.runUntilAllBlocked(DeterministicRunner.DEFAULT_DEADLOCK_DETECTION_TIMEOUT_MS);
-    assertEquals("done", status);
-    assertTrue(d.isDone());
-    assertEquals("Callback should return true twice (once per condition)", 2, trueCount.get());
-  }
 }