diff --git a/client/client.go b/client/client.go index ff88e338..75035c72 100644 --- a/client/client.go +++ b/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. diff --git a/client/client_test.go b/client/client_test.go index eb77d215..4ebee829 100644 --- a/client/client_test.go +++ b/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)