gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-06 05:17:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(internode): pin msgpack gate rollback coverage (#5203)

Browse Source 
Add test coverage for the msgpack-only request/fleet confirmation truth table and exact request JSON policy manifest.

Pin request and response rollback behavior so removing either gate restores JSON compatibility while both gates enter msgpack-only for eligible fields.

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-07-24 23:16:23 +08:00
committed by GitHub
parent cda443bd81
commit 187a060919
3 changed files with 219 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/cluster/rpc/remote_disk.rs
+35
View File
@@ -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();
crates/protos/src/lib.rs
+55
View File
@@ -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::<Vec<_>>();
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
rustfs/src/storage/rpc/node_service/disk.rs
+129
View File
@@ -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::<SamplePayload>(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::<ReadMultipleResp>(&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::<BatchReadVersionResp>(&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);
},
);
}
}