chore: Use only one lock

Spin up docker

Signed-off-by: Javier Aliaga <javier@diagrid.io>

Author: javier-aliaga

bindings/sftp/client.go

@@ -18,18 +18,28 @@ import (
 	"fmt"
 	"os"
 	"sync"
+	"sync/atomic"
 
 	sftpClient "github.com/pkg/sftp"
 	"golang.org/x/crypto/ssh"
 )
 
 type Client struct {
-	sshClient  *ssh.Client
-	sftpClient *sftpClient.Client
-	address    string
-	config     *ssh.ClientConfig
-	lock       sync.RWMutex
-	rLock      sync.Mutex
+	sshClient         *ssh.Client
+	sftpClient        *sftpClient.Client
+	address           string
+	config            *ssh.ClientConfig
+	lock              sync.RWMutex
+	reconnectionCount atomic.Uint64
+}
+
+type sftpError struct {
+	err               error
+	reconnectionCount uint64
+}
+
+func (s sftpError) Error() string {
+	return s.err.Error()
 }
 
 func newClient(address string, config *ssh.ClientConfig) (*Client, error) {
@@ -66,12 +76,15 @@ func (c *Client) Close() error {
 func (c *Client) list(path string) ([]os.FileInfo, error) {
 	var fi []os.FileInfo
 
-	fn := func() error {
+	fn := func() *sftpError {
 		var err error
 		c.lock.RLock()
 		defer c.lock.RUnlock()
 		fi, err = c.sftpClient.ReadDir(path)
-		return err
+		if err != nil {
+			return &sftpError{err: err, reconnectionCount: c.reconnectionCount.Load()}
+		}
+		return nil
 	}
 
 	err := withReconnection(c, fn)
@@ -87,17 +100,17 @@ func (c *Client) create(path string) (*sftpClient.File, string, error) {
 
 	var file *sftpClient.File
 
-	createFn := func() error {
+	createFn := func() *sftpError {
 		c.lock.RLock()
 		defer c.lock.RUnlock()
 		cErr := c.sftpClient.MkdirAll(dir)
 		if cErr != nil {
-			return fmt.Errorf("sftp binding error: error create dir %s: %w", dir, cErr)
+			return &sftpError{err: fmt.Errorf("sftp binding error: error create dir %s: %w", dir, cErr), reconnectionCount: c.reconnectionCount.Load()}
 		}
 
 		file, cErr = c.sftpClient.Create(path)
 		if cErr != nil {
-			return fmt.Errorf("sftp binding error: error create file %s: %w", path, cErr)
+			return &sftpError{err: fmt.Errorf("sftp binding error: error create file %s: %w", path, cErr), reconnectionCount: c.reconnectionCount.Load()}
 		}
 
 		return nil
@@ -114,12 +127,16 @@ func (c *Client) create(path string) (*sftpClient.File, string, error) {
 func (c *Client) get(path string) (*sftpClient.File, error) {
 	var f *sftpClient.File
 
-	fn := func() error {
+	fn := func() *sftpError {
 		var err error
 		c.lock.RLock()
+
 		defer c.lock.RUnlock()
 		f, err = c.sftpClient.Open(path)
-		return err
+		if err != nil {
+			return &sftpError{err: err, reconnectionCount: c.reconnectionCount.Load()}
+		}
+		return nil
 	}
 
 	err := withReconnection(c, fn)
@@ -131,12 +148,15 @@ func (c *Client) get(path string) (*sftpClient.File, error) {
 }
 
 func (c *Client) delete(path string) error {
-	fn := func() error {
+	fn := func() *sftpError {
 		var err error
 		c.lock.RLock()
 		defer c.lock.RUnlock()
 		err = c.sftpClient.Remove(path)
-		return err
+		if err != nil {
+			return &sftpError{err: err, reconnectionCount: c.reconnectionCount.Load()}
+		}
+		return nil
 	}
 
 	err := withReconnection(c, fn)
@@ -157,7 +177,7 @@ func (c *Client) ping() error {
 	return nil
 }
 
-func withReconnection(c *Client, fn func() error) error {
+func withReconnection(c *Client, fn func() *sftpError) error {
 	err := fn()
 	if err == nil {
 		return nil
@@ -167,7 +187,7 @@ func withReconnection(c *Client, fn func() error) error {
 		return err
 	}
 
-	rErr := doReconnect(c)
+	rErr := doReconnect(c, err.reconnectionCount)
 	if rErr != nil {
 		return errors.Join(err, rErr)
 	}
@@ -180,33 +200,14 @@ func withReconnection(c *Client, fn func() error) error {
 	return nil
 }
 
-// 1) c.rLock (sync.Mutex) — reconnect serialization:
-//   - Ensures only one goroutine performs the reconnect sequence at a time
-//     (ping/check, dial SSH, create SFTP client), preventing a thundering herd
-//     of concurrent reconnect attempts.
-//   - Does NOT protect day-to-day client usage; it only coordinates who
-//     is allowed to perform a reconnect.
-//
-// 2) c.lock (sync.RWMutex) — data-plane safety and atomic swap:
-//   - Guards reads/writes of the active client handles (sshClient, sftpClient).
-//   - Regular operations hold RLock while using the clients.
-//   - Reconnect performs a short critical section with Lock to atomically swap
-//     the client pointers; old clients are closed after unlocking to keep the
-//     critical section small and avoid blocking readers.
-//
-// Why not a single RWMutex?
-//   - If we used only c.lock and held it while dialing/handshaking, all I/O would
-//     be blocked for the entire network operation, increasing latency and risk of
-//     contention. Worse, reconnects triggered while a caller holds RLock could
-//     deadlock or starve the writer.
-//   - Separating concerns allows: (a) fast, minimal swap under c.lock, and
-//     (b) serialized reconnect work under c.rLock without blocking readers.
-func doReconnect(c *Client) error {
-	c.rLock.Lock()
-	defer c.rLock.Unlock()
+func doReconnect(c *Client, reconnectionCount uint64) error {
+	// No need to reconnect as it has been reconnected
+	if reconnectionCount != c.reconnectionCount.Load() {
+		return nil
+	}
 
 	err := c.ping()
-	if !shouldReconnect(err) {
+	if err == nil {
 		return nil
 	}
 
@@ -222,11 +223,19 @@ func doReconnect(c *Client) error {
 	}
 
 	// Swap under short lock; close old clients after unlocking.
+	// Close new clients if not swapped
 	c.lock.Lock()
-	oldSftp := c.sftpClient
-	oldSSH := c.sshClient
-	c.sftpClient = newSftpClient
-	c.sshClient = sshClient
+	var oldSftp *sftpClient.Client
+	var oldSSH *ssh.Client
+	if reconnectionCount == c.reconnectionCount.Load() {
+		oldSftp = c.sftpClient
+		oldSSH = c.sshClient
+		c.sftpClient = newSftpClient
+		c.sshClient = sshClient
+		c.reconnectionCount.Add(1)
+		sshClient = nil
+		newSftpClient = nil
+	}
 	c.lock.Unlock()
 
 	if oldSftp != nil {
@@ -236,13 +245,21 @@ func doReconnect(c *Client) error {
 		_ = oldSSH.Close()
 	}
 
+	if newSftpClient != nil {
+		_ = newSftpClient.Close()
+	}
+
+	if sshClient != nil {
+		_ = sshClient.Close()
+	}
+
 	return nil
 }
 
 func newSSHClient(address string, config *ssh.ClientConfig) (*ssh.Client, error) {
 	sshClient, err := ssh.Dial("tcp", address, config)
 	if err != nil {
-		return nil, fmt.Errorf("sftp binding error: error create ssh client: %w", err)
+		return nil, fmt.Errorf("sftp binding error: error dialing ssh server: %w", err)
 	}
 	return sshClient, nil
 }

bindings/sftp/docker-compose.yaml

@@ -0,0 +1,11 @@
+services:
+  sftp:
+    image:
+      atmoz/sftp
+    environment:
+      - SFTP_USERS=foo:pass:1001:1001:upload
+    volumes:
+      - ./upload:/home/foo/upload
+    ports:
+      - "2222:22"
+

bindings/sftp/sftp_integration_test.go

@@ -1,3 +1,5 @@
+//go:build integration_test
+
 /*
 Copyright 2025 The Dapr Authors
 Licensed under the Apache License, Version 2.0 (the "License");
@@ -14,40 +16,57 @@ limitations under the License.
 package sftp
 
 import (
+	"context"
 	"encoding/json"
+	"math/rand"
 	"os"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/dapr/components-contrib/bindings"
+	"github.com/dapr/components-contrib/tests/certification/flow"
+	"github.com/dapr/components-contrib/tests/certification/flow/dockercompose"
 	sftp "github.com/dapr/components-contrib/tests/utils/sftpproxy"
 )
 
-const ProxySftp = "0.0.0.0:2223"
-
-var connectionStringEnvKey = "DAPR_TEST_SFTP_CONNSTRING"
+const (
+	ProxySftp        = "0.0.0.0:2223"
+	ConnectionString = "0.0.0.0:2222"
+)
 
-// Run docker from the file location as the upload folder is relative to the test
-// cd proxy
-// docker run --name sftp -v ./upload:/home/foo/upload -p 2222:22 -d atmoz/sftp foo:pass:1001
-// export DAPR_TEST_SFTP_CONNSTRING=localhost:2222
 func TestIntegrationCases(t *testing.T) {
-	connectionString := os.Getenv(connectionStringEnvKey)
-	if connectionString == "" {
-		t.Skipf("sftp binding integration skipped. To enable this test, define the connection string using environment variable '%[1]s' (example 'export %[1]s=\"localhost:2222\")'", connectionStringEnvKey)
-	}
-
+	cleanUp := setupSftp(t)
+	defer cleanUp()
+	time.Sleep(1 * time.Second)
 	t.Run("List operation", testListOperation)
 	t.Run("Create operation", testCreateOperation)
 	t.Run("Reconnections", testReconnect)
 }
 
+func setupSftp(t *testing.T) func() {
+	dc := dockercompose.New("sftp", "docker-compose.yaml")
+	ctx := flow.Context{
+		T:       t,
+		Context: t.Context(),
+		Flow:    nil,
+	}
+	err := dc.Up(ctx)
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return func() { dc.Down(ctx) }
+}
+
 func testListOperation(t *testing.T) {
 	proxy := &sftp.Proxy{
 		ListenAddr:   ProxySftp,
-		UpstreamAddr: os.Getenv(connectionStringEnvKey),
+		UpstreamAddr: ConnectionString,
 	}
 
 	defer proxy.Close()
@@ -82,7 +101,7 @@ func testListOperation(t *testing.T) {
 func testCreateOperation(t *testing.T) {
 	proxy := &sftp.Proxy{
 		ListenAddr:   ProxySftp,
-		UpstreamAddr: os.Getenv(connectionStringEnvKey),
+		UpstreamAddr: ConnectionString,
 	}
 	defer proxy.Close()
 	go proxy.ListenAndServe()
@@ -114,7 +133,7 @@ func testCreateOperation(t *testing.T) {
 	require.NoError(t, err)
 	assert.NotNil(t, r.Data)
 
-	file, err := os.Stat("./proxy/upload/test.txt")
+	file, err := os.Stat("./upload/test.txt")
 	require.NoError(t, err)
 	assert.Equal(t, "test.txt", file.Name())
 	assert.EqualValues(t, 1, proxy.ReconnectionCount.Load())
@@ -123,7 +142,7 @@ func testCreateOperation(t *testing.T) {
 func testReconnect(t *testing.T) {
 	proxy := &sftp.Proxy{
 		ListenAddr:   ProxySftp,
-		UpstreamAddr: os.Getenv(connectionStringEnvKey),
+		UpstreamAddr: ConnectionString,
 	}
 	defer proxy.Close()
 	go proxy.ListenAndServe()
@@ -218,4 +237,60 @@ func testReconnect(t *testing.T) {
 
 		assert.EqualValues(t, numReconnects+1, proxy.ReconnectionCount.Load())
 	})
+
+	t.Run("Parallel ops - reconnection", func(t *testing.T) {
+		numReconnects := proxy.ReconnectionCount.Load()
+		ctx, cancelFn := context.WithCancel(t.Context())
+		opCount := atomic.Int32{}
+		opFailed := atomic.Int32{}
+		for range 10 {
+			go func(ctx context.Context) {
+				for {
+					select {
+					case <-ctx.Done():
+						return
+					case <-time.After(time.Duration(500*rand.Float32()) * time.Millisecond):
+						opCount.Add(1)
+						r, err := c.Invoke(t.Context(), &bindings.InvokeRequest{Operation: bindings.ListOperation})
+						if err != nil {
+							opFailed.Add(1)
+							break
+						}
+
+						assert.NotNil(t, r.Data)
+					}
+				}
+			}(ctx)
+		}
+
+		go func(ctx context.Context) {
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case <-time.After(100 * time.Millisecond):
+					_ = proxy.KillServerConn()
+				}
+			}
+
+		}(ctx)
+
+		time.Sleep(time.Second * 5)
+		cancelFn()
+
+		totalOps := opCount.Load()
+		failedOps := opFailed.Load()
+
+		// Calculate 5% tolerance
+		tolerance := float64(totalOps) * 0.05
+
+		// Assert that failed operations are within 1% of total operations
+		assert.InDelta(t, 0, failedOps, tolerance,
+			"Expected less than 1%% of operations to fail. Total: %d, Failed: %d (%.2f%%)",
+			totalOps, failedOps, (float64(failedOps)/float64(totalOps))*100)
+
+		expectedReconnects := numReconnects + 10
+		currentReconnects := proxy.ReconnectionCount.Load()
+		assert.InDelta(t, expectedReconnects, currentReconnects, 2.0, "Expected %d reconnections, got %d", expectedReconnects, currentReconnects)
+	})
 }