mirror of
https://github.com/gotenberg/gotenberg.git
synced 2026-10-17 09:43:17 +01:00
fix(webhook): bound webhook delivery with its own retry budget
This commit is contained in:
@@ -31,10 +31,33 @@ type client struct {
|
|||||||
extraHttpHeaders map[string]string
|
extraHttpHeaders map[string]string
|
||||||
startTime time.Time
|
startTime time.Time
|
||||||
|
|
||||||
|
// deliveryTimeout bounds one delivery including retries. See
|
||||||
|
// [Webhook.deliveryTimeout].
|
||||||
|
deliveryTimeout time.Duration
|
||||||
|
|
||||||
client *retryablehttp.Client
|
client *retryablehttp.Client
|
||||||
logger *slog.Logger
|
logger *slog.Logger
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// deliveryContext returns the context one delivery runs on.
|
||||||
|
//
|
||||||
|
// It keeps the values of ctx, so trace propagation and logging correlation
|
||||||
|
// survive, and replaces its cancellation with a fresh budget. Threading the
|
||||||
|
// conversion context straight through does not work: a delivery starts after
|
||||||
|
// the handler returned, so that deadline may already be spent and the callback
|
||||||
|
// would fail without a single attempt.
|
||||||
|
func (c client) deliveryContext(ctx context.Context) (context.Context, context.CancelFunc) {
|
||||||
|
timeout := c.deliveryTimeout
|
||||||
|
if timeout <= 0 {
|
||||||
|
// An unset budget would expire the delivery before its first attempt.
|
||||||
|
// [Webhook.deliveryTimeout] never returns a non-positive value, so this
|
||||||
|
// only guards a caller that builds a client without one.
|
||||||
|
timeout = minDeliveryTimeout
|
||||||
|
}
|
||||||
|
|
||||||
|
return context.WithTimeout(context.WithoutCancel(ctx), timeout)
|
||||||
|
}
|
||||||
|
|
||||||
// send call the webhook either to send the success response or the error response.
|
// send call the webhook either to send the success response or the error response.
|
||||||
func (c client) send(ctx context.Context, body io.Reader, headers map[string]string, errored bool) error {
|
func (c client) send(ctx context.Context, body io.Reader, headers map[string]string, errored bool) error {
|
||||||
url := c.url
|
url := c.url
|
||||||
@@ -57,6 +80,9 @@ func (c client) send(ctx context.Context, body io.Reader, headers map[string]str
|
|||||||
spanName = fmt.Sprintf("%s Webhook Error", method)
|
spanName = fmt.Sprintf("%s Webhook Error", method)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
ctx, cancel := c.deliveryContext(ctx)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
tracer := gotenberg.Tracer()
|
tracer := gotenberg.Tracer()
|
||||||
ctx, span := tracer.Start(ctx, spanName,
|
ctx, span := tracer.Start(ctx, spanName,
|
||||||
trace.WithSpanKind(trace.SpanKindClient),
|
trace.WithSpanKind(trace.SpanKindClient),
|
||||||
@@ -64,7 +90,10 @@ func (c client) send(ctx context.Context, body io.Reader, headers map[string]str
|
|||||||
)
|
)
|
||||||
defer span.End()
|
defer span.End()
|
||||||
|
|
||||||
req, err := retryablehttp.NewRequest(method, url, body)
|
// The request must carry ctx: retryablehttp.NewRequest builds on
|
||||||
|
// [context.Background], and its wait between attempts is a select on the
|
||||||
|
// request context, so a contextless request cannot be interrupted.
|
||||||
|
req, err := retryablehttp.NewRequestWithContext(ctx, method, url, body)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
span.RecordError(err)
|
span.RecordError(err)
|
||||||
span.SetStatus(codes.Error, err.Error())
|
span.SetStatus(codes.Error, err.Error())
|
||||||
@@ -165,6 +194,9 @@ func (c client) sendEvent(ctx context.Context, correlationIdHeader, correlationI
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
ctx, cancel := c.deliveryContext(ctx)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
tracer := gotenberg.Tracer()
|
tracer := gotenberg.Tracer()
|
||||||
ctx, span := tracer.Start(ctx, "POST Webhook Event",
|
ctx, span := tracer.Start(ctx, "POST Webhook Event",
|
||||||
trace.WithSpanKind(trace.SpanKindClient),
|
trace.WithSpanKind(trace.SpanKindClient),
|
||||||
@@ -172,7 +204,7 @@ func (c client) sendEvent(ctx context.Context, correlationIdHeader, correlationI
|
|||||||
)
|
)
|
||||||
defer span.End()
|
defer span.End()
|
||||||
|
|
||||||
req, err := retryablehttp.NewRequest(http.MethodPost, c.eventsUrl, b)
|
req, err := retryablehttp.NewRequestWithContext(ctx, http.MethodPost, c.eventsUrl, b)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
span.RecordError(err)
|
span.RecordError(err)
|
||||||
span.SetStatus(codes.Error, err.Error())
|
span.SetStatus(codes.Error, err.Error())
|
||||||
|
|||||||
@@ -222,6 +222,7 @@ func webhookMiddleware(w *Webhook) api.Middleware {
|
|||||||
eventsUrl: webhookEventsUrl,
|
eventsUrl: webhookEventsUrl,
|
||||||
extraHttpHeaders: extraHttpHeaders,
|
extraHttpHeaders: extraHttpHeaders,
|
||||||
startTime: startTime,
|
startTime: startTime,
|
||||||
|
deliveryTimeout: w.deliveryTimeout(),
|
||||||
|
|
||||||
client: &retryablehttp.Client{
|
client: &retryablehttp.Client{
|
||||||
HTTPClient: gotenberg.NewOutboundHttpClient(w.clientTimeout, w.allowList, w.denyList, w.enableEnvironmentProxy, ipOpts...),
|
HTTPClient: gotenberg.NewOutboundHttpClient(w.clientTimeout, w.allowList, w.denyList, w.enableEnvironmentProxy, ipOpts...),
|
||||||
@@ -230,7 +231,10 @@ func webhookMiddleware(w *Webhook) api.Middleware {
|
|||||||
RetryWaitMax: w.retryMaxWait,
|
RetryWaitMax: w.retryMaxWait,
|
||||||
Logger: gotenberg.NewLeveledLogger(ctx.Log()),
|
Logger: gotenberg.NewLeveledLogger(ctx.Log()),
|
||||||
CheckRetry: retryablehttp.DefaultRetryPolicy,
|
CheckRetry: retryablehttp.DefaultRetryPolicy,
|
||||||
Backoff: retryablehttp.DefaultBackoff,
|
// Not DefaultBackoff: it returns a remote Retry-After
|
||||||
|
// verbatim, ignoring --webhook-retry-max-wait (env
|
||||||
|
// WEBHOOK_RETRY_MAX_WAIT).
|
||||||
|
Backoff: gotenberg.ClampedBackoff,
|
||||||
},
|
},
|
||||||
logger: ctx.Log(),
|
logger: ctx.Log(),
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -92,6 +92,27 @@ func (w *Webhook) Provision(ctx *gotenberg.Context) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// minDeliveryTimeout is the floor for [Webhook.deliveryTimeout], so that a
|
||||||
|
// deliberately tiny --webhook-client-timeout (env WEBHOOK_CLIENT_TIMEOUT) never
|
||||||
|
// leaves a delivery with no budget at all.
|
||||||
|
const minDeliveryTimeout = 1 * time.Second
|
||||||
|
|
||||||
|
// deliveryTimeout bounds one webhook delivery including its retries. It is the
|
||||||
|
// worst case a correctly behaving remote produces: one client timeout per
|
||||||
|
// attempt, plus the capped wait between attempts.
|
||||||
|
//
|
||||||
|
// A delivery runs after the handler returned, so it cannot borrow the
|
||||||
|
// conversion deadline. Without this budget it would have none, because
|
||||||
|
// [retryablehttp] builds its requests on [context.Background].
|
||||||
|
func (w *Webhook) deliveryTimeout() time.Duration {
|
||||||
|
timeout := w.clientTimeout*time.Duration(w.maxRetry+1) + w.retryMaxWait*time.Duration(w.maxRetry)
|
||||||
|
if timeout < minDeliveryTimeout {
|
||||||
|
return minDeliveryTimeout
|
||||||
|
}
|
||||||
|
|
||||||
|
return timeout
|
||||||
|
}
|
||||||
|
|
||||||
// Middlewares returns the middleware.
|
// Middlewares returns the middleware.
|
||||||
func (w *Webhook) Middlewares() ([]api.Middleware, error) {
|
func (w *Webhook) Middlewares() ([]api.Middleware, error) {
|
||||||
if w.disable {
|
if w.disable {
|
||||||
|
|||||||
95
pkg/modules/webhook/webhook_test.go
Normal file
95
pkg/modules/webhook/webhook_test.go
Normal file
@@ -0,0 +1,95 @@
|
|||||||
|
package webhook
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestWebhook_deliveryTimeout(t *testing.T) {
|
||||||
|
for _, tc := range []struct {
|
||||||
|
scenario string
|
||||||
|
clientTimeout time.Duration
|
||||||
|
maxRetry int
|
||||||
|
retryMaxWait time.Duration
|
||||||
|
want time.Duration
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
scenario: "shipped defaults",
|
||||||
|
clientTimeout: 30 * time.Second,
|
||||||
|
maxRetry: 4,
|
||||||
|
retryMaxWait: 30 * time.Second,
|
||||||
|
want: 270 * time.Second,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
scenario: "no retry is one client timeout",
|
||||||
|
clientTimeout: 30 * time.Second,
|
||||||
|
maxRetry: 0,
|
||||||
|
retryMaxWait: 30 * time.Second,
|
||||||
|
want: 30 * time.Second,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
scenario: "a zero client timeout still gets a budget",
|
||||||
|
clientTimeout: 0,
|
||||||
|
maxRetry: 0,
|
||||||
|
retryMaxWait: 0,
|
||||||
|
want: minDeliveryTimeout,
|
||||||
|
},
|
||||||
|
} {
|
||||||
|
t.Run(tc.scenario, func(t *testing.T) {
|
||||||
|
w := &Webhook{
|
||||||
|
clientTimeout: tc.clientTimeout,
|
||||||
|
maxRetry: tc.maxRetry,
|
||||||
|
retryMaxWait: tc.retryMaxWait,
|
||||||
|
}
|
||||||
|
if got := w.deliveryTimeout(); got != tc.want {
|
||||||
|
t.Fatalf("deliveryTimeout() = %s, want %s", got, tc.want)
|
||||||
|
}
|
||||||
|
if w.deliveryTimeout() <= 0 {
|
||||||
|
t.Fatal("deliveryTimeout() must always be positive")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// A delivery must not inherit the conversion deadline. It runs after the
|
||||||
|
// handler returned, so that deadline is often already spent, which would fail
|
||||||
|
// the callback without a single attempt.
|
||||||
|
func TestClient_deliveryContext_IgnoresAnExpiredParentDeadline(t *testing.T) {
|
||||||
|
expired, cancelExpired := context.WithTimeout(context.Background(), -1*time.Second)
|
||||||
|
defer cancelExpired()
|
||||||
|
|
||||||
|
if expired.Err() == nil {
|
||||||
|
t.Fatal("expected the parent context to be expired")
|
||||||
|
}
|
||||||
|
|
||||||
|
c := client{deliveryTimeout: 30 * time.Second}
|
||||||
|
ctx, cancel := c.deliveryContext(expired)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
t.Fatalf("delivery context inherited the expired parent: %v", ctx.Err())
|
||||||
|
}
|
||||||
|
|
||||||
|
deadline, ok := ctx.Deadline()
|
||||||
|
if !ok {
|
||||||
|
t.Fatal("delivery context has no deadline, so a delivery would be unbounded")
|
||||||
|
}
|
||||||
|
if remaining := time.Until(deadline); remaining <= 0 {
|
||||||
|
t.Fatalf("delivery budget = %s, want a positive value", remaining)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The delivery context must still be bounded, so a hostile remote cannot hold
|
||||||
|
// the goroutine and its output file open indefinitely.
|
||||||
|
func TestClient_deliveryContext_IsBounded(t *testing.T) {
|
||||||
|
c := client{deliveryTimeout: 50 * time.Millisecond}
|
||||||
|
ctx, cancel := c.deliveryContext(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
case <-time.After(5 * time.Second):
|
||||||
|
t.Fatal("delivery context never expired")
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user