Async.function control flow in command queue pattern

Community Support
,
1

I’m implementing a command queue pattern in my Temporal workflow similar to the Injecting work into the main Workflow pattern described in the docs. My workflow has a main event loop that processes commands from a queue, and an update method that adds commands to the queue.

Here’s a simplified version of my implementation:

val commandQueue: Queue<Execution> = LinkedList()

@WorkflowMethod
fun myWorkflowMethod() {
    while (!currentState.isTerminal) {
        Workflow.await { commandQueue.isNotEmpty() }
        executeTopCommand() // polls the top of the queue and calls an activity
    }
}

@UpdateMethod
fun post(command: Command): ExecutionRecord {
    val execution = postCommand(command)
    return Async.function {
        execution.completedRecord[TIMEOUT]
    }
        .exceptionally { execution.record }
        .get()
}

data class Execution(
    val record: ExecutionRecord, // contains the command and some metadata
    val completedRecord: CompletablePromise<ExecutionRecord>, // contains the result of a completed command
)

My question

Does the Async.function in the post method yield control back to the main workflow thread immediately? Or does it block until the completedRecord promise resolves or times out?

I want to ensure that:

  1. The main workflow loop can continue processing other commands in the queue while a post is waiting for a result
  2. Multiple simultaneous calls to post don’t block each other

Any insights or best practices around this pattern would be appreciated!

2

Promise.get yields control to the Temporal event loop. So nothing is blocked.

3

Just to add for the use case, since inside the update handler itself both postCommand and the .get on promise is blocking you could run into max concurrent update limit per workflow execution.
Dynamic config:
history.maxInFlightUpdates
default 10
So if these two blocking calls inside update handler take some time while you receive a burst of updates which you don’t reject some, be mindful of hitting limit.

If this is a concern maybe using sliding window pattern (sample here) could be alternative option?

4

Thank you maxim!

5

That’s super good to know, I didn’t know about that limit. Few questions around this:

  1. What happens to an incoming request if the limit is reached?
  2. Is there a limit to how high we can configure that the maxInFlightUpdates value? What is the tradeoff of increasing it?
  3. Is there any way to query Temporal at runtime in the update method to determine how many in-flight requests there currently are? I’m thinking we could have a dynamic approach where if we’re near the limit we immediately return without creating a promise.