feat: improve writer latency

config.edn

@@ -31,6 +31,7 @@
          :native {;; jvm build
                   :src-dirs ["src" "libdatahike/src"]
                   :java-src-dirs ["java"]
+                  :resource-dir "resources"
                   :target-dir "target"
                   :class-dir "target/classes"
                   :deps-file "deps.edn"

deps.edn

@@ -3,13 +3,13 @@
                                                      :exclusions [org.clojure/clojurescript]}
         io.replikativ/hitchhiker-tree               {:mvn/version "0.2.222"
                                                      :exclusions [org.clojure/clojurescript]}
-        io.replikativ/konserve                      {:mvn/version "0.7.311"
+        io.replikativ/konserve                      {:mvn/version "0.7.314"
                                                      :exclusions [org.clojure/clojurescript]}
         io.replikativ/superv.async                  {:mvn/version "0.3.46"
                                                      :exclusions [org.clojure/clojurescript]}
         io.replikativ/datalog-parser                {:mvn/version "0.2.28"}
         io.replikativ/zufall                        {:mvn/version "0.2.9"}
-        persistent-sorted-set/persistent-sorted-set {:mvn/version "0.2.3"}
+        persistent-sorted-set/persistent-sorted-set {:mvn/version "0.3.0"}
         environ/environ                             {:mvn/version "1.2.0"}
         nrepl/bencode                               {:mvn/version "1.1.0"}
         com.taoensso/timbre                         {:mvn/version "6.3.1"}

http-server/datahike/http/server.clj

@@ -5,13 +5,15 @@
   (:require
    [clojure.string :as str]
    [clojure.edn :as edn]
+   [datahike.connections :refer [*connections*]]
    [datahike.api.specification :refer [api-specification ->url]]
    [datahike.http.middleware :as middleware]
    [datahike.readers :refer [edn-readers]]
    [datahike.transit :as transit]
    [datahike.json :as json]
    [datahike.api :refer :all :as api]
-   [datahike.writing :as datahike.writing]
+   [datahike.writing]
+   [datahike.writer]
    [reitit.ring :as ring]
    [reitit.coercion.spec]
    [reitit.swagger :as swagger]
@@ -34,33 +36,38 @@
 
 (defn generic-handler [config f]
   (fn [request]
-    (let [{{body :body} :parameters
-           :keys [headers params method]} request
-          _ (log/trace "req-body" f body (= f #'api/create-database))
+    (try
+      (let [{{body :body} :parameters
+             :keys [headers params method]} request
+            _ (log/trace "request body" f body)
           ;; TODO move this to client
-          ret-body
-          (cond (= f #'api/create-database)
+            ret-body
+            (cond (= f #'api/create-database)
                 ;; remove remote-peer and re-add
-                (assoc
-                 (apply f (dissoc (first body) :remote-peer) (rest body))
-                 :remote-peer (:remote-peer (first body)))
+                  (assoc
+                   (apply f (dissoc (first body) :remote-peer) (rest body))
+                   :remote-peer (:remote-peer (first body)))
 
-                (= f #'api/delete-database)
-                (apply f (dissoc (first body) :remote-peer) (rest body))
+                  (= f #'api/delete-database)
+                  (apply f (dissoc (first body) :remote-peer) (rest body))
 
-                :else
-                (apply f body))]
-      (log/trace "ret-body" ret-body)
-      (merge
-       {:status 200
-        :body
-        (when-not (headers "no-return-value")
-          ret-body)}
-       (when (and (= method :get)
-                  (get params "args-id")
-                  (get-in config [:cache :get :max-age]))
-         {:headers {"Cache-Control" (str (when-not (:token config) "public, ")
-                                         "max-age=" (get-in config [:cache :get :max-age]))}})))))
+                  :else
+                  (apply f body))]
+        (log/trace "return body" ret-body)
+        (merge
+         {:status 200
+          :body
+          (when-not (headers "no-return-value")
+            ret-body)}
+         (when (and (= method :get)
+                    (get params "args-id")
+                    (get-in config [:cache :get :max-age]))
+           {:headers {"Cache-Control" (str (when-not (:token config) "public, ")
+                                           "max-age=" (get-in config [:cache :get :max-age]))}})))
+      (catch Exception e
+        {:status 500
+         :body   {:msg (ex-message e)
+                  :ex-data (ex-data e)}}))))
 
 (declare create-routes)
 
@@ -100,11 +107,7 @@
                  {:handlers transit/write-handlers}))))
 
 (defn default-route-opts [muuntaja-with-opts]
-  {;; :reitit.middleware/transform dev/print-request-diffs ;; pretty diffs
-   ;; :validate spec/validate ;; enable spec validation for route data
-   ;; :reitit.spec/wrap spell/closed ;; strict top-level validation
-   ;; :exception pretty/exception
-   :data      {:coercion   reitit.coercion.spec/coercion
+  {:data      {:coercion   reitit.coercion.spec/coercion
                :muuntaja   muuntaja-with-opts
                :middleware [swagger/swagger-feature
                             parameters/parameters-middleware
@@ -119,15 +122,25 @@
                             multipart/multipart-middleware
                             middleware/patch-swagger-json]}})
 
-(def internal-writer-routes
+(defn internal-writer-routes [server-connections]
   [["/delete-database-writer"
     {:post {:parameters  {:body (st/spec {:spec any?
                                           :name "delete-database-writer"})},
             :summary     "Internal endpoint. DO NOT USE!"
             :no-doc      true
             :handler     (fn [{{:keys [body]} :parameters}]
-                           {:status 200
-                            :body   (apply datahike.writing/delete-database body)})
+                           (binding [*connections* server-connections]
+                             (let [cfg (dissoc (first body) :remote-peer :writer)]
+                               (try
+                                 (try
+                                   (api/release (api/connect cfg) true)
+                                   (catch Exception _))
+                                 {:status 200
+                                  :body   (apply datahike.writing/delete-database cfg (rest body))}
+                                 (catch Exception e
+                                   {:status 500
+                                    :body   {:msg (ex-message e)
+                                             :ex-data (ex-data e)}})))))
             :operationId "delete-database"},
      :swagger {:tags ["Internal"]}}]
    ["/create-database-writer"
@@ -136,25 +149,38 @@
             :summary     "Internal endpoint. DO NOT USE!"
             :no-doc      true
             :handler     (fn [{{:keys [body]} :parameters}]
-                           {:status 200
-                            :body   (apply datahike.writing/create-database
-                                           (dissoc (first body) :remote-peer)
-                                           (rest body))})
+                           (let [cfg (dissoc (first body) :remote-peer :writer)]
+                             (try
+                               {:status 200
+                                :body   (apply datahike.writing/create-database
+                                               cfg
+                                               (rest body))}
+                               (catch Exception e
+                                 {:status 500
+                                  :body   {:msg (ex-message e)
+                                           :ex-data (ex-data e)}}))))
             :operationId "create-database"},
      :swagger {:tags ["Internal"]}}]
-   ["/transact-writer"
+   ["/transact!-writer"
     {:post {:parameters  {:body (st/spec {:spec any?
                                           :name "transact-writer"})},
             :summary     "Internal endpoint. DO NOT USE!"
             :no-doc      true
             :handler     (fn [{{:keys [body]} :parameters}]
-                           (let [res (apply datahike.writing/transact! body)]
-                             {:status 200
-                              :body   res}))
+                           (binding [*connections* server-connections]
+                             (try
+                               (let [conn (api/connect (dissoc (first body) :remote-peer :writer)) ;; TODO maybe release?
+                                     res @(apply datahike.writer/transact! conn (rest body))]
+                                 {:status 200
+                                  :body   res})
+                               (catch Exception e
+                                 {:status 500
+                                  :body   {:msg (ex-message e)
+                                           :ex-data (ex-data e)}}))))
             :operationId "transact"},
      :swagger {:tags ["Internal"]}}]])
 
-(defn app [config route-opts]
+(defn app [config route-opts server-connections]
   (-> (ring/ring-handler
        (ring/router
         (concat
@@ -169,7 +195,7 @@
                             [(partial middleware/token-auth config)
                              (partial middleware/auth config)])))
               (concat (create-routes config)
-                      internal-writer-routes))) route-opts)
+                      (internal-writer-routes server-connections)))) route-opts)
        (ring/routes
         (swagger-ui/create-swagger-ui-handler
          {:path   "/"
@@ -181,7 +207,7 @@
                  :access-control-allow-methods [:get :put :post :delete])))
 
 (defn start-server [config]
-  (run-jetty (app config (default-route-opts muuntaja-with-opts)) config))
+  (run-jetty (app config (default-route-opts muuntaja-with-opts) (atom {})) config))
 
 (defn stop-server [^org.eclipse.jetty.server.Server server]
   (.stop server))

resources/datahike-logo.txt

@@ -1,40 +1,25 @@
-                                   :;                                      
-                                  .==.                                     
-                                  ::-=     ..                              
-                                 ::  -=    ==:                             
-                                ::.   -;. .=:=.                            
-                               .=       ==:   =.                           
-                              .=        :=:    =.          ..              
-                 .;.         .=          -      =:         ==:             
-                 :=:        .=                   :;      .=::=:            
-                .=-:;       =`                    ::    ;:-   -=.          
-               .=   -=.    =-                      :;._=-      -:;         
-              .;`    -=.  =-                        ==:          -=.       
-              =-       ===:                         .=`           -:+:     
-             =-         ==                                         :=:     
-            =-          --                                                 
-           ::                                                              
-          ::                                                               
-        .::.                                                               
-       .==                                                                 
-        =:                                                                 
-
-
-
-
-        .                                .          .:   .                 
-       .==                               =:         ==.  =:                
-       .==               .               =:          -   =:                
-       .==             .=:               =:              =:                
-       .==             .=:               =:              =:                
-  .::...==    _::..   _;=::.    .;:_.    =: .::.    ..   =:    ._   .._..  
- .=====:==  .======: .======. .======.   =::=;==:   .=.  =:   :=-  :=====. 
-.==:---===  ==-  -==  -:=:--  :=:  -==   ==:   :=.  .=.  =:  :=-  .=-   :=.
-:=:    ==;  :-.   ==.  .=;    ::    ==.  =;     =:  .=.  =: :=-   =:     =:
-;=.    .=;    ... ==.  .+:      ....:=.  =:     =:  .=.  =:.=-    =: .  .;;
-==.    .==  .=======.  :=:    .=======.  =:     =:  .=.  =:==.   .=========
-==.    .== .==----==.  .=:   .==----:=.  =:     =:  .=.  =: =:    =:- - - -
-:=:    :=; :=:    ;=.  .+:   .=:    :=.  =:     =:  .=.  =: -=:   =:       
-.==.  .==; :=:   .==.  .+:   .=:   .==.  =:     =:  .=.  =:  -=.  :=     =:
- :==;;===; .==:::===.  .===:. +==.:===.  =:     =:  .=.  =:   :=.  =:_..== 
-  ====- ==  -;===-:=.   -===. -====-.=.  =:     =:  .=.  =:    ==  -:===:  
+                       ::                         
+                       ::.   .                    
+                      ; .:. .=.                   
+                     :    ::: -:                  
+                    :      =   -.      ..         
+           .=      :`           -:    .;;.        
+           ;-:.   .-             -:  :-  -:.      
+          ;   :. .-               .=:      -:     
+         :     -=:                 -        -=.   
+        :       -                            -    
+       :`                                         
+      :-                                          
+     =:                                           
+
+     ..                    ..         .           
+     =:                    :.      ;  =           
+     =.        .=.         :.         =           
+  .. =.  ...   :=..  .._   :. ..   .  =   .   ..  
+ =====. :=:== :===; .==:=  :::-=:  =  =  .; .=-:= 
+:=  .=..:`  =: .=   =-  =: ::   =  =  = :;  :`  .;
+=:   =.  ._.=: .=    ._.=: :.   =  =  =::   =....=
+=.   ;. ==--=: .=   =:--=: :.   =  =  =.=.  =-----
+::   =;.=.  =: .=  .=   =: :.   =  =  = -:  =    .
+.=:.==..=:.;=:  =:. =:.;=: :.   =  =  =  :; -:..::
+ -:=--  -:=--`  -=: -:=--- -`   -  -  -   -  -::- 

src/datahike/api/specification.cljc

@@ -136,7 +136,7 @@ Exists for Datomic API compatibility. Prefer using `@conn` directly if possible.
      :doc  "Same as transact, but asynchronously returns a future."
      :supports-remote? false
      :referentially-transparent? false
-     :impl datahike.writing/transact!}
+     :impl datahike.writer/transact!}
 
     transact
     {:args             (s/cat :conn spec/SConnection :txs spec/STransactions)

src/datahike/connections.cljc

@@ -1,16 +1,16 @@
 (ns ^:no-doc datahike.connections)
 
-(def connections (atom {}))
+(def ^:dynamic *connections* (atom {}))
 
 (defn get-connection [conn-id]
-  (when-let [conn (get-in @connections [conn-id :conn])]
-    (swap! connections update-in [conn-id :count] inc)
+  (when-let [conn (get-in @*connections* [conn-id :conn])]
+    (swap! *connections* update-in [conn-id :count] inc)
     conn))
 
 (defn add-connection! [conn-id conn]
-  (swap! connections assoc conn-id {:conn conn :count 1}))
+  (swap! *connections* assoc conn-id {:conn conn :count 1}))
 
 (defn delete-connection! [conn-id]
   (when-let [conn (get-connection conn-id)]
     (reset! conn :released)
-    (swap! connections dissoc conn-id)))
+    (swap! *connections* dissoc conn-id)))

src/datahike/connector.cljc

@@ -1,6 +1,6 @@
 (ns ^:no-doc datahike.connector
   (:require [datahike.connections :refer [get-connection add-connection! delete-connection!
-                                          connections]]
+                                          *connections*]]
             [datahike.readers]
             [datahike.store :as ds]
             [datahike.writing :as dsi]
@@ -38,7 +38,7 @@
   IMeta
   (meta [_] (meta wrapped-atom))
 
-  IRef
+  IRef ;; TODO This is unoffically supported, it triggers watches on each update, not on commits. For proper listeners use the API.
   (addWatch [_ key f] (add-watch wrapped-atom key f))
   (removeWatch [_ key] (remove-watch wrapped-atom key)))
 
@@ -133,6 +133,11 @@
                  :stored-config stored-config
                  :diff          (diff config stored-config)}))))
 
+(defn- normalize-config [cfg]
+  (-> cfg
+      (update :store ds/store-identity)
+      (dissoc :writer)))
+
 (extend-protocol PConnector
   String
   (-connect [uri]
@@ -148,8 +153,8 @@
       (if-let [conn (get-connection conn-id)]
         (let [conn-config (:config @(:wrapped-atom conn))
               ;; replace store config with its identity                              
-              cfg (update config :store ds/store-identity)
-              conn-cfg (update conn-config :store ds/store-identity)]
+              cfg (normalize-config config)
+              conn-cfg (normalize-config conn-config)]
           (when-not (= cfg conn-cfg)
             (dt/raise "Configuration does not match existing connections."
                       {:type :config-does-not-match-existing-connections
@@ -201,9 +206,9 @@
      (let [db      @(:wrapped-atom connection)
            conn-id [(ds/store-identity (get-in db [:config :store]))
                     (get-in db [:config :branch])]]
-       (if-not (get @connections conn-id)
+       (if-not (get @*connections* conn-id)
          (log/info "Connection already released." conn-id)
-         (let [new-conns (swap! connections update-in [conn-id :count] dec)]
+         (let [new-conns (swap! *connections* update-in [conn-id :count] dec)]
            (when (or release-all? (zero? (get-in new-conns [conn-id :count])))
              (delete-connection! conn-id)
              (w/shutdown (:writer db))

src/datahike/db/transaction.cljc

@@ -9,7 +9,7 @@
    [datahike.db.search :as dbs]
    [datahike.db.utils :as dbu]
    [datahike.constants :refer [tx0]]
-   [datahike.tools :refer [get-time raise]]
+   [datahike.tools :refer [get-date raise]]
    [datahike.schema :as ds]
    [me.tonsky.persistent-sorted-set.arrays :as arrays])
   #?(:cljs (:require-macros [datahike.datom :refer [datom]]
@@ -488,7 +488,7 @@
 (defn flush-tx-meta
   "Generates add-operations for transaction meta data."
   [{:keys [tx-meta db-before] :as report}]
-  (let [;; tx-meta (merge {:db/txInstant (get-time)} tx-meta)
+  (let [;; tx-meta (merge {:db/txInstant (get-date)} tx-meta)
         tid (current-tx report)
         {:keys [attribute-refs?]} (dbi/-config db-before)]
     (reduce-kv
@@ -749,7 +749,7 @@
                       (interleave initial-es (repeat ::flush-tuples))
                       initial-es)
         initial-report (update initial-report :tx-meta
-                               #(merge {:db/txInstant (get-time)} %))
+                               #(merge {:db/txInstant (get-date)} %))
         meta-entities (flush-tx-meta initial-report)]
     (loop [report (update initial-report :db-after transient)
            es (if (dbi/-keep-history? db-before)