Clojurians Log v2

what is >count here?

a discrete flow of counts, such as (m/seed [1 2 3])

or (m/reductions + 0 (m/seed [1 2 3]))

I got it working for a test case using m/reductions + 0, with m/seed & m/sleep (simulating IO delay), but when I pass in this pages function (which returns an m/ap), it only yields the final value. Usage:

(dom/text "Count thing: "
  (e/server
    (let [user-eid (:db/id (d/entity db [:eacl/id signed-in-user-id]))]
      (->>
        (counting/pages db {:subject       (->user user-eid)
                            :permission    :view
                            :resource/type :server})
        (m/reductions + 0)
        (m/relieve {})
        (e/input)))))

(defn pages
  ([db filters] (pages db filters nil))
  ([db filters cursor]
   (let [page-limit 10000]
     (m/ap
       (loop [cursor cursor]
         (let [{:as         page
                page-count  :count
                next-cursor :cursor} (m/? (api db
                                            (assoc filters
                                              :limit page-limit
                                              :cursor cursor)))]
           (if (< page-count page-limit) ; count less than limit signals end of stream
             page-count
             (m/amb page-count
               (recur next-cursor)))))))))

OK, I got it working with a dirty initial m/amb for starting value, but feels non-optimal. Would be cleaner if the call-site iterated and called m/reductions + methinks. Usage:

(->> (counting/pages db {:subject       (->user user-eid)
                         :permission    :view
                         :resource/type :server})
  (m/eduction) ; do we need this?
  (m/relieve {})
  (e/input))

this is above my missionary level to find the issue but here are some ideas • your flow must have an initial value for Electric, e/input accepts only continuous flows (i.e. must be defined for all time) • I am not sure how api is defined and it may matter • it seems like the computation is too eager, at some point the upstream source needs to be asynchronous for electric to ever see intermediate states

All good, thanks – I got it working 🙂: if the pages fn returns only page-count only (no accumulation), I can ditch the m/amb with initial acc value of 0, because m/reductions + 0 has an initial value. Check it ooout (39s demo):

👏

how can I know that the flow is still computing so I can append a "+" i.e. "500k+" instead of "500k" so an intermediate value does not look like the final value while it may still be counting?

I think i need to see inside api, does your datasource indicate termination?

indices are exhausted once page-count for any iterative count < limit:

(defn api
  [db filters]                                              ; filters includes limit.
  {:pre [(pos? (:limit filters))]}                          ; EACL allows :limit -1, but we want a finite limit here.
  (m/via m/blk (impl/count-resources db filters)))

(defn pages
  ([db filters] (pages db filters nil))
  ([db filters cursor]
   (let [page-limit (or (:limit filters) 1000)]
     (m/ap
       (loop [limit page-limit
              cursor cursor]
         (let [{:as         page
                page-count  :count
                next-cursor :cursor} (m/? (api db
                                            (assoc filters
                                              :limit limit
                                              :cursor cursor)))]
           (if (< page-count page-limit) ; <== termination logic here
             page-count
             (m/amb page-count
               (recur
                 (min (round-up-to-1000 (* limit 1.1)) 50000) ; max out at 50k limit.
                 next-cursor)))))))))

I guess I could return a tuple of [done? page-count]

impl/count-resources needs to signal when the count is final and there are no more resources to count (or that it reached the limit and terminated early)

impl/count-resources takes filters incl. limit and returns {:keys [count cursor]} and signals exhaustion when (< count limit). does it need to do something special with a Missionary construct?

i believe missionary eductions support clojure.core/reduced and clojure.core/reduced? to terminate a discrete flow

I dont think (< count limit) is a sufficient condition

you need to distinguish between • "small collection is fully counted" • "large collection reached the limit and truncated" • "collection is still counting and has not reached limit"

if there are 10 matching resources, and you query with limit 3, the counts will yield [3 3 3 1], so the final 1 < 3 means EACL terminal resource indices are exhausted, which is when the recur stops. if there are 10 resources, and you query with limit 100, you get one [10] which is less than 100, so also terminates if there are 10 resources, and you query with limit 10, you'll get [10 0], also terminates

currently there is no truncation

is it possible to time the total reduction from the caller while it is yielding intermediate results? I can't use my old timer helpers because it will measure each iterative count call, instead of the recursive query I guess I can pass start-time to pages and it returns end-time (or time delta) when it terminates, along with page-count.

I've got it working with EACL's iterative count-resources to return & sum-reduce over page-count, but I don't know how to render the intermediate sums in Electric. m/? + m/reduce will await all results. How to yield the intermediate sums?

Your question is: given a batch query that streams incremental results as they become available, in the style of a lazy seq or eduction, how can Electric render the first page as it becomes available without waiting for the entire collection to be realized. That gives you the ability also to perform eductions over the seq to compute derived state such as a running count over records seen so far. You probably also want this in a virtual scroll compatible way.

in fact here is an old prototype:

> m/? + m/reduce will await all results. How to yield the intermediate sums? Your specific question is: given a discrete stream of sums, how to render that as a reactive count in electric?

(e/server (e/input (m/relieve {} >count)))