From 5dfb5d1c6c5e70e18b05bbbd9569afefa3560a99 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E6=80=A7=E9=A9=8A?= Date: Thu, 5 Nov 2026 15:48:22 +0800 Subject: [PATCH] Routines follow the clock in each timezone; channels resume where they stopped; pages find checks the bot --- src/channels.rs | 45 ++++++++++++++++++++++++++++++++------------- src/pages.rs | 4 +++- src/routines.rs | 48 ++++++++++++++++++++++++++++++++++++++++++------ src/triggers.rs | 42 +++++++++++++++++++++++++++++++++++------- src/voice.rs | 2 +- 5 files changed, 113 insertions(+), 28 deletions(-) diff --git a/src/channels.rs b/src/channels.rs index d133a48..f78a594 100644 --- a/src/channels.rs +++ b/src/channels.rs @@ -143,6 +143,12 @@ fn agent(timeout: u64) -> ureq::Agent { ureq::Agent::config_builder().timeout_global(Some(Duration::from_secs(timeout))).http_status_as_error(false).build().into() } +/// Seconds to wait after a 429 ("rate:N" from get_json), bounded so a bogus value can't park +/// the channel forever. +fn rate_secs(e: &str) -> u64 { + e.split(':').nth(1).and_then(|s| s.parse().ok()).unwrap_or(10).clamp(1, 300) +} + fn get_json(url: &str, auth: Option<&str>, timeout: u64) -> Result { let mut r = agent(timeout).get(url); if let Some(a) = auth { @@ -152,7 +158,10 @@ fn get_json(url: &str, auth: Option<&str>, timeout: u64) -> Result(&text).ok().and_then(|v| v["parameters"]["retry_after"].as_u64()).unwrap_or(10); + return Err(format!("rate:{ra}")); } if !(200..300).contains(&status) { return Err(format!("HTTP {status}：{}", text.chars().take(160).collect::())); @@ -457,8 +466,9 @@ fn poll(c: Channel, gen: u64, tx: Sender, repaint: Arc { if cursor.is_empty() { - // Start from now: older messages aren't answered. - cursor = v["messages"][0]["ts"].as_str().unwrap_or("0").to_string(); + // Start from now: older messages aren't answered. (An empty channel + // has no ts to stand on — "0" made the next round replay history.) + cursor = v["messages"].get(0).and_then(|m| m["ts"].as_str()).map(String::from).unwrap_or_else(|| chrono::Utc::now().timestamp().to_string()); set_status(c.id, true, "已連上，等待新訊息", 0); } else { let (next, items) = slack_new(&v, &cursor, &own); @@ -469,7 +479,7 @@ fn poll(c: Channel, gen: u64, tx: Sender, repaint: Arc set_status(c.id, false, &format!("Slack 說：{}", v["error"].as_str().unwrap_or("不明錯誤")), 0), - Err(e) if e == "rate" => std::thread::sleep(Duration::from_secs(20)), + Err(e) if e.starts_with("rate") => std::thread::sleep(Duration::from_secs(rate_secs(&e))), Err(e) => set_status(c.id, false, &e, 0), } std::thread::sleep(Duration::from_secs(POLL_SECS)); @@ -497,7 +507,7 @@ fn poll(c: Channel, gen: u64, tx: Sender, repaint: Arc std::thread::sleep(Duration::from_secs(10)), + Err(e) if e.starts_with("rate") => std::thread::sleep(Duration::from_secs(rate_secs(&e))), Err(e) => set_status(c.id, false, &e, 0), } std::thread::sleep(Duration::from_secs(POLL_SECS)); @@ -530,7 +540,7 @@ fn poll(c: Channel, gen: u64, tx: Sender, repaint: Arc std::thread::sleep(Duration::from_secs(20)), + Err(e) if e.starts_with("rate") => std::thread::sleep(Duration::from_secs(rate_secs(&e))), Err(e) => set_status(c.id, false, &e, 0), } } @@ -655,12 +665,20 @@ fn poll(c: Channel, gen: u64, tx: Sender, repaint: Arc { - // Skip what's waiting from before: start after the latest update. - let mut offset = get_json(&format!("{}/bot{token}/getUpdates?offset=-1", tg()), None, 15) - .ok() - .and_then(|v| v["result"][0]["update_id"].as_i64()) - .map(|u| u + 1) - .unwrap_or(0); + // Skip what's waiting from before: start after the latest update. If this acknowledge + // fails, offset 0 would replay the whole backlog on the first poll — retry it. + let mut offset = loop { + match get_json(&format!("{}/bot{token}/getUpdates?offset=-1", tg()), None, 15) { + Ok(v) => break v["result"].as_array().and_then(|a| a.last()).and_then(|u| u["update_id"].as_i64()).map(|u| u + 1).unwrap_or(1), + Err(e) => { + set_status(c.id, false, &format!("正在連上 Telegram：{e}"), 0); + if !alive(gen) { + return; + } + std::thread::sleep(Duration::from_secs(rate_secs(&e))); + } + } + }; set_status(c.id, true, "已連上，等待新訊息", 0); while alive(gen) { let url = format!("{}/bot{token}/getUpdates?offset={offset}&timeout={TELEGRAM_WAIT_SECS}", tg()); @@ -677,7 +695,8 @@ fn poll(c: Channel, gen: u64, tx: Sender, repaint: Arc { set_status(c.id, false, &e, 0); - std::thread::sleep(Duration::from_secs(5)); + // A 429 says how long to back off; hammering every 5 s keeps it limited. + std::thread::sleep(Duration::from_secs(if e.starts_with("rate") { rate_secs(&e) } else { 5 })); } } } diff --git a/src/pages.rs b/src/pages.rs index 59a0c75..c8e52df 100644 --- a/src/pages.rs +++ b/src/pages.rs @@ -87,8 +87,10 @@ pub fn describe(name: &str, args: &Value) -> String { } fn find<'a>(pages: &'a [Page], bot: u64, id: Option, title: &str) -> Option<&'a Page> { + // A page belongs to the bot that made it: ids are guesses a model can make, and reading + // (or overwriting) another bot's page by id must not work. match id { - Some(i) => pages.iter().find(|p| p.id == i), + Some(i) => pages.iter().find(|p| p.id == i && p.bot == bot), None => pages.iter().find(|p| p.bot == bot && same_title(&p.title, title)), } } diff --git a/src/routines.rs b/src/routines.rs index 8ae24e2..3321cb6 100644 --- a/src/routines.rs +++ b/src/routines.rs @@ -130,6 +130,26 @@ pub fn validate(mut s: Schedule) -> Result { if s.frequency == Frequency::Minutely && s.interval < MIN_MINUTES { return Err(format!("每 N 分鐘最少 {MIN_MINUTES} 分鐘")); } + // Keep the "next run" maths (which searches interval-sized steps) bounded, and absurd + // intervals out: a huge hourly interval used to spin the UI thread for minutes on create. + let cap: u32 = match s.frequency { + Frequency::Minutely => 7 * 24 * 60, + Frequency::Hourly => 31 * 24, + Frequency::Daily => 366, + Frequency::Weekly => 53, + Frequency::Monthly => 120, + Frequency::Yearly => 10, + }; + if s.interval > cap { + return Err(format!("間隔太大（這種頻率最多每 {cap} 單位一次）")); + } + if s.frequency == Frequency::Yearly { + let (m, d) = (s.month.unwrap_or(1), s.month_day.unwrap_or(1)); + // 2 月 31 日永遠不會來：這個排程建立即失效，當初卻毫無警告。 + if chrono::NaiveDate::from_ymd_opt(2024, m, d).is_none() { + return Err(format!("每年 {m} 月 {d} 日不存在，無法排入")); + } + } match s.frequency { Frequency::Weekly if s.days_of_week.is_empty() => return Err("每週的例行任務要指定星期幾（daysOfWeek，例如 [\"MO\",\"WE\"]）".into()), Frequency::Monthly if s.month_day.is_none() && s.weekday_of_month.is_none() => s.month_day = Some(1), @@ -245,7 +265,8 @@ fn day_matches(s: &Schedule, anchor: NaiveDate, d: NaiveDate) -> bool { pub fn next_run(s: &Schedule, created: i64, after: i64) -> Option { if s.once { let first = exact_next(s, created, created - 1)?; - let first = flex(s, created, first); + // A one-time run is never flexed: flexing its only moment into the past made the task + // born already-late and it died silently (next_run = None). return (first > after).then_some(first); } // Look a little earlier so a time pulled forward by "左右" isn't skipped. @@ -280,10 +301,16 @@ fn exact_next(s: &Schedule, created: i64, after: i64) -> Option { } if s.frequency == Frequency::Hourly { let minute = parse_time(&s.time).map(|t| chrono::Timelike::minute(&t) as i64).unwrap_or(0); - let hour0 = created.div_euclid(3600); - let mut h = after.div_euclid(3600); + // Count hours in the schedule's own timezone: its hour line sits at a different + // epoch minute in +05:30/+05:45 — anchoring hours to UTC ran those zones 30/45 late. + let tz = tz_of(&s.timezone); + let at = Utc.timestamp_opt(created, 0).single()?.with_timezone(&tz); + // Offset seconds of that zone at `created` (local wall clock minus UTC). + let off = (at.naive_local() - at.naive_utc()).num_seconds() as i64; + let hour0 = (created + off).div_euclid(3600); + let mut h = (after + off).div_euclid(3600); for _ in 0..(n * 2 + 2) { - let ts = h * 3600 + minute * 60; + let ts = h * 3600 + minute * 60 - off; if ts > after && (h - hour0).rem_euclid(n) == 0 { return Some(ts); } @@ -300,8 +327,17 @@ fn exact_next(s: &Schedule, created: i64, after: i64) -> Option { if !day_matches(s, anchor, d) { continue; } - // Skipped or doubled clock times (summer time) take the earliest valid moment. - let Some(at) = tz.from_local_datetime(&d.and_time(time)).earliest() else { continue }; + // Skipped or doubled clock times (summer time): a doubled time takes its first + // occurrence; a skipped one (spring-forward) runs just after the gap, not never. + let local = d.and_time(time); + let at = match tz.from_local_datetime(&local) { + chrono::LocalResult::Single(at) => at, + chrono::LocalResult::Ambiguous(at, _) => at, + chrono::LocalResult::None => match tz.from_local_datetime(&(local + Duration::hours(1))).earliest() { + Some(at) => at, + None => continue, + }, + }; let ts = at.timestamp(); if ts > after { return Some(ts); diff --git a/src/triggers.rs b/src/triggers.rs index f9d0267..4021d79 100644 --- a/src/triggers.rs +++ b/src/triggers.rs @@ -315,9 +315,12 @@ pub fn github_runs(runs: &Value, seen: &mut HashSet, first: bool) -> Vec`. pub fn status_changes(prev: &mut HashMap, items: &[(String, String, String)], first: bool) -> Vec<(String, String)> { let mut out = vec![]; + // An empty prev is always the baseline: a poll that follows a failed first round must + // record, not fire every existing item as "created" (history replay). + let baseline = first || prev.is_empty(); for (id, status, title) in items { match prev.get(id) { - None if !first => out.push(("created".to_string(), title.clone())), + None if !baseline => out.push(("created".to_string(), title.clone())), Some(old) if old != status => out.push((status.clone(), format!("{title}（{old} → {status}）"))), _ => {} } @@ -437,6 +440,9 @@ fn sleep_alive(gen: u64, secs: u64) -> bool { pub fn restart(watches: Vec, slack_token: String, tx: Sender, repaint: Arc) { let gen = GENERATION.fetch_add(1, Ordering::SeqCst) + 1; webhooks().lock().unwrap().clear(); + // Fresh router too: old entries held senders of the previous generation (a leak, and a + // webhook could fire a routine that no longer exists). + *ROUTER_TX.lock().unwrap() = None; for w in watches { start_one(w.bot_id, w.routine_id, w.trigger, gen, slack_token.clone(), tx.clone(), repaint.clone()); } @@ -497,6 +503,9 @@ fn watch_loop(routine_id: u64, t: Trigger, gen: u64, slack_token: String, fire: let mut slack_own = String::new(); let mut slack_reacted: HashSet = HashSet::new(); loop { + // first must only advance after a round that actually reached the service: a failed + // first round used to make the next one replay the whole history as new events. + let mut reached = true; let wait = match &t { Trigger::Github { repo, token, events, users, branch, pr } => { let mut h = vec![("Accept", "application/vnd.github+json".to_string())]; @@ -515,7 +524,7 @@ fn watch_loop(routine_id: u64, t: Trigger, gen: u64, slack_token: String, fire: cursor = next; note(routine_id, "已連上 GitHub，等待事件"); } - Err(e) => note(routine_id, &format!("GitHub：{e}")), + Err(e) => { reached = false; note(routine_id, &format!("GitHub：{e}")) }, } if wanted(events, "ci_passed") || wanted(events, "ci_failed") { if let Ok(v) = get(&format!("https://api.github.com/repos/{repo}/actions/runs?per_page=20"), &h) { @@ -560,7 +569,7 @@ fn watch_loop(routine_id: u64, t: Trigger, gen: u64, slack_token: String, fire: note(routine_id, "已連上 Slack，等待訊息"); } Ok(v) => note(routine_id, &format!("Slack 說：{}", v["error"].as_str().unwrap_or("不明錯誤"))), - Err(e) => note(routine_id, &format!("Slack：{e}")), + Err(e) => { reached = false; note(routine_id, &format!("Slack：{e}")) }, } SLACK_POLL_SECS } @@ -583,7 +592,7 @@ fn watch_loop(routine_id: u64, t: Trigger, gen: u64, slack_token: String, fire: } note(routine_id, "已連上 Linear，等待變化"); } - Err(e) => note(routine_id, &format!("Linear：{e}")), + Err(e) => { reached = false; note(routine_id, &format!("Linear：{e}")) }, } SERVICE_POLL_SECS } @@ -620,7 +629,7 @@ fn watch_loop(routine_id: u64, t: Trigger, gen: u64, slack_token: String, fire: } note(routine_id, "已連上 Sentry，等待變化"); } - Err(e) => note(routine_id, &format!("Sentry：{e}")), + Err(e) => { reached = false; note(routine_id, &format!("Sentry：{e}")) }, } SERVICE_POLL_SECS } @@ -642,7 +651,7 @@ fn watch_loop(routine_id: u64, t: Trigger, gen: u64, slack_token: String, fire: } note(routine_id, "已連上 PagerDuty，等待事故"); } - Err(e) => note(routine_id, &format!("PagerDuty：{e}")), + Err(e) => { reached = false; note(routine_id, &format!("PagerDuty：{e}")) }, } SERVICE_POLL_SECS } @@ -663,6 +672,7 @@ fn watch_loop(routine_id: u64, t: Trigger, gen: u64, slack_token: String, fire: } note(routine_id, "正在看「郵件」的收件匣"); } else { + reached = false; note(routine_id, &o); } SERVICE_POLL_SECS @@ -679,13 +689,31 @@ fn watch_loop(routine_id: u64, t: Trigger, gen: u64, slack_token: String, fire: } note(routine_id, "正在看「行事曆」"); } else { + reached = false; note(routine_id, &o); } CALENDAR_POLL_SECS } Trigger::Webhook { .. } | Trigger::Group { .. } => return, }; - first = false; + if reached { + first = false; + } + // These sets grow with every event seen: a watcher that runs for months must stay + // bounded (ids grow, so keep the newest range; the react/calendar sets are only + // short-term de-duplication, so a cap that clears is harmless). + if seen.len() > 4096 { + let m = *seen.iter().max().unwrap_or(&0); + seen.retain(|id| *id + 100_000 >= m); + } + if prev.len() > 4096 { + // status_changes treats an empty set as a quiet baseline (records, fires nothing), + // so clearing is safe and keeps the map from growing with every issue ever seen. + prev.clear(); + } + if slack_reacted.len() > 4096 { + slack_reacted.clear(); + } if !sleep_alive(gen, wait) { return; } diff --git a/src/voice.rs b/src/voice.rs index 29c0816..3c59f2a 100644 --- a/src/voice.rs +++ b/src/voice.rs @@ -163,7 +163,7 @@ impl Recording { } match said.trim() { "denied" => { - let _ = Command::new("open").arg("x-apple.systempreferences:com.apple.preference.security?Privacy_Microphone").spawn(); + let _ = crate::computer::spawn_reaped(&mut Command::new("open").arg("x-apple.systempreferences:com.apple.preference.security?Privacy_Microphone")); Err("macOS 還沒讓 AnyBot 使用麥克風。我已經打開「系統設定 → 隱私權與安全性 → 麥克風」：把 AnyBot（或「終端機」）打開，再試一次。".into()) } "fail" => Err("找不到可以用的麥克風。".into()),