git.iko.soy/crush
on main help sign in
feat(db): refuse to open a data directory in use by another crush
6923820Christian Rocha
internal/db/connect.go+35−3
compare with main
unifiedsplit
@@ -6,6 +6,7 @@ import (
6 "embed"
6 "embed"
7 "fmt"
7 "fmt"
8 "log/slog"
8 "log/slog"
9+ "os"
9 "path/filepath"
10 "path/filepath"
10 "sync"
11 "sync"
11 "testing"
12 "testing"
@@ -38,10 +39,16 @@ func init() {
38 }
39 }
39}
40}
40
41
41−// connEntry holds a shared database connection and its reference count.
42+// connEntry holds a shared database connection, its reference count,
43+// and the data-directory lock that gates access to this entry. The
44+// lock is acquired exactly once when the entry is created and released
45+// when the last reference is dropped, which lets the same process open
46+// the same data directory concurrently while still blocking a second
47+// crush process from racing the storage.
42type connEntry struct {
48type connEntry struct {
43 db *sql.DB
49 db *sql.DB
44 refCount int
50 refCount int
51+ lock *dataDirLock
45}
52}
46
53
47var (
54var (
@@ -76,8 +83,23 @@ func Connect(ctx context.Context, dataDir string) (*sql.DB, error) {
76 return entry.db, nil
83 return entry.db, nil
77 }
84 }
78
85
86+ // Take the per-data-directory lock before opening the database so
87+ // we fail fast and with a clear error rather than racing another
88+ // crush process on the same SQLite file. The lock is released when
89+ // the matching Release call drops the refcount to zero. Ensuring
90+ // the data directory exists is required because the lock file
91+ // lives inside it.
92+ if err := os.MkdirAll(dataDir, 0o700); err != nil {
93+ return nil, fmt.Errorf("failed to create data directory %q: %w", dataDir, err)
94+ }
95+ lock, err := acquireDataDirLock(dataDir)
96+ if err != nil {
97+ return nil, err
98+ }
99+
79 conn, err := openDB(dbPath)
100 conn, err := openDB(dbPath)
80 if err != nil {
101 if err != nil {
102+ lock.release()
81 return nil, err
103 return nil, err
82 }
104 }
83
105
@@ -90,22 +112,25 @@ func Connect(ctx context.Context, dataDir string) (*sql.DB, error) {
90
112
91 if err = conn.PingContext(ctx); err != nil {
113 if err = conn.PingContext(ctx); err != nil {
92 conn.Close()
114 conn.Close()
115+ lock.release()
93 return nil, fmt.Errorf("failed to connect to database: %w", err)
116 return nil, fmt.Errorf("failed to connect to database: %w", err)
94 }
117 }
95
118
96 if err := initGoose(); err != nil {
119 if err := initGoose(); err != nil {
97 conn.Close()
120 conn.Close()
121+ lock.release()
98 slog.Error("Failed to initialize goose", "error", err)
122 slog.Error("Failed to initialize goose", "error", err)
99 return nil, fmt.Errorf("failed to initialize goose: %w", err)
123 return nil, fmt.Errorf("failed to initialize goose: %w", err)
100 }
124 }
101
125
102 if err := goose.Up(conn, "migrations"); err != nil {
126 if err := goose.Up(conn, "migrations"); err != nil {
103 conn.Close()
127 conn.Close()
128+ lock.release()
104 slog.Error("Failed to apply migrations", "error", err)
129 slog.Error("Failed to apply migrations", "error", err)
105 return nil, fmt.Errorf("failed to apply migrations: %w", err)
130 return nil, fmt.Errorf("failed to apply migrations: %w", err)
106 }
131 }
107
132
108− pool[absPath] = &connEntry{db: conn, refCount: 1}
133+ pool[absPath] = &connEntry{db: conn, refCount: 1, lock: lock}
109 return conn, nil
134 return conn, nil
110}
135}
111
136
@@ -133,7 +158,11 @@ func Release(dataDir string) error {
133 }
158 }
134
159
135 delete(pool, absPath)
160 delete(pool, absPath)
136− return entry.db.Close()
161+ closeErr := entry.db.Close()
162+ if entry.lock != nil {
163+ entry.lock.release()
164+ }
165+ return closeErr
137}
166}
138
167
139// ResetPool closes all pooled connections and clears the pool. This is
168// ResetPool closes all pooled connections and clears the pool. This is
@@ -143,6 +172,9 @@ func ResetPool() {
143 defer poolMu.Unlock()
172 defer poolMu.Unlock()
144 for path, entry := range pool {
173 for path, entry := range pool {
145 entry.db.Close()
174 entry.db.Close()
175+ if entry.lock != nil {
176+ entry.lock.release()
177+ }
146 delete(pool, path)
178 delete(pool, path)
147 }
179 }
148}
180}
internal/db/connect_test.go+116
@@ -2,6 +2,8 @@ package db
2
2
3import (
3import (
4 "context"
4 "context"
5+ "errors"
6+ "path/filepath"
5 "testing"
7 "testing"
6
8
7 "github.com/stretchr/testify/require"
9 "github.com/stretchr/testify/require"
@@ -52,3 +54,117 @@ func TestRelease_NoopForUnknownDataDir(t *testing.T) {
52
54
53 require.NoError(t, Release("/nonexistent/path"), "releasing unknown data dir should not error")
55 require.NoError(t, Release("/nonexistent/path"), "releasing unknown data dir should not error")
54}
56}
57+
58+// TestConnect_FailsWhenDataDirLocked simulates a second crush process by
59+// taking the data-dir lock directly via the OS primitive on a separate
60+// file descriptor and then asserting that Connect surfaces a clean
61+// ErrDataDirLocked instead of opening the database under contention.
62+func TestConnect_FailsWhenDataDirLocked(t *testing.T) {
63+ t.Cleanup(ResetPool)
64+
65+ dataDir := t.TempDir()
66+ lockPath := filepath.Join(dataDir, dataDirLockFile)
67+
68+ release, err := tryFileLock(lockPath)
69+ require.NoError(t, err, "expected to take the data-dir lock for the first time")
70+ t.Cleanup(release)
71+
72+ _, err = Connect(context.Background(), dataDir)
73+ require.Error(t, err, "Connect must refuse to open a locked data dir")
74+ require.ErrorIs(t, err, ErrDataDirLocked)
75+}
76+
77+// TestConnect_SucceedsAfterContenderReleases ensures the lock is purely
78+// advisory and that a clean release lets the next Connect proceed.
79+func TestConnect_SucceedsAfterContenderReleases(t *testing.T) {
80+ t.Cleanup(ResetPool)
81+
82+ dataDir := t.TempDir()
83+ lockPath := filepath.Join(dataDir, dataDirLockFile)
84+
85+ release, err := tryFileLock(lockPath)
86+ require.NoError(t, err)
87+
88+ _, err = Connect(context.Background(), dataDir)
89+ require.ErrorIs(t, err, ErrDataDirLocked)
90+
91+ release()
92+
93+ conn, err := Connect(context.Background(), dataDir)
94+ require.NoError(t, err, "Connect should succeed once the contender releases the lock")
95+ require.NoError(t, conn.PingContext(context.Background()))
96+ require.NoError(t, Release(dataDir))
97+}
98+
99+// TestConnect_LockReleasedOnFinalRelease confirms that closing the last
100+// reference to a pool entry also drops the OS lock, so subsequent
101+// processes can take the data dir.
102+func TestConnect_LockReleasedOnFinalRelease(t *testing.T) {
103+ t.Cleanup(ResetPool)
104+
105+ dataDir := t.TempDir()
106+ lockPath := filepath.Join(dataDir, dataDirLockFile)
107+
108+ conn, err := Connect(context.Background(), dataDir)
109+ require.NoError(t, err)
110+ require.NoError(t, conn.PingContext(context.Background()))
111+
112+ // Holding the in-process entry must keep the OS lock held so a
113+ // "second process" (simulated by a fresh tryFileLock call) is
114+ // rejected.
115+ _, lockErr := tryFileLock(lockPath)
116+ require.Error(t, lockErr)
117+ require.True(t, errors.Is(lockErr, errLockContended), "expected contended lock while pool entry is live")
118+
119+ require.NoError(t, Release(dataDir))
120+
121+ // After the final release the lock is free again.
122+ release, err := tryFileLock(lockPath)
123+ require.NoError(t, err, "expected lock to be released after final Release")
124+ release()
125+}
126+
127+// TestConnect_SharedPoolDoesNotReacquireLock makes sure that subsequent
128+// in-process Connect calls reuse the existing OS lock through refcount,
129+// not by re-acquiring it. The simplest observable signal of correctness
130+// is that the second Connect does not error and the lock is still held
131+// after a single Release.
132+func TestConnect_SharedPoolDoesNotReacquireLock(t *testing.T) {
133+ t.Cleanup(ResetPool)
134+
135+ dataDir := t.TempDir()
136+ lockPath := filepath.Join(dataDir, dataDirLockFile)
137+
138+ _, err := Connect(context.Background(), dataDir)
139+ require.NoError(t, err)
140+
141+ _, err = Connect(context.Background(), dataDir)
142+ require.NoError(t, err)
143+
144+ // Drop one reference; lock must still be held.
145+ require.NoError(t, Release(dataDir))
146+ _, lockErr := tryFileLock(lockPath)
147+ require.ErrorIs(t, lockErr, errLockContended)
148+
149+ require.NoError(t, Release(dataDir))
150+}
151+
152+// TestConnect_SkipLockEnvBypassesAcquisition exercises the escape
153+// hatch used by users on filesystems where flock is unreliable.
154+func TestConnect_SkipLockEnvBypassesAcquisition(t *testing.T) {
155+ t.Cleanup(ResetPool)
156+
157+ dataDir := t.TempDir()
158+ lockPath := filepath.Join(dataDir, dataDirLockFile)
159+
160+ release, err := tryFileLock(lockPath)
161+ require.NoError(t, err)
162+ t.Cleanup(release)
163+
164+ t.Setenv("CRUSH_SKIP_DATADIR_LOCK", "1")
165+
166+ conn, err := Connect(context.Background(), dataDir)
167+ require.NoError(t, err, "skip-lock env should bypass contention")
168+ require.NoError(t, conn.PingContext(context.Background()))
169+ require.NoError(t, Release(dataDir))
170+}
internal/db/datadirlock.gonew file+125
@@ -0,0 +1,125 @@
1+package db
2+
3+import (
4+ "encoding/json"
5+ "errors"
6+ "fmt"
7+ "os"
8+ "path/filepath"
9+ "strconv"
10+ "time"
11+
12+ "github.com/charmbracelet/crush/internal/version"
13+)
14+
15+// ErrDataDirLocked is returned by Connect when the data directory is
16+// already in use by another crush process.
17+var ErrDataDirLocked = errors.New("data directory already in use by another crush process")
18+
19+// dataDirLockFile is the name of the lock file inside the data
20+// directory. It lives next to crush.db so users can `ls` and find it.
21+const dataDirLockFile = "crush.lock"
22+
23+// dataDirOwnerInfo is the JSON payload written into the lock file by
24+// the process that currently owns it. It is purely informational; the
25+// authoritative state of ownership is the operating system flock on
26+// the file descriptor.
27+type dataDirOwnerInfo struct {
28+ PID int `json:"pid"`
29+ Version string `json:"version,omitempty"`
30+ StartedAt string `json:"started_at,omitempty"`
31+}
32+
33+// dataDirLock represents an acquired exclusive lock on a data
34+// directory. release closes the underlying file descriptor which the
35+// kernel uses to drop the OS-level lock.
36+type dataDirLock struct {
37+ release func()
38+}
39+
40+// acquireDataDirLock takes an exclusive non-blocking lock on
41+// {dataDir}/crush.lock. If the lock is already held by another
42+// process, it returns ErrDataDirLocked wrapped with a diagnostic that
43+// includes whatever owner info that process wrote.
44+//
45+// Acquisition is skipped (returning a no-op lock) when
46+// CRUSH_SKIP_DATADIR_LOCK is set to a truthy value. This is intended
47+// as an escape hatch for hostile filesystems that do not implement
48+// advisory locking; it should not be used in normal operation.
49+func acquireDataDirLock(dataDir string) (*dataDirLock, error) {
50+ if skipDataDirLock() {
51+ return &dataDirLock{release: func() {}}, nil
52+ }
53+
54+ path := filepath.Join(dataDir, dataDirLockFile)
55+ release, err := tryFileLock(path)
56+ if err != nil {
57+ if errors.Is(err, errLockContended) {
58+ return nil, contendedLockError(dataDir, path)
59+ }
60+ return nil, fmt.Errorf("failed to lock data directory %q: %w", dataDir, err)
61+ }
62+
63+ // Record ownership metadata so a contending process can identify
64+ // us. Failures here are non-fatal: the OS-level lock is what
65+ // actually guarantees mutual exclusion, and a missing/partial JSON
66+ // payload only degrades diagnostics.
67+ if err := writeOwnerInfo(path); err != nil {
68+ // Best-effort; log via stderr only when running in a debug
69+ // context would be invasive here, so we silently swallow.
70+ _ = err
71+ }
72+
73+ return &dataDirLock{release: release}, nil
74+}
75+
76+// skipDataDirLock reports whether the data-dir lock should be bypassed.
77+func skipDataDirLock() bool {
78+ v, _ := strconv.ParseBool(os.Getenv("CRUSH_SKIP_DATADIR_LOCK"))
79+ return v
80+}
81+
82+// writeOwnerInfo truncates and rewrites the lock file with the current
83+// process's identifying information. It is called only after the lock
84+// is held.
85+func writeOwnerInfo(path string) error {
86+ info := dataDirOwnerInfo{
87+ PID: os.Getpid(),
88+ Version: version.Version,
89+ StartedAt: time.Now().UTC().Format(time.RFC3339),
90+ }
91+ payload, err := json.MarshalIndent(info, "", " ")
92+ if err != nil {
93+ return err
94+ }
95+ payload = append(payload, '\n')
96+ return os.WriteFile(path, payload, 0o600)
97+}
98+
99+// readOwnerInfo returns the lock file's recorded owner, if it parses.
100+// A missing or malformed file yields an empty struct and no error;
101+// the caller decides what to surface to the user.
102+func readOwnerInfo(path string) dataDirOwnerInfo {
103+ raw, err := os.ReadFile(path)
104+ if err != nil || len(raw) == 0 {
105+ return dataDirOwnerInfo{}
106+ }
107+ var info dataDirOwnerInfo
108+ _ = json.Unmarshal(raw, &info)
109+ return info
110+}
111+
112+// contendedLockError builds a wrapped ErrDataDirLocked annotated with
113+// whatever owner metadata is currently in the lock file.
114+func contendedLockError(dataDir, lockPath string) error {
115+ info := readOwnerInfo(lockPath)
116+ details := ""
117+ switch {
118+ case info.PID != 0 && info.StartedAt != "":
119+ details = fmt.Sprintf(" (owner pid=%d version=%s started_at=%s)",
120+ info.PID, info.Version, info.StartedAt)
121+ case info.PID != 0:
122+ details = fmt.Sprintf(" (owner pid=%d)", info.PID)
123+ }
124+ return fmt.Errorf("%w: %s%s", ErrDataDirLocked, dataDir, details)
125+}
internal/db/datadirlock_unix.gonew file+45
@@ -0,0 +1,45 @@
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.gonew file+46
@@ -0,0 +1,46 @@
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+}