feat: configurable maximum replication message size

[makarski]

Summary

Makes the largest accepted replication message configurable. Today the worker accepts any message up to the hard-coded MAX_MESSAGE_SIZE (1 GiB) and sizes its payload buffer from the length the server announces, so a single message can take up to 1 GiB of memory before the consumer sees it. A consumer running with a fixed memory budget has no way to lower that bound.

Changes

• ReplicationConfig::max_message_size and with_max_message_size(): the largest message the worker accepts. A message over the limit ends the stream with a protocol error, before any of its payload is buffered.
• MessageReader::with_max_message_size(): the same limit for direct use of the reader. The streaming loop builds its reader with the configured limit.
• The default stays MAX_MESSAGE_SIZE, so behavior is unchanged unless the limit is set.
• CHANGELOG entry under Unreleased.

Tests

• message_reader_rejects_a_message_over_its_configured_limit: a 10-byte payload against a limit of 8 is refused, and the error names the limit.
• message_reader_accepts_a_message_at_its_configured_limit: a payload exactly at the limit is read.
• max_message_size_defaults_to_the_protocol_limit_and_builder_sets_it: config default and builder.
• A doc test on with_max_message_size().

Setting the limit below the size of the largest row or logical message the publication can produce will end the stream on that message, which the field's documentation says.

[Copilot]

Copilot review overview

🟡 Changes recommended

The new configuration currently permits increasing the limit above the existing hard safety cap, which can reintroduce memory-exhaustion risk from malformed length fields.

Get a fresh assessment by requesting another Copilot review.

Review effort: Lite
Findings: 2

Open (2)

• Cap replication message size at the protocol limit · New
• Clamp configured message size to the protocol safety cap · New

What changed in this PR

This PR adds a configurable upper bound on the size of a single replication/backend message that the worker (and MessageReader) will accept, so consumers can cap worst-case per-message memory usage instead of relying on the hard-coded 1 GiB default.

Changes:

• Add ReplicationConfig::max_message_size plus with_max_message_size() builder, defaulting to MAX_MESSAGE_SIZE (1 GiB).
• Add MessageReader::with_max_message_size() and wire the worker’s streaming loop to use the configured limit.
• Add tests for the new size-limit behavior and document the change in CHANGELOG.md.

File	Description
src/​protocol/​framing.rs	Introduces a configurable max payload size in MessageReader and adds unit tests for limit enforcement.
src/​config.rs	Adds max_message_size to ReplicationConfig, a builder method, and config tests/doc example.
src/​client/​worker.rs	Applies the configured max message size to the streaming loop’s MessageReader.
CHANGELOG.md	Documents the new configuration option under Unreleased.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[makarski]

Agreed, fixed in 6c9206a (reader side in f35dbd8): ReplicationConfig::with_max_message_size() and MessageReader::with_max_message_size() now both cap the value at MAX_MESSAGE_SIZE, so the setting can only lower the limit. Covered by max_message_size_cannot_be_raised_above_the_protocol_limit and message_reader_limit_cannot_be_raised_above_the_protocol_cap.

[makarski]

Agreed, fixed in f35dbd8: both ReplicationConfig::with_max_message_size() and MessageReader::with_max_message_size() now cap the value at MAX_MESSAGE_SIZE, so the setting can only lower the limit. Covered by max_message_size_cannot_be_raised_above_the_protocol_limit and message_reader_limit_cannot_be_raised_above_the_protocol_cap.

[vnvo]

thanks for the PR, just one potential issue worth fixing:
MessageReader is left desynched after rejecting an oversized message

src/protocol/framing.rs:144-150

if a payload exceeds self.max_payload the reader resets hdr_filled = 0 and returns a Protocol error, but the payload bytes are never consumed from the stream.
they are still sitting there, the next call to read() takes its 5-byte header from the middle of the discarded payload.

the sibling reset at framing.rs:134-136 documents the intent:

// Reset so the reader is reusable after a protocol error is
// surfaced (callers typically tear down on this anyway).

[makarski]

Thanks, good catch. Fixed in 2568b16. When read() refuses an oversized message it now records the payload bytes still on the wire, and the next read() drops them first (in 8 KiB chunks, without buffering, and cancel-safely) before reading the next header. So the reader stays aligned and is reusable after the error, as the sibling reset intends. Covered by message_reader_stays_aligned_after_refusing_an_oversized_message and message_reader_skips_a_refused_payload_across_short_reads.