From c17ea577e773713276a929f61ac1bdf579f7100d Mon Sep 17 00:00:00 2001 From: Shankar Date: Sun, 19 Apr 2026 01:04:56 +0000 Subject: [PATCH] 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 --- cmd/server/main.go | 11 + internal/config/config.go | 38 +++ internal/config/config_test.go | 124 +++++++++ internal/integration/lifecycle_test.go | 19 ++ internal/repository/interfaces.go | 5 + internal/repository/postgres/job.go | 36 +++ internal/scheduler/scheduler.go | 93 ++++++- internal/scheduler/scheduler_test.go | 217 +++++++++++++++ internal/service/job.go | 76 ++++++ internal/service/job_test.go | 355 +++++++++++++++++++++++++ internal/service/testutil_test.go | 56 +++- internal/service/verification_test.go | 4 + 12 files changed, 1020 insertions(+), 14 deletions(-) diff --git a/cmd/server/main.go b/cmd/server/main.go index 276cac9..022d256 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -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) diff --git a/internal/config/config.go b/internal/config/config.go index f9f4506..93c8a07 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -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 } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index e8bce5c..b7b8a2c 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -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) + } + }) + } +} diff --git a/internal/integration/lifecycle_test.go b/internal/integration/lifecycle_test.go index a574cf3..f343ebb 100644 --- a/internal/integration/lifecycle_test.go +++ b/internal/integration/lifecycle_test.go @@ -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 } diff --git a/internal/repository/interfaces.go b/internal/repository/interfaces.go index 81e0bbe..b201e04 100644 --- a/internal/repository/interfaces.go +++ b/internal/repository/interfaces.go @@ -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. diff --git a/internal/repository/postgres/job.go b/internal/repository/postgres/job.go index 4e521d7..d535713 100644 --- a/internal/repository/postgres/job.go +++ b/internal/repository/postgres/job.go @@ -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 diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 2c4e54c..0c8a459 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -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. diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go index 1e4692e..daf08cd 100644 --- a/internal/scheduler/scheduler_test.go +++ b/internal/scheduler/scheduler_test.go @@ -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") +} diff --git a/internal/service/job.go b/internal/service/job.go index e034469..e57d84e 100644 --- a/internal/service/job.go +++ b/internal/service/job.go @@ -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) diff --git a/internal/service/job_test.go b/internal/service/job_test.go index b0c1197..a67e1ed 100644 --- a/internal/service/job_test.go +++ b/internal/service/job_test.go @@ -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)) + } +} diff --git a/internal/service/testutil_test.go b/internal/service/testutil_test.go index 51633d4..b530dfb 100644 --- a/internal/service/testutil_test.go +++ b/internal/service/testutil_test.go @@ -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() diff --git a/internal/service/verification_test.go b/internal/service/verification_test.go index b5785c4..58f1f9d 100644 --- a/internal/service/verification_test.go +++ b/internal/service/verification_test.go @@ -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}