修复部署启动时空锁恢复
This commit is contained in:
parent
bf9d8e63fd
commit
000991fc65
@ -651,7 +651,7 @@ func (s *Service) recoverTransaction(ctx context.Context, record *TransactionRec
|
||||
return err
|
||||
}
|
||||
if record.Phase == TransactionPhaseAcquiring || record.Phase == TransactionPhaseRollbackVerified || record.Phase == TransactionPhaseCommitVerified {
|
||||
return releaseTransactionLocks(recoveryContext, client, *record, true)
|
||||
return releaseTransactionLocks(recoveryContext, client, *record, true, record.Phase == TransactionPhaseAcquiring)
|
||||
}
|
||||
if record.Phase == TransactionPhaseCommitting {
|
||||
return s.recoverCommittedTransaction(recoveryContext, record, server, client)
|
||||
@ -679,7 +679,7 @@ func (s *Service) recoverTransaction(ctx context.Context, record *TransactionRec
|
||||
if err := s.persistRecoveryPhase(recoveryContext, record, TransactionPhaseRollbackVerified); err != nil {
|
||||
return err
|
||||
}
|
||||
return releaseTransactionLocks(recoveryContext, client, *record, true)
|
||||
return releaseTransactionLocks(recoveryContext, client, *record, true, false)
|
||||
}
|
||||
|
||||
func (s *Service) recoverCommittedTransaction(ctx context.Context, record *TransactionRecord, server Server, client Client) error {
|
||||
@ -719,7 +719,7 @@ func (s *Service) recoverCommittedTransaction(ctx context.Context, record *Trans
|
||||
if err := s.persistRecoveryPhase(ctx, record, TransactionPhaseCommitVerified); err != nil {
|
||||
return err
|
||||
}
|
||||
return releaseTransactionLocks(ctx, client, *record, true)
|
||||
return releaseTransactionLocks(ctx, client, *record, true, false)
|
||||
}
|
||||
|
||||
func restoreTransactionFiles(ctx context.Context, client Client, record TransactionRecord) error {
|
||||
@ -771,13 +771,32 @@ func restoreTransactionFiles(ctx context.Context, client Client, record Transact
|
||||
return nil
|
||||
}
|
||||
|
||||
func releaseTransactionLocks(ctx context.Context, client Client, record TransactionRecord, verifyOwner bool) error {
|
||||
func releaseTransactionLocks(ctx context.Context, client Client, record TransactionRecord, verifyOwner, allowMissingOwner bool) error {
|
||||
for index := len(record.LockPaths) - 1; index >= 0; index-- {
|
||||
lockPath := record.LockPaths[index]
|
||||
if verifyOwner {
|
||||
if err := checkOwnedLock(ctx, client, lockPath, record.Evidence.LockToken, false); err != nil {
|
||||
if err := checkOwnedLock(ctx, client, lockPath, record.Evidence.LockToken, allowMissingOwner); err != nil {
|
||||
return fmt.Errorf("无法证明事务仍拥有部署锁 %s:%w", lockPath, err)
|
||||
}
|
||||
if allowMissingOwner {
|
||||
reader, ok := client.(EvidenceClient)
|
||||
if !ok {
|
||||
return fmt.Errorf("无法读取部署锁归属:%s", lockPath)
|
||||
}
|
||||
owner, err := reader.ReadLockOwner(ctx, lockPath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("读取部署锁归属失败:%w", err)
|
||||
}
|
||||
if owner == "" {
|
||||
exists, err := client.Exists(ctx, lockPath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("检查部署锁目录失败:%w", err)
|
||||
}
|
||||
if !exists {
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := client.Unlock(ctx, lockPath); err != nil {
|
||||
return fmt.Errorf("释放部署锁失败:%w", err)
|
||||
|
||||
@ -171,3 +171,41 @@ func TestDeployPartialLockConflictReleasesOwnedLocks(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecoverPendingAcquiringCleansLocksWithoutOwner(t *testing.T) {
|
||||
server := testServer("acquiring-missing-owner").Normalize()
|
||||
client := newMemoryClient()
|
||||
id := "acquiring-missing-owner-transaction"
|
||||
paths := []string{server.CertificatePath, server.CertificateChainPath, server.PrivateKeyPath}
|
||||
locks := make([]string, 0, len(paths))
|
||||
files := make(map[string]string, len(paths))
|
||||
for _, path := range paths {
|
||||
lockPath := path + ".scert-lock"
|
||||
locks = append(locks, lockPath)
|
||||
client.files[lockPath] = memoryFile{}
|
||||
files[path] = contentDigest([]byte("old"))
|
||||
}
|
||||
journal := &memoryTransactionJournal{pending: []TransactionRecord{{
|
||||
ID: id, TaskID: "task", AttemptID: "attempt", AttemptNumber: 1, ResultIndex: 0,
|
||||
ServerID: server.ID, ServerName: server.Name, Paths: paths, LockPaths: locks,
|
||||
Phase: TransactionPhaseAcquiring, Status: TransactionStatusRecoveryRequired,
|
||||
Evidence: &RecoveryEvidence{
|
||||
Version: 1, LockToken: contentDigest([]byte("scert-lock:" + id)),
|
||||
Address: server.Address, Port: server.Port, HostKeyFingerprint: server.HostKeyFingerprint,
|
||||
ValidationCommand: append([]string(nil), server.ValidationCommand...),
|
||||
ReloadCommand: append([]string(nil), server.ReloadCommand...), Files: files,
|
||||
},
|
||||
}}}
|
||||
service := NewServiceWithJournal(memoryFactory{clients: map[string]*memoryClient{server.ID: client}}, journal)
|
||||
if err := service.RecoverPending(t.Context(), func(context.Context, string) (Server, error) { return server, nil }); err != nil {
|
||||
t.Fatalf("无归属的中断获取阶段恢复失败:%v", err)
|
||||
}
|
||||
for _, lockPath := range locks {
|
||||
if _, ok := client.files[lockPath]; ok {
|
||||
t.Fatalf("无归属锁目录未清理:%s", lockPath)
|
||||
}
|
||||
}
|
||||
if len(journal.finished) != 1 || journal.finished[0].status != TransactionStatusRolledBack {
|
||||
t.Fatalf("无归属锁恢复未完成回滚:%+v", journal.finished)
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user