package routewatch import ( "context" "errors" "log/slog" "sync" "time" "git.eeqj.de/sneak/routewatch/internal/database" ) // Database maintenance configuration constants. const ( // checkpointInterval is how often to run WAL checkpoint. // Frequent checkpoints keep the WAL small, improving read performance. // Under heavy write load, we need aggressive checkpointing. checkpointInterval = 5 * time.Second // vacuumInterval is how often to run incremental vacuum. // Since incremental vacuum only frees ~1000 pages (~4MB) per run, // we run it frequently to keep up with deletions. vacuumInterval = 10 * time.Minute // analyzeInterval is how often to run ANALYZE. analyzeInterval = 1 * time.Hour // checkpointTimeout is the max time for WAL checkpoint. checkpointTimeout = 10 * time.Second // vacuumTimeout is the max time for incremental vacuum (should be quick). vacuumTimeout = 30 * time.Second // analyzeTimeout is the max time for ANALYZE. analyzeTimeout = 5 * time.Minute ) // checkpointIncompleteMessage is logged, at debug level like a completed // checkpoint, when a checkpoint leaves part of the WAL for a later one: while // reads overlap, most do. The database warns when the WAL file stays over its // size limit. const checkpointIncompleteMessage = "WAL checkpoint incomplete" // DBMaintainer handles background database maintenance tasks. type DBMaintainer struct { db database.Store logger *slog.Logger stopCh chan struct{} done chan struct{} // closed when the maintenance goroutine returns // Stats tracking statsMu sync.Mutex lastCheckpoint time.Time lastVacuum time.Time lastAnalyze time.Time checkpointCount int vacuumCount int analyzeCount int lastCheckpointError error lastVacuumError error lastAnalyzeError error } // NewDBMaintainer creates a new database maintainer. func NewDBMaintainer(db database.Store, logger *slog.Logger) *DBMaintainer { return &DBMaintainer{ db: db, logger: logger.With("component", "db_maintainer"), stopCh: make(chan struct{}), done: make(chan struct{}), } } // Start begins the background maintenance goroutine. It runs until Stop: // the end of ctx does not stop it. func (m *DBMaintainer) Start(ctx context.Context) { go m.run(context.WithoutCancel(ctx)) m.logger.Info("Database maintainer started", "checkpoint_interval", checkpointInterval, "vacuum_interval", vacuumInterval, "analyze_interval", analyzeInterval, ) } // Stop stops the maintainer and waits, until ctx ends, for a maintenance task // in progress to finish. func (m *DBMaintainer) Stop(ctx context.Context) { close(m.stopCh) select { case <-m.done: m.logger.Info("Database maintainer stopped") case <-ctx.Done(): m.logger.Warn("Stop deadline passed before the database maintainer " + "stopped") } } // run is the main background loop. func (m *DBMaintainer) run(ctx context.Context) { defer close(m.done) // Use different timers for each task checkpointTimer := time.NewTimer(checkpointInterval) defer checkpointTimer.Stop() vacuumTimer := time.NewTimer(vacuumInterval) defer vacuumTimer.Stop() analyzeTimer := time.NewTimer(analyzeInterval) defer analyzeTimer.Stop() for { select { case <-m.stopCh: return case <-checkpointTimer.C: m.runCheckpoint(ctx) checkpointTimer.Reset(checkpointInterval) case <-vacuumTimer.C: m.runVacuum(ctx) vacuumTimer.Reset(vacuumInterval) case <-analyzeTimer.C: m.runAnalyze(ctx) analyzeTimer.Reset(analyzeInterval) } } } // runCheckpoint performs a WAL checkpoint to keep the WAL file small. func (m *DBMaintainer) runCheckpoint(ctx context.Context) { duration, err := runWithTimeout(ctx, checkpointTimeout, m.db.Checkpoint) m.recordResult(&m.lastCheckpoint, &m.lastCheckpointError, &m.checkpointCount, err) switch { case errors.Is(err, database.ErrCheckpointIncomplete): m.logger.Debug(checkpointIncompleteMessage, "error", err, "duration", duration) case err != nil: m.logger.Error("WAL checkpoint failed", "error", err, "duration", duration) default: m.logger.Debug("WAL checkpoint completed", "duration", duration) } } // runVacuum performs an incremental vacuum operation on the database. func (m *DBMaintainer) runVacuum(ctx context.Context) { m.logger.Debug("Running incremental vacuum") duration, err := runWithTimeout(ctx, vacuumTimeout, m.db.Vacuum) m.recordResult(&m.lastVacuum, &m.lastVacuumError, &m.vacuumCount, err) if err != nil { m.logger.Error("Incremental vacuum failed", "error", err, "duration", duration) } else { m.logger.Debug("Incremental vacuum completed", "duration", duration) } } // runAnalyze performs an ANALYZE operation on the database. func (m *DBMaintainer) runAnalyze(ctx context.Context) { m.logger.Info("Starting database ANALYZE") duration, err := runWithTimeout(ctx, analyzeTimeout, m.db.Analyze) m.recordResult(&m.lastAnalyze, &m.lastAnalyzeError, &m.analyzeCount, err) if err != nil { m.logger.Error("ANALYZE failed", "error", err, "duration", duration) } else { m.logger.Info("ANALYZE completed", "duration", duration) } } // runWithTimeout runs operation with a context that ends after timeout and // returns how long it took and its error. func runWithTimeout( ctx context.Context, timeout time.Duration, operation func(context.Context) error, ) (time.Duration, error) { ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel() startTime := time.Now() err := operation(ctx) return time.Since(startTime), err } // recordResult stores when a maintenance task last ran and its error, and // counts it when it succeeded. func (m *DBMaintainer) recordResult( lastRun *time.Time, lastErr *error, successCount *int, err error, ) { m.statsMu.Lock() defer m.statsMu.Unlock() *lastRun = time.Now() *lastErr = err if err == nil { *successCount++ } }