PYCBC-1813: Fix async txn callback leaks and add unraisable

Changes
--------
* Add complete_pending_operation() as the shared tail of every async
  txn completion path, folding in handle_returning_void,
  handle_returning_transaction_get_result,
  handle_returning_transaction_get_multi_result,
  handle_returning_query_result, and transaction_commit's inline
  completion lambda; report a raised callback/errback via
  PyErr_WriteUnraisable instead of dropping it silently
* Fix transaction_op's 7 early-return paths and its default: case to
  Py_XDECREF the Py_XINCREF'd callback/errback before returning; the
  default: case also now returns instead of falling through, which
  previously hung the sync path on fut.get() waiting on a promise
  nothing would set
* Fix the same leak in transaction_get_multi_op's analogous
  unrecognized-op_type fallback
* Add PyErr_WriteUnraisable at the 5 remaining connection.cxx/
  connection.hxx sites where a raised user callback/errback was
  discarded unreported

Change-Id: I416d7e4319f41b338b7e335853734d8942a7d7d7
Reviewed-on: https://review.couchbase.org/c/couchbase-python-client/+/250600
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Dimitris Christodoulou <dimitris.christodoulou@couchbase.com>

Author: thejcfactor

src/connection.cxx

@@ -113,6 +113,10 @@ Connection::handle_connection_operation_callback(std::error_code ec,
       }
     } else if (pyObj_errback != nullptr) {
       PyObject* ret = PyObject_CallFunctionObjArgs(pyObj_errback, result, nullptr);
+      if (ret == nullptr) {
+        // errback raised; nothing else observes this IO thread's exception state.
+        PyErr_WriteUnraisable(pyObj_errback);
+      }
       Py_XDECREF(ret);
       Py_XDECREF(result);
     } else if (barrier) {
@@ -132,6 +136,10 @@ Connection::handle_connection_operation_callback(std::error_code ec,
 
     if (pyObj_callback != nullptr) {
       PyObject* ret = PyObject_CallFunctionObjArgs(pyObj_callback, result, nullptr);
+      if (ret == nullptr) {
+        // callback raised; nothing else observes this IO thread's exception state.
+        PyErr_WriteUnraisable(pyObj_callback);
+      }
       Py_XDECREF(ret);
       Py_XDECREF(result);
     } else if (barrier) {

src/connection.hxx

@@ -142,6 +142,10 @@ private:
       }
     } else if (pyObj_callback != nullptr) {
       PyObject* ret = PyObject_CallFunctionObjArgs(pyObj_callback, result, nullptr);
+      if (ret == nullptr) {
+        // callback raised; nothing else observes this IO thread's exception state.
+        PyErr_WriteUnraisable(pyObj_callback);
+      }
       Py_XDECREF(ret);
       Py_XDECREF(result);
     } else if (barrier) {
@@ -265,7 +269,12 @@ private:
             PyObject_TypeCheck(result, &pycbc_exception_type) ? pyObj_errback : pyObj_callback;
 
           if (target_handler != nullptr) {
-            Py_XDECREF(PyObject_CallFunctionObjArgs(target_handler, result, nullptr));
+            PyObject* ret = PyObject_CallFunctionObjArgs(target_handler, result, nullptr);
+            if (ret == nullptr) {
+              // callback/errback raised; nothing else observes this IO thread's exception state.
+              PyErr_WriteUnraisable(target_handler);
+            }
+            Py_XDECREF(ret);
           } else if (barrier != nullptr) {
             barrier->set_value(result);
             result = nullptr; // Reference transferred to barrier
@@ -703,6 +712,10 @@ Connection::execute_streaming_op(PyObject* kwargs)
           // The python side will determine if the query has rows or returned an error
           if (callback != nullptr) {
             PyObject* result = PyObject_CallFunction(callback, "O", Py_True);
+            if (result == nullptr) {
+              // callback raised; nothing else observes this IO thread's exception state.
+              PyErr_WriteUnraisable(callback);
+            }
             Py_XDECREF(result);
           }