Retryable Async Request in YTsaurus Flow (C++)

The pipeline is similar to Async Request, but it adds retry logic using timers and the internal YsonState.

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 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):

You save the request in the state with FailedAttempts = 0.
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 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

events → TStateKeeper → request (request generation).
request → TRequestProcessor → response (processing with retries).
response → TStateKeeper → state (result accumulation).

In the 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

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

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);
    }
}

Retryable Async Request in YTsaurus Flow (C++)

Difference from Async Request

Pipeline Components

TRequestProcessor

TStateKeeper

Retry Pattern

Pipeline Structure

Source Code

TRequestProcessor

TStateKeeper

See also