fix(core): reject unsupported list options instead of ignoring them

bindings/cpp/tests/basic_test.cpp

@@ -265,9 +265,6 @@ TEST(OpenDALOptionsTest, ListOptions) {
   opendal::ListOptions options;
   options.recursive = true;
   options.limit = 16;
-  options.start_after = "list_options/";
-  options.versions = true;
-  options.deleted = true;
 
   auto entries = op.List("list_options/", options);
   std::unordered_set<std::string> paths;
@@ -284,6 +281,23 @@ TEST(OpenDALOptionsTest, ListOptions) {
   EXPECT_TRUE(paths.find("list_options/nested/file") != paths.end());
 }
 
+TEST(OpenDALOptionsTest, ListOptionsUnsupportedByService) {
+  opendal::Operator op("memory");
+  op.Write("list_unsupported/file", "hello");
+
+  opendal::ListOptions start_after;
+  start_after.start_after = "list_unsupported/file";
+  EXPECT_THROW(op.List("list_unsupported/", start_after), std::exception);
+
+  opendal::ListOptions versions;
+  versions.versions = true;
+  EXPECT_THROW(op.List("list_unsupported/", versions), std::exception);
+
+  opendal::ListOptions deleted;
+  deleted.deleted = true;
+  EXPECT_THROW(op.List("list_unsupported/", deleted), std::exception);
+}
+
 TEST(OpenDALOptionsTest, DeleteOptions) {
   opendal::Operator op("memory");
   op.CreateDir("delete_options/");

bindings/ruby/test/lister_test.rb

@@ -67,11 +67,9 @@ class ListerTest < ActiveSupport::TestCase
     assert_equal ["/", "sample", "sub/"], lists
   end
 
-  test "lists the directory with start_after" do
-    lister = @op.list("", start_after: "sub/")
+  test "rejects start_after when the service does not support it" do
+    error = assert_raises(RuntimeError) { @op.list("", start_after: "sub/") }
 
-    lists = lister.map(&:to_h).map { |e| e[:path] }.sort
-
-    assert_equal ["/", "sample", "sub/"], lists # fs backend doesn't support start_after
+    assert_match(/does not support the operation list with the arguments start_after/, error.message)
   end
 end

core/core/src/layers/correctness_check.rs

@@ -160,6 +160,26 @@ impl CorrectnessService {
 
         Ok(())
     }
+
+    fn check_list_args(&self, args: &OpList) -> Result<()> {
+        let capability = self.capability();
+        let scheme = self.info().scheme();
+        if args.start_after().is_some() && !capability.list_with_start_after {
+            return Err(new_unsupported_error(
+                scheme,
+                Operation::List,
+                "start_after",
+            ));
+        }
+        if args.versions() && !capability.list_with_versions {
+            return Err(new_unsupported_error(scheme, Operation::List, "versions"));
+        }
+        if args.deleted() && !capability.list_with_deleted {
+            return Err(new_unsupported_error(scheme, Operation::List, "deleted"));
+        }
+
+        Ok(())
+    }
 }
 
 impl Service for CorrectnessService {
@@ -419,6 +439,7 @@ impl Service for CorrectnessService {
     }
 
     fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
+        self.check_list_args(&args)?;
         self.inner.list(ctx, path, args)
     }
 
@@ -846,6 +867,43 @@ mod tests {
         assert!(res.is_ok())
     }
 
+    #[tokio::test]
+    async fn test_list() {
+        let op = new_test_operator(Capability {
+            list: true,
+            ..Default::default()
+        });
+        let res = op.list_with("path/").start_after("path/key").await;
+        assert_eq!(res.unwrap_err().kind(), ErrorKind::Unsupported);
+        let res = op.list_with("path/").versions(true).await;
+        assert_eq!(res.unwrap_err().kind(), ErrorKind::Unsupported);
+        let res = op.list_with("path/").deleted(true).await;
+        assert_eq!(res.unwrap_err().kind(), ErrorKind::Unsupported);
+
+        let op = new_test_operator(Capability {
+            list: true,
+            list_with_start_after: true,
+            list_with_versions: true,
+            list_with_deleted: true,
+            ..Default::default()
+        });
+        assert!(op.list_with("path/").start_after("path/key").await.is_ok());
+        assert!(op.list_with("path/").versions(true).await.is_ok());
+        assert!(op.list_with("path/").deleted(true).await.is_ok());
+    }
+
+    /// `limit` is a backend hint and `recursive` is simulated by `SimulateLayer`,
+    /// so neither is gated on a capability.
+    #[tokio::test]
+    async fn test_list_limit_and_recursive_need_no_capability() {
+        let op = new_test_operator(Capability {
+            list: true,
+            ..Default::default()
+        });
+        assert!(op.list_with("path/").limit(1).await.is_ok());
+        assert!(op.list_with("path/").recursive(true).await.is_ok());
+    }
+
     #[tokio::test]
     async fn test_compose() {
         let op = new_test_operator(Capability {

core/core/src/types/operator/operator_futures.rs

@@ -1416,6 +1416,9 @@ impl<F: Future<Output = Result<Vec<Entry>>>> FutureList<F> {
 
     /// The start_after passes to underlying service to specify the specified key
     /// to start listing from.
+    ///
+    /// Requires [`Capability::list_with_start_after`]; otherwise the operation
+    /// fails with [`ErrorKind::Unsupported`].
     pub fn start_after(mut self, v: &str) -> Self {
         self.args.start_after = Some(v.to_string());
         self
@@ -1441,6 +1444,9 @@ impl<F: Future<Output = Result<Vec<Entry>>>> FutureList<F> {
     /// If `false`, version information will be omitted from the `list` results.
     ///
     /// Default to `false`
+    ///
+    /// Requires [`Capability::list_with_versions`]; otherwise the operation
+    /// fails with [`ErrorKind::Unsupported`].
     pub fn versions(mut self, v: bool) -> Self {
         self.args.versions = v;
         self
@@ -1454,6 +1460,9 @@ impl<F: Future<Output = Result<Vec<Entry>>>> FutureList<F> {
     ///
     /// If `true`, subsequent `list` operations will include deleted files or versions.
     /// If `false`, deleted files or versions will be excluded from the `list` results.
+    ///
+    /// Requires [`Capability::list_with_deleted`]; otherwise the operation
+    /// fails with [`ErrorKind::Unsupported`].
     pub fn deleted(mut self, v: bool) -> Self {
         self.args.deleted = v;
         self
@@ -1477,6 +1486,9 @@ impl<F: Future<Output = Result<Lister>>> FutureLister<F> {
 
     /// The start_after passes to underlying service to specify the specified key
     /// to start listing from.
+    ///
+    /// Requires [`Capability::list_with_start_after`]; otherwise the operation
+    /// fails with [`ErrorKind::Unsupported`].
     pub fn start_after(mut self, v: &str) -> Self {
         self.args.start_after = Some(v.to_string());
         self
@@ -1502,6 +1514,9 @@ impl<F: Future<Output = Result<Lister>>> FutureLister<F> {
     /// If `false`, version information will be omitted from the `list` results.
     ///
     /// Default to `false`
+    ///
+    /// Requires [`Capability::list_with_versions`]; otherwise the operation
+    /// fails with [`ErrorKind::Unsupported`].
     pub fn versions(mut self, v: bool) -> Self {
         self.args.versions = v;
         self
@@ -1515,6 +1530,9 @@ impl<F: Future<Output = Result<Lister>>> FutureLister<F> {
     ///
     /// If `true`, subsequent `list` operations will include deleted files or versions.
     /// If `false`, deleted files or versions will be excluded from the `list` results.
+    ///
+    /// Requires [`Capability::list_with_deleted`]; otherwise the operation
+    /// fails with [`ErrorKind::Unsupported`].
     pub fn deleted(mut self, v: bool) -> Self {
         self.args.deleted = v;
         self

core/core/src/types/options.rs

@@ -133,12 +133,18 @@ pub struct ListOptions {
     pub limit: Option<usize>,
     /// The start_after passes to underlying service to specify the specified key
     /// to start listing from.
+    ///
+    /// Requires [`Capability::list_with_start_after`](crate::Capability::list_with_start_after).
     pub start_after: Option<String>,
     /// Whether to list recursively under the prefix; default `false`.
     pub recursive: bool,
-    /// Include object versions when supported by the backend; default `false`.
+    /// Include object versions; default `false`.
+    ///
+    /// Requires [`Capability::list_with_versions`](crate::Capability::list_with_versions).
     pub versions: bool,
-    /// Include delete markers when supported by version-aware backends; default `false`.
+    /// Include delete markers; default `false`.
+    ///
+    /// Requires [`Capability::list_with_deleted`](crate::Capability::list_with_deleted).
     pub deleted: bool,
 }