rogee/gochat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(HH-591): bound notification shutdown XACK (#154)

Browse Source 
Co-authored-by: Rogee <rogee@ipao.vip>
This commit is contained in:
Rogee
2026-08-24 11:36:24 +08:00
committed by GitHub
co-authored by rogee
parent 92ace6b041
commit 57f8a64fd8
6 changed files with 99 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/app/shutdown.go
+1 -1
View File
@@ -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 {
backend/internal/service/coverage21_test.go
+1 -1
View File
@@ -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) {
backend/internal/service/coverage29_test.go
+1 -1
View File
@@ -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 ----------
backend/internal/service/coverage37_test.go
+1 -1
View File
@@ -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 ===
backend/internal/service/notification_delivery_lifecycle_test.go
+64 -7
View File
@@ -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 := &notificationAckGate{
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")
backend/internal/service/notification_delivery_service.go
+31 -14
View File
@@ -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 ---