feat(storage): optional fsync durability + stale temp-file sweep for local FS (follow-up to #3649)

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 ".<object>.tmp-<random>".
+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 ".<object>.tmp-<random>": 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
 }

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