Add restartable dev server lifecycle

devserver/devserver.go

@@ -103,10 +103,14 @@ type Server struct {
 	ports    Ports
 	logger   *zap.SugaredLogger
 	workDir  string
-	cancel   context.CancelFunc
-	done     chan error
-	stopOnce sync.Once
-	stopErr  error
+	launch   func(context.Context) (context.CancelFunc, chan error, error)
+	ready    func(context.Context) error
+
+	mu      sync.Mutex
+	cancel  context.CancelFunc
+	done    chan error
+	stopped bool
+	stopErr error
 }
 
 // Start clones+builds the requested ref and runs the server, returning once
@@ -130,13 +134,9 @@ func Start(ctx context.Context, opts Options) (*Server, error) {
 
 	// Roll back partially-acquired resources on any error before the Server
 	// struct takes ownership. After `success = true`, Stop() handles teardown.
-	var cancel context.CancelFunc
 	success := false
 	defer func() {
 		if !success {
-			if cancel != nil {
-				cancel()
-			}
 			_ = os.RemoveAll(workDir)
 		}
 	}()
@@ -150,7 +150,7 @@ func Start(ctx context.Context, opts Options) (*Server, error) {
 	}
 	serverPorts := newPorts(host, ports)
 	frontendAddr := net.JoinHostPort(serverPorts.Host, strconv.Itoa(serverPorts.FrontendGRPC))
-	frontendHTTPAddr := net.JoinHostPort(host, strconv.Itoa(ports[portFrontendHTTP]))
+	frontendHTTPAddr := net.JoinHostPort(serverPorts.Host, strconv.Itoa(serverPorts.FrontendHTTP))
 	dynConfigPath, err := writeDynamicConfig(workDir, frontendHTTPAddr, opts.DynamicConfigValues)
 	if err != nil {
 		return nil, fmt.Errorf("devserver: write dynamic config: %w", err)
@@ -171,45 +171,42 @@ func Start(ctx context.Context, opts Options) (*Server, error) {
 		return nil, fmt.Errorf("devserver: build: %w", err)
 	}
 
-	// Launch the server subprocess. runCtx is derived from the caller's ctx
-	// so cancelling it (or letting it expire) tears the server down. Stop()
-	// also cancels runCtx.
-	var runCtx context.Context
-	runCtx, cancel = context.WithCancel(ctx)
-	cmd := exec.CommandContext(runCtx, binaryPath,
-		"--allow-no-auth",
-		"start",
-	)
-	cmd.Env = serverEnv
 	output := cmp.Or(opts.Output, io.Discard)
-	cmd.Stdout = output
-	cmd.Stderr = output
-	cmd.Cancel = func() error { return cmd.Process.Signal(syscall.SIGTERM) }
-	cmd.WaitDelay = 15 * time.Second
-
-	opts.Logger.Infof("Starting temporal server (ref %s, frontend %s)", opts.Ref, frontendAddr)
-	if err := cmd.Start(); err != nil {
-		cancel()
-		return nil, fmt.Errorf("devserver: start: %w", err)
-	}
-
-	done := make(chan error, 1)
-	go func() { done <- cmd.Wait() }()
-
 	s := &Server{
 		frontend: frontendAddr,
 		ports:    serverPorts,
 		logger:   opts.Logger,
 		workDir:  workDir,
-		cancel:   cancel,
-		done:     done,
 	}
-	success = true
+	s.launch = func(launchCtx context.Context) (context.CancelFunc, chan error, error) {
+		runCtx, runCancel := context.WithCancel(launchCtx)
+		cmd := exec.CommandContext(runCtx, binaryPath,
+			"--allow-no-auth",
+			"start",
+		)
+		cmd.Env = serverEnv
+		cmd.Stdout = output
+		cmd.Stderr = output
+		cmd.Cancel = func() error { return cmd.Process.Signal(syscall.SIGTERM) }
+		cmd.WaitDelay = 15 * time.Second
+
+		opts.Logger.Infof("Starting temporal server (ref %s, frontend %s)", opts.Ref, frontendAddr)
+		if err := cmd.Start(); err != nil {
+			runCancel()
+			return nil, nil, err
+		}
 
-	if err := s.registerNamespace(ctx, opts.Namespace); err != nil {
-		_ = s.Stop()
-		return nil, fmt.Errorf("devserver: register namespace %q: %w", opts.Namespace, err)
+		done := make(chan error, 1)
+		go func() { done <- cmd.Wait() }()
+		return runCancel, done, nil
 	}
+	s.ready = func(readyCtx context.Context) error {
+		return s.registerNamespace(readyCtx, opts.Namespace)
+	}
+	if err := s.startProcessLocked(ctx); err != nil {
+		return nil, err
+	}
+	success = true
 	return s, nil
 }
 
@@ -224,19 +221,72 @@ func (s *Server) Ports() Ports {
 	return s.ports
 }
 
-// Stop signals the server to terminate and waits for it to exit. The
-// per-run work directory is removed. Safe to call more than once; subsequent
-// calls return the same error as the first.
+// Restart gracefully terminates the server process and starts it again with
+// the same binary, configuration, ports, output, and work directory.
+func (s *Server) Restart(ctx context.Context) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.stopped {
+		return errors.New("devserver: cannot restart a stopped server")
+	}
+	if err := s.stopProcessLocked(); err != nil {
+		return fmt.Errorf("devserver: stop for restart: %w", err)
+	}
+	if err := s.startProcessLocked(ctx); err != nil {
+		return fmt.Errorf("devserver: restart: %w", err)
+	}
+	return nil
+}
+
+// Stop signals the server to terminate and waits for it to exit. The per-run
+// work directory is removed. Stop is final and safe to call more than once;
+// subsequent calls return the same error as the first.
 func (s *Server) Stop() error {
-	s.stopOnce.Do(func() {
-		s.cancel()
-		err := <-s.done
-		_ = os.RemoveAll(s.workDir)
-		s.stopErr = classifyExitErr(err)
-	})
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.stopped {
+		return s.stopErr
+	}
+	s.stopped = true
+	s.stopErr = s.stopProcessLocked()
+	_ = os.RemoveAll(s.workDir)
 	return s.stopErr
 }
 
+func (s *Server) startProcessLocked(ctx context.Context) error {
+	cancel, done, err := s.launch(ctx)
+	if err != nil {
+		return fmt.Errorf("start: %w", err)
+	}
+	s.cancel = cancel
+	s.done = done
+	if err := s.ready(ctx); err != nil {
+		_ = s.stopProcessLocked()
+		return fmt.Errorf("wait for frontend: %w", err)
+	}
+	return nil
+}
+
+func (s *Server) stopProcessLocked() error {
+	if s.cancel == nil {
+		return nil
+	}
+	s.cancel()
+	err := <-s.done
+	s.cancel = nil
+	s.done = nil
+	return classifyStopErr(err)
+}
+
+func classifyStopErr(err error) error {
+	if errors.Is(err, context.Canceled) {
+		return nil
+	}
+	return classifyExitErr(err)
+}
+
 // classifyExitErr treats a clean exit or termination by our own SIGTERM
 // (sent by cmd.Cancel) as success, and surfaces anything else — including
 // non-zero exit codes that didn't come from a signal, and SIGKILL that
@@ -258,11 +308,12 @@ func classifyExitErr(err error) error {
 func (s *Server) registerNamespace(ctx context.Context, namespace string) error {
 	ctx, cancel := context.WithCancelCause(ctx)
 	defer cancel(nil)
+	done := s.done
 
 	go func() {
 		select {
-		case procErr := <-s.done:
-			s.done <- procErr
+		case procErr := <-done:
+			done <- procErr
 			cancel(fmt.Errorf("process exited before namespace registration completed: %w", procErr))
 		case <-ctx.Done():
 		}

devserver/devserver_restart_test.go

@@ -0,0 +1,150 @@
+package devserver
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+type fakeProcesses struct {
+	mu         sync.Mutex
+	launches   int
+	active     int
+	maxActive  int
+	failLaunch int
+	output     *bytes.Buffer
+}
+
+func (f *fakeProcesses) launch(context.Context) (context.CancelFunc, chan error, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.launches++
+	if f.launches == f.failLaunch {
+		return nil, nil, errors.New("launch failed")
+	}
+	f.active++
+	f.maxActive = max(f.maxActive, f.active)
+	_, _ = f.output.WriteString("started\n")
+	done := make(chan error, 1)
+	var once sync.Once
+	return func() {
+		once.Do(func() {
+			f.mu.Lock()
+			f.active--
+			f.mu.Unlock()
+			done <- context.Canceled
+		})
+	}, done, nil
+}
+
+func TestServerStopIgnoresCanceledCommandContext(t *testing.T) {
+	workDir := t.TempDir()
+	server := &Server{
+		cancel:  func() {},
+		done:    make(chan error, 1),
+		workDir: workDir,
+	}
+	server.done <- context.Canceled
+
+	require.NoError(t, server.Stop())
+}
+
+func newFakeServer(t *testing.T, processes *fakeProcesses) *Server {
+	t.Helper()
+	server := &Server{
+		frontend: "127.0.0.1:7233",
+		ports:    newPorts("127.0.0.1", [portCount]int{7233, 7234, 7243, 7235, 7236, 7237, 7238, 7239, 7240}),
+		workDir:  t.TempDir(),
+		launch:   processes.launch,
+		ready:    func(context.Context) error { return nil },
+	}
+	require.NoError(t, server.startProcessLocked(t.Context()))
+	return server
+}
+
+func TestServerRestartPreservesPortsAndOutput(t *testing.T) {
+	var output bytes.Buffer
+	processes := &fakeProcesses{output: &output}
+	server := newFakeServer(t, processes)
+	wantPorts := server.Ports()
+
+	require.NoError(t, server.Restart(t.Context()))
+	require.Equal(t, wantPorts, server.Ports())
+	require.Equal(t, "started\nstarted\n", output.String())
+	require.NoError(t, server.Stop())
+}
+
+func TestServerRepeatedRestarts(t *testing.T) {
+	processes := &fakeProcesses{output: &bytes.Buffer{}}
+	server := newFakeServer(t, processes)
+
+	for range 5 {
+		require.NoError(t, server.Restart(t.Context()))
+	}
+	require.NoError(t, server.Stop())
+	require.Equal(t, 6, processes.launches)
+	require.Zero(t, processes.active)
+}
+
+func TestServerRestartFailure(t *testing.T) {
+	processes := &fakeProcesses{output: &bytes.Buffer{}, failLaunch: 2}
+	server := newFakeServer(t, processes)
+
+	err := server.Restart(t.Context())
+	require.ErrorContains(t, err, "launch failed")
+	require.Zero(t, processes.active)
+	require.NoError(t, server.Stop())
+}
+
+func TestServerLifecycleCallsAreSerialized(t *testing.T) {
+	processes := &fakeProcesses{output: &bytes.Buffer{}}
+	server := newFakeServer(t, processes)
+
+	var wg sync.WaitGroup
+	for range 10 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			_ = server.Restart(t.Context())
+		}()
+	}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		_ = server.Stop()
+	}()
+	wg.Wait()
+
+	require.Equal(t, 1, processes.maxActive)
+	require.Zero(t, processes.active)
+}
+
+func TestServerRestartCancellation(t *testing.T) {
+	processes := &fakeProcesses{output: &bytes.Buffer{}}
+	server := newFakeServer(t, processes)
+	server.ready = func(ctx context.Context) error {
+		<-ctx.Done()
+		return context.Cause(ctx)
+	}
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+
+	err := server.Restart(ctx)
+	require.ErrorIs(t, err, context.Canceled)
+	require.Zero(t, processes.active)
+	require.NoError(t, server.Stop())
+}
+
+func TestServerRejectsRestartAfterStop(t *testing.T) {
+	processes := &fakeProcesses{output: &bytes.Buffer{}}
+	server := newFakeServer(t, processes)
+
+	require.NoError(t, server.Stop())
+	require.NoError(t, server.Stop())
+	require.ErrorContains(t, server.Restart(t.Context()), "stopped server")
+	require.Equal(t, 1, processes.launches)
+}