Repository: wilkerlucio/pathom
Pull request #141 (2020-01-29 13:54:44)

Fix caching issues when using batch resolvers with the parallel-reader
Hi,

I've reported a related issue on the slack channel recently about a
batch resolver sporadically being called with `nil` input leading to
`not-found` in the results of a query. Sadly I can't reproduce it with
a minimal example, but came up with a fix which at least resolves the
issue for me and doesn't break any tests.

In the current version `parallel-batch` removes already cached items
from the `processing-sequence`. This can lead to the resolver being
called with `nil` when all the items are already in the
cache. Furthermore this leads to no result being returned for these
items.

This PR first groups the items into cached and uncached
categories. The uncached items are processed as before, except when
there are no such items the resolver call is omitted. The cached items
don't require further processing, except their result is returned
after reading it from the cache.

Could you please give me a bit of feedback on this approach? Thanks!


Patch:
diff --git a/src/com/wsscode/pathom/connect.cljc b/src/com/wsscode/pathom/connect.cljc
index b3c75f04..37583e48 100644
--- a/src/com/wsscode/pathom/connect.cljc
+++ b/src/com/wsscode/pathom/connect.cljc
@@ -1103,29 +1103,37 @@
       (pt/tracing env (assoc trace-data ::pt/event ::call-resolver-batch)
         (if (p/cache-contains? env [resolver-sym e params])
           (<! (p/cache-read env [resolver-sym e params]))
-          (let [items-map      (->> (<? (map-async-serial #(entity-select-keys env % input) processing-sequence))
-                                    (into [] (comp
-                                               (map-indexed vector)
-                                               (filter #(all-values-valid? (second %) input))
-                                               (remove #(p/cache-contains? env [resolver-sym (second %) params]))))
-                                    (group-input-indexes))
-                items          (keys items-map)
-                _              (pt/trace env {::pt/event ::batch-items-ready
-                                              ::items    items})
-                channels       (into [] (map (fn [resolver-input]
-                                               (let [ch (async/promise-chan)]
-                                                 (p/cache-hit env [resolver-sym resolver-input params] ch)
-                                                 ch))) items)
-
-                batch-result   (try
-                                 (p.async/throw-err (<?maybe (call-resolver env items)))
-                                 (catch #?(:clj Throwable :cljs :default) e
-                                   (parallel-batch-error env e)))
-
-                _              (pt/trace env {::pt/event    ::batch-result-ready
-                                              ::items-count (count batch-result)})
-
-                linked-results (zipmap items (mapv vector channels batch-result))]
+          (let [valid-inputs     (into [] (comp
+                                           (map-indexed vector)
+                                           (filter #(all-values-valid? (second %) input)))
+                                       (<? (map-async-serial #(entity-select-keys env % input)
+                                                             processing-sequence)))
+                items-map        (group-input-indexes valid-inputs)
+
+                {cached true
+                 uncached false} (group-by #(p/cache-contains? env [resolver-sym % params])
+                                           (keys items-map))
+
+                _                (pt/trace env {::pt/event ::batch-items-ready
+                                                ::items    uncached})
+
+                channels         (mapv (fn [resolver-input]
+                                         (let [ch (async/promise-chan)]
+                                           (p/cache-hit env [resolver-sym resolver-input params] ch)
+                                           ch))
+                                       uncached)
+
+                batch-result     (when (seq uncached)
+                                   (try
+                                     (p.async/throw-err (<?maybe (call-resolver env uncached)))
+                                     (catch #?(:clj Throwable :cljs :default) e
+                                       (parallel-batch-error env e))))
+
+                _                (pt/trace env {::pt/event    ::batch-result-ready
+                                                ::items-count (count batch-result)})
+
+                linked-results   (zipmap uncached (mapv vector channels batch-result))
+                cached-set       (set cached)]
 
             (if (and (not= ::p/reader-error (first batch-result))
                      (>= (count path) 2))
@@ -1140,14 +1148,16 @@
                           cache
                           (get items-map item)))
                       cache
-                      (zipmap items batch-result))))))
+                      (zipmap uncached batch-result))))))
 
             (doseq [[_ [ch value]] linked-results]
               (if value
                 (async/put! ch (or value {}))
                 (async/close! ch)))
 
-            (second (get linked-results e [nil {}]))))))))
+            (if (contains? cached-set e)
+              (<! (p/cache-read env [resolver-sym e params]))
+              (second (get linked-results e [nil {}])))))))))
 
 (defn parallel-reader
   [{::keys    [indexes max-resolver-weight]


Test patch:
diff --git a/test/com/wsscode/pathom/connect_test.cljc b/test/com/wsscode/pathom/connect_test.cljc
index 40af77e9..c8e642bb 100644
--- a/test/com/wsscode/pathom/connect_test.cljc
+++ b/test/com/wsscode/pathom/connect_test.cljc
@@ -3651,4 +3651,44 @@
                                                    :c 2}))]
                                ::p/env       {::pp/external-wait-ignore-timeout 200}}
                   '[{:>/bla [:b :c]}])
-                #:>{:bla {:c 2, :b 1}}))))))
+                #:>{:bla {:c 2, :b 1}})))
+
+       (testing "fix concurrency-related caching issues for items repeated across batches"
+         (let [things (into (sorted-map) (map #(do [% (char (+ (int \a) %))])
+                                              (range 26)))
+               to-thing-list (fn [ids]
+                               {:list-of-things (map #(do {:thing-id %}) ids)})
+               thing-list-resolver
+               (pc/resolver 'thing-list-batches
+                            {::pc/output [:thing-batches]}
+                            (fn [env _]
+                              (go {:thing-batches
+                                   (take 100
+                                         (cycle [(to-thing-list (take 20 (keys things)))
+                                                 (to-thing-list (drop 6 (keys things)))
+                                                 (to-thing-list (drop 20 (keys things)))]))})))
+               thing-value-resolver
+               (pc/resolver 'thing-value
+                            {::pc/input #{:thing-id}
+                             ::pc/output [:thing-value]
+                             ::pc/batch? true}
+                            (fn [env input]
+                              (let [c (async/chan)]
+                                (future
+                                  (Thread/sleep 1500)
+                                  (async/put! c
+                                              (if (sequential? input)
+                                                (mapv (fn [v] {:thing-value (get things (:thing-id v))})
+                                                      input)
+                                                {:thing-value (get things (:thing-id input))})))
+                                c)))
+               ;; run the test multiple times to increase the likelihood of the issue occurring
+               results (pmap (fn [_]
+                               (quick-parser {::pc/register [thing-list-resolver thing-value-resolver]}
+                                             '[{:thing-batches [{:list-of-things [:thing-value]}]}]))
+                             (range 5))]
+           (is (empty? (for [{:keys [thing-batches]} results
+                             {:keys [list-of-things]} thing-batches
+                             {:keys [thing-value]} list-of-things
+                             :when (= thing-value ::p/not-found)]
+                         thing-value))))))))