4 Commits

Author SHA1 Message Date
Jay df0a968907 add logger to auth manager 2026-06-04 16:03:10 -04:00
Jay dda3bfd5b7 cleanup: flakey tests and race conditions 2026-06-04 12:07:03 -04:00
Jay b5b0ec77ba chore: bump go-honeywood to v0.5.1 2026-06-03 23:26:51 -04:00
Jay c11e40b48a chore: bump go-mana-component to v0.2.0, go-roots-ws to v0.2.0 2026-06-03 23:23:54 -04:00
8 changed files with 48 additions and 18 deletions
+22 -5
View File
@@ -2,6 +2,8 @@ package prism
import ( import (
"context" "context"
"git.wisehodl.dev/jay/go-mana-component"
"log/slog"
"sync" "sync"
"time" "time"
@@ -30,14 +32,18 @@ type AuthManager struct {
inbox <-chan InboxMessage inbox <-chan InboxMessage
events <-chan OutboundPoolEvent events <-chan OutboundPoolEvent
ctx context.Context ctx context.Context
cancel context.CancelFunc cancel context.CancelFunc
mu sync.Mutex mu sync.Mutex
wg sync.WaitGroup wg sync.WaitGroup
handler slog.Handler
logger *slog.Logger
} }
func NewAuthManager(e *Envoy, respond func(string) ([]byte, error)) *AuthManager { func NewAuthManager(e *Envoy, respond func(string) ([]byte, error)) *AuthManager {
ctx, cancel := context.WithCancel(e.Context()) ctx, cancel := context.WithCancel(
component.MustExtend(e.Context(), "auth_manager"))
m := &AuthManager{ m := &AuthManager{
envoy: e, envoy: e,
respond: respond, respond: respond,
@@ -46,6 +52,13 @@ func NewAuthManager(e *Envoy, respond func(string) ([]byte, error)) *AuthManager
ctx: ctx, ctx: ctx,
cancel: cancel, cancel: cancel,
} }
if e.Handler() != nil {
comp := component.FromContext(ctx)
m.handler = e.Handler()
m.logger = slog.New(m.handler).With(slog.Any("component", comp))
}
m.wg.Go(m.routeInbox) m.wg.Go(m.routeInbox)
m.wg.Go(m.handleEvents) m.wg.Go(m.handleEvents)
return m return m
@@ -85,6 +98,10 @@ func (m *AuthManager) routeInbox() {
continue continue
} }
if m.logger != nil {
m.logger.Info("responded to auth", "challenge", challenge)
}
m.envoy.Observer().Record(msg.ID, AuthResponseSent{At: time.Now()}) m.envoy.Observer().Record(msg.ID, AuthResponseSent{At: time.Now()})
} }
} }
+8 -1
View File
@@ -16,8 +16,11 @@ func TestAuthManager(t *testing.T) {
p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs)) p, envoy := newMockEnvoy(t, WithEmbassyObserver(obs))
signed := []byte(`{"kind":22242}`) signed := []byte(`{"kind":22242}`)
var mu sync.Mutex
var calledWith string var calledWith string
callback := func(challenge string) ([]byte, error) { callback := func(challenge string) ([]byte, error) {
mu.Lock()
defer mu.Unlock()
calledWith = challenge calledWith = challenge
return signed, nil return signed, nil
} }
@@ -28,7 +31,11 @@ func TestAuthManager(t *testing.T) {
p.connect() p.connect()
p.receive([]byte(envelope.EncloseAuthChallenge("abc"))) 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") "callback not called with challenge")
Eventually(t, func() bool { Eventually(t, func() bool {
+3 -3
View File
@@ -3,9 +3,9 @@ module git.wisehodl.dev/jay/go-mana-prism
go 1.25.0 go 1.25.0
require ( require (
git.wisehodl.dev/jay/go-honeybee v0.5.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 github.com/stretchr/testify v1.11.1
) )
+6
View File
@@ -1,9 +1,15 @@
git.wisehodl.dev/jay/go-honeybee v0.5.0 h1:tAhXMLJnxMMhWJstA1f3OYVsdQf0rstl5/AaJVzuCWk= git.wisehodl.dev/jay/go-honeybee v0.5.0 h1:tAhXMLJnxMMhWJstA1f3OYVsdQf0rstl5/AaJVzuCWk=
git.wisehodl.dev/jay/go-honeybee v0.5.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 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.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 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.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 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= 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= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+2 -2
View File
@@ -139,8 +139,6 @@ func (p *EventPublisher) Close() {
p.wg.Wait() 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 { func (p *EventPublisher) trySend(entry *pendingEntry) error {
entry.mu.Lock() entry.mu.Lock()
defer entry.mu.Unlock() defer entry.mu.Unlock()
@@ -212,9 +210,11 @@ func (p *EventPublisher) handleEvents() {
p.mu.Lock() p.mu.Lock()
var toSend []*pendingEntry var toSend []*pendingEntry
for _, e := range p.pending { for _, e := range p.pending {
e.mu.Lock()
if !e.sent { if !e.sent {
toSend = append(toSend, e) toSend = append(toSend, e)
} }
e.mu.Unlock()
} }
p.mu.Unlock() p.mu.Unlock()
+3 -1
View File
@@ -364,7 +364,9 @@ func TestEventPublisher(t *testing.T) {
}, "Publish did not return a send error") }, "Publish did not return a send error")
assert.ErrorContains(t, err, "send failed") 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) { t.Run("disconnect then send failure on reconnect", func(t *testing.T) {
+2 -2
View File
@@ -157,11 +157,11 @@ func NewRequestManager(e *Envoy) *RequestManager {
ctx: ctx, ctx: ctx,
cancel: cancel, cancel: cancel,
observer: e.Observer(), observer: e.Observer(),
handler: e.Handler(),
} }
if m.handler != nil { if e.Handler() != nil {
comp := component.FromContext(ctx) comp := component.FromContext(ctx)
m.handler = e.Handler()
m.logger = slog.New(m.handler).With(slog.Any("component", comp)) m.logger = slog.New(m.handler).With(slog.Any("component", comp))
} }
+2 -4
View File
@@ -333,15 +333,13 @@ func TestRequestManager_Stream(t *testing.T) {
p.connect() p.connect()
Eventually(t, envoy.IsConnected, "envoy should be reconnected") 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(`{}`)} filters := [][]byte{[]byte(`{}`)}
id, _, _, err := m.Stream(filters) id, _, _, err := m.Stream(filters)
assert.NoError(t, err) assert.NoError(t, err)
Eventually(t, func() bool { 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") }, "expected ReqSendFailed observable")
failed := EventsOf[ReqSendFailed](obs) failed := EventsOf[ReqSendFailed](obs)
assert.Equal(t, id, failed[0].SubID) assert.Equal(t, id, failed[0].SubID)