git.iko.soy/crush
on main help sign in
refactor(lock): add canonical internal/lock package, migrate db and cmd callers
7ce0936Kieran Klukas
internal/cmd/root.go+2−1
compare with main
unifiedsplit
@@ -29,6 +29,7 @@ import (
29 "github.com/charmbracelet/crush/internal/config"
29 "github.com/charmbracelet/crush/internal/config"
30 "github.com/charmbracelet/crush/internal/db"
30 "github.com/charmbracelet/crush/internal/db"
31 "github.com/charmbracelet/crush/internal/event"
31 "github.com/charmbracelet/crush/internal/event"
32+ "github.com/charmbracelet/crush/internal/lock"
32 crushlog "github.com/charmbracelet/crush/internal/log"
33 crushlog "github.com/charmbracelet/crush/internal/log"
33 "github.com/charmbracelet/crush/internal/projects"
34 "github.com/charmbracelet/crush/internal/projects"
34 "github.com/charmbracelet/crush/internal/proto"
35 "github.com/charmbracelet/crush/internal/proto"
@@ -466,7 +467,7 @@ func spawnAndWaitReady(cmd *cobra.Command, hostURL *url.URL) error {
466 if err != nil {
467 if err != nil {
467 return err
468 return err
468 }
469 }
469− release, err := acquireSpawnLock(filepath.Join(chDir, "start.lock"))
470+ release, err := lock.File(cmd.Context(), filepath.Join(chDir, "start.lock"))
470 if err != nil {
471 if err != nil {
471 // If the lock itself is unavailable, fall back to the
472 // If the lock itself is unavailable, fall back to the
472 // unsynchronized path rather than blocking the user.
473 // unsynchronized path rather than blocking the user.
internal/cmd/spawnlock_other.godeleted−28
@@ -1,28 +0,0 @@
1−//go:build !windows
2−
3−package cmd
4−
5−import (
6− "fmt"
7− "os"
8−
9− "golang.org/x/sys/unix"
10−)
11−
12−// acquireSpawnLock takes an exclusive flock on the given file (creating
13−// it if necessary) and returns a release function that unlocks and
14−// closes the file. Blocks until the lock is acquired.
15−func acquireSpawnLock(path string) (func(), error) {
16− f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o600)
17− if err != nil {
18− return nil, fmt.Errorf("open spawn lock %q: %v", path, err)
19− }
20− if err := unix.Flock(int(f.Fd()), unix.LOCK_EX); err != nil {
21− _ = f.Close()
22− return nil, fmt.Errorf("flock spawn lock %q: %v", path, err)
23− }
24− return func() {
25− _ = unix.Flock(int(f.Fd()), unix.LOCK_UN)
26− _ = f.Close()
27− }, nil
28−}
internal/cmd/spawnlock_windows.godeleted−32
@@ -1,32 +0,0 @@
1−//go:build windows
2−
3−package cmd
4−
5−import (
6− "fmt"
7− "math"
8− "os"
9−
10− "golang.org/x/sys/windows"
11−)
12−
13−// acquireSpawnLock takes an exclusive lock on the given file (creating
14−// it if necessary) using LockFileEx, and returns a release function
15−// that unlocks and closes the file. Blocks until the lock is acquired.
16−func acquireSpawnLock(path string) (func(), error) {
17− f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o600)
18− if err != nil {
19− return nil, fmt.Errorf("open spawn lock %q: %v", path, err)
20− }
21− h := windows.Handle(f.Fd())
22− ol := new(windows.Overlapped)
23− if err := windows.LockFileEx(h, windows.LOCKFILE_EXCLUSIVE_LOCK, 0, math.MaxUint32, math.MaxUint32, ol); err != nil {
24− _ = f.Close()
25− return nil, fmt.Errorf("LockFileEx spawn lock %q: %v", path, err)
26− }
27− return func() {
28− ol := new(windows.Overlapped)
29− _ = windows.UnlockFileEx(windows.Handle(f.Fd()), 0, math.MaxUint32, math.MaxUint32, ol)
30− _ = f.Close()
31− }, nil
32−}
internal/config/store.go+143−65
@@ -8,10 +8,13 @@ import (
8 "os"
8 "os"
9 "path/filepath"
9 "path/filepath"
10 "slices"
10 "slices"
11+ "sync"
12+ "time"
11
13
12 "charm.land/catwalk/pkg/catwalk"
14 "charm.land/catwalk/pkg/catwalk"
13 hyperp "github.com/charmbracelet/crush/internal/agent/hyper"
15 hyperp "github.com/charmbracelet/crush/internal/agent/hyper"
14 "github.com/charmbracelet/crush/internal/env"
16 "github.com/charmbracelet/crush/internal/env"
17+ "github.com/charmbracelet/crush/internal/lock"
15 "github.com/charmbracelet/crush/internal/oauth"
18 "github.com/charmbracelet/crush/internal/oauth"
16 "github.com/charmbracelet/crush/internal/oauth/copilot"
19 "github.com/charmbracelet/crush/internal/oauth/copilot"
17 "github.com/charmbracelet/crush/internal/oauth/hyper"
20 "github.com/charmbracelet/crush/internal/oauth/hyper"
@@ -19,6 +22,11 @@ import (
19 "github.com/tidwall/sjson"
22 "github.com/tidwall/sjson"
20)
23)
21
24
25+// configLockDeadline bounds how long lockConfig waits for the
26+// cross-process flock before giving up. A few seconds is plenty for
27+// honest contention; longer suggests something is wedged.
28+const configLockDeadline = 5 * time.Second
29+
22// fileSnapshot captures metadata about a config file at a point in time.
30// fileSnapshot captures metadata about a config file at a point in time.
23type fileSnapshot struct {
31type fileSnapshot struct {
24 Path string
32 Path string
@@ -37,6 +45,11 @@ type RuntimeOverrides struct {
37// ConfigStore is the single entry point for all config access. It owns the
45// ConfigStore is the single entry point for all config access. It owns the
38// pure-data Config, runtime state (working directory, resolver, known
46// pure-data Config, runtime state (working directory, resolver, known
39// providers), and persistence to both global and workspace config files.
47// providers), and persistence to both global and workspace config files.
48+//
49+// mu serialises all config file mutations (SetConfigFields,
50+// RemoveConfigField, RefreshOAuthToken) to prevent both in-process
51+// goroutine races and, together with the shared lock.File, cross-process
52+// races on the config file.
40type ConfigStore struct {
53type ConfigStore struct {
41 config *Config
54 config *Config
42 workingDir string
55 workingDir string
@@ -50,6 +63,8 @@ type ConfigStore struct {
50 snapshots map[string]fileSnapshot // path -> snapshot at last capture
63 snapshots map[string]fileSnapshot // path -> snapshot at last capture
51 autoReloadDisabled bool // set during load/reload to prevent re-entrancy
64 autoReloadDisabled bool // set during load/reload to prevent re-entrancy
52 reloadInProgress bool // set during reload to avoid disk writes mid-reload
65 reloadInProgress bool // set during reload to avoid disk writes mid-reload
66+
67+ mu sync.Mutex
53}
68}
54
69
55// Config returns the pure-data config struct (read-only after load).
70// Config returns the pure-data config struct (read-only after load).
@@ -95,6 +110,74 @@ func (s *ConfigStore) LoadedPaths() []string {
95 return slices.Clone(s.loadedPaths)
110 return slices.Clone(s.loadedPaths)
96}
111}
97
112
113+// lockConfig acquires both the in-process mutex and a cross-process flock
114+// on the config file for the given scope. Callers that need to do I/O
115+// between reading and writing (e.g. an HTTP token exchange) must use
116+// lockConfig explicitly rather than atomicWrite.
117+//
118+// The returned release function drops both locks. Callers must call it
119+// as soon as the file access is complete — no I/O should be performed
120+// while the lock is held.
121+func (s *ConfigStore) lockConfig(scope Scope) (func(), error) {
122+ s.mu.Lock()
123+ path, err := s.configPath(scope)
124+ if err != nil {
125+ s.mu.Unlock()
126+ return nil, err
127+ }
128+ if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
129+ s.mu.Unlock()
130+ return nil, fmt.Errorf("create config directory: %w", err)
131+ }
132+ ctx, cancel := context.WithTimeout(context.Background(), configLockDeadline)
133+ defer cancel()
134+ release, err := lock.File(ctx, path+".lock")
135+ if err != nil {
136+ s.mu.Unlock()
137+ return nil, fmt.Errorf("acquire config lock: %w", err)
138+ }
139+ return func() {
140+ release()
141+ s.mu.Unlock()
142+ }, nil
143+}
144+
145+// atomicWrite handles the lock-read-transform-write-unlock cycle for
146+// config file mutations. The fn callback receives the current file
147+// contents (raw bytes, or {} if the file is missing) and must return the
148+// new contents. fn must be pure — no I/O, no network calls.
149+func (s *ConfigStore) atomicWrite(scope Scope, fn func(current []byte) ([]byte, error)) error {
150+ unlock, err := s.lockConfig(scope)
151+ if err != nil {
152+ return err
153+ }
154+ defer unlock()
155+
156+ path, err := s.configPath(scope)
157+ if err != nil {
158+ return err
159+ }
160+
161+ data, err := os.ReadFile(path)
162+ if err != nil {
163+ if os.IsNotExist(err) {
164+ data = []byte("{}")
165+ } else {
166+ return fmt.Errorf("read config file: %w", err)
167+ }
168+ }
169+
170+ newData, err := fn(data)
171+ if err != nil {
172+ return err
173+ }
174+
175+ if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
176+ return fmt.Errorf("create config directory: %w", err)
177+ }
178+ return atomicWriteFile(path, newData, 0o600)
179+}
180+
98// configPath returns the file path for the given scope.
181// configPath returns the file path for the given scope.
99func (s *ConfigStore) configPath(scope Scope) (string, error) {
182func (s *ConfigStore) configPath(scope Scope) (string, error) {
100 switch scope {
183 switch scope {
@@ -134,32 +217,23 @@ func (s *ConfigStore) SetConfigField(scope Scope, key string, value any) error {
134// config to keep in-memory state fresh. This is preferred over multiple
217// config to keep in-memory state fresh. This is preferred over multiple
135// SetConfigField calls when writing several fields atomically to avoid
218// SetConfigField calls when writing several fields atomically to avoid
136// intermediate reloads with partial state.
219// intermediate reloads with partial state.
220+//
221+// The write is protected by an in-process mutex and a cross-process flock
222+// to prevent races between concurrent writers in different processes.
137func (s *ConfigStore) SetConfigFields(scope Scope, kv map[string]any) error {
223func (s *ConfigStore) SetConfigFields(scope Scope, kv map[string]any) error {
138− path, err := s.configPath(scope)
224+ err := s.atomicWrite(scope, func(data []byte) ([]byte, error) {
139− if err != nil {
140− return fmt.Errorf("%v: %w", kv, err)
141− }
142− data, err := os.ReadFile(path)
143− if err != nil {
144− if os.IsNotExist(err) {
145− data = []byte("{}")
146− } else {
147− return fmt.Errorf("failed to read config file: %w", err)
148− }
149− }
150−
151− newValue := string(data)
225+ v := string(data)
152− for key, value := range kv {
226+ for key, value := range kv {
227+ var sErr error
153− newValue, err = sjson.Set(newValue, key, value)
228+ v, sErr = sjson.Set(v, key, value)
154− if err != nil {
229+ if sErr != nil {
155− return fmt.Errorf("failed to set config field %s: %w", key, err)
230+ return nil, fmt.Errorf("failed to set config field %s: %w", key, sErr)
231+ }
156 }
232 }
157− }
233+ return []byte(v), nil
158− if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
234+ })
159− return fmt.Errorf("failed to create config directory %q: %w", path, err)
235+ if err != nil {
160− }
236+ return err
161− if err := atomicWriteFile(path, []byte(newValue), 0o600); err != nil {
162− return fmt.Errorf("failed to write config file: %w", err)
163 }
237 }
164
238
165 // Auto-reload to keep in-memory state fresh after config edits.
239 // Auto-reload to keep in-memory state fresh after config edits.
@@ -176,28 +250,20 @@ func (s *ConfigStore) SetConfigFields(scope Scope, kv map[string]any) error {
176// RemoveConfigField removes a key from the config file for the given scope.
250// RemoveConfigField removes a key from the config file for the given scope.
177// After a successful write, it automatically reloads config to keep in-memory
251// After a successful write, it automatically reloads config to keep in-memory
178// state fresh.
252// state fresh.
253+//
254+// The write is protected by an in-process mutex and a cross-process flock.
179func (s *ConfigStore) RemoveConfigField(scope Scope, key string) error {
255func (s *ConfigStore) RemoveConfigField(scope Scope, key string) error {
180− path, err := s.configPath(scope)
256+ err := s.atomicWrite(scope, func(data []byte) ([]byte, error) {
181− if err != nil {
182− return fmt.Errorf("%s: %w", key, err)
183− }
184− data, err := os.ReadFile(path)
185− if err != nil {
186− return fmt.Errorf("failed to read config file: %w", err)
187− }
188−
189− newValue, err := sjson.Delete(string(data), key)
257+ v, sErr := sjson.Delete(string(data), key)
258+ if sErr != nil {
259+ return nil, fmt.Errorf("failed to delete config field %s: %w", key, sErr)
260+ }
261+ return []byte(v), nil
262+ })
190 if err != nil {
263 if err != nil {
191− return fmt.Errorf("failed to delete config field %s: %w", key, err)
264+ return err
192− }
193− if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
194− return fmt.Errorf("failed to create config directory %q: %w", path, err)
195− }
196− if err := atomicWriteFile(path, []byte(newValue), 0o600); err != nil {
197− return fmt.Errorf("failed to write config file: %w", err)
198 }
265 }
199
266
200− // Auto-reload to keep in-memory state fresh after config edits.
201 if err := s.autoReload(context.Background()); err != nil {
267 if err := s.autoReload(context.Background()); err != nil {
202 slog.Warn("Config file updated but failed to reload in-memory state", "error", err)
268 slog.Warn("Config file updated but failed to reload in-memory state", "error", err)
203 }
269 }
@@ -300,12 +366,14 @@ func (s *ConfigStore) SetProviderAPIKey(scope Scope, providerID string, apiKey a
300}
366}
301
367
302// RefreshOAuthToken refreshes the OAuth token for the given provider.
368// RefreshOAuthToken refreshes the OAuth token for the given provider.
303−// Before making an external refresh request, it checks the config file on
369+//
370+// It uses two-phase locking: the pre-check (reading the config file to
304−// disk to see if another Crush session has already refreshed the token. If
371+// see if another process already refreshed) happens under the config
305−// a newer, unexpired token is found, it is used instead of refreshing. If
372+// lock, then the HTTP exchange runs without any lock held, and finally
373+// the result is persisted via SetConfigFields (which acquires the lock
306−// the exchange fails (e.g. because another session already rotated the
374+// internally). If the exchange fails — e.g. because another process
307−// refresh token), the disk is re-checked to recover the other session's
375+// already rotated the refresh token — the disk is re-checked under lock
308−// token.
376+// to recover the other process's token.
309func (s *ConfigStore) RefreshOAuthToken(ctx context.Context, scope Scope, providerID string) error {
377func (s *ConfigStore) RefreshOAuthToken(ctx context.Context, scope Scope, providerID string) error {
310 providerConfig, exists := s.config.Providers.Get(providerID)
378 providerConfig, exists := s.config.Providers.Get(providerID)
311 if !exists {
379 if !exists {
@@ -316,16 +384,22 @@ func (s *ConfigStore) RefreshOAuthToken(ctx context.Context, scope Scope, provid
316 return fmt.Errorf("provider %s does not have an OAuth token", providerID)
384 return fmt.Errorf("provider %s does not have an OAuth token", providerID)
317 }
385 }
318
386
319− // Check if another session refreshed the token recently by reading
387+ // Phase 1: Pre-check under lock — did another process already refresh?
320− // the current token from the config file on disk.
388+ release, lockErr := s.lockConfig(scope)
389+ if lockErr != nil {
390+ slog.Warn("Failed to lock config for pre-check, proceeding anyway", "provider", providerID, "error", lockErr)
391+ } else {
321− newToken, err := s.loadTokenFromDisk(scope, providerID)
392+ diskToken, err := s.loadTokenFromDisk(scope, providerID)
393+ release()
322− if err != nil {
394+ if err != nil {
323− slog.Warn("Failed to read token from config file, proceeding with refresh", "provider", providerID, "error", err)
395+ slog.Warn("Failed to read token from config file", "provider", providerID, "error", err)
324− } else if newToken != nil && !newToken.IsExpired() && newToken.AccessToken != providerConfig.OAuthToken.AccessToken {
396+ } else if diskToken != nil && !diskToken.IsExpired() && diskToken.AccessToken != providerConfig.OAuthToken.AccessToken {
325− slog.Info("Using token refreshed by another session", "provider", providerID)
397+ slog.Info("Using token refreshed by another session", "provider", providerID)
326− return s.applyToken(providerConfig, newToken, providerID)
398+ return s.applyToken(providerConfig, diskToken, providerID)
399+ }
327 }
400 }
328
401
402+ // Phase 2: HTTP exchange — no lock held.
329 var refreshedToken *oauth.Token
403 var refreshedToken *oauth.Token
330 var refreshErr error
404 var refreshErr error
331 switch providerID {
405 switch providerID {
@@ -337,15 +411,19 @@ func (s *ConfigStore) RefreshOAuthToken(ctx context.Context, scope Scope, provid
337 return fmt.Errorf("OAuth refresh not supported for provider %s", providerID)
411 return fmt.Errorf("OAuth refresh not supported for provider %s", providerID)
338 }
412 }
339 if refreshErr != nil {
413 if refreshErr != nil {
414+ // Phase 3: Fallback — re-check disk under lock. The exchange may
340− // The exchange may have failed because another session already
415+ // have failed because another process already rotated the refresh
341− // rotated the refresh token. Re-read the config file and use the
416+ // token.
342− // other session's token if available.
417+ if release, lockErr := s.lockConfig(scope); lockErr == nil {
343− if diskToken, diskErr := s.loadTokenFromDisk(scope, providerID); diskErr == nil &&
418+ diskToken, diskErr := s.loadTokenFromDisk(scope, providerID)
419+ release()
420+ if diskErr == nil &&
344− diskToken != nil &&
421+ diskToken != nil &&
345− !diskToken.IsExpired() &&
422+ !diskToken.IsExpired() &&
346− diskToken.AccessToken != providerConfig.OAuthToken.AccessToken {
423+ diskToken.AccessToken != providerConfig.OAuthToken.AccessToken {
347− slog.Info("Using token refreshed by another session after exchange failure", "provider", providerID)
424+ slog.Info("Using token refreshed by another session after exchange failure", "provider", providerID)
348− return s.applyToken(providerConfig, diskToken, providerID)
425+ return s.applyToken(providerConfig, diskToken, providerID)
426+ }
349 }
427 }
350 return fmt.Errorf("failed to refresh OAuth token for provider %s: %w", providerID, refreshErr)
428 return fmt.Errorf("failed to refresh OAuth token for provider %s: %w", providerID, refreshErr)
351 }
429 }
internal/config/store_test.go+57
@@ -3,6 +3,7 @@ package config
3import (
3import (
4 "context"
4 "context"
5 "errors"
5 "errors"
6+ "fmt"
6 "os"
7 "os"
7 "path/filepath"
8 "path/filepath"
8 "testing"
9 "testing"
@@ -11,6 +12,7 @@ import (
11 "github.com/charmbracelet/crush/internal/csync"
12 "github.com/charmbracelet/crush/internal/csync"
12 "github.com/charmbracelet/crush/internal/oauth"
13 "github.com/charmbracelet/crush/internal/oauth"
13 "github.com/stretchr/testify/require"
14 "github.com/stretchr/testify/require"
15+ "github.com/tidwall/gjson"
14)
16)
15
17
16func TestConfigStore_ConfigPath_GlobalAlwaysWorks(t *testing.T) {
18func TestConfigStore_ConfigPath_GlobalAlwaysWorks(t *testing.T) {
@@ -729,3 +731,58 @@ func TestRefreshOAuthToken_UsesDiskTokenWhenDifferent(t *testing.T) {
729 require.Equal(t, "newer-access-token", updatedConfig.OAuthToken.AccessToken)
731 require.Equal(t, "newer-access-token", updatedConfig.OAuthToken.AccessToken)
730 require.Equal(t, "refresh-abc", updatedConfig.OAuthToken.RefreshToken)
732 require.Equal(t, "refresh-abc", updatedConfig.OAuthToken.RefreshToken)
731}
733}
734+
735+// TestConfigStore_SetConfigFields_concurrent verifies that concurrent writes do
736+// not lose data when protected by the in-process mutex and cross-process flock.
737+func TestConfigStore_SetConfigFields_concurrent(t *testing.T) {
738+ t.Parallel()
739+
740+ dir := t.TempDir()
741+ configPath := filepath.Join(dir, "crush.json")
742+ require.NoError(t, os.MkdirAll(filepath.Dir(configPath), 0o755))
743+ require.NoError(t, os.WriteFile(configPath, []byte("{}"), 0o600))
744+
745+ store := &ConfigStore{
746+ config: &Config{
747+ Providers: csync.NewMap[string, ProviderConfig](),
748+ Models: make(map[SelectedModelType]SelectedModel),
749+ },
750+ globalDataPath: configPath,
751+ workingDir: dir,
752+ }
753+
754+ const (
755+ numGoroutines = 20
756+ fieldsPerRoutine = 5
757+ )
758+
759+ errs := make(chan error, numGoroutines)
760+ for i := 0; i < numGoroutines; i++ {
761+ go func(id int) {
762+ kv := make(map[string]any, fieldsPerRoutine)
763+ for j := 0; j < fieldsPerRoutine; j++ {
764+ key := fmt.Sprintf("goroutine_%d_field_%d", id, j)
765+ kv[key] = fmt.Sprintf("value_%d_%d", id, j)
766+ }
767+ errs <- store.SetConfigFields(ScopeGlobal, kv)
768+ }(i)
769+ }
770+
771+ for i := 0; i < numGoroutines; i++ {
772+ require.NoError(t, <-errs)
773+ }
774+
775+ // Verify all fields are present in the config file.
776+ data, err := os.ReadFile(configPath)
777+ require.NoError(t, err)
778+
779+ for i := 0; i < numGoroutines; i++ {
780+ for j := 0; j < fieldsPerRoutine; j++ {
781+ key := fmt.Sprintf("goroutine_%d_field_%d", i, j)
782+ expectedValue := fmt.Sprintf("value_%d_%d", i, j)
783+ result := gjson.Get(string(data), key)
784+ require.True(t, result.Exists(), "key %s should exist", key)
785+ require.Equal(t, expectedValue, result.String(), "key %s should have the correct value", key)
786+ }
787+ }
788+}
internal/db/connect_test.go+13−12
@@ -6,6 +6,7 @@ import (
6 "path/filepath"
6 "path/filepath"
7 "testing"
7 "testing"
8
8
9+ "github.com/charmbracelet/crush/internal/lock"
9 "github.com/stretchr/testify/require"
10 "github.com/stretchr/testify/require"
10)
11)
11
12
@@ -65,7 +66,7 @@ func TestConnect_FailsWhenDataDirLocked(t *testing.T) {
65 dataDir := t.TempDir()
66 dataDir := t.TempDir()
66 lockPath := filepath.Join(dataDir, dataDirLockFile)
67 lockPath := filepath.Join(dataDir, dataDirLockFile)
67
68
68− release, err := tryFileLock(lockPath)
69+ release, err := lock.TryFile(lockPath)
69 require.NoError(t, err, "expected to take the data-dir lock for the first time")
70 require.NoError(t, err, "expected to take the data-dir lock for the first time")
70 t.Cleanup(release)
71 t.Cleanup(release)
71
72
@@ -82,7 +83,7 @@ func TestConnect_SucceedsAfterContenderReleases(t *testing.T) {
82 dataDir := t.TempDir()
83 dataDir := t.TempDir()
83 lockPath := filepath.Join(dataDir, dataDirLockFile)
84 lockPath := filepath.Join(dataDir, dataDirLockFile)
84
85
85− release, err := tryFileLock(lockPath)
86+ release, err := lock.TryFile(lockPath)
86 require.NoError(t, err)
87 require.NoError(t, err)
87
88
88 _, err = Connect(context.Background(), dataDir, WithDataDirLock(true))
89 _, err = Connect(context.Background(), dataDir, WithDataDirLock(true))
@@ -110,16 +111,16 @@ func TestConnect_LockReleasedOnFinalRelease(t *testing.T) {
110 require.NoError(t, conn.PingContext(context.Background()))
111 require.NoError(t, conn.PingContext(context.Background()))
111
112
112 // Holding the in-process entry must keep the OS lock held so a
113 // Holding the in-process entry must keep the OS lock held so a
113− // "second process" (simulated by a fresh tryFileLock call) is
114+ // "second process" (simulated by a fresh lock.TryFile call) is
114 // rejected.
115 // rejected.
115− _, lockErr := tryFileLock(lockPath)
116+ _, lockErr := lock.TryFile(lockPath)
116 require.Error(t, lockErr)
117 require.Error(t, lockErr)
117− require.True(t, errors.Is(lockErr, errLockContended), "expected contended lock while pool entry is live")
118+ require.True(t, errors.Is(lockErr, lock.ErrContended), "expected contended lock while pool entry is live")
118
119
119 require.NoError(t, Release(dataDir))
120 require.NoError(t, Release(dataDir))
120
121
121 // After the final release the lock is free again.
122 // After the final release the lock is free again.
122− release, err := tryFileLock(lockPath)
123+ release, err := lock.TryFile(lockPath)
123 require.NoError(t, err, "expected lock to be released after final Release")
124 require.NoError(t, err, "expected lock to be released after final Release")
124 release()
125 release()
125}
126}
@@ -143,8 +144,8 @@ func TestConnect_SharedPoolDoesNotReacquireLock(t *testing.T) {
143
144
144 // Drop one reference; lock must still be held.
145 // Drop one reference; lock must still be held.
145 require.NoError(t, Release(dataDir))
146 require.NoError(t, Release(dataDir))
146− _, lockErr := tryFileLock(lockPath)
147+ _, lockErr := lock.TryFile(lockPath)
147− require.ErrorIs(t, lockErr, errLockContended)
148+ require.ErrorIs(t, lockErr, lock.ErrContended)
148
149
149 require.NoError(t, Release(dataDir))
150 require.NoError(t, Release(dataDir))
150}
151}
@@ -157,7 +158,7 @@ func TestConnect_SkipLockEnvBypassesAcquisition(t *testing.T) {
157 dataDir := t.TempDir()
158 dataDir := t.TempDir()
158 lockPath := filepath.Join(dataDir, dataDirLockFile)
159 lockPath := filepath.Join(dataDir, dataDirLockFile)
159
160
160− release, err := tryFileLock(lockPath)
161+ release, err := lock.TryFile(lockPath)
161 require.NoError(t, err)
162 require.NoError(t, err)
162 t.Cleanup(release)
163 t.Cleanup(release)
163
164
@@ -171,7 +172,7 @@ func TestConnect_SkipLockEnvBypassesAcquisition(t *testing.T) {
171
172
172// TestConnect_DefaultIgnoresContendedLock confirms that without
173// TestConnect_DefaultIgnoresContendedLock confirms that without
173// WithDataDirLock(true) the lock file is irrelevant: a contender can
174// WithDataDirLock(true) the lock file is irrelevant: a contender can
174−// hold tryFileLock and Connect still succeeds. This pins the
175+// hold lock.TryFile and Connect still succeeds. This pins the
175// local-mode default to its pre-lock behavior.
176// local-mode default to its pre-lock behavior.
176func TestConnect_DefaultIgnoresContendedLock(t *testing.T) {
177func TestConnect_DefaultIgnoresContendedLock(t *testing.T) {
177 t.Cleanup(ResetPool)
178 t.Cleanup(ResetPool)
@@ -179,7 +180,7 @@ func TestConnect_DefaultIgnoresContendedLock(t *testing.T) {
179 dataDir := t.TempDir()
180 dataDir := t.TempDir()
180 lockPath := filepath.Join(dataDir, dataDirLockFile)
181 lockPath := filepath.Join(dataDir, dataDirLockFile)
181
182
182− release, err := tryFileLock(lockPath)
183+ release, err := lock.TryFile(lockPath)
183 require.NoError(t, err, "expected to take the data-dir lock for the first time")
184 require.NoError(t, err, "expected to take the data-dir lock for the first time")
184 t.Cleanup(release)
185 t.Cleanup(release)
185
186
@@ -199,7 +200,7 @@ func TestConnect_ServerPathFailsWhenDataDirLocked(t *testing.T) {
199 dataDir := t.TempDir()
200 dataDir := t.TempDir()
200 lockPath := filepath.Join(dataDir, dataDirLockFile)
201 lockPath := filepath.Join(dataDir, dataDirLockFile)
201
202
202− release, err := tryFileLock(lockPath)
203+ release, err := lock.TryFile(lockPath)
203 require.NoError(t, err, "expected to take the data-dir lock for the first time")
204 require.NoError(t, err, "expected to take the data-dir lock for the first time")
204 t.Cleanup(release)
205 t.Cleanup(release)
205
206
internal/db/datadirlock.go+3−2
@@ -10,6 +10,7 @@ import (
10 "strconv"
10 "strconv"
11 "time"
11 "time"
12
12
13+ "github.com/charmbracelet/crush/internal/lock"
13 "github.com/charmbracelet/crush/internal/version"
14 "github.com/charmbracelet/crush/internal/version"
14)
15)
15
16
@@ -53,9 +54,9 @@ func acquireDataDirLock(dataDir string) (*dataDirLock, error) {
53 }
54 }
54
55
55 path := filepath.Join(dataDir, dataDirLockFile)
56 path := filepath.Join(dataDir, dataDirLockFile)
56− release, err := tryFileLock(path)
57+ release, err := lock.TryFile(path)
57 if err != nil {
58 if err != nil {
58− if errors.Is(err, errLockContended) {
59+ if errors.Is(err, lock.ErrContended) {
59 return nil, contendedLockError(dataDir, path)
60 return nil, contendedLockError(dataDir, path)
60 }
61 }
61 return nil, fmt.Errorf("failed to lock data directory %q: %w", dataDir, err)
62 return nil, fmt.Errorf("failed to lock data directory %q: %w", dataDir, err)
internal/db/datadirlock_unix.godeleted−45
@@ -1,45 +0,0 @@
1−//go:build !windows
2−
3−package db
4−
5−import (
6− "errors"
7− "fmt"
8− "os"
9−
10− "golang.org/x/sys/unix"
11−)
12−
13−// errLockContended is returned by tryFileLock when the lock is already
14−// held by another open file description (typically another process).
15−var errLockContended = errors.New("file lock is held by another process")
16−
17−// tryFileLock takes an exclusive non-blocking BSD flock on path,
18−// creating the file if necessary. On success it returns a release
19−// function that drops the lock and closes the descriptor. When the
20−// lock is contended it returns errLockContended.
21−//
22−// BSD flock is advisory and per-open-file-description, so it does not
23−// interfere with the byte-range locks SQLite itself uses on the same
24−// file's siblings (crush.db, crush.db-wal, crush.db-shm). The lock is
25−// also released automatically by the kernel when the file descriptor
26−// is closed, including on process crash, so we do not need any
27−// explicit stale-lock recovery.
28−func tryFileLock(path string) (func(), error) {
29− f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o600)
30− if err != nil {
31− return nil, fmt.Errorf("open lock file: %w", err)
32− }
33− if err := unix.Flock(int(f.Fd()), unix.LOCK_EX|unix.LOCK_NB); err != nil {
34− _ = f.Close()
35− if errors.Is(err, unix.EWOULDBLOCK) {
36− return nil, errLockContended
37− }
38− return nil, fmt.Errorf("flock: %w", err)
39− }
40− return func() {
41− // Closing the descriptor releases the flock atomically.
42− _ = unix.Flock(int(f.Fd()), unix.LOCK_UN)
43− _ = f.Close()
44− }, nil
45−}
internal/db/datadirlock_windows.godeleted−46
@@ -1,46 +0,0 @@
1−//go:build windows
2−
3−package db
4−
5−import (
6− "errors"
7− "fmt"
8− "math"
9− "os"
10−
11− "golang.org/x/sys/windows"
12−)
13−
14−// errLockContended is returned by tryFileLock when the lock is held
15−// by another process.
16−var errLockContended = errors.New("file lock is held by another process")
17−
18−// tryFileLock takes an exclusive non-blocking lock on path via
19−// LockFileEx. On success it returns a release function that unlocks
20−// and closes the descriptor.
21−//
22−// The flags combine LOCKFILE_EXCLUSIVE_LOCK with LOCKFILE_FAIL_IMMEDIATELY
23−// to mirror the BSD LOCK_EX|LOCK_NB semantics used on POSIX. The lock
24−// is released when the file handle closes, including on process exit,
25−// which gives us automatic stale-lock recovery without any bookkeeping.
26−func tryFileLock(path string) (func(), error) {
27− f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o600)
28− if err != nil {
29− return nil, fmt.Errorf("open lock file: %w", err)
30− }
31− h := windows.Handle(f.Fd())
32− ol := new(windows.Overlapped)
33− flags := uint32(windows.LOCKFILE_EXCLUSIVE_LOCK | windows.LOCKFILE_FAIL_IMMEDIATELY)
34− if err := windows.LockFileEx(h, flags, 0, math.MaxUint32, math.MaxUint32, ol); err != nil {
35− _ = f.Close()
36− if errors.Is(err, windows.ERROR_LOCK_VIOLATION) || errors.Is(err, windows.ERROR_IO_PENDING) {
37− return nil, errLockContended
38− }
39− return nil, fmt.Errorf("LockFileEx: %w", err)
40− }
41− return func() {
42− ol := new(windows.Overlapped)
43− _ = windows.UnlockFileEx(windows.Handle(f.Fd()), 0, math.MaxUint32, math.MaxUint32, ol)
44− _ = f.Close()
45− }, nil
46−}
internal/lock/lock.gonew file+75
@@ -0,0 +1,75 @@
1+// Package lock provides cross-process advisory file locking.
2+//
3+// File acquires an exclusive lock on the file at path, blocking until
4+// the context is cancelled (or its deadline elapses). TryFile does the
5+// same but returns ErrContended immediately if the lock is already
6+// held. In both cases the returned release function drops the lock and
7+// closes the underlying file descriptor.
8+//
9+// The lock is released automatically by the kernel on process
10+// termination (including crash), so no stale-lock recovery is needed.
11+//
12+// The lock file at path is created if it does not exist. It is never
13+// unlinked — flock is keyed by inode, not path, and unlinking could
14+// create a window where two processes lock different inodes at the
15+// same path.
16+//
17+// This is the canonical file-locking helper for Crush. Callers should
18+// prefer it over rolling their own platform-specific code.
19+package lock
20+
21+import (
22+ "context"
23+ "errors"
24+ "fmt"
25+ "os"
26+)
27+
28+// ErrContended is returned by TryFile when the lock is already held by
29+// another process.
30+var ErrContended = errors.New("file lock is held by another process")
31+
32+// File acquires an exclusive advisory lock on the file at path, blocking
33+// until the lock is acquired or ctx is cancelled. It returns a release
34+// function that drops the lock and closes the underlying file descriptor.
35+//
36+// Pass a context with a deadline (e.g. context.WithTimeout) to bound the
37+// wait. Pass context.Background() to block indefinitely.
38+func File(ctx context.Context, path string) (func(), error) {
39+ f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o600)
40+ if err != nil {
41+ return nil, fmt.Errorf("open lock file %q: %w", path, err)
42+ }
43+
44+ release, err := lockFile(ctx, f)
45+ if err != nil {
46+ f.Close()
47+ return nil, err
48+ }
49+
50+ return func() {
51+ release()
52+ f.Close()
53+ }, nil
54+}
55+
56+// TryFile is like File but returns ErrContended immediately if the lock
57+// is already held by another process. Use this when you want to fail
58+// fast rather than wait.
59+func TryFile(path string) (func(), error) {
60+ f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o600)
61+ if err != nil {
62+ return nil, fmt.Errorf("open lock file %q: %w", path, err)
63+ }
64+
65+ release, err := tryLockFile(f)
66+ if err != nil {
67+ f.Close()
68+ return nil, err
69+ }
70+
71+ return func() {
72+ release()
73+ f.Close()
74+ }, nil
75+}
internal/lock/lock_test.gonew file+168
@@ -0,0 +1,168 @@
1+package lock
2+
3+import (
4+ "context"
5+ "errors"
6+ "path/filepath"
7+ "sync"
8+ "testing"
9+ "time"
10+
11+ "github.com/stretchr/testify/require"
12+)
13+
14+func TestTryFile_AcquiresWhenFree(t *testing.T) {
15+ t.Parallel()
16+ path := filepath.Join(t.TempDir(), "test.lock")
17+
18+ release, err := TryFile(path)
19+ require.NoError(t, err)
20+ require.NotNil(t, release)
21+ release()
22+}
23+
24+func TestTryFile_ReturnsErrContendedWhenHeld(t *testing.T) {
25+ t.Parallel()
26+ path := filepath.Join(t.TempDir(), "test.lock")
27+
28+ release, err := TryFile(path)
29+ require.NoError(t, err)
30+ t.Cleanup(release)
31+
32+ _, err = TryFile(path)
33+ require.ErrorIs(t, err, ErrContended)
34+}
35+
36+func TestTryFile_ReacquireAfterRelease(t *testing.T) {
37+ t.Parallel()
38+ path := filepath.Join(t.TempDir(), "test.lock")
39+
40+ release, err := TryFile(path)
41+ require.NoError(t, err)
42+ release()
43+
44+ release2, err := TryFile(path)
45+ require.NoError(t, err)
46+ t.Cleanup(release2)
47+}
48+
49+func TestFile_AcquiresWhenFree(t *testing.T) {
50+ t.Parallel()
51+ path := filepath.Join(t.TempDir(), "test.lock")
52+
53+ release, err := File(context.Background(), path)
54+ require.NoError(t, err)
55+ t.Cleanup(release)
56+}
57+
58+func TestFile_BlocksThenSucceeds(t *testing.T) {
59+ t.Parallel()
60+ path := filepath.Join(t.TempDir(), "test.lock")
61+
62+ release, err := TryFile(path)
63+ require.NoError(t, err)
64+
65+ // Release the lock after a short delay so the blocking acquirer
66+ // can complete within the test timeout.
67+ go func() {
68+ time.Sleep(150 * time.Millisecond)
69+ release()
70+ }()
71+
72+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
73+ defer cancel()
74+ release2, err := File(ctx, path)
75+ require.NoError(t, err, "should acquire after first releases")
76+ release2()
77+}
78+
79+func TestFile_RespectsContextDeadline(t *testing.T) {
80+ t.Parallel()
81+ path := filepath.Join(t.TempDir(), "test.lock")
82+
83+ release, err := TryFile(path)
84+ require.NoError(t, err)
85+ t.Cleanup(release)
86+
87+ ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
88+ defer cancel()
89+ start := time.Now()
90+ _, err = File(ctx, path)
91+ elapsed := time.Since(start)
92+ require.Error(t, err)
93+ require.True(t, errors.Is(err, context.DeadlineExceeded), "expected deadline exceeded, got %v", err)
94+ require.Less(t, elapsed, 1*time.Second, "should return promptly after deadline")
95+}
96+
97+func TestFile_RespectsContextCancellation(t *testing.T) {
98+ t.Parallel()
99+ path := filepath.Join(t.TempDir(), "test.lock")
100+
101+ release, err := TryFile(path)
102+ require.NoError(t, err)
103+ t.Cleanup(release)
104+
105+ ctx, cancel := context.WithCancel(context.Background())
106+ done := make(chan error, 1)
107+ go func() {
108+ _, err := File(ctx, path)
109+ done <- err
110+ }()
111+
112+ time.Sleep(50 * time.Millisecond)
113+ cancel()
114+
115+ select {
116+ case err := <-done:
117+ require.ErrorIs(t, err, context.Canceled)
118+ case <-time.After(2 * time.Second):
119+ t.Fatal("File did not return after context cancellation")
120+ }
121+}
122+
123+// TestFile_ConcurrentAcquirers verifies that multiple blocking acquirers
124+// queue up correctly: each gets the lock in turn, exactly one at a time.
125+func TestFile_ConcurrentAcquirers(t *testing.T) {
126+ t.Parallel()
127+ path := filepath.Join(t.TempDir(), "test.lock")
128+
129+ const n = 5
130+ var (
131+ mu sync.Mutex
132+ inside int
133+ maxSeen int
134+ finished int
135+ )
136+ var wg sync.WaitGroup
137+ wg.Add(n)
138+ for range n {
139+ go func() {
140+ defer wg.Done()
141+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
142+ defer cancel()
143+ release, err := File(ctx, path)
144+ if err != nil {
145+ t.Errorf("acquire failed: %v", err)
146+ return
147+ }
148+ mu.Lock()
149+ inside++
150+ if inside > maxSeen {
151+ maxSeen = inside
152+ }
153+ mu.Unlock()
154+
155+ time.Sleep(20 * time.Millisecond)
156+
157+ mu.Lock()
158+ inside--
159+ finished++
160+ mu.Unlock()
161+ release()
162+ }()
163+ }
164+ wg.Wait()
165+
166+ require.Equal(t, n, finished)
167+ require.Equal(t, 1, maxSeen, "lock must be mutually exclusive")
168+}
internal/lock/lock_unix.gonew file+45
@@ -0,0 +1,45 @@
1+//go:build !windows
2+
3+package lock
4+
5+import (
6+ "context"
7+ "errors"
8+ "fmt"
9+ "os"
10+ "time"
11+
12+ "golang.org/x/sys/unix"
13+)
14+
15+// retrySleep is the interval between non-blocking flock retries in the
16+// blocking File path. Small enough that contention resolution feels
17+// snappy; large enough that we don't burn a CPU spinning.
18+const retrySleep = 100 * time.Millisecond
19+
20+func lockFile(ctx context.Context, f *os.File) (func(), error) {
21+ for {
22+ err := unix.Flock(int(f.Fd()), unix.LOCK_EX|unix.LOCK_NB)
23+ if err == nil {
24+ return func() { _ = unix.Flock(int(f.Fd()), unix.LOCK_UN) }, nil
25+ }
26+ if !errors.Is(err, unix.EWOULDBLOCK) {
27+ return nil, fmt.Errorf("flock: %w", err)
28+ }
29+ select {
30+ case <-ctx.Done():
31+ return nil, fmt.Errorf("acquire lock: %w", ctx.Err())
32+ case <-time.After(retrySleep):
33+ }
34+ }
35+}
36+
37+func tryLockFile(f *os.File) (func(), error) {
38+ if err := unix.Flock(int(f.Fd()), unix.LOCK_EX|unix.LOCK_NB); err != nil {
39+ if errors.Is(err, unix.EWOULDBLOCK) {
40+ return nil, ErrContended
41+ }
42+ return nil, fmt.Errorf("flock: %w", err)
43+ }
44+ return func() { _ = unix.Flock(int(f.Fd()), unix.LOCK_UN) }, nil
45+}
internal/lock/lock_windows.gonew file+57
@@ -0,0 +1,57 @@
1+//go:build windows
2+
3+package lock
4+
5+import (
6+ "context"
7+ "errors"
8+ "fmt"
9+ "math"
10+ "os"
11+ "time"
12+
13+ "golang.org/x/sys/windows"
14+)
15+
16+// retrySleep is the interval between non-blocking lock retries in the
17+// blocking File path.
18+const retrySleep = 100 * time.Millisecond
19+
20+func lockFile(ctx context.Context, f *os.File) (func(), error) {
21+ h := windows.Handle(f.Fd())
22+ for {
23+ ol := new(windows.Overlapped)
24+ flags := uint32(windows.LOCKFILE_EXCLUSIVE_LOCK | windows.LOCKFILE_FAIL_IMMEDIATELY)
25+ err := windows.LockFileEx(h, flags, 0, math.MaxUint32, math.MaxUint32, ol)
26+ if err == nil {
27+ return func() {
28+ ol := new(windows.Overlapped)
29+ _ = windows.UnlockFileEx(windows.Handle(f.Fd()), 0, math.MaxUint32, math.MaxUint32, ol)
30+ }, nil
31+ }
32+ if !errors.Is(err, windows.ERROR_LOCK_VIOLATION) && !errors.Is(err, windows.ERROR_IO_PENDING) {
33+ return nil, fmt.Errorf("LockFileEx: %w", err)
34+ }
35+ select {
36+ case <-ctx.Done():
37+ return nil, fmt.Errorf("acquire lock: %w", ctx.Err())
38+ case <-time.After(retrySleep):
39+ }
40+ }
41+}
42+
43+func tryLockFile(f *os.File) (func(), error) {
44+ h := windows.Handle(f.Fd())
45+ ol := new(windows.Overlapped)
46+ flags := uint32(windows.LOCKFILE_EXCLUSIVE_LOCK | windows.LOCKFILE_FAIL_IMMEDIATELY)
47+ if err := windows.LockFileEx(h, flags, 0, math.MaxUint32, math.MaxUint32, ol); err != nil {
48+ if errors.Is(err, windows.ERROR_LOCK_VIOLATION) || errors.Is(err, windows.ERROR_IO_PENDING) {
49+ return nil, ErrContended
50+ }
51+ return nil, fmt.Errorf("LockFileEx: %w", err)
52+ }
53+ return func() {
54+ ol := new(windows.Overlapped)
55+ _ = windows.UnlockFileEx(windows.Handle(f.Fd()), 0, math.MaxUint32, math.MaxUint32, ol)
56+ }, nil
57+}