mirror/hatchet
1
0
Fork 0
mirror of https://github.com/hatchet-dev/hatchet.git synced 2025-12-29 12:49:45 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix: race condition on err in pgmq (#1198)

Browse Source
This commit is contained in:
abelanger5
2025-01-18 08:20:24 -08:00
committed by GitHub
parent 9efd56c7de
commit 61ae067014
1 changed files with 2 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
internal/msgqueue/postgres/msgqueue.go
View File

@@ -150,7 +150,7 @@ func (p *PostgresMessageQueue) Subscribe(queue msgqueue.Queue, preAck msgqueue.A
}() }()
doTask := func(task msgqueue.Message, ackId *int64) error { doTask := func(task msgqueue.Message, ackId *int64) error {
err = preAck(&task) err := preAck(&task)
if err != nil { if err != nil {
p.l.Error().Err(err).Msg("error pre-acking message") p.l.Error().Err(err).Msg("error pre-acking message")
@@ -229,7 +229,7 @@ func (p *PostgresMessageQueue) Subscribe(queue msgqueue.Queue, preAck msgqueue.A
// start the listener // start the listener
go func() { go func() {
err = p.repo.Listen(subscribeCtx, queue.Name(), func(ctx context.Context, notification *repository.PubMessage) error { err := p.repo.Listen(subscribeCtx, queue.Name(), func(ctx context.Context, notification *repository.PubMessage) error {
// if this is an exchange queue, and the message starts with JSON '{', then we process the message directly // if this is an exchange queue, and the message starts with JSON '{', then we process the message directly
if queue.FanoutExchangeKey() != "" && len(notification.Payload) >= 1 && notification.Payload[0] == '{' { if queue.FanoutExchangeKey() != "" && len(notification.Payload) >= 1 && notification.Payload[0] == '{' {
var task msgqueue.Message var task msgqueue.Message