make sure we don't block sending this ack

hatchet-dev · Jan 23, 2025 · 21770f8 · 21770f8
1 parent 5a051bd
commit 21770f8
Showing 1 changed file with 6 additions and 3 deletions.
diff --git a/internal/msgqueue/rabbitmq/rabbitmq.go b/internal/msgqueue/rabbitmq/rabbitmq.go
@@ -452,10 +452,13 @@ func (t *MessageQueueImpl) startPublishing() func() error {
 
 						err = pub.PublishWithContext(ctx, "", msg.q.Name(), false, false, pubMsg)
 
-						// retry failed delivery on the next session
 						if err != nil {
-							msg.ackChan <- ack{e: &err}
-							t.msgs <- msg
+							select {
+							case msg.ackChan <- ack{e: &err}:
+								t.msgs <- msg
+							case <-time.After(100 * time.Millisecond):
+								t.l.Error().Msgf("ack channel blocked for %s", msg.ID)
+							}
 							return
 						}