diff --git a/pkg/service/ingress.go b/pkg/service/ingress.go index 528309c68..a4abb641b 100644 --- a/pkg/service/ingress.go +++ b/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, diff --git a/pkg/service/ingress_test.go b/pkg/service/ingress_test.go index 86d29d2e2..f391e4482 100644 --- a/pkg/service/ingress_test.go +++ b/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()) + }) +} diff --git a/pkg/service/redisstore.go b/pkg/service/redisstore.go index 936f403a1..272f3e9ab 100644 --- a/pkg/service/redisstore.go +++ b/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 { diff --git a/pkg/service/redisstore_ingress_test.go b/pkg/service/redisstore_ingress_test.go new file mode 100644 index 000000000..4f02af2d0 --- /dev/null +++ b/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") +}