From fb51d648700bb44bb03ab1fa1c8a93191e886956 Mon Sep 17 00:00:00 2001 From: "Raphael (manual office deploy after cloud-state fix)" Date: Sat, 25 Jul 2026 10:20:00 -0700 Subject: [PATCH] streaming: re-establish the genesis lifecycle after Redis state loss MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A bound generation-one handle on the flat key observing a wholly absent lifecycle record is state loss, not destruction: intentional Destroy leaves a tombstone. The genesis identity is the one incarnation that can prove continuity — the flat key is generation one by construction, so every concurrent re-establishment converges on the identical record — so verification, publication, and consumer-group recovery now atomically re-assert it and retry once. Later generations still cannot prove continuity and remain destroyed, as do tombstoned generations. This restores no-restart recovery from full Redis state loss for long-running publishers and sinks. --- streaming/sink_recovery.go | 38 +++++--- streaming/sink_recovery_test.go | 1 - streaming/stream_lifecycle.go | 151 ++++++++++++++++++++++++----- streaming/stream_lifecycle_test.go | 65 +++++++++++++ 4 files changed, 217 insertions(+), 38 deletions(-) create mode 100644 streaming/stream_lifecycle_test.go diff --git a/streaming/sink_recovery.go b/streaming/sink_recovery.go index 03d1b3d..5e75fd4 100644 --- a/streaming/sink_recovery.go +++ b/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 diff --git a/streaming/sink_recovery_test.go b/streaming/sink_recovery_test.go index 3b08446..5a7ccc7 100644 --- a/streaming/sink_recovery_test.go +++ b/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, "") diff --git a/streaming/stream_lifecycle.go b/streaming/stream_lifecycle.go index 43fc6fc..ab851bb 100644 --- a/streaming/stream_lifecycle.go +++ b/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. diff --git a/streaming/stream_lifecycle_test.go b/streaming/stream_lifecycle_test.go new file mode 100644 index 0000000..f4ec71e --- /dev/null +++ b/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) +}