Fix money-batch settlement when requests exceed batch size

core/src/main/java/io/temporal/samples/moneybatch/AccountTransferWorkflow.java

@@ -9,7 +9,7 @@
 public interface AccountTransferWorkflow {
 
   @WorkflowMethod
-  void deposit(String toAccountId, int batchSize);
+  void deposit(String toAccountId, int batchSize, BatchState state);
 
   @SignalMethod
   void withdraw(String fromAccountId, String referenceId, int amountCents);

core/src/main/java/io/temporal/samples/moneybatch/AccountTransferWorkflowImpl.java

@@ -2,9 +2,17 @@
 
 import io.temporal.activity.ActivityOptions;
 import io.temporal.common.RetryOptions;
+import io.temporal.failure.ApplicationFailure;
+import io.temporal.workflow.Async;
+import io.temporal.workflow.Promise;
 import io.temporal.workflow.Workflow;
+import io.temporal.workflow.WorkflowInit;
 import java.time.Duration;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 
 public class AccountTransferWorkflowImpl implements AccountTransferWorkflow {
@@ -21,25 +29,60 @@ public class AccountTransferWorkflowImpl implements AccountTransferWorkflow {
 
   private final Account account = Workflow.newActivityStub(Account.class, options);
 
-  private Set<String> references = new HashSet<>();
+  private final Set<String> references;
+  private final Deque<BatchState.Withdrawal> pending;
   private int balance;
   private int count;
 
+  @WorkflowInit
+  public AccountTransferWorkflowImpl(String toAccount, int batchSize, BatchState state) {
+    // Signals can run before deposit(), including immediately after Continue-As-New.
+    BatchState initial = state == null ? new BatchState() : state;
+    references = new HashSet<>(initial.references);
+    pending = new ArrayDeque<>(initial.pending);
+  }
+
   @Override
-  public void deposit(String toAccount, int batchSize) {
-    Workflow.await(() -> count == batchSize);
+  public void deposit(String toAccount, int batchSize, BatchState state) {
+    if (batchSize <= 0) {
+      throw ApplicationFailure.newNonRetryableFailure(
+          "batchSize must be positive", "InvalidBatchSize");
+    }
+    Workflow.await(() -> pending.size() >= batchSize);
+    // Select the whole batch before yielding to any Activity. Later signals only enter pending.
+    List<BatchState.Withdrawal> batch = new ArrayList<>();
+    for (int i = 0; i < batchSize; i++) {
+      batch.add(pending.removeFirst());
+    }
+    List<Promise<Void>> withdrawals = new ArrayList<>();
+    for (BatchState.Withdrawal withdrawal : batch) {
+      withdrawals.add(
+          Async.procedure(
+              () -> {
+                account.withdraw(
+                    withdrawal.fromAccountId, withdrawal.referenceId, withdrawal.amountCents);
+                balance += withdrawal.amountCents;
+                count++;
+              }));
+    }
+    Promise<Void> completed = Promise.allOf(withdrawals);
+    // Only the selected N requests can increment count. Also wake on an Activity failure.
+    Workflow.await(() -> count == batchSize || completed.isCompleted());
+    completed.get();
     String referenceId = Workflow.randomUUID().toString();
     account.deposit(toAccount, referenceId, balance);
+
+    if (!pending.isEmpty()) {
+      Workflow.continueAsNew(toAccount, batchSize, new BatchState(pending, references));
+    }
   }
 
   @Override
   public void withdraw(String fromAccountId, String referenceId, int amountCents) {
     if (!references.add(referenceId)) {
       return; // duplicate
     }
-    account.withdraw(fromAccountId, referenceId, amountCents);
-    balance += amountCents;
-    count++;
+    pending.addLast(new BatchState.Withdrawal(fromAccountId, referenceId, amountCents));
   }
 
   @Override

core/src/main/java/io/temporal/samples/moneybatch/BatchState.java

@@ -0,0 +1,34 @@
+package io.temporal.samples.moneybatch;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/** Pending requests and deduplication state carried to the next batch execution. */
+public class BatchState {
+  public List<Withdrawal> pending = new ArrayList<>();
+  public Set<String> references = new HashSet<>();
+
+  public BatchState() {}
+
+  public BatchState(Collection<Withdrawal> pending, Set<String> references) {
+    this.pending = new ArrayList<>(pending);
+    this.references = new HashSet<>(references);
+  }
+
+  public static class Withdrawal {
+    public String fromAccountId;
+    public String referenceId;
+    public int amountCents;
+
+    public Withdrawal() {}
+
+    public Withdrawal(String fromAccountId, String referenceId, int amountCents) {
+      this.fromAccountId = fromAccountId;
+      this.referenceId = referenceId;
+      this.amountCents = amountCents;
+    }
+  }
+}

core/src/main/java/io/temporal/samples/moneybatch/README.md

@@ -8,6 +8,31 @@ The sample also demonstrates the *signal with start* way of starting Workflows.
 If the Workflow is already running, it just receives the Signal. If it is not running, then it is started first, and then the signal is delivered to it.
 You can think about *signal with start* as a lazy way to create Workflows when signaling them.
 
+Each Workflow Execution settles exactly one batch. The Signal handler queues requests and ignores
+duplicate reference IDs. Once a full batch is available, the Workflow selects exactly `batchSize`
+requests, executes their withdrawals in parallel, and deposits their combined amount. Its completed
+withdrawal count cannot exceed the batch size, so the `count == batchSize` condition remains valid.
+
+Requests received while withdrawals or the deposit are running stay in the queue without executing
+their withdrawals. After a successful deposit, the Workflow completes if the queue is empty.
+Otherwise, it uses Continue-As-New to pass the pending requests and reference IDs to the next
+execution. For example, with a batch size of five and six unique requests, the first execution
+settles five and the next waits for four more requests before withdrawing and settling its batch.
+Waiting for that full batch is expected; a client waiting for the Workflow result also follows
+Continue-As-New and waits until the execution chain completes.
+
+`TransferRequester` supplies an empty `BatchState` on the initial start. A continued execution uses
+the carried state, initialized before any Signals are handled. Reference IDs are deduplicated
+throughout that execution chain; a separate start after completion begins with fresh state.
+
+This implementation changes Activity scheduling and does not preserve replay compatibility with
+executions started by the original implementation. Use it for new executions; upgrading existing
+executions requires a separate versioning or migration strategy.
+
+Reference IDs remain in the carried state for the whole execution chain. Continue-As-New limits
+history per execution but does not bound this state or the pending queue. Bounding reference
+retention would require choosing a finite deduplication window or using external durable storage.
+
 **How to run the Money Batch Sample**
 
 Money Batch sample has three separate processes. One to host Workflow Executions,

core/src/main/java/io/temporal/samples/moneybatch/TransferRequester.java

@@ -42,7 +42,7 @@ public static void main(String[] args) {
         workflowClient.newWorkflowStub(AccountTransferWorkflow.class, options);
     // Signal with start sends a signal to a workflow starting it if not yet running
     BatchRequest request = workflowClient.newSignalWithStartRequest();
-    request.add(transferWorkflow::deposit, to, BATCH_SIZE);
+    request.add(transferWorkflow::deposit, to, BATCH_SIZE, new BatchState());
     request.add(transferWorkflow::withdraw, from, reference, amountCents);
     workflowClient.signalWithStart(request);