Workaround for "Duplicate RW transaction" LMDB exception.

Although LMDB supports concurrent RO and RW transactions (but no more
than one writer), the existing lmdb access code here seems to be written
assuming an LMDB RW transaction is not allowed to coexist with any other
transaction, even RO ones.

The logic to decide whether a transaction would be RO or RW was relying
upon the way the database was opened at mdb_dbi_open() time, which turns
out to be (almost always) read-write, despite the kind of the upcoming
requests.

We now pass a boolean further in MDBEnv::openDB() to specify the intent
of the operation, rather than trusting the shared environment only.

This allows us to build more MDBROTransaction objects where
MDBRWTransaction objects would have been used previously The existing
test would raise an exception, should there be existing RO transactions.

ext/lmdb-safe/lmdb-safe.cc

@@ -201,7 +201,7 @@ std::shared_ptr<MDBEnv> getMDBEnv(const char* fname, int flags, int mode, uint64
 }
 
 
-MDBDbi MDBEnv::openDB(const string_view dbname, int flags)
+MDBDbi MDBEnv::openDB(const string_view dbname, int flags, bool readonly)
 {
   unsigned int envflags;
   mdb_env_get_flags(d_env, &envflags);
@@ -210,7 +210,14 @@ MDBDbi MDBEnv::openDB(const string_view dbname, int flags)
   */
   std::lock_guard<std::mutex> l(d_openmut);
 
-  if(!(envflags & MDB_RDONLY)) {
+  // For a variety of reasons, such as the need to automagically convert older
+  // databases, the mdb environment may not have MDB_RDONLY set. But if our
+  // caller will only perform read-only transactions, give it a
+  // MDBROTransaction.
+  // If we don't do this, recursive lookups may end up triggering the
+  // "a writer can not exist with anyone else" exception in
+  // MDBRWTransactionImpl::openRWTransaction() below.
+  if(!(envflags & MDB_RDONLY) && !readonly) {
     auto rwt = getRWTransaction();
     MDBDbi ret = rwt->openDB(dbname, flags);
     rwt->commit();
@@ -235,8 +242,9 @@ MDBRWTransactionImpl::MDBRWTransactionImpl(MDBEnv *parent, MDB_txn *txn):
 MDB_txn *MDBRWTransactionImpl::openRWTransaction(MDBEnv *env, MDB_txn *parent, int flags)
 {
   MDB_txn *result;
-  if(env->getROTX() || env->getRWTX())
-    throw std::runtime_error("Duplicate RW transaction");
+  if(env->getROTX() != 0 || env->getRWTX() != 0) {
+    throw std::runtime_error("RW transaction while other transactions are active");
+  }
 
   if(int rc=mdb_txn_begin(env->d_env, parent, flags, &result))
     throw std::runtime_error("Unable to start RW transaction: "+std::string(mdb_strerror(rc)));

ext/lmdb-safe/lmdb-safe.hh

@@ -77,7 +77,7 @@ public:
     // but, elsewhere, docs say database handles do not need to be closed?
   }
 
-  MDBDbi openDB(string_view dbname, int flags);
+  MDBDbi openDB(string_view dbname, int flags, bool readonly = false);
 
   MDBRWTransaction getRWTransaction();
   MDBROTransaction getROTransaction();

modules/lmdbbackend/lmdbbackend.cc

@@ -743,7 +743,7 @@ LMDBBackend::LMDBBackend(const std::string& suffix)
       d_tkdb = std::make_shared<tkdb_t>(d_tdomains->getEnv(), "keydata_v5");
       d_ttsig = std::make_shared<ttsig_t>(d_tdomains->getEnv(), "tsig_v5");
 
-      auto pdnsdbi = d_tdomains->getEnv()->openDB("pdns", MDB_CREATE);
+      auto pdnsdbi = d_tdomains->getEnv()->openDB("pdns", MDB_CREATE, false);
 
       opened = true;
 
@@ -1255,7 +1255,7 @@ std::shared_ptr<LMDBBackend::RecordsRWTransaction> LMDBBackend::getRecordsRWTran
   if (!shard.env) {
     shard.env = getMDBEnv((getArg("filename") + "-" + std::to_string(id % s_shards)).c_str(),
                           MDB_NOSUBDIR | d_asyncFlag, 0600);
-    shard.dbi = shard.env->openDB("records_v5", MDB_CREATE);
+    shard.dbi = shard.env->openDB("records_v5", MDB_CREATE, false);
   }
   auto ret = std::make_shared<RecordsRWTransaction>(shard.env->getRWTransaction());
   ret->db = std::make_shared<RecordsDB>(shard);
@@ -1272,7 +1272,7 @@ std::shared_ptr<LMDBBackend::RecordsROTransaction> LMDBBackend::getRecordsROTran
     }
     shard.env = getMDBEnv((getArg("filename") + "-" + std::to_string(id % s_shards)).c_str(),
                           MDB_NOSUBDIR | d_asyncFlag, 0600);
-    shard.dbi = shard.env->openDB("records_v5", MDB_CREATE);
+    shard.dbi = shard.env->openDB("records_v5", MDB_CREATE, true);
   }
 
   if (rwtxn) {