Implement PieceNotifier and improve download scheduling logic (#1941)

* feat(task): wait for in-flight pieces before scheduling download

- Before scheduling, check if the task is prefetched and wait for
  in-flight pieces to finish writing to local storage
- Poll local storage at `DEFAULT_WAIT_FOR_PIECE_FINISHED_INTERVAL`
  intervals up to `DEFAULT_WAIT_FOR_IN_FLIGHT_PIECE_TIMEOUT` (500ms)
- Stop early if all interested pieces finish or none remain in-flight
- Reduces scheduler stalls caused by missing candidate parents during
  concurrent prefetch downloads

Signed-off-by: Gaius <gaius.qi@gmail.com>

* refactor: move prefetch task started log to after task initiation

- Move "prefetch task started" log to just before the actual download
  begins
- Ensures the log reflects task initiation more accurately
- Upgrade instrument log level from `debug` to default (`info`) for
  proxy handlers

Signed-off-by: Gaius <gaius.qi@gmail.com>

* style(grpc): downgrade "task is already prefetched" log to debug level

Changes `info!` to `debug!` for "task is already prefetched" in both
download and upload gRPC handlers, reducing log noise for a
normal/expected condition.

Signed-off-by: Gaius <gaius.qi@gmail.com>

* refactor: remove redundant `level = "debug"` from instrument macros

- Replace `level = "debug"` with default in all `#[instrument]` macros
- Add missing `#[instrument(skip_all)]` to `HuggingFace` and
  `ModelScope` backends
- Downgrade a 416-retry log from `info!` to `debug!` in HTTP backend
- Remove unused `info` import in `http.rs

Signed-off-by: Gaius <gaius.qi@gmail.com>

* feat(storage): add PieceNotifier for event-driven piece completion

- Add `PieceNotifier` struct using `DashMap<String, Arc<Notify>>` to
  enable event-driven piece completion signaling
- Provide `register`, `get`, and `remove_and_notify` methods to manage
  per-piece notifiers across concurrent downloads
- Add `DEFAULT_WAIT_FOR_PIECE_FINISHED_FALLBACK_INTERVAL` (1s) as a
  safety re-check interval in case notifications are missed
- Attach `piece_notifier` field to `Storage` struct
- Demote several `#[instrument]` spans to `level = "debug"` to reduce
  trace noise; remove `#[instrument]` from non-async constructor methods

Signed-off-by: Gaius <gaius.qi@gmail.com>

* refactor(storage): extract PieceNotifier and replace polling with notify

- Move `PieceNotifier` into its own `piece_notifier` module.
- Replace interval-based polling loops with `Notify`-driven wakeups,
  falling back to a timeout interval.
- Register notifier before piece metadata creation; call
  `remove_and_notify` on finish, failure, or error paths.
- Expose `in_flight_piece_notifier()` for external waiters.
- Add tests covering notify-driven wake, fail-on-failure, and
  fast-fail for stale (orphaned) piece metadata.

Signed-off-by: Gaius <gaius.qi@gmail.com>

* refactor(storage): replace polling with notifier in wait_for_cache_piece

- Replace interval-based polling with `Notify`-based notifier in
  `wait_for_cache_piece_finished`
- Register notifier on `download_cache_piece_started`; notify on
  finish, fail, or timeout
- Add fallback sleep interval and double-check loop for race-free
  `None` notifier case
- Notify and clean up on `download_cache_piece_failed` to unblock
  waiters
- Elevate several `#[instrument(level = "debug")]` to default (info)
  level

Signed-off-by: Gaius <gaius.qi@gmail.com>

* refactor: move DEFAULT_WAIT_FOR_PIECE_FINISHED_INTERVAL to dfdaemon_upload

- Moves `DEFAULT_WAIT_FOR_PIECE_FINISHED_INTERVAL` from
  `dragonfly-client-storage` to `dfdaemon_upload` where it's actually
  used
- Simplifies call sites by removing the `dragonfly_client_storage::`
  prefix

Signed-off-by: Gaius <gaius.qi@gmail.com>

* style: remove redundant inline comments from tests

- Remove inline comments that restate obvious code intent
- Affected files: `lib.rs`, `piece_notifier.rs`, and `task.rs` test
  modules
- Comments describing setup steps, assertions, and no-ops are dropped

Signed-off-by: Gaius <gaius.qi@gmail.com>

* fix(storage): prevent failed download from erasing finished piece

- Guard `download_piece_failed` to skip deletion if piece is already
  finished
- Guard `wait_for_piece_finished_failed` with the same check
- Prevents a late/duplicate downloader from erasing a piece completed by
  the winner
- Add unit tests in both `metadata` and `storage` layers to cover the
  fix

Signed-off-by: Gaius <gaius.qi@gmail.com>

* fix(piece): move scopeguard after finished piece check

- Register scopeguard only when piece is not already finished
- Avoids calling `download_piece_failed` on successfully cached pieces
- Removes need for manual `ScopeGuard::into_inner` on early return

Signed-off-by: Gaius <gaius.qi@gmail.com>

* feat(task): add direct source download for seed peer range tasks

- Rename `download_and_ensure_announcement` to `download_with_scheduler`
- For seed peer range tasks where range length ≤ max piece length,
  bypass the scheduler and download directly from source
- Return early after direct source download, skipping scheduler logic

Signed-off-by: Gaius <gaius.qi@gmail.com>

* fix(metric): change slow task detection from Level1 to Level2 threshold

- Change slow task detection from `Level1` to `Level2` size threshold
  for both upload and download tasks
- Add `#[instrument(skip_all)]` to upload/download finished metric
  functions
- Fix trailing comma in `warn!` macro call
- Update `.gitignore` to exclude `.claude` and `.claude-analysis` files

Signed-off-by: Gaius <gaius.qi@gmail.com>

* feat(storage): use atomic claim to prevent duplicate piece downloads

- Replace `register` with atomic `claim` returning `Owner` or `InFlight`
- `Owner` callers proceed to download; `InFlight` callers wait on the
  existing notifier and retry in a loop
- Add double-check after winning the claim to skip download if piece
  already finished
- Apply the same pattern to all four piece download entry points
- Add concurrency test asserting exactly one downloader wins per piece

Signed-off-by: Gaius <gaius.qi@gmail.com>

* fix(fs): run fallocate in spawn_blocking to avoid blocking async runtime

- Move `fallocate` syscall into `spawn_blocking` to prevent blocking
  the async runtime
- Clone the file handle before passing it into the blocking task
- Map `JoinError` to `io::Error` after awaiting the task

Signed-off-by: Gaius <gaius.qi@gmail.com>

* fix(fs): simplify fallocate return with double question mark operator

- Replace explicit `return` with implicit return using `??` operator
- Collapse `map_err(...)?` + `return` into a single `??` expression
- Remove redundant `#[cfg(not(target_os = "linux"))]` attribute before
  `Ok(())

Signed-off-by: Gaius <gaius.qi@gmail.com>

---------

Signed-off-by: Gaius <gaius.qi@gmail.com>