in source/client/python/api-v0.1/api/state_table_dynamodb.py [0:0]
def refresh_ttl_for_ongoing_task(self, task_id, agent_id, new_expirtaion_timestamp):
"""
Function updates TTL for the task
Approximately equivalent to SQL:
Alter table state_table where TaskId == wu.getTaskId()
set HeartbeatExpirationTimestamp = expiration_timestamp
condition to status == Running and OwnerID == SelfWorkerID
Args:
task_id: the task to update
agent_id: expected worker id that is working on this task.
new_expirtaion_timestamp: a timestamp in the "future" when this task will be considered expired.
Returns:
True if successful
Throws:
StateTableException on throttling
StateTableException on condition
Exception for all other errors
"""
session_id = self.__get_session_id_from_task_id(task_id)
refresh_is_successful = True
try:
self.state_table.update_item(
Key={
'task_id': task_id
},
UpdateExpression="SET #var_heartbeat_expiration_timestamp = :val3",
ExpressionAttributeValues={
':val3': new_expirtaion_timestamp,
},
ExpressionAttributeNames={
"#var_heartbeat_expiration_timestamp": "heartbeat_expiration_timestamp",
},
ConditionExpression=Key('task_status').eq(
self.__make_task_state_from_session_id(TASK_STATE_PROCESSING, session_id)
) & Key('task_owner').eq(agent_id)
)
except ClientError as e:
if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
task_row = self.get_task_by_id(task_id, consistent_read=True)
msg = f"Could not update TTL on the own task [{task_id}] agent: [{agent_id}] state: [{self.__make_task_state_from_session_id(TASK_STATE_PROCESSING, session_id)}], did TTL Lambda re-assigned it? TaskRow: [{task_row}] {e}"
logging.warning(msg)
raise StateTableException(e, msg, caused_by_condition=True)
elif e.response['Error']['Code'] in ["ThrottlingException", "ProvisionedThroughputExceededException"]:
msg = f"Could not update TTL on the own task [{task_id}] agent: [{agent_id}], Throttling Exception {e}"
logging.warning(msg)
raise StateTableException(e, msg, caused_by_throttling=True)
else:
msg = f"Could not update TTL on the own task [{task_id}] agent: [{agent_id}]: {e}"
logging.error(msg)
raise Exception(e)
except Exception as e:
msg = f"Could not update TTL on the own task [{task_id}]: {e}"
logging.error(msg)
raise e
return refresh_is_successful