mptyl/ThothII
1
0
Fork 0
Code Issues 25 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix: harden thothctl pi lifecycle

Browse Source
This commit is contained in:
marcopan
2026-08-04 18:40:50 +02:00
parent 20e59b8d32
commit 8fde1f81c7
8 changed files with 308 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/thothctl/internal/pi/update.go
+93 -16
View File
@@ -9,6 +9,7 @@ import (
"regexp"
"sort"
"strings"
"time"
)
var (
@@ -45,6 +46,11 @@ type Result struct {
// Update performs a recoverable core-only Pi update using the default Compose command layout.
func Update(ctx context.Context, runner Runner, request Request) (Result, error) {
lock, err := acquireLock(request.StatePath)
if err != nil {
return Result{StatePath: request.StatePath}, err
}
defer lock.Release()
if request.StatePath == "" {
return Result{}, errors.New("update state path is required")
}
@@ -73,13 +79,30 @@ func Update(ctx context.Context, runner Runner, request Request) (Result, error)
if err != nil {
return Result{StatePath: request.StatePath}, err
}
frontendStopped := false
if running {
if !request.Drain {
return Result{StatePath: request.StatePath}, ErrActiveSessions
}
running, err = activeSessions(ctx, runner)
if err != nil {
return Result{StatePath: request.StatePath}, err
stopped, stopErr := runCompose(ctx, runner, "stop", "frontend")
if stopErr != nil {
return Result{StatePath: request.StatePath}, commandError("frontend admission gate", stopped, stopErr)
}
frontendStopped = true
defer func() {
if frontendStopped {
_, _ = runCompose(context.Background(), runner, "up", "--detach", "frontend")
}
}()
for attempts := 0; attempts < 30; attempts++ {
running, err = activeSessions(ctx, runner)
if err != nil {
return Result{StatePath: request.StatePath}, err
}
if !running {
break
}
time.Sleep(time.Second)
}
if running {
return Result{StatePath: request.StatePath}, ErrActiveSessions
@@ -119,6 +142,17 @@ func Update(ctx context.Context, runner Runner, request Request) (Result, error)
_ = writeState(request.StatePath, state)
return Result{Phase: PhaseFailed, StatePath: request.StatePath}, err
}
if frontendStopped {
running, err = activeSessions(ctx, runner)
if err != nil {
return Result{Phase: PhaseFailed, StatePath: request.StatePath}, err
}
if running {
state.Phase, state.Error = PhaseFailed, "new session admitted while draining"
_ = writeState(request.StatePath, state)
return Result{Phase: PhaseFailed, StatePath: request.StatePath}, ErrActiveSessions
}
}
if err := recreateCore(ctx, runner); err != nil {
state.Phase, state.Error = PhaseFailed, "core recreation failed"
_ = writeState(request.StatePath, state)
@@ -129,7 +163,7 @@ func Update(ctx context.Context, runner Runner, request Request) (Result, error)
if err := writeState(request.StatePath, state); err != nil {
return Result{Phase: PhaseRecreated, StatePath: request.StatePath}, err
}
if err := verifyCandidate(ctx, runner, request.Version, previous.Volumes, previous.ConfigurationSHA); err != nil {
if err := verifyCandidate(ctx, runner, request.Version, previous); err != nil {
return rollbackAfterFailure(ctx, runner, request.StatePath, state, err)
}
state.Phase, state.Error = PhaseVerified, ""
@@ -141,6 +175,11 @@ func Update(ctx context.Context, runner Runner, request Request) (Result, error)
// Rollback restores the image recorded in durable update state. It is safe for interrupted runs.
func Rollback(ctx context.Context, runner Runner, statePath string, confirm bool) (Result, error) {
lock, err := acquireLock(statePath)
if err != nil {
return Result{StatePath: statePath}, err
}
defer lock.Release()
if !confirm {
return Result{StatePath: statePath}, ErrConfirmationRequired
}
@@ -208,15 +247,24 @@ func runningImage(ctx context.Context, runner Runner, reference string) (Image,
if err != nil || strings.TrimSpace(image.Stdout) == "" {
return Image{}, commandError("running core image check", image, err)
}
mounts, err := runner.Run(ctx, []string{"inspect", "--format", "{{range .Mounts}}{{if eq .Type \"volume\"}}{{.Name}}{{\"\\n\"}}{{end}}{{end}}", id}, nil)
mounts, err := runner.Run(ctx, []string{"inspect", "--format", "{{json .Mounts}}", id}, nil)
if err != nil {
return Image{}, commandError("core volume check", mounts, err)
}
volumes := nonEmptyLines(mounts.Stdout)
if len(volumes) == 0 {
return Image{}, errors.New("core has no named volumes to preserve")
var contract []Mount
if err := json.Unmarshal([]byte(mounts.Stdout), &contract); err != nil {
return Image{}, errors.New("core returned invalid persistence mount data")
}
return Image{ID: strings.TrimSpace(image.Stdout), Reference: reference, Volumes: volumes}, nil
if len(contract) == 0 {
return Image{}, errors.New("core has no persistence mounts to preserve")
}
volumes := make([]string, 0, len(contract))
for _, mount := range contract {
if mount.Type == "volume" && mount.Name != "" {
volumes = append(volumes, mount.Name)
}
}
return Image{ID: strings.TrimSpace(image.Stdout), Reference: reference, Volumes: volumes, Mounts: contract}, nil
}
func prepareCandidate(ctx context.Context, runner Runner, request Request, reference string) error {
@@ -239,14 +287,14 @@ func prepareCandidate(ctx context.Context, runner Runner, request Request, refer
}
func recreateCore(ctx context.Context, runner Runner) error {
result, err := runCompose(ctx, runner, "up", "--detach", "--no-deps", "--force-recreate", "core")
result, err := runCompose(ctx, runner, "up", "--detach", "--wait", "--wait-timeout", "45", "--no-deps", "--force-recreate", "core")
if err != nil {
return commandError("core recreation", result, err)
}
return nil
}
func verifyCandidate(ctx context.Context, runner Runner, wanted string, previousVolumes []string, configurationSHA string) error {
func verifyCandidate(ctx context.Context, runner Runner, wanted string, previous Image) error {
health, err := runCompose(ctx, runner, "exec", "-T", "core", "curl", "-fsS", "http://127.0.0.1:8787/health")
if err != nil {
return commandError("core health check", health, err)
@@ -265,15 +313,15 @@ func verifyCandidate(ctx context.Context, runner Runner, wanted string, previous
if err != nil {
return err
}
if configured.ConfigurationSHA != configurationSHA {
if configured.ConfigurationSHA != previous.ConfigurationSHA {
return errors.New("external endpoint configuration changed during Pi update")
}
after, err := runningImage(ctx, runner, configured.Reference)
if err != nil {
return err
}
if !sameStrings(previousVolumes, after.Volumes) {
return errors.New("core volume set changed during Pi update")
if !sameMounts(previous.Mounts, after.Mounts) {
return errors.New("core persistence mount contract changed during Pi update")
}
return nil
}
@@ -294,8 +342,20 @@ func restore(ctx context.Context, runner Runner, previous Image) error {
if err != nil {
return err
}
if !sameStrings(previous.Volumes, after.Volumes) {
return errors.New("core volume set changed during rollback")
if after.ID != previous.ID {
return errors.New("rollback core image does not match recorded previous image")
}
if configured.ConfigurationSHA != previous.ConfigurationSHA {
return errors.New("external endpoint configuration drift prevents rollback proof")
}
if !sameMounts(previous.Mounts, after.Mounts) {
return errors.New("core persistence mount contract changed during rollback")
}
if err := Doctor(ctx, runner); err != nil {
return err
}
if err := Test(ctx, runner); err != nil {
return err
}
return nil
}
@@ -316,3 +376,20 @@ func sameStrings(left, right []string) bool {
sort.Strings(right)
return strings.Join(left, "\x00") == strings.Join(right, "\x00")
}
func sameMounts(left, right []Mount) bool {
if len(left) != len(right) {
return false
}
key := func(m Mount) string {
return m.Type + "\x00" + m.Name + "\x00" + m.Source + "\x00" + m.Destination + "\x00" + fmt.Sprint(m.RW)
}
a, b := make([]string, len(left)), make([]string, len(right))
for i := range left {
a[i] = key(left[i])
}
for i := range right {
b[i] = key(right[i])
}
return sameStrings(a, b)
}