perf(daemon): preserve telemetry source collection state

internal/daemon/server.go

@@ -20,24 +20,28 @@ import (
 	"github.com/janekbaraniewski/openusage/internal/core"
 	"github.com/janekbaraniewski/openusage/internal/exporter"
 	"github.com/janekbaraniewski/openusage/internal/providers"
+	"github.com/janekbaraniewski/openusage/internal/providers/shared"
 	"github.com/janekbaraniewski/openusage/internal/telemetry"
 )
 
 type Service struct {
 	cfg Config
 	ctx context.Context
 
-	store        *telemetry.Store
-	pipeline     *telemetry.Pipeline
-	quotaIngest  *telemetry.QuotaSnapshotIngestor
-	providerByID map[string]core.UsageProvider
-	exp          *exporter.Exporter
+	store            *telemetry.Store
+	pipeline         *telemetry.Pipeline
+	quotaIngest      *telemetry.QuotaSnapshotIngestor
+	providerByID     map[string]core.UsageProvider
+	telemetrySources map[string]shared.TelemetrySource
+	exp              *exporter.Exporter
 
 	spoolMu     sync.Mutex // guards spool filesystem operations (read/write/cleanup)
 	logThrottle *core.LogThrottle
 
 	rmCache       *readModelCache
-	dataIngested  atomic.Bool // set when new data is ingested; read model loop skips refresh when clean
+	dataIngested  atomic.Bool   // set when new data is ingested; read model loop skips refresh when clean
+	dataVersion   atomic.Uint64 // increments after each successful ingest so HTTP caches refresh only when stale
+	collectNow    chan struct{} // coalesced source-change requests consumed by the single collect loop
 	pollScheduler *PollScheduler
 
 	pollStateMu sync.Mutex
@@ -49,6 +53,14 @@ type Service struct {
 	clock core.Clock
 }
 
+func (s *Service) markDataIngested() {
+	if s == nil {
+		return
+	}
+	s.dataVersion.Add(1)
+	s.dataIngested.Store(true)
+}
+
 // now is the canonical "what time is it?" hook for the daemon. Code that
 // stamps snap.Timestamp, persists state, or computes deadlines should call
 // this rather than time.Now(). Pure observability paths (request duration
@@ -131,19 +143,22 @@ func startService(ctx context.Context, cfg Config) (*Service, error) {
 		}
 	}
 
+	telemetrySources := telemetrySourcesBySystem()
 	svc := &Service{
-		cfg:           cfg,
-		ctx:           ctx,
-		store:         store,
-		pipeline:      telemetry.NewPipeline(store, telemetry.NewSpool(cfg.SpoolDir)),
-		quotaIngest:   telemetry.NewQuotaSnapshotIngestor(store),
-		providerByID:  providersByID(),
-		exp:           exp,
-		logThrottle:   core.NewLogThrottle(200, 10*time.Minute),
-		rmCache:       newReadModelCache(),
-		pollScheduler: newPollScheduler(cfg.PollInterval),
-		pollState:     make(map[string]*providerPollState),
-		clock:         core.SystemClock{},
+		cfg:              cfg,
+		ctx:              ctx,
+		store:            store,
+		pipeline:         telemetry.NewPipeline(store, telemetry.NewSpool(cfg.SpoolDir)),
+		quotaIngest:      telemetry.NewQuotaSnapshotIngestor(store),
+		providerByID:     providersByID(),
+		telemetrySources: telemetrySources,
+		exp:              exp,
+		logThrottle:      core.NewLogThrottle(200, 10*time.Minute),
+		rmCache:          newReadModelCache(),
+		collectNow:       make(chan struct{}, 1),
+		pollScheduler:    newPollScheduler(cfg.PollInterval),
+		pollState:        make(map[string]*providerPollState),
+		clock:            core.SystemClock{},
 	}
 
 	svc.infof(
@@ -154,7 +169,7 @@ func startService(ctx context.Context, cfg Config) (*Service, error) {
 		svc.cfg.SpoolDir,
 		svc.cfg.CollectInterval,
 		svc.cfg.PollInterval,
-		telemetrySourceCount(),
+		len(telemetrySources),
 		len(svc.providerByID),
 	)
 
@@ -178,6 +193,7 @@ func startService(ctx context.Context, cfg Config) (*Service, error) {
 	go svc.runSpoolMaintenanceLoop(ctx)
 	go svc.runHookSpoolLoop(ctx)
 	go svc.runRetentionLoop(ctx)
+	go svc.runTelemetryPayloadMaintenanceLoop(ctx)
 
 	if svc.exp != nil {
 		go svc.exp.Start(ctx)

internal/daemon/server_collect.go

@@ -18,31 +18,40 @@ func (s *Service) runCollectLoop(ctx context.Context) {
 	s.infof("collect_loop_start", "interval=%s", interval)
 	s.collectAndFlush(ctx)
 	for {
+		triggeredByChange := false
 		select {
 		case <-ctx.Done():
 			s.infof("collect_loop_stop", "reason=context_done")
 			return
 		case <-time.After(interval):
-			collected := s.collectAndFlush(ctx)
-			if collected == 0 {
-				consecutiveEmpty++
-				if consecutiveEmpty >= 3 {
-					newInterval := interval * 2
-					if newInterval > maxInterval {
-						newInterval = maxInterval
-					}
-					if newInterval != interval {
-						interval = newInterval
-						s.infof("collect_backoff", "interval=%s empty_cycles=%d", interval, consecutiveEmpty)
-					}
+		case <-s.collectNow:
+			triggeredByChange = true
+		}
+
+		collected := s.collectAndFlush(ctx)
+		if collected == 0 {
+			// Noisy filesystem notifications must not accelerate periodic
+			// backoff; they already received an immediate collection attempt.
+			if triggeredByChange {
+				continue
+			}
+			consecutiveEmpty++
+			if consecutiveEmpty >= 3 {
+				newInterval := interval * 2
+				if newInterval > maxInterval {
+					newInterval = maxInterval
 				}
-			} else {
-				if consecutiveEmpty > 0 && interval != s.cfg.CollectInterval {
-					s.infof("collect_reset", "interval=%s→%s collected=%d", interval, s.cfg.CollectInterval, collected)
+				if newInterval != interval {
+					interval = newInterval
+					s.infof("collect_backoff", "interval=%s empty_cycles=%d", interval, consecutiveEmpty)
 				}
-				consecutiveEmpty = 0
-				interval = s.cfg.CollectInterval
 			}
+		} else {
+			if consecutiveEmpty > 0 && interval != s.cfg.CollectInterval {
+				s.infof("collect_reset", "interval=%s→%s collected=%d", interval, s.cfg.CollectInterval, collected)
+			}
+			consecutiveEmpty = 0
+			interval = s.cfg.CollectInterval
 		}
 	}
 }
@@ -85,7 +94,7 @@ func (s *Service) collectAndFlush(ctx context.Context) int {
 	if accountsErr != nil {
 		warnings = append(warnings, fmt.Sprintf("collector account config: %v", accountsErr))
 	}
-	collectors, collectorWarnings := buildCollectors(accounts)
+	collectors, collectorWarnings := buildCollectorsFromSources(accounts, s.telemetrySources)
 	warnings = append(warnings, collectorWarnings...)
 
 	for _, collector := range collectors {
@@ -98,18 +107,17 @@ func (s *Service) collectAndFlush(ctx context.Context) int {
 		allReqs = append(allReqs, reqs...)
 	}
 
-	// No ingest-time age filter: local-file sources re-import the last ~90d of
-	// history each cycle, which is fine because the hot window (retention_days)
-	// is ≥ that lookback, so re-imported events land inside the window and are
-	// never the tug-of-war target. Detail past the hot window is downsampled
-	// into usage_rollup_daily and then pruned (see pruneOldData).
+	// No ingest-time age filter: local-file sources import their available
+	// history on the first cycle and emit only changed-file events afterward.
+	// Detail past the hot window is downsampled into usage_rollup_daily and then
+	// pruned (see pruneOldData).
 	direct, retries := s.ingestBatch(ctx, allReqs)
 	if direct.ingested > 0 {
-		s.dataIngested.Store(true)
+		s.markDataIngested()
 	}
 	flush, enqueued, flushWarnings := s.flushBacklog(ctx, retries, backlogFlushLimit)
 	if flush.Ingested > 0 {
-		s.dataIngested.Store(true)
+		s.markDataIngested()
 	}
 	warnings = append(warnings, flushWarnings...)
 
@@ -126,46 +134,54 @@ func (s *Service) collectAndFlush(ctx context.Context) int {
 		for _, warning := range warnings {
 			s.warnf("collect_warning", "message=%q", warning)
 		}
-		s.pruneTelemetryOrphans(ctx)
 		return totalCollected
 	}
 
 	if durationMs >= 1500 && s.shouldLog("collect_slow", 30*time.Second) {
 		s.infof("collect_idle_slow", "duration_ms=%d", durationMs)
 	}
 
-	s.pruneTelemetryOrphans(ctx)
 	return totalCollected
 }
 
-func (s *Service) pruneTelemetryOrphans(ctx context.Context) {
-	if s == nil || s.store == nil {
-		return
-	}
-	if !s.shouldLog("prune_orphan_raw_events_tick", 45*time.Second) {
-		return
-	}
+const telemetryPayloadMaintenanceBatchSize = 10000
 
-	const pruneBatchSize = 10000
-	pruneCtx, cancel := context.WithTimeout(ctx, 4*time.Second)
-	defer cancel()
+func nextTelemetryPayloadMaintenanceDelay(backlog bool) time.Duration {
+	if backlog {
+		return 10 * time.Minute
+	}
+	return 6 * time.Hour
+}
 
-	removed, err := s.store.PruneOrphanRawEvents(pruneCtx, pruneBatchSize)
-	if err != nil {
-		if s.shouldLog("prune_orphan_raw_events_error", 20*time.Second) {
-			s.warnf("prune_orphan_raw_events_error", "error=%v", err)
-		}
+func (s *Service) runTelemetryPayloadMaintenanceLoop(ctx context.Context) {
+	if s == nil || s.store == nil {
 		return
 	}
-	if removed > 0 {
-		s.infof("prune_orphan_raw_events", "removed=%d batch_size=%d", removed, pruneBatchSize)
-	}
-
-	payloadCtx, payloadCancel := context.WithTimeout(ctx, 4*time.Second)
-	defer payloadCancel()
-	pruned, pruneErr := s.store.PruneRawEventPayloads(payloadCtx, 1, pruneBatchSize)
-	if pruneErr == nil && pruned > 0 {
-		s.infof("prune_raw_payloads", "pruned=%d", pruned)
+	// Keep large-table cleanup off the hot collection path and away from
+	// startup. Orphan cleanup already runs after retention deletes canonical
+	// events; this loop only strips old raw payload bodies in bounded batches.
+	timer := time.NewTimer(10 * time.Minute)
+	defer timer.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+			maintenanceCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+			pruned, err := s.store.PruneRawEventPayloads(maintenanceCtx, 1, telemetryPayloadMaintenanceBatchSize)
+			cancel()
+			if err != nil {
+				if s.shouldLog("prune_raw_payloads_error", 30*time.Minute) {
+					s.warnf("prune_raw_payloads_error", "error=%v", err)
+				}
+				timer.Reset(10 * time.Minute)
+				continue
+			}
+			if pruned > 0 {
+				s.infof("prune_raw_payloads", "pruned=%d batch_size=%d", pruned, telemetryPayloadMaintenanceBatchSize)
+			}
+			timer.Reset(nextTelemetryPayloadMaintenanceDelay(pruned >= telemetryPayloadMaintenanceBatchSize))
+		}
 	}
 }
 
@@ -216,6 +232,13 @@ func (s *Service) pruneOldData(ctx context.Context) (complete bool) {
 
 	pruneCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
 	defer cancel()
+	if removed, changeErr := s.store.PruneUsageEventChanges(pruneCtx, 50000); changeErr != nil {
+		if s.shouldLog("usage_change_log_prune_error", 30*time.Second) {
+			s.warnf("usage_change_log_prune_error", "error=%v", changeErr)
+		}
+	} else if removed > 0 {
+		s.infof("usage_change_log_prune", "removed=%d", removed)
+	}
 
 	// Thin and trim the balance observation series independently of usage
 	// events — it grows on its own poll cadence and has its own retention floor.

internal/daemon/server_http.go

@@ -60,6 +60,9 @@ func (s *Service) handleHook(w http.ResponseWriter, r *http.Request) {
 	}
 
 	tally, _ := s.ingestBatch(r.Context(), parsed.Requests)
+	if tally.ingested > 0 {
+		s.markDataIngested()
+	}
 	warnings := append([]string(nil), parsed.Warnings...)
 	if tally.failed > 0 {
 		warnings = append(warnings, fmt.Sprintf("%d ingest failures", tally.failed))
@@ -119,23 +122,24 @@ func (s *Service) handleReadModel(w http.ResponseWriter, r *http.Request) {
 	}
 
 	cacheKey := ReadModelRequestKey(req)
-	if cached, cachedAt, ok := s.rmCache.get(cacheKey); ok {
+	if cached, cachedAt, cachedVersion, ok := s.rmCache.get(cacheKey); ok {
 		core.Tracef("[read_model] cache hit key=%s age=%s providers=%d", cacheKey, time.Since(cachedAt).Round(time.Millisecond), len(cached))
 		for id, snap := range cached {
 			core.Tracef("[read_model]   %s: %d metrics", id, len(snap.Metrics))
 		}
 		writeJSON(w, http.StatusOK, ReadModelResponse{Snapshots: cached})
-		if time.Since(cachedAt) > 2*time.Second {
+		if shouldRefreshCachedReadModel(cachedAt, cachedVersion, s.dataVersion.Load(), time.Now()) {
 			s.refreshReadModelCacheAsync(s.serviceContext(r.Context()), cacheKey, req, 60*time.Second)
 		}
 		return
 	}
 
+	computeVersion := s.dataVersion.Load()
 	computeCtx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
 	snapshots, err := s.computeReadModel(computeCtx, req)
 	cancel()
 	if err == nil && len(snapshots) > 0 {
-		s.rmCache.set(cacheKey, snapshots)
+		s.rmCache.set(cacheKey, snapshots, computeVersion)
 		writeJSON(w, http.StatusOK, ReadModelResponse{Snapshots: snapshots})
 		return
 	}