streaming: let a new handle snapshot a stream that was recreated

streaming/exact_publication.go

@@ -206,7 +206,10 @@ return {generation, physical, deadline or "", effective_retention, 1, event_id}
 
 	// snapshotScript binds an unbound handle using the same zero-migration
 	// lifecycle contract, verifies a bound handle, and returns one XRANGE
-	// result at the operation's single Redis linearization point.
+	// result at the operation's single Redis linearization point. Only a bound
+	// handle (non-empty generation) is checked against a generation and
+	// physical key: an unbound handle carries the generation-1 key, which a
+	// stream destroyed and created again no longer uses.
 	snapshotScript = redis.NewScript(`
 local state = redis.call("HGET", KEYS[1], "state")
 local generation = redis.call("HGET", KEYS[1], "generation")
@@ -215,8 +218,7 @@ if not generation then
 end
 local physical = redis.call("HGET", KEYS[1], ARGV[4])
 local deadline = redis.call("HGET", KEYS[1], ARGV[5])
-if state ~= ARGV[1] or (ARGV[2] ~= "" and generation ~= ARGV[2])
-or (ARGV[3] ~= "" and physical ~= ARGV[3]) then
+if state ~= ARGV[1] or (ARGV[2] ~= "" and (generation ~= ARGV[2] or physical ~= ARGV[3])) then
     return redis.error_reply("STREAMDESTROYED")
 end
 local retention = redis.call("HGET", KEYS[1], ARGV[6])

streaming/exact_publication_test.go

@@ -495,6 +495,39 @@ func TestExactPublicationOptionValidation(t *testing.T) {
 	require.NoError(t, ttlStream.Destroy(ctx))
 }
 
+// TestSnapshotReadsRecreatedStreamFromNewHandle verifies that a new handle
+// reads a stream that was destroyed and created again, whose events live
+// under a generation-scoped key, and that a handle still bound to the
+// destroyed generation is rejected.
+func TestSnapshotReadsRecreatedStreamFromNewHandle(t *testing.T) {
+	rdb := ptesting.NewRedisClient(t)
+	defer ptesting.CleanupRedis(t, rdb, false, "")
+	ctx := ptesting.NewTestContext(t)
+	first, err := NewStream(t.Name(), rdb)
+	require.NoError(t, err)
+	_, err = first.Add(ctx, "old", []byte("one"))
+	require.NoError(t, err)
+	require.NoError(t, first.Destroy(ctx))
+
+	second, err := NewStream(t.Name(), rdb)
+	require.NoError(t, err)
+	_, err = second.Add(ctx, "new", []byte("two"))
+	require.NoError(t, err)
+	require.Equal(t, "2", second.Generation())
+
+	fresh, err := NewStream(t.Name(), rdb)
+	require.NoError(t, err)
+	events, err := fresh.Snapshot(ctx)
+	require.NoError(t, err)
+	require.Len(t, events, 1)
+	require.Equal(t, "new", events[0].EventName())
+	require.Equal(t, "2", fresh.Generation())
+
+	_, err = first.Snapshot(ctx)
+	require.ErrorIs(t, err, ErrStreamDestroyed)
+	require.NoError(t, second.Destroy(ctx))
+}
+
 func TestSnapshotDoesNotInitializeAbsentStream(t *testing.T) {
 	rdb := ptesting.NewRedisClient(t)
 	defer ptesting.CleanupRedis(t, rdb, false, "")