Fix stale connection errors (NoHttpResponseException)

src/ductile/conn.clj

@@ -6,20 +6,19 @@
             [ductile.schemas :refer [ConnectParams ESConn]]
             [schema.core :as s]))
 
-(def default-timeout 30000)
-
-(defn cm-options [{:keys [timeout]}]
-  {:timeout timeout
-   :threads 100
-   :default-per-route 100})
+(def default-connection-ttl 60)
+(def default-validate-after-inactivity 5000)
+(def default-threads 100)
+(def default-per-route 100)
+(def default-connection-timeout 10000)
 
 (def default-opts
   {:as :json
    :content-type :json
    :throw-exceptions false})
 
 (s/defn make-http-opts :- (s/pred map?)
-  ([{:keys [cm auth]} :- (s/maybe ESConn)
+  ([{:keys [cm auth timeouts]} :- (s/maybe ESConn)
     opts :- (s/pred map?)
     query-params-keys :- (s/maybe (s/pred coll?))
     form-params :- (s/maybe (s/pred map?))
@@ -29,35 +28,81 @@
      body (assoc :body body)
      form-params (assoc :form-params form-params)
      cm (assoc :connection-manager cm)
+     (:connection-timeout timeouts) (assoc :connection-timeout (:connection-timeout timeouts))
+     (:socket-timeout timeouts) (assoc :socket-timeout (:socket-timeout timeouts))
      (seq query-params-keys) (assoc :query-params
                                     (select-keys opts query-params-keys))))
   ([conn opts query-params-keys] (make-http-opts conn opts query-params-keys nil nil))
   ([conn opts] (make-http-opts conn opts [] nil nil))
   ([conn] (make-http-opts conn {} [] nil nil)))
 
-(defn make-connection-manager [cm-options]
-  (make-reusable-conn-manager cm-options))
+(defn make-connection-manager
+  "Create a reusable connection manager with the given options.
+
+   Options:
+   - :connection-ttl - how long connections live in the pool in seconds
+   - :threads - max total connections
+   - :default-per-route - max connections per route
+   - :insecure? - allow insecure SSL (self-signed certs)
+   - :validate-after-inactivity - check idle connections before reuse (in ms)
+     Helps prevent NoHttpResponseException from stale connections."
+  [{:keys [connection-ttl validate-after-inactivity] :as opts}]
+  (let [;; clj-http uses :timeout for TTL
+        cm-opts (-> opts
+                    (dissoc :connection-ttl :validate-after-inactivity)
+                    (assoc :timeout connection-ttl))
+        conn-mgr (make-reusable-conn-manager cm-opts)]
+    (when (some? validate-after-inactivity)
+      (.setValidateAfterInactivity conn-mgr (int validate-after-inactivity)))
+    conn-mgr))
 
 (s/defn connect :- ESConn
   "Instantiate an ES conn from ConnectParams props.
 
   To intercept all ES HTTP requests, set :request-fn
   to function with the same interface as the 1-argument
-  arity of `clj-http.client/request`."
-  [{:keys [protocol host port timeout version engine auth request-fn]
+  arity of `clj-http.client/request`.
+
+  Connection pool options:
+  - :connection-ttl - how long connections live in the pool in seconds (default: 60)
+  - :validate-after-inactivity - check idle connections before reuse, in ms (default: 5000)
+    Prevents NoHttpResponseException from stale connections closed server-side.
+  - :threads - max total connections in pool (default: 100)
+  - :default-per-route - max connections per route (default: 100)
+  - :insecure? - allow insecure SSL connections, e.g. self-signed certs (default: false)
+
+  Request timeout options (applied to every request):
+  - :connection-timeout - time to establish TCP connection in ms (default: 10000)
+  - :socket-timeout - time to wait for data in ms (default: none, for long-running operations)"
+  [{:keys [protocol host port connection-ttl version engine auth request-fn
+           validate-after-inactivity threads default-per-route insecure?
+           connection-timeout socket-timeout]
     :or {protocol :http
          request-fn client/request
-         timeout default-timeout
+         connection-ttl default-connection-ttl
+         validate-after-inactivity default-validate-after-inactivity
+         threads default-threads
+         default-per-route default-per-route
+         insecure? false
+         connection-timeout default-connection-timeout
          version 7
          engine :elasticsearch}} :- ConnectParams]
-  (let [conn {:cm (make-connection-manager
-                   (cm-options {:timeout timeout}))
+  (let [timeouts (cond-> {}
+                   connection-timeout (assoc :connection-timeout connection-timeout)
+                   socket-timeout (assoc :socket-timeout socket-timeout))
+        conn {:cm (make-connection-manager
+                   {:connection-ttl connection-ttl
+                    :validate-after-inactivity validate-after-inactivity
+                    :threads threads
+                    :default-per-route default-per-route
+                    :insecure? insecure?})
               :request-fn request-fn
               :uri (format "%s://%s:%s" (name protocol) host port)
               :version version
               :engine engine}]
     (cond-> conn
-      auth (assoc :auth (auth/http-options auth)))))
+      auth (assoc :auth (auth/http-options auth))
+      (seq timeouts) (assoc :timeouts timeouts))))
 
 (s/defn close [conn :- ESConn]
   (-> conn :cm shutdown-manager))

src/ductile/schemas.clj

@@ -2,8 +2,7 @@
   "All ES related schemas should be defined here"
   (:require [schema.core :as s]
             [schema-tools.core :as st])
-  (:import [org.apache.http.impl.conn PoolingClientConnectionManager
-            PoolingHttpClientConnectionManager]))
+  (:import [org.apache.http.impl.conn PoolingHttpClientConnectionManager]))
 
 (s/defschema RequestFn
   "A function implementing the 1-argument
@@ -33,20 +32,29 @@
     (s/optional-key :authorization) s/Str
     (s/optional-key :version) s/Int
     (s/optional-key :engine) (s/enum :elasticsearch :opensearch)
-    (s/optional-key :timeout) s/Int
+    ;; Connection pool options
+    (s/optional-key :connection-ttl) s/Int
+    (s/optional-key :validate-after-inactivity) s/Int
+    (s/optional-key :threads) s/Int
+    (s/optional-key :default-per-route) s/Int
+    (s/optional-key :insecure?) s/Bool
+    ;; Request timeout options
+    (s/optional-key :connection-timeout) s/Int
+    (s/optional-key :socket-timeout) s/Int
     (s/optional-key :auth) AuthParams
     (s/optional-key :request-fn) RequestFn}))
 
 (s/defschema ESConn
   "an ES conn is a map with a
    connection manager and an index name"
-  {:cm (s/either PoolingClientConnectionManager
-                 PoolingHttpClientConnectionManager)
+  {:cm PoolingHttpClientConnectionManager
    :uri s/Str
    :version s/Int
    :engine (s/enum :elasticsearch :opensearch)
    :request-fn RequestFn
-   (s/optional-key :auth) (s/pred map?)})
+   (s/optional-key :auth) (s/pred map?)
+   (s/optional-key :timeouts) {(s/optional-key :connection-timeout) s/Int
+                               (s/optional-key :socket-timeout) s/Int}})
 
 (s/defschema Refresh
   "ES refresh parameter, see

test/ductile/conn_test.clj

@@ -1,7 +1,10 @@
 (ns ductile.conn-test
   (:require [clojure.test :refer [deftest is testing use-fixtures]]
             [ductile.conn :as sut]
-            [schema.test :refer [validate-schemas]]))
+            [ductile.capabilities :as caps]
+            [ductile.test-helpers :as helpers]
+            [schema.test :refer [validate-schemas]])
+  (:import [org.apache.http.impl.conn PoolingHttpClientConnectionManager]))
 
 (use-fixtures :once validate-schemas)
 
@@ -26,7 +29,8 @@
               :param-2 "param-2"}
         common {:as :json
                 :content-type :json
-                :throw-exceptions false}]
+                :throw-exceptions false
+                :connection-timeout 10000}]
 
     (is (= (assoc common
                   :connection-manager cm
@@ -66,7 +70,7 @@
                   :connection-manager cm)
            (sut/make-http-opts conn-wo-auth opts [])
            (sut/make-http-opts conn-wo-auth opts nil)))
-    (is (= (assoc common
+    (is (= (assoc (dissoc common :connection-timeout)
                   :query-params {:param-1 "param-1"})
            (sut/make-http-opts nil opts [:param-1])))))
 
@@ -134,3 +138,159 @@
           "OpenSearch version 2 should be set correctly")
       (is (= :opensearch (:engine conn))
           "Engine should be :opensearch"))))
+
+(deftest validate-after-inactivity-test
+  (testing "validate-after-inactivity defaults to 5000ms"
+    (let [conn (sut/connect {:host "127.0.0.1"
+                             :port 9200})
+          ^PoolingHttpClientConnectionManager cm (:cm conn)]
+      (is (= 5000 (.getValidateAfterInactivity cm))
+          "Default validate-after-inactivity should be 5000ms")))
+
+  (testing "validate-after-inactivity can be customized"
+    (let [conn (sut/connect {:host "127.0.0.1"
+                             :port 9200
+                             :validate-after-inactivity 10000})
+          ^PoolingHttpClientConnectionManager cm (:cm conn)]
+      (is (= 10000 (.getValidateAfterInactivity cm))
+          "Custom validate-after-inactivity should be set")))
+
+  (testing "validate-after-inactivity can be disabled with 0"
+    (let [conn (sut/connect {:host "127.0.0.1"
+                             :port 9200
+                             :validate-after-inactivity 0})
+          ^PoolingHttpClientConnectionManager cm (:cm conn)]
+      ;; When 0 is passed, validation is effectively disabled
+      (is (= 0 (.getValidateAfterInactivity cm))
+          "Zero validate-after-inactivity disables validation"))))
+
+(deftest connection-pool-options-test
+  (testing "pool size defaults"
+    (let [conn (sut/connect {:host "127.0.0.1" :port 9200})
+          ^PoolingHttpClientConnectionManager cm (:cm conn)]
+      (is (= 100 (.getMaxTotal cm))
+          "Default max total connections should be 100")
+      (is (= 100 (.getDefaultMaxPerRoute cm))
+          "Default max per route should be 100")))
+
+  (testing "pool sizes can be customized"
+    (let [conn (sut/connect {:host "127.0.0.1"
+                             :port 9200
+                             :threads 50
+                             :default-per-route 25})
+          ^PoolingHttpClientConnectionManager cm (:cm conn)]
+      (is (= 50 (.getMaxTotal cm))
+          "Custom max total connections")
+      (is (= 25 (.getDefaultMaxPerRoute cm))
+          "Custom max per route")))
+
+  (testing "insecure? option is accepted"
+    ;; Just verify it doesn't throw - actual SSL behavior tested in integration
+    (is (some? (sut/connect {:host "127.0.0.1"
+                             :port 9200
+                             :insecure? true}))
+        "Connection with insecure? should succeed")))
+
+(deftest request-timeout-options-test
+  (testing "connection-timeout defaults to 10000ms"
+    (let [conn (sut/connect {:host "127.0.0.1" :port 9200})]
+      (is (= 10000 (get-in conn [:timeouts :connection-timeout]))
+          "Default connection-timeout should be 10000ms")
+      (is (nil? (get-in conn [:timeouts :socket-timeout]))
+          "socket-timeout should not be set by default")))
+
+  (testing "connection-timeout is stored in conn"
+    (let [conn (sut/connect {:host "127.0.0.1"
+                             :port 9200
+                             :connection-timeout 5000})]
+      (is (= 5000 (get-in conn [:timeouts :connection-timeout]))
+          "connection-timeout should be stored")))
+
+  (testing "socket-timeout is stored in conn"
+    (let [conn (sut/connect {:host "127.0.0.1"
+                             :port 9200
+                             :socket-timeout 10000})]
+      (is (= 10000 (get-in conn [:timeouts :socket-timeout]))
+          "socket-timeout should be stored")))
+
+  (testing "both timeouts can be set together"
+    (let [conn (sut/connect {:host "127.0.0.1"
+                             :port 9200
+                             :connection-timeout 5000
+                             :socket-timeout 10000})]
+      (is (= {:connection-timeout 5000 :socket-timeout 10000}
+             (:timeouts conn))
+          "Both timeouts should be stored")))
+
+  (testing "timeouts are applied to http options"
+    (let [conn (sut/connect {:host "127.0.0.1"
+                             :port 9200
+                             :connection-timeout 5000
+                             :socket-timeout 10000})
+          http-opts (sut/make-http-opts conn {})]
+      (is (= 5000 (:connection-timeout http-opts))
+          "connection-timeout should be in http opts")
+      (is (= 10000 (:socket-timeout http-opts))
+          "socket-timeout should be in http opts"))))
+
+(defn- connect-with-validation
+  "Connect to test ES/OpenSearch with specified validate-after-inactivity setting"
+  [engine version validate-after-inactivity]
+  (sut/connect
+   (cond-> {:host "localhost"
+            :port (helpers/engine-port engine version)
+            :version version
+            :engine engine
+            :validate-after-inactivity validate-after-inactivity}
+     (= engine :elasticsearch) (assoc :auth helpers/basic-auth-opts))))
+
+(deftest ^:integration validate-after-inactivity-behavior-test
+  (testing "connection validation allows request to succeed after idle period"
+    (helpers/for-each-es-version
+      "requests succeed with validation enabled after idle"
+      nil
+      ;; Create a connection with validation enabled (short: 1ms)
+      (let [conn-with-validation (connect-with-validation engine version 1)
+            ^PoolingHttpClientConnectionManager cm (:cm conn-with-validation)]
+        (try
+          ;; Verify validation setting is applied
+          (is (= 1 (.getValidateAfterInactivity cm))
+              "Validation should be set to 1ms")
+
+          ;; Make initial request to establish a pooled connection
+          (is (map? (caps/get-cluster-info conn-with-validation))
+              "Initial request should succeed")
+
+          ;; Wait longer than validate-after-inactivity to trigger validation
+          (Thread/sleep 10)
+
+          ;; Make another request - validation will check the connection
+          (is (map? (caps/get-cluster-info conn-with-validation))
+              "Second request should succeed after validation")
+
+          ;; Verify pool is working correctly
+          (let [stats (.getTotalStats cm)]
+            (is (>= (.getMax stats) 1)
+                "Connection pool should be configured"))
+          (finally
+            (sut/close conn-with-validation))))))
+
+  (testing "multiple requests work correctly with connection pooling and validation"
+    (helpers/for-each-es-version
+      "sequential requests with validation"
+      nil
+      (let [conn (connect-with-validation engine version 100)
+            ^PoolingHttpClientConnectionManager cm (:cm conn)]
+        (try
+          ;; Make multiple requests with delays to exercise pool behavior
+          (dotimes [_ 5]
+            (is (map? (caps/get-cluster-info conn))
+                "Each request should succeed")
+            (Thread/sleep 50))
+
+          ;; Verify no connection leaks
+          (let [stats (.getTotalStats cm)]
+            (is (<= (.getLeased stats) 0)
+                "All connections should be returned to pool after requests"))
+          (finally
+            (sut/close conn)))))))