Управление потоком и аренда

Это руководство ограничивает, сколько сообщений подписка держит одновременно, и сохраняет аренду медленного сообщения, пока сервис работает с ним или ждёт его.

Pub/Sub берёт в аренду каждое сообщение, которое доставляет через streaming pull. Пока аренда действует, подписка не доставляет сообщение никому другому. Клиент продлевает аренду в фоне, пока сообщение не завершено, но не дольше предела. Когда аренда истекает, сообщение уходит другому потребителю, даже если первый ещё работает с ним.

Пример - сервис отчётов: построение отчёта занимает время, а его данных иногда ещё нет.

Подписка

Макросы

src/main.rs

/// How long a report whose data is late waits before it is tried again.
const DATA_LATE: Duration = Duration::from_secs(20);

#[subscriber(
    GooglePubSub::new("reports-renderer")
        .create_with_topic("reports")
        .max_outstanding(10)
        .ack_extension(Duration::from_mins(1))
        .max_lease(Duration::from_mins(30))
)]
async fn render(request: &ReportRequest) -> HandlerOutcome {
    if !request.data_ready {
        tracing::info!(request.id, "report data late, report held");
        return HandlerOutcome::retry_after(DATA_LATE);
    }
    tracing::info!(request.id, "report rendered");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// How long a report whose data is late waits before it is tried again.
const DATA_LATE: Duration = Duration::from_secs(20);

struct Render;

impl Handle<ReportRequest> for Render {
    fn handle(
        &self,
        request: &ReportRequest,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if !request.data_ready {
            tracing::info!(request.id, "report data late, report held");
            return ready(Err(HandlerOutcome::retry_after(DATA_LATE)));
        }
        tracing::info!(request.id, "report rendered");
        ready(Ok(()))
    }
}

fn reports() -> GooglePubSub {
    GooglePubSub::new("reports-renderer")
        .create_with_topic("reports")
        .max_outstanding(10)
        .ack_extension(Duration::from_mins(1))
        .max_lease(Duration::from_mins(30))
}

Все три настройки - шаги дескриптора GooglePubSub.

max_outstanding(10) ограничивает число незавершённых сообщений, которые сервис держит одновременно. На пределе Pub/Sub перестаёт отправлять сообщения и продолжает по мере их завершения. Сообщение, которое держит retry_after, учитывается в пределе, пока не вернётся. По умолчанию в клиенте 1000. Небольшой предел не даёт медленной работе скапливаться в одной реплике, пока другие простаивают.

ack_extension - насколько продлевается аренда за одно продление, здесь на минуту. Клиент держит это значение в пределах от 10 секунд до 10 минут. Долгое продление означает меньше обращений к Pub/Sub. Короткое быстрее передаёт сообщение другому потребителю, когда реплика упала.

max_lease - как долго клиент продлевает одну аренду, здесь 30 минут, а по умолчанию час. Это и самая долгая задержка, которую может запросить retry_after: задержка проходит под арендой.

Ожидание под арендой

render запрашивает отчёт, данные которого опаздывают, снова через 20 секунд с retry_after. Сервис держит сообщение на время задержки, а клиент продлевает его аренду, поэтому тем временем его не получает другой потребитель. Когда задержка истекает, сервис отправляет nack, и Pub/Sub доставляет сообщение снова. Задержка дольше max_lease пережила бы аренду, поэтому сервис отказывает в ней, и сообщение возвращается сразу.

Макросы

src/main.rs

/// The service on `broker`; `app` runs it against the local emulator.
fn service(broker: PubSubBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("reports", "0.1.0")).with_broker(broker, |b| {
        b.include(render);
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(PubSubBroker::new("my-project").emulator("localhost:8085"))
}

Вручную

src/main.rs

/// The service on `broker`; `app` runs it against the local emulator.
fn service(broker: PubSubBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("reports", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber(reports(), Render).build());
    })
}

fn app() -> impl App<State = ()> {
    service(PubSubBroker::new("my-project").emulator("localhost:8085"))
}

fn main() -> ExitCode {
    run_main(app)
}

Приложение строит service(broker), поэтому тест запускает тот же сервис внутри процесса и на эмуляторе.

Тестирование

Тест публикует отчёт, данные которого опаздывают, и проверяет, что он возвращается один раз, через 20 секунд. Внутри процесса задержка проходит на остановленных часах. У внутрипроцессного Pub/Sub нет аренды и управления потоком, поэтому тот же тест выполняется и на эмуляторе, где 20 секунд настоящие. Там ожидание вдвое дольше 10-секундного срока подтверждения подписки, а отчёт возвращается только один раз: его держала аренда.

docker compose up -d
PUBSUB_TEST_HOST=localhost:8085 cargo test -- --test-threads=1

Тест на эмуляторе читает его адрес из PUBSUB_TEST_HOST и пропускается, если переменная не задана. Как одно тело теста выполняется в обоих режимах, показывает руководство Тестирование сервиса на Pub/Sub.