surface honeybee.EventDialFailed through the prism event pipeline

auth.go

@@ -30,7 +30,7 @@ type AuthManager struct {
 	challenge string
 
 	inbox  <-chan InboxMessage
-	events <-chan OutboundPoolEvent
+	events <-chan PoolEvent
 
 	ctx     context.Context
 	cancel  context.CancelFunc

embassy.go

@@ -29,6 +29,7 @@ type EmbassyEventKind int
 const (
 	EventConnected EmbassyEventKind = iota
 	EventDisconnected
+	EventDialFailed
 	EventEmbassyUnknown
 )
 
@@ -38,15 +39,18 @@ func mapEmbassyEvent(kind honeybee.PoolEventKind) EmbassyEventKind {
 		return EventConnected
 	case honeybee.EventDisconnected:
 		return EventDisconnected
+	case honeybee.EventDialFailed:
+		return EventDialFailed
 	default:
 		return EventEmbassyUnknown
 	}
 }
 
-type OutboundPoolEvent struct {
+type PoolEvent struct {
 	ID   string
 	Kind EmbassyEventKind
 	At   time.Time
+	Err  error
 }
 
 type InboxMessage struct {
@@ -85,7 +89,7 @@ func WithEmbassyObserver(o observer.Observer) EmbassyOption {
 type Embassy struct {
 	pool      EmbassyPlugin
 	envoys    map[string]*Envoy
-	eventSubs map[string]chan<- OutboundPoolEvent
+	eventSubs map[string]chan<- PoolEvent
 	inboxSubs map[string]chan<- InboxMessage
 
 	ctx      context.Context
@@ -111,7 +115,7 @@ func NewEmbassy(
 	e := &Embassy{
 		pool:      pool,
 		envoys:    make(map[string]*Envoy),
-		eventSubs: make(map[string]chan<- OutboundPoolEvent),
+		eventSubs: make(map[string]chan<- PoolEvent),
 		inboxSubs: make(map[string]chan<- InboxMessage),
 		ctx:       ctx,
 		cancel:    cancel,
@@ -205,8 +209,8 @@ func (e *Embassy) send(url string, data []byte) error {
 	return e.pool.Send(url, data)
 }
 
-func (e *Embassy) subscribeEventsLock(url string) <-chan OutboundPoolEvent {
+func (e *Embassy) subscribeEventsLock(url string) <-chan PoolEvent {
-	ch := make(chan OutboundPoolEvent)
+	ch := make(chan PoolEvent)
 	e.eventSubs[url] = ch
 	return ch
 }
@@ -261,8 +265,8 @@ func (e *Embassy) routeEvents() {
 			select {
 			case <-e.ctx.Done():
 				return
-			case sub <- OutboundPoolEvent{
+			case sub <- PoolEvent{
-				ID: ev.ID, Kind: mapEmbassyEvent(ev.Kind), At: ev.At,
+				ID: ev.ID, Kind: mapEmbassyEvent(ev.Kind), At: ev.At, Err: ev.Err,
 			}:
 			}
 		}
@@ -312,9 +316,9 @@ type Envoy struct {
 	terminate         func()
 	queue             chan []byte
 	send              func(data []byte) error
-	events            <-chan OutboundPoolEvent
+	events            <-chan PoolEvent
 	inbox             <-chan InboxMessage
-	eventSubs         []chan<- OutboundPoolEvent
+	eventSubs         []chan<- PoolEvent
 	labelledInboxSubs map[string][]chan<- InboxMessage
 	inboxSubs         []chan<- InboxMessage
 
@@ -332,7 +336,7 @@ func newEnvoy(
 	url string,
 	terminate func(),
 	send func(data []byte) error,
-	events <-chan OutboundPoolEvent,
+	events <-chan PoolEvent,
 	inbox <-chan InboxMessage,
 	observer observer.Observer,
 	handler slog.Handler,
@@ -411,10 +415,10 @@ func (e *Envoy) Send(data []byte) error {
 	return e.send(data)
 }
 
-func (e *Envoy) SubscribeEvents() <-chan OutboundPoolEvent {
+func (e *Envoy) SubscribeEvents() <-chan PoolEvent {
 	e.mu.Lock()
 	defer e.mu.Unlock()
-	ch := make(chan OutboundPoolEvent)
+	ch := make(chan PoolEvent)
 	e.eventSubs = append(e.eventSubs, ch)
 	return ch
 }

embassy_test.go

@@ -1,6 +1,7 @@
 package prism
 
 import (
+	"errors"
 	"git.wisehodl.dev/jay/go-honeybee"
 	"git.wisehodl.dev/jay/go-roots-ws"
 	"github.com/stretchr/testify/assert"
@@ -91,3 +92,58 @@ func TestEmbassy_Dispatch(t *testing.T) {
 	envoy = embassy.Call(p.url)
 	assert.Nil(t, envoy)
 }
+
+func TestEmbassy_DialFailed_Routing(t *testing.T) {
+	t.Run("routed to correct subscriber with Err set", func(t *testing.T) {
+		p, envoy := newMockEnvoy(t)
+		sub := envoy.SubscribeEvents()
+
+		dialErr := errors.New("connection refused")
+		p.dialFail(dialErr)
+
+		var got PoolEvent
+		Eventually(t, func() bool {
+			select {
+			case ev := <-sub:
+				got = ev
+				return true
+			default:
+				return false
+			}
+		}, "expected EventDialFailed to be routed to subscriber")
+
+		assert.Equal(t, EventDialFailed, got.Kind)
+		assert.ErrorIs(t, got.Err, dialErr)
+		assert.Equal(t, p.url, got.ID)
+	})
+
+	t.Run("not routed to unrelated envoy subscriber", func(t *testing.T) {
+		p := newMockPool(t)
+		url1 := "wss://peer1"
+		url2 := "wss://peer2"
+
+		added := make(chan struct{}, 2)
+		p.plugin.Connect = func(url string, opts ...honeybee.ConnectOption) error {
+			added <- struct{}{}
+			return nil
+		}
+
+		emb := NewEmbassy(p.ctx, p.plugin)
+		emb.Dispatch(url1)
+		emb.Dispatch(url2)
+
+		envoy2 := emb.Call(url2)
+		sub2 := envoy2.SubscribeEvents()
+
+		p.dialFail(errors.New("refused"))
+
+		Never(t, func() bool {
+			select {
+			case <-sub2:
+				return true
+			default:
+				return false
+			}
+		}, "EventDialFailed for a different peer must not reach unrelated subscriber")
+	})
+}

envoy_test.go

@@ -60,12 +60,12 @@ func TestEnvoy_IsConnected(t *testing.T) {
 	ctx := component.MustNew(context.Background(), "prism", "test")
 	mu := sync.RWMutex{}
 	url := "wss://test"
-	events := make(chan OutboundPoolEvent)
+	events := make(chan PoolEvent)
 
 	envoy := newEnvoy(ctx, url, nil, nil, events, nil, nil, nil)
 	eventSub := envoy.SubscribeEvents()
 
-	gotEvents := []OutboundPoolEvent{}
+	gotEvents := []PoolEvent{}
 	go func() {
 		for ev := range eventSub {
 			mu.Lock()
@@ -74,7 +74,7 @@ func TestEnvoy_IsConnected(t *testing.T) {
 		}
 	}()
 
-	events <- OutboundPoolEvent{
+	events <- PoolEvent{
 		ID: url, Kind: EventConnected, At: time.Now()}
 
 	Eventually(t, func() bool {
@@ -86,12 +86,12 @@ func TestEnvoy_Events(t *testing.T) {
 	ctx := component.MustNew(context.Background(), "prism", "test")
 	mu := sync.RWMutex{}
 	url := "wss://test"
-	events := make(chan OutboundPoolEvent)
+	events := make(chan PoolEvent)
 
 	envoy := newEnvoy(ctx, url, nil, nil, events, nil, nil, nil)
 	eventSub := envoy.SubscribeEvents()
 
-	gotEvents := []OutboundPoolEvent{}
+	gotEvents := []PoolEvent{}
 	go func() {
 		for ev := range eventSub {
 			mu.Lock()
@@ -100,7 +100,7 @@ func TestEnvoy_Events(t *testing.T) {
 		}
 	}()
 
-	events <- OutboundPoolEvent{
+	events <- PoolEvent{
 		ID: url, Kind: EventConnected, At: time.Now()}
 
 	Eventually(t, func() bool {

helpers_test.go

@@ -95,6 +95,11 @@ func (p *mockPool) disconnect() {
 		ID: p.url, Kind: honeybee.EventDisconnected, At: time.Now()}
 }
 
+func (p *mockPool) dialFail(err error) {
+	p.events <- honeybee.PoolEvent{
+		ID: p.url, Kind: honeybee.EventDialFailed, Err: err, At: time.Now()}
+}
+
 func (p *mockPool) receive(data []byte) {
 	p.inbox <- honeybee.InboxMessage{
 		ID:         p.url,

publish.go

@@ -69,7 +69,7 @@ type EventPublisher struct {
 	pending map[string]*pendingEntry // event id -> pending entry
 
 	inbox  <-chan InboxMessage
-	events <-chan OutboundPoolEvent
+	events <-chan PoolEvent
 
 	ctx    context.Context
 	cancel context.CancelFunc

request.go

@@ -115,7 +115,7 @@ type RequestManager struct {
 	reqs map[string]*request
 
 	envoy  *Envoy
-	events <-chan OutboundPoolEvent
+	events <-chan PoolEvent
 	inbox  <-chan InboxMessage
 
 	ctx      context.Context