B1: data-migration engine (MigrateAll + MigrateApp) + backup mutual-exclusion
internal/stacks/migrate.go: crash-safe, resumable namespace migration over the controller's /mnt RW mount. Two entry points (whole-namespace + per-app) share one journaled pipeline: validate -> stop -> copy (rsync -a --checksum, additive; conflict- merge walk for non-app content) -> verify -> flip+redeploy (RedeployFromEnv) -> cleanup. CLEANUP (the only destructive step) is gated on all units verified AND all apps redeployed. Single-flight; mutual exclusion with the backup orchestrator (Change 3). Non-hollow tests incl. mutation-proven collision + cleanup-gate companions. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -15,6 +15,7 @@ import (
|
||||
|
||||
"gitea.dooplex.hu/admin/felhom-controller/internal/config"
|
||||
"gitea.dooplex.hu/admin/felhom-controller/internal/crypto"
|
||||
"gitea.dooplex.hu/admin/felhom-controller/internal/settings"
|
||||
)
|
||||
|
||||
// ContainerState represents the current state of a container.
|
||||
@@ -85,6 +86,15 @@ type Manager struct {
|
||||
mu sync.RWMutex
|
||||
encKey []byte // AES-256 key for encrypting sensitive values in app.yaml
|
||||
infraMu sync.Mutex // single-flight guard for EnsureBaseStack (base-infra bring-up/self-heal)
|
||||
|
||||
// Migration engine (B1): single-flight + live job + deps wired via SetMigrationDeps.
|
||||
migrateMu sync.Mutex
|
||||
migrating bool
|
||||
migJob *MigrationJob
|
||||
settings *settings.Settings
|
||||
sysDataPath string
|
||||
backupRunning func() bool // mutual exclusion with the backup orchestrator (Change 3)
|
||||
testSeams *migSeams // nil in production; tests inject fakes
|
||||
}
|
||||
|
||||
// NewManager creates a new stack manager.
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,398 @@
|
||||
package stacks
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"gitea.dooplex.hu/admin/felhom-controller/internal/appbackup"
|
||||
"gitea.dooplex.hu/admin/felhom-controller/internal/config"
|
||||
"gitea.dooplex.hu/admin/felhom-controller/internal/settings"
|
||||
)
|
||||
|
||||
// --- pure helpers: siblingName / lowest-free-N / merge walk ---
|
||||
|
||||
func TestSiblingName(t *testing.T) {
|
||||
cases := map[string]string{
|
||||
"/d/foo.txt": "/d/foo(1).txt",
|
||||
"/d/backup.tar.gz": "/d/backup.tar(1).gz", // single (last) extension
|
||||
"/d/README": "/d/README(1)",
|
||||
}
|
||||
for in, want := range cases {
|
||||
if got := siblingName(filepath.FromSlash(in), 1); got != filepath.FromSlash(want) {
|
||||
t.Errorf("siblingName(%q) = %q, want %q", in, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func writeFile(t *testing.T, path, content string) {
|
||||
t.Helper()
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(path, []byte(content), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func readFile(t *testing.T, path string) string {
|
||||
t.Helper()
|
||||
b, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatalf("read %s: %v", path, err)
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
|
||||
// TestMergeWalk_Conflicts covers differ→(1), identical→dedup, and re-run idempotency (no (1)(1)).
|
||||
func TestMergeWalk_Conflicts(t *testing.T) {
|
||||
lg := log.New(os.Stderr, "", 0)
|
||||
src := t.TempDir()
|
||||
dst := t.TempDir()
|
||||
|
||||
// identical file at same rel path → should dedup (no sibling)
|
||||
writeFile(t, filepath.Join(src, "same.txt"), "IDENTICAL")
|
||||
writeFile(t, filepath.Join(dst, "same.txt"), "IDENTICAL")
|
||||
// differing file at same rel path → should land as differ(1).txt, original untouched
|
||||
writeFile(t, filepath.Join(src, "differ.txt"), "SRC")
|
||||
writeFile(t, filepath.Join(dst, "differ.txt"), "DST")
|
||||
// fresh file absent at dst → copied as-is
|
||||
writeFile(t, filepath.Join(src, "sub", "new.txt"), "NEW")
|
||||
|
||||
if err := walkMerge(lg, src, dst, nil, false, nil); err != nil {
|
||||
t.Fatalf("walkMerge: %v", err)
|
||||
}
|
||||
if got := readFile(t, filepath.Join(dst, "same.txt")); got != "IDENTICAL" {
|
||||
t.Errorf("identical file mutated: %q", got)
|
||||
}
|
||||
if pathExists(filepath.Join(dst, "same(1).txt")) {
|
||||
t.Errorf("identical file should NOT create a (1) sibling")
|
||||
}
|
||||
if got := readFile(t, filepath.Join(dst, "differ.txt")); got != "DST" {
|
||||
t.Errorf("existing target overwritten: %q (must never overwrite)", got)
|
||||
}
|
||||
if got := readFile(t, filepath.Join(dst, "differ(1).txt")); got != "SRC" {
|
||||
t.Errorf("differing file should land as differ(1).txt, got %q", got)
|
||||
}
|
||||
if got := readFile(t, filepath.Join(dst, "sub", "new.txt")); got != "NEW" {
|
||||
t.Errorf("fresh file = %q", got)
|
||||
}
|
||||
|
||||
// Re-run must be idempotent: SRC now equals differ(1).txt, so no differ(1)(1).txt accrues.
|
||||
if err := walkMerge(lg, src, dst, nil, false, nil); err != nil {
|
||||
t.Fatalf("walkMerge re-run: %v", err)
|
||||
}
|
||||
if pathExists(filepath.Join(dst, "differ(1)(1).txt")) {
|
||||
t.Errorf("re-run proliferated a (1)(1) sibling — NOT idempotent")
|
||||
}
|
||||
|
||||
// A THIRD distinct content for the same rel path → lowest-free N = (2), not (1)(1).
|
||||
writeFile(t, filepath.Join(src, "differ.txt"), "THIRD")
|
||||
if err := walkMerge(lg, src, dst, nil, false, nil); err != nil {
|
||||
t.Fatalf("walkMerge third: %v", err)
|
||||
}
|
||||
if got := readFile(t, filepath.Join(dst, "differ(2).txt")); got != "THIRD" {
|
||||
t.Errorf("third distinct content should be differ(2).txt, got %q", got)
|
||||
}
|
||||
if pathExists(filepath.Join(dst, "differ(1)(1).txt")) {
|
||||
t.Errorf("must not create differ(1)(1).txt")
|
||||
}
|
||||
}
|
||||
|
||||
// TestMergeWalk_AssertOnly: assert passes when every source file has an identical counterpart, and
|
||||
// fails when one does not.
|
||||
func TestMergeWalk_AssertOnly(t *testing.T) {
|
||||
lg := log.New(os.Stderr, "", 0)
|
||||
src := t.TempDir()
|
||||
dst := t.TempDir()
|
||||
writeFile(t, filepath.Join(src, "a.txt"), "A")
|
||||
writeFile(t, filepath.Join(dst, "a.txt"), "A")
|
||||
if err := walkMerge(lg, src, dst, nil, true, nil); err != nil {
|
||||
t.Errorf("assert should pass when all counterparts identical: %v", err)
|
||||
}
|
||||
// add a source file with no counterpart at dst
|
||||
writeFile(t, filepath.Join(src, "missing.txt"), "M")
|
||||
if err := walkMerge(lg, src, dst, nil, true, nil); err == nil {
|
||||
t.Errorf("assert must FAIL when a source file has no identical counterpart")
|
||||
}
|
||||
// a (N)-sibling counterpart also satisfies assert
|
||||
writeFile(t, filepath.Join(dst, "missing(1).txt"), "M")
|
||||
if err := walkMerge(lg, src, dst, nil, true, nil); err != nil {
|
||||
t.Errorf("assert should accept a (N)-sibling counterpart: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestMergeWalk_SkipDirs confirms app appdata dirs are pruned (handled by rsync).
|
||||
func TestMergeWalk_SkipDirs(t *testing.T) {
|
||||
lg := log.New(os.Stderr, "", 0)
|
||||
src := t.TempDir()
|
||||
dst := t.TempDir()
|
||||
appdata := filepath.Join(src, "appdata", "romm")
|
||||
writeFile(t, filepath.Join(appdata, "rom.bin"), "ROM")
|
||||
writeFile(t, filepath.Join(src, "media", "movie.mkv"), "VID")
|
||||
skip := map[string]bool{filepath.Clean(appdata): true}
|
||||
if err := walkMerge(lg, src, dst, skip, false, nil); err != nil {
|
||||
t.Fatalf("walkMerge: %v", err)
|
||||
}
|
||||
if pathExists(filepath.Join(dst, "appdata", "romm", "rom.bin")) {
|
||||
t.Errorf("skipped appdata dir should NOT be merged")
|
||||
}
|
||||
if !pathExists(filepath.Join(dst, "media", "movie.mkv")) {
|
||||
t.Errorf("non-app content should be merged")
|
||||
}
|
||||
}
|
||||
|
||||
// --- orchestration: phase machine via injected seams + real temp-dir FS ---
|
||||
|
||||
func newMigManager(t *testing.T, target string) *Manager {
|
||||
t.Helper()
|
||||
lg := log.New(os.Stderr, "", 0)
|
||||
cfg := &config.Config{}
|
||||
cfg.Paths.DataDir = t.TempDir()
|
||||
cfg.Paths.SystemDataPath = "/mnt/sys_drive"
|
||||
m := &Manager{cfg: cfg, logger: lg, stacks: map[string]*Stack{}, sysDataPath: cfg.Paths.SystemDataPath}
|
||||
sett, err := settings.Load(filepath.Join(t.TempDir(), "settings.json"), lg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if target != "" {
|
||||
_ = sett.AddStoragePath(settings.StoragePath{Path: target, Schedulable: true})
|
||||
}
|
||||
m.settings = sett
|
||||
return m
|
||||
}
|
||||
|
||||
// runJob runs a manually-built job synchronously to a terminal phase.
|
||||
func (m *Manager) runJobSync(j *MigrationJob) {
|
||||
_ = m.acquireMigrating()
|
||||
m.setJob(j)
|
||||
m.runMigration(context.Background(), j)
|
||||
}
|
||||
|
||||
func newAllJob(srcNS, dstNS, source, target string, apps ...string) *MigrationJob {
|
||||
j := &MigrationJob{
|
||||
Scope: "all", Phase: PhaseStop, Source: source, Target: target,
|
||||
SourceNS: srcNS, TargetNS: dstNS, Apps: apps, Units: map[string]*MigUnit{},
|
||||
}
|
||||
for _, a := range apps {
|
||||
j.Units[a] = &MigUnit{App: a, State: UnitPending}
|
||||
}
|
||||
j.Units[nonAppUnit] = &MigUnit{App: nonAppUnit, State: UnitPending}
|
||||
return j
|
||||
}
|
||||
|
||||
// TestMigration_CleanupOnlyAfterRedeploy: a redeploy failure aborts BEFORE cleanup; source intact.
|
||||
func TestMigration_CleanupOnlyAfterRedeploy(t *testing.T) {
|
||||
m := newMigManager(t, "/mnt/target")
|
||||
srcNS := t.TempDir()
|
||||
dstNS := t.TempDir()
|
||||
// real source data so we can assert it survives an abort
|
||||
writeFile(t, filepath.Join(appbackup.AppDataDir(srcNS, "romm"), "rom.bin"), "ROM")
|
||||
|
||||
// seams: copy/verify succeed (no-op), redeploy FAILS for romm.
|
||||
m.testSeams = &migSeams{
|
||||
stop: func(string) error { return nil },
|
||||
copy: func(_ context.Context, _, _ string, _ func(int64)) error { return nil },
|
||||
verify: func(_ context.Context, _, _ string) error { return nil },
|
||||
flipRedeploy: func(name, _ string) error {
|
||||
return os.ErrPermission // simulate app failing to come up
|
||||
},
|
||||
}
|
||||
j := newAllJob(srcNS, dstNS, "/mnt/source", "/mnt/target", "romm")
|
||||
m.runJobSync(j)
|
||||
|
||||
if j.Phase != PhaseAborted {
|
||||
t.Fatalf("phase = %s, want aborted", j.Phase)
|
||||
}
|
||||
if !pathExists(appbackup.AppDataDir(srcNS, "romm")) {
|
||||
t.Errorf("SOURCE was deleted despite redeploy failure — cleanup must not run before redeploy")
|
||||
}
|
||||
}
|
||||
|
||||
// TestMigration_HappyPathCleansSource: full success removes the source only after verify+redeploy.
|
||||
func TestMigration_HappyPathCleansSource(t *testing.T) {
|
||||
m := newMigManager(t, "/mnt/target")
|
||||
srcNS := t.TempDir()
|
||||
dstNS := t.TempDir()
|
||||
writeFile(t, filepath.Join(appbackup.AppDataDir(srcNS, "romm"), "rom.bin"), "ROM")
|
||||
writeFile(t, filepath.Join(srcNS, "media", "movie.mkv"), "VID") // non-app content
|
||||
|
||||
flips := []string{}
|
||||
m.testSeams = &migSeams{
|
||||
stop: func(string) error { return nil },
|
||||
copy: func(_ context.Context, _, _ string, _ func(int64)) error { return nil },
|
||||
verify: func(_ context.Context, _, _ string) error { return nil },
|
||||
flipRedeploy: func(name, target string) error {
|
||||
flips = append(flips, name+"→"+target)
|
||||
return nil
|
||||
},
|
||||
}
|
||||
j := newAllJob(srcNS, dstNS, "/mnt/source", "/mnt/target", "romm")
|
||||
m.runJobSync(j)
|
||||
|
||||
if j.Phase != PhaseDone {
|
||||
t.Fatalf("phase = %s (err=%s), want done", j.Phase, j.Error)
|
||||
}
|
||||
if pathExists(appbackup.AppDataDir(srcNS, "romm")) {
|
||||
t.Errorf("source appdata should be removed after success")
|
||||
}
|
||||
if pathExists(filepath.Join(srcNS, "media")) {
|
||||
t.Errorf("source non-app content should be removed after success")
|
||||
}
|
||||
if len(flips) != 1 || flips[0] != "romm→/mnt/target" {
|
||||
t.Errorf("flipRedeploy calls = %v", flips)
|
||||
}
|
||||
}
|
||||
|
||||
// TestMigration_VerifyCatchesCorruption: a verify failure aborts; source intact, no cleanup.
|
||||
func TestMigration_VerifyCatchesCorruption(t *testing.T) {
|
||||
m := newMigManager(t, "/mnt/target")
|
||||
srcNS := t.TempDir()
|
||||
dstNS := t.TempDir()
|
||||
writeFile(t, filepath.Join(appbackup.AppDataDir(srcNS, "romm"), "rom.bin"), "ROM")
|
||||
|
||||
cleanupReached := false
|
||||
m.testSeams = &migSeams{
|
||||
stop: func(string) error { return nil },
|
||||
copy: func(_ context.Context, _, _ string, _ func(int64)) error { return nil },
|
||||
verify: func(_ context.Context, _, _ string) error { return os.ErrInvalid }, // corruption
|
||||
flipRedeploy: func(string, string) error {
|
||||
cleanupReached = true // would only be reachable past verify
|
||||
return nil
|
||||
},
|
||||
}
|
||||
j := newAllJob(srcNS, dstNS, "/mnt/source", "/mnt/target", "romm")
|
||||
m.runJobSync(j)
|
||||
|
||||
if j.Phase != PhaseAborted {
|
||||
t.Fatalf("phase = %s, want aborted", j.Phase)
|
||||
}
|
||||
if cleanupReached {
|
||||
t.Errorf("flip/redeploy ran despite a verify failure")
|
||||
}
|
||||
if !pathExists(appbackup.AppDataDir(srcNS, "romm")) {
|
||||
t.Errorf("source deleted despite verify failure")
|
||||
}
|
||||
}
|
||||
|
||||
// TestMigration_Resume: a journal at Phase=verify with units already copied must NOT re-copy.
|
||||
func TestMigration_Resume(t *testing.T) {
|
||||
m := newMigManager(t, "/mnt/target")
|
||||
srcNS := t.TempDir()
|
||||
dstNS := t.TempDir()
|
||||
writeFile(t, filepath.Join(appbackup.AppDataDir(srcNS, "romm"), "rom.bin"), "ROM")
|
||||
|
||||
copied := 0
|
||||
m.testSeams = &migSeams{
|
||||
stop: func(string) error { return nil },
|
||||
copy: func(_ context.Context, _, _ string, _ func(int64)) error {
|
||||
copied++ // must NOT be called on resume from verify
|
||||
return nil
|
||||
},
|
||||
verify: func(_ context.Context, _, _ string) error { return nil },
|
||||
flipRedeploy: func(string, string) error { return nil },
|
||||
}
|
||||
j := newAllJob(srcNS, dstNS, "/mnt/source", "/mnt/target", "romm")
|
||||
j.Phase = PhaseVerify
|
||||
j.Units["romm"].State = UnitCopied
|
||||
j.Units[nonAppUnit].State = UnitCopied
|
||||
m.runJobSync(j)
|
||||
|
||||
if j.Phase != PhaseDone {
|
||||
t.Fatalf("phase = %s (err=%s), want done", j.Phase, j.Error)
|
||||
}
|
||||
if copied != 0 {
|
||||
t.Errorf("resume re-copied %d subtree(s); should re-copy none", copied)
|
||||
}
|
||||
}
|
||||
|
||||
// TestMigValidate_CollisionRefused: an existing app dir at target refuses, naming the app. (Companion:
|
||||
// without the collision guard this returns nil, so the test would fail — non-hollow.)
|
||||
func TestMigValidate_CollisionRefused(t *testing.T) {
|
||||
m := newMigManager(t, "/mnt/target")
|
||||
dstNS := t.TempDir()
|
||||
m.testSeams = &migSeams{} // skip the rsync-binary check
|
||||
// pre-existing app dir at the target namespace
|
||||
writeFile(t, filepath.Join(appbackup.AppDataDir(dstNS, "romm"), "x"), "x")
|
||||
|
||||
j := &MigrationJob{Scope: "all", Source: "/mnt/source", Target: "/mnt/target",
|
||||
SourceNS: t.TempDir(), TargetNS: dstNS, Apps: []string{"romm"}, Units: map[string]*MigUnit{}}
|
||||
if err := m.migValidate(j); err == nil {
|
||||
t.Fatalf("expected collision refusal")
|
||||
} else if !contains(err.Error(), "romm") {
|
||||
t.Errorf("collision error must name the app, got %q", err)
|
||||
}
|
||||
|
||||
// No collision (fresh target) → validate passes.
|
||||
j.TargetNS = t.TempDir()
|
||||
if err := m.migValidate(j); err != nil {
|
||||
t.Errorf("validate should pass without collision: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestMigValidate_BackupExclusion: refuse to start while a backup is running (Change 3).
|
||||
func TestMigValidate_BackupExclusion(t *testing.T) {
|
||||
m := newMigManager(t, "/mnt/target")
|
||||
m.testSeams = &migSeams{}
|
||||
m.backupRunning = func() bool { return true }
|
||||
j := &MigrationJob{Scope: "all", Source: "/mnt/source", Target: "/mnt/target",
|
||||
SourceNS: t.TempDir(), TargetNS: t.TempDir(), Apps: nil, Units: map[string]*MigUnit{}}
|
||||
if err := m.migValidate(j); err == nil || !contains(err.Error(), "mentés") {
|
||||
t.Errorf("expected backup-in-progress refusal, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestMigrate_SingleFlight: a second Start while one is active is refused.
|
||||
func TestMigrate_SingleFlight(t *testing.T) {
|
||||
m := newMigManager(t, "/mnt/target")
|
||||
_ = m.acquireMigrating() // simulate an active migration
|
||||
if _, err := m.MigrateApp(context.Background(), "romm", "/mnt/target"); err == nil || !contains(err.Error(), "folyamatban") {
|
||||
t.Errorf("second migration should be refused, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAppSourceNS_SSDToDrive: an app with no HDD_PATH resolves to the system/SSD namespace.
|
||||
func TestAppSourceNS_SSDToDrive(t *testing.T) {
|
||||
m := newMigManager(t, "")
|
||||
// Use filepath.Clean on expectations so the test holds on both Linux (the deploy target) and
|
||||
// Windows (dev), where Clean uses backslashes.
|
||||
src, ns := m.appSourceNS(&AppConfig{Env: map[string]string{}})
|
||||
if src != filepath.Clean("/mnt/sys_drive") {
|
||||
t.Errorf("SSD app source = %q, want %q", src, filepath.Clean("/mnt/sys_drive"))
|
||||
}
|
||||
wantNS := appbackup.NamespaceRoot("/mnt/sys_drive", false) // SSD → felhom-data subdir
|
||||
if ns != wantNS {
|
||||
t.Errorf("SSD app ns = %q, want %q", ns, wantNS)
|
||||
}
|
||||
// a drive-resident app uses its mount root as the namespace
|
||||
src2, ns2 := m.appSourceNS(&AppConfig{Env: map[string]string{"HDD_PATH": "/mnt/felhom-usb"}})
|
||||
if src2 != filepath.Clean("/mnt/felhom-usb") || ns2 != filepath.Clean("/mnt/felhom-usb") {
|
||||
t.Errorf("drive app resolved to src=%q ns=%q", src2, ns2)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCleanupGate: the cleanup gate refuses until all units verified AND all apps redeployed.
|
||||
func TestCleanupGate(t *testing.T) {
|
||||
m := newMigManager(t, "")
|
||||
j := &MigrationJob{Scope: "all", Apps: []string{"romm"}, Units: map[string]*MigUnit{
|
||||
"romm": {App: "romm", State: UnitVerified}, // verified but NOT redeployed
|
||||
nonAppUnit: {App: nonAppUnit, State: UnitVerified},
|
||||
}}
|
||||
if err := m.migCleanupAllowed(j); err == nil {
|
||||
t.Errorf("gate must refuse cleanup when an app is not redeployed")
|
||||
}
|
||||
j.Units["romm"].State = UnitRedeployed
|
||||
if err := m.migCleanupAllowed(j); err != nil {
|
||||
t.Errorf("gate should allow cleanup once all verified+redeployed: %v", err)
|
||||
}
|
||||
// non-app unit only verified is required; if it regressed below verified, refuse
|
||||
j.Units[nonAppUnit].State = UnitCopied
|
||||
if err := m.migCleanupAllowed(j); err == nil {
|
||||
t.Errorf("gate must refuse when non-app content is not verified")
|
||||
}
|
||||
}
|
||||
|
||||
func contains(s, sub string) bool { return strings.Contains(s, sub) }
|
||||
Reference in New Issue
Block a user