Merge pull request #1060 from openziti/fix-races

Fix various races detected by the race detector

controller/sync_strats/sync_instant.go

@@ -19,7 +19,6 @@ package sync_strats
 import (
 	"encoding/json"
 	"fmt"
-	"google.golang.org/protobuf/proto"
 	"github.com/lucsky/cuid"
 	"github.com/michaelquigley/pfxlog"
 	"github.com/openziti/channel"
@@ -30,9 +29,11 @@ import (
 	"github.com/openziti/edge/pb/edge_ctrl_pb"
 	"github.com/openziti/fabric/build"
 	"github.com/openziti/fabric/controller/network"
-	"github.com/openziti/storage/ast"
+	"github.com/openziti/foundation/util/concurrenz"
 	"github.com/openziti/foundation/util/debugz"
+	"github.com/openziti/storage/ast"
 	"go.etcd.io/bbolt"
+	"google.golang.org/protobuf/proto"
 	"strings"
 	"sync"
 	"time"
@@ -87,7 +88,8 @@ type InstantStrategy struct {
 	routerConnectedQueue     chan *RouterSender
 	receivedClientHelloQueue chan *RouterSender
 
-	stop chan struct{}
+	stopNotify chan struct{}
+	stopped    concurrenz.AtomicBoolean
 }
 
 func NewInstantStrategy(ae *env.AppEnv, options InstantStrategyOptions) *InstantStrategy {
@@ -124,7 +126,7 @@ func NewInstantStrategy(ae *env.AppEnv, options InstantStrategyOptions) *Instant
 		routerConnectedQueue:     make(chan *RouterSender, options.MaxQueuedRouterConnects),
 		receivedClientHelloQueue: make(chan *RouterSender, options.MaxQueuedClientHellos),
 
-		stop: make(chan struct{}, 0),
+		stopNotify: make(chan struct{}, 0),
 	}
 
 	strategy.helloHandler = handler_edge_ctrl.NewHelloHandler(ae, strategy.ReceiveClientHello)
@@ -150,9 +152,8 @@ func (strategy *InstantStrategy) Type() env.RouterSyncStrategyType {
 }
 
 func (strategy *InstantStrategy) Stop() {
-	if strategy.stop != nil {
-		close(strategy.stop)
-		strategy.stop = nil
+	if strategy.stopped.CompareAndSwap(false, true) {
+		close(strategy.stopNotify)
 	}
 }
 
@@ -267,7 +268,7 @@ func (strategy *InstantStrategy) SessionDeleted(session *persistence.Session) {
 func (strategy *InstantStrategy) startHandleRouterConnectWorker() {
 	for {
 		select {
-		case <-strategy.stop:
+		case <-strategy.stopNotify:
 			return
 		case rtx := <-strategy.routerConnectedQueue:
 			func() {
@@ -287,7 +288,7 @@ func (strategy *InstantStrategy) startHandleRouterConnectWorker() {
 func (strategy *InstantStrategy) startSynchronizeWorker() {
 	for {
 		select {
-		case <-strategy.stop:
+		case <-strategy.stopNotify:
 			return
 		case rtx := <-strategy.receivedClientHelloQueue:
 			func() {

router/handler_edge_ctrl/apiSessionAdded.go

@@ -75,8 +75,11 @@ func (h *apiSessionAddedHandler) HandleReceive(msg *channel.Message, _ channel.C
 	go func() {
 		req := &edge_ctrl_pb.ApiSessionAdded{}
 		if err := proto.Unmarshal(msg.Body, req); err == nil {
-			if req.IsFullState {
+			for _, session := range req.ApiSessions {
+				h.sm.AddApiSession(session)
+			}
 
+			if req.IsFullState {
 				reqWithState := &apiSessionAddedWithState{
 					ApiSessionAdded: req,
 				}
@@ -98,10 +101,6 @@ func (h *apiSessionAddedHandler) HandleReceive(msg *channel.Message, _ channel.C
 				}
 				h.reqChan <- reqWithState
 			}
-
-			for _, session := range req.ApiSessions {
-				h.sm.AddApiSession(session)
-			}
 		} else {
 			pfxlog.Logger().Panic("could not convert message as api session added")
 		}

router/xgress_edge/certchecker_test.go

@@ -254,11 +254,11 @@ func Test_CertExpirationChecker(t *testing.T) {
 			certChecker := newCertChecker()
 			certChecker.timeoutDuration = 10 * time.Millisecond
 
-			invoked := false
+			var invoked concurrenz.AtomicBoolean
 
 			extender := &stubExtender{
 				done: func() error {
-					invoked = true
+					invoked.Set(true)
 					certChecker.id.Cert().Leaf.NotAfter = time.Now().AddDate(1, 0, 0)
 					certChecker.id.ServerCert().Leaf.NotAfter = time.Now().AddDate(1, 0, 0)
 					return errors.New("test")
@@ -275,7 +275,7 @@ func Test_CertExpirationChecker(t *testing.T) {
 
 			time.Sleep(200 * time.Millisecond)
 
-			req.True(invoked)
+			req.True(invoked.Get())
 
 			certChecker.closeNotify <- struct{}{}
 		})

tunnel/dns/server.go

@@ -84,6 +84,17 @@ func NewDnsServer(addr string) Resolver {
 		Net:  "udp",
 	}
 
+	names := make(map[string]net.IP)
+	r := &resolver{
+		server:     s,
+		names:      names,
+		ips:        make(map[string]string),
+		namesMtx:   sync.Mutex{},
+		domains:    make(map[string]*domainEntry),
+		domainsMtx: sync.Mutex{},
+	}
+	s.Handler = r
+
 	errChan := make(chan error)
 	go func() {
 		errChan <- s.ListenAndServe()
@@ -100,17 +111,6 @@ func NewDnsServer(addr string) Resolver {
 		log.Infof("dns server running at %s", s.Addr)
 	}
 
-	names := make(map[string]net.IP)
-	r := &resolver{
-		server:     s,
-		names:      names,
-		ips:        make(map[string]string),
-		namesMtx:   sync.Mutex{},
-		domains:    make(map[string]*domainEntry),
-		domainsMtx: sync.Mutex{},
-	}
-	s.Handler = r
-
 	const resolverConfigHelp = "ziti-tunnel runs an internal DNS server which must be first in the host's\n" +
 		"resolver configuration. On systems that use NetManager/dhclient, this can\n" +
 		"be achieved by adding the following to /etc/dhcp/dhclient.conf:\n" +
@@ -150,13 +150,22 @@ func (r *resolver) testSystemResolver() error {
 	return nil
 }
 
-func (r *resolver) getAddress(name string) (net.IP, error) {
+func (r *resolver) getHostnameIp(name string) (net.IP, bool) {
+	r.namesMtx.Lock()
+	defer r.namesMtx.Unlock()
 	canonical := strings.ToLower(name)
-	a, ok := r.names[canonical]
+	ip, found := r.names[canonical]
+	return ip, found
+}
+
+func (r *resolver) getAddress(name string) (net.IP, error) {
+	a, ok := r.getHostnameIp(name)
 	if ok {
 		return a, nil
 	}
 
+	canonical := strings.ToLower(name)
+
 	r.domainsMtx.Lock()
 	defer r.domainsMtx.Unlock()
 	for {
@@ -174,7 +183,7 @@ func (r *resolver) getAddress(name string) (net.IP, error) {
 				return nil, err
 			}
 			log.Debugf("assigned %v => %v", name, ip)
-			r.AddHostname(name[:len(name)-1], ip)
+			_ = r.AddHostname(name[:len(name)-1], ip) // this resolver impl never returns an error
 			return ip, err
 		}
 	}