Merge pull request #223 from TheBlueMatt/2018-10-chanmanager-serialize

Implement and document Channel/ChannelManager (de)serialization

fuzz/fuzz_targets/chanmon_deser_target.rs

@@ -1,13 +1,20 @@
 // This file is auto-generated by gen_target.sh based on msg_target_template.txt
 // To modify it, modify msg_target_template.txt and run gen_target.sh instead.
 
+extern crate bitcoin;
 extern crate lightning;
 
+use bitcoin::util::hash::Sha256dHash;
+
 use lightning::ln::channelmonitor;
 use lightning::util::reset_rng_state;
-use lightning::util::ser::{Readable, Writer};
+use lightning::util::ser::{ReadableArgs, Writer};
+
+mod utils;
+use utils::test_logger;
 
 use std::io::Cursor;
+use std::sync::Arc;
 
 struct VecWriter(Vec<u8>);
 impl Writer for VecWriter {
@@ -23,10 +30,13 @@ impl Writer for VecWriter {
 #[inline]
 pub fn do_test(data: &[u8]) {
 	reset_rng_state();
-	if let Ok(monitor) = channelmonitor::ChannelMonitor::read(&mut Cursor::new(data)) {
+	let logger = Arc::new(test_logger::TestLogger{});
+	if let Ok((latest_block_hash, monitor)) = <(Sha256dHash, channelmonitor::ChannelMonitor)>::read(&mut Cursor::new(data), logger.clone()) {
 		let mut w = VecWriter(Vec::new());
 		monitor.write_for_disk(&mut w).unwrap();
-		assert!(channelmonitor::ChannelMonitor::read(&mut Cursor::new(&w.0)).unwrap() == monitor);
+		let deserialized_copy = <(Sha256dHash, channelmonitor::ChannelMonitor)>::read(&mut Cursor::new(&w.0), logger.clone()).unwrap();
+		assert!(latest_block_hash == deserialized_copy.0);
+		assert!(monitor == deserialized_copy.1);
 		w.0.clear();
 		monitor.write_for_watchtower(&mut w).unwrap();
 	}

src/chain/keysinterface.rs

@@ -78,6 +78,15 @@ pub struct ChannelKeys {
 	pub commitment_seed: [u8; 32],
 }
 
+impl_writeable!(ChannelKeys, 0, {
+	funding_key,
+	revocation_base_key,
+	payment_base_key,
+	delayed_payment_base_key,
+	htlc_base_key,
+	commitment_seed
+});
+
 impl ChannelKeys {
 	/// Generate a set of lightning keys needed to operate a channel by HKDF-expanding a given
 	/// random 32-byte seed

src/ln/channel.rs

@@ -4,7 +4,9 @@ use bitcoin::blockdata::transaction::{TxIn, TxOut, Transaction, SigHashType};
 use bitcoin::blockdata::opcodes;
 use bitcoin::util::hash::{Sha256dHash, Hash160};
 use bitcoin::util::bip143;
-use bitcoin::network::serialize::BitcoinHash;
+use bitcoin::network;
+use bitcoin::network::serialize::{BitcoinHash, RawDecoder, RawEncoder};
+use bitcoin::network::encodable::{ConsensusEncodable, ConsensusDecodable};
 
 use secp256k1::key::{PublicKey,SecretKey};
 use secp256k1::{Secp256k1,Message,Signature};
@@ -13,7 +15,7 @@ use secp256k1;
 use crypto::digest::Digest;
 
 use ln::msgs;
-use ln::msgs::{ErrorAction, HandleError};
+use ln::msgs::{DecodeError, ErrorAction, HandleError};
 use ln::channelmonitor::ChannelMonitor;
 use ln::channelmanager::{PendingHTLCStatus, HTLCSource, HTLCFailReason, HTLCFailureMsg, PendingForwardHTLCInfo, RAACommitmentOrder};
 use ln::chan_utils::{TxCreationKeys,HTLCOutputInCommitment,HTLC_SUCCESS_TX_WEIGHT,HTLC_TIMEOUT_TX_WEIGHT};
@@ -22,7 +24,7 @@ use chain::chaininterface::{FeeEstimator,ConfirmationTarget};
 use chain::transaction::OutPoint;
 use chain::keysinterface::{ChannelKeys, KeysInterface};
 use util::{transaction_utils,rng};
-use util::ser::Writeable;
+use util::ser::{Readable, ReadableArgs, Writeable, Writer, WriterWriteAdaptor};
 use util::sha2::Sha256;
 use util::logger::Logger;
 use util::errors::APIError;
@@ -306,8 +308,9 @@ pub(super) struct Channel {
 	/// could miss the funding_tx_confirmed_in block as well, but it serves as a useful fallback.
 	funding_tx_confirmed_in: Option<Sha256dHash>,
 	short_channel_id: Option<u64>,
-	/// Used to deduplicate block_connected callbacks
+	/// Used to deduplicate block_connected callbacks, also used to verify consistency during
-	last_block_connected: Sha256dHash,
+	/// ChannelManager deserialization (hence pub(super))
+	pub(super) last_block_connected: Sha256dHash,
 	funding_tx_confirmations: u64,
 
 	their_dust_limit_satoshis: u64,
@@ -438,7 +441,7 @@ impl Channel {
 		let secp_ctx = Secp256k1::new();
 		let channel_monitor = ChannelMonitor::new(&chan_keys.revocation_base_key, &chan_keys.delayed_payment_base_key,
 		                                          &chan_keys.htlc_base_key, BREAKDOWN_TIMEOUT,
-		                                          keys_provider.get_destination_script());
+		                                          keys_provider.get_destination_script(), logger.clone());
 
 		Ok(Channel {
 			user_id: user_id,
@@ -600,7 +603,7 @@ impl Channel {
 		let secp_ctx = Secp256k1::new();
 		let mut channel_monitor = ChannelMonitor::new(&chan_keys.revocation_base_key, &chan_keys.delayed_payment_base_key,
 		                                              &chan_keys.htlc_base_key, BREAKDOWN_TIMEOUT,
-		                                              keys_provider.get_destination_script());
+		                                              keys_provider.get_destination_script(), logger.clone());
 		channel_monitor.set_their_base_keys(&msg.htlc_basepoint, &msg.delayed_payment_basepoint);
 		channel_monitor.set_their_to_self_delay(msg.to_self_delay);
 
@@ -2560,6 +2563,18 @@ impl Channel {
 		self.feerate_per_kw
 	}
 
+	pub fn get_cur_local_commitment_transaction_number(&self) -> u64 {
+		self.cur_local_commitment_transaction_number + 1
+	}
+
+	pub fn get_cur_remote_commitment_transaction_number(&self) -> u64 {
+		self.cur_remote_commitment_transaction_number + 1 - if self.channel_state & (ChannelState::AwaitingRemoteRevoke as u32) != 0 { 1 } else { 0 }
+	}
+
+	pub fn get_revoked_remote_commitment_transaction_number(&self) -> u64 {
+		self.cur_remote_commitment_transaction_number + 2
+	}
+
 	//TODO: Testing purpose only, should be changed in another way after #81
 	#[cfg(test)]
 	pub fn get_local_keys(&self) -> &ChannelKeys {
@@ -2671,9 +2686,10 @@ impl Channel {
 	/// Only returns an ErrorAction of DisconnectPeer, if Err.
 	pub fn block_connected(&mut self, header: &BlockHeader, height: u32, txn_matched: &[&Transaction], indexes_of_txn_matched: &[u32]) -> Result<Option<msgs::FundingLocked>, HandleError> {
 		let non_shutdown_state = self.channel_state & (!MULTI_STATE_FLAGS);
-		if self.funding_tx_confirmations > 0 {
+		if header.bitcoin_hash() != self.last_block_connected {
-			if header.bitcoin_hash() != self.last_block_connected {
+			self.last_block_connected = header.bitcoin_hash();
-				self.last_block_connected = header.bitcoin_hash();
+			self.channel_monitor.last_block_hash = self.last_block_connected;
+			if self.funding_tx_confirmations > 0 {
 				self.funding_tx_confirmations += 1;
 				if self.funding_tx_confirmations == Channel::derive_minimum_depth(self.channel_value_satoshis*1000, self.value_to_self_msat) as u64 {
 					let need_commitment_update = if non_shutdown_state == ChannelState::FundingSent as u32 {
@@ -2754,6 +2770,8 @@ impl Channel {
 		if Some(header.bitcoin_hash()) == self.funding_tx_confirmed_in {
 			self.funding_tx_confirmations = Channel::derive_minimum_depth(self.channel_value_satoshis*1000, self.value_to_self_msat) as u64 - 1;
 		}
+		self.last_block_connected = header.bitcoin_hash();
+		self.channel_monitor.last_block_hash = self.last_block_connected;
 		false
 	}
 
@@ -3227,6 +3245,494 @@ impl Channel {
 	}
 }
 
+const SERIALIZATION_VERSION: u8 = 1;
+const MIN_SERIALIZATION_VERSION: u8 = 1;
+
+impl Writeable for InboundHTLCRemovalReason {
+	fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ::std::io::Error> {
+		match self {
+			&InboundHTLCRemovalReason::FailRelay(ref error_packet) => {
+				0u8.write(writer)?;
+				error_packet.write(writer)?;
+			},
+			&InboundHTLCRemovalReason::FailMalformed((ref onion_hash, ref err_code)) => {
+				1u8.write(writer)?;
+				onion_hash.write(writer)?;
+				err_code.write(writer)?;
+			},
+			&InboundHTLCRemovalReason::Fulfill(ref payment_preimage) => {
+				2u8.write(writer)?;
+				payment_preimage.write(writer)?;
+			},
+		}
+		Ok(())
+	}
+}
+
+impl<R: ::std::io::Read> Readable<R> for InboundHTLCRemovalReason {
+	fn read(reader: &mut R) -> Result<Self, DecodeError> {
+		Ok(match <u8 as Readable<R>>::read(reader)? {
+			0 => InboundHTLCRemovalReason::FailRelay(Readable::read(reader)?),
+			1 => InboundHTLCRemovalReason::FailMalformed((Readable::read(reader)?, Readable::read(reader)?)),
+			2 => InboundHTLCRemovalReason::Fulfill(Readable::read(reader)?),
+			_ => return Err(DecodeError::InvalidValue),
+		})
+	}
+}
+
+impl Writeable for Channel {
+	fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ::std::io::Error> {
+		// Note that we write out as if remove_uncommitted_htlcs_and_mark_paused had just been
+		// called but include holding cell updates (and obviously we don't modify self).
+
+		writer.write_all(&[SERIALIZATION_VERSION; 1])?;
+		writer.write_all(&[MIN_SERIALIZATION_VERSION; 1])?;
+
+		self.user_id.write(writer)?;
+
+		self.channel_id.write(writer)?;
+		(self.channel_state | ChannelState::PeerDisconnected as u32).write(writer)?;
+		self.channel_outbound.write(writer)?;
+		self.announce_publicly.write(writer)?;
+		self.channel_value_satoshis.write(writer)?;
+
+		self.local_keys.write(writer)?;
+		self.shutdown_pubkey.write(writer)?;
+
+		self.cur_local_commitment_transaction_number.write(writer)?;
+		self.cur_remote_commitment_transaction_number.write(writer)?;
+		self.value_to_self_msat.write(writer)?;
+
+		self.received_commitment_while_awaiting_raa.write(writer)?;
+
+		let mut dropped_inbound_htlcs = 0;
+		for htlc in self.pending_inbound_htlcs.iter() {
+			if let InboundHTLCState::RemoteAnnounced(_) = htlc.state {
+				dropped_inbound_htlcs += 1;
+			}
+		}
+		(self.pending_inbound_htlcs.len() as u64 - dropped_inbound_htlcs).write(writer)?;
+		for htlc in self.pending_inbound_htlcs.iter() {
+			htlc.htlc_id.write(writer)?;
+			htlc.amount_msat.write(writer)?;
+			htlc.cltv_expiry.write(writer)?;
+			htlc.payment_hash.write(writer)?;
+			match &htlc.state {
+				&InboundHTLCState::RemoteAnnounced(_) => {}, // Drop
+				&InboundHTLCState::AwaitingRemoteRevokeToAnnounce(ref htlc_state) => {
+					1u8.write(writer)?;
+					htlc_state.write(writer)?;
+				},
+				&InboundHTLCState::AwaitingAnnouncedRemoteRevoke(ref htlc_state) => {
+					2u8.write(writer)?;
+					htlc_state.write(writer)?;
+				},
+				&InboundHTLCState::Committed => {
+					3u8.write(writer)?;
+				},
+				&InboundHTLCState::LocalRemoved(ref removal_reason) => {
+					4u8.write(writer)?;
+					removal_reason.write(writer)?;
+				},
+			}
+		}
+
+		macro_rules! write_option {
+			($thing: expr) => {
+				match &$thing {
+					&None => 0u8.write(writer)?,
+					&Some(ref v) => {
+						1u8.write(writer)?;
+						v.write(writer)?;
+					},
+				}
+			}
+		}
+
+		(self.pending_outbound_htlcs.len() as u64).write(writer)?;
+		for htlc in self.pending_outbound_htlcs.iter() {
+			htlc.htlc_id.write(writer)?;
+			htlc.amount_msat.write(writer)?;
+			htlc.cltv_expiry.write(writer)?;
+			htlc.payment_hash.write(writer)?;
+			htlc.source.write(writer)?;
+			write_option!(htlc.fail_reason);
+			match &htlc.state {
+				&OutboundHTLCState::LocalAnnounced(ref onion_packet) => {
+					0u8.write(writer)?;
+					onion_packet.write(writer)?;
+				},
+				&OutboundHTLCState::Committed => {
+					1u8.write(writer)?;
+				},
+				&OutboundHTLCState::RemoteRemoved => {
+					2u8.write(writer)?;
+				},
+				&OutboundHTLCState::AwaitingRemoteRevokeToRemove => {
+					3u8.write(writer)?;
+				},
+				&OutboundHTLCState::AwaitingRemovedRemoteRevoke => {
+					4u8.write(writer)?;
+				},
+			}
+		}
+
+		(self.holding_cell_htlc_updates.len() as u64).write(writer)?;
+		for update in self.holding_cell_htlc_updates.iter() {
+			match update {
+				&HTLCUpdateAwaitingACK::AddHTLC { ref amount_msat, ref cltv_expiry, ref payment_hash, ref source, ref onion_routing_packet, time_created: _ } => {
+					0u8.write(writer)?;
+					amount_msat.write(writer)?;
+					cltv_expiry.write(writer)?;
+					payment_hash.write(writer)?;
+					source.write(writer)?;
+					onion_routing_packet.write(writer)?;
+					// time_created is not serialized - we re-init the timeout upon deserialization
+				},
+				&HTLCUpdateAwaitingACK::ClaimHTLC { ref payment_preimage, ref htlc_id } => {
+					1u8.write(writer)?;
+					payment_preimage.write(writer)?;
+					htlc_id.write(writer)?;
+				},
+				&HTLCUpdateAwaitingACK::FailHTLC { ref htlc_id, ref err_packet } => {
+					2u8.write(writer)?;
+					htlc_id.write(writer)?;
+					err_packet.write(writer)?;
+				}
+			}
+		}
+
+		self.monitor_pending_revoke_and_ack.write(writer)?;
+		self.monitor_pending_commitment_signed.write(writer)?;
+		match self.monitor_pending_order {
+			None => 0u8.write(writer)?,
+			Some(RAACommitmentOrder::CommitmentFirst) => 1u8.write(writer)?,
+			Some(RAACommitmentOrder::RevokeAndACKFirst) => 2u8.write(writer)?,
+		}
+
+		(self.monitor_pending_forwards.len() as u64).write(writer)?;
+		for &(ref pending_forward, ref htlc_id) in self.monitor_pending_forwards.iter() {
+			pending_forward.write(writer)?;
+			htlc_id.write(writer)?;
+		}
+
+		(self.monitor_pending_failures.len() as u64).write(writer)?;
+		for &(ref htlc_source, ref payment_hash, ref fail_reason) in self.monitor_pending_failures.iter() {
+			htlc_source.write(writer)?;
+			payment_hash.write(writer)?;
+			fail_reason.write(writer)?;
+		}
+
+		write_option!(self.pending_update_fee);
+		write_option!(self.holding_cell_update_fee);
+
+		self.next_local_htlc_id.write(writer)?;
+		(self.next_remote_htlc_id - dropped_inbound_htlcs).write(writer)?;
+		self.channel_update_count.write(writer)?;
+		self.feerate_per_kw.write(writer)?;
+
+		(self.last_local_commitment_txn.len() as u64).write(writer)?;
+		for tx in self.last_local_commitment_txn.iter() {
+			if let Err(e) = tx.consensus_encode(&mut RawEncoder::new(WriterWriteAdaptor(writer))) {
+				match e {
+					network::serialize::Error::Io(e) => return Err(e),
+					_ => panic!("last_local_commitment_txn must have been well-formed!"),
+				}
+			}
+		}
+
+		match self.last_sent_closing_fee {
+			Some((feerate, fee)) => {
+				1u8.write(writer)?;
+				feerate.write(writer)?;
+				fee.write(writer)?;
+			},
+			None => 0u8.write(writer)?,
+		}
+
+		write_option!(self.funding_tx_confirmed_in);
+		write_option!(self.short_channel_id);
+
+		self.last_block_connected.write(writer)?;
+		self.funding_tx_confirmations.write(writer)?;
+
+		self.their_dust_limit_satoshis.write(writer)?;
+		self.our_dust_limit_satoshis.write(writer)?;
+		self.their_max_htlc_value_in_flight_msat.write(writer)?;
+		self.their_channel_reserve_satoshis.write(writer)?;
+		self.their_htlc_minimum_msat.write(writer)?;
+		self.our_htlc_minimum_msat.write(writer)?;
+		self.their_to_self_delay.write(writer)?;
+		self.their_max_accepted_htlcs.write(writer)?;
+
+		write_option!(self.their_funding_pubkey);
+		write_option!(self.their_revocation_basepoint);
+		write_option!(self.their_payment_basepoint);
+		write_option!(self.their_delayed_payment_basepoint);
+		write_option!(self.their_htlc_basepoint);
+		write_option!(self.their_cur_commitment_point);
+
+		write_option!(self.their_prev_commitment_point);
+		self.their_node_id.write(writer)?;
+
+		write_option!(self.their_shutdown_scriptpubkey);
+
+		self.channel_monitor.write_for_disk(writer)?;
+		Ok(())
+	}
+}
+
+impl<R : ::std::io::Read> ReadableArgs<R, Arc<Logger>> for Channel {
+	fn read(reader: &mut R, logger: Arc<Logger>) -> Result<Self, DecodeError> {
+		let _ver: u8 = Readable::read(reader)?;
+		let min_ver: u8 = Readable::read(reader)?;
+		if min_ver > SERIALIZATION_VERSION {
+			return Err(DecodeError::UnknownVersion);
+		}
+
+		let user_id = Readable::read(reader)?;
+
+		let channel_id = Readable::read(reader)?;
+		let channel_state = Readable::read(reader)?;
+		let channel_outbound = Readable::read(reader)?;
+		let announce_publicly = Readable::read(reader)?;
+		let channel_value_satoshis = Readable::read(reader)?;
+
+		let local_keys = Readable::read(reader)?;
+		let shutdown_pubkey = Readable::read(reader)?;
+
+		let cur_local_commitment_transaction_number = Readable::read(reader)?;
+		let cur_remote_commitment_transaction_number = Readable::read(reader)?;
+		let value_to_self_msat = Readable::read(reader)?;
+
+		let received_commitment_while_awaiting_raa = Readable::read(reader)?;
+
+		let pending_inbound_htlc_count: u64 = Readable::read(reader)?;
+		let mut pending_inbound_htlcs = Vec::with_capacity(cmp::min(pending_inbound_htlc_count as usize, OUR_MAX_HTLCS as usize));
+		for _ in 0..pending_inbound_htlc_count {
+			pending_inbound_htlcs.push(InboundHTLCOutput {
+				htlc_id: Readable::read(reader)?,
+				amount_msat: Readable::read(reader)?,
+				cltv_expiry: Readable::read(reader)?,
+				payment_hash: Readable::read(reader)?,
+				state: match <u8 as Readable<R>>::read(reader)? {
+					1 => InboundHTLCState::AwaitingRemoteRevokeToAnnounce(Readable::read(reader)?),
+					2 => InboundHTLCState::AwaitingAnnouncedRemoteRevoke(Readable::read(reader)?),
+					3 => InboundHTLCState::Committed,
+					4 => InboundHTLCState::LocalRemoved(Readable::read(reader)?),
+					_ => return Err(DecodeError::InvalidValue),
+				},
+			});
+		}
+
+		macro_rules! read_option { () => {
+			match <u8 as Readable<R>>::read(reader)? {
+				0 => None,
+				1 => Some(Readable::read(reader)?),
+				_ => return Err(DecodeError::InvalidValue),
+			}
+		} }
+
+		let pending_outbound_htlc_count: u64 = Readable::read(reader)?;
+		let mut pending_outbound_htlcs = Vec::with_capacity(cmp::min(pending_outbound_htlc_count as usize, OUR_MAX_HTLCS as usize));
+		for _ in 0..pending_outbound_htlc_count {
+			pending_outbound_htlcs.push(OutboundHTLCOutput {
+				htlc_id: Readable::read(reader)?,
+				amount_msat: Readable::read(reader)?,
+				cltv_expiry: Readable::read(reader)?,
+				payment_hash: Readable::read(reader)?,
+				source: Readable::read(reader)?,
+				fail_reason: read_option!(),
+				state: match <u8 as Readable<R>>::read(reader)? {
+					0 => OutboundHTLCState::LocalAnnounced(Box::new(Readable::read(reader)?)),
+					1 => OutboundHTLCState::Committed,
+					2 => OutboundHTLCState::RemoteRemoved,
+					3 => OutboundHTLCState::AwaitingRemoteRevokeToRemove,
+					4 => OutboundHTLCState::AwaitingRemovedRemoteRevoke,
+					_ => return Err(DecodeError::InvalidValue),
+				},
+			});
+		}
+
+		let holding_cell_htlc_update_count: u64 = Readable::read(reader)?;
+		let mut holding_cell_htlc_updates = Vec::with_capacity(cmp::min(holding_cell_htlc_update_count as usize, OUR_MAX_HTLCS as usize*2));
+		for _ in 0..holding_cell_htlc_update_count {
+			holding_cell_htlc_updates.push(match <u8 as Readable<R>>::read(reader)? {
+				0 => HTLCUpdateAwaitingACK::AddHTLC {
+					amount_msat: Readable::read(reader)?,
+					cltv_expiry: Readable::read(reader)?,
+					payment_hash: Readable::read(reader)?,
+					source: Readable::read(reader)?,
+					onion_routing_packet: Readable::read(reader)?,
+					time_created: Instant::now(),
+				},
+				1 => HTLCUpdateAwaitingACK::ClaimHTLC {
+					payment_preimage: Readable::read(reader)?,
+					htlc_id: Readable::read(reader)?,
+				},
+				2 => HTLCUpdateAwaitingACK::FailHTLC {
+					htlc_id: Readable::read(reader)?,
+					err_packet: Readable::read(reader)?,
+				},
+				_ => return Err(DecodeError::InvalidValue),
+			});
+		}
+
+		let monitor_pending_revoke_and_ack = Readable::read(reader)?;
+		let monitor_pending_commitment_signed = Readable::read(reader)?;
+
+		let monitor_pending_order = match <u8 as Readable<R>>::read(reader)? {
+			0 => None,
+			1 => Some(RAACommitmentOrder::CommitmentFirst),
+			2 => Some(RAACommitmentOrder::RevokeAndACKFirst),
+			_ => return Err(DecodeError::InvalidValue),
+		};
+
+		let monitor_pending_forwards_count: u64 = Readable::read(reader)?;
+		let mut monitor_pending_forwards = Vec::with_capacity(cmp::min(monitor_pending_forwards_count as usize, OUR_MAX_HTLCS as usize));
+		for _ in 0..monitor_pending_forwards_count {
+			monitor_pending_forwards.push((Readable::read(reader)?, Readable::read(reader)?));
+		}
+
+		let monitor_pending_failures_count: u64 = Readable::read(reader)?;
+		let mut monitor_pending_failures = Vec::with_capacity(cmp::min(monitor_pending_failures_count as usize, OUR_MAX_HTLCS as usize));
+		for _ in 0..monitor_pending_failures_count {
+			monitor_pending_failures.push((Readable::read(reader)?, Readable::read(reader)?, Readable::read(reader)?));
+		}
+
+		let pending_update_fee = read_option!();
+		let holding_cell_update_fee = read_option!();
+
+		let next_local_htlc_id = Readable::read(reader)?;
+		let next_remote_htlc_id = Readable::read(reader)?;
+		let channel_update_count = Readable::read(reader)?;
+		let feerate_per_kw = Readable::read(reader)?;
+
+		let last_local_commitment_txn_count: u64 = Readable::read(reader)?;
+		let mut last_local_commitment_txn = Vec::with_capacity(cmp::min(last_local_commitment_txn_count as usize, OUR_MAX_HTLCS as usize*2 + 1));
+		for _ in 0..last_local_commitment_txn_count {
+			last_local_commitment_txn.push(match Transaction::consensus_decode(&mut RawDecoder::new(reader.by_ref())) {
+				Ok(tx) => tx,
+				Err(_) => return Err(DecodeError::InvalidValue),
+			});
+		}
+
+		let last_sent_closing_fee = match <u8 as Readable<R>>::read(reader)? {
+			0 => None,
+			1 => Some((Readable::read(reader)?, Readable::read(reader)?)),
+			_ => return Err(DecodeError::InvalidValue),
+		};
+
+		let funding_tx_confirmed_in = read_option!();
+		let short_channel_id = read_option!();
+
+		let last_block_connected = Readable::read(reader)?;
+		let funding_tx_confirmations = Readable::read(reader)?;
+
+		let their_dust_limit_satoshis = Readable::read(reader)?;
+		let our_dust_limit_satoshis = Readable::read(reader)?;
+		let their_max_htlc_value_in_flight_msat = Readable::read(reader)?;
+		let their_channel_reserve_satoshis = Readable::read(reader)?;
+		let their_htlc_minimum_msat = Readable::read(reader)?;
+		let our_htlc_minimum_msat = Readable::read(reader)?;
+		let their_to_self_delay = Readable::read(reader)?;
+		let their_max_accepted_htlcs = Readable::read(reader)?;
+
+		let their_funding_pubkey = read_option!();
+		let their_revocation_basepoint = read_option!();
+		let their_payment_basepoint = read_option!();
+		let their_delayed_payment_basepoint = read_option!();
+		let their_htlc_basepoint = read_option!();
+		let their_cur_commitment_point = read_option!();
+
+		let their_prev_commitment_point = read_option!();
+		let their_node_id = Readable::read(reader)?;
+
+		let their_shutdown_scriptpubkey = read_option!();
+		let (monitor_last_block, channel_monitor) = ReadableArgs::read(reader, logger.clone())?;
+		// We drop the ChannelMonitor's last block connected hash cause we don't actually bother
+		// doing full block connection operations on the internal CHannelMonitor copies
+		if monitor_last_block != last_block_connected {
+			return Err(DecodeError::InvalidValue);
+		}
+
+		Ok(Channel {
+			user_id,
+
+			channel_id,
+			channel_state,
+			channel_outbound,
+			secp_ctx: Secp256k1::new(),
+			announce_publicly,
+			channel_value_satoshis,
+
+			local_keys,
+			shutdown_pubkey,
+
+			cur_local_commitment_transaction_number,
+			cur_remote_commitment_transaction_number,
+			value_to_self_msat,
+
+			received_commitment_while_awaiting_raa,
+			pending_inbound_htlcs,
+			pending_outbound_htlcs,
+			holding_cell_htlc_updates,
+
+			monitor_pending_revoke_and_ack,
+			monitor_pending_commitment_signed,
+			monitor_pending_order,
+			monitor_pending_forwards,
+			monitor_pending_failures,
+
+			pending_update_fee,
+			holding_cell_update_fee,
+			next_local_htlc_id,
+			next_remote_htlc_id,
+			channel_update_count,
+			feerate_per_kw,
+
+			#[cfg(debug_assertions)]
+			max_commitment_tx_output_local: ::std::sync::Mutex::new((0, 0)),
+			#[cfg(debug_assertions)]
+			max_commitment_tx_output_remote: ::std::sync::Mutex::new((0, 0)),
+
+			last_local_commitment_txn,
+
+			last_sent_closing_fee,
+
+			funding_tx_confirmed_in,
+			short_channel_id,
+			last_block_connected,
+			funding_tx_confirmations,
+
+			their_dust_limit_satoshis,
+			our_dust_limit_satoshis,
+			their_max_htlc_value_in_flight_msat,
+			their_channel_reserve_satoshis,
+			their_htlc_minimum_msat,
+			our_htlc_minimum_msat,
+			their_to_self_delay,
+			their_max_accepted_htlcs,
+
+			their_funding_pubkey,
+			their_revocation_basepoint,
+			their_payment_basepoint,
+			their_delayed_payment_basepoint,
+			their_htlc_basepoint,
+			their_cur_commitment_point,
+
+			their_prev_commitment_point,
+			their_node_id,
+
+			their_shutdown_scriptpubkey,
+
+			channel_monitor,
+
+			logger,
+		})
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	use bitcoin::util::hash::{Sha256dHash, Hash160};

src/ln/channelmonitor.rs

@@ -16,6 +16,8 @@ use bitcoin::blockdata::transaction::{TxIn,TxOut,SigHashType,Transaction};
 use bitcoin::blockdata::transaction::OutPoint as BitcoinOutPoint;
 use bitcoin::blockdata::script::Script;
 use bitcoin::network::serialize;
+use bitcoin::network::serialize::BitcoinHash;
+use bitcoin::network::encodable::{ConsensusDecodable, ConsensusEncodable};
 use bitcoin::util::hash::Sha256dHash;
 use bitcoin::util::bip143;
 
@@ -31,7 +33,8 @@ use ln::chan_utils::HTLCOutputInCommitment;
 use chain::chaininterface::{ChainListener, ChainWatchInterface, BroadcasterInterface};
 use chain::transaction::OutPoint;
 use chain::keysinterface::SpendableOutputDescriptor;
-use util::ser::{Readable, Writer};
+use util::logger::Logger;
+use util::ser::{ReadableArgs, Readable, Writer, Writeable, WriterWriteAdaptor, U48};
 use util::sha2::Sha256;
 use util::{byte_utils, events};
 
@@ -112,12 +115,13 @@ pub struct SimpleManyChannelMonitor<Key> {
 }
 
 impl<Key : Send + cmp::Eq + hash::Hash> ChainListener for SimpleManyChannelMonitor<Key> {
-	fn block_connected(&self, _header: &BlockHeader, height: u32, txn_matched: &[&Transaction], _indexes_of_txn_matched: &[u32]) {
+	fn block_connected(&self, header: &BlockHeader, height: u32, txn_matched: &[&Transaction], _indexes_of_txn_matched: &[u32]) {
+		let block_hash = header.bitcoin_hash();
 		let mut new_events: Vec<events::Event> = Vec::with_capacity(0);
 		{
-			let monitors = self.monitors.lock().unwrap();
+			let mut monitors = self.monitors.lock().unwrap();
-			for monitor in monitors.values() {
+			for monitor in monitors.values_mut() {
-				let (txn_outputs, spendable_outputs) = monitor.block_connected(txn_matched, height, &*self.broadcaster);
+				let (txn_outputs, spendable_outputs) = monitor.block_connected(txn_matched, height, &block_hash, &*self.broadcaster);
 				if spendable_outputs.len() > 0 {
 					new_events.push(events::Event::SpendableOutputs {
 						outputs: spendable_outputs,
@@ -239,6 +243,7 @@ const MIN_SERIALIZATION_VERSION: u8 = 1;
 ///
 /// You MUST ensure that no ChannelMonitors for a given channel anywhere contain out-of-date
 /// information and are actively monitoring the chain.
+#[derive(Clone)]
 pub struct ChannelMonitor {
 	funding_txo: Option<(OutPoint, Script)>,
 	commitment_transaction_number_obscure_factor: u64,
@@ -259,7 +264,7 @@ pub struct ChannelMonitor {
 	/// spending. Thus, in order to claim them via revocation key, we track all the remote
 	/// commitment transactions which we find on-chain, mapping them to the commitment number which
 	/// can be used to derive the revocation key and claim the transactions.
-	remote_commitment_txn_on_chain: Mutex<HashMap<Sha256dHash, u64>>,
+	remote_commitment_txn_on_chain: HashMap<Sha256dHash, (u64, Vec<Script>)>,
 	/// Cache used to make pruning of payment_preimages faster.
 	/// Maps payment_hash values to commitment numbers for remote transactions for non-revoked
 	/// remote transactions (ie should remain pretty small).
@@ -273,39 +278,22 @@ pub struct ChannelMonitor {
 	prev_local_signed_commitment_tx: Option<LocalSignedTx>,
 	current_local_signed_commitment_tx: Option<LocalSignedTx>,
 
+	// Used just for ChannelManager to make sure it has the latest channel data during
+	// deserialization
+	current_remote_commitment_number: u64,
+
 	payment_preimages: HashMap<[u8; 32], [u8; 32]>,
 
 	destination_script: Script,
+
+	// We simply modify last_block_hash in Channel's block_connected so that serialization is
+	// consistent but hopefully the users' copy handles block_connected in a consistent way.
+	// (we do *not*, however, update them in insert_combine to ensure any local user copies keep
+	// their last_block_hash from its state and not based on updated copies that didn't run through
+	// the full block_connected).
+	pub(crate) last_block_hash: Sha256dHash,
 	secp_ctx: Secp256k1<secp256k1::All>, //TODO: dedup this a bit...
-}
+	logger: Arc<Logger>,
-impl Clone for ChannelMonitor {
-	fn clone(&self) -> Self {
-		ChannelMonitor {
-			funding_txo: self.funding_txo.clone(),
-			commitment_transaction_number_obscure_factor: self.commitment_transaction_number_obscure_factor.clone(),
-
-			key_storage: self.key_storage.clone(),
-			their_htlc_base_key: self.their_htlc_base_key.clone(),
-			their_delayed_payment_base_key: self.their_delayed_payment_base_key.clone(),
-			their_cur_revocation_points: self.their_cur_revocation_points.clone(),
-
-			our_to_self_delay: self.our_to_self_delay,
-			their_to_self_delay: self.their_to_self_delay,
-
-			old_secrets: self.old_secrets.clone(),
-			remote_claimable_outpoints: self.remote_claimable_outpoints.clone(),
-			remote_commitment_txn_on_chain: Mutex::new((*self.remote_commitment_txn_on_chain.lock().unwrap()).clone()),
-			remote_hash_commitment_number: self.remote_hash_commitment_number.clone(),
-
-			prev_local_signed_commitment_tx: self.prev_local_signed_commitment_tx.clone(),
-			current_local_signed_commitment_tx: self.current_local_signed_commitment_tx.clone(),
-
-			payment_preimages: self.payment_preimages.clone(),
-
-			destination_script: self.destination_script.clone(),
-			secp_ctx: self.secp_ctx.clone(),
-		}
-	}
 }
 
 #[cfg(any(test, feature = "fuzztarget"))]
@@ -322,8 +310,10 @@ impl PartialEq for ChannelMonitor {
 			self.our_to_self_delay != other.our_to_self_delay ||
 			self.their_to_self_delay != other.their_to_self_delay ||
 			self.remote_claimable_outpoints != other.remote_claimable_outpoints ||
+			self.remote_commitment_txn_on_chain != other.remote_commitment_txn_on_chain ||
 			self.remote_hash_commitment_number != other.remote_hash_commitment_number ||
 			self.prev_local_signed_commitment_tx != other.prev_local_signed_commitment_tx ||
+			self.current_remote_commitment_number != other.current_remote_commitment_number ||
 			self.current_local_signed_commitment_tx != other.current_local_signed_commitment_tx ||
 			self.payment_preimages != other.payment_preimages ||
 			self.destination_script != other.destination_script
@@ -335,15 +325,13 @@ impl PartialEq for ChannelMonitor {
 					return false
 				}
 			}
-			let us = self.remote_commitment_txn_on_chain.lock().unwrap();
+			true
-			let them = other.remote_commitment_txn_on_chain.lock().unwrap();
-			*us == *them
 		}
 	}
 }
 
 impl ChannelMonitor {
-	pub(super) fn new(revocation_base_key: &SecretKey, delayed_payment_base_key: &SecretKey, htlc_base_key: &SecretKey, our_to_self_delay: u16, destination_script: Script) -> ChannelMonitor {
+	pub(super) fn new(revocation_base_key: &SecretKey, delayed_payment_base_key: &SecretKey, htlc_base_key: &SecretKey, our_to_self_delay: u16, destination_script: Script, logger: Arc<Logger>) -> ChannelMonitor {
 		ChannelMonitor {
 			funding_txo: None,
 			commitment_transaction_number_obscure_factor: 0,
@@ -364,16 +352,19 @@ impl ChannelMonitor {
 
 			old_secrets: [([0; 32], 1 << 48); 49],
 			remote_claimable_outpoints: HashMap::new(),
-			remote_commitment_txn_on_chain: Mutex::new(HashMap::new()),
+			remote_commitment_txn_on_chain: HashMap::new(),
 			remote_hash_commitment_number: HashMap::new(),
 
 			prev_local_signed_commitment_tx: None,
 			current_local_signed_commitment_tx: None,
+			current_remote_commitment_number: 1 << 48,
 
 			payment_preimages: HashMap::new(),
-
 			destination_script: destination_script,
+
+			last_block_hash: Default::default(),
 			secp_ctx: Secp256k1::new(),
+			logger,
 		}
 	}
 
@@ -486,6 +477,7 @@ impl ChannelMonitor {
 			self.remote_hash_commitment_number.insert(htlc.payment_hash, commitment_number);
 		}
 		self.remote_claimable_outpoints.insert(unsigned_commitment_tx.txid(), htlc_outputs);
+		self.current_remote_commitment_number = commitment_number;
 	}
 
 	/// Informs this monitor of the latest local (ie broadcastable) commitment transaction. The
@@ -543,6 +535,8 @@ impl ChannelMonitor {
 		if our_min_secret > other_min_secret {
 			self.provide_secret(other_min_secret, other.get_secret(other_min_secret).unwrap(), None)?;
 		}
+		// TODO: We should use current_remote_commitment_number and the commitment number out of
+		// local transactions to decide how to merge
 		if our_min_secret >= other_min_secret {
 			self.their_cur_revocation_points = other.their_cur_revocation_points;
 			for (txid, htlcs) in other.remote_claimable_outpoints.drain() {
@@ -556,6 +550,7 @@ impl ChannelMonitor {
 			}
 			self.payment_preimages = other.payment_preimages;
 		}
+		self.current_remote_commitment_number = cmp::min(self.current_remote_commitment_number, other.current_remote_commitment_number);
 		Ok(())
 	}
 
@@ -597,6 +592,20 @@ impl ChannelMonitor {
 		}
 	}
 
+	/// Gets the sets of all outpoints which this ChannelMonitor expects to hear about spends of.
+	/// Generally useful when deserializing as during normal operation the return values of
+	/// block_connected are sufficient to ensure all relevant outpoints are being monitored (note
+	/// that the get_funding_txo outpoint and transaction must also be monitored for!).
+	pub fn get_monitored_outpoints(&self) -> Vec<(Sha256dHash, u32, &Script)> {
+		let mut res = Vec::with_capacity(self.remote_commitment_txn_on_chain.len() * 2);
+		for (ref txid, &(_, ref outputs)) in self.remote_commitment_txn_on_chain.iter() {
+			for (idx, output) in outputs.iter().enumerate() {
+				res.push(((*txid).clone(), idx as u32, output));
+			}
+		}
+		res
+	}
+
 	/// Serializes into a vec, with various modes for the exposed pub fns
 	fn write<W: Writer>(&self, writer: &mut W, for_local_storage: bool) -> Result<(), ::std::io::Error> {
 		//TODO: We still write out all the serialization here manually instead of using the fancy
@@ -608,8 +617,7 @@ impl ChannelMonitor {
 			&Some((ref outpoint, ref script)) => {
 				writer.write_all(&outpoint.txid[..])?;
 				writer.write_all(&byte_utils::be16_to_array(outpoint.index))?;
-				writer.write_all(&byte_utils::be64_to_array(script.len() as u64))?;
+				script.write(writer)?;
-				writer.write_all(&script[..])?;
 			},
 			&None => {
 				// We haven't even been initialized...not sure why anyone is serializing us, but
@@ -619,7 +627,7 @@ impl ChannelMonitor {
 		}
 
 		// Set in initial Channel-object creation, so should always be set by now:
-		writer.write_all(&byte_utils::be48_to_array(self.commitment_transaction_number_obscure_factor))?;
+		U48(self.commitment_transaction_number_obscure_factor).write(writer)?;
 
 		match self.key_storage {
 			KeyStorage::PrivMode { ref revocation_base_key, ref htlc_base_key, ref delayed_payment_base_key, ref prev_latest_per_commitment_point, ref latest_per_commitment_point } => {
@@ -684,7 +692,7 @@ impl ChannelMonitor {
 		}
 
 		writer.write_all(&byte_utils::be64_to_array(self.remote_claimable_outpoints.len() as u64))?;
-		for (txid, htlc_outputs) in self.remote_claimable_outpoints.iter() {
+		for (ref txid, ref htlc_outputs) in self.remote_claimable_outpoints.iter() {
 			writer.write_all(&txid[..])?;
 			writer.write_all(&byte_utils::be64_to_array(htlc_outputs.len() as u64))?;
 			for htlc_output in htlc_outputs.iter() {
@@ -692,19 +700,20 @@ impl ChannelMonitor {
 			}
 		}
 
-		{
+		writer.write_all(&byte_utils::be64_to_array(self.remote_commitment_txn_on_chain.len() as u64))?;
-			let remote_commitment_txn_on_chain = self.remote_commitment_txn_on_chain.lock().unwrap();
+		for (ref txid, &(commitment_number, ref txouts)) in self.remote_commitment_txn_on_chain.iter() {
-			writer.write_all(&byte_utils::be64_to_array(remote_commitment_txn_on_chain.len() as u64))?;
+			writer.write_all(&txid[..])?;
-			for (txid, commitment_number) in remote_commitment_txn_on_chain.iter() {
+			writer.write_all(&byte_utils::be48_to_array(commitment_number))?;
-				writer.write_all(&txid[..])?;
+			(txouts.len() as u64).write(writer)?;
-				writer.write_all(&byte_utils::be48_to_array(*commitment_number))?;
+			for script in txouts.iter() {
+				script.write(writer)?;
 			}
 		}
 
 		if for_local_storage {
 			writer.write_all(&byte_utils::be64_to_array(self.remote_hash_commitment_number.len() as u64))?;
-			for (payment_hash, commitment_number) in self.remote_hash_commitment_number.iter() {
+			for (ref payment_hash, commitment_number) in self.remote_hash_commitment_number.iter() {
-				writer.write_all(payment_hash)?;
+				writer.write_all(*payment_hash)?;
 				writer.write_all(&byte_utils::be48_to_array(*commitment_number))?;
 			}
 		} else {
@@ -713,9 +722,12 @@ impl ChannelMonitor {
 
 		macro_rules! serialize_local_tx {
 			($local_tx: expr) => {
-				let tx_ser = serialize::serialize(&$local_tx.tx).unwrap();
+				if let Err(e) = $local_tx.tx.consensus_encode(&mut serialize::RawEncoder::new(WriterWriteAdaptor(writer))) {
-				writer.write_all(&byte_utils::be64_to_array(tx_ser.len() as u64))?;
+					match e {
-				writer.write_all(&tx_ser)?;
+						serialize::Error::Io(e) => return Err(e),
+						_ => panic!("local tx must have been well-formed!"),
+					}
+				}
 
 				writer.write_all(&$local_tx.revocation_key.serialize())?;
 				writer.write_all(&$local_tx.a_htlc_key.serialize())?;
@@ -746,23 +758,41 @@ impl ChannelMonitor {
 			writer.write_all(&[0; 1])?;
 		}
 
+		if for_local_storage {
+			writer.write_all(&byte_utils::be48_to_array(self.current_remote_commitment_number))?;
+		} else {
+			writer.write_all(&byte_utils::be48_to_array(0))?;
+		}
+
 		writer.write_all(&byte_utils::be64_to_array(self.payment_preimages.len() as u64))?;
 		for payment_preimage in self.payment_preimages.values() {
 			writer.write_all(payment_preimage)?;
 		}
 
-		writer.write_all(&byte_utils::be64_to_array(self.destination_script.len() as u64))?;
+		self.last_block_hash.write(writer)?;
-		writer.write_all(&self.destination_script[..])?;
+		self.destination_script.write(writer)?;
 
 		Ok(())
 	}
 
 	/// Writes this monitor into the given writer, suitable for writing to disk.
+	///
+	/// Note that the deserializer is only implemented for (Sha256dHash, ChannelMonitor), which
+	/// tells you the last block hash which was block_connect()ed. You MUST rescan any blocks along
+	/// the "reorg path" (ie not just starting at the same height but starting at the highest
+	/// common block that appears on your best chain as well as on the chain which contains the
+	/// last block hash returned) upon deserializing the object!
 	pub fn write_for_disk<W: Writer>(&self, writer: &mut W) -> Result<(), ::std::io::Error> {
 		self.write(writer, true)
 	}
 
 	/// Encodes this monitor into the given writer, suitable for sending to a remote watchtower
+	///
+	/// Note that the deserializer is only implemented for (Sha256dHash, ChannelMonitor), which
+	/// tells you the last block hash which was block_connect()ed. You MUST rescan any blocks along
+	/// the "reorg path" (ie not just starting at the same height but starting at the highest
+	/// common block that appears on your best chain as well as on the chain which contains the
+	/// last block hash returned) upon deserializing the object!
 	pub fn write_for_watchtower<W: Writer>(&self, writer: &mut W) -> Result<(), ::std::io::Error> {
 		self.write(writer, false)
 	}
@@ -792,11 +822,21 @@ impl ChannelMonitor {
 		min
 	}
 
+	pub(super) fn get_cur_remote_commitment_number(&self) -> u64 {
+		self.current_remote_commitment_number
+	}
+
+	pub(super) fn get_cur_local_commitment_number(&self) -> u64 {
+		if let &Some(ref local_tx) = &self.current_local_signed_commitment_tx {
+			0xffff_ffff_ffff - ((((local_tx.tx.input[0].sequence as u64 & 0xffffff) << 3*8) | (local_tx.tx.lock_time as u64 & 0xffffff)) ^ self.commitment_transaction_number_obscure_factor)
+		} else { 0xffff_ffff_ffff }
+	}
+
 	/// Attempts to claim a remote commitment transaction's outputs using the revocation key and
 	/// data in remote_claimable_outpoints. Will directly claim any HTLC outputs which expire at a
 	/// height > height + CLTV_SHARED_CLAIM_BUFFER. In any case, will install monitoring for
 	/// HTLC-Success/HTLC-Timeout transactions.
-	fn check_spend_remote_transaction(&self, tx: &Transaction, height: u32) -> (Vec<Transaction>, (Sha256dHash, Vec<TxOut>), Vec<SpendableOutputDescriptor>) {
+	fn check_spend_remote_transaction(&mut self, tx: &Transaction, height: u32) -> (Vec<Transaction>, (Sha256dHash, Vec<TxOut>), Vec<SpendableOutputDescriptor>) {
 		// Most secp and related errors trying to create keys means we have no hope of constructing
 		// a spend transaction...so we return no transactions to broadcast
 		let mut txn_to_broadcast = Vec::new();
@@ -936,7 +976,7 @@ impl ChannelMonitor {
 			if !inputs.is_empty() || !txn_to_broadcast.is_empty() { // ie we're confident this is actually ours
 				// We're definitely a remote commitment transaction!
 				watch_outputs.append(&mut tx.output.clone());
-				self.remote_commitment_txn_on_chain.lock().unwrap().insert(commitment_txid, commitment_number);
+				self.remote_commitment_txn_on_chain.insert(commitment_txid, (commitment_number, tx.output.iter().map(|output| { output.script_pubkey.clone() }).collect()));
 			}
 			if inputs.is_empty() { return (txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs); } // Nothing to be done...probably a false positive/local tx
 
@@ -973,7 +1013,7 @@ impl ChannelMonitor {
 			// not being generated by the above conditional. Thus, to be safe, we go ahead and
 			// insert it here.
 			watch_outputs.append(&mut tx.output.clone());
-			self.remote_commitment_txn_on_chain.lock().unwrap().insert(commitment_txid, commitment_number);
+			self.remote_commitment_txn_on_chain.insert(commitment_txid, (commitment_number, tx.output.iter().map(|output| { output.script_pubkey.clone() }).collect()));
 
 			if let Some(revocation_points) = self.their_cur_revocation_points {
 				let revocation_point_option =
@@ -1276,7 +1316,24 @@ impl ChannelMonitor {
 		(Vec::new(), Vec::new())
 	}
 
-	fn block_connected(&self, txn_matched: &[&Transaction], height: u32, broadcaster: &BroadcasterInterface)-> (Vec<(Sha256dHash, Vec<TxOut>)>, Vec<SpendableOutputDescriptor>) {
+	/// Used by ChannelManager deserialization to broadcast the latest local state if it's copy of
+	/// the Channel was out-of-date.
+	pub(super) fn get_latest_local_commitment_txn(&self) -> Vec<Transaction> {
+		if let &Some(ref local_tx) = &self.current_local_signed_commitment_tx {
+			let mut res = vec![local_tx.tx.clone()];
+			match self.key_storage {
+				KeyStorage::PrivMode { ref delayed_payment_base_key, ref prev_latest_per_commitment_point, .. } => {
+					res.append(&mut self.broadcast_by_local_state(local_tx, prev_latest_per_commitment_point, &Some(*delayed_payment_base_key)).0);
+				},
+				_ => panic!("Can only broadcast by local channelmonitor"),
+			};
+			res
+		} else {
+			Vec::new()
+		}
+	}
+
+	fn block_connected(&mut self, txn_matched: &[&Transaction], height: u32, block_hash: &Sha256dHash, broadcaster: &BroadcasterInterface)-> (Vec<(Sha256dHash, Vec<TxOut>)>, Vec<SpendableOutputDescriptor>) {
 		let mut watch_outputs = Vec::new();
 		let mut spendable_outputs = Vec::new();
 		for tx in txn_matched {
@@ -1300,9 +1357,8 @@ impl ChannelMonitor {
 						txn = remote_txn;
 					}
 				} else {
-					let remote_commitment_txn_on_chain = self.remote_commitment_txn_on_chain.lock().unwrap();
+					if let Some(&(commitment_number, _)) = self.remote_commitment_txn_on_chain.get(&prevout.txid) {
-					if let Some(commitment_number) = remote_commitment_txn_on_chain.get(&prevout.txid) {
+						let (tx, spendable_output) = self.check_spend_remote_htlc(tx, commitment_number);
-						let (tx, spendable_output) = self.check_spend_remote_htlc(tx, *commitment_number);
 						if let Some(tx) = tx {
 							txn.push(tx);
 						}
@@ -1337,6 +1393,7 @@ impl ChannelMonitor {
 				}
 			}
 		}
+		self.last_block_hash = block_hash.clone();
 		(watch_outputs, spendable_outputs)
 	}
 
@@ -1373,27 +1430,10 @@ impl ChannelMonitor {
 	}
 }
 
-impl<R: ::std::io::Read> Readable<R> for ChannelMonitor {
+const MAX_ALLOC_SIZE: usize = 64*1024;
-	fn read(reader: &mut R) -> Result<Self, DecodeError> {
-		// TODO: read_to_end and then deserializing from that vector is really dumb, we should
-		// actually use the fancy serialization framework we have instead of hacking around it.
-		let mut datavec = Vec::new();
-		reader.read_to_end(&mut datavec)?;
-		let data = &datavec;
-
-		let mut read_pos = 0;
-		macro_rules! read_bytes {
-			($byte_count: expr) => {
-				{
-					if ($byte_count as usize) > data.len() - read_pos {
-						return Err(DecodeError::ShortRead);
-					}
-					read_pos += $byte_count as usize;
-					&data[read_pos - $byte_count as usize..read_pos]
-				}
-			}
-		}
 
+impl<R: ::std::io::Read> ReadableArgs<R, Arc<Logger>> for (Sha256dHash, ChannelMonitor) {
+	fn read(reader: &mut R, logger: Arc<Logger>) -> Result<Self, DecodeError> {
 		let secp_ctx = Secp256k1::new();
 		macro_rules! unwrap_obj {
 			($key: expr) => {
@@ -1404,8 +1444,8 @@ impl<R: ::std::io::Read> Readable<R> for ChannelMonitor {
 			}
 		}
 
-		let _ver = read_bytes!(1)[0];
+		let _ver: u8 = Readable::read(reader)?;
-		let min_ver = read_bytes!(1)[0];
+		let min_ver: u8 = Readable::read(reader)?;
 		if min_ver > SERIALIZATION_VERSION {
 			return Err(DecodeError::UnknownVersion);
 		}
@@ -1413,31 +1453,26 @@ impl<R: ::std::io::Read> Readable<R> for ChannelMonitor {
 		// Technically this can fail and serialize fail a round-trip, but only for serialization of
 		// barely-init'd ChannelMonitors that we can't do anything with.
 		let outpoint = OutPoint {
-			txid: Sha256dHash::from(read_bytes!(32)),
+			txid: Readable::read(reader)?,
-			index: byte_utils::slice_to_be16(read_bytes!(2)),
+			index: Readable::read(reader)?,
 		};
-		let script_len = byte_utils::slice_to_be64(read_bytes!(8));
+		let funding_txo = Some((outpoint, Readable::read(reader)?));
-		let funding_txo = Some((outpoint, Script::from(read_bytes!(script_len).to_vec())));
+		let commitment_transaction_number_obscure_factor = <U48 as Readable<R>>::read(reader)?.0;
-		let commitment_transaction_number_obscure_factor = byte_utils::slice_to_be48(read_bytes!(6));
 
-		let key_storage = match read_bytes!(1)[0] {
+		let key_storage = match <u8 as Readable<R>>::read(reader)? {
 			0 => {
-				let revocation_base_key = unwrap_obj!(SecretKey::from_slice(&secp_ctx, read_bytes!(32)));
+				let revocation_base_key = Readable::read(reader)?;
-				let htlc_base_key = unwrap_obj!(SecretKey::from_slice(&secp_ctx, read_bytes!(32)));
+				let htlc_base_key = Readable::read(reader)?;
-				let delayed_payment_base_key = unwrap_obj!(SecretKey::from_slice(&secp_ctx, read_bytes!(32)));
+				let delayed_payment_base_key = Readable::read(reader)?;
-				let prev_latest_per_commitment_point = match read_bytes!(1)[0] {
+				let prev_latest_per_commitment_point = match <u8 as Readable<R>>::read(reader)? {
-						0 => None,
+					0 => None,
-						1 => {
+					1 => Some(Readable::read(reader)?),
-							Some(unwrap_obj!(PublicKey::from_slice(&secp_ctx, read_bytes!(33))))
+					_ => return Err(DecodeError::InvalidValue),
-						},
-						_ => return Err(DecodeError::InvalidValue),
 				};
-				let latest_per_commitment_point = match read_bytes!(1)[0] {
+				let latest_per_commitment_point = match <u8 as Readable<R>>::read(reader)? {
-						0 => None,
+					0 => None,
-						1 => {
+					1 => Some(Readable::read(reader)?),
-							Some(unwrap_obj!(PublicKey::from_slice(&secp_ctx, read_bytes!(33))))
+					_ => return Err(DecodeError::InvalidValue),
-						},
-						_ => return Err(DecodeError::InvalidValue),
 				};
 				KeyStorage::PrivMode {
 					revocation_base_key,
@@ -1450,45 +1485,41 @@ impl<R: ::std::io::Read> Readable<R> for ChannelMonitor {
 			_ => return Err(DecodeError::InvalidValue),
 		};
 
-		let their_htlc_base_key = Some(unwrap_obj!(PublicKey::from_slice(&secp_ctx, read_bytes!(33))));
+		let their_htlc_base_key = Some(Readable::read(reader)?);
-		let their_delayed_payment_base_key = Some(unwrap_obj!(PublicKey::from_slice(&secp_ctx, read_bytes!(33))));
+		let their_delayed_payment_base_key = Some(Readable::read(reader)?);
 
 		let their_cur_revocation_points = {
-			let first_idx = byte_utils::slice_to_be48(read_bytes!(6));
+			let first_idx = <U48 as Readable<R>>::read(reader)?.0;
 			if first_idx == 0 {
 				None
 			} else {
-				let first_point = unwrap_obj!(PublicKey::from_slice(&secp_ctx, read_bytes!(33)));
+				let first_point = Readable::read(reader)?;
-				let second_point_slice = read_bytes!(33);
+				let second_point_slice: [u8; 33] = Readable::read(reader)?;
 				if second_point_slice[0..32] == [0; 32] && second_point_slice[32] == 0 {
 					Some((first_idx, first_point, None))
 				} else {
-					Some((first_idx, first_point, Some(unwrap_obj!(PublicKey::from_slice(&secp_ctx, second_point_slice)))))
+					Some((first_idx, first_point, Some(unwrap_obj!(PublicKey::from_slice(&secp_ctx, &second_point_slice)))))
 				}
 			}
 		};
 
-		let our_to_self_delay = byte_utils::slice_to_be16(read_bytes!(2));
+		let our_to_self_delay: u16 = Readable::read(reader)?;
-		let their_to_self_delay = Some(byte_utils::slice_to_be16(read_bytes!(2)));
+		let their_to_self_delay: Option<u16> = Some(Readable::read(reader)?);
 
 		let mut old_secrets = [([0; 32], 1 << 48); 49];
 		for &mut (ref mut secret, ref mut idx) in old_secrets.iter_mut() {
-			secret.copy_from_slice(read_bytes!(32));
+			*secret = Readable::read(reader)?;
-			*idx = byte_utils::slice_to_be64(read_bytes!(8));
+			*idx = Readable::read(reader)?;
 		}
 
 		macro_rules! read_htlc_in_commitment {
 			() => {
 				{
-					let offered = match read_bytes!(1)[0] {
+					let offered: bool = Readable::read(reader)?;
-						0 => false, 1 => true,
+					let amount_msat: u64 = Readable::read(reader)?;
-						_ => return Err(DecodeError::InvalidValue),
+					let cltv_expiry: u32 = Readable::read(reader)?;
-					};
+					let payment_hash: [u8; 32] = Readable::read(reader)?;
-					let amount_msat = byte_utils::slice_to_be64(read_bytes!(8));
+					let transaction_output_index: u32 = Readable::read(reader)?;
-					let cltv_expiry = byte_utils::slice_to_be32(read_bytes!(4));
-					let mut payment_hash = [0; 32];
-					payment_hash[..].copy_from_slice(read_bytes!(32));
-					let transaction_output_index = byte_utils::slice_to_be32(read_bytes!(4));
 
 					HTLCOutputInCommitment {
 						offered, amount_msat, cltv_expiry, payment_hash, transaction_output_index
@@ -1497,14 +1528,12 @@ impl<R: ::std::io::Read> Readable<R> for ChannelMonitor {
 			}
 		}
 
-		let remote_claimable_outpoints_len = byte_utils::slice_to_be64(read_bytes!(8));
+		let remote_claimable_outpoints_len: u64 = Readable::read(reader)?;
-		if remote_claimable_outpoints_len > data.len() as u64 / 64 { return Err(DecodeError::BadLengthDescriptor); }
+		let mut remote_claimable_outpoints = HashMap::with_capacity(cmp::min(remote_claimable_outpoints_len as usize, MAX_ALLOC_SIZE / 64));
-		let mut remote_claimable_outpoints = HashMap::with_capacity(remote_claimable_outpoints_len as usize);
 		for _ in 0..remote_claimable_outpoints_len {
-			let txid = Sha256dHash::from(read_bytes!(32));
+			let txid: Sha256dHash = Readable::read(reader)?;
-			let outputs_count = byte_utils::slice_to_be64(read_bytes!(8));
+			let outputs_count: u64 = Readable::read(reader)?;
-			if outputs_count > data.len() as u64 / 32 { return Err(DecodeError::BadLengthDescriptor); }
+			let mut outputs = Vec::with_capacity(cmp::min(outputs_count as usize, MAX_ALLOC_SIZE / 32));
-			let mut outputs = Vec::with_capacity(outputs_count as usize);
 			for _ in 0..outputs_count {
 				outputs.push(read_htlc_in_commitment!());
 			}
@@ -1513,24 +1542,26 @@ impl<R: ::std::io::Read> Readable<R> for ChannelMonitor {
 			}
 		}
 
-		let remote_commitment_txn_on_chain_len = byte_utils::slice_to_be64(read_bytes!(8));
+		let remote_commitment_txn_on_chain_len: u64 = Readable::read(reader)?;
-		if remote_commitment_txn_on_chain_len > data.len() as u64 / 32 { return Err(DecodeError::BadLengthDescriptor); }
+		let mut remote_commitment_txn_on_chain = HashMap::with_capacity(cmp::min(remote_commitment_txn_on_chain_len as usize, MAX_ALLOC_SIZE / 32));
-		let mut remote_commitment_txn_on_chain = HashMap::with_capacity(remote_commitment_txn_on_chain_len as usize);
 		for _ in 0..remote_commitment_txn_on_chain_len {
-			let txid = Sha256dHash::from(read_bytes!(32));
+			let txid: Sha256dHash = Readable::read(reader)?;
-			let commitment_number = byte_utils::slice_to_be48(read_bytes!(6));
+			let commitment_number = <U48 as Readable<R>>::read(reader)?.0;
-			if let Some(_) = remote_commitment_txn_on_chain.insert(txid, commitment_number) {
+			let outputs_count = <u64 as Readable<R>>::read(reader)?;
+			let mut outputs = Vec::with_capacity(cmp::min(outputs_count as usize, MAX_ALLOC_SIZE / 8));
+			for _ in 0..outputs_count {
+				outputs.push(Readable::read(reader)?);
+			}
+			if let Some(_) = remote_commitment_txn_on_chain.insert(txid, (commitment_number, outputs)) {
 				return Err(DecodeError::InvalidValue);
 			}
 		}
 
-		let remote_hash_commitment_number_len = byte_utils::slice_to_be64(read_bytes!(8));
+		let remote_hash_commitment_number_len: u64 = Readable::read(reader)?;
-		if remote_hash_commitment_number_len > data.len() as u64 / 32 { return Err(DecodeError::BadLengthDescriptor); }
+		let mut remote_hash_commitment_number = HashMap::with_capacity(cmp::min(remote_hash_commitment_number_len as usize, MAX_ALLOC_SIZE / 32));
-		let mut remote_hash_commitment_number = HashMap::with_capacity(remote_hash_commitment_number_len as usize);
 		for _ in 0..remote_hash_commitment_number_len {
-			let mut txid = [0; 32];
+			let txid: [u8; 32] = Readable::read(reader)?;
-			txid[..].copy_from_slice(read_bytes!(32));
+			let commitment_number = <U48 as Readable<R>>::read(reader)?.0;
-			let commitment_number = byte_utils::slice_to_be48(read_bytes!(6));
 			if let Some(_) = remote_hash_commitment_number.insert(txid, commitment_number) {
 				return Err(DecodeError::InvalidValue);
 			}
@@ -1539,29 +1570,29 @@ impl<R: ::std::io::Read> Readable<R> for ChannelMonitor {
 		macro_rules! read_local_tx {
 			() => {
 				{
-					let tx_len = byte_utils::slice_to_be64(read_bytes!(8));
+					let tx = match Transaction::consensus_decode(&mut serialize::RawDecoder::new(reader.by_ref())) {
-					let tx_ser = read_bytes!(tx_len);
+						Ok(tx) => tx,
-					let tx: Transaction = unwrap_obj!(serialize::deserialize(tx_ser));
+						Err(e) => match e {
-					if serialize::serialize(&tx).unwrap() != tx_ser {
+							serialize::Error::Io(ioe) => return Err(DecodeError::Io(ioe)),
-						// We check that the tx re-serializes to the same form to ensure there is
+							_ => return Err(DecodeError::InvalidValue),
-						// no extra data, and as rust-bitcoin doesn't handle the 0-input ambiguity
+						},
-						// all that well.
+					};
+
+					if tx.input.is_empty() {
+						// Ensure tx didn't hit the 0-input ambiguity case.
 						return Err(DecodeError::InvalidValue);
 					}
 
-					let revocation_key = unwrap_obj!(PublicKey::from_slice(&secp_ctx, read_bytes!(33)));
+					let revocation_key = Readable::read(reader)?;
-					let a_htlc_key = unwrap_obj!(PublicKey::from_slice(&secp_ctx, read_bytes!(33)));
+					let a_htlc_key = Readable::read(reader)?;
-					let b_htlc_key = unwrap_obj!(PublicKey::from_slice(&secp_ctx, read_bytes!(33)));
+					let b_htlc_key = Readable::read(reader)?;
-					let delayed_payment_key = unwrap_obj!(PublicKey::from_slice(&secp_ctx, read_bytes!(33)));
+					let delayed_payment_key = Readable::read(reader)?;
-					let feerate_per_kw = byte_utils::slice_to_be64(read_bytes!(8));
+					let feerate_per_kw: u64 = Readable::read(reader)?;
 
-					let htlc_outputs_len = byte_utils::slice_to_be64(read_bytes!(8));
+					let htlc_outputs_len: u64 = Readable::read(reader)?;
-					if htlc_outputs_len > data.len() as u64 / 128 { return Err(DecodeError::BadLengthDescriptor); }
+					let mut htlc_outputs = Vec::with_capacity(cmp::min(htlc_outputs_len as usize, MAX_ALLOC_SIZE / 128));
-					let mut htlc_outputs = Vec::with_capacity(htlc_outputs_len as usize);
 					for _ in 0..htlc_outputs_len {
-						htlc_outputs.push((read_htlc_in_commitment!(),
+						htlc_outputs.push((read_htlc_in_commitment!(), Readable::read(reader)?, Readable::read(reader)?));
-								unwrap_obj!(Signature::from_compact(&secp_ctx, read_bytes!(64))),
-								unwrap_obj!(Signature::from_compact(&secp_ctx, read_bytes!(64)))));
 					}
 
 					LocalSignedTx {
@@ -1572,7 +1603,7 @@ impl<R: ::std::io::Read> Readable<R> for ChannelMonitor {
 			}
 		}
 
-		let prev_local_signed_commitment_tx = match read_bytes!(1)[0] {
+		let prev_local_signed_commitment_tx = match <u8 as Readable<R>>::read(reader)? {
 			0 => None,
 			1 => {
 				Some(read_local_tx!())
@@ -1580,7 +1611,7 @@ impl<R: ::std::io::Read> Readable<R> for ChannelMonitor {
 			_ => return Err(DecodeError::InvalidValue),
 		};
 
-		let current_local_signed_commitment_tx = match read_bytes!(1)[0] {
+		let current_local_signed_commitment_tx = match <u8 as Readable<R>>::read(reader)? {
 			0 => None,
 			1 => {
 				Some(read_local_tx!())
@@ -1588,13 +1619,13 @@ impl<R: ::std::io::Read> Readable<R> for ChannelMonitor {
 			_ => return Err(DecodeError::InvalidValue),
 		};
 
-		let payment_preimages_len = byte_utils::slice_to_be64(read_bytes!(8));
+		let current_remote_commitment_number = <U48 as Readable<R>>::read(reader)?.0;
-		if payment_preimages_len > data.len() as u64 / 32 { return Err(DecodeError::InvalidValue); }
+
-		let mut payment_preimages = HashMap::with_capacity(payment_preimages_len as usize);
+		let payment_preimages_len: u64 = Readable::read(reader)?;
+		let mut payment_preimages = HashMap::with_capacity(cmp::min(payment_preimages_len as usize, MAX_ALLOC_SIZE / 32));
 		let mut sha = Sha256::new();
 		for _ in 0..payment_preimages_len {
-			let mut preimage = [0; 32];
+			let preimage: [u8; 32] = Readable::read(reader)?;
-			preimage[..].copy_from_slice(read_bytes!(32));
 			sha.reset();
 			sha.input(&preimage);
 			let mut hash = [0; 32];
@@ -1604,10 +1635,10 @@ impl<R: ::std::io::Read> Readable<R> for ChannelMonitor {
 			}
 		}
 
-		let destination_script_len = byte_utils::slice_to_be64(read_bytes!(8));
+		let last_block_hash: Sha256dHash = Readable::read(reader)?;
-		let destination_script = Script::from(read_bytes!(destination_script_len).to_vec());
+		let destination_script = Readable::read(reader)?;
 
-		Ok(ChannelMonitor {
+		Ok((last_block_hash.clone(), ChannelMonitor {
 			funding_txo,
 			commitment_transaction_number_obscure_factor,
 
@@ -1621,17 +1652,20 @@ impl<R: ::std::io::Read> Readable<R> for ChannelMonitor {
 
 			old_secrets,
 			remote_claimable_outpoints,
-			remote_commitment_txn_on_chain: Mutex::new(remote_commitment_txn_on_chain),
+			remote_commitment_txn_on_chain,
 			remote_hash_commitment_number,
 
 			prev_local_signed_commitment_tx,
 			current_local_signed_commitment_tx,
+			current_remote_commitment_number,
 
 			payment_preimages,
 
 			destination_script,
+			last_block_hash,
 			secp_ctx,
-		})
+			logger,
+		}))
 	}
 
 }
@@ -1645,9 +1679,11 @@ mod tests {
 	use ln::channelmonitor::ChannelMonitor;
 	use ln::chan_utils::{HTLCOutputInCommitment, TxCreationKeys};
 	use util::sha2::Sha256;
+	use util::test_utils::TestLogger;
 	use secp256k1::key::{SecretKey,PublicKey};
 	use secp256k1::{Secp256k1, Signature};
 	use rand::{thread_rng,Rng};
+	use std::sync::Arc;
 
 	#[test]
 	fn test_per_commitment_storage() {
@@ -1655,6 +1691,7 @@ mod tests {
 		let mut secrets: Vec<[u8; 32]> = Vec::new();
 		let mut monitor: ChannelMonitor;
 		let secp_ctx = Secp256k1::new();
+		let logger = Arc::new(TestLogger::new());
 
 		macro_rules! test_secrets {
 			() => {
@@ -1670,7 +1707,7 @@ mod tests {
 
 		{
 			// insert_secret correct sequence
-			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new());
+			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new(), logger.clone());
 			secrets.clear();
 
 			secrets.push([0; 32]);
@@ -1716,7 +1753,7 @@ mod tests {
 
 		{
 			// insert_secret #1 incorrect
-			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new());
+			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new(), logger.clone());
 			secrets.clear();
 
 			secrets.push([0; 32]);
@@ -1732,7 +1769,7 @@ mod tests {
 
 		{
 			// insert_secret #2 incorrect (#1 derived from incorrect)
-			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new());
+			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new(), logger.clone());
 			secrets.clear();
 
 			secrets.push([0; 32]);
@@ -1758,7 +1795,7 @@ mod tests {
 
 		{
 			// insert_secret #3 incorrect
-			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new());
+			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new(), logger.clone());
 			secrets.clear();
 
 			secrets.push([0; 32]);
@@ -1784,7 +1821,7 @@ mod tests {
 
 		{
 			// insert_secret #4 incorrect (1,2,3 derived from incorrect)
-			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new());
+			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new(), logger.clone());
 			secrets.clear();
 
 			secrets.push([0; 32]);
@@ -1830,7 +1867,7 @@ mod tests {
 
 		{
 			// insert_secret #5 incorrect
-			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new());
+			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new(), logger.clone());
 			secrets.clear();
 
 			secrets.push([0; 32]);
@@ -1866,7 +1903,7 @@ mod tests {
 
 		{
 			// insert_secret #6 incorrect (5 derived from incorrect)
-			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new());
+			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new(), logger.clone());
 			secrets.clear();
 
 			secrets.push([0; 32]);
@@ -1912,7 +1949,7 @@ mod tests {
 
 		{
 			// insert_secret #7 incorrect
-			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new());
+			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new(), logger.clone());
 			secrets.clear();
 
 			secrets.push([0; 32]);
@@ -1958,7 +1995,7 @@ mod tests {
 
 		{
 			// insert_secret #8 incorrect
-			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new());
+			monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new(), logger.clone());
 			secrets.clear();
 
 			secrets.push([0; 32]);
@@ -2006,6 +2043,7 @@ mod tests {
 	#[test]
 	fn test_prune_preimages() {
 		let secp_ctx = Secp256k1::new();
+		let logger = Arc::new(TestLogger::new());
 		let dummy_sig = Signature::from_der(&secp_ctx, &hex::decode("3045022100fa86fa9a36a8cd6a7bb8f06a541787d51371d067951a9461d5404de6b928782e02201c8b7c334c10aed8976a3a465be9a28abff4cb23acbf00022295b378ce1fa3cd").unwrap()[..]).unwrap();
 
 		macro_rules! dummy_keys {
@@ -2076,7 +2114,7 @@ mod tests {
 
 		// Prune with one old state and a local commitment tx holding a few overlaps with the
 		// old state.
-		let mut monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new());
+		let mut monitor = ChannelMonitor::new(&SecretKey::from_slice(&secp_ctx, &[42; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[43; 32]).unwrap(), &SecretKey::from_slice(&secp_ctx, &[44; 32]).unwrap(), 0, Script::new(), logger.clone());
 		monitor.set_their_to_self_delay(10);
 
 		monitor.provide_latest_local_commitment_tx_info(dummy_tx.clone(), dummy_keys!(), 0, preimages_to_local_htlcs!(preimages[0..10]));

src/ln/router.rs

@@ -13,9 +13,9 @@ use bitcoin::blockdata::opcodes;
 
 use chain::chaininterface::{ChainError, ChainWatchInterface};
 use ln::channelmanager;
-use ln::msgs::{ErrorAction,HandleError,RoutingMessageHandler,NetAddress,GlobalFeatures};
+use ln::msgs::{DecodeError,ErrorAction,HandleError,RoutingMessageHandler,NetAddress,GlobalFeatures};
 use ln::msgs;
-use util::ser::Writeable;
+use util::ser::{Writeable, Readable};
 use util::logger::Logger;
 
 use std::cmp;
@@ -47,6 +47,37 @@ pub struct Route {
 	pub hops: Vec<RouteHop>,
 }
 
+impl Writeable for Route {
+	fn write<W: ::util::ser::Writer>(&self, writer: &mut W) -> Result<(), ::std::io::Error> {
+		(self.hops.len() as u8).write(writer)?;
+		for hop in self.hops.iter() {
+			hop.pubkey.write(writer)?;
+			hop.short_channel_id.write(writer)?;
+			hop.fee_msat.write(writer)?;
+			hop.cltv_expiry_delta.write(writer)?;
+		}
+		Ok(())
+	}
+}
+
+impl<R: ::std::io::Read> Readable<R> for Route {
+	fn read(reader: &mut R) -> Result<Route, DecodeError> {
+		let hops_count: u8 = Readable::read(reader)?;
+		let mut hops = Vec::with_capacity(hops_count as usize);
+		for _ in 0..hops_count {
+			hops.push(RouteHop {
+				pubkey: Readable::read(reader)?,
+				short_channel_id: Readable::read(reader)?,
+				fee_msat: Readable::read(reader)?,
+				cltv_expiry_delta: Readable::read(reader)?,
+			});
+		}
+		Ok(Route {
+			hops
+		})
+	}
+}
+
 struct DirectionalChannelInfo {
 	src_node_id: PublicKey,
 	last_update: u32,

src/util/ser.rs

@@ -2,19 +2,19 @@
 //! as ChannelsManagers and ChannelMonitors.
 
 use std::result::Result;
-use std::io::Read;
+use std::io::{Read, Write};
 use std::collections::HashMap;
 use std::hash::Hash;
 
 use secp256k1::{Secp256k1, Signature};
-use secp256k1::key::PublicKey;
+use secp256k1::key::{PublicKey, SecretKey};
 use bitcoin::util::hash::Sha256dHash;
 use bitcoin::blockdata::script::Script;
 use std::marker::Sized;
 use ln::msgs::DecodeError;
 use util::byte_utils;
 
-use util::byte_utils::{be64_to_array, be32_to_array, be16_to_array, slice_to_be16, slice_to_be32, slice_to_be64};
+use util::byte_utils::{be64_to_array, be48_to_array, be32_to_array, be16_to_array, slice_to_be16, slice_to_be32, slice_to_be48, slice_to_be64};
 
 const MAX_BUF_SIZE: usize = 64 * 1024;
 
@@ -30,7 +30,7 @@ pub trait Writer {
 	fn size_hint(&mut self, size: usize);
 }
 
-impl<W: ::std::io::Write> Writer for W {
+impl<W: Write> Writer for W {
 	#[inline]
 	fn write_all(&mut self, buf: &[u8]) -> Result<(), ::std::io::Error> {
 		<Self as ::std::io::Write>::write_all(self, buf)
@@ -39,6 +39,20 @@ impl<W: ::std::io::Write> Writer for W {
 	fn size_hint(&mut self, _size: usize) { }
 }
 
+pub(crate) struct WriterWriteAdaptor<'a, W: Writer + 'a>(pub &'a mut W);
+impl<'a, W: Writer + 'a> Write for WriterWriteAdaptor<'a, W> {
+	fn write_all(&mut self, buf: &[u8]) -> Result<(), ::std::io::Error> {
+		self.0.write_all(buf)
+	}
+	fn write(&mut self, buf: &[u8]) -> Result<usize, ::std::io::Error> {
+		self.0.write_all(buf)?;
+		Ok(buf.len())
+	}
+	fn flush(&mut self) -> Result<(), ::std::io::Error> {
+		Ok(())
+	}
+}
+
 struct VecWriter(Vec<u8>);
 impl Writer for VecWriter {
 	fn write_all(&mut self, buf: &[u8]) -> Result<(), ::std::io::Error> {
@@ -82,6 +96,32 @@ pub trait Readable<R>
 	fn read(reader: &mut R) -> Result<Self, DecodeError>;
 }
 
+/// A trait that various higher-level rust-lightning types implement allowing them to be read in
+/// from a Read given some additional set of arguments which is required to deserialize.
+pub trait ReadableArgs<R, P>
+	where Self: Sized,
+	      R: Read
+{
+	/// Reads a Self in from the given Read
+	fn read(reader: &mut R, params: P) -> Result<Self, DecodeError>;
+}
+
+pub(crate) struct U48(pub u64);
+impl Writeable for U48 {
+	#[inline]
+	fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ::std::io::Error> {
+		writer.write_all(&be48_to_array(self.0))
+	}
+}
+impl<R: Read> Readable<R> for U48 {
+	#[inline]
+	fn read(reader: &mut R) -> Result<U48, DecodeError> {
+		let mut buf = [0; 6];
+		reader.read_exact(&mut buf)?;
+		Ok(U48(slice_to_be48(&buf)))
+	}
+}
+
 macro_rules! impl_writeable_primitive {
 	($val_type:ty, $meth_write:ident, $len: expr, $meth_read:ident) => {
 		impl Writeable for $val_type {
@@ -300,6 +340,24 @@ impl<R: Read> Readable<R> for PublicKey {
 	}
 }
 
+impl Writeable for SecretKey {
+	fn write<W: Writer>(&self, w: &mut W) -> Result<(), ::std::io::Error> {
+		let mut ser = [0; 32];
+		ser.copy_from_slice(&self[..]);
+		ser.write(w)
+	}
+}
+
+impl<R: Read> Readable<R> for SecretKey {
+	fn read(r: &mut R) -> Result<Self, DecodeError> {
+		let buf: [u8; 32] = Readable::read(r)?;
+		match SecretKey::from_slice(&Secp256k1::without_caps(), &buf) {
+			Ok(key) => Ok(key),
+			Err(_) => return Err(DecodeError::InvalidValue),
+		}
+	}
+}
+
 impl Writeable for Sha256dHash {
 	fn write<W: Writer>(&self, w: &mut W) -> Result<(), ::std::io::Error> {
 		self.as_bytes().write(w)

src/util/ser_macros.rs

@@ -1,17 +1,19 @@
 macro_rules! impl_writeable {
 	($st:ident, $len: expr, {$($field:ident),*}) => {
-		impl Writeable for $st {
+		impl ::util::ser::Writeable for $st {
-			fn write<W: Writer>(&self, w: &mut W) -> Result<(), ::std::io::Error> {
+			fn write<W: ::util::ser::Writer>(&self, w: &mut W) -> Result<(), ::std::io::Error> {
-				w.size_hint($len);
+				if $len != 0 {
+					w.size_hint($len);
+				}
 				$( self.$field.write(w)?; )*
 				Ok(())
 			}
 		}
 
-		impl<R: Read> Readable<R> for $st {
+		impl<R: ::std::io::Read> ::util::ser::Readable<R> for $st {
-			fn read(r: &mut R) -> Result<Self, DecodeError> {
+			fn read(r: &mut R) -> Result<Self, ::ln::msgs::DecodeError> {
 				Ok(Self {
-					$($field: Readable::read(r)?),*
+					$($field: ::util::ser::Readable::read(r)?),*
 				})
 			}
 		}
@@ -29,7 +31,7 @@ macro_rules! impl_writeable_len_match {
 			}
 		}
 
-		impl<R: Read> Readable<R> for $st {
+		impl<R: ::std::io::Read> Readable<R> for $st {
 			fn read(r: &mut R) -> Result<Self, DecodeError> {
 				Ok(Self {
 					$($field: Readable::read(r)?),*

src/util/test_utils.rs

@@ -6,9 +6,10 @@ use ln::msgs;
 use ln::msgs::{HandleError};
 use util::events;
 use util::logger::{Logger, Level, Record};
-use util::ser::{Readable, Writer};
+use util::ser::{ReadableArgs, Writer};
 
 use bitcoin::blockdata::transaction::Transaction;
+use bitcoin::util::hash::Sha256dHash;
 
 use secp256k1::PublicKey;
 
@@ -55,7 +56,8 @@ impl channelmonitor::ManyChannelMonitor for TestChannelMonitor {
 		// to a watchtower and disk...
 		let mut w = VecWriter(Vec::new());
 		monitor.write_for_disk(&mut w).unwrap();
-		assert!(channelmonitor::ChannelMonitor::read(&mut ::std::io::Cursor::new(&w.0)).unwrap() == monitor);
+		assert!(<(Sha256dHash, channelmonitor::ChannelMonitor)>::read(
+				&mut ::std::io::Cursor::new(&w.0), Arc::new(TestLogger::new())).unwrap().1 == monitor);
 		w.0.clear();
 		monitor.write_for_watchtower(&mut w).unwrap(); // This at least shouldn't crash...
 		self.added_monitors.lock().unwrap().push((funding_txo, monitor.clone()));