Skip to content
Merged
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
132 changes: 88 additions & 44 deletions src/aws_durable_execution_sdk_python/concurrency.py
Original file line number Diff line number Diff line change
Expand Up @@ -386,25 +386,63 @@ def fail_task(self) -> None:
with self._lock:
self.failure_count += 1

def should_complete(self) -> bool:
"""Check if execution should complete."""
def should_continue(self) -> bool:
"""
Check if we should continue starting new tasks (based on failure tolerance).
Matches TypeScript shouldContinue() logic.
"""
with self._lock:
# Success condition
if self.success_count >= self.min_successful:
return True
# If no completion config, only continue if no failures
if (
self.tolerated_failure_count is None
and self.tolerated_failure_percentage is None
):
return self.failure_count == 0

# Failure conditions
if self._is_failure_condition_reached(
tolerated_count=self.tolerated_failure_count,
tolerated_percentage=self.tolerated_failure_percentage,
failure_count=self.failure_count,
# Check failure count tolerance
if (
self.tolerated_failure_count is not None
and self.failure_count > self.tolerated_failure_count
):
return True
return False

# Impossible to succeed condition
# TODO: should this keep running? TS doesn't currently handle this either.
remaining_tasks = self.total_tasks - self.success_count - self.failure_count
return self.success_count + remaining_tasks < self.min_successful
# Check failure percentage tolerance
if self.tolerated_failure_percentage is not None and self.total_tasks > 0:
failure_percentage = (self.failure_count / self.total_tasks) * 100
if failure_percentage > self.tolerated_failure_percentage:
return False

return True

def is_complete(self) -> bool:
"""
Check if execution should complete (based on completion criteria).
Matches TypeScript isComplete() logic.
"""
with self._lock:
completed_count = self.success_count + self.failure_count

# All tasks completed
if completed_count == self.total_tasks:
# Complete if no failure tolerance OR no failures OR min successful reached
return (
(
self.tolerated_failure_count is None
and self.tolerated_failure_percentage is None
)
or self.failure_count == 0
or self.success_count >= self.min_successful
)

# when we breach min successful, we've completed
return self.success_count >= self.min_successful

def should_complete(self) -> bool:
"""
Check if execution should complete.
Combines TypeScript shouldContinue() and isComplete() logic.
"""
return self.is_complete() or not self.should_continue()

def is_all_completed(self) -> bool:
"""True if all tasks completed successfully."""
Expand Down Expand Up @@ -690,40 +728,46 @@ def _on_task_complete(
self._completion_event.set()

def _create_result(self) -> BatchResult[ResultType]:
"""Build the final BatchResult."""
batch_items: list[BatchItem[ResultType]] = []
completed_branches: list[ExecutableWithState] = []
failed_branches: list[ExecutableWithState] = []
"""
Build the final BatchResult.

When this function executes, we've terminated the upper/parent context for whatever reason.
It follows that our items can be only in 3 states, Completed, Failed and Started (in all of the possible forms).
We tag each branch based on its observed value at the time of completion of the parent / upper context, and pass the
results to BatchResult.

Any inference wrt completion reason is left up to BatchResult, keeping the logic inference isolated.
"""
batch_items: list[BatchItem[ResultType]] = []
for executable in self.executables_with_state:
if executable.status is BranchStatus.COMPLETED:
completed_branches.append(executable)
batch_items.append(
BatchItem(
executable.index, BatchItemStatus.SUCCEEDED, executable.result
match executable.status:
case BranchStatus.COMPLETED:
batch_items.append(
BatchItem(
executable.index,
BatchItemStatus.SUCCEEDED,
executable.result,
)
)
)
elif executable.status is BranchStatus.FAILED:
failed_branches.append(executable)
batch_items.append(
BatchItem(
executable.index,
BatchItemStatus.FAILED,
error=ErrorObject.from_exception(executable.error),
case BranchStatus.FAILED:
batch_items.append(
BatchItem(
executable.index,
BatchItemStatus.FAILED,
error=ErrorObject.from_exception(executable.error),
)
)
case (
BranchStatus.PENDING
| BranchStatus.RUNNING
| BranchStatus.SUSPENDED
| BranchStatus.SUSPENDED_WITH_TIMEOUT
):
batch_items.append(
BatchItem(executable.index, BatchItemStatus.STARTED)
)
)

completion_reason: CompletionReason = (
CompletionReason.ALL_COMPLETED
if self.counters.is_all_completed()
else (
CompletionReason.MIN_SUCCESSFUL_REACHED
if self.counters.is_min_successful_reached()
else CompletionReason.FAILURE_TOLERANCE_EXCEEDED
)
)

return BatchResult(batch_items, completion_reason)
return BatchResult.from_items(batch_items, self.completion_config)

def _execute_item_in_child_context(
self,
Expand Down
Loading