bridge: fix shutdown of not running bridge

Author: jkralik

bridge/device/cloud/manager.go

@@ -25,6 +25,7 @@ import (
 	"fmt"
 	"reflect"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/google/uuid"
@@ -96,6 +97,7 @@ type Manager struct {
 	signedIn           bool
 	resourcesPublished bool
 	done               chan struct{}
+	stopped            atomic.Bool
 	trigger            chan bool
 	loop               *eventloop.Loop
 }
@@ -529,7 +531,10 @@ func (c *Manager) connect(ctx context.Context) error {
 }
 
 func (c *Manager) Close() {
-	c.done <- struct{}{}
+	if !c.stopped.CompareAndSwap(false, true) {
+		return
+	}
+	close(c.done)
 }
 
 func (c *Manager) Unregister() {

bridge/device/device.go

@@ -23,6 +23,7 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"sync/atomic"
 
 	"github.com/google/uuid"
 	"github.com/plgd-dev/device/v2/bridge/device/cloud"
@@ -69,6 +70,7 @@ type Device struct {
 	loop              *eventloop.Loop
 	runLoop           bool
 	done              chan struct{}
+	stopped           atomic.Bool
 }
 
 func NewLogger(id uuid.UUID, level pkgLog.Level) pkgLog.Logger {
@@ -296,6 +298,9 @@ func (d *Device) HandleRequest(req *net.Request) (*pool.Message, error) {
 }
 
 func (d *Device) Close() {
+	if !d.stopped.CompareAndSwap(false, true) {
+		return
+	}
 	if d.cloudManager != nil {
 		d.cloudManager.Close()
 	}
@@ -306,9 +311,6 @@ func (d *Device) Close() {
 		resource.Close()
 	}
 	if d.runLoop {
-		select {
-		case d.done <- struct{}{}:
-		default:
-		}
+		close(d.done)
 	}
 }

bridge/net/network.go

@@ -54,6 +54,8 @@ type Net struct {
 
 	servers coAPServers
 	serving atomic.Bool
+	stopped atomic.Bool
+	wg      sync.WaitGroup
 	done    chan struct{}
 	cache   *coapCache.Cache[int32, bool]
 }
@@ -357,7 +359,9 @@ func New(cfg Config, handler RequestHandler, logger log.Logger) (*Net, error) {
 		cache:   coapCache.NewCache[int32, bool](),
 	}
 	m.DefaultHandle(mux.HandlerFunc(n.ServeCOAP))
+	n.wg.Add(1)
 	go func() {
+		defer n.wg.Done()
 		for {
 			select {
 			case <-n.done:
@@ -417,10 +421,12 @@ func (n *Net) GetEndpoints(cm *net.ControlMessage, localAddr string) schema.Endp
 }
 
 func (n *Net) Serve() error {
+	if n.stopped.Load() {
+		return fmt.Errorf("already stopped")
+	}
 	if !n.serving.CompareAndSwap(false, true) {
 		return fmt.Errorf("already serving")
 	}
-	defer close(n.done)
 	var wg sync.WaitGroup
 	errCh := make(chan error, len(n.servers))
 	wg.Add(len(n.servers))
@@ -446,10 +452,14 @@ func (n *Net) Serve() error {
 }
 
 func (n *Net) Close() error {
+	if !n.stopped.CompareAndSwap(false, true) {
+		return nil
+	}
+	close(n.done)
+	n.wg.Wait()
 	if !n.serving.Load() {
 		return n.servers.Close()
 	}
 	n.servers.Stop()
-	<-n.done
 	return nil
 }

bridge/service/service.go

@@ -20,6 +20,7 @@ package service
 
 import (
 	"fmt"
+	"sync"
 
 	"github.com/google/uuid"
 	"github.com/plgd-dev/device/v2/bridge/device"
@@ -66,7 +67,7 @@ type Service struct {
 	cfg                Config
 	net                *net.Net
 	devices            *coapSync.Map[uuid.UUID, Device]
-	done               chan struct{}
+	wg                 sync.WaitGroup
 	onDiscoveryDevices func(req *net.Request)
 }
 
@@ -143,7 +144,6 @@ func New(cfg Config, opts ...Option) (*Service, error) {
 		cfg:                cfg,
 		devices:            coapSync.NewMap[uuid.UUID, Device](),
 		onDiscoveryDevices: o.onDiscoveryDevices,
-		done:               make(chan struct{}),
 	}
 	n, err := net.New(cfg.API.CoAP.Config, c.DefaultRequestHandler, o.logger)
 	if err != nil {
@@ -155,7 +155,8 @@ func New(cfg Config, opts ...Option) (*Service, error) {
 }
 
 func (c *Service) Serve() error {
-	defer close(c.done)
+	c.wg.Add(1)
+	defer c.wg.Done()
 	return c.net.Serve()
 }
 
@@ -164,7 +165,7 @@ func (c *Service) Shutdown() error {
 	if err != nil {
 		return err
 	}
-	<-c.done
+	c.wg.Wait()
 	devices := c.devices.LoadAndDeleteAll()
 	for _, d := range devices {
 		d.Close()