Manage exceptions in bcc callback functions

Exceptions are not propagated in ctype callback functions used by bcc.
This means an AssertionError exception raised by check_equal() to signal
a failed assertion is not getting caught and properly logged. Instead,
the error is logged to stdout and execution of the handler stops.

The current workaround to check whether all check_equal() assertions in
a callback succeeded is to increment a success counter after the
assertions (which only gets incremented if none exception is raised and
stops execution). Then, outside the callback, the success counter can be
used to check whether a callback executed successfully.

One issue with the described workaround is that when an exception
occurs, there is no way of telling which of the check_equal() statements
caused the exception; moreover, there is no way of inspecting how the
pieces of data that got compared in check_equal() differed (often
a crucial clue when debugging what went wrong).

Two fixes to this problem come to mind. The first involves having the
callback function make event data accessible outside the callback and
inspecting the event using check_equal() outside the callback. This
solution still requires a counter in the callback to tell whether
a callback was actually executed or if instead the call to
perf_buffer_poll() timed out.

The second fix entails wrapping all relevant check_equal() statements
inside callback functions into try-catch blocks and manually logging
AssertionErrors. While not as elegant in terms of design, this approach
can be more pragmatic for more complex tests (e.g., ones involving
multiple events, events of different types, or the order of events).

The solution proposed here is to select the most pragmatic fix on
a case-by-case basis: Tests in interface_usdt_net.py,
interface_usdt_mempool.py and interface_usdt_validation.py have been
refactored to use the first approach, while the second approach was
chosen for interface_usdt_utxocache.py (partly to provide a reference
for the second approach, but mainly because the utxocache tests are the
most intricate tests, and refactoring them to use the first approach
would negatively impact their readability). Lastly,
interface_usdt_coinselection.py was kept unchanged because it does not
use check_equal() statements inside callback functions.

4 files changed, 88 insertions, 90 deletions

diff --git a/test/functional/interface_usdt_mempool.py b/test/functional/interface_usdt_mempool.py
index 542849d558..62fdc28360 100755
--- a/test/functional/interface_usdt_mempool.py
+++ b/test/functional/interface_usdt_mempool.py
@@ -139,6 +139,7 @@ class MempoolTracepointTest(BitcoinTestFramework):
 
         EXPECTED_ADDED_EVENTS = 1
         handled_added_events = 0
+        event = None
 
         self.log.info("Hooking into mempool:added tracepoint...")
         node = self.nodes[0]
@@ -147,11 +148,8 @@ class MempoolTracepointTest(BitcoinTestFramework):
         bpf = BPF(text=MEMPOOL_TRACEPOINTS_PROGRAM, usdt_contexts=[ctx], debug=0)
 
         def handle_added_event(_, data, __):
-            nonlocal handled_added_events
+            nonlocal event, handled_added_events
             event = bpf["added_events"].event(data)
-            assert_equal(txid, bytes(event.hash)[::-1].hex())
-            assert_equal(vsize, event.vsize)
-            assert_equal(fee, event.fee)
             handled_added_events += 1
 
         bpf["added_events"].open_perf_buffer(handle_added_event)
@@ -159,9 +157,6 @@ class MempoolTracepointTest(BitcoinTestFramework):
         self.log.info("Sending transaction...")
         fee = Decimal(31200)
         tx = self.wallet.send_self_transfer(from_node=node, fee=fee / COIN)
-        # expected data
-        txid = tx["txid"]
-        vsize = tx["tx"].get_vsize()
 
         self.log.info("Polling buffer...")
         bpf.perf_buffer_poll(timeout=200)
@@ -169,10 +164,13 @@ class MempoolTracepointTest(BitcoinTestFramework):
         self.log.info("Cleaning up mempool...")
         self.generate(node, 1)
 
-        bpf.cleanup()
-
         self.log.info("Ensuring mempool:added event was handled successfully...")
         assert_equal(EXPECTED_ADDED_EVENTS, handled_added_events)
+        assert_equal(bytes(event.hash)[::-1].hex(), tx["txid"])
+        assert_equal(event.vsize, tx["tx"].get_vsize())
+        assert_equal(event.fee, fee)
+
+        bpf.cleanup()
         self.generate(self.wallet, 1)
 
     def removed_test(self):
@@ -181,6 +179,7 @@ class MempoolTracepointTest(BitcoinTestFramework):
 
         EXPECTED_REMOVED_EVENTS = 1
         handled_removed_events = 0
+        event = None
 
         self.log.info("Hooking into mempool:removed tracepoint...")
         node = self.nodes[0]
@@ -189,13 +188,8 @@ class MempoolTracepointTest(BitcoinTestFramework):
         bpf = BPF(text=MEMPOOL_TRACEPOINTS_PROGRAM, usdt_contexts=[ctx], debug=0)
 
         def handle_removed_event(_, data, __):
-            nonlocal handled_removed_events
+            nonlocal event, handled_removed_events
             event = bpf["removed_events"].event(data)
-            assert_equal(txid, bytes(event.hash)[::-1].hex())
-            assert_equal(reason, event.reason.decode("UTF-8"))
-            assert_equal(vsize, event.vsize)
-            assert_equal(fee, event.fee)
-            assert_equal(entry_time, event.entry_time)
             handled_removed_events += 1
 
         bpf["removed_events"].open_perf_buffer(handle_removed_event)
@@ -203,10 +197,7 @@ class MempoolTracepointTest(BitcoinTestFramework):
         self.log.info("Sending transaction...")
         fee = Decimal(31200)
         tx = self.wallet.send_self_transfer(from_node=node, fee=fee / COIN)
-        # expected data
         txid = tx["txid"]
-        reason = "expiry"
-        vsize = tx["tx"].get_vsize()
 
         self.log.info("Fast-forwarding time to mempool expiry...")
         entry_time = node.getmempoolentry(txid)["time"]
@@ -220,10 +211,15 @@ class MempoolTracepointTest(BitcoinTestFramework):
         self.log.info("Polling buffer...")
         bpf.perf_buffer_poll(timeout=200)
 
-        bpf.cleanup()
-
         self.log.info("Ensuring mempool:removed event was handled successfully...")
         assert_equal(EXPECTED_REMOVED_EVENTS, handled_removed_events)
+        assert_equal(bytes(event.hash)[::-1].hex(), txid)
+        assert_equal(event.reason.decode("UTF-8"), "expiry")
+        assert_equal(event.vsize, tx["tx"].get_vsize())
+        assert_equal(event.fee, fee)
+        assert_equal(event.entry_time, entry_time)
+
+        bpf.cleanup()
         self.generate(self.wallet, 1)
 
     def replaced_test(self):
@@ -232,6 +228,7 @@ class MempoolTracepointTest(BitcoinTestFramework):
 
         EXPECTED_REPLACED_EVENTS = 1
         handled_replaced_events = 0
+        event = None
 
         self.log.info("Hooking into mempool:replaced tracepoint...")
         node = self.nodes[0]
@@ -240,15 +237,8 @@ class MempoolTracepointTest(BitcoinTestFramework):
         bpf = BPF(text=MEMPOOL_TRACEPOINTS_PROGRAM, usdt_contexts=[ctx], debug=0)
 
         def handle_replaced_event(_, data, __):
-            nonlocal handled_replaced_events
+            nonlocal event, handled_replaced_events
             event = bpf["replaced_events"].event(data)
-            assert_equal(replaced_txid, bytes(event.replaced_hash)[::-1].hex())
-            assert_equal(replaced_vsize, event.replaced_vsize)
-            assert_equal(replaced_fee, event.replaced_fee)
-            assert_equal(replaced_entry_time, event.replaced_entry_time)
-            assert_equal(replacement_txid, bytes(event.replacement_hash)[::-1].hex())
-            assert_equal(replacement_vsize, event.replacement_vsize)
-            assert_equal(replacement_fee, event.replacement_fee)
             handled_replaced_events += 1
 
         bpf["replaced_events"].open_perf_buffer(handle_replaced_event)
@@ -267,21 +257,20 @@ class MempoolTracepointTest(BitcoinTestFramework):
             from_node=node, utxo_to_spend=utxo, fee=replacement_fee / COIN
         )
 
-        # expected data
-        replaced_txid = original_tx["txid"]
-        replaced_vsize = original_tx["tx"].get_vsize()
-        replaced_fee = original_fee
-        replaced_entry_time = entry_time
-        replacement_txid = replacement_tx["txid"]
-        replacement_vsize = replacement_tx["tx"].get_vsize()
-
         self.log.info("Polling buffer...")
         bpf.perf_buffer_poll(timeout=200)
 
-        bpf.cleanup()
-
         self.log.info("Ensuring mempool:replaced event was handled successfully...")
         assert_equal(EXPECTED_REPLACED_EVENTS, handled_replaced_events)
+        assert_equal(bytes(event.replaced_hash)[::-1].hex(), original_tx["txid"])
+        assert_equal(event.replaced_vsize, original_tx["tx"].get_vsize())
+        assert_equal(event.replaced_fee, original_fee)
+        assert_equal(event.replaced_entry_time, entry_time)
+        assert_equal(bytes(event.replacement_hash)[::-1].hex(), replacement_tx["txid"])
+        assert_equal(event.replacement_vsize, replacement_tx["tx"].get_vsize())
+        assert_equal(event.replacement_fee, replacement_fee)
+
+        bpf.cleanup()
         self.generate(self.wallet, 1)
 
     def rejected_test(self):
@@ -290,6 +279,7 @@ class MempoolTracepointTest(BitcoinTestFramework):
 
         EXPECTED_REJECTED_EVENTS = 1
         handled_rejected_events = 0
+        event = None
 
         self.log.info("Adding P2P connection...")
         node = self.nodes[0]
@@ -301,10 +291,8 @@ class MempoolTracepointTest(BitcoinTestFramework):
         bpf = BPF(text=MEMPOOL_TRACEPOINTS_PROGRAM, usdt_contexts=[ctx], debug=0)
 
         def handle_rejected_event(_, data, __):
-            nonlocal handled_rejected_events
+            nonlocal event, handled_rejected_events
             event = bpf["rejected_events"].event(data)
-            assert_equal(txid, bytes(event.hash)[::-1].hex())
-            assert_equal(reason, event.reason.decode("UTF-8"))
             handled_rejected_events += 1
 
         bpf["rejected_events"].open_perf_buffer(handle_rejected_event)
@@ -313,17 +301,15 @@ class MempoolTracepointTest(BitcoinTestFramework):
         tx = self.wallet.create_self_transfer(fee_rate=Decimal(0))
         node.p2ps[0].send_txs_and_test([tx["tx"]], node, success=False)
 
-        # expected data
-        txid = tx["tx"].hash
-        reason = "min relay fee not met"
-
         self.log.info("Polling buffer...")
         bpf.perf_buffer_poll(timeout=200)
 
-        bpf.cleanup()
-
         self.log.info("Ensuring mempool:rejected event was handled successfully...")
         assert_equal(EXPECTED_REJECTED_EVENTS, handled_rejected_events)
+        assert_equal(bytes(event.hash)[::-1].hex(), tx["tx"].hash)
+        assert_equal(event.reason.decode("UTF-8"), "min relay fee not met")
+
+        bpf.cleanup()
         self.generate(self.wallet, 1)
 
     def run_test(self):
diff --git a/test/functional/interface_usdt_net.py b/test/functional/interface_usdt_net.py
index 2235da702b..d1f94637c9 100755
--- a/test/functional/interface_usdt_net.py
+++ b/test/functional/interface_usdt_net.py
@@ -116,13 +116,10 @@ class NetTracepointTest(BitcoinTestFramework):
                          fn_name="trace_outbound_message")
         bpf = BPF(text=net_tracepoints_program, usdt_contexts=[ctx], debug=0)
 
-        # The handle_* function is a ctypes callback function called from C. When
-        # we assert in the handle_* function, the AssertError doesn't propagate
-        # back to Python. The exception is ignored. We manually count and assert
-        # that the handle_* functions succeeded.
         EXPECTED_INOUTBOUND_VERSION_MSG = 1
         checked_inbound_version_msg = 0
         checked_outbound_version_msg = 0
+        events = []
 
         def check_p2p_message(event, inbound):
             nonlocal checked_inbound_version_msg, checked_outbound_version_msg
@@ -142,12 +139,13 @@ class NetTracepointTest(BitcoinTestFramework):
                     checked_outbound_version_msg += 1
 
         def handle_inbound(_, data, __):
+            nonlocal events
             event = ctypes.cast(data, ctypes.POINTER(P2PMessage)).contents
-            check_p2p_message(event, True)
+            events.append((event, True))
 
         def handle_outbound(_, data, __):
             event = ctypes.cast(data, ctypes.POINTER(P2PMessage)).contents
-            check_p2p_message(event, False)
+            events.append((event, False))
 
         bpf["inbound_messages"].open_perf_buffer(handle_inbound)
         bpf["outbound_messages"].open_perf_buffer(handle_outbound)
@@ -158,12 +156,15 @@ class NetTracepointTest(BitcoinTestFramework):
         bpf.perf_buffer_poll(timeout=200)
 
         self.log.info(
-            "check that we got both an inbound and outbound version message")
+            "check receipt and content of in- and outbound version messages")
+        for event, inbound in events:
+            check_p2p_message(event, inbound)
         assert_equal(EXPECTED_INOUTBOUND_VERSION_MSG,
                      checked_inbound_version_msg)
         assert_equal(EXPECTED_INOUTBOUND_VERSION_MSG,
                      checked_outbound_version_msg)
 
+
         bpf.cleanup()
 
 
diff --git a/test/functional/interface_usdt_utxocache.py b/test/functional/interface_usdt_utxocache.py
index 6774db7c5f..5f2ba49026 100755
--- a/test/functional/interface_usdt_utxocache.py
+++ b/test/functional/interface_usdt_utxocache.py
@@ -188,13 +188,16 @@ class UTXOCacheTracepointTest(BitcoinTestFramework):
             nonlocal handle_uncache_succeeds
             event = ctypes.cast(data, ctypes.POINTER(UTXOCacheChange)).contents
             self.log.info(f"handle_utxocache_uncache(): {event}")
-            assert_equal(block_1_coinbase_txid, bytes(event.txid[::-1]).hex())
-            assert_equal(0, event.index)  # prevout index
-            assert_equal(EARLY_BLOCK_HEIGHT, event.height)
-            assert_equal(50 * COIN, event.value)
-            assert_equal(True, event.is_coinbase)
-
-            handle_uncache_succeeds += 1
+            try:
+                assert_equal(block_1_coinbase_txid, bytes(event.txid[::-1]).hex())
+                assert_equal(0, event.index)  # prevout index
+                assert_equal(EARLY_BLOCK_HEIGHT, event.height)
+                assert_equal(50 * COIN, event.value)
+                assert_equal(True, event.is_coinbase)
+            except AssertionError:
+                self.log.exception("Assertion failed")
+            else:
+                handle_uncache_succeeds += 1
 
         bpf["utxocache_uncache"].open_perf_buffer(handle_utxocache_uncache)
 
@@ -260,24 +263,32 @@ class UTXOCacheTracepointTest(BitcoinTestFramework):
             event = ctypes.cast(data, ctypes.POINTER(UTXOCacheChange)).contents
             self.log.info(f"handle_utxocache_add(): {event}")
             add = expected_utxocache_adds.pop(0)
-            assert_equal(add["txid"], bytes(event.txid[::-1]).hex())
-            assert_equal(add["index"], event.index)
-            assert_equal(add["height"], event.height)
-            assert_equal(add["value"], event.value)
-            assert_equal(add["is_coinbase"], event.is_coinbase)
-            handle_add_succeeds += 1
+            try:
+                assert_equal(add["txid"], bytes(event.txid[::-1]).hex())
+                assert_equal(add["index"], event.index)
+                assert_equal(add["height"], event.height)
+                assert_equal(add["value"], event.value)
+                assert_equal(add["is_coinbase"], event.is_coinbase)
+            except AssertionError:
+                self.log.exception("Assertion failed")
+            else:
+                handle_add_succeeds += 1
 
         def handle_utxocache_spent(_, data, __):
             nonlocal handle_spent_succeeds
             event = ctypes.cast(data, ctypes.POINTER(UTXOCacheChange)).contents
             self.log.info(f"handle_utxocache_spent(): {event}")
             spent = expected_utxocache_spents.pop(0)
-            assert_equal(spent["txid"], bytes(event.txid[::-1]).hex())
-            assert_equal(spent["index"], event.index)
-            assert_equal(spent["height"], event.height)
-            assert_equal(spent["value"], event.value)
-            assert_equal(spent["is_coinbase"], event.is_coinbase)
-            handle_spent_succeeds += 1
+            try:
+                assert_equal(spent["txid"], bytes(event.txid[::-1]).hex())
+                assert_equal(spent["index"], event.index)
+                assert_equal(spent["height"], event.height)
+                assert_equal(spent["value"], event.value)
+                assert_equal(spent["is_coinbase"], event.is_coinbase)
+            except AssertionError:
+                self.log.exception("Assertion failed")
+            else:
+                handle_spent_succeeds += 1
 
         bpf["utxocache_add"].open_perf_buffer(handle_utxocache_add)
         bpf["utxocache_spent"].open_perf_buffer(handle_utxocache_spent)
diff --git a/test/functional/interface_usdt_validation.py b/test/functional/interface_usdt_validation.py
index 4323aef771..f9d9b525cd 100755
--- a/test/functional/interface_usdt_validation.py
+++ b/test/functional/interface_usdt_validation.py
@@ -85,13 +85,10 @@ class ValidationTracepointTest(BitcoinTestFramework):
                     self.sigops,
                     self.duration)
 
-        # The handle_* function is a ctypes callback function called from C. When
-        # we assert in the handle_* function, the AssertError doesn't propagate
-        # back to Python. The exception is ignored. We manually count and assert
-        # that the handle_* functions succeeded.
         BLOCKS_EXPECTED = 2
         blocks_checked = 0
         expected_blocks = dict()
+        events = []
 
         self.log.info("hook into the validation:block_connected tracepoint")
         ctx = USDT(pid=self.nodes[0].process.pid)
@@ -101,19 +98,10 @@ class ValidationTracepointTest(BitcoinTestFramework):
                   usdt_contexts=[ctx], debug=0)
 
         def handle_blockconnected(_, data, __):
-            nonlocal expected_blocks, blocks_checked
+            nonlocal events, blocks_checked
             event = ctypes.cast(data, ctypes.POINTER(Block)).contents
             self.log.info(f"handle_blockconnected(): {event}")
-            block_hash = bytes(event.hash[::-1]).hex()
-            block = expected_blocks[block_hash]
-            assert_equal(block["hash"], block_hash)
-            assert_equal(block["height"], event.height)
-            assert_equal(len(block["tx"]), event.transactions)
-            assert_equal(len([tx["vin"] for tx in block["tx"]]), event.inputs)
-            assert_equal(0, event.sigops)  # no sigops in coinbase tx
-            # only plausibility checks
-            assert event.duration > 0
-            del expected_blocks[block_hash]
+            events.append(event)
             blocks_checked += 1
 
         bpf["block_connected"].open_perf_buffer(
@@ -126,12 +114,24 @@ class ValidationTracepointTest(BitcoinTestFramework):
             expected_blocks[block_hash] = self.nodes[0].getblock(block_hash, 2)
 
         bpf.perf_buffer_poll(timeout=200)
-        bpf.cleanup()
 
-        self.log.info(f"check that we traced {BLOCKS_EXPECTED} blocks")
+        self.log.info(f"check that we correctly traced {BLOCKS_EXPECTED} blocks")
+        for event in events:
+            block_hash = bytes(event.hash[::-1]).hex()
+            block = expected_blocks[block_hash]
+            assert_equal(block["hash"], block_hash)
+            assert_equal(block["height"], event.height)
+            assert_equal(len(block["tx"]), event.transactions)
+            assert_equal(len([tx["vin"] for tx in block["tx"]]), event.inputs)
+            assert_equal(0, event.sigops)  # no sigops in coinbase tx
+            # only plausibility checks
+            assert event.duration > 0
+            del expected_blocks[block_hash]
         assert_equal(BLOCKS_EXPECTED, blocks_checked)
         assert_equal(0, len(expected_blocks))
 
+        bpf.cleanup()
+
 
 if __name__ == '__main__':
     ValidationTracepointTest().main()