false EOF in paginated listings: gather_results infers completion from post-filter count, not producer stop reason

cargo test -p rustfs-ecstore --lib list_path_gather_results_reports_false_eof -- --nocapture

running 1 test
test store::list_objects::test::list_path_gather_results_reports_false_eof_when_producer_stops_at_its_own_limit_after_heavy_filtering ... ok

diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs
index 9a1de1c..f741004 100644
--- a/crates/ecstore/src/store/list_objects.rs
+++ b/crates/ecstore/src/store/list_objects.rs
@@ -7900,6 +7900,81 @@ mod test {
         assert!(cancel.is_cancelled());
     }
 
+    #[tokio::test]
+    async fn list_path_gather_results_reports_false_eof_when_producer_stops_at_its_own_limit_after_heavy_filtering() {
+        // Reproduces the "deeper protocol-level issue" flagged during review of
+        // rustfs/rustfs#7049: gather_results infers whether the listing is
+        // truncated purely from whether ITS OWN `entries.len() >= opts.limit`
+        // was hit. It has no way to learn that the upstream producer (the disk
+        // scan_dir walker, whose per-disk scan is capped independently -- see
+        // `per_disk_limit = opts.limit + 4 + opts.limit / 16` in set_disks list
+        // path) stopped early because IT hit its own, smaller limit deep inside
+        // a subdirectory, rather than because the object tree was actually
+        // exhausted.
+        //
+        // This test simulates that: the producer sends a batch of raw entries
+        // that are entirely filtered out downstream (delete markers, with
+        // incl_deleted = false) and then closes its channel -- exactly what
+        // happens today once a per-disk scan limit is reached mid-subtree, well
+        // before the user's requested `opts.limit` of candidate entries is
+        // reached.
+        let (entry_tx, entry_rx) = mpsc::channel(32);
+        let (result_tx, mut result_rx) = mpsc::channel(1);
+        let cancel = CancellationToken::new();
+
+        let mod_time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+
+        // Producer scans 20 raw entries and then closes the channel, standing
+        // in for a per-disk scan_dir call that hit its own internal limit deep
+        // inside a much larger subtree it never finished walking.
+        for i in 0..20 {
+            entry_tx
+                .send(test_delete_marker_meta_entry(&format!("obj-{i:03}"), mod_time))
+                .await
+                .expect("test entry should be queued");
+        }
+        drop(entry_tx); // channel closes: simulates the producer stopping early
+
+        let opts = ListPathOptions {
+            bucket: "bucket".to_owned(),
+            limit: 100,          // user asked for up to 100 keys
+            incl_deleted: false, // delete markers are filtered out downstream
+            ..Default::default()
+        };
+
+        let state = gather_results(cancel.clone(), opts, entry_rx, result_tx)
+            .await
+            .expect("gather_results should not error");
+
+        let result = result_rx
+            .try_recv()
+            .expect("gather_results should have sent exactly one page");
+
+        // All 20 raw entries were filtered out (all delete markers), so the
+        // page gather_results hands back is empty...
+        assert_eq!(result.entries.unwrap().entries().len(), 0);
+
+        // ...and gather_results reports this as a definitive, non-truncated
+        // EOF: state is InputClosed (not LimitReached), and every call site in
+        // this file treats `err.is_none()` as "the disk may have more" (see
+        // `disk_has_more`). Here `err` is `Some(Unexpected)`, i.e. "no more
+        // data" -- even though the simulated producer only ever scanned 20
+        // entries out of what could be a much larger real subtree, and stopped
+        // for reasons gather_results was never told about. A real disk-side
+        // scan_dir stopping at its own per-disk limit after filtering discards
+        // most raw entries (the linked report measured ~62%) produces this
+        // exact shape, and results in objects being silently dropped from
+        // paginated listings with IsTruncated=false.
+        assert_eq!(state, GatherResultsState::InputClosed);
+        assert!(
+            result.err.is_some(),
+            "gather_results cannot distinguish \"producer hit its own limit\" \
+             from \"producer genuinely exhausted the tree\": both produce this \
+             same InputClosed/err=Some(..) shape, which downstream code reads \
+             as a confident, non-truncated end of listing (false EOF)"
+        );
+    }
+
     #[tokio::test]
     async fn list_path_gather_results_keeps_marker_entry_for_version_marker_listing() {
         let (entry_tx, entry_rx) = mpsc::channel(4);