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()
 }
@@ -146,6 +150,9 @@ func (c *Core) RemoveEventListener(id int) {
 
  listeners := []*EventListener{}
  for _, listener := range c.evtListeners {
+ if listener.id == id {
+ close(listener.channel)
+ }
  if listener.id != id {
  listeners = append(listeners, listener)
  }
@@ -175,6 +182,9 @@ func (c *Core) RemoveServerPollListener(id int) {
 
  listeners := []*ServerPollListener{}
  for _, listener := range c.serverPollListeners {
+ if listener.id == id {
+ close(listener.channel)
+ }
  if listener.id != id {
  listeners = append(listeners, listener)
  }

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,16 @@ 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 {
+ c.mux.Lock()
+ for _, listener := range c.evtListeners {
+ close(listener.channel)
+ }
+ c.mux.Unlock()
+ return nil
  }
+ c.handleServerEvent(evt)
  }
 }
 
@@ -47,6 +49,9 @@ func (c *Core) handleServerEvent(evt *event.Event) {
 
  c.log.Info().Fields(fields).Msg("Event Received")
 
+ c.mux.Lock()
+ defer c.mux.Unlock()
+
  for _, listener := range c.evtListeners {
  listener.channel <- evt
  }
@@ -59,6 +64,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 {
@@ -77,9 +85,11 @@ func (c *Core) pollForDatabaseUpdates() error {
 
  errCount = 0
 
+ c.mux.Lock()
  for _, listener := range c.serverPollListeners {
  listener.channel <- response
  }
+ c.mux.Unlock()
 
  time.Sleep(pollTime)
  }

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,
  }