External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(desktop): fall back to legacy thread paging

Browse Source 
Retry thread reads without bounds or ordering extensions when a relay omits
the requested bounds overlay. Preserve fail-closed handling for overlays that
are present but invalid, and pin 401-reply legacy convergence.

Co-authored-by: Carl <c7ebe626f000404285d3686e1dc74cc07cc60a9754a150041ba132e14bd3e2ec@buzz.block.builderlab.xyz>
Signed-off-by: Wes <wesbillman@users.noreply.github.com>
This commit is contained in:
Wes
2026-08-13 15:27:10 -06:00
co-authored by Carl
parent db381e8a70
commit 3712a7fa73
3 changed files with 220 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
desktop/src-tauri/src/commands/messages.rs
+30 -6
View File
@@ -264,10 +264,31 @@ pub async fn get_thread_replies(
cap,
thread_order.as_deref(),
cursor.as_ref(),
true,
);
let events = query_relay(&state, &[serde_json::Value::Object(filter)]).await?;
thread_page::parse(events, &root_event_id)
match thread_page::parse(events, &root_event_id) {
Ok(page) => Ok(page),
Err(thread_page::ParseError::MissingBounds) => {
// Relays predating the bounds extension ignore the opt-in fields.
// Retry their historical oldest-first contract so Desktop releases
// remain safe during partial rollout and relay rollback.
let legacy_filter = build_thread_replies_filter(
&root_event_id,
channel_id.as_deref(),
depth_limit.unwrap_or(64),
cap,
None,
cursor.as_ref(),
false,
);
let legacy_events =
query_relay(&state, &[serde_json::Value::Object(legacy_filter)]).await?;
Ok(thread_page::parse_legacy(legacy_events, cap))
}
Err(error) => Err(error.to_string()),
}
}
/// Build the relay `/query` filter for the server-side thread-subtree read.
@@ -292,6 +313,7 @@ fn build_thread_replies_filter(
cap: u32,
thread_order: Option<&str>,
cursor: Option<&crate::models::ThreadCursor>,
include_bounds: bool,
) -> serde_json::Map<String, serde_json::Value> {
let mut filter = serde_json::Map::new();
filter.insert("#e".to_string(), serde_json::json!([root_event_id]));
@@ -300,11 +322,13 @@ fn build_thread_replies_filter(
// defaults it to a deep-but-bounded value so nested replies aren't dropped.
filter.insert("depth_limit".to_string(), serde_json::json!(depth_limit));
filter.insert("limit".to_string(), serde_json::json!(cap));
// Opt into the relay's authoritative raw-scan bounds overlay. Older clients
// do not understand protocol events in this response and must not receive it.
filter.insert("thread_bounds".to_string(), serde_json::json!(true));
if matches!(thread_order, Some("newest")) {
filter.insert("thread_order".to_string(), serde_json::json!("newest"));
// Opt into authoritative raw-scan bounds only on the modern attempt.
// Legacy retries omit both extensions to preserve the old relay contract.
if include_bounds {
filter.insert("thread_bounds".to_string(), serde_json::json!(true));
if matches!(thread_order, Some("newest")) {
filter.insert("thread_order".to_string(), serde_json::json!("newest"));
}
}
if let Some(cid) = channel_id {
filter.insert("#h".to_string(), serde_json::json!([cid]));
desktop/src-tauri/src/commands/messages/thread_page.rs
+54 -7
View File
@@ -1,8 +1,26 @@
use nostr::Event;
use serde::Deserialize;
use std::fmt;
use crate::models::{ThreadCursor, ThreadRepliesResponse};
#[derive(Debug, PartialEq, Eq)]
pub(super) enum ParseError {
MissingBounds,
InvalidBounds(String),
}
impl fmt::Display for ParseError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::MissingBounds => {
formatter.write_str("Thread response omitted its bounds overlay")
}
Self::InvalidBounds(message) => formatter.write_str(message),
}
}
}
#[derive(Deserialize)]
struct BoundsCursor {
created_at: i64,
@@ -18,13 +36,15 @@ struct BoundsPayload {
pub(super) fn parse(
events: Vec<Event>,
root_event_id: &str,
) -> Result<ThreadRepliesResponse, String> {
) -> Result<ThreadRepliesResponse, ParseError> {
let mut replies = Vec::new();
let mut bounds = None;
for event in events {
if event.kind.as_u16() as u32 == buzz_core_pkg::kind::KIND_THREAD_BOUNDS {
if bounds.is_some() {
return Err("Thread response contained multiple bounds overlays".to_string());
return Err(ParseError::InvalidBounds(
"Thread response contained multiple bounds overlays".to_string(),
));
}
let matches_root = event.tags.iter().any(|tag| {
let values = tag.as_slice();
@@ -32,23 +52,28 @@ pub(super) fn parse(
&& values.get(1).map(String::as_str) == Some(root_event_id)
});
if !matches_root {
return Err("Thread bounds overlay does not match the requested root".to_string());
return Err(ParseError::InvalidBounds(
"Thread bounds overlay does not match the requested root".to_string(),
));
}
bounds = Some(
serde_json::from_str::<BoundsPayload>(&event.content)
.map_err(|error| format!("Invalid thread bounds overlay: {error}"))?,
serde_json::from_str::<BoundsPayload>(&event.content).map_err(|error| {
ParseError::InvalidBounds(format!("Invalid thread bounds overlay: {error}"))
})?,
);
} else if let Ok(value) = serde_json::to_value(&event) {
replies.push(value);
}
}
let bounds = bounds.ok_or_else(|| "Thread response omitted its bounds overlay".to_string())?;
let bounds = bounds.ok_or(ParseError::MissingBounds)?;
let next_cursor = bounds.next_cursor.map(|cursor| ThreadCursor {
created_at: cursor.created_at,
event_id: cursor.id,
});
if bounds.has_more != next_cursor.is_some() {
return Err("Thread bounds has_more and next_cursor disagree".to_string());
return Err(ParseError::InvalidBounds(
"Thread bounds has_more and next_cursor disagree".to_string(),
));
}
Ok(ThreadRepliesResponse {
events: replies,
@@ -56,3 +81,25 @@ pub(super) fn parse(
next_cursor,
})
}
pub(super) fn parse_legacy(events: Vec<Event>, cap: u32) -> ThreadRepliesResponse {
let has_more = events.len() as u32 >= cap;
let next_cursor = if has_more {
events.last().map(|event| ThreadCursor {
created_at: event.created_at.as_secs() as i64,
event_id: event.id.to_hex(),
})
} else {
None
};
let events = events
.iter()
.filter_map(|event| serde_json::to_value(event).ok())
.collect();
ThreadRepliesResponse {
events,
has_more,
next_cursor,
}
}
desktop/src-tauri/src/commands/messages_tests.rs
+136 -4
View File
@@ -1,16 +1,21 @@
use super::*;
use nostr::{EventBuilder, Kind, Tag};
fn thread_page_event(kind: u32, content: &str, root: Option<&str>) -> Event {
fn thread_page_event_at(kind: u32, content: &str, root: Option<&str>, created_at: u64) -> Event {
let tags = root
.map(|root| vec![Tag::parse(["e", root]).expect("valid root tag")])
.unwrap_or_default();
EventBuilder::new(Kind::Custom(kind as u16), content)
.tags(tags)
.custom_created_at(nostr::Timestamp::from(created_at))
.sign_with_keys(&Keys::generate())
.expect("test event should sign")
}
fn thread_page_event(kind: u32, content: &str, root: Option<&str>) -> Event {
thread_page_event_at(kind, content, root, 1_700_000_000)
}
fn bounds_event(root: &str, content: &str) -> Event {
thread_page_event(buzz_core_pkg::kind::KIND_THREAD_BOUNDS, content, Some(root))
}
@@ -161,7 +166,8 @@ fn thread_replies_filter_carries_non_p_gated_kinds_to_clear_the_gate() {
// and every kind MUST be non-p-gated (else the gate still fires). The
// Playwright mock does not model p-gating, so this unit test is the
// only guard against the client/relay auth contract drifting.
let filter = build_thread_replies_filter("root-hex", Some("channel-1"), 64, 200, None, None);
let filter =
build_thread_replies_filter("root-hex", Some("channel-1"), 64, 200, None, None, true);
let kinds = filter
.get("kinds")
@@ -191,8 +197,15 @@ fn thread_replies_filter_pages_with_composite_cursor() {
created_at: 1_700_000_000,
event_id: "abcd".to_string(),
};
let filter =
build_thread_replies_filter("root-hex", None, 64, 200, Some("newest"), Some(&cursor));
let filter = build_thread_replies_filter(
"root-hex",
None,
64,
200,
Some("newest"),
Some(&cursor),
true,
);
assert_eq!(filter["thread_order"], serde_json::json!("newest"));
assert_eq!(filter["thread_cursor"], serde_json::json!(1_700_000_000));
assert_eq!(filter["thread_cursor_id"], serde_json::json!("abcd"));
@@ -202,6 +215,120 @@ fn thread_replies_filter_pages_with_composite_cursor() {
);
}
#[test]
fn legacy_thread_filter_omits_modern_extensions_but_keeps_composite_cursor() {
let cursor = crate::models::ThreadCursor {
created_at: 1_700_000_000,
event_id: "abcd".to_string(),
};
let filter = build_thread_replies_filter(
"root-hex",
Some("channel-1"),
64,
200,
None,
Some(&cursor),
false,
);
assert!(!filter.contains_key("thread_bounds"));
assert!(!filter.contains_key("thread_order"));
assert_eq!(filter["thread_cursor"], serde_json::json!(1_700_000_000));
assert_eq!(filter["thread_cursor_id"], serde_json::json!("abcd"));
}
#[test]
fn legacy_thread_pages_converge_past_two_same_second_boundaries() {
let mut all_events: Vec<Event> = (0..401)
.map(|_| thread_page_event_at(9, "reply", Some("root"), 1_700_000_000))
.collect();
// Old relays order and keyset on (created_at ASC, event_id ASC).
all_events.sort_by_key(|event| (event.created_at.as_secs(), event.id.to_hex()));
let expected_ids: Vec<String> = all_events.iter().map(|event| event.id.to_hex()).collect();
let mut received_ids = Vec::new();
let mut cursor: Option<crate::models::ThreadCursor> = None;
loop {
let modern_filter = build_thread_replies_filter(
"root",
Some("channel-1"),
64,
200,
Some("newest"),
cursor.as_ref(),
true,
);
assert_eq!(modern_filter["thread_bounds"], serde_json::json!(true));
assert_eq!(modern_filter["thread_order"], serde_json::json!("newest"));
let select_old_relay_page = |request: &serde_json::Map<String, serde_json::Value>| {
let cursor_created_at = request
.get("thread_cursor")
.and_then(|value| value.as_i64());
let cursor_id = request
.get("thread_cursor_id")
.and_then(|value| value.as_str());
all_events
.iter()
.filter(|event| match (cursor_created_at, cursor_id) {
(Some(created_at), Some(event_id)) => {
(event.created_at.as_secs() as i64, event.id.to_hex())
> (created_at, event_id.to_string())
}
_ => true,
})
.take(200)
.cloned()
.collect::<Vec<_>>()
};
// An old relay ignores the modern extensions and omits the overlay.
let modern_response = select_old_relay_page(&modern_filter);
assert!(matches!(
thread_page::parse(modern_response, "root"),
Err(thread_page::ParseError::MissingBounds)
));
let legacy_filter = build_thread_replies_filter(
"root",
Some("channel-1"),
64,
200,
None,
cursor.as_ref(),
false,
);
assert!(!legacy_filter.contains_key("thread_bounds"));
assert!(!legacy_filter.contains_key("thread_order"));
if let Some(expected_cursor) = cursor.as_ref() {
assert_eq!(
legacy_filter["thread_cursor"],
serde_json::json!(expected_cursor.created_at)
);
assert_eq!(
legacy_filter["thread_cursor_id"],
serde_json::json!(expected_cursor.event_id)
);
}
let page = thread_page::parse_legacy(select_old_relay_page(&legacy_filter), 200);
received_ids.extend(
page.events
.iter()
.map(|event| event["id"].as_str().expect("event id").to_string()),
);
if !page.has_more {
assert!(page.next_cursor.is_none());
break;
}
cursor = page.next_cursor;
}
assert_eq!(received_ids, expected_ids);
let unique_ids: std::collections::HashSet<_> = received_ids.iter().collect();
assert_eq!(unique_ids.len(), 401);
}
#[test]
fn thread_page_parser_accepts_valid_empty_and_non_empty_pages() {
let empty = thread_page::parse(
@@ -243,6 +370,7 @@ fn thread_page_parser_rejects_missing_duplicate_and_wrong_root_bounds() {
assert!(thread_page::parse(vec![reply], "root")
.err()
.expect("missing bounds must fail closed")
.to_string()
.contains("omitted"));
let terminal = r#"{"has_more":false,"next_cursor":null}"#;
@@ -255,12 +383,14 @@ fn thread_page_parser_rejects_missing_duplicate_and_wrong_root_bounds() {
)
.err()
.expect("duplicate bounds must fail closed")
.to_string()
.contains("multiple"));
assert!(
thread_page::parse(vec![bounds_event("other-root", terminal)], "root")
.err()
.expect("mismatched root must fail closed")
.to_string()
.contains("does not match")
);
}
@@ -271,6 +401,7 @@ fn thread_page_parser_rejects_malformed_and_inconsistent_bounds() {
thread_page::parse(vec![bounds_event("root", "not-json")], "root")
.err()
.expect("malformed bounds must fail closed")
.to_string()
.contains("Invalid thread bounds")
);
@@ -282,6 +413,7 @@ fn thread_page_parser_rejects_malformed_and_inconsistent_bounds() {
thread_page::parse(vec![bounds_event("root", content)], "root")
.err()
.expect("cursor presence must agree with has_more")
.to_string()
.contains("disagree")
);
}