diff --git a/core/src/main/java/io/temporal/samples/moneybatch/AccountTransferWorkflow.java b/core/src/main/java/io/temporal/samples/moneybatch/AccountTransferWorkflow.java index c3ebc35a..2d256596 100644 --- a/core/src/main/java/io/temporal/samples/moneybatch/AccountTransferWorkflow.java +++ b/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); diff --git a/core/src/main/java/io/temporal/samples/moneybatch/AccountTransferWorkflowImpl.java b/core/src/main/java/io/temporal/samples/moneybatch/AccountTransferWorkflowImpl.java index fe1ab124..d54dc41c 100644 --- a/core/src/main/java/io/temporal/samples/moneybatch/AccountTransferWorkflowImpl.java +++ b/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,15 +29,52 @@ public class AccountTransferWorkflowImpl implements AccountTransferWorkflow { private final Account account = Workflow.newActivityStub(Account.class, options); - private Set references = new HashSet<>(); + private final Set references; + private final Deque 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 batch = new ArrayList<>(); + for (int i = 0; i < batchSize; i++) { + batch.add(pending.removeFirst()); + } + List> 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 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 @@ -37,9 +82,7 @@ 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 diff --git a/core/src/main/java/io/temporal/samples/moneybatch/BatchState.java b/core/src/main/java/io/temporal/samples/moneybatch/BatchState.java new file mode 100644 index 00000000..18dace25 --- /dev/null +++ b/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 pending = new ArrayList<>(); + public Set references = new HashSet<>(); + + public BatchState() {} + + public BatchState(Collection pending, Set 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; + } + } +} diff --git a/core/src/main/java/io/temporal/samples/moneybatch/README.md b/core/src/main/java/io/temporal/samples/moneybatch/README.md index 701ef66c..9477bfa3 100644 --- a/core/src/main/java/io/temporal/samples/moneybatch/README.md +++ b/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, diff --git a/core/src/main/java/io/temporal/samples/moneybatch/TransferRequester.java b/core/src/main/java/io/temporal/samples/moneybatch/TransferRequester.java index 31b2f507..a536113d 100644 --- a/core/src/main/java/io/temporal/samples/moneybatch/TransferRequester.java +++ b/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); diff --git a/core/src/test/java/io/temporal/samples/moneybatch/TransferWorkflowTest.java b/core/src/test/java/io/temporal/samples/moneybatch/TransferWorkflowTest.java index 282a2136..b62f14e3 100644 --- a/core/src/test/java/io/temporal/samples/moneybatch/TransferWorkflowTest.java +++ b/core/src/test/java/io/temporal/samples/moneybatch/TransferWorkflowTest.java @@ -1,47 +1,59 @@ package io.temporal.samples.moneybatch; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.enums.v1.EventType; +import io.temporal.api.history.v1.HistoryEvent; import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowFailedException; import io.temporal.client.WorkflowOptions; import io.temporal.client.WorkflowStub; +import io.temporal.client.WorkflowTargetOptions; +import io.temporal.failure.ApplicationFailure; import io.temporal.testing.TestWorkflowRule; +import java.util.HashSet; +import java.util.List; import java.util.Random; +import java.util.Set; import java.util.UUID; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.junit.Rule; import org.junit.Test; public class TransferWorkflowTest { + private static final String WORKFLOW_ID = "moneybatch-749-reproduction"; @Rule - public TestWorkflowRule testWorkflowRule = + public TestWorkflowRule rule = TestWorkflowRule.newBuilder() .setWorkflowTypes(AccountTransferWorkflowImpl.class) .setDoNotStart(true) .build(); - @Test - public void testTransfer() { + @Test(timeout = 30000) + public void testTransfer() throws Exception { Account activities = mock(Account.class); - testWorkflowRule.getWorker().registerActivitiesImplementations(activities); - testWorkflowRule.getTestEnvironment().start(); + rule.getWorker().registerActivitiesImplementations(activities); + rule.getTestEnvironment().start(); String from = "account1"; String to = "account2"; int batchSize = 5; WorkflowOptions options = - WorkflowOptions.newBuilder() - .setTaskQueue(testWorkflowRule.getTaskQueue()) - .setWorkflowId(to) - .build(); + WorkflowOptions.newBuilder().setTaskQueue(rule.getTaskQueue()).setWorkflowId(to).build(); AccountTransferWorkflow transferWorkflow = - testWorkflowRule - .getWorkflowClient() - .newWorkflowStub(AccountTransferWorkflow.class, options); - WorkflowClient.start(transferWorkflow::deposit, to, batchSize); + rule.getWorkflowClient().newWorkflowStub(AccountTransferWorkflow.class, options); + WorkflowClient.start(transferWorkflow::deposit, to, batchSize, new BatchState()); Random random = new Random(); int total = 0; for (int i = 0; i < batchSize; i++) { @@ -50,9 +62,172 @@ public void testTransfer() { total += amountCents; } // Wait for workflow to finish - WorkflowStub.fromTyped(transferWorkflow).getResult(Void.class); + WorkflowStub.fromTyped(transferWorkflow).getResult(5, TimeUnit.SECONDS, Void.class); verify(activities).deposit(eq("account2"), any(), eq(total)); - testWorkflowRule.getTestEnvironment().shutdown(); + rule.getTestEnvironment().shutdown(); + } + + @Test(timeout = 30000) + public void carriesSixthRequestWithoutWithdrawingItUntilNextBatchIsFull() throws Exception { + RecordingAccount account = new RecordingAccount(false); + AccountTransferWorkflow workflow = newWorkflow(account); + WorkflowExecution first = start(workflow); + // Buffer all six signals before polling starts, so one task sees more than a batch. + send(workflow, 0, 6); + rule.getTestEnvironment().start(); + String nextRun = awaitContinuedRun(first.getRunId()); + + assertEquals(List.of(1500), account.deposits); + assertEquals(5, account.withdrawals.size()); + assertEquals( + Set.of("request-0", "request-1", "request-2", "request-3", "request-4"), + new HashSet<>(account.withdrawals)); + assertFalse(account.withdrawals.contains("request-5")); + assertEquals(0, workflow.getCount()); + assertEquals(0, workflow.getBalance()); + + // Both already-settled and carried references must remain deduplicated across runs. + workflow.withdraw("source", "request-0", 99999); + workflow.withdraw("source", "request-5", 99999); + send(workflow, 6, 10); + finish(workflow); + + assertEquals(List.of(1500, 4000), account.deposits); + assertEquals(10, account.withdrawals.size()); + assertEquals(10, new HashSet<>(account.withdrawals).size()); + assertRunCompleted(nextRun); + } + + @Test(timeout = 30000) + public void preservesSignalsReceivedWhileDepositIsBlocked() throws Exception { + RecordingAccount account = new RecordingAccount(true); + AccountTransferWorkflow workflow = newWorkflow(account); + WorkflowExecution first = start(workflow); + send(workflow, 0, 5); + rule.getTestEnvironment().start(); + try { + assertTrue(account.firstDeposit.await(5, TimeUnit.SECONDS)); + send(workflow, 5, 7); + workflow.withdraw("source", "request-0", 99999); + assertEquals(5, workflow.getCount()); + assertEquals(1500, workflow.getBalance()); + assertEquals(5, account.withdrawals.size()); + } finally { + account.releaseDeposit.countDown(); + } + String nextRun = awaitContinuedRun(first.getRunId()); + assertEquals(0, workflow.getCount()); + assertEquals(5, account.withdrawals.size()); + send(workflow, 7, 10); + finish(workflow); + + assertEquals(List.of(1500, 4000), account.deposits); + assertEquals(10, new HashSet<>(account.withdrawals).size()); + assertRunCompleted(nextRun); + } + + @Test(timeout = 30000) + public void propagatesWithdrawalFailureInsteadOfWaitingForeverForCount() throws Exception { + RecordingAccount account = + new RecordingAccount(false) { + @Override + public synchronized void withdraw(String accountId, String referenceId, int amountCents) { + throw ApplicationFailure.newNonRetryableFailure("withdrawal rejected", "Rejected"); + } + }; + AccountTransferWorkflow workflow = newWorkflow(account); + start(workflow); + send(workflow, 0, 5); + rule.getTestEnvironment().start(); + assertThrows(WorkflowFailedException.class, () -> finish(workflow)); + assertTrue(account.deposits.isEmpty()); + } + + private AccountTransferWorkflow newWorkflow(Account account) { + rule.getWorker().registerActivitiesImplementations(account); + return rule.getWorkflowClient() + .newWorkflowStub( + AccountTransferWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowId(WORKFLOW_ID) + .setTaskQueue(rule.getTaskQueue()) + .build()); + } + + private WorkflowExecution start(AccountTransferWorkflow workflow) { + return WorkflowClient.start(workflow::deposit, "destination", 5, new BatchState()); + } + + private void send(AccountTransferWorkflow workflow, int from, int to) { + for (int i = from; i < to; i++) { + workflow.withdraw("source", "request-" + i, (i + 1) * 100); + } + } + + private void finish(AccountTransferWorkflow workflow) throws Exception { + WorkflowStub.fromTyped(workflow).getResult(5, TimeUnit.SECONDS, Void.class); + } + + private String awaitContinuedRun(String runId) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + do { + for (HistoryEvent event : + rule.getWorkflowClient().fetchHistory(WORKFLOW_ID, runId).getEvents()) { + if (event.getEventType() == EventType.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW) { + return event.getWorkflowExecutionContinuedAsNewEventAttributes().getNewExecutionRunId(); + } + } + Thread.sleep(10); + } while (System.nanoTime() < deadline); + throw new AssertionError("Expected Continue-As-New for run " + runId); + } + + private void assertRunCompleted(String runId) { + List events = + rule.getWorkflowClient().fetchHistory(WORKFLOW_ID, runId).getEvents(); + assertEquals( + EventType.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED, + events.get(events.size() - 1).getEventType()); + assertEquals( + 5, + rule.getWorkflowClient() + .newWorkflowStub( + AccountTransferWorkflow.class, + WorkflowTargetOptions.newBuilder() + .setWorkflowId(WORKFLOW_ID) + .setRunId(runId) + .build()) + .getCount()); + } + + private static class RecordingAccount implements Account { + final List withdrawals = new CopyOnWriteArrayList<>(); + final List deposits = new CopyOnWriteArrayList<>(); + final CountDownLatch firstDeposit = new CountDownLatch(1); + final CountDownLatch releaseDeposit; + + RecordingAccount(boolean blockDeposit) { + releaseDeposit = new CountDownLatch(blockDeposit ? 1 : 0); + } + + @Override + public synchronized void withdraw(String accountId, String referenceId, int amountCents) { + withdrawals.add(referenceId); + } + + @Override + public void deposit(String accountId, String referenceId, int amountCents) { + deposits.add(amountCents); + firstDeposit.countDown(); + try { + if (!releaseDeposit.await(4, TimeUnit.SECONDS)) { + throw new IllegalStateException("Deposit was not released"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + } } }