dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-16 02:29:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use request id to make api idempotence on sdk retry (#4694)

Browse Source 
* Use request id to make api idempotence on sdk retry

Derive resource id from request id
This commit is contained in:
cnderrauber
2026-08-12 09:28:39 +08:00
committed by GitHub
parent 223587e140
commit fad2cc4afe
9 changed files with 143 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
pkg/service/agent_dispatch_service.go
+1 -1
View File
@@ -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/clients.go
+1 -1
View File
@@ -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)
}
pkg/service/ingress.go
+1
View File
@@ -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:
pkg/service/ioservice_ingress.go
+4 -3
View File
@@ -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) {
pkg/service/requestid.go
+53
View File
@@ -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))
}
pkg/service/requestid_test.go
+70
View File
@@ -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)
}
}
pkg/service/servicefakes/fake_ioclient.go
+10 -10
View File
@@ -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}
}