diff --git a/controller/cmd/controller/main.go b/controller/cmd/controller/main.go index 9d8f0df..d3a6e8a 100644 --- a/controller/cmd/controller/main.go +++ b/controller/cmd/controller/main.go @@ -226,6 +226,13 @@ func main() { backupMgr.SetVersion(Version) } + // --- Wire the data-migration engine (B1) + backup↔migration mutual exclusion (Change 3) --- + stackMgr.SetMigrationDeps(sett, func() bool { return backupMgr != nil && backupMgr.IsRunning() }) + if backupMgr != nil { + backupMgr.SetMigrationRunningCheck(stackMgr.IsMigrating) + } + stackMgr.RecoverMigration(ctx) + // --- Initialize alert manager --- alertMgr := web.NewAlertManager(logger) diff --git a/controller/internal/backup/backup.go b/controller/internal/backup/backup.go index 6c0b8b1..ff4a662 100644 --- a/controller/internal/backup/backup.go +++ b/controller/internal/backup/backup.go @@ -36,6 +36,11 @@ type Manager struct { discoverDBs func(ctx context.Context) ([]DiscoveredDB, error) importDBDump func(ctx context.Context, db DiscoveredDB, dumpPath string) error + // migrationRunning, if set, reports whether a data migration is in progress. The scheduled + // backup paths skip when it returns true (Change 3 — backup ↔ migration mutual exclusion), so a + // nightly dump/Tier-2 can't race a migration copy/cleanup on the same drive. + migrationRunning func() bool + mu sync.Mutex lastDBDump *DBDumpStatus running bool @@ -158,8 +163,23 @@ func (m *Manager) groupStacksByDrive() map[string][]StackSummary { return result } +// SetMigrationRunningCheck wires the mutual-exclusion guard (Change 3): when fn() reports a +// migration is active, the scheduled backup paths skip rather than race it. +func (m *Manager) SetMigrationRunningCheck(fn func() bool) { + m.migrationRunning = fn +} + +// migrationActive reports whether a migration is in progress (false when no check is wired). +func (m *Manager) migrationActive() bool { + return m.migrationRunning != nil && m.migrationRunning() +} + // RunDBDumps discovers and dumps all databases to per-drive, per-app paths. func (m *Manager) RunDBDumps(ctx context.Context) error { + if m.migrationActive() { + m.logger.Printf("[INFO] [backup] DB dump kihagyva: migráció folyamatban") + return nil + } if err := m.acquireRunning(); err != nil { return err } diff --git a/controller/internal/backup/migration_exclusion_test.go b/controller/internal/backup/migration_exclusion_test.go new file mode 100644 index 0000000..c760cc5 --- /dev/null +++ b/controller/internal/backup/migration_exclusion_test.go @@ -0,0 +1,38 @@ +package backup + +import ( + "context" + "io" + "log" + "testing" + + "gitea.dooplex.hu/admin/felhom-controller/internal/config" +) + +// TestRunDBDumps_SkippedWhileMigrating asserts the scheduled DB-dump path SKIPS when a migration is +// active (Change 3 — backup ↔ migration mutual exclusion), rather than racing the migration. The +// companion: without the migrationActive guard, RunDBDumps would proceed (and a non-nil migration +// check would not matter) — this test FAILS because lastDBDump would be set / discovery attempted. +func TestRunDBDumps_SkippedWhileMigrating(t *testing.T) { + cfg := &config.Config{} + cfg.Paths.SystemDataPath = "/mnt/sys_drive" + m := NewManager(cfg, nil, log.New(io.Discard, "", 0)) + m.SetMigrationRunningCheck(func() bool { return true }) + + if err := m.RunDBDumps(context.Background()); err != nil { + t.Fatalf("RunDBDumps should skip cleanly, got %v", err) + } + // Skipped before runDBDumpsInternal → no status recorded and the running flag never taken. + if m.lastDBDump != nil { + t.Errorf("DB dump ran despite an active migration (lastDBDump set)") + } + if m.IsRunning() { + t.Errorf("running flag left set after a skipped dump") + } + + // With no migration active, the guard does not block (it proceeds into discovery). + m.SetMigrationRunningCheck(func() bool { return false }) + if m.migrationActive() { + t.Errorf("migrationActive should be false when the check returns false") + } +} diff --git a/controller/internal/backup/tier2.go b/controller/internal/backup/tier2.go index b443b37..fe6482a 100644 --- a/controller/internal/backup/tier2.go +++ b/controller/internal/backup/tier2.go @@ -200,6 +200,10 @@ func (m *Manager) RunAllTier2() { if m.stackProvider == nil { return } + if m.migrationActive() { + m.logger.Printf("[INFO] [backup] Tier 2 kihagyva: migráció folyamatban") + return + } var n int for _, stack := range m.stackProvider.ListDeployedStacks() { if m.stackProvider.GetStackHDDPath(stack.Name) == "" { diff --git a/controller/internal/stacks/manager.go b/controller/internal/stacks/manager.go index 087922a..38944a4 100644 --- a/controller/internal/stacks/manager.go +++ b/controller/internal/stacks/manager.go @@ -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. diff --git a/controller/internal/stacks/migrate.go b/controller/internal/stacks/migrate.go new file mode 100644 index 0000000..98e0cf7 --- /dev/null +++ b/controller/internal/stacks/migrate.go @@ -0,0 +1,1028 @@ +package stacks + +// Data-migration engine (TASK B1). Moves an app-data felhom-data NAMESPACE from one storage drive +// to another, in-process over the controller's /mnt:/mnt:rslave RW mount. Crash-safe + resumable via +// a single durable journal (/migration.json); one migration (app OR all) at a time. +// +// Two entry points share ONE pipeline: +// - MigrateAll(source, target): the whole namespace — every app on the source drive + the +// conflict-merge walk for non-app/customer content. Used by the decommission flow. +// - MigrateApp(app, target): one app's data subtree only (appdata + its recovery unit). Handles +// drive→drive AND SSD→drive (an SSD-resident app gaining HDD_PATH for the first time). +// +// Move primitive = additive `rsync -a --checksum` (NO --delete — never destroys the target). The +// non-app content uses a custom conflict-merge walk (skip-identical / rename-on-differ to a +// `(N)` sibling; never overwrite). The ONLY destructive step is CLEANUP, which removes the +// SOURCE and runs ONLY after every unit verified AND every app redeployed. + +import ( + "bufio" + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "io/fs" + "log" + "os" + "os/exec" + "path/filepath" + "sort" + "strconv" + "strings" + "time" + + "gitea.dooplex.hu/admin/felhom-controller/internal/appbackup" + "gitea.dooplex.hu/admin/felhom-controller/internal/settings" + "gitea.dooplex.hu/admin/felhom-controller/internal/system" +) + +const ( + migrationJournalFile = "migration.json" + migrateCopyTimeout = 6 * time.Hour + migrateVerifyTimeout = 60 * time.Minute + nonAppUnit = "" // map key for the non-app-content unit (MigrateAll only) +) + +// MigrationState is the per-unit progress state, persisted in the journal. +type MigrationState string + +const ( + UnitPending MigrationState = "pending" + UnitCopied MigrationState = "copied" + UnitVerified MigrationState = "verified" + UnitFlipped MigrationState = "flipped" + UnitRedeployed MigrationState = "redeployed" + UnitCleaned MigrationState = "cleaned" +) + +func stateRank(s MigrationState) int { + switch s { + case UnitCopied: + return 1 + case UnitVerified: + return 2 + case UnitFlipped: + return 3 + case UnitRedeployed: + return 4 + case UnitCleaned: + return 5 + default: + return 0 // pending / unknown + } +} + +// MigrationPhase is the pipeline phase, persisted so RecoverMigration resumes from the right step. +type MigrationPhase string + +const ( + PhaseStop MigrationPhase = "stop" + PhaseCopy MigrationPhase = "copy" + PhaseVerify MigrationPhase = "verify" + PhaseFlip MigrationPhase = "flip" + PhaseRedeploy MigrationPhase = "redeploy" + PhaseCleanup MigrationPhase = "cleanup" + PhaseDone MigrationPhase = "done" + PhaseAborted MigrationPhase = "aborted" +) + +// MigUnit is one journal unit: an affected app, or the non-app-content sentinel (""). +type MigUnit struct { + App string `json:"app"` + State MigrationState `json:"state"` + Error string `json:"error,omitempty"` +} + +// MigrationJob is BOTH the live status (returned by MigrationStatus, polled by the UI) and the +// durable journal (written to /migration.json atomically at every transition). +type MigrationJob struct { + ID string `json:"id"` + Scope string `json:"scope"` // "app" | "all" + Phase MigrationPhase `json:"phase"` + Source string `json:"source"` // source drive path (the HDD_PATH value, or the SSD path) + Target string `json:"target"` // target drive path + SourceNS string `json:"source_ns"` // resolved felhom-data namespace root on source + TargetNS string `json:"target_ns"` + Apps []string `json:"apps"` // affected app names, stable order + Units map[string]*MigUnit `json:"units"` + CurrentApp string `json:"current_app,omitempty"` + BytesTotal int64 `json:"bytes_total"` + BytesDone int64 `json:"bytes_done"` + Error string `json:"error,omitempty"` + StartedAt time.Time `json:"started_at"` + UpdatedAt time.Time `json:"updated_at"` + FinishedAt time.Time `json:"finished_at,omitempty"` +} + +func (j *MigrationJob) clone() *MigrationJob { + cp := *j + cp.Apps = append([]string(nil), j.Apps...) + cp.Units = make(map[string]*MigUnit, len(j.Units)) + for k, u := range j.Units { + uc := *u + cp.Units[k] = &uc + } + return &cp +} + +// migSeams lets tests drive the pipeline without docker/rsync. Nil = use the real implementations. +type migSeams struct { + copy func(ctx context.Context, src, dst string, onBytes func(int64)) error + verify func(ctx context.Context, src, dst string) error + stop func(name string) error + flipRedeploy func(name, target string) error +} + +// SetMigrationDeps wires the registry + the backup-running check (mutual exclusion, Change 3). +// Call once from main after SetEncryptionKey, before RecoverMigration. +func (m *Manager) SetMigrationDeps(sett *settings.Settings, backupRunning func() bool) { + m.mu.Lock() + defer m.mu.Unlock() + m.settings = sett + m.backupRunning = backupRunning + m.sysDataPath = m.cfg.Paths.SystemDataPath +} + +// IsMigrating reports whether a migration is in progress (used by the backup orchestrator's guard). +func (m *Manager) IsMigrating() bool { + m.migrateMu.Lock() + defer m.migrateMu.Unlock() + return m.migrating +} + +func (m *Manager) acquireMigrating() error { + m.migrateMu.Lock() + defer m.migrateMu.Unlock() + if m.migrating { + return fmt.Errorf("migráció már folyamatban") + } + m.migrating = true + return nil +} + +func (m *Manager) releaseMigrating() { + m.migrateMu.Lock() + m.migrating = false + m.migrateMu.Unlock() +} + +func (m *Manager) setJob(j *MigrationJob) { + m.migrateMu.Lock() + m.migJob = j + m.migrateMu.Unlock() +} + +// MigrationStatus returns a deep copy of the live job (nil when idle). +func (m *Manager) MigrationStatus() *MigrationJob { + m.migrateMu.Lock() + defer m.migrateMu.Unlock() + if m.migJob == nil { + return nil + } + return m.migJob.clone() +} + +// MigrateAll moves the whole namespace off sourcePath onto targetPath. +func (m *Manager) MigrateAll(ctx context.Context, sourcePath, targetPath string) (string, error) { + return m.startMigration("all", sourcePath, "", targetPath) +} + +// MigrateApp moves a single app's data subtree onto targetPath. +func (m *Manager) MigrateApp(ctx context.Context, appName, targetPath string) (string, error) { + return m.startMigration("app", "", appName, targetPath) +} + +func (m *Manager) startMigration(scope, sourcePath, appName, targetPath string) (string, error) { + if err := m.acquireMigrating(); err != nil { + return "", err + } + launched := false + defer func() { + if !launched { + m.releaseMigrating() + } + }() + + j := &MigrationJob{ + Scope: scope, + Target: filepath.Clean(targetPath), + Units: map[string]*MigUnit{}, + StartedAt: time.Now().UTC(), + } + j.ID = "mig-" + j.StartedAt.Format("20060102-150405") + + // Resolve affected apps + the source namespace. + var apps []string + switch scope { + case "all": + j.Source = filepath.Clean(sourcePath) + j.SourceNS = appbackup.NamespaceRoot(j.Source, m.inGuest(j.Source)) + apps = append(apps, m.appsOnDrive(j.Source)...) + j.Units[nonAppUnit] = &MigUnit{App: nonAppUnit, State: UnitPending} + case "app": + cfg := m.LoadAppConfigByName(appName) + if cfg == nil { + return "", fmt.Errorf("alkalmazás nem található: %s", appName) + } + j.Source, j.SourceNS = m.appSourceNS(cfg) + apps = []string{appName} + default: + return "", fmt.Errorf("ismeretlen migrációs hatókör: %s", scope) + } + + j.TargetNS = appbackup.NamespaceRoot(j.Target, m.inGuest(j.Target)) + sort.Strings(apps) + j.Apps = apps + for _, a := range apps { + j.Units[a] = &MigUnit{App: a, State: UnitPending} + } + + // VALIDATE (synchronous — refusals reach the caller and change nothing). + if err := m.migValidate(j); err != nil { + return "", err + } + + j.Phase = PhaseStop + j.UpdatedAt = time.Now().UTC() + m.setJob(j) + if err := m.writeJournal(j); err != nil { + return "", fmt.Errorf("migrációs napló írása: %w", err) + } + + launched = true + go m.runMigration(context.Background(), j) + return j.ID, nil +} + +// inGuest reports whether drivePath is a user drive (its in-guest mount IS the felhom-data namespace +// root) vs the system/SSD path (which holds a felhom-data SUBDIR). Compares cleaned paths so the +// decision is stable regardless of slash style. +func (m *Manager) inGuest(drivePath string) bool { + return filepath.Clean(drivePath) != filepath.Clean(m.sysDataPath) +} + +// appSourceNS resolves an app's current source drive path + felhom-data namespace root from its +// config. An app with no HDD_PATH lives on the system/SSD path (the SSD→drive case). +func (m *Manager) appSourceNS(cfg *AppConfig) (src, ns string) { + src = cfg.Env["HDD_PATH"] + if src == "" { + src = m.sysDataPath + } + src = filepath.Clean(src) + return src, appbackup.NamespaceRoot(src, m.inGuest(src)) +} + +// appsOnDrive returns the names of deployed apps whose HDD_PATH equals sourcePath. +func (m *Manager) appsOnDrive(sourcePath string) []string { + var out []string + for _, st := range m.GetStacks() { + if !st.Deployed { + continue + } + cfg := m.LoadAppConfigByName(st.Name) + if cfg != nil && cfg.Env["HDD_PATH"] == sourcePath { + out = append(out, st.Name) + } + } + return out +} + +// migValidate runs the pre-flight checks. Any failure aborts before any side-effect. +func (m *Manager) migValidate(j *MigrationJob) error { + if m.backupRunning != nil && m.backupRunning() { + return fmt.Errorf("biztonsági mentés folyamatban, próbáld újra") + } + if m.testSeams == nil { // real rsync is only required when not using injected copy/verify seams + if _, err := exec.LookPath("rsync"); err != nil { + return fmt.Errorf("az rsync nem érhető el a rendszeren") + } + } + if j.Target == j.Source { + return fmt.Errorf("a cél és a forrás tároló megegyezik") + } + if m.settings == nil || !m.settings.IsStoragePathSchedulable(j.Target) { + return fmt.Errorf("a céltároló nem elérhető vagy nem választható") + } + // App-dir collision: refuse if the same app dir already exists at the target. + var collide []string + for _, app := range j.Apps { + if pathExists(appbackup.AppDataDir(j.TargetNS, app)) { + collide = append(collide, app) + } + } + if len(collide) > 0 { + return fmt.Errorf("ütközés a céltárolón — már létezik ezeknek az alkalmazásoknak az adata: %s", strings.Join(collide, ", ")) + } + // Free-space check (best-effort; GetDiskUsage is nil on non-linux). + need := m.migSourceSize(j) + j.BytesTotal = need + if du := system.GetDiskUsage(j.Target); du != nil { + avail := int64(du.AvailGB * 1e9) + if need > avail { + return fmt.Errorf("nincs elég hely a céltárolón (kb. %d GB szükséges, %.1f GB szabad)", need/1_000_000_000, du.AvailGB) + } + } + return nil +} + +// migSourceSize estimates the bytes to move (conservative; dedup may write less). +func (m *Manager) migSourceSize(j *MigrationJob) int64 { + if j.Scope == "all" { + return dirBytes(j.SourceNS) + } + var total int64 + for _, app := range j.Apps { + total += dirBytes(appbackup.AppDataDir(j.SourceNS, app)) + total += dirBytes(appbackup.RecoveryUnitPath(j.SourceNS, app)) + } + return total +} + +// runMigration drives the phase machine. Used both for a fresh start and for resume; each phase is +// idempotent and skips already-completed units, so re-entry never re-copies or re-removes. +func (m *Manager) runMigration(ctx context.Context, j *MigrationJob) { + defer m.releaseMigrating() + for { + var err error + switch j.Phase { + case PhaseStop: + err = m.migStop(j) + if err == nil { + j.Phase = PhaseCopy + } + case PhaseCopy: + err = m.migCopy(ctx, j) + if err == nil { + j.Phase = PhaseVerify + } + case PhaseVerify: + err = m.migVerify(ctx, j) + if err == nil { + j.Phase = PhaseFlip + } + case PhaseFlip, PhaseRedeploy: + err = m.migFlipRedeploy(j) + if err == nil { + j.Phase = PhaseCleanup + } + case PhaseCleanup: + err = m.migCleanup(j) + if err == nil { + j.Phase = PhaseDone + j.FinishedAt = time.Now().UTC() + } + case PhaseDone, PhaseAborted: + return + default: + err = fmt.Errorf("ismeretlen migrációs fázis: %q", j.Phase) + } + if err != nil { + m.migAbort(j, err) + return + } + j.CurrentApp = "" + if perr := m.persistJob(j); perr != nil { + m.logger.Printf("[ERROR] [migrate] journal write failed: %v", perr) + } + if j.Phase == PhaseDone { + m.logger.Printf("[INFO] [migrate] %s complete: %s → %s (%d app(s))", j.ID, j.Source, j.Target, len(j.Apps)) + return + } + } +} + +func (m *Manager) migAbort(j *MigrationJob, cause error) { + j.Phase = PhaseAborted + j.Error = cause.Error() + j.FinishedAt = time.Now().UTC() + _ = m.persistJob(j) + m.logger.Printf("[ERROR] [migrate] %s ABORTED at copy/verify/flip — source intact: %v", j.ID, cause) +} + +// migStop stops every affected app (idempotent — compose down on a stopped stack is a no-op). +func (m *Manager) migStop(j *MigrationJob) error { + for _, app := range j.Apps { + if err := m.doStop(app); err != nil { + return fmt.Errorf("alkalmazás leállítása sikertelen (%s): %w", app, err) + } + } + return nil +} + +// migCopy copies each app subtree (rsync, additive) and — for MigrateAll — the non-app content via +// the conflict-merge walk. Skips units already past `copied` so resume never re-copies. +func (m *Manager) migCopy(ctx context.Context, j *MigrationJob) error { + for _, app := range j.Apps { + u := j.Units[app] + if stateRank(u.State) >= stateRank(UnitCopied) { + continue + } + j.CurrentApp = app + _ = m.persistJob(j) + // appdata subtree (collision-free post-validate) + if err := m.copySubtree(ctx, j, appbackup.AppDataDir(j.SourceNS, app), appbackup.AppDataDir(j.TargetNS, app)); err != nil { + u.Error = err.Error() + return fmt.Errorf("másolás sikertelen (%s appdata): %w", app, err) + } + // the app's recovery unit (db-dumps + volume-dumps + compose + manifest) + if err := m.copySubtree(ctx, j, appbackup.RecoveryUnitPath(j.SourceNS, app), appbackup.RecoveryUnitPath(j.TargetNS, app)); err != nil { + u.Error = err.Error() + return fmt.Errorf("másolás sikertelen (%s mentés): %w", app, err) + } + u.State = UnitCopied + _ = m.persistJob(j) + } + if j.Scope == "all" { + u := j.Units[nonAppUnit] + if stateRank(u.State) < stateRank(UnitCopied) { + j.CurrentApp = "" + if err := walkMerge(m.logger, j.SourceNS, j.TargetNS, m.appDataSkipSet(j), false, func(b int64) { j.BytesDone += b }); err != nil { + u.Error = err.Error() + return fmt.Errorf("ügyfél-adatok összefésülése sikertelen: %w", err) + } + u.State = UnitCopied + _ = m.persistJob(j) + } + } + return nil +} + +// copySubtree rsyncs src→dst if src exists; a missing source is a no-op (nothing to move). +func (m *Manager) copySubtree(ctx context.Context, j *MigrationJob, src, dst string) error { + if !pathExists(src) { + return nil + } + return m.doCopy(ctx, src, dst, func(b int64) { j.BytesDone = b }) +} + +// migVerify confirms every source byte landed at the target before any source mutation. +func (m *Manager) migVerify(ctx context.Context, j *MigrationJob) error { + for _, app := range j.Apps { + u := j.Units[app] + if stateRank(u.State) >= stateRank(UnitVerified) { + continue + } + if err := m.verifySubtree(ctx, appbackup.AppDataDir(j.SourceNS, app), appbackup.AppDataDir(j.TargetNS, app)); err != nil { + u.Error = err.Error() + return fmt.Errorf("ellenőrzés sikertelen (%s appdata): %w", app, err) + } + if err := m.verifySubtree(ctx, appbackup.RecoveryUnitPath(j.SourceNS, app), appbackup.RecoveryUnitPath(j.TargetNS, app)); err != nil { + u.Error = err.Error() + return fmt.Errorf("ellenőrzés sikertelen (%s mentés): %w", app, err) + } + u.State = UnitVerified + _ = m.persistJob(j) + } + if j.Scope == "all" { + u := j.Units[nonAppUnit] + if stateRank(u.State) < stateRank(UnitVerified) { + // assert-only merge walk: every source file has a content-identical counterpart at target. + if err := walkMerge(m.logger, j.SourceNS, j.TargetNS, m.appDataSkipSet(j), true, nil); err != nil { + u.Error = err.Error() + return fmt.Errorf("ügyfél-adatok ellenőrzése sikertelen: %w", err) + } + u.State = UnitVerified + _ = m.persistJob(j) + } + } + return nil +} + +func (m *Manager) verifySubtree(ctx context.Context, src, dst string) error { + if !pathExists(src) { + return nil + } + return m.doVerify(ctx, src, dst) +} + +// migFlipRedeploy rewrites each app's HDD_PATH to the target and redeploys it (RedeployFromEnv as one +// idempotent unit). Journals `flipped` before the call and `redeployed` after, so resume re-calls. +func (m *Manager) migFlipRedeploy(j *MigrationJob) error { + for _, app := range j.Apps { + u := j.Units[app] + if stateRank(u.State) >= stateRank(UnitRedeployed) { + continue + } + j.CurrentApp = app + u.State = UnitFlipped + _ = m.persistJob(j) + if err := m.doFlipRedeploy(app, j.Target); err != nil { + u.Error = err.Error() + return fmt.Errorf("újratelepítés sikertelen (%s): %w", app, err) + } + u.State = UnitRedeployed + _ = m.persistJob(j) + } + return nil +} + +// migCleanup removes the SOURCE content. THE irreversible step: gated on every unit verified AND +// every app redeployed. Idempotent (RemoveAll of an absent path is a no-op). +func (m *Manager) migCleanup(j *MigrationJob) error { + if err := m.migCleanupAllowed(j); err != nil { + return err + } + for _, app := range j.Apps { + u := j.Units[app] + if u.State == UnitCleaned { + continue + } + if err := os.RemoveAll(appbackup.AppDataDir(j.SourceNS, app)); err != nil { + return fmt.Errorf("forrás törlése sikertelen (%s appdata): %w", app, err) + } + if err := os.RemoveAll(appbackup.RecoveryUnitPath(j.SourceNS, app)); err != nil { + return fmt.Errorf("forrás törlése sikertelen (%s mentés): %w", app, err) + } + u.State = UnitCleaned + _ = m.persistJob(j) + } + if j.Scope == "all" { + u := j.Units[nonAppUnit] + if u.State != UnitCleaned { + // Remove every remaining child of the source namespace (the non-app/customer content). + entries, err := os.ReadDir(j.SourceNS) + if err != nil && !os.IsNotExist(err) { + return fmt.Errorf("forrás névtér olvasása sikertelen: %w", err) + } + for _, e := range entries { + if err := os.RemoveAll(filepath.Join(j.SourceNS, e.Name())); err != nil { + return fmt.Errorf("forrás törlése sikertelen (%s): %w", e.Name(), err) + } + } + u.State = UnitCleaned + _ = m.persistJob(j) + } + } + return nil +} + +// migCleanupAllowed is the load-bearing gate: cleanup runs ONLY when all units are verified AND all +// apps are redeployed. +func (m *Manager) migCleanupAllowed(j *MigrationJob) error { + for key, u := range j.Units { + if stateRank(u.State) < stateRank(UnitVerified) { + return fmt.Errorf("cleanup gate: a(z) %q egység nincs ellenőrizve (állapot=%s)", key, u.State) + } + } + for _, app := range j.Apps { + if stateRank(j.Units[app].State) < stateRank(UnitRedeployed) { + return fmt.Errorf("cleanup gate: a(z) %q alkalmazás nincs újratelepítve (állapot=%s)", app, j.Units[app].State) + } + } + return nil +} + +// appDataSkipSet returns the source appdata dirs (rsync'd separately) to skip in the merge walk. +func (m *Manager) appDataSkipSet(j *MigrationJob) map[string]bool { + skip := map[string]bool{} + for _, app := range j.Apps { + skip[filepath.Clean(appbackup.AppDataDir(j.SourceNS, app))] = true + } + return skip +} + +// RecoverMigration resumes a crashed migration on startup (no-op if none or terminal). +func (m *Manager) RecoverMigration(ctx context.Context) { + j, err := m.loadJournal() + if err != nil { + m.logger.Printf("[WARN] [migrate] could not read migration journal: %v", err) + return + } + if j == nil || j.Phase == PhaseDone || j.Phase == PhaseAborted { + return + } + if err := m.acquireMigrating(); err != nil { + return + } + m.logger.Printf("[WARN] [migrate] resuming interrupted migration %s from phase=%s", j.ID, j.Phase) + m.setJob(j) + go m.runMigration(context.Background(), j) +} + +// --- seam dispatch (real impls; tests override via m.testSeams) --- + +func (m *Manager) doStop(name string) error { + if m.testSeams != nil && m.testSeams.stop != nil { + return m.testSeams.stop(name) + } + return m.StopStack(name) +} + +func (m *Manager) doCopy(ctx context.Context, src, dst string, onBytes func(int64)) error { + if m.testSeams != nil && m.testSeams.copy != nil { + return m.testSeams.copy(ctx, src, dst, onBytes) + } + return rsyncCopy(ctx, src, dst, onBytes) +} + +func (m *Manager) doVerify(ctx context.Context, src, dst string) error { + if m.testSeams != nil && m.testSeams.verify != nil { + return m.testSeams.verify(ctx, src, dst) + } + return rsyncVerify(ctx, src, dst) +} + +func (m *Manager) doFlipRedeploy(name, target string) error { + if m.testSeams != nil && m.testSeams.flipRedeploy != nil { + return m.testSeams.flipRedeploy(name, target) + } + cfg := m.LoadAppConfigByName(name) + if cfg == nil { + return fmt.Errorf("app config not found") + } + if err := m.RedeployFromEnv(name, flipEnv(cfg.Env, target)); err != nil { + return err + } + if !m.waitHealthy(name) { + return fmt.Errorf("az alkalmazás nem indult el az új tárhelyen") + } + return nil +} + +// flipEnv returns a copy of env with HDD_PATH set to target. +func flipEnv(env map[string]string, target string) map[string]string { + out := make(map[string]string, len(env)+1) + for k, v := range env { + out[k] = v + } + out["HDD_PATH"] = target + return out +} + +// waitHealthy polls until the stack is up (running/unhealthy) or times out. +func (m *Manager) waitHealthy(name string) bool { + deadline := time.Now().Add(90 * time.Second) + for { + _ = m.RefreshStatus() + if st, ok := m.GetStack(name); ok { + if st.State == StateRunning || st.State == StateUnhealthy { + return true + } + } + if time.Now().After(deadline) { + if st, ok := m.GetStack(name); ok { + return st.State == StateRunning || st.State == StateUnhealthy || st.State == StateStarting + } + return false + } + time.Sleep(3 * time.Second) + } +} + +// --- journal persistence --- + +func (m *Manager) journalPath() string { + return filepath.Join(m.cfg.Paths.DataDir, migrationJournalFile) +} + +// persistJob stamps UpdatedAt and writes the journal atomically. +func (m *Manager) persistJob(j *MigrationJob) error { + j.UpdatedAt = time.Now().UTC() + return m.writeJournal(j) +} + +func (m *Manager) writeJournal(j *MigrationJob) error { + data, err := json.MarshalIndent(j, "", " ") + if err != nil { + return err + } + if err := os.MkdirAll(filepath.Dir(m.journalPath()), 0o755); err != nil { + return err + } + tmp := m.journalPath() + ".tmp" + if err := os.WriteFile(tmp, data, 0o600); err != nil { + os.Remove(tmp) + return err + } + return os.Rename(tmp, m.journalPath()) +} + +func (m *Manager) loadJournal() (*MigrationJob, error) { + data, err := os.ReadFile(m.journalPath()) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, err + } + var j MigrationJob + if err := json.Unmarshal(data, &j); err != nil { + return nil, err + } + return &j, nil +} + +// --- rsync + filesystem primitives (additive — NO --delete) --- + +// rsyncCopy mirrors src→dst additively with `rsync -a --checksum` (NEVER --delete). onBytes receives +// the running transferred-bytes total parsed from --info=progress2 (best-effort). +func rsyncCopy(ctx context.Context, src, dst string, onBytes func(int64)) error { + if err := os.MkdirAll(dst, 0o755); err != nil { + return fmt.Errorf("mkdir %s: %w", dst, err) + } + cctx, cancel := context.WithTimeout(ctx, migrateCopyTimeout) + defer cancel() + cmd := exec.CommandContext(cctx, "rsync", "-a", "--checksum", "--info=progress2", + strings.TrimRight(src, "/")+"/", strings.TrimRight(dst, "/")+"/") + stdout, err := cmd.StdoutPipe() + if err != nil { + return err + } + var stderr bytes.Buffer + cmd.Stderr = &stderr + if err := cmd.Start(); err != nil { + return err + } + scanner := bufio.NewScanner(stdout) + for scanner.Scan() { + if b, ok := parseProgress2Bytes(scanner.Text()); ok && onBytes != nil { + onBytes(b) + } + } + if err := cmd.Wait(); err != nil { + return fmt.Errorf("rsync: %v: %s", err, strings.TrimSpace(stderr.String())) + } + return nil +} + +// rsyncVerify dry-runs `rsync -ani --checksum` and fails if ANY content transfer/create is still +// pending (itemize lines beginning with <, >, or c). Attr-only `.`-prefixed lines are ignored. +func rsyncVerify(ctx context.Context, src, dst string) error { + cctx, cancel := context.WithTimeout(ctx, migrateVerifyTimeout) + defer cancel() + cmd := exec.CommandContext(cctx, "rsync", "-a", "-n", "-i", "--checksum", + strings.TrimRight(src, "/")+"/", strings.TrimRight(dst, "/")+"/") + out, err := cmd.CombinedOutput() + if err != nil { + return fmt.Errorf("verify rsync: %v: %s", err, strings.TrimSpace(string(out))) + } + var pending []string + for _, ln := range strings.Split(string(out), "\n") { + ln = strings.TrimRight(ln, "\r") + if isPendingTransfer(ln) { + pending = append(pending, strings.TrimSpace(ln)) + } + } + if len(pending) > 0 { + return fmt.Errorf("%d függőben lévő átvitel maradt, pl. %q", len(pending), pending[0]) + } + return nil +} + +// isPendingTransfer reports whether an rsync -i itemize line denotes a content transfer/create +// (file or dir). Attr-only changes ('.' prefix) and messages ('*') are NOT transfers. +func isPendingTransfer(line string) bool { + if len(line) < 2 { + return false + } + switch line[0] { + case '>', '<', 'c': + return true + default: + return false + } +} + +// parseProgress2Bytes extracts the running transferred-bytes total from an --info=progress2 line +// (first field is a comma-grouped byte count, e.g. "1,234,567 45% ..."). +func parseProgress2Bytes(line string) (int64, bool) { + fields := strings.Fields(strings.TrimSpace(line)) + if len(fields) == 0 { + return 0, false + } + digits := strings.ReplaceAll(fields[0], ",", "") + n, err := strconv.ParseInt(digits, 10, 64) + if err != nil { + return 0, false + } + return n, true +} + +// walkMerge merges source ns into target ns with the conflict rules (Change: non-app content only). +// assertOnly=true performs no writes — it verifies every source regular file has a content-identical +// counterpart at target (the file itself or a `(N)` sibling), returning an error if any does not. +// skip names absolute SOURCE dirs (app appdata dirs) handled by rsync, which are pruned. +func walkMerge(lg *log.Logger, srcNS, dstNS string, skip map[string]bool, assertOnly bool, onBytes func(int64)) error { + if !pathExists(srcNS) { + return nil + } + return filepath.WalkDir(srcNS, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + if filepath.Clean(path) == filepath.Clean(srcNS) { + return nil + } + rel, rerr := filepath.Rel(srcNS, path) + if rerr != nil { + return rerr + } + dst := filepath.Join(dstNS, rel) + + if d.IsDir() { + if skip[filepath.Clean(path)] { + return filepath.SkipDir + } + if assertOnly { + return nil + } + return os.MkdirAll(dst, 0o755) + } + + // Symlink: recreate-if-absent (copy mode); ignored in assert mode. + if d.Type()&fs.ModeSymlink != 0 { + if assertOnly { + return nil + } + return mergeSymlink(path, dst) + } + if !d.Type().IsRegular() { + if lg != nil { + lg.Printf("[WARN] [migrate] skipping non-regular file %s", path) + } + return nil + } + + srcSum, serr := fileSum(path) + if serr != nil { + return serr + } + matched, merr := matchesExistingTarget(dst, srcSum) + if merr != nil { + return merr + } + if matched { + return nil // dedup / already present + } + if assertOnly { + return fmt.Errorf("a forrásfájlnak nincs azonos másolata a célon: %s", rel) + } + out := dst + if pathExists(dst) { + out = lowestFreeSibling(dst) // never overwrite an existing target file + } + n, cerr := copyFile(path, out) + if cerr != nil { + return cerr + } + if onBytes != nil { + onBytes(n) + } + return nil + }) +} + +// matchesExistingTarget reports whether srcSum equals the checksum of dst OR any `(N)` sibling. +func matchesExistingTarget(dst, srcSum string) (bool, error) { + if pathExists(dst) { + s, err := fileSum(dst) + if err != nil { + return false, err + } + if s == srcSum { + return true, nil + } + } + for n := 1; ; n++ { + sib := siblingName(dst, n) + if !pathExists(sib) { + break + } + s, err := fileSum(sib) + if err != nil { + return false, err + } + if s == srcSum { + return true, nil + } + } + return false, nil +} + +// lowestFreeSibling returns the first `(N)` name (N≥1) that does not exist. +func lowestFreeSibling(dst string) string { + for n := 1; ; n++ { + sib := siblingName(dst, n) + if !pathExists(sib) { + return sib + } + } +} + +// siblingName builds `(N)` using the LAST extension (foo.tar.gz → stem "foo.tar", ext ".gz"). +func siblingName(dst string, n int) string { + dir := filepath.Dir(dst) + base := filepath.Base(dst) + ext := filepath.Ext(base) + stem := strings.TrimSuffix(base, ext) + return filepath.Join(dir, fmt.Sprintf("%s(%d)%s", stem, n, ext)) +} + +func mergeSymlink(src, dst string) error { + target, err := os.Readlink(src) + if err != nil { + return err + } + if existing, lerr := os.Readlink(dst); lerr == nil { + if existing == target { + return nil // identical symlink already present + } + // differing symlink → rename-sibling + dst = lowestFreeSibling(dst) + } else if pathExists(dst) { + dst = lowestFreeSibling(dst) + } + if err := os.MkdirAll(filepath.Dir(dst), 0o755); err != nil { + return err + } + return os.Symlink(target, dst) +} + +// copyFile streams src→dst via a temp file + chmod + fsync + atomic rename. Returns bytes written. +func copyFile(src, dst string) (int64, error) { + in, err := os.Open(src) + if err != nil { + return 0, err + } + defer in.Close() + fi, err := in.Stat() + if err != nil { + return 0, err + } + if err := os.MkdirAll(filepath.Dir(dst), 0o755); err != nil { + return 0, err + } + tmp := dst + ".felhom-mig.tmp" + out, err := os.OpenFile(tmp, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, fi.Mode().Perm()) + if err != nil { + return 0, err + } + n, cerr := io.Copy(out, in) + if cerr != nil { + out.Close() + os.Remove(tmp) + return 0, cerr + } + if err := out.Sync(); err != nil { + out.Close() + os.Remove(tmp) + return 0, err + } + if err := out.Close(); err != nil { + os.Remove(tmp) + return 0, err + } + if err := os.Chmod(tmp, fi.Mode().Perm()); err != nil { + os.Remove(tmp) + return 0, err + } + if err := os.Rename(tmp, dst); err != nil { + os.Remove(tmp) + return 0, err + } + return n, nil +} + +// fileSum returns the hex sha256 of a file (streaming). +func fileSum(path string) (string, error) { + f, err := os.Open(path) + if err != nil { + return "", err + } + defer f.Close() + h := sha256.New() + if _, err := io.Copy(h, f); err != nil { + return "", err + } + return hex.EncodeToString(h.Sum(nil)), nil +} + +func pathExists(p string) bool { + _, err := os.Stat(p) + return err == nil +} + +// dirBytes returns the total size of a directory via `du -sb` (0 if absent/error). +func dirBytes(dir string) int64 { + if !pathExists(dir) { + return 0 + } + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + out, err := exec.CommandContext(ctx, "du", "-sb", dir).Output() + if err != nil { + return 0 + } + fields := strings.Fields(string(out)) + if len(fields) == 0 { + return 0 + } + n, err := strconv.ParseInt(fields[0], 10, 64) + if err != nil { + return 0 + } + return n +} diff --git a/controller/internal/stacks/migrate_test.go b/controller/internal/stacks/migrate_test.go new file mode 100644 index 0000000..7f867f4 --- /dev/null +++ b/controller/internal/stacks/migrate_test.go @@ -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) }