Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,8 @@ to docs, or any other relevant information.
### Fixed
- Test server now honors retry expiration deadlines that fall exactly on a whole second. Previously such deadlines were
ignored and retries were scheduled past them instead of failing with `RETRY_STATE_TIMEOUT`.
- Test server no longer disables time skipping when a child workflow closes after its parent, for example an abandoned
child. Previously every later sleep or timer on that test server ran in real time.

## Previous releases

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1385,7 +1385,6 @@ public void childWorkflowFailed(
child.action(StateMachines.Action.FAIL, ctx, a, 0);
childWorkflows.remove(a.getInitiatedEventId());
scheduleWorkflowTask(ctx);
ctx.unlockTimer("childWorkflowFailed");
});
}

Expand All @@ -1398,7 +1397,6 @@ public void childWorkflowTimedOut(
child.action(Action.TIME_OUT, ctx, a.getRetryState(), 0);
childWorkflows.remove(a.getInitiatedEventId());
scheduleWorkflowTask(ctx);
ctx.unlockTimer("childWorkflowTimedOut");
});
}

Expand All @@ -1424,7 +1422,6 @@ public void childWorkflowCompleted(
child.action(StateMachines.Action.COMPLETE, ctx, a, 0);
childWorkflows.remove(a.getInitiatedEventId());
scheduleWorkflowTask(ctx);
ctx.unlockTimer("childWorkflowCompleted");
});
}

Expand All @@ -1437,7 +1434,6 @@ public void childWorkflowCanceled(
child.action(StateMachines.Action.CANCEL, ctx, a, 0);
childWorkflows.remove(a.getInitiatedEventId());
scheduleWorkflowTask(ctx);
ctx.unlockTimer("childWorkflowCanceled");
});
}

Expand Down Expand Up @@ -1594,7 +1590,8 @@ private void processFailWorkflowExecution(
workflowTaskStateMachine.getData().workflowCompleted = true;
processWorkflowCompletionCallbacks(ctx);
if (parent.isPresent()) {
ctx.lockTimer("processFailWorkflowExecution notify parent"); // unlocked by the parent
// Keeps time from skipping until the parent schedules a workflow task.
ctx.lockTimer("processFailWorkflowExecution notify parent");
ChildWorkflowExecutionFailedEventAttributes a =
ChildWorkflowExecutionFailedEventAttributes.newBuilder()
.setInitiatedEventId(parentChildInitiatedEventId.getAsLong())
Expand All @@ -1617,6 +1614,9 @@ private void processFailWorkflowExecution(
}
} catch (Throwable e) {
log.error("Failure reporting child failure", e);
} finally {
// Released even if the parent already closed, so time skipping is not stuck.
timerService.unlockTimeSkipping("processFailWorkflowExecution notify parent");
}
});
}
Expand All @@ -1641,7 +1641,8 @@ private void processCompleteWorkflowExecution(
// years
workflow.getData().runTimerCancellationHandle.apply();
if (parent.isPresent()) {
ctx.lockTimer("processCompleteWorkflowExecution notify parent"); // unlocked by the parent
// Keeps time from skipping until the parent schedules a workflow task.
ctx.lockTimer("processCompleteWorkflowExecution notify parent");
ChildWorkflowExecutionCompletedEventAttributes a =
ChildWorkflowExecutionCompletedEventAttributes.newBuilder()
.setInitiatedEventId(parentChildInitiatedEventId.getAsLong())
Expand All @@ -1665,6 +1666,9 @@ private void processCompleteWorkflowExecution(
}
} catch (Throwable e) {
log.error("Failure reporting child completion", e);
} finally {
// Released even if the parent already closed, so time skipping is not stuck.
timerService.unlockTimeSkipping("processCompleteWorkflowExecution notify parent");
}
});
}
Expand Down Expand Up @@ -1718,7 +1722,8 @@ private void processCancelWorkflowExecution(
workflowTaskStateMachine.getData().workflowCompleted = true;
processWorkflowCompletionCallbacks(ctx);
if (parent.isPresent()) {
ctx.lockTimer("processCancelWorkflowExecution notify parent"); // unlocked by the parent
// Keeps time from skipping until the parent schedules a workflow task.
ctx.lockTimer("processCancelWorkflowExecution notify parent");
ChildWorkflowExecutionCanceledEventAttributes a =
ChildWorkflowExecutionCanceledEventAttributes.newBuilder()
.setInitiatedEventId(parentChildInitiatedEventId.getAsLong())
Expand All @@ -1742,6 +1747,9 @@ private void processCancelWorkflowExecution(
}
} catch (Throwable e) {
log.error("Failure reporting child cancellation", e);
} finally {
// Released even if the parent already closed, so time skipping is not stuck.
timerService.unlockTimeSkipping("processCancelWorkflowExecution notify parent");
}
});
}
Expand Down Expand Up @@ -2646,7 +2654,8 @@ private void timeoutWorkflow() {
workflowTaskStateMachine.getData().workflowCompleted = true;
processWorkflowCompletionCallbacks(ctx);
if (parent.isPresent()) {
ctx.lockTimer("timeoutWorkflow notify parent"); // unlocked by the parent
// Keeps time from skipping until the parent schedules a workflow task.
ctx.lockTimer("timeoutWorkflow notify parent");
}
ForkJoinPool.commonPool().execute(() -> reportWorkflowTimeoutToParent(ctx));
});
Expand Down Expand Up @@ -2677,6 +2686,9 @@ private void reportWorkflowTimeoutToParent(RequestContext ctx) {
}
} catch (Exception e) {
log.error("Failure reporting child timing out", e);
} finally {
// Released even if the parent already closed, so time skipping is not stuck.
timerService.unlockTimeSkipping("timeoutWorkflow notify parent");
}
}

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,146 @@
package io.temporal.testserver.functional.timeskipping;

import static org.junit.Assume.assumeFalse;

import io.temporal.api.enums.v1.ParentClosePolicy;
import io.temporal.client.WorkflowFailedException;
import io.temporal.client.WorkflowOptions;
import io.temporal.client.WorkflowStub;
import io.temporal.failure.ApplicationFailure;
import io.temporal.testing.internal.SDKTestWorkflowRule;
import io.temporal.workflow.Async;
import io.temporal.workflow.ChildWorkflowOptions;
import io.temporal.workflow.Workflow;
import io.temporal.workflow.WorkflowInterface;
import io.temporal.workflow.WorkflowMethod;
import java.time.Duration;
import java.util.UUID;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;

/**
* An abandoned child that closes after its parent must not disable time skipping. The test server
* locks time skipping when a child closes until the parent processes the notification, so the lock
* has to be released when the parent is already closed. If it leaks, the final sleep runs in real
* time and the test times out.
*/
public class AbandonedChildTimeSkippingTest {
private static final Duration CHILD_SLEEP = Duration.ofHours(1);

@Rule
public SDKTestWorkflowRule testWorkflowRule =
SDKTestWorkflowRule.newBuilder()
.setWorkflowTypes(ParentWorkflowImpl.class, ChildWorkflowImpl.class)
.build();

@Before
public void checkRealServer() {
assumeFalse("Time skipping is a test server feature", SDKTestWorkflowRule.useExternalService);
}

@Test
public void timeSkippingWorksAfterAbandonedChildCompletes() {
String childId = startParentWithAbandonedChild(ChildOutcome.COMPLETE);
WorkflowStub child = testWorkflowRule.getWorkflowClient().newUntypedWorkflowStub(childId);
child.getResult(Void.class);
assertTimeSkippingWorks();
}

@Test
public void timeSkippingWorksAfterAbandonedChildFails() {
String childId = startParentWithAbandonedChild(ChildOutcome.FAIL);
WorkflowStub child = testWorkflowRule.getWorkflowClient().newUntypedWorkflowStub(childId);
awaitFailure(child);
assertTimeSkippingWorks();
}

@Test
public void timeSkippingWorksAfterAbandonedChildIsCanceled() {
String childId = startParentWithAbandonedChild(ChildOutcome.CANCEL);
WorkflowStub child = testWorkflowRule.getWorkflowClient().newUntypedWorkflowStub(childId);
child.cancel();
awaitFailure(child);
assertTimeSkippingWorks();
}

@Test
public void timeSkippingWorksAfterAbandonedChildTimesOut() {
String childId = startParentWithAbandonedChild(ChildOutcome.TIME_OUT);
WorkflowStub child = testWorkflowRule.getWorkflowClient().newUntypedWorkflowStub(childId);
awaitFailure(child);
assertTimeSkippingWorks();
}

private String startParentWithAbandonedChild(ChildOutcome outcome) {
String childId = UUID.randomUUID().toString();
ParentWorkflow parent =
testWorkflowRule
.getWorkflowClient()
.newWorkflowStub(
ParentWorkflow.class,
WorkflowOptions.newBuilder().setTaskQueue(testWorkflowRule.getTaskQueue()).build());
// The parent returns as soon as the child has started.
parent.execute(childId, outcome);
return childId;
}

private static void awaitFailure(WorkflowStub child) {
try {
child.getResult(Void.class);
} catch (WorkflowFailedException expected) {
return;
}
throw new AssertionError("Child workflow was expected to fail");
}

private void assertTimeSkippingWorks() {
// Blocks in real time, exceeding the test timeout, if time skipping is locked.
testWorkflowRule.getTestEnvironment().sleep(Duration.ofDays(1));
}

public enum ChildOutcome {
COMPLETE,
FAIL,
CANCEL,
TIME_OUT
}

@WorkflowInterface
public interface ParentWorkflow {
@WorkflowMethod
void execute(String childId, ChildOutcome outcome);
}

@WorkflowInterface
public interface ChildWorkflow {
@WorkflowMethod
void execute(ChildOutcome outcome);
}

public static class ParentWorkflowImpl implements ParentWorkflow {
@Override
public void execute(String childId, ChildOutcome outcome) {
ChildWorkflowOptions.Builder options =
ChildWorkflowOptions.newBuilder()
.setWorkflowId(childId)
.setParentClosePolicy(ParentClosePolicy.PARENT_CLOSE_POLICY_ABANDON);
if (outcome == ChildOutcome.TIME_OUT) {
options.setWorkflowRunTimeout(CHILD_SLEEP.dividedBy(2));
}
ChildWorkflow child = Workflow.newChildWorkflowStub(ChildWorkflow.class, options.build());
Async.procedure(child::execute, outcome);
Workflow.getWorkflowExecution(child).get();
}
}

public static class ChildWorkflowImpl implements ChildWorkflow {
@Override
public void execute(ChildOutcome outcome) {
Workflow.sleep(CHILD_SLEEP);
if (outcome == ChildOutcome.FAIL) {
throw ApplicationFailure.newFailure("Child failed", "ChildFailure");
}
}
}
}
Loading