|
100 | 100 | (defonce ^:private pub-counters (atom {})) ; k -> AtomicLong (next publishingId) |
101 | 101 | (defonce ^:private env-generation (atom 0)) ; bumped each Environment rebuild |
102 | 102 |
|
| 103 | +(defn- retryable? |
| 104 | + "True when the throwable (or a cause in its chain) is a StreamException -- a |
| 105 | + transient broker-side condition (stream/leader not yet available right after |
| 106 | + create or during a partition) worth retrying. A programming error (NPE, |
| 107 | + ClassCast) is not a StreamException, so it falls through and crashes fast. The |
| 108 | + walk is nil-safe and depth-bounded so a cyclic or self-referential cause chain |
| 109 | + (which async stream-client errors can produce) cannot spin forever -- under the |
| 110 | + leader-move crash storm an unbounded walk hung the whole run." |
| 111 | + [^Throwable e] |
| 112 | + (->> (iterate (fn [^Throwable t] (when t (.getCause t))) e) |
| 113 | + (take 64) |
| 114 | + (take-while some?) |
| 115 | + (some #(instance? com.rabbitmq.stream.StreamException %)) |
| 116 | + boolean)) |
| 117 | + |
103 | 118 | (defn with-retry |
104 | | - "Calls thunk, retrying on any exception until op-retry-ms elapses, then |
105 | | - rethrows. For transient 'leader/stream not available' errors right after |
106 | | - create or during a partition." |
| 119 | + "Calls thunk, retrying transient stream errors (see retryable?) until |
| 120 | + op-retry-ms elapses, then rethrows. A non-transient exception is rethrown |
| 121 | + immediately, so a client bug fails fast instead of after a 10s retry loop." |
107 | 122 | [thunk] |
108 | 123 | (let [deadline (+ (System/currentTimeMillis) op-retry-ms)] |
109 | 124 | (loop [] |
110 | 125 | (let [r (try {:val (thunk)} |
111 | 126 | (catch Exception e |
112 | | - (if (< (System/currentTimeMillis) deadline) :retry (throw e))))] |
| 127 | + (if (and (retryable? e) (< (System/currentTimeMillis) deadline)) |
| 128 | + :retry |
| 129 | + (throw e))))] |
113 | 130 | (if (= r :retry) (do (Thread/sleep 250) (recur)) (:val r)))))) |
114 | 131 |
|
115 | 132 | (defn get-env ^Environment [test] |
|
139 | 156 | (reset! pub-counters {}) |
140 | 157 | (swap! env-generation inc)))) |
141 | 158 |
|
| 159 | +(defn reset-run-state! |
| 160 | + "Resets all run-scoped shared connection state (the Environment, declared |
| 161 | + streams, producers, publishingId counters, generation) so a second test in the |
| 162 | + same JVM (a REPL session or `lein run test-all`) starts clean rather than |
| 163 | + carrying stale stream declarations or publishingId counters against a |
| 164 | + freshly-wiped cluster. (tiering-stats and authoritative-reads are reset |
| 165 | + alongside this in db/setup!.)" |
| 166 | + [] |
| 167 | + (locking shared-env |
| 168 | + (when-let [e @shared-env] (try (.close e) (catch Exception _))) |
| 169 | + (reset! shared-env nil) |
| 170 | + (reset! declared-streams #{}) |
| 171 | + (reset! shared-producers {}) |
| 172 | + (reset! pub-counters {}) |
| 173 | + (reset! env-generation 0))) |
| 174 | + |
142 | 175 | (defn ensure-stream! [env k] |
143 | 176 | (when-not (contains? @declared-streams k) |
144 | 177 | ;; create() is idempotent for a matching stream. |
|
253 | 286 | ;; Authoritative end-to-end read (for the durability checker) |
254 | 287 | ;; --------------------------------------------------------------------------- |
255 | 288 |
|
256 | | -;; Reading the whole history of ~90 streams, some served from S3, needs a far |
257 | | -;; more generous quiescence wait than the steady-state catch-up above. |
258 | | -(def auth-quiet-ms 3000) ; deliveries idle this long => every stream at its tail |
259 | | -(def auth-grace-ms 8000) ; tolerate this long before any delivery (S3 warm-up) |
260 | | -(def auth-cap-ms 180000) ; hard cap so a genuinely stuck stream cannot hang |
| 289 | +;; Default cap on the authoritative read (overridable via --auth-read-timeout-sec). |
| 290 | +;; The read waits for an observable target (each consumer reaching the stream's |
| 291 | +;; committed offset), not idle time, so it cannot truncate mid-backfill; the cap |
| 292 | +;; only bounds a stream that genuinely never reaches its target. |
| 293 | +(def auth-read-cap-ms-default 180000) |
261 | 294 |
|
262 | 295 | ;; The authoritative read of every jepsen stream, captured once after the run |
263 | 296 | ;; while the brokers are still up (db/log-files), for the durability checker: |
|
266 | 299 | ;; offsets unreliable. |
267 | 300 | (defonce authoritative-reads (atom {})) |
268 | 301 |
|
269 | | -(defn- await-quiescent! |
270 | | - "Waits until the total buffered count across queues holds steady for |
271 | | - auth-quiet-ms (every stream reached its tail), bounded by auth-cap-ms, with an |
272 | | - auth-grace-ms grace before the first delivery for streams served from S3." |
273 | | - [queues] |
274 | | - (let [total (fn [] (reduce (fn [a ^java.util.Collection q] (+ a (.size q))) 0 queues)) |
275 | | - start (System/currentTimeMillis) |
276 | | - deadline (+ start auth-cap-ms)] |
277 | | - (loop [last-total (total), last-change start] |
278 | | - (let [now (System/currentTimeMillis) |
279 | | - t (total)] |
280 | | - (cond |
281 | | - (>= now deadline) nil |
282 | | - (not= t last-total) (do (Thread/sleep 100) (recur t now)) |
283 | | - (and (> (- now last-change) auth-quiet-ms) |
284 | | - (or (pos? t) (> (- now start) auth-grace-ms))) nil |
285 | | - :else (do (Thread/sleep 100) (recur t last-change))))))) |
| 302 | +(defn- await-targets! |
| 303 | + "Waits until every stream's max delivered offset reaches its committed-offset |
| 304 | + target, bounded by cap-ms. `state` is {key {:q queue :maxoff AtomicLong}}; |
| 305 | + `targets` is {key committed-offset}. A stream that never reaches its target |
| 306 | + waits out the cap, after which the durability checker reports the loss." |
| 307 | + [state targets cap-ms] |
| 308 | + (let [deadline (+ (System/currentTimeMillis) cap-ms) |
| 309 | + met? (fn [] (every? (fn [[k {:keys [^AtomicLong maxoff]}]] |
| 310 | + (>= (.get maxoff) (get targets k -1))) |
| 311 | + state))] |
| 312 | + (loop [] |
| 313 | + (when (and (not (met?)) (< (System/currentTimeMillis) deadline)) |
| 314 | + (Thread/sleep 100) |
| 315 | + (recur))))) |
286 | 316 |
|
287 | 317 | (defn read-streams-fully |
288 | | - "Authoritative end-to-end read of stream keys `ks` for the durability checker. |
289 | | - Builds a FRESH Environment (the run's shared one may be wedged from topology |
290 | | - churn), subscribes each stream from the beginning, waits for deliveries to go |
291 | | - quiet (every stream at its tail), and returns {key -> vector of [offset value] |
292 | | - in delivery order}. Closes the Environment before returning." |
293 | | - [test ks] |
294 | | - (let [uris (java.util.ArrayList. |
295 | | - (map #(str "rabbitmq-stream://guest:guest@" (name %) ":" stream-port) |
296 | | - (:nodes test))) |
297 | | - env (-> (Environment/builder) (.uris uris) (.build)) |
298 | | - queues (reduce (fn [m k] (assoc m k (ConcurrentLinkedQueue.))) {} ks)] |
| 318 | + "Authoritative end-to-end read of the streams in `targets` (key -> committed |
| 319 | + offset) for the durability checker. Builds a FRESH Environment (the run's |
| 320 | + shared one may be wedged from topology churn), subscribes each stream from the |
| 321 | + beginning, and waits until each consumer has delivered up to the stream's |
| 322 | + committed offset -- an observable target rather than an idle-time guess, so a |
| 323 | + mid-backfill quiet period cannot truncate the read and report false loss. |
| 324 | + Returns {key -> vector of [offset value] in delivery order}. Closes the |
| 325 | + Environment before returning." |
| 326 | + [test targets] |
| 327 | + (let [cap-ms (* 1000 (or (:auth-read-timeout-sec test) |
| 328 | + (quot auth-read-cap-ms-default 1000))) |
| 329 | + uris (java.util.ArrayList. |
| 330 | + (map #(str "rabbitmq-stream://guest:guest@" (name %) ":" stream-port) |
| 331 | + (:nodes test))) |
| 332 | + env (-> (Environment/builder) (.uris uris) (.build)) |
| 333 | + state (into {} (map (fn [k] [k {:q (ConcurrentLinkedQueue.) :maxoff (AtomicLong. -1)}]) |
| 334 | + (keys targets)))] |
299 | 335 | (try |
300 | | - (doseq [[k q] queues] |
| 336 | + (doseq [[k {:keys [^ConcurrentLinkedQueue q ^AtomicLong maxoff]}] state] |
301 | 337 | (-> env (.consumerBuilder) (.stream (stream-name k)) |
302 | 338 | (.offset (OffsetSpecification/first)) |
303 | 339 | (.messageHandler |
304 | 340 | (reify com.rabbitmq.stream.MessageHandler |
305 | 341 | (handle [_ ctx msg] |
306 | | - (.add q [(.offset ctx) (decode-value (.getBodyAsBinary msg))])))) |
| 342 | + ;; Deliveries arrive in offset order from `first`, so set wins. |
| 343 | + (.add q [(.offset ctx) (decode-value (.getBodyAsBinary msg))]) |
| 344 | + (.set maxoff (.offset ctx))))) |
307 | 345 | (.build))) |
308 | | - (await-quiescent! (vals queues)) |
309 | | - (into {} (map (fn [[k q]] [k (drain! q)]) queues)) |
| 346 | + (await-targets! state targets cap-ms) |
| 347 | + (into {} (map (fn [[k {:keys [^ConcurrentLinkedQueue q]}]] [k (drain! q)]) state)) |
310 | 348 | (finally (.close env))))) |
311 | 349 |
|
312 | 350 | (defn capture-authoritative-reads! |
313 | | - "Reads keys `ks` end-to-end and stores the result in authoritative-reads." |
314 | | - [test ks] |
315 | | - (reset! authoritative-reads (read-streams-fully test ks))) |
| 351 | + "Reads the streams in `targets` (key -> committed offset) end-to-end and stores |
| 352 | + the result in authoritative-reads." |
| 353 | + [test targets] |
| 354 | + (reset! authoritative-reads (read-streams-fully test targets))) |
316 | 355 |
|
317 | 356 | (defrecord Client [node ^Environment env env-gen consumers buffers assigned] |
318 | 357 | client/Client |
|
0 commit comments