streaming: support exact publication on TTL streams

streaming/README.md

@@ -74,23 +74,22 @@ of at most 256 bytes. The length-delimited name, topic, and payload may total at
 most 1 MiB. The first call records those exact canonical bytes with the event
 ID; no digest is used. An exact retry returns the same ID, while reusing the key
 for different bytes returns `ErrIdempotencyConflict`. The event and its
-generation-scoped dedupe record
-share one Redis-absolute deadline, checked with Redis `TIME`, so retrying an
-ambiguous client result is safe even after max-length trimming removes the
-event itself. The returned ID remains the publication result even when that
-event is no longer retained.
-
-`AddOnce` takes no deadline argument: the active stream generation's immutable
-deadline is the sole expiry authority. A handle may adopt that existing
-deadline by omitting retention options; an explicitly configured handle must
-match it exactly.
-Ordinary `Add` calls use that same absolute expiry and never extend it. At or
-after expiry, Add, AddOnce, and
-Snapshot return `ErrDeadlineElapsed`; Sink.Close treats expiry as terminal and
-still closes its local subscriptions. Reuse of the logical name requires
-explicit `Destroy` followed by construction of a new generation. The lifecycle
-record intentionally survives expiry until Destroy so stale handles remain
-fenced.
+generation-scoped dedupe record share the stream's finite expiry, so retrying
+an ambiguous client result is safe even after max-length trimming removes the
+event itself. The returned ID remains the publication result while that retry
+record is retained.
+
+`AddOnce` accepts generations configured with an absolute deadline, a fixed
+TTL, or a sliding TTL. A handle may adopt the active retention by omitting
+retention options; an explicitly configured handle must match the complete
+retention contract. Ordinary `Add` calls never extend an absolute deadline or
+fixed TTL. On a sliding TTL, every ordinary or exact publication refreshes the
+stream, dedupe records, and recovery metadata together. At or after an absolute
+deadline, Add, AddOnce, and Snapshot return `ErrDeadlineElapsed`; Sink.Close
+treats expiry as terminal and still closes its local subscriptions. Reuse of
+that deadline-owned logical name requires explicit `Destroy` followed by
+construction of a new generation. The lifecycle record intentionally survives
+expiry until Destroy so stale handles remain fenced.
 
 `Stream.Snapshot` performs one generation-fenced Lua `XRANGE COUNT MaxLen+1`
 and returns currently retained immutable `SnapshotEvent` values in Redis ID

streaming/exact_publication.go

@@ -33,9 +33,10 @@ const (
 )
 
 var (
-	// addOnceScript verifies the stream generation and absolute deadline,
+	// addOnceScript verifies the stream generation and finite retention,
 	// resolves the generation-scoped idempotency record, and publishes exactly
-	// once. Redis TIME is authoritative for deadline admission.
+	// once. Redis TIME is authoritative for absolute-deadline admission and for
+	// aligning retry metadata with fixed or sliding stream TTLs.
 	addOnceScript = redis.NewScript(`
 local state = redis.call("HGET", KEYS[1], "state")
 local generation = redis.call("HGET", KEYS[1], "generation")
@@ -52,9 +53,6 @@ if ARGV[2] ~= "" then
     end
 else
     if not generation then
-        if ARGV[8] == "" then
-            return redis.error_reply("STREAMDEADLINEREQUIRED")
-        end
         generation = "1"
         physical = ARGV[4]
         recreate = true
@@ -73,12 +71,26 @@ end
 if retention and ARGV[18] == "1" and retention ~= ARGV[16] then
     return redis.error_reply("STREAMCONFIGMISMATCH")
 end
-if deadline then
+local effective_retention = retention or ARGV[16]
+local retention_mode = string.match(effective_retention, "|mode=([^|]+)|")
+local retention_value = tonumber(string.match(effective_retention, "|value=(%d+)|"))
+local retention_sliding = string.match(effective_retention, "|sliding=([^|]+)$")
+local ttl = 0
+local ttl_sliding = false
+if ttl_owned == "1" then
+    if retention_mode ~= "ttl" or not retention_value or retention_value <= 0 then
+        return redis.error_reply("STREAMCONFIGMISMATCH")
+    end
+    ttl = retention_value
+    ttl_sliding = retention_sliding == "true"
+elseif deadline then
     if ARGV[8] ~= "" and deadline ~= ARGV[8] then
         return redis.error_reply("STREAMDEADLINECONFLICT")
     end
-elseif ttl_owned == "1" then
-    return redis.error_reply("STREAMDEADLINECONFLICT")
+elseif retention_mode == "ttl" and retention_value and retention_value > 0 then
+    ttl = retention_value
+    ttl_sliding = retention_sliding == "true"
+    ttl_owned = "1"
 else
     if ARGV[8] == "" then
         return redis.error_reply("STREAMDEADLINEREQUIRED")
@@ -87,9 +99,18 @@ else
 end
 local now = redis.call("TIME")
 local now_ms = tonumber(now[1]) * 1000 + math.floor(tonumber(now[2]) / 1000)
-if now_ms >= tonumber(deadline) then
+if deadline and now_ms >= tonumber(deadline) then
     return redis.error_reply("DEADLINEELAPSED")
 end
+local expiry_deadline = deadline
+if ttl > 0 then
+    local remaining = redis.call("PTTL", physical)
+    if ttl_sliding or remaining < 0 then
+        expiry_deadline = now_ms + ttl
+    else
+        expiry_deadline = now_ms + remaining
+    end
+end
 
 local dedupe = ARGV[4] .. ":generation:" .. generation .. ":idempotency"
 local recovery = physical .. ":sink-recovery:" .. generation
@@ -116,15 +137,16 @@ if existing then
     if existing_identity ~= ARGV[17] then
         return redis.error_reply("IDEMPOTENCYCONFLICT")
     end
-    redis.call("PEXPIREAT", physical, deadline)
-    redis.call("PEXPIREAT", dedupe, deadline)
-    redis.call("PEXPIREAT", recovery, deadline)
+    redis.call("HSET", recovery, "=deadline", expiry_deadline)
+    redis.call("PEXPIREAT", physical, expiry_deadline)
+    redis.call("PEXPIREAT", dedupe, expiry_deadline)
+    redis.call("PEXPIREAT", recovery, expiry_deadline)
     local existing_resources = redis.call("SMEMBERS", resources_key)
     for _, resource in ipairs(existing_resources) do
-        redis.call("PEXPIREAT", resource, deadline)
+        redis.call("PEXPIREAT", resource, expiry_deadline)
     end
-    redis.call("PEXPIREAT", resources_key, deadline)
-    return {generation, physical, deadline, retention, 0, event_id}
+    redis.call("PEXPIREAT", resources_key, expiry_deadline)
+    return {generation, physical, deadline or "", effective_retention, 0, event_id}
 end
 
 if recreate then
@@ -135,14 +157,21 @@ if recreate then
         "generation", generation,
         "state", ARGV[1],
         ARGV[5], physical,
-        ARGV[6], deadline,
         ARGV[15], ARGV[16])
-    redis.call("HDEL", KEYS[1], ARGV[7])
+    if deadline then
+        redis.call("HSET", KEYS[1], ARGV[6], deadline)
+        redis.call("HDEL", KEYS[1], ARGV[7])
+    else
+        redis.call("HDEL", KEYS[1], ARGV[6])
+        redis.call("HSET", KEYS[1], ARGV[7], "1")
+    end
 elseif redis.call("HGET", KEYS[1], ARGV[5]) == false then
     redis.call("HSET", KEYS[1], ARGV[5], physical)
+end
+if deadline and redis.call("HGET", KEYS[1], ARGV[6]) == false then
     redis.call("HSET", KEYS[1], ARGV[6], deadline)
-elseif redis.call("HGET", KEYS[1], ARGV[6]) == false then
-    redis.call("HSET", KEYS[1], ARGV[6], deadline)
+elseif ttl > 0 then
+    redis.call("HSET", KEYS[1], ARGV[7], "1")
 end
 if not retention then
     redis.call("HSET", KEYS[1], ARGV[15], ARGV[16])
@@ -161,18 +190,18 @@ else
         "n", ARGV[11], "p", ARGV[12])
 end
 redis.call("HSET", dedupe, ARGV[10], event_id .. "\0" .. ARGV[17])
-redis.call("HSET", recovery, "=deadline", deadline)
+redis.call("HSET", recovery, "=deadline", expiry_deadline)
 redis.call("SADD", resources_key, dedupe, recovery)
 
-redis.call("PEXPIREAT", physical, deadline)
-redis.call("PEXPIREAT", dedupe, deadline)
-redis.call("PEXPIREAT", recovery, deadline)
+redis.call("PEXPIREAT", physical, expiry_deadline)
+redis.call("PEXPIREAT", dedupe, expiry_deadline)
+redis.call("PEXPIREAT", recovery, expiry_deadline)
 local resources = redis.call("SMEMBERS", resources_key)
 for _, resource in ipairs(resources) do
-    redis.call("PEXPIREAT", resource, deadline)
+    redis.call("PEXPIREAT", resource, expiry_deadline)
 end
-redis.call("PEXPIREAT", resources_key, deadline)
-return {generation, physical, deadline, retention or ARGV[16], 1, event_id}
+redis.call("PEXPIREAT", resources_key, expiry_deadline)
+return {generation, physical, deadline or "", effective_retention, 1, event_id}
 `)
 
 	// snapshotScript binds an unbound handle using the same zero-migration
@@ -222,9 +251,10 @@ return {1, generation, physical, deadline or "", retention, events}
 
 // AddOnce publishes one event for idempotencyKey in this stream generation.
 // The first call stores the event ID and exact length-delimited event identity
-// until the generation deadline. Exact retries return that ID; content changes
-// return ErrIdempotencyConflict. The active generation must be deadline-owned;
-// a handle with explicit retention options must match that immutable deadline.
+// until the generation expires. Exact retries return that ID; content changes
+// return ErrIdempotencyConflict. The active generation must have an absolute
+// deadline or a finite TTL, and explicit retention options must match its
+// immutable retention contract.
 func (s *Stream) AddOnce(
 	ctx context.Context,
 	idempotencyKey string,

streaming/exact_publication_test.go

@@ -102,6 +102,74 @@ func TestAddOnceConflictAndAmbiguousCommitRetry(t *testing.T) {
 	require.NoError(t, stream.Destroy(ctx))
 }
 
+func TestAddOnceSharesFixedAndSlidingTTLLifetimes(t *testing.T) {
+	for _, test := range []struct {
+		name       string
+		streamOpts []options.Stream
+		extends    bool
+	}{
+		{
+			name: "sliding ttl",
+			streamOpts: []options.Stream{
+				options.WithStreamMaxLen(100),
+				options.WithStreamSlidingTTL(2 * time.Second),
+			},
+			extends: true,
+		},
+		{
+			name: "fixed ttl",
+			streamOpts: []options.Stream{
+				options.WithStreamMaxLen(100),
+				options.WithStreamTTL(2 * time.Second),
+			},
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			rdb := ptesting.NewRedisClient(t)
+			defer ptesting.CleanupRedis(t, rdb, false, "")
+			ctx := ptesting.NewTestContext(t)
+			streamName := t.Name()
+
+			firstWriter, err := NewStream(streamName, rdb, test.streamOpts...)
+			require.NoError(t, err)
+			_, err = firstWriter.Add(ctx, "ordinary", []byte("before"))
+			require.NoError(t, err)
+
+			keyedWriter, err := NewStream(streamName, rdb, test.streamOpts...)
+			require.NoError(t, err)
+			eventID, err := keyedWriter.AddOnce(ctx, "stable-key", "keyed", []byte("payload"))
+			require.NoError(t, err)
+
+			dedupeKey := streamKey(streamName) + ":generation:1:idempotency"
+			before := rdb.PTTL(ctx, dedupeKey).Val()
+			require.Positive(t, before)
+			time.Sleep(100 * time.Millisecond)
+
+			laterWriter, err := NewStream(streamName, rdb, test.streamOpts...)
+			require.NoError(t, err)
+			_, err = laterWriter.Add(ctx, "ordinary", []byte("after"))
+			require.NoError(t, err)
+			after := rdb.PTTL(ctx, dedupeKey).Val()
+			require.Positive(t, after)
+			if test.extends {
+				require.Greater(t, after, before-50*time.Millisecond)
+			} else {
+				require.Less(t, after, before-75*time.Millisecond)
+			}
+
+			retryWriter, err := NewStream(streamName, rdb, test.streamOpts...)
+			require.NoError(t, err)
+			retryID, err := retryWriter.AddOnce(ctx, "stable-key", "keyed", []byte("payload"))
+			require.NoError(t, err)
+			require.Equal(t, eventID, retryID)
+			require.EqualValues(t, 3, rdb.XLen(ctx, streamKey(streamName)).Val())
+
+			_, err = retryWriter.AddOnce(ctx, "stable-key", "keyed", []byte("changed"))
+			require.ErrorIs(t, err, ErrIdempotencyConflict)
+		})
+	}
+}
+
 func TestAddOnceMetadataSurvivesMaxLenAndScriptFlush(t *testing.T) {
 	rdb := ptesting.NewRedisClient(t)
 	defer ptesting.CleanupRedis(t, rdb, false, "")

streaming/stream_lifecycle.go

@@ -277,7 +277,16 @@ if deadline then
     redis.call("PEXPIREAT", KEYS[4], deadline)
 elseif ttl > 0 then
     if ARGV[10] == "1" then
+        local now = redis.call("TIME")
+        local expiry_deadline = tonumber(now[1]) * 1000 + math.floor(tonumber(now[2]) / 1000) + ttl
+        redis.call("HSET", KEYS[3], "=deadline", expiry_deadline)
+        redis.call("SADD", KEYS[4], KEYS[3])
         redis.call("PEXPIRE", KEYS[2], ttl)
+        local resources = redis.call("SMEMBERS", KEYS[4])
+        for _, resource in ipairs(resources) do
+            redis.call("PEXPIRE", resource, ttl)
+        end
+        redis.call("PEXPIRE", KEYS[4], ttl)
     elseif redis.call("PTTL", KEYS[2]) == -1 then
         redis.call("PEXPIRE", KEYS[2], ttl)
     end