fix(persistence): load sessions resiliently and bound file-descriptor use

Two robustness fixes for the recurring persistence reports (#284, #292, #318):

- Make LoadSessions best-effort per session. Previously a single missing or
  unreadable per-session file (e.g. a partial session directory after a
  mid-write restart or a K8s volume race) made the whole load fail and dropped
  EVERY persisted session. Now each session's history and mocks load
  independently: a missing/corrupt file is logged and skipped, other sessions
  and the session name/metadata still load.
- Bound persistence concurrency with errgroup SetLimit(16) on both the load and
  the store paths, so a large number of sessions can no longer exhaust the
  process's file descriptors ("too many open files").

Tests (server/services/persistence_test.go, pass under -race):
- TestLoadSessionsResilientToMissingFiles: a partial session dir no longer wipes
  the others.
- TestStoreAndLoadManySessions: round-trips more sessions than the limit without
  deadlocking or losing data.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: gwleclerc

server/services/persistence.go

@@ -2,7 +2,6 @@ package services
 
 import (
 	"fmt"
-	"io"
 	"log/slog"
 	"os"
 	"path/filepath"
@@ -17,6 +16,12 @@ const (
 	historyFileName  = "history.yml"
 	mocksFileName    = "mocks.yml"
 	sessionsFileName = "sessions.yml"
+
+	// maxPersistenceConcurrency bounds how many session files are read/written in parallel.
+	// Persistence spawns work per session, so without a cap a large number of sessions could
+	// exhaust the process's file descriptors ("too many open files"). errgroup.Go blocks once
+	// this many goroutines are in flight.
+	maxPersistenceConcurrency = 16
 )
 
 type Persistence interface {
@@ -110,6 +115,7 @@ func (p *persistence) StoreSessions(sessions types.Sessions) {
 		return
 	}
 	var sessionsGroup errgroup.Group
+	sessionsGroup.SetLimit(maxPersistenceConcurrency)
 	for i := range sessions {
 		session := sessions[i]
 		sessionsGroup.Go(func() error {
@@ -146,79 +152,76 @@ func (p *persistence) LoadSessions() (types.Sessions, error) {
 	if _, err := os.Stat(p.persistenceDirectory); os.IsNotExist(err) {
 		return nil, err
 	}
-	file, err := os.Open(filepath.Join(p.persistenceDirectory, sessionsFileName))
-	if err != nil {
-		return nil, err
-	}
-	defer file.Close()
-	bytes, err := io.ReadAll(file)
+	data, err := os.ReadFile(filepath.Join(p.persistenceDirectory, sessionsFileName))
 	if err != nil {
 		return nil, err
 	}
 	var sessions types.Sessions
-	err = yaml.Unmarshal(bytes, &sessions)
-	if err != nil {
+	if err := yaml.Unmarshal(data, &sessions); err != nil {
 		return nil, err
 	}
-	var sessionsGroup errgroup.Group
+
+	// Load each session's history and mocks independently and resiliently: a missing or
+	// unreadable per-session file must never discard the other sessions. Previously a single
+	// incomplete session directory made the whole load fail and wiped every persisted session on
+	// restart (the recurring persistence reports). The session name always loads from the
+	// summary, and whatever history/mocks are readable are kept.
 	var sessionsLock sync.Mutex
+	var group errgroup.Group
+	group.SetLimit(maxPersistenceConcurrency)
 	for i := range sessions {
 		session := sessions[i]
-		sessionsGroup.Go(func() error {
-			historyFile, err := os.Open(filepath.Join(p.persistenceDirectory, session.ID, historyFileName))
-			if err != nil {
-				slog.Error(fmt.Sprintf("Unable to open history file for session %q", session.ID), "error", err)
-				return err
-			}
-			defer historyFile.Close()
-			bytes, err := io.ReadAll(historyFile)
+		group.Go(func() error {
+			history, err := loadPersistedFile[types.History](p.persistenceDirectory, session.ID, historyFileName)
 			if err != nil {
-				return err
-			}
-			var history types.History
-			err = yaml.Unmarshal(bytes, &history)
-			if err != nil {
-				return err
+				logPersistedLoadError("history", session.ID, err)
+				return nil
 			}
 			sessionsLock.Lock()
 			session.History = history
 			sessionsLock.Unlock()
 			return nil
 		})
-		sessionsGroup.Go(func() error {
-			mocksFile, err := os.Open(filepath.Join(p.persistenceDirectory, session.ID, mocksFileName))
+		group.Go(func() error {
+			mocks, err := loadPersistedFile[types.Mocks](p.persistenceDirectory, session.ID, mocksFileName)
 			if err != nil {
-				slog.Error(fmt.Sprintf("Unable to open mocks file for session %q", session.ID), "error", err)
-				return err
+				logPersistedLoadError("mocks", session.ID, err)
+				return nil
 			}
-			defer mocksFile.Close()
-			bytes, err := io.ReadAll(mocksFile)
-			if err != nil {
-				return err
-			}
-			var mocks types.Mocks
-			err = yaml.Unmarshal(bytes, &mocks)
-			if err != nil {
-				return err
-			}
-
 			// mocks are stored as a stack so we need to reverse the list from mocks file
 			orderedMocks := make(types.Mocks, 0, len(mocks))
-			for i := len(mocks) - 1; i >= 0; i-- {
-				orderedMocks = append(orderedMocks, mocks[i])
+			for j := len(mocks) - 1; j >= 0; j-- {
+				orderedMocks = append(orderedMocks, mocks[j])
 			}
 			sessionsLock.Lock()
 			session.Mocks = orderedMocks
 			sessionsLock.Unlock()
 			return nil
 		})
 	}
-	if err := sessionsGroup.Wait(); err != nil {
-		return nil, err
-	}
+	_ = group.Wait() // per-file errors are handled above; the load itself never fails here
 	return sessions, nil
 }
 
+// loadPersistedFile reads and YAML-decodes a per-session persistence file (history or mocks).
+func loadPersistedFile[T any](dir, sessionID, name string) (T, error) {
+	var v T
+	data, err := os.ReadFile(filepath.Join(dir, sessionID, name))
+	if err != nil {
+		return v, err
+	}
+	return v, yaml.Unmarshal(data, &v)
+}
+
+// logPersistedLoadError reports a per-session load problem without aborting the whole load.
+func logPersistedLoadError(kind, sessionID string, err error) {
+	if os.IsNotExist(err) {
+		slog.Debug(fmt.Sprintf("No %s file for session %q, treating as empty", kind, sessionID))
+		return
+	}
+	slog.Warn(fmt.Sprintf("Unable to load %s for session %q, ignoring it", kind, sessionID), "error", err)
+}
+
 func (p *persistence) createSessionDirectory(sessionID string) error {
 	slog.Debug(fmt.Sprintf("Create directory for session %q", sessionID))
 	sessionDirectory := filepath.Join(p.persistenceDirectory, sessionID)

server/services/persistence_test.go

@@ -0,0 +1,98 @@
+package services
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/smocker-dev/smocker/server/types"
+)
+
+// TestLoadSessionsResilientToMissingFiles locks the fix for the recurring persistence bug:
+// a single incomplete session directory (e.g. a missing mocks.yml after a partial write or a
+// mid-write pod restart) used to make LoadSessions fail and drop EVERY persisted session.
+// Loading must now be best-effort per session: other sessions still load, and a session with a
+// missing file keeps its name and whatever else is readable.
+func TestLoadSessionsResilientToMissingFiles(t *testing.T) {
+	dir := t.TempDir()
+
+	write := func(name, content string) {
+		t.Helper()
+		full := filepath.Join(dir, name)
+		if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.WriteFile(full, []byte(content), 0o644); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	write("sessions.yml", "- id: sess-full\n  name: Full\n  date: 2020-01-01T00:00:00Z\n"+
+		"- id: sess-partial\n  name: Partial\n  date: 2020-01-01T00:00:00Z\n")
+	// Complete session: has both mocks and history.
+	write("sess-full/mocks.yml", "- request:\n    path: /a\n  response:\n    status: 200\n")
+	write("sess-full/history.yml", "[]\n")
+	// Partial session: history only, mocks.yml is missing (the failure case).
+	write("sess-partial/history.yml", "[]\n")
+
+	sessions, err := NewPersistence(dir).LoadSessions()
+	if err != nil {
+		t.Fatalf("LoadSessions must not fail because of a partial session dir: %v", err)
+	}
+	if len(sessions) != 2 {
+		t.Fatalf("expected both sessions to load, got %d", len(sessions))
+	}
+
+	byName := map[string]*types.Session{}
+	for _, s := range sessions {
+		byName[s.Name] = s
+	}
+	if full := byName["Full"]; full == nil || len(full.Mocks) != 1 {
+		t.Errorf("complete session should keep its mock, got %+v", full)
+	}
+	if partial := byName["Partial"]; partial == nil {
+		t.Error("partial session must still be loaded (by name)")
+	} else if len(partial.Mocks) != 0 {
+		t.Errorf("partial session with missing mocks.yml should have no mocks, got %d", len(partial.Mocks))
+	}
+}
+
+// TestStoreAndLoadManySessions round-trips more sessions than maxPersistenceConcurrency to
+// ensure the file-descriptor bounding (errgroup SetLimit) throttles correctly without
+// deadlocking or dropping data.
+func TestStoreAndLoadManySessions(t *testing.T) {
+	dir := t.TempDir()
+	const n = maxPersistenceConcurrency*3 + 1 // comfortably above the concurrency limit
+
+	sessions := make(types.Sessions, 0, n)
+	for i := 0; i < n; i++ {
+		sessions = append(sessions, &types.Session{
+			ID:   fmt.Sprintf("sess-%d", i),
+			Name: fmt.Sprintf("Session %d", i),
+			Mocks: types.Mocks{{
+				Request: types.MockRequest{
+					Path:   types.StringMatcher{Matcher: "ShouldEqual", Value: fmt.Sprintf("/p%d", i)},
+					Method: types.StringMatcher{Matcher: "ShouldEqual", Value: "GET"},
+				},
+				Response: &types.MockResponse{Status: 200},
+			}},
+		})
+	}
+
+	p := NewPersistence(dir)
+	p.StoreSessions(sessions)
+
+	loaded, err := p.LoadSessions()
+	if err != nil {
+		t.Fatalf("LoadSessions: %v", err)
+	}
+	if len(loaded) != n {
+		t.Fatalf("expected %d sessions, got %d", n, len(loaded))
+	}
+	for _, s := range loaded {
+		if len(s.Mocks) != 1 {
+			t.Fatalf("session %q should have 1 mock, got %d", s.ID, len(s.Mocks))
+		}
+	}
+}