feat(frontend): memory throttling for frontend messages

[zwang28]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

This PR is a follow-up to #20470. It aims to address the frontend OOM issue caused by a surge of large incoming queries, which may occur due to numerous concurrent DML or a few extremely large DML.

A MessageMemoryManager has been implemented in the frontend. It tracks memory usage of frontend messages and provides hint of whether this message would cause memory constraint violation. For any message of size n,

• If n <= min_filter_bytes, the message won't add to the memory constraint, because the message is too small that it can always be accepted. e.g. SELECT is usually small relative in size and should not be rejected.
• If n > max_filter_bytes, the message won't add to the memory constraint, because the message is too large that it's better to be rejected immediately.
• Otherwise, the message will add size n to the memory constraint. See MessageMemoryManager::add.

The frontend will follow MessageMemoryManager's hint and respond client with error immediately, without processing the message any further.

Tests similar to those in #20470 have demonstrated the effectiveness of this PR.

Checklist

• I have written necessary rustdoc comments.
• I have added necessary unit tests and integration tests.
• I have added test labels as necessary.
• I have added fuzzing tests or opened an issue to track them.
• My PR contains breaking changes.
• My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.
• My PR contains critical fixes that are necessary to be merged into the latest release.

Documentation

• My PR needs documentation updates.

Release note

PR Overview

This PR implements memory throttling for frontend messages to mitigate OOM issues by introducing a MessageMemoryManager in the frontend and propagating its usage through the pgwire protocol. Key changes include:

• Adding MessageMemoryManager and MessageMemoryGuard to track and limit memory usage for messages.
• Updating pgwire protocol and message handling to integrate memory throttling and server throttle responses.
• Extending configuration and documentation to include new memory-related thresholds.

Reviewed Changes

File	Description
src/utils/pgwire/src/pg_protocol.rs	Added MessageMemoryManager, integrated memory management in read_message, and adjusted error handling for throttling.
src/utils/pgwire/src/pg_message.rs	Introduced new message types and refactored header/body reading functions.
src/common/src/config.rs	Added new fields to configure frontend message memory limits.
src/frontend/src/lib.rs	Updated connection context creation to include MessageMemoryManager.
src/utils/pgwire/src/error.rs	Added ServerThrottle error variant.
src/config/example.toml	Updated example configuration with memory thresholds.
src/frontend/src/scheduler/local.rs	Adjusted drop order for memory release in query execution.
src/config/docs.md	Documented the new memory-related configuration options.
src/utils/pgwire/src/pg_server.rs	Updated connection setup to use the new ConnectionContext.

Copilot reviewed 9 out of 9 changed files in this pull request and generated no comments.

Comments suppressed due to low confidence (3)

src/utils/pgwire/src/pg_protocol.rs:111

• Consider whether using Ordering::Relaxed is sufficient for concurrent updates to current_running_bytes. If ordering issues could arise, a stronger memory ordering might be necessary to ensure consistency.

            .fetch_add(bytes, Ordering::Relaxed);

src/utils/pgwire/src/pg_protocol.rs:597

• [nitpick] Enhance the error message by including the actual message size (or payload size) along with the limit to provide clearer context for troubleshooting throttling issues.

return Err(PsqlError::ServerThrottle(anyhow::anyhow!(format!("frontend_throttling_filter_max_bytes {} has been exceeded, please either reduce the message size or increase the limit", self.message_memory_manager.max_filter_bytes)).into()));

src/frontend/src/scheduler/local.rs:122

• [nitpick] Verify that the new drop order—dropping plan_node before dropping self—is intentional and does not inadvertently affect the lifespan of large objects or resource cleanup.

drop(self);

[BugenZhao]

Suggested change

	return Err(PsqlError::ServerThrottle(anyhow::anyhow!(format!("frontend_throttling_filter_max_bytes {} has been exceeded, please either reduce the message size or increase the limit", self.message_memory_manager.max_filter_bytes)).into()));
	return Err(PsqlError::ServerThrottle(anyhow::anyhow!("frontend_throttling_filter_max_bytes {} has been exceeded, please either reduce the message size or increase the limit", self.message_memory_manager.max_filter_bytes).into()));