Move the allocation plan in to memory

Rather than using an in memory SQLite table, since I think this might be
causing issues when opening new SQLite reader threads.

2 files changed, 52 insertions, 234 deletions

diff --git a/guix-build-coordinator/datastore/abstract.scm b/guix-build-coordinator/datastore/abstract.scm
index 68fc654..799485e 100644
--- a/guix-build-coordinator/datastore/abstract.scm
+++ b/guix-build-coordinator/datastore/abstract.scm
@@ -9,8 +9,7 @@
             datastore-new-agent
             datastore-new-agent-password
             datastore-list-agent-builds
-            datastore-agent-password-exists?
-            datastore-list-allocation-plan-builds))
+            datastore-agent-password-exists?))
 
 (define-class <abstract-datastore> ())
 
@@ -24,5 +23,3 @@
 (define-generic datastore-agent-password-exists?)
 (define-generic datastore-agent-list-unprocessed-builds)
 (define-generic datastore-list-agent-builds)
-(define-generic datastore-agent-replace-build-allocation-plan)
-(define-generic datastore-list-allocation-plan-builds)
diff --git a/guix-build-coordinator/datastore/sqlite.scm b/guix-build-coordinator/datastore/sqlite.scm
index 96751a5..344befb 100644
--- a/guix-build-coordinator/datastore/sqlite.scm
+++ b/guix-build-coordinator/datastore/sqlite.scm
@@ -87,7 +87,7 @@
             datastore-replace-agent-tags
             datastore-list-processed-builds
             datastore-list-unprocessed-builds
-            datastore-find-first-unallocated-deferred-build
+            datastore-find-deferred-build
             datastore-fetch-prioritised-unprocessed-builds
             datastore-insert-unprocessed-hook-event
             datastore-count-unprocessed-hook-events
@@ -96,16 +96,13 @@
             datastore-delete-unprocessed-hook-event
             datastore-list-agent-builds
             datastore-agent-for-build
-            datastore-count-build-allocation-plan-entries
             datastore-replace-build-allocation-plan
-            datastore-remove-build-from-allocation-plan
             datastore-count-allocated-builds
             datastore-agent-requested-systems
             datastore-update-agent-requested-systems
             datastore-fetch-build-to-allocate
-            datastore-insert-to-allocated-builds
-            datastore-remove-builds-from-plan
-            datastore-list-allocation-plan-builds))
+            datastore-check-if-derivation-conflicts?
+            datastore-insert-to-allocated-builds))
 
 (define-class <sqlite-datastore> (<abstract-datastore>)
   database-file
@@ -152,18 +149,6 @@
           (sqlite-exec db "PRAGMA temp_store = MEMORY;")
           (sqlite-exec db "PRAGMA foreign_keys = ON;")
 
-          (sqlite-exec db "ATTACH DATABASE 'file:/mem?vfs=memdb' AS mem;")
-
-          (sqlite-exec
-           db
-           "
-CREATE TABLE IF NOT EXISTS mem.build_allocation_plan (
-       build_id INTEGER NOT NULL,
-       agent_id TEXT NOT NULL,
-       ordering INTEGER NOT NULL,
-       PRIMARY KEY (agent_id, build_id)
-);")
-
           (list db)))
       #:name "ds write"
       #:destructor
@@ -1742,42 +1727,6 @@ WHERE build_id = :build_id"
      #t)
    rest))
 
-(define-method (datastore-remove-build-from-allocation-plan
-                (datastore <sqlite-datastore>)
-                uuid)
-  (define (update-build-allocation-plan-metrics)
-    (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))))
-
-  (call-with-worker-thread/delay-logging
-   (slot-ref datastore 'worker-writer-thread-channel)
-   (lambda (db)
-     (let ((statement (sqlite-prepare
-                       db
-                       "
-DELETE FROM build_allocation_plan WHERE build_id = :build_id"
-                       #:cache? #t)))
-
-       (sqlite-bind-arguments
-        statement
-        #:build_id (db-find-build-id db uuid))
-
-       (sqlite-step-and-reset statement)
-
-       (unless (= 0 (changes-count db))
-         (update-build-allocation-plan-metrics)))))
-  #t)
-
 (define-method (datastore-count-build-results
                 (datastore <sqlite-datastore>))
   (call-with-worker-thread
@@ -3024,8 +2973,9 @@ ORDER BY priority DESC"
 
          builds)))))
 
-(define-method (datastore-find-first-unallocated-deferred-build
-                (datastore <sqlite-datastore>))
+(define-method (datastore-find-deferred-build
+                (datastore <sqlite-datastore>)
+                select?)
   (call-with-worker-thread
    (slot-ref datastore 'worker-reader-thread-channel)
    (lambda (db)
@@ -3040,25 +2990,29 @@ INNER JOIN derivations
 WHERE processed = 0
   AND canceled = 0
   AND deferred_until IS NOT NULL
-  AND builds.id NOT IN (SELECT build_id FROM build_allocation_plan)
-ORDER BY deferred_until ASC
-LIMIT 1"
+ORDER BY deferred_until ASC"
              #:cache? #t)))
 
-       (match (sqlite-step-and-reset statement)
-         (#(uuid derivation_name priority created_at deferred_until)
-          `((uuid            . ,uuid)
-            (derivation-name . ,derivation_name)
-            (priority        . ,priority)
-            (created-at      . ,(if (string? created_at)
-                                    (string->date created_at
-                                                  "~Y-~m-~d ~H:~M:~S")
-                                    #f))
-            (deferred-until . ,(if (string? deferred_until)
-                                   (string->date deferred_until
-                                                 "~Y-~m-~d ~H:~M:~S")
-                                   #f))))
-         (#f #f))))))
+       (let loop ((row (sqlite-step statement)))
+         (match row
+           (#(uuid derivation_name priority created_at deferred_until)
+            (let ((res
+                   (select?
+                    `((uuid            . ,uuid)
+                      (derivation-name . ,derivation_name)
+                      (priority        . ,priority)
+                      (created-at      . ,(if (string? created_at)
+                                              (string->date created_at
+                                                            "~Y-~m-~d ~H:~M:~S")
+                                              #f))
+                      (deferred-until . ,(if (string? deferred_until)
+                                             (string->date deferred_until
+                                                           "~Y-~m-~d ~H:~M:~S")
+                                             #f))))))
+              (if res
+                  res
+                  (loop (sqlite-step statement)))))
+           (#f #f)))))))
 
 (define-method (datastore-fetch-prioritised-unprocessed-builds
                 (datastore <sqlite-datastore>))
@@ -3230,103 +3184,6 @@ DELETE FROM unprocessed_hook_events WHERE id = :id"
        (sqlite-step-and-reset statement))))
   #t)
 
-(define-method (datastore-count-build-allocation-plan-entries
-                (datastore <sqlite-datastore>))
-  (call-with-worker-thread
-   (slot-ref datastore 'worker-reader-thread-channel)
-   (lambda (db)
-     (let ((statement
-            (sqlite-prepare
-             db
-             "
-SELECT agent_id, COUNT(*)
-FROM build_allocation_plan
-GROUP BY agent_id"
-             #:cache? #t)))
-
-       (let ((result
-              (sqlite-map
-               (match-lambda
-                 (#(agent_id count)
-                  (cons agent_id count)))
-               statement)))
-         (sqlite-reset statement)
-
-         result)))))
-
-(define-method (datastore-replace-build-allocation-plan
-                (datastore <sqlite-datastore>)
-                planned-builds)
-  (define (clear-current-plan db)
-    (sqlite-exec
-     db
-     "DELETE FROM build_allocation_plan"))
-
-  (define insert-sql
-    (call-with-worker-thread
-     (slot-ref datastore 'worker-reader-thread-channel)
-     (lambda (db)
-       (string-append
-        "
-INSERT INTO build_allocation_plan (build_id, agent_id, ordering) VALUES "
-        (string-join
-         (map (match-lambda
-                ((build-uuid agent-id ordering)
-                 (simple-format
-                  #f
-                  "('~A', '~A', ~A)"
-                  (db-find-build-id db build-uuid)
-                  agent-id
-                  ordering)))
-              planned-builds)
-         ", ")
-        ";"))))
-
-  (define (insert-new-plan db planned-builds)
-    (sqlite-exec
-     db
-     insert-sql))
-
-  (datastore-call-with-transaction
-   datastore
-   (lambda (db)
-     (clear-current-plan db)
-     (unless (null? planned-builds)
-       (insert-new-plan db planned-builds)))
-   #:duration-metric-name "replace_build_allocation_plan")
-
-  (let* ((agent-ids
-          (map (lambda (agent-details)
-                 (assq-ref agent-details 'uuid))
-               (datastore-list-agents datastore)))
-         (counts-by-agent
-          (make-vector (length agent-ids) 0)))
-    (for-each
-     (match-lambda
-       ((_ agent-id _)
-        (let ((index (list-index (lambda (list-agent-id)
-                                   (string=? agent-id list-agent-id))
-                                 agent-ids)))
-          (vector-set! counts-by-agent
-                       index
-                       (+ (vector-ref counts-by-agent
-                                      index)
-                          1)))))
-     planned-builds)
-
-    (let ((metric
-           (metrics-registry-fetch-metric (slot-ref datastore 'metrics-registry)
-                                          "build_allocation_plan_total")))
-      (for-each
-       (lambda (index agent-id)
-         (metric-set metric
-                     (vector-ref counts-by-agent index)
-                     #:label-values
-                     `((agent_id . ,agent-id))))
-       (iota (length agent-ids))
-       agent-ids)))
-  #t)
-
 (define-method (datastore-count-allocated-builds
                 (datastore <sqlite-datastore>))
   (call-with-worker-thread/delay-logging
@@ -3426,34 +3283,40 @@ INSERT INTO build_allocation_agent_requested_systems (agent_id, system_id) VALUE
 
 (define-method (datastore-fetch-build-to-allocate
                 (datastore <sqlite-datastore>)
-                agent-id)
-  (datastore-call-with-transaction
-   datastore
+                build-id)
+  (call-with-worker-thread
+   (slot-ref datastore 'worker-reader-thread-channel)
    (lambda (db)
      (let ((statement
             (sqlite-prepare
              db
-             ;; This needs to guard against the plan being out of date
              "
 SELECT builds.uuid, derivations.id, derivations.name,
        unprocessed_builds_with_derived_priorities.derived_priority
 FROM builds
-INNER JOIN build_allocation_plan
-  ON builds.id = build_allocation_plan.build_id
 INNER JOIN derivations
   ON builds.derivation_id = derivations.id
-INNER JOIN build_allocation_agent_requested_systems
-  ON build_allocation_agent_requested_systems.agent_id = :agent_id
- AND build_allocation_agent_requested_systems.system_id = derivations.system_id
 LEFT JOIN unprocessed_builds_with_derived_priorities
   ON unprocessed_builds_with_derived_priorities.build_id = builds.id
-WHERE build_allocation_plan.agent_id = :agent_id
+WHERE builds.uuid = :uuid
   AND builds.processed = 0
   AND builds.canceled = 0
-  AND builds.id NOT IN (SELECT build_id FROM allocated_builds)
-ORDER BY build_allocation_plan.ordering ASC"
-             #:cache? #t))
-           (output-conflicts-statement
+  AND builds.id NOT IN (SELECT build_id FROM allocated_builds)"
+             #:cache? #t)))
+       (sqlite-bind-arguments
+        statement
+        #:uuid build-id)
+
+       (sqlite-step-and-reset statement)))))
+
+(define-method (datastore-check-if-derivation-conflicts?
+                (datastore <sqlite-datastore>)
+                agent-id
+                derivation-id)
+  (call-with-worker-thread
+   (slot-ref datastore 'worker-reader-thread-channel)
+   (lambda (db)
+     (let ((statement
             (sqlite-prepare
              db
              "
@@ -3471,34 +3334,11 @@ WHERE build_derivation_outputs.derivation_id = :derivation_id
       allocated_builds_derivation_outputs.output_id"
              #:cache? #t)))
 
-       (define (get-build-to-allocate)
-         (let loop ((build-details (sqlite-step statement)))
-           (match build-details
-             (#f #f)
-             (#(uuid derivation-id derivation-name derived_priority)
-
-              (sqlite-bind-arguments output-conflicts-statement
-                                     #:agent_id agent-id
-                                     #:derivation_id derivation-id)
-
-              (if (eq? (sqlite-step-and-reset output-conflicts-statement)
-                       #f)
-                  `((uuid . ,uuid)
-                    (derivation_name . ,derivation-name)
-                    (derived_priority . ,derived_priority))
-                  (loop (sqlite-step statement)))))))
-
-       (sqlite-bind-arguments
-        statement
-        #:agent_id agent-id)
-
-       (let ((result (get-build-to-allocate)))
-         (sqlite-reset statement)
-
-         result)))
+       (sqlite-bind-arguments statement
+                              #:agent_id agent-id
+                              #:derivation_id derivation-id)
 
-   #:readonly? #t
-   #:duration-metric-name "fetch_builds_to_allocate"))
+       (->bool (sqlite-step-and-reset statement))))))
 
 (define-method (datastore-insert-to-allocated-builds
                 (datastore <sqlite-datastore>)
@@ -3523,25 +3363,6 @@ INSERT INTO allocated_builds (build_id, agent_id) VALUES "
         ", ")
        ";")))))
 
-(define-method (datastore-remove-builds-from-plan
-                (datastore <sqlite-datastore>)
-                build-uuids)
-  (call-with-worker-thread
-   (slot-ref datastore 'worker-writer-thread-channel)
-   (lambda (db)
-     (sqlite-exec
-      db
-      (string-append
-       "
-DELETE FROM build_allocation_plan
-WHERE build_id IN ("
-       (string-join
-        (map (lambda (build-uuid)
-               (number->string (db-find-build-id db build-uuid)))
-             build-uuids)
-        ", ")
-       ")")))))
-
 (define-method (datastore-list-allocation-plan-builds
                 (datastore <sqlite-datastore>)
                 .