Support (de)serializing payment_data in onion TLVs and track them
[rust-lightning] / lightning / src / ln / channelmanager.rs
index 12ac136af0a7c7aa468904cb7cfd6da8c0f559a5..4f2fbad19e5edf06bd88827694d77e5ee6b8072e 100644 (file)
@@ -30,7 +30,7 @@ use chain::transaction::OutPoint;
 use ln::channel::{Channel, ChannelError};
 use ln::channelmonitor::{ChannelMonitor, ChannelMonitorUpdateErr, ManyChannelMonitor, CLTV_CLAIM_BUFFER, LATENCY_GRACE_PERIOD_BLOCKS, ANTI_REORG_DELAY};
 use ln::router::Route;
-use ln::features::InitFeatures;
+use ln::features::{InitFeatures, NodeFeatures};
 use ln::msgs;
 use ln::onion_utils;
 use ln::msgs::{ChannelMessageHandler, DecodeError, LightningError};
@@ -57,19 +57,33 @@ use std::ops::Deref;
 // forward the HTLC with information it will give back to us when it does so, or if it should Fail
 // the HTLC with the relevant message for the Channel to handle giving to the remote peer.
 //
-// When a Channel forwards an HTLC to its peer, it will give us back the PendingForwardHTLCInfo
-// which we will use to construct an outbound HTLC, with a relevant HTLCSource::PreviousHopData
-// filled in to indicate where it came from (which we can use to either fail-backwards or fulfill
-// the HTLC backwards along the relevant path).
+// Once said HTLC is committed in the Channel, if the PendingHTLCStatus indicated Forward, the
+// Channel will return the PendingHTLCInfo back to us, and we will create an HTLCForwardInfo
+// with it to track where it came from (in case of onwards-forward error), waiting a random delay
+// before we forward it.
+//
+// We will then use HTLCForwardInfo's PendingHTLCInfo to construct an outbound HTLC, with a
+// relevant HTLCSource::PreviousHopData filled in to indicate where it came from (which we can use
+// to either fail-backwards or fulfill the HTLC backwards along the relevant path).
 // Alternatively, we can fill an outbound HTLC with a HTLCSource::OutboundRoute indicating this is
 // our payment, which we can use to decode errors or inform the user that the payment was sent.
-/// Stores the info we will need to send when we want to forward an HTLC onwards
+
 #[derive(Clone)] // See Channel::revoke_and_ack for why, tl;dr: Rust bug
-pub(super) struct PendingForwardHTLCInfo {
-       onion_packet: Option<msgs::OnionPacket>,
+enum PendingForwardReceiveHTLCInfo {
+       Forward {
+               onion_packet: msgs::OnionPacket,
+               short_channel_id: u64, // This should be NonZero<u64> eventually
+       },
+       Receive {
+               payment_data: Option<msgs::FinalOnionHopData>,
+       },
+}
+
+#[derive(Clone)] // See Channel::revoke_and_ack for why, tl;dr: Rust bug
+pub(super) struct PendingHTLCInfo {
+       type_data: PendingForwardReceiveHTLCInfo,
        incoming_shared_secret: [u8; 32],
        payment_hash: PaymentHash,
-       short_channel_id: u64,
        pub(super) amt_to_forward: u64,
        pub(super) outgoing_cltv_value: u32,
 }
@@ -83,10 +97,22 @@ pub(super) enum HTLCFailureMsg {
 /// Stores whether we can't forward an HTLC or relevant forwarding info
 #[derive(Clone)] // See Channel::revoke_and_ack for why, tl;dr: Rust bug
 pub(super) enum PendingHTLCStatus {
-       Forward(PendingForwardHTLCInfo),
+       Forward(PendingHTLCInfo),
        Fail(HTLCFailureMsg),
 }
 
+pub(super) enum HTLCForwardInfo {
+       AddHTLC {
+               prev_short_channel_id: u64,
+               prev_htlc_id: u64,
+               forward_info: PendingHTLCInfo,
+       },
+       FailHTLC {
+               htlc_id: u64,
+               err_packet: msgs::OnionErrorPacket,
+       },
+}
+
 /// Tracks the inbound corresponding to an outbound HTLC
 #[derive(Clone, PartialEq)]
 pub(super) struct HTLCPreviousHopData {
@@ -95,6 +121,12 @@ pub(super) struct HTLCPreviousHopData {
        incoming_packet_shared_secret: [u8; 32],
 }
 
+struct ClaimableHTLC {
+       src: HTLCPreviousHopData,
+       value: u64,
+       payment_data: Option<msgs::FinalOnionHopData>,
+}
+
 /// Tracks the inbound corresponding to an outbound HTLC
 #[derive(Clone, PartialEq)]
 pub(super) enum HTLCSource {
@@ -231,18 +263,6 @@ impl MsgHandleErrInternal {
 /// second to 30 seconds, but people expect lightning to be, you know, kinda fast, sadly.
 const MIN_HTLC_RELAY_HOLDING_CELL_MILLIS: u64 = 100;
 
-pub(super) enum HTLCForwardInfo {
-       AddHTLC {
-               prev_short_channel_id: u64,
-               prev_htlc_id: u64,
-               forward_info: PendingForwardHTLCInfo,
-       },
-       FailHTLC {
-               htlc_id: u64,
-               err_packet: msgs::OnionErrorPacket,
-       },
-}
-
 /// For events which result in both a RevokeAndACK and a CommitmentUpdate, by default they should
 /// be sent in the order they appear in the return value, however sometimes the order needs to be
 /// variable at runtime (eg Channel::channel_reestablish needs to re-send messages in the order
@@ -262,14 +282,13 @@ pub(super) struct ChannelHolder<ChanSigner: ChannelKeys> {
        /// short channel id -> forward infos. Key of 0 means payments received
        /// Note that while this is held in the same mutex as the channels themselves, no consistency
        /// guarantees are made about the existence of a channel with the short id here, nor the short
-       /// ids in the PendingForwardHTLCInfo!
+       /// ids in the PendingHTLCInfo!
        pub(super) forward_htlcs: HashMap<u64, Vec<HTLCForwardInfo>>,
-       /// payment_hash -> Vec<(amount_received, htlc_source)> for tracking things that were to us and
-       /// can be failed/claimed by the user
+       /// Tracks things that were to us and can be failed/claimed by the user
        /// Note that while this is held in the same mutex as the channels themselves, no consistency
        /// guarantees are made about the channels given here actually existing anymore by the time you
        /// go to read them!
-       pub(super) claimable_htlcs: HashMap<PaymentHash, Vec<(u64, HTLCPreviousHopData)>>,
+       claimable_htlcs: HashMap<PaymentHash, Vec<ClaimableHTLC>>,
        /// Messages to send to peers - pushed to in the same lock that they are generated in (except
        /// for broadcast messages, where ordering isn't as strict).
        pub(super) pending_msg_events: Vec<events::MessageSendEvent>,
@@ -355,6 +374,8 @@ pub struct ChannelManager<ChanSigner: ChannelKeys, M: Deref> where M::Target: Ma
        channel_state: Mutex<ChannelHolder<ChanSigner>>,
        our_network_key: SecretKey,
 
+       last_node_announcement_serial: AtomicUsize,
+
        /// The bulk of our storage will eventually be here (channels and message queues and the like).
        /// If we are connected to a peer we always at least have an entry here, even if no channels
        /// are currently open with that peer.
@@ -571,7 +592,7 @@ macro_rules! handle_monitor_err {
                                                        } else if $resend_commitment { "commitment" }
                                                        else if $resend_raa { "RAA" }
                                                        else { "nothing" },
-                                               (&$failed_forwards as &Vec<(PendingForwardHTLCInfo, u64)>).len(),
+                                               (&$failed_forwards as &Vec<(PendingHTLCInfo, u64)>).len(),
                                                (&$failed_fails as &Vec<(HTLCSource, PaymentHash, HTLCFailReason)>).len());
                                if !$resend_commitment {
                                        debug_assert!($action_type == RAACommitmentOrder::RevokeAndACKFirst || !$resend_raa);
@@ -649,6 +670,8 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
                        }),
                        our_network_key: keys_manager.get_node_secret(),
 
+                       last_node_announcement_serial: AtomicUsize::new(0),
+
                        per_peer_state: RwLock::new(HashMap::new()),
 
                        pending_events: Mutex::new(Vec::new()),
@@ -958,15 +981,20 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
                                        return_err!("Upstream node set CLTV to the wrong value", 18, &byte_utils::be32_to_array(msg.cltv_expiry));
                                }
 
+                               let payment_data = match next_hop_data.format {
+                                       msgs::OnionHopDataFormat::Legacy { .. } => None,
+                                       msgs::OnionHopDataFormat::NonFinalNode { .. } => return_err!("Got non final data with an HMAC of 0", 0x4000 | 22, &[0;0]),
+                                       msgs::OnionHopDataFormat::FinalNode { payment_data } => payment_data,
+                               };
+
                                // Note that we could obviously respond immediately with an update_fulfill_htlc
                                // message, however that would leak that we are the recipient of this payment, so
                                // instead we stay symmetric with the forwarding case, only responding (after a
                                // delay) once they've send us a commitment_signed!
 
-                               PendingHTLCStatus::Forward(PendingForwardHTLCInfo {
-                                       onion_packet: None,
+                               PendingHTLCStatus::Forward(PendingHTLCInfo {
+                                       type_data: PendingForwardReceiveHTLCInfo::Receive { payment_data },
                                        payment_hash: msg.payment_hash.clone(),
-                                       short_channel_id: 0,
                                        incoming_shared_secret: shared_secret,
                                        amt_to_forward: next_hop_data.amt_to_forward,
                                        outgoing_cltv_value: next_hop_data.outgoing_cltv_value,
@@ -1010,15 +1038,17 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
                                let short_channel_id = match next_hop_data.format {
                                        msgs::OnionHopDataFormat::Legacy { short_channel_id } => short_channel_id,
                                        msgs::OnionHopDataFormat::NonFinalNode { short_channel_id } => short_channel_id,
-                                       msgs::OnionHopDataFormat::FinalNode => {
+                                       msgs::OnionHopDataFormat::FinalNode { .. } => {
                                                return_err!("Final Node OnionHopData provided for us as an intermediary node", 0x4000 | 22, &[0;0]);
                                        },
                                };
 
-                               PendingHTLCStatus::Forward(PendingForwardHTLCInfo {
-                                       onion_packet: Some(outgoing_packet),
+                               PendingHTLCStatus::Forward(PendingHTLCInfo {
+                                       type_data: PendingForwardReceiveHTLCInfo::Forward {
+                                               onion_packet: outgoing_packet,
+                                               short_channel_id: short_channel_id,
+                                       },
                                        payment_hash: msg.payment_hash.clone(),
-                                       short_channel_id: short_channel_id,
                                        incoming_shared_secret: shared_secret,
                                        amt_to_forward: next_hop_data.amt_to_forward,
                                        outgoing_cltv_value: next_hop_data.outgoing_cltv_value,
@@ -1026,8 +1056,11 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
                        };
 
                channel_state = Some(self.channel_state.lock().unwrap());
-               if let &PendingHTLCStatus::Forward(PendingForwardHTLCInfo { ref onion_packet, ref short_channel_id, ref amt_to_forward, ref outgoing_cltv_value, .. }) = &pending_forward_info {
-                       if onion_packet.is_some() { // If short_channel_id is 0 here, we'll reject them in the body here
+               if let &PendingHTLCStatus::Forward(PendingHTLCInfo { ref type_data, ref amt_to_forward, ref outgoing_cltv_value, .. }) = &pending_forward_info {
+                       // If short_channel_id is 0 here, we'll reject them in the body here (which is
+                       // important as various things later assume we are a ::Receive if short_channel_id is
+                       // non-0.
+                       if let &PendingForwardReceiveHTLCInfo::Forward { ref short_channel_id, .. } = type_data {
                                let id_option = channel_state.as_ref().unwrap().short_to_id.get(&short_channel_id).cloned();
                                let forwarding_id = match id_option {
                                        None => { // unknown_next_peer
@@ -1163,6 +1196,9 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
                let onion_keys = secp_call!(onion_utils::construct_onion_keys(&self.secp_ctx, &route, &session_priv),
                                APIError::RouteError{err: "Pubkey along hop was maliciously selected"});
                let (onion_payloads, htlc_msat, htlc_cltv) = onion_utils::build_onion_payloads(&route, cur_height)?;
+               if onion_utils::route_size_insane(&onion_payloads) {
+                       return Err(APIError::RouteError{err: "Route had too large size once"});
+               }
                let onion_packet = onion_utils::construct_onion_packet(onion_payloads, onion_keys, prng_seed, &payment_hash);
 
                let _ = self.total_consistency_lock.read().unwrap();
@@ -1311,6 +1347,37 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
                })
        }
 
+       /// Generates a signed node_announcement from the given arguments and creates a
+       /// BroadcastNodeAnnouncement event.
+       ///
+       /// RGB is a node "color" and alias a printable human-readable string to describe this node to
+       /// humans. They carry no in-protocol meaning.
+       ///
+       /// addresses represent the set (possibly empty) of socket addresses on which this node accepts
+       /// incoming connections.
+       pub fn broadcast_node_announcement(&self, rgb: [u8; 3], alias: [u8; 32], addresses: msgs::NetAddressSet) {
+               let _ = self.total_consistency_lock.read().unwrap();
+
+               let announcement = msgs::UnsignedNodeAnnouncement {
+                       features: NodeFeatures::supported(),
+                       timestamp: self.last_node_announcement_serial.fetch_add(1, Ordering::AcqRel) as u32,
+                       node_id: self.get_our_node_id(),
+                       rgb, alias,
+                       addresses: addresses.to_vec(),
+                       excess_address_data: Vec::new(),
+                       excess_data: Vec::new(),
+               };
+               let msghash = hash_to_message!(&Sha256dHash::hash(&announcement.encode()[..])[..]);
+
+               let mut channel_state = self.channel_state.lock().unwrap();
+               channel_state.pending_msg_events.push(events::MessageSendEvent::BroadcastNodeAnnouncement {
+                       msg: msgs::NodeAnnouncement {
+                               signature: self.secp_ctx.sign(&msghash, &self.our_network_key),
+                               contents: announcement
+                       },
+               });
+       }
+
        /// Processes HTLCs which are pending waiting on random forward delay.
        ///
        /// Should only really ever be called in response to a PendingHTLCsForwardable event.
@@ -1357,22 +1424,25 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
                                                let mut fail_htlc_msgs = Vec::new();
                                                for forward_info in pending_forwards.drain(..) {
                                                        match forward_info {
-                                                               HTLCForwardInfo::AddHTLC { prev_short_channel_id, prev_htlc_id, forward_info } => {
-                                                                       log_trace!(self, "Adding HTLC from short id {} with payment_hash {} to channel with short id {} after delay", log_bytes!(forward_info.payment_hash.0), prev_short_channel_id, short_chan_id);
+                                                               HTLCForwardInfo::AddHTLC { prev_short_channel_id, prev_htlc_id, forward_info: PendingHTLCInfo {
+                                                                               type_data: PendingForwardReceiveHTLCInfo::Forward {
+                                                                                       onion_packet, ..
+                                                                               }, incoming_shared_secret, payment_hash, amt_to_forward, outgoing_cltv_value }, } => {
+                                                                       log_trace!(self, "Adding HTLC from short id {} with payment_hash {} to channel with short id {} after delay", log_bytes!(payment_hash.0), prev_short_channel_id, short_chan_id);
                                                                        let htlc_source = HTLCSource::PreviousHopData(HTLCPreviousHopData {
                                                                                short_channel_id: prev_short_channel_id,
                                                                                htlc_id: prev_htlc_id,
-                                                                               incoming_packet_shared_secret: forward_info.incoming_shared_secret,
+                                                                               incoming_packet_shared_secret: incoming_shared_secret,
                                                                        });
-                                                                       match chan.get_mut().send_htlc(forward_info.amt_to_forward, forward_info.payment_hash, forward_info.outgoing_cltv_value, htlc_source.clone(), forward_info.onion_packet.unwrap()) {
+                                                                       match chan.get_mut().send_htlc(amt_to_forward, payment_hash, outgoing_cltv_value, htlc_source.clone(), onion_packet) {
                                                                                Err(e) => {
                                                                                        if let ChannelError::Ignore(msg) = e {
-                                                                                               log_trace!(self, "Failed to forward HTLC with payment_hash {}: {}", log_bytes!(forward_info.payment_hash.0), msg);
+                                                                                               log_trace!(self, "Failed to forward HTLC with payment_hash {}: {}", log_bytes!(payment_hash.0), msg);
                                                                                        } else {
                                                                                                panic!("Stated return value requirements in send_htlc() were not met");
                                                                                        }
                                                                                        let chan_update = self.get_channel_update(chan.get()).unwrap();
-                                                                                       failed_forwards.push((htlc_source, forward_info.payment_hash, 0x1000 | 7, Some(chan_update)));
+                                                                                       failed_forwards.push((htlc_source, payment_hash, 0x1000 | 7, Some(chan_update)));
                                                                                        continue;
                                                                                },
                                                                                Ok(update_add) => {
@@ -1391,6 +1461,9 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
                                                                                }
                                                                        }
                                                                },
+                                                               HTLCForwardInfo::AddHTLC { .. } => {
+                                                                       panic!("short_channel_id != 0 should imply any pending_forward entries are of type Forward");
+                                                               },
                                                                HTLCForwardInfo::FailHTLC { htlc_id, err_packet } => {
                                                                        log_trace!(self, "Failing HTLC back to channel with short id {} after delay", short_chan_id);
                                                                        match chan.get_mut().get_update_fail_htlc(htlc_id, err_packet) {
@@ -1470,21 +1543,27 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
                                } else {
                                        for forward_info in pending_forwards.drain(..) {
                                                match forward_info {
-                                                       HTLCForwardInfo::AddHTLC { prev_short_channel_id, prev_htlc_id, forward_info } => {
+                                                       HTLCForwardInfo::AddHTLC { prev_short_channel_id, prev_htlc_id, forward_info: PendingHTLCInfo {
+                                                                       type_data: PendingForwardReceiveHTLCInfo::Receive { payment_data },
+                                                                       incoming_shared_secret, payment_hash, amt_to_forward, .. }, } => {
                                                                let prev_hop_data = HTLCPreviousHopData {
                                                                        short_channel_id: prev_short_channel_id,
                                                                        htlc_id: prev_htlc_id,
-                                                                       incoming_packet_shared_secret: forward_info.incoming_shared_secret,
-                                                               };
-                                                               match channel_state.claimable_htlcs.entry(forward_info.payment_hash) {
-                                                                       hash_map::Entry::Occupied(mut entry) => entry.get_mut().push((forward_info.amt_to_forward, prev_hop_data)),
-                                                                       hash_map::Entry::Vacant(entry) => { entry.insert(vec![(forward_info.amt_to_forward, prev_hop_data)]); },
+                                                                       incoming_packet_shared_secret: incoming_shared_secret,
                                                                };
+                                                               channel_state.claimable_htlcs.entry(payment_hash).or_insert(Vec::new()).push(ClaimableHTLC {
+                                                                       src: prev_hop_data,
+                                                                       value: amt_to_forward,
+                                                                       payment_data,
+                                                               });
                                                                new_events.push(events::Event::PaymentReceived {
-                                                                       payment_hash: forward_info.payment_hash,
-                                                                       amt: forward_info.amt_to_forward,
+                                                                       payment_hash: payment_hash,
+                                                                       amt: amt_to_forward,
                                                                });
                                                        },
+                                                       HTLCForwardInfo::AddHTLC { .. } => {
+                                                               panic!("short_channel_id == 0 should imply any pending_forward entries are of type Receive");
+                                                       },
                                                        HTLCForwardInfo::FailHTLC { .. } => {
                                                                panic!("Got pending fail of our own HTLC");
                                                        }
@@ -1549,11 +1628,11 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
                let mut channel_state = Some(self.channel_state.lock().unwrap());
                let removed_source = channel_state.as_mut().unwrap().claimable_htlcs.remove(payment_hash);
                if let Some(mut sources) = removed_source {
-                       for (recvd_value, htlc_with_hash) in sources.drain(..) {
+                       for htlc in sources.drain(..) {
                                if channel_state.is_none() { channel_state = Some(self.channel_state.lock().unwrap()); }
                                self.fail_htlc_backwards_internal(channel_state.take().unwrap(),
-                                               HTLCSource::PreviousHopData(htlc_with_hash), payment_hash,
-                                               HTLCFailReason::Reason { failure_code: 0x4000 | 15, data: byte_utils::be64_to_array(recvd_value).to_vec() });
+                                               HTLCSource::PreviousHopData(htlc.src), payment_hash,
+                                               HTLCFailReason::Reason { failure_code: 0x4000 | 15, data: byte_utils::be64_to_array(htlc.value).to_vec() });
                        }
                        true
                } else { false }
@@ -1677,17 +1756,17 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
                let mut channel_state = Some(self.channel_state.lock().unwrap());
                let removed_source = channel_state.as_mut().unwrap().claimable_htlcs.remove(&payment_hash);
                if let Some(mut sources) = removed_source {
-                       for (received_amount, htlc_with_hash) in sources.drain(..) {
+                       for htlc in sources.drain(..) {
                                if channel_state.is_none() { channel_state = Some(self.channel_state.lock().unwrap()); }
-                               if received_amount < expected_amount || received_amount > expected_amount * 2 {
-                                       let mut htlc_msat_data = byte_utils::be64_to_array(received_amount).to_vec();
+                               if htlc.value < expected_amount || htlc.value > expected_amount * 2 {
+                                       let mut htlc_msat_data = byte_utils::be64_to_array(htlc.value).to_vec();
                                        let mut height_data = byte_utils::be32_to_array(self.latest_block_height.load(Ordering::Acquire) as u32).to_vec();
                                        htlc_msat_data.append(&mut height_data);
                                        self.fail_htlc_backwards_internal(channel_state.take().unwrap(),
-                                                                        HTLCSource::PreviousHopData(htlc_with_hash), &payment_hash,
+                                                                        HTLCSource::PreviousHopData(htlc.src), &payment_hash,
                                                                         HTLCFailReason::Reason { failure_code: 0x4000|15, data: htlc_msat_data });
                                } else {
-                                       self.claim_funds_internal(channel_state.take().unwrap(), HTLCSource::PreviousHopData(htlc_with_hash), payment_preimage);
+                                       self.claim_funds_internal(channel_state.take().unwrap(), HTLCSource::PreviousHopData(htlc.src), payment_preimage);
                                }
                        }
                        true
@@ -2145,7 +2224,7 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
                                        // If the update_add is completely bogus, the call will Err and we will close,
                                        // but if we've sent a shutdown and they haven't acknowledged it yet, we just
                                        // want to reject the new HTLC and fail it backwards instead of forwarding.
-                                       if let PendingHTLCStatus::Forward(PendingForwardHTLCInfo { incoming_shared_secret, .. }) = pending_forward_info {
+                                       if let PendingHTLCStatus::Forward(PendingHTLCInfo { incoming_shared_secret, .. }) = pending_forward_info {
                                                let chan_update = self.get_channel_update(chan.get());
                                                pending_forward_info = PendingHTLCStatus::Fail(HTLCFailureMsg::Relay(msgs::UpdateFailHTLC {
                                                        channel_id: msg.channel_id,
@@ -2275,7 +2354,7 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
        }
 
        #[inline]
-       fn forward_htlcs(&self, per_source_pending_forwards: &mut [(u64, Vec<(PendingForwardHTLCInfo, u64)>)]) {
+       fn forward_htlcs(&self, per_source_pending_forwards: &mut [(u64, Vec<(PendingHTLCInfo, u64)>)]) {
                for &mut (prev_short_channel_id, ref mut pending_forwards) in per_source_pending_forwards {
                        let mut forward_event = None;
                        if !pending_forwards.is_empty() {
@@ -2284,7 +2363,10 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
                                        forward_event = Some(Duration::from_millis(MIN_HTLC_RELAY_HOLDING_CELL_MILLIS))
                                }
                                for (forward_info, prev_htlc_id) in pending_forwards.drain(..) {
-                                       match channel_state.forward_htlcs.entry(forward_info.short_channel_id) {
+                                       match channel_state.forward_htlcs.entry(match forward_info.type_data {
+                                                       PendingForwardReceiveHTLCInfo::Forward { short_channel_id, .. } => short_channel_id,
+                                                       PendingForwardReceiveHTLCInfo::Receive { .. } => 0,
+                                       }) {
                                                hash_map::Entry::Occupied(mut entry) => {
                                                        entry.get_mut().push(HTLCForwardInfo::AddHTLC { prev_short_channel_id, prev_htlc_id, forward_info });
                                                },
@@ -2906,6 +2988,7 @@ impl<ChanSigner: ChannelKeys, M: Deref + Sync + Send> ChannelMessageHandler for
                                        &events::MessageSendEvent::SendShutdown { ref node_id, .. } => node_id != their_node_id,
                                        &events::MessageSendEvent::SendChannelReestablish { ref node_id, .. } => node_id != their_node_id,
                                        &events::MessageSendEvent::BroadcastChannelAnnouncement { .. } => true,
+                                       &events::MessageSendEvent::BroadcastNodeAnnouncement { .. } => true,
                                        &events::MessageSendEvent::BroadcastChannelUpdate { .. } => true,
                                        &events::MessageSendEvent::HandleError { ref node_id, .. } => node_id != their_node_id,
                                        &events::MessageSendEvent::PaymentFailureNetworkUpdate { .. } => true,
@@ -2986,25 +3069,42 @@ impl<ChanSigner: ChannelKeys, M: Deref + Sync + Send> ChannelMessageHandler for
 const SERIALIZATION_VERSION: u8 = 1;
 const MIN_SERIALIZATION_VERSION: u8 = 1;
 
-impl Writeable for PendingForwardHTLCInfo {
+impl Writeable for PendingHTLCInfo {
        fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ::std::io::Error> {
-               self.onion_packet.write(writer)?;
+               match &self.type_data {
+                       &PendingForwardReceiveHTLCInfo::Forward { ref onion_packet, ref short_channel_id } => {
+                               0u8.write(writer)?;
+                               onion_packet.write(writer)?;
+                               short_channel_id.write(writer)?;
+                       },
+                       &PendingForwardReceiveHTLCInfo::Receive { ref payment_data } => {
+                               1u8.write(writer)?;
+                               payment_data.write(writer)?;
+                       },
+               }
                self.incoming_shared_secret.write(writer)?;
                self.payment_hash.write(writer)?;
-               self.short_channel_id.write(writer)?;
                self.amt_to_forward.write(writer)?;
                self.outgoing_cltv_value.write(writer)?;
                Ok(())
        }
 }
 
-impl<R: ::std::io::Read> Readable<R> for PendingForwardHTLCInfo {
-       fn read(reader: &mut R) -> Result<PendingForwardHTLCInfo, DecodeError> {
-               Ok(PendingForwardHTLCInfo {
-                       onion_packet: Readable::read(reader)?,
+impl<R: ::std::io::Read> Readable<R> for PendingHTLCInfo {
+       fn read(reader: &mut R) -> Result<PendingHTLCInfo, DecodeError> {
+               Ok(PendingHTLCInfo {
+                       type_data: match Readable::read(reader)? {
+                               0u8 => PendingForwardReceiveHTLCInfo::Forward {
+                                       onion_packet: Readable::read(reader)?,
+                                       short_channel_id: Readable::read(reader)?,
+                               },
+                               1u8 => PendingForwardReceiveHTLCInfo::Receive {
+                                       payment_data: Readable::read(reader)?,
+                               },
+                               _ => return Err(DecodeError::InvalidValue),
+                       },
                        incoming_shared_secret: Readable::read(reader)?,
                        payment_hash: Readable::read(reader)?,
-                       short_channel_id: Readable::read(reader)?,
                        amt_to_forward: Readable::read(reader)?,
                        outgoing_cltv_value: Readable::read(reader)?,
                })
@@ -3205,9 +3305,10 @@ impl<ChanSigner: ChannelKeys + Writeable, M: Deref> Writeable for ChannelManager
                for (payment_hash, previous_hops) in channel_state.claimable_htlcs.iter() {
                        payment_hash.write(writer)?;
                        (previous_hops.len() as u64).write(writer)?;
-                       for &(recvd_amt, ref previous_hop) in previous_hops.iter() {
-                               recvd_amt.write(writer)?;
-                               previous_hop.write(writer)?;
+                       for htlc in previous_hops.iter() {
+                               htlc.src.write(writer)?;
+                               htlc.value.write(writer)?;
+                               htlc.payment_data.write(writer)?;
                        }
                }
 
@@ -3219,6 +3320,8 @@ impl<ChanSigner: ChannelKeys + Writeable, M: Deref> Writeable for ChannelManager
                        peer_state.latest_features.write(writer)?;
                }
 
+               (self.last_node_announcement_serial.load(Ordering::Acquire) as u32).write(writer)?;
+
                Ok(())
        }
 }
@@ -3347,7 +3450,11 @@ impl<'a, R : ::std::io::Read, ChanSigner: ChannelKeys + Readable<R>, M: Deref> R
                        let previous_hops_len: u64 = Readable::read(reader)?;
                        let mut previous_hops = Vec::with_capacity(cmp::min(previous_hops_len as usize, 2));
                        for _ in 0..previous_hops_len {
-                               previous_hops.push((Readable::read(reader)?, Readable::read(reader)?));
+                               previous_hops.push(ClaimableHTLC {
+                                       src: Readable::read(reader)?,
+                                       value: Readable::read(reader)?,
+                                       payment_data: Readable::read(reader)?,
+                               });
                        }
                        claimable_htlcs.insert(payment_hash, previous_hops);
                }
@@ -3362,6 +3469,8 @@ impl<'a, R : ::std::io::Read, ChanSigner: ChannelKeys + Readable<R>, M: Deref> R
                        per_peer_state.insert(peer_pubkey, Mutex::new(peer_state));
                }
 
+               let last_node_announcement_serial: u32 = Readable::read(reader)?;
+
                let channel_manager = ChannelManager {
                        genesis_hash,
                        fee_estimator: args.fee_estimator,
@@ -3381,6 +3490,8 @@ impl<'a, R : ::std::io::Read, ChanSigner: ChannelKeys + Readable<R>, M: Deref> R
                        }),
                        our_network_key: args.keys_manager.get_node_secret(),
 
+                       last_node_announcement_serial: AtomicUsize::new(last_node_announcement_serial as usize),
+
                        per_peer_state: RwLock::new(per_peer_state),
 
                        pending_events: Mutex::new(Vec::new()),