diff --git a/internal/proxy/client.go b/internal/proxy/client.go index ff02f63..c43be7a 100644 --- a/internal/proxy/client.go +++ b/internal/proxy/client.go @@ -8,7 +8,9 @@ import ( "net" "net/http" "net/url" + "sort" "strconv" + "sync" "time" "github.com/dalbodeule/hop-gate/internal/dtls" @@ -22,6 +24,9 @@ type ClientProxy struct { HTTPClient *http.Client Logger logging.Logger LocalTarget string // e.g. "127.0.0.1:8080" + + sendersMu sync.Mutex + streamSenders map[protocol.StreamID]*streamSender } // NewClientProxy 는 기본 HTTP 클라이언트 및 로거를 사용해 ClientProxy 를 생성합니다. (ko) @@ -46,8 +51,9 @@ func NewClientProxy(logger logging.Logger, localTarget string) *ClientProxy { ExpectContinueTimeout: 1 * time.Second, }, }, - Logger: logger.With(logging.Fields{"component": "client_proxy"}), - LocalTarget: localTarget, + Logger: logger.With(logging.Fields{"component": "client_proxy"}), + LocalTarget: localTarget, + streamSenders: make(map[protocol.StreamID]*streamSender), } } @@ -56,6 +62,88 @@ func NewClientProxy(logger logging.Logger, localTarget string) *ClientProxy { // StartLoop reads protocol.Envelope messages from the DTLS session; for HTTP/stream // messages it performs local HTTP requests and writes back responses over the DTLS // tunnel. (en) +type streamSender struct { + mu sync.Mutex + outstanding map[uint64][]byte +} + +func newStreamSender() *streamSender { + return &streamSender{ + outstanding: make(map[uint64][]byte), + } +} + +func (s *streamSender) register(seq uint64, data []byte) { + s.mu.Lock() + defer s.mu.Unlock() + + if s.outstanding == nil { + s.outstanding = make(map[uint64][]byte) + } + buf := make([]byte, len(data)) + copy(buf, data) + s.outstanding[seq] = buf +} + +func (s *streamSender) handleAck(ack *protocol.StreamAck) map[uint64][]byte { + s.mu.Lock() + defer s.mu.Unlock() + + if s.outstanding == nil { + return nil + } + + // 연속 수신 완료 구간(seq <= AckSeq)은 outstanding 에서 제거합니다. + for seq := range s.outstanding { + if seq <= ack.AckSeq { + delete(s.outstanding, seq) + } + } + + // LostSeqs 가 비어 있으면 재전송할 것이 없습니다. + if len(ack.LostSeqs) == 0 { + return nil + } + + // LostSeqs 에 포함된 시퀀스 중, 아직 outstanding 에 남아 있는 것들만 재전송 대상으로 선택합니다. + lost := make(map[uint64][]byte, len(ack.LostSeqs)) + for _, seq := range ack.LostSeqs { + if data, ok := s.outstanding[seq]; ok { + buf := make([]byte, len(data)) + copy(buf, data) + lost[seq] = buf + } + } + return lost +} + +func (p *ClientProxy) registerStreamSender(id protocol.StreamID, sender *streamSender) { + p.sendersMu.Lock() + defer p.sendersMu.Unlock() + if p.streamSenders == nil { + p.streamSenders = make(map[protocol.StreamID]*streamSender) + } + p.streamSenders[id] = sender +} + +func (p *ClientProxy) unregisterStreamSender(id protocol.StreamID) { + p.sendersMu.Lock() + defer p.sendersMu.Unlock() + if p.streamSenders == nil { + return + } + delete(p.streamSenders, id) +} + +func (p *ClientProxy) getStreamSender(id protocol.StreamID) *streamSender { + p.sendersMu.Lock() + defer p.sendersMu.Unlock() + if p.streamSenders == nil { + return nil + } + return p.streamSenders[id] +} + func (p *ClientProxy) StartLoop(ctx context.Context, sess dtls.Session) error { if ctx == nil { ctx = context.Background() @@ -109,6 +197,44 @@ func (p *ClientProxy) StartLoop(ctx context.Context, sess dtls.Session) error { }) return err } + case protocol.MessageTypeStreamAck: + sa := env.StreamAck + if sa == nil { + log.Error("received stream_ack envelope with nil payload", nil) + return fmt.Errorf("stream_ack payload is nil") + } + streamID := protocol.StreamID(sa.ID) + sender := p.getStreamSender(streamID) + if sender == nil { + log.Warn("received stream_ack for unknown stream", logging.Fields{ + "stream_id": sa.ID, + }) + continue + } + lost := sender.handleAck(sa) + // LostSeqs 를 기반으로 선택적 재전송 수행 + for seq, data := range lost { + retryEnv := protocol.Envelope{ + Type: protocol.MessageTypeStreamData, + StreamData: &protocol.StreamData{ + ID: streamID, + Seq: seq, + Data: data, + }, + } + if err := codec.Encode(sess, &retryEnv); err != nil { + log.Error("failed to retransmit stream_data after stream_ack", logging.Fields{ + "stream_id": streamID, + "seq": seq, + "error": err.Error(), + }) + return err + } + log.Info("retransmitted stream_data after stream_ack", logging.Fields{ + "stream_id": streamID, + "seq": seq, + }) + } default: log.Error("received unsupported envelope type from server", logging.Fields{ "type": env.Type, @@ -187,6 +313,10 @@ func (p *ClientProxy) handleStreamRequest(ctx context.Context, sess dtls.Session } streamID := so.ID + // 이 스트림에 대한 송신 측 ARQ 상태를 준비하고, StartLoop 에서 들어오는 StreamAck 와 연동합니다. + sender := newStreamSender() + p.registerStreamSender(streamID, sender) + defer p.unregisterStreamSender(streamID) // Pseudo-header 에서 HTTP 메타데이터를 추출합니다. (ko) // Extract HTTP metadata from pseudo-headers. (en) @@ -222,7 +352,17 @@ func (p *ClientProxy) handleStreamRequest(ctx context.Context, sess dtls.Session // 요청 바디를 StreamData/StreamClose 프레임에서 모두 읽어 메모리에 적재합니다. (ko) // Read the entire request body from StreamData/StreamClose frames into memory. (en) - var bodyBuf bytes.Buffer + // + // 동시에 수신 측 ARQ 상태( expectedSeq / out-of-order 버퍼 / LostSeqs )를 관리하고 + // StreamAck 를 전송해 선택적 재전송(Selective Retransmission)을 유도합니다. + var ( + bodyBuf bytes.Buffer + expectedSeq uint64 + received = make(map[uint64][]byte) + lost = make(map[uint64]struct{}) + ) + const maxLostReport = 32 + for { var env protocol.Envelope if err := codec.Decode(sess, &env); err != nil { @@ -241,11 +381,91 @@ func (p *ClientProxy) handleStreamRequest(ctx context.Context, sess dtls.Session if sd.ID != streamID { return fmt.Errorf("stream_data for unexpected stream id %q (expected %q)", sd.ID, streamID) } - if len(sd.Data) > 0 { - if _, err := bodyBuf.Write(sd.Data); err != nil { - return fmt.Errorf("buffer stream_data: %w", err) + + // 수신 측 ARQ: Seq 에 따라 분기 + switch { + case sd.Seq == expectedSeq: + // 기대하던 순서의 프레임: 바로 bodyBuf 에 기록하고, 이후 버퍼된 연속 프레임도 flush. + if len(sd.Data) > 0 { + if _, err := bodyBuf.Write(sd.Data); err != nil { + return fmt.Errorf("buffer stream_data: %w", err) + } + } + expectedSeq++ + for { + data, ok := received[expectedSeq] + if !ok { + break + } + if len(data) > 0 { + if _, err := bodyBuf.Write(data); err != nil { + return fmt.Errorf("buffer reordered stream_data: %w", err) + } + } + delete(received, expectedSeq) + delete(lost, expectedSeq) + expectedSeq++ + } + + // AckSeq 이전 구간의 lost 항목 정리 + for seq := range lost { + if seq < expectedSeq { + delete(lost, seq) + } + } + + case sd.Seq > expectedSeq: + // 앞선 일부 Seq 들이 누락된 상태: 현재 프레임을 버퍼링하고 missing seq 들을 lost 에 추가. + if len(sd.Data) > 0 { + buf := make([]byte, len(sd.Data)) + copy(buf, sd.Data) + received[sd.Seq] = buf + } + for seq := expectedSeq; seq < sd.Seq && len(lost) < maxLostReport; seq++ { + if _, ok := lost[seq]; !ok { + lost[seq] = struct{}{} + } + } + + default: + // sd.Seq < expectedSeq 인 경우: 이미 처리했거나 Ack 로 커버된 프레임 → 무시. + } + + // 수신 측 StreamAck 전송: + // - AckSeq: 0부터 시작해 연속으로 수신 완료한 마지막 시퀀스 (expectedSeq-1) + // - LostSeqs: 현재 윈도우 내에서 누락된 시퀀스 중 상한 개수(maxLostReport)까지만 포함 + var ackSeq uint64 + if expectedSeq == 0 { + ackSeq = 0 + } else { + ackSeq = expectedSeq - 1 + } + + lostSeqs := make([]uint64, 0, len(lost)) + for seq := range lost { + if seq >= expectedSeq { + lostSeqs = append(lostSeqs, seq) } } + if len(lostSeqs) > 0 { + sort.Slice(lostSeqs, func(i, j int) bool { return lostSeqs[i] < lostSeqs[j] }) + if len(lostSeqs) > maxLostReport { + lostSeqs = lostSeqs[:maxLostReport] + } + } + + ackEnv := protocol.Envelope{ + Type: protocol.MessageTypeStreamAck, + StreamAck: &protocol.StreamAck{ + ID: streamID, + AckSeq: ackSeq, + LostSeqs: lostSeqs, + }, + } + if err := codec.Encode(sess, &ackEnv); err != nil { + return fmt.Errorf("send stream ack: %w", err) + } + case protocol.MessageTypeStreamClose: sc := env.StreamClose if sc == nil { @@ -322,6 +542,8 @@ haveBody: Data: []byte("HopGate: " + errMsg), }, } + // 에러 응답 프레임도 ARQ 대상에 등록합니다. + sender.register(0, dataEnv.StreamData.Data) if err2 := codec.Encode(sess, &dataEnv); err2 != nil { logReq.Error("failed to encode stream response data envelope (error path)", logging.Fields{ "error": err2.Error(), @@ -390,6 +612,9 @@ haveBody: n, err := res.Body.Read(chunk) if n > 0 { dataCopy := append([]byte(nil), chunk[:n]...) + // 송신 측 ARQ: Seq 별 payload 를 기록해 두었다가, StreamAck 의 LostSeqs 를 기반으로 재전송할 수 있습니다. + sender.register(seq, dataCopy) + dataEnv := protocol.Envelope{ Type: protocol.MessageTypeStreamData, StreamData: &protocol.StreamData{ diff --git a/progress.md b/progress.md index eb5af47..c396244 100644 --- a/progress.md +++ b/progress.md @@ -7,12 +7,14 @@ This document tracks implementation progress against the HopGate architecture an ## 1. High-level Status / 상위 수준 상태 -- 아키텍처 문서 및 README 정리 완료 (ko/en 병기). - Architecture and README are documented in both Korean and English. -- 서버/클라이언트 엔트리 포인트, DTLS 핸드셰이크, 기본 PostgreSQL/ent 스키마까지 1차 뼈대 구현 완료. - First skeleton implementation is done for server/client entrypoints, DTLS handshake, and basic PostgreSQL/ent schema. -- 실제 Proxy 동작(HTTP ↔ DTLS 터널링), Admin API의 비즈니스 로직, 실 ACME 연동 등은 아직 남아 있음. - Actual proxying (HTTP ↔ DTLS tunneling), admin API business logic, and real ACME integration are still pending. +- 아키텍처 문서 및 README 정리 완료 (ko/en 병기). + Architecture and README are documented in both Korean and English. +- 서버/클라이언트 엔트리 포인트, DTLS 핸드셰이크, 기본 PostgreSQL/ent 스키마까지 1차 뼈대 구현 완료. + First skeleton implementation is done for server/client entrypoints, DTLS handshake, and basic PostgreSQL/ent schema. +- 기본 Proxy 동작(HTTP ↔ DTLS 터널링), Admin API 비즈니스 로직, ACME 기반 인증서 관리는 구현 완료된 상태. + Core proxying (HTTP ↔ DTLS tunneling), admin API business logic, and ACME-based certificate management are implemented. +- 스트림 ARQ, Observability, Hardening, ACME 고급 전략 등은 아직 남아 있는 다음 단계 작업이다. + Stream-level ARQ, observability, hardening, and advanced ACME operational strategies remain as next-step work items. --- @@ -39,12 +41,12 @@ This document tracks implementation progress against the HopGate architecture an ### 2.2 Server / Client Entrypoints -- 서버 메인: [`cmd/server/main.go`](cmd/server/main.go) - - 서버 설정 로드 (`LoadServerConfigFromEnv`). - - PostgreSQL 연결 및 ent 스키마 init (`store.OpenPostgresFromEnv`). - - Debug 모드 시 self-signed localhost cert 생성 (`dtls.NewSelfSignedLocalhostConfig`). - - DTLS 서버 생성 (`dtls.NewPionServer`) 및 Accept + Handshake 루프 (`PerformServerHandshake`). - - DummyDomainValidator 사용해 도메인/API Key 조합을 임시로 모두 허용. +- 서버 메인: [`cmd/server/main.go`](cmd/server/main.go) + - 서버 설정 로드 (`LoadServerConfigFromEnv`). + - PostgreSQL 연결 및 ent 스키마 init (`store.OpenPostgresFromEnv`). + - Debug 모드 시 self-signed localhost cert 생성 (`dtls.NewSelfSignedLocalhostConfig`). + - DTLS 서버 생성 (`dtls.NewPionServer`) 및 Accept + Handshake 루프 (`PerformServerHandshake`). + - ent 기반 `DomainValidator` + `domainGateValidator` 를 사용해 `(domain, client_api_key)` 조합과 DNS/IP(옵션) 검증을 수행. - 클라이언트 메인: [`cmd/client/main.go`](cmd/client/main.go) - CLI + env 병합 설정 (우선순위: CLI > env). @@ -115,13 +117,13 @@ This document tracks implementation progress against the HopGate architecture an - `RegisterDomain(ctx, domain, memo) (clientAPIKey string, err error)` - `UnregisterDomain(ctx, domain, clientAPIKey string) error` -- HTTP Handler: [`internal/admin/http.go`](internal/admin/http.go) - - `Authorization: Bearer {ADMIN_API_KEY}` 검증. +- HTTP Handler: [`internal/admin/http.go`](internal/admin/http.go) + - `Authorization: Bearer {ADMIN_API_KEY}` 검증. - 엔드포인트: - - `POST /api/v1/admin/domains/register` - - `POST /api/v1/admin/domains/unregister` - - JSON request/response 구조 정의 및 기본 에러 처리. - - 아직 실제 서비스/라우터 wiring, ent 기반 구현 미완성. + - `POST /api/v1/admin/domains/register` + - `POST /api/v1/admin/domains/unregister` + - JSON request/response 구조 정의 및 기본 에러 처리. + - 실제 서비스(`DomainService`) 및 라우터 wiring, ent 기반 구현이 완료되어 도메인 등록/해제가 동작. --- @@ -224,10 +226,11 @@ This document tracks implementation progress against the HopGate architecture an ### 3.3 Proxy Core / HTTP Tunneling -- [x] 서버 측 Proxy 구현 확장: [`internal/proxy/server.go`](internal/proxy/server.go) - - HTTP/HTTPS 리스너와 DTLS 세션 매핑 구현. - - `Router` 구현체 추가 (도메인/패스 → 클라이언트/서비스). - - 요청/응답을 `internal/protocol` 구조체로 직렬화/역직렬화. +- [ ] 서버 측 Proxy 구현 확장: [`internal/proxy/server.go`](internal/proxy/server.go) + - 현재 `ServerProxy` / `Router` 인터페이스와 `NewHTTPServer` 만 정의되어 있고, + 실제 HTTP/HTTPS 리스너와 DTLS 세션 매핑 로직은 [`cmd/server/main.go`](cmd/server/main.go) 의 + `newHTTPHandler` / `dtlsSessionWrapper.ForwardHTTP` 안에 위치합니다. + - Proxy 코어 로직을 proxy 레이어로 이동하는 리팩터링은 아직 진행되지 않았습니다. (3.6 항목과 연동) - [x] 클라이언트 측 Proxy 구현 확장: [`internal/proxy/client.go`](internal/proxy/client.go) - DTLS 세션에서 `protocol.Request` 수신 → 로컬 HTTP 호출 → `protocol.Response` 전송 루프 구현. @@ -242,11 +245,11 @@ This document tracks implementation progress against the HopGate architecture an #### 3.3A Stream-based DTLS Tunneling / 스트림 기반 DTLS 터널링 -현재 HTTP 터널링은 **단일 JSON Envelope + 단일 DTLS 쓰기** 방식(요청/응답 바디 전체를 한 번에 전송)이므로, -대용량 응답 바디에서 UDP MTU 한계로 인한 `sendto: message too long` 문제가 발생할 수 있습니다. -프로덕션 전 단계에서 이 한계를 제거하기 위해, DTLS 위 애플리케이션 프로토콜을 **완전히 스트림/프레임 기반**으로 재설계합니다. -The current tunneling model uses a **single JSON envelope + single DTLS write per HTTP message**, which can hit UDP MTU limits (`sendto: message too long`) for large bodies. -Before production, we will redesign the application protocol over DTLS to be fully **stream/frame-based**. +초기 HTTP 터널링 설계는 **단일 JSON Envelope + 단일 DTLS 쓰기** 방식(요청/응답 바디 전체를 한 번에 전송)이었고, +대용량 응답 바디에서 UDP MTU 한계로 인한 `sendto: message too long` 문제가 발생할 수 있었습니다. +이 한계를 제거하기 위해, 현재 코드는 DTLS 위 애플리케이션 프로토콜을 **스트림/프레임 기반**으로 재설계하여 `StreamOpen` / `StreamData` / `StreamClose` 를 사용합니다. +The initial tunneling model used a **single JSON envelope + single DTLS write per HTTP message**, which could hit UDP MTU limits (`sendto: message too long`) for large bodies. +The current implementation uses a **stream/frame-based** protocol over DTLS (`StreamOpen` / `StreamData` / `StreamClose`), and this section documents its constraints and further improvements (e.g. ARQ). 고려해야 할 제약 / Constraints: @@ -312,51 +315,31 @@ The following tasks describe concrete work items to be implemented on the `featu ##### 3.3A.2 애플리케이션 레벨 ARQ 설계 (Selective Retransmission) ##### 3.3A.2 Application-level ARQ (Selective Retransmission) -- [x] 수신 측 스트림 상태 관리 로직 설계 - - 스트림별로 다음 상태를 유지합니다. - For each stream, maintain: - - `expectedSeq` (다음에 연속으로 기대하는 Seq, 초기값 0) - `expectedSeq` – next contiguous sequence expected (starts at 0) - - `received` (map[uint64][]byte) – 도착했지만 아직 순서가 맞지 않은 chunk 버퍼 - `received` – buffer for out-of-order chunks - - `lastAckSent`, `lostBuffer` – 마지막 ACK 상태 및 누락 시퀀스 기록 - `lastAckSent`, `lostBuffer` – last acknowledged seq and known missing sequences - - `StreamData{ID, Seq, Data}` 수신 시: - When receiving `StreamData{ID, Seq, Data}`: - - `Seq == expectedSeq` 인 경우: 바로 상위(HTTP Body writer)에 전달 후, - `expectedSeq++` 하면서 `received` map 에 쌓인 연속된 Seq 들을 순서대로 flush. - If `Seq == expectedSeq`, deliver to the HTTP body writer, increment `expectedSeq`, and flush any contiguous buffered seqs. - - `Seq > expectedSeq` 인 경우: `received[Seq] = Data` 로 버퍼링하고, - `expectedSeq` ~ `Seq-1` 구간 중 비어 있는 Seq 들을 `lostBuffer` 에 추가. - If `Seq > expectedSeq`, buffer as out-of-order and mark missing seqs in `lostBuffer`. +- [x] 수신 측 ARQ 상태 관리 구현 + - 스트림별로 `expectedSeq`, out-of-order chunk 버퍼(`received`), 누락 시퀀스 집합(`lost`)을 유지하면서, + in-order / out-of-order 프레임을 구분해 HTTP 바디 버퍼에 순서대로 쌓습니다. + - For each stream, maintain `expectedSeq`, an out-of-order buffer (`received`), and a lost-sequence set (`lost`), + delivering in-order frames directly to the HTTP body buffer while buffering/reordering out-of-order ones. -- [x] 수신 측 StreamAck 전송 정책 - - 주기적 타이머 또는 일정 수의 프레임 처리 후에 `StreamAck` 를 전송합니다. - Send `StreamAck` periodically or after processing N frames: - - `AckSeq = expectedSeq - 1` (연속 수신 완료 지점) - `AckSeq = expectedSeq - 1` – last contiguous sequence received - - `LostSeqs` 는 윈도우 내 손실 시퀀스 중 상한 개수까지만 포함 (과도한 길이 방지). - `LostSeqs` should only include a bounded set of missing seqs within the receive window. +- [x] 수신 측 StreamAck 전송 정책 구현 + - 각 `StreamData` 수신 시점에 `AckSeq = expectedSeq - 1` 과 현재 윈도우에서 누락된 시퀀스 일부(`LostSeqs`, 상한 개수 적용)를 포함한 + `StreamAck{AckSeq, LostSeqs}` 를 전송해 선택적 재전송을 유도합니다. + - On every `StreamData` frame, send `StreamAck{AckSeq, LostSeqs}` where `AckSeq = expectedSeq - 1` and `LostSeqs` + contains a bounded set (up to a fixed limit) of missing sequence numbers in the current receive window. -- [x] 송신 측 재전송 로직 - - 스트림별로 다음 상태를 유지합니다. - For each stream on the sender: - - `sendSeq` – 송신에 사용할 다음 Seq (0부터 시작) - - `outstanding` – map[seq]*FrameState (`data`, `lastSentAt`, `retryCount` 포함) - - 새 chunk 전송 시: - On new chunk: - - `seq := sendSeq`, `sendSeq++`, `outstanding[seq] = FrameState{...}`, - `StreamData{ID, Seq: seq, Data}` 전송. +- [x] 송신 측 재전송 로직 구현 (StreamAck 기반) + - 응답 스트림 송신 측에서 스트림별 `streamSender` 를 두고, `outstanding[seq] = payload` 로 아직 Ack 되지 않은 프레임을 추적합니다. - `StreamAck{AckSeq, LostSeqs}` 수신 시: - On receiving `StreamAck`: - - `seq <= AckSeq` 인 `outstanding` 항목은 **모두 삭제** (해당 지점까지 연속 수신으로 간주). - Delete all `outstanding` entries with `seq <= AckSeq`. - - `LostSeqs` 에 포함된 시퀀스는 즉시 재전송 (`retryCount++`, `lastSentAt = now` 업데이트). - Retransmit frames whose seqs are listed in `LostSeqs`. - - 타임아웃 기반 재전송: - Timeout-based retransmission: - - 주기적으로 `outstanding` 을 순회하며 `now - lastSentAt > RTO` 인 프레임을 재전송 (단순 고정 RTO 로 시작). - Periodically scan `outstanding` and retransmit frames that exceed a fixed RTO. + - `seq <= AckSeq` 인 항목은 모두 제거하고, + - `LostSeqs` 에 포함된 시퀀스에 대해서만 `StreamData{ID, Seq, Data}` 를 재전송합니다. + - A per-stream `streamSender` tracks `outstanding[seq] = payload` for unacknowledged frames. Upon receiving + `StreamAck{AckSeq, LostSeqs}`, it deletes all `seq <= AckSeq` and retransmits only frames whose sequence + numbers appear in `LostSeqs`. + +> Note: 현재 구현은 StreamAck 기반 **선택적 재전송(Selective Retransmission)** 까지 포함하며, +> 별도의 RTO(재전송 타이머) 기반 백그라운드 재전송 루프는 향후 확장 여지로 남겨둔 상태입니다. +> Note: The current implementation covers StreamAck-based **selective retransmission**; a separate RTO-based +> background retransmission loop is left as a potential future enhancement. --- @@ -364,10 +347,8 @@ The following tasks describe concrete work items to be implemented on the `featu ##### 3.3A.3 HTTP ↔ stream mapping (server/client) - [x] 서버 → 클라이언트 요청 스트림: [`cmd/server/main.go`](cmd/server/main.go:200) - - 현재 `ForwardHTTP` 는 단일 `HTTPRequest`/`HTTPResponse` 를 처리하는 구조입니다. - Currently `ForwardHTTP` handles a single `HTTPRequest`/`HTTPResponse` pair. - - 스트림 모드에서는 다음과 같이 바꿉니다. - In stream mode: + - `ForwardHTTP` 는 스트림 기반 HTTP 요청/응답을 처리하도록 구현되어 있으며, 동작은 다음과 같습니다. + `ForwardHTTP` is implemented in stream mode and behaves as follows: - HTTP 요청 수신 시: - 새로운 `StreamID` 를 발급합니다 (세션별 증가). Generate a new `StreamID` per incoming HTTP request on the DTLS session. @@ -406,20 +387,19 @@ The following tasks describe concrete work items to be implemented on the `featu ##### 3.3A.4 JSON → 바이너리 직렬화로의 잠재적 전환 (2단계) ##### 3.3A.4 JSON → binary serialization (potential phase 2) -- [ ] JSON 기반 스트림 프로토콜의 1단계 구현/안정화 이후, 직렬화 포맷 재검토 - - 현재는 디버깅/호환성 관점에서 JSON `Envelope` + base64 `[]byte` encoding 이 유리합니다. - For now, JSON `Envelope` + base64-encoded `[]byte` is convenient for debugging and compatibility. - - HTTP 바디 chunk 가 MTU-safe 크기(예: 4KiB)로 제한되므로, JSON 오버헤드는 수용 가능합니다. - Since body chunks are bounded to a safe MTU-sized payload, JSON overhead is acceptable initially. -- [ ] 필요 시 length-prefix 이진 프레임(Protobuf 등)으로 전환 +- [x] JSON 기반 스트림 프로토콜의 1단계 구현/안정화 이후, 직렬화 포맷 재검토 및 Protobuf 전환 + - 현재는 JSON 대신 Protobuf length-prefix `Envelope` 포맷을 기본으로 사용합니다. + The runtime now uses a Protobuf-based, length-prefixed `Envelope` format instead of JSON. + - HTTP/스트림 payload 는 여전히 MTU-safe 크기(예: 4KiB, `StreamChunkSize`)로 제한되어 있어, 단일 프레임이 과도하게 커지지 않습니다. + HTTP/stream payloads remain bounded to an MTU-safe size (e.g. 4KiB via `StreamChunkSize`), so individual frames stay small. +- [x] length-prefix 이진 프레임(Protobuf)으로 전환 - 동일한 logical model (`StreamOpen` / `StreamData(seq)` / `StreamClose` / `StreamAck`)을 유지한 채, - wire-format 만 Protobuf 또는 MsgPack 등의 length-prefix binary 프레이밍으로 교체할 수 있습니다. - We can later keep the same logical model and swap the wire format for Protobuf or other length-prefix binary framing. -- [x] 이 전환은 `internal/protocol` 내 직렬화 레이어를 얇은 abstraction 으로 감싸 구현할 수 있습니다. - - 현재는 [`internal/protocol/codec.go`](internal/protocol/codec.go:1) 에 `WireCodec` 인터페이스와 JSON 기반 `DefaultCodec` 을 도입하여, - 추후 Protobuf/이진 포맷으로 교체할 때 호출자는 `protocol.DefaultCodec` 만 사용하도록 분리해 두었습니다. - - This has been prepared via [`internal/protocol/codec.go`](internal/protocol/codec.go:1), which introduces a `WireCodec` interface - and a JSON-based `DefaultCodec` so that future Protobuf/binary codecs can be swapped in behind the same API. + wire-format 을 Protobuf length-prefix binary 프레이밍으로 교체했고, 이는 `protobufCodec` 으로 구현되었습니다. + We now keep the same logical model while using Protobuf length-prefixed framing via `protobufCodec`. +- [x] 이 전환은 `internal/protocol` 내 직렬화 레이어를 얇은 abstraction 으로 감싸 구현했습니다. + - [`internal/protocol/codec.go`](internal/protocol/codec.go:130) 에 `WireCodec` 인터페이스와 Protobuf 기반 `DefaultCodec` 을 도입해, + 호출자는 `protocol.DefaultCodec` 만 사용하고, JSON codec 은 보조 용도로만 남아 있습니다. + In [`internal/protocol/codec.go`](internal/protocol/codec.go:130), the `WireCodec` abstraction and Protobuf-based `DefaultCodec` allow callers to use only `protocol.DefaultCodec` while JSON remains as an auxiliary codec. --- @@ -442,10 +422,10 @@ The following tasks describe concrete work items to be implemented on the `featu ### 3.5 Observability / 관측성 -- [ ] Prometheus 메트릭 노출 및 서버 wiring +- [x] Prometheus 메트릭 노출 및 서버 wiring - `cmd/server/main.go` 에 Prometheus `/metrics` 엔드포인트 추가 (예: promhttp.Handler). - - DTLS 세션 수, DTLS 핸드셰이크 성공/실패 수, HTTP/Proxy 요청 수 및 에러 수에 대한 카운터/게이지 메트릭 정의. - - 도메인, 클라이언트 ID, request_id 등의 라벨 설계 및 현재 구조적 로깅 필드와 일관성 유지. + - DTLS 핸드셰이크 성공/실패 수, HTTP 요청 수, HTTP 요청 지연, Proxy 에러 수에 대한 메트릭을 정의합니다. + - 메트릭 라벨은 메서드/상태 코드/결과/에러 타입 등에 한정되며, 도메인/클라이언트 ID/request_id 는 구조적 로그 필드로만 노출됩니다. - [ ] Loki/Grafana 대시보드 및 쿼리 예시 - Loki/Promtail 구성을 가정한 주요 로그 쿼리 예시 정리(도메인, 클라이언트 ID, request_id 기준). @@ -482,9 +462,11 @@ The following tasks describe concrete work items to be implemented on the `featu ### Milestone 2 — Full HTTP Tunneling (프락시 동작 완성) -- [ ] 서버 Proxy 코어 구현 및 HTTPS ↔ DTLS 라우팅. -- [ ] 클라이언트 Proxy 루프 구현 및 로컬 서비스 연동. -- [ ] End-to-end HTTP 요청/응답 터널링 E2E 테스트. +- [x] 서버 Proxy 코어 구현 및 HTTPS ↔ DTLS 라우팅. + - 현재 `cmd/server/main.go` 의 `newHTTPHandler` / `dtlsSessionWrapper.ForwardHTTP` 경로에서 동작합니다. +- [x] 클라이언트 Proxy 루프 구현 및 로컬 서비스 연동. + - `cmd/client/main.go` + [`ClientProxy.StartLoop()`](internal/proxy/client.go:59) 를 통해 DTLS 세션 위에서 로컬 서비스와 연동됩니다. +- [ ] End-to-end HTTP 요청/응답 터널링 E2E 테스트. ### Milestone 3 — ACME + TLS/DTLS 정식 인증 @@ -494,7 +476,10 @@ The following tasks describe concrete work items to be implemented on the `featu ### Milestone 4 — Observability & Hardening -- [ ] Prometheus/Loki/Grafana 통합. +- [ ] Prometheus/Loki/Grafana 통합. + - Prometheus 메트릭 정의 및 `/metrics` 엔드포인트는 이미 구현 및 동작 중이며, + Loki/Promtail/Grafana 대시보드 및 운영 통합 작업은 아직 남아 있습니다. + - [ ] 에러/리트라이/타임아웃 정책 정교화. - [ ] 보안/구성 최종 점검 및 문서화.