diff --git a/.gitea/workflows/build.yml b/.gitea/workflows/build.yml index e8000e1..445c909 100644 --- a/.gitea/workflows/build.yml +++ b/.gitea/workflows/build.yml @@ -16,13 +16,8 @@ jobs: steps: - uses: actions/checkout@v4 - - name: Test - run: | - docker run --rm \ - -v "$PWD:/app" \ - -w /app \ - clojure:temurin-21-tools-deps-alpine \ - clj -M:test + - name: Unit and PostgreSQL fencing tests + run: bash scripts/test-fencing.sh - name: Generate immutable tags id: tags diff --git a/.gitea/workflows/pr.yml b/.gitea/workflows/pr.yml new file mode 100644 index 0000000..ed3cfb3 --- /dev/null +++ b/.gitea/workflows/pr.yml @@ -0,0 +1,14 @@ +name: pr +on: + pull_request: + branches: [master] +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + persist-credentials: false + - name: Unit and PostgreSQL fencing tests + run: bash scripts/test-fencing.sh +# No registry, deployment, or production credentials in PR jobs. diff --git a/btcdata-deploy.md b/btcdata-deploy.md index a49eef7..508da8b 100644 --- a/btcdata-deploy.md +++ b/btcdata-deploy.md @@ -1,5 +1,32 @@ # btcdata Kubernetes deployment plan +## 2026-09-19 status and proposed recovery + +Kubernetes deployment already exists: one `Recreate` replica on citadel, eligible +on citadel and srv1893094. Both nodes were Ready. At 14:03 UTC the live price +WebSocket upgraded successfully but returned data last recorded at 08:36:42 UTC. +Harbor was refusing connections and Flux apps reconciliation was dependency-blocked. +No production restart or relocation was performed during this investigation. + +The `fix/websocket-failover` application change proposes bounded Binance reconnects, +60-second in-memory freshness readiness, and PostgreSQL single-writer fencing. +Every application query uses the session holding advisory lock `728194106`; loss +of that session cannot cause writes through another pooled connection. Competing +instances fail startup rather than becoming simultaneous writers. This requires +all writers to run the fenced version; legacy deployments must stay stopped. + +Companion deployment changes in `magnus/deploy` add a **suspended** watchdog: +three failed minute checks after startup grace request relocation to the other +healthy node, with a ten-minute cooldown. See `apps/btcdata/FAILOVER.md` in that +repository for review gates, acceptance tests, limitations, and rollback. This is +proposed code, not a claim that automatic failover is enabled in production. + +Run `make test` for unit tests. Run `bash scripts/test-fencing.sh` for unit tests +plus PostgreSQL lock-contention, session-loss, and pooled-lock-release tests in +isolated disposable containers. No production credentials are required. + +The remaining sections below are the original migration plan, not current status. + ## Goal and interpretation Deploy `btcdata` through the existing Flux-managed two-node k3s cluster (`srv1893094` and `citadel`), with **exactly one btcdata pod running at a time**. The pod may run on either node and Kubernetes may recreate it on the other node after failure. diff --git a/resources/system.edn b/resources/system.edn index 524269d..6aa5099 100644 --- a/resources/system.edn +++ b/resources/system.edn @@ -43,36 +43,41 @@ :db {:datasource #ig/ref :db.sql/connection} :migrate-on-init? true} + :writer/query-fn + {:conn #ig/ref :db.sql/connection + :query-fn #ig/ref :db.sql/query-fn + :migrations #ig/ref :db.sql/migrations} + :reitit.routes/api {:base-path "/api" - :query-fn #ig/ref :db.sql/query-fn + :query-fn #ig/ref :writer/query-fn :binance #ig/ref :ws/binance :frankfurter #ig/ref :frankfurter/rates :strike #ig/ref :strike/ticker} :ws/binance - {:query-fn #ig/ref :db.sql/query-fn + {:query-fn #ig/ref :writer/query-fn :uri "wss://stream.binance.com:9443/ws/btcusdt@trade" :enabled? #or [#env BINANCE_ENABLED "true"]} :kraken/ohlc - {:query-fn #ig/ref :db.sql/query-fn + {:query-fn #ig/ref :writer/query-fn :enabled? #or [#env KRAKEN_ENABLED "true"]} :kraken/ohlc-day - {:query-fn #ig/ref :db.sql/query-fn + {:query-fn #ig/ref :writer/query-fn :enabled? #or [#env KRAKEN_ENABLED "true"]} :kraken/ohlc-minute - {:query-fn #ig/ref :db.sql/query-fn + {:query-fn #ig/ref :writer/query-fn :enabled? #or [#env KRAKEN_ENABLED "true"]} :frankfurter/rates {:url #or [#env FRANKFURTER_URL "http://localhost:8080"] - :query-fn #ig/ref :db.sql/query-fn} + :query-fn #ig/ref :writer/query-fn} :strike/ticker - {:query-fn #ig/ref :db.sql/query-fn + {:query-fn #ig/ref :writer/query-fn :api-key #env STRIKE_API_KEY :sats-eur-amount #or [#env SATS_EUR_AMOUNT "55"] :enabled? #or [#env STRIKE_ENABLED "true"]}} diff --git a/scripts/test-fencing.sh b/scripts/test-fencing.sh new file mode 100644 index 0000000..45909fd --- /dev/null +++ b/scripts/test-fencing.sh @@ -0,0 +1,32 @@ +#!/usr/bin/env bash +# Disposable isolated PostgreSQL; no production credentials or published ports. +set -euo pipefail +name="btcdata-test-${GITHUB_RUN_ID:-$$}-${GITHUB_RUN_ATTEMPT:-1}" +cleanup() { + docker rm -f "${name}-runner" "$name" >/dev/null 2>&1 || true + docker network rm "$name" >/dev/null 2>&1 || true +} +trap cleanup EXIT +docker network create "$name" >/dev/null +docker run -d --name "$name" --network "$name" \ + -e POSTGRES_HOST_AUTH_METHOD=trust postgres:16-alpine >/dev/null +ready=false +for _ in $(seq 1 60); do + if docker exec "$name" pg_isready -U postgres >/dev/null 2>&1; then + ready=true + break + fi + sleep 1 +done +if [ "$ready" != true ]; then + echo 'Disposable PostgreSQL did not become ready' >&2 + exit 1 +fi +# The CI workspace is inside the runner container, not on the Docker host. +# Copy only test inputs; never mount a nonexistent host workspace or copy .env/.git. +docker create --name "${name}-runner" --network "$name" -w /app \ + -e "BTCDATA_TEST_JDBC_URL=jdbc:postgresql://$name:5432/postgres?user=postgres" \ + clojure:temurin-21-tools-deps-alpine clj -M:test >/dev/null +COPYFILE_DISABLE=1 tar --no-xattrs -cf - deps.edn src resources test env/test | docker cp - "${name}-runner:/app" +docker start --attach "${name}-runner" +exit "$(docker inspect --format '{{.State.ExitCode}}' "${name}-runner")" diff --git a/src/clj/pmagnus/btcdata/core.clj b/src/clj/pmagnus/btcdata/core.clj index d4bdb8b..e0201fd 100644 --- a/src/clj/pmagnus/btcdata/core.clj +++ b/src/clj/pmagnus/btcdata/core.clj @@ -16,6 +16,7 @@ [pmagnus.btcdata.web.routes.api] ;; WebSocket clients [pmagnus.btcdata.ws.binance] + [pmagnus.btcdata.writer] ;; Pollers [pmagnus.btcdata.kraken.ohlc] [pmagnus.btcdata.kraken.ohlc-daily] diff --git a/src/clj/pmagnus/btcdata/web/controllers/health.clj b/src/clj/pmagnus/btcdata/web/controllers/health.clj index a9df489..2f21a98 100644 --- a/src/clj/pmagnus/btcdata/web/controllers/health.clj +++ b/src/clj/pmagnus/btcdata/web/controllers/health.clj @@ -1,7 +1,8 @@ (ns pmagnus.btcdata.web.controllers.health (:require [clojure.tools.logging :as log] - [ring.util.http-response :as response]) + [ring.util.http-response :as response] + [pmagnus.btcdata.ws.binance :as binance]) (:import [java.lang.management ManagementFactory])) @@ -12,11 +13,13 @@ :app {:status "up"}})) (defn readiness-check! - "Report readiness only while the application can query PostgreSQL." - [query-fn _req] + "Require the lock-owning database session and a fresh upstream stream." + [query-fn component _req] (try (query-fn :check-database {}) - (response/ok {:status "ready"}) + (if (binance/fresh? component) + (response/ok {:status "ready"}) + (response/service-unavailable {:status "not ready" :reason "binance stale"})) (catch Exception e (log/warn e "Database readiness check failed") (response/service-unavailable {:status "not ready"})))) diff --git a/src/clj/pmagnus/btcdata/web/routes/api.clj b/src/clj/pmagnus/btcdata/web/routes/api.clj index c482ce0..8afdf6a 100644 --- a/src/clj/pmagnus/btcdata/web/routes/api.clj +++ b/src/clj/pmagnus/btcdata/web/routes/api.clj @@ -181,7 +181,7 @@ ["/health" {:get health/healthcheck!}] ["/ready" - {:get (fn [req] (health/readiness-check! (:query-fn opts) req))}] + {:get (fn [req] (health/readiness-check! (:query-fn opts) (:binance opts) req))}] ["/price/ws" {:get (fn [req] (ws-price-handler opts req)) :no-doc true diff --git a/src/clj/pmagnus/btcdata/writer.clj b/src/clj/pmagnus/btcdata/writer.clj new file mode 100644 index 0000000..a2777b9 --- /dev/null +++ b/src/clj/pmagnus/btcdata/writer.clj @@ -0,0 +1,36 @@ +(ns pmagnus.btcdata.writer + "Fence all application queries with one PostgreSQL session advisory lock. + Queries use the lock-owning connection itself: a lost session cannot keep + writing through a different pooled connection. Never reconnect this session." + (:require [integrant.core :as ig] + [next.jdbc :as jdbc])) + +(def lock-id 728194106) + +(defn guarded-query [connection query-fn] + (fn [query params] + (locking connection + (query-fn connection query params)))) + +(defmethod ig/init-key :writer/query-fn + [_ {:keys [conn query-fn]}] + (let [connection (jdbc/get-connection conn)] + (try + ;; Bound lock/query stalls; a failed session is never replaced behind the fence. + (.setNetworkTimeout connection + (reify java.util.concurrent.Executor (execute [_ task] (.run ^Runnable task))) + 10000) + (when-not (:locked (jdbc/execute-one! connection + ["select pg_try_advisory_lock(?) as locked" lock-id])) + (throw (ex-info "Another btcdata writer holds the database lock" {}))) + (with-meta (guarded-query connection query-fn) {:connection connection}) + (catch Exception e + (.close connection) + (throw e))))) + +(defmethod ig/halt-key! :writer/query-fn [_ query-fn] + (let [connection (:connection (meta query-fn))] + (locking connection + (try + (jdbc/execute-one! connection ["select pg_advisory_unlock(?)" lock-id]) + (finally (.close connection)))))) diff --git a/src/clj/pmagnus/btcdata/ws/binance.clj b/src/clj/pmagnus/btcdata/ws/binance.clj index beeb8a4..c4e5553 100644 --- a/src/clj/pmagnus/btcdata/ws/binance.clj +++ b/src/clj/pmagnus/btcdata/ws/binance.clj @@ -1,97 +1,102 @@ -(ns pmagnus.btcdata.ws.binance - (:require - [clojure.data.json :as json] - [clojure.tools.logging :as log] - [integrant.core :as ig]) - (:import - [java.net URI] - [java.net.http HttpClient WebSocket WebSocket$Listener] - [java.util.concurrent CompletableFuture CompletionStage])) - -(defn- save-price! [query-fn price] - (try - (query-fn :insert-binance-price! {:price price}) - (catch Exception e - (log/error e "Failed to save BTC price")))) - -(defn- connect! - "Opens a WebSocket to Binance trade stream and returns the WebSocket instance. - `state` is an atom with keys :running?, :last-write, :buffer. - `latest-price` is an atom reset with each throttled price for SSE consumers." - [uri query-fn state latest-price] - (let [client (HttpClient/newHttpClient) - listener (reify WebSocket$Listener - (onOpen [_ ws] - (log/info "Binance WebSocket connected") - (.request ws 1)) - - (onText [_ ws data last?] - (let [buf (:buffer @state)] - (.append buf data) - (when last? - (let [text (str buf)] - (.setLength buf 0) - (try - (let [msg (json/read-str text :key-fn keyword) - price (some-> (:p msg) bigdec)] - (when price - (let [now (System/currentTimeMillis)] - (when (> (- now (:last-write @state)) 5000) - (swap! state assoc :last-write now) - (log/info "BTC price:" (str price)) - (save-price! query-fn price) - (let [prev-price (:price @latest-price)] - (reset! latest-price - {:price price - :prev-price prev-price - :recorded-at (java.time.Instant/now)})))))) - (catch Exception e - (log/error e "Failed to parse Binance message"))))) - (let [^CompletionStage cf (CompletableFuture/completedFuture nil)] - (.request ws 1) - cf))) - - (onClose [_ _ws code reason] - (log/warn "Binance WebSocket closed:" code reason) - (when (:running? @state) - (future - (Thread/sleep 3000) - (when (:running? @state) - (log/info "Reconnecting to Binance...") - (try - (let [ws (connect! uri query-fn state latest-price)] - (swap! state assoc :ws ws)) - (catch Exception e - (log/error e "Binance reconnect failed"))))))) - - (onError [_ _ws error] - (log/error error "Binance WebSocket error")))] - (-> (.newWebSocketBuilder client) - (.buildAsync (URI. uri) listener) - (.join)))) - -(defmethod ig/init-key :ws/binance - [_ {:keys [query-fn uri enabled?] :or {enabled? "true"}}] - (if-not (= enabled? "true") - (do (log/info "Binance WebSocket disabled") - {:state (atom {:running? false}) - :latest-price (atom nil)}) - (do (log/info "Starting Binance WebSocket listener:" uri) - (let [latest-price (atom nil) - state (atom {:running? true - :last-write 0 - :buffer (StringBuilder.) - :ws nil}) - ws (connect! uri query-fn state latest-price)] - (swap! state assoc :ws ws) - {:state state - :latest-price latest-price})))) - -(defmethod ig/halt-key! :ws/binance - [_ {:keys [state]}] - (log/info "Stopping Binance WebSocket listener") - (swap! state assoc :running? false) - (when-let [ws (:ws @state)] - (try - (.sendClose ws WebSocket/NORMAL_CLOSURE "shutting down") - (catch Exception _)))) +(ns pmagnus.btcdata.ws.binance + (:require [clojure.data.json :as json] + [clojure.tools.logging :as log] + [integrant.core :as ig]) + (:import [java.net URI] + [java.net.http HttpClient WebSocket WebSocket$Listener] + [java.time Duration Instant] + [java.util.concurrent CompletableFuture TimeUnit])) + +(def stale-after-ms 60000) + +(defn fresh? + "Only in-process stream data counts; database fallback is not stream health." + ([component] (fresh? component (System/currentTimeMillis))) + ([{:keys [state latest-price]} now] + (let [{:keys [enabled? connected?]} @state + recorded-at (:recorded-at @latest-price)] + (or (false? enabled?) + (and connected? (some? recorded-at) + (<= 0 (- now (.toEpochMilli ^Instant recorded-at)) stale-after-ms)))))) + +(defn- connect! [client uri query-fn state latest-price generation] + (let [buffer (StringBuilder.) + current? #(and (:running? @state) (= generation (:generation @state))) + listener + (reify WebSocket$Listener + (onOpen [_ ws] + (if (current?) + (do (swap! state assoc :connected? true) + (log/info "Binance WebSocket connected") + (.request ws 1)) + (.abort ws))) + (onText [_ ws data last?] + (when (current?) + (.append buffer data) + (when last? + (let [text (str buffer)] + (.setLength buffer 0) + (try + (when-let [price (some-> (json/read-str text :key-fn keyword) :p bigdec)] + (let [now (System/currentTimeMillis)] + (swap! state assoc :last-message now) + (when (> (- now (:last-write @state)) 5000) + ;; Publish only after persistence succeeds. + (query-fn :insert-binance-price! {:price price}) + (swap! state assoc :last-write now) + (reset! latest-price {:price price + :prev-price (:price @latest-price) + :recorded-at (Instant/now)})))) + (catch Exception e + (log/error e "Failed to process Binance price")))))) + (.request ws 1) + (CompletableFuture/completedFuture nil)) + (onClose [_ _ code reason] + (when (current?) + (swap! state assoc :connected? false) + (log/warn "Binance WebSocket closed:" code reason)) + (CompletableFuture/completedFuture nil)) + (onError [_ _ error] + (when (current?) + (swap! state assoc :connected? false) + (log/warn error "Binance WebSocket error"))))] + (-> (.newWebSocketBuilder ^HttpClient client) + (.connectTimeout (Duration/ofSeconds 10)) + (.buildAsync (URI. uri) listener) + (.get 15 TimeUnit/SECONDS)))) + +(defmethod ig/init-key :ws/binance + [_ {:keys [query-fn uri enabled?] :or {enabled? "true"}}] + (let [enabled (= enabled? "true") + latest-price (atom nil) + state (atom {:enabled? enabled :running? enabled :connected? false + :last-write 0 :last-message 0 :generation nil :ws nil}) + client (when enabled (-> (HttpClient/newBuilder) + (.connectTimeout (Duration/ofSeconds 10)) .build)) + worker + (when enabled + (future + (try + (while (:running? @state) + (let [generation (Object.)] + (swap! state assoc :generation generation :connected? false + :last-message (System/currentTimeMillis)) + (try + (let [ws (connect! client uri query-fn state latest-price generation)] + (swap! state assoc :ws ws) + (try + (while (and (:running? @state) (:connected? @state) + (< (- (System/currentTimeMillis) (:last-message @state)) stale-after-ms)) + (Thread/sleep 1000)) + (finally (.abort ^WebSocket ws)))) + (catch InterruptedException e (throw e)) + (catch Exception e (log/warn e "Binance connection failed; retrying")) + (finally (swap! state assoc :connected? false :generation nil :ws nil)))) + (when (:running? @state) (Thread/sleep 3000))) + (catch InterruptedException _))))] + {:state state :latest-price latest-price :worker worker})) + +(defmethod ig/halt-key! :ws/binance [_ {:keys [state worker]}] + (swap! state assoc :running? false :connected? false :generation nil) + (when-let [ws (:ws @state)] (.abort ^WebSocket ws)) + (when worker (future-cancel worker))) diff --git a/test/clj/pmagnus/btcdata/web/controllers/health_test.clj b/test/clj/pmagnus/btcdata/web/controllers/health_test.clj index 64a8ed0..c855ebc 100644 --- a/test/clj/pmagnus/btcdata/web/controllers/health_test.clj +++ b/test/clj/pmagnus/btcdata/web/controllers/health_test.clj @@ -1,26 +1,24 @@ (ns pmagnus.btcdata.web.controllers.health-test - (:require - [clojure.test :refer [deftest is testing]] - [pmagnus.btcdata.web.controllers.health :as health])) + (:require [clojure.test :refer [deftest is]] + [pmagnus.btcdata.web.controllers.health :as health])) + +(def disabled-stream {:state (atom {:enabled? false}) :latest-price (atom nil)}) (deftest healthcheck-is-process-only - (let [response (health/healthcheck! {})] - (is (= 200 (:status response))) - (is (= "up" (get-in response [:body :app :status]))))) + (is (= 200 (:status (health/healthcheck! {}))))) (deftest readiness-check-tests-database - (testing "ready when the database query succeeds" - (let [called (atom nil) - query-fn (fn [query params] - (reset! called [query params]) - {:healthy 1}) - response (health/readiness-check! query-fn {})] - (is (= 200 (:status response))) - (is (= {:status "ready"} (:body response))) - (is (= [:check-database {}] @called)))) + (let [called (atom nil) + query-fn (fn [query params] (reset! called [query params]))] + (is (= 200 (:status (health/readiness-check! query-fn disabled-stream {})))) + (is (= [:check-database {}] @called))) + (is (= 503 (:status (health/readiness-check! + (fn [_ _] (throw (ex-info "database unavailable" {}))) + disabled-stream {}))))) - (testing "not ready when the database query fails" - (let [query-fn (fn [_ _] (throw (ex-info "database unavailable" {}))) - response (health/readiness-check! query-fn {})] - (is (= 503 (:status response))) - (is (= {:status "not ready"} (:body response)))))) +(deftest stale-stream-is-not-ready + (is (= 503 (:status (health/readiness-check! + (fn [_ _] {}) + {:state (atom {:enabled? true :connected? true}) + :latest-price (atom {:recorded-at java.time.Instant/EPOCH})} + {}))))) diff --git a/test/clj/pmagnus/btcdata/writer_test.clj b/test/clj/pmagnus/btcdata/writer_test.clj new file mode 100644 index 0000000..a6a1124 --- /dev/null +++ b/test/clj/pmagnus/btcdata/writer_test.clj @@ -0,0 +1,52 @@ +(ns pmagnus.btcdata.writer-test + (:require [clojure.test :refer [deftest is]] + [integrant.core :as ig] + [conman.core :as conman] + [next.jdbc :as jdbc] + [pmagnus.btcdata.writer :as writer])) + +(deftest queries-use-lock-session + (let [connection (Object.) + seen (atom nil) + query (writer/guarded-query connection + (fn [& args] (reset! seen args) :result))] + (is (= :result (query :insert-price {:price 1}))) + (is (= [connection :insert-price {:price 1}] @seen)))) + +(deftest postgres-session-fencing + ;; Optional disposable PostgreSQL integration test; never use production here. + (if-let [url (System/getenv "BTCDATA_TEST_JDBC_URL")] + (let [ds (jdbc/get-datasource {:jdbcUrl url}) + opts {:conn ds :query-fn (fn [conn _ _] + (jdbc/execute-one! conn ["select 1 as healthy"]))} + first-writer (ig/init-key :writer/query-fn opts)] + (try + (is (= 1 (:healthy (first-writer :check {})))) + (is (thrown-with-msg? clojure.lang.ExceptionInfo #"Another btcdata writer" + (ig/init-key :writer/query-fn opts))) + ;; Closing the owning session cannot transparently reconnect for writes. + (.close (:connection (meta first-writer))) + (is (thrown? Exception (first-writer :check {}))) + (let [replacement (ig/init-key :writer/query-fn opts)] + (try + (is (= 1 (:healthy (replacement :check {})))) + (finally (ig/halt-key! :writer/query-fn replacement)))) + (finally + (when-not (.isClosed (:connection (meta first-writer))) + (ig/halt-key! :writer/query-fn first-writer))))) + (println "SKIP PostgreSQL fencing integration: set BTCDATA_TEST_JDBC_URL"))) + +(deftest pooled-session-releases-lock-on-halt + (when-let [url (System/getenv "BTCDATA_TEST_JDBC_URL")] + (let [pool (conman/connect! {:jdbc-url url :maximum-pool-size 3}) + opts {:conn pool :query-fn (fn [_ _ _] :ok)}] + (try + (let [leader (ig/init-key :writer/query-fn opts)] + (is (thrown? Exception (ig/init-key :writer/query-fn opts))) + (ig/halt-key! :writer/query-fn leader)) + ;; An independent connection can acquire after pooled-session shutdown. + (let [other-opts (assoc opts :conn (jdbc/get-datasource {:jdbcUrl url})) + leader (ig/init-key :writer/query-fn other-opts)] + (is (= :ok (leader :check {}))) + (ig/halt-key! :writer/query-fn leader)) + (finally (conman/disconnect! pool)))))) diff --git a/test/clj/pmagnus/btcdata/ws/binance_test.clj b/test/clj/pmagnus/btcdata/ws/binance_test.clj new file mode 100644 index 0000000..90cb34d --- /dev/null +++ b/test/clj/pmagnus/btcdata/ws/binance_test.clj @@ -0,0 +1,41 @@ +(ns pmagnus.btcdata.ws.binance-test + (:require [clojure.test :refer [deftest is]] + [integrant.core :as ig] + [pmagnus.btcdata.ws.binance :as binance]) + (:import [java.time Instant])) + +(deftest freshness-boundaries + (let [state (atom {:enabled? true :connected? true}) + price (atom {:recorded-at (Instant/ofEpochMilli 100000)}) + component {:state state :latest-price price}] + (is (binance/fresh? component 100000)) + (is (binance/fresh? component 160000)) + (is (not (binance/fresh? component 160001))) + (is (not (binance/fresh? component 99999))) + (swap! state assoc :connected? false) + (is (not (binance/fresh? component 100000))) + (reset! price nil) + (is (not (binance/fresh? component 100000))))) + +(deftest disabled-stream-does-not-connect + (let [component (ig/init-key :ws/binance {:enabled? "false" :uri "invalid"})] + (is (binance/fresh? component)) + (is (nil? (:worker component))) + (ig/halt-key! :ws/binance component) + (is (false? (:running? @(:state component)))))) + +(deftest initial-failure-is-retried-and-worker-stops + (let [attempts (atom 0) + connect-var (ns-resolve 'pmagnus.btcdata.ws.binance 'connect!)] + (with-redefs-fn + {connect-var (fn [& _] (swap! attempts inc) (throw (ex-info "test failure" {})))} + (fn [] + (let [component (ig/init-key :ws/binance {:enabled? "true" :uri "test"})] + (try + (let [deadline (+ (System/currentTimeMillis) 6000)] + (while (and (< @attempts 2) (< (System/currentTimeMillis) deadline)) + (Thread/sleep 20))) + (is (>= @attempts 2)) + (is (not (binance/fresh? component))) + (finally (ig/halt-key! :ws/binance component))) + (is (future-cancelled? (:worker component))))))))