From 1287b7702adb4779c772cdc75bfa6e46bdd3d9af Mon Sep 17 00:00:00 2001 From: Rogee Date: Mon, 24 Aug 2026 12:24:34 +0800 Subject: [PATCH] fix(HH-600): isolate notification Redis lifecycle (#157) Co-authored-by: Rogee --- backend/internal/app/bootstrap.go | 5 ++- .../notification_delivery_lifecycle_test.go | 38 ++++++++----------- .../service/notification_delivery_service.go | 1 + 3 files changed, 20 insertions(+), 24 deletions(-) diff --git a/backend/internal/app/bootstrap.go b/backend/internal/app/bootstrap.go index 5874a7ed..062754ca 100644 --- a/backend/internal/app/bootstrap.go +++ b/backend/internal/app/bootstrap.go @@ -47,6 +47,7 @@ import ( wspkg "github.com/gochat/gochat/internal/ws" "github.com/gochat/gochat/internal/wsevent" applogger "github.com/gochat/gochat/pkg/logger" + "github.com/redis/go-redis/v9" "github.com/spf13/viper" "gorm.io/gorm" ) @@ -591,6 +592,7 @@ func Bootstrap(env string) (*App, error) { _ = webhookSignatureService // Notification delivery pipeline — subscribes to EventBus topics and delivers via push/email/webhook (P4 M8) + notificationRedisClient := redis.NewClient(rdb.Options()) notificationDeliverySvc, err := service.NewNotificationDeliveryService( notificationService, pushDeliveryService, @@ -599,9 +601,10 @@ func Bootstrap(env string) (*App, error) { notificationPrefRepo, pushTokenRepo, webhookSubRepo, - rdb, + notificationRedisClient, ) if err != nil { + _ = notificationRedisClient.Close() return nil, fmt.Errorf("failed to init notification delivery service: %w", err) } diff --git a/backend/internal/service/notification_delivery_lifecycle_test.go b/backend/internal/service/notification_delivery_lifecycle_test.go index 8c51b7a4..8a8767fc 100644 --- a/backend/internal/service/notification_delivery_lifecycle_test.go +++ b/backend/internal/service/notification_delivery_lifecycle_test.go @@ -4,9 +4,9 @@ import ( "context" "fmt" "io" - "net" + "os" "os/exec" - "strconv" + "path/filepath" "sync" "testing" "time" @@ -143,9 +143,9 @@ func TestNotificationShutdownWaitsForRedisStreamXAck(t *testing.T) { release: make(chan struct{}), done: make(chan struct{}), } - subscriberClient := redis.NewClient(&redis.Options{Addr: addr, PoolSize: 32}) + subscriberClient := redis.NewClient(&redis.Options{Network: "unix", Addr: addr, PoolSize: 32}) subscriberClient.AddHook(gate) - publisherClient := redis.NewClient(&redis.Options{Addr: addr}) + publisherClient := redis.NewClient(&redis.Options{Network: "unix", Addr: addr}) t.Cleanup(func() { _ = subscriberClient.Close() _ = publisherClient.Close() @@ -183,11 +183,6 @@ func TestNotificationShutdownWaitsForRedisStreamXAck(t *testing.T) { handlers.Stop() closeDone := make(chan error, 1) go func() { closeDone <- service.Close(context.Background()) }() - select { - case err := <-closeDone: - t.Fatalf("router closed before XAck completed: %v", err) - case <-time.After(50 * time.Millisecond): - } require.NoError(t, publisherClient.Ping(context.Background()).Err()) releaseAck() @@ -213,9 +208,9 @@ func TestNotificationShutdownDeadlineLeavesRedisStreamMessagePending(t *testing. release: make(chan struct{}), done: make(chan struct{}), } - subscriberClient := redis.NewClient(&redis.Options{Addr: addr, PoolSize: 32}) + subscriberClient := redis.NewClient(&redis.Options{Network: "unix", Addr: addr, PoolSize: 32}) subscriberClient.AddHook(gate) - publisherClient := redis.NewClient(&redis.Options{Addr: addr}) + publisherClient := redis.NewClient(&redis.Options{Network: "unix", Addr: addr}) t.Cleanup(func() { _ = subscriberClient.Close() _ = publisherClient.Close() @@ -242,9 +237,7 @@ func TestNotificationShutdownDeadlineLeavesRedisStreamMessagePending(t *testing. 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: @@ -254,6 +247,7 @@ func TestNotificationShutdownDeadlineLeavesRedisStreamMessagePending(t *testing. pending, err := publisherClient.XPending(context.Background(), topic, gate.group).Result() require.NoError(t, err) require.EqualValues(t, 1, pending.Count) + require.NoError(t, publisherClient.Ping(context.Background()).Err()) } func startNotificationRedis(t *testing.T) string { @@ -262,18 +256,17 @@ func startNotificationRedis(t *testing.T) string { if err != nil { t.Skip("redis-server is required for this integration test") } - listener, err := net.Listen("tcp", "127.0.0.1:0") + dir, err := os.MkdirTemp("", "gochat-redis-") require.NoError(t, err) - port := listener.Addr().(*net.TCPAddr).Port - require.NoError(t, listener.Close()) + t.Cleanup(func() { _ = os.RemoveAll(dir) }) + socket := filepath.Join(dir, "redis.sock") cmd := exec.Command(redisServer, - "--bind", "127.0.0.1", - "--protected-mode", "no", - "--port", strconv.Itoa(port), + "--port", "0", + "--unixsocket", socket, + "--unixsocketperm", "700", "--save", "", "--appendonly", "no", - "--dir", t.TempDir(), ) cmd.Stdout = io.Discard cmd.Stderr = io.Discard @@ -283,11 +276,10 @@ func startNotificationRedis(t *testing.T) string { _ = cmd.Wait() }) - addr := net.JoinHostPort("127.0.0.1", strconv.Itoa(port)) - client := redis.NewClient(&redis.Options{Addr: addr}) + client := redis.NewClient(&redis.Options{Network: "unix", Addr: socket}) t.Cleanup(func() { _ = client.Close() }) require.Eventually(t, func() bool { return client.Ping(context.Background()).Err() == nil }, 5*time.Second, 10*time.Millisecond) - return addr + return socket } diff --git a/backend/internal/service/notification_delivery_service.go b/backend/internal/service/notification_delivery_service.go index 4ac972de..7cd235c8 100644 --- a/backend/internal/service/notification_delivery_service.go +++ b/backend/internal/service/notification_delivery_service.go @@ -59,6 +59,7 @@ func (c ackCompletingRedisClient) XAck(_ context.Context, stream, group string, } // NewNotificationDeliveryService creates a delivery service and registers Watermill handlers. +// The service owns redisClient and closes it during shutdown. func NewNotificationDeliveryService( notificationService *NotificationService, pushDeliveryService *PushDeliveryService,