[POSSIBLE BUG] - Publisher get frozen when publishing without any delay

[sujit-baniya]

I'm using BadgerDB as persistent DB and when trying to publish around 500K messages with Persistent messages for testing without any delay, it causes publisher to freeze.

package main

import (
    "bytes"
    "context"
    "fmt"
    "log"
    "time"

    amqp "github.com/oarkflow/amqp/amqp091"

    grabbit "github.com/oarkflow/amqp"
)

func OnPubReattempting(name string, retry int) bool {
    log.Printf("callback_redo: {%s} retry count {%d}", name, retry)
    return true // want continuing
}

// OnNotifyPublish CallbackNotifyPublish
func OnNotifyPublish(confirm amqp.Confirmation, ch *grabbit.Channel) {
    log.Printf("callback: publish confirmed status [%v] from queue [%s]\n", confirm.Ack, ch.Queue())
}

// OnNotifyReturn CallbackNotifyReturn
func OnNotifyReturn(_ amqp.Return, ch *grabbit.Channel) {
    log.Printf("callback: publish returned from queue [%s]\n", ch.Queue())
}

func PublishMsg(publisher *grabbit.Publisher, start, end int) {
    message := amqp.Publishing{DeliveryMode: amqp.Persistent} // This just runs for some messages and just freezes
    // message := amqp.Publishing{}
    message.Headers = map[string]any{
        "next-queue": "I'm loving it",
    }
    data := make([]byte, 0, 64)
    buff := bytes.NewBuffer(data)

    for i := start; i < end; i++ {
        // <-time.After(1 * time.Millisecond)
        buff.Reset()
        buff.WriteString(fmt.Sprintf("test number %04d", i))
        message.Body = buff.Bytes()
        log.Println("going to send:", buff.String())

        if err := publisher.Publish(message); err != nil {
            log.Println("publishing failed with: ", err)
        }
    }
}

func main() {
    ctxMaster, ctxCancel := context.WithCancel(context.TODO())
    conn := grabbit.NewConnection("amqp://guest:guest@localhost:5672", amqp.Config{}, grabbit.WithConnectionCtx(ctxMaster))
    pubOpt := grabbit.DefaultPublisherOptions()
    pubOpt.WithKey("workload").WithContext(ctxMaster).WithConfirmationsCount(20)

    topos := make([]*grabbit.TopologyOptions, 0, 8)
    topos = append(topos, &grabbit.TopologyOptions{
        Name:          "workload",
        IsDestination: true,
        Durable:       true,
        Declare:       true,
    })
    publisher := grabbit.NewPublisher(conn, pubOpt,
        grabbit.WithChannelCtx(ctxMaster),
        grabbit.WithChannelTopology(topos),
        grabbit.OnChannelRecovering(OnPubReattempting),
        grabbit.OnPublishSuccess(OnNotifyPublish),
        grabbit.OnPublishFailure(OnNotifyReturn),
    )
    if !publisher.AwaitAvailable(30*time.Second, 1*time.Second) {
        log.Println("publisher not ready yet")
        ctxCancel()
        return
    }

    PublishMsg(publisher, 0, 500000)
}

Screenshot

Seems Channel is closed before all messages are being published

[valinurovam]

It's looke like some issue is here

log.Printf("callback: publish confirmed status [%v] from queue [%s]\n", confirm.Ack, ch.Queue())

channel.Queue method has a lock

func (ch *Channel) Queue() string {
    ch.baseChan.mu.RLock()
    defer ch.baseChan.mu.RUnlock()

    return ch.queue
}

If you change ch.Queue() for simple string, for example "queueName" - all is good

[valinurovam]

It looks like lock/unluck on ch.Queue is useless and make some decrease in performance

[sujit-baniya]

@valinurovam Thank you for response.

So you mean, If I remove lock, it might work?

Question: Is lock necessary there? Because I don't see any operations that might cause deadlock

[valinurovam]

Looks like deadlock somewhere. If you change method like this

// Queue returns the active (as indicated by [IsDestination] option in topology options) queue name.
// Useful for finding the server assigned name.
func (ch *Channel) Queue() string {
    fmt.Println("try to get lock... ")
    ch.baseChan.mu.RLock()
    fmt.Println(" locked.")
    defer func() {
        fmt.Println(" unlocked.\n")
        ch.baseChan.mu.RUnlock()
    }()

    return ch.queue
}

Logs

try to get lock... 
2023/12/29 19:11:40 going to send: test number 1623
 locked.
 unlocked.

callback: publish confirmed status [true] from queue [workload]
try to get lock... 
2023/12/29 19:11:40 going to send: test number 1624
 locked.
 unlocked.

callback: publish confirmed status [true] from queue [workload]
2023/12/29 19:11:40 going to send: test number 1625
2023/12/29 19:11:40 going to send: test number 1626
2023/12/29 19:11:40 going to send: test number 1627
2023/12/29 19:11:40 going to send: test number 1628
try to get lock... 
2023/12/29 19:11:40 going to send: test number 1629
 locked.
 unlocked.

callback: publish confirmed status [true] from queue [workload]
try to get lock...