Client <> Query <> Subquery Bookkeeping

Question

Context

• I am making a system, where different clients issue queries
• A query is resolved by issuing a set of subqueries
• I have an invalidation-worker, which gets notified when subqueries go stale

Goal

• When a subquery goes stale, I want to notify the clients which have made this subquery

Solution

To do this, I am thinking of keeping a mapping. Here's a rough solution you can play with in the REPL:

(ns play
  (:require [clojure.core.async :as a :refer [go <! go-loop >!]]))

(def recordings (atom {}))

(defn record-subquery! [client-id query-n subquery-n]
  (swap! recordings update subquery-n
         (fn [prev]
           (let [prev (or prev #{})]
             (conj prev [client-id query-n])))))

(defn go-subquery [client-id query-n subquery-n]
  (go
    (<! (a/timeout (rand-int 2000)))
    (record-subquery! client-id query-n subquery-n)
    {:client-id client-id
     :query-n query-n
     :subquery-n subquery-n}))

(defn go-query [client-id query-n]
  (go
    (let [subquery-ns (range query-n (+ query-n 5))]
      {:client-id client-id
       :query-n query-n
       :subqueries (->> subquery-ns
                        (map (partial go-subquery client-id query-n))
                        a/merge
                        (a/into [])
                        <!)})))

(comment
  (go (prn (<! (go-query :a 1)))))

(def client-chans {:a (a/chan)
                   :b (a/chan)})

(defn client-worker [client-id query-chan]
  (go-loop []
    (when-some [q (<! query-chan)]
      (prn (format "queried id = %s q = %s" client-id (<! (go-query client-id q))))
      (recur))))

(def invalidation-chan (a/chan))

(defn invalidation-broadcaster []
  (go (loop []
        (<! (a/timeout 1500))
        (when (>! invalidation-chan (rand-int 10))
          (recur)))))

(defn invalidation-worker [chan]
  (go-loop []
    (when-some [sq-id (<! chan)]
      (let [subs (->> sq-id (@recordings))]
        (prn (format "invalidating sq-id = %s subs = %s" sq-id subs))
        (doseq [[client-id query-n] subs]
          (>! (client-id client-chans) query-n))
        (recur)))))

(comment
  (do (client-worker :a (:a client-chans))
      (client-worker :b (:b client-chans))
      (invalidation-worker invalidation-chan)
      (invalidation-broadcaster))

  (a/close! invalidation-chan)
  (go (>! (:a client-chans) 1)))

Problem with the solution

I am a bit sad that record-subquery! is nested under go-subquery. This makes go-query stateful. I do it though to avoid the following race condition:

T0: go-query starts 
T1: subquery-1 completes 
T2: subquery-1 is invalidated 
T3: subquery-2 completes 
T4: go-query completes

In this scenario, we would miss the T2 update.

Would you do this differently?

Answer 1

invariant

You didn't write down any invariants, and the problem statement was not entirely clear. Here's what I heard.

We return valid query results, built up from sub-queries that are time consuming. Each sub-query may be invalidated at any time, which affects the overall result.

This sounds racy to me, as invalidation messages can be sent at any time right up until a query result is about to be returned. The strongest promise the current code could make is to say that immediately before returning it checked for invalidations and found an empty queue.

stronger promise

Suppose the channel marks all incoming messages with a timestamp or Lamport clock. And it returns a similar counter when reporting that zero messages remain. Now we can sensibly describe the validity of a query result by annotating it with that counter. There may be subsequent invalidation messages, but they will be marked with counter values greater than the one in our valid result.

---

queue depth

Consider specifying a "large" value for the channel size so that invalidators are unlikely to block. The current default size of 1 invites lock-step progress between producer and consumer.

---

post-processing

Consider performing all (time consuming!) sub-queries unconditionally, and then doing (quick) validity checks afterward, in the interest of simplicity. The idea would be to go through a tight validation loop which can exit as soon as it sees that the channel is empty.