fix: data race in Close/signalEviction, persistence flush on StopPersistence

- atomic.Bool guard prevents send on closed evictSignal channel
- StopPersistence drains AOF channel and flushes buffered writes before stopping
- Added Flush() to PersistenceEngine interface and HybridEngine
- All 9 scenario tests pass with -race

Author: rishabhverma17

internal/persistence/hybrid_engine.go

@@ -22,6 +22,9 @@ type PersistenceEngine interface {
 	Start(ctx context.Context) error
 	Stop() error
 
+	// Flush ensures all buffered writes are persisted to disk
+	Flush() error
+
 	// Maintenance
 	Compact() error
 	GetStats() *PersistenceStats
@@ -451,3 +454,21 @@ func (he *HybridEngine) Compact() error {
 	// This will be implemented when we have cache data access
 	return fmt.Errorf("compaction requires cache integration")
 }
+
+// Flush ensures all buffered AOF writes are persisted to disk
+func (he *HybridEngine) Flush() error {
+	if he.aofManager == nil {
+		return nil
+	}
+	he.aofManager.mu.Lock()
+	defer he.aofManager.mu.Unlock()
+	if he.aofManager.writer != nil {
+		if err := he.aofManager.writer.Flush(); err != nil {
+			return err
+		}
+		if he.aofManager.currentLog != nil {
+			return he.aofManager.currentLog.Sync()
+		}
+	}
+	return nil
+}

internal/storage/basic_store.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"reflect"
 	"sync"
+	"sync/atomic"
 	"time"
 	"unsafe"
 
@@ -100,6 +101,7 @@ type BasicStore struct {
 	// Background eviction
 	evictSignal chan struct{} // Signal background evictor to run
 	evictDone   chan struct{} // Closed when background evictor exits
+	closing     atomic.Bool  // Set to true during Close() to prevent sends on closed channels
 
 	// Background AOF
 	aofChan chan *persistence.LogEntry // Buffered channel for async AOF writes
@@ -600,6 +602,9 @@ func (s *BasicStore) Delete(key string) error {
 
 // signalEviction sends a non-blocking signal to the background evictor
 func (s *BasicStore) signalEviction() {
+	if s.closing.Load() {
+		return // Store is shutting down, don't send on closed channel
+	}
 	select {
 	case s.evictSignal <- struct{}{}:
 	default:
@@ -740,6 +745,9 @@ func (s *BasicStore) IsTombstoned(key string) bool {
 
 // Close shuts down the store and cleans up resources
 func (s *BasicStore) Close() error {
+	// Mark as closing to prevent signalEviction from sending on closed channel
+	s.closing.Store(true)
+
 	// Stop cleanup goroutine and background evictor
 	select {
 	case s.stopCleanup <- true:
@@ -754,6 +762,11 @@ func (s *BasicStore) Close() error {
 	close(s.aofChan)
 	<-s.aofDone
 
+	// Flush persistence engine to ensure all buffered writes hit disk
+	if s.persistEngine != nil {
+		_ = s.persistEngine.Flush()
+	}
+
 	// Clear all items
 	_ = s.Clear()
 

internal/storage/persistence_integration.go

@@ -47,12 +47,30 @@ func (s *BasicStore) getSnapshotData() map[string]interface{} {
 	return data
 }
 
-// StopPersistence gracefully stops the persistence engine
+// StopPersistence gracefully stops the persistence engine.
+// Drains the background AOF channel and flushes buffered writes before stopping.
 func (s *BasicStore) StopPersistence() error {
 	if s.persistEngine == nil {
 		return nil
 	}
 
+	// Drain any pending AOF channel entries to the persistence engine
+	for {
+		select {
+		case entry, ok := <-s.aofChan:
+			if !ok {
+				goto done
+			}
+			_ = s.persistEngine.WriteEntry(entry)
+		default:
+			goto done
+		}
+	}
+done:
+
+	// Flush buffered writes to disk
+	_ = s.persistEngine.Flush()
+
 	return s.persistEngine.Stop()
 }