diff --git a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto
index 91f15c6af..0055aa419 100644
--- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto
+++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto
@@ -43,6 +43,12 @@ service StreamService {
// wait limit (`streamnotifier.pollMaxWait`). The reader then reads again with `arm`. After the
// wait limit, that read finds records whose notification was lost, or arms again.
rpc WaitForNotification(WaitForNotificationRequest) returns (WaitForNotificationResponse);
+ // Reads again the exact range an earlier read answered, for a positions-only Workflow reader on
+ // replay, and checks it against that read's digest. Waits for nothing and checks no owner.
+ rpc ReadRange(ReadRangeRequest) returns (ReadResponse);
+ // Ends a positions-only reader's hold on a stream's records after a grace, so retention takes
+ // them again.
+ rpc ReleaseFloor(ReleaseFloorRequest) returns (ReleaseFloorResponse);
// What the store can do beyond the log every store gives, and its limits.
rpc DescribeStore(DescribeStoreRequest) returns (DescribeStoreResponse);
}
@@ -90,6 +96,11 @@ message RedisStoreConfig {
uint32 blocking_reads_per_node = 5;
// How long a short call waits for its reply. Unset uses the default.
google.protobuf.Duration response_timeout = 6;
+ // How long a positions-only reader holds a stream's records after its last read. Unset uses
+ // `retention`.
+ google.protobuf.Duration floor_lease = 7;
+ // How long a hold lasts after its reader released it. Unset keeps one day.
+ google.protobuf.Duration floor_release_grace = 8;
// Runs with no Lua, so the store keeps no producer state and no last position, and Core takes
// their fallbacks. For tests of the fallbacks on a real store.
bool plain = 10;
@@ -197,6 +208,10 @@ message ReadRequest {
// finds nothing, the store arms the stream in the same step, so the next append notifies the
// notifier. A reader that doesn't wait on the notifier leaves it unset.
bool arm = 6;
+ // For a positions-only Workflow reader: names the reader, as its run id and its subscription
+ // id. The store holds the records after this read's start against retention for that holder,
+ // until `ReleaseFloor` and its grace, or until the holder's lease passes with no read.
+ string floor_holder = 7;
// The most stored record bytes one answer holds past its first record. Zero puts no bound. A
// Workflow reader that records its answers in History sets it, so an answer stays below the
// server's blob size limit.
@@ -249,8 +264,46 @@ message ReadResponse {
temporal.api.failure.v1.Failure close_failure = 8;
// Set with `done` to the `closed_by` of the topic's first close.
string closed_by = 9;
+ // SHA-256 over the answer's records, as `ReadRange` checks it: per record its cursor, its stale
+ // flag, and for a stored record its kind, producer, attempt, sequence and plaintext content
+ // hash, or for a supersession its producer and both attempts. It leaves out the encoded body.
+ bytes records_digest = 10;
+ // The first run of the chain the read read. `ReadRange` takes it, so replay needs no describe.
+ string chain_first_run_id = 11;
+ // The cursor this answer reads strictly after, with `$end` resolved. With `cursor`, it is the
+ // range `ReadRange` reads again.
+ string start_cursor = 12;
+}
+
+message ReadRangeRequest {
+ // The stream, as the read named it.
+ StreamAddress stream = 1;
+ // The read's `chain_first_run_id`.
+ string chain_first_run_id = 2;
+ // The read's `start_cursor` and its answer's `cursor`.
+ string after = 3;
+ string to = 4;
+ // How many stored records the answer held, not counting supersessions.
+ uint32 stored_records = 5;
+ // The read's input `state`.
+ bytes state = 6;
+ // The answer's `records_digest`.
+ bytes records_digest = 7;
+ // As in `ReadRequest`, so a replay holds the records for its own run.
+ string floor_holder = 8;
+ // Set when the read's `after` was `$end`. That read looked up no record at the start it found,
+ // and the replay must not either, or a newer attempt there reads as superseded.
+ bool start_from_end = 9;
+}
+
+message ReleaseFloorRequest {
+ StreamAddress stream = 1;
+ string chain_first_run_id = 2;
+ string floor_holder = 3;
}
+message ReleaseFloorResponse {}
+
message LatestRequest {
StreamAddress stream = 1;
}
@@ -442,6 +495,12 @@ enum StreamFailureKind {
// the same call fails the same way. A lang built before this kind maps it, as any kind it
// doesn't know, to a storage failure.
STREAM_FAILURE_KIND_INVALID_ARGUMENT = 12;
+ // A replay read found other records than the read it repeats: other content, another count
+ // or another end. Lang reports it as nondeterminism.
+ STREAM_FAILURE_KIND_REPLAY_MISMATCH = 13;
+ // A replay read found that the store no longer holds the range the read it repeats got
+ // (REC-01), such as records that retention trimmed or a deleted log.
+ STREAM_FAILURE_KIND_REPLAY_RANGE_LOST = 14;
}
message StreamFailure {
diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md
index 2b1cfcfc8..135690044 100644
--- a/crates/sdk-core/CHANGELOG.md
+++ b/crates/sdk-core/CHANGELOG.md
@@ -95,6 +95,15 @@ relevant information.
tells the stream's notifier that first close too, not its own.
`StreamStore::close_topic_with` and `close_stream` take a `StoredClose`, and
`close_topic_with` answers the close the store keeps.
+* Experimental: a positions-only Workflow reader can record what it read in place of the records.
+ `ReadResponse` gains `records_digest`, `chain_first_run_id` and `start_cursor`, and the new
+ `ReadRange` stream call reads that exact range again on replay and checks the digest. Other
+ records fail as `REPLAY_MISMATCH`, and records the store no longer holds as
+ `REPLAY_RANGE_LOST` (REC-01). `ReadRequest.floor_holder` holds the records a reader read
+ against retention, under a lease that each read renews, and the new `ReleaseFloor` call ends
+ the hold after a grace. The Redis store keeps holds in a `floors` hash per topic, and its append
+ script trims around them. `RedisStoreConfig` gains `floor_lease` and `floor_release_grace`.
+ Lang sets `ReadRangeRequest.start_from_end` when the read asked for `$end`.
* Experimental: the stream service's `RetireProducer` call drops one producer attempt's dedupe
state on a stream, which the Redis store keeps as one field of the topic's meta hash until
retention. A producer that writes once per short-lived attempt calls it after the attempt's
diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md
index 8365ff4ff..a9fd8be57 100644
--- a/crates/streams/docs/redis.md
+++ b/crates/streams/docs/redis.md
@@ -36,6 +36,27 @@ reaches the end sees the outcome without a wake. An append that the close refuse
beyond the documented ones, since the script uses `HGET`, `HSET` and `PEXPIRE` only. An SDK's
provider ignores it.
+**Floors.** A positions-only Workflow reader records which range it read, not the records, and
+reads the range again on replay. So the store holds those records past retention while the
+reader's run can replay. A hash `:t::floors` keeps one field per reader, its run id
+and subscription id, with `|`: the lowest position the reader read after, and
+the end of its lease. A read or a replay read renews the lease, which lasts the retention by
+default, and extends the log's and the meta's expiry to it. `ReleaseFloor` moves the end to now
+plus a release grace, one day by default. The append script's trim reads the hash first. It drops
+the ended holds, trims to the lower of the retention floor and the lowest held position, and keeps
+the log until the latest lease ends. A hold from the beginning holds `0-1`, so it keeps the whole
+topic. Any position below `0-1` holds `0-1` too. A topic keeps at most 1,000 live holders, and a
+new holder past them is refused, since every append reads them all. The scripts use `HGET`,
+`HSET`, `HGETALL`, `HLEN`, `HDEL`, `PTTL` and `PEXPIRE`, which the documented ACL allows. Without
+holds the hash doesn't exist and the trim is as before. The append script's hash changes again.
+A stream close releases no hold, since the readers of a closed stream still replay. Only the
+reader's `ReleaseFloor` and its grace, or a lapsed lease, end one.
+
+**Mixed providers and floors.** An SDK's own Redis provider trims with its own append script and
+doesn't read the floors hash, so its appends can trim records that a positions-only reader holds.
+A topic that a positions-only reader reads must take its appends only through Core's store. Other
+providers on such a topic are unsupported.
+
**Retired producers.** A field stays until a write's trim finds it older than the retention, so
a topic written by many short-lived producers holds one field per producer attempt of the last
retention window, about 235 bytes each. `RetireProducer` deletes one attempt's field with
@@ -63,7 +84,7 @@ percent-encoded. A Redis ACL user for one namespace's applications, with the rul
conformance suite runs as:
```text
-ACL SETUSER streams-app on >secret resetkeys ~temporal-streams:{my-ns:* resetchannels -@all +evalsha +eval +script|load +multi +exec +xadd +xread +xrevrange +xrange +xtrim +xlen +hget +hset +hgetall +hincrby +hdel +hscan +rpush +lrange +exists +del +unlink +pexpire +pttl +time +info +config|get +ping +hello +client|setinfo +cluster|slots
+ACL SETUSER streams-app on >secret resetkeys ~temporal-streams:{my-ns:* resetchannels -@all +evalsha +eval +script|load +multi +exec +xadd +xread +xrevrange +xrange +xtrim +xlen +hget +hset +hgetall +hlen +hincrby +hdel +hscan +rpush +lrange +exists +del +unlink +pexpire +pttl +time +info +config|get +ping +hello +client|setinfo +cluster|slots
```
A cluster client reads the topology with `CLUSTER SLOTS` first. An ACL user lives on one node,
diff --git a/crates/streams/lua/append.lua b/crates/streams/lua/append.lua
index cc7ce94b3..a18e369a0 100644
--- a/crates/streams/lua/append.lua
+++ b/crates/streams/lua/append.lua
@@ -27,7 +27,7 @@ end
revive(KEYS[1], KEYS[2])
-- Redis doesn't roll a script back, so whatever can fail runs before the first record lands,
-- and the dedupe field goes last. A retry then never finds a batch that only half landed.
-local floor_ms = trim(KEYS[1], KEYS[2], retention)
+local floor_ms, life = trim(KEYS[1], KEYS[2], retention, KEYS[4])
local first, last
for i = 6, #ARGV do
last = redis.call('XADD', KEYS[1], '*', 'r', ARGV[i])
@@ -35,7 +35,7 @@ for i = 6, #ARGV do
first = last
end
end
-keep(KEYS[1], KEYS[2], #ARGV - 5, last, retention, grace, floor_ms)
+keep(KEYS[1], KEYS[2], #ARGV - 5, last, grace, floor_ms, life)
redis.call('HSET', KEYS[2], field,
ARGV[4] .. '|' .. (#ARGV - 5) .. '|' .. first .. '|' .. last .. '|' .. digest_arg)
return {first, last, redis.call('HGET', KEYS[2], 'arm') or '0'}
diff --git a/crates/streams/lua/floor.lua b/crates/streams/lua/floor.lua
new file mode 100644
index 000000000..d2f762c78
--- /dev/null
+++ b/crates/streams/lua/floor.lua
@@ -0,0 +1,28 @@
+local function clock_ms()
+ local time = redis.call('TIME')
+ return tonumber(time[1]) * 1000 + math.floor(tonumber(time[2]) / 1000)
+end
+local function id_below(a, b)
+ local am, as = string.match(a, '^(%d+)%-(%d+)$')
+ local bm, bs = string.match(b, '^(%d+)%-(%d+)$')
+ am, as, bm, bs = tonumber(am), tonumber(as), tonumber(bm), tonumber(bs)
+ return am < bm or (am == bm and as < bs)
+end
+local function held(floors, floor, retention)
+ local now = clock_ms()
+ local life = retention
+ local entries = redis.call('HGETALL', floors)
+ for i = 1, #entries, 2 do
+ local id, until_ms = string.match(entries[i + 1], '^(%d+%-%d+)|(%d+)$')
+ until_ms = tonumber(until_ms)
+ if not until_ms or until_ms <= now then
+ redis.call('HDEL', floors, entries[i])
+ else
+ if id_below(id, floor) then
+ floor = id
+ end
+ life = math.max(life, until_ms - now)
+ end
+ end
+ return floor, life
+end
diff --git a/crates/streams/lua/hold.lua b/crates/streams/lua/hold.lua
new file mode 100644
index 000000000..364ddffcb
--- /dev/null
+++ b/crates/streams/lua/hold.lua
@@ -0,0 +1,35 @@
+local now = clock_ms()
+local lease, grace = tonumber(ARGV[3]), tonumber(ARGV[4])
+local id = ARGV[2]
+local held_value = redis.call('HGET', KEYS[1], ARGV[1])
+if not held_value and redis.call('HLEN', KEYS[1]) >= tonumber(ARGV[5]) then
+ -- Lapsed holds count only until a scan drops them.
+ local entries = redis.call('HGETALL', KEYS[1])
+ for i = 1, #entries, 2 do
+ local until_ms = tonumber(string.match(entries[i + 1], '|(%d+)$'))
+ if not until_ms or until_ms <= now then
+ redis.call('HDEL', KEYS[1], entries[i])
+ end
+ end
+ if redis.call('HLEN', KEYS[1]) >= tonumber(ARGV[5]) then
+ return redis.error_reply('STREAMS_HOLDERS topic already has ' .. ARGV[5] ..
+ ' floor holders, so a positions-only reader can not start on it')
+ end
+end
+if held_value then
+ local old = string.match(held_value, '^(%d+%-%d+)|')
+ if old and id_below(old, id) then
+ id = old
+ end
+end
+redis.call('HSET', KEYS[1], ARGV[1], id .. '|' .. (now + lease))
+local function extend(key, ms, create)
+ local ttl = redis.call('PTTL', key)
+ if (ttl >= 0 or (create and ttl == -1)) and ttl < ms then
+ redis.call('PEXPIRE', key, ms)
+ end
+end
+extend(KEYS[1], lease, true)
+extend(KEYS[2], lease, false)
+extend(KEYS[3], lease + grace, false)
+return 1
diff --git a/crates/streams/lua/keep.lua b/crates/streams/lua/keep.lua
index 22a8873ca..f22280c3b 100644
--- a/crates/streams/lua/keep.lua
+++ b/crates/streams/lua/keep.lua
@@ -23,20 +23,22 @@ local function sweep_sessions(meta, floor_ms)
end
redis.call('HSET', meta, 'hwscan', found[1])
end
-local function trim(log, meta, retention)
+local function trim(log, meta, retention, floors)
local floor_ms = now_ms() - retention
local floor = floor_ms .. '-0'
+ local life
+ floor, life = held(floors, floor, retention)
local doomed = redis.call('XREVRANGE', log, '(' .. floor, '-', 'COUNT', 1)
if #doomed > 0 then
redis.call('XTRIM', log, 'MINID', floor)
redis.call('HSET', meta, 'trimmed', doomed[1][1])
end
- return floor_ms
+ return floor_ms, life
end
-local function keep(log, meta, added, last, retention, grace, floor_ms)
- redis.call('PEXPIRE', log, retention)
+local function keep(log, meta, added, last, grace, floor_ms, life)
+ redis.call('PEXPIRE', log, life)
redis.call('HINCRBY', meta, 'added', added)
redis.call('HSET', meta, 'last', last)
- redis.call('PEXPIRE', meta, retention + grace)
+ redis.call('PEXPIRE', meta, life + grace)
sweep_sessions(meta, floor_ms)
end
diff --git a/crates/streams/lua/release.lua b/crates/streams/lua/release.lua
new file mode 100644
index 000000000..3841c125a
--- /dev/null
+++ b/crates/streams/lua/release.lua
@@ -0,0 +1,10 @@
+local held_value = redis.call('HGET', KEYS[1], ARGV[1])
+if not held_value then
+ return 0
+end
+local id, until_ms = string.match(held_value, '^(%d+%-%d+)|(%d+)$')
+local release_ms = clock_ms() + tonumber(ARGV[2])
+if id and tonumber(until_ms) > release_ms then
+ redis.call('HSET', KEYS[1], ARGV[1], id .. '|' .. release_ms)
+end
+return 1
diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs
index 04826e18f..a5e545c0c 100644
--- a/crates/streams/src/handle.rs
+++ b/crates/streams/src/handle.rs
@@ -6,14 +6,15 @@
use crate::{
BEGINNING, END, MAX_CLOSE_BYTES, ReadTarget, StreamError, StreamResult, StreamStore,
WORKFLOW_OWNER_KIND, activity_producer_id, append_batch, check_append_digest,
- check_store_limits, close_topic, mint_cursor,
+ check_store_limits, close_topic, cursor_position, mint_cursor,
owner::{OwnerClient, OwnerDescription, OwnerError},
proto::{
ActivityProducer, AppendRequest, AppendResponse, ChainId, CloseRequest, CloseResponse,
DeleteOwnerRequest, DeleteOwnerResponse, LatestRequest, LatestResponse, NamedProducer,
- ReadRequest, ReadResponse, RetireProducerRequest, RetireProducerResponse,
- StoreAppendRequest, StoreLatestRequest, StoredClose, StreamAddress, StreamFailureKind,
- StreamOwnerKind, append_request, retire_producer_request,
+ ReadRangeRequest, ReadRequest, ReadResponse, ReleaseFloorRequest, ReleaseFloorResponse,
+ RetireProducerRequest, RetireProducerResponse, StoreAppendRequest, StoreLatestRequest,
+ StoredClose, StreamAddress, StreamFailureKind, StreamOwnerKind, append_request,
+ read_record, retire_producer_request,
},
reader::{ReadState, read_with},
stored_append_record, stream_hash,
@@ -27,6 +28,9 @@ use std::{
use temporalio_common::protos::temporal::api::enums::v1::WorkflowExecutionStatus;
use tokio::time::Instant;
+/// The longest `floor_holder` a read takes.
+const MAX_FLOOR_HOLDER_BYTES: usize = 512;
+
/// The longest `closed_by` a close takes.
const MAX_CLOSED_BY_BYTES: usize = 256;
@@ -311,6 +315,10 @@ impl Streams {
topic: stream.topic.clone(),
stream_hash: self.hash(stream),
};
+ if !request.floor_holder.is_empty() {
+ self.hold_floor(&target, &request.after, &request.floor_holder)
+ .await?;
+ }
let minimum = self.options.owner_check_min;
let mut page_request = request.clone();
let mut topic_close = None;
@@ -381,6 +389,179 @@ impl Streams {
})
}
+ /// Holds the records after `after` against retention for `holder`, a positions-only reader.
+ async fn hold_floor(&self, target: &ReadTarget, after: &str, holder: &str) -> StreamResult<()> {
+ if holder.len() > MAX_FLOOR_HOLDER_BYTES {
+ return Err(StreamError::invalid_argument(format!(
+ "a floor holder is at most {MAX_FLOOR_HOLDER_BYTES} bytes, not {}",
+ holder.len()
+ )));
+ }
+ let position = if after == END {
+ self.store
+ .latest(StoreLatestRequest {
+ chain: Some(target.chain.clone()),
+ topic: target.topic.clone(),
+ })
+ .await?
+ .position
+ } else {
+ cursor_position(after, self.store.name(), &target.stream_hash)?
+ .unwrap_or_default()
+ .to_string()
+ };
+ self.store
+ .hold_floor(&target.chain, &target.topic, holder, &position)
+ .await
+ }
+
+ /// Reads again the exact range of an earlier answer and checks it against that answer's
+ /// digest (RR-2). It runs the read that the answer came from, with the same input state and
+ /// the start that read found, so it gives the same supersessions and stale marks, and it waits
+ /// for nothing and checks no owner. Records with other content fail as a mismatch. Records the store no longer holds
+ /// fail as lost (REC-01).
+ pub async fn read_range(&self, request: ReadRangeRequest) -> StreamResult {
+ let stream = address(request.stream.as_ref())?;
+ if request.chain_first_run_id.is_empty() {
+ return Err(StreamError::invalid_argument(
+ "a replay read needs the chain its read read",
+ ));
+ }
+ if request.after == END {
+ return Err(StreamError::invalid_argument(
+ "a replay read starts at the read's resolved start_cursor, not at $end",
+ ));
+ }
+ let mut state = ReadState::decode_from(&request.state)?;
+ if state.first_run_id.is_empty() {
+ state.first_run_id = request.chain_first_run_id.clone();
+ }
+ if request.start_from_end {
+ state.started = true;
+ }
+ let target = ReadTarget {
+ chain: chain_of(stream, &request.chain_first_run_id),
+ topic: stream.topic.clone(),
+ stream_hash: self.hash(stream),
+ };
+ if !request.floor_holder.is_empty() {
+ self.hold_floor(&target, &request.after, &request.floor_holder)
+ .await?;
+ }
+ let expected = request.stored_records as usize;
+ let lost = |message: String| {
+ StreamError::new(
+ StreamFailureKind::ReplayRangeLost,
+ format!("REC-01: {message}"),
+ )
+ };
+ let mismatch =
+ |message: String| StreamError::new(StreamFailureKind::ReplayMismatch, message);
+ let page = if expected == 0 {
+ ReadResponse {
+ cursor: request.after.clone(),
+ records_digest: crate::reader::records_digest(&[]),
+ chain_first_run_id: request.chain_first_run_id.clone(),
+ start_cursor: request.after.clone(),
+ ..Default::default()
+ }
+ } else {
+ let page_request = ReadRequest {
+ stream: Some(stream.clone()),
+ after: request.after.clone(),
+ max_records: request.stored_records,
+ ..Default::default()
+ };
+ match read_with(&*self.store, &target, &page_request, &mut state).await {
+ Ok((page, _, _)) => page,
+ Err(error)
+ if matches!(
+ error.kind,
+ StreamFailureKind::Expired | StreamFailureKind::NotFound
+ ) =>
+ {
+ return Err(lost(error.message));
+ }
+ Err(error) => return Err(error),
+ }
+ };
+ let stored = page
+ .records
+ .iter()
+ .filter(|record| matches!(record.record, Some(read_record::Record::Stored(_))))
+ .count();
+ if stored < expected {
+ return Err(lost(format!(
+ "the store holds {stored} of the {expected} records after {:?} that the read got",
+ request.after
+ )));
+ }
+ if page.cursor != request.to || page.records_digest != request.records_digest {
+ // A read from the beginning has no position to check, so a trim since the read shows
+ // as other records, which end elsewhere or differ.
+ if request.after.is_empty()
+ && self
+ .store
+ .dropped_after(&target.chain, &target.topic, "")
+ .await?
+ .is_some()
+ {
+ return Err(lost(
+ "retention trimmed the start of the stream, which the read began at"
+ .to_string(),
+ ));
+ }
+ // A store that replicates late can lose acked records in a failover. The record the
+ // read ended at is gone then, and records written since read as other records.
+ if let Some(end) = cursor_position(&request.to, self.store.name(), &target.stream_hash)?
+ && self
+ .store
+ .record_at(&target.chain, &target.topic, end)
+ .await?
+ .is_none()
+ {
+ return Err(lost(format!(
+ "the store no longer holds the record at {:?} that the read ended at",
+ request.to
+ )));
+ }
+ if page.cursor != request.to {
+ return Err(mismatch(format!(
+ "the replayed read ends at {:?}, not at {:?}",
+ page.cursor, request.to
+ )));
+ }
+ return Err(mismatch(
+ "the replayed records differ from the records the read got".to_string(),
+ ));
+ }
+ Ok(ReadResponse {
+ state: state.encode_to_vec(),
+ ..page
+ })
+ }
+
+ /// Ends a positions-only reader's hold after the store's release grace.
+ pub async fn release_floor(
+ &self,
+ request: ReleaseFloorRequest,
+ ) -> StreamResult {
+ let stream = address(request.stream.as_ref())?;
+ if request.chain_first_run_id.is_empty() || request.floor_holder.is_empty() {
+ return Err(StreamError::invalid_argument(
+ "a release needs the chain and the floor holder",
+ ));
+ }
+ self.store
+ .release_floor(
+ &chain_of(stream, &request.chain_first_run_id),
+ &stream.topic,
+ &request.floor_holder,
+ )
+ .await?;
+ Ok(ReleaseFloorResponse {})
+ }
+
/// The chain a read follows, or `None` when Temporal doesn't know its owner yet, within
/// [StreamsOptions::owner_start_grace] of the read's first owner check. It asks again as
/// an idle read does until `deadline`. After the grace the read fails as not found.
@@ -1890,6 +2071,211 @@ mod tests {
setup.streams.append(append("p", 1, "1")).await.unwrap();
}
+ /// The replay read that repeats the live `read` of `request` that answered `page`.
+ fn replay_of(request: &ReadRequest, page: &ReadResponse) -> ReadRangeRequest {
+ ReadRangeRequest {
+ stream: Some(stream("out")),
+ chain_first_run_id: page.chain_first_run_id.clone(),
+ after: page.start_cursor.clone(),
+ to: page.cursor.clone(),
+ stored_records: page
+ .records
+ .iter()
+ .filter(|r| matches!(r.record, Some(read_record::Record::Stored(_))))
+ .count() as u32,
+ state: request.state.clone(),
+ records_digest: page.records_digest.clone(),
+ floor_holder: String::new(),
+ start_from_end: request.after == crate::END,
+ }
+ }
+
+ fn attempt(producer: &str, attempt: i64, sequence: i64, value: &str) -> AppendRequest {
+ AppendRequest {
+ producer: Some(Producer::Named(NamedProducer {
+ producer_id: producer.to_string(),
+ attempt,
+ })),
+ ..append(producer, sequence, value)
+ }
+ }
+
+ /// A read after the stream's first record, with its replay request.
+ async fn first_page(setup: &Setup, holder: &str) -> (ReadRequest, ReadResponse) {
+ let first = setup.streams.append(append("p", 1, "1")).await.unwrap();
+ setup.streams.append(append("p", 2, "2")).await.unwrap();
+ // A newer attempt makes the read synthesize a supersession.
+ setup.streams.append(attempt("p", 2, 1, "3")).await.unwrap();
+ let request = ReadRequest {
+ floor_holder: holder.to_string(),
+ ..read_after(&first.last_cursor, &[], Duration::ZERO)
+ };
+ let page = setup.streams.read(request.clone()).await.unwrap();
+ (request, page)
+ }
+
+ #[tokio::test]
+ async fn a_replay_read_gives_the_records_the_read_got() {
+ let setup = setup(quick());
+ let (request, page) = first_page(&setup, "").await;
+ assert!(
+ page.records
+ .iter()
+ .any(|r| matches!(r.record, Some(read_record::Record::Superseded(_)))),
+ "{page:?}"
+ );
+ assert_eq!(page.chain_first_run_id, "run-1");
+ assert_eq!(page.start_cursor, request.after);
+ assert_eq!(page.records_digest.len(), 32);
+ // More records after the range change nothing in it.
+ setup.streams.append(attempt("p", 2, 2, "4")).await.unwrap();
+ let replayed = setup
+ .streams
+ .read_range(replay_of(&request, &page))
+ .await
+ .unwrap();
+ assert_eq!(replayed.records, page.records);
+ assert_eq!(replayed.cursor, page.cursor);
+ assert_eq!(replayed.records_digest, page.records_digest);
+ }
+
+ #[tokio::test]
+ async fn a_replay_read_of_a_read_from_the_end_gives_the_records_the_read_got() {
+ let setup = setup(quick());
+ setup.streams.append(attempt("p", 1, 1, "1")).await.unwrap();
+ let request = read_after(crate::END, &[], Duration::from_secs(5));
+ // The read finds its start at the end, then a newer attempt of the producer there lands.
+ let (page, _) = tokio::join!(setup.streams.read(request.clone()), async {
+ tokio::time::sleep(Duration::from_millis(50)).await;
+ setup.streams.append(attempt("p", 2, 1, "2")).await.unwrap();
+ });
+ let page = page.unwrap();
+ assert_eq!(page.records.len(), 1, "no supersession, {page:?}");
+ let replay = replay_of(&request, &page);
+ assert!(replay.start_from_end);
+ let replayed = setup.streams.read_range(replay).await.unwrap();
+ assert_eq!(replayed.records, page.records);
+ }
+
+ #[tokio::test]
+ async fn a_replay_read_needs_no_owner() {
+ let setup = setup(quick());
+ let (request, page) = first_page(&setup, "").await;
+ // History dropped the owner, as for a Replayer with only the store.
+ setup.owners.end("wf", WorkflowExecutionStatus::Completed);
+ let before = setup.owners.describes();
+ setup
+ .streams
+ .read_range(replay_of(&request, &page))
+ .await
+ .unwrap();
+ assert_eq!(setup.owners.describes(), before);
+ }
+
+ #[tokio::test]
+ async fn a_replay_read_with_other_records_is_a_mismatch() {
+ let setup = setup(quick());
+ let (request, page) = first_page(&setup, "").await;
+ let mut other_digest = replay_of(&request, &page);
+ other_digest.records_digest[0] ^= 1;
+ let mut other_end = replay_of(&request, &page);
+ other_end.stored_records -= 1;
+ for replay in [other_digest, other_end] {
+ let error = setup.streams.read_range(replay).await.unwrap_err();
+ assert_eq!(error.kind, StreamFailureKind::ReplayMismatch, "{error}");
+ }
+ }
+
+ #[tokio::test]
+ async fn a_replay_read_of_records_retention_dropped_is_rec_01() {
+ let setup = setup(quick());
+ let (request, page) = first_page(&setup, "").await;
+ setup.store.truncate(&chain("run-1"), "out", 1);
+ let error = setup
+ .streams
+ .read_range(replay_of(&request, &page))
+ .await
+ .unwrap_err();
+ assert_eq!(error.kind, StreamFailureKind::ReplayRangeLost, "{error}");
+ assert!(error.message.starts_with("REC-01"), "{error}");
+ }
+
+ #[tokio::test]
+ async fn a_replay_read_from_a_trimmed_start_is_rec_01() {
+ let setup = setup(quick());
+ setup.streams.append(append("p", 1, "1")).await.unwrap();
+ setup.streams.append(append("p", 2, "2")).await.unwrap();
+ let request = read_after("", &[], Duration::ZERO);
+ let page = setup.streams.read(request.clone()).await.unwrap();
+ assert_eq!(page.start_cursor, "");
+ // Later records stand where the range began, so the replay finds as many as it wants.
+ setup.store.truncate(&chain("run-1"), "out", 0);
+ for sequence in 3..=4 {
+ setup
+ .streams
+ .append(append("p", sequence, &sequence.to_string()))
+ .await
+ .unwrap();
+ }
+ let error = setup
+ .streams
+ .read_range(replay_of(&request, &page))
+ .await
+ .unwrap_err();
+ assert_eq!(error.kind, StreamFailureKind::ReplayRangeLost, "{error}");
+ }
+
+ #[tokio::test]
+ async fn a_floor_keeps_what_a_positions_only_reader_read_until_it_is_released() {
+ let setup = setup(quick());
+ let (request, page) = first_page(&setup, "reader-run/1").await;
+ setup.store.truncate(&chain("run-1"), "out", 1);
+ let replayed = setup
+ .streams
+ .read_range(ReadRangeRequest {
+ floor_holder: "reader-run/1".to_string(),
+ ..replay_of(&request, &page)
+ })
+ .await
+ .unwrap();
+ assert_eq!(replayed.records, page.records, "the floor kept the range");
+ setup
+ .streams
+ .release_floor(ReleaseFloorRequest {
+ stream: Some(stream("out")),
+ chain_first_run_id: "run-1".to_string(),
+ floor_holder: "reader-run/1".to_string(),
+ })
+ .await
+ .unwrap();
+ setup.store.truncate(&chain("run-1"), "out", 1);
+ let error = setup
+ .streams
+ .read_range(replay_of(&request, &page))
+ .await
+ .unwrap_err();
+ assert_eq!(error.kind, StreamFailureKind::ReplayRangeLost, "{error}");
+ }
+
+ #[tokio::test]
+ async fn a_replay_read_needs_its_chain_and_a_resolved_start() {
+ let setup = setup(quick());
+ let (request, page) = first_page(&setup, "").await;
+ for replay in [
+ ReadRangeRequest {
+ chain_first_run_id: String::new(),
+ ..replay_of(&request, &page)
+ },
+ ReadRangeRequest {
+ after: crate::END.to_string(),
+ ..replay_of(&request, &page)
+ },
+ ] {
+ let error = setup.streams.read_range(replay).await.unwrap_err();
+ assert_eq!(error.kind, StreamFailureKind::InvalidArgument, "{error}");
+ }
+ }
+
#[tokio::test]
async fn only_the_pass_that_waits_out_the_read_arms() {
// Owner checks every 20 ms cut the wait into passes. One arm for the whole wait keeps a
diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs
index 5518b6cb5..88ab174ef 100644
--- a/crates/streams/src/lib.rs
+++ b/crates/streams/src/lib.rs
@@ -46,8 +46,8 @@ pub use record::{
};
pub use service::StreamService;
pub use store::{
- Capabilities, CloseOutcome, ConditionForms, MAX_BATCH_BYTES, MAX_CLOSE_BYTES, MAX_RECORD_BYTES,
- NewestOrArm, StreamStore, empty_close_result,
+ Capabilities, CloseOutcome, ConditionForms, MAX_BATCH_BYTES, MAX_CLOSE_BYTES,
+ MAX_FLOOR_HOLDERS, MAX_RECORD_BYTES, NewestOrArm, StreamStore, empty_close_result,
};
/// The protos stores and lang exchange.
diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs
index 31db743d5..8a5486526 100644
--- a/crates/streams/src/memory.rs
+++ b/crates/streams/src/memory.rs
@@ -100,6 +100,9 @@ struct Chain {
/// Kept apart from the topics, so an arm on a topic with no records leaves it unknown to
/// reads and to the topic list.
arms: HashMap,
+ /// Per topic, the position each positions-only reader holds the records after. `None` holds
+ /// the whole topic.
+ floors: HashMap>>,
}
/// A topic's arm: the generation a reader set, zero once cleared, and the last one handed out.
@@ -166,15 +169,23 @@ impl MemoryStore {
}
/// Drops all but the newest `keep` records of a topic, standing in for a store's retention.
+ /// It keeps the records a positions-only reader holds, from the position it holds on.
pub fn truncate(&self, chain: &ChainId, topic: &str, keep: usize) {
let mut chains = self.chains.lock().unwrap();
- let Some(topic) = chains
- .get_mut(&chain_key(chain))
- .and_then(|chain| chain.topics.get_mut(topic))
- else {
+ let Some(chain) = chains.get_mut(&chain_key(chain)) else {
return;
};
- let drop = topic.records.len().saturating_sub(keep);
+ let held = chain
+ .floors
+ .get(topic)
+ .and_then(|holders| holders.values().map(|position| position.unwrap_or(0)).min());
+ let Some(topic) = chain.topics.get_mut(topic) else {
+ return;
+ };
+ let mut drop = topic.records.len().saturating_sub(keep);
+ if let Some(held) = held {
+ drop = drop.min(held.saturating_sub(topic.base) as usize);
+ }
topic.records.drain(..drop);
topic.base += drop as u64;
}
@@ -471,6 +482,46 @@ impl StreamStore for MemoryStore {
.map(|offset| offset.to_string()))
}
+ async fn hold_floor(
+ &self,
+ chain: &ChainId,
+ topic: &str,
+ holder: &str,
+ position: &str,
+ ) -> StreamResult<()> {
+ let position = parse_position(position)?;
+ let mut chains = self.chains.lock().unwrap();
+ let holders = chains
+ .entry(chain_key(chain))
+ .or_default()
+ .floors
+ .entry(topic.to_string())
+ .or_default();
+ if holders.len() >= crate::MAX_FLOOR_HOLDERS && !holders.contains_key(holder) {
+ return Err(StreamError::refused(too_many_holders(topic)));
+ }
+ let held = holders.entry(holder.to_string()).or_insert(position);
+ *held = match (*held, position) {
+ (Some(held), Some(position)) => Some(held.min(position)),
+ _ => None,
+ };
+ Ok(())
+ }
+
+ /// The memory store has no clock for retention, so a release ends the hold at once.
+ async fn release_floor(&self, chain: &ChainId, topic: &str, holder: &str) -> StreamResult<()> {
+ if let Some(holders) = self
+ .chains
+ .lock()
+ .unwrap()
+ .get_mut(&chain_key(chain))
+ .and_then(|chain| chain.floors.get_mut(topic))
+ {
+ holders.remove(holder);
+ }
+ Ok(())
+ }
+
async fn close_chain(&self, chain: &ChainId) -> StreamResult<()> {
self.producer_state()?;
self.chains
@@ -595,6 +646,15 @@ impl StreamStore for MemoryStore {
}
}
+/// Why a new floor holder is refused.
+fn too_many_holders(topic: &str) -> String {
+ format!(
+ "topic {topic:?} already has {} floor holders, so a positions-only reader can't start \
+ on it; let readers finish or Continue-as-New",
+ crate::MAX_FLOOR_HOLDERS
+ )
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -753,4 +813,21 @@ mod tests {
.unwrap_err();
assert_eq!(error.kind, StreamFailureKind::Unsupported);
}
+
+ #[tokio::test]
+ async fn a_topic_refuses_a_floor_holder_past_the_cap() {
+ let store = MemoryStore::new();
+ for holder in 0..crate::MAX_FLOOR_HOLDERS {
+ store
+ .hold_floor(&chain(), "out", &holder.to_string(), "")
+ .await
+ .unwrap();
+ }
+ let error = store
+ .hold_floor(&chain(), "out", "one-more", "")
+ .await
+ .unwrap_err();
+ assert_eq!(error.kind, StreamFailureKind::Refused);
+ store.hold_floor(&chain(), "out", "0", "").await.unwrap();
+ }
}
diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs
index 700552f6e..70d67737d 100644
--- a/crates/streams/src/notify.rs
+++ b/crates/streams/src/notify.rs
@@ -942,6 +942,20 @@ impl StreamStore for NotifyingStore {
self.inner.delete_owner(request).await
}
+ async fn hold_floor(
+ &self,
+ chain: &ChainId,
+ topic: &str,
+ holder: &str,
+ position: &str,
+ ) -> StreamResult<()> {
+ self.inner.hold_floor(chain, topic, holder, position).await
+ }
+
+ async fn release_floor(&self, chain: &ChainId, topic: &str, holder: &str) -> StreamResult<()> {
+ self.inner.release_floor(chain, topic, holder).await
+ }
+
async fn close_stream(
&self,
chain: &ChainId,
diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs
index b9940a9ac..04b56b9f2 100644
--- a/crates/streams/src/reader.rs
+++ b/crates/streams/src/reader.rs
@@ -77,7 +77,7 @@ pub(crate) struct ReadState {
producers: HashMap,
/// Set by the first call, so later calls don't look up the resume record again.
#[prost(bool, tag = "2")]
- started: bool,
+ pub(crate) started: bool,
/// The first run of the chain the read follows, so later calls don't resolve it again.
#[prost(string, tag = "3")]
pub(crate) first_run_id: String,
@@ -380,7 +380,7 @@ pub(crate) async fn read_with(
}
let (mut closed, armed, mut close) = (page.closed, page.armed, page.close);
let mut records = Vec::with_capacity(page.records.len());
- let mut previous = start;
+ let mut previous = start.clone();
let max_bytes = usize::try_from(request.max_bytes).unwrap_or(usize::MAX);
let (mut bytes, mut kept) = (0usize, 0usize);
for stored in page.records {
@@ -449,7 +449,6 @@ pub(crate) async fn read_with(
state.fit_within(MAX_STATE_BYTES);
Ok((
ReadResponse {
- records,
cursor: previous,
state: Vec::new(),
done: false,
@@ -461,6 +460,10 @@ pub(crate) async fn read_with(
} else {
0
},
+ records_digest: records_digest(&records),
+ records,
+ chain_first_run_id: target.chain.first_run_id.clone(),
+ start_cursor: start.clone(),
..Default::default()
},
closed,
@@ -468,6 +471,45 @@ pub(crate) async fn read_with(
))
}
+/// The digest of a read's answer that a replay read checks (RR-1). It covers what a reader acts
+/// on and leaves out the encoded body, so a codec change keeps it. Each field has a length
+/// prefix, so two answers can't give the same bytes.
+pub(crate) fn records_digest(records: &[ReadRecord]) -> Vec {
+ use sha2::{Digest, Sha256};
+ let mut digest = Sha256::new();
+ let mut field = |bytes: &[u8]| {
+ digest.update((bytes.len() as u64).to_be_bytes());
+ digest.update(bytes);
+ };
+ for record in records {
+ field(record.cursor.as_bytes());
+ field(&[u8::from(record.stale)]);
+ match &record.record {
+ Some(read_record::Record::Stored(stored)) => {
+ field(b"stored");
+ field(&stored.kind.to_be_bytes());
+ field(stored.producer_id.as_bytes());
+ field(&stored.attempt.to_be_bytes());
+ field(&stored.sequence.to_be_bytes());
+ field(
+ stored
+ .metadata
+ .get(temporalio_common::streams::CONTENT_HASH_KEY)
+ .map_or(&[][..], |hash| hash.data.as_slice()),
+ );
+ }
+ Some(read_record::Record::Superseded(superseded)) => {
+ field(b"superseded");
+ field(superseded.producer_id.as_bytes());
+ field(&superseded.previous_attempt.to_be_bytes());
+ field(&superseded.attempt.to_be_bytes());
+ }
+ None => field(b"none"),
+ }
+ }
+ digest.finalize().to_vec()
+}
+
/// The first poll interval of a read on a store with no long poll, and the longest.
const POLL_FIRST: Duration = Duration::from_millis(50);
const POLL_MAX: Duration = Duration::from_secs(1);
diff --git a/crates/streams/src/redis/errors.rs b/crates/streams/src/redis/errors.rs
index c094d00e9..7dab0a44d 100644
--- a/crates/streams/src/redis/errors.rs
+++ b/crates/streams/src/redis/errors.rs
@@ -44,6 +44,7 @@ pub(crate) fn stream_error(error: &RedisError, call: Call) -> StreamError {
Some("STREAMS_DIVERGENT") => StreamError::new(StreamFailureKind::ProducerDivergent, detail),
Some("STREAMS_STALE") => StreamError::new(StreamFailureKind::ProducerStale, detail),
Some("STREAMS_CLOSED") => StreamError::closed(detail),
+ Some("STREAMS_HOLDERS") => StreamError::refused(detail),
Some(code) if TRANSIENT.contains(&code) => match call {
Call::Append => StreamError::outcome_unknown(format!(
"Redis could not take the batch for now, so it may or may not have landed: \
diff --git a/crates/streams/src/redis/keys.rs b/crates/streams/src/redis/keys.rs
index 548b86983..b26cf27ee 100644
--- a/crates/streams/src/redis/keys.rs
+++ b/crates/streams/src/redis/keys.rs
@@ -7,6 +7,7 @@
//! ```text
//! :{::}:t: the log
//! :{::}:t::meta its meta
+//! :{::}:t::floors the readers that hold it
//! :{::}:chain the close flag
//! ```
//!
@@ -76,6 +77,11 @@ impl ChainKeys {
format!("{}:chain", self.base)
}
+ /// The positions-only readers that hold a topic's records, so a replay finds them again.
+ pub(crate) fn floors(&self, topic: &str) -> String {
+ format!("{}:floors", self.log(topic))
+ }
+
/// The pattern that matches every topic key of the chain, logs and metas.
pub(crate) fn topics_pattern(&self) -> String {
format!("{}:t:*", self.base)
diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs
index 6a90be565..15b2c94cb 100644
--- a/crates/streams/src/redis/mod.rs
+++ b/crates/streams/src/redis/mod.rs
@@ -70,6 +70,11 @@ pub struct RedisStoreOptions {
pub blocking_reads_per_node: usize,
/// How long a short call waits for its reply.
pub response_timeout: Duration,
+ /// How long a positions-only reader's hold lasts after its last read. `None` uses the
+ /// retention.
+ pub floor_lease: Option,
+ /// How long a hold lasts after its reader released it.
+ pub floor_release_grace: Duration,
/// Runs with no Lua: no dedupe entries, no close flags and no topic meta. Core then takes
/// the fallbacks of producer state and the kept last position, as on a store that has
/// only a log. For tests of those fallbacks on a real store.
@@ -86,6 +91,8 @@ impl RedisStoreOptions {
retention: Duration::from_secs(7 * 24 * 60 * 60),
blocking_reads_per_node: 256,
response_timeout: Duration::from_secs(5),
+ floor_lease: None,
+ floor_release_grace: Duration::from_secs(24 * 60 * 60),
plain: false,
}
}
@@ -98,6 +105,8 @@ pub struct RedisStore {
prefix: String,
retention_ms: u64,
grace_ms: u64,
+ floor_lease_ms: u64,
+ floor_release_grace_ms: u64,
plain: bool,
}
@@ -157,6 +166,10 @@ impl RedisStore {
prefix: options.key_prefix,
retention_ms,
grace_ms: TOMBSTONE_GRACE.as_millis() as u64,
+ floor_lease_ms: options
+ .floor_lease
+ .map_or(retention_ms, |lease| millis(lease).max(1)),
+ floor_release_grace_ms: millis(options.floor_release_grace),
plain: options.plain,
})
}
@@ -595,6 +608,10 @@ fn wait_of(request: &StoreReadRequest) -> Duration {
})
}
+fn millis(duration: Duration) -> u64 {
+ u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
+}
+
/// A topic's arm as a script answered it, zero when none.
fn generation(arm: &str) -> StreamResult {
arm.parse()
@@ -638,6 +655,7 @@ impl StreamStore for RedisStore {
.key(keys.log(&request.topic))
.key(keys.meta(&request.topic))
.key(keys.chain())
+ .key(keys.floors(&request.topic))
.arg(self.retention_ms)
.arg(self.grace_ms)
.arg(session_field(&request.producer_id, request.attempt))
@@ -828,6 +846,47 @@ impl StreamStore for RedisStore {
Ok(Some(trimmed))
}
+ async fn hold_floor(
+ &self,
+ chain: &ChainId,
+ topic: &str,
+ holder: &str,
+ position: &str,
+ ) -> StreamResult<()> {
+ let keys = ChainKeys::new(&self.prefix, chain);
+ // Redis never stores `0-0`, so `0-1` holds every record, and the trim's exclusive range
+ // below it stays valid. Below it, a crafted `0-0` would fail every later append.
+ let position = if position.is_empty() || entry(position)? < (0, 1) {
+ "0-1"
+ } else {
+ position
+ };
+ scripts::HOLD_SCRIPT
+ .key(keys.floors(topic))
+ .key(keys.log(topic))
+ .key(keys.meta(topic))
+ .arg(holder)
+ .arg(position)
+ .arg(self.floor_lease_ms)
+ .arg(self.grace_ms)
+ .arg(crate::MAX_FLOOR_HOLDERS)
+ .invoke_async::(&mut self.shared.clone())
+ .await
+ .mapped(Call::Write)
+ .map(drop)
+ }
+
+ async fn release_floor(&self, chain: &ChainId, topic: &str, holder: &str) -> StreamResult<()> {
+ scripts::RELEASE_SCRIPT
+ .key(ChainKeys::new(&self.prefix, chain).floors(topic))
+ .arg(holder)
+ .arg(self.floor_release_grace_ms)
+ .invoke_async::(&mut self.shared.clone())
+ .await
+ .mapped(Call::Write)
+ .map(drop)
+ }
+
async fn close_chain(&self, chain: &ChainId) -> StreamResult<()> {
self.producer_state()?;
let keys = ChainKeys::new(&self.prefix, chain);
diff --git a/crates/streams/src/redis/scripts.rs b/crates/streams/src/redis/scripts.rs
index fdf1492a8..f5122475e 100644
--- a/crates/streams/src/redis/scripts.rs
+++ b/crates/streams/src/redis/scripts.rs
@@ -2,14 +2,16 @@
//!
//! The append script is the one the SDKs sent, with one more check, the topic's close flag, so a
//! batch can't land after the topic's close returned (DD-16). It also reports the topic's arm, so
-//! a producer notifies only while a reader sleeps (DD-118). Redis doesn't roll a script back, so
-//! it runs every step that can fail before its first record lands. `keep.lua` is the prelude of
-//! every script that writes a log.
+//! a producer notifies only while a reader sleeps (DD-118), and its trim keeps what positions-only
+//! readers hold. Redis doesn't roll a script back, so it runs every step that can fail before its
+//! first record lands. `keep.lua` is the prelude of every script that writes a log, and
+//! `floor.lua` of every script that reads or writes the holds.
use redis::Script;
use std::sync::LazyLock;
pub(crate) const APPEND: &str = concat!(
+ include_str!("../../lua/floor.lua"),
include_str!("../../lua/keep.lua"),
include_str!("../../lua/append.lua")
);
@@ -23,6 +25,18 @@ pub(crate) static DISARM_SCRIPT: LazyLock