---
metadata:
  - name: generator
    content: Diplodoc Platform v5.63.3
alternate:
  - https://ytsaurus.tech/docs/en/flow/cpp/examples/retryable_async_request.md
  - https://ytsaurus.tech/docs/ru/flow/cpp/examples/retryable_async_request.md
  - href: https://ytsaurus.tech/docs/en/flow/cpp/examples/retryable_async_request.md
    type: text/markdown
    title: Markdown version
  - href: https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65
    rel: describedby
---
> **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65

<!-- source: en/_includes/flow/cpp/examples/retryable_async_request.md -->
# Retryable Async Request in YTsaurus Flow (C++)

The [pipeline](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#pipeline) is similar to [Async Request](https://ytsaurus.tech/docs/en/flow/cpp/examples/async_request.md), but it adds retry logic using [timers](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#timer) and the internal `YsonState`.

[Source code](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/cpp/retryable_async_request)

## Difference from Async Request

The key difference is that `TRequestProcessor` is hosted by `TProcessFunctionComputation` rather than `TProcessFunctionSwiftMapComputation` because it needs to:
- Store the internal [state](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#state) to track the number of failed attempts.
- Use timers for retries at a specified interval.

## Pipeline Components

### TRequestProcessor

`TRequestProcessor` implements `IProcessFunction` and uses `TMutableStateKeyClient<TDelayedRequestState>` to store the internal state (Internal YsonState). You initialize the state in `Init(const IRuntimeInitContextPtr& initContext)` via `initContext->InitClient(RequestStateClient_, "request_state")`.

When you process an incoming message (`ProcessMessage`):
1. You save the request in the state with `FailedAttempts = 0`.
2. You call `TryRequest` to perform the attempt.

The `TryRequest` method contains the main retry logic:
- If the request fails (determined by `IsRequestSuccessful`), you increment the `FailedAttempts` counter and set a timer via `output->AddTimer(GetNextAttempt(context))`.
- If the request succeeds, you create a `TResponseMessage`, reset the state via `state.Clear()`, and send the response.

When the timer fires (`ProcessTimer`), you call `TryRequest` again with the current state.

### TStateKeeper

It’s fully similar to `TStateKeeper` from the [Async Request](https://ytsaurus.tech/docs/en/flow/cpp/examples/async_request.md) example: it accepts incoming events and responses and stores the accumulated result in the external state.

## Retry Pattern

The retry logic is based on the following elements:

- **TDelayedRequestState** — a descendant of `NYTree::TYsonStruct`, it stores `FailedAttempts` and the `Request` itself.
- **TMutableStateKeyClient** — a client for working with the internal YsonState. Unlike `TSimpleExternalStateManager`, the state is stored in Flow’s internal tables, not in an external user table.
- **Timers** — on a failed attempt, you set a timer with a `Delay` via `output->AddTimer(GetNextAttempt(context))`.
- **Constants** — `MaxRetries = 3` defines the simulated success period: a request succeeds immediately or after no more than two retries. `Delay = 5` defines the delay between attempts.

You calculate the time for the next attempt via `context->GetCurrentTimestamp()`, which ensures correct operation with Flow’s system time.

## Pipeline Structure

1. **events** → `TStateKeeper` → **request** (request generation).
2. **request** → `TRequestProcessor` → **response** (processing with retries).
3. **response** → `TStateKeeper` → state (result accumulation).

In the [spec](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#spec-and-dynamic-spec) for `TRequestProcessor`, you must register the `timer_streams` section to support retries. The process function class is specified in `processing_function`.

## Source Code

### TRequestProcessor

```cpp
void TRequestProcessor::Init(const IRuntimeInitContextPtr& initContext)
{
    initContext->InitClient<TDelayedRequestState>(RequestStateClient_, "request_state");
}

void TRequestProcessor::ProcessMessage(
    const TInputMessageConstPtr& message,
    const IOutputCollectorPtr& output,
    const IRuntimeContextPtr& context)
{
    auto request = context->ConvertToYsonMessage<TRequestMessage>(message);
    auto state = RequestStateClient_.GetState(message->Key);
    state->Request = request;
    state->FailedAttempts = 0;
    TryRequest(state, output, context);
}

void TRequestProcessor::ProcessTimer(
    const TInputTimerConstPtr& timer,
    const IOutputCollectorPtr& output,
    const IRuntimeContextPtr& context)
{
    auto state = RequestStateClient_.GetState(timer->Key);
    TryRequest(state, output, context);
}

TSystemTimestamp TRequestProcessor::GetNextAttempt(const IRuntimeContextPtr& context) const
{
    return TSystemTimestamp{context->GetCurrentTimestamp().Underlying() + Delay};
}

bool TRequestProcessor::IsRequestSuccessful(ui64 requestId, i64 failedAttempts) const
{
    return (requestId + failedAttempts) % MaxRetries == 0;
}

void TRequestProcessor::TryRequest(
    TStateAccessor<TDelayedRequestState>& state,
    const IOutputCollectorPtr& output,
    const IRuntimeContextPtr& context) const
{
    if (!IsRequestSuccessful(state->Request->RequestId, state->FailedAttempts)) {
        state->FailedAttempts += 1;
        output->AddTimer(GetNextAttempt(context));
        YT_TLOG_DEBUG("Failed request")
            .With("RequestId", state->Request->RequestId)
            .With("FailedAttempts", state->FailedAttempts);
        return;
    }

    auto response = New<TResponseMessage>();
    response->RequestId = state->Request->RequestId;
    response->Key = state->Request->Key;
    response->Length = std::ssize(state->Request->Request);
    YT_TLOG_DEBUG("Processed request")
        .With("RequestId", state->Request->RequestId)
        .With("FailedAttempts", state->FailedAttempts);
    state.Clear();
    output->AddMessage(context->ConvertToMessage(response));
}
```

### TStateKeeper

```cpp
void TStateKeeper::Init(const IRuntimeInitContextPtr& initContext)
{
    initContext->InitExternalStateClient(StateClient_, "/state");
}

void TStateKeeper::ProcessMessage(
    const TInputMessageConstPtr& message,
    const IOutputCollectorPtr& output,
    const IRuntimeContextPtr& context)
{
    if (message->StreamId == "event") {
        auto event = context->ConvertToYsonMessage<TEventMessage>(message);
        auto request = New<TRequestMessage>();
        request->RequestId = RandomNumber<ui64>();
        request->Key = event->Key;
        request->Request = event->Data;
        output->AddMessage(context->ConvertToMessage(request));
        YT_TLOG_DEBUG("Send request")
            .With("RequestId", request->RequestId);
    } else if (message->StreamId == "response") {
        auto response = context->ConvertToYsonMessage<TResponseMessage>(message);
        YT_TLOG_DEBUG("Received response")
            .With("RequestId", response->RequestId);
        auto state = StateClient_.GetState(message->Key);
        i64 totalLength = state->GetColumnValue<std::optional<i64>>("total_length").value_or(0);
        totalLength += response->Length;
        TPayloadBuilder builder(state->Schema);
        builder.Set(totalLength, "total_length");
        state->Payload = builder.Finish();
    } else {
        THROW_ERROR_EXCEPTION("Unexpected stream_id")
            .With("stream_id", message->StreamId);
    }
}
```
<!-- endsource: en/_includes/flow/cpp/examples/retryable_async_request.md -->

<!-- source: en/_includes/flow/cpp/examples/retryable_async_request_also.md -->
## See also

- [Quick Start (C++)](https://ytsaurus.tech/docs/en/flow/cpp/getting-started.md)
- [Timers](https://ytsaurus.tech/docs/en/flow/concepts/timers.md)
- [Example: Asynchronous Requests (C++)](https://ytsaurus.tech/docs/en/flow/cpp/examples/async_request.md)
<!-- endsource: en/_includes/flow/cpp/examples/retryable_async_request_also.md -->