Routines follow the clock in each timezone; channels resume where they stopped; pages find checks the bot

This commit is contained in:
王性驊 2026-11-05 15:48:22 +08:00
parent ca30f708ed
commit 5dfb5d1c6c
5 changed files with 113 additions and 28 deletions

View File

@ -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<Value, String> {
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<Value, String
let status = resp.status().as_u16();
let text = resp.body_mut().read_to_string().unwrap_or_default();
if status == 429 {
return Err("rate".into());
// Carry the server's own "wait N seconds" (Telegram and Slack both send it); a fixed
// 5-second retry used to hammer the API and stay rate-limited.
let ra = serde_json::from_str::<Value>(&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::<String>()));
@ -457,8 +466,9 @@ fn poll(c: Channel, gen: u64, tx: Sender<Incoming>, repaint: Arc<dyn Fn() + Send
match get_json(&url, Some(&auth), 20) {
Ok(v) if v["ok"].as_bool() == Some(true) => {
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<Incoming>, repaint: Arc<dyn Fn() + Send
}
}
Ok(v) => 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<Incoming>, repaint: Arc<dyn Fn() + Send
}
}
}
Err(e) if e == "rate" => 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<Incoming>, repaint: Arc<dyn Fn() + Send
}
}
}
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),
}
}
@ -655,12 +665,20 @@ fn poll(c: Channel, gen: u64, tx: Sender<Incoming>, repaint: Arc<dyn Fn() + Send
}
}
Kind::Telegram => {
// 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<Incoming>, repaint: Arc<dyn Fn() + Send
}
Err(e) => {
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 }));
}
}
}

View File

@ -87,8 +87,10 @@ pub fn describe(name: &str, args: &Value) -> String {
}
fn find<'a>(pages: &'a [Page], bot: u64, id: Option<u64>, 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)),
}
}

View File

@ -130,6 +130,26 @@ pub fn validate(mut s: Schedule) -> Result<Schedule, String> {
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<i64> {
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<i64> {
}
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<i64> {
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);

View File

@ -315,9 +315,12 @@ pub fn github_runs(runs: &Value, seen: &mut HashSet<u64>, first: bool) -> Vec<Ev
/// status changes. Events are `created` and `<status>`.
pub fn status_changes(prev: &mut HashMap<String, String>, 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<Watch>, slack_token: String, tx: Sender<Fired>, repaint: Arc<dyn Fn() + Send + Sync>) {
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<String> = 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,
};
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;
}

View File

@ -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()),