streaming: re-establish the genesis lifecycle after Redis state loss

streaming/sink_recovery.go

@@ -193,7 +193,22 @@ func ensureConsumerGroup(ctx context.Context, stream *Stream, group, configuredS
 	if err := stream.ensureGeneration(ctx); err != nil {
 		return false, err
 	}
-	absent, err := ensureConsumerGroupScript.Run(
+	absent, err := runEnsureConsumerGroup(ctx, stream, group, configuredStart)
+	if err != nil {
+		lifecycleErr := stream.lifecycleError(err)
+		if !stream.reestablishLostGenesis(ctx, lifecycleErr) {
+			return false, ensureConsumerGroupError(stream, group, lifecycleErr)
+		}
+		if absent, err = runEnsureConsumerGroup(ctx, stream, group, configuredStart); err != nil {
+			return false, ensureConsumerGroupError(stream, group, stream.lifecycleError(err))
+		}
+	}
+	return absent == 1, nil
+}
+
+// runEnsureConsumerGroup executes the fenced group recovery script once.
+func runEnsureConsumerGroup(ctx context.Context, stream *Stream, group, configuredStart string) (int64, error) {
+	return ensureConsumerGroupScript.Run(
 		ctx,
 		stream.rdb,
 		[]string{stream.lifecycleKey, stream.key, recoveryCursorKey(stream)},
@@ -206,16 +221,17 @@ func ensureConsumerGroup(ctx context.Context, stream *Stream, group, configuredS
 		streamPhysicalKey,
 		streamDeadlineKey,
 	).Int64()
-	if err != nil {
-		return false, fmt.Errorf(
-			"failed to ensure Redis consumer group %q for stream %q generation %s: %w",
-			group,
-			stream.Name,
-			stream.generation,
-			stream.lifecycleError(err),
-		)
-	}
-	return absent == 1, nil
+}
+
+// ensureConsumerGroupError wraps group recovery failures with their identity.
+func ensureConsumerGroupError(stream *Stream, group string, err error) error {
+	return fmt.Errorf(
+		"failed to ensure Redis consumer group %q for stream %q generation %s: %w",
+		group,
+		stream.Name,
+		stream.generation,
+		err,
+	)
 }
 
 // XAck atomically acknowledges IDs and advances this generation's shared

streaming/sink_recovery_test.go

@@ -362,7 +362,6 @@ func TestSinkConsumerRotationRegistersEveryStreamOrRollsBack(t *testing.T) {
 	}
 }
 
-
 func TestSinkStreamMutationRollback(t *testing.T) {
 	rdb := ptesting.NewRedisClient(t)
 	defer ptesting.CleanupRedis(t, rdb, false, "")

streaming/stream_lifecycle.go

@@ -5,6 +5,7 @@ package streaming
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strconv"
 	"strings"
@@ -180,6 +181,34 @@ if deadline then
     end
 end
 return 1
+`)
+
+	// reestablishGenesisScript re-creates a lifecycle record that Redis lost
+	// entirely, exactly as the flat generation-one identity the caller still
+	// holds. It writes nothing when any record exists: a concurrent
+	// re-establishment converges on the same values and a destroy tombstone
+	// stays terminal, both decided by the retried operation's own fence.
+	//
+	// KEYS: [1]=lifecycle
+	// ARGV: [1]=active state [2]=physical field [3]=flat key [4]=config field
+	//	[5]=retention [6]=deadline field [7]=deadline ms [8]=ttl-owned field
+	//	[9]=ttl owned
+	reestablishGenesisScript = redis.NewScript(`
+if redis.call("EXISTS", KEYS[1]) == 1 then
+    return 0
+end
+redis.call("HSET", KEYS[1],
+    "state", ARGV[1],
+    "generation", "1",
+    ARGV[2], ARGV[3],
+    ARGV[4], ARGV[5])
+if ARGV[7] ~= "" then
+    redis.call("HSET", KEYS[1], ARGV[6], ARGV[7])
+end
+if ARGV[9] == "1" then
+    redis.call("HSET", KEYS[1], ARGV[8], "1")
+end
+return 1
 `)
 
 	// addStreamEventScript verifies the generation, appends one event, and
@@ -397,7 +426,16 @@ func (s *Stream) verifyGeneration(ctx context.Context) error {
 	if err := s.ensureGeneration(ctx); err != nil {
 		return err
 	}
-	err := verifyStreamScript.Run(
+	err := s.lifecycleError(s.runVerifyGeneration(ctx))
+	if err == nil || !s.reestablishLostGenesis(ctx, err) {
+		return err
+	}
+	return s.lifecycleError(s.runVerifyGeneration(ctx))
+}
+
+// runVerifyGeneration executes the exact-generation fence once.
+func (s *Stream) runVerifyGeneration(ctx context.Context) error {
+	return verifyStreamScript.Run(
 		ctx,
 		s.rdb,
 		[]string{s.lifecycleKey},
@@ -409,7 +447,52 @@ func (s *Stream) verifyGeneration(ctx context.Context) error {
 		streamConfigKey,
 		s.retention,
 	).Err()
-	return s.lifecycleError(err)
+}
+
+// reestablishLostGenesis re-asserts the lifecycle record for the flat
+// generation-one identity this bound handle already holds after Redis lost
+// the record entirely (state loss, not Destroy: a destroy tombstone still
+// exists and stays terminal). Only the genesis identity can prove continuity
+// — the flat key is generation one by construction, so every concurrent
+// re-establishment converges on the identical record — while later
+// generations cannot and remain destroyed. It reports whether the failed
+// operation is worth one retry.
+func (s *Stream) reestablishLostGenesis(ctx context.Context, opErr error) bool {
+	if !errors.Is(opErr, ErrStreamDestroyed) {
+		return false
+	}
+	if s.generation != "1" || s.key != streamKey(s.Name) {
+		return false
+	}
+	deadline := ""
+	if !s.deadline.IsZero() {
+		deadline = strconv.FormatInt(s.deadline.UnixMilli(), 10)
+	}
+	created, err := reestablishGenesisScript.Run(
+		ctx,
+		s.rdb,
+		[]string{s.lifecycleKey},
+		streamStateActive,
+		streamPhysicalKey,
+		s.key,
+		streamConfigKey,
+		s.retention,
+		streamDeadlineKey,
+		deadline,
+		streamTTLOwnedKey,
+		boolString(s.ttl > 0),
+	).Int64()
+	if err != nil {
+		s.logger.Error(fmt.Errorf("re-establish stream lifecycle after state loss: %w", err))
+		return false
+	}
+	if created == 1 {
+		s.logger.Info("re-established stream lifecycle after Redis state loss")
+	}
+	// A record now exists either way: re-created here, re-established by a
+	// concurrent genesis holder, or a terminal tombstone. The retried
+	// operation's fence decides.
+	return true
 }
 
 // verifyExistingGeneration loads without creating, then verifies the exact
@@ -438,8 +521,45 @@ func (s *Stream) addEvent(
 	if err != nil {
 		return "", err
 	}
-	topicPresent := topic != ""
-	result, err := addStreamEventScript.Run(
+	result, err := s.runAddEvent(ctx, name, payload, onlyIfExists, topic)
+	if err != nil {
+		addErr := s.lifecycleError(err)
+		if !s.reestablishLostGenesis(ctx, addErr) {
+			return "", addErr
+		}
+		if result, err = s.runAddEvent(ctx, name, payload, onlyIfExists, topic); err != nil {
+			return "", s.lifecycleError(err)
+		}
+	}
+	if len(result) == 0 {
+		return "", fmt.Errorf("add stream event script returned no status")
+	}
+	status, ok := result[0].(int64)
+	if !ok {
+		return "", fmt.Errorf("add stream event script returned invalid status %T", result[0])
+	}
+	if status == 0 {
+		return "", nil
+	}
+	if len(result) != 2 {
+		return "", fmt.Errorf("add stream event script returned %d values for successful add", len(result))
+	}
+	id, ok := result[1].(string)
+	if !ok {
+		return "", fmt.Errorf("add stream event script returned invalid event ID %T", result[1])
+	}
+	return id, nil
+}
+
+// runAddEvent executes the fenced publication script once.
+func (s *Stream) runAddEvent(
+	ctx context.Context,
+	name string,
+	payload []byte,
+	onlyIfExists bool,
+	topic string,
+) ([]any, error) {
+	return addStreamEventScript.Run(
 		ctx,
 		s.rdb,
 		[]string{
@@ -454,7 +574,7 @@ func (s *Stream) addEvent(
 		name,
 		payload,
 		boolString(onlyIfExists),
-		boolString(topicPresent),
+		boolString(topic != ""),
 		topic,
 		strconv.FormatInt(s.ttl.Milliseconds(), 10),
 		boolString(s.ttlSliding),
@@ -463,27 +583,6 @@ func (s *Stream) addEvent(
 		streamConfigKey,
 		s.retention,
 	).Slice()
-	if err != nil {
-		return "", s.lifecycleError(err)
-	}
-	if len(result) == 0 {
-		return "", fmt.Errorf("add stream event script returned no status")
-	}
-	status, ok := result[0].(int64)
-	if !ok {
-		return "", fmt.Errorf("add stream event script returned invalid status %T", result[0])
-	}
-	if status == 0 {
-		return "", nil
-	}
-	if len(result) != 2 {
-		return "", fmt.Errorf("add stream event script returned %d values for successful add", len(result))
-	}
-	id, ok := result[1].(string)
-	if !ok {
-		return "", fmt.Errorf("add stream event script returned invalid event ID %T", result[1])
-	}
-	return id, nil
 }
 
 // removeEvents atomically verifies this generation and deletes event IDs.

streaming/stream_lifecycle_test.go

@@ -0,0 +1,65 @@
+// Stream lifecycle tests prove generation identity survives Redis state loss
+// for the genesis handle while intentional destruction stays terminal.
+package streaming
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"goa.design/pulse/pulse"
+	"goa.design/pulse/streaming/options"
+	ptesting "goa.design/pulse/testing"
+)
+
+func TestGenesisLifecycleReestablishedAfterStateLoss(t *testing.T) {
+	rdb := ptesting.NewRedisClient(t)
+	defer ptesting.CleanupRedis(t, rdb, false, "")
+	ctx := ptesting.NewTestContext(t)
+	stream, err := NewStream(t.Name(), rdb, options.WithStreamLogger(pulse.ClueLogger(ctx)))
+	require.NoError(t, err)
+	sink, err := stream.NewSink(ctx, "sink", options.WithSinkBlockDuration(testBlockDuration))
+	require.NoError(t, err)
+	defer cleanupSink(t, ctx, stream, sink)
+	events := sink.Subscribe()
+
+	_, err = stream.Add(ctx, "before", []byte("before"))
+	require.NoError(t, err)
+	require.NoError(t, sink.Ack(ctx, receiveSinkEvent(t, events)))
+
+	// Redis loses every key: lifecycle, stream data, groups, and cursors.
+	// The bound genesis handles must re-assert their identity and resume.
+	require.NoError(t, rdb.FlushDB(ctx).Err())
+
+	// The sink's read loop re-establishes the genesis lifecycle and recreates
+	// its consumer group; the flushed recovery cursor is gone, so the group
+	// restarts at the tail and only later events are deliverable.
+	require.Eventually(t, func() bool {
+		return consumerGroupExists(ctx, rdb, stream.key, sink.Name)
+	}, max, delay, "sink should re-establish the lost genesis lifecycle and group")
+
+	afterID, err := stream.Add(ctx, "after", []byte("after"))
+	require.NoError(t, err, "publisher should re-establish the lost genesis lifecycle")
+	require.Equal(t, "1", stream.Generation())
+
+	recovered := receiveSinkEvent(t, events)
+	require.Equal(t, afterID, recovered.ID)
+	require.Equal(t, "after", recovered.EventName)
+	require.NoError(t, sink.Ack(ctx, recovered))
+}
+
+func TestDestroyedGenesisStaysTerminal(t *testing.T) {
+	rdb := ptesting.NewRedisClient(t)
+	defer ptesting.CleanupRedis(t, rdb, false, "")
+	ctx := ptesting.NewTestContext(t)
+	stream, err := NewStream(t.Name(), rdb, options.WithStreamLogger(pulse.ClueLogger(ctx)))
+	require.NoError(t, err)
+	require.NoError(t, stream.Open(ctx))
+	require.NoError(t, stream.Destroy(ctx))
+
+	// A destroy tombstone is intentional termination, not state loss: the
+	// bound handle must not resurrect the generation.
+	_, err = stream.Add(ctx, "after", []byte("after"))
+	require.ErrorIs(t, err, ErrStreamDestroyed)
+	require.ErrorIs(t, stream.verifyGeneration(ctx), ErrStreamDestroyed)
+}