From bf6cfe1370c2181dd764bf5c4fb1baae6f6b54f8 Mon Sep 17 00:00:00 2001 From: Ihar Hrachyshka Date: Thu, 5 Feb 2026 18:00:44 -0500 Subject: [PATCH 1/4] client: fix send on closed channel panic during disconnect When handleDisconnectNotification() closes the trafficSeen channel, there's a race window where transact() may attempt to send on the already-closed channel, causing a panic: goroutine 1 (transact): goroutine 2 (handleDisconnectNotification): if trafficSeen != nil { close(trafficSeen) trafficSeen <- struct{}{} // PANIC: send on closed channel } This was observed in CI as a coredump during e2e tests when the services controller was syncing a service while the OVSDB connection was being closed. Fix this by adding a disconnected flag (protected by a mutex) that is set before closing the trafficSeen channel. The transact() function checks this flag before attempting to send on the channel. This follows the same pattern as the isShutdown() mechanism proposed in: https://github.com/ovn-kubernetes/libovsdb/pull/457 A test using a TCP proxy to simulate abrupt network disconnections is included to verify the fix. Signed-off-by: Ihar Hrachyshka Assisted-by: opus (claude-opus-4-5-20251101) --- client/client.go | 27 +++++- client/client_test.go | 188 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 214 insertions(+), 1 deletion(-) diff --git a/client/client.go b/client/client.go index 9d05e773..3b53b127 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{} @@ -426,6 +431,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 +853,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,6 +1245,9 @@ func (o *ovsdbClient) handleDisconnectNotification() { <-o.rpcClient.DisconnectNotify() // close the stopCh, which will stop the cache event processor close(o.stopCh) + // Set disconnected before closing trafficSeen to prevent + // send on closed channel panic in transact() + o.setDisconnected(true) if o.trafficSeen != nil { close(o.trafficSeen) } @@ -1352,6 +1362,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 after it's been closed. +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") + } + }) + } +} + From d09d51b05e5f74e123ca009b728db91ada116a07 Mon Sep 17 00:00:00 2001 From: Ihar Hrachyshka Date: Thu, 5 Feb 2026 19:21:16 -0500 Subject: [PATCH 2/4] client: don't close trafficSeen channel on disconnect Instead of closing trafficSeen channel during disconnect, rely on the disconnected flag to stop sends and let handleInactivityProbes() exit via stopCh (which is closed first). Closing the channel races with concurrent sends in transact(), even with the disconnected flag check, because the check and send are not atomic: WARNING: DATA RACE Write at 0x00c0000a6160 by goroutine 7312: runtime.recvDirect() client.(*ovsdbClient).handleDisconnectNotification() client/client.go:1255 <-- close(o.trafficSeen) Previous read at 0x00c0000a6160 by goroutine 7315: runtime.chansend1() client.(*ovsdbClient).transact() client/client.go:861 <-- o.trafficSeen <- struct{}{} The channel doesn't need to be explicitly closed: - handleInactivityProbes() watches stopCh which is closed first - The channel will be garbage collected when no longer referenced - A new channel is created on reconnection anyway Signed-off-by: Ihar Hrachyshka Assisted-by: opus (claude-opus-4-5-20251101) --- client/client.go | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/client/client.go b/client/client.go index 3b53b127..5fdade39 100644 --- a/client/client.go +++ b/client/client.go @@ -108,7 +108,7 @@ type ovsdbClient struct { 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 + // to prevent further sends to trafficSeen channel disconnected bool disconnectedMutex sync.Mutex @@ -1245,12 +1245,11 @@ func (o *ovsdbClient) handleDisconnectNotification() { <-o.rpcClient.DisconnectNotify() // close the stopCh, which will stop the cache event processor close(o.stopCh) - // Set disconnected before closing trafficSeen to prevent - // send on closed channel panic in transact() + // 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) - if o.trafficSeen != nil { - close(o.trafficSeen) - } o.metrics.numDisconnects.Inc() // wait for client related handlers to shutdown o.handlerShutdown.Wait() @@ -1363,7 +1362,7 @@ func (o *ovsdbClient) Close() { } // isDisconnected returns true if the client is in the process of disconnecting. -// This is used to prevent sending on the trafficSeen channel after it's been closed. +// This is used to prevent sending on the trafficSeen channel during disconnect. func (o *ovsdbClient) isDisconnected() bool { o.disconnectedMutex.Lock() defer o.disconnectedMutex.Unlock() From 3f9d9f6017ea9c8658f4aee945c71c20c2b66783 Mon Sep 17 00:00:00 2001 From: Ihar Hrachyshka Date: Thu, 5 Feb 2026 19:21:39 -0500 Subject: [PATCH 3/4] client: start handleDisconnectNotification after all WaitGroup Add() calls Move the goroutine spawn for handleDisconnectNotification() to after all handlerShutdown.Add() calls complete. This fixes a race between Add() and Wait() on the same WaitGroup during reconnection. The race occurs because handleDisconnectNotification() calls Wait() on handlerShutdown, but it was spawned before all Add() calls: WARNING: DATA RACE Write at 0x00c00058e158 by goroutine 14049: client.(*ovsdbClient).handleDisconnectNotification() client/client.go:1256 <-- o.handlerShutdown.Wait() Previous read at 0x00c00058e158 by goroutine 13633: client.(*ovsdbClient).connect() client/client.go:317 <-- o.handlerShutdown.Add(1) Per sync.WaitGroup documentation: "calls with a positive delta that occur when the counter is zero must happen before a Wait." Signed-off-by: Ihar Hrachyshka Assisted-by: opus (claude-opus-4-5-20251101) --- client/client.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/client/client.go b/client/client.go index 5fdade39..965f30a0 100644 --- a/client/client.go +++ b/client/client.go @@ -312,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() @@ -329,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 } From 8cf016d0a86be35aabdfa04567dedb6c57030f4d Mon Sep 17 00:00:00 2001 From: Ihar Hrachyshka Date: Thu, 5 Feb 2026 19:22:05 -0500 Subject: [PATCH 4/4] cache: add read locks to Mapper() and DatabaseModel() Add RLock protection to Mapper() and DatabaseModel() methods to prevent races with Purge() which writes to t.dbModel under the write lock. Without the lock, concurrent access during reconnection causes a race: WARNING: DATA RACE Read at 0x00c0006c95d0 by goroutine 1279: cache.(*TableCache).DatabaseModel() cache/cache.go:886 <-- return t.dbModel (no lock) Previous write at 0x00c0006c95d0 by goroutine 1277: cache.(*TableCache).Purge() cache/cache.go:1017 <-- t.dbModel = dbModel (with lock) Signed-off-by: Ihar Hrachyshka Assisted-by: opus (claude-opus-4-5-20251101) --- cache/cache.go | 4 ++++ 1 file changed, 4 insertions(+) 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 }