DebTik/internal/safeapply/manager.go

// Package safeapply provides durable, serialized rollback transactions.
// Callers must supply fixed, trusted paths and an apply function that restores
// the corresponding runtime state. It is not a general backup/restore API.
package safeapply

import (
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"time"

	"golang.org/x/sys/unix"
)

var ErrPending = errors.New("a safe-apply transaction is already pending")
var ErrNoPending = errors.New("no safe-apply transaction is pending")
var ErrExpired = errors.New("safe-apply transaction has expired")
var ErrToken = errors.New("invalid safe-apply transaction token")

type fileSnapshot struct {
	Path   string      `json:"path"`
	Data   []byte      `json:"data,omitempty"`
	Mode   os.FileMode `json:"mode,omitempty"`
	Exists bool        `json:"exists"`
}
type transaction struct {
	Token    string         `json:"token"`
	Deadline time.Time      `json:"deadline"`
	Files    []fileSnapshot `json:"files"`
}
type Status struct {
	Pending  bool      `json:"pending"`
	Deadline time.Time `json:"deadline,omitempty"`
}

// Manager serializes operations across processes by locking a file in dir.
// apply is invoked after snapshot files are restored and must return an error
// when runtime recovery fails. The transaction remains pending on any error.
type Manager struct {
	dir   string
	paths []string
	apply func() error
	now   func() time.Time
}

func New(dir string, paths []string, apply func() error) *Manager {
	return &Manager{dir: dir, paths: append([]string(nil), paths...), apply: apply, now: time.Now}
}
func (m *Manager) locked(fn func() error) error {
	if err := os.MkdirAll(m.dir, 0700); err != nil {
		return err
	}
	f, err := os.OpenFile(filepath.Join(m.dir, "lock"), os.O_CREATE|os.O_RDWR, 0600)
	if err != nil {
		return err
	}
	defer f.Close()
	if err = unix.Flock(int(f.Fd()), unix.LOCK_EX); err != nil {
		return err
	}
	defer unix.Flock(int(f.Fd()), unix.LOCK_UN)
	return fn()
}
func (m *Manager) read() (*transaction, error) {
	data, err := os.ReadFile(filepath.Join(m.dir, "pending.json"))
	if os.IsNotExist(err) {
		return nil, ErrNoPending
	}
	if err != nil {
		return nil, err
	}
	var tx transaction
	if err = json.Unmarshal(data, &tx); err != nil {
		return nil, fmt.Errorf("invalid transaction: %w", err)
	}
	if tx.Token == "" || tx.Deadline.IsZero() || len(tx.Files) != len(m.paths) {
		return nil, errors.New("invalid transaction state")
	}
	for i, p := range m.paths {
		if tx.Files[i].Path != p {
			return nil, errors.New("transaction paths do not match configured paths")
		}
	}
	return &tx, nil
}
func atomicWrite(path string, data []byte, mode os.FileMode) error {
	dir := filepath.Dir(path)
	if err := os.MkdirAll(dir, 0700); err != nil {
		return err
	}
	f, err := os.CreateTemp(dir, ".safeapply-*")
	if err != nil {
		return err
	}
	name := f.Name()
	defer os.Remove(name)
	if err = f.Chmod(mode.Perm()); err == nil {
		_, err = f.Write(data)
	}
	if err == nil {
		err = f.Sync()
	}
	if closeErr := f.Close(); err == nil {
		err = closeErr
	}
	if err != nil {
		return err
	}
	if err = os.Rename(name, path); err != nil {
		return err
	}
	d, err := os.Open(dir)
	if err != nil {
		return err
	}
	defer d.Close()
	return d.Sync()
}

// Begin snapshots the fixed set of paths before any changes are applied.
func (m *Manager) Begin(duration time.Duration) (string, time.Time, error) {
	var token string
	var deadline time.Time
	err := m.locked(func() error {
		_, err := m.read()
		if err == nil {
			return ErrPending
		}
		if !errors.Is(err, ErrNoPending) {
			return err
		}
		if duration <= 0 || len(m.paths) == 0 || m.apply == nil {
			return errors.New("invalid safe-apply configuration")
		}
		tx := transaction{Deadline: m.now().UTC().Add(duration)}
		random := make([]byte, 32)
		if _, err := rand.Read(random); err != nil {
			return err
		}
		tx.Token = hex.EncodeToString(random)
		for _, path := range m.paths {
			if !filepath.IsAbs(path) || filepath.Clean(path) != path {
				return fmt.Errorf("invalid snapshot path %q", path)
			}
			snap := fileSnapshot{Path: path}
			info, err := os.Lstat(path)
			if err == nil {
				if !info.Mode().IsRegular() {
					return fmt.Errorf("snapshot path is not a regular file: %s", path)
				}
				snap.Exists = true
				snap.Mode = info.Mode().Perm()
				snap.Data, err = os.ReadFile(path)
				if err != nil {
					return err
				}
			}
			if err != nil && !os.IsNotExist(err) {
				return err
			}
			tx.Files = append(tx.Files, snap)
		}
		data, err := json.Marshal(tx)
		if err != nil {
			return err
		}
		if err = atomicWrite(filepath.Join(m.dir, "pending.json"), data, 0600); err != nil {
			return err
		}
		token, deadline = tx.Token, tx.Deadline
		return nil
	})
	return token, deadline, err
}

// Confirm removes the pending transaction only before its deadline.
func (m *Manager) Confirm(token string) error {
	return m.locked(func() error {
		tx, err := m.read()
		if err != nil {
			return err
		}
		if tx.Token != token {
			return ErrToken
		}
		if !m.now().Before(tx.Deadline) {
			return ErrExpired
		}
		return m.remove()
	})
}
func (m *Manager) remove() error {
	if err := os.Remove(filepath.Join(m.dir, "pending.json")); err != nil {
		return err
	}
	d, err := os.Open(m.dir)
	if err != nil {
		return err
	}
	defer d.Close()
	return d.Sync()
}
func (m *Manager) Status() (Status, error) {
	var status Status
	err := m.locked(func() error {
		tx, err := m.read()
		if errors.Is(err, ErrNoPending) {
			return nil
		}
		if err != nil {
			return err
		}
		status = Status{true, tx.Deadline}
		return nil
	})
	return status, err
}

// RecoverExpired is intended to be run by an independent watchdog as well as
// at startup. It is safe to retry after a partial file or runtime failure.
func (m *Manager) RecoverExpired() (bool, error) {
	recovered := false
	err := m.locked(func() error {
		tx, err := m.read()
		if errors.Is(err, ErrNoPending) {
			return nil
		}
		if err != nil {
			return err
		}
		if m.now().Before(tx.Deadline) {
			return nil
		}
		for _, f := range tx.Files {
			if f.Exists {
				if err = atomicWrite(f.Path, f.Data, f.Mode); err != nil {
					return err
				}
			} else if err = os.Remove(f.Path); err != nil && !os.IsNotExist(err) {
				return err
			}
		}
		if m.apply == nil {
			return errors.New("runtime recovery is not configured")
		}
		if err = m.apply(); err != nil {
			return fmt.Errorf("runtime recovery failed: %w", err)
		}
		if err = m.remove(); err != nil {
			return err
		}
		recovered = true
		return nil
	})
	return recovered, err
}