Merge pull request 'Restart btcdata after losing its writer session' (#3) from fix/restart-closed-writer-session into master

Reviewed-on: #2

.gitea/workflows/build.yml

@@ -16,13 +16,8 @@ jobs:
     steps:
       - uses: actions/checkout@v4
 
-      - name: Test
+      - name: Unit and PostgreSQL fencing tests
-        run: |
+        run: bash scripts/test-fencing.sh
-          docker run --rm \
-            -v "$PWD:/app" \
-            -w /app \
-            clojure:temurin-21-tools-deps-alpine \
-            clj -M:test
 
       - name: Generate immutable tags
         id: tags

.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.

README.md

@@ -9,6 +9,33 @@ Bitcoin data backend service. Fetches BTC prices from Binance (WebSocket) and Kr
 | `GET /api/price/ws` | WebSocket — sends HTML fragments with live price updates (HTMX-compatible) |
 | `GET /api/price/latest` | Latest price as JSON |
 | `GET /api/health` | Health check |
+| `GET /api/price/day/:date` | Completed UTC BTC/USD daily OHLC, e.g. `/api/price/day/2020-01-01` |
+| `GET /api/price/day?date=YYYY-MM-DD` | Same daily OHLC endpoint using a query parameter |
+| `GET /api/swagger.json` | Swagger specification |
+| `GET /swagger` | Interactive Swagger UI |
+
+### Daily OHLC
+
+Dates must be valid `YYYY-MM-DD` values strictly before today in UTC. Both
+`kraken_day` and `kraken_day_history` are checked before contacting Kraken.
+Missing recent candles are fetched from the OHLC API and saved in `kraken_day`.
+Older days outside Kraken's 720-candle window are calculated from paginated
+historical trades and saved in `kraken_day_history` for subsequent requests.
+Only trades in `[00:00 UTC, next midnight UTC)` contribute; VWAP is volume-weighted.
+
+Example response (prices and volume are JSON numbers):
+
+```json
+{"date":"2020-01-01","pair":"BTC/USD","open":7200,"high":7300,"low":7100,"close":7250,"vwap":7230.5,"volume":123.45,"trade_count":1000}
+```
+
+Errors are JSON: 400 for invalid, current or future dates, 404 for days without
+trades, and 502 for Kraken failures. Failed calculations are not cached.
+Historical requests wait for completion and can take several minutes; clients
+and reverse proxies need sufficiently long request timeouts. Kraken requests
+have a 60-second per-page timeout and historical pages are spaced one second apart.
+
+The `kraken_day_history` migration runs automatically on application startup.
 
 ## Quick Start
 

btcdata-deploy.md

@@ -1,5 +1,87 @@
 # btcdata Kubernetes deployment plan
 
+## 2026-10-09 incident: closed writer session never restarted
+
+The citadel pod was Running but not Ready, and both btcprice pods were unready.
+All writer queries failed with `Connection is closed`. Readiness correctly
+returned 503, but `/api/health` always returned 200, so Kubernetes did not restart
+the process. The initial connection-loss cause was not established from the
+available logs. Fenced connections deliberately never reconnect in place.
+
+Recovery deleted only the failed pod (normal graceful deletion); its Deployment
+recreated it, and btcdata and both btcprice pods became Ready. No lock was
+bypassed or database backend terminated for this incident.
+
+The source fix makes `/api/health` return 503 when the writer session is closed
+(or absent). This checks local JDBC state only: no SQL, external APIs, or Binance
+freshness are used in liveness. Existing probes will then restart an irrecoverable
+writer after three failed checks. Readiness retains database/freshness checks.
+Unit tests passed: 21 tests, 63 assertions. PostgreSQL integration tests were
+not run because the local Docker daemon was unavailable.
+
+The recovered pod still runs `main-20260930155338`, not the new source changes.
+Flux apps reconciliation is blocked on infrastructure health checks for image
+repositories; building/publishing and a successful Flux rollout remain necessary
+to activate the liveness and keepalive source fixes.
+
+## 2026-10-07 incident: orphaned writer session
+
+The `sejs/btcdata` pod on citadel entered CrashLoopBackOff with
+`Another btcdata writer holds the database lock`. The prior pod on srv1893094
+had been removed after readiness/network failures, but its PostgreSQL backend
+(PID 3531439, client 100.75.252.111) still held advisory lock 728194106. No old
+btcdata Java process or database TCP connection remained on that node.
+
+Recovery terminated only that verified orphaned backend. Kubernetes restarted
+btcdata successfully; btcdata and both btcprice pods became Ready. Do not bypass
+the advisory lock or terminate its owner without first verifying the old writer
+is stopped.
+
+Production database `btcprod` now has connection defaults:
+`tcp_keepalives_idle=60`, `tcp_keepalives_interval=10`,
+`tcp_keepalives_count=3`, and `tcp_user_timeout=90000` (milliseconds).
+These apply to **new connections**, not the already-running recovered session.
+Server-side dead-peer detection prevents a disappeared node from retaining the
+lock for the default multi-hour TCP timeout; it does not evict a live competing
+writer. The application now also sets these values on its lock-owning session
+before acquiring the fence (source change; not yet built/deployed).
+
+Validation: 20 unit tests / 58 assertions passed; disposable PostgreSQL integration
+tests were not run because the local Docker daemon was unavailable. New
+production connection settings were verified with `SHOW`.
+
+Rollback database defaults, if necessary, with `ALTER DATABASE btcprod RESET
+<setting>` for each of the four settings above; new connections inherit the
+server defaults again. Reverting the application change requires a normal image
+build and Flux rollout. The orphan-session termination is not reversible.
+
+## 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.

deploy-citadel.sh

@@ -1,47 +0,0 @@
-#!/usr/bin/env bash
-set -euo pipefail
-
-HOST="${BTCDATA_DEPLOY_HOST:-citadel}"
-APP_DIR="${BTCDATA_DEPLOY_DIR:-/home/pmp/apps/btcdata}"
-IMAGE_NAME="${BTCDATA_IMAGE_NAME:-$(basename "$APP_DIR")-btcdata:latest}"
-PLATFORM="${BTCDATA_PLATFORM:-}"
-NETWORK="${BTCDATA_NETWORK:-btc-network}"
-FILES=(docker-compose.yml .env)
-
-if [[ -z "$PLATFORM" ]]; then
-  remote_arch="$(ssh "$HOST" uname -m)"
-  case "$remote_arch" in
-  x86_64 | amd64) PLATFORM="linux/amd64" ;;
-  aarch64 | arm64) PLATFORM="linux/arm64" ;;
-  *)
-    echo "Unsupported target architecture: $remote_arch" >&2
-    echo "Set BTCDATA_PLATFORM manually, e.g. BTCDATA_PLATFORM=linux/amd64" >&2
-    exit 1
-    ;;
-  esac
-fi
-
-for file in "${FILES[@]}"; do
-  if [[ ! -f "$file" ]]; then
-    echo "Missing required file: $file" >&2
-    exit 1
-  fi
-done
-
-echo "Building $IMAGE_NAME locally for $PLATFORM..."
-docker build --platform "$PLATFORM" -t "$IMAGE_NAME" .
-
-echo "Preparing $APP_DIR on $HOST..."
-ssh "$HOST" "mkdir -p '$APP_DIR'"
-
-echo "Ensuring external network $NETWORK exists on $HOST..."
-ssh "$HOST" "docker network inspect '$NETWORK' >/dev/null 2>&1 || docker network create '$NETWORK'"
-
-echo "Copying docker-compose.yml and .env to $HOST:$APP_DIR..."
-scp "${FILES[@]}" "$HOST:$APP_DIR/"
-
-echo "Copying Docker image $IMAGE_NAME to $HOST..."
-docker save "$IMAGE_NAME" | ssh "$HOST" docker load
-
-echo "Starting docker compose on $HOST in $APP_DIR..."
-ssh "$HOST" "cd '$APP_DIR' && docker compose up -d --no-build"

resources/migrations/20260317000000-create-kraken-day-history.down.sql

@@ -0,0 +1 @@
+DROP TABLE kraken_day_history;

resources/migrations/20260317000000-create-kraken-day-history.up.sql

@@ -0,0 +1,11 @@
+CREATE TABLE kraken_day_history (
+  ts          BIGINT         NOT NULL PRIMARY KEY,
+  open        NUMERIC(18,8)  NOT NULL,
+  high        NUMERIC(18,8)  NOT NULL,
+  low         NUMERIC(18,8)  NOT NULL,
+  close       NUMERIC(18,8)  NOT NULL,
+  vwap        NUMERIC(18,8)  NOT NULL,
+  volume      NUMERIC(24,8)  NOT NULL,
+  trade_count INTEGER        NOT NULL,
+  created_at  TIMESTAMPTZ    NOT NULL DEFAULT NOW()
+);

resources/queries.sql

@@ -47,6 +47,26 @@ SET open = EXCLUDED.open,
 -- :doc Get the most recent Kraken daily candle by timestamp
 SELECT ts FROM kraken_day ORDER BY ts DESC LIMIT 1
 
+-- :name get-kraken-day-by-ts :? :1
+SELECT ts, open, high, low, close, vwap, volume, trade_count
+FROM kraken_day WHERE ts = :ts
+
+-- :name get-kraken-day-history-by-ts :? :1
+SELECT ts, open, high, low, close, vwap, volume, trade_count
+FROM kraken_day_history WHERE ts = :ts
+
+-- :name upsert-kraken-day-history! :! :n
+INSERT INTO kraken_day_history (ts, open, high, low, close, vwap, volume, trade_count)
+VALUES (:ts, :open, :high, :low, :close, :vwap, :volume, :trade-count)
+ON CONFLICT (ts) DO UPDATE
+SET open = EXCLUDED.open,
+    high = EXCLUDED.high,
+    low = EXCLUDED.low,
+    close = EXCLUDED.close,
+    vwap = EXCLUDED.vwap,
+    volume = EXCLUDED.volume,
+    trade_count = EXCLUDED.trade_count
+
 -- :name insert-strike-price! :! :n
 -- :doc Insert a Strike ticker snapshot
 INSERT INTO strike_price (rates, quote_sats) VALUES (:rates, :quote-sats)

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"]}}

scripts/fetch-year.sh

@@ -0,0 +1,154 @@
+#!/usr/bin/env bash
+# Usage: ./scripts/fetch-year.sh YEAR [--reverse]
+# Fetches sequentially; prints JSON responses to stdout and errors to stderr.
+set -euo pipefail
+
+reverse=false
+year=
+for arg in "$@"; do
+  case "$arg" in
+    --reverse) reverse=true ;;
+    *)
+      if [[ -n $year ]]; then
+        echo "Usage: $0 YEAR [--reverse]" >&2
+        exit 1
+      fi
+      year=$arg
+      ;;
+  esac
+done
+if [[ -z $year ]]; then
+  echo "Usage: $0 YEAR [--reverse]" >&2
+  exit 1
+fi
+if [[ ! $year =~ ^[0-9]{4}$ || $year == 0000 ]]; then
+  echo "YEAR must be a four-digit year from 0001 to 9999." >&2
+  exit 1
+fi
+
+year_number=$((10#$year))
+days_in_month=(31 28 31 30 31 30 31 31 30 31 30 31)
+if (( year_number % 400 == 0 || (year_number % 4 == 0 && year_number % 100 != 0) )); then
+  days_in_month[1]=29
+fi
+
+# Reuse a reachable service, or keep a port-forward alive while fetching.
+service_ready() {
+  curl --fail --silent --output /dev/null --connect-timeout 1 --max-time 2 \
+    'http://localhost:4101/api/health'
+}
+
+forward_pid=
+cleanup() {
+  if [[ -n $forward_pid ]]; then
+    kill "$forward_pid" 2>/dev/null || true
+    wait "$forward_pid" 2>/dev/null || true
+    forward_pid=
+  fi
+}
+trap cleanup EXIT
+trap 'exit 130' INT
+trap 'exit 143' TERM
+
+wait_for_pod() {
+  local selector
+  # Use the service's selector rather than assuming a particular pod label.
+  if ! selector=$(kubectl -n sejs get svc btcdata --request-timeout=10s \
+    -o go-template='{{range $key, $value := .spec.selector}}{{$key}}={{$value}},{{end}}'); then
+    return 1
+  fi
+  selector=${selector%,}
+  if [[ -z $selector ]]; then
+    echo "Service sejs/btcdata has no pod selector." >&2
+    return 1
+  fi
+  echo "Waiting up to 5 minutes for btcdata pods to become Ready..." >&2
+  kubectl -n sejs wait --for=condition=Ready pod --selector="$selector" \
+    --timeout=300s --request-timeout=310s >&2
+}
+
+start_forward() {
+  local attempt
+  cleanup
+  if ! command -v kubectl >/dev/null 2>&1; then
+    echo "Service unavailable on localhost:4101 and kubectl is not installed." >&2
+    return 1
+  fi
+  if ! wait_for_pod; then
+    echo "btcdata pod readiness check failed; not opening a tunnel." >&2
+    return 1
+  fi
+  echo "Starting kubectl port-forward for sejs/btcdata..." >&2
+  kubectl -n sejs port-forward svc/btcdata 4101:4101 >&2 &
+  forward_pid=$!
+
+  for (( attempt = 0; attempt < 30; attempt++ )); do
+    if ! kill -0 "$forward_pid" 2>/dev/null; then
+      echo "Port-forward exited before the service became ready." >&2
+      cleanup
+      return 1
+    fi
+    if service_ready; then
+      return 0
+    fi
+    sleep 1
+  done
+  echo "Timed out waiting for the service on localhost:4101." >&2
+  cleanup
+  return 1
+}
+
+ensure_service() {
+  local retry
+  if service_ready; then
+    return 0
+  fi
+  for (( retry = 1; retry <= 3; retry++ )); do
+    echo "Recovering port-forward (attempt $retry/3)..." >&2
+    if start_forward; then
+      return 0
+    fi
+    if (( retry < 3 )); then
+      sleep 2
+    fi
+  done
+  echo "Unable to restore the service on localhost:4101." >&2
+  return 1
+}
+
+fetch_date() {
+  curl --fail --silent --show-error --connect-timeout 5 -X GET \
+    "http://localhost:4101/api/price/day/$1" \
+    -H "accept: application/json" --write-out '\n'
+}
+
+ensure_service || exit 1
+
+# Generate dates without relying on platform-specific date command options.
+status=0
+for (( month_index = 0; month_index < 12; month_index++ )); do
+  month=$((month_index + 1))
+  if [[ $reverse == true ]]; then
+    month=$((12 - month_index))
+  fi
+  days=${days_in_month[month - 1]}
+  for (( day_index = 0; day_index < days; day_index++ )); do
+    day=$((day_index + 1))
+    if [[ $reverse == true ]]; then
+      day=$((days - day_index))
+    fi
+    printf -v date '%s-%02d-%02d' "$year" "$month" "$day"
+    if ! fetch_date "$date"; then
+      if ! service_ready; then
+        ensure_service || exit 1
+        echo "Retrying $date after tunnel recovery..." >&2
+        if fetch_date "$date"; then
+          continue
+        fi
+      fi
+      echo "Failed to fetch $date" >&2
+      status=1
+    fi
+  done
+done
+exit "$status"

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")"

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]

src/clj/pmagnus/btcdata/kraken/day.clj

@@ -0,0 +1,111 @@
+(ns pmagnus.btcdata.kraken.day
+  "On-demand completed UTC candles. Database failures are not upstream failures."
+  (:require [clojure.data.json :as json])
+  (:import [java.net URI]
+           [java.net.http HttpClient HttpRequest HttpResponse$BodyHandlers]
+           [java.time Duration LocalDate ZoneOffset]
+           [java.math RoundingMode]))
+
+(defonce ^:private client
+  (delay (-> (HttpClient/newBuilder)
+             (.connectTimeout (Duration/ofSeconds 15))
+             (.build))))
+
+(defn- upstream-error [message]
+  (ex-info message {:type ::upstream}))
+
+(defn request!
+  "Fetch a Kraken public result, with a per-page timeout (not an overall deadline)."
+  [endpoint]
+  (try
+    (let [request (-> (HttpRequest/newBuilder)
+                      (.uri (URI/create (str "https://api.kraken.com/0/public/" endpoint)))
+                      (.timeout (Duration/ofSeconds 60))
+                      (.header "Accept" "application/json")
+                      (.GET) (.build))
+          response (.send ^HttpClient @client request (HttpResponse$BodyHandlers/ofString))]
+      (when-not (= 200 (.statusCode response))
+        (throw (upstream-error "Kraken returned an HTTP error")))
+      (let [body (json/read-str (.body response) :key-fn keyword :bigdec true)]
+        (when (or (seq (:error body)) (not (map? (:result body))))
+          (throw (upstream-error "Invalid or unsuccessful Kraken response")))
+        (:result body)))
+    (catch InterruptedException e
+      (.interrupt (Thread/currentThread))
+      (throw (ex-info "Kraken request interrupted" {:type ::upstream} e)))
+    (catch Exception e
+      (throw (ex-info "Failed to fetch Kraken data" {:type ::upstream} e)))))
+
+(defn- pair-rows [result]
+  (let [rows (:XXBTZUSD result)]
+    (when-not (vector? rows)
+      (throw (upstream-error "Missing BTC/USD data in Kraken response")))
+    rows))
+
+(defn- candle [[ts open high low close vwap volume trade-count]]
+  {:ts (long ts) :open (bigdec open) :high (bigdec high)
+   :low (bigdec low) :close (bigdec close) :vwap (bigdec vwap)
+   :volume (bigdec volume) :trade-count (long trade-count)})
+
+(defn- add-trade [acc [price volume]]
+  (let [p (bigdec price) v (bigdec volume)]
+    (-> (or acc {:open p :high p :low p :volume 0M :value 0M :trade-count 0})
+        (assoc :close p)
+        (update :high max p)
+        (update :low min p)
+        (update :volume + v)
+        (update :value + (* p v))
+        (update :trade-count inc))))
+
+(defn pause! [] (Thread/sleep 1000))
+
+(defn- historical-candle! [start]
+  (let [end (+ start 86400)]
+    (loop [cursor (str (*' (max 0 start) 1000000000)) acc nil]
+      (let [result (request! (str "Trades?pair=XBTUSD&since=" cursor))
+            rows (pair-rows result)
+            acc (reduce add-trade acc
+                        (filter (fn [row]
+                                  (let [ts (bigdec (nth row 2))]
+                                    (and (<= start ts) (< ts end))))
+                                rows))
+            done? (or (empty? rows) (some #(>= (bigdec (nth % 2)) end) rows))]
+        (if done?
+          (when acc
+            (when-not (pos? (:volume acc))
+              (throw (upstream-error "Kraken returned zero daily volume")))
+            (-> acc
+                (assoc :ts start :vwap (.divide ^java.math.BigDecimal (:value acc)
+                                               ^java.math.BigDecimal (:volume acc)
+                                               8 RoundingMode/HALF_UP))
+                (dissoc :value)))
+          (let [next-cursor (str (:last result))]
+            (when-not (> (bigint next-cursor) (bigint cursor))
+              (throw (upstream-error "Kraken trade pagination did not advance")))
+            (pause!)
+            (recur next-cursor acc)))))))
+
+(defn- fetch-candle! [ts]
+  (try
+    (let [rows (pair-rows (request! (str "OHLC?pair=XBTUSD&interval=1440&since=" (max 0 ts))))
+          match (some #(when (= ts (long (first %))) %) rows)]
+      (cond
+        match [:upsert-kraken-day! (candle match)]
+        (or (empty? rows) (< ts (long (ffirst rows))))
+        (when-let [c (historical-candle! ts)] [:upsert-kraken-day-history! c])
+        :else nil))
+    (catch Exception e
+      (throw (ex-info "Failed to obtain Kraken daily candle" {:type ::upstream} e)))))
+
+(defn find-day!
+  "Look up both caches before fetching. Returns a candle or nil. Date must be completed."
+  [query-fn ^LocalDate date]
+  (let [ts (.toEpochSecond (.atStartOfDay date) ZoneOffset/UTC)
+        params {:ts ts}]
+    (or (query-fn :get-kraken-day-by-ts params)
+        (query-fn :get-kraken-day-history-by-ts params)
+        (when-let [[save-query c] (fetch-candle! ts)]
+          (query-fn save-query c)
+          ;; Read stored numeric precision so first and cached responses match.
+          (query-fn (if (= save-query :upsert-kraken-day!)
+                      :get-kraken-day-by-ts :get-kraken-day-history-by-ts) params)))))

src/clj/pmagnus/btcdata/web/controllers/day.clj

@@ -0,0 +1,45 @@
+(ns pmagnus.btcdata.web.controllers.day
+  (:require [pmagnus.btcdata.kraken.day :as day])
+  (:import [java.time LocalDate ZoneOffset]
+           [java.time.format DateTimeParseException]))
+
+(defn- response [status body]
+  {:status status :body body})
+
+(defn- parse-date [s]
+  (when (and (string? s) (re-matches #"[0-9]{4}-[0-9]{2}-[0-9]{2}" s))
+    (try (LocalDate/parse s) (catch DateTimeParseException _ nil))))
+
+(defn handle [{:keys [query-fn]} req]
+  (let [s (or (get-in req [:path-params :date])
+              (get-in req [:query-params "date"]))
+        date (parse-date s)]
+    (cond
+      (nil? date) (response 400 {:error "Invalid date, use YYYY-MM-DD"})
+      (not (.isBefore date (LocalDate/now ZoneOffset/UTC)))
+      (response 400 {:error "Only completed UTC days are allowed"})
+      :else
+      (try
+        (if-let [row (day/find-day! query-fn date)]
+          (response 200 (assoc (select-keys row [:open :high :low :close :vwap :volume :trade_count])
+                               :date (str date) :pair "BTC/USD"))
+          (response 404 {:error "No BTC/USD trading data for this day"}))
+        (catch clojure.lang.ExceptionInfo e
+          (if (= ::day/upstream (:type (ex-data e)))
+            (response 502 {:error "Failed to fetch or calculate Kraken daily data"})
+            (throw e)))))))
+
+(def response-schema
+  [:map [:date :string] [:pair :string]
+   [:open number?] [:high number?] [:low number?] [:close number?]
+   [:vwap number?] [:volume number?] [:trade_count int?]])
+
+(defn route [opts parameter-location]
+  {:get {:summary "BTC/USD daily OHLC for a completed UTC date"
+         :description "Checks stored candles first. Missing recent candles are fetched from Kraken; older days are calculated from historical trades and cached. Historical requests wait for completion and may take several minutes."
+         :parameters {parameter-location [:map [:date :string]]}
+         :responses {200 {:body response-schema}
+                     400 {:body [:map [:error :string]]}
+                     404 {:body [:map [:error :string]]}
+                     502 {:body [:map [:error :string]]}}
+         :handler (partial handle opts)}})

src/clj/pmagnus/btcdata/web/controllers/health.clj

@@ -1,22 +1,31 @@
 (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]
+   [pmagnus.btcdata.writer :as writer])
   (:import
    [java.lang.management ManagementFactory]))
 
-(defn healthcheck! [_req]
+(defn healthcheck!
-  (response/ok
+  "No remote dependency checks. A permanently closed writer session needs restart."
-   {:time    (str (java.time.Instant/now))
+  [query-fn _req]
-    :up-time (.. ManagementFactory getRuntimeMXBean getUptime)
+  (if (writer/session-live? query-fn)
-    :app     {:status "up"}}))
+    (response/ok
+     {:time    (str (java.time.Instant/now))
+      :up-time (.. ManagementFactory getRuntimeMXBean getUptime)
+      :app     {:status "up"}})
+    (response/service-unavailable {:status "restart required"
+                                   :reason "writer session closed"})))
 
 (defn readiness-check!
-  "Report readiness only while the application can query PostgreSQL."
+  "Require the lock-owning database session and a fresh upstream stream."
-  [query-fn _req]
+  [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"}))))

src/clj/pmagnus/btcdata/web/handler.clj

@@ -2,6 +2,7 @@
   (:require
    [integrant.core :as ig]
    [reitit.ring :as ring]
+   [reitit.swagger-ui :as swagger-ui]
    [pmagnus.btcdata.web.middleware.core :as middleware]))
 
 (defmethod ig/init-key :handler/ring
@@ -10,7 +11,8 @@
    (router)
    (ring/routes
     (when (some? api-path)
-      (reitit.ring/create-default-handler))
+      (swagger-ui/create-swagger-ui-handler
+       {:path "/swagger" :url (str api-path "/swagger.json")}))
     (ring/create-default-handler
      {:not-found          (constantly {:status 404 :body "Page not found"})
       :method-not-allowed (constantly {:status 405 :body "Not allowed"})

src/clj/pmagnus/btcdata/web/routes/api.clj

@@ -4,6 +4,7 @@
    [integrant.core :as ig]
    [pmagnus.btcdata.frankfurter.rates :as rates]
    [pmagnus.btcdata.web.controllers.health :as health]
+   [pmagnus.btcdata.web.controllers.day :as day]
    [pmagnus.btcdata.web.controllers.transactions :as tx]
    [pmagnus.btcdata.web.middleware.exception :as exception]
    [pmagnus.btcdata.web.middleware.formats :as formats]
@@ -179,15 +180,17 @@
            :swagger {:info {:title "btcdata API"}}
            :handler (swagger/create-swagger-handler)}}]
    ["/health"
-    {:get health/healthcheck!}]
+    {:get (fn [req] (health/healthcheck! (:query-fn opts) req))}]
    ["/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
      :middleware []}]
    ["/price/latest"
     {:get (fn [req] (latest-price-handler opts req))}]
+   ["/price/day" (day/route opts :query)]
+   ["/price/day/:date" (day/route opts :path)]
    ["/currencies/:date"
     {:get (fn [req] (currencies-handler opts req))}]
    ["/strike/ws"

src/clj/pmagnus/btcdata/writer.clj

@@ -0,0 +1,55 @@
+(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 configure-session! [connection]
+  ;; Client-side network timeouts cannot reap a server session left behind by a
+  ;; dead node. Server TCP keepalives release its fence after ~90s of silence.
+  (doseq [sql ["SET tcp_keepalives_idle = 60"
+               "SET tcp_keepalives_interval = 10"
+               "SET tcp_keepalives_count = 3"
+               "SET tcp_user_timeout = 90000"]]
+    (jdbc/execute-one! connection [sql])))
+
+(defn session-live?
+  "Local-only liveness check; a closed fenced session can never recover in place."
+  [query-fn]
+  (if-let [connection (:connection (meta query-fn))]
+    (try
+      (not (.isClosed ^java.sql.Connection connection))
+      (catch Exception _ false))
+    false))
+
+(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)
+      (configure-session! connection)
+      (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))))))

src/clj/pmagnus/btcdata/ws/binance.clj

@@ -1,97 +1,102 @@
-(ns pmagnus.btcdata.ws.binance
+(ns pmagnus.btcdata.ws.binance
-  (:require
+  (:require [clojure.data.json :as json]
-   [clojure.data.json :as json]
+            [clojure.tools.logging :as log]
-   [clojure.tools.logging :as log]
+            [integrant.core :as ig])
-   [integrant.core :as ig])
+  (:import [java.net URI]
-  (:import
+           [java.net.http HttpClient WebSocket WebSocket$Listener]
-   [java.net URI]
+           [java.time Duration Instant]
-   [java.net.http HttpClient WebSocket WebSocket$Listener]
+           [java.util.concurrent CompletableFuture TimeUnit]))
-   [java.util.concurrent CompletableFuture CompletionStage]))
+
-
+(def stale-after-ms 60000)
-(defn- save-price! [query-fn price]
+
-  (try
+(defn fresh?
-    (query-fn :insert-binance-price! {:price price})
+  "Only in-process stream data counts; database fallback is not stream health."
-    (catch Exception e
+  ([component] (fresh? component (System/currentTimeMillis)))
-      (log/error e "Failed to save BTC price"))))
+  ([{:keys [state latest-price]} now]
-
+   (let [{:keys [enabled? connected?]} @state
-(defn- connect!
+         recorded-at (:recorded-at @latest-price)]
-  "Opens a WebSocket to Binance trade stream and returns the WebSocket instance.
+     (or (false? enabled?)
-   `state` is an atom with keys :running?, :last-write, :buffer.
+         (and connected? (some? recorded-at)
-   `latest-price` is an atom reset with each throttled price for SSE consumers."
+              (<= 0 (- now (.toEpochMilli ^Instant recorded-at)) stale-after-ms))))))
-  [uri query-fn state latest-price]
+
-  (let [client   (HttpClient/newHttpClient)
+(defn- connect! [client uri query-fn state latest-price generation]
-        listener (reify WebSocket$Listener
+  (let [buffer (StringBuilder.)
-                   (onOpen [_ ws]
+        current? #(and (:running? @state) (= generation (:generation @state)))
-                     (log/info "Binance WebSocket connected")
+        listener
-                     (.request ws 1))
+        (reify WebSocket$Listener
-
+          (onOpen [_ ws]
-                   (onText [_ ws data last?]
+            (if (current?)
-                     (let [buf (:buffer @state)]
+              (do (swap! state assoc :connected? true)
-                       (.append buf data)
+                  (log/info "Binance WebSocket connected")
-                       (when last?
+                  (.request ws 1))
-                         (let [text (str buf)]
+              (.abort ws)))
-                           (.setLength buf 0)
+          (onText [_ ws data last?]
-                           (try
+            (when (current?)
-                             (let [msg   (json/read-str text :key-fn keyword)
+              (.append buffer data)
-                                   price (some-> (:p msg) bigdec)]
+              (when last?
-                               (when price
+                (let [text (str buffer)]
-                                 (let [now (System/currentTimeMillis)]
+                  (.setLength buffer 0)
-                                   (when (> (- now (:last-write @state)) 5000)
+                  (try
-                                     (swap! state assoc :last-write now)
+                    (when-let [price (some-> (json/read-str text :key-fn keyword) :p bigdec)]
-                                     (log/info "BTC price:" (str price))
+                      (let [now (System/currentTimeMillis)]
-                                     (save-price! query-fn price)
+                        (swap! state assoc :last-message now)
-                                     (let [prev-price (:price @latest-price)]
+                        (when (> (- now (:last-write @state)) 5000)
-                                       (reset! latest-price
+                          ;; Publish only after persistence succeeds.
-                                               {:price      price
+                          (query-fn :insert-binance-price! {:price price})
-                                                :prev-price prev-price
+                          (swap! state assoc :last-write now)
-                                                :recorded-at (java.time.Instant/now)}))))))
+                          (reset! latest-price {:price price
-                             (catch Exception e
+                                                :prev-price (:price @latest-price)
-                               (log/error e "Failed to parse Binance message")))))
+                                                :recorded-at (Instant/now)}))))
-                       (let [^CompletionStage cf (CompletableFuture/completedFuture nil)]
+                    (catch Exception e
-                         (.request ws 1)
+                      (log/error e "Failed to process Binance price"))))))
-                         cf)))
+            (.request ws 1)
-
+            (CompletableFuture/completedFuture nil))
-                   (onClose [_ _ws code reason]
+          (onClose [_ _ code reason]
-                     (log/warn "Binance WebSocket closed:" code reason)
+            (when (current?)
-                     (when (:running? @state)
+              (swap! state assoc :connected? false)
-                       (future
+              (log/warn "Binance WebSocket closed:" code reason))
-                         (Thread/sleep 3000)
+            (CompletableFuture/completedFuture nil))
-                         (when (:running? @state)
+          (onError [_ _ error]
-                           (log/info "Reconnecting to Binance...")
+            (when (current?)
-                           (try
+              (swap! state assoc :connected? false)
-                             (let [ws (connect! uri query-fn state latest-price)]
+              (log/warn error "Binance WebSocket error"))))]
-                               (swap! state assoc :ws ws))
+    (-> (.newWebSocketBuilder ^HttpClient client)
-                             (catch Exception e
+        (.connectTimeout (Duration/ofSeconds 10))
-                               (log/error e "Binance reconnect failed")))))))
+        (.buildAsync (URI. uri) listener)
-
+        (.get 15 TimeUnit/SECONDS))))
-                   (onError [_ _ws error]
+
-                     (log/error error "Binance WebSocket error")))]
+(defmethod ig/init-key :ws/binance
-    (-> (.newWebSocketBuilder client)
+  [_ {:keys [query-fn uri enabled?] :or {enabled? "true"}}]
-        (.buildAsync (URI. uri) listener)
+  (let [enabled (= enabled? "true")
-        (.join))))
+        latest-price (atom nil)
-
+        state (atom {:enabled? enabled :running? enabled :connected? false
-(defmethod ig/init-key :ws/binance
+                     :last-write 0 :last-message 0 :generation nil :ws nil})
-  [_ {:keys [query-fn uri enabled?] :or {enabled? "true"}}]
+        client (when enabled (-> (HttpClient/newBuilder)
-  (if-not (= enabled? "true")
+                                 (.connectTimeout (Duration/ofSeconds 10)) .build))
-    (do (log/info "Binance WebSocket disabled")
+        worker
-        {:state        (atom {:running? false})
+        (when enabled
-         :latest-price (atom nil)})
+          (future
-    (do (log/info "Starting Binance WebSocket listener:" uri)
+            (try
-        (let [latest-price (atom nil)
+              (while (:running? @state)
-              state        (atom {:running?   true
+                (let [generation (Object.)]
-                                  :last-write 0
+                  (swap! state assoc :generation generation :connected? false
-                                  :buffer     (StringBuilder.)
+                         :last-message (System/currentTimeMillis))
-                                  :ws         nil})
+                  (try
-              ws           (connect! uri query-fn state latest-price)]
+                    (let [ws (connect! client uri query-fn state latest-price generation)]
-          (swap! state assoc :ws ws)
+                      (swap! state assoc :ws ws)
-          {:state        state
+                      (try
-           :latest-price latest-price}))))
+                        (while (and (:running? @state) (:connected? @state)
-
+                                    (< (- (System/currentTimeMillis) (:last-message @state)) stale-after-ms))
-(defmethod ig/halt-key! :ws/binance
+                          (Thread/sleep 1000))
-  [_ {:keys [state]}]
+                        (finally (.abort ^WebSocket ws))))
-  (log/info "Stopping Binance WebSocket listener")
+                    (catch InterruptedException e (throw e))
-  (swap! state assoc :running? false)
+                    (catch Exception e (log/warn e "Binance connection failed; retrying"))
-  (when-let [ws (:ws @state)]
+                    (finally (swap! state assoc :connected? false :generation nil :ws nil))))
-    (try
+                (when (:running? @state) (Thread/sleep 3000)))
-      (.sendClose ws WebSocket/NORMAL_CLOSURE "shutting down")
+              (catch InterruptedException _))))]
-      (catch Exception _))))
+    {: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)))

test/clj/pmagnus/btcdata/kraken/day_test.clj

@@ -0,0 +1,77 @@
+(ns pmagnus.btcdata.kraken.day-test
+  (:require [clojure.test :refer [deftest is]]
+            [pmagnus.btcdata.kraken.day :as day])
+  (:import [java.time LocalDate ZoneOffset]))
+
+(def date (LocalDate/parse "2020-01-01"))
+(def ts (.toEpochSecond (.atStartOfDay date) ZoneOffset/UTC))
+
+(defn database [store writes]
+  (fn [q p]
+    (case q
+      :get-kraken-day-by-ts (get @store [:recent (:ts p)])
+      :get-kraken-day-history-by-ts (get @store [:history (:ts p)])
+      (:upsert-kraken-day! :upsert-kraken-day-history!)
+      (let [table (if (= q :upsert-kraken-day!) :recent :history)]
+        (swap! writes conj q)
+        (swap! store assoc [table (:ts p)]
+               (-> p (assoc :trade_count (:trade-count p)) (dissoc :trade-count)))))))
+
+(deftest cached-data-never-fetches
+  (doseq [table [:recent :history]]
+    (let [row {:ts ts} store (atom {[table ts] row})]
+      (with-redefs [day/request! (fn [_] (throw (Exception. "Unexpected fetch")))]
+        (is (= row (day/find-day! (database store (atom [])) date)))))))
+
+(deftest recent-candle-is-persisted
+  (let [store (atom {}) writes (atom []) q (database store writes)]
+    (with-redefs [day/request! (fn [_] {:XXBTZUSD [[ts "10" "12" "9" "11" "10.5" "2" 3]]})]
+      (is (= 11M (:close (day/find-day! q date))))
+      (is (= [:upsert-kraken-day!] @writes)))))
+
+(deftest historical-pagination-and-cache
+  (let [store (atom {}) writes (atom []) q (database store writes)
+        calls (atom []) cursor (str (*' (+ ts 5) 1000000000))]
+    (with-redefs [day/pause! (fn [])
+                  day/request!
+                  (fn [url]
+                    (swap! calls conj url)
+                    (case (count @calls)
+                      1 {:XXBTZUSD [[(+ ts 86400) "1" "1" "1" "1" "1" "1" 1]]}
+                      2 {:XXBTZUSD [["10" "2" ts] ["20" "1" (+ ts 5)]] :last cursor}
+                      3 {:XXBTZUSD [["5" "1" (+ ts 86399.9M)]
+                                   ["999" "1" (+ ts 86400)]] :last "unused"}))]
+      (let [result (day/find-day! q date)]
+        (is (= {:open 10M :high 20M :low 5M :close 5M :volume 4M
+                :vwap 11.25000000M :trade_count 3 :ts ts} result))
+        (is (= result (day/find-day! q date)))
+        (is (= 3 (count @calls)))
+        (is (.endsWith (last @calls) cursor))
+        (is (= [:upsert-kraken-day-history!] @writes))))))
+
+(deftest failed-or-empty-history-is-not-persisted
+  (doseq [result [{:XXBTZUSD [] :last "0"}
+                  {:XXBTZUSD [["1" "1" ts]] :last "0"}
+                  {:wrong []}]]
+    (let [writes (atom []) calls (atom 0)]
+      (with-redefs [day/request! (fn [_]
+                                  (if (= 1 (swap! calls inc))
+                                    {:XXBTZUSD [[(+ ts 86400)]]} result))]
+        (if (= [] (:XXBTZUSD result))
+          (is (nil? (day/find-day! (database (atom {}) writes) date)))
+          (is (thrown? clojure.lang.ExceptionInfo
+                       (day/find-day! (database (atom {}) writes) date))))
+        (is (empty? @writes))))))
+
+(deftest upstream-failure-does-not-save-a-partial-day
+  (let [writes (atom []) calls (atom 0)]
+    (with-redefs [day/pause! (fn [])
+                  day/request! (fn [_]
+                                 (case (swap! calls inc)
+                                   1 {:XXBTZUSD [[(+ ts 86400)]]}
+                                   2 {:XXBTZUSD [["10" "1" ts]]
+                                      :last (str (*' (inc ts) 1000000000))}
+                                   (throw (Exception. "Connection failed"))))]
+      (is (thrown? clojure.lang.ExceptionInfo
+                   (day/find-day! (database (atom {}) writes) date)))
+      (is (empty? @writes)))))

test/clj/pmagnus/btcdata/web/controllers/day_test.clj

@@ -0,0 +1,54 @@
+(ns pmagnus.btcdata.web.controllers.day-test
+  (:require [clojure.test :refer [deftest is]]
+            [clojure.data.json :as json]
+            [integrant.core :as ig]
+            [pmagnus.btcdata.kraken.day :as day]
+            [pmagnus.btcdata.web.controllers.day :as controller]
+            [pmagnus.btcdata.web.routes.api :as api]
+            [pmagnus.btcdata.web.handler]
+            [reitit.ring :as ring]))
+
+(def app
+  (ring/ring-handler
+   (ring/router [(ig/init-key :reitit.routes/api {:base-path "/api" :query-fn nil})])))
+
+(defn request [uri query]
+  (app {:request-method :get :uri uri :query-string query
+        :headers {"accept" "application/json"}}))
+
+(deftest routes-and-numeric-json
+  (with-redefs [day/find-day! (fn [_ _]
+                              {:open 10M :high 12M :low 9M :close 11M
+                               :vwap 10.5M :volume 2M :trade_count 3})]
+    (doseq [[uri query] [["/api/price/day/2020-01-01" nil]
+                        ["/api/price/day" "date=2020-01-01"]]]
+      (let [r (request uri query) body (json/read-str (slurp (:body r)) :key-fn keyword)]
+        (is (= 200 (:status r)))
+        (is (= "BTC/USD" (:pair body)))
+        (is (= "2020-01-01" (:date body)))
+        (is (number? (:open body)))))))
+
+(deftest invalid-dates
+  (doseq [s [nil "bad" "2020-02-30" "2020-1-01" "9999-01-01"
+             (str (java.time.LocalDate/now java.time.ZoneOffset/UTC))]]
+    (with-redefs [day/find-day! (fn [& _] (throw (Exception. "Must not fetch")))]
+      (is (= 400 (:status (controller/handle {} {:query-params {"date" s}})))))))
+
+(deftest error-responses
+  (with-redefs [day/find-day! (constantly nil)]
+    (is (= 404 (:status (request "/api/price/day/2020-01-01" nil)))))
+  (with-redefs [day/find-day! (fn [& _] (throw (ex-info "upstream" {:type ::day/upstream})))]
+    (is (= 502 (:status (request "/api/price/day/2020-01-01" nil))))))
+
+(deftest swagger-documents-both-routes
+  (let [r (request "/api/swagger.json" nil)
+        body (json/read-str (slurp (:body r)))]
+    (is (= 200 (:status r)))
+    (is (get-in body ["paths" "/api/price/day" "get"]))
+    (is (get-in body ["paths" "/api/price/day/{date}" "get"]))))
+
+(deftest swagger-ui-is-served
+  (let [handler (ig/init-key :handler/ring
+                            {:router #(ring/router []) :api-path "/api"})]
+    (is (= 302 (:status (handler {:request-method :get :uri "/swagger" :headers {}}))))
+    (is (= 200 (:status (handler {:request-method :get :uri "/swagger/" :headers {}}))))))

test/clj/pmagnus/btcdata/web/controllers/health_test.clj

@@ -1,26 +1,34 @@
 (ns pmagnus.btcdata.web.controllers.health-test
-  (:require
+  (:require [clojure.test :refer [deftest is]]
-   [clojure.test :refer [deftest is testing]]
+            [pmagnus.btcdata.web.controllers.health :as health]))
-   [pmagnus.btcdata.web.controllers.health :as health]))
 
-(deftest healthcheck-is-process-only
+(def disabled-stream {:state (atom {:enabled? false}) :latest-price (atom nil)})
-  (let [response (health/healthcheck! {})]
+
-    (is (= 200 (:status response)))
+(deftest liveness-check-is-local-and-detects-lost-writer
-    (is (= "up" (get-in response [:body :app :status])))))
+  (let [closed? (atom false)
+        connection (proxy [java.sql.Connection] []
+                     (isClosed [] @closed?))
+        query-fn (with-meta (fn [& _] (throw (Exception. "Must not query database")))
+                   {:connection connection})]
+    (is (= 200 (:status (health/healthcheck! query-fn {}))))
+    (reset! closed? true)
+    (is (= 503 (:status (health/healthcheck! query-fn {}))))
+    (is (= "writer session closed"
+           (get-in (health/healthcheck! query-fn {}) [:body :reason]))))
+  (is (= 503 (:status (health/healthcheck! (fn [& _]) {})))))
 
 (deftest readiness-check-tests-database
-  (testing "ready when the database query succeeds"
+  (let [called (atom nil)
-    (let [called   (atom nil)
+        query-fn (fn [query params] (reset! called [query params]))]
-          query-fn (fn [query params]
+    (is (= 200 (:status (health/readiness-check! query-fn disabled-stream {}))))
-                     (reset! called [query params])
+    (is (= [:check-database {}] @called)))
-                     {:healthy 1})
+  (is (= 503 (:status (health/readiness-check!
-          response (health/readiness-check! query-fn {})]
+                       (fn [_ _] (throw (ex-info "database unavailable" {})))
-      (is (= 200 (:status response)))
+                       disabled-stream {})))))
-      (is (= {:status "ready"} (:body response)))
-      (is (= [:check-database {}] @called))))
 
-  (testing "not ready when the database query fails"
+(deftest stale-stream-is-not-ready
-    (let [query-fn (fn [_ _] (throw (ex-info "database unavailable" {})))
+  (is (= 503 (:status (health/readiness-check!
-          response (health/readiness-check! query-fn {})]
+                       (fn [_ _] {})
-      (is (= 503 (:status response)))
+                       {:state (atom {:enabled? true :connected? true})
-      (is (= {:status "not ready"} (:body response))))))
+                        :latest-price (atom {:recorded-at java.time.Instant/EPOCH})}
+                       {})))))

test/clj/pmagnus/btcdata/writer_test.clj

@@ -0,0 +1,80 @@
+(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 session-liveness-fails-closed
+  (is (false? (writer/session-live? (fn [& _]))))
+  (let [connection (proxy [java.sql.Connection] []
+                     (isClosed [] (throw (java.sql.SQLException. "broken"))))]
+    (is (false? (writer/session-live? (with-meta (fn [& _])
+                                       {:connection connection}))))))
+
+(deftest server-session-has-bounded-dead-peer-detection
+  (let [connection (Object.)
+        calls (atom [])]
+    (with-redefs [jdbc/execute-one! (fn [conn sql] (swap! calls conj [conn sql]))]
+      (writer/configure-session! connection))
+    (is (= (mapv #(vector connection [%])
+                 ["SET tcp_keepalives_idle = 60"
+                  "SET tcp_keepalives_interval = 10"
+                  "SET tcp_keepalives_count = 3"
+                  "SET tcp_user_timeout = 90000"])
+           @calls))))
+
+(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 {}))))
+        (let [connection (:connection (meta first-writer))]
+          (doseq [[setting expected] [["tcp_keepalives_idle" "60"]
+                                     ["tcp_keepalives_interval" "10"]
+                                     ["tcp_keepalives_count" "3"]
+                                     ["tcp_user_timeout" "90000"]]]
+            (is (= expected (:value (jdbc/execute-one! connection
+                                     ["select current_setting(?) as value" setting]))))))
+        (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.
+        (is (writer/session-live? first-writer))
+        (.close (:connection (meta first-writer)))
+        (is (false? (writer/session-live? 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))))))

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))))))))