eric/narratio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 15 Wiki Activity

Make atomic file replacement crash durable

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-10 17:44:38 +00:00
parent 1dccf5f140
commit 59f3fe3d1d
13 changed files with 502 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
docs/internal/fileops.md Normal file
View File

@@ -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.

8
docs/internal/manifest.md
View File

@@ -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

2
docs/internal/overview.md
View File

@@ -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

2
docs/roadmap/implementation.md
View File

@@ -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 |

40
internal/adapters/subprocess/run.go
View File

@@ -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
}

37
internal/adapters/whisperx/http.go
View File

@@ -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
}

244
internal/fileops/fileops.go
View File

@@ -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()
}

229
internal/fileops/fileops_test.go
View File

@@ -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)

7
internal/fileops/sync_directory_unix.go
View File

@@ -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
}

9
internal/fileops/sync_directory_unsupported.go
View File

@@ -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)
}

7
internal/fileops/sync_directory_windows.go
View File

@@ -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
}

90
internal/manifest/store.go
View File

@@ -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 {

25
internal/manifest/store_test.go
View File

@@ -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()