fix(ssf): 支持断开后安全重连

重建 Stream 时仅重置流级 Verification 状态，保留历史收据和撤销水位；允许显式恢复失败连接，并在进程重启后按持久化时间继续 bootstrap。\n\n已通过 pnpm test、pnpm lint、pnpm build、go vet、PostgreSQL 集成测试以及本地真实断开重连和停机重试验收。
2026-07-16 12:30:14 +08:00 · 2026-07-16 12:30:14 +08:00 · 8e33d1d33e
commit 8e33d1d33e
parent 192e924dfb
4 changed files with 127 additions and 15 deletions
--- a/apps/api/internal/securityevents/connection_manager.go
+++ b/apps/api/internal/securityevents/connection_manager.go
@ -167,6 +167,9 @@ func NewConnectionManager(ctx context.Context, repository ConnectionRepository,
 			_, _ = repository.UpdateSecurityEventConnectionLifecycle(ctx, connection.ConnectionID, "degraded", &category)
 		}
 	}
+	if connection.LifecycleStatus == "bootstrap" {
+		go manager.finishBootstrap(connection.ConnectionID)
+	}
 	return manager, nil
 }

@ -193,7 +196,7 @@ func (m *ConnectionManager) Connect(ctx context.Context, issuer, idempotencyKey
 		if existing.TransmitterIssuer != issuer {
 			return ConnectionView{}, store.ErrSecurityEventConnectionConflict
 		}
-		if existing.StreamID != nil {
+		if existing.StreamID != nil && !resumableConnectionLifecycle(existing.LifecycleStatus) {
 			return m.view(existing), nil
 		}
 		return m.resumeConnecting(ctx, existing)
@ -223,6 +226,15 @@ func (m *ConnectionManager) Connect(ctx context.Context, issuer, idempotencyKey
 	return m.resumeConnecting(ctx, connection)
 }

+func resumableConnectionLifecycle(status string) bool {
+	switch status {
+	case "connecting", "verifying", "degraded", "error":
+		return true
+	default:
+		return false
+	}
+}
+
 func (m *ConnectionManager) resumeConnecting(ctx context.Context, connection store.SecurityEventConnection) (ConnectionView, error) {
 	secret, err := m.secrets.Get(ctx, connection.CredentialRef)
 	if err != nil {
@ -594,12 +606,19 @@ func (m *ConnectionManager) finishAutomaticVerification(connectionID string, con
 }

 func (m *ConnectionManager) finishBootstrap(connectionID string) {
-	select {
-	case <-m.ctx.Done():
-		return
-	case <-time.After(m.config.BootstrapDuration):
-	}
 	connection, err := m.repository.SecurityEventConnection(m.ctx)
+	if err != nil || connection.ConnectionID != connectionID || connection.LifecycleStatus != "bootstrap" {
+		return
+	}
+	delay := m.config.BootstrapDuration - time.Since(connection.UpdatedAt)
+	if delay > 0 {
+		select {
+		case <-m.ctx.Done():
+			return
+		case <-time.After(delay):
+		}
+	}
+	connection, err = m.repository.SecurityEventConnection(m.ctx)
 	if err == nil && connection.ConnectionID == connectionID && connection.LifecycleStatus == "bootstrap" {
 		mode, _, metricsErr := m.repository.SecurityEventMetrics(m.ctx, connection.TransmitterIssuer, valueOrEmpty(connection.Audience))
 		if metricsErr == nil && mode == "push_healthy" {
--- a/apps/api/internal/securityevents/connection_manager_test.go
+++ b/apps/api/internal/securityevents/connection_manager_test.go
@ -130,12 +130,59 @@ func TestConnectionManagerBootstrapOnlyBecomesEnabledWhenPushIsHealthy(t *testin
 	}
 }

+func TestConnectionManagerResumesBootstrapAfterRestart(t *testing.T) {
+	repository := &memoryConnectionRepository{mode: "push_healthy", connection: &store.SecurityEventConnection{
+		ConnectionID: uuid.NewString(), TransmitterIssuer: "https://auth.example/ssf",
+		LifecycleStatus: "bootstrap", Version: 1, UpdatedAt: time.Now().Add(-time.Second),
+	}}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	if _, err := NewConnectionManager(ctx, repository, &memorySecretStore{}, ConnectionManagerConfig{
+		OIDCEnabled: true, OIDCIssuer: "https://auth.example/issuer", OIDCTenantID: uuid.NewString(),
+		BootstrapDuration: 10 * time.Millisecond,
+	}, &Metrics{}); err != nil {
+		t.Fatal(err)
+	}
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		connection, err := repository.SecurityEventConnection(ctx)
+		if err == nil && connection.LifecycleStatus == "enabled" {
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatal("bootstrap lifecycle did not resume after process restart")
+}
+
 func TestRemoteNotFoundIsRecognizedForIdempotentDisconnect(t *testing.T) {
 	if !isRemoteHTTPStatus(remoteHTTPError{status: http.StatusNotFound}, http.StatusNotFound) {
 		t.Fatal("remote 404 was not recognized as an already-deleted Stream")
 	}
 }

+func TestConnectionLifecycleCanResumeOnlyIncompleteOrFailedConnections(t *testing.T) {
+	for _, test := range []struct {
+		status string
+		want   bool
+	}{
+		{status: "connecting", want: true},
+		{status: "verifying", want: true},
+		{status: "degraded", want: true},
+		{status: "error", want: true},
+		{status: "bootstrap"},
+		{status: "enabled"},
+		{status: "rotating"},
+		{status: "disconnect_pending"},
+		{status: "retiring"},
+	} {
+		t.Run(test.status, func(t *testing.T) {
+			if got := resumableConnectionLifecycle(test.status); got != test.want {
+				t.Fatalf("resumableConnectionLifecycle(%q)=%t want %t", test.status, got, test.want)
+			}
+		})
+	}
+}
+
 func TestTransmitterSSRFProtectionBlocksReservedNetworks(t *testing.T) {
 	for _, address := range []string{
 		"127.0.0.1", "10.0.0.1", "169.254.169.254", "100.64.0.1", "192.0.2.1", "198.51.100.1", "203.0.113.1", "2001:db8::1",
--- a/apps/api/internal/store/security_events.go
+++ b/apps/api/internal/store/security_events.go
@ -108,19 +108,38 @@ WHERE session.gateway_user_id = gateway_user.id
 }

 func (s *Store) EnsureSecurityEventStreamState(ctx context.Context, issuer, audience, streamID string) error {
-	tag, err := s.pool.Exec(ctx, `
-INSERT INTO gateway_security_event_stream_state(issuer,audience,stream_id,mode)
-VALUES($1,$2,$3::uuid,'bootstrap')
-ON CONFLICT(issuer,audience) DO UPDATE
-SET stream_id=EXCLUDED.stream_id,updated_at=now()
-WHERE gateway_security_event_stream_state.stream_id=EXCLUDED.stream_id`, issuer, audience, streamID)
+	tx, err := s.pool.Begin(ctx)
 	if err != nil {
 		return err
 	}
-	if tag.RowsAffected() == 0 {
-		return errors.New("security event stream id cannot be changed")
+	defer func() { _ = tx.Rollback(ctx) }()
+	if _, err := tx.Exec(ctx, `
+INSERT INTO gateway_security_event_stream_state(issuer,audience,stream_id,mode)
+VALUES($1,$2,$3::uuid,'bootstrap')
+ON CONFLICT(issuer,audience) DO NOTHING`, issuer, audience, streamID); err != nil {
+		return err
 	}
-	return nil
+	var currentStreamID string
+	if err := tx.QueryRow(ctx, `SELECT stream_id::text FROM gateway_security_event_stream_state
+		WHERE issuer=$1 AND audience=$2 FOR UPDATE`, issuer, audience).Scan(&currentStreamID); err != nil {
+		return err
+	}
+	if currentStreamID != streamID {
+		if _, err := tx.Exec(ctx, `DELETE FROM gateway_security_event_verification_challenges
+			WHERE issuer=$1 AND audience=$2`, issuer, audience); err != nil {
+			return err
+		}
+		if _, err := tx.Exec(ctx, `
+UPDATE gateway_security_event_stream_state
+SET stream_id=$3::uuid,mode='bootstrap',stream_status='unknown',
+    last_verification_at=NULL,bootstrap_until=NULL,consecutive_verifications=0,
+    pending_state_hash=NULL,pending_state_created_at=NULL,fallback_since=NULL,
+    last_error_category=NULL,created_at=now(),updated_at=now()
+WHERE issuer=$1 AND audience=$2`, issuer, audience, streamID); err != nil {
+			return err
+		}
+	}
+	return tx.Commit(ctx)
 }

 func (s *Store) BeginSecurityEventVerification(ctx context.Context, issuer, audience string, stateHash []byte, now time.Time) error {
--- a/apps/api/internal/store/security_events_integration_test.go
+++ b/apps/api/internal/store/security_events_integration_test.go
@ -155,4 +155,31 @@ func TestSecurityEventWatermarkVerificationAndFallbackLifecycle(t *testing.T) {
 	if lifecycle != "enabled" || lastError != nil {
 		t.Fatalf("recovered connection lifecycle=%q lastError=%v", lifecycle, lastError)
 	}
+	replacementStreamID := uuid.NewString()
+	if err := db.EnsureSecurityEventStreamState(ctx, issuer, audience, replacementStreamID); err != nil {
+		t.Fatalf("rebind retained receiver state: %v", err)
+	}
+	var reboundStreamID, reboundMode, reboundStatus string
+	var reboundVerification *time.Time
+	if err := db.pool.QueryRow(ctx, `SELECT stream_id::text,mode,stream_status,last_verification_at
+		FROM gateway_security_event_stream_state WHERE issuer=$1 AND audience=$2`, issuer, audience).
+		Scan(&reboundStreamID, &reboundMode, &reboundStatus, &reboundVerification); err != nil {
+		t.Fatal(err)
+	}
+	if reboundStreamID != replacementStreamID || reboundMode != "bootstrap" || reboundStatus != "unknown" || reboundVerification != nil {
+		t.Fatalf("rebound state stream=%q mode=%q status=%q verification=%v",
+			reboundStreamID, reboundMode, reboundStatus, reboundVerification)
+	}
+	var retainedReceipts, retainedWatermarks int
+	if err := db.pool.QueryRow(ctx, `SELECT count(*) FROM gateway_security_event_receipts
+		WHERE issuer=$1`, issuer).Scan(&retainedReceipts); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.pool.QueryRow(ctx, `SELECT count(*) FROM gateway_oidc_revocation_watermarks
+		WHERE issuer=$1 AND tenant_id=$2 AND subject=$3`, subjectIssuer, tenantID, subject).Scan(&retainedWatermarks); err != nil {
+		t.Fatal(err)
+	}
+	if retainedReceipts == 0 || retainedWatermarks != 1 {
+		t.Fatalf("rebind discarded security history receipts=%d watermarks=%d", retainedReceipts, retainedWatermarks)
+	}
 }