tests: lnpeer.htlc_switch: don't fulfill htlc until add is irrevocable

This adds a failing test, where the HTLC switch fulfills an HTLC too soon,
before the corresponding 'update_add_htlc' is irrevocably committed.

electrum/lnchannel.py

@@ -997,6 +997,7 @@ class Channel(AbstractChannel):
             self.hm.recv_rev()
             self.config[REMOTE].current_per_commitment_point=self.config[REMOTE].next_per_commitment_point
             self.config[REMOTE].next_per_commitment_point=revocation.next_per_commitment_point
+        assert new_ctn == self.get_oldest_unrevoked_ctn(REMOTE)
         # lnworker callbacks
         if self.lnworker:
             sent = self.hm.sent_in_ctn(new_ctn)

electrum/lnhtlc.py

@@ -313,6 +313,46 @@ class HTLCManager:
                 return True
         return False
 
+    @with_lock
+    def is_add_htlc_irrevocably_committed_yet(
+            self,
+            *,
+            ctx_owner: HTLCOwner = None,
+            htlc_proposer: HTLCOwner,
+            htlc_id: int,
+    ) -> bool:
+        """Returns whether `add_htlc` was irrevocably committed to `ctx_owner's` ctx.
+        If `ctx_owner` is None, both parties' ctxs are checked.
+        """
+        in_local = self._is_add_htlc_irrevocably_committed_yet(
+            ctx_owner=LOCAL, htlc_proposer=htlc_proposer, htlc_id=htlc_id)
+        in_remote = self._is_add_htlc_irrevocably_committed_yet(
+            ctx_owner=REMOTE, htlc_proposer=htlc_proposer, htlc_id=htlc_id)
+        if ctx_owner is None:
+            return in_local and in_remote
+        elif ctx_owner == LOCAL:
+            return in_local
+        elif ctx_owner == REMOTE:
+            return in_remote
+        else:
+            raise Exception(f"unexpected ctx_owner: {ctx_owner!r}")
+
+    @with_lock
+    def _is_add_htlc_irrevocably_committed_yet(
+            self,
+            *,
+            ctx_owner: HTLCOwner,
+            htlc_proposer: HTLCOwner,
+            htlc_id: int,
+    ) -> bool:
+        htlc_id = int(htlc_id)
+        if htlc_id >= self.get_next_htlc_id(htlc_proposer):
+            return False
+        ctns = self.log[htlc_proposer]['locked_in'][htlc_id]
+        if ctns[ctx_owner] is None:
+            return False
+        return ctns[ctx_owner] <= self.ctn_oldest_unrevoked(ctx_owner)
+
     @with_lock
     def htlcs_by_direction(self, subject: HTLCOwner, direction: Direction,
                            ctn: int = None) -> Dict[int, UpdateAddHtlc]:

electrum/lnpeer.py

@@ -1432,6 +1432,7 @@ class Peer(Logger):
     def fulfill_htlc(self, chan: Channel, htlc_id: int, preimage: bytes):
         self.logger.info(f"_fulfill_htlc. chan {chan.short_channel_id}. htlc_id {htlc_id}")
         assert chan.can_send_ctx_updates(), f"cannot send updates: {chan.short_channel_id}"
+        assert chan.hm.is_add_htlc_irrevocably_committed_yet(htlc_proposer=REMOTE, htlc_id=htlc_id)
         chan.settle_htlc(preimage, htlc_id)
         self.send_message("update_fulfill_htlc",
                           channel_id=chan.channel_id,
@@ -1665,6 +1666,7 @@ class Peer(Logger):
                 done = set()
                 unfulfilled = chan.hm.log.get('unfulfilled_htlcs', {})
                 for htlc_id, (local_ctn, remote_ctn, onion_packet_hex, forwarding_info) in unfulfilled.items():
+                    # FIXME this test is not sufficient:
                     if chan.get_oldest_unrevoked_ctn(LOCAL) <= local_ctn:
                         continue
                     if chan.get_oldest_unrevoked_ctn(REMOTE) <= remote_ctn:

electrum/lnworker.py

@@ -1299,6 +1299,7 @@ class LNWallet(LNWorker):
             self.set_payment_status(bfh(key), status)
 
     async def await_payment(self, payment_hash: bytes) -> BarePaymentAttemptLog:
+        # note side-effect: Future is created and added here (defaultdict):
         payment_attempt = await self.pending_payments[payment_hash]
         self.pending_payments.pop(payment_hash)
         return payment_attempt

electrum/tests/test_lnpeer.py

@@ -467,6 +467,72 @@ class TestPeer(ElectrumTestCase):
         with self.assertRaises(PaymentDone):
             run(f())
 
+    @needs_test_with_all_chacha20_implementations
+    def test_payment_race(self):
+        """Alice and Bob pay each other simultaneously.
+        They both send 'update_add_htlc' and receive each other's update
+        before sending 'commitment_signed'. Neither party should fulfill
+        the respective HTLCs until those are irrevocably committed to.
+        """
+        alice_channel, bob_channel = create_test_channels()
+        p1, p2, w1, w2, _q1, _q2 = self.prepare_peers(alice_channel, bob_channel)
+        async def pay():
+            await asyncio.wait_for(p1.initialized, 1)
+            await asyncio.wait_for(p2.initialized, 1)
+            # prep
+            _maybe_send_commitment1 = p1.maybe_send_commitment
+            _maybe_send_commitment2 = p2.maybe_send_commitment
+            pay_req2 = await self.prepare_invoice(w2)
+            lnaddr2 = lndecode(pay_req2, expected_hrp=constants.net.SEGWIT_HRP)
+            pay_req1 = await self.prepare_invoice(w1)
+            lnaddr1 = lndecode(pay_req1, expected_hrp=constants.net.SEGWIT_HRP)
+            # alice sends htlc BUT NOT COMMITMENT_SIGNED
+            p1.maybe_send_commitment = lambda x: None
+            p1.pay(
+                route=w1._create_route_from_invoice(decoded_invoice=lnaddr2),
+                chan=alice_channel,
+                amount_msat=lnaddr2.get_amount_msat(),
+                payment_hash=lnaddr2.paymenthash,
+                min_final_cltv_expiry=lnaddr2.get_min_final_cltv_expiry(),
+                payment_secret=lnaddr2.payment_secret,
+            )
+            w1.pending_payments[lnaddr2.paymenthash] = asyncio.Future()
+            p1.maybe_send_commitment = _maybe_send_commitment1
+            # bob sends htlc BUT NOT COMMITMENT_SIGNED
+            p2.maybe_send_commitment = lambda x: None
+            p2.pay(
+                route=w2._create_route_from_invoice(decoded_invoice=lnaddr1),
+                chan=bob_channel,
+                amount_msat=lnaddr1.get_amount_msat(),
+                payment_hash=lnaddr1.paymenthash,
+                min_final_cltv_expiry=lnaddr1.get_min_final_cltv_expiry(),
+                payment_secret=lnaddr1.payment_secret,
+            )
+            w2.pending_payments[lnaddr1.paymenthash] = asyncio.Future()
+            p2.maybe_send_commitment = _maybe_send_commitment2
+            # sleep a bit so that they both receive msgs sent so far
+            await asyncio.sleep(0.1)
+            # now they both send COMMITMENT_SIGNED
+            p1.maybe_send_commitment(alice_channel)
+            p2.maybe_send_commitment(bob_channel)
+
+            payment_attempt1 = await w1.await_payment(lnaddr2.paymenthash)
+            assert payment_attempt1.success
+            payment_attempt2 = await w2.await_payment(lnaddr1.paymenthash)
+            assert payment_attempt2.success
+            raise PaymentDone()
+
+        async def f():
+            async with TaskGroup() as group:
+                await group.spawn(p1._message_loop())
+                await group.spawn(p1.htlc_switch())
+                await group.spawn(p2._message_loop())
+                await group.spawn(p2.htlc_switch())
+                await asyncio.sleep(0.01)
+                await group.spawn(pay())
+        with self.assertRaises(PaymentDone):
+            run(f())
+
     #@unittest.skip("too expensive")
     #@needs_test_with_all_chacha20_implementations
     def test_payments_stresstest(self):