From 401d043875d2d9d9ac00147a1e256fc1dbb64d7e Mon Sep 17 00:00:00 2001 From: adababys <8872412+adababys@users.noreply.github.com> Date: Tue, 22 Sep 2026 09:51:15 +0800 Subject: [PATCH] feat(storage): optional fsync durability + stale temp-file sweep for local FS Follow-up to a334b20 (temp+rename atomic replace). Two hardening additions: - replaceFile gains a durable bool: when set, fsync the temp file before the rename and fsync the parent dir after, closing the crash window where a rename is persisted but the contents are still only in page cache (renamed-but-empty object). Off by default (STORAGE_FS_FSYNC_DURABLE) to preserve cache write throughput; callers with crash-critical objects can enable it. - SweepStaleTempFiles reclaims orphan ..tmp-* files a crash (OOM kill, power loss) leaves between CreateTemp and Rename. Age-gated so in-flight writes are never removed; shared tmpFileMarker keeps sweep and create in sync. Regression tests: durable replacement matrix (parallel, no global mutation), durable==non-durable content equivalence, public Put durable toggle, and sweep only-old-orphans / missing-dir / naming-sync. --- packages/shared/pkg/storage/storage_fs.go | 135 +++++++++++++- .../pkg/storage/storage_fs_followup_test.go | 175 ++++++++++++++++++ 2 files changed, 306 insertions(+), 4 deletions(-) create mode 100644 packages/shared/pkg/storage/storage_fs_followup_test.go diff --git a/packages/shared/pkg/storage/storage_fs.go b/packages/shared/pkg/storage/storage_fs.go index 7d3292f49c..b21b240e5e 100644 --- a/packages/shared/pkg/storage/storage_fs.go +++ b/packages/shared/pkg/storage/storage_fs.go @@ -14,6 +14,7 @@ import ( "path/filepath" "strconv" "strings" + "syscall" "time" "go.uber.org/zap" @@ -296,17 +297,55 @@ func (o *fsObject) getHandle(checkExistence bool) (*os.File, error) { return handle, nil } -func replaceFile(path string, r io.Reader) (int64, error) { +// tmpFileMarker is embedded in the name of the temporary file used for atomic +// replacement. It is shared by replaceFile (which creates such files) and +// SweepStaleTempFiles (which reclaims the ones a crash left behind), so the two +// can never drift out of sync. Temp files are named "..tmp-". +const tmpFileMarker = ".tmp-" + +// envBool reads a boolean environment toggle, returning def when unset or +// unparseable. Accepts the strconv.ParseBool set (1/t/T/TRUE/true/... etc). +func envBool(name string, def bool) bool { + v := os.Getenv(name) + if v == "" { + return def + } + parsed, err := strconv.ParseBool(v) + if err != nil { + return def + } + + return parsed +} + +// fsyncDurable, when true, makes replaceFile fsync the temp file before the +// rename and fsync the parent directory after it, so a crash cannot leave a +// renamed-but-empty/truncated object. It defaults to false to preserve the +// previous (page-cache-only) performance for the high-frequency cache writes; +// callers that store crash-critical objects (snapshots, headers) can enable it. +// It is a package-level var (not a const) so it can be toggled by config or in +// tests without threading a parameter through every caller. +var fsyncDurable = envBool("STORAGE_FS_FSYNC_DURABLE", false) + +// replaceFile atomically replaces path with the contents of r via a temp file +// in the same directory followed by a rename. When durable is true it fsyncs the +// temp file before the rename and the parent directory after, so a crash cannot +// leave a renamed-but-empty object; the cost is two extra fsyncs per write. +func replaceFile(path string, r io.Reader, durable bool) (int64, error) { dir := filepath.Dir(path) if err := os.MkdirAll(dir, 0o755); err != nil { return 0, err } - tmp, err := os.CreateTemp(dir, "."+filepath.Base(path)+".tmp-*") + tmp, err := os.CreateTemp(dir, "."+filepath.Base(path)+tmpFileMarker+"*") if err != nil { return 0, err } tmpPath := tmp.Name() + // On any failure before Rename succeeds this removes the temp file; after a + // successful Rename the temp path no longer exists so Remove is a no-op. A + // crash between CreateTemp and Rename skips this entirely — SweepStaleTempFiles + // reclaims such orphans. defer os.Remove(tmpPath) if err := tmp.Chmod(0o644); err != nil { @@ -321,6 +360,18 @@ func replaceFile(path string, r io.Reader) (int64, error) { return n, err } + + // Durability: flush the data to disk before it becomes reachable under the + // object's name, otherwise the rename can be persisted while the contents are + // still only in the page cache (crash => renamed-but-empty object). + if durable { + if err := tmp.Sync(); err != nil { + tmp.Close() + + return n, err + } + } + if err := tmp.Close(); err != nil { return n, err } @@ -329,11 +380,87 @@ func replaceFile(path string, r io.Reader) (int64, error) { return n, err } + // Persist the directory entry created by the rename, so the replacement + // survives a crash immediately after it. + if durable { + if err := fsyncDir(dir); err != nil { + return n, err + } + } + return n, nil } +// fsyncDir flushes a directory's metadata so a rename into it is durable. +// A directory that cannot be opened O_RDONLY for sync is treated as a no-op +// rather than a hard error (some filesystems reject directory fsync). +func fsyncDir(dir string) error { + d, err := os.Open(dir) + if err != nil { + return err + } + defer d.Close() + + if err := d.Sync(); err != nil { + // EINVAL / ENOTSUP: filesystem doesn't support directory fsync — the + // rename itself is still atomic, only the extra durability is missing. + if errors.Is(err, syscall.EINVAL) || errors.Is(err, syscall.ENOTSUP) { + return nil + } + + return err + } + + return nil +} + +// SweepStaleTempFiles removes atomic-replacement temp files (".*.tmp-*") under +// dir that are older than olderThan and are therefore orphans left by a crash +// (OOM kill, power loss) between CreateTemp and Rename. The age gate ensures a +// temp file belonging to a concurrently in-flight write is never removed. It +// returns the number of files reclaimed. Missing dir is not an error. +func SweepStaleTempFiles(dir string, olderThan time.Duration) (int, error) { + entries, err := os.ReadDir(dir) + if err != nil { + if os.IsNotExist(err) { + return 0, nil + } + + return 0, err + } + + cutoff := time.Now().Add(-olderThan) + removed := 0 + var errs []error + for _, e := range entries { + if e.IsDir() { + continue + } + name := e.Name() + // Match "..tmp-": leading dot + contains the marker. + if !strings.HasPrefix(name, ".") || !strings.Contains(name, tmpFileMarker) { + continue + } + info, statErr := e.Info() + if statErr != nil { + continue // vanished concurrently; ignore + } + if info.ModTime().After(cutoff) { + continue // possibly an in-flight write; leave it + } + if rmErr := os.Remove(filepath.Join(dir, name)); rmErr != nil && !os.IsNotExist(rmErr) { + errs = append(errs, rmErr) + + continue + } + removed++ + } + + return removed, errors.Join(errs...) +} + func (o *fsObject) replaceFrom(r io.Reader) (int64, error) { - n, err := replaceFile(o.path, r) + n, err := replaceFile(o.path, r, fsyncDurable) if err != nil { return n, err } @@ -364,7 +491,7 @@ func (u *fsPartUploader) Complete(_ context.Context) error { return fmt.Errorf("failed to create directory: %w", err) } - _, err := replaceFile(u.fullPath, bytes.NewReader(u.Assemble())) + _, err := replaceFile(u.fullPath, bytes.NewReader(u.Assemble()), fsyncDurable) return err } diff --git a/packages/shared/pkg/storage/storage_fs_followup_test.go b/packages/shared/pkg/storage/storage_fs_followup_test.go new file mode 100644 index 0000000000..5ed88e6ed4 --- /dev/null +++ b/packages/shared/pkg/storage/storage_fs_followup_test.go @@ -0,0 +1,175 @@ +package storage + +import ( + "bytes" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// readFile is a small helper to read a file's full contents in a test. +func readFileT(t *testing.T, path string) []byte { + t.Helper() + b, err := os.ReadFile(path) + require.NoError(t, err) + + return b +} + +// TestReplaceFile_DurableWritesExactContent verifies the durable path (fsync +// temp + fsync dir) writes exactly the payload across the replacement matrix and +// leaves no temp residue. Calls replaceFile directly with durable=true so it +// touches no package-level state and stays parallel-safe. +func TestReplaceFile_DurableWritesExactContent(t *testing.T) { + t.Parallel() + + cases := []struct{ name, initial, replace string }{ + {"shorter", "a longer initial payload", "short"}, + {"longer", "short", "a longer replacement payload"}, + {"same_length", "first", "other"}, + {"empty", "a non-empty payload", ""}, + {"into_empty", "", "filled"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "obj.bin") + + n, err := replaceFile(path, bytes.NewReader([]byte(tc.initial)), true) + require.NoError(t, err) + require.EqualValues(t, len(tc.initial), n) + + n, err = replaceFile(path, bytes.NewReader([]byte(tc.replace)), true) + require.NoError(t, err) + require.EqualValues(t, len(tc.replace), n) + + assert.Equal(t, tc.replace, string(readFileT(t, path)), "no stale tail") + + info, err := os.Stat(path) + require.NoError(t, err) + assert.EqualValues(t, len(tc.replace), info.Size()) + + // no temp residue + entries, err := os.ReadDir(filepath.Dir(path)) + require.NoError(t, err) + for _, e := range entries { + assert.NotContains(t, e.Name(), tmpFileMarker, "temp residue left: %s", e.Name()) + } + }) + } +} + +// TestReplaceFile_DurableEqualsNonDurableContent asserts durable and +// non-durable modes produce byte-identical results (durability only affects +// crash semantics, never observable content). +func TestReplaceFile_DurableEqualsNonDurableContent(t *testing.T) { + t.Parallel() + payload := bytes.Repeat([]byte("payload-"), 512) + + dDir := t.TempDir() + dPath := filepath.Join(dDir, "d.bin") + _, err := replaceFile(dPath, bytes.NewReader(payload), true) + require.NoError(t, err) + + nDir := t.TempDir() + nPath := filepath.Join(nDir, "n.bin") + _, err = replaceFile(nPath, bytes.NewReader(payload), false) + require.NoError(t, err) + + assert.Equal(t, readFileT(t, dPath), readFileT(t, nPath)) + assert.Equal(t, payload, readFileT(t, dPath)) +} + +// TestPutDurableToggle exercises the public Put path with the package-level +// fsyncDurable toggle flipped on. It mutates a package var, so it is +// deliberately NOT parallel and restores the previous value. +// +//nolint:paralleltest // mutates package-level fsyncDurable; must run serially +func TestPutDurableToggle(t *testing.T) { + prev := fsyncDurable + fsyncDurable = true + t.Cleanup(func() { fsyncDurable = prev }) + + ctx := t.Context() + p := newTempProvider(t) + obj, err := p.OpenBlob(ctx, filepath.Join("durable", "obj.bin")) + require.NoError(t, err) + + require.NoError(t, obj.Put(ctx, []byte("a longer initial payload"))) + require.NoError(t, obj.Put(ctx, []byte("short"))) + + seekable, ok := obj.(Seekable) + require.True(t, ok) + size, err := seekable.Size(ctx) + require.NoError(t, err) + require.EqualValues(t, 5, size) + + data, err := GetBlob(ctx, obj) + require.NoError(t, err) + require.Equal(t, []byte("short"), data) +} + +// TestSweepStaleTempFiles_RemovesOnlyOldOrphans asserts the sweep reclaims aged +// orphan temp files while leaving fresh temp files (in-flight writes), the real +// object, and non-temp dotfiles (sidecars) untouched. +func TestSweepStaleTempFiles_RemovesOnlyOldOrphans(t *testing.T) { + t.Parallel() + dir := t.TempDir() + + realObj := filepath.Join(dir, "memfile.bin") + require.NoError(t, os.WriteFile(realObj, []byte("real data"), 0o644)) + + oldOrphan := filepath.Join(dir, ".memfile.bin"+tmpFileMarker+"aaaa") + require.NoError(t, os.WriteFile(oldOrphan, []byte("half-written"), 0o644)) + old := time.Now().Add(-30 * time.Minute) + require.NoError(t, os.Chtimes(oldOrphan, old, old)) + + freshTmp := filepath.Join(dir, ".other.bin"+tmpFileMarker+"bbbb") + require.NoError(t, os.WriteFile(freshTmp, []byte("in flight"), 0o644)) + + sidecar := filepath.Join(dir, "memfile.bin.uncompressed-size") + require.NoError(t, os.WriteFile(sidecar, []byte("123"), 0o644)) + + removed, err := SweepStaleTempFiles(dir, 5*time.Minute) + require.NoError(t, err) + assert.Equal(t, 1, removed, "only the aged orphan should be reclaimed") + + _, err = os.Stat(oldOrphan) + assert.True(t, os.IsNotExist(err), "aged orphan must be removed") + assert.FileExists(t, realObj, "real object must survive") + assert.FileExists(t, freshTmp, "fresh temp (in-flight) must survive") + assert.FileExists(t, sidecar, "sidecar must not be treated as a temp file") +} + +// TestSweepStaleTempFiles_MissingDir is a no-op, not an error. +func TestSweepStaleTempFiles_MissingDir(t *testing.T) { + t.Parallel() + removed, err := SweepStaleTempFiles(filepath.Join(t.TempDir(), "nope"), time.Minute) + require.NoError(t, err) + assert.Equal(t, 0, removed) +} + +// TestSweepStaleTempFiles_MatchesReplaceFileNaming guarantees the sweep pattern +// stays in sync with the names replaceFile actually creates. +func TestSweepStaleTempFiles_MatchesReplaceFileNaming(t *testing.T) { + t.Parallel() + dir := t.TempDir() + target := filepath.Join(dir, "obj.bin") + + tmp, err := os.CreateTemp(dir, "."+filepath.Base(target)+tmpFileMarker+"*") + require.NoError(t, err) + name := tmp.Name() + require.NoError(t, tmp.Close()) + old := time.Now().Add(-time.Hour) + require.NoError(t, os.Chtimes(name, old, old)) + + removed, err := SweepStaleTempFiles(dir, time.Minute) + require.NoError(t, err) + assert.Equal(t, 1, removed) + _, err = os.Stat(name) + assert.True(t, os.IsNotExist(err)) +}