diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index 9f91c7e20..df8f8d076 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -3074,6 +3074,41 @@ mod tests { ); } + #[test] + fn compat_json_restores_json_when_either_msgpack_only_gate_is_removed() { + with_internode_msgpack_env( + [ + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")), + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, Some("true")), + ], + || { + let resp = sample_read_multiple_resp("file", b"data"); + let json = compat_json(&resp).expect("compat_json should encode"); + + assert!(json.is_empty(), "both gates should enter msgpack-only send mode"); + }, + ); + + for vars in [ + [ + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")), + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, Some("false")), + ], + [ + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("false")), + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, Some("true")), + ], + ] { + with_internode_msgpack_env(vars, || { + let resp = sample_read_multiple_resp("file", b"data"); + let json = compat_json(&resp).expect("compat_json should encode"); + + assert!(!json.is_empty(), "removing either gate should restore old-peer JSON compatibility"); + assert_eq!(json, serde_json::to_string(&resp).expect("json should encode")); + }); + } + } + #[test] fn read_multiple_response_decode_reports_corrupt_msgpack_item() { let endpoint = sample_remote_endpoint(); diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index cd1ef509f..b3515a608 100644 --- a/crates/protos/src/lib.rs +++ b/crates/protos/src/lib.rs @@ -1193,6 +1193,36 @@ mod tests { } } + #[test] + fn request_compat_send_site_manifest_pins_exact_json_policies() { + let mut policies = REQUEST_COMPAT_SEND_SITES + .iter() + .map(|send_site| (send_site.field.message, send_site.field.json_field, send_site.policy)) + .collect::>(); + policies.sort_by_key(|(message, json_field, _)| (*message, *json_field)); + + assert_eq!( + policies, + [ + ( + "BatchReadVersionRequest", + "batch_read_version_req", + RequestJsonPolicy::MsgpackOnlyEligible + ), + ("DeleteVersionRequest", "file_info", RequestJsonPolicy::AlwaysDualWriteUntilFallbackZero,), + ("DeleteVersionRequest", "opts", RequestJsonPolicy::AlwaysDualWriteUntilFallbackZero), + ("DeleteVersionsRequest", "opts", RequestJsonPolicy::AlwaysDualWriteUntilFallbackZero,), + ("DeleteVersionsRequest", "versions", RequestJsonPolicy::AlwaysDualWriteUntilFallbackZero,), + ("ReadMultipleRequest", "read_multiple_req", RequestJsonPolicy::MsgpackOnlyEligible), + ("ReadVersionRequest", "opts", RequestJsonPolicy::MsgpackOnlyEligible), + ("RenameDataRequest", "file_info", RequestJsonPolicy::MsgpackOnlyEligible), + ("UpdateMetadataRequest", "file_info", RequestJsonPolicy::MsgpackOnlyEligible), + ("UpdateMetadataRequest", "opts", RequestJsonPolicy::MsgpackOnlyEligible), + ("WriteMetadataRequest", "file_info", RequestJsonPolicy::MsgpackOnlyEligible), + ] + ); + } + #[test] fn response_compat_send_site_manifest_covers_node_proto_bin_fields() { let mut manifest_fields = RESPONSE_COMPAT_SEND_SITES @@ -1294,6 +1324,31 @@ mod tests { reset_internode_rpc_msgpack_only_cache(); } + #[test] + fn internode_rpc_msgpack_only_requires_request_and_fleet_confirmation() { + for (requested, fleet_confirmed, expected) in [ + (None, None, false), + (Some("true"), None, false), + (None, Some("true"), false), + (Some("true"), Some("false"), false), + (Some("false"), Some("true"), false), + (Some("true"), Some("true"), true), + ] { + reset_internode_rpc_msgpack_only_cache(); + temp_env::with_vars( + [ + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, requested), + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, fleet_confirmed), + ], + || { + assert_eq!(internode_rpc_msgpack_only(), expected); + }, + ); + } + + reset_internode_rpc_msgpack_only_cache(); + } + #[tokio::test] async fn get_channel_for_class_bulk_reuses_control_cache_when_isolation_disabled() { // Isolation defaults off, so a Bulk request must reuse the control channel keyed by the diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index e2bc29a8f..44990045a 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/rustfs/src/storage/rpc/node_service/disk.rs @@ -1263,6 +1263,44 @@ mod tests { ); } + #[test] + fn compat_response_json_restores_json_when_either_msgpack_only_gate_is_removed() { + let payload = SamplePayload { + name: "rollback".to_string(), + count: 12, + }; + + with_internode_msgpack_env( + [ + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")), + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, Some("true")), + ], + || { + let json = compat_response_json(&payload).expect("compat response json should encode"); + + assert!(json.is_empty(), "both gates should enter msgpack-only response mode"); + }, + ); + + for vars in [ + [ + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")), + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, Some("false")), + ], + [ + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("false")), + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, Some("true")), + ], + ] { + with_internode_msgpack_env(vars, || { + let json = compat_response_json(&payload).expect("compat response json should encode"); + + assert!(!json.is_empty(), "removing either gate should restore old-peer JSON compatibility"); + assert_eq!(json, serde_json::to_string(&payload).expect("json should encode")); + }); + } + } + #[test] fn decode_msgpack_or_json_fails_closed_on_corrupt_non_empty_msgpack() { let err = decode_msgpack_or_json::(b"not-msgpack", r#"{"name":"json","count":1}"#, "SamplePayload") @@ -1330,6 +1368,52 @@ mod tests { assert_eq!(msgpack_decoded.data, responses[0].data); } + #[test] + fn encode_read_multiple_response_payloads_respects_msgpack_only_gate_and_rollback() { + let responses = vec![ReadMultipleResp { + bucket: "bucket".to_string(), + prefix: "prefix".to_string(), + file: "gate".to_string(), + exists: true, + data: b"payload".to_vec(), + ..Default::default() + }]; + + with_internode_msgpack_env( + [ + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")), + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, Some("true")), + ], + || { + let (json_payloads, msgpack_payloads) = + encode_read_multiple_response_payloads(&responses).expect("read multiple responses should encode"); + + assert_eq!(json_payloads, vec![String::new()]); + assert_eq!(msgpack_payloads.len(), responses.len()); + let decoded = decode_msgpack_or_json::(&msgpack_payloads[0], "", "ReadMultipleResp") + .expect("msgpack read multiple response should decode"); + assert_eq!(decoded.file, responses[0].file); + }, + ); + + with_internode_msgpack_env( + [ + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")), + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, Some("false")), + ], + || { + let (json_payloads, msgpack_payloads) = + encode_read_multiple_response_payloads(&responses).expect("read multiple responses should encode"); + + assert!(!json_payloads[0].is_empty(), "rollback should restore response JSON"); + assert_eq!(msgpack_payloads.len(), responses.len()); + let json_decoded: ReadMultipleResp = + serde_json::from_str(&json_payloads[0]).expect("json read multiple response should decode"); + assert_eq!(json_decoded.file, responses[0].file); + }, + ); + } + #[test] fn encode_batch_read_version_response_payloads_keeps_json_and_msgpack_in_sync() { let responses = vec![BatchReadVersionResp { @@ -1356,4 +1440,49 @@ mod tests { assert_eq!(msgpack_decoded.path, responses[0].path); assert_eq!(msgpack_decoded.error, responses[0].error); } + + #[test] + fn encode_batch_read_version_response_payloads_respects_msgpack_only_gate_and_rollback() { + let responses = vec![BatchReadVersionResp { + index: 4, + path: "object-b".to_string(), + version_id: "version-b".to_string(), + success: true, + ..Default::default() + }]; + + with_internode_msgpack_env( + [ + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")), + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, Some("true")), + ], + || { + let (json_payloads, msgpack_payloads) = + encode_batch_read_version_response_payloads(&responses).expect("batch read version responses should encode"); + + assert_eq!(json_payloads, vec![String::new()]); + assert_eq!(msgpack_payloads.len(), responses.len()); + let decoded = decode_msgpack_or_json::(&msgpack_payloads[0], "", "BatchReadVersionResp") + .expect("msgpack batch read version response should decode"); + assert_eq!(decoded.path, responses[0].path); + }, + ); + + with_internode_msgpack_env( + [ + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("false")), + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, Some("true")), + ], + || { + let (json_payloads, msgpack_payloads) = + encode_batch_read_version_response_payloads(&responses).expect("batch read version responses should encode"); + + assert!(!json_payloads[0].is_empty(), "rollback should restore response JSON"); + assert_eq!(msgpack_payloads.len(), responses.len()); + let json_decoded: BatchReadVersionResp = + serde_json::from_str(&json_payloads[0]).expect("json batch read version response should decode"); + assert_eq!(json_decoded.path, responses[0].path); + }, + ); + } }