tests: fix flaky LN tests (MPP timeout and orphaned tasks)

Two fixes:

1. test_trampoline_mpp_consolidation: set dave.MPP_EXPIRY=120 when
   test_mpp_consolidation=True. With MPP_EXPIRY=2s and sequential HTLC
   commitment rounds, Dave times out before the second HTLC arrives.

2. test_reestablish_fake_data: use explicit Task references in the
   payment setup phase so that loop tasks (message loops, htlc_switch)
   are cancelled in a finally block regardless of whether pay() succeeds
   or raises. Without this, asyncio.gather leaves orphaned tasks running
   across sub-test iterations when a payment fails, causing interference

tests/test_lnpeer.py

@@ -774,15 +774,29 @@ class TestPeerDirect(TestPeer):
             alice_channel, bob_channel = create_test_channels(alice_lnwallet=alice_lnwallet, bob_lnwallet=bob_lnwallet)
             p1, p2, w1, w2 = self.prepare_peers(alice_channel, bob_channel)
             # first make some payments, to bump the channel ctns a bit
+            loop_tasks = [
+                asyncio.ensure_future(p1._message_loop()),
+                asyncio.ensure_future(p2._message_loop()),
+                asyncio.ensure_future(p1.htlc_switch()),
+                asyncio.ensure_future(p2.htlc_switch()),
+            ]
             async def pay():
                 for pnum in range(2):
                     lnaddr, pay_req = self.prepare_invoice(w2)
                     result, log = await w1.pay_invoice(pay_req)
                     self.assertEqual(result, True)
                 gath.cancel()
-            gath = asyncio.gather(pay(), p1._message_loop(), p2._message_loop(), p1.htlc_switch(), p2.htlc_switch())
-            with self.assertRaises(asyncio.CancelledError):
-                await gath
+            gath = asyncio.gather(pay(), *loop_tasks)
+            try:
+                with self.assertRaises(asyncio.CancelledError):
+                    await gath
+            finally:
+                # cancel loop_tasks explicitly: asyncio.gather does not cancel
+                # its child tasks when it fails due to an exception in pay(),
+                # so without this orphaned tasks accumulate across sub-test iterations
+                for t in loop_tasks:
+                    t.cancel()
+                await asyncio.gather(*loop_tasks, return_exceptions=True)
             for chan in (alice_channel, bob_channel):
                 chan.peer_state = PeerState.DISCONNECTED
 
@@ -2730,6 +2744,7 @@ class TestPeerForwarding(TestPeer):
         graph = self.prepare_chans_and_peers_in_graph(graph_definition)
         if test_mpp_consolidation:
             graph.workers['dave'].features |= LnFeatures.BASIC_MPP_OPT
+            graph.workers['dave'].MPP_EXPIRY = 120  # HTLCs arrive at Dave sequentially; give enough time for both to arrive
             graph.workers['alice'].network.config.TEST_FORCE_MPP = True # trampoline must wait until all incoming htlcs are received before sending outgoing htlcs
             graph.workers['bob'].network.config.TEST_FORCE_MPP = True   # trampoline must wait until all outgoing htlcs have failed before failing incoming htlcs
         if is_legacy: