Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions src/agent_ws.rs
Original file line number Diff line number Diff line change
Expand Up @@ -816,14 +816,17 @@ fn ping_tasks_message(app: &App, node_id: i64) -> String {
json!({"jsonrpc": "2.0", "method": "ping.tasks", "params": tasks}).to_string()
}

/// Pushes the current probe list to every connected agent, so a panel edit takes
/// effect without waiting for a reconnect.
pub fn push_ping_tasks(app: &App) {
/// Pushes the current probe list to the connected agents, so a panel edit takes
/// effect without waiting for a reconnect. `only` limits it to one node, for an
/// edit that changed that node alone: every push is a query and a message per
/// agent.
pub fn push_ping_tasks(app: &App, only: Option<i64>) {
let connected: Vec<(i64, mpsc::Sender<String>)> = app
.agents
.read()
.unwrap_or_else(|e| e.into_inner())
.iter()
.filter(|(id, _)| only.is_none_or(|only| only == **id))
.map(|(id, agent)| (*id, agent.tx.clone()))
.collect();
for (node_id, sender) in connected {
Expand Down
29 changes: 27 additions & 2 deletions src/api.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1189,17 +1189,42 @@ pub async fn save_ping_task(_: Admin, State(app): State<Shared>, Json(mut task):
}
match app.db.save_ping_task(&task) {
Ok(id) => {
agent_ws::push_ping_tasks(&app);
agent_ws::push_ping_tasks(&app, None);
Json(json!({"id": id})).into_response()
}
Err(e) => fail(e),
}
}

/// The probes ticked for one node, and `base`, those it had when the editor
/// opened; see `Db::set_node_ping_tasks`.
#[derive(Deserialize)]
pub struct NodeProbes {
tasks: Vec<i64>,
base: Vec<i64>,
}

pub async fn set_node_ping_tasks(
_: Admin,
State(app): State<Shared>,
Path(id): Path<i64>,
body: Result<Json<NodeProbes>, JsonRejection>,
) -> Response {
let Ok(Json(probes)) = body else { return bad("监控列表格式不对") };
match app.db.set_node_ping_tasks(id, &probes.tasks, &probes.base) {
Ok(true) => {
agent_ws::push_ping_tasks(&app, Some(id));
Json(json!({"ok": true})).into_response()
}
Ok(false) => no_such_node(),
Err(e) => fail(e),
}
}

pub async fn delete_ping_task(_: Admin, State(app): State<Shared>, Path(id): Path<i64>) -> Response {
match app.db.delete_ping_task(id) {
Ok(()) => {
agent_ws::push_ping_tasks(&app);
agent_ws::push_ping_tasks(&app, None);
Json(json!({"ok": true})).into_response()
}
Err(e) => fail(e),
Expand Down
120 changes: 106 additions & 14 deletions src/db.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1943,17 +1943,33 @@ impl Db {
_ => e.into(),
})?;
}
// Queried from the table after the rows are in rather than counted from
// the request: an update changes this task's own assignments, so
// arithmetic on the way in would have to subtract them again. The
// transaction makes this atomic with the write, and bailing here rolls it
// back.
// By name: the panel identifies nodes by name and never shows an id.
Self::refuse_crowded(&tx, None)?;
// The next node created receives every auto-joining probe at once.
let joining: i64 =
tx.query_row("SELECT COUNT(*) FROM ping_task WHERE auto_join", [], |r| r.get(0))?;
if joining > Self::MAX_PROBES_PER_NODE {
refuse!("新节点会自动加入 {joining} 个探测任务,agent 最多只跑 {} 个", Self::MAX_PROBES_PER_NODE);
}
tx.commit()?;
Ok(id)
}

/// Refuses a write that leaves some node over [`Self::MAX_PROBES_PER_NODE`],
/// or `node` alone when the write touched only that node's rows: another
/// node already over the limit is not this edit's to fix.
///
/// Queried from the table after the rows are in rather than counted from the
/// request: an update changes the caller's own assignments, so arithmetic on
/// the way in would have to subtract them again. Run inside the write's
/// transaction, so refusing rolls it back. By name: the panel identifies
/// nodes by name and never shows an id.
fn refuse_crowded(tx: &rusqlite::Transaction, node: Option<i64>) -> Result<()> {
let crowded: Option<String> = tx
.query_row(
"SELECT n.name FROM ping_node p JOIN node n ON n.id = p.node_id
WHERE ?2 IS NULL OR p.node_id = ?2
GROUP BY p.node_id HAVING COUNT(*) > ?1 LIMIT 1",
[Self::MAX_PROBES_PER_NODE],
params![Self::MAX_PROBES_PER_NODE, node],
|r| r.get(0),
)
.optional()?;
Expand All @@ -1963,14 +1979,38 @@ impl Db {
Self::MAX_PROBES_PER_NODE
);
}
// The next node created receives every auto-joining probe at once.
let joining: i64 =
tx.query_row("SELECT COUNT(*) FROM ping_task WHERE auto_join", [], |r| r.get(0))?;
if joining > Self::MAX_PROBES_PER_NODE {
refuse!("新节点会自动加入 {joining} 个探测任务,agent 最多只跑 {} 个", Self::MAX_PROBES_PER_NODE);
Ok(())
}

/// One node's probes, edited from the node's side: `save_ping_task` with
/// the roles swapped. Only the change from `base` is applied, as there, so a
/// probe assigned to this node elsewhere while the editor was open keeps the
/// assignment. `false` when the node does not exist.
pub fn set_node_ping_tasks(&self, node: i64, tasks: &[i64], base: &[i64]) -> Result<bool> {
let mut conn = self.conn();
let tx = conn.transaction()?;
if tx.query_row("SELECT 1 FROM node WHERE id=?1", [node], |_| Ok(())).optional()?.is_none() {
return Ok(false);
}
for task in base.iter().filter(|t| !tasks.contains(t)) {
tx.execute("DELETE FROM ping_node WHERE task_id=?1 AND node_id=?2", params![task, node])?;
}
for task in tasks.iter().filter(|t| !base.contains(t)) {
// The node was checked above, so a key failing here is the probe.
tx.execute(
"INSERT OR IGNORE INTO ping_node (task_id, node_id) VALUES (?1,?2)",
params![task, node],
)
.map_err(|e| match e.sqlite_error_code() {
Some(rusqlite::ErrorCode::ConstraintViolation) => {
anyhow::Error::from(e).context(crate::Shown("有监控已被删除,刷新后重试".into()))
}
_ => e.into(),
})?;
}
Self::refuse_crowded(&tx, Some(node))?;
tx.commit()?;
Ok(id)
Ok(true)
}

/// Deletes a probe and the results filed under it.
Expand All @@ -1997,7 +2037,7 @@ impl Db {
/// Ordered, because the agent keeps the first [`Self::MAX_PROBES_PER_NODE`]
/// as its backstop against a hub requesting hundreds. Unordered, a list at
/// that boundary could yield a different subset on each push, restarting half
/// the timers each time; `save_ping_task` prevents reaching that boundary,
/// the timers each time; `refuse_crowded` prevents reaching that boundary,
/// and this makes the backstop deterministic should a database arrive there
/// by another route.
///
Expand Down Expand Up @@ -3985,6 +4025,58 @@ mod tests {
assert!(db.ping_tasks().unwrap().is_empty());
}

/// The node-side editor sends its ticks and what it opened with; only the
/// difference is written.
#[test]
fn a_node_edit_applies_only_the_probes_it_changed() {
let db = db();
let (a, other) = (node(&db, 1), node(&db, 1));
let probe = |db: &Db, nodes| {
db.save_ping_task(&PingTask {
name: "p".into(),
target: "1.1.1.1:443".into(),
interval: 60,
nodes,
..Default::default()
})
.unwrap()
};
let (kept, dropped, added, elsewhere) =
(probe(&db, vec![a]), probe(&db, vec![a, other]), probe(&db, vec![]), probe(&db, vec![]));
// Assigned from the probe's side while this node's editor was open.
db.save_ping_task(&PingTask {
id: elsewhere,
name: "p".into(),
target: "1.1.1.1:443".into(),
interval: 60,
nodes: vec![a],
base: Some(vec![]),
..Default::default()
})
.unwrap();

assert!(db.set_node_ping_tasks(a, &[kept, added], &[kept, dropped]).unwrap());
let ids =
|n| db.ping_tasks_for(n).unwrap().iter().map(|t| t["id"].as_i64().unwrap()).collect::<Vec<_>>();
assert_eq!(ids(a), [kept, added, elsewhere]);
assert_eq!(ids(other), [dropped], "an assignment on another node is unchanged");

assert!(!db.set_node_ping_tasks(a + 100, &[kept], &[]).unwrap());
db.delete_ping_task(added).unwrap();
assert!(db.set_node_ping_tasks(other, &[added], &[]).is_err(), "a deleted probe is refused");
let many: Vec<i64> = (0..Db::MAX_PROBES_PER_NODE).map(|_| probe(&db, vec![])).collect();
assert!(db.set_node_ping_tasks(other, &many, &[]).is_err(), "65 probes on one node");
assert_eq!(ids(other), [dropped], "a refusal writes nothing");

// Over the limit by another route: editing a different node still saves.
for task in &many {
db.conn()
.execute("INSERT INTO ping_node (task_id, node_id) VALUES (?1,?2)", params![task, other])
.unwrap();
}
assert!(db.set_node_ping_tasks(a, &[kept], &[kept, added]).unwrap());
}

/// The agent caps the probe list it will run and drops the remainder with
/// nothing but a line in its own journal. The hub knows the total, so the hub
/// issues the refusal; otherwise the panel lists probes that never ran and
Expand Down
1 change: 1 addition & 0 deletions src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -504,6 +504,7 @@ async fn main() -> Result<()> {
.route("/api/nodes/{id}", put(api::update_node).delete(api::delete_node))
.route("/api/nodes/{id}/token", post(api::reset_token))
.route("/api/nodes/{id}/traffic", put(api::patch_traffic))
.route("/api/nodes/{id}/ping-tasks", put(api::set_node_ping_tasks))
.route("/api/ping-tasks", get(api::ping_tasks).post(api::save_ping_task))
.route("/api/ping-tasks/order", put(api::reorder_ping_tasks))
.route("/api/ping-tasks/{id}", delete(api::delete_ping_task))
Expand Down
Loading
Loading