From 59f3fe3d1d6f6b2e768f3274ecb86c2a6fca22e5 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 10 Aug 2026 17:44:38 +0000 Subject: [PATCH] Make atomic file replacement crash durable --- docs/internal/fileops.md | 35 +++ docs/internal/manifest.md | 8 +- docs/internal/overview.md | 2 +- docs/roadmap/implementation.md | 2 +- internal/adapters/subprocess/run.go | 40 +-- internal/adapters/whisperx/http.go | 37 +-- internal/fileops/fileops.go | 244 +++++++++++++----- internal/fileops/fileops_test.go | 229 ++++++++++++++++ internal/fileops/sync_directory_unix.go | 7 +- .../fileops/sync_directory_unsupported.go | 9 +- internal/fileops/sync_directory_windows.go | 7 +- internal/manifest/store.go | 90 +------ internal/manifest/store_test.go | 25 ++ 13 files changed, 502 insertions(+), 233 deletions(-) create mode 100644 docs/internal/fileops.md diff --git a/docs/internal/fileops.md b/docs/internal/fileops.md new file mode 100644 index 0000000..09b2943 --- /dev/null +++ b/docs/internal/fileops.md @@ -0,0 +1,35 @@ +# Internal: File Operations + +`internal/fileops` owns the narrow mechanics for durable replacement of one +byte file. Callers keep ownership of serialization, validation, cancellation, +and destination-directory policy. + +## Replacement Contract + +`ReplaceFileAtomic` requires an existing destination directory. It creates a +sibling temporary file, writes the complete byte sequence, applies the +caller-supplied mode, syncs and closes the file, runs an optional pre-rename +check, replaces the destination with a rename, then syncs the containing +directory. + +The pre-rename check is the last point at which a caller can cancel without +installing a new destination. A failure before the rename leaves the old +destination unchanged and removes the temporary file; any cleanup failure is +returned alongside the primary failure. A failure after the rename may leave +the new file visible, but it is not reported as crash-durable. + +Replacement follows the operating system's same-filesystem rename semantics. +If a platform cannot replace an existing destination, the operation returns an +error and never removes the old file as an emulation step. + +## Directory-Sync Support + +Linux and macOS attempt to sync the destination directory. Windows opens the +directory with backup semantics and flushes its buffers. If either operation +is unavailable for the platform, directory handle, or filesystem, +`ErrDirectorySyncUnsupported` is returned. Narratio does not treat that result +as successful crash-durable replacement. + +`WriteFileAtomic`, copy helpers, and downloaded temporary-file installation +retain their compatibility behavior of creating the destination parent with +the repository's workspace permissions before using this contract. diff --git a/docs/internal/manifest.md b/docs/internal/manifest.md index f3e0e33..e2b322f 100644 --- a/docs/internal/manifest.md +++ b/docs/internal/manifest.md @@ -48,9 +48,15 @@ The model admits these stage states: - validates loaded documents; - normalizes missing maps/stage records; -- writes atomically via temp file + rename; +- writes through a sibling temporary file, syncing the completed file and + destination directory after atomic replacement; - updates `updated_at` on save. +If the operating system or filesystem cannot sync a directory, save returns an +explicit error instead of claiming crash-durable replacement. A returned error +after the rename can therefore leave the new manifest visible but not confirmed +durable; callers must reload it before retrying. + ## Execution Semantics The application runner marks an executing stage running and then succeeded or diff --git a/docs/internal/overview.md b/docs/internal/overview.md index eb28dca..860fa87 100644 --- a/docs/internal/overview.md +++ b/docs/internal/overview.md @@ -35,7 +35,7 @@ progress and artifact services resolve durable inputs and outputs. | Artifacts and paths | `internal/artifacts`, `internal/pathsafe` | Artifact identities and resolution, local and remote path/key models, current-state discovery, and confined relative destinations. | | Previous-session cache | `internal/previouscache` | Deterministic planning and materialization requirements for configured previous-session inputs. | | Artifact policy | `internal/artifactpolicy` | Source and destination policy, configured artifact identity validation, and publish destination safety. | -| Shared models and file operations | `internal/artifactmodel`, `internal/contracts`, `internal/fileops` | Transcript and artifact data contracts plus narrow atomic filesystem helpers. | +| Shared models and file operations | `internal/artifactmodel`, `internal/contracts`, [`internal/fileops`](fileops.md) | Transcript and artifact data contracts plus durable single-file replacement helpers; unsupported directory syncing is reported explicitly. | | Logging | `internal/logging` | Application logger construction and shared structured logging behavior. | The application boundary composes concrete implementations. Stages depend on diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 108edea..6f8703f 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -18,7 +18,7 @@ All stages are pending when this plan is created. | ---: | --- | --- | --- | | 1 | Align data classification and group workspace modes | RSK-004 | Completed | | 2 | Enforce safe identifiers and fuzz path/source contracts | COR-002, TST-013 | Completed | -| 3 | Consolidate crash-durable atomic file replacement | RSK-002, DUP-001, DUP-005 | Pending | +| 3 | Consolidate crash-durable atomic file replacement | RSK-002, DUP-001, DUP-005 | Completed | | 4 | Add confined destination and download/install capabilities | COR-003, DUP-003, TST-003 | Pending | | 5 | Confine recursive cleanup and replace sentinel locks | RSK-003 | Pending | | 6 | Harden API-key file acquisition | RSK-010 | Pending | diff --git a/internal/adapters/subprocess/run.go b/internal/adapters/subprocess/run.go index fcbb7bb..c99a5c3 100644 --- a/internal/adapters/subprocess/run.go +++ b/internal/adapters/subprocess/run.go @@ -128,48 +128,14 @@ func WriteYAMLAtomic(path string, value any, perm os.FileMode) error { return nil } -// WriteFileAtomic writes bytes via same-directory temp file + atomic rename. +// WriteFileAtomic writes bytes through the shared durable replacement primitive. func WriteFileAtomic(path string, data []byte, perm os.FileMode) error { if strings.TrimSpace(path) == "" { return fmt.Errorf("write file: path is required") } - - dir := filepath.Dir(path) - if err := fileops.EnsureWorkspaceDirectory(dir); err != nil { - return fmt.Errorf("create parent directory %q: %w", dir, err) + if err := fileops.WriteFileAtomic(path, data, perm); err != nil { + return fmt.Errorf("write file %q: %w", path, err) } - - base := filepath.Base(path) - tmp, err := os.CreateTemp(dir, "."+base+".tmp-*") - if err != nil { - return fmt.Errorf("create temp file: %w", err) - } - tmpPath := tmp.Name() - removeTmp := true - defer func() { - if removeTmp { - _ = os.Remove(tmpPath) - } - }() - - if _, err := tmp.Write(data); err != nil { - _ = tmp.Close() - return fmt.Errorf("write temp file: %w", err) - } - if err := tmp.Sync(); err != nil { - _ = tmp.Close() - return fmt.Errorf("sync temp file: %w", err) - } - if err := tmp.Close(); err != nil { - return fmt.Errorf("close temp file: %w", err) - } - if err := os.Chmod(tmpPath, perm); err != nil { - return fmt.Errorf("chmod temp file: %w", err) - } - if err := os.Rename(tmpPath, path); err != nil { - return fmt.Errorf("rename temp file: %w", err) - } - removeTmp = false return nil } diff --git a/internal/adapters/whisperx/http.go b/internal/adapters/whisperx/http.go index 59a280c..4c51eb7 100644 --- a/internal/adapters/whisperx/http.go +++ b/internal/adapters/whisperx/http.go @@ -282,41 +282,8 @@ func writeFileAtomic(path string, data []byte, perm os.FileMode) error { if strings.TrimSpace(path) == "" { return fmt.Errorf("path is required") } - dir := filepath.Dir(path) - if err := fileops.EnsureWorkspaceDirectory(dir); err != nil { - return fmt.Errorf("create parent dir %q: %w", dir, err) + if err := fileops.WriteFileAtomic(path, data, perm); err != nil { + return fmt.Errorf("write file %q: %w", path, err) } - - base := filepath.Base(path) - tmp, err := os.CreateTemp(dir, "."+base+".tmp-*") - if err != nil { - return fmt.Errorf("create temp file: %w", err) - } - tmpPath := tmp.Name() - removeTmp := true - defer func() { - if removeTmp { - _ = os.Remove(tmpPath) - } - }() - - if _, err := tmp.Write(data); err != nil { - _ = tmp.Close() - return fmt.Errorf("write temp file: %w", err) - } - if err := tmp.Sync(); err != nil { - _ = tmp.Close() - return fmt.Errorf("sync temp file: %w", err) - } - if err := tmp.Close(); err != nil { - return fmt.Errorf("close temp file: %w", err) - } - if err := os.Chmod(tmpPath, perm); err != nil { - return fmt.Errorf("chmod temp file: %w", err) - } - if err := os.Rename(tmpPath, path); err != nil { - return fmt.Errorf("rename temp file: %w", err) - } - removeTmp = false return nil } diff --git a/internal/fileops/fileops.go b/internal/fileops/fileops.go index bca653f..a7d0d8b 100644 --- a/internal/fileops/fileops.go +++ b/internal/fileops/fileops.go @@ -1,8 +1,10 @@ package fileops import ( + "bytes" "crypto/sha256" "encoding/hex" + "errors" "fmt" "io" "os" @@ -10,7 +12,32 @@ import ( "strings" ) -// WriteFileAtomic writes data to dst atomically via temp file + rename. +// ErrDirectorySyncUnsupported reports that the platform or filesystem cannot +// make a renamed directory entry crash-durable. The replacement may be visible, +// but callers must not treat it as having the same durability guarantee. +var ErrDirectorySyncUnsupported = errors.New("directory sync is unsupported") + +// ReplaceFileOptions describes the caller-owned parts of a byte-file +// replacement. The destination parent must already exist. BeforeRename runs +// after the temporary file is complete and durable, but before replacement is +// committed; callers can use it for cancellation or other commit checks. +type ReplaceFileOptions struct { + Mode os.FileMode + BeforeRename func() error +} + +// ReplaceFileAtomic durably replaces dst with data. It creates a sibling +// temporary file, writes and syncs its bytes and mode, atomically renames it, +// then syncs the destination parent. On platforms where replacement of an +// existing destination or directory syncing is unavailable, it returns an +// error rather than claiming an equivalent durability guarantee. +func ReplaceFileAtomic(dst string, data []byte, options ReplaceFileOptions) error { + return replaceFileAtomicWithOperations(dst, data, options, systemAtomicReplacementOperations) +} + +// WriteFileAtomic creates the destination parent with workspace permissions +// and durably replaces dst with data. Callers that own parent-directory policy +// or need a pre-commit check should use ReplaceFileAtomic directly. func WriteFileAtomic(dst string, data []byte, perm os.FileMode) error { if strings.TrimSpace(dst) == "" { return fmt.Errorf("destination path is required") @@ -18,48 +45,17 @@ func WriteFileAtomic(dst string, data []byte, perm os.FileMode) error { if err := EnsureWorkspaceDirectory(filepath.Dir(dst)); err != nil { return fmt.Errorf("create destination directory: %w", err) } - - base := filepath.Base(dst) - tmp, err := os.CreateTemp(filepath.Dir(dst), "."+base+".tmp-*") - if err != nil { - return fmt.Errorf("create temp file: %w", err) - } - tmpPath := tmp.Name() - removeTmp := true - defer func() { - if removeTmp { - _ = os.Remove(tmpPath) - } - }() - - if _, err := tmp.Write(data); err != nil { - _ = tmp.Close() - return fmt.Errorf("write temp file: %w", err) - } - if err := tmp.Sync(); err != nil { - _ = tmp.Close() - return fmt.Errorf("sync temp file: %w", err) - } - if err := tmp.Close(); err != nil { - return fmt.Errorf("close temp file: %w", err) - } - if err := os.Chmod(tmpPath, perm); err != nil { - return fmt.Errorf("set temp file permissions: %w", err) - } - if err := os.Rename(tmpPath, dst); err != nil { - return fmt.Errorf("install temp file: %w", err) - } - removeTmp = false - return nil + return ReplaceFileAtomic(dst, data, ReplaceFileOptions{Mode: perm}) } -// CopyFileAtomic copies src to dst atomically via temp file + rename. +// CopyFileAtomic copies src to dst through the durable replacement sequence. func CopyFileAtomic(src, dst string, perm os.FileMode) error { _, err := CopyFileAtomicWithChecksum(src, dst, perm) return err } -// CopyFileAtomicWithChecksum copies src to dst atomically and returns the SHA-256 checksum. +// CopyFileAtomicWithChecksum copies src to dst through the durable replacement +// sequence and returns the SHA-256 checksum of the copied bytes. func CopyFileAtomicWithChecksum(src, dst string, perm os.FileMode) (string, error) { if strings.TrimSpace(src) == "" || strings.TrimSpace(dst) == "" { return "", fmt.Errorf("source and destination paths are required") @@ -75,42 +71,22 @@ func CopyFileAtomicWithChecksum(src, dst string, perm os.FileMode) (string, erro return "", fmt.Errorf("create destination directory: %w", err) } - base := filepath.Base(dst) - tmp, err := os.CreateTemp(filepath.Dir(dst), "."+base+".tmp-*") - if err != nil { - return "", fmt.Errorf("create temp file: %w", err) - } - tmpPath := tmp.Name() - removeTmp := true - defer func() { - if removeTmp { - _ = os.Remove(tmpPath) - } - }() - digest := sha256.New() - if _, err := io.Copy(io.MultiWriter(tmp, digest), in); err != nil { - _ = tmp.Close() - return "", fmt.Errorf("copy file: %w", err) + err = replaceFileFromReaderWithOperations( + dst, + io.TeeReader(in, digest), + ReplaceFileOptions{Mode: perm}, + systemAtomicReplacementOperations, + ) + if err != nil { + return "", err } - if err := tmp.Sync(); err != nil { - _ = tmp.Close() - return "", fmt.Errorf("sync temp file: %w", err) - } - if err := tmp.Close(); err != nil { - return "", fmt.Errorf("close temp file: %w", err) - } - if err := os.Chmod(tmpPath, perm); err != nil { - return "", fmt.Errorf("set temp file permissions: %w", err) - } - if err := os.Rename(tmpPath, dst); err != nil { - return "", fmt.Errorf("install temp file: %w", err) - } - removeTmp = false return hex.EncodeToString(digest.Sum(nil)), nil } -// InstallDownloadedTempFile installs a previously downloaded temp file at dst. +// InstallDownloadedTempFile installs a fully downloaded sibling temporary file +// at dst. On failure before the rename, the caller retains ownership of tmpPath +// and is responsible for cleanup. func InstallDownloadedTempFile(tmpPath, dst string, perm os.FileMode) error { if strings.TrimSpace(tmpPath) == "" || strings.TrimSpace(dst) == "" { return fmt.Errorf("temp and destination paths are required") @@ -118,11 +94,137 @@ func InstallDownloadedTempFile(tmpPath, dst string, perm os.FileMode) error { if err := EnsureWorkspaceDirectory(filepath.Dir(dst)); err != nil { return fmt.Errorf("create destination directory: %w", err) } - if err := os.Chmod(tmpPath, perm); err != nil { - return fmt.Errorf("set temp file permissions: %w", err) + return installDownloadedTempFileWithOperations(tmpPath, dst, perm, systemAtomicReplacementOperations) +} + +type temporaryFile interface { + io.Writer + Name() string + Sync() error + Close() error +} + +type atomicReplacementOperations struct { + createTemp func(string, string) (temporaryFile, error) + chmod func(string, os.FileMode) error + rename func(string, string) error + remove func(string) error + syncDirectory func(string) error + syncFile func(string) error +} + +var systemAtomicReplacementOperations = atomicReplacementOperations{ + createTemp: func(dir, pattern string) (temporaryFile, error) { + return os.CreateTemp(dir, pattern) + }, + chmod: os.Chmod, + rename: os.Rename, + remove: os.Remove, + syncDirectory: syncDirectory, + syncFile: syncFile, +} + +func replaceFileAtomicWithOperations( + dst string, + data []byte, + options ReplaceFileOptions, + ops atomicReplacementOperations, +) error { + return replaceFileFromReaderWithOperations(dst, bytes.NewReader(data), options, ops) +} + +func replaceFileFromReaderWithOperations( + dst string, + source io.Reader, + options ReplaceFileOptions, + ops atomicReplacementOperations, +) (resultErr error) { + if strings.TrimSpace(dst) == "" { + return fmt.Errorf("destination path is required") } - if err := os.Rename(tmpPath, dst); err != nil { - return fmt.Errorf("install downloaded file: %w", err) + + dir := filepath.Dir(dst) + tmp, err := ops.createTemp(dir, "."+filepath.Base(dst)+".tmp-*") + if err != nil { + return fmt.Errorf("create temp file: %w", err) + } + tmpPath := tmp.Name() + removeTmp := true + defer func() { + if removeTmp { + if err := ops.remove(tmpPath); err != nil { + cleanupErr := fmt.Errorf("remove temporary file: %w", err) + if resultErr == nil { + resultErr = cleanupErr + } else { + resultErr = errors.Join(resultErr, cleanupErr) + } + } + } + }() + + if _, err := io.Copy(tmp, source); err != nil { + return closeTemporaryFileAfterError(tmp, fmt.Errorf("write temp file: %w", err)) + } + if err := ops.chmod(tmpPath, options.Mode); err != nil { + return closeTemporaryFileAfterError(tmp, fmt.Errorf("set temp file permissions: %w", err)) + } + if err := tmp.Sync(); err != nil { + return closeTemporaryFileAfterError(tmp, fmt.Errorf("sync temp file: %w", err)) + } + if err := tmp.Close(); err != nil { + return fmt.Errorf("close temp file: %w", err) + } + if options.BeforeRename != nil { + if err := options.BeforeRename(); err != nil { + return fmt.Errorf("check before file replacement: %w", err) + } + } + if err := ops.rename(tmpPath, dst); err != nil { + return fmt.Errorf("install temp file: %w", err) + } + removeTmp = false + if err := ops.syncDirectory(dir); err != nil { + return fmt.Errorf("sync destination directory after replacement: %w", err) } return nil } + +func closeTemporaryFileAfterError(tmp temporaryFile, operationErr error) error { + if err := tmp.Close(); err != nil { + return errors.Join(operationErr, fmt.Errorf("close temporary file: %w", err)) + } + return operationErr +} + +func installDownloadedTempFileWithOperations( + tmpPath, dst string, + perm os.FileMode, + ops atomicReplacementOperations, +) error { + if err := ops.chmod(tmpPath, perm); err != nil { + return fmt.Errorf("set temp file permissions: %w", err) + } + if err := ops.syncFile(tmpPath); err != nil { + return fmt.Errorf("sync downloaded temp file: %w", err) + } + if err := ops.rename(tmpPath, dst); err != nil { + return fmt.Errorf("install downloaded file: %w", err) + } + if err := ops.syncDirectory(filepath.Dir(dst)); err != nil { + return fmt.Errorf("sync destination directory after downloaded-file replacement: %w", err) + } + return nil +} + +func syncFile(path string) error { + file, err := os.Open(path) + if err != nil { + return err + } + if err := file.Sync(); err != nil { + _ = file.Close() + return err + } + return file.Close() +} diff --git a/internal/fileops/fileops_test.go b/internal/fileops/fileops_test.go index 07e1f6b..cdd796d 100644 --- a/internal/fileops/fileops_test.go +++ b/internal/fileops/fileops_test.go @@ -1,12 +1,16 @@ package fileops import ( + "errors" + "io" "os" "path/filepath" "strings" "testing" ) +var errAtomicReplacementTest = errors.New("atomic replacement test failure") + func TestWriteFileAtomicOverwritesAndLeavesNoTempFile(t *testing.T) { root := t.TempDir() dst := filepath.Join(root, "out", "value.txt") @@ -111,6 +115,231 @@ func TestInstallDownloadedTempFileSetsPermissions(t *testing.T) { } } +func TestReplaceFileAtomicOrdersDurableOperations(t *testing.T) { + events := make([]string, 0, 8) + tmp := &recordingTemporaryFile{name: "/work/.result.tmp-1", events: &events} + ops := recordingAtomicReplacementOperations(&events, tmp) + + err := replaceFileAtomicWithOperations( + "/work/result", + []byte("replacement"), + ReplaceFileOptions{ + Mode: 0o640, + BeforeRename: func() error { + events = append(events, "check") + return nil + }, + }, + ops, + ) + if err != nil { + t.Fatalf("replaceFileAtomicWithOperations() error = %v", err) + } + + want := []string{"create", "write", "chmod:rw-r-----", "sync", "close", "check", "rename", "sync-directory"} + if !equalStrings(events, want) { + t.Fatalf("operation order = %v, want %v", events, want) + } +} + +func TestReplaceFileAtomicFailuresCleanUninstalledTemporaryFile(t *testing.T) { + tests := []struct { + name string + configure func(*recordingTemporaryFile, *atomicReplacementOperations) + wantErr error + wantRemove bool + }{ + { + name: "short write", + configure: func(tmp *recordingTemporaryFile, _ *atomicReplacementOperations) { + tmp.shortWrite = true + }, + wantErr: io.ErrShortWrite, + wantRemove: true, + }, + { + name: "write", + configure: func(tmp *recordingTemporaryFile, _ *atomicReplacementOperations) { + tmp.writeErr = errAtomicReplacementTest + }, + wantErr: errAtomicReplacementTest, + wantRemove: true, + }, + { + name: "sync", + configure: func(tmp *recordingTemporaryFile, _ *atomicReplacementOperations) { + tmp.syncErr = errAtomicReplacementTest + }, + wantErr: errAtomicReplacementTest, + wantRemove: true, + }, + { + name: "rename", + configure: func(_ *recordingTemporaryFile, ops *atomicReplacementOperations) { + ops.rename = func(string, string) error { return errAtomicReplacementTest } + }, + wantErr: errAtomicReplacementTest, + wantRemove: true, + }, + { + name: "directory sync", + configure: func(_ *recordingTemporaryFile, ops *atomicReplacementOperations) { + ops.syncDirectory = func(string) error { return ErrDirectorySyncUnsupported } + }, + wantErr: ErrDirectorySyncUnsupported, + wantRemove: false, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + events := make([]string, 0, 8) + tmp := &recordingTemporaryFile{name: "/work/.result.tmp-1", events: &events} + ops := recordingAtomicReplacementOperations(&events, tmp) + test.configure(tmp, &ops) + + err := replaceFileAtomicWithOperations("/work/result", []byte("replacement"), ReplaceFileOptions{Mode: 0o640}, ops) + if !errors.Is(err, test.wantErr) { + t.Fatalf("replaceFileAtomicWithOperations() error = %v, want %v", err, test.wantErr) + } + gotRemove := containsString(events, "remove") + if gotRemove != test.wantRemove { + t.Fatalf("remove temporary file = %t, want %t; operations = %v", gotRemove, test.wantRemove, events) + } + }) + } +} + +func TestReplaceFileAtomicReportsCleanupFailureWithPrimaryFailure(t *testing.T) { + events := make([]string, 0, 5) + tmp := &recordingTemporaryFile{ + name: "/work/.result.tmp-1", + events: &events, + writeErr: errAtomicReplacementTest, + } + ops := recordingAtomicReplacementOperations(&events, tmp) + cleanupErr := errors.New("temporary cleanup failed") + ops.remove = func(string) error { + events = append(events, "remove") + return cleanupErr + } + + err := replaceFileAtomicWithOperations("/work/result", []byte("replacement"), ReplaceFileOptions{Mode: 0o640}, ops) + if !errors.Is(err, errAtomicReplacementTest) { + t.Fatalf("replacement error = %v, want write failure", err) + } + if !errors.Is(err, cleanupErr) { + t.Fatalf("replacement error = %v, want cleanup failure", err) + } +} + +func TestInstallDownloadedTempFileOrdersDurableOperations(t *testing.T) { + events := make([]string, 0, 4) + ops := atomicReplacementOperations{ + chmod: func(string, os.FileMode) error { + events = append(events, "chmod") + return nil + }, + syncFile: func(string) error { + events = append(events, "sync-file") + return nil + }, + rename: func(string, string) error { + events = append(events, "rename") + return nil + }, + syncDirectory: func(string) error { + events = append(events, "sync-directory") + return nil + }, + } + + if err := installDownloadedTempFileWithOperations("/work/.download.tmp", "/work/result", 0o640, ops); err != nil { + t.Fatalf("installDownloadedTempFileWithOperations() error = %v", err) + } + want := []string{"chmod", "sync-file", "rename", "sync-directory"} + if !equalStrings(events, want) { + t.Fatalf("operation order = %v, want %v", events, want) + } +} + +type recordingTemporaryFile struct { + name string + events *[]string + shortWrite bool + writeErr error + syncErr error +} + +func (f *recordingTemporaryFile) Write(data []byte) (int, error) { + *f.events = append(*f.events, "write") + if f.writeErr != nil { + return 0, f.writeErr + } + if f.shortWrite { + return len(data) - 1, nil + } + return len(data), nil +} + +func (f *recordingTemporaryFile) Name() string { return f.name } + +func (f *recordingTemporaryFile) Sync() error { + *f.events = append(*f.events, "sync") + return f.syncErr +} + +func (f *recordingTemporaryFile) Close() error { + *f.events = append(*f.events, "close") + return nil +} + +func recordingAtomicReplacementOperations(events *[]string, tmp temporaryFile) atomicReplacementOperations { + return atomicReplacementOperations{ + createTemp: func(string, string) (temporaryFile, error) { + *events = append(*events, "create") + return tmp, nil + }, + chmod: func(_ string, mode os.FileMode) error { + *events = append(*events, "chmod:"+mode.Perm().String()[1:]) + return nil + }, + rename: func(string, string) error { + *events = append(*events, "rename") + return nil + }, + remove: func(string) error { + *events = append(*events, "remove") + return nil + }, + syncDirectory: func(string) error { + *events = append(*events, "sync-directory") + return nil + }, + } +} + +func containsString(values []string, target string) bool { + for _, value := range values { + if value == target { + return true + } + } + return false +} + +func equalStrings(got, want []string) bool { + if len(got) != len(want) { + return false + } + for i := range got { + if got[i] != want[i] { + return false + } + } + return true +} + func assertNoMatchingTempFiles(t *testing.T, dir, prefix string) { t.Helper() entries, err := os.ReadDir(dir) diff --git a/internal/fileops/sync_directory_unix.go b/internal/fileops/sync_directory_unix.go index d5df543..2b00587 100644 --- a/internal/fileops/sync_directory_unix.go +++ b/internal/fileops/sync_directory_unix.go @@ -4,6 +4,7 @@ package fileops import ( "errors" + "fmt" "os" "syscall" ) @@ -16,10 +17,10 @@ func syncDirectory(path string) error { defer func() { _ = directory.Close() }() err = directory.Sync() - // Some Unix filesystems do not implement directory syncing. Only their - // explicit unsupported-operation errors are safe to treat as best effort. + // Some Unix filesystems do not implement directory syncing. Report this + // explicitly so callers do not confuse visible replacement with a durable one. if errors.Is(err, syscall.EINVAL) || errors.Is(err, syscall.ENOTSUP) { - return nil + return fmt.Errorf("%w for %q: %w", ErrDirectorySyncUnsupported, path, err) } return err } diff --git a/internal/fileops/sync_directory_unsupported.go b/internal/fileops/sync_directory_unsupported.go index 8fd8ccb..300d35e 100644 --- a/internal/fileops/sync_directory_unsupported.go +++ b/internal/fileops/sync_directory_unsupported.go @@ -2,6 +2,11 @@ package fileops -func syncDirectory(string) error { - return checkAtomicDirectoryPromotionSupport() +import ( + "fmt" + "runtime" +) + +func syncDirectory(path string) error { + return fmt.Errorf("%w for %q on %s", ErrDirectorySyncUnsupported, path, runtime.GOOS) } diff --git a/internal/fileops/sync_directory_windows.go b/internal/fileops/sync_directory_windows.go index 8a41421..693bbb9 100644 --- a/internal/fileops/sync_directory_windows.go +++ b/internal/fileops/sync_directory_windows.go @@ -4,6 +4,7 @@ package fileops import ( "errors" + "fmt" "golang.org/x/sys/windows" ) @@ -29,12 +30,12 @@ func syncDirectory(path string) error { err = windows.FlushFileBuffers(directory) // Windows filesystems may reject flushing a directory handle even when it - // was opened correctly. Preserve every error except the documented forms - // that mean this operation is unavailable for the handle or filesystem. + // was opened correctly. Report documented unavailable forms explicitly so + // callers do not confuse visible replacement with a durable one. if errors.Is(err, windows.ERROR_INVALID_FUNCTION) || errors.Is(err, windows.ERROR_INVALID_HANDLE) || errors.Is(err, windows.ERROR_NOT_SUPPORTED) { - return nil + return fmt.Errorf("%w for %q: %w", ErrDirectorySyncUnsupported, path, err) } return err } diff --git a/internal/manifest/store.go b/internal/manifest/store.go index 7c58618..0b64df7 100644 --- a/internal/manifest/store.go +++ b/internal/manifest/store.go @@ -63,7 +63,7 @@ func (s *LocalStore) Load(ctx context.Context, path string) (*Manifest, error) { return &m, nil } -// Save writes the manifest to path atomically via temp file + rename. +// Save writes the manifest to path through the durable file replacement primitive. func (s *LocalStore) Save(ctx context.Context, path string, m *Manifest) error { if err := checkContext(ctx); err != nil { return err @@ -95,46 +95,9 @@ func (s *LocalStore) Save(ctx context.Context, path string, m *Manifest) error { } data = append(data, '\n') - dir := filepath.Dir(path) - if err := fileops.EnsureWorkspaceDirectory(dir); err != nil { - return fmt.Errorf("save manifest: create directory %q: %w", dir, err) + if err := writeJSONAtomically(ctx, path, data); err != nil { + return fmt.Errorf("save manifest: %w", err) } - - tmp, err := os.CreateTemp(dir, ".manifest.json.tmp-*") - if err != nil { - return fmt.Errorf("save manifest: create temp file: %w", err) - } - tmpName := tmp.Name() - removeTmp := true - defer func() { - if removeTmp { - _ = os.Remove(tmpName) - } - }() - - if _, err := tmp.Write(data); err != nil { - _ = tmp.Close() - return fmt.Errorf("save manifest: write temp file: %w", err) - } - if err := tmp.Sync(); err != nil { - _ = tmp.Close() - return fmt.Errorf("save manifest: sync temp file: %w", err) - } - if err := tmp.Close(); err != nil { - return fmt.Errorf("save manifest: close temp file: %w", err) - } - if err := os.Chmod(tmpName, fileops.WorkspaceFileMode); err != nil { - return fmt.Errorf("save manifest: set temp file permissions: %w", err) - } - if err := checkContext(ctx); err != nil { - return err - } - - if err := os.Rename(tmpName, path); err != nil { - return fmt.Errorf("save manifest: rename temp file: %w", err) - } - removeTmp = false - return nil } @@ -221,7 +184,7 @@ func (s *LocalStore) SaveRun(ctx context.Context, path string, m *RunManifest) e } data = append(data, '\n') - return writeJSONAtomically(ctx, path, ".run-manifest.json.tmp-*", data) + return writeJSONAtomically(ctx, path, data) } func validateLoadedManifest(m *Manifest) error { @@ -330,48 +293,17 @@ func normalizeRunManifest(m *RunManifest) { } } -func writeJSONAtomically(ctx context.Context, path, tempPattern string, data []byte) error { +func writeJSONAtomically(ctx context.Context, path string, data []byte) error { dir := filepath.Dir(path) if err := fileops.EnsureWorkspaceDirectory(dir); err != nil { return fmt.Errorf("create directory %q: %w", dir, err) } - - tmp, err := os.CreateTemp(dir, tempPattern) - if err != nil { - return fmt.Errorf("create temp file: %w", err) - } - tmpName := tmp.Name() - removeTmp := true - defer func() { - if removeTmp { - _ = os.Remove(tmpName) - } - }() - - if _, err := tmp.Write(data); err != nil { - _ = tmp.Close() - return fmt.Errorf("write temp file: %w", err) - } - if err := tmp.Sync(); err != nil { - _ = tmp.Close() - return fmt.Errorf("sync temp file: %w", err) - } - if err := tmp.Close(); err != nil { - return fmt.Errorf("close temp file: %w", err) - } - if err := os.Chmod(tmpName, fileops.WorkspaceFileMode); err != nil { - return fmt.Errorf("set temp file permissions: %w", err) - } - if err := checkContext(ctx); err != nil { - return err - } - - if err := os.Rename(tmpName, path); err != nil { - return fmt.Errorf("rename temp file: %w", err) - } - removeTmp = false - - return nil + return fileops.ReplaceFileAtomic(path, data, fileops.ReplaceFileOptions{ + Mode: fileops.WorkspaceFileMode, + BeforeRename: func() error { + return checkContext(ctx) + }, + }) } func checkContext(ctx context.Context) error { diff --git a/internal/manifest/store_test.go b/internal/manifest/store_test.go index f7f79ac..42ad5b7 100644 --- a/internal/manifest/store_test.go +++ b/internal/manifest/store_test.go @@ -3,6 +3,7 @@ package manifest import ( "context" "encoding/json" + "errors" "os" "path/filepath" "strings" @@ -205,6 +206,30 @@ func TestLocalStoreSaveAtomicPractical(t *testing.T) { } } +func TestWriteJSONAtomicallyChecksCancellationBeforeReplacement(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + dir := t.TempDir() + path := filepath.Join(dir, "manifest.json") + err := writeJSONAtomically(ctx, path, []byte("{\"session_id\":\"session\"}\n")) + if !errors.Is(err, context.Canceled) { + t.Fatalf("writeJSONAtomically() error = %v, want context cancellation", err) + } + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Fatalf("replacement destination exists after cancellation: stat err = %v", err) + } + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("ReadDir() error = %v", err) + } + for _, entry := range entries { + if strings.HasPrefix(entry.Name(), ".manifest.json.tmp-") { + t.Fatalf("temporary manifest remained after cancellation: %s", entry.Name()) + } + } +} + func TestLoadRejectsInvalidManifest(t *testing.T) { store := &LocalStore{} ctx := context.Background()