BCAI/easyai-ai-gateway
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(ssf): 恢复健康连接生命周期

Browse Source 
连续两次有效 Verification 将 Stream 状态恢复为 push_healthy 时，同一事务同步把 degraded 连接恢复为 enabled、清除陈旧错误并推进版本，避免管理页永久误报降级。\n\n验证：真实 PostgreSQL 集成测试；pnpm test；pnpm lint；pnpm build；真实 Transmitter/Introspection 故障与恢复演练。
This commit is contained in:
chengcheng 2026-07-16 12:06:43 +08:00
parent 29f24222fd
commit 192e924dfb
2 changed files with 35 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
apps/api/internal/store/security_events.go
View File

@ -210,6 +210,18 @@ WHERE issuer=$1 AND audience=$2 AND stream_id=$3::uuid
WHERE issuer=$1 AND audience=$2 AND state_hash=$3`, issuer, audience, stateHash); err != nil {
return false, err
}
if _, err := tx.Exec(ctx, `
UPDATE gateway_security_event_connections connection
SET lifecycle_status='enabled',last_error_category=NULL,version=version+1,updated_at=now()
WHERE connection.transmitter_issuer=$1 AND connection.audience=$2
AND connection.lifecycle_status='degraded'
AND EXISTS(
SELECT 1 FROM gateway_security_event_stream_state state
WHERE state.issuer=$1 AND state.audience=$2
AND state.mode='push_healthy' AND state.stream_status='enabled'
)`, issuer, audience); err != nil {
return false, err
}
return true, tx.Commit(ctx)
}

23
apps/api/internal/store/security_events_integration_test.go
View File

@ -42,6 +42,7 @@ func TestSecurityEventWatermarkVerificationAndFallbackLifecycle(t *testing.T) {
audience := "urn:easyai:ssf:receiver:" + uuid.NewString()
streamID, tenantID, subject := uuid.NewString(), uuid.NewString(), uuid.NewString()
t.Cleanup(func() {
_, _ = db.pool.Exec(context.Background(), `DELETE FROM gateway_security_event_connections WHERE transmitter_issuer=$1`, issuer)
_, _ = db.pool.Exec(context.Background(), `DELETE FROM gateway_security_event_receipts WHERE issuer=$1`, issuer)
_, _ = db.pool.Exec(context.Background(), `DELETE FROM gateway_oidc_revocation_watermarks WHERE issuer=$1`, subjectIssuer)
_, _ = db.pool.Exec(context.Background(), `DELETE FROM gateway_security_event_stream_state WHERE issuer=$1`, issuer)
@ -132,4 +133,26 @@ func TestSecurityEventWatermarkVerificationAndFallbackLifecycle(t *testing.T) {
if err != nil || !evaluation.RequireIntrospection || evaluation.Mode != "introspection_fallback" {
t.Fatalf("fallback evaluation=%#v error=%v", evaluation, err)
}
connectionID := uuid.NewString()
if _, err := db.pool.Exec(ctx, `DELETE FROM gateway_security_event_connections`); err != nil {
t.Fatal(err)
}
if _, err := db.pool.Exec(ctx, `INSERT INTO gateway_security_event_connections(
connection_id,transmitter_issuer,endpoint_url,audience,stream_id,credential_ref,
lifecycle_status,last_error_category,idempotency_key)
VALUES($1,$2,'https://gateway.test/api/v1/security-events/ssf',$3,$4,'test-secret-ref',
'degraded','bootstrap_verification_stale',$5)`, connectionID, issuer, audience, streamID, uuid.NewString()); err != nil {
t.Fatal(err)
}
confirmAt("recovery-verification-one", now.Add(362*time.Second))
confirmAt("recovery-verification-two", now.Add(363*time.Second))
var lifecycle string
var lastError *string
if err := db.pool.QueryRow(ctx, `SELECT lifecycle_status,last_error_category
FROM gateway_security_event_connections WHERE connection_id=$1`, connectionID).Scan(&lifecycle, &lastError); err != nil {
t.Fatal(err)
}
if lifecycle != "enabled" || lastError != nil {
t.Fatalf("recovered connection lifecycle=%q lastError=%v", lifecycle, lastError)
}
}