diff --git a/pkg/certmanager/acme.go b/pkg/certmanager/acme.go index 1b9fa6c43..fe2e828fb 100644 --- a/pkg/certmanager/acme.go +++ b/pkg/certmanager/acme.go @@ -182,8 +182,6 @@ func (s *ACMEService) registerAccount(ctx context.Context) error { return nil } -// StartHTTPChallenge creates an ACME order, accepts the HTTP-01 challenge, and -// returns the persisted challenge metadata. It never waits for order completion. func (s *ACMEService) StartHTTPChallenge(ctx context.Context, domain string) (*HTTPChallenge, error) { started := time.Now() @@ -223,15 +221,6 @@ func (s *ACMEService) StartHTTPChallenge(ctx context.Context, domain string) (*H return nil, fmt.Errorf("cannot get challenge response: %w", err) } - challenge1 := &acme.Challenge{ - URI: challenge.URI, - Token: challenge.Token, - } - - if _, err := s.client.Accept(ctx, challenge1); err != nil && !isChallengeAlreadyValid(err) { - return nil, s.handleError(provisionPhaseCreateOrder, started, "cannot accept challenge", err) - } - s.metrics.observeStep(provisionPhaseCreateOrder, provisionResultOK, started) return &HTTPChallenge{ @@ -243,7 +232,23 @@ func (s *ACMEService) StartHTTPChallenge(ctx context.Context, domain string) (*H }, nil } -// PollOrder performs a single GetOrder call and classifies the result. +func (s *ACMEService) AcceptHTTPChallenge(ctx context.Context, challenge *HTTPChallenge) error { + started := time.Now() + + acceptChallenge := &acme.Challenge{ + URI: challenge.URL, + Token: challenge.Token, + } + + if _, err := s.client.Accept(ctx, acceptChallenge); err != nil && !isChallengeAlreadyValid(err) { + return s.handleError(provisionPhaseCreateOrder, started, "cannot accept challenge", err) + } + + s.metrics.observeStep(provisionPhaseCreateOrder, provisionResultOK, started) + + return nil +} + func (s *ACMEService) PollOrder(ctx context.Context, orderURL string) (*OrderPollResult, error) { started := time.Now() @@ -279,7 +284,6 @@ func (s *ACMEService) PollOrder(ctx context.Context, orderURL string) (*OrderPol return result, nil } -// IssueCertificate finalizes a ready order or fetches a valid certificate. func (s *ACMEService) IssueCertificate( ctx context.Context, challenge *HTTPChallenge, diff --git a/pkg/certmanager/acme_errors.go b/pkg/certmanager/acme_errors.go index 69611a5db..c8b80e880 100644 --- a/pkg/certmanager/acme_errors.go +++ b/pkg/certmanager/acme_errors.go @@ -23,6 +23,7 @@ package certmanager import ( "errors" "fmt" + "math" "net/http" "strconv" "time" @@ -72,11 +73,6 @@ func (e *ACMEError) Is(target error) bool { return e != nil && e.rateLimited && target == ErrACMERateLimited } -// RetryAfter returns how long callers should wait before retrying. -// For rate-limited errors it honors the ACME Retry-After value whenever the -// header is present and parseable — including a zero (or past) value, which the -// CA uses to permit an immediate retry. It falls back to defaultCooldown only -// when the header is absent or invalid. Non-rate-limited errors return 0. func (e *ACMEError) RetryAfter() time.Duration { if e == nil || !e.rateLimited { return 0 @@ -131,10 +127,6 @@ func newACMEError(op string, err error) *ACMEError { if _, ok := acme.RateLimit(acmeErr); ok { out.rateLimited = true - // acme.RateLimit collapses "Retry-After: 0", an invalid header, and a - // missing header all to a zero duration, so inspect the header directly - // to tell an explicit zero (immediate retry) apart from an absent one - // (fall back to defaultCooldown in RetryAfter). if retryAfter, ok := parseRetryAfter(acmeErr.Header); ok { out.retryAfter = retryAfter out.retryAfterSet = true @@ -158,7 +150,17 @@ func parseRetryAfter(header http.Header) (time.Duration, bool) { return 0, false } - if seconds, err := strconv.Atoi(value); err == nil { + // The delta-seconds form is an unsigned decimal integer (RFC 9110 §10.2.3). + // Parsing it as unsigned rejects negative or otherwise malformed values so + // they fall back to the caller's default cooldown instead of collapsing to a + // zero/negative duration that would disable the rate-limit cooldown. A value + // larger than time.Duration can hold is clamped to the maximum duration. + if seconds, err := strconv.ParseUint(value, 10, 64); err == nil { + maxSeconds := uint64(math.MaxInt64 / int64(time.Second)) + if seconds > maxSeconds { + return time.Duration(math.MaxInt64), true + } + return time.Duration(seconds) * time.Second, true } diff --git a/pkg/certmanager/acme_test.go b/pkg/certmanager/acme_test.go index be90f69d7..e71857817 100644 --- a/pkg/certmanager/acme_test.go +++ b/pkg/certmanager/acme_test.go @@ -52,7 +52,7 @@ func TestNewMetrics_SharedRegistererDoesNotPanic(t *testing.T) { assert.Same(t, first.provisionSteps, second.provisionSteps) assert.Same(t, first.acmeErrors, second.acmeErrors) assert.Same(t, first.stepDuration, second.stepDuration) - assert.Equal(t, first.acmeCooldown, second.acmeCooldown) + assert.Same(t, first.acmeCooldown, second.acmeCooldown) } func TestNewACMEError_RateLimited(t *testing.T) { diff --git a/pkg/certmanager/metrics.go b/pkg/certmanager/metrics.go index 3f1afb20c..0b78abe56 100644 --- a/pkg/certmanager/metrics.go +++ b/pkg/certmanager/metrics.go @@ -22,6 +22,7 @@ package certmanager import ( "errors" + "strings" "time" "github.com/prometheus/client_golang/prometheus" @@ -127,11 +128,49 @@ func (m *metrics) observeStep(phase provisionPhase, result provisionResult, star } func (m *metrics) recordACMEError(problemType string) { - if problemType == "" { - problemType = "unknown" + m.acmeErrors.WithLabelValues(normalizeProblemType(problemType)).Inc() +} + +const acmeProblemTypePrefix = "urn:ietf:params:acme:error:" + +var knownACMEProblemTypes = map[string]struct{}{ + "accountdoesnotexist": {}, + "alreadyrevoked": {}, + "badcsr": {}, + "badnonce": {}, + "badpublickey": {}, + "badrevocationreason": {}, + "badsignaturealgorithm": {}, + "caa": {}, + "compound": {}, + "connection": {}, + "dns": {}, + "externalaccountrequired": {}, + "incorrectresponse": {}, + "invalidcontact": {}, + "malformed": {}, + "ordernotready": {}, + "ratelimited": {}, + "rejectedidentifier": {}, + "serverinternal": {}, + "tls": {}, + "unauthorized": {}, + "unsupportedcontact": {}, + "unsupportedidentifier": {}, + "useractionrequired": {}, +} + +func normalizeProblemType(problemType string) string { + suffix, ok := strings.CutPrefix(strings.ToLower(problemType), acmeProblemTypePrefix) + if !ok { + return "unknown" } - m.acmeErrors.WithLabelValues(problemType).Inc() + if _, ok := knownACMEProblemTypes[suffix]; !ok { + return "unknown" + } + + return suffix } func (m *metrics) setCooldown(active bool) { diff --git a/pkg/certmanager/provision_worker.go b/pkg/certmanager/provision_worker.go index 01ba6dae5..cb2aefe92 100644 --- a/pkg/certmanager/provision_worker.go +++ b/pkg/certmanager/provision_worker.go @@ -43,10 +43,14 @@ const ( maxProvisioningRetries = 3 dnsExchangeTimeout = 10 * time.Second processTickTimeout = 90 * time.Second - // persistFailureTimeout bounds the retry-outcome write-back. It runs on a - // context detached from the process tick deadline so a timed-out attempt can - // still record its failure. - persistFailureTimeout = 15 * time.Second + persistFailureTimeout = 15 * time.Second + + // provisioningPollLease is how long a claimed PROVISIONING row with an open + // order stays ineligible for re-claim after each attempt. It must exceed + // processTickTimeout: the claim's row lock is released before Process runs, + // so a lease shorter than the maximum processing window would let another + // worker claim and process the same row concurrently. + provisioningPollLease = processTickTimeout + 30*time.Second tracerName = "go.probo.inc/probo/pkg/certmanager" ) @@ -120,7 +124,7 @@ func (h *provisionHandler) Claim(ctx context.Context) (coredata.Certificate, err if err := h.pg.WithTx( ctx, func(ctx context.Context, tx pg.Tx) error { - if err := certificate.LoadNextForProvisioningForUpdateSkipLocked(ctx, tx); err != nil { + if err := certificate.LoadNextForProvisioningForUpdateSkipLocked(ctx, tx, provisioningPollLease); err != nil { return err } @@ -209,11 +213,57 @@ func (h *provisionHandler) processBeginChallenge( return h.persistFailure(ctx, certificate.ID, err) } - return h.pg.WithTx( + persisted, err := h.persistChallenge(ctx, certificate, challenge) + if err != nil { + h.recordSpanError(span, err, classifyProvisioningError(err)) + return err + } + + if !persisted { + return nil + } + + // The key authorization is now committed and served by the challenge + // handler, so it is safe to ask the CA to begin validation. Accepting + // earlier risks the CA hitting the token before this instance can serve it, + // yielding a 404 and an invalid order. + if err := h.acmeService.AcceptHTTPChallenge(ctx, challenge); err != nil { + errorCode := classifyProvisioningError(err) + h.logACMEOutcome(ctx, certificate, provisionPhaseCreateOrder, err, errorCode) + h.recordSpanError(span, err, errorCode) + + return h.persistFailure(ctx, certificate.ID, err) + } + + h.logger.InfoCtx( + ctx, + "HTTP challenge accepted, waiting for order validation", + log.String("hostname", certificate.Hostname), + log.String("certificate_id", certificate.ID.String()), + ) + + return nil +} + +// persistChallenge stores the HTTP-01 challenge metadata and flips the row to +// PROVISIONING. It takes a blocking write-back lock so the metadata is persisted +// once any competing transaction commits; a genuinely deleted row is the only +// no-op. It reports whether the row was persisted (false when the row is gone or +// has moved on to another status). +func (h *provisionHandler) persistChallenge( + ctx context.Context, + certificate coredata.Certificate, + challenge *HTTPChallenge, +) (bool, error) { + persisted := false + + err := h.pg.WithTx( ctx, func(ctx context.Context, tx pg.Tx) error { + persisted = false + row := &coredata.Certificate{} - if err := row.LoadByIDForUpdateSkipLocked(ctx, tx, coredata.NewNoScope(), certificate.ID); err != nil { + if err := row.LoadByIDForUpdate(ctx, tx, coredata.NewNoScope(), certificate.ID); err != nil { if errors.Is(err, coredata.ErrResourceNotFound) { return nil } @@ -237,16 +287,16 @@ func (h *provisionHandler) processBeginChallenge( return fmt.Errorf("cannot update certificate with challenge: %w", err) } - h.logger.InfoCtx( - ctx, - "HTTP challenge accepted, waiting for order validation", - log.String("hostname", row.Hostname), - log.String("certificate_id", row.ID.String()), - ) + persisted = true return nil }, ) + if err != nil { + return false, err + } + + return persisted, nil } func (h *provisionHandler) processPollOrder( @@ -301,13 +351,68 @@ func (h *provisionHandler) processPollOrder( h.recordSpanError(span, err, errorCode) return h.persistFailure(ctx, certificate.ID, err) - case OrderPollStatusReady, OrderPollStatusValid: + case OrderPollStatusReady: return h.issueCertificate(ctx, certificate, challenge, poll) + case OrderPollStatusValid: + // A VALID order observed while polling means a previous attempt already + // finalized it at the CA but the certificate/private-key write never + // landed (crash or a failed transaction). The private key generated for + // that finalize is gone, so fetching the issued certificate now would + // pair it with a freshly generated key and break TLS loading. Abandon + // the unrecoverable order and start a new one. + return h.abandonRecoveredValidOrder(ctx, span, certificate) default: return nil } } +// abandonRecoveredValidOrder clears the ACME order state and returns the row to +// PENDING so the next tick mints a fresh order (and a matching private key). +func (h *provisionHandler) abandonRecoveredValidOrder( + ctx context.Context, + span trace.Span, + certificate coredata.Certificate, +) error { + h.logger.WarnCtx( + ctx, + "abandoning recovered valid ACME order without a matching private key, restarting provisioning", + log.String("hostname", certificate.Hostname), + log.String("certificate_id", certificate.ID.String()), + ) + span.AddEvent("abandoned recovered valid order without matching private key") + + return h.pg.WithTx( + ctx, + func(ctx context.Context, tx pg.Tx) error { + row := &coredata.Certificate{} + if err := row.LoadByIDForUpdate(ctx, tx, coredata.NewNoScope(), certificate.ID); err != nil { + if errors.Is(err, coredata.ErrResourceNotFound) { + return nil + } + + return fmt.Errorf("cannot load certificate %q: %w", certificate.ID, err) + } + + if row.Status != coredata.CertificateStatusProvisioning { + return nil + } + + row.HTTPChallengeToken = nil + row.HTTPChallengeKeyAuth = nil + row.HTTPChallengeURL = nil + row.HTTPOrderURL = nil + row.ProvisioningError = nil + row.Status = coredata.CertificateStatusPending + + if err := row.Update(ctx, tx, coredata.NewNoScope()); err != nil { + return fmt.Errorf("cannot reset certificate after abandoning valid order: %w", err) + } + + return nil + }, + ) +} + func (h *provisionHandler) issueCertificate( ctx context.Context, certificate coredata.Certificate, @@ -408,12 +513,6 @@ func (h *provisionHandler) persistFailure( ) error { errorCode := classifyProvisioningError(provisionErr) - // Process runs each tick under processTickTimeout. When that deadline fires - // mid-attempt, the same expired context reaches here, and the write-back - // silently fails — so the retry budget never advances and the certificate - // stays retriable forever. Detach from the tick deadline (and cancellation) - // and bound the write with its own timeout so repeated timeouts still make - // progress toward FAILED. ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), persistFailureTimeout) defer cancel() @@ -527,7 +626,7 @@ func (h *provisionHandler) RecoverStale(ctx context.Context) error { ctx, func(ctx context.Context, tx pg.Tx) error { var certificates coredata.Certificates - if err := certificates.ListStaleProvisioning(ctx, tx, coredata.NewNoScope()); err != nil { + if err := certificates.ListStaleProvisioning(ctx, tx, coredata.NewNoScope(), ProvisioningErrorACMERateLimited); err != nil { return fmt.Errorf("cannot load stale provisioning certificates: %w", err) } diff --git a/pkg/coredata/certificate.go b/pkg/coredata/certificate.go index 57973e711..933e3a92b 100644 --- a/pkg/coredata/certificate.go +++ b/pkg/coredata/certificate.go @@ -808,7 +808,14 @@ func (certificates *Certificates) ListStaleProvisioning( ctx context.Context, conn pg.Querier, scope Scoper, + rateLimitedErrorCode string, ) error { + // Rate-limited rows are intentionally left in PROVISIONING with their order + // URL preserved so the next attempt resumes the same order once the ACME + // cooldown (up to an hour) elapses. Resetting them at the 10-minute stale + // threshold would discard that resumable order and mint a new one straight + // into the same rate limit, so exclude them from that branch. They are still + // recoverable via the 24-hour safety net if they get truly stuck. q := ` SELECT id, @@ -832,7 +839,11 @@ FROM WHERE %s AND ( - (status IN (@provisioning_status, @renewing_status) AND updated_at < CURRENT_TIMESTAMP - INTERVAL '10 minutes') + ( + status IN (@provisioning_status, @renewing_status) + AND updated_at < CURRENT_TIMESTAMP - INTERVAL '10 minutes' + AND (provisioning_error IS NULL OR provisioning_error != @rate_limited_error_code) + ) OR (ssl_retry_count > 0 AND ssl_last_attempt_at < CURRENT_TIMESTAMP - INTERVAL '24 hours') ) @@ -843,10 +854,11 @@ WHERE q = fmt.Sprintf(q, scope.SQLFragment()) args := pgx.NamedArgs{ - "provisioning_status": string(CertificateStatusProvisioning), - "renewing_status": string(CertificateStatusRenewing), - "failed_status": string(CertificateStatusFailed), - "active_status": string(CertificateStatusActive), + "provisioning_status": string(CertificateStatusProvisioning), + "renewing_status": string(CertificateStatusRenewing), + "failed_status": string(CertificateStatusFailed), + "active_status": string(CertificateStatusActive), + "rate_limited_error_code": rateLimitedErrorCode, } maps.Copy(args, scope.SQLArguments()) @@ -868,12 +880,17 @@ WHERE func (c *Certificate) LoadNextForProvisioningForUpdateSkipLocked( ctx context.Context, tx pg.Tx, + pollLease time.Duration, ) error { - // PROVISIONING rows with an open order poll every ~30s. Pending/Renewing - // rows (and provisioning rows without an order) use exponential backoff - // from ssl_last_attempt_at: 15m * 2^min(retry,5). Ordinary failures only - // reach retry counts 0–2 before FAILED; higher exponents are unused by the - // current failure budget but keep the SQL ceiling defensive. + // PROVISIONING rows with an open order become eligible again only after + // pollLease elapses since the last attempt. The caller sizes pollLease to + // exceed the maximum Process window: the claim's FOR UPDATE lock is released + // before Process runs, so a shorter interval would let another worker claim + // the same row while its poll/issue is still in flight. Pending/Renewing + // rows (and provisioning rows without an order) use exponential backoff from + // ssl_last_attempt_at: 15m * 2^min(retry,5). Ordinary failures only reach + // retry counts 0–2 before FAILED; higher exponents are unused by the current + // failure budget but keep the SQL ceiling defensive. q := ` SELECT id, @@ -901,7 +918,7 @@ WHERE OR ( status = @provisioning_status AND http_order_url IS NOT NULL - AND ssl_last_attempt_at < CURRENT_TIMESTAMP - INTERVAL '30 seconds' + AND ssl_last_attempt_at < CURRENT_TIMESTAMP - make_interval(secs => @poll_lease_seconds) ) OR ( NOT ( @@ -929,6 +946,7 @@ FOR UPDATE SKIP LOCKED string(CertificateStatusRenewing), }, "provisioning_status": string(CertificateStatusProvisioning), + "poll_lease_seconds": pollLease.Seconds(), }, ) if err != nil {