mirror of
https://github.com/shankar0123/certctl.git
synced 2026-06-07 16:51:31 +00:00
I-003: job timeout reaper closes AwaitingCSR/AwaitingApproval gap
Add 11th always-on scheduler loop that transitions jobs stuck in
AwaitingCSR (default 24h TTL) or AwaitingApproval (default 168h TTL)
to Failed. I-001's retry loop then auto-promotes eligible Failed jobs
back to Pending. No new status enum, no schema migration.
- JobRepository.ListTimedOutAwaitingJobs with per-status cutoff WHERE
- JobService.ReapTimedOutJobs mirrors RetryFailedJobs structure
- Scheduler jobTimeoutLoop with atomic.Bool idempotency guard, 2m
per-tick context, WaitGroup shutdown drain
- Config: CERTCTL_JOB_TIMEOUT_INTERVAL (10m), CERTCTL_JOB_AWAITING_CSR_TIMEOUT
(24h), CERTCTL_JOB_AWAITING_APPROVAL_TIMEOUT (168h)
- Audit event per transition: actor=system, actorType=System,
action=job_timeout, details={old_status, new_status, timeout_reason,
age_hours}
- 14 new tests: 3 config, 7 service, 4 scheduler
This commit is contained in:
@@ -469,6 +469,17 @@ func main() {
|
||||
"sources", cloudDiscoveryService.SourceCount())
|
||||
}
|
||||
|
||||
|
||||
// Wire job timeout reaper (I-003)
|
||||
sched.SetJobReaperService(jobService)
|
||||
sched.SetJobTimeoutInterval(cfg.Scheduler.JobTimeoutInterval)
|
||||
sched.SetAwaitingCSRTimeout(cfg.Scheduler.AwaitingCSRTimeout)
|
||||
sched.SetAwaitingApprovalTimeout(cfg.Scheduler.AwaitingApprovalTimeout)
|
||||
logger.Info("job timeout reaper enabled",
|
||||
"interval", cfg.Scheduler.JobTimeoutInterval.String(),
|
||||
"csr_timeout", cfg.Scheduler.AwaitingCSRTimeout.String(),
|
||||
"approval_timeout", cfg.Scheduler.AwaitingApprovalTimeout.String())
|
||||
|
||||
// Start scheduler
|
||||
logger.Info("starting scheduler")
|
||||
startedChan := sched.Start(ctx)
|
||||
|
||||
@@ -715,6 +715,29 @@ type SchedulerConfig struct {
|
||||
// had no caller prior to this loop being wired).
|
||||
// Setting: CERTCTL_SCHEDULER_RETRY_INTERVAL environment variable.
|
||||
RetryInterval time.Duration
|
||||
|
||||
// JobTimeoutInterval is how often the reaper loop sweeps AwaitingCSR and
|
||||
// AwaitingApproval jobs for TTL expiration. Default: 10 minutes. Minimum: 1
|
||||
// second. Timed-out jobs are transitioned to Failed with a descriptive error
|
||||
// message; I-001's retry loop then auto-promotes eligible Failed jobs back
|
||||
// to Pending (closes coverage gap I-003).
|
||||
// Setting: CERTCTL_JOB_TIMEOUT_INTERVAL environment variable.
|
||||
JobTimeoutInterval time.Duration
|
||||
|
||||
// AwaitingCSRTimeout is the maximum age an AwaitingCSR job can remain in
|
||||
// that state before the reaper transitions it to Failed. Default: 24 hours.
|
||||
// An agent that hasn't submitted a CSR within this window is presumed
|
||||
// unreachable. Minimum: 1 second.
|
||||
// Setting: CERTCTL_JOB_AWAITING_CSR_TIMEOUT environment variable.
|
||||
AwaitingCSRTimeout time.Duration
|
||||
|
||||
// AwaitingApprovalTimeout is the maximum age an AwaitingApproval job can
|
||||
// remain in that state before the reaper transitions it to Failed. Default:
|
||||
// 168 hours (7 days). Reviewers who haven't approved within this window
|
||||
// force the renewal to fail loudly rather than silently stall. Minimum: 1
|
||||
// second.
|
||||
// Setting: CERTCTL_JOB_AWAITING_APPROVAL_TIMEOUT environment variable.
|
||||
AwaitingApprovalTimeout time.Duration
|
||||
}
|
||||
|
||||
// LogConfig contains logging configuration.
|
||||
@@ -816,6 +839,9 @@ func Load() (*Config, error) {
|
||||
AgentHealthCheckInterval: getEnvDuration("CERTCTL_SCHEDULER_AGENT_HEALTH_CHECK_INTERVAL", 2*time.Minute),
|
||||
NotificationProcessInterval: getEnvDuration("CERTCTL_SCHEDULER_NOTIFICATION_PROCESS_INTERVAL", 1*time.Minute),
|
||||
RetryInterval: getEnvDuration("CERTCTL_SCHEDULER_RETRY_INTERVAL", 5*time.Minute),
|
||||
JobTimeoutInterval: getEnvDuration("CERTCTL_JOB_TIMEOUT_INTERVAL", 10*time.Minute),
|
||||
AwaitingCSRTimeout: getEnvDuration("CERTCTL_JOB_AWAITING_CSR_TIMEOUT", 24*time.Hour),
|
||||
AwaitingApprovalTimeout: getEnvDuration("CERTCTL_JOB_AWAITING_APPROVAL_TIMEOUT", 168*time.Hour),
|
||||
},
|
||||
Log: LogConfig{
|
||||
Level: getEnv("CERTCTL_LOG_LEVEL", "info"),
|
||||
@@ -1087,6 +1113,18 @@ func (c *Config) Validate() error {
|
||||
return fmt.Errorf("retry interval must be at least 1 second")
|
||||
}
|
||||
|
||||
if c.Scheduler.JobTimeoutInterval < 1*time.Second {
|
||||
return fmt.Errorf("job timeout interval must be at least 1 second")
|
||||
}
|
||||
|
||||
if c.Scheduler.AwaitingCSRTimeout < 1*time.Second {
|
||||
return fmt.Errorf("awaiting CSR timeout must be at least 1 second")
|
||||
}
|
||||
|
||||
if c.Scheduler.AwaitingApprovalTimeout < 1*time.Second {
|
||||
return fmt.Errorf("awaiting approval timeout must be at least 1 second")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"log/slog"
|
||||
"os"
|
||||
"testing"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -329,6 +330,9 @@ func TestValidate_ValidConfig(t *testing.T) {
|
||||
AgentHealthCheckInterval: 2 * time.Minute,
|
||||
NotificationProcessInterval: 1 * time.Minute,
|
||||
RetryInterval: 5 * time.Minute,
|
||||
JobTimeoutInterval: 10 * time.Minute,
|
||||
AwaitingCSRTimeout: 24 * time.Hour,
|
||||
AwaitingApprovalTimeout: 168 * time.Hour,
|
||||
},
|
||||
}
|
||||
if err := cfg.Validate(); err != nil {
|
||||
@@ -349,6 +353,9 @@ func TestValidate_AuthTypeNone(t *testing.T) {
|
||||
AgentHealthCheckInterval: 2 * time.Minute,
|
||||
NotificationProcessInterval: 1 * time.Minute,
|
||||
RetryInterval: 5 * time.Minute,
|
||||
JobTimeoutInterval: 10 * time.Minute,
|
||||
AwaitingCSRTimeout: 24 * time.Hour,
|
||||
AwaitingApprovalTimeout: 168 * time.Hour,
|
||||
},
|
||||
}
|
||||
if err := cfg.Validate(); err != nil {
|
||||
@@ -708,3 +715,120 @@ func TestGetEnvBool(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
// I-003: Job timeout reaper configuration tests
|
||||
func TestConfig_Scheduler_JobTimeoutDefaults(t *testing.T) {
|
||||
clearCertctlEnv(t)
|
||||
setMinimalValidEnv(t)
|
||||
// Explicitly unset the three I-003 env vars to exercise the default path.
|
||||
t.Setenv("CERTCTL_JOB_TIMEOUT_INTERVAL", "")
|
||||
t.Setenv("CERTCTL_JOB_AWAITING_CSR_TIMEOUT", "")
|
||||
t.Setenv("CERTCTL_JOB_AWAITING_APPROVAL_TIMEOUT", "")
|
||||
|
||||
cfg, err := Load()
|
||||
if err != nil {
|
||||
t.Fatalf("Load() error: %v", err)
|
||||
}
|
||||
|
||||
if cfg.Scheduler.JobTimeoutInterval != 10*time.Minute {
|
||||
t.Errorf("JobTimeoutInterval = %v, want 10m", cfg.Scheduler.JobTimeoutInterval)
|
||||
}
|
||||
if cfg.Scheduler.AwaitingCSRTimeout != 24*time.Hour {
|
||||
t.Errorf("AwaitingCSRTimeout = %v, want 24h", cfg.Scheduler.AwaitingCSRTimeout)
|
||||
}
|
||||
if cfg.Scheduler.AwaitingApprovalTimeout != 168*time.Hour {
|
||||
t.Errorf("AwaitingApprovalTimeout = %v, want 168h", cfg.Scheduler.AwaitingApprovalTimeout)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConfig_Scheduler_JobTimeoutEnvOverride(t *testing.T) {
|
||||
clearCertctlEnv(t)
|
||||
setMinimalValidEnv(t)
|
||||
t.Setenv("CERTCTL_JOB_TIMEOUT_INTERVAL", "15m")
|
||||
t.Setenv("CERTCTL_JOB_AWAITING_CSR_TIMEOUT", "48h")
|
||||
t.Setenv("CERTCTL_JOB_AWAITING_APPROVAL_TIMEOUT", "336h")
|
||||
|
||||
cfg, err := Load()
|
||||
if err != nil {
|
||||
t.Fatalf("Load() error: %v", err)
|
||||
}
|
||||
|
||||
if cfg.Scheduler.JobTimeoutInterval != 15*time.Minute {
|
||||
t.Errorf("JobTimeoutInterval = %v, want 15m", cfg.Scheduler.JobTimeoutInterval)
|
||||
}
|
||||
if cfg.Scheduler.AwaitingCSRTimeout != 48*time.Hour {
|
||||
t.Errorf("AwaitingCSRTimeout = %v, want 48h", cfg.Scheduler.AwaitingCSRTimeout)
|
||||
}
|
||||
if cfg.Scheduler.AwaitingApprovalTimeout != 336*time.Hour {
|
||||
t.Errorf("AwaitingApprovalTimeout = %v, want 336h", cfg.Scheduler.AwaitingApprovalTimeout)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConfig_Scheduler_JobTimeoutValidation(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
field string
|
||||
value time.Duration
|
||||
wantErrMsg string
|
||||
}{
|
||||
{
|
||||
"JobTimeoutInterval too small",
|
||||
"JobTimeoutInterval",
|
||||
500 * time.Millisecond,
|
||||
"job timeout interval must be at least 1 second",
|
||||
},
|
||||
{
|
||||
"AwaitingCSRTimeout too small",
|
||||
"AwaitingCSRTimeout",
|
||||
500 * time.Millisecond,
|
||||
"awaiting CSR timeout must be at least 1 second",
|
||||
},
|
||||
{
|
||||
"AwaitingApprovalTimeout too small",
|
||||
"AwaitingApprovalTimeout",
|
||||
500 * time.Millisecond,
|
||||
"awaiting approval timeout must be at least 1 second",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
// Start from a fully valid config so the I-003 timeout checks
|
||||
// are the only potential failure point.
|
||||
cfg := &Config{
|
||||
Server: ServerConfig{Port: 8080},
|
||||
Database: DatabaseConfig{URL: "postgres://localhost/certctl", MaxConnections: 25},
|
||||
Log: LogConfig{Level: "info", Format: "json"},
|
||||
Auth: AuthConfig{Type: "api-key", Secret: "test-secret"},
|
||||
Keygen: KeygenConfig{Mode: "agent"},
|
||||
Scheduler: SchedulerConfig{
|
||||
RenewalCheckInterval: 1 * time.Minute,
|
||||
JobProcessorInterval: 1 * time.Minute,
|
||||
AgentHealthCheckInterval: 1 * time.Minute,
|
||||
NotificationProcessInterval: 1 * time.Minute,
|
||||
RetryInterval: 1 * time.Minute,
|
||||
JobTimeoutInterval: 10 * time.Minute,
|
||||
AwaitingCSRTimeout: 24 * time.Hour,
|
||||
AwaitingApprovalTimeout: 168 * time.Hour,
|
||||
},
|
||||
}
|
||||
|
||||
// Override the specific field under test
|
||||
switch tt.field {
|
||||
case "JobTimeoutInterval":
|
||||
cfg.Scheduler.JobTimeoutInterval = tt.value
|
||||
case "AwaitingCSRTimeout":
|
||||
cfg.Scheduler.AwaitingCSRTimeout = tt.value
|
||||
case "AwaitingApprovalTimeout":
|
||||
cfg.Scheduler.AwaitingApprovalTimeout = tt.value
|
||||
}
|
||||
|
||||
err := cfg.Validate()
|
||||
if err == nil {
|
||||
t.Fatalf("Validate() = nil, want error containing %q", tt.wantErrMsg)
|
||||
}
|
||||
if !strings.Contains(err.Error(), tt.wantErrMsg) {
|
||||
t.Errorf("Validate() error = %q, want to contain %q", err.Error(), tt.wantErrMsg)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -742,6 +742,25 @@ func (m *mockJobRepository) ClaimPendingByAgentID(ctx context.Context, agentID s
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// ListTimedOutAwaitingJobs is the I-003 integration-mock stub. Returns jobs whose
|
||||
// created_at predates the relevant cutoff for their status.
|
||||
func (m *mockJobRepository) ListTimedOutAwaitingJobs(ctx context.Context, csrCutoff, approvalCutoff time.Time) ([]*domain.Job, error) {
|
||||
var jobs []*domain.Job
|
||||
for _, j := range m.jobs {
|
||||
switch j.Status {
|
||||
case domain.JobStatusAwaitingCSR:
|
||||
if j.CreatedAt.Before(csrCutoff) {
|
||||
jobs = append(jobs, j)
|
||||
}
|
||||
case domain.JobStatusAwaitingApproval:
|
||||
if j.CreatedAt.Before(approvalCutoff) {
|
||||
jobs = append(jobs, j)
|
||||
}
|
||||
}
|
||||
}
|
||||
return jobs, nil
|
||||
}
|
||||
|
||||
type mockAuditRepository struct {
|
||||
events []*domain.AuditEvent
|
||||
}
|
||||
|
||||
@@ -151,6 +151,11 @@ type JobRepository interface {
|
||||
// to Running) and locks AwaitingCSR jobs against concurrent observers (leaving state intact,
|
||||
// since the CSR-submission path drives the next transition). H-6 (CWE-362) race remediation.
|
||||
ClaimPendingByAgentID(ctx context.Context, agentID string) ([]*domain.Job, error)
|
||||
// ListTimedOutAwaitingJobs returns jobs stuck in AwaitingCSR (created before csrCutoff) or
|
||||
// AwaitingApproval (created before approvalCutoff). The reaper loop transitions them to
|
||||
// Failed; I-001's retry loop then auto-promotes eligible Failed jobs back to Pending.
|
||||
// I-003 coverage-gap closure.
|
||||
ListTimedOutAwaitingJobs(ctx context.Context, csrCutoff, approvalCutoff time.Time) ([]*domain.Job, error)
|
||||
}
|
||||
|
||||
// RenewalPolicyRepository defines operations for managing renewal policies.
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/shankar0123/certctl/internal/domain"
|
||||
@@ -570,6 +571,41 @@ func (r *JobRepository) ClaimPendingByAgentID(ctx context.Context, agentID strin
|
||||
return append(pendingJobs, csrJobs...), nil
|
||||
}
|
||||
|
||||
// ListTimedOutAwaitingJobs returns jobs stuck in AwaitingCSR or AwaitingApproval past
|
||||
// their respective cutoff timestamps (created_at < cutoff). The reaper loop transitions
|
||||
// them to Failed; I-001's retry loop then auto-promotes eligible Failed jobs back to
|
||||
// Pending. I-003 coverage-gap closure.
|
||||
func (r *JobRepository) ListTimedOutAwaitingJobs(ctx context.Context, csrCutoff, approvalCutoff time.Time) ([]*domain.Job, error) {
|
||||
rows, err := r.db.QueryContext(ctx, `
|
||||
SELECT id, type, certificate_id, target_id, agent_id, status, attempts, max_attempts,
|
||||
last_error, scheduled_at, started_at, completed_at, created_at
|
||||
FROM jobs
|
||||
WHERE (status = $1 AND created_at < $2)
|
||||
OR (status = $3 AND created_at < $4)
|
||||
ORDER BY created_at ASC
|
||||
`, domain.JobStatusAwaitingCSR, csrCutoff, domain.JobStatusAwaitingApproval, approvalCutoff)
|
||||
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to query timed-out awaiting jobs: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var jobs []*domain.Job
|
||||
for rows.Next() {
|
||||
job, err := scanJob(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
jobs = append(jobs, job)
|
||||
}
|
||||
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("error iterating timed-out job rows: %w", err)
|
||||
}
|
||||
|
||||
return jobs, nil
|
||||
}
|
||||
|
||||
// scanJob scans a job from a row or rows
|
||||
func scanJob(scanner interface {
|
||||
Scan(...interface{}) error
|
||||
|
||||
@@ -58,6 +58,11 @@ type CloudDiscoveryServicer interface {
|
||||
DiscoverAll(ctx context.Context) (int, []error)
|
||||
}
|
||||
|
||||
// JobReaperService defines the interface for job timeout reaping used by the scheduler.
|
||||
type JobReaperService interface {
|
||||
ReapTimedOutJobs(ctx context.Context, csrTTL, approvalTTL time.Duration) error
|
||||
}
|
||||
|
||||
// Scheduler manages background jobs and periodic tasks for the certificate control plane.
|
||||
// It runs multiple concurrent loops for renewal checks, job processing, agent health checks,
|
||||
// and notification processing.
|
||||
@@ -70,6 +75,7 @@ type Scheduler struct {
|
||||
digestService DigestServicer
|
||||
healthCheckService HealthCheckServicer
|
||||
cloudDiscoveryService CloudDiscoveryServicer
|
||||
jobReaper JobReaperService
|
||||
logger *slog.Logger
|
||||
|
||||
// Configurable tick intervals
|
||||
@@ -83,6 +89,9 @@ type Scheduler struct {
|
||||
digestInterval time.Duration
|
||||
healthCheckInterval time.Duration
|
||||
cloudDiscoveryInterval time.Duration
|
||||
jobTimeoutInterval time.Duration
|
||||
awaitingCSRTimeout time.Duration
|
||||
awaitingApprovalTimeout time.Duration
|
||||
|
||||
// Idempotency guards: prevent duplicate execution of slow jobs
|
||||
renewalCheckRunning atomic.Bool
|
||||
@@ -95,6 +104,7 @@ type Scheduler struct {
|
||||
digestRunning atomic.Bool
|
||||
healthCheckRunning atomic.Bool
|
||||
cloudDiscoveryRunning atomic.Bool
|
||||
jobTimeoutRunning atomic.Bool
|
||||
|
||||
// Graceful shutdown: wait for in-flight work to complete
|
||||
wg sync.WaitGroup
|
||||
@@ -128,6 +138,7 @@ func NewScheduler(
|
||||
digestInterval: 24 * time.Hour,
|
||||
healthCheckInterval: 60 * time.Second,
|
||||
cloudDiscoveryInterval: 6 * time.Hour,
|
||||
jobTimeoutInterval: 10 * time.Minute,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -201,6 +212,26 @@ func (s *Scheduler) SetCloudDiscoveryInterval(d time.Duration) {
|
||||
s.cloudDiscoveryInterval = d
|
||||
}
|
||||
|
||||
|
||||
// SetJobReaperService sets the job reaper service (I-003).
|
||||
func (s *Scheduler) SetJobReaperService(jr JobReaperService) {
|
||||
s.jobReaper = jr
|
||||
}
|
||||
|
||||
// SetJobTimeoutInterval sets the job timeout reaper tick interval (I-003).
|
||||
func (s *Scheduler) SetJobTimeoutInterval(d time.Duration) {
|
||||
s.jobTimeoutInterval = d
|
||||
}
|
||||
|
||||
// SetAwaitingCSRTimeout sets the AwaitingCSR TTL (I-003).
|
||||
func (s *Scheduler) SetAwaitingCSRTimeout(d time.Duration) {
|
||||
s.awaitingCSRTimeout = d
|
||||
}
|
||||
|
||||
// SetAwaitingApprovalTimeout sets the AwaitingApproval TTL (I-003).
|
||||
func (s *Scheduler) SetAwaitingApprovalTimeout(d time.Duration) {
|
||||
s.awaitingApprovalTimeout = d
|
||||
}
|
||||
// Start initiates all background scheduler loops. It returns a channel that signals
|
||||
// when the scheduler has started all loops. The scheduler runs until the context is cancelled.
|
||||
func (s *Scheduler) Start(ctx context.Context) <-chan struct{} {
|
||||
@@ -211,10 +242,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).
|
||||
// Base count is 6: renewal, job processor, job retry (I-001),
|
||||
// agent health, notification, short-lived expiry. Optional loops
|
||||
// Base count is 7: renewal, job processor, job retry (I-001),
|
||||
// job timeout (I-003), agent health, notification, short-lived expiry. Optional loops
|
||||
// (network scan, digest, health check, cloud discovery) add to this.
|
||||
loopCount := 6
|
||||
loopCount := 7
|
||||
if s.networkScanService != nil {
|
||||
loopCount++
|
||||
}
|
||||
@@ -232,6 +263,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.jobTimeoutLoop(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) }()
|
||||
@@ -413,6 +445,61 @@ func (s *Scheduler) runJobRetry(ctx context.Context) {
|
||||
}
|
||||
}
|
||||
|
||||
// jobTimeoutLoop runs every jobTimeoutInterval and transitions jobs stuck in
|
||||
// AwaitingCSR or AwaitingApproval to Failed if they exceed their TTL. I-001's
|
||||
// retry loop then auto-promotes eligible Failed jobs back to Pending. Closes
|
||||
// coverage gap I-003. Uses atomic.Bool to prevent duplicate execution.
|
||||
func (s *Scheduler) jobTimeoutLoop(ctx context.Context) {
|
||||
ticker := time.NewTicker(s.jobTimeoutInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
// Run immediately on start (with idempotency guard)
|
||||
s.jobTimeoutRunning.Store(true)
|
||||
s.wg.Add(1)
|
||||
go func() {
|
||||
defer s.wg.Done()
|
||||
defer s.jobTimeoutRunning.Store(false)
|
||||
s.runJobTimeout(ctx)
|
||||
}()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
if !s.jobTimeoutRunning.CompareAndSwap(false, true) {
|
||||
s.logger.Warn("job timeout reaper still running, skipping tick")
|
||||
continue
|
||||
}
|
||||
s.wg.Add(1)
|
||||
go func() {
|
||||
defer s.wg.Done()
|
||||
defer s.jobTimeoutRunning.Store(false)
|
||||
s.runJobTimeout(ctx)
|
||||
}()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// runJobTimeout executes a single job timeout reaping cycle with error recovery.
|
||||
// When no JobReaperService has been wired (e.g. in tests that don't exercise
|
||||
// I-003) the call is a safe no-op, preserving the always-on loop topology
|
||||
// described in I-003 without forcing every consumer to wire a reaper.
|
||||
func (s *Scheduler) runJobTimeout(ctx context.Context) {
|
||||
if s.jobReaper == nil {
|
||||
return
|
||||
}
|
||||
opCtx, cancel := context.WithTimeout(ctx, 2*time.Minute)
|
||||
defer cancel()
|
||||
if err := s.jobReaper.ReapTimedOutJobs(opCtx, s.awaitingCSRTimeout, s.awaitingApprovalTimeout); err != nil {
|
||||
s.logger.Error("job timeout reaper failed",
|
||||
"error", err,
|
||||
"interval", s.jobTimeoutInterval.String())
|
||||
} else {
|
||||
s.logger.Debug("job timeout reaper 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.
|
||||
|
||||
@@ -85,6 +85,13 @@ type mockJobService struct {
|
||||
retryMaxRetriesSeen []int
|
||||
retrySlowDelay time.Duration
|
||||
retryShouldError bool
|
||||
|
||||
// Timeout reaper tracking (coverage gap I-003)
|
||||
reapCallCount int
|
||||
reapCallTimes []time.Time
|
||||
reapSlowDelay time.Duration
|
||||
reapShouldError bool
|
||||
reapCtxHasDeadline bool
|
||||
}
|
||||
|
||||
func (m *mockJobService) ProcessPendingJobs(ctx context.Context) error {
|
||||
@@ -131,6 +138,33 @@ func (m *mockJobService) RetryFailedJobs(ctx context.Context, maxRetries int) er
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
// ReapTimedOutJobs is the scheduler-driven counterpart to ProcessPendingJobs that
|
||||
// covers coverage gap I-003: JobService.ReapTimedOutJobs (via JobReaperService interface)
|
||||
// had no runtime caller prior to the jobTimeoutLoop being wired.
|
||||
func (m *mockJobService) ReapTimedOutJobs(ctx context.Context, csrTTL, approvalTTL time.Duration) error {
|
||||
m.mu.Lock()
|
||||
m.reapCallCount++
|
||||
m.reapCallTimes = append(m.reapCallTimes, time.Now())
|
||||
// Track whether context has a deadline set
|
||||
_, hasDeadline := ctx.Deadline()
|
||||
m.reapCtxHasDeadline = hasDeadline
|
||||
m.mu.Unlock()
|
||||
|
||||
if m.reapSlowDelay > 0 {
|
||||
select {
|
||||
case <-time.After(m.reapSlowDelay):
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
if m.reapShouldError {
|
||||
return context.Canceled
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// mockAgentService is a mock implementation for testing.
|
||||
type mockAgentService struct {
|
||||
mu sync.Mutex
|
||||
@@ -1141,3 +1175,186 @@ func TestScheduler_JobRetryLoop_WaitForCompletion(t *testing.T) {
|
||||
}
|
||||
t.Logf("retry loop graceful shutdown completed in %v after %d in-flight sweep(s)", elapsed, retryCount)
|
||||
}
|
||||
|
||||
// TestScheduler_JobTimeoutLoop_NormalTick verifies that the job timeout reaper
|
||||
// loop ticks at the specified interval (coverage gap I-003).
|
||||
func TestScheduler_JobTimeoutLoop_NormalTick(t *testing.T) {
|
||||
logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
|
||||
renewalMock := &mockRenewalService{}
|
||||
jobMock := &mockJobService{}
|
||||
agentMock := &mockAgentService{}
|
||||
notificationMock := &mockNotificationService{}
|
||||
networkMock := &mockNetworkScanService{}
|
||||
|
||||
sched := NewScheduler(renewalMock, jobMock, agentMock, notificationMock, networkMock, logger)
|
||||
sched.SetRenewalCheckInterval(10 * time.Second)
|
||||
sched.SetJobProcessorInterval(10 * time.Second)
|
||||
sched.SetAgentHealthCheckInterval(10 * time.Second)
|
||||
sched.SetNotificationProcessInterval(10 * time.Second)
|
||||
sched.SetNetworkScanInterval(10 * time.Second)
|
||||
sched.SetJobRetryInterval(10 * time.Second)
|
||||
sched.SetJobTimeoutInterval(50 * time.Millisecond)
|
||||
sched.SetAwaitingCSRTimeout(24 * time.Hour)
|
||||
sched.SetAwaitingApprovalTimeout(168 * time.Hour)
|
||||
sched.SetJobReaperService(jobMock)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
<-sched.Start(ctx)
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
cancel()
|
||||
if err := sched.WaitForCompletion(2 * time.Second); err != nil {
|
||||
t.Fatalf("WaitForCompletion: %v", err)
|
||||
}
|
||||
|
||||
jobMock.mu.Lock()
|
||||
count := jobMock.reapCallCount
|
||||
jobMock.mu.Unlock()
|
||||
if count < 2 {
|
||||
t.Fatalf("expected >= 2 reap calls, got %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
// TestScheduler_JobTimeoutLoop_IdempotencyGuard verifies that the timeout reaper
|
||||
// uses an atomic guard to prevent concurrent execution (coverage gap I-003).
|
||||
func TestScheduler_JobTimeoutLoop_IdempotencyGuard(t *testing.T) {
|
||||
logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
|
||||
renewalMock := &mockRenewalService{}
|
||||
jobMock := &mockJobService{
|
||||
reapSlowDelay: 150 * time.Millisecond,
|
||||
}
|
||||
agentMock := &mockAgentService{}
|
||||
notificationMock := &mockNotificationService{}
|
||||
networkMock := &mockNetworkScanService{}
|
||||
|
||||
sched := NewScheduler(renewalMock, jobMock, agentMock, notificationMock, networkMock, logger)
|
||||
sched.SetRenewalCheckInterval(10 * time.Second)
|
||||
sched.SetJobProcessorInterval(10 * time.Second)
|
||||
sched.SetAgentHealthCheckInterval(10 * time.Second)
|
||||
sched.SetNotificationProcessInterval(10 * time.Second)
|
||||
sched.SetNetworkScanInterval(10 * time.Second)
|
||||
sched.SetJobRetryInterval(10 * time.Second)
|
||||
sched.SetJobTimeoutInterval(50 * time.Millisecond)
|
||||
sched.SetAwaitingCSRTimeout(24 * time.Hour)
|
||||
sched.SetAwaitingApprovalTimeout(168 * time.Hour)
|
||||
sched.SetJobReaperService(jobMock)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
<-sched.Start(ctx)
|
||||
time.Sleep(400 * time.Millisecond)
|
||||
|
||||
jobMock.mu.Lock()
|
||||
reapCount := jobMock.reapCallCount
|
||||
jobMock.mu.Unlock()
|
||||
|
||||
if reapCount > 3 {
|
||||
t.Logf("WARNING: reap called %d times in 400ms with 50ms interval and 150ms sweep — guard may not be working", reapCount)
|
||||
}
|
||||
|
||||
t.Logf("job timeout idempotency guard: %d calls in 400ms (50ms interval, 150ms sweep)", reapCount)
|
||||
|
||||
cancel()
|
||||
if err := sched.WaitForCompletion(2 * time.Second); err != nil {
|
||||
t.Fatalf("WaitForCompletion should succeed: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestScheduler_JobTimeoutLoop_ShutdownDrainsInFlight verifies that shutdown waits
|
||||
// for an in-flight timeout reaper to complete (coverage gap I-003).
|
||||
func TestScheduler_JobTimeoutLoop_ShutdownDrainsInFlight(t *testing.T) {
|
||||
logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
|
||||
renewalMock := &mockRenewalService{}
|
||||
jobMock := &mockJobService{
|
||||
reapSlowDelay: 100 * time.Millisecond,
|
||||
}
|
||||
agentMock := &mockAgentService{}
|
||||
notificationMock := &mockNotificationService{}
|
||||
networkMock := &mockNetworkScanService{}
|
||||
|
||||
sched := NewScheduler(renewalMock, jobMock, agentMock, notificationMock, networkMock, logger)
|
||||
sched.SetRenewalCheckInterval(10 * time.Second)
|
||||
sched.SetJobProcessorInterval(10 * time.Second)
|
||||
sched.SetAgentHealthCheckInterval(10 * time.Second)
|
||||
sched.SetNotificationProcessInterval(10 * time.Second)
|
||||
sched.SetNetworkScanInterval(10 * time.Second)
|
||||
sched.SetJobRetryInterval(10 * time.Second)
|
||||
sched.SetJobTimeoutInterval(50 * time.Millisecond)
|
||||
sched.SetAwaitingCSRTimeout(24 * time.Hour)
|
||||
sched.SetAwaitingApprovalTimeout(168 * time.Hour)
|
||||
sched.SetJobReaperService(jobMock)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
<-sched.Start(ctx)
|
||||
|
||||
// Let the immediate-start timeout reaper goroutine begin its 100ms sweep.
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
|
||||
// Initiate shutdown mid-sweep.
|
||||
cancel()
|
||||
|
||||
start := time.Now()
|
||||
err := sched.WaitForCompletion(5 * time.Second)
|
||||
elapsed := time.Since(start)
|
||||
|
||||
if err != nil {
|
||||
t.Fatalf("WaitForCompletion should not error: %v", err)
|
||||
}
|
||||
if elapsed > 5*time.Second {
|
||||
t.Fatalf("WaitForCompletion took longer than expected: %v", elapsed)
|
||||
}
|
||||
|
||||
jobMock.mu.Lock()
|
||||
reapCount := jobMock.reapCallCount
|
||||
jobMock.mu.Unlock()
|
||||
|
||||
if reapCount < 1 {
|
||||
t.Fatalf("expected timeout reaper to have started at least once before shutdown, got %d", reapCount)
|
||||
}
|
||||
t.Logf("timeout reaper graceful shutdown completed in %v after %d in-flight sweep(s)", elapsed, reapCount)
|
||||
}
|
||||
|
||||
// TestScheduler_JobTimeoutLoop_ContextDeadlineRespected verifies that the timeout
|
||||
// reaper receives a context with a deadline set for each tick (coverage gap I-003).
|
||||
func TestScheduler_JobTimeoutLoop_ContextDeadlineRespected(t *testing.T) {
|
||||
logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
|
||||
renewalMock := &mockRenewalService{}
|
||||
jobMock := &mockJobService{}
|
||||
agentMock := &mockAgentService{}
|
||||
notificationMock := &mockNotificationService{}
|
||||
networkMock := &mockNetworkScanService{}
|
||||
|
||||
sched := NewScheduler(renewalMock, jobMock, agentMock, notificationMock, networkMock, logger)
|
||||
sched.SetRenewalCheckInterval(10 * time.Second)
|
||||
sched.SetJobProcessorInterval(10 * time.Second)
|
||||
sched.SetAgentHealthCheckInterval(10 * time.Second)
|
||||
sched.SetNotificationProcessInterval(10 * time.Second)
|
||||
sched.SetNetworkScanInterval(10 * time.Second)
|
||||
sched.SetJobRetryInterval(10 * time.Second)
|
||||
sched.SetJobTimeoutInterval(50 * time.Millisecond)
|
||||
sched.SetAwaitingCSRTimeout(24 * time.Hour)
|
||||
sched.SetAwaitingApprovalTimeout(168 * time.Hour)
|
||||
sched.SetJobReaperService(jobMock)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
<-sched.Start(ctx)
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
cancel()
|
||||
if err := sched.WaitForCompletion(2 * time.Second); err != nil {
|
||||
t.Fatalf("WaitForCompletion: %v", err)
|
||||
}
|
||||
|
||||
jobMock.mu.Lock()
|
||||
hasDeadline := jobMock.reapCtxHasDeadline
|
||||
jobMock.mu.Unlock()
|
||||
|
||||
if !hasDeadline {
|
||||
t.Fatal("expected timeout reaper context to have a deadline set, but none found")
|
||||
}
|
||||
t.Log("timeout reaper context deadline verified")
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"time"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
@@ -236,6 +237,81 @@ func (s *JobService) RetryFailedJobs(ctx context.Context, maxRetries int) error
|
||||
return nil
|
||||
}
|
||||
|
||||
// ReapTimedOutJobs transitions jobs stuck in AwaitingCSR or AwaitingApproval
|
||||
// to Failed if they've exceeded their TTL. I-001's retry loop then auto-promotes
|
||||
// eligible Failed jobs back to Pending (closes coverage gap I-003).
|
||||
func (s *JobService) ReapTimedOutJobs(ctx context.Context, csrTTL, approvalTTL time.Duration) error {
|
||||
s.logger.Debug("reaping timed-out jobs", "csr_ttl", csrTTL, "approval_ttl", approvalTTL)
|
||||
|
||||
now := time.Now()
|
||||
csrCutoff := now.Add(-csrTTL)
|
||||
approvalCutoff := now.Add(-approvalTTL)
|
||||
|
||||
timedOutJobs, err := s.jobRepo.ListTimedOutAwaitingJobs(ctx, csrCutoff, approvalCutoff)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to fetch timed-out jobs: %w", err)
|
||||
}
|
||||
|
||||
var reaped int
|
||||
|
||||
for _, job := range timedOutJobs {
|
||||
oldStatus := job.Status
|
||||
var (
|
||||
newErrMsg string
|
||||
reason string
|
||||
ttl time.Duration
|
||||
)
|
||||
switch job.Status {
|
||||
case domain.JobStatusAwaitingCSR:
|
||||
ttl = csrTTL
|
||||
reason = "csr_timeout"
|
||||
newErrMsg = fmt.Sprintf("timed out in %s after %s", oldStatus, csrTTL)
|
||||
case domain.JobStatusAwaitingApproval:
|
||||
ttl = approvalTTL
|
||||
reason = "approval_timeout"
|
||||
newErrMsg = fmt.Sprintf("timed out in %s after %s", oldStatus, approvalTTL)
|
||||
default:
|
||||
continue
|
||||
}
|
||||
_ = ttl
|
||||
|
||||
job.Status = domain.JobStatusFailed
|
||||
job.LastError = &newErrMsg
|
||||
|
||||
if err := s.jobRepo.Update(ctx, job); err != nil {
|
||||
s.logger.Error("failed to transition timed-out job",
|
||||
"job_id", job.ID,
|
||||
"old_status", oldStatus,
|
||||
"error", err)
|
||||
continue
|
||||
}
|
||||
|
||||
if s.auditService != nil {
|
||||
ageHours := time.Since(job.CreatedAt).Hours()
|
||||
if auditErr := s.auditService.RecordEvent(ctx, "system", domain.ActorTypeSystem,
|
||||
"job_timeout", "job", job.ID,
|
||||
map[string]interface{}{
|
||||
"old_status": string(oldStatus),
|
||||
"new_status": string(domain.JobStatusFailed),
|
||||
"timeout_reason": reason,
|
||||
"age_hours": ageHours,
|
||||
}); auditErr != nil {
|
||||
s.logger.Error("failed to record job timeout audit event",
|
||||
"job_id", job.ID,
|
||||
"error", auditErr)
|
||||
}
|
||||
}
|
||||
|
||||
reaped++
|
||||
}
|
||||
|
||||
s.logger.Info("job timeout reaper completed",
|
||||
"reaped", reaped,
|
||||
"total_timed_out", len(timedOutJobs))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetJobStatus returns the current status of a job.
|
||||
func (s *JobService) GetJobStatus(ctx context.Context, jobID string) (*domain.Job, error) {
|
||||
job, err := s.jobRepo.Get(ctx, jobID)
|
||||
|
||||
@@ -6,6 +6,8 @@ import (
|
||||
"errors"
|
||||
"log/slog"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -576,3 +578,356 @@ func TestJobService_RetryFailedJobs_NoAuditServiceOK(t *testing.T) {
|
||||
t.Errorf("expected status Pending after retry, got %s", got)
|
||||
}
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// ReapTimedOutJobs Tests (I-003 coverage closure)
|
||||
// =============================================================================
|
||||
|
||||
func TestJobService_ReapTimedOutJobs_AwaitingCSRTransitionsAndAudits(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
now := time.Now()
|
||||
job := &domain.Job{
|
||||
ID: "job-reap-csr-1",
|
||||
Type: domain.JobTypeRenewal,
|
||||
CertificateID: "cert-001",
|
||||
Status: domain.JobStatusAwaitingCSR,
|
||||
CreatedAt: now.Add(-36 * time.Hour), // 36 hours old
|
||||
ScheduledAt: now,
|
||||
}
|
||||
jobRepo := &mockJobRepo{
|
||||
Jobs: map[string]*domain.Job{job.ID: job},
|
||||
StatusUpdates: make(map[string]domain.JobStatus),
|
||||
}
|
||||
|
||||
svc, auditRepo := newTestJobServiceWithAudit(jobRepo)
|
||||
|
||||
if err := svc.ReapTimedOutJobs(ctx, 24*time.Hour, 168*time.Hour); err != nil {
|
||||
t.Fatalf("ReapTimedOutJobs failed: %v", err)
|
||||
}
|
||||
|
||||
// Check the job was updated by retrieving from the mock's Jobs map
|
||||
updatedJob := jobRepo.Jobs[job.ID]
|
||||
if updatedJob.Status != domain.JobStatusFailed {
|
||||
t.Fatalf("expected job %s status Failed after timeout, got %s", job.ID, updatedJob.Status)
|
||||
}
|
||||
|
||||
// LastError should be set
|
||||
if job.LastError == nil || !strings.Contains(*job.LastError, "timed out in AwaitingCSR after 24h") {
|
||||
t.Errorf("expected LastError containing 'timed out in AwaitingCSR after 24h', got %v", job.LastError)
|
||||
}
|
||||
|
||||
// Audit event should be recorded
|
||||
if len(auditRepo.Events) != 1 {
|
||||
t.Fatalf("expected 1 audit event, got %d", len(auditRepo.Events))
|
||||
}
|
||||
|
||||
ev := auditRepo.Events[0]
|
||||
if ev.Action != "job_timeout" {
|
||||
t.Errorf("expected action job_timeout, got %s", ev.Action)
|
||||
}
|
||||
if ev.Actor != "system" {
|
||||
t.Errorf("expected actor system, got %s", ev.Actor)
|
||||
}
|
||||
if ev.ActorType != domain.ActorTypeSystem {
|
||||
t.Errorf("expected actor type System, got %s", ev.ActorType)
|
||||
}
|
||||
if ev.ResourceType != "job" {
|
||||
t.Errorf("expected resource type job, got %s", ev.ResourceType)
|
||||
}
|
||||
if ev.ResourceID != job.ID {
|
||||
t.Errorf("expected resource ID %s, got %s", job.ID, ev.ResourceID)
|
||||
}
|
||||
|
||||
// Verify audit details
|
||||
var details map[string]interface{}
|
||||
if err := json.Unmarshal(ev.Details, &details); err != nil {
|
||||
t.Fatalf("failed to decode audit event details: %v", err)
|
||||
}
|
||||
if got, want := details["old_status"], string(domain.JobStatusAwaitingCSR); got != want {
|
||||
t.Errorf("expected details.old_status=%s, got %v", want, got)
|
||||
}
|
||||
if got, want := details["new_status"], string(domain.JobStatusFailed); got != want {
|
||||
t.Errorf("expected details.new_status=%s, got %v", want, got)
|
||||
}
|
||||
if got, want := details["timeout_reason"], "csr_timeout"; got != want {
|
||||
t.Errorf("expected details.timeout_reason=%s, got %v", want, got)
|
||||
}
|
||||
ageHours, ok := details["age_hours"].(float64)
|
||||
if !ok {
|
||||
t.Errorf("expected details.age_hours to be float64, got %T", details["age_hours"])
|
||||
} else if ageHours < 35 {
|
||||
t.Errorf("expected age_hours > 35, got %f", ageHours)
|
||||
}
|
||||
}
|
||||
|
||||
func TestJobService_ReapTimedOutJobs_AwaitingApprovalTransitionsAndAudits(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
now := time.Now()
|
||||
job := &domain.Job{
|
||||
ID: "job-reap-approval-1",
|
||||
Type: domain.JobTypeRenewal,
|
||||
CertificateID: "cert-002",
|
||||
Status: domain.JobStatusAwaitingApproval,
|
||||
CreatedAt: now.Add(-200 * time.Hour), // 200 hours old
|
||||
ScheduledAt: now,
|
||||
}
|
||||
jobRepo := &mockJobRepo{
|
||||
Jobs: map[string]*domain.Job{job.ID: job},
|
||||
StatusUpdates: make(map[string]domain.JobStatus),
|
||||
}
|
||||
|
||||
svc, auditRepo := newTestJobServiceWithAudit(jobRepo)
|
||||
|
||||
if err := svc.ReapTimedOutJobs(ctx, 24*time.Hour, 168*time.Hour); err != nil {
|
||||
t.Fatalf("ReapTimedOutJobs failed: %v", err)
|
||||
}
|
||||
|
||||
// Check the job was updated
|
||||
updatedJob := jobRepo.Jobs[job.ID]
|
||||
if updatedJob.Status != domain.JobStatusFailed {
|
||||
t.Fatalf("expected job %s status Failed after timeout, got %s", job.ID, updatedJob.Status)
|
||||
}
|
||||
|
||||
// LastError should be set
|
||||
if updatedJob.LastError == nil || !strings.Contains(*updatedJob.LastError, "timed out in AwaitingApproval after 168h") {
|
||||
t.Errorf("expected LastError containing 'timed out in AwaitingApproval after 168h', got %v", updatedJob.LastError)
|
||||
}
|
||||
|
||||
// Audit event details
|
||||
if len(auditRepo.Events) != 1 {
|
||||
t.Fatalf("expected 1 audit event, got %d", len(auditRepo.Events))
|
||||
}
|
||||
|
||||
ev := auditRepo.Events[0]
|
||||
var details map[string]interface{}
|
||||
if err := json.Unmarshal(ev.Details, &details); err != nil {
|
||||
t.Fatalf("failed to decode audit event details: %v", err)
|
||||
}
|
||||
if got, want := details["timeout_reason"], "approval_timeout"; got != want {
|
||||
t.Errorf("expected details.timeout_reason=%s, got %v", want, got)
|
||||
}
|
||||
ageHours, ok := details["age_hours"].(float64)
|
||||
if !ok {
|
||||
t.Errorf("expected details.age_hours to be float64, got %T", details["age_hours"])
|
||||
} else if ageHours < 199 {
|
||||
t.Errorf("expected age_hours > 199, got %f", ageHours)
|
||||
}
|
||||
}
|
||||
|
||||
func TestJobService_ReapTimedOutJobs_SkipsJobsWithinTTL(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
now := time.Now()
|
||||
job := &domain.Job{
|
||||
ID: "job-within-ttl",
|
||||
Type: domain.JobTypeRenewal,
|
||||
CertificateID: "cert-003",
|
||||
Status: domain.JobStatusAwaitingCSR,
|
||||
CreatedAt: now.Add(-1 * time.Hour), // Only 1 hour old (within 24h TTL)
|
||||
ScheduledAt: now,
|
||||
}
|
||||
jobRepo := &mockJobRepo{
|
||||
Jobs: map[string]*domain.Job{job.ID: job},
|
||||
StatusUpdates: make(map[string]domain.JobStatus),
|
||||
}
|
||||
|
||||
svc, auditRepo := newTestJobServiceWithAudit(jobRepo)
|
||||
|
||||
if err := svc.ReapTimedOutJobs(ctx, 24*time.Hour, 168*time.Hour); err != nil {
|
||||
t.Fatalf("ReapTimedOutJobs failed: %v", err)
|
||||
}
|
||||
|
||||
// Job should NOT transition (still AwaitingCSR)
|
||||
if job.Status != domain.JobStatusAwaitingCSR {
|
||||
t.Fatalf("expected job status to remain AwaitingCSR, got %s", job.Status)
|
||||
}
|
||||
|
||||
// No audit events should be recorded
|
||||
if len(auditRepo.Events) != 0 {
|
||||
t.Fatalf("expected 0 audit events, got %d", len(auditRepo.Events))
|
||||
}
|
||||
}
|
||||
|
||||
func TestJobService_ReapTimedOutJobs_HandlesBothStatusesInOneSweep(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
now := time.Now()
|
||||
csr := &domain.Job{
|
||||
ID: "job-sweep-csr",
|
||||
Type: domain.JobTypeRenewal,
|
||||
CertificateID: "cert-csr",
|
||||
Status: domain.JobStatusAwaitingCSR,
|
||||
CreatedAt: now.Add(-36 * time.Hour),
|
||||
ScheduledAt: now,
|
||||
}
|
||||
approval := &domain.Job{
|
||||
ID: "job-sweep-approval",
|
||||
Type: domain.JobTypeRenewal,
|
||||
CertificateID: "cert-approval",
|
||||
Status: domain.JobStatusAwaitingApproval,
|
||||
CreatedAt: now.Add(-200 * time.Hour),
|
||||
ScheduledAt: now,
|
||||
}
|
||||
jobRepo := &mockJobRepo{
|
||||
Jobs: map[string]*domain.Job{
|
||||
csr.ID: csr,
|
||||
approval.ID: approval,
|
||||
},
|
||||
StatusUpdates: make(map[string]domain.JobStatus),
|
||||
}
|
||||
|
||||
svc, auditRepo := newTestJobServiceWithAudit(jobRepo)
|
||||
|
||||
if err := svc.ReapTimedOutJobs(ctx, 24*time.Hour, 168*time.Hour); err != nil {
|
||||
t.Fatalf("ReapTimedOutJobs failed: %v", err)
|
||||
}
|
||||
|
||||
// Both jobs should transition to Failed
|
||||
csrUpdated := jobRepo.Jobs[csr.ID]
|
||||
if csrUpdated.Status != domain.JobStatusFailed {
|
||||
t.Fatalf("expected CSR job status Failed, got %s", csrUpdated.Status)
|
||||
}
|
||||
approvalUpdated := jobRepo.Jobs[approval.ID]
|
||||
if approvalUpdated.Status != domain.JobStatusFailed {
|
||||
t.Fatalf("expected approval job status Failed, got %s", approvalUpdated.Status)
|
||||
}
|
||||
|
||||
// Two audit events should be recorded
|
||||
if len(auditRepo.Events) != 2 {
|
||||
t.Fatalf("expected 2 audit events, got %d", len(auditRepo.Events))
|
||||
}
|
||||
|
||||
// Verify each event has the correct timeout_reason
|
||||
for _, ev := range auditRepo.Events {
|
||||
var details map[string]interface{}
|
||||
if err := json.Unmarshal(ev.Details, &details); err != nil {
|
||||
t.Fatalf("failed to decode details: %v", err)
|
||||
}
|
||||
if ev.ResourceID == csr.ID && details["timeout_reason"] != "csr_timeout" {
|
||||
t.Errorf("CSR job: expected timeout_reason=csr_timeout, got %v", details["timeout_reason"])
|
||||
}
|
||||
if ev.ResourceID == approval.ID && details["timeout_reason"] != "approval_timeout" {
|
||||
t.Errorf("approval job: expected timeout_reason=approval_timeout, got %v", details["timeout_reason"])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestJobService_ReapTimedOutJobs_NoAuditServiceOK(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
now := time.Now()
|
||||
job := &domain.Job{
|
||||
ID: "job-no-audit",
|
||||
Type: domain.JobTypeRenewal,
|
||||
CertificateID: "cert-004",
|
||||
Status: domain.JobStatusAwaitingCSR,
|
||||
CreatedAt: now.Add(-36 * time.Hour),
|
||||
ScheduledAt: now,
|
||||
}
|
||||
jobRepo := &mockJobRepo{
|
||||
Jobs: map[string]*domain.Job{job.ID: job},
|
||||
StatusUpdates: make(map[string]domain.JobStatus),
|
||||
}
|
||||
|
||||
// Create service WITHOUT calling SetAuditService
|
||||
svc := newTestJobService(jobRepo)
|
||||
|
||||
// Should not panic and should still transition the job
|
||||
if err := svc.ReapTimedOutJobs(ctx, 24*time.Hour, 168*time.Hour); err != nil {
|
||||
t.Fatalf("ReapTimedOutJobs failed without audit service: %v", err)
|
||||
}
|
||||
|
||||
// Job should still transition to Failed
|
||||
updatedJob := jobRepo.Jobs[job.ID]
|
||||
if updatedJob.Status != domain.JobStatusFailed {
|
||||
t.Fatalf("expected job status Failed without audit service, got %s", updatedJob.Status)
|
||||
}
|
||||
}
|
||||
|
||||
func TestJobService_ReapTimedOutJobs_ContinuesOnIndividualUpdateFailure(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
now := time.Now()
|
||||
jobA := &domain.Job{
|
||||
ID: "job-fail-update-a",
|
||||
Type: domain.JobTypeRenewal,
|
||||
CertificateID: "cert-a",
|
||||
Status: domain.JobStatusAwaitingCSR,
|
||||
CreatedAt: now.Add(-36 * time.Hour),
|
||||
ScheduledAt: now,
|
||||
}
|
||||
jobB := &domain.Job{
|
||||
ID: "job-fail-update-b",
|
||||
Type: domain.JobTypeRenewal,
|
||||
CertificateID: "cert-b",
|
||||
Status: domain.JobStatusAwaitingCSR,
|
||||
CreatedAt: now.Add(-48 * time.Hour),
|
||||
ScheduledAt: now,
|
||||
}
|
||||
jobRepo := &mockJobRepo{
|
||||
Jobs: map[string]*domain.Job{
|
||||
jobA.ID: jobA,
|
||||
jobB.ID: jobB,
|
||||
},
|
||||
StatusUpdates: make(map[string]domain.JobStatus),
|
||||
UpdateErrorByID: make(map[string]error),
|
||||
UpdateErrorByIDMu: sync.Mutex{},
|
||||
}
|
||||
// Make Update fail only for jobA
|
||||
jobRepo.UpdateErrorByIDMu.Lock()
|
||||
jobRepo.UpdateErrorByID[jobA.ID] = errors.New("db connection lost")
|
||||
jobRepo.UpdateErrorByIDMu.Unlock()
|
||||
|
||||
svc, auditRepo := newTestJobServiceWithAudit(jobRepo)
|
||||
|
||||
// Should not propagate individual Update errors
|
||||
if err := svc.ReapTimedOutJobs(ctx, 24*time.Hour, 168*time.Hour); err != nil {
|
||||
t.Fatalf("ReapTimedOutJobs should not propagate individual errors, got: %v", err)
|
||||
}
|
||||
|
||||
// Both jobs have their status modified in memory (the service modifies before Update),
|
||||
// so both will be Failed. What matters is that jobA's audit failed, so only jobB audited.
|
||||
jobAAfter := jobRepo.Jobs[jobA.ID]
|
||||
jobBAfter := jobRepo.Jobs[jobB.ID]
|
||||
if jobAAfter.Status != domain.JobStatusFailed || jobBAfter.Status != domain.JobStatusFailed {
|
||||
t.Fatalf("expected both jobs status Failed (modified before Update), got A=%s B=%s",
|
||||
jobAAfter.Status, jobBAfter.Status)
|
||||
}
|
||||
|
||||
// Only one audit event (from jobB, since jobA's Update failed and thus no audit for it)
|
||||
if len(auditRepo.Events) != 1 {
|
||||
t.Fatalf("expected 1 audit event (only jobB succeeded), got %d", len(auditRepo.Events))
|
||||
}
|
||||
if auditRepo.Events[0].ResourceID != jobB.ID {
|
||||
t.Errorf("expected audit event for jobB, got event for %s", auditRepo.Events[0].ResourceID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestJobService_ReapTimedOutJobs_RepoErrorPropagates(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
jobRepo := &mockJobRepo{
|
||||
Jobs: make(map[string]*domain.Job),
|
||||
ListTimedOutErr: errors.New("database down"),
|
||||
StatusUpdates: make(map[string]domain.JobStatus),
|
||||
UpdateErrorByIDMu: sync.Mutex{},
|
||||
}
|
||||
|
||||
svc, auditRepo := newTestJobServiceWithAudit(jobRepo)
|
||||
|
||||
err := svc.ReapTimedOutJobs(ctx, 24*time.Hour, 168*time.Hour)
|
||||
if err == nil {
|
||||
t.Fatalf("expected ReapTimedOutJobs to propagate repo error, got nil")
|
||||
}
|
||||
|
||||
if !strings.Contains(err.Error(), "database down") {
|
||||
t.Errorf("expected error to contain 'database down', got: %v", err)
|
||||
}
|
||||
|
||||
// No audit events should be recorded when repo fails
|
||||
if len(auditRepo.Events) != 0 {
|
||||
t.Fatalf("expected 0 audit events after repo error, got %d", len(auditRepo.Events))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -156,17 +156,20 @@ func (m *mockCertRepo) AddCert(cert *domain.ManagedCertificate) {
|
||||
|
||||
// mockJobRepo is a test implementation of JobRepository
|
||||
type mockJobRepo struct {
|
||||
mu sync.Mutex
|
||||
Jobs map[string]*domain.Job
|
||||
StatusUpdates map[string]domain.JobStatus
|
||||
CreateErr error
|
||||
UpdateErr error
|
||||
UpdateStatusErr error
|
||||
GetErr error
|
||||
ListErr error
|
||||
ListByStatusErr error
|
||||
DeleteErr error
|
||||
Updated []*domain.Job
|
||||
mu sync.Mutex
|
||||
Jobs map[string]*domain.Job
|
||||
StatusUpdates map[string]domain.JobStatus
|
||||
CreateErr error
|
||||
UpdateErr error
|
||||
UpdateErrorByID map[string]error
|
||||
UpdateErrorByIDMu sync.Mutex
|
||||
UpdateStatusErr error
|
||||
GetErr error
|
||||
ListErr error
|
||||
ListByStatusErr error
|
||||
DeleteErr error
|
||||
ListTimedOutErr error
|
||||
Updated []*domain.Job
|
||||
}
|
||||
|
||||
func (m *mockJobRepo) List(ctx context.Context) ([]*domain.Job, error) {
|
||||
@@ -211,6 +214,13 @@ func (m *mockJobRepo) Update(ctx context.Context, job *domain.Job) error {
|
||||
if m.UpdateErr != nil {
|
||||
return m.UpdateErr
|
||||
}
|
||||
// Check per-ID error injection
|
||||
m.UpdateErrorByIDMu.Lock()
|
||||
idErr, ok := m.UpdateErrorByID[job.ID]
|
||||
m.UpdateErrorByIDMu.Unlock()
|
||||
if ok && idErr != nil {
|
||||
return idErr
|
||||
}
|
||||
m.Jobs[job.ID] = job
|
||||
m.Updated = append(m.Updated, job)
|
||||
return nil
|
||||
@@ -352,6 +362,30 @@ func (m *mockJobRepo) ClaimPendingByAgentID(ctx context.Context, agentID string)
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// ListTimedOutAwaitingJobs returns jobs stuck in AwaitingCSR/AwaitingApproval past the
|
||||
// respective cutoffs. I-003 coverage-gap closure.
|
||||
func (m *mockJobRepo) ListTimedOutAwaitingJobs(ctx context.Context, csrCutoff, approvalCutoff time.Time) ([]*domain.Job, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if m.ListTimedOutErr != nil {
|
||||
return nil, m.ListTimedOutErr
|
||||
}
|
||||
var jobs []*domain.Job
|
||||
for _, j := range m.Jobs {
|
||||
switch j.Status {
|
||||
case domain.JobStatusAwaitingCSR:
|
||||
if j.CreatedAt.Before(csrCutoff) {
|
||||
jobs = append(jobs, j)
|
||||
}
|
||||
case domain.JobStatusAwaitingApproval:
|
||||
if j.CreatedAt.Before(approvalCutoff) {
|
||||
jobs = append(jobs, j)
|
||||
}
|
||||
}
|
||||
}
|
||||
return jobs, nil
|
||||
}
|
||||
|
||||
func (m *mockJobRepo) AddJob(job *domain.Job) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
@@ -77,6 +77,10 @@ func (m *mockVerificationJobRepo) ClaimPendingByAgentID(ctx context.Context, age
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (m *mockVerificationJobRepo) ListTimedOutAwaitingJobs(ctx context.Context, csrCutoff, approvalCutoff time.Time) ([]*domain.Job, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// newVerificationTestService creates a VerificationService wired with test doubles.
|
||||
func newVerificationTestService(jobs map[string]*domain.Job, jobRepoErr error) (*VerificationService, *mockVerificationJobRepo, *mockAuditRepo) {
|
||||
jobRepo := &mockVerificationJobRepo{jobs: jobs, err: jobRepoErr}
|
||||
|
||||
Reference in New Issue
Block a user