Microservice orchestration
Coordinate concurrent service calls, accept an in-flight state update, and finish through an external event or a timeout fallback.
Product requirements
The process starts with a data value and must:
- Call API 1, then persist the input in the Data Attribute.
- Start CallAPI2 and CallAPI3 concurrently.
- Let CallAPI2 read the current data, call API 2, and end its branch.
- Keep CallAPI3 waiting for either one Ready Channel message or a 24-hour durable Timer.
- Allow a caller to invoke the Swap RPC while the Flow is running. The RPC must return the previous value and replace it with the new value atomically.
- After a Ready message, call API 3 with the latest data and complete the Flow.
- If the Timer wins, call API 3, continue to CallAPI4, call API 4 with the latest data, and then complete.
The wait is durable. A Worker or Dex restart does not lose the Channel message, Timer deadline, persisted data, or current Step. A Ready message normally arrives from another service after its asynchronous work is complete. If it never arrives, the Timer selects the fallback path.
Potential failures include:
- A downstream API can be unavailable, time out, or reject its input. The Step must report that failure so its configured retry and failure policy can act on it.
- A downstream call can succeed while its response is lost. Production API operations must be idempotent because Dex may retry the Step.
- A Swap RPC or Ready publish can target an unknown or completed Flow and return a client error. Callers should retain the Flow ID and handle that response.
- No Ready message may arrive. This is an expected outcome, not lost progress; the durable Timer eventually routes the Flow through CallAPI4.
- CallAPI2 and CallAPI3 can observe different values when Swap runs between their Attribute reads. The process intentionally guarantees that each read is durable, not that both concurrent branches share a snapshot.
The examples use local dependency stubs. Production adapters should add timeouts, idempotency keys, and the retry or recovery policy appropriate for each API.
Flow design
This interactive definition graph is generated from the runnable Python Flow.
The Flow has one fan-out point and two ways to finish:
CallAPI1 ─┬─▶ CallAPI2 ─▶ DeadEnd
│
└─▶ Wait(Ready | 24h Timer) ─▶ CallAPI3 ─┬─▶ Complete
└─▶ CallAPI4 ─▶ Complete
(Timer only)
It combines these design patterns:
- Static parallel Steps fan out two known service calls without placing both operations in one retry boundary.
- Durable external-event gate uses Wait.anyOf with a Channel and Timer. The Channel handles the normal callback; the Timer guarantees bounded waiting.
- Timeout fallback sends only the timed-out execution to CallAPI4. The normal Channel path skips that fallback.
- Request-response interaction exposes Swap as a typed RPC so a caller can change persisted state and receive the previous value in one operation.
- Durable shared state stores the current payload in an Attribute. Both parallel branches read it at execution time instead of carrying a stale copy through the graph.
- Branch termination lets CallAPI2 return DeadEnd while the waiting branch owns the business completion result.
Each downstream API has its own Step. A retry of API 3 therefore does not repeat API 1 or API 2, and the Step execution graph shows exactly which call is waiting, retrying, or complete.
Core implementation
The tabs show the Flow signature, topology, persistence schema, Swap RPC, parallel fan-out, and Channel-or-Timer wait. Each linked example also includes its dependency stub, HTTP controller, and end-to-end integration test.
class OrchestrationFlow(Flow[str]):
data = Attribute("data", str)
ready = Channel[None]("Ready", type(None))
def __init__(self, service: MyDependencyService) -> None:
self.service = service
self.call_api4 = CallAPI4(service, self.data)
self.call_api3 = CallAPI3(service, self.data, self.ready, self.call_api4)
self.call_api2 = CallAPI2(service, self.data)
self.call_api1 = CallAPI1(service, self.data, self.call_api2, self.call_api3)
def get_steps(self) -> StepList[str]:
return StepList.start_step(self.call_api1).other_steps(
self.call_api2,
self.call_api3,
self.call_api4,
)
def get_persistence_schema(self) -> PersistenceSchema:
return PersistenceSchema.of(self.data, self.ready)
@rpc
def swap(self, context: Context, new_data: str) -> RPCResult[str]:
old_data = self.data.get(context)
self.data.set(context, new_data)
return RPCResult(old_data)
class CallAPI1(Step[str]):
def execute(self, context: Context, input: str) -> StepDecision:
self.service.call_api1(input)
self.data.set(context, input)
return go_to_many(
StepMovement.of(CallAPI2, None),
StepMovement.of(CallAPI3, None),
)
class CallAPI3(Step[None]):
def wait_for(self, context: Context, input: None) -> Wait:
return Wait.any_of(
Timer.by_duration(timedelta(hours=24)),
self.ready.for_one(),
)
def execute(self, context: Context, input: None) -> StepDecision:
value = self.data.get(context)
self.service.call_api3(value)
if context.has_timer_fired():
return go_to(CallAPI4, None)
return graceful_complete(value)
Example: examples/python/dex_examples/products/microservices/orchestration_flow.py
Demo
Run Dex and any language example server, then start Microservice orchestration from the examples playground. For the run below, the Flow started with test initial data, the Swap RPC replaced it with updated in-flight data, and a Ready message released the wait. CallAPI1, CallAPI2, and CallAPI3 completed; CallAPI4 did not run because the Channel won before the 24-hour Timer.
