Process Batch Bundle Entries With a Sliding Window

.github/scripts/large-batch.sh

@@ -0,0 +1,83 @@
+#!/bin/bash
+set -euo pipefail
+
+#
+# This script sends one large batch bundle and checks that every entry is
+# processed and that the response entries keep the order of the request
+# entries.
+#
+# The entries of a batch bundle are processed concurrently with a sliding
+# window. So the bundle interleaves writes with reads, because the reads are
+# much faster than the writes and make the entries complete out of order. The
+# response entries have to keep the bundle order nonetheless.
+#
+
+script_dir="$(dirname "$(readlink -f "$0")")"
+. "$script_dir/util.sh"
+
+base="http://localhost:8080/fhir"
+
+# the bundle holds one read after each write, so it has twice as many entries
+num_writes="${1:-2000}"
+num_entries=$((num_writes * 2))
+
+prefix="large-batch"
+
+# tag marking the patients written by this script, so that all of them can be
+# counted with a single search
+tag_system="http://acme.org/codes"
+tag_code="$prefix"
+
+bundle() {
+  jq -nc --arg prefix "$prefix" --arg system "$tag_system" --arg code "$tag_code" \
+    --argjson num "$num_entries" '
+    {
+      resourceType: "Bundle",
+      type: "batch",
+      entry: [
+        range($num) |
+        if . % 2 == 0 then
+          {
+            resource: {
+              resourceType: "Patient",
+              id: "\($prefix)-\(.)",
+              meta: {tag: [{system: $system, code: $code}]}
+            },
+            request: {method: "PUT", url: "Patient/\($prefix)-\(.)"}
+          }
+        else
+          {request: {method: "GET", url: "Patient?_summary=count"}}
+        end
+      ]
+    }'
+}
+
+start="$(date +%s)"
+result="$(bundle | transact "$base")"
+echo "ℹ️ processed $num_entries batch bundle entries in $(($(date +%s) - start)) s"
+
+test "resource type" "$(echo "$result" | jq -r '.resourceType')" "Bundle"
+test "bundle type" "$(echo "$result" | jq -r '.type')" "batch-response"
+test "number of response entries" "$(echo "$result" | jq -r '.entry | length')" "$num_entries"
+
+test "distinct statuses of the write entries" \
+  "$(echo "$result" | jq -r '[.entry | to_entries[] | select(.key % 2 == 0) | .value.response.status] | unique | join(",")')" \
+  "201"
+
+test "distinct statuses of the read entries" \
+  "$(echo "$result" | jq -r '[.entry | to_entries[] | select(.key % 2 == 1) | .value.response.status] | unique | join(",")')" \
+  "200"
+
+# every write entry has to answer with the location of the patient that the
+# request entry at that very position created
+test "number of write entries at the wrong position" \
+  "$(echo "$result" | jq -r --arg prefix "$prefix" '
+      [.entry | to_entries[]
+       | select(.key % 2 == 0)
+       | select((.value.response.location // "" | split("/") | .[5]) != "\($prefix)-\(.key)")]
+      | length')" \
+  "0"
+
+test "number of created patients" \
+  "$(search_strict "$base/Patient?_tag=$tag_system|$tag_code&_summary=count" | jq -r '.total')" \
+  "$num_writes"

.github/workflows/build.yml

@@ -1568,6 +1568,9 @@ jobs:
     - name: Batch Metadata
       run: .github/scripts/batch-metadata.sh
 
+    - name: Large Batch Bundle
+      run: .github/scripts/large-batch.sh
+
     - name: Transaction
       run: .github/scripts/transaction.sh
 

docs/api/interaction/batch.md

@@ -1,6 +1,6 @@
 # Batch
 
-The batch interaction allows submitting a set of actions to be performed as a single HTTP request. The semantics of the individual actions described in the `batch` Bundle are identical of the semantics of the corresponding individual request. The actions are performed in order but can be interleaved by other individual requests or actions from other batch interactions.
+The batch interaction allows submitting a set of actions to be performed as a single HTTP request. The semantics of the individual actions described in the `batch` Bundle are identical of the semantics of the corresponding individual request. The actions are independent of each other, are performed in no particular order and can be interleaved by other individual requests or actions from other batch interactions.
 
 ```
 POST [base]
@@ -20,6 +20,20 @@ The request body has to be a Bundle of type `batch`. The following methods are s
 
 The methods `HEAD` and `PATCH` are not supported and will result in an error entry with status `422`, unknown methods in an error entry with status `400`.
 
+## Processing Rules <Badge type="warning" text="Since 1.11.0"/>
+
+FHIR requires that there are [no interdependencies][1] between the entries of a batch bundle: the success or failure of one action must not alter the success or failure or the resulting content of another one. Blaze relies on that rule and processes the entries concurrently. That concurrency is about submitting the actions; the transactions they create are still applied one at a time as described in the note above.
+
+* **No order.** The actions aren't performed in the order of the bundle, and no order between any two of them is guaranteed. Two actions writing the same resource are applied in an undefined order.
+* **Reads aren't separated from writes.** In a [transaction](transaction.md) bundle, the `GET` actions are executed after the writes, against the new database state. In a batch, a `GET` is just another independent action, so whether it sees a write of the same bundle is undefined. Use a transaction bundle if an action has to read what the same bundle writes.
+* **Bounded concurrency.** At most 64 actions are processed at a time, and never more than half of the maximum number of in-flight transactions set by the [environment variable](../../deployment/environment-variables.md) `DB_MAX_IN_FLIGHT_TRANSACTIONS`. So a single batch can't take more than half of the places for in-flight transactions away from other clients.
+* **Backpressure.** A write action rejected because that maximum is reached is retried a few times. If it still doesn't get one of the places, it keeps its `503` in the response bundle, where it then really means that the server is saturated.
+
+None of this affects the response bundle, whose entries always keep the order of the request entries.
+
+> [!NOTE]
+> Up to version 1.10.1, the actions of a batch were processed one after another in the order of the bundle. Bundles relying on that order were already outside what FHIR guarantees, but they can behave differently now.
+
 ## Response
 
 Blaze returns a `200 OK` with a Bundle of type `batch-response`. The response bundle contains one entry for each entry of the request bundle, in the same order. Each entry contains the status and further details like `ETag` value, last modified time and location under `Bundle.entry.response`. For successful reads and searches, the resource or search result Bundle is returned under `Bundle.entry.resource`.
@@ -52,3 +66,5 @@ In contrast to the [transaction](transaction.md) interaction, actions in a batch
   ]
 }
 ```
+
+[1]: <https://hl7.org/fhir/R4/http.html#brules>

docs/deployment/environment-variables.md

@@ -473,6 +473,8 @@ Timeout in milliseconds for all reading FHIR interactions acquiring the newest d
 
 The maximum number of transactions that were submitted but are not yet indexed. Submitting a transaction while that maximum is reached will return a 503 Service Unavailable response instead, asking the client to try again later. Because such a transaction is rejected before any of its data is written, lowering this value bounds the memory the not yet indexed transactions occupy. Please increase it if you experience such 503 responses under a write load your system can otherwise sustain.
 
+The value also bounds how many entries of a batch bundle Blaze processes concurrently. That number is at most 64 and never more than half of this maximum, so that a single batch bundle can't take more than half of the places for in-flight transactions away from other clients.
+
 **Default:** 1024
 
 #### `DB_SEARCH_PARAM_BUNDLE` <Badge type="warning" text="Since 0.21"/>

modules/async/deps.edn

@@ -15,7 +15,10 @@
   {:extra-paths ["test"]
 
    :extra-deps
-   {blaze/test-util
+   {blaze/module-test-util
+    {:local/root "../module-test-util"}
+
+    blaze/test-util
     {:local/root "../test-util"}}}
 
   :kaocha

modules/async/src/blaze/async/comp.clj

@@ -6,7 +6,6 @@
   (:require
    [blaze.anomaly :as ba]
    [clojure.math :as math]
-   [cognitect.anomalies :as anom]
    [taoensso.timbre :as log])
   (:import
    [java.util.concurrent CompletionStage CompletableFuture TimeUnit CompletionException]))
@@ -299,10 +298,7 @@
     (fn [~binding-form]
       ~@body)))
 
-(defn- retryable? [{::anom/keys [category]}]
-  (#{::anom/busy} category))
-
-(defn- retry* [future-fn action-name max-retries num-retry]
+(defn- retry* [future-fn retryable? action-name max-retries num-retry]
   (-> (future-fn)
       (exceptionally-compose
        (fn [e]
@@ -312,24 +308,37 @@
              (-> (future)
                  (complete-on-timeout!
                   nil delay TimeUnit/MILLISECONDS)
-                 (then-compose
+                 (then-compose-async
                   (fn [_]
-                    (retry* future-fn action-name max-retries (inc num-retry))))))
+                    (retry* future-fn retryable? action-name max-retries
+                            (inc num-retry))))))
            e)))))
 
 (defn retry
   "Returns a CompletionStage that, when the CompletionStage as result of calling
   the function `f` with no arguments completes normally will complete with its
   result.
 
-  Otherwise retries by calling `f` again with no arguments if the anomaly is of
-  category `::anom/busy`. Waits between retries starting with 100 ms growing
-  exponentially.
+  Otherwise retries by calling `f` again with no arguments if calling the
+  function `retryable?` with the anomaly returns true. Retries every busy
+  anomaly if `retryable?` isn't given. Waits between retries starting with
+  100 ms growing exponentially.
+
+  Only anomalies for which retrying is safe should be matched by `retryable?`.
+  The category alone often doesn't tell that. A busy anomaly for example is
+  returned by a request that was rejected before it had any effect as well as by
+  one that timed out after it maybe had one already.
+
+  The retry is done using the default asynchronous execution facility and not on
+  the thread that completes the wait. That thread schedules all timeouts of the
+  JVM, so `f` doing any work on it would delay every other timeout.
 
   Please be aware that `num-retries` shouldn't be higher than the max stack
   depth. Otherwise, the CompletionStage would fail with a StackOverflowException."
-  [f action-name num-retries]
-  (retry* f action-name num-retries 0))
+  ([f action-name num-retries]
+   (retry f action-name num-retries ba/busy?))
+  ([f action-name num-retries retryable?]
+   (retry* f retryable? action-name num-retries 0)))
 
 (defn retry2
   "Returns a CompletionStage that, when the CompletionStage as result of calling

modules/async/src/blaze/async/comp_spec.clj

@@ -120,7 +120,8 @@
   :ret ac/completable-future?)
 
 (s/fdef ac/retry
-  :args (s/cat :f ifn? :action-name string? :num-retries pos-int?)
+  :args (s/cat :f ifn? :action-name string? :num-retries pos-int?
+               :retryable? (s/? ifn?))
   :ret ac/completable-future?)
 
 (s/fdef ac/retry2

modules/async/test/blaze/async/comp_test.clj

@@ -4,10 +4,13 @@
    [blaze.async.comp :as ac :refer [do-sync do-async]]
    [blaze.async.comp-spec]
    [blaze.executors :as ex]
+   [blaze.module.test-util :as mtu]
    [blaze.test-util :as tu :refer [given-failed-future]]
    [clojure.spec.test.alpha :as st]
+   [clojure.string :as str]
    [clojure.test :as test :refer [deftest is testing]]
-   [cognitect.anomalies :as anom])
+   [cognitect.anomalies :as anom]
+   [juxt.iota :refer [given]])
   (:import
    [java.util.concurrent TimeUnit]))
 
@@ -16,6 +19,12 @@
 
 (test/use-fixtures :each tu/fixture)
 
+(defn- delay-scheduler-thread?
+  "Returns true if `thread-name` is the name of the thread that schedules all
+  timeouts of the JVM."
+  [thread-name]
+  (str/includes? thread-name "delayScheduler"))
+
 (deftest completed-future-test
   (testing "on completed future"
     (is (= ::x @(ac/completed-future ::x))))
@@ -384,6 +393,34 @@
         (given-failed-future (ac/retry future-fn "action-114844" 1)
           ::anom/category := ::anom/busy))))
 
+  (testing "with a retryable predicate"
+    (testing "an anomaly the predicate matches is retried"
+      (let [counter (atom 0)
+            future-fn #(ac/completed-future
+                        (let [n (swap! counter inc)]
+                          (if (= 2 n) ::x (ba/conflict))))]
+        (is (= ::x @(ac/retry future-fn "action-114844" 1 ba/conflict?)))))
+
+    (testing "a busy anomaly the predicate doesn't match isn't retried"
+      (let [counter (atom 0)
+            future-fn #(ac/completed-future
+                        (let [n (swap! counter inc)]
+                          (if (= 2 n) ::x (ba/busy))))]
+        (given-failed-future (ac/retry future-fn "action-114844" 1 ba/conflict?)
+          ::anom/category := ::anom/busy)
+
+        (is (= 1 @counter)))))
+
+  (testing "the retry doesn't happen on the thread that schedules the delay
+            because that single thread schedules all timeouts of the JVM"
+    (let [counter (atom 0)
+          future-fn #(ac/completed-future
+                      (let [n (swap! counter inc)]
+                        (if (= 2 n) {:result ::x} (ba/busy))))]
+      (given @(mtu/assoc-thread-name (ac/retry future-fn "action-114844" 1))
+        [meta :thread-name] :!? delay-scheduler-thread?
+        :result := ::x)))
+
   (testing "times"
     (testing "with second call successful"
       (let [counter (atom 0)

modules/db/src/blaze/db/api.clj

@@ -56,9 +56,13 @@
   transaction in case of success or will complete exceptionally with an anomaly
   in case of a transaction error or other errors. Fails with an unavailable
   anomaly if `node` is closed, because its indexing loop wouldn't pick the
-  transaction up anymore. Fails with a busy anomaly if `node` already has the
-  maximum number of transactions submitted but not yet indexed, before anything
-  of the transaction is written.
+  transaction up anymore. Fails with a busy anomaly of
+  `:blaze.db.anom/category` `:submit-rejected` if `node` already has the maximum
+  number of transactions submitted but not yet indexed. Such a rejection happens
+  before anything of the transaction is written, so it's the only failure of a
+  transaction that can be safely retried by submitting it again. Every other
+  busy anomaly, especially the one of a timeout, can also mean that the
+  transaction was applied after all.
 
   Functions applied after the returned future are executed on the common
   ForkJoinPool."

modules/db/src/blaze/db/node.clj

@@ -72,6 +72,7 @@
    [blaze.anomaly :as ba :refer [if-ok]]
    [blaze.async.comp :as ac :refer [do-sync]]
    [blaze.coll.core :as coll]
+   [blaze.db.anom :as-alias db-anom]
    [blaze.db.api :as d]
    [blaze.db.impl.batch-db :as batch-db]
    [blaze.db.impl.codec :as codec]
@@ -304,17 +305,20 @@
   "Returns a function that tries to take one of the places for in-flight
   transactions.
 
-  That function returns a busy anomaly and counts the rejection if the maximum
-  number of in-flight transactions is already reached. It's the only point at
-  which a transaction is turned away, so it has to be called before any data of
-  the transaction is stored."
+  That function returns a rejection anomaly and counts the rejection if the
+  maximum number of in-flight transactions is already reached. It's the only
+  point at which a transaction is turned away, so it has to be called before any
+  data of the transaction is stored. That's also why the anomaly is categorized
+  as a rejection, which tells a caller that submitting the transaction again is
+  safe."
   [node-name state max-in-flight-transactions]
   (fn []
     (let [[old new] (swap-vals! state acquire-in-flight
                                 max-in-flight-transactions)]
       (when (identical? old new)
         (prom/inc! submit-rejections-total node-name)
-        (ba/busy (max-in-flight-msg max-in-flight-transactions))))))
+        (ba/busy (max-in-flight-msg max-in-flight-transactions)
+                 ::db-anom/category :submit-rejected)))))
 
 (defn- settle-in-flight
   "Stops counting the transaction as submitting and starts counting it as

modules/db/src/blaze/db/spec.clj

@@ -1,5 +1,6 @@
 (ns blaze.db.spec
   (:require
+   [blaze.db.anom :as-alias db-anom]
    [blaze.db.impl.index.resource-handle :as rh]
    [blaze.db.impl.protocols :as p]
    [blaze.db.node.protocols :as np]
@@ -117,3 +118,6 @@
 
 (s/def :blaze.db/max-in-flight-transactions
   pos-int?)
+
+(s/def ::db-anom/category
+  #{:submit-rejected})

modules/db/test/blaze/db/node_test.clj

@@ -5,6 +5,7 @@
    [blaze.async.comp-spec]
    [blaze.async.flow :as flow]
    [blaze.async.flow-spec]
+   [blaze.db.anom :as-alias db-anom]
    [blaze.db.api :as d]
    [blaze.db.api-spec]
    [blaze.db.impl.db-spec]
@@ -580,7 +581,8 @@
                     third submit is still rejected"
             (given-failed-future (submit-patient node "2")
               ::anom/category := ::anom/busy
-              ::anom/message := "The maximum number of 2 in-flight transactions is reached. Please try again later."))
+              ::anom/message := "The maximum number of 2 in-flight transactions is reached. Please try again later."
+              ::db-anom/category := :submit-rejected))
 
           (testing "no resource content is stored for the rejected transaction"
             (is (= puts @put-count))))

modules/interaction/src/blaze/interaction/transaction.clj

@@ -296,7 +296,7 @@
 (defmethod m/pre-init-spec :blaze.interaction/transaction [_]
   (s/keys :req-un [:blaze.db/node ::rest-api/batch-handler :blaze/clock :blaze/rng-fn
                    ::rest-api/db-sync-timeout]
-          :opt-un [:blaze/validator]))
+          :opt-un [:blaze/validator :blaze.db/max-in-flight-transactions]))
 
 (defmethod ig/init-key :blaze.interaction/transaction [_ context]
   (log/info "Init FHIR transaction interaction handler")

modules/interaction/test/blaze/interaction/transaction_test.clj

@@ -258,6 +258,13 @@
       [:cause-data ::s/problems 0 :via] := [:blaze.rest-api/db-sync-timeout]
       [:cause-data ::s/problems 0 :val] := ::invalid))
 
+  (testing "invalid max-in-flight-transactions"
+    (given-failed-system (assoc-in config [:blaze.interaction/transaction :max-in-flight-transactions] ::invalid)
+      :key := :blaze.interaction/transaction
+      :reason := ::ig/build-failed-spec
+      [:cause-data ::s/problems 0 :via] := [:blaze.db/max-in-flight-transactions]
+      [:cause-data ::s/problems 0 :val] := ::invalid))
+
   (testing "invalid validator"
     (given-failed-system (assoc-in config [:blaze.interaction/transaction :validator] ::invalid)
       :key := :blaze.interaction/transaction