chore: Handle sftp reconnections

bindings/sftp/client.go

@@ -0,0 +1,264 @@
+/*
+Copyright 2025 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package sftp
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"sync"
+
+	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
+}
+
+func newClient(address string, config *ssh.ClientConfig) (*Client, error) {
+	if address == "" || config == nil {
+		return nil, errors.New("sftp binding error: client not initialized")
+	}
+
+	sshClient, err := newSSHClient(address, config)
+	if err != nil {
+		return nil, err
+	}
+
+	newSftpClient, err := sftpClient.NewClient(sshClient)
+	if err != nil {
+		_ = sshClient.Close()
+		return nil, fmt.Errorf("sftp binding error: error create sftp client: %w", err)
+	}
+
+	return &Client{
+		sshClient:  sshClient,
+		sftpClient: newSftpClient,
+		address:    address,
+		config:     config,
+	}, nil
+}
+
+func (c *Client) Close() error {
+	_ = c.sshClient.Close()
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	return c.sftpClient.Close()
+}
+
+func (c *Client) list(path string) ([]os.FileInfo, error) {
+	var fi []os.FileInfo
+
+	fn := func() error {
+		var err error
+		c.lock.RLock()
+		defer c.lock.RUnlock()
+		fi, err = c.sftpClient.ReadDir(path)
+		return err
+	}
+
+	err := withReconnection(c, fn)
+	if err != nil {
+		return nil, err
+	}
+
+	return fi, nil
+}
+
+func (c *Client) create(path string) (*sftpClient.File, string, error) {
+	dir, fileName := sftpClient.Split(path)
+
+	var file *sftpClient.File
+
+	createFn := func() error {
+		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)
+		}
+
+		file, cErr = c.sftpClient.Create(path)
+		if cErr != nil {
+			return fmt.Errorf("sftp binding error: error create file %s: %w", path, cErr)
+		}
+
+		return nil
+	}
+
+	rErr := withReconnection(c, createFn)
+	if rErr != nil {
+		return nil, "", rErr
+	}
+
+	return file, fileName, nil
+}
+
+func (c *Client) get(path string) (*sftpClient.File, error) {
+	var f *sftpClient.File
+
+	fn := func() error {
+		var err error
+		c.lock.RLock()
+		defer c.lock.RUnlock()
+		f, err = c.sftpClient.Open(path)
+		return err
+	}
+
+	err := withReconnection(c, fn)
+	if err != nil {
+		return nil, err
+	}
+
+	return f, nil
+}
+
+func (c *Client) delete(path string) error {
+	fn := func() error {
+		var err error
+		c.lock.RLock()
+		defer c.lock.RUnlock()
+		err = c.sftpClient.Remove(path)
+		return err
+	}
+
+	err := withReconnection(c, fn)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (c *Client) ping() error {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+	_, err := c.sftpClient.Getwd()
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func withReconnection(c *Client, fn func() error) error {
+	err := fn()
+	if err == nil {
+		return nil
+	}
+
+	if !shouldReconnect(err) {
+		return err
+	}
+
+	rErr := doReconnect(c)
+	if rErr != nil {
+		return errors.Join(err, rErr)
+	}
+
+	err = fn()
+	if err != nil {
+		return err
+	}
+
+	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()
+
+	err := c.ping()
+	if !shouldReconnect(err) {
+		return nil
+	}
+
+	sshClient, err := newSSHClient(c.address, c.config)
+	if err != nil {
+		return err
+	}
+
+	newSftpClient, err := sftpClient.NewClient(sshClient)
+	if err != nil {
+		_ = sshClient.Close()
+		return fmt.Errorf("sftp binding error: error create sftp client: %w", err)
+	}
+
+	// Swap under short lock; close old clients after unlocking.
+	c.lock.Lock()
+	oldSftp := c.sftpClient
+	oldSSH := c.sshClient
+	c.sftpClient = newSftpClient
+	c.sshClient = sshClient
+	c.lock.Unlock()
+
+	if oldSftp != nil {
+		_ = oldSftp.Close()
+	}
+	if oldSSH != nil {
+		_ = oldSSH.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 sshClient, nil
+}
+
+// shouldReconnect returns true if the error looks like a transport-level failure
+func shouldReconnect(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	// SFTP status errors that are logical, not connectivity (avoid reconnect)
+	if errors.Is(err, sftpClient.ErrSSHFxPermissionDenied) ||
+		errors.Is(err, sftpClient.ErrSSHFxNoSuchFile) ||
+		errors.Is(err, sftpClient.ErrSSHFxOpUnsupported) {
+		return false
+	}
+
+	return true
+}

bindings/sftp/sftp.go

@@ -1,3 +1,16 @@
+/*
+Copyright 2025 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package sftp
 
 import (
@@ -25,9 +38,9 @@ const (
 
 // Sftp is a binding for file operations on sftp server.
 type Sftp struct {
-	metadata   *sftpMetadata
-	logger     logger.Logger
-	sftpClient *sftpClient.Client
+	metadata *sftpMetadata
+	logger   logger.Logger
+	c        *Client
 }
 
 // sftpMetadata defines the sftp metadata.
@@ -115,19 +128,12 @@ func (sftp *Sftp) Init(_ context.Context, metadata bindings.Metadata) error {
 		HostKeyCallback: hostKeyCallback,
 	}
 
-	sshClient, err := ssh.Dial("tcp", m.Address, config)
-	if err != nil {
-		return fmt.Errorf("sftp binding error: error create ssh client: %w", err)
-	}
-
-	newSftpClient, err := sftpClient.NewClient(sshClient)
+	sftp.metadata = m
+	sftp.c, err = newClient(m.Address, config)
 	if err != nil {
-		return fmt.Errorf("sftp binding error: error create sftp client: %w", err)
+		return fmt.Errorf("sftp binding error: create sftp client error: %w", err)
 	}
 
-	sftp.metadata = m
-	sftp.sftpClient = newSftpClient
-
 	return nil
 }
 
@@ -161,14 +167,9 @@ func (sftp *Sftp) create(_ context.Context, req *bindings.InvokeRequest) (*bindi
 		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
-	dir, fileName := sftpClient.Split(path)
+	c := sftp.c
 
-	err = sftp.sftpClient.MkdirAll(dir)
-	if err != nil {
-		return nil, fmt.Errorf("sftp binding error: error create dir %s: %w", dir, err)
-	}
-
-	file, err := sftp.sftpClient.Create(path)
+	file, fileName, err := c.create(path)
 	if err != nil {
 		return nil, fmt.Errorf("sftp binding error: error create file %s: %w", path, err)
 	}
@@ -211,7 +212,9 @@ func (sftp *Sftp) list(_ context.Context, req *bindings.InvokeRequest) (*binding
 		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
-	files, err := sftp.sftpClient.ReadDir(path)
+	c := sftp.c
+
+	files, err := c.list(path)
 	if err != nil {
 		return nil, fmt.Errorf("sftp binding error: error read dir %s: %w", path, err)
 	}
@@ -246,7 +249,9 @@ func (sftp *Sftp) get(_ context.Context, req *bindings.InvokeRequest) (*bindings
 		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
-	file, err := sftp.sftpClient.Open(path)
+	c := sftp.c
+
+	file, err := c.get(path)
 	if err != nil {
 		return nil, fmt.Errorf("sftp binding error: error open file %s: %w", path, err)
 	}
@@ -272,7 +277,9 @@ func (sftp *Sftp) delete(_ context.Context, req *bindings.InvokeRequest) (*bindi
 		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
-	err = sftp.sftpClient.Remove(path)
+	c := sftp.c
+
+	err = c.delete(path)
 	if err != nil {
 		return nil, fmt.Errorf("sftp binding error: error remove file %s: %w", path, err)
 	}
@@ -296,7 +303,7 @@ func (sftp *Sftp) Invoke(ctx context.Context, req *bindings.InvokeRequest) (*bin
 }
 
 func (sftp *Sftp) Close() error {
-	return sftp.sftpClient.Close()
+	return sftp.c.Close()
 }
 
 func (metadata sftpMetadata) getPath(requestMetadata map[string]string) (path string, err error) {