feat(thothctl): add safe Pi core restart
This commit is contained in:
@@ -0,0 +1,235 @@
|
|||||||
|
package pi
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
errInterruptedRestart = errors.New("a previous Pi restart is incomplete; recover lifecycle maintenance before restarting again")
|
||||||
|
errRestartImageDrift = errors.New("core image changed during Pi restart")
|
||||||
|
errRestartConfigurationDrift = errors.New("external endpoint configuration changed during Pi restart")
|
||||||
|
errRestartMountDrift = errors.New("core persistence mount contract changed during Pi restart")
|
||||||
|
)
|
||||||
|
|
||||||
|
type RestartRequest struct {
|
||||||
|
StatePath string
|
||||||
|
UpdateStatePath string
|
||||||
|
Confirm bool
|
||||||
|
Drain bool
|
||||||
|
}
|
||||||
|
|
||||||
|
type RestartResult struct {
|
||||||
|
StatePath string
|
||||||
|
Version string
|
||||||
|
}
|
||||||
|
|
||||||
|
func Restart(ctx context.Context, runner Runner, request RestartRequest) (RestartResult, error) {
|
||||||
|
return restartWithHooks(ctx, runner, request, defaultLifecycleHooks)
|
||||||
|
}
|
||||||
|
|
||||||
|
func restartWithHooks(
|
||||||
|
ctx context.Context,
|
||||||
|
runner Runner,
|
||||||
|
request RestartRequest,
|
||||||
|
hooks lifecycleHooks,
|
||||||
|
) (result RestartResult, retErr error) {
|
||||||
|
if err := validateRestartStatePaths(request.StatePath, request.UpdateStatePath); err != nil {
|
||||||
|
return RestartResult{}, err
|
||||||
|
}
|
||||||
|
lock, err := acquireLock(request.StatePath)
|
||||||
|
if err != nil {
|
||||||
|
return RestartResult{StatePath: request.StatePath}, err
|
||||||
|
}
|
||||||
|
defer lock.Release()
|
||||||
|
if !request.Confirm {
|
||||||
|
return RestartResult{StatePath: request.StatePath}, ErrConfirmationRequired
|
||||||
|
}
|
||||||
|
if state, err := readState(request.UpdateStatePath); err == nil && stateNeedsRecovery(state) {
|
||||||
|
return RestartResult{StatePath: request.StatePath}, ErrInterruptedUpdate
|
||||||
|
} else if err != nil && !errors.Is(err, os.ErrNotExist) {
|
||||||
|
return RestartResult{StatePath: request.StatePath}, err
|
||||||
|
}
|
||||||
|
if state, err := readState(request.StatePath); err == nil && state.MutationStarted {
|
||||||
|
return RestartResult{StatePath: request.StatePath}, errInterruptedRestart
|
||||||
|
} else if err != nil && !errors.Is(err, os.ErrNotExist) {
|
||||||
|
return RestartResult{StatePath: request.StatePath}, err
|
||||||
|
}
|
||||||
|
if err := setMaintenance(ctx, runner, true); err != nil {
|
||||||
|
return RestartResult{StatePath: request.StatePath}, err
|
||||||
|
}
|
||||||
|
clearMaintenance := true
|
||||||
|
mutationStarted := false
|
||||||
|
var state State
|
||||||
|
defer func() {
|
||||||
|
if !clearMaintenance {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if clearErr := setMaintenance(context.Background(), runner, false); clearErr != nil {
|
||||||
|
result.StatePath = request.StatePath
|
||||||
|
if mutationStarted {
|
||||||
|
clearMaintenance = false
|
||||||
|
cause := clearErr
|
||||||
|
if writeErr := hooks.writeState(request.StatePath, state); writeErr != nil {
|
||||||
|
cause = errors.Join(cause, fmt.Errorf("restart recovery state could not be restored: %w", writeErr))
|
||||||
|
}
|
||||||
|
retErr = errors.Join(retErr, recoveryRequired("Pi restart maintenance cleanup failed", cause))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
retErr = errors.Join(retErr, fmt.Errorf("maintenance admission gate could not be cleared: %w", clearErr))
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
if err := waitForInactiveSessions(ctx, runner, request.Drain, hooks.sleep); err != nil {
|
||||||
|
return RestartResult{StatePath: request.StatePath}, err
|
||||||
|
}
|
||||||
|
if err := Doctor(ctx, runner); err != nil {
|
||||||
|
return RestartResult{StatePath: request.StatePath}, err
|
||||||
|
}
|
||||||
|
version, err := Status(ctx, runner)
|
||||||
|
if err != nil {
|
||||||
|
return RestartResult{StatePath: request.StatePath}, err
|
||||||
|
}
|
||||||
|
configured, err := renderedCore(ctx, runner)
|
||||||
|
if err != nil {
|
||||||
|
return RestartResult{StatePath: request.StatePath, Version: version}, err
|
||||||
|
}
|
||||||
|
previous, err := runningImage(ctx, runner, configured.Reference)
|
||||||
|
if err != nil {
|
||||||
|
return RestartResult{StatePath: request.StatePath, Version: version}, err
|
||||||
|
}
|
||||||
|
previous.ConfigurationSHA = configured.ConfigurationSHA
|
||||||
|
state = State{
|
||||||
|
Transaction: lifecycleTransaction(request.StatePath),
|
||||||
|
Phase: PhasePreflight,
|
||||||
|
Target: Target{Version: version, Source: "restart"},
|
||||||
|
Previous: previous,
|
||||||
|
}
|
||||||
|
if err := hooks.writeState(request.StatePath, state); err != nil {
|
||||||
|
return RestartResult{StatePath: request.StatePath, Version: version}, err
|
||||||
|
}
|
||||||
|
if running, err := activeSessions(ctx, runner); err != nil {
|
||||||
|
return RestartResult{StatePath: request.StatePath, Version: version}, err
|
||||||
|
} else if running {
|
||||||
|
return RestartResult{StatePath: request.StatePath, Version: version}, ErrActiveSessions
|
||||||
|
}
|
||||||
|
state.MutationStarted = true
|
||||||
|
if err := hooks.writeState(request.StatePath, state); err != nil {
|
||||||
|
return RestartResult{StatePath: request.StatePath, Version: version}, err
|
||||||
|
}
|
||||||
|
mutationStarted = true
|
||||||
|
clearMaintenance = false
|
||||||
|
if err := recreateCore(ctx, runner); err != nil {
|
||||||
|
return RestartResult{StatePath: request.StatePath, Version: version}, recoveryRequired("Pi restart core recreation failed", err)
|
||||||
|
}
|
||||||
|
if err := ensureMaintenance(ctx, runner); err != nil {
|
||||||
|
return RestartResult{StatePath: request.StatePath, Version: version}, recoveryRequired("Pi restart maintenance proof failed", err)
|
||||||
|
}
|
||||||
|
state.Phase = PhaseRecreated
|
||||||
|
if err := hooks.writeState(request.StatePath, state); err != nil {
|
||||||
|
return RestartResult{StatePath: request.StatePath, Version: version}, recoveryRequired("Pi restart recreation state could not be recorded", err)
|
||||||
|
}
|
||||||
|
if err := verifyRestart(ctx, runner, version, previous); err != nil {
|
||||||
|
return RestartResult{StatePath: request.StatePath, Version: version}, recoveryRequired("Pi restart verification failed", err)
|
||||||
|
}
|
||||||
|
state.Phase = PhaseVerified
|
||||||
|
if err := hooks.writeState(request.StatePath, state); err != nil {
|
||||||
|
return RestartResult{StatePath: request.StatePath, Version: version}, recoveryRequired("Pi restart verification state could not be recorded", err)
|
||||||
|
}
|
||||||
|
if err := hooks.removeFile(request.StatePath); err != nil {
|
||||||
|
return RestartResult{StatePath: request.StatePath, Version: version}, recoveryRequired("Pi restart recovery state could not be removed", err)
|
||||||
|
}
|
||||||
|
clearMaintenance = true
|
||||||
|
return RestartResult{StatePath: request.StatePath, Version: version}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func verifyRestart(ctx context.Context, runner Runner, wanted string, previous Image) error {
|
||||||
|
if err := Doctor(ctx, runner); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
version, err := Status(ctx, runner)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if version != wanted {
|
||||||
|
return errors.New("Pi version changed during core restart")
|
||||||
|
}
|
||||||
|
configured, err := renderedCore(ctx, runner)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
after, err := runningImage(ctx, runner, configured.Reference)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if after.ID != previous.ID {
|
||||||
|
return errRestartImageDrift
|
||||||
|
}
|
||||||
|
if configured.ConfigurationSHA != previous.ConfigurationSHA {
|
||||||
|
return errRestartConfigurationDrift
|
||||||
|
}
|
||||||
|
if !sameMounts(previous.Mounts, after.Mounts) {
|
||||||
|
return errRestartMountDrift
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func RecoverLifecycleMaintenance(
|
||||||
|
ctx context.Context,
|
||||||
|
runner Runner,
|
||||||
|
updateStatePath string,
|
||||||
|
restartStatePath string,
|
||||||
|
confirm bool,
|
||||||
|
) error {
|
||||||
|
if err := validateRestartStatePaths(restartStatePath, updateStatePath); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if !confirm {
|
||||||
|
return ErrConfirmationRequired
|
||||||
|
}
|
||||||
|
lock, err := acquireLock(restartStatePath)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer lock.Release()
|
||||||
|
|
||||||
|
restartState, restartErr := readState(restartStatePath)
|
||||||
|
if restartErr == nil {
|
||||||
|
if restartState.MutationStarted {
|
||||||
|
if restartState.Target.Source != "restart" || restartState.Target.Version == "" {
|
||||||
|
return errors.New("restart recovery state is incomplete")
|
||||||
|
}
|
||||||
|
if err := ensureMaintenance(ctx, runner); err != nil {
|
||||||
|
return recoveryRequired("Pi restart maintenance recovery failed", err)
|
||||||
|
}
|
||||||
|
if err := verifyRestart(ctx, runner, restartState.Target.Version, restartState.Previous); err != nil {
|
||||||
|
return recoveryRequired("Pi restart recovery verification failed", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if err := durableRemove(restartStatePath); err != nil {
|
||||||
|
return recoveryRequired("Pi restart recovery state could not be removed", err)
|
||||||
|
}
|
||||||
|
} else if !errors.Is(restartErr, os.ErrNotExist) {
|
||||||
|
return restartErr
|
||||||
|
}
|
||||||
|
return recoverMaintenanceLocked(ctx, runner, updateStatePath)
|
||||||
|
}
|
||||||
|
|
||||||
|
func validateRestartStatePaths(restartStatePath, updateStatePath string) error {
|
||||||
|
if restartStatePath == "" {
|
||||||
|
return errors.New("restart state path is required")
|
||||||
|
}
|
||||||
|
if updateStatePath == "" {
|
||||||
|
return errors.New("update state path is required")
|
||||||
|
}
|
||||||
|
if filepath.Clean(restartStatePath) == filepath.Clean(updateStatePath) {
|
||||||
|
return fmt.Errorf("%w: restart and update state paths must remain separate", ErrInvalidRequest)
|
||||||
|
}
|
||||||
|
if lifecycleLockPath(restartStatePath) != lifecycleLockPath(updateStatePath) {
|
||||||
|
return fmt.Errorf("%w: restart and update state paths must share one lifecycle control directory", ErrInvalidRequest)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
@@ -0,0 +1,292 @@
|
|||||||
|
package pi
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestRestartRequiresConfirmationWithoutInvokingCompose(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
fake := newFakeRunner()
|
||||||
|
_, err := Restart(context.Background(), fake, RestartRequest{
|
||||||
|
StatePath: filepath.Join(dir, "restart-state.json"),
|
||||||
|
UpdateStatePath: filepath.Join(dir, "update-state.json"),
|
||||||
|
})
|
||||||
|
if !errors.Is(err, ErrConfirmationRequired) {
|
||||||
|
t.Fatalf("Restart() error = %v, want ErrConfirmationRequired", err)
|
||||||
|
}
|
||||||
|
assertNotCalled(t, fake.calls, "compose")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRestartDrainsRecreatesOnlyCoreAndRetainsImage(t *testing.T) {
|
||||||
|
fake := newFakeRunner()
|
||||||
|
fake.activeSessions = true
|
||||||
|
dir := t.TempDir()
|
||||||
|
hooks := defaultLifecycleHooks
|
||||||
|
hooks.sleep = func(time.Duration) { fake.activeSessions = false }
|
||||||
|
|
||||||
|
result, err := restartWithHooks(context.Background(), fake, RestartRequest{
|
||||||
|
StatePath: filepath.Join(dir, "restart-state.json"),
|
||||||
|
UpdateStatePath: filepath.Join(dir, "update-state.json"),
|
||||||
|
Confirm: true,
|
||||||
|
Drain: true,
|
||||||
|
}, hooks)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if result.Version != fake.version {
|
||||||
|
t.Fatalf("version = %q, want %q", result.Version, fake.version)
|
||||||
|
}
|
||||||
|
assertCalled(t, fake.calls, "up --detach --wait --wait-timeout 45 --no-deps --force-recreate core")
|
||||||
|
assertNotCalled(t, fake.calls, "build --pull")
|
||||||
|
assertNotCalled(t, fake.calls, "pull ")
|
||||||
|
assertNotCalled(t, fake.calls, "frontend")
|
||||||
|
if _, err := os.Stat(result.StatePath); !errors.Is(err, os.ErrNotExist) {
|
||||||
|
t.Fatalf("successful restart state still exists: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRestartRefusesActiveSessionsWithoutDrain(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
fake := newFakeRunner()
|
||||||
|
fake.activeSessions = true
|
||||||
|
|
||||||
|
_, err := Restart(context.Background(), fake, RestartRequest{
|
||||||
|
StatePath: filepath.Join(dir, "restart-state.json"),
|
||||||
|
UpdateStatePath: filepath.Join(dir, "update-state.json"),
|
||||||
|
Confirm: true,
|
||||||
|
})
|
||||||
|
if !errors.Is(err, ErrActiveSessions) {
|
||||||
|
t.Fatalf("Restart() error = %v, want ErrActiveSessions", err)
|
||||||
|
}
|
||||||
|
if fake.maintenance {
|
||||||
|
t.Fatal("maintenance remained active after refusing pre-mutation restart")
|
||||||
|
}
|
||||||
|
if fake.recreated {
|
||||||
|
t.Fatal("core was recreated with active sessions")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRestartRefusesInterruptedUpdateOrRestartState(t *testing.T) {
|
||||||
|
for _, stateFile := range []string{"update-state.json", "restart-state.json"} {
|
||||||
|
t.Run(stateFile, func(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
fake := newFakeRunner()
|
||||||
|
previous := stateImageForTest(t, fake)
|
||||||
|
writeStateForTest(t, filepath.Join(dir, stateFile), State{
|
||||||
|
Transaction: "interrupted",
|
||||||
|
Phase: PhaseRecreated,
|
||||||
|
Target: Target{Version: fake.version, Source: "restart"},
|
||||||
|
Previous: previous,
|
||||||
|
MutationStarted: true,
|
||||||
|
})
|
||||||
|
fake.calls = nil
|
||||||
|
|
||||||
|
_, err := Restart(context.Background(), fake, RestartRequest{
|
||||||
|
StatePath: filepath.Join(dir, "restart-state.json"),
|
||||||
|
UpdateStatePath: filepath.Join(dir, "update-state.json"),
|
||||||
|
Confirm: true,
|
||||||
|
})
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("Restart() accepted interrupted lifecycle state")
|
||||||
|
}
|
||||||
|
assertNotCalled(t, fake.calls, "compose")
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRestartPreflightFailureNeverRecreatesCoreAndClearsMaintenance(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
fake := newFakeRunner()
|
||||||
|
fake.fail = "preflight"
|
||||||
|
|
||||||
|
_, err := Restart(context.Background(), fake, RestartRequest{
|
||||||
|
StatePath: filepath.Join(dir, "restart-state.json"),
|
||||||
|
UpdateStatePath: filepath.Join(dir, "update-state.json"),
|
||||||
|
Confirm: true,
|
||||||
|
})
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("Restart() error = nil, want preflight failure")
|
||||||
|
}
|
||||||
|
if fake.recreated || fake.currentImage != "sha256:old" {
|
||||||
|
t.Fatalf("preflight failure mutated core: recreated=%t image=%q", fake.recreated, fake.currentImage)
|
||||||
|
}
|
||||||
|
if fake.maintenance {
|
||||||
|
t.Fatal("maintenance remained active after preflight failure")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRestartPostRecreateFailureKeepsMaintenanceAndRecoveryState(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
fake := newFakeRunner()
|
||||||
|
fake.fail = "health"
|
||||||
|
statePath := filepath.Join(dir, "restart-state.json")
|
||||||
|
|
||||||
|
_, err := Restart(context.Background(), fake, RestartRequest{
|
||||||
|
StatePath: statePath,
|
||||||
|
UpdateStatePath: filepath.Join(dir, "update-state.json"),
|
||||||
|
Confirm: true,
|
||||||
|
})
|
||||||
|
var recovery *RecoveryRequiredError
|
||||||
|
if !errors.As(err, &recovery) {
|
||||||
|
t.Fatalf("Restart() error = %v, want RecoveryRequiredError", err)
|
||||||
|
}
|
||||||
|
if !fake.recreated {
|
||||||
|
t.Fatal("post-recreate failure occurred before core recreation")
|
||||||
|
}
|
||||||
|
if !fake.maintenance {
|
||||||
|
t.Fatal("maintenance was cleared after post-recreate failure")
|
||||||
|
}
|
||||||
|
state, stateErr := readState(statePath)
|
||||||
|
if stateErr != nil || !state.MutationStarted {
|
||||||
|
t.Fatalf("restart recovery state = %+v, %v; want durable mutation state", state, stateErr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRestartMaintenanceClearFailureRestoresRecoveryState(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
fake := newFakeRunner()
|
||||||
|
fake.fail = "maintenance-clear"
|
||||||
|
statePath := filepath.Join(dir, "restart-state.json")
|
||||||
|
|
||||||
|
_, err := Restart(context.Background(), fake, RestartRequest{
|
||||||
|
StatePath: statePath,
|
||||||
|
UpdateStatePath: filepath.Join(dir, "update-state.json"),
|
||||||
|
Confirm: true,
|
||||||
|
})
|
||||||
|
var recovery *RecoveryRequiredError
|
||||||
|
if !errors.As(err, &recovery) {
|
||||||
|
t.Fatalf("Restart() error = %v, want RecoveryRequiredError", err)
|
||||||
|
}
|
||||||
|
if !fake.maintenance {
|
||||||
|
t.Fatal("maintenance was cleared despite deactivation failure")
|
||||||
|
}
|
||||||
|
state, stateErr := readState(statePath)
|
||||||
|
if stateErr != nil || state.Phase != PhaseVerified || !state.MutationStarted {
|
||||||
|
t.Fatalf("restart recovery state = %+v, %v; want durable verified mutation state", state, stateErr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRecoverLifecycleMaintenanceVerifiesAndClearsRestartState(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
fake := newFakeRunner()
|
||||||
|
fake.maintenance = true
|
||||||
|
previous := stateImageForTest(t, fake)
|
||||||
|
restartStatePath := filepath.Join(dir, "restart-state.json")
|
||||||
|
updateStatePath := filepath.Join(dir, "update-state.json")
|
||||||
|
writeStateForTest(t, restartStatePath, State{
|
||||||
|
Transaction: "restart-recovery",
|
||||||
|
Phase: PhaseRecreated,
|
||||||
|
Target: Target{Version: fake.version, Source: "restart"},
|
||||||
|
Previous: previous,
|
||||||
|
MutationStarted: true,
|
||||||
|
})
|
||||||
|
candidate := previous
|
||||||
|
candidate.Reference = "thothii-core:thothctl-recover-candidate"
|
||||||
|
writeStateForTest(t, updateStatePath, State{
|
||||||
|
Transaction: "update-recovery",
|
||||||
|
Phase: PhasePromoting,
|
||||||
|
Target: Target{Version: fake.version, Source: string(BuildSource)},
|
||||||
|
Previous: previous,
|
||||||
|
Candidate: candidate,
|
||||||
|
MutationStarted: true,
|
||||||
|
})
|
||||||
|
if err := writeLifecycleOverride(currentImageOverridePath(updateStatePath), candidate.Reference); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
fake.calls = nil
|
||||||
|
|
||||||
|
if err := RecoverLifecycleMaintenance(context.Background(), fake, updateStatePath, restartStatePath, true); err != nil {
|
||||||
|
t.Fatalf("RecoverLifecycleMaintenance() error = %v", err)
|
||||||
|
}
|
||||||
|
if _, err := os.Stat(restartStatePath); !errors.Is(err, os.ErrNotExist) {
|
||||||
|
t.Fatalf("restart recovery state still exists: %v", err)
|
||||||
|
}
|
||||||
|
if fake.maintenance {
|
||||||
|
t.Fatal("maintenance remained active after both lifecycle states were verified")
|
||||||
|
}
|
||||||
|
updateState, err := readState(updateStatePath)
|
||||||
|
if err != nil || updateState.Phase != PhaseVerified {
|
||||||
|
t.Fatalf("update recovery state = %+v, %v; want verified image rollback metadata", updateState, err)
|
||||||
|
}
|
||||||
|
deactivate := callIndex(fake.calls, "/internal/maintenance/deactivate")
|
||||||
|
lastVerification := lastCallIndexBefore(fake.calls, "/pi-management/test", deactivate)
|
||||||
|
if deactivate < 0 || lastVerification < 0 {
|
||||||
|
t.Fatalf("calls = %v; want verification before maintenance deactivation", fake.calls)
|
||||||
|
}
|
||||||
|
verificationCount := 0
|
||||||
|
for index := 0; index < deactivate; index++ {
|
||||||
|
if strings.Contains(fake.calls[index], "/internal/maintenance/deactivate") {
|
||||||
|
t.Fatalf("maintenance reopened before combined verification: %v", fake.calls)
|
||||||
|
}
|
||||||
|
if strings.Contains(fake.calls[index], "/pi-management/test") {
|
||||||
|
verificationCount++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if verificationCount < 3 {
|
||||||
|
t.Fatalf("verification calls before maintenance deactivation = %d, want restart, update, and final proofs: %v", verificationCount, fake.calls)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRestartRejectsImageConfigurationAndMountDrift(t *testing.T) {
|
||||||
|
for _, test := range []struct {
|
||||||
|
failure string
|
||||||
|
want error
|
||||||
|
}{
|
||||||
|
{failure: "image-drift", want: errRestartImageDrift},
|
||||||
|
{failure: "config-drift", want: errRestartConfigurationDrift},
|
||||||
|
{failure: "mount-drift", want: errRestartMountDrift},
|
||||||
|
} {
|
||||||
|
t.Run(test.failure, func(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
fake := newFakeRunner()
|
||||||
|
fake.fail = test.failure
|
||||||
|
statePath := filepath.Join(dir, "restart-state.json")
|
||||||
|
|
||||||
|
_, err := Restart(context.Background(), fake, RestartRequest{
|
||||||
|
StatePath: statePath,
|
||||||
|
UpdateStatePath: filepath.Join(dir, "update-state.json"),
|
||||||
|
Confirm: true,
|
||||||
|
})
|
||||||
|
if !errors.Is(err, test.want) {
|
||||||
|
t.Fatalf("Restart() error = %v, want errors.Is(..., %v)", err, test.want)
|
||||||
|
}
|
||||||
|
if !fake.maintenance {
|
||||||
|
t.Fatal("maintenance was cleared after restart identity drift")
|
||||||
|
}
|
||||||
|
if _, stateErr := os.Stat(statePath); stateErr != nil {
|
||||||
|
t.Fatalf("restart recovery state missing after drift: %v", stateErr)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRestartPreservesVerifiedUpdateState(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
fake := newFakeRunner()
|
||||||
|
updateStatePath := filepath.Join(dir, "update-state.json")
|
||||||
|
writeStateForTest(t, updateStatePath, State{
|
||||||
|
Transaction: "verified-update",
|
||||||
|
Phase: PhaseVerified,
|
||||||
|
Target: Target{Version: fake.version, Source: string(BuildSource)},
|
||||||
|
Previous: stateImageForTest(t, fake),
|
||||||
|
})
|
||||||
|
before := readStateBytes(t, updateStatePath)
|
||||||
|
|
||||||
|
if _, err := Restart(context.Background(), fake, RestartRequest{
|
||||||
|
StatePath: filepath.Join(dir, "restart-state.json"),
|
||||||
|
UpdateStatePath: updateStatePath,
|
||||||
|
Confirm: true,
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
after := readStateBytes(t, updateStatePath)
|
||||||
|
if string(after) != string(before) {
|
||||||
|
t.Fatal("restart changed verified update rollback metadata")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -51,7 +51,7 @@ type Mount struct {
|
|||||||
Options string `json:"options,omitempty"`
|
Options string `json:"options,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// Target records the immutable input selected by the operator. Source is either build or a
|
// Target records the immutable input selected by the operator. Source is build, restart, or a
|
||||||
// digest-pinned image reference; it intentionally never contains credentials.
|
// digest-pinned image reference; it intentionally never contains credentials.
|
||||||
type Target struct {
|
type Target struct {
|
||||||
Version string `json:"version"`
|
Version string `json:"version"`
|
||||||
|
|||||||
@@ -119,28 +119,9 @@ func updateWithHooks(ctx context.Context, runner Runner, request Request, hooks
|
|||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
running, err := activeSessions(ctx, runner)
|
if err := waitForInactiveSessions(ctx, runner, request.Drain, hooks.sleep); err != nil {
|
||||||
if err != nil {
|
|
||||||
return Result{StatePath: request.StatePath}, err
|
return Result{StatePath: request.StatePath}, err
|
||||||
}
|
}
|
||||||
if running {
|
|
||||||
if !request.Drain {
|
|
||||||
return Result{StatePath: request.StatePath}, ErrActiveSessions
|
|
||||||
}
|
|
||||||
for attempts := 0; attempts < 30; attempts++ {
|
|
||||||
running, err = activeSessions(ctx, runner)
|
|
||||||
if err != nil {
|
|
||||||
return Result{StatePath: request.StatePath}, err
|
|
||||||
}
|
|
||||||
if !running {
|
|
||||||
break
|
|
||||||
}
|
|
||||||
hooks.sleep(time.Second)
|
|
||||||
}
|
|
||||||
if running {
|
|
||||||
return Result{StatePath: request.StatePath}, ErrActiveSessions
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if err := Doctor(ctx, runner); err != nil {
|
if err := Doctor(ctx, runner); err != nil {
|
||||||
return Result{StatePath: request.StatePath}, err
|
return Result{StatePath: request.StatePath}, err
|
||||||
}
|
}
|
||||||
@@ -192,7 +173,7 @@ func updateWithHooks(ctx context.Context, runner Runner, request Request, hooks
|
|||||||
result, retErr, clearMaintenance = failPreparation(request.StatePath, overridePath, state, err, hooks)
|
result, retErr, clearMaintenance = failPreparation(request.StatePath, overridePath, state, err, hooks)
|
||||||
return result, retErr
|
return result, retErr
|
||||||
}
|
}
|
||||||
running, err = activeSessions(ctx, runner)
|
running, err := activeSessions(ctx, runner)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
result, retErr, clearMaintenance = failPreparation(request.StatePath, overridePath, state, err, hooks)
|
result, retErr, clearMaintenance = failPreparation(request.StatePath, overridePath, state, err, hooks)
|
||||||
return result, retErr
|
return result, retErr
|
||||||
@@ -554,6 +535,30 @@ func activeSessions(ctx context.Context, runner Runner) (bool, error) {
|
|||||||
return false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func waitForInactiveSessions(ctx context.Context, runner Runner, drain bool, sleep func(time.Duration)) error {
|
||||||
|
running, err := activeSessions(ctx, runner)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if !running {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if !drain {
|
||||||
|
return ErrActiveSessions
|
||||||
|
}
|
||||||
|
for attempts := 0; attempts < 30; attempts++ {
|
||||||
|
running, err = activeSessions(ctx, runner)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if !running {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
sleep(time.Second)
|
||||||
|
}
|
||||||
|
return ErrActiveSessions
|
||||||
|
}
|
||||||
|
|
||||||
func runningImage(ctx context.Context, runner Runner, reference string) (Image, error) {
|
func runningImage(ctx context.Context, runner Runner, reference string) (Image, error) {
|
||||||
container, err := runCompose(ctx, runner, "ps", "-q", "core")
|
container, err := runCompose(ctx, runner, "ps", "-q", "core")
|
||||||
if err != nil || strings.TrimSpace(container.Stdout) == "" {
|
if err != nil || strings.TrimSpace(container.Stdout) == "" {
|
||||||
@@ -808,6 +813,10 @@ func RecoverMaintenance(ctx context.Context, runner Runner, statePath string, co
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
defer lock.Release()
|
defer lock.Release()
|
||||||
|
return recoverMaintenanceLocked(ctx, runner, statePath)
|
||||||
|
}
|
||||||
|
|
||||||
|
func recoverMaintenanceLocked(ctx context.Context, runner Runner, statePath string) error {
|
||||||
state, stateErr := readState(statePath)
|
state, stateErr := readState(statePath)
|
||||||
if stateErr == nil {
|
if stateErr == nil {
|
||||||
transactionOverride := lifecycleOverridePath(statePath, state.Transaction)
|
transactionOverride := lifecycleOverridePath(statePath, state.Transaction)
|
||||||
|
|||||||
@@ -863,6 +863,7 @@ type fakeRunner struct {
|
|||||||
labelVersion string
|
labelVersion string
|
||||||
activeSessions bool
|
activeSessions bool
|
||||||
built bool
|
built bool
|
||||||
|
recreated bool
|
||||||
currentImage string
|
currentImage string
|
||||||
mountsJSON string
|
mountsJSON string
|
||||||
sessionsWire string
|
sessionsWire string
|
||||||
@@ -911,7 +912,7 @@ func (f *fakeRunner) Run(_ context.Context, args []string, _ io.Reader) (compose
|
|||||||
f.execFailuresWhileStopped++
|
f.execFailuresWhileStopped++
|
||||||
return compose.Result{ExitCode: 1}, errors.New("core service is not running")
|
return compose.Result{ExitCode: 1}, errors.New("core service is not running")
|
||||||
}
|
}
|
||||||
if f.built && f.fail != "compensation" && strings.Contains(call, "image tag sha256:old") {
|
if (f.built || f.recreated) && f.fail != "compensation" && strings.Contains(call, "image tag sha256:old") {
|
||||||
f.fail = ""
|
f.fail = ""
|
||||||
}
|
}
|
||||||
if f.fail == "preflight" && strings.Contains(call, "config --format json") {
|
if f.fail == "preflight" && strings.Contains(call, "config --format json") {
|
||||||
@@ -926,22 +927,22 @@ func (f *fakeRunner) Run(_ context.Context, args []string, _ io.Reader) (compose
|
|||||||
if f.fail == "tag" && len(args) >= 4 && args[0] == "image" && args[1] == "tag" && strings.Contains(args[3], "-candidate") {
|
if f.fail == "tag" && len(args) >= 4 && args[0] == "image" && args[1] == "tag" && strings.Contains(args[3], "-candidate") {
|
||||||
return compose.Result{ExitCode: 1}, errors.New("tag token=secret")
|
return compose.Result{ExitCode: 1}, errors.New("tag token=secret")
|
||||||
}
|
}
|
||||||
if f.fail == "health" && f.built && strings.Contains(call, "curl -fsS http://127.0.0.1:8787/health") {
|
if f.fail == "health" && (f.built || f.recreated) && strings.Contains(call, "curl -fsS http://127.0.0.1:8787/health") {
|
||||||
return compose.Result{ExitCode: 1}, errors.New("health token=secret")
|
return compose.Result{ExitCode: 1}, errors.New("health token=secret")
|
||||||
}
|
}
|
||||||
if f.fail == "compensation" && f.built && !f.rollbackPrepared && strings.Contains(call, "curl -fsS http://127.0.0.1:8787/health") {
|
if f.fail == "compensation" && (f.built || f.recreated) && !f.rollbackPrepared && strings.Contains(call, "curl -fsS http://127.0.0.1:8787/health") {
|
||||||
return compose.Result{ExitCode: 1}, errors.New("candidate health failure")
|
return compose.Result{ExitCode: 1}, errors.New("candidate health failure")
|
||||||
}
|
}
|
||||||
if f.fail == "version" && f.built && strings.Contains(call, "pi --version") && strings.Contains(call, "exec") {
|
if f.fail == "version" && (f.built || f.recreated) && strings.Contains(call, "pi --version") && strings.Contains(call, "exec") {
|
||||||
return compose.Result{ExitCode: 1}, errors.New("version token=secret")
|
return compose.Result{ExitCode: 1}, errors.New("version token=secret")
|
||||||
}
|
}
|
||||||
if f.fail == "smoke" && f.built && strings.Contains(call, "127.0.0.1:8787/pi-management/test") {
|
if f.fail == "smoke" && (f.built || f.recreated) && strings.Contains(call, "127.0.0.1:8787/pi-management/test") {
|
||||||
return compose.Result{ExitCode: 1}, errors.New("smoke token=secret")
|
return compose.Result{ExitCode: 1}, errors.New("smoke token=secret")
|
||||||
}
|
}
|
||||||
switch {
|
switch {
|
||||||
case strings.Contains(call, "config --format json"):
|
case strings.Contains(call, "config --format json"):
|
||||||
endpoint := "https://llm.example.invalid"
|
endpoint := "https://llm.example.invalid"
|
||||||
if f.fail == "config-drift" && f.currentImage == "sha256:candidate" {
|
if f.fail == "config-drift" && (f.built || f.recreated) {
|
||||||
endpoint = "https://drift.example.invalid"
|
endpoint = "https://drift.example.invalid"
|
||||||
}
|
}
|
||||||
return compose.Result{Stdout: `{"services":{"core":{"image":"` + selectedCoreReference(args, f.configuredImage) + `","environment":{"THT_LLM_URL":"` + endpoint + `"}}}}`}, nil
|
return compose.Result{Stdout: `{"services":{"core":{"image":"` + selectedCoreReference(args, f.configuredImage) + `","environment":{"THT_LLM_URL":"` + endpoint + `"}}}}`}, nil
|
||||||
@@ -953,11 +954,14 @@ func (f *fakeRunner) Run(_ context.Context, args []string, _ io.Reader) (compose
|
|||||||
case strings.Contains(call, "ps -q core"):
|
case strings.Contains(call, "ps -q core"):
|
||||||
return compose.Result{Stdout: "core-container\n"}, nil
|
return compose.Result{Stdout: "core-container\n"}, nil
|
||||||
case strings.Contains(call, "inspect --format {{.Image}}"):
|
case strings.Contains(call, "inspect --format {{.Image}}"):
|
||||||
|
if f.fail == "image-drift" && f.recreated {
|
||||||
|
return compose.Result{Stdout: "sha256:unexpected\n"}, nil
|
||||||
|
}
|
||||||
return compose.Result{Stdout: f.currentImage + "\n"}, nil
|
return compose.Result{Stdout: f.currentImage + "\n"}, nil
|
||||||
case strings.Contains(call, "io.thothii.pi.version"):
|
case strings.Contains(call, "io.thothii.pi.version"):
|
||||||
return compose.Result{Stdout: f.labelVersion + "\n"}, nil
|
return compose.Result{Stdout: f.labelVersion + "\n"}, nil
|
||||||
case strings.Contains(call, "inspect --format {{json .Mounts}}"):
|
case strings.Contains(call, "inspect --format {{json .Mounts}}"):
|
||||||
if f.fail == "mount-drift" && f.currentImage == "sha256:candidate" {
|
if f.fail == "mount-drift" && (f.built || f.recreated) {
|
||||||
return compose.Result{Stdout: `[{"Type":"volume","Name":"wrong-settings","Source":"wrong-settings","Destination":"/data/settings","RW":true}]`}, nil
|
return compose.Result{Stdout: `[{"Type":"volume","Name":"wrong-settings","Source":"wrong-settings","Destination":"/data/settings","RW":true}]`}, nil
|
||||||
}
|
}
|
||||||
if f.mountsJSON != "" {
|
if f.mountsJSON != "" {
|
||||||
@@ -1046,6 +1050,7 @@ func (f *fakeRunner) Run(_ context.Context, args []string, _ io.Reader) (compose
|
|||||||
}
|
}
|
||||||
return compose.Result{}, nil
|
return compose.Result{}, nil
|
||||||
case containsArg(args, "up"):
|
case containsArg(args, "up"):
|
||||||
|
f.recreated = true
|
||||||
f.maintenanceAtRecreate = append(f.maintenanceAtRecreate, f.maintenance)
|
f.maintenanceAtRecreate = append(f.maintenanceAtRecreate, f.maintenance)
|
||||||
reference := selectedCoreReference(args, f.configuredImage)
|
reference := selectedCoreReference(args, f.configuredImage)
|
||||||
if id, ok := f.tags[reference]; ok {
|
if id, ok := f.tags[reference]; ok {
|
||||||
|
|||||||
Reference in New Issue
Block a user