Assistant/runtime/supervisor/durable_store.go

package supervisor

import (
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"sync"
)

// DurableStore is supervisor-owned external control-plane state. It is not a
// candidate Assistant process-local recovery cache.
type DurableStore struct{ path string }

func NewDurableStore(path string) (*DurableStore, error) {
	if path == "" || !filepath.IsAbs(path) {
		return nil, errors.New("supervisor state path must be absolute")
	}
	return &DurableStore{path: path}, nil
}

func (s *DurableStore) Save(value Lifeboat) error {
	if s == nil || s.path == "" {
		return errors.New("supervisor state store is unavailable")
	}
	payload, err := json.Marshal(value)
	if err != nil {
		return err
	}
	if err := os.MkdirAll(filepath.Dir(s.path), 0700); err != nil {
		return err
	}
	temporary, err := os.CreateTemp(filepath.Dir(s.path), ".lifeboat-*.tmp")
	if err != nil {
		return err
	}
	temporaryName := temporary.Name()
	defer os.Remove(temporaryName)
	if err = temporary.Chmod(0600); err == nil {
		_, err = temporary.Write(payload)
	}
	if closeErr := temporary.Close(); err == nil {
		err = closeErr
	}
	if err != nil {
		return err
	}
	return os.Rename(temporaryName, s.path)
}

func (s *DurableStore) Load() (Lifeboat, error) {
	if s == nil || s.path == "" {
		return Lifeboat{}, errors.New("supervisor state store is unavailable")
	}
	payload, err := os.ReadFile(s.path)
	if err != nil {
		return Lifeboat{}, err
	}
	var value Lifeboat
	if err := json.Unmarshal(payload, &value); err != nil {
		return Lifeboat{}, err
	}
	return value, nil
}

// PersistentLifeboat reloads before every transition and commits after it.
type PersistentLifeboat struct {
	store *DurableStore
	mu    sync.Mutex
}

func NewPersistentLifeboat(store *DurableStore) *PersistentLifeboat {
	return &PersistentLifeboat{store: store}
}
func (p *PersistentLifeboat) transition(fn func(*Lifeboat) bool) (bool, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	value, err := p.store.Load()
	if err != nil && !os.IsNotExist(err) {
		return false, err
	}
	ok := fn(&value)
	if !ok {
		return false, nil
	}
	if err := p.store.Save(value); err != nil {
		return false, err
	}
	return true, nil
}
func (p *PersistentLifeboat) Start(candidate Release) (bool, error) {
	return p.transition(func(value *Lifeboat) bool { return value.Start(candidate) })
}

func (p *PersistentLifeboat) StartSet(candidate ReleaseSet) (bool, error) {
	return p.transition(func(value *Lifeboat) bool { return value.StartSet(candidate) })
}

func (p *PersistentLifeboat) EstablishBaseline(release Release) (bool, error) {
	return p.transition(func(value *Lifeboat) bool { return value.EstablishBaseline(release) })
}

func (p *PersistentLifeboat) EstablishBaselineSet(release ReleaseSet) (bool, error) {
	return p.transition(func(value *Lifeboat) bool { return value.EstablishBaselineSet(release) })
}
func (p *PersistentLifeboat) Record(observation Observation) (bool, error) {
	return p.transition(func(value *Lifeboat) bool { return value.Record(observation) })
}

func (p *PersistentLifeboat) RecordSet(observation ReleaseSetObservation) (bool, error) {
	return p.transition(func(value *Lifeboat) bool { return value.RecordSet(observation) })
}
func (p *PersistentLifeboat) Rollback() (bool, error) {
	return p.transition(func(value *Lifeboat) bool { return value.Rollback() })
}
func (p *PersistentLifeboat) SetTraffic(mode TrafficMode, stableUpstream, candidateUpstream string) (bool, error) {
	return p.transition(func(value *Lifeboat) bool { return value.SetTraffic(mode, stableUpstream, candidateUpstream) })
}

func (p *PersistentLifeboat) SetTrafficForSets(mode TrafficMode) (bool, error) {
	return p.transition(func(value *Lifeboat) bool { return value.SetTrafficForSets(mode) })
}

func (p *PersistentLifeboat) Promote() (bool, error) {
	return p.transition(func(value *Lifeboat) bool {
		if value.Current.State != ReleaseHealthy || value.Current.ID == "" {
			return false
		}
		return value.SetTraffic(TrafficCandidate, value.StableUpstream, value.CandidateUpstream)
	})
}

func (p *PersistentLifeboat) PromoteSet() (bool, error) {
	return p.transition(func(value *Lifeboat) bool { return value.PromoteSet() })
}

func (p *PersistentLifeboat) RollbackTraffic() (bool, error) {
	return p.transition(func(value *Lifeboat) bool {
		if !value.Rollback() {
			return false
		}
		return value.SetTraffic(TrafficStable, value.StableUpstream, value.CandidateUpstream)
	})
}

func (p *PersistentLifeboat) RollbackSet() (bool, error) {
	return p.transition(func(value *Lifeboat) bool { return value.RollbackSet() })
}

func (p *PersistentLifeboat) Snapshot() (Lifeboat, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.store.Load()
}