Fix racing in Open (#171)

* Lock allocator assignment in openComplete

Related to #170. There is an edge case where the background frame
handler receives an error before `Open` completes. In this case, there
is a data race to assign the value of the allocator. The destructor in
shutdown is already a critical section. Adding a tiny critical section
in openComplete to protect the allocator.

Signed-off-by: Aitor Perez Cedres <[email protected]>

* Add a critical section in openTune

Related to #170. There is an edge case where the TCP connection between
client-server is setup, and AMQP handshake starts, up to the point right
after sending Tune frame. At this point, the background frame reader
receives an error from the TCP socket and starts the shutdown sequence.
At the same time, the Tune function continues (as it has not completed)
and attempts to set the ChannelMax field in the connection struct. At
the same time, the shutdown sequence initiated by the error in the frame
handler reads the ChannelMax field. This creates a race.

A potential solution is to add a critical section in tune to protect
access to ChannelMax field. The destructor in the shutdown sequence is
already a critical section, protected by the struct mutex.

Signed-off-by: Aitor Perez Cedres <[email protected]>

---------

Signed-off-by: Aitor Perez Cedres <[email protected]>

Makefile

@@ -16,8 +16,8 @@ fmt: ## Run go fmt against code
 	go fmt ./...
 
 .PHONY: tests
-tests: ## Run all tests and requires a running rabbitmq-server
-	go test -cpu 1,2 -race -v -tags integration
+tests: ## Run all tests and requires a running rabbitmq-server. Use GO_TEST_FLAGS to add extra flags to go test
+	go test -race -v -tags integration $(GO_TEST_FLAGS)
 
 .PHONY: check
 check:

connection.go

@@ -912,13 +912,17 @@ func (c *Connection) openTune(config Config, auth Authentication) error {
 		return ErrCredentials
 	}
 
+	// Edge case that may race with c.shutdown()
+	// https://github.com/rabbitmq/amqp091-go/issues/170
+	c.m.Lock()
 	// When the server and client both use default 0, then the max channel is
 	// only limited by uint16.
 	c.Config.ChannelMax = pick(config.ChannelMax, int(tune.ChannelMax))
 	if c.Config.ChannelMax == 0 {
 		c.Config.ChannelMax = defaultChannelMax
 	}
 	c.Config.ChannelMax = min(c.Config.ChannelMax, maxChannelMax)
+	c.m.Unlock()
 
 	// Frame size includes headers and end byte (len(payload)+8), even if
 	// this is less than FrameMinSize, use what the server sends because the
@@ -974,7 +978,9 @@ func (c *Connection) openComplete() error {
 		_ = deadliner.SetDeadline(time.Time{})
 	}
 
+	c.m.Lock()
 	c.allocator = newAllocator(1, c.Config.ChannelMax)
+	c.m.Unlock()
 	return nil
 }