diff --git a/README.md b/README.md index 642fec7..49232e8 100644 --- a/README.md +++ b/README.md @@ -92,6 +92,34 @@ creates them before opening SQLite. New filesystem databases on POSIX systems are created with mode `0600`; existing files keep their permissions. SQLite URI paths (`file:...`) and in-memory databases are passed through unchanged. +### Background refresh + +`:refresh-after` opts into read-triggered background refresh: + +```clojure +{:max-age (* 30 60) + :refresh-after 60} ; eligible for refresh after one minute + +{:max-age (* 30 60) + :refresh-after {:before-max-age 120}} ; eligible after twenty-eight minutes +``` + +Both forms use seconds and must resolve to an age above zero and below +`:max-age`. The map is normalized at cache construction. On a read: + +- Before `:refresh-after`, return the cached value. +- At or after `:refresh-after`, return it and start or join a background refresh. +- Missing or at/after the stored `:max-age`, wait for the shared computation. + +Refreshes run on Clojure's future pool and remain shared until stored. Success +resets the entry's age. Failure logs a warning and retains the old value until +its original expiry; later reads retry. Blocking failures propagate. + +Omitting `:refresh-after` or setting it to `nil` preserves existing behavior. +There is no timer: idle entries are not refreshed. TTL remains an idle eviction +policy. Existing databases need no migration; entries retain their stored TTL +and max-age until replaced. + ### Supported types The cache uses [transit-canon](https://github.com/latacora/transit-canon) for serialization, so out of the box it supports: diff --git a/src/com/latacora/sqlite_cache/core.clj b/src/com/latacora/sqlite_cache/core.clj index d1300fd..9b288cd 100644 --- a/src/com/latacora/sqlite_cache/core.clj +++ b/src/com/latacora/sqlite_cache/core.clj @@ -24,7 +24,8 @@ [com.latacora.sqlite-cache.maintenance :as maint]) (:import (java.nio.file FileAlreadyExistsException) - (java.util.concurrent LinkedBlockingQueue TimeUnit))) + (java.util.concurrent LinkedBlockingQueue TimeUnit) + (java.util.logging Logger Level))) ;; This currently does not use clojure.core.cache, but it probably could stand ;; to be compatible with it. The problem with clojure.core.cache is that it's @@ -61,14 +62,8 @@ ;; ... and hence all entries would be cold. (defn ^:private get! - "Gets a value from the cache based on the provided options and cache arguments. - - Options should include: - - conn: The database connection - - func-name: The function name for which the cache is being checked - - Returns the cached value if found and not stale, or nil otherwise." - [opts cache-args] + "Returns an entry with its value and refresh status, or nil when missing or stale." + [{:keys [read-conn func-name handlers refresh-after]} cache-args] ;; This function has to deal with a subtle concurrency problem. Originally, ;; the attempt to get a cache entry and the hit counter increment happened ;; inside of a transaction. If two threads try to get something form the @@ -95,21 +90,22 @@ ;; hit counter: it's not critical for correctness, just an approximate ;; statistical measure. ;; - ;; For now, we're assuming that hit updates won't block for long enough to - ;; warrant shunting them off to a separate thread for processing, and that - ;; hits can be done out-of-transaction and still be almost always correct: - ;; close enough for this purpose. - (let [{:keys [read-conn func-name handlers]} opts - q (-> (h/select :id :result) + ;; Hit updates are queued to the writer thread after lookup. Keeping writes + ;; out of the read transaction avoids the lock upgrade described above. + (let [refresh? (if refresh-after + (maint/in-the-past [:+ :created-at refresh-after]) + [:inline 0]) + q (-> (h/select :id :result [refresh? :refresh?]) (h/from :cache) (h/where [:= :function func-name] [:= :args cache-args] [:not maint/stale?])) - {:keys [id result]} (db/exec-one! read-conn q)] + {:keys [id result refresh?]} (db/exec-one! read-conn q)] (when id - (queue-write! (assoc opts :op-fn hit! :entry-id id)) - (ser/deserialize result {:handlers (:read handlers)})))) + {:id id + :value (ser/deserialize result {:handlers (:read handlers)}) + :refresh? (pos? refresh?)}))) (defn ^:private hit! "Updates hit count and last-hit time for a cache entry." @@ -143,7 +139,7 @@ (swap! put-queue dissoc queue-key))))) (def ^:private put-queue - "A queue map of `[db func args]` to a delay that will compute the result, put it + "A queue map of `[db func-name func args]` to a delay that will compute the result, put it in the cache, and remove itself from this queue. This is used to prevent more than one thread from trying to compute the same @@ -178,32 +174,68 @@ (fn [_k _atom old new] (println "put-queue changed from" old "to" new)))) +(defn- computation! + "Returns `[shared-delay new?]` for a computation keyed by + `[db func-name func cache-args]`. Registration is atomic; `new?` identifies + the caller that registered the delay. Refresh callers schedule it only when + new; callers needing a value dereference it to run or join the computation. + + On dereference, rechecks the cache because another refresh may have been + stored since the caller's lookup. A fresh entry supplies the value without + calling `func`. Otherwise, computes with `args` and queues an asynchronous + write; the value is available before that write completes. + + The registration remains until `put!` finishes its database write, so readers + arriving before persistence share the result. Reusing a fresh entry or a + failure during lookup, computation, or enqueueing removes it immediately." + [{:keys [func func-name db] :as opts} cache-args serialized-args args] + (let [queue-key [db func-name func cache-args] + computation + (delay + (try + ;; A refresh may have been stored since the caller read the old entry. + (let [entry (get! opts serialized-args)] + (if (and entry (not (:refresh? entry))) + (do + (swap! put-queue dissoc queue-key) + (:value entry)) + (let [result (apply func args)] + (queue-write! (assoc opts :op-fn put! :serialized-args serialized-args + :result result :queue-key queue-key)) + result))) + (catch Throwable e + (swap! put-queue dissoc queue-key) + (throw e)))) + [before after] (swap-vals! put-queue update queue-key #(or % computation))] + [(get after queue-key) (not (contains? before queue-key))])) + (defn ^:private cached - [{:keys [func args-cache-key db handlers] :as opts} & args] + [{:keys [args-cache-key handlers func-name] :as opts} & args] (let [cache-args (args-cache-key args) serialized-args (ser/serialize cache-args {:handlers (:write handlers)})] - (or - (get! opts serialized-args) - ;; Prevent more than 1 thread from trying to compute the same thing at once. - (let [queue-key [db func cache-args]] - (-> - (swap! - put-queue update queue-key - (fn [maybe-delay] - (or - maybe-delay ;; Another thread is already working on this. - (delay - (try - (let [result (apply func args)] - ;; Asynchronously write to cache, return immediately, rely on put-queue delay for other, simultaneous requests of the same data - (queue-write! (assoc opts :op-fn put! :serialized-args serialized-args :result result :queue-key queue-key)) - result) - (catch Exception e - ;; Clear put-queue entry on error so future calls can retry - (swap! put-queue dissoc queue-key) - (throw e))))))) - (get queue-key) - (deref)))))) + (if-let [{:keys [id value refresh?]} (get! opts serialized-args)] + (do + (queue-write! (assoc opts :op-fn hit! :entry-id id)) + (when refresh? + (let [[computation new?] (computation! opts cache-args serialized-args args)] + (when new? + (future + (try + @computation + (catch Throwable e + (.log (Logger/getLogger "com.latacora.sqlite-cache") Level/WARNING + (str "Background refresh failed for " func-name) e))))))) + value) + (let [[computation _] (computation! opts cache-args serialized-args args)] + @computation)))) + +(defn- normalize-refresh-after + [opts] + (update opts :refresh-after + (fn [refresh-after] + (if (map? refresh-after) + (- (:max-age opts) (:before-max-age refresh-after)) + refresh-after)))) (def default-opts "Default options for the cache function. @@ -248,6 +280,9 @@ - func-name: The function name for identification in the cache - ttl: Time to live in seconds - max-age: Maximum age in seconds + - refresh-after: Optional age in seconds at which a hit starts a background + refresh, or {:before-max-age seconds}. Must resolve to an age above zero + and below max-age. Missing or expired entries wait for the same computation. - args-cache-key: Function to transform arguments into cache key - handlers: Optional extra transit handlers for types not covered by the built-in defaults (which already include `java.time.*`, @@ -261,7 +296,7 @@ canonical handlers (map/set sorting, Long->BigInt) always win — those are required for cache-key correctness." [opts] - (let [{:keys [db] :as opts} (merge default-opts opts) + (let [{:keys [db] :as opts} (normalize-refresh-after (merge default-opts opts)) _ (prepare-db-file! db) read-conn (jdbc/get-connection db) write-conn (jdbc/get-connection db) diff --git a/test/com/latacora/sqlite_cache/refresh_test.clj b/test/com/latacora/sqlite_cache/refresh_test.clj new file mode 100644 index 0000000..fa9004b --- /dev/null +++ b/test/com/latacora/sqlite_cache/refresh_test.clj @@ -0,0 +1,270 @@ +(ns com.latacora.sqlite-cache.refresh-test + (:require + [clojure.test :as t] + [com.latacora.sqlite-cache.core :as c] + [com.latacora.sqlite-cache.test-utils :as tu]) + (:import (clojure.lang ExceptionInfo))) + +(defn- with-refresh-jobs [test-fn] + (let [jobs (atom []) + run-next! (fn [] + (let [[before _] (swap-vals! jobs #(subvec % 1))] + @(first before)))] + (with-redefs [clojure.core/future-call + (fn [f] + (let [run (bound-fn [] (f)) + job (delay (run))] + (swap! jobs conj job) + job))] + (test-fn jobs run-next!)))) + +(defn- in-flight [cached-fn & args] + (let [{:keys [db func-name func args-cache-key]} (meta cached-fn)] + (get @@#'c/put-queue [db func-name func (args-cache-key args)]))) + +(t/deftest refresh-option-normalization-test + (doseq [[option expected] [[nil nil] [1 1] [9 9] + [{:before-max-age 1} 9] + [{:before-max-age 9} 1]]] + (tu/with-harness + {:cache-opts {:max-age 10 :refresh-after option}} + (fn [{:keys [base-cached-fn]}] + (t/is (= expected (:refresh-after (meta base-cached-fn)))))))) + +(t/deftest refresh-boundaries-test + (doseq [refresh-after [5 {:before-max-age 5}]] + (with-refresh-jobs + (fn [jobs run-next!] + (let [calls (atom 0)] + (tu/with-harness + {:cache-opts {:max-age 10 :refresh-after refresh-after} + :f (fn [& _] (swap! calls inc))} + (fn [{:keys [cached-fn base-cached-fn advance-clock! sync-write-queue!]}] + (t/is (= 1 (cached-fn :key))) + (advance-clock! 4) + (t/is (= 1 (cached-fn :key))) + (t/is (empty? @jobs)) + (advance-clock! 1) + (t/is (= 1 (cached-fn :key))) + (dotimes [_ 10] (t/is (= 1 (cached-fn :key)))) + (t/is (= 1 @calls) "refresh is queued, not evaluated on the caller") + (t/is (= 1 (count @jobs)) "one background job per key") + (run-next!) + (sync-write-queue!) + (t/is (= 2 (cached-fn :key))) + (t/is (= 2 @calls)) + (t/is (false? (:stale? (c/entry-status base-cached-fn :key))))))))))) + +(t/deftest expired-read-evaluates-queued-refresh-test + (with-refresh-jobs + (fn [jobs run-next!] + (let [calls (atom 0)] + (tu/with-harness + {:cache-opts {:max-age 10 :refresh-after 5} + :f (fn [& _] (swap! calls inc))} + (fn [{:keys [cached-fn advance-clock!]}] + (t/is (= 1 (cached-fn :key))) + (advance-clock! 5) + (t/is (= 1 (cached-fn :key))) + (t/is (= 1 (count @jobs))) + (advance-clock! 5) + (t/is (= 2 (cached-fn :key)) "expired caller evaluates the queued shared delay") + (run-next!) + (t/is (= 2 @calls) "background job reuses the caller's computation"))))))) + +(t/deftest expired-read-joins-running-refresh-test + (let [calls (atom 0) + started (promise) + release (promise)] + (tu/with-harness + {:cache-opts {:max-age 10 :refresh-after 5} + :f (fn [& _] + (let [n (swap! calls inc)] + (when (= n 2) + (deliver started true) + @release) + n))} + (fn [{:keys [cached-fn base-cached-fn advance-clock! sync-write-queue!]}] + (let [waiter (atom nil)] + (try + (t/is (= 1 (cached-fn :key))) + (advance-clock! 5) + (t/is (= 1 (cached-fn :key))) + @started + (let [pending (in-flight base-cached-fn :key) + joined (promise) + original-computation! @#'c/computation!] + (advance-clock! 5) + (with-redefs [c/computation! + (fn [& args] + (let [result (apply original-computation! args)] + (deliver joined (first result)) + result))] + (reset! waiter (future (cached-fn :key))) + (t/is (identical? pending @joined)) + (t/is (not (realized? pending)) "refresh remains blocked until explicitly released") + (deliver release true) + (t/is (= 2 @@waiter))) + (t/is (= 2 @calls))) + (finally + (deliver release true) + (when @waiter @@waiter) + (sync-write-queue!)))))))) + +(t/deftest refresh-failure-retains-value-and-retries-test + (with-refresh-jobs + (fn [_ run-next!] + (let [calls (atom 0)] + (tu/with-harness + {:cache-opts {:max-age 10 :refresh-after 5} + :f (fn [& _] + (let [n (swap! calls inc)] + (when (= n 2) (throw (ex-info "Refresh failed" {}))) + n))} + (fn [{:keys [cached-fn base-cached-fn advance-clock! sync-write-queue!]}] + (t/is (= 1 (cached-fn :key))) + (let [created-at (:created-at (c/entry-status base-cached-fn :key))] + (advance-clock! 5) + (t/is (= 1 (cached-fn :key))) + (run-next!) + (t/is (= created-at (:created-at (c/entry-status base-cached-fn :key)))) + (t/is (= 1 (cached-fn :key)) "failed refresh leaves the previous value available") + (run-next!) + (sync-write-queue!) + (t/is (= 3 (cached-fn :key))) + (t/is (= 3 @calls))))))))) + +(t/deftest falsey-values-are-cache-hits-test + (doseq [refresh-after [nil 5] + value [nil false]] + (with-refresh-jobs + (fn [jobs run-next!] + (tu/with-harness + {:cache-opts {:max-age 10 :refresh-after refresh-after} + :f (constantly value)} + (fn [{:keys [cached-fn n-calls advance-clock! sync-write-queue!]}] + (t/is (= value (cached-fn :key))) + (t/is (= value (cached-fn :key))) + (t/is (= 1 @n-calls)) + (advance-clock! 5) + (t/is (= value (cached-fn :key))) + (when refresh-after + (t/is (= 1 (count @jobs))) + (run-next!) + (sync-write-queue!) + (t/is (= value (cached-fn :key))) + (t/is (= 2 @n-calls))))))))) + +(t/deftest refresh-remains-shared-until-stored-test + (with-refresh-jobs + (fn [_ run-next!] + (let [calls (atom 0)] + (tu/with-harness + {:cache-opts {:max-age 10 :refresh-after 5} + :f (fn [& _] (swap! calls inc))} + (fn [{:keys [cached-fn base-cached-fn advance-clock! sync-write-queue!]}] + (t/is (= 1 (cached-fn :key))) + (let [writes (atom []) + queue-write! @#'c/queue-write!] + (with-redefs [c/queue-write! + (fn [op] + (if (= (:op-fn op) @#'c/put!) + (swap! writes conj op) + (queue-write! op)))] + (advance-clock! 5) + (t/is (= 1 (cached-fn :key))) + (let [pending (in-flight base-cached-fn :key)] + (run-next!) + (dotimes [_ 10] (t/is (= 1 (cached-fn :key)))) + (t/is (identical? pending (in-flight base-cached-fn :key))) + (advance-clock! 5) + (t/is (= 2 (cached-fn :key))) + (t/is (= 2 @calls)))) + (doseq [op @writes] (queue-write! op)) + (sync-write-queue!) + (t/is (= 2 (cached-fn :key)))))))))) + +(t/deftest refresh-shares-transformed-keys-without-blocking-other-keys-test + (with-refresh-jobs + (fn [jobs run-next!] + (let [calls (atom {})] + (tu/with-harness + {:cache-opts {:max-age 10 :refresh-after 5 :args-cache-key sort} + :f (fn [& args] + (let [key (sort args) + n (get (swap! calls update key (fnil inc 0)) key)] + {:key key :generation n :print-length *print-length*}))} + (fn [{:keys [cached-fn base-cached-fn advance-clock! sync-write-queue!]}] + (t/is (= 1 (:generation (cached-fn 1 2)))) + (advance-clock! 5) + (binding [*print-length* 7] + (t/is (= 1 (:generation (cached-fn 2 1))))) + (t/is (= 1 (:generation (cached-fn 1 2)))) + (t/is (= 1 (count @jobs))) + (t/is (identical? (in-flight base-cached-fn 1 2) (in-flight base-cached-fn 2 1))) + (t/is (= 1 (:generation (cached-fn 3 4))) "another key can compute before the refresh runs") + (run-next!) + (sync-write-queue!) + (t/is (= 7 (:print-length (cached-fn 2 1)))) + (t/is (= {'(1 2) 2 '(3 4) 1} @calls)))))))) + +(t/deftest existing-entries-can-opt-into-refresh-test + (with-refresh-jobs + (fn [_ run-next!] + (let [calls (atom 0)] + (tu/with-harness + {:cache-opts {:max-age 10} + :f (fn [& _] (swap! calls inc))} + (fn [{:keys [cached-fn base-cached-fn db advance-clock!]}] + (t/is (= 1 (cached-fn :key))) + (advance-clock! 5) + (t/is (= 1 (cached-fn :key))) + (t/is (= 1 @calls)) + (let [refreshing (c/cache {:db db + :func (:func (meta base-cached-fn)) + :func-name "my-func" + :max-age 10 + :refresh-after {:before-max-age 5}})] + (t/is (= 1 (refreshing :key))) + (run-next!) + (tu/sync-write-queue! refreshing) + (t/is (= 2 (cached-fn :key)))))))))) + +(t/deftest blocking-computation-failure-after-expiry-test + (let [fail? (atom false)] + (tu/with-harness + {:cache-opts {:max-age 10 :refresh-after 5} + :f (fn [& _] + (if @fail? (throw (ex-info "Upstream unavailable" {})) :value))} + (fn [{:keys [cached-fn base-cached-fn advance-clock!]}] + (t/is (= :value (cached-fn :key))) + (reset! fail? true) + (advance-clock! 10) + (t/is (thrown-with-msg? ExceptionInfo #"Upstream unavailable" (cached-fn :key))) + (t/is (:stale? (c/entry-status base-cached-fn :key))) + (reset! fail? false) + (t/is (= :value (cached-fn :key))))))) + +(t/deftest named-caches-do-not-share-refresh-computations-test + (with-refresh-jobs + (fn [jobs run-next!] + (let [calls (atom 0)] + (tu/with-harness + {:cache-opts {:max-age 10 :refresh-after 5} + :f (fn [& _] (swap! calls inc))} + (fn [{:keys [cached-fn base-cached-fn db advance-clock! sync-write-queue!]}] + (let [other (c/cache {:db db :func (:func (meta base-cached-fn)) + :func-name "other" :max-age 10 :refresh-after 5})] + (t/is (= 1 (cached-fn :key))) + (t/is (= 2 (other :key))) + (tu/sync-write-queue! other) + (advance-clock! 5) + (t/is (= 1 (cached-fn :key))) + (t/is (= 2 (other :key))) + (t/is (= 2 (count @jobs)) "separate database entries each need a refresh") + (run-next!) + (sync-write-queue!) + (run-next!) + (tu/sync-write-queue! other) + (t/is (= 3 (cached-fn :key))) + (t/is (= 4 (other :key))))))))))