bkinnightskytw
/
agent-orchestrator
mirror of https://github.com/ComposioHQ/agent-orchestrator.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: address notifier review cleanup

This commit is contained in:
whoisasx 2026-06-01 03:09:55 +05:30
parent 041c8c8f7f
commit d39e8e0da3
5 changed files with 93 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
backend/internal/notification/retry.go
View File

@ -1,8 +1,9 @@
package notification
import (
crand "crypto/rand"
"encoding/binary"
"math"
"math/rand"
"strings"
"time"
@ -30,7 +31,7 @@ func RetryPolicyFromConfig(cfg config.NotificationRetryConfig) RetryPolicy {
LeaseTTL: settings.Retry.LeaseTTL,
BatchSize: settings.Retry.BatchSize,
Jitter: retryJitterFraction,
RandFloat64: rand.Float64,
RandFloat64: cryptoRandFloat64,
}
}
@ -69,13 +70,24 @@ func (p RetryPolicy) BackoffDelay(attempt int) time.Duration {
}
randFloat := p.RandFloat64
if randFloat == nil {
randFloat = rand.Float64
randFloat = cryptoRandFloat64
}
// rand in [0,1) -> factor in [1-jitter, 1+jitter)
factor := 1 - p.Jitter + (2 * p.Jitter * randFloat())
return time.Duration(float64(delay) * factor)
}
func cryptoRandFloat64() float64 {
var b [8]byte
if _, err := crand.Read(b[:]); err != nil {
// Fall back to a time-derived value only if the OS CSPRNG fails. The
// fallback still avoids math/rand's deterministic process seed.
return float64(time.Now().UnixNano()&((1<<53)-1)) / float64(1<<53)
}
// Match math/rand.Float64's 53 bits of precision in [0,1).
return float64(binary.BigEndian.Uint64(b[:])>>11) / float64(1<<53)
}
func (p RetryPolicy) NextAttemptAt(now time.Time, attempt int) time.Time {
return now.Add(p.BackoffDelay(attempt))
}

9
backend/internal/notification/settings.go
View File

@ -16,11 +16,7 @@ type staticSettings struct {
}
func SettingsFromConfig(cfg config.Config) SettingsProvider {
settings := cfg.Notifications
if isZeroNotificationConfig(settings) {
settings = config.DefaultNotificationConfig()
}
return staticSettings{cfg: NormalizeSettings(settings)}
return staticSettings{cfg: NormalizeSettings(cfg.Notifications)}
}
func StaticSettings(cfg config.NotificationConfig) SettingsProvider {
@ -35,9 +31,6 @@ func (s staticSettings) Settings(context.Context) config.NotificationConfig {
// explicit route overrides, including an explicit empty route list.
func NormalizeSettings(in config.NotificationConfig) config.NotificationConfig {
def := config.DefaultNotificationConfig()
if isZeroNotificationConfig(in) {
return cloneSettings(def)
}
out := in
if isZeroDashboardConfig(in.Dashboard) {

14
backend/internal/notification/settings_test.go
View File

@ -9,15 +9,25 @@ import (
)
func TestSettingsFromConfigDefaultsWhenUnset(t *testing.T) {
got := SettingsFromConfig(config.Config{}).Settings(context.Background())
got := SettingsFromConfig(config.Config{Notifications: config.DefaultNotificationConfig()}).Settings(context.Background())
if !got.Enabled || !got.Desktop.Enabled || !got.Dashboard.Enabled {
t.Fatalf("zero config should resolve safe enabled defaults: %+v", got)
t.Fatalf("default config should resolve safe enabled defaults: %+v", got)
}
if got.Retry.MaxAttempts != 5 || got.Retry.BatchSize != 50 {
t.Fatalf("retry defaults = %+v", got.Retry)
}
}
func TestSettingsFromConfigPreservesExplicitGlobalDisable(t *testing.T) {
got := SettingsFromConfig(config.Config{Notifications: config.NotificationConfig{Enabled: false}}).Settings(context.Background())
if got.Enabled {
t.Fatalf("explicit disabled notifications should stay disabled: %+v", got)
}
if got.Retry.MaxAttempts != 5 || got.Routing.Priorities == nil {
t.Fatalf("disabled config should still receive non-global defaults: %+v", got)
}
}
func TestNormalizeSettingsPreservesExplicitEmptyRoute(t *testing.T) {
cfg := config.DefaultNotificationConfig()
cfg.Routing.Priorities[ports.PriorityUrgent] = []string{}

103
backend/internal/storage/sqlite/notification_delivery_store.go
View File

@ -70,39 +70,29 @@ func (s *Store) EnqueueDelivery(ctx context.Context, row notification.DeliveryRo
s.writeMu.Lock()
defer s.writeMu.Unlock()
insert := `INSERT INTO notification_deliveries (
id, notification_id, notification_seq, project_id, session_id,
route_name, sink, destination_key, request_json,
status, attempts, max_attempts, next_attempt_at, lease_owner, lease_expires_at,
last_error_code, last_error, external_id,
created_at, updated_at, delivered_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(notification_id, route_name, destination_key) DO NOTHING
RETURNING ` + deliveryColumns
got, err := scanDelivery(s.writeDB.QueryRowContext(ctx, insert,
row.ID,
string(row.NotificationID),
row.NotificationSeq,
string(row.ProjectID),
string(row.SessionID),
row.RouteName,
row.Sink,
row.DestinationKey,
string(row.RequestJSON),
string(row.Status),
row.Attempts,
row.MaxAttempts,
row.NextAttemptAt,
row.LeaseOwner,
nullTime(row.LeaseExpiresAt),
row.LastErrorCode,
row.LastError,
row.ExternalID,
row.CreatedAt,
row.UpdatedAt,
nullTime(row.DeliveredAt),
))
got, err := s.qw.InsertNotificationDelivery(ctx, gen.InsertNotificationDeliveryParams{
ID: row.ID,
NotificationID: string(row.NotificationID),
NotificationSeq: row.NotificationSeq,
ProjectID: string(row.ProjectID),
SessionID: string(row.SessionID),
RouteName: row.RouteName,
Sink: row.Sink,
DestinationKey: row.DestinationKey,
RequestJson: string(row.RequestJSON),
Status: string(row.Status),
Attempts: int64(row.Attempts),
MaxAttempts: int64(row.MaxAttempts),
NextAttemptAt: row.NextAttemptAt,
LeaseOwner: row.LeaseOwner,
LeaseExpiresAt: nullTime(row.LeaseExpiresAt),
LastErrorCode: row.LastErrorCode,
LastError: row.LastError,
ExternalID: row.ExternalID,
CreatedAt: row.CreatedAt,
UpdatedAt: row.UpdatedAt,
DeliveredAt: nullTime(row.DeliveredAt),
})
if errors.Is(err, sql.ErrNoRows) {
existing, readErr := s.getDeliveryByUniqueLocked(ctx, row.NotificationID, row.RouteName, row.DestinationKey)
if readErr != nil {
@ -113,7 +103,7 @@ RETURNING ` + deliveryColumns
if err != nil {
return notification.DeliveryRow{}, false, fmt.Errorf("insert notification delivery: %w", err)
}
return got, true, nil
return deliveryFromGen(got), true, nil
}
func (s *Store) ClaimDueDeliveries(ctx context.Context, sink string, owner string, now time.Time, limit int, lease time.Duration) ([]notification.DeliveryRow, error) {
@ -298,14 +288,14 @@ WHERE id = ? AND status NOT IN ('sent','failed','skipped','cancelled')`, reason,
}
func (s *Store) GetDelivery(ctx context.Context, id string) (notification.DeliveryRow, bool, error) {
row, err := scanDelivery(s.readDB.QueryRowContext(ctx, `SELECT `+deliveryColumns+` FROM notification_deliveries WHERE id = ?`, id))
row, err := s.qr.GetNotificationDelivery(ctx, id)
if errors.Is(err, sql.ErrNoRows) {
return notification.DeliveryRow{}, false, nil
}
if err != nil {
return notification.DeliveryRow{}, false, fmt.Errorf("get notification delivery %s: %w", id, err)
}
return row, true, nil
return deliveryFromGen(row), true, nil
}
func (s *Store) ListDeliveries(ctx context.Context, filter DeliveryFilter) ([]notification.DeliveryRow, error) {
@ -375,13 +365,15 @@ func (s *Store) updateDelivery(ctx context.Context, what string, query string, a
}
func (s *Store) getDeliveryByUniqueLocked(ctx context.Context, id domain.NotificationID, routeName, destinationKey string) (notification.DeliveryRow, error) {
row, err := scanDelivery(s.writeDB.QueryRowContext(ctx, `SELECT `+deliveryColumns+`
FROM notification_deliveries
WHERE notification_id = ? AND route_name = ? AND destination_key = ?`, string(id), routeName, destinationKey))
row, err := s.qw.GetNotificationDeliveryByUnique(ctx, gen.GetNotificationDeliveryByUniqueParams{
NotificationID: string(id),
RouteName: routeName,
DestinationKey: destinationKey,
})
if err != nil {
return notification.DeliveryRow{}, fmt.Errorf("get notification delivery by unique key: %w", err)
}
return row, nil
return deliveryFromGen(row), nil
}
type rowScanner interface {
@ -437,3 +429,34 @@ func scanDelivery(scanner rowScanner) (notification.DeliveryRow, error) {
}
return row, nil
}
func deliveryFromGen(r gen.NotificationDelivery) notification.DeliveryRow {
row := notification.DeliveryRow{
ID: r.ID,
NotificationID: domain.NotificationID(r.NotificationID),
NotificationSeq: r.NotificationSeq,
ProjectID: domain.ProjectID(r.ProjectID),
SessionID: domain.SessionID(r.SessionID),
RouteName: r.RouteName,
Sink: r.Sink,
DestinationKey: r.DestinationKey,
RequestJSON: []byte(r.RequestJson),
Status: notification.DeliveryStatus(r.Status),
Attempts: int(r.Attempts),
MaxAttempts: int(r.MaxAttempts),
NextAttemptAt: r.NextAttemptAt,
LeaseOwner: r.LeaseOwner,
LastErrorCode: r.LastErrorCode,
LastError: r.LastError,
ExternalID: r.ExternalID,
CreatedAt: r.CreatedAt,
UpdatedAt: r.UpdatedAt,
}
if r.LeaseExpiresAt.Valid {
row.LeaseExpiresAt = r.LeaseExpiresAt.Time
}
if r.DeliveredAt.Valid {
row.DeliveredAt = r.DeliveredAt.Time
}
return row
}

4
backend/notifier_wiring.go
View File

@ -14,10 +14,10 @@ type notifierStack struct {
done <-chan struct{}
}
func startNotifier(ctx context.Context, cfg config.Config, store *sqlite.Store, log *slog.Logger) (*notifierStack, error) {
func startNotifier(ctx context.Context, cfg config.Config, store *sqlite.Store, log *slog.Logger) *notifierStack {
mgr := notification.NewManager(store, notification.SettingsFromConfig(cfg), log)
done := mgr.Start(ctx)
return &notifierStack{Manager: mgr, done: done}, nil
return &notifierStack{Manager: mgr, done: done}
}
func (s *notifierStack) Stop() {