Exception (505) Reason: "UNEXPECTED_FRAME - expected content header for class 60, got non content header frame instead"

[imirkin]

I've been a long-time user of the streadway/amqp library. Recently updated to this repository.

Since then, I have one instance of the error in the subject being sent as a close notification (Exception (505) Reason: "UNEXPECTED_FRAME - expected content header for class 60, got non content header frame instead"). This happened very shortly after program start (within the first second).

Searching around online, the most common explanation seems to be "you're using concurrency, but the library isn't designed for concurrency". However this library has a lock on Publish(), which serializes everything. I don't think the explanation is so simple in this instance.

My usage is fairly vanilla. There are pub-sub "fanout" exchanges. The publish happens with

c.Publish("", key, true, false,
	amqp091.Publishing{
		DeliveryMode: amqp091.Persistent,
		ContentType:  contentType,
		Body:         body,
	})

The message itself is binary. It may be small (several bytes), or it may be large (100kb+, but probably not as big as 1MB).

What I'm looking for are some plausible theories as to how this could happen. Is there some sort of race at startup? Are we meant to wait for something after ExchangeDeclare? We also have a consumer subscription on the same Channel. I don't think consuming messages can cause this sort of thing, but maybe in some weird instance?

Is there some additional information I should be looking at?

[michaelklishin]

You publishing concurrently with other operations on a shared channel. No client supports that. It's been discussed hundreds of times on rabbitmq-users but to reiterate: a publish is at least two, usually three or more frames. They must all go in sequence (on a single channel) on the wire, interleaving with any other frame would be considered invalid.

So look for places that need synchronisation in your code. Or delay publishing by N seconds before topology is set up. Or something like that.

[imirkin]

Also why did this get converted into a discussion? The library performs synchronization, but that synchronization is failing somewhere... so bug in library, no? I was looking for help in identifying opportunities for the bug in the library, since after some digging around it all looked pretty safe.

[michaelklishin]

Because we use issues for actionable items. It means there's a solution that the community is willing to accept, not just a problem statement without an executable example that can be used to reproduce (at least some of the time).

This is open source software. You are welcome to investigate what's going on and propose a solution (or submit a PR). Putting
pressure on others is not how open source works.

[michaelklishin]

I don't see how a channel-wide lock during a publish would help with two methods that arrive from the server:

• basic.ack for publisher confirms
• basic.return for returns

Publishing (basic.publish and the content frames) is entirely asynchronous in the protocol by design.

It's the dispatch of inbound methods that matters here, unless you can provide evidence of the contrary.

[imirkin]

Sounds good on the discussion point. A bit different than what I've seen before, but wtvr, all the same in the end.

I'm still weak on what you're proposing the illegal sequence might be. Please note that I'm not an expert on AMQP 0.9 (or any version for that matter).

The error is generated on the MQ server, so I assumed it was in response to something the client sent. Are you saying that there's a potential race between publishing and receiving messages?

The incoming message processing happens here from what I can tell:

https://github.com/rabbitmq/amqp091-go/blob/main/channel.go#L272

and according to the comments, it ends up being all serialized too. But perhaps it can race against a publish? But since it's an async protocol, as you mention, unclear why that would matter.

[michaelklishin]

There is no illegal sequence w.r.t. basic.ack vs. basic.return sent by RabbitMQ nodes. They are single-frame methods and
any interleaving would be fine. I don't remember why the decision to send returns before acks was made but it generally makes sense to me.

You can inspect what is sent using Wireshark which has a very decent dissector for AMQP 0-9-1. From there, the question is how those methods are dispatched and why the sequence is indeterministic (not perfectly reproducible). It's been a while since I've touched this client but the Channel#dispatch(message) function linked above
makes a lot of sense as a starting point.

Neither of those methods can race against a publish because basic.ack in this case and basic.return (always) are sent by RabbitMQ, while publishes are sent by clients.

That dispatch function accepts an inbound frame and does something. When a basic.ack comes in right after a basic.return,
that "something" creates a race condition of some kind (that's my best guess). That's what we need to understand first,
before any solution would be considered and eventually filed as an actionable issue.

Those frames are sent on the same channel sequentially, and go from a channel process on the RabbitMQ to the socket in the same sequence. So there is no room for a similar race condition.

[michaelklishin]

As I generally expected, a basic.return is dispatched to a Go channel:

	case *basicReturn:
		ret := newReturn(*m)
		ch.notifyM.RLock()
		for _, c := range ch.returns {
			c <- *ret
		}
		ch.notifyM.RUnlock()

and basic.ack (or nack) are stored in a data structure that keeps track of confirms:

	case *basicAck:
		if ch.confirming {
			if m.Multiple {
				ch.confirms.Multiple(Confirmation{m.DeliveryTag, true})
			} else {
				ch.confirms.One(Confirmation{m.DeliveryTag, true})
			}
		}

	case *basicNack:
		if ch.confirming {
			if m.Multiple {
				ch.confirms.Multiple(Confirmation{m.DeliveryTag, false})
			} else {
				ch.confirms.One(Confirmation{m.DeliveryTag, false})
			}
		}

// ...

func (c *confirms) One(confirmed Confirmation) {
	c.m.Lock()
	defer c.m.Unlock()

	c.deferredConfirmations.Confirm(confirmed)

	if c.expecting == confirmed.DeliveryTag {
		c.confirm(confirmed)
	} else {
		c.sequencer[confirmed.DeliveryTag] = confirmed
	}
	c.resequence()
}

I can only speculate what kind of behavior you see since there never was a code example (some published message properties are not relevant) but at least one the methods is handed over to a Go channel, which by definition can create a race condition with
the goroutine executing the dispatch.

[michaelklishin]

Where do you go from here? We do not guess in this community.

1. Put together a repo that demonstrates the behavior
2. Take a traffic capture and see what the server actually sends when that example runs
3. Add debug logging to the dispatch and basic.return, (inbound) basic.ack methods

then you would have something to work with.

[imirkin]

Unfortunately I don't have a reproducer. The error in question has happened a grand total of once - so a tcpdump is out. What I was hoping to get were some ideas for what could realistically be happening, given the IMO vanilla usage, so that I could write a reproducer application that would hit the problem.

Anyways, I guess I'll try some stuff, see if I can repro.