@@ -9,6 +9,8 @@ import (
99 "io"
1010 "io/fs"
1111 "log/slog"
12+ "net"
13+ "net/http"
1214 "net/url"
1315 "os"
1416 "os/exec"
@@ -357,22 +359,7 @@ func connectToServer(cmd *cobra.Command) (*client.Client, *proto.Workspace, func
357359
358360 ws, err := c.CreateWorkspace(ctx, wsReq)
359361 if err != nil {
360− // The server socket may exist before the HTTP handler is ready.
361− // Retry a few times with a short backoff.
362− for range 5 {
363− select {
364− case <-ctx.Done():
365− return nil, nil, nil, ctx.Err()
366− case <-time.After(200 * time.Millisecond):
367− }
368− ws, err = c.CreateWorkspace(ctx, wsReq)
369− if err == nil {
370− break
371− }
372− }
373− if err != nil {
374− return nil, nil, nil, fmt.Errorf("failed to create workspace: %v", err)
375− }
362+ return nil, nil, nil, fmt.Errorf("failed to create workspace: %v", err)
376363 }
377364
378365 if shouldEnableMetrics(ws.Config) {
@@ -396,55 +383,239 @@ func ensureServer(cmd *cobra.Command, hostURL *url.URL) error {
396383 switch hostURL.Scheme {
397384 case "unix", "npipe":
398385 needsStart := false
399− if _, err := os.Stat(hostURL.Host); err != nil && errors.Is(err, fs.ErrNotExist) {
386+ _, statErr := os.Stat(hostURL.Host)
387+ switch {
388+ case statErr == nil:
389+ restarted, err := restartIfStale(cmd, hostURL)
390+ if err != nil {
391+ slog.Warn("Failed to check server version", "error", err)
392+ }
393+ needsStart = restarted || err != nil
394+ case errors.Is(statErr, fs.ErrNotExist):
400395 needsStart = true
401− } else if err == nil {
402− if err := restartIfStale(cmd, hostURL); err != nil {
403− slog.Warn("Failed to check server version, restarting", "error", err)
404− needsStart = true
396+ default:
397+ slog.Warn("Unexpected error stat'ing server socket, attempting cleanup",
398+ "path", hostURL.Host, "error", statErr)
399+ if err := os.Remove(hostURL.Host); err != nil && !errors.Is(err, fs.ErrNotExist) {
400+ return fmt.Errorf("failed to remove stale server socket %q: %v", hostURL.Host, err)
405401 }
402+ needsStart = true
406403 }
407404
408405 if needsStart {
409− if err := startDetachedServer(cmd); err != nil {
410− return err
406+ if err := spawnAndWaitReady(cmd, hostURL); err != nil {
407+ return fmt.Errorf("failed to initialize crush server: %v", err)
411408 }
409+ return nil
412410 }
413411
414− var err error
415− for range 10 {
416− _, err = os.Stat(hostURL.Host)
417− if err == nil {
418− break
419− }
420− select {
421− case <-cmd.Context().Done():
422− return cmd.Context().Err()
423− case <-time.After(100 * time.Millisecond):
412+ if err := waitForServerReady(cmd.Context(), hostURL); err != nil {
413+ return fmt.Errorf("failed to initialize crush server: %v", err)
414+ }
415+ }
416+
417+ return nil
418+}
419+
420+// spawnAndWaitReady serializes the spawn-and-wait-for-readiness sequence
421+// across concurrent clients via an exclusive flock on
422+// $XDG_CACHE_HOME/crush/server-<safeHost>/start.lock.
423+//
424+// After acquiring the lock it re-probes readiness so that a client that
425+// blocked while another client was spawning can skip its own spawn and
426+// just use the now-running server. The lock is held only for the
427+// duration of "spawn + readiness probe" and released before the caller
428+// resumes its normal lifetime.
429+func spawnAndWaitReady(cmd *cobra.Command, hostURL *url.URL) error {
430+ chDir, err := perHostServerDir(hostURL)
431+ if err != nil {
432+ return err
433+ }
434+ release, err := acquireSpawnLock(filepath.Join(chDir, "start.lock"))
435+ if err != nil {
436+ // If the lock itself is unavailable, fall back to the
437+ // unsynchronized path rather than blocking the user.
438+ slog.Warn("Failed to acquire spawn lock, proceeding without single-flight", "error", err)
439+ if err := startDetachedServer(cmd, hostURL); err != nil {
440+ return err
441+ }
442+ return waitForServerReady(cmd.Context(), hostURL)
443+ }
444+ defer release()
445+
446+ // Another client may have just finished spawning while we were
447+ // waiting on the lock; if the server is already responsive, skip
448+ // the spawn entirely.
449+ probeCtx, cancel := context.WithTimeout(cmd.Context(), 200*time.Millisecond)
450+ probeErr := quickHealthProbe(probeCtx, hostURL)
451+ cancel()
452+ if probeErr == nil {
453+ return nil
454+ }
455+
456+ if err := startDetachedServer(cmd, hostURL); err != nil {
457+ return err
458+ }
459+ return waitForServerReady(cmd.Context(), hostURL)
460+}
461+
462+// quickHealthProbe issues a single readiness request with the caller's
463+// context and returns nil iff the server is responsive right now.
464+func quickHealthProbe(ctx context.Context, hostURL *url.URL) error {
465+ httpClient, reqURL, err := readinessHTTPClient(hostURL)
466+ if err != nil {
467+ return err
468+ }
469+ return probeHealth(ctx, httpClient, reqURL, hostURL)
470+}
471+
472+// perHostServerDir returns (and creates) the cache directory used for
473+// per-host server state (logs, start.lock, etc.). The path is derived
474+// from the parsed host URL rather than the global flag so the same key
475+// is computed regardless of where the host came from.
476+func perHostServerDir(hostURL *url.URL) (string, error) {
477+ chDir := filepath.Join(config.GlobalCacheDir(), "server-"+safeHostName(hostURL))
478+ if err := os.MkdirAll(chDir, 0o700); err != nil {
479+ return "", fmt.Errorf("failed to create server working directory: %v", err)
480+ }
481+ return chDir, nil
482+}
483+
484+// safeHostName returns a filesystem-safe identifier for hostURL,
485+// suitable for use as a directory name. It mirrors the input shape of
486+// the --host flag so client and server compute the same key.
487+func safeHostName(hostURL *url.URL) string {
488+ return safeNameRegexp.ReplaceAllString(
489+ hostURL.Scheme+"://"+hostURL.Host+hostURL.Path, "_")
490+}
491+
492+// serverReadyTimeout returns the total budget for the readiness probe.
493+// Overridable via CRUSH_SERVER_READY_TIMEOUT (parsed as a Go duration).
494+func serverReadyTimeout() time.Duration {
495+ const def = 10 * time.Second
496+ v := os.Getenv("CRUSH_SERVER_READY_TIMEOUT")
497+ if v == "" {
498+ return def
499+ }
500+ d, err := time.ParseDuration(v)
501+ if err != nil || d <= 0 {
502+ return def
503+ }
504+ return d
505+}
506+
507+// waitForServerReady polls GET /v1/health until the server responds with
508+// any 2xx status or the total timeout elapses. Each attempt uses a short
509+// per-attempt timeout so a hung listener doesn't burn the whole budget.
510+//
511+// The HTTP transport is built to mirror how *client.Client dials so the
512+// same unix socket / npipe / tcp setups all work uniformly here.
513+func waitForServerReady(ctx context.Context, hostURL *url.URL) error {
514+ httpClient, reqURL, err := readinessHTTPClient(hostURL)
515+ if err != nil {
516+ return err
517+ }
518+
519+ const perAttempt = 100 * time.Millisecond
520+ deadline := time.Now().Add(serverReadyTimeout())
521+
522+ var lastErr error
523+ for {
524+ if err := ctx.Err(); err != nil {
525+ return err
526+ }
527+ if time.Now().After(deadline) {
528+ if lastErr != nil {
529+ return lastErr
424530 }
531+ return fmt.Errorf("timed out waiting for server readiness")
425532 }
426− if err != nil {
427− return fmt.Errorf("failed to initialize crush server: %v", err)
533+
534+ attemptCtx, cancel := context.WithTimeout(ctx, perAttempt)
535+ err := probeHealth(attemptCtx, httpClient, reqURL, hostURL)
536+ cancel()
537+ if err == nil {
538+ return nil
539+ }
540+ lastErr = err
541+
542+ select {
543+ case <-ctx.Done():
544+ return ctx.Err()
545+ case <-time.After(perAttempt):
428546 }
429547 }
548+}
430549
550+// readinessHTTPClient builds an *http.Client whose transport dials the
551+// server using the same scheme-aware logic as *client.Client (unix
552+// socket, named pipe, or tcp).
553+func readinessHTTPClient(hostURL *url.URL) (*http.Client, string, error) {
554+ c, err := client.NewClient("", hostURL.Scheme, hostURL.Host)
555+ if err != nil {
556+ return nil, "", err
557+ }
558+
559+ tr := http.DefaultTransport.(*http.Transport).Clone()
560+ tr.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) {
561+ return c.Dial(ctx, network, addr)
562+ }
563+ if hostURL.Scheme == "unix" || hostURL.Scheme == "npipe" {
564+ tr.DisableCompression = true
565+ }
566+
567+ httpClient := &http.Client{Transport: tr}
568+
569+ // For unix sockets / named pipes we still need a syntactically valid
570+ // HTTP URL; the actual address is resolved by the dialer.
571+ host := hostURL.Host
572+ if hostURL.Scheme == "unix" || hostURL.Scheme == "npipe" {
573+ host = client.DummyHost
574+ }
575+ reqURL := (&url.URL{Scheme: "http", Host: host, Path: "/v1/health"}).String()
576+ return httpClient, reqURL, nil
577+}
578+
579+// probeHealth issues a single GET to the readiness endpoint and treats
580+// any 2xx response as success.
581+func probeHealth(ctx context.Context, h *http.Client, reqURL string, hostURL *url.URL) error {
582+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL, nil)
583+ if err != nil {
584+ return err
585+ }
586+ if hostURL.Scheme == "unix" || hostURL.Scheme == "npipe" {
587+ req.Host = client.DummyHost
588+ }
589+ rsp, err := h.Do(req)
590+ if err != nil {
591+ return err
592+ }
593+ defer rsp.Body.Close()
594+ _, _ = io.Copy(io.Discard, rsp.Body)
595+ if rsp.StatusCode < 200 || rsp.StatusCode >= 300 {
596+ return fmt.Errorf("server health check failed: %s", rsp.Status)
597+ }
431598 return nil
432599}
433600
434601// restartIfStale checks whether the running server matches the current
435602// client version. When they differ, it sends a shutdown command and
436603// removes the stale socket so the caller can start a fresh server.
437−func restartIfStale(cmd *cobra.Command, hostURL *url.URL) error {
604+//
605+// It returns restarted=true when it has shut down a stale server and the
606+// caller must spawn a new one. When the server matches the client version
607+// (or the check itself fails), restarted is false.
608+func restartIfStale(cmd *cobra.Command, hostURL *url.URL) (restarted bool, err error) {
438609 c, err := client.NewClient("", hostURL.Scheme, hostURL.Host)
439610 if err != nil {
440− return err
611+ return false, err
441612 }
442613 vi, err := c.VersionInfo(cmd.Context())
443614 if err != nil {
444− return err
615+ return false, err
445616 }
446617 if vi.Version == version.Version {
447− return nil
618+ return false, nil
448619 }
449620 slog.Info("Server version mismatch, restarting",
450621 "server", vi.Version,
@@ -458,27 +629,26 @@ func restartIfStale(cmd *cobra.Command, hostURL *url.URL) error {
458629 }
459630 select {
460631 case <-cmd.Context().Done():
461− return cmd.Context().Err()
632+ return true, cmd.Context().Err()
462633 case <-time.After(100 * time.Millisecond):
463634 }
464635 }
465636 // Force-remove if the socket is still lingering.
466637 _ = os.Remove(hostURL.Host)
467− return nil
638+ return true, nil
468639}
469640
470641var safeNameRegexp = regexp.MustCompile(`[^a-zA-Z0-9._-]`)
471642
472−func startDetachedServer(cmd *cobra.Command) error {
643+func startDetachedServer(cmd *cobra.Command, hostURL *url.URL) error {
473644 exe, err := os.Executable()
474645 if err != nil {
475646 return fmt.Errorf("failed to get executable path: %v", err)
476647 }
477648
478− safeClientHost := safeNameRegexp.ReplaceAllString(clientHost, "_")
479− chDir := filepath.Join(config.GlobalCacheDir(), "server-"+safeClientHost)
480− if err := os.MkdirAll(chDir, 0o700); err != nil {
481− return fmt.Errorf("failed to create server working directory: %v", err)
649+ chDir, err := perHostServerDir(hostURL)
650+ if err != nil {
651+ return err
482652 }
483653
484654 cmdArgs := []string{"server"}
@@ -486,7 +656,11 @@ func startDetachedServer(cmd *cobra.Command) error {
486656 cmdArgs = append(cmdArgs, "--host", clientHost)
487657 }
488658
489− c := exec.CommandContext(cmd.Context(), exe, cmdArgs...)
659+ // Use context.Background() so the parent's context cancellation does not
660+ // kill the spawned server. detachProcess (Setsid on !windows,
661+ // DETACHED_PROCESS on windows) is what truly detaches the child from
662+ // this process's lifetime.
663+ c := exec.CommandContext(context.Background(), exe, cmdArgs...)
490664 stdoutPath := filepath.Join(chDir, "stdout.log")
491665 stderrPath := filepath.Join(chDir, "stderr.log")
492666 detachProcess(c)