docs: 更新资源治理验证记录

docs/生产图片视频内存事故与处理记录-2026-07-31.md

@@ -84,4 +84,46 @@
 - 上线后必须观察 `P2024` 数量、数据库连接数、素材筛选耗时、后台任务队列长度、容器 RSS 和宿主机可用内存；若 `P2024` 仍持续出现，需要继续定位剩余长事务或连接池等待。
 - 宿主机 `middlewared` 异常内存、无 Swap 等运维风险仍未由代码解决，继续按本文“仍需运维处理的风险”执行。
 
-截至本记录更新时，本轮代码已完成本地验证，尚未合并到 `main`，也尚未部署生产。
+### 发布与真实验收状态
+
+- 本轮代码已合并并部署到生产，当前生产版本为 `f5ea65778489942a7fef92ce48e631b2e7f55d6f`。
+- 生产应用容器运行正常，发布后未发现新的 OOM、容器重启、上传 5xx、Prisma 上传报错或图片预览处理异常。
+- 2026-08-01 用户已使用本人 Edge 浏览器和真实文件完成上传，确认上传成功；此前“更换 Edge 仍失败”的用户问题正式关闭。
+
+## 2026-08-01 第三阶段生产宿主机只读排查
+
+### 当前结论
+
+应用代码修复和真实上传验收已经完成，但生产宿主机仍处于 P1 高风险状态。当前最重要的问题不是新版本应用，而是对象存储路径、TrueNAS 管理服务内存和 MinIO 后台负载。
+
+1. MinIO 配置的数据目录是 `/mnt/tank/apps/minio/data`，但生产的 `/mnt/tank` 并不是数据池挂载点或软链接。该目录实际位于 TrueNAS 启动池数据集 `freenas-boot/ROOT/25.04.2.4/mnt`。
+2. 约 258GB 的 MinIO 素材数据因此实际存放在系统启动盘，而真正的业务数据池是 `/mnt/PoolOne`，当前仍有约 22TB 可用空间。
+3. MinIO 在没有明显公网请求、主应用 CPU 接近空闲时，仍周期性占用约 2300%～2600% CPU。同期磁盘读取和网络流量很低，现有证据高度指向 MinIO 内部后台扫描、对象校验或该版本内部循环；由于尚未取得 scanner/trace 指标，不能把某一个内部任务写成已完全证实的根因。
+4. TrueNAS `middlewared` 约占用 7.23GB 私有匿名内存。连续约 54 分钟采样没有继续快速增长，当前也没有活动管理任务，因此更像历史任务后没有归还给系统的原生内存、缓存或内存碎片；仅凭本次非侵入排查不能断言它仍在持续泄漏。
+5. 宿主机没有 Swap，且历史记录已经确认多次整机 OOM。当前虽然应用正常，但不适合继续进行批量转码、大文件集中上传或其它高负载发布。
+
+### 与应用任务的关系
+
+- 任务详情后台完成、视频转码、考核调研、作品评审、白底图、媒体代理和动态 ZIP 都会访问 MinIO。
+- 排查时任务详情后台队列为空，生产公网 S3 同期请求量也很小，无法解释 MinIO 持续占用二十多个 CPU 核；所以当前不能把持续高 CPU 归因于某个用户上传任务。
+- 公网流量汇总任务是另一个独立性能问题：它每 5 分钟重复扫描约 165 万行日志，并写入约 3.4 万条汇总记录，单轮约 2 分 40 秒。它不调用 MinIO，不是 MinIO 高 CPU 的直接原因，但会额外占用应用、数据库和 ZFS 资源。
+
+### 后续处理顺序与授权边界
+
+1. 低风险代码减负：优化公网流量汇总，限制失败补偿删除、动态 ZIP 和视频上传的资源并发，并补充脱敏后的 S3 操作耗时、字节、重试和并发观测。
+2. 中风险维护：在无活动管理任务的维护窗口重启 `middlewared`，回收约 7GB 内存；这只用于恢复安全余量，不视为永久修复。
+3. 高风险永久修复：在 `PoolOne` 创建专用 MinIO 数据集，先在线预复制，再在维护窗口停止写入、完成最终增量同步和切换。接口地址、桶名和现有凭据保持不变，团队成员不需要修改配置。
+4. 原启动池数据在业务验收完成前保持只读并保留，用于快速回滚；不得在迁移同一阶段直接删除。
+5. 若迁移后 MinIO 仍持续高 CPU，再采集授权后的 scanner/trace/profile 证据，决定是否调整扫描速度或升级 MinIO；不把盲目重启当作根治。
+
+### 已完成但尚未发布的代码减负
+
+- 公网流量 5 分钟汇总首次建立当天基线后，后续按域名分别从上次已看到的日志时间向前保留重叠窗口，只重算最近的完整时间桶；某一个域名同步落后不会被其它域名的进度跳过。
+- 日志文件在扫描期间发生替换、轮转或读取失败时，会放弃该域名当轮的部分结果且不推进其进度点，避免把不完整统计覆盖进数据库。
+- 已成功完成的昨日汇总不再每 5 分钟重复全量扫描；进度保存在现有持久上传卷中，蓝绿容器共用同一份非敏感进度。昨日结果每 1 小时强制重新核验一次，最坏陈旧时间有明确上限，也可通过 `--force-full-scan` 立即重建。
+- 考核调研、作品评审以及考核附件清理统一改为最多 3 个并发 S3 删除；单个对象删除失败仍会继续清理其它对象，并保留原来的逐项成功/失败结果。
+- 没有修改部署脚本、MinIO 配置、接口地址、桶名或凭据。
+
+本地验证：合入最新 `main` 后，全量 Jest 137 个测试套件、883 项测试全部通过；TypeScript 类型检查通过；相关 ESLint 通过；流量汇总脚本 9 项 Python 单元测试通过；`git diff --check` 通过。
+
+本阶段未重启服务、未迁移或删除生产数据、未修改 MinIO/TrueNAS 配置，也未修改部署脚本。上述代码减负仍需单独提交、发布和生产观察后才能标记为线上完成。

lib/__tests__/s3-delete-pool.test.ts

@@ -0,0 +1,75 @@
+/** @jest-environment node */
+
+import { deleteFromS3 } from '@/lib/s3'
+import {
+  S3_COMPENSATION_DELETE_CONCURRENCY,
+  deleteS3ObjectsWithLimit,
+} from '@/lib/s3-delete-pool'
+
+jest.mock('@/lib/s3', () => ({
+  deleteFromS3: jest.fn(),
+}))
+
+const flushPromises = () => new Promise<void>((resolve) => setImmediate(resolve))
+
+describe('deleteS3ObjectsWithLimit', () => {
+  beforeEach(() => {
+    jest.clearAllMocks()
+  })
+
+  test('默认最多同时执行 3 个删除请求', async () => {
+    const releases: Array<() => void> = []
+    let activeCount = 0
+    let maxActiveCount = 0
+
+    ;(deleteFromS3 as jest.Mock).mockImplementation(() => new Promise<void>((resolve) => {
+      activeCount += 1
+      maxActiveCount = Math.max(maxActiveCount, activeCount)
+      releases.push(() => {
+        activeCount -= 1
+        resolve()
+      })
+    }))
+
+    const urls = Array.from(
+      { length: 8 },
+      (_, index) => `https://s3.example/bucket/file-${index + 1}.webp`,
+    )
+    const pending = deleteS3ObjectsWithLimit(urls)
+
+    await flushPromises()
+    expect(deleteFromS3).toHaveBeenCalledTimes(S3_COMPENSATION_DELETE_CONCURRENCY)
+
+    while ((deleteFromS3 as jest.Mock).mock.calls.length < urls.length) {
+      releases.splice(0).forEach((release) => release())
+      await flushPromises()
+    }
+    releases.splice(0).forEach((release) => release())
+
+    const results = await pending
+    expect(maxActiveCount).toBe(S3_COMPENSATION_DELETE_CONCURRENCY)
+    expect(deleteFromS3).toHaveBeenCalledTimes(urls.length)
+    expect(results).toHaveLength(urls.length)
+    expect(results.every((result) => result.status === 'fulfilled')).toBe(true)
+  })
+
+  test('单个删除失败时仍继续清理其余对象并返回 settled 结果', async () => {
+    const failure = new Error('MinIO delete failed')
+    ;(deleteFromS3 as jest.Mock)
+      .mockRejectedValueOnce(failure)
+      .mockResolvedValue(undefined)
+
+    const results = await deleteS3ObjectsWithLimit([
+      'https://s3.example/bucket/failed.webp',
+      'https://s3.example/bucket/success-1.webp',
+      'https://s3.example/bucket/success-2.webp',
+    ])
+
+    expect(deleteFromS3).toHaveBeenCalledTimes(3)
+    expect(results).toEqual([
+      { status: 'rejected', reason: failure },
+      { status: 'fulfilled', value: undefined },
+      { status: 'fulfilled', value: undefined },
+    ])
+  })
+})

lib/performance-survey-service.ts

@@ -87,7 +87,8 @@ import {
   validatePerformanceSurveyProcessedImages,
   type PerformanceSurveyProcessedImageDeclaration,
 } from '@/lib/performance-survey-processed-images'
-import { deleteFromS3, uploadFileToS3, uploadToS3 } from '@/lib/s3'
+import { uploadFileToS3, uploadToS3 } from '@/lib/s3'
+import { deleteS3ObjectsWithLimit } from '@/lib/s3-delete-pool'
 import {
   deleteOwnedUploadSession,
   getOwnedUploadSession,
@@ -2701,9 +2702,7 @@ export async function removePerformanceSurveyPublicDashboardAssessment(
   })
 
   clearPerformanceSurveyDepartmentCache()
-  const attachmentCleanup = await Promise.allSettled(
-    result.attachmentUrls.map((url) => deleteFromS3(url)),
-  )
+  const attachmentCleanup = await deleteS3ObjectsWithLimit(result.attachmentUrls)
   const failedAttachmentCount = attachmentCleanup.filter((item) => item.status === 'rejected').length
   if (failedAttachmentCount > 0) {
     console.error(
@@ -4092,7 +4091,7 @@ async function createPerformanceSurveyWorkItemFromProcessedUploads(
       submittedAt: created.submittedAt,
     }
   } catch (error) {
-    await Promise.allSettled(uploadedUrls.map((url) => deleteFromS3(url)))
+    await deleteS3ObjectsWithLimit(uploadedUrls)
     if (error instanceof PerformanceSurveyProcessedImageError) {
       throw new PerformanceSurveyApiError(error.message)
     }
@@ -4277,7 +4276,7 @@ async function createPerformanceSurveyWorkItemFromSources(
       submittedAt: created.submittedAt,
     }
   } catch (error) {
-    await Promise.allSettled(uploadedUrls.map((url) => deleteFromS3(url)))
+    await deleteS3ObjectsWithLimit(uploadedUrls)
     if (error instanceof PerformanceSurveyImageProcessingError) {
       throw new PerformanceSurveyApiError(error.message)
     }

lib/s3-delete-pool.ts

@@ -0,0 +1,40 @@
+import { deleteFromS3 } from '@/lib/s3'
+
+export const S3_COMPENSATION_DELETE_CONCURRENCY = 3
+
+/**
+ * 失败补偿需要尽量删除所有已上传对象，但不能一次向 MinIO 发出无限量请求。
+ * 返回值保持 Promise.allSettled 的语义：单个删除失败不会阻断其余对象清理。
+ */
+export async function deleteS3ObjectsWithLimit(
+  fileUrls: string[],
+  concurrency = S3_COMPENSATION_DELETE_CONCURRENCY,
+): Promise<PromiseSettledResult<void>[]> {
+  if (fileUrls.length === 0) {
+    return []
+  }
+
+  const normalizedConcurrency = Number.isFinite(concurrency)
+    ? Math.max(1, Math.floor(concurrency))
+    : S3_COMPENSATION_DELETE_CONCURRENCY
+  const workerCount = Math.min(normalizedConcurrency, fileUrls.length)
+  const results = new Array<PromiseSettledResult<void>>(fileUrls.length)
+  let nextIndex = 0
+
+  const worker = async () => {
+    while (nextIndex < fileUrls.length) {
+      const index = nextIndex
+      nextIndex += 1
+
+      try {
+        await deleteFromS3(fileUrls[index])
+        results[index] = { status: 'fulfilled', value: undefined }
+      } catch (reason) {
+        results[index] = { status: 'rejected', reason }
+      }
+    }
+  }
+
+  await Promise.all(Array.from({ length: workerCount }, () => worker()))
+  return results
+}

lib/work-review-service.ts

@@ -26,7 +26,8 @@ import {
   validatePerformanceSurveyProcessedImages,
   type PerformanceSurveyProcessedImageDeclaration,
 } from '@/lib/performance-survey-processed-images'
-import { deleteFromS3, uploadFileToS3 } from '@/lib/s3'
+import { uploadFileToS3 } from '@/lib/s3'
+import { deleteS3ObjectsWithLimit } from '@/lib/s3-delete-pool'
 import {
   deleteOwnedUploadSession,
   getOwnedUploadedFile,
@@ -715,7 +716,7 @@ async function withStoredImages<T>(input: {
 
     return await input.task(imageGroups)
   } catch (error) {
-    await Promise.allSettled(uploadedUrls.map((url) => deleteFromS3(url)))
+    await deleteS3ObjectsWithLimit(uploadedUrls)
     throw error
   } finally {
     await cleanupProcessedUploads(input.userId, uploads)

scripts/__tests__/test_public_traffic_rollup.py

@@ -0,0 +1,283 @@
+import importlib.util
+import io
+import os
+import tempfile
+import unittest
+from contextlib import redirect_stdout
+from datetime import datetime, timedelta
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+
+SCRIPT_PATH = Path(__file__).resolve().parents[1] / "public-traffic-rollup-from-nginx.py"
+SPEC = importlib.util.spec_from_file_location("public_traffic_rollup", SCRIPT_PATH)
+ROLLUP = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(ROLLUP)
+
+
+def nginx_line(timestamp, path="/asset.webp", status=200, body_bytes=100):
+    rendered = timestamp.strftime("%d/%b/%Y:%H:%M:%S %z")
+    return (
+        f'127.0.0.1 - - [{rendered}] "GET {path} HTTP/1.1" '
+        f'{status} {body_bytes} "-" "test" request_length=42 bytes_sent={body_bytes}\n'
+    )
+
+
+class PublicTrafficRollupTest(unittest.TestCase):
+    def make_args(self, log_dir, state_dir, overlap=15):
+        return SimpleNamespace(
+            log_dir=str(log_dir),
+            domains="sc.jiansun.vip",
+            granularity="5m",
+            source_server="test-server",
+            state_dir=str(state_dir),
+            recent_overlap_minutes=overlap,
+        )
+
+    def test_reverse_reader_preserves_complete_lines(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "sample.log"
+            expected = [f"line-{index}-" + ("x" * 47) for index in range(250)]
+            path.write_text("\n".join(expected) + "\n", encoding="utf-8")
+
+            actual = [line.rstrip("\n") for line in ROLLUP.iter_reverse_lines(path, block_size=113)]
+
+            self.assertEqual(actual, list(reversed(expected)))
+
+    def test_incremental_scan_reads_only_overlap_and_keeps_complete_buckets(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            log_path = root / "sc.jiansun.vip.access.log"
+            timestamps = [
+                datetime(2026, 8, 1, 9, 0, tzinfo=ROLLUP.CHINA_TZ),
+                datetime(2026, 8, 1, 10, 10, tzinfo=ROLLUP.CHINA_TZ),
+                datetime(2026, 8, 1, 10, 14, tzinfo=ROLLUP.CHINA_TZ),
+                datetime(2026, 8, 1, 10, 17, tzinfo=ROLLUP.CHINA_TZ),
+                datetime(2026, 8, 1, 10, 19, tzinfo=ROLLUP.CHINA_TZ),
+            ]
+            log_path.write_text("".join(nginx_line(value) for value in timestamps), encoding="utf-8")
+            os.utime(log_path, (timestamps[-1].timestamp(), timestamps[-1].timestamp()))
+            args = self.make_args(root, root / "state", overlap=5)
+            business_date = timestamps[-1].date()
+            state = {
+                "lastLogTimestamps": {
+                    "sc.jiansun.vip": ROLLUP.iso_utc(timestamps[-1]),
+                },
+            }
+
+            recent_starts = ROLLUP.incremental_starts(
+                args,
+                state,
+                business_date,
+                business_date,
+                ("sc.jiansun.vip",),
+            )
+            rows, scanned, matched, _, newest, _, complete = ROLLUP.aggregate(
+                args,
+                business_date,
+                business_date,
+                recent_since_by_domain=recent_starts,
+            )
+
+            self.assertEqual(
+                recent_starts["sc.jiansun.vip"],
+                datetime(2026, 8, 1, 10, 10, tzinfo=ROLLUP.CHINA_TZ),
+            )
+            self.assertEqual(matched, 4)
+            self.assertEqual(sum(row["totalRequests"] for row in rows), 4)
+            self.assertLess(scanned, len(timestamps) + 1)
+            self.assertEqual(newest["sc.jiansun.vip"], timestamps[-1])
+            self.assertTrue(complete["sc.jiansun.vip"])
+
+    def test_incremental_scan_tolerates_a_physically_late_older_log_line(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            log_path = root / "sc.jiansun.vip.access.log"
+            newer = [
+                datetime(2026, 8, 1, 10, 16, tzinfo=ROLLUP.CHINA_TZ),
+                datetime(2026, 8, 1, 10, 17, tzinfo=ROLLUP.CHINA_TZ),
+            ]
+            physically_late_but_old = datetime(2026, 8, 1, 9, 0, tzinfo=ROLLUP.CHINA_TZ)
+            log_path.write_text(
+                "".join(nginx_line(value) for value in [*newer, physically_late_but_old]),
+                encoding="utf-8",
+            )
+            os.utime(log_path, (newer[-1].timestamp(), newer[-1].timestamp()))
+            args = self.make_args(root, root / "state", overlap=5)
+            business_date = newer[-1].date()
+
+            rows, _, matched, _, newest, _, complete = ROLLUP.aggregate(
+                args,
+                business_date,
+                business_date,
+                recent_since_by_domain={
+                    "sc.jiansun.vip": datetime(
+                        2026, 8, 1, 10, 10, tzinfo=ROLLUP.CHINA_TZ,
+                    ),
+                },
+            )
+
+            self.assertEqual(matched, 2)
+            self.assertEqual(sum(row["totalRequests"] for row in rows), 2)
+            self.assertEqual(newest["sc.jiansun.vip"], newer[-1])
+            self.assertTrue(complete["sc.jiansun.vip"])
+
+    def test_state_round_trip_is_atomic_and_reusable(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            path = root / "state" / "checkpoint.json"
+            expected = {"lastLogTimestamp": "2026-08-01T02:19:00Z", "completed": True}
+
+            ROLLUP.save_state(path, expected)
+
+            self.assertEqual(ROLLUP.load_state(path), expected)
+            self.assertEqual(list(path.parent.glob(f".{path.name}.*")), [])
+
+    def test_checkpoint_key_is_stable_across_blue_green_container_names(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            business_date = datetime(2026, 8, 1, tzinfo=ROLLUP.CHINA_TZ).date()
+            args_blue = self.make_args(root, root / "state")
+            args_green = self.make_args(root, root / "state")
+            args_blue.source_server = "blue-container-id"
+            args_green.source_server = "green-container-id"
+
+            blue_path = ROLLUP.state_path(
+                args_blue,
+                business_date,
+                business_date,
+                ("sc.jiansun.vip",),
+            )
+            green_path = ROLLUP.state_path(
+                args_green,
+                business_date,
+                business_date,
+                ("sc.jiansun.vip",),
+            )
+
+            self.assertEqual(blue_path, green_path)
+
+    def test_completed_day_is_not_rescanned_every_five_minutes(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            state_dir = root / "state"
+            business_date = datetime.now(ROLLUP.CHINA_TZ).date() - timedelta(days=1)
+            timestamp = datetime.combine(
+                business_date,
+                datetime.min.time(),
+                tzinfo=ROLLUP.CHINA_TZ,
+            ).replace(hour=12)
+            after_day = timestamp + timedelta(hours=13)
+            log_path = root / "sc.jiansun.vip.access.log"
+            log_path.write_text(
+                nginx_line(timestamp) + nginx_line(after_day),
+                encoding="utf-8",
+            )
+            argv = [
+                str(SCRIPT_PATH),
+                "--granularity", "day",
+                "--start-date", business_date.isoformat(),
+                "--end-date", business_date.isoformat(),
+                "--log-dir", str(root),
+                "--state-dir", str(state_dir),
+                "--domains", "sc.jiansun.vip",
+                "--token-env", "TEST_ROLLUP_TOKEN",
+                "--source-server", "test-server",
+            ]
+
+            with patch.dict(os.environ, {"TEST_ROLLUP_TOKEN": "test-token"}), \
+                    patch.object(ROLLUP.sys, "argv", argv), \
+                    patch.object(ROLLUP, "post_rows", return_value=1) as post_rows:
+                first_output = io.StringIO()
+                with redirect_stdout(first_output):
+                    ROLLUP.main()
+                second_output = io.StringIO()
+                with redirect_stdout(second_output):
+                    ROLLUP.main()
+
+            first_result = ROLLUP.json.loads(first_output.getvalue())
+            second_result = ROLLUP.json.loads(second_output.getvalue())
+            self.assertEqual(first_result["scanMode"], "full")
+            self.assertEqual(first_result["matchedLines"], 1)
+            self.assertTrue(second_result["skipped"])
+            self.assertEqual(second_result["scanMode"], "cached")
+            post_rows.assert_called_once()
+
+    def test_domain_checkpoint_does_not_follow_a_faster_domain(self):
+        args = SimpleNamespace(granularity="5m", recent_overlap_minutes=15)
+        business_date = datetime(2026, 8, 1, tzinfo=ROLLUP.CHINA_TZ).date()
+        state = {
+            "lastLogTimestamps": {
+                "sc.jiansun.vip": "2026-08-01T02:20:00Z",
+                "s3.jiansun.vip": "2026-08-01T01:00:00Z",
+            },
+        }
+
+        starts = ROLLUP.incremental_starts(
+            args,
+            state,
+            business_date,
+            business_date,
+            ("sc.jiansun.vip", "s3.jiansun.vip", "cp.jiansun.vip"),
+        )
+
+        self.assertEqual(
+            starts["sc.jiansun.vip"],
+            datetime(2026, 8, 1, 10, 5, tzinfo=ROLLUP.CHINA_TZ),
+        )
+        self.assertEqual(
+            starts["s3.jiansun.vip"],
+            datetime(2026, 8, 1, 8, 45, tzinfo=ROLLUP.CHINA_TZ),
+        )
+        self.assertIsNone(starts["cp.jiansun.vip"])
+
+    def test_day_cache_has_a_bounded_revalidation_window(self):
+        now = datetime(2026, 8, 2, 12, 0, tzinfo=ROLLUP.CHINA_TZ)
+        args = SimpleNamespace(granularity="day", day_cache_hours=1.0)
+        end_date = now.date() - timedelta(days=1)
+
+        self.assertTrue(ROLLUP.is_fresh_day_cache(
+            args,
+            {"completedAt": ROLLUP.iso_utc(now - timedelta(minutes=30))},
+            end_date,
+            now,
+        ))
+        self.assertFalse(ROLLUP.is_fresh_day_cache(
+            args,
+            {"completedAt": ROLLUP.iso_utc(now - timedelta(hours=2))},
+            end_date,
+            now,
+        ))
+
+    def test_changed_source_file_drops_partial_domain_rows(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            timestamp = datetime(2026, 8, 1, 10, 0, tzinfo=ROLLUP.CHINA_TZ)
+            log_path = root / "sc.jiansun.vip.access.log"
+            log_path.write_text(nginx_line(timestamp), encoding="utf-8")
+            args = self.make_args(root, root / "state")
+            initial = {"sc.jiansun.vip": ((str(log_path), 100, 1),)}
+            changed = {"sc.jiansun.vip": ((str(log_path), 200, 2),)}
+
+            with patch.object(
+                ROLLUP,
+                "snapshot_domain_paths",
+                side_effect=[
+                    (initial, {"sc.jiansun.vip": True}),
+                    (changed, {"sc.jiansun.vip": True}),
+                ],
+            ):
+                rows, _, matched, _, _, _, complete = ROLLUP.aggregate(
+                    args,
+                    timestamp.date(),
+                    timestamp.date(),
+                )
+
+            self.assertEqual(rows, [])
+            self.assertEqual(matched, 0)
+            self.assertFalse(complete["sc.jiansun.vip"])
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/public-traffic-rollup-from-nginx.py

@@ -1,11 +1,13 @@
 #!/usr/bin/env python3
 import argparse
 import gzip
+import hashlib
 import json
 import os
 import re
 import socket
 import sys
+import tempfile
 import urllib.error
 import urllib.parse
 import urllib.request
@@ -16,6 +18,8 @@ from pathlib import Path
 
 CHINA_TZ = timezone(timedelta(hours=8))
 DOMAINS = ("sc.jiansun.vip", "s3.jiansun.vip", "cp.jiansun.vip")
+RECENT_STOP_OLDER_LINES = 2048
+STATE_VERSION = 2
 LOG_RE = re.compile(
     r'^(?P<remote>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] '
     r'"(?P<method>[A-Z]+) (?P<target>\S+) HTTP/[^"]+" '
@@ -34,9 +38,41 @@ def parse_args():
     parser.add_argument("--source-server", default=socket.gethostname())
     parser.add_argument("--batch-size", type=int, default=1000)
     parser.add_argument("--domains", default=",".join(DOMAINS))
+    parser.add_argument(
+        "--state-dir",
+        default=default_state_dir(),
+    )
+    parser.add_argument(
+        "--recent-overlap-minutes",
+        type=int,
+        default=int(os.environ.get("PUBLIC_TRAFFIC_ROLLUP_OVERLAP_MINUTES", "15")),
+        help="Overlap used by incremental 5-minute rollups.",
+    )
+    parser.add_argument(
+        "--force-full-scan",
+        action="store_true",
+        help="Ignore local checkpoints and rebuild the requested range.",
+    )
+    parser.add_argument(
+        "--day-cache-hours",
+        type=float,
+        default=float(os.environ.get("PUBLIC_TRAFFIC_ROLLUP_DAY_CACHE_HOURS", "1")),
+        help="Revalidate completed day rollups after this many hours.",
+    )
     return parser.parse_args()
 
 
+def default_state_dir():
+    configured = os.environ.get("PUBLIC_TRAFFIC_ROLLUP_STATE_DIR", "").strip()
+    if configured:
+        return configured
+
+    persistent_uploads = Path("/app/uploads")
+    if persistent_uploads.is_dir() and os.access(persistent_uploads, os.W_OK):
+        return str(persistent_uploads / ".public-traffic-rollup")
+    return "/tmp/public-traffic-rollup"
+
+
 def parse_date(value):
     return datetime.strptime(value, "%Y-%m-%d").date()
 
@@ -63,6 +99,29 @@ def open_log(path):
     return open(path, "rt", encoding="utf-8", errors="replace")
 
 
+def iter_reverse_lines(path, block_size=64 * 1024):
+    """Yield an uncompressed log from newest line to oldest without loading it all."""
+    with open(path, "rb") as handle:
+        handle.seek(0, os.SEEK_END)
+        position = handle.tell()
+        remainder = b""
+
+        while position > 0:
+            read_size = min(block_size, position)
+            position -= read_size
+            handle.seek(position)
+            block = handle.read(read_size) + remainder
+            parts = block.split(b"\n")
+            remainder = parts[0]
+            for raw_line in reversed(parts[1:]):
+                raw_line = raw_line.removesuffix(b"\r")
+                if raw_line:
+                    yield raw_line.decode("utf-8", errors="replace") + "\n"
+
+        if remainder:
+            yield remainder.removesuffix(b"\r").decode("utf-8", errors="replace")
+
+
 def infer_domain(path):
     name = Path(path).name
     for domain in DOMAINS:
@@ -151,75 +210,246 @@ def iso_utc(value):
     return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
 
 
-def aggregate(args):
-    domains = tuple(item.strip() for item in args.domains.split(",") if item.strip())
+def resolve_range(args):
     start_default, end_default = default_dates(args.granularity)
     start_date = parse_date(args.start_date) if args.start_date else start_default
     end_date = parse_date(args.end_date) if args.end_date else end_default
+    return start_date, end_date
+
+
+def floor_five_minutes(value):
+    return value.replace(minute=(value.minute // 5) * 5, second=0, microsecond=0)
+
+
+def state_path(args, start_date, end_date, domains):
+    identity = json.dumps(
+        {
+            "stateVersion": STATE_VERSION,
+            "granularity": args.granularity,
+            "startDate": start_date.isoformat(),
+            "endDate": end_date.isoformat(),
+            "domains": domains,
+            "logDir": str(Path(args.log_dir).resolve()),
+        },
+        sort_keys=True,
+    ).encode("utf-8")
+    digest = hashlib.sha256(identity).hexdigest()[:20]
+    return Path(args.state_dir) / f"{args.granularity}-{digest}.json"
+
+
+def load_state(path):
+    try:
+        with open(path, "rt", encoding="utf-8") as handle:
+            value = json.load(handle)
+        return value if isinstance(value, dict) else {}
+    except (OSError, ValueError, TypeError):
+        return {}
+
+
+def save_state(path, value):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
+    try:
+        with os.fdopen(descriptor, "wt", encoding="utf-8") as handle:
+            json.dump(value, handle, ensure_ascii=False, sort_keys=True)
+            handle.write("\n")
+        os.replace(temporary_name, path)
+    except Exception:
+        try:
+            os.unlink(temporary_name)
+        except OSError:
+            pass
+        raise
+
+
+def parse_state_timestamp(value):
+    if not isinstance(value, str) or not value:
+        return None
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(CHINA_TZ)
+    except ValueError:
+        return None
+
+
+def incremental_starts(args, state, start_date, end_date, domains):
+    if args.granularity != "5m" or start_date != end_date:
+        return {domain: None for domain in domains}
+
+    saved_timestamps = state.get("lastLogTimestamps")
+    if not isinstance(saved_timestamps, dict):
+        saved_timestamps = {}
+
+    overlap = timedelta(minutes=max(5, args.recent_overlap_minutes))
+    day_start = datetime.combine(start_date, datetime.min.time(), tzinfo=CHINA_TZ)
+    starts = {}
+    for domain in domains:
+        last_timestamp = parse_state_timestamp(saved_timestamps.get(domain))
+        if not last_timestamp or last_timestamp.date() != start_date:
+            starts[domain] = None
+            continue
+        starts[domain] = max(day_start, floor_five_minutes(last_timestamp - overlap))
+    return starts
+
+
+def path_signature(path):
+    stat = path.stat()
+    return (str(path), stat.st_size, stat.st_mtime_ns)
+
+
+def collect_domain_paths(log_dir, domains):
+    paths_by_domain = {domain: [] for domain in domains}
+    for path in iter_log_paths(log_dir, domains):
+        domain = infer_domain(path)
+        if domain in paths_by_domain:
+            paths_by_domain[domain].append(path)
+    return paths_by_domain
+
+
+def snapshot_domain_paths(paths_by_domain):
+    snapshots = {}
+    complete = {}
+    for domain, paths in paths_by_domain.items():
+        signatures = []
+        domain_complete = bool(paths)
+        for path in paths:
+            try:
+                signatures.append(path_signature(path))
+            except OSError:
+                domain_complete = False
+        snapshots[domain] = tuple(signatures)
+        complete[domain] = domain_complete and len(signatures) == len(paths)
+    return snapshots, complete
+
+
+def aggregate(args, start_date, end_date, recent_since_by_domain=None):
+    domains = tuple(item.strip() for item in args.domains.split(",") if item.strip())
+    if recent_since_by_domain is None:
+        recent_since_by_domain = {domain: None for domain in domains}
 
     rows = {}
     scanned_lines = 0
     matched_lines = 0
-    for path in iter_log_paths(args.log_dir, domains):
-        domain = infer_domain(path)
-        try:
-            handle = open_log(path)
-        except OSError as exc:
-            print(f"WARN: cannot open {path}: {exc}", file=sys.stderr)
-            continue
-        with handle:
-            for line in handle:
-                scanned_lines += 1
-                entry = parse_line(line, domain)
-                if not entry:
+    matched_lines_by_domain = {domain: 0 for domain in domains}
+    max_log_timestamps = {domain: None for domain in domains}
+    max_seen_timestamps = {domain: None for domain in domains}
+    paths_by_domain = collect_domain_paths(args.log_dir, domains)
+    initial_snapshots, domain_complete = snapshot_domain_paths(paths_by_domain)
+
+    for domain, domain_paths in paths_by_domain.items():
+        recent_since = recent_since_by_domain.get(domain)
+        for path in domain_paths:
+            reverse_order = recent_since is not None and not str(path).endswith(".gz")
+            consecutive_older_lines = 0
+            try:
+                if recent_since is not None and path.stat().st_mtime < recent_since.timestamp():
                     continue
-                entry_date = parse_date(entry["businessDate"])
-                if entry_date < start_date or entry_date > end_date:
-                    continue
-                matched_lines += 1
-                bucket_start, bucket_end = bucket_for(entry["timestamp"], args.granularity)
-                key = (
-                    iso_utc(bucket_start),
-                    iso_utc(bucket_end),
-                    entry["businessDate"],
-                    args.granularity,
-                    entry["domain"],
-                    entry["method"],
-                    entry["path"],
-                )
-                row = rows.setdefault(
-                    key,
-                    {
-                        "bucketStart": key[0],
-                        "bucketEnd": key[1],
-                        "businessDate": key[2],
-                        "granularity": key[3],
-                        "domain": key[4],
-                        "method": key[5],
-                        "path": key[6],
-                        "responseBytes": 0,
-                        "requestBytes": None,
-                        "requestBytesKnownCount": 0,
-                        "totalRequests": 0,
-                        "statusCounts": defaultdict(int),
-                        "sourceServer": args.source_server,
-                        "generatedAt": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
-                        "frozen": args.granularity == "day",
-                    },
-                )
-                row["responseBytes"] += entry["responseBytes"]
-                row["totalRequests"] += 1
-                row["statusCounts"][entry["statusBucket"]] += 1
-                if entry["requestBytes"] is not None:
-                    row["requestBytes"] = (row["requestBytes"] or 0) + entry["requestBytes"]
-                    row["requestBytesKnownCount"] += 1
+                if reverse_order:
+                    lines = iter_reverse_lines(path)
+                    handle = None
+                else:
+                    handle = open_log(path)
+                    lines = handle
+                try:
+                    for line in lines:
+                        scanned_lines += 1
+                        entry = parse_line(line, domain)
+                        if not entry:
+                            continue
+                        timestamp = entry["timestamp"]
+                        current_max_seen = max_seen_timestamps[domain]
+                        if current_max_seen is None or timestamp > current_max_seen:
+                            max_seen_timestamps[domain] = timestamp
+                        if recent_since is not None and timestamp < recent_since:
+                            if reverse_order:
+                                consecutive_older_lines += 1
+                                if consecutive_older_lines >= RECENT_STOP_OLDER_LINES:
+                                    break
+                            continue
+                        consecutive_older_lines = 0
+                        entry_date = timestamp.date()
+                        if entry_date < start_date or entry_date > end_date:
+                            continue
+                        matched_lines += 1
+                        matched_lines_by_domain[domain] += 1
+                        current_max_log = max_log_timestamps[domain]
+                        if current_max_log is None or timestamp > current_max_log:
+                            max_log_timestamps[domain] = timestamp
+                        bucket_start, bucket_end = bucket_for(timestamp, args.granularity)
+                        key = (
+                            iso_utc(bucket_start),
+                            iso_utc(bucket_end),
+                            entry["businessDate"],
+                            args.granularity,
+                            entry["domain"],
+                            entry["method"],
+                            entry["path"],
+                        )
+                        row = rows.setdefault(
+                            key,
+                            {
+                                "bucketStart": key[0],
+                                "bucketEnd": key[1],
+                                "businessDate": key[2],
+                                "granularity": key[3],
+                                "domain": key[4],
+                                "method": key[5],
+                                "path": key[6],
+                                "responseBytes": 0,
+                                "requestBytes": None,
+                                "requestBytesKnownCount": 0,
+                                "totalRequests": 0,
+                                "statusCounts": defaultdict(int),
+                                "sourceServer": args.source_server,
+                                "generatedAt": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
+                                "frozen": args.granularity == "day",
+                            },
+                        )
+                        row["responseBytes"] += entry["responseBytes"]
+                        row["totalRequests"] += 1
+                        row["statusCounts"][entry["statusBucket"]] += 1
+                        if entry["requestBytes"] is not None:
+                            row["requestBytes"] = (row["requestBytes"] or 0) + entry["requestBytes"]
+                            row["requestBytesKnownCount"] += 1
+                finally:
+                    if handle is not None:
+                        handle.close()
+            except OSError as exc:
+                print(f"WARN: cannot open {path}: {exc}", file=sys.stderr)
+                domain_complete[domain] = False
+                continue
+
+    final_paths_by_domain = collect_domain_paths(args.log_dir, domains)
+    final_snapshots, final_complete = snapshot_domain_paths(final_paths_by_domain)
+    for domain in domains:
+        if not final_complete[domain] or initial_snapshots[domain] != final_snapshots[domain]:
+            domain_complete[domain] = False
+
+    if not all(domain_complete.values()):
+        rows = {
+            key: row
+            for key, row in rows.items()
+            if domain_complete.get(row["domain"], False)
+        }
+        matched_lines = sum(
+            count
+            for domain, count in matched_lines_by_domain.items()
+            if domain_complete.get(domain, False)
+        )
 
     normalized_rows = []
     for row in rows.values():
         row["statusCounts"] = dict(row["statusCounts"])
         normalized_rows.append(row)
 
-    return normalized_rows, scanned_lines, matched_lines
+    return (
+        normalized_rows,
+        scanned_lines,
+        matched_lines,
+        matched_lines_by_domain,
+        max_log_timestamps,
+        max_seen_timestamps,
+        domain_complete,
+    )
 
 
 def post_rows(endpoint, token, rows, batch_size):
@@ -246,18 +476,109 @@ def post_rows(endpoint, token, rows, batch_size):
     return accepted
 
 
+def is_fresh_day_cache(args, state, end_date, now):
+    if args.granularity != "day" or end_date >= now.date():
+        return False
+    completed_at = parse_state_timestamp(state.get("completedAt"))
+    if completed_at is None:
+        return False
+    cache_lifetime = timedelta(hours=max(0.0, args.day_cache_hours))
+    cache_age = now - completed_at
+    return timedelta(0) <= cache_age < cache_lifetime
+
+
 def main():
     args = parse_args()
     token = os.environ.get(args.token_env, "").strip()
     if not token:
         raise SystemExit(f"Missing token env: {args.token_env}")
 
-    rows, scanned_lines, matched_lines = aggregate(args)
+    domains = tuple(item.strip() for item in args.domains.split(",") if item.strip())
+    start_date, end_date = resolve_range(args)
+    checkpoint_path = state_path(args, start_date, end_date, domains)
+    state = load_state(checkpoint_path)
+
+    now = datetime.now(CHINA_TZ)
+    if not args.force_full_scan and is_fresh_day_cache(args, state, end_date, now):
+        print(json.dumps({
+            "ok": True,
+            "granularity": args.granularity,
+            "rows": 0,
+            "accepted": 0,
+            "scannedLines": 0,
+            "matchedLines": 0,
+            "scanMode": "cached",
+            "skipped": True,
+        }, ensure_ascii=False))
+        return
+
+    if args.force_full_scan:
+        recent_since_by_domain = {domain: None for domain in domains}
+    else:
+        recent_since_by_domain = incremental_starts(args, state, start_date, end_date, domains)
+
+    (
+        rows,
+        scanned_lines,
+        matched_lines,
+        matched_lines_by_domain,
+        max_log_timestamps,
+        max_seen_timestamps,
+        domain_complete,
+    ) = aggregate(
+        args,
+        start_date,
+        end_date,
+        recent_since_by_domain=recent_since_by_domain,
+    )
     if rows:
         accepted = post_rows(args.endpoint, token, rows, max(1, args.batch_size))
     else:
         accepted = 0
 
+    next_state = dict(state)
+    saved_timestamps = next_state.get("lastLogTimestamps")
+    if not isinstance(saved_timestamps, dict):
+        saved_timestamps = {}
+    else:
+        saved_timestamps = dict(saved_timestamps)
+    for domain in domains:
+        max_timestamp = max_log_timestamps.get(domain)
+        if domain_complete.get(domain, False) and max_timestamp is not None:
+            saved_timestamps[domain] = iso_utc(max_timestamp)
+    if saved_timestamps:
+        next_state["lastLogTimestamps"] = saved_timestamps
+
+    if args.granularity == "day" and end_date < now.date():
+        day_end = datetime.combine(
+            end_date + timedelta(days=1),
+            datetime.min.time(),
+            tzinfo=CHINA_TZ,
+        )
+        day_is_fully_observed = all(
+            domain_complete.get(domain, False)
+            and max_seen_timestamps.get(domain) is not None
+            and max_seen_timestamps[domain] >= day_end
+            for domain in domains
+        )
+        if day_is_fully_observed:
+            next_state["completedAt"] = iso_utc(now)
+        else:
+            next_state.pop("completedAt", None)
+    if next_state != state:
+        save_state(checkpoint_path, next_state)
+
+    incremental_domain_count = sum(
+        recent_since is not None
+        for recent_since in recent_since_by_domain.values()
+    )
+    if incremental_domain_count == 0:
+        scan_mode = "full"
+    elif incremental_domain_count == len(domains):
+        scan_mode = "incremental"
+    else:
+        scan_mode = "mixed"
+
     print(json.dumps({
         "ok": True,
         "granularity": args.granularity,
@@ -265,6 +586,13 @@ def main():
         "accepted": accepted,
         "scannedLines": scanned_lines,
         "matchedLines": matched_lines,
+        "matchedLinesByDomain": matched_lines_by_domain,
+        "domainComplete": domain_complete,
+        "scanMode": scan_mode,
+        "recentSinceByDomain": {
+            domain: iso_utc(recent_since) if recent_since is not None else None
+            for domain, recent_since in recent_since_by_domain.items()
+        },
     }, ensure_ascii=False))