diff --git a/tools/tht/internal/backup/create.go b/tools/tht/internal/backup/create.go index d0b9692d..1c22f8d8 100644 --- a/tools/tht/internal/backup/create.go +++ b/tools/tht/internal/backup/create.go @@ -70,12 +70,13 @@ type archiveRunner interface { } type dependencies struct { - runner archiveRunner - now func() time.Time - homeDir func() (string, error) - revision func(context.Context, string) (string, error) - sleep func(time.Duration) - rename func(string, string) error + runner archiveRunner + now func() time.Time + homeDir func() (string, error) + revision func(context.Context, string) (string, error) + sleep func(time.Duration) + reserveOutput func(string) (*archiveReservation, error) + publishReserved func(*archiveReservation, string) error } // Create creates an archive with the real Docker command boundary. It performs no shell @@ -94,13 +95,14 @@ func Create(ctx context.Context, installation config.Installation, request Creat } return strings.TrimSpace(string(value)), nil }, - sleep: time.Sleep, - rename: os.Rename, + sleep: time.Sleep, + reserveOutput: reserveArchiveOutput, + publishReserved: publishReservedArchive, }) } func createWithDependencies(ctx context.Context, installation config.Installation, request CreateRequest, dependencies dependencies) (result Result, resultErr error) { - if dependencies.runner == nil || dependencies.now == nil || dependencies.homeDir == nil || dependencies.revision == nil || dependencies.sleep == nil || dependencies.rename == nil { + if dependencies.runner == nil || dependencies.now == nil || dependencies.homeDir == nil || dependencies.revision == nil || dependencies.sleep == nil || dependencies.reserveOutput == nil || dependencies.publishReserved == nil { return Result{}, errors.New("backup dependencies are incomplete") } if request.IncludeSecrets && !request.Confirm { @@ -142,6 +144,16 @@ func createWithDependencies(ctx context.Context, installation config.Installatio if err := ensureNewArchivePath(output); err != nil { return Result{}, err } + reservation, err := dependencies.reserveOutput(output) + if err != nil { + return Result{}, err + } + published := false + defer func() { + if !published { + _ = reservation.RemoveIfOwned() + } + }() rendered, err := renderedConfiguration(ctx, installation, dependencies.runner) if err != nil { @@ -201,9 +213,10 @@ func createWithDependencies(ctx context.Context, installation config.Installatio Images: images, Volumes: volumes, } - if err := writeArchive(ctx, output, installation, request, secretPaths, manifest, volumes, dependencies); err != nil { + if err := writeArchive(ctx, output, reservation, installation, request, secretPaths, manifest, volumes, dependencies); err != nil { return Result{}, err } + published = true if wasRunning { if err := composeStartAndVerify(ctx, installation, dependencies.runner); err != nil { return Result{}, err @@ -286,15 +299,90 @@ func ensureNewArchivePath(output string) error { if err := os.MkdirAll(filepath.Dir(output), 0o700); err != nil { return fmt.Errorf("create backup directory: %w", err) } - if info, err := os.Lstat(output); err == nil || !errors.Is(err, os.ErrNotExist) { - if err == nil && info.Mode().IsRegular() { - return errors.New("backup output already exists") - } - return errors.New("backup output path is unavailable") + return nil +} + +// archiveReservation claims an output path without replacing an existing archive. Its file +// identity is retained so cleanup never removes a path another process took over. +type archiveReservation struct { + path string + file *os.File + info os.FileInfo +} + +func reserveArchiveOutput(output string) (*archiveReservation, error) { + file, err := os.OpenFile(output, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600) + if errors.Is(err, os.ErrExist) { + return nil, errors.New("backup output already exists") + } + if err != nil { + return nil, fmt.Errorf("reserve backup output: %w", err) + } + info, statErr := file.Stat() + if statErr != nil { + _ = file.Close() + return nil, fmt.Errorf("inspect reserved backup output: %w", statErr) + } + return &archiveReservation{path: output, file: file, info: info}, nil +} + +func publishReservedArchive(reservation *archiveReservation, temporaryPath string) error { + if reservation == nil || reservation.file == nil { + return errors.New("backup output reservation is unavailable") + } + temporary, err := os.Open(temporaryPath) + if err != nil { + return fmt.Errorf("open temporary backup archive: %w", err) + } + defer temporary.Close() + if err := reservation.file.Truncate(0); err != nil { + return fmt.Errorf("prepare reserved backup output: %w", err) + } + if _, err := reservation.file.Seek(0, io.SeekStart); err != nil { + return fmt.Errorf("seek reserved backup output: %w", err) + } + if _, err := io.Copy(reservation.file, temporary); err != nil { + return fmt.Errorf("write reserved backup output: %w", err) + } + if err := reservation.file.Sync(); err != nil { + return fmt.Errorf("fsync reserved backup output: %w", err) + } + if err := reservation.file.Close(); err != nil { + return fmt.Errorf("close reserved backup output: %w", err) + } + reservation.file = nil + current, err := os.Stat(reservation.path) + if err != nil || !os.SameFile(reservation.info, current) { + return errors.New("backup output ownership changed before publication") } return nil } +// RemoveIfOwned removes the reservation only when the output path still names the exact file +// created by this invocation. It is safe to call after another process has claimed the path. +func (reservation *archiveReservation) RemoveIfOwned() error { + if reservation == nil { + return nil + } + if reservation.file != nil { + if err := reservation.file.Close(); err != nil { + return err + } + reservation.file = nil + } + current, err := os.Stat(reservation.path) + if errors.Is(err, os.ErrNotExist) { + return nil + } + if err != nil { + return err + } + if !os.SameFile(reservation.info, current) { + return nil + } + return os.Remove(reservation.path) +} + type renderedCompose struct { Volumes map[string]struct { Name string `json:"name"` @@ -398,11 +486,74 @@ func imageIdentities(ctx context.Context, installation config.Installation, runn } func installationRunning(ctx context.Context, installation config.Installation, runner archiveRunner) (bool, error) { - result, err := runner.Run(ctx, installation.ComposeArgs("ps", "--status", "running", "-q"), nil) + result, err := runner.Run(ctx, installation.ComposeArgs("ps", "--all", "--format", "json"), nil) if err != nil { - return false, dockerError("inspect running services", result, err) + return false, dockerError("inspect service states", result, err) } - return strings.TrimSpace(result.Stdout) != "", nil + states, err := backupServiceStates(result.Stdout) + if err != nil { + return false, err + } + running := false + coreRunning := false + for _, service := range states { + switch service.State { + case "exited", "dead": + continue + case "running": + running = true + if service.Service == "core" { + coreRunning = true + } + default: + return false, fmt.Errorf("service %q is %q and is not safely quiesced; stop the installation before backup", service.Service, service.State) + } + } + if running && !coreRunning { + return false, errors.New("core is not running while other services are active; stop the installation before backup") + } + return running, nil +} + +type backupServiceState struct { + Service string `json:"Service"` + State string `json:"State"` +} + +func backupServiceStates(value string) ([]backupServiceState, error) { + trimmed := strings.TrimSpace(value) + if trimmed == "" || trimmed == "[]" { + return nil, nil + } + var array []backupServiceState + if err := json.Unmarshal([]byte(trimmed), &array); err == nil { + return normalizeBackupServiceStates(array) + } + decoder := json.NewDecoder(strings.NewReader(trimmed)) + var states []backupServiceState + for { + var state backupServiceState + err := decoder.Decode(&state) + if errors.Is(err, io.EOF) { + break + } + if err != nil { + return nil, errors.New("Docker Compose returned invalid service states") + } + states = append(states, state) + } + return normalizeBackupServiceStates(states) +} + +func normalizeBackupServiceStates(states []backupServiceState) ([]backupServiceState, error) { + for index := range states { + states[index].Service = strings.TrimSpace(states[index].Service) + states[index].State = strings.ToLower(strings.TrimSpace(states[index].State)) + if states[index].Service == "" || states[index].State == "" { + return nil, errors.New("Docker Compose returned invalid service states") + } + } + return states, nil } func maintenance(ctx context.Context, installation config.Installation, runner archiveRunner, activate bool) error { @@ -480,14 +631,13 @@ func runCompose(ctx context.Context, installation config.Installation, runner ar return nil } -func writeArchive(ctx context.Context, output string, installation config.Installation, request CreateRequest, secretPaths []string, manifest Manifest, volumes []VolumeMetadata, dependencies dependencies) (resultErr error) { +func writeArchive(ctx context.Context, output string, reservation *archiveReservation, installation config.Installation, request CreateRequest, secretPaths []string, manifest Manifest, volumes []VolumeMetadata, dependencies dependencies) (resultErr error) { directory := filepath.Dir(output) temporary, err := os.CreateTemp(directory, ".tht-backup-*.tmp") if err != nil { return fmt.Errorf("create temporary backup archive: %w", err) } temporaryPath := temporary.Name() - published := false closedFile := false defer func() { var closeErr error @@ -497,10 +647,7 @@ func writeArchive(ctx context.Context, output string, installation config.Instal if closeErr != nil && resultErr == nil { resultErr = closeErr } - if !published { - _ = os.Remove(temporaryPath) - _ = os.Remove(output) - } + _ = os.Remove(temporaryPath) }() if err := temporary.Chmod(0o600); err != nil { return fmt.Errorf("protect temporary backup archive: %w", err) @@ -632,10 +779,9 @@ func writeArchive(ctx context.Context, output string, installation config.Instal return fmt.Errorf("close backup archive: %w", err) } closedFile = true - if err := dependencies.rename(temporaryPath, output); err != nil { + if err := dependencies.publishReserved(reservation, temporaryPath); err != nil { return fmt.Errorf("publish backup archive: %w", err) } - published = true return nil } diff --git a/tools/tht/internal/backup/create_test.go b/tools/tht/internal/backup/create_test.go index 08528d4e..0e0ebcad 100644 --- a/tools/tht/internal/backup/create_test.go +++ b/tools/tht/internal/backup/create_test.go @@ -176,7 +176,7 @@ func TestCreateCleansIncompleteArchiveAfterAtomicPublishFailure(t *testing.T) { runner := newBackupRunner(fixture.installation, false) directory := t.TempDir() dependencies := testDependencies(t, runner) - dependencies.rename = func(string, string) error { return errors.New("publish failed") } + dependencies.publishReserved = func(*archiveReservation, string) error { return errors.New("publish failed") } if _, err := createWithDependencies(context.Background(), fixture.installation, CreateRequest{Output: filepath.Join(directory, "failed.zip")}, dependencies); err == nil { t.Fatal("Create() succeeded after publish failure") @@ -184,6 +184,39 @@ func TestCreateCleansIncompleteArchiveAfterAtomicPublishFailure(t *testing.T) { assertNoBackupArtifacts(t, directory) } +func TestCreateDoesNotOverwriteOrDeleteAnOutputCreatedBeforeReservation(t *testing.T) { + fixture := newBackupFixture(t, "local") + runner := newBackupRunner(fixture.installation, false) + directory := t.TempDir() + output := filepath.Join(directory, "race.zip") + dependencies := testDependencies(t, runner) + reserve := dependencies.reserveOutput + dependencies.reserveOutput = func(path string) (*archiveReservation, error) { + if err := os.WriteFile(path, []byte("created-by-another-process"), 0o600); err != nil { + return nil, err + } + return reserve(path) + } + + if _, err := createWithDependencies(context.Background(), fixture.installation, CreateRequest{Output: output}, dependencies); err == nil { + t.Fatal("Create() succeeded after another process claimed the output path") + } + contents, err := os.ReadFile(output) + if err != nil { + t.Fatalf("racing output was removed: %v", err) + } + if got := string(contents); got != "created-by-another-process" { + t.Fatalf("racing output = %q, want unchanged content", got) + } + entries, err := os.ReadDir(directory) + if err != nil { + t.Fatal(err) + } + if len(entries) != 1 || entries[0].Name() != "race.zip" { + t.Fatalf("temporary backup artifacts remain after race: %v", entries) + } +} + func TestCreateExcludesExternalSecretPayloadsByDefaultButRecordsDigests(t *testing.T) { fixture := newBackupFixture(t, "local") runner := newBackupRunner(fixture.installation, false) @@ -341,6 +374,26 @@ func TestCreateHonorsTheSharedInstallationLifecycleLock(t *testing.T) { } } +func TestCreateRefusesMutableNonRunningServiceStates(t *testing.T) { + for _, state := range []string{"paused", "restarting", "created", "removing"} { + t.Run(state, func(t *testing.T) { + fixture := newBackupFixture(t, "local") + runner := newBackupRunner(fixture.installation, false) + runner.serviceStates = map[string]string{"core": "running", "frontend": state} + dependencies := testDependencies(t, runner) + dependencies.sleep = func(time.Duration) { t.Fatal("unsafe service state reached the drain loop") } + + _, err := createWithDependencies(context.Background(), fixture.installation, CreateRequest{Output: filepath.Join(t.TempDir(), "unsafe.zip")}, dependencies) + if err == nil || !strings.Contains(err.Error(), "not safely quiesced") { + t.Fatalf("Create() error = %v, want unsafe service-state refusal", err) + } + if runner.stopCount != 0 || runner.streams != 0 { + t.Fatalf("unsafe state was not refused before snapshot: stops=%d streams=%d", runner.stopCount, runner.streams) + } + }) + } +} + type backupFixture struct { root string installationID string @@ -437,6 +490,7 @@ type fakeBackupRunner struct { stopCount int startCount int healthChecks int + serviceStates map[string]string } func newBackupRunner(installation config.Installation, running bool) *fakeBackupRunner { @@ -474,16 +528,25 @@ func (runner *fakeBackupRunner) Run(_ context.Context, args []string, _ io.Reade return compose.Result{Stdout: string(encoded)}, nil case strings.Contains(command, " images --format json"): return compose.Result{Stdout: "{\"Service\":\"core\",\"Repository\":\"thothii-core\",\"Tag\":\"test\",\"ID\":\"sha256:core\"}\n{\"Service\":\"frontend\",\"Repository\":\"thothii-frontend\",\"Tag\":\"test\",\"ID\":\"sha256:frontend\"}\n"}, nil - case strings.Contains(command, " ps --status running -q core"): - if runner.coreRunning { - return compose.Result{Stdout: "core-id\n"}, nil + case strings.Contains(command, " ps --all --format json"): + runner.healthChecks++ + states := runner.serviceStates + if states == nil { + if runner.running { + return compose.Result{Stdout: healthyServicesPayload()}, nil + } + return compose.Result{}, nil } - return compose.Result{}, nil - case strings.Contains(command, " ps --status running -q"): - if runner.running { - return compose.Result{Stdout: "container-id\n"}, nil + services := make([]string, 0, len(states)) + for service := range states { + services = append(services, service) } - return compose.Result{}, nil + sort.Strings(services) + lines := make([]string, 0, len(services)) + for _, service := range services { + lines = append(lines, fmt.Sprintf(`{"Service":%q,"State":%q}`, service, states[service])) + } + return compose.Result{Stdout: strings.Join(lines, "\n")}, nil case strings.Contains(command, "/internal/maintenance/status"): return compose.Result{Stdout: fmt.Sprintf(`{"active":%t,"admissions":0,"recoveryRequired":false}`, runner.maintenance)}, nil case strings.Contains(command, "/internal/maintenance/activate"): @@ -553,12 +616,13 @@ func (runner *fakeBackupRunner) SessionInventoryScope() string { func testDependencies(t *testing.T, runner archiveRunner) dependencies { t.Helper() return dependencies{ - runner: runner, - now: func() time.Time { return time.Date(2026, 8, 16, 8, 11, 12, 0, time.UTC) }, - homeDir: func() (string, error) { return t.TempDir(), nil }, - revision: func(context.Context, string) (string, error) { return testRevision, nil }, - sleep: func(time.Duration) {}, - rename: os.Rename, + runner: runner, + now: func() time.Time { return time.Date(2026, 8, 16, 8, 11, 12, 0, time.UTC) }, + homeDir: func() (string, error) { return t.TempDir(), nil }, + revision: func(context.Context, string) (string, error) { return testRevision, nil }, + sleep: func(time.Duration) {}, + reserveOutput: reserveArchiveOutput, + publishReserved: publishReservedArchive, } } diff --git a/tools/tht/internal/backup/manifest.go b/tools/tht/internal/backup/manifest.go index 50191cff..38e388ac 100644 --- a/tools/tht/internal/backup/manifest.go +++ b/tools/tht/internal/backup/manifest.go @@ -7,6 +7,7 @@ import ( "encoding/json" "errors" "fmt" + "io" "path" "regexp" "sort" @@ -193,7 +194,8 @@ func DecodeManifest(value []byte) (Manifest, error) { if err := decoder.Decode(&manifest); err != nil { return Manifest{}, fmt.Errorf("decode backup manifest: %w", err) } - if decoder.More() { + var trailing any + if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) { return Manifest{}, errors.New("backup manifest contains trailing data") } if err := manifest.Finalize(); err != nil { diff --git a/tools/tht/internal/backup/manifest_test.go b/tools/tht/internal/backup/manifest_test.go index 80fa069d..77f07f34 100644 --- a/tools/tht/internal/backup/manifest_test.go +++ b/tools/tht/internal/backup/manifest_test.go @@ -125,7 +125,8 @@ func TestDecodeManifestRejectsASecondDocument(t *testing.T) { if err != nil { t.Fatal(err) } - if _, err := DecodeManifest(append(encoded, encoded...)); err == nil { + withSecondDocument := append(append([]byte(nil), encoded...), append([]byte(" \n"), encoded...)...) + if _, err := DecodeManifest(withSecondDocument); err == nil { t.Fatal("DecodeManifest() accepted a second JSON document") } }