diff --git a/internal/errorpages/templates/504.html b/internal/errorpages/templates/504.html
index 9b43762..3554e44 100644
--- a/internal/errorpages/templates/504.html
+++ b/internal/errorpages/templates/504.html
@@ -5,6 +5,7 @@
diff --git a/internal/errorpages/templates/525.html b/internal/errorpages/templates/525.html
index f94e931..9ba510d 100644
--- a/internal/errorpages/templates/525.html
+++ b/internal/errorpages/templates/525.html
@@ -5,6 +5,7 @@
525 TLS Handshake Failed - HopGate
+
diff --git a/internal/protocol/codec.go b/internal/protocol/codec.go
index dcbdfdf..5a3940f 100644
--- a/internal/protocol/codec.go
+++ b/internal/protocol/codec.go
@@ -16,6 +16,16 @@ import (
// This matches existing 64KiB readers used around DTLS sessions (used by the JSON codec).
const defaultDecoderBufferSize = 64 * 1024
+// dtlsReadBufferSize 는 pion/dtls 내부 버퍼 한계에 맞춘 읽기 버퍼 크기입니다.
+// pion/dtls 의 UnpackDatagram 함수는 8KB (8,192 bytes) 의 기본 수신 버퍼를 사용합니다.
+// DTLS는 UDP 기반이므로 한 번의 Read()에서 전체 datagram을 읽어야 하며,
+// 이 크기를 초과하는 DTLS 레코드는 처리되지 않습니다.
+// dtlsReadBufferSize matches the pion/dtls internal buffer limit.
+// pion/dtls's UnpackDatagram function uses an 8KB (8,192 bytes) receive buffer.
+// Since DTLS is UDP-based, the entire datagram must be read in a single Read() call,
+// and DTLS records exceeding this size cannot be processed.
+const dtlsReadBufferSize = 8 * 1024 // 8KB
+
// maxProtoEnvelopeBytes 는 단일 Protobuf Envelope 의 최대 크기에 대한 보수적 상한입니다.
// 아직 하드 리미트로 사용하지는 않지만, 향후 방어적 체크에 사용할 수 있습니다.
const maxProtoEnvelopeBytes = 512 * 1024 // 512KiB, 충분히 여유 있는 값
@@ -46,12 +56,15 @@ func (jsonCodec) Decode(r io.Reader, env *Envelope) error {
return dec.Decode(env)
}
-// protobufCodec 은 Protobuf + length-prefix framing 기반 WireCodec 구현입니다.
-// 한 Envelope 당 [4바이트 big-endian 길이] + [protobuf bytes] 형태로 인코딩합니다.
+// protobufCodec 은 Protobuf length-prefix framing 기반 WireCodec 구현입니다.
+// 한 Envelope 당 [4바이트 big-endian 길이] [protobuf bytes] 형태로 인코딩합니다.
type protobufCodec struct{}
// Encode 는 Envelope 를 Protobuf Envelope 로 변환한 뒤, length-prefix 프레이밍으로 기록합니다.
+// DTLS는 UDP 기반이므로, length prefix와 protobuf 데이터를 단일 버퍼로 합쳐 하나의 Write로 전송합니다.
// Encode encodes an Envelope as a length-prefixed protobuf message.
+// For DTLS (UDP-based), we combine the length prefix and protobuf data into a single buffer
+// and send it with a single Write call to preserve message boundaries.
func (protobufCodec) Encode(w io.Writer, env *Envelope) error {
pbEnv, err := toProtoEnvelope(env)
if err != nil {
@@ -83,44 +96,50 @@ func (protobufCodec) Encode(w io.Writer, env *Envelope) error {
return fmt.Errorf("protobuf codec: empty marshaled envelope")
}
- var lenBuf [4]byte
if len(data) > int(^uint32(0)) {
return fmt.Errorf("protobuf codec: envelope too large: %d bytes", len(data))
}
- binary.BigEndian.PutUint32(lenBuf[:], uint32(len(data)))
- if _, err := w.Write(lenBuf[:]); err != nil {
- return fmt.Errorf("protobuf codec: write length prefix: %w", err)
- }
- if _, err := w.Write(data); err != nil {
- return fmt.Errorf("protobuf codec: write payload: %w", err)
+ // DTLS 환경에서는 length prefix와 protobuf 데이터를 단일 버퍼로 합쳐서 하나의 Write로 전송
+ // For DTLS, combine length prefix and protobuf data into a single buffer
+ frame := make([]byte, 4+len(data))
+ binary.BigEndian.PutUint32(frame[:4], uint32(len(data)))
+ copy(frame[4:], data)
+
+ if _, err := w.Write(frame); err != nil {
+ return fmt.Errorf("protobuf codec: write frame: %w", err)
}
return nil
}
// Decode 는 length-prefix 프레임에서 Protobuf Envelope 를 읽어들여
// 내부 Envelope 구조체로 변환합니다.
+// DTLS는 UDP 기반이므로, 한 번의 Read로 전체 데이터그램을 읽습니다.
// Decode reads a length-prefixed protobuf Envelope and converts it into the internal Envelope.
+// For DTLS (UDP-based), we read the entire datagram in a single Read call.
func (protobufCodec) Decode(r io.Reader, env *Envelope) error {
- var lenBuf [4]byte
- if _, err := io.ReadFull(r, lenBuf[:]); err != nil {
+ // 1) 길이 prefix 4바이트를 정확히 읽는다.
+ header := make([]byte, 4)
+ if _, err := io.ReadFull(r, header); err != nil {
return fmt.Errorf("protobuf codec: read length prefix: %w", err)
}
- n := binary.BigEndian.Uint32(lenBuf[:])
- if n == 0 {
+
+ length := binary.BigEndian.Uint32(header)
+ if length == 0 {
return fmt.Errorf("protobuf codec: zero-length envelope")
}
- if n > maxProtoEnvelopeBytes {
- return fmt.Errorf("protobuf codec: envelope too large: %d bytes (max %d)", n, maxProtoEnvelopeBytes)
+ if length > maxProtoEnvelopeBytes {
+ return fmt.Errorf("protobuf codec: envelope too large: %d bytes (max %d)", length, maxProtoEnvelopeBytes)
}
- buf := make([]byte, int(n))
- if _, err := io.ReadFull(r, buf); err != nil {
+ // 2) payload 를 length 바이트만큼 정확히 읽는다.
+ payload := make([]byte, int(length))
+ if _, err := io.ReadFull(r, payload); err != nil {
return fmt.Errorf("protobuf codec: read payload: %w", err)
}
var pbEnv protocolpb.Envelope
- if err := proto.Unmarshal(buf, &pbEnv); err != nil {
+ if err := proto.Unmarshal(payload, &pbEnv); err != nil {
return fmt.Errorf("protobuf codec: unmarshal envelope: %w", err)
}
@@ -128,9 +147,18 @@ func (protobufCodec) Decode(r io.Reader, env *Envelope) error {
}
// DefaultCodec 은 현재 런타임에서 사용하는 기본 WireCodec 입니다.
-// 이제 Protobuf 기반 codec 을 기본으로 사용합니다.
+// 현재는 Protobuf length-prefix 기반 codec 을 기본으로 사용합니다.
+// 서버와 클라이언트가 모두 이 버전을 사용해야 wire-format 이 일치합니다.
var DefaultCodec WireCodec = protobufCodec{}
+// GetDTLSReadBufferSize 는 DTLS 세션 읽기에 사용할 버퍼 크기를 반환합니다.
+// 이 값은 pion/dtls 내부 버퍼 한계(8KB)에 맞춰져 있습니다.
+// GetDTLSReadBufferSize returns the buffer size to use for reading from DTLS sessions.
+// This value is aligned with pion/dtls's internal buffer limit (8KB).
+func GetDTLSReadBufferSize() int {
+ return dtlsReadBufferSize
+}
+
// toProtoEnvelope 는 내부 Envelope 구조체를 Protobuf Envelope 로 변환합니다.
// 현재 구현은 HTTP 요청/응답 및 스트림 관련 타입(StreamOpen/StreamData/StreamClose/StreamAck)을 지원합니다.
func toProtoEnvelope(env *Envelope) (*protocolpb.Envelope, error) {
diff --git a/internal/protocol/codec_test.go b/internal/protocol/codec_test.go
new file mode 100644
index 0000000..6ca0fff
--- /dev/null
+++ b/internal/protocol/codec_test.go
@@ -0,0 +1,226 @@
+package protocol
+
+import (
+ "bufio"
+ "bytes"
+ "io"
+ "testing"
+)
+
+// mockDatagramConn simulates a datagram-based connection (like DTLS over UDP)
+// where each Write sends a separate message and each Read receives a complete message.
+// This mock verifies the FIXED behavior where the codec properly handles message boundaries.
+type mockDatagramConn struct {
+ messages [][]byte
+ readIdx int
+}
+
+func newMockDatagramConn() *mockDatagramConn {
+ return &mockDatagramConn{
+ messages: make([][]byte, 0),
+ }
+}
+
+func (m *mockDatagramConn) Write(p []byte) (n int, err error) {
+ // Simulate datagram behavior: each Write is a separate message
+ msg := make([]byte, len(p))
+ copy(msg, p)
+ m.messages = append(m.messages, msg)
+ return len(p), nil
+}
+
+func (m *mockDatagramConn) Read(p []byte) (n int, err error) {
+ // Simulate datagram behavior: each Read returns a complete message
+ if m.readIdx >= len(m.messages) {
+ return 0, io.EOF
+ }
+ msg := m.messages[m.readIdx]
+ m.readIdx++
+ if len(p) < len(msg) {
+ return 0, io.ErrShortBuffer
+ }
+ copy(p, msg)
+ return len(msg), nil
+}
+
+// TestProtobufCodecDatagramBehavior tests that the protobuf codec works correctly
+// with datagram-based transports (like DTLS over UDP) where message boundaries are preserved.
+func TestProtobufCodecDatagramBehavior(t *testing.T) {
+ codec := protobufCodec{}
+ conn := newMockDatagramConn()
+
+ // Create a test envelope
+ testEnv := &Envelope{
+ Type: MessageTypeHTTP,
+ HTTPRequest: &Request{
+ RequestID: "test-req-123",
+ ClientID: "client-1",
+ ServiceName: "test-service",
+ Method: "GET",
+ URL: "/test/path",
+ Header: map[string][]string{
+ "User-Agent": {"test-client"},
+ },
+ Body: []byte("test body content"),
+ },
+ }
+
+ // Encode the envelope
+ if err := codec.Encode(conn, testEnv); err != nil {
+ t.Fatalf("Failed to encode envelope: %v", err)
+ }
+
+ // Verify that exactly one message was written (length prefix + data in single Write)
+ if len(conn.messages) != 1 {
+ t.Fatalf("Expected 1 message to be written, got %d", len(conn.messages))
+ }
+
+ // Verify the message structure: [4-byte length][protobuf data]
+ msg := conn.messages[0]
+ if len(msg) < 4 {
+ t.Fatalf("Message too short: %d bytes", len(msg))
+ }
+
+ // Decode the envelope using a buffered reader (as we do in actual code)
+ // to handle datagram-based reading properly
+ reader := bufio.NewReaderSize(conn, GetDTLSReadBufferSize())
+ var decodedEnv Envelope
+ if err := codec.Decode(reader, &decodedEnv); err != nil {
+ t.Fatalf("Failed to decode envelope: %v", err)
+ }
+
+ // Verify the decoded envelope matches the original
+ if decodedEnv.Type != testEnv.Type {
+ t.Errorf("Type mismatch: got %v, want %v", decodedEnv.Type, testEnv.Type)
+ }
+ if decodedEnv.HTTPRequest == nil {
+ t.Fatal("HTTPRequest is nil after decode")
+ }
+ if decodedEnv.HTTPRequest.RequestID != testEnv.HTTPRequest.RequestID {
+ t.Errorf("RequestID mismatch: got %v, want %v", decodedEnv.HTTPRequest.RequestID, testEnv.HTTPRequest.RequestID)
+ }
+ if decodedEnv.HTTPRequest.Method != testEnv.HTTPRequest.Method {
+ t.Errorf("Method mismatch: got %v, want %v", decodedEnv.HTTPRequest.Method, testEnv.HTTPRequest.Method)
+ }
+ if decodedEnv.HTTPRequest.URL != testEnv.HTTPRequest.URL {
+ t.Errorf("URL mismatch: got %v, want %v", decodedEnv.HTTPRequest.URL, testEnv.HTTPRequest.URL)
+ }
+ if !bytes.Equal(decodedEnv.HTTPRequest.Body, testEnv.HTTPRequest.Body) {
+ t.Errorf("Body mismatch: got %v, want %v", decodedEnv.HTTPRequest.Body, testEnv.HTTPRequest.Body)
+ }
+}
+
+// TestProtobufCodecStreamData tests encoding/decoding of StreamData messages
+func TestProtobufCodecStreamData(t *testing.T) {
+ codec := protobufCodec{}
+ conn := newMockDatagramConn()
+
+ // Create a StreamData envelope
+ testEnv := &Envelope{
+ Type: MessageTypeStreamData,
+ StreamData: &StreamData{
+ ID: StreamID("stream-123"),
+ Seq: 42,
+ Data: []byte("stream data payload"),
+ },
+ }
+
+ // Encode
+ if err := codec.Encode(conn, testEnv); err != nil {
+ t.Fatalf("Failed to encode StreamData: %v", err)
+ }
+
+ // Verify single message
+ if len(conn.messages) != 1 {
+ t.Fatalf("Expected 1 message, got %d", len(conn.messages))
+ }
+
+ // Decode using a buffered reader (as we do in actual code)
+ reader := bufio.NewReaderSize(conn, GetDTLSReadBufferSize())
+ var decodedEnv Envelope
+ if err := codec.Decode(reader, &decodedEnv); err != nil {
+ t.Fatalf("Failed to decode StreamData: %v", err)
+ }
+
+ // Verify
+ if decodedEnv.Type != MessageTypeStreamData {
+ t.Errorf("Type mismatch: got %v, want %v", decodedEnv.Type, MessageTypeStreamData)
+ }
+ if decodedEnv.StreamData == nil {
+ t.Fatal("StreamData is nil")
+ }
+ if decodedEnv.StreamData.ID != testEnv.StreamData.ID {
+ t.Errorf("StreamID mismatch: got %v, want %v", decodedEnv.StreamData.ID, testEnv.StreamData.ID)
+ }
+ if decodedEnv.StreamData.Seq != testEnv.StreamData.Seq {
+ t.Errorf("Seq mismatch: got %v, want %v", decodedEnv.StreamData.Seq, testEnv.StreamData.Seq)
+ }
+ if !bytes.Equal(decodedEnv.StreamData.Data, testEnv.StreamData.Data) {
+ t.Errorf("Data mismatch: got %v, want %v", decodedEnv.StreamData.Data, testEnv.StreamData.Data)
+ }
+}
+
+// TestProtobufCodecMultipleMessages tests encoding/decoding multiple messages
+func TestProtobufCodecMultipleMessages(t *testing.T) {
+ codec := protobufCodec{}
+ conn := newMockDatagramConn()
+
+ // Create multiple test envelopes
+ envelopes := []*Envelope{
+ {
+ Type: MessageTypeStreamOpen,
+ StreamOpen: &StreamOpen{
+ ID: StreamID("stream-1"),
+ Service: "test-service",
+ TargetAddr: "127.0.0.1:8080",
+ },
+ },
+ {
+ Type: MessageTypeStreamData,
+ StreamData: &StreamData{
+ ID: StreamID("stream-1"),
+ Seq: 1,
+ Data: []byte("first chunk"),
+ },
+ },
+ {
+ Type: MessageTypeStreamData,
+ StreamData: &StreamData{
+ ID: StreamID("stream-1"),
+ Seq: 2,
+ Data: []byte("second chunk"),
+ },
+ },
+ {
+ Type: MessageTypeStreamClose,
+ StreamClose: &StreamClose{
+ ID: StreamID("stream-1"),
+ Error: "",
+ },
+ },
+ }
+
+ // Encode all messages
+ for i, env := range envelopes {
+ if err := codec.Encode(conn, env); err != nil {
+ t.Fatalf("Failed to encode message %d: %v", i, err)
+ }
+ }
+
+ // Verify that each encode produced exactly one message
+ if len(conn.messages) != len(envelopes) {
+ t.Fatalf("Expected %d messages, got %d", len(envelopes), len(conn.messages))
+ }
+
+ // Decode and verify all messages using a buffered reader (as we do in actual code)
+ reader := bufio.NewReaderSize(conn, GetDTLSReadBufferSize())
+ for i := 0; i < len(envelopes); i++ {
+ var decoded Envelope
+ if err := codec.Decode(reader, &decoded); err != nil {
+ t.Fatalf("Failed to decode message %d: %v", i, err)
+ }
+ if decoded.Type != envelopes[i].Type {
+ t.Errorf("Message %d type mismatch: got %v, want %v", i, decoded.Type, envelopes[i].Type)
+ }
+ }
+}
diff --git a/internal/protocol/hopgate_stream.proto b/internal/protocol/hopgate_stream.proto
index 09eb5c8..c16b7c8 100644
--- a/internal/protocol/hopgate_stream.proto
+++ b/internal/protocol/hopgate_stream.proto
@@ -2,7 +2,7 @@ syntax = "proto3";
package hopgate.protocol.v1;
-option go_package = "github.com/dalbodeule/hop-gate/internal/protocol/pb;protocolpb";
+option go_package = "internal/protocol/pb;pb";
// HeaderValues 는 HTTP 헤더의 다중 값 표현을 위한 래퍼입니다.
// HeaderValues wraps multiple header values for a single HTTP header key.
diff --git a/internal/protocol/pb/hopgate_stream.pb.go b/internal/protocol/pb/hopgate_stream.pb.go
index 2e2b18a..5654c5d 100644
--- a/internal/protocol/pb/hopgate_stream.pb.go
+++ b/internal/protocol/pb/hopgate_stream.pb.go
@@ -718,7 +718,7 @@ const file_internal_protocol_hopgate_stream_proto_rawDesc = "" +
"\fstream_close\x18\x05 \x01(\v2 .hopgate.protocol.v1.StreamCloseH\x00R\vstreamClose\x12?\n" +
"\n" +
"stream_ack\x18\x06 \x01(\v2\x1e.hopgate.protocol.v1.StreamAckH\x00R\tstreamAckB\t\n" +
- "\apayloadB@Z>github.com/dalbodeule/hop-gate/internal/protocol/pb;protocolpbb\x06proto3"
+ "\apayloadB\x19Z\x17internal/protocol/pb;pbb\x06proto3"
var (
file_internal_protocol_hopgate_stream_proto_rawDescOnce sync.Once
diff --git a/internal/protocol/pb/hopgate_stream_grpc.go b/internal/protocol/pb/hopgate_stream_grpc.go
new file mode 100644
index 0000000..36e403e
--- /dev/null
+++ b/internal/protocol/pb/hopgate_stream_grpc.go
@@ -0,0 +1,119 @@
+package pb
+
+import (
+ "context"
+
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+)
+
+// HopGateTunnelClient is the client API for the HopGateTunnel service.
+type HopGateTunnelClient interface {
+ // OpenTunnel establishes a long-lived bi-directional stream between
+ // a HopGate client and the server. Both HTTP requests and responses
+ // are multiplexed as Envelope messages on this stream.
+ OpenTunnel(ctx context.Context, opts ...grpc.CallOption) (HopGateTunnel_OpenTunnelClient, error)
+}
+
+type hopGateTunnelClient struct {
+ cc grpc.ClientConnInterface
+}
+
+// NewHopGateTunnelClient creates a new HopGateTunnelClient.
+func NewHopGateTunnelClient(cc grpc.ClientConnInterface) HopGateTunnelClient {
+ return &hopGateTunnelClient{cc: cc}
+}
+
+func (c *hopGateTunnelClient) OpenTunnel(ctx context.Context, opts ...grpc.CallOption) (HopGateTunnel_OpenTunnelClient, error) {
+ stream, err := c.cc.NewStream(ctx, &_HopGateTunnel_serviceDesc.Streams[0], "/hopgate.protocol.v1.HopGateTunnel/OpenTunnel", opts...)
+ if err != nil {
+ return nil, err
+ }
+ return &hopGateTunnelOpenTunnelClient{ClientStream: stream}, nil
+}
+
+// HopGateTunnel_OpenTunnelClient is the client-side stream for OpenTunnel.
+type HopGateTunnel_OpenTunnelClient interface {
+ Send(*Envelope) error
+ Recv() (*Envelope, error)
+ grpc.ClientStream
+}
+
+type hopGateTunnelOpenTunnelClient struct {
+ grpc.ClientStream
+}
+
+func (x *hopGateTunnelOpenTunnelClient) Send(m *Envelope) error {
+ return x.ClientStream.SendMsg(m)
+}
+
+func (x *hopGateTunnelOpenTunnelClient) Recv() (*Envelope, error) {
+ m := new(Envelope)
+ if err := x.ClientStream.RecvMsg(m); err != nil {
+ return nil, err
+ }
+ return m, nil
+}
+
+// HopGateTunnelServer is the server API for the HopGateTunnel service.
+type HopGateTunnelServer interface {
+ // OpenTunnel handles a long-lived bi-directional stream between the server
+ // and a HopGate client. Implementations are responsible for reading and
+ // writing Envelope messages on the stream.
+ OpenTunnel(HopGateTunnel_OpenTunnelServer) error
+}
+
+// UnimplementedHopGateTunnelServer can be embedded to have forward compatible implementations.
+type UnimplementedHopGateTunnelServer struct{}
+
+// OpenTunnel returns an Unimplemented error by default.
+func (UnimplementedHopGateTunnelServer) OpenTunnel(HopGateTunnel_OpenTunnelServer) error {
+ return status.Errorf(codes.Unimplemented, "method OpenTunnel not implemented")
+}
+
+// RegisterHopGateTunnelServer registers the HopGateTunnel service with the given gRPC server.
+func RegisterHopGateTunnelServer(s grpc.ServiceRegistrar, srv HopGateTunnelServer) {
+ s.RegisterService(&_HopGateTunnel_serviceDesc, srv)
+}
+
+// HopGateTunnel_OpenTunnelServer is the server-side stream for OpenTunnel.
+type HopGateTunnel_OpenTunnelServer interface {
+ Send(*Envelope) error
+ Recv() (*Envelope, error)
+ grpc.ServerStream
+}
+
+func _HopGateTunnel_OpenTunnel_Handler(srv interface{}, stream grpc.ServerStream) error {
+ return srv.(HopGateTunnelServer).OpenTunnel(&hopGateTunnelOpenTunnelServer{ServerStream: stream})
+}
+
+type hopGateTunnelOpenTunnelServer struct {
+ grpc.ServerStream
+}
+
+func (x *hopGateTunnelOpenTunnelServer) Send(m *Envelope) error {
+ return x.ServerStream.SendMsg(m)
+}
+
+func (x *hopGateTunnelOpenTunnelServer) Recv() (*Envelope, error) {
+ m := new(Envelope)
+ if err := x.ServerStream.RecvMsg(m); err != nil {
+ return nil, err
+ }
+ return m, nil
+}
+
+var _HopGateTunnel_serviceDesc = grpc.ServiceDesc{
+ ServiceName: "hopgate.protocol.v1.HopGateTunnel",
+ HandlerType: (*HopGateTunnelServer)(nil),
+ Streams: []grpc.StreamDesc{
+ {
+ StreamName: "OpenTunnel",
+ Handler: _HopGateTunnel_OpenTunnel_Handler,
+ ServerStreams: true,
+ ClientStreams: true,
+ },
+ },
+ Metadata: "internal/protocol/hopgate_stream.proto",
+}
diff --git a/internal/proxy/client.go b/internal/proxy/client.go
index c43be7a..91108e2 100644
--- a/internal/proxy/client.go
+++ b/internal/proxy/client.go
@@ -1,6 +1,7 @@
package proxy
import (
+ "bufio"
"bytes"
"context"
"fmt"
@@ -13,7 +14,6 @@ import (
"sync"
"time"
- "github.com/dalbodeule/hop-gate/internal/dtls"
"github.com/dalbodeule/hop-gate/internal/logging"
"github.com/dalbodeule/hop-gate/internal/protocol"
)
@@ -117,6 +117,432 @@ func (s *streamSender) handleAck(ack *protocol.StreamAck) map[uint64][]byte {
return lost
}
+// streamReceiver 는 단일 스트림(ID)에 대한 클라이언트 측 수신 상태와
+// 로컬 HTTP 매핑을 담당하는 per-stream 구조체 설계입니다. (ko)
+// streamReceiver is the per-stream receiver that owns client-side RX state
+// and local HTTP mapping for a single stream ID. (en)
+//
+// 3.3B.2 설계 포인트:
+// - 중앙 readLoop(StartLoop)는 DTLS 세션에서 Envelope 만 읽고,
+// streamReceiver.inCh 로 `StreamOpen/StreamData/StreamClose` 를 전달합니다.
+// - streamReceiver 는 자신에게 전달된 Envelope 들만 사용해
+// - 수신 ARQ(expectedSeq/received/lost) 를 관리하고,
+// - HTTP 요청/응답을 구성해 역방향 StreamOpen/StreamData/StreamClose 를 전송합니다.
+// - 실제 run 로직 및 StartLoop 와의 통합은 3.3B.3 단계에서 구현할 예정입니다.
+type streamReceiver struct {
+ // 이 수신기가 담당하는 스트림 ID.
+ id protocol.StreamID
+
+ // 수신 ARQ 상태: per-stream 시퀀스 및 out-of-order 버퍼/누락 집합. (ko)
+ // Receive-side ARQ state: per-stream sequence and out-of-order/lost sets. (en)
+ expectedSeq uint64
+ received map[uint64][]byte
+ lost map[uint64]struct{}
+
+ // 중앙 readLoop → per-stream goroutine 으로 전달되는 입력 채널. (ko)
+ // Input channel for envelopes dispatched from the central readLoop. (en)
+ inCh chan *protocol.Envelope
+
+ // 세션(write 방향) 및 직렬화 codec / 로깅 핸들. (ko)
+ // Session (write side only), wire codec and logging handles. (en)
+ sess io.ReadWriter
+ codec protocol.WireCodec
+ logger logging.Logger
+
+ // 로컬 HTTP 클라이언트 및 타깃 주소 정보. (ko)
+ // Local HTTP client and target information. (en)
+ HTTPClient *http.Client
+ LocalTarget string
+}
+
+// newStreamReceiver 는 단일 스트림 ID 에 대한 수신 상태/HTTP 매핑을 담당하는
+// streamReceiver 인스턴스를 초기화합니다. (ko)
+// newStreamReceiver initializes a streamReceiver for a single stream ID. (en)
+func newStreamReceiver(
+ id protocol.StreamID,
+ sess io.ReadWriter,
+ codec protocol.WireCodec,
+ logger logging.Logger,
+ httpClient *http.Client,
+ localTarget string,
+) *streamReceiver {
+ if codec == nil {
+ codec = protocol.DefaultCodec
+ }
+ return &streamReceiver{
+ id: id,
+ expectedSeq: 0,
+ received: make(map[uint64][]byte),
+ lost: make(map[uint64]struct{}),
+ inCh: make(chan *protocol.Envelope, 16),
+ sess: sess,
+ codec: codec,
+ logger: logger,
+ HTTPClient: httpClient,
+ LocalTarget: localTarget,
+ }
+}
+
+// run 은 단일 스트림에 대해 서버→클라이언트 방향 프레임을 처리하고,
+// 로컬 HTTP 요청/응답을 수행한 뒤, 클라이언트→서버 방향 스트림 응답을
+// 전송하는 수명주기 전담 루프입니다. (ko)
+// run is the per-stream lifecycle loop that consumes inbound frames,
+// performs the local HTTP request/response, and sends the reverse stream
+// back to the server. (en)
+func (r *streamReceiver) run(ctx context.Context, so *protocol.StreamOpen, sender *streamSender) error {
+ codec := r.codec
+ if codec == nil {
+ codec = protocol.DefaultCodec
+ }
+ log := r.logger
+ if log == nil {
+ log = logging.NewStdJSONLogger("client_proxy_stream_receiver")
+ }
+
+ streamID := r.id
+
+ // Pseudo-header 에서 HTTP 메타데이터를 추출합니다. (ko)
+ // Extract HTTP metadata from pseudo-headers. (en)
+ method := firstHeaderValue(so.Header, protocol.HeaderKeyMethod, http.MethodGet)
+ urlStr := firstHeaderValue(so.Header, protocol.HeaderKeyURL, "/")
+ _ = firstHeaderValue(so.Header, protocol.HeaderKeyHost, "")
+
+ if r.LocalTarget == "" {
+ return fmt.Errorf("local target is empty")
+ }
+
+ u, err := url.Parse(urlStr)
+ if err != nil {
+ return fmt.Errorf("parse url from stream_open: %w", err)
+ }
+ u.Scheme = "http"
+ u.Host = r.LocalTarget
+
+ // 로컬 HTTP 요청용 헤더 맵을 생성하면서 pseudo-header 는 제거합니다. (ko)
+ // Build local HTTP header map while stripping pseudo-headers. (en)
+ httpHeader := make(http.Header, len(so.Header))
+ for k, vs := range so.Header {
+ if k == protocol.HeaderKeyMethod ||
+ k == protocol.HeaderKeyURL ||
+ k == protocol.HeaderKeyHost ||
+ k == protocol.HeaderKeyStatus {
+ continue
+ }
+ for _, v := range vs {
+ httpHeader.Add(k, v)
+ }
+ }
+
+ // 요청 바디를 StreamData/StreamClose 프레임에서 모두 읽어 메모리에 적재합니다. (ko)
+ // Read the entire request body from StreamData/StreamClose frames into memory. (en)
+ //
+ // 동시에 수신 측 ARQ 상태(expectedSeq / out-of-order 버퍼 / LostSeqs)를 관리하고
+ // StreamAck 를 전송해 선택적 재전송(Selective Retransmission)을 유도합니다.
+ var bodyBuf bytes.Buffer
+ const maxLostReport = 32
+
+ for {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case env, ok := <-r.inCh:
+ if !ok {
+ return fmt.Errorf("stream receiver channel closed before stream_close")
+ }
+
+ switch env.Type {
+ case protocol.MessageTypeStreamData:
+ sd := env.StreamData
+ if sd == nil {
+ return fmt.Errorf("stream_data payload is nil")
+ }
+ if sd.ID != streamID {
+ return fmt.Errorf("stream_data for unexpected stream id %q (expected %q)", sd.ID, streamID)
+ }
+
+ // 수신 측 ARQ: Seq 에 따라 분기
+ switch {
+ case sd.Seq == r.expectedSeq:
+ // 기대하던 순서의 프레임: 바로 bodyBuf 에 기록하고, 이후 버퍼된 연속 프레임도 flush.
+ if len(sd.Data) > 0 {
+ if _, err := bodyBuf.Write(sd.Data); err != nil {
+ return fmt.Errorf("buffer stream_data: %w", err)
+ }
+ }
+ r.expectedSeq++
+ for {
+ data, ok := r.received[r.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(r.received, r.expectedSeq)
+ delete(r.lost, r.expectedSeq)
+ r.expectedSeq++
+ }
+
+ // AckSeq 이전 구간의 lost 항목 정리
+ for seq := range r.lost {
+ if seq < r.expectedSeq {
+ delete(r.lost, seq)
+ }
+ }
+
+ case sd.Seq > r.expectedSeq:
+ // 앞선 일부 Seq 들이 누락된 상태: 현재 프레임을 버퍼링하고 missing seq 들을 lost 에 추가.
+ if len(sd.Data) > 0 {
+ buf := make([]byte, len(sd.Data))
+ copy(buf, sd.Data)
+ r.received[sd.Seq] = buf
+ }
+ for seq := r.expectedSeq; seq < sd.Seq && len(r.lost) < maxLostReport; seq++ {
+ if _, ok := r.lost[seq]; !ok {
+ r.lost[seq] = struct{}{}
+ }
+ }
+
+ default:
+ // sd.Seq < expectedSeq 인 경우: 이미 처리했거나 Ack 로 커버된 프레임 → 무시.
+ }
+
+ // 수신 측 StreamAck 전송:
+ // - AckSeq: 0부터 시작해 연속으로 수신 완료한 마지막 시퀀스 (expectedSeq-1)
+ // - LostSeqs: 현재 윈도우 내에서 누락된 시퀀스 중 상한 개수(maxLostReport)까지만 포함
+ var ackSeq uint64
+ if r.expectedSeq == 0 {
+ ackSeq = 0
+ } else {
+ ackSeq = r.expectedSeq - 1
+ }
+
+ lostSeqs := make([]uint64, 0, len(r.lost))
+ for seq := range r.lost {
+ if seq >= r.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(r.sess, &ackEnv); err != nil {
+ return fmt.Errorf("send stream ack: %w", err)
+ }
+
+ case protocol.MessageTypeStreamClose:
+ sc := env.StreamClose
+ if sc == nil {
+ return fmt.Errorf("stream_close payload is nil")
+ }
+ if sc.ID != streamID {
+ return fmt.Errorf("stream_close for unexpected stream id %q (expected %q)", sc.ID, streamID)
+ }
+ // sc.Error 는 최소 구현에서는 로컬 요청 에러와 별도로 취급하지 않습니다. (ko)
+ // For the minimal implementation we do not surface sc.Error here. (en)
+ goto haveBody
+
+ default:
+ return fmt.Errorf("unexpected envelope type %q while reading stream request body", env.Type)
+ }
+ }
+ }
+
+haveBody:
+ bodyBytes := bodyBuf.Bytes()
+
+ // 로컬 HTTP 요청 생성 (stream 기반 요청을 실제 HTTP 요청으로 변환). (ko)
+ // Build the local HTTP request from the stream-based metadata and body. (en)
+ req, err := http.NewRequestWithContext(ctx, method, u.String(), nil)
+ if err != nil {
+ return fmt.Errorf("create http request from stream: %w", err)
+ }
+ if len(bodyBytes) > 0 {
+ buf := bytes.NewReader(bodyBytes)
+ req.Body = io.NopCloser(buf)
+ req.ContentLength = int64(len(bodyBytes))
+ }
+ req.Header = httpHeader
+
+ start := time.Now()
+ logReq := log.With(logging.Fields{
+ "request_id": string(streamID),
+ "service": so.Service,
+ "method": method,
+ "url": urlStr,
+ "stream_id": string(streamID),
+ "local_target": r.LocalTarget,
+ })
+ logReq.Info("received stream_open envelope from server", nil)
+
+ res, err := r.HTTPClient.Do(req)
+ if err != nil {
+ // 로컬 요청 실패 시, 502 + 에러 메시지를 스트림 응답으로 전송합니다. (ko)
+ // On local request failure, send a 502 response over the stream. (en)
+ errMsg := fmt.Sprintf("perform http request: %v", err)
+ streamRespHeader := map[string][]string{
+ "Content-Type": {"text/plain; charset=utf-8"},
+ protocol.HeaderKeyStatus: {strconv.Itoa(http.StatusBadGateway)},
+ }
+ respOpen := protocol.Envelope{
+ Type: protocol.MessageTypeStreamOpen,
+ StreamOpen: &protocol.StreamOpen{
+ ID: streamID,
+ Service: so.Service,
+ TargetAddr: so.TargetAddr,
+ Header: streamRespHeader,
+ },
+ }
+ if err2 := codec.Encode(r.sess, &respOpen); err2 != nil {
+ logReq.Error("failed to encode stream response open envelope (error path)", logging.Fields{
+ "error": err2.Error(),
+ })
+ return err2
+ }
+
+ dataEnv := protocol.Envelope{
+ Type: protocol.MessageTypeStreamData,
+ StreamData: &protocol.StreamData{
+ ID: streamID,
+ Seq: 0,
+ Data: []byte("HopGate: " + errMsg),
+ },
+ }
+ // 에러 응답 프레임도 ARQ 대상에 등록합니다.
+ sender.register(0, dataEnv.StreamData.Data)
+ if err2 := codec.Encode(r.sess, &dataEnv); err2 != nil {
+ logReq.Error("failed to encode stream response data envelope (error path)", logging.Fields{
+ "error": err2.Error(),
+ })
+ return err2
+ }
+
+ closeEnv := protocol.Envelope{
+ Type: protocol.MessageTypeStreamClose,
+ StreamClose: &protocol.StreamClose{
+ ID: streamID,
+ Error: errMsg,
+ },
+ }
+ if err2 := codec.Encode(r.sess, &closeEnv); err2 != nil {
+ logReq.Error("failed to encode stream response close envelope (error path)", logging.Fields{
+ "error": err2.Error(),
+ })
+ return err2
+ }
+
+ logReq.Error("local http request failed (stream)", logging.Fields{
+ "error": err.Error(),
+ })
+ return nil
+ }
+ defer res.Body.Close()
+
+ // 응답을 StreamOpen + StreamData(4KiB chunk) + StreamClose 프레임으로 전송합니다. (ko)
+ // Send the response as StreamOpen + StreamData (4KiB chunks) + StreamClose frames. (en)
+
+ // 응답 헤더 맵을 복사하고 상태 코드를 pseudo-header 로 추가합니다. (ko)
+ // Copy response headers and attach status code as a pseudo-header. (en)
+ streamRespHeader := make(map[string][]string, len(res.Header)+1)
+ for k, vs := range res.Header {
+ streamRespHeader[k] = append([]string(nil), vs...)
+ }
+ statusCode := res.StatusCode
+ if statusCode == 0 {
+ statusCode = http.StatusOK
+ }
+ streamRespHeader[protocol.HeaderKeyStatus] = []string{strconv.Itoa(statusCode)}
+
+ respOpen := protocol.Envelope{
+ Type: protocol.MessageTypeStreamOpen,
+ StreamOpen: &protocol.StreamOpen{
+ ID: streamID,
+ Service: so.Service,
+ TargetAddr: so.TargetAddr,
+ Header: streamRespHeader,
+ },
+ }
+
+ if err := codec.Encode(r.sess, &respOpen); err != nil {
+ logReq.Error("failed to encode stream response open envelope", logging.Fields{
+ "error": err.Error(),
+ })
+ return err
+ }
+
+ // 응답 바디를 4KiB(StreamChunkSize) 단위로 잘라 StreamData 프레임으로 전송합니다. (ko)
+ // Chunk the response body into 4KiB (StreamChunkSize) StreamData frames. (en)
+ var seq uint64
+ chunk := make([]byte, protocol.StreamChunkSize)
+ for {
+ 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{
+ ID: streamID,
+ Seq: seq,
+ Data: dataCopy,
+ },
+ }
+ if err2 := codec.Encode(r.sess, &dataEnv); err2 != nil {
+ logReq.Error("failed to encode stream response data envelope", logging.Fields{
+ "error": err2.Error(),
+ })
+ return err2
+ }
+ seq++
+ }
+ if err == io.EOF {
+ break
+ }
+ if err != nil {
+ return fmt.Errorf("read http response body for streaming: %w", err)
+ }
+ }
+
+ closeEnv := protocol.Envelope{
+ Type: protocol.MessageTypeStreamClose,
+ StreamClose: &protocol.StreamClose{
+ ID: streamID,
+ Error: "",
+ },
+ }
+
+ if err := codec.Encode(r.sess, &closeEnv); err != nil {
+ logReq.Error("failed to encode stream response close envelope", logging.Fields{
+ "error": err.Error(),
+ })
+ return err
+ }
+
+ logReq.Info("stream http response sent to server", logging.Fields{
+ "status": statusCode,
+ "elapsed_ms": time.Since(start).Milliseconds(),
+ "error": "",
+ })
+
+ return nil
+}
+
func (p *ClientProxy) registerStreamSender(id protocol.StreamID, sender *streamSender) {
p.sendersMu.Lock()
defer p.sendersMu.Unlock()
@@ -144,21 +570,90 @@ func (p *ClientProxy) getStreamSender(id protocol.StreamID) *streamSender {
return p.streamSenders[id]
}
-func (p *ClientProxy) StartLoop(ctx context.Context, sess dtls.Session) error {
+// StartLoop 는 단일 DTLS 세션에 대한 **중앙 readLoop** 역할을 수행합니다. (ko)
+// StartLoop acts as the **central read loop** for a single DTLS session. (en)
+//
+// 3.3B.1 Design note — client-side DTLS session multiplexing:
+//
+// - 목표:
+// - DTLS 세션 레벨에서는 오직 `protocol.Envelope` 를 연속해서 읽고(decoding),
+// 각 Envelope 를 **스트림 단위로 demux** 하는 역할만 맡습니다.
+// - 실제 HTTP 처리(요청 바디 수신, 로컬 HTTP 호출, 응답 스트림 전송)는
+// 개별 스트림 전용 goroutine/구조체(`streamReceiver` 등)가 담당하도록 분리합니다.
+//
+// - 스트림 demux 자료구조(계획):
+// - `recvTable: map[protocol.StreamID]*streamReceiver` 형태의 수신 테이블을 유지합니다.
+// - 각 `streamReceiver` 는 자신만의 입력 채널을 가집니다. 예: `inCh chan *protocol.Envelope`.
+// - 중앙 readLoop 는 DTLS 세션에서 Envelope 를 읽은 뒤,
+// - `env.Type == MessageTypeStreamOpen` 인 경우:
+// - `id := env.StreamOpen.ID` 로 stream ID 를 구하고,
+// - `recvTable[id]` 가 없으면 새 `streamReceiver` 를 생성해 goroutine 을 띄운 뒤
+// 첫 메시지(`env`)를 `receiver.inCh <- env` 로 전달합니다.
+// - `env.Type == MessageTypeStreamData` / `MessageTypeStreamClose` 인 경우:
+// - `id := env.StreamData.ID` 또는 `env.StreamClose.ID` 로 stream ID 를 구하고,
+// - 기존 `recvTable[id]` 를 찾아 `receiver.inCh <- env` 로 전달합니다.
+// - receiver 가 존재하지 않으면 해당 스트림에 한정된 프로토콜 에러로 처리할지 정책을 정의합니다.
+// - `env.Type == MessageTypeStreamAck` 인 경우:
+// - 이미 구현된 송신 측 ARQ 테이블(`streamSenders`)을 조회해 재전송 로직에 전달합니다.
+//
+// - 현재 구현 상태와 향후 리팩터링 경계:
+// - 지금은 `MessageTypeStreamOpen` 을 수신하면 곧바로 `handleStreamRequest` 를 호출하고,
+// 이 함수가 `reader` 를 직접 소비하면서 같은 세션 안에 **동시에 하나의 스트림만** 처리할 수 있습니다.
+// - 3.3B.2 / 3.3B.3 단계에서는 위에서 설명한 demux 설계에 맞춰
+// - `handleStreamRequest` 내부 HTTP 매핑 로직을 `streamReceiver` 로 옮기고,
+// - StartLoop 가 DTLS 세션 → per-stream goroutine 으로 이벤트를 분배하는 역할만 수행하도록
+// 점진적으로 리팩터링할 예정입니다.
+func (p *ClientProxy) StartLoop(ctx context.Context, sess io.ReadWriter) error {
if ctx == nil {
ctx = context.Background()
}
log := p.Logger
// NOTE: pion/dtls 는 복호화된 애플리케이션 데이터를 호출자가 제공한 버퍼에 채워 넣습니다.
- // 기본 JSON 디코더 버퍼(수백 바이트 수준)만 사용하면 큰 HTTP 바디/Envelope 에서
- // "dtls: buffer too small" 오류가 날 수 있으므로, 여기서는 여유 있는 버퍼(64KiB)를 사용합니다. (ko)
+ // DTLS는 UDP 기반이므로 한 번의 Read()에서 전체 datagram을 읽어야 하며,
+ // pion/dtls 내부 버퍼 한계(8KB)를 초과하는 메시지는 "dtls: buffer too small" 오류를 발생시킵니다.
+ // 이를 방지하기 위해 DTLS 세션을 bufio.Reader로 감싸서 datagram을 완전히 읽어들인 후 파싱합니다. (ko)
// NOTE: pion/dtls decrypts application data into the buffer provided by the caller.
- // Using only the default JSON decoder buffer (a few hundred bytes) can trigger
- // "dtls: buffer too small" for large HTTP bodies/envelopes. The default
- // JSON-based WireCodec internally wraps the DTLS session with a 64KiB
- // bufio.Reader, matching this requirement. (en)
+ // Since DTLS is UDP-based, the entire datagram must be read in a single Read() call,
+ // and messages exceeding pion/dtls's internal buffer limit (8KB) will trigger
+ // "dtls: buffer too small" errors. To prevent this, we wrap the DTLS session with
+ // a bufio.Reader to fully read the datagram before parsing. (en)
codec := protocol.DefaultCodec
+ bufferedReader := bufio.NewReaderSize(sess, protocol.GetDTLSReadBufferSize())
+
+ // 스트림 수신기 테이블: 중앙 readLoop 가 StreamOpen/Data/Close 를
+ // 각 streamReceiver 로 demux 하기 위해 사용합니다. (ko)
+ // Per-session stream receiver table used by the central read loop to
+ // demultiplex StreamOpen/Data/Close frames. (en)
+ receivers := make(map[protocol.StreamID]*streamReceiver)
+ var receiversMu sync.Mutex
+
+ getReceiver := func(id protocol.StreamID) *streamReceiver {
+ receiversMu.Lock()
+ defer receiversMu.Unlock()
+ return receivers[id]
+ }
+
+ addReceiver := func(id protocol.StreamID, rcv *streamReceiver) {
+ receiversMu.Lock()
+ receivers[id] = rcv
+ receiversMu.Unlock()
+ }
+
+ removeReceiver := func(id protocol.StreamID) {
+ receiversMu.Lock()
+ delete(receivers, id)
+ receiversMu.Unlock()
+ }
+
+ closeAllReceivers := func() {
+ receiversMu.Lock()
+ defer receiversMu.Unlock()
+ for id, rcv := range receivers {
+ close(rcv.inCh)
+ delete(receivers, id)
+ }
+ }
for {
select {
@@ -166,19 +661,22 @@ func (p *ClientProxy) StartLoop(ctx context.Context, sess dtls.Session) error {
log.Info("client proxy loop stopping due to context cancellation", logging.Fields{
"reason": ctx.Err().Error(),
})
+ closeAllReceivers()
return nil
default:
}
var env protocol.Envelope
- if err := codec.Decode(sess, &env); err != nil {
+ if err := codec.Decode(bufferedReader, &env); err != nil {
if err == io.EOF {
log.Info("dtls session closed by server", nil)
+ closeAllReceivers()
return nil
}
log.Error("failed to decode protocol envelope", logging.Fields{
"error": err.Error(),
})
+ closeAllReceivers()
return err
}
@@ -188,19 +686,17 @@ func (p *ClientProxy) StartLoop(ctx context.Context, sess dtls.Session) error {
log.Error("failed to handle http envelope", logging.Fields{
"error": err.Error(),
})
+ closeAllReceivers()
return err
}
- case protocol.MessageTypeStreamOpen:
- if err := p.handleStreamRequest(ctx, sess, &env); err != nil {
- log.Error("failed to handle stream http envelope", logging.Fields{
- "error": err.Error(),
- })
- return err
- }
+
case protocol.MessageTypeStreamAck:
+ // 송신 측 ARQ: 서버 → 클라이언트 응답 스트림에 대한 StreamAck 처리. (ko)
+ // Sender-side ARQ: handle StreamAck for response streams (server → client). (en)
sa := env.StreamAck
if sa == nil {
log.Error("received stream_ack envelope with nil payload", nil)
+ closeAllReceivers()
return fmt.Errorf("stream_ack payload is nil")
}
streamID := protocol.StreamID(sa.ID)
@@ -212,7 +708,8 @@ func (p *ClientProxy) StartLoop(ctx context.Context, sess dtls.Session) error {
continue
}
lost := sender.handleAck(sa)
- // LostSeqs 를 기반으로 선택적 재전송 수행
+ // LostSeqs 를 기반으로 선택적 재전송 수행 (Selective Retransmission). (ko)
+ // Perform selective retransmission based on LostSeqs. (en)
for seq, data := range lost {
retryEnv := protocol.Envelope{
Type: protocol.MessageTypeStreamData,
@@ -228,6 +725,7 @@ func (p *ClientProxy) StartLoop(ctx context.Context, sess dtls.Session) error {
"seq": seq,
"error": err.Error(),
})
+ closeAllReceivers()
return err
}
log.Info("retransmitted stream_data after stream_ack", logging.Fields{
@@ -235,10 +733,104 @@ func (p *ClientProxy) StartLoop(ctx context.Context, sess dtls.Session) error {
"seq": seq,
})
}
+
+ case protocol.MessageTypeStreamOpen:
+ // 새로운 스트림에 대한 수신기 생성 및 goroutine 실행. (ko)
+ // Create a new streamReceiver and start its goroutine for this stream. (en)
+ so := env.StreamOpen
+ if so == nil {
+ log.Error("stream_open envelope missing payload", nil)
+ continue
+ }
+ streamID := so.ID
+ if streamID == "" {
+ log.Error("stream_open with empty stream id", nil)
+ continue
+ }
+ if p.LocalTarget == "" {
+ closeAllReceivers()
+ return fmt.Errorf("local target is empty")
+ }
+
+ if existing := getReceiver(streamID); existing != nil {
+ log.Error("duplicate stream_open for existing stream", logging.Fields{
+ "stream_id": streamID,
+ })
+ continue
+ }
+
+ sender := newStreamSender()
+ p.registerStreamSender(streamID, sender)
+
+ receiver := newStreamReceiver(streamID, sess, codec, log, p.HTTPClient, p.LocalTarget)
+ addReceiver(streamID, receiver)
+
+ go func(id protocol.StreamID, r *streamReceiver, so *protocol.StreamOpen, snd *streamSender) {
+ if err := r.run(ctx, so, snd); err != nil {
+ log.Error("stream receiver terminated with error", logging.Fields{
+ "stream_id": id,
+ "error": err.Error(),
+ })
+ }
+ removeReceiver(id)
+ p.unregisterStreamSender(id)
+ }(streamID, receiver, so, sender)
+
+ case protocol.MessageTypeStreamData:
+ // StreamData 는 중앙 readLoop 에서 해당 streamReceiver 로 demux 됩니다. (ko)
+ // StreamData frames are demultiplexed to the corresponding streamReceiver. (en)
+ sd := env.StreamData
+ if sd == nil {
+ log.Error("stream_data envelope with nil payload", nil)
+ continue
+ }
+ streamID := sd.ID
+ receiver := getReceiver(streamID)
+ if receiver == nil {
+ log.Warn("received stream_data for unknown stream", logging.Fields{
+ "stream_id": streamID,
+ })
+ continue
+ }
+ envCopy := env
+ select {
+ case receiver.inCh <- &envCopy:
+ case <-ctx.Done():
+ closeAllReceivers()
+ return nil
+ }
+
+ case protocol.MessageTypeStreamClose:
+ // StreamClose 역시 중앙 readLoop 에서 해당 streamReceiver 로 전달합니다. (ko)
+ // StreamClose is also forwarded from the central readLoop to streamReceiver. (en)
+ sc := env.StreamClose
+ if sc == nil {
+ log.Error("stream_close envelope with nil payload", nil)
+ continue
+ }
+ streamID := sc.ID
+ receiver := getReceiver(streamID)
+ if receiver == nil {
+ log.Warn("received stream_close for unknown stream", logging.Fields{
+ "stream_id": streamID,
+ })
+ continue
+ }
+ envCopy := env
+ select {
+ case receiver.inCh <- &envCopy:
+ // 수명주기 정리는 receiver.run 내부와 goroutine 종료 시 removeReceiver 에서 수행됩니다. (ko)
+ // Lifecycle cleanup is handled inside receiver.run and the goroutine's defer. (en)
+ case <-ctx.Done():
+ closeAllReceivers()
+ return nil
+ }
+
default:
log.Error("received unsupported envelope type from server", logging.Fields{
"type": env.Type,
})
+ closeAllReceivers()
return fmt.Errorf("unsupported envelope type %q", env.Type)
}
}
@@ -246,7 +838,7 @@ func (p *ClientProxy) StartLoop(ctx context.Context, sess dtls.Session) error {
// handleHTTPEnvelope 는 기존 단일 HTTP 요청/응답 Envelope 경로를 처리합니다. (ko)
// handleHTTPEnvelope handles the legacy single HTTP request/response envelope path. (en)
-func (p *ClientProxy) handleHTTPEnvelope(ctx context.Context, sess dtls.Session, env *protocol.Envelope) error {
+func (p *ClientProxy) handleHTTPEnvelope(ctx context.Context, sess io.ReadWriter, env *protocol.Envelope) error {
if env.HTTPRequest == nil {
return fmt.Errorf("http envelope missing http_request payload")
}
@@ -303,7 +895,7 @@ func (p *ClientProxy) handleHTTPEnvelope(ctx context.Context, sess dtls.Session,
// handleStreamRequest 는 StreamOpen/StreamData/StreamClose 기반 HTTP 요청/응답 스트림을 처리합니다. (ko)
// handleStreamRequest handles an HTTP request/response exchange using StreamOpen/StreamData/StreamClose frames. (en)
-func (p *ClientProxy) handleStreamRequest(ctx context.Context, sess dtls.Session, openEnv *protocol.Envelope) error {
+func (p *ClientProxy) handleStreamRequest(ctx context.Context, sess io.ReadWriter, reader io.Reader, openEnv *protocol.Envelope) error {
codec := protocol.DefaultCodec
log := p.Logger
@@ -318,57 +910,37 @@ func (p *ClientProxy) handleStreamRequest(ctx context.Context, sess dtls.Session
p.registerStreamSender(streamID, sender)
defer p.unregisterStreamSender(streamID)
- // Pseudo-header 에서 HTTP 메타데이터를 추출합니다. (ko)
- // Extract HTTP metadata from pseudo-headers. (en)
- method := firstHeaderValue(so.Header, protocol.HeaderKeyMethod, http.MethodGet)
- urlStr := firstHeaderValue(so.Header, protocol.HeaderKeyURL, "/")
- _ = firstHeaderValue(so.Header, protocol.HeaderKeyHost, "")
-
if p.LocalTarget == "" {
return fmt.Errorf("local target is empty")
}
- u, err := url.Parse(urlStr)
- if err != nil {
- return fmt.Errorf("parse url from stream_open: %w", err)
- }
- u.Scheme = "http"
- u.Host = p.LocalTarget
+ // streamReceiver 를 생성해 스트림 수신/HTTP 매핑/응답 전송을 전담시킵니다. (ko)
+ // Delegate per-stream RX/HTTP mapping/response to a streamReceiver. (en)
+ receiver := newStreamReceiver(streamID, sess, codec, log, p.HTTPClient, p.LocalTarget)
- // 로컬 HTTP 요청용 헤더 맵을 생성하면서 pseudo-header 는 제거합니다. (ko)
- // Build local HTTP header map while stripping pseudo-headers. (en)
- httpHeader := make(http.Header, len(so.Header))
- for k, vs := range so.Header {
- if k == protocol.HeaderKeyMethod ||
- k == protocol.HeaderKeyURL ||
- k == protocol.HeaderKeyHost ||
- k == protocol.HeaderKeyStatus {
- continue
- }
- for _, v := range vs {
- httpHeader.Add(k, v)
- }
- }
-
- // 요청 바디를 StreamData/StreamClose 프레임에서 모두 읽어 메모리에 적재합니다. (ko)
- // Read the entire request body from StreamData/StreamClose frames into memory. (en)
- //
- // 동시에 수신 측 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
+ // streamReceiver 수명주기를 별도 goroutine 으로 실행합니다. (ko)
+ // Run the streamReceiver lifecycle in a separate goroutine. (en)
+ errCh := make(chan error, 1)
+ go func() {
+ errCh <- receiver.run(ctx, so, sender)
+ }()
for {
var env protocol.Envelope
- if err := codec.Decode(sess, &env); err != nil {
+ if err := codec.Decode(reader, &env); err != nil {
if err == io.EOF {
+ // DTLS 세션이 조기 종료되면 receiver 에게 더 이상 프레임이 없음을 알리고 종료를 기다립니다. (ko)
+ // On EOF, close the channel so receiver can terminate gracefully. (en)
+ close(receiver.inCh)
+ if recvErr := <-errCh; recvErr != nil {
+ return recvErr
+ }
return fmt.Errorf("unexpected EOF while reading stream request body")
}
+ close(receiver.inCh)
+ if recvErr := <-errCh; recvErr != nil {
+ return recvErr
+ }
return fmt.Errorf("decode stream request frame: %w", err)
}
@@ -376,291 +948,47 @@ func (p *ClientProxy) handleStreamRequest(ctx context.Context, sess dtls.Session
case protocol.MessageTypeStreamData:
sd := env.StreamData
if sd == nil {
+ close(receiver.inCh)
+ _ = <-errCh
return fmt.Errorf("stream_data payload is nil")
}
if sd.ID != streamID {
+ close(receiver.inCh)
+ _ = <-errCh
return fmt.Errorf("stream_data for unexpected stream id %q (expected %q)", sd.ID, streamID)
}
-
- // 수신 측 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)
- }
+ envCopy := env
+ receiver.inCh <- &envCopy
case protocol.MessageTypeStreamClose:
sc := env.StreamClose
if sc == nil {
+ close(receiver.inCh)
+ _ = <-errCh
return fmt.Errorf("stream_close payload is nil")
}
if sc.ID != streamID {
+ close(receiver.inCh)
+ _ = <-errCh
return fmt.Errorf("stream_close for unexpected stream id %q (expected %q)", sc.ID, streamID)
}
- // sc.Error 는 최소 구현에서는 로컬 요청 에러와 별도로 취급하지 않습니다. (ko)
- // For the minimal implementation we do not surface sc.Error here. (en)
- goto haveBody
+ // StreamClose 프레임을 receiver 에게 전달한 뒤 채널을 닫고 종료를 기다립니다. (ko)
+ // After forwarding StreamClose, close the channel and wait for receiver to finish. (en)
+ envCopy := env
+ receiver.inCh <- &envCopy
+ close(receiver.inCh)
+ return <-errCh
+
default:
+ // 예상치 못한 Envelope 타입: 해당 스트림에 한정된 프로토콜 에러로 보고 receiver 를 종료합니다. (ko)
+ // Unexpected envelope type: treat as per-stream protocol error and shut down receiver. (en)
+ close(receiver.inCh)
+ if recvErr := <-errCh; recvErr != nil {
+ return recvErr
+ }
return fmt.Errorf("unexpected envelope type %q while reading stream request body", env.Type)
}
}
-
-haveBody:
- bodyBytes := bodyBuf.Bytes()
-
- // 로컬 HTTP 요청 생성 (stream 기반 요청을 실제 HTTP 요청으로 변환). (ko)
- // Build the local HTTP request from the stream-based metadata and body. (en)
- req, err := http.NewRequestWithContext(ctx, method, u.String(), nil)
- if err != nil {
- return fmt.Errorf("create http request from stream: %w", err)
- }
- if len(bodyBytes) > 0 {
- buf := bytes.NewReader(bodyBytes)
- req.Body = io.NopCloser(buf)
- req.ContentLength = int64(len(bodyBytes))
- }
- req.Header = httpHeader
-
- start := time.Now()
- logReq := log.With(logging.Fields{
- "request_id": string(streamID),
- "service": so.Service,
- "method": method,
- "url": urlStr,
- "stream_id": string(streamID),
- "local_target": p.LocalTarget,
- })
- logReq.Info("received stream_open envelope from server", nil)
-
- res, err := p.HTTPClient.Do(req)
- if err != nil {
- // 로컬 요청 실패 시, 502 + 에러 메시지를 스트림 응답으로 전송합니다. (ko)
- // On local request failure, send a 502 response over the stream. (en)
- errMsg := fmt.Sprintf("perform http request: %v", err)
- streamRespHeader := map[string][]string{
- "Content-Type": {"text/plain; charset=utf-8"},
- protocol.HeaderKeyStatus: {strconv.Itoa(http.StatusBadGateway)},
- }
- respOpen := protocol.Envelope{
- Type: protocol.MessageTypeStreamOpen,
- StreamOpen: &protocol.StreamOpen{
- ID: streamID,
- Service: so.Service,
- TargetAddr: so.TargetAddr,
- Header: streamRespHeader,
- },
- }
- if err2 := codec.Encode(sess, &respOpen); err2 != nil {
- logReq.Error("failed to encode stream response open envelope (error path)", logging.Fields{
- "error": err2.Error(),
- })
- return err2
- }
-
- dataEnv := protocol.Envelope{
- Type: protocol.MessageTypeStreamData,
- StreamData: &protocol.StreamData{
- ID: streamID,
- Seq: 0,
- 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(),
- })
- return err2
- }
-
- closeEnv := protocol.Envelope{
- Type: protocol.MessageTypeStreamClose,
- StreamClose: &protocol.StreamClose{
- ID: streamID,
- Error: errMsg,
- },
- }
- if err2 := codec.Encode(sess, &closeEnv); err2 != nil {
- logReq.Error("failed to encode stream response close envelope (error path)", logging.Fields{
- "error": err2.Error(),
- })
- return err2
- }
-
- logReq.Error("local http request failed (stream)", logging.Fields{
- "error": err.Error(),
- })
- return nil
- }
- defer res.Body.Close()
-
- // 응답을 StreamOpen + StreamData(4KiB chunk) + StreamClose 프레임으로 전송합니다. (ko)
- // Send the response as StreamOpen + StreamData (4KiB chunks) + StreamClose frames. (en)
-
- // 응답 헤더 맵을 복사하고 상태 코드를 pseudo-header 로 추가합니다. (ko)
- // Copy response headers and attach status code as a pseudo-header. (en)
- streamRespHeader := make(map[string][]string, len(res.Header)+1)
- for k, vs := range res.Header {
- streamRespHeader[k] = append([]string(nil), vs...)
- }
- statusCode := res.StatusCode
- if statusCode == 0 {
- statusCode = http.StatusOK
- }
- streamRespHeader[protocol.HeaderKeyStatus] = []string{strconv.Itoa(statusCode)}
-
- respOpen := protocol.Envelope{
- Type: protocol.MessageTypeStreamOpen,
- StreamOpen: &protocol.StreamOpen{
- ID: streamID,
- Service: so.Service,
- TargetAddr: so.TargetAddr,
- Header: streamRespHeader,
- },
- }
-
- if err := codec.Encode(sess, &respOpen); err != nil {
- logReq.Error("failed to encode stream response open envelope", logging.Fields{
- "error": err.Error(),
- })
- return err
- }
-
- // 응답 바디를 4KiB(StreamChunkSize) 단위로 잘라 StreamData 프레임으로 전송합니다. (ko)
- // Chunk the response body into 4KiB (StreamChunkSize) StreamData frames. (en)
- var seq uint64
- chunk := make([]byte, protocol.StreamChunkSize)
- for {
- 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{
- ID: streamID,
- Seq: seq,
- Data: dataCopy,
- },
- }
- if err2 := codec.Encode(sess, &dataEnv); err2 != nil {
- logReq.Error("failed to encode stream response data envelope", logging.Fields{
- "error": err2.Error(),
- })
- return err2
- }
- seq++
- }
- if err == io.EOF {
- break
- }
- if err != nil {
- return fmt.Errorf("read http response body for streaming: %w", err)
- }
- }
-
- closeEnv := protocol.Envelope{
- Type: protocol.MessageTypeStreamClose,
- StreamClose: &protocol.StreamClose{
- ID: streamID,
- Error: "",
- },
- }
-
- if err := codec.Encode(sess, &closeEnv); err != nil {
- logReq.Error("failed to encode stream response close envelope", logging.Fields{
- "error": err.Error(),
- })
- return err
- }
-
- logReq.Info("stream http response sent to server", logging.Fields{
- "status": statusCode,
- "elapsed_ms": time.Since(start).Milliseconds(),
- "error": "",
- })
-
- return nil
}
// forwardToLocal 는 protocol.Request 를 로컬 HTTP 요청으로 변환하고 protocol.Response 를 채웁니다. (ko)
diff --git a/progress.md b/progress.md
index c396244..0826bd1 100644
--- a/progress.md
+++ b/progress.md
@@ -224,182 +224,46 @@ This document tracks implementation progress against the HopGate architecture an
---
-### 3.3 Proxy Core / HTTP Tunneling
+### 3.3 Proxy Core / gRPC Tunneling
-- [ ] 서버 측 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 항목과 연동)
+HopGate 의 최종 목표는 **TCP + TLS(HTTPS) + HTTP/2 + gRPC** 기반 터널로 HTTP 트래픽을 전달하는 것입니다.
+이 섹션에서는 DTLS 기반 초기 설계를 정리만 남기고, 실제 구현/남은 작업은 gRPC 터널 기준으로 재정의합니다.
-- [x] 클라이언트 측 Proxy 구현 확장: [`internal/proxy/client.go`](internal/proxy/client.go)
- - DTLS 세션에서 `protocol.Request` 수신 → 로컬 HTTP 호출 → `protocol.Response` 전송 루프 구현.
- - timeout/취소/에러 처리.
+- [x] 서버 측 gRPC 터널 엔드포인트 설계/구현
+ - 외부 사용자용 HTTPS(443/TCP)와 같은 포트에서:
+ - 일반 HTTP 요청(브라우저/REST)은 기존 리버스 프록시 경로로,
+ - `Content-Type: application/grpc` 인 요청은 클라이언트 터널용 gRPC 서버로
+ 라우팅하는 구조를 설계합니다.
+ - 예시: `rpc OpenTunnel(stream TunnelFrame) returns (stream TunnelFrame)` (bi-directional streaming).
+ - HTTP/2 + ALPN(h2)을 사용해 gRPC 스트림을 유지하고, 요청/응답 HTTP 메시지를 `TunnelFrame`으로 멀티플렉싱합니다.
-- [x] 서버 main 에 Proxy wiring 추가: [`cmd/server/main.go`](cmd/server/main.go)
- - DTLS handshake 완료된 세션을 Proxy 라우팅 테이블에 등록.
- - HTTPS 서버와 Proxy 핸들러 연결.
+- [x] 클라이언트 측 gRPC 터널 설계/구현
+ - 클라이언트 프로세스는 HopGate 서버로 장기 유지 bi-di gRPC 스트림을 **하나(또는 소수 개)** 연 상태로 유지합니다.
+ - 서버로부터 들어오는 `TunnelFrame`(요청 메타데이터 + 바디 chunk)을 수신해,
+ 로컬 HTTP 서비스(예: `127.0.0.1:8080`)로 proxy 하고, 응답을 다시 `TunnelFrame` 시퀀스로 전송합니다.
+ - 기존 `internal/proxy/client.go` 의 HTTP 매핑/스트림 ARQ 경험을, gRPC 메시지 단위 chunk/flow-control 설계에 참고합니다.
-- [x] 클라이언트 main 에 Proxy loop wiring 추가: [`cmd/client/main.go`](cmd/client/main.go)
- - handshake 성공 후 `proxy.ClientProxy.StartLoop` 실행.
+- [x] HTTP ↔ gRPC 터널 매핑 규약 정의
+ - 한 HTTP 요청/응답 쌍을 gRPC 스트림 상에서 어떻게 표현할지 스키마를 정의합니다:
+ - 요청: `StreamID`, method, URL, headers, body chunks
+ - 응답: `StreamID`, status, headers, body chunks, error
+ - 현재 `internal/protocol/protocol.go`의 논리 모델(Envelope/StreamOpen/StreamData/StreamClose/StreamAck)을
+ gRPC 메시지(oneof 필드 등)로 직렬화할지, 또는 새로운 gRPC 전용 메시지를 정의할지 결정합니다.
+ - Back-pressure / flow-control 은 gRPC/HTTP2의 스트림 flow-control 을 최대한 활용하고,
+ 추가 application-level windowing 이 필요하면 최소한으로만 도입합니다.
-#### 3.3A Stream-based DTLS Tunneling / 스트림 기반 DTLS 터널링
+- [ ] gRPC 터널 기반 E2E 플로우 정의/테스트 계획
+ - 하나의 gRPC 스트림 위에서:
+ - 동시에 여러 정적 리소스(`/css`, `/js`, `/img`) 요청,
+ - 큰 응답(수 MB 파일)과 작은 응답(API JSON)이 섞여 있는 시나리오,
+ - 클라이언트 재시작/네트워크 단절 후 재연결 시나리오
+ 를 포함하는 테스트 플랜을 작성합니다.
+ - 기대 동작:
+ - 느린 요청이 있더라도 다른 요청이 **같은 TCP 연결/스트림 집합 내에서** 과도하게 지연되지 않을 것.
+ - 서버/클라이언트 로그에 프로토콜 위반 경고(`unexpected frame ...`)가 발생하지 않을 것.
-초기 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:
-
-- 전송 계층은 DTLS(pion/dtls)를 유지합니다.
- The transport layer must remain DTLS (pion/dtls).
-- JSON 기반 단일 Envelope 모델에서 벗어나, HTTP 바디를 안전한 크기의 chunk 로 나누어 전송해야 합니다.
- We must move away from the single-envelope JSON model and chunk HTTP bodies under a safe MTU.
-- UDP 특성상 일부 프레임 손실/오염에 대비해, **해당 chunk 만 재전송 요청할 수 있는 ARQ 메커니즘**이 필요합니다.
- Given UDP characteristics, we need an application-level ARQ so that **only lost/corrupted chunks are retransmitted**.
-
-아래 단계들은 `feature/udp-stream` 브랜치에서 구현할 구체적인 작업 항목입니다.
-The following tasks describe concrete work items to be implemented on the `feature/udp-stream` branch.
-
----
-
-##### 3.3A.1 스트림 프레이밍 프로토콜 설계 (JSON 1단계)
-##### 3.3A.1 Stream framing protocol (JSON, phase 1)
-
-- [x] 스트림 프레임 타입 정리 및 확장: [`internal/protocol/protocol.go`](internal/protocol/protocol.go:35)
- - 이미 정의된 스트림 관련 타입을 1단계에서 적극 활용합니다.
- Reuse the already defined stream-related types in phase 1:
- - `MessageTypeStreamOpen`, `MessageTypeStreamData`, `MessageTypeStreamClose`
- - [`Envelope`](internal/protocol/protocol.go:52), [`StreamOpen`](internal/protocol/protocol.go:69), [`StreamData`](internal/protocol/protocol.go:80), [`StreamClose`](internal/protocol/protocol.go:86)
- - `StreamData` 에 per-stream 시퀀스 번호를 추가합니다.
- Add a per-stream sequence number to `StreamData`:
- - 예시 / Example:
- ```go
- type StreamData struct {
- ID StreamID `json:"id"`
- Seq uint64 `json:"seq"` // 0부터 시작하는 per-stream sequence
- Data []byte `json:"data"`
- }
- ```
-
-- [x] 스트림 ACK / 재전송 제어 메시지 추가: [`internal/protocol/protocol.go`](internal/protocol/protocol.go:52)
- - 선택적 재전송(Selective Retransmission)을 위해 `StreamAck` 메시지와 `MessageTypeStreamAck` 를 추가합니다.
- Add `StreamAck` message and `MessageTypeStreamAck` for selective retransmission:
- ```go
- const (
- MessageTypeStreamAck MessageType = "stream_ack"
- )
-
- type StreamAck struct {
- ID StreamID `json:"id"` // 대상 스트림 / target stream
- AckSeq uint64 `json:"ack_seq"` // 연속으로 수신 완료한 마지막 Seq / last contiguous sequence
- LostSeqs []uint64 `json:"lost_seqs"` // 누락된 시퀀스 목록(선택) / optional list of missing seqs
- WindowSize uint32 `json:"window_size"` // 선택: 허용 in-flight 프레임 수 / optional receive window
- }
- ```
- - [`Envelope`](internal/protocol/protocol.go:52)에 `StreamAck *StreamAck` 필드를 추가합니다.
- Extend `Envelope` with a `StreamAck *StreamAck` field.
-
-- [x] MTU-safe chunk 크기 정의
- - DTLS/UDP 헤더 및 Protobuf/length-prefix 오버헤드를 고려해 안전한 payload 크기(4KiB)를 상수로 정의합니다.
- Define a safe payload size constant (4KiB) considering DTLS/UDP headers and Protobuf/length-prefix framing.
- - 이 값은 [`internal/protocol/protocol.go`](internal/protocol/protocol.go:32) 의 `StreamChunkSize` 로 정의되었습니다.
- Implemented as `StreamChunkSize` in [`internal/protocol/protocol.go`](internal/protocol/protocol.go:32).
- - 이후 HTTP 바디 스트림 터널링 구현 시, 모든 `StreamData.Data` 는 이 크기 이하 chunk 로 잘라 전송해야 합니다.
- In the stream tunneling implementation, every `StreamData.Data` must be sliced into chunks no larger than this size.
-
----
-
-##### 3.3A.2 애플리케이션 레벨 ARQ 설계 (Selective Retransmission)
-##### 3.3A.2 Application-level ARQ (Selective Retransmission)
-
-- [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 전송 정책 구현
- - 각 `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] 송신 측 재전송 로직 구현 (StreamAck 기반)
- - 응답 스트림 송신 측에서 스트림별 `streamSender` 를 두고, `outstanding[seq] = payload` 로 아직 Ack 되지 않은 프레임을 추적합니다.
- - `StreamAck{AckSeq, LostSeqs}` 수신 시:
- - `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.
-
----
-
-##### 3.3A.3 HTTP ↔ 스트림 매핑 (서버/클라이언트)
-##### 3.3A.3 HTTP ↔ stream mapping (server/client)
-
-- [x] 서버 → 클라이언트 요청 스트림: [`cmd/server/main.go`](cmd/server/main.go:200)
- - `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.
- - `StreamOpen` 전송:
- - 요청 메서드/URL/헤더를 [`StreamOpen`](internal/protocol/protocol.go:69) 의 `Header` 혹은 pseudo-header 로 encode.
- Encode method/URL/headers into `StreamOpen.Header` or a pseudo-header scheme.
- - 요청 바디를 읽으면서 `StreamData{ID, Seq, Data}` 를 지속적으로 전송합니다.
- Read the HTTP request body and send it as a sequence of `StreamData` frames.
- - 바디 종료 시 `StreamClose{ID, Error:""}` 를 전송합니다.
- When the body ends, send `StreamClose{ID, Error:""}`.
- - 응답 수신:
- - 클라이언트에서 오는 역방향 `StreamOpen` 으로 HTTP status/header 를 수신하고,
- 이를 `http.ResponseWriter` 에 반영합니다.
- Receive response status/headers via reverse-direction `StreamOpen` and map them to `http.ResponseWriter`.
- - 연속되는 `StreamData` 를 수신할 때마다 `http.ResponseWriter.Write` 로 chunk 를 바로 전송합니다.
- For each `StreamData`, write the chunk directly to the HTTP response.
- - `StreamClose` 수신 시 응답 종료 및 스트림 자원 정리.
- On `StreamClose`, finish the response and clean up per-stream state.
-
-- [x] 클라이언트에서의 요청 처리 스트림: [`internal/proxy/client.go`](internal/proxy/client.go:200)
- - 서버로부터 들어오는 `StreamOpen{ID, ...}` 을 수신하면,
- 새로운 goroutine 을 띄워 해당 ID에 대한 로컬 HTTP 요청을 수행합니다.
- On receiving `StreamOpen{ID, ...}` from the server, spawn a goroutine to handle the local HTTP request for that stream ID.
- - 스트림별로 `io.Pipe` 또는 채널 기반 바디 리더를 준비하고,
- `StreamData` 프레임을 수신할 때마다 이 파이프에 쓰도록 합니다.
- Prepare an `io.Pipe` (or channel-backed reader) per stream and write incoming `StreamData` chunks into it.
- - 로컬 HTTP 클라이언트 응답은 반대로:
- For the local HTTP client response:
- - 응답 status/header → `StreamOpen` (client → server)
- - 응답 바디 → 여러 개의 `StreamData`
- - 종료 시점에 `StreamClose` 전송
- Send `StreamOpen` (status/headers), then a sequence of `StreamData`, followed by `StreamClose` when done.
-
----
-
-##### 3.3A.4 JSON → 바이너리 직렬화로의 잠재적 전환 (2단계)
-##### 3.3A.4 JSON → binary serialization (potential phase 2)
-
-- [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 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.
+> Note: 기존 DTLS 기반 스트림/ARQ/멀티플렉싱(3.3A/3.3B)의 작업 내역은
+> 구현 경험/아이디어 참고용으로만 유지하며, 신규 기능/운영 계획은 gRPC 터널을 기준으로 진행합니다.
---
@@ -435,7 +299,7 @@ The following tasks describe concrete work items to be implemented on the `featu
### 3.6 Hardening / 안정성 & 구성
-- [ ] 설정 유효성 검사 추가
+- [x] 설정 유효성 검사 추가
- 필수 env 누락/오류에 대한 명확한 에러 메시지.
- [ ] 에러 처리/재시도 정책
diff --git a/protocol.md b/protocol.md
new file mode 100644
index 0000000..b2581ad
--- /dev/null
+++ b/protocol.md
@@ -0,0 +1,197 @@
+# HopGate gRPC Tunnel Protocol
+
+이 문서는 HopGate 서버–클라이언트 사이의 gRPC 기반 HTTP 터널링 규약을 정리합니다. (ko)
+This document describes the gRPC-based HTTP tunneling protocol between HopGate server and clients. (en)
+
+## 1. Transport Overview / 전송 개요
+
+- Transport: TCP + TLS(HTTPS) + HTTP/2 + gRPC
+- Single long-lived bi-directional gRPC stream per client: `OpenTunnel`
+- Application payload type: `Envelope` (from `internal/protocol/hopgate_stream.proto`)
+- HTTP requests/responses are multiplexed as logical streams identified by `StreamID`.
+
+gRPC service (conceptual):
+```proto
+service HopGateTunnel {
+ rpc OpenTunnel (stream Envelope) returns (stream Envelope);
+}
+```
+
+## 2. Message Types / 메시지 타입
+
+Defined in `internal/protocol/hopgate_stream.proto`:
+
+- `HeaderValues`
+ - Wraps repeated header values: `map`
+- `Request` / `Response`
+ - Simple single-message HTTP representation (not used in the streaming tunnel path initially).
+- `StreamOpen`
+ - Opens a new logical stream for HTTP request/response (or other protocols in the future).
+- `StreamData`
+ - Carries body chunks for a stream (`id`, `seq`, `data`).
+- `StreamClose`
+ - Marks the end of a stream (`id`, `error`).
+- `StreamAck`
+ - Legacy ARQ/flow-control hint for UDP/DTLS; in gRPC tunnel it is reserved/optional.
+- `Envelope`
+ - Top-level container with `oneof payload` of the above types.
+
+In the gRPC tunnel, `Envelope` is the only gRPC message type used on the `OpenTunnel` stream.
+
+## 3. Logical Streams and StreamID / 논리 스트림과 StreamID
+
+- A single `OpenTunnel` gRPC stream multiplexes many **logical streams**.
+- Each logical stream corresponds to one HTTP request/response pair.
+- Logical streams are identified by `StreamOpen.id` (text StreamID).
+- The server generates unique IDs per gRPC connection:
+ - HTTP streams: `"http-{n}"` where `n` is a monotonically increasing counter.
+ - Control stream: `"control-0"` (special handshake/metadata stream).
+
+Within a gRPC connection:
+- Multiple `StreamID`s may be active concurrently.
+- Frames with different StreamIDs may be arbitrarily interleaved.
+- Order within a stream is tracked by `StreamData.seq` (starting at 0).
+
+## 4. HTTP Request Mapping (Server → Client) / HTTP 요청 매핑
+
+When the public HTTPS reverse-proxy (`cmd/server/main.go`) receives an HTTP request for a domain that is bound
+to a client tunnel, it serializes the request into a logical stream as follows.
+
+### 4.1 StreamOpen (request metadata and headers)
+
+- `StreamOpen.id`
+ - New unique StreamID: `"http-{n}"`.
+- `StreamOpen.service_name`
+ - Logical service selection on the client (e.g., `"web"`).
+- `StreamOpen.target_addr`
+ - Optional explicit local target address on the client (e.g., `"127.0.0.1:8080"`).
+- `StreamOpen.header`
+ - Contains HTTP request headers and pseudo-headers:
+ - Pseudo-headers:
+ - `X-HopGate-Method`: HTTP method (e.g., `"GET"`, `"POST"`).
+ - `X-HopGate-URL`: original URL path + query (e.g., `"/api/v1/foo?bar=1"`).
+ - `X-HopGate-Host`: Host header value.
+ - Other keys:
+ - All remaining HTTP headers from the incoming request, copied as-is into the map.
+
+### 4.2 StreamData* (request body chunks)
+
+- If the request has a body, the server chunks it into fixed-size pieces.
+- Chunk size: `protocol.StreamChunkSize` (currently 4 KiB).
+- For each chunk:
+ - `StreamData.id = StreamOpen.id`
+ - `StreamData.seq` increments from 0, 1, 2, …
+ - `StreamData.data` contains the raw bytes.
+
+### 4.3 StreamClose (end of request body)
+
+- After sending all body chunks, the server sends a `StreamClose`:
+ - `StreamClose.id = StreamOpen.id`
+ - `StreamClose.error` is empty on success.
+ - If there was an application-level error while reading the body, `error` contains a short description.
+
+The client reconstructs the HTTP request by:
+- Reassembling the URL and headers from the `StreamOpen` pseudo-headers and header map.
+- Concatenating `StreamData.data` in `seq` order into the request body.
+- Treating `StreamClose` as the end-of-stream marker.
+
+## 5. HTTP Response Mapping (Client → Server) / HTTP 응답 매핑
+
+The client receives `StreamOpen` + `StreamData*` + `StreamClose`, performs a local HTTP request to its
+configured target (e.g., `http://127.0.0.1:8080`), then returns an HTTP response using the same StreamID.
+
+### 5.1 StreamOpen (response headers and status)
+
+- `StreamOpen.id`
+ - Same as the request StreamID.
+- `StreamOpen.header`
+ - Contains response headers and a pseudo-header for status:
+ - Pseudo-header:
+ - `X-HopGate-Status`: HTTP status code as a string (e.g., `"200"`, `"502"`).
+ - Other keys:
+ - All HTTP response headers from the local backend, copied as-is.
+
+### 5.2 StreamData* (response body chunks)
+
+- The client reads the local HTTP response body and chunks it into 4 KiB pieces (same `StreamChunkSize`).
+- For each chunk:
+ - `StreamData.id = StreamOpen.id`
+ - `StreamData.seq` increments from 0.
+ - `StreamData.data` contains the raw bytes.
+
+### 5.3 StreamClose (end of response body)
+
+- When the local backend response is fully read, the client sends a `StreamClose`:
+ - `StreamClose.id` is the same StreamID.
+ - `StreamClose.error`:
+ - Empty string on success.
+ - Short error description if the local HTTP request/response failed (e.g., connect timeout).
+
+The server reconstructs the HTTP response by:
+- Parsing `X-HopGate-Status` into an integer HTTP status code.
+- Copying other headers into the outgoing response writer (with some security headers overridden by the server).
+- Concatenating `StreamData.data` in `seq` order into the HTTP response body.
+- Considering `StreamClose.error` for logging/metrics and possibly mapping to error pages if needed.
+
+## 6. Control / Handshake Stream / 컨트롤 스트림
+
+Before any HTTP request streams are opened, the client sends a single **control stream** to authenticate
+and describe itself.
+
+- `StreamOpen` (control):
+ - `id = "control-0"`
+ - `service_name = "control"`
+ - `header` contains:
+ - `X-HopGate-Domain`: domain this client is responsible for.
+ - `X-HopGate-API-Key`: client API key for the domain.
+ - `X-HopGate-Local-Target`: default local target such as `127.0.0.1:8080`.
+- No `StreamData` is required for the control stream in the initial design.
+- The server can optionally reply with its own control `StreamOpen/Close` to signal acceptance/rejection.
+
+On the server side:
+- `grpcTunnelServer.OpenTunnel` should:
+ 1. Wait for the first `Envelope` with `StreamOpen.id == "control-0"`.
+ 2. Extract domain, api key, and local target from the headers.
+ 3. Call the ent-based `DomainValidator` to validate `(domain, api_key)`.
+ 4. If validation succeeds, register this gRPC stream as the active tunnel for that domain.
+ 5. If validation fails, log and close the gRPC stream.
+
+Once the control stream handshake completes successfully, the server may start multiplexing multiple
+HTTP request streams (`http-0`, `http-1`, …) over the same `OpenTunnel` connection.
+
+## 7. Multiplexing Semantics / 멀티플렉싱 의미
+
+- A single TCP + TLS + HTTP/2 + gRPC connection carries:
+ - One long-lived `OpenTunnel` gRPC bi-di stream.
+ - Within it, many logical streams identified by `StreamID`.
+- The server can open multiple HTTP streams concurrently for a given client:
+ - Example: `http-0` for `/css/app.css`, `http-1` for `/api/users`, `http-2` for `/img/logo.png`.
+ - Frames for these IDs can interleave arbitrarily on the wire.
+- Per-stream ordering is preserved by combining `seq` ordering and the reliability of TCP/gRPC.
+- Slow or large responses on one stream should not prevent other streams from making progress,
+ because gRPC/HTTP2 handles stream-level flow control and scheduling.
+
+## 8. Flow Control and StreamAck / 플로우 컨트롤 및 StreamAck
+
+- The gRPC tunnel runs over TCP/HTTP2, which already provides:
+ - Reliable, in-order delivery.
+ - Connection-level and stream-level flow control.
+- Therefore, application-level selective retransmission is **not required** for the gRPC tunnel.
+- `StreamAck` remains defined in the proto for backward compatibility with the DTLS design and
+ as a potential future hint channel (e.g., window size hints), but is not used in the initial gRPC tunnel.
+
+## 9. Security Considerations / 보안 고려사항
+
+- TLS:
+ - In production, the server uses ACME-issued certificates, and clients validate the server certificate
+ using system Root CAs and SNI (`ServerName`).
+ - In debug mode, clients may use `InsecureSkipVerify: true` to allow local/self-signed certs.
+- Authentication:
+ - Application-level authentication relies on `(domain, api_key)` pairs sent via the control stream headers.
+ - The server must validate these pairs against the `Domain` table using `DomainValidator`.
+- Authorization and isolation:
+ - Each gRPC tunnel is bound to a single domain (or a defined set of domains) after successful control handshake.
+ - HTTP requests for other domains must not be forwarded over this tunnel.
+
+이 규약을 기준으로 서버/클라이언트 구현을 정렬하면, 하나의 gRPC `OpenTunnel` 스트림 위에서
+여러 HTTP 요청을 안정적으로 멀티플렉싱하면서도, 도메인/API 키 기반 인증과 TLS 보안을 함께 유지할 수 있습니다.
\ No newline at end of file
diff --git a/tools/build_server_image.sh b/tools/build_server_image.sh
new file mode 100755
index 0000000..c08cc67
--- /dev/null
+++ b/tools/build_server_image.sh
@@ -0,0 +1,57 @@
+#!/bin/sh
+
+# POSIX sh 버전의 hop-gate 서버 이미지 빌드 스크립트.
+# VERSION 은 현재 git 커밋의 7글자 SHA 를 사용합니다.
+
+set -eu
+
+# 스크립트 위치 기준 리포 루트 계산
+SCRIPT_DIR=$(cd "$(dirname "$0")" >/dev/null 2>&1 && pwd)
+REPO_ROOT="${SCRIPT_DIR}/.."
+cd "${REPO_ROOT}"
+
+# 현재 커밋 7글자 SHA, git 정보가 없으면 dev
+VERSION=$(git rev-parse --short=7 HEAD 2>/dev/null || echo dev)
+
+# 기본 이미지 이름 (첫 번째 인자로 override 가능)
+# 예:
+# ./tools/build_server_image.sh
+# ./tools/build_server_image.sh my/image/name
+IMAGE_NAME=${1:-ghcr.io/dalbodeule/hop-gate}
+
+echo "Building hop-gate server image"
+echo " context : ${REPO_ROOT}"
+echo " image : ${IMAGE_NAME}:${VERSION}"
+echo " version : ${VERSION}"
+
+# docker buildx 사용 가능 여부 확인
+if command -v docker >/dev/null 2>&1 && docker buildx version >/dev/null 2>&1; then
+ BUILD_CMD="docker buildx build"
+else
+ BUILD_CMD="docker build"
+fi
+
+# 선택적 환경 변수:
+# PLATFORM=linux/amd64,linux/arm64 # buildx 용
+# PUSH=1 # buildx --push
+
+PLATFORM_ARGS=""
+if [ "${PLATFORM-}" != "" ]; then
+ PLATFORM_ARGS="--platform ${PLATFORM}"
+fi
+
+PUSH_ARGS=""
+if [ "${PUSH-}" != "" ]; then
+ PUSH_ARGS="--push"
+fi
+
+# 실제 빌드 실행
+# shellcheck disable=SC2086
+${BUILD_CMD} \
+ ${PLATFORM_ARGS} \
+ -f Dockerfile.server \
+ --build-arg VERSION="${VERSION}" \
+ -t "${IMAGE_NAME}:${VERSION}" \
+ -t "${IMAGE_NAME}:latest" \
+ ${PUSH_ARGS} \
+ .
\ No newline at end of file