diff --git a/.github/workflows/rustfs-performance-test.yml b/.github/workflows/rustfs-performance-test.yml index d698b27e0..699b583cf 100644 --- a/.github/workflows/rustfs-performance-test.yml +++ b/.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 diff --git a/crates/targets/src/store.rs b/crates/targets/src/store.rs index 4a19295ae..0a98524c5 100644 --- a/crates/targets/src/store.rs +++ b/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::::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"); diff --git a/scripts/test_security_workflow.py b/scripts/test_security_workflow.py index ba483571f..e9c70f44c 100644 --- a/scripts/test_security_workflow.py +++ b/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()