JSCBC-1522: Close and destroy transactions objects off the JS thread

Changes
-------
* Run the core's transactions close() from Transactions::jsClose on a
  detached thread and call back through a CallCookie, so removing the
  client record from each bucket no longer blocks the event loop;
  close inline, as before, when no thread can be started
* Instance now owns every core transactions object created on its
  connection; the Transactions and Transaction wrappers hold aliasing
  pointers that keep the instance alive instead, so collecting a
  Transactions wrapper never frees an object a live transaction still
  uses, and no finalizer runs the core's destructor on the JS thread
* Close each object through Instance::closeTransactions, which holds a
  lock for the whole close, so an explicit close and the connection's
  shutdown are serialized and close it once (two concurrent closes end
  the process)
* Instance::shutdown closes every transactions object still open
  before the cluster, and destroys them while the cluster is open, or
  once the IO thread has stopped when an operation is still in flight
  and so still uses its object
* Count a transaction operation from its start until its callback is
  invoked or its handler is destroyed, through a TransactionOp shared
  by every copy of its RefCallCookie
* Document on Transactions.run that an operation in flight at a worker
  terminate or process exit is cancelled, and lost transaction cleanup
  decides the outcome
* Add a vars option to expectProcessExits for passing extra
  environment variables to an exit fixture
* Add tests that the close callback runs after the call returns, that
  two concurrent closes both resolve, and that terminating a worker
  while its cluster is closing still exits
* Add a test that commits after the Transactions wrapper that started
  the transaction has been collected
* Add a test that terminates workers with transactions still open, and
  after collecting their connection, and checks that each client is
  removed from the client record
* Add test/suites/transactions/teardownstall.test.ts, which stalls a
  worker's IO thread, with a commit queued behind the stall and with
  its transactions closed, and checks each terminate returns at its
  bound

Change-Id: I2c7d8a4deb1098d805f1b7e902177ad70b747add
Reviewed-on: https://review.couchbase.org/c/couchnode/+/256439
Reviewed-by: Mateusz <matt.wozakowski@couchbase.com>
Tested-by: Jared Casey <jared.casey@couchbase.com>

Author: thejcfactor

lib/transactions.ts

@@ -1271,6 +1271,12 @@ export class Transactions {
   /**
    * Executes a transaction.
    *
+   * When the worker thread running a transaction is terminated, or the process
+   * exits, while one of its operations is in flight, that operation is
+   * cancelled. The outcome is then decided by lost transaction cleanup rather
+   * than by the caller, so a transaction past its commit point may still commit
+   * after `terminate()` has returned.
+   *
    * @param logicFn The transaction lambda to execute.
    * @param config Configuration operations for the transaction.
    */

src/connection.hpp

@@ -133,6 +133,11 @@ class Connection : public Napi::ObjectWrap<Connection>
         return _instance->_cluster;
     }
 
+    const std::shared_ptr<Instance> &instance() const
+    {
+        return _instance;
+    }
+
     std::pair<std::optional<std::string>, std::optional<std::string>>
     getClusterLabels();
 

src/instance.cpp

@@ -1,12 +1,34 @@
 #include "instance.hpp"
+#include <algorithm>
+#include <core/transactions.hxx>
 #include <future>
 #include <system_error>
 
 namespace couchnode
 {
 
+TransactionOp::TransactionOp(std::shared_ptr<Count> count)
+    : _count(std::move(count))
+{
+    ++*_count;
+}
+
+TransactionOp::~TransactionOp()
+{
+    complete();
+}
+
+void TransactionOp::complete()
+{
+    if (_completed.exchange(true)) {
+        return;
+    }
+    --*_count;
+}
+
 Instance::Instance()
     : _cluster(couchbase::core::cluster(_io))
+    , _transactionOps(std::make_shared<TransactionOp::Count>(0))
 {
     _ioThread = std::thread([this]() {
         try {
@@ -25,6 +47,58 @@ Instance::~Instance()
 {
 }
 
+std::shared_ptr<cbcoretxns::transactions> Instance::addTransactions(
+    std::shared_ptr<cbcoretxns::transactions> transactions)
+{
+    std::lock_guard<std::mutex> lock(_transactionsMutex);
+    auto *raw = transactions.get();
+    _transactions.push_back({std::move(transactions)});
+    return std::shared_ptr<cbcoretxns::transactions>(shared_from_this(), raw);
+}
+
+void Instance::closeTransactions(const cbcoretxns::transactions *transactions)
+{
+    std::lock_guard<std::mutex> lock(_transactionsMutex);
+    auto owned = std::find_if(
+        _transactions.begin(), _transactions.end(),
+        [transactions](const auto &o) { return o.impl.get() == transactions; });
+    if (owned == _transactions.end() || owned->closed) {
+        return;
+    }
+    owned->impl->close();
+    owned->closed = true;
+}
+
+std::shared_ptr<TransactionOp> Instance::beginTransactionOp()
+{
+    return std::make_shared<TransactionOp>(_transactionOps);
+}
+
+bool Instance::transactionOpsInFlight() const
+{
+    return *_transactionOps != 0;
+}
+
+void Instance::closeAllTransactions()
+{
+    std::lock_guard<std::mutex> lock(_transactionsMutex);
+    for (auto &owned : _transactions) {
+        if (!owned.closed) {
+            owned.impl->close();
+            owned.closed = true;
+        }
+    }
+}
+
+void Instance::destroyTransactions()
+{
+    std::lock_guard<std::mutex> lock(_transactionsMutex);
+    // The core's destructor closes each object again. That repeats the client
+    // record removal: a round trip while the cluster is open, and retries
+    // against `cluster_closed` for up to 500 ms per keyspace once it is not.
+    _transactions.clear();
+}
+
 void Instance::shutdown(std::function<void()> done)
 {
     std::call_once(_shutdownOnce, [this, &done]() {
@@ -34,11 +108,20 @@ void Instance::shutdown(std::function<void()> done)
         // reference keeps the last release off the IO thread.
         try {
             std::thread([self = shared_from_this(), done]() mutable {
+                self->closeAllTransactions();
+                // An operation still in flight uses its transactions object, so
+                // that is destroyed once the IO thread has stopped instead. The
+                // cluster close cancels the operation, and a transaction past its
+                // commit point is then settled by lost transaction cleanup.
+                if (!self->transactionOpsInFlight()) {
+                    self->destroyTransactions();
+                }
                 auto closed = std::make_shared<std::promise<void>>();
                 auto closing = closed->get_future();
                 self->_cluster.close([closed]() { closed->set_value(); });
                 closing.wait();
                 self->_ioThread.join();
+                self->destroyTransactions();
                 // Released before `done`, so a caller that waits on `done` and
                 // holds its own reference destroys the instance itself.
                 self.reset();

src/instance.hpp

@@ -1,15 +1,47 @@
 #pragma once
 #include <asio/io_context.hpp>
+#include <atomic>
 #include <core/cluster.hxx>
 #include <core/logger/logger.hxx>
+#include <cstddef>
 #include <functional>
 #include <memory>
 #include <mutex>
 #include <thread>
+#include <vector>
+
+namespace couchbase::core::transactions
+{
+class transactions;
+} // namespace couchbase::core::transactions
+namespace cbcoretxns = couchbase::core::transactions;
 
 namespace couchnode
 {
 
+// One transaction operation in flight. It counts from when the operation
+// starts until it completes or its handler is destroyed without completing it,
+// whichever comes first, so a handler the core keeps after calling it does not
+// count as in flight for as long as the core keeps it.
+class TransactionOp
+{
+public:
+    using Count = std::atomic<std::size_t>;
+
+    explicit TransactionOp(std::shared_ptr<Count> count);
+    ~TransactionOp();
+
+    TransactionOp(const TransactionOp &) = delete;
+    TransactionOp &operator=(const TransactionOp &) = delete;
+
+    // Stops counting the operation. Only the first call does anything.
+    void complete();
+
+private:
+    std::shared_ptr<Count> _count;
+    std::atomic<bool> _completed{false};
+};
+
 // Owned through `std::shared_ptr`. The last reference must never be dropped on
 // the IO thread, since destroying the instance destroys the IO context that
 // thread is running.
@@ -19,19 +51,54 @@ class Instance : public std::enable_shared_from_this<Instance>
     Instance();
     ~Instance();
 
-    // Closes the cluster and joins the IO thread, on a helper thread that holds
-    // a reference until the join returns. Only the first call does anything;
-    // `done` runs on the helper thread after the join and may be empty. When no
-    // helper thread can be started, the close is abandoned, the instance is
-    // leaked, and `done` runs before this returns.
+    // Takes ownership of a transactions object and returns a pointer to it
+    // that keeps this instance alive instead. The instance destroys the object
+    // in `shutdown`, on its helper thread, so a wrapper's finalizer never runs
+    // the core's destructor on the JS thread. That destructor closes the
+    // object again, which waits on the cluster.
+    std::shared_ptr<cbcoretxns::transactions>
+    addTransactions(std::shared_ptr<cbcoretxns::transactions> transactions);
+
+    // Closes one transactions object, unless it was already closed. Waits on
+    // cluster requests, so it belongs on a thread of its own. Holds a lock for
+    // the whole close, since the core's close cannot run concurrently and
+    // `shutdown` must not close the cluster under it.
+    void closeTransactions(const cbcoretxns::transactions *transactions);
+
+    // Counts a transaction operation as in flight until the returned
+    // operation completes or is destroyed.
+    std::shared_ptr<TransactionOp> beginTransactionOp();
+
+    // On a helper thread that holds a reference until it is done: closes the
+    // transactions objects, then the cluster, and joins the IO thread. Only
+    // the first call does anything; `done` runs on the helper thread after the
+    // join and may be empty. When no helper thread can be started, the close
+    // is abandoned, the instance is leaked, and `done` runs before this
+    // returns.
+    //
+    // It runs once nothing on the JS side can reach this instance's objects:
+    // at environment teardown, or when the connection is collected, which an
+    // operation's pending callback prevents.
     void shutdown(std::function<void()> done);
 
     asio::io_context _io;
     std::thread _ioThread;
     couchbase::core::cluster _cluster;
 
 private:
+    struct OwnedTransactions {
+        std::shared_ptr<cbcoretxns::transactions> impl;
+        bool closed{false};
+    };
+
+    bool transactionOpsInFlight() const;
+    void closeAllTransactions();
+    void destroyTransactions();
+
     std::once_flag _shutdownOnce;
+    std::mutex _transactionsMutex;
+    std::vector<OwnedTransactions> _transactions;
+    std::shared_ptr<TransactionOp::Count> _transactionOps;
 };
 
 } // namespace couchnode