Lock OAuth refresh across concurrent ECA processes (#462)

Multiple eca server processes sharing ~/.cache/eca/db.transit.json all
read the same refresh token at startup, then each independently POSTs to
the provider when the access token nears expiry. OAuth refresh tokens
are single-use, so the first request wins, the others receive
invalid_grant, and the losers' in-memory state never resyncs from disk.

Wrap renew-auth! in a cross-process advisory lock (JVM mutex plus
FileChannel.lock on a sidecar of the global cache file). Inside the
lock, re-read the :auth slice from disk via sync-auth-from-cache! and
only call login-step :login/renew-token when the freshly-loaded
expires-at is still in the past. Losers now silently adopt the winner's
rotated tokens instead of POSTing with a stale refresh token.

Generic across all providers that route through the shared renew-auth!
entry point (Anthropic Max/Console, OpenAI ChatGPT, Copilot). MCP token
refresh is unchanged.

🤖 Generated with [eca](https://eca.dev)

Co-Authored-By: eca-agent <git@eca.dev>

Author: ericdallo

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Unreleased
 
+- Fix `invalid_grant` errors when multiple ECA processes race to refresh the same OAuth token. #462
+
 ## 0.135.0
 
 - Persist chats more durably: save after every assistant segment / tool output / error / rollback, and write the cache via atomic tmp+rename so long chats no longer get lost on crash or restart.

src/eca/db.clj

@@ -8,7 +8,8 @@
    [eca.metrics :as metrics]
    [eca.shared :as shared])
   (:import
-   [java.io OutputStream]
+   [java.io OutputStream RandomAccessFile]
+   [java.nio.channels FileChannel FileLock]
    [java.nio.file AtomicMoveNotSupportedException CopyOption Files StandardCopyOption]))
 
 (set! *warn-on-reflection* true)
@@ -262,6 +263,63 @@
       (assoc :version version)
       (upsert-cache! (transit-global-db-file) metrics)))
 
+(def ^:private global-cache-lock-sentinel (Object.))
+
+(defn ^:private global-cache-lock-file []
+  (io/file (cache/global-dir) "db.transit.json.lock"))
+
+(defn with-global-cache-lock-fn
+  "Run `f` while holding both a JVM-wide mutex and an OS advisory exclusive
+   lock on a sidecar of the global cache file. The JVM mutex avoids
+   `OverlappingFileLockException` when two threads in the same ECA server
+   race a renew; the file lock serializes across `eca server` processes
+   that share `~/.cache/eca/`. Blocks until both are acquired."
+  [f]
+  (locking global-cache-lock-sentinel
+    (let [^java.io.File lock-file (global-cache-lock-file)
+          _ (io/make-parents lock-file)
+          ^RandomAccessFile raf (RandomAccessFile. lock-file "rw")
+          ^FileChannel channel (.getChannel raf)
+          lock-ref (volatile! nil)]
+      (try
+        (vreset! lock-ref ^FileLock (.lock channel))
+        (f)
+        (finally
+          (when-let [^FileLock lock @lock-ref]
+            (try (.release lock)
+                 (catch Throwable e
+                   (logger/warn logger-tag "Could not release global cache lock" e))))
+          (try (.close channel) (catch Throwable _))
+          (try (.close raf) (catch Throwable _)))))))
+
+(defmacro with-global-cache-lock
+  "See `with-global-cache-lock-fn`. Runs `body` while holding the lock."
+  [& body]
+  `(with-global-cache-lock-fn (fn [] ~@body)))
+
+(defn sync-auth-from-cache!
+  "Re-read the global cache from disk and, if its `:auth` entry for `provider`
+   has a different `:expires-at` than the in-memory copy, overwrite the
+   in-memory `[:auth provider]` with the disk version. This lets a process
+   that lost a token-refresh race adopt the winner's freshly rotated tokens
+   instead of POSTing with a stale refresh token.
+
+   Returns a truthy value when in-memory state was updated."
+  [db* provider metrics]
+  (try
+    (when-let [disk-auth (some-> (read-global-cache metrics) :auth (get provider))]
+      (let [in-mem-auth (get-in @db* [:auth provider])]
+        (when (and (:expires-at disk-auth)
+                   (not= (:expires-at disk-auth) (:expires-at in-mem-auth)))
+          (logger/info logger-tag
+                       (format "Adopting %s auth tokens refreshed by peer process (expires-at %s)"
+                               provider (:expires-at disk-auth)))
+          (swap! db* assoc-in [:auth provider] disk-auth)
+          true)))
+    (catch Throwable e
+      (logger/warn logger-tag "Could not sync auth from cache" e)
+      false)))
+
 (defn cleanup-old-chats!
   "Deletes chats older than retention-days from the db and flushes the workspace cache.
    When retention-days is non-positive, cleanup is disabled."

src/eca/features/login.clj

@@ -3,10 +3,13 @@
    [clojure.string :as string]
    [eca.config :as config]
    [eca.db :as db]
+   [eca.logger :as logger]
    [eca.messenger :as messenger]
    [eca.models :as models]
    [eca.shared :refer [multi-str]]))
 
+(def ^:private logger-tag "[LOGIN]")
+
 (defmulti login-step (fn [ctx] [(:provider ctx) (:step ctx)]))
 
 (defmethod login-step :default [{:keys [send-msg!]}]
@@ -78,14 +81,27 @@
    {:keys [db* messenger config metrics]}
    {:keys [on-error]}]
   (try
-    (login-step
-     {:provider provider
-      :metrics metrics
-      :messenger messenger
-      :config config
-      :step :login/renew-token
-      :db* db*})
-    (db/update-global-cache! @db* metrics)
+    ;; Serialize across ECA processes that share `~/.cache/eca/db.transit.json`.
+    ;; OAuth refresh tokens are single-use, so two concurrent processes
+    ;; both POSTing with the same token would have one win and the other
+    ;; receive `invalid_grant`. Holding the cache lock + re-reading disk
+    ;; lets the loser adopt the winner's rotated tokens instead. #462
+    (db/with-global-cache-lock
+      (db/sync-auth-from-cache! db* provider metrics)
+      (let [expires-at (get-in @db* [:auth provider :expires-at])
+            now-plus-60 (+ 60 (quot (System/currentTimeMillis) 1000))]
+        (if (and expires-at (> (long expires-at) now-plus-60))
+          (logger/info logger-tag
+                       (format "Skipping %s renew; peer process already refreshed." provider))
+          (do
+            (login-step
+             {:provider provider
+              :metrics metrics
+              :messenger messenger
+              :config config
+              :step :login/renew-token
+              :db* db*})
+            (db/update-global-cache! @db* metrics)))))
     (catch Exception e
       (on-error (.getMessage e)))))
 

test/eca/db_test.clj

@@ -4,6 +4,7 @@
    [clojure.java.io :as io]
    [clojure.test :refer [deftest is testing]]
    [cognitect.transit :as transit]
+   [eca.cache :as cache]
    [eca.db :as db])
   (:import
    [java.io File]))
@@ -97,6 +98,94 @@
         (is (= db/version (:version (read-transit-file cache-file))))
         (finally (fs/delete-tree tmpdir))))))
 
+(deftest sync-auth-from-cache!-adopts-fresher-disk-tokens-test
+  (testing "when on-disk :auth has a different :expires-at, in-memory state is overwritten"
+    (let [tmpdir (str (fs/create-temp-dir))]
+      (with-redefs [cache/global-dir (constantly (io/file tmpdir))]
+        (try
+          (let [fresh-auth {:type :auth/oauth
+                            :mode :max
+                            :step :login/done
+                            :refresh-token "fresh-refresh"
+                            :api-key "fresh-access"
+                            :expires-at 9999999999}
+                disk-db (atom {:auth {"anthropic" fresh-auth}})
+                _ (db/update-global-cache! @disk-db nil)
+                stale-mem (atom {:auth {"anthropic" {:type :auth/oauth
+                                                     :mode :max
+                                                     :step :login/done
+                                                     :refresh-token "stale-refresh"
+                                                     :api-key "stale-access"
+                                                     :expires-at 1000}}})
+                updated? (db/sync-auth-from-cache! stale-mem "anthropic" nil)]
+            (is updated?)
+            (is (= "fresh-refresh" (get-in @stale-mem [:auth "anthropic" :refresh-token])))
+            (is (= "fresh-access" (get-in @stale-mem [:auth "anthropic" :api-key])))
+            (is (= 9999999999 (get-in @stale-mem [:auth "anthropic" :expires-at]))))
+          (finally (fs/delete-tree tmpdir)))))))
+
+(deftest sync-auth-from-cache!-noop-when-disk-matches-memory-test
+  (testing "when on-disk :expires-at matches memory, no update happens"
+    (let [tmpdir (str (fs/create-temp-dir))]
+      (with-redefs [cache/global-dir (constantly (io/file tmpdir))]
+        (try
+          (let [auth {:type :auth/oauth
+                      :refresh-token "same"
+                      :api-key "same"
+                      :expires-at 7777}
+                disk-db (atom {:auth {"anthropic" auth}})
+                _ (db/update-global-cache! @disk-db nil)
+                mem (atom {:auth {"anthropic" auth}})
+                updated? (db/sync-auth-from-cache! mem "anthropic" nil)]
+            (is (not updated?))
+            (is (= auth (get-in @mem [:auth "anthropic"]))))
+          (finally (fs/delete-tree tmpdir)))))))
+
+(deftest sync-auth-from-cache!-noop-when-no-disk-cache-test
+  (testing "when no global cache file exists, returns falsy and leaves memory untouched"
+    (let [tmpdir (str (fs/create-temp-dir))]
+      (with-redefs [cache/global-dir (constantly (io/file tmpdir))]
+        (try
+          (let [mem (atom {:auth {"anthropic" {:refresh-token "x" :expires-at 1}}})
+                updated? (db/sync-auth-from-cache! mem "anthropic" nil)]
+            (is (not updated?))
+            (is (= "x" (get-in @mem [:auth "anthropic" :refresh-token]))))
+          (finally (fs/delete-tree tmpdir)))))))
+
+(deftest with-global-cache-lock-runs-body-and-releases-test
+  (testing "the lock can be acquired sequentially without leaking handles"
+    (let [tmpdir (str (fs/create-temp-dir))]
+      (with-redefs [cache/global-dir (constantly (io/file tmpdir))]
+        (try
+          (let [ran (atom 0)]
+            (db/with-global-cache-lock (swap! ran inc))
+            (db/with-global-cache-lock (swap! ran inc))
+            (is (= 2 @ran)))
+          (finally (fs/delete-tree tmpdir)))))))
+
+(deftest with-global-cache-lock-serializes-concurrent-threads-test
+  (testing "two threads acquiring the lock cannot interleave inside the body"
+    (let [tmpdir (str (fs/create-temp-dir))]
+      (with-redefs [cache/global-dir (constantly (io/file tmpdir))]
+        (try
+          (let [inside (atom 0)
+                max-inside (atom 0)
+                iterations 50
+                worker (fn []
+                         (dotimes [_ iterations]
+                           (db/with-global-cache-lock
+                             (let [v (swap! inside inc)]
+                               (swap! max-inside max v)
+                               ;; small spin so a non-locking impl would observe overlap
+                               (Thread/sleep 1)
+                               (swap! inside dec)))))
+                f1 (future (worker))
+                f2 (future (worker))]
+            @f1 @f2
+            (is (= 1 @max-inside)
+                "no two threads should ever be inside the locked body at the same time"))
+          (finally (fs/delete-tree tmpdir)))))))
+
 (deftest normalize-preserves-empty-message-chats-test
   (let [normalize @#'db/normalize-db-for-workspace-write
         result (normalize {:chats {"empty" {:id "empty" :messages []}

test/eca/features/login_test.clj

@@ -1,6 +1,10 @@
 (ns eca.features.login-test
   (:require
+   [babashka.fs :as fs]
+   [clojure.java.io :as io]
    [clojure.test :refer [deftest is testing]]
+   [eca.cache :as cache]
+   [eca.db :as db]
    [eca.features.login :as login]
    [hato.client :as http]
    [matcher-combinators.test :refer [match?]]))
@@ -67,3 +71,82 @@
                                1 {}}}
 
                       @db*)))))))
+
+(def ^:private renew-auth!* @#'login/renew-auth!)
+
+(deftest renew-auth!-skips-refresh-when-peer-already-refreshed-test
+  (testing "if disk holds fresher tokens than memory, renew-auth! adopts them and does NOT call login-step"
+    (let [tmpdir (str (fs/create-temp-dir))]
+      (with-redefs [cache/global-dir (constantly (io/file tmpdir))]
+        (try
+          (let [fresh {:type :auth/oauth :mode :max :step :login/done
+                       :refresh-token "peer-fresh" :api-key "peer-access"
+                       :expires-at 9999999999}
+                stale {:type :auth/oauth :mode :max :step :login/done
+                       :refresh-token "mine-stale" :api-key "mine-access"
+                       :expires-at 1000}
+                _ (db/update-global-cache! {:auth {"anthropic" fresh}} nil)
+                db* (atom {:auth {"anthropic" stale}})
+                step-calls (atom 0)
+                on-error-msgs (atom [])]
+            (with-redefs [login/login-step (fn [_ctx] (swap! step-calls inc))]
+              (renew-auth!* "anthropic"
+                            {:db* db* :messenger nil :config nil :metrics nil}
+                            {:on-error #(swap! on-error-msgs conj %)}))
+            (is (zero? @step-calls)
+                "login-step should not be invoked when disk has fresh tokens")
+            (is (empty? @on-error-msgs))
+            (is (= "peer-fresh" (get-in @db* [:auth "anthropic" :refresh-token])))
+            (is (= "peer-access" (get-in @db* [:auth "anthropic" :api-key])))
+            (is (= 9999999999 (get-in @db* [:auth "anthropic" :expires-at]))))
+          (finally (fs/delete-tree tmpdir)))))))
+
+(deftest renew-auth!-refreshes-when-disk-also-stale-test
+  (testing "when memory and disk are both expired, renew-auth! invokes login-step exactly once and persists"
+    (let [tmpdir (str (fs/create-temp-dir))]
+      (with-redefs [cache/global-dir (constantly (io/file tmpdir))]
+        (try
+          (let [stale {:type :auth/oauth :mode :max :step :login/done
+                       :refresh-token "stale" :api-key "stale-access"
+                       :expires-at 1000}
+                _ (db/update-global-cache! {:auth {"anthropic" stale}} nil)
+                db* (atom {:auth {"anthropic" stale}})
+                step-calls (atom 0)
+                on-error-msgs (atom [])]
+            (with-redefs [login/login-step
+                          (fn [{:keys [db* provider]}]
+                            (swap! step-calls inc)
+                            (swap! db* update-in [:auth provider] merge
+                                   {:refresh-token "rotated"
+                                    :api-key "rotated-access"
+                                    :expires-at 9999999999}))]
+              (renew-auth!* "anthropic"
+                            {:db* db* :messenger nil :config nil :metrics nil}
+                            {:on-error #(swap! on-error-msgs conj %)}))
+            (is (= 1 @step-calls))
+            (is (empty? @on-error-msgs))
+            (is (= "rotated" (get-in @db* [:auth "anthropic" :refresh-token])))
+            ;; And the rotated tokens should have been written to disk so the next
+            ;; process in the race sees them.
+            (let [reloaded (atom {:auth {"anthropic" {}}})]
+              (db/sync-auth-from-cache! reloaded "anthropic" nil)
+              (is (= "rotated" (get-in @reloaded [:auth "anthropic" :refresh-token])))))
+          (finally (fs/delete-tree tmpdir)))))))
+
+(deftest renew-auth!-calls-on-error-when-login-step-throws-test
+  (testing "exceptions from the provider refresh propagate to on-error and do not persist"
+    (let [tmpdir (str (fs/create-temp-dir))]
+      (with-redefs [cache/global-dir (constantly (io/file tmpdir))]
+        (try
+          (let [stale {:type :auth/oauth :refresh-token "stale" :expires-at 1000}
+                _ (db/update-global-cache! {:auth {"anthropic" stale}} nil)
+                db* (atom {:auth {"anthropic" stale}})
+                on-error-msgs (atom [])]
+            (with-redefs [login/login-step
+                          (fn [_ctx] (throw (ex-info "Anthropic refresh token failed" {})))]
+              (renew-auth!* "anthropic"
+                            {:db* db* :messenger nil :config nil :metrics nil}
+                            {:on-error #(swap! on-error-msgs conj %)}))
+            (is (= 1 (count @on-error-msgs)))
+            (is (= "Anthropic refresh token failed" (first @on-error-msgs))))
+          (finally (fs/delete-tree tmpdir)))))))