From ea3ae0693e1290cdf33eaf6e0948397c617efe65 Mon Sep 17 00:00:00 2001
From: Graham Christensen <graham@grahamc.com>
Date: Thu, 26 Aug 2021 21:53:51 -0400
Subject: [PATCH] Hook up the retryable tasks with hydra-notify

---
 src/lib/Hydra/Schema/ResultSet/TaskRetries.pm |  1 +
 src/lib/Hydra/TaskDispatcher.pm               |  2 +-
 src/script/hydra-notify                       | 10 +++++++++-
 3 files changed, 11 insertions(+), 2 deletions(-)

diff --git a/src/lib/Hydra/Schema/ResultSet/TaskRetries.pm b/src/lib/Hydra/Schema/ResultSet/TaskRetries.pm
index 2284c64e..10d89cd8 100644
--- a/src/lib/Hydra/Schema/ResultSet/TaskRetries.pm
+++ b/src/lib/Hydra/Schema/ResultSet/TaskRetries.pm
@@ -6,6 +6,7 @@ use utf8;
 use base 'DBIx::Class::ResultSet';
 use List::Util qw(max);
 use Hydra::Math qw(exponential_backoff);
+use Hydra::Task;
 
 =head2 get_seconds_to_next_retry
 
diff --git a/src/lib/Hydra/TaskDispatcher.pm b/src/lib/Hydra/TaskDispatcher.pm
index 7c472b20..310fe6d6 100644
--- a/src/lib/Hydra/TaskDispatcher.pm
+++ b/src/lib/Hydra/TaskDispatcher.pm
@@ -254,7 +254,7 @@ sub failure {
     my $event_labels = $self->prom_labels_for_task($task);
 
     if (defined($task->{"record"})) {
-        if ($task->{"record"}->{"attempts"} > 100) {
+        if ($task->{"record"}->attempts > 100) {
             $self->{"prometheus"}->inc("notify_plugin_drop", $event_labels);
             $task->{"record"}->delete();
         } else {
diff --git a/src/script/hydra-notify b/src/script/hydra-notify
index de8f64f2..3459cf2c 100755
--- a/src/script/hydra-notify
+++ b/src/script/hydra-notify
@@ -101,10 +101,12 @@ for my $build ($db->resultset('Builds')->search(
     $task_dispatcher->dispatch_event($event);
 }
 
+my $taskretries = $db->resultset('TaskRetries');
+
 # Process incoming notifications.
 while (!$queued_only) {
     $prom->inc("event_loop_iterations");
-    my $messages = $listener->block_for_messages();
+    my $messages = $listener->block_for_messages($taskretries->get_seconds_to_next_retry());
     while (my $message = $messages->()) {
         $prom->set("event_received", time());
         my $channelName = $message->{"channel"};
@@ -128,4 +130,10 @@ while (!$queued_only) {
             print STDERR "error processing message '$payload' on channel '$channelName': $@\n";
         }
     }
+
+    my $task = $taskretries->getRetryableTask();
+    if (defined($task)) {
+        $task_dispatcher->dispatchTask($task);
+    }
+
 }