恢复与定位

本指南让服务从上次运行停下的地方继续，并把正在运行的订阅移到文件中的另一个位置。

流文件不记录读取者读到了哪里。在这种传输上，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 在一个真实文件上运行服务两次：第二次运行在第一次结束时的订单处打开，并丢弃它。测试文件或管道服务讲解这两种模式。

检查点​

从检查点继续​

移动正在运行的订阅​

测试恢复​

检查点

从检查点继续

移动正在运行的订阅

测试恢复