diff --git a/pkg/certmanager/provision_worker.go b/pkg/certmanager/provision_worker.go index bc86e5f8e..d00a79412 100644 --- a/pkg/certmanager/provision_worker.go +++ b/pkg/certmanager/provision_worker.go @@ -32,6 +32,7 @@ import ( "go.probo.inc/probo/pkg/coredata" "go.probo.inc/probo/pkg/crypto/cipher" "go.probo.inc/probo/pkg/gid" + "golang.org/x/crypto/acme" ) const ( @@ -56,16 +57,26 @@ const ( ) type ( - provisionHandler struct { - pg *pg.Client - acmeService *ACMEService - encryptionKey cipher.EncryptionKey + provisionCore struct { + pg *pg.Client + acmeService *ACMEService + logger *log.Logger + tracer trace.Tracer + } + + beginChallengeHandler struct { + provisionCore + cnameTarget string caaIssuerDomain string resolverAddr string managedBaseDomain string - logger *log.Logger - tracer trace.Tracer + } + + pollOrderHandler struct { + provisionCore + + encryptionKey cipher.EncryptionKey } provisioningOutcome struct { @@ -77,14 +88,14 @@ type ( ) var ( - _ worker.Handler[coredata.Certificate] = (*provisionHandler)(nil) - _ worker.StaleRecoverer = (*provisionHandler)(nil) + _ worker.Handler[coredata.Certificate] = (*beginChallengeHandler)(nil) + _ worker.Handler[coredata.Certificate] = (*pollOrderHandler)(nil) + _ worker.StaleRecoverer = (*pollOrderHandler)(nil) ) -func NewProvisionWorker( +func NewBeginChallengeWorker( pgClient *pg.Client, acmeService *ACMEService, - encryptionKey cipher.EncryptionKey, cnameTarget string, caaIssuerDomain string, resolverAddr string, @@ -92,16 +103,17 @@ func NewProvisionWorker( logger *log.Logger, opts ...worker.Option, ) *worker.Worker[coredata.Certificate] { - h := &provisionHandler{ - pg: pgClient, - acmeService: acmeService, - encryptionKey: encryptionKey, + h := &beginChallengeHandler{ + provisionCore: provisionCore{ + pg: pgClient, + acmeService: acmeService, + logger: logger, + tracer: otel.Tracer(tracerName), + }, cnameTarget: cnameTarget, caaIssuerDomain: caaIssuerDomain, resolverAddr: resolverAddr, managedBaseDomain: managedBaseDomain, - logger: logger, - tracer: otel.Tracer(tracerName), } opts = append(opts, worker.WithMaxConcurrency(1)) @@ -114,7 +126,34 @@ func NewProvisionWorker( ) } -func (h *provisionHandler) Claim(ctx context.Context) (coredata.Certificate, error) { +func NewPollOrderWorker( + pgClient *pg.Client, + acmeService *ACMEService, + encryptionKey cipher.EncryptionKey, + logger *log.Logger, + opts ...worker.Option, +) *worker.Worker[coredata.Certificate] { + h := &pollOrderHandler{ + provisionCore: provisionCore{ + pg: pgClient, + acmeService: acmeService, + logger: logger, + tracer: otel.Tracer(tracerName), + }, + encryptionKey: encryptionKey, + } + + opts = append(opts, worker.WithMaxConcurrency(1)) + + return worker.New( + "certificate-poll-worker", + h, + logger, + opts..., + ) +} + +func (h *beginChallengeHandler) Claim(ctx context.Context) (coredata.Certificate, error) { if h.acmeService.InCooldown() { return coredata.Certificate{}, worker.ErrNoTask } @@ -124,7 +163,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, provisioningPollLease); err != nil { + if err := certificate.LoadNextForBeginChallengeForUpdateSkipLocked(ctx, tx); err != nil { return err } @@ -148,24 +187,10 @@ func (h *provisionHandler) Claim(ctx context.Context) (coredata.Certificate, err return certificate, nil } -func (h *provisionHandler) Process(ctx context.Context, certificate coredata.Certificate) error { +func (h *beginChallengeHandler) Process(ctx context.Context, certificate coredata.Certificate) error { ctx, cancel := context.WithTimeout(ctx, processTickTimeout) defer cancel() - switch certificate.Status { - case coredata.CertificateStatusPending, coredata.CertificateStatusRenewing: - return h.processBeginChallenge(ctx, certificate) - case coredata.CertificateStatusProvisioning: - return h.processPollOrder(ctx, certificate) - default: - return nil - } -} - -func (h *provisionHandler) processBeginChallenge( - ctx context.Context, - certificate coredata.Certificate, -) error { ctx, span := h.tracer.Start(ctx, "certmanager.create_order") defer span.End() @@ -250,7 +275,7 @@ func (h *provisionHandler) processBeginChallenge( // 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( +func (h *beginChallengeHandler) persistChallenge( ctx context.Context, certificate coredata.Certificate, challenge *HTTPChallenge, @@ -299,7 +324,201 @@ func (h *provisionHandler) persistChallenge( return persisted, nil } -func (h *provisionHandler) processPollOrder( +func (h *beginChallengeHandler) loadSkipDNSChecks(ctx context.Context, hostname string) (bool, error) { + var skip bool + + err := h.pg.WithConn( + ctx, + func(ctx context.Context, conn pg.Querier) error { + var err error + + skip, err = h.skipsDNSChecks(ctx, conn, hostname) + + return err + }, + ) + if err != nil { + return false, err + } + + return skip, nil +} + +func (h *beginChallengeHandler) checkDNSConfiguration(ctx context.Context, hostname string) error { + customerFQDN := hostname + if !strings.HasSuffix(customerFQDN, ".") { + customerFQDN = customerFQDN + "." + } + + expectedFQDN := h.cnameTarget + if !strings.HasSuffix(expectedFQDN, ".") { + expectedFQDN = expectedFQDN + "." + } + + msg := &dns.Msg{MsgHeader: dns.MsgHeader{ID: dns.ID(), RecursionDesired: true}} + msg.Question = []dns.RR{&dns.CNAME{Hdr: dns.Header{Name: customerFQDN, Class: dns.ClassINET}}} + + dnsCtx, cancel := context.WithTimeout(ctx, dnsExchangeTimeout) + defer cancel() + + client := dns.NewClient() + + resp, _, err := client.Exchange(dnsCtx, msg, "udp", h.resolverAddr) + if err != nil { + return fmt.Errorf("cannot exchange dns message: %w", err) + } + + if len(resp.Answer) == 0 { + return fmt.Errorf("no cname records found for domain %q", hostname) + } + + if len(resp.Answer) > 1 { + return fmt.Errorf("multiple cname records found for domain %q", hostname) + } + + resolvedRecord, ok := resp.Answer[0].(*dns.CNAME) + if !ok { + return fmt.Errorf("first answer is not a cname record for domain %q", hostname) + } + + if !strings.EqualFold(expectedFQDN, resolvedRecord.Target) { + return fmt.Errorf( + "cname target mismatch: domain %q resolves to %q, expected %q", + hostname, + resolvedRecord.Target, + expectedFQDN, + ) + } + + return nil +} + +func (h *beginChallengeHandler) checkCAARecords(ctx context.Context, hostname string) error { + fqdn := hostname + if !strings.HasSuffix(fqdn, ".") { + fqdn = fqdn + "." + } + + msg := &dns.Msg{MsgHeader: dns.MsgHeader{ID: dns.ID(), RecursionDesired: true}} + msg.Question = []dns.RR{&dns.CAA{Hdr: dns.Header{Name: fqdn, Class: dns.ClassINET}}} + + dnsCtx, cancel := context.WithTimeout(ctx, dnsExchangeTimeout) + defer cancel() + + client := dns.NewClient() + + resp, _, err := client.Exchange( + dnsCtx, + msg, + "udp", + h.resolverAddr, + ) + if err != nil { + return fmt.Errorf("cannot exchange dns message for caa records: %w", err) + } + + var caaRecords []*dns.CAA + + for _, rr := range resp.Answer { + if caa, ok := rr.(*dns.CAA); ok { + caaRecords = append(caaRecords, caa) + } + } + + if len(caaRecords) == 0 { + return nil + } + + for _, caa := range caaRecords { + if caa.Tag == "issue" { + issuer, _, _ := strings.Cut(caa.Value, ";") + if strings.EqualFold(strings.TrimSpace(issuer), h.caaIssuerDomain) { + return nil + } + } + } + + return fmt.Errorf( + "%w: domain %q by %q", + ErrCAANotPermitted, + hostname, + h.caaIssuerDomain, + ) +} + +func (h *beginChallengeHandler) skipsDNSChecks( + ctx context.Context, + conn pg.Querier, + hostname string, +) (bool, error) { + if h.managedBaseDomain == "" { + return false, nil + } + + suffix := "." + h.managedBaseDomain + if hostname != h.managedBaseDomain && !strings.HasSuffix(hostname, suffix) { + return false, nil + } + + domain := &coredata.CustomDomain{} + + err := domain.LoadByDomain(ctx, conn, coredata.NewNoScope(), hostname) + if errors.Is(err, coredata.ErrResourceNotFound) { + return true, nil + } + + if err != nil { + return false, fmt.Errorf("cannot load custom domain: %w", err) + } + + return domain.Managed, nil +} + +func (h *pollOrderHandler) Claim(ctx context.Context) (coredata.Certificate, error) { + // Deliberately NOT gated by acmeService.InCooldown(). A cooldown is entered + // when minting a NEW order hits the CA's rate limit (see + // beginChallengeHandler.Claim); advancing an order already in flight only + // polls/finalizes that specific order and does not mint new ones. Gating + // this claim too would stall every other tenant's in-flight provisioning + // for up to the cooldown duration just because one unrelated hostname + // tripped a limit. + var certificate coredata.Certificate + + if err := h.pg.WithTx( + ctx, + func(ctx context.Context, tx pg.Tx) error { + if err := certificate.LoadNextForPollOrderForUpdateSkipLocked(ctx, tx, provisioningPollLease); err != nil { + return err + } + + now := time.Now() + certificate.SSLLastAttemptAt = &now + + if err := certificate.Update(ctx, tx, coredata.NewNoScope()); err != nil { + return fmt.Errorf("cannot stamp certificate claim: %w", err) + } + + return nil + }, + ); err != nil { + if errors.Is(err, coredata.ErrResourceNotFound) { + return coredata.Certificate{}, worker.ErrNoTask + } + + return coredata.Certificate{}, err + } + + return certificate, nil +} + +func (h *pollOrderHandler) Process(ctx context.Context, certificate coredata.Certificate) error { + ctx, cancel := context.WithTimeout(ctx, processTickTimeout) + defer cancel() + + return h.processPollOrder(ctx, certificate) +} + +func (h *pollOrderHandler) processPollOrder( ctx context.Context, certificate coredata.Certificate, ) error { @@ -337,16 +556,21 @@ func (h *provisionHandler) processPollOrder( case OrderPollStatusNotReady: // Re-accept best-effort: if a prior tick committed the challenge but its // Accept never reached the CA, the order would otherwise stay pending - // forever since this path only polls. Re-accepting a pending challenge - // starts validation; one already processing is a no-op at the CA. - if err := h.acmeService.AcceptHTTPChallenge(ctx, challenge); err != nil { - h.logger.WarnCtx( - ctx, - "re-accepting HTTP challenge for not-ready order failed, will poll again", - log.String("hostname", certificate.Hostname), - log.String("certificate_id", certificate.ID.String()), - log.Error(err), - ) + // forever since this path only polls. Only do this while the order is + // still PENDING: once it has moved to PROCESSING, the CA has already + // registered the Accept and rejects a second one with + // malformed/"Only pending challenges may be validated" (RFC 8555), so + // re-accepting there is not a no-op and just produces noisy failures. + if poll.Order.Status == acme.StatusPending { + if err := h.acmeService.AcceptHTTPChallenge(ctx, challenge); err != nil { + h.logger.WarnCtx( + ctx, + "re-accepting HTTP challenge for not-ready order failed, will poll again", + log.String("hostname", certificate.Hostname), + log.String("certificate_id", certificate.ID.String()), + log.Error(err), + ) + } } h.logger.InfoCtx( @@ -382,7 +606,7 @@ func (h *provisionHandler) processPollOrder( // 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( +func (h *pollOrderHandler) abandonRecoveredValidOrder( ctx context.Context, span trace.Span, certificate coredata.Certificate, @@ -427,7 +651,7 @@ func (h *provisionHandler) abandonRecoveredValidOrder( ) } -func (h *provisionHandler) issueCertificate( +func (h *pollOrderHandler) issueCertificate( ctx context.Context, certificate coredata.Certificate, challenge *HTTPChallenge, @@ -520,7 +744,7 @@ func (h *provisionHandler) issueCertificate( ) } -func (h *provisionHandler) persistFailure( +func (h *provisionCore) persistFailure( ctx context.Context, certificateID gid.GID, provisionErr error, @@ -591,9 +815,9 @@ func isImmediateRetryRateLimit(err error) bool { // - Normal transient errors: ssl_retry_count increments; at maxProvisioningRetries // the domain becomes FAILED. // - Rate limits: never increment ssl_retry_count and never mark FAILED; the -// in-process ACME cooldown gates Claim. When an order URL is present -// it is preserved so the next tick resumes polling instead of minting a new -// order. +// in-process ACME cooldown gates the begin-challenge worker's Claim. When an +// order URL is present it is preserved so the next tick resumes polling +// instead of minting a new order. // // DNS/CAA misconfig is intentionally non-terminal so customer DNS fixes // auto-recover on the claim backoff schedule. @@ -654,7 +878,7 @@ func decideProvisioningOutcome( } } -func (h *provisionHandler) RecoverStale(ctx context.Context) error { +func (h *pollOrderHandler) RecoverStale(ctx context.Context) error { return h.pg.WithTx( ctx, func(ctx context.Context, tx pg.Tx) error { @@ -685,7 +909,7 @@ func (h *provisionHandler) RecoverStale(ctx context.Context) error { ) } -func (h *provisionHandler) logACMEOutcome( +func (h *provisionCore) logACMEOutcome( ctx context.Context, certificate coredata.Certificate, phase provisionPhase, @@ -740,7 +964,7 @@ func (h *provisionHandler) logACMEOutcome( ) } -func (h *provisionHandler) setCertificateSpanAttributes(span trace.Span, certificate coredata.Certificate) { +func (h *provisionCore) setCertificateSpanAttributes(span trace.Span, certificate coredata.Certificate) { span.SetAttributes( attribute.String("certificate.id", certificate.ID.String()), attribute.String("certificate.hostname", certificate.Hostname), @@ -748,7 +972,7 @@ func (h *provisionHandler) setCertificateSpanAttributes(span trace.Span, certifi ) } -func (h *provisionHandler) recordSpanError(span trace.Span, err error, errorCode string) { +func (h *provisionCore) recordSpanError(span trace.Span, err error, errorCode string) { if err == nil { return } @@ -771,26 +995,6 @@ func (h *provisionHandler) recordSpanError(span trace.Span, err error, errorCode ) } -func (h *provisionHandler) loadSkipDNSChecks(ctx context.Context, hostname string) (bool, error) { - var skip bool - - err := h.pg.WithConn( - ctx, - func(ctx context.Context, conn pg.Querier) error { - var err error - - skip, err = h.skipsDNSChecks(ctx, conn, hostname) - - return err - }, - ) - if err != nil { - return false, err - } - - return skip, nil -} - func stringValue(value *string) string { if value == nil { return "" @@ -799,137 +1003,7 @@ func stringValue(value *string) string { return *value } -func (h *provisionHandler) checkDNSConfiguration(ctx context.Context, hostname string) error { - customerFQDN := hostname - if !strings.HasSuffix(customerFQDN, ".") { - customerFQDN = customerFQDN + "." - } - - expectedFQDN := h.cnameTarget - if !strings.HasSuffix(expectedFQDN, ".") { - expectedFQDN = expectedFQDN + "." - } - - msg := &dns.Msg{MsgHeader: dns.MsgHeader{ID: dns.ID(), RecursionDesired: true}} - msg.Question = []dns.RR{&dns.CNAME{Hdr: dns.Header{Name: customerFQDN, Class: dns.ClassINET}}} - - dnsCtx, cancel := context.WithTimeout(ctx, dnsExchangeTimeout) - defer cancel() - - client := dns.NewClient() - - resp, _, err := client.Exchange(dnsCtx, msg, "udp", h.resolverAddr) - if err != nil { - return fmt.Errorf("cannot exchange dns message: %w", err) - } - - if len(resp.Answer) == 0 { - return fmt.Errorf("no cname records found for domain %q", hostname) - } - - if len(resp.Answer) > 1 { - return fmt.Errorf("multiple cname records found for domain %q", hostname) - } - - resolvedRecord, ok := resp.Answer[0].(*dns.CNAME) - if !ok { - return fmt.Errorf("first answer is not a cname record for domain %q", hostname) - } - - if !strings.EqualFold(expectedFQDN, resolvedRecord.Target) { - return fmt.Errorf( - "cname target mismatch: domain %q resolves to %q, expected %q", - hostname, - resolvedRecord.Target, - expectedFQDN, - ) - } - - return nil -} - -func (h *provisionHandler) checkCAARecords(ctx context.Context, hostname string) error { - fqdn := hostname - if !strings.HasSuffix(fqdn, ".") { - fqdn = fqdn + "." - } - - msg := &dns.Msg{MsgHeader: dns.MsgHeader{ID: dns.ID(), RecursionDesired: true}} - msg.Question = []dns.RR{&dns.CAA{Hdr: dns.Header{Name: fqdn, Class: dns.ClassINET}}} - - dnsCtx, cancel := context.WithTimeout(ctx, dnsExchangeTimeout) - defer cancel() - - client := dns.NewClient() - - resp, _, err := client.Exchange( - dnsCtx, - msg, - "udp", - h.resolverAddr, - ) - if err != nil { - return fmt.Errorf("cannot exchange dns message for caa records: %w", err) - } - - var caaRecords []*dns.CAA - - for _, rr := range resp.Answer { - if caa, ok := rr.(*dns.CAA); ok { - caaRecords = append(caaRecords, caa) - } - } - - if len(caaRecords) == 0 { - return nil - } - - for _, caa := range caaRecords { - if caa.Tag == "issue" { - issuer, _, _ := strings.Cut(caa.Value, ";") - if strings.EqualFold(strings.TrimSpace(issuer), h.caaIssuerDomain) { - return nil - } - } - } - - return fmt.Errorf( - "%w: domain %q by %q", - ErrCAANotPermitted, - hostname, - h.caaIssuerDomain, - ) -} - -func (h *provisionHandler) skipsDNSChecks( - ctx context.Context, - conn pg.Querier, - hostname string, -) (bool, error) { - if h.managedBaseDomain == "" { - return false, nil - } - - suffix := "." + h.managedBaseDomain - if hostname != h.managedBaseDomain && !strings.HasSuffix(hostname, suffix) { - return false, nil - } - - domain := &coredata.CustomDomain{} - - err := domain.LoadByDomain(ctx, conn, coredata.NewNoScope(), hostname) - if errors.Is(err, coredata.ErrResourceNotFound) { - return true, nil - } - - if err != nil { - return false, fmt.Errorf("cannot load custom domain: %w", err) - } - - return domain.Managed, nil -} - -func (h *provisionHandler) resetStaleCertificate( +func (h *pollOrderHandler) resetStaleCertificate( ctx context.Context, tx pg.Tx, certificate *coredata.Certificate, diff --git a/pkg/certmanager/service.go b/pkg/certmanager/service.go index cc7ca56d1..6918f2927 100644 --- a/pkg/certmanager/service.go +++ b/pkg/certmanager/service.go @@ -34,12 +34,13 @@ type ( // a generic core service and knows nothing about the resources a // certificate protects; callers reference a certificate by its ID. Service struct { - pg *pg.Client - acmeService *ACMEService - encryptionKey cipher.EncryptionKey - logger *log.Logger - provisionWorker *worker.Worker[coredata.Certificate] - renewWorker *worker.Worker[coredata.Certificate] + pg *pg.Client + acmeService *ACMEService + encryptionKey cipher.EncryptionKey + logger *log.Logger + beginChallengeWorker *worker.Worker[coredata.Certificate] + pollOrderWorker *worker.Worker[coredata.Certificate] + renewWorker *worker.Worker[coredata.Certificate] } // Config holds the SSL provisioning parameters for the service workers. @@ -75,15 +76,21 @@ func NewService( acmeService: acmeService, encryptionKey: encryptionKey, logger: logger, - provisionWorker: NewProvisionWorker( + beginChallengeWorker: NewBeginChallengeWorker( pgClient, acmeService, - encryptionKey, cfg.CnameTarget, cfg.CAAIssuerDomain, cfg.ResolverAddr, cfg.ManagedBaseDomain, - logger.Named("provision-worker"), + logger.Named("begin-challenge-worker"), + worker.WithInterval(provisionInterval), + ), + pollOrderWorker: NewPollOrderWorker( + pgClient, + acmeService, + encryptionKey, + logger.Named("poll-order-worker"), worker.WithInterval(provisionInterval), ), renewWorker: NewRenewWorker( @@ -100,7 +107,13 @@ func (s *Service) Run(ctx context.Context) error { g.Go( func() error { - return s.provisionWorker.Run(gctx) + return s.beginChallengeWorker.Run(gctx) + }, + ) + + g.Go( + func() error { + return s.pollOrderWorker.Run(gctx) }, ) diff --git a/pkg/coredata/certificate.go b/pkg/coredata/certificate.go index 933e3a92b..630e5372d 100644 --- a/pkg/coredata/certificate.go +++ b/pkg/coredata/certificate.go @@ -877,20 +877,15 @@ WHERE return nil } -func (c *Certificate) LoadNextForProvisioningForUpdateSkipLocked( +// LoadNextForBeginChallengeForUpdateSkipLocked selects the next PENDING/RENEWING +// row eligible to start a new ACME order. These rows 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. +func (c *Certificate) LoadNextForBeginChallengeForUpdateSkipLocked( ctx context.Context, tx pg.Tx, - pollLease time.Duration, ) error { - // 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, @@ -913,18 +908,90 @@ FROM certificates WHERE status = ANY(@statuses) + AND ( + ssl_last_attempt_at IS NULL + OR ssl_last_attempt_at < CURRENT_TIMESTAMP - ( + INTERVAL '15 minutes' * (POWER(2, LEAST(ssl_retry_count, 5))::int) + ) + ) +ORDER BY + updated_at ASC +LIMIT 1 +FOR UPDATE SKIP LOCKED +` + + rows, err := tx.Query( + ctx, + q, + pgx.StrictNamedArgs{ + "statuses": []string{ + string(CertificateStatusPending), + string(CertificateStatusRenewing), + }, + }, + ) + if err != nil { + return fmt.Errorf("cannot query certificate begin-challenge queue: %w", err) + } + + certificate, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[Certificate]) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return ErrResourceNotFound + } + + return fmt.Errorf("cannot collect certificate: %w", err) + } + + *c = certificate + + return nil +} + +// LoadNextForPollOrderForUpdateSkipLocked selects the next PROVISIONING row +// eligible to poll its ACME order. A row with an open order becomes 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. A row +// without an open order is a defensive edge case (the state machine always +// pairs PROVISIONING with an order) and falls back to the same exponential +// backoff as begin-challenge rows. +func (c *Certificate) LoadNextForPollOrderForUpdateSkipLocked( + ctx context.Context, + tx pg.Tx, + pollLease time.Duration, +) error { + q := ` +SELECT + id, + hostname, + http_challenge_token, + http_challenge_key_auth, + http_challenge_url, + http_order_url, + ssl_certificate, + encrypted_ssl_private_key, + ssl_certificate_chain, + status, + ssl_expires_at, + ssl_retry_count, + ssl_last_attempt_at, + provisioning_error, + created_at, + updated_at +FROM + certificates +WHERE + status = @provisioning_status AND ( ssl_last_attempt_at IS NULL OR ( - status = @provisioning_status - AND http_order_url IS NOT NULL + http_order_url IS NOT NULL AND ssl_last_attempt_at < CURRENT_TIMESTAMP - make_interval(secs => @poll_lease_seconds) ) OR ( - NOT ( - status = @provisioning_status - AND http_order_url IS NOT NULL - ) + http_order_url IS NULL AND ssl_last_attempt_at < CURRENT_TIMESTAMP - ( INTERVAL '15 minutes' * (POWER(2, LEAST(ssl_retry_count, 5))::int) ) @@ -940,17 +1007,12 @@ FOR UPDATE SKIP LOCKED ctx, q, pgx.StrictNamedArgs{ - "statuses": []string{ - string(CertificateStatusPending), - string(CertificateStatusProvisioning), - string(CertificateStatusRenewing), - }, "provisioning_status": string(CertificateStatusProvisioning), "poll_lease_seconds": pollLease.Seconds(), }, ) if err != nil { - return fmt.Errorf("cannot query certificate provisioning queue: %w", err) + return fmt.Errorf("cannot query certificate poll-order queue: %w", err) } certificate, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[Certificate])