Pipeline Resource Indexing Across Transactions

[alexanderkiel]

The database node indexes one transaction at a time. index-tx-data! dispatches the resource indexing of a transaction to the resource indexer executor, indexes the transaction itself, stores the transaction entries and then blocks in wait-for-resources until the resources of that very transaction are indexed, before it commits and moves on to the next transaction. Nothing overlaps across transactions, although poll! already delivers up to 50 transactions at once.

Measurement

Measured on LEA79 at 5300 tx/s with the transaction load test, which submits two resources per transaction. Of the 189 µs per transaction:

	per tx
index-resources (wall, brackets the next two)	~181 µs
├ index-transactions	~28 µs	overlapped
├ store-tx-entries	~14 µs	overlapped
└ blocked in ac/join	~139 µs	74 % of the loop
store-tx-success-entries	~7 µs
poll-tx-log	~0.9 µs

The index-resources histogram brackets index-transactions and store-tx-entries, so the values are not additive. They are derived from the shares of the measured sum and should be re-read from rate(_sum)/rate(_count) before they are used as a baseline.

Because a transaction of the load test contains only two resources, the resource indexing fans out only two tasks, so at most two of the four threads of the resource indexer executor are ever active. Raising DB_RESOURCE_INDEXER_THREADS cannot recover the idle time — the serialization itself has to go.

Goal: peak throughput under concurrency, at the upper end of the load test sweep. Single transaction latency at concurrency 1 is unaffected either way.

Second problem: unbounded fan-out per transaction

Independent of the serialization, the resource indexing of a single transaction is unbounded in two ways. index-resources fans out one executor task per resource, and on a node without local payload it first fetches every resource of the transaction in one rs/multi-get. The Cassandra resource store doesn't batch that call, it issues one query per key:

https://github.com/samply/blaze/blob/main/modules/db-resource-store-cassandra/src/blaze/db/resource_store/cassandra.clj#L80-L81

So a transaction with 100.000 resources issues 100.000 concurrent Cassandra queries and holds every decoded resource in memory before a single index entry is written. Both the query concurrency and the resource memory scale with the transaction size and nothing bounds them.

This issue therefore does two things: it pipelines the resource indexing across transactions, and it makes the resource indexing of a transaction proceed in bounded chunks. The second is what turns the look-ahead budget into an actual bound instead of a suggestion.

Design

The indexing loop keeps at most a bounded number of resources dispatched but not yet awaited, across transaction boundaries and within a single transaction. Resources are dispatched in chunks, and a chunk is also the unit of the resource store fetch.

Resource indexer API

The resource indexer currently indexes a whole transaction: index-resources takes tx-data, decides which resources to index, fetches them from the resource store if the transaction carries no local payload, and fans out one executor task per resource. That leaves the fan-out rule inside the indexer while the node has to bound it.

The indexer becomes a leaf service for a single resource instead:

(defn index-resource
  "Returns a CompletableFuture that completes after the `resource` with `hash` is
  indexed."
  [resource-indexer last-updated hash resource])

index-resources, index-resources* and async-index-resource go away and cmd-rs-keys moves to the node unchanged. The node builds the work list, which it can do because it already holds the resource store.

This builds on #4102, which aligned both paths on excluding keep commands. Without it the work list on the node side would have had to reproduce an asymmetry between the two paths.

(defn- work
  "Returns a tuple of the chunks of resources of `tx-data` that have to be indexed
  and a function fetching the resources of one chunk.

  Uses the local payload if this node submitted the transaction and fetches the
  resources from the resource store otherwise. Both cases cover the same set of
  resources: kept resources are excluded because they didn't change and so are
  already indexed, and commands like delete carry no hash and are excluded that
  way."
  [{:keys [resource-store]} chunk-size {:keys [tx-cmds] payload :local-payload}]
  (if payload
    [(partition-all chunk-size payload) ac/completed-future]
    [(partition-all chunk-size (cmd-rs-keys tx-cmds :complete))
     #(-> (rs/multi-get resource-store %) (ac/then-apply hash-resource-pairs))]))

(defn- index-chunk!
  "Starts indexing one chunk of resources, returning a future that completes after
  all resources of that chunk are indexed."
  [resource-indexer last-updated fetch chunk]
  (-> (fetch chunk)
      (ac/then-compose
       (fn [entries]
         (ac/all-of
          (mapv (fn [[hash resource]]
                  (resource-indexer/index-resource resource-indexer last-updated
                                                   hash resource))
                entries))))))

The size of a chunk is known before it is dispatched, in both paths, so the loop can account for it without asking anybody.

Budget and chunk size

Both are derived from the width of the executor, which the resource indexer owns. blaze.executors gains pool-size (.getCorePoolSize; io-pool is a plain ThreadPoolExecutor, as the thread pool executor collector already assumes):

(def ^:private ^:const look-ahead-factor 8)
(def ^:private ^:const chunk-factor 2)

(defn look-ahead-resources
  "Returns the maximum number of resources the indexing loop may have dispatched
  but not yet awaited."
  [{:keys [executor]}]
  (* look-ahead-factor (ex/pool-size executor)))

(defn chunk-size
  "Returns the number of resources the indexing loop dispatches and fetches at
  once."
  [{:keys [executor]}]
  (* chunk-factor (ex/pool-size executor)))

At 16 threads that is a window of 128 resources in chunks of 32, at the default of 4 threads a window of 32 in chunks of 8. Four chunks fit into the window, so the fetch of the next chunk overlaps the indexing of the current one and the pool doesn't idle between chunks. No new environment variable, DB_RESOURCE_INDEXER_THREADS governs both.

Indexing loop

index-batch! replaces the run! over the polled batch. It walks the transactions of the batch in order and keeps the following state:

• in-flight — the number of resources dispatched but not yet awaited
• a FIFO of dispatched chunks, each with its size and its future
• the transactions of the batch in order, each with the chunks it has left to dispatch and the futures of the chunks it has dispatched

One step of the loop:

1. If in-flight is below the budget and any transaction still has a chunk left to dispatch, dispatch the next chunk — earliest transaction first — add its future to that transaction and to the FIFO, and add its size to in-flight.
2. Otherwise await the head of the FIFO, subtract its size from in-flight and drop it.
3. When the head transaction has no chunk left to dispatch and none outstanding, apply it: index-tx, store-tx-entries!, commit-success!, exactly as today but without waiting for resources, because they are already indexed. Then check :run? and continue with the next transaction.

Invariants:

• in-flight never exceeds the budget, whatever the size of a transaction. There is no escape hatch for large transactions.
• At most one chunk worth of resources of any transaction is being fetched at a time, so the query concurrency against the resource store is bounded by the chunk size and the resource memory by the window.
• Chunks are awaited in dispatch order, and transactions are applied in order, so the order of the observable effects is unchanged.
• The loop drains its outstanding chunks in a finally, awaiting them and discarding results and exceptions, so no resource indexing task outlives the indexing loop — on a normal end, on a stop via :run? and on an exception. The drain is bounded by the window.

index-tx-data! keeps everything order dependent and no longer dispatches or waits:

(defn- index-tx-data!
  [{:keys [node-name kv-store read-only-matcher] :as node}
   {:keys [t instant] :as tx-data}]
  (let [result (index-tx node-name {:db-before (np/-db node)
                                    :read-only-matcher read-only-matcher} tx-data)]
    (if (ba/anomaly? result)
      (commit-error! node t result)
      (do
        (store-tx-entries! node-name kv-store result)
        (commit-success! node t instant)))))

wait-for-resources goes away. index-loop keeps its while (:run? @state) as the outer guard, the per transaction check inside index-batch! makes closing responsive regardless of the batch size. Abandoned transactions are durable in the transaction log and are indexed at the next start, which is what closed-before-indexed-msg already tells callers.

Metrics

No new metric. index-resources keeps measuring the actual duration of the resource indexing of one transaction: the timer starts when the first chunk of that transaction is dispatched and is observed by a when-complete on the aggregate of its chunk futures, so it measures when the resources were actually indexed and not when the loop got around to awaiting them. It includes the resource store fetch, as it does today, and now also the time chunks spend queued.

Two consequences, documented rather than fixed:

• under a full window the value reads higher than today's ~181 µs without anything being slower
• it is observed even for transactions whose index-tx failed, where today the observation is dropped because it happens in wait-for-resources, which the error path skips. The resource indexing did happen and did cost time.

The residual wait, today's ~139 µs, is no longer measured anywhere. The evidence that the pipeline works is the throughput together with thread_pool_executor_active_count of the resource indexer executor rising towards DB_RESOURCE_INDEXER_THREADS.

Both Durations panels of the dashboard get a description stating that the ops overlap — index-resources runs concurrently with index-transactions and store-tx-entries of its own transaction and with the ops of later transactions — so that the durations must not be summed to obtain the time of the indexing thread. The defhistogram docstring of the duration-seconds histogram of the node gets the same statement, it currently documents no op semantics at all. The Indexer Utilization panel is unaffected, it is 1 - rate(…op="poll-tx-log") and the meaning of that op doesn't change.

Error handling and ordering

• index-tx anomaly — unchanged. commit-error! writes the error entry and advance-error-t! releases the in-flight place. The chunks of that transaction are not awaited by the apply step, they are drained like any other outstanding chunk.
• Resource indexing failure — the chunk future completes exceptionally and the await in the loop rethrows, index-loop catches, fail! and finish-indexing! run. Identical to today. Because chunks are awaited in dispatch order, a failure surfaces in transaction order even if it completed earlier, so the order of the observable effects doesn't change.
• release-in-flight stays driven by the commit, so the in-flight accounting of the node needs no change.
• Leftovers — transactions that were dispatched but never committed have written index entries. Those are keyed by t and unreachable without TxSuccess, and the next start re-indexes from the committed t, rewriting them idempotently. Same class of leftover as today, now bounded by the window instead of by the size of a transaction.
• No cancellation. allOf doesn't propagate cancellation to its components, and cancelling wouldn't stop running tasks anyway. With the window bounding the drain there is little left to win. It composes cleanly later if the drain ever shows up.

Poll batch size

max-poll-size stays at 50. Raising or removing it would recover the stall at the poll boundary — one stall per batch, about 1,4 % — but costs retained buffer memory: trim-buffer runs only at the start of -poll, so the acknowledgement lag equals the batch size and the buffer sawtooths between max-in-flight and max-in-flight plus the batch size. Decide it from blaze_db_tx_log_buffer_entries{state="retained"} during the next load test run instead of now, it is a one constant change either way.

Files

File	Change
modules/executor/src/blaze/executors.clj	add pool-size, import ThreadPoolExecutor
modules/executor/src/blaze/executors_spec.clj	fdef for pool-size
modules/db/src/blaze/db/node/resource_indexer.clj	index-resource replaces index-resources, index-resources* and async-index-resource; cmd-rs-keys moves to node.clj unchanged; add look-ahead-resources and chunk-size
modules/db/test/blaze/db/node/resource_indexer_spec.clj	fdefs for index-resource, look-ahead-resources and chunk-size
modules/db/test/blaze/db/node/resource_indexer_test.clj	rewritten for the per resource API
modules/db/src/blaze/db/node.clj	work, index-chunk!, index-batch!, drain!, changed index-tx-data!, poll-and-index!, duration-seconds docstring; wait-for-resources removed
modules/monitoring/dashboard.edn	description on both Durations panels
modules/db/test/blaze/db/node_test.clj	new tests, existing with-redefs switched to index-resource
docs/implementation/database.md	paragraph on the look-ahead and the chunking under Apply-Time Mechanics

Testing

Written first, each failing against today's code.

1. Dispatch precedes apply. A reified TxLog returning a fixed batch of four single resource transactions, with with-redefs on resource-indexer/index-resource recording each resource and returning a future gated on at least two transactions having been dispatched. Today the first await never completes and the bounded deref times out, with the look-ahead the node reaches t = 4. Also asserts that the dispatch order equals the transaction order.
2. The window is never exceeded. The stub tracks the number of concurrently dispatched but not yet completed resources and records the maximum. For a batch of transactions totalling far more resources than the window, that maximum has to stay at or below look-ahead-resources.
3. A single large transaction is chunked. One transaction with more resources than the window, without local payload, against a resource store stub that records the size of every multi-get. No call may exceed chunk-size, and the recorded maximum of concurrently outstanding resources has to stay within the window. This is the memory and query concurrency bound and it fails today, where one multi-get receives every key of the transaction.
4. No task outlives the loop. The stub counts starts and finishes, after closing they have to be equal, exercising the drain in the finally.
5. Stopping mid batch. Close while the first transaction is still being indexed, assert that t is below the end of the batch, reopen against the same key-value store and assert that the remainder is indexed.
6. Resource selection per path. Both paths select the same set once the work list moves into the node: a transaction with a local payload indexes the payload, one without indexes the commands that have a hash and aren't keep. Guards the rule Skip Resource Indexing for Keep Commands #4102 established while it crosses the boundary.
7. Existing behaviour. The failing resource indexer tests in node_test.clj keep passing unchanged, they are the regression guard for the failure ordering.

Verification: make fmt, make lint, make -C modules/db test, make -C modules/executor test, make test-coverage at ≥ 95 % forms.

Expected effect and non-goals

With the wait removed, the throughput of the indexing loop should be bound by the resource indexer executor at roughly 11.000 tx/s instead of 5300, and DB_RESOURCE_INDEXER_THREADS becomes effective again. Contention on RocksDB and on the state atom of the node will take a part of that, so the load test has to show what is left.

Independent of throughput, the chunking bounds two things that scale with the transaction size today: the number of concurrent queries against the resource store and the number of decoded resources held in memory while indexing one transaction. Both matter most in a distributed setup, where every node fetches the resources of every transaction from Cassandra.

Non-goals: node.clj is about 1300 lines and hosts the indexing loop, the publishing loop, the submit path, the waiters and the in-flight accounting. Extracting the indexing loop into its own namespace would make this logic directly unit testable, but that is a refactoring beyond this issue. max-poll-size and cancellation on close are deliberate follow-ups, each gated on a measurement named above.