feat: paginate step functions executions

This commit is contained in:
Zakaria
2026-09-27 23:35:21 -04:00
parent 0aef7fb2d8
commit cede01969e
5 changed files with 564 additions and 15 deletions
@@ -431,10 +431,13 @@ public class StepFunctionsJsonHandler {
}
private Response handleListExecutions(JsonNode request) {
List<Execution> list = service.listExecutions(request.path("stateMachineArn").asText());
Integer maxResults = optionalPageSize(request);
String statusFilter = optionalText(request, "statusFilter");
PaginatedResult<Execution> page = service.listExecutions(request.path("stateMachineArn").asText(),
statusFilter, maxResults, optionalNextToken(request));
ObjectNode response = objectMapper.createObjectNode();
ArrayNode array = response.putArray("executions");
for (Execution e : list) {
for (Execution e : page.items()) {
ObjectNode item = array.addObject();
item.put("executionArn", e.getExecutionArn());
item.put("stateMachineArn", e.getStateMachineArn());
@@ -449,6 +452,9 @@ public class StepFunctionsJsonHandler {
item.put("startDate", e.getStartDate());
if (e.getStopDate() != null) item.put("stopDate", e.getStopDate());
}
if (page.nextToken() != null) {
response.put("nextToken", page.nextToken());
}
return Response.ok(response).build();
}
@@ -464,18 +470,14 @@ public class StepFunctionsJsonHandler {
}
private Response handleGetExecutionHistory(JsonNode request) {
var arn = request.path("executionArn").asText();
var includeExecutionData = request.path("includeExecutionData").asBoolean(true);
var live = service.getExecutionHistory(arn);
List<HistoryEvent> events;
// Branch and iteration threads append under the history's own monitor while a client reads.
synchronized (live) {
events = new ArrayList<>(live);
}
String arn = request.path("executionArn").asText();
boolean includeExecutionData = request.path("includeExecutionData").asBoolean(true);
boolean reverseOrder = request.path("reverseOrder").asBoolean(false);
PaginatedResult<HistoryEvent> page = service.getExecutionHistory(arn, includeExecutionData, reverseOrder,
optionalPageSize(request), optionalNextToken(request));
ObjectNode response = objectMapper.createObjectNode();
ArrayNode array = response.putArray("events");
for (HistoryEvent e : events) {
for (HistoryEvent e : page.items()) {
ObjectNode item = array.addObject();
item.put("id", e.getId());
item.put("timestamp", e.getTimestamp());
@@ -491,9 +493,34 @@ public class StepFunctionsJsonHandler {
item.set(historyEventDetailsField(e.getType()), objectMapper.valueToTree(details));
}
}
if (page.nextToken() != null) {
response.put("nextToken", page.nextToken());
}
return Response.ok(response).build();
}
private static Integer optionalPageSize(JsonNode request) {
JsonNode maxResults = request.get("maxResults");
if (maxResults == null || maxResults.isNull()) {
return null;
}
if (!maxResults.isIntegralNumber() || !maxResults.canConvertToInt()) {
throw new AwsException("ValidationException", "maxResults must be an integer.", 400);
}
return maxResults.intValue();
}
private static String optionalNextToken(JsonNode request) {
JsonNode nextToken = request.get("nextToken");
if (nextToken == null || nextToken.isNull()) {
return null;
}
if (!nextToken.isTextual() || nextToken.textValue().isEmpty()) {
throw new AwsException("InvalidToken", "The provided nextToken is not valid.", 400);
}
return nextToken.textValue();
}
private static final Set<String> EXECUTION_DATA_FIELDS =
Set.of("input", "inputDetails", "output", "outputDetails");
@@ -31,6 +31,7 @@ import jakarta.inject.Inject;
import org.jboss.logging.Logger;
import java.math.BigDecimal;
import java.time.Clock;
import java.time.Instant;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
@@ -50,6 +51,9 @@ public class StepFunctionsService implements Resettable, ResourceProvider {
private static final Logger LOG = Logger.getLogger(StepFunctionsService.class);
private static final int HISTORY_PERSIST_CHECKPOINT = 100;
private static final int PAGINATION_DEFAULT_PAGE_SIZE = 100;
private static final int PAGINATION_MAX_PAGE_SIZE = 1000;
private static final long PAGINATION_TOKEN_TTL_MILLIS = TimeUnit.HOURS.toMillis(24);
private final StorageBackend<String, StateMachine> stateMachineStore;
private final StorageBackend<String, StateMachineAlias> stateMachineAliasStore;
@@ -63,6 +67,8 @@ public class StepFunctionsService implements Resettable, ResourceProvider {
// When each pending token last showed progress, so a Task's HeartbeatSeconds can bound the gap
// between heartbeats instead of the whole wait.
private final Map<String, Long> taskHeartbeatNanos = new ConcurrentHashMap<>();
private final Map<String, HistoryPaginationToken> historyPaginationTokens = new ConcurrentHashMap<>();
private final Map<String, ListExecutionsPaginationToken> listExecutionsPaginationTokens = new ConcurrentHashMap<>();
// Serializes the StartExecution check-and-create so two concurrent calls with the same name cannot
// both create and launch an execution; also enables AWS idempotent-success for STANDARD workflows.
private final Object executionStartLock = new Object();
@@ -70,6 +76,7 @@ public class StepFunctionsService implements Resettable, ResourceProvider {
private final AslExecutor aslExecutor;
private final ObjectMapper objectMapper;
private final SfnMockLoader mockLoader;
private final Clock clock;
// Fields that are valid only in JSONPath mode. Validated against real AWS:
// creating a JSONata state machine with any of these fields returns SCHEMA_VALIDATION_FAILED.
@@ -125,8 +132,8 @@ public class StepFunctionsService implements Resettable, ResourceProvider {
@Inject
public StepFunctionsService(StorageFactory storageFactory, RegionResolver regionResolver,
AslExecutor aslExecutor, ObjectMapper objectMapper,
SfnMockLoader mockLoader) {
AslExecutor aslExecutor, ObjectMapper objectMapper,
SfnMockLoader mockLoader, Clock clock) {
this.stateMachineStore = storageFactory.create("stepfunctions", "sfn-state-machines.json",
new TypeReference<Map<String, StateMachine>>() {});
this.executionStore = storageFactory.create("stepfunctions", "sfn-executions.json",
@@ -144,6 +151,13 @@ public class StepFunctionsService implements Resettable, ResourceProvider {
this.aslExecutor = aslExecutor;
this.objectMapper = objectMapper;
this.mockLoader = mockLoader;
this.clock = clock;
}
StepFunctionsService(StorageFactory storageFactory, RegionResolver regionResolver,
AslExecutor aslExecutor, ObjectMapper objectMapper,
SfnMockLoader mockLoader) {
this(storageFactory, regionResolver, aslExecutor, objectMapper, mockLoader, Clock.systemUTC());
}
public void clear() {
@@ -152,6 +166,8 @@ public class StepFunctionsService implements Resettable, ResourceProvider {
pendingTaskTokens.values().forEach(f -> f.completeExceptionally(new RuntimeException("StepFunctionsService cleared")));
pendingTaskTokens.clear();
taskHeartbeatNanos.clear();
historyPaginationTokens.clear();
listExecutionsPaginationTokens.clear();
}
@Override
@@ -942,8 +958,15 @@ public class StepFunctionsService implements Resettable, ResourceProvider {
}
public List<Execution> listExecutions(String stateMachineArn) {
return listExecutions(stateMachineArn, null, null, null).items();
}
public PaginatedResult<Execution> listExecutions(
String stateMachineArn, String statusFilter, Integer maxResults, String nextToken) {
validatePageSize(maxResults);
validateExecutionStatusFilter(statusFilter);
QualifiedStateMachineArn qualifiedArn = parseQualifiedStateMachineArn(stateMachineArn);
return executionStore.scan(k -> executionStore.get(k).map(execution -> {
List<Execution> executions = executionStore.scan(k -> executionStore.get(k).map(execution -> {
if (qualifiedArn == null) {
return Objects.equals(execution.getStateMachineArn(), stateMachineArn);
}
@@ -951,6 +974,13 @@ public class StepFunctionsService implements Resettable, ResourceProvider {
? Objects.equals(execution.getStateMachineVersionArn(), stateMachineArn)
: Objects.equals(execution.getStateMachineAliasArn(), stateMachineArn);
}).orElse(false));
if (statusFilter != null) {
executions.removeIf(execution -> !statusFilter.equals(execution.getStatus()));
}
executions.sort(Comparator.comparingDouble(StepFunctionsService::executionSortDate)
.reversed()
.thenComparing(Execution::getExecutionArn, Comparator.reverseOrder()));
return paginateExecutions(executions, stateMachineArn, statusFilter, maxResults, nextToken);
}
/**
@@ -1045,6 +1075,143 @@ public class StepFunctionsService implements Resettable, ResourceProvider {
key -> new ExecutionHistory(exec.getHistory(), () -> { }, isTerminal(exec.getStatus())));
}
public PaginatedResult<HistoryEvent> getExecutionHistory(
String arn, boolean includeExecutionData, boolean reverseOrder, Integer maxResults, String nextToken) {
validatePageSize(maxResults);
List<HistoryEvent> live = getExecutionHistory(arn);
List<HistoryEvent> events;
synchronized (live) {
events = new ArrayList<>(live);
}
events.sort(Comparator.comparingLong(HistoryEvent::getId));
if (reverseOrder) {
Collections.reverse(events);
}
return paginateHistory(events, arn, reverseOrder, includeExecutionData, maxResults, nextToken);
}
private static double executionSortDate(Execution execution) {
return execution.getStartDate();
}
private static void validateExecutionStatusFilter(String statusFilter) {
if (statusFilter != null && !Set.of("RUNNING", "SUCCEEDED", "FAILED", "TIMED_OUT", "ABORTED",
"PENDING_REDRIVE").contains(statusFilter)) {
throw new AwsException("ValidationException", "Invalid statusFilter: " + statusFilter, 400);
}
if ("PENDING_REDRIVE".equals(statusFilter)) {
throw new AwsException("ValidationException", "PENDING_REDRIVE requires a mapRunArn.", 400);
}
}
private static void validatePageSize(Integer maxResults) {
if (maxResults != null && (maxResults < 0 || maxResults > PAGINATION_MAX_PAGE_SIZE)) {
throw new AwsException("ValidationException", "maxResults must be between 0 and "
+ PAGINATION_MAX_PAGE_SIZE, 400);
}
}
private PaginatedResult<Execution> paginateExecutions(List<Execution> all, String stateMachineArn,
String statusFilter, Integer maxResults,
String nextToken) {
ListExecutionsPaginationToken cursor = listExecutionsCursor(stateMachineArn, statusFilter, maxResults,
nextToken);
List<Execution> remaining = cursor == null ? all : all.stream()
.filter(execution -> isAfterExecutionCursor(execution, cursor))
.toList();
int limit = maxResults == null || maxResults == 0 ? PAGINATION_DEFAULT_PAGE_SIZE : maxResults;
int end = Math.min(limit, remaining.size());
String token = end < remaining.size()
? issueListExecutionsToken(stateMachineArn, statusFilter, maxResults, remaining.get(end - 1))
: null;
return new PaginatedResult<>(new ArrayList<>(remaining.subList(0, end)), token);
}
private ListExecutionsPaginationToken listExecutionsCursor(String stateMachineArn, String statusFilter,
Integer maxResults, String nextToken) {
if (nextToken == null || nextToken.isEmpty()) {
return null;
}
ListExecutionsPaginationToken token = listExecutionsPaginationTokens.get(nextToken);
if (token == null || token.expiresAtMillis() < clock.millis()
|| !token.matches(stateMachineArn, statusFilter, maxResults)) {
listExecutionsPaginationTokens.remove(nextToken);
throw new AwsException("InvalidToken", "The provided nextToken is not valid.", 400);
}
return token;
}
private static boolean isAfterExecutionCursor(Execution execution, ListExecutionsPaginationToken cursor) {
int dateComparison = Double.compare(executionSortDate(execution), cursor.sortDate());
return dateComparison < 0 || dateComparison == 0
&& execution.getExecutionArn().compareTo(cursor.executionArn()) < 0;
}
private String issueListExecutionsToken(String stateMachineArn, String statusFilter, Integer maxResults,
Execution lastExecution) {
String token = UUID.randomUUID().toString();
listExecutionsPaginationTokens.put(token, new ListExecutionsPaginationToken(stateMachineArn, statusFilter,
maxResults, executionSortDate(lastExecution), lastExecution.getExecutionArn(),
clock.millis() + PAGINATION_TOKEN_TTL_MILLIS));
return token;
}
private PaginatedResult<HistoryEvent> paginateHistory(List<HistoryEvent> all, String executionArn,
boolean reverseOrder, boolean includeExecutionData,
Integer maxResults, String nextToken) {
int limit = maxResults == null || maxResults == 0 ? PAGINATION_DEFAULT_PAGE_SIZE : maxResults;
HistoryPaginationToken cursor = historyCursor(executionArn, reverseOrder, includeExecutionData, maxResults,
nextToken);
List<HistoryEvent> remaining = cursor == null ? all : all.stream()
.filter(event -> reverseOrder ? event.getId() < cursor.eventId() : event.getId() > cursor.eventId())
.toList();
int end = Math.min(limit, remaining.size());
String token = end < remaining.size()
? issueHistoryToken(executionArn, reverseOrder, includeExecutionData, maxResults,
remaining.get(end - 1).getId())
: null;
return new PaginatedResult<>(new ArrayList<>(remaining.subList(0, end)), token);
}
private HistoryPaginationToken historyCursor(String executionArn, boolean reverseOrder,
boolean includeExecutionData, Integer maxResults, String nextToken) {
if (nextToken == null || nextToken.isEmpty()) {
return null;
}
HistoryPaginationToken token = historyPaginationTokens.get(nextToken);
if (token == null || token.expiresAtMillis() < clock.millis()
|| !token.matches(executionArn, reverseOrder, includeExecutionData, maxResults)) {
historyPaginationTokens.remove(nextToken);
throw new AwsException("InvalidToken", "The provided nextToken is not valid.", 400);
}
return token;
}
private String issueHistoryToken(String executionArn, boolean reverseOrder, boolean includeExecutionData,
Integer maxResults, long eventId) {
String token = UUID.randomUUID().toString();
historyPaginationTokens.put(token, new HistoryPaginationToken(executionArn, reverseOrder, includeExecutionData,
maxResults, eventId, clock.millis() + PAGINATION_TOKEN_TTL_MILLIS));
return token;
}
private record HistoryPaginationToken(String executionArn, boolean reverseOrder, boolean includeExecutionData,
Integer maxResults, long eventId, long expiresAtMillis) {
private boolean matches(String executionArn, boolean reverseOrder,
boolean includeExecutionData, Integer maxResults) {
return this.executionArn.equals(executionArn) && this.reverseOrder == reverseOrder
&& this.includeExecutionData == includeExecutionData && Objects.equals(this.maxResults, maxResults);
}
}
private record ListExecutionsPaginationToken(String stateMachineArn, String statusFilter, Integer maxResults,
double sortDate, String executionArn, long expiresAtMillis) {
private boolean matches(String stateMachineArn, String statusFilter, Integer maxResults) {
return this.stateMachineArn.equals(stateMachineArn) && Objects.equals(this.statusFilter, statusFilter)
&& Objects.equals(this.maxResults, maxResults);
}
}
private static boolean isTerminal(String status) {
return !"RUNNING".equals(status);
}
@@ -9,6 +9,7 @@ import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import static io.restassured.RestAssured.given;
@@ -331,6 +332,114 @@ class StepFunctionsExecutionHistoryIntegrationTest {
assertEquals(4L, events.get(4).path("previousEventId").asLong());
}
@Test
void listExecutions_pagesNewestFirstWithStatusFilterAndRejectsInvalidTokens() throws Exception {
String definition = """
{"StartAt":"Done","States":{"Done":{"Type":"Pass","End":true}}}
""";
String stateMachineArn = createStateMachine("execution-list-pagination-test", definition);
List<String> executionArns = new ArrayList<>();
for (int i = 0; i < 26; i++) {
executionArns.add(startExecution(stateMachineArn));
}
for (String executionArn : executionArns) {
waitForExecution(executionArn);
}
Response first = listExecutions(stateMachineArn, "SUCCEEDED", 10, null);
String secondToken = first.jsonPath().getString("nextToken");
Response second = listExecutions(stateMachineArn, "SUCCEEDED", 10, secondToken);
String thirdToken = second.jsonPath().getString("nextToken");
Response third = listExecutions(stateMachineArn, "SUCCEEDED", 10, thirdToken);
List<String> returnedArns = new ArrayList<>();
for (Response page : List.of(first, second, third)) {
returnedArns.addAll(page.jsonPath().getList("executions.executionArn"));
List<Number> dates = page.jsonPath().getList("executions.startDate");
for (int i = 1; i < dates.size(); i++) {
assertTrue(dates.get(i - 1).doubleValue() >= dates.get(i).doubleValue(), page.body().asString());
}
}
assertEquals(26, returnedArns.size());
assertEquals(26, new HashSet<>(returnedArns).size());
assertEquals(null, third.jsonPath().getString("nextToken"));
assertEquals(26, listExecutions(stateMachineArn, "SUCCEEDED", 0, null)
.jsonPath().getList("executions.executionArn").size());
given()
.header("X-Amz-Target", "AWSStepFunctions.ListExecutions")
.contentType(SFN_CONTENT_TYPE)
.body("""
{"stateMachineArn":"%s","statusFilter":"RUNNING","maxResults":10,"nextToken":"%s"}
""".formatted(stateMachineArn, secondToken))
.when()
.post("/")
.then()
.statusCode(400)
.body("__type", equalTo("InvalidToken"));
given()
.header("X-Amz-Target", "AWSStepFunctions.ListExecutions")
.contentType(SFN_CONTENT_TYPE)
.body("""
{"stateMachineArn":"%s","maxResults":1001}
""".formatted(stateMachineArn))
.when()
.post("/")
.then()
.statusCode(400)
.body("__type", equalTo("ValidationException"));
}
@Test
void getExecutionHistory_pagesAscendingAndReverseOrder() throws Exception {
StringBuilder states = new StringBuilder();
for (int i = 1; i <= 10; i++) {
states.append("\"P").append(i).append("\":{\"Type\":\"Pass\",");
if (i == 10) {
states.append("\"End\":true}");
} else {
states.append("\"Next\":\"P").append(i + 1).append("\"}");
}
if (i < 10) {
states.append(',');
}
}
String definition = "{\"StartAt\":\"P1\",\"States\":{" + states + "}}";
String stateMachineArn = createStateMachine("execution-history-pagination-test", definition);
String executionArn = startExecution(stateMachineArn);
waitForExecution(executionArn);
Response first = getExecutionHistoryPage(executionArn, false, 7, null);
Response second = getExecutionHistoryPage(executionArn, false, 7, first.jsonPath().getString("nextToken"));
Response third = getExecutionHistoryPage(executionArn, false, 7, second.jsonPath().getString("nextToken"));
Response fourth = getExecutionHistoryPage(executionArn, false, 7, third.jsonPath().getString("nextToken"));
List<Number> eventIds = new ArrayList<>();
for (Response page : List.of(first, second, third, fourth)) {
eventIds.addAll(page.jsonPath().getList("events.id"));
}
assertEquals(22, eventIds.size());
for (int i = 0; i < eventIds.size(); i++) {
assertEquals(i + 1L, eventIds.get(i).longValue());
}
Response reverse = getExecutionHistoryPage(executionArn, true, 7, null);
assertEquals(22L, reverse.jsonPath().getLong("events[0].id"));
assertEquals(16L, reverse.jsonPath().getLong("events[6].id"));
given()
.header("X-Amz-Target", "AWSStepFunctions.GetExecutionHistory")
.contentType(SFN_CONTENT_TYPE)
.body("""
{"executionArn":"%s","maxResults":7,"nextToken":"not-a-token"}
""".formatted(executionArn))
.when()
.post("/")
.then()
.statusCode(400)
.body("__type", equalTo("InvalidToken"));
}
private Response getExecutionHistory(String executionArn) {
Response response = given()
.header("X-Amz-Target", "AWSStepFunctions.GetExecutionHistory")
@@ -347,6 +456,35 @@ class StepFunctionsExecutionHistoryIntegrationTest {
return response;
}
private Response getExecutionHistoryPage(String executionArn, boolean reverseOrder, int maxResults,
String nextToken) {
String nextTokenField = nextToken == null ? "" : ",\"nextToken\":\"" + nextToken + "\"";
Response response = given()
.header("X-Amz-Target", "AWSStepFunctions.GetExecutionHistory")
.contentType(SFN_CONTENT_TYPE)
.body("""
{"executionArn":"%s","reverseOrder":%s,"maxResults":%d%s}
""".formatted(executionArn, reverseOrder, maxResults, nextTokenField))
.when()
.post("/");
response.then().statusCode(200);
return response;
}
private Response listExecutions(String stateMachineArn, String statusFilter, int maxResults, String nextToken) {
String nextTokenField = nextToken == null ? "" : ",\"nextToken\":\"" + nextToken + "\"";
Response response = given()
.header("X-Amz-Target", "AWSStepFunctions.ListExecutions")
.contentType(SFN_CONTENT_TYPE)
.body("""
{"stateMachineArn":"%s","statusFilter":"%s","maxResults":%d%s}
""".formatted(stateMachineArn, statusFilter, maxResults, nextTokenField))
.when()
.post("/");
response.then().statusCode(200);
return response;
}
private Response describeExecution(String executionArn) {
Response response = given()
.header("X-Amz-Target", "AWSStepFunctions.DescribeExecution")
@@ -0,0 +1,208 @@
package io.github.hectorvent.floci.services.stepfunctions;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.github.hectorvent.floci.core.common.AwsException;
import io.github.hectorvent.floci.core.common.PaginatedResult;
import io.github.hectorvent.floci.core.common.RegionResolver;
import io.github.hectorvent.floci.core.storage.AccountAwareStorageBackend;
import io.github.hectorvent.floci.core.storage.StorageFactory;
import io.github.hectorvent.floci.services.stepfunctions.model.Activity;
import io.github.hectorvent.floci.services.stepfunctions.model.Execution;
import io.github.hectorvent.floci.services.stepfunctions.model.HistoryEvent;
import io.github.hectorvent.floci.services.stepfunctions.model.MapRun;
import io.github.hectorvent.floci.services.stepfunctions.model.StateMachine;
import io.github.hectorvent.floci.services.stepfunctions.model.StateMachineAlias;
import org.junit.jupiter.api.Test;
import java.time.Clock;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
class StepFunctionsListExecutionsPaginationTest {
private static final String ACCOUNT = "000000000000";
private static final String STATE_MACHINE_ARN =
"arn:aws:states:us-east-1:000000000000:stateMachine:test";
@Test
void continuesAfterNewExecutionsWithoutDuplicatingOrSkippingExistingExecutions() {
Fixture fixture = fixture();
List<String> existingArns = new ArrayList<>();
for (int i = 1; i <= 5; i++) {
String arn = executionArn("existing-" + i);
existingArns.add(arn);
fixture.executions().put(arn, succeededExecution(arn, i));
}
PaginatedResult<Execution> first = fixture.service().listExecutions(
STATE_MACHINE_ARN, "SUCCEEDED", 2, null);
fixture.executions().put(executionArn("new-1"), succeededExecution(executionArn("new-1"), 7));
fixture.executions().put(executionArn("new-2"), succeededExecution(executionArn("new-2"), 6));
PaginatedResult<Execution> second = fixture.service().listExecutions(
STATE_MACHINE_ARN, "SUCCEEDED", 2, first.nextToken());
PaginatedResult<Execution> third = fixture.service().listExecutions(
STATE_MACHINE_ARN, "SUCCEEDED", 2, second.nextToken());
List<String> returned = new ArrayList<>();
for (PaginatedResult<Execution> page : List.of(first, second, third)) {
page.items().forEach(execution -> returned.add(execution.getExecutionArn()));
}
assertEquals(existingArns.size(), returned.stream().filter(existingArns::contains).count());
assertEquals(existingArns.size(), new HashSet<>(returned).stream().filter(existingArns::contains).count());
}
@Test
void rejectsTokenAfterTwentyFourHours() {
Fixture fixture = fixture();
fixture.executions().put(executionArn("one"), succeededExecution(executionArn("one"), 2));
fixture.executions().put(executionArn("two"), succeededExecution(executionArn("two"), 1));
String token = fixture.service().listExecutions(STATE_MACHINE_ARN, "SUCCEEDED", 1, null).nextToken();
fixture.clock().advanceSeconds(24 * 60 * 60 + 1);
AwsException exception = assertThrows(AwsException.class, () -> fixture.service().listExecutions(
STATE_MACHINE_ARN, "SUCCEEDED", 1, token));
assertEquals("InvalidToken", exception.getErrorCode());
}
@Test
void ordersSameMillisecondExecutionsByArnAcrossPages() {
Fixture fixture = fixture();
List<String> expected = new ArrayList<>();
for (int i = 0; i < 12; i++) {
String arn = executionArn("same-millisecond-%02d".formatted(i));
expected.add(arn);
Execution execution = succeededExecution(arn, 100 - i);
execution.setStartDate(42);
fixture.executions().put(arn, execution);
}
expected.sort(Comparator.reverseOrder());
PaginatedResult<Execution> first = fixture.service().listExecutions(STATE_MACHINE_ARN, "SUCCEEDED", 4, null);
PaginatedResult<Execution> second = fixture.service().listExecutions(
STATE_MACHINE_ARN, "SUCCEEDED", 4, first.nextToken());
PaginatedResult<Execution> third = fixture.service().listExecutions(
STATE_MACHINE_ARN, "SUCCEEDED", 4, second.nextToken());
List<String> returned = new ArrayList<>();
for (PaginatedResult<Execution> page : List.of(first, second, third)) {
page.items().forEach(execution -> returned.add(execution.getExecutionArn()));
}
assertEquals(expected, returned);
}
@Test
void reverseHistoryPagingDoesNotShiftWhenRunningExecutionAppendsEvents() {
Fixture fixture = fixture();
String executionArn = executionArn("running-history");
Execution execution = new Execution();
execution.setExecutionArn(executionArn);
execution.setStateMachineArn(STATE_MACHINE_ARN);
execution.setStatus("RUNNING");
execution.setHistory(history(10));
fixture.executions().put(executionArn, execution);
PaginatedResult<HistoryEvent> first = fixture.service().getExecutionHistory(
executionArn, true, true, 3, null);
List<HistoryEvent> liveHistory = fixture.service().getExecutionHistory(executionArn);
liveHistory.add(historyEvent(11));
liveHistory.add(historyEvent(12));
PaginatedResult<HistoryEvent> second = fixture.service().getExecutionHistory(
executionArn, true, true, 3, first.nextToken());
PaginatedResult<HistoryEvent> third = fixture.service().getExecutionHistory(
executionArn, true, true, 3, second.nextToken());
PaginatedResult<HistoryEvent> fourth = fixture.service().getExecutionHistory(
executionArn, true, true, 3, third.nextToken());
List<Long> returned = new ArrayList<>();
for (PaginatedResult<HistoryEvent> page : List.of(first, second, third, fourth)) {
page.items().forEach(event -> returned.add(event.getId()));
}
assertEquals(List.of(10L, 9L, 8L, 7L, 6L, 5L, 4L, 3L, 2L, 1L), returned);
}
private static Fixture fixture() {
AccountAwareStorageBackend<Execution> executions = AccountAwareStorageBackend.inMemory(ACCOUNT);
StorageFactory factory = mock(StorageFactory.class);
doReturn(AccountAwareStorageBackend.<StateMachine>inMemory(ACCOUNT))
.doReturn(executions)
.doReturn(AccountAwareStorageBackend.<Activity>inMemory(ACCOUNT))
.doReturn(AccountAwareStorageBackend.<MapRun>inMemory(ACCOUNT))
.doReturn(AccountAwareStorageBackend.<StateMachineAlias>inMemory(ACCOUNT))
.when(factory).create(anyString(), anyString(), any());
MutableClock clock = new MutableClock(Instant.parse("2026-01-01T00:00:00Z"));
StepFunctionsService service = new StepFunctionsService(factory, new RegionResolver("us-east-1", ACCOUNT),
null, new ObjectMapper(), mock(SfnMockLoader.class), clock);
return new Fixture(service, executions, clock);
}
private static String executionArn(String name) {
return "arn:aws:states:us-east-1:000000000000:execution:test:" + name;
}
private static Execution succeededExecution(String arn, double stopDate) {
Execution execution = new Execution();
execution.setExecutionArn(arn);
execution.setStateMachineArn(STATE_MACHINE_ARN);
execution.setStatus("SUCCEEDED");
execution.setStopDate(stopDate);
return execution;
}
private static List<HistoryEvent> history(int count) {
List<HistoryEvent> events = new ArrayList<>();
for (int i = 1; i <= count; i++) {
events.add(historyEvent(i));
}
return events;
}
private static HistoryEvent historyEvent(long id) {
HistoryEvent event = new HistoryEvent();
event.setId(id);
return event;
}
private record Fixture(StepFunctionsService service, AccountAwareStorageBackend<Execution> executions,
MutableClock clock) {
}
private static final class MutableClock extends Clock {
private Instant instant;
private MutableClock(Instant instant) {
this.instant = instant;
}
@Override
public ZoneOffset getZone() {
return ZoneOffset.UTC;
}
@Override
public Clock withZone(ZoneId zone) {
return this;
}
@Override
public Instant instant() {
return instant;
}
private void advanceSeconds(long seconds) {
instant = instant.plusSeconds(seconds);
}
}
}