Detect and recover from invalid database environment.

Sometimes wrong databases are returned for the name, probably related
to threading/incorrect usage of lmdb.
For the time being we recover from that by detecting it and retrying.

3 files changed, 57 insertions, 4 deletions

diff --git a/common/queryrunner.cpp b/common/queryrunner.cpp
index ea17176..fd966a9 100644
--- a/common/queryrunner.cpp
+++ b/common/queryrunner.cpp
@@ -408,9 +408,19 @@ QPair<qint64, qint64> QueryWorker<DomainType>::load(const Sink::Query &query, co
     QTime time;
     time.start();
 
-    Sink::Storage storage(Sink::storageLocation(), mResourceInstanceIdentifier);
-    storage.setDefaultErrorHandler([](const Sink::Storage::Error &error) { Warning() << "Error during query: " << error.store << error.message; });
-    auto transaction = storage.createTransaction(Sink::Storage::ReadOnly);
+    Sink::Storage::Transaction transaction;
+    {
+        Sink::Storage storage(Sink::storageLocation(), mResourceInstanceIdentifier);
+        storage.setDefaultErrorHandler([](const Sink::Storage::Error &error) { Warning() << "Error during query: " << error.store << error.message; });
+        transaction = storage.createTransaction(Sink::Storage::ReadOnly);
+    }
+
+    //FIXME this is a temporary measure to recover from a failure to open the named databases correctly.
+    //Once the actual problem is fixed it will be enough to simply crash if we open the wrong database (which we check in openDatabase already).
+    while (!transaction.validateNamedDatabases()) {
+        Sink::Storage storage(Sink::storageLocation(), mResourceInstanceIdentifier);
+        transaction = storage.createTransaction(Sink::Storage::ReadOnly);
+    }
     auto db = Storage::mainDatabase(transaction, mBufferType);
 
     QSet<QByteArray> remainingFilters;
diff --git a/common/storage.h b/common/storage.h
index 87573e2..0527c4f 100644
--- a/common/storage.h
+++ b/common/storage.h
@@ -141,6 +141,7 @@ public:
         void abort();
 
         QList<QByteArray> getDatabaseNames() const;
+        bool validateNamedDatabases();
 
         NamedDatabase openDatabase(const QByteArray &name = QByteArray("default"),
             const std::function<void(const Storage::Error &error)> &errorHandler = std::function<void(const Storage::Error &error)>(), bool allowDuplicates = false) const;
diff --git a/common/storage_lmdb.cpp b/common/storage_lmdb.cpp
index cc8b28d..ae4bfba 100644
--- a/common/storage_lmdb.cpp
+++ b/common/storage_lmdb.cpp
@@ -437,6 +437,43 @@ void Storage::Transaction::abort()
     d->transaction = nullptr;
 }
 
+//Ensure that we opened the correct database by comparing the expected identifier with the one
+//we write to the database on first open.
+static bool ensureCorrectDb(Storage::NamedDatabase &database, const QByteArray &db, bool readOnly)
+{
+    bool openedTheWrongDatabase = false;
+    auto count = database.scan("__internal_dbname", [db, &openedTheWrongDatabase](const QByteArray &key, const QByteArray &value) ->bool {
+        if (value != db) {
+            Warning() << "Opened the wrong database, got " << value << " instead of " << db;
+            openedTheWrongDatabase = true;
+        }
+        return false;
+    },
+    [](const Storage::Error &error) -> bool{
+        return false;
+    }, false);
+    //This is the first time we open this database in a write transaction, write the db name
+    if (!count) {
+        if (!readOnly) {
+            database.write("__internal_dbname", db);
+        }
+    }
+    return !openedTheWrongDatabase;
+}
+
+bool Storage::Transaction::validateNamedDatabases()
+{
+    auto databases = getDatabaseNames();
+    for (const auto &dbName : databases) {
+        auto db = openDatabase(dbName);
+        if (!db) {
+            Warning() << "Failed to open the database: " << dbName;
+            return false;
+        }
+    }
+    return true;
+}
+
 Storage::NamedDatabase Storage::Transaction::openDatabase(const QByteArray &db, const std::function<void(const Storage::Error &error)> &errorHandler, bool allowDuplicates) const
 {
     if (!d) {
@@ -450,7 +487,12 @@ Storage::NamedDatabase Storage::Transaction::openDatabase(const QByteArray &db,
         delete p;
         return Storage::NamedDatabase();
     }
-    return Storage::NamedDatabase(p);
+    auto database = Storage::NamedDatabase(p);
+    if (!ensureCorrectDb(database, db, d->requestedRead)) {
+        Warning() << "Failed to open the database";
+        return Storage::NamedDatabase();
+    }
+    return database;
 }
 
 QList<QByteArray> Storage::Transaction::getDatabaseNames() const