Возобновление и перемотка

Это руководство возобновляет работу сервиса с места, где остановился его прошлый запуск, и переносит работающую подписку в другое место файла.

Файл потока не хранит, докуда дочитал читатель. ack и nack на этом транспорте возвращают Unsupported (Подтверждение), поэтому перезапущенная подписка открывается там, где указывает её точка монтирования. Сервис, которому нельзя обработать заказ дважды, сам хранит свою контрольную точку.

Пример - сервис выставления счетов: он читает заказы магазина и выставляет счёт на каждый ровно один раз, в том числе между перезапусками.

Контрольная точка

Макросы

src/checkpoint.rs

/// The sequence of the last order the service invoiced, kept in a file of its own.
#[derive(Debug, Clone)]
struct Checkpoint {
    path: Arc<Path>,
}

impl Checkpoint {
    fn new(path: impl AsRef<Path>) -> Self {
        Self {
            path: path.as_ref().into(),
        }
    }

    /// The stored sequence; `None` until the first order is invoiced.
    fn load(&self) -> Option<u64> {
        fs::read_to_string(&self.path).ok()?.trim().parse().ok()
    }

    fn store(&self, sequence: u64) -> io::Result<()> {
        fs::write(&self.path, sequence.to_string())
    }
}

/// The app's state: each field reaches a handler as `State<FieldType>`.
#[derive(Debug, Clone, FromRef)]
struct Invoicing {
    checkpoint: Checkpoint,
}

Вручную

src/checkpoint.rs

/// The sequence of the last order the service invoiced, kept in a file of its own.
#[derive(Debug, Clone)]
struct Checkpoint {
    path: Arc<Path>,
}

impl Checkpoint {
    fn new(path: impl AsRef<Path>) -> Self {
        Self {
            path: path.as_ref().into(),
        }
    }

    /// The stored sequence; `None` until the first order is invoiced.
    fn load(&self) -> Option<u64> {
        fs::read_to_string(&self.path).ok()?.trim().parse().ok()
    }

    fn store(&self, sequence: u64) -> io::Result<()> {
        fs::write(&self.path, sequence.to_string())
    }
}

#[derive(Debug, Clone)]
struct Invoicing {
    checkpoint: Checkpoint,
}

Контрольная точка - это порядковый номер последнего заказа, на который сервис выставил счёт, в отдельном файле. Invoicing - состояние приложения, и обработчик берёт контрольную точку из него. Так же работает любое хранилище, которое уже есть у сервиса: строка в базе данных, ключ в кеше.

Возобновление с контрольной точки

Макросы

src/main.rs

#[subscriber(FileStream::new("orders"))]
async fn invoice(
    order: &Order,
    Ctx(at): Ctx<Position>,
    State(checkpoint): State<Checkpoint>,
) -> HandlerOutcome {
    let FilePosition::Sequence(sequence) = at else {
        return HandlerOutcome::drop();
    };
    // The run opens at the stored sequence, so the order the last run ended on comes again.
    if checkpoint.load().is_some_and(|stored| sequence <= stored) {
        return HandlerOutcome::drop();
    }
    tracing::info!(order.id, sequence, "invoiced");
    if let Err(error) = checkpoint.store(sequence) {
        tracing::error!(order.id, sequence, %error, "checkpoint not stored");
    }
    HandlerOutcome::ack()
}

fn invoicing(file: FileBroker, checkpoint: Checkpoint) -> impl App<State = Invoicing> {
    let start = checkpoint
        .load()
        .map_or(FilePosition::beginning(), FilePosition::sequence);
    RustStream::new(AppInfo::new("invoicing", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Invoicing { checkpoint }))
        .with_broker(file, |b| {
            b.include(invoice.start_at(start));
        })
}

fn main() -> ExitCode {
    run_main(|| {
        invoicing(
            FileBroker::new("/var/lib/shop/orders.ss"),
            Checkpoint::new("/var/lib/shop/invoicing.checkpoint"),
        )
    })
}

Вручную

src/main.rs

struct Invoice;

impl Handle<Order, (), (), FileContext, Invoicing> for Invoice {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, FileContext, Invoicing>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let FilePosition::Sequence(sequence) = ctx.context(Position) else {
            return ready(Err(HandlerOutcome::drop()));
        };
        let checkpoint = &ctx.state().checkpoint;
        // The run opens at the stored sequence, so the order the last run ended on comes again.
        if checkpoint.load().is_some_and(|stored| sequence <= stored) {
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(order.id, sequence, "invoiced");
        if let Err(error) = checkpoint.store(sequence) {
            tracing::error!(order.id, sequence, %error, "checkpoint not stored");
        }
        ready(Ok(()))
    }
}

fn invoicing(file: FileBroker, checkpoint: Checkpoint) -> impl App<State = Invoicing> {
    let start = checkpoint
        .load()
        .map_or(FilePosition::beginning(), FilePosition::sequence);
    RustStream::new(AppInfo::new("invoicing", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Invoicing { checkpoint }))
        .with_broker(file, |b| {
            b.include(
                subscriber(FileStream::new("orders"), Invoice)
                    .start_at(start)
                    .build(),
            );
        })
}

fn main() -> ExitCode {
    run_main(|| {
        invoicing(
            FileBroker::new("/var/lib/shop/orders.ss"),
            Checkpoint::new("/var/lib/shop/invoicing.checkpoint"),
        )
    })
}

invoicing читает контрольную точку до запуска сервиса и передаёт её в start_at в точке монтирования. У первого запуска контрольной точки нет, и он читает файл с начала. FilePosition::sequence(n) доставляет и само сообщение n, поэтому запуск, открытый на сохранённом номере, ещё раз получает заказ, на котором закончился прошлый. invoice читает порядковый номер доставки через Ctx<Position> и отбрасывает заказ с номером не больше контрольной точки. На каждый другой заказ выставляется счёт, и его номер становится новой контрольной точкой.

Контрольная точка записывается после счёта. Если запуск останавливается между ними, следующий запуск снова выставляет счёт на этот заказ, поэтому само выставление счёта должно допускать повтор. Запись контрольной точки до счёта даёт обратный компромисс: счёт на заказ никогда не выставляется дважды, но один заказ может потеряться.

FilePosition::timestamp(ms) открывает подписку на первом сообщении, записанном после заданного момента, в миллисекундах от начала эпохи Unix. Эта позиция подходит сервису, который восстанавливает своё представление данных с момента времени, а не с сохранённого номера.

Перемотка работающей подписки

Обработчик перематывает собственную подписку через Ctx<SeekHandle>. seek(FilePosition::end()) пропускает накопленное и продолжает с конца журнала. seek(FilePosition::sequence(n)) возвращается к сообщению n или перескакивает вперёд через ряд сообщений, о которых известно, что они не обрабатываются. Доставки, которые уже стоят в очереди со старого места, отбрасываются, и следующая доставка приходит с нового. Перемотка на порядковый номер дальше того, который файл запишет следующим, возвращает ошибку: живая подписка тогда заканчивается, а воспроизведение остаётся на месте. Пакетный обработчик получает тот же дескриптор через FileBatchContext.

Тестирование возобновления

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