From 50099f2425a779851a524d4425fd0237530f9dff Mon Sep 17 00:00:00 2001 From: Ihar Hrachyshka Date: Thu, 5 Feb 2026 18:00:44 -0500 Subject: [PATCH 1/3] 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: 1. Adding a disconnected flag (protected by a mutex) that is set before stopping handlers in handleDisconnectNotification() 2. Checking this flag in transact() before sending to trafficSeen 3. Not closing trafficSeen in handleDisconnectNotification() - instead let handleInactivityProbes() clean up when it exits via stopCh The disconnected flag prevents sends to trafficSeen without the race condition that would occur if we closed the channel explicitly. 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. The test is disabled by default in short mode because it revealed several data races in other places in this library as well as in rpc2 dependency. The test is disabled for now to avoid breaking CI job, with the expectation that eventually it will be enabled once known races are fixed. Fixes #464 Signed-off-by: Ihar Hrachyshka Assisted-by: opus (claude-opus-4-5-20251101) --- client/client.go | 32 ++++++- client/client_test.go | 203 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 231 insertions(+), 4 deletions(-) diff --git a/client/client.go b/client/client.go index ff88e338..7247740a 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{} @@ -431,6 +436,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{}) } @@ -858,7 +865,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 +1258,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 +1374,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..1dbba4a6 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" @@ -2064,6 +2067,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) From 747607a2fdd08585c0c5143839544fffe5f921ad Mon Sep 17 00:00:00 2001 From: Ihar Hrachyshka Date: Wed, 4 Mar 2026 12:28:01 -0500 Subject: [PATCH 2/3] client: start handleDisconnectNotification after all WaitGroup Add() calls Move the start of handleDisconnectNotification goroutine to after all handlerShutdown.Add() calls. This prevents a race where handleDisconnectNotification could call handlerShutdown.Wait() during reconnection before all handlers have been registered, causing a premature return from Wait(). The goroutine must be started after all handler registration is complete to ensure Wait() will block until all handlers have actually shut down. 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 7247740a..2b638c0e 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 608484034552bb24d65fab0343b0f4894d44ae88 Mon Sep 17 00:00:00 2001 From: Ihar Hrachyshka Date: Wed, 4 Mar 2026 12:50:23 -0500 Subject: [PATCH 3/3] client: fix SetOption() to check connected flag instead of rpcClient TestOpsWaitForReconnect was failing because after Disconnect() is called, the connected flag is set to false immediately, but rpcClient is only set to nil later by handleDisconnectNotification(). This creates a window where SetOption() would check rpcClient != nil and fail even though the client is disconnected. Fix by checking the connected flag instead of rpcClient in SetOption(). This is the correct check since connected accurately reflects the client state from the caller's perspective. Fixes TestOpsWaitForReconnect integration test. Signed-off-by: Ihar Hrachyshka Assisted-by: opus (claude-opus-4-5-20251101) --- client/client.go | 2 +- client/client_test.go | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/client/client.go b/client/client.go index 2b638c0e..75035c72 100644 --- a/client/client.go +++ b/client/client.go @@ -586,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) diff --git a/client/client_test.go b/client/client_test.go index 1dbba4a6..4ebee829 100644 --- a/client/client_test.go +++ b/client/client_test.go @@ -950,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")