feat: Delivery timeout

Author: alexluong

cmd/e2e/configs/basic.go

@@ -67,6 +67,7 @@ func Basic(t *testing.T) *config.Config {
 		LogMaxConcurrency:               3,
 		RetryIntervalSeconds:            1,
 		RetryMaxCount:                   3,
+		DeliveryTimeoutSeconds:          5,
 		LogBatcherDelayThresholdSeconds: 1,
 		LogBatcherItemCountThreshold:    100,
 		MaxDestinationsPerTenant:        20,

internal/config/config.go

@@ -43,6 +43,7 @@ type Config struct {
 	LogMaxConcurrency               int
 	RetryIntervalSeconds            int
 	RetryMaxCount                   int
+	DeliveryTimeoutSeconds          int
 	LogBatcherDelayThresholdSeconds int
 	LogBatcherItemCountThreshold    int
 	DestinationMetadataPath         string
@@ -83,6 +84,7 @@ var defaultConfig = map[string]any{
 	"LOGMQ_MAX_CONCURRENCY":               1,
 	"RETRY_INTERVAL_SECONDS":              30,
 	"MAX_RETRY_COUNT":                     10,
+	"DELIVERY_TIMEOUT_SECONDS":            5,
 	"LOG_BATCHER_DELAY_THRESHOLD_SECONDS": 5,
 	"LOG_BATCHER_ITEM_COUNT_THRESHOLD":    100,
 	"MAX_DESTINATIONS_PER_TENANT":         20,
@@ -198,6 +200,7 @@ func Parse(flags Flags) (*Config, error) {
 		LogMaxConcurrency:               mustInt(viper, "LOGMQ_MAX_CONCURRENCY"),
 		RetryIntervalSeconds:            mustInt(viper, "RETRY_INTERVAL_SECONDS"),
 		RetryMaxCount:                   mustInt(viper, "MAX_RETRY_COUNT"),
+		DeliveryTimeoutSeconds:          mustInt(viper, "DELIVERY_TIMEOUT_SECONDS"),
 		LogBatcherDelayThresholdSeconds: mustInt(viper, "LOG_BATCHER_DELAY_THRESHOLD_SECONDS"),
 		LogBatcherItemCountThreshold:    mustInt(viper, "LOG_BATCHER_ITEM_COUNT_THRESHOLD"),
 		DestinationMetadataPath:         viper.GetString("DESTINATION_METADATA_PATH"),

internal/destregistry/registry.go

@@ -55,12 +55,14 @@ type registry struct {
 	metadata       map[string]*metadata.ProviderMetadata
 	providers      map[string]Provider
 	publishers     *lru.Cache[string, Publisher]
+	config         Config
 }
 
 type Config struct {
 	DestinationMetadataPath string
 	PublisherCacheSize      int
 	PublisherTTL            time.Duration
+	DeliveryTimeout         time.Duration
 }
 
 func NewRegistry(cfg *Config, logger *otelzap.Logger) *registry {
@@ -70,6 +72,9 @@ func NewRegistry(cfg *Config, logger *otelzap.Logger) *registry {
 	if cfg.PublisherTTL == 0 {
 		cfg.PublisherTTL = defaultPublisherTTL
 	}
+	if cfg.DeliveryTimeout == 0 {
+		cfg.DeliveryTimeout = defaultDeliveryTimeout
+	}
 
 	onEvict := func(key string, p Publisher) {
 		if err := p.Close(); err != nil {
@@ -87,6 +92,7 @@ func NewRegistry(cfg *Config, logger *otelzap.Logger) *registry {
 		metadata:       make(map[string]*metadata.ProviderMetadata),
 		providers:      make(map[string]Provider),
 		publishers:     cache,
+		config:         *cfg,
 	}
 }
 
@@ -115,11 +121,26 @@ func (r *registry) PublishEvent(ctx context.Context, destination *models.Destina
 	if err != nil {
 		return err
 	}
-	if err := publisher.Publish(ctx, event); err != nil {
+
+	// Create a new context with timeout
+	timeoutCtx, cancel := context.WithTimeout(ctx, r.config.DeliveryTimeout)
+	defer cancel()
+
+	if err := publisher.Publish(timeoutCtx, event); err != nil {
 		var publishErr *ErrDestinationPublishAttempt
 		if errors.As(err, &publishErr) {
 			return publishErr
 		}
+		if errors.Is(err, context.DeadlineExceeded) {
+			return &ErrDestinationPublishAttempt{
+				Err:      err,
+				Provider: destination.Type,
+				Data: map[string]interface{}{
+					"error":   "timeout",
+					"timeout": r.config.DeliveryTimeout.String(),
+				},
+			}
+		}
 		return &ErrUnexpectedPublishError{Err: err}
 	}
 	return nil
@@ -203,4 +224,5 @@ func (r *registry) ObfuscateDestination(destination *models.Destination) (*model
 var (
 	defaultPublisherCacheSize = 10000
 	defaultPublisherTTL       = time.Minute
+	defaultDeliveryTimeout    = 5 * time.Second
 )

internal/destregistry/registry_test.go

@@ -18,11 +18,13 @@ import (
 type mockProvider struct {
 	createCount int32
 	*destregistry.BaseProvider
+	publishDelay time.Duration
 }
 
 type mockPublisher struct {
-	id     int64
-	closed bool
+	id           int64
+	closed       bool
+	publishDelay time.Duration
 }
 
 var mockPublisherID int64
@@ -90,10 +92,20 @@ func (p *mockProvider) Validate(ctx context.Context, dest *models.Destination) e
 
 func (p *mockProvider) CreatePublisher(ctx context.Context, dest *models.Destination) (destregistry.Publisher, error) {
 	atomic.AddInt32(&p.createCount, 1)
-	return newMockPublisher(), nil
+	pub := newMockPublisher()
+	pub.publishDelay = p.publishDelay
+	return pub, nil
+}
+
+func (p *mockPublisher) Publish(ctx context.Context, event *models.Event) error {
+	select {
+	case <-time.After(p.publishDelay):
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
 }
 
-func (p *mockPublisher) Publish(ctx context.Context, event *models.Event) error { return nil }
 func (p *mockPublisher) Close() error {
 	p.closed = true
 	return nil
@@ -576,3 +588,57 @@ func TestObfuscateDestination(t *testing.T) {
 	assert.Equal(t, "****", obfuscated.Credentials["token"])                // 3 chars
 	assert.Equal(t, "****", obfuscated.Credentials["code"])                 // 4 chars
 }
+
+func TestPublishEventTimeout(t *testing.T) {
+	timeout := 100 * time.Millisecond
+	logger := testutil.CreateTestLogger(t)
+
+	t.Run("should not return timeout error when publish completes within timeout", func(t *testing.T) {
+		registry := destregistry.NewRegistry(&destregistry.Config{
+			DeliveryTimeout: timeout,
+		}, logger)
+
+		provider, err := newMockProvider()
+		require.NoError(t, err)
+		provider.publishDelay = timeout / 2
+		err = registry.RegisterProvider("test", provider)
+		require.NoError(t, err)
+
+		destination := &models.Destination{
+			Type: "test",
+		}
+		event := &models.Event{}
+
+		err = registry.PublishEvent(context.Background(), destination, event)
+		assert.NoError(t, err)
+	})
+
+	t.Run("should return timeout error when publish exceeds timeout", func(t *testing.T) {
+		registry := destregistry.NewRegistry(&destregistry.Config{
+			DeliveryTimeout: timeout,
+		}, logger)
+
+		provider, err := newMockProvider()
+		require.NoError(t, err)
+		provider.publishDelay = timeout * 2
+		err = registry.RegisterProvider("test", provider)
+		require.NoError(t, err)
+
+		destination := &models.Destination{
+			Type: "test",
+		}
+		event := &models.Event{}
+
+		err = registry.PublishEvent(context.Background(), destination, event)
+		assert.Error(t, err)
+
+		var publishErr *destregistry.ErrDestinationPublishAttempt
+		assert.ErrorAs(t, err, &publishErr)
+		assert.Equal(t, "test", publishErr.Provider)
+
+		data, ok := publishErr.Data.(map[string]interface{})
+		assert.True(t, ok, "Expected Data to be map[string]interface{}")
+		assert.Equal(t, "timeout", data["error"])
+		assert.Equal(t, timeout.String(), data["timeout"])
+	})
+}

internal/services/api/api.go

@@ -46,6 +46,7 @@ func NewService(ctx context.Context, wg *sync.WaitGroup, cfg *config.Config, log
 
 	registry := destregistry.NewRegistry(&destregistry.Config{
 		DestinationMetadataPath: cfg.DestinationMetadataPath,
+		DeliveryTimeout:         time.Duration(cfg.DeliveryTimeoutSeconds) * time.Second,
 	}, logger)
 	if err := destregistrydefault.RegisterDefault(registry, destregistrydefault.RegisterDefaultDestinationOptions{
 		Webhook: &destregistrydefault.DestWebhookConfig{

internal/services/delivery/delivery.go

@@ -64,6 +64,7 @@ func NewService(ctx context.Context,
 	if handler == nil {
 		registry := destregistry.NewRegistry(&destregistry.Config{
 			DestinationMetadataPath: cfg.DestinationMetadataPath,
+			DeliveryTimeout:         time.Duration(cfg.DeliveryTimeoutSeconds) * time.Second,
 		}, logger)
 		if err := destregistrydefault.RegisterDefault(registry, destregistrydefault.RegisterDefaultDestinationOptions{
 			Webhook: &destregistrydefault.DestWebhookConfig{