Fail an approval request fast when no subscriber received the prompt

This commit is contained in:
Viktor Liu
2026-08-29 16:00:36 +02:00
parent e104cef490
commit d0d8813dcb
5 changed files with 45 additions and 6 deletions
+11 -2
View File
@@ -90,13 +90,15 @@ var ErrDenied = errors.New("approval denied")
// EventPublisher is the subset of peer.Status used to emit prompts. // EventPublisher is the subset of peer.Status used to emit prompts.
type EventPublisher interface { type EventPublisher interface {
// PublishEvent reports whether the event reached at least one live
// subscriber. A prompt nobody received is one nobody can answer.
PublishEvent( PublishEvent(
severity proto.SystemEvent_Severity, severity proto.SystemEvent_Severity,
category proto.SystemEvent_Category, category proto.SystemEvent_Category,
msg string, msg string,
userMsg string, userMsg string,
metadata map[string]string, metadata map[string]string,
) ) bool
HasEventSubscribers() bool HasEventSubscribers() bool
} }
@@ -171,7 +173,14 @@ func (b *Broker) Request(ctx context.Context, p Prompt) (Decision, error) {
if subject == "" { if subject == "" {
subject = fmt.Sprintf("%s connection requires approval", p.Kind) subject = fmt.Sprintf("%s connection requires approval", p.Kind)
} }
b.pub.PublishEvent(proto.SystemEvent_INFO, proto.SystemEvent_APPROVAL, subject, subject, meta) // A subscriber whose queue is full drops the event silently. Waiting out the
// timeout for a prompt that never appeared spends the caller's whole
// approval window before denying, and tells the user it timed out rather
// than that nothing ever asked them.
if !b.pub.PublishEvent(proto.SystemEvent_INFO, proto.SystemEvent_APPROVAL, subject, subject, meta) {
log.Warnf("approval request %s (%s) reached no subscriber; denying without waiting", id, p.Kind)
return zero, ErrNoSubscriber
}
log.Debugf("approval request %s (%s) emitted: %s", id, p.Kind, subject) log.Debugf("approval request %s (%s) emitted: %s", id, p.Kind, subject)
timer := time.NewTimer(timeout) timer := time.NewTimer(timeout)
+25 -1
View File
@@ -20,6 +20,9 @@ type fakePublisher struct {
mu sync.Mutex mu sync.Mutex
subscribers bool subscribers bool
events []*proto.SystemEvent events []*proto.SystemEvent
// dropped makes PublishEvent report that no subscriber received the event,
// standing in for a subscriber whose queue is full.
dropped bool
} }
func (p *fakePublisher) PublishEvent( func (p *fakePublisher) PublishEvent(
@@ -28,7 +31,7 @@ func (p *fakePublisher) PublishEvent(
msg string, msg string,
userMsg string, userMsg string,
metadata map[string]string, metadata map[string]string,
) { ) bool {
p.mu.Lock() p.mu.Lock()
p.events = append(p.events, &proto.SystemEvent{ p.events = append(p.events, &proto.SystemEvent{
Severity: severity, Severity: severity,
@@ -37,7 +40,9 @@ func (p *fakePublisher) PublishEvent(
UserMessage: userMsg, UserMessage: userMsg,
Metadata: metadata, Metadata: metadata,
}) })
dropped := p.dropped
p.mu.Unlock() p.mu.Unlock()
return !dropped
} }
func (p *fakePublisher) HasEventSubscribers() bool { func (p *fakePublisher) HasEventSubscribers() bool {
@@ -60,6 +65,25 @@ func (p *fakePublisher) eventCount() int {
return len(p.events) return len(p.events)
} }
// A subscriber can exist and still not receive the prompt, because its event
// queue is full and the publisher drops rather than blocks. Waiting out the
// approval window for a dialog that never opened wastes the caller's whole
// timeout and then reports it as one, so the broker refuses straight away.
func TestRequestUndeliveredPromptFailsFast(t *testing.T) {
pub := &fakePublisher{subscribers: true, dropped: true}
b := New(pub)
start := time.Now()
_, err := b.Request(context.Background(), Prompt{Kind: KindVNC, Subject: "test"})
assert.ErrorIs(t, err, ErrNoSubscriber)
assert.Less(t, time.Since(start), time.Second, "must not wait out the approval timeout")
b.mu.Lock()
pending := len(b.pending)
b.mu.Unlock()
assert.Equal(t, 0, pending, "no waiter must be left behind")
}
// TestRequestNoSubscriberFailsClosed is the core fail-closed invariant: // TestRequestNoSubscriberFailsClosed is the core fail-closed invariant:
// when the UI is not subscribed, the broker must refuse without emitting // when the UI is not subscribed, the broker must refuse without emitting
// an event or arming a waiter. A regression here is a silent bypass. // an event or arming a waiter. A regression here is a silent bypass.
+3 -1
View File
@@ -76,13 +76,15 @@ var (
// owned by sessionwatch, not the caller. // owned by sessionwatch, not the caller.
type StatusRecorder interface { type StatusRecorder interface {
SetSessionExpiresAt(deadline time.Time) SetSessionExpiresAt(deadline time.Time)
// PublishEvent reports whether any subscriber received the event; the
// watcher does not act on it.
PublishEvent( PublishEvent(
severity cProto.SystemEvent_Severity, severity cProto.SystemEvent_Severity,
category cProto.SystemEvent_Category, category cProto.SystemEvent_Category,
message string, message string,
userMessage string, userMessage string,
metadata map[string]string, metadata map[string]string,
) ) bool
} }
// Watcher observes the latest session deadline and fires two warnings // Watcher observes the latest session deadline and fires two warnings
@@ -64,7 +64,7 @@ func (r *fakeRecorder) PublishEvent(
message string, message string,
_ string, _ string,
metadata map[string]string, metadata map[string]string,
) { ) bool {
r.mu.Lock() r.mu.Lock()
defer r.mu.Unlock() defer r.mu.Unlock()
r.events = append(r.events, event{ r.events = append(r.events, event{
@@ -74,6 +74,7 @@ func (r *fakeRecorder) PublishEvent(
message: message, message: message,
meta: metadata, meta: metadata,
}) })
return true
} }
func (r *fakeRecorder) snapshot() []event { func (r *fakeRecorder) snapshot() []event {
+4 -1
View File
@@ -1306,7 +1306,7 @@ func (d *Status) PublishEvent(
msg string, msg string,
userMsg string, userMsg string,
metadata map[string]string, metadata map[string]string,
) { ) bool {
event := &proto.SystemEvent{ event := &proto.SystemEvent{
Id: uuid.New().String(), Id: uuid.New().String(),
Severity: severity, Severity: severity,
@@ -1322,15 +1322,18 @@ func (d *Status) PublishEvent(
d.eventQueue.Add(event) d.eventQueue.Add(event)
delivered := false
for _, stream := range d.eventStreams { for _, stream := range d.eventStreams {
select { select {
case stream <- event: case stream <- event:
delivered = true
default: default:
log.Debugf("event stream buffer full, skipping event: %v", event) log.Debugf("event stream buffer full, skipping event: %v", event)
} }
} }
log.Debugf("event published: %v", event) log.Debugf("event published: %v", event)
return delivered
} }
// SubscribeToEvents returns a new event subscription // SubscribeToEvents returns a new event subscription