perf(store): move Update fsyncs out of the flock hold

storage/json/json.go

@@ -2,8 +2,12 @@ package json
 
 import (
 	"context"
+	stdjson "encoding/json"
 	"errors"
+	"fmt"
 	"io/fs"
+	"os"
+	"path/filepath"
 
 	"github.com/projecteru2/core/log"
 
@@ -12,6 +16,8 @@ import (
 	"github.com/cocoonstack/cocoon/utils"
 )
 
+const prevSuffix = ".prev"
+
 var _ storage.Store[struct{}] = (*Store[struct{}])(nil)
 
 // Store is a lock-guarded JSON file holding one value of type T.
@@ -27,7 +33,7 @@ func New[T any](filePath string, locker lock.Locker) *Store[T] {
 
 // ReadRaw loads the JSON file unlocked.
 func (s *Store[T]) ReadRaw(fn func(*T) error) error {
-	data, err := s.load()
+	data, _, err := s.load()
 	if err != nil {
 		return err
 	}
@@ -44,14 +50,27 @@ func (s *Store[T]) With(ctx context.Context, fn func(*T) error) error {
 	return s.withLocked(ctx, func() error { return s.ReadRaw(fn) })
 }
 
-// Update runs fn read-modify-write under the store lock.
+// Update runs fn read-modify-write under the store lock; all fsyncs run after release so waiters never queue behind stable-storage flushes, and load falls back to the .prev generation if a crash tears the unsynced rename.
 func (s *Store[T]) Update(ctx context.Context, fn func(*T) error) error {
-	return s.withLocked(ctx, func() error { return s.writeRaw(fn, utils.Sync) })
+	if err := s.UpdateNoDirSync(ctx, fn); err != nil {
+		return err
+	}
+	return utils.SyncParentDir(filepath.Dir(s.filePath))
 }
 
-// UpdateNoDirSync runs fn read-modify-write under the store lock, fsyncing the file but not the parent dir.
+// UpdateNoDirSync is Update without the parent-dir fsync, for placeholder states GC re-derives after power loss.
 func (s *Store[T]) UpdateNoDirSync(ctx context.Context, fn func(*T) error) error {
-	return s.withLocked(ctx, func() error { return s.writeRaw(fn, utils.NoSync) })
+	if err := s.withLocked(ctx, func() error { return s.writeRaw(fn, utils.NoSync) }); err != nil {
+		return err
+	}
+	// Main first so a .prev sync failure still leaves the caller's own generation durable; syncing .prev too closes the double-tear window when the previous writer died before its own post-release fsync, near-free on an already-durable inode.
+	if err := utils.SyncFile(s.filePath); err != nil {
+		return err
+	}
+	if err := utils.SyncFile(s.filePath + prevSuffix); err != nil && !errors.Is(err, fs.ErrNotExist) {
+		return err
+	}
+	return nil
 }
 
 func (s *Store[T]) TryLock(ctx context.Context) (bool, error) {
@@ -63,7 +82,7 @@ func (s *Store[T]) Unlock(ctx context.Context) error {
 }
 
 func (s *Store[T]) writeRaw(fn func(*T) error, sync utils.SyncMode) error {
-	data, err := s.load()
+	data, recovered, err := s.load()
 	if err != nil {
 		return err
 	}
@@ -73,16 +92,47 @@ func (s *Store[T]) writeRaw(fn func(*T) error, sync utils.SyncMode) error {
 	if sync == utils.Sync {
 		return utils.AtomicWriteJSON(s.filePath, data)
 	}
-	return utils.AtomicWriteJSONNoDirSync(s.filePath, data)
+	// Keep the durable previous generation reachable: the caller fsyncs only after release, and a crash in between may tear the fresh file on filesystems without rename-over data ordering. A recovered main is the torn one — rotating it in would destroy the only good generation — and the rotation goes link+rename so a .prev exists at every instant.
+	if !recovered {
+		tmp := s.filePath + prevSuffix + ".tmp"
+		if err := os.Remove(tmp); err != nil && !errors.Is(err, fs.ErrNotExist) {
+			return err
+		}
+		switch err := os.Link(s.filePath, tmp); {
+		case errors.Is(err, fs.ErrNotExist):
+		case err != nil:
+			return err
+		default:
+			if err := os.Rename(tmp, s.filePath+prevSuffix); err != nil {
+				return err
+			}
+		}
+	}
+	return utils.AtomicWriteJSONNoSync(s.filePath, data)
 }
 
-func (s *Store[T]) load() (*T, error) {
+// load reads the store; recovered reports that main was undecodable and the .prev generation was served instead. Read errors fail closed — only a decode failure of successfully read bytes is eligible for fallback.
+func (s *Store[T]) load() (*T, bool, error) {
 	var data T
-	if err := utils.ReadJSONFile(s.filePath, &data); err != nil && !errors.Is(err, fs.ErrNotExist) {
-		return nil, err
+	recovered := false
+	raw, err := os.ReadFile(s.filePath) //nolint:gosec
+	switch {
+	case errors.Is(err, fs.ErrNotExist):
+	case err != nil:
+		return nil, false, fmt.Errorf("read %s: %w", s.filePath, err)
+	default:
+		if decodeErr := stdjson.Unmarshal(raw, &data); decodeErr != nil {
+			var prev T
+			if prevErr := utils.ReadJSONFile(s.filePath+prevSuffix, &prev); prevErr != nil {
+				return nil, false, errors.Join(fmt.Errorf("decode %s: %w", s.filePath, decodeErr), prevErr)
+			}
+			data = prev
+			recovered = true
+			log.WithFunc("storage.json.load").Warnf(context.Background(), "%s undecodable (%v); recovered the previous generation", s.filePath, decodeErr)
+		}
 	}
 	initData(&data)
-	return &data, nil
+	return &data, recovered, nil
 }
 
 func (s *Store[T]) withLocked(ctx context.Context, fn func() error) error {

storage/json/json_test.go

@@ -182,3 +182,112 @@ func newTestStore(t *testing.T, dir, name string) *Store[testData] {
 	lockPath := filepath.Join(dir, name+".lock")
 	return New[testData](dataPath, flock.New(lockPath))
 }
+
+func TestLoadRecoversPrevGeneration(t *testing.T) {
+	dir := t.TempDir()
+	dataPath := filepath.Join(dir, "data.json")
+	s := New[testData](dataPath, flock.New(filepath.Join(dir, "data.lock")))
+
+	for i := 1; i <= 2; i++ {
+		if err := s.Update(t.Context(), func(d *testData) error {
+			d.Count = i
+			return nil
+		}); err != nil {
+			t.Fatalf("update %d: %v", i, err)
+		}
+	}
+	if err := os.WriteFile(dataPath, []byte("{torn"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := s.ReadRaw(func(d *testData) error {
+		if d.Count != 1 {
+			t.Fatalf("recovered Count = %d, want previous generation 1", d.Count)
+		}
+		return nil
+	}); err != nil {
+		t.Fatalf("read after tear: %v", err)
+	}
+}
+
+func TestUpdateAfterRecoveryKeepsGoodPrev(t *testing.T) {
+	dir := t.TempDir()
+	dataPath := filepath.Join(dir, "data.json")
+	s := New[testData](dataPath, flock.New(filepath.Join(dir, "data.lock")))
+
+	for i := 1; i <= 2; i++ {
+		if err := s.Update(t.Context(), func(d *testData) error {
+			d.Count = i
+			return nil
+		}); err != nil {
+			t.Fatalf("update %d: %v", i, err)
+		}
+	}
+	if err := os.WriteFile(dataPath, []byte("{torn"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+
+	// The recovery update must not rotate the torn main into .prev.
+	if err := s.Update(t.Context(), func(d *testData) error {
+		d.Count += 10
+		return nil
+	}); err != nil {
+		t.Fatalf("recovery update: %v", err)
+	}
+	if err := s.ReadRaw(func(d *testData) error {
+		if d.Count != 11 {
+			t.Fatalf("post-recovery Count = %d, want 11", d.Count)
+		}
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	// A second tear must still find a decodable generation behind it.
+	if err := os.WriteFile(dataPath, []byte("{torn again"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.ReadRaw(func(d *testData) error {
+		if d.Count != 1 {
+			t.Fatalf("second recovery Count = %d, want 1", d.Count)
+		}
+		return nil
+	}); err != nil {
+		t.Fatalf("second recovery: %v", err)
+	}
+}
+
+func TestLoadReadErrorFailsClosed(t *testing.T) {
+	dir := t.TempDir()
+	dataPath := filepath.Join(dir, "data.json")
+	s := New[testData](dataPath, flock.New(filepath.Join(dir, "data.lock")))
+
+	if err := s.Update(t.Context(), func(d *testData) error { d.Count = 1; return nil }); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Update(t.Context(), func(d *testData) error { d.Count = 2; return nil }); err != nil {
+		t.Fatal(err)
+	}
+	// Replace main with a directory: the read itself fails, so .prev must NOT be served.
+	if err := os.Remove(dataPath); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Mkdir(dataPath, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.ReadRaw(func(*testData) error { return nil }); err == nil {
+		t.Fatal("want error for unreadable main even with a valid .prev")
+	}
+}
+
+func TestLoadTornWithoutPrevFails(t *testing.T) {
+	dir := t.TempDir()
+	dataPath := filepath.Join(dir, "data.json")
+	if err := os.WriteFile(dataPath, []byte("{torn"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	s := New[testData](dataPath, flock.New(filepath.Join(dir, "data.lock")))
+	if err := s.ReadRaw(func(*testData) error { return nil }); err == nil {
+		t.Fatal("want error for torn file without a previous generation")
+	}
+}

storage/storage.go

@@ -13,9 +13,9 @@ type Initer interface {
 type Store[T any] interface {
 	// With loads under lock and passes to fn; *T's Init() runs first if implemented; lock held for fn's duration.
 	With(ctx context.Context, fn func(*T) error) error
-	// Update performs a read-modify-write under lock; persists only if fn returns nil.
+	// Update performs a read-modify-write under lock; persists only if fn returns nil. Fsyncs run after the lock releases; a torn main heals from the .prev generation on load. Declared crash contract: with several relaxed writers all killed before their post-release fsyncs, surviving old-or-new content relies on ext4-class rename-over data ordering (auto_da_alloc).
 	Update(ctx context.Context, fn func(*T) error) error
-	// UpdateNoDirSync is Update without the rename's parent-dir fsync: content is never torn, only the mutation may roll back on power failure — for placeholder states GC re-derives. Requires a metadata-journaling filesystem (ext4/XFS) where a crashed rename resolves to the old or new entry, never to none.
+	// UpdateNoDirSync is Update without the rename's parent-dir fsync, for placeholder states GC re-derives after power loss.
 	UpdateNoDirSync(ctx context.Context, fn func(*T) error) error
 
 	// ReadRaw deserializes and passes to fn without locking; caller must already hold the lock via TryLock.

utils/atomic.go

@@ -20,27 +20,22 @@ type SyncMode bool
 
 // AtomicWriteFile writes data via temp + fsync + rename so readers never see a partial file.
 func AtomicWriteFile(path string, data []byte, perm os.FileMode) error {
-	return atomicWriteFile(path, data, perm, true, true)
+	return atomicWriteFile(path, data, perm, true)
 }
 
 // AtomicWriteFileNoSync is AtomicWriteFile without fsyncs, for transient run-dir files regenerated on the next launch.
 func AtomicWriteFileNoSync(path string, data []byte, perm os.FileMode) error {
-	return atomicWriteFile(path, data, perm, false, false)
+	return atomicWriteFile(path, data, perm, false)
 }
 
 // AtomicWriteJSON marshals v to JSON and writes it atomically.
 func AtomicWriteJSON(path string, v any) error {
-	return atomicWriteJSON(path, v, true, true)
+	return atomicWriteJSON(path, v, true)
 }
 
 // AtomicWriteJSONNoSync marshals v and writes it atomically without fsync (see AtomicWriteFileNoSync).
 func AtomicWriteJSONNoSync(path string, v any) error {
-	return atomicWriteJSON(path, v, false, false)
-}
-
-// AtomicWriteJSONNoDirSync fsyncs the file but not the parent dir: content is never torn on any filesystem, only the rename may be lost to power failure.
-func AtomicWriteJSONNoDirSync(path string, v any) error {
-	return atomicWriteJSON(path, v, true, false)
+	return atomicWriteJSON(path, v, false)
 }
 
 // ReadJSONFile loads path and unmarshals it into v.
@@ -70,6 +65,16 @@ func SyncParentDir(dir string) error {
 	return nil
 }
 
+// SyncFile fsyncs path.
+func SyncFile(path string) (err error) {
+	f, err := os.Open(path) //nolint:gosec // path is a cocoon-managed snapshot file
+	if err != nil {
+		return err
+	}
+	defer func() { err = errors.Join(err, f.Close()) }()
+	return f.Sync()
+}
+
 // SyncTree fsyncs every regular file directly under dir (snapshot dirs are flat), then dir and its parent — the durable-before-kill barrier hibernate needs, since a bare rename fsyncs nothing.
 func SyncTree(dir string) error {
 	entries, err := os.ReadDir(dir)
@@ -80,7 +85,7 @@ func SyncTree(dir string) error {
 		if !e.Type().IsRegular() {
 			continue
 		}
-		if err := syncFile(filepath.Join(dir, e.Name())); err != nil {
+		if err := SyncFile(filepath.Join(dir, e.Name())); err != nil {
 			return err
 		}
 	}
@@ -90,7 +95,7 @@ func SyncTree(dir string) error {
 	return SyncParentDir(filepath.Dir(dir))
 }
 
-func atomicWriteFile(path string, data []byte, perm os.FileMode, fileSync, dirSync bool) error {
+func atomicWriteFile(path string, data []byte, perm os.FileMode, sync bool) error {
 	dir := filepath.Dir(path)
 	tmp, err := os.CreateTemp(dir, ".tmp-*")
 	if err != nil {
@@ -108,7 +113,7 @@ func atomicWriteFile(path string, data []byte, perm os.FileMode, fileSync, dirSy
 	if _, err = tmp.Write(data); err != nil {
 		return fmt.Errorf("write temp file: %w", err)
 	}
-	if fileSync {
+	if sync {
 		if err = tmp.Sync(); err != nil {
 			return fmt.Errorf("sync temp file: %w", err)
 		}
@@ -122,28 +127,19 @@ func atomicWriteFile(path string, data []byte, perm os.FileMode, fileSync, dirSy
 	if err = os.Rename(tmpPath, path); err != nil {
 		return fmt.Errorf("rename temp to target: %w", err)
 	}
-	if dirSync {
+	if sync {
 		if err = SyncParentDir(dir); err != nil {
 			return fmt.Errorf("sync parent dir: %w", err)
 		}
 	}
 	return nil
 }
 
-func atomicWriteJSON(path string, v any, fileSync, dirSync bool) error {
+func atomicWriteJSON(path string, v any, sync bool) error {
 	data, err := json.Marshal(v)
 	if err != nil {
 		return fmt.Errorf("marshal JSON: %w", err)
 	}
 	data = append(data, '\n')
-	return atomicWriteFile(path, data, 0o644, fileSync, dirSync)
-}
-
-func syncFile(path string) (err error) {
-	f, err := os.Open(path) //nolint:gosec // path is a cocoon-managed snapshot file
-	if err != nil {
-		return err
-	}
-	defer func() { err = errors.Join(err, f.Close()) }()
-	return f.Sync()
+	return atomicWriteFile(path, data, 0o644, sync)
 }