perf(logs): 日志事件按窗口合并推送,抽屉改为虚拟滚动

- 后端 LogSink 把逐行事件按 16 KiB / 100 ms 合并成一条 log 事件,文件写入保持即时;后台 tick 与析构 flush 保证低输出和收尾都不丢尾部
- 前端新增 LogViewer:固定行高虚拟滚动、跟随底部/回到最新,增量按 rAF 合并,长日志 DOM 节点数恒定
- 日志抽屉改为行缓冲(lib/logLines)+ 可见窗口计算(lib/logWindow)
- 详情区改 flex 布局,日志面板吃掉剩余高度,回到最新按钮不再被裁到面板外
- 测试:Rust 单元 52→57,Vitest 13→28,Playwright 4→5(1500 行虚拟滚动)
- 文档:README 补充合并推送与虚拟滚动,新增 ADR-0008
This commit is contained in:
2026-10-27 22:38:53 +08:00
parent 14fb3563c5
commit 07ca14ea34
12 changed files with 867 additions and 59 deletions
+215 -29
View File
@@ -37,6 +37,10 @@ use crate::{
const MAX_LOG_BYTES: u64 = 5 * 1024 * 1024;
const MAX_LOG_LINE_BYTES: usize = 256 * 1024;
const MAX_HTTP_LOG_BODY_BYTES: usize = 64 * 1024;
/// 合并推送的日志事件上限:攒够这么多字节就立即发一个事件。
const LOG_EVENT_BATCH_BYTES: usize = 16 * 1024;
/// 合并推送的时间窗口:超过这个时间还没攒满也发出去,保证低输出时也有实时感。
const LOG_EVENT_BATCH_INTERVAL: Duration = Duration::from_millis(100);
const TERMINATION_WAIT_SECS: u64 = 5;
const CREATE_NO_WINDOW: u32 = 0x0800_0000;
const DEFAULT_RETENTION_DAYS: u64 = 30;
@@ -175,13 +179,28 @@ struct RunningHandle {
#[derive(Clone)]
struct LogSink {
file: Arc<AsyncMutex<tokio::fs::File>>,
size: Arc<AtomicU64>,
truncated: Arc<AtomicBool>,
inner: Arc<LogSinkInner>,
}
struct LogSinkInner {
file: AsyncMutex<tokio::fs::File>,
size: AtomicU64,
truncated: AtomicBool,
run_id: i64,
events: broadcast::Sender<ExecutionEvent>,
secrets: Arc<Vec<String>>,
secrets: Vec<String>,
encoding: OutputEncoding,
/// 已经写进文件、但还没通过 SSE 推送出去的字节。
pending: Mutex<PendingBatch>,
}
/// 等待合并推送的一段日志。`offset` 是这段内容在日志文件中的起始偏移。
#[derive(Default)]
struct PendingBatch {
offset: u64,
bytes: usize,
opened_at: Option<Instant>,
text: String,
}
struct ScriptOutcome {
@@ -466,7 +485,7 @@ impl TaskExecutor {
let secrets = sensitive_values(&task.action);
let sink = match LogSink::create(
&log_path,
task,
&task.id,
run_id,
self.inner.events.clone(),
secrets,
@@ -631,7 +650,7 @@ impl TaskExecutor {
let log_path = self.log_path(&task.id, run_id);
let sink = match LogSink::create(
&log_path,
task,
&task.id,
run_id,
self.inner.events.clone(),
sensitive_values(&task.action),
@@ -812,7 +831,7 @@ impl TaskExecutor {
impl LogSink {
async fn create(
path: &Path,
task: &TaskResponse,
task_id: &str,
run_id: i64,
events: broadcast::Sender<ExecutionEvent>,
secrets: Vec<String>,
@@ -822,8 +841,7 @@ impl LogSink {
tokio::fs::create_dir_all(parent).await?;
}
let header = format!(
"[system] TaskPilot run {run_id} for task {} started at {}\n",
task.id,
"[system] TaskPilot run {run_id} for task {task_id} started at {}\n",
Utc::now().to_rfc3339()
);
tokio::fs::write(path, header.as_bytes()).await?;
@@ -834,45 +852,59 @@ impl LogSink {
let mut secrets = secrets;
secrets.sort_by_key(|value| std::cmp::Reverse(value.len()));
Ok(Self {
file: Arc::new(AsyncMutex::new(file)),
size: Arc::new(AtomicU64::new(header.len() as u64)),
truncated: Arc::new(AtomicBool::new(false)),
let inner = Arc::new(LogSinkInner {
file: AsyncMutex::new(file),
size: AtomicU64::new(header.len() as u64),
truncated: AtomicBool::new(false),
run_id,
events,
secrets: Arc::new(secrets),
secrets,
encoding,
})
pending: Mutex::new(PendingBatch::default()),
});
spawn_pending_flusher(&inner);
Ok(Self { inner })
}
fn decode_output(&self, bytes: &[u8]) -> String {
self.encoding.decode(bytes)
self.inner.encoding.decode(bytes)
}
async fn append(&self, text: &str) {
let redacted = redact_text(text, &self.secrets);
let redacted = redact_text(text, &self.inner.secrets);
let bytes = redacted.as_bytes();
let current = self.size.load(Ordering::Relaxed);
let mut file = self.inner.file.lock().await;
let current = self.inner.size.load(Ordering::Relaxed);
if current + bytes.len() as u64 > MAX_LOG_BYTES {
if !self.truncated.swap(true, Ordering::Relaxed) {
if !self.inner.truncated.swap(true, Ordering::Relaxed) {
let notice = format!("[system] 日志超过 {MAX_LOG_BYTES} 字节,后续输出已丢弃\n");
let mut file = self.file.lock().await;
let _ = file.write_all(notice.as_bytes()).await;
}
return;
}
let mut file = self.file.lock().await;
if file.write_all(bytes).await.is_err() {
return;
}
let next = self.size.fetch_add(bytes.len() as u64, Ordering::Relaxed) + bytes.len() as u64;
let _ = self.events.send(ExecutionEvent::Log {
run_id: self.run_id,
offset: current,
next_offset: next,
content: redacted,
});
let next = current + bytes.len() as u64;
self.inner.size.store(next, Ordering::Relaxed);
// 文件写入是串行的,所以 pending 也在文件锁内更新,保证推送的偏移连续。
let mut pending = lock_pending(&self.inner.pending);
if pending.bytes == 0 {
pending.offset = current;
pending.opened_at = Some(Instant::now());
}
pending.bytes += bytes.len();
pending.text.push_str(&redacted);
let batch_full = pending.bytes >= LOG_EVENT_BATCH_BYTES;
let window_elapsed = pending
.opened_at
.is_some_and(|opened| opened.elapsed() >= LOG_EVENT_BATCH_INTERVAL);
if batch_full || window_elapsed {
emit_pending(&self.inner, &mut pending);
}
}
async fn system(&self, message: &str) {
@@ -884,6 +916,57 @@ impl LogSink {
}
}
impl Drop for LogSinkInner {
fn drop(&mut self) {
let mut pending = lock_pending(&self.pending);
emit_pending(self, &mut pending);
}
}
fn lock_pending(pending: &Mutex<PendingBatch>) -> std::sync::MutexGuard<'_, PendingBatch> {
pending
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
/// 把待推送的日志合并成一个事件发出去,并清空缓冲区。
fn emit_pending(inner: &LogSinkInner, pending: &mut PendingBatch) {
if pending.bytes == 0 {
return;
}
let batch = std::mem::take(pending);
let _ = inner.events.send(ExecutionEvent::Log {
run_id: inner.run_id,
offset: batch.offset,
next_offset: batch.offset + batch.bytes as u64,
content: batch.text,
});
}
fn flush_pending_if_due(inner: &LogSinkInner) {
let mut pending = lock_pending(&inner.pending);
let due = pending
.opened_at
.is_some_and(|opened| opened.elapsed() >= LOG_EVENT_BATCH_INTERVAL);
if due {
emit_pending(inner, &mut pending);
}
}
/// 低输出量时不会触发"攒满即发",靠这个后台任务按时间窗口把日志推出去。
fn spawn_pending_flusher(inner: &Arc<LogSinkInner>) {
let weak = Arc::downgrade(inner);
tokio::spawn(async move {
loop {
sleep(LOG_EVENT_BATCH_INTERVAL / 2).await;
let Some(inner) = weak.upgrade() else {
break;
};
flush_pending_if_due(&inner);
}
});
}
fn lock_running(
running: &Mutex<HashMap<String, RunningHandle>>,
) -> std::sync::MutexGuard<'_, HashMap<String, RunningHandle>> {
@@ -1273,4 +1356,107 @@ mod tests {
remove_running_if(&running, "task", Some(2));
assert!(!lock_running(&running).contains_key("task"));
}
async fn test_sink() -> (
LogSink,
broadcast::Receiver<ExecutionEvent>,
tempfile::TempDir,
PathBuf,
) {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join("run.log");
let (events, receiver) = broadcast::channel(32);
let sink = LogSink::create(
&path,
"task",
7,
events,
Vec::new(),
OutputEncoding::from_env(),
)
.await
.unwrap();
(sink, receiver, temp, path)
}
async fn recv_log(receiver: &mut broadcast::Receiver<ExecutionEvent>) -> (u64, u64, String) {
let event = timeout(Duration::from_secs(2), receiver.recv())
.await
.expect("log event should arrive")
.expect("log channel should stay open");
match event {
ExecutionEvent::Log {
run_id,
offset,
next_offset,
content,
} => {
assert_eq!(run_id, 7);
(offset, next_offset, content)
}
other => panic!("expected log event, got {other:?}"),
}
}
#[tokio::test]
async fn coalesces_rapid_output_into_a_single_log_event() {
let (sink, mut receiver, _temp, path) = test_sink().await;
sink.append("first line\n").await;
sink.append("second line\n").await;
drop(sink);
let (offset, next_offset, content) = recv_log(&mut receiver).await;
assert_eq!(content, "first line\nsecond line\n");
assert_eq!(next_offset - offset, content.len() as u64);
let on_disk = std::fs::read_to_string(&path).unwrap();
assert!(on_disk.ends_with(&content));
}
#[tokio::test]
async fn flushes_the_pending_batch_when_the_sink_is_dropped() {
let (sink, mut receiver, _temp, _path) = test_sink().await;
sink.append("only line\n").await;
drop(sink);
let (offset, next_offset, content) = recv_log(&mut receiver).await;
assert_eq!(content, "only line\n");
assert_eq!(next_offset - offset, content.len() as u64);
}
#[tokio::test]
async fn flushes_a_full_batch_without_waiting_for_the_window() {
let (sink, mut receiver, _temp, _path) = test_sink().await;
sink.append(&"x".repeat(LOG_EVENT_BATCH_BYTES)).await;
let (_offset, _next_offset, content) = recv_log(&mut receiver).await;
assert_eq!(content.len(), LOG_EVENT_BATCH_BYTES);
}
#[tokio::test]
async fn flushes_low_volume_output_within_the_time_window() {
let (sink, mut receiver, _temp, _path) = test_sink().await;
sink.append("quiet output\n").await;
let (_offset, _next_offset, content) = recv_log(&mut receiver).await;
assert_eq!(content, "quiet output\n");
}
#[tokio::test]
async fn batches_keep_contiguous_offsets_across_flushes() {
let (sink, mut receiver, _temp, _path) = test_sink().await;
sink.append(&"x".repeat(LOG_EVENT_BATCH_BYTES)).await;
let (first_offset, first_next, first_content) = recv_log(&mut receiver).await;
sink.append("tail\n").await;
let (second_offset, second_next, second_content) = recv_log(&mut receiver).await;
assert_eq!(first_next - first_offset, first_content.len() as u64);
assert_eq!(second_offset, first_next);
assert_eq!(second_next - second_offset, second_content.len() as u64);
assert_eq!(second_content, "tail\n");
}
}