Use fibers when processing new revisions

Just have one fiber at the moment, but this will enable using fibers for
parallelism in the future.

Fibers seemed to cause problems with the logging setup, which was a bit odd in
the first place. So move logging to the parent process which is better anyway.

guix-data-service/jobs.scm

@@ -20,11 +20,93 @@
   #:use-module (ice-9 format)
   #:use-module (ice-9 threads)
   #:use-module (ice-9 atomic)
+  #:use-module (ice-9 textual-ports)
+  #:use-module (squee)
+  #:use-module (guix-data-service utils)
+  #:use-module (guix-data-service database)
   #:use-module (guix-data-service jobs load-new-guix-revision)
-  #:export (process-jobs
+  #:export (log-for-job
+            count-log-parts
+            combine-log-parts!
+
+            process-jobs
 
             default-max-processes))
 
+(define (log-part-sequence-name job-id)
+  (simple-format #f "load_new_guix_revision_job_log_parts_id_seq_~A" job-id))
+
+(define (start-thread-for-process-output job-id)
+  (define (insert conn job_id s)
+    (exec-query
+     conn
+     (string-append "
+INSERT INTO load_new_guix_revision_job_log_parts (id, job_id, contents)
+VALUES (nextval('" (log-part-sequence-name job_id) "'), $1, $2)")
+     (list job_id s)))
+
+
+  (match (pipe)
+    ((port-to-read-from . port-to-write-to)
+
+     (setvbuf port-to-read-from 'line)
+     (setvbuf port-to-write-to 'line)
+
+     (let ((flags (fcntl port-to-read-from F_GETFL)))
+       (fcntl port-to-read-from F_SETFL (logior O_NONBLOCK flags)))
+     (let ((flags (fcntl port-to-write-to F_GETFL)))
+       (fcntl port-to-write-to F_SETFL (logior O_NONBLOCK flags)))
+
+     (call-with-new-thread
+      (lambda ()
+        (with-postgresql-connection
+         (simple-format #f "~A job logging" job-id)
+         (lambda (logging-conn)
+           (exec-query
+            logging-conn
+            (simple-format #f "CREATE SEQUENCE IF NOT EXISTS ~A"
+                           (log-part-sequence-name job-id)))
+           (exec-query
+            logging-conn
+            "DELETE FROM load_new_guix_revision_job_log_parts WHERE job_id = $1"
+            (list job-id))
+
+           (let loop ((line (get-line port-to-read-from)))
+             (let ((line-with-newline
+                    (string-append line "\n")))
+               (catch #t
+                 (lambda ()
+                   (insert logging-conn job-id line-with-newline)
+                   (display line-with-newline))
+                 (lambda (key . args)
+                   (display
+                    (simple-format
+                     #f
+                     "
+error: ~A: ~A
+error: could not insert log part: '~A'\n\n"
+                     key args line))
+                   (catch #t
+                     (lambda ()
+                       (insert
+                        logging-conn
+                        job-id
+                        (simple-format
+                         #f
+                         "
+guix-data-service: error: missing log line: ~A
+\n" key)))
+                     (lambda _
+                       #t)))))
+             (loop (get-line port-to-read-from)))))))
+
+     port-to-write-to)))
+
+(define (cleanup-logging conn job-id)
+  (drop-log-parts-sequence conn job-id)
+  (with-time-logging "vacuuming log parts"
+    (vacuum-log-parts-table conn)))
+
 (define* (process-jobs conn #:key max-processes
                        latest-branch-revision-max-processes
                        skip-system-tests?)
@@ -32,29 +114,137 @@
     (fetch-unlocked-jobs conn))
 
   (define (process-job job-id)
-    (apply execlp
+    (let ((log-port (start-thread-for-process-output job-id)))
-           "guix-data-service-process-job"
+      (spawn
-           "guix-data-service-process-job"
+       "guix-data-service-process-job"
-           job-id
+       (cons* "guix-data-service-process-job"
-           (if skip-system-tests?
+              job-id
-               '("--skip-system-tests")
+              (if skip-system-tests?
-               '())))
+                  '("--skip-system-tests")
+                  '()))
+       #:output log-port
+       #:error log-port)))
+
+  (define (post-job job-id)
+    (when (> (count-log-parts conn job-id)
+             0)
+      (combine-log-parts! conn job-id)
+      (cleanup-logging conn job-id)))
 
   (define (handle-job-failure job-id)
     (record-job-event conn job-id "failure")
     (display (simple-format #f "recording failure for job ~A\n" job-id)
-             (current-error-port))
+             (current-error-port)))
-    (when (> (count-log-parts conn job-id)
-             0)
-      (combine-log-parts! conn job-id)))
 
   (process-jobs-concurrently fetch-new-jobs
                              process-job
+                             post-job
                              handle-job-failure
                              #:max-processes max-processes
                              #:priority-max-processes
                              latest-branch-revision-max-processes))
 
+
+(define* (log-for-job conn job-id
+                      #:key
+                      character-limit
+                      start-character)
+  (define (sql-html-escape s)
+    (string-append
+     "replace("
+     (string-append
+      "replace("
+      (string-append
+       "replace("
+       s
+       ",'&','&')")
+      ",'<','&lt;')")
+     ",'>','&gt;')"))
+
+  (define (get-characters s)
+    (if start-character
+        (simple-format #f "substr(~A, ~A, ~A)"
+                       s start-character
+                       character-limit)
+        (simple-format #f "right(~A, ~A)" s character-limit)))
+
+  (define log-query
+    (string-append
+     "SELECT "
+     (sql-html-escape (get-characters "contents"))
+     " FROM load_new_guix_revision_job_logs"
+     " WHERE job_id = $1 AND contents IS NOT NULL"))
+
+  (define parts-query
+    (string-append
+     "SELECT "
+     (sql-html-escape
+      (get-characters "STRING_AGG(contents, '' ORDER BY id ASC)"))
+     " FROM load_new_guix_revision_job_log_parts WHERE job_id = $1"))
+
+  (match (exec-query conn log-query (list job-id))
+    (((contents))
+     contents)
+    (()
+     (match (exec-query conn parts-query (list job-id))
+       (((contents))
+        contents)))))
+
+(define (count-log-parts conn job-id)
+  (match (exec-query
+          conn
+          "
+SELECT COUNT(*)
+FROM load_new_guix_revision_job_log_parts
+WHERE job_id = $1"
+          (list job-id))
+    (((id))
+     (string->number id))))
+
+(define (combine-log-parts! conn job-id)
+  (with-postgresql-transaction
+   conn
+   (lambda (conn)
+     (exec-query
+      conn
+      (string-append
+       "
+UPDATE load_new_guix_revision_job_logs SET contents = (
+  SELECT STRING_AGG(contents, '' ORDER BY id ASC)
+  FROM load_new_guix_revision_job_log_parts
+  WHERE job_id = $1
+  GROUP BY job_id
+)
+WHERE job_id = $1")
+      (list job-id))
+     (exec-query
+      conn
+      "DELETE FROM load_new_guix_revision_job_log_parts WHERE job_id = $1"
+      (list job-id)))))
+
+(define (drop-log-parts-sequence conn job-id)
+  (with-postgresql-transaction
+   conn
+   (lambda (conn)
+     (exec-query conn
+                 "SET LOCAL lock_timeout = '10s'")
+     (with-exception-handler
+         (lambda (exn)
+           (simple-format (current-error-port)
+                          "error when dropping sequence: ~A"
+                          exn))
+       (lambda ()
+         (exec-query conn
+                     (string-append
+                      "DROP SEQUENCE IF EXISTS "
+                      (log-part-sequence-name job-id))))
+       #:unwind? #t))))
+
+(define (vacuum-log-parts-table conn)
+  (exec-query
+   conn
+   "VACUUM load_new_guix_revision_job_log_parts"))
+
 (define default-max-processes
   (max (round (/ (current-processor-count)
                  4))
@@ -67,6 +257,7 @@
 (define* (process-jobs-concurrently
           fetch-new-jobs
           process-job
+          post-job
           handle-job-failure
           #:key
           (max-processes default-max-processes)
@@ -108,9 +299,10 @@
            ;; No process to wait for
            #f)
           ((pid . status)
-           (unless (eq? status 0)
+           (match (hash-ref processes pid)
-             (match (hash-ref processes pid)
+             ((_ (id))
-               ((_ (id))
+              (post-job id)
+              (unless (eq? status 0)
                 (simple-format (current-error-port)
                                "pid ~A (job: ~A) failed with status ~A\n"
                                pid id status)
@@ -161,20 +353,6 @@
         (kill pid SIGTERM)))
      processes))
 
-  (define (fork-and-process-job job-args)
-    (match (primitive-fork)
-      (0
-       (dynamic-wind
-         (const #t)
-         (lambda ()
-           (apply process-job job-args))
-         (lambda ()
-           (primitive-exit 127))))
-      (pid
-       (hashv-set! processes pid
-                   (list (current-time) job-args))
-       #t)))
-
   (define exit?
     (make-atomic-box #f))
 
@@ -206,6 +384,9 @@
                       (if priority?
                           priority-max-processes
                           max-processes))
-               (fork-and-process-job (list job-id))))))
+               (let ((pid (process-job job-id)))
+                 (peek "PID" pid)
+                 (hashv-set! processes pid
+                             (list (current-time) (list job-id))))))))
         jobs)))
     (sleep 15)))

guix-data-service/jobs/load-new-guix-revision.scm

@@ -19,13 +19,16 @@
   #:use-module (srfi srfi-1)
   #:use-module (srfi srfi-11)
   #:use-module (srfi srfi-43)
+  #:use-module (srfi srfi-71)
   #:use-module (ice-9 match)
   #:use-module (ice-9 threads)
+  #:use-module (ice-9 custom-ports)
   #:use-module (ice-9 textual-ports)
   #:use-module (ice-9 hash-table)
   #:use-module (rnrs exceptions)
   #:use-module (json)
   #:use-module (squee)
+  #:use-module (fibers)
   #:use-module (guix monads)
   #:use-module (guix store)
   #:use-module (guix channels)
@@ -61,10 +64,7 @@
   #:use-module (guix-data-service model package-metadata)
   #:use-module (guix-data-service model derivation)
   #:use-module (guix-data-service model system-test)
-  #:export (log-for-job
+  #:export (fetch-unlocked-jobs
-            count-log-parts
-            combine-log-parts!
-            fetch-unlocked-jobs
             process-load-new-guix-revision-job
             select-load-new-guix-revision-job-metrics
             select-job-for-commit
@@ -77,259 +77,6 @@
             enqueue-load-new-guix-revision-job
             most-recent-n-load-new-guix-revision-jobs))
 
-(define (log-part-sequence-name job-id)
-  (simple-format #f "load_new_guix_revision_job_log_parts_id_seq_~A" job-id))
-
-(define* (log-port job-id conn
-                   #:key
-                   delete-existing-log-parts?
-                   real-output-port)
-  (define output-port
-    (or real-output-port
-        (current-output-port)))
-
-  (define buffer "")
-
-  (define (insert job_id s)
-    (exec-query
-     conn
-     (string-append
-      "
-INSERT INTO load_new_guix_revision_job_log_parts (id, job_id, contents)
-VALUES (nextval('" (log-part-sequence-name job_id) "'), $1, $2)")
-     (list job_id s)))
-
-  (define (log-string s)
-    (if (string-contains s "\n")
-        (let ((output (string-append buffer s)))
-          (set! buffer "") ; clear the buffer
-          (catch #t
-            (lambda ()
-              (insert job-id output)
-              (display output output-port))
-            (lambda (key . args)
-              (display
-               (simple-format
-                #f
-                "
-error: ~A: ~A
-error: could not insert log part: '~A'\n\n"
-                key args output)
-               output-port)
-              (catch #t
-                (lambda ()
-                  (insert
-                   job-id
-                   (simple-format
-                    #f
-                    "
-guix-data-service: error: missing log line: ~A
-\n" key)))
-                (lambda ()
-                  #t)))))
-        (set! buffer (string-append buffer s))))
-
-  (exec-query
-   conn
-   (simple-format #f "CREATE SEQUENCE IF NOT EXISTS ~A"
-                  (log-part-sequence-name job-id)))
-  (when delete-existing-log-parts?
-    ;; TODO, this is useful when re-running jobs, but I'm not sure that should
-    ;; be a thing, jobs should probably be only attempted once.
-    (exec-query
-     conn
-     "DELETE FROM load_new_guix_revision_job_log_parts WHERE job_id = $1"
-     (list job-id)))
-
-  (let ((port
-         (make-soft-port
-          (vector (lambda (c)
-                    (set! buffer (string-append buffer (string c))))
-                  log-string
-                  (lambda ()
-                    (force-output output-port))
-                  #f ; fetch one character
-                  (lambda ()
-                    ;; close port
-                    #f)
-                  #f) ; number of characters that can be read
-          "w")))
-    (setvbuf port 'line)
-    port))
-
-(define (setup-port-for-inferior-error-output job-id real-output-port)
-  (define (insert conn job_id s)
-    (exec-query
-     conn
-     (string-append "
-INSERT INTO load_new_guix_revision_job_log_parts (id, job_id, contents)
-VALUES (nextval('" (log-part-sequence-name job_id) "'), $1, $2)")
-     (list job_id s)))
-
-  (match (pipe)
-    ((port-to-read-from . port-to-write-to)
-
-     (setvbuf port-to-read-from 'line)
-     (setvbuf port-to-write-to 'line)
-     (call-with-new-thread
-      (lambda ()
-        (catch 'system-error
-          (lambda ()
-            (set-thread-name "inferior logging"))
-          (const #t))
-
-        (with-postgresql-connection
-         (simple-format #f "~A inferior error logging" job-id)
-         (lambda (logging-conn)
-           (let loop ((line (get-line port-to-read-from)))
-             (let ((line-with-newline
-                    (string-append line "\n")))
-               (catch #t
-                 (lambda ()
-                   (insert logging-conn job-id line-with-newline)
-                   (display line-with-newline real-output-port))
-                 (lambda (key . args)
-                   (display
-                    (simple-format
-                     #f
-                     "
-error: ~A: ~A
-error: could not insert log part: '~A'\n\n"
-                     key args line)
-                    real-output-port)
-                   (catch #t
-                     (lambda ()
-                       (insert
-                        logging-conn
-                        job-id
-                        (simple-format
-                         #f
-                         "
-guix-data-service: error: missing log line: ~A
-\n" key)))
-                     (lambda ()
-                       #t)))))
-             (loop (get-line port-to-read-from)))))))
-
-     port-to-write-to)))
-
-(define real-error-port
-  (make-parameter (current-error-port)))
-
-(define inferior-error-port
-  (make-parameter (current-error-port)))
-
-(define* (log-for-job conn job-id
-                      #:key
-                      character-limit
-                      start-character)
-  (define (sql-html-escape s)
-    (string-append
-     "replace("
-     (string-append
-      "replace("
-      (string-append
-       "replace("
-       s
-       ",'&','&')")
-      ",'<','&lt;')")
-     ",'>','&gt;')"))
-
-  (define (get-characters s)
-    (if start-character
-        (simple-format #f "substr(~A, ~A, ~A)"
-                       s start-character
-                       character-limit)
-        (simple-format #f "right(~A, ~A)" s character-limit)))
-
-  (define log-query
-    (string-append
-     "SELECT "
-     (sql-html-escape (get-characters "contents"))
-     " FROM load_new_guix_revision_job_logs"
-     " WHERE job_id = $1 AND contents IS NOT NULL"))
-
-  (define parts-query
-    (string-append
-     "SELECT "
-     (sql-html-escape
-      (get-characters "STRING_AGG(contents, '' ORDER BY id ASC)"))
-     " FROM load_new_guix_revision_job_log_parts WHERE job_id = $1"))
-
-  (match (exec-query conn log-query (list job-id))
-    (((contents))
-     contents)
-    (()
-     (match (exec-query conn parts-query (list job-id))
-       (((contents))
-        contents)))))
-
-(define (insert-empty-log-entry conn job-id)
-  (exec-query
-   conn
-   "DELETE FROM load_new_guix_revision_job_logs WHERE job_id = $1"
-   (list job-id))
-  (exec-query
-   conn
-   "INSERT INTO load_new_guix_revision_job_logs (job_id, contents) VALUES
-($1, NULL)"
-   (list job-id)))
-
-(define (count-log-parts conn job-id)
-  (match (exec-query
-          conn
-          "
-SELECT COUNT(*)
-FROM load_new_guix_revision_job_log_parts
-WHERE job_id = $1"
-          (list job-id))
-    (((id))
-     (string->number id))))
-
-(define (combine-log-parts! conn job-id)
-  (with-postgresql-transaction
-   conn
-   (lambda (conn)
-     (exec-query
-      conn
-      (string-append
-       "
-UPDATE load_new_guix_revision_job_logs SET contents = (
-  SELECT STRING_AGG(contents, '' ORDER BY id ASC)
-  FROM load_new_guix_revision_job_log_parts
-  WHERE job_id = $1
-  GROUP BY job_id
-)
-WHERE job_id = $1")
-      (list job-id))
-     (exec-query
-      conn
-      "DELETE FROM load_new_guix_revision_job_log_parts WHERE job_id = $1"
-      (list job-id)))))
-
-(define (drop-log-parts-sequence conn job-id)
-  (with-postgresql-transaction
-   conn
-   (lambda (conn)
-     (exec-query conn
-                 "SET LOCAL lock_timeout = '10s'")
-     (with-exception-handler
-         (lambda (exn)
-           (simple-format (current-error-port)
-                          "error when dropping sequence: ~A"
-                          exn))
-       (lambda ()
-         (exec-query conn
-                     (string-append
-                      "DROP SEQUENCE IF EXISTS "
-                      (log-part-sequence-name job-id))))
-       #:unwind? #t))))
-
-(define (vacuum-log-parts-table conn)
-  (exec-query
-   conn
-   "VACUUM load_new_guix_revision_job_log_parts"))
-
 (define inferior-package-id
   (@@ (guix inferior) inferior-package-id))
 
@@ -845,9 +592,12 @@ WHERE job_id = $1")
                 (a-name (inferior-package-name a))
                 (b-name (inferior-package-name b))
                 (a-version (inferior-package-version a))
-                (b-version (inferior-package-version b)))
+                (b-version (inferior-package-version b))
+                (a-replacement (inferior-package-replacement a))
+                (b-replacement (inferior-package-replacement b)))
            (if (and (string=? a-name b-name)
-                    (string=? a-version b-version))
+                    (string=? a-version b-version)
+                    (eq? a-replacement b-replacement))
                (begin
                  (simple-format (current-error-port)
                                 "warning: ignoring duplicate package: ~A (~A)\n"
@@ -926,7 +676,17 @@ WHERE job_id = $1")
 
     (list->vector deduplicated-packages)))
 
-(define* (all-inferior-packages-data inf packages #:key (process-replacements? #t))
+(define* (all-inferior-packages-data inf packages)
+  (define inferior-package-id->packages-index-hash-table
+    (let ((hash-table (make-hash-table)))
+      (vector-for-each
+       (lambda (i pkg)
+         (hash-set! hash-table
+                    (inferior-package-id pkg)
+                    i))
+       packages)
+      hash-table))
+
   (let* ((package-license-data
           (with-time-logging "fetching inferior package license metadata"
             (inferior-packages->license-data inf)))
@@ -944,27 +704,21 @@ WHERE job_id = $1")
                        (cdr translated-package-descriptions-and-synopsis))))
              packages)))
          (package-replacement-data
-          (if process-replacements?
+          (vector-map
-              (vector-map
+           (lambda (_ pkg)
-               (lambda (_ package)
+             (let ((replacement (inferior-package-replacement pkg)))
-                 (let ((replacement (inferior-package-replacement package)))
+               (if replacement
-                   (if replacement
+                   ;; I'm not sure if replacements can themselves be
-                       ;; I'm not sure if replacements can themselves be
+                   ;; replaced, but I do know for sure that there are
-                       ;; replaced, but I do know for sure that there are
+                   ;; infinite chains of replacements (python(2)-urllib3
-                       ;; infinite chains of replacements (python(2)-urllib3
+                   ;; in 7c4c781aa40c42d4cd10b8d9482199f3db345e1b for
-                       ;; in 7c4c781aa40c42d4cd10b8d9482199f3db345e1b for
+                   ;; example).
-                       ;; example).
+                   ;;
-                       ;;
+                   ;; So this might be #f in these cases
-                       ;; This code currently just capures the first level
+                   (hash-ref inferior-package-id->packages-index-hash-table
-                       ;; of replacements
+                             (inferior-package-id pkg))
-                       (first
+                   #f)))
-                        (all-inferior-packages-data
+           packages)))
-                         inf
-                         (vector replacement)
-                         #:process-replacements? #f))
-                       #f)))
-               packages)
-              #f)))
 
     `((names        . ,(vector-map (lambda (_ pkg) (inferior-package-name pkg))
                                    packages))
@@ -972,53 +726,57 @@ WHERE job_id = $1")
                                    packages))
       (license-data . ,package-license-data)
       (metadata     . ,package-metadata)
-      (replacemnets . ,package-replacement-data))))
+      (replacements . ,package-replacement-data))))
 
 (define (insert-packages conn inferior-packages-data)
-  (let*-values
+  (let* ((names (assq-ref inferior-packages-data 'names))
-      (((package-license-set-ids)
+         (versions (assq-ref inferior-packages-data 'versions))
-        (inferior-packages->license-set-ids
+         (package-license-set-ids
-         conn
+          (inferior-packages->license-set-ids
-         (inferior-packages->license-id-lists
-          conn
-          ;; TODO Don't needlessly convert
-          (vector->list
-           (assq-ref inferior-packages-data 'license-data)))))
-       ((all-package-metadata-ids new-package-metadata-ids)
-        (with-time-logging "inserting package metadata entries"
-          (inferior-packages->package-metadata-ids
            conn
-           ;; TODO Don't needlessly convert
+           (inferior-packages->license-id-lists
-           (vector->list
+            conn
-            (assq-ref inferior-packages-data 'metadata))
+            ;; TODO Don't needlessly convert
-           package-license-set-ids)))
+            (vector->list
-       ((replacement-ids)
+             (assq-ref inferior-packages-data 'license-data)))))
-        (or (and=> (assq-ref inferior-packages-data 'replacements)
+         (all-package-metadata-ids
-                   (lambda (all-replacement-data)
+          new-package-metadata-ids
-                     (with-time-logging "inserting package replacements"
+          (with-time-logging "inserting package metadata entries"
-                       (vector-map
+            (inferior-packages->package-metadata-ids
-                        (lambda (_ replacement-data)
+             conn
-                          (if replacement-data
+             ;; TODO Don't needlessly convert
-                              (first
+             (vector->list
-                               (insert-packages conn (list replacement-data)))
+              (assq-ref inferior-packages-data 'metadata))
-                              (cons "integer" NULL)))
+             package-license-set-ids)))
-                        all-replacement-data))))
+         (replacement-package-ids
-            (make-vector (length package-license-set-ids)
+          (vector-map
-                         (cons "integer" NULL)))))
+           (lambda (_ package-index-or-false)
+             (if package-index-or-false
+                 (first
+                  (inferior-packages->package-ids
+                   conn
+                   (list (list (vector-ref names package-index-or-false)
+                               (vector-ref versions package-index-or-false)
+                               (list-ref all-package-metadata-ids
+                                         package-index-or-false)
+                               (cons "integer" NULL)))))
+                 (cons "integer" NULL)))
+           (assq-ref inferior-packages-data 'replacements))))
 
     (unless (null? new-package-metadata-ids)
-      (with-time-logging "fetching package metadata tsvector entries"
+      (with-time-logging "inserting package metadata tsvector entries"
         (insert-package-metadata-tsvector-entries
          conn new-package-metadata-ids)))
 
-    (with-time-logging "getting package-ids"
+    (with-time-logging "getting package-ids (without replacements)"
-      (inferior-packages->package-ids
+      (list->vector
-       conn
+       (inferior-packages->package-ids
-       ;; TODO Do this more efficiently
+        conn
-       (zip (vector->list (assq-ref inferior-packages-data 'names))
+        ;; TODO Do this more efficiently
-            (vector->list (assq-ref inferior-packages-data 'versions))
+        (zip (vector->list names)
-            all-package-metadata-ids
+             (vector->list versions)
-            (vector->list replacement-ids))))))
+             all-package-metadata-ids
+             (vector->list replacement-package-ids)))))))
 
 (define (insert-lint-warnings conn
                               package-ids
@@ -1082,11 +840,11 @@ WHERE job_id = $1")
               conn
               derivations-vector)))
 
-          (insert-package-derivations conn
+       (insert-package-derivations conn
-                                 (car system-and-target)
+                                   (car system-and-target)
-                                 (or (cdr system-and-target) "")
+                                   (or (cdr system-and-target) "")
-                                 package-ids
+                                   package-ids
-                                 derivation-ids)))
+                                   derivation-ids)))
    inferior-packages-system-and-target-to-derivations-alist))
 
 (define guix-store-path
@@ -1213,8 +971,7 @@ WHERE job_id = $1")
                        "SSL_CERT_DIR=" (nss-certs-store-path store))))
                (begin
                  (simple-format #t "debug: using open-inferior\n")
-                 (open-inferior (guix-store-path store)
+                 (open-inferior (guix-store-path store))))))
-                                #:error-port (inferior-error-port))))))
 
       (define (start-inferior-and-return-derivation-file-names)
         ;; /etc is only missing if open-inferior/container has been used
@@ -1344,8 +1101,7 @@ WHERE job_id = $1")
                                           '("/gnu/store"))
                  (begin
                    (simple-format #t "debug: using open-inferior\n")
-                   (open-inferior store-path
+                   (open-inferior store-path)))))
-                                  #:error-port (inferior-error-port))))))
     (inferior-eval '(use-modules (srfi srfi-1)
                                  (srfi srfi-34)
                                  (guix grafts)
@@ -1400,8 +1156,7 @@ WHERE job_id = $1")
                     (begin
                       (setenv "GUIX_LOCPATH" guix-locpath)
                       (simple-format #t "debug: using open-inferior\n")
-                      (open-inferior store-path
+                      (open-inferior store-path))))))
-                                     #:error-port (inferior-error-port)))))))
     (setenv "GUIX_LOCPATH" guix-locpath) ; restore GUIX_LOCPATH
 
     (when (eq? inf #f)
@@ -2040,44 +1795,6 @@ SKIP LOCKED")
 
 (prevent-inlining-for-tests with-store-connection)
 
-(define (setup-logging id thunk)
-  (let* ((previous-output-port (current-output-port))
-         (previous-error-port (current-error-port))
-         (result
-          (with-postgresql-connection
-           (simple-format #f "load-new-guix-revision ~A logging" id)
-           (lambda (logging-conn)
-             (insert-empty-log-entry logging-conn id)
-             (let ((logging-port
-                    (log-port id logging-conn
-                              #:delete-existing-log-parts? #t)))
-               (set-current-output-port logging-port)
-               (set-current-error-port logging-port)
-               (let ((result
-                      (parameterize ((current-build-output-port logging-port)
-                                     (real-error-port previous-error-port)
-                                     (inferior-error-port
-                                      (setup-port-for-inferior-error-output
-                                       id previous-error-port)))
-                        (thunk))))
-                 (set-current-output-port previous-output-port)
-                 (set-current-error-port previous-error-port)
-
-                 ;; This can happen with GC, so do it explicitly
-                 (close-port logging-port)
-
-                 (combine-log-parts! logging-conn id)
-
-                 result))))))
-    result))
-
-(define (cleanup-logging id conn)
-  (drop-log-parts-sequence conn id)
-  (with-time-logging "vacuuming log parts"
-    (vacuum-log-parts-table conn)))
-
-(prevent-inlining-for-tests setup-logging)
-
 (define* (process-load-new-guix-revision-job id #:key skip-system-tests?)
   (with-postgresql-connection
    (simple-format #f "load-new-guix-revision ~A" id)
@@ -2104,38 +1821,30 @@ SKIP LOCKED")
 
         (if (eq?
              (with-time-logging (string-append "processing revision " commit)
-               (setup-logging
+               (with-exception-handler
-                id
+                   (const #f)
-                (lambda ()
+                 (lambda ()
-                  (with-exception-handler
+                   (with-throw-handler #t
-                      (const #f)
+                     (lambda ()
-                    (lambda ()
+                       (with-store-connection
-                      (with-throw-handler #t
+                        (lambda (store)
-                        (lambda ()
+                          (load-new-guix-revision conn
-                          (with-store-connection
+                                                  store
-                           (lambda (store)
+                                                  git-repository-id
-                             (load-new-guix-revision conn
+                                                  commit
-                                                     store
+                                                  #:skip-system-tests? #t))))
-                                                     git-repository-id
+                     (lambda (key . args)
-                                                     commit
+                       (simple-format (current-error-port)
-                                                     #:skip-system-tests?
+                                      "error: load-new-guix-revision: ~A ~A\n"
-                                                     skip-system-tests?))))
+                                      key args)
-                        (lambda (key . args)
+                       (backtrace))))
-                          (simple-format (current-error-port)
+                 #:unwind? #t))
-                                         "error: load-new-guix-revision: ~A ~A\n"
-                                         key args)
-                          (backtrace))))
-                    #:unwind? #t))))
              #t)
             (begin
               (record-job-succeeded conn id)
               (record-job-event conn id "success")
               (exec-query conn "COMMIT")
 
-              (with-time-logging
-                  "cleanup logging"
-                (cleanup-logging id conn))
-
               (with-time-logging
                   "vacuuming package derivations by guix revision range table"
                 (vacuum-package-derivations-table conn))
@@ -2170,10 +1879,6 @@ SKIP LOCKED")
               (exec-query conn "ROLLBACK")
               (record-job-event conn id "failure")
 
-              (with-time-logging
-                  "cleanup logging"
-                (cleanup-logging id conn))
-
               #f)))
        (()
         (exec-query conn "ROLLBACK")

guix-data-service/model/package.scm

@@ -264,12 +264,11 @@ INSERT INTO packages (name, version, package_metadata_id) VALUES "
 RETURNING id"))
 
 (define (inferior-packages->package-ids conn package-entries)
-  (list->vector
+  (insert-missing-data-and-return-all-ids
-   (insert-missing-data-and-return-all-ids
+   conn
-    conn
+   "packages"
-    "packages"
+   '(name version package_metadata_id replacement_package_id)
-    '(name version package_metadata_id replacement_package_id)
+   package-entries))
-    package-entries)))
 
 (define (select-package-versions-for-revision conn
                                               commit

guix-data-service/web/jobs/controller.scm

@@ -23,6 +23,7 @@
   #:use-module (guix-data-service web controller)
   #:use-module (guix-data-service web query-parameters)
   #:use-module (guix-data-service web util)
+  #:use-module (guix-data-service jobs)
   #:use-module (guix-data-service jobs load-new-guix-revision)
   #:use-module (guix-data-service web jobs html)
   #:export (jobs-controller))

scripts/guix-data-service-process-job.in

@@ -25,6 +25,8 @@
 (use-modules (srfi srfi-1)
              (srfi srfi-37)
              (ice-9 match)
+             (ice-9 suspendable-ports)
+             (fibers)
              (guix-data-service database)
              (guix-data-service data-deletion)
              (guix-data-service model package-derivation-by-guix-revision-range)
@@ -38,12 +40,21 @@
 ;; Make stack traces more useful
 (setenv "COLUMNS" "256")
 
+(install-suspendable-ports!)
+
 (define %options
   (list (option '("skip-system-tests") #f #f
                 (lambda (opt name _ result)
-                  (alist-cons 'skip-system-tests #t result)))))
+                  (alist-cons 'skip-system-tests #t result)))
+        (option '("parallelism") #t #f
+                (lambda (opt name arg result)
+                  (alist-cons 'parallelism
+                              (string->number arg)
+                              (alist-delete 'parallelism
+                                            result))))))
 
-(define %default-options '())
+(define %default-options
+  '((parallelism . 1)))
 
 (define (parse-options args)
   (args-fold
@@ -62,6 +73,11 @@
 (let ((opts (parse-options (cdr (program-arguments)))))
   (match (assq-ref opts 'arguments)
     ((job)
-     (process-load-new-guix-revision-job
+     (run-fibers
-      job
+      (lambda ()
-      #:skip-system-tests? (assq-ref opts 'skip-system-tests)))))
+        (process-load-new-guix-revision-job
+         job
+         #:skip-system-tests? (assq-ref opts 'skip-system-tests)))
+      #:hz 0
+      #:parallelism (assq-ref opts 'parallelism)
+      #:drain? #t))))

tests/jobs-load-new-guix-revision.scm

@@ -48,43 +48,37 @@
 
        (mock
         ((guix-data-service jobs load-new-guix-revision)
-         setup-logging
+         channel-derivations-by-system->guix-store-item
-         (lambda (conn thunk)
+         (lambda (store channel-derivations-by-system)
-           (thunk)))
+           "/gnu/store/test"))
 
         (mock
          ((guix-data-service jobs load-new-guix-revision)
-          channel-derivations-by-system->guix-store-item
+          extract-information-from
-          (lambda (store channel-derivations-by-system)
+          (lambda* (conn store guix-revision-id commit
-            "/gnu/store/test"))
+                         guix-source store-path
+                         #:key skip-system-tests?)
+            #t))
 
          (mock
-          ((guix-data-service jobs load-new-guix-revision)
+          ((guix-data-service model channel-instance)
-           extract-information-from
+           insert-channel-instances
-           (lambda* (conn store guix-revision-id commit
+           (lambda (conn guix-revision-id derivations-by-system)
-                          guix-source store-path
-                          #:key skip-system-tests?)
              #t))
 
           (mock
-           ((guix-data-service model channel-instance)
+           ((guix channels)
-            insert-channel-instances
+            channel-news-for-commit
-            (lambda (conn guix-revision-id derivations-by-system)
+            (lambda (channel commit)
-              #t))
+              '()))
 
-           (mock
+           (match (enqueue-load-new-guix-revision-job
-            ((guix channels)
+                   conn
-             channel-news-for-commit
+                   (git-repository-url->git-repository-id conn "test-url")
-             (lambda (channel commit)
+                   "test-commit"
-               '()))
+                   "test-source")
-
+             ((id)
-            (match (enqueue-load-new-guix-revision-job
+              (process-load-new-guix-revision-job id))))))))))
-                    conn
-                    (git-repository-url->git-repository-id conn "test-url")
-                    "test-commit"
-                    "test-source")
-              ((id)
-               (process-load-new-guix-revision-job id)))))))))))
 
    (exec-query conn "TRUNCATE guix_revisions CASCADE")
    (exec-query conn "TRUNCATE load_new_guix_revision_jobs CASCADE")