cleanup: flakey tests and race conditions

AGENTS.md

@@ -0,0 +1,4 @@
+# go-mana-prism
+
+## Build
+- This repo is part of a Go workspace at `../go.work`; resolve local packages from there before searching elsewhere.

auth.go

@@ -0,0 +1,106 @@
+package prism
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"git.wisehodl.dev/jay/go-roots-ws"
+)
+
+type ChallengeReceived struct {
+	Challenge string
+	At        time.Time
+}
+
+type AuthResponseSent struct {
+	At time.Time
+}
+
+type AuthResponseFailed struct {
+	Err error
+	At  time.Time
+}
+
+type AuthManager struct {
+	envoy     *Envoy
+	respond   func(challenge string) ([]byte, error)
+	challenge string
+
+	inbox  <-chan InboxMessage
+	events <-chan OutboundPoolEvent
+
+	ctx    context.Context
+	cancel context.CancelFunc
+	mu     sync.Mutex
+	wg     sync.WaitGroup
+}
+
+func NewAuthManager(e *Envoy, respond func(string) ([]byte, error)) *AuthManager {
+	ctx, cancel := context.WithCancel(e.Context())
+	m := &AuthManager{
+		envoy:   e,
+		respond: respond,
+		inbox:   e.SubscribeInbox([]string{"AUTH"}),
+		events:  e.SubscribeEvents(),
+		ctx:     ctx,
+		cancel:  cancel,
+	}
+	m.wg.Go(m.routeInbox)
+	m.wg.Go(m.handleEvents)
+	return m
+}
+
+func (m *AuthManager) Close() {
+	m.cancel()
+	m.wg.Wait()
+}
+
+func (m *AuthManager) routeInbox() {
+	for {
+		select {
+		case <-m.ctx.Done():
+			return
+		case msg := <-m.inbox:
+			challenge, err := envelope.FindAuthChallenge(msg.Data)
+			if err != nil {
+				continue
+			}
+
+			m.mu.Lock()
+			m.challenge = challenge
+			m.mu.Unlock()
+
+			now := time.Now()
+			m.envoy.Observer().Record(msg.ID, ChallengeReceived{Challenge: challenge, At: now})
+
+			signed, err := m.respond(challenge)
+			if err != nil {
+				m.envoy.Observer().Record(msg.ID, AuthResponseFailed{Err: err, At: time.Now()})
+				continue
+			}
+
+			if err := m.envoy.Send([]byte(envelope.EncloseAuthResponse(signed))); err != nil {
+				m.envoy.Observer().Record(msg.ID, AuthResponseFailed{Err: err, At: time.Now()})
+				continue
+			}
+
+			m.envoy.Observer().Record(msg.ID, AuthResponseSent{At: time.Now()})
+		}
+	}
+}
+
+func (m *AuthManager) handleEvents() {
+	for {
+		select {
+		case <-m.ctx.Done():
+			return
+		case ev := <-m.events:
+			if ev.Kind == EventDisconnected {
+				m.mu.Lock()
+				m.challenge = ""
+				m.mu.Unlock()
+			}
+		}
+	}
+}

auth_test.go

@@ -0,0 +1,208 @@
+package prism
+
+import (
+	"bytes"
+	"errors"
+	"sync"
+	"testing"
+
+	"git.wisehodl.dev/jay/go-roots-ws"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAuthManager(t *testing.T) {
+	t.Run("challenge triggers callback and send", func(t *testing.T) {
+		obs := &mockObserver{}
+		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
+		}
+
+		m := NewAuthManager(envoy, callback)
+		t.Cleanup(m.Close)
+
+		p.connect()
+		p.receive([]byte(envelope.EncloseAuthChallenge("abc")))
+
+		Eventually(t, func() bool {
+			mu.Lock()
+			defer mu.Unlock()
+			return calledWith == "abc"
+		},
+			"callback not called with challenge")
+
+		Eventually(t, func() bool {
+			select {
+			case got := <-p.sent:
+				return bytes.Contains(got, signed)
+			default:
+				return false
+			}
+		}, "AUTH response envelope not sent")
+
+		assert.Len(t, EventsOf[ChallengeReceived](obs), 1)
+		assert.Len(t, EventsOf[AuthResponseSent](obs), 1)
+	})
+
+	t.Run("callback error recorded", func(t *testing.T) {
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
+
+		callback := func(challenge string) ([]byte, error) {
+			return nil, errors.New("signing failed")
+		}
+
+		m := NewAuthManager(envoy, callback)
+		t.Cleanup(m.Close)
+
+		p.connect()
+		p.receive([]byte(envelope.EncloseAuthChallenge("abc")))
+
+		Eventually(t, func() bool {
+			return len(EventsOf[AuthResponseFailed](obs)) == 1
+		}, "AuthResponseFailed not recorded")
+
+		Never(t, func() bool {
+			select {
+			case <-p.sent:
+				return true
+			default:
+				return false
+			}
+		}, "no AUTH envelope should be sent on callback error")
+
+		assert.Len(t, EventsOf[ChallengeReceived](obs), 1)
+	})
+
+	t.Run("send error recorded", func(t *testing.T) {
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
+
+		callback := func(challenge string) ([]byte, error) {
+			return []byte(`{"kind":22242}`), nil
+		}
+
+		m := NewAuthManager(envoy, callback)
+		t.Cleanup(m.Close)
+
+		p.setSendError(errors.New("send failed"))
+		p.connect()
+		p.receive([]byte(envelope.EncloseAuthChallenge("abc")))
+
+		Eventually(t, func() bool {
+			return len(EventsOf[AuthResponseFailed](obs)) == 1
+		}, "AuthResponseFailed not recorded")
+	})
+
+	t.Run("disconnect resets state", func(t *testing.T) {
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
+
+		var mu sync.Mutex
+		var calls []string
+		callback := func(challenge string) ([]byte, error) {
+			mu.Lock()
+			calls = append(calls, challenge)
+			mu.Unlock()
+			return []byte(`{"kind":22242}`), nil
+		}
+
+		m := NewAuthManager(envoy, callback)
+		t.Cleanup(m.Close)
+
+		p.connect()
+		p.receive([]byte(envelope.EncloseAuthChallenge("abc")))
+		Eventually(t, func() bool {
+			mu.Lock()
+			defer mu.Unlock()
+			return len(calls) == 1 && calls[0] == "abc"
+		}, "first callback not fired")
+
+		p.disconnect()
+		// drain the sent channel so p.sent doesn't block the second send
+		select {
+		case <-p.sent:
+		default:
+		}
+
+		p.connect()
+		p.receive([]byte(envelope.EncloseAuthChallenge("def")))
+		Eventually(t, func() bool {
+			mu.Lock()
+			defer mu.Unlock()
+			return len(calls) == 2 && calls[1] == "def"
+		}, "second callback not fired with new challenge")
+	})
+
+	t.Run("replacement challenge", func(t *testing.T) {
+		p, envoy := newMockEnvoy(t)
+
+		var mu sync.Mutex
+		var calls []string
+		callback := func(challenge string) ([]byte, error) {
+			mu.Lock()
+			calls = append(calls, challenge)
+			mu.Unlock()
+			return []byte(`{"kind":22242}`), nil
+		}
+
+		m := NewAuthManager(envoy, callback)
+		t.Cleanup(m.Close)
+
+		p.connect()
+		p.receive([]byte(envelope.EncloseAuthChallenge("abc")))
+		p.receive([]byte(envelope.EncloseAuthChallenge("def")))
+
+		Eventually(t, func() bool {
+			mu.Lock()
+			defer mu.Unlock()
+			return len(calls) == 2 && calls[1] == "def"
+		}, "callback not called twice with both challenges")
+
+		assert.GreaterOrEqual(t, len(p.sent), 2)
+	})
+
+	t.Run("malformed AUTH ignored", func(t *testing.T) {
+		p, envoy := newMockEnvoy(t)
+
+		called := false
+		callback := func(challenge string) ([]byte, error) {
+			called = true
+			return []byte(`{"kind":22242}`), nil
+		}
+
+		m := NewAuthManager(envoy, callback)
+		t.Cleanup(m.Close)
+
+		p.connect()
+		// EncloseAuthResponse wraps a JSON object as the second element — not a string
+		p.receive([]byte(envelope.EncloseAuthResponse([]byte(`{}`))))
+
+		Never(t, func() bool { return called }, "callback should not be invoked for malformed AUTH")
+	})
+
+	t.Run("close cleans up", func(t *testing.T) {
+		p, envoy := newMockEnvoy(t)
+
+		called := false
+		callback := func(challenge string) ([]byte, error) {
+			called = true
+			return []byte(`{"kind":22242}`), nil
+		}
+
+		m := NewAuthManager(envoy, callback)
+		p.connect()
+		m.Close()
+
+		p.receive([]byte(envelope.EncloseAuthChallenge("abc")))
+
+		Never(t, func() bool { return called }, "callback should not fire after Close")
+	})
+}

embassy.go

@@ -17,7 +17,7 @@ import (
 // ----------------------------------------------------------------------------
 
 type EmbassyPlugin struct {
-	Connect func(url string) error
+	Connect func(url string, opts ...honeybee.ConnectOption) error
 	Remove  func(url string) error
 	Send    func(url string, data []byte) error
 	Events  <-chan honeybee.PoolEvent
@@ -124,9 +124,8 @@ func NewEmbassy(
 		e.logger = slog.New(cfg.handler).With(slog.Any("component", comp))
 	}
 
-	e.wg.Add(2)
+	e.wg.Go(e.routeEvents)
-	go e.routeEvents()
+	e.wg.Go(e.routeInbox)
-	go e.routeInbox()
 
 	return e
 }
@@ -237,8 +236,6 @@ func (e *Embassy) unsubscribeInboxLock(url string) {
 }
 
 func (e *Embassy) routeEvents() {
-	defer e.wg.Done()
-
 	for {
 		select {
 		case <-e.ctx.Done():
@@ -273,8 +270,6 @@ func (e *Embassy) routeEvents() {
 }
 
 func (e *Embassy) routeInbox() {
-	defer e.wg.Done()
-
 	for {
 		select {
 		case <-e.ctx.Done():
@@ -355,17 +350,16 @@ func newEnvoy(
 		ctx:               ctx,
 		cancel:            cancel,
 		observer:          observer,
-		handler:           handler,
 	}
 
 	if handler != nil {
 		comp := component.FromContext(ctx)
-		e.logger = slog.New(handler).With(slog.Any("component", comp)).With("peer", url)
+		e.handler = handler.WithAttrs([]slog.Attr{slog.String("peer", url)})
+		e.logger = slog.New(e.handler).With(slog.Any("component", comp))
 	}
 
-	e.wg.Add(2)
+	e.wg.Go(e.publishEvents)
-	go e.publishEvents()
+	e.wg.Go(e.routeInbox)
-	go e.routeInbox()
 
 	return e
 }
@@ -440,8 +434,6 @@ func (e *Envoy) SubscribeInbox(labels []string) <-chan InboxMessage {
 }
 
 func (e *Envoy) publishEvents() {
-	defer e.wg.Done()
-
 	for {
 		select {
 		case <-e.ctx.Done():
@@ -473,8 +465,6 @@ func (e *Envoy) publishEvents() {
 }
 
 func (e *Envoy) routeInbox() {
-	defer e.wg.Done()
-
 	for {
 		select {
 		case <-e.ctx.Done():

go.mod

@@ -3,9 +3,9 @@ module git.wisehodl.dev/jay/go-mana-prism
 go 1.25.0
 
 require (
-	git.wisehodl.dev/jay/go-honeybee v0.4.0
+	git.wisehodl.dev/jay/go-honeybee v0.5.1
-	git.wisehodl.dev/jay/go-mana-component v0.1.0
+	git.wisehodl.dev/jay/go-mana-component v0.2.0
-	git.wisehodl.dev/jay/go-roots-ws v0.1.0
+	git.wisehodl.dev/jay/go-roots-ws v0.2.0
 	github.com/stretchr/testify v1.11.1
 )
 

go.sum

@@ -1,9 +1,15 @@
-git.wisehodl.dev/jay/go-honeybee v0.4.0 h1:+aA2GbCQzgB2mzx0fcEmg319T2KmdHKsSxF8aXAbG7E=
+git.wisehodl.dev/jay/go-honeybee v0.5.0 h1:tAhXMLJnxMMhWJstA1f3OYVsdQf0rstl5/AaJVzuCWk=
-git.wisehodl.dev/jay/go-honeybee v0.4.0/go.mod h1:91H66sH5t1BHERGMihybeL1CDMu6vJgGuxjkUxfaqi4=
+git.wisehodl.dev/jay/go-honeybee v0.5.0/go.mod h1:91H66sH5t1BHERGMihybeL1CDMu6vJgGuxjkUxfaqi4=
+git.wisehodl.dev/jay/go-honeybee v0.5.1 h1:PuJeIOysCj8FV2GKfi2UL9xITOYJhkftFze8IdOuJZ0=
+git.wisehodl.dev/jay/go-honeybee v0.5.1/go.mod h1:lOz4/P3ultn/PgFz1JiaetxMXvff48GiLK0b+s1rvUc=
 git.wisehodl.dev/jay/go-mana-component v0.1.0 h1:wWYN5MzC9Hq3tEt4z7FjrwNuQz3rZY3RWAmgmNE8EZE=
 git.wisehodl.dev/jay/go-mana-component v0.1.0/go.mod h1:r2ZaTjKzwV5JJfC5boikxtjAKusPrzlJU/7qul0EUqA=
+git.wisehodl.dev/jay/go-mana-component v0.2.0 h1:uuk8YeFH/yQCtcpotkgQERDwFGY85/rNCUgk6hCGhW4=
+git.wisehodl.dev/jay/go-mana-component v0.2.0/go.mod h1:r2ZaTjKzwV5JJfC5boikxtjAKusPrzlJU/7qul0EUqA=
 git.wisehodl.dev/jay/go-roots-ws v0.1.0 h1:p1veCkpOmL26N//Qz7ekJOYj1Ck30ai4OKq9dxLjodk=
 git.wisehodl.dev/jay/go-roots-ws v0.1.0/go.mod h1:ANQOOP13lHs2uQwYhrSQGAlL7+zR6QvbLzNPmNBJssQ=
+git.wisehodl.dev/jay/go-roots-ws v0.2.0 h1:XzlrGu0//09SrTWWd6A8b/bGck2EZnJq2JOA+UJql28=
+git.wisehodl.dev/jay/go-roots-ws v0.2.0/go.mod h1:ANQOOP13lHs2uQwYhrSQGAlL7+zR6QvbLzNPmNBJssQ=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=

helpers_test.go

@@ -5,11 +5,14 @@ import (
 	"git.wisehodl.dev/jay/go-honeybee"
 	"git.wisehodl.dev/jay/go-mana-component"
 	"github.com/stretchr/testify/assert"
+	"sync"
 	"testing"
 	"time"
 )
 
+// ----------------------------------------------------------------------------
 // Async Helpers
+// ----------------------------------------------------------------------------
 
 const (
 	TestTimeout         = 2 * time.Second
@@ -27,6 +30,10 @@ func Never(t *testing.T, condition func() bool, msg string) {
 	assert.Never(t, condition, NegativeTestTimeout, TestTick, msg)
 }
 
+// ----------------------------------------------------------------------------
+// Mock Pool
+// ----------------------------------------------------------------------------
+
 type mockPool struct {
 	plugin  EmbassyPlugin
 	ctx     context.Context
@@ -36,40 +43,46 @@ type mockPool struct {
 	events  chan honeybee.PoolEvent
 	inbox   chan honeybee.InboxMessage
 	sent    chan []byte
+	sendMu  sync.Mutex
+	sendErr error
 }
 
-// Mock Pool
-
 func newMockPool(t *testing.T) *mockPool {
 	t.Helper()
 
-	ctx := component.MustNew(context.Background(), "prism", "test")
+	p := &mockPool{}
-	url := "wss://test"
 
-	added := make(chan struct{})
+	p.ctx = component.MustNew(context.Background(), "prism", "test")
-	removed := make(chan struct{})
+	p.url = "wss://test"
-	events := make(chan honeybee.PoolEvent, 16)
-	inbox := make(chan honeybee.InboxMessage, 16)
-	sent := make(chan []byte, 16)
 
-	plugin := EmbassyPlugin{
+	p.added = make(chan struct{})
-		Connect: func(url string) error { close(added); return nil },
+	p.removed = make(chan struct{})
-		Remove:  func(url string) error { close(removed); return nil },
+	p.events = make(chan honeybee.PoolEvent, 16)
-		Send:    func(_ string, data []byte) error { sent <- data; return nil },
+	p.inbox = make(chan honeybee.InboxMessage, 16)
-		Events:  events,
+	p.sent = make(chan []byte, 16)
-		Inbox:   inbox,
+	sendFunc := func(_ string, data []byte) error {
+		p.sendMu.Lock()
+		err := p.sendErr
+		p.sendMu.Unlock()
+		if err != nil {
+			return err
+		}
+		p.sent <- data
+		return nil
 	}
 
-	return &mockPool{
+	p.plugin = EmbassyPlugin{
-		plugin:  plugin,
+		Connect: func(url string, opts ...honeybee.ConnectOption) error {
-		ctx:     ctx,
+			close(p.added)
-		url:     url,
+			return nil
-		added:   added,
+		},
-		removed: removed,
+		Remove: func(url string) error { close(p.removed); return nil },
-		events:  events,
+		Send:   sendFunc,
-		inbox:   inbox,
+		Events: p.events,
-		sent:    sent,
+		Inbox:  p.inbox,
 	}
+
+	return p
 }
 
 func (p *mockPool) connect() {
@@ -90,15 +103,56 @@ func (p *mockPool) receive(data []byte) {
 	}
 }
 
-// MockEnvoy
+func (p *mockPool) setSendError(err error) {
+	p.sendMu.Lock()
+	defer p.sendMu.Unlock()
+	p.sendErr = err
+}
 
-func newMockEnvoy(t *testing.T) (*mockPool, *Envoy) {
+// ----------------------------------------------------------------------------
+// MockEnvoy
+// ----------------------------------------------------------------------------
+
+func newMockEnvoy(t *testing.T, opts ...EmbassyOption) (*mockPool, *Envoy) {
 	t.Helper()
 
 	p := newMockPool(t)
-	emb := NewEmbassy(p.ctx, p.plugin)
+	emb := NewEmbassy(p.ctx, p.plugin, opts...)
 	err := emb.Dispatch(p.url)
 	assert.NoError(t, err)
 	envoy := emb.Call(p.url)
 	return p, envoy
 }
+
+// ----------------------------------------------------------------------------
+// MockObserver
+// ----------------------------------------------------------------------------
+
+type observerRecord struct {
+	peerID string
+	event  any
+}
+
+type mockObserver struct {
+	mu      sync.Mutex
+	records []observerRecord
+}
+
+func (o *mockObserver) Record(peerID string, event any) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.records = append(o.records, observerRecord{peerID, event})
+}
+
+// EventsOf returns all recorded events of type T.
+func EventsOf[T any](o *mockObserver) []T {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	var out []T
+	for _, r := range o.records {
+		if v, ok := r.event.(T); ok {
+			out = append(out, v)
+		}
+	}
+	return out
+}

notice.go

@@ -0,0 +1,75 @@
+package prism
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"git.wisehodl.dev/jay/go-roots-ws"
+)
+
+type NoticeReceived struct {
+	Message string
+	At      time.Time
+}
+
+type Notice struct {
+	PeerID    string
+	Message   string
+	Timestamp time.Time
+}
+
+type NoticeHandler struct {
+	envoy   *Envoy
+	inbox   <-chan InboxMessage
+	notices chan Notice
+
+	ctx    context.Context
+	cancel context.CancelFunc
+	wg     sync.WaitGroup
+}
+
+func NewNoticeHandler(e *Envoy) *NoticeHandler {
+	ctx, cancel := context.WithCancel(e.Context())
+	h := &NoticeHandler{
+		envoy:   e,
+		inbox:   e.SubscribeInbox([]string{"NOTICE"}),
+		notices: make(chan Notice, 16),
+		ctx:     ctx,
+		cancel:  cancel,
+	}
+	h.wg.Go(h.route)
+	return h
+}
+
+func (h *NoticeHandler) Notices() <-chan Notice { return h.notices }
+
+func (h *NoticeHandler) Close() {
+	h.cancel()
+	h.wg.Wait()
+	close(h.notices)
+}
+
+func (h *NoticeHandler) route() {
+	for {
+		select {
+		case <-h.ctx.Done():
+			return
+		case msg := <-h.inbox:
+			message, err := envelope.FindNotice(msg.Data)
+			if err != nil {
+				continue
+			}
+			now := time.Now()
+			h.notices <- Notice{
+				PeerID:    msg.ID,
+				Message:   message,
+				Timestamp: now,
+			}
+			h.envoy.Observer().Record(msg.ID, NoticeReceived{
+				Message: message,
+				At:      now,
+			})
+		}
+	}
+}

notice_test.go

@@ -0,0 +1,75 @@
+package prism
+
+import (
+	"testing"
+
+	"git.wisehodl.dev/jay/go-roots-ws"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNoticeHandler(t *testing.T) {
+	t.Run("notice delivered", func(t *testing.T) {
+		p, envoy := newMockEnvoy(t)
+		h := NewNoticeHandler(envoy)
+		t.Cleanup(h.Close)
+
+		p.receive([]byte(envelope.EncloseNotice("hello")))
+
+		var got Notice
+		Eventually(t, func() bool {
+			select {
+			case got = <-h.Notices():
+				return true
+			default:
+				return false
+			}
+		}, "notice not delivered")
+
+		assert.Equal(t, "hello", got.Message)
+		assert.Equal(t, p.url, got.PeerID)
+		assert.False(t, got.Timestamp.IsZero())
+	})
+
+	t.Run("malformed ignored", func(t *testing.T) {
+		p, envoy := newMockEnvoy(t)
+		h := NewNoticeHandler(envoy)
+		t.Cleanup(h.Close)
+
+		p.receive([]byte("not json"))
+
+		Never(t, func() bool {
+			select {
+			case <-h.Notices():
+				return true
+			default:
+				return false
+			}
+		}, "notices channel should remain empty")
+	})
+
+	t.Run("observer notified", func(t *testing.T) {
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
+		h := NewNoticeHandler(envoy)
+		t.Cleanup(h.Close)
+
+		p.receive([]byte(envelope.EncloseNotice("hello")))
+
+		Eventually(t, func() bool {
+			events := EventsOf[NoticeReceived](obs)
+			return len(events) == 1 &&
+				events[0].Message == "hello" &&
+				!events[0].At.IsZero()
+		}, "NoticeReceived observable not recorded")
+	})
+
+	t.Run("close cleans up", func(t *testing.T) {
+		_, envoy := newMockEnvoy(t)
+		h := NewNoticeHandler(envoy)
+
+		h.Close()
+
+		_, ok := <-h.Notices()
+		assert.False(t, ok, "notices channel should be closed")
+	})
+}

publish.go

@@ -0,0 +1,245 @@
+package prism
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
+
+	"git.wisehodl.dev/jay/go-roots-ws"
+)
+
+// ----------------------------------------------------------------------------
+// Observables
+// ----------------------------------------------------------------------------
+
+type PublishDispatched struct {
+	EventID string
+	At      time.Time
+}
+
+type PublishAccepted struct {
+	EventID string
+	At      time.Time
+}
+
+type PublishRejected struct {
+	EventID string
+	Message string
+	At      time.Time
+}
+
+type PublishTimeout struct {
+	EventID string
+	At      time.Time
+}
+
+type PublishSendFailed struct {
+	EventID string
+	Err     error
+	At      time.Time
+}
+
+// ----------------------------------------------------------------------------
+// Types
+// ----------------------------------------------------------------------------
+
+type publishResult struct {
+	accepted bool
+	message  string
+	err      error
+}
+
+type pendingEntry struct {
+	eventID   string
+	eventJSON []byte
+	sent      bool
+	result    chan publishResult
+	mu        sync.Mutex
+	once      sync.Once
+	timer     *time.Timer
+}
+
+// ----------------------------------------------------------------------------
+// Event Publisher
+// ----------------------------------------------------------------------------
+
+type EventPublisher struct {
+	envoy   *Envoy
+	pending map[string]*pendingEntry // event id -> pending entry
+
+	inbox  <-chan InboxMessage
+	events <-chan OutboundPoolEvent
+
+	ctx    context.Context
+	cancel context.CancelFunc
+	mu     sync.Mutex
+	wg     sync.WaitGroup
+}
+
+func NewEventPublisher(e *Envoy) *EventPublisher {
+	ctx, cancel := context.WithCancel(e.Context())
+	p := &EventPublisher{
+		envoy:   e,
+		pending: make(map[string]*pendingEntry),
+		inbox:   e.SubscribeInbox([]string{"OK"}),
+		events:  e.SubscribeEvents(),
+		ctx:     ctx,
+		cancel:  cancel,
+	}
+	p.wg.Go(p.routeInbox)
+	p.wg.Go(p.handleEvents)
+	return p
+}
+
+func (p *EventPublisher) Publish(eventID string, eventJSON []byte, timeout time.Duration) (bool, string, error) {
+	entry := &pendingEntry{
+		eventID:   eventID,
+		eventJSON: eventJSON,
+		result:    make(chan publishResult, 1),
+	}
+
+	entry.timer = time.AfterFunc(timeout, func() {
+		p.mu.Lock()
+		p.deregister(eventID)
+		p.mu.Unlock()
+		p.envoy.Observer().Record(p.envoy.PeerID(),
+			PublishTimeout{EventID: eventID, At: time.Now()})
+		p.deliver(entry, publishResult{err: errors.New("publish timeout")})
+	})
+
+	p.mu.Lock()
+	p.pending[eventID] = entry
+	connected := p.envoy.IsConnected()
+	p.mu.Unlock()
+
+	if connected {
+		if err := p.trySend(entry); err != nil {
+			p.mu.Lock()
+			p.deregister(eventID)
+			p.mu.Unlock()
+			p.deliver(entry, publishResult{err: err})
+		}
+	}
+
+	r := <-entry.result
+	return r.accepted, r.message, r.err
+}
+
+func (p *EventPublisher) Close() {
+	p.cancel()
+
+	p.mu.Lock()
+	for id, e := range p.pending {
+		p.deregister(id)
+		p.deliver(e, publishResult{err: errors.New("publisher closed")})
+	}
+	p.mu.Unlock()
+
+	p.wg.Wait()
+}
+
+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
+	}
+	entry.sent = true
+	p.envoy.Observer().Record(p.envoy.PeerID(),
+		PublishDispatched{EventID: entry.eventID, At: time.Now()})
+	return nil
+}
+
+func (p *EventPublisher) deliver(entry *pendingEntry, result publishResult) {
+	entry.once.Do(func() {
+		entry.timer.Stop()
+		entry.result <- result
+	})
+}
+
+func (p *EventPublisher) routeInbox() {
+	for {
+		select {
+		case <-p.ctx.Done():
+			return
+		case msg := <-p.inbox:
+			eventID, accepted, message, err := envelope.FindOK(msg.Data)
+			if err != nil {
+				continue
+			}
+
+			p.mu.Lock()
+			entry, ok := p.pending[eventID]
+			if ok {
+				p.deregister(eventID)
+			}
+			p.mu.Unlock()
+
+			if !ok {
+				continue
+			}
+
+			if accepted {
+				p.envoy.Observer().Record(msg.ID,
+					PublishAccepted{EventID: eventID, At: time.Now()})
+			} else {
+				p.envoy.Observer().Record(msg.ID,
+					PublishRejected{EventID: eventID, Message: message, At: time.Now()})
+			}
+			p.deliver(entry, publishResult{accepted: accepted, message: message})
+		}
+	}
+}
+
+func (p *EventPublisher) handleEvents() {
+	for {
+		select {
+		case <-p.ctx.Done():
+			return
+		case ev := <-p.events:
+			switch ev.Kind {
+			case EventConnected:
+				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()
+
+				for _, e := range toSend {
+					if err := p.trySend(e); err != nil {
+						p.mu.Lock()
+						p.deregister(e.eventID)
+						p.mu.Unlock()
+						p.deliver(e, publishResult{err: err})
+					}
+				}
+
+			case EventDisconnected:
+				p.mu.Lock()
+				for _, e := range p.pending {
+					e.mu.Lock()
+					e.sent = false
+					e.mu.Unlock()
+				}
+				p.mu.Unlock()
+			}
+		}
+	}
+}
+
+func (p *EventPublisher) deregister(eventID string) {
+	delete(p.pending, eventID)
+}

publish_test.go

@@ -0,0 +1,474 @@
+package prism
+
+import (
+	"errors"
+	"testing"
+
+	"git.wisehodl.dev/jay/go-roots-ws"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestEventPublisher(t *testing.T) {
+	t.Run("publish accepted", func(t *testing.T) {
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
+		p.connect()
+
+		pub := NewEventPublisher(envoy)
+		t.Cleanup(pub.Close)
+
+		const id = "aaaa1111"
+		eventJSON := []byte(`{"id":"aaaa1111","kind":1}`)
+
+		type result struct {
+			accepted bool
+			message  string
+			err      error
+		}
+		ch := make(chan result, 1)
+		go func() {
+			accepted, msg, err := pub.Publish(id, eventJSON, TestTimeout)
+			ch <- result{accepted, msg, err}
+		}()
+
+		var sent []byte
+		Eventually(t, func() bool {
+			select {
+			case sent = <-p.sent:
+				return true
+			default:
+				return false
+			}
+		}, "EVENT envelope not sent")
+		assert.Contains(t, string(sent), id)
+
+		p.receive([]byte(envelope.EncloseOK(id, true, "")))
+
+		Eventually(t, func() bool {
+			select {
+			case r := <-ch:
+				return r.accepted && r.message == "" && r.err == nil
+			default:
+				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) {
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
+		p.connect()
+
+		pub := NewEventPublisher(envoy)
+		t.Cleanup(pub.Close)
+
+		const id = "bbbb2222"
+		eventJSON := []byte(`{"id":"bbbb2222","kind":1}`)
+
+		type result struct {
+			accepted bool
+			message  string
+			err      error
+		}
+		ch := make(chan result, 1)
+		go func() {
+			accepted, msg, err := pub.Publish(id, eventJSON, TestTimeout)
+			ch <- result{accepted, msg, err}
+		}()
+
+		Eventually(t, func() bool {
+			select {
+			case <-p.sent:
+				return true
+			default:
+				return false
+			}
+		}, "EVENT envelope not sent")
+
+		p.receive([]byte(envelope.EncloseOK(id, false, "rejected: not on whitelist")))
+
+		Eventually(t, func() bool {
+			select {
+			case r := <-ch:
+				return !r.accepted &&
+					r.message == "rejected: not on whitelist" &&
+					r.err == nil
+			default:
+				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) {
+		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
+		}()
+
+		var err error
+		Eventually(t, func() bool {
+			select {
+			case err = <-ch:
+				return err != nil
+			default:
+				return false
+			}
+		}, "Publish did not return a timeout error")
+
+		assert.ErrorContains(t, err, "publish timeout")
+		assert.Len(t, EventsOf[PublishDispatched](obs), 1)
+		assert.Len(t, EventsOf[PublishTimeout](obs), 1)
+	})
+
+	t.Run("concurrent correlation", func(t *testing.T) {
+		p, envoy := newMockEnvoy(t)
+		p.connect()
+
+		pub := NewEventPublisher(envoy)
+		t.Cleanup(pub.Close)
+
+		type result struct {
+			accepted bool
+			message  string
+			err      error
+		}
+
+		chA := make(chan result, 1)
+		chB := make(chan result, 1)
+
+		go func() {
+			accepted, msg, err := pub.Publish(
+				"aaaa", []byte(`{"id":"aaaa"}`), TestTimeout)
+			chA <- result{accepted, msg, err}
+		}()
+		go func() {
+			accepted, msg, err := pub.Publish(
+				"bbbb", []byte(`{"id":"bbbb"}`), TestTimeout)
+			chB <- result{accepted, msg, err}
+		}()
+
+		// drain both EVENT envelopes before feeding OKs
+		Eventually(t, func() bool { return len(p.sent) == 2 },
+			"expected two EVENT envelopes")
+		<-p.sent
+		<-p.sent
+
+		// feed OK for B first, then A
+		p.receive([]byte(envelope.EncloseOK("bbbb", true, "b-msg")))
+		p.receive([]byte(envelope.EncloseOK("aaaa", false, "a-msg")))
+
+		Eventually(t, func() bool {
+			select {
+			case r := <-chA:
+				return !r.accepted &&
+					r.message == "a-msg" &&
+					r.err == nil
+			default:
+				return false
+			}
+		}, "caller A did not receive its result")
+
+		Eventually(t, func() bool {
+			select {
+			case r := <-chB:
+				return r.accepted && r.message == "b-msg" && r.err == nil
+			default:
+				return false
+			}
+		}, "caller B did not receive its result")
+	})
+
+	t.Run("publish while disconnected times out", func(t *testing.T) {
+		p, envoy := newMockEnvoy(t)
+		// do not connect
+
+		pub := NewEventPublisher(envoy)
+		t.Cleanup(pub.Close)
+
+		ch := make(chan error, 1)
+		go func() {
+			_, _, err := pub.Publish(
+				"dddd4444", []byte(`{"id":"dddd4444"}`), NegativeTestTimeout)
+			ch <- err
+		}()
+
+		Eventually(t, func() bool {
+			select {
+			case err := <-ch:
+				return err != nil
+			default:
+				return false
+			}
+		}, "Publish did not return a timeout error")
+
+		Never(t, func() bool {
+			select {
+			case <-p.sent:
+				return true
+			default:
+				return false
+			}
+		}, "EVENT envelope should not be sent while disconnected")
+	})
+
+	t.Run("held event sent on connect", func(t *testing.T) {
+		p, envoy := newMockEnvoy(t)
+		// do not connect
+
+		pub := NewEventPublisher(envoy)
+		t.Cleanup(pub.Close)
+
+		const id = "eeee5555"
+		eventJSON := []byte(`{"id":"eeee5555"}`)
+
+		type result struct {
+			accepted bool
+			message  string
+			err      error
+		}
+		ch := make(chan result, 1)
+		go func() {
+			accepted, msg, err := pub.Publish(id, eventJSON, TestTimeout)
+			ch <- result{accepted, msg, err}
+		}()
+
+		Never(t, func() bool {
+			select {
+			case <-p.sent:
+				return true
+			default:
+				return false
+			}
+		}, "EVENT envelope should not be sent before connect")
+
+		p.connect()
+
+		Eventually(t, func() bool {
+			select {
+			case <-p.sent:
+				return true
+			default:
+				return false
+			}
+		}, "EVENT envelope not sent after connect")
+
+		p.receive([]byte(envelope.EncloseOK(id, true, "")))
+
+		Eventually(t, func() bool {
+			select {
+			case r := <-ch:
+				return r.accepted && r.message == "" && r.err == nil
+			default:
+				return false
+			}
+		}, "Publish did not return accepted result after connect")
+	})
+
+	t.Run("resend after disconnect if OK not received in time", func(t *testing.T) {
+		p, envoy := newMockEnvoy(t)
+		p.connect()
+
+		pub := NewEventPublisher(envoy)
+		t.Cleanup(pub.Close)
+
+		const id = "ffff6666"
+		eventJSON := []byte(`{"id":"ffff6666"}`)
+
+		type result struct {
+			accepted bool
+			err      error
+		}
+		ch := make(chan result, 1)
+		go func() {
+			accepted, _, err := pub.Publish(id, eventJSON, TestTimeout)
+			ch <- result{accepted, err}
+		}()
+
+		// first EVENT sent
+		Eventually(t, func() bool {
+			select {
+			case <-p.sent:
+				return true
+			default:
+				return false
+			}
+		}, "first EVENT envelope not sent")
+
+		p.disconnect()
+		p.connect()
+
+		// second EVENT sent after reconnect
+		// because OK was not yet received
+		Eventually(t, func() bool {
+			select {
+			case <-p.sent:
+				return true
+			default:
+				return false
+			}
+		}, "second EVENT envelope not sent after reconnect")
+
+		p.receive([]byte(envelope.EncloseOK(id, true, "")))
+
+		Eventually(t, func() bool {
+			select {
+			case r := <-ch:
+				return r.accepted && r.err == nil
+			default:
+				return false
+			}
+		}, "Publish did not return accepted result after resend")
+	})
+
+	t.Run("send failure delivers error", func(t *testing.T) {
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
+		p.connect()
+		p.setSendError(errors.New("send failed"))
+
+		pub := NewEventPublisher(envoy)
+		t.Cleanup(pub.Close)
+
+		ch := make(chan error, 1)
+		go func() {
+			_, _, err := pub.Publish(
+				"gggg7777", []byte(`{"id":"gggg7777"}`), TestTimeout)
+			ch <- err
+		}()
+
+		var err error
+		Eventually(t, func() bool {
+			select {
+			case err = <-ch:
+				return err != nil
+			default:
+				return false
+			}
+		}, "Publish did not return a send error")
+
+		assert.ErrorContains(t, err, "send failed")
+
+		// 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) {
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
+		p.connect()
+
+		pub := NewEventPublisher(envoy)
+		t.Cleanup(pub.Close)
+
+		ch := make(chan error, 1)
+		go func() {
+			_, _, err := pub.Publish(
+				"hhhh8888", []byte(`{"id":"hhhh8888"}`), TestTimeout)
+			ch <- err
+		}()
+
+		// first EVENT sent successfully
+		Eventually(t, func() bool {
+			select {
+			case <-p.sent:
+				return true
+			default:
+				return false
+			}
+		}, "first EVENT envelope not sent")
+
+		p.disconnect()
+		p.setSendError(errors.New("send failed on reconnect"))
+		p.connect()
+
+		var err error
+		Eventually(t, func() bool {
+			select {
+			case err = <-ch:
+				return err != nil
+			default:
+				return false
+			}
+		}, "Publish did not return a send error after reconnect")
+
+		assert.ErrorContains(t, err, "send failed on reconnect")
+		assert.Len(t, EventsOf[PublishSendFailed](obs), 1)
+	})
+
+	t.Run("close cancels pending", func(t *testing.T) {
+		p, envoy := newMockEnvoy(t)
+		p.connect()
+
+		pub := NewEventPublisher(envoy)
+
+		ch := make(chan error, 1)
+		go func() {
+			_, _, err := pub.Publish(
+				"iiii9999", []byte(`{"id":"iiii9999"}`), TestTimeout)
+			ch <- err
+		}()
+
+		// wait for EVENT to be sent before closing
+		Eventually(t, func() bool {
+			select {
+			case <-p.sent:
+				return true
+			default:
+				return false
+			}
+		}, "EVENT envelope not sent")
+
+		pub.Close()
+
+		var err error
+		Eventually(t, func() bool {
+			select {
+			case err = <-ch:
+				return err != nil
+			default:
+				return false
+			}
+		}, "Publish did not return a cancellation error")
+
+		assert.ErrorContains(t, err, "closed")
+	})
+
+	t.Run("unknown OK ignored", func(t *testing.T) {
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
+		p.connect()
+
+		pub := NewEventPublisher(envoy)
+		t.Cleanup(pub.Close)
+
+		p.receive([]byte(envelope.EncloseOK("no-such-id", true, "")))
+
+		Never(t, func() bool {
+			pub.mu.Lock()
+			pendingLen := len(pub.pending)
+			pub.mu.Unlock()
+			return pendingLen > 0 ||
+				len(EventsOf[PublishAccepted](obs)) > 0 ||
+				len(EventsOf[PublishRejected](obs)) > 0 ||
+				len(EventsOf[PublishTimeout](obs)) > 0 ||
+				len(EventsOf[PublishSendFailed](obs)) > 0
+		}, "unknown OK should produce no side effects")
+	})
+}

request.go

@@ -165,9 +165,8 @@ func NewRequestManager(e *Envoy) *RequestManager {
 		m.logger = slog.New(m.handler).With(slog.Any("component", comp))
 	}
 
-	m.wg.Add(2)
+	m.wg.Go(m.handleEvents)
-	go m.handleEvents()
+	m.wg.Go(m.routeInbox)
-	go m.routeInbox()
 
 	return m
 }
@@ -262,6 +261,9 @@ func (m *RequestManager) Query(
 			// query timed out
 			m.observer.Record(m.envoy.PeerID(),
 				MissedEOSE{SubID: id, At: time.Now()})
+			if m.logger != nil {
+				m.logger.Warn("missed eose", "req", id)
+			}
 			m.Cancel(id)
 			return result, nil, nil
 		}
@@ -278,7 +280,18 @@ func (m *RequestManager) Cancel(id string) error {
 	}
 
 	if req.active {
-		go m.envoy.Send(envelope.EncloseClose(id))
+		go func() {
+			err := m.envoy.Send(envelope.EncloseClose(id))
+			if err != nil {
+				if m.logger != nil {
+					m.logger.Warn("close send failed", "req", req.id, "error", err)
+				}
+				return
+			}
+			if m.logger != nil {
+				m.logger.Debug("close sent", "req", req.id)
+			}
+		}()
 		req.active = false
 	}
 
@@ -317,10 +330,16 @@ func (m *RequestManager) activate(req *request) {
 		if err != nil {
 			m.observer.Record(m.envoy.PeerID(),
 				ReqSendFailed{SubID: req.id, Err: err, At: time.Now()})
+			if m.logger != nil {
+				m.logger.Warn("req send failed", "req", req.id, "error", err)
+			}
 			return
 		}
 		m.observer.Record(m.envoy.PeerID(),
 			ReqDispatched{SubID: req.id, DispatchedAt: time.Now()})
+		if m.logger != nil {
+			m.logger.Debug("req sent", "req", req.id)
+		}
 	}()
 }
 
@@ -334,8 +353,6 @@ func (m *RequestManager) deregister(req *request) {
 }
 
 func (m *RequestManager) handleEvents() {
-	defer m.wg.Done()
-
 	for {
 		select {
 		case <-m.ctx.Done():
@@ -361,8 +378,6 @@ func (m *RequestManager) handleEvents() {
 }
 
 func (m *RequestManager) routeInbox() {
-	defer m.wg.Done()
-
 	for {
 		select {
 		case <-m.ctx.Done():
@@ -450,7 +465,18 @@ func (m *RequestManager) routeEOSE(msg InboxMessage) {
 		req.active = false
 		close(req.buffer)
 		delete(m.reqs, req.id)
-		go m.envoy.Send(envelope.EncloseClose(subID))
+		go func() {
+			err := m.envoy.Send(envelope.EncloseClose(subID))
+			if err != nil {
+				if m.logger != nil {
+					m.logger.Warn("close send failed", "req", req.id, "error", err)
+				}
+				return
+			}
+			if m.logger != nil {
+				m.logger.Debug("close sent", "req", req.id)
+			}
+		}()
 	}
 }
 
@@ -473,6 +499,9 @@ func (m *RequestManager) routeClosed(msg InboxMessage) {
 			ReceivedAt: receivedAt,
 			Message:    message,
 		})
+	if m.logger != nil {
+		m.logger.Warn("req closed by peer", "req", req.id, "message", message)
+	}
 	req.closedOnce.Do(func() {
 		req.closed <- ReqClosed{
 			PeerID:     msg.ID,

request_test.go

@@ -1,6 +1,7 @@
 package prism
 
 import (
+	"errors"
 	"git.wisehodl.dev/jay/go-roots-ws"
 	"github.com/stretchr/testify/assert"
 	"testing"
@@ -78,7 +79,8 @@ func TestRequestManager_Options(t *testing.T) {
 
 func TestRequestManager_Stream(t *testing.T) {
 	t.Run("sends req when connected", func(t *testing.T) {
-		p, envoy := newMockEnvoy(t)
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
 
 		p.connect()
 		Eventually(t, envoy.IsConnected, "envoy should be connected")
@@ -104,6 +106,12 @@ func TestRequestManager_Stream(t *testing.T) {
 		}, "expected REQ send")
 
 		assert.Equal(t, []byte(envelope.EncloseReq(id, filters)), got)
+
+		Eventually(t, func() bool {
+			return len(EventsOf[ReqDispatched](obs)) == 1
+		}, "expected ReqDispatched observable")
+		dispatched := EventsOf[ReqDispatched](obs)
+		assert.Equal(t, id, dispatched[0].SubID)
 	})
 
 	t.Run("does not send req when disconnected", func(t *testing.T) {
@@ -130,7 +138,8 @@ func TestRequestManager_Stream(t *testing.T) {
 	})
 
 	t.Run("forwards events to caller", func(t *testing.T) {
-		p, envoy := newMockEnvoy(t)
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
 		p.connect()
 		Eventually(t, envoy.IsConnected, "envoy should be connected")
 
@@ -172,10 +181,18 @@ func TestRequestManager_Stream(t *testing.T) {
 		assert.Len(t, got, 2)
 		assert.Equal(t, eventA, got[0].Data)
 		assert.Equal(t, eventB, got[1].Data)
+
+		Eventually(t, func() bool {
+			return len(EventsOf[FirstEventReceived](obs)) == 1
+		}, "expected FirstEventReceived observable")
+		first := EventsOf[FirstEventReceived](obs)
+		assert.Equal(t, id, first[0].SubID)
+		assert.False(t, first[0].ReceivedAt.IsZero())
 	})
 
 	t.Run("ignores eose", func(t *testing.T) {
-		p, envoy := newMockEnvoy(t)
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
 		p.connect()
 		Eventually(t, envoy.IsConnected, "envoy should be connected")
 
@@ -225,10 +242,17 @@ func TestRequestManager_Stream(t *testing.T) {
 				return false
 			}
 		}, "expected event after eose")
+
+		Eventually(t, func() bool {
+			return len(EventsOf[StreamEOSEReceived](obs)) == 1
+		}, "expected StreamEOSEReceived observable")
+		eoseEvents := EventsOf[StreamEOSEReceived](obs)
+		assert.Equal(t, id, eoseEvents[0].SubID)
 	})
 
 	t.Run("closed deregisters and signals caller", func(t *testing.T) {
-		p, envoy := newMockEnvoy(t)
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
 		p.connect()
 		Eventually(t, envoy.IsConnected, "envoy should be connected")
 
@@ -261,6 +285,13 @@ func TestRequestManager_Stream(t *testing.T) {
 		}, "expected closed signal")
 		assert.Equal(t, "error: test", got.Data)
 
+		Eventually(t, func() bool {
+			return len(EventsOf[ClosedReceived](obs)) == 1
+		}, "expected ClosedReceived observable")
+		closedEvents := EventsOf[ClosedReceived](obs)
+		assert.Equal(t, id, closedEvents[0].SubID)
+		assert.Equal(t, "error: test", closedEvents[0].Message)
+
 		Eventually(t, func() bool {
 			select {
 			case _, ok := <-events:
@@ -284,6 +315,45 @@ func TestRequestManager_Stream(t *testing.T) {
 		m.mu.RUnlock()
 		assert.False(t, ok, "registration should be removed from reqs")
 	})
+
+	t.Run("send failure recorded", func(t *testing.T) {
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
+
+		p.connect()
+		Eventually(t, envoy.IsConnected, "envoy should be connected")
+
+		m := NewRequestManager(envoy)
+		t.Cleanup(func() { m.Close() })
+
+		// disconnect, arm send error, reconnect so the send attempt
+		// happens after the error is in place
+		p.disconnect()
+		p.setSendError(errors.New("send failed"))
+		p.connect()
+		Eventually(t, envoy.IsConnected, "envoy should be reconnected")
+
+		filters := [][]byte{[]byte(`{}`)}
+		id, _, _, err := m.Stream(filters)
+		assert.NoError(t, err)
+
+		Eventually(t, func() bool {
+			// 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)
+		assert.NotNil(t, failed[0].Err)
+
+		Never(t, func() bool {
+			select {
+			case <-p.sent:
+				return true
+			default:
+				return false
+			}
+		}, "no REQ envelope should have been sent")
+	})
 }
 
 func TestRequestManager_Cancel(t *testing.T) {
@@ -377,7 +447,8 @@ func TestRequestManager_Cancel(t *testing.T) {
 
 func TestRequestManager_Query(t *testing.T) {
 	t.Run("returns events and nil closed on eose", func(t *testing.T) {
-		p, envoy := newMockEnvoy(t)
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
 		p.connect()
 		Eventually(t, envoy.IsConnected, "envoy should be connected")
 
@@ -387,6 +458,7 @@ func TestRequestManager_Query(t *testing.T) {
 		filters := [][]byte{[]byte(`{}`)}
 		eventData := []byte(`{"id":"abc"}`)
 
+		var querySubID string
 		go func() {
 			// wait for the REQ to arrive, extract the sub ID
 			reqBytes := <-p.sent
@@ -395,6 +467,7 @@ func TestRequestManager_Query(t *testing.T) {
 				t.Errorf("FindReq: %v", err)
 				return
 			}
+			querySubID = subID
 			// inject three EVENTs then EOSE
 			for range 3 {
 				raw := envelope.EncloseSubscriptionEvent(subID, eventData)
@@ -409,6 +482,12 @@ func TestRequestManager_Query(t *testing.T) {
 		assert.Len(t, events, 3)
 		assert.Nil(t, closed)
 
+		Eventually(t, func() bool {
+			return len(EventsOf[QueryEOSEReceived](obs)) == 1
+		}, "expected QueryEOSEReceived observable")
+		qeose := EventsOf[QueryEOSEReceived](obs)
+		assert.Equal(t, querySubID, qeose[0].SubID)
+
 		// CLOSE envelope should have been sent after EOSE
 		var closeEnv []byte
 		select {
@@ -451,7 +530,8 @@ func TestRequestManager_Query(t *testing.T) {
 	})
 
 	t.Run("returns partial events on timeout", func(t *testing.T) {
-		p, envoy := newMockEnvoy(t)
+		obs := &mockObserver{}
+		p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
 		p.connect()
 		Eventually(t, envoy.IsConnected, "envoy should be connected")
 
@@ -483,6 +563,9 @@ func TestRequestManager_Query(t *testing.T) {
 		assert.GreaterOrEqual(t, elapsed, queryTimeout)
 		assert.Len(t, events, 2)
 		assert.Nil(t, closed)
+
+		missed := EventsOf[MissedEOSE](obs)
+		assert.Len(t, missed, 1)
 	})
 
 	t.Run("connects within timeout and returns events", func(t *testing.T) {