fix: Limit max bundle size (#2355)

## Why is this change needed?

`submitBulkMessages` and `submitMessageBundle` were not validating size
of the bundle and allowing unbounded input.

## Merge Checklist

_Choose all relevant options below by adding an `x` now or at any time
before submitting for review_

- [x] PR title adheres to the [conventional
commits](https://www.conventionalcommits.org/en/v1.0.0/) standard
- [x] PR has a
[changeset](https://github.com/farcasterxyz/hub-monorepo/blob/main/CONTRIBUTING.md#35-adding-changesets)
- [x] PR has been tagged with a change label(s) (i.e. documentation,
feature, bugfix, or chore)
- [ ] PR includes
[documentation](https://github.com/farcasterxyz/hub-monorepo/blob/main/CONTRIBUTING.md#32-writing-docs)
if necessary.


<!-- start pr-codex -->

---

## PR-Codex overview
This PR focuses on limiting the size of message bundles in the
application to enhance performance and prevent overload.

### Detailed summary
- Added a new export `MAX_BUNDLE_SIZE` in
`apps/hubble/src/network/p2p/bundleCreator.ts`.
- Implemented a check in `apps/hubble/src/hubble.ts` to reject message
bundles exceeding `MAX_BUNDLE_SIZE`.
- Added a similar check in `apps/hubble/src/rpc/server.ts` for gRPC
message submissions.

> ✨ Ask PR-Codex anything about this PR by commenting with `/codex {your
question}`

<!-- end pr-codex -->

.changeset/real-hounds-divide.md

@@ -0,0 +1,5 @@
+---
+"@farcaster/hubble": patch
+---
+
+fix: Limit message bundle size

apps/hubble/src/hubble.ts

@@ -98,6 +98,7 @@ import { diagnosticReporter } from "./utils/diagnosticReport.js";
 import { startupCheck, StartupCheckStatus } from "./utils/startupCheck.js";
 import { AddressInfo } from "node:net";
 import { MeasureSyncHealthJobScheduler } from "./network/sync/syncHealthJob.js";
+import { MAX_BUNDLE_SIZE } from "./network/p2p/bundleCreator.js";
 
 export type HubSubmitSource =
   | "gossip"
@@ -1867,6 +1868,16 @@ export class Hub implements HubInterface {
       );
     }
 
+    if (messageBundle.messages.length > MAX_BUNDLE_SIZE) {
+      log.warn(
+        { bundleSize: messageBundle.messages.length, maxBundleSize: MAX_BUNDLE_SIZE },
+        "submitMessageBundle rejected: Message bundle is too large",
+      );
+      return messageBundle.messages.map(() =>
+        err(new HubError("bad_request.invalid_param", "Message bundle is too large")),
+      );
+    }
+
     const start = Date.now();
     const allResults: Map<number, HubResult<number>> = new Map();
 

apps/hubble/src/network/p2p/bundleCreator.ts

@@ -2,7 +2,7 @@ import { Message, MessageBundle } from "@farcaster/hub-nodejs";
 import { GossipPublishResult, LibP2PNode } from "./gossipNodeWorker.js";
 import { blake3Truncate160 } from "../../utils/crypto.js";
 
-const MAX_BUNDLE_SIZE = 256;
+export const MAX_BUNDLE_SIZE = 256;
 const DEFAULT_BUNDLE_TIME_MS = 1000;
 
 export class BundleCreator {

apps/hubble/src/rpc/server.ts

@@ -86,6 +86,7 @@ import axios from "axios";
 import { fidFromEvent } from "../storage/stores/storeEventHandler.js";
 import { rustErrorToHubError } from "../rustfunctions.js";
 import { handleUnaryCall, sendUnaryData, ServerDuplexStream, ServerUnaryCall } from "@grpc/grpc-js";
+import { MAX_BUNDLE_SIZE } from "../network/p2p/bundleCreator.js";
 
 const HUBEVENTS_READER_TIMEOUT = 1 * 60 * 60 * 1000; // 1 hour
 const STREAM_METHODS_TIMEOUT = 8 * 1000; // 2 seconds
@@ -960,6 +961,16 @@ export default class Server {
           return;
         }
 
+        if (call.request.messages.length > MAX_BUNDLE_SIZE) {
+          logger.warn({ total: call.request.messages.length }, "gRPC submitBulkMessages received too many messages");
+          callback(
+            toServiceError(
+              new HubError("bad_request.validation_failure", `Too many messages. Max is ${MAX_BUNDLE_SIZE}`),
+            ),
+          );
+          return;
+        }
+
         const submissionTime = getFarcasterTime();
         if (submissionTime.isErr()) {
           callback(toServiceError(submissionTime.error));