优化历史表轮换;修复D1 flush 锁问题;新增总览批量唤醒接口并移除逐服务器 WebSocket。
This commit is contained in:
1 parent
55bc7beb8d
commit
c834deb077
16 files changed
+220
-112
No files matched your search
@@ -26,8 +26,9 @@
|
||||
## 部署到 Cloudflare
|
||||
|
||||
1. Fork 本仓库并打开 [Workers & Pages](https://dash.cloudflare.com/?to=/:account/workers-and-pages/create),选择 **Import a repository** 后连接仓库。
|
||||
2. 部署命令修改为 `bun run deploy`。
|
||||
3. 在 **高级设置** 中填写下表变量后部署。
|
||||
2. 在构建设置中将 **Build command(构建命令)清空**。Cloudflare 可能自动识别为 `bun run build`,需要手动删除;如果控制台不允许留空,可填写 `true`。
|
||||
3. 将 **Deploy command(部署命令)** 修改为 `bun run deploy`。该命令会按需构建前端和 Worker、应用 D1 迁移并完成部署,无需单独执行构建命令;同时配置 `bun run build` 会导致前端重复构建。
|
||||
4. 在 **高级设置** 中填写下表变量后部署。
|
||||
|
||||
| 变量 | 必填/可选 | 说明 |
|
||||
| --- | ---: | --- |
|
||||
|
||||
@@ -188,11 +188,12 @@ export function AppProvider({ children }: { children: ReactNode }) {
|
||||
servers: serversRef.current,
|
||||
})),
|
||||
onConnectedChange: (connected) => { liveConnectedRef.current = connected; },
|
||||
wakeTargets: () => {
|
||||
onOverviewConnected: () => {
|
||||
const now = Date.now() / 1000;
|
||||
return serversRef.current
|
||||
const serverIds = serversRef.current
|
||||
.filter((server) => server.timestamp && now - server.timestamp <= config.offline_threshold_seconds)
|
||||
.map((server) => server.id);
|
||||
return serverIds.length ? api.wakeServers(serverIds) : Promise.resolve();
|
||||
},
|
||||
});
|
||||
}, [access, config.offline_threshold_seconds, loading, selectedId]);
|
||||
@@ -211,8 +212,9 @@ export function AppProvider({ children }: { children: ReactNode }) {
|
||||
link.rel = "icon";
|
||||
document.head.append(link);
|
||||
}
|
||||
link.href = config.favicon_url || config.logo_url || "/logo.svg";
|
||||
}, [config.favicon_url, config.logo_url]);
|
||||
link.removeAttribute("type");
|
||||
link.href = config.logo_url || "/logo.svg";
|
||||
}, [config.logo_url]);
|
||||
|
||||
useEffect(() => {
|
||||
const media = matchMedia("(prefers-color-scheme: dark)");
|
||||
|
||||
@@ -50,8 +50,9 @@ function AdminApp() {
|
||||
link.rel = "icon";
|
||||
document.head.append(link);
|
||||
}
|
||||
link.href = config?.favicon_url || config?.logo_url || "/logo.svg";
|
||||
}, [config?.favicon_url, config?.logo_url]);
|
||||
link.removeAttribute("type");
|
||||
link.href = config?.logo_url || "/logo.svg";
|
||||
}, [config?.logo_url]);
|
||||
|
||||
if (error) return <div className={`admin-loading ${dark ? "admin-dark" : ""}`}><span>{error}</span><button className="secondary-btn" onClick={() => void loadConfig()}>重试</button></div>;
|
||||
if (!config) return <div className={`admin-loading ${dark ? "admin-dark" : ""}`}>正在加载管理面板</div>;
|
||||
|
||||
@@ -41,6 +41,10 @@ export const api = {
|
||||
bootstrap: () => request<Bootstrap>("/api/bootstrap"),
|
||||
config: () => request<Config>("/api/config"),
|
||||
exchangeRates: () => request<ExchangeRates>("/api/exchange-rates"),
|
||||
wakeServers: (serverIds: string[]) => request<void>("/api/live/wake", {
|
||||
method: "POST",
|
||||
body: JSON.stringify({ server_ids: serverIds }),
|
||||
}),
|
||||
refreshExchangeRates: () => request<ExchangeRates>("/api/admin/exchange-rates/refresh", { method: "POST" }, true),
|
||||
servers: () => request<{ servers: Server[] }>("/api/servers"),
|
||||
adminServers: () => request<{ servers: AdminServer[] }>("/api/admin/servers", {}, true),
|
||||
|
||||
@@ -620,7 +620,7 @@ export function AdminPanel({
|
||||
<div className="section-title"><Eye size={15} />外观与展示</div>
|
||||
<div className="form-grid"><label><span>站点名称</span><input required value={settings.site_name} onChange={(event) => updateSettings("site_name", event.target.value)} /></label><label><span>站点描述</span><input value={settings.site_description} onChange={(event) => updateSettings("site_description", event.target.value)} /></label></div>
|
||||
<label><span>站点公告</span><textarea rows={3} maxLength={1000} value={settings.site_announcement} onChange={(event) => updateSettings("site_announcement", event.target.value)} /></label>
|
||||
<div className="form-grid three"><label><span>界面语言</span><select value={settings.locale} onChange={(event) => updateSettings("locale", event.target.value as Settings["locale"])}><option value="zh-CN">简体中文</option><option value="en">English</option></select></label><label><span>站点 Logo 地址</span><input type="url" maxLength={1000} value={settings.logo_url} onChange={(event) => updateSettings("logo_url", event.target.value)} placeholder="https://example.com/logo.svg" /></label><label><span>浏览器图标地址</span><input type="url" maxLength={1000} value={settings.favicon_url} onChange={(event) => updateSettings("favicon_url", event.target.value)} placeholder="https://example.com/favicon.png" /></label></div>
|
||||
<div className="form-grid"><label><span>界面语言</span><select value={settings.locale} onChange={(event) => updateSettings("locale", event.target.value as Settings["locale"])}><option value="zh-CN">简体中文</option><option value="en">English</option></select></label><label><span>站点 Logo / 浏览器图标地址</span><input type="url" maxLength={1000} value={settings.logo_url} onChange={(event) => updateSettings("logo_url", event.target.value)} placeholder="https://example.com/logo.svg" /></label></div>
|
||||
<div className="form-grid"><label><span>离线判定(秒)</span><input type="number" min="30" max="3600" value={settings.offline_threshold_seconds} onChange={(event) => updateSettings("offline_threshold_seconds", Number(event.target.value))} /></label><label><span>历史保留(天)</span><input type="number" min="1" max="30" value={settings.history_retention_days} onChange={(event) => updateSettings("history_retention_days", Number(event.target.value))} /></label></div>
|
||||
</> : null}
|
||||
|
||||
|
||||
@@ -5,7 +5,6 @@ export const demoConfig: Config = {
|
||||
site_description: "边缘节点与核心服务运行状态",
|
||||
site_announcement: "",
|
||||
logo_url: "",
|
||||
favicon_url: "",
|
||||
locale: "zh-CN",
|
||||
public_dashboard: true,
|
||||
offline_threshold_seconds: 180,
|
||||
|
||||
+11
-45
@@ -3,16 +3,13 @@ import type { Server } from "./types";
|
||||
|
||||
const RECONNECT_DELAY = 3_000;
|
||||
const HEARTBEAT_INTERVAL = 30_000;
|
||||
const WAKE_CONCURRENCY = 6;
|
||||
const WAKE_TIMEOUT = 5_000;
|
||||
const WAKE_SETTLE = 100;
|
||||
|
||||
export interface LiveTransportHandlers {
|
||||
onServer: (server: Server) => void;
|
||||
onBatch: (updates: BatchUpdate[], cached: boolean) => void;
|
||||
onConnectedChange: (connected: boolean) => void;
|
||||
/** Online servers that should be woken when the overview connects. */
|
||||
wakeTargets: () => string[];
|
||||
/** Wakes online Agents through one batched HTTP request. */
|
||||
onOverviewConnected: () => Promise<void>;
|
||||
}
|
||||
|
||||
export interface LiveTransportOptions {
|
||||
@@ -31,8 +28,8 @@ function endpoint(serverId: string | null): URL {
|
||||
|
||||
/**
|
||||
* WSS-only live feed. Returns a disposer; the caller owns when the connection
|
||||
* starts and stops. Reconnects on close, and on the overview it opens a short
|
||||
* socket per online server so hibernating Durable Objects resume reporting.
|
||||
* starts and stops. Reconnects on close, and asks the Worker to wake online
|
||||
* Agents in one batch when the overview connection opens.
|
||||
*/
|
||||
export function connectLive(
|
||||
{ serverId }: LiveTransportOptions,
|
||||
@@ -40,48 +37,20 @@ export function connectLive(
|
||||
): () => void {
|
||||
const sockets: WebSocket[] = [];
|
||||
const reconnects: number[] = [];
|
||||
const wakeTimers: number[] = [];
|
||||
const wakeSockets = new Set<WebSocket>();
|
||||
let cancelled = false;
|
||||
let wakeInFlight = false;
|
||||
|
||||
const reportConnected = () => {
|
||||
handlers.onConnectedChange(sockets.some((socket) => socket.readyState === WebSocket.OPEN));
|
||||
};
|
||||
|
||||
const wakeOverviewAgents = () => {
|
||||
if (serverId || cancelled) return;
|
||||
const queue = Array.from(new Set(handlers.wakeTargets()));
|
||||
let cursor = 0;
|
||||
let active = 0;
|
||||
const launch = () => {
|
||||
while (!cancelled && active < WAKE_CONCURRENCY && cursor < queue.length) {
|
||||
const socket = new WebSocket(endpoint(queue[cursor++]));
|
||||
wakeSockets.add(socket);
|
||||
active += 1;
|
||||
let finished = false;
|
||||
let timeout = 0;
|
||||
const finish = () => {
|
||||
if (finished) return;
|
||||
finished = true;
|
||||
clearTimeout(timeout);
|
||||
wakeSockets.delete(socket);
|
||||
active -= 1;
|
||||
socket.onopen = null;
|
||||
socket.onclose = null;
|
||||
socket.onerror = null;
|
||||
if (socket.readyState === WebSocket.OPEN || socket.readyState === WebSocket.CONNECTING) {
|
||||
socket.close();
|
||||
}
|
||||
launch();
|
||||
};
|
||||
timeout = window.setTimeout(finish, WAKE_TIMEOUT);
|
||||
wakeTimers.push(timeout);
|
||||
socket.onopen = () => wakeTimers.push(window.setTimeout(finish, WAKE_SETTLE));
|
||||
socket.onclose = finish;
|
||||
socket.onerror = finish;
|
||||
}
|
||||
};
|
||||
launch();
|
||||
if (serverId || cancelled || wakeInFlight) return;
|
||||
wakeInFlight = true;
|
||||
void Promise.resolve()
|
||||
.then(handlers.onOverviewConnected)
|
||||
.catch(() => { /* Best effort; the live socket remains usable. */ })
|
||||
.finally(() => { wakeInFlight = false; });
|
||||
};
|
||||
|
||||
const connect = () => {
|
||||
@@ -126,9 +95,6 @@ export function connectLive(
|
||||
handlers.onConnectedChange(false);
|
||||
clearInterval(heartbeat);
|
||||
reconnects.forEach(clearTimeout);
|
||||
wakeTimers.forEach(clearTimeout);
|
||||
wakeSockets.forEach((socket) => socket.close());
|
||||
wakeSockets.clear();
|
||||
sockets.forEach((socket) => socket.close());
|
||||
sockets.length = 0;
|
||||
};
|
||||
|
||||
@@ -2,8 +2,8 @@ export interface Config {
|
||||
site_name: string;
|
||||
site_description: string;
|
||||
site_announcement: string;
|
||||
/** 站点 Logo,同时用作浏览器标签图标。空则回落到 /logo.svg。 */
|
||||
logo_url: string;
|
||||
favicon_url: string;
|
||||
locale: "zh-CN" | "en";
|
||||
public_dashboard: boolean;
|
||||
offline_threshold_seconds: number;
|
||||
|
||||
@@ -272,7 +272,6 @@ INSERT INTO settings(id, value, updated_at) VALUES (
|
||||
"site_description": "轻量、实时的服务器运行状态",
|
||||
"site_announcement": "",
|
||||
"logo_url": "",
|
||||
"favicon_url": "",
|
||||
"locale": "zh-CN",
|
||||
"public_dashboard": "true",
|
||||
"history_cache_version": "0",
|
||||
|
||||
@@ -188,6 +188,27 @@ try {
|
||||
throw new Error(`Invalid Agent config: ${JSON.stringify(config)}`);
|
||||
}
|
||||
if (!configOnly) {
|
||||
const wakeHintPromise = waitForJsonMessage(
|
||||
agent,
|
||||
"batched overview wake hint",
|
||||
(message) => message.type === "ack" && message.realtimeHint === true,
|
||||
);
|
||||
const wakeResponse = await fetch(new URL("/api/live/wake", baseUrl), {
|
||||
method: "POST",
|
||||
headers: {
|
||||
Authorization: `Bearer ${adminToken}`,
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
body: JSON.stringify({ server_ids: [serverId] }),
|
||||
});
|
||||
if (wakeResponse.status !== 204) {
|
||||
throw new Error(`Batch wake returned HTTP ${wakeResponse.status}: ${await wakeResponse.text()}`);
|
||||
}
|
||||
const wakeHint = await wakeHintPromise;
|
||||
if (wakeHint.nextWssReportAfterMs !== 5_000) {
|
||||
throw new Error(`Invalid batch wake hint: ${JSON.stringify(wakeHint)}`);
|
||||
}
|
||||
|
||||
const timestamp = Math.floor(Date.now() / 1_000);
|
||||
const baseSample = {
|
||||
timestamp,
|
||||
|
||||
@@ -159,6 +159,11 @@ struct AgentConfigRow {
|
||||
auto_update: i64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct ServerIdRow {
|
||||
id: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub struct AgentLiveContext {
|
||||
pub report_interval: i64,
|
||||
@@ -365,7 +370,6 @@ pub struct SettingsView {
|
||||
pub site_description: String,
|
||||
pub site_announcement: String,
|
||||
pub logo_url: String,
|
||||
pub favicon_url: String,
|
||||
pub locale: String,
|
||||
pub public_dashboard: bool,
|
||||
pub offline_threshold_seconds: i64,
|
||||
@@ -512,6 +516,27 @@ pub async fn list_servers(db: &D1Database, include_hidden: bool) -> Result<Vec<S
|
||||
db.prepare(query).all().await?.results()
|
||||
}
|
||||
|
||||
/// Returns only requested server IDs that exist and are visible, preserving request order.
|
||||
pub async fn visible_server_ids(db: &D1Database, ids: &[String]) -> Result<Vec<String>> {
|
||||
if ids.is_empty() {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
let requested = serde_json::to_string(ids)?;
|
||||
let rows = db
|
||||
.prepare(
|
||||
r#"SELECT s.id
|
||||
FROM json_each(?1) requested
|
||||
JOIN servers s ON s.id = CAST(requested.value AS TEXT)
|
||||
WHERE s.hidden = 0
|
||||
ORDER BY CAST(requested.key AS INTEGER)"#,
|
||||
)
|
||||
.bind(&[text(&requested)])?
|
||||
.all()
|
||||
.await?
|
||||
.results::<ServerIdRow>()?;
|
||||
Ok(rows.into_iter().map(|row| row.id).collect())
|
||||
}
|
||||
|
||||
pub async fn get_server(
|
||||
db: &D1Database,
|
||||
id: &str,
|
||||
@@ -1204,7 +1229,6 @@ pub async fn settings(
|
||||
site_description: string_setting(&values, "site_description", "轻量、实时的服务器运行状态"),
|
||||
site_announcement: string_setting(&values, "site_announcement", ""),
|
||||
logo_url: string_setting(&values, "logo_url", ""),
|
||||
favicon_url: string_setting(&values, "favicon_url", ""),
|
||||
locale: string_setting(&values, "locale", "zh-CN"),
|
||||
public_dashboard: bool_setting(&values, "public_dashboard", true),
|
||||
offline_threshold_seconds: values
|
||||
@@ -1280,9 +1304,6 @@ pub async fn update_settings(
|
||||
if let Some(value) = input.logo_url.as_deref() {
|
||||
push_setting!("logo_url", value.trim());
|
||||
}
|
||||
if let Some(value) = input.favicon_url.as_deref() {
|
||||
push_setting!("favicon_url", value.trim());
|
||||
}
|
||||
if let Some(value) = input.locale.as_deref() {
|
||||
push_setting!("locale", value.trim());
|
||||
}
|
||||
@@ -1658,24 +1679,22 @@ pub async fn cleanup_history(db: &D1Database, retention_days: i64) -> Result<()>
|
||||
let delete_archive = db
|
||||
.prepare("DELETE FROM metric_history_hourly WHERE timestamp < ?1")
|
||||
.bind(&[number(cutoff)])?;
|
||||
// DDL 不和上面的语句放进同一个 batch:batch 是隐式事务,而 SQLite 不允许
|
||||
// 在事务里 ALTER TABLE ... RENAME。压缩+删除先提交,轮换单独走;轮换中途
|
||||
// 失败下一次 cron 重来即可,重复压缩无害(见上)。
|
||||
// 压缩和保留期删除先原子提交;轮换随后在另一个原子 batch 中完成。
|
||||
// 两个阶段都可独立重试,且轮换失败时不会留下缺少当前表的半完成状态。
|
||||
db.batch(vec![compact, delete_archive]).await?;
|
||||
rotate_recent_history(db).await
|
||||
}
|
||||
|
||||
/// 丢弃上一代分钟历史,把当前代降级为上一代,再建一张空的当前代。
|
||||
async fn rotate_recent_history(db: &D1Database) -> Result<()> {
|
||||
db.prepare("DROP TABLE IF EXISTS metric_history_old")
|
||||
.run()
|
||||
.await?;
|
||||
db.prepare("ALTER TABLE metric_history RENAME TO metric_history_old")
|
||||
.run()
|
||||
.await?;
|
||||
db.prepare(metric_history_ddl("metric_history"))
|
||||
.run()
|
||||
.await?;
|
||||
// D1 batch guarantees transactional execution, including SQLite DDL. If any
|
||||
// statement fails, the prior pair of recent-history tables remains intact.
|
||||
db.batch(vec![
|
||||
db.prepare("DROP TABLE IF EXISTS metric_history_old"),
|
||||
db.prepare("ALTER TABLE metric_history RENAME TO metric_history_old"),
|
||||
db.prepare(metric_history_ddl("metric_history")),
|
||||
])
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
+95
-22
@@ -1,6 +1,7 @@
|
||||
use std::cell::RefCell;
|
||||
use std::collections::HashSet;
|
||||
|
||||
use futures_util::{stream, StreamExt};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use worker::*;
|
||||
|
||||
@@ -610,6 +611,29 @@ pub struct LiveHub {
|
||||
active_d1_flushes: RefCell<HashSet<String>>,
|
||||
}
|
||||
|
||||
struct ActiveD1FlushGuard<'a> {
|
||||
active: &'a RefCell<HashSet<String>>,
|
||||
server_id: String,
|
||||
}
|
||||
|
||||
impl<'a> ActiveD1FlushGuard<'a> {
|
||||
fn acquire(active: &'a RefCell<HashSet<String>>, server_id: &str) -> Option<Self> {
|
||||
if !active.borrow_mut().insert(server_id.to_string()) {
|
||||
return None;
|
||||
}
|
||||
Some(Self {
|
||||
active,
|
||||
server_id: server_id.to_string(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for ActiveD1FlushGuard<'_> {
|
||||
fn drop(&mut self) {
|
||||
self.active.borrow_mut().remove(&self.server_id);
|
||||
}
|
||||
}
|
||||
|
||||
impl LiveHub {
|
||||
async fn broadcast_to_global_dashboard(&self, server_id: &str, payload: &str) -> bool {
|
||||
let result = async {
|
||||
@@ -664,7 +688,7 @@ impl LiveHub {
|
||||
.is_empty()
|
||||
}
|
||||
|
||||
fn hint_agents(&self, server_id: Option<&str>) {
|
||||
fn hint_agents(&self, server_id: Option<&str>, force_realtime: bool) {
|
||||
let sockets = server_id.map_or_else(
|
||||
|| self.state.get_websockets_with_tag("agents"),
|
||||
|server_id| {
|
||||
@@ -680,7 +704,8 @@ impl LiveHub {
|
||||
let Some(agent_server_id) = attachment.server_id.as_deref() else {
|
||||
continue;
|
||||
};
|
||||
let realtime_active = !attachment.hidden && self.dashboard_active_for(agent_server_id);
|
||||
let realtime_active = !attachment.hidden
|
||||
&& (force_realtime || self.dashboard_active_for(agent_server_id));
|
||||
let Ok(payload) = live_ack_payload(
|
||||
now,
|
||||
false,
|
||||
@@ -846,6 +871,10 @@ impl DurableObject for LiveHub {
|
||||
}
|
||||
return Response::empty();
|
||||
}
|
||||
Some("hint-agents") => {
|
||||
self.hint_agents(None, true);
|
||||
return Response::empty();
|
||||
}
|
||||
Some("record-alert-sample") => {
|
||||
let record = req.json::<AlertRecordRequest>().await?;
|
||||
self.record_alert_point(&record.server_id, &record.point)
|
||||
@@ -947,7 +976,7 @@ impl DurableObject for LiveHub {
|
||||
pair.server.serialize_attachment(dashboard_attachment())?;
|
||||
self.state.accept_websocket_with_tags(&pair.server, &[&tag]);
|
||||
self.replay_latest(&pair.server, server_id.as_deref());
|
||||
self.hint_agents(server_id.as_deref());
|
||||
self.hint_agents(server_id.as_deref(), false);
|
||||
Response::from_websocket(pair.client)
|
||||
}
|
||||
|
||||
@@ -1043,16 +1072,21 @@ impl DurableObject for LiveHub {
|
||||
|
||||
let due_for_d1 = received_at.saturating_sub(attachment.last_d1_write_at)
|
||||
>= HISTORY_WRITE_INTERVAL_SECONDS;
|
||||
let flush_active = self.active_d1_flushes.borrow().contains(&server_id);
|
||||
let mut flush = if !unpersisted_reports.is_empty()
|
||||
&& (due_for_d1 || attachment.d1_flush.is_some())
|
||||
&& !flush_active
|
||||
{
|
||||
let database = self.env.d1("DB")?;
|
||||
begin_d1_flush(&mut attachment).map(|snapshot| (database, snapshot))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let should_flush =
|
||||
!unpersisted_reports.is_empty() && (due_for_d1 || attachment.d1_flush.is_some());
|
||||
let mut flush_guard = None;
|
||||
let mut flush = None;
|
||||
if should_flush {
|
||||
// Acquire before any external await below. Durable Object events may
|
||||
// interleave while broadcasting or writing D1.
|
||||
if let Some(guard) = ActiveD1FlushGuard::acquire(&self.active_d1_flushes, &server_id) {
|
||||
let database = self.env.d1("DB")?;
|
||||
flush = begin_d1_flush(&mut attachment).map(|snapshot| (database, snapshot));
|
||||
if flush.is_some() {
|
||||
flush_guard = Some(guard);
|
||||
}
|
||||
}
|
||||
}
|
||||
if flush.is_some() {
|
||||
if !trim_socket_attachment(&mut attachment) {
|
||||
return ws.close(Some(1011), Some("live state exceeds attachment limit"));
|
||||
@@ -1081,9 +1115,6 @@ impl DurableObject for LiveHub {
|
||||
let mut persistence_error = false;
|
||||
let mut attachment_serialized = false;
|
||||
if let Some((database, snapshot)) = flush.take() {
|
||||
self.active_d1_flushes
|
||||
.borrow_mut()
|
||||
.insert(server_id.clone());
|
||||
let history_point = snapshot
|
||||
.history
|
||||
.point()
|
||||
@@ -1110,7 +1141,6 @@ impl DurableObject for LiveHub {
|
||||
}
|
||||
Ok(inserted) => (true, inserted),
|
||||
};
|
||||
self.active_d1_flushes.borrow_mut().remove(&server_id);
|
||||
|
||||
let Some(mut current_attachment) = ws.deserialize_attachment::<SocketAttachment>()?
|
||||
else {
|
||||
@@ -1134,6 +1164,7 @@ impl DurableObject for LiveHub {
|
||||
attachment = current_attachment;
|
||||
attachment_serialized = true;
|
||||
persisted = write_succeeded;
|
||||
drop(flush_guard.take());
|
||||
|
||||
if newly_written {
|
||||
match crate::db::has_enabled_alert_rules(&database).await {
|
||||
@@ -1314,6 +1345,36 @@ pub async fn disconnect_agents(env: &Env, server_ids: &[String]) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn wake_agents(env: &Env, server_ids: &[String]) -> Result<()> {
|
||||
if server_ids.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
let namespace = env.durable_object("LIVE_HUB")?;
|
||||
let results = stream::iter(server_ids.iter())
|
||||
.map(|server_id| {
|
||||
let namespace = &namespace;
|
||||
async move {
|
||||
let stub = namespace
|
||||
.id_from_name(&server_hub_name(server_id))?
|
||||
.get_stub()?;
|
||||
let headers = Headers::new();
|
||||
headers.set("X-Live-Action", "hint-agents")?;
|
||||
let mut init = RequestInit::new();
|
||||
init.with_method(Method::Post).with_headers(headers);
|
||||
let req = Request::new_with_init("https://live.internal/hint-agents", &init)?;
|
||||
let response = stub.fetch_with_request(req).await?;
|
||||
ensure_live_success(&response, "hint-agents")
|
||||
}
|
||||
})
|
||||
.buffer_unordered(6)
|
||||
.collect::<Vec<_>>()
|
||||
.await;
|
||||
for result in results {
|
||||
result?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn upgrade(req: Request, env: &Env) -> Result<Response> {
|
||||
let namespace = env.durable_object("LIVE_HUB")?;
|
||||
let server_id = req
|
||||
@@ -1403,15 +1464,17 @@ mod tests {
|
||||
agent_wss_interval_ms, apply_live_traffic, batch_update_parts_at, begin_d1_flush,
|
||||
cached_update_value, consume_agent_message_budget, finish_d1_flush, live_ack_payload,
|
||||
next_d1_write_ms, outbound_close_code, trim_cached_live_samples, trim_socket_attachment,
|
||||
update_alert_window, AgentMessageBudget, AlertEvaluationServer, AlertWindowSample,
|
||||
CachedLiveSample, SocketAttachment, TrafficAttachment, ALERT_WINDOW_SECONDS,
|
||||
BROWSER_OMITTED_REPORT_FIELDS, CACHED_LIVE_TTL_SECONDS, MAX_AGENT_BYTES_PER_WINDOW,
|
||||
MAX_AGENT_MESSAGES_PER_WINDOW, MAX_AGENT_SAMPLES_PER_WINDOW, MAX_CACHED_LIVE_BYTES,
|
||||
MAX_CACHED_LIVE_SAMPLES, MAX_SERIALIZED_ATTACHMENT_BYTES,
|
||||
update_alert_window, ActiveD1FlushGuard, AgentMessageBudget, AlertEvaluationServer,
|
||||
AlertWindowSample, CachedLiveSample, SocketAttachment, TrafficAttachment,
|
||||
ALERT_WINDOW_SECONDS, BROWSER_OMITTED_REPORT_FIELDS, CACHED_LIVE_TTL_SECONDS,
|
||||
MAX_AGENT_BYTES_PER_WINDOW, MAX_AGENT_MESSAGES_PER_WINDOW, MAX_AGENT_SAMPLES_PER_WINDOW,
|
||||
MAX_CACHED_LIVE_BYTES, MAX_CACHED_LIVE_SAMPLES, MAX_SERIALIZED_ATTACHMENT_BYTES,
|
||||
};
|
||||
use crate::db::HistoryMetricAggregate;
|
||||
use crate::latency::LatencyMetricAggregates;
|
||||
use crate::models::{AgentLatencyResult, AgentReport, AlertRuleView, HistoryPoint};
|
||||
use std::cell::RefCell;
|
||||
use std::collections::HashSet;
|
||||
|
||||
fn alert_point(timestamp: i64, cpu: f64, net_in: f64) -> HistoryPoint {
|
||||
HistoryPoint {
|
||||
@@ -1499,6 +1562,16 @@ mod tests {
|
||||
assert_eq!(outbound_close_code(5000), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn d1_flush_guard_is_exclusive_and_releases_on_drop() {
|
||||
let active = RefCell::new(HashSet::new());
|
||||
let guard = ActiveD1FlushGuard::acquire(&active, "node-a").expect("first flush lock");
|
||||
assert!(ActiveD1FlushGuard::acquire(&active, "node-a").is_none());
|
||||
assert!(ActiveD1FlushGuard::acquire(&active, "node-b").is_some());
|
||||
drop(guard);
|
||||
assert!(ActiveD1FlushGuard::acquire(&active, "node-a").is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bounds_agent_messages_per_connection_window() {
|
||||
let mut budget = AgentMessageBudget::default();
|
||||
|
||||
@@ -55,7 +55,6 @@ pub struct SettingsInput {
|
||||
pub site_description: Option<String>,
|
||||
pub site_announcement: Option<String>,
|
||||
pub logo_url: Option<String>,
|
||||
pub favicon_url: Option<String>,
|
||||
pub locale: Option<String>,
|
||||
pub public_dashboard: Option<bool>,
|
||||
pub offline_threshold_seconds: Option<i64>,
|
||||
|
||||
+1
-7
@@ -663,6 +663,7 @@ async fn settings_patch(req: &mut Request, ctx: &RouteContext) -> Result<Respons
|
||||
{
|
||||
return error("站点公告不能超过 1000 个字符", 400);
|
||||
}
|
||||
// logo_url 同时作为站点 Logo 和浏览器标签图标使用。
|
||||
if input
|
||||
.logo_url
|
||||
.as_deref()
|
||||
@@ -789,13 +790,6 @@ async fn settings_patch(req: &mut Request, ctx: &RouteContext) -> Result<Respons
|
||||
{
|
||||
return error("界面语言仅支持简体中文或 English", 400);
|
||||
}
|
||||
if input
|
||||
.favicon_url
|
||||
.as_deref()
|
||||
.is_some_and(|value| !valid_public_asset_url(value))
|
||||
{
|
||||
return error("站点图标必须使用 HTTPS 地址", 400);
|
||||
}
|
||||
let password_hash = if input
|
||||
.new_password
|
||||
.as_ref()
|
||||
|
||||
+34
-2
@@ -1,5 +1,6 @@
|
||||
use std::collections::HashMap;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
|
||||
use serde::Deserialize;
|
||||
use worker::{Method, Request, Response, Result};
|
||||
|
||||
use crate::auth::{
|
||||
@@ -15,6 +16,13 @@ use crate::{
|
||||
set_admin_session_cookie, store_history_response, API_JSON_MAX_BYTES,
|
||||
};
|
||||
|
||||
const MAX_WAKE_SERVER_IDS: usize = 500;
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct WakeServersRequest {
|
||||
server_ids: Vec<String>,
|
||||
}
|
||||
|
||||
fn request_hostname(req: &Request) -> Option<String> {
|
||||
req.url()
|
||||
.ok()?
|
||||
@@ -55,6 +63,9 @@ pub(crate) async fn route(mut req: Request, ctx: &RouteContext) -> Result<RouteO
|
||||
}
|
||||
return Ok(RouteOutcome::Handled(live::upgrade(req, &ctx.env).await?));
|
||||
}
|
||||
if method == Method::Post && path == "/api/live/wake" {
|
||||
return Ok(RouteOutcome::Handled(wake_servers(&mut req, ctx).await?));
|
||||
}
|
||||
if method == Method::Get && path == "/api/servers" {
|
||||
return Ok(RouteOutcome::Handled(servers(ctx).await?));
|
||||
}
|
||||
@@ -84,7 +95,6 @@ fn config_value(ctx: &RouteContext) -> serde_json::Value {
|
||||
"site_description": settings.site_description,
|
||||
"site_announcement": settings.site_announcement,
|
||||
"logo_url": settings.logo_url,
|
||||
"favicon_url": settings.favicon_url,
|
||||
"locale": settings.locale,
|
||||
"public_dashboard": settings.public_dashboard,
|
||||
"offline_threshold_seconds": settings.offline_threshold_seconds,
|
||||
@@ -294,6 +304,28 @@ async fn servers(ctx: &RouteContext) -> Result<Response> {
|
||||
json(&serde_json::json!({ "servers": servers }), 200)
|
||||
}
|
||||
|
||||
async fn wake_servers(req: &mut Request, ctx: &RouteContext) -> Result<Response> {
|
||||
if let Some(denied) = public_access_denied(ctx) {
|
||||
return denied;
|
||||
}
|
||||
let input: WakeServersRequest = match request_json(req, API_JSON_MAX_BYTES).await {
|
||||
Ok(value) => value,
|
||||
Err(_) => return error("请求格式无效", 400),
|
||||
};
|
||||
if input.server_ids.len() > MAX_WAKE_SERVER_IDS
|
||||
|| input
|
||||
.server_ids
|
||||
.iter()
|
||||
.any(|id| id.is_empty() || id.len() > 80 || id.contains('/'))
|
||||
|| input.server_ids.iter().collect::<HashSet<_>>().len() != input.server_ids.len()
|
||||
{
|
||||
return error("节点列表无效", 400);
|
||||
}
|
||||
let visible_ids = db::visible_server_ids(&ctx.database, &input.server_ids).await?;
|
||||
live::wake_agents(&ctx.env, &visible_ids).await?;
|
||||
no_content()
|
||||
}
|
||||
|
||||
async fn history(req: &Request, ctx: &RouteContext) -> Result<Response> {
|
||||
if let Some(denied) = public_access_denied(ctx) {
|
||||
return denied;
|
||||
|
||||
+4
-6
@@ -63,15 +63,13 @@
|
||||
"crons": ["*/5 * * * *"]
|
||||
},
|
||||
"observability": {
|
||||
"enabled": true,
|
||||
"enabled": false,
|
||||
"logs": {
|
||||
"enabled": true,
|
||||
"head_sampling_rate": 1,
|
||||
"invocation_logs": true
|
||||
"enabled": false,
|
||||
"invocation_logs": false
|
||||
},
|
||||
"traces": {
|
||||
"enabled": true,
|
||||
"head_sampling_rate": 0.05
|
||||
"enabled": false
|
||||
}
|
||||
},
|
||||
"dev": {
|
||||
|
||||
Reference in new issue
Block a user