Move most functionality out of the start script

As this can make debugging easier.

1 files changed, 258 insertions, 0 deletions

diff --git a/nar-herder/server.scm b/nar-herder/server.scm
index 525d70b..7aa4115 100644
--- a/nar-herder/server.scm
+++ b/nar-herder/server.scm
@@ -28,26 +28,38 @@
   #:use-module (ice-9 binary-ports)
   #:use-module (rnrs bytevectors)
   #:use-module (web uri)
+  #:use-module (web client)
   #:use-module (web response)
   #:use-module (web request)
   #:use-module (logging logger)
   #:use-module (fibers)
+  #:use-module (fibers channels)
+  #:use-module (fibers scheduler)
+  #:use-module (fibers conditions)
+  #:use-module (fibers operations)
   #:use-module (prometheus)
   #:use-module (json)
   #:use-module ((system foreign)
                 #:select (bytevector->pointer pointer->bytevector))
   #:use-module (guix store)
   #:use-module (guix base32)
+  #:use-module (guix progress)
   #:use-module (guix serialization)
   #:use-module ((guix utils)
                 #:select (decompressed-port))
   #:use-module ((guix build utils)
                 #:select (dump-port))
+  #:use-module ((guix build syscalls) #:select (set-thread-name))
   #:use-module (nar-herder database)
   #:use-module (nar-herder storage)
+  #:use-module (nar-herder utils)
+  #:use-module (nar-herder mirror)
+  #:use-module (nar-herder recent-changes)
+  #:use-module (nar-herder cached-compression)
   #:use-module (ice-9 textual-ports)
   #:export (%compression-options
 
+            run-nar-herder-service
             make-request-handler))
 
 (define %compression-options
@@ -537,3 +549,249 @@
        (values (build-response #:code 404)
                "404")))))
 
+(define* (run-nar-herder-service opts lgr)
+  (define (download-database)
+    (let ((database-uri
+           (string->uri
+            (string-append (assq-ref opts 'mirror)
+                           "/latest-database-dump"))))
+      (with-fibers-port-timeouts
+       (lambda ()
+         (call-with-values
+             (lambda ()
+               (simple-format (current-error-port)
+                              "starting downloading the database\n")
+               (let ((port
+                      socket
+                      (open-socket-for-uri* database-uri)))
+                 (http-get database-uri
+                           #:port port
+                           #:streaming? #t)))
+           (lambda (response body)
+             (when (not (= (response-code response) 200))
+               (error "unable to fetch database from mirror"))
+
+             (let* ((reporter (progress-reporter/file
+                               (uri->string database-uri)
+                               (response-content-length response)
+                               (current-error-port)))
+                    (port
+                     (progress-report-port
+                      reporter
+                      body
+                      #:download-size (response-content-length response))))
+
+               (call-with-output-file (assq-ref opts 'database)
+                 (lambda (output-port)
+                   (dump-port port output-port)))
+
+               (close-port port))
+
+             (simple-format (current-error-port)
+                            "finished downloading the database\n"))))
+       #:timeout 30)))
+
+  (define metrics-registry
+    (make-metrics-registry
+     #:namespace
+     "narherder"))
+
+  (and=>
+   (assq-ref opts 'mirror)
+   (lambda (mirror)
+     (let ((database-file (assq-ref opts 'database)))
+       (if (file-exists? database-file)
+           (begin
+             ;; TODO Open the database, and check if the
+             ;; latest changes in the database are visible on
+             ;; the source to mirror. If they're not, then
+             ;; delete the database and download it to get
+             ;; back in sync
+
+             #f)
+           (download-database)))))
+
+  (let* ((database (setup-database (assq-ref opts 'database)
+                                   metrics-registry))
+         (canonical-storage (and=> (assq-ref opts 'storage)
+                                   canonicalize-path))
+
+         (enabled-cached-compressions
+          (let ((explicit-cached-compression-directories
+                 (filter-map
+                  (match-lambda
+                    (('cached-compression-directory . details) details)
+                    (_ #f))
+                  opts))
+                (cached-compression-directories-max-sizes
+                 (filter-map
+                  (match-lambda
+                    (('cached-compression-directory-max-size . details) details)
+                    (_ #f))
+                  opts)))
+            (filter-map
+             (match-lambda
+               (('cached-compression . details)
+                (let ((compression
+                       (assq-ref details 'type)))
+                  (cons compression
+                        `(,@(alist-delete 'type details)
+                          (directory
+                           . ,(or (assq-ref explicit-cached-compression-directories
+                                            compression)
+                                  (simple-format #f "/var/cache/nar-herder/nar/~A"
+                                                 compression)))
+                          (directory-max-size
+                           . ,(assq-ref cached-compression-directories-max-sizes
+                                        compression))))))
+               (_ #f))
+             opts)))
+
+         (cached-compression-min-uses
+          (assq-ref opts 'cached-compression-min-uses))
+
+         (maybe-trigger-creation-of-compressed-nars
+          (if (null? enabled-cached-compressions)
+              #f
+              (make-maybe-trigger-creation-of-compressed-nars
+               database
+               metrics-registry
+               (or (assq-ref opts 'cached-compression-nar-source)
+                   canonical-storage)
+               enabled-cached-compressions
+               cached-compression-min-uses
+               #:cached-compression-workers
+               (assq-ref opts 'cached-compression-workers)))))
+
+    (if (string=? (assq-ref opts 'database-dump)
+                  "disabled")
+        (log-msg 'INFO "database dump disabled")
+        (when (not (file-exists? (assq-ref opts 'database-dump)))
+          (log-msg 'INFO "dumping database...")
+          (dump-database database (assq-ref opts 'database-dump))))
+
+    (let ((finished? (make-condition)))
+      (call-with-new-thread
+       (lambda ()
+         (catch 'system-error
+           (lambda ()
+             (set-thread-name "maintenance"))
+           (const #t))
+
+         (run-fibers
+          (lambda ()
+            (start-recent-change-removal-and-database-dump-fiber
+             database
+             (let ((filename (assq-ref opts 'database-dump)))
+               (if (string=? filename "disabled")
+                   #f
+                   filename))
+             (* 24 3600)                     ; 24 hours
+             (assq-ref opts 'recent-changes-limit))
+
+            (let ((mirror-channel
+                   (and=>
+                    (assq-ref opts 'mirror)
+                    (lambda (mirror)
+                      (start-fetch-changes-fiber database
+                                                 canonical-storage
+                                                 mirror
+                                                 metrics-registry)
+
+                      (when (assq-ref opts 'storage)
+                        (start-mirroring-fiber database
+                                               mirror
+                                               (assq-ref opts 'storage-limit)
+                                               canonical-storage
+                                               metrics-registry)))))
+                  (removal-channel
+                   (let ((nar-removal-criteria
+                          (filter-map
+                           (match-lambda
+                             ((key . val)
+                              (if (eq? key 'storage-nar-removal-criteria)
+                                  val
+                                  #f)))
+                           opts)))
+                     (when (and (assq-ref opts 'storage)
+                                (number? (assq-ref opts 'storage-limit))
+                                (not (null? nar-removal-criteria)))
+                       (start-nar-removal-fiber database
+                                                canonical-storage
+                                                (assq-ref opts 'storage-limit)
+                                                metrics-registry
+                                                nar-removal-criteria))))
+                  (addition-channel (make-channel)))
+
+              (spawn-fiber
+               (lambda ()
+                 (while #t
+                   (match (get-message addition-channel)
+                     (('addition file)
+                      (when mirror-channel
+                        (put-message mirror-channel
+                                     `(fetch ,file)))
+                      (when removal-channel
+                        (spawn-fiber
+                         (lambda ()
+                           (sleep 60)
+                           (put-message removal-channel
+                                        `(remove-from-storage ,file))
+                           (sleep (* 5 60))
+                           (put-message removal-channel
+                                        `(remove-from-storage ,file))
+                           (sleep (* 15 60))
+                           (put-message removal-channel
+                                        `(remove-from-storage ,file))
+                           (sleep 3600)
+                           (put-message removal-channel
+                                        `(remove-from-storage ,file))))))))))
+
+              (start-recent-change-listener-fiber
+               database
+               addition-channel
+               removal-channel))
+
+            (log-msg 'DEBUG "finished maintenance setup")
+            (wait finished?))
+          #:parallelism 1)))
+
+      (call-with-sigint
+       (lambda ()
+         (run-fibers
+          (lambda ()
+            (let* ((current (current-scheduler))
+                   (schedulers
+                    (cons current (scheduler-remote-peers current))))
+              (for-each
+               (lambda (i sched)
+                 (spawn-fiber
+                  (lambda ()
+                    (catch 'system-error
+                      (lambda ()
+                        (set-thread-name
+                         (string-append "fibers " (number->string i))))
+                      (const #t)))
+                  sched))
+               (iota (length schedulers))
+               schedulers))
+
+            (log-msg 'INFO "starting server, listening on "
+                     (assq-ref opts 'host) ":" (assq-ref opts 'port))
+
+            (run-server/patched
+             (make-request-handler database
+                                   canonical-storage
+                                   #:ttl (assq-ref opts 'narinfo-ttl)
+                                   #:negative-ttl (assq-ref opts 'narinfo-negative-ttl)
+                                   #:logger lgr
+                                   #:metrics-registry metrics-registry
+                                   #:maybe-trigger-creation-of-compressed-nars
+                                   maybe-trigger-creation-of-compressed-nars)
+             #:host (assq-ref opts 'host)
+             #:port (assq-ref opts 'port))
+
+            (wait finished?))
+          #:hz 0
+          #:parallelism 4))
+       finished?))))