as: Implement backoff for cooldown period of disabled webhooks

Author: vlasebian

api/ttn/lorawan/v3/api.md

@@ -1248,6 +1248,8 @@ Application Server configuration.
 | ----- | ---- | ----- | ----------- |
 | `unhealthy_attempts_threshold` | [`int64`](#int64) |  |  |
 | `unhealthy_retry_interval` | [`google.protobuf.Duration`](#google.protobuf.Duration) |  |  |
+| `unhealthy_retry_backoff_interval_min` | [`google.protobuf.Duration`](#google.protobuf.Duration) |  |  |
+| `unhealthy_retry_backoff_interval_max` | [`google.protobuf.Duration`](#google.protobuf.Duration) |  |  |
 
 ### <a name="ttn.lorawan.v3.DecodeDownlinkRequest">Message `DecodeDownlinkRequest`</a>
 

api/ttn/lorawan/v3/api.swagger.json

@@ -17358,6 +17358,12 @@
         },
         "unhealthy_retry_interval": {
           "type": "string"
+        },
+        "unhealthy_retry_backoff_interval_min": {
+          "type": "string"
+        },
+        "unhealthy_retry_backoff_interval_max": {
+          "type": "string"
         }
       }
     },

api/ttn/lorawan/v3/applicationserver.proto

@@ -102,9 +102,10 @@ message AsConfiguration {
   message Webhooks {
     int64 unhealthy_attempts_threshold = 1;
     google.protobuf.Duration unhealthy_retry_interval = 2;
-
     reserved 3;
     reserved "queue";
+    google.protobuf.Duration unhealthy_retry_backoff_interval_min = 4;
+    google.protobuf.Duration unhealthy_retry_backoff_interval_max = 5;
   }
   Webhooks webhooks = 2;
 }

pkg/applicationserver/config.go

@@ -155,21 +155,25 @@ type UplinkStorageConfig struct {
 
 // WebhooksConfig defines the configuration of the webhooks integration.
 type WebhooksConfig struct {
-	Registry                   web.WebhookRegistry `name:"-"`
-	Target                     string              `name:"target" description:"Target of the integration (direct)"`
-	Timeout                    time.Duration       `name:"timeout" description:"Wait timeout of the target to process the request"` // nolint:lll
-	QueueSize                  int                 `name:"queue-size" description:"Number of requests to queue"`
-	Workers                    int                 `name:"workers" description:"Number of workers to process requests"`
-	UnhealthyAttemptsThreshold int                 `name:"unhealthy-attempts-threshold" description:"Number of failed webhook attempts before the webhook is disabled"` // nolint:lll
-	UnhealthyRetryInterval     time.Duration       `name:"unhealthy-retry-interval" description:"Time interval after which disabled webhooks may execute again"`
-	Templates                  web.TemplatesConfig `name:"templates" description:"The store of the webhook templates"`
-	Downlinks                  web.DownlinksConfig `name:"downlink" description:"The downlink queue operations configuration"`
+	Registry                         web.WebhookRegistry `name:"-"`
+	Target                           string              `name:"target" description:"Target of the integration (direct)"`
+	Timeout                          time.Duration       `name:"timeout" description:"Wait timeout of the target to process the request"` // nolint:lll
+	QueueSize                        int                 `name:"queue-size" description:"Number of requests to queue"`
+	Workers                          int                 `name:"workers" description:"Number of workers to process requests"`
+	UnhealthyAttemptsThreshold       int                 `name:"unhealthy-attempts-threshold" description:"Number of failed webhook attempts before the webhook is disabled"` // nolint:lll
+	UnhealthyRetryInterval           time.Duration       `name:"unhealthy-retry-interval" description:"Time interval after which disabled webhooks may execute again"`
+	UnhealthyRetryBackoffIntervalMin time.Duration       `name:"unhealthy-retry-min-backoff-interval" description:"Minimum backoff interval for retrying unhealthy webhooks"`
+	UnhealthyRetryBackoffIntervalMax time.Duration       `name:"unhealthy-retry-max-backoff-interval" description:"Maximum backoff interval for retrying unhealthy webhooks"`
+	Templates                        web.TemplatesConfig `name:"templates" description:"The store of the webhook templates"`
+	Downlinks                        web.DownlinksConfig `name:"downlink" description:"The downlink queue operations configuration"`
 }
 
 func (c WebhooksConfig) toProto() *ttnpb.AsConfiguration_Webhooks {
 	return &ttnpb.AsConfiguration_Webhooks{
-		UnhealthyAttemptsThreshold: int64(c.UnhealthyAttemptsThreshold),
-		UnhealthyRetryInterval:     durationpb.New(c.UnhealthyRetryInterval),
+		UnhealthyAttemptsThreshold:       int64(c.UnhealthyAttemptsThreshold),
+		UnhealthyRetryInterval:           durationpb.New(c.UnhealthyRetryInterval),
+		UnhealthyRetryBackoffIntervalMin: durationpb.New(c.UnhealthyRetryBackoffIntervalMin),
+		UnhealthyRetryBackoffIntervalMax: durationpb.New(c.UnhealthyRetryBackoffIntervalMax),
 	}
 }
 
@@ -273,7 +277,7 @@ func (c WebhooksConfig) NewWebhooks(ctx context.Context, server io.Server) (web.
 	if c.UnhealthyAttemptsThreshold > 0 || c.UnhealthyRetryInterval > 0 {
 		registry := web.NewHealthStatusRegistry(c.Registry)
 		registry = web.NewCachedHealthStatusRegistry(registry)
-		target = sink.NewHealthCheckSink(target, registry, c.UnhealthyAttemptsThreshold, c.UnhealthyRetryInterval)
+		target = sink.NewHealthCheckSink(target, registry, c.UnhealthyAttemptsThreshold, c.UnhealthyRetryInterval, c.UnhealthyRetryBackoffIntervalMin, c.UnhealthyRetryBackoffIntervalMax)
 	}
 	if c.QueueSize > 0 || c.Workers > 0 {
 		target = sink.NewPooledSink(ctx, server, target, c.Workers, c.QueueSize)

pkg/applicationserver/io/web/sink/health.go

@@ -37,6 +37,8 @@ type healthCheckSink struct {
 
 	unhealthyAttemptsThreshold int
 	unhealthyRetryInterval     time.Duration
+	minBackoffInterval         time.Duration
+	maxBackoffInterval         time.Duration
 }
 
 // Process runs the health checks and sends the request to the underlying sink
@@ -82,8 +84,11 @@ func (hcs *healthCheckSink) preRunCheck(ctx context.Context) (healthState, error
 	case h.GetUnhealthy() != nil:
 		h := h.GetUnhealthy()
 		monitorOnly := hcs.unhealthyAttemptsThreshold <= 0 || hcs.unhealthyRetryInterval <= 0
-		nextAttemptAt := ttnpb.StdTime(h.LastFailedAttemptAt).Add(hcs.unhealthyRetryInterval)
+
+		backoffInterval := hcs.calculateBackoffInterval(h.FailedAttempts)
+		nextAttemptAt := ttnpb.StdTime(h.LastFailedAttemptAt).Add(backoffInterval)
 		retryIntervalPassed := time.Now().After(nextAttemptAt)
+
 		switch {
 		case monitorOnly:
 			// The system only monitors the health status, but does not block execution.
@@ -116,6 +121,36 @@ func (hcs *healthCheckSink) preRunCheck(ctx context.Context) (healthState, error
 	}
 }
 
+// calculateBackoffInterval calculates the retry interval using exponential backoff.
+// For webhooks that have been disabled (failedAttempts >= threshold), it uses exponential backoff
+// based on how many times the webhook has been disabled. The interval doubles on each disabled period
+// and is capped between minBackoffInterval and maxBackoffInterval.
+func (hcs *healthCheckSink) calculateBackoffInterval(failedAttempts uint64) time.Duration {
+	if failedAttempts < uint64(hcs.unhealthyAttemptsThreshold) {
+		return hcs.unhealthyRetryInterval
+	}
+
+	// Calculate how many times the webhook has been disabled.
+	disabledPeriods := failedAttempts / uint64(hcs.unhealthyAttemptsThreshold)
+
+	// Use exponential backoff: baseInterval * 2^disabledPeriods, capped at max.
+	backoffInterval := hcs.unhealthyRetryInterval
+	for i := uint64(1); i < disabledPeriods; i++ {
+		backoffInterval *= 2
+		if backoffInterval > hcs.maxBackoffInterval {
+			backoffInterval = hcs.maxBackoffInterval
+			break
+		}
+	}
+
+	// Ensure the backoff is at least the minimum.
+	if backoffInterval < hcs.minBackoffInterval {
+		backoffInterval = hcs.minBackoffInterval
+	}
+
+	return backoffInterval
+}
+
 // executeAndRecord runs the provided request using the underlying sink and records the health status.
 func (hcs *healthCheckSink) executeAndRecord(
 	ctx context.Context, req *http.Request, lastKnownState healthState,
@@ -164,14 +199,19 @@ func (hcs *healthCheckSink) executeAndRecord(
 }
 
 // NewHealthCheckSink creates a Sink that records the health status of the webhooks and stops them from executing if
-// too many fail in a specified interval of time.
+// too many fail in a specified interval of time. The cooldown period uses exponential backoff, starting from
+// unhealthyRetryInterval and doubling on each subsequent disabled period, capped between minBackoffInterval and
+// maxBackoffInterval.
 func NewHealthCheckSink(
-	sink Sink, registry HealthStatusRegistry, unhealthyAttemptsThreshold int, unhealthyRetryInterval time.Duration,
+	sink Sink, registry HealthStatusRegistry, unhealthyAttemptsThreshold int, unhealthyRetryInterval,
+	minBackoffInterval, maxBackoffInterval time.Duration,
 ) Sink {
 	return &healthCheckSink{
 		sink:                       sink,
 		registry:                   registry,
 		unhealthyAttemptsThreshold: unhealthyAttemptsThreshold,
 		unhealthyRetryInterval:     unhealthyRetryInterval,
+		minBackoffInterval:         minBackoffInterval,
+		maxBackoffInterval:         maxBackoffInterval,
 	}
 }

pkg/applicationserver/io/web/sink/health_test.go

@@ -66,7 +66,7 @@ func TestHealthCheckSink(t *testing.T) { // nolint:gocyclo
 	target := mock.New(sinkCh)
 
 	registry := web.NewHealthStatusRegistry(webhookRegistry)
-	healthSink := sink.NewHealthCheckSink(target, registry, 4, 8*timeout)
+	healthSink := sink.NewHealthCheckSink(target, registry, 4, 8*timeout, 4*timeout, 32*timeout)
 
 	ctx = internal.WithWebhookData(ctx, &internal.WebhookData{
 		EndDeviceIDs: registeredDeviceID,
@@ -151,7 +151,7 @@ func TestHealthCheckSink(t *testing.T) { // nolint:gocyclo
 		}
 	}
 
-	// We wait for the cooldown period to pass.
+	// We wait for the first cooldown period to pass (8 * timeout).
 	time.Sleep(8 * timeout)
 
 	// The sink should now do one attempt, and fail the rest.
@@ -191,8 +191,11 @@ func TestHealthCheckSink(t *testing.T) { // nolint:gocyclo
 		}
 	}
 
-	// We wait for the cooldown period to pass.
-	time.Sleep(8 * timeout)
+	// We wait for the second cooldown period to pass.
+	// Since the webhook has been disabled twice now (failedAttempts went from 4 to 5, then to 8),
+	// the backoff interval doubles from 8*timeout to 16*timeout.
+	// We wait for 16*timeout to ensure the retry period has passed.
+	time.Sleep(16 * timeout)
 
 	// We reset the error and expect the health status to recover.
 	target.SetError(nil)