git.iko.soy/crush
on main help sign in
fix(server): serialize concurrent server spawns with a per-host lock
d41c118Christian Rocha
internal/cmd/root.go+70−6
compare with main
unifiedsplit
@@ -402,9 +402,10 @@ func ensureServer(cmd *cobra.Command, hostURL *url.URL) error {
402 }
402 }
403
403
404 if needsStart {
404 if needsStart {
405− if err := startDetachedServer(cmd); err != nil {
405+ if err := spawnAndWaitReady(cmd, hostURL); err != nil {
406− return err
406+ return fmt.Errorf("failed to initialize crush server: %v", err)
407 }
407 }
408+ return nil
408 }
409 }
409
410
410 if err := waitForServerReady(cmd.Context(), hostURL); err != nil {
411 if err := waitForServerReady(cmd.Context(), hostURL); err != nil {
@@ -415,6 +416,70 @@ func ensureServer(cmd *cobra.Command, hostURL *url.URL) error {
415 return nil
416 return nil
416}
417}
417
418
419+// spawnAndWaitReady serializes the spawn-and-wait-for-readiness sequence
420+// across concurrent clients via an exclusive flock on
421+// $XDG_CACHE_HOME/crush/server-<safeHost>/start.lock.
422+//
423+// After acquiring the lock it re-probes readiness so that a client that
424+// blocked while another client was spawning can skip its own spawn and
425+// just use the now-running server. The lock is held only for the
426+// duration of "spawn + readiness probe" and released before the caller
427+// resumes its normal lifetime.
428+func spawnAndWaitReady(cmd *cobra.Command, hostURL *url.URL) error {
429+ chDir, err := perHostServerDir()
430+ if err != nil {
431+ return err
432+ }
433+ release, err := acquireSpawnLock(filepath.Join(chDir, "start.lock"))
434+ if err != nil {
435+ // If the lock itself is unavailable, fall back to the
436+ // unsynchronized path rather than blocking the user.
437+ slog.Warn("Failed to acquire spawn lock, proceeding without single-flight", "error", err)
438+ if err := startDetachedServer(cmd); err != nil {
439+ return err
440+ }
441+ return waitForServerReady(cmd.Context(), hostURL)
442+ }
443+ defer release()
444+
445+ // Another client may have just finished spawning while we were
446+ // waiting on the lock; if the server is already responsive, skip
447+ // the spawn entirely.
448+ probeCtx, cancel := context.WithTimeout(cmd.Context(), 200*time.Millisecond)
449+ probeErr := quickHealthProbe(probeCtx, hostURL)
450+ cancel()
451+ if probeErr == nil {
452+ return nil
453+ }
454+
455+ if err := startDetachedServer(cmd); err != nil {
456+ return err
457+ }
458+ return waitForServerReady(cmd.Context(), hostURL)
459+}
460+
461+// quickHealthProbe issues a single readiness request with the caller's
462+// context and returns nil iff the server is responsive right now.
463+func quickHealthProbe(ctx context.Context, hostURL *url.URL) error {
464+ httpClient, reqURL, err := readinessHTTPClient(hostURL)
465+ if err != nil {
466+ return err
467+ }
468+ return probeHealth(ctx, httpClient, reqURL, hostURL)
469+}
470+
471+// perHostServerDir returns (and creates) the cache directory used for
472+// per-host server state (logs, start.lock, etc.). It mirrors the path
473+// computed in startDetachedServer so both code paths stay in sync.
474+func perHostServerDir() (string, error) {
475+ safeClientHost := safeNameRegexp.ReplaceAllString(clientHost, "_")
476+ chDir := filepath.Join(config.GlobalCacheDir(), "server-"+safeClientHost)
477+ if err := os.MkdirAll(chDir, 0o700); err != nil {
478+ return "", fmt.Errorf("failed to create server working directory: %v", err)
479+ }
480+ return chDir, nil
481+}
482+
418// serverReadyTimeout returns the total budget for the readiness probe.
483// serverReadyTimeout returns the total budget for the readiness probe.
419// Overridable via CRUSH_SERVER_READY_TIMEOUT (parsed as a Go duration).
484// Overridable via CRUSH_SERVER_READY_TIMEOUT (parsed as a Go duration).
420func serverReadyTimeout() time.Duration {
485func serverReadyTimeout() time.Duration {
@@ -572,10 +637,9 @@ func startDetachedServer(cmd *cobra.Command) error {
572 return fmt.Errorf("failed to get executable path: %v", err)
637 return fmt.Errorf("failed to get executable path: %v", err)
573 }
638 }
574
639
575− safeClientHost := safeNameRegexp.ReplaceAllString(clientHost, "_")
640+ chDir, err := perHostServerDir()
576− chDir := filepath.Join(config.GlobalCacheDir(), "server-"+safeClientHost)
577− if err := os.MkdirAll(chDir, 0o700); err != nil {
641+ if err != nil {
578− return fmt.Errorf("failed to create server working directory: %v", err)
642+ return err
579 }
643 }
580
644
581 cmdArgs := []string{"server"}
645 cmdArgs := []string{"server"}
internal/cmd/spawnlock_other.gonew file+28
@@ -0,0 +1,28 @@
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.gonew file+32
@@ -0,0 +1,32 @@
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+}