Use request id to make api idempotence on sdk retry

Derive resource id from request id

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,

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)
 }
 

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)
 			}

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,

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))
+}

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)
+	}
+}

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 "<new>" 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
 	}