Long-running entity workflows with internal scheduling

Community Support
1

I’m creating a long-running entity workflow as described at Very Long-Running Workflows | Temporal Technologies. What is the best way to implement the functionality described in that post as “Wait for Billing Period”? For example, lets say I want a workflow to do something every Friday at 7pm.

The approaches I have considered are:

Approach 1 would be to simply Workflow.sleep for a duration equal to the target time minus the current time.

Approach 2 would be to create a separate scheduled workflow that runs at the target time, and sends the entity workflow a signal which causes it to start its work.

Any other good approaches to this?

Related to this is perhaps the requirement to insert a new activity in between the current time and the next scheduled action for the entity.

With approach 1, would changing the sleep time to a lower value and adding a new action before the one the sleep is currently waiting on (still for some time in the future, so no previous activities are affected) be an indeterministic change?

The same scenario with adding a new signal using approach #2 would I think be no problem.

Which approach is easier to manage long-term?

Thanks!

2

I would use a single workflow approach. It is easy to create a timer and call a function when it fires to avoid blocking sleep. Which SDK are you using, Java?

3

Yes, Java.

4

You can do something like:

Workflow.newTimer(...).thenApply(()->handleTimeout());
...
function handleTimeout() {
}
5

And if I want to update the workflow to schedule a new activity that should occur before the next timer currently scheduled, is there a way to do that?

So, say for example, it is currently Monday at 5pm, I have a timer scheduled for Friday at 5pm, but a new business requirement arises and I need the long-running workflow to do something on Wednesday at 8pm.

I believe new timers can only be scheduled when the workflow wakes up, which won’t happen until the previous timer fires, so with the approach @maxim suggested, the Wednesday at 8pm timer will be missed, right?

The solution I have in mind will fire a timer on a much more frequent basis (say, every minute), which will wake the workflow up and allow it to determine what the next thing to do is. So instead of using:

Workflow.newTimer(...).thenApply(()->doSomethingFriday5pm());

the workflow would instead do something like (in pseudo-code):

var expectedSchedule = listOf(...)

while (true) {
  currentTime = Workflow.currentTimeMillis()
  action = sideEffect { actionAt(currentTime) }
  if (action != null) execute(action)
  Workflow.sleep(1, TimeUnit.MINUTES)
}

function actionAt(at): ExpectedEvent? {
  return expectedEvents.removeFirstOrNull { it -> it.at > at }
}

And then I believe updates to expectedSchedule, either via signals or via code updates, will be deterministic, and new actions can be added to the list in advance of existing actions, and because of sideEffect, I can remove older actions from the list as well.

It seems like there should be a simpler way though, and one that perhaps will cost less in terms of actions executed on the cluster?

6

Firing a timer every minute is not recommended as it will grow the workflow history even if the workflow is not doing anything useful.

So, say for example, it is currently Monday at 5pm, I have a timer scheduled for Friday at 5pm, but a new business requirement arises and I need the long-running workflow to do something on Wednesday at 8pm.

You have two options. You either cancel the original timer or await call and reschedule for the new one. Or you create a second timer that fires asynchronously at the new time.

I don’t understand why your workflow code uses an external variable and side effect. I would recommend maintaining the list of expectedEvents as a workflow field and update it from a signal or update handler that is used to notify about the schedule changes. Then change the Workflow.sleep to a Workflow.await which blocks on a variable that is set to true from an update handler.

See this sample which implements similar idea: samples-java/core/src/main/java/io/temporal/samples/updatabletimer at main · temporalio/samples-java · GitHub

1 Like
7

Thank you, I can see I wasn’t thinking in the right direction. I’ve updated the implementation based on the information you provided. The fundamental realization to me was that timers can be cancelled / terminated based on conditions. This is the updated core logic within the workflow:

  private val scheduledActions: TreeSet<ScheduledAction> =  TreeSet(...)

  private tailrec fun awaitNextAction(): ScheduledAction {
    val currentTime = Instant.ofEpochMilli(Workflow.currentTimeMillis())
    val action = scheduledActions.firstOrNull()

    return if (action == null) {
      log.info("Awaiting any action")
      Workflow.await { scheduledActions.isNotEmpty() }
      awaitNextAction()
    } else {
      val duration = action.at - currentTime
      if (duration <= ACTION_EXEC_WITHIN) {
        log.info("Action $action scheduled to run soon in $duration, triggering now")
        action
      } else {
        log.info("Awaiting $duration for action $action to trigger")
        val conditionTriggered = Workflow.await(duration.toJavaDuration()) { 
          scheduledActions.firstOrNull() != action
        }
        if (!conditionTriggered) {
          action
        } else {
          log.info("Action $action is no longer the first action scheduled")
          awaitNextAction()
        }
      }
    }
  }

and the awaitNextAction is simply called in the workflow in an infinite loop which executes the action it returns and calls it again.

The workflow variable scheduledActions can be updated via signals or queried, of course, or updated within the workflow based on business logic and the results of previous actions. It seems to work well, but feedback welcome.

8

This looks good. I would also consider calling continue-as-new after several iterations if the number of actions is unbounded.

1 Like