[AV-1829]: Add node process failure management (#150)

* start adding some checks for node stop

* avoid closing the notification chan

* get to work health up to a good point

* notification of node failure events to user he decides

* use new channel in example

* fix unit tests

* add stopped node check on server healthy check

* keep dead status of process and let used find out on that

* add management of node failure on server

* fix unit tests

* fix e2e test

* Update local/node.go

Co-authored-by: Dan Laine <daniel.laine@avalabs.org>

* fix after applying suggested changes

* add more sync to local node process

* check this causes unit test error at CI

* add state to process manage struct

* add process exit code to notification msg

* fix stuff related to changing notification msg

* fix unit test

* add ctx to node Wait, network RemoveNode

* fix unit test

* kill all descendant of wait context failure

* address PR comment

* move context wait to Stop

* fix unit test issue

* use Await4 to avoid race conditions

* improve comments

* nit

* rm comments regarding Ctrl+C special case and improve code for that

* address PR comment

* move node process to its own file

* add missing file

* reduce number of channels

* address PR comments

* address PR comments

* change Alive method for Status method

* address PR comments

* avoid using of syscal

* address PR comments

* avoid syscall SIGNAL stuff

* address PR comments

* fix examples delay

* fix log format; go mod tidy

* Dan's pass; remove unexpectedStopChan (#158)

* move status to its own package; nits

* combine newNodeProcess and Start

* update nodeProcess

* remove unexpected stop chan

* nit

* fix test

* fix local network stop context cancellation

* fix node stop deadlock

* make Stop return exit code rather than error

* remove return value from killDescendants

* nits; bump timeout

* lint

* add wait time to Status to be notified on common avalanchego startup
failures

* fix lock in Status()

* fix lock code

* increase first status call wait time

* remove first call to status check

Co-authored-by: Dan Laine <daniel.laine@avalabs.org>

examples/local/fivenodenetwork/main.go

@@ -33,7 +33,7 @@ func shutdownOnSignal(
 	sig := <-signalChan
 	log.Info("got OS signal %s", sig)
 	if err := n.Stop(context.Background()); err != nil {
-		log.Debug("error while stopping network: %s", err)
+		log.Info("error stopping network: %s", err)
 	}
 	signal.Reset()
 	close(signalChan)
@@ -73,7 +73,7 @@ func run(log logging.Logger, binaryPath string) error {
 	}
 	defer func() { // Stop the network when this function returns
 		if err := nw.Stop(context.Background()); err != nil {
-			log.Debug("error stopping network: %w", err)
+			log.Info("error stopping network: %s", err)
 		}
 	}()
 

examples/local/indepth/main.go

@@ -17,7 +17,8 @@ import (
 )
 
 const (
-	healthyTimeout = 2 * time.Minute
+	healthyTimeout    = 2 * time.Minute
+	removeNodeTimeout = 10 * time.Second
 )
 
 var goPath = os.ExpandEnv("$GOPATH")
@@ -35,7 +36,7 @@ func shutdownOnSignal(
 	sig := <-signalChan
 	log.Info("got OS signal %s", sig)
 	if err := n.Stop(context.Background()); err != nil {
-		log.Debug("error while stopping network: %s", err)
+		log.Info("error stopping network: %s", err)
 	}
 	signal.Reset()
 	close(signalChan)
@@ -77,7 +78,7 @@ func run(log logging.Logger, binaryPath string) error {
 	}
 	defer func() { // Stop the network when this function returns
 		if err := nw.Stop(context.Background()); err != nil {
-			log.Debug("error stopping network: %w", err)
+			log.Info("error stopping network: %s", err)
 		}
 	}()
 
@@ -96,7 +97,6 @@ func run(log logging.Logger, binaryPath string) error {
 	log.Info("waiting for all nodes to report healthy...")
 	if err := nw.Healthy(ctx); err != nil {
 		return err
-
 	}
 
 	// Print the node names
@@ -143,7 +143,9 @@ func run(log logging.Logger, binaryPath string) error {
 	// Remove one node
 	nodeToRemove := nodeNames[3]
 	log.Info("removing node %q", nodeToRemove)
-	if err := nw.RemoveNode(nodeToRemove); err != nil {
+	removeNodeCtx, removeNodeCtxCancel := context.WithTimeout(context.Background(), removeNodeTimeout)
+	defer removeNodeCtxCancel()
+	if err := nw.RemoveNode(removeNodeCtx, nodeToRemove); err != nil {
 		return err
 	}
 

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.10.3
 	github.com/onsi/ginkgo/v2 v2.1.3
 	github.com/onsi/gomega v1.19.0
+	github.com/shirou/gopsutil v3.21.4-0.20210419000835-c7a38de76ee5+incompatible
 	github.com/otiai10/copy v1.7.0
 	github.com/prometheus/client_golang v1.12.2
 	github.com/spf13/cobra v1.3.0
@@ -82,7 +83,6 @@ require (
 	github.com/prometheus/procfs v0.7.3 // indirect
 	github.com/rjeczalik/notify v0.9.2 // indirect
 	github.com/rs/cors v1.7.0 // indirect
-	github.com/shirou/gopsutil v3.21.4-0.20210419000835-c7a38de76ee5+incompatible // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
 	github.com/spf13/afero v1.6.0 // indirect
 	github.com/spf13/cast v1.4.1 // indirect

local/mocks/node_process.go

@@ -1,52 +1,56 @@
-// Code generated by mockery v2.9.4. DO NOT EDIT.
+// Code generated by mockery v2.12.0. DO NOT EDIT.
 
 package mocks
 
-import mock "github.com/stretchr/testify/mock"
+import (
+	context "context"
+
+	mock "github.com/stretchr/testify/mock"
+
+	status "github.com/ava-labs/avalanche-network-runner/network/node/status"
+
+	testing "testing"
+)
 
 // NodeProcess is an autogenerated mock type for the NodeProcess type
 type NodeProcess struct {
 	mock.Mock
 }
 
-// Start provides a mock function with given fields:
-func (_m *NodeProcess) Start() error {
+// Status provides a mock function with given fields:
+func (_m *NodeProcess) Status() status.Status {
 	ret := _m.Called()
 
-	var r0 error
-	if rf, ok := ret.Get(0).(func() error); ok {
+	var r0 status.Status
+	if rf, ok := ret.Get(0).(func() status.Status); ok {
 		r0 = rf()
 	} else {
-		r0 = ret.Error(0)
+		r0 = ret.Get(0).(status.Status)
 	}
 
 	return r0
 }
 
-// Stop provides a mock function with given fields:
-func (_m *NodeProcess) Stop() error {
-	ret := _m.Called()
+// Stop provides a mock function with given fields: ctx
+func (_m *NodeProcess) Stop(ctx context.Context) int {
+	ret := _m.Called(ctx)
 
-	var r0 error
-	if rf, ok := ret.Get(0).(func() error); ok {
-		r0 = rf()
+	var r0 int
+	if rf, ok := ret.Get(0).(func(context.Context) int); ok {
+		r0 = rf(ctx)
 	} else {
-		r0 = ret.Error(0)
+		r0 = ret.Get(0).(int)
 	}
 
 	return r0
 }
 
-// Wait provides a mock function with given fields:
-func (_m *NodeProcess) Wait() error {
-	ret := _m.Called()
+// NewNodeProcess creates a new instance of NodeProcess. It also registers the testing.TB interface on the mock and a cleanup function to assert the mocks expectations.
+func NewNodeProcess(t testing.TB) *NodeProcess {
+	mock := &NodeProcess{}
+	mock.Mock.Test(t)
 
-	var r0 error
-	if rf, ok := ret.Get(0).(func() error); ok {
-		r0 = rf()
-	} else {
-		r0 = ret.Error(0)
-	}
+	t.Cleanup(func() { mock.AssertExpectations(t) })
 
-	return r0
+	return mock
 }

local/network.go

@@ -20,6 +20,7 @@ import (
 	"github.com/ava-labs/avalanche-network-runner/api"
 	"github.com/ava-labs/avalanche-network-runner/network"
 	"github.com/ava-labs/avalanche-network-runner/network/node"
+	"github.com/ava-labs/avalanche-network-runner/network/node/status"
 	"github.com/ava-labs/avalanche-network-runner/utils"
 	"github.com/ava-labs/avalanchego/config"
 	"github.com/ava-labs/avalanchego/staking"
@@ -198,10 +199,11 @@ func init() {
 
 // NodeProcessCreator is an interface for new node process creation
 type NodeProcessCreator interface {
-	NewNodeProcess(config node.Config, args ...string) (NodeProcess, error)
+	NewNodeProcess(config node.Config, log logging.Logger, args ...string) (NodeProcess, error)
 }
 
 type nodeProcessCreator struct {
+	log logging.Logger
 	// If this node's stdout or stderr are redirected, [colorPicker] determines
 	// the color of logs printed to stdout and/or stderr
 	colorPicker utils.ColorPicker
@@ -216,7 +218,7 @@ type nodeProcessCreator struct {
 // NewNodeProcess creates a new process of the passed binary
 // If the config has redirection set to `true` for either StdErr or StdOut,
 // the output will be redirected and colored
-func (npc *nodeProcessCreator) NewNodeProcess(config node.Config, args ...string) (NodeProcess, error) {
+func (npc *nodeProcessCreator) NewNodeProcess(config node.Config, log logging.Logger, args ...string) (NodeProcess, error) {
 	// Start the AvalancheGo node and pass it the flags defined above
 	cmd := exec.Command(config.BinaryPath, args...)
 	// assign a new color to this process (might not be used if the config isn't set for it)
@@ -238,7 +240,7 @@ func (npc *nodeProcessCreator) NewNodeProcess(config node.Config, args ...string
 		// redirect stderr and assign a color to the text
 		utils.ColorAndPrepend(stderr, npc.stderr, config.Name, color)
 	}
-	return &nodeProcessImpl{cmd: cmd}, nil
+	return newNodeProcess(config.Name, npc.log, cmd)
 }
 
 // NewNetwork returns a new network that uses the given log.
@@ -257,6 +259,7 @@ func NewNetwork(
 		api.NewAPIClient,
 		&nodeProcessCreator{
 			colorPicker: utils.NewColorPicker(),
+			log:         log,
 			stdout:      os.Stdout,
 			stderr:      os.Stderr,
 		},
@@ -497,14 +500,14 @@ func (ln *localNetwork) addNode(nodeConfig node.Config) (node.Node, error) {
 	}
 
 	// Start the AvalancheGo node and pass it the flags defined above
-	nodeProcess, err := ln.nodeProcessCreator.NewNodeProcess(nodeConfig, nodeData.flags...)
+	nodeProcess, err := ln.nodeProcessCreator.NewNodeProcess(nodeConfig, ln.log, nodeData.flags...)
 	if err != nil {
-		return nil, fmt.Errorf("couldn't create new node process: %s", err)
+		return nil, fmt.Errorf(
+			"couldn't create new node process with binary %q and flags %v: %w",
+			nodeConfig.BinaryPath, nodeData.flags, err,
+		)
 	}
 	ln.log.Debug("starting node %q with \"%s %s\"", nodeConfig.Name, nodeConfig.BinaryPath, nodeData.flags)
-	if err := nodeProcess.Start(); err != nil {
-		return nil, fmt.Errorf("could not execute cmd \"%s %s\": %w", nodeConfig.BinaryPath, nodeData.flags, err)
-	}
 
 	// Create a wrapper for this node so we can reference it later
 	node := &localNode{
@@ -563,18 +566,24 @@ func (ln *localNetwork) Healthy(ctx context.Context) error {
 	errGr, ctx := errgroup.WithContext(ctx)
 	for _, node := range ln.nodes {
 		node := node
+		nodeName := node.GetName()
 		errGr.Go(func() error {
 			// Every [healthCheckFreq], query node for health status.
 			// Do this until ctx timeout or network closed.
 			for {
+				if node.Status() != status.Running {
+					// If we had stopped this node ourselves, it wouldn't be in [ln.nodes].
+					// Since it is, it means the node stopped unexpectedly.
+					return fmt.Errorf("node %q stopped unexpectedly", nodeName)
+				}
 				health, err := node.client.HealthAPI().Health(ctx)
 				if err == nil && health.Healthy {
-					ln.log.Debug("node %q became healthy", node.name)
+					ln.log.Debug("node %q became healthy", nodeName)
 					return nil
 				}
 				select {
 				case <-ctx.Done():
-					return fmt.Errorf("node %q failed to become healthy within timeout, or network stopped", node.GetName())
+					return fmt.Errorf("node %q failed to become healthy within timeout, or network stopped", nodeName)
 				case <-time.After(healthCheckFreq):
 				}
 			}
@@ -651,40 +660,31 @@ func (ln *localNetwork) Stop(ctx context.Context) error {
 
 // Assumes [ln.lock] is held.
 func (ln *localNetwork) stop(ctx context.Context) error {
-	ctx, cancel := context.WithTimeout(ctx, stopTimeout)
-	defer cancel()
 	errs := wrappers.Errs{}
 	for nodeName := range ln.nodes {
-		select {
-		case <-ctx.Done():
-			// In practice we'll probably never time out here,
-			// and the caller probably won't cancel a call
-			// to stop(), but we include this to respect the
-			// network.Network interface.
-			return ctx.Err()
-		default:
-		}
-		if err := ln.removeNode(nodeName); err != nil {
+		stopCtx, stopCtxCancel := context.WithTimeout(ctx, stopTimeout)
+		if err := ln.removeNode(stopCtx, nodeName); err != nil {
 			ln.log.Error("error stopping node %q: %s", nodeName, err)
 			errs.Add(err)
 		}
+		stopCtxCancel()
 	}
 	ln.log.Info("done stopping network")
 	return errs.Err
 }
 
 // Sends a SIGTERM to the given node and removes it from this network.
-func (ln *localNetwork) RemoveNode(nodeName string) error {
+func (ln *localNetwork) RemoveNode(ctx context.Context, nodeName string) error {
 	ln.lock.Lock()
 	defer ln.lock.Unlock()
 	if ln.stopCalled() {
 		return network.ErrStopped
 	}
-	return ln.removeNode(nodeName)
+	return ln.removeNode(ctx, nodeName)
 }
 
 // Assumes [ln.lock] is held.
-func (ln *localNetwork) removeNode(nodeName string) error {
+func (ln *localNetwork) removeNode(ctx context.Context, nodeName string) error {
 	ln.log.Debug("removing node %q", nodeName)
 	node, ok := ln.nodes[nodeName]
 	if !ok {
@@ -698,11 +698,8 @@ func (ln *localNetwork) removeNode(nodeName string) error {
 	// cchain eth api uses a websocket connection and must be closed before stopping the node,
 	// to avoid errors logs at client
 	node.client.CChainEthAPI().Close()
-	if err := node.process.Stop(); err != nil {
-		return fmt.Errorf("error sending SIGTERM to node %s: %w", nodeName, err)
-	}
-	if err := node.process.Wait(); err != nil {
-		return fmt.Errorf("node %q stopped with error: %w", nodeName, err)
+	if exitCode := node.process.Stop(ctx); exitCode != 0 {
+		return fmt.Errorf("node %q exited with exit code: %d", nodeName, exitCode)
 	}
 	return nil
 }

local/network_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/ava-labs/avalanche-network-runner/local/mocks"
 	"github.com/ava-labs/avalanche-network-runner/network"
 	"github.com/ava-labs/avalanche-network-runner/network/node"
+	"github.com/ava-labs/avalanche-network-runner/network/node/status"
 	"github.com/ava-labs/avalanche-network-runner/utils"
 	"github.com/ava-labs/avalanchego/api/health"
 	healthmocks "github.com/ava-labs/avalanchego/api/health/mocks"
@@ -44,23 +45,19 @@ var (
 
 type localTestSuccessfulNodeProcessCreator struct{}
 
-func (*localTestSuccessfulNodeProcessCreator) NewNodeProcess(config node.Config, flags ...string) (NodeProcess, error) {
-	return newMockProcessSuccessful(config, flags...)
+func (*localTestSuccessfulNodeProcessCreator) NewNodeProcess(config node.Config, log logging.Logger, flags ...string) (NodeProcess, error) {
+	return newMockProcessSuccessful(config, log, flags...)
 }
 
 type localTestFailedStartProcessCreator struct{}
 
-func (*localTestFailedStartProcessCreator) NewNodeProcess(config node.Config, flags ...string) (NodeProcess, error) {
-	process := &mocks.NodeProcess{}
-	process.On("Start").Return(errors.New("Start failed"))
-	process.On("Wait").Return(nil)
-	process.On("Stop").Return(nil)
-	return process, nil
+func (*localTestFailedStartProcessCreator) NewNodeProcess(config node.Config, log logging.Logger, flags ...string) (NodeProcess, error) {
+	return nil, errors.New("error on purpose for test")
 }
 
 type localTestProcessUndefNodeProcessCreator struct{}
 
-func (*localTestProcessUndefNodeProcessCreator) NewNodeProcess(config node.Config, flags ...string) (NodeProcess, error) {
+func (*localTestProcessUndefNodeProcessCreator) NewNodeProcess(config node.Config, log logging.Logger, flags ...string) (NodeProcess, error) {
 	return newMockProcessUndef(config, flags...)
 }
 
@@ -69,11 +66,11 @@ type localTestFlagCheckProcessCreator struct {
 	assert        *assert.Assertions
 }
 
-func (lt *localTestFlagCheckProcessCreator) NewNodeProcess(config node.Config, flags ...string) (NodeProcess, error) {
+func (lt *localTestFlagCheckProcessCreator) NewNodeProcess(config node.Config, log logging.Logger, flags ...string) (NodeProcess, error) {
 	if ok := lt.assert.EqualValues(lt.expectedFlags, config.Flags); !ok {
 		return nil, errors.New("assertion failed: flags not equal value")
 	}
-	return newMockProcessSuccessful(config, flags...)
+	return newMockProcessSuccessful(config, log, flags...)
 }
 
 // Returns an API client where:
@@ -110,11 +107,11 @@ func newMockProcessUndef(node.Config, ...string) (NodeProcess, error) {
 }
 
 // Returns a NodeProcess that always returns nil
-func newMockProcessSuccessful(node.Config, ...string) (NodeProcess, error) {
+func newMockProcessSuccessful(node.Config, logging.Logger, ...string) (NodeProcess, error) {
 	process := &mocks.NodeProcess{}
-	process.On("Start").Return(nil)
 	process.On("Wait").Return(nil)
-	process.On("Stop").Return(nil)
+	process.On("Stop", mock.Anything).Return(0)
+	process.On("Status").Return(status.Running)
 	return process, nil
 }
 
@@ -160,7 +157,7 @@ func newLocalTestOneNodeCreator(assert *assert.Assertions, networkConfig network
 
 // Assert that the node's config is being passed correctly
 // to the function that starts the node process.
-func (lt *localTestOneNodeCreator) NewNodeProcess(config node.Config, flags ...string) (NodeProcess, error) {
+func (lt *localTestOneNodeCreator) NewNodeProcess(config node.Config, log logging.Logger, flags ...string) (NodeProcess, error) {
 	lt.assert.True(config.IsBeacon)
 	expectedConfig := lt.networkConfig.NodeConfigs[0]
 	lt.assert.EqualValues(expectedConfig.ChainConfigFiles, config.ChainConfigFiles)
@@ -176,7 +173,7 @@ func (lt *localTestOneNodeCreator) NewNodeProcess(config node.Config, flags ...s
 		lt.assert.True(ok)
 		lt.assert.EqualValues(v, gotV)
 	}
-	return lt.successCreator.NewNodeProcess(config, flags...)
+	return lt.successCreator.NewNodeProcess(config, log, flags...)
 }
 
 // Start a network with one node.
@@ -601,7 +598,7 @@ func TestNetworkNodeOps(t *testing.T) {
 	for _, nodeConfig := range networkConfig.NodeConfigs {
 		_, err := net.GetNode(nodeConfig.Name)
 		assert.NoError(err)
-		err = net.RemoveNode(nodeConfig.Name)
+		err = net.RemoveNode(context.Background(), nodeConfig.Name)
 		assert.NoError(err)
 		removedNodes[nodeConfig.Name] = struct{}{}
 		delete(runningNodes, nodeConfig.Name)
@@ -630,16 +627,16 @@ func TestNodeNotFound(t *testing.T) {
 	_, err = net.GetNode(networkConfig.NodeConfigs[1].Name)
 	assert.Error(err)
 	// remove non-existent node
-	err = net.RemoveNode(networkConfig.NodeConfigs[1].Name)
+	err = net.RemoveNode(context.Background(), networkConfig.NodeConfigs[1].Name)
 	assert.Error(err)
 	// remove node
-	err = net.RemoveNode(networkConfig.NodeConfigs[0].Name)
+	err = net.RemoveNode(context.Background(), networkConfig.NodeConfigs[0].Name)
 	assert.NoError(err)
 	// get removed node
 	_, err = net.GetNode(networkConfig.NodeConfigs[0].Name)
 	assert.Error(err)
 	// remove already-removed node
-	err = net.RemoveNode(networkConfig.NodeConfigs[0].Name)
+	err = net.RemoveNode(context.Background(), networkConfig.NodeConfigs[0].Name)
 	assert.Error(err)
 }
 
@@ -673,7 +670,7 @@ func TestStoppedNetwork(t *testing.T) {
 	_, err = net.GetNodeNames()
 	assert.EqualValues(network.ErrStopped, err)
 	// RemoveNode failure
-	assert.EqualValues(network.ErrStopped, net.RemoveNode(networkConfig.NodeConfigs[0].Name))
+	assert.EqualValues(network.ErrStopped, net.RemoveNode(context.Background(), networkConfig.NodeConfigs[0].Name))
 	// Healthy failure
 	assert.EqualValues(awaitNetworkHealthy(net, defaultHealthyTimeout), network.ErrStopped)
 	_, err = net.GetAllNodes()
@@ -801,7 +798,7 @@ type lockedBuffer struct {
 
 // Write is locked for the lockedBuffer
 func (m *lockedBuffer) Write(b []byte) (int, error) {
-	defer func() { close(m.writtenCh) }()
+	defer close(m.writtenCh)
 	return m.Buffer.Write(b)
 }
 
@@ -815,6 +812,7 @@ func TestChildCmdRedirection(t *testing.T) {
 		writtenCh: make(chan struct{}),
 	}
 	npc := &nodeProcessCreator{
+		log:         logging.NoLog{},
 		stdout:      buf,
 		stderr:      buf,
 		colorPicker: utils.NewColorPicker(),
@@ -836,43 +834,37 @@ func TestChildCmdRedirection(t *testing.T) {
 
 	// now create the node process and check it will be prepended and colored
 	testConfig := node.Config{
-		BinaryPath:     "echo",
+		BinaryPath:     "sh",
 		RedirectStdout: true,
 		RedirectStderr: true,
 		Name:           mockNodeName,
 	}
-	proc, err := npc.NewNodeProcess(testConfig, testOutput)
+	// Sleep for a second after echoing so that we have a chance to read from the stdout pipe
+	// before it closes when the process exits and Wait() returns.
+	// See https://pkg.go.dev/os/exec#Cmd.StdoutPipe
+	proc, err := npc.NewNodeProcess(testConfig, logging.NoLog{}, "-c", fmt.Sprintf("echo %s && sleep 1", testOutput))
 	if err != nil {
 		t.Fatal(err)
 	}
-	if err = proc.Start(); err != nil {
-		t.Fatal(err)
-	}
 
 	// lock read access to the buffer
 	<-buf.writtenCh
-	newResult := buf.String()
+	result := buf.String()
 
 	// wait for the process to finish.
-	// Note that, according to the specification of StdoutPipe
-	// and StderrPipe, we have to wait until after we read from
-	// the pipe before calling Wait.
-	// See https://pkg.go.dev/os/exec#Cmd.StdoutPipe
-	if err = proc.Wait(); err != nil {
-		t.Fatal(err)
-	}
+	_ = proc.Stop(context.Background())
 
 	// now do the checks:
 	// the new string should contain the node name
-	if !strings.Contains(newResult, mockNodeName) {
+	if !strings.Contains(result, mockNodeName) {
 		t.Fatalf("expected subcommand to contain node name %s, but it didn't", mockNodeName)
 	}
 
 	// and it should have a specific length:
 	//             the actual output   + the color terminal escape sequence      + node name    + []<space> + color terminal reset escape sequence
 	expectedLen := len(expectedResult) + len(utils.NewColorPicker().NextColor()) + len(mockNodeName) + 3 + len(logging.Reset)
-	if len(newResult) != expectedLen {
-		t.Fatalf("expected string length to be %d, but it was %d", expectedLen, len(newResult))
+	if len(result) != expectedLen {
+		t.Fatalf("expected string length to be %d, but it was %d", expectedLen, len(result))
 	}
 }
 
@@ -1250,7 +1242,7 @@ func TestRemoveBeacon(t *testing.T) {
 	assert.NoError(err)
 
 	// remove the beacon node from the network
-	err = net.RemoveNode(networkConfig.NodeConfigs[0].Name)
+	err = net.RemoveNode(context.Background(), networkConfig.NodeConfigs[0].Name)
 	assert.NoError(err)
 	assert.Equal(0, net.bootstraps.Len())
 }

local/node.go

@@ -5,13 +5,12 @@ import (
 	"crypto"
 	"fmt"
 	"net"
-	"os/exec"
 	"path/filepath"
-	"syscall"
 	"time"
 
 	"github.com/ava-labs/avalanche-network-runner/api"
 	"github.com/ava-labs/avalanche-network-runner/network/node"
+	"github.com/ava-labs/avalanche-network-runner/network/node/status"
 	"github.com/ava-labs/avalanchego/ids"
 	"github.com/ava-labs/avalanchego/message"
 	"github.com/ava-labs/avalanchego/network/peer"
@@ -29,47 +28,18 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
-// interface compliance
 var (
-	_ node.Node   = (*localNode)(nil)
-	_ NodeProcess = (*nodeProcessImpl)(nil)
 	_ getConnFunc = defaultGetConnFunc
+	_ node.Node   = (*localNode)(nil)
 )
 
 type getConnFunc func(context.Context, node.Node) (net.Conn, error)
 
-// NodeProcess as an interface so we can mock running
-// AvalancheGo binaries in tests
-type NodeProcess interface {
-	// Start this process
-	Start() error
-	// Send a SIGTERM to this process
-	Stop() error
-	// Returns when the process finishes exiting
-	Wait() error
-}
-
 const (
 	peerMsgQueueBufferSize      = 1024
 	peerResourceTrackerDuration = 10 * time.Second
 )
 
-type nodeProcessImpl struct {
-	cmd *exec.Cmd
-}
-
-func (p *nodeProcessImpl) Start() error {
-	return p.cmd.Start()
-}
-
-func (p *nodeProcessImpl) Wait() error {
-	return p.cmd.Wait()
-}
-
-func (p *nodeProcessImpl) Stop() error {
-	return p.cmd.Process.Signal(syscall.SIGTERM)
-}
-
 // Gives access to basic node info, and to most avalanchego apis
 type localNode struct {
 	// Must be unique across all nodes in this network.
@@ -225,6 +195,10 @@ func (node *localNode) GetAPIPort() uint16 {
 	return node.apiPort
 }
 
+func (node *localNode) Status() status.Status {
+	return node.process.Status()
+}
+
 // See node.Node
 func (node *localNode) GetBinaryPath() string {
 	return node.config.BinaryPath

local/node_process.go

@@ -0,0 +1,158 @@
+package local
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/exec"
+	"sync"
+
+	"github.com/ava-labs/avalanche-network-runner/network/node/status"
+	"github.com/ava-labs/avalanchego/utils/logging"
+	"github.com/shirou/gopsutil/process"
+)
+
+var _ NodeProcess = (*nodeProcess)(nil)
+
+// NodeProcess as an interface so we can mock running
+// AvalancheGo binaries in tests
+type NodeProcess interface {
+	// Sends a SIGINT to this process and returns the process's
+	// exit code.
+	// If [ctx] is cancelled, sends a SIGKILL to this process and descendants.
+	// We assume sending a SIGKILL to a process will always successfully kill it.
+	// Subsequent calls to [Stop] have no effect.
+	Stop(ctx context.Context) int
+	// Returns the status of the process.
+	Status() status.Status
+}
+
+type nodeProcess struct {
+	name string
+	log  logging.Logger
+	lock sync.RWMutex
+	cmd  *exec.Cmd
+	// Process status
+	state status.Status
+	// Closed when the process exits.
+	closedOnStop chan struct{}
+}
+
+func newNodeProcess(name string, log logging.Logger, cmd *exec.Cmd) (*nodeProcess, error) {
+	np := &nodeProcess{
+		name:            name,
+		log:             log,
+		cmd:             cmd,
+		closedOnStop:    make(chan struct{}),
+	}
+	return np, np.start()
+}
+
+// Start this process.
+// Must only be called once.
+func (p *nodeProcess) start() error {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
+	p.state = status.Running
+	if err := p.cmd.Start(); err != nil {
+		p.state = status.Stopped
+		close(p.closedOnStop)
+		return fmt.Errorf("couldn't start process: %w", err)
+	}
+
+	go p.awaitExit()
+	return nil
+}
+
+// Wait for the process to exit.
+// When it does, update the state and close [p.closedOnStop]
+func (p *nodeProcess) awaitExit() {
+	if err := p.cmd.Wait(); err != nil {
+		p.log.Debug("node %q returned error on wait: %s", p.name, err)
+	}
+
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
+	p.state = status.Stopped
+	close(p.closedOnStop)
+}
+
+func (p *nodeProcess) Stop(ctx context.Context) int {
+	p.lock.Lock()
+
+	// The process is already stopped.
+	if p.state == status.Stopped {
+		exitCode := p.cmd.ProcessState.ExitCode()
+		p.lock.Unlock()
+		return exitCode
+	}
+
+	// There's another call to Stop executing right now.
+	// Wait for it to finish.
+	if p.state == status.Stopping {
+		p.lock.Unlock()
+		<-p.closedOnStop
+		p.lock.RLock()
+		defer p.lock.RUnlock()
+
+		return p.cmd.ProcessState.ExitCode()
+	}
+
+	p.state = status.Stopping
+	proc := p.cmd.Process
+	// We have to unlock here so that [p.awaitExit] can grab the lock
+	// and close [p.closedOnStop].
+	p.lock.Unlock()
+
+	if err := proc.Signal(os.Interrupt); err != nil {
+		p.log.Warn("sending SIGINT errored: %w", err)
+	}
+
+	select {
+	case <-ctx.Done():
+		p.log.Warn("context cancelled while waiting for node %q to stop", p.name)
+		killDescendants(int32(proc.Pid), p.log)
+		if err := proc.Signal(os.Kill); err != nil {
+			p.log.Warn("sending SIGKILL errored: %w", err)
+		}
+	case <-p.closedOnStop:
+	}
+
+	<-p.closedOnStop
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+
+	return p.cmd.ProcessState.ExitCode()
+
+}
+
+func (p *nodeProcess) Status() status.Status {
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+
+	return p.state
+}
+
+func killDescendants(pid int32, log logging.Logger) {
+	procs, err := process.Processes()
+	if err != nil {
+		log.Warn("couldn't get processes: %s", err)
+		return
+	}
+	for _, proc := range procs {
+		ppid, err := proc.Ppid()
+		if err != nil {
+			log.Warn("couldn't get process ID: %s", err)
+			continue
+		}
+		if ppid != pid {
+			continue
+		}
+		killDescendants(proc.Pid, log)
+		if err := proc.Kill(); err != nil {
+			log.Warn("error killing process %d: %s", proc.Pid, err)
+		}
+	}
+}

network/network.go

@@ -24,7 +24,7 @@ type Network interface {
 	AddNode(node.Config) (node.Node, error)
 	// Stop the node with this name.
 	// Returns ErrStopped if Stop() was previously called.
-	RemoveNode(name string) error
+	RemoveNode(ctx context.Context, name string) error
 	// Return the node with this name.
 	// Returns ErrStopped if Stop() was previously called.
 	GetNode(name string) (node.Node, error)

network/node/node.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 
 	"github.com/ava-labs/avalanche-network-runner/api"
+	"github.com/ava-labs/avalanche-network-runner/network/node/status"
 	"github.com/ava-labs/avalanchego/config"
 	"github.com/ava-labs/avalanchego/ids"
 	"github.com/ava-labs/avalanchego/network/peer"
@@ -34,6 +35,8 @@ type Node interface {
 	// It's left to the caller to maintain a reference to the returned peer.
 	// The caller should call StartClose() on the peer when they're done with it.
 	AttachPeer(ctx context.Context, handler router.InboundHandler) (peer.Peer, error)
+	// Return the state of the node process
+	Status() status.Status
 	// Return this node's avalanchego binary path
 	GetBinaryPath() string
 	// Return this node's db dir

network/node/status/status.go

@@ -0,0 +1,26 @@
+package status
+
+// The status of a node.
+type Status byte
+
+const (
+	// Process is running and hasn't been asked to stop.
+	Running Status = iota + 1
+	// Process has been asked to stop.
+	Stopping
+	// Process has exited.
+	Stopped
+)
+
+func (s Status) String() string {
+	switch s {
+	case Running:
+		return "running"
+	case Stopping:
+		return "stopping"
+	case Stopped:
+		return "stopped"
+	default:
+		return "invalid status"
+	}
+}

server/custom_vms.go

@@ -505,7 +505,7 @@ func (lc *localNetwork) restartNodesWithWhitelistedSubnets(
 
 		lc.customVMRestartMu.Lock()
 		zap.L().Info("removing and adding back the node for whitelisted subnets", zap.String("node-name", nodeName))
-		if err := lc.nw.RemoveNode(nodeName); err != nil {
+		if err := lc.nw.RemoveNode(ctx, nodeName); err != nil {
 			lc.customVMRestartMu.Unlock()
 			return err
 		}

server/server.go

@@ -82,14 +82,13 @@ var (
 	ErrNotBootstrapped                    = errors.New("not bootstrapped")
 	ErrNodeNotFound                       = errors.New("node not found")
 	ErrPeerNotFound                       = errors.New("peer not found")
-	ErrUnexpectedType                     = errors.New("unexpected type")
 	ErrStatusCanceled                     = errors.New("gRPC stream status canceled")
 )
 
 const (
 	MinNodes            uint32 = 1
 	DefaultNodes        uint32 = 5
-	StopOnSignalTimeout        = 2 * time.Second
+	stopOnSignalTimeout        = 2 * time.Second
 
 	rootDataDirPrefix = "network-runner-root-data"
 )
@@ -201,7 +200,7 @@ func (s *server) Run(rootCtx context.Context) (err error) {
 	}
 
 	if s.network != nil {
-		stopCtx, stopCtxCancel := context.WithTimeout(context.Background(), StopOnSignalTimeout)
+		stopCtx, stopCtxCancel := context.WithTimeout(context.Background(), stopOnSignalTimeout)
 		defer stopCtxCancel()
 		s.network.stop(stopCtx)
 		zap.L().Warn("network stopped")
@@ -218,15 +217,15 @@ func (s *server) Ping(ctx context.Context, req *rpcpb.PingRequest) (*rpcpb.PingR
 	return &rpcpb.PingResponse{Pid: int32(os.Getpid())}, nil
 }
 
-const DefaultStartTimeout = 5 * time.Minute
+const defaultStartTimeout = 5 * time.Minute
 
 func (s *server) Start(ctx context.Context, req *rpcpb.StartRequest) (*rpcpb.StartResponse, error) {
 	// if timeout is too small or not set, default to 5-min
-	if deadline, ok := ctx.Deadline(); !ok || time.Until(deadline) < DefaultStartTimeout {
+	if deadline, ok := ctx.Deadline(); !ok || time.Until(deadline) < defaultStartTimeout {
 		var cancel context.CancelFunc
-		ctx, cancel = context.WithTimeout(context.Background(), DefaultStartTimeout)
+		ctx, cancel = context.WithTimeout(context.Background(), defaultStartTimeout)
 		_ = cancel // don't call since "start" is async, "curl" may not specify timeout
-		zap.L().Info("received start request with default timeout", zap.String("timeout", DefaultStartTimeout.String()))
+		zap.L().Info("received start request with default timeout", zap.String("timeout", defaultStartTimeout.String()))
 	} else {
 		zap.L().Info("received start request with existing timeout", zap.String("deadline", deadline.String()))
 	}
@@ -415,11 +414,11 @@ func (s *server) waitChAndUpdateClusterInfo(waitMsg string, readyCh chan struct{
 
 func (s *server) CreateBlockchains(ctx context.Context, req *rpcpb.CreateBlockchainsRequest) (*rpcpb.CreateBlockchainsResponse, error) {
 	// if timeout is too small or not set, default to 5-min
-	if deadline, ok := ctx.Deadline(); !ok || time.Until(deadline) < DefaultStartTimeout {
+	if deadline, ok := ctx.Deadline(); !ok || time.Until(deadline) < defaultStartTimeout {
 		var cancel context.CancelFunc
-		ctx, cancel = context.WithTimeout(context.Background(), DefaultStartTimeout)
+		ctx, cancel = context.WithTimeout(context.Background(), defaultStartTimeout)
 		_ = cancel // don't call since "start" is async, "curl" may not specify timeout
-		zap.L().Info("received start request with default timeout", zap.String("timeout", DefaultStartTimeout.String()))
+		zap.L().Info("received start request with default timeout", zap.String("timeout", defaultStartTimeout.String()))
 	} else {
 		zap.L().Info("received start request with existing timeout", zap.String("deadline", deadline.String()))
 	}
@@ -507,11 +506,11 @@ func (s *server) CreateBlockchains(ctx context.Context, req *rpcpb.CreateBlockch
 
 func (s *server) CreateSubnets(ctx context.Context, req *rpcpb.CreateSubnetsRequest) (*rpcpb.CreateSubnetsResponse, error) {
 	// if timeout is too small or not set, default to 5-min
-	if deadline, ok := ctx.Deadline(); !ok || time.Until(deadline) < DefaultStartTimeout {
+	if deadline, ok := ctx.Deadline(); !ok || time.Until(deadline) < defaultStartTimeout {
 		var cancel context.CancelFunc
-		ctx, cancel = context.WithTimeout(context.Background(), DefaultStartTimeout)
+		ctx, cancel = context.WithTimeout(context.Background(), defaultStartTimeout)
 		_ = cancel // don't call since "start" is async, "curl" may not specify timeout
-		zap.L().Info("received start request with default timeout", zap.String("timeout", DefaultStartTimeout.String()))
+		zap.L().Info("received start request with default timeout", zap.String("timeout", defaultStartTimeout.String()))
 	} else {
 		zap.L().Info("received start request with existing timeout", zap.String("deadline", deadline.String()))
 	}
@@ -570,6 +569,7 @@ func (s *server) Health(ctx context.Context, req *rpcpb.HealthRequest) (*rpcpb.H
 
 	s.clusterInfo.NodeNames = s.network.nodeNames
 	s.clusterInfo.NodeInfos = s.network.nodeInfos
+	s.clusterInfo.Healthy = true
 
 	return &rpcpb.HealthResponse{ClusterInfo: s.clusterInfo}, nil
 }
@@ -802,7 +802,7 @@ func (s *server) RemoveNode(ctx context.Context, req *rpcpb.RemoveNodeRequest) (
 		return nil, ErrNodeNotFound
 	}
 
-	if err := s.network.nw.RemoveNode(req.Name); err != nil {
+	if err := s.network.nw.RemoveNode(ctx, req.Name); err != nil {
 		return nil, err
 	}
 
@@ -811,6 +811,7 @@ func (s *server) RemoveNode(ctx context.Context, req *rpcpb.RemoveNodeRequest) (
 		return nil, err
 	}
 
+	s.clusterInfo.Healthy = true
 	s.clusterInfo.NodeNames = s.network.nodeNames
 	s.clusterInfo.NodeInfos = s.network.nodeInfos
 
@@ -875,7 +876,7 @@ func (s *server) RestartNode(ctx context.Context, req *rpcpb.RestartNodeRequest)
 
 	// now remove the node before restart
 	zap.L().Info("removing the node")
-	if err := s.network.nw.RemoveNode(req.Name); err != nil {
+	if err := s.network.nw.RemoveNode(ctx, req.Name); err != nil {
 		return nil, err
 	}
 
@@ -893,6 +894,7 @@ func (s *server) RestartNode(ctx context.Context, req *rpcpb.RestartNodeRequest)
 	// update with the new config
 	s.clusterInfo.NodeNames = s.network.nodeNames
 	s.clusterInfo.NodeInfos = s.network.nodeInfos
+	s.clusterInfo.Healthy = true
 
 	return &rpcpb.RestartNodeResponse{ClusterInfo: s.clusterInfo}, nil
 }
@@ -1014,11 +1016,11 @@ func (s *server) SendOutboundMessage(ctx context.Context, req *rpcpb.SendOutboun
 
 func (s *server) LoadSnapshot(ctx context.Context, req *rpcpb.LoadSnapshotRequest) (*rpcpb.LoadSnapshotResponse, error) {
 	// if timeout is too small or not set, default to 5-min
-	if deadline, ok := ctx.Deadline(); !ok || time.Until(deadline) < DefaultStartTimeout {
+	if deadline, ok := ctx.Deadline(); !ok || time.Until(deadline) < defaultStartTimeout {
 		var cancel context.CancelFunc
-		ctx, cancel = context.WithTimeout(context.Background(), DefaultStartTimeout)
+		ctx, cancel = context.WithTimeout(context.Background(), defaultStartTimeout)
 		_ = cancel // don't call since "start" is async, "curl" may not specify timeout
-		zap.L().Info("received start request with default timeout", zap.String("timeout", DefaultStartTimeout.String()))
+		zap.L().Info("received start request with default timeout", zap.String("timeout", defaultStartTimeout.String()))
 	} else {
 		zap.L().Info("received start request with existing timeout", zap.String("deadline", deadline.String()))
 	}