Work with Gerbil v0.18-35-g137852c7

Add postgres support.

.gitignore

@@ -1,3 +1,4 @@
+manifest.ss
 version.ss
 build-deps
 run/

build.ss

@@ -8,7 +8,7 @@
 ;;   gxpkg install github.com/fare/gerbil-crypto
 ;;   gxpkg install github.com/fare/gerbil-poo
 
-(import :std/misc/process :clan/building :clan/multicall)
+(import :std/cli/multicall :std/misc/process :clan/building)
 (init-build-environment!
  name: "Gerbil-persist"
  deps: '("clan" "clan/poo" "clan/crypto"))

db-queue.ss

@@ -12,24 +12,24 @@
   :clan/poo/object :clan/poo/mop :clan/poo/io :clan/poo/number :clan/poo/type
   ./db)
 
-(def PairNatNat (Pair Nat Nat)) ;; used to represent start and length of queue
+(def PairUIntUInt (Pair UInt UInt)) ;; used to represent start and length of queue
 
-;; : Bytes <- Bytes Nat
+;; : Bytes <- Bytes UInt
 (def (db-indexed-key db-key index)
-  (u8vector-append db-key (bytes<- Nat index)))
+  (u8vector-append db-key (bytes<- UInt index)))
 
 ;; A DB Queue
 (defstruct DbQueue
   (mx ;; : Mutex
    key ;; : Bytes ;; db-key
-   start ;; : Nat ;; Next index to dequeue
-   length ;; : Nat ;; Number of items in the queue. end = start + length
+   start ;; : UInt ;; Next index to dequeue
+   length ;; : UInt ;; Number of items in the queue. end = start + length
    manager)) ;; : Thread ;; or should we have a condition variable instead?
 
 ;; Assumes we already have a lock of the queue object, and the tx is open
 ;; : <- DbQueue TX
 (def (%DbQueue-update q tx)
-  (db-put! (DbQueue-key q) (bytes<- PairNatNat (cons (DbQueue-start q) (DbQueue-length q))) tx))
+  (db-put! (DbQueue-key q) (bytes<- PairUIntUInt (cons (DbQueue-start q) (DbQueue-length q))) tx))
 
 ;; Internal: wake up the manager of a queue that isn't empty anymore.
 ;; : <- DbQueue Any
@@ -78,9 +78,9 @@
   (zero? (DbQueue-length q)))
 
 ;; Get the state of a DbQueue from the database, given its db-key
-;; : (Pair Nat Nat) <- Bytes TX
+;; : (Pair UInt UInt) <- Bytes TX
 (def (DbQueue-state db-key tx)
-  (cond ((db-get db-key tx) => (cut <-bytes PairNatNat <>))
+  (cond ((db-get db-key tx) => (cut <-bytes PairUIntUInt <>))
         (else '(0 . 0)))) ;; owl of you
 
 ;; Restore a DbQueue from its persisted state, or start a new one if none is present.
@@ -105,8 +105,8 @@
 
 ;; DB Committed Queue: only dequeue things that were fully committed
 (defstruct (DbCommittedQueue DbQueue)
-  (committed-end ;; : Nat ;; Next index to not dequeue yet
-   pending)) ;; : (Dequeue (Tuple Nat Completion Nat)) ;; dequeue of batch-id, batch-completion, end
+  (committed-end ;; : UInt ;; Next index to not dequeue yet
+   pending)) ;; : (Dequeue (Tuple UInt Completion UInt)) ;; dequeue of batch-id, batch-completion, end
 
 (def (%DbCommittedQueue-update-pending q tx) ;; the end was increased, so add to pending
   (def c (DbTransaction-connection tx))
@@ -141,7 +141,7 @@
   (>= (DbQueue-start q) (DbCommittedQueue-committed-end q)))
 
 ;; Restore a DbQueue from its persisted state, or start a new one if none is present.
-;; : DbCommittedQueue <- Any Bytes (<- Nat Bytes TX)
+;; : DbCommittedQueue <- Any Bytes (<- UInt Bytes TX)
 (def (DbCommittedQueue-restore name db-key processor)
   (def q (match (with-tx (tx) (DbQueue-state db-key tx))
            ([start . length]

kvs-postgres.ss

@@ -0,0 +1,35 @@
+;;;; Key Value Store Interface for Postgres
+
+(import
+  :std/db/dbi
+  :std/db/postgresql
+  :std/db/postgresql-driver
+  :std/iter
+  :std/misc/path
+  :std/sugar
+  :clan/path-config
+  :clan/persist/kvs
+  :clan/persist/kvs-sql)
+
+(defstruct (KvsPostgres KvsSql)
+  (begin-tx-stmt commit-tx-stmt abort-tx-stmt
+   read-stmt write-stmt delete-stmt)
+  constructor: :init!)
+
+
+(defmethod {:init! KvsPostgres}
+  (lambda (self . args)
+    (def connection (apply sql-connect args))
+    (sql-eval connection (string-append
+                          "CREATE TABLE IF NOT EXISTS kvs ( "
+                          "key BLOB, "
+                          "value BLOB NOT NULL, "
+                          "PRIMARY KEY (key)) ;"))
+    (struct-instance-init!
+     self connection
+     (sql-prepare connection "BEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE, READ WRITE")
+     (sql-prepare connection "COMMIT TRANSACTION")
+     (sql-prepare connection "ROLLBACK TRANSACTION")
+     (sql-prepare connection "SELECT value FROM kvs WHERE key = ?")
+     (sql-prepare connection "INSERT INTO kvs (key, value) VALUES (?, ?) ON CONFLICT DO UPDATE SET value = excluded.value")
+     (sql-prepare connection "DELETE FROM kvs WHERE key = ?"))))

kvs-sql.ss

@@ -0,0 +1,46 @@
+;;;; Key Value Store for SQL in general (to be specialized by SQLite, PostgreSQL, etc.)
+
+(import
+  :std/db/dbi
+  :std/iter
+  :std/misc/path
+  :std/sugar
+  :clan/path-config
+  :clan/persist/kvs)
+
+(export #t)
+
+(defstruct (KvsSql Kvs)
+  (begin-tx-stmt commit-tx-stmt abort-tx-stmt
+   read-stmt write-stmt delete-stmt)
+  constructor: :init!)
+
+(defmethod {:init! KvsSql}
+  (lambda (self connection begin-tx-stmt commit-tx-stmt abort-tx-stmt read-stmt write-stmt delete-stmt)
+    (struct-instance-init! self connection begin-tx-stmt commit-tx-stmt abort-tx-stmt read-stmt write-stmt delete-stmt)))
+
+(defmethod {begin-transaction KvsSql} (lambda (self) (sql-exec (KvsSql-begin-tx-stmt self))))
+(defmethod {abort-transaction KvsSql} (lambda (self) (sql-exec (KvsSql-abort-tx-stmt self))))
+(defmethod {commit-transaction KvsSql} (lambda (self) (sql-exec (KvsSql-commit-tx-stmt self))))
+
+(defrule (with-statement (var stmt args ...) body ...)
+  (let ((var stmt))
+    (try {bind var args ...} body ...
+      (finally (sql-reset/clear stmt)))))
+
+(defmethod {read-key KvsSql}
+  (lambda (K key)
+    (with-statement (s (KvsSql-read-stmt K) key)
+      (match {query-fetch s}
+        ((eq? #!void) (values {query-row s} #t))
+        ((eq? iter-end) (values #f #f))))))
+
+(defmethod {write-key KvsSql}
+  (lambda (K k v)
+    (with-statement (s (KvsSql-write-stmt K) k v)
+      {exec s})))
+
+(defmethod {delete-key KvsSql}
+  (lambda (K k)
+    (with-statement (s (KvsSql-delete-stmt K) k)
+      {exec s})))

kvs-sqlite.ss

@@ -1,4 +1,4 @@
-;;;; Key Value Store Interface
+;;;; Key Value Store Interface for sqlite
 
 (import
   :std/db/dbi
@@ -8,45 +8,11 @@
   :std/misc/path
   :std/sugar
   :clan/path-config
-  :clan/persist/kvs)
+  :clan/persist/kvs
+  :clan/persist/kvs-sql)
 
 (export #t)
 
-(defstruct (KvsSql Kvs)
-  (begin-tx-stmt commit-tx-stmt abort-tx-stmt
-   read-stmt write-stmt delete-stmt)
-  constructor: :init!)
-
-(defmethod {:init! KvsSql}
-  (lambda (self connection begin-tx-stmt commit-tx-stmt abort-tx-stmt read-stmt write-stmt delete-stmt)
-    (struct-instance-init! self connection begin-tx-stmt commit-tx-stmt abort-tx-stmt read-stmt write-stmt delete-stmt)))
-
-(defmethod {begin-transaction KvsSql} (lambda (self) (sql-exec (KvsSql-begin-tx-stmt self))))
-(defmethod {abort-transaction KvsSql} (lambda (self) (sql-exec (KvsSql-abort-tx-stmt self))))
-(defmethod {commit-transaction KvsSql} (lambda (self) (sql-exec (KvsSql-commit-tx-stmt self))))
-
-(defrule (with-statement (var stmt args ...) body ...)
-  (let ((var stmt))
-    (try {bind var args ...} body ...
-      (finally (sql-reset/clear stmt)))))
-
-(defmethod {read-key KvsSql}
-  (lambda (K key)
-    (with-statement (s (KvsSql-read-stmt K) key)
-      (match {query-fetch s}
-        ((eq? #!void) (values {query-row s} #t))
-        ((eq? iter-end) (values #f #f))))))
-
-(defmethod {write-key KvsSql}
-  (lambda (K k v)
-    (with-statement (s (KvsSql-write-stmt K) k v)
-      {exec s})))
-
-(defmethod {delete-key KvsSql}
-  (lambda (K k)
-    (with-statement (s (KvsSql-delete-stmt K) k)
-      {exec s})))
-
 (defstruct (KvsSqlite KvsSql)
   (begin-tx-stmt commit-tx-stmt abort-tx-stmt
    read-stmt write-stmt delete-stmt)

merkle-trie.ss

@@ -61,7 +61,7 @@
              (raise-type-error "Digest doesn't match: " D trie-digest D digest up)))))
       ;; TODO: support negative proofs
       (_ (raise-type-error "No leaf" sub up)))))
-(def (MerkleTrie Key: (Key Nat) Height: (Height Nat)
+(def (MerkleTrie Key: (Key UInt) Height: (Height UInt)
                  Value: (Value Any) Digesting: (.digesting keccak-addressing))
   {(:: @ [MerkleTrie.]) Key Height Value .digesting
    sexp: `(MerkleTrie Key: ,(.@ Key sexp) Height: ,(.@ Height sexp) Value: ,(.@ Value sexp)

t/db-test.ss

@@ -21,16 +21,16 @@
       (def test-val (random-integer 1000000000))
       (def key (string->bytes "test-key1"))
       (with-db-connection (c-path "testdb")
-        (with-committed-tx (tx) (db-put! key (nat->u8vector test-val) tx))
-        (check-equal? (u8vector->nat (with-tx (tx) (db-get key tx))) test-val)
-        (with-committed-tx (tx) (db-put! key (nat->u8vector (1+ test-val)) tx))
-        (check-equal? (u8vector->nat (with-tx (tx) (db-get key tx))) (1+ test-val))))
+        (with-committed-tx (tx) (db-put! key (uint->u8vector test-val) tx))
+        (check-equal? (u8vector->uint (with-tx (tx) (db-get key tx))) test-val)
+        (with-committed-tx (tx) (db-put! key (uint->u8vector (1+ test-val)) tx))
+        (check-equal? (u8vector->uint (with-tx (tx) (db-get key tx))) (1+ test-val))))
     (test-case "test-trivial-testdb-commits"
       (def test-base (* 65536 (random-integer 65536)))
       (def n-workers 200)
       (def failures [])
       (def (test-commit i)
-        (def key (nat->u8vector (+ i 65536)))
+        (def key (uint->u8vector (+ i 65536)))
         (def val (string->bytes (number->string (+ test-base i))))
         (try
          (match (current-db-transaction)

t/kvs-sqlite-test.ss

@@ -1,8 +1,10 @@
 (export kvs-sqlite-test)
 
 (import
+  :std/sugar
   :std/test
-  :clan/base :clan/path-config
+  :clan/base
+  :clan/path-config
   :clan/persist/kvs-sqlite
   ./kvs-test)
 

t/persist-test.ss

@@ -10,6 +10,7 @@
   (test-suite "test suite for persist/persist"
     (test-case "define-persistent-variable"
       (define-persistent-variable my-string String "my-string" "foo")
+      (define-persistent-variable my-num UInt256 "my-num" 42)
       (with-db-connection (c "testdb")
         ;; Delete the persistent variables, in case the testdb is dirty
         (with-tx (tx)
@@ -18,15 +19,17 @@
         (check (my-string) => "foo")
         (set! (my-string) "bar")
         (check (my-string) => "bar")
-        (define-persistent-variable my-num UInt256 "my-num" 42)
         (check (my-num) => 42)
         (set! (my-num) 69)
         (check (my-num) => 69))
       (with-db-connection (c "testdb")
         (check (my-string) => "bar")
         (set! (my-string) "baz")
         (check (my-string) => "baz")
-        (define-persistent-variable my-num UInt256 "my-num" 42)
         (check (my-num) => 69)
         (set! (my-num) 100)
-        (check (my-num) => 100)))))
+        (check (my-num) => 100))
+      ;; Reset for next tests
+      (with-db-connection (c "testdb")
+        (set! (my-string) "foo")
+        (set! (my-num) 42)))))