gao/ljwx-shiti
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sync): add night-window worker mode and 3am nightly e2e job

Browse Source
This commit is contained in:
bruno
2026-02-28 21:49:12 +08:00
parent 664ab7319b
commit 8da293bc8b
8 changed files with 172 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/build-and-notify.yml
+6
View File
@@ -1,6 +1,12 @@
name: build-and-notify
on:
workflow_dispatch:
inputs:
branch:
description: "Branch to run against"
required: false
default: "master"
push:
branches:
- main
docs/acceptance-checklist.md
+1
View File
@@ -65,6 +65,7 @@ bash scripts/ops/check-sync-autopr.sh
- [ ] 网络故障进入 `FAILED_RETRYABLE`，并写入 `next_retry_at`
- [ ] 认证/镜像不存在进入 `FAILED_FATAL`
- [ ] 重试超过 `MAX_RETRIES` 后转 `FAILED_FATAL`
- [ ] 若启用 `SYNC_WINDOW_ENABLED=true`，窗口外不拉取 GHCR，仅等待窗口内执行
## E. Digest 一致性
docs/architecture.md
+1
View File
@@ -75,6 +75,7 @@ Webhook 必填字段：
- MVP：单机 sqlite + 单 worker
- 稳定期：systemd 或 compose 常驻
- 扩展期：迁移 PostgreSQL、支持多 worker、引入分布式锁与多 Harbor 目标
- 若 GHCR 日间链路慢：启用夜间同步窗口（`SYNC_WINDOW_*`），白天仅入队，夜间集中执行同步
## 9. 多服务器统一代码
scripts/nightly-e2e.sh
Executable
+61
View File
@@ -0,0 +1,61 @@
#!/usr/bin/env bash
set -euo pipefail
REPO="${REPO:-BrunoGao/ljwx-shiti}"
WORKFLOW="${WORKFLOW:-build-and-notify.yml}"
BRANCH="${BRANCH:-master}"
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
ROOT_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)"
cd "${ROOT_DIR}"
REPORT_PATH="${REPORT_PATH:-/tmp/ljwx-shiti-e2e-nightly-$(date +%Y%m%d-%H%M%S).md}"
log() {
echo "[$(date +'%Y-%m-%d %H:%M:%S')] $*"
}
require_cmd() {
local cmd="$1"
if ! command -v "$cmd" >/dev/null 2>&1; then
log "missing command: ${cmd}"
exit 1
fi
}
require_cmd gh
require_cmd jq
log "trigger workflow ${WORKFLOW} repo=${REPO} branch=${BRANCH}"
gh workflow run "${WORKFLOW}" --repo "${REPO}" --ref "${BRANCH}"
sleep 8
RUN_ID="$(
gh run list \
--repo "${REPO}" \
--workflow "${WORKFLOW}" \
--branch "${BRANCH}" \
--limit 10 \
--json databaseId,status,createdAt \
| jq -r 'map(select(.status == "queued" or .status == "in_progress")) | .[0].databaseId // empty'
)"
if [[ -z "${RUN_ID}" ]]; then
log "failed to resolve new workflow run id"
exit 1
fi
log "watch workflow run=${RUN_ID}"
gh run watch "${RUN_ID}" --repo "${REPO}" --exit-status
log "run full e2e verification run_id=${RUN_ID}"
"${ROOT_DIR}/scripts/verify-e2e-deploy.sh" \
--repo "${REPO}" \
--workflow "${WORKFLOW}" \
--branch "${BRANCH}" \
--run-id "${RUN_ID}" \
--report "${REPORT_PATH}" \
--require-sync-logs
log "nightly e2e report: ${REPORT_PATH}"
sync-service/.env.example
+5
View File
@@ -29,6 +29,11 @@ WORKER_POLL_SECONDS=3
MAX_RETRIES=6
RETRY_BASE_SECONDS=10
RETRY_MAX_SECONDS=600
SYNC_WINDOW_ENABLED=false
SYNC_WINDOW_TIMEZONE=UTC
SYNC_WINDOW_START=00:00
SYNC_WINDOW_END=06:00
SYNC_WINDOW_LOG_INTERVAL_SECONDS=300
# Deploy Auto PR (optional)
DEPLOY_AUTOPR_ENABLED=false
sync-service/app/config.py
+8
View File
@@ -39,6 +39,14 @@ class Settings(BaseSettings):
max_retries: int = Field(default=6, validation_alias="MAX_RETRIES")
retry_base_seconds: int = Field(default=10, validation_alias="RETRY_BASE_SECONDS")
retry_max_seconds: int = Field(default=600, validation_alias="RETRY_MAX_SECONDS")
sync_window_enabled: bool = Field(default=False, validation_alias="SYNC_WINDOW_ENABLED")
sync_window_timezone: str = Field(default="UTC", validation_alias="SYNC_WINDOW_TIMEZONE")
sync_window_start: str = Field(default="00:00", validation_alias="SYNC_WINDOW_START")
sync_window_end: str = Field(default="06:00", validation_alias="SYNC_WINDOW_END")
sync_window_log_interval_seconds: int = Field(
default=300,
validation_alias="SYNC_WINDOW_LOG_INTERVAL_SECONDS",
)
deploy_autopr_enabled: bool = Field(default=False, validation_alias="DEPLOY_AUTOPR_ENABLED")
deploy_autopr_github_token: str = Field(default="", validation_alias="DEPLOY_AUTOPR_GITHUB_TOKEN")
sync-service/app/db.py
+23
View File
@@ -232,6 +232,29 @@ class Database:
task["last_error"] = None
return task
def requeue_syncing_tasks(self, reason: str) -> int:
now = utc_now_iso()
with self._connect() as conn:
cursor = conn.execute(
"""
UPDATE tasks
SET status = ?,
retry_count = retry_count + 1,
next_retry_at = ?,
last_error = ?,
updated_at = ?
WHERE status = ?
""",
(
TaskStatus.FAILED_RETRYABLE.value,
now,
self._truncate_error(reason),
now,
TaskStatus.SYNCING.value,
),
)
return int(cursor.rowcount)
def mark_verified(self, task_id: str) -> None:
now = utc_now_iso()
with self._connect() as conn:
sync-service/app/worker.py
+67
View File
@@ -3,7 +3,10 @@ from __future__ import annotations
import logging
import signal
import threading
import time
from datetime import datetime
from typing import Any
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from app.config import Settings, get_settings
from app.db import Database, utc_after_seconds_iso
@@ -30,6 +33,10 @@ class SyncWorker:
self._stop_event = threading.Event()
self._thread: threading.Thread | None = None
self._running = False
self._window_tz = self._resolve_window_timezone()
self._window_start_minutes = self._parse_hhmm(self.settings.sync_window_start)
self._window_end_minutes = self._parse_hhmm(self.settings.sync_window_end)
self._last_window_log_at = 0.0
def start(self) -> None:
if self._thread and self._thread.is_alive():
@@ -53,9 +60,16 @@ class SyncWorker:
def run_forever(self) -> None:
logger.info("sync worker started")
recovered = self.db.requeue_syncing_tasks("worker startup recovery")
if recovered > 0:
logger.warning("recovered %s stale SYNCING task(s) on startup", recovered)
self._running = True
try:
while not self._stop_event.is_set():
if not self._is_sync_window_open():
self._stop_event.wait(self.settings.worker_poll_seconds)
continue
task = self.db.claim_next_task()
if task is None:
self._stop_event.wait(self.settings.worker_poll_seconds)
@@ -192,6 +206,59 @@ class SyncWorker:
str(exc),
)
@staticmethod
def _parse_hhmm(value: str) -> int:
try:
hour_text, minute_text = value.strip().split(":", maxsplit=1)
hour = int(hour_text)
minute = int(minute_text)
except ValueError as exc:
raise ValueError(f"invalid HH:MM value: {value}") from exc
if hour < 0 or hour > 23 or minute < 0 or minute > 59:
raise ValueError(f"invalid HH:MM value: {value}")
return hour * 60 + minute
def _resolve_window_timezone(self) -> ZoneInfo:
try:
return ZoneInfo(self.settings.sync_window_timezone)
except ZoneInfoNotFoundError:
logger.warning(
"invalid SYNC_WINDOW_TIMEZONE=%s, fallback to UTC",
self.settings.sync_window_timezone,
)
return ZoneInfo("UTC")
def _is_sync_window_open(self) -> bool:
if not self.settings.sync_window_enabled:
return True
now_local = datetime.now(self._window_tz)
now_minutes = now_local.hour * 60 + now_local.minute
start = self._window_start_minutes
end = self._window_end_minutes
if start == end:
return True
if start < end:
is_open = start <= now_minutes < end
else:
is_open = now_minutes >= start or now_minutes < end
if is_open:
return True
current = time.monotonic()
if current - self._last_window_log_at >= max(30, self.settings.sync_window_log_interval_seconds):
self._last_window_log_at = current
logger.info(
"sync window closed, waiting for %s %s-%s",
self.settings.sync_window_timezone,
self.settings.sync_window_start,
self.settings.sync_window_end,
)
return False
def main() -> None:
settings = get_settings()