Repository: sxyazi/yazi
File: yazi-scheduler/src/file/file.rs (Rust)

Code before:
				})
			},
			|err| {
				self.ops.out(id, FileOutCut::Deform(err));
			},
		)
		.await?;

		Ok(self.ops.out(id, FileOutCut::Succ))
	}

	pub(crate) async fn cut_do(&self, mut task: FileInCut) -> Result<(), FileOutCutDo> {
		ok_or_not_found!(provider::remove_file(&task.to).await); // FIXME: integrate into copy_with_progress
		let mut it = provider::copy_with_progress(&task.from, &task.to, task.cha.unwrap()).await?;

		while let Some(res) = it.recv().await {
			match res {
				Ok(0) => {
					provider::remove_file(&task.from).await.ok();
					break;
				}
				Ok(n) => self.ops.out(task.id, FileOutCutDo::Adv(n)),
				Err(e) if e.kind() == NotFound => {
					warn!("Cut task partially done: {task:?}");
					break;
				}
				// Operation not permitted (os error 1)
				// Attribute not found (os error 93)
				Err(e)
					if task.retry < YAZI.tasks.bizarre_retry
						&& matches!(e.raw_os_error(), Some(1) | Some(93)) =>
				{
					task.retry += 1;
					self.ops.out(task.id, FileOutCutDo::Log(format!("Retrying due to error: {e}")));
					return Ok(self.queue(task, LOW));
				}
				Err(e) => Err(e)?,
			}
		}
		Ok(self.ops.out(task.id, FileOutCutDo::Succ))
	}

	pub(crate) fn link(&self, task: FileInLink) -> Result<(), FileOutLink> {
		Ok(self.queue(task, NORMAL))
	}

	pub(crate) async fn link_do(&self, task: FileInLink) -> Result<(), FileOutLink> {
		let mut src: PathCow = if task.resolve {
			ok_or_not_found!(
				provider::read_link(&task.from).await,
				return Ok(self.ops.out(task.id, FileOutLink::Succ))

Review comment (bug):
Race condition with bounded channel: The channel has capacity `len` (line 123), and the receiving loop starts after all tasks are queued (lines 124-126). If tasks execute and complete faster than the receiving loop starts and processes messages (line 128), the channel could fill up. When additional tasks try to send notifications via `try_send()` in the Drop impl (in.rs:47), they will fail silently because the channel is full. The loop will then block indefinitely waiting for messages that were never sent.

While this is unlikely for SFTP operations (which are typically slow), it's theoretically possible for tasks that fail immediately or for very fast operations.

Consider dropping `tx` before entering the receiving loop and modifying the loop to handle `recv()` returning `None`:
```rust
drop(tx);
for _ in 0..len {
    if rx.recv().await.is_none() {
        break;  // All senders dropped
    }
}
```
```suggestion
			drop(tx);
			for _ in 0..len {
				if rx.recv().await.is_none() {
					break;
				}
```

Code after:
				if nofollow {
					self.queue(task.into_link(), NORMAL);
				} else {
					match (cha.is_link(), reorder) {
						(_, false) => self.queue(task, LOW),
						(true, true) => links.push(task),
						(false, true) => files.push(task),
					}
				};

				async { Ok(()) }
			},
			|err| {
				self.ops.out(id, FileOutCut::Deform(err));
			},
		)
		.await?;

		if !links.is_empty() {
			let len = links.len();
			let (tx, mut rx) = mpsc::channel(len);
			for task in links {
				self.queue(task.with_drop(&tx), LOW);
			}
			for _ in 0..len {
				rx.recv().await;
			}
		}

		for task in files {
			self.queue(task, LOW);
		}

		Ok(self.ops.out(id, FileOutCut::Succ))
	}

	pub(crate) async fn cut_do(&self, mut task: FileInCut) -> Result<(), FileOutCutDo> {
		ok_or_not_found!(provider::remove_file(&task.to).await);
		let mut it = provider::copy_with_progress(&task.from, &task.to, task.cha.unwrap()).await?;

		while let Some(res) = it.recv().await {
			match res {
				Ok(0) => {
					provider::remove_file(&task.from).await.ok();
					break;
				}
				Ok(n) => self.ops.out(task.id, FileOutCutDo::Adv(n)),
				Err(e) if e.kind() == NotFound => {
					warn!("Cut task partially done: {task:?}");
					break;
				}