fix(channels): add periodic sweep for pending message compaction

Auto-compaction stopped working after server restarts because provider
resolution failure was silent and there was no retry mechanism. Add a
10-minute periodic sweep in the flusher loop that checks DB for groups
exceeding the compaction threshold, and warn when compaction provider
cannot be resolved at startup.
This commit is contained in:
viettranx committed 2026-03-23 15:45:35 +07:00
1 parent ed840dd6d7
commit 9aaec59527
4 files changed
+50 -2

No files matched your search

+3 -2
View File
@@ -27,8 +27,9 @@ const maxHistoryKeys = 1000
const DefaultGroupHistoryLimit = 100
const (
flushInterval = 3 * time.Second // periodic flush interval
flushBatchMax = 20 // flush when buffer reaches this size
flushInterval = 3 * time.Second // periodic flush interval
flushBatchMax = 20 // flush when buffer reaches this size
compactSweepInterval = 10 * time.Minute // periodic compaction sweep for post-restart safety
)
// HistoryEntry represents a single tracked group message.
+31
View File
@@ -129,6 +129,37 @@ func CompactGroup(ctx context.Context, s store.PendingMessageStore, channelName,
return remaining, nil
}
// sweepCompaction checks DB for groups exceeding compaction threshold.
// Called periodically from flushLoop as a safety net for post-restart scenarios
// where RAM is empty but DB has accumulated messages.
func (ph *PendingHistory) sweepCompaction() {
cfg := ph.compactionCfg
if cfg == nil || cfg.Provider == nil || ph.store == nil {
return
}
threshold := cfg.Threshold
if threshold <= 0 {
threshold = DefaultGroupHistoryLimit
}
ctx, cancel := context.WithTimeout(ph.tenantCtx(), 30*time.Second)
defer cancel()
groups, err := ph.store.ListGroups(ctx)
if err != nil {
slog.Warn("compaction.sweep_failed", "channel", ph.channelName, "error", err)
return
}
for _, g := range groups {
if g.ChannelName != ph.channelName {
continue
}
if g.MessageCount > threshold {
ph.MaybeCompact(g.HistoryKey, g.MessageCount, cfg)
}
}
}
// runCompaction performs LLM-based summarization of old messages.
// Wraps CompactGroup with flush + RAM update + threshold check.
func (ph *PendingHistory) runCompaction(historyKey string, cfg *CompactionConfig) {
+6
View File
@@ -73,11 +73,15 @@ func (ph *PendingHistory) signalFlush() {
}
// flushLoop runs in background, flushing buffer periodically or when signaled.
// Also runs periodic compaction sweep as a safety net for post-restart scenarios.
func (ph *PendingHistory) flushLoop() {
defer close(ph.stopped)
ticker := time.NewTicker(flushInterval)
defer ticker.Stop()
compactTicker := time.NewTicker(compactSweepInterval)
defer compactTicker.Stop()
for {
select {
case <-ph.stopCh:
@@ -87,6 +91,8 @@ func (ph *PendingHistory) flushLoop() {
ph.flushNow()
case <-ticker.C:
ph.flushNow()
case <-compactTicker.C:
ph.sweepCompaction()
}
}
}
+10
View File
@@ -282,6 +282,16 @@ func (l *InstanceLoader) loadInstance(ctx context.Context, inst store.ChannelIns
pc.SetPendingCompaction(cc)
slog.Debug("pending compaction configured", "channel", inst.Name, "provider", p.Name(), "model", model,
"threshold", cc.Threshold, "keep_recent", cc.KeepRecent, "max_tokens", cc.MaxTokens)
} else {
attemptedProvider := ""
if l.pendingCompactCfg != nil {
attemptedProvider = l.pendingCompactCfg.Provider
}
if attemptedProvider == "" && ag != nil {
attemptedProvider = ag.Provider
}
slog.Warn("pending compaction not configured: provider/model unavailable",
"channel", inst.Name, "agent_id", inst.AgentID, "attempted_provider", attemptedProvider)
}
}
l.manager.RegisterChannel(inst.Name, ch)