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

fix(identity): 允许清理未收到 Manifest 的配对

Browse Source 
0090 的 Manifest 约束要求无 issuer 的 Revision 保持 draft，原取消流程将其改为 failed，导致取消事务被数据库拒绝。现在由 cancelled Pairing 记录终态，并仅允许明确标记 pairing_cancelled 的预 Manifest 草稿完成清理。\n\n验证：IdentityPairing PostgreSQL 定向测试通过；PairingService 定向测试通过。
This commit is contained in:
chengcheng
2026-07-28 18:42:13 +08:00
parent 206e367dbf
commit 021ee9ab8f
2 changed files with 67 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/internal/store/identity_pairing.go
+8 -2
View File
@@ -320,7 +320,10 @@ FROM gateway_identity_configuration_revisions WHERE id=$1::uuid FOR UPDATE`, exc
if revision.State != identity.RevisionDraft && revision.State != identity.RevisionValidated && revision.State != identity.RevisionFailed { if revision.State != identity.RevisionDraft && revision.State != identity.RevisionValidated && revision.State != identity.RevisionFailed {
return identity.PairingExchange{}, identity.ErrPairingNotCancellable return identity.PairingExchange{}, identity.ErrPairingNotCancellable
} }
if _, err := tx.Exec(ctx, `UPDATE gateway_identity_configuration_revisions SET state='failed', // Before Manifest delivery, issuer is NULL and the schema requires the
// revision to remain draft. The cancelled pairing is the terminal record.
if _, err := tx.Exec(ctx, `UPDATE gateway_identity_configuration_revisions SET
state=CASE WHEN issuer IS NULL AND state IN ('draft','failed') THEN 'draft' ELSE 'failed' END,
last_error_category='pairing_cancelled',last_trace_id=NULLIF($2,''),last_audit_id=NULLIF($3,''), last_error_category='pairing_cancelled',last_trace_id=NULLIF($2,''),last_audit_id=NULLIF($3,''),
version=version+1,updated_at=now() WHERE id=$1::uuid`, exchange.RevisionID, traceID, auditID); err != nil { version=version+1,updated_at=now() WHERE id=$1::uuid`, exchange.RevisionID, traceID, auditID); err != nil {
return identity.PairingExchange{}, err return identity.PairingExchange{}, err
@@ -358,7 +361,10 @@ WHERE id=$1::uuid AND version=$2 AND status='cancelled' AND cleanup_status='pend
return identity.PairingExchange{}, err return identity.PairingExchange{}, err
} }
tag, err := tx.Exec(ctx, `UPDATE gateway_identity_configuration_revisions SET machine_credential_ref=NULL, tag, err := tx.Exec(ctx, `UPDATE gateway_identity_configuration_revisions SET machine_credential_ref=NULL,
session_encryption_key_ref=NULL,version=version+1,updated_at=now() WHERE id=$1::uuid AND state='failed'`, revisionID) session_encryption_key_ref=NULL,version=version+1,updated_at=now()
WHERE id=$1::uuid AND (
state='failed' OR (state='draft' AND issuer IS NULL AND last_error_category='pairing_cancelled')
)`, revisionID)
if err != nil { if err != nil {
return identity.PairingExchange{}, err return identity.PairingExchange{}, err
} }
apps/api/internal/store/identity_pairing_integration_test.go
+59 -9
View File
@@ -16,6 +16,51 @@ import (
"github.com/jackc/pgx/v5/pgxpool" "github.com/jackc/pgx/v5/pgxpool"
) )
func TestIdentityPairingCancellationPreservesPreManifestDraftUntilCleanupCompletes(t *testing.T) {
db := newIdentityPairingPostgresTestStore(t)
ctx := context.Background()
revisionID := uuid.NewString()
pairingID := uuid.NewString()
if _, err := db.pool.Exec(ctx, `
INSERT INTO gateway_identity_configuration_revisions (
id,state,auth_center_url,local_tenant_key,public_base_url,web_base_url
) VALUES ($1::uuid,'draft','https://auth.test.example','','https://gateway.test.example',
'https://gateway-web.test.example')`, revisionID); err != nil {
t.Fatalf("seed pre-manifest revision: %v", err)
}
if _, err := db.CreateIdentityPairingExchange(ctx, identity.PairingExchange{
ID: pairingID, RevisionID: revisionID, RemoteExchangeID: uuid.NewString(),
ExchangeTokenRef: "identity-exchange-" + pairingID,
Status: identity.PairingMetadataPending, CleanupStatus: identity.PairingCleanupNone,
RemoteVersion: 1, ExpiresAt: time.Now().Add(time.Hour),
}); err != nil {
t.Fatalf("seed pre-manifest pairing: %v", err)
}
cancelled, err := db.CancelIdentityPairingExchange(ctx, pairingID, 1, "trace-cancel", "audit-cancel")
if err != nil {
t.Fatalf("cancel pre-manifest pairing: %v", err)
}
if cancelled.Status != identity.PairingCancelled || cancelled.CleanupStatus != identity.PairingCleanupPending {
t.Fatalf("cancelled pairing state=%#v", cancelled)
}
cleaned, err := db.CompleteIdentityPairingCleanup(ctx, pairingID, cancelled.Version)
if err != nil {
t.Fatalf("complete pre-manifest pairing cleanup: %v", err)
}
if cleaned.CleanupStatus != identity.PairingCleanupCompleted {
t.Fatalf("cleaned pairing state=%#v", cleaned)
}
revision, err := db.IdentityConfigurationRevision(ctx, revisionID)
if err != nil {
t.Fatalf("load cancelled pre-manifest revision: %v", err)
}
if revision.State != identity.RevisionDraft || revision.Issuer != "" || revision.LastErrorCategory != "pairing_cancelled" {
t.Fatalf("cancelled pre-manifest revision=%#v", revision)
}
}
func TestIdentityPairingCancellationKeepsOlderInactivePairingBlockedUntilEveryCleanupCompletes(t *testing.T) { func TestIdentityPairingCancellationKeepsOlderInactivePairingBlockedUntilEveryCleanupCompletes(t *testing.T) {
db := newIdentityPairingPostgresTestStore(t) db := newIdentityPairingPostgresTestStore(t)
ctx := context.Background() ctx := context.Background()
@@ -211,9 +256,10 @@ func TestIdentityPairingStartRejectsAnyActiveConfigurationBeforeReservation(t *t
db := newIdentityPairingPostgresTestStore(t) db := newIdentityPairingPostgresTestStore(t)
ctx := context.Background() ctx := context.Background()
if _, err := db.pool.Exec(ctx, `INSERT INTO gateway_identity_configuration_revisions ( if _, err := db.pool.Exec(ctx, `INSERT INTO gateway_identity_configuration_revisions (
id,state,auth_center_url,local_tenant_key,public_base_url,web_base_url id,state,auth_center_url,issuer,tenant_id,local_tenant_key,public_base_url,web_base_url
) VALUES ($1::uuid,'active','https://auth.test.example','default','https://gateway.test.example', ) VALUES ($1::uuid,'active','https://auth.test.example','https://issuer.test.example',$2,
'https://gateway-web.test.example')`, uuid.NewString()); err != nil { 'default','https://gateway.test.example','https://gateway-web.test.example')`,
uuid.NewString(), uuid.NewString()); err != nil {
t.Fatalf("seed active identity revision: %v", err) t.Fatalf("seed active identity revision: %v", err)
} }
@@ -389,9 +435,10 @@ func TestConcurrentIdentityPairingCommitRejectsActiveRevisionCommittedAheadOfLif
waitForIdentityLifecycleLockWaiters(t, ctx, db, 1) waitForIdentityLifecycleLockWaiters(t, ctx, db, 1)
activeID := uuid.NewString() activeID := uuid.NewString()
if _, err := gate.Exec(ctx, `INSERT INTO gateway_identity_configuration_revisions ( if _, err := gate.Exec(ctx, `INSERT INTO gateway_identity_configuration_revisions (
id,state,auth_center_url,local_tenant_key,public_base_url,web_base_url id,state,auth_center_url,issuer,tenant_id,local_tenant_key,public_base_url,web_base_url
) VALUES ($1::uuid,'active','https://auth.test.example','default','https://active-gateway.test.example', ) VALUES ($1::uuid,'active','https://auth.test.example','https://issuer.test.example',$2,
'https://active-gateway-web.test.example')`, activeID); err != nil { 'default','https://active-gateway.test.example','https://active-gateway-web.test.example')`,
activeID, uuid.NewString()); err != nil {
t.Fatalf("commit active revision ahead of pairing commit: %v", err) t.Fatalf("commit active revision ahead of pairing commit: %v", err)
} }
if err := gate.Commit(ctx); err != nil { if err := gate.Commit(ctx); err != nil {
@@ -493,9 +540,10 @@ func TestCompletedIdentityPairingRestoreUsesCanonicalVisibleRevision(t *testing.
revisionID, pairingID := uuid.NewString(), uuid.NewString() revisionID, pairingID := uuid.NewString(), uuid.NewString()
if _, err := db.pool.Exec(ctx, ` if _, err := db.pool.Exec(ctx, `
INSERT INTO gateway_identity_configuration_revisions ( INSERT INTO gateway_identity_configuration_revisions (
id,state,auth_center_url,local_tenant_key,public_base_url,web_base_url,created_at,updated_at id,state,auth_center_url,issuer,tenant_id,local_tenant_key,public_base_url,web_base_url,created_at,updated_at
) VALUES ($1::uuid,'validated','https://auth.test.example','default','https://gateway.test.example', ) VALUES ($1::uuid,'validated','https://auth.test.example','https://issuer.test.example',$2,
'https://gateway-web.test.example',$2,$2)`, revisionID, createdAt); err != nil { 'default','https://gateway.test.example','https://gateway-web.test.example',$3,$3)`,
revisionID, uuid.NewString(), createdAt); err != nil {
t.Fatalf("seed completed revision %d: %v", index, err) t.Fatalf("seed completed revision %d: %v", index, err)
} }
if _, err := db.CreateIdentityPairingExchange(ctx, identity.PairingExchange{ if _, err := db.CreateIdentityPairingExchange(ctx, identity.PairingExchange{
@@ -687,11 +735,13 @@ func newIdentityPairingPostgresTestStore(t *testing.T) *Store {
"0001_init.sql", "0001_init.sql",
"0017_task_record_enrichment.sql", "0017_task_record_enrichment.sql",
"0061_oidc_server_sessions.sql", "0061_oidc_server_sessions.sql",
"0063_oidc_security_events.sql",
"0065_identity_configuration_revisions.sql", "0065_identity_configuration_revisions.sql",
"0066_identity_onboarding_exchanges.sql", "0066_identity_onboarding_exchanges.sql",
"0067_identity_secret_cleanup_queue.sql", "0067_identity_secret_cleanup_queue.sql",
"0068_identity_pairing_start_reservation.sql", "0068_identity_pairing_start_reservation.sql",
"0069_billing_correctness_v2.sql", "0069_billing_correctness_v2.sql",
"0090_oidc_multi_tenant_identity.sql",
} { } {
migration, err := os.ReadFile(filepath.Join(migrationDirectory, migrationName)) migration, err := os.ReadFile(filepath.Join(migrationDirectory, migrationName))
if err != nil { if err != nil {