toor
27ca6e08a0
现象：点「开始下载」后，单个画廊图片较多时整个程序无报错直接退出。
只在图片多的画廊上偶发。
根因：pageState.snapshotLocked 用浅拷贝把 Task 交给 UI：
t := *s.task // Pages map 仍是同一块内存
快照被 EventsEmit 做 JSON 序列化时，其他页 worker 正在写同一个 map，
撞上就抛 Go runtime 的致命错误：
fatal error: concurrent map iteration and map write
这是 runtime 级 fatal error，不可 recover，直接杀进程 —— 所以没有任何
报错提示。图片越多、页并发越高，并发写窗口越长，越容易触发。
同一问题还有第二处：Store.UpsertTask 存的是原值（含活的 map），
而 Flush() 会在后台 json.Marshal 整个 state，同样会撞上。
修复：
- snapshotLocked 深拷贝 Pages map 与 FailedList slice
- UpsertTask 存前先 clone()，让 store 完全拥有自己的数据
- 另加一个兜底 recover，防止将来再有 panic 直接杀掉进程
测试：
- TestSnapshotDoesNotSharePagesMap：验证快照隔离（修复前失败）
- TestSnapshotRaceWindowClosed：持续写原 task + 持续序列化快照，
精准复现崩溃窗口 —— 实测修复前 fatal error 崩进程，修复后通过
- 大画廊压测：200 页 × 页并发 20，以及 5 画廊 × 60 页 × 并发 16
测试从 62 增至 67。
507 lines
12 KiB
Go
507 lines
12 KiB
Go
package core
|
|
|
|
// 持久化：JSON 状态文件（原子写）
|
|
//
|
|
// 为什么不用 SQLite：断点续传的真正依据是「文件系统里文件在不在」，
|
|
// 状态文件只需记住画廊级别的进度。几百个任务的规模下 JSON 完全够用。
|
|
//
|
|
// 写文件用「临时文件 + rename」保证原子性。
|
|
|
|
import (
|
|
"encoding/json"
|
|
"os"
|
|
"path/filepath"
|
|
"sync"
|
|
"time"
|
|
|
|
"eh-downloader-wails/internal/eh"
|
|
)
|
|
|
|
// Settings 用户设置
|
|
type Settings struct {
|
|
DownloadRoot string `json:"downloadRoot"`
|
|
Proxy string `json:"proxy"`
|
|
ProxyMode string `json:"proxyMode"` // off | http | socks5
|
|
// GalleryConcurrency 同时处理几个画廊
|
|
GalleryConcurrency int `json:"galleryConcurrency"`
|
|
// PageConcurrency 每个画廊内同时下几张图（总上限 = 两者相乘）
|
|
PageConcurrency int `json:"pageConcurrency"`
|
|
// RequestIntervalMs 全局请求间隔（毫秒，0 = 不限流）
|
|
RequestIntervalMs int `json:"requestIntervalMs"`
|
|
Timeout int `json:"timeout"` // 毫秒
|
|
Xres string `json:"xres"` // auto | org | 数字(长边上限)
|
|
BridgePort int `json:"bridgePort"`
|
|
DirTemplate string `json:"dirTemplate"`
|
|
|
|
// 内部字段：是否已探测过代理（不暴露给前端）
|
|
Probed bool `json:"_probed"`
|
|
}
|
|
|
|
// FailedItem 单页失败记录
|
|
type FailedItem struct {
|
|
Page int `json:"page"`
|
|
File string `json:"file"`
|
|
Error string `json:"error"`
|
|
}
|
|
|
|
// Task 一个画廊的下载任务
|
|
type Task struct {
|
|
GID string `json:"gid"`
|
|
Token string `json:"token"`
|
|
Raw string `json:"raw"`
|
|
Title string `json:"title"`
|
|
Status string `json:"status"` // queued|running|downloading|partial|done|failed
|
|
Done int `json:"done"`
|
|
Failed int `json:"failed"`
|
|
TotalPages int `json:"totalPages"`
|
|
Error string `json:"error"`
|
|
Dir string `json:"dir"`
|
|
Pages map[string]string `json:"pages"`
|
|
FailedList []FailedItem `json:"failedList"`
|
|
Selected bool `json:"selected"`
|
|
CreatedAt int64 `json:"createdAt"`
|
|
UpdatedAt int64 `json:"updatedAt"`
|
|
}
|
|
|
|
// State 状态文件顶层结构
|
|
type State struct {
|
|
Settings Settings `json:"settings"`
|
|
Tasks []Task `json:"tasks"`
|
|
}
|
|
|
|
// DefaultSettings 返回默认设置
|
|
func DefaultSettings() Settings {
|
|
return Settings{
|
|
DownloadRoot: "",
|
|
Proxy: "",
|
|
ProxyMode: "off",
|
|
GalleryConcurrency: 5,
|
|
PageConcurrency: 8,
|
|
RequestIntervalMs: 0,
|
|
Timeout: 30000,
|
|
Xres: "auto",
|
|
BridgePort: 37777,
|
|
DirTemplate: "{title}",
|
|
}
|
|
}
|
|
|
|
// Store 是带锁的 JSON 状态存储，并发安全
|
|
type Store struct {
|
|
mu sync.RWMutex
|
|
path string
|
|
state State
|
|
|
|
flushTimer *time.Timer
|
|
dirty bool
|
|
}
|
|
|
|
// NewStore 打开（或新建）状态文件
|
|
func NewStore(filePath string) *Store {
|
|
s := &Store{path: filePath}
|
|
s.load()
|
|
return s
|
|
}
|
|
|
|
func (s *Store) load() {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
s.state = State{Settings: DefaultSettings(), Tasks: []Task{}}
|
|
|
|
raw, err := os.ReadFile(s.path)
|
|
if err == nil {
|
|
var loaded State
|
|
if json.Unmarshal(raw, &loaded) == nil {
|
|
// 补齐缺失字段（向后兼容）
|
|
def := DefaultSettings()
|
|
if loaded.Settings.DownloadRoot == "" {
|
|
loaded.Settings.DownloadRoot = def.DownloadRoot
|
|
}
|
|
if loaded.Settings.GalleryConcurrency <= 0 {
|
|
loaded.Settings.GalleryConcurrency = def.GalleryConcurrency
|
|
}
|
|
if loaded.Settings.PageConcurrency <= 0 {
|
|
loaded.Settings.PageConcurrency = def.PageConcurrency
|
|
}
|
|
if loaded.Settings.RequestIntervalMs < 0 {
|
|
loaded.Settings.RequestIntervalMs = 0
|
|
}
|
|
if loaded.Settings.Timeout <= 0 {
|
|
loaded.Settings.Timeout = def.Timeout
|
|
}
|
|
if loaded.Settings.BridgePort <= 0 {
|
|
loaded.Settings.BridgePort = def.BridgePort
|
|
}
|
|
if loaded.Settings.DirTemplate == "" {
|
|
loaded.Settings.DirTemplate = def.DirTemplate
|
|
}
|
|
if loaded.Settings.Xres == "" {
|
|
loaded.Settings.Xres = def.Xres
|
|
}
|
|
if loaded.Tasks == nil {
|
|
loaded.Tasks = []Task{}
|
|
}
|
|
s.state = loaded
|
|
}
|
|
}
|
|
}
|
|
|
|
// Flush 原子写：先写临时文件再 rename
|
|
func (s *Store) Flush() {
|
|
s.mu.RLock()
|
|
data, err := json.MarshalIndent(s.state, "", " ")
|
|
path := s.path
|
|
s.dirty = false
|
|
s.mu.RUnlock()
|
|
|
|
if err != nil {
|
|
return
|
|
}
|
|
tmp := path + ".tmp"
|
|
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
|
return
|
|
}
|
|
if err := os.WriteFile(tmp, data, 0o644); err != nil {
|
|
return
|
|
}
|
|
_ = os.Rename(tmp, path)
|
|
}
|
|
|
|
// ScheduleFlush 合并短时间内的多次写，降低磁盘压力
|
|
func (s *Store) ScheduleFlush(delay int) {
|
|
if delay <= 0 {
|
|
delay = 400
|
|
}
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.flushTimer != nil {
|
|
return
|
|
}
|
|
s.dirty = true
|
|
s.flushTimer = time.AfterFunc(time.Duration(delay)*time.Millisecond, func() {
|
|
s.mu.Lock()
|
|
s.flushTimer = nil
|
|
s.mu.Unlock()
|
|
s.Flush()
|
|
})
|
|
}
|
|
|
|
// Stop 停掉待执行的延迟写（退出前调用）
|
|
func (s *Store) Stop() {
|
|
s.mu.Lock()
|
|
if s.flushTimer != nil {
|
|
s.flushTimer.Stop()
|
|
s.flushTimer = nil
|
|
}
|
|
s.mu.Unlock()
|
|
}
|
|
|
|
// Settings 取当前设置（返回副本）
|
|
func (s *Store) Settings() Settings {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
return s.state.Settings
|
|
}
|
|
|
|
// UpdateSettings 合并设置
|
|
func (s *Store) UpdateSettings(patch map[string]interface{}) Settings {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
m, _ := json.Marshal(s.state.Settings)
|
|
var cur map[string]interface{}
|
|
_ = json.Unmarshal(m, &cur)
|
|
for k, v := range patch {
|
|
cur[k] = v
|
|
}
|
|
m2, _ := json.Marshal(cur)
|
|
var next Settings
|
|
if json.Unmarshal(m2, &next) == nil {
|
|
s.state.Settings = next
|
|
}
|
|
s.startFlushLocked()
|
|
return s.state.Settings
|
|
}
|
|
|
|
func (s *Store) startFlushLocked() {
|
|
if s.flushTimer != nil {
|
|
return
|
|
}
|
|
s.dirty = true
|
|
s.flushTimer = time.AfterFunc(400*time.Millisecond, func() {
|
|
s.mu.Lock()
|
|
s.flushTimer = nil
|
|
s.mu.Unlock()
|
|
s.Flush()
|
|
})
|
|
}
|
|
|
|
// Tasks 取全部任务（深拷贝，防止调用方改到内部状态）
|
|
func (s *Store) Tasks() []Task {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
out := make([]Task, 0, len(s.state.Tasks))
|
|
for _, t := range s.state.Tasks {
|
|
out = append(out, t.clone())
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (t Task) clone() Task {
|
|
c := t
|
|
if t.Pages != nil {
|
|
c.Pages = make(map[string]string, len(t.Pages))
|
|
for k, v := range t.Pages {
|
|
c.Pages[k] = v
|
|
}
|
|
}
|
|
if t.FailedList != nil {
|
|
c.FailedList = append([]FailedItem(nil), t.FailedList...)
|
|
}
|
|
return c
|
|
}
|
|
|
|
// FindTask 找任务
|
|
func (s *Store) FindTask(gid, token string) *Task {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
for i := range s.state.Tasks {
|
|
if s.state.Tasks[i].GID == gid && s.state.Tasks[i].Token == token {
|
|
c := s.state.Tasks[i].clone()
|
|
return &c
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// FindTaskByGID 只按 gid 找（清单恢复用 —— token 可能对不上）
|
|
func (s *Store) FindTaskByGID(gid string) *Task {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
for i := range s.state.Tasks {
|
|
if s.state.Tasks[i].GID == gid {
|
|
c := s.state.Tasks[i].clone()
|
|
return &c
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// AddTasks 批量加任务（已存在的跳过），返回新增数
|
|
func (s *Store) AddTasks(refs []eh.GalleryRef) int {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
added := 0
|
|
now := time.Now().UnixMilli()
|
|
for _, r := range refs {
|
|
if r.GID == "" || r.Token == "" {
|
|
continue
|
|
}
|
|
exists := false
|
|
for _, t := range s.state.Tasks {
|
|
if t.GID == r.GID && t.Token == r.Token {
|
|
exists = true
|
|
break
|
|
}
|
|
}
|
|
if exists {
|
|
continue
|
|
}
|
|
s.state.Tasks = append([]Task{{
|
|
GID: r.GID,
|
|
Token: r.Token,
|
|
Raw: r.Raw,
|
|
Title: r.Title,
|
|
Status: "queued",
|
|
Pages: map[string]string{},
|
|
FailedList: []FailedItem{},
|
|
CreatedAt: now,
|
|
UpdatedAt: now,
|
|
}}, s.state.Tasks...)
|
|
added++
|
|
}
|
|
s.startFlushLocked()
|
|
return added
|
|
}
|
|
|
|
// SetTitle 给已有任务补标题（油猴推送时带标题来的）
|
|
func (s *Store) SetTitle(gid, token, title string) bool {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
for i := range s.state.Tasks {
|
|
if s.state.Tasks[i].GID == gid && s.state.Tasks[i].Token == token {
|
|
if s.state.Tasks[i].Title == "" && title != "" {
|
|
s.state.Tasks[i].Title = truncate(title, 300)
|
|
s.startFlushLocked()
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func truncate(s string, n int) string {
|
|
r := []rune(s)
|
|
if len(r) <= n {
|
|
return s
|
|
}
|
|
return string(r[:n])
|
|
}
|
|
|
|
// UpsertTask 更新/插入任务。
|
|
//
|
|
// ⚠️ 存进去前必须 clone() —— 否则外部（页级 worker）继续改同一个 map，
|
|
// 而 store 的 Flush() 会在后台 json.Marshal 整个 state —— 撞上就是
|
|
// fatal error: concurrent map write（不可 recover，进程闪退）。
|
|
func (s *Store) UpsertTask(t Task) Task {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
t.UpdatedAt = time.Now().UnixMilli()
|
|
if t.Pages == nil {
|
|
t.Pages = map[string]string{}
|
|
}
|
|
if t.FailedList == nil {
|
|
t.FailedList = []FailedItem{}
|
|
}
|
|
|
|
for i := range s.state.Tasks {
|
|
if s.state.Tasks[i].GID == t.GID && s.state.Tasks[i].Token == t.Token {
|
|
if t.CreatedAt == 0 {
|
|
t.CreatedAt = s.state.Tasks[i].CreatedAt
|
|
}
|
|
s.state.Tasks[i] = t.clone()
|
|
s.startFlushLocked()
|
|
return t.clone()
|
|
}
|
|
}
|
|
|
|
if t.CreatedAt == 0 {
|
|
t.CreatedAt = t.UpdatedAt
|
|
}
|
|
stored := t.clone()
|
|
s.state.Tasks = append([]Task{stored}, s.state.Tasks...)
|
|
s.startFlushLocked()
|
|
return stored.clone()
|
|
}
|
|
|
|
// RemoveTask 删单个任务
|
|
func (s *Store) RemoveTask(gid, token string) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
for i := range s.state.Tasks {
|
|
if s.state.Tasks[i].GID == gid && s.state.Tasks[i].Token == token {
|
|
s.state.Tasks = append(s.state.Tasks[:i], s.state.Tasks[i+1:]...)
|
|
s.startFlushLocked()
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// ClearAll 清空任务列表（或只删勾选的）。
|
|
// ⚠️ 只清任务列表，本地已下载的文件一个都不动。
|
|
func (s *Store) ClearAll(onlySelected bool) int {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
before := len(s.state.Tasks)
|
|
kept := make([]Task, 0, before)
|
|
for _, t := range s.state.Tasks {
|
|
if onlySelected && !t.Selected {
|
|
kept = append(kept, t)
|
|
}
|
|
}
|
|
s.state.Tasks = kept
|
|
s.startFlushLocked()
|
|
return before - len(s.state.Tasks)
|
|
}
|
|
|
|
// ClearFinished 只清 done。partial/failed 保留（里面还有失败清单要处理）。
|
|
func (s *Store) ClearFinished() int {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
before := len(s.state.Tasks)
|
|
kept := make([]Task, 0, before)
|
|
for _, t := range s.state.Tasks {
|
|
if t.Status != "done" {
|
|
kept = append(kept, t)
|
|
}
|
|
}
|
|
s.state.Tasks = kept
|
|
s.startFlushLocked()
|
|
return before - len(s.state.Tasks)
|
|
}
|
|
|
|
// CountDone 数一下已完成
|
|
func (s *Store) CountDone() int {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
n := 0
|
|
for _, t := range s.state.Tasks {
|
|
if t.Status == "done" {
|
|
n++
|
|
}
|
|
}
|
|
return n
|
|
}
|
|
|
|
// RequeueFailed 把失败/部分失败/中断的重新排队
|
|
func (s *Store) RequeueFailed() int {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
n := 0
|
|
for i := range s.state.Tasks {
|
|
switch s.state.Tasks[i].Status {
|
|
case "failed", "partial", "downloading", "running":
|
|
s.state.Tasks[i].Status = "queued"
|
|
s.state.Tasks[i].Error = ""
|
|
n++
|
|
}
|
|
}
|
|
if n > 0 {
|
|
s.startFlushLocked()
|
|
}
|
|
return n
|
|
}
|
|
|
|
// RecoverOnStart 启动时把残留的 running/downloading 归位
|
|
func (s *Store) RecoverOnStart() int {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
n := 0
|
|
for i := range s.state.Tasks {
|
|
if s.state.Tasks[i].Status == "running" || s.state.Tasks[i].Status == "downloading" {
|
|
s.state.Tasks[i].Status = "queued"
|
|
n++
|
|
}
|
|
}
|
|
if n > 0 {
|
|
s.startFlushLocked()
|
|
}
|
|
return n
|
|
}
|
|
|
|
// SelectAll 标记勾选状态
|
|
func (s *Store) SelectAll(gids []string) []Task {
|
|
set := make(map[string]bool, len(gids))
|
|
for _, g := range gids {
|
|
set[g] = true
|
|
}
|
|
s.mu.Lock()
|
|
for i := range s.state.Tasks {
|
|
s.state.Tasks[i].Selected = set[s.state.Tasks[i].GID]
|
|
}
|
|
s.startFlushLocked()
|
|
out := make([]Task, 0, len(s.state.Tasks))
|
|
for _, t := range s.state.Tasks {
|
|
out = append(out, t.clone())
|
|
}
|
|
s.mu.Unlock()
|
|
return out
|
|
}
|