client: fix send on closed channel panic during disconnect

client/client.go

@@ -107,6 +107,11 @@ type ovsdbClient struct {
 	shutdown      bool
 	shutdownMutex sync.Mutex
 
+	// disconnected is set to true when the client is disconnecting,
+	// before trafficSeen channel is closed, to prevent send on closed channel panic
+	disconnected      bool
+	disconnectedMutex sync.Mutex
+
 	handlerShutdown *sync.WaitGroup
 
 	trafficSeen chan struct{}
@@ -307,7 +312,6 @@ func (o *ovsdbClient) connect(ctx context.Context, reconnect bool) error {
 		}
 	}
 
-	go o.handleDisconnectNotification()
 	if o.options.inactivityTimeout > 0 {
 		o.handlerShutdown.Add(1)
 		go o.handleInactivityProbes()
@@ -324,6 +328,10 @@ func (o *ovsdbClient) connect(ctx context.Context, reconnect bool) error {
 		}(db)
 	}
 
+	// handleDisconnectNotification must be started after all Add() calls
+	// to handlerShutdown, since it will call Wait() on reconnection
+	go o.handleDisconnectNotification()
+
 	o.connected = true
 	return nil
 }
@@ -431,6 +439,8 @@ func (o *ovsdbClient) tryEndpoint(ctx context.Context, u *url.URL) (string, erro
 // Should only be called when the mutex is held
 func (o *ovsdbClient) createRPC2Client(conn net.Conn) {
 	o.stopCh = make(chan struct{})
+	// Reset disconnected flag before creating new trafficSeen channel
+	o.setDisconnected(false)
 	if o.options.inactivityTimeout > 0 {
 		o.trafficSeen = make(chan struct{})
 	}
@@ -576,7 +586,7 @@ func (o *ovsdbClient) UpdateEndpoints(endpoints []string) {
 func (o *ovsdbClient) SetOption(opt Option) error {
 	o.rpcMutex.RLock()
 	defer o.rpcMutex.RUnlock()
-	if o.rpcClient != nil {
+	if o.connected {
 		return fmt.Errorf("cannot set option when client is connected")
 	}
 	return opt(o.options)
@@ -858,7 +868,7 @@ func (o *ovsdbClient) transact(ctx context.Context, dbName string, skipChWrite b
 		return nil, err
 	}
 
-	if !skipChWrite && o.trafficSeen != nil {
+	if !skipChWrite && o.trafficSeen != nil && !o.isDisconnected() {
 		select {
 		case o.trafficSeen <- struct{}{}:
 		default:
@@ -1251,9 +1261,11 @@ func (o *ovsdbClient) handleDisconnectNotification() {
 	<-o.rpcClient.DisconnectNotify()
 	// close the stopCh, which will stop the cache event processor
 	close(o.stopCh)
-	if o.trafficSeen != nil {
-		close(o.trafficSeen)
-	}
+	// Set disconnected to prevent further sends to trafficSeen channel.
+	// We don't close trafficSeen because handleInactivityProbes() will
+	// exit via stopCh (already closed above), and closing trafficSeen
+	// would race with concurrent sends in transact().
+	o.setDisconnected(true)
 	o.metrics.numDisconnects.Inc()
 	// wait for client related handlers to shutdown
 	o.handlerShutdown.Wait()
@@ -1365,6 +1377,21 @@ func (o *ovsdbClient) Close() {
 	o.rpcClient.Close()
 }
 
+// isDisconnected returns true if the client is in the process of disconnecting.
+// This is used to prevent sending on the trafficSeen channel during disconnection.
+func (o *ovsdbClient) isDisconnected() bool {
+	o.disconnectedMutex.Lock()
+	defer o.disconnectedMutex.Unlock()
+	return o.disconnected
+}
+
+// setDisconnected sets the disconnected state of the client.
+func (o *ovsdbClient) setDisconnected(val bool) {
+	o.disconnectedMutex.Lock()
+	defer o.disconnectedMutex.Unlock()
+	o.disconnected = val
+}
+
 // Ensures the cache is consistent by evaluating that the client is connected
 // and the monitor is fully setup, with the cache populated. Caller must hold
 // the database's cache mutex for reading.

client/client_test.go

@@ -5,12 +5,15 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"log"
 	"math/rand"
+	"net"
 	"os"
 	"reflect"
 	"sort"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -947,6 +950,7 @@ func TestSetOption(t *testing.T) {
 	require.NoError(t, err)
 
 	o.rpcClient = &rpc2.Client{}
+	o.connected = true
 
 	err = o.SetOption(WithEndpoint("tcp::6641"))
 	assert.EqualError(t, err, "cannot set option when client is connected")
@@ -2064,6 +2068,206 @@ func TestGetSelectResultsByIndex(t *testing.T) {
 	})
 }
 
+// tcpProxy creates a TCP proxy that can be used to simulate network failures.
+// It forwards connections between a client and server, and can be closed
+// abruptly to simulate network disconnection.
+type tcpProxy struct {
+	listener net.Listener
+	target   string
+	conns    []net.Conn
+	connsMu  sync.Mutex
+}
+
+func newTCPProxy(t *testing.T, targetUnixSocket string) *tcpProxy {
+	// Create a TCP listener on a random port
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	proxy := &tcpProxy{
+		listener: listener,
+		target:   targetUnixSocket,
+	}
+
+	go proxy.acceptLoop()
+	return proxy
+}
+
+func (p *tcpProxy) acceptLoop() {
+	for {
+		clientConn, err := p.listener.Accept()
+		if err != nil {
+			return // Listener closed
+		}
+
+		// Connect to the target unix socket
+		serverConn, err := net.Dial("unix", p.target)
+		if err != nil {
+			clientConn.Close()
+			continue
+		}
+
+		p.connsMu.Lock()
+		p.conns = append(p.conns, clientConn, serverConn)
+		p.connsMu.Unlock()
+
+		// Bidirectional copy (errors ignored; connections will be closed on proxy shutdown)
+		go func() { _, _ = io.Copy(serverConn, clientConn) }()
+		go func() { _, _ = io.Copy(clientConn, serverConn) }()
+	}
+}
+
+func (p *tcpProxy) Addr() string {
+	return p.listener.Addr().String()
+}
+
+func (p *tcpProxy) CloseAllConnections() {
+	p.connsMu.Lock()
+	defer p.connsMu.Unlock()
+	for _, conn := range p.conns {
+		conn.Close()
+	}
+	p.conns = nil
+}
+
+func (p *tcpProxy) Close() {
+	p.listener.Close()
+	p.CloseAllConnections()
+}
+
+// TestTransactDuringDisconnectNoPanic verifies that the actual transact() code
+// doesn't panic when abrupt network disconnects occur.
+//
+// This test uses a TCP proxy to simulate real network failures by abruptly
+// closing connections while operations are in flight. This is more realistic
+// than calling Disconnect() which performs a graceful shutdown.
+//
+// The race condition occurs when:
+// 1. transact() completes an RPC call successfully
+// 2. transact() is about to send to trafficSeen
+// 3. Network failure triggers handleDisconnectNotification() which sets disconnected flag
+// 4. Without the fix, transact() would send to trafficSeen after it becomes invalid
+//
+// The fix prevents this by:
+// - Setting a disconnected flag before stopping handlers
+// - Checking this flag in transact() before sending to trafficSeen
+// - Not closing trafficSeen in handleDisconnectNotification to avoid races
+//
+// Note: This race has an extremely small timing window (nanoseconds). The test
+// may need multiple runs to trigger it reliably. Run with: go test -count=10
+// Without the fix, the test typically fails within a few runs with:
+// "panic: send on closed channel" at client.go in the transact() function.
+//
+// TODO: This test is currently skipped in short mode (e.g., CI) because it exposes
+// multiple pre-existing race conditions in this library and in github.com/cenkalti/rpc2
+// when run with -race enabled. These races should be fixed in separate commits, then
+// this skip can be removed.
+//
+// To run this test explicitly: go test -v -run TestTransactDuringDisconnectNoPanic ./client
+func TestTransactDuringDisconnectNoPanic(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping test in short mode due to unrelated race conditions (see TODO above)")
+	}
+
+	var defSchema ovsdb.DatabaseSchema
+	err := json.Unmarshal([]byte(schema), &defSchema)
+	require.NoError(t, err)
+
+	// Run multiple iterations to increase chance of hitting the race
+	for iter := 0; iter < 100; iter++ {
+		t.Run(fmt.Sprintf("iteration-%d", iter), func(t *testing.T) {
+			server, sock := newOVSDBServer(t, defDB, defSchema)
+			defer server.Close()
+
+			// Create a TCP proxy to the unix socket so we can abruptly close connections
+			proxy := newTCPProxy(t, sock)
+			defer proxy.Close()
+
+			endpoint := fmt.Sprintf("tcp:%s", proxy.Addr())
+			client, err := newOVSDBClient(defDB,
+				WithEndpoint(endpoint),
+				// Enable inactivity check so trafficSeen channel is created
+				WithInactivityCheck(50*time.Millisecond, 25*time.Millisecond, backoff.NewConstantBackOff(time.Millisecond)),
+			)
+			require.NoError(t, err)
+
+			err = client.Connect(context.Background())
+			require.NoError(t, err)
+			defer client.Close()
+
+			var wg sync.WaitGroup
+			stop := make(chan struct{})
+
+			// Launch goroutines that continuously call Transact
+			for i := 0; i < 50; i++ {
+				wg.Add(1)
+				go func(id int) {
+					defer wg.Done()
+					j := 0
+					for {
+						select {
+						case <-stop:
+							return
+						default:
+							bridge := &Bridge{
+								UUID:        fmt.Sprintf("test-bridge-%d-%d-%d", iter, id, j),
+								Name:        fmt.Sprintf("br-test-%d-%d-%d", iter, id, j),
+								ExternalIDs: map[string]string{"test": "disconnect-race"},
+							}
+							ops, err := client.Create(bridge)
+							if err == nil {
+								ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
+								_, _ = client.Transact(ctx, ops...)
+								cancel()
+							}
+							j++
+						}
+					}
+				}(i)
+			}
+
+			// Goroutines that abruptly close connections via the proxy
+			// This simulates network failures while operations are in flight
+			for i := 0; i < 20; i++ {
+				wg.Add(1)
+				go func() {
+					defer wg.Done()
+					for j := 0; j < 50; j++ {
+						select {
+						case <-stop:
+							return
+						default:
+							// Random delay to vary timing
+							time.Sleep(time.Duration(rand.Intn(500)) * time.Microsecond)
+							// Abruptly close all connections - this triggers handleDisconnectNotification
+							proxy.CloseAllConnections()
+							// Small delay then let client reconnect
+							time.Sleep(time.Duration(rand.Intn(1000)) * time.Microsecond)
+							// Client will auto-reconnect on next Transact attempt
+						}
+					}
+				}()
+			}
+
+			// Let test run for a while
+			time.Sleep(200 * time.Millisecond)
+			close(stop)
+
+			// Wait for goroutines with timeout
+			done := make(chan struct{})
+			go func() {
+				wg.Wait()
+				close(done)
+			}()
+
+			select {
+			case <-done:
+			case <-time.After(10 * time.Second):
+				t.Log("timeout waiting for goroutines")
+			}
+		})
+	}
+}
+
 func TestConditionalWhereListWaitsForCacheConsistency(t *testing.T) {
 	ovs, err := newOVSDBClient(defDB)
 	require.NoError(t, err)