Move build allocation complexity out of the datastore

And in to the coordinator module. This will make adding more datastore's
easier.

3 files changed, 180 insertions, 162 deletions

diff --git a/guix-build-coordinator/coordinator.scm b/guix-build-coordinator/coordinator.scm
index 349597f..5e56f09 100644
--- a/guix-build-coordinator/coordinator.scm
+++ b/guix-build-coordinator/coordinator.scm
@@ -596,6 +596,27 @@
 
 (define (fetch-builds build-coordinator agent systems
                       max-builds deprecated-requested-count)
+  (define datastore
+    (build-coordinator-datastore build-coordinator))
+
+  (define (allocate-one-build agent-id)
+    (let ((build-details (datastore-fetch-build-to-allocate datastore agent-id)))
+      (if build-details
+          (let ((build-id (assq-ref build-details 'uuid)))
+            (datastore-insert-to-allocated-builds datastore agent-id (list build-id))
+            (datastore-remove-builds-from-plan datastore (list build-id))
+            build-details)
+          #f)))
+
+  (define (allocate-several-builds agent-id count)
+    (let loop ((builds '()))
+      (if (= (length builds) count)
+          builds
+          (let ((build-details (allocate-one-build agent-id)))
+            (if build-details
+                (loop (cons build-details builds))
+                builds)))))
+
   (call-with-duration-metric
    (build-coordinator-metrics-registry build-coordinator)
    "coordinator_fetch_builds_duration_seconds"
@@ -607,11 +628,49 @@
        (when update-made
          (trigger-build-allocation build-coordinator)))
 
-     (datastore-allocate-builds-to-agent
-      (build-coordinator-datastore build-coordinator)
-      agent
-      max-builds
-      deprecated-requested-count))))
+     (datastore-call-with-transaction
+      datastore
+      (lambda _
+        (let* ((initially-allocated-builds
+                (datastore-select-allocated-builds datastore agent))
+               (start-count
+                (length initially-allocated-builds))
+               (target-count (or max-builds
+                                 (+ start-count
+                                    deprecated-requested-count))))
+          (if (< start-count target-count)
+              (let ((new-builds
+                     (allocate-several-builds agent
+                                              (- target-count start-count))))
+                (unless (null? new-builds)
+                  (let ((allocation-plan-metric
+                         (metrics-registry-fetch-metric
+                          (slot-ref datastore 'metrics-registry)
+                          "build_allocation_plan_total")))
+                    (for-each
+                     (match-lambda
+                       ((agent-id . count)
+                        (metric-set allocation-plan-metric
+                                    count
+                                    #:label-values
+                                    `((agent_id . ,agent-id)))))
+                     (datastore-count-build-allocation-plan-entries datastore))))
+
+                ;; Previously allocate builds just returned newly allocated
+                ;; builds, but if max-builds is provided, return all the
+                ;; builds. This means the agent can handle this in a idempotent
+                ;; manor.
+                (if max-builds
+                    (append initially-allocated-builds
+                            new-builds)
+                    new-builds))
+              ;; Previously allocate builds just returned newly allocated builds,
+              ;; but if max-builds is provided, return all the builds. This means
+              ;; the agent can handle this in a idempotent manor.
+              (if max-builds
+                  initially-allocated-builds
+                  '()))))
+      #:duration-metric-name "allocate_builds_to_agent"))))
 
 (define (agent-details datastore agent-id)
   (let ((agent (datastore-find-agent datastore agent-id))
diff --git a/guix-build-coordinator/datastore.scm b/guix-build-coordinator/datastore.scm
index 913d53a..14eed3e 100644
--- a/guix-build-coordinator/datastore.scm
+++ b/guix-build-coordinator/datastore.scm
@@ -66,7 +66,10 @@
 (re-export datastore-count-allocated-builds)
 (re-export datastore-agent-requested-systems)
 (re-export datastore-update-agent-requested-systems)
-(re-export datastore-allocate-builds-to-agent)
+(re-export datastore-fetch-build-to-allocate)
+(re-export datastore-insert-to-allocated-builds)
+(re-export datastore-remove-builds-from-plan)
+(re-export datastore-select-allocated-builds)
 (re-export datastore-list-allocation-plan-builds)
 
 (define* (database-uri->datastore database
diff --git a/guix-build-coordinator/datastore/sqlite.scm b/guix-build-coordinator/datastore/sqlite.scm
index bc3dd58..8515da8 100644
--- a/guix-build-coordinator/datastore/sqlite.scm
+++ b/guix-build-coordinator/datastore/sqlite.scm
@@ -73,7 +73,10 @@
             datastore-count-allocated-builds
             datastore-agent-requested-systems
             datastore-update-agent-requested-systems
-            datastore-allocate-builds-to-agent
+            datastore-fetch-build-to-allocate
+            datastore-insert-to-allocated-builds
+            datastore-remove-builds-from-plan
+            datastore-select-allocated-builds
             datastore-list-allocation-plan-builds))
 
 (define-class <sqlite-datastore> (<abstract-datastore>)
@@ -2141,17 +2144,17 @@ INSERT INTO build_allocation_agent_requested_systems (agent_id, system) VALUES "
 
          #t))))
 
-(define-method (datastore-allocate-builds-to-agent
+(define-method (datastore-fetch-build-to-allocate
                 (datastore <sqlite-datastore>)
-                agent-id
-                max-builds
-                deprecated-requested-count)
-  (define (fetch-build db)
-    (let ((statement
-           (sqlite-prepare
-            db
-            ;; This needs to guard against the plan being out of date
-            "
+                agent-id)
+  (datastore-call-with-transaction
+   datastore
+   (lambda (db)
+     (let ((statement
+            (sqlite-prepare
+             db
+             ;; This needs to guard against the plan being out of date
+             "
 SELECT builds.uuid, builds.derivation_name
 FROM builds
 INNER JOIN build_allocation_plan
@@ -2166,11 +2169,11 @@ WHERE build_allocation_plan.agent_id = :agent_id
   AND builds.canceled = 0
   AND builds.uuid NOT IN (SELECT build_id FROM allocated_builds)
 ORDER BY build_allocation_plan.ordering DESC"
-            #:cache? #t))
-          (output-conflicts-statement
-           (sqlite-prepare
-            db
-            "
+             #:cache? #t))
+           (output-conflicts-statement
+            (sqlite-prepare
+             db
+             "
 SELECT 1
 FROM derivation_outputs AS build_derivation_outputs
 INNER JOIN allocated_builds
@@ -2183,165 +2186,118 @@ INNER JOIN derivation_outputs AS allocated_builds_derivation_outputs
 WHERE build_derivation_outputs.derivation_name = :derivation_name
   AND build_derivation_outputs.output =
       allocated_builds_derivation_outputs.output"
-            #:cache? #t)))
+             #:cache? #t)))
 
-      (sqlite-bind-arguments
-       statement
-       #:agent_id agent-id)
+       (sqlite-bind-arguments
+        statement
+        #:agent_id agent-id)
 
-      (let ((builds
-             (sqlite-fold
-              (lambda (row result)
-                (cons
-                 (match row
-                   (#f #f)
-                   (#(uuid derivation_name)
-                    `((uuid            . ,uuid)
-                      (derivation-name . ,derivation_name))))
-                 result))
-              '()
-              statement)))
-        (sqlite-reset statement)
+       (let ((builds
+              (sqlite-fold
+               (lambda (row result)
+                 (cons
+                  (match row
+                    (#f #f)
+                    (#(uuid derivation_name)
+                     `((uuid            . ,uuid)
+                       (derivation-name . ,derivation_name))))
+                  result))
+               '()
+               statement)))
+         (sqlite-reset statement)
 
-        (find (lambda (build-details)
-                (sqlite-bind-arguments
-                 output-conflicts-statement
-                 #:derivation_name (assq-ref build-details
-                                             'derivation-name))
+         (find (lambda (build-details)
+                 (sqlite-bind-arguments
+                  output-conflicts-statement
+                  #:derivation_name (assq-ref build-details
+                                              'derivation-name))
 
-                (let ((result (sqlite-step output-conflicts-statement)))
-                  (sqlite-reset output-conflicts-statement)
+                 (let ((result (sqlite-step output-conflicts-statement)))
+                   (sqlite-reset output-conflicts-statement)
 
-                  (match result
-                    (#f #t)
-                    (_ #f))))
-              builds))))
+                   (match result
+                     (#f #t)
+                     (_ #f))))
+               builds))))))
 
-  (define (insert-to-allocated-builds db agent-id build-ids)
-    (sqlite-exec
-     db
-     (string-append
-      "
+(define-method (datastore-insert-to-allocated-builds
+                (datastore <sqlite-datastore>)
+                agent-id
+                build-ids)
+  (call-with-worker-thread
+   (slot-ref datastore 'worker-writer-channel)
+   (lambda (db)
+     (sqlite-exec
+      db
+      (string-append
+       "
 INSERT INTO allocated_builds (build_id, agent_id) VALUES "
-      (string-join
-       (map (lambda (build-id)
-              (simple-format
-               #f
-               "('~A', '~A')"
-               build-id
-               agent-id))
-            build-ids)
-       ", ")
-      ";")))
+       (string-join
+        (map (lambda (build-id)
+               (simple-format
+                #f
+                "('~A', '~A')"
+                build-id
+                agent-id))
+             build-ids)
+        ", ")
+       ";")))))
 
-  (define (remove-builds-from-plan db build-ids)
-    (sqlite-exec
-     db
-     (string-append
-      "
+(define-method (datastore-remove-builds-from-plan
+                (datastore <sqlite-datastore>)
+                build-ids)
+  (call-with-worker-thread
+   (slot-ref datastore 'worker-writer-channel)
+   (lambda (db)
+     (sqlite-exec
+      db
+      (string-append
+       "
 DELETE FROM build_allocation_plan
 WHERE build_id IN ("
-      (string-join
-       (map (lambda (build-id)
-              (string-append "'" build-id "'"))
-            build-ids)
-       ", ")
-      ")")))
-
-  (define (allocate-one-build db agent-id)
-    (let ((build-details (fetch-build db)))
-      (if build-details
-          (let ((build-id (assq-ref build-details 'uuid)))
-            (insert-to-allocated-builds db agent-id (list build-id))
-            (remove-builds-from-plan db (list build-id))
-            build-details)
-          #f)))
-
-  (define (allocate-several-builds db agent-id count)
-    (let loop ((builds '()))
-      (if (= (length builds) count)
-          builds
-          (let ((build-details (allocate-one-build db agent-id)))
-            (if build-details
-                (loop (cons build-details builds))
-                builds)))))
-
-  (define (select-allocated-builds db agent-id)
-    (let ((statement
-           (sqlite-prepare
-            db
-            "
+       (string-join
+        (map (lambda (build-id)
+               (string-append "'" build-id "'"))
+             build-ids)
+        ", ")
+       ")")))))
+
+(define-method (datastore-select-allocated-builds
+                (datastore <sqlite-datastore>)
+                agent-id)
+  (call-with-worker-thread
+   (slot-ref datastore 'worker-reader-channel)
+   (lambda (db)
+     (let ((statement
+            (sqlite-prepare
+             db
+             "
 SELECT builds.uuid, builds.derivation_name
 FROM allocated_builds
 INNER JOIN builds
   ON allocated_builds.build_id = builds.uuid
 WHERE agent_id = :agent_id"
-            #:cache? #t)))
-
-      (sqlite-bind-arguments
-       statement
-       #:agent_id agent-id)
+             #:cache? #t)))
 
-      (let ((result
-             (sqlite-fold
-              (lambda (row result)
-                (cons
-                 (match row
-                   (#(uuid derivation_name)
-                    `((uuid            . ,uuid)
-                      (derivation-name . ,derivation_name))))
-                 result))
-              '()
-              statement)))
+       (sqlite-bind-arguments
+        statement
+        #:agent_id agent-id)
 
-        (sqlite-reset statement)
+       (let ((result
+              (sqlite-fold
+               (lambda (row result)
+                 (cons
+                  (match row
+                    (#(uuid derivation_name)
+                     `((uuid            . ,uuid)
+                       (derivation-name . ,derivation_name))))
+                  result))
+               '()
+               statement)))
 
-        result)))
+         (sqlite-reset statement)
 
-  (datastore-call-with-transaction
-   datastore
-   (lambda (db)
-     (let* ((initially-allocated-builds
-             (select-allocated-builds db agent-id))
-            (start-count
-             (length initially-allocated-builds))
-            (target-count (or max-builds
-                              (+ start-count
-                                 deprecated-requested-count))))
-       (if (< start-count target-count)
-           (let ((new-builds
-                  (allocate-several-builds db
-                                           agent-id
-                                           (- target-count start-count))))
-             (unless (null? new-builds)
-               (let ((allocation-plan-metric
-                      (metrics-registry-fetch-metric
-                       (slot-ref datastore 'metrics-registry)
-                       "build_allocation_plan_total")))
-                 (for-each
-                  (match-lambda
-                    ((agent-id . count)
-                     (metric-set allocation-plan-metric
-                                 count
-                                 #:label-values
-                                 `((agent_id . ,agent-id)))))
-                  (datastore-count-build-allocation-plan-entries datastore))))
-
-             ;; Previously allocate builds just returned newly allocated
-             ;; builds, but if max-builds is provided, return all the
-             ;; builds. This means the agent can handle this in a idempotent
-             ;; manor.
-             (if max-builds
-                 (append initially-allocated-builds
-                         new-builds)
-                 new-builds))
-           ;; Previously allocate builds just returned newly allocated builds,
-           ;; but if max-builds is provided, return all the builds. This means
-           ;; the agent can handle this in a idempotent manor.
-           (if max-builds
-               initially-allocated-builds
-               '()))))
-   #:duration-metric-name "allocate_builds_to_agent"))
+         result)))))
 
 (define-method (datastore-list-allocation-plan-builds
                 (datastore <sqlite-datastore>)