feat(plugin): add protocol proxy mode

internal/adminhttp/api.go

@@ -34,6 +34,7 @@ type APIHandlers struct {
 	PluginsList     http.HandlerFunc
 	PluginItem      SegmentHandlerFunc
 	PluginAction    SegmentHandlerFunc
+	PluginDraining  SegmentHandlerFunc
 	PluginDispatch  http.HandlerFunc
 }
 
@@ -86,6 +87,10 @@ func NewAPIHandler(prefix string, handlers APIHandlers) http.HandlerFunc {
 			callHandler(w, r, handlers.PluginDispatch)
 		case strings.HasPrefix(path, "/plugins/"):
 			pluginPath := strings.TrimPrefix(path, "/plugins/")
+			if strings.HasSuffix(pluginPath, "/draining/force-close") {
+				callSegmentHandler(w, r, handlers.PluginDraining, strings.TrimSuffix(pluginPath, "/draining/force-close"))
+				return
+			}
 			if strings.Count(pluginPath, "/") == 1 {
 				callSegmentHandler(w, r, handlers.PluginAction, pluginPath)
 				return

internal/adminhttp/api_test.go

@@ -35,6 +35,7 @@ func TestNewAPIHandlerRoutesRequests(t *testing.T) {
 		{name: "plugins list", method: http.MethodGet, path: "/admin/api/plugins", wantCall: "plugins_list"},
 		{name: "plugin item", method: http.MethodPut, path: "/admin/api/plugins/upstream-rewrite", wantCall: "plugin_item", wantSegment: "upstream-rewrite"},
 		{name: "plugin action", method: http.MethodPost, path: "/admin/api/plugins/upstream-rewrite/enable", wantCall: "plugin_action", wantSegment: "upstream-rewrite/enable"},
+		{name: "plugin draining force close", method: http.MethodPost, path: "/admin/api/plugins/mc-auth-proxy/draining/force-close", wantCall: "plugin_draining", wantSegment: "mc-auth-proxy"},
 		{name: "plugin dispatch", method: http.MethodGet, path: "/admin/api/plugins/dispatch-plan", wantCall: "plugin_dispatch"},
 	}
 
@@ -68,6 +69,7 @@ func TestNewAPIHandlerRoutesRequests(t *testing.T) {
 				PluginsList:     recordCall(&gotCall, "plugins_list"),
 				PluginItem:      recordSegmentCall(&gotCall, &gotSegment, "plugin_item"),
 				PluginAction:    recordSegmentCall(&gotCall, &gotSegment, "plugin_action"),
+				PluginDraining:  recordSegmentCall(&gotCall, &gotSegment, "plugin_draining"),
 				PluginDispatch:  recordCall(&gotCall, "plugin_dispatch"),
 			})
 

internal/pluginmanager/artifact.go

@@ -188,9 +188,9 @@ func (s ArtifactStore) ValidateAndStore(upload ArtifactUpload) (ArtifactRecord,
 	if err != nil {
 		return ArtifactRecord{}, err
 	}
-	capabilities := manifest.Capabilities
-	if len(capabilities) == 0 {
-		capabilities = json.RawMessage(`{}`)
+	capabilities, err := capabilitiesSummaryJSON(manifest.Capabilities)
+	if err != nil {
+		return ArtifactRecord{}, err
 	}
 	now := s.now().Unix()
 	return ArtifactRecord{
@@ -219,6 +219,38 @@ func (s ArtifactStore) ValidateAndStore(upload ArtifactUpload) (ArtifactRecord,
 	}, nil
 }
 
+func capabilitiesSummaryJSON(raw json.RawMessage) ([]byte, error) {
+	summary := CapabilitySummary{
+		UpstreamConnect: UpstreamConnectCapability{Mode: UpstreamModeDialer},
+	}
+	if len(raw) == 0 {
+		return json.Marshal(summary)
+	}
+	summary.Raw = append(json.RawMessage(nil), raw...)
+	var caps struct {
+		UpstreamConnect UpstreamConnectCapability `json:"upstream_connect"`
+		Minecraft       *MinecraftCapability      `json:"minecraft"`
+	}
+	if err := json.Unmarshal(raw, &caps); err != nil {
+		return nil, fmt.Errorf("invalid capabilities: %w", err)
+	}
+	if caps.UpstreamConnect.Mode != "" {
+		summary.UpstreamConnect.Mode = caps.UpstreamConnect.Mode
+	}
+	if caps.Minecraft != nil {
+		summary.Minecraft = caps.Minecraft
+		if summary.Minecraft.UnsupportedPolicy == "" {
+			summary.Minecraft.UnsupportedPolicy = summary.Minecraft.ProtocolVersions.UnsupportedPolicy
+		}
+	}
+	switch summary.UpstreamConnect.Mode {
+	case UpstreamModeDialer, UpstreamModeProtocolProxy:
+	default:
+		return nil, fmt.Errorf("unsupported upstream_connect.mode %q", summary.UpstreamConnect.Mode)
+	}
+	return json.Marshal(summary)
+}
+
 func validateManifest(manifest Manifest) error {
 	switch {
 	case manifest.SchemaVersion != SchemaVersion:

internal/pluginmanager/artifact_test.go

@@ -91,7 +91,14 @@ func TestArtifactStoreRejectsUnsafePackage(t *testing.T) {
 }
 
 func testManifestBytes(t *testing.T, pluginID string) []byte {
+	return testManifestBytesWithCapabilities(t, pluginID, json.RawMessage(`{"extension_points":["upstream.connect/v1"]}`))
+}
+
+func testManifestBytesWithCapabilities(t *testing.T, pluginID string, capabilities json.RawMessage) []byte {
 	t.Helper()
+	if len(capabilities) == 0 {
+		capabilities = json.RawMessage(`{"extension_points":["upstream.connect/v1"]}`)
+	}
 	manifest := Manifest{
 		SchemaVersion: SchemaVersion,
 		ID:            pluginID,
@@ -111,7 +118,7 @@ func testManifestBytes(t *testing.T, pluginID string) []byte {
 			Type: "hook",
 			Key:  ExtensionUpstreamConnect,
 		}},
-		Capabilities:  json.RawMessage(`{"extension_points":["upstream.connect/v1"]}`),
+		Capabilities:  capabilities,
 		ConfigSchema:  json.RawMessage(`{"type":"object"}`),
 		RuntimeLimits: RuntimeLimits{HandlerTimeoutMS: 3000},
 	}

internal/pluginmanager/manager.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	stdplugin "plugin"
 	"reflect"
@@ -76,6 +77,11 @@ type Manager struct {
 	mu       sync.Mutex
 	loaded   map[string]*loadedPlugin
 	snapshot atomic.Value
+
+	proxyMu     sync.Mutex
+	proxySeq    uint64
+	proxyConns  map[uint64]*proxyConnection
+	drainingIDs map[string]bool
 }
 
 type loadedPlugin struct {
@@ -87,18 +93,52 @@ type loadedPlugin struct {
 }
 
 type upstreamHandler struct {
+	pluginID            string
+	artifactID          string
+	priority            int
+	handlerID           string
+	mode                string
+	timeout             time.Duration
+	initialWriteTimeout time.Duration
+	accept              func(api.UpstreamConnectRequest) bool
+	handle              func(api.UpstreamConnectRequest) (net.Conn, error)
+
+	calls          atomic.Uint64
+	errors         atomic.Uint64
+	panics         atomic.Uint64
+	timeouts       atomic.Uint64
+	blocked        atomic.Uint64
+	activeProxy    atomic.Int64
+	proxyStarted   atomic.Uint64
+	proxyCompleted atomic.Uint64
+	proxyErrors    atomic.Uint64
+	proxyBytesIn   atomic.Uint64
+	proxyBytesOut  atomic.Uint64
+	proxyDuration  atomic.Uint64
+}
+
+type proxyConnection struct {
+	id         uint64
 	pluginID   string
 	artifactID string
-	priority   int
 	handlerID  string
-	timeout    time.Duration
-	accept     func(api.UpstreamConnectRequest) bool
-	handle     func(api.UpstreamConnectRequest) (net.Conn, error)
+	handler    *upstreamHandler
+	client     net.Conn
+	endpoint   net.Conn
+	startedAt  time.Time
+	draining   bool
+}
 
-	calls    atomic.Uint64
-	errors   atomic.Uint64
-	panics   atomic.Uint64
-	timeouts atomic.Uint64
+type ProxyConnectionHandle struct {
+	manager *Manager
+	id      uint64
+}
+
+type ProxyConnectionStats struct {
+	BytesToPlugin int64
+	BytesToClient int64
+	Duration      time.Duration
+	Err           error
 }
 
 type Options struct {
@@ -115,12 +155,14 @@ func New(options Options) *Manager {
 		adapter = GoPluginAdapter{}
 	}
 	manager := &Manager{
-		repo:       NewRepository(options.DB),
-		store:      NewArtifactStore(options.ArtifactRoot),
-		adapter:    adapter,
-		handleConn: options.HandleConn,
-		wg:         options.WaitGroup,
-		loaded:     make(map[string]*loadedPlugin),
+		repo:        NewRepository(options.DB),
+		store:       NewArtifactStore(options.ArtifactRoot),
+		adapter:     adapter,
+		handleConn:  options.HandleConn,
+		wg:          options.WaitGroup,
+		loaded:      make(map[string]*loadedPlugin),
+		proxyConns:  make(map[uint64]*proxyConnection),
+		drainingIDs: make(map[string]bool),
 	}
 	manager.publish(nil)
 	return manager
@@ -207,6 +249,7 @@ func (m *Manager) Enable(ctx context.Context, actor, pluginID string) (PluginRec
 	if err := m.markEnabled(ctx, loaded); err != nil {
 		return PluginRecord{}, err
 	}
+	m.clearDrainingLocked(pluginID)
 	m.publish(next)
 	_ = m.repo.UpdateArtifactStatus(ctx, loaded.artifact.ID, ArtifactStatusLoaded, "")
 	_ = m.repo.RecordOperation(ctx, pluginID, loaded.artifact.ID, "enable", "succeeded", actor, "plugin enabled", map[string]any{
@@ -229,13 +272,20 @@ func (m *Manager) Disable(ctx context.Context, actor, pluginID string) (PluginRe
 		return PluginRecord{}, err
 	}
 	m.removeFromDispatchLocked(pluginID)
+	m.markDrainingLocked(pluginID)
 	if loaded := m.loaded[pluginID]; loaded != nil && loaded.instance != nil {
 		if err := loaded.instance.Destroy(); err != nil {
 			_ = m.repo.RecordOperation(ctx, pluginID, loaded.artifact.ID, "disable", "warning", actor, err.Error(), nil)
 		}
 	}
+	runtimeState := RuntimeDisabled
+	if m.activeProxyCountLocked(pluginID) > 0 {
+		runtimeState = RuntimeDraining
+	}
 	delete(m.loaded, pluginID)
-	if err := m.repo.MarkRuntime(ctx, pluginID, RuntimeDisabled, "", "", pluginRecord.DesiredGeneration, "", nil, nil); err != nil {
+	if err := m.repo.MarkRuntime(ctx, pluginID, runtimeState, "", "", pluginRecord.DesiredGeneration, "", map[string]any{
+		"active_proxy_connections": m.activeProxyCountLocked(pluginID),
+	}, nil); err != nil {
 		return PluginRecord{}, err
 	}
 	_ = m.repo.RecordOperation(ctx, pluginID, pluginRecord.DesiredArtifactID, "disable", "succeeded", actor, "plugin disabled", nil)
@@ -251,6 +301,7 @@ func (m *Manager) Delete(ctx context.Context, actor, pluginID string) error {
 		return err
 	}
 	m.removeFromDispatchLocked(pluginID)
+	m.markDrainingLocked(pluginID)
 	if loaded := m.loaded[pluginID]; loaded != nil && loaded.instance != nil {
 		_ = loaded.instance.Destroy()
 	}
@@ -288,6 +339,7 @@ func (m *Manager) Reconcile(ctx context.Context) error {
 		}
 		nextByPlugin[pluginRecord.ID] = loaded.handlers
 		_ = m.markEnabled(ctx, loaded)
+		m.clearDrainingLocked(pluginRecord.ID)
 	}
 	m.publish(flattenHandlers(nextByPlugin))
 	return nil
@@ -305,6 +357,7 @@ func (m *Manager) ConnectUpstream(ctx context.Context, req api.UpstreamConnectRe
 	if req.Context == nil {
 		req.Context = ctx
 	}
+	req.InitialData = append([]byte(nil), req.InitialData...)
 	for _, handler := range handlers {
 		accepted, err := handler.accepts(req)
 		if err != nil {
@@ -321,12 +374,54 @@ func (m *Manager) ConnectUpstream(ctx context.Context, req api.UpstreamConnectRe
 			return UpstreamResult{Handled: true}, err
 		}
 		if conn != nil {
-			return UpstreamResult{Conn: conn, Handled: true}, nil
+			result := UpstreamResult{
+				Conn:      conn,
+				Handled:   true,
+				Mode:      handler.mode,
+				PluginID:  handler.pluginID,
+				HandlerID: handler.handlerID,
+			}
+			if handler.mode == UpstreamModeProtocolProxy {
+				return m.startProtocolProxy(ctx, handler, result, req)
+			}
+			return result, nil
 		}
 	}
 	return UpstreamResult{}, nil
 }
 
+func (m *Manager) startProtocolProxy(ctx context.Context, handler *upstreamHandler, result UpstreamResult, req api.UpstreamConnectRequest) (UpstreamResult, error) {
+	endpoint := result.Conn
+	initial := append([]byte(nil), req.InitialData...)
+	if len(initial) > 0 {
+		if handler.initialWriteTimeout > 0 {
+			_ = endpoint.SetWriteDeadline(time.Now().Add(handler.initialWriteTimeout))
+			defer endpoint.SetWriteDeadline(time.Time{})
+		}
+		if err := writeAll(endpoint, initial); err != nil {
+			handler.proxyErrors.Add(1)
+			_ = endpoint.Close()
+			return UpstreamResult{Handled: true, Mode: handler.mode, PluginID: handler.pluginID, HandlerID: handler.handlerID}, fmt.Errorf("plugin %s protocol-proxy initial replay failed: %w", handler.pluginID, err)
+		}
+	}
+
+	handle := m.TrackProxyConnection(result, req.Source, endpoint)
+	if handle == nil {
+		_ = endpoint.Close()
+		return UpstreamResult{Handled: true, Mode: handler.mode, PluginID: handler.pluginID, HandlerID: handler.handlerID}, fmt.Errorf("plugin %s protocol-proxy tracking failed", handler.pluginID)
+	}
+	runProtocolProxy(ctx, handle, req.Source, endpoint)
+
+	return UpstreamResult{
+		Handled:         true,
+		Mode:            handler.mode,
+		PluginID:        handler.pluginID,
+		HandlerID:       handler.handlerID,
+		InitialDataSent: len(initial) > 0,
+		Proxied:         true,
+	}, nil
+}
+
 func (h *upstreamHandler) accepts(req api.UpstreamConnectRequest) (accepted bool, err error) {
 	if h.accept == nil {
 		return true, nil
@@ -366,6 +461,106 @@ func (m *Manager) DispatchPlan(ctx context.Context) DispatchPlan {
 	return plan
 }
 
+func (m *Manager) TrackProxyConnection(result UpstreamResult, client, endpoint net.Conn) *ProxyConnectionHandle {
+	if result.Mode != UpstreamModeProtocolProxy || client == nil || endpoint == nil {
+		return nil
+	}
+	handler := m.findHandler(result.PluginID, result.HandlerID)
+	if handler == nil {
+		return nil
+	}
+	id := atomic.AddUint64(&m.proxySeq, 1)
+	proxyConn := &proxyConnection{
+		id:         id,
+		pluginID:   result.PluginID,
+		artifactID: handler.artifactID,
+		handlerID:  result.HandlerID,
+		handler:    handler,
+		client:     client,
+		endpoint:   endpoint,
+		startedAt:  time.Now(),
+	}
+	handler.activeProxy.Add(1)
+	handler.proxyStarted.Add(1)
+	m.proxyMu.Lock()
+	proxyConn.draining = m.drainingIDs[result.PluginID]
+	m.proxyConns[id] = proxyConn
+	m.proxyMu.Unlock()
+	return &ProxyConnectionHandle{manager: m, id: id}
+}
+
+func (h *ProxyConnectionHandle) Finish(stats ProxyConnectionStats) {
+	if h == nil || h.manager == nil {
+		return
+	}
+	h.manager.finishProxyConnection(h.id, stats)
+}
+
+func (m *Manager) ForceCloseDraining(ctx context.Context, actor, pluginID string) (int, error) {
+	_ = ctx
+	var conns []*proxyConnection
+	m.proxyMu.Lock()
+	for _, conn := range m.proxyConns {
+		if conn.pluginID == pluginID && conn.draining {
+			conns = append(conns, conn)
+		}
+	}
+	m.proxyMu.Unlock()
+	for _, conn := range conns {
+		_ = conn.client.Close()
+		_ = conn.endpoint.Close()
+	}
+	_ = m.repo.RecordOperation(ctx, pluginID, "", "force_close_draining", "succeeded", actor, "draining protocol-proxy connections force closed", map[string]any{
+		"closed": len(conns),
+	})
+	return len(conns), nil
+}
+
+func (m *Manager) findHandler(pluginID, handlerID string) *upstreamHandler {
+	value := m.snapshot.Load()
+	if handlers, ok := value.([]*upstreamHandler); ok {
+		for _, handler := range handlers {
+			if handler.pluginID == pluginID && handler.handlerID == handlerID {
+				return handler
+			}
+		}
+	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if loaded := m.loaded[pluginID]; loaded != nil {
+		for _, handler := range loaded.handlers {
+			if handler.handlerID == handlerID {
+				return handler
+			}
+		}
+	}
+	return nil
+}
+
+func (m *Manager) finishProxyConnection(id uint64, stats ProxyConnectionStats) {
+	m.proxyMu.Lock()
+	proxyConn := m.proxyConns[id]
+	delete(m.proxyConns, id)
+	m.proxyMu.Unlock()
+	if proxyConn == nil || proxyConn.handler == nil {
+		return
+	}
+	proxyConn.handler.activeProxy.Add(-1)
+	proxyConn.handler.proxyCompleted.Add(1)
+	if stats.Err != nil {
+		proxyConn.handler.proxyErrors.Add(1)
+	}
+	if stats.BytesToPlugin > 0 {
+		proxyConn.handler.proxyBytesIn.Add(uint64(stats.BytesToPlugin))
+	}
+	if stats.BytesToClient > 0 {
+		proxyConn.handler.proxyBytesOut.Add(uint64(stats.BytesToClient))
+	}
+	if stats.Duration > 0 {
+		proxyConn.handler.proxyDuration.Add(uint64(stats.Duration.Milliseconds()))
+	}
+}
+
 func (m *Manager) loadLocked(ctx context.Context, pluginRecord PluginRecord) (*loadedPlugin, error) {
 	if loaded := m.loaded[pluginRecord.ID]; loaded != nil &&
 		loaded.artifact.ID == pluginRecord.DesiredArtifactID &&
@@ -426,6 +621,35 @@ func (m *Manager) removeFromDispatchLocked(pluginID string) {
 	m.publish(flattenHandlers(current))
 }
 
+func (m *Manager) markDrainingLocked(pluginID string) {
+	m.proxyMu.Lock()
+	defer m.proxyMu.Unlock()
+	m.drainingIDs[pluginID] = true
+	for _, conn := range m.proxyConns {
+		if conn.pluginID == pluginID {
+			conn.draining = true
+		}
+	}
+}
+
+func (m *Manager) clearDrainingLocked(pluginID string) {
+	m.proxyMu.Lock()
+	delete(m.drainingIDs, pluginID)
+	m.proxyMu.Unlock()
+}
+
+func (m *Manager) activeProxyCountLocked(pluginID string) int {
+	m.proxyMu.Lock()
+	defer m.proxyMu.Unlock()
+	count := 0
+	for _, conn := range m.proxyConns {
+		if conn.pluginID == pluginID {
+			count++
+		}
+	}
+	return count
+}
+
 func (m *Manager) publish(handlers []*upstreamHandler) {
 	sort.SliceStable(handlers, func(i, j int) bool {
 		if handlers[i].priority != handlers[j].priority {
@@ -441,31 +665,42 @@ func (m *Manager) publish(handlers []*upstreamHandler) {
 
 func buildHandlers(pluginRecord PluginRecord, artifact ArtifactRecord, gateway *Gateway) []*upstreamHandler {
 	timeout := DefaultHandlerTimeout
+	initialWriteTimeout := DefaultInitialWriteTimeout
 	var manifest Manifest
-	if err := json.Unmarshal([]byte(artifact.MetadataJSON), &manifest); err == nil && manifest.RuntimeLimits.HandlerTimeoutMS > 0 {
-		timeout = time.Duration(manifest.RuntimeLimits.HandlerTimeoutMS) * time.Millisecond
+	if err := json.Unmarshal([]byte(artifact.MetadataJSON), &manifest); err == nil {
+		if manifest.RuntimeLimits.HandlerTimeoutMS > 0 {
+			timeout = time.Duration(manifest.RuntimeLimits.HandlerTimeoutMS) * time.Millisecond
+		}
+		if manifest.RuntimeLimits.InitialWriteTimeoutMS > 0 {
+			initialWriteTimeout = time.Duration(manifest.RuntimeLimits.InitialWriteTimeoutMS) * time.Millisecond
+		}
 	}
 	var handlers []*upstreamHandler
+	mode := upstreamModeFromArtifact(artifact)
 	if hook, ok := gateway.UpstreamConnectHandler(); ok {
 		handlers = append(handlers, &upstreamHandler{
-			pluginID:   pluginRecord.ID,
-			artifactID: artifact.ID,
-			priority:   pluginRecord.Priority,
-			handlerID:  "upstream.connect/v1",
-			timeout:    timeout,
-			accept:     hook.Acceptor(),
-			handle:     hook.Handler(),
+			pluginID:            pluginRecord.ID,
+			artifactID:          artifact.ID,
+			priority:            pluginRecord.Priority,
+			handlerID:           "upstream.connect/v1",
+			mode:                mode,
+			timeout:             timeout,
+			initialWriteTimeout: initialWriteTimeout,
+			accept:              hook.Acceptor(),
+			handle:              hook.Handler(),
 		})
 	}
 	if hook, ok := gateway.LegacyUpstreamHandler(); ok {
 		acceptor := hook.Acceptor()
 		handler := hook.Handler()
 		handlers = append(handlers, &upstreamHandler{
-			pluginID:   pluginRecord.ID,
-			artifactID: artifact.ID,
-			priority:   pluginRecord.Priority,
-			handlerID:  "legacy-upstream",
-			timeout:    timeout,
+			pluginID:            pluginRecord.ID,
+			artifactID:          artifact.ID,
+			priority:            pluginRecord.Priority,
+			handlerID:           "legacy-upstream",
+			mode:                UpstreamModeDialer,
+			timeout:             timeout,
+			initialWriteTimeout: initialWriteTimeout,
 			accept: func(req api.UpstreamConnectRequest) bool {
 				return acceptor(req.Source, req.Upstream)
 			},
@@ -477,6 +712,19 @@ func buildHandlers(pluginRecord PluginRecord, artifact ArtifactRecord, gateway *
 	return handlers
 }
 
+func upstreamModeFromArtifact(artifact ArtifactRecord) string {
+	var summary CapabilitySummary
+	if err := json.Unmarshal([]byte(artifact.CapabilitiesSummaryJSON), &summary); err == nil {
+		switch summary.UpstreamConnect.Mode {
+		case UpstreamModeProtocolProxy:
+			return UpstreamModeProtocolProxy
+		case UpstreamModeDialer:
+			return UpstreamModeDialer
+		}
+	}
+	return UpstreamModeDialer
+}
+
 func (h *upstreamHandler) invoke(req api.UpstreamConnectRequest) (conn net.Conn, err error) {
 	h.calls.Add(1)
 	ctx := req.Context
@@ -508,6 +756,9 @@ func (h *upstreamHandler) invoke(req api.UpstreamConnectRequest) (conn net.Conn,
 		go closeLateConn(done)
 		return nil, ctx.Err()
 	case result := <-done:
+		if errors.Is(result.err, api.ErrBlocked) {
+			h.blocked.Add(1)
+		}
 		if result.err != nil && !errors.Is(result.err, api.ErrPass) {
 			h.errors.Add(1)
 		}
@@ -527,6 +778,116 @@ func closeLateConn(done <-chan result) {
 	}
 }
 
+type proxyCopyResult struct {
+	toPlugin bool
+	bytes    int64
+	err      error
+}
+
+type closeWriter interface {
+	CloseWrite() error
+}
+
+type closeReader interface {
+	CloseRead() error
+}
+
+func runProtocolProxy(ctx context.Context, handle *ProxyConnectionHandle, client, endpoint net.Conn) {
+	start := time.Now()
+	defer client.Close()
+	defer endpoint.Close()
+	done := make(chan proxyCopyResult, 2)
+	stopContext := make(chan struct{})
+	if ctx != nil {
+		go func() {
+			select {
+			case <-ctx.Done():
+				_ = client.Close()
+				_ = endpoint.Close()
+			case <-stopContext:
+			}
+		}()
+	}
+
+	go copyProtocolProxy(endpoint, client, true, done)
+	go copyProtocolProxy(client, endpoint, false, done)
+
+	var stats ProxyConnectionStats
+	for i := 0; i < 2; i++ {
+		result := <-done
+		if result.toPlugin {
+			stats.BytesToPlugin += result.bytes
+		} else {
+			stats.BytesToClient += result.bytes
+		}
+		if result.err != nil && !errors.Is(result.err, io.EOF) && stats.Err == nil {
+			stats.Err = result.err
+		}
+	}
+	close(stopContext)
+	stats.Duration = time.Since(start)
+	handle.Finish(stats)
+}
+
+func copyProtocolProxy(dst io.Writer, src io.Reader, toPlugin bool, done chan<- proxyCopyResult) {
+	result := proxyCopyResult{toPlugin: toPlugin}
+	defer func() {
+		if rec := recover(); rec != nil {
+			result.err = fmt.Errorf("protocol-proxy copy panic: %v", rec)
+		}
+		closeRead(src)
+		if toPlugin {
+			closeWriteOnly(dst)
+		} else {
+			closeWrite(dst)
+		}
+		done <- result
+	}()
+	result.bytes, result.err = copyForward(dst, src)
+}
+
+func copyForward(dst io.Writer, src io.Reader) (int64, error) {
+	return io.Copy(dst, src)
+}
+
+func writeAll(w io.Writer, buf []byte) error {
+	for len(buf) > 0 {
+		n, err := w.Write(buf)
+		if n > 0 {
+			buf = buf[n:]
+		}
+		if err != nil {
+			return err
+		}
+		if n == 0 {
+			return io.ErrShortWrite
+		}
+	}
+	return nil
+}
+
+func closeWrite(conn any) {
+	if closer, ok := conn.(closeWriter); ok {
+		_ = closer.CloseWrite()
+		return
+	}
+	if closer, ok := conn.(io.Closer); ok {
+		_ = closer.Close()
+	}
+}
+
+func closeWriteOnly(conn any) {
+	if closer, ok := conn.(closeWriter); ok {
+		_ = closer.CloseWrite()
+	}
+}
+
+func closeRead(conn any) {
+	if closer, ok := conn.(closeReader); ok {
+		_ = closer.CloseRead()
+	}
+}
+
 func flattenHandlers(byPlugin map[string][]*upstreamHandler) []*upstreamHandler {
 	var handlers []*upstreamHandler
 	for _, pluginHandlers := range byPlugin {
@@ -539,16 +900,25 @@ func handlerSummaries(handlers []*upstreamHandler) []DispatchHandlerSummary {
 	summaries := make([]DispatchHandlerSummary, 0, len(handlers))
 	for _, handler := range handlers {
 		summaries = append(summaries, DispatchHandlerSummary{
-			PluginID:       handler.pluginID,
-			ArtifactID:     handler.artifactID,
-			Priority:       handler.priority,
-			HandlerID:      handler.handlerID,
-			ExtensionPoint: ExtensionUpstreamConnect,
-			TimeoutMS:      handler.timeout.Milliseconds(),
-			Calls:          handler.calls.Load(),
-			Errors:         handler.errors.Load(),
-			Panics:         handler.panics.Load(),
-			Timeouts:       handler.timeouts.Load(),
+			PluginID:        handler.pluginID,
+			ArtifactID:      handler.artifactID,
+			Priority:        handler.priority,
+			HandlerID:       handler.handlerID,
+			ExtensionPoint:  ExtensionUpstreamConnect,
+			Mode:            handler.mode,
+			TimeoutMS:       handler.timeout.Milliseconds(),
+			Calls:           handler.calls.Load(),
+			Errors:          handler.errors.Load(),
+			Panics:          handler.panics.Load(),
+			Timeouts:        handler.timeouts.Load(),
+			Blocked:         handler.blocked.Load(),
+			ActiveProxy:     handler.activeProxy.Load(),
+			ProxyStarted:    handler.proxyStarted.Load(),
+			ProxyCompleted:  handler.proxyCompleted.Load(),
+			ProxyErrors:     handler.proxyErrors.Load(),
+			ProxyBytesIn:    handler.proxyBytesIn.Load(),
+			ProxyBytesOut:   handler.proxyBytesOut.Load(),
+			ProxyDurationMS: handler.proxyDuration.Load(),
 		})
 	}
 	return summaries

internal/pluginmanager/manager_test.go

@@ -1,12 +1,16 @@
 package pluginmanager
 
 import (
+	"bytes"
 	"context"
 	"database/sql"
+	"encoding/json"
 	"errors"
+	"io"
 	"net"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/tursom/mc-gateway/internal/admindb"
 	"github.com/tursom/mc-gateway/plugin/api"
@@ -109,6 +113,46 @@ func TestManagerErrPassContinuesToNextHandler(t *testing.T) {
 	}
 }
 
+func TestManagerErrBlockedStopsDispatch(t *testing.T) {
+	adapter := &fakeAdapter{
+		handlers: map[string]api.UpstreamConnectHandler{
+			"plugin-a": func(api.UpstreamConnectRequest) (net.Conn, error) {
+				return nil, api.ErrBlocked
+			},
+			"plugin-b": func(api.UpstreamConnectRequest) (net.Conn, error) {
+				return newMemoryConn(), nil
+			},
+		},
+	}
+	manager := newManagerForTest(t, adapter)
+	artifactA := uploadTestArtifact(t, manager, "plugin-a")
+	artifactB := uploadTestArtifact(t, manager, "plugin-b")
+	if _, err := manager.SetDesired(context.Background(), "admin", "plugin-a", artifactA.ID, DesiredEnabled, `{}`, 10); err != nil {
+		t.Fatalf("SetDesired(a) error = %v", err)
+	}
+	if _, err := manager.SetDesired(context.Background(), "admin", "plugin-b", artifactB.ID, DesiredEnabled, `{}`, 20); err != nil {
+		t.Fatalf("SetDesired(b) error = %v", err)
+	}
+	if err := manager.Reconcile(context.Background()); err != nil {
+		t.Fatalf("Reconcile() error = %v", err)
+	}
+
+	result, err := manager.ConnectUpstream(context.Background(), api.UpstreamConnectRequest{Host: "play.example", Upstream: "backend"})
+	if !errors.Is(err, api.ErrBlocked) {
+		t.Fatalf("ConnectUpstream() error = %v, want ErrBlocked", err)
+	}
+	if !result.Handled {
+		t.Fatalf("ConnectUpstream() = %+v, want handled", result)
+	}
+	plan := manager.DispatchPlan(context.Background())
+	if got := plan.Handlers[0].Blocked; got != 1 {
+		t.Fatalf("blocked count = %d, want 1", got)
+	}
+	if got := plan.Handlers[1].Calls; got != 0 {
+		t.Fatalf("second handler calls = %d, want 0", got)
+	}
+}
+
 func TestManagerPanicDoesNotReplaceExistingDispatch(t *testing.T) {
 	adapter := &fakeAdapter{
 		handlers: map[string]api.UpstreamConnectHandler{
@@ -240,6 +284,176 @@ func TestAcceptorPanicIsRecovered(t *testing.T) {
 	}
 }
 
+func TestProtocolProxyTrackDrainAndForceClose(t *testing.T) {
+	clientGateway, clientSide := net.Pipe()
+	defer clientSide.Close()
+	pluginGateway, pluginSide := net.Pipe()
+	defer pluginSide.Close()
+	handlerReturned := make(chan struct{})
+	adapter := &fakeAdapter{
+		handlers: map[string]api.UpstreamConnectHandler{
+			"plugin-a": func(req api.UpstreamConnectRequest) (net.Conn, error) {
+				go func() {
+					buf := make([]byte, len(req.InitialData))
+					if _, err := io.ReadFull(pluginSide, buf); err != nil {
+						t.Errorf("plugin side initial read error = %v", err)
+					}
+					close(handlerReturned)
+					_, _ = pluginSide.Read(make([]byte, 1))
+				}()
+				return pluginGateway, nil
+			},
+		},
+	}
+	manager := newManagerForTest(t, adapter)
+	artifact := uploadTestArtifactWithCapabilities(t, manager, "plugin-a", json.RawMessage(`{"upstream_connect":{"mode":"protocol-proxy"}}`))
+	if _, err := manager.SetDesired(context.Background(), "admin", "plugin-a", artifact.ID, DesiredEnabled, `{}`, 10); err != nil {
+		t.Fatalf("SetDesired() error = %v", err)
+	}
+	if _, err := manager.Enable(context.Background(), "admin", "plugin-a"); err != nil {
+		t.Fatalf("Enable() error = %v", err)
+	}
+	errCh := make(chan error, 1)
+	go func() {
+		_, err := manager.ConnectUpstream(context.Background(), api.UpstreamConnectRequest{
+			Host:        "play.example",
+			Upstream:    "backend",
+			Source:      clientGateway,
+			InitialData: []byte("hello"),
+		})
+		errCh <- err
+	}()
+	<-handlerReturned
+	waitForPluginManagerTest(t, func() bool {
+		return manager.DispatchPlan(context.Background()).Handlers[0].ActiveProxy == 1
+	})
+	plan := manager.DispatchPlan(context.Background())
+	if got := plan.Handlers[0].ActiveProxy; got != 1 {
+		t.Fatalf("active proxy = %d, want 1", got)
+	}
+	if _, err := manager.Disable(context.Background(), "admin", "plugin-a"); err != nil {
+		t.Fatalf("Disable() error = %v", err)
+	}
+	closed, err := manager.ForceCloseDraining(context.Background(), "admin", "plugin-a")
+	if err != nil {
+		t.Fatalf("ForceCloseDraining() error = %v", err)
+	}
+	if closed != 1 {
+		t.Fatalf("ForceCloseDraining() = %d, want 1", closed)
+	}
+	if err := <-errCh; err != nil {
+		t.Fatalf("ConnectUpstream() error = %v", err)
+	}
+	waitForPluginManagerTest(t, func() bool {
+		return manager.activeProxyCountLocked("plugin-a") == 0
+	})
+}
+
+func TestProtocolProxyReplaysInitialAndForwardsClientBytes(t *testing.T) {
+	clientGateway, clientSide := net.Pipe()
+	defer clientSide.Close()
+
+	initial := []byte("initial-handshake")
+	next := []byte("login-start")
+	pluginRead := make(chan []byte, 1)
+	adapter := &fakeAdapter{
+		handlers: map[string]api.UpstreamConnectHandler{
+			"plugin-a": func(api.UpstreamConnectRequest) (net.Conn, error) {
+				gatewayEnd, pluginEnd := net.Pipe()
+				go func() {
+					defer pluginEnd.Close()
+					buf := make([]byte, len(initial)+len(next))
+					if _, err := io.ReadFull(pluginEnd, buf); err != nil {
+						t.Errorf("plugin read error = %v", err)
+						return
+					}
+					pluginRead <- buf
+				}()
+				return gatewayEnd, nil
+			},
+		},
+	}
+	manager := newManagerForTest(t, adapter)
+	enableProtocolProxyTestPlugin(t, manager, "plugin-a")
+
+	errCh := make(chan error, 1)
+	go func() {
+		_, err := manager.ConnectUpstream(context.Background(), api.UpstreamConnectRequest{
+			Source:      clientGateway,
+			InitialData: initial,
+		})
+		errCh <- err
+	}()
+	if _, err := clientSide.Write(next); err != nil {
+		t.Fatalf("client write error = %v", err)
+	}
+	got := <-pluginRead
+	if !bytes.Equal(got, append(append([]byte(nil), initial...), next...)) {
+		t.Fatalf("plugin bytes = %q, want initial+next", got)
+	}
+	_ = clientSide.Close()
+	if err := <-errCh; err != nil {
+		t.Fatalf("ConnectUpstream() error = %v", err)
+	}
+	plan := manager.DispatchPlan(context.Background())
+	if got := plan.Handlers[0].ProxyBytesIn; got != uint64(len(next)) {
+		t.Fatalf("proxy bytes in = %d, want %d", got, len(next))
+	}
+}
+
+func TestProtocolProxyInitialWriteTimeoutClosesUnreadableConn(t *testing.T) {
+	reader, writer := net.Pipe()
+	defer reader.Close()
+	adapter := &fakeAdapter{
+		handlers: map[string]api.UpstreamConnectHandler{
+			"plugin-a": func(api.UpstreamConnectRequest) (net.Conn, error) {
+				return writer, nil
+			},
+		},
+	}
+	manager := newManagerForTest(t, adapter)
+	artifact := uploadTestArtifactWithCapabilities(t, manager, "plugin-a", json.RawMessage(`{"upstream_connect":{"mode":"protocol-proxy"}}`))
+	if _, err := manager.SetDesired(context.Background(), "admin", "plugin-a", artifact.ID, DesiredEnabled, `{"unused":true}`, 10); err != nil {
+		t.Fatalf("SetDesired() error = %v", err)
+	}
+	if _, err := manager.Enable(context.Background(), "admin", "plugin-a"); err != nil {
+		t.Fatalf("Enable() error = %v", err)
+	}
+	sourceGateway, sourceClient := net.Pipe()
+	defer sourceGateway.Close()
+	defer sourceClient.Close()
+
+	done := make(chan error, 1)
+	go func() {
+		initial := bytes.Repeat([]byte("x"), 2*1024*1024)
+		_, err := manager.ConnectUpstream(context.Background(), api.UpstreamConnectRequest{
+			Source:      sourceGateway,
+			InitialData: initial,
+		})
+		done <- err
+	}()
+	select {
+	case err := <-done:
+		if err == nil {
+			t.Fatal("ConnectUpstream() error = nil, want initial replay failure")
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("ConnectUpstream() did not return after initial write deadline")
+	}
+}
+
+func enableProtocolProxyTestPlugin(t *testing.T, manager *Manager, pluginID string) ArtifactRecord {
+	t.Helper()
+	artifact := uploadTestArtifactWithCapabilities(t, manager, pluginID, json.RawMessage(`{"upstream_connect":{"mode":"protocol-proxy"}}`))
+	if _, err := manager.SetDesired(context.Background(), "admin", pluginID, artifact.ID, DesiredEnabled, `{}`, 10); err != nil {
+		t.Fatalf("SetDesired() error = %v", err)
+	}
+	if _, err := manager.Enable(context.Background(), "admin", pluginID); err != nil {
+		t.Fatalf("Enable() error = %v", err)
+	}
+	return artifact
+}
+
 func newManagerForTest(t *testing.T, adapter RuntimeAdapter) *Manager {
 	t.Helper()
 	db := openPluginManagerTestDB(t)
@@ -264,9 +478,14 @@ func openPluginManagerTestDB(t *testing.T) *sql.DB {
 }
 
 func uploadTestArtifact(t *testing.T, manager *Manager, pluginID string) ArtifactRecord {
+	t.Helper()
+	return uploadTestArtifactWithCapabilities(t, manager, pluginID, nil)
+}
+
+func uploadTestArtifactWithCapabilities(t *testing.T, manager *Manager, pluginID string, capabilities json.RawMessage) ArtifactRecord {
 	t.Helper()
 	packagePath := writeTestMCGP(t, map[string][]byte{
-		"manifest.json": testManifestBytes(t, pluginID),
+		"manifest.json": testManifestBytesWithCapabilities(t, pluginID, capabilities),
 		"plugin.so":     []byte("fake plugin bytes " + pluginID),
 	})
 	artifact, err := manager.UploadArtifact(context.Background(), ArtifactUpload{
@@ -280,6 +499,23 @@ func uploadTestArtifact(t *testing.T, manager *Manager, pluginID string) Artifac
 	return artifact
 }
 
+func waitForPluginManagerTest(t *testing.T, done func() bool) {
+	t.Helper()
+	deadline := time.After(2 * time.Second)
+	ticker := time.NewTicker(time.Millisecond)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-deadline:
+			t.Fatal("timed out waiting for plugin manager condition")
+		case <-ticker.C:
+			if done() {
+				return
+			}
+		}
+	}
+}
+
 type fakeAdapter struct {
 	loads    int
 	handlers map[string]api.UpstreamConnectHandler

internal/pluginmanager/types.go

@@ -20,6 +20,9 @@ const (
 
 	ExtensionUpstreamConnect = "upstream.connect/v1"
 
+	UpstreamModeDialer        = "dialer"
+	UpstreamModeProtocolProxy = "protocol-proxy"
+
 	ArtifactStatusUploaded  = "uploaded"
 	ArtifactStatusValidated = "validated"
 	ArtifactStatusLoadable  = "loadable"
@@ -36,14 +39,16 @@ const (
 	RuntimeEnabled   = "enabled"
 	RuntimeFailed    = "failed"
 	RuntimeDisabled  = "disabled"
+	RuntimeDraining  = "draining"
 
-	DefaultPriority           = 100
-	DefaultHandlerTimeout     = 3 * time.Second
-	DefaultManifestMaxBytes   = 256 * 1024
-	DefaultPackageMaxBytes    = 64 * 1024 * 1024
-	DefaultPackageMaxEntries  = 2048
-	DefaultExtractedMaxBytes  = 256 * 1024 * 1024
-	DefaultNonRuntimeMaxBytes = 16 * 1024 * 1024
+	DefaultPriority            = 100
+	DefaultHandlerTimeout      = 3 * time.Second
+	DefaultManifestMaxBytes    = 256 * 1024
+	DefaultPackageMaxBytes     = 64 * 1024 * 1024
+	DefaultPackageMaxEntries   = 2048
+	DefaultExtractedMaxBytes   = 256 * 1024 * 1024
+	DefaultNonRuntimeMaxBytes  = 16 * 1024 * 1024
+	DefaultInitialWriteTimeout = time.Second
 )
 
 var (
@@ -85,7 +90,40 @@ type ExtensionPoint struct {
 }
 
 type RuntimeLimits struct {
-	HandlerTimeoutMS int `json:"handler_timeout_ms"`
+	HandlerTimeoutMS      int `json:"handler_timeout_ms"`
+	InitialWriteTimeoutMS int `json:"initial_write_timeout_ms"`
+}
+
+type CapabilitySummary struct {
+	UpstreamConnect UpstreamConnectCapability `json:"upstream_connect,omitempty"`
+	Minecraft       *MinecraftCapability      `json:"minecraft,omitempty"`
+	Raw             json.RawMessage           `json:"raw,omitempty"`
+}
+
+type UpstreamConnectCapability struct {
+	Mode string `json:"mode,omitempty"`
+}
+
+type MinecraftCapability struct {
+	ProtocolVersions  MinecraftProtocolVersions `json:"protocol_versions,omitempty"`
+	States            map[string]string         `json:"states,omitempty"`
+	AuthModes         []string                  `json:"auth_modes,omitempty"`
+	Forwarding        MinecraftForwarding       `json:"forwarding,omitempty"`
+	UnsupportedPolicy string                    `json:"unsupported_policy,omitempty"`
+	Modded            map[string]string         `json:"modded,omitempty"`
+}
+
+type MinecraftProtocolVersions struct {
+	Min               int    `json:"min,omitempty"`
+	Max               int    `json:"max,omitempty"`
+	Tested            []int  `json:"tested,omitempty"`
+	UnsupportedPolicy string `json:"unsupported_policy,omitempty"`
+}
+
+type MinecraftForwarding struct {
+	Supported      []string `json:"supported,omitempty"`
+	Default        string   `json:"default,omitempty"`
+	RequiresSecret bool     `json:"requires_secret,omitempty"`
 }
 
 type ArtifactRecord struct {
@@ -163,21 +201,35 @@ type DispatchPlan struct {
 }
 
 type DispatchHandlerSummary struct {
-	PluginID       string `json:"plugin_id"`
-	ArtifactID     string `json:"artifact_id"`
-	Priority       int    `json:"priority"`
-	HandlerID      string `json:"handler_id"`
-	ExtensionPoint string `json:"extension_point"`
-	TimeoutMS      int64  `json:"timeout_ms"`
-	Calls          uint64 `json:"calls"`
-	Errors         uint64 `json:"errors"`
-	Panics         uint64 `json:"panics"`
-	Timeouts       uint64 `json:"timeouts"`
+	PluginID        string `json:"plugin_id"`
+	ArtifactID      string `json:"artifact_id"`
+	Priority        int    `json:"priority"`
+	HandlerID       string `json:"handler_id"`
+	ExtensionPoint  string `json:"extension_point"`
+	Mode            string `json:"mode"`
+	TimeoutMS       int64  `json:"timeout_ms"`
+	Calls           uint64 `json:"calls"`
+	Errors          uint64 `json:"errors"`
+	Panics          uint64 `json:"panics"`
+	Timeouts        uint64 `json:"timeouts"`
+	Blocked         uint64 `json:"blocked"`
+	ActiveProxy     int64  `json:"active_proxy_connections"`
+	ProxyStarted    uint64 `json:"proxy_connections_started"`
+	ProxyCompleted  uint64 `json:"proxy_connections_completed"`
+	ProxyErrors     uint64 `json:"proxy_errors"`
+	ProxyBytesIn    uint64 `json:"proxy_bytes_in"`
+	ProxyBytesOut   uint64 `json:"proxy_bytes_out"`
+	ProxyDurationMS uint64 `json:"proxy_duration_ms"`
 }
 
 type UpstreamResult struct {
-	Conn    net.Conn
-	Handled bool
+	Conn            net.Conn
+	Handled         bool
+	Mode            string
+	PluginID        string
+	HandlerID       string
+	InitialDataSent bool
+	Proxied         bool
 }
 
 type Gateway struct {