From 284469f4a82c619888f3f68c1368568386667b35 Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Mon, 14 Sep 2026 21:15:40 +0000 Subject: [PATCH 1/5] Roll channel and node reminders across 6-hourly buckets Clients backdate everything we send them by a week and prune anything they haven't heard about in two weeks, so a channel direction lives for seven days past the header of the last snapshot that wrote it. Since we never send value-identical re-broadcasts, an unchanging channel is only written by a reminder. Reminders were previously included only in the snapshot generated at 00:00 UTC every fifth day (or for scopes above 50 hours), i.e. in one three-hour window out of forty for any client syncing at least every two days. Clients that missed that window, or synced daily at a fixed hour outside it, lost every stable channel a week after their last reminder, and channels can never be re-announced to an incremental client afterwards. Even clients that caught every window only kept a channel that had mutated once and then gone quiet if a reminder happened to fall in the single day between the six-day eligibility age and the seven-day client lifetime. Instead, every channel and node is now assigned to one of 24 buckets (a mixed hash of the SCID, or the node id's x coordinate), one bucket becomes due every six hours, and a snapshot includes the reminders of every bucket that became due within its (last_sync, reference] window. Consecutive snapshots thus chain without gaps and any client that syncs at least once every six days is reminded of every stable channel once per six-day cycle, regardless of which snapshots it fetches. Clients further behind receive all buckets at once, which subsumes the old scope rule. The channel-side six-day streak-age gate (and the window-function query that computed it) is dropped: with a six-day rotation it recreated the timing lottery for freshly mutated channels, and skipping recently mutated channels saves few bytes since those mostly carry an incremental update anyway. Node reminders keep their eligibility semantics but only evaluate the 14-day announcement history for nodes whose bucket is due. Co-Authored-By: Claude Fable 5.1 --- README.md | 3 + src/config.rs | 17 +++- src/lookup.rs | 250 +++++++++++++++++++++------------------------- src/tests/mod.rs | 254 ++++++++++++++++++++++++++++++++++++++++++++++- 4 files changed, 383 insertions(+), 141 deletions(-) diff --git a/README.md b/README.md index eec22860..c88dfb6f 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,9 @@ Then, any channel that has only had an announcement but never an update is dropp every channel whose first update was seen after the given timestamp is collected alongside its announcement. +To avoid unchanging data being eventually pruned by clients, "reminder" updates are added in a +rolling fashion. + Finally, all channel update transitions are evaluated and collected into either a full or an incremental update. diff --git a/src/config.rs b/src/config.rs index 8195a31d..3f1fc897 100644 --- a/src/config.rs +++ b/src/config.rs @@ -25,11 +25,22 @@ pub(crate) const SCHEMA_VERSION: i32 = 17; pub(crate) const SYMLINK_GRANULARITY_INTERVAL: u32 = 3600 * 3; // three hours pub(crate) const MAX_SNAPSHOT_SCOPE: u32 = 3600 * 24 * 21; // three weeks // generate symlinks based on a 3-hour-granularity -/// If the last update in either direction was more than six days ago, we send a reminder -/// That reminder may be either in the form of a channel announcement, or in the form of empty -/// updates in both directions. + +/// Clients backdate everything we send them by a week and prune anything they haven't heard about +/// in two weeks, so from the client's perspective each channel direction lives for a bit under 7 +/// days past the last snapshot that wrote it. If the reference update a client would use to apply +/// an incremental update is older than this, it may have been pruned, and we send a full update +/// instead. The same age is used to decide whether a node announcement warrants a reminder. pub(crate) const CHANNEL_REMINDER_AGE: Duration = Duration::from_secs(6 * 24 * 60 * 60); +/// The reminders are sent in rolling fashion, with each channel and node assigned one of +/// [`REMINDER_BUCKET_COUNT`] buckets. One bucket becomes due every [`REMINDER_SLOT_INTERVAL`], so +/// every channel and node is reminded once per [`CHANNEL_REMINDER_AGE`]. +pub(crate) const REMINDER_SLOT_INTERVAL: Duration = Duration::from_secs(6 * 60 * 60); +pub(crate) const REMINDER_BUCKET_COUNT: u64 = CHANNEL_REMINDER_AGE.as_secs() / REMINDER_SLOT_INTERVAL.as_secs(); +const _: () = assert!(CHANNEL_REMINDER_AGE.as_secs() % REMINDER_SLOT_INTERVAL.as_secs() == 0); +const _: () = assert!(REMINDER_BUCKET_COUNT < 64, "reminder buckets are tracked in a u64 bitmask"); + /// The interval after which graph data gets pruned after it was first seen /// This should match the LDK default pruning interval, which is 14 days pub(crate) const PRUNE_INTERVAL: Duration = Duration::from_secs(14 * 24 * 60 * 60); diff --git a/src/lookup.rs b/src/lookup.rs index 93671ce1..9683c68d 100644 --- a/src/lookup.rs +++ b/src/lookup.rs @@ -11,7 +11,7 @@ use tokio_postgres::Client; use futures::StreamExt; use hex_conservative::DisplayHex; -use lightning::{log_debug, log_gossip, log_info}; +use lightning::{log_gossip, log_info}; use lightning::types::features::NodeFeatures; use lightning::util::logger::Logger; @@ -50,7 +50,8 @@ pub(super) struct ChannelDelta { pub(super) announcement: Option, pub(super) updates: (Option, Option), pub(super) first_bidirectional_updates_seen: Option, - /// The seen timestamp of the older of the two latest directional updates + /// Whether this channel's reminder bucket became due within the snapshot's window, in which + /// case each direction without a real update to send gets a flags-only reminder update pub(super) requires_reminder: bool, } @@ -103,22 +104,41 @@ impl Default for DirectedUpdateDelta { } } -fn should_snapshot_include_reminders(last_sync_timestamp: u32, current_timestamp: u64, logger: &L) -> bool where L::Target: Logger { - let current_hour = current_timestamp / 3600; - let current_day = current_timestamp / (24 * 3600); +/// The reminder buckets that became due within the window `(last_sync_timestamp, current_timestamp]`, +/// as a bitmask. +pub(super) fn reminder_buckets_due(last_sync_timestamp: u32, current_timestamp: u64) -> u64 { + assert!(config::REMINDER_BUCKET_COUNT < 64); - log_debug!(logger, "Current day index: {}", current_day); - log_debug!(logger, "Current hour: {}", current_hour); + // the first slot boundary strictly after the last sync, and the last one at or before now + let first_due_slot = (last_sync_timestamp as u64) / config::REMINDER_SLOT_INTERVAL.as_secs() + 1; + let last_due_slot = current_timestamp / config::REMINDER_SLOT_INTERVAL.as_secs(); + if last_due_slot < first_due_slot { + return 0; + } + if last_due_slot - first_due_slot + 1 >= config::REMINDER_BUCKET_COUNT { + return (1u64 << config::REMINDER_BUCKET_COUNT) - 1; + } - // every 5th day at midnight - let is_reminder_hour = (current_hour % 24) == 0; - let is_reminder_day = (current_day % 5) == 0; + let mut due_buckets = 0u64; + for slot in first_due_slot..=last_due_slot { + due_buckets |= 1u64 << (slot % config::REMINDER_BUCKET_COUNT); + } + due_buckets +} - let snapshot_scope = current_timestamp.saturating_sub(last_sync_timestamp as u64); - let is_reminder_scope = snapshot_scope > (50 * 3600); - log_debug!(logger, "Snapshot scope: {}s", snapshot_scope); +pub(super) fn channel_reminder_bucket_flag(short_channel_id: u64) -> u64 { + // splitmix64 finalizer + let mut mixed = short_channel_id; + mixed = (mixed ^ (mixed >> 30)).wrapping_mul(0xbf58476d1ce4e5b9); + mixed = (mixed ^ (mixed >> 27)).wrapping_mul(0x94d049bb133111eb); + mixed ^= mixed >> 31; + 1u64 << (mixed % config::REMINDER_BUCKET_COUNT) +} - (is_reminder_hour && is_reminder_day) || is_reminder_scope +/// Node ids are compressed public keys, whose x coordinate is already uniformly distributed. +pub(super) fn node_reminder_bucket_flag(node_id: &NodeId) -> u64 { + let x_coordinate_prefix: [u8; 8] = node_id.as_slice()[1..9].try_into().unwrap(); + 1u64 << (u64::from_be_bytes(x_coordinate_prefix) % config::REMINDER_BUCKET_COUNT) } /// Fetch all the channel announcements that are presently in the network graph, regardless of @@ -144,7 +164,8 @@ pub(super) async fn fetch_channel_announcements(delta_set: &mut DeltaS let current_timestamp = snapshot_reference_timestamp.unwrap_or(SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()); log_info!(logger, "Current timestamp: {}", current_timestamp); - let include_reminders = should_snapshot_include_reminders(last_sync_timestamp, current_timestamp, &logger); + let due_reminder_buckets = reminder_buckets_due(last_sync_timestamp, current_timestamp); + log_info!(logger, "Reminder buckets due in this snapshot: {:#b}", due_reminder_buckets); log_info!(logger, "Obtaining corresponding database entries"); let start = Instant::now(); @@ -226,112 +247,57 @@ pub(super) async fn fetch_channel_announcements(delta_set: &mut DeltaS log_info!(logger, "Fetched {} update rows of the first update in a new direction in {:?}", newer_oldest_directional_update_count, start.elapsed()); } - if include_reminders { - // THIS STEP IS USED TO DETERMINE IF A REMINDER UPDATE SHOULD BE SENT - - log_info!(logger, "Annotating channel announcements whose latest channel update in a given direction occurred more than six days ago"); - // Steps: - // — Obtain all updates, distinct by (scid, direction), ordered by seen DESC - // — From those updates, select distinct by (scid), ordered by seen ASC (to obtain the older one per direction) - let reminder_threshold_timestamp = current_timestamp.checked_sub(config::CHANNEL_REMINDER_AGE.as_secs()).unwrap() as f64; - - log_info!(logger, "Fetch first time we saw the current value combination for each direction (prior mutations excepted)"); - let reminder_lookup_threshold_timestamp = current_timestamp.checked_sub(config::PRUNE_INTERVAL.as_secs()).unwrap() as f64; - let params: [&(dyn tokio_postgres::types::ToSql + Sync); 2] = [&channel_ids, &reminder_lookup_threshold_timestamp]; - - /* - What exactly is the below query doing? - - First, the inner query groups all channel updates by their scid/direction combination, - and then sorts those in reverse chronological order by the "seen" column. - - Then, each row is annotated based on whether its subsequent row for the same scid/direction - combination has a different value for any one of these six fields: - disable, cltv_expiry_delta, htlc_minimum_msat, fee_base_msat, fee_proportional_millionths, htlc_maximum_msat - Those are simply the properties we use to keep track of channel mutations. - - The outer query takes all of those results and selects the first value that has a distinct - successor for each scid/direction combination. That yields the first instance at which - a given channel configuration was received after any prior mutations. - - Knowing that, we can check whether or not there have been any mutations within the - reminder requirement window. Because we only care about that window (and potentially the - 2-week-window), we pre-filter the scanned updates by only those that were received within - 3x the timeframe that we consider necessitates reminders. - */ - - let start = Instant::now(); - let mutated_updates = client.query_raw(" - SELECT DISTINCT ON (short_channel_id, direction) short_channel_id, direction, blob_signed, CAST(EXTRACT('epoch' from seen) AS BIGINT) AS seen FROM ( - SELECT short_channel_id, direction, timestamp, seen, blob_signed, COALESCE ( - disable<>lead(disable) OVER w1 - OR - cltv_expiry_delta<>lead(cltv_expiry_delta) OVER w1 - OR - htlc_minimum_msat<>lead(htlc_minimum_msat) OVER w1 - OR - fee_base_msat<>lead(fee_base_msat) OVER w1 - OR - fee_proportional_millionths<>lead(fee_proportional_millionths) OVER w1 - OR - htlc_maximum_msat<>lead(htlc_maximum_msat) OVER w1, - TRUE - ) has_distinct_successor - FROM channel_updates - WHERE short_channel_id = any($1) AND seen >= TO_TIMESTAMP($2) - WINDOW w1 AS (PARTITION BY short_channel_id, direction ORDER BY seen DESC) - ) _ - WHERE has_distinct_successor - ORDER BY short_channel_id ASC, direction ASC, timestamp DESC - ", params).await.unwrap(); - - let mut pinned_updates = Box::pin(mutated_updates); - let mut older_latest_directional_update_count = 0; - while let Some(row_res) = pinned_updates.next().await { - let current_row = row_res.unwrap(); - let seen = current_row.get::<_, i64>("seen") as u32; - - if seen < reminder_threshold_timestamp as u32 { - let blob: Vec = current_row.get("blob_signed"); - let mut readable = Cursor::new(&blob); - let unsigned_channel_update = ChannelUpdate::read(&mut readable).unwrap().contents; - - let scid = unsigned_channel_update.short_channel_id; - let direction: bool = current_row.get("direction"); - - let current_channel_delta = delta_set.entry(scid).or_insert(ChannelDelta::default()); - - // We might be able to get away with not using this - (*current_channel_delta).requires_reminder = true; - older_latest_directional_update_count += 1; + if due_reminder_buckets != 0 { + let read_only_graph = network_graph.read_only(); + let mut reminder_channel_count = 0; + for scid in channel_ids.iter().map(|scid| *scid as u64) { + if due_reminder_buckets & channel_reminder_bucket_flag(scid) == 0 { + continue; + } - if let Some(current_channel_info) = network_graph.read_only().channel(scid) { - if current_channel_info.one_to_two.is_none() || current_channel_info.two_to_one.is_none() { - // we don't send reminders if we don't have bidirectional update data + match delta_set.get(&scid) { + Some(current_channel_delta) => { + let is_new_announcement = current_channel_delta.announcement.as_ref() + .map(|announcement| announcement.seen >= last_sync_timestamp) + .unwrap_or(false); + if is_new_announcement || current_channel_delta.first_bidirectional_updates_seen.is_some() { + // the client is about to receive the announcement alongside full updates + // for this channel, so a reminder would be pointless continue; } - - if let Some(info) = current_channel_info.one_to_two.as_ref() { - let flags: u8 = if info.enabled { 0 } else { 2 }; - let current_update = (*current_channel_delta).updates.0.get_or_insert(DirectedUpdateDelta::default()); - current_update.serialization_update_flags = Some(flags); - } - - if let Some(info) = current_channel_info.two_to_one.as_ref() { - let flags: u8 = if info.enabled { 1 } else { 3 }; - let current_update = (*current_channel_delta).updates.1.get_or_insert(DirectedUpdateDelta::default()); - current_update.serialization_update_flags = Some(flags); - } - } else { - // we don't send reminders if we don't have the channel + }, + None => { + // we don't have the announcement in the database (yet), so this channel is + // going to be dropped from the delta anyway continue; } - - log_gossip!(logger, "Reminder requirement triggered by update for channel {} in direction {}", scid, direction); } + + // the graph may have changed since the channel ids were collected; we don't send + // reminders if we don't have bidirectional update data + let (one_to_two, two_to_one) = match read_only_graph.channel(scid) { + Some(channel_info) => match (channel_info.one_to_two.as_ref(), channel_info.two_to_one.as_ref()) { + (Some(one_to_two), Some(two_to_one)) => (one_to_two, two_to_one), + _ => continue, + }, + None => continue, + }; + + let current_channel_delta = delta_set.get_mut(&scid).unwrap(); + (*current_channel_delta).requires_reminder = true; + + let flags: u8 = if one_to_two.enabled { 0 } else { 2 }; + let current_update = (*current_channel_delta).updates.0.get_or_insert(DirectedUpdateDelta::default()); + current_update.serialization_update_flags = Some(flags); + + let flags: u8 = if two_to_one.enabled { 1 } else { 3 }; + let current_update = (*current_channel_delta).updates.1.get_or_insert(DirectedUpdateDelta::default()); + current_update.serialization_update_flags = Some(flags); + + log_gossip!(logger, "Reminder due for channel {}", scid); + reminder_channel_count += 1; } - log_info!(logger, "Fetched {} update rows of the latest update in the less recently updated direction in {:?}", - older_latest_directional_update_count, start.elapsed()); + log_info!(logger, "Annotated {} channels for reminders", reminder_channel_count); } } @@ -566,32 +532,44 @@ pub(super) async fn fetch_node_updates(network_graph: &Network let reminder_inclusion_threshold_timestamp = current_timestamp.checked_sub(config::CHANNEL_REMINDER_AGE.as_secs()).unwrap() as u32; let reminder_lookup_threshold_timestamp = current_timestamp.checked_sub(config::PRUNE_INTERVAL.as_secs()).unwrap() as u32; - // this is the timestamp we need to fetch all relevant updates - let include_reminders = should_snapshot_include_reminders(last_sync_timestamp, current_timestamp, &logger); - let effective_threshold_timestamp = if include_reminders { - std::cmp::min(last_sync_timestamp, reminder_lookup_threshold_timestamp) as f64 - } else { - // If we include reminders, the decision logic is as follows: - // If the pre-sync update was more than 6 days ago, serialize in full. - // Otherwise: - // If the last mutation occurred after the last sync, serialize the mutated properties. - // Otherwise: - // If the last mutation occurred more than 6 days ago, serialize as a reminder. - // Otherwise, don't serialize at all. - last_sync_timestamp as f64 - }; + // Nodes whose reminder bucket became due within this snapshot's window are candidates for a + // reminder. For those, the decision logic is as follows: + // If the pre-sync update was more than 6 days ago, serialize in full. + // Otherwise: + // If the last mutation occurred after the last sync, serialize the mutated properties. + // Otherwise: + // If the last mutation occurred more than 6 days ago, serialize as a reminder. + // Otherwise, don't serialize at all. + // Determining when the last mutation occurred requires looking at the announcements from the + // whole prune interval, rather than just since the last sync, for those nodes only. + let due_reminder_buckets = reminder_buckets_due(last_sync_timestamp, current_timestamp); + log_info!(logger, "Reminder buckets due in this snapshot: {:#b}", due_reminder_buckets); + let reminder_node_ids: Vec = delta_set.keys() + .filter(|node_id| due_reminder_buckets & node_reminder_bucket_flag(node_id) != 0) + .map(|node_id| node_id.as_slice().to_lower_hex_string()) + .collect(); + let reminder_lookup_threshold_timestamp_float = std::cmp::min(last_sync_timestamp, reminder_lookup_threshold_timestamp) as f64; // get all the intermediate node updates // (to calculate the set of mutated fields for snapshotting, where intermediate updates may // have been omitted) let start = Instant::now(); - let params: [&(dyn tokio_postgres::types::ToSql + Sync); 2] = [&node_ids, &effective_threshold_timestamp]; + let params: [&(dyn tokio_postgres::types::ToSql + Sync); 4] = [&node_ids, &last_sync_timestamp_float, &reminder_node_ids, &reminder_lookup_threshold_timestamp_float]; let intermediate_updates = client.query_raw(" - SELECT announcement_signed, CAST(EXTRACT('epoch' from seen) AS BIGINT) AS seen - FROM node_announcements - WHERE - public_key = ANY($1) AND - seen >= TO_TIMESTAMP($2) + SELECT announcement_signed, seen FROM ( + SELECT public_key, timestamp, announcement_signed, CAST(EXTRACT('epoch' from seen) AS BIGINT) AS seen + FROM node_announcements + WHERE + public_key = ANY($1) AND + seen >= TO_TIMESTAMP($2) + UNION ALL + SELECT public_key, timestamp, announcement_signed, CAST(EXTRACT('epoch' from seen) AS BIGINT) AS seen + FROM node_announcements + WHERE + public_key = ANY($3) AND + seen >= TO_TIMESTAMP($4) AND + seen < TO_TIMESTAMP($2) + ) _ ORDER BY public_key ASC, timestamp DESC ", params).await.unwrap(); let mut pinned_updates = Box::pin(intermediate_updates); @@ -603,6 +581,7 @@ pub(super) async fn fetch_node_updates(network_graph: &Network let mut has_address_set_changed = false; let mut has_feature_set_changed = false; let mut latest_mutation_timestamp = None; + let mut is_reminder_due = false; while let Some(row_res) = pinned_updates.next().await { let intermediate_update = row_res.unwrap(); intermediate_update_count += 1; @@ -623,6 +602,7 @@ pub(super) async fn fetch_node_updates(network_graph: &Network has_address_set_changed = false; has_feature_set_changed = false; latest_mutation_timestamp = None; + is_reminder_due = node_reminder_bucket_flag(&node_id) & due_reminder_buckets != 0; // this is the highest timestamp value, so set the seen timestamp accordingly current_node_delta.latest_details.as_mut().map(|d| d.seen.replace(current_seen_timestamp)); @@ -652,7 +632,7 @@ pub(super) async fn fetch_node_updates(network_graph: &Network features: has_feature_set_changed, })); } - } else if include_reminders && latest_mutation_timestamp.unwrap_or(u32::MAX) <= reminder_inclusion_threshold_timestamp { + } else if is_reminder_due && latest_mutation_timestamp.unwrap_or(u32::MAX) <= reminder_inclusion_threshold_timestamp { // only send a reminder if the latest mutation occurred at least 6 days ago current_node_delta.strategy = Some(NodeSerializationStrategy::Reminder); } diff --git a/src/tests/mod.rs b/src/tests/mod.rs index 0379aaa2..27084f22 100644 --- a/src/tests/mod.rs +++ b/src/tests/mod.rs @@ -16,8 +16,9 @@ use lightning::routing::gossip::{NetworkGraph, NodeAlias, NodeId}; use lightning::types::features::{ChannelFeatures, NodeFeatures}; use lightning::util::ser::Writeable; use lightning_rapid_gossip_sync::RapidGossipSync; -use crate::{calculate_delta, config, serialize_delta, serialize_empty_blob}; +use crate::{calculate_delta, config, lookup, serialize_delta, serialize_empty_blob}; use crate::persistence::GossipPersister; +use crate::serialization::NodeSerializationStrategy; use crate::snapshot::Snapshotter; use crate::types::{GossipMessage, tests::TestLogger}; @@ -657,11 +658,15 @@ async fn test_channel_reminders() { let channel_count = network_graph.read_only().channels().len(); assert_eq!(channel_count, 2); - let delta = calculate_delta(&network_graph, timestamp - channel_reminder_delta + 15, None, logger.clone()).await; + // use a reference timestamp that puts the snapshot scope past a full reminder rotation period + // (the reminder age), + // so that every bucket is due + let reference_timestamp = (timestamp + 15) as u64 + 3600; + let delta = calculate_delta(&network_graph, timestamp - channel_reminder_delta + 15, Some(reference_timestamp), logger.clone()).await; let serialization = serialize_delta(&delta, 1, logger.clone()); logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 0 update rows of the first update in a new direction", 1); - logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 4 update rows of the latest update in the less recently updated direction", 1); + logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Annotated 2 channels for reminders", 1); logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched + processed 2 reference rows", 1); logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched + processed intermediate rows (2)", 1); @@ -678,6 +683,249 @@ async fn test_channel_reminders() { clean_test_db().await; } +#[test] +fn test_reminder_bucket_rotation() { + let slot = config::REMINDER_SLOT_INTERVAL.as_secs(); + let period = config::CHANNEL_REMINDER_AGE.as_secs(); + let bucket_count = config::REMINDER_BUCKET_COUNT; + let all_buckets = (1u64 << bucket_count) - 1; + + // a reference timestamp on the slot grid, i. e. a snapshot generated right at a slot boundary + let reference = (current_time() as u64 / slot) * slot; + let reference_bucket = (reference / slot) % bucket_count; + + // a window that contains exactly one slot boundary reminds exactly that bucket + let due = lookup::reminder_buckets_due((reference - 3 * 3600) as u32, reference); + assert_eq!(due, 1u64 << reference_bucket); + assert_ne!(due & (1u64 << reference_bucket), 0); + assert_eq!(due & (1u64 << ((reference_bucket + 1) % bucket_count)), 0); + assert_eq!(lookup::reminder_buckets_due((reference - slot) as u32, reference), due); + + // a window that contains no slot boundary reminds nothing + assert_eq!(lookup::reminder_buckets_due(reference as u32, reference + 3 * 3600), 0); + assert_eq!(lookup::reminder_buckets_due((reference - 3 * 3600) as u32, reference - 1), 0); + + // consecutive windows chain: they are disjoint and their union is the combined window + let combined = lookup::reminder_buckets_due((reference - 24 * 3600) as u32, reference); + let first_half = lookup::reminder_buckets_due((reference - 24 * 3600) as u32, reference - 9 * 3600); + let second_half = lookup::reminder_buckets_due((reference - 9 * 3600) as u32, reference); + assert_eq!(combined.count_ones() as u64, 24 * 3600 / slot); + assert_eq!(first_half & second_half, 0); + assert_eq!(first_half | second_half, combined); + + // a client that has been away for a full rotation period (or longer, or never synced) gets all + assert_eq!(lookup::reminder_buckets_due((reference - period + slot) as u32, reference).count_ones() as u64, bucket_count - 1); + assert_eq!(lookup::reminder_buckets_due((reference - period) as u32, reference), all_buckets); + assert_eq!(lookup::reminder_buckets_due((reference - 3 * period) as u32, reference), all_buckets); + assert_eq!(lookup::reminder_buckets_due(0, reference), all_buckets); + + // realistically shaped short channel ids (block height, tx index, output index 0 or 1) are + // spread evenly across buckets + let mut bucket_sizes = vec![0u64; bucket_count as usize]; + for block_height in 800_000u64..800_100 { + for tx_index in 0u64..50 { + for output_index in 0u64..2 { + let scid = (block_height << 40) | (tx_index << 16) | output_index; + bucket_sizes[lookup::channel_reminder_bucket_flag(scid).trailing_zeros() as usize] += 1; + } + } + } + let expected_bucket_size = 100 * 50 * 2 / bucket_count; + for bucket_size in bucket_sizes { + assert!(bucket_size > expected_bucket_size * 7 / 10, "bucket size {} too small", bucket_size); + assert!(bucket_size < expected_bucket_size * 13 / 10, "bucket size {} too large", bucket_size); + } +} + +/// Channel reminders are spread across channels: a channel is reminded only in snapshots whose +/// window contains the slot boundary at which its bucket becomes due, and consecutive windows chain. +#[tokio::test] +async fn test_channel_reminder_rotation() { + let _sanitizer = SchemaSanitizer::new(); + + let logger = Arc::new(TestLogger::new()); + let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let (mut persister, receiver) = GossipPersister::new(logger.clone()).await; + + let slot = config::REMINDER_SLOT_INTERVAL.as_secs(); + let period = config::CHANNEL_REMINDER_AGE.as_secs(); + let bucket_count = config::REMINDER_BUCKET_COUNT; + + // the reference timestamp of a snapshot generated at the most recent slot boundary + let reference = (current_time() as u64 / slot) * slot; + let reference_bucket = (reference / slot) % bucket_count; + let previous_bucket = (reference / slot + bucket_count - 1) % bucket_count; + let idle_bucket = (reference / slot + bucket_count - 2) % bucket_count; + + let scid_in_bucket = |bucket: u64| (1u64..).find(|scid| lookup::channel_reminder_bucket_flag(*scid) == 1u64 << bucket).unwrap(); + let due_scid = scid_in_bucket(reference_bucket); + let previously_due_scid = scid_in_bucket(previous_bucket); + let idle_scid = scid_in_bucket(idle_bucket); + + // all channels were last updated ten days ago, so no snapshot below contains a real update + let seen = (reference - 10 * 24 * 3600) as u32; + { // seed the db + for scid in [due_scid, previously_due_scid, idle_scid] { + let announcement = generate_channel_announcement(scid); + let update_1 = generate_update(scid, false, seen, 0, 0, 0, 5, 0); + let update_2 = generate_update(scid, true, seen, 0, 0, 0, 3, 0); + + network_graph.update_channel_from_announcement_no_lookup(&announcement).unwrap(); + network_graph.update_channel_unsigned(&update_1.contents).unwrap(); + network_graph.update_channel_unsigned(&update_2.contents).unwrap(); + + receiver.send(GossipMessage::ChannelAnnouncement(announcement, 100, Some(seen))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_1, Some(seen))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_2, Some(seen))).await.unwrap(); + } + drop(receiver); + persister.persist_gossip().await; + } + assert_eq!(network_graph.read_only().channels().len(), 3); + + let reminded_scids = |delta: &crate::serialization::SerializationSet| { + let mut scids: Vec = delta.updates.iter().map(|update| update.scid()).collect(); + scids.sort(); + scids + }; + + // a three-hour window ending on the slot boundary reminds exactly the bucket that became due + let delta = calculate_delta(&network_graph, (reference - 3 * 3600) as u32, Some(reference), logger.clone()).await; + let serialization = serialize_delta(&delta, 1, logger.clone()); + assert_eq!(serialization.channel_announcement_count, 0); + assert_eq!(serialization.update_count, 2); + assert_eq!(serialization.update_count_full, 0); + assert_eq!(serialization.update_count_incremental, 2); + assert_eq!(reminded_scids(&delta), vec![due_scid, due_scid]); + + // the following three-hour window contains no slot boundary: nothing is reminded + let delta = calculate_delta(&network_graph, reference as u32, Some(reference + 3 * 3600), logger.clone()).await; + assert_eq!(delta.updates.len(), 0); + + // the window ending on the previous slot boundary reminded the previous bucket + let delta = calculate_delta(&network_graph, (reference - 9 * 3600) as u32, Some(reference - 6 * 3600), logger.clone()).await; + assert_eq!(reminded_scids(&delta), vec![previously_due_scid, previously_due_scid]); + + // a client that skipped a snapshot receives both buckets at once + let delta = calculate_delta(&network_graph, (reference - 12 * 3600) as u32, Some(reference), logger.clone()).await; + let mut expected_scids = vec![due_scid, due_scid, previously_due_scid, previously_due_scid]; + expected_scids.sort(); + assert_eq!(reminded_scids(&delta), expected_scids); + + // a client that has been away for a whole rotation period is reminded of everything + let delta = calculate_delta(&network_graph, (reference - period) as u32, Some(reference), logger.clone()).await; + let serialization = serialize_delta(&delta, 1, logger.clone()); + assert_eq!(serialization.channel_announcement_count, 0); + assert_eq!(serialization.update_count, 6); + assert_eq!(serialization.update_count_incremental, 6); + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + + clean_test_db().await; +} + +/// Node reminders follow the same rotation as channel reminders. +#[tokio::test] +async fn test_node_reminder_rotation() { + let _sanitizer = SchemaSanitizer::new(); + + let logger = Arc::new(TestLogger::new()); + let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let (mut persister, receiver) = GossipPersister::new(logger.clone()).await; + + let slot = config::REMINDER_SLOT_INTERVAL.as_secs(); + let bucket_count = config::REMINDER_BUCKET_COUNT; + let day = 24 * 3600; + + let reference = (current_time() as u64 / slot) * slot; + let reference_bucket = (reference / slot) % bucket_count; + let previous_bucket = (reference / slot + bucket_count - 1) % bucket_count; + let idle_bucket = (reference / slot + bucket_count - 2) % bucket_count; + + let secp_context = Secp256k1::new(); + let node_id = |key: &SecretKey| NodeId::from_pubkey(&key.public_key(&secp_context)); + let key_in_bucket = |bucket: u64| (1u8..=255) + .map(|byte| SecretKey::from_slice(&[byte; 32]).unwrap()) + .find(|key| lookup::node_reminder_bucket_flag(&node_id(key)) == 1u64 << bucket) + .unwrap(); + let due_key = key_in_bucket(reference_bucket); + let previously_due_key = key_in_bucket(previous_bucket); + // the counterparty never announces itself, so it is not part of any node delta + let counterparty_key = key_in_bucket(idle_bucket); + + { // seed the db + let channel_seen = (reference - 10 * day) as u32; + for (scid, key) in [(1u64, &due_key), (2u64, &previously_due_key)] { + let announcement = generate_channel_announcement_between_nodes(scid, key, &counterparty_key); + let update_1 = generate_update(scid, false, channel_seen, 0, 0, 0, 5, 0); + let update_2 = generate_update(scid, true, channel_seen, 0, 0, 0, 3, 0); + + network_graph.update_channel_from_announcement_no_lookup(&announcement).unwrap(); + network_graph.update_channel_unsigned(&update_1.contents).unwrap(); + network_graph.update_channel_unsigned(&update_2.contents).unwrap(); + + receiver.send(GossipMessage::ChannelAnnouncement(announcement, 100, Some(channel_seen))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_1, Some(channel_seen))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_2, Some(channel_seen))).await.unwrap(); + + // both nodes changed their addresses four days ago and have been stable since, which + // is what makes them eligible for a reminder (rather than a full re-serialization) + let mut old_announcement = generate_node_announcement(Some(key.clone())); + old_announcement.contents.timestamp = (reference - 10 * day) as u32; + network_graph.update_node_from_unsigned_announcement(&old_announcement.contents).unwrap(); + receiver.send(GossipMessage::NodeAnnouncement(old_announcement.clone(), Some(old_announcement.contents.timestamp))).await.unwrap(); + + let mut current_announcement = generate_node_announcement(Some(key.clone())); + current_announcement.contents.timestamp = (reference - 4 * day) as u32; + current_announcement.contents.addresses.push(SocketAddress::TcpIpV4 { addr: [127, 0, 0, 1], port: 9735 }); + network_graph.update_node_from_unsigned_announcement(¤t_announcement.contents).unwrap(); + receiver.send(GossipMessage::NodeAnnouncement(current_announcement.clone(), Some(current_announcement.contents.timestamp))).await.unwrap(); + } + drop(receiver); + persister.persist_gossip().await; + } + + let reminded_nodes = |delta: &crate::serialization::SerializationSet| { + let mut node_ids: Vec = delta.node_mutations.iter() + .filter(|(_, node_delta)| matches!(node_delta.strategy, Some(NodeSerializationStrategy::Reminder))) + .map(|(node_id, _)| node_id.clone()) + .collect(); + node_ids.sort(); + node_ids + }; + + // a three-hour window ending on the slot boundary reminds exactly the node whose bucket became due + let delta = calculate_delta(&network_graph, (reference - 3 * 3600) as u32, Some(reference), logger.clone()).await; + let serialization = serialize_delta(&delta, 2, logger.clone()); + assert_eq!(serialization.message_count, 0); + assert_eq!(serialization.node_announcement_count, 1); + assert_eq!(serialization.node_update_count, 0); + assert_eq!(delta.node_mutations.len(), 1); + assert_eq!(reminded_nodes(&delta), vec![node_id(&due_key)]); + + // the following three-hour window contains no slot boundary: nothing is reminded + let delta = calculate_delta(&network_graph, reference as u32, Some(reference + 3 * 3600), logger.clone()).await; + assert_eq!(delta.node_mutations.len(), 0); + + // the window ending on the previous slot boundary reminded the other node + let delta = calculate_delta(&network_graph, (reference - 9 * 3600) as u32, Some(reference - 6 * 3600), logger.clone()).await; + assert_eq!(reminded_nodes(&delta), vec![node_id(&previously_due_key)]); + + // a client that skipped a snapshot is reminded of both at once + let delta = calculate_delta(&network_graph, (reference - 12 * 3600) as u32, Some(reference), logger.clone()).await; + let mut expected_node_ids = vec![node_id(&due_key), node_id(&previously_due_key)]; + expected_node_ids.sort(); + assert_eq!(reminded_nodes(&delta), expected_node_ids); + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + + clean_test_db().await; +} + #[tokio::test] async fn test_full_snapshot_recency() { let _sanitizer = SchemaSanitizer::new(); From 4112972ae1d6e69d089bf043239331b2ad191998 Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Mon, 14 Sep 2026 23:52:55 +0000 Subject: [PATCH 2/5] Re-announce channels whose updates resume after a prune-length gap A channel_announcement is sent to an incremental client only if its `seen` timestamp, or the moment the channel first had updates in both directions, is newer than the client's last sync. Both are one-time events, so a channel that leaves and later re-enters our snapshots is never announced again. That is exactly what happens when one peer of a channel goes offline for a month: after two weeks its direction is pruned from our network graph, which drops the channel altogether and rejects its updates, so clients stop receiving reminders for it and prune it a week later. When the peer comes back it re-broadcasts the announcement and both peers update the channel again, and it re-enters our graph, but every client that synced in the meantime only ever receives updates for a channel it no longer has, and drops them. Instead of the first update per direction, we now look, per direction, at the first update seen after the client's last sync and check whether it had a predecessor within the prune interval. If not, updates in that direction (re)started after the last sync, either because the channel is new (the previous behaviour) or because it had been pruned and has come back, and the channel is announced alongside full updates in both directions. Since we only ever drop a channel after a full prune interval without updates and then block it for another week, every resumption after a prune implies such a gap in the update history. Co-Authored-By: Claude Fable 5.1 --- README.md | 3 +- src/lookup.rs | 104 +++++++++++++++++++++-------------- src/serialization.rs | 6 ++- src/tests/mod.rs | 126 +++++++++++++++++++++++++++++++++++++++++-- 4 files changed, 193 insertions(+), 46 deletions(-) diff --git a/README.md b/README.md index c88dfb6f..6348a360 100644 --- a/README.md +++ b/README.md @@ -72,7 +72,8 @@ seen prior to the given timestamp, the latest known updates, and, if necessary, updates are collected. Then, any channel that has only had an announcement but never an update is dropped. Additionally, -every channel whose first update was seen after the given timestamp is collected alongside its +every channel whose updates in either direction (re)started after the given timestamp, having never +been seen before or having been absent for at least the prune interval, is collected alongside its announcement. To avoid unchanging data being eventually pruned by clients, "reminder" updates are added in a diff --git a/src/lookup.rs b/src/lookup.rs index 9683c68d..006b5b57 100644 --- a/src/lookup.rs +++ b/src/lookup.rs @@ -49,7 +49,10 @@ pub(super) struct DirectedUpdateDelta { pub(super) struct ChannelDelta { pub(super) announcement: Option, pub(super) updates: (Option, Option), - pub(super) first_bidirectional_updates_seen: Option, + /// The `seen` timestamp at which this channel most recently (re)gained updates in a direction + /// that had previously not had any for at least the prune interval (or ever), provided that + /// happened after the client's last sync. Such a channel needs to be (re-)announced. + pub(super) updates_resumed_seen: Option, /// Whether this channel's reminder bucket became due within the snapshot's window, in which /// case each direction without a real update to send gets a flags-only reminder update pub(super) requires_reminder: bool, @@ -77,7 +80,7 @@ impl Default for ChannelDelta { Self { announcement: None, updates: (None, None), - first_bidirectional_updates_seen: None, + updates_resumed_seen: None, requires_reminder: false, } } @@ -143,8 +146,8 @@ pub(super) fn node_reminder_bucket_flag(node_id: &NodeId) -> u64 { /// Fetch all the channel announcements that are presently in the network graph, regardless of /// whether they had been seen before. -/// Also include all announcements for which the first update was announced -/// after `last_sync_timestamp` +/// Also include all announcements for which updates in either direction (re)started after +/// `last_sync_timestamp`, be it because the channel is new or because it had been pruned pub(super) async fn fetch_channel_announcements(delta_set: &mut DeltaSet, network_graph: &NetworkGraph, client: &Client, last_sync_timestamp: u32, snapshot_reference_timestamp: Option, logger: L) where L::Target: Logger { log_info!(logger, "Obtaining channel ids from network graph"); let channel_ids = { @@ -196,55 +199,76 @@ pub(super) async fn fetch_channel_announcements(delta_set: &mut DeltaS log_info!(logger, "Fetched {} announcement rows in {:?}", announcement_count, start.elapsed()); { - // THIS STEP IS USED TO DETERMINE IF A CHANNEL SHOULD BE OMITTED FROM THE DELTA - - log_info!(logger, "Annotating channel announcements whose oldest channel update in a given direction occurred after the last sync"); - // Steps: - // — Obtain all updates, distinct by (scid, direction), ordered by seen DESC // to find the oldest update in a given direction - // — From those updates, select distinct by (scid), ordered by seen DESC (to obtain the newer one per direction) - // This will allow us to mark the first time updates in both directions were seen - - // here is where the channels whose first update in either direction occurred after - // `last_seen_timestamp` are added to the selection + log_info!(logger, "Annotating channels whose updates in a direction (re)started after the last sync"); + // Clients only receive a channel's announcement once, and prune the channel if our + // snapshots stop covering it, which happens when one of its peers stops announcing + // for two weeks. When that peer comes back we need to provide clients a fresh + // update. + // + // We detect this from the update history alone: for each direction, take the first update + // seen at or after the last sync and check whether it had a predecessor within the prune + // interval. If not, updates in that direction (re)started after the client's last sync, be + // it because the channel is brand new or because it was pruned and has come back. As we + // only ever drop channels after a full prune interval without updates, and block them from + // being re-added for another week, any such resumption implies a gap in the history. + let prune_interval_seconds = (config::PRUNE_INTERVAL.as_secs() - 60 * 60 * 24) as f64; let start = Instant::now(); - let params: [&(dyn tokio_postgres::types::ToSql + Sync); 2] = - [&channel_ids, &last_sync_timestamp_float]; - let newer_oldest_directional_updates = client.query_raw(" + let params: [&(dyn tokio_postgres::types::ToSql + Sync); 3] = + [&channel_ids, &last_sync_timestamp_float, &prune_interval_seconds]; + let resumed_directional_updates = client.query_raw(" SELECT scids.short_channel_id, CAST(EXTRACT('epoch' from GREATEST(dir0.seen, dir1.seen)) AS BIGINT) AS seen FROM unnest($1::bigint[]) AS scids(short_channel_id) - CROSS JOIN LATERAL ( - SELECT seen - FROM channel_updates - WHERE short_channel_id = scids.short_channel_id AND direction = false - ORDER BY seen ASC - LIMIT 1 - ) dir0 - CROSS JOIN LATERAL ( - SELECT seen - FROM channel_updates - WHERE short_channel_id = scids.short_channel_id AND direction = true - ORDER BY seen ASC - LIMIT 1 - ) dir1 - WHERE GREATEST(dir0.seen, dir1.seen) >= TO_TIMESTAMP($2) + LEFT JOIN LATERAL ( + SELECT first_recent.seen + FROM ( + SELECT seen + FROM channel_updates + WHERE short_channel_id = scids.short_channel_id AND direction = false AND seen >= TO_TIMESTAMP($2) + ORDER BY seen ASC + LIMIT 1 + ) first_recent + WHERE NOT EXISTS ( + SELECT 1 + FROM channel_updates AS predecessor + WHERE predecessor.short_channel_id = scids.short_channel_id AND predecessor.direction = false + AND predecessor.seen < first_recent.seen + AND predecessor.seen >= first_recent.seen - $3 * INTERVAL '1 second' + ) + ) dir0 ON TRUE + LEFT JOIN LATERAL ( + SELECT first_recent.seen + FROM ( + SELECT seen + FROM channel_updates + WHERE short_channel_id = scids.short_channel_id AND direction = true AND seen >= TO_TIMESTAMP($2) + ORDER BY seen ASC + LIMIT 1 + ) first_recent + WHERE NOT EXISTS ( + SELECT 1 + FROM channel_updates AS predecessor + WHERE predecessor.short_channel_id = scids.short_channel_id AND predecessor.direction = true + AND predecessor.seen < first_recent.seen + AND predecessor.seen >= first_recent.seen - $3 * INTERVAL '1 second' + ) + ) dir1 ON TRUE + WHERE dir0.seen IS NOT NULL OR dir1.seen IS NOT NULL ", params).await.unwrap(); - let mut pinned_updates = Box::pin(newer_oldest_directional_updates); + let mut pinned_updates = Box::pin(resumed_directional_updates); - let mut newer_oldest_directional_update_count = 0; + let mut resumed_directional_update_count = 0; while let Some(row_res) = pinned_updates.next().await { let current_row = row_res.unwrap(); let scid: i64 = current_row.get("short_channel_id"); let current_seen_timestamp = current_row.get::<_, i64>("seen") as u32; - // the newer of the two oldest seen directional updates came after last sync timestamp let current_channel_delta = delta_set.entry(scid as u64).or_insert(ChannelDelta::default()); - // first time a channel was seen in both directions - (*current_channel_delta).first_bidirectional_updates_seen = Some(current_seen_timestamp); + (*current_channel_delta).updates_resumed_seen = Some(current_seen_timestamp); - newer_oldest_directional_update_count += 1; + resumed_directional_update_count += 1; } - log_info!(logger, "Fetched {} update rows of the first update in a new direction in {:?}", newer_oldest_directional_update_count, start.elapsed()); + log_info!(logger, "Fetched {} update rows of the first update in a (re)started direction in {:?}", resumed_directional_update_count, start.elapsed()); } if due_reminder_buckets != 0 { @@ -260,7 +284,7 @@ pub(super) async fn fetch_channel_announcements(delta_set: &mut DeltaS let is_new_announcement = current_channel_delta.announcement.as_ref() .map(|announcement| announcement.seen >= last_sync_timestamp) .unwrap_or(false); - if is_new_announcement || current_channel_delta.first_bidirectional_updates_seen.is_some() { + if is_new_announcement || current_channel_delta.updates_resumed_seen.is_some() { // the client is about to receive the announcement alongside full updates // for this channel, so a reminder would be pointless continue; diff --git a/src/serialization.rs b/src/serialization.rs index 6be4f8d1..56681d7e 100644 --- a/src/serialization.rs +++ b/src/serialization.rs @@ -158,8 +158,10 @@ pub(super) fn serialize_delta_set(channel_delta_set: DeltaSet, node_delta_set: N let current_announcement_seen = channel_announcement_delta.seen; let is_new_announcement = current_announcement_seen >= last_sync_timestamp; - let is_newly_included_announcement = if let Some(first_update_seen) = channel_delta.first_bidirectional_updates_seen { - first_update_seen >= last_sync_timestamp + // the channel's updates (re)started after the last sync, so the client is either unaware + // of the channel or has pruned it in the meantime + let is_newly_included_announcement = if let Some(updates_resumed_seen) = channel_delta.updates_resumed_seen { + updates_resumed_seen >= last_sync_timestamp } else { false }; diff --git a/src/tests/mod.rs b/src/tests/mod.rs index 27084f22..8785bd50 100644 --- a/src/tests/mod.rs +++ b/src/tests/mod.rs @@ -511,7 +511,7 @@ async fn test_unidirectional_intermediate_update_consideration() { let delta = calculate_delta(&network_graph, timestamp + 1, None, logger.clone()).await; let serialization = serialize_delta(&delta, 1, logger.clone()); - logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 1 update rows of the first update in a new direction", 1); + logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 1 update rows of the first update in a (re)started direction", 1); logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched + processed 1 reference rows", 1); logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched + processed intermediate rows (2)", 1); @@ -578,7 +578,7 @@ async fn test_bidirectional_intermediate_update_consideration() { let delta = calculate_delta(&network_graph, timestamp + 1, None, logger.clone()).await; let serialization = serialize_delta(&delta, 1, logger.clone()); - logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 0 update rows of the first update in a new direction", 1); + logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 0 update rows of the first update in a (re)started direction", 1); logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched + processed 2 reference rows", 1); logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched + processed intermediate rows (2)", 1); @@ -665,7 +665,7 @@ async fn test_channel_reminders() { let delta = calculate_delta(&network_graph, timestamp - channel_reminder_delta + 15, Some(reference_timestamp), logger.clone()).await; let serialization = serialize_delta(&delta, 1, logger.clone()); - logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 0 update rows of the first update in a new direction", 1); + logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 0 update rows of the first update in a (re)started direction", 1); logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Annotated 2 channels for reminders", 1); logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched + processed 2 reference rows", 1); logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched + processed intermediate rows (2)", 1); @@ -926,6 +926,126 @@ async fn test_node_reminder_rotation() { clean_test_db().await; } +/// A channel one of whose peers was offline long enough for the channel to be pruned (by us and +/// by clients) must be announced again, with full updates, once that peer is back. +#[tokio::test] +async fn test_channel_with_resumed_updates_is_reannounced() { + let _sanitizer = SchemaSanitizer::new(); + + let logger = Arc::new(TestLogger::new()); + let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let (mut persister, receiver) = GossipPersister::new(logger.clone()).await; + + let short_channel_id = 1; + let timestamp = current_time(); + let day = 24 * 3600; + + { // seed the db + // the channel was announced a month ago, and both peers last updated it then (the gossip + // timestamps are kept recent so that the test graph itself doesn't prune the channel) + let announcement = generate_channel_announcement(short_channel_id); + let old_update_1 = generate_update(short_channel_id, false, timestamp - 2 * day, 0, 0, 0, 5, 0); + let old_update_2 = generate_update(short_channel_id, true, timestamp - 2 * day, 0, 0, 0, 10, 0); + network_graph.update_channel_from_announcement_no_lookup(&announcement).unwrap(); + receiver.send(GossipMessage::ChannelAnnouncement(announcement, 100, Some(timestamp - 30 * day))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(old_update_1, Some(timestamp - 30 * day))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(old_update_2, Some(timestamp - 30 * day))).await.unwrap(); + + // today, both peers are back and update the channel again + let update_1 = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 6, 0); + let update_2 = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 11, 0); + network_graph.update_channel_unsigned(&update_1.contents).unwrap(); + network_graph.update_channel_unsigned(&update_2.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_1, Some(timestamp))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_2, Some(timestamp))).await.unwrap(); + + drop(receiver); + persister.persist_gossip().await; + } + + // a client that last synced three hours ago has long since pruned the channel + let delta = calculate_delta(&network_graph, timestamp - 3 * 3600, Some(timestamp as u64), logger.clone()).await; + let serialization = serialize_delta(&delta, 1, logger.clone()); + logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 1 update rows of the first update in a (re)started direction", 1); + + assert_eq!(serialization.channel_announcement_count, 1); + assert_eq!(serialization.update_count, 2); + assert_eq!(serialization.update_count_full, 2); + + // such a client gets the channel back in full + let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let rgs = RapidGossipSync::new(&client_graph, logger.clone()); + rgs.update_network_graph(&serialization.data).unwrap(); + { + let readonly_graph = client_graph.read_only(); + let channel = readonly_graph.channels().get(&short_channel_id).unwrap(); + assert_eq!(channel.one_to_two.as_ref().unwrap().fees.base_msat, 6); + assert_eq!(channel.two_to_one.as_ref().unwrap().fees.base_msat, 11); + } + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + + clean_test_db().await; +} + +/// If the two peers of a pruned channel come back at different times, clients that synced in +/// between have only seen the first direction resume, and still need the announcement. +#[tokio::test] +async fn test_channel_with_staggered_resumed_updates_is_reannounced() { + let _sanitizer = SchemaSanitizer::new(); + + let logger = Arc::new(TestLogger::new()); + let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let (mut persister, receiver) = GossipPersister::new(logger.clone()).await; + + let short_channel_id = 1; + let timestamp = current_time(); + let day = 24 * 3600; + + { // seed the db + let announcement = generate_channel_announcement(short_channel_id); + let old_update_1 = generate_update(short_channel_id, false, timestamp - 2 * day, 0, 0, 0, 5, 0); + let old_update_2 = generate_update(short_channel_id, true, timestamp - 2 * day, 0, 0, 0, 10, 0); + network_graph.update_channel_from_announcement_no_lookup(&announcement).unwrap(); + receiver.send(GossipMessage::ChannelAnnouncement(announcement, 100, Some(timestamp - 30 * day))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(old_update_1, Some(timestamp - 30 * day))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(old_update_2, Some(timestamp - 30 * day))).await.unwrap(); + + // the first peer is back now, the second one only two hours later + let update_1 = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 6, 0); + let update_2 = generate_update(short_channel_id, true, timestamp + 2 * 3600, 0, 0, 0, 11, 0); + network_graph.update_channel_unsigned(&update_1.contents).unwrap(); + network_graph.update_channel_unsigned(&update_2.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_1, Some(timestamp))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_2, Some(timestamp + 2 * 3600))).await.unwrap(); + + drop(receiver); + persister.persist_gossip().await; + } + + // a client that synced an hour from now, between the two peers returning, and fetches a + // snapshot generated three hours from now + let delta = calculate_delta(&network_graph, timestamp + 3600, Some((timestamp + 3 * 3600) as u64), logger.clone()).await; + let serialization = serialize_delta(&delta, 1, logger.clone()); + + assert_eq!(serialization.channel_announcement_count, 1); + assert_eq!(serialization.update_count, 2); + assert_eq!(serialization.update_count_full, 2); + + // while a client that synced after both peers were back is not sent the channel again + let delta = calculate_delta(&network_graph, timestamp + 3 * 3600, Some((timestamp + 6 * 3600) as u64), logger.clone()).await; + assert_eq!(delta.announcements.len(), 0); + assert_eq!(delta.updates.len(), 0); + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + + clean_test_db().await; +} + #[tokio::test] async fn test_full_snapshot_recency() { let _sanitizer = SchemaSanitizer::new(); From 5e9bf584934f6396fb1a7f8ef7b02603e8a3fe9e Mon Sep 17 00:00:00 2001 From: Vincenzo Palazzo Date: Tue, 6 Oct 2026 16:04:54 -0700 Subject: [PATCH 3/5] Publish the v2 reference-timestamp symlink v2 clients persist the snapshot header and request that timestamp next. The empty delta was written for both versions, but its public symlink was created only in the v1 directory, so the v2 follow-up was HTTP 404. Fixes lightningdevkit/rapid-gossip-sync-server#108 --- src/snapshot.rs | 14 ++++++++++---- src/tests/mod.rs | 20 ++++++++++++++++++++ 2 files changed, 30 insertions(+), 4 deletions(-) diff --git a/src/snapshot.rs b/src/snapshot.rs index 7f88fd08..fabb01c3 100644 --- a/src/snapshot.rs +++ b/src/snapshot.rs @@ -144,10 +144,16 @@ impl Snapshotter where L::Target: Logger { fs::write(&dummy_snapshot_path_v1, dummy_snapshot_v1).unwrap(); fs::write(&dummy_snapshot_path_v2, dummy_snapshot_v2).unwrap(); - let dummy_symlink_path = format!("{}/{}.bin", pending_symlink_directory, reference_timestamp); - let relative_dummy_snapshot_path = format!("{}/{}", relative_symlink_to_snapshot_path, dummy_filename); - log_info!(self.logger, "Symlinking dummy: {} -> {}", dummy_symlink_path, relative_dummy_snapshot_path); - symlink(&relative_dummy_snapshot_path, &dummy_symlink_path).unwrap(); + // The timestamp written into every snapshot header must exist for + // both protocol versions. Real deltas already do this in the loop + // below. This symlink used to be v1-only, so a v2 client that + // stored `reference_timestamp` got HTTP 404. + for (suffix, path_to_root) in suffixes { + let dummy_symlink_path = format!("{}{}/{}.bin", pending_symlink_directory, suffix, reference_timestamp); + let relative_dummy_snapshot_path = format!("{}{}{}/{}", path_to_root, relative_symlink_to_snapshot_path, suffix, dummy_filename); + log_info!(self.logger, "Symlinking dummy: {} -> {}", dummy_symlink_path, relative_dummy_snapshot_path); + symlink(&relative_dummy_snapshot_path, &dummy_symlink_path).unwrap(); + } } // Number of intervals since Jan 1, 2022, a few months before RGS server was released. diff --git a/src/tests/mod.rs b/src/tests/mod.rs index 0379aaa2..1ed79730 100644 --- a/src/tests/mod.rs +++ b/src/tests/mod.rs @@ -1213,6 +1213,26 @@ async fn test_full_snapshot_persistence() { // the update result must be a multiple of our snapshot granularity assert_eq!(update_result % config::snapshot_generation_interval(), 0); + // v2 clients request this generation's reference timestamp next. + // That symlink used to be v1-only, so the follow-up sync was HTTP 404. + let v2_dir = std::env::current_dir().unwrap().join(&cache_path).join("symlinks/v2"); + let reference_symlink = fs::read_dir(&v2_dir) + .unwrap() + .flatten() + .filter_map(|e| { + let ts = e.path().file_stem().and_then(|s| s.to_str()).and_then(|s| s.parse::().ok())?; + (ts != 0).then_some((ts, e.path())) + }) + .max_by_key(|(ts, _)| *ts) + .unwrap() + .1; + let v2_empty = fs::read(&reference_symlink).unwrap(); + let v2_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let v2_rgs = RapidGossipSync::new(&v2_graph, logger.clone()); + let applied = v2_rgs.update_network_graph(&v2_empty).unwrap(); + assert_eq!(applied % config::snapshot_generation_interval(), 0); + assert!(v2_graph.read_only().channels().is_empty()); + let readonly_graph = client_graph.read_only(); let channels = readonly_graph.channels(); let client_channel_count = channels.len(); From 73cf3b5dab3440674e86c90e7481e99532662d20 Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Fri, 9 Oct 2026 21:55:32 +0000 Subject: [PATCH 4/5] Run tests against a throwaway local Postgres instead of a service Rather than relying on a `postgres` service container, which requires the CI runner to be able to run docker images, add `ci/ci-tests.sh`, which uses the locally installed Postgres server binaries to create a temporary database cluster, listening only on a Unix socket inside its (private) temporary directory, runs `cargo test` against it, and deletes it again once the tests complete. As it doesn't need a running Postgres or any database configuration, it is also the easiest way to run the tests locally. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/build.yml | 18 +------------ README.md | 7 ++++++ ci/ci-tests.sh | 50 +++++++++++++++++++++++++++++++++++++ 3 files changed, 58 insertions(+), 17 deletions(-) create mode 100755 ci/ci-tests.sh diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index f1e74a83..81303dcb 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -40,18 +40,6 @@ jobs: - stable - 1.63.0 runs-on: ubuntu-latest - services: - postgres: - image: postgres - env: - POSTGRES_PASSWORD: postgres - options: >- - --health-cmd pg_isready - --health-interval 10s - --health-timeout 5s - --health-retries 5 - ports: - - 5432:5432 steps: - name: Checkout source code uses: actions/checkout@v3 @@ -63,10 +51,6 @@ jobs: profile: minimal - name: Run tests run: | - RUST_BACKTRACE=1 cargo test --verbose --color always -- --show-output + RUST_BACKTRACE=1 ./ci/ci-tests.sh --verbose --color always -- --show-output env: - RAPID_GOSSIP_TEST_DB_HOST: localhost - RAPID_GOSSIP_TEST_DB_NAME: postgres - RAPID_GOSSIP_TEST_DB_USER: postgres - RAPID_GOSSIP_TEST_DB_PASSWORD: postgres RAPID_GOSSIP_SYNC_SERVER_LOG_LEVEL: gossip diff --git a/README.md b/README.md index 6348a360..8a869c36 100644 --- a/README.md +++ b/README.md @@ -82,6 +82,13 @@ rolling fashion. Finally, all channel update transitions are evaluated and collected into either a full or an incremental update. +## Testing + +The tests require a Postgres database. `ci/ci-tests.sh` starts a throwaway Postgres instance in a +temporary directory, runs `cargo test` against it (passing along any arguments), and deletes the +instance afterwards. It needs the Postgres server binaries (`initdb`, `pg_ctl` and `postgres`), +which it looks for in `$POSTGRES_BIN_DIR`, then `$PATH`, then `/usr/lib/postgresql/*/bin`. + ## License [Apache 2.0](LICENSE-APACHE.md) or [MIT](LICENSE-MIT.md), [at your option](LICENSE.md). diff --git a/ci/ci-tests.sh b/ci/ci-tests.sh new file mode 100755 index 00000000..e5131f63 --- /dev/null +++ b/ci/ci-tests.sh @@ -0,0 +1,50 @@ +#!/bin/bash +set -eo pipefail + +# Runs the test suite against a throwaway PostgreSQL instance. The database cluster lives in a +# temporary directory, only listens on a Unix socket within that directory, and is deleted once +# the tests complete. +# +# The PostgreSQL server binaries (initdb, pg_ctl and postgres) are taken from $POSTGRES_BIN_DIR if +# set, otherwise from $PATH, falling back to Debian's /usr/lib/postgresql//bin. +# +# Any arguments are passed through to `cargo test`. + +if [ -n "$POSTGRES_BIN_DIR" ]; then + PATH="$POSTGRES_BIN_DIR:$PATH" +elif ! command -v initdb > /dev/null; then + for DIR in /usr/lib/postgresql/*/bin; do + [ -x "$DIR/initdb" ] && POSTGRES_BIN_DIR="$DIR" + done + [ -n "$POSTGRES_BIN_DIR" ] && PATH="$POSTGRES_BIN_DIR:$PATH" +fi +for BIN in initdb pg_ctl postgres; do + if ! command -v "$BIN" > /dev/null; then + echo "Could not find $BIN, set POSTGRES_BIN_DIR to the directory containing the PostgreSQL server binaries" >&2 + exit 1 + fi +done + +PG_DIR="$(mktemp -d)" +cleanup() { + pg_ctl stop -D "$PG_DIR/data" -m immediate > /dev/null 2>&1 || true + rm -rf -- "$PG_DIR" +} +trap cleanup EXIT + +# Only our user can reach the socket in the (0700) temp directory, so trust auth is fine. +initdb -D "$PG_DIR/data" -U postgres --auth=trust --encoding=UTF8 --locale=C --no-sync > /dev/null +cat >> "$PG_DIR/data/postgresql.conf" < /dev/null; then + cat "$PG_DIR/postgres.log" >&2 + exit 1 +fi + +export RAPID_GOSSIP_TEST_DB_HOST="$PG_DIR" +export RAPID_GOSSIP_TEST_DB_USER=postgres +export RAPID_GOSSIP_TEST_DB_NAME=postgres +cargo test "$@" From edd4bf2f5661f871224da396a2c78203e45f719a Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Fri, 9 Oct 2026 22:04:46 +0000 Subject: [PATCH 5/5] Switch to forgejo runners --- .github/workflows/build.yml | 20 ++++++-------------- 1 file changed, 6 insertions(+), 14 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 81303dcb..3e987a1b 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -20,16 +20,12 @@ jobs: - stable - 1.63.0 - beta - runs-on: ubuntu-latest + runs-on: debian-trixie steps: - name: Checkout source code - uses: actions/checkout@v3 + uses: https://data.forgejo.org/actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 - name: Install Rust ${{ matrix.toolchain }} toolchain - uses: actions-rs/toolchain@v1 - with: - toolchain: ${{ matrix.toolchain }} - override: true - profile: minimal + run: rustup default ${{ matrix.toolchain }} - name: Build on Rust ${{ matrix.toolchain }} run: | cargo build --verbose --color always @@ -39,16 +35,12 @@ jobs: toolchain: - stable - 1.63.0 - runs-on: ubuntu-latest + runs-on: debian-trixie steps: - name: Checkout source code - uses: actions/checkout@v3 + uses: https://data.forgejo.org/actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 - name: Install Rust ${{ matrix.toolchain }} toolchain - uses: actions-rs/toolchain@v1 - with: - toolchain: ${{ matrix.toolchain }} - override: true - profile: minimal + run: rustup default ${{ matrix.toolchain }} - name: Run tests run: | RUST_BACKTRACE=1 ./ci/ci-tests.sh --verbose --color always -- --show-output