Rust in Everything
基于 async-stream 0.3.6

哪里可以用

主线位置：主线的反面，什么时候不该用它。stream! 的惰性是运行时一段的性质：块只在消费者 poll 时执行，消费者丢掉流，执行就停。下面三处候选都要先过这一关：这件事该不该跟着消费者的生死走。

flowchart TB
  q1{"有人在实时看逐项结果吗"}
  q1 -->|没有| no["不用：一次返回报告或写库就够"]
  q1 -->|有| q2{"消费者走了，活还要不要干完"}
  q2 -->|要| ch["spawn 干活，通道或 watch 推进度"]
  q2 -->|不要| st["try_stream! 产出，交给 Sse"]

候选一：搜索增量重建的进度

reindex_incremental 收集文件和数据库两类文档，和上一版 manifest 做差分，写入索引，最后只返回一个 IncrementalReport { upserts, deletes, elapsed_ms }。它有两个调用方：

可以把它改成 try_stream!，每处理完一批 yield 一条进度，交给 axum 的 Sse 推给后台页面。下面这个独立的小程序把这件事做了一遍，用睡眠模拟分批重建，验证这条路在 async-stream 0.3.6 + axum 0.8.9 上实际是什么样子：

async-stream = "=0.3.6"
axum = "=0.8.9"
futures = "0.3"
tokio = { version = "=1.54.1", features = ["full"] }

use std::time::Duration;

use async_stream::try_stream;
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::{routing::get, Router};
use futures::Stream;

// 假装在分批重建索引：每批 50ms，第 `fail_at` 批失败
async fn index_batch(batch: u32, fail_at: Option<u32>) -> Result<usize, String> {
  tokio::time::sleep(Duration::from_millis(50)).await;
  if Some(batch) == fail_at {
    return Err(format!("batch {batch} failed"));
  }
  Ok(100)
}

fn progress(fail_at: Option<u32>) -> impl Stream<Item = Result<Event, String>> {
  try_stream! {
    let mut done = 0;
    for batch in 1..=3 {
      done += index_batch(batch, fail_at).await?;
      println!("  [server] batch {batch} indexed");
      yield Event::default().event("progress").data(format!("{batch}/3 docs={done}"));
    }
    yield Event::default().event("done").data(format!("docs={done}"));
  }
}

async fn ok() -> Sse<impl Stream<Item = Result<Event, String>>> {
  Sse::new(progress(None)).keep_alive(KeepAlive::default())
}

async fn fail() -> Sse<impl Stream<Item = Result<Event, String>>> {
  Sse::new(progress(Some(2))).keep_alive(KeepAlive::default())
}

#[tokio::main]
async fn main() {
  let app = Router::new().route("/reindex", get(ok)).route("/reindex-fail", get(fail));
  let listener = tokio::net::TcpListener::bind("127.0.0.1:38411").await.expect("bind");
  println!("listening on {}", listener.local_addr().expect("addr"));
  axum::serve(listener, app).await.expect("serve");
}

axum 0.8.9 的 Sse 要求流是 Stream<Item = Result<Event, E>> + Send + 'static、E: Into<BoxError>，不要求 Sync（SseBody 里自己包了一层 SyncWrapper），所以 0.3.6 里 AsyncStream 的 Sync 跟着 async 块走（poll_next）对这里没有影响。错误类型用 String，它能转成 BoxError。

服务只绑 127.0.0.1，用 curl 测三种情况。正常跑完：

$ curl -sS -N -i http://127.0.0.1:38411/reindex
HTTP/1.1 200 OK
content-type: text/event-stream
cache-control: no-cache
transfer-encoding: chunked
date: Sat, 31 Oct 2026 13:22:15 GMT

event: progress
data: 1/3 docs=100

event: progress
data: 2/3 docs=200

event: progress
data: 3/3 docs=300

event: done
data: docs=300

第 2 批失败，? 把 Err 交给 axum：

$ curl -sS -N http://127.0.0.1:38411/reindex-fail
event: progress
data: 1/3 docs=100

curl: (18) transfer closed with outstanding read data remaining

客户端在 75 毫秒时断开（--max-time 0.075），服务端再等 0.5 秒后看日志，三次结果相同：

curl: (28) Operation timed out after 82 milliseconds with 36 bytes received
-- server log
listening on 127.0.0.1:38411
  [server] batch 1 indexed

两个现象都直接来自前面几章：

  • 出错时客户端看不到错误。 ? 把 Err 作为最后一项交出（? 之后流就结束了），axum 拿到 Err 就中断响应体，curl 报传输未完成，batch 2 failed 这句话没有到达客户端。想让页面显示错误，就别用 ?，自己 match，yield 一个 event("error") 的事件，再 return。
  • 客户端一断，活就停了。 断开后 axum 丢掉响应体，AsyncStream 随之被 drop，第 2 批正在等的那次睡眠连同整个 async 块一起被丢掉，batch 2 indexed 再也没出现。这就是惰性：流不被 poll 就不执行（为什么需要它）。

第二个现象对搜索重建是个问题：管理员关掉页面，重建就停在半路。真实的 reindex_incremental 先改索引、最后才保存 manifest，停在中间会怎样，这里没有验证；但「关页面会中断重建」本身就不是想要的语义。如果真要推进度，形状应当是 spawn 一个任务干活、用 tokio::sync::watch 或通道发进度，SSE 只订阅进度，断开不影响任务。那正是为什么需要它里的「通道 + spawn」，不是 stream!。

现在不用：重建按 mtime 差分，结果里带 elapsed_ms，平时由定时器触发，没有人盯着进度看；手动触发的 search_reindex 也只要最后的报告。

候选二：论文解读逐篇生成

generate_day 取当天待生成的论文，逐篇调用模型，每篇之间等 ARXIV_INTERVAL（3 秒），结果写库，统计累加进 stats。逐篇产出结果在形状上正好是一个流：for 循环里每篇 yield 一个结果。

现在不用：它跑在单独的 papers-ingest 程序里，文件头注释写明由服务器的 cron 每天跑一次；结果写进数据库，页面读库就够了，没有一个在线的消费者要逐篇接收。单篇失败已经是「记下来、继续下一篇」（mark_failed 后进入下一轮循环），这层语义不需要流来表达。如果哪天要在后台页面上手动触发并实时看进度，它和候选一的问题一样：每篇要先抓全文、再调两次模型（generate_one），不应该因为关掉页面就停在半路，还是 spawn + 进度通道的形状。

候选三：全表扫描改分页流

搜索索引收集论坛主题时，collect_topics 用 .all(&db) 一次把所有已审核的主题读进内存，再逐个转成索引文档。可以改成 .paginate(&db, n).into_stream()，一页一页地读、一页一页地转，内存里同时只有一页。这正好用上 sea-orm 里那段 stream!（依赖树里的 async-stream）。

现在没有证据说该改：本卷不连服务器，没有查线上 topics 表的行数，也没有量过这次 .all() 的内存峰值。另外，collect_topics 的结果最后还是要整体交给差分（diff_for_reindex 拿的是完整的 Vec），只把读库这一步改成流，峰值省不下多少。要判断值不值得改，先要有一个数：线上行数，或者重建时的内存峰值。这个数本卷没有，所以下面不写这一条的阈值。

重估条件

「本站第一次出现流式 HTTP 响应」是一个可以直接检查的事实。在框架仓库根目录执行：

git grep -qE 'Sse::new|Body::from_stream|into_stream\(|ServerEvents' -- crates
echo $?

在 ceffc19 上退出码是 1（没有匹配），这一节的结论成立。退出码变成 0，说明有人加了 SSE、流式响应体、into_stream 或 dioxus 的 ServerEvents，那时回到这一节，看新代码属于上面哪一种形状。这条命令反向验证过：同一个模式对上面那个 SSE 示例的源文件执行，退出码是 0。

小结

  • 三处候选（搜索重建进度、论文逐篇生成、全表扫描）现在都不用 async-stream：前两处没有在线的消费者，结果写日志、写库就够；第三处没有行数或内存的数据，没有证据说该改。
  • 实测 try_stream! + axum 0.8.9 Sse：正常时逐条推送；? 出错时 axum 中断响应，客户端只看到 curl: (18)，看不到错误内容；客户端断开后流被 drop，第 2 批再也没执行。
  • 所以「进度推送」不一定是 stream! 的场景：活要干完的，用 spawn 干活、通道或 watch 推进度；活可以跟着连接一起停的，才用 try_stream! 直接交给 Sse。
  • 重估条件是一条 git grep，ceffc19 上退出码为 1，对示例源文件为 0；变成 0 时回来重看。

💬 关联讨论

发起讨论
加载中...