diff --git a/streaming/README.md b/streaming/README.md index 5cfd174..3a1567f 100644 --- a/streaming/README.md +++ b/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 diff --git a/streaming/exact_publication.go b/streaming/exact_publication.go index e837cb0..8724374 100644 --- a/streaming/exact_publication.go +++ b/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, diff --git a/streaming/exact_publication_test.go b/streaming/exact_publication_test.go index 6ef87ea..727cc4e 100644 --- a/streaming/exact_publication_test.go +++ b/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, "") diff --git a/streaming/stream_lifecycle.go b/streaming/stream_lifecycle.go index ab851bb..61b4bb1 100644 --- a/streaming/stream_lifecycle.go +++ b/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