test(auth): bound restore lifecycle gates

tools/tht/internal/backup/restore_test.go

@@ -317,42 +317,160 @@ type restoreLifecycleTestOutcome struct {
 	err    error
 }
 
+const restoreLifecycleTestWait = 2 * time.Second
+
+func stopRestoreLifecycleTimer(timer *time.Timer) {
+	if !timer.Stop() {
+		select {
+		case <-timer.C:
+		default:
+		}
+	}
+}
+
+func waitRestoreLifecycleSignal(ctx context.Context, signal <-chan struct{}, description string) error {
+	timer := time.NewTimer(restoreLifecycleTestWait)
+	defer stopRestoreLifecycleTimer(timer)
+	select {
+	case <-signal:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return fmt.Errorf("timed out waiting for %s", description)
+	}
+}
+
+func sendRestoreLifecycleSignal(ctx context.Context, signal chan<- struct{}, description string) error {
+	timer := time.NewTimer(restoreLifecycleTestWait)
+	defer stopRestoreLifecycleTimer(timer)
+	select {
+	case signal <- struct{}{}:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return fmt.Errorf("timed out releasing %s", description)
+	}
+}
+
+func waitRestoreLifecycleOutcome(ctx context.Context, done <-chan restoreLifecycleTestOutcome) (restoreLifecycleTestOutcome, bool, error) {
+	timer := time.NewTimer(restoreLifecycleTestWait)
+	defer stopRestoreLifecycleTimer(timer)
+	select {
+	case outcome := <-done:
+		return outcome, true, nil
+	case <-ctx.Done():
+		select {
+		case outcome := <-done:
+			return outcome, true, nil
+		default:
+		}
+		return restoreLifecycleTestOutcome{}, false, ctx.Err()
+	case <-timer.C:
+		return restoreLifecycleTestOutcome{}, false, errors.New("timed out waiting for restore worker outcome")
+	}
+}
+
+func waitRestoreLifecycleStage(ctx context.Context, stages <-chan string, done <-chan restoreLifecycleTestOutcome) (string, *restoreLifecycleTestOutcome, error) {
+	timer := time.NewTimer(restoreLifecycleTestWait)
+	defer stopRestoreLifecycleTimer(timer)
+	select {
+	case stage := <-stages:
+		return stage, nil, nil
+	case outcome := <-done:
+		return "", &outcome, nil
+	case <-ctx.Done():
+		select {
+		case outcome := <-done:
+			return "", &outcome, nil
+		default:
+		}
+		return "", nil, ctx.Err()
+	case <-timer.C:
+		return "", nil, errors.New("timed out waiting for lifecycle stage")
+	}
+}
+
+func waitRestoreError(ctx context.Context, done <-chan error) (error, bool, error) {
+	timer := time.NewTimer(restoreLifecycleTestWait)
+	defer stopRestoreLifecycleTimer(timer)
+	select {
+	case err := <-done:
+		return err, true, nil
+	case <-ctx.Done():
+		select {
+		case err := <-done:
+			return err, true, nil
+		default:
+		}
+		return nil, false, ctx.Err()
+	case <-timer.C:
+		return nil, false, errors.New("timed out waiting for restore worker error")
+	}
+}
+
+func waitRestoreAdmissionAttempt(ctx context.Context, attempt <-chan bool) (bool, error) {
+	timer := time.NewTimer(restoreLifecycleTestWait)
+	defer stopRestoreLifecycleTimer(timer)
+	select {
+	case allowed := <-attempt:
+		return allowed, nil
+	case <-ctx.Done():
+		return false, ctx.Err()
+	case <-timer.C:
+		return false, errors.New("timed out waiting for restore admission attempt")
+	}
+}
+
+func releaseRestoreLifecycleGate(release chan<- struct{}) {
+	select {
+	case release <- struct{}{}:
+	default:
+	}
+}
+
+func joinRestoreLifecycleWorker(cancel context.CancelFunc, cancelGate context.CancelFunc, release chan<- struct{}, done <-chan restoreLifecycleTestOutcome) (restoreLifecycleTestOutcome, bool, error) {
+	cancel()
+	if cancelGate != nil {
+		cancelGate()
+	}
+	releaseRestoreLifecycleGate(release)
+	joinContext, stop := context.WithTimeout(context.Background(), restoreLifecycleTestWait)
+	defer stop()
+	return waitRestoreLifecycleOutcome(joinContext, done)
+}
+
+func joinRestoreErrorWorker(cancel context.CancelFunc, done <-chan error) (error, bool, error) {
+	cancel()
+	joinContext, stop := context.WithTimeout(context.Background(), restoreLifecycleTestWait)
+	defer stop()
+	return waitRestoreError(joinContext, done)
+}
+
 func releaseLifecycleStage(ctx context.Context, release chan<- struct{}, done <-chan restoreLifecycleTestOutcome) (restoreLifecycleTestOutcome, bool, error) {
 	select {
 	case outcome := <-done:
 		return outcome, true, nil
 	default:
 	}
-	select {
-	case release <- struct{}{}:
-		return restoreLifecycleTestOutcome{}, false, nil
-	default:
-	}
 
-	timer := time.NewTimer(2 * time.Second)
+	timer := time.NewTimer(restoreLifecycleTestWait)
-	defer timer.Stop()
+	defer stopRestoreLifecycleTimer(timer)
-	var ctxErr error
 	select {
 	case outcome := <-done:
 		return outcome, true, nil
 	case release <- struct{}{}:
 		return restoreLifecycleTestOutcome{}, false, nil
 	case <-ctx.Done():
-		ctxErr = ctx.Err()
-		ctx = nil
-	case <-timer.C:
-		if ctxErr != nil {
-			return restoreLifecycleTestOutcome{}, false, ctxErr
-		}
-		return restoreLifecycleTestOutcome{}, false, errors.New("timed out releasing lifecycle stage")
-	}
 		select {
 		case outcome := <-done:
 			return outcome, true, nil
-	case release <- struct{}{}:
+		default:
-		return restoreLifecycleTestOutcome{}, false, nil
+		}
+		return restoreLifecycleTestOutcome{}, false, ctx.Err()
 	case <-timer.C:
-		return restoreLifecycleTestOutcome{}, false, ctxErr
+		return restoreLifecycleTestOutcome{}, false, errors.New("timed out releasing lifecycle stage")
 	}
 }
 
@@ -370,13 +488,112 @@ func TestReleaseLifecycleStageReturnsPrematureWorkerOutcome(t *testing.T) {
 	}
 }
 
+func TestReleaseLifecycleStageHonorsCancellation(t *testing.T) {
+	caller, cancel := context.WithCancel(context.Background())
+	cancel()
+	started := time.Now()
+	got, terminal, err := releaseLifecycleStage(caller, make(chan struct{}), make(chan restoreLifecycleTestOutcome))
+	if !errors.Is(err, context.Canceled) || terminal {
+		t.Fatalf("releaseLifecycleStage() = outcome %#v, terminal %t, err %v; want prompt cancellation", got, terminal, err)
+	}
+	if elapsed := time.Since(started); elapsed >= time.Second {
+		t.Fatalf("releaseLifecycleStage() cancellation took %s; want bounded prompt return", elapsed)
+	}
+}
+
+func TestRestoreLifecycleCancellationJoinsWithWithheldGate(t *testing.T) {
+	fixture := newBackupFixture(t, "local")
+	installation := fixture.installation
+	archive := filepath.Join(t.TempDir(), "restore.zip")
+	writePreflightArchive(t, archive, preflightArchiveSpec{
+		installationID: fixture.installationID,
+		entries:        []preflightArchiveEntry{{path: "configuration/operator.env", body: []byte("safe")}},
+	})
+	runner := &lifecycleGateRunner{fakeBackupRunner: newBackupRunner(installation, true)}
+	stages := make(chan string)
+	continueStage := make(chan struct{})
+	caller, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	gate := func(stage string) error {
+		select {
+		case stages <- stage:
+		case <-caller.Done():
+			return caller.Err()
+		}
+		select {
+		case <-continueStage:
+			return nil
+		case <-caller.Done():
+			return caller.Err()
+		}
+	}
+	runner.beforeFinalMaintenanceRelease = func() { _ = gate("final-barrier-release") }
+	deps := restoreTestDependencies(t, runner)
+	deps.acquireTransaction = lifecycle.AcquireTransaction
+	deps.checkpoint = func(context.Context, *lifecycle.Transaction, config.Installation, CreateRequest) (Result, error) {
+		if err := gate("checkpoint"); err != nil {
+			return Result{}, err
+		}
+		return Result{Path: filepath.Join(t.TempDir(), "checkpoint.zip")}, nil
+	}
+
+	done := make(chan restoreLifecycleTestOutcome, 1)
+	go func() {
+		result, err := restoreWithDependencies(caller, installation, RestoreRequest{Archive: archive, Confirm: true}, deps)
+		done <- restoreLifecycleTestOutcome{result: result, err: err}
+	}()
+
+	var failures []error
+	var outcome restoreLifecycleTestOutcome
+	workerJoined := false
+	stage, terminal, waitErr := waitRestoreLifecycleStage(caller, stages, done)
+	if terminal != nil {
+		outcome = *terminal
+		workerJoined = true
+		failures = append(failures, fmt.Errorf("restore ended before withheld checkpoint release: %w", outcome.err))
+	} else if waitErr != nil {
+		failures = append(failures, fmt.Errorf("withheld checkpoint gate: %w", waitErr))
+	} else {
+		if stage != "checkpoint" {
+			failures = append(failures, fmt.Errorf("lifecycle stage = %q, want checkpoint", stage))
+		}
+		diagnosticContext, stopDiagnostic := context.WithTimeout(caller, 100*time.Millisecond)
+		_, diagnosticJoined, diagnosticErr := waitRestoreLifecycleOutcome(diagnosticContext, done)
+		stopDiagnostic()
+		if diagnosticJoined {
+			failures = append(failures, errors.New("restore bypassed the intentionally withheld checkpoint release"))
+		} else if !errors.Is(diagnosticErr, context.DeadlineExceeded) {
+			failures = append(failures, fmt.Errorf("withheld checkpoint diagnostic = %w, want deadline", diagnosticErr))
+		}
+	}
+
+	if !workerJoined {
+		outcome, workerJoined, waitErr = joinRestoreLifecycleWorker(cancel, cancel, continueStage, done)
+		if waitErr != nil {
+			failures = append(failures, fmt.Errorf("cancelled restore worker join: %w", waitErr))
+		}
+	}
+
+	if err := lifecycleLockFreeAfterTerminalRestore(installation); err != nil {
+		failures = append(failures, err)
+	}
+	if !workerJoined {
+		failures = append(failures, errors.New("restore worker outcome was never observed"))
+	} else if !errors.Is(outcome.err, context.Canceled) {
+		failures = append(failures, fmt.Errorf("restore cancellation outcome = %v, want context.Canceled", outcome.err))
+	}
+	if len(failures) != 0 {
+		t.Fatal(errors.Join(failures...))
+	}
+}
+
 func TestRestoreLifecycleLockExcludesCompetingTransactionsUntilTerminalCleanup(t *testing.T) {
 	targetFailure := errors.New("target restore failed")
 	recoveryFailure := errors.New("recovery restore failed")
 	for _, scenario := range []struct {
 		name            string
 		stages          []string
-		configure       func(*restoreDependencies, *lifecycleGateRunner, func(string), context.CancelFunc)
+		configure       func(*restoreDependencies, *lifecycleGateRunner, func(string) error, context.CancelFunc)
 		wantErrors      []error
 		wantVerified    bool
 		wantMaintenance bool
@@ -386,14 +603,12 @@ func TestRestoreLifecycleLockExcludesCompetingTransactionsUntilTerminalCleanup(t
 			stages:          []string{"checkpoint", "target", "verification", "checkpoint-cleanup", "final-barrier-release"},
 			wantVerified:    true,
 			wantMaintenance: false,
-			configure: func(deps *restoreDependencies, _ *lifecycleGateRunner, gate func(string), _ context.CancelFunc) {
+			configure: func(deps *restoreDependencies, _ *lifecycleGateRunner, gate func(string) error, _ context.CancelFunc) {
 				deps.restoreFile = func(context.Context, config.Installation, ArchiveEntryMetadata, io.Reader) error {
-					gate("target")
+					return gate("target")
-					return nil
 				}
 				deps.verify["health"] = func(context.Context, config.Installation, archiveRunner) error {
-					gate("verification")
+					return gate("verification")
-					return nil
 				}
 			},
 		},
@@ -402,13 +617,17 @@ func TestRestoreLifecycleLockExcludesCompetingTransactionsUntilTerminalCleanup(t
 			stages:          []string{"checkpoint", "target-failure", "recovery", "checkpoint-cleanup", "final-barrier-release"},
 			wantErrors:      []error{targetFailure},
 			wantMaintenance: false,
-			configure: func(deps *restoreDependencies, runner *lifecycleGateRunner, gate func(string), _ context.CancelFunc) {
+			configure: func(deps *restoreDependencies, runner *lifecycleGateRunner, gate func(string) error, _ context.CancelFunc) {
 				deps.restoreFile = func(context.Context, config.Installation, ArchiveEntryMetadata, io.Reader) error {
-					gate("target-failure")
+					if err := gate("target-failure"); err != nil {
+						return err
+					}
 					return targetFailure
 				}
 				deps.recover = func(context.Context, config.Installation, PreflightResult, *stagedArchive, bool) error {
-					gate("recovery")
+					if err := gate("recovery"); err != nil {
+						return err
+					}
 					runner.running, runner.coreRunning = true, true
 					return nil
 				}
@@ -419,13 +638,17 @@ func TestRestoreLifecycleLockExcludesCompetingTransactionsUntilTerminalCleanup(t
 			stages:          []string{"checkpoint", "target-failure", "recovery-failure", "checkpoint-cleanup"},
 			wantErrors:      []error{targetFailure, recoveryFailure},
 			wantMaintenance: true,
-			configure: func(deps *restoreDependencies, _ *lifecycleGateRunner, gate func(string), _ context.CancelFunc) {
+			configure: func(deps *restoreDependencies, _ *lifecycleGateRunner, gate func(string) error, _ context.CancelFunc) {
 				deps.restoreFile = func(context.Context, config.Installation, ArchiveEntryMetadata, io.Reader) error {
-					gate("target-failure")
+					if err := gate("target-failure"); err != nil {
+						return err
+					}
 					return targetFailure
 				}
 				deps.recover = func(context.Context, config.Installation, PreflightResult, *stagedArchive, bool) error {
-					gate("recovery-failure")
+					if err := gate("recovery-failure"); err != nil {
+						return err
+					}
 					return recoveryFailure
 				}
 			},
@@ -435,14 +658,18 @@ func TestRestoreLifecycleLockExcludesCompetingTransactionsUntilTerminalCleanup(t
 			stages:          []string{"checkpoint", "target-cancel", "recovery", "checkpoint-cleanup", "final-barrier-release"},
 			wantErrors:      []error{context.Canceled},
 			wantMaintenance: false,
-			configure: func(deps *restoreDependencies, runner *lifecycleGateRunner, gate func(string), cancel context.CancelFunc) {
+			configure: func(deps *restoreDependencies, runner *lifecycleGateRunner, gate func(string) error, cancel context.CancelFunc) {
 				deps.restoreFile = func(context.Context, config.Installation, ArchiveEntryMetadata, io.Reader) error {
-					gate("target-cancel")
+					if err := gate("target-cancel"); err != nil {
+						return err
+					}
 					cancel()
 					return context.Canceled
 				}
 				deps.recover = func(context.Context, config.Installation, PreflightResult, *stagedArchive, bool) error {
-					gate("recovery")
+					if err := gate("recovery"); err != nil {
+						return err
+					}
 					runner.running, runner.coreRunning = true, true
 					return nil
 				}
@@ -467,24 +694,36 @@ func TestRestoreLifecycleLockExcludesCompetingTransactionsUntilTerminalCleanup(t
 			stages := make(chan string, 1)
 			continueStage := make(chan struct{}, 1)
 			caller, cancel := context.WithCancel(context.Background())
+			gateContext, cancelGate := context.WithCancel(context.Background())
 			defer cancel()
-			gate := func(stage string) {
+			defer cancelGate()
-				stages <- stage
+			gate := func(stage string) error {
-				<-continueStage
+				select {
+				case stages <- stage:
+				case <-gateContext.Done():
+					return gateContext.Err()
 				}
-			runner.beforeFinalMaintenanceRelease = func() { gate("final-barrier-release") }
+				select {
+				case <-continueStage:
+					return nil
+				case <-gateContext.Done():
+					return gateContext.Err()
+				}
+			}
+			runner.beforeFinalMaintenanceRelease = func() { _ = gate("final-barrier-release") }
 			deps := restoreTestDependencies(t, runner)
 			deps.prepareRecovery = func(ctx context.Context, target config.Installation, _ string) (PreflightResult, error) {
 				return Preflight(ctx, target, PreflightRequest{Archive: recoveryArchive, Confirm: true, AllowExternalSecrets: true}, permissivePreflightDependencies())
 			}
 			deps.acquireTransaction = lifecycle.AcquireTransaction
 			deps.checkpoint = func(context.Context, *lifecycle.Transaction, config.Installation, CreateRequest) (Result, error) {
-				gate("checkpoint")
+				if err := gate("checkpoint"); err != nil {
+					return Result{}, err
+				}
 				return Result{Path: filepath.Join(t.TempDir(), "checkpoint.zip")}, nil
 			}
 			deps.cleanupCheckpoint = func(string) error {
-				gate("checkpoint-cleanup")
+				return gate("checkpoint-cleanup")
-				return nil
 			}
 			scenario.configure(&deps, runner, gate, cancel)
 
@@ -496,43 +735,38 @@ func TestRestoreLifecycleLockExcludesCompetingTransactionsUntilTerminalCleanup(t
 
 			var failures []error
 			var terminal *restoreLifecycleTestOutcome
+			workerJoined := false
+			aborted := false
 			for stageIndex, wantStage := range scenario.stages {
-				receivedStage := false
+				stage, outcome, waitErr := waitRestoreLifecycleStage(gateContext, stages, done)
-				select {
+				if outcome != nil {
-				case stage := <-stages:
+					terminal = outcome
-					receivedStage = true
+					workerJoined = true
+					failures = append(failures, fmt.Errorf("restore ended before lifecycle stage %q: %w", wantStage, outcome.err))
+					break
+				}
+				if waitErr != nil {
+					failures = append(failures, fmt.Errorf("lifecycle stage %q: %w", wantStage, waitErr))
+					aborted = true
+					break
+				}
 				if stage != wantStage {
 					failures = append(failures, fmt.Errorf("lifecycle stage = %q, want %q", stage, wantStage))
 				}
-				case outcome := <-done:
-					terminal = &outcome
-					failures = append(failures, fmt.Errorf("restore ended before lifecycle stage %q: %w", wantStage, outcome.err))
-				case <-time.After(2 * time.Second):
-					select {
-					case outcome := <-done:
-						terminal = &outcome
-						failures = append(failures, fmt.Errorf("restore ended before lifecycle stage %q: %w", wantStage, outcome.err))
-					default:
-						failures = append(failures, fmt.Errorf("timed out waiting for lifecycle stage %q", wantStage))
-						cancel()
-					}
-				}
-				if terminal != nil || !receivedStage {
-					break
-				}
 				if err := competingRestoreAndBackupEntry(installation, archive, t); err != nil {
 					failures = append(failures, fmt.Errorf("%s: %w", wantStage, err))
 				}
-				outcome, workerDone, err := releaseLifecycleStage(caller, continueStage, done)
+				releasedOutcome, workerDone, err := releaseLifecycleStage(gateContext, continueStage, done)
 				if err != nil {
 					failures = append(failures, fmt.Errorf("release lifecycle stage %q: %w", wantStage, err))
-					cancel()
+					aborted = true
 					break
 				}
 				if workerDone {
-					terminal = &outcome
+					terminal = &releasedOutcome
+					workerJoined = true
 					if stageIndex+1 < len(scenario.stages) {
-						failures = append(failures, fmt.Errorf("restore ended before lifecycle stage %q: %w", scenario.stages[stageIndex+1], outcome.err))
+						failures = append(failures, fmt.Errorf("restore ended before lifecycle stage %q: %w", scenario.stages[stageIndex+1], releasedOutcome.err))
 					}
 					break
 				}
@@ -540,12 +774,21 @@ func TestRestoreLifecycleLockExcludesCompetingTransactionsUntilTerminalCleanup(t
 			var result restoreLifecycleTestOutcome
 			if terminal != nil {
 				result = *terminal
+			} else if aborted {
+				var joinErr error
+				result, workerJoined, joinErr = joinRestoreLifecycleWorker(cancel, cancelGate, continueStage, done)
+				if joinErr != nil {
+					failures = append(failures, joinErr)
+				}
 			} else {
-				select {
+				var waitErr error
-				case result = <-done:
+				result, workerJoined, waitErr = waitRestoreLifecycleOutcome(context.Background(), done)
-				case <-time.After(2 * time.Second):
+				if waitErr != nil {
-					failures = append(failures, errors.New("timed out waiting for restore worker outcome"))
+					failures = append(failures, waitErr)
-					cancel()
+					result, workerJoined, waitErr = joinRestoreLifecycleWorker(cancel, cancelGate, continueStage, done)
+					if waitErr != nil {
+						failures = append(failures, waitErr)
+					}
 				}
 			}
 			for _, wantErr := range scenario.wantErrors {
@@ -562,6 +805,9 @@ func TestRestoreLifecycleLockExcludesCompetingTransactionsUntilTerminalCleanup(t
 			if err := lifecycleLockFreeAfterTerminalRestore(installation); err != nil {
 				failures = append(failures, err)
 			}
+			if !workerJoined {
+				failures = append(failures, errors.New("restore worker outcome was never joined"))
+			}
 			if len(failures) != 0 {
 				t.Fatal(errors.Join(failures...))
 			}
@@ -620,6 +866,8 @@ func TestRestoreCannotApplyAStaleCheckpointOverAnInterleavedRestore(t *testing.T
 	recoveryObserved := ""
 	checkpointEntered := make(chan struct{})
 	continueCheckpoint := make(chan struct{})
+	firstContext, firstCancel := context.WithCancel(context.Background())
+	defer firstCancel()
 	firstRunner := newBackupRunner(installation, true)
 	firstDeps := restoreTestDependencies(t, firstRunner)
 	recoveryArchive := filepath.Join(t.TempDir(), "first-recovery.zip")
@@ -631,10 +879,14 @@ func TestRestoreCannotApplyAStaleCheckpointOverAnInterleavedRestore(t *testing.T
 		return Preflight(ctx, target, PreflightRequest{Archive: recoveryArchive, Confirm: true, AllowExternalSecrets: true}, permissivePreflightDependencies())
 	}
 	firstDeps.acquireTransaction = lifecycle.AcquireTransaction
-	firstDeps.checkpoint = func(context.Context, *lifecycle.Transaction, config.Installation, CreateRequest) (Result, error) {
+	firstDeps.checkpoint = func(ctx context.Context, _ *lifecycle.Transaction, _ config.Installation, _ CreateRequest) (Result, error) {
 		checkpointState = targetState
 		close(checkpointEntered)
-		<-continueCheckpoint
+		select {
+		case <-continueCheckpoint:
+		case <-ctx.Done():
+			return Result{}, ctx.Err()
+		}
 		return Result{Path: filepath.Join(t.TempDir(), "first-checkpoint.zip")}, nil
 	}
 	firstDeps.restoreFile = func(context.Context, config.Installation, ArchiveEntryMetadata, io.Reader) error {
@@ -649,13 +901,18 @@ func TestRestoreCannotApplyAStaleCheckpointOverAnInterleavedRestore(t *testing.T
 
 	done := make(chan error, 1)
 	go func() {
-		_, err := restoreWithDependencies(context.Background(), installation, RestoreRequest{Archive: archive, Confirm: true}, firstDeps)
+		_, err := restoreWithDependencies(firstContext, installation, RestoreRequest{Archive: archive, Confirm: true}, firstDeps)
 		done <- err
 	}()
-	select {
+	if err := waitRestoreLifecycleSignal(context.Background(), checkpointEntered, "first restore recovery checkpoint"); err != nil {
-	case <-checkpointEntered:
+		_, _, joinErr := joinRestoreErrorWorker(firstCancel, done)
-	case <-time.After(2 * time.Second):
+		if joinErr != nil {
-		t.Fatal("first restore did not begin its recovery checkpoint")
+			err = errors.Join(err, joinErr)
+		}
+		if lockErr := lifecycleLockFreeAfterTerminalRestore(installation); lockErr != nil {
+			err = errors.Join(err, lockErr)
+		}
+		t.Fatal(err)
 	}
 
 	interleavedCheckpoint := false
@@ -672,8 +929,29 @@ func TestRestoreCannotApplyAStaleCheckpointOverAnInterleavedRestore(t *testing.T
 		return nil
 	}
 	_, interleavedErr := restoreWithDependencies(context.Background(), installation, RestoreRequest{Archive: archive, Confirm: true}, interleavedDeps)
-	close(continueCheckpoint)
+	if err := sendRestoreLifecycleSignal(context.Background(), continueCheckpoint, "first restore checkpoint"); err != nil {
-	firstErr := <-done
+		_, _, joinErr := joinRestoreErrorWorker(firstCancel, done)
+		if joinErr != nil {
+			err = errors.Join(err, joinErr)
+		}
+		if lockErr := lifecycleLockFreeAfterTerminalRestore(installation); lockErr != nil {
+			err = errors.Join(err, lockErr)
+		}
+		t.Fatal(err)
+	}
+	firstErr, joined, waitErr := waitRestoreError(context.Background(), done)
+	if waitErr != nil {
+		firstErr, joined, waitErr = joinRestoreErrorWorker(firstCancel, done)
+	}
+	if waitErr != nil {
+		if lockErr := lifecycleLockFreeAfterTerminalRestore(installation); lockErr != nil {
+			waitErr = errors.Join(waitErr, lockErr)
+		}
+		t.Fatal(waitErr)
+	}
+	if !joined {
+		t.Fatal("first restore worker outcome was never joined")
+	}
 
 	if !errors.Is(interleavedErr, lifecycle.ErrLocked) || interleavedCheckpoint || interleavedMutation {
 		t.Fatalf("interleaved restore was admitted: error=%v checkpoint=%t mutation=%t", interleavedErr, interleavedCheckpoint, interleavedMutation)
@@ -989,28 +1267,29 @@ func TestRestoreKeepsAdmissionBarrierActiveUntilVerificationCommits(t *testing.T
 	deps := restoreTestDependencies(t, runner)
 	verificationEntered := make(chan struct{})
 	allowVerification := make(chan struct{})
-	deps.verify["health"] = func(context.Context, config.Installation, archiveRunner) error {
+	caller, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	deps.verify["health"] = func(ctx context.Context, _ config.Installation, _ archiveRunner) error {
 		close(verificationEntered)
-		<-allowVerification
+		select {
+		case <-allowVerification:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
 		if !runner.maintenance {
 			return errors.New("maintenance barrier was removed before verification committed")
 		}
 		return nil
 	}
 
-	type restoreOutcome struct {
+	done := make(chan restoreLifecycleTestOutcome, 1)
-		result RestoreResult
-		err    error
-	}
-	done := make(chan restoreOutcome, 1)
 	go func() {
-		result, err := restoreWithDependencies(context.Background(), installation, RestoreRequest{Archive: archive, Confirm: true}, deps)
+		result, err := restoreWithDependencies(caller, installation, RestoreRequest{Archive: archive, Confirm: true}, deps)
-		done <- restoreOutcome{result: result, err: err}
+		done <- restoreLifecycleTestOutcome{result: result, err: err}
 	}()
-	select {
+	var failures []error
-	case <-verificationEntered:
+	if err := waitRestoreLifecycleSignal(context.Background(), verificationEntered, "post-start verification"); err != nil {
-	case <-time.After(2 * time.Second):
+		failures = append(failures, err)
-		t.Fatal("restore did not reach post-start verification")
 	}
 
 	// A newly admitted operation observes the same durable barrier as the backend gate. No
@@ -1019,17 +1298,44 @@ func TestRestoreKeepsAdmissionBarrierActiveUntilVerificationCommits(t *testing.T
 	go func() {
 		admissionAttempt <- !runner.maintenance
 	}()
-	admissionAllowed := <-admissionAttempt
+	admissionAllowed, err := waitRestoreAdmissionAttempt(context.Background(), admissionAttempt)
-	close(allowVerification)
+	if err != nil {
-	outcome := <-done
+		failures = append(failures, err)
+	}
+	if err := sendRestoreLifecycleSignal(context.Background(), allowVerification, "post-start verification"); err != nil {
+		failures = append(failures, err)
+	}
+	outcome, workerJoined, err := waitRestoreLifecycleOutcome(context.Background(), done)
+	if err != nil {
+		failures = append(failures, err)
+	}
+	if !workerJoined {
+		joinedOutcome, joined, joinErr := joinRestoreLifecycleWorker(cancel, nil, nil, done)
+		if joined {
+			outcome = joinedOutcome
+			workerJoined = true
+		}
+		if joinErr != nil {
+			failures = append(failures, joinErr)
+		}
+	}
 	if admissionAllowed {
-		t.Fatal("a new operation could enter while restore verification was still in progress")
+		failures = append(failures, errors.New("a new operation could enter while restore verification was still in progress"))
 	}
 	if outcome.err != nil || !outcome.result.Verified {
-		t.Fatalf("restore result = %#v, %v; want successful verified restore", outcome.result, outcome.err)
+		failures = append(failures, fmt.Errorf("restore result = %#v, %v; want successful verified restore", outcome.result, outcome.err))
 	}
 	if runner.maintenance {
-		t.Fatal("maintenance barrier remained active after successful verification")
+		failures = append(failures, errors.New("maintenance barrier remained active after successful verification"))
+	}
+	if err := lifecycleLockFreeAfterTerminalRestore(installation); err != nil {
+		failures = append(failures, err)
+	}
+	if !workerJoined {
+		failures = append(failures, errors.New("restore worker outcome was never joined"))
+	}
+	if len(failures) != 0 {
+		t.Fatal(errors.Join(failures...))
 	}
 }
 

tools/tht/internal/safeio/private_windows_test.go

@@ -4,7 +4,6 @@ package safeio
 
 import (
 	"errors"
-	"fmt"
 	"os"
 	"path/filepath"
 	"runtime"
@@ -155,7 +154,7 @@ func TestOwnerOnlyDACLNativeShape(t *testing.T) {
 	}
 	parents, target, err := openCanonicalWindowsParent(directory)
 	if err != nil {
-		t.Fatalf("openCanonicalWindowsParent() error = %T %v; relative-open matrix: %s", err, err, windowsRelativeOpenMatrix(directory))
+		t.Fatalf("openCanonicalWindowsParent() error = %T %v", err, err)
 	}
 	defer parents.Close()
 	handle, err := openWindowsRelativeComponent(
@@ -214,58 +213,6 @@ func TestOwnerOnlyDACLNativeShape(t *testing.T) {
 	}
 }
 
-func windowsRelativeOpenMatrix(path string) string {
-	volume := filepath.VolumeName(path)
-	root := volume + `\`
-	components := strings.Split(strings.TrimPrefix(path, root), `\`)
-	if volume == "" || len(components) == 0 || components[0] == "" {
-		return "invalid diagnostic path"
-	}
-	rootHandle, err := openWindowsComponentWithAccess(root, true, windows.GENERIC_READ)
-	if err != nil {
-		return fmt.Sprintf("root=%T %v", err, err)
-	}
-	defer windows.CloseHandle(rootHandle)
-	type attempt struct {
-		name      string
-		access    uint32
-		options   uint32
-		shareMode uint32
-	}
-	attempts := []attempt{
-		{name: "generic-read-directory", access: windows.GENERIC_READ, options: windows.FILE_DIRECTORY_FILE},
-		{name: "generic-read-sync-directory", access: windows.GENERIC_READ, options: windows.FILE_DIRECTORY_FILE | windows.FILE_SYNCHRONOUS_IO_NONALERT},
-		{name: "generic-read-reparse-directory", access: windows.GENERIC_READ, options: windows.FILE_DIRECTORY_FILE | windows.FILE_OPEN_REPARSE_POINT},
-		{name: "generic-read-all-options", access: windows.GENERIC_READ, options: windows.FILE_DIRECTORY_FILE | windows.FILE_SYNCHRONOUS_IO_NONALERT | windows.FILE_OPEN_REPARSE_POINT},
-		{name: "file-read-all-options", access: windows.FILE_GENERIC_READ, options: windows.FILE_DIRECTORY_FILE | windows.FILE_SYNCHRONOUS_IO_NONALERT | windows.FILE_OPEN_REPARSE_POINT},
-		{name: "traverse-read-all-options", access: windows.GENERIC_READ | windows.FILE_TRAVERSE, options: windows.FILE_DIRECTORY_FILE | windows.FILE_SYNCHRONOUS_IO_NONALERT | windows.FILE_OPEN_REPARSE_POINT},
-		{name: "generic-read-all-share", access: windows.GENERIC_READ, options: windows.FILE_DIRECTORY_FILE | windows.FILE_SYNCHRONOUS_IO_NONALERT | windows.FILE_OPEN_REPARSE_POINT, shareMode: windows.FILE_SHARE_READ | windows.FILE_SHARE_WRITE | windows.FILE_SHARE_DELETE},
-	}
-	results := make([]string, 0, len(attempts))
-	for _, test := range attempts {
-		shareMode := test.shareMode
-		if shareMode == 0 {
-			shareMode = windowsRetainedHandleShareMode
-		}
-		handle, openErr := openWindowsRelativeObjectWithShareMode(
-			rootHandle,
-			components[0],
-			test.access,
-			windows.FILE_OPEN,
-			test.options,
-			nil,
-			shareMode,
-		)
-		if openErr == nil {
-			windows.CloseHandle(handle)
-			results = append(results, test.name+"=ok")
-			continue
-		}
-		results = append(results, fmt.Sprintf("%s=%T %v", test.name, openErr, openErr))
-	}
-	return strings.Join(results, "; ")
-}
-
 func TestWindowsPrivateRegularCleanupClosesAfterDeleteDispositionFailure(t *testing.T) {
 	var calls []string
 	err := finishWindowsPrivateRegularCleanup(