refactor: timeout + reconciler tickers onto bgRunner
Last two raw tickers; now they get shutdown, panic isolation, and /health/tasks like everything else.
This commit is contained in:
Fimeg
parent
1af5a9a184
commit
6a4be4d12f
3 changed files with 32 additions and 79 deletions
|
|
@ -6,7 +6,6 @@ import (
|
|||
"errors"
|
||||
"log"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/Fimeg/RedFlag/server/internal/database/queries"
|
||||
"github.com/Fimeg/RedFlag/server/internal/models"
|
||||
|
|
@ -22,14 +21,14 @@ import (
|
|||
// then releases it for the slow per-item work, then re-acquires to clear the
|
||||
// flag. This prevents overlapping ReconcileAll runs without holding the lock
|
||||
// across HTTP calls, DB writes, or the repology rate-limit sleep.
|
||||
//
|
||||
// Periodic execution is managed by the taskrunner; register ReconcileAll with
|
||||
// bgRunner.Every so the runner owns the ticker, shutdown, and panic isolation.
|
||||
type Reconciler struct {
|
||||
agentSWQueries *queries.AgentTrackedSoftwareQueries
|
||||
upstreamQ *queries.UpstreamQueries
|
||||
reconcileQ *queries.ReconciliationQueries
|
||||
aliasCache *upstream.RepologyCache
|
||||
interval time.Duration
|
||||
shutdown chan struct{}
|
||||
stopOnce sync.Once
|
||||
mu sync.Mutex
|
||||
running bool
|
||||
}
|
||||
|
|
@ -45,34 +44,6 @@ func NewReconciler(
|
|||
upstreamQ: uq,
|
||||
reconcileQ: rq,
|
||||
aliasCache: ac,
|
||||
interval: time.Hour,
|
||||
shutdown: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Reconciler) Start(ctx context.Context) {
|
||||
go r.loop(ctx)
|
||||
}
|
||||
|
||||
func (r *Reconciler) Stop() {
|
||||
r.stopOnce.Do(func() { close(r.shutdown) })
|
||||
}
|
||||
|
||||
func (r *Reconciler) loop(ctx context.Context) {
|
||||
log.Printf("[INFO] [reconciler] started interval=%s", r.interval)
|
||||
r.ReconcileAll(ctx)
|
||||
|
||||
t := time.NewTicker(r.interval)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-r.shutdown:
|
||||
return
|
||||
case <-t.C:
|
||||
r.ReconcileAll(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -3,7 +3,6 @@ package services
|
|||
import (
|
||||
"fmt"
|
||||
"log"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/Fimeg/RedFlag/server/internal/database/queries"
|
||||
|
|
@ -11,20 +10,18 @@ import (
|
|||
"github.com/gofrs/uuid/v5"
|
||||
)
|
||||
|
||||
// TimeoutService handles timeout management for long-running operations
|
||||
// TimeoutService handles timeout management for long-running operations.
|
||||
// Periodic execution is managed by the taskrunner; call CheckTimeouts on each tick.
|
||||
type TimeoutService struct {
|
||||
commandQueries *queries.CommandQueries
|
||||
updateQueries *queries.UpdateQueries
|
||||
agentQueries *queries.AgentQueries
|
||||
securitySettings *SecuritySettingsService // optional; overrides updateTimeout per-tick via operational.update_stuck_minutes
|
||||
ticker *time.Ticker
|
||||
stopChan chan bool
|
||||
stopOnce sync.Once
|
||||
sentTimeout time.Duration // For commands already sent to agents
|
||||
pendingTimeout time.Duration // For commands stuck in queue
|
||||
receivedTimeout time.Duration // For commands received by agent but not completed (Migration 033 §4)
|
||||
updateTimeout time.Duration // Default for agents stuck in is_updating=true; overridden by settings if wired
|
||||
checkInterval time.Duration // How often to check for timeouts
|
||||
sentTimeout time.Duration // For commands already sent to agents
|
||||
pendingTimeout time.Duration // For commands stuck in queue
|
||||
receivedTimeout time.Duration // For commands received by agent but not completed (Migration 033 §4)
|
||||
updateTimeout time.Duration // Default for agents stuck in is_updating=true; overridden by settings if wired
|
||||
checkInterval time.Duration // Stored for callers; scheduling is handled by bgRunner
|
||||
}
|
||||
|
||||
// SetSecuritySettings injects the settings service. When wired, the
|
||||
|
|
@ -75,36 +72,18 @@ func NewTimeoutService(cq *queries.CommandQueries, uq *queries.UpdateQueries, aq
|
|||
receivedTimeout: receivedTimeout,
|
||||
updateTimeout: updateTimeout,
|
||||
checkInterval: checkInterval,
|
||||
stopChan: make(chan bool),
|
||||
}
|
||||
}
|
||||
|
||||
// Start begins the timeout monitoring service
|
||||
func (ts *TimeoutService) Start() {
|
||||
log.Printf("[INFO] [server] [timeout] service_started sent_timeout=%v pending_timeout=%v received_timeout=%v update_timeout=%v check_interval=%v",
|
||||
ts.sentTimeout, ts.pendingTimeout, ts.receivedTimeout, ts.updateTimeout, ts.checkInterval)
|
||||
|
||||
ts.ticker = time.NewTicker(ts.checkInterval)
|
||||
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
case <-ts.ticker.C:
|
||||
ts.checkForTimeouts()
|
||||
ts.checkForReceivedTimeouts()
|
||||
ts.reconcileAgentUpdates()
|
||||
case <-ts.stopChan:
|
||||
ts.ticker.Stop()
|
||||
log.Println("Timeout service stopped")
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// Stop stops the timeout monitoring service
|
||||
func (ts *TimeoutService) Stop() {
|
||||
ts.stopOnce.Do(func() { close(ts.stopChan) })
|
||||
// CheckTimeouts runs one full sweep: sent/pending command timeouts, received
|
||||
// command timeouts, and stuck-updating agent reconciliation. Register this with
|
||||
// bgRunner.Every so the runner owns the ticker, shutdown, and panic isolation.
|
||||
func (ts *TimeoutService) CheckTimeouts() {
|
||||
log.Printf("[INFO] [server] [timeout] check_start sent_timeout=%v pending_timeout=%v received_timeout=%v update_timeout=%v",
|
||||
ts.sentTimeout, ts.pendingTimeout, ts.receivedTimeout, ts.effectiveUpdateTimeout())
|
||||
ts.checkForTimeouts()
|
||||
ts.checkForReceivedTimeouts()
|
||||
ts.reconcileAgentUpdates()
|
||||
}
|
||||
|
||||
// checkForTimeouts checks for commands that have been running too long
|
||||
|
|
|
|||
Loading…
Reference in a new issue