[fusion] better consolidation logic

- At most 3 outputs when we consolidate.
- Therefore consolidation has an endpoint.
- Try to slow down consolidations in late stages.

(Maybe max_outputs should be made into a gui/config option
for the power gamers?)

plugins/fusion/fusion.py

@@ -72,12 +72,15 @@
 # The largest 'excess fee' that we are willing to pay in a fusion (fees beyond
 # those needed to pay for our components' inclusion)
 MAX_EXCESS_FEE = 10000
-# Even if the server allows more, put at most this many inputs+outputs.
+# Even if the server allows more, put at most this many inputs+outputs+blanks
 MAX_COMPONENTS = 40
 # The largest total fee we are willing to pay (our contribution to transaction
 # size should not exceed 7 kB even with 40 largest components).
 MAX_FEE = MAX_COMPONENT_FEERATE * 7 + MAX_EXCESS_FEE
 
+# For privacy reasons, don't submit less than this many inputs+outputs
+MIN_TX_COMPONENTS = 11
+
 def can_fuse_from(wallet):
     """We can only fuse from wallets that are p2pkh, and where we are able
     to extract the private key."""
@@ -254,6 +257,7 @@ class Fusion(threading.Thread, PrintError):
     """
     stopping=False
     stopping_if_not_running=False
+    max_outputs = None
     status=('setup', None) # will always be 2-tuple; second param has extra details
 
     def __init__(self, plugin, target_wallet, server_host, server_port, server_ssl, tor_host, tor_port):
@@ -526,22 +530,33 @@ def greet(self,):
             raise FusionError('excessive min excess fee from server')
         if self.min_excess_fee > self.max_excess_fee:
             raise FusionError('bad config on server: fees')
+        if self.num_components < MIN_TX_COMPONENTS * 1.5:
+            raise FusionError('bad config on server: num_components')
 
     def allocate_outputs(self,):
         assert self.status[0] in ('setup', 'connecting')
-        num_inputs = len(self.coins)
 
         # fix the input selection
         self.inputs = tuple(self.coins.items())
+        num_inputs = len(self.inputs)
+
+        # For obfuscation, when there are few inputs we want to have many outputs,
+        # and vice versa. Many of both is even better, of course.
+        min_outputs = max(MIN_TX_COMPONENTS - num_inputs, 1)
 
         maxcomponents = min(self.num_components, MAX_COMPONENTS)
         max_outputs = maxcomponents - num_inputs
         if max_outputs < 1:
             raise FusionError('Too many inputs (%d >= %d)'%(num_inputs, maxcomponents))
 
-        # For obfuscation, when there are few inputs we want to have many outputs,
-        # and vice versa. Many of both is even better, of course.
-        min_outputs = max(11 - num_inputs, 1)
+        if self.max_outputs is not None:
+            assert self.max_outputs >= 1
+            if self.max_outputs < min_outputs:
+                raise FusionError('Too few inputs (%d) for output count constraint (<=%d)'%(num_inputs, self.max_outputs))
+            max_outputs = min(self.max_outputs, max_outputs)
+
+        if max_outputs < min_outputs:
+            raise FusionError('Impossible output count range (%d, %d)'%(min_outputs, max_outputs))
 
         # how much input value do we bring to the table (after input & player fees)
         sum_inputs_value = sum(v for (_,_), (p,v) in self.inputs)

plugins/fusion/plugin.py

@@ -42,7 +42,7 @@
 from electroncash import Network
 
 from .conf import Conf, Global
-from .fusion import Fusion, can_fuse_from, can_fuse_to, is_tor_port
+from .fusion import Fusion, can_fuse_from, can_fuse_to, is_tor_port, MIN_TX_COMPONENTS
 from .server import FusionServer
 from .covert import limiter
 
@@ -69,6 +69,8 @@
 # how many autofusions can be running per-wallet
 MAX_AUTOFUSIONS_PER_WALLET = 10
 
+CONSOLIDATE_MAX_OUTPUTS = MIN_TX_COMPONENTS // 3
+
 pnp = None
 def get_upnp():
     """ return an initialized UPnP singleton """
@@ -201,28 +203,49 @@ def select_random_coins(wallet, fraction, eligible):
 
     return result
 
-def get_target_params_1(wallet, eligible):
+def get_target_params_1(wallet, wallet_conf, active_autofusions, eligible):
     """ WIP -- TODO: Rename this function. """
     wallet_conf = Conf(wallet)
     mode = wallet_conf.fusion_mode
 
+    # Note each fusion 'consumes' a certain number of coins by freezing them,
+    # so that the next fusion has less eligible coins to work with. So each
+    # call to this may see a smaller n_coins.
     n_coins = sum(len(acoins) for addr,acoins in eligible)
     if mode == 'normal':
         return max(2, round(n_coins / DEFAULT_MAX_COINS)), False
     elif mode == 'fan-out':
         return max(4, math.ceil(n_coins / (COIN_FRACTION_FUDGE_FACTOR*0.65))), False
     elif mode == 'consolidate':
-        num_threads = math.trunc(n_coins / (COIN_FRACTION_FUDGE_FACTOR*1.5))
-        return num_threads, num_threads <= 1
+        if len(eligible) < MIN_TX_COMPONENTS - CONSOLIDATE_MAX_OUTPUTS:
+            # Too few eligible buckets to make an effective consolidation.
+            return 0, False
+
+        # In the latter stages of consolidation, only do one fusion
+        # at a time with all-confirmed rule, to make sure each fusion's outputs
+        # may be consumed by the subsequent one.
+        # To avoid weird loops, try to calculate the TOTAL number of coins
+        # that are either 1) eligible or 2) being fused. (Should stay constant
+        # as fusions are added/cancelled)
+        n_total = n_coins + sum(len(f.inputs) for f in active_autofusions)
+        if n_total < DEFAULT_MAX_COINS*3:
+            return 1, True
+
+        # If coins are scarce then don't make more autofusions unless we
+        # have none.
+        if n_coins < DEFAULT_MAX_COINS*2:
+            return 1, False
+
+        # We still have lots of coins left, so request another autofusion.
+        return MAX_AUTOFUSIONS_PER_WALLET, False
     else:  # 'custom'
         target_num_auto = wallet_conf.queued_autofuse
         confirmed_only = wallet_conf.autofuse_confirmed_only
         return int(target_num_auto), bool(confirmed_only)
 
 
-def get_target_params_2(wallet, eligible, sum_value):
+def get_target_params_2(wallet_conf, sum_value):
     """ WIP -- TODO: Rename this function. """
-    wallet_conf = Conf(wallet)
     mode = wallet_conf.fusion_mode
 
     fraction = 0.1
@@ -456,7 +479,7 @@ def remove_wallet(self, wallet):
         return [f for f in fusions if f.status[0] not in ('complete', 'failed')]
 
 
-    def create_fusion(self, source_wallet, password, coins, target_wallet = None):
+    def create_fusion(self, source_wallet, password, coins, target_wallet = None, max_outputs = None):
         """ Create a new Fusion object with current server/tor settings. Once created
         you must call fusion.start() to launch it.
 
@@ -485,6 +508,7 @@ def create_fusion(self, source_wallet, password, coins, target_wallet = None):
         target_wallet._fusions.add(fusion)
         source_wallet._fusions.add(fusion)
         fusion.add_coins_from_wallet(source_wallet, password, coins)
+        fusion.max_outputs = max_outputs
         with self.lock:
             self.fusions[fusion] = time.time()
         return fusion
@@ -532,29 +556,34 @@ def run(self, ):
             with wallet.lock:
                 if not hasattr(wallet, '_fusions'):
                     continue
-                num_auto = 0
                 for f in list(wallet._fusions_auto):
                     if f.status[0] in ('complete', 'failed'):
                         wallet._fusions_auto.discard(f)
-                    else:
-                        num_auto += 1
+                active_autofusions = list(wallet._fusions_auto)
+                num_auto = len(active_autofusions)
+                wallet_conf = Conf(wallet)
                 eligible, ineligible, sum_value, has_unconfirmed, has_coinbase = select_coins(wallet)
-                target_num_auto, confirmed_only = get_target_params_1(wallet, eligible)
-                #self.print_error("params1", target_num_auto, confirmed_only)
+                target_num_auto, confirmed_only = get_target_params_1(wallet, wallet_conf, active_autofusions, eligible)
                 if confirmed_only and has_unconfirmed:
                     for f in list(wallet._fusions_auto):
                         f.stop('Wallet has unconfirmed coins... waiting.', not_if_running = True)
                     continue
                 if num_auto < min(target_num_auto, MAX_AUTOFUSIONS_PER_WALLET):
                     # we don't have enough auto-fusions running, so start one
-                    fraction = get_target_params_2(wallet, eligible, sum_value)
-                    #self.print_error("params2", fraction)
+                    fraction = get_target_params_2(wallet_conf, sum_value)
                     coins = [c for l in select_random_coins(wallet, fraction, eligible) for c in l]
                     if not coins:
                         self.print_error("auto-fusion skipped due to lack of coins")
                         continue
+                    if wallet_conf.fusion_mode == 'consolidate':
+                        max_outputs = CONSOLIDATE_MAX_OUTPUTS
+                        if len(coins) < (MIN_TX_COMPONENTS - max_outputs):
+                            self.print_error("consolidating auto-fusion skipped due to lack of unrelated coins")
+                            continue
+                    else:
+                        max_outputs = None
                     try:
-                        f = self.create_fusion(wallet, password, coins)
+                        f = self.create_fusion(wallet, password, coins, max_outputs = max_outputs)
                         f.start(inactive_timeout = AUTOFUSE_INACTIVE_TIMEOUT)
                         self.print_error("started auto-fusion")
                     except RuntimeError as e: