diff --git a/backend-go/cmd/poller/main.go b/backend-go/cmd/poller/main.go index 8fb8714..ffab945 100644 --- a/backend-go/cmd/poller/main.go +++ b/backend-go/cmd/poller/main.go @@ -138,6 +138,8 @@ func runCycle( return } + evaluator.WaitForNotifications() + duration := time.Since(startTime) log.Printf("[%s] Cycle complete: %d observations, %d alerts fired in %s", time.Now().UTC().Format(time.RFC3339), diff --git a/backend-go/internal/protocols/ethereum/jsonrpc.go b/backend-go/internal/protocols/ethereum/jsonrpc.go index b8a26f2..4c87eda 100644 --- a/backend-go/internal/protocols/ethereum/jsonrpc.go +++ b/backend-go/internal/protocols/ethereum/jsonrpc.go @@ -18,7 +18,7 @@ import ( const ( rpcTimeoutMS = 30000 rpcMaxRetries = 3 - rpcRetryBaseMS = 1000 + rpcRetryBaseMS = 2000 ) type JsonRpcEthereum struct { diff --git a/backend-go/internal/services/evaluator.go b/backend-go/internal/services/evaluator.go index 0f151a7..1af176b 100644 --- a/backend-go/internal/services/evaluator.go +++ b/backend-go/internal/services/evaluator.go @@ -4,8 +4,11 @@ import ( "context" "fmt" "log" + "sync" "time" + "golang.org/x/sync/semaphore" + "github.com/kjannette/koin-ping/backend-go/internal/domain" "github.com/kjannette/koin-ping/backend-go/internal/models" "github.com/kjannette/koin-ping/backend-go/internal/notifications" @@ -17,6 +20,7 @@ const ( notificationTimeout = 30 * time.Second notificationMaxRetries = 3 notificationRetryBase = time.Second + maxConcurrentNotifications = 5 ) type EvaluatorService struct { @@ -27,6 +31,8 @@ type EvaluatorService struct { notifConfigs *models.NotificationConfigModel resendAPIKey string emailFrom string + notifSem *semaphore.Weighted + notifWg sync.WaitGroup } func NewEvaluatorService( @@ -46,6 +52,7 @@ func NewEvaluatorService( notifConfigs: notifConfigs, resendAPIKey: resendAPIKey, emailFrom: emailFrom, + notifSem: semaphore.NewWeighted(maxConcurrentNotifications), } } @@ -189,7 +196,14 @@ func (s *EvaluatorService) fireAlert(ctx context.Context, rule domain.AlertRule, if addr != nil { userID := addr.UserID address := addr.Address + if err := s.notifSem.Acquire(ctx, 1); err != nil { + log.Printf("Failed to acquire notification semaphore for rule %d: %v", rule.ID, err) + return nil + } + s.notifWg.Add(1) go func() { + defer s.notifSem.Release(1) + defer s.notifWg.Done() notifCtx, cancel := context.WithTimeout(context.Background(), notificationTimeout) defer cancel() s.sendNotification(notifCtx, userID, message, obs, addressLabel, rule, address) @@ -199,6 +213,11 @@ func (s *EvaluatorService) fireAlert(ctx context.Context, rule domain.AlertRule, return nil } +// WaitForNotifications blocks until all in-flight notification goroutines finish. +func (s *EvaluatorService) WaitForNotifications() { + s.notifWg.Wait() +} + func (s *EvaluatorService) buildNotifiers(cfg *domain.NotificationConfig) []notifications.Notifier { var notifiers []notifications.Notifier