fix(gitops): recover safe pre-side-effect continuation retries
This commit is contained in:
@@ -446,7 +446,12 @@ class ContinuationExecutor:
|
|||||||
"lease_seconds": request.lease_seconds,
|
"lease_seconds": request.lease_seconds,
|
||||||
}
|
}
|
||||||
|
|
||||||
def _validate_preflight(self, binding: ContinuationRuntimeBinding) -> None:
|
def _validate_preflight(
|
||||||
|
self,
|
||||||
|
binding: ContinuationRuntimeBinding,
|
||||||
|
*,
|
||||||
|
allow_same_generation: bool = False,
|
||||||
|
) -> None:
|
||||||
request = binding.request
|
request = binding.request
|
||||||
normalized = validate_writer_lease_request_v2(self._request_mapping(request), self.authority)
|
normalized = validate_writer_lease_request_v2(self._request_mapping(request), self.authority)
|
||||||
expected_normalized = {
|
expected_normalized = {
|
||||||
@@ -465,7 +470,7 @@ class ContinuationExecutor:
|
|||||||
raise ContinuationExecutionError("LEASE_NOT_ACTIVE")
|
raise ContinuationExecutionError("LEASE_NOT_ACTIVE")
|
||||||
if now >= binding.lease_expires_at:
|
if now >= binding.lease_expires_at:
|
||||||
raise ContinuationExecutionError("LEASE_EXPIRED")
|
raise ContinuationExecutionError("LEASE_EXPIRED")
|
||||||
if request.generation <= self._max_generation(request.target_branch):
|
if request.generation <= self._max_generation(request.target_branch) and not allow_same_generation:
|
||||||
raise ContinuationExecutionError("GENERATION_NOT_MONOTONIC")
|
raise ContinuationExecutionError("GENERATION_NOT_MONOTONIC")
|
||||||
if not self.workspace.is_detached():
|
if not self.workspace.is_detached():
|
||||||
raise ContinuationExecutionError("WORKTREE_NOT_DETACHED")
|
raise ContinuationExecutionError("WORKTREE_NOT_DETACHED")
|
||||||
@@ -495,17 +500,26 @@ class ContinuationExecutor:
|
|||||||
current = self.controller.store.read(submitted.job_id)
|
current = self.controller.store.read(submitted.job_id)
|
||||||
if JobState(current["state"]) == JobState.COMPLETED:
|
if JobState(current["state"]) == JobState.COMPLETED:
|
||||||
return self.controller.store._receipt(current, duplicate=True).as_dict()
|
return self.controller.store._receipt(current, duplicate=True).as_dict()
|
||||||
if JobState(current["state"]) in {JobState.EXECUTING, JobState.RECOVERABLE_FAILURE}:
|
retry_same_generation = False
|
||||||
|
if JobState(current["state"]) in {
|
||||||
|
JobState.EXECUTING,
|
||||||
|
JobState.RECOVERABLE_FAILURE,
|
||||||
|
JobState.RECONCILING,
|
||||||
|
}:
|
||||||
receipt = json.loads(current["receipt_json"]) if current["receipt_json"] else {}
|
receipt = json.loads(current["receipt_json"]) if current["receipt_json"] else {}
|
||||||
|
if not receipt.get("recovery_safe_before_git_side_effect"):
|
||||||
|
raise ContinuationExecutionError("RECONCILIATION_REQUIRED")
|
||||||
if any(key in receipt for key in ("authority_oid", "desired_remote_oid", "git_remote_oid", "api_branch_oid")):
|
if any(key in receipt for key in ("authority_oid", "desired_remote_oid", "git_remote_oid", "api_branch_oid")):
|
||||||
raise ContinuationExecutionError("RECONCILIATION_REQUIRED")
|
raise ContinuationExecutionError("RECONCILIATION_REQUIRED")
|
||||||
|
retry_same_generation = True
|
||||||
|
if JobState(current["state"]) != JobState.RECONCILING:
|
||||||
self.controller.advance(
|
self.controller.advance(
|
||||||
submitted.job_id,
|
submitted.job_id,
|
||||||
JobState.RECONCILING,
|
JobState.RECONCILING,
|
||||||
side_effect_state="PRE_SIDE_EFFECT_RETRY",
|
side_effect_state="PRE_SIDE_EFFECT_RETRY",
|
||||||
receipt={"recovery_safe_before_git_side_effect": True},
|
receipt={"recovery_safe_before_git_side_effect": True},
|
||||||
)
|
)
|
||||||
self._validate_preflight(binding)
|
self._validate_preflight(binding, allow_same_generation=retry_same_generation)
|
||||||
request = binding.request
|
request = binding.request
|
||||||
projected = False
|
projected = False
|
||||||
try:
|
try:
|
||||||
|
|||||||
Reference in New Issue
Block a user