Merge #677: Handle push tx

4440ffb Taker broadcasts tx after unconfirm_timeout_sec. (Adam Gibson)
d209d4d Fixes #368. random-peer and not-self tx broadcast. (Adam Gibson)

jmbase/jmbase/commands.py

@@ -220,3 +220,11 @@ class JMTXReceived(JMCommand):
     arguments = [(b'nick', Unicode()),
                  (b'txhex', Unicode()),
                  (b'offer', Unicode())]
+
+class JMTXBroadcast(JMCommand):
+    """ Accept a bitcoin transaction
+    sent over the wire by a counterparty
+    and relay it to the client for network
+    broadcast.
+    """
+    arguments = [(b'txhex', Unicode())]

jmbase/test/test_commands.py

@@ -111,7 +111,9 @@ class JMTestServerProtocol(JMBaseProtocol):
                                         hashlen=4,
                                         max_encoded=5,
                                         hostid="hostid2")
-        self.defaultCallbacks(d3)        
+        self.defaultCallbacks(d3)
+        d4 = self.callRemote(JMTXBroadcast, txhex="deadbeef")
+        self.defaultCallbacks(d4)
         return {'accepted': True}
             
 
@@ -216,6 +218,11 @@ class JMTestClientProtocol(JMBaseProtocol):
         self.defaultCallbacks(d)
         return {'accepted': True}
 
+    @JMTXBroadcast.responder
+    def on_JM_TX_BROADCAST(self, txhex):
+        show_receipt("JMTXBROADCAST", txhex)
+        return {"accepted": True}
+
 class JMTestClientProtocolFactory(protocol.ClientFactory):
     protocol = JMTestClientProtocol
 

jmclient/jmclient/client_protocol.py

@@ -301,6 +301,22 @@ class JMMakerClientProtocol(JMClientProtocol):
             self.defaultCallbacks(d)
         return {"accepted": True}
 
+    @commands.JMTXBroadcast.responder
+    def on_JM_TX_BROADCAST(self, txhex):
+        """ Makers have no issue broadcasting anything,
+        so only need to prevent crashes.
+        Note in particular we don't check the return value,
+        since the transaction being accepted or not is not
+        our (maker)'s concern.
+        """
+        try:
+            txbin = hextobin(txhex)
+            jm_single().bc_interface.pushtx(txbin)
+        except:
+            jlog.info("We received an invalid transaction broadcast "
+                      "request: " + txhex)
+        return {"accepted": True}
+
     def tx_match(self, txd):
         for k,v in self.finalized_offers.items():
             # Tx considered defined by its output set

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).
-# self = broadcast transaction with your bitcoin node's ip
+# options: self, random-peer, not-self.
+# 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.txid, self.unconfirm_callback,
-                        "unconfirmed",
-        "transaction with txid: " + str(self.txid) + " not broadcast.")
+                        self.handle_unbroadcast_transaction, self.txid, self.latest_tx)
 
         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,20 +836,19 @@ 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:
             if nick_to_use:
-                # TODO option not currently functional
-                return (nick_to_use, self.latest_tx.serialize())
+                return (nick_to_use, bintohex(self.latest_tx.serialize()))
         #if push was not successful, return None
 
     def self_sign_and_push(self):

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
-                # processed and so do nothing.
+                return False
+            # 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

jmdaemon/jmdaemon/daemon_protocol.py

@@ -9,6 +9,7 @@ from .protocol import (COMMAND_PREFIX, ORDER_KEYS, NICK_HASH_LENGTH,
                        COMMITMENT_PREFIXES)
 from .irc import IRCMessageChannel
 
+from jmbase import hextobin
 from jmbase.commands import *
 from twisted.protocols import amp
 from twisted.internet import reactor, ssl
@@ -432,9 +433,16 @@ class JMDaemonServerProtocol(amp.AMP, OrderbookWatch):
 
     @maker_only
     def on_push_tx(self, nick, txhex):
-        """Not yet implemented; ignore rather than raise.
+        """Broadcast unquestioningly, except checking
+        hex format.
 	"""
-        log.msg('received pushtx message, ignoring, TODO')
+        try:
+            dummy = hextobin(txhex)
+        except:
+            return
+        d = self.callRemote(JMTXBroadcast,
+                            txhex=txhex)
+        self.defaultCallbacks(d)
 
     @maker_only
     def on_seen_tx(self, nick, txhex):