Use recent changes to trigger nar mirroring and removal

As this should be more efficient than repeatedly looking through all
the nars in the database.

3 files changed, 164 insertions, 52 deletions

diff --git a/nar-herder/recent-changes.scm b/nar-herder/recent-changes.scm
index ed3bd34..0615cd8 100644
--- a/nar-herder/recent-changes.scm
+++ b/nar-herder/recent-changes.scm
@@ -18,12 +18,18 @@
 
 (define-module (nar-herder recent-changes)
   #:use-module (srfi srfi-1)
+  #:use-module (ice-9 match)
   #:use-module (ice-9 threads)
   #:use-module ((guix build syscalls)
                 #:select (set-thread-name))
   #:use-module (fibers)
+  #:use-module (fibers channels)
+  #:use-module (logging logger)
+  #:use-module (web uri)
+  #:use-module (guix narinfo)
   #:use-module (nar-herder database)
-  #:export (start-recent-change-removal-and-database-dump-fiber))
+  #:export (start-recent-change-removal-and-database-dump-fiber
+            start-recent-change-listener-fiber))
 
 (define (start-recent-change-removal-and-database-dump-fiber database
                                                              database-dump-filename
@@ -75,4 +81,64 @@
              (sleep check-interval)))
          #:unwind? #t)))))
 
+(define (start-recent-change-listener-fiber database
+                                            mirror-channel
+                                            removal-channel)
+  (define (process-addition-change change-details)
+    (let ((narinfo
+           (call-with-input-string
+               (assq-ref change-details 'data)
+             (lambda (port)
+               (read-narinfo port
+                             "https://narherderdummyvalue")))))
+      (for-each
+       (lambda (uri)
+         (log-msg 'DEBUG "processing recent change triggered fetch of " (uri-path uri))
+         (put-message mirror-channel (list 'fetch (uri-path uri))))
+       (narinfo-uris narinfo))))
+
+  (define (process-removal-change change-details)
+    (log-msg 'DEBUG "processing recent change triggered removal of "
+             (assq-ref change-details 'data))
+    (put-message removal-channel
+                 (list 'remove (assq-ref change-details 'data))))
+
+  (spawn-fiber
+   (lambda ()
+     (log-msg 'DEBUG "starting to listen for recent changes")
+     (let ((after-initial
+            (database-select-latest-recent-change-datetime database)))
+       (let loop ((after after-initial)
+                  (last-processed-recent-changes
+                   (database-select-recent-changes database after-initial)))
+         (sleep 10)
+
+         (match
+             (with-exception-handler
+                 (lambda (exn)
+                   (log-msg 'ERROR "exception in recent change listener " exn)
+                   #f)
+               (lambda ()
+                 (let ((recent-changes
+                        (database-select-recent-changes database after)))
+                   (for-each
+                    (lambda (change-details)
+                      (when (not (member change-details
+                                         last-processed-recent-changes))
+                        (let ((change (assq-ref change-details 'change)))
+                          (cond
+                           ((string=? change "addition")
+                            (process-addition-change change-details))
+                           ((string=? change "removal")
+                            (process-removal-change change-details))
+                           (else #f)))))
+                    recent-changes)
+
+                   recent-changes))
+               #:unwind? #t)
+           (#f (loop after '()))
+           (recent-changes
+            (loop (assq-ref (last recent-changes)
+                            'datetime)
+                  recent-changes))))))))
 
diff --git a/nar-herder/storage.scm b/nar-herder/storage.scm
index a88f0f1..44b326d 100644
--- a/nar-herder/storage.scm
+++ b/nar-herder/storage.scm
@@ -26,6 +26,7 @@
   #:use-module (web uri)
   #:use-module (web client)
   #:use-module (web response)
+  #:use-module (fibers)
   #:use-module (fibers channels)
   #:use-module (logging logger)
   #:use-module (logging port-log)
@@ -487,16 +488,37 @@ When VERIFY-CERTIFICATE? is true, verify HTTPS server certificates."
   (when (null? nar-removal-criteria)
     (error "must be some removal criteria"))
 
-  (spawn-fiber
-   (lambda ()
-     (while #t
-       (with-exception-handler
-           (lambda (exn)
-             (log-msg 'ERROR "nar removal pass failed " exn))
-         run-removal-pass
-         #:unwind? #t)
-
-       (sleep 300)))))
+  (let ((channel (make-channel)))
+    (spawn-fiber
+     (lambda ()
+       (while #t
+         (match (get-message channel)
+           ('full-pass
+            (with-exception-handler
+                (lambda (exn)
+                  (log-msg 'ERROR "nar removal pass failed " exn))
+              run-removal-pass
+              #:unwind? #t))
+           (('remove file)
+            (with-exception-handler
+                (lambda (exn)
+                  (log-msg 'ERROR "failed to remove " file ": " exn))
+              (lambda ()
+                (remove-nar-from-storage
+                 storage-root
+                 (uri-decode file))
+                (update-nar-files-metric metrics-registry
+                                         '()
+                                         #:removed-count 1))
+              #:unwind? #t))))))
+
+    (spawn-fiber
+     (lambda ()
+       (while #t
+         (put-message channel 'full-pass)
+         (sleep (* 60 60 24)))))
+
+    channel))
 
 (define (start-mirroring-fiber database mirror storage-limit storage-root
                                metrics-registry)
@@ -697,14 +719,32 @@ When VERIFY-CERTIFICATE? is true, verify HTTPS server certificates."
         (log-msg 'DEBUG "finished mirror pass (any change? " any-change? ")")
         any-change?)))
 
-  (spawn-fiber
-   (lambda ()
-     (while #t
-       (unless (with-exception-handler
-                   (lambda (exn)
-                     (log-msg 'ERROR "mirror pass failed " exn)
-                     #t)
-                 run-mirror-pass
-                 #:unwind? #t)
-         ;; Only sleep where there were no changes or exceptions
-         (sleep 300))))))
+  (let ((channel (make-channel)))
+    (spawn-fiber
+     (lambda ()
+       (while #t
+         (match (get-message channel)
+           ('full-pass
+            (with-exception-handler
+                (lambda (exn)
+                  (log-msg 'ERROR "mirror pass failed " exn))
+              run-mirror-pass
+              #:unwind? #t))
+           (('fetch file)
+            (with-exception-handler
+                (lambda (exn)
+                  (log-msg 'ERROR "failed to mirror " file ": " exn))
+              (lambda ()
+                (fetch-file file)
+                (update-nar-files-metric metrics-registry
+                                         '()
+                                         #:fetched-count 1))
+              #:unwind? #t))))))
+
+    (spawn-fiber
+     (lambda ()
+       (while #t
+         (put-message channel 'full-pass)
+         (sleep (* 60 60 24)))))
+
+    channel))
diff --git a/scripts/nar-herder.in b/scripts/nar-herder.in
index 0736592..5991d02 100644
--- a/scripts/nar-herder.in
+++ b/scripts/nar-herder.in
@@ -629,37 +629,43 @@
                 (* 24 3600)                     ; 24 hours
                 (assq-ref opts 'recent-changes-limit))
 
-               (and=> (assq-ref opts 'mirror)
-                      (lambda (mirror)
-                        (start-fetch-changes-fiber database
+               (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
-                                                   mirror
-                                                   metrics-registry)
-
-                        (when (assq-ref opts 'storage)
-                          (start-mirroring-fiber database
-                                                 mirror
-                                                 (assq-ref opts 'storage-limit)
-                                                 canonical-storage
-                                                 metrics-registry))))
-
-
-               (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)))
+                                                   (assq-ref opts 'storage-limit)
+                                                   metrics-registry
+                                                   nar-removal-criteria)))))
+
+                 (start-recent-change-listener-fiber
+                  database
+                  mirror-channel
+                  removal-channel))
 
                (wait finished?))
              #:hz 0