refactor(events): replace time.Sleep with ReadySignal for deterministic test synchronization

Author: vfusco

internal/events/eventstest/contract_suite.go

@@ -15,7 +15,12 @@ import (
 
 // PublisherSubscriberFactory creates a matched Publisher and Subscriber
 // pair for testing. The factory is called once per test case.
-type PublisherSubscriberFactory func(t *testing.T) (events.Publisher, events.Subscriber)
+//
+// The returned ready channel is closed (or receives a value) when the
+// subscriber is connected and ready to receive notifications. For
+// memory.Bus, return a pre-closed channel. For PostgreSQL, use
+// SubscriberConfig.ReadySignal.
+type PublisherSubscriberFactory func(t *testing.T) (events.Publisher, events.Subscriber, <-chan struct{})
 
 // ContractSuite verifies that any Publisher+Subscriber implementation
 // satisfies the events library's behavioral contract. Run it against
@@ -25,18 +30,28 @@ type ContractSuite struct {
 	Factory PublisherSubscriberFactory
 }
 
-func (s *ContractSuite) TestSinglePublishSubscribe() {
-	pub, sub := s.Factory(s.T())
-	ch := sub.Subscribe(events.ChannelInputReceived)
+// startListening starts sub.Listen and waits for the subscriber to be ready.
+func (s *ContractSuite) startListening(sub events.Subscriber, ready <-chan struct{}) {
 	ctx := s.T().Context()
 	go sub.Listen(ctx) //nolint:errcheck
-	time.Sleep(200 * time.Millisecond)
+	select {
+	case <-ready:
+	case <-time.After(5 * time.Second):
+		s.Fail("subscriber did not become ready")
+	}
+}
+
+func (s *ContractSuite) TestSinglePublishSubscribe() {
+	pub, sub, ready := s.Factory(s.T())
+	ch := sub.Subscribe(events.ChannelInputReceived)
+	s.startListening(sub, ready)
 
 	expected := events.Notification{
 		Channel:       events.ChannelInputReceived,
 		ApplicationID: 42,
 		EpochIndex:    7,
 	}
+	ctx := s.T().Context()
 	pub.Publish(ctx, expected)
 
 	select {
@@ -49,12 +64,11 @@ func (s *ContractSuite) TestSinglePublishSubscribe() {
 }
 
 func (s *ContractSuite) TestChannelIsolation() {
-	pub, sub := s.Factory(s.T())
+	pub, sub, ready := s.Factory(s.T())
 	ch := sub.Subscribe(events.ChannelInputReceived)
-	ctx := s.T().Context()
-	go sub.Listen(ctx) //nolint:errcheck
-	time.Sleep(200 * time.Millisecond)
+	s.startListening(sub, ready)
 
+	ctx := s.T().Context()
 	// Publish on a different channel.
 	pub.Publish(ctx, events.Notification{
 		Channel:       events.ChannelClaimComputed,
@@ -75,12 +89,11 @@ func (s *ContractSuite) TestChannelIsolation() {
 }
 
 func (s *ContractSuite) TestBufferOverflowDropsWithoutBlocking() {
-	pub, sub := s.Factory(s.T())
+	pub, sub, ready := s.Factory(s.T())
 	ch := sub.Subscribe(events.ChannelInputReceived)
-	ctx := s.T().Context()
-	go sub.Listen(ctx) //nolint:errcheck
-	time.Sleep(200 * time.Millisecond)
+	s.startListening(sub, ready)
 
+	ctx := s.T().Context()
 	// Publish more than buffer size (64) without reading.
 	for i := range 100 {
 		pub.Publish(ctx, events.Notification{
@@ -99,13 +112,12 @@ func (s *ContractSuite) TestBufferOverflowDropsWithoutBlocking() {
 }
 
 func (s *ContractSuite) TestMultipleSubscriptions() {
-	pub, sub := s.Factory(s.T())
+	pub, sub, ready := s.Factory(s.T())
 	ch1 := sub.Subscribe(events.ChannelInputReceived)
 	ch2 := sub.Subscribe(events.ChannelClaimComputed)
-	ctx := s.T().Context()
-	go sub.Listen(ctx) //nolint:errcheck
-	time.Sleep(200 * time.Millisecond)
+	s.startListening(sub, ready)
 
+	ctx := s.T().Context()
 	pub.Publish(ctx, events.Notification{
 		Channel:       events.ChannelInputReceived,
 		ApplicationID: 1,
@@ -130,13 +142,12 @@ func (s *ContractSuite) TestMultipleSubscriptions() {
 }
 
 func (s *ContractSuite) TestSubscribeWithFilterByAppID() {
-	pub, sub := s.Factory(s.T())
+	pub, sub, ready := s.Factory(s.T())
 	filter := events.SubscriptionFilter{ApplicationIDs: []int64{42}}
 	ch := sub.SubscribeWithFilter(filter, events.ChannelInputReceived)
-	ctx := s.T().Context()
-	go sub.Listen(ctx) //nolint:errcheck
-	time.Sleep(200 * time.Millisecond)
+	s.startListening(sub, ready)
 
+	ctx := s.T().Context()
 	// Publish for a non-matching app.
 	pub.Publish(ctx, events.Notification{
 		Channel:       events.ChannelInputReceived,

internal/events/eventstest/property_suite.go

@@ -24,9 +24,9 @@ type PropertySuite struct {
 	suite.Suite
 	Factory PublisherSubscriberFactory
 
-	// SettleTime is the time to wait after subscribe+listen for connection
-	// readiness, and after publish for delivery. For memory.Bus this is 0;
-	// for PG LISTEN/NOTIFY this should be ~200ms.
+	// SettleTime is the time to wait after publish for async delivery.
+	// For memory.Bus this is 0; for PG LISTEN/NOTIFY this should be ~200ms.
+	// Connection readiness uses the factory's ready channel instead.
 	SettleTime time.Duration
 }
 
@@ -37,12 +37,15 @@ func (s *PropertySuite) settle() {
 	}
 }
 
-// startListening starts sub.Listen in a background goroutine and waits
-// for the connection to be established. Must be called after Subscribe.
-func (s *PropertySuite) startListening(sub events.Subscriber) {
+// startListening starts sub.Listen and waits for the subscriber to be ready.
+func (s *PropertySuite) startListening(sub events.Subscriber, ready <-chan struct{}) {
 	ctx := s.T().Context()
 	go sub.Listen(ctx) //nolint:errcheck
-	s.settle()
+	select {
+	case <-ready:
+	case <-time.After(5 * time.Second):
+		s.Fail("subscriber did not become ready")
+	}
 }
 
 // TestP1NoWorkLossUnderEventLoss verifies that for any sequence of published
@@ -53,9 +56,9 @@ func (s *PropertySuite) TestP1NoWorkLossUnderEventLoss() {
 		n := rand.IntN(20) + 1 //nolint:gosec
 		db := make(map[int64]bool)
 
-		pub, sub := s.Factory(s.T())
+		pub, sub, ready := s.Factory(s.T())
 		ch := sub.Subscribe(events.ChannelInputReceived)
-		s.startListening(sub)
+		s.startListening(sub, ready)
 
 		ctx := s.T().Context()
 		for i := range n {
@@ -106,9 +109,9 @@ func (s *PropertySuite) TestP5ChannelIsolation() {
 	allChannels := events.AllChannels()
 
 	for _, subChannel := range allChannels {
-		pub, sub := s.Factory(s.T())
+		pub, sub, ready := s.Factory(s.T())
 		ch := sub.Subscribe(subChannel)
-		s.startListening(sub)
+		s.startListening(sub, ready)
 
 		ctx := s.T().Context()
 		for _, pubChannel := range allChannels {
@@ -142,14 +145,14 @@ func (s *PropertySuite) TestP6PipelineEventualDelivery() {
 	const syncInterval = 3
 
 	for range 10 {
-		pub, sub := s.Factory(s.T())
+		pub, sub, ready := s.Factory(s.T())
 		ctx := s.T().Context()
 
 		// Set up a 3-stage pipeline: input -> processed -> computed.
 		stage1Ch := sub.Subscribe(events.ChannelInputReceived)
 		stage2Ch := sub.Subscribe(events.ChannelInputsProcessed)
 		stage3Ch := sub.Subscribe(events.ChannelClaimComputed)
-		s.startListening(sub)
+		s.startListening(sub, ready)
 
 		// Simulated DB: each stage writes its output here.
 		stage1Done := false

internal/events/memory/bus_test.go

@@ -202,9 +202,11 @@ func TestBusDefaultBufferSize(t *testing.T) {
 
 func TestMemoryBusContract(t *testing.T) {
 	suite.Run(t, &eventstest.ContractSuite{
-		Factory: func(_ *testing.T) (events.Publisher, events.Subscriber) {
+		Factory: func(_ *testing.T) (events.Publisher, events.Subscriber, <-chan struct{}) {
 			bus := NewBus(64)
-			return bus, bus
+			ready := make(chan struct{})
+			close(ready) // memory bus is immediately ready
+			return bus, bus, ready
 		},
 	})
 }

internal/events/postgres/property_test.go

@@ -26,17 +26,18 @@ func TestPropertySuitePostgres(t *testing.T) {
 	connStr := getTestConnString(t)
 
 	suite.Run(t, &eventstest.PropertySuite{
-		Factory: func(t *testing.T) (events.Publisher, events.Subscriber) {
+		Factory: func(t *testing.T) (events.Publisher, events.Subscriber, <-chan struct{}) {
 			ctx := context.Background()
 			pool, err := pgxpool.New(ctx, connStr)
 			require.NoError(t, err)
 			t.Cleanup(pool.Close)
 
+			ready := make(chan struct{}, 1)
 			pub := NewPublisher(pool, slog.Default())
-			sub := NewSubscriber(connStr, slog.Default(), nil)
+			sub := NewSubscriber(connStr, slog.Default(), &SubscriberConfig{ReadySignal: ready})
 			t.Cleanup(func() { _ = sub.Close() })
 
-			return pub, sub
+			return pub, sub, ready
 		},
 		SettleTime: 200 * time.Millisecond,
 	})

internal/events/postgres/subscriber.go

@@ -46,6 +46,11 @@ type SubscriberConfig struct {
 	// Default: 64. Increase for systems with many applications where a single
 	// block could generate notifications for all of them.
 	BufferSize int
+
+	// ReadySignal, if non-nil, receives a value each time the subscriber
+	// successfully connects and issues LISTEN commands. This enables tests
+	// to wait for readiness without time.Sleep.
+	ReadySignal chan<- struct{}
 }
 
 // subscription holds the delivery channel, subscribed channels, and optional
@@ -64,6 +69,7 @@ type Subscriber struct {
 	logger           *slog.Logger
 	bufferSize       int
 	heartbeatTimeout time.Duration
+	readySignal      chan<- struct{}
 
 	mu            sync.Mutex
 	channels      []events.Channel
@@ -74,19 +80,22 @@ type Subscriber struct {
 func NewSubscriber(connString string, logger *slog.Logger, cfg *SubscriberConfig) *Subscriber {
 	heartbeat := defaultHeartbeatTimeout
 	bufSize := defaultBufferSize
+	var readySig chan<- struct{}
 	if cfg != nil {
 		if cfg.HeartbeatTimeout > 0 {
 			heartbeat = cfg.HeartbeatTimeout
 		}
 		if cfg.BufferSize > 0 {
 			bufSize = cfg.BufferSize
 		}
+		readySig = cfg.ReadySignal
 	}
 	return &Subscriber{
 		connString:       connString,
 		logger:           logger,
 		bufferSize:       bufSize,
 		heartbeatTimeout: heartbeat,
+		readySignal:      readySig,
 	}
 }
 
@@ -197,6 +206,12 @@ func (s *Subscriber) listenLoop(
 	}
 
 	s.logger.Info("Event listener connected", "channels", channels)
+	if s.readySignal != nil {
+		select {
+		case s.readySignal <- struct{}{}:
+		default:
+		}
+	}
 
 	for {
 		waitCtx, cancel := context.WithTimeout(ctx, s.heartbeatTimeout)

internal/events/postgres/subscriber_test.go

@@ -361,17 +361,18 @@ func TestPostgresContract(t *testing.T) {
 	connStr := getTestConnString(t)
 
 	suite.Run(t, &eventstest.ContractSuite{
-		Factory: func(_ *testing.T) (events.Publisher, events.Subscriber) {
+		Factory: func(_ *testing.T) (events.Publisher, events.Subscriber, <-chan struct{}) {
 			ctx := context.Background()
 			pool, err := pgxpool.New(ctx, connStr)
 			require.NoError(t, err)
 			t.Cleanup(pool.Close)
 
+			ready := make(chan struct{}, 1)
 			pub := NewPublisher(pool, slog.Default())
-			sub := NewSubscriber(connStr, slog.Default(), nil)
+			sub := NewSubscriber(connStr, slog.Default(), &SubscriberConfig{ReadySignal: ready})
 			t.Cleanup(func() { _ = sub.Close() })
 
-			return pub, sub
+			return pub, sub, ready
 		},
 	})
 }

internal/events/property_test.go

@@ -25,10 +25,12 @@ import (
 // in internal/events/postgres/property_test.go.
 func TestPropertySuiteMemory(t *testing.T) {
 	suite.Run(t, &eventstest.PropertySuite{
-		Factory: func(t *testing.T) (events.Publisher, events.Subscriber) {
+		Factory: func(t *testing.T) (events.Publisher, events.Subscriber, <-chan struct{}) {
 			bus := memory.NewBus(64)
 			t.Cleanup(func() { _ = bus.Close() })
-			return bus, bus
+			ready := make(chan struct{})
+			close(ready) // memory bus is immediately ready
+			return bus, bus, ready
 		},
 	})
 }