Feat: expose retry count

examples/simple/worker.py

@@ -15,7 +15,7 @@ class MyWorkflow:
     def __init__(self):
         self.my_value = "test"
 
-    @hatchet.step()
+    @hatchet.step(timeout="2s", retries=3)
     def step1(self, context: Context):
         print("executed step1")
         time.sleep(10)

hatchet_sdk/clients/dispatcher.py

@@ -80,6 +80,7 @@ def __init__(
         action_id: str,
         action_payload: str,
         action_type: ActionType,
+        retry_count: int,
     ):
         self.worker_id = worker_id
         self.workflow_run_id = workflow_run_id
@@ -93,6 +94,7 @@ def __init__(
         self.action_id = action_id
         self.action_payload = action_payload
         self.action_type = action_type
+        self.retry_count = retry_count
 
 
 class WorkerActionListener:
@@ -264,6 +266,7 @@ async def _generator(self) -> AsyncGenerator[Action, None]:
                         action_id=assigned_action.actionId,
                         action_payload=action_payload,
                         action_type=action_type,
+                        retry_count=assigned_action.retryCount,
                     )
 
                     yield action

hatchet_sdk/clients/rest/models/step_run_event_reason.py

@@ -41,6 +41,7 @@ class StepRunEventReason(str, Enum):
     REASSIGNED = "REASSIGNED"
     TIMED_OUT = "TIMED_OUT"
     SLOT_RELEASED = "SLOT_RELEASED"
+    RETRIED_BY_USER = "RETRIED_BY_USER"
 
     @classmethod
     def from_json(cls, json_str: str) -> Self:

hatchet_sdk/clients/rest/models/worker.py

@@ -39,6 +39,11 @@ class Worker(BaseModel):
         description="The time this worker last sent a heartbeat.",
         alias="lastHeartbeatAt",
     )
+    last_listener_established: Optional[datetime] = Field(
+        default=None,
+        description="The time this worker last sent a heartbeat.",
+        alias="lastListenerEstablished",
+    )
     actions: Optional[List[StrictStr]] = Field(
         default=None, description="The actions this worker can perform."
     )
@@ -71,6 +76,7 @@ class Worker(BaseModel):
         "metadata",
         "name",
         "lastHeartbeatAt",
+        "lastListenerEstablished",
         "actions",
         "recentStepRuns",
         "status",
@@ -156,6 +162,7 @@ def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:
                 ),
                 "name": obj.get("name"),
                 "lastHeartbeatAt": obj.get("lastHeartbeatAt"),
+                "lastListenerEstablished": obj.get("lastListenerEstablished"),
                 "actions": obj.get("actions"),
                 "recentStepRuns": (
                     [StepRun.from_dict(_item) for _item in obj["recentStepRuns"]]

hatchet_sdk/context.py

@@ -209,3 +209,6 @@ def refresh_timeout(self, increment_by: str):
             )
         except Exception as e:
             logger.error(f"Error refreshing timeout: {e}")
+
+    def retry_count(self):
+        return self.action.retry_count

hatchet_sdk/dispatcher_pb2.pyi

@@ -92,7 +92,7 @@ class WorkerRegisterResponse(_message.Message):
     def __init__(self, tenantId: _Optional[str] = ..., workerId: _Optional[str] = ..., workerName: _Optional[str] = ...) -> None: ...
 
 class AssignedAction(_message.Message):
-    __slots__ = ("tenantId", "workflowRunId", "getGroupKeyRunId", "jobId", "jobName", "jobRunId", "stepId", "stepRunId", "actionId", "actionType", "actionPayload", "stepName")
+    __slots__ = ("tenantId", "workflowRunId", "getGroupKeyRunId", "jobId", "jobName", "jobRunId", "stepId", "stepRunId", "actionId", "actionType", "actionPayload", "stepName", "retryCount")
     TENANTID_FIELD_NUMBER: _ClassVar[int]
     WORKFLOWRUNID_FIELD_NUMBER: _ClassVar[int]
     GETGROUPKEYRUNID_FIELD_NUMBER: _ClassVar[int]
@@ -105,6 +105,7 @@ class AssignedAction(_message.Message):
     ACTIONTYPE_FIELD_NUMBER: _ClassVar[int]
     ACTIONPAYLOAD_FIELD_NUMBER: _ClassVar[int]
     STEPNAME_FIELD_NUMBER: _ClassVar[int]
+    RETRYCOUNT_FIELD_NUMBER: _ClassVar[int]
     tenantId: str
     workflowRunId: str
     getGroupKeyRunId: str
@@ -117,7 +118,8 @@ class AssignedAction(_message.Message):
     actionType: ActionType
     actionPayload: str
     stepName: str
-    def __init__(self, tenantId: _Optional[str] = ..., workflowRunId: _Optional[str] = ..., getGroupKeyRunId: _Optional[str] = ..., jobId: _Optional[str] = ..., jobName: _Optional[str] = ..., jobRunId: _Optional[str] = ..., stepId: _Optional[str] = ..., stepRunId: _Optional[str] = ..., actionId: _Optional[str] = ..., actionType: _Optional[_Union[ActionType, str]] = ..., actionPayload: _Optional[str] = ..., stepName: _Optional[str] = ...) -> None: ...
+    retryCount: int
+    def __init__(self, tenantId: _Optional[str] = ..., workflowRunId: _Optional[str] = ..., getGroupKeyRunId: _Optional[str] = ..., jobId: _Optional[str] = ..., jobName: _Optional[str] = ..., jobRunId: _Optional[str] = ..., stepId: _Optional[str] = ..., stepRunId: _Optional[str] = ..., actionId: _Optional[str] = ..., actionType: _Optional[_Union[ActionType, str]] = ..., actionPayload: _Optional[str] = ..., stepName: _Optional[str] = ..., retryCount: _Optional[int] = ...) -> None: ...
 
 class WorkerListenRequest(_message.Message):
     __slots__ = ("workerId",)

lint.sh

@@ -0,0 +1 @@
+pre-commit run --all-files || pre-commit run --all-files