Bound API submissions and proposer preparations

crates/beacon_api/src/beacon/operations.rs

@@ -295,7 +295,7 @@ pub(crate) mod tests {
     use crate::{
         http::router::Outcome,
         submission::{
-            SubmissionFailure,
+            MAX_SUBMISSION_ENTRIES, SubmissionFailure,
             tests::{SLOT, ctx, posted_committees, validator_signature},
         },
         testing::{body, dispatch, dispatch_into, posting, status_code, submissions},
@@ -402,6 +402,22 @@ pub(crate) mod tests {
         assert_eq!(failures[0]["message"], "attester_index is past the validator count");
     }
 
+    #[test]
+    fn body_past_the_submission_entry_cap_is_a_400() {
+        let ctx = ctx();
+        let at_cap = format!("[{}]", vec![entry(1, 0); MAX_SUBMISSION_ENTRIES].join(","));
+        assert!(matches!(submit(&ctx, &at_cap).0, Outcome::AwaitingVerdicts(_)));
+
+        let past_cap = format!("[{}]", vec![entry(1, 0); MAX_SUBMISSION_ENTRIES + 1].join(","));
+        let (outcome, response) = submit(&ctx, &past_cap);
+        assert_eq!(outcome, Outcome::Response(None));
+        assert_eq!(status_code(&response), "400");
+        assert_eq!(
+            body(&response),
+            b"{\"code\":400,\"message\":\"too many entries in request body\"}"
+        );
+    }
+
     /// Entries the posted shuffling cannot place are reported by their place
     /// in the body, and the rest of the submission still goes out.
     #[test]

crates/beacon_api/src/server.rs

@@ -39,8 +39,8 @@ use crate::{
     node::peers::Peer,
     routes::{ROUTES, SERVED_WHILE_SYNCING},
     submission::{
-        AcceptedEntry, BuilderForward, Integration, RecentImports, SubmissionFailure,
-        failure_message,
+        AcceptedEntry, BuilderForward, Integration, MAX_SUBMISSION_ENTRIES, RecentImports,
+        Submission, SubmissionFailure, failure_message,
     },
     validator::{
         aggregate_attestation::AggregateRequest, execution_payload_envelope::EnvelopeRequest,
@@ -54,6 +54,10 @@ const MAX_SWEEP_INTERVAL: Duration = Duration::from_secs(1);
 // that matter, such as blocks, are useless after one 12 s slot anyway.
 const REQUEST_DEADLINE: Duration = Duration::from_secs(60);
 
+/// Two of the largest bodies at once: half the `beacon_api_requests` queue,
+/// leaving the rest to the blocks and requests that share it.
+const MAX_OUTSTANDING_SUBMISSIONS: usize = 2 * MAX_SUBMISSION_ENTRIES;
+
 /// nginx's `lingering_close` caps: how long a connection that has already
 /// answered may wait between the peer's bytes, and how long the whole drain
 /// may run before the slot is taken back.
@@ -1530,14 +1534,23 @@ impl BeaconApi {
                         self.want_payload_attributes(true, emit);
                     }
                 }
+                Outcome::AwaitingVerdicts(submission) if !self.admits(&submission) => {
+                    silver_log::warn!(
+                        entries = submission.accepted.len(),
+                        "submission refused: too many entries awaiting their verdicts"
+                    );
+                    let requests = self.deferring(token);
+                    let mut resp = Response::new(requests.http.write_buf_mut(), submissions);
+                    resp.error(503, "too many submissions awaiting publication, retry shortly");
+                    self.resume_writing(token);
+                }
                 deferred => {
                     let (pending, ids_used) = Pending::defer(deferred, self.next_request_id, emit);
                     self.next_request_id += ids_used;
                     if matches!(&pending, Pending::Submission(p) if p.integration.awaits_import()) {
                         self.awaiting_import.push(token);
                     }
-                    let requests = conn.requests_mut().expect("only a request handler defers");
-                    requests.pending = Some(pending);
+                    self.deferring(token).pending = Some(pending);
                 }
             },
             Ok(true) => {
@@ -1553,6 +1566,33 @@ impl BeaconApi {
         true
     }
 
+    fn deferring(&mut self, token: Token) -> &mut Requests {
+        self.connections
+            .get_mut(&token)
+            .and_then(Connection::requests_mut)
+            .expect("only a request handler defers")
+    }
+
+    /// A body of one entry is admitted past the budget: the connection cap
+    /// bounds those, and a refused block would cost its proposal.
+    fn admits(&self, submission: &Submission) -> bool {
+        submission.body_len() == 1 ||
+            self.outstanding_submissions() + submission.accepted.len() <=
+                MAX_OUTSTANDING_SUBMISSIONS
+    }
+
+    /// Counted from the connections' own waits, so a connection that closes
+    /// or times out gives its share back with it.
+    fn outstanding_submissions(&self) -> usize {
+        self.connections
+            .values()
+            .filter_map(|conn| match conn.pending()? {
+                Pending::Submission(pending) => Some(pending.unanswered),
+                _ => None,
+            })
+            .sum()
+    }
+
     /// Connections close in any order while the cursor only advances, so the
     /// offset it lands on may still be held. The range holds every socket the
     /// server can register at once, so probing forward ends on a free one.
@@ -3855,6 +3895,70 @@ mod tests {
         );
     }
 
+    const ATTESTATIONS: &str = "/eth/v2/beacon/pool/attestations";
+
+    fn post_attestations(server: &mut Server, count: usize) -> TcpStream {
+        let client = connect(tcp_addr(server));
+        let body = format!("[{}]", vec![entry(1, 0); count].join(","));
+        let writer = client.try_clone().unwrap();
+        std::thread::spawn(move || post(writer, ATTESTATIONS, &body));
+        client
+    }
+
+    fn submitted_after(server: &mut Server, count: usize, msg: &str) {
+        pump_until(server, msg, |s| s.requests.len() == count);
+    }
+
+    fn refused(server: &mut Server, count: usize) -> Vec<u8> {
+        let client = post_attestations(server, count);
+        let response = serve(server, std::thread::spawn(move || read_to_eof(client)), "refusal");
+        assert!(response.starts_with(b"HTTP/1.1 503 Service Unavailable\r\n"), "{response:?}");
+        response
+    }
+
+    /// A body's share of the budget comes back when its entries are answered
+    /// or when its connection is reaped.
+    #[test]
+    fn submissions_past_the_outstanding_budget_are_answered_503() {
+        let mut server = server_with(64, LONG_TIMEOUT);
+        server.api.ctx = submission::ctx();
+        server.submissions = TCache::producer(TCacheId::BoundaryProcessing, 1 << 24);
+
+        let first = post_attestations(&mut server, MAX_SUBMISSION_ENTRIES);
+        submitted_after(&mut server, MAX_SUBMISSION_ENTRIES, "the first body");
+        let second = post_attestations(&mut server, MAX_SUBMISSION_ENTRIES);
+        submitted_after(&mut server, MAX_OUTSTANDING_SUBMISSIONS, "the second body");
+
+        let response = refused(&mut server, 2);
+        assert!(body(&response).starts_with(b"{\"code\":503,"));
+        assert_eq!(server.requests.len(), MAX_OUTSTANDING_SUBMISSIONS, "none of it was sent");
+
+        let _lone = post_attestations(&mut server, 1);
+        submitted_after(&mut server, MAX_OUTSTANDING_SUBMISSIONS + 1, "a body of one entry");
+
+        let first_ids: Vec<_> =
+            server.requests.drain(..MAX_SUBMISSION_ENTRIES).map(|r| request_id(&r)).collect();
+        for request_id in first_ids {
+            server.answer(BeaconApiResponse::LocalGossipResponse {
+                request_id,
+                response: Ok(()),
+                block_root: None,
+            });
+        }
+        let response = serve(&mut server, std::thread::spawn(move || read_to_eof(first)), "200");
+        assert!(response.starts_with(b"HTTP/1.1 200 OK\r\n"), "{response:?}");
+        let _admitted = post_attestations(&mut server, 2);
+        submitted_after(&mut server, MAX_SUBMISSION_ENTRIES + 3, "a body within the budget");
+
+        refused(&mut server, MAX_SUBMISSION_ENTRIES);
+        drop(second);
+        server.api.idle.timeout = Duration::ZERO;
+        pump_until(&mut server, "idle connections reaped", |s| s.api.connections.is_empty());
+        server.api.idle.timeout = LONG_TIMEOUT;
+        let _largest = post_attestations(&mut server, MAX_SUBMISSION_ENTRIES);
+        submitted_after(&mut server, 2 * MAX_SUBMISSION_ENTRIES + 3, "the largest body");
+    }
+
     #[test]
     fn entry_published_on_several_topics_takes_one_request_id() {
         let ssz = submissions().write_with(1, |_| {}).unwrap();

crates/beacon_api/src/submission.rs

@@ -4,15 +4,20 @@ use serde::Deserialize;
 use silver_beacon_state_data::{B256, Slot, SpecConfig};
 use silver_common::{
     BroadcastValidation, GossipTopic, LocalGossipFailure, PayloadValidationStatus, TCacheRead,
-    ssz_hash::HashedBody,
+    json_fields::each_array_entry, ssz_hash::HashedBody,
 };
 
 use crate::{
     ctx::ApiCtx,
     entry_signatures::SignatureCheck,
-    http::{ids::each_body_entry, response::Response, router::Request},
+    http::{response::Response, router::Request},
 };
 
+/// Each accepted entry is a message per topic on the 2^14-deep
+/// `beacon_api_requests` queue. A body of single-topic entries is a quarter of
+/// it at most: one slot's attestations for 131,072 validators.
+pub(crate) const MAX_SUBMISSION_ENTRIES: usize = 4096;
+
 pub(crate) fn post_submission<'a, T: Deserialize<'a> + SubmittedEntry>(
     req: &Request<'a>,
     ctx: &ApiCtx,
@@ -21,8 +26,9 @@ pub(crate) fn post_submission<'a, T: Deserialize<'a> + SubmittedEntry>(
     if !reads_submission(req, resp) {
         return;
     }
-    let parsed =
-        each_body_entry(req.body, |body_index, entry: T| submit(ctx, resp, body_index, entry));
+    let parsed = each_array_entry(req.body, MAX_SUBMISSION_ENTRIES, |body_index, entry: T| {
+        submit(ctx, resp, body_index, entry)
+    });
     match parsed {
         Err(message) => return resp.error(400, message),
         Ok(0) => return resp.error(400, "the body must name at least one entry"),

crates/beacon_api/src/validator/registration.rs

@@ -33,26 +33,37 @@ pub(crate) fn post_register_validator(req: &Request<'_>, ctx: &ApiCtx, resp: &mu
     resp.await_sidecar(SidecarCall::RegisterValidators);
 }
 
+/// An index past the registry is dropped, as Lighthouse drops it: it cannot
+/// propose within the epochs a preparation is kept, and clients resend each
+/// epoch.
 pub(crate) fn post_prepare_beacon_proposer(
     req: &Request<'_>,
-    _ctx: &ApiCtx,
+    ctx: &ApiCtx,
     resp: &mut Response<'_>,
 ) {
     let Some(preparations) = received(req.body, resp, ProposerPreparation::well_formed) else {
         return;
     };
-    silver_log::debug!(count = preparations.len(), "proposer preparations received");
-    if preparations.is_empty() {
+    let registry_len = ctx.read_state(|view| view.validators.count()) as u64;
+    let records: Vec<_> = preparations
+        .iter()
+        .map(ProposerPreparation::record)
+        .filter(|record| record.validator_index < registry_len)
+        .collect();
+    silver_log::debug!(
+        count = preparations.len(),
+        registered = records.len(),
+        "proposer preparations received"
+    );
+    if records.is_empty() {
         return resp.ok();
     }
 
     resp.submit(
-        preparations.len() * PreparationRecord::SIZE,
+        records.len() * PreparationRecord::SIZE,
         |out| {
-            for (record, preparation) in
-                out.chunks_exact_mut(PreparationRecord::SIZE).zip(&preparations)
-            {
-                preparation.record().encode(record);
+            for (chunk, record) in out.chunks_exact_mut(PreparationRecord::SIZE).zip(&records) {
+                record.encode(chunk);
             }
         },
         |preparations| BeaconApiRequest::ProposerPreparations { preparations },
@@ -100,13 +111,15 @@ impl ProposerPreparation<'_> {
 
 #[cfg(test)]
 mod tests {
-    use silver_beacon_state_data::SLOTS_PER_EPOCH;
+    use silver_beacon_state_data::{
+        BeaconState, BeaconStateOwner, EpochStateFinalized, SLOTS_PER_EPOCH, SpecConfig, ValSeed,
+    };
     use silver_common::TCacheProducer;
     use silver_httpcore::ParsedRequest;
 
     use super::*;
     use crate::{
-        ctx::anchor_ctx,
+        ctx::{anchor_ctx, test_ctx},
         http::{ids::MAX_BODY_IDS, router::Outcome},
         testing::{dispatch, dispatch_into, posting, submissions},
     };
@@ -154,9 +167,8 @@ mod tests {
     }
 
     /// Every receipt endpoint answers the same bodyless 200: neither schema
-    /// declares content under it. Neither reads beacon state, so
-    /// the whole suite runs against a node that has published none — which is
-    /// when a validator client first sends these.
+    /// declares content under it. The suite runs against an empty anchor,
+    /// which is when a validator client first sends these.
     #[test]
     fn well_formed_body_is_acknowledged_with_a_bodyless_200() {
         for (path, entry) in entries() {
@@ -238,16 +250,28 @@ mod tests {
         assert_bad_request(&json_post(PREPARE, &at_cap), "invalid entry in request body");
     }
 
+    fn ctx_with_validators(count: usize) -> ApiCtx {
+        let seeds: Vec<_> = (0..count).map(|_| ValSeed::default()).collect();
+        let state = BeaconState::for_test(EpochStateFinalized::default(), &seeds, 0);
+        let mut owner = BeaconStateOwner::new(state);
+        let anchor = owner.roll_fresh();
+        owner.publish_state_id(anchor);
+        test_ctx(&SpecConfig::mainnet(), owner.reader())
+    }
+
+    fn preparation_of(validator_index: u64, fee_recipient: u8) -> String {
+        format!(
+            "{{\"validator_index\":\"{validator_index}\",\"fee_recipient\":\"0x{}\"}}",
+            hex::encode([fee_recipient; 20])
+        )
+    }
+
     #[test]
     fn preparations_are_forwarded_through_the_cache() {
-        let other = format!(
-            "{{\"validator_index\":\"9\",\"fee_recipient\":\"0x{}\"}}",
-            hex::encode([0x11; 20])
-        );
-        let body = format!("[{},{other}]", preparation());
+        let body = format!("[{},{}]", preparation(), preparation_of(9, 0x11));
         let mut submissions = submissions();
         let (outcome, response) =
-            dispatch_into(&anchor_ctx(), &posting(PREPARE, &body), &mut submissions);
+            dispatch_into(&ctx_with_validators(10), &posting(PREPARE, &body), &mut submissions);
         assert_eq!(response, BODYLESS_OK);
         let Outcome::Response(Some(BeaconApiRequest::ProposerPreparations { preparations })) =
             outcome
@@ -261,6 +285,30 @@ mod tests {
         ]));
     }
 
+    #[test]
+    fn preparations_past_the_registry_are_dropped() {
+        let body = format!("[{},{}]", preparation_of(10, 0x11), preparation_of(9, 0x22));
+        let mut submissions = submissions();
+        let (outcome, response) =
+            dispatch_into(&ctx_with_validators(10), &posting(PREPARE, &body), &mut submissions);
+        assert_eq!(response, BODYLESS_OK);
+        let Outcome::Response(Some(BeaconApiRequest::ProposerPreparations { preparations })) =
+            outcome
+        else {
+            panic!("{outcome:?}")
+        };
+        let written = submissions.read_buffer(preparations).unwrap();
+        assert!(
+            PreparationRecord::decode_all(written)
+                .eq([PreparationRecord { validator_index: 9, fee_recipient: [0x22; 20] }])
+        );
+
+        let body = format!("[{},{}]", preparation_of(10, 0x11), preparation_of(u64::MAX, 0x22));
+        let (outcome, response) = dispatch(&ctx_with_validators(10), &posting(PREPARE, &body));
+        assert_eq!(response, BODYLESS_OK);
+        assert_eq!(outcome, Outcome::Response(None), "nothing reaches the state tile");
+    }
+
     #[test]
     fn registrations_wait_on_the_sidecar() {
         let mut ctx = anchor_ctx();