diff --git a/docs/services/step-functions.md b/docs/services/step-functions.md index cf59569..2c67b7f 100644 --- a/docs/services/step-functions.md +++ b/docs/services/step-functions.md @@ -48,6 +48,15 @@ changes to `StateMachineName` or `StateMachineType` use replacement semantics. ## Execution history events +## Execution pagination + +`ListExecutions` and `GetExecutionHistory` support `maxResults` from 0 to 1000, with AWS's +default page size of 100. Both return opaque `nextToken` values bound to the request arguments +and valid for 24 hours. `ListExecutions` is newest first, ordered by start date and then execution +ARN. The same-millisecond tie-break is covered by focused unit tests; the live acceptance suite +does not force identical execution timestamps. `GetExecutionHistory` is ascending by event ID by +default and supports `reverseOrder`. + `GetExecutionHistory` emits an event family around a Task state's `TaskStateEntered` and `TaskStateExited` pair. Which family depends on the resource type. diff --git a/src/main/java/io/github/hectorvent/floci/services/stepfunctions/StepFunctionsJsonHandler.java b/src/main/java/io/github/hectorvent/floci/services/stepfunctions/StepFunctionsJsonHandler.java index a5697d0..a65dfc7 100644 --- a/src/main/java/io/github/hectorvent/floci/services/stepfunctions/StepFunctionsJsonHandler.java +++ b/src/main/java/io/github/hectorvent/floci/services/stepfunctions/StepFunctionsJsonHandler.java @@ -431,10 +431,13 @@ public class StepFunctionsJsonHandler { } private Response handleListExecutions(JsonNode request) { - List list = service.listExecutions(request.path("stateMachineArn").asText()); + Integer maxResults = optionalPageSize(request); + String statusFilter = optionalText(request, "statusFilter"); + PaginatedResult 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 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 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 EXECUTION_DATA_FIELDS = Set.of("input", "inputDetails", "output", "outputDetails"); diff --git a/src/main/java/io/github/hectorvent/floci/services/stepfunctions/StepFunctionsService.java b/src/main/java/io/github/hectorvent/floci/services/stepfunctions/StepFunctionsService.java index cb3f8e6..a119179 100644 --- a/src/main/java/io/github/hectorvent/floci/services/stepfunctions/StepFunctionsService.java +++ b/src/main/java/io/github/hectorvent/floci/services/stepfunctions/StepFunctionsService.java @@ -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 stateMachineStore; private final StorageBackend 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 taskHeartbeatNanos = new ConcurrentHashMap<>(); + private final Map historyPaginationTokens = new ConcurrentHashMap<>(); + private final Map 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>() {}); 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 listExecutions(String stateMachineArn) { + return listExecutions(stateMachineArn, null, null, null).items(); + } + + public PaginatedResult 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 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 getExecutionHistory( + String arn, boolean includeExecutionData, boolean reverseOrder, Integer maxResults, String nextToken) { + validatePageSize(maxResults); + List live = getExecutionHistory(arn); + List 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 paginateExecutions(List all, String stateMachineArn, + String statusFilter, Integer maxResults, + String nextToken) { + ListExecutionsPaginationToken cursor = listExecutionsCursor(stateMachineArn, statusFilter, maxResults, + nextToken); + List 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 paginateHistory(List 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 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); } diff --git a/src/test/java/io/github/hectorvent/floci/services/stepfunctions/StepFunctionsExecutionHistoryIntegrationTest.java b/src/test/java/io/github/hectorvent/floci/services/stepfunctions/StepFunctionsExecutionHistoryIntegrationTest.java index fd0fb48..ad24d53 100644 --- a/src/test/java/io/github/hectorvent/floci/services/stepfunctions/StepFunctionsExecutionHistoryIntegrationTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/stepfunctions/StepFunctionsExecutionHistoryIntegrationTest.java @@ -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 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 returnedArns = new ArrayList<>(); + for (Response page : List.of(first, second, third)) { + returnedArns.addAll(page.jsonPath().getList("executions.executionArn")); + List 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 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") diff --git a/src/test/java/io/github/hectorvent/floci/services/stepfunctions/StepFunctionsListExecutionsPaginationTest.java b/src/test/java/io/github/hectorvent/floci/services/stepfunctions/StepFunctionsListExecutionsPaginationTest.java new file mode 100644 index 0000000..37c6597 --- /dev/null +++ b/src/test/java/io/github/hectorvent/floci/services/stepfunctions/StepFunctionsListExecutionsPaginationTest.java @@ -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 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 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 second = fixture.service().listExecutions( + STATE_MACHINE_ARN, "SUCCEEDED", 2, first.nextToken()); + PaginatedResult third = fixture.service().listExecutions( + STATE_MACHINE_ARN, "SUCCEEDED", 2, second.nextToken()); + + List returned = new ArrayList<>(); + for (PaginatedResult 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 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 first = fixture.service().listExecutions(STATE_MACHINE_ARN, "SUCCEEDED", 4, null); + PaginatedResult second = fixture.service().listExecutions( + STATE_MACHINE_ARN, "SUCCEEDED", 4, first.nextToken()); + PaginatedResult third = fixture.service().listExecutions( + STATE_MACHINE_ARN, "SUCCEEDED", 4, second.nextToken()); + + List returned = new ArrayList<>(); + for (PaginatedResult 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 first = fixture.service().getExecutionHistory( + executionArn, true, true, 3, null); + List liveHistory = fixture.service().getExecutionHistory(executionArn); + liveHistory.add(historyEvent(11)); + liveHistory.add(historyEvent(12)); + PaginatedResult second = fixture.service().getExecutionHistory( + executionArn, true, true, 3, first.nextToken()); + PaginatedResult third = fixture.service().getExecutionHistory( + executionArn, true, true, 3, second.nextToken()); + PaginatedResult fourth = fixture.service().getExecutionHistory( + executionArn, true, true, 3, third.nextToken()); + + List returned = new ArrayList<>(); + for (PaginatedResult 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 executions = AccountAwareStorageBackend.inMemory(ACCOUNT); + StorageFactory factory = mock(StorageFactory.class); + doReturn(AccountAwareStorageBackend.inMemory(ACCOUNT)) + .doReturn(executions) + .doReturn(AccountAwareStorageBackend.inMemory(ACCOUNT)) + .doReturn(AccountAwareStorageBackend.inMemory(ACCOUNT)) + .doReturn(AccountAwareStorageBackend.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 history(int count) { + List 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 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); + } + } +}