diff --git a/pkg/service/agent_dispatch_service.go b/pkg/service/agent_dispatch_service.go index b15bc55a3..f2dc581f5 100644 --- a/pkg/service/agent_dispatch_service.go +++ b/pkg/service/agent_dispatch_service.go @@ -84,7 +84,7 @@ func (ag *AgentDispatchService) CreateDispatch(ctx context.Context, req *livekit } dispatch := &livekit.AgentDispatch{ - Id: guid.New(guid.AgentDispatchPrefix), + Id: DeterministicID(guid.AgentDispatchPrefix, RequestID(ctx)), AgentName: req.AgentName, Room: req.Room, Metadata: req.Metadata, diff --git a/pkg/service/egress.go b/pkg/service/egress.go index 697aecbb4..2a72fd377 100644 --- a/pkg/service/egress.go +++ b/pkg/service/egress.go @@ -231,6 +231,15 @@ func (s *EgressService) startEgress(ctx context.Context, req *rpc.StartEgressReq return nil, ErrEgressNotConnected } + // Dedup SDK retries: when the client sent a request id, the egress id is + // derived from it, so an already-launched egress means this is a retry — + // return it instead of starting a second egress. + if s.io != nil && req.EgressId != "" && RequestID(ctx) != "" { + if existing, err := s.io.GetEgress(ctx, &rpc.GetEgressRequest{EgressId: req.EgressId}); err == nil && existing != nil { + return existing, nil + } + } + return s.launcher.StartEgress(ctx, req) } diff --git a/pkg/service/egressid.go b/pkg/service/egressid.go index a06d8c515..af0923013 100644 --- a/pkg/service/egressid.go +++ b/pkg/service/egressid.go @@ -16,7 +16,7 @@ func TwirpEgressID() *twirp.ServerHooks { RequestRouted: func(ctx context.Context) (context.Context, error) { // generate egressID for start egress methods for tracing egress failure before it reaches egress service. if isStartEgressMethod(ctx) { - egressID := guid.New(guid.EgressPrefix) + egressID := DeterministicID(guid.EgressPrefix, RequestID(ctx)) ctx = WithEgressID(ctx, egressID) AppendLogFields(ctx, "egressID", egressID) } diff --git a/pkg/service/ingress.go b/pkg/service/ingress.go index 2b82453e1..2648ae1dc 100644 --- a/pkg/service/ingress.go +++ b/pkg/service/ingress.go @@ -26,7 +26,6 @@ import ( "github.com/livekit/protocol/logger" "github.com/livekit/protocol/rpc" "github.com/livekit/protocol/utils" - "github.com/livekit/protocol/utils/guid" "github.com/livekit/psrpc" ) @@ -140,13 +139,14 @@ func (s *IngressService) CreateIngressWithUrl(ctx context.Context, urlStr string urlStr = urlObj.String() } + reqID := RequestID(ctx) var sk string if req.InputType != livekit.IngressInput_URL_INPUT { - sk = guid.New("") + sk = DeterministicID("", saltRequestID(reqID, "streamkey")) } info := &livekit.IngressInfo{ - IngressId: guid.New(utils.IngressPrefix), + IngressId: DeterministicID(utils.IngressPrefix, reqID), Name: req.Name, StreamKey: sk, Url: urlStr, diff --git a/pkg/service/requestid.go b/pkg/service/requestid.go new file mode 100644 index 000000000..91ce68952 --- /dev/null +++ b/pkg/service/requestid.go @@ -0,0 +1,64 @@ +// 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 + +import ( + "context" + + "github.com/livekit/protocol/utils/guid" +) + +// RequestIDAttribute is the participant attribute stamped with the request id +// for calls that join a room before dialing. A +// retry finds the existing participant with a matching value and skips the dial. +const RequestIDAttribute = "lk.request_id" + +type requestIDKey struct{} + +// WithRequestID stores the client idempotency id on the context and propagates +// it to downstream services via outgoing metadata. A no-op when id is empty. +func WithRequestID(ctx context.Context, id string) context.Context { + if id == "" { + return ctx + } + return context.WithValue(ctx, requestIDKey{}, id) +} + +func RequestID(ctx context.Context) string { + if id, ok := ctx.Value(requestIDKey{}).(string); ok && id != "" { + return id + } + return "" +} + +// saltRequestID derives a distinct deterministic seed from the request id for a +// secondary value (e.g. an ingress stream key) so it doesn't collide with the +// primary resource id derived from the same request id. Empty in, empty out, so +// DeterministicID still falls back to a random value when no request id is set. +func saltRequestID(requestID, salt string) string { + if requestID == "" { + return "" + } + return requestID + "/" + salt +} + +// DeterministicID derives a stable resource id from the request id — retries of the same logical call +// yield the same id and dedup at the store. With no request id it falls back to a random id. +func DeterministicID(prefix, requestID string) string { + if requestID == "" { + return guid.New(prefix) + } + return guid.Hash(prefix, []byte(requestID)) +} diff --git a/pkg/service/requestid_test.go b/pkg/service/requestid_test.go new file mode 100644 index 000000000..f1358fd4f --- /dev/null +++ b/pkg/service/requestid_test.go @@ -0,0 +1,87 @@ +// 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 + +import ( + "context" + "strings" + "testing" +) + +func TestDeterministicID(t *testing.T) { + const prefix = "EG_" + + // Same request id -> same id: this is what makes an SDK-retried create dedup. + a := DeterministicID(prefix, "req_123") + if b := DeterministicID(prefix, "req_123"); a != b { + t.Fatalf("not stable for same request id: %q != %q", a, b) + } + if !strings.HasPrefix(a, prefix) { + t.Fatalf("id %q missing prefix %q", a, prefix) + } + + // Different request ids -> different ids. + if c := DeterministicID(prefix, "req_456"); c == a { + t.Fatalf("different request ids produced the same id: %q", c) + } + + // Different prefixes -> different ids (so e.g. a room and identity derived + // from the same request id don't collide). + if d := DeterministicID("IN_", "req_123"); d == a { + t.Fatalf("different prefixes produced the same id: %q", d) + } + + // Empty request id -> random (each call differs), still prefixed. Preserves + // non-idempotent behavior when the client sent no request id. + r1 := DeterministicID(prefix, "") + r2 := DeterministicID(prefix, "") + if r1 == r2 { + t.Fatalf("empty request id should produce random ids, got equal: %q", r1) + } + if !strings.HasPrefix(r1, prefix) { + t.Fatalf("random id %q missing prefix %q", r1, prefix) + } +} + +func TestSaltRequestID(t *testing.T) { + // A salted seed derives a distinct-but-stable id (e.g. an ingress stream key + // vs its ingress id), so the two don't collide. + id := DeterministicID("IN_", "req_123") + sk := DeterministicID("", saltRequestID("req_123", "streamkey")) + if id == sk { + t.Fatalf("salted derivation collided with unsalted: %q", id) + } + if sk2 := DeterministicID("", saltRequestID("req_123", "streamkey")); sk2 != sk { + t.Fatalf("salted derivation not stable: %q != %q", sk, sk2) + } + // Empty in -> empty out, so DeterministicID still falls back to random. + if got := saltRequestID("", "streamkey"); got != "" { + t.Fatalf("saltRequestID(\"\", ...) = %q, want empty", got) + } +} + +func TestRequestID(t *testing.T) { + if got := RequestID(context.Background()); got != "" { + t.Fatalf("RequestID on bare context = %q, want empty", got) + } + ctx := WithRequestID(context.Background(), "req_abc") + if got := RequestID(ctx); got != "req_abc" { + t.Fatalf("RequestID = %q, want req_abc", got) + } + // An empty id is a no-op (the client sent none). + if got := RequestID(WithRequestID(context.Background(), "")); got != "" { + t.Fatalf("WithRequestID(\"\") should be a no-op, got %q", got) + } +} diff --git a/pkg/service/sip.go b/pkg/service/sip.go index b24206056..3ccc5e36c 100644 --- a/pkg/service/sip.go +++ b/pkg/service/sip.go @@ -139,6 +139,16 @@ func (s *SIPService) CreateSIPInboundTrunk(ctx context.Context, req *livekit.Cre // Keep ID empty still, so that validation can print "" instead of a non-existent ID in the error. + // Derive a deterministic id from the client request id so an SDK-retried + // create resolves to the same trunk. If it already exists, this is a retry: + // return it directly. + trunkID := DeterministicID(utils.SIPTrunkPrefix, RequestID(ctx)) + if RequestID(ctx) != "" { + if existing, err := s.store.LoadSIPInboundTrunk(ctx, trunkID); err == nil && existing != nil { + return existing, nil + } + } + // Validate all trunks including the new one first. it, err := ListSIPInboundTrunk(ctx, s.store, &livekit.ListSIPInboundTrunkRequest{ Numbers: req.GetTrunk().GetNumbers(), @@ -152,7 +162,7 @@ func (s *SIPService) CreateSIPInboundTrunk(ctx context.Context, req *livekit.Cre } // Now we can generate ID and store. - info.SipTrunkId = guid.New(utils.SIPTrunkPrefix) + info.SipTrunkId = trunkID if err := s.store.StoreSIPInboundTrunk(ctx, info); err != nil { return nil, err } @@ -177,7 +187,7 @@ func (s *SIPService) CreateSIPOutboundTrunk(ctx context.Context, req *livekit.Cr AppendLogFields(ctx, "trunk", logger.Proto(info)) // No additional validation needed for outbound. - info.SipTrunkId = guid.New(utils.SIPTrunkPrefix) + info.SipTrunkId = DeterministicID(utils.SIPTrunkPrefix, RequestID(ctx)) if err := s.store.StoreSIPOutboundTrunk(ctx, info); err != nil { return nil, err } @@ -447,6 +457,16 @@ func (s *SIPService) CreateSIPDispatchRule(ctx context.Context, req *livekit.Cre info := req.DispatchRuleInfo() info.SipDispatchRuleId = "" + // Derive a deterministic id from the client request id so an SDK-retried + // create resolves to the same rule. If it already exists, this is a retry: + // return it directly. + ruleID := DeterministicID(utils.SIPDispatchRulePrefix, RequestID(ctx)) + if RequestID(ctx) != "" { + if existing, err := s.store.LoadSIPDispatchRule(ctx, ruleID); err == nil && existing != nil { + return existing, nil + } + } + // Validate all rules including the new one first. it, err := ListSIPDispatchRule(ctx, s.store, &livekit.ListSIPDispatchRuleRequest{ TrunkIds: req.TrunkIds, @@ -460,7 +480,7 @@ func (s *SIPService) CreateSIPDispatchRule(ctx context.Context, req *livekit.Cre } // Now we can generate ID and store. - info.SipDispatchRuleId = guid.New(utils.SIPDispatchRulePrefix) + info.SipDispatchRuleId = ruleID if err := s.store.StoreSIPDispatchRule(ctx, info); err != nil { return nil, err }