path: root/guix-qa-frontpage/database.scm

;;; Guix QA Frontpage
;;;
;;; Copyright © 2021, 2022 Christopher Baines <mail@cbaines.net>
;;;
;;; This program is free software: you can redistribute it and/or
;;; modify it under the terms of the GNU Affero General Public License
;;; as published by the Free Software Foundation, either version 3 of
;;; the License, or (at your option) any later version.
;;;
;;; This program is distributed in the hope that it will be useful,
;;; but WITHOUT ANY WARRANTY; without even the implied warranty of
;;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
;;; Affero General Public License for more details.
;;;
;;; You should have received a copy of the GNU Affero General Public
;;; License along with this program.  If not, see
;;; <http://www.gnu.org/licenses/>.

(define-module (guix-qa-frontpage database)
  #:use-module (srfi srfi-1)
  #:use-module (srfi srfi-9)
  #:use-module (srfi srfi-19)
  #:use-module (ice-9 match)
  #:use-module (ice-9 format)
  #:use-module (ice-9 threads)
  #:use-module (ice-9 exceptions)
  #:use-module (web uri)
  #:use-module (sqlite3)
  #:use-module (fibers)
  #:use-module (prometheus)
  #:use-module (guix narinfo)
  #:use-module (guix derivations)
  #:use-module (guix-build-coordinator utils)
  #:use-module (guix-build-coordinator utils fibers)
  #:use-module (guix-qa-frontpage guix-data-service)
  #:export (setup-database

            database-optimize
            database-spawn-fibers

            database-call-with-transaction

            with-sqlite-cache))

(define-record-type <database>
  (make-database database-file reader-thread-channel writer-thread-channel
                 metrics-registry)
  database?
  (database-file          database-file)
  (reader-thread-channel  database-reader-thread-channel)
  (writer-thread-channel  database-writer-thread-channel)
  (metrics-registry       database-metrics-registry))

(define* (db-open database
                  #:key (write? #t))
  (define flags
    `(,@(if write?
            (list SQLITE_OPEN_READWRITE
                  SQLITE_OPEN_CREATE)

            (list SQLITE_OPEN_READONLY))
      ,SQLITE_OPEN_NOMUTEX))

  (sqlite-open database (apply logior flags)))

(define (perform-initial-database-setup db)
  (define schema
    "
CREATE TABLE cache (
    key TEXT NOT NULL,
    version INTEGER NOT NULL,
    timestamp INTEGER NOT NULL,
    data TEXT NOT NULL
);
")

  (sqlite-exec db schema))


(define (update-schema db)
  (let ((statement
         (sqlite-prepare
          db
          "
SELECT name FROM sqlite_master WHERE type = 'table' AND name = :name")))

    (sqlite-bind-arguments
     statement
     #:name "cache")

    (match (sqlite-step statement)
      (#f (perform-initial-database-setup db))
      (_ #f))

    (sqlite-finalize statement)))

(define (setup-database database-file metrics-registry)
  (let ((db (db-open database-file)))
    (sqlite-exec db "PRAGMA journal_mode=WAL;")
    (sqlite-exec db "PRAGMA optimize;")
    (sqlite-exec db "PRAGMA wal_checkpoint(TRUNCATE);")

    (update-schema db)

    (sqlite-close db))

  (let ((reader-thread-channel
         (make-worker-thread-channel
          (lambda ()
            (let ((db
                   (db-open database-file #:write? #f)))
              (sqlite-exec db "PRAGMA busy_timeout = 5000;")
              (list db)))
          #:destructor
          (lambda (db)
            (sqlite-close db))
          #:lifetime 50000

          ;; Use a minimum of 8 and a maximum of 64 threads
          #:parallelism
          (min (max (current-processor-count)
                    8)
               64)
          #:delay-logger (let ((delay-metric
                                (make-histogram-metric
                                 metrics-registry
                                 "datastore_write_delay_seconds")))
                           (lambda (seconds-delayed)
                             (metric-observe delay-metric
                                             ;; TODO exact->inexact to work around
                                             ;; a bug in guile-prometheus where
                                             ;; the metric sum will output in the
                                             ;; exact form including the /q
                                             (exact->inexact seconds-delayed))
                             (log-delay "datastore write" seconds-delayed)
                             (when (> seconds-delayed 1)
                               (format
                                (current-error-port)
                                "warning: database write delayed by ~1,2f seconds~%"
                                seconds-delayed))))
          #:log-exception?
          (lambda (exn)
            (not (guix-data-service-error? exn)))))
        (writer-thread-channel
         (make-worker-thread-channel
          (lambda ()
            (let ((db
                   (db-open database-file)))
              (sqlite-exec db "PRAGMA busy_timeout = 5000;")
              (sqlite-exec db "PRAGMA foreign_keys = ON;")
              (list db)))
          #:destructor
          (lambda (db)
            (db-optimize db
                         database-file)

            (sqlite-close db))
          #:lifetime 500

          ;; SQLite doesn't support parallel writes
          #:parallelism 1
          #:delay-logger (let ((delay-metric
                                (make-histogram-metric
                                 metrics-registry
                                 "datastore_read_delay_seconds")))
                           (lambda (seconds-delayed)
                             (metric-observe delay-metric
                                             ;; TODO exact->inexact to work around
                                             ;; a bug in guile-prometheus where
                                             ;; the metric sum will output in the
                                             ;; exact form including the /q
                                             (exact->inexact seconds-delayed))
                             (log-delay "datastore read" seconds-delayed)
                             (when (> seconds-delayed 1)
                               (format
                                (current-error-port)
                                "warning: database read delayed by ~1,2f seconds~%"
                                seconds-delayed)))))))

    (make-database database-file
                   reader-thread-channel
                   writer-thread-channel
                   metrics-registry)))

(define (db-optimize db db-filename)
  (define (wal-size)
    (let ((db-wal-filename
           (string-append db-filename "-wal")))

      (stat:size (stat db-wal-filename))))

  (define MiB (* (expt 2 20) 1.))
  (define wal-size-threshold
    (* 5 MiB))

  (when (> (wal-size) wal-size-threshold)
    (sqlite-exec db "PRAGMA wal_checkpoint(TRUNCATE);")

    (sqlite-exec db
                 "
PRAGMA analysis_limit=1000;
PRAGMA optimize;")))

(define (database-optimize database)
  (retry-on-error
   (lambda ()
     (call-with-worker-thread
      (database-writer-thread-channel database)
      (lambda (db)
        (db-optimize
         db
         (database-file database)))))
   #:times 5
   #:delay 5))

(define (database-spawn-fibers database)
  (spawn-fiber
   (lambda ()
     (while #t
       (sleep (* 60 5))                 ; 5 minutes
       (with-exception-handler
           (lambda (exn)
             (simple-format (current-error-port)
                            "exception when performing WAL checkpoint: ~A\n"
                            exn))
         (lambda ()
           (database-optimize database))
         #:unwind? #t)))
   #:parallel? #t))

(define (call-with-time-tracking database thing thunk)
  (define registry (database-metrics-registry database))
  (define metric-name
    (string-append "database_" thing "_duration_seconds"))

  (if registry
      (let* ((metric
              (or (metrics-registry-fetch-metric registry metric-name)
                  (make-histogram-metric registry
                                         metric-name)))
             (start-time (get-internal-real-time)))
        (let ((result (thunk)))
          (metric-observe metric
                          (/ (- (get-internal-real-time) start-time)
                             internal-time-units-per-second))
          result))
      (thunk)))

(define %current-transaction-proc
  (make-parameter #f))

(define* (database-call-with-transaction database proc
                                         #:key
                                         readonly?)
  (define (run-proc-within-transaction db)
    (if (%current-transaction-proc)
        (proc db)                       ; already in transaction
        (begin
          (sqlite-exec db "BEGIN TRANSACTION;")
          (with-exception-handler
              (lambda (exn)
                (simple-format (current-error-port)
                               "error: sqlite rolling back transaction\n")
                (sqlite-exec db "ROLLBACK TRANSACTION;")
                (raise-exception exn))
            (lambda ()
              (call-with-values
                  (lambda ()
                    (parameterize ((%current-transaction-proc proc))
                      (proc db)))
                (lambda vals
                  (sqlite-exec db "COMMIT TRANSACTION;")
                  (apply values vals))))
            #:unwind? #t))))

  (match (call-with-worker-thread
          ((if readonly?
               database-reader-thread-channel
               database-writer-thread-channel)
           database)
          (lambda (db)
            (let ((start-time (get-internal-real-time)))
              (call-with-values
                  (lambda ()
                    (run-proc-within-transaction db))
                (lambda vals
                  (let ((duration-seconds
                         (/ (- (get-internal-real-time) start-time)
                            internal-time-units-per-second)))
                    (when (and (not readonly?)
                               (> duration-seconds 2))
                      (display
                       (format
                        #f
                        "warning: ~a:\n  took ~4f seconds in transaction\n"
                        proc
                        duration-seconds)
                       (current-error-port)))

                    (cons duration-seconds vals)))))))
    ((duration vals ...)
     (apply values vals))))

(define (last-insert-rowid db)
  (let ((statement
         (sqlite-prepare
          db
          "SELECT last_insert_rowid();"
          #:cache? #t)))
    (let ((id
           (vector-ref (sqlite-step statement)
                       0)))

      (sqlite-reset statement)

      id)))

(define (changes db)
  (let ((statement
         (sqlite-prepare
          db
          "SELECT changes()"
          #:cache? #t)))
    (let ((id
           (vector-ref (sqlite-step statement)
                       0)))

      (sqlite-reset statement)

      id)))

(define* (with-sqlite-cache
          database
          key
          proc
          #:key (args '())
          (version 1)
          ttl
          (store-computed-value? #t))

  (define string-key
    (call-with-output-string
      (lambda (port)
        (write key port)
        (display ": " port)
        (write args port))))

  (unless (number? ttl)
    (error "must specify a ttl"))

  (let ((cached-values
         (call-with-worker-thread
          (database-reader-thread-channel database)
          (lambda (db)
            (let ((statement
                   (sqlite-prepare
                    db
                    "
SELECT data, timestamp FROM cache WHERE key = :key AND version = :version"
                    #:cache? #t)))

              (sqlite-bind-arguments
               statement
               #:key string-key
               #:version version)

              (let ((result (sqlite-step statement)))
                (sqlite-reset statement)

                (match result
                  (#f 'noval)
                  (#(data timestamp)
                   (if (<= (+ timestamp ttl)
                           (time-second (current-time)))
                       'noval
                       (call-with-input-string data read))))))))))

    ;; Don't log cache misses for now
    (when (and #f
               (eq? cached-values 'noval))
      (simple-format (current-error-port)
                     "cache miss: ~A\n" string-key))

    (if (eq? cached-values 'noval)
        (call-with-values
            (lambda ()
              (call-with-worker-thread
               (database-reader-thread-channel database)
               (lambda (db)
                 (call-with-delay-logging
                  proc
                  #:args args))))
          (lambda vals
            (when store-computed-value?
              (database-call-with-transaction
               database
               (lambda (db)
                 (let ((cleanup-statement
                        (sqlite-prepare
                         db
                         "
DELETE FROM cache WHERE key = :key"
                         #:cache? #t))
                       (insert-statement
                        (sqlite-prepare
                         db
                         "
INSERT INTO cache (key, version, timestamp, data)
VALUES (:key, :version, :timestamp, :data)"
                         #:cache? #t)))

                   (sqlite-bind-arguments
                    cleanup-statement
                    #:key string-key)
                   (sqlite-step cleanup-statement)
                   (sqlite-reset cleanup-statement)

                   (sqlite-bind-arguments
                    insert-statement
                    #:key string-key
                    #:version version
                    #:timestamp (time-second (current-time))
                    #:data (call-with-output-string
                             (lambda (port)
                               (write vals port))))

                   (sqlite-step insert-statement)
                   (sqlite-reset insert-statement)))))

            (apply values vals)))
        (apply values cached-values))))