diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index f1e74a8..3e987a1 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,34 +35,14 @@ jobs: toolchain: - 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 + 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 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 eec2286..8a869c3 100644 --- a/README.md +++ b/README.md @@ -72,12 +72,23 @@ 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 +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 0000000..e5131f6 --- /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 "$@" diff --git a/src/config.rs b/src/config.rs index 8195a31..3f1fc89 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 93671ce..006b5b5 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; @@ -49,8 +49,12 @@ 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 of the older of the two latest directional updates + /// 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, } @@ -76,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, } } @@ -103,28 +107,47 @@ 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 /// 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 = { @@ -144,7 +167,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(); @@ -175,163 +199,129 @@ 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 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.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; } - - 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 +556,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 +605,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 +626,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 +656,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/serialization.rs b/src/serialization.rs index 6be4f8d..56681d7 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/snapshot.rs b/src/snapshot.rs index 7f88fd0..fabb01c 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 0379aaa..4950405 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}; @@ -510,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); @@ -577,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); @@ -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", "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); @@ -678,6 +683,369 @@ 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; +} + +/// 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(); @@ -1213,6 +1581,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();