★ FEATURED ARTICLE 1. 从buzz这个名字说起一个用Rust重写的Nostr中继到底在解决什么问题第一次看到buzz这个项目名加上关键词里的Nostr、relay、Rust、workspace我大概能猜到这是一个用Rust语言实现的Nostr中继服务。Nostr这个协议圈子里的人应该不陌生——它本质上是一套极简的社交协议核心就两样东西客户端负责签名和展示中继relay负责存储和转发事件。整个协议没有中心服务器没有账号体系你的身份就是一对密钥你发的内容就是一条条被签名的event。那为什么还要专门用Rust重写一个relay我实际跑过几个不同语言实现的中继之后体会很深。Nostr中继看起来简单——收event、存event、按订阅条件推event——但真正上线之后你会发现瓶颈根本不在业务逻辑而在连接管理、并发写入、内存占用和长时间运行的稳定性上。用Node.js或者Python写的中继几百个并发WebSocket连接加上高频事件写入内存曲线就开始往上飙GC一抖延迟就上来了。Rust在这个场景下的优势非常直接没有GC停顿、内存占用可控、异步运行时tokio处理海量长连接是它的主场。buzz这个项目名本身就带着一种嗡嗡作响、持续不断的意象很贴合中继服务那种7x24小时不停接收和广播事件的状态。而关键词里的workspace则暗示这个项目大概率采用了Cargo workspace的组织方式——也就是把relay核心、协议类型定义、存储层、配置管理等拆成多个crate放在同一个workspace里统一管理。这种结构在Rust项目里很常见好处是编译时可以并行、依赖可以共享、各模块边界清晰。这篇文章我打算从一个实际动手的角度把用Rust从零搭一个Nostr relay这件事拆开讲透。包括Nostr relay的核心协议行为到底是什么、Rust workspace该怎么划分、WebSocket连接和订阅管理怎么写、事件存储选型怎么权衡、以及我在实际调试过程中踩过的那些坑。不管你是刚入门Rust想找个真实项目练手还是已经在跑Nostr中继想换一套更稳的实现这篇内容应该都能给你一些可以直接抄作业的东西。2. Nostr relay的协议行为拆解别被简单两个字骗了2.1 一条event从进来到出去中间到底发生了什么很多人对Nostr relay的理解停留在就是个转发服务器但真正写起来你会发现relay要处理的事情比想象中多。客户端通过WebSocket连上来之后会发几种类型的消息最核心的是这三种EVENT客户端提交一条签名好的事件relay要验证签名、验证事件ID、检查是否重复、然后决定存不存、转发给哪些订阅者。REQ客户端发起一个订阅带一个subscription id和一组filter条件relay要记住这个订阅并且把历史匹配的事件推回去之后有新事件也要实时推。CLOSE客户端关闭某个订阅relay要把对应的订阅状态清理掉。看起来就三个动作但每个动作背后都有一堆细节。比如EVENT进来的时候你得先算事件的sha256得到id验证id和客户端声称的是否一致然后用schnorr签名验证event的sig字段再检查created_at是不是在合理范围内防止有人伪造未来时间的事件把订阅流刷爆还要看这个event的kind和tags是否命中某些relay的准入策略。我见过不少新手写的relay直接把收到的event往数据库一塞就完事结果跑一段时间发现数据库里全是重复事件、垃圾事件查询越来越慢。这就是没把协议行为吃透的后果。2.2 filter匹配逻辑Nostr relay最容易被写错的地方REQ里的filter是Nostr协议里最需要仔细实现的部分。一个filter可以包含这些字段ids、authors、kinds、since、until、limit以及各种#single-letter形式的tag过滤比如#e、#p。匹配规则是同一个filter内部不同字段之间是AND关系同一个字段的多个值是OR关系。举个例子{ authors: [pubkey_a, pubkey_b], kinds: [1, 7], since: 1700000000 }这条filter的意思是作者是a或b并且kind是1或7并且created_at大于等于1700000000。三个条件同时满足才算命中。而一个REQ可以带多个filter多个filter之间是OR关系——只要命中任意一个filter这条event就该推给这个订阅。我踩过的坑就在这里早期实现的时候我把多个filter的匹配写成了必须全部命中结果客户端订阅了我关注的人发的帖子和提到我的消息两组filter实际只收到了同时满足两组的event几乎什么都收不到。这个bug非常隐蔽因为单filter测试完全正常只有多filter场景才暴露。2.3 订阅的生命周期与背压问题订阅不是一次性的。客户端发一个REQ之后这个订阅会一直活着直到客户端发CLOSE或者连接断开。这意味着relay要为每个连接维护一个订阅表每个订阅对应一组filter。这里有个很容易被忽略的问题背压backpressure。如果某个客户端网络很慢relay还在拼命往它的WebSocket写事件写缓冲区就会堆积最终吃光内存。正确的做法是给每个连接的发送队列设一个上限超过就丢弃或者断开这个连接。Rust的tokio里可以用mpsc::channel配合一个有界队列来实现写满之后要么await等待要么直接drop。提示NIP-11relay信息文档里其实建议relay声明自己的限制比如最大订阅数、最大filter数、最大事件大小。这些限制不是摆设是保护relay不被单个客户端拖垮的关键。3. 用Cargo workspace组织项目模块边界怎么划才不别扭3.1 为什么这个项目适合用workspace而不是单crate一个Nostr relay如果全塞进一个crate里代码量上去之后会非常难受。协议类型定义、存储抽象、WebSocket处理、配置解析、CLI入口这些东西的依赖和编译特性都不一样。比如协议类型那部分你可能希望它是纯数据、不依赖tokio存储层你可能想同时支持内存和SQLite两种后端WebSocket层则强依赖tokio和网络库。Cargo workspace的价值就在这里每个crate独立编译、独立测试依赖关系显式声明。我一般会这样划分crate名称职责关键依赖buzz-core协议类型、event验证、filter匹配serde, secp256k1, sha2buzz-store存储抽象trait 内存/SQLite实现rusqlite或sqlx, async-traitbuzz-relayWebSocket服务、连接与订阅管理tokio, tokio-tungstenitebuzz-config配置结构体与加载serde, tomlbuzz-bin可执行入口组装以上模块以上全部这样划分之后buzz-core可以单独写单元测试不需要起网络buzz-store可以针对trait写两套实现测试时用内存版生产用SQLite版buzz-relay只关心连接和消息路由不关心数据存哪。3.2 workspace的Cargo.toml怎么写才干净根目录的Cargo.toml用[workspace]声明成员同时可以用[workspace.dependencies]统一管理依赖版本避免各个crate里版本号打架[workspace] members [buzz-core, buzz-store, buzz-relay, buzz-config, buzz-bin] resolver 2 [workspace.dependencies] tokio { version 1, features [full] } serde { version 1, features [derive] } serde_json 1 secp256k1 { version 0.29, features [rand, global-context] } sha2 0.10然后在子crate里这样引用[dependencies] tokio { workspace true } serde { workspace true }这个做法我强烈推荐。以前我在多个crate里各自写tokio版本结果一个用1.35一个用1.38编译时Cargo会拉两份tokio进来编译时间翻倍不说还可能因为feature不一致出现诡异问题。统一到workspace级别之后整个项目只有一份依赖图。3.3 编译缓存与增量构建的实操心得workspace有个很实际的好处target目录是共享的。你在根目录跑cargo build所有crate的产物都放在同一个target下切换crate编译时不会重复编译公共依赖。但这里有个坑如果你在workspace根目录跑cargo build它会编译所有成员如果你只想编译某个crate用cargo build -p buzz-relay。我刚开始用workspace的时候每次改一行relay代码都跑全量build等得想砸键盘。后来改成-p指定包配合cargo check先做类型检查开发效率高了一大截。另外cargo check比cargo build快很多因为它不生成最终二进制只做类型检查。日常开发我基本是cargo check -p buzz-relay循环只有要实际跑起来的时候才cargo run -p buzz-bin。4. WebSocket连接与订阅管理tokio下的并发模型设计4.1 每个连接一个task还是全局事件循环这是设计relay时第一个要做的架构决策。两种常见方案方案A每个连接spawn一个tokio task。每个WebSocket连接独立处理读写订阅状态存在这个task的局部变量里。新事件到来时通过一个全局的broadcast channel通知所有连接task各task自己判断是否命中订阅。方案B全局单事件循环。所有连接注册到一个中心结构里事件到来时由中心统一分发。我实际两种都写过最后倾向方案A。原因是方案A的并发模型更简单每个连接的状态是局部的不需要锁连接断开时task自然结束资源回收干净。方案B虽然分发逻辑集中但中心结构会成为锁竞争的热点而且连接状态和中心状态分离容易出bug。方案A的关键是用tokio::sync::broadcast做事件广播。relay收到一条合法event后往broadcast channel发一份所有连接task都订阅了这个channel各自收到后用自己的订阅filter判断要不要推给客户端。let (tx, _rx) tokio::sync::broadcast::channel::ArcEvent(1024); // 每个连接task里 let mut rx tx.subscribe(); loop { tokio::select! { Some(msg) ws_stream.next() { /* 处理客户端消息 */ } Ok(event) rx.recv() { if matches_any_subscription(event, subscriptions) { ws_stream.send(to_client_msg(event)).await?; } } } }4.2 broadcast channel的容量选择与lag处理broadcast::channel的容量参数很关键。设太小慢消费者会收到RecvError::Lagged也就是它跟不上了中间的事件被丢弃。设太大内存占用高。我的经验值是1024到4096之间。Nostr的事件流在热门relay上可能每秒几十到几百条1024的缓冲能扛住短时突发。收到Lagged的时候不要直接panic正确做法是记录日志然后让这个连接重新拉一次历史或者干脆断开让它重连。因为一旦lag了这个连接的事件流就出现空洞了继续推反而会让客户端状态不一致。match rx.recv().await { Ok(event) { /* 正常推送 */ } Err(broadcast::error::RecvError::Lagged(n)) { tracing::warn!(connection lagged, dropped {} events, n); // 可选触发重新同步或断开 } Err(broadcast::error::RecvError::Closed) break, }4.3 订阅表的并发访问什么时候需要锁在方案A里订阅表是连接task的局部变量读写都在同一个task里天然不需要锁。但有个例外如果relay需要支持服务端主动关闭某订阅或者统计每个订阅的推送量就可能需要跨task访问。我的建议是尽量把订阅表保持在连接task内部跨task的需求通过消息传递解决而不是共享内存加锁。比如要统计推送量连接task自己维护一个计数器定期通过一个mpsc channel上报给统计模块。这样整个系统里几乎没有共享可变状态调试起来轻松很多。注意Rust的借用检查器在写这种并发代码时其实是帮手。如果你发现某段代码非要加ArcMutex才能编译过往往说明你的状态划分有问题值得停下来重新想想。5. 事件存储选型内存、SQLite还是别的5.1 先想清楚relay的定位再选存储存储选型没有标准答案取决于你这个relay是干什么的。我把它分成三类个人relay只服务自己事件量小内存存储完全够用重启丢数据也无所谓。社区relay服务一小群人事件量中等需要持久化SQLite是甜点区。公共relay开放注册事件量大需要正经的数据库PostgreSQL之类。buzz这种项目如果目标是做一个可长期运行的中继SQLite是很务实的选择。它单文件、零运维、支持并发读写入用WAL模式之后性能也够看。关键是Rust生态里rusqlite和sqlx都很成熟。5.2 存储抽象trait的设计为了让存储可替换我会在buzz-store里定义一个trait#[async_trait] pub trait EventStore: Send Sync { async fn insert(self, event: Event) - Resultbool; // 返回是否新插入 async fn query(self, filters: [Filter]) - ResultVecEvent; async fn delete(self, id: EventId) - Result(); async fn count(self) - Resultu64; }insert返回bool表示是否是新的去重这个设计很重要因为relay收到重复事件时不应该重复广播。内存实现用一个RwLockHashMapEventId, Event加一个按时间排序的索引就够了。SQLite实现则要注意建索引created_at、kind、pubkey这几个字段经常出现在filter里不建索引查询会全表扫描。CREATE TABLE events ( id TEXT PRIMARY KEY, pubkey TEXT NOT NULL, created_at INTEGER NOT NULL, kind INTEGER NOT NULL, tags TEXT NOT NULL, content TEXT NOT NULL, sig TEXT NOT NULL ); CREATE INDEX idx_events_created_at ON events(created_at); CREATE INDEX idx_events_pubkey ON events(pubkey); CREATE INDEX idx_events_kind ON events(kind);5.3 tag过滤的存储难题Nostr的tag过滤#e、#p等是存储层最头疼的部分。因为tags是变长的数组存在一个字段里查询时要么全表扫描解析JSON要么单独建一张tag表。我推荐单独建表CREATE TABLE event_tags ( event_id TEXT NOT NULL, tag_name TEXT NOT NULL, tag_value TEXT NOT NULL ); CREATE INDEX idx_tags_lookup ON event_tags(tag_name, tag_value);查询#p过滤时先在这张表里查出匹配的event_id集合再和主表的其他条件做交集。这样虽然多一次join但比解析JSON快得多。实测在十万级事件量下tag查询从秒级降到毫秒级。提示写入event和写入tags要放在同一个事务里否则可能出现event写进去了但tags没写全的情况导致这条event永远查不到。6. 调试与压测中暴露的真实问题6.1 签名验证的性能陷阱secp256k1的schnorr签名验证是CPU密集操作单次大概几十到几百微秒。看起来不多但如果relay每秒收到上千条event光验证签名就能吃满一个核。我实测下来用secp256k1crate的global-context特性可以复用上下文比每次新建上下文快不少。另外可以考虑把签名验证放到tokio::task::spawn_blocking里避免阻塞异步运行时。但要注意spawn_blocking有线程池开销事件量不大的时候反而更慢需要根据实际负载权衡。还有一个优化点先验证event idsha256很快id不对直接拒绝根本不用走到签名验证。这样能挡掉大量格式错误或恶意构造的事件。6.2 重复事件的判定与竞态去重看起来简单——查一下id存不存在就行。但在并发场景下两个连接同时提交同一条event两个task同时查数据库都发现不存在然后都去插入就会触发主键冲突。处理方式有两种一是依赖数据库的主键约束插入失败就当作重复二是在内存里维护一个近期event id的LRU集合先查内存再查库。我倾向两者结合内存LRU挡掉绝大部分重复数据库约束兜底。// 插入时用 INSERT OR IGNORE根据 affected rows 判断是否新插入 let affected conn.execute( INSERT OR IGNORE INTO events (...) VALUES (...), params![...], )?; Ok(affected 0)6.3 连接数暴涨时的文件描述符耗尽这个坑我踩得很惨。压测的时候模拟几千个并发连接跑一会儿relay就报too many open files。原因是Linux默认的fd上限是1024每个WebSocket连接占一个fd加上数据库连接、日志文件等很快就到顶了。解决办法是调整系统的ulimit -n生产环境一般设到65535甚至更高。同时relay本身也要设一个最大连接数限制超过就拒绝新连接而不是硬扛到崩溃。ulimit -n 65535另外tokio的运行时线程数也要注意。默认它按CPU核数起worker线程连接数极高的时候可以考虑适当调大但不要盲目因为线程切换本身也有成本。6.4 日志与可观测性别等出问题才想起来relay这种长跑服务日志是命根子。我用tracingtracing-subscriber按级别输出生产环境默认info需要排查时临时开debug。关键要记录的东西连接建立与断开带连接id、订阅创建与关闭、事件接收与拒绝带拒绝原因、广播lag事件、存储错误。这些日志在排查为什么某条消息没收到这类问题时是决定性的。tracing::info!(conn_id %id, connection established); tracing::debug!(conn_id %id, sub_id %sub, subscription created); tracing::warn!(event_id %eid, reason invalid sig, event rejected);7. 从能跑到好用几个值得投入的优化方向7.1 NIP-11 relay信息文档客户端连上relay之前通常会先发一个HTTP GET请求带Accept: application/nostrjson期望拿到relay的元信息。这个就是NIP-11。返回内容包括relay名称、描述、管理员pubkey、支持的NIP列表、限制参数等。实现起来不难就是在WebSocket升级之前拦截HTTP请求判断Accept头返回一段JSON。但它的价值很大客户端能据此知道你的relay支持哪些功能、有什么限制用户体验直接上一个台阶。async fn handle_request(req: RequestBody) - ResultResponseBody { if req.headers().get(accept).map(|v| v.as_bytes()) Some(bapplication/nostrjson) { let info RelayInfo { name: buzz, /* ... */ }; return Ok(Response::new(Body::from(serde_json::to_string(info)?))); } // 否则走WebSocket升级 }7.2 限制策略把保护做在前面一个开放的relay如果不设限制很容易被单个客户端拖垮。我建议至少设这几项限制项建议值说明最大连接数按机器配置超过直接拒绝单连接最大订阅数20防止订阅表爆炸单filter最大limit5000防止一次拉全库最大事件大小64KB防止超大content单IP连接数10防止单点刷连接这些限制通过NIP-11暴露出去客户端会自觉遵守。不遵守的客户端直接拒绝不用客气。7.3 优雅关闭别让数据丢在最后一秒relay重启的时候如果直接kill进程正在写入的事件可能丢正在推送的消息可能断。正确做法是监听SIGTERM收到后停止接受新连接等待现有写入完成再退出。tokio里可以用tokio::signal::unix::signal监听信号配合一个CancellationToken通知所有task收尾。这个功能平时看不出价值但每次部署更新的时候都会感谢自己写了它。let mut sigterm signal(SignalKind::terminate())?; tokio::select! { _ sigterm.recv() { tracing::info!(shutting down gracefully); cancel.cancel(); // 等待task结束 } }8. 我在这个项目里最想分享的几条经验写到这里把几个我认为最值钱的经验单独拎出来说说。第一协议实现一定要对着NIP逐条核对。Nostr的NIP文档写得很细但很多人只看个大概就动手结果filter匹配、event验证这些地方全是偏差。我建议把相关的NIP打印出来实现一个功能就勾掉一条别嫌麻烦。第二Rust的workspace不是越大越好。我一开始恨不得把每个小功能都拆成crate结果crate之间循环依赖、trait定义到处飞反而更难维护。后来收敛到五六个crate边界清晰编译也快。拆分的标准应该是这个模块会不会被独立测试或独立替换而不是它看起来像不像一个独立的东西。第三压测要趁早。我第一版relay功能全对单连接测试完美结果一上并发就各种问题。后来养成习惯功能写完先跑一轮简单压测比如用脚本模拟几百个连接持续发事件很多并发bug在早期就暴露了修起来成本低得多。第四存储层的去重和索引是性能命门。relay跑久了慢十有八九是存储层的问题。去重没做好数据库里全是重复数据索引没建对查询全表扫描。这两件事在项目初期就要做对后期补代价很大。最后说个小的cargo clippy和cargo fmt一定要配到CI里。Rust的clippy能揪出很多你意识不到的写法问题比如不必要的clone、可以简化的match、潜在的性能陷阱。我每次提交前跑一遍clippy长期下来代码质量提升非常明显。这个习惯看起来不起眼但坚持半年之后回头看会庆幸自己做了。
阅读完成 · 觉得有帮助？
