chanbackup: only store backups for peers with current/previous channels.

This seems fair.

Signed-off-by: Rusty Russell <[email protected]>

Author: rustyrussell

plugins/chanbackup.c

@@ -388,9 +388,12 @@ static struct command_result *peer_after_listdatastore(struct command *cmd,
 						       const u8 *hexdata,
 						       struct node_id *nodeid)
 {
+        struct out_req *req;
+
+	/* We use an empty record to indicate we *would* store data
+	 * for this peer, until we actually get data from them */
         if (tal_bytelen(hexdata) == 0)
         	return command_hook_success(cmd);
-        struct out_req *req;
 
 	if (!chanbackup(cmd->plugin)->peer_backup)
 		return command_hook_success(cmd);
@@ -582,6 +585,15 @@ static struct command_result *after_staticbackup(struct command *cmd,
 	return send_outreq(req);
 }
 
+static struct command_result *datastore_create_done(struct command *cmd,
+						    const char *method,
+						    const char *buf,
+						    const jsmntok_t *result,
+						    void *unused)
+{
+	return notification_or_hook_done(cmd);
+}
+
 static struct command_result *json_state_changed(struct command *cmd,
 					         const char *buf,
 					         const jsmntok_t *params)
@@ -604,6 +616,32 @@ static struct command_result *json_state_changed(struct command *cmd,
 		return send_outreq(req);
 	}
 
+	/* Once it has a normal channel, we will start storing its
+	 * backups: put an empty record in place. */
+	if (json_tok_streq(buf, statetok, "CHANNELD_NORMAL")) {
+		const jsmntok_t *nodeid_tok;
+		struct node_id node_id;
+
+		nodeid_tok = json_get_member(buf, notiftok, "peer_id");
+		if (!json_to_node_id(buf, nodeid_tok, &node_id))
+			plugin_err(cmd->plugin, "Invalid peer_id in %.*s",
+				   json_tok_full_len(notiftok),
+				   json_tok_full(buf, notiftok));
+
+		/* Might already exist, that's OK, just don't overwrite! */
+		return jsonrpc_set_datastore_binary(cmd,
+					     	    tal_fmt(cmd,
+						    	    "chanbackup/peers/%s",
+					     	     	    fmt_node_id(tmpctx,
+									&node_id)),
+						    /* Empty record */
+						    tal_arr(tmpctx, u8, 0),
+						    "must-create",
+					     	    datastore_create_done,
+					     	    datastore_create_done,
+						    NULL);
+	}
+
 	return notification_or_hook_done(cmd);
 }
 
@@ -723,7 +761,10 @@ static struct command_result *handle_your_peer_storage(struct command *cmd,
 		 *      - MUST store the message.
 		 *    - MAY store the message anyway.
 		 */
-		/* FIXME: Only store peers we want to! */
+		/* We store if we have a datastore slot for it
+		 * (otherwise, this fails).  We create those once it
+		 * has a channel, though the user could also create an
+		 * empty one if they wanted to */
 
 		/* BOLT #1:
 		 * - If it does store the message:
@@ -737,7 +778,7 @@ static struct command_result *handle_your_peer_storage(struct command *cmd,
 					     	     	    fmt_node_id(tmpctx,
 									&node_id)),
 						    payload_deserialise,
-						    "create-or-replace",
+						    "must-replace",
 					     	    datastore_success,
 					     	    datastore_failed,
 						    "Saving chanbackup/peers/");

tests/test_misc.py

@@ -3171,6 +3171,15 @@ def test_restorefrompeer(node_factory, bitcoind):
     assert l1.daemon.is_in_log('Peer storage sent!')
     assert l2.daemon.is_in_log('Peer storage sent!')
 
+    # Note: each node may or may not send peer_storage_retrieval: if it
+    # receives storage fast enough, it will, otherwise not.
+    l1.rpc.disconnect(l2.info['id'], force=True)
+    l1.rpc.connect(l2.info['id'], 'localhost', l2.port)
+    l1.daemon.wait_for_logs(['peer_out WIRE_PEER_STORAGE',
+                             'peer_in WIRE_PEER_STORAGE'])
+    l2.daemon.wait_for_logs(['peer_out WIRE_PEER_STORAGE',
+                             'peer_in WIRE_PEER_STORAGE'])
+
     l1.stop()
     os.unlink(os.path.join(l1.daemon.lightning_dir, TEST_NETWORK, "lightningd.sqlite3"))
 

tests/test_plugin.py

@@ -4558,3 +4558,94 @@ def write_all(fd, bytestr):
 
     with pytest.raises(RpcError, match="maybe encrypted"):
         l1.rpc.exposesecret(passphrase=password)
+
+
+def test_peer_storage(node_factory, bitcoind):
+    """Test that we offer and re-xmit peer storage for our peers if they have a channel or are explicitly enabled"""
+    l1, l2, l3 = node_factory.get_nodes(3,
+                                        opts={'experimental-peer-storage': None,
+                                              'may_reconnect': True,
+                                              'dev-no-reconnect': None})
+
+    # Connect them, no peer storage yet anyway.
+    l1.rpc.connect(l2.info['id'], 'localhost', l2.port)
+
+    assert not l1.daemon.is_in_log(r'WIRE_PEER_STORAGE_RETRIEVAL')
+    assert not l2.daemon.is_in_log(r'WIRE_PEER_STORAGE_RETRIEVAL')
+
+    # And they won't store.
+    assert l1.rpc.listdatastore(['chanbackup', 'peers', l2.info['id']]) == {'datastore': []}
+    assert l2.rpc.listdatastore(['chanbackup', 'peers', l1.info['id']]) == {'datastore': []}
+
+    # Reconnect, still no xmit.
+    l1.rpc.disconnect(l2.info['id'])
+    l1.rpc.connect(l2.info['id'], 'localhost', l2.port)
+
+    # Give it time to make sure we *would* catch it if it happened.
+    time.sleep(10)
+    assert not l1.daemon.is_in_log(r'WIRE_PEER_STORAGE_RETRIEVAL')
+    assert not l2.daemon.is_in_log(r'WIRE_PEER_STORAGE_RETRIEVAL')
+
+    # But we can force it manually by creating an empty one.
+    l1.rpc.datastore(['chanbackup', 'peers', l2.info['id']], hex='')
+    assert l1.rpc.listdatastore(['chanbackup', 'peers', l2.info['id']])['datastore'][0]['hex'] == ''
+    l1.restart()
+    l1.rpc.connect(l2.info['id'], 'localhost', l2.port)
+
+    # Note: l1 may or may not send peer_storage_retrieval: if it
+    # receives storage fast enough, it will, otherwise not.
+    wait_for(lambda: l1.rpc.listdatastore(['chanbackup', 'peers', l2.info['id']])['datastore'][0]['hex'] != '')
+
+    # Next reconnect, l1 will send it back.
+    l1.rpc.disconnect(l2.info['id'])
+    l1.rpc.connect(l2.info['id'], 'localhost', l2.port)
+
+    l1.daemon.wait_for_log(r'peer_out WIRE_PEER_STORAGE_RETRIEVAL')
+    l2.daemon.wait_for_log(r'peer_in WIRE_PEER_STORAGE_RETRIEVAL')
+    assert not l1.daemon.is_in_log(r'peer_in WIRE_PEER_STORAGE_RETRIEVAL')
+    assert not l2.daemon.is_in_log(r'peer_out WIRE_PEER_STORAGE_RETRIEVAL')
+
+    # It must be valid.
+    l2.daemon.wait_for_log(r'Received peer_storage from peer')
+    assert not l2.daemon.is_in_log(r'PeerStorageFailed')
+
+    # l2 will only store if we have an ESTABLISHED channel.
+    l1.openchannel(l2, confirm=False, wait_for_announce=False)
+    assert l2.rpc.listdatastore(['chanbackup', 'peers', l1.info['id']]) == {'datastore': []}
+
+    # But it will create an entry once it hits CHANNELD_NORMAL.
+    bitcoind.generate_block(1, wait_for_mempool=1)
+    wait_for(lambda: l2.rpc.listdatastore(['chanbackup', 'peers', l1.info['id']]) != {'datastore': []})
+
+    # Now it will store l1's backup when channels change.
+    l1.openchannel(l3, wait_for_announce=False)
+    wait_for(lambda: l2.rpc.listdatastore(['chanbackup', 'peers', l1.info['id']])['datastore'][0]['hex'] != '')
+
+    # Now every time we reconnect, both sides restore.
+    l1.restart()
+    l1.rpc.connect(l2.info['id'], 'localhost', l2.port)
+
+    l1.daemon.wait_for_logs([r'peer_out WIRE_PEER_STORAGE_RETRIEVAL',
+                             r'peer_in WIRE_PEER_STORAGE_RETRIEVAL'])
+    l2.daemon.wait_for_logs([r'peer_out WIRE_PEER_STORAGE_RETRIEVAL',
+                             r'peer_in WIRE_PEER_STORAGE_RETRIEVAL'])
+
+    # Even if we close channel, and it's long forgotten, we will store for
+    # them as a courtesy.
+    l1.rpc.close(l2.info['id'])
+    assert len(l1.rpc.listpeerchannels()['channels']) == 2
+    bitcoind.generate_block(100, wait_for_mempool=1)
+    wait_for(lambda: len(l1.rpc.listpeerchannels()['channels']) == 1)
+
+    # Now try restarting l2 and connecting that way instead.
+    l2.restart()
+    l2.rpc.connect(l1.info['id'], 'localhost', l1.port)
+
+    l1.daemon.wait_for_logs([r'peer_out WIRE_PEER_STORAGE_RETRIEVAL',
+                             r'peer_in WIRE_PEER_STORAGE_RETRIEVAL'])
+    l2.daemon.wait_for_logs([r'peer_out WIRE_PEER_STORAGE_RETRIEVAL',
+                             r'peer_in WIRE_PEER_STORAGE_RETRIEVAL'])
+
+    # This should never happen
+    assert not l1.daemon.is_in_log(r'PeerStorageFailed')
+    assert not l2.daemon.is_in_log(r'PeerStorageFailed')