mirrors/rust-lightning
1
0
Fork 0
mirror of https://github.com/lightningdevkit/rust-lightning.git synced 2025-03-26 21:42:35 +01:00
Code Issues Projects Releases Packages Wiki Activity

Track historical liquidity update time separately from the bounds

Browse source 
In the next commit, we'll start to use the new
`ScoreUpdate::decay_liquidity_certainty` to decay our bounds in the
background. This will result in the `last_updated` field getting
updated regularly on decay, rather than only on update. While this
isn't an issue for the regular liquidity bounds, it poses a problem
for the historical liquidity buckets, which are decayed on a
separate (and by default much longer) timer. If we didn't move to
tracking their decays separately, we'd never let the `last_updated`
field get old enough for the historical buckets to decay at all.

Instead, here we introduce a new `Duration` in the
`ChannelLiquidity` which tracks the last time the historical
liquidity buckets were last updated. We initialize it to a copy of
`last_updated` on deserialization if it is missing.
This commit is contained in:
Matt Corallo 2023-10-02 19:44:36 +00:00
parent b84842a904
commit f0f8194719
1 changed files with 50 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

68
lightning/src/routing/scoring.rs
View file

@ -805,11 +805,14 @@ struct ChannelLiquidity<T: Time> {
/// Upper channel liquidity bound in terms of an offset from the effective capacity. /// Upper channel liquidity bound in terms of an offset from the effective capacity.
max_liquidity_offset_msat: u64, max_liquidity_offset_msat: u64,
min_liquidity_offset_history: HistoricalBucketRangeTracker,
max_liquidity_offset_history: HistoricalBucketRangeTracker,
/// Time when the liquidity bounds were last modified. /// Time when the liquidity bounds were last modified.
last_updated: T, last_updated: T,
min_liquidity_offset_history: HistoricalBucketRangeTracker, /// Time when the historical liquidity bounds were last modified.
max_liquidity_offset_history: HistoricalBucketRangeTracker, offset_history_last_updated: T,
} }
/// A snapshot of [`ChannelLiquidity`] in one direction assuming a certain channel capacity and /// A snapshot of [`ChannelLiquidity`] in one direction assuming a certain channel capacity and
@ -820,6 +823,7 @@ struct DirectedChannelLiquidity<L: Deref<Target = u64>, BRT: Deref<Target = Hist
liquidity_history: HistoricalMinMaxBuckets<BRT>, liquidity_history: HistoricalMinMaxBuckets<BRT>,
capacity_msat: u64, capacity_msat: u64,
last_updated: U, last_updated: U,
offset_history_last_updated: U,
now: T, now: T,
decay_params: ProbabilisticScoringDecayParameters, decay_params: ProbabilisticScoringDecayParameters,
} }
@ -858,7 +862,7 @@ impl<G: Deref<Target = NetworkGraph<L>>, L: Deref, T: Time> ProbabilisticScorerU
let dir_liq = liq.as_directed(source, target, amt, self.decay_params); let dir_liq = liq.as_directed(source, target, amt, self.decay_params);
let (min_buckets, max_buckets) = dir_liq.liquidity_history let (min_buckets, max_buckets) = dir_liq.liquidity_history
.get_decayed_buckets(now, *dir_liq.last_updated, .get_decayed_buckets(now, *dir_liq.offset_history_last_updated,
self.decay_params.historical_no_updates_half_life) self.decay_params.historical_no_updates_half_life)
.unwrap_or(([0; 32], [0; 32])); .unwrap_or(([0; 32], [0; 32]));
@ -955,7 +959,7 @@ impl<G: Deref<Target = NetworkGraph<L>>, L: Deref, T: Time> ProbabilisticScorerU
let (min_buckets, mut max_buckets) = let (min_buckets, mut max_buckets) =
dir_liq.liquidity_history.get_decayed_buckets( dir_liq.liquidity_history.get_decayed_buckets(
dir_liq.now, *dir_liq.last_updated, dir_liq.now, *dir_liq.offset_history_last_updated,
self.decay_params.historical_no_updates_half_life self.decay_params.historical_no_updates_half_life
)?; )?;
@ -988,7 +992,7 @@ impl<G: Deref<Target = NetworkGraph<L>>, L: Deref, T: Time> ProbabilisticScorerU
let dir_liq = liq.as_directed(source, target, capacity_msat, self.decay_params); let dir_liq = liq.as_directed(source, target, capacity_msat, self.decay_params);
return dir_liq.liquidity_history.calculate_success_probability_times_billion( return dir_liq.liquidity_history.calculate_success_probability_times_billion(
dir_liq.now, *dir_liq.last_updated, dir_liq.now, *dir_liq.offset_history_last_updated,
self.decay_params.historical_no_updates_half_life, &params, amount_msat, self.decay_params.historical_no_updates_half_life, &params, amount_msat,
capacity_msat capacity_msat
).map(|p| p as f64 / (1024 * 1024 * 1024) as f64); ).map(|p| p as f64 / (1024 * 1024 * 1024) as f64);
@ -1008,6 +1012,7 @@ impl<T: Time> ChannelLiquidity<T> {
min_liquidity_offset_history: HistoricalBucketRangeTracker::new(), min_liquidity_offset_history: HistoricalBucketRangeTracker::new(),
max_liquidity_offset_history: HistoricalBucketRangeTracker::new(), max_liquidity_offset_history: HistoricalBucketRangeTracker::new(),
last_updated: T::now(), last_updated: T::now(),
offset_history_last_updated: T::now(),
} }
} }
@ -1034,6 +1039,7 @@ impl<T: Time> ChannelLiquidity<T> {
}, },
capacity_msat, capacity_msat,
last_updated: &self.last_updated, last_updated: &self.last_updated,
offset_history_last_updated: &self.offset_history_last_updated,
now: T::now(), now: T::now(),
decay_params: decay_params, decay_params: decay_params,
} }
@ -1062,6 +1068,7 @@ impl<T: Time> ChannelLiquidity<T> {
}, },
capacity_msat, capacity_msat,
last_updated: &mut self.last_updated, last_updated: &mut self.last_updated,
offset_history_last_updated: &mut self.offset_history_last_updated,
now: T::now(), now: T::now(),
decay_params: decay_params, decay_params: decay_params,
} }
@ -1197,7 +1204,8 @@ impl<L: Deref<Target = u64>, BRT: Deref<Target = HistoricalBucketRangeTracker>,
if score_params.historical_liquidity_penalty_multiplier_msat != 0 || if score_params.historical_liquidity_penalty_multiplier_msat != 0 ||
score_params.historical_liquidity_penalty_amount_multiplier_msat != 0 { score_params.historical_liquidity_penalty_amount_multiplier_msat != 0 {
if let Some(cumulative_success_prob_times_billion) = self.liquidity_history if let Some(cumulative_success_prob_times_billion) = self.liquidity_history
.calculate_success_probability_times_billion(self.now, *self.last_updated, .calculate_success_probability_times_billion(
self.now, *self.offset_history_last_updated,
self.decay_params.historical_no_updates_half_life, score_params, amount_msat, self.decay_params.historical_no_updates_half_life, score_params, amount_msat,
self.capacity_msat) self.capacity_msat)
{ {
@ -1316,7 +1324,7 @@ impl<L: DerefMut<Target = u64>, BRT: DerefMut<Target = HistoricalBucketRangeTrac
/// state"), we allow the caller to set an offset applied to our liquidity bounds which /// state"), we allow the caller to set an offset applied to our liquidity bounds which
/// represents the amount of the successful payment we just made. /// represents the amount of the successful payment we just made.
fn update_history_buckets(&mut self, bucket_offset_msat: u64) { fn update_history_buckets(&mut self, bucket_offset_msat: u64) {
let half_lives = self.now.duration_since(*self.last_updated).as_secs() let half_lives = self.now.duration_since(*self.offset_history_last_updated).as_secs()
.checked_div(self.decay_params.historical_no_updates_half_life.as_secs()) .checked_div(self.decay_params.historical_no_updates_half_life.as_secs())
.map(|v| v.try_into().unwrap_or(u32::max_value())).unwrap_or(u32::max_value()); .map(|v| v.try_into().unwrap_or(u32::max_value())).unwrap_or(u32::max_value());
self.liquidity_history.min_liquidity_offset_history.time_decay_data(half_lives); self.liquidity_history.min_liquidity_offset_history.time_decay_data(half_lives);
@ -1341,6 +1349,7 @@ impl<L: DerefMut<Target = u64>, BRT: DerefMut<Target = HistoricalBucketRangeTrac
self.decayed_offset_msat(*self.max_liquidity_offset_msat) self.decayed_offset_msat(*self.max_liquidity_offset_msat)
}; };
*self.last_updated = self.now; *self.last_updated = self.now;
*self.offset_history_last_updated = self.now;
} }
/// Adjusts the upper bound of the channel liquidity balance in this direction. /// Adjusts the upper bound of the channel liquidity balance in this direction.
@ -1352,6 +1361,7 @@ impl<L: DerefMut<Target = u64>, BRT: DerefMut<Target = HistoricalBucketRangeTrac
self.decayed_offset_msat(*self.min_liquidity_offset_msat) self.decayed_offset_msat(*self.min_liquidity_offset_msat)
}; };
*self.last_updated = self.now; *self.last_updated = self.now;
*self.offset_history_last_updated = self.now;
} }
} }
@ -1983,9 +1993,9 @@ mod bucketed_history {
} }
impl<D: Deref<Target = HistoricalBucketRangeTracker>> HistoricalMinMaxBuckets<D> { impl<D: Deref<Target = HistoricalBucketRangeTracker>> HistoricalMinMaxBuckets<D> {
pub(super) fn get_decayed_buckets<T: Time>(&self, now: T, last_updated: T, half_life: Duration) pub(super) fn get_decayed_buckets<T: Time>(&self, now: T, offset_history_last_updated: T, half_life: Duration)
-> Option<([u16; 32], [u16; 32])> { -> Option<([u16; 32], [u16; 32])> {
let (_, required_decays) = self.get_total_valid_points(now, last_updated, half_life)?; let (_, required_decays) = self.get_total_valid_points(now, offset_history_last_updated, half_life)?;
let mut min_buckets = *self.min_liquidity_offset_history; let mut min_buckets = *self.min_liquidity_offset_history;
min_buckets.time_decay_data(required_decays); min_buckets.time_decay_data(required_decays);
@ -1994,9 +2004,9 @@ mod bucketed_history {
Some((min_buckets.buckets, max_buckets.buckets)) Some((min_buckets.buckets, max_buckets.buckets))
} }
#[inline] #[inline]
pub(super) fn get_total_valid_points<T: Time>(&self, now: T, last_updated: T, half_life: Duration) pub(super) fn get_total_valid_points<T: Time>(&self, now: T, offset_history_last_updated: T, half_life: Duration)
-> Option<(u64, u32)> { -> Option<(u64, u32)> {
let required_decays = now.duration_since(last_updated).as_secs() let required_decays = now.duration_since(offset_history_last_updated).as_secs()
.checked_div(half_life.as_secs()) .checked_div(half_life.as_secs())
.map_or(u32::max_value(), |decays| cmp::min(decays, u32::max_value() as u64) as u32); .map_or(u32::max_value(), |decays| cmp::min(decays, u32::max_value() as u64) as u32);
@ -2019,7 +2029,7 @@ mod bucketed_history {
#[inline] #[inline]
pub(super) fn calculate_success_probability_times_billion<T: Time>( pub(super) fn calculate_success_probability_times_billion<T: Time>(
&self, now: T, last_updated: T, half_life: Duration, &self, now: T, offset_history_last_updated: T, half_life: Duration,
params: &ProbabilisticScoringFeeParameters, amount_msat: u64, capacity_msat: u64 params: &ProbabilisticScoringFeeParameters, amount_msat: u64, capacity_msat: u64
) -> Option<u64> { ) -> Option<u64> {
// If historical penalties are enabled, we try to calculate a probability of success // If historical penalties are enabled, we try to calculate a probability of success
@ -2035,7 +2045,7 @@ mod bucketed_history {
// Check if all our buckets are zero, once decayed and treat it as if we had no data. We // Check if all our buckets are zero, once decayed and treat it as if we had no data. We
// don't actually use the decayed buckets, though, as that would lose precision. // don't actually use the decayed buckets, though, as that would lose precision.
let (total_valid_points_tracked, _) let (total_valid_points_tracked, _)
= self.get_total_valid_points(now, last_updated, half_life)?; = self.get_total_valid_points(now, offset_history_last_updated, half_life)?;
let mut cumulative_success_prob_times_billion = 0; let mut cumulative_success_prob_times_billion = 0;
// Special-case the 0th min bucket - it generally means we failed a payment, so only // Special-case the 0th min bucket - it generally means we failed a payment, so only
@ -2128,6 +2138,8 @@ ReadableArgs<(ProbabilisticScoringDecayParameters, G, L)> for ProbabilisticScore
impl<T: Time> Writeable for ChannelLiquidity<T> { impl<T: Time> Writeable for ChannelLiquidity<T> {
#[inline] #[inline]
fn write<W: Writer>(&self, w: &mut W) -> Result<(), io::Error> { fn write<W: Writer>(&self, w: &mut W) -> Result<(), io::Error> {
let offset_history_duration_since_epoch =
T::duration_since_epoch() - self.offset_history_last_updated.elapsed();
let duration_since_epoch = T::duration_since_epoch() - self.last_updated.elapsed(); let duration_since_epoch = T::duration_since_epoch() - self.last_updated.elapsed();
write_tlv_fields!(w, { write_tlv_fields!(w, {
(0, self.min_liquidity_offset_msat, required), (0, self.min_liquidity_offset_msat, required),
@ -2137,6 +2149,7 @@ impl<T: Time> Writeable for ChannelLiquidity<T> {
(4, duration_since_epoch, required), (4, duration_since_epoch, required),
(5, Some(self.min_liquidity_offset_history), option), (5, Some(self.min_liquidity_offset_history), option),
(7, Some(self.max_liquidity_offset_history), option), (7, Some(self.max_liquidity_offset_history), option),
(9, offset_history_duration_since_epoch, required),
}); });
Ok(()) Ok(())
} }
@ -2152,6 +2165,7 @@ impl<T: Time> Readable for ChannelLiquidity<T> {
let mut min_liquidity_offset_history: Option<HistoricalBucketRangeTracker> = None; let mut min_liquidity_offset_history: Option<HistoricalBucketRangeTracker> = None;
let mut max_liquidity_offset_history: Option<HistoricalBucketRangeTracker> = None; let mut max_liquidity_offset_history: Option<HistoricalBucketRangeTracker> = None;
let mut duration_since_epoch = Duration::from_secs(0); let mut duration_since_epoch = Duration::from_secs(0);
let mut offset_history_duration_since_epoch = None;
read_tlv_fields!(r, { read_tlv_fields!(r, {
(0, min_liquidity_offset_msat, required), (0, min_liquidity_offset_msat, required),
(1, legacy_min_liq_offset_history, option), (1, legacy_min_liq_offset_history, option),
@ -2160,6 +2174,7 @@ impl<T: Time> Readable for ChannelLiquidity<T> {
(4, duration_since_epoch, required), (4, duration_since_epoch, required),
(5, min_liquidity_offset_history, option), (5, min_liquidity_offset_history, option),
(7, max_liquidity_offset_history, option), (7, max_liquidity_offset_history, option),
(9, offset_history_duration_since_epoch, option),
}); });
// On rust prior to 1.60 `Instant::duration_since` will panic if time goes backwards. // On rust prior to 1.60 `Instant::duration_since` will panic if time goes backwards.
// We write `last_updated` as wallclock time even though its ultimately an `Instant` (which // We write `last_updated` as wallclock time even though its ultimately an `Instant` (which
@ -2173,6 +2188,13 @@ impl<T: Time> Readable for ChannelLiquidity<T> {
let last_updated = if wall_clock_now > duration_since_epoch { let last_updated = if wall_clock_now > duration_since_epoch {
now - (wall_clock_now - duration_since_epoch) now - (wall_clock_now - duration_since_epoch)
} else { now }; } else { now };
let offset_history_duration_since_epoch =
offset_history_duration_since_epoch.unwrap_or(duration_since_epoch);
let offset_history_last_updated = if wall_clock_now > offset_history_duration_since_epoch {
now - (wall_clock_now - offset_history_duration_since_epoch)
} else { now };
if min_liquidity_offset_history.is_none() { if min_liquidity_offset_history.is_none() {
if let Some(legacy_buckets) = legacy_min_liq_offset_history { if let Some(legacy_buckets) = legacy_min_liq_offset_history {
min_liquidity_offset_history = Some(legacy_buckets.into_current()); min_liquidity_offset_history = Some(legacy_buckets.into_current());
@ -2193,6 +2215,7 @@ impl<T: Time> Readable for ChannelLiquidity<T> {
min_liquidity_offset_history: min_liquidity_offset_history.unwrap(), min_liquidity_offset_history: min_liquidity_offset_history.unwrap(),
max_liquidity_offset_history: max_liquidity_offset_history.unwrap(), max_liquidity_offset_history: max_liquidity_offset_history.unwrap(),
last_updated, last_updated,
offset_history_last_updated,
}) })
} }
} }
@ -2368,18 +2391,21 @@ mod tests {
fn liquidity_bounds_directed_from_lowest_node_id() { fn liquidity_bounds_directed_from_lowest_node_id() {
let logger = TestLogger::new(); let logger = TestLogger::new();
let last_updated = SinceEpoch::now(); let last_updated = SinceEpoch::now();
let offset_history_last_updated = SinceEpoch::now();
let network_graph = network_graph(&logger); let network_graph = network_graph(&logger);
let decay_params = ProbabilisticScoringDecayParameters::default(); let decay_params = ProbabilisticScoringDecayParameters::default();
let mut scorer = ProbabilisticScorer::new(decay_params, &network_graph, &logger) let mut scorer = ProbabilisticScorer::new(decay_params, &network_graph, &logger)
.with_channel(42, .with_channel(42,
ChannelLiquidity { ChannelLiquidity {
min_liquidity_offset_msat: 700, max_liquidity_offset_msat: 100, last_updated, min_liquidity_offset_msat: 700, max_liquidity_offset_msat: 100,
last_updated, offset_history_last_updated,
min_liquidity_offset_history: HistoricalBucketRangeTracker::new(), min_liquidity_offset_history: HistoricalBucketRangeTracker::new(),
max_liquidity_offset_history: HistoricalBucketRangeTracker::new(), max_liquidity_offset_history: HistoricalBucketRangeTracker::new(),
}) })
.with_channel(43, .with_channel(43,
ChannelLiquidity { ChannelLiquidity {
min_liquidity_offset_msat: 700, max_liquidity_offset_msat: 100, last_updated, min_liquidity_offset_msat: 700, max_liquidity_offset_msat: 100,
last_updated, offset_history_last_updated,
min_liquidity_offset_history: HistoricalBucketRangeTracker::new(), min_liquidity_offset_history: HistoricalBucketRangeTracker::new(),
max_liquidity_offset_history: HistoricalBucketRangeTracker::new(), max_liquidity_offset_history: HistoricalBucketRangeTracker::new(),
}); });
@ -2446,12 +2472,14 @@ mod tests {
fn resets_liquidity_upper_bound_when_crossed_by_lower_bound() { fn resets_liquidity_upper_bound_when_crossed_by_lower_bound() {
let logger = TestLogger::new(); let logger = TestLogger::new();
let last_updated = SinceEpoch::now(); let last_updated = SinceEpoch::now();
let offset_history_last_updated = SinceEpoch::now();
let network_graph = network_graph(&logger); let network_graph = network_graph(&logger);
let decay_params = ProbabilisticScoringDecayParameters::default(); let decay_params = ProbabilisticScoringDecayParameters::default();
let mut scorer = ProbabilisticScorer::new(decay_params, &network_graph, &logger) let mut scorer = ProbabilisticScorer::new(decay_params, &network_graph, &logger)
.with_channel(42, .with_channel(42,
ChannelLiquidity { ChannelLiquidity {
min_liquidity_offset_msat: 200, max_liquidity_offset_msat: 400, last_updated, min_liquidity_offset_msat: 200, max_liquidity_offset_msat: 400,
last_updated, offset_history_last_updated,
min_liquidity_offset_history: HistoricalBucketRangeTracker::new(), min_liquidity_offset_history: HistoricalBucketRangeTracker::new(),
max_liquidity_offset_history: HistoricalBucketRangeTracker::new(), max_liquidity_offset_history: HistoricalBucketRangeTracker::new(),
}); });
@ -2505,12 +2533,14 @@ mod tests {
fn resets_liquidity_lower_bound_when_crossed_by_upper_bound() { fn resets_liquidity_lower_bound_when_crossed_by_upper_bound() {
let logger = TestLogger::new(); let logger = TestLogger::new();
let last_updated = SinceEpoch::now(); let last_updated = SinceEpoch::now();
let offset_history_last_updated = SinceEpoch::now();
let network_graph = network_graph(&logger); let network_graph = network_graph(&logger);
let decay_params = ProbabilisticScoringDecayParameters::default(); let decay_params = ProbabilisticScoringDecayParameters::default();
let mut scorer = ProbabilisticScorer::new(decay_params, &network_graph, &logger) let mut scorer = ProbabilisticScorer::new(decay_params, &network_graph, &logger)
.with_channel(42, .with_channel(42,
ChannelLiquidity { ChannelLiquidity {
min_liquidity_offset_msat: 200, max_liquidity_offset_msat: 400, last_updated, min_liquidity_offset_msat: 200, max_liquidity_offset_msat: 400,
last_updated, offset_history_last_updated,
min_liquidity_offset_history: HistoricalBucketRangeTracker::new(), min_liquidity_offset_history: HistoricalBucketRangeTracker::new(),
max_liquidity_offset_history: HistoricalBucketRangeTracker::new(), max_liquidity_offset_history: HistoricalBucketRangeTracker::new(),
}); });
@ -2616,6 +2646,7 @@ mod tests {
fn constant_penalty_outside_liquidity_bounds() { fn constant_penalty_outside_liquidity_bounds() {
let logger = TestLogger::new(); let logger = TestLogger::new();
let last_updated = SinceEpoch::now(); let last_updated = SinceEpoch::now();
let offset_history_last_updated = SinceEpoch::now();
let network_graph = network_graph(&logger); let network_graph = network_graph(&logger);
let params = ProbabilisticScoringFeeParameters { let params = ProbabilisticScoringFeeParameters {
liquidity_penalty_multiplier_msat: 1_000, liquidity_penalty_multiplier_msat: 1_000,
@ -2628,7 +2659,8 @@ mod tests {
let scorer = ProbabilisticScorer::new(decay_params, &network_graph, &logger) let scorer = ProbabilisticScorer::new(decay_params, &network_graph, &logger)
.with_channel(42, .with_channel(42,
ChannelLiquidity { ChannelLiquidity {
min_liquidity_offset_msat: 40, max_liquidity_offset_msat: 40, last_updated, min_liquidity_offset_msat: 40, max_liquidity_offset_msat: 40,
last_updated, offset_history_last_updated,
min_liquidity_offset_history: HistoricalBucketRangeTracker::new(), min_liquidity_offset_history: HistoricalBucketRangeTracker::new(),
max_liquidity_offset_history: HistoricalBucketRangeTracker::new(), max_liquidity_offset_history: HistoricalBucketRangeTracker::new(),
}); });