Circuitbreaker support

It is now possible to control the node limits of a circuitbreaker
instance dynamically.

README.md

@@ -11,7 +11,8 @@ See [INSTALL.md](/INSTALL.md)
 charge-lnd takes only a minimal set of parameters:
 
 ```
-usage: charge-lnd [-h] [--lnddir LNDDIR] [--grpc GRPC] [--dry-run] [--check] [-v] -c CONFIG
+usage: charge-lnd [-h] [--lnddir LNDDIR] [--tlscert TLS_CERT_PATH] [--macaroon MACAROON_PATH] [--grpc GRPC]
+                  [--circuitbreaker CIRCUITBREAKER] [--dry-run] [--check] [-v] -c CONFIG
 
 optional arguments:
   -h, --help            show this help message and exit
@@ -21,6 +22,8 @@ optional arguments:
   --macaroon MACAROON_PATH
                         (default [lnddir]/data/chain/bitcoin/mainnet/charge-lnd.macaroon) path to lnd auth macaroons
   --grpc GRPC           (default localhost:10009) lnd gRPC endpoint
+  --circuitbreaker CIRCUITBREAKER
+                        (optional, no default) circuitbreaker gRPC endpoint host:port
   --dry-run             Do not perform actions (for testing), print what we would do to
                         stdout
   --check               Do not perform actions, only check config file for valid syntax
@@ -217,7 +220,24 @@ All strategies (except the ignore strategy) will apply the following properties
 | **max_htlc_msat_ratio** | Maximum size of HTLC to allow as a fraction of total channel capacity | 0..1 |
 | **time_lock_delta** | Time Lock Delta | # blocks |
 | **min_fee_ppm_delta** | Minimum change in fees (ppm) before updating channel | ppm delta |
+| **cb_max_hourly_rate** | Circuitbreaker: maximum number of incoming htlcs per hour | # hourly rate |
+| **cb_max_pending** | Circuitbreaker: maximum number of incoming htlcs at the same time | # incoming pending htlcs |
+| **cb_mode** | Circuitbreaker: mode (0 - FAIL; 1 - QUEUE; 2 - QUEUE_PEER_INITIATED; 3 - BLOCK) | 0..3 |
+| **cb_clear_limit** | Circuitbreaker: delete the peer limit and fallback to the default limit | true |
 
+### Circuitbreaker Support
+Optionally, it is also possible to dynamically control the [circuitbreaker](https://github.com/lightningequipment/circuitbreaker) limits for individual peers. However, the default limit of the circuitbreaker cannot currently be changed with `charge-lnd`.
+
+### One channel for the peer
+
+If any of the properties `cb_max_hourly_rate`, `cb_max_pending`, or `cb_mode` are set, the node limit will be adjusted. It should be noted that 0 for the first two properties is interpreted as infinite. Those properties that are not currently set will be set according to the respective default limits. If no node limit is to be set, but a limit has already been set, the reset to the default limit can be performed by setting `cb_clear_limit`.
+
+### Multiple channels for the peer
+
+It can happen that different properties for individual channels are chosen when there are multiple channels with one peer. Since the circuitbreaker monitors its limits at the peer level, we aggregate the properties into a node limit:
+- `cb_max_hourly_rate` and `cb_max_pending` are added, considering that 0 corresponds to infinite.
+- `cb_mode` is set to the most conservative mode, i.e., the first mode from the following list is used: `MODE_BLOCK`, `MODE_FAIL`, `MODE_QUEUE`, `MODE_QUEUE_PEER_INITIATED`.
+- The node limit will be deleted if `cb_clear_limit` is set for a channel.
 
 ## Contributing
 

charge_lnd/charge_lnd.py

@@ -11,6 +11,7 @@ from .lnd import Lnd
 from .policy import Policies
 from .strategy import is_defined
 from .config import Config
+from .circuitbreaker import Circuitbreaker
 import charge_lnd.fmt as fmt
 
 def debug(message):
@@ -35,9 +36,19 @@ def main():
 
     lnd = Lnd(arguments.lnddir, arguments.grpc, arguments.tls_cert_path, arguments.macaroon_path)
     if not lnd.valid:
-        debug("Could not connect to gRPC endpoint")
+        debug("Could not connect to lnd gRPC endpoint")
         return False
 
+    cb = None
+    if arguments.circuitbreaker:
+        cb = Circuitbreaker(arguments.circuitbreaker)
+        if not cb.valid:
+            debug("Could not connect to circuitbreaker gRPC endpoint")
+            return False
+        if cb.get_info().node_key != lnd.get_info().identity_pubkey:
+            debug("node_key of circuitbreaker is different from pubkey of lnd")
+            return False
+
     policies = Policies(lnd, config)
 
     my_pubkey = lnd.get_own_pubkey()
@@ -50,6 +61,9 @@ def main():
 
         chp = policy.strategy.execute(channel)
 
+        if cb and is_defined(chp.circuitbreaker_params):
+            cb.apply_params(chp.circuitbreaker_params, channel.remote_pubkey)
+
         if channel.chan_id in lnd.feereport:
             (current_base_fee_msat, current_fee_ppm) = lnd.feereport[channel.chan_id]
 
@@ -144,8 +158,73 @@ def main():
                     s = ' ➜ ' + fmt.col_hi(chp.time_lock_delta)
                 print("  time_lock_delta:         %s%s" % (fmt.col_hi(my_policy.time_lock_delta), s) )
 
+    if cb:
+         update_circuitbreaker(cb, lnd, arguments)
+
     return True
 
+# Updates the circuitbreaker backend with all necessary limit changes.
+def update_circuitbreaker(cb: Circuitbreaker, lnd: Lnd, arguments):
+    clear_limits, update_limits = cb.get_limit_updates()
+
+    # The for loop is only for printing the changes to the current state.
+    for peer in clear_limits + list(update_limits.keys()):
+        limit_current = cb.get_limit(peer)
+
+        is_deleted = peer in clear_limits and limit_current is not None
+        is_new = peer in update_limits and limit_current is None
+
+        is_update_candidate = peer in update_limits and limit_current is not None
+        is_updated = False
+        if is_update_candidate:
+            is_updated_rate = update_limits[peer].max_hourly_rate != limit_current.max_hourly_rate
+            is_updated_pending = update_limits[peer].max_pending != limit_current.max_pending
+            is_updated_mode = update_limits[peer].mode != limit_current.mode
+            is_updated = any([is_updated_rate, is_updated_pending, is_updated_mode])
+
+        is_changed = any([is_deleted, is_new, is_updated])
+
+        if is_changed or (is_update_candidate and arguments.verbose):
+            print(fmt.print_node(lnd.get_node_info(peer)))
+            print("  service:                 circuitbreaker")
+
+        if is_deleted:
+            s = ' ➜ ' + fmt.col_hi("default")
+            print("  max_hourly_rate:         %s%s" % (fmt.col_hi(limit_current.max_hourly_rate), s) )
+            print("  max_pending:             %s%s" % (fmt.col_hi(limit_current.max_pending), s) )
+            print("  mode:                    %s%s" % (fmt.col_hi(limit_current.mode), s) )
+
+        if is_new:
+            s = fmt.col_hi("default") + ' ➜ '
+            print("  max_hourly_rate:         %s%s" % (s, fmt.col_hi(update_limits[peer].max_hourly_rate)) )
+            print("  max_pending:             %s%s" % (s, fmt.col_hi(update_limits[peer].max_pending)) )
+            print("  mode:                    %s%s" % (s, fmt.col_hi(update_limits[peer].mode)) )
+
+        if is_updated or (is_update_candidate and arguments.verbose):
+            if is_updated_rate or arguments.verbose:
+                s = ''
+                if is_updated_rate:
+                    s = ' ➜ ' + fmt.col_hi(update_limits[peer].max_hourly_rate)
+                print("  max_hourly_rate:         %s%s" % (fmt.col_hi(limit_current.max_hourly_rate), s) )
+
+            if is_updated_pending or arguments.verbose:
+                s = ''
+                if is_updated_pending:
+                    s = ' ➜ ' + fmt.col_hi(update_limits[peer].max_pending)
+                print("  max_pending:             %s%s" % (fmt.col_hi(limit_current.max_pending), s) )
+
+            if is_updated_mode or arguments.verbose:
+                s = ''
+                if is_updated_mode:
+                    s = ' ➜ ' + fmt.col_hi(update_limits[peer].mode)
+                print("  mode:                    %s%s" % (fmt.col_hi(limit_current.mode), s) )
+
+    # Eventually, we are updating the circuitbreaker backend.
+    if not arguments.dry_run:
+        cb.clear_limits(clear_limits)
+        cb.update_limits(update_limits)
+
+
 def get_argument_parser():
     parser = argparse.ArgumentParser()
     parser.add_argument("--lnddir",
@@ -162,6 +241,9 @@ def get_argument_parser():
                         default="localhost:10009",
                         dest="grpc",
                         help="(default localhost:10009) lnd gRPC endpoint")
+    parser.add_argument("--circuitbreaker",
+                        dest="circuitbreaker",
+                        help="(optional, no default) circuitbreaker gRPC endpoint host:port")
     parser.add_argument("--dry-run",
                         dest="dry_run",
                         action="store_true",

charge_lnd/circuitbreaker.py

@@ -0,0 +1,140 @@
+import grpc
+import operator
+import sys
+
+from .grpc_generated import circuitbreaker_pb2_grpc as cbrpc, circuitbreaker_pb2 as cb
+from typing import Optional
+from types import SimpleNamespace
+
+MESSAGE_SIZE_MB = 50 * 1024 * 1024
+
+
+def debug(message):
+    sys.stderr.write(message + "\n")
+
+# Aggregation function for two circuitbreaker modes, choosing
+# the most conservative one.
+def add_cb_modes(x, y):
+    # MODE_BLOCK > MODE_FAIL > MODE_QUEUE > MODE_QUEUE_PEER_INITIATED
+    prio={3: 0, 0: 1, 1: 2, 2: 3}
+
+    if prio[x] < prio[y]:
+        return x
+    return y
+
+# Aggregation of two numbers with 0 is equal to infinity.
+def add_with_infty(x, y):
+    if x == 0 or y == 0:
+        return 0
+    return x + y
+
+# Aggregation of two operands with an arbitragy operator,
+# where both operands can be none.
+def add_with_none(x, y, op):
+    if x is None:
+        s = y
+    elif y is None:
+        s = x
+    else:
+        s = op(x,y)
+    return s
+
+class CircuitbreakerParams(SimpleNamespace):
+    max_hourly_rate: Optional[int] = None
+    max_pending: Optional[int] = None
+    mode: Optional[int] = None
+    clear_limit: Optional[bool] = None
+
+    # In the case that we have several channels open for a peer,
+    # we would like to aggregate the limit conservatively.
+    def __add__(self, add):
+        sum = CircuitbreakerParams()
+        sum.max_hourly_rate = add_with_none(self.max_hourly_rate, add.max_hourly_rate, add_with_infty)
+        sum.max_pending = add_with_none(self.max_pending, add.max_pending, add_with_infty)
+        sum.mode = add_with_none(self.mode, add.mode, add_cb_modes)
+        sum.clear_limit = add_with_none(self.clear_limit, add.clear_limit, operator.or_)
+
+        return sum
+
+class Circuitbreaker:
+    def __init__(self, server):
+        channel_options = [
+            ('grpc.max_message_length', MESSAGE_SIZE_MB),
+            ('grpc.max_receive_message_length', MESSAGE_SIZE_MB)
+        ]
+        grpc_channel = grpc.insecure_channel(server, channel_options)
+        self.stub = cbrpc.ServiceStub(grpc_channel)
+        self.info = None
+        self.valid = True
+        # dict of existing limits
+        self.dict_limits = None
+        # dict of circuitbreaker params per peer
+        self.peer_params = {}
+
+        try:
+            _ = self.get_info()
+        except grpc._channel._InactiveRpcError:
+            self.valid = False
+
+    def get_info(self):
+        if not self.info:
+            self.info = self.stub.GetInfo(cb.GetInfoRequest())
+        return self.info
+
+    def list_limits(self):
+        return self.stub.ListLimits(cb.ListLimitsRequest())
+
+    # Returns the current limit. limit can be either a pubkey
+    # of a peer or 'default' for the default limit. We are returning
+    # None if no peer limit is set.
+    def get_limit(self, limitid):
+        if self.dict_limits is None:
+            self.dict_limits = {}
+            list_limits = self.list_limits()
+            for l in list_limits.limits:
+                if l.HasField("limit"):
+                    self.dict_limits[l.node] = l.limit
+                else:
+                    self.dict_limits[l.node] = None
+            self.dict_limits["default"] = list_limits.default_limit
+
+        return self.dict_limits.get(limitid)
+
+    # Updating the internal circuitbreaker limits for a peer but not
+    # sending updates to the backend.
+    def apply_params(self, params, peerid):
+        if peerid not in self.peer_params:
+            self.peer_params[peerid] = CircuitbreakerParams()
+        
+        self.peer_params[peerid] += params
+
+    # Returns the necessary limit deletes and updates to make backend
+    # consistent with the current params.
+    def get_limit_updates(self):
+        clear_limits = []
+        update_limits = {}
+        for k, v in self.peer_params.items():
+            if not any([
+                v.max_hourly_rate is not None,
+                v.max_pending is not None,
+                v.mode is not None
+            ]):
+                if v.clear_limit is not None and v.clear_limit:
+                    clear_limits.append(k)
+                continue
+
+            limit_ref = self.get_limit("default")
+            limit = cb.Limit(
+                max_hourly_rate=limit_ref.max_hourly_rate if v.max_hourly_rate is None else v.max_hourly_rate,
+                max_pending=limit_ref.max_pending if v.max_pending is None else v.max_pending,
+                mode=limit_ref.mode if v.mode is None else v.mode
+            )
+            update_limits[k] = limit
+
+        return clear_limits, update_limits
+
+    def update_limits(self, limits):
+        return self.stub.UpdateLimits(cb.UpdateLimitsRequest(limits=limits))
+
+    def clear_limits(self, nodes):
+        return self.stub.ClearLimits(cb.ClearLimitsRequest(nodes=nodes))

charge_lnd/strategy.py

@@ -6,6 +6,7 @@ from types import SimpleNamespace
 
 from . import fmt
 from .config import Config
+from .circuitbreaker import CircuitbreakerParams
 
 def debug(message):
     sys.stderr.write(message + "\n")
@@ -25,6 +26,7 @@ class ChanParams(SimpleNamespace):
     inbound_base_fee_msat: Optional[Union[str, int]] = DONTCARE
     inbound_fee_ppm: Optional[Union[str, int]] = DONTCARE
     disabled: Optional[Union[str, bool]] = DONTCARE
+    circuitbreaker_params: Optional[Union[str, CircuitbreakerParams]] = DONTCARE
 
 def strategy(_func=None,*,name):
     def register_strategy(func):
@@ -56,6 +58,13 @@ class StrategyDelegate:
                 result.time_lock_delta = self.policy.getint('time_lock_delta')
             if result.disabled == DONTCARE:
                 result.disabled = False
+            if result.circuitbreaker_params == DONTCARE:
+                result.circuitbreaker_params = CircuitbreakerParams(
+                    max_hourly_rate=self.policy.getint('cb_max_hourly_rate'),
+                    max_pending=self.policy.getint('cb_max_pending'),
+                    mode=self.policy.getint('cb_mode'),
+                    clear_limit=self.policy.getbool('cb_clear_limit')
+                )
 
             return result
         except Exception as e:
@@ -86,7 +95,8 @@ def strategy_ignore(channel, policy, **kwargs):
         time_lock_delta=KEEP,
         inbound_base_fee_msat=KEEP,
         inbound_fee_ppm=KEEP,
-        disabled=KEEP
+        disabled=KEEP,
+        circuitbreaker_params=KEEP
     )
 
 @strategy(name = 'ignore_fees')
@@ -244,4 +254,6 @@ def strategy_disable(channel, policy, **kwargs):
 
     chanparams = strategy_ignore(channel, policy)
     chanparams.disabled=True
+    # We want to allow changes to the Circuitbreaker params, such as blocking incoming htlcs.
+    chanparams.circuitbreaker_params = DONTCARE
     return chanparams