Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
256 changes: 116 additions & 140 deletions fuzz/src/chanmon_consistency.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -142,17 +142,29 @@ impl Writer for VecWriter {
}
}

/// The LDK API requires that any time we tell it we're done persisting a `ChannelMonitor[Update]`
/// we never pass it in as the "latest" `ChannelMonitor` on startup. However, we can pass
/// out-of-date monitors as long as we never told LDK we finished persisting them, which we do by
/// storing both old `ChannelMonitor`s and ones that are "being persisted" here.
///
/// Note that such "being persisted" `ChannelMonitor`s are stored in `ChannelManager` and will
/// simply be replayed on startup.
struct LatestMonitorState {
/// The latest monitor id which we told LDK we've persisted
persisted_monitor_id: u64,
/// The latest serialized `ChannelMonitor` that we told LDK we persisted.
persisted_monitor: Vec<u8>,
/// A set of (monitor id, serialized `ChannelMonitor`)s which we're currently "persisting",
/// from LDK's perspective.
pending_monitors: Vec<(u64, Vec<u8>)>,
}

struct TestChainMonitor {
pub logger: Arc<dyn Logger>,
pub keys: Arc<KeyProvider>,
pub persister: Arc<TestPersister>,
pub chain_monitor: Arc<chainmonitor::ChainMonitor<TestChannelSigner, Arc<dyn chain::Filter>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>, Arc<TestPersister>>>,
// If we reload a node with an old copy of ChannelMonitors, the ChannelManager deserialization
// logic will automatically force-close our channels for us (as we don't have an up-to-date
// monitor implying we are not able to punish misbehaving counterparties). Because this test
// "fails" if we ever force-close a channel, we avoid doing so, always saving the latest
// fully-serialized monitor state here, as well as the corresponding update_id.
pub latest_monitors: Mutex<HashMap<OutPoint, (u64, Vec<u8>)>>,
pub latest_monitors: Mutex<HashMap<OutPoint, LatestMonitorState>>,
}
impl TestChainMonitor {
pub fn new(broadcaster: Arc<TestBroadcaster>, logger: Arc<dyn Logger>, feeest: Arc<FuzzEstimator>, persister: Arc<TestPersister>, keys: Arc<KeyProvider>) -> Self {
Expand All@@ -169,22 +181,47 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
fn watch_channel(&self, funding_txo: OutPoint, monitor: channelmonitor::ChannelMonitor<TestChannelSigner>) -> Result<chain::ChannelMonitorUpdateStatus, ()> {
let mut ser = VecWriter(Vec::new());
monitor.write(&mut ser).unwrap();
if let Some(_) = self.latest_monitors.lock().unwrap().insert(funding_txo, (monitor.get_latest_update_id(), ser.0)) {
let monitor_id = monitor.get_latest_update_id();
let res = self.chain_monitor.watch_channel(funding_txo, monitor);
let state = match res {
Ok(chain::ChannelMonitorUpdateStatus::Completed) => {
LatestMonitorState {
persisted_monitor_id: monitor_id, persisted_monitor: ser.0,
pending_monitors: Vec::new(),
}
},
Ok(chain::ChannelMonitorUpdateStatus::InProgress) =>
panic!("The test currently doesn't test initial-persistence via the async pipeline"),
Ok(chain::ChannelMonitorUpdateStatus::UnrecoverableError) => panic!(),
Err(()) => panic!(),
};
if self.latest_monitors.lock().unwrap().insert(funding_txo, state).is_some() {
panic!("Already had monitor pre-watch_channel");
}
self.chain_monitor.watch_channel(funding_txo, monitor)
res
}

fn update_channel(&self, funding_txo: OutPoint, update: &channelmonitor::ChannelMonitorUpdate) -> chain::ChannelMonitorUpdateStatus {
let mut map_lock = self.latest_monitors.lock().unwrap();
let map_entry = map_lock.get_mut(&funding_txo).expect("Didn't have monitor on update call");
let latest_monitor_data = map_entry.pending_monitors.last().as_ref().map(|(_, data)| data).unwrap_or(&map_entry.persisted_monitor);
let deserialized_monitor = <(BlockHash, channelmonitor::ChannelMonitor<TestChannelSigner>)>::
read(&mut Cursor::new(&map_entry.1), (&*self.keys, &*self.keys)).unwrap().1;
read(&mut Cursor::new(&latest_monitor_data), (&*self.keys, &*self.keys)).unwrap().1;
deserialized_monitor.update_monitor(update, &&TestBroadcaster{}, &&FuzzEstimator { ret_val: atomic::AtomicU32::new(253) }, &self.logger).unwrap();
let mut ser = VecWriter(Vec::new());
deserialized_monitor.write(&mut ser).unwrap();
*map_entry = (update.update_id, ser.0);
self.chain_monitor.update_channel(funding_txo, update)
let res = self.chain_monitor.update_channel(funding_txo, update);
match res {
chain::ChannelMonitorUpdateStatus::Completed => {
map_entry.persisted_monitor_id = update.update_id;
map_entry.persisted_monitor = ser.0;
},
chain::ChannelMonitorUpdateStatus::InProgress => {
map_entry.pending_monitors.push((update.update_id, ser.0));
},
chain::ChannelMonitorUpdateStatus::UnrecoverableError => panic!(),
}
res
}

fn release_pending_monitor_events(&self) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, Option<PublicKey>)> {
Expand DownExpand Up@@ -511,9 +548,15 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {

let mut monitors = new_hash_map();
let mut old_monitors = $old_monitors.latest_monitors.lock().unwrap();
for (outpoint, (update_id, monitor_ser)) in old_monitors.drain() {
monitors.insert(outpoint, <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(&mut Cursor::new(&monitor_ser), (&*$keys_manager, &*$keys_manager)).expect("Failed to read monitor").1);
chain_monitor.latest_monitors.lock().unwrap().insert(outpoint, (update_id, monitor_ser));
for (outpoint, mut prev_state) in old_monitors.drain() {
monitors.insert(outpoint, <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(
&mut Cursor::new(&prev_state.persisted_monitor), (&*$keys_manager, &*$keys_manager)
).expect("Failed to read monitor").1);
// Wipe any `ChannelMonitor`s which we never told LDK we finished persisting,
// considering them discarded. LDK should replay these for us as they're stored in
// the `ChannelManager`.
prev_state.pending_monitors.clear();
chain_monitor.latest_monitors.lock().unwrap().insert(outpoint, prev_state);
}
let mut monitor_refs = new_hash_map();
for (outpoint, monitor) in monitors.iter_mut() {
Expand DownExpand Up@@ -1040,6 +1083,43 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
} }
}

let complete_first = |v: &mut Vec<_>| if !v.is_empty() { Some(v.remove(0)) } else { None };
let complete_second = |v: &mut Vec<_>| if v.len() > 1 { Some(v.remove(1)) } else { None };
let complete_monitor_update = |
monitor: &Arc<TestChainMonitor>, chan_funding,
compl_selector: &dyn Fn(&mut Vec<(u64, Vec<u8>)>) -> Option<(u64, Vec<u8>)>,
| {
if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
assert!(
state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
"updates should be sorted by id"
);
if let Some((id, data)) = compl_selector(&mut state.pending_monitors) {
monitor.chain_monitor.channel_monitor_updated(*chan_funding, id).unwrap();
if id > state.persisted_monitor_id {
state.persisted_monitor_id = id;
state.persisted_monitor = data;
}
}
}
};

let complete_all_monitor_updates = |monitor: &Arc<TestChainMonitor>, chan_funding| {
Comment thread
tnull marked this conversation as resolved.
if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
assert!(
state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
"updates should be sorted by id"
);
for (id, data) in state.pending_monitors.drain(..) {
monitor.chain_monitor.channel_monitor_updated(*chan_funding, id).unwrap();
if id > state.persisted_monitor_id {
state.persisted_monitor_id = id;
state.persisted_monitor = data;
}
}
}
};

let v = get_slice!(1)[0];
out.locked_write(format!("READ A BYTE! HANDLING INPUT {:x}...........\n", v).as_bytes());
match v {
Expand All@@ -1054,30 +1134,10 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
0x05 => *monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed,
0x06 => *monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed,

0x08 => {
if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[0].process_monitor_events();
}
},
0x09 => {
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[1].process_monitor_events();
}
},
0x0a => {
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[1].process_monitor_events();
}
},
0x0b => {
if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_c.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[2].process_monitor_events();
}
},
0x08 => complete_all_monitor_updates(&monitor_a, &chan_1_funding),
0x09 => complete_all_monitor_updates(&monitor_b, &chan_1_funding),
0x0a => complete_all_monitor_updates(&monitor_b, &chan_2_funding),
0x0b => complete_all_monitor_updates(&monitor_c, &chan_2_funding),

0x0c => {
if !chan_a_disconnected {
Expand DownExpand Up@@ -1285,119 +1345,35 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
},
0x89 => { fee_est_c.ret_val.store(253, atomic::Ordering::Release); nodes[2].maybe_update_chan_fees(); },

0xf0 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf1 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf2 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf0 => complete_monitor_update(&monitor_a, &chan_1_funding, &complete_first),
0xf1 => complete_monitor_update(&monitor_a, &chan_1_funding, &complete_second),
0xf2 => complete_monitor_update(&monitor_a, &chan_1_funding, &Vec::pop),

0xf4 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf5 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf6 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf4 => complete_monitor_update(&monitor_b, &chan_1_funding, &complete_first),
0xf5 => complete_monitor_update(&monitor_b, &chan_1_funding, &complete_second),
0xf6 => complete_monitor_update(&monitor_b, &chan_1_funding, &Vec::pop),

0xf8 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf9 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xfa => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf8 => complete_monitor_update(&monitor_b, &chan_2_funding, &complete_first),
0xf9 => complete_monitor_update(&monitor_b, &chan_2_funding, &complete_second),
0xfa => complete_monitor_update(&monitor_b, &chan_2_funding, &Vec::pop),

0xfc => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfd => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfe => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfc => complete_monitor_update(&monitor_c, &chan_2_funding, &complete_first),
0xfd => complete_monitor_update(&monitor_c, &chan_2_funding, &complete_second),
0xfe => complete_monitor_update(&monitor_c, &chan_2_funding, &Vec::pop),

0xff => {
// Test that no channel is in a stuck state where neither party can send funds even
// after we resolve all pending events.
// First make sure there are no pending monitor updates, resetting the error state
// and calling force_channel_monitor_updated for each monitor.
// First make sure there are no pending monitor updates and further update
// operations complete.
*monitor_a.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
*monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
*monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;

if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[0].process_monitor_events();
}
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[1].process_monitor_events();
}
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[1].process_monitor_events();
}
if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_c.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[2].process_monitor_events();
}
complete_all_monitor_updates(&monitor_a, &chan_1_funding);
complete_all_monitor_updates(&monitor_b, &chan_1_funding);
complete_all_monitor_updates(&monitor_b, &chan_2_funding);
complete_all_monitor_updates(&monitor_c, &chan_2_funding);

// Next, make sure peers are all connected to each other
if chan_a_disconnected {
Expand Down
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Add copy buttons to all
 blocks\n(function() {\n function addCopyButtons() {\n document.querySelectorAll('pre code').forEach(function(codeBlock) {\n if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;\n codeBlock.parentElement.setAttribute('data-copy-added', 'true');\n \n var btn = document.createElement('button');\n btn.textContent = 'Copy';\n btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';\n btn.onmouseover = function() { this.style.opacity = '1'; };\n btn.onmouseout = function() { this.style.opacity = '0.7'; };\n btn.onclick = function() {\n navigator.clipboard.writeText(codeBlock.textContent).then(function() {\n btn.textContent = 'Copied!';\n setTimeout(function() { btn.textContent = 'Copy'; }, 1500);\n });\n };\n codeBlock.parentElement.style.position = 'relative';\n codeBlock.parentElement.appendChild(btn);\n });\n }\n \n addCopyButtons();\n \n // Re-run on dynamic content\n var observer = new MutationObserver(addCopyButtons);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Add Copy Buttons to Code Blocks");
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
256 changes: 116 additions & 140 deletions fuzz/src/chanmon_consistency.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -142,17 +142,29 @@ impl Writer for VecWriter {
}
}

/// The LDK API requires that any time we tell it we're done persisting a `ChannelMonitor[Update]`
/// we never pass it in as the "latest" `ChannelMonitor` on startup. However, we can pass
/// out-of-date monitors as long as we never told LDK we finished persisting them, which we do by
/// storing both old `ChannelMonitor`s and ones that are "being persisted" here.
///
/// Note that such "being persisted" `ChannelMonitor`s are stored in `ChannelManager` and will
/// simply be replayed on startup.
struct LatestMonitorState {
/// The latest monitor id which we told LDK we've persisted
persisted_monitor_id: u64,
/// The latest serialized `ChannelMonitor` that we told LDK we persisted.
persisted_monitor: Vec<u8>,
/// A set of (monitor id, serialized `ChannelMonitor`)s which we're currently "persisting",
/// from LDK's perspective.
pending_monitors: Vec<(u64, Vec<u8>)>,
}

struct TestChainMonitor {
pub logger: Arc<dyn Logger>,
pub keys: Arc<KeyProvider>,
pub persister: Arc<TestPersister>,
pub chain_monitor: Arc<chainmonitor::ChainMonitor<TestChannelSigner, Arc<dyn chain::Filter>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>, Arc<TestPersister>>>,
// If we reload a node with an old copy of ChannelMonitors, the ChannelManager deserialization
// logic will automatically force-close our channels for us (as we don't have an up-to-date
// monitor implying we are not able to punish misbehaving counterparties). Because this test
// "fails" if we ever force-close a channel, we avoid doing so, always saving the latest
// fully-serialized monitor state here, as well as the corresponding update_id.
pub latest_monitors: Mutex<HashMap<OutPoint, (u64, Vec<u8>)>>,
pub latest_monitors: Mutex<HashMap<OutPoint, LatestMonitorState>>,
}
impl TestChainMonitor {
pub fn new(broadcaster: Arc<TestBroadcaster>, logger: Arc<dyn Logger>, feeest: Arc<FuzzEstimator>, persister: Arc<TestPersister>, keys: Arc<KeyProvider>) -> Self {
Expand All@@ -169,22 +181,47 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
fn watch_channel(&self, funding_txo: OutPoint, monitor: channelmonitor::ChannelMonitor<TestChannelSigner>) -> Result<chain::ChannelMonitorUpdateStatus, ()> {
let mut ser = VecWriter(Vec::new());
monitor.write(&mut ser).unwrap();
if let Some(_) = self.latest_monitors.lock().unwrap().insert(funding_txo, (monitor.get_latest_update_id(), ser.0)) {
let monitor_id = monitor.get_latest_update_id();
let res = self.chain_monitor.watch_channel(funding_txo, monitor);
let state = match res {
Ok(chain::ChannelMonitorUpdateStatus::Completed) => {
LatestMonitorState {
persisted_monitor_id: monitor_id, persisted_monitor: ser.0,
pending_monitors: Vec::new(),
}
},
Ok(chain::ChannelMonitorUpdateStatus::InProgress) =>
panic!("The test currently doesn't test initial-persistence via the async pipeline"),
Ok(chain::ChannelMonitorUpdateStatus::UnrecoverableError) => panic!(),
Err(()) => panic!(),
};
if self.latest_monitors.lock().unwrap().insert(funding_txo, state).is_some() {
panic!("Already had monitor pre-watch_channel");
}
self.chain_monitor.watch_channel(funding_txo, monitor)
res
}

fn update_channel(&self, funding_txo: OutPoint, update: &channelmonitor::ChannelMonitorUpdate) -> chain::ChannelMonitorUpdateStatus {
let mut map_lock = self.latest_monitors.lock().unwrap();
let map_entry = map_lock.get_mut(&funding_txo).expect("Didn't have monitor on update call");
let latest_monitor_data = map_entry.pending_monitors.last().as_ref().map(|(_, data)| data).unwrap_or(&map_entry.persisted_monitor);
let deserialized_monitor = <(BlockHash, channelmonitor::ChannelMonitor<TestChannelSigner>)>::
read(&mut Cursor::new(&map_entry.1), (&*self.keys, &*self.keys)).unwrap().1;
read(&mut Cursor::new(&latest_monitor_data), (&*self.keys, &*self.keys)).unwrap().1;
deserialized_monitor.update_monitor(update, &&TestBroadcaster{}, &&FuzzEstimator { ret_val: atomic::AtomicU32::new(253) }, &self.logger).unwrap();
let mut ser = VecWriter(Vec::new());
deserialized_monitor.write(&mut ser).unwrap();
*map_entry = (update.update_id, ser.0);
self.chain_monitor.update_channel(funding_txo, update)
let res = self.chain_monitor.update_channel(funding_txo, update);
match res {
chain::ChannelMonitorUpdateStatus::Completed => {
map_entry.persisted_monitor_id = update.update_id;
map_entry.persisted_monitor = ser.0;
},
chain::ChannelMonitorUpdateStatus::InProgress => {
map_entry.pending_monitors.push((update.update_id, ser.0));
},
chain::ChannelMonitorUpdateStatus::UnrecoverableError => panic!(),
}
res
}

fn release_pending_monitor_events(&self) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, Option<PublicKey>)> {
Expand DownExpand Up@@ -511,9 +548,15 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {

let mut monitors = new_hash_map();
let mut old_monitors = $old_monitors.latest_monitors.lock().unwrap();
for (outpoint, (update_id, monitor_ser)) in old_monitors.drain() {
monitors.insert(outpoint, <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(&mut Cursor::new(&monitor_ser), (&*$keys_manager, &*$keys_manager)).expect("Failed to read monitor").1);
chain_monitor.latest_monitors.lock().unwrap().insert(outpoint, (update_id, monitor_ser));
for (outpoint, mut prev_state) in old_monitors.drain() {
monitors.insert(outpoint, <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(
&mut Cursor::new(&prev_state.persisted_monitor), (&*$keys_manager, &*$keys_manager)
).expect("Failed to read monitor").1);
// Wipe any `ChannelMonitor`s which we never told LDK we finished persisting,
// considering them discarded. LDK should replay these for us as they're stored in
// the `ChannelManager`.
prev_state.pending_monitors.clear();
chain_monitor.latest_monitors.lock().unwrap().insert(outpoint, prev_state);
}
let mut monitor_refs = new_hash_map();
for (outpoint, monitor) in monitors.iter_mut() {
Expand DownExpand Up@@ -1040,6 +1083,43 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
} }
}

let complete_first = |v: &mut Vec<_>| if !v.is_empty() { Some(v.remove(0)) } else { None };
let complete_second = |v: &mut Vec<_>| if v.len() > 1 { Some(v.remove(1)) } else { None };
let complete_monitor_update = |
monitor: &Arc<TestChainMonitor>, chan_funding,
compl_selector: &dyn Fn(&mut Vec<(u64, Vec<u8>)>) -> Option<(u64, Vec<u8>)>,
| {
if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
assert!(
state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
"updates should be sorted by id"
);
if let Some((id, data)) = compl_selector(&mut state.pending_monitors) {
monitor.chain_monitor.channel_monitor_updated(*chan_funding, id).unwrap();
if id > state.persisted_monitor_id {
state.persisted_monitor_id = id;
state.persisted_monitor = data;
}
}
}
};

let complete_all_monitor_updates = |monitor: &Arc<TestChainMonitor>, chan_funding| {
Comment thread
tnull marked this conversation as resolved.
if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
assert!(
state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
"updates should be sorted by id"
);
for (id, data) in state.pending_monitors.drain(..) {
monitor.chain_monitor.channel_monitor_updated(*chan_funding, id).unwrap();
if id > state.persisted_monitor_id {
state.persisted_monitor_id = id;
state.persisted_monitor = data;
}
}
}
};

let v = get_slice!(1)[0];
out.locked_write(format!("READ A BYTE! HANDLING INPUT {:x}...........\n", v).as_bytes());
match v {
Expand All@@ -1054,30 +1134,10 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
0x05 => *monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed,
0x06 => *monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed,

0x08 => {
if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[0].process_monitor_events();
}
},
0x09 => {
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[1].process_monitor_events();
}
},
0x0a => {
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[1].process_monitor_events();
}
},
0x0b => {
if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_c.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[2].process_monitor_events();
}
},
0x08 => complete_all_monitor_updates(&monitor_a, &chan_1_funding),
0x09 => complete_all_monitor_updates(&monitor_b, &chan_1_funding),
0x0a => complete_all_monitor_updates(&monitor_b, &chan_2_funding),
0x0b => complete_all_monitor_updates(&monitor_c, &chan_2_funding),

0x0c => {
if !chan_a_disconnected {
Expand DownExpand Up@@ -1285,119 +1345,35 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
},
0x89 => { fee_est_c.ret_val.store(253, atomic::Ordering::Release); nodes[2].maybe_update_chan_fees(); },

0xf0 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf1 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf2 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf0 => complete_monitor_update(&monitor_a, &chan_1_funding, &complete_first),
0xf1 => complete_monitor_update(&monitor_a, &chan_1_funding, &complete_second),
0xf2 => complete_monitor_update(&monitor_a, &chan_1_funding, &Vec::pop),

0xf4 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf5 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf6 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf4 => complete_monitor_update(&monitor_b, &chan_1_funding, &complete_first),
0xf5 => complete_monitor_update(&monitor_b, &chan_1_funding, &complete_second),
0xf6 => complete_monitor_update(&monitor_b, &chan_1_funding, &Vec::pop),

0xf8 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf9 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xfa => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf8 => complete_monitor_update(&monitor_b, &chan_2_funding, &complete_first),
0xf9 => complete_monitor_update(&monitor_b, &chan_2_funding, &complete_second),
0xfa => complete_monitor_update(&monitor_b, &chan_2_funding, &Vec::pop),

0xfc => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfd => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfe => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfc => complete_monitor_update(&monitor_c, &chan_2_funding, &complete_first),
0xfd => complete_monitor_update(&monitor_c, &chan_2_funding, &complete_second),
0xfe => complete_monitor_update(&monitor_c, &chan_2_funding, &Vec::pop),

0xff => {
// Test that no channel is in a stuck state where neither party can send funds even
// after we resolve all pending events.
// First make sure there are no pending monitor updates, resetting the error state
// and calling force_channel_monitor_updated for each monitor.
// First make sure there are no pending monitor updates and further update
// operations complete.
*monitor_a.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
*monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
*monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;

if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[0].process_monitor_events();
}
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[1].process_monitor_events();
}
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[1].process_monitor_events();
}
if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_c.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[2].process_monitor_events();
}
complete_all_monitor_updates(&monitor_a, &chan_1_funding);
complete_all_monitor_updates(&monitor_b, &chan_1_funding);
complete_all_monitor_updates(&monitor_b, &chan_2_funding);
complete_all_monitor_updates(&monitor_c, &chan_2_funding);

// Next, make sure peers are all connected to each other
if chan_a_disconnected {
Expand Down
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Force GitHub README to respect dark mode\n(function() {\n var style = document.createElement('style');\n style.textContent = '\n .markdown-body {\n color-scheme: dark light;\n }\n .markdown-body pre { background: #161b22 !important; }\n .markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }\n .markdown-body table th, .markdown-body table td { border-color: #30363d !important; }\n .markdown-body img { background: #0d1117; }\n .markdown-body blockquote { border-left-color: #8b949e; }\n .markdown-body hr { border-color: #30363d; }\n ';\n document.head.appendChild(style);\n})();", "GitHub Dark Mode README Fix"); } } catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
256 changes: 116 additions & 140 deletions fuzz/src/chanmon_consistency.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -142,17 +142,29 @@ impl Writer for VecWriter {
}
}

/// The LDK API requires that any time we tell it we're done persisting a `ChannelMonitor[Update]`
/// we never pass it in as the "latest" `ChannelMonitor` on startup. However, we can pass
/// out-of-date monitors as long as we never told LDK we finished persisting them, which we do by
/// storing both old `ChannelMonitor`s and ones that are "being persisted" here.
///
/// Note that such "being persisted" `ChannelMonitor`s are stored in `ChannelManager` and will
/// simply be replayed on startup.
struct LatestMonitorState {
/// The latest monitor id which we told LDK we've persisted
persisted_monitor_id: u64,
/// The latest serialized `ChannelMonitor` that we told LDK we persisted.
persisted_monitor: Vec<u8>,
/// A set of (monitor id, serialized `ChannelMonitor`)s which we're currently "persisting",
/// from LDK's perspective.
pending_monitors: Vec<(u64, Vec<u8>)>,
}

struct TestChainMonitor {
pub logger: Arc<dyn Logger>,
pub keys: Arc<KeyProvider>,
pub persister: Arc<TestPersister>,
pub chain_monitor: Arc<chainmonitor::ChainMonitor<TestChannelSigner, Arc<dyn chain::Filter>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>, Arc<TestPersister>>>,
// If we reload a node with an old copy of ChannelMonitors, the ChannelManager deserialization
// logic will automatically force-close our channels for us (as we don't have an up-to-date
// monitor implying we are not able to punish misbehaving counterparties). Because this test
// "fails" if we ever force-close a channel, we avoid doing so, always saving the latest
// fully-serialized monitor state here, as well as the corresponding update_id.
pub latest_monitors: Mutex<HashMap<OutPoint, (u64, Vec<u8>)>>,
pub latest_monitors: Mutex<HashMap<OutPoint, LatestMonitorState>>,
}
impl TestChainMonitor {
pub fn new(broadcaster: Arc<TestBroadcaster>, logger: Arc<dyn Logger>, feeest: Arc<FuzzEstimator>, persister: Arc<TestPersister>, keys: Arc<KeyProvider>) -> Self {
Expand All@@ -169,22 +181,47 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
fn watch_channel(&self, funding_txo: OutPoint, monitor: channelmonitor::ChannelMonitor<TestChannelSigner>) -> Result<chain::ChannelMonitorUpdateStatus, ()> {
let mut ser = VecWriter(Vec::new());
monitor.write(&mut ser).unwrap();
if let Some(_) = self.latest_monitors.lock().unwrap().insert(funding_txo, (monitor.get_latest_update_id(), ser.0)) {
let monitor_id = monitor.get_latest_update_id();
let res = self.chain_monitor.watch_channel(funding_txo, monitor);
let state = match res {
Ok(chain::ChannelMonitorUpdateStatus::Completed) => {
LatestMonitorState {
persisted_monitor_id: monitor_id, persisted_monitor: ser.0,
pending_monitors: Vec::new(),
}
},
Ok(chain::ChannelMonitorUpdateStatus::InProgress) =>
panic!("The test currently doesn't test initial-persistence via the async pipeline"),
Ok(chain::ChannelMonitorUpdateStatus::UnrecoverableError) => panic!(),
Err(()) => panic!(),
};
if self.latest_monitors.lock().unwrap().insert(funding_txo, state).is_some() {
panic!("Already had monitor pre-watch_channel");
}
self.chain_monitor.watch_channel(funding_txo, monitor)
res
}

fn update_channel(&self, funding_txo: OutPoint, update: &channelmonitor::ChannelMonitorUpdate) -> chain::ChannelMonitorUpdateStatus {
let mut map_lock = self.latest_monitors.lock().unwrap();
let map_entry = map_lock.get_mut(&funding_txo).expect("Didn't have monitor on update call");
let latest_monitor_data = map_entry.pending_monitors.last().as_ref().map(|(_, data)| data).unwrap_or(&map_entry.persisted_monitor);
let deserialized_monitor = <(BlockHash, channelmonitor::ChannelMonitor<TestChannelSigner>)>::
read(&mut Cursor::new(&map_entry.1), (&*self.keys, &*self.keys)).unwrap().1;
read(&mut Cursor::new(&latest_monitor_data), (&*self.keys, &*self.keys)).unwrap().1;
deserialized_monitor.update_monitor(update, &&TestBroadcaster{}, &&FuzzEstimator { ret_val: atomic::AtomicU32::new(253) }, &self.logger).unwrap();
let mut ser = VecWriter(Vec::new());
deserialized_monitor.write(&mut ser).unwrap();
*map_entry = (update.update_id, ser.0);
self.chain_monitor.update_channel(funding_txo, update)
let res = self.chain_monitor.update_channel(funding_txo, update);
match res {
chain::ChannelMonitorUpdateStatus::Completed => {
map_entry.persisted_monitor_id = update.update_id;
map_entry.persisted_monitor = ser.0;
},
chain::ChannelMonitorUpdateStatus::InProgress => {
map_entry.pending_monitors.push((update.update_id, ser.0));
},
chain::ChannelMonitorUpdateStatus::UnrecoverableError => panic!(),
}
res
}

fn release_pending_monitor_events(&self) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, Option<PublicKey>)> {
Expand DownExpand Up@@ -511,9 +548,15 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {

let mut monitors = new_hash_map();
let mut old_monitors = $old_monitors.latest_monitors.lock().unwrap();
for (outpoint, (update_id, monitor_ser)) in old_monitors.drain() {
monitors.insert(outpoint, <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(&mut Cursor::new(&monitor_ser), (&*$keys_manager, &*$keys_manager)).expect("Failed to read monitor").1);
chain_monitor.latest_monitors.lock().unwrap().insert(outpoint, (update_id, monitor_ser));
for (outpoint, mut prev_state) in old_monitors.drain() {
monitors.insert(outpoint, <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(
&mut Cursor::new(&prev_state.persisted_monitor), (&*$keys_manager, &*$keys_manager)
).expect("Failed to read monitor").1);
// Wipe any `ChannelMonitor`s which we never told LDK we finished persisting,
// considering them discarded. LDK should replay these for us as they're stored in
// the `ChannelManager`.
prev_state.pending_monitors.clear();
chain_monitor.latest_monitors.lock().unwrap().insert(outpoint, prev_state);
}
let mut monitor_refs = new_hash_map();
for (outpoint, monitor) in monitors.iter_mut() {
Expand DownExpand Up@@ -1040,6 +1083,43 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
} }
}

let complete_first = |v: &mut Vec<_>| if !v.is_empty() { Some(v.remove(0)) } else { None };
let complete_second = |v: &mut Vec<_>| if v.len() > 1 { Some(v.remove(1)) } else { None };
let complete_monitor_update = |
monitor: &Arc<TestChainMonitor>, chan_funding,
compl_selector: &dyn Fn(&mut Vec<(u64, Vec<u8>)>) -> Option<(u64, Vec<u8>)>,
| {
if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
assert!(
state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
"updates should be sorted by id"
);
if let Some((id, data)) = compl_selector(&mut state.pending_monitors) {
monitor.chain_monitor.channel_monitor_updated(*chan_funding, id).unwrap();
if id > state.persisted_monitor_id {
state.persisted_monitor_id = id;
state.persisted_monitor = data;
}
}
}
};

let complete_all_monitor_updates = |monitor: &Arc<TestChainMonitor>, chan_funding| {
Comment thread
tnull marked this conversation as resolved.
if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
assert!(
state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
"updates should be sorted by id"
);
for (id, data) in state.pending_monitors.drain(..) {
monitor.chain_monitor.channel_monitor_updated(*chan_funding, id).unwrap();
if id > state.persisted_monitor_id {
state.persisted_monitor_id = id;
state.persisted_monitor = data;
}
}
}
};

let v = get_slice!(1)[0];
out.locked_write(format!("READ A BYTE! HANDLING INPUT {:x}...........\n", v).as_bytes());
match v {
Expand All@@ -1054,30 +1134,10 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
0x05 => *monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed,
0x06 => *monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed,

0x08 => {
if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[0].process_monitor_events();
}
},
0x09 => {
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[1].process_monitor_events();
}
},
0x0a => {
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[1].process_monitor_events();
}
},
0x0b => {
if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_c.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[2].process_monitor_events();
}
},
0x08 => complete_all_monitor_updates(&monitor_a, &chan_1_funding),
0x09 => complete_all_monitor_updates(&monitor_b, &chan_1_funding),
0x0a => complete_all_monitor_updates(&monitor_b, &chan_2_funding),
0x0b => complete_all_monitor_updates(&monitor_c, &chan_2_funding),

0x0c => {
if !chan_a_disconnected {
Expand DownExpand Up@@ -1285,119 +1345,35 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
},
0x89 => { fee_est_c.ret_val.store(253, atomic::Ordering::Release); nodes[2].maybe_update_chan_fees(); },

0xf0 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf1 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf2 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf0 => complete_monitor_update(&monitor_a, &chan_1_funding, &complete_first),
0xf1 => complete_monitor_update(&monitor_a, &chan_1_funding, &complete_second),
0xf2 => complete_monitor_update(&monitor_a, &chan_1_funding, &Vec::pop),

0xf4 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf5 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf6 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf4 => complete_monitor_update(&monitor_b, &chan_1_funding, &complete_first),
0xf5 => complete_monitor_update(&monitor_b, &chan_1_funding, &complete_second),
0xf6 => complete_monitor_update(&monitor_b, &chan_1_funding, &Vec::pop),

0xf8 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf9 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xfa => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf8 => complete_monitor_update(&monitor_b, &chan_2_funding, &complete_first),
0xf9 => complete_monitor_update(&monitor_b, &chan_2_funding, &complete_second),
0xfa => complete_monitor_update(&monitor_b, &chan_2_funding, &Vec::pop),

0xfc => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfd => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfe => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfc => complete_monitor_update(&monitor_c, &chan_2_funding, &complete_first),
0xfd => complete_monitor_update(&monitor_c, &chan_2_funding, &complete_second),
0xfe => complete_monitor_update(&monitor_c, &chan_2_funding, &Vec::pop),

0xff => {
// Test that no channel is in a stuck state where neither party can send funds even
// after we resolve all pending events.
// First make sure there are no pending monitor updates, resetting the error state
// and calling force_channel_monitor_updated for each monitor.
// First make sure there are no pending monitor updates and further update
// operations complete.
*monitor_a.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
*monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
*monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;

if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[0].process_monitor_events();
}
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[1].process_monitor_events();
}
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[1].process_monitor_events();
}
if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_c.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[2].process_monitor_events();
}
complete_all_monitor_updates(&monitor_a, &chan_1_funding);
complete_all_monitor_updates(&monitor_b, &chan_1_funding);
complete_all_monitor_updates(&monitor_b, &chan_2_funding);
complete_all_monitor_updates(&monitor_c, &chan_2_funding);

// Next, make sure peers are all connected to each other
if chan_a_disconnected {
Expand Down
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Highlight search terms from Google/DuckDuckGo/Bing referrer\n(function() {\n var ref = document.referrer;\n var terms = [];\n \n if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {\n var url = new URL(ref);\n var q = url.searchParams.get('q') || url.searchParams.get('p');\n if (q) {\n terms = q.split(/\\s+/).filter(function(t) { return t.length > 2; });\n }\n }\n \n if (terms.length === 0) return;\n \n var style = document.createElement('style');\n style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';\n document.head.appendChild(style);\n \n function highlight(node) {\n if (node.nodeType === 3) { // text node\n var text = node.textContent;\n var found = false;\n terms.forEach(function(term) {\n var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\\]\\\\]/g, '\\\\') + ')', 'gi');\n if (regex.test(text)) {\n found = true;\n var frag = document.createDocumentFragment();\n var parts = text.split(regex);\n parts.forEach(function(part, i) {\n if (i % 2 === 0) {\n frag.appendChild(document.createTextNode(part));\n } else {\n var span = document.createElement('span');\n span.className = 'userscript-highlight';\n span.textContent = part;\n frag.appendChild(span);\n }\n });\n node.parentNode.replaceChild(frag, node);\n }\n });\n } else if (node.nodeType === 1 && node.childNodes) { // element\n var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];\n if (!skipTags.includes(node.tagName)) {\n Array.from(node.childNodes).forEach(highlight);\n }\n }\n }\n \n highlight(document.body);\n \n // Re-highlight on dynamic content\n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1 || node.nodeType === 3) highlight(node);\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Highlight Search Terms"); } } catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
256 changes: 116 additions & 140 deletions fuzz/src/chanmon_consistency.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -142,17 +142,29 @@ impl Writer for VecWriter {
}
}

/// The LDK API requires that any time we tell it we're done persisting a `ChannelMonitor[Update]`
/// we never pass it in as the "latest" `ChannelMonitor` on startup. However, we can pass
/// out-of-date monitors as long as we never told LDK we finished persisting them, which we do by
/// storing both old `ChannelMonitor`s and ones that are "being persisted" here.
///
/// Note that such "being persisted" `ChannelMonitor`s are stored in `ChannelManager` and will
/// simply be replayed on startup.
struct LatestMonitorState {
/// The latest monitor id which we told LDK we've persisted
persisted_monitor_id: u64,
/// The latest serialized `ChannelMonitor` that we told LDK we persisted.
persisted_monitor: Vec<u8>,
/// A set of (monitor id, serialized `ChannelMonitor`)s which we're currently "persisting",
/// from LDK's perspective.
pending_monitors: Vec<(u64, Vec<u8>)>,
}

struct TestChainMonitor {
pub logger: Arc<dyn Logger>,
pub keys: Arc<KeyProvider>,
pub persister: Arc<TestPersister>,
pub chain_monitor: Arc<chainmonitor::ChainMonitor<TestChannelSigner, Arc<dyn chain::Filter>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>, Arc<TestPersister>>>,
// If we reload a node with an old copy of ChannelMonitors, the ChannelManager deserialization
// logic will automatically force-close our channels for us (as we don't have an up-to-date
// monitor implying we are not able to punish misbehaving counterparties). Because this test
// "fails" if we ever force-close a channel, we avoid doing so, always saving the latest
// fully-serialized monitor state here, as well as the corresponding update_id.
pub latest_monitors: Mutex<HashMap<OutPoint, (u64, Vec<u8>)>>,
pub latest_monitors: Mutex<HashMap<OutPoint, LatestMonitorState>>,
}
impl TestChainMonitor {
pub fn new(broadcaster: Arc<TestBroadcaster>, logger: Arc<dyn Logger>, feeest: Arc<FuzzEstimator>, persister: Arc<TestPersister>, keys: Arc<KeyProvider>) -> Self {
Expand All@@ -169,22 +181,47 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
fn watch_channel(&self, funding_txo: OutPoint, monitor: channelmonitor::ChannelMonitor<TestChannelSigner>) -> Result<chain::ChannelMonitorUpdateStatus, ()> {
let mut ser = VecWriter(Vec::new());
monitor.write(&mut ser).unwrap();
if let Some(_) = self.latest_monitors.lock().unwrap().insert(funding_txo, (monitor.get_latest_update_id(), ser.0)) {
let monitor_id = monitor.get_latest_update_id();
let res = self.chain_monitor.watch_channel(funding_txo, monitor);
let state = match res {
Ok(chain::ChannelMonitorUpdateStatus::Completed) => {
LatestMonitorState {
persisted_monitor_id: monitor_id, persisted_monitor: ser.0,
pending_monitors: Vec::new(),
}
},
Ok(chain::ChannelMonitorUpdateStatus::InProgress) =>
panic!("The test currently doesn't test initial-persistence via the async pipeline"),
Ok(chain::ChannelMonitorUpdateStatus::UnrecoverableError) => panic!(),
Err(()) => panic!(),
};
if self.latest_monitors.lock().unwrap().insert(funding_txo, state).is_some() {
panic!("Already had monitor pre-watch_channel");
}
self.chain_monitor.watch_channel(funding_txo, monitor)
res
}

fn update_channel(&self, funding_txo: OutPoint, update: &channelmonitor::ChannelMonitorUpdate) -> chain::ChannelMonitorUpdateStatus {
let mut map_lock = self.latest_monitors.lock().unwrap();
let map_entry = map_lock.get_mut(&funding_txo).expect("Didn't have monitor on update call");
let latest_monitor_data = map_entry.pending_monitors.last().as_ref().map(|(_, data)| data).unwrap_or(&map_entry.persisted_monitor);
let deserialized_monitor = <(BlockHash, channelmonitor::ChannelMonitor<TestChannelSigner>)>::
read(&mut Cursor::new(&map_entry.1), (&*self.keys, &*self.keys)).unwrap().1;
read(&mut Cursor::new(&latest_monitor_data), (&*self.keys, &*self.keys)).unwrap().1;
deserialized_monitor.update_monitor(update, &&TestBroadcaster{}, &&FuzzEstimator { ret_val: atomic::AtomicU32::new(253) }, &self.logger).unwrap();
let mut ser = VecWriter(Vec::new());
deserialized_monitor.write(&mut ser).unwrap();
*map_entry = (update.update_id, ser.0);
self.chain_monitor.update_channel(funding_txo, update)
let res = self.chain_monitor.update_channel(funding_txo, update);
match res {
chain::ChannelMonitorUpdateStatus::Completed => {
map_entry.persisted_monitor_id = update.update_id;
map_entry.persisted_monitor = ser.0;
},
chain::ChannelMonitorUpdateStatus::InProgress => {
map_entry.pending_monitors.push((update.update_id, ser.0));
},
chain::ChannelMonitorUpdateStatus::UnrecoverableError => panic!(),
}
res
}

fn release_pending_monitor_events(&self) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, Option<PublicKey>)> {
Expand DownExpand Up@@ -511,9 +548,15 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {

let mut monitors = new_hash_map();
let mut old_monitors = $old_monitors.latest_monitors.lock().unwrap();
for (outpoint, (update_id, monitor_ser)) in old_monitors.drain() {
monitors.insert(outpoint, <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(&mut Cursor::new(&monitor_ser), (&*$keys_manager, &*$keys_manager)).expect("Failed to read monitor").1);
chain_monitor.latest_monitors.lock().unwrap().insert(outpoint, (update_id, monitor_ser));
for (outpoint, mut prev_state) in old_monitors.drain() {
monitors.insert(outpoint, <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(
&mut Cursor::new(&prev_state.persisted_monitor), (&*$keys_manager, &*$keys_manager)
).expect("Failed to read monitor").1);
// Wipe any `ChannelMonitor`s which we never told LDK we finished persisting,
// considering them discarded. LDK should replay these for us as they're stored in
// the `ChannelManager`.
prev_state.pending_monitors.clear();
chain_monitor.latest_monitors.lock().unwrap().insert(outpoint, prev_state);
}
let mut monitor_refs = new_hash_map();
for (outpoint, monitor) in monitors.iter_mut() {
Expand DownExpand Up@@ -1040,6 +1083,43 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
} }
}

let complete_first = |v: &mut Vec<_>| if !v.is_empty() { Some(v.remove(0)) } else { None };
let complete_second = |v: &mut Vec<_>| if v.len() > 1 { Some(v.remove(1)) } else { None };
let complete_monitor_update = |
monitor: &Arc<TestChainMonitor>, chan_funding,
compl_selector: &dyn Fn(&mut Vec<(u64, Vec<u8>)>) -> Option<(u64, Vec<u8>)>,
| {
if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
assert!(
state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
"updates should be sorted by id"
);
if let Some((id, data)) = compl_selector(&mut state.pending_monitors) {
monitor.chain_monitor.channel_monitor_updated(*chan_funding, id).unwrap();
if id > state.persisted_monitor_id {
state.persisted_monitor_id = id;
state.persisted_monitor = data;
}
}
}
};

let complete_all_monitor_updates = |monitor: &Arc<TestChainMonitor>, chan_funding| {
Comment thread
tnull marked this conversation as resolved.
if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
assert!(
state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
"updates should be sorted by id"
);
for (id, data) in state.pending_monitors.drain(..) {
monitor.chain_monitor.channel_monitor_updated(*chan_funding, id).unwrap();
if id > state.persisted_monitor_id {
state.persisted_monitor_id = id;
state.persisted_monitor = data;
}
}
}
};

let v = get_slice!(1)[0];
out.locked_write(format!("READ A BYTE! HANDLING INPUT {:x}...........\n", v).as_bytes());
match v {
Expand All@@ -1054,30 +1134,10 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
0x05 => *monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed,
0x06 => *monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed,

0x08 => {
if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[0].process_monitor_events();
}
},
0x09 => {
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[1].process_monitor_events();
}
},
0x0a => {
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[1].process_monitor_events();
}
},
0x0b => {
if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_c.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[2].process_monitor_events();
}
},
0x08 => complete_all_monitor_updates(&monitor_a, &chan_1_funding),
0x09 => complete_all_monitor_updates(&monitor_b, &chan_1_funding),
0x0a => complete_all_monitor_updates(&monitor_b, &chan_2_funding),
0x0b => complete_all_monitor_updates(&monitor_c, &chan_2_funding),

0x0c => {
if !chan_a_disconnected {
Expand DownExpand Up@@ -1285,119 +1345,35 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
},
0x89 => { fee_est_c.ret_val.store(253, atomic::Ordering::Release); nodes[2].maybe_update_chan_fees(); },

0xf0 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf1 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf2 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf0 => complete_monitor_update(&monitor_a, &chan_1_funding, &complete_first),
0xf1 => complete_monitor_update(&monitor_a, &chan_1_funding, &complete_second),
0xf2 => complete_monitor_update(&monitor_a, &chan_1_funding, &Vec::pop),

0xf4 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf5 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf6 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf4 => complete_monitor_update(&monitor_b, &chan_1_funding, &complete_first),
0xf5 => complete_monitor_update(&monitor_b, &chan_1_funding, &complete_second),
0xf6 => complete_monitor_update(&monitor_b, &chan_1_funding, &Vec::pop),

0xf8 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf9 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xfa => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf8 => complete_monitor_update(&monitor_b, &chan_2_funding, &complete_first),
0xf9 => complete_monitor_update(&monitor_b, &chan_2_funding, &complete_second),
0xfa => complete_monitor_update(&monitor_b, &chan_2_funding, &Vec::pop),

0xfc => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfd => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfe => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfc => complete_monitor_update(&monitor_c, &chan_2_funding, &complete_first),
0xfd => complete_monitor_update(&monitor_c, &chan_2_funding, &complete_second),
0xfe => complete_monitor_update(&monitor_c, &chan_2_funding, &Vec::pop),

0xff => {
// Test that no channel is in a stuck state where neither party can send funds even
// after we resolve all pending events.
// First make sure there are no pending monitor updates, resetting the error state
// and calling force_channel_monitor_updated for each monitor.
// First make sure there are no pending monitor updates and further update
// operations complete.
*monitor_a.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
*monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
*monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;

if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[0].process_monitor_events();
}
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[1].process_monitor_events();
}
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[1].process_monitor_events();
}
if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_c.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[2].process_monitor_events();
}
complete_all_monitor_updates(&monitor_a, &chan_1_funding);
complete_all_monitor_updates(&monitor_b, &chan_1_funding);
complete_all_monitor_updates(&monitor_b, &chan_2_funding);
complete_all_monitor_updates(&monitor_c, &chan_2_funding);

// Next, make sure peers are all connected to each other
if chan_a_disconnected {
Expand Down
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Strip utm_, fbclid, gclid, etc. from all links on page\n(function() {\n var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',\n 'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',\n 'ref', 'ref_src', 'source', 'medium', 'campaign'];\n \n function cleanUrl(url) {\n try {\n var u = new URL(url, window.location.origin);\n var changed = false;\n trackingParams.forEach(function(p) {\n if (u.searchParams.has(p)) {\n u.searchParams.delete(p);\n changed = true;\n }\n });\n return changed ? u.toString() : url;\n } catch (e) {\n return url;\n }\n }\n \n function cleanLinks() {\n document.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n \n cleanLinks();\n \n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1) {\n if (node.tagName === 'A') cleanLinks();\n node.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Remove Tracking Parameters from Links"); } } catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); } })(); (function(){ try { var __m = "youtube.com"; var __re = new RegExp('^' + "youtube\\.com" + '
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
256 changes: 116 additions & 140 deletions fuzz/src/chanmon_consistency.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -142,17 +142,29 @@ impl Writer for VecWriter {
}
}

/// The LDK API requires that any time we tell it we're done persisting a `ChannelMonitor[Update]`
/// we never pass it in as the "latest" `ChannelMonitor` on startup. However, we can pass
/// out-of-date monitors as long as we never told LDK we finished persisting them, which we do by
/// storing both old `ChannelMonitor`s and ones that are "being persisted" here.
///
/// Note that such "being persisted" `ChannelMonitor`s are stored in `ChannelManager` and will
/// simply be replayed on startup.
struct LatestMonitorState {
/// The latest monitor id which we told LDK we've persisted
persisted_monitor_id: u64,
/// The latest serialized `ChannelMonitor` that we told LDK we persisted.
persisted_monitor: Vec<u8>,
/// A set of (monitor id, serialized `ChannelMonitor`)s which we're currently "persisting",
/// from LDK's perspective.
pending_monitors: Vec<(u64, Vec<u8>)>,
}

struct TestChainMonitor {
pub logger: Arc<dyn Logger>,
pub keys: Arc<KeyProvider>,
pub persister: Arc<TestPersister>,
pub chain_monitor: Arc<chainmonitor::ChainMonitor<TestChannelSigner, Arc<dyn chain::Filter>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>, Arc<TestPersister>>>,
// If we reload a node with an old copy of ChannelMonitors, the ChannelManager deserialization
// logic will automatically force-close our channels for us (as we don't have an up-to-date
// monitor implying we are not able to punish misbehaving counterparties). Because this test
// "fails" if we ever force-close a channel, we avoid doing so, always saving the latest
// fully-serialized monitor state here, as well as the corresponding update_id.
pub latest_monitors: Mutex<HashMap<OutPoint, (u64, Vec<u8>)>>,
pub latest_monitors: Mutex<HashMap<OutPoint, LatestMonitorState>>,
}
impl TestChainMonitor {
pub fn new(broadcaster: Arc<TestBroadcaster>, logger: Arc<dyn Logger>, feeest: Arc<FuzzEstimator>, persister: Arc<TestPersister>, keys: Arc<KeyProvider>) -> Self {
Expand All@@ -169,22 +181,47 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
fn watch_channel(&self, funding_txo: OutPoint, monitor: channelmonitor::ChannelMonitor<TestChannelSigner>) -> Result<chain::ChannelMonitorUpdateStatus, ()> {
let mut ser = VecWriter(Vec::new());
monitor.write(&mut ser).unwrap();
if let Some(_) = self.latest_monitors.lock().unwrap().insert(funding_txo, (monitor.get_latest_update_id(), ser.0)) {
let monitor_id = monitor.get_latest_update_id();
let res = self.chain_monitor.watch_channel(funding_txo, monitor);
let state = match res {
Ok(chain::ChannelMonitorUpdateStatus::Completed) => {
LatestMonitorState {
persisted_monitor_id: monitor_id, persisted_monitor: ser.0,
pending_monitors: Vec::new(),
}
},
Ok(chain::ChannelMonitorUpdateStatus::InProgress) =>
panic!("The test currently doesn't test initial-persistence via the async pipeline"),
Ok(chain::ChannelMonitorUpdateStatus::UnrecoverableError) => panic!(),
Err(()) => panic!(),
};
if self.latest_monitors.lock().unwrap().insert(funding_txo, state).is_some() {
panic!("Already had monitor pre-watch_channel");
}
self.chain_monitor.watch_channel(funding_txo, monitor)
res
}

fn update_channel(&self, funding_txo: OutPoint, update: &channelmonitor::ChannelMonitorUpdate) -> chain::ChannelMonitorUpdateStatus {
let mut map_lock = self.latest_monitors.lock().unwrap();
let map_entry = map_lock.get_mut(&funding_txo).expect("Didn't have monitor on update call");
let latest_monitor_data = map_entry.pending_monitors.last().as_ref().map(|(_, data)| data).unwrap_or(&map_entry.persisted_monitor);
let deserialized_monitor = <(BlockHash, channelmonitor::ChannelMonitor<TestChannelSigner>)>::
read(&mut Cursor::new(&map_entry.1), (&*self.keys, &*self.keys)).unwrap().1;
read(&mut Cursor::new(&latest_monitor_data), (&*self.keys, &*self.keys)).unwrap().1;
deserialized_monitor.update_monitor(update, &&TestBroadcaster{}, &&FuzzEstimator { ret_val: atomic::AtomicU32::new(253) }, &self.logger).unwrap();
let mut ser = VecWriter(Vec::new());
deserialized_monitor.write(&mut ser).unwrap();
*map_entry = (update.update_id, ser.0);
self.chain_monitor.update_channel(funding_txo, update)
let res = self.chain_monitor.update_channel(funding_txo, update);
match res {
chain::ChannelMonitorUpdateStatus::Completed => {
map_entry.persisted_monitor_id = update.update_id;
map_entry.persisted_monitor = ser.0;
},
chain::ChannelMonitorUpdateStatus::InProgress => {
map_entry.pending_monitors.push((update.update_id, ser.0));
},
chain::ChannelMonitorUpdateStatus::UnrecoverableError => panic!(),
}
res
}

fn release_pending_monitor_events(&self) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, Option<PublicKey>)> {
Expand DownExpand Up@@ -511,9 +548,15 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {

let mut monitors = new_hash_map();
let mut old_monitors = $old_monitors.latest_monitors.lock().unwrap();
for (outpoint, (update_id, monitor_ser)) in old_monitors.drain() {
monitors.insert(outpoint, <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(&mut Cursor::new(&monitor_ser), (&*$keys_manager, &*$keys_manager)).expect("Failed to read monitor").1);
chain_monitor.latest_monitors.lock().unwrap().insert(outpoint, (update_id, monitor_ser));
for (outpoint, mut prev_state) in old_monitors.drain() {
monitors.insert(outpoint, <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(
&mut Cursor::new(&prev_state.persisted_monitor), (&*$keys_manager, &*$keys_manager)
).expect("Failed to read monitor").1);
// Wipe any `ChannelMonitor`s which we never told LDK we finished persisting,
// considering them discarded. LDK should replay these for us as they're stored in
// the `ChannelManager`.
prev_state.pending_monitors.clear();
chain_monitor.latest_monitors.lock().unwrap().insert(outpoint, prev_state);
}
let mut monitor_refs = new_hash_map();
for (outpoint, monitor) in monitors.iter_mut() {
Expand DownExpand Up@@ -1040,6 +1083,43 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
} }
}

let complete_first = |v: &mut Vec<_>| if !v.is_empty() { Some(v.remove(0)) } else { None };
let complete_second = |v: &mut Vec<_>| if v.len() > 1 { Some(v.remove(1)) } else { None };
let complete_monitor_update = |
monitor: &Arc<TestChainMonitor>, chan_funding,
compl_selector: &dyn Fn(&mut Vec<(u64, Vec<u8>)>) -> Option<(u64, Vec<u8>)>,
| {
if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
assert!(
state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
"updates should be sorted by id"
);
if let Some((id, data)) = compl_selector(&mut state.pending_monitors) {
monitor.chain_monitor.channel_monitor_updated(*chan_funding, id).unwrap();
if id > state.persisted_monitor_id {
state.persisted_monitor_id = id;
state.persisted_monitor = data;
}
}
}
};

let complete_all_monitor_updates = |monitor: &Arc<TestChainMonitor>, chan_funding| {
Comment thread
tnull marked this conversation as resolved.
if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
assert!(
state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
"updates should be sorted by id"
);
for (id, data) in state.pending_monitors.drain(..) {
monitor.chain_monitor.channel_monitor_updated(*chan_funding, id).unwrap();
if id > state.persisted_monitor_id {
state.persisted_monitor_id = id;
state.persisted_monitor = data;
}
}
}
};

let v = get_slice!(1)[0];
out.locked_write(format!("READ A BYTE! HANDLING INPUT {:x}...........\n", v).as_bytes());
match v {
Expand All@@ -1054,30 +1134,10 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
0x05 => *monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed,
0x06 => *monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed,

0x08 => {
if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[0].process_monitor_events();
}
},
0x09 => {
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[1].process_monitor_events();
}
},
0x0a => {
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[1].process_monitor_events();
}
},
0x0b => {
if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_c.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[2].process_monitor_events();
}
},
0x08 => complete_all_monitor_updates(&monitor_a, &chan_1_funding),
0x09 => complete_all_monitor_updates(&monitor_b, &chan_1_funding),
0x0a => complete_all_monitor_updates(&monitor_b, &chan_2_funding),
0x0b => complete_all_monitor_updates(&monitor_c, &chan_2_funding),

0x0c => {
if !chan_a_disconnected {
Expand DownExpand Up@@ -1285,119 +1345,35 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
},
0x89 => { fee_est_c.ret_val.store(253, atomic::Ordering::Release); nodes[2].maybe_update_chan_fees(); },

0xf0 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf1 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf2 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf0 => complete_monitor_update(&monitor_a, &chan_1_funding, &complete_first),
0xf1 => complete_monitor_update(&monitor_a, &chan_1_funding, &complete_second),
0xf2 => complete_monitor_update(&monitor_a, &chan_1_funding, &Vec::pop),

0xf4 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf5 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf6 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf4 => complete_monitor_update(&monitor_b, &chan_1_funding, &complete_first),
0xf5 => complete_monitor_update(&monitor_b, &chan_1_funding, &complete_second),
0xf6 => complete_monitor_update(&monitor_b, &chan_1_funding, &Vec::pop),

0xf8 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf9 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xfa => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf8 => complete_monitor_update(&monitor_b, &chan_2_funding, &complete_first),
0xf9 => complete_monitor_update(&monitor_b, &chan_2_funding, &complete_second),
0xfa => complete_monitor_update(&monitor_b, &chan_2_funding, &Vec::pop),

0xfc => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfd => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfe => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfc => complete_monitor_update(&monitor_c, &chan_2_funding, &complete_first),
0xfd => complete_monitor_update(&monitor_c, &chan_2_funding, &complete_second),
0xfe => complete_monitor_update(&monitor_c, &chan_2_funding, &Vec::pop),

0xff => {
// Test that no channel is in a stuck state where neither party can send funds even
// after we resolve all pending events.
// First make sure there are no pending monitor updates, resetting the error state
// and calling force_channel_monitor_updated for each monitor.
// First make sure there are no pending monitor updates and further update
// operations complete.
*monitor_a.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
*monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
*monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;

if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[0].process_monitor_events();
}
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[1].process_monitor_events();
}
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[1].process_monitor_events();
}
if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_c.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[2].process_monitor_events();
}
complete_all_monitor_updates(&monitor_a, &chan_1_funding);
complete_all_monitor_updates(&monitor_b, &chan_1_funding);
complete_all_monitor_updates(&monitor_b, &chan_2_funding);
complete_all_monitor_updates(&monitor_c, &chan_2_funding);

// Next, make sure peers are all connected to each other
if chan_a_disconnected {
Expand Down
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Auto-enable theater mode on YouTube\n(function() {\n function tryTheater() {\n var btn = document.querySelector('button[aria-label=\"Theater mode\"], ytd-player #player button[title=\"Theater mode\"]');\n if (btn && !btn.classList.contains('activated')) {\n btn.click();\n }\n }\n \n // Try immediately\n tryTheater();\n \n // Try after navigation (SPA)\n var lastUrl = location.href;\n setInterval(function() {\n if (location.href !== lastUrl) {\n lastUrl = location.href;\n setTimeout(tryTheater, 500);\n }\n }, 1000);\n \n // Also try on player load\n var observer = new MutationObserver(tryTheater);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "YouTube Theater Mode Default"); } } catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
256 changes: 116 additions & 140 deletions fuzz/src/chanmon_consistency.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -142,17 +142,29 @@ impl Writer for VecWriter {
}
}

/// The LDK API requires that any time we tell it we're done persisting a `ChannelMonitor[Update]`
/// we never pass it in as the "latest" `ChannelMonitor` on startup. However, we can pass
/// out-of-date monitors as long as we never told LDK we finished persisting them, which we do by
/// storing both old `ChannelMonitor`s and ones that are "being persisted" here.
///
/// Note that such "being persisted" `ChannelMonitor`s are stored in `ChannelManager` and will
/// simply be replayed on startup.
struct LatestMonitorState {
/// The latest monitor id which we told LDK we've persisted
persisted_monitor_id: u64,
/// The latest serialized `ChannelMonitor` that we told LDK we persisted.
persisted_monitor: Vec<u8>,
/// A set of (monitor id, serialized `ChannelMonitor`)s which we're currently "persisting",
/// from LDK's perspective.
pending_monitors: Vec<(u64, Vec<u8>)>,
}

struct TestChainMonitor {
pub logger: Arc<dyn Logger>,
pub keys: Arc<KeyProvider>,
pub persister: Arc<TestPersister>,
pub chain_monitor: Arc<chainmonitor::ChainMonitor<TestChannelSigner, Arc<dyn chain::Filter>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>, Arc<TestPersister>>>,
// If we reload a node with an old copy of ChannelMonitors, the ChannelManager deserialization
// logic will automatically force-close our channels for us (as we don't have an up-to-date
// monitor implying we are not able to punish misbehaving counterparties). Because this test
// "fails" if we ever force-close a channel, we avoid doing so, always saving the latest
// fully-serialized monitor state here, as well as the corresponding update_id.
pub latest_monitors: Mutex<HashMap<OutPoint, (u64, Vec<u8>)>>,
pub latest_monitors: Mutex<HashMap<OutPoint, LatestMonitorState>>,
}
impl TestChainMonitor {
pub fn new(broadcaster: Arc<TestBroadcaster>, logger: Arc<dyn Logger>, feeest: Arc<FuzzEstimator>, persister: Arc<TestPersister>, keys: Arc<KeyProvider>) -> Self {
Expand All@@ -169,22 +181,47 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
fn watch_channel(&self, funding_txo: OutPoint, monitor: channelmonitor::ChannelMonitor<TestChannelSigner>) -> Result<chain::ChannelMonitorUpdateStatus, ()> {
let mut ser = VecWriter(Vec::new());
monitor.write(&mut ser).unwrap();
if let Some(_) = self.latest_monitors.lock().unwrap().insert(funding_txo, (monitor.get_latest_update_id(), ser.0)) {
let monitor_id = monitor.get_latest_update_id();
let res = self.chain_monitor.watch_channel(funding_txo, monitor);
let state = match res {
Ok(chain::ChannelMonitorUpdateStatus::Completed) => {
LatestMonitorState {
persisted_monitor_id: monitor_id, persisted_monitor: ser.0,
pending_monitors: Vec::new(),
}
},
Ok(chain::ChannelMonitorUpdateStatus::InProgress) =>
panic!("The test currently doesn't test initial-persistence via the async pipeline"),
Ok(chain::ChannelMonitorUpdateStatus::UnrecoverableError) => panic!(),
Err(()) => panic!(),
};
if self.latest_monitors.lock().unwrap().insert(funding_txo, state).is_some() {
panic!("Already had monitor pre-watch_channel");
}
self.chain_monitor.watch_channel(funding_txo, monitor)
res
}

fn update_channel(&self, funding_txo: OutPoint, update: &channelmonitor::ChannelMonitorUpdate) -> chain::ChannelMonitorUpdateStatus {
let mut map_lock = self.latest_monitors.lock().unwrap();
let map_entry = map_lock.get_mut(&funding_txo).expect("Didn't have monitor on update call");
let latest_monitor_data = map_entry.pending_monitors.last().as_ref().map(|(_, data)| data).unwrap_or(&map_entry.persisted_monitor);
let deserialized_monitor = <(BlockHash, channelmonitor::ChannelMonitor<TestChannelSigner>)>::
read(&mut Cursor::new(&map_entry.1), (&*self.keys, &*self.keys)).unwrap().1;
read(&mut Cursor::new(&latest_monitor_data), (&*self.keys, &*self.keys)).unwrap().1;
deserialized_monitor.update_monitor(update, &&TestBroadcaster{}, &&FuzzEstimator { ret_val: atomic::AtomicU32::new(253) }, &self.logger).unwrap();
let mut ser = VecWriter(Vec::new());
deserialized_monitor.write(&mut ser).unwrap();
*map_entry = (update.update_id, ser.0);
self.chain_monitor.update_channel(funding_txo, update)
let res = self.chain_monitor.update_channel(funding_txo, update);
match res {
chain::ChannelMonitorUpdateStatus::Completed => {
map_entry.persisted_monitor_id = update.update_id;
map_entry.persisted_monitor = ser.0;
},
chain::ChannelMonitorUpdateStatus::InProgress => {
map_entry.pending_monitors.push((update.update_id, ser.0));
},
chain::ChannelMonitorUpdateStatus::UnrecoverableError => panic!(),
}
res
}

fn release_pending_monitor_events(&self) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, Option<PublicKey>)> {
Expand DownExpand Up@@ -511,9 +548,15 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {

let mut monitors = new_hash_map();
let mut old_monitors = $old_monitors.latest_monitors.lock().unwrap();
for (outpoint, (update_id, monitor_ser)) in old_monitors.drain() {
monitors.insert(outpoint, <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(&mut Cursor::new(&monitor_ser), (&*$keys_manager, &*$keys_manager)).expect("Failed to read monitor").1);
chain_monitor.latest_monitors.lock().unwrap().insert(outpoint, (update_id, monitor_ser));
for (outpoint, mut prev_state) in old_monitors.drain() {
monitors.insert(outpoint, <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(
&mut Cursor::new(&prev_state.persisted_monitor), (&*$keys_manager, &*$keys_manager)
).expect("Failed to read monitor").1);
// Wipe any `ChannelMonitor`s which we never told LDK we finished persisting,
// considering them discarded. LDK should replay these for us as they're stored in
// the `ChannelManager`.
prev_state.pending_monitors.clear();
chain_monitor.latest_monitors.lock().unwrap().insert(outpoint, prev_state);
}
let mut monitor_refs = new_hash_map();
for (outpoint, monitor) in monitors.iter_mut() {
Expand DownExpand Up@@ -1040,6 +1083,43 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
} }
}

let complete_first = |v: &mut Vec<_>| if !v.is_empty() { Some(v.remove(0)) } else { None };
let complete_second = |v: &mut Vec<_>| if v.len() > 1 { Some(v.remove(1)) } else { None };
let complete_monitor_update = |
monitor: &Arc<TestChainMonitor>, chan_funding,
compl_selector: &dyn Fn(&mut Vec<(u64, Vec<u8>)>) -> Option<(u64, Vec<u8>)>,
| {
if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
assert!(
state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
"updates should be sorted by id"
);
if let Some((id, data)) = compl_selector(&mut state.pending_monitors) {
monitor.chain_monitor.channel_monitor_updated(*chan_funding, id).unwrap();
if id > state.persisted_monitor_id {
state.persisted_monitor_id = id;
state.persisted_monitor = data;
}
}
}
};

let complete_all_monitor_updates = |monitor: &Arc<TestChainMonitor>, chan_funding| {
Comment thread
tnull marked this conversation as resolved.
if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
assert!(
state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
"updates should be sorted by id"
);
for (id, data) in state.pending_monitors.drain(..) {
monitor.chain_monitor.channel_monitor_updated(*chan_funding, id).unwrap();
if id > state.persisted_monitor_id {
state.persisted_monitor_id = id;
state.persisted_monitor = data;
}
}
}
};

let v = get_slice!(1)[0];
out.locked_write(format!("READ A BYTE! HANDLING INPUT {:x}...........\n", v).as_bytes());
match v {
Expand All@@ -1054,30 +1134,10 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
0x05 => *monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed,
0x06 => *monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed,

0x08 => {
if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[0].process_monitor_events();
}
},
0x09 => {
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[1].process_monitor_events();
}
},
0x0a => {
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[1].process_monitor_events();
}
},
0x0b => {
if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_c.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[2].process_monitor_events();
}
},
0x08 => complete_all_monitor_updates(&monitor_a, &chan_1_funding),
0x09 => complete_all_monitor_updates(&monitor_b, &chan_1_funding),
0x0a => complete_all_monitor_updates(&monitor_b, &chan_2_funding),
0x0b => complete_all_monitor_updates(&monitor_c, &chan_2_funding),

0x0c => {
if !chan_a_disconnected {
Expand DownExpand Up@@ -1285,119 +1345,35 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
},
0x89 => { fee_est_c.ret_val.store(253, atomic::Ordering::Release); nodes[2].maybe_update_chan_fees(); },

0xf0 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf1 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf2 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf0 => complete_monitor_update(&monitor_a, &chan_1_funding, &complete_first),
0xf1 => complete_monitor_update(&monitor_a, &chan_1_funding, &complete_second),
0xf2 => complete_monitor_update(&monitor_a, &chan_1_funding, &Vec::pop),

0xf4 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf5 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf6 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf4 => complete_monitor_update(&monitor_b, &chan_1_funding, &complete_first),
0xf5 => complete_monitor_update(&monitor_b, &chan_1_funding, &complete_second),
0xf6 => complete_monitor_update(&monitor_b, &chan_1_funding, &Vec::pop),

0xf8 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf9 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xfa => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf8 => complete_monitor_update(&monitor_b, &chan_2_funding, &complete_first),
0xf9 => complete_monitor_update(&monitor_b, &chan_2_funding, &complete_second),
0xfa => complete_monitor_update(&monitor_b, &chan_2_funding, &Vec::pop),

0xfc => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfd => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfe => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfc => complete_monitor_update(&monitor_c, &chan_2_funding, &complete_first),
0xfd => complete_monitor_update(&monitor_c, &chan_2_funding, &complete_second),
0xfe => complete_monitor_update(&monitor_c, &chan_2_funding, &Vec::pop),

0xff => {
// Test that no channel is in a stuck state where neither party can send funds even
// after we resolve all pending events.
// First make sure there are no pending monitor updates, resetting the error state
// and calling force_channel_monitor_updated for each monitor.
// First make sure there are no pending monitor updates and further update
// operations complete.
*monitor_a.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
*monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
*monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;

if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[0].process_monitor_events();
}
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[1].process_monitor_events();
}
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[1].process_monitor_events();
}
if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_c.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[2].process_monitor_events();
}
complete_all_monitor_updates(&monitor_a, &chan_1_funding);
complete_all_monitor_updates(&monitor_b, &chan_1_funding);
complete_all_monitor_updates(&monitor_b, &chan_2_funding);
complete_all_monitor_updates(&monitor_c, &chan_2_funding);

// Next, make sure peers are all connected to each other
if chan_a_disconnected {
Expand Down
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Remove or un-stick sticky/fixed headers that block content\n(function() {\n function unstick() {\n document.querySelectorAll('header, nav, [role=\"banner\"], .header, .navbar, .sticky, .fixed-top, [style*=\"position: fixed\"], [style*=\"position:sticky\"]').forEach(function(el) {\n if (el.style.position === 'fixed' || el.style.position === 'sticky' || \n getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {\n el.style.position = 'static';\n el.style.top = 'auto';\n el.style.zIndex = 'auto';\n }\n });\n }\n \n unstick();\n \n var observer = new MutationObserver(unstick);\n observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });\n})();", "Kill Sticky Headers"); } } catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
256 changes: 116 additions & 140 deletions fuzz/src/chanmon_consistency.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -142,17 +142,29 @@ impl Writer for VecWriter {
}
}

/// The LDK API requires that any time we tell it we're done persisting a `ChannelMonitor[Update]`
/// we never pass it in as the "latest" `ChannelMonitor` on startup. However, we can pass
/// out-of-date monitors as long as we never told LDK we finished persisting them, which we do by
/// storing both old `ChannelMonitor`s and ones that are "being persisted" here.
///
/// Note that such "being persisted" `ChannelMonitor`s are stored in `ChannelManager` and will
/// simply be replayed on startup.
struct LatestMonitorState {
/// The latest monitor id which we told LDK we've persisted
persisted_monitor_id: u64,
/// The latest serialized `ChannelMonitor` that we told LDK we persisted.
persisted_monitor: Vec<u8>,
/// A set of (monitor id, serialized `ChannelMonitor`)s which we're currently "persisting",
/// from LDK's perspective.
pending_monitors: Vec<(u64, Vec<u8>)>,
}

struct TestChainMonitor {
pub logger: Arc<dyn Logger>,
pub keys: Arc<KeyProvider>,
pub persister: Arc<TestPersister>,
pub chain_monitor: Arc<chainmonitor::ChainMonitor<TestChannelSigner, Arc<dyn chain::Filter>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>, Arc<TestPersister>>>,
// If we reload a node with an old copy of ChannelMonitors, the ChannelManager deserialization
// logic will automatically force-close our channels for us (as we don't have an up-to-date
// monitor implying we are not able to punish misbehaving counterparties). Because this test
// "fails" if we ever force-close a channel, we avoid doing so, always saving the latest
// fully-serialized monitor state here, as well as the corresponding update_id.
pub latest_monitors: Mutex<HashMap<OutPoint, (u64, Vec<u8>)>>,
pub latest_monitors: Mutex<HashMap<OutPoint, LatestMonitorState>>,
}
impl TestChainMonitor {
pub fn new(broadcaster: Arc<TestBroadcaster>, logger: Arc<dyn Logger>, feeest: Arc<FuzzEstimator>, persister: Arc<TestPersister>, keys: Arc<KeyProvider>) -> Self {
Expand All@@ -169,22 +181,47 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
fn watch_channel(&self, funding_txo: OutPoint, monitor: channelmonitor::ChannelMonitor<TestChannelSigner>) -> Result<chain::ChannelMonitorUpdateStatus, ()> {
let mut ser = VecWriter(Vec::new());
monitor.write(&mut ser).unwrap();
if let Some(_) = self.latest_monitors.lock().unwrap().insert(funding_txo, (monitor.get_latest_update_id(), ser.0)) {
let monitor_id = monitor.get_latest_update_id();
let res = self.chain_monitor.watch_channel(funding_txo, monitor);
let state = match res {
Ok(chain::ChannelMonitorUpdateStatus::Completed) => {
LatestMonitorState {
persisted_monitor_id: monitor_id, persisted_monitor: ser.0,
pending_monitors: Vec::new(),
}
},
Ok(chain::ChannelMonitorUpdateStatus::InProgress) =>
panic!("The test currently doesn't test initial-persistence via the async pipeline"),
Ok(chain::ChannelMonitorUpdateStatus::UnrecoverableError) => panic!(),
Err(()) => panic!(),
};
if self.latest_monitors.lock().unwrap().insert(funding_txo, state).is_some() {
panic!("Already had monitor pre-watch_channel");
}
self.chain_monitor.watch_channel(funding_txo, monitor)
res
}

fn update_channel(&self, funding_txo: OutPoint, update: &channelmonitor::ChannelMonitorUpdate) -> chain::ChannelMonitorUpdateStatus {
let mut map_lock = self.latest_monitors.lock().unwrap();
let map_entry = map_lock.get_mut(&funding_txo).expect("Didn't have monitor on update call");
let latest_monitor_data = map_entry.pending_monitors.last().as_ref().map(|(_, data)| data).unwrap_or(&map_entry.persisted_monitor);
let deserialized_monitor = <(BlockHash, channelmonitor::ChannelMonitor<TestChannelSigner>)>::
read(&mut Cursor::new(&map_entry.1), (&*self.keys, &*self.keys)).unwrap().1;
read(&mut Cursor::new(&latest_monitor_data), (&*self.keys, &*self.keys)).unwrap().1;
deserialized_monitor.update_monitor(update, &&TestBroadcaster{}, &&FuzzEstimator { ret_val: atomic::AtomicU32::new(253) }, &self.logger).unwrap();
let mut ser = VecWriter(Vec::new());
deserialized_monitor.write(&mut ser).unwrap();
*map_entry = (update.update_id, ser.0);
self.chain_monitor.update_channel(funding_txo, update)
let res = self.chain_monitor.update_channel(funding_txo, update);
match res {
chain::ChannelMonitorUpdateStatus::Completed => {
map_entry.persisted_monitor_id = update.update_id;
map_entry.persisted_monitor = ser.0;
},
chain::ChannelMonitorUpdateStatus::InProgress => {
map_entry.pending_monitors.push((update.update_id, ser.0));
},
chain::ChannelMonitorUpdateStatus::UnrecoverableError => panic!(),
}
res
}

fn release_pending_monitor_events(&self) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, Option<PublicKey>)> {
Expand DownExpand Up@@ -511,9 +548,15 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {

let mut monitors = new_hash_map();
let mut old_monitors = $old_monitors.latest_monitors.lock().unwrap();
for (outpoint, (update_id, monitor_ser)) in old_monitors.drain() {
monitors.insert(outpoint, <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(&mut Cursor::new(&monitor_ser), (&*$keys_manager, &*$keys_manager)).expect("Failed to read monitor").1);
chain_monitor.latest_monitors.lock().unwrap().insert(outpoint, (update_id, monitor_ser));
for (outpoint, mut prev_state) in old_monitors.drain() {
monitors.insert(outpoint, <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(
&mut Cursor::new(&prev_state.persisted_monitor), (&*$keys_manager, &*$keys_manager)
).expect("Failed to read monitor").1);
// Wipe any `ChannelMonitor`s which we never told LDK we finished persisting,
// considering them discarded. LDK should replay these for us as they're stored in
// the `ChannelManager`.
prev_state.pending_monitors.clear();
chain_monitor.latest_monitors.lock().unwrap().insert(outpoint, prev_state);
}
let mut monitor_refs = new_hash_map();
for (outpoint, monitor) in monitors.iter_mut() {
Expand DownExpand Up@@ -1040,6 +1083,43 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
} }
}

let complete_first = |v: &mut Vec<_>| if !v.is_empty() { Some(v.remove(0)) } else { None };
let complete_second = |v: &mut Vec<_>| if v.len() > 1 { Some(v.remove(1)) } else { None };
let complete_monitor_update = |
monitor: &Arc<TestChainMonitor>, chan_funding,
compl_selector: &dyn Fn(&mut Vec<(u64, Vec<u8>)>) -> Option<(u64, Vec<u8>)>,
| {
if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
assert!(
state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
"updates should be sorted by id"
);
if let Some((id, data)) = compl_selector(&mut state.pending_monitors) {
monitor.chain_monitor.channel_monitor_updated(*chan_funding, id).unwrap();
if id > state.persisted_monitor_id {
state.persisted_monitor_id = id;
state.persisted_monitor = data;
}
}
}
};

let complete_all_monitor_updates = |monitor: &Arc<TestChainMonitor>, chan_funding| {
Comment thread
tnull marked this conversation as resolved.
if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
assert!(
state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
"updates should be sorted by id"
);
for (id, data) in state.pending_monitors.drain(..) {
monitor.chain_monitor.channel_monitor_updated(*chan_funding, id).unwrap();
if id > state.persisted_monitor_id {
state.persisted_monitor_id = id;
state.persisted_monitor = data;
}
}
}
};

let v = get_slice!(1)[0];
out.locked_write(format!("READ A BYTE! HANDLING INPUT {:x}...........\n", v).as_bytes());
match v {
Expand All@@ -1054,30 +1134,10 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
0x05 => *monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed,
0x06 => *monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed,

0x08 => {
if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[0].process_monitor_events();
}
},
0x09 => {
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[1].process_monitor_events();
}
},
0x0a => {
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[1].process_monitor_events();
}
},
0x0b => {
if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_c.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[2].process_monitor_events();
}
},
0x08 => complete_all_monitor_updates(&monitor_a, &chan_1_funding),
0x09 => complete_all_monitor_updates(&monitor_b, &chan_1_funding),
0x0a => complete_all_monitor_updates(&monitor_b, &chan_2_funding),
0x0b => complete_all_monitor_updates(&monitor_c, &chan_2_funding),

0x0c => {
if !chan_a_disconnected {
Expand DownExpand Up@@ -1285,119 +1345,35 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
},
0x89 => { fee_est_c.ret_val.store(253, atomic::Ordering::Release); nodes[2].maybe_update_chan_fees(); },

0xf0 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf1 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf2 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf0 => complete_monitor_update(&monitor_a, &chan_1_funding, &complete_first),
0xf1 => complete_monitor_update(&monitor_a, &chan_1_funding, &complete_second),
0xf2 => complete_monitor_update(&monitor_a, &chan_1_funding, &Vec::pop),

0xf4 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf5 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf6 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf4 => complete_monitor_update(&monitor_b, &chan_1_funding, &complete_first),
0xf5 => complete_monitor_update(&monitor_b, &chan_1_funding, &complete_second),
0xf6 => complete_monitor_update(&monitor_b, &chan_1_funding, &Vec::pop),

0xf8 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf9 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xfa => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf8 => complete_monitor_update(&monitor_b, &chan_2_funding, &complete_first),
0xf9 => complete_monitor_update(&monitor_b, &chan_2_funding, &complete_second),
0xfa => complete_monitor_update(&monitor_b, &chan_2_funding, &Vec::pop),

0xfc => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfd => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfe => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfc => complete_monitor_update(&monitor_c, &chan_2_funding, &complete_first),
0xfd => complete_monitor_update(&monitor_c, &chan_2_funding, &complete_second),
0xfe => complete_monitor_update(&monitor_c, &chan_2_funding, &Vec::pop),

0xff => {
// Test that no channel is in a stuck state where neither party can send funds even
// after we resolve all pending events.
// First make sure there are no pending monitor updates, resetting the error state
// and calling force_channel_monitor_updated for each monitor.
// First make sure there are no pending monitor updates and further update
// operations complete.
*monitor_a.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
*monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
*monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;

if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[0].process_monitor_events();
}
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[1].process_monitor_events();
}
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[1].process_monitor_events();
}
if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_c.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[2].process_monitor_events();
}
complete_all_monitor_updates(&monitor_a, &chan_1_funding);
complete_all_monitor_updates(&monitor_b, &chan_1_funding);
complete_all_monitor_updates(&monitor_b, &chan_2_funding);
complete_all_monitor_updates(&monitor_c, &chan_2_funding);

// Next, make sure peers are all connected to each other
if chan_a_disconnected {
Expand Down
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Universal Dark Mode - works on any site\n(function() {\n var enabled = true;\n \n function applyDarkMode() {\n if (!enabled) return;\n \n // Create style element if it doesn't exist\n var style = document.getElementById('universal-dark-mode-style');\n if (!style) {\n style = document.createElement('style');\n style.id = 'universal-dark-mode-style';\n document.head.appendChild(style);\n }\n \n // Dark mode CSS - inverts colors but preserves images/video\n style.textContent = '\n /* Invert everything except media */\n html {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #1a1a2e !important;\n }\n \n /* Restore images, videos, iframes, canvas */\n img, video, iframe, canvas, svg, picture, [style*=\"background-image\"] {\n filter: invert(1) hue-rotate(180deg) !important;\n }\n \n /* Preserve specific elements that should not be inverted */\n .no-dark-mode, .no-dark-mode *,\n [data-theme=\"light\"], [data-theme=\"light\"],\n .ace_editor, .ace_editor *,\n .CodeMirror, .CodeMirror *,\n .monaco-editor, .monaco-editor *,\n .markdown-body pre, .markdown-body pre *,\n .highlight, .highlight *,\n pre code, pre code * {\n filter: none !important;\n }\n \n /* Fix common UI elements */\n .modal, .popup, .dropdown-menu, .tooltip, .popover {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #2d2d44 !important;\n border-color: #444 !important;\n }\n \n /* Scrollbars */\n ::-webkit-scrollbar { background: #1a1a2e !important; }\n ::-webkit-scrollbar-thumb { background: #444 !important; }\n ::-webkit-scrollbar-thumb:hover { background: #555 !important; }\n \n /* Selection */\n ::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ';\n }\n \n function removeDarkMode() {\n var style = document.getElementById('universal-dark-mode-style');\n if (style) style.remove();\n }\n \n // Toggle with Alt+Shift+D\n document.addEventListener('keydown', function(e) {\n if (e.altKey && e.shiftKey && e.key === 'D') {\n e.preventDefault();\n enabled = !enabled;\n if (enabled) {\n applyDarkMode();\n console.log('[Universal Dark Mode] Enabled');\n } else {\n removeDarkMode();\n console.log('[Universal Dark Mode] Disabled');\n }\n }\n });\n \n // Apply on load\n applyDarkMode();\n \n // Re-apply on dynamic content\n var observer = new MutationObserver(function(mutations) {\n if (enabled && !document.getElementById('universal-dark-mode-style')) {\n applyDarkMode();\n }\n });\n observer.observe(document.head, { childList: true });\n \n console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');\n})();", "Universal Dark Mode"); } } catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); } })(); })();
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
256 changes: 116 additions & 140 deletions fuzz/src/chanmon_consistency.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -142,17 +142,29 @@ impl Writer for VecWriter {
}
}

/// The LDK API requires that any time we tell it we're done persisting a `ChannelMonitor[Update]`
/// we never pass it in as the "latest" `ChannelMonitor` on startup. However, we can pass
/// out-of-date monitors as long as we never told LDK we finished persisting them, which we do by
/// storing both old `ChannelMonitor`s and ones that are "being persisted" here.
///
/// Note that such "being persisted" `ChannelMonitor`s are stored in `ChannelManager` and will
/// simply be replayed on startup.
struct LatestMonitorState {
/// The latest monitor id which we told LDK we've persisted
persisted_monitor_id: u64,
/// The latest serialized `ChannelMonitor` that we told LDK we persisted.
persisted_monitor: Vec<u8>,
/// A set of (monitor id, serialized `ChannelMonitor`)s which we're currently "persisting",
/// from LDK's perspective.
pending_monitors: Vec<(u64, Vec<u8>)>,
}

struct TestChainMonitor {
pub logger: Arc<dyn Logger>,
pub keys: Arc<KeyProvider>,
pub persister: Arc<TestPersister>,
pub chain_monitor: Arc<chainmonitor::ChainMonitor<TestChannelSigner, Arc<dyn chain::Filter>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>, Arc<TestPersister>>>,
// If we reload a node with an old copy of ChannelMonitors, the ChannelManager deserialization
// logic will automatically force-close our channels for us (as we don't have an up-to-date
// monitor implying we are not able to punish misbehaving counterparties). Because this test
// "fails" if we ever force-close a channel, we avoid doing so, always saving the latest
// fully-serialized monitor state here, as well as the corresponding update_id.
pub latest_monitors: Mutex<HashMap<OutPoint, (u64, Vec<u8>)>>,
pub latest_monitors: Mutex<HashMap<OutPoint, LatestMonitorState>>,
}
impl TestChainMonitor {
pub fn new(broadcaster: Arc<TestBroadcaster>, logger: Arc<dyn Logger>, feeest: Arc<FuzzEstimator>, persister: Arc<TestPersister>, keys: Arc<KeyProvider>) -> Self {
Expand All@@ -169,22 +181,47 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
fn watch_channel(&self, funding_txo: OutPoint, monitor: channelmonitor::ChannelMonitor<TestChannelSigner>) -> Result<chain::ChannelMonitorUpdateStatus, ()> {
let mut ser = VecWriter(Vec::new());
monitor.write(&mut ser).unwrap();
if let Some(_) = self.latest_monitors.lock().unwrap().insert(funding_txo, (monitor.get_latest_update_id(), ser.0)) {
let monitor_id = monitor.get_latest_update_id();
let res = self.chain_monitor.watch_channel(funding_txo, monitor);
let state = match res {
Ok(chain::ChannelMonitorUpdateStatus::Completed) => {
LatestMonitorState {
persisted_monitor_id: monitor_id, persisted_monitor: ser.0,
pending_monitors: Vec::new(),
}
},
Ok(chain::ChannelMonitorUpdateStatus::InProgress) =>
panic!("The test currently doesn't test initial-persistence via the async pipeline"),
Ok(chain::ChannelMonitorUpdateStatus::UnrecoverableError) => panic!(),
Err(()) => panic!(),
};
if self.latest_monitors.lock().unwrap().insert(funding_txo, state).is_some() {
panic!("Already had monitor pre-watch_channel");
}
self.chain_monitor.watch_channel(funding_txo, monitor)
res
}

fn update_channel(&self, funding_txo: OutPoint, update: &channelmonitor::ChannelMonitorUpdate) -> chain::ChannelMonitorUpdateStatus {
let mut map_lock = self.latest_monitors.lock().unwrap();
let map_entry = map_lock.get_mut(&funding_txo).expect("Didn't have monitor on update call");
let latest_monitor_data = map_entry.pending_monitors.last().as_ref().map(|(_, data)| data).unwrap_or(&map_entry.persisted_monitor);
let deserialized_monitor = <(BlockHash, channelmonitor::ChannelMonitor<TestChannelSigner>)>::
read(&mut Cursor::new(&map_entry.1), (&*self.keys, &*self.keys)).unwrap().1;
read(&mut Cursor::new(&latest_monitor_data), (&*self.keys, &*self.keys)).unwrap().1;
deserialized_monitor.update_monitor(update, &&TestBroadcaster{}, &&FuzzEstimator { ret_val: atomic::AtomicU32::new(253) }, &self.logger).unwrap();
let mut ser = VecWriter(Vec::new());
deserialized_monitor.write(&mut ser).unwrap();
*map_entry = (update.update_id, ser.0);
self.chain_monitor.update_channel(funding_txo, update)
let res = self.chain_monitor.update_channel(funding_txo, update);
match res {
chain::ChannelMonitorUpdateStatus::Completed => {
map_entry.persisted_monitor_id = update.update_id;
map_entry.persisted_monitor = ser.0;
},
chain::ChannelMonitorUpdateStatus::InProgress => {
map_entry.pending_monitors.push((update.update_id, ser.0));
},
chain::ChannelMonitorUpdateStatus::UnrecoverableError => panic!(),
}
res
}

fn release_pending_monitor_events(&self) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, Option<PublicKey>)> {
Expand DownExpand Up@@ -511,9 +548,15 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {

let mut monitors = new_hash_map();
let mut old_monitors = $old_monitors.latest_monitors.lock().unwrap();
for (outpoint, (update_id, monitor_ser)) in old_monitors.drain() {
monitors.insert(outpoint, <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(&mut Cursor::new(&monitor_ser), (&*$keys_manager, &*$keys_manager)).expect("Failed to read monitor").1);
chain_monitor.latest_monitors.lock().unwrap().insert(outpoint, (update_id, monitor_ser));
for (outpoint, mut prev_state) in old_monitors.drain() {
monitors.insert(outpoint, <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(
&mut Cursor::new(&prev_state.persisted_monitor), (&*$keys_manager, &*$keys_manager)
).expect("Failed to read monitor").1);
// Wipe any `ChannelMonitor`s which we never told LDK we finished persisting,
// considering them discarded. LDK should replay these for us as they're stored in
// the `ChannelManager`.
prev_state.pending_monitors.clear();
chain_monitor.latest_monitors.lock().unwrap().insert(outpoint, prev_state);
}
let mut monitor_refs = new_hash_map();
for (outpoint, monitor) in monitors.iter_mut() {
Expand DownExpand Up@@ -1040,6 +1083,43 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
} }
}

let complete_first = |v: &mut Vec<_>| if !v.is_empty() { Some(v.remove(0)) } else { None };
let complete_second = |v: &mut Vec<_>| if v.len() > 1 { Some(v.remove(1)) } else { None };
let complete_monitor_update = |
monitor: &Arc<TestChainMonitor>, chan_funding,
compl_selector: &dyn Fn(&mut Vec<(u64, Vec<u8>)>) -> Option<(u64, Vec<u8>)>,
| {
if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
assert!(
state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
"updates should be sorted by id"
);
if let Some((id, data)) = compl_selector(&mut state.pending_monitors) {
monitor.chain_monitor.channel_monitor_updated(*chan_funding, id).unwrap();
if id > state.persisted_monitor_id {
state.persisted_monitor_id = id;
state.persisted_monitor = data;
}
}
}
};

let complete_all_monitor_updates = |monitor: &Arc<TestChainMonitor>, chan_funding| {
Comment thread
tnull marked this conversation as resolved.
if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
assert!(
state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
"updates should be sorted by id"
);
for (id, data) in state.pending_monitors.drain(..) {
monitor.chain_monitor.channel_monitor_updated(*chan_funding, id).unwrap();
if id > state.persisted_monitor_id {
state.persisted_monitor_id = id;
state.persisted_monitor = data;
}
}
}
};

let v = get_slice!(1)[0];
out.locked_write(format!("READ A BYTE! HANDLING INPUT {:x}...........\n", v).as_bytes());
match v {
Expand All@@ -1054,30 +1134,10 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
0x05 => *monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed,
0x06 => *monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed,

0x08 => {
if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[0].process_monitor_events();
}
},
0x09 => {
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[1].process_monitor_events();
}
},
0x0a => {
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[1].process_monitor_events();
}
},
0x0b => {
if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_c.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[2].process_monitor_events();
}
},
0x08 => complete_all_monitor_updates(&monitor_a, &chan_1_funding),
0x09 => complete_all_monitor_updates(&monitor_b, &chan_1_funding),
0x0a => complete_all_monitor_updates(&monitor_b, &chan_2_funding),
0x0b => complete_all_monitor_updates(&monitor_c, &chan_2_funding),

0x0c => {
if !chan_a_disconnected {
Expand DownExpand Up@@ -1285,119 +1345,35 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
},
0x89 => { fee_est_c.ret_val.store(253, atomic::Ordering::Release); nodes[2].maybe_update_chan_fees(); },

0xf0 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf1 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf2 => {
let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[0].process_monitor_events();
}
0xf0 => complete_monitor_update(&monitor_a, &chan_1_funding, &complete_first),
0xf1 => complete_monitor_update(&monitor_a, &chan_1_funding, &complete_second),
0xf2 => complete_monitor_update(&monitor_a, &chan_1_funding, &Vec::pop),

0xf4 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf5 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf6 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf4 => complete_monitor_update(&monitor_b, &chan_1_funding, &complete_first),
0xf5 => complete_monitor_update(&monitor_b, &chan_1_funding, &complete_second),
0xf6 => complete_monitor_update(&monitor_b, &chan_1_funding, &Vec::pop),

0xf8 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf9 => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xfa => {
let pending_updates = monitor_b.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[1].process_monitor_events();
}
0xf8 => complete_monitor_update(&monitor_b, &chan_2_funding, &complete_first),
0xf9 => complete_monitor_update(&monitor_b, &chan_2_funding, &complete_second),
0xfa => complete_monitor_update(&monitor_b, &chan_2_funding, &Vec::pop),

0xfc => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(0) {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfd => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.get(1) {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfe => {
let pending_updates = monitor_c.chain_monitor.list_pending_monitor_updates().remove(&chan_2_funding).unwrap();
if let Some(id) = pending_updates.last() {
monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id).unwrap();
}
nodes[2].process_monitor_events();
}
0xfc => complete_monitor_update(&monitor_c, &chan_2_funding, &complete_first),
0xfd => complete_monitor_update(&monitor_c, &chan_2_funding, &complete_second),
0xfe => complete_monitor_update(&monitor_c, &chan_2_funding, &Vec::pop),

0xff => {
// Test that no channel is in a stuck state where neither party can send funds even
// after we resolve all pending events.
// First make sure there are no pending monitor updates, resetting the error state
// and calling force_channel_monitor_updated for each monitor.
// First make sure there are no pending monitor updates and further update
// operations complete.
*monitor_a.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
*monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
*monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;

if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[0].process_monitor_events();
}
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
nodes[1].process_monitor_events();
}
if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_b.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[1].process_monitor_events();
}
if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
monitor_c.chain_monitor.force_channel_monitor_updated(chan_2_funding, *id);
nodes[2].process_monitor_events();
}
complete_all_monitor_updates(&monitor_a, &chan_1_funding);
complete_all_monitor_updates(&monitor_b, &chan_1_funding);
complete_all_monitor_updates(&monitor_b, &chan_2_funding);
complete_all_monitor_updates(&monitor_c, &chan_2_funding);

// Next, make sure peers are all connected to each other
if chan_a_disconnected {
Expand Down
Loading