Repository navigation
fix: export visible replay completions #869
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
zhongkechen
wants to merge
4
commits into
main
Choose a base branch
from
codex/fix-854-otel-replay-completions
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
Open
Changes from all commits
Commits
Show all changes
4 commits
Select commit
Hold shift + click to select a range
626c175
fix: export visible replay completions
zhongkechen 2b18fa1
fix: resolve replay completion merge conflicts
zhongkechen 7f018e4
test(otel): fail when a completion aborts invocation End
zhongkechen 8fde831
fix(otel): contain completion failures at invocation End
zhongkechen File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
137 changes: 137 additions & 0 deletions
137
...-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/_invocation_state.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,137 @@ | ||
| """Visible completion metadata owned by one instrumentation invocation.""" | ||
|
|
||
| from collections.abc import Mapping | ||
| from graphlib import TopologicalSorter | ||
|
|
||
| from aws_durable_execution_sdk_python.plugin import ( | ||
| InvocationStartInfo, | ||
| OperationInfo, | ||
| OperationStatus, | ||
| OperationType, | ||
| ) | ||
|
|
||
|
|
||
| _TERMINAL_STATUSES = frozenset( | ||
| { | ||
| OperationStatus.SUCCEEDED, | ||
| OperationStatus.FAILED, | ||
| OperationStatus.TIMED_OUT, | ||
| OperationStatus.STOPPED, | ||
| OperationStatus.CANCELLED, | ||
| } | ||
| ) | ||
|
|
||
|
|
||
| class CompletionState: | ||
| """Pending deliveries for one invocation, guarded by its span-registry lock. | ||
|
|
||
| Callers hold that lock for every access, including parent registration and | ||
| dequeue. Span rendering runs outside the lock, between claim and release. | ||
| Only complete records successful delivery; a failed attempt stays pending. | ||
| """ | ||
|
|
||
| def __init__(self) -> None: | ||
| self.pending: dict[str, OperationInfo] = {} | ||
| # An ordered ID index; completion payloads live only in pending. | ||
| self.children_by_parent: dict[str, dict[str, None]] = {} | ||
| self.in_flight: set[str] = set() | ||
| self.delivered: set[str] = set() | ||
|
|
||
| def remember(self, updates: Mapping[str, OperationInfo]) -> None: | ||
| """Keep explicit terminal updates that have not already been delivered.""" | ||
| for operation_id, operation in updates.items(): | ||
| if ( | ||
| operation.status in _TERMINAL_STATUSES | ||
| and not operation.is_replayed | ||
| and operation_id not in self.delivered | ||
| ): | ||
| self.pending[operation_id] = operation | ||
| self._index_child(operation) | ||
|
|
||
| def _index_child(self, operation: OperationInfo) -> None: | ||
| if operation.parent_id is not None: | ||
| self.children_by_parent.setdefault(operation.parent_id, {})[ | ||
| operation.operation_id | ||
| ] = None | ||
|
|
||
| def take_children(self, parent_id: str) -> list[OperationInfo]: | ||
| """Dequeue children in observation order, retaining undelivered payloads.""" | ||
| return [ | ||
| self.pending[operation_id] | ||
| for operation_id in self.children_by_parent.pop(parent_id, {}) | ||
| ] | ||
|
|
||
| def claim( | ||
| self, operation: OperationInfo, *, parent_available: bool, repeat: bool = False | ||
| ) -> bool: | ||
| """Claim delivery, or defer it until its parent is available. | ||
|
|
||
| A newly opened context segment may repeat a delivered operation, but | ||
| it must still respect another caller's in-flight claim. | ||
| """ | ||
| operation_id = operation.operation_id | ||
| if operation_id in self.in_flight or ( | ||
| operation_id in self.delivered and not repeat | ||
| ): | ||
| return False | ||
| self.pending[operation_id] = operation | ||
| if not parent_available: | ||
| self._index_child(operation) | ||
| return False | ||
| self.in_flight.add(operation_id) | ||
| return True | ||
|
|
||
| def complete(self, operation: OperationInfo) -> None: | ||
| """Record success and remove the payload and any remaining index entry.""" | ||
| operation_id = operation.operation_id | ||
| self.delivered.add(operation_id) | ||
| self.pending.pop(operation_id, None) | ||
| if operation.parent_id is not None: | ||
| siblings = self.children_by_parent.get(operation.parent_id) | ||
| if siblings is not None: | ||
| siblings.pop(operation_id, None) | ||
| if not siblings: | ||
| del self.children_by_parent[operation.parent_id] | ||
|
|
||
| def release(self, operation_id: str) -> None: | ||
| """Release a claim after either successful or failed rendering.""" | ||
| self.in_flight.discard(operation_id) | ||
|
|
||
| def clear(self) -> None: | ||
| """Release all invocation-owned delivery metadata.""" | ||
| self.pending.clear() | ||
| self.children_by_parent.clear() | ||
| self.in_flight.clear() | ||
| self.delivered.clear() | ||
|
|
||
|
|
||
| def completed_input_parent_ids(info: InvocationStartInfo) -> set[str]: | ||
| """Visible completed contexts may parent input or later checkpoint updates.""" | ||
| return { | ||
| operation.operation_id | ||
| for operation in info.operations.values() | ||
| if operation.operation_type is OperationType.CONTEXT | ||
| and operation.status in _TERMINAL_STATUSES | ||
| } | ||
|
|
||
|
|
||
| def parent_first( | ||
| operations: Mapping[str, OperationInfo], | ||
| ) -> list[OperationInfo]: | ||
| """Materialize pending ancestors before referencing them for descendants. | ||
|
|
||
| Durable operations form a parent tree. Missing ancestors remain unresolved; | ||
| their view-specific fallback is chosen by the plugin, without inventing an | ||
| operation or treating its non-recording reference as an exported completion. | ||
| """ | ||
| order = TopologicalSorter( | ||
| { | ||
| operation_id: ( | ||
| (operation.parent_id,) | ||
| if operation.parent_id is not None and operation.parent_id in operations | ||
| else () | ||
| ) | ||
| for operation_id, operation in operations.items() | ||
| } | ||
| ) | ||
| return [operations[operation_id] for operation_id in order.static_order()] | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Codex AI review · Finding
arf_v1_slgcwqvuap7vij4rfw7nejw7q6[P2] Revisit the untested parent-index pruning branch
This breaks CONTRIBUTING rule 4, “Every branch has a test that fails without it.” Removing this branch leaves empty parent-index entries, but every index-emptiness assertion runs after invocation End unconditionally clears the state. The sibling regression also checks delivery only after finalization, whose scan of
pendingmasks an incorrectly removed sibling index. Consequently, the tests do not establish the new index-retention contract. Return to the design phase under rule 8, step 6, as rule 9 requires, and justify the necessary state transitions with tests that distinguish them from final cleanup.To suppress repeats, reply
/ai verdict deferred,/ai verdict false-positive, or/ai verdict out-of-scopeon the first nonblank line. Add rationale on later lines. A new/ai verdict acceptedallows future repeats again.