HH-463: wait for redisstream XAck during shutdown (#149)
Co-authored-by: Rogee <rogee@ipao.vip>
This commit is contained in:
Rogee
@@ -2,13 +2,21 @@ package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"os/exec"
|
||||
"strconv"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/ThreeDotsLabs/watermill"
|
||||
"github.com/ThreeDotsLabs/watermill-redisstream/pkg/redisstream"
|
||||
"github.com/ThreeDotsLabs/watermill/message"
|
||||
"github.com/gochat/gochat/internal/lifecycle"
|
||||
"github.com/gochat/gochat/internal/pubsub"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"github.com/stretchr/testify/require"
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
@@ -21,6 +29,35 @@ type notificationLifecycleSubscriber struct {
|
||||
ctx context.Context
|
||||
}
|
||||
|
||||
type notificationAckGate struct {
|
||||
topic string
|
||||
started chan struct{}
|
||||
release chan struct{}
|
||||
done chan struct{}
|
||||
group string
|
||||
startOnce sync.Once
|
||||
doneOnce sync.Once
|
||||
}
|
||||
|
||||
func (g *notificationAckGate) DialHook(next redis.DialHook) redis.DialHook { return next }
|
||||
func (g *notificationAckGate) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
|
||||
return next
|
||||
}
|
||||
func (g *notificationAckGate) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
|
||||
return func(ctx context.Context, cmd redis.Cmder) error {
|
||||
args := cmd.Args()
|
||||
if cmd.Name() != "xack" || len(args) < 3 || fmt.Sprint(args[1]) != g.topic {
|
||||
return next(ctx, cmd)
|
||||
}
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
func (s *notificationLifecycleSubscriber) Subscribe(ctx context.Context, _ string) (<-chan *message.Message, error) {
|
||||
s.ctx = ctx
|
||||
s.once.Do(func() { close(s.ready) })
|
||||
@@ -91,3 +128,109 @@ func TestNotificationCloseTimeoutWaitsForHandlerBeforeDependenciesClose(t *testi
|
||||
require.Error(t, sqlDB.Ping())
|
||||
require.NoError(t, <-runDone)
|
||||
}
|
||||
|
||||
func TestNotificationShutdownWaitsForRedisStreamXAck(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)
|
||||
|
||||
releaseAck := sync.OnceFunc(func() { close(gate.release) })
|
||||
t.Cleanup(func() {
|
||||
releaseAck()
|
||||
_ = service.Close()
|
||||
})
|
||||
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.Wait()
|
||||
pending, err := publisherClient.XPending(context.Background(), topic, gate.group).Result()
|
||||
require.NoError(t, err)
|
||||
require.EqualValues(t, 1, pending.Count)
|
||||
|
||||
handlers.Stop()
|
||||
closeDone := make(chan error, 1)
|
||||
go func() { closeDone <- service.Close() }()
|
||||
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()
|
||||
require.NoError(t, <-closeDone)
|
||||
require.NoError(t, <-runDone)
|
||||
select {
|
||||
case <-gate.done:
|
||||
default:
|
||||
t.Fatal("router closed without completing XAck")
|
||||
}
|
||||
pending, err = publisherClient.XPending(context.Background(), topic, gate.group).Result()
|
||||
require.NoError(t, err)
|
||||
require.Zero(t, pending.Count)
|
||||
require.Error(t, subscriberClient.Ping(context.Background()).Err())
|
||||
}
|
||||
|
||||
func startNotificationRedis(t *testing.T) string {
|
||||
t.Helper()
|
||||
redisServer, err := exec.LookPath("redis-server")
|
||||
if err != nil {
|
||||
t.Skip("redis-server is required for this integration test")
|
||||
}
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
require.NoError(t, err)
|
||||
port := listener.Addr().(*net.TCPAddr).Port
|
||||
require.NoError(t, listener.Close())
|
||||
|
||||
cmd := exec.Command(redisServer,
|
||||
"--bind", "127.0.0.1",
|
||||
"--protected-mode", "no",
|
||||
"--port", strconv.Itoa(port),
|
||||
"--save", "",
|
||||
"--appendonly", "no",
|
||||
"--dir", t.TempDir(),
|
||||
)
|
||||
cmd.Stdout = io.Discard
|
||||
cmd.Stderr = io.Discard
|
||||
require.NoError(t, cmd.Start())
|
||||
t.Cleanup(func() {
|
||||
_ = cmd.Process.Kill()
|
||||
_ = cmd.Wait()
|
||||
})
|
||||
|
||||
addr := net.JoinHostPort("127.0.0.1", strconv.Itoa(port))
|
||||
client := redis.NewClient(&redis.Options{Addr: addr})
|
||||
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
|
||||
}
|
||||
|
||||
@@ -46,6 +46,19 @@ type NotificationDeliveryService struct {
|
||||
handlers *lifecycle.HandlerGroup
|
||||
}
|
||||
|
||||
type ackCompletingRedisClient struct{ redis.UniversalClient }
|
||||
|
||||
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...)
|
||||
}
|
||||
|
||||
// NewNotificationDeliveryService creates a delivery service and registers Watermill handlers.
|
||||
func NewNotificationDeliveryService(
|
||||
notificationService *NotificationService,
|
||||
@@ -58,10 +71,13 @@ func NewNotificationDeliveryService(
|
||||
redisClient redis.UniversalClient,
|
||||
) (*NotificationDeliveryService, error) {
|
||||
loggerAdapter := &deliveryWatermillAdapter{}
|
||||
if redisClient == nil {
|
||||
return nil, fmt.Errorf("failed to create delivery subscriber: redis client is empty")
|
||||
}
|
||||
|
||||
subscriber, err := redisstream.NewSubscriber(
|
||||
redisstream.SubscriberConfig{
|
||||
Client: redisClient,
|
||||
Client: ackCompletingRedisClient{redisClient},
|
||||
ConsumerGroup: "notif-delivery-" + watermill.NewUUID(),
|
||||
},
|
||||
loggerAdapter,
|
||||
|
||||
Reference in New Issue
Block a user