fix(targets): surface truncated store batches instead of returning partial entries (#4423)

fix(targets): surface truncated store batches instead of silently returning partial entries (backlog#967)

get_multiple deserializes a persisted batch by pulling item_count items
from a concatenated JSON stream. When the stream ended early (a truncated
or corrupted batch file), the partial-read branch only logged a warn! and
broke out of the loop, returning Ok(partial). Callers then treated the
batch as fully delivered, deleted the store entry, and permanently lost
the remaining events with no error.

Return StoreError::Deserialization as soon as the stream ends before
item_count items are read, so the caller keeps the entry on disk and can
retry rather than silently dropping events. The empty-file case already
returned an error and remains covered by this path.

Add a regression test that writes a 3-item batch, truncates the file at a
clean boundary after two items, and asserts get_multiple returns Err and
leaves the batch file in place.

Refs: https://github.com/rustfs/backlog/issues/967

crates/targets/src/store.rs

@@ -545,28 +545,27 @@ where
                     return Err(StoreError::Deserialization(format!("Failed to deserialize item in batch: {e}")));
                 }
                 None => {
-                    // Reached end of stream sooner than item_count
+                    // Reached end of stream before deserializing `item_count` items: the
-                    if items.len() < key.item_count && !items.is_empty() {
+                    // batch file was truncated or corrupted. This MUST be surfaced as an
-                        // Partial read
+                    // error rather than silently returning the partial set. If we returned
-                        warn!(
+                    // `Ok(partial)`, the caller would treat the batch as fully delivered,
-                            event = EVENT_TARGET_STORE_STATE,
+                    // delete the store entry, and permanently lose the missing events.
-                            component = LOG_COMPONENT_TARGETS,
+                    warn!(
-                            subsystem = LOG_SUBSYSTEM_STORE,
+                        event = EVENT_TARGET_STORE_STATE,
-                            action = "read_batch",
+                        component = LOG_COMPONENT_TARGETS,
-                            key = %key,
+                        subsystem = LOG_SUBSYSTEM_STORE,
-                            expected_items = key.item_count,
+                        action = "read_batch",
-                            actual_items = items.len(),
+                        key = %key,
-                            reason = "partial_batch_read",
+                        expected_items = key.item_count,
-                            "target store state"
+                        actual_items = items.len(),
-                        );
+                        reason = "truncated_batch_read",
-                        // Depending on strictness, this could be an error.
+                        "target store state"
-                    } else if items.is_empty() {
+                    );
-                        // No items at all, but file existed
+                    return Err(StoreError::Deserialization(format!(
-                        return Err(StoreError::Deserialization(format!(
+                        "Truncated batch for key {key}: expected {} items but only deserialized {}",
-                            "No items deserialized for key {key} though file existed."
+                        key.item_count,
-                        )));
+                        items.len()
-                    }
+                    )));
-                    break;
                 }
             }
         }
@@ -800,6 +799,41 @@ mod tests {
         let _ = store.delete();
     }
 
+    #[test]
+    fn get_multiple_errors_on_truncated_batch_instead_of_partial_success() {
+        let dir = temp_store_dir("truncated-batch");
+        let store = QueueStore::<String>::new_with_compression(&dir, 8, ".test", false);
+        store.open().unwrap();
+
+        let items = vec!["aa".to_string(), "bb".to_string(), "cc".to_string()];
+        let key = store.put_multiple(items).unwrap();
+        assert_eq!(key.item_count, 3);
+
+        // Truncate the batch file at a clean boundary after the first two serialized
+        // items (`"aa""bb"`), so the read of the third item hits end-of-stream — the
+        // exact "partial read" condition that previously returned Ok silently.
+        let prefix_len =
+            serde_json::to_vec(&"aa".to_string()).unwrap().len() + serde_json::to_vec(&"bb".to_string()).unwrap().len();
+        let path = store.file_path(&key);
+        let file = std::fs::OpenOptions::new().write(true).open(&path).unwrap();
+        file.set_len(prefix_len as u64).unwrap();
+        drop(file);
+
+        // A truncated batch must be reported as an error, not silently returned as a
+        // partial-but-successful result that would drop the missing "cc" event.
+        let err = store.get_multiple(&key).unwrap_err();
+        assert!(
+            matches!(err, StoreError::Deserialization(_)),
+            "expected Deserialization error, got {err:?}"
+        );
+
+        // Because get_multiple failed, the batch entry is still on disk for the caller
+        // to retry — the missing events are not silently discarded.
+        assert!(store.file_path(&key).exists());
+
+        let _ = store.delete();
+    }
+
     #[test]
     fn concurrent_put_raw_respects_entry_limit() {
         let dir = temp_store_dir("concurrent-limit");