Add support for incrementally tracking blocked builds

This will hopefully provide a less expensive way of finding out if a scheduled
build is probably blocked by other builds failing or being canceled.

By working this out when the build events are recieved, it should be more
feasible to include information about whether builds are likely blocked or not
in various places (e.g. revision comparisons).

2 files changed, 408 insertions, 54 deletions

diff --git a/guix-data-service/model/blocked-builds.scm b/guix-data-service/model/blocked-builds.scm
new file mode 100644
index 0000000..bde410f
--- /dev/null
+++ b/guix-data-service/model/blocked-builds.scm
@@ -0,0 +1,303 @@
+;;; Guix Data Service -- Information about Guix over time
+;;; Copyright © 2022 Christopher Baines <mail@cbaines.net>
+;;;
+;;; This program is free software: you can redistribute it and/or
+;;; modify it under the terms of the GNU Affero General Public License
+;;; as published by the Free Software Foundation, either version 3 of
+;;; the License, or (at your option) any later version.
+;;;
+;;; This program is distributed in the hope that it will be useful,
+;;; but WITHOUT ANY WARRANTY; without even the implied warranty of
+;;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+;;; Affero General Public License for more details.
+;;;
+;;; You should have received a copy of the GNU Affero General Public
+;;; License along with this program.  If not, see
+;;; <http://www.gnu.org/licenses/>.
+
+(define-module (guix-data-service model blocked-builds)
+  #:use-module (srfi srfi-1)
+  #:use-module (ice-9 match)
+  #:use-module (squee)
+  #:use-module (guix-data-service database)
+  #:use-module (guix-data-service utils)
+  #:use-module (guix-data-service model utils)
+  #:use-module (guix-data-service model build)
+  #:export (handle-populating-blocked-builds-for-scheduled-builds
+            handle-populating-blocked-builds-for-build-failures
+            handle-removing-blocking-build-entries-for-successful-builds
+
+            backfill-blocked-builds))
+
+(define (select-blocked-derivation-output-details-set-ids-for-blocking-build
+         conn
+         build-server-id
+         blocking-derivation-output-details-set-id)
+  (define query
+    "
+WITH RECURSIVE all_derivations(id, file_name) AS (
+  (
+    SELECT derivations.id, derivations.file_name
+    FROM derivations
+    INNER JOIN derivations_by_output_details_set
+      ON derivations.id = derivations_by_output_details_set.derivation_id
+    WHERE derivation_output_details_set_id = $2
+    LIMIT 1
+  )
+  UNION
+    SELECT derivations.id, derivations.file_name
+    FROM all_derivations
+    INNER JOIN derivation_outputs
+      ON all_derivations.id = derivation_outputs.derivation_id
+    INNER JOIN derivation_inputs
+      ON derivation_outputs.id = derivation_inputs.derivation_output_id
+    INNER JOIN derivations
+      ON derivation_inputs.derivation_id = derivations.id
+)
+SELECT builds.derivation_output_details_set_id
+FROM all_derivations
+INNER JOIN derivations_by_output_details_set
+  ON all_derivations.id = derivations_by_output_details_set.derivation_id
+INNER JOIN builds
+  ON builds.build_server_id = $1
+ AND builds.derivation_output_details_set_id
+      = derivations_by_output_details_set.derivation_output_details_set_id
+INNER JOIN latest_build_status
+  ON builds.id = latest_build_status.build_id
+ AND status = 'scheduled'")
+
+  (exec-query conn
+              query
+              (list build-server-id
+                    blocking-derivation-output-details-set-id)))
+
+(define (select-blocking-builds-for-build-id conn build-id build-server-id)
+  (define query
+    "
+WITH RECURSIVE all_derivations(id, file_name) AS (
+    SELECT derivations.id, derivations.file_name
+    FROM derivations
+    WHERE id IN (
+      -- Look up by the builds.derivation_output_details_set_id, since that'll
+      -- work even if the derivation for the build isn't known.
+      SELECT derivation_id
+      FROM derivations_by_output_details_set
+      INNER JOIN builds
+        ON builds.derivation_output_details_set_id
+           = derivations_by_output_details_set.derivation_output_details_set_id
+      WHERE builds.id = $1
+    )
+  UNION
+    SELECT derivations.id, derivations.file_name
+    FROM all_derivations
+    INNER JOIN derivation_inputs
+      ON all_derivations.id = derivation_inputs.derivation_id
+    INNER JOIN derivation_outputs
+      ON derivation_inputs.derivation_output_id = derivation_outputs.id
+    INNER JOIN derivations
+      ON derivation_outputs.derivation_id = derivations.id
+)
+SELECT derivations_by_output_details_set.derivation_output_details_set_id
+FROM all_derivations
+INNER JOIN derivations_by_output_details_set
+  ON all_derivations.id = derivations_by_output_details_set.derivation_id
+INNER JOIN builds
+  ON derivations_by_output_details_set.derivation_output_details_set_id =
+     builds.derivation_output_details_set_id
+ AND builds.build_server_id = $2
+INNER JOIN latest_build_status
+  ON builds.id = latest_build_status.build_id
+ AND latest_build_status.status IN (
+       'failed', 'canceled', 'failed-dependency',
+       'failed-other'
+     )
+WHERE NOT EXISTS (
+    SELECT 1
+    FROM builds AS successful_builds
+    INNER JOIN build_status AS successful_builds_build_status
+      ON successful_builds.id = successful_builds_build_status.build_id
+    WHERE successful_builds.derivation_output_details_set_id =
+          builds.derivation_output_details_set_id
+      AND successful_builds.build_server_id = $2
+      AND successful_builds_build_status.status = 'succeeded'
+)")
+
+  (exec-query conn
+              query
+              (list (number->string build-id)
+                    build-server-id)))
+
+(define (insert-blocked-builds conn data)
+  (define (create-partitions)
+    (for-each
+     (lambda (build-server-id)
+       (exec-query
+        conn
+        (string-append
+         "
+CREATE TABLE IF NOT EXISTS blocked_builds_build_server__"
+         (number->string build-server-id) "
+PARTITION OF blocked_builds FOR VALUES IN ("
+         (number->string build-server-id)
+         ")")))
+     (delete-duplicates
+      (map (lambda (fields)
+             (string->number (car fields)))
+           data)
+      =)))
+
+  (define (try-insert)
+    (exec-query
+     conn
+     (string-append
+      "
+INSERT INTO blocked_builds (
+  build_server_id,
+  blocked_derivation_output_details_set_id,
+  blocking_derivation_output_details_set_id
+)
+VALUES "
+      (string-join
+       (map (match-lambda
+              ((a b c)
+               (simple-format #f "(~A, ~A, ~A)" a b c)))
+            data)
+       ", ")
+      "
+ON CONFLICT DO NOTHING")
+     '()))
+
+  (unless (null? data)
+    (with-exception-handler
+        (lambda (exn)
+          (create-partitions)
+
+          (try-insert))
+      try-insert
+      #:unwind? #t)))
+
+(define (handle-populating-blocked-builds-for-scheduled-builds conn build-ids)
+  (define (get-build-details build-id)
+    (define query
+      "
+SELECT build_server_id, derivation_output_details_set_id
+FROM builds
+WHERE id = $1")
+
+    (exec-query conn query (list (number->string build-id))))
+
+  (for-each
+   (lambda (build-id)
+     (match (get-build-details build-id)
+       (((build-server-id blocked-derivation-output-details-set-id))
+        (let ((blocking-derivation-output-details-set-ids
+               (select-blocking-builds-for-build-id conn build-id build-server-id)))
+
+          (unless (null? blocking-derivation-output-details-set-ids)
+            (insert-blocked-builds
+             conn
+             (map
+              (lambda (blocking-derivation-output-details-set-id)
+                (list build-server-id
+                      blocked-derivation-output-details-set-id
+                      blocking-derivation-output-details-set-id))
+              blocking-derivation-output-details-set-ids)))))))
+   build-ids)
+
+  #t)
+
+(define (handle-populating-blocked-builds-for-build-failures conn build-ids)
+  (define build-build-server-id-and-derivation-output-details-set-ids-query
+    (string-append
+     "
+SELECT builds.build_server_id, builds.derivation_output_details_set_id
+FROM builds
+INNER JOIN build_status
+  ON build_status.build_id = builds.id
+  -- This should only be run on builds that have failed, but double check here
+ AND status IN ('failed', 'failed-dependency', 'failed-other', 'canceled')
+WHERE builds.id IN ("
+     (string-join (map number->string build-ids) ", ")
+     ")
+  AND builds.derivation_output_details_set_id IS NOT NULL
+  AND NOT EXISTS (
+    SELECT 1
+    FROM builds AS builds_for_same_output
+    INNER JOIN build_status AS builds_for_same_output_build_status
+      ON builds_for_same_output.id
+           = builds_for_same_output_build_status.build_id
+     AND builds_for_same_output_build_status.status = 'succeeded'
+    WHERE builds_for_same_output.derivation_output_details_set_id
+            = builds.derivation_output_details_set_id
+)"))
+
+  (for-each
+   (match-lambda
+     ((build-server-id blocking-derivation-output-details-set-id)
+      (let ((blocked-derivation-output-details-set-ids
+             (select-blocked-derivation-output-details-set-ids-for-blocking-build
+              conn
+              build-server-id
+              blocking-derivation-output-details-set-id)))
+        (insert-blocked-builds
+         conn
+         (map
+          (lambda (blocked-derivation-output-details-set-id)
+            (list build-server-id
+                  blocked-derivation-output-details-set-id
+                  blocking-derivation-output-details-set-id))
+          blocked-derivation-output-details-set-ids)))))
+   (exec-query
+    conn
+    build-build-server-id-and-derivation-output-details-set-ids-query
+    '())))
+
+(define (handle-removing-blocking-build-entries-for-successful-builds conn build-ids)
+  (define query
+    (string-append
+     "
+DELETE FROM blocked_builds
+WHERE EXISTS (
+  SELECT 1
+  FROM builds
+  WHERE builds.id IN (" (string-join
+                         (map number->string build-ids)
+                         ", ")
+  ")
+    AND EXISTS (
+          SELECT 1
+          FROM build_status
+          WHERE build_status.build_id = builds.id
+            AND build_status.status = 'succeeded'
+        )
+    AND blocked_builds.build_server_id = builds.build_server_id
+    AND blocked_builds.blocking_derivation_output_details_set_id
+         = builds.derivation_output_details_set_id
+)"))
+
+  (exec-query conn query '()))
+
+(define (backfill-blocked-builds conn)
+  (define query
+    "
+SELECT build_id
+FROM latest_build_status
+ WHERE status IN ('failed', 'failed-dependency', 'failed-other', 'canceled')")
+
+  (let ((build-ids
+         (map car (exec-query conn query '()))))
+    (chunk-for-each!
+     (lambda (ids)
+       (with-postgresql-transaction
+        conn
+        (lambda (conn)
+          (exec-query
+           conn
+           "LOCK TABLE blocked_builds IN SHARE MODE")
+
+          (handle-populating-blocked-builds-for-build-failures
+           conn
+           (map string->number ids))
+          (simple-format #t "processed chunk...\n"))))
+     1000
+     build-ids)))
diff --git a/guix-data-service/web/build-server/controller.scm b/guix-data-service/web/build-server/controller.scm
index 2514f53..7c2ace6 100644
--- a/guix-data-service/web/build-server/controller.scm
+++ b/guix-data-service/web/build-server/controller.scm
@@ -20,6 +20,7 @@
   #:use-module (ice-9 match)
   #:use-module (rnrs bytevectors)
   #:use-module (json)
+  #:use-module (fibers)
   #:use-module (guix-data-service utils)
   #:use-module (guix-data-service database)
   #:use-module (guix-data-service web render)
@@ -29,6 +30,7 @@
   #:use-module (guix-data-service model build)
   #:use-module (guix-data-service model build-server)
   #:use-module (guix-data-service model build-status)
+  #:use-module (guix-data-service model blocked-builds)
   #:use-module (guix-data-service model nar)
   #:use-module (guix-data-service model build-server-token-seed)
   #:use-module (guix-data-service web util)
@@ -118,63 +120,112 @@
   (define build-server-id
     (string->number build-server-id-string))
 
+  (define (spawn-fiber-for-build-handler handler
+                                         statuses
+                                         data
+                                         build-ids)
+    (let ((ids
+           (delete-duplicates
+            (filter-map
+             (lambda (build-id item-data)
+               (if (and (string=? (assoc-ref item-data "type")
+                                  "build")
+                        (member (assoc-ref item-data "event")
+                                statuses))
+                   build-id
+                   #f))
+             build-ids
+             data)
+            =)))
+      (unless (null? ids)
+        (spawn-fiber
+         (lambda ()
+           (parallel-via-thread-pool-channel
+            (with-thread-postgresql-connection
+             (lambda (conn)
+               (handler conn ids)))))))))
+
   (define (handle-derivation-events conn items)
-    (unless (null? items)
-      (let ((build-ids
-             (insert-builds
-              conn
-              build-server-id
-              (map (lambda (item)
-                     (assoc-ref item "derivation"))
-                   items)
-              (map (lambda (item)
-                     (and=>
-                      (assoc-ref item "derivation_outputs")
-                      (lambda (outputs)
-                        (map
-                         (lambda (output)
-                           `((path           . ,(assoc-ref output "output"))
-                             (hash_algorithm
-                              . ,(or (assoc-ref output "hash_algorithm")
-                                     NULL))
-                             (hash           . ,(or (assoc-ref output "hash")
-                                                    NULL))
-                             (recursive      . ,(assoc-ref output "recursive"))))
-                         (vector->list outputs)))))
-                   items)
-              (map (lambda (item)
-                     (assoc-ref item "build_id"))
-                   items))))
-        (insert-build-statuses
-         conn
-         build-ids
-         (map
-          (lambda (item-data)
-            (list (assoc-ref item-data "timestamp")
-                  (assoc-ref item-data "event")))
-          items)
-         #:transaction? #f))))
+    (if (null? items)
+        '()
+        (let ((build-ids
+               (insert-builds
+                conn
+                build-server-id
+                (map (lambda (item)
+                       (assoc-ref item "derivation"))
+                     items)
+                (map (lambda (item)
+                       (and=>
+                        (assoc-ref item "derivation_outputs")
+                        (lambda (outputs)
+                          (map
+                           (lambda (output)
+                             `((path           . ,(assoc-ref output "output"))
+                               (hash_algorithm
+                                . ,(or (assoc-ref output "hash_algorithm")
+                                       NULL))
+                               (hash           . ,(or (assoc-ref output "hash")
+                                                      NULL))
+                               (recursive      . ,(assoc-ref output "recursive"))))
+                           (vector->list outputs)))))
+                     items)
+                (map (lambda (item)
+                       (assoc-ref item "build_id"))
+                     items))))
+          (insert-build-statuses
+           conn
+           build-ids
+           (map
+            (lambda (item-data)
+              (list (assoc-ref item-data "timestamp")
+                    (assoc-ref item-data "event")))
+            items)
+           #:transaction? #f)
+
+          build-ids)))
 
   (define (process-items items)
-    (parallel-via-thread-pool-channel
-     (with-thread-postgresql-connection
-      (lambda (conn)
-        (with-postgresql-transaction
-         conn
-         (lambda (conn)
-           (handle-derivation-events
-            conn
-            (filter (lambda (item)
-                      (let ((type (assoc-ref item "type")))
-                        (if type
-                            (string=? type "build")
-                            (begin
-                              (simple-format
-                               (current-error-port)
-                               "warning: unknown type for event: ~A\n"
-                               item)
-                              #f))))
-                    items))))))))
+    (define filtered-items
+      (filter (lambda (item)
+                (let ((type (assoc-ref item "type")))
+                  (if type
+                      (string=? type "build")
+                      (begin
+                        (simple-format
+                         (current-error-port)
+                         "warning: unknown type for event: ~A\n"
+                         item)
+                        #f))))
+              items))
+
+    (letpar& ((build-ids
+               (with-thread-postgresql-connection
+                (lambda (conn)
+                  (with-postgresql-transaction
+                   conn
+                   (lambda (conn)
+                     (handle-derivation-events
+                      conn
+                      filtered-items)))))))
+
+      (spawn-fiber-for-build-handler
+       handle-removing-blocking-build-entries-for-successful-builds
+       '("succeeded")
+       items
+       build-ids)
+
+      (spawn-fiber-for-build-handler
+       handle-populating-blocked-builds-for-scheduled-builds
+       '("scheduled")
+       items
+       build-ids)
+
+      (spawn-fiber-for-build-handler
+       handle-populating-blocked-builds-for-build-failures
+       '("failed" "failed-dependency" "canceled")
+       items
+       build-ids)))
 
   (if (any-invalid-query-parameters? parsed-query-parameters)
       (render-json