connection block when write connection reset by peer

[yywing]

Describe the bug

version: v1.8.1

NotifyClose got error：

time="2023-09-26T10:09:02Z" level=warning msg="Monitor NotifyClose: Exception (501) Reason: \"write tcp 169.254.1.1:47048->169.254.1.5:5672: write: connection reset by peer\"" func=mq

pprof block:

goroutine 52 [select, 101 minutes]:
github.com/rabbitmq/amqp091-go.(*Channel).call(0xc00041eea0, {0xcbf328, 0xc00040d420}, {0xc0003553f8, 0x1, 0x0?})
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/channel.go:178 +0xff
github.com/rabbitmq/amqp091-go.(*Channel).open(...)
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/channel.go:167
github.com/rabbitmq/amqp091-go.(*Connection).openChannel(0xc000302be8?)
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:833 +0x98
github.com/rabbitmq/amqp091-go.(*Connection).Channel(...)
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:854
....

goroutine 214 [chan receive, 101 minutes]:
github.com/rabbitmq/amqp091-go.(*Connection).call(0xc0002bcdc0, {0xcbf440?, 0xc000336a00?}, {0xc000116d60, 0x1, 0xc000116d88?})
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:866 +0xee
github.com/rabbitmq/amqp091-go.(*Connection).closeWith(0xc0002bcdc0, 0x1129120)
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:443 +0x10a
github.com/rabbitmq/amqp091-go.(*Connection).dispatchClosed(0xc000414947?, {0xcbc258?, 0xc00031ef48?})
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:692 +0x1f9
github.com/rabbitmq/amqp091-go.(*Connection).dispatchN(0xc0002bcdc0, {0xcbc258?, 0xc00031ef48?})
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:664 +0x1e5
github.com/rabbitmq/amqp091-go.(*Connection).demux(0xc000116f28?, {0xcbc258, 0xc00031ef48})
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:613 +0x5b
github.com/rabbitmq/amqp091-go.(*Connection).reader(0xc0002bcdc0, {0xcba440?, 0xc00029c168?})
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:717 +0x23d
created by github.com/rabbitmq/amqp091-go.Open
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:268 +0x34c

Reproduction steps

close mq connection by hand

Expected behavior

no block

Additional context

There are two goroutine are wired. Maybe there is two bug?

Code like:

var conn *amqp.Connection
var mqAddr string

go func() {
   for {
      ch := conn.Channel()
      <-channel.NotifyClose(make(chan *amqp.Error, 1))
      channel.Close()
   }
}

go func() {
   for {
      conn := amqp.Dial(mqAddr)
      <-conn.NotifyClose(make(chan *amqp.Error, 1))
      conn.Close()
   }
}

[lukebakken]

Please help us help you by providing enough information that we can easily reproduce this issue:

• Share the source code to a runnable program that exhibits this issue. Provide the code via a git repository that I can clone, compile, and run. The code you provided is insufficient!
• What does "close mq connection by hand" mean, exactly? Did you close the connection using rabbitmqctl the RabbitMQ management web UI, or some other means?
• How did you get the pprof block output?

[yywing]

close mq connection by hand

rabbitmqctl close_connection <connection pid>

How did you get the pprof block output?

curl http://localhost/debug/pprof/goroutine?debug=2

And I try write a demo to exhibits this issue.

[yywing]

goroutine 52 block reason:

1. old := connection.Channel() open channel id 1
2. old.Close() delete contection.channels id 1,and release id 1
3. new := connection.Channel() open new channel id 1 and add id 1 to contection.channels
4. old.Close() will delete contection.channels id 1 which id point to new.
5. new channel will leak, connection.shutdown() nerver close new channel

amqp091-go/channel.go

Lines 464 to 470 in 831d90b

	/*
	Close initiate a clean channel closure by sending a close message with the error
	code set to '200'.
	It is safe to call this method multiple times.
	*/
	func (ch *Channel) Close() error {

not safe enough - -

[yywing]

goroutine 214 block reason：

goroutine 214 [chan receive, 101 minutes]:
github.com/rabbitmq/amqp091-go.(*Connection).call(0xc0002bcdc0, {0xcbf440?, 0xc000336a00?}, {0xc000116d60, 0x1, 0xc000116d88?})
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:866 +0xee
github.com/rabbitmq/amqp091-go.(*Connection).closeWith(0xc0002bcdc0, 0x1129120)
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:443 +0x10a
github.com/rabbitmq/amqp091-go.(*Connection).dispatchClosed(0xc000414947?, {0xcbc258?, 0xc00031ef48?})
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:692 +0x1f9
github.com/rabbitmq/amqp091-go.(*Connection).dispatchN(0xc0002bcdc0, {0xcbc258?, 0xc00031ef48?})
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:664 +0x1e5
github.com/rabbitmq/amqp091-go.(*Connection).demux(0xc000116f28?, {0xcbc258, 0xc00031ef48})
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:613 +0x5b
github.com/rabbitmq/amqp091-go.(*Connection).reader(0xc0002bcdc0, {0xcba440?, 0xc00029c168?})
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:717 +0x23d
created by github.com/rabbitmq/amqp091-go.Open
	/etst/.gvm/pkgsets/go1.19/global/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:268 +0x34c

amqp091-go/connection.go

Lines 436 to 450 in 831d90b

	func (c *Connection) closeWith(err *Error) error {
	if c.IsClosed() {
	return ErrClosed
	}
	defer c.shutdown(err)
	return c.call(
	&connectionClose{
	ReplyCode: uint16(err.Code),
	ReplyText: err.Reason,
	},
	&connectionCloseOk{},
	)
	}

In some case, reader will call c.closeWith, and block at c.call at <- c.rpc. But only reader write to <-c.rpc - -

[yywing]

code for this issue：main.go.txt

[yywing]

my pr is ready #223

[lukebakken]

Thank you! 🏅