From 7efcb8d6ca1b189aa131109a3ed0881ff89fdaf2 Mon Sep 17 00:00:00 2001 From: silverwind Date: Mon, 27 Jul 2026 16:12:06 +0200 Subject: [PATCH] test(pubsub): stop racing the Redis SUBSCRIBE ack (#38661) `RedisBroker.Subscribe` returns before the server acks `SUBSCRIBE`, so a publish right after it can be dropped, making `TestRedisBroker/CrossBroker` fail intermittently on loaded CI runners ([example](https://github.com/go-gitea/gitea/actions/runs/30257188479/job/89948425118)). Each scenario now uses its own topic and waits for `PUBSUB NUMSUB` before publishing. `MemoryBroker` registers synchronously and skips the wait. --- services/pubsub/broker_test.go | 55 ++++++++++++++++++++++++---------- services/pubsub/redis.go | 6 ++-- services/pubsub/redis_test.go | 9 +++--- 3 files changed, 48 insertions(+), 22 deletions(-) diff --git a/services/pubsub/broker_test.go b/services/pubsub/broker_test.go index 6e6ea3d3f2..d2fc36e9dc 100644 --- a/services/pubsub/broker_test.go +++ b/services/pubsub/broker_test.go @@ -22,15 +22,16 @@ type newBrokerFunc func(t *testing.T) Broker func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Duration) { t.Run("PublishWithoutSubscribers", func(t *testing.T) { b := newBroker(t) - b.Publish("nobody", []byte("msg")) // must not block or panic + b.Publish(t.Name(), []byte("msg")) // must not block or panic }) t.Run("SubscribeReceivesPublished", func(t *testing.T) { b := newBroker(t) - ch, cancel := b.Subscribe("topic") + ch, cancel := b.Subscribe(t.Name()) defer cancel() + waitSubscribed(t, b, t.Name()) - b.Publish("topic", []byte("hello")) + b.Publish(t.Name(), []byte("hello")) assert.Equal(t, []byte("hello"), recvWithin(t, ch, recvTimeout)) }) @@ -39,12 +40,13 @@ func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Dur const n = 3 channels := make([]<-chan []byte, n) for i := range n { - ch, cancel := b.Subscribe("topic") + ch, cancel := b.Subscribe(t.Name()) defer cancel() channels[i] = ch } + waitSubscribed(t, b, t.Name()) // all local subscribers share one Redis SUBSCRIBE - b.Publish("topic", []byte("broadcast")) + b.Publish(t.Name(), []byte("broadcast")) for i, ch := range channels { assert.Equal(t, []byte("broadcast"), recvWithin(t, ch, recvTimeout), "subscriber %d", i) } @@ -52,30 +54,33 @@ func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Dur t.Run("TopicIsolation", func(t *testing.T) { b := newBroker(t) - chA, cancelA := b.Subscribe("a") + topicA, topicB := t.Name()+"-a", t.Name()+"-b" + chA, cancelA := b.Subscribe(topicA) defer cancelA() - chB, cancelB := b.Subscribe("b") + chB, cancelB := b.Subscribe(topicB) defer cancelB() + waitSubscribed(t, b, topicA) + waitSubscribed(t, b, topicB) - b.Publish("a", []byte("only-a")) + b.Publish(topicA, []byte("only-a")) assert.Equal(t, []byte("only-a"), recvWithin(t, chA, recvTimeout)) assertQuiet(t, chB, 100*time.Millisecond) // topic b must stay silent }) t.Run("CancelStopsDelivery", func(t *testing.T) { b := newBroker(t) - ch, cancel := b.Subscribe("topic") + ch, cancel := b.Subscribe(t.Name()) cancel() _, ok := <-ch assert.False(t, ok, "channel must be closed after cancel") - b.Publish("topic", []byte("after-cancel")) // must not panic or block + b.Publish(t.Name(), []byte("after-cancel")) // must not panic or block }) t.Run("CancelIsIdempotent", func(t *testing.T) { b := newBroker(t) - _, cancel := b.Subscribe("topic") + _, cancel := b.Subscribe(t.Name()) cancel() assert.NotPanics(t, cancel, "cancel must be safe to call more than once") }) @@ -84,17 +89,18 @@ func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Dur // the backend's own test file. t.Run("HasTopicSubscribers", func(t *testing.T) { b := newBroker(t) - _, cancel := b.Subscribe("topic") + _, cancel := b.Subscribe(t.Name()) defer cancel() - assert.True(t, b.HasTopicSubscribers("topic"), "must report subscribers while one is live") + assert.True(t, b.HasTopicSubscribers(t.Name()), "must report subscribers while one is live") }) t.Run("SlowSubscriberDropsWithoutBlocking", func(t *testing.T) { b := newBroker(t) - _, cancelSlow := b.Subscribe("topic") // never drained, buffer overflows + _, cancelSlow := b.Subscribe(t.Name()) // never drained, buffer overflows defer cancelSlow() - fast, cancelFast := b.Subscribe("topic") + fast, cancelFast := b.Subscribe(t.Name()) defer cancelFast() + waitSubscribed(t, b, t.Name()) // Drain fast concurrently so it keeps up while slow's buffer fills. got := make(chan struct{}, 1) @@ -122,7 +128,7 @@ func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Dur published := make(chan struct{}) go func() { for i := range n { - b.Publish("topic", []byte{byte(i)}) + b.Publish(t.Name(), []byte{byte(i)}) } close(published) }() @@ -141,6 +147,23 @@ func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Dur }) } +// waitSubscribed blocks until Redis registers the subscription, since +// RedisBroker.Subscribe returns before the server acks SUBSCRIBE and a publish +// right after it would race it. Topics are per-scenario, so any subscriber is +// this one. No-op for MemoryBroker, which registers synchronously. +func waitSubscribed(t *testing.T, b Broker, topic string) { + t.Helper() + rb, ok := b.(*RedisBroker) + if !ok { + return + } + channel := redisChannelForTopic(topic) + require.Eventually(t, func() bool { + res, err := rb.client.PubSubNumSub(t.Context(), channel).Result() + return err == nil && res[channel] > 0 + }, 5*time.Second, 10*time.Millisecond, "redis did not register a subscriber on %q", topic) +} + // recvWithin returns the next message or fails if none arrives before timeout. func recvWithin(t *testing.T, ch <-chan []byte, timeout time.Duration) []byte { t.Helper() diff --git a/services/pubsub/redis.go b/services/pubsub/redis.go index ce0c4305cd..eb7798298e 100644 --- a/services/pubsub/redis.go +++ b/services/pubsub/redis.go @@ -94,8 +94,10 @@ func (b *RedisBroker) Subscribe(topic string) (<-chan []byte, func()) { // other Subscribe/cancel calls aren't blocked on the network round-trip. // graceful.ShutdownContext so the reader loop dies cleanly on Gitea // shutdown even if every local subscriber has already cancelled. - // readLoop consumes the SUBSCRIBE ack; don't wait for it here, a direct - // ps.Receive blocks for its whole timeout instead of returning on the ack. + // readLoop consumes the SUBSCRIBE ack; don't wait for it here, that would put + // a Redis round-trip in the WebSocket handshake to close a sub-millisecond + // window in which a publish is missed - harmless, since a client receives + // nothing at all until its handshake completes. ctx, cancelCtx := context.WithCancel(graceful.GetManager().ShutdownContext()) ps := b.client.Subscribe(ctx, redisChannelForTopic(topic)) b.mu.Lock() diff --git a/services/pubsub/redis_test.go b/services/pubsub/redis_test.go index 99ff902a30..5da2a0c868 100644 --- a/services/pubsub/redis_test.go +++ b/services/pubsub/redis_test.go @@ -39,14 +39,14 @@ func TestRedisBroker(t *testing.T) { // state once the last local subscriber cancels. t.Run("CancelCleansTopicState", func(t *testing.T) { b := newBroker(t).(*RedisBroker) - ch, cancel := b.Subscribe("topic") + ch, cancel := b.Subscribe(t.Name()) cancel() _, ok := <-ch assert.False(t, ok, "channel must be closed after cancel") b.mu.RLock() - _, present := b.topics["topic"] + _, present := b.topics[t.Name()] b.mu.RUnlock() assert.False(t, present, "topic state must be removed after last subscriber cancels") }) @@ -56,10 +56,11 @@ func TestRedisBroker(t *testing.T) { t.Run("CrossBroker", func(t *testing.T) { publisher, subscriber := newBroker(t), newBroker(t) - ch, cancel := subscriber.Subscribe("topic") + ch, cancel := subscriber.Subscribe(t.Name()) defer cancel() + waitSubscribed(t, subscriber, t.Name()) - publisher.Publish("topic", []byte("cross-process")) + publisher.Publish(t.Name(), []byte("cross-process")) select { case msg := <-ch: