Fixed crashes due to concurrently running queries.

A single QueryRunner should never have multiple workers running at the
same time. We did not properly enforce this in case of incremental
updates coming in.

The only way I managed to reproduce the crash:
* Open a large folder with lots of unread mail in kube
* Select a mail in the maillist and hold the down button
* This will:
    * Repeatedly call fetch more
    * Trigger lot's of mark as read modifications that result in
    notifications.
    * Eventually it crashes somewhere in EntityStore, likely because
    of concurrent access of the filter structure which is shared through
    the state.

We now ensure in the single threaded portion of the code that we only
ever run one worker at a time. If we did receive an update during,
we remember that change and fetch more once we're done.

To be able to call fetch again that portion was also factored out into a
separate function.

1 files changed, 75 insertions, 9 deletions

diff --git a/sinksh/syntax_modules/sink_selftest.cpp b/sinksh/syntax_modules/sink_selftest.cpp
index 21dfbff..8ad4f60 100644
--- a/sinksh/syntax_modules/sink_selftest.cpp
+++ b/sinksh/syntax_modules/sink_selftest.cpp
@@ -46,28 +46,94 @@ bool selfTest(const QStringList &args_, State &state)
         return false;
     }
 
+    using namespace Sink::ApplicationDomain;
     auto options = SyntaxTree::parseOptions(args_);
     if (options.positionalArguments.contains("stresstest")) {
         auto resource = SinkshUtils::parseUid(options.options.value("resource").first().toUtf8());
         qWarning() << "Stresstest on resource: " << resource;
-        Sink::Query query;
-        query.resourceFilter(resource);
-        query.limit(100);
-
         auto models = QSharedPointer<QList<QSharedPointer<QAbstractItemModel>>>::create();
-        for (int i = 0; i < 50; i++) {
+
+        //Simulate the maillist, where we scroll down and trigger lots of fetchMore calls
+        {
+            Sink::Query query;
+            query.resourceFilter(resource);
+            query.limit(100);
+            query.request<Mail::Subject>();
+            query.request<Mail::Sender>();
+            query.request<Mail::To>();
+            query.request<Mail::Cc>();
+            query.request<Mail::Bcc>();
+            query.request<Mail::Date>();
+            query.request<Mail::Unread>();
+            query.request<Mail::Important>();
+            query.request<Mail::Draft>();
+            query.request<Mail::Sent>();
+            query.request<Mail::Trash>();
+            query.request<Mail::Folder>();
+            query.sort<Mail::Date>();
+            query.reduce<Mail::ThreadId>(Sink::Query::Reduce::Selector::max<Mail::Date>())
+                .count("count")
+                .collect<Mail::Unread>("unreadCollected")
+                .collect<Mail::Important>("importantCollected");
+
             auto model = Sink::Store::loadModel<Sink::ApplicationDomain::Mail>(query);
-            *models << model;
+            models->append(model);
             QObject::connect(model.data(), &QAbstractItemModel::dataChanged, [models, model, &state](const QModelIndex &start, const QModelIndex &end, const QVector<int> &roles) {
                 if (roles.contains(Sink::Store::ChildrenFetchedRole)) {
-                    models->removeAll(model);
-                    qWarning() << "Model complete: " << models->count();
+                    if (!model->canFetchMore({})) {
+                        qWarning() << "Model complete: " << models->count();
+                        models->removeAll(model);
+                    } else {
+                        qWarning() << "Fetching more";
+                        //Simulate superfluous fetchMore calls
+                        for (int i = 0; i < 10; i++) {
+                            model->fetchMore({});
+                        }
+                        return;
+                    }
                     if (models->isEmpty()) {
                         state.commandFinished();
                     }
                 }
-                    });
+            });
+
         }
+
+        //Simluate lot's of mailviewers doing a bunch of queries
+        {
+            Sink::Query query;
+            query.resourceFilter(resource);
+            query.limit(10);
+            query.request<Mail::Subject>();
+            query.request<Mail::Sender>();
+            query.request<Mail::To>();
+            query.request<Mail::Cc>();
+            query.request<Mail::Bcc>();
+            query.request<Mail::Date>();
+            query.request<Mail::Unread>();
+            query.request<Mail::Important>();
+            query.request<Mail::Draft>();
+            query.request<Mail::Sent>();
+            query.request<Mail::Trash>();
+            query.request<Mail::Folder>();
+            query.sort<Sink::ApplicationDomain::Mail::Date>();
+            query.bloom<Sink::ApplicationDomain::Mail::ThreadId>();
+
+            for (int i = 0; i < 50; i++) {
+                auto model = Sink::Store::loadModel<Sink::ApplicationDomain::Mail>(query);
+                *models << model;
+                QObject::connect(model.data(), &QAbstractItemModel::dataChanged, [models, model, &state](const QModelIndex &start, const QModelIndex &end, const QVector<int> &roles) {
+                    if (roles.contains(Sink::Store::ChildrenFetchedRole)) {
+                        models->removeAll(model);
+                        qWarning() << "Model complete: " << models->count();
+                        if (models->isEmpty()) {
+                            state.commandFinished();
+                        }
+                    }
+                        });
+            }
+        }
+
         return true;
     }
     return false;