From fad2cc4afeb2f812d2f0ffa211f03d7d5a209b02 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Wed, 12 Aug 2026 09:28:39 +0800 Subject: [PATCH] Use request id to make api idempotence on sdk retry (#4694) * Use request id to make api idempotence on sdk retry Derive resource id from request id --- go.mod | 2 +- go.sum | 4 +- pkg/service/agent_dispatch_service.go | 2 +- pkg/service/clients.go | 2 +- pkg/service/ingress.go | 1 + pkg/service/ioservice_ingress.go | 7 ++- pkg/service/requestid.go | 53 +++++++++++++++++ pkg/service/requestid_test.go | 70 +++++++++++++++++++++++ pkg/service/servicefakes/fake_ioclient.go | 20 +++---- 9 files changed, 143 insertions(+), 18 deletions(-) create mode 100644 pkg/service/requestid.go create mode 100644 pkg/service/requestid_test.go diff --git a/go.mod b/go.mod index d2f1ac99f..093625e43 100644 --- a/go.mod +++ b/go.mod @@ -21,7 +21,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0 - github.com/livekit/protocol v1.50.5-0.20260805073019-6c35920ed51c + github.com/livekit/protocol v1.50.5-0.20260811022948-6ad5660c9f8f github.com/livekit/psrpc v0.7.3 github.com/mackerelio/go-osstat v0.2.8 github.com/magefile/mage v1.17.2 diff --git a/go.sum b/go.sum index b11dbfce9..19ddb1945 100644 --- a/go.sum +++ b/go.sum @@ -160,8 +160,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0 h1:XHNNzebIKZRkLimla/hFGrAIX5EMWHctrgt3hLw7s+I= github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU= -github.com/livekit/protocol v1.50.5-0.20260805073019-6c35920ed51c h1:s/DmilyBXEqM02LvZMnaAIK8IfxhTXjMkSNSuWwe8+c= -github.com/livekit/protocol v1.50.5-0.20260805073019-6c35920ed51c/go.mod h1:DeGUkqNGJhLz16DOP3MfKrJtYn/1PiYfrQMsJb3rzDs= +github.com/livekit/protocol v1.50.5-0.20260811022948-6ad5660c9f8f h1:1Tod/QfZbX2Y6qvmVm392uYvddWaY8bz4hbnSgpEcMc= +github.com/livekit/protocol v1.50.5-0.20260811022948-6ad5660c9f8f/go.mod h1:edX/q09IZsPPR6SRK1xqA2lcdskpXIZfVvZAPDVV4yw= github.com/livekit/psrpc v0.7.3 h1:bekuZt/ZQzg8+/M8G6G5jq7bvV9fAKdPHSOZeTwrIIc= github.com/livekit/psrpc v0.7.3/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw= github.com/mackerelio/go-osstat v0.2.8 h1:I2duicTaCGWoM53XwAwA9OIe1inu0xnVs8/pqOWWVr4= 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/clients.go b/pkg/service/clients.go index d8fb33789..406bf5e74 100644 --- a/pkg/service/clients.go +++ b/pkg/service/clients.go @@ -28,6 +28,6 @@ type IOClient interface { CreateEgress(ctx context.Context, info *livekit.EgressInfo) (*emptypb.Empty, error) GetEgress(ctx context.Context, req *rpc.GetEgressRequest) (*livekit.EgressInfo, error) ListEgress(ctx context.Context, req *livekit.ListEgressRequest) (*livekit.ListEgressResponse, error) - CreateIngress(ctx context.Context, req *livekit.IngressInfo) (*emptypb.Empty, error) + CreateIngress(ctx context.Context, req *livekit.IngressInfo) (*rpc.CreateIngressResponse, error) UpdateIngressState(ctx context.Context, req *rpc.UpdateIngressStateRequest) (*emptypb.Empty, error) } diff --git a/pkg/service/ingress.go b/pkg/service/ingress.go index 2b82453e1..4036b0e1e 100644 --- a/pkg/service/ingress.go +++ b/pkg/service/ingress.go @@ -192,6 +192,7 @@ func (s *IngressService) CreateIngressWithUrl(ctx context.Context, urlStr string } // The Ingress instance will create the ingress object when handling the URL pull ingress } else { + // TODO-jie: ingress retry idempotency: generate ingress key by request-id, and return the ingress object from CreateIngress. _, err = s.io.CreateIngress(ctx, info) switch err { case nil: diff --git a/pkg/service/ioservice_ingress.go b/pkg/service/ioservice_ingress.go index b802c888f..c1fe59555 100644 --- a/pkg/service/ioservice_ingress.go +++ b/pkg/service/ioservice_ingress.go @@ -18,13 +18,14 @@ import ( "context" "errors" + "google.golang.org/protobuf/types/known/emptypb" + "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/rpc" - "google.golang.org/protobuf/types/known/emptypb" ) -func (s *IOInfoService) CreateIngress(ctx context.Context, info *livekit.IngressInfo) (*emptypb.Empty, error) { +func (s *IOInfoService) CreateIngress(ctx context.Context, info *livekit.IngressInfo) (*rpc.CreateIngressResponse, error) { if s.is == nil { return nil, ErrIngressNotConnected } @@ -36,7 +37,7 @@ func (s *IOInfoService) CreateIngress(ctx context.Context, info *livekit.Ingress s.telemetry.IngressCreated(ctx, info) - return &emptypb.Empty{}, nil + return &rpc.CreateIngressResponse{Info: info}, nil } func (s *IOInfoService) GetIngressInfo(ctx context.Context, req *rpc.GetIngressInfoRequest) (*rpc.GetIngressInfoResponse, error) { diff --git a/pkg/service/requestid.go b/pkg/service/requestid.go new file mode 100644 index 000000000..bde50b1f6 --- /dev/null +++ b/pkg/service/requestid.go @@ -0,0 +1,53 @@ +// 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 "" +} + +// 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..f3ed1398d --- /dev/null +++ b/pkg/service/requestid_test.go @@ -0,0 +1,70 @@ +// 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 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/servicefakes/fake_ioclient.go b/pkg/service/servicefakes/fake_ioclient.go index 0cdda4127..6a04a2c29 100644 --- a/pkg/service/servicefakes/fake_ioclient.go +++ b/pkg/service/servicefakes/fake_ioclient.go @@ -26,18 +26,18 @@ type FakeIOClient struct { result1 *emptypb.Empty result2 error } - CreateIngressStub func(context.Context, *livekit.IngressInfo) (*emptypb.Empty, error) + CreateIngressStub func(context.Context, *livekit.IngressInfo) (*rpc.CreateIngressResponse, error) createIngressMutex sync.RWMutex createIngressArgsForCall []struct { arg1 context.Context arg2 *livekit.IngressInfo } createIngressReturns struct { - result1 *emptypb.Empty + result1 *rpc.CreateIngressResponse result2 error } createIngressReturnsOnCall map[int]struct { - result1 *emptypb.Empty + result1 *rpc.CreateIngressResponse result2 error } GetEgressStub func(context.Context, *rpc.GetEgressRequest) (*livekit.EgressInfo, error) @@ -151,7 +151,7 @@ func (fake *FakeIOClient) CreateEgressReturnsOnCall(i int, result1 *emptypb.Empt }{result1, result2} } -func (fake *FakeIOClient) CreateIngress(arg1 context.Context, arg2 *livekit.IngressInfo) (*emptypb.Empty, error) { +func (fake *FakeIOClient) CreateIngress(arg1 context.Context, arg2 *livekit.IngressInfo) (*rpc.CreateIngressResponse, error) { fake.createIngressMutex.Lock() ret, specificReturn := fake.createIngressReturnsOnCall[len(fake.createIngressArgsForCall)] fake.createIngressArgsForCall = append(fake.createIngressArgsForCall, struct { @@ -177,7 +177,7 @@ func (fake *FakeIOClient) CreateIngressCallCount() int { return len(fake.createIngressArgsForCall) } -func (fake *FakeIOClient) CreateIngressCalls(stub func(context.Context, *livekit.IngressInfo) (*emptypb.Empty, error)) { +func (fake *FakeIOClient) CreateIngressCalls(stub func(context.Context, *livekit.IngressInfo) (*rpc.CreateIngressResponse, error)) { fake.createIngressMutex.Lock() defer fake.createIngressMutex.Unlock() fake.CreateIngressStub = stub @@ -190,28 +190,28 @@ func (fake *FakeIOClient) CreateIngressArgsForCall(i int) (context.Context, *liv return argsForCall.arg1, argsForCall.arg2 } -func (fake *FakeIOClient) CreateIngressReturns(result1 *emptypb.Empty, result2 error) { +func (fake *FakeIOClient) CreateIngressReturns(result1 *rpc.CreateIngressResponse, result2 error) { fake.createIngressMutex.Lock() defer fake.createIngressMutex.Unlock() fake.CreateIngressStub = nil fake.createIngressReturns = struct { - result1 *emptypb.Empty + result1 *rpc.CreateIngressResponse result2 error }{result1, result2} } -func (fake *FakeIOClient) CreateIngressReturnsOnCall(i int, result1 *emptypb.Empty, result2 error) { +func (fake *FakeIOClient) CreateIngressReturnsOnCall(i int, result1 *rpc.CreateIngressResponse, result2 error) { fake.createIngressMutex.Lock() defer fake.createIngressMutex.Unlock() fake.CreateIngressStub = nil if fake.createIngressReturnsOnCall == nil { fake.createIngressReturnsOnCall = make(map[int]struct { - result1 *emptypb.Empty + result1 *rpc.CreateIngressResponse result2 error }) } fake.createIngressReturnsOnCall[i] = struct { - result1 *emptypb.Empty + result1 *rpc.CreateIngressResponse result2 error }{result1, result2} }