Merge pull request #8051 from SivanCola/fix/session-sidecar-durability

fix(desktop): make session sidecars durable / 持久化并串行化会话 sidecar

desktop/session_sidecar_durability_test.go

@@ -0,0 +1,194 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"reasonix/internal/filelock"
+	"reasonix/internal/fileutil"
+)
+
+func TestSessionSidecarSavesUseDurableAtomicWrite(t *testing.T) {
+	tests := []struct {
+		name string
+		path func(string) string
+		save func(string) error
+	}{
+		{
+			name: "titles",
+			path: sessionTitlesPath,
+			save: func(dir string) error {
+				return saveSessionTitles(dir, map[string]string{"session.jsonl": "Durable title"})
+			},
+		},
+		{
+			name: "displays",
+			path: sessionDisplayPath,
+			save: func(dir string) error {
+				return saveSessionDisplays(dir, sessionDisplayMap{
+					"session.jsonl": {messageDisplayKey("expanded prompt"): "visible prompt"},
+				})
+			},
+		},
+		{
+			name: "planner displays",
+			path: sessionPlannerDisplayPath,
+			save: func(dir string) error {
+				return saveSessionPlannerDisplays(dir, sessionPlannerDisplayMap{
+					"session.jsonl": {{UserHash: "prompt-digest", Messages: []HistoryMessage{}}},
+				})
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			dir := t.TempDir()
+			path := tt.path(dir)
+			previousCrashPoint := fileutil.CrashPoint
+			t.Cleanup(func() { fileutil.CrashPoint = previousCrashPoint })
+			usedDurableWrite := false
+			fileutil.CrashPoint = func(op, gotPath string) {
+				if op == "atomic-write" && gotPath == path {
+					usedDurableWrite = true
+				}
+			}
+
+			if err := tt.save(dir); err != nil {
+				t.Fatalf("save %s sidecar: %v", tt.name, err)
+			}
+			if !usedDurableWrite {
+				t.Fatalf("%s sidecar bypassed fileutil.AtomicWriteFile", tt.name)
+			}
+			if runtime.GOOS != "windows" {
+				info, err := os.Stat(path)
+				if err != nil {
+					t.Fatalf("stat %s sidecar: %v", tt.name, err)
+				}
+				if got := info.Mode().Perm(); got != 0o600 {
+					t.Fatalf("%s sidecar mode = %o, want 600", tt.name, got)
+				}
+			}
+		})
+	}
+}
+
+func TestSetSessionTitleHonorsSidecarLock(t *testing.T) {
+	dir := t.TempDir()
+	release, err := filelock.Acquire(context.Background(), sessionTitlesPath(dir)+".lock")
+	if err != nil {
+		t.Fatalf("acquire title sidecar lock: %v", err)
+	}
+	previousTimeout := sessionTitlesQueueTimeout
+	sessionTitlesQueueTimeout = 40 * time.Millisecond
+	t.Cleanup(func() { sessionTitlesQueueTimeout = previousTimeout })
+
+	err = setSessionTitle(dir, filepath.Join(dir, "locked.jsonl"), "Locked")
+	if err == nil {
+		release()
+		t.Fatal("setSessionTitle succeeded while the title sidecar lock was held")
+	}
+	release()
+	if err := setSessionTitle(dir, filepath.Join(dir, "locked.jsonl"), "Saved"); err != nil {
+		t.Fatalf("setSessionTitle after release: %v", err)
+	}
+}
+
+func TestRecordSessionDisplayExternalLockUsesShortBudget(t *testing.T) {
+	if os.Getenv("REASONIX_DISPLAY_LOCK_HELPER") != "" {
+		dir := os.Getenv("REASONIX_DISPLAY_LOCK_DIR")
+		release, err := filelock.Acquire(context.Background(), sessionDisplayPath(dir)+".lock")
+		if err != nil {
+			t.Fatalf("acquire external display lock: %v", err)
+		}
+		defer release()
+		if err := os.WriteFile(filepath.Join(dir, "display-lock.ready"), []byte("ready"), 0o600); err != nil {
+			t.Fatal(err)
+		}
+		if !waitForPlannerDisplayTestFile(filepath.Join(dir, "display-lock.release"), 10*time.Second) {
+			t.Fatal("timed out waiting to release external display lock")
+		}
+		return
+	}
+
+	dir := t.TempDir()
+	var output strings.Builder
+	cmd := exec.Command(os.Args[0], "-test.run=^TestRecordSessionDisplayExternalLockUsesShortBudget$")
+	cmd.Env = append(os.Environ(),
+		"REASONIX_DISPLAY_LOCK_HELPER=1",
+		"REASONIX_DISPLAY_LOCK_DIR="+dir,
+	)
+	cmd.Stdout = &output
+	cmd.Stderr = &output
+	if err := cmd.Start(); err != nil {
+		t.Fatalf("start external display-lock helper: %v", err)
+	}
+	releasePath := filepath.Join(dir, "display-lock.release")
+	releaseHelper := func() {
+		_ = os.WriteFile(releasePath, []byte("release"), 0o600)
+	}
+	if !waitForPlannerDisplayTestFile(filepath.Join(dir, "display-lock.ready"), 5*time.Second) {
+		releaseHelper()
+		_ = cmd.Wait()
+		t.Fatalf("external display-lock helper did not start: %s", output.String())
+	}
+
+	previousTimeout := sessionDisplayExternalLockTimeout
+	sessionDisplayExternalLockTimeout = 100 * time.Millisecond
+	t.Cleanup(func() { sessionDisplayExternalLockTimeout = previousTimeout })
+	started := time.Now()
+	err := recordSessionDisplay(dir, filepath.Join(dir, "session.jsonl"), "expanded prompt", "visible prompt")
+	elapsed := time.Since(started)
+	releaseHelper()
+	if waitErr := cmd.Wait(); waitErr != nil {
+		t.Fatalf("external display-lock helper failed: %v\n%s", waitErr, output.String())
+	}
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("record display error = %v, want external lock deadline exceeded", err)
+	}
+	if elapsed >= 2*time.Second {
+		t.Fatalf("record display waited %v, want the short external lock budget", elapsed)
+	}
+}
+
+func TestSetSessionTitleSerializesConcurrentUpdates(t *testing.T) {
+	dir := t.TempDir()
+	const sessions = 32
+	errs := make(chan error, sessions)
+	var wg sync.WaitGroup
+	for i := range sessions {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			path := filepath.Join(dir, fmt.Sprintf("session-%02d.jsonl", i))
+			errs <- setSessionTitle(dir, path, fmt.Sprintf("Title %02d", i))
+		}(i)
+	}
+	wg.Wait()
+	close(errs)
+	for err := range errs {
+		if err != nil {
+			t.Fatalf("setSessionTitle: %v", err)
+		}
+	}
+
+	titles := loadSessionTitles(dir)
+	if len(titles) != sessions {
+		t.Fatalf("title count = %d, want %d: %#v", len(titles), sessions, titles)
+	}
+	for i := range sessions {
+		key := fmt.Sprintf("session-%02d.jsonl", i)
+		if got := titles[key]; got != fmt.Sprintf("Title %02d", i) {
+			t.Fatalf("%s title = %q, want retained concurrent value", key, got)
+		}
+	}
+}

desktop/sessions.go

@@ -33,6 +33,21 @@ const sessionPlannerDisplayFile = ".planner-display.json"
 const sessionTrashDir = ".trash"
 const sessionTrashMetaFile = ".trash-meta.json"
 
+const (
+	// Durable sidecar publication includes an fsync while holding the update
+	// lock. Keep enough queue budget for a burst of in-process writers on
+	// slower Windows disks, but fail external contention quickly so turn
+	// completion and retry-queue handoff do not stall behind another process.
+	sessionSidecarQueueTimeout        = 5 * time.Second
+	sessionSidecarExternalLockTimeout = 750 * time.Millisecond
+)
+
+var (
+	sessionTitlesQueueTimeout                = sessionSidecarQueueTimeout
+	sessionPlannerDisplayExternalLockTimeout = sessionSidecarExternalLockTimeout
+	sessionDisplayExternalLockTimeout        = sessionSidecarExternalLockTimeout
+)
+
 func sessionTitlesPath(dir string) string  { return filepath.Join(dir, sessionTitlesFile) }
 func sessionDisplayPath(dir string) string { return filepath.Join(dir, sessionDisplayFile) }
 func sessionTrashPath(dir string) string   { return filepath.Join(dir, sessionTrashDir) }
@@ -73,7 +88,33 @@ func loadSessionTitlesForUpdate(dir string) (map[string]string, error) {
 	return loadStringMapForUpdate(sessionTitlesPath(dir))
 }
 
-// saveSessionTitles writes the map atomically (temp file + rename).
+func updateSessionTitles(dir string, mutate func(map[string]string) bool) error {
+	if strings.TrimSpace(dir) == "" {
+		return errors.New("title directory is empty")
+	}
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), sessionTitlesQueueTimeout)
+	defer cancel()
+	release, err := filelock.AcquireWithExternalTimeout(ctx, sessionTitlesPath(dir)+".lock", sessionSidecarExternalLockTimeout)
+	if err != nil {
+		return fmt.Errorf("lock title sidecar: %w", err)
+	}
+	defer release()
+
+	m, err := loadSessionTitlesForUpdate(dir)
+	if err != nil {
+		return err
+	}
+	if !mutate(m) {
+		return nil
+	}
+	return saveSessionTitles(dir, m)
+}
+
+// saveSessionTitles writes the map durably and atomically. Keep this on the
+// shared helper so the temporary file is fsynced before it is published.
 func saveSessionTitles(dir string, m map[string]string) error {
 	b, err := json.MarshalIndent(m, "", "  ")
 	if err != nil {
@@ -82,21 +123,7 @@ func saveSessionTitles(dir string, m map[string]string) error {
 	if err := os.MkdirAll(dir, 0o755); err != nil {
 		return err
 	}
-	tmp, err := os.CreateTemp(dir, ".titles.*.tmp")
-	if err != nil {
-		return err
-	}
-	tmpPath := tmp.Name()
-	if _, err := tmp.Write(b); err != nil {
-		tmp.Close()
-		os.Remove(tmpPath)
-		return err
-	}
-	if err := tmp.Close(); err != nil {
-		os.Remove(tmpPath)
-		return err
-	}
-	return fileutil.ReplaceFile(tmpPath, sessionTitlesPath(dir))
+	return fileutil.AtomicWriteFile(sessionTitlesPath(dir), b, 0o600)
 }
 
 // setSessionTitle sets (or, with an empty title, clears) a session's custom name.
@@ -105,17 +132,22 @@ func setSessionTitle(dir, sessionPath, title string) error {
 	if err != nil {
 		return err
 	}
-	m, err := loadSessionTitlesForUpdate(dir)
-	if err != nil {
-		return err
-	}
 	key := filepath.Base(sessionPath)
-	if strings.TrimSpace(title) == "" {
-		delete(m, key)
-	} else {
-		m[key] = strings.TrimSpace(title)
-	}
-	return saveSessionTitles(dir, m)
+	return updateSessionTitles(dir, func(m map[string]string) bool {
+		title = strings.TrimSpace(title)
+		if title == "" {
+			if _, ok := m[key]; !ok {
+				return false
+			}
+			delete(m, key)
+			return true
+		}
+		if m[key] == title {
+			return false
+		}
+		m[key] = title
+		return true
+	})
 }
 
 // deleteSessionFile moves a session's .jsonl and file sidecars into the local
@@ -598,15 +630,14 @@ func purgeTrashedSessionFile(dir, path string) error {
 	if err := os.RemoveAll(itemDir); err != nil {
 		return err
 	}
-	m, err := loadSessionTitlesForUpdate(dir)
-	if err != nil {
-		return err
-	}
-	if _, ok := m[key]; ok {
-		delete(m, key)
-		if err := saveSessionTitles(dir, m); err != nil {
-			return err
+	if err := updateSessionTitles(dir, func(m map[string]string) bool {
+		if _, ok := m[key]; !ok {
+			return false
 		}
+		delete(m, key)
+		return true
+	}); err != nil {
+		return err
 	}
 	if err := removeSessionDisplayKey(dir, key); err != nil {
 		return err
@@ -950,10 +981,7 @@ type plannerDisplayTurn struct {
 	Messages []HistoryMessage `json:"messages"`
 }
 
-var (
-	sessionPlannerDisplayLockTimeout = 750 * time.Millisecond
-	errCorruptSessionPlannerDisplay  = errors.New("corrupt planner display sidecar")
-)
+var errCorruptSessionPlannerDisplay = errors.New("corrupt planner display sidecar")
 
 // sessionPlannerDisplayUpdateAfterLoad is a subprocess-test seam. Production
 // leaves it nil; tests use it to force two independent processes into the old
@@ -1018,21 +1046,7 @@ func saveSessionPlannerDisplays(dir string, m sessionPlannerDisplayMap) error {
 	if err := os.MkdirAll(dir, 0o755); err != nil {
 		return err
 	}
-	tmp, err := os.CreateTemp(dir, ".planner-display.*.tmp")
-	if err != nil {
-		return err
-	}
-	tmpPath := tmp.Name()
-	if _, err := tmp.Write(b); err != nil {
-		tmp.Close()
-		os.Remove(tmpPath)
-		return err
-	}
-	if err := tmp.Close(); err != nil {
-		os.Remove(tmpPath)
-		return err
-	}
-	return fileutil.ReplaceFile(tmpPath, sessionPlannerDisplayPath(dir))
+	return fileutil.AtomicWriteFile(sessionPlannerDisplayPath(dir), b, 0o600)
 }
 
 func saveOrRemoveSessionPlannerDisplays(dir string, m sessionPlannerDisplayMap) error {
@@ -1053,9 +1067,9 @@ func updateSessionPlannerDisplays(dir string, recoverCorrupt bool, mutate func(s
 	if err := os.MkdirAll(dir, 0o755); err != nil {
 		return err
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), sessionPlannerDisplayLockTimeout)
+	ctx, cancel := context.WithTimeout(context.Background(), sessionSidecarQueueTimeout)
 	defer cancel()
-	release, err := filelock.Acquire(ctx, sessionPlannerDisplayPath(dir)+".lock")
+	release, err := filelock.AcquireWithExternalTimeout(ctx, sessionPlannerDisplayPath(dir)+".lock", sessionPlannerDisplayExternalLockTimeout)
 	if err != nil {
 		return fmt.Errorf("lock planner display sidecar: %w", err)
 	}
@@ -1155,21 +1169,7 @@ func saveSessionDisplays(dir string, m sessionDisplayMap) error {
 	if err := os.MkdirAll(dir, 0o755); err != nil {
 		return err
 	}
-	tmp, err := os.CreateTemp(dir, ".display.*.tmp")
-	if err != nil {
-		return err
-	}
-	tmpPath := tmp.Name()
-	if _, err := tmp.Write(b); err != nil {
-		tmp.Close()
-		os.Remove(tmpPath)
-		return err
-	}
-	if err := tmp.Close(); err != nil {
-		os.Remove(tmpPath)
-		return err
-	}
-	return fileutil.ReplaceFile(tmpPath, sessionDisplayPath(dir))
+	return fileutil.AtomicWriteFile(sessionDisplayPath(dir), b, 0o600)
 }
 
 func saveOrRemoveSessionDisplays(dir string, m sessionDisplayMap) error {
@@ -1194,9 +1194,9 @@ func updateSessionDisplays(dir string, mutate func(sessionDisplayMap) bool) erro
 	if err := os.MkdirAll(dir, 0o755); err != nil {
 		return err
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), 750*time.Millisecond)
+	ctx, cancel := context.WithTimeout(context.Background(), sessionSidecarQueueTimeout)
 	defer cancel()
-	release, err := filelock.Acquire(ctx, sessionDisplayPath(dir)+".lock")
+	release, err := filelock.AcquireWithExternalTimeout(ctx, sessionDisplayPath(dir)+".lock", sessionDisplayExternalLockTimeout)
 	if err != nil {
 		return fmt.Errorf("lock display sidecar: %w", err)
 	}

desktop/sessions_test.go

@@ -1243,7 +1243,7 @@ func TestRecordSessionPlannerDisplayConcurrentPreservesEverySession(t *testing.T
 func TestRecordSessionPlannerDisplayCrossProcessPreservesEverySession(t *testing.T) {
 	if role := os.Getenv("REASONIX_PLANNER_DISPLAY_HELPER"); role != "" {
 		dir := os.Getenv("REASONIX_PLANNER_DISPLAY_DIR")
-		sessionPlannerDisplayLockTimeout = 5 * time.Second
+		sessionPlannerDisplayExternalLockTimeout = 5 * time.Second
 		attempted := filepath.Join(dir, role+".attempted")
 		loaded := filepath.Join(dir, role+".loaded")
 		release := filepath.Join(dir, role+".release")

internal/filelock/filelock.go

@@ -36,6 +36,21 @@ var localRegistry = struct {
 // function is called. It serializes both goroutines in this process and other
 // Reasonix processes, and never waits past ctx's deadline.
 func Acquire(ctx context.Context, path string) (func(), error) {
+	return acquire(ctx, path, 0)
+}
+
+// AcquireWithExternalTimeout obtains an exclusive lock while keeping the
+// in-process queue and cross-process file-lock budgets separate. ctx bounds
+// only the wait for another goroutine in this process; externalTimeout starts
+// after that queue is acquired and bounds retries against other processes.
+func AcquireWithExternalTimeout(ctx context.Context, path string, externalTimeout time.Duration) (func(), error) {
+	if externalTimeout <= 0 {
+		return nil, errors.New("external file lock timeout must be positive")
+	}
+	return acquire(ctx, path, externalTimeout)
+}
+
+func acquire(ctx context.Context, path string, externalTimeout time.Duration) (func(), error) {
 	if ctx == nil {
 		ctx = context.Background()
 	}
@@ -48,6 +63,12 @@ func Acquire(ctx context.Context, path string) (func(), error) {
 		return nil, err
 	}
 	_ = local
+	fileCtx := ctx
+	cancel := func() {}
+	if externalTimeout > 0 {
+		fileCtx, cancel = context.WithTimeout(context.Background(), externalTimeout)
+	}
+	defer cancel()
 
 	for {
 		releaseFile, err := tryLockFile(key)
@@ -67,7 +88,7 @@ func Acquire(ctx context.Context, path string) (func(), error) {
 		timer := time.NewTimer(retryInterval)
 		select {
 		case <-timer.C:
-		case <-ctx.Done():
+		case <-fileCtx.Done():
 			if !timer.Stop() {
 				select {
 				case <-timer.C:
@@ -75,7 +96,7 @@ func Acquire(ctx context.Context, path string) (func(), error) {
 				}
 			}
 			releaseLocal()
-			return nil, fmt.Errorf("acquire file lock: %w", ctx.Err())
+			return nil, fmt.Errorf("acquire file lock: %w", fileCtx.Err())
 		}
 	}
 }

internal/filelock/filelock_test.go

@@ -29,6 +29,34 @@ func TestAcquireHonorsDeadlineAndRecoversAfterRelease(t *testing.T) {
 	secondRelease()
 }
 
+func TestAcquireWithExternalTimeoutBoundsOnlyFileLockRetries(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "state.lock")
+	releaseExternal, err := tryLockFile(path)
+	if err != nil {
+		t.Fatalf("hold external file lock: %v", err)
+	}
+	defer releaseExternal()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	started := time.Now()
+	_, err = AcquireWithExternalTimeout(ctx, path, 60*time.Millisecond)
+	elapsed := time.Since(started)
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("external acquire error = %v, want deadline exceeded", err)
+	}
+	if elapsed >= time.Second {
+		t.Fatalf("external acquire waited %v, want the short external budget", elapsed)
+	}
+}
+
+func TestAcquireWithExternalTimeoutRejectsInvalidBudget(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "state.lock")
+	if _, err := AcquireWithExternalTimeout(context.Background(), path, 0); err == nil {
+		t.Fatal("zero external timeout should be rejected")
+	}
+}
+
 func TestLocalRegistryReclaimsReleasedEntries(t *testing.T) {
 	before := RegistrySizeForTest()
 	path := filepath.Join(t.TempDir(), "ephemeral.lock")