From a22967b7d0ee561423610724057380579c8dc05a Mon Sep 17 00:00:00 2001 From: Kostiantyn <32730812+kkovaletp@users.noreply.github.com> Date: Wed, 11 Jun 2025 17:09:22 +0300 Subject: [PATCH] Refactor the Periodic Scanner module (#1210) * Initial fix of a race condition * better logging * some refactoring and graceful shutdown * fix more potential race condition places; a better app shutdown * add test and fix some issues * some housekeeping * more descriptive asserts * Address review suggestions * Better server shutdown logic + better message about the .env file not found * Create context inside the shutdown goroutine; removed unnecessary shutdown from the main function * Move HTTP server to the main process and remove unnecessary code * log shutdown error if any * Get back to the local `scanner` variable * Protecting scanner with mutex * Adding the mutex to the shutdown function * Addressing the race condition at the init stage --------- Co-authored-by: Konstantin Koval --- api/go.mod | 1 + api/go.sum | 2 + .../periodic_scanner/periodic_scanner.go | 133 +++++++-- .../periodic_scanner/periodic_scanner_test.go | 279 ++++++++++++++++++ api/server.go | 42 ++- 5 files changed, 428 insertions(+), 29 deletions(-) create mode 100644 api/scanner/periodic_scanner/periodic_scanner_test.go diff --git a/api/go.mod b/api/go.mod index 50c9ec6d..dc01592c 100644 --- a/api/go.mod +++ b/api/go.mod @@ -53,6 +53,7 @@ require ( github.com/rogpeppe/go-internal v1.12.0 // indirect github.com/russross/blackfriday/v2 v2.1.0 // indirect github.com/sosodev/duration v1.3.1 // indirect + github.com/stretchr/objx v0.5.2 // indirect github.com/urfave/cli/v2 v2.27.6 // indirect github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1 // indirect golang.org/x/mod v0.24.0 // indirect diff --git a/api/go.sum b/api/go.sum index b7858be5..2edcc265 100644 --- a/api/go.sum +++ b/api/go.sum @@ -86,6 +86,8 @@ github.com/sergi/go-diff v1.3.1/go.mod h1:aMJSSKb2lpPvRNec0+w3fl7LP9IOFzdc9Pa4NF github.com/sosodev/duration v1.3.1 h1:qtHBDMQ6lvMQsL15g4aopM4HEfOaYuhWBw3NPTtlqq4= github.com/sosodev/duration v1.3.1/go.mod h1:RQIBBX0+fMLc/D9+Jb/fwvVmo0eZvDDEERAikUR6SDg= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY= +github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= diff --git a/api/scanner/periodic_scanner/periodic_scanner.go b/api/scanner/periodic_scanner/periodic_scanner.go index f5e6963c..ae7c80da 100644 --- a/api/scanner/periodic_scanner/periodic_scanner.go +++ b/api/scanner/periodic_scanner/periodic_scanner.go @@ -1,26 +1,39 @@ package periodic_scanner import ( - "log" + "fmt" "sync" "time" "github.com/photoview/photoview/api/graphql/models" + "github.com/photoview/photoview/api/log" "github.com/photoview/photoview/api/scanner/scanner_queue" "gorm.io/gorm" ) +type ScannerQueue interface { + AddAllToQueue() error +} + +type RealScannerQueue struct{} + +func (r *RealScannerQueue) AddAllToQueue() error { + return scanner_queue.AddAllToQueue() +} + type periodicScanner struct { ticker *time.Ticker + tickerLocker sync.Mutex ticker_changed chan bool - mutex *sync.Mutex + done chan struct{} db *gorm.DB + scannerQueue ScannerQueue } var mainPeriodicScanner *periodicScanner = nil +var mainPeriodicScannerLocker sync.Mutex func getPeriodicScanInterval(db *gorm.DB) (time.Duration, error) { - var siteInfo models.SiteInfo if err := db.First(&siteInfo).Error; err != nil { return 0, err @@ -29,9 +42,12 @@ func getPeriodicScanInterval(db *gorm.DB) (time.Duration, error) { return time.Duration(siteInfo.PeriodicScanInterval) * time.Second, nil } -func InitializePeriodicScanner(db *gorm.DB) error { +func InitializePeriodicScannerWithQueue(db *gorm.DB, queue ScannerQueue) error { + mainPeriodicScannerLocker.Lock() + defer mainPeriodicScannerLocker.Unlock() + if mainPeriodicScanner != nil { - panic("periodic scanner has already been initialized") + return fmt.Errorf("periodic scanner has already been initialized") } scanInterval, err := getPeriodicScanInterval(db) @@ -42,51 +58,118 @@ func InitializePeriodicScanner(db *gorm.DB) error { mainPeriodicScanner = &periodicScanner{ db: db, ticker_changed: make(chan bool), - mutex: &sync.Mutex{}, + done: make(chan struct{}), + tickerLocker: sync.Mutex{}, + scannerQueue: queue, } - go scanIntervalRunner() + go mainPeriodicScanner.scanIntervalRunner() + + var newTicker *time.Ticker = nil + if scanInterval > 0 { + newTicker = time.NewTicker(scanInterval) + log.Info(nil, "Periodic scan interval changed: "+scanInterval.String()) + } else { + log.Info(nil, "Periodic scan interval changed: disabled") + } + + mainPeriodicScanner.ticker = newTicker + + select { + case mainPeriodicScanner.ticker_changed <- true: + default: + // Channel might be full, but that's okay + } - ChangePeriodicScanInterval(scanInterval) return nil } +func InitializePeriodicScanner(db *gorm.DB) error { + return InitializePeriodicScannerWithQueue(db, &RealScannerQueue{}) +} + func ChangePeriodicScanInterval(duration time.Duration) { var newTicker *time.Ticker = nil if duration > 0 { newTicker = time.NewTicker(duration) - log.Printf("Periodic scan interval changed: %s", duration.String()) + log.Info(nil, "Periodic scan interval changed: "+duration.String()) } else { - log.Print("Periodic scan interval changed: disabled") + log.Info(nil, "Periodic scan interval changed: disabled") } - { - mainPeriodicScanner.mutex.Lock() - defer mainPeriodicScanner.mutex.Unlock() + mainPeriodicScannerLocker.Lock() + scanner := mainPeriodicScanner + mainPeriodicScannerLocker.Unlock() + if scanner != nil { + scanner.tickerLocker.Lock() + defer scanner.tickerLocker.Unlock() + if scanner.ticker != nil { + scanner.ticker.Stop() + } + + scanner.ticker = newTicker + select { + case scanner.ticker_changed <- true: + default: + // Channel might be full, but that's okay + } + } +} + +// ShutdownPeriodicScanner gracefully shuts down the periodic scanner +func ShutdownPeriodicScanner() { + mainPeriodicScannerLocker.Lock() + defer mainPeriodicScannerLocker.Unlock() + + if mainPeriodicScanner != nil { + log.Info(nil, "Shutting down periodic scanner") + + // Signal the runner goroutine to stop + close(mainPeriodicScanner.done) + + // Stop the ticker if it exists + mainPeriodicScanner.tickerLocker.Lock() if mainPeriodicScanner.ticker != nil { mainPeriodicScanner.ticker.Stop() + mainPeriodicScanner.ticker = nil } + mainPeriodicScanner.tickerLocker.Unlock() - mainPeriodicScanner.ticker = newTicker - mainPeriodicScanner.ticker_changed <- true + // Reset the global scanner + mainPeriodicScanner = nil } } -func scanIntervalRunner() { +func (ps *periodicScanner) scanIntervalRunner() { for { - log.Print("Scan interval runner: Waiting for signal") - if mainPeriodicScanner.ticker != nil { + log.Info(nil, "Scan interval runner: Waiting for signal") + + ps.tickerLocker.Lock() + ticker := ps.ticker + ps.tickerLocker.Unlock() + + if ticker != nil { select { - case <-mainPeriodicScanner.ticker_changed: - log.Print("Scan interval runner: New ticker detected") - case <-mainPeriodicScanner.ticker.C: - log.Print("Scan interval runner: Starting periodic scan") - scanner_queue.AddAllToQueue() + case <-ps.done: + log.Info(nil, "Scan interval runner: Shutting down") + return + case <-ps.ticker_changed: + log.Info(nil, "Scan interval runner: New ticker detected") + case <-ticker.C: + log.Info(nil, "Scan interval runner: Starting periodic scan") + if err := ps.scannerQueue.AddAllToQueue(); err != nil { + log.Error(nil, "Scan interval runner: Failed to add all users to queue", "error", err) + } } } else { - <-mainPeriodicScanner.ticker_changed - log.Print("Scan interval runner: New ticker detected") + select { + case <-ps.done: + log.Info(nil, "Scan interval runner: Shutting down") + return + case <-ps.ticker_changed: + log.Info(nil, "Scan interval runner: New ticker detected") + } } } } diff --git a/api/scanner/periodic_scanner/periodic_scanner_test.go b/api/scanner/periodic_scanner/periodic_scanner_test.go new file mode 100644 index 00000000..2dd48a42 --- /dev/null +++ b/api/scanner/periodic_scanner/periodic_scanner_test.go @@ -0,0 +1,279 @@ +package periodic_scanner + +import ( + "errors" + "sync" + "testing" + "time" + + "github.com/photoview/photoview/api/graphql/models" + "github.com/photoview/photoview/api/test_utils" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "gorm.io/gorm" +) + +// MockScannerQueue implements the ScannerQueue interface for testing +type MockScannerQueue struct { + mock.Mock +} + +func (m *MockScannerQueue) AddAllToQueue() error { + return m.Called().Error(0) +} + +func TestMain(m *testing.M) { + test_utils.UnitTestRun(m) +} + +func resetPeriodicScanner() { + + if mainPeriodicScanner != nil { + select { + case <-mainPeriodicScanner.done: + // Already closed + default: + close(mainPeriodicScanner.done) + } + if mainPeriodicScanner.ticker != nil { + mainPeriodicScanner.ticker.Stop() + } + mainPeriodicScanner = nil + } +} + +func createTestSiteInfo(db *gorm.DB, interval int) error { + siteInfo := models.SiteInfo{ + InitialSetup: false, + PeriodicScanInterval: interval, + ConcurrentWorkers: 1, + } + return db.Create(&siteInfo).Error +} + +func TestGetPeriodicScanInterval(t *testing.T) { + db := test_utils.DatabaseTest(t) + + t.Run("successful retrieval", func(t *testing.T) { + assert.NoError(t, createTestSiteInfo(db, 300), "Failed to create test site info with 300 second interval") + + duration, err := getPeriodicScanInterval(db) + assert.NoError(t, err, "Failed to retrieve periodic scan interval from database") + assert.Equal(t, 300*time.Second, duration, + "Periodic scan interval should be 300 seconds but got %v", duration) + }) + + t.Run("database error - no site info", func(t *testing.T) { + db.Exec("DELETE FROM site_info") + + duration, err := getPeriodicScanInterval(db) + assert.Error(t, err, "Expected error when no site info exists in database") + assert.Equal(t, time.Duration(0), duration, + "Duration should be zero when database error occurs, but got %v", duration) + }) +} + +func TestInitializePeriodicScanner(t *testing.T) { + db := test_utils.DatabaseTest(t) + + t.Run("successful initialization with injection", func(t *testing.T) { + defer resetPeriodicScanner() + + mockQueue := &MockScannerQueue{} + assert.NoError(t, createTestSiteInfo(db, 300), "Failed to create test site info with 300 second interval") + assert.NoError(t, InitializePeriodicScannerWithQueue(db, mockQueue), + "Failed to initialize periodic scanner with mock queue") + + // Verify initialization + assert.NotNil(t, mainPeriodicScanner, "mainPeriodicScanner should not be nil after successful initialization") + assert.NotNil(t, mainPeriodicScanner.scannerQueue, "Scanner queue should not be nil after initialization") + assert.Equal(t, mockQueue, mainPeriodicScanner.scannerQueue, "Scanner should use the injected mock queue instance") + + // Verify ticker is set up + mainPeriodicScanner.tickerLocker.Lock() + tickerExists := mainPeriodicScanner.ticker != nil + mainPeriodicScanner.tickerLocker.Unlock() + assert.True(t, tickerExists, "Ticker should be created and set up after scanner initialization") + }) + + t.Run("backward compatibility with original function", func(t *testing.T) { + defer resetPeriodicScanner() + + assert.NoError(t, createTestSiteInfo(db, 300), "Failed to create test site info with 300 second interval") + assert.NoError(t, InitializePeriodicScanner(db), + "Failed to initialize periodic scanner using original function") + + // Verify it uses RealScannerQueue + assert.NotNil(t, mainPeriodicScanner, + "mainPeriodicScanner should be initialized by original InitializePeriodicScanner function") + assert.IsType(t, &RealScannerQueue{}, mainPeriodicScanner.scannerQueue, + "Original InitializePeriodicScanner should use RealScannerQueue by default") + }) + + t.Run("double initialization error", func(t *testing.T) { + defer resetPeriodicScanner() + + mockQueue := &MockScannerQueue{} + assert.NoError(t, createTestSiteInfo(db, 300), "Failed to create test site info with 300 second interval") + assert.NoError(t, InitializePeriodicScannerWithQueue(db, mockQueue), + "Failed first initialization for double initialization test") + + err := InitializePeriodicScannerWithQueue(db, mockQueue) + assert.Error(t, err, "Second initialization attempt should return an error") + assert.Contains(t, err.Error(), "already been initialized", + "Double initialization error should contain 'already been initialized' message") + }) +} + +func TestScanIntervalRunnerWithMocking(t *testing.T) { + t.Run("runner calls queue on ticker events", func(t *testing.T) { + mockQueue := &MockScannerQueue{} + + // Use a channel to synchronize and count calls + callChan := make(chan struct{}, 5) // Buffer for multiple calls + mockQueue.On("AddAllToQueue").Return(nil).Run(func(args mock.Arguments) { + select { + case callChan <- struct{}{}: + default: + // Channel full, but that's okay + } + }).Maybe() + + ps := &periodicScanner{ + ticker: time.NewTicker(50 * time.Millisecond), + ticker_changed: make(chan bool, 1), + done: make(chan struct{}), + tickerLocker: sync.Mutex{}, + scannerQueue: mockQueue, + } + + // Start runner + go ps.scanIntervalRunner() + + // Wait for at least one call with timeout + select { + case <-callChan: + // Success - at least one call received + case <-time.After(200 * time.Millisecond): + t.Fatal("Expected at least one call to AddAllToQueue within 200ms timeout, but ticker events were not processed") + } + + // Proper cleanup - stop ticker first, then close done + ps.ticker.Stop() + close(ps.done) + + // Give time for goroutine to finish + time.Sleep(50 * time.Millisecond) + + // Verify the queue was called as expected + mockQueue.AssertExpectations(t) + }) + + t.Run("runner handles queue errors gracefully", func(t *testing.T) { + mockQueue := &MockScannerQueue{} + // Mock queue to return an error + mockQueue.On("AddAllToQueue").Return(errors.New("queue error")).Maybe() + + ps := &periodicScanner{ + ticker: time.NewTicker(30 * time.Millisecond), + ticker_changed: make(chan bool, 1), + done: make(chan struct{}), + tickerLocker: sync.Mutex{}, + scannerQueue: mockQueue, + } + + // Start runner and let it run briefly + go ps.scanIntervalRunner() + time.Sleep(80 * time.Millisecond) + + // Proper cleanup + ps.ticker.Stop() + close(ps.done) + time.Sleep(50 * time.Millisecond) + + // Test passes if no panic occurred - errors should be logged gracefully + mockQueue.AssertExpectations(t) + }) + + t.Run("runner responds to shutdown signal", func(t *testing.T) { + mockQueue := &MockScannerQueue{} + + ps := &periodicScanner{ + ticker: nil, // No ticker to avoid timing issues + ticker_changed: make(chan bool, 1), + done: make(chan struct{}), + tickerLocker: sync.Mutex{}, + scannerQueue: mockQueue, + } + + runnerDone := make(chan bool) + go func() { + ps.scanIntervalRunner() + close(runnerDone) + }() + + close(ps.done) + + select { + case <-runnerDone: + // Success + case <-time.After(1 * time.Second): + t.Fatal("scanIntervalRunner goroutine did not exit within 1 second after closing done channel") + } + + mockQueue.AssertExpectations(t) + }) + + t.Run("runner responds to ticker changes", func(t *testing.T) { + mockQueue := &MockScannerQueue{} + + ps := &periodicScanner{ + ticker: nil, // Start without ticker + ticker_changed: make(chan bool, 1), + done: make(chan struct{}), + tickerLocker: sync.Mutex{}, + scannerQueue: mockQueue, + } + + runnerDone := make(chan bool) + go func() { + ps.scanIntervalRunner() + close(runnerDone) + }() + + // Send ticker change signal + select { + case ps.ticker_changed <- true: + case <-time.After(100 * time.Millisecond): + t.Fatal("Could not send ticker change signal within 100ms - channel may be blocked") + } + + // Clean shutdown + close(ps.done) + + // Wait for completion + select { + case <-runnerDone: + // Success + case <-time.After(500 * time.Millisecond): + t.Fatal("scanIntervalRunner did not exit within 500ms after receiving ticker change signal and shutdown") + } + + // Test passes if no deadlock occurs + mockQueue.AssertExpectations(t) + }) +} + +func TestRealScannerQueue(t *testing.T) { + t.Run("real scanner queue interface compliance", func(t *testing.T) { + queue := &RealScannerQueue{} + + // Just verify it implements the interface correctly + var _ ScannerQueue = queue + + // Test that it doesn't panic when created + assert.NotNil(t, queue, "RealScannerQueue instance should not be nil after creation") + + // We don't test the actual call since it requires external setup + }) +} diff --git a/api/server.go b/api/server.go index 60974a27..bab3046a 100644 --- a/api/server.go +++ b/api/server.go @@ -1,9 +1,14 @@ package main import ( + "context" "log" "net/http" + "os" + "os/signal" "path" + "syscall" + "time" "github.com/gorilla/handlers" "github.com/gorilla/mux" @@ -27,11 +32,10 @@ import ( ) func main() { - log.Println("Starting Photoview...") if err := godotenv.Load(); err != nil { - log.Println("No .env file found") + log.Println("No .env file found. If Photoview runs in Docker, this is expected and correct.") } terminateWorkers := executable_worker.Initialize() @@ -64,7 +68,6 @@ func main() { } rootRouter := mux.NewRouter() - rootRouter.Use(dataloader.Middleware(db)) rootRouter.Use(auth.Middleware(db)) rootRouter.Use(server.LoggingMiddleware) @@ -116,7 +119,38 @@ func main() { } - log.Panic(http.ListenAndServe(apiListenURL.Host, handlers.CompressHandler(rootRouter))) + srv := &http.Server{ + Addr: apiListenURL.Host, + Handler: handlers.CompressHandler(rootRouter), + } + + setupGracefulShutdown(srv) + + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Panicf("HTTP server failed: %s", err) + } +} + +func setupGracefulShutdown(server *http.Server) { + c := make(chan os.Signal, 1) + signal.Notify(c, os.Interrupt, syscall.SIGTERM) + + go func() { + <-c + log.Println("Shutting down Photoview...") + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) // Wait for 1m to shutdown + defer cancel() + + // Shutdown scanners in correct order + periodic_scanner.ShutdownPeriodicScanner() + scanner_queue.CloseScannerQueue() + + if err := server.Shutdown(ctx); err != nil { + log.Printf("Server shutdown error: %s", err) + } else { + log.Println("Shutdown complete") + } + }() } func logUIendpointURL() {