[Jetty] Let a positions-only reader replay its reads from the store.

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 {

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

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 `<base>:t:<topic>:floors` keeps one field per reader, its run id
+and subscription id, with `<entry id>|<until ms>`: 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,

crates/streams/lua/append.lua

@@ -27,15 +27,15 @@ 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])
   if not first then
     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'}

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

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

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

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