feat(gateway): report provider lifecycle

2026-07-29 10:55:19 +07:00
parent 61a17ff42b
commit 8a302cd8cc
7 changed files with 99 additions and 21 deletions
@@ -76,6 +76,15 @@ func (c *ControlPlaneClient) Release(ctx context.Context, authority protocol.Ses
 	return err
 }

+func (c *ControlPlaneClient) ReportProviderState(ctx context.Context, state protocol.ProviderState) error {
+	payload, err := protocol.EncodeProviderState(state)
+	if err != nil {
+		return err
+	}
+	_, err = c.post(ctx, "/api/v1/gateway/provider-state", payload)
+	return err
+}
+
 func (c *ControlPlaneClient) post(ctx context.Context, path string, payload []byte) ([]byte, error) {
 	if c == nil || c.HTTPClient == nil || c.BaseURL == "" {
 		return nil, errors.New("control-plane client is not configured")
@@ -14,6 +14,7 @@ import (
 	"net"
 	"os"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -238,7 +239,8 @@ func TestAdmissionQUICMTLSRelayAndCleanup(t *testing.T) {
 	fake := NewFakeApollo(FakeApolloConfig{Now: time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)})
 	authority := protocol.SessionAuthority{Version: "1", SessionID: "session-1", GatewayID: "gateway-1", Audience: "versevdi-gateway", ReconnectSequence: 0, ExpiresAt: time.Now().Add(5 * time.Second).UTC().Format(time.RFC3339Nano), Capabilities: DefaultCapabilities(), ProviderProfile: ProviderProfileApollo, ProviderIdentity: fake.config.Identity.Key()}
 	admission := &oneTimeAdmission{authority: authority, released: make(chan struct{})}
-	server, err := NewServer(ServerConfig{ListenAddress: "127.0.0.1:0", TLSConfig: serverTLS, GatewayID: "gateway-1", Capabilities: DefaultCapabilities(), ProviderCapabilities: DefaultCapabilities(), Admission: admission, Provider: fake})
+	reporter := &recordingProviderStateReporter{}
+	server, err := NewServer(ServerConfig{ListenAddress: "127.0.0.1:0", TLSConfig: serverTLS, GatewayID: "gateway-1", Capabilities: DefaultCapabilities(), ProviderCapabilities: DefaultCapabilities(), Admission: admission, ProviderStateReporter: reporter, Provider: fake})
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -246,7 +248,7 @@ func TestAdmissionQUICMTLSRelayAndCleanup(t *testing.T) {
 	defer cancel()
 	serveDone := make(chan error, 1)
 	go func() { serveDone <- server.Serve(ctx) }()
-	request := protocol.TunnelAdmissionRequest{Version: "1", SessionID: "session-1", GatewayID: "gateway-1", Audience: "versevdi-gateway", Grant: strings.Repeat("g", 64), ReconnectSequence: 0, ClientNonce: "nonce-0000000001", Capabilities: DefaultCapabilities()}
+	request := protocol.TunnelAdmissionRequest{Version: "1", SessionID: "session-1", GatewayID: "gateway-1", Audience: "versevdi-gateway", Grant: strings.Repeat("g", 64), ReconnectSequence: 0, ClientNonce: "nonce-0000000001", DeviceSignature: strings.Repeat("s", 86), Capabilities: DefaultCapabilities()}
 	client, err := Dial(context.Background(), server.Addr().String(), clientTLS, request)
 	if err != nil {
 		_ = server.Close()
@@ -282,6 +284,10 @@ func TestAdmissionQUICMTLSRelayAndCleanup(t *testing.T) {
 	if err := <-serveDone; err != nil {
 		t.Fatal(err)
 	}
+	states := reporter.States()
+	if len(states) != 3 || states[0].State != ProviderStateStarting || states[1].State != ProviderStateReady || states[2].State != ProviderStateTerminated {
+		t.Fatalf("provider states = %#v", states)
+	}
 }

 func testTLS(t *testing.T) (*tls.Config, *tls.Config) {
@@ -322,6 +328,24 @@ type oneTimeAdmission struct {
 	released  chan struct{}
 }

+type recordingProviderStateReporter struct {
+	mu     sync.Mutex
+	states []protocol.ProviderState
+}
+
+func (r *recordingProviderStateReporter) ReportProviderState(_ context.Context, state protocol.ProviderState) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.states = append(r.states, state)
+	return nil
+}
+
+func (r *recordingProviderStateReporter) States() []protocol.ProviderState {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return append([]protocol.ProviderState(nil), r.states...)
+}
+
 func (a *oneTimeAdmission) Admit(context.Context, protocol.TunnelAdmissionRequest) (protocol.SessionAuthority, error) {
 	if !a.used.CompareAndSwap(false, true) {
 		return protocol.SessionAuthority{}, ErrAdmissionRejected
@@ -338,12 +338,12 @@ func (f *FakeApollo) Setup(context.Context, LaunchRequest) ([]byte, error) {
 	return []byte("RTSP/1.0 200 OK\r\nSession: fixture-session\r\nTransport: RTP/AVP/TCP;interleaved=0-1\r\n\r\n"), nil
 }

-func (f *FakeApollo) Open(context.Context, LaunchRequest, RTSPResponse) (ProviderSession, error) {
+func (f *FakeApollo) Open(_ context.Context, request LaunchRequest, _ RTSPResponse) (ProviderSession, error) {
 	session := &fakeSession{
 		failure: f.config.Failure,
 		video:   make(chan []byte, 16),
 		audio:   make(chan []byte, 16),
-		state:   protocol.ProviderState{Version: "1", State: ProviderStateStarting, Channels: []string{"video", "audio", "input", "feedback"}},
+		state:   protocol.ProviderState{Version: "1", SessionID: request.SessionID, State: ProviderStateStarting, Channels: []string{"video", "audio", "input", "feedback"}},
 		pressed: make(map[string]struct{}),
 	}
 	for _, payload := range f.config.Video {
@@ -44,18 +44,23 @@ func (f AdmissionFunc) Admit(ctx context.Context, request protocol.TunnelAdmissi

 func (AdmissionFunc) Release(context.Context, protocol.SessionAuthority) error { return nil }

+type ProviderStateReporter interface {
+	ReportProviderState(context.Context, protocol.ProviderState) error
+}
+
 type ServerConfig struct {
-	ListenAddress        string
-	TLSConfig            *tls.Config
-	QUICConfig           *quic.Config
-	GatewayID            string
-	Capabilities         protocol.CapabilityProfile
-	ProviderCapabilities protocol.CapabilityProfile
-	Admission            Admission
-	Provider             Provider
-	ProviderProfile      string
-	ProviderIdentity     string
-	PacerKbps            int64
+	ListenAddress         string
+	TLSConfig             *tls.Config
+	QUICConfig            *quic.Config
+	GatewayID             string
+	Capabilities          protocol.CapabilityProfile
+	ProviderCapabilities  protocol.CapabilityProfile
+	Admission             Admission
+	ProviderStateReporter ProviderStateReporter
+	Provider              Provider
+	ProviderProfile       string
+	ProviderIdentity      string
+	PacerKbps             int64
 }

 type Server struct {
@@ -212,13 +217,26 @@ func (s *Server) handleConnection(parent context.Context, connection *quic.Conn)
 		_ = writeStableError(stream, "no_capability_overlap", err, false)
 		return
 	}
+	if err := s.reportProviderState(ctx, protocol.ProviderState{Version: "1", SessionID: request.SessionID, State: ProviderStateStarting, CleanupPending: false, Channels: []string{"video", "audio", "input", "feedback"}}); err != nil {
+		_ = s.config.Admission.Release(context.Background(), authority)
+		_ = writeStableError(stream, "provider_state_unavailable", err, true)
+		return
+	}
 	providerSession, err := s.config.Provider.Start(ctx, LaunchRequest{SessionID: request.SessionID, Capabilities: selected, ProviderProfile: authority.ProviderProfile, ProviderIdentity: authority.ProviderIdentity})
 	if err != nil {
 		s.metrics.ProviderErrors.Add(1)
+		_ = s.reportProviderState(context.Background(), protocol.ProviderState{Version: "1", SessionID: request.SessionID, State: ProviderStateFailed, CleanupPending: false, Channels: []string{"video", "audio", "input", "feedback"}})
 		_ = s.config.Admission.Release(context.Background(), authority)
 		_ = writeStableError(stream, stableProviderCode(err), err, errors.Is(err, context.DeadlineExceeded))
 		return
 	}
+	if err := s.reportProviderState(ctx, providerSession.State()); err != nil {
+		_ = providerSession.ReleaseAll(context.Background())
+		_ = providerSession.Terminate(context.Background())
+		_ = s.config.Admission.Release(context.Background(), authority)
+		_ = writeStableError(stream, "provider_state_unavailable", err, true)
+		return
+	}
 	authority.Capabilities = selected
 	authorityBytes, err := protocol.EncodeSessionAuthority(authority)
 	if err != nil || writeWire(stream, authorityBytes, defaultHelloLimit) != nil {
@@ -237,6 +255,16 @@ func (s *Server) handleConnection(parent context.Context, connection *quic.Conn)
 	session.run()
 }

+func (s *Server) reportProviderState(ctx context.Context, state protocol.ProviderState) error {
+	if s.config.ProviderStateReporter == nil {
+		return nil
+	}
+	if err := state.Validate(); err != nil {
+		return err
+	}
+	return s.config.ProviderStateReporter.ReportProviderState(ctx, state)
+}
+
 func (s *Server) validateAuthority(authority protocol.SessionAuthority, request protocol.TunnelAdmissionRequest) error {
 	if err := authority.Validate(); err != nil {
 		return err
@@ -467,15 +495,32 @@ func (s *gatewaySession) cleanup() {
 		s.cancel()
 		cleanupCtx, cancel := context.WithTimeout(context.Background(), time.Second)
 		defer cancel()
-		if err := s.provider.ReleaseAll(cleanupCtx); err != nil {
+		releaseInputsErr := s.provider.ReleaseAll(cleanupCtx)
+		if releaseInputsErr != nil {
 			s.server.metrics.ProviderErrors.Add(1)
 		}
-		if err := s.provider.Terminate(cleanupCtx); err != nil {
+		terminateErr := s.provider.Terminate(cleanupCtx)
+		if terminateErr != nil {
 			s.server.metrics.ProviderErrors.Add(1)
 		}
+		state := s.provider.State()
+		if releaseInputsErr != nil || terminateErr != nil {
+			state.State = ProviderStateCleanup
+			state.CleanupPending = true
+			if err := s.server.reportProviderState(cleanupCtx, state); err != nil {
+				s.server.metrics.ProviderErrors.Add(1)
+			} else if err := s.server.config.Admission.Release(cleanupCtx, s.authority); err != nil {
+				s.server.metrics.ProviderErrors.Add(1)
+			}
+			_ = s.connection.CloseWithError(applicationError, "session closed")
+			return
+		}
 		if err := s.server.config.Admission.Release(cleanupCtx, s.authority); err != nil {
 			s.server.metrics.ProviderErrors.Add(1)
 		}
+		if err := s.server.reportProviderState(cleanupCtx, state); err != nil {
+			s.server.metrics.ProviderErrors.Add(1)
+		}
 		_ = s.connection.CloseWithError(applicationError, "session closed")
 	})
 }