接入完整部署互斥与持久化人工占用防护

This commit is contained in:
xuans 2026-10-03 20:18:35 +08:00
parent d801fb3833
commit b3f27e88df
12 changed files with 430 additions and 2 deletions

View File

@ -0,0 +1,85 @@
package deployment
import (
"context"
"fmt"
)
// acquireOperation 覆盖整个部署、站点核验及收尾;人工执行器必须共用此入口。
// 它不替代数据库持久化占用和远端文件锁,等待可以取消,不能提前释放。
func (s *Service) acquireOperation(ctx context.Context) (func(), error) {
if s == nil {
return nil, ErrRecoveryDispositionUnavailable
}
if err := ctx.Err(); err != nil {
return nil, err
}
s.operationOnce.Do(func() { s.operationGate = make(chan struct{}, 1) })
select {
case <-ctx.Done():
return nil, ctx.Err()
case s.operationGate <- struct{}{}:
if err := ctx.Err(); err != nil {
<-s.operationGate
return nil, err
}
return func() { <-s.operationGate }, nil
}
}
func operationFailure(targets []Target, err error) Report {
report := Report{Total: len(targets), Failed: len(targets), Status: "failed", Results: make([]Result, 0, len(targets))}
for _, target := range targets {
report.Results = append(report.Results, Result{
ServerID: target.Server.ID, Server: target.Server.Name, Status: "failed",
FailureCode: FailureUnavailable, Message: fmt.Sprintf("未获得部署操作许可:%v", err),
})
}
return report
}
type recoveryClaimReader interface {
HasRecoveryClaim(context.Context, string) (bool, error)
}
func (s *Service) checkRecoveryClaim(ctx context.Context, id string) error {
if err := ctx.Err(); err != nil {
return err
}
reader, ok := s.journal.(recoveryClaimReader)
if !ok {
return nil
}
claimed, err := reader.HasRecoveryClaim(ctx, id)
if err != nil {
return ErrRecoveryDispositionUnavailable
}
if claimed {
return ErrRecoveryDispositionBusy
}
return ctx.Err()
}
// checkDeploymentClaims 在连接远端前拦截仍有人工占用的服务器。
// 同地址同端口的别名也须检查;凭据或路径修改不能绕过原服务器占用。
func (s *Service) checkDeploymentClaims(ctx context.Context, server Server) error {
if _, ok := s.journal.(recoveryClaimReader); !ok {
return nil
}
records, err := s.journal.ListPendingDeploymentTransactions(ctx)
if err != nil {
return ErrRecoveryDispositionUnavailable
}
for _, record := range records {
sameServer := record.ServerID == server.ID
if record.Evidence != nil {
sameServer = sameServer || (record.Evidence.Address == server.Address && record.Evidence.Port == server.Port)
}
if sameServer {
if err := s.checkRecoveryClaim(ctx, record.ID); err != nil {
return err
}
}
}
return ctx.Err()
}

View File

@ -0,0 +1,243 @@
package deployment
import (
"context"
"errors"
"maps"
"reflect"
"sync"
"sync/atomic"
"testing"
"time"
)
type operationWitnessContext struct {
context.Context
once sync.Once
waiting chan struct{}
}
func (c *operationWitnessContext) Done() <-chan struct{} {
c.once.Do(func() { close(c.waiting) })
return c.Context.Done()
}
type operationFactory struct {
client Client
calls atomic.Int32
}
func (f *operationFactory) Connect(context.Context, Server) (Client, error) {
f.calls.Add(1)
return f.client, nil
}
type operationCloseClient struct {
*memoryClient
closeHook func()
}
func (c *operationCloseClient) Close() error {
if c.closeHook != nil {
c.closeHook()
}
return nil
}
type operationJournal struct {
memoryTransactionJournal
commitHook func()
}
func (j *operationJournal) UpdateDeploymentTransaction(ctx context.Context, item TransactionRecord) error {
if item.Phase == TransactionPhaseCommitting && j.commitHook != nil {
j.commitHook()
}
return j.memoryTransactionJournal.UpdateDeploymentTransaction(ctx, item)
}
func awaitOperationSignal(t *testing.T, signal <-chan struct{}) {
t.Helper()
select {
case <-signal:
case <-time.After(5 * time.Second):
t.Fatal("operation did not reach expected stage")
}
}
func TestOperationGateCoversVerifiedLifecycle(t *testing.T) {
for _, stage := range []string{"verify", "commit", "restore", "close"} {
t.Run(stage, func(t *testing.T) {
entered, resume, done := make(chan struct{}), make(chan struct{}), make(chan struct{})
var unblock sync.Once
release := func() { unblock.Do(func() { close(resume) }) }
defer release()
var hookOnce sync.Once
hook := func() { hookOnce.Do(func() { close(entered); <-resume }) }
server, memory := testServer("gate"), newMemoryClient()
seedVerifiedFiles(memory, server)
client := &operationCloseClient{memoryClient: memory}
journal := &operationJournal{}
if stage == "close" {
client.closeHook = hook
}
if stage == "commit" {
journal.commitHook = hook
}
factory := &operationFactory{client: client}
service := NewServiceWithJournal(factory, journal)
ctx := WithTransactionContext(t.Context(), TransactionContext{TaskID: "task", AttemptID: "attempt", AttemptNumber: 1})
artifact := testArtifact(t)
var report Report
var runErr error
go func() {
defer close(done)
report, runErr = service.DeployVerified(ctx, []Target{{Server: server}}, artifact, Verification{
Verify: func(context.Context) error {
if stage == "verify" {
hook()
}
if stage == "restore" {
return errors.New("injected verification failure")
}
return nil
},
VerifyRecovery: func(context.Context) error { hook(); return nil },
})
}()
awaitOperationSignal(t, entered)
calls := factory.calls.Load()
for _, operation := range []string{"deploy", "verified", "recover"} {
waitCtx, cancel := context.WithCancel(t.Context())
witness := &operationWitnessContext{Context: waitCtx, waiting: make(chan struct{})}
otherDone := make(chan struct{})
var otherErr error
var otherReport Report
go func() {
defer close(otherDone)
if operation == "recover" {
otherErr = service.RecoverPending(witness, func(context.Context, string) (Server, error) { return server, nil })
} else if operation == "verified" {
otherReport, otherErr = service.DeployVerified(witness, []Target{{Server: server}}, artifact, Verification{Verify: func(context.Context) error { return nil }})
} else {
otherReport = service.Deploy(witness, []Target{{Server: server}}, artifact)
}
}()
awaitOperationSignal(t, witness.waiting)
cancel()
awaitOperationSignal(t, otherDone)
if factory.calls.Load() != calls || (operation != "deploy" && !errors.Is(otherErr, context.Canceled)) || (operation != "recover" && otherReport.Status != "failed") {
t.Fatalf("concurrent %s entered during %s: %+v %v", operation, stage, otherReport, otherErr)
}
}
release()
awaitOperationSignal(t, done)
if stage == "restore" {
if runErr == nil || report.RecoveryStatus != "success" {
t.Fatalf("restore failed: %+v %v", report, runErr)
}
} else if runErr != nil || report.Status != "success" {
t.Fatalf("deployment failed: %+v %v", report, runErr)
}
unlock, err := service.acquireOperation(t.Context())
if err != nil {
t.Fatal(err)
}
unlock()
})
}
}
type claimedOperationJournal struct {
memoryTransactionJournal
claimed bool
claimErr error
listErr error
}
func (j *claimedOperationJournal) HasRecoveryClaim(context.Context, string) (bool, error) {
return j.claimed, j.claimErr
}
func (j *claimedOperationJournal) ListPendingDeploymentTransactions(ctx context.Context) ([]TransactionRecord, error) {
if j.listErr != nil {
return nil, j.listErr
}
return j.memoryTransactionJournal.ListPendingDeploymentTransactions(ctx)
}
func TestRecoveryClaimPreventsAutomaticMutation(t *testing.T) {
for _, failure := range []bool{false, true} {
t.Run(map[bool]string{false: "claimed", true: "read_failure"}[failure], func(t *testing.T) {
server, client, record := recoveryPreflightFixture()
original := maps.Clone(client.files)
journal := &claimedOperationJournal{memoryTransactionJournal: memoryTransactionJournal{pending: []TransactionRecord{record}}, claimed: true}
want := ErrRecoveryDispositionBusy
if failure {
journal.claimErr = errors.New("private database error")
want = ErrRecoveryDispositionUnavailable
}
factory := &operationFactory{client: client}
service := NewServiceWithJournal(factory, journal)
err := service.RecoverPending(t.Context(), func(context.Context, string) (Server, error) { t.Error("claimed record resolved"); return server, nil })
if !errors.Is(err, want) || factory.calls.Load() != 0 || len(journal.updated) != 0 || len(journal.finished) != 0 || !reflect.DeepEqual(original, client.files) {
t.Fatalf("claimed recovery changed state: %v", err)
}
})
}
}
func TestRecoveryClaimPreventsDeploymentBeforeConnection(t *testing.T) {
for _, scenario := range []string{"same_server", "alias", "claim_failure", "list_failure"} {
t.Run(scenario, func(t *testing.T) {
server, client, record := recoveryPreflightFixture()
journal := &claimedOperationJournal{memoryTransactionJournal: memoryTransactionJournal{pending: []TransactionRecord{record}}, claimed: true}
if scenario == "alias" {
server.ID = "alias"
}
if scenario == "claim_failure" {
journal.claimErr = errors.New("private claim error")
}
if scenario == "list_failure" {
journal.listErr = errors.New("private list error")
}
factory := &operationFactory{client: client}
service := NewServiceWithJournal(factory, journal)
report := service.Deploy(t.Context(), []Target{{Server: server}}, testArtifact(t))
if report.Status != "failed" || report.Failed != 1 || factory.calls.Load() != 0 || len(journal.pending) != 1 {
t.Fatalf("claimed deployment entered remote: %+v", report)
}
})
}
}
func TestUnclaimedRecoveryStillRuns(t *testing.T) {
server, client, record := recoveryPreflightFixture()
journal := &claimedOperationJournal{memoryTransactionJournal: memoryTransactionJournal{pending: []TransactionRecord{record}}}
factory := &operationFactory{client: client}
service := NewServiceWithJournal(factory, journal)
service.VerifyRecoveryTarget = func(context.Context, Server, RecoveryBaseline, bool) error { return nil }
if err := service.RecoverPending(t.Context(), func(context.Context, string) (Server, error) { return server, nil }); err != nil {
t.Fatal(err)
}
if factory.calls.Load() != 1 || len(journal.finished) != 1 || journal.finished[0].status != TransactionStatusRolledBack {
t.Fatal("unclaimed recovery was suppressed")
}
}
func TestCanceledOperationDoesNotLeakGate(t *testing.T) {
service := NewService(nil)
canceled, cancel := context.WithCancel(t.Context())
cancel()
if _, err := service.acquireOperation(canceled); !errors.Is(err, context.Canceled) {
t.Fatal(err)
}
for range 3 {
ctx, stop := context.WithTimeout(t.Context(), time.Second)
unlock, err := service.acquireOperation(ctx)
stop()
if err != nil {
t.Fatal(err)
}
unlock()
}
}

View File

@ -11,6 +11,7 @@ import (
"os"
"sort"
"strings"
"sync"
"time"
)
@ -57,6 +58,8 @@ type Report struct {
}
type Service struct {
operationOnce sync.Once
operationGate chan struct{}
factory ClientFactory
journal TransactionJournal
now func() time.Time
@ -72,6 +75,11 @@ func NewServiceWithJournal(factory ClientFactory, journal TransactionJournal) *S
}
func (s *Service) Deploy(ctx context.Context, targets []Target, artifact Artifact) Report {
release, err := s.acquireOperation(ctx)
if err != nil {
return operationFailure(targets, err)
}
defer release()
return s.deploy(ctx, targets, artifact, nil)
}
@ -150,6 +158,11 @@ func (s *Service) deployOne(ctx context.Context, server Server, domainName strin
result.Message = err.Error()
return result
}
if err := s.checkDeploymentClaims(ctx, server); err != nil {
result.FailureCode = FailureUnavailable
result.Message = err.Error()
return result
}
client, err := s.factory.Connect(ctx, server)
if err != nil {
result.FailureCode = FailureConnection
@ -752,6 +765,11 @@ func (s *Service) RecoverPending(ctx context.Context, resolve func(context.Conte
if s == nil || s.journal == nil {
return nil
}
release, err := s.acquireOperation(ctx)
if err != nil {
return err
}
defer release()
if resolve == nil {
return errors.New("部署事务恢复缺少服务器解析器")
}
@ -761,6 +779,10 @@ func (s *Service) RecoverPending(ctx context.Context, resolve func(context.Conte
}
var recoveryErrors []error
for _, record := range records {
if err := s.checkRecoveryClaim(ctx, record.ID); err != nil {
recoveryErrors = append(recoveryErrors, fmt.Errorf("事务 %s 未执行自动恢复:%w", record.ID, err))
continue
}
if err := s.recoverTransaction(ctx, &record, resolve); err != nil {
recoveryErrors = append(recoveryErrors, fmt.Errorf("事务 %s 恢复失败:%w", record.ID, err))
record.Status = TransactionStatusRecoveryRequired

View File

@ -31,6 +31,11 @@ func (s *Service) DeployVerified(ctx context.Context, targets []Target, artifact
if verification.Verify == nil {
return Report{Status: "failed"}, errors.New("缺少部署后站点校验,未执行推送")
}
release, err := s.acquireOperation(ctx)
if err != nil {
return operationFailure(targets, err), err
}
defer release()
pending := make([]*pendingDeployment, 0, len(targets))
// 异常退出时关闭连接但保留现场;不得以释放锁掩盖未确认的事务。
defer func() {

View File

@ -12,6 +12,7 @@ import (
type lateCancellationContext struct {
context.Context
armed atomic.Bool
errCalls atomic.Int32
}
@ -20,7 +21,7 @@ func (c *lateCancellationContext) Done() <-chan struct{} {
}
func (c *lateCancellationContext) Err() error {
if c.errCalls.Add(1) > 2 {
if c.armed.Load() && c.errCalls.Add(1) > 1 {
return context.Canceled
}
return nil
@ -49,7 +50,7 @@ func TestVerifiedDeploymentCommitUsesIndependentContextAfterCancellationRace(t *
artifact := testArtifact(t)
service := NewServiceWithJournal(fixedClientFactory{client}, journal)
report, err := service.DeployVerified(ctx, []Target{{Server: server}}, artifact, Verification{
Verify: func(context.Context) error { return nil },
Verify: func(context.Context) error { parent.armed.Store(true); return nil },
})
if err != nil || report.Status != "success" || !journal.commitContextOK {
t.Fatalf("unexpected deployment result: %+v, %v", report, err)

View File

@ -8,6 +8,7 @@ import (
"encoding/json"
"errors"
"fmt"
"reflect"
"strings"
"sync"
"sync/atomic"
@ -244,6 +245,22 @@ func TestRecoveryDispositionClaimSurvivesDatabaseReopen(t *testing.T) {
}
defer reopened.Close()
repo = NewSQLDeploymentTransactionRepository(reopened)
before, err := repo.ListPendingDeploymentTransactions(t.Context())
if err != nil {
t.Fatal(err)
}
service := deployment.NewServiceWithJournal(nil, repo)
err = service.RecoverPending(t.Context(), func(context.Context, string) (deployment.Server, error) {
t.Error("restart attempted automatic recovery of claimed transaction")
return deployment.Server{}, errors.New("unexpected server lookup")
})
if !errors.Is(err, deployment.ErrRecoveryDispositionBusy) {
t.Fatalf("restart did not honor durable claim: %v", err)
}
after, err := repo.ListPendingDeploymentTransactions(t.Context())
if err != nil || !reflect.DeepEqual(before, after) {
t.Fatal("restart modified claimed original transaction", err)
}
if claimed, err := repo.HasRecoveryClaim(t.Context(), intent.TransactionID); err != nil || !claimed {
t.Fatal("incomplete operation lost its persistent claim", err)
}

View File

@ -1,5 +1,14 @@
# Linux 生产流程测试记录
## 2026年10月3日20:17 完整操作互斥回归
- [x] `scert-operation-20261003-2011.service` 成功结束;`operation-race-vet.log` 中部署包竞态约 3.189 秒,`OPERATION_RACE_EXIT=0`、`OPERATION_VET_EXIT=0`,整组约 43.506 秒。
- [x] 补充竞态:部署约 1.086 秒、仓储约 2.016 秒,`OPERATION_EXTRA_RACE_EXIT=0`;实际覆盖 SQLite 关库重开后服务启动识别占用及原事务不变。
- [~] `scert-operation-full-20261003-2012.service` 正执行最新后端全量;日志为第一台隔离目录中的 `operation-full.log`。未修改正在运行任务的源码。
- [x] `service.go`、`operation.go`、`operation_test.go`、`recovery_disposition_repository_test.go` 的本地与远端 SHA-256 一致;测试目录 Git 索引仅用于源码边界测试,不是正式部署。
- [ ] 本轮未替换线上正式包,不代表公共入口、首次 Web 流程或生产验收通过。
## 2026年10月3日20:04 MySQL 修复复测与磁盘复核
- [x] 原任务 `scert-disposition-extra-20261003-1945.service` 已失败终止:补充竞态通过,MySQL 迁移十二场景通过;仓储三个审计注入因错误 1419 失败。原日志保留,不记整组通过。

View File

@ -1,5 +1,13 @@
# Scert 开发任务清单
## 2026年10月3日20:17 当前开发进度
- [x] 开发完整部署操作互斥及持久化占用检查;部署竞态、完整静态检查和补充并发/取消/数据库重开专项通过。
- [~] 原生 Linux 最新后端全量仍运行中,不把专项替代全量或实机验收。
- [ ] 下一开发任务:一次性本机确认、三类动作执行器、受保护本机协议及终端接入;复用已建立的持久化审计与共同互斥。
- [ ] PostgreSQL 实库及正式候选包生产门禁仍待完成;Linux 范围不变,未改线上业务数据库。
## 2026年10月3日20:04 开发与回归状态
- [x] MySQL 审计故障注入改用隔离实例管理连接,受限业务连接仍执行全部仓储操作;六场景复测通过。

View File

@ -1,5 +1,13 @@
# Scert 整改任务执行清单
## 2026年10月3日20:17 当前执行状态
- [x] 完整部署互斥和人工占用防护已编码;部署竞态、完整静态检查、补充竞态均通过。
- [x] SQLite 实际关库重开后由真实部署服务识别遗留占用,不连接远端、不改原事务;无占用仍正常恢复。
- [~] 最新后端全量运行中。
- [ ] 一次性本机确认、动作执行器及本机交互、PostgreSQL 实库、正式包与生产门禁仍未完成。
## 2026年10月3日20:04 当前执行状态
- [x] 隔离 MySQL 迁移十二场景通过,仓储三动作及三个审计失败回滚场景修复后通过。

View File

@ -1,5 +1,18 @@
# Scert 整改、合并与优化清单
## 2026年10月3日20:17 最新执行进度
本节覆盖下方历史状态;仅 Linux,第一台部署测试、第二台证书推送,Smail 为 `web.cdis.cn`。总体生产验收未通过。
- [x] R4 完整进程内互斥已接入 `Deploy`、`DeployVerified`、`RecoverPending`,覆盖推送、HTTPS、提交、回滚与连接收尾;等待可取消,不提前释放。
- [x] 自动恢复识别持久化人工占用,查询失败或已有占用均不连接远端、不更新原事务;普通部署在连接前检查同服务器及同地址端口别名。
- [x] 原生 Linux 部署包竞态和后端完整静态检查通过;补充并发、取消、未占用正常恢复、SQLite 关闭重开后的真实服务占用拦截竞态通过。
- [~] 最新后端全量正在 `scert-operation-full-20261003-2012.service` 执行,尚未据此记通过;四个关键源文件本地与测试机 SHA-256 一致。
- [x] 前一批 MySQL 迁移十二场景及仓储六场景通过;代码与文档检查点 `d801fb3` 已推送。
- [ ] R3 下一步:一次性确认绑定受保护本机身份和准备摘要,在共同互斥及持久化占用内重新核验,执行明确的记录、提交或回滚动作;再接入本机协议和终端。
- [ ] R4 PostgreSQL 实库、执行器并发/重启边界,以及 R5 正式候选包和三数据库首次 Web、系统邮件、DNS/ACME、公共 HTTPS、多对多部署、失败回滚等生产门禁仍待完成。
## 2026年10月3日20:04 最新执行进度
本节覆盖下方历史状态;仍仅处理 Linux,生产验收未通过。

View File

@ -1,5 +1,14 @@
# 部署事务人工处置设计
## 2026年10月3日20:17 互斥接入状态
- [x] `Service` 的共同操作门覆盖普通部署、站点核验、异常回滚、提交收尾及自动恢复;等待可取消,异常退出先收尾再释放。
- [x] 自动恢复在解析服务器之前查询持久化占用;失败关闭且不改原事务。普通部署在连接之前检查同服务器及同地址端口别名占用。
- [x] 竞态、静态检查及数据库重开集成专项通过,最新全量运行中。
- [ ] 人工执行器必须加入同一操作门,消费一次性确认后建立持久化意图、重新核验准备绑定,并按动作执行真实文件/锁/严格 HTTPS 核验;失败不静默切换动作。
- [ ] 本机协议、终端、动作级不确定结果与重启矩阵、PostgreSQL 实库及正式生产门禁仍未闭环。
## 2026年10月3日20:04 实现与待接入边界
- [x] 持久化 MySQL 迁移十二场景、仓储三动作及审计失败回滚六场景通过。

View File

@ -1,5 +1,13 @@
# 部署事务人工核查任务清单
## 2026年10月3日20:17 最新任务状态
- [x] R4-A 部署及自动恢复进程内互斥、持久化占用识别已接入;涵盖站点核验、提交、恢复核验和连接收尾,等待可取消。
- [x] 部署竞态、完整静态检查及数据库关闭重开后真实服务拒绝自动重放通过。
- [~] 最新后端全量仍运行中;R4-A 整体还需人工执行器接入同一互斥,不能标整项复审通过。
- [ ] R3 一次性本机确认、记录/提交/回滚三动作及本机协议与终端,R4 PostgreSQL 实库,R5 正式包验收。
## 2026年10月3日20:04 最新任务状态
- [x] 持久化补充竞态、MySQL 迁移十二场景和仓储六场景通过;仓储审计故障注入权限问题已在测试夹具层修复。