Долгие задачи, сбои и форма захвата

Это руководство выбирает, как таблица очереди держит задачу, пока над ней работает обработчик, и показывает, что происходит с задачей, которая выполняется минуты или процесс которой аварийно завершился.

Захваченная задача не должна попадать в другие захваты, пока не завершится, и должна вернуться, если её процесс остановился. Для этого описание таблицы выбирает одну из трёх форм, а обработчик во всех них один и тот же.

Три формы

	Блокировка строк	Аренда	Рекомендательная блокировка
Как выбирается	ничем: форма по умолчанию	#[field(locked_until)]	#[inbox(advisory_lock = "..")]
Чем держит задачу	транзакцией захвата, FOR UPDATE SKIP LOCKED	сроком, записанным в строку	блокировкой ключа строки, которую держит сеанс
Соединения	одно на задачу в работе, на всё время обработчика	одно на оператор	одно на задачу в работе
После сбоя	задача возвращается сразу	задача возвращается, когда истекает её аренда	задача возвращается сразу
Сбой тратит попытку	нет	да	да
Базы данных	Postgres, MySQL, MariaDB	все	все

Блокировка строк держит транзакцию захвата открытой, пока работает обработчик, а завершение её фиксирует. Транзакция аварийно завершившегося процесса откатывается, поэтому его задачи сразу возвращаются, а их попытки не засчитываются. Цена - соединение на каждую задачу в работе. В SQLite нет блокировок строк, поэтому таблица на SQLite указывает одну из двух других форм.

Аренда фиксирует захват сразу, поэтому задача в работе не держит ни соединения, ни транзакции. Она подходит для долгих задач, и это форма очереди на SQLite. Рекомендательная блокировка держит каждую задачу по ключу: ключ по id держит каждую задачу отдельно, а ключ по другому столбцу пускает задачи с одинаковым ключом в работу по одной.

Отчёт, который строится минуты

src/main.rs

const SCHEMA: &str = "CREATE TABLE IF NOT EXISTS report_jobs (
    id INTEGER PRIMARY KEY,
    attempt INTEGER NOT NULL DEFAULT 1,
    locked_until TEXT,
    payload BLOB NOT NULL
)";

Макросы

src/main.rs

use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::Sqlite;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `report_jobs`: `locked_until` claims it under a lease.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "report_jobs")]
struct RenderReport {
    #[field(id, generated)]
    id: i64,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// How long the row of a crashed process stays out of the queue.
const LEASE: Duration = Duration::from_mins(1);

#[subscriber(InboxQueue::<RenderReport>::new("reports").lease(LEASE))]
async fn render(report: &Report, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    if attempt > Some(1) {
        tracing::warn!(report.id, ?attempt, "rendering again after a lost attempt");
    }
    draw(report).await;
    tracing::info!(report.id, "report rendered");
    HandlerOutcome::ack()
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    RustStream::new(AppInfo::new("reports", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<RenderReport>("reports"), |b| {
            b.include(render);
        })
}

Вручную

src/main.rs

use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::Sqlite;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `report_jobs`: `.lease(..)` claims it under a lease.
#[derive(sqlx::FromRow)]
struct RenderReport {
    id: i64,
    attempt: i16,
    payload: Vec<u8>,
}

impl InboxTable for RenderReport {
    type Id = i64;
    type Table = InboxSpec<(Attempt, Lease<DateTime<Utc>>, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("report_jobs", Column::new("id").generated())
        .attempt(Column::new("attempt").generated())
        .lease(Column::new("locked_until"))
        .payload(Column::new("payload"));

    fn id(&self) -> &i64 {
        &self.id
    }
}

impl AttemptRow for RenderReport {
    type Attempt = i16;

    fn attempt(&self) -> &i16 {
        &self.attempt
    }
}

impl PayloadRow for RenderReport {
    type Column = Vec<u8>;

    fn payload(&self) -> &[u8] {
        &self.payload
    }
}

/// How long the row of a crashed process stays out of the queue.
const LEASE: Duration = Duration::from_mins(1);

struct Render;

impl Handle<Report, (), (), InboxContext> for Render {
    async fn handle(
        &self,
        report: &Report,
        _outs: &(),
        ctx: &mut Context<'_, InboxContext>,
    ) -> Result<(), HandlerOutcome> {
        let Ctx(attempt) = Ctx::<keys::Attempt>::from_context(ctx).await?;
        if attempt > Some(1) {
            tracing::warn!(report.id, ?attempt, "rendering again after a lost attempt");
        }
        draw(report).await;
        tracing::info!(report.id, "report rendered");
        Ok(())
    }
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    RustStream::new(AppInfo::new("reports", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<RenderReport>("reports"), |b| {
            let render = subscriber("reports", Render)
                .map_source(|_| InboxQueue::<RenderReport>::new("reports").lease(LEASE));
            b.include(render.build());
        })
}

Захват записывает срок аренды в locked_until и засчитывает попытку. Пока работает render, подписка продлевает аренду раз в половину её срока, поэтому отчёт может строиться пять минут при аренде в одну минуту. Срок служит токеном владения доставкой: подтверждение удаляет строку, только пока строка всё ещё держит эту аренду. InboxQueue::lease задаёт аренду одной подписки, а SqlxBroker::lease - аренду брокера; если её не задать, она равна тридцати секундам, в целых секундах.

Что делает сбой

Процесс, который аварийно завершился посреди отчёта, перестаёт продлевать аренду. Строка остаётся вне очереди, пока аренда не истечёт, здесь минуту, а затем её берёт следующий захват. Этот захват засчитывает ещё одну попытку, поэтому render видит вторую попытку и пишет её в журнал. Значит, аренда - это время, на которое сбой задерживает задачу: короткая аренда возвращает её раньше ценой большего числа продлений. Живой процесс, который перестал продлевать аренду, потому что дольше её срока не мог связаться с базой данных, теряет задачу: её берёт другой захват, а его запоздалое завершение возвращает ошибку SqlxBrokerError::LeaseLost.

Блокировка строк и рекомендательная блокировка заканчиваются вместе со своим сеансом, поэтому сбой сразу возвращает задачи. При завершении работы брокер перестаёт продлевать аренды: задача в работе сохраняет свою аренду и завершается, если успевает, а иначе возвращается, когда аренда истекает. Рекомендательные блокировки снимаются при завершении работы (Завершение работы).

Пул для каждой формы

Блокировка строк и рекомендательная блокировка держат соединение на каждую задачу в работе, поэтому подписке с workers(8) только для её задач нужно восемь соединений. Аренда держит соединение только на время каждого оператора. Каждая подписка оставляет в пуле одно соединение для обработчиков, их публикаций и завершений (Соединения). На Postgres рекомендательной блокировке нужно прямое соединение или PgBouncer в режиме сеанса, потому что блокировка принадлежит сеансу. На SQLite ключи задач в работе хранит процесс, поэтому один файл базы данных обслуживает один процесс. Как рассчитать пул, показывает руководство Работа в промышленной среде.

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

Тесты в файле примера запускают сервис на SQLite в памяти. Первый останавливает часы, поэтому пять минут построения отчёта проходят мгновенно, а продления аренды выполняются, когда часы доходят до них; отчёт доставляется один раз. Второй записывает две строки, которые захватил другой процесс: одну с истёкшей арендой и одну с действующей. Возвращается только первая. Эти строки не будят ни одну подписку, поэтому тест работает на реальных часах через TestApp::start_live и ждёт дольше интервала опроса.