Skip to main content

Advanced: Back Pressure for High Availability

Use a short SubmitRequestFlow when the target parent may reject an overloaded request.

Definition graph

SubmitRequestFlow

Valid

python · examples/python/dex_examples/patterns/parallel-subflows/submit_request_flow.py

The client starts SubmitRequestFlow instead of retrying an RPC in memory. SubmitStep partitions the request, calls the selected parent's SendRequest RPC, and completes only after the parent accepts it. A false response fails the Step, so the normal Step retry policy durably backs off and tries again. The same retry handles parent-start races: FlowAlreadyStarted fails SubmitStep, and its next execution invokes the RPC on the active parent. If the submitting process or Worker restarts, Dex preserves the retry.

This separates admission from processing: parent Flows remain bounded, while submissions remain durable and independently retryable.

Core implementation​

class SubmitStep(Step[SubmitRequestInput]):
    def __init__(
        self,
        client_provider: Callable[[], AsyncClient],
        parent_flow: AdvancedShortLiveParentFlow,
    ) -> None:
        self.client_provider = client_provider
        self.parent_flow = parent_flow

    async def execute(  # type: ignore[override]
        self, context: AsyncContext, input: SubmitRequestInput
    ) -> StepDecision:
        if not input.parent_ids:
            raise ValueError("at least one parent Flow ID is required")
        parent_id = input.parent_ids[partition(input.request, len(input.parent_ids))]
        accepted = await enqueue_request(
            self.client_provider(), self.parent_flow, parent_id, input.request
        )
        if not accepted:
            raise RuntimeError(f"parent {parent_id} rejected the request")
        return graceful_complete(parent_id)


async def enqueue_request(
    client: AsyncClient,
    parent_flow: AdvancedShortLiveParentFlow,
    parent_id: str,
    request: str,
) -> bool:
    try:
        return await client.invoke_rpc(parent_flow.send_request, parent_id, request)
    except FlowNotActiveError:
        await client.start_flow(
            parent_flow,
            parent_id,
            ParentInput([request], DEFAULT_CONCURRENCY),
            StartFlowOptions(id_reuse_policy=IdReusePolicy.ALLOW_IF_NOT_RUNNING),
        )
        return True


class SubmitRequestFlow(Flow[SubmitRequestInput]):
    def __init__(
        self,
        client_provider: Callable[[], AsyncClient],
        parent_flow: AdvancedShortLiveParentFlow,
    ) -> None:
        self.submit = SubmitStep(client_provider, parent_flow)

    def get_steps(self) -> StepList[SubmitRequestInput]:
        return StepList.start_step(self.submit)

Example: examples/python/dex_examples/patterns/parallel-subflows/submit_request_flow.py