lab/multi-agent-paper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs: redesign T1 client creation with transport selector pattern, remove grpc.ClientConn from QUIC path

Browse Source
This commit is contained in:
Godopu
2026-07-06 11:55:11 +09:00
parent c6dab0a901
commit f2a1611390
1 changed files with 150 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gRPC_Based_Interface/IMPL_DESIGN.md
+150 -59
View File
@@ -39,7 +39,7 @@
- `net/http` 인터페이스 호환 → HTTP/3 서버를 표준 핸들러로 운영 가능
- Connection migration 내장: 핸드오버(IP 변경) 시 세션 유지
- QUIC Datagram 확장(RFC 9221) 지원 → 저지연 V2X 메시지에 활용 가능
- gRPC-Go 트랜스포트 레이어 교체 방식으로 통합 (`WithContextDialer` + `quic.DialAddr`)
- TransportSelector 기반의 사전 프로브 및 분기 방식으로 통합 (UDP 성공 시 connect-go/HTTP3 RoundTripper, 실패 시 grpc-go/HTTP2 폴백)
#### gRPC-Go (`google.golang.org/grpc`)
- 공식 Google Go gRPC 구현 — Interceptor Chain, Health Checking, Service Config(retry policy) 내장
@@ -554,16 +554,19 @@ var defaultDeadlines = map[string]int64{
T1 클라우드와 T2 엣지 게이트웨이 간 무선 링크의 신뢰성을 보장하기 위해, 기본 UDP/QUIC 연결 수립 실패 시 TCP/HTTP/2 경로로 자동 전환하는 전송 폴백 메커니즘을 구현한다. 본 구조는 포트나 전송 프로토콜이 상이하여 단일 핸드셰이크 내 ALPN 강등 협상이 불가하므로, 클라이언트 단의 다이얼러에서 명시적인 연결 실패 감지 후 폴백을 수행한다.
##### 1. 클라이언트 측 전송 어댑터 및 폴백 다이얼러 (`internal/gateway/fallback_dialer.go`)
##### 1. 클라이언트 측 전송 어댑터 및 트랜스포트 셀렉터 (`internal/gateway/fallback_dialer.go`)
공식 gRPC-Go의 단일 커넥션 다이얼러 방식은 하나의 QUIC 스트림으로 HTTP/2 전체 프레임을 터널링할 경우 QUIC 세션 내부에서 TCP 수준의 HOLB를 그대로 재현하는 설계 결함(naive 구현)을 갖는다. 이를 방지하기 위해 각 RPC 호출을 독립된 QUIC 스트림에 1:1로 매핑하는 구조를 채택하며, 다음의 두 가지 대안을 명세한다:
- **(a) 권장 구현 — HTTP/3 의미론 기반 매핑 (connect-go 활용)**:
`quic-go/http3` 패키지의 `http3.RoundTripper`를 기반으로 동작하는 HTTP/3 클라이언트를 활용하여 gRPC와 와이어 프로토콜 수준에서 호환되도록 구성한다. 이 방식은 HTTP/3 명세에 따라 **개별 RPC 호출(Request/Response)이 네트워크 수준의 독립된 QUIC 스트림으로 자동 매핑**되므로 패킷 유실 시에도 다른 RPC 스트림이 정체되지 않는다.
`quic-go/http3` 패키지의 `http3.RoundTripper`를 기반으로 동작하는 HTTP/3 클라이언트를 활용하여 gRPC와 와이어 프로토콜 수준에서 호환되도록 구성한다. 이 방식은 HTTP/3 명세에 따라 **개별 RPC 호출(Request/Response)이 네트워크 수준의 독립된 QUIC 스트림으로 자동 매핑**되므로 패킷 유실 시에도 다른 RPC 스트림이 정체되지 않는다. 이 권장 구현의 Go 코드 세부 사양은 부록 [internal/cloud/grpc_quic_client.go](file:///Users/godopu16/PuKi/lab/canary_projects/multi-agent-paper/gRPC_Based_Interface/IMPL_DESIGN.md#L1498)의 `NewT1Client` QUIC 분기 구현(RoundTripper 및 http.Client 연계)을 통해 구체화된다.
- **(b) 대안 구현 — per-RPC 커스텀 ClientTransport**:
gRPC-Go의 커스텀 `ClientTransport` 인터페이스를 직접 구현하여, 클라이언트가 RPC를 개시할 때마다 `quic.Connection.OpenStreamSync`를 호출하여 새 QUIC 스트림을 동적으로 개방하고, 해당 스트림에 length-prefixed gRPC 바이트 프레임을 직접 매핑하여 송수신한다. 단, 이 대안은 gRPC-Go 내부의 private transport API 변동에 따른 유지보수 리스크가 있음을 명시한다.
아래는 다이얼 실패 감지 및 `tls.Config` 분리 복제(`Clone`)가 적용된 폴백 다이얼러 구현 명세이다:
> [!IMPORTANT]
> **구조적 제약 규정**: QUIC 경로 클라이언트 기동 시, 기존의 naive 터널링 회귀를 원천적으로 방지하기 위해 `grpc.DialContext` 및 `WithContextDialer` API의 사용을 구조적으로 금지한다. QUIC 전송은 오직 `http3.RoundTripper`와 와이어 수준에서 호환되는 HTTP/3 기반 클라이언트(예: `connect-go`)를 통해서만 독립 QUIC 스트림으로 개별 RPC를 매핑해야 한다.
아래는 다이얼 실패 감지 및 `tls.Config` 분리 복제(`Clone`)가 적용된 트랜스포트 셀렉터 구현 명세이다:
```go
package gateway
@@ -571,62 +574,53 @@ package gateway
import (
"context"
"crypto/tls"
"fmt"
"net"
"time"
"github.com/quic-go/quic-go"
)
// FallbackDialer는 QUIC 연결 시도 실패 시 TCP/HTTP2로 전환하는 다이얼러이다.
type FallbackDialer struct {
// TransportKind는 연결에 사용될 최적의 전송 방식을 정의한다.
type TransportKind int
const (
TransportQUIC TransportKind = iota // UDP/QUIC (HTTP/3)
TransportH2 // TCP/HTTP2 (폴백)
)
// TransportSelector는 연결 수립 전 UDP/QUIC의 가용성을 사전에 프로브하는 셀렉터이다.
type TransportSelector struct {
quicTLSConfig *tls.Config // QUIC 전용 ALPN (h3)
tcpTLSConfig *tls.Config // TCP 전용 ALPN (h2)
timeout time.Duration
}
func NewFallbackDialer(baseConfig *tls.Config, timeout time.Duration) *FallbackDialer {
func NewTransportSelector(baseConfig *tls.Config, timeout time.Duration) *TransportSelector {
// 호출자 tls.Config의 사이드 이펙트 방지를 위해 복제(Clone) 수행
quicTLS := baseConfig.Clone()
quicTLS.NextProtos = []string{"h3"} // QUIC은 h3 프로토콜에 한정
tcpTLS := baseConfig.Clone()
tcpTLS.NextProtos = []string{"h2"} // TCP/HTTP2는 h2 프로토콜에 한정
return &FallbackDialer{
return &TransportSelector{
quicTLSConfig: quicTLS,
tcpTLSConfig: tcpTLS,
timeout: timeout,
}
}
// DialContext는 UDP/QUIC 다이얼을 시도하고 실패 시 0-RTT 이내에 TCP 기반 HTTP/2 연결로 폴백한다.
func (d *FallbackDialer) DialContext(ctx context.Context, addr string) (net.Conn, error) {
quicCtx, cancel := context.WithTimeout(ctx, d.timeout)
// Probe는 UDP/QUIC 다이얼을 시도하여 가용성을 판정하고, 즉시 연결을 해제한 뒤 적절한 TransportKind를 반환한다.
// (성공 시 커넥션을 즉시 닫고 TransportQUIC 반환. 0-RTT/세션 재개 각주 및 프로브 커넥션 캐싱 변형 병기 가능)
func (s *TransportSelector) Probe(ctx context.Context, addr string) (TransportKind, error) {
probeCtx, cancel := context.WithTimeout(ctx, s.timeout)
defer cancel()
// 1. QUIC(HTTP/3) Dial 시도
// 각 RPC 호출마다 개별 스트림을 매핑하는 connect-go 또는 커스텀 Transport 구조의 세션을 관리
quicConn, err := quic.DialAddr(quicCtx, addr, d.quicTLSConfig, nil)
// 1. QUIC(HTTP/3) Dial 프로브 시도
conn, err := quic.DialAddr(probeCtx, addr, s.quicTLSConfig, nil)
if err == nil {
// 단일 커넥션 내부 스트림 생성을 래핑하는 커넥션 핸들러 반환
return &quicSessionWrapper{conn: quicConn}, nil
// 프로브 성공 후 커넥션 종료 (HTTP/3 RoundTripper에서 정식으로 독립 연결을 맺고 세션을 관리하도록 함)
conn.CloseWithError(0, "probe finished")
return TransportQUIC, nil
}
// 2. QUIC 다이얼 실패 또는 UDP 블로킹 감지 시 TCP/HTTP/2 Fallback 실행
dialer := &net.Dialer{Timeout: d.timeout}
tcpConn, err := dialer.DialContext(ctx, "tcp", addr)
if err != nil {
return nil, fmt.Errorf("both QUIC dial and HTTP/2 fallback failed: %w", err)
}
tlsConn := tls.Client(tcpConn, d.tcpTLSConfig)
if err := tlsConn.HandshakeContext(ctx); err != nil {
tcpConn.Close()
return nil, err
}
return tlsConn, nil
// 2. 다이얼 실패 시 h2 폴백 지시
return TransportH2, nil
}
```
@@ -690,6 +684,13 @@ func (s *GatewayServer) Start() error {
}
```
##### 2.1. 전송 계층 단일화 대안 및 런타임 강등 정책
- **M3-alt 대안: connect-go + http2.Transport 기반 폴백 단일화**:
기본 구현은 QUIC 경로에는 `connect-go`를, 폴백 TCP 경로에는 표준 `grpc-go`를 분리 적용하여 구현하나, 설계 대안으로 **폴백 TCP 경로마저 connect-go + http2.Transport 스택으로 단일화**하는 방안을 고려할 수 있다. 이 경우 클라이언트 스택은 `http.Client` 하나로 완전히 단일화되어 `connectT1Client` 1종만 구현하면 되며, 서버 측의 단일 공용 HTTP 핸들러 공유 구조와 구조적인 대칭성을 완벽히 만족하게 된다.
- **M4 정책: 런타임 강등 및 복구 정책**:
T1 클라이언트 기동 이후 연속 3회 이상 네트워크 타임아웃 또는 전송 계층 에러가 검출될 경우, 클라이언트는 현재 커넥션을 파기하고 즉시 `TransportSelector.Probe`를 재수수행하여 네트워크 가용성을 다시 판정한 후 T1Client 인스턴스를 재생성한다. 단, 이 과정에서의 기존 실행 중인 RPC 세션 정보 복구 및 상태 동기화의 상세 알고리즘은 향후 연구 범위로 위임한다.
##### 3. 소거 실험용 비교군 명세 (Naive Single-Stream Tunneling)
스트림 매핑 유무가 HOLB 해소에 미치는 실질적인 영향을 입증하는 소거 실험(Ablation Study) 비교군 구성을 위해, 기존에 논의된 단일 QUIC 스트림 고정 터널링(Naive) 코드를 아래와 같이 별도로 명세한다. 이 명세는 평가 단계(Evaluation)에서 패킷 손실 시 TCP와 동등한 수준의 지연시간 병목이 유발됨을 입증하는 대조군으로 기능한다.
@@ -1488,23 +1489,70 @@ func (c *quicStreamConn) SetDeadline(t interface{}) error { return nil }
func (c *quicStreamConn) SetReadDeadline(t interface{}) error { return nil }
func (c *quicStreamConn) SetWriteDeadline(t interface{}) error { return nil }
// T1Client는 T1 클라우드에서 T2 게이트웨이로 연결하는 gRPC 클라이언트다.
type T1Client struct {
conn *grpc.ClientConn
agentClient agentv1.AgentServiceClient
telemetryClient telemetryv1.TelemetryServiceClient
// TransportKind는 감지된 전송 프로토콜 종류를 나타낸다.
type TransportKind int
const (
TransportQUIC TransportKind = iota
TransportH2
)
// T1Client는 T1 클라우드에서 T2 게이트웨이로 연결하는 gRPC 클라이언트 인터페이스이다.
type T1Client interface {
ExecuteTask(ctx context.Context, task *agentv1.AgentTask) (*agentv1.AgentResult, error)
StreamTelemetry(ctx context.Context, query *telemetryv1.TelemetryQuery, handler func(*telemetryv1.DataPoint) error) error
Close() error
}
// NewT1Client는 T2 게이트웨이에 gRPC over QUIC 연결을 수립한다.
func NewT1Client(ctx context.Context, gatewayAddr string, tlsConfig *tls.Config) (*T1Client, error) {
dialer := NewQUICDialer(tlsConfig)
// NewT1Client는 TransportSelector를 통해 QUIC 가용성을 우선 프로브하고 최적의 클라이언트 스택을 구성한다.
func NewT1Client(ctx context.Context, gatewayAddr string, tlsConfig *tls.Config) (T1Client, error) {
// T1 클라이언트 수립 시, QUIC 경로에서는 grpc.Dial 및 WithContextDialer 사용이 구조적으로 금지된다.
selector := NewTransportSelector(tlsConfig, 3*time.Second)
kind, err := selector.Probe(ctx, gatewayAddr)
if err != nil {
kind = TransportH2
}
conn, err := grpc.DialContext(
ctx,
gatewayAddr,
grpc.WithContextDialer(dialer.DialContext),
grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)),
// Retry policy: 무선 손실 시 지수 백오프
if kind == TransportQUIC {
// (a) 권장 구현 — connect-go + HTTP/3 RoundTripper 구성 (grpc.ClientConn 미사용)
quicTLS := tlsConfig.Clone()
quicTLS.NextProtos = []string{"h3"}
h3RoundTripper := &http3.RoundTripper{
TLSClientConfig: quicTLS,
}
httpClient := &http.Client{
Transport: h3RoundTripper,
Timeout: 10 * time.Second,
}
return &connectT1Client{
httpClient: httpClient,
agentClient: agentv1.NewAgentServiceClient(httpClient),
telemetryClient: telemetryv1.NewTelemetryServiceClient(httpClient),
}, nil
}
// (b) 폴백 구현 — grpc-go over HTTP/2 구성 (TCP 전용 다이얼러 적용)
tcpTLS := tlsConfig.Clone()
tcpTLS.NextProtos = []string{"h2"}
dialer := func(ctx context.Context, addr string) (net.Conn, error) {
d := &net.Dialer{Timeout: 3 * time.Second}
conn, err := d.DialContext(ctx, "tcp", addr)
if err != nil {
return nil, err
}
tlsConn := tls.Client(conn, tcpTLS)
if err := tlsConn.HandshakeContext(ctx); err != nil {
conn.Close()
return nil, err
}
return tlsConn, nil
}
conn, err := grpc.DialContext(ctx, gatewayAddr,
grpc.WithContextDialer(dialer),
grpc.WithTransportCredentials(insecure.NewCredentials()), // dialer단에서 tls 핸드셰이크 처리 완료
grpc.WithDefaultServiceConfig(`{
"methodConfig": [{
"name": [{}],
@@ -1519,23 +1567,28 @@ func NewT1Client(ctx context.Context, gatewayAddr string, tlsConfig *tls.Config)
}`),
)
if err != nil {
return nil, fmt.Errorf("grpc dial: %w", err)
return nil, fmt.Errorf("tcp fallback dial failed: %w", err)
}
return &T1Client{
return &grpcT1Client{
conn: conn,
agentClient: agentv1.NewAgentServiceClient(conn),
telemetryClient: telemetryv1.NewTelemetryServiceClient(conn),
}, nil
}
// ExecuteTask는 T2 게이트웨이에 AgentTask를 전송한다.
func (c *T1Client) ExecuteTask(ctx context.Context, task *agentv1.AgentTask) (*agentv1.AgentResult, error) {
// connectT1Client는 connect-go/HTTP3 기반의 T1Client 구현체이다. (RPC 단위 독립 QUIC 스트림 매핑)
type connectT1Client struct {
httpClient *http.Client
agentClient agentv1.AgentServiceClient
telemetryClient telemetryv1.TelemetryServiceClient
}
func (c *connectT1Client) ExecuteTask(ctx context.Context, task *agentv1.AgentTask) (*agentv1.AgentResult, error) {
return c.agentClient.Execute(ctx, task)
}
// StreamTelemetry는 T2 게이트웨이에서 텔레메트리 스트림을 구독한다.
func (c *T1Client) StreamTelemetry(
func (c *connectT1Client) StreamTelemetry(
ctx context.Context,
query *telemetryv1.TelemetryQuery,
handler func(*telemetryv1.DataPoint) error,
@@ -1544,7 +1597,6 @@ func (c *T1Client) StreamTelemetry(
if err != nil {
return fmt.Errorf("stream start: %w", err)
}
for {
dp, err := stream.Recv()
if err != nil {
@@ -1556,8 +1608,47 @@ func (c *T1Client) StreamTelemetry(
}
}
// Close는 gRPC 연결을 종료한다.
func (c *T1Client) Close() error { return c.conn.Close() }
func (c *connectT1Client) Close() error {
if transport, ok := c.httpClient.Transport.(*http3.RoundTripper); ok {
return transport.Close()
}
return nil
}
// grpcT1Client는 gRPC-Go/HTTP2 기반의 T1Client 폴백 구현체이다.
type grpcT1Client struct {
conn *grpc.ClientConn
agentClient agentv1.AgentServiceClient
telemetryClient telemetryv1.TelemetryServiceClient
}
func (c *grpcT1Client) ExecuteTask(ctx context.Context, task *agentv1.AgentTask) (*agentv1.AgentResult, error) {
return c.agentClient.Execute(ctx, task)
}
func (c *grpcT1Client) StreamTelemetry(
ctx context.Context,
query *telemetryv1.TelemetryQuery,
handler func(*telemetryv1.DataPoint) error,
) error {
stream, err := c.telemetryClient.Stream(ctx, query)
if err != nil {
return fmt.Errorf("stream start: %w", err)
}
for {
dp, err := stream.Recv()
if err != nil {
return fmt.Errorf("stream recv: %w", err)
}
if err := handler(dp); err != nil {
return err
}
}
}
func (c *grpcT1Client) Close() error {
return c.conn.Close()
}
```
---