Fix a race that can cause hydra-queue-runner to ignore newly added builds

As @dtzWill discovered, with the concurrent hydra-evaluator, there can
be multiple active transactions adding builds to the database. As a
result, builds can become visible in a non-monotonically increasing
order, breaking the queue monitor's assumption that build IDs only go
up.

The fix is to have hydra-eval-jobset provide the lowest build ID it
just added in the builds_added notification, and have the queue
monitor check from there.

Fixes #496.

src/hydra-queue-runner/queue-monitor.cc

@@ -48,8 +48,10 @@ void State::queueMonitorLoop()
         } else
             conn->get_notifs();
 
-        if (buildsAdded.get())
+        if (auto lowestId = buildsAdded.get()) {
+            lastBuildId = std::stoi(*lowestId) - 1;
             printMsg(lvlTalkative, "got notification: new builds added to the queue");
+        }
         if (buildsRestarted.get()) {
             printMsg(lvlTalkative, "got notification: builds restarted");
             lastBuildId = 0; // check all builds
@@ -75,7 +77,7 @@ struct PreviousFailure : public std::exception {
 bool State::getQueuedBuilds(Connection & conn,
     ref<Store> destStore, unsigned int & lastBuildId)
 {
-    printMsg(lvlInfo, format("checking the queue for builds > %1%...") % lastBuildId);
+    printInfo("checking the queue for builds > %d...", lastBuildId);
 
     /* Grab the queued builds from the database, but don't process
        them yet (since we don't want a long-running transaction). */

src/libhydra/db.hh

@@ -21,18 +21,23 @@ struct Connection : pqxx::connection
 };
 
 
-struct receiver : public pqxx::notification_receiver
+class receiver : public pqxx::notification_receiver
 {
-    bool status = false;
+    std::experimental::optional<std::string> status;
+
+public:
+
     receiver(pqxx::connection_base & c, const std::string & channel)
         : pqxx::notification_receiver(c, channel) { }
+
     void operator() (const std::string & payload, int pid) override
     {
-        status = true;
+        status = payload;
     };
-    bool get() {
-        bool b = status;
-        status = false;
-        return b;
+
+    std::experimental::optional<std::string> get() {
+        auto s = status;
+        status = std::experimental::nullopt;
+        return s;
     }
 };

src/script/hydra-eval-jobset

@@ -25,6 +25,8 @@ STDERR->autoflush(1);
 binmode STDERR, ":encoding(utf8)";
 
 my $db = Hydra::Model::DB->new();
+my $notifyAdded = $db->storage->dbh->prepare("notify builds_added, ?");
+
 my $config = getHydraConfig();
 
 my $plugins = [Hydra::Plugin->instantiate(db => $db, config => $config)];
@@ -720,6 +722,14 @@ sub checkJobsetWrapped {
 
             print STDERR "  created new eval ", $ev->id, "\n";
             $ev->builds->update({iscurrent => 1});
+
+            # Wake up hydra-queue-runner.
+            my $lowestId;
+            while (my ($id, $x) = each %buildMap) {
+                $lowestId = $id if $x->{new} && (!defined $lowestId || $id < $lowestId);
+            }
+            $notifyAdded->execute($lowestId) if defined $lowestId;
+
         } else {
             print STDERR "  created cached eval ", $ev->id, "\n";
             $prevEval->builds->update({iscurrent => 1}) if defined $prevEval;

src/sql/hydra.sql

@@ -233,9 +233,6 @@ create table Builds (
 
 #ifdef POSTGRESQL
 
-create function notifyBuildsAdded() returns trigger as 'begin notify builds_added; return null; end;' language plpgsql;
-create trigger BuildsAdded after insert on Builds execute procedure notifyBuildsAdded();
-
 create function notifyBuildsDeleted() returns trigger as 'begin notify builds_deleted; return null; end;' language plpgsql;
 create trigger BuildsDeleted after delete on Builds execute procedure notifyBuildsDeleted();
 

src/sql/upgrade-54.sql

@@ -0,0 +1,2 @@
+drop trigger BuildsAdded on Builds;
+drop function notifyBuildsAdded();