SFTP input - last file not deleted

[joffreychambrin]

Hi 👋

I have discovered an issue when the sftp input component is configured with watcher enabled and delete_on_finish enabled.

Steps to reproduce:

• Upload 10 files to the SFTP
• Run the benthos pipeline (see below for config)
• After the files have been consumed by the pipeline, check the SFTP => one file is still there even if it has been consumed by benthos.

Here is the configuration file I am using

cache_resources:
  - label: memory
    memory:
      default_ttl: 5m
      compaction_interval: 60s
      init_values: {}

input:
  sftp:
    address: ${SFTP_HOST}
    credentials:
      username: ${SFTP_USERNAME}
      password: ${SFTP_PASSWORD}
    paths:
      - "folder/**"
    scanner:
      to_the_end: {}
    delete_on_finish: true
    watcher:
      enabled: true
      minimum_age: 1s
      poll_interval: 20ms
      cache: memory

pipeline:
  processors:
    - log:
        level: INFO
        message: "message received"

output:
  label: ""
  stdout:
    codec: lines

[joffreychambrin]

After looking at the code, it seems to be related to this code: internal/impl/sftp/input.go

	if s.scanner, err = s.scannerCtor.Create(file, func(ctx context.Context, aErr error) (outErr error) {
		_ = s.pathProvider.Ack(ctx, nextPath, aErr)
		if aErr != nil {
			return nil
		}
		if s.deleteOnFinish {
			s.scannerMut.Lock()
			client := s.client
			if client == nil {

The mutex seems to be locked and so the last file is not being deleted

Maybe I am completely wrong, but in this case, do you need the mutex? #2436

[Jeffail]

Hey @joffreychambrin, we need some form of mutex there as we can't otherwise guarantee race conditions there. However, it should definitely be possible to fix this by reducing the scope of those locks.