Skip to content

Commit ded5362

Browse files
author
Morven Cao
committed
Fix PubSub race condition with sequential event processing.
Signed-off-by: Morven Cao <lcao@redhat.com>
1 parent e1fbdd7 commit ded5362

2 files changed

Lines changed: 191 additions & 7 deletions

File tree

‎pkg/cloudevents/generic/options/v2/pubsub/transport.go‎

Lines changed: 23 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package pubsub
33
import (
44
"context"
55
"fmt"
6+
"sync"
67

78
"cloud.google.com/go/pubsub/v2"
89
cloudevents "github.com/cloudevents/sdk-go/v2"
@@ -135,36 +136,51 @@ func (o *pubsubTransport) Send(ctx context.Context, evt cloudevents.Event) error
135136
func (o *pubsubTransport) Receive(ctx context.Context, fn options.ReceiveHandlerFn) error {
136137
errChan := make(chan error)
137138

139+
// Use a mutex to ensure sequential processing across both subscribers.
140+
// This prevents race conditions when concurrent events for the same
141+
// resource arrive on different subscriptions.
142+
var mu sync.Mutex
143+
138144
// start the subscriber for spec/status updates
139-
go o.receiveFromSubscriber(ctx, o.subscriber, fn, errChan)
145+
go o.receiveFromSubscriber(ctx, o.subscriber, fn, &mu, errChan)
140146

141147
// start the resync subscriber for resync events
142-
go o.receiveFromSubscriber(ctx, o.resyncSubscriber, fn, errChan)
148+
go o.receiveFromSubscriber(ctx, o.resyncSubscriber, fn, &mu, errChan)
143149

144-
// Return the error from either subscriber (including context cancellation).
150+
// Return the first error from either subscriber (including context cancellation).
145151
// We return errors directly instead of writing to the transport errorChan because
146152
// Pub/Sub client has internal retry logic for transient errors. Only non-retryable
147153
// errors or context cancellation will be returned here.
148154
return <-errChan
149155
}
150156

151157
// receiveFromSubscriber handles receiving messages from a subscriber.
158+
// It uses a mutex to ensure sequential processing across all subscribers.
152159
func (o *pubsubTransport) receiveFromSubscriber(
153160
ctx context.Context,
154161
subscriber *pubsub.Subscriber,
155162
fn options.ReceiveHandlerFn,
163+
mu *sync.Mutex,
156164
errChan chan<- error,
157165
) {
158166
logger := klog.FromContext(ctx)
159167
err := subscriber.Receive(ctx, func(ctx context.Context, msg *pubsub.Message) {
160168
evt, err := Decode(msg)
161169
if err != nil {
162-
// also send ACK on decode error since redelivery won't fix it.
170+
// ACK decode errors immediately since redelivery won't fix them.
163171
logger.Error(err, "failed to decode pubsub message")
164-
} else {
165-
fn(ctx, evt)
172+
msg.Ack()
173+
return
166174
}
167-
// send ACK after all receiver handlers complete.
175+
176+
// Lock to ensure sequential processing across both subscribers.
177+
// This prevents race conditions when concurrent events for the same
178+
// resource arrive on different subscriptions.
179+
mu.Lock()
180+
defer mu.Unlock()
181+
fn(ctx, evt)
182+
183+
// ACK after successful processing.
168184
msg.Ack()
169185
})
170186

‎test/integration/cloudevents/manifestworkclients_resync_pubsub_test.go‎

Lines changed: 168 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,168 @@
1+
package cloudevents
2+
3+
import (
4+
"context"
5+
"fmt"
6+
"time"
7+
8+
"github.com/onsi/ginkgo"
9+
"github.com/onsi/gomega"
10+
11+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
12+
"k8s.io/apimachinery/pkg/util/rand"
13+
14+
workv1 "open-cluster-management.io/api/work/v1"
15+
16+
"open-cluster-management.io/sdk-go/pkg/cloudevents/clients/common"
17+
"open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work"
18+
agentcodec "open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work/agent/codec"
19+
"open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work/payload"
20+
sourcecodec "open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work/source/codec"
21+
"open-cluster-management.io/sdk-go/pkg/cloudevents/generic"
22+
"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/clients"
23+
"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/v2/pubsub"
24+
"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
25+
"open-cluster-management.io/sdk-go/test/integration/cloudevents/agent"
26+
"open-cluster-management.io/sdk-go/test/integration/cloudevents/util"
27+
)
28+
29+
// emptyManifestWorkLister is a simple lister that returns empty list
30+
// Used for testing where we only need to publish events, not list resources
31+
type emptyManifestWorkLister struct{}
32+
33+
func (e *emptyManifestWorkLister) List(ctx context.Context, options types.ListOptions) ([]*workv1.ManifestWork, error) {
34+
return []*workv1.ManifestWork{}, nil
35+
}
36+
37+
// This test simulates the PubSub race condition by sending create_request and resync_response events concurrently
38+
// to a running work agent, to ensure pubsub transport can handle messages arriving on the different subscriptions simultaneously.
39+
var _ = ginkgo.Describe("ManifestWork Clients Test - Resync PubSub", func() {
40+
ginkgo.Context("Concurrent message delivery on pubsub", func() {
41+
var ctx context.Context
42+
var cancel context.CancelFunc
43+
44+
var sourceID string
45+
var clusterName string
46+
var workName string
47+
48+
var agentClientHolder *work.ClientHolder
49+
var sourceCloudEventsClient generic.CloudEventsClient[*workv1.ManifestWork]
50+
51+
ginkgo.BeforeEach(func() {
52+
ctx, cancel = context.WithCancel(context.Background())
53+
sourceID = fmt.Sprintf("mw-pubsub-race-%s", rand.String(5))
54+
clusterName = fmt.Sprintf("cluster-race-%s", rand.String(5))
55+
workName = "race-test-work"
56+
57+
// Setup PubSub topics and subscriptions
58+
gomega.Expect(setupTopicsAndSubscriptions(ctx, clusterName, sourceID)).ToNot(gomega.HaveOccurred())
59+
60+
// Start the agent and keep it running
61+
ginkgo.By("starting the agent")
62+
pubsubAgentOptions := util.NewPubSubAgentOptions(pubsubServer.Addr, pubsubProjectID, clusterName, true)
63+
var err error
64+
agentClientHolder, _, err = agent.StartWorkAgent(ctx, clusterName, pubsubAgentOptions, agentcodec.NewManifestBundleCodec())
65+
gomega.Expect(err).ToNot(gomega.HaveOccurred())
66+
67+
// wait for agent ready
68+
<-time.After(time.Second)
69+
70+
// Create a pure CloudEvents client (source) to send events directly
71+
ginkgo.By("creating pure cloudevents client to send events")
72+
pubsubSourceOptions := util.NewPubSubSourceOptions(pubsubServer.Addr, pubsubProjectID, sourceID, true)
73+
sourceOptions := pubsub.NewSourceOptions(pubsubSourceOptions, sourceID)
74+
75+
// Use a simple lister that returns empty list (we're not listing, just publishing)
76+
lister := &emptyManifestWorkLister{}
77+
hashGetter := func(obj *workv1.ManifestWork) (string, error) {
78+
return "", nil // not used for this test
79+
}
80+
81+
sourceCloudEventsClient, err = clients.NewCloudEventSourceClient(
82+
ctx,
83+
sourceOptions,
84+
lister,
85+
hashGetter,
86+
sourcecodec.NewManifestBundleCodec(),
87+
)
88+
gomega.Expect(err).ToNot(gomega.HaveOccurred())
89+
90+
// wait for source client ready
91+
<-time.After(time.Second)
92+
})
93+
94+
ginkgo.AfterEach(func() {
95+
cancel()
96+
})
97+
98+
ginkgo.It("should handle concurrent create_request and resync_response without race", func() {
99+
// Create the manifestwork that we'll send events for
100+
work := util.NewManifestWork(clusterName, workName, true)
101+
work.UID = "test-uid-123"
102+
work.Labels = map[string]string{
103+
common.CloudEventsOriginalSourceLabelKey: sourceID,
104+
}
105+
106+
// Define the event types we'll send concurrently
107+
createRequest := types.CloudEventsType{
108+
CloudEventsDataType: payload.ManifestBundleEventDataType,
109+
SubResource: types.SubResourceSpec,
110+
Action: types.CreateRequestAction,
111+
}
112+
113+
resyncResponse := types.CloudEventsType{
114+
CloudEventsDataType: payload.ManifestBundleEventDataType,
115+
SubResource: types.SubResourceSpec,
116+
Action: types.ResyncResponseAction,
117+
}
118+
119+
// THE RACE CONDITION TEST:
120+
// Send both create_request and resync_response events concurrently.
121+
// Both arrive on the agent's sourceevents and sourcebroadcast subscriptions from the source.
122+
// Without the sequential channel fix: two goroutines in receiveFromSubscriber
123+
// could process these concurrently, causing race conditions in the agent store
124+
// and potential resource version conflicts when controllers patch the work.
125+
// With the fix: events are funneled through a single channel for sequential processing.
126+
ginkgo.By("sending create_request and resync_response concurrently")
127+
128+
start := make(chan struct{})
129+
errChan := make(chan error, 2)
130+
131+
// Send create_request
132+
go func() {
133+
<-start
134+
err := sourceCloudEventsClient.Publish(ctx, createRequest, work)
135+
errChan <- err
136+
}()
137+
138+
// Send resync_response
139+
go func() {
140+
<-start
141+
err := sourceCloudEventsClient.Publish(ctx, resyncResponse, work.DeepCopy())
142+
errChan <- err
143+
}()
144+
145+
close(start)
146+
147+
// Wait for both publishes to complete
148+
for range 2 {
149+
err := <-errChan
150+
gomega.Expect(err).ToNot(gomega.HaveOccurred())
151+
}
152+
153+
// Verify the work is correctly applied in the agent despite concurrent message delivery
154+
ginkgo.By("verifying manifestwork is applied correctly")
155+
gomega.Eventually(func() error {
156+
retrievedWork, err := agentClientHolder.ManifestWorks(clusterName).Get(
157+
ctx, workName, metav1.GetOptions{})
158+
if err != nil {
159+
return err
160+
}
161+
if len(retrievedWork.Spec.Workload.Manifests) == 0 {
162+
return fmt.Errorf("work has no manifests")
163+
}
164+
return nil
165+
}, 10*time.Second, 500*time.Millisecond).Should(gomega.Succeed())
166+
})
167+
})
168+
})

0 commit comments

Comments
 (0)