Rogeeandrogee
798ea43c2f
* HH-442: isolate runtime processes and harden shutdown * HH-442: harden worker shutdown races * HH-442: gate dependency shutdown on active handlers --------- Co-authored-by: Rogee <rogee@ipao.vip>
94 lines
2.6 KiB
Go
94 lines
2.6 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/ThreeDotsLabs/watermill"
|
|
"github.com/ThreeDotsLabs/watermill/message"
|
|
"github.com/gochat/gochat/internal/lifecycle"
|
|
"github.com/stretchr/testify/require"
|
|
"gorm.io/driver/sqlite"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
type notificationLifecycleSubscriber struct {
|
|
ready chan struct{}
|
|
messages chan *message.Message
|
|
once sync.Once
|
|
ctx context.Context
|
|
}
|
|
|
|
func (s *notificationLifecycleSubscriber) Subscribe(ctx context.Context, _ string) (<-chan *message.Message, error) {
|
|
s.ctx = ctx
|
|
s.once.Do(func() { close(s.ready) })
|
|
return s.messages, nil
|
|
}
|
|
|
|
func (*notificationLifecycleSubscriber) Close() error { return nil }
|
|
|
|
func (s *notificationLifecycleSubscriber) Send(msg *message.Message) {
|
|
msg.SetContext(s.ctx)
|
|
s.messages <- msg
|
|
}
|
|
|
|
func TestNotificationCloseTimeoutWaitsForHandlerBeforeDependenciesClose(t *testing.T) {
|
|
db, err := gorm.Open(sqlite.Open("file:notification-shutdown-order?mode=memory&cache=shared"), &gorm.Config{})
|
|
require.NoError(t, err)
|
|
router, err := message.NewRouter(message.RouterConfig{CloseTimeout: 10 * time.Millisecond}, watermill.NopLogger{})
|
|
require.NoError(t, err)
|
|
subscriber := ¬ificationLifecycleSubscriber{
|
|
ready: make(chan struct{}),
|
|
messages: make(chan *message.Message),
|
|
}
|
|
handlers := lifecycle.NewHandlerGroup()
|
|
service := &NotificationDeliveryService{router: router, subscriber: subscriber, handlers: handlers}
|
|
|
|
started := make(chan struct{})
|
|
cancelled := make(chan struct{})
|
|
probeDependency := make(chan struct{})
|
|
handlerDBErr := make(chan error, 1)
|
|
router.AddNoPublisherHandler("blocked", "blocked", subscriber, service.track(func(msg *message.Message) error {
|
|
close(started)
|
|
<-msg.Context().Done()
|
|
close(cancelled)
|
|
<-probeDependency
|
|
var one int
|
|
handlerDBErr <- db.Raw("SELECT 1").Scan(&one).Error
|
|
return nil
|
|
}))
|
|
|
|
runDone := make(chan error, 1)
|
|
go func() { runDone <- service.Start(handlers.Context()) }()
|
|
<-subscriber.ready
|
|
subscriber.Send(message.NewMessage("blocked", nil))
|
|
<-started
|
|
|
|
handlers.Stop()
|
|
<-cancelled
|
|
require.ErrorContains(t, service.Close(), "router close timeout")
|
|
|
|
waitDone := make(chan struct{})
|
|
go func() {
|
|
handlers.Wait()
|
|
close(waitDone)
|
|
}()
|
|
select {
|
|
case <-waitDone:
|
|
t.Fatal("handler gate opened before the blocked handler completed")
|
|
case <-time.After(20 * time.Millisecond):
|
|
}
|
|
require.NoError(t, db.Exec("SELECT 1").Error)
|
|
|
|
close(probeDependency)
|
|
require.NoError(t, <-handlerDBErr)
|
|
<-waitDone
|
|
sqlDB, err := db.DB()
|
|
require.NoError(t, err)
|
|
require.NoError(t, sqlDB.Close())
|
|
require.Error(t, sqlDB.Ping())
|
|
require.NoError(t, <-runDone)
|
|
}
|