Return an error when the ingress store exhausts its retries; tolerate missing state in UpdateIngress/DeleteIngress (#5049)

- pkg/service/redisstore.go: after the retry loop, return an error wrapping redis.TxFailedErr instead of nil. CreateIngress now fails instead of returning an ingress that was never written, StoreIngress stops before creating the orphan state key, and a dropped UpdateIngressState reaches the caller's existing error log. 

- pkg/service/ingress.go: the Fix nil pointer dereference in IOInfoService.UpdateIngressState on concurrent DeleteIngress #4806 guard applied to the two Twirp handlers, treating a nil State as empty. UpdateIngress then applies and stores the request; DeleteIngress deletes without calling the ingress service.

- pkg/service/redisstore.go: UpdateIngress now fails with ErrIngressNotFound when the ingress no longer exists, checked inside the WATCH on the ingress hash so a later delete fails the EXEC. Without it an update racing a  DeleteIngress recreated the deleted ingress and its stream key; this was possible before this PR whenever the delete landed after LoadIngress, and the guard in 2 widened the window to the gap between its two reads.
The guard stays at the call sites to match #4806 and keep API responses unchanged.

pkg/service/ingress.go

@@ -309,6 +309,12 @@ func (s *IngressService) UpdateIngress(ctx context.Context, req *livekit.UpdateI
 		logger.Errorw("could not load ingress info", err)
 		return nil, err
 	}
+	if info.State == nil {
+		// LoadIngress reads the info and the state from separate keys, so the state
+		// can be missing: a concurrent DeleteIngress, or a create whose state write
+		// failed. Treat it as an empty state, as IOInfoService.UpdateIngressState does.
+		info.State = &livekit.IngressState{}
+	}
 
 	if !info.Reusable {
 		logger.Infow("ingress update attempted on non reusable ingress", "ingressID", info.IngressId)
@@ -397,6 +403,10 @@ func (s *IngressService) DeleteIngress(ctx context.Context, req *livekit.DeleteI
 	if err != nil {
 		return nil, err
 	}
+	if info.State == nil {
+		// The state key can be missing, see UpdateIngress.
+		info.State = &livekit.IngressState{}
+	}
 
 	switch info.State.Status {
 	case livekit.IngressState_ENDPOINT_BUFFERING,

pkg/service/ingress_test.go

@@ -22,10 +22,12 @@ import (
 
 	"github.com/livekit/protocol/auth"
 	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/rpc"
 
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/service"
 	"github.com/livekit/livekit-server/pkg/service/servicefakes"
+	"github.com/livekit/livekit-server/pkg/telemetry/telemetryfakes"
 )
 
 func TestCreateURLPullIngressScheme(t *testing.T) {
@@ -94,3 +96,62 @@ func TestCreateURLPullIngressScheme(t *testing.T) {
 		require.Contains(t, err.Error(), "invalid url scheme rtsp")
 	})
 }
+
+// unusedIngressClient is a non-nil rpc.IngressClient for handlers that must not
+// reach the ingress service: any call on it panics on the nil embedded interfaces.
+type unusedIngressClient struct {
+	rpc.IngressInternalClient
+	rpc.IngressHandlerClient
+}
+
+func (unusedIngressClient) Close() {}
+
+// LoadIngress reads the ingress info and its state from separate keys and
+// returns the info with a nil State when the state key is missing (a concurrent
+// DeleteIngress, or a create whose state write failed). The API handlers must
+// treat that as an empty state instead of panicking, like
+// IOInfoService.UpdateIngressState does.
+func TestIngressAPIWithNilState(t *testing.T) {
+	newService := func() (*service.IngressService, *servicefakes.FakeIngressStore) {
+		store := &servicefakes.FakeIngressStore{}
+		store.LoadIngressReturns(&livekit.IngressInfo{
+			IngressId:           "IN_test",
+			StreamKey:           "sk",
+			RoomName:            "testroom",
+			ParticipantIdentity: "ingress",
+			Reusable:            true,
+		}, nil)
+
+		svc := service.NewIngressServiceWithIngressLauncher(
+			&config.IngressConfig{},
+			"nodeID",
+			nil,
+			unusedIngressClient{},
+			store,
+			&servicefakes.FakeIOClient{},
+			&telemetryfakes.FakeTelemetryService{},
+			&servicefakes.FakeIngressLauncher{},
+		)
+		return svc, store
+	}
+
+	adminCtx := service.WithGrants(context.Background(), &auth.ClaimGrants{Video: &auth.VideoGrant{IngressAdmin: true}}, "")
+
+	t.Run("UpdateIngress", func(t *testing.T) {
+		svc, store := newService()
+
+		info, err := svc.UpdateIngress(adminCtx, &livekit.UpdateIngressRequest{IngressId: "IN_test", Name: "renamed"})
+		require.NoError(t, err)
+		require.Equal(t, "renamed", info.Name)
+		require.Equal(t, 1, store.UpdateIngressCallCount())
+	})
+
+	t.Run("DeleteIngress", func(t *testing.T) {
+		svc, store := newService()
+
+		info, err := svc.DeleteIngress(adminCtx, &livekit.DeleteIngressRequest{IngressId: "IN_test"})
+		require.NoError(t, err)
+		require.Equal(t, livekit.IngressState_ENDPOINT_INACTIVE, info.State.Status)
+		require.Equal(t, 1, store.DeleteIngressCallCount())
+	})
+}

pkg/service/redisstore.go

@@ -531,7 +531,7 @@ func parseEgressEnded(value string) (roomName string, endedAt int64, err error)
 }
 
 func (s *RedisStore) StoreIngress(ctx context.Context, info *livekit.IngressInfo) error {
-	err := s.storeIngress(ctx, info)
+	err := s.storeIngress(ctx, info, false)
 	if err != nil {
 		return err
 	}
@@ -539,7 +539,9 @@ func (s *RedisStore) StoreIngress(ctx context.Context, info *livekit.IngressInfo
 	return s.storeIngressState(ctx, info.IngressId, nil)
 }
 
-func (s *RedisStore) storeIngress(_ context.Context, info *livekit.IngressInfo) error {
+// storeIngress writes the ingress info. With mustExist it fails with ErrIngressNotFound
+// instead of recreating an ingress that was deleted since the caller loaded it.
+func (s *RedisStore) storeIngress(_ context.Context, info *livekit.IngressInfo, mustExist bool) error {
 	if info.IngressId == "" {
 		return errors.New("Missing IngressId")
 	}
@@ -563,6 +565,11 @@ func (s *RedisStore) storeIngress(_ context.Context, info *livekit.IngressInfo)
 		oldInfo, err := s.loadIngress(tx, info.IngressId)
 		switch err {
 		case ErrIngressNotFound:
+			if mustExist {
+				// Checked inside the WATCH: a delete landing after this point
+				// fails the EXEC instead of being overwritten.
+				return err
+			}
 			// Ingress doesn't exist yet
 		case nil:
 			oldRoom = oldInfo.RoomName
@@ -613,7 +620,7 @@ func (s *RedisStore) storeIngress(_ context.Context, info *livekit.IngressInfo)
 		}
 	}
 
-	return nil
+	return psrpc.NewError(psrpc.Unavailable, errors.Wrapf(redis.TxFailedErr, "could not store ingress %s after %d attempts", info.IngressId, maxRetries))
 }
 
 func (s *RedisStore) storeIngressState(_ context.Context, ingressId string, state *livekit.IngressState) error {
@@ -688,7 +695,7 @@ func (s *RedisStore) storeIngressState(_ context.Context, ingressId string, stat
 		}
 	}
 
-	return nil
+	return psrpc.NewError(psrpc.Unavailable, errors.Wrapf(redis.TxFailedErr, "could not store ingress state %s after %d attempts", ingressId, maxRetries))
 }
 
 func (s *RedisStore) loadIngress(c redis.Cmdable, ingressId string) (*livekit.IngressInfo, error) {
@@ -828,7 +835,7 @@ func (s *RedisStore) ListIngress(_ context.Context, roomName livekit.RoomName) (
 }
 
 func (s *RedisStore) UpdateIngress(ctx context.Context, info *livekit.IngressInfo) error {
-	return s.storeIngress(ctx, info)
+	return s.storeIngress(ctx, info, true)
 }
 
 func (s *RedisStore) UpdateIngressState(ctx context.Context, ingressId string, state *livekit.IngressState) error {

pkg/service/redisstore_ingress_test.go

@@ -0,0 +1,164 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package service_test
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/redis/go-redis/v9"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/livekit/protocol/livekit"
+
+	"github.com/livekit/livekit-server/pkg/service"
+)
+
+// watchRival is a go-redis hook that writes to every key the hooked client
+// WATCHes, from another connection, right after the WATCH is sent. Every
+// optimistic WATCH/MULTI/EXEC transaction the hooked client attempts therefore
+// fails with redis.TxFailedErr, the same way a busy cluster can defeat all of
+// RedisStore's retries.
+type watchRival struct {
+	write func(ctx context.Context, key string) error
+}
+
+func (r *watchRival) DialHook(next redis.DialHook) redis.DialHook { return next }
+
+func (r *watchRival) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
+	return next
+}
+
+func (r *watchRival) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
+	return func(ctx context.Context, cmd redis.Cmder) error {
+		if err := next(ctx, cmd); err != nil {
+			return err
+		}
+		if strings.EqualFold(cmd.Name(), "watch") {
+			for _, key := range cmd.Args()[1:] {
+				if err := r.write(ctx, key.(string)); err != nil {
+					return err
+				}
+			}
+		}
+		return nil
+	}
+}
+
+// RedisStore.storeIngress and storeIngressState retry their optimistic
+// transaction a fixed number of times on redis.TxFailedErr. When every attempt
+// loses the lock they must report it: returning nil makes the caller believe
+// the write landed, so CreateIngress answers with an ingress that does not
+// exist and leaves an orphan state key, and UpdateIngressState silently drops
+// a state transition.
+func TestIngressStoreReportsLostOptimisticLock(t *testing.T) {
+	ctx := context.Background()
+	rc := redisClient(t)
+	rs := service.NewRedisStore(rc)
+
+	// Every transaction attempted through this store loses its optimistic lock.
+	rival := &watchRival{}
+	contendedClient := redis.NewClient(&redis.Options{Addr: rc.Options().Addr})
+	contendedClient.AddHook(rival)
+	t.Cleanup(func() { _ = contendedClient.Close() })
+	contended := service.NewRedisStore(contendedClient)
+
+	info := &livekit.IngressInfo{
+		IngressId: "IN_lost_lock",
+		StreamKey: "sk_lost_lock",
+		RoomName:  "room_lost_lock",
+	}
+	t.Cleanup(func() { _ = rs.DeleteIngress(ctx, info) })
+
+	t.Run("StoreIngress", func(t *testing.T) {
+		// Any other ingress written to the hash invalidates the WATCH on IngressKey.
+		other, err := proto.Marshal(&livekit.IngressInfo{IngressId: "IN_rival"})
+		require.NoError(t, err)
+		t.Cleanup(func() { _ = rc.HDel(ctx, service.IngressKey, "IN_rival").Err() })
+		rival.write = func(ctx context.Context, key string) error {
+			if key != service.IngressKey {
+				return nil
+			}
+			return rc.HSet(ctx, key, "IN_rival", other).Err()
+		}
+
+		err = contended.StoreIngress(ctx, info)
+
+		_, loadErr := rs.LoadIngress(ctx, info.IngressId)
+		require.ErrorIs(t, loadErr, service.ErrIngressNotFound, "no attempt could have written the ingress")
+		require.ErrorIs(t, err, redis.TxFailedErr, "StoreIngress returned nil although the ingress was never written")
+		exists, err := rc.Exists(ctx, service.IngressStatePrefix+info.IngressId).Result()
+		require.NoError(t, err)
+		require.Zero(t, exists, "a state key was left behind for an ingress that was never written")
+	})
+
+	t.Run("UpdateIngressState", func(t *testing.T) {
+		require.NoError(t, rs.StoreIngress(ctx, info))
+
+		// An empty payload is a valid zero IngressState: writing it invalidates the
+		// WATCH on the state key without tripping the out-of-date checks.
+		stateKey := service.IngressStatePrefix + info.IngressId
+		rival.write = func(ctx context.Context, key string) error {
+			if key != stateKey {
+				return nil
+			}
+			return rc.Set(ctx, key, "", 0).Err()
+		}
+
+		err := contended.UpdateIngressState(ctx, info.IngressId, &livekit.IngressState{
+			Status: livekit.IngressState_ENDPOINT_PUBLISHING,
+		})
+
+		loaded, loadErr := rs.LoadIngress(ctx, info.IngressId)
+		require.NoError(t, loadErr)
+		require.Equal(t, livekit.IngressState_ENDPOINT_INACTIVE, loaded.State.Status, "no attempt could have written the state")
+		require.ErrorIs(t, err, redis.TxFailedErr, "UpdateIngressState returned nil although the state was never written")
+	})
+}
+
+// RedisStore.UpdateIngress must not recreate an ingress that was deleted after the
+// caller loaded it. The existence check runs inside the WATCH on the ingress hash,
+// so a delete landing after the check fails the transaction instead.
+func TestUpdateIngressDoesNotRecreateDeletedIngress(t *testing.T) {
+	ctx := context.Background()
+	rc := redisClient(t)
+	rs := service.NewRedisStore(rc)
+
+	info := &livekit.IngressInfo{
+		IngressId: "IN_deleted",
+		StreamKey: "sk_deleted",
+		RoomName:  "room_deleted",
+	}
+	require.NoError(t, rs.StoreIngress(ctx, info))
+	t.Cleanup(func() { _ = rs.DeleteIngress(ctx, info) })
+
+	// what an API handler holds when a concurrent DeleteIngress lands
+	loaded, err := rs.LoadIngress(ctx, info.IngressId)
+	require.NoError(t, err)
+	require.NoError(t, rs.DeleteIngress(ctx, info))
+
+	loaded.Name = "renamed"
+	require.ErrorIs(t, rs.UpdateIngress(ctx, loaded), service.ErrIngressNotFound)
+
+	_, err = rs.LoadIngress(ctx, info.IngressId)
+	require.ErrorIs(t, err, service.ErrIngressNotFound, "the deleted ingress must not reappear")
+	_, err = rs.LoadIngressFromStreamKey(ctx, info.StreamKey)
+	require.ErrorIs(t, err, service.ErrIngressNotFound, "nor its stream key")
+	members, err := rc.SMembers(ctx, service.RoomIngressPrefix+info.RoomName).Result()
+	require.NoError(t, err)
+	require.Empty(t, members, "nor its room membership")
+}