rebalance: support custom CLN version strings

rebalance/clnutils.py

@@ -9,6 +9,9 @@ def cln_parse_rpcversion(string):
     to ubuntu style with version 22.11  `yy.mm[.patch][-mod]`
     make sure we can read all of them for (the next 80 years).
     """
+    if not isinstance(string, str):
+        return None
+
     rpcversion = string
     if rpcversion.startswith("v"):  # strip leading 'v'
         rpcversion = rpcversion[1:]
@@ -19,5 +22,11 @@ def cln_parse_rpcversion(string):
     if rpcversion.count(".") == 1:  # imply patch version 0 if not given
         rpcversion = rpcversion + ".0"
 
-    # split and convert numeric string parts to actual integers
-    return list(map(int, rpcversion.split(".")))
+    # Custom builds may replace the version with a branch or package name.
+    # Callers must feature-detect RPC compatibility when no version is available.
+    try:
+        version = list(map(int, rpcversion.split(".")))
+    except ValueError:
+        return None
+
+    return version if len(version) >= 2 else None

rebalance/rebalance.py

@@ -22,6 +22,28 @@
 plugin.rebalance_stop_by_event = False
 plugin.threadids = {}
 
+REQUIRED_RPCS = {
+    "askrene-create-layer",
+    "askrene-disable-node",
+    "askrene-inform-channel",
+    "askrene-remove-layer",
+    "askrene-update-channel",
+    "delinvoice",
+    "getroutes",
+    "invoice",
+    "listchannels",
+    "listconfigs",
+    "listforwards",
+    "listfunds",
+    "listinvoices",
+    "listnodes",
+    "listpays",
+    "listpeerchannels",
+    "listpeers",
+    "sendpay",
+    "waitsendpay",
+}
+
 
 def rebalance_stopping():
     return (
@@ -36,76 +58,52 @@ def get_thread_id_str():
     return f"{plugin.threadids.get(threading.get_ident(), 0):{'0' + str(max_digits)}}"
 
 
-# The route msat helpers are needed because older versions of cln
-# have different field names, that were replaced in newer versions
+# The route helpers support both getroutes path formats. Custom builds may not
+# expose a parseable version, so infer the format from the returned path.
+def route_uses_modern_fields(obj):
+    return "node_id_out" in obj
+
+
 def route_set_msat(obj, msat):
-    if (
-        plugin.rpcversion[0] > 26
-        or plugin.rpcversion[0] == 26
-        and plugin.rpcversion[1] >= 6
-    ):
+    if route_uses_modern_fields(obj):
         obj["amount_in_msat"] = Millisatoshi(msat)
     else:
         obj["amount_msat"] = Millisatoshi(msat)
 
 
 def route_set_out_msat(obj, msat):
-    if (
-        plugin.rpcversion[0] > 26
-        or plugin.rpcversion[0] == 26
-        and plugin.rpcversion[1] >= 6
-    ):
+    if route_uses_modern_fields(obj):
         obj["amount_out_msat"] = Millisatoshi(msat)
 
 
 def route_get_msat(r):
-    if (
-        plugin.rpcversion[0] > 26
-        or plugin.rpcversion[0] == 26
-        and plugin.rpcversion[1] >= 6
-    ):
+    if route_uses_modern_fields(r):
         return r["amount_in_msat"]
     else:
         return r["amount_msat"]
 
 
 def route_set_delay(obj, delay):
-    if (
-        plugin.rpcversion[0] > 26
-        or plugin.rpcversion[0] == 26
-        and plugin.rpcversion[1] >= 6
-    ):
+    if route_uses_modern_fields(obj):
         obj["cltv_out"] = delay
 
 
 def route_set_in_delay(obj, delay):
-    if (
-        plugin.rpcversion[0] > 26
-        or plugin.rpcversion[0] == 26
-        and plugin.rpcversion[1] >= 6
-    ):
+    if route_uses_modern_fields(obj):
         obj["cltv_in"] = delay
     else:
         obj["delay"] = delay
 
 
 def route_get_id(r):
-    if (
-        plugin.rpcversion[0] > 26
-        or plugin.rpcversion[0] == 26
-        and plugin.rpcversion[1] >= 6
-    ):
+    if route_uses_modern_fields(r):
         return r["node_id_out"]
     else:
         return r.get("next_node_id") or r["id"]
 
 
 def route_get_scid(r):
-    if (
-        plugin.rpcversion[0] > 26
-        or plugin.rpcversion[0] == 26
-        and plugin.rpcversion[1] >= 6
-    ):
+    if route_uses_modern_fields(r):
         return r["short_channel_id_dir"].split("/")[0]
     else:
         scidd = r.get("short_channel_id_dir")
@@ -116,11 +114,7 @@ def route_get_scid(r):
 
 
 def getroutes_to_sendpay(route, msat):
-    if (
-        plugin.rpcversion[0] < 26
-        or plugin.rpcversion[0] == 26
-        and plugin.rpcversion[1] < 6
-    ):
+    if not route_uses_modern_fields(route[0]):
         sendpay_route = []
         for i, r in enumerate(route):
             sr = {}
@@ -220,7 +214,7 @@ def amounts_from_scid(scid):
 def peer_from_scid(short_channel_id, my_node_id, payload):
     channels = plugin.rpc.listpeerchannels().get("channels")
     for ch in channels:
-        if ch["short_channel_id"] == short_channel_id:
+        if ch.get("short_channel_id") == short_channel_id:
             return ch["peer_id"]
     raise RpcError(
         "rebalance",
@@ -441,12 +435,6 @@ def rebalance(
         "debug",
     )
 
-    is_26_06_or_later = (
-        plugin.rpcversion[0] > 26
-        or plugin.rpcversion[0] == 26
-        and plugin.rpcversion[1] >= 6
-    )
-
     start_ts = int(time.time())
     label = "Rebalance-" + str(uuid.uuid4())
     description = "%s to %s" % (outgoing_scid, incoming_scid)
@@ -467,32 +455,6 @@ def rebalance(
 
     try:
         while int(time.time()) - start_ts < retry_for and not rebalance_stopping():
-            if is_26_06_or_later:
-                route_out = {
-                    "node_id_out": outgoing_node_id,
-                    "short_channel_id_dir": out_alias
-                    + "/"
-                    + str(int(not my_node_id < outgoing_node_id)),
-                }
-                route_in = {
-                    "node_id_out": my_node_id,
-                    "short_channel_id_dir": in_alias
-                    + "/"
-                    + str(int(not incoming_node_id < my_node_id)),
-                }
-            else:
-                route_out = {
-                    "next_node_id": outgoing_node_id,
-                    "short_channel_id_dir": out_alias
-                    + "/"
-                    + str(int(not my_node_id < outgoing_node_id)),
-                }
-                route_in = {
-                    "next_node_id": my_node_id,
-                    "short_channel_id_dir": in_alias
-                    + "/"
-                    + str(int(not incoming_node_id < my_node_id)),
-                }
             count += 1
             try:
                 time_start = time.time()
@@ -516,6 +478,23 @@ def rebalance(
                     raise e
 
             route_mid = r["routes"][0]["path"]
+            id_field = (
+                "node_id_out"
+                if route_uses_modern_fields(route_mid[0])
+                else "next_node_id"
+            )
+            route_out = {
+                id_field: outgoing_node_id,
+                "short_channel_id_dir": out_alias
+                + "/"
+                + str(int(not my_node_id < outgoing_node_id)),
+            }
+            route_in = {
+                id_field: my_node_id,
+                "short_channel_id_dir": in_alias
+                + "/"
+                + str(int(not incoming_node_id < my_node_id)),
+            }
             route = [route_out] + route_mid + [route_in]
             setup_routing_fees(route, msatoshi)
             route = getroutes_to_sendpay(route, msatoshi)
@@ -1225,7 +1204,20 @@ def init(options: dict, configuration: dict, plugin: Plugin, **kwargs):
     plugin.getinfo = plugin.rpc.getinfo()
     plugin.rpcversion = cln_parse_rpcversion(plugin.getinfo.get("version"))
 
-    if (
+    if plugin.rpcversion is None:
+        rpc_commands = {
+            entry["command"].split()[0] for entry in plugin.rpc.help()["help"]
+        }
+        missing_commands = REQUIRED_RPCS - rpc_commands
+        if missing_commands:
+            missing = ", ".join(sorted(missing_commands))
+            return {"disable": f"Required RPC commands unavailable: {missing}"}
+        plugin.log(
+            f"Unparseable CLN version '{plugin.getinfo.get('version')}'; "
+            "validated required RPC commands instead",
+            "warn",
+        )
+    elif (
         plugin.rpcversion[0] < 25
         or plugin.rpcversion[0] == 25
         and plugin.rpcversion[1] < 9
@@ -1303,4 +1295,5 @@ def init(options: dict, configuration: dict, plugin: Plugin, **kwargs):
     "string",
 )
 
-plugin.run()
+if __name__ == "__main__":
+    plugin.run()

rebalance/test_clnutils.py

@@ -0,0 +1,129 @@
+import importlib.util
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from pyln.client import Millisatoshi
+
+from clnutils import cln_parse_rpcversion
+
+
+spec = importlib.util.spec_from_file_location(
+    "rebalance_plugin", Path(__file__).with_name("rebalance.py")
+)
+rebalance = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(rebalance)
+
+
+@pytest.mark.parametrize(
+    "version, expected",
+    [
+        ("v25.09", [25, 9, 0]),
+        ("v26.06.1", [26, 6, 1]),
+        ("v26.06rc2", [26, 6, 0]),
+        ("v26.06.1-custom", [26, 6, 1]),
+        ("v25", None),
+        ("opencode/spenderp-defer-awaiting-recovery", None),
+        (None, None),
+    ],
+)
+def test_cln_parse_rpcversion(version, expected):
+    assert cln_parse_rpcversion(version) == expected
+
+
+class FakeRpc:
+    def __init__(self, commands):
+        self.commands = commands
+
+    def getinfo(self):
+        return {"version": "opencode/spenderp-defer-awaiting-recovery"}
+
+    def help(self):
+        return {"help": [{"command": command} for command in self.commands]}
+
+    def listconfigs(self):
+        return {
+            "configs": {
+                "cltv-final": {"value_int": 18},
+                "fee-base": {"value_int": 1000},
+                "fee-per-satoshi": {"value_int": 10},
+            }
+        }
+
+
+def fake_plugin(commands):
+    return SimpleNamespace(rpc=FakeRpc(commands), log=lambda *args: None)
+
+
+def test_init_accepts_custom_version_with_required_rpcs():
+    plugin = fake_plugin(rebalance.REQUIRED_RPCS)
+    options = {
+        "rebalance-getroute": None,
+        "rebalance-maxhops": None,
+        "rebalance-msatfactor": None,
+        "rebalance-erringnodes": "5",
+        "rebalance-threads": "8",
+    }
+
+    assert rebalance.init(options, {}, plugin) is None
+    assert plugin.rpcversion is None
+    assert plugin.cltv_final == 18
+    assert plugin.fee_base == Millisatoshi(1000)
+    assert plugin.fee_ppm == 10
+    assert not plugin.mutex.locked()
+
+
+def test_init_rejects_custom_version_without_required_rpc():
+    plugin = fake_plugin(rebalance.REQUIRED_RPCS - {"getroutes"})
+
+    assert rebalance.init({}, {}, plugin) == {
+        "disable": "Required RPC commands unavailable: getroutes"
+    }
+
+
+def test_modern_route_is_passed_to_sendpay_unchanged():
+    route = [
+        {
+            "node_id_out": "node-a",
+            "short_channel_id_dir": "1x1x1/0",
+            "amount_in_msat": Millisatoshi(2000),
+            "amount_out_msat": Millisatoshi(1000),
+            "cltv_in": 24,
+            "cltv_out": 18,
+        }
+    ]
+
+    assert rebalance.getroutes_to_sendpay(route, Millisatoshi(1000)) is route
+
+
+def test_legacy_route_is_converted_for_sendpay():
+    rebalance.plugin.cltv_final = 18
+    route = [
+        {
+            "next_node_id": "node-a",
+            "short_channel_id_dir": "1x1x1/0",
+            "amount_msat": Millisatoshi(2000),
+            "delay": 24,
+        },
+        {
+            "next_node_id": "node-b",
+            "short_channel_id_dir": "2x2x2/1",
+            "amount_msat": Millisatoshi(1000),
+            "delay": 18,
+        },
+    ]
+
+    assert rebalance.getroutes_to_sendpay(route, Millisatoshi(1000)) == [
+        {
+            "id": "node-a",
+            "channel": "1x1x1",
+            "delay": 18,
+            "amount_msat": Millisatoshi(1000),
+        },
+        {
+            "id": "node-b",
+            "channel": "2x2x2",
+            "delay": 18,
+            "amount_msat": Millisatoshi(1000),
+        },
+    ]