补充数据库生命周期专项回归
增加请求取消、Claim 清理重试、批量清理容量和并发竞争等数据库验证生命周期测试，作为 P0-10 待复审证据。
This commit is contained in:
parent
b4b761f1ed
commit
1620a83461
@ -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)
|
||||
|
||||
249
backend/internal/bootstrap/database_lifecycle_test.go
Normal file
249
backend/internal/bootstrap/database_lifecycle_test.go
Normal file
@ -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("拒绝验证时破坏了待清理状态")
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user