From dda3bfd5b729ad9d4aff63dacae0d602be2ff58b Mon Sep 17 00:00:00 2001
From: Jay <jay@wisehodl.dev>
Date: Thu, 4 Jun 2026 12:07:03 -0400
Subject: [PATCH] cleanup: flakey tests and race conditions

---
 auth_test.go    | 9 ++++++++-
 publish.go      | 4 ++--
 publish_test.go | 4 +++-
 request_test.go | 6 ++----
 4 files changed, 15 insertions(+), 8 deletions(-)

diff --git a/auth_test.go b/auth_test.go
index df43952..9f2b65c 100644
--- a/auth_test.go
+++ b/auth_test.go
@@ -16,8 +16,11 @@ func TestAuthManager(t *testing.T) {
 		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
 
 		signed := []byte(`{"kind":22242}`)
+		var mu sync.Mutex
 		var calledWith string
 		callback := func(challenge string) ([]byte, error) {
+			mu.Lock()
+			defer mu.Unlock()
 			calledWith = challenge
 			return signed, nil
 		}
@@ -28,7 +31,11 @@ func TestAuthManager(t *testing.T) {
 		p.connect()
 		p.receive([]byte(envelope.EncloseAuthChallenge("abc")))
 
-		Eventually(t, func() bool { return calledWith == "abc" },
+		Eventually(t, func() bool {
+			mu.Lock()
+			defer mu.Unlock()
+			return calledWith == "abc"
+		},
 			"callback not called with challenge")
 
 		Eventually(t, func() bool {
diff --git a/publish.go b/publish.go
index 6dc77df..3a254ff 100644
--- a/publish.go
+++ b/publish.go
@@ -139,8 +139,6 @@ func (p *EventPublisher) Close() {
 	p.wg.Wait()
 }
 
-// trySend sends the EVENT envelope. It does not modify entry.sent — callers
-// are responsible for setting it under p.mu before and/or after this call.
 func (p *EventPublisher) trySend(entry *pendingEntry) error {
 	entry.mu.Lock()
 	defer entry.mu.Unlock()
@@ -212,9 +210,11 @@ func (p *EventPublisher) handleEvents() {
 				p.mu.Lock()
 				var toSend []*pendingEntry
 				for _, e := range p.pending {
+					e.mu.Lock()
 					if !e.sent {
 						toSend = append(toSend, e)
 					}
+					e.mu.Unlock()
 				}
 				p.mu.Unlock()
 
diff --git a/publish_test.go b/publish_test.go
index ce6e5a5..e667758 100644
--- a/publish_test.go
+++ b/publish_test.go
@@ -364,7 +364,9 @@ func TestEventPublisher(t *testing.T) {
 		}, "Publish did not return a send error")
 
 		assert.ErrorContains(t, err, "send failed")
-		assert.Len(t, EventsOf[PublishSendFailed](obs), 1)
+
+		// on connect event handler may race with Publish()
+		assert.GreaterOrEqual(t, len(EventsOf[PublishSendFailed](obs)), 1)
 	})
 
 	t.Run("disconnect then send failure on reconnect", func(t *testing.T) {
diff --git a/request_test.go b/request_test.go
index 5a54bc2..d24cbc1 100644
--- a/request_test.go
+++ b/request_test.go
@@ -333,15 +333,13 @@ func TestRequestManager_Stream(t *testing.T) {
 		p.connect()
 		Eventually(t, envoy.IsConnected, "envoy should be reconnected")
 
-		// prevent activate() race between Stream and on-connect handler
-		time.Sleep(20 * time.Millisecond)
-
 		filters := [][]byte{[]byte(`{}`)}
 		id, _, _, err := m.Stream(filters)
 		assert.NoError(t, err)
 
 		Eventually(t, func() bool {
-			return len(EventsOf[ReqSendFailed](obs)) == 1
+			// on connect event handler may race with Stream()
+			return len(EventsOf[ReqSendFailed](obs)) >= 1
 		}, "expected ReqSendFailed observable")
 		failed := EventsOf[ReqSendFailed](obs)
 		assert.Equal(t, id, failed[0].SubID)