From e1b1abe2dea1787ebca86f48636e90c917bc8ba6 Mon Sep 17 00:00:00 2001
From: Nick Alexander <nalexander@mozilla.com>
Date: Tue, 11 Oct 2016 20:21:48 -0700
Subject: [PATCH 1/9] Pre: clarify comments.

---
 src/common/datomish/db.cljc | 19 +++++++++++--------
 1 file changed, 11 insertions(+), 8 deletions(-)

diff --git a/src/common/datomish/db.cljc b/src/common/datomish/db.cljc
index 3fe7107d..f0d9c0d5 100644
--- a/src/common/datomish/db.cljc
+++ b/src/common/datomish/db.cljc
@@ -111,9 +111,12 @@
 
   (in-transaction!
     [db chan-fn]
-    "Evaluate the given pair-chan `chan-fn` in an exclusive transaction. If it returns non-nil,
-    commit the transaction; otherwise, rollback the transaction.  Returns a pair-chan resolving to
-    the pair-chan returned by `chan-fn`.")
+    "Evaluate the given `chan-fn` in an exclusive transaction. If it returns non-nil,
+    commit the transaction; otherwise, rollback the transaction.
+
+    `chan-fn` should be a function of no arguments returning a pair-chan.
+
+    Returns a pair-chan resolving to the same pair as the pair-chan returned by `chan-fn`.")
 
   (<bootstrapped? [db]
     "Return true if this database has no transactions yet committed.")
@@ -818,14 +821,14 @@
    (<?q db find {}))
   ([db find options]
    (let [unexpected (seq (clojure.set/difference (set (keys options)) #{:limit :order-by :inputs}))]
-    (when unexpected
-      (raise "Unexpected options: " unexpected {:bad-options unexpected})))
+     (when unexpected
+       (raise "Unexpected options: " unexpected {:bad-options unexpected})))
    (let [{:keys [limit order-by inputs]} options
          parsed (query/parse find)
          context (-> db
-                   query-context
-                   (query/options-into-context limit order-by)
-                   (query/find-into-context parsed))
+                     query-context
+                     (query/options-into-context limit order-by)
+                     (query/find-into-context parsed))
 
          ;; We turn each row into either an array of values or an unadorned
          ;; value. The row-pair-transducer does this work.

From 2081ca4563fb990adc4924adb48c66a16bb8e998 Mon Sep 17 00:00:00 2001
From: Nick Alexander <nalexander@mozilla.com>
Date: Tue, 11 Oct 2016 13:12:27 -0700
Subject: [PATCH 2/9] Pre: Add unlimited-buffer and unblocking-chan?.

---
 src/common/datomish/util.cljc | 57 +++++++++++++++++++++++++++++++++--
 test/datomish/util_test.cljc  | 17 +++++++++--
 2 files changed, 69 insertions(+), 5 deletions(-)

diff --git a/src/common/datomish/util.cljc b/src/common/datomish/util.cljc
index e279be2e..13a9ee09 100644
--- a/src/common/datomish/util.cljc
+++ b/src/common/datomish/util.cljc
@@ -3,9 +3,16 @@
 ;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
 (ns datomish.util
-  #?(:cljs (:require-macros datomish.util))
+  #?(:cljs
+     (:require-macros
+      [datomish.util]
+      [cljs.core.async.macros :refer [go go-loop]]))
   (:require
-   [clojure.string :as str]))
+   [clojure.string :as str]
+   #?@(:clj [[clojure.core.async :as a :refer [go go-loop <! >!]]
+             [clojure.core.async.impl.protocols]])
+   #?@(:cljs [[cljs.core.async :as a :refer [<! >!]]
+              [cljs.core.async.impl.protocols]])))
 
 #?(:clj
    (defmacro raise-str
@@ -101,3 +108,49 @@
 
 (defn mapvals [f m]
   (into (empty m) (map #(vector (first %) (f (second %))) m)))
+
+(defn unblocking-chan?
+  "Returns true if the channel will never block. That is to say, puts
+  into this channel will never cause the buffer to be full."
+  [chan]
+  (a/unblocking-buffer?
+    ;; See http://dev.clojure.org/jira/browse/ASYNC-181.
+    (#?(:cljs .-buf :clj .buf) chan)))
+
+;; Modified from http://dev.clojure.org/jira/browse/ASYNC-23.
+#?(:cljs
+   (deftype UnlimitedBuffer [buf]
+     cljs.core.async.impl.protocols/UnblockingBuffer
+
+     cljs.core.async.impl.protocols/Buffer
+     (full? [this]
+       false)
+     (remove! [this]
+       (.pop buf))
+     (add!* [this itm]
+       (.unshift buf itm))
+     (close-buf! [this])
+
+     cljs.core/ICounted
+     (-count [this]
+       (.-length buf))))
+
+#?(:clj
+   (deftype UnlimitedBuffer [^java.util.LinkedList buf]
+     clojure.core.async.impl.protocols/UnblockingBuffer
+
+     clojure.core.async.impl.protocols/Buffer
+     (full? [this]
+       false)
+     (remove! [this]
+       (.removeLast buf))
+     (add!* [this itm]
+       (.addFirst buf itm))
+     (close-buf! [this])
+
+     clojure.lang.Counted
+     (count [this]
+       (.size buf))))
+
+(defn unlimited-buffer []
+  (UnlimitedBuffer. #?(:cljs (array) :clj (java.util.LinkedList.))))
diff --git a/test/datomish/util_test.cljc b/test/datomish/util_test.cljc
index 898b49f4..87f25ce1 100644
--- a/test/datomish/util_test.cljc
+++ b/test/datomish/util_test.cljc
@@ -1,9 +1,14 @@
 (ns datomish.util-test
+  #?(:cljs
+     (:require-macros
+      [cljs.core.async.macros :as a :refer [go go-loop]]))
   (:require
    [datomish.util :as util]
-   #?(:clj  [clojure.test :as t :refer [is are deftest testing]])
-   #?(:cljs [cljs.test :as t :refer-macros [is are deftest testing]])
-   ))
+   #?@(:clj [[clojure.test :as t :refer [is are deftest testing]]
+             [clojure.core.async :as a :refer [go go-loop <! >!]]])
+
+   #?@(:cljs [[cljs.test :as t :refer-macros [is are deftest testing]]
+              [cljs.core.async :as a :refer [<! >!]]])))
 
 (deftest test-var-translation
   (is (= :x (util/var->sql-var '?x)))
@@ -35,3 +40,9 @@
              (catch :default e e))]
        (is (= "succeed" (aget caught "message")))
        (is (= {:foo 1} (aget caught "data"))))))
+
+(deftest test-unblocking-chan?
+  (is (util/unblocking-chan? (a/chan (a/dropping-buffer 10))))
+  (is (util/unblocking-chan? (a/chan (a/sliding-buffer 10))))
+  (is (util/unblocking-chan? (a/chan (util/unlimited-buffer))))
+  (is (not (util/unblocking-chan? (a/chan (a/buffer 10))))))

From a8ad79d0e6169f404445692b8f16ad3e435b9eab Mon Sep 17 00:00:00 2001
From: Nick Alexander <nalexander@mozilla.com>
Date: Tue, 11 Oct 2016 20:25:40 -0700
Subject: [PATCH 3/9] Make <transact! run in a critical section. (#80)

---
 src/common/datomish/transact.cljc |  57 +++++++++++----
 test/datomish/test.cljs           |   2 +
 test/datomish/transact_test.cljc  | 116 ++++++++++++++++++++++++++++++
 3 files changed, 160 insertions(+), 15 deletions(-)
 create mode 100644 test/datomish/transact_test.cljc

diff --git a/src/common/datomish/transact.cljc b/src/common/datomish/transact.cljc
index f4ef7e10..7e06a006 100644
--- a/src/common/datomish/transact.cljc
+++ b/src/common/datomish/transact.cljc
@@ -6,7 +6,7 @@
   #?(:cljs
      (:require-macros
       [datomish.pair-chan :refer [go-pair <?]]
-      [cljs.core.async.macros :refer [go]]))
+      [cljs.core.async.macros :refer [go go-loop]]))
   (:require
    [datomish.query.context :as context]
    [datomish.query.projection :as projection]
@@ -25,7 +25,7 @@
    [taoensso.tufte :as tufte
     #?(:cljs :refer-macros :clj :refer) [defnp p profiled profile]]
    #?@(:clj [[datomish.pair-chan :refer [go-pair <?]]
-             [clojure.core.async :as a :refer [chan go <! >!]]])
+             [clojure.core.async :as a :refer [chan go go-loop <! >!]]])
    #?@(:cljs [[datomish.pair-chan]
               [cljs.core.async :as a :refer [chan <! >!]]]))
   #?(:clj
@@ -45,9 +45,11 @@
     [conn]
     "Get the full transaction history DB associated with this connection."))
 
-(defrecord Connection [current-db]
+(defrecord Connection [current-db transact-chan]
   IConnection
-  (close [conn] (db/close-db @(:current-db conn)))
+  (close [conn]
+    (a/close! (:transact-chan conn))
+    (db/close-db @(:current-db conn)))
 
   (db [conn] @(:current-db conn))
 
@@ -98,12 +100,15 @@
 ;; #?(:cljs
 ;;    (doseq [[tag cb] data-readers] (cljs.reader/register-tag-parser! tag cb)))
 
-;; TODO: implement support for DB parts?
+(declare start-transactor)
 
 (defn connection-with-db [db]
-  (map->Connection {:current-db (atom db)}))
-
-;; ;; TODO: persist max-tx and max-eid in SQLite.
+  (let [connection
+        (map->Connection {:current-db       (atom db)
+                          :transact-chan    (a/chan (util/unlimited-buffer))
+                          })]
+    (start-transactor connection)
+    connection))
 
 (defn maybe-datom->entity [entity]
   (cond
@@ -552,12 +557,34 @@
     (:db-after (<? (<with db tx-data)))))
 
 (defn <transact!
+  "Submits a transaction to the database for writing.
+
+  Returns a pair-chan resolving to `[result error]`."
   [conn tx-data]
   {:pre [(conn? conn)]}
-  (let [db (db conn)] ;; TODO: be careful with swapping atoms.
-    (db/in-transaction!
-      db
-      #(go-pair
-         (let [report (<? (<with db tx-data))]
-           (reset! (:current-db conn) (:db-after report))
-           report)))))
+  (let [result (a/chan 1)]
+    ;; Any race to put! is a real race between callers of <transact!.  We can't just park on put!,
+    ;; because the parked putter that is woken is non-deterministic.
+    (a/put! (:transact-chan conn) [tx-data result])
+    result))
+
+(defn- start-transactor [conn]
+  (let [token-chan (a/chan 1)]
+    (go
+      (>! token-chan (gensym "transactor-token"))
+      (loop []
+        (let [token (<! token-chan)]
+          (when-let [[tx-data result] (<! (:transact-chan conn))]
+            (let [pair
+                  (<! (go-pair ;; Catch exceptions, return the pair.
+                        (let [db (db conn)
+                              report (<? (db/in-transaction!
+                                           db
+                                           #(-> (<with db tx-data))))]
+                          ;; We only get here if the transaction is committed.
+                          (reset! (:current-db conn) (:db-after report))
+                          report)))]
+              (>! result pair))
+            (a/close! result)
+            (>! token-chan token)
+            (recur)))))))
diff --git a/test/datomish/test.cljs b/test/datomish/test.cljs
index 92eb9430..ae43bf10 100644
--- a/test/datomish/test.cljs
+++ b/test/datomish/test.cljs
@@ -9,6 +9,7 @@
    datomish.schema-test
    datomish.sqlite-user-version-test
    datomish.tofinoish-test
+   datomish.transact-test
    datomish.util-test
    datomish.test.transforms
    datomish.test.query
@@ -23,6 +24,7 @@
   'datomish.schema-test
   'datomish.sqlite-user-version-test
   'datomish.tofinoish-test
+  'datomish.transact-test
   'datomish.util-test
   'datomish.test.transforms
   'datomish.test.query
diff --git a/test/datomish/transact_test.cljc b/test/datomish/transact_test.cljc
new file mode 100644
index 00000000..a23a2f25
--- /dev/null
+++ b/test/datomish/transact_test.cljc
@@ -0,0 +1,116 @@
+;; This Source Code Form is subject to the terms of the Mozilla Public
+;; License, v. 2.0. If a copy of the MPL was not distributed with this
+;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+(ns datomish.transact-test
+  #?(:cljs
+     (:require-macros
+      [datomish.pair-chan :refer [go-pair <?]]
+      [datomish.node-tempfile-macros :refer [with-tempfile]]
+      [cljs.core.async.macros :as a :refer [go go-loop]]))
+  (:require
+   [datomish.api :as d]
+   [datomish.db.debug :refer [<datoms-after <datoms>= <transactions-after <shallow-entity <fulltext-values]]
+   [datomish.util :as util #?(:cljs :refer-macros :clj :refer) [raise cond-let]]
+   [datomish.schema :as ds]
+   [datomish.simple-schema]
+   [datomish.sqlite :as s]
+   [datomish.sqlite-schema]
+   [datomish.datom]
+   #?@(:clj [[datomish.jdbc-sqlite]
+             [datomish.pair-chan :refer [go-pair <?]]
+             [tempfile.core :refer [tempfile with-tempfile]]
+             [datomish.test-macros :refer [deftest-async deftest-db]]
+             [clojure.test :as t :refer [is are deftest testing]]
+             [clojure.core.async :as a :refer [go go-loop <! >!]]])
+   #?@(:cljs [[datomish.js-sqlite]
+              [datomish.pair-chan]
+              [datomish.test-macros :refer-macros [deftest-async deftest-db]]
+              [datomish.node-tempfile :refer [tempfile]]
+              [cljs.test :as t :refer-macros [is are deftest testing async]]
+              [cljs.core.async :as a :refer [<! >!]]]))
+  #?(:clj
+     (:import [clojure.lang ExceptionInfo]))
+  #?(:clj
+     (:import [datascript.db DB])))
+
+#?(:cljs
+   (def Throwable js/Error))
+
+(defn- tempids [tx]
+  (into {} (map (juxt (comp :idx first) second) (:tempids tx))))
+
+(def test-schema
+  [{:db/id        (d/id-literal :db.part/user)
+    :db/ident     :x
+    :db/unique    :db.unique/identity
+    :db/valueType :db.type/long
+    :db.install/_attribute :db.part/db}
+   {:db/id        (d/id-literal :db.part/user)
+    :db/ident     :name
+    :db/unique    :db.unique/identity
+    :db/valueType :db.type/string
+    :db.install/_attribute :db.part/db}
+   {:db/id          (d/id-literal :db.part/user)
+    :db/ident       :y
+    :db/cardinality :db.cardinality/many
+    :db/valueType   :db.type/long
+    :db.install/_attribute :db.part/db}
+   {:db/id          (d/id-literal :db.part/user)
+    :db/ident       :aka
+    :db/cardinality :db.cardinality/many
+    :db/valueType   :db.type/string
+    :db.install/_attribute :db.part/db}
+   {:db/id        (d/id-literal :db.part/user)
+    :db/ident     :age
+    :db/valueType :db.type/long
+    :db.install/_attribute :db.part/db}
+   {:db/id        (d/id-literal :db.part/user)
+    :db/ident     :email
+    :db/unique    :db.unique/identity
+    :db/valueType :db.type/string
+    :db.install/_attribute :db.part/db}
+   {:db/id        (d/id-literal :db.part/user)
+    :db/ident     :spouse
+    :db/unique    :db.unique/value
+    :db/valueType :db.type/string
+    :db.install/_attribute :db.part/db}
+   {:db/id          (d/id-literal :db.part/user)
+    :db/ident       :friends
+    :db/cardinality :db.cardinality/many
+    :db/valueType   :db.type/ref
+    :db.install/_attribute :db.part/db}
+   ])
+
+(deftest-db test-overlapping-transacts conn
+  (let [{tx0 :tx} (<? (d/<transact! conn test-schema))
+        report0   (<? (d/<transact! conn [{:db/id (d/id-literal :db.part/user -1)
+                                           :name  "Petr"}]))
+        id0       (get (tempids report0) -1)
+        n         5
+        make-t    (fn [i]
+                    ;; Be aware that a go block with a parking operation here
+                    ;; can change the order of transaction evaluation, since the
+                    ;; parking operation will be unparked non-deterministically.
+                    (d/<transact! conn [{:db/id (d/id-literal :db.part/user -1)
+                                         :name  "Petr"
+                                         :email (str "@" i)}]))]
+
+    ;; Wait for all transactions to complete.
+    (<! (a/into []
+                (a/merge
+                  (map make-t (range n)))))
+
+    ;; Transactions should be processed in order.  This is an awkward way to
+    ;; express the expected data, but it's robust in the face of changing default
+    ;; identities, transaction numbers, and values of n.
+    (is (= (concat [[id0 :name "Petr" (+ 1 tx0) 1]
+                    [id0 :email "@0" (+ 2 tx0) 1]]
+                   (mapcat
+                     #(-> [[id0 :email (str "@" %) (+ 3 % tx0) 0]
+                           [id0 :email (str "@" (inc %)) (+ 3 % tx0) 1]])
+                     (range 0 (dec n))))
+
+           (filter #(not= :db/txInstant (second %)) (<? (<transactions-after (d/db conn) tx0)))))))
+
+#_ (time (t/run-tests))

From e5917406b4441f9eb7fe3f0d8c4de2d906d1bfd7 Mon Sep 17 00:00:00 2001
From: Nick Alexander <nalexander@mozilla.com>
Date: Tue, 11 Oct 2016 20:29:43 -0700
Subject: [PATCH 4/9] Add {un}listen{-chan}! to connection. (#61)

---
 src/common/datomish/api.cljc      |  4 ++
 src/common/datomish/transact.cljc | 53 ++++++++++++++++-
 test/datomish/transact_test.cljc  | 96 +++++++++++++++++++++++++++++++
 3 files changed, 152 insertions(+), 1 deletion(-)

diff --git a/src/common/datomish/api.cljc b/src/common/datomish/api.cljc
index 567335e0..2e012f84 100644
--- a/src/common/datomish/api.cljc
+++ b/src/common/datomish/api.cljc
@@ -26,6 +26,10 @@
 
 (def <transact! transact/<transact!)
 
+(def listen! transact/listen!)
+(def listen-chan! transact/listen-chan!)
+(def unlisten-chan! transact/unlisten-chan!)
+
 ;; TODO: use Potemkin, or a subset of Potemkin that is CLJS friendly (like
 ;; https://github.com/ztellman/potemkin/issues/31) to improve this re-exporting process.
 (def <close transact/close)
diff --git a/src/common/datomish/transact.cljc b/src/common/datomish/transact.cljc
index 7e06a006..58e0d33a 100644
--- a/src/common/datomish/transact.cljc
+++ b/src/common/datomish/transact.cljc
@@ -103,8 +103,18 @@
 (declare start-transactor)
 
 (defn connection-with-db [db]
-  (let [connection
+  ;; Puts to listener-source may park if listener-mult can't distribute them fast enough.  Since the
+  ;; underlying taps are asserted to be be unblocking, the parking time should be very short.
+  (let [listener-source
+        (a/chan 1)
+
+        listener-mult
+        (a/mult listener-source) ;; Just for tapping.
+
+        connection
         (map->Connection {:current-db       (atom db)
+                          :listener-source  listener-source
+                          :listener-mult    listener-mult
                           :transact-chan    (a/chan (util/unlimited-buffer))
                           })]
     (start-transactor connection)
@@ -583,8 +593,49 @@
                                            #(-> (<with db tx-data))))]
                           ;; We only get here if the transaction is committed.
                           (reset! (:current-db conn) (:db-after report))
+                          (>! (:listener-source conn) report)
                           report)))]
               (>! result pair))
             (a/close! result)
             (>! token-chan token)
             (recur)))))))
+
+(defn listen-chan!
+  "Put reports successfully transacted against the given connection onto the given channel.
+
+  The listener sink channel must be unblocking.
+
+  Returns the channel listened to, for future unlistening."
+  [conn listener-sink]
+  {:pre [(conn? conn)]}
+  (when-not (util/unblocking-chan? listener-sink)
+    (raise "Listener sinks must be channels backed by unblocking buffers"
+           {:error :transact/bad-listener :listener-sink listener-sink}))
+  ;; Tapping an already registered sink is a no-op.
+  (a/tap (:listener-mult conn) listener-sink)
+  listener-sink)
+
+(defn- -listen-chan [f]
+  (let [c (a/chan (a/sliding-buffer 10))]
+    (go-loop []
+      (when-let [v (<! c)]
+        (do
+          (f v)
+          (recur))))
+    c))
+
+(defn listen!
+  "Evaluate the given function with reports successfully transacted against the given connection.
+
+  `f` should be a function of one argument, the transaction report.
+
+  Returns the channel listened to, for future calls to `unlisten-chan!`."
+  ([conn f]
+   {:pre [(fn? f)]}
+   (listen-chan! conn (-listen-chan f))))
+
+(defn unlisten-chan! [conn listener-sink]
+  "Stop putting reports successfully transacted against the given connection onto the given channel."
+  {:pre [(conn? conn)]}
+  ;; Untapping an un-registered sink is a no-op.
+  (a/untap (:listener-mult conn) listener-sink))
diff --git a/test/datomish/transact_test.cljc b/test/datomish/transact_test.cljc
index a23a2f25..05e28698 100644
--- a/test/datomish/transact_test.cljc
+++ b/test/datomish/transact_test.cljc
@@ -113,4 +113,100 @@
 
            (filter #(not= :db/txInstant (second %)) (<? (<transactions-after (d/db conn) tx0)))))))
 
+(deftest-db test-listeners conn
+  (let [{tx0 :tx} (<? (d/<transact! conn test-schema))
+
+        c1  (a/chan (a/dropping-buffer 5))
+        c2  (a/chan (a/dropping-buffer 5))]
+
+    (testing "no listeners is okay"
+      ;; So that we can upsert to concrete entids.
+      (<? (d/<transact! conn [[:db/add 101 :name "Ivan"]
+                              [:db/add 102 :name "Petr"]])))
+
+    (testing "listeners are added, not accidentally notified of events before they were added"
+      (d/listen-chan! conn c1)
+      (d/listen-chan! conn c2)
+      ;; This is not authoritative, because in an error situation a report may
+      ;; be put! to a listener tap outside the expected flow.  We should witness
+      ;; such an occurrence later in the test.
+      (is (= nil (a/poll! c1)))
+      (is (= nil (a/poll! c2))))
+
+    (testing "unlistening to unrecognized key is ignored"
+      (d/unlisten-chan! conn (a/chan)))
+
+    (testing "listeners observe reports"
+      (<? (d/<transact! conn [[:db/add (d/id-literal :db.part/user -1) :name "Ivan"]]))
+      (is (= {-1 101}
+             (tempids (<! c1))))
+      (is (= {-1 101}
+             (tempids (<! c2))))
+      ;; Again, not authoritative.
+      (is (= nil (a/poll! c1)))
+      (is (= nil (a/poll! c2))))
+
+    (testing "unlisten removes correct listener"
+      (d/unlisten-chan! conn c1)
+
+      (<? (d/<transact! conn [[:db/add (d/id-literal :db.part/user -2) :name "Petr"]]))
+      (is (= {-2 102}
+             (tempids (<! c2))))
+      ;; Again, not authoritative.
+      (is (= nil (a/poll! c1))))
+
+    (testing "returning to no listeners is okay"
+      (d/unlisten-chan! conn c2)
+
+      (<? (d/<transact! conn [[:db/add (d/id-literal :db.part/user -1) :name "Petr"]]))
+
+      ;; Again, not authoritative.
+      (is (= nil (a/poll! c1)))
+      (is (= nil (a/poll! c2)))
+
+      ;; This should be authoritative, however.  We should be able to put! due
+      ;; to the size of the buffer, and we should take! what we put!.
+      (>! c1 :token-1)
+      (is (= :token-1 (<! c1)))
+      (>! c2 :token-1)
+      (is (= :token-1 (<! c2))))
+
+    (testing "complains about blocking channels"
+      (is (thrown-with-msg?
+            ExceptionInfo #"unblocking buffers"
+            (d/listen-chan! conn (a/chan 1)))))
+    ))
+
+(deftest-db test-transact-in-listener conn
+  (let [{tx0 :tx} (<? (d/<transact! conn test-schema))
+
+        ;; So that we can see all transactions.
+        lc  (a/chan (a/dropping-buffer 5))
+
+        ;; A oneshot listener, to prevent infinite recursion.
+        ofl (atom false)
+        ol  (fn [report]
+              (when (compare-and-set! ofl false true)
+                ;; Asynchronously throw another transaction at the wall.  This
+                ;; upserts to the earlier one.
+                (d/<transact! conn [{:db/id (d/id-literal :db.part/user -1) :name "Ivan" :email "@1"}])))
+        ]
+
+    (testing "that we can invoke <transact! from within a listener"
+      (d/listen-chan! conn lc)
+      (d/listen! conn ol)
+
+      ;; Transact once to get started, and so that we can upsert against concrete ids.
+      (<? (d/<transact! conn [{:db/id 101 :name "Ivan"}]))
+      (is (= (+ 1 tx0) (:tx (<! lc))))
+
+      ;; The listener should have kicked off another transaction, but we can't
+      ;; wait for it explicitly.  However, we can wait for the report to hit the
+      ;; listening channel.
+      (let [r (<! lc)]
+        (is (= (+ 2 tx0) (:tx r)))
+        (is (= {-1 101}
+               (tempids r)))
+        (is (= nil (a/poll! lc)))))))
+
 #_ (time (t/run-tests))

From cea0e3d60f4c39840db6e892d78c8f871a8894e6 Mon Sep 17 00:00:00 2001
From: Nick Alexander <nalexander@mozilla.com>
Date: Thu, 13 Oct 2016 12:00:49 -0700
Subject: [PATCH 5/9] Review comment: return pair-chan; accept a result chan
 and close? flag.

---
 src/common/datomish/transact.cljc | 22 +++++++++++++---------
 test/datomish/transact_test.cljc  |  4 ++--
 2 files changed, 15 insertions(+), 11 deletions(-)

diff --git a/src/common/datomish/transact.cljc b/src/common/datomish/transact.cljc
index 58e0d33a..bc62ff05 100644
--- a/src/common/datomish/transact.cljc
+++ b/src/common/datomish/transact.cljc
@@ -570,13 +570,16 @@
   "Submits a transaction to the database for writing.
 
   Returns a pair-chan resolving to `[result error]`."
-  [conn tx-data]
-  {:pre [(conn? conn)]}
-  (let [result (a/chan 1)]
-    ;; Any race to put! is a real race between callers of <transact!.  We can't just park on put!,
-    ;; because the parked putter that is woken is non-deterministic.
-    (a/put! (:transact-chan conn) [tx-data result])
-    result))
+  ([conn tx-data]
+   (<transact! conn tx-data (a/chan 1) true))
+  ([conn tx-data result close?]
+   {:pre [(conn? conn)]}
+   ;; Any race to put! is a real race between callers of <transact!.  We can't just park on put!,
+   ;; because the parked putter that is woken is non-deterministic.
+   (a/put! (:transact-chan conn) [tx-data result close?])
+   (go-pair
+     ;; We want to return a pair-chan, no matter what kind of channel result is.
+     (<? result))))
 
 (defn- start-transactor [conn]
   (let [token-chan (a/chan 1)]
@@ -584,7 +587,7 @@
       (>! token-chan (gensym "transactor-token"))
       (loop []
         (let [token (<! token-chan)]
-          (when-let [[tx-data result] (<! (:transact-chan conn))]
+          (when-let [[tx-data result close?] (<! (:transact-chan conn))]
             (let [pair
                   (<! (go-pair ;; Catch exceptions, return the pair.
                         (let [db (db conn)
@@ -596,8 +599,9 @@
                           (>! (:listener-source conn) report)
                           report)))]
               (>! result pair))
-            (a/close! result)
             (>! token-chan token)
+            (when close?
+              (a/close! result))
             (recur)))))))
 
 (defn listen-chan!
diff --git a/test/datomish/transact_test.cljc b/test/datomish/transact_test.cljc
index 05e28698..1436eea9 100644
--- a/test/datomish/transact_test.cljc
+++ b/test/datomish/transact_test.cljc
@@ -98,8 +98,8 @@
 
     ;; Wait for all transactions to complete.
     (<! (a/into []
-                (a/merge
-                  (map make-t (range n)))))
+                (a/merge ;; pair-chan's never stop providing values; use take to force close.
+                  (map #(a/take 1 (make-t %)) (range n)))))
 
     ;; Transactions should be processed in order.  This is an awkward way to
     ;; express the expected data, but it's robust in the face of changing default

From b20c70fc2a6c7a747e645bd6c4e87ee659e23047 Mon Sep 17 00:00:00 2001
From: Nick Alexander <nalexander@mozilla.com>
Date: Thu, 13 Oct 2016 12:23:25 -0700
Subject: [PATCH 6/9] Review comment: ensure report is non-nil after
 in-transaction!.

---
 src/common/datomish/transact.cljc | 12 ++++++++----
 test/datomish/transact_test.cljc  | 12 ++++++++++++
 2 files changed, 20 insertions(+), 4 deletions(-)

diff --git a/src/common/datomish/transact.cljc b/src/common/datomish/transact.cljc
index bc62ff05..68da51dc 100644
--- a/src/common/datomish/transact.cljc
+++ b/src/common/datomish/transact.cljc
@@ -586,7 +586,7 @@
     (go
       (>! token-chan (gensym "transactor-token"))
       (loop []
-        (let [token (<! token-chan)]
+        (when-let [token (<! token-chan)]
           (when-let [[tx-data result close?] (<! (:transact-chan conn))]
             (let [pair
                   (<! (go-pair ;; Catch exceptions, return the pair.
@@ -594,10 +594,14 @@
                               report (<? (db/in-transaction!
                                            db
                                            #(-> (<with db tx-data))))]
-                          ;; We only get here if the transaction is committed.
-                          (reset! (:current-db conn) (:db-after report))
-                          (>! (:listener-source conn) report)
+                          (when report
+                            ;; <with returns non-nil or throws, but we still check report just in
+                            ;; case.  Here, in-transaction! function completed and returned non-nil,
+                            ;; so the transaction has committed.
+                            (reset! (:current-db conn) (:db-after report))
+                            (>! (:listener-source conn) report))
                           report)))]
+              ;; Even when report is nil (transaction not committed), pair is non-nil.
               (>! result pair))
             (>! token-chan token)
             (when close?
diff --git a/test/datomish/transact_test.cljc b/test/datomish/transact_test.cljc
index 1436eea9..34970f8b 100644
--- a/test/datomish/transact_test.cljc
+++ b/test/datomish/transact_test.cljc
@@ -209,4 +209,16 @@
                (tempids r)))
         (is (= nil (a/poll! lc)))))))
 
+(deftest-db test-failing-transacts conn
+  (let [{tx0 :tx} (<? (d/<transact! conn test-schema))]
+    (testing "failing transact throws"
+      (is (thrown-with-msg?
+            ExceptionInfo #"expected :db.type/string"
+            (<? (d/<transact! conn [{:db/id (d/id-literal :db.part/user -1) :name 1}])))))
+
+    (testing "transaction after bad transaction is applied"
+      (<? (d/<transact! conn [{:db/id 101 :name "Petr"}]))
+      (is (= (<? (<datoms-after (d/db conn) tx0))
+             #{[101 :name "Petr"]})))))
+
 #_ (time (t/run-tests))

From f02d508370be3f48f0d61541089a4aebe4d88c11 Mon Sep 17 00:00:00 2001
From: Nick Alexander <nalexander@mozilla.com>
Date: Thu, 13 Oct 2016 12:45:29 -0700
Subject: [PATCH 7/9] Review comment: ensure <transact! after <close is
 rejected.

---
 src/common/datomish/transact.cljc | 27 ++++++++++++++++++---------
 test/datomish/transact_test.cljc  | 21 +++++++++++++++++++++
 2 files changed, 39 insertions(+), 9 deletions(-)

diff --git a/src/common/datomish/transact.cljc b/src/common/datomish/transact.cljc
index 68da51dc..04c56e69 100644
--- a/src/common/datomish/transact.cljc
+++ b/src/common/datomish/transact.cljc
@@ -35,7 +35,9 @@
 (defprotocol IConnection
   (close
     [conn]
-    "Close this connection. Returns a pair channel of [nil error].")
+    "Close this connection. Returns a pair channel of [nil error].
+
+    Closing a closed connection is a no-op.")
 
   (db
     [conn]
@@ -45,11 +47,15 @@
     [conn]
     "Get the full transaction history DB associated with this connection."))
 
-(defrecord Connection [current-db transact-chan]
+(defrecord Connection [closed? current-db transact-chan]
   IConnection
   (close [conn]
-    (a/close! (:transact-chan conn))
-    (db/close-db @(:current-db conn)))
+    (if (compare-and-set! (:closed? conn) false true)
+      (do
+        ;; This immediately stops <transact! enqueueing new work.
+        (a/close! (:transact-chan conn))
+        (db/close-db @(:current-db conn)))
+      (go [nil nil])))
 
   (db [conn] @(:current-db conn))
 
@@ -112,7 +118,8 @@
         (a/mult listener-source) ;; Just for tapping.
 
         connection
-        (map->Connection {:current-db       (atom db)
+        (map->Connection {:closed?          (atom false)
+                          :current-db       (atom db)
                           :listener-source  listener-source
                           :listener-mult    listener-mult
                           :transact-chan    (a/chan (util/unlimited-buffer))
@@ -576,10 +583,12 @@
    {:pre [(conn? conn)]}
    ;; Any race to put! is a real race between callers of <transact!.  We can't just park on put!,
    ;; because the parked putter that is woken is non-deterministic.
-   (a/put! (:transact-chan conn) [tx-data result close?])
-   (go-pair
-     ;; We want to return a pair-chan, no matter what kind of channel result is.
-     (<? result))))
+   (let [closed? (not (a/put! (:transact-chan conn) [tx-data result close?]))]
+     (go-pair
+       ;; We want to return a pair-chan, no matter what kind of channel result is.
+       (if closed?
+         (raise "Connection is closed" {:error :transact/connection-closed})
+         (<? result))))))
 
 (defn- start-transactor [conn]
   (let [token-chan (a/chan 1)]
diff --git a/test/datomish/transact_test.cljc b/test/datomish/transact_test.cljc
index 34970f8b..ed211711 100644
--- a/test/datomish/transact_test.cljc
+++ b/test/datomish/transact_test.cljc
@@ -221,4 +221,25 @@
       (is (= (<? (<datoms-after (d/db conn) tx0))
              #{[101 :name "Petr"]})))))
 
+;; We don't use deftest-db in order to be able to close the connection ourselves.
+(deftest-async test-transact-after-close
+  (with-tempfile [t (tempfile)]
+    (let [conn (<? (d/<connect t))
+          {tx0 :tx} (<? (d/<transact! conn test-schema))]
+      (try
+        (testing "transaction before close is applied"
+          (<? (d/<transact! conn [{:db/id 101 :name "Petr"}]))
+          (is (= (<? (<datoms-after (d/db conn) tx0))
+                 #{[101 :name "Petr"]})))
+        (finally
+          (<? (d/<close conn))))
+
+      (testing "transact after close throws"
+        (is (thrown-with-msg?
+              ExceptionInfo #"Connection is closed"
+              (<? (d/<transact! conn [{:db/id (d/id-literal :db.part/user -1) :name "Petr"}])))))
+
+      ;; Closing a closed connection is a no-op.
+      (<? (d/<close conn)))))
+
 #_ (time (t/run-tests))

From 032bfafec226869b3153ce7dc358b66363c49455 Mon Sep 17 00:00:00 2001
From: Nick Alexander <nalexander@mozilla.com>
Date: Thu, 13 Oct 2016 14:24:49 -0700
Subject: [PATCH 8/9] Review comment: fail pending transactions after closing
 connection.

This is pretty difficult to test robustly, but here's a stab at it.
---
 src/common/datomish/transact.cljc | 54 ++++++++++++++++++++-----------
 test/datomish/transact_test.cljc  | 36 +++++++++++++++++++++
 2 files changed, 71 insertions(+), 19 deletions(-)

diff --git a/src/common/datomish/transact.cljc b/src/common/datomish/transact.cljc
index 04c56e69..3c369ba7 100644
--- a/src/common/datomish/transact.cljc
+++ b/src/common/datomish/transact.cljc
@@ -50,12 +50,18 @@
 (defrecord Connection [closed? current-db transact-chan]
   IConnection
   (close [conn]
-    (if (compare-and-set! (:closed? conn) false true)
-      (do
-        ;; This immediately stops <transact! enqueueing new work.
-        (a/close! (:transact-chan conn))
-        (db/close-db @(:current-db conn)))
-      (go [nil nil])))
+    (go-pair ;; Always want to return a pair-chan.
+      (when (compare-and-set! (:closed? conn) false true)
+        (let [result (a/chan 1)]
+          ;; Ask for the underlying database to be closed while (usually, after) draining the queue.
+          ;; Invariant: we see :sentinel-close in the transactor queue at most once.
+          (a/put! (:transact-chan conn) [:sentinel-close nil result true])
+          ;; This immediately stops <transact! enqueueing new transactions.
+          (a/close! (:transact-chan conn))
+          ;; The transactor will close the underlying DB after draining the queue; by waiting for
+          ;; result, we can raise any error from closing the DB and ensure that the DB is really
+          ;; closed after waiting for the connection to close.
+          (<? result)))))
 
   (db [conn] @(:current-db conn))
 
@@ -583,7 +589,7 @@
    {:pre [(conn? conn)]}
    ;; Any race to put! is a real race between callers of <transact!.  We can't just park on put!,
    ;; because the parked putter that is woken is non-deterministic.
-   (let [closed? (not (a/put! (:transact-chan conn) [tx-data result close?]))]
+   (let [closed? (not (a/put! (:transact-chan conn) [:sentinel-transact tx-data result close?]))]
      (go-pair
        ;; We want to return a pair-chan, no matter what kind of channel result is.
        (if closed?
@@ -596,20 +602,30 @@
       (>! token-chan (gensym "transactor-token"))
       (loop []
         (when-let [token (<! token-chan)]
-          (when-let [[tx-data result close?] (<! (:transact-chan conn))]
+          (when-let [[sentinel tx-data result close?] (<! (:transact-chan conn))]
             (let [pair
                   (<! (go-pair ;; Catch exceptions, return the pair.
-                        (let [db (db conn)
-                              report (<? (db/in-transaction!
-                                           db
-                                           #(-> (<with db tx-data))))]
-                          (when report
-                            ;; <with returns non-nil or throws, but we still check report just in
-                            ;; case.  Here, in-transaction! function completed and returned non-nil,
-                            ;; so the transaction has committed.
-                            (reset! (:current-db conn) (:db-after report))
-                            (>! (:listener-source conn) report))
-                          report)))]
+                        (case sentinel
+                          :sentinel-close
+                          ;; Time to close the underlying DB.
+                          (<? (db/close-db @(:current-db conn)))
+
+                          ;; Default: process the transaction.
+                          (do
+                            (when @(:closed? conn)
+                              ;; Drain enqueued transactions.
+                              (raise "Connection is closed" {:error :transact/connection-closed}))
+                            (let [db (db conn)
+                                  report (<? (db/in-transaction!
+                                               db
+                                               #(-> (<with db tx-data))))]
+                              (when report
+                                ;; <with returns non-nil or throws, but we still check report just in
+                                ;; case.  Here, in-transaction! function completed and returned non-nil,
+                                ;; so the transaction has committed.
+                                (reset! (:current-db conn) (:db-after report))
+                                (>! (:listener-source conn) report))
+                              report)))))]
               ;; Even when report is nil (transaction not committed), pair is non-nil.
               (>! result pair))
             (>! token-chan token)
diff --git a/test/datomish/transact_test.cljc b/test/datomish/transact_test.cljc
index ed211711..732e1a7c 100644
--- a/test/datomish/transact_test.cljc
+++ b/test/datomish/transact_test.cljc
@@ -242,4 +242,40 @@
       ;; Closing a closed connection is a no-op.
       (<? (d/<close conn)))))
 
+;; We don't use deftest-db in order to be able to close the connection ourselves.
+(deftest-async test-transact-queued-before-close
+  (with-tempfile [t (tempfile)]
+    (let [conn      (<? (d/<connect t))
+          {tx0 :tx} (<? (d/<transact! conn test-schema))
+
+          n      100
+          make-t (fn [i]
+                   (d/<transact! conn [{:db/id (d/id-literal :db.part/user -1)
+                                        :name  "Petr"
+                                        :email (str "@" i)}]))]
+      (try
+        (testing "close while outstanding transactions are pending"
+          ;; It's not really possible to ensure that at least one of the transactions is not
+          ;; serviced before we close, so we just start "a lot" and wait for them all to resolve.
+          (let [ts (mapv make-t (range n))]
+            ;; Give a little time for some to succeed, and then wait for one, non-deterministically.
+            (<! (a/timeout 10))
+            (a/alts! ts)
+            (<? (d/<close conn))
+
+            ;; We should have some successes and some failures.
+            (let [ps    (a/into []
+                                (a/merge ;; pair-chan's never stop providing values; use take to force close.
+                                  (map (partial a/take 1) ts)))
+                  rs    (group-by (comp some? second) (<! ps))
+                  xs    (get rs false)
+                  es    (get rs true)
+                  [v e] (first es)]
+              (is (> (count xs) 0))
+              (is (> (count es) 0))
+              (is (= {:error :transact/connection-closed} (ex-data e))))))
+        (finally
+          ;; Closing a closed connection is a no-op.
+          (<? (d/<close conn)))))))
+
 #_ (time (t/run-tests))

From a4dd7e4e9c98fad629430b40a1b2f07af8ce41db Mon Sep 17 00:00:00 2001
From: Nick Alexander <nalexander@mozilla.com>
Date: Thu, 13 Oct 2016 14:28:23 -0700
Subject: [PATCH 9/9] Review comment: make a large-ish dropping buffer for JS
 listen! consumers.

---
 src/common/datomish/transact.cljc | 12 ++++++++----
 1 file changed, 8 insertions(+), 4 deletions(-)

diff --git a/src/common/datomish/transact.cljc b/src/common/datomish/transact.cljc
index 3c369ba7..81daefc8 100644
--- a/src/common/datomish/transact.cljc
+++ b/src/common/datomish/transact.cljc
@@ -648,8 +648,9 @@
   (a/tap (:listener-mult conn) listener-sink)
   listener-sink)
 
-(defn- -listen-chan [f]
-  (let [c (a/chan (a/sliding-buffer 10))]
+(defn- -listen-chan
+  [f n]
+  (let [c (a/chan (a/dropping-buffer n))]
     (go-loop []
       (when-let [v (<! c)]
         (do
@@ -664,8 +665,11 @@
 
   Returns the channel listened to, for future calls to `unlisten-chan!`."
   ([conn f]
-   {:pre [(fn? f)]}
-   (listen-chan! conn (-listen-chan f))))
+   ;; Decently large buffer before dropping, for JS consumers.
+   (listen! conn f 1024))
+  ([conn f n]
+   {:pre [(fn? f) (pos? n)]}
+   (listen-chan! conn (-listen-chan f n))))
 
 (defn unlisten-chan! [conn listener-sink]
   "Stop putting reports successfully transacted against the given connection onto the given channel."