Azure / Azure/azure-sdk-for-rust
[Event Hubs] send_event does not enforce the AMQP link maximum that batches enforce
- Dominant language
- Rust
- Stars
- 884
- Forks
- 365
- Avg merge
- 2d 19h
- Merged PRs (30d)
- 112
Description
## Summary
`ProducerClient::send_event` accepts an event that is larger than the AMQP link maximum, and the service stores it. `EventDataBatch::try_add_event_data` refuses the same event. The two publish paths disagree about the same limit.
## Motivation
A live run against a Standard namespace read the link maximum as `max_allowed: 1048576` bytes. A single 2 MiB event sent with `send_event` returned `Ok(())`, and the partition tail then moved by one, so the event reached the partition. The batch path returned `Ok(false)` for the same body. A caller that publishes one event at a time therefore gets no size protection from the client, the behavior depends on what the broker chooses to accept, and it differs from the batch path in the same crate. The .NET client rejects an oversized publication on both paths.
## Proposal
- Compare the encoded message size against the sender link maximum inside `send_event`, the same comparison `EventDataBatch::try_add_event_data` already makes.
- Return an error that names the requested size and the maximum, so the message matches the existing `ErrorKind::InvalidBatchSize { requested, max_allowed }`.
- Keep the maximum readable from one place, because the batch path and the single event path must not drift again.
### Reproduction
Read the maximum from the batch error, then send one event above it.
```rust
let error = producer.create_batch(Some(EventDataBatchOptions {
max_size_in_bytes: Some(512 * 1024 * 1024),
partition_id: Some("0".to_string()),
..Default::default()
})).await.err().unwrap();
// error.kind == InvalidBatchSize { requested: 536870912, max_allowed: 1048576 }
let before = producer.get_partition_properties("0").await?.last_enqueued_sequence_number;
producer.send_event(
EventData::builder().with_body(vec![0xABu8; 2 * 1024 * 1024]).build(),
Some(SendEventOptions { partition_id: Some("0".to_string()) }),
).await?; // returns Ok
let after = producer.get_partition_properties("0").await?.last_enqueued_sequence_number;
assert_eq!(after, before); // fails: the tail moved, so the event landed
```
## Why the client must do this check itself
The AMQP layer below never refuses the message. In `fe2o3-amqp` 0.14.0, which `Cargo.lock` pins, `src/link/sender_link.rs:38` computes
```rust
let more = (self.max_message_size != 0) && (payload.len() as u64 > self.max_message_size);
```
and when `more` is true it splits the payload into several transfer frames and sends them all. It treats `max_message_size` as a fragmentation boundary rather than as a limit, so an oversized message is transmitted instead of rejected. That is why the 2 MiB event reached the partition rather than raising `amqp:link:message-size-exceeded`.
One related detail that matters for the check. `max_message_size()` maps a reported `0` to `None` at `sender_link.rs:644`, and the Event Hubs attach passes no sender options, so the effective maximum is whatever the broker sends. `None` therefore means no limit, and the client must send in that case rather than refuse.
Whether `fe2o3-amqp` is right to fragment on `max-message-size` is a separate question for that project, and this issue does not depend on the answer. Even if the layer below starts rejecting, a clear client side error that names the size and the limit is better than a link detach.
Contributor guide
Assessment
This issue has not been assessed yet.