fix(directory-watcher): address race conditions and state consistency issues

This commit fixes several critical and moderate issues identified in code review:

**Critical Fixes:**
1. **Meta-path collision detection**: Add validation in NewDBsFromDirectoryConfig
   to detect when multiple databases would share the same meta-path, which
   would cause replication state corruption. Returns clear error message
   identifying the conflicting databases.

2. **Store.AddDB documentation**: Improve comments explaining the double-check
   locking pattern used to handle concurrent additions of the same database.
   The pattern prevents duplicates while avoiding holding locks during slow
   Open() operations.

**Moderate Fixes:**
3. **Directory removal state consistency**: Refactor removeDatabase and
   removeDatabasesUnder to only delete from local map after successful
   Store.RemoveDB. Prevents inconsistent state if removal fails.

4. **Context propagation**: Replace context.Background() with dm.ctx in
   directory_watcher.go for proper cancellation during shutdown.

**Testing:**
- All unit tests pass
- Integration test failures are pre-existing on this branch, not introduced
  by these changes (verified by testing before/after)

Fixes identified in PR #827 code review.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: corylanou

cmd/litestream/directory_watcher.go

@@ -280,18 +280,21 @@ func (dm *DirectoryMonitor) handlePotentialDatabase(path string) {
 func (dm *DirectoryMonitor) removeDatabase(path string) {
 	dm.mu.Lock()
 	db := dm.dbs[path]
-	delete(dm.dbs, path)
 	dm.mu.Unlock()
 
 	if db == nil {
 		return
 	}
 
-	if err := dm.store.RemoveDB(context.Background(), db.Path()); err != nil {
+	if err := dm.store.RemoveDB(dm.ctx, db.Path()); err != nil {
 		dm.logger.Error("remove database from store", "path", path, "error", err)
 		return
 	}
 
+	dm.mu.Lock()
+	delete(dm.dbs, path)
+	dm.mu.Unlock()
+
 	dm.logger.Info("removed database from replication", "path", path)
 }
 
@@ -300,21 +303,29 @@ func (dm *DirectoryMonitor) removeDatabasesUnder(dir string) {
 
 	dm.mu.Lock()
 	var toClose []*litestream.DB
+	var toClosePaths []string
 	for path, db := range dm.dbs {
 		if path == dir || strings.HasPrefix(path, prefix) {
 			toClose = append(toClose, db)
-			delete(dm.dbs, path)
+			toClosePaths = append(toClosePaths, path)
 		}
 	}
 	dm.mu.Unlock()
 
-	for _, db := range toClose {
+	// Remove from store first, only delete from local map on success
+	for i, db := range toClose {
 		if db == nil {
 			continue
 		}
-		if err := dm.store.RemoveDB(context.Background(), db.Path()); err != nil {
+		if err := dm.store.RemoveDB(dm.ctx, db.Path()); err != nil {
 			dm.logger.Error("remove database from store", "path", db.Path(), "error", err)
+			continue
 		}
+
+		// Only remove from local map after successful store removal
+		dm.mu.Lock()
+		delete(dm.dbs, toClosePaths[i])
+		dm.mu.Unlock()
 	}
 }
 

cmd/litestream/main.go

@@ -605,11 +605,22 @@ func NewDBsFromDirectoryConfig(dbc *DBConfig) ([]*litestream.DB, error) {
 
 	// Create DB instances for each found database
 	var dbs []*litestream.DB
+	metaPaths := make(map[string]string)
+
 	for _, dbPath := range dbPaths {
 		db, err := newDBFromDirectoryEntry(dbc, dirPath, dbPath)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create DB for %s: %w", dbPath, err)
 		}
+
+		// Validate unique meta-path to prevent replication state corruption
+		if mp := db.MetaPath(); mp != "" {
+			if existingDB, exists := metaPaths[mp]; exists {
+				return nil, fmt.Errorf("meta-path collision: databases %s and %s would share meta-path %s, causing replication state corruption", existingDB, dbPath, mp)
+			}
+			metaPaths[mp] = dbPath
+		}
+
 		dbs = append(dbs, db)
 	}
 

store.go

@@ -164,6 +164,7 @@ func (s *Store) AddDB(db *DB) error {
 		return fmt.Errorf("db required")
 	}
 
+	// First check: see if database already exists
 	s.mu.Lock()
 	for _, existing := range s.dbs {
 		if existing.Path() == db.Path() {
@@ -176,15 +177,21 @@ func (s *Store) AddDB(db *DB) error {
 	// Apply store-wide retention settings before opening the database.
 	db.L0Retention = s.L0Retention
 
+	// Open the database without holding the lock to avoid blocking other operations.
+	// The double-check pattern below handles the race condition.
 	if err := db.Open(); err != nil {
 		return fmt.Errorf("open db: %w", err)
 	}
 
+	// Second check: verify database wasn't added by another goroutine while we were opening.
+	// If it was, close our instance and return without error.
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	for _, existing := range s.dbs {
 		if existing.Path() == db.Path() {
+			// Another goroutine added this database while we were opening.
+			// Close our instance to avoid resource leaks.
 			if err := db.Close(context.Background()); err != nil {
 				slog.Error("close duplicate db", "path", db.Path(), "error", err)
 			}