244 lines
8.1 KiB
Go
244 lines
8.1 KiB
Go
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()
|
|
}
|
|
}
|