From 69f9219819f613ed6b0098fb506262e1ca19a8bd Mon Sep 17 00:00:00 2001 From: Jay Date: Thu, 4 Jun 2026 17:46:47 -0400 Subject: [PATCH] surface honeybee.EventDialFailed through the prism event pipeline --- auth.go | 2 +- embassy.go | 28 ++++++++++++++----------- embassy_test.go | 56 +++++++++++++++++++++++++++++++++++++++++++++++++ envoy_test.go | 12 +++++------ helpers_test.go | 5 +++++ publish.go | 2 +- request.go | 2 +- 7 files changed, 86 insertions(+), 21 deletions(-) diff --git a/auth.go b/auth.go index 18bc878..c377b84 100644 --- a/auth.go +++ b/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 diff --git a/embassy.go b/embassy.go index 029ba04..d2ff8bb 100644 --- a/embassy.go +++ b/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 { - ch := make(chan OutboundPoolEvent) +func (e *Embassy) subscribeEventsLock(url string) <-chan PoolEvent { + 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{ - ID: ev.ID, Kind: mapEmbassyEvent(ev.Kind), At: ev.At, + case sub <- PoolEvent{ + 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 } diff --git a/embassy_test.go b/embassy_test.go index 3744132..b630f73 100644 --- a/embassy_test.go +++ b/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") + }) +} diff --git a/envoy_test.go b/envoy_test.go index faaca82..9fa40c9 100644 --- a/envoy_test.go +++ b/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 { diff --git a/helpers_test.go b/helpers_test.go index 395b7c4..0b6b760 100644 --- a/helpers_test.go +++ b/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, diff --git a/publish.go b/publish.go index 3a254ff..247cddc 100644 --- a/publish.go +++ b/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 diff --git a/request.go b/request.go index 2e2be13..5640a74 100644 --- a/request.go +++ b/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