diff --git a/backend/internal/app/shutdown.go b/backend/internal/app/shutdown.go index e59e93c4..f808dd3c 100644 --- a/backend/internal/app/shutdown.go +++ b/backend/internal/app/shutdown.go @@ -50,7 +50,7 @@ func (a *App) shutdown(ctx context.Context) error { // Step 0.5: Close notification delivery service (Watermill router + subscriber) if a.notificationDeliverySvc != nil && a.notificationRunning.Swap(false) { - if closeErr := a.notificationDeliverySvc.Close(); closeErr != nil { + if closeErr := a.notificationDeliverySvc.Close(ctx); closeErr != nil { applogger.L().Errorf("Notification delivery service close error: %v", closeErr) shutdownErrs = append(shutdownErrs, closeErr) } else { diff --git a/backend/internal/service/coverage21_test.go b/backend/internal/service/coverage21_test.go index 6e4cb6ec..b3898d46 100644 --- a/backend/internal/service/coverage21_test.go +++ b/backend/internal/service/coverage21_test.go @@ -2746,7 +2746,7 @@ func TestNotifDeliveryService_StructLiteral_Cov21(t *testing.T) { func TestNotifDeliveryService_Close_Nil_Cov21(t *testing.T) { defer func() { _ = recover() }() svc := &NotificationDeliveryService{} - _ = svc.Close() + _ = svc.Close(context.Background()) } func TestNotifDeliveryService_Start_Nil_Cov21(t *testing.T) { diff --git a/backend/internal/service/coverage29_test.go b/backend/internal/service/coverage29_test.go index b91d661d..dff486fb 100644 --- a/backend/internal/service/coverage29_test.go +++ b/backend/internal/service/coverage29_test.go @@ -3109,7 +3109,7 @@ func TestNotificationDeliveryService_StructLit_Cov29(t *testing.T) { func TestNotificationDeliveryService_Close_Cov29(t *testing.T) { svc := &NotificationDeliveryService{} defer func() { _ = recover() }() - _ = svc.Close() + _ = svc.Close(context.Background()) } // ---------- AppliedSlaService ---------- diff --git a/backend/internal/service/coverage37_test.go b/backend/internal/service/coverage37_test.go index fa6ea128..092e0736 100644 --- a/backend/internal/service/coverage37_test.go +++ b/backend/internal/service/coverage37_test.go @@ -1284,7 +1284,7 @@ func TestNotificationDeliveryService_Start_Cov37(t *testing.T) { func TestNotificationDeliveryService_Close_Cov37(t *testing.T) { svc := &NotificationDeliveryService{} - safeCall_Cov37(func() { _ = svc.Close() }) + safeCall_Cov37(func() { _ = svc.Close(context.Background()) }) } // === NotificationService === diff --git a/backend/internal/service/notification_delivery_lifecycle_test.go b/backend/internal/service/notification_delivery_lifecycle_test.go index c53c8cd0..8c51b7a4 100644 --- a/backend/internal/service/notification_delivery_lifecycle_test.go +++ b/backend/internal/service/notification_delivery_lifecycle_test.go @@ -51,10 +51,15 @@ func (g *notificationAckGate) ProcessHook(next redis.ProcessHook) redis.ProcessH } g.group = fmt.Sprint(args[2]) g.startOnce.Do(func() { close(g.started) }) - <-g.release - err := next(ctx, cmd) - g.doneOnce.Do(func() { close(g.done) }) - return err + select { + case <-g.release: + err := next(ctx, cmd) + g.doneOnce.Do(func() { close(g.done) }) + return err + case <-ctx.Done(): + g.doneOnce.Do(func() { close(g.done) }) + return ctx.Err() + } } } @@ -105,7 +110,7 @@ func TestNotificationCloseTimeoutWaitsForHandlerBeforeDependenciesClose(t *testi handlers.Stop() <-cancelled - require.ErrorContains(t, service.Close(), "router close timeout") + require.ErrorContains(t, service.Close(context.Background()), "router close timeout") waitDone := make(chan struct{}) go func() { @@ -154,7 +159,7 @@ func TestNotificationShutdownWaitsForRedisStreamXAck(t *testing.T) { releaseAck := sync.OnceFunc(func() { close(gate.release) }) t.Cleanup(func() { releaseAck() - _ = service.Close() + _ = service.Close(context.Background()) }) runDone := make(chan error, 1) go func() { runDone <- service.Start(handlers.Context()) }() @@ -177,7 +182,7 @@ func TestNotificationShutdownWaitsForRedisStreamXAck(t *testing.T) { handlers.Stop() closeDone := make(chan error, 1) - go func() { closeDone <- service.Close() }() + go func() { closeDone <- service.Close(context.Background()) }() select { case err := <-closeDone: t.Fatalf("router closed before XAck completed: %v", err) @@ -199,6 +204,58 @@ func TestNotificationShutdownWaitsForRedisStreamXAck(t *testing.T) { require.Error(t, subscriberClient.Ping(context.Background()).Err()) } +func TestNotificationShutdownDeadlineLeavesRedisStreamMessagePending(t *testing.T) { + addr := startNotificationRedis(t) + topic := pubsub.TopicSystemNotification + gate := ¬ificationAckGate{ + topic: topic, + started: make(chan struct{}), + release: make(chan struct{}), + done: make(chan struct{}), + } + subscriberClient := redis.NewClient(&redis.Options{Addr: addr, PoolSize: 32}) + subscriberClient.AddHook(gate) + publisherClient := redis.NewClient(&redis.Options{Addr: addr}) + t.Cleanup(func() { + _ = subscriberClient.Close() + _ = publisherClient.Close() + }) + + service, err := NewNotificationDeliveryService(nil, nil, nil, nil, nil, nil, nil, subscriberClient) + require.NoError(t, err) + handlers := lifecycle.NewHandlerGroup() + service.SetHandlerGroup(handlers) + runDone := make(chan error, 1) + go func() { runDone <- service.Start(handlers.Context()) }() + <-service.router.Running() + + publisher, err := redisstream.NewPublisher(redisstream.PublisherConfig{Client: publisherClient}, watermill.NopLogger{}) + require.NoError(t, err) + t.Cleanup(func() { _ = publisher.Close() }) + require.NoError(t, publisher.Publish(topic, message.NewMessage(watermill.NewUUID(), []byte("{")))) + select { + case <-gate.started: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for XAck") + } + + handlers.Stop() + shutdownCtx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + started := time.Now() + require.ErrorIs(t, service.Close(shutdownCtx), context.DeadlineExceeded) + require.Less(t, time.Since(started), 500*time.Millisecond) + require.NoError(t, <-runDone) + select { + case <-gate.done: + default: + t.Fatal("shutdown returned before the blocked XAck stopped") + } + pending, err := publisherClient.XPending(context.Background(), topic, gate.group).Result() + require.NoError(t, err) + require.EqualValues(t, 1, pending.Count) +} + func startNotificationRedis(t *testing.T) string { t.Helper() redisServer, err := exec.LookPath("redis-server") diff --git a/backend/internal/service/notification_delivery_service.go b/backend/internal/service/notification_delivery_service.go index 49d3a20a..4ac972de 100644 --- a/backend/internal/service/notification_delivery_service.go +++ b/backend/internal/service/notification_delivery_service.go @@ -44,19 +44,18 @@ type NotificationDeliveryService struct { router *message.Router subscriber message.Subscriber handlers *lifecycle.HandlerGroup + cancelAcks context.CancelFunc + redisClient redis.UniversalClient } -type ackCompletingRedisClient struct{ redis.UniversalClient } +type ackCompletingRedisClient struct { + redis.UniversalClient + ctx context.Context +} -func (c ackCompletingRedisClient) XAck(ctx context.Context, stream, group string, ids ...string) *redis.IntCmd { - // Subscriber.Close cancels ctx while waiting; the in-flight XAck must finish first. - ackCtx := context.WithoutCancel(ctx) - if deadline, ok := ctx.Deadline(); ok { - var cancel context.CancelFunc - ackCtx, cancel = context.WithDeadline(ackCtx, deadline) - defer cancel() - } - return c.UniversalClient.XAck(ackCtx, stream, group, ids...) +func (c ackCompletingRedisClient) XAck(_ context.Context, stream, group string, ids ...string) *redis.IntCmd { + // Subscriber shutdown must not interrupt XACK before the application deadline. + return c.UniversalClient.XAck(c.ctx, stream, group, ids...) } // NewNotificationDeliveryService creates a delivery service and registers Watermill handlers. @@ -74,20 +73,23 @@ func NewNotificationDeliveryService( if redisClient == nil { return nil, fmt.Errorf("failed to create delivery subscriber: redis client is empty") } + ackCtx, cancelAcks := context.WithCancel(context.Background()) subscriber, err := redisstream.NewSubscriber( redisstream.SubscriberConfig{ - Client: ackCompletingRedisClient{redisClient}, + Client: ackCompletingRedisClient{UniversalClient: redisClient, ctx: ackCtx}, ConsumerGroup: "notif-delivery-" + watermill.NewUUID(), }, loggerAdapter, ) if err != nil { + cancelAcks() return nil, fmt.Errorf("failed to create delivery subscriber: %w", err) } router, err := message.NewRouter(message.RouterConfig{}, loggerAdapter) if err != nil { + cancelAcks() return nil, fmt.Errorf("failed to create delivery router: %w", err) } @@ -101,6 +103,8 @@ func NewNotificationDeliveryService( webhookSubRepo: webhookSubRepo, router: router, subscriber: subscriber, + cancelAcks: cancelAcks, + redisClient: redisClient, } s.registerHandlers() @@ -347,13 +351,26 @@ func (s *NotificationDeliveryService) Start(ctx context.Context) error { return s.router.Run(ctx) } -// Close shuts down the delivery router and subscriber. -func (s *NotificationDeliveryService) Close() error { +// Close shuts down the delivery router and subscriber within ctx's budget. +func (s *NotificationDeliveryService) Close(ctx context.Context) error { + cancelAcks := s.cancelAcks + if cancelAcks == nil { + cancelAcks = func() {} + } + stopDeadline := context.AfterFunc(ctx, func() { + cancelAcks() + if s.redisClient != nil { + _ = s.redisClient.Close() + } + }) + defer stopDeadline() + defer cancelAcks() + routerErr := s.router.Close() if routerErr != nil { applogger.L().Errorf("notif-delivery: failed to close router: %v", routerErr) } - return errors.Join(routerErr, s.subscriber.Close()) + return errors.Join(routerErr, s.subscriber.Close(), ctx.Err()) } // --- Watermill logger adapter for delivery service ---