Move more logic around submitting builds in to the coordinator

Originally I was trying to keep the implementation details of the datastore in
the datastore modules, but this approach starts to crack as you cope with more
and more complicated transactions.

This change should help resolve issues around getting the coordinator logic in
to the coordinator module, and simplifying the SQLite datastore in preparation
for adding PostgreSQL support.

3 files changed, 138 insertions, 113 deletions

diff --git a/guix-build-coordinator/coordinator.scm b/guix-build-coordinator/coordinator.scm
index 8393fd9..3797349 100644
--- a/guix-build-coordinator/coordinator.scm
+++ b/guix-build-coordinator/coordinator.scm
@@ -277,6 +277,55 @@
            (not (null? builds-for-output))))
        (datastore-find-derivation-outputs datastore derivation-file))))
 
+  (define (store-build derivation-name
+                       uuid
+                       priority
+                       tags)
+    (datastore-insert-build datastore
+                            derivation-name
+                            uuid
+                            priority)
+    (datastore-insert-unprocessed-hook-event datastore
+                                             "build-submitted"
+                                             (list uuid))
+    (unless (null? tags)
+      (datastore-insert-build-tags datastore
+                                   uuid
+                                   tags))
+    #t)
+
+  (define build-id
+    (or requested-uuid
+        (random-v4-uuid)))
+
+  (define (perform-datastore-changes)
+    ;; Actually create a build
+    (when ensure-all-related-derivation-outputs-have-builds?
+      (let ((derivations-lacking-builds
+             (datastore-list-related-derivations-with-no-build-for-outputs
+              datastore
+              derivation-file)))
+        (for-each
+         (lambda (related-derivation)
+           (let ((related-uuid (random-v4-uuid)))
+             (simple-format #t "submtiting ~A for related ~A\n"
+                            related-uuid
+                            related-derivation)
+             (store-build related-derivation
+                          related-uuid
+                          ;; Let the scheduler take care of
+                          ;; the prioritisation
+                          0
+                          tags)))
+         derivations-lacking-builds)))
+
+    (store-build derivation-file
+                 build-id
+                 priority
+                 tags)
+
+    #t)
+
   (call-with-duration-metric
    (build-coordinator-metrics-registry build-coordinator)
    "coordinator_submit_build_duration_seconds"
@@ -287,41 +336,18 @@
          (if (and ignore-if-build-for-outputs-exists?
                   (build-for-output-already-exists?))
              '((no-build-submitted . build-already-exists-for-a-output))
-
-             ;; Actually create a build
-             (let ((uuid (or requested-uuid (random-v4-uuid))))
-               (when ensure-all-related-derivation-outputs-have-builds?
-                 (let ((derivations-lacking-builds
-                        (datastore-list-related-derivations-with-no-build-for-outputs
-                         datastore
-                         derivation-file)))
-                   (for-each
-                    (lambda (related-derivation)
-                      (let ((related-uuid (random-v4-uuid)))
-                        (simple-format #t "submtiting ~A for related ~A\n"
-                                       related-uuid
-                                       related-derivation)
-                        (datastore-store-build datastore
-                                               related-derivation
-                                               related-uuid
-                                               ;; Let the scheduler take care of
-                                               ;; the prioritisation
-                                               0
-                                               tags)))
-                    derivations-lacking-builds)))
-
-               (datastore-store-build datastore
-                                      derivation-file
-                                      uuid
-                                      priority
-                                      tags)
+             (begin
+               (datastore-call-with-transaction datastore
+                                                perform-datastore-changes
+                                                #:duration-metric-name
+                                                "store_build")
 
                (build-coordinator-prompt-hook-processing-for-event
                 build-coordinator
                 'build-submitted)
                (trigger-build-allocation build-coordinator)
 
-               `((build-submitted . ,uuid))))))))
+               `((build-submitted . ,build-id))))))))
 
 (define (cancel-build build-coordinator uuid)
   (datastore-cancel-build
diff --git a/guix-build-coordinator/datastore.scm b/guix-build-coordinator/datastore.scm
index c39a454..e0be527 100644
--- a/guix-build-coordinator/datastore.scm
+++ b/guix-build-coordinator/datastore.scm
@@ -14,7 +14,7 @@
 (re-export datastore-update)
 (re-export datastore-call-with-transaction)
 (re-export datastore-store-derivation)
-(re-export datastore-store-build)
+(re-export datastore-insert-build)
 (re-export datastore-cancel-build)
 (re-export datastore-new-agent)
 (re-export datastore-list-agents)
@@ -38,6 +38,7 @@
 (re-export datastore-count-builds)
 (re-export datastore-for-each-build)
 (re-export datastore-find-build)
+(re-export datastore-insert-build-tags)
 (re-export datastore-fetch-build-tags)
 (re-export datastore-find-build-result)
 (re-export datastore-find-build-derivation-system)
@@ -45,6 +46,7 @@
 (re-export datastore-list-processed-builds)
 (re-export datastore-list-unprocessed-builds)
 (re-export datastore-fetch-build-ids-and-propagated-priorities-for-unprocessed-builds)
+(re-export datastore-insert-unprocessed-hook-event)
 (re-export datastore-count-unprocessed-hook-events)
 (re-export datastore-list-unprocessed-hook-events)
 (re-export datastore-delete-unprocessed-hook-event)
diff --git a/guix-build-coordinator/datastore/sqlite.scm b/guix-build-coordinator/datastore/sqlite.scm
index 6e216ed..9a1a8ec 100644
--- a/guix-build-coordinator/datastore/sqlite.scm
+++ b/guix-build-coordinator/datastore/sqlite.scm
@@ -23,11 +23,12 @@
             datastore-list-related-derivations-with-no-build-for-outputs
             datastore-list-failed-builds-with-blocking-count
             datastore-list-builds-for-derivation-recursive-inputs
-            datastore-store-build
+            datastore-insert-build
             datastore-cancel-build
             datastore-count-builds
             datastore-for-each-build
             datastore-find-build
+            datastore-insert-build-tags
             datastore-fetch-build-tags
             datastore-find-build-result
             datastore-find-build-derivation-system
@@ -58,6 +59,7 @@
             datastore-list-processed-builds
             datastore-list-unprocessed-builds
             datastore-fetch-build-ids-and-propagated-priorities-for-unprocessed-builds
+            datastore-insert-unprocessed-hook-event
             datastore-count-unprocessed-hook-events
             datastore-list-unprocessed-hook-events
             datastore-delete-unprocessed-hook-event
@@ -585,81 +587,59 @@ INNER JOIN related_derivations
 
          result)))))
 
-(define-method (datastore-store-build
+(define-method (datastore-insert-build-tags
                 (datastore <sqlite-datastore>)
-                derivation-name
-                uuid
-                priority
+                build-id
                 tags)
-  (define (insert-tags db build-id tags)
-    (let ((insert-tag-statement
-           (sqlite-prepare
-            db
-            "
+  (call-with-worker-thread
+   (slot-ref datastore 'worker-writer-thread-channel)
+   (lambda (db)
+     (let ((insert-tag-statement
+            (sqlite-prepare
+             db
+             "
 INSERT INTO tags (\"key\", \"value\") VALUES (:tagkey, :tagvalue)"
-            #:cache? #t))
-          (find-tag-statement
-           (sqlite-prepare
-            db
-            "
+             #:cache? #t))
+           (find-tag-statement
+            (sqlite-prepare
+             db
+             "
 SELECT id FROM tags WHERE key = :tag_key AND value = :tag_value"
-            #:cache? #t))
-          (build-tags-statement
-           (sqlite-prepare
-            db
-            "
+             #:cache? #t))
+           (build-tags-statement
+            (sqlite-prepare
+             db
+             "
 INSERT INTO build_tags (build_id, tag_id) VALUES (:build_id, :tag_id)"
-            #:cache? #t)))
-
-      (define (tag->id key value)
-        (sqlite-bind-arguments find-tag-statement
-                               #:tag_key key
-                               #:tag_value value)
-        (let ((result
-               (match (sqlite-step find-tag-statement)
-                 (#(id) id)
-                 (#f
-                  (sqlite-bind-arguments insert-tag-statement
-                                         #:tagkey key
-                                         #:tagvalue value)
-                  (sqlite-step insert-tag-statement)
-                  (sqlite-reset insert-tag-statement)
-                  (last-insert-rowid db)))))
-          (sqlite-reset find-tag-statement)
-
-          result))
-
-      (for-each
-       (match-lambda
-         ((key . value)
-          (sqlite-bind-arguments build-tags-statement
-                                 #:build_id build-id
-                                 #:tag_id (tag->id key value))
-          (sqlite-step build-tags-statement)
-          (sqlite-reset build-tags-statement)))
-       tags)))
-
-  (define (handle-inserting-unprocessed-hook-event db build-id)
-    (insert-unprocessed-hook-event
-     db
-     "build-submitted"
-     (list build-id)))
-
-  (datastore-call-with-transaction
-   datastore
-   (lambda (db)
-     (insert-build db uuid derivation-name priority)
-     (handle-inserting-unprocessed-hook-event db uuid)
-     (unless (null? tags)
-       (insert-tags db uuid tags)))
-   #:duration-metric-name "store_build")
+             #:cache? #t)))
 
-  (metric-increment
-   (metrics-registry-fetch-metric (slot-ref datastore 'metrics-registry)
-                                  "builds_total")
-   #:label-values `((system . ,(datastore-find-derivation-system
-                                datastore
-                                derivation-name))))
+       (define (tag->id key value)
+         (sqlite-bind-arguments find-tag-statement
+                                #:tag_key key
+                                #:tag_value value)
+         (let ((result
+                (match (sqlite-step find-tag-statement)
+                  (#(id) id)
+                  (#f
+                   (sqlite-bind-arguments insert-tag-statement
+                                          #:tagkey key
+                                          #:tagvalue value)
+                   (sqlite-step insert-tag-statement)
+                   (sqlite-reset insert-tag-statement)
+                   (last-insert-rowid db)))))
+           (sqlite-reset find-tag-statement)
+
+           result))
+
+       (for-each
+        (match-lambda
+          ((key . value)
+           (sqlite-bind-arguments build-tags-statement
+                                  #:build_id build-id
+                                  #:tag_id (tag->id key value))
+           (sqlite-step build-tags-statement)
+           (sqlite-reset build-tags-statement)))
+        tags))))
   #t)
 
 (define-method (datastore-cancel-build
@@ -1662,6 +1642,17 @@ GROUP BY builds_with_derived_priority.uuid"
                unprocessed-build-ids)))
    #:readonly? #t))
 
+(define-method (datastore-insert-unprocessed-hook-event
+                (datastore <sqlite-datastore>)
+                event
+                arguments)
+  (call-with-worker-thread
+   (slot-ref datastore 'worker-writer-thread-channel)
+   (lambda (db)
+     (insert-unprocessed-hook-event db
+                                    event
+                                    arguments))))
+
 (define (insert-unprocessed-hook-event
          db
          event
@@ -2694,23 +2685,29 @@ INSERT INTO derivation_outputs (derivation_name, name, output) VALUES "
                  derivation-outputs))
   #t)
 
-(define (insert-build db uuid derivation-name priority)
-  (let ((statement
-         (sqlite-prepare
-          db
-          "
+(define-method (datastore-insert-build
+                (datastore <sqlite-datastore>)
+                uuid derivation-name priority)
+  (call-with-worker-thread
+   (slot-ref datastore 'worker-writer-thread-channel)
+   (lambda (db)
+     (let ((statement
+            (sqlite-prepare
+             db
+             "
 INSERT INTO builds (uuid, derivation_name, priority, created_at)
 VALUES (:uuid, :derivation_name, :priority, datetime('now'))"
-          #:cache? #t)))
+             #:cache? #t)))
 
-    (sqlite-bind-arguments
-     statement
-     #:uuid uuid
-     #:derivation_name derivation-name
-     #:priority priority)
+       (sqlite-bind-arguments
+        statement
+        #:uuid uuid
+        #:derivation_name derivation-name
+        #:priority priority)
 
-    (sqlite-step statement)
-    (sqlite-reset statement)))
+       (sqlite-step statement)
+       (sqlite-reset statement))))
+  #t)
 
 (define (insert-agent db uuid description)
   (let ((statement