Skip to content
This repository was archived by the owner on Jan 27, 2026. It is now read-only.

Commit 4733cbc

Browse files
JannikStJannikSt
authored
improve validator e2e tests (#405)
* improve validator e2e tests
1 parent 260c4d1 commit 4733cbc

1 file changed

Lines changed: 65 additions & 59 deletions

File tree

  • crates/validator/src/validators/synthetic_data

‎crates/validator/src/validators/synthetic_data/mod.rs‎

Lines changed: 65 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -1459,26 +1459,27 @@ mod tests {
14591459
..Default::default()
14601460
};
14611461

1462-
let file_sha = "c257e3d3fe866a00df1285f8bbbe601fed6b85229d983bbbb75e19a068346641";
1463-
let group_id = "3450756714426841564";
1462+
const FILE_SHA: &str = "c257e3d3fe866a00df1285f8bbbe601fed6b85229d983bbbb75e19a068346641";
1463+
const GROUP_ID: &str = "3450756714426841564";
1464+
const NODE_ADDRESS: &str = "0x0000000000000000000000000000000000000000";
14641465

14651466
let mock_storage = MockStorageProvider::new();
14661467
mock_storage.add_file(
1467-
&format!("Qwen/Qwen0.6/dataset/samplingn-{}-1-0-0.parquet", group_id),
1468+
&format!("Qwen/Qwen0.6/dataset/samplingn-{}-1-0-0.parquet", GROUP_ID),
14681469
"file1",
14691470
);
14701471
mock_storage.add_mapping_file(
1471-
file_sha,
1472-
&format!("Qwen/Qwen0.6/dataset/samplingn-{}-1-0-0.parquet", group_id),
1472+
FILE_SHA,
1473+
&format!("Qwen/Qwen0.6/dataset/samplingn-{}-1-0-0.parquet", GROUP_ID),
14731474
);
14741475
server
14751476
.mock(
14761477
"POST",
1477-
"/validategroup/dataset/samplingn-3450756714426841564-1-0.parquet",
1478+
format!("/validategroup/dataset/samplingn-{}-1-0.parquet", GROUP_ID).as_str(),
14781479
)
14791480
.match_body(mockito::Matcher::Json(serde_json::json!({
1480-
"file_shas": [file_sha],
1481-
"group_id": group_id,
1481+
"file_shas": [FILE_SHA],
1482+
"group_id": GROUP_ID,
14821483
"file_number": 0,
14831484
"group_size": 1
14841485
})))
@@ -1488,7 +1489,7 @@ mod tests {
14881489
server
14891490
.mock(
14901491
"GET",
1491-
"/statusgroup/dataset/samplingn-3450756714426841564-1-0.parquet",
1492+
format!("/statusgroup/dataset/samplingn-{}-1-0.parquet", GROUP_ID).as_str(),
14921493
)
14931494
.with_status(200)
14941495
.with_body(r#"{"status": "accept"}"#)
@@ -1515,12 +1516,10 @@ mod tests {
15151516
Some(metrics_context),
15161517
);
15171518

1518-
let work_keys: Vec<String> = vec![file_sha.to_string()];
1519-
let work_keys_2: Vec<String> = work_keys.clone();
1520-
let work_keys_3: Vec<String> = work_keys.clone();
1519+
let work_keys: Vec<String> = vec![FILE_SHA.to_string()];
15211520

15221521
let work_info = WorkInfo {
1523-
node_id: Address::from_str("0x0000000000000000000000000000000000000000").unwrap(),
1522+
node_id: Address::from_str(NODE_ADDRESS).unwrap(),
15241523
..Default::default()
15251524
};
15261525
for work_key in work_keys.clone() {
@@ -1529,29 +1528,29 @@ mod tests {
15291528
.await?;
15301529
}
15311530

1532-
let plan = validator.build_validation_plan(work_keys).await?;
1531+
let plan = validator.build_validation_plan(work_keys.clone()).await?;
15331532
assert_eq!(plan.group_trigger_tasks.len(), 1);
1534-
assert_eq!(plan.group_trigger_tasks[0].group_id, group_id);
1533+
assert_eq!(plan.group_trigger_tasks[0].group_id, GROUP_ID);
15351534
let metrics_0 = export_metrics().unwrap();
15361535
assert!(metrics_0
15371536
.contains("validator_work_keys_to_process{pool_id=\"0\",validator_id=\"0\"} 1"));
15381537

1539-
let group = validator.get_group(file_sha).await?;
1538+
let group = validator.get_group(FILE_SHA).await?;
15401539
assert!(group.is_some());
15411540
let group = group.unwrap();
1542-
assert_eq!(group.group_id, group_id);
1541+
assert_eq!(group.group_id, GROUP_ID);
15431542
assert_eq!(group.group_size, 1);
15441543
assert_eq!(group.file_number, 0);
15451544

15461545
let result = validator.process_group_task(group).await;
15471546
assert!(result.is_ok());
15481547

15491548
let cache_status = validator
1550-
.get_work_validation_status_from_redis(file_sha)
1549+
.get_work_validation_status_from_redis(FILE_SHA)
15511550
.await?;
15521551
assert_eq!(cache_status, Some(ValidationResult::Unknown));
15531552

1554-
let plan_2 = validator.build_validation_plan(work_keys_2).await?;
1553+
let plan_2 = validator.build_validation_plan(work_keys.clone()).await?;
15551554
assert_eq!(plan_2.group_trigger_tasks.len(), 0);
15561555
assert_eq!(plan_2.group_status_check_tasks.len(), 1);
15571556

@@ -1566,23 +1565,26 @@ mod tests {
15661565
assert!(result.is_ok());
15671566

15681567
let cache_status = validator
1569-
.get_work_validation_status_from_redis(file_sha)
1568+
.get_work_validation_status_from_redis(FILE_SHA)
15701569
.await?;
15711570
assert_eq!(cache_status, Some(ValidationResult::Accept));
15721571

1573-
let plan_3 = validator.build_validation_plan(work_keys_3).await?;
1572+
let plan_3 = validator.build_validation_plan(work_keys.clone()).await?;
15741573
assert_eq!(plan_3.group_trigger_tasks.len(), 0);
15751574
assert_eq!(plan_3.group_status_check_tasks.len(), 0);
15761575
let metrics_2 = export_metrics().unwrap();
15771576
assert!(metrics_2
15781577
.contains("validator_work_keys_to_process{pool_id=\"0\",validator_id=\"0\"} 0"));
15791578
assert!(metrics_2.contains("toploc_config_name=\"Qwen/Qwen0.6\""));
1580-
assert!(metrics_2.contains("validator_group_work_units_check_total{group_id=\"3450756714426841564\",pool_id=\"0\",result=\"match\",toploc_config_name=\"Qwen/Qwen0.6\",validator_id=\"0\"} 1"));
1579+
assert!(metrics_2.contains(&format!("validator_group_work_units_check_total{{group_id=\"{}\",pool_id=\"0\",result=\"match\",toploc_config_name=\"Qwen/Qwen0.6\",validator_id=\"0\"}} 1", GROUP_ID)));
15811580

15821581
Ok(())
15831582
}
15841583

15851584
#[tokio::test]
1585+
// This test accepts a groups submission via the toploc server which returns a total output flops count
1586+
// One of the nodes claims more work units than the output_flops_count / number of nodes
1587+
// The validator should reject the node and accept the other node
15861588
async fn test_group_e2e_work_unit_mismatch() -> Result<(), Error> {
15871589
let mut server = Server::new_async().await;
15881590
let (store, contracts) = setup_test_env()?;
@@ -1593,35 +1595,39 @@ mod tests {
15931595
..Default::default()
15941596
};
15951597

1596-
let file_sha_1 = "c257e3d3fe866a00df1285f8bbbe601fed6b85229d983bbbb75e19a068346641";
1597-
let file_sha_2 = "88e4672c19e5a10bff2e23d223f8bfc38ae1425feaa18db9480e631a4fd98edf";
1598-
let group_id = "3450756714426841564";
1598+
const HONEST_NODE_ADDRESS: &str = "0x0000000000000000000000000000000000000001";
1599+
const HONEST_FILE_SHA: &str =
1600+
"c257e3d3fe866a00df1285f8bbbe601fed6b85229d983bbbb75e19a068346641";
1601+
const EXCESSIVE_FILE_SHA: &str =
1602+
"88e4672c19e5a10bff2e23d223f8bfc38ae1425feaa18db9480e631a4fd98edf";
1603+
const EXCESSIVE_NODE_ADDRESS: &str = "0x0000000000000000000000000000000000000002";
1604+
const GROUP_ID: &str = "3456714426841564";
15991605

16001606
let mock_storage = MockStorageProvider::new();
16011607
mock_storage.add_file(
1602-
&format!("Qwen/Qwen0.6/dataset/samplingn-{}-2-0-0.parquet", group_id),
1608+
&format!("Qwen/Qwen0.6/dataset/samplingn-{}-2-0-0.parquet", GROUP_ID),
16031609
"file1",
16041610
);
16051611
mock_storage.add_file(
1606-
&format!("Qwen/Qwen0.6/dataset/samplingn-{}-2-0-1.parquet", group_id),
1612+
&format!("Qwen/Qwen0.6/dataset/samplingn-{}-2-0-1.parquet", GROUP_ID),
16071613
"file2",
16081614
);
16091615
mock_storage.add_mapping_file(
1610-
file_sha_1,
1611-
&format!("Qwen/Qwen0.6/dataset/samplingn-{}-2-0-0.parquet", group_id),
1616+
HONEST_FILE_SHA,
1617+
&format!("Qwen/Qwen0.6/dataset/samplingn-{}-2-0-0.parquet", GROUP_ID),
16121618
);
16131619
mock_storage.add_mapping_file(
1614-
file_sha_2,
1615-
&format!("Qwen/Qwen0.6/dataset/samplingn-{}-2-0-1.parquet", group_id),
1620+
EXCESSIVE_FILE_SHA,
1621+
&format!("Qwen/Qwen0.6/dataset/samplingn-{}-2-0-1.parquet", GROUP_ID),
16161622
);
16171623
server
16181624
.mock(
16191625
"POST",
1620-
"/validategroup/dataset/samplingn-3450756714426841564-2-0.parquet",
1626+
format!("/validategroup/dataset/samplingn-{}-2-0.parquet", GROUP_ID).as_str(),
16211627
)
16221628
.match_body(mockito::Matcher::Json(serde_json::json!({
1623-
"file_shas": [file_sha_1, file_sha_2],
1624-
"group_id": group_id,
1629+
"file_shas": [HONEST_FILE_SHA, EXCESSIVE_FILE_SHA],
1630+
"group_id": GROUP_ID,
16251631
"file_number": 0,
16261632
"group_size": 2
16271633
})))
@@ -1631,7 +1637,7 @@ mod tests {
16311637
server
16321638
.mock(
16331639
"GET",
1634-
"/statusgroup/dataset/samplingn-3450756714426841564-2-0.parquet",
1640+
format!("/statusgroup/dataset/samplingn-{}-2-0.parquet", GROUP_ID).as_str(),
16351641
)
16361642
.with_status(200)
16371643
.with_body(r#"{"status": "accept", "input_flops": 1, "output_flops": 2000}"#)
@@ -1658,57 +1664,58 @@ mod tests {
16581664
Some(metrics_context),
16591665
);
16601666

1661-
let work_keys: Vec<String> = vec![file_sha_1.to_string(), file_sha_2.to_string()];
1662-
let work_keys_2: Vec<String> = work_keys.clone();
1663-
let work_keys_3: Vec<String> = work_keys.clone();
1667+
let work_keys: Vec<String> =
1668+
vec![HONEST_FILE_SHA.to_string(), EXCESSIVE_FILE_SHA.to_string()];
16641669

16651670
// Node 1 claims exact amount (1000 work units)
1671+
const EXPECTED_WORK_UNITS: u64 = 1000;
1672+
const EXCESSIVE_WORK_UNITS: u64 = 1500;
16661673
let work_info_1 = WorkInfo {
1667-
node_id: Address::from_str("0x0000000000000000000000000000000000000001").unwrap(),
1668-
work_units: U256::from(1000),
1674+
node_id: Address::from_str(HONEST_NODE_ADDRESS).unwrap(),
1675+
work_units: U256::from(EXPECTED_WORK_UNITS),
16691676
..Default::default()
16701677
};
16711678
// Node 2 claims too much (1500 work units instead of 1000)
16721679
let work_info_2 = WorkInfo {
1673-
node_id: Address::from_str("0x0000000000000000000000000000000000000002").unwrap(),
1674-
work_units: U256::from(1500),
1680+
node_id: Address::from_str(EXCESSIVE_NODE_ADDRESS).unwrap(),
1681+
work_units: U256::from(EXCESSIVE_WORK_UNITS),
16751682
..Default::default()
16761683
};
16771684

16781685
validator
1679-
.update_work_info_in_redis(file_sha_1, &work_info_1)
1686+
.update_work_info_in_redis(HONEST_FILE_SHA, &work_info_1)
16801687
.await?;
16811688
validator
1682-
.update_work_info_in_redis(file_sha_2, &work_info_2)
1689+
.update_work_info_in_redis(EXCESSIVE_FILE_SHA, &work_info_2)
16831690
.await?;
16841691

1685-
let plan = validator.build_validation_plan(work_keys).await?;
1692+
let plan = validator.build_validation_plan(work_keys.clone()).await?;
16861693
assert_eq!(plan.group_trigger_tasks.len(), 1);
1687-
assert_eq!(plan.group_trigger_tasks[0].group_id, group_id);
1694+
assert_eq!(plan.group_trigger_tasks[0].group_id, GROUP_ID);
16881695
let metrics_0 = export_metrics().unwrap();
16891696
assert!(metrics_0
16901697
.contains("validator_work_keys_to_process{pool_id=\"0\",validator_id=\"0\"} 2"));
16911698

1692-
let group = validator.get_group(file_sha_1).await?;
1699+
let group = validator.get_group(HONEST_FILE_SHA).await?;
16931700
assert!(group.is_some());
16941701
let group = group.unwrap();
1695-
assert_eq!(group.group_id, group_id);
1702+
assert_eq!(group.group_id, GROUP_ID);
16961703
assert_eq!(group.group_size, 2);
16971704
assert_eq!(group.file_number, 0);
16981705

16991706
let result = validator.process_group_task(group).await;
17001707
assert!(result.is_ok());
17011708

17021709
let cache_status_1 = validator
1703-
.get_work_validation_status_from_redis(file_sha_1)
1710+
.get_work_validation_status_from_redis(HONEST_FILE_SHA)
17041711
.await?;
17051712
assert_eq!(cache_status_1, Some(ValidationResult::Unknown));
17061713
let cache_status_2 = validator
1707-
.get_work_validation_status_from_redis(file_sha_2)
1714+
.get_work_validation_status_from_redis(EXCESSIVE_FILE_SHA)
17081715
.await?;
17091716
assert_eq!(cache_status_2, Some(ValidationResult::Unknown));
17101717

1711-
let plan_2 = validator.build_validation_plan(work_keys_2).await?;
1718+
let plan_2 = validator.build_validation_plan(work_keys.clone()).await?;
17121719
assert_eq!(plan_2.group_trigger_tasks.len(), 0);
17131720
assert_eq!(plan_2.group_status_check_tasks.len(), 1);
17141721

@@ -1720,31 +1727,30 @@ mod tests {
17201727
let result = validator
17211728
.process_group_status_check(plan_2.group_status_check_tasks[0].clone())
17221729
.await;
1723-
println!("result: {:?}", result);
17241730
assert!(result.is_ok());
17251731

17261732
// Node 1 should be accepted (exact claim)
17271733
let cache_status_1 = validator
1728-
.get_work_validation_status_from_redis(file_sha_1)
1734+
.get_work_validation_status_from_redis(HONEST_FILE_SHA)
17291735
.await?;
17301736
assert_eq!(cache_status_1, Some(ValidationResult::Accept));
17311737

1732-
// Node 2 should be rejected (wrong claim)
1738+
// Node 2 should be rejected (excessive claim)
17331739
let cache_status_2 = validator
1734-
.get_work_validation_status_from_redis(file_sha_2)
1740+
.get_work_validation_status_from_redis(EXCESSIVE_FILE_SHA)
17351741
.await?;
17361742
assert_eq!(cache_status_2, Some(ValidationResult::Reject));
17371743

1738-
let plan_3 = validator.build_validation_plan(work_keys_3).await?;
1744+
// Make sure there is no more work to process
1745+
let plan_3 = validator.build_validation_plan(work_keys.clone()).await?;
17391746
assert_eq!(plan_3.group_trigger_tasks.len(), 0);
17401747
assert_eq!(plan_3.group_status_check_tasks.len(), 0);
17411748
let metrics_2 = export_metrics().unwrap();
1742-
println!("metrics_2: {}", metrics_2);
1749+
assert!(metrics_2.contains(&format!("validator_group_validations_total{{group_id=\"{}\",pool_id=\"0\",result=\"accept\",toploc_config_name=\"Qwen/Qwen0.6\",validator_id=\"0\"}} 1", GROUP_ID)));
17431750
assert!(metrics_2
17441751
.contains("validator_work_keys_to_process{pool_id=\"0\",validator_id=\"0\"} 0"));
17451752
assert!(metrics_2.contains("toploc_config_name=\"Qwen/Qwen0.6\""));
1746-
assert!(metrics_2.contains("validator_group_work_units_check_total{group_id=\"3450756714426841564\",pool_id=\"0\",result=\"mismatch\",toploc_config_name=\"Qwen/Qwen0.6\",validator_id=\"0\"} 1"));
1747-
assert!(metrics_2.contains("validator_group_work_units_check_total{group_id=\"3450756714426841564\",pool_id=\"0\",result=\"match\",toploc_config_name=\"Qwen/Qwen0.6\",validator_id=\"0\"} 1"));
1753+
assert!(metrics_2.contains(&format!("validator_group_work_units_check_total{{group_id=\"{}\",pool_id=\"0\",result=\"mismatch\",toploc_config_name=\"Qwen/Qwen0.6\",validator_id=\"0\"}} 1", GROUP_ID)));
17481754

17491755
Ok(())
17501756
}

0 commit comments

Comments
 (0)