События и сверка после перезапуска

Объяснение: как гейтвей отдает боту события, как бот догоняет пропущенное после обрыва и что он обязан сделать, если догнать нельзя. Если вы пишете бота, прочитайте страницу целиком: на ней весь порядок действий после сбоя. Как подключиться и в каком порядке вызывать: Подключить бота.

Что приходит в потоке

Бот открывает вызов Events и читает кадры четырех видов:

epoch - случайное число, новое при каждом запуске процесса гейтвея. seq - порядковый номер события в пределах этого запуска. Пара epoch и seq однозначно называет место в потоке. Бот хранит ее после каждого обработанного события.

Кольцо. Гейтвей держит в памяти последние события: не больше 10 000 штук и не старше 5 минут (настраивается в [gateway.ring], см. Конфиг). Из него и берется то, что бот пропустил, пока был отключен.

Сначала поток, потом подписки. Бот открывает Events, дожидается Ready и только потом подписывается. Кадр Ready означает, что с этого момента поток записан в гейтвее. Подпишись бот раньше, события между подпиской и открытием потока не увидел бы никто, и заявка, поставленная в эту щель, осталась бы без своих состояний.

Бот на Python этих кадров не видит. Пакет brokapi читает поток сам: Ready, номера, обрыв и Lagged он закрывает переоткрытием потока с последнего номера, а Gap отдает событием brokapi.Gap в обычном client.events. Торговля после него закрыта (Degraded) до reconcile. Подробно: Подключить бота.

Что получает бот после обрыва

Бот переподключается к Events с epoch и last_seq последнего обработанного события. Схема показывает, что гейтвей ответит.

flowchart TB
  open["бот открывает Events<br/>с epoch и last_seq"] --> same{"epoch<br/>тот же?"}
  same -->|"нет"| g1["Gap: NEW_EPOCH"]
  same -->|"да"| ring{"last_seq<br/>в кольце?"}
  ring -->|"старше кольца"| g2["Gap: TOO_OLD"]
  ring -->|"впереди гейтвея"| g3["Gap: AHEAD"]
  ring -->|"да"| ok["пропущенное из кольца,<br/>потом живые события"]
  g1 --> rec["Reconcile и чтение счета"]
  g2 --> rec
  g3 --> rec

Три ветки кончаются Gap, и во всех бот делает одно и то же: сверку. Единственная ветка без Gap - та, где сверка не нужна: бот получает пропущенные события по порядку, а за ними живые, без дыр. Без epoch в запросе поток только живой: ничего из прошлого бот не получит и не ждет.

После Gap поток идет с текущего места: номера в нем продолжаются, но события между last_seq и этим местом бот не увидит никогда. Поэтому бот не может считать, что знает состояние счета. Он делает Reconcile и заново читает то, что ему нужно (портфель, активные заявки).

Что случилось и что делать

Перезапуск гейтвея и открытые заявки

Заявки, стоящие у брокера, перезапуск гейтвея не задевает. Что происходит:

1. Перед отправкой каждой заявки гейтвей записывает ее cid в журнал на диске (journal.jsonl в каталоге гейтвея).
2. После старта гейтвей сначала сверяет журнал с брокером и только потом принимает команды. До этого Health отвечает starting, остальные вызовы (кроме Health и Caps) получают UNAVAILABLE с причиной GATEWAY_STARTING. Если сверка не вышла (лимит, сеть), она повторяется с паузой от полсекунды до 30 секунд, а последняя ошибка видна в Health.detail и в brokapi status <гейтвей>.
3. Заявку закрывает в журнале то, что точно знает об ее конце: окончательный отказ брокера, финальное событие заявки или сверка (closed_offline). Заявки, про которые неизвестно, чем они кончились, остаются открытыми.
4. Бот после Gap делает свой Reconcile со своим списком cid. Заявка, у которой не было ответа, находится там же: она в списке бота, и сверка скажет, стоит ли она у брокера.

Почему кольцо не записывается на диск. Пока процесс гейтвея мертв, события брокера никто не принимает, и диск их не вернет. Кольцо на диске спасло бы только хвост между последним прочитанным ботом событием и падением, а сверку после перезапуска бот делает в любом случае: состояние счета берется у брокера, заявки держит журнал. Цена диска - запись на каждое событие. Поэтому после перезапуска процесса бот получает Gap с причиной NEW_EPOCH и сверяется.

Чего сверка не возвращает. Исполнения заявок, которые прошли, пока гейтвея не было, сверка называет closed_offline и не пересказывает. Состояние такой заявки спросите вызовом OrderState, а сделки - вызовом Executions за интервал простоя.

Как это выглядит в коде

Бот на Rust: открыть поток, подписаться, при Gap подписаться заново и сверить, при Lagged переподключиться с места остановки. Пример запускается как тест: crates/gateway/examples/bot.rs, подробности и остальной код - в Подключить бота.

/// Reads the events of the gateway until `done` is true, and does what the stream asks for on the way.
///
/// `resume` is where the previous stream stopped (`None` at the first start). `journal` is the bot's own list of the orders
/// it placed and has not seen end.
pub async fn follow(
    bot: &RemoteClient,
    mut resume: Option<Resume>,
    journal: &[ClientOrderId],
    mut on_event: impl FnMut(&Envelope) -> bool,
) -> Result<u32, Error> {
    let mut gaps = 0;
    loop {
        // Open the stream first: from the moment `events` returns, the gateway keeps what happens for this bot.
        let mut stream = bot.events(resume, true).await?;
        // Then subscribe. Asking again is harmless, the gateway counts the request once for the bot.
        bot.subscribe_order_state().await?;
        while let Some(next) = stream.next().await {
            match next? {
                RemoteEvent::Event { envelope, .. } => {
                    if on_event(&envelope) {
                        return Ok(gaps);
                    }
                }
                RemoteEvent::Gap(gap) => {
                    // The gateway no longer has the events after our number. Whatever they said, the account is
                    // not what we last saw: ask the broker. A new epoch means a new process, which holds no subscriptions.
                    gaps += 1;
                    if gap.reason == pb::gap::Reason::NewEpoch as i32 {
                        bot.subscribe_order_state().await?;
                    }
                    bot.reconcile(journal).await?;
                }
                RemoteEvent::Lagged { .. } => break, // we read too slowly: reconnect from the resume point
            }
        }
        // The stream ended (the gateway restarted, the connection broke) or we lagged: continue where we stopped.
        resume = Some(stream.resume_point());
    }
}