From 9672ca14c6294a3f4657f1fd3d3a080111b0a3aa Mon Sep 17 00:00:00 2001 From: Jay Date: Sun, 9 Aug 2026 19:10:41 -0400 Subject: [PATCH] ensure entry is registered before timeout is armed. In the case of a zero timeout, the AfterFunc would attempt to deregister an unregistered entry. Then a zombie entry would be added that would never be deregistered. --- publish.go | 21 ++++++++++++--------- 1 file changed, 12 insertions(+), 9 deletions(-) diff --git a/publish.go b/publish.go index ac6babc..89073f4 100644 --- a/publish.go +++ b/publish.go @@ -99,18 +99,21 @@ func (p *EventPublisher) Publish(eventID string, eventJSON []byte, timeout time. 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.URL(), - 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() + + // use timeout if positive and nonzero + if timeout > time.Duration(0) { + entry.timer = time.AfterFunc(timeout, func() { + p.mu.Lock() + p.deregister(eventID) + p.mu.Unlock() + p.envoy.Observer().Record(p.envoy.URL(), + PublishTimeout{EventID: eventID, At: time.Now()}) + p.deliver(entry, publishResult{err: errors.New("publish timeout")}) + }) + } p.mu.Unlock() if connected {