mirrors/takahe
1
1
Fork 0
mirror of https://github.com/jointakahe/takahe.git synced 2024-11-29 02:31:00 +00:00
Code Issues Projects Releases Packages Wiki Activity

Add error states, use for fanout

Browse source
This commit is contained in:
Andrew Godwin 2022-12-20 06:23:50 +00:00
parent 94216f241a
commit 985bb95aca
3 changed files with 23 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
activities/models/fan_out.py
View file

@ -8,10 +8,12 @@ from stator.models import State, StateField, StateGraph, StatorModel
class FanOutStates(StateGraph): class FanOutStates(StateGraph):
new = State(try_interval=300) new = State(try_interval=600)
sent = State() sent = State()
failed = State()
new.transitions_to(sent) new.transitions_to(sent)
new.times_out_to(failed, seconds=86400 * 3)
@classmethod @classmethod
async def handle_new(cls, instance: "FanOut"): async def handle_new(cls, instance: "FanOut"):

10
stator/graph.py
View file

@ -98,6 +98,8 @@ class State:
self.force_initial = force_initial self.force_initial = force_initial
self.parents: set["State"] = set() self.parents: set["State"] = set()
self.children: set["State"] = set() self.children: set["State"] = set()
self.timeout_state: State | None = None
self.timeout_value: int | None = None
def _add_to_graph(self, graph: type[StateGraph], name: str): def _add_to_graph(self, graph: type[StateGraph], name: str):
self.graph = graph self.graph = graph
@ -124,6 +126,14 @@ class State:
self.children.add(other) self.children.add(other)
other.parents.add(other) other.parents.add(other)
def times_out_to(self, other: "State", seconds: int):
if self.timeout_state is not None:
raise ValueError("Timeout state already set!")
self.timeout_state = other
self.timeout_value = seconds
self.children.add(other)
other.parents.add(other)
@property @property
def initial(self): def initial(self):
return self.force_initial or (not self.parents) return self.force_initial or (not self.parents)

12
stator/models.py
View file

@ -159,7 +159,7 @@ class StatorModel(models.Model):
""" """
Attempts to transition the current state by running its handler(s). Attempts to transition the current state by running its handler(s).
""" """
current_state = self.state_graph.states[self.state] current_state: State = self.state_graph.states[self.state]
# If it's a manual progression state don't even try # If it's a manual progression state don't even try
# We shouldn't really be here in this case, but it could be a race condition # We shouldn't really be here in this case, but it could be a race condition
if current_state.externally_progressed: if current_state.externally_progressed:
@ -168,7 +168,7 @@ class StatorModel(models.Model):
) )
return None return None
try: try:
next_state = await current_state.handler(self) next_state = await current_state.handler(self) # type: ignore
except BaseException as e: except BaseException as e:
await exceptions.acapture_exception(e) await exceptions.acapture_exception(e)
traceback.print_exc() traceback.print_exc()
@ -184,6 +184,14 @@ class StatorModel(models.Model):
) )
await self.atransition_perform(next_state) await self.atransition_perform(next_state)
return next_state return next_state
# See if it timed out
if (
current_state.timeout_value
and current_state.timeout_value
<= (timezone.now() - self.state_changed).total_seconds()
):
await self.atransition_perform(current_state.timeout_state)
return current_state.timeout_state
await self.__class__.objects.filter(pk=self.pk).aupdate( await self.__class__.objects.filter(pk=self.pk).aupdate(
state_attempted=timezone.now(), state_attempted=timezone.now(),
state_locked_until=None, state_locked_until=None,