diff --git a/cache/cache.go b/cache/cache.go index 4840aa24..bac4c1a6 100644 --- a/cache/cache.go +++ b/cache/cache.go @@ -878,11 +878,15 @@ func NewTableCache(dbModel model.DatabaseModel, data Data, logger *logr.Logger) // Mapper returns the mapper func (t *TableCache) Mapper() mapper.Mapper { + t.mutex.RLock() + defer t.mutex.RUnlock() return t.dbModel.Mapper } // DatabaseModel returns the DatabaseModelRequest func (t *TableCache) DatabaseModel() model.DatabaseModel { + t.mutex.RLock() + defer t.mutex.RUnlock() return t.dbModel } diff --git a/client/client.go b/client/client.go index 9d05e773..965f30a0 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, + // to prevent further sends to trafficSeen channel + 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 } @@ -426,6 +434,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{}) } @@ -846,7 +856,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: @@ -1238,9 +1248,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() @@ -1352,6 +1364,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 disconnect. +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 c20bc8e0..68c05045 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" @@ -2063,3 +2066,188 @@ func TestGetSelectResultsByIndex(t *testing.T) { assert.Contains(t, err.Error(), "index 5 is out of range") }) } + +// 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 + go io.Copy(serverConn, clientConn) + go 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 closes trafficSeen +// 4. transact() sends to closed trafficSeen -> panic (without the fix) +// +// 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. +func TestTransactDuringDisconnectNoPanic(t *testing.T) { + 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") + } + }) + } +} +