(defn latest-db [!conn]
  (m/relieve {}
    (m/observe
      (fn [!]
        (! (d/db !conn))
        (let [q (d/tx-report-queue !conn)
              t (Thread. ^Runnable
                  (fn []
                    (! (:db-after (.take ^Queue q)))
                    (recur)))]
          (.start t)
          #(do (.interrupt t) (.join t)
               (d/remove-tx-report-queue !conn)))))))

your version is correct too

How do the above approaches compare to this:

@U09FL65DK ^

I also don't understand the reference to m/signal (the upcoming primitive in missionary-next), can you spell it out for me please

datomic shares a single queue to all subscribers, so when there is more than one subscriber only one can observe a given change. m/signal acts as a proxy publisher enforcing a single subscription

For completeness, this is a modified version of Leo's latest-db above, that silences log spam — JVM threads, when they terminate due to uncaught exception, the thread calls getUncaughtExceptionHandler which seems to have a JVM default impl that prints the exception to stderr. So we silence that.

Actually this seems to leak the thread on cancellation? I think we want to use m/via per J’s original post, and pair it with the defonce singleton pattern for correct broadcast to multiple sessions. That would mean cancellation is handled by m/via , no threads ever die (since we're leveraging missionary threadpool) and therefore there shouldnt be any logspam, right?

the thread is interrupted on cancellation, why should it leak ?

Oh, the exception is signaling the thread is being harvested, i read it wrong and thought it was silencing the ThreatInterruptException that causes the thread to terminate in the first place (IIRC)