From 8fbc58eaaab741370b41b2a7dee49b31b4979536 Mon Sep 17 00:00:00 2001 From: XingfenD Date: Mon, 14 Sep 2026 19:40:53 +0800 Subject: [PATCH] fix(redis,scanner): atomic IncrWindow Lua, lock rand/ctx fixes, ScanLock renewal, single-flight, error logging (B1,B2,B3,B9,B10,B11) --- backend/internal/redispkg/redis.go | 89 ++++++++++++++++++--- backend/internal/redispkg/redis_test.go | 100 ++++++++++++++++++++++++ backend/internal/scanner/scanner.go | 55 ++++++++++--- 3 files changed, 223 insertions(+), 21 deletions(-) diff --git a/backend/internal/redispkg/redis.go b/backend/internal/redispkg/redis.go index abae72e..f698c8d 100644 --- a/backend/internal/redispkg/redis.go +++ b/backend/internal/redispkg/redis.go @@ -44,18 +44,25 @@ func (r *R) Set(ctx context.Context, key, val string, ttl time.Duration) { } } +// incrWindowScript atomically increments and sets TTL on first value, +// preventing the INCR+EXPIRE race that could leave keys without TTL (B1). +var incrWindowScript = redis.NewScript(` +local n = redis.call('INCR', KEYS[1]) +if n == 1 then + redis.call('EXPIRE', KEYS[1], ARGV[1]) +end +return n +`) + func (r *R) IncrWindow(ctx context.Context, key string, ttl time.Duration) int { if r.c == nil { return 1 } - n, err := r.c.Incr(ctx, key).Result() + n, err := incrWindowScript.Run(ctx, r.c, []string{key}, int(ttl.Seconds())).Int() if err != nil { - return 1 + return 1 // fail-open } - if n == 1 { - r.c.Expire(ctx, key, ttl) - } - return int(n) + return n } func (r *R) Lock(ctx context.Context, key string, ttl time.Duration) (func(), bool) { @@ -64,7 +71,11 @@ func (r *R) Lock(ctx context.Context, key string, ttl time.Duration) (func(), bo return noop, true } b := make([]byte, 8) - rand.Read(b) + if _, err := rand.Read(b); err != nil { + // B2: rand failure → degrade to no-lock instead of using a zero token. + log.Printf("rand.Read failed: %v (proceeding without lock)", err) + return noop, true + } tok := hex.EncodeToString(b) ok, err := r.c.SetNX(ctx, key, tok, ttl).Result() if err != nil { // spec §9: Redis 故障降级放行,锁只做尽力去重 @@ -75,8 +86,68 @@ func (r *R) Lock(ctx context.Context, key string, ttl time.Duration) (func(), bo return noop, false // 锁被持有,别的副本在扫 } return func() { - r.c.Eval(ctx, + // B3: use WithoutCancel so unlock survives caller cancellation. + if err := r.c.Eval(context.WithoutCancel(ctx), "if redis.call('get',KEYS[1])==ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end", - []string{key}, tok) + []string{key}, tok).Err(); err != nil { + log.Printf("redis unlock %s: %v", key, err) + } }, true } + +// ScanLock acquires a distributed lock with automatic renewal. +// The lock is renewed every ttl/2 until unlock is called. +// Returns (unlock, true) on success, (noop, true) on redis failure (degrade), +// or (noop, false) if the lock is already held. +func (r *R) ScanLock(ctx context.Context, key string, ttl time.Duration) (func(), bool) { + noop := func() {} + if r.c == nil { + return noop, true + } + b := make([]byte, 8) + if _, err := rand.Read(b); err != nil { + log.Printf("rand.Read failed: %v (proceeding without lock)", err) + return noop, true + } + tok := hex.EncodeToString(b) + ok, err := r.c.SetNX(ctx, key, tok, ttl).Result() + if err != nil { + log.Printf("redis scanlock %s: %v (proceeding without lock)", key, err) + return noop, true + } + if !ok { + return noop, false + } + + // Start renewal goroutine. + done := make(chan struct{}) + go func() { + ticker := time.NewTicker(ttl / 2) + defer ticker.Stop() + for { + select { + case <-done: + return + case <-ticker.C: + // Renew only if we still own the lock. + if err := r.c.Eval(context.Background(), + `if redis.call('get',KEYS[1])==ARGV[1] then + return redis.call('expire',KEYS[1],ARGV[2]) + else return 0 end`, + []string{key}, tok, int(ttl.Seconds())).Err(); err != nil { + log.Printf("redis scanlock renew %s: %v", key, err) + } + } + } + }() + + unlock := func() { + close(done) // stop renewal + if err := r.c.Eval(context.WithoutCancel(ctx), + "if redis.call('get',KEYS[1])==ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end", + []string{key}, tok).Err(); err != nil { + log.Printf("redis scanlock unlock %s: %v", key, err) + } + } + return unlock, true +} diff --git a/backend/internal/redispkg/redis_test.go b/backend/internal/redispkg/redis_test.go index 8d65092..62dbf1f 100644 --- a/backend/internal/redispkg/redis_test.go +++ b/backend/internal/redispkg/redis_test.go @@ -2,6 +2,7 @@ package redispkg import ( "context" + "os" "testing" "time" ) @@ -21,6 +22,12 @@ func TestDisabledIsSafe(t *testing.T) { t.Fatal("disabled Lock must always acquire") } un() + // ScanLock disabled mode. + un2, ok2 := r.ScanLock(ctx, "slk", time.Second) + if !ok2 { + t.Fatal("disabled ScanLock must always acquire") + } + un2() } func TestDeadRedisLockFailsOpen(t *testing.T) { @@ -31,3 +38,96 @@ func TestDeadRedisLockFailsOpen(t *testing.T) { } un() } + +func TestDeadRedisScanLockFailsOpen(t *testing.T) { + r := New("redis://127.0.0.1:16399") + un, ok := r.ScanLock(context.Background(), "slk", time.Second) + if !ok { + t.Fatal("ScanLock on redis error must fail open") + } + un() +} + +func newLiveRedis(t *testing.T) *R { + t.Helper() + url := os.Getenv("REDIS_URL") + if url == "" { + t.Skip("REDIS_URL not set") + } + return New(url) +} + +func TestIncrWindowSetsTTL(t *testing.T) { + r := newLiveRedis(t) + ctx := context.Background() + key := "test:incrwindow:ttl:" + t.Name() + r.c.Del(ctx, key) + + n := r.IncrWindow(ctx, key, 5*time.Second) + if n != 1 { + t.Fatalf("first call = %d, want 1", n) + } + ttl, err := r.c.TTL(ctx, key).Result() + if err != nil { + t.Fatal(err) + } + if ttl <= 0 { + t.Fatalf("TTL should be positive after first increment, got %v", ttl) + } + + // Second call: n=2, TTL should still be positive. + n = r.IncrWindow(ctx, key, 5*time.Second) + if n != 2 { + t.Fatalf("second call = %d, want 2", n) + } + r.c.Del(ctx, key) +} + +func TestScanLockRenewal(t *testing.T) { + r := newLiveRedis(t) + ctx := context.Background() + key := "test:scanlock:" + t.Name() + r.c.Del(ctx, key) + + unlock, ok := r.ScanLock(ctx, key, 2*time.Second) + if !ok { + t.Fatal("should acquire") + } + // Wait 3 seconds — without renewal, the lock would expire at 2s. + time.Sleep(3 * time.Second) + + // A second attempt should fail (lock still held by first, renewed). + _, ok2 := r.ScanLock(ctx, key, 2*time.Second) + if ok2 { + t.Fatal("second acquire should fail — lock should have been renewed") + } + unlock() + r.c.Del(ctx, key) +} + +func TestScanLockMutualExclusion(t *testing.T) { + r := newLiveRedis(t) + ctx := context.Background() + key := "test:scanlock:mutex:" + t.Name() + r.c.Del(ctx, key) + + unlock1, ok1 := r.ScanLock(ctx, key, 10*time.Second) + if !ok1 { + t.Fatal("first should acquire") + } + + _, ok2 := r.ScanLock(ctx, key, 10*time.Second) + if ok2 { + t.Fatal("second should not acquire while first holds") + } + + unlock1() + // After unlock, a new acquire should succeed. + time.Sleep(50 * time.Millisecond) + unlock3, ok3 := r.ScanLock(ctx, key, 10*time.Second) + if !ok3 { + t.Fatal("should acquire after unlock") + } + unlock3() + r.c.Del(ctx, key) +} diff --git a/backend/internal/scanner/scanner.go b/backend/internal/scanner/scanner.go index e794c52..3543f9c 100644 --- a/backend/internal/scanner/scanner.go +++ b/backend/internal/scanner/scanner.go @@ -10,6 +10,7 @@ import ( "os" "path/filepath" "strings" + "sync" "time" "booklib/internal/bookfile" @@ -22,6 +23,9 @@ type Scanner struct { st *store.Store cfg *config.Config rdb *redispkg.R + // B9-②: per-library single-flight — concurrent scan triggers for the same + // library are merged into one execution, even without redis. + flights sync.Map // map[int64]*sync.WaitGroup } func New(st *store.Store, cfg *config.Config, rdb *redispkg.R) *Scanner { @@ -42,7 +46,7 @@ func (s *Scanner) Run(ctx context.Context) { continue } for _, l := range libs { - s.ScanLibrary(ctx, l) + s.scanOnce(ctx, l) } } } @@ -54,12 +58,28 @@ func (s *Scanner) ScanLibraryByID(ctx context.Context, id int64) { log.Printf("scan: library %d: %v", id, err) return } + s.scanOnce(ctx, lib) +} + +// scanOnce ensures only one scan per library runs concurrently in this process. +// Concurrent callers block until the in-flight scan completes (B9-②). +func (s *Scanner) scanOnce(ctx context.Context, lib store.Library) { + wg := &sync.WaitGroup{} + wg.Add(1) + if existing, loaded := s.flights.LoadOrStore(lib.ID, wg); loaded { + existing.(*sync.WaitGroup).Wait() + return + } + defer func() { + s.flights.Delete(lib.ID) + wg.Done() + }() s.ScanLibrary(ctx, lib) } func (s *Scanner) ScanLibrary(ctx context.Context, lib store.Library) { - // ponytail: 5min lock TTL; a scan longer than this lets another replica join — refresh mid-walk if libs ever outgrow it - unlock, ok := s.rdb.Lock(ctx, fmt.Sprintf("scan:%d", lib.ID), 5*time.Minute) + // B9-①: ScanLock auto-renews every TTL/2 during long scans. + unlock, ok := s.rdb.ScanLock(ctx, fmt.Sprintf("scan:%d", lib.ID), 5*time.Minute) if !ok { return // 别的副本在扫 } @@ -137,8 +157,6 @@ func titleOf(rel string) string { return strings.TrimSpace(strings.ReplaceAll(strings.TrimSuffix(base, filepath.Ext(base)), "_", " ")) } -// cbz 完整性判定集中在 add/update:PageIndex 失败 → state=error。 -// InsertBook/UpdateBookFile 的 SQL 已把 state 重置为 ready(Task 2),无需显式清 error。 func (s *Scanner) add(ctx context.Context, libID int64, root, rel string, ds diskStat) { format := bookfile.FormatFromExt(filepath.Base(rel)) pageCount := 0 @@ -147,7 +165,7 @@ func (s *Scanner) add(ctx context.Context, libID int64, root, rel string, ds dis idx, err := s.zipIndex(root, rel) pageCount = len(idx) idxErr = err - if idxErr == nil && pageCount == 0 { // 视频/文档 zip 不是漫画,空白 reader 没有意义 + if idxErr == nil && pageCount == 0 { idxErr = errors.New("no images in archive") } } @@ -157,7 +175,10 @@ func (s *Scanner) add(ctx context.Context, libID int64, root, rel string, ds dis return } if idxErr != nil { - s.st.SetBookState(ctx, id, "error", idxErr.Error()) + // B10: log SetBookState errors instead of discarding. + if e := s.st.SetBookState(ctx, id, "error", idxErr.Error()); e != nil { + log.Printf("scan: SetBookState error for %s: %v", rel, e) + } return } s.cover(ctx, id, root, rel, format, ds) @@ -171,7 +192,7 @@ func (s *Scanner) update(ctx context.Context, libID, bookID int64, root, rel str idx, err := s.zipIndex(root, rel) pageCount = len(idx) idxErr = err - if idxErr == nil && pageCount == 0 { // 视频/文档 zip 不是漫画,空白 reader 没有意义 + if idxErr == nil && pageCount == 0 { idxErr = errors.New("no images in archive") } } @@ -180,7 +201,9 @@ func (s *Scanner) update(ctx context.Context, libID, bookID int64, root, rel str return } if idxErr != nil { - s.st.SetBookState(ctx, bookID, "error", idxErr.Error()) + if e := s.st.SetBookState(ctx, bookID, "error", idxErr.Error()); e != nil { + log.Printf("scan: SetBookState error for %s: %v", rel, e) + } return } s.cover(ctx, bookID, root, rel, format, ds) @@ -199,7 +222,8 @@ func (s *Scanner) zipIndex(root, rel string) ([]string, error) { return bookfile.PageIndex(f, st.Size()) } -// cover 失败(坏 epub、无图等)只 log — 书的 state 由 PageIndex 判定,封面缺了有占位 SVG 兜底 +// cover writes the cover image to the cache dir. B11: all write errors are +// checked and logged; orphan .tmp files are cleaned only on failure path. func (s *Scanner) cover(ctx context.Context, id int64, root, rel, format string, ds diskStat) { var img []byte var ext string @@ -223,8 +247,15 @@ func (s *Scanner) cover(ctx context.Context, id int64, root, rel, format string, } tmp := filepath.Join(dir, "cover"+ext+".tmp") dst := filepath.Join(dir, "cover"+ext) - if e := os.WriteFile(tmp, img, 0o644); e == nil { - os.Rename(tmp, dst) + if e := os.WriteFile(tmp, img, 0o644); e != nil { + log.Printf("scan: write cover tmp %s: %v", rel, e) + os.Remove(tmp) + return + } + if e := os.Rename(tmp, dst); e != nil { + log.Printf("scan: rename cover %s: %v", rel, e) + os.Remove(tmp) + return } }