Fixes race conditions

Makefile

@@ -8,12 +8,18 @@ all: $(PREFIX)/ops
 $(PREFIX)/ops: $(go_deps)
 	cd cli && go build -ldflags '-s -w' -o $(@)
 
+$(PREFIX)/ops-dev: $(go_deps)
+	cd cli && go build -race -ldflags '-s -w' -o $(@)
+
 .PHONY: ops
 ops: $(PREFIX)/ops
 
+.PHONY: dev
+dev: $(PREFIX)/ops-dev
+
 .PHONY: test
 test:
-	go test -v ./...
+	go test -v -race ./...
 
 .PHONY: mock
 mock:

internal/core/core.go

@@ -30,6 +30,7 @@ type Core struct {
 	discovery           discovery.Service
 	serverService       server.Service
 	log                 logger.Logger
+	eventSubscription   int
 	evtListeners        []*EventListener
 	serverPollListeners []*ServerPollListener
 	nextListenerId      int
@@ -64,6 +65,9 @@ func New(
 
 func (c *Core) Stop() error {
 	c.discovery.Stop()
+	if c.eventSubscription != 0 {
+		c.serverService.StopStream(c.eventSubscription)
+	}
 	c.cancel()
 	return c.ctx.Err()
 }

internal/core/core_test.go

@@ -1,8 +1,8 @@
 package core_test
 
 import (
+	"sync"
 	"testing"
-	"time"
 
 	"github.com/golang/mock/gomock"
 	"github.com/robgonnella/ops/internal/config"
@@ -159,7 +159,7 @@ func TestCore(t *testing.T) {
 	})
 
 	t.Run("registers and removes event listener", func(st *testing.T) {
-		evtChan := make(chan *event.Event, 1)
+		evtChan := make(chan *event.Event)
 		id := coreService.RegisterEventListener(evtChan)
 
 		assert.Equal(st, 1, id)
@@ -168,7 +168,7 @@ func TestCore(t *testing.T) {
 	})
 
 	t.Run("registers and removes server listener", func(st *testing.T) {
-		serverChan := make(chan []*server.Server, 1)
+		serverChan := make(chan []*server.Server)
 
 		id := coreService.RegisterServerPollListener(serverChan)
 
@@ -178,16 +178,39 @@ func TestCore(t *testing.T) {
 	})
 
 	t.Run("monitors network", func(st *testing.T) {
-		defer coreService.Stop()
+		wg := sync.WaitGroup{}
+		wg.Add(2)
 
-		mockServerService.EXPECT().StreamEvents(gomock.Any())
+		mockServerService.EXPECT().StreamEvents(gomock.Any()).Return(1)
 		mockServerService.EXPECT().GetAllServersInNetworkTargets(conf.Targets)
-		mockScanner.EXPECT().Scan()
+		mockScanner.EXPECT().Scan().DoAndReturn(func() ([]*discovery.DiscoveryResult, error) {
+			defer func() {
+				coreService.Stop()
+				wg.Done()
+			}()
+			return []*discovery.DiscoveryResult{
+				{
+					ID:       "id",
+					Hostname: "hostname",
+					IP:       "ip",
+					OS:       "os",
+					Status:   server.StatusOnline,
+					Ports: []discovery.Port{
+						{
+							ID:     22,
+							Status: discovery.PortOpen,
+						},
+					},
+				},
+			}, nil
+		})
 		mockScanner.EXPECT().Stop()
-		mockServerService.EXPECT().StopStream(gomock.Any()).AnyTimes()
+		mockServerService.EXPECT().StopStream(1).Do(func(int) {
+			wg.Done()
+		})
 
 		go coreService.Monitor()
 
-		time.Sleep(time.Millisecond * 10)
+		wg.Wait()
 	})
 }

internal/core/monitor.go

@@ -10,12 +10,10 @@ import (
 
 // Run runs the sequence driver for the HostInstallStage
 func (c *Core) Monitor() error {
-	evtReceiveChan := make(chan *event.Event, 100)
+	evtReceiveChan := make(chan *event.Event)
 
 	// create event subscription
-	subscription := c.serverService.StreamEvents(evtReceiveChan)
-
-	defer c.serverService.StopStream(subscription)
+	c.eventSubscription = c.serverService.StreamEvents(evtReceiveChan)
 
 	// Start network scanner
 	go c.discovery.MonitorNetwork()
@@ -24,12 +22,14 @@ func (c *Core) Monitor() error {
 	go c.pollForDatabaseUpdates()
 
 	for {
-		select {
-		case <-c.ctx.Done():
-			return c.ctx.Err()
-		case evt := <-evtReceiveChan:
-			c.handleServerEvent(evt)
+		evt, ok := <-evtReceiveChan
+		if !ok {
+			for _, listener := range c.evtListeners {
+				close(listener.channel)
+			}
+			return nil
 		}
+		c.handleServerEvent(evt)
 	}
 }
 
@@ -59,6 +59,9 @@ func (c *Core) pollForDatabaseUpdates() error {
 	for {
 		select {
 		case <-c.ctx.Done():
+			for _, listener := range c.serverPollListeners {
+				close(listener.channel)
+			}
 			return c.ctx.Err()
 		default:
 			if errCount >= 5 {

internal/server/service.go

@@ -1,9 +1,9 @@
 package server
 
 import (
-	"context"
 	"errors"
 	"net"
+	"sync"
 
 	"github.com/imdario/mergo"
 	"github.com/robgonnella/ops/internal/config"
@@ -37,23 +37,21 @@ func filterChannels(channels []*eventChannel, fn func(c *eventChannel) bool) []*
 
 // ServerService represents our server service implementation
 type ServerService struct {
-	ctx      context.Context
 	log      logger.Logger
 	repo     Repo
 	evtChans []*eventChannel
+	mux      sync.Mutex
 }
 
 // NewService returns a new instance server service
 func NewService(conf config.Config, repo Repo) *ServerService {
 	log := logger.New()
 
-	ctx := context.Background()
-
 	return &ServerService{
-		ctx:      ctx,
 		log:      log,
 		repo:     repo,
 		evtChans: []*eventChannel{},
+		mux:      sync.Mutex{},
 	}
 }
 
@@ -175,14 +173,22 @@ func (s *ServerService) StreamEvents(send chan *event.Event) int {
 		send: send,
 	}
 
+	s.mux.Lock()
 	s.evtChans = append(s.evtChans, evtChan)
+	s.mux.Unlock()
 
 	return evtChan.id
 }
 
 func (s *ServerService) StopStream(id int) {
+	s.mux.Lock()
+	defer s.mux.Unlock()
+
 	s.log.Info().Int("channelID", id).Msg("Filtering channel")
 	s.evtChans = filterChannels(s.evtChans, func(c *eventChannel) bool {
+		if c.id == id {
+			close(c.send)
+		}
 		return c.id != id
 	})
 }
@@ -193,6 +199,8 @@ func (s *ServerService) GetServer(id string) (*Server, error) {
 }
 
 func (s *ServerService) sendServerUpdateEvent(server *Server) {
+	s.mux.Lock()
+	defer s.mux.Unlock()
 	for _, clientChan := range s.evtChans {
 		clientChan.send <- &event.Event{
 			Type:    event.SeverUpdate,

internal/server/service_test.go

@@ -127,7 +127,7 @@ func TestServerService(t *testing.T) {
 	})
 
 	t.Run("streams events", func(st *testing.T) {
-		evtChan := make(chan *event.Event, 1)
+		evtChan := make(chan *event.Event)
 
 		streamID := service.StreamEvents(evtChan)
 

internal/ui/component/context.go

@@ -7,6 +7,7 @@ import (
 	"github.com/gdamore/tcell/v2"
 	"github.com/rivo/tview"
 	"github.com/robgonnella/ops/internal/config"
+	"github.com/robgonnella/ops/internal/logger"
 	"github.com/robgonnella/ops/internal/ui/key"
 	"github.com/robgonnella/ops/internal/ui/style"
 )
@@ -21,6 +22,8 @@ func NewConfigContext(
 	onSelect func(id int),
 	onDelete func(name string, id int),
 ) *ConfigContext {
+	log := logger.New()
+
 	colHeaders := []string{"ID", "Name", "Target", "SSH-User", "SSH-Identity", "Overrides"}
 	table := createTable("Context", colHeaders)
 
@@ -31,7 +34,12 @@ func NewConfigContext(
 			idStr := table.GetCell(row, 0).Text
 			name := table.GetCell(row, 1).Text
 
-			id, _ := strconv.Atoi(idStr)
+			id, err := strconv.Atoi(idStr)
+
+			if err != nil {
+				log.Error().Err(err).Msg("failed to delete context")
+				return nil
+			}
 
 			onDelete(name, id)
 
@@ -41,7 +49,11 @@ func NewConfigContext(
 		if evt.Key() == key.KeyEnter {
 			row, _ := table.GetSelection()
 			idStr := table.GetCell(row, 0).Text
-			id, _ := strconv.Atoi(idStr)
+			id, err := strconv.Atoi(idStr)
+			if err != nil {
+				log.Error().Err(err).Msg("failed to select new context")
+				return nil
+			}
 			onSelect(id)
 			return nil
 		}

internal/ui/component/event.go

@@ -1,7 +1,6 @@
 package component
 
 import (
-	"context"
 	"strconv"
 
 	"github.com/rivo/tview"
@@ -11,8 +10,6 @@ import (
 )
 
 type EventTable struct {
-	ctx           context.Context
-	cancel        context.CancelFunc
 	table         *tview.Table
 	columnHeaders []string
 	count         uint
@@ -31,11 +28,7 @@ func NewEventTable() *EventTable {
 		"STATUS",
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
-
 	return &EventTable{
-		ctx:           ctx,
-		cancel:        cancel,
 		table:         createTable("events", columnHeaders),
 		columnHeaders: columnHeaders,
 		count:         0,

internal/ui/component/server.go

@@ -1,8 +1,6 @@
 package component
 
 import (
-	"context"
-
 	"github.com/gdamore/tcell/v2"
 	"github.com/rivo/tview"
 	"github.com/robgonnella/ops/internal/server"
@@ -11,17 +9,13 @@ import (
 )
 
 type ServerTable struct {
-	ctx           context.Context
-	cancel        context.CancelFunc
 	table         *tview.Table
 	columnHeaders []string
 }
 
 func NewServerTable(OnSSH func(ip string)) *ServerTable {
 	columnHeaders := []string{"HOSTNAME", "IP", "ID", "OS", "SSH", "STATUS"}
 
-	ctx, cancel := context.WithCancel(context.Background())
-
 	table := createTable("servers", columnHeaders)
 
 	table.SetInputCapture(func(evt *tcell.EventKey) *tcell.EventKey {
@@ -36,8 +30,6 @@ func NewServerTable(OnSSH func(ip string)) *ServerTable {
 	})
 
 	return &ServerTable{
-		ctx:           ctx,
-		cancel:        cancel,
 		table:         table,
 		columnHeaders: columnHeaders,
 	}

internal/ui/launch.go

@@ -62,7 +62,13 @@ func (u *UI) Launch() error {
 		log.Fatal().Err(err).Msg("failed to create app core")
 	}
 
-	u.view = newView(*userIP, appCore)
+	allConfigs, err := appCore.GetConfigs()
+
+	if err != nil {
+		log.Fatal().Err(err).Msg("failed to retrieve configs")
+	}
+
+	u.view = newView(*userIP, allConfigs, appCore)
 
 	os.Stdout, _ = os.Open(os.DevNull)
 	os.Stderr, _ = os.Open(os.DevNull)