Deflake flagged_driver_test by expanding mutex (#6062)

Author: evankanderson

internal/events/flagged_driver.go

@@ -80,15 +80,15 @@ func (f *flaggedDriver) Subscribe(ctx context.Context, topic string) (<-chan *me
 					base = nil
 					continue
 				}
-				out <- msg
 				f.readMessages.Add(ctx, 1, metric.WithAttributes(attribute.Bool("experiment", false)))
+				out <- msg
 			case msg, ok := <-experiment:
 				if !ok {
 					experiment = nil
 					continue
 				}
-				out <- msg
 				f.readMessages.Add(ctx, 1, metric.WithAttributes(attribute.Bool("experiment", true)))
+				out <- msg
 			case <-ctx.Done():
 				return
 			}
@@ -149,5 +149,7 @@ func makeFlaggedDriver(ctx context.Context, cfg *serverconfig.EventConfig, flagC
 	return ret, ret, closer, nil
 }
 
-var _ message.Publisher = (*flaggedDriver)(nil)
-var _ message.Subscriber = (*flaggedDriver)(nil)
+var (
+	_ message.Publisher  = (*flaggedDriver)(nil)
+	_ message.Subscriber = (*flaggedDriver)(nil)
+)

internal/events/flagged_driver_test.go

@@ -109,6 +109,9 @@ alternate_message_driver:
 			t.Parallel()
 			ctx := context.Background()
 
+			metricMutex.Lock() // Can run in parallel with non-metric-measuring tests
+			defer metricMutex.Unlock()
+
 			config := serverconfig.Config{
 				Events: serverconfig.EventConfig{
 					Driver: constants.FlaggedDriver,
@@ -153,9 +156,6 @@ alternate_message_driver:
 			})
 			<-eventer.Running()
 
-			metricMutex.Lock()
-			defer metricMutex.Unlock()
-
 			sendBefore := getMetric(t, reader, "events_published", tt.sendExperiment)
 			readBefore := getMetric(t, reader, "events_read", tt.sendExperiment)