test: EventPublisher publish accepted/rejected/timeout with observer assertions

publish.go

@@ -55,6 +55,7 @@ type pendingEntry struct {
 	eventJSON []byte
 	sent      bool
 	result    chan publishResult
+	mu        sync.Mutex
 	once      sync.Once
 	timer     *time.Timer
 }
@@ -138,18 +139,22 @@ 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()
+	if entry.sent {
+		return nil
+	}
+
 	err := p.envoy.Send([]byte(envelope.EncloseEvent(entry.eventJSON)))
 	if err != nil {
 		p.envoy.Observer().Record(p.envoy.PeerID(),
 			PublishSendFailed{EventID: entry.eventID, Err: err, At: time.Now()})
 		return err
 	}
-
-	p.mu.Lock()
 	entry.sent = true
-	p.mu.Unlock()
-
 	p.envoy.Observer().Record(p.envoy.PeerID(),
 		PublishDispatched{EventID: entry.eventID, At: time.Now()})
 	return nil
@@ -225,7 +230,9 @@ func (p *EventPublisher) handleEvents() {
 			case EventDisconnected:
 				p.mu.Lock()
 				for _, e := range p.pending {
+					e.mu.Lock()
 					e.sent = false
+					e.mu.Unlock()
 				}
 				p.mu.Unlock()
 			}

publish_test.go

@@ -9,7 +9,8 @@ import (
 
 func TestEventPublisher(t *testing.T) {
 	t.Run("publish accepted", func(t *testing.T) {
-		p, envoy := newMockEnvoy(t)
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
 		p.connect()
 
 		pub := NewEventPublisher(envoy)
@@ -50,10 +51,14 @@ func TestEventPublisher(t *testing.T) {
 				return false
 			}
 		}, "Publish did not return accepted result")
+
+		assert.Len(t, EventsOf[PublishDispatched](obs), 1)
+		assert.Len(t, EventsOf[PublishAccepted](obs), 1)
 	})
 
 	t.Run("publish rejected", func(t *testing.T) {
-		p, envoy := newMockEnvoy(t)
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
 		p.connect()
 
 		pub := NewEventPublisher(envoy)
@@ -94,11 +99,39 @@ func TestEventPublisher(t *testing.T) {
 				return false
 			}
 		}, "Publish did not return rejected result")
+
+		assert.Len(t, EventsOf[PublishDispatched](obs), 1)
+		assert.Len(t, EventsOf[PublishRejected](obs), 1)
 	})
 
 	t.Run("publish timeout", func(t *testing.T) {
-		// pub.Publish(id, json, 75*time.Millisecond) — do not feed OK
-		// assert err != nil (timeout error) after ~75ms
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
+		p.connect()
+
+		pub := NewEventPublisher(envoy)
+		t.Cleanup(pub.Close)
+
+		const id = "cccc3333"
+		eventJSON := []byte(`{"id":"cccc3333","kind":1}`)
+
+		ch := make(chan error, 1)
+		go func() {
+			_, _, err := pub.Publish(id, eventJSON, NegativeTestTimeout)
+			ch <- err
+		}()
+
+		Eventually(t, func() bool {
+			select {
+			case err := <-ch:
+				return err != nil
+			default:
+				return false
+			}
+		}, "Publish did not return a timeout error")
+
+		assert.Len(t, EventsOf[PublishDispatched](obs), 1)
+		assert.Len(t, EventsOf[PublishTimeout](obs), 1)
 	})
 
 	t.Run("concurrent correlation", func(t *testing.T) {