unit test repro of issue #2266

go-runtime/ftl/ftltest/testdata/go/pubsub/pubsub.go

@@ -11,6 +11,10 @@ import (
 
 //ftl:export
 var Topic = ftl.Topic[Event]("topic")
+
+//ftl:export
+var Topic2 = ftl.Topic[Event]("topic2")
+
 var subscription = ftl.Subscription(Topic, "subscription")
 
 //ftl:data
@@ -23,6 +27,11 @@ func PublishToTopicOne(ctx context.Context, event Event) error {
 	return Topic.Publish(ctx, event)
 }
 
+//ftl:verb
+func PublishToTopicTwo(ctx context.Context, event Event) error {
+	return Topic2.Publish(ctx, event)
+}
+
 //ftl:subscribe subscription
 func ErrorsAfterASecond(ctx context.Context, event Event) error {
 	time.Sleep(1 * time.Second)

go-runtime/ftl/ftltest/testdata/go/pubsub/pubsub_test.go

@@ -26,6 +26,17 @@ func TestSubscriberReturningErrors(t *testing.T) {
 	assert.Equal(t, count, len(ftltest.EventsForTopic(ctx, Topic)))
 }
 
+func TestPropagateEvents(t *testing.T) {
+	// Test that we can publish multiple events, which will take time to consume, and that we track each error
+	ctx := ftltest.Context(
+		ftltest.WithSubscriber(subscription, PublishToTopicTwo),
+	)
+	assert.NoError(t, PublishToTopicOne(ctx, Event{Value: "propagation-test"}))
+	ftltest.WaitForSubscriptionsToComplete(ctx)
+	assert.Equal(t, 1, len(ftltest.EventsForTopic(ctx, Topic)))
+	assert.Equal(t, 1, len(ftltest.EventsForTopic(ctx, Topic2)))
+}
+
 func TestMultipleMultipleFakeSubscribers(t *testing.T) {
 	// Test that multiple adhoc subscribers can be added for a subscription
 	count := 5