1. 1//! Concurrent edits to `.vak/feeds.toml` all land, and every reader sees a
2. 2//! whole document: each edit is a read-modify-write held under one lock and
3. 3//! published through a temporary file no other write shares.
4. 4
5. 5#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
6. 6
7. 7use std::path::Path;
8. 8
9. 9use vak_core::Core;
10. 10
11. 11fn source_ids(path: &Path) -> Vec<String> {
12. 12 let text = std::fs::read_to_string(path).unwrap();
13. 13 let document: toml::Table = toml::from_str(&text)
14. 14 .unwrap_or_else(|error| panic!("feeds.toml does not parse: {error}\n{text}"));
15. 15 document
16. 16 .get("sources")
17. 17 .and_then(toml::Value::as_array)
18. 18 .map(|sources| {
19. 19 sources
20. 20 .iter()
21. 21 .filter_map(|source| source.get("id").and_then(toml::Value::as_str))
22. 22 .map(ToOwned::to_owned)
23. 23 .collect()
24. 24 })
25. 25 .unwrap_or_default()
26. 26}
27. 27
28. 28#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
29. 29async fn concurrent_feed_source_adds_all_land() {
30. 30 let dir = tempfile::tempdir().unwrap();
31. 31 let cwd = dir.path().join("ws");
32. 32 std::fs::create_dir_all(cwd.join(".vak")).unwrap();
33. 33 let config = cwd.join(".vak").join("feeds.toml");
34. 34 // A key no feed handler writes; every edit must carry it through.
35. 35 std::fs::write(&config, "# kept\n[general]\nowner_note = \"mine\"\n").unwrap();
36. 36
37. 37 vak_config::paths::isolate_home_for_tests();
38. 38 let core = Core::new_with_trust(cwd.clone(), true).expect("core");
39. 39 core.set_sessions_home(dir.path().join("home"));
40. 40
41. 41 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
42. 42 let addr = listener.local_addr().unwrap();
43. 43 let app = vak_server::router(core);
44. 44 tokio::spawn(async move {
45. 45 axum::serve(listener, app).await.unwrap();
46. 46 });
47. 47 let base = format!("http://{addr}");
48. 48 let client = reqwest::Client::new();
49. 49
50. 50 const ROUNDS: usize = 20;
51. 51 const WRITERS: usize = 8;
52. 52 let mut expected = Vec::new();
53. 53 for round in 0..ROUNDS {
54. 54 let mut requests = Vec::new();
55. 55 for writer in 0..WRITERS {
56. 56 let id = format!("src-{round}-{writer}");
57. 57 expected.push(id.clone());
58. 58 let client = client.clone();
59. 59 let url = format!("{base}/feeds/sources");
60. 60 requests.push(tokio::spawn(async move {
61. 61 client
62. 62 .post(url)
63. 63 .json(&serde_json::json!({
64. 64 "id": id,
65. 65 "name": format!("Feed {id}"),
66. 66 "type": "rss",
67. 67 "url": "https://example.com/feed.xml",
68. 68 }))
69. 69 .send()
70. 70 .await
71. 71 .unwrap()
72. 72 .status()
73. 73 }));
74. 74 }
75. 75 for request in requests {
76. 76 assert_eq!(request.await.unwrap(), 200);
77. 77 }
78. 78 let ids = source_ids(&config);
79. 79 for id in &expected {
80. 80 assert!(ids.contains(id), "round {round}: {id} was lost");
81. 81 }
82. 82 assert_eq!(ids.len(), expected.len(), "round {round}: {ids:?}");
83. 83 }
84. 84
85. 85 let text = std::fs::read_to_string(&config).unwrap();
86. 86 assert!(text.contains("owner_note = \"mine\""));
87. 87 let strays: Vec<_> = std::fs::read_dir(cwd.join(".vak"))
88. 88 .unwrap()
89. 89 .filter_map(Result::ok)
90. 90 .map(|entry| entry.file_name().to_string_lossy().into_owned())
91. 91 .filter(|name| name.ends_with(".tmp"))
92. 92 .collect();
93. 93 assert!(strays.is_empty(), "temporary files left behind: {strays:?}");
94. 94}
95. 95