fix(executor): fall back to full scan when sparse range scan finds no results

Collections without a secondary index on the queried field return empty
from the sparse range scan. Previously the handler returned that empty
result to the client. Now it falls back to a full collection scan,
applies the same sort used by the document sort handler, truncates to
the requested limit, and encodes the result — matching the behavior the
control plane expects when no index exists.

Also fix doc_format's MessagePack detection: zerompk can spuriously
succeed on raw JSON bytes by interpreting the leading '{' byte (0x7B)
as a fixint. Guard the branch to only accept Object values, so JSON
payloads stored in document collections are not misidentified as
MessagePack and garbled on read.

Widen sort_rows visibility from pub(super) to pub(in crate::data::executor)
so the snapshot handler can reuse it without duplication.

Author: farhan-syah

nodedb/src/data/executor/doc_format.rs

@@ -74,7 +74,9 @@ pub(super) fn json_to_msgpack(bytes: &[u8]) -> Vec<u8> {
     }
 
     // Try decoding as nodedb_types::Value (zerompk tagged format) and transcode.
-    if let Ok(val) = nodedb_types::value_from_msgpack(bytes) {
+    // Only accept if the result is an Object — zerompk can spuriously "succeed"
+    // on JSON bytes by interpreting the first byte (e.g. '{' = 0x7B) as a fixint.
+    if let Ok(val @ nodedb_types::Value::Object(_)) = nodedb_types::value_from_msgpack(bytes) {
         let json: serde_json::Value = val.into();
         if let Ok(mp) = nodedb_types::json_to_msgpack(&json) {
             return mp;

nodedb/src/data/executor/handlers/control/snapshot.rs

@@ -80,24 +80,73 @@ impl CoreLoop {
         limit: usize,
     ) -> Response {
         debug!(core = self.core_id, %collection, %field, limit, "range scan");
-        match self
+
+        // Try index-backed range scan first.
+        let results = match self
             .sparse
             .range_scan(tid, collection, field, lower, upper, limit)
         {
-            Ok(results) => match super::super::super::response_codec::encode(&results) {
-                Ok(payload) => self.response_with_payload(task, payload),
+            Ok(r) => r,
+            Err(e) => {
+                warn!(core = self.core_id, error = %e, "sparse range scan failed");
+                return self.response_error(
+                    task,
+                    ErrorCode::Internal {
+                        detail: e.to_string(),
+                    },
+                );
+            }
+        };
+
+        // If the index returned nothing, fall back to full scan + sort.
+        // This handles collections without a secondary index on `field`.
+        if results.is_empty() {
+            let scan_result = self.scan_collection(tid, collection, limit.max(1000));
+            match scan_result {
+                Ok(mut docs) => {
+                    super::super::document::sort::sort_rows(
+                        &mut docs,
+                        &[(field.to_string(), true)],
+                    );
+                    docs.truncate(limit);
+                    let rows: Vec<_> = docs
+                        .iter()
+                        .map(|(id, val)| {
+                            let data = super::super::super::doc_format::decode_document(val)
+                                .unwrap_or(serde_json::Value::Null);
+                            super::super::super::response_codec::DocumentRow {
+                                id: id.clone(),
+                                data,
+                            }
+                        })
+                        .collect();
+                    match super::super::super::response_codec::encode(&rows) {
+                        Ok(payload) => return self.response_with_payload(task, payload),
+                        Err(e) => {
+                            return self.response_error(
+                                task,
+                                ErrorCode::Internal {
+                                    detail: e.to_string(),
+                                },
+                            );
+                        }
+                    }
+                }
                 Err(e) => {
-                    warn!(core = self.core_id, error = %e, "range scan serialization failed");
-                    self.response_error(
+                    return self.response_error(
                         task,
                         ErrorCode::Internal {
                             detail: e.to_string(),
                         },
-                    )
+                    );
                 }
-            },
+            }
+        }
+
+        match super::super::super::response_codec::encode(&results) {
+            Ok(payload) => self.response_with_payload(task, payload),
             Err(e) => {
-                warn!(core = self.core_id, error = %e, "sparse range scan failed");
+                warn!(core = self.core_id, error = %e, "range scan serialization failed");
                 self.response_error(
                     task,
                     ErrorCode::Internal {

nodedb/src/data/executor/handlers/document/sort.rs

@@ -169,7 +169,7 @@ pub(super) fn compare_docs_by_keys_binary(
 /// Each entry is `Option<(usize, usize)>` — byte range of the sort key value.
 type SortKeyOffsets = Vec<Option<(usize, usize)>>;
 
-pub(super) fn sort_rows(rows: &mut [(String, Vec<u8>)], sort_keys: &[(String, bool)]) {
+pub(in crate::data::executor) fn sort_rows(rows: &mut [(String, Vec<u8>)], sort_keys: &[(String, bool)]) {
     if sort_keys.is_empty() {
         return;
     }