Taker broadcasts tx after unconfirm_timeout_sec.

Prior to this commit, if non-self broadcast was enabled
but the counterparty chosen did not broadcast, the transaction
would remain unbroadcast.
After this commit, the Taker checks, after the configured
value of TIMEOUT.unconfirm_timeout_sec (default 90s), the
Taker will broadcast the transaction.
Also amended default config comment for this function.

jmclient/jmclient/configure.py

@@ -244,10 +244,12 @@ absurd_fee_per_kb = 350000
 # spends from unconfirmed inputs, which may then get malleated or double-spent!
 # other counterparties are likely to reject unconfirmed inputs... don't do it.
 
-# options: self, random-peer, not-self (note: currently, ONLY 'self' works).
+# options: self, random-peer, not-self.
-# self = broadcast transaction with your bitcoin node's ip
+# self = broadcast transaction with your own bitcoin node.
 # random-peer = everyone who took part in the coinjoin has a chance of broadcasting
 # not-self = never broadcast with your own ip
+# note: if your counterparties do not support it, you will fall back
+# to broadcasting via your own node.
 tx_broadcast = self
 
 # If makers do not respond while creating a coinjoin transaction,

jmclient/jmclient/taker.py

@@ -781,6 +781,27 @@ class Taker(object):
         if not success:
             jlog.error("Failed to sign transaction: " + msg)
 
+    def handle_unbroadcast_transaction(self, txid, tx):
+        """ The wallet service will handle dangling
+        callbacks for transactions but we want to reattempt
+        broadcast in case the cause of the problem is a
+        counterparty who refused to broadcast it for us.
+        """
+        if not self.wallet_service.check_callback_called(
+            self.txid, self.unconfirm_callback, "unconfirmed",
+            "transaction with txid: " + str(self.txid) + " not broadcast."):
+            # we now know the transaction was not pushed, so we reinstigate
+            # the cancelledcallback with the same logic as explained
+            # in Taker.push():
+            self.wallet_service.register_callbacks([self.unconfirm_callback],
+                                                   txid, "unconfirmed")
+            if not self.push_ourselves():
+                jlog.error("Failed to broadcast transaction: ")
+                jlog.info(btc.human_readable_transaction(tx))
+
+    def push_ourselves(self):
+        return jm_single().bc_interface.pushtx(self.latest_tx.serialize())
+
     def push(self):
         jlog.debug('\n' + bintohex(self.latest_tx.serialize()))
         self.txid = bintohex(self.latest_tx.GetTxid()[::-1])
@@ -802,15 +823,12 @@ class Taker(object):
         task.deferLater(reactor,
                         float(jm_single().config.getint(
                             "TIMEOUT", "unconfirm_timeout_sec")),
-                        self.wallet_service.check_callback_called,
+                        self.handle_unbroadcast_transaction, self.txid, self.latest_tx)
-                        self.txid, self.unconfirm_callback,
-                        "unconfirmed",
-        "transaction with txid: " + str(self.txid) + " not broadcast.")
 
         tx_broadcast = jm_single().config.get('POLICY', 'tx_broadcast')
         nick_to_use = None
         if tx_broadcast == 'self':
-            pushed = jm_single().bc_interface.pushtx(self.latest_tx.serialize())
+            pushed = self.push_ourselves()
         elif tx_broadcast in ['random-peer', 'not-self']:
             n = len(self.maker_utxo_data)
             if tx_broadcast == 'random-peer':
@@ -818,14 +836,14 @@ class Taker(object):
             else:
                 i = random.randrange(n)
             if i == n:
-                pushed = jm_single().bc_interface.pushtx(self.latest_tx.serialize())
+                pushed = self.push_ourselves()
             else:
                 nick_to_use = list(self.maker_utxo_data.keys())[i]
                 pushed = True
         else:
             jlog.info("Only self, random-peer and not-self broadcast "
                       "methods supported. Reverting to self-broadcast.")
-            pushed = jm_single().bc_interface.pushtx(self.latest_tx.serialize())
+            pushed = self.push_ourselves()
         if not pushed:
             self.on_finished_callback(False, fromtx=True)
         else:

jmclient/jmclient/wallet_service.py

@@ -377,6 +377,7 @@ class WalletService(Service):
         """ Intended to be a deferred Task to be scheduled some
         set time after the callback was registered. "all" type
         callbacks do not expire and are not included.
+        If the callback was previously called, return True, otherwise False.
         """
         assert cbtype in ["unconfirmed", "confirmed"]
         if txinfo in self.callbacks[cbtype]:
@@ -389,8 +390,10 @@ class WalletService(Service):
                 # this never occurs, although their presence should
                 # not cause a functional error.
                 jlog.info("Timed out: " + msg)
-                # if callback is not in the list, it was already
+                return False
-                # processed and so do nothing.
+            # if callback is not in the list, it was already
+            # processed and so do nothing.
+        return True
 
     def log_new_tx(self, removed_utxos, added_utxos, txid):
         """ Changes to the wallet are logged at INFO level by