mirror of
https://github.com/shankar0123/certctl.git
synced 2026-06-14 10:49:22 +00:00
Close I-001 (RetryFailedJobs never invoked) coverage-gap finding
Operator decision answered as Option A: JobService.RetryFailedJobs is
now wired into the scheduler as an always-on 10th loop. Prior to this
commit the method was implemented, unit-tested, and exported but had
zero runtime callers — any job that transitioned to status=Failed stayed
Failed forever regardless of how many attempts it had remaining.
Scheduler — 10th loop:
internal/scheduler/scheduler.go grows a jobRetryLoop alongside the
existing nine loops (renewal, jobs, health, notifications, short-lived,
network scan, digest, health check, cloud discovery). The loop follows
the established run-immediately-then-tick pattern (same shape as
jobProcessorLoop), gated by a sync/atomic.Bool idempotency guard and
joined into the scheduler's sync.WaitGroup so WaitForCompletion drains
it on graceful shutdown. Each tick runs under a 2-minute context
timeout mirroring jobProcessorLoop's opCtx budget. The runJobRetry
helper invokes jobService.RetryFailedJobs(ctx, 3) — the advisory
maxRetries cap is belt-and-suspenders; per-job eligibility is still
enforced inside the service via Attempts < MaxAttempts.
The JobServicer scheduler-interface gains RetryFailedJobs so the
scheduler's dependency surface stays explicit and mockable.
Service — audit trail per retry:
internal/service/job.go:RetryFailedJobs now emits an audit event for
every Failed→Pending transition. Following the house convention used
by all scheduler-emitted events, actor='system' and actorType=
domain.ActorTypeSystem; action='job_retry'; details capture
old_status, new_status, attempts, max_attempts. JobService carries an
optional *AuditService (SetAuditService) that nil-guards to preserve
test-wiring ergonomics — existing tests that construct JobService
without an audit service continue to pass unchanged.
Config — env var with sane default:
internal/config/config.go:SchedulerConfig grows RetryInterval, wired
to CERTCTL_SCHEDULER_RETRY_INTERVAL with a 5-minute default. Validate
rejects intervals below 1 second (matches other scheduler interval
validators).
Server wiring:
cmd/server/main.go calls jobService.SetAuditService(auditService)
after JobService construction and sched.SetJobRetryInterval(
cfg.Scheduler.RetryInterval) alongside the other SetXxxInterval calls.
Regression coverage:
internal/service/job_test.go (3 new)
- TestJobService_RetryFailedJobs_EligibleJobTransitionsAndAudits
- TestJobService_RetryFailedJobs_SkipsJobsAtMaxAttempts
- TestJobService_RetryFailedJobs_NoAuditServiceOK
internal/scheduler/scheduler_test.go (3 new)
- TestScheduler_JobRetryLoop_CallsService
- TestScheduler_JobRetryLoop_IdempotencyGuard
- TestScheduler_JobRetryLoop_WaitForCompletion
The service tests assert status transitions, attempt-cap short-
circuiting, and audit event shape (actor='system', action='job_retry',
details keys). The scheduler tests assert the loop invokes the service,
the atomic.Bool guard skips overlapping ticks with the expected
'still running, skipping tick' log, and WaitForCompletion drains the
in-flight tick on Stop.
Residual follow-up (not in scope for this commit):
internal/service/renewal.go:RetryFailedJobs is a parallel dead-code
duplicate of the same logic on RenewalService — untested and has no
runtime caller. The audit finding called this out as 'implemented
twice'. Removing it is a separate cleanup and does not block the
Option-A wiring this commit delivers.
Files:
cmd/server/main.go — SetAuditService + SetJobRetryInterval
internal/config/config.go — RetryInterval field + env + validate
internal/scheduler/scheduler.go — 10th loop, interface, field, setter
internal/scheduler/scheduler_test.go — 3 new scheduler-loop tests
internal/service/job.go — RetryFailedJobs audit emission + SetAuditService
internal/service/job_test.go — 3 new service-layer tests
This commit is contained in:
@@ -16,8 +16,16 @@ type RenewalServicer interface {
|
||||
}
|
||||
|
||||
// JobServicer defines the interface for job processing used by the scheduler.
|
||||
//
|
||||
// RetryFailedJobs was added to close coverage gap I-001: JobService.RetryFailedJobs
|
||||
// existed and was unit-tested but had no runtime caller prior to this loop being
|
||||
// wired. The scheduler now drives it on an independent tick so failed jobs whose
|
||||
// attempt counter is below MaxAttempts are periodically reset to Pending for the
|
||||
// job processor to pick up again. maxRetries is advisory (per-job gating uses
|
||||
// each job's own Attempts/MaxAttempts fields).
|
||||
type JobServicer interface {
|
||||
ProcessPendingJobs(ctx context.Context) error
|
||||
RetryFailedJobs(ctx context.Context, maxRetries int) error
|
||||
}
|
||||
|
||||
// AgentServicer defines the interface for agent health checks used by the scheduler.
|
||||
@@ -67,6 +75,7 @@ type Scheduler struct {
|
||||
// Configurable tick intervals
|
||||
renewalCheckInterval time.Duration
|
||||
jobProcessorInterval time.Duration
|
||||
jobRetryInterval time.Duration
|
||||
agentHealthCheckInterval time.Duration
|
||||
notificationProcessInterval time.Duration
|
||||
shortLivedExpiryCheckInterval time.Duration
|
||||
@@ -78,6 +87,7 @@ type Scheduler struct {
|
||||
// Idempotency guards: prevent duplicate execution of slow jobs
|
||||
renewalCheckRunning atomic.Bool
|
||||
jobProcessorRunning atomic.Bool
|
||||
jobRetryRunning atomic.Bool
|
||||
agentHealthCheckRunning atomic.Bool
|
||||
notificationProcessRunning atomic.Bool
|
||||
shortLivedExpiryCheckRunning atomic.Bool
|
||||
@@ -110,6 +120,7 @@ func NewScheduler(
|
||||
// Default intervals
|
||||
renewalCheckInterval: 1 * time.Hour,
|
||||
jobProcessorInterval: 30 * time.Second,
|
||||
jobRetryInterval: 5 * time.Minute,
|
||||
agentHealthCheckInterval: 2 * time.Minute,
|
||||
notificationProcessInterval: 1 * time.Minute,
|
||||
shortLivedExpiryCheckInterval: 30 * time.Second,
|
||||
@@ -141,6 +152,13 @@ func (s *Scheduler) SetJobProcessorInterval(d time.Duration) {
|
||||
s.jobProcessorInterval = d
|
||||
}
|
||||
|
||||
// SetJobRetryInterval configures the interval for the failed-job retry loop
|
||||
// (coverage gap I-001). Defaults to 5 minutes; honors
|
||||
// CERTCTL_SCHEDULER_RETRY_INTERVAL when wired from config.
|
||||
func (s *Scheduler) SetJobRetryInterval(d time.Duration) {
|
||||
s.jobRetryInterval = d
|
||||
}
|
||||
|
||||
// SetAgentHealthCheckInterval configures the interval for agent health checks.
|
||||
func (s *Scheduler) SetAgentHealthCheckInterval(d time.Duration) {
|
||||
s.agentHealthCheckInterval = d
|
||||
@@ -193,7 +211,10 @@ func (s *Scheduler) Start(ctx context.Context) <-chan struct{} {
|
||||
|
||||
// Track all loop goroutines in the WaitGroup so WaitForCompletion
|
||||
// blocks until they've fully exited (prevents test races).
|
||||
loopCount := 5
|
||||
// Base count is 6: renewal, job processor, job retry (I-001),
|
||||
// agent health, notification, short-lived expiry. Optional loops
|
||||
// (network scan, digest, health check, cloud discovery) add to this.
|
||||
loopCount := 6
|
||||
if s.networkScanService != nil {
|
||||
loopCount++
|
||||
}
|
||||
@@ -210,6 +231,7 @@ func (s *Scheduler) Start(ctx context.Context) <-chan struct{} {
|
||||
|
||||
go func() { defer s.wg.Done(); s.renewalCheckLoop(ctx) }()
|
||||
go func() { defer s.wg.Done(); s.jobProcessorLoop(ctx) }()
|
||||
go func() { defer s.wg.Done(); s.jobRetryLoop(ctx) }()
|
||||
go func() { defer s.wg.Done(); s.agentHealthCheckLoop(ctx) }()
|
||||
go func() { defer s.wg.Done(); s.notificationProcessLoop(ctx) }()
|
||||
go func() { defer s.wg.Done(); s.shortLivedExpiryCheckLoop(ctx) }()
|
||||
@@ -334,6 +356,63 @@ func (s *Scheduler) runJobProcessor(ctx context.Context) {
|
||||
}
|
||||
}
|
||||
|
||||
// jobRetryLoop runs every jobRetryInterval and transitions eligible Failed jobs
|
||||
// back to Pending so the job processor can pick them up again. Closes coverage
|
||||
// gap I-001 — JobService.RetryFailedJobs had no runtime caller prior to this
|
||||
// loop being wired. Runs immediately on start, then every interval.
|
||||
// Uses atomic.Bool to prevent duplicate execution if the previous retry sweep
|
||||
// is still running.
|
||||
func (s *Scheduler) jobRetryLoop(ctx context.Context) {
|
||||
ticker := time.NewTicker(s.jobRetryInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
// Run immediately on start (with idempotency guard)
|
||||
s.jobRetryRunning.Store(true)
|
||||
s.wg.Add(1)
|
||||
go func() {
|
||||
defer s.wg.Done()
|
||||
defer s.jobRetryRunning.Store(false)
|
||||
s.runJobRetry(ctx)
|
||||
}()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
if !s.jobRetryRunning.CompareAndSwap(false, true) {
|
||||
s.logger.Warn("job retry still running, skipping tick")
|
||||
continue
|
||||
}
|
||||
s.wg.Add(1)
|
||||
go func() {
|
||||
defer s.wg.Done()
|
||||
defer s.jobRetryRunning.Store(false)
|
||||
s.runJobRetry(ctx)
|
||||
}()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// runJobRetry executes a single failed-job retry cycle with error recovery.
|
||||
// Uses the same 2-minute per-tick timeout as runJobProcessor; RetryFailedJobs
|
||||
// issues one SELECT and one UPDATE per eligible job (cheap), so this headroom
|
||||
// covers very large failure backlogs without starving the loop.
|
||||
func (s *Scheduler) runJobRetry(ctx context.Context) {
|
||||
opCtx, cancel := context.WithTimeout(ctx, 2*time.Minute)
|
||||
defer cancel()
|
||||
// maxRetries is advisory at the service layer (per-job gating uses each
|
||||
// job's own Attempts/MaxAttempts). Passing 3 matches the conventional
|
||||
// default seen across the codebase's job creation paths.
|
||||
if err := s.jobService.RetryFailedJobs(opCtx, 3); err != nil {
|
||||
s.logger.Error("job retry failed",
|
||||
"error", err,
|
||||
"interval", s.jobRetryInterval.String())
|
||||
} else {
|
||||
s.logger.Debug("job retry completed")
|
||||
}
|
||||
}
|
||||
|
||||
// agentHealthCheckLoop runs every agentHealthCheckInterval and marks stale agents as offline.
|
||||
// An agent is considered stale if it hasn't sent a heartbeat within the health check interval.
|
||||
// If an error occurs, it logs the error but continues running.
|
||||
|
||||
Reference in New Issue
Block a user