chore: integrate main for error identity fix

.github/workflows/rustfs-performance-test.yml

@@ -82,6 +82,12 @@ jobs:
   performance-test:
     runs-on: pf-testing
     timeout-minutes: 900
+    env:
+      RUSTFS_BENCH_SCRIPT: ${{ github.workspace }}/auto-testing/rustfs_performance_testing.sh
+      RUSTFS_WARP_METHODS: ${{ inputs.test_method }}
+      RUSTFS_WARP_SIZES: ${{ inputs.object_size }}
+      RUSTFS_WARP_DURATION: ${{ inputs.warp_duration || '5m' }}
+      RUSTFS_WARP_CONCURRENCY: ${{ inputs.warp_concurrency || '64' }}
     # Run on manual dispatch, or when the nightly build completed successfully.
     # Skipped when nightly failed.
     if: ${{ github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch' }}
@@ -158,19 +164,15 @@ jobs:
       - name: Run benchmark (GET/PUT/MIXED)
         id: benchmark
         run: |
-          # Empty on automatic (workflow_run) runs -> full 30 rounds.
-          # Manual dispatch can restrict method(s)/size(s).
-          export WARP_METHODS="${{ inputs.test_method }}"
-          export WARP_SIZES="${{ inputs.object_size }}"
           ./auto-testing/rustfs_performance_test.sh \
             --step 5 -y \
-            --warp-duration "${{ inputs.warp_duration || '5m' }}" \
-            --warp-concurrency "${{ inputs.warp_concurrency || '64' }}" \
             --log-file "${LOG_FILE}"
 
       - name: Analyze results
         if: ${{ steps.benchmark.conclusion == 'success' }}
         run: |
+          export WARP_METHODS="${RUSTFS_WARP_METHODS}" WARP_SIZES="${RUSTFS_WARP_SIZES}"
+          export WARP_DURATION="${RUSTFS_WARP_DURATION}" WARP_CONCURRENCY="${RUSTFS_WARP_CONCURRENCY}"
           ./auto-testing/rustfs_performance_test.sh --step 6 -y --log-file "${LOG_FILE:-/dev/null}"
 
       - name: Collect RustFS version info

crates/targets/src/store.rs

@@ -919,6 +919,22 @@ where
             }
         }
 
+        // The filename's item count is untrusted. Reject a payload that contains
+        // more items than advertised instead of returning success and allowing the
+        // caller to delete the entry with trailing events still in the file.
+        match deserializer.next() {
+            None => {}
+            Some(Ok(_)) => {
+                return Err(StoreError::Deserialization(format!(
+                    "Batch for key {key} contains more than {} items",
+                    key.item_count
+                )));
+            }
+            Some(Err(e)) => {
+                return Err(StoreError::Deserialization(format!("Failed to deserialize trailing batch item: {e}")));
+            }
+        }
+
         if items.is_empty() && key.item_count > 0 {
             return Err(StoreError::Deserialization("No items found".to_string()));
         }
@@ -1381,6 +1397,39 @@ mod tests {
         let _ = store.delete();
     }
 
+    #[test]
+    fn get_multiple_errors_on_batch_with_trailing_items_instead_of_partial_success() {
+        let dir = temp_store_dir("trailing-batch-items");
+        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 original_key = store.put_multiple(items).unwrap();
+        assert_eq!(original_key.item_count, 3);
+
+        // Keep the three-item payload but make its filename claim that it contains
+        // only two items, simulating a corrupt or otherwise untrusted queue key.
+        let original_path = store.file_path(&original_key);
+        let advertised_key = Key {
+            item_count: 2,
+            ..original_key
+        };
+        let advertised_path = store.file_path(&advertised_key);
+        std::fs::rename(&original_path, &advertised_path).unwrap();
+
+        let err = store.get_multiple(&advertised_key).unwrap_err();
+        assert!(
+            matches!(err, StoreError::Deserialization(_)),
+            "expected Deserialization error, got {err:?}"
+        );
+
+        // Because get_multiple failed, the batch entry remains available for
+        // inspection or recovery instead of being silently discarded.
+        assert!(advertised_path.exists());
+
+        let _ = store.delete();
+    }
+
     #[test]
     fn concurrent_put_raw_respects_entry_limit() {
         let dir = temp_store_dir("concurrent-limit");

scripts/test_security_workflow.py

@@ -837,6 +837,55 @@ emit_step_result() {
             self.assertIn(value, contents)
         self.assertNotIn("OLD RUN EVIDENCE", contents)
 
+    def test_performance_commands_bind_runner_selection_and_preserve_failures(self) -> None:
+        self.prepare("performance")
+        source = self.source.splitlines()
+        job = yaml_block(source, "performance-test", 2)
+        runner = WorkflowSteps()
+        runner.directory = self.directory / "workspace with spaces"
+        scripts = runner.directory / "auto-testing"
+        scripts.mkdir(parents=True)
+        wrapper = scripts / "rustfs_performance_test.sh"
+        wrapper.write_text(f"#!{sys.executable}\nimport json, os, sys\n" +
+                           "print(json.dumps({'args': sys.argv[1:], 'env': {key: os.environ.get(key) for key in " +
+                           "('RUSTFS_BENCH_SCRIPT', 'RUSTFS_WARP_METHODS', 'RUSTFS_WARP_SIZES', " +
+                           "'RUSTFS_WARP_DURATION', 'RUSTFS_WARP_CONCURRENCY', 'WARP_METHODS', " +
+                           "'WARP_SIZES', 'WARP_DURATION', 'WARP_CONCURRENCY')}}))\n" +
+                           "sys.exit(int(os.environ['FAKE_BENCH_EXIT']))\n")
+        wrapper.chmod(0o755)
+        runner.steps = named_steps(job)
+        for methods, sizes, duration, concurrency in (
+            ("get", "1KiB", "1s", "7"), ("all", "all", "5m", "64"), ("", "", "5m", "64")
+        ):
+            runner.context = {"github.workspace": str(runner.directory), "inputs.test_method": methods,
+                              "inputs.object_size": sizes, "inputs.warp_duration || '5m'": duration,
+                              "inputs.warp_concurrency || '64'": concurrency}
+            runner.env = {**self.env, "RUSTFS_BENCH_SCRIPT": "/unverified/home-script.sh",
+                          "RUSTFS_WARP_METHODS": "put", "RUSTFS_WARP_SIZES": "64MiB",
+                          "RUSTFS_WARP_DURATION": "99h", "RUSTFS_WARP_CONCURRENCY": "2",
+                          "WARP_DURATION": "88h", "WARP_CONCURRENCY": "3", "WARP_METHODS": "mixed", "WARP_SIZES": "32MiB",
+                          "LOG_FILE": str(self.directory / "suite.log")}
+            runner.env.update(runner.step_env(job, indent=4))
+            for step, number in (("Run benchmark (GET/PUT/MIXED)", "5"), ("Analyze results", "6")):
+                for code in (0, 42):
+                    with self.subTest(methods=methods, sizes=sizes, step=step, exit=code):
+                        runner.env["FAKE_BENCH_EXIT"] = str(code)
+                        result = runner.run_step(step)
+                        self.assertEqual(result.returncode, code, result.stderr)
+                        invocation = json.loads(result.stdout)
+                        expected = ["--step", number, "-y", "--log-file", runner.env["LOG_FILE"]]
+                        self.assertEqual(invocation["args"], expected)
+                        self.assertEqual(invocation["env"]["RUSTFS_BENCH_SCRIPT"], str(scripts / "rustfs_performance_testing.sh"))
+                        self.assertEqual(invocation["env"]["RUSTFS_WARP_METHODS"], methods)
+                        self.assertEqual(invocation["env"]["RUSTFS_WARP_SIZES"], sizes)
+                        self.assertEqual(invocation["env"]["RUSTFS_WARP_DURATION"], duration)
+                        self.assertEqual(invocation["env"]["RUSTFS_WARP_CONCURRENCY"], concurrency)
+                        if number == "6":
+                            self.assertEqual(invocation["env"]["WARP_METHODS"], methods)
+                            self.assertEqual(invocation["env"]["WARP_SIZES"], sizes)
+                            self.assertEqual(invocation["env"]["WARP_DURATION"], duration)
+                            self.assertEqual(invocation["env"]["WARP_CONCURRENCY"], concurrency)
+
 
 if __name__ == "__main__":
     unittest.main()