Retryable Async Request в YTsaurus Flow (Go)

Расширение примера Async Request: обработчик запросов повторяет каждую неудачную попытку через таймеры. Запрос целиком сохраняется во внутреннем стейте и повторяется с фиксированной задержкой, пока внешний сервис на него не ответит.

Исходный код

Структура

Пайплайн состоит из трёх компьютейшенов:

injector — нативный сорс (TSwiftPassthroughOrderedSourceComputation), объявленный прямо в спеке: он читает очередь и публикует события в стрим event. Go-кода у него нет.
state (stateKeeper) — transform-компьютейшен, сгруппированный по key: на каждое событие он открывает запрос в стриме request, а пришедший в стриме response ответ прибавляет к суммарной длине ответов ключа во внешнем стейте /state.
processor (requestProcessor) — transform-компьютейшен, сгруппированный по request_id: он делает попытки обращения к внешнему сервису и публикует ответ в стрим response.

Два компьютейшена сгруппированы по разным ключам не случайно: processor партиционируется по идентификатору запроса, поэтому запросы одного ключа повторяются независимо друг от друга. Таймерный стрим delay объявлен в спеке с allow_timer_self_dependency = %true — компьютейшен ставит таймеры сам себе.

`main.go`

Точка входа: создание пайплайна и регистрация обоих компьютейшенов компаньона.

Исходный код: main.go


type eventMessage struct {
	flow.YSONMessage
	Key  uint64 `yson:"key"`
	Data string `yson:"data"`
}

type requestMessage struct {
	flow.YSONMessage
	RequestID uint64 `yson:"request_id"`
	Key       uint64 `yson:"key"`
	Request   string `yson:"request"`
}

type responseMessage struct {
	flow.YSONMessage
	RequestID uint64 `yson:"request_id"`
	Key       uint64 `yson:"key"`
	Length    int64  `yson:"length"`
}

func main() {
	pipeline := flow.NewPipeline()
	pipeline.AddStreams(
		flow.NewYSONStream[eventMessage](eventStreamID),
		flow.NewYSONStream[requestMessage](requestStreamID),
		flow.NewYSONStream[responseMessage](responseStreamID),
	)
	pipeline.Add(
		flow.NewRowComputation("state", &stateKeeper{}),
		flow.NewRowComputation("processor", &requestProcessor{}),
	)

	if err := pipeline.Run(); err != nil {
		fmt.Fprintf(os.Stderr, "retryable_async_request: %v\n", err)
		os.Exit(1)
	}
}

`state_keeper.go`

Маршрутизация стримов и накопление результатов во внешнем стейте: ветвление по msg.StreamID разводит событие и ответ по двум обработчикам.


type totalLengthState struct {
	TotalLength int64 `yson:"total_length"`
}

type stateKeeper struct{}

var _ flow.RowFunction = (*stateKeeper)(nil)

func (s *stateKeeper) OnMessage(
	ctx context.Context,
	rt flow.Runtime,
	msg flow.ExtendedMessage,
	out flow.OutputCollector,
) error {
	switch msg.StreamID {
	case eventStreamID:
		return s.sendRequest(rt, msg, out)
	case responseStreamID:
		return s.bookResponse(rt, msg)
	default:
		return fmt.Errorf("unhandled stream %q", msg.StreamID)
	}
}

func (s *stateKeeper) sendRequest(rt flow.Runtime, msg flow.ExtendedMessage, out flow.OutputCollector) error {
	var event eventMessage
	if err := msg.ConvertTo(&event); err != nil {
		return err
	}

	request := flow.NewYSONMessage[requestMessage](requestStreamID)
	request.RequestID = rand.Uint64()
	request.Key = event.Key
	request.Request = event.Data
	encoded, err := flow.ConvertFrom(rt, request)
	if err != nil {
		return err
	}
	out.AddMessage(encoded)
	return nil
}

func (s *stateKeeper) bookResponse(rt flow.Runtime, msg flow.ExtendedMessage) error {
	var response responseMessage
	if err := msg.ConvertTo(&response); err != nil {
		return err
	}

	state, err := flow.OpenExternalState(rt, totalStateName, msg)
	if err != nil {
		return err
	}
	var total totalLengthState
	if _, err := state.ConvertTo(&total); err != nil {
		return err
	}
	total.TotalLength += response.Length
	return state.ConvertFrom(&total)
}

`request_processor.go`

Значение стейта — обычная Go-структура с YSON-тегами. В ней лежит всё, что нужно повтору, включая счётчик неудачных попыток, поэтому исходное сообщение повтору уже не требуется:


type requestState struct {
	RequestID      uint64 `yson:"request_id"`
	Key            uint64 `yson:"key"`
	Request        string `yson:"request"`
	FailedAttempts int64  `yson:"failed_attempts"`
}

Логика повторов: OnMessage открывает запрос и делает первую попытку, OnTimer читает запрос из стейта и повторяет её. Общий для обоих обработчиков метод attempt либо ставит таймер на следующую попытку, либо публикует ответ и очищает стейт:


type requestProcessor struct{}

var (
	_ flow.RowFunction      = (*requestProcessor)(nil)
	_ flow.RowTimerFunction = (*requestProcessor)(nil)
)

func (p *requestProcessor) OnMessage(
	ctx context.Context,
	rt flow.Runtime,
	msg flow.ExtendedMessage,
	out flow.OutputCollector,
) error {
	state, err := flow.OpenYSONState[requestState](rt, requestStateName, msg)
	if err != nil {
		return err
	}

	var input requestMessage
	if err := msg.ConvertTo(&input); err != nil {
		return err
	}
	request := requestState{
		RequestID: input.RequestID,
		Key:       input.Key,
		Request:   input.Request,
	}
	return p.attempt(rt, state, request, out)
}

func (p *requestProcessor) OnTimer(
	ctx context.Context,
	rt flow.Runtime,
	timer flow.Timer,
	out flow.OutputCollector,
) error {
	state, err := flow.OpenYSONState[requestState](rt, requestStateName, timer)
	if err != nil {
		return err
	}

	if state.Empty() {
		return nil
	}
	return p.attempt(rt, state, *state.Value(), out)
}

func (p *requestProcessor) attempt(
	rt flow.Runtime,
	state *flow.YSONState[requestState],
	request requestState,
	out flow.OutputCollector,
) error {
	if !succeeds(request) {
		request.FailedAttempts++
		*state.Value() = request
		out.AddTimer(flow.TimerRequest{TriggerTimestamp: uint64(time.Now().Add(retryDelay).Unix())})
		return nil
	}

	response := flow.NewYSONMessage[responseMessage](responseStreamID)
	response.RequestID = request.RequestID
	response.Key = request.Key
	response.Length = int64(len(request.Request))
	encoded, err := flow.ConvertFrom(rt, response)
	if err != nil {
		return err
	}
	out.AddMessage(encoded)

	state.Clear()
	return nil
}

func succeeds(request requestState) bool {
	return (request.RequestID+uint64(request.FailedAttempts))%attemptsToSucceed == 0
}

Ключевые паттерны

Повторы через таймеры: out.AddTimer(flow.TimerRequest{TriggerTimestamp: ...}) откладывает следующую попытку на retryDelay, а OnTimer её выполняет. Это стандартный способ реализовать повторы в Flow без внешних очередей. Пустой StreamID в flow.TimerRequest означает единственный таймерный стрим компьютейшена.
Счётчик попыток живёт в стейте вместе с данными запроса, поэтому переживает рестарт воркера: после перезапуска повтор продолжается с того же места.
Вход один раз преобразуется в requestMessage через msg.ConvertTo(&input), а retry-логика работает с отдельной структурой requestState, содержащей счётчик неудачных попыток.
Один и тот же стейт открывается и по сообщению, и по таймеру: flow.OpenYSONState[requestState](rt, requestStateName, msg) и flow.OpenYSONState[requestState](rt, requestStateName, timer) — оба входа несут ключ группировки, по которому стейт и адресуется.
Очистка стейта при успехе: после state.Clear() сработавший позже таймер видит state.Empty() и ничего не делает — устаревшее срабатывание безвредно.
Разделение ответственности: stateKeeper ничего не знает о повторах, вся логика инкапсулирована в requestProcessor, поэтому стратегию повторов можно поменять, не трогая учёт результатов.
Детерминированная симуляция сбоев: succeeds(request) стоит на месте настоящего клиента внешнего сервиса — в реальном коде здесь был бы HTTP-вызов.

См. также

Async Request

Wait Click Join