Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fix #275 #276

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Fix #275 #276

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
edc28c0
🧪 Add a test to show #275
Mar 23, 2022
3ed2575
🐛 Fix #275
Mar 23, 2022
f8f852e
🚧 Try to fix other pubsub test broken with previous fix
Mar 23, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 8 additions & 3 deletions pubsub/gochannel/pubsub.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,7 +154,7 @@ func (g *GoChannel) sendMessage(topic string, message *message.Message) (<-chan

wg.Add(1)
go func() {
subscriber.sendMessageToSubscriber(message, logFields)
subscriber.sendMessageToSubscriber(message, logFields, g.config.BlockPublishUntilSubscriberAck)
wg.Done()
}()
}
Expand Down Expand Up @@ -236,7 +236,7 @@ func (g *GoChannel) Subscribe(ctx context.Context, topic string) (<-chan *messag
msg := g.persistedMessages[topic][i]
logFields := watermill.LogFields{"message_uuid": msg.UUID, "topic": topic}

go s.sendMessageToSubscriber(msg, logFields)
go s.sendMessageToSubscriber(msg, logFields, g.config.BlockPublishUntilSubscriberAck)
}
}

Expand Down Expand Up @@ -341,7 +341,7 @@ func (s *subscriber) Close() {
close(s.outputChannel)
}

func (s *subscriber) sendMessageToSubscriber(msg *message.Message, logFields watermill.LogFields) {
func (s *subscriber) sendMessageToSubscriber(msg *message.Message, logFields watermill.LogFields, blockPublishUntilSubscriberAck bool) {
s.sending.Lock()
defer s.sending.Unlock()

Expand Down Expand Up @@ -370,6 +370,11 @@ SendToSubscriber:
return
}

if !blockPublishUntilSubscriberAck {
s.logger.Trace("Sent message to subscriber without ack", logFields)
return
}

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The select below isn't really blocking execution, as it's running in another goroutine. BlockPublishUntilSubscriberAck is checked in Publish and that's where it blocks.

Can you share more about your use case?

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@m110 the current implementation will cause deadlock on all subscribers even if one of them did Ack/Nack a message. However I believe only the subscriber that does not Ack/Nack should lose the message leaving the rest processing goes through. So the setting to block publishing will deadlock if one subscriber does not Ack and I don't think that the intention.

Copy link

@Tochemey Tochemey Dec 27, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My suggestion will be to timeout when the BlockPublishUntilSubscriberAck is set to avoid the deadlock. That means when an Ack/Nack is not received after some time for a given subscriber the message is lost because the timeout will kick in and the context will be cancelled.

select {
case <-msgToSend.Acked():
s.logger.Trace("Message acked", logFields)
Expand Down
92 changes: 79 additions & 13 deletions pubsub/gochannel/pubsub_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,72 @@ func TestPublishSubscribe_block_until_ack(t *testing.T) {
}
}

func TestPublishSubscribe_do_not_block_without_ack_required(t *testing.T) {
Comment on lines 110 to +111
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we use a small function name to make it more concise?

t.Helper()

messagesCount := 10
subscribersCount := 10

pubSub := gochannel.NewGoChannel(
gochannel.Config{
OutputChannelBuffer: int64(messagesCount),
Persistent: true,
},
watermill.NewStdLogger(true, false),
)

allSent := sync.WaitGroup{}
allSent.Add(messagesCount)
allReceived := sync.WaitGroup{}

sentMessages := message.Messages{}
subscriberReceivedCh := make(chan message.Messages, subscribersCount)
for i := 0; i < subscribersCount; i++ {
allReceived.Add(1)

go func(i int) {
allMsgReceived := make(message.Messages, 0)
msgs, err := pubSub.Subscribe(context.Background(), "topic")
require.NoError(t, err)

for received := range msgs {
allMsgReceived = append(allMsgReceived, received)
if len(allMsgReceived) >= len(sentMessages) {
break
}
}
subscriberReceivedCh <- allMsgReceived
allReceived.Done()
}(i)
}

go func() {
for i := 0; i < messagesCount; i++ {
msg := message.NewMessage(watermill.NewUUID(), nil)
sentMessages = append(sentMessages, msg)

go func() {
require.NoError(t, pubSub.Publish("topic", msg))
allSent.Done()
}()
}
}()

log.Println("waiting for all sent")
allSent.Wait()

log.Println("waiting for all received")
allReceived.Wait()

close(subscriberReceivedCh)

log.Println("asserting")

for subMsgs := range subscriberReceivedCh {
tests.AssertAllMessagesReceived(t, sentMessages, subMsgs)
}
}

func TestPublishSubscribe_race_condition_on_subscribe(t *testing.T) {
testsCount := 15
if testing.Short() {
Expand Down Expand Up @@ -228,19 +294,6 @@ func testPublishSubscribeSubRace(t *testing.T) {
allSent.Add(messagesCount)
allReceived := sync.WaitGroup{}

sentMessages := message.Messages{}
go func() {
for i := 0; i < messagesCount; i++ {
msg := message.NewMessage(watermill.NewUUID(), nil)
sentMessages = append(sentMessages, msg)

go func() {
require.NoError(t, pubSub.Publish("topic", msg))
allSent.Done()
}()
}
}()

subscriberReceivedCh := make(chan message.Messages, subscribersCount)
for i := 0; i < subscribersCount; i++ {
allReceived.Add(1)
Expand All @@ -256,6 +309,19 @@ func testPublishSubscribeSubRace(t *testing.T) {
}()
}

sentMessages := message.Messages{}
go func() {
for i := 0; i < messagesCount; i++ {
msg := message.NewMessage(watermill.NewUUID(), nil)
sentMessages = append(sentMessages, msg)

go func() {
require.NoError(t, pubSub.Publish("topic", msg))
allSent.Done()
}()
}
}()

log.Println("waiting for all sent")
allSent.Wait()

Expand Down
12 changes: 6 additions & 6 deletions pubsub/tests/test_pubsub.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -477,12 +477,12 @@ func TestResendOnError(
var publishedMessages message.Messages
allMessagesSent := make(chan struct{})

publishedMessages = PublishSimpleMessages(t, messagesToSend, pub, topicName)
close(allMessagesSent)

messages, err := sub.Subscribe(context.Background(), topicName)
require.NoError(t, err)

publishedMessages = PublishSimpleMessages(t, messagesToSend, pub, topicName)
close(allMessagesSent)

NackLoop:
for i := 0; i < nacksCount; i++ {
select {
Expand Down Expand Up @@ -524,6 +524,9 @@ func TestNoAck(
require.NoError(t, subscribeInitializer.SubscribeInitialize(topicName))
}

messages, err := sub.Subscribe(context.Background(), topicName)
require.NoError(t, err)

for i := 0; i < 2; i++ {
id := watermill.NewUUID()
log.Printf("sending %s", id)
Expand All @@ -534,9 +537,6 @@ func TestNoAck(
require.NoError(t, err)
}

messages, err := sub.Subscribe(context.Background(), topicName)
require.NoError(t, err)

receivedMessage := make(chan struct{})
unlockAck := make(chan struct{}, 1)
go func() {
Expand Down