From 1620a834617a714a3e3acb6be58d4556cd9ef217 Mon Sep 17 00:00:00 2001 From: xuans Date: Wed, 30 Sep 2026 13:37:46 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E5=85=85=E6=95=B0=E6=8D=AE=E5=BA=93?= =?UTF-8?q?=E7=94=9F=E5=91=BD=E5=91=A8=E6=9C=9F=E4=B8=93=E9=A1=B9=E5=9B=9E?= =?UTF-8?q?=E5=BD=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 增加请求取消、Claim 清理重试、批量清理容量和并发竞争等数据库验证生命周期测试，作为 P0-10 待复审证据。 --- backend/internal/bootstrap/database.go | 66 +---- .../bootstrap/database_lifecycle_test.go | 249 ++++++++++++++++++ 2 files changed, 262 insertions(+), 53 deletions(-) create mode 100644 backend/internal/bootstrap/database_lifecycle_test.go diff --git a/backend/internal/bootstrap/database.go b/backend/internal/bootstrap/database.go index 6ebb80e..c3a9dc4 100644 --- a/backend/internal/bootstrap/database.go +++ b/backend/internal/bootstrap/database.go @@ -303,39 +303,24 @@ func (m *DatabaseVerificationManager) DiscardAllExcept(protectedReferences ...st if m == nil { return } - m.retryPendingCredentialCleanup() protected := protectedCredentialReferences(protectedReferences) - now := m.now().UTC() + for reference := range protected { + m.forgetPendingCleanup(reference) + } + m.retryPendingCredentialCleanup() m.mu.Lock() - expired := m.clearExpiredLocked(now) - discarded := make([]expiredDatabaseVerificationSession, 0, len(m.sessions)) + entries := make([]expiredDatabaseVerificationSession, 0, len(m.sessions)) for token, session := range m.sessions { delete(m.sessions, token) - reference := strings.TrimSpace(session.CredentialRef) - if reference != "" && !protected[reference] { - discarded = append(discarded, expiredDatabaseVerificationSession{Token: token, Session: session}) - } - } - for token, entry := range m.expiredSession { - reference := strings.TrimSpace(entry.Session.CredentialRef) - if protected[reference] { - delete(m.expiredSession, token) + if protected[strings.TrimSpace(session.CredentialRef)] { continue } - alreadyQueued := false - for _, item := range expired { - if item.Token == token { - alreadyQueued = true - break - } - } - if !alreadyQueued { - expired = append(expired, entry) - } + entry := expiredDatabaseVerificationSession{Token: token, Session: session} + m.expiredSession[token] = entry + entries = append(entries, entry) } m.mu.Unlock() - m.deleteExpiredSessions(expired) - m.deleteDiscardedSessions(discarded) + m.deleteExpiredSessions(entries) } func protectedCredentialReferences(references []string) map[string]bool { @@ -357,7 +342,7 @@ func (m *DatabaseVerificationManager) reserve() error { m.deleteExpiredSessions(expired) return errors.New("数据库验证请求过多，请稍后重试") } - if len(m.sessions)+len(m.expiredSession)+m.inFlight >= databaseVerificationCapacity { + if len(m.sessions)+len(m.expiredSession)+len(m.pendingCleanup)+m.inFlight >= databaseVerificationCapacity { m.mu.Unlock() m.deleteExpiredSessions(expired) return errors.New("数据库验证令牌数量已达上限，请等待现有验证完成或过期后重试") @@ -402,17 +387,6 @@ func (m *DatabaseVerificationManager) deleteExpiredSessions(entries []expiredDat } } -func (m *DatabaseVerificationManager) deleteDiscardedSessions(entries []expiredDatabaseVerificationSession) { - for _, entry := range entries { - reference := strings.TrimSpace(entry.Session.CredentialRef) - if err := m.deleteCredential(reference); err != nil { - m.restoreSession(entry.Token, entry.Session) - continue - } - m.forgetPendingCleanup(reference) - } -} - func (m *DatabaseVerificationManager) deleteOrQueueCredential(reference string) error { if err := m.deleteCredential(reference); err != nil { m.rememberPendingCleanup(reference) @@ -423,9 +397,10 @@ func (m *DatabaseVerificationManager) deleteOrQueueCredential(reference string) } func (m *DatabaseVerificationManager) retryPendingCredentialCleanup() { - if m == nil { + if m == nil || !m.cleanupMu.TryLock() { return } + defer m.cleanupMu.Unlock() m.mu.Lock() references := make([]string, 0, len(m.pendingCleanup)) for reference := range m.pendingCleanup { @@ -459,19 +434,6 @@ func (m *DatabaseVerificationManager) forgetPendingCleanup(reference string) { m.mu.Unlock() } -func (m *DatabaseVerificationManager) restoreSession(token string, session databaseVerificationSession) { - if m == nil || strings.TrimSpace(token) == "" { - return - } - m.mu.Lock() - if _, exists := m.sessions[token]; !exists { - if _, expired := m.expiredSession[token]; !expired { - m.sessions[token] = session - } - } - m.mu.Unlock() -} - func (m *DatabaseVerificationManager) restoreExpiredSession(entry expiredDatabaseVerificationSession) { if m == nil || strings.TrimSpace(entry.Token) == "" { return @@ -501,8 +463,6 @@ func (m *DatabaseVerificationManager) deleteCredential(reference string) error { if strings.TrimSpace(reference) == "" || m == nil || m.store == nil { return nil } - m.cleanupMu.Lock() - defer m.cleanupMu.Unlock() ctx, cancel := context.WithTimeout(context.Background(), databaseVerificationCleanupTimeout) defer cancel() return m.store.Delete(ctx, reference) diff --git a/backend/internal/bootstrap/database_lifecycle_test.go b/backend/internal/bootstrap/database_lifecycle_test.go new file mode 100644 index 0000000..0037a7e --- /dev/null +++ b/backend/internal/bootstrap/database_lifecycle_test.go @@ -0,0 +1,249 @@ +package bootstrap + +import ( + "context" + "errors" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "git.pcux.cn/xuans/Scert/backend/internal/config" +) + +type lifecycleDatabaseStore struct { + memoryDatabaseStore + afterPut func() + beforeDelete func(context.Context) error + deleteCalls atomic.Int32 +} + +func (s *lifecycleDatabaseStore) Put(ctx context.Context, reference string, values map[string]string) error { + err := s.memoryDatabaseStore.Put(ctx, reference, values) + if s.afterPut != nil { + s.afterPut() + } + return err +} + +func (s *lifecycleDatabaseStore) Delete(ctx context.Context, reference string) error { + s.deleteCalls.Add(1) + if s.beforeDelete != nil { + if err := s.beforeDelete(ctx); err != nil { + return err + } + } + return s.memoryDatabaseStore.Delete(ctx, reference) +} + +func databaseLifecycleInput() DatabaseInput { + return DatabaseInput{ + Driver: "mysql", Host: "db.example.test", Database: "scert", + Username: "owner", Password: "test-password", + } +} + +func TestDatabaseCancellationDoesNotPublishSession(t *testing.T) { + for _, phase := range []string{"before", "after_put", "after_verify"} { + t.Run(phase, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + store := &lifecycleDatabaseStore{} + manager := NewDatabaseVerificationManager(config.Default(t.TempDir()), store) + calls := 0 + manager.verify = func(context.Context, config.DatabaseConfig) error { + calls++ + if phase == "after_verify" { + cancel() + } + return nil + } + if phase == "before" { + cancel() + } else if phase == "after_put" { + store.afterPut = cancel + } + result, err := manager.Verify(ctx, databaseLifecycleInput()) + if !errors.Is(err, context.Canceled) || result.VerificationToken != "" { + t.Fatalf("取消请求不得发布令牌：%v", err) + } + if phase != "after_verify" && calls != 0 { + t.Fatal("已取消请求仍执行了数据库连接验证") + } + if len(store.values) != 0 || len(manager.sessions) != 0 || manager.inFlight != 0 { + t.Fatal("取消请求遗留了凭据、会话或并发占位") + } + }) + } +} + +func TestDatabaseClaimCleanupFailureRetriesThroughPublicOperation(t *testing.T) { + store := &memoryDatabaseStore{deleteFailures: 1} + manager := NewDatabaseVerificationManager(config.Default(t.TempDir()), store) + manager.verify = func(context.Context, config.DatabaseConfig) error { return nil } + result, err := manager.Verify(t.Context(), databaseLifecycleInput()) + if err != nil { + t.Fatal(err) + } + _, cleanup, err := manager.Claim(result.VerificationToken) + if err != nil { + t.Fatal(err) + } + if err := cleanup(); err == nil || len(manager.pendingCleanup) != 1 { + t.Fatal("事务回收失败必须保留待清理引用") + } + manager.DiscardAll() + if len(store.values) != 0 || len(manager.pendingCleanup) != 0 { + t.Fatal("后续公开操作未重试事务遗留凭据") + } + if _, err := manager.Resolve(result.VerificationToken); err == nil { + t.Fatal("事务已领取的令牌不得因清理重试重新生效") + } +} + +func TestDatabaseBatchCleanupProtectsExpiredFormalCredential(t *testing.T) { + store := &memoryDatabaseStore{} + manager := NewDatabaseVerificationManager(config.Default(t.TempDir()), store) + manager.verify = func(context.Context, config.DatabaseConfig) error { return nil } + result, err := manager.Verify(t.Context(), databaseLifecycleInput()) + if err != nil { + t.Fatal(err) + } + manager.now = func() time.Time { return result.ExpiresAt } + manager.DiscardAllExcept(result.CredentialRef) + if _, exists := store.values[result.CredentialRef]; !exists { + t.Fatal("批量清理错误删除了过期令牌关联的正式凭据") + } + if len(manager.sessions) != 0 || len(manager.expiredSession) != 0 { + t.Fatal("保护正式凭据不应保留初始化令牌") + } +} + +func TestDatabaseCleaningSessionsReserveCapacity(t *testing.T) { + for _, mode := range []string{"expiry", "discard", "batch"} { + t.Run(mode, func(t *testing.T) { + started, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{}) + var startOnce, releaseOnce sync.Once + unblock := func() { releaseOnce.Do(func() { close(release) }) } + defer unblock() + store := &lifecycleDatabaseStore{beforeDelete: func(ctx context.Context) error { + startOnce.Do(func() { close(started) }) + select { + case <-release: + return nil + case <-ctx.Done(): + return ctx.Err() + } + }} + manager := NewDatabaseVerificationManager(config.Default(t.TempDir()), store) + var verifyCalls atomic.Int32 + manager.verify = func(context.Context, config.DatabaseConfig) error { verifyCalls.Add(1); return nil } + now := time.Now().UTC() + manager.now = func() time.Time { return now } + for i := range databaseVerificationCapacity { + expires := now.Add(time.Hour) + if mode == "expiry" && i == 0 { + expires = now + } + manager.sessions[fmt.Sprint(i)] = databaseVerificationSession{ + CredentialRef: fmt.Sprint("credential-", i), ExpiresAt: expires, + } + } + go func() { + defer close(done) + switch mode { + case "expiry": + _, _ = manager.Resolve("0") + case "discard": + _ = manager.Discard("0") + case "batch": + manager.DiscardAll() + } + }() + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("凭据删除未开始") + } + if _, err := manager.Verify(t.Context(), DatabaseInput{Driver: "sqlite"}); err == nil { + t.Fatal("清理中的会话未计入容量，错误接受了新验证") + } + if verifyCalls.Load() != 0 { + t.Fatal("容量已满时仍进行了数据库验证") + } + unblock() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("清理无法完成") + } + if len(manager.expiredSession) != 0 { + t.Fatal("清理成功后仍保留容量占位") + } + }) + } +} + +func TestDatabaseBatchDoesNotTakeOverConcurrentExpiryCleanup(t *testing.T) { + started, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{}) + var startOnce, releaseOnce sync.Once + unblock := func() { releaseOnce.Do(func() { close(release) }) } + defer unblock() + store := &lifecycleDatabaseStore{beforeDelete: func(ctx context.Context) error { + startOnce.Do(func() { close(started) }) + select { + case <-release: + return nil + case <-ctx.Done(): + return ctx.Err() + } + }} + manager := NewDatabaseVerificationManager(config.Default(t.TempDir()), store) + manager.verify = func(context.Context, config.DatabaseConfig) error { return nil } + result, err := manager.Verify(t.Context(), databaseLifecycleInput()) + if err != nil { + t.Fatal(err) + } + manager.now = func() time.Time { return result.ExpiresAt } + go func() { defer close(done); _, _ = manager.Resolve(result.VerificationToken) }() + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("过期清理未开始") + } + batchDone := make(chan struct{}) + go func() { defer close(batchDone); manager.DiscardAll() }() + select { + case <-batchDone: + case <-time.After(5 * time.Second): + t.Fatal("批量清理重复接管了正在删除的过期会话") + } + unblock() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("过期清理未结束") + } + if store.deleteCalls.Load() != 1 || len(manager.expiredSession) != 0 { + t.Fatal("单个过期会话被重复删除或遗留清理状态") + } +} + +func TestDatabasePendingCleanupCountsTowardCapacity(t *testing.T) { + store := &lifecycleDatabaseStore{beforeDelete: func(context.Context) error { return errors.New("测试删除失败") }} + manager := NewDatabaseVerificationManager(config.Default(t.TempDir()), store) + manager.verify = func(context.Context, config.DatabaseConfig) error { + t.Fatal("待回收凭据占满容量时不应连接数据库") + return nil + } + for i := range databaseVerificationCapacity { + manager.pendingCleanup[fmt.Sprint("pending-", i)] = struct{}{} + } + if _, err := manager.Verify(t.Context(), DatabaseInput{Driver: "sqlite"}); err == nil { + t.Fatal("未回收凭据未计入验证容量") + } + if manager.inFlight != 0 || len(manager.pendingCleanup) != databaseVerificationCapacity { + t.Fatal("拒绝验证时破坏了待清理状态") + } +}