Track history of where channel liquidities have been in the past

This introduces two new fields to the `ChannelLiquidity` struct -
`min_liquidity_offset_history` and `max_liquidity_offset_history`,
both an array of 8 `u16`s. Each entry represents the proportion of
time that we spent with the min or max liquidity offset in the
given 1/8th of the channel's liquidity range. ie the first bucket
in `min_liquidity_offset_history` represents the proportion of time
we've thought the channel's minimum liquidity is lower than 1/8th's
the channel's capacity.

Each bucket is stored, effectively, as a fixed-point number with
5 bits for the fractional part, which is incremented by one (ie 32)
each time we update our liquidity estimates and decide our
estimates are in that bucket. We then decay each bucket by
2047/2048.

Thus, memory of a payment sticks around for more than 8,000
data points, though the majority of that memory decays after 1,387
data points.

Author: TheBlueMatt

lightning/src/routing/scoring.rs

@@ -64,6 +64,7 @@ use util::time::Time;
 use prelude::*;
 use core::fmt;
 use core::cell::{RefCell, RefMut};
+use core::convert::TryInto;
 use core::ops::{Deref, DerefMut};
 use core::time::Duration;
 use io::{self, Read};
@@ -446,13 +447,18 @@ struct ChannelLiquidity<T: Time> {
 
 	/// Time when the liquidity bounds were last modified.
 	last_updated: T,
+
+	min_liquidity_offset_history: [u16; 8],
+	max_liquidity_offset_history: [u16; 8],
 }
 
 /// A snapshot of [`ChannelLiquidity`] in one direction assuming a certain channel capacity and
 /// decayed with a given half life.
-struct DirectedChannelLiquidity<L: Deref<Target = u64>, T: Time, U: Deref<Target = T>> {
+struct DirectedChannelLiquidity<L: Deref<Target = u64>, LA: Deref<Target = [u16; 8]>, T: Time, U: Deref<Target = T>> {
 	min_liquidity_offset_msat: L,
 	max_liquidity_offset_msat: L,
+	min_liquidity_offset_history: LA,
+	max_liquidity_offset_history: LA,
 	capacity_msat: u64,
 	last_updated: U,
 	now: T,
@@ -593,6 +599,8 @@ impl<T: Time> ChannelLiquidity<T> {
 		Self {
 			min_liquidity_offset_msat: 0,
 			max_liquidity_offset_msat: 0,
+			min_liquidity_offset_history: [0; 8],
+			max_liquidity_offset_history: [0; 8],
 			last_updated: T::now(),
 		}
 	}
@@ -601,16 +609,21 @@ impl<T: Time> ChannelLiquidity<T> {
 	/// `capacity_msat`.
 	fn as_directed(
 		&self, source: &NodeId, target: &NodeId, capacity_msat: u64, half_life: Duration
-	) -> DirectedChannelLiquidity<&u64, T, &T> {
-		let (min_liquidity_offset_msat, max_liquidity_offset_msat) = if source < target {
-			(&self.min_liquidity_offset_msat, &self.max_liquidity_offset_msat)
-		} else {
-			(&self.max_liquidity_offset_msat, &self.min_liquidity_offset_msat)
-		};
+	) -> DirectedChannelLiquidity<&u64, &[u16; 8], T, &T> {
+		let (min_liquidity_offset_msat, max_liquidity_offset_msat, min_liquidity_offset_history, max_liquidity_offset_history) =
+			if source < target {
+				(&self.min_liquidity_offset_msat, &self.max_liquidity_offset_msat,
+					&self.min_liquidity_offset_history, &self.max_liquidity_offset_history)
+			} else {
+				(&self.max_liquidity_offset_msat, &self.min_liquidity_offset_msat,
+					&self.max_liquidity_offset_history, &self.min_liquidity_offset_history)
+			};
 
 		DirectedChannelLiquidity {
 			min_liquidity_offset_msat,
 			max_liquidity_offset_msat,
+			min_liquidity_offset_history,
+			max_liquidity_offset_history,
 			capacity_msat,
 			last_updated: &self.last_updated,
 			now: T::now(),
@@ -622,16 +635,21 @@ impl<T: Time> ChannelLiquidity<T> {
 	/// `capacity_msat`.
 	fn as_directed_mut(
 		&mut self, source: &NodeId, target: &NodeId, capacity_msat: u64, half_life: Duration
-	) -> DirectedChannelLiquidity<&mut u64, T, &mut T> {
-		let (min_liquidity_offset_msat, max_liquidity_offset_msat) = if source < target {
-			(&mut self.min_liquidity_offset_msat, &mut self.max_liquidity_offset_msat)
-		} else {
-			(&mut self.max_liquidity_offset_msat, &mut self.min_liquidity_offset_msat)
-		};
+	) -> DirectedChannelLiquidity<&mut u64, &mut [u16; 8], T, &mut T> {
+		let (min_liquidity_offset_msat, max_liquidity_offset_msat, min_liquidity_offset_history, max_liquidity_offset_history) =
+			if source < target {
+				(&mut self.min_liquidity_offset_msat, &mut self.max_liquidity_offset_msat,
+					&mut self.min_liquidity_offset_history, &mut self.max_liquidity_offset_history)
+			} else {
+				(&mut self.max_liquidity_offset_msat, &mut self.min_liquidity_offset_msat,
+					&mut self.max_liquidity_offset_history, &mut self.min_liquidity_offset_history)
+			};
 
 		DirectedChannelLiquidity {
 			min_liquidity_offset_msat,
 			max_liquidity_offset_msat,
+			min_liquidity_offset_history,
+			max_liquidity_offset_history,
 			capacity_msat,
 			last_updated: &mut self.last_updated,
 			now: T::now(),
@@ -652,7 +670,7 @@ const PRECISION_LOWER_BOUND_DENOMINATOR: u64 = approx::LOWER_BITS_BOUND;
 const AMOUNT_PENALTY_DIVISOR: u64 = 1 << 20;
 const BASE_AMOUNT_PENALTY_DIVISOR: u64 = 1 << 30;
 
-impl<L: Deref<Target = u64>, T: Time, U: Deref<Target = T>> DirectedChannelLiquidity<L, T, U> {
+impl<L: Deref<Target = u64>, LA: Deref<Target = [u16; 8]>, T: Time, U: Deref<Target = T>> DirectedChannelLiquidity<L, LA, T, U> {
 	/// Returns a liquidity penalty for routing the given HTLC `amount_msat` through the channel in
 	/// this direction.
 	fn penalty_msat(&self, amount_msat: u64, params: &ProbabilisticScoringParameters) -> u64 {
@@ -722,7 +740,7 @@ impl<L: Deref<Target = u64>, T: Time, U: Deref<Target = T>> DirectedChannelLiqui
 	}
 }
 
-impl<L: DerefMut<Target = u64>, T: Time, U: DerefMut<Target = T>> DirectedChannelLiquidity<L, T, U> {
+impl<L: DerefMut<Target = u64>, LA: DerefMut<Target = [u16; 8]>, T: Time, U: DerefMut<Target = T>> DirectedChannelLiquidity<L, LA, T, U> {
 	/// Adjusts the channel liquidity balance bounds when failing to route `amount_msat`.
 	fn failed_at_channel<Log: Deref>(&mut self, amount_msat: u64, chan_descr: fmt::Arguments, logger: &Log) where Log::Target: Logger {
 		if amount_msat < self.max_liquidity_msat() {
@@ -750,6 +768,41 @@ impl<L: DerefMut<Target = u64>, T: Time, U: DerefMut<Target = T>> DirectedChanne
 		self.set_max_liquidity_msat(max_liquidity_msat);
 	}
 
+	fn update_history_buckets(&mut self) {
+		// We have 8 leaky buckets for min and max liquidity. Each bucket tracks the amount of time
+		// we spend in each bucket as a fixed-point number with 5 bits for the fractional part.
+		//
+		// Each time we update our liquidity estimate, we add 32 to the buckets for the current
+		// min and max liquidity offset positions.
+		//
+		// We then decay each bucket by multiplying by 2047/2048. This ensures we can't actually
+		// reach 65,535 - when we get to 63,457 adding 32 and decaying by 2047/2048 leaves us back
+		// at 63,457.
+		//
+		// In total, this allows us to track data for the last 8,000 or so payments across a given
+		// channel.
+		debug_assert!(*self.min_liquidity_offset_msat <= self.capacity_msat);
+		if let Some(v) = self.min_liquidity_offset_history
+			.get_mut((self.min_liquidity_offset_msat.saturating_sub(1) * 8 / self.capacity_msat)
+				.try_into().unwrap_or(32))
+		{
+			*v = v.saturating_add(8);
+			for e in self.min_liquidity_offset_history.iter_mut() {
+				*e = ((*e as u64) * 2047 / 2048) as u16;
+			}
+		}
+		debug_assert!(*self.max_liquidity_offset_msat <= self.capacity_msat);
+		if let Some(v) = self.max_liquidity_offset_history
+			.get_mut((self.max_liquidity_offset_msat.saturating_sub(1) * 8 / self.capacity_msat)
+				.try_into().unwrap_or(32))
+		{
+			*v = v.saturating_add(8);
+			for e in self.max_liquidity_offset_history.iter_mut() {
+				*e = ((*e as u64) * 2047 / 2048) as u16;
+			}
+		}
+	}
+
 	/// Adjusts the lower bound of the channel liquidity balance in this direction.
 	fn set_min_liquidity_msat(&mut self, amount_msat: u64) {
 		*self.min_liquidity_offset_msat = amount_msat;
@@ -759,6 +812,7 @@ impl<L: DerefMut<Target = u64>, T: Time, U: DerefMut<Target = T>> DirectedChanne
 			self.decayed_offset_msat(*self.max_liquidity_offset_msat)
 		};
 		*self.last_updated = self.now;
+		self.update_history_buckets();
 	}
 
 	/// Adjusts the upper bound of the channel liquidity balance in this direction.
@@ -770,6 +824,7 @@ impl<L: DerefMut<Target = u64>, T: Time, U: DerefMut<Target = T>> DirectedChanne
 			self.decayed_offset_msat(*self.min_liquidity_offset_msat)
 		};
 		*self.last_updated = self.now;
+		self.update_history_buckets();
 	}
 }
 
@@ -1236,7 +1291,9 @@ impl<T: Time> Writeable for ChannelLiquidity<T> {
 		let duration_since_epoch = T::duration_since_epoch() - self.last_updated.elapsed();
 		write_tlv_fields!(w, {
 			(0, self.min_liquidity_offset_msat, required),
+			(1, Some(self.min_liquidity_offset_history), option),
 			(2, self.max_liquidity_offset_msat, required),
+			(3, Some(self.max_liquidity_offset_history), option),
 			(4, duration_since_epoch, required),
 		});
 		Ok(())
@@ -1248,10 +1305,14 @@ impl<T: Time> Readable for ChannelLiquidity<T> {
 	fn read<R: Read>(r: &mut R) -> Result<Self, DecodeError> {
 		let mut min_liquidity_offset_msat = 0;
 		let mut max_liquidity_offset_msat = 0;
+		let mut min_liquidity_offset_history = Some([0; 8]);
+		let mut max_liquidity_offset_history = Some([0; 8]);
 		let mut duration_since_epoch = Duration::from_secs(0);
 		read_tlv_fields!(r, {
 			(0, min_liquidity_offset_msat, required),
+			(1, min_liquidity_offset_history, option),
 			(2, max_liquidity_offset_msat, required),
+			(3, max_liquidity_offset_history, option),
 			(4, duration_since_epoch, required),
 		});
 		// On rust prior to 1.60 `Instant::duration_since` will panic if time goes backwards.
@@ -1269,6 +1330,8 @@ impl<T: Time> Readable for ChannelLiquidity<T> {
 		Ok(Self {
 			min_liquidity_offset_msat,
 			max_liquidity_offset_msat,
+			min_liquidity_offset_history: min_liquidity_offset_history.unwrap(),
+			max_liquidity_offset_history: max_liquidity_offset_history.unwrap(),
 			last_updated,
 		})
 	}
@@ -1459,11 +1522,13 @@ mod tests {
 		let mut scorer = ProbabilisticScorer::new(params, &network_graph, &logger)
 			.with_channel(42,
 				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,
+					min_liquidity_offset_history: [0; 8], max_liquidity_offset_history: [0; 8],
 				})
 			.with_channel(43,
 				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,
+					min_liquidity_offset_history: [0; 8], max_liquidity_offset_history: [0; 8],
 				});
 		let source = source_node_id();
 		let target = target_node_id();
@@ -1534,7 +1599,8 @@ mod tests {
 		let mut scorer = ProbabilisticScorer::new(params, &network_graph, &logger)
 			.with_channel(42,
 				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,
+					min_liquidity_offset_history: [0; 8], max_liquidity_offset_history: [0; 8],
 				});
 		let source = source_node_id();
 		let target = target_node_id();
@@ -1592,7 +1658,8 @@ mod tests {
 		let mut scorer = ProbabilisticScorer::new(params, &network_graph, &logger)
 			.with_channel(42,
 				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,
+					min_liquidity_offset_history: [0; 8], max_liquidity_offset_history: [0; 8],
 				});
 		let source = source_node_id();
 		let target = target_node_id();
@@ -1699,7 +1766,8 @@ mod tests {
 		let scorer = ProbabilisticScorer::new(params, &network_graph, &logger)
 			.with_channel(42,
 				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,
+					min_liquidity_offset_history: [0; 8], max_liquidity_offset_history: [0; 8],
 				});
 		let source = source_node_id();
 		let target = target_node_id();

lightning/src/util/ser.rs

@@ -515,6 +515,29 @@ impl_array!(PUBLIC_KEY_SIZE); // for PublicKey
 impl_array!(COMPACT_SIGNATURE_SIZE); // for Signature
 impl_array!(1300); // for OnionPacket.hop_data
 
+impl Writeable for [u16; 8] {
+	#[inline]
+	fn write<W: Writer>(&self, w: &mut W) -> Result<(), io::Error> {
+		for v in self.iter() {
+			w.write_all(&v.to_be_bytes())?
+		}
+		Ok(())
+	}
+}
+
+impl Readable for [u16; 8] {
+	#[inline]
+	fn read<R: Read>(r: &mut R) -> Result<Self, DecodeError> {
+		let mut buf = [0u8; 16];
+		r.read_exact(&mut buf)?;
+		let mut res = [0u16; 8];
+		for (idx, v) in res.iter_mut().enumerate() {
+			*v = (buf[idx] as u16) << 8 | (buf[idx + 1] as u16)
+		}
+		Ok(res)
+	}
+}
+
 // HashMap
 impl<K, V> Writeable for HashMap<K, V>
 	where K: Writeable + Eq + Hash,