feat(amqp10): implement exchange declaration

[sharathb5]

Implement AMQP 1.0 exchange declaration through the public Subscribe path,
track connection-scoped exchanges, preserve redeclaration race handling, and
add unit and integration coverage.

[sharathb5]

Comment from #226: Remove knownExchanges, knownQueues, and knownBindings because track.EntityTracker replaces them.

BrokerDetails previously had three separate concurrent maps for exchanges, queues, and bindings. I confirmed that these fields no longer had any consumers in the AMQP 1.0 package.

I removed all three fields. Connection-scoped entity state now uses:

Go
knownEntities *track.EntityTracker
Show more lines

I did not add binding tracking because the old knownBindings map was unused and EntityTracker does not currently track bindings.

Status: Addressed in feature-amqp10 through e9c5bc0.

[sharathb5]

Comment from #226: Remove initialization of the legacy entity maps in connect().

Connection setup previously created separate concurrent maps for exchanges, queues, and bindings. Those initializers were no longer needed after moving to EntityTracker.

I removed those initializers. Connection setup now initializes the shared tracker with:

bd.knownEntities = track.New()

Status: Addressed in feature-amqp10 through e9c5bc0.

[sharathb5]

Comment from #226: Standardize on “exists” terminology because that is what the tracker API uses.

The helper previously used the name exchangeKnown, while the tracker method is ExchangeExists.

I renamed the helper to:

func (bd *BrokerDetails) exchangeExists(name string) bool {
return bd.entityTracker().ExchangeExists(name)
}

I also updated the declareExchange call sites to use exchangeExists.

Status: Addressed in feature-amqp10 through e9c5bc0.

[sharathb5]

Comment from #226: How are the remaining AMQP 0.9.1 exchange declaration arguments handled by the AMQP 1.0 specification?

I checked the pinned RabbitMQ AMQP 1.0 client implementation.

The Arke code supplies:

the exchange name through Name
the exchange type through the selected concrete specification
auto-delete through IsAutoDelete
optional arguments through Arguments

The client itself always sends durable: true. The AMQP 1.0 management API does not expose internal or noWait. The internal field is omitted, leaving RabbitMQ’s normal false default, and noWait does not apply to this management request. When Arguments is nil, the client omits the arguments field.

Based on that behavior, no additional Arke argument mapping was needed.

Status: Investigated and answered. No code change was required

[sharathb5]

Comment from #226: Does ErrPreconditionFailed mean the exchange already existed or something else?

ErrPreconditionFailed does not merely mean that an exchange with the name already exists. The client returns this error for an HTTP 409 when an existing exchange has incompatible settings.

A compatible redeclaration is accepted as successful. The implementation therefore returns ErrPreconditionFailed directly rather than allowing the post-error tracker check to hide a real configuration conflict.

The relevant behavior is:

if errors.Is(err, rabbitmqamqp.ErrPreconditionFailed) {
return err
}

Status: Investigated and preserved in the current implementation.

[sharathb5]

Comment from #226: Add an integration test that creates an exchange so the AMQP 1.0 behavior can be compared with AMQP 0.9.1.

I added a temporary package-local integration test named:

TestAMQP10DeclareExchangeIntegration

The test:

connects to the configured RabbitMQ broker
creates a uniquely named topic exchange
calls declareExchange a second time
deletes the exchange during cleanup

I also moved Test_ProviderConnectsToBroker into a separate integration-tagged file, as suggested. The moved test body was not changed.

The integration-tagged tests compile locally. I did not claim a live-broker result because the RabbitMQ integration environment was not available locally.

Status: Addressed in feature-amqp10 through e9c5bc0.

[sharathb5]

For context, PR #226 was accidentally pushed directly to feature-amqp10, so the implementation and review updates are already in the base branch and do not appear in #227’s diff. PR #227 contains the later CI fixes. I’m copying the #226 comments here individually so reviewers can verify how each was handled.

[rsperl]

Comment from #226: Does ErrPreconditionFailed mean the exchange already existed or something else?

ErrPreconditionFailed does not merely mean that an exchange with the name already exists. The client returns this error for an HTTP 409 when an existing exchange has incompatible settings.

A compatible redeclaration is accepted as successful. The implementation therefore returns ErrPreconditionFailed directly rather than allowing the post-error tracker check to hide a real configuration conflict.

The relevant behavior is:

if errors.Is(err, rabbitmqamqp.ErrPreconditionFailed) { return err }

Status: Investigated and preserved in the current implementation.

We have a test that ensures that if a consumer declares an existing entity with different parameters, we ignore that particular kind of error:

arke/internal/provider/connectors/amqp091/amqp091_test.go

Line 2500 in 61ae814

func Test_IgnoreDeclareError(t *testing.T) {

It may be that we should not ignore all precondition failed, but if it's a redeclaration with different args, we do ignore that error.

[rsperl]

Comment from #226: Add an integration test that creates an exchange so the AMQP 1.0 behavior can be compared with AMQP 0.9.1.

I added a temporary package-local integration test named:

TestAMQP10DeclareExchangeIntegration

The test:

connects to the configured RabbitMQ broker
creates a uniquely named topic exchange
calls declareExchange a second time
deletes the exchange during cleanup

I also moved Test_ProviderConnectsToBroker into a separate integration-tagged file, as suggested. The moved test body was not changed.

The integration-tagged tests compile locally. I did not claim a live-broker result because the RabbitMQ integration environment was not available locally.

Status: Addressed in feature-amqp10 through e9c5bc0.

You can run make compose_only to spin up arke and rabbitmq in docker locally. You can then run the integration tests against them. You may have to build arke manually if you are on mac arm64:

GOOS=linux go build -o build/linux/arke ./cmd
make compose_only
cd tests/integration
ARKE_BROKER_TYPE=rabbitmq-amqp10 go test -count=1 -tags=integration -run Test_ProviderConnectsToBroker ./...

[rsperl]

All integration tests should go in ./tests/integration/*_test.go.

[rsperl]

I don't see an implementation of the rabbitMQAMQP10ManagementShim interface.

[rsperl]

In amqp091, an exchange is declared with an amqp channel with this signature:

// amqp091
func (ch *amqp.Channel) ExchangeDeclare(name string, kind string, durable bool, autoDelete bool, internal bool, noWait bool, args amqp.Table) error

In amqp10, a management client class DeclareExchange with this signature:

// amqp10
func (a *AmqpManagement) DeclareExchange(ctx context.Context, exchangeSpecification IExchangeSpecification) (*AmqpExchangeInfo, error)

When I compare how this parameters map, I see

param	amqp091	amqp10
name	ExchangeDeclare arg	IExchangeSpecification
type	ExchangeDeclare arg	IExchangeSpecification
durable	ExchangeDeclare arg	IExchangeSpecification
autodelete	ExchangeDeclare arg	IExchangeSpecification
internal	ExchangeDeclare arg	not exposed, but doesn't matter
noWait	ExchangeDeclare arg	??
args (not used by us)	ExchangeDeclare arg	IExchangeSpecification

You said noWait doesn't apply to this management request -- this parameter says to not return until the exchange is created. Do we know for sure that the mgmt request in amqp10 does not return until the exchange is created?

[sharathb5]

I don't see an implementation of the rabbitMQAMQP10ManagementShim interface.

The shim implementationis *rabbitmqamqp.AmqpManagement, which is returned by the connection’s Management() method. Since it was implicit, I added a compile-time assertion to make it explicit and ensure the client continues to satisfy the interface.

[sharathb5]

All integration tests should go in ./tests/integration/*_test.go.

I removed the integration test from the amqp10 package so the integration tests stay under tests/integration. Since declareExchange is private and none of the public AMQP 1.0 provider methods call it yet, moving that exact test would mean either exporting internal behavior or adding a test-only production hook, so I left Test_ProviderConnectsToBroker under tests/integration and verified that it passes against the local environment.

[sharathb5]

In amqp091, an exchange is declared with an amqp channel with this signature:

// amqp091
func (ch *amqp.Channel) ExchangeDeclare(name string, kind string, durable bool, autoDelete bool, internal bool, noWait bool, args amqp.Table) error

In amqp10, a management client class DeclareExchange with this signature:

// amqp10
func (a *AmqpManagement) DeclareExchange(ctx context.Context, exchangeSpecification IExchangeSpecification) (*AmqpExchangeInfo, error)

When I compare how this parameters map, I see

param amqp091 amqp10
name ExchangeDeclare arg IExchangeSpecification
type ExchangeDeclare arg IExchangeSpecification
durable ExchangeDeclare arg IExchangeSpecification
autodelete ExchangeDeclare arg IExchangeSpecification
internal ExchangeDeclare arg not exposed, but doesn't matter
noWait ExchangeDeclare arg ??
args (not used by us) ExchangeDeclare arg IExchangeSpecification
You said noWait doesn't apply to this management request -- this parameter says to not return until the exchange is created. Do we know for sure that the mgmt request in amqp10 does not return until the exchange is created?

I looked this through the pinned client implementation. DeclareExchange sends a correlated management request and waits for the matching RabbitMQ response before returning successfully. The response is then accepted and its status is validated. So although the AMQP 1.0 API does not expose a separate noWait option, the operation is still request-response and does not return immediately after sending the declaration.

[sharathb5]

Comment from #226: Does ErrPreconditionFailed mean the exchange already existed or something else?
ErrPreconditionFailed does not merely mean that an exchange with the name already exists. The client returns this error for an HTTP 409 when an existing exchange has incompatible settings.
A compatible redeclaration is accepted as successful. The implementation therefore returns ErrPreconditionFailed directly rather than allowing the post-error tracker check to hide a real configuration conflict.
The relevant behavior is:
if errors.Is(err, rabbitmqamqp.ErrPreconditionFailed) { return err }
Status: Investigated and preserved in the current implementation.

We have a test that ensures that if a consumer declares an existing entity with different parameters, we ignore that particular kind of error:

arke/internal/provider/connectors/amqp091/amqp091_test.go

Line 2500 in 61ae814

func Test_IgnoreDeclareError(t *testing.T) {

It may be that we should not ignore all precondition failed, but if it's a redeclaration with different args, we do ignore that error.

I compared this with the AMQP 0.9.1 behavior. AMQP 0.9.1 can ignore only the specific case where the error contains both PRECONDITION_FAILED and inequivalent arg.

The AMQP 1.0 client converts the relevant HTTP 409 responses into the same ErrPreconditionFailed value and does not preserve the broker’s error details. Because of that, there isn’t a safe way to distinguish an inequivalent redeclaration from another precondition failure. I left the error propagated rather than ignoring every precondition failure.

[sharathb5]

Comment from #226: Add an integration test that creates an exchange so the AMQP 1.0 behavior can be compared with AMQP 0.9.1.
I added a temporary package-local integration test named:
TestAMQP10DeclareExchangeIntegration
The test:
connects to the configured RabbitMQ broker
creates a uniquely named topic exchange
calls declareExchange a second time
deletes the exchange during cleanup
I also moved Test_ProviderConnectsToBroker into a separate integration-tagged file, as suggested. The moved test body was not changed.
The integration-tagged tests compile locally. I did not claim a live-broker result because the RabbitMQ integration environment was not available locally.
Status: Addressed in feature-amqp10 through e9c5bc0.

You can run make compose_only to spin up arke and rabbitmq in docker locally. You can then run the integration tests against them. You may have to build arke manually if you are on mac arm64:

GOOS=linux go build -o build/linux/arke ./cmd
make compose_only
cd tests/integration
ARKE_BROKER_TYPE=rabbitmq-amqp10 go test -count=1 -tags=integration -run Test_ProviderConnectsToBroker ./...

I ran the local Compose setup and verified that Test_ProviderConnectsToBroker passes with ARKE_BROKER_TYPE=rabbitmq-amqp10.

I also removed the package-local declaration integration test so all integration-tagged tests remain under tests/integration. Since declareExchange is package-private and no current public AMQP 1.0 provider operation calls it, moving that exact test into tests/integration would require exporting internal behavior or adding a production test hook. I left the public connection integration test in place for now.

[rsperl]

I don't see an implementation of the rabbitMQAMQP10ManagementShim interface.

The shim implementationis *rabbitmqamqp.AmqpManagement, which is returned by the connection’s Management() method. Since it was implicit, I added a compile-time assertion to make it explicit and ensure the client continues to satisfy the interface.

Ok - I see where I missed it. You can remove the assertion. I was used to seeing a shim, then the shim implementation, but that's not actually needed.

[rsperl]

Comment from #226: Add an integration test that creates an exchange so the AMQP 1.0 behavior can be compared with AMQP 0.9.1.
I added a temporary package-local integration test named:
TestAMQP10DeclareExchangeIntegration
The test:
connects to the configured RabbitMQ broker
creates a uniquely named topic exchange
calls declareExchange a second time
deletes the exchange during cleanup
I also moved Test_ProviderConnectsToBroker into a separate integration-tagged file, as suggested. The moved test body was not changed.
The integration-tagged tests compile locally. I did not claim a live-broker result because the RabbitMQ integration environment was not available locally.
Status: Addressed in feature-amqp10 through e9c5bc0.

You can run make compose_only to spin up arke and rabbitmq in docker locally. You can then run the integration tests against them. You may have to build arke manually if you are on mac arm64:

GOOS=linux go build -o build/linux/arke ./cmd
make compose_only
cd tests/integration
ARKE_BROKER_TYPE=rabbitmq-amqp10 go test -count=1 -tags=integration -run Test_ProviderConnectsToBroker ./...

I ran the local Compose setup and verified that Test_ProviderConnectsToBroker passes with ARKE_BROKER_TYPE=rabbitmq-amqp10.

I also removed the package-local declaration integration test so all integration-tagged tests remain under tests/integration. Since declareExchange is package-private and no current public AMQP 1.0 provider operation calls it, moving that exact test into tests/integration would require exporting internal behavior or adding a production test hook. I left the public connection integration test in place for now.

In the Provider.Subscribe stub, you could add the declareExchange, then in the integration test call prov.Subscribe.

[rsperl]

Comment from #226: Does ErrPreconditionFailed mean the exchange already existed or something else?
ErrPreconditionFailed does not merely mean that an exchange with the name already exists. The client returns this error for an HTTP 409 when an existing exchange has incompatible settings.
A compatible redeclaration is accepted as successful. The implementation therefore returns ErrPreconditionFailed directly rather than allowing the post-error tracker check to hide a real configuration conflict.
The relevant behavior is:
if errors.Is(err, rabbitmqamqp.ErrPreconditionFailed) { return err }
Status: Investigated and preserved in the current implementation.

We have a test that ensures that if a consumer declares an existing entity with different parameters, we ignore that particular kind of error:

arke/internal/provider/connectors/amqp091/amqp091_test.go

Line 2500 in 61ae814

func Test_IgnoreDeclareError(t *testing.T) {

It may be that we should not ignore all precondition failed, but if it's a redeclaration with different args, we do ignore that error.

I compared this with the AMQP 0.9.1 behavior. AMQP 0.9.1 can ignore only the specific case where the error contains both PRECONDITION_FAILED and inequivalent arg.

The AMQP 1.0 client converts the relevant HTTP 409 responses into the same ErrPreconditionFailed value and does not preserve the broker’s error details. Because of that, there isn’t a safe way to distinguish an inequivalent redeclaration from another precondition failure. I left the error propagated rather than ignoring every precondition failure.

We need to figure out a way to capture this. This would be a regression to not allow it that particular error to pass.

[bithckr]

I think if we check to see if the exchange exists before we declare, then most of the following code can go away. If the exchange does exist, I think we should add it to the tracker.

[sharathb5]

I checked, but the client doesn’t have a broker-side exchange lookup. Adding one through the HTTP API would add another dependency and still leave a race between checking and declaring. I think the current declare-first flow is safer unless there’s another API you had in mind.

[rsperl]

I'm not sure what "client doesn’t have a broker-side exchange lookup" means.
I'm already having to use the http mgmt api for source stats.

[sharathb5]

ok I think I misunderstood, I was treating the HTTP mgmt API as a new dependency, but it’s already scoped for SourceStats. Since we are doing that later, I’ll keep the current declare-first flow

[rsperl]

I wonder if rabbitMQAMQP10ManagementShim should go away and just add its one method DeclareExchange to the connection, since the management interface is tied to the amqp10 connection, and is fairly limited in what it can do.

Or should we have an implementation rabbitMQAMQP10Management that uses the real management connection but doesn't expose it (maybe it just returns an error).

[sharathb5]

ill make a quick commit to just add it to DeclareExchange lmk what you think

[bithckr]

Comment from #226: How are the remaining AMQP 0.9.1 exchange declaration arguments handled by the AMQP 1.0 specification?

I checked the pinned RabbitMQ AMQP 1.0 client implementation.

The Arke code supplies:

the exchange name through Name the exchange type through the selected concrete specification auto-delete through IsAutoDelete optional arguments through Arguments

The client itself always sends durable: true. The AMQP 1.0 management API does not expose internal or noWait. The internal field is omitted, leaving RabbitMQ’s normal false default, and noWait does not apply to this management request. When Arguments is nil, the client omits the arguments field.

Based on that behavior, no additional Arke argument mapping was needed.

Status: Investigated and answered. No code change was required

We should consider deprecating Address.auto_delete. I don't remember why we introduced it.