DAL node: bound committee fetch level and decouple P2P receive callback

What

Two changes to the DAL node's shard reception path:

1. Decouple the receive loop (gs_transport_connection.ml, node_context.ml): the P2P receive loop (transport_layer_inputs_handler) called recv_any, awaited the application callback, enqueued In_message to the Gossipsub worker, then repeated. Because recv_any listens across all connections, a slow callback on one peer delayed ingestion from all other peers. The callback is now spawned with Lwt.dont_wait so the loop continues immediately; errors are logged via the app_in_callback_failed Warning event and exceptions via Format.eprintf. Concurrent fetches for the same level are deduplicated: a single in-flight promise is shared, so a burst of shards at the same level triggers exactly one L1 RPC.

2. Bound the committee fetch level (daemon.ml, event.ml): the attester branch of shards_in_handler derives a committee level from the incoming message header and calls Node_context.fetch_committees. On a cache miss this issues a synchronous L1 context/dal/shards RPC. The cache is populated by block processing within the normal attestation window; messages with levels beyond that window always miss. The fetch is now skipped when committee_level > head_level + attestation_lag + validation_slack. A Notice event is emitted when the guard fires.

Why

1. Callbacks from different peers now run concurrently rather than serialising on the shared receive loop. The in-flight deduplication prevents a burst of concurrent callbacks from amplifying a cold-cache miss into many parallel L1 RPCs for the same level.

2. Messages with implausibly far-future levels always miss the committee cache and trigger L1 RPCs without bound. The guard limits fetches to the range the block handler legitimately pre-fetches.

How

• Commit 1: replace let* _res = app_in_callback ... in with Lwt.dont_wait, and move In_message dispatch immediately after recv_any. The committee cache is pre-populated by block processing before shards for that level arrive, so the reordering is semantically equivalent for well-formed traffic. Also adds committee_fetch_in_flight : Committee_fetch_tbl.t (a size-bounded Aches.Vache.Map) to the node context; fetch_committees checks it on cache miss and shares the existing promise if one is already in flight. Lwt.on_any removes the entry when the promise settles.

• Commit 2: compute committee_level = slot_level + attestation_lag - 1 and max_committee_level = head_level + attestation_lag + validation_slack, and emit + return_unit early when committee_level > max_committee_level. The window is the forward dual of the outdated-message check in Message_validation: that check drops messages whose slot level is more than attestation_lag + validation_slack behind the head; this drops fetches that far ahead.

Manually testing the MR

The guard (change 2) can in principle be checked by running a controller/attester DAL node against a sandboxed L1 and having a peer emit shard messages carrying far-future levels (e.g. slot level 1,000,000). Confirm that no context/dal/shards RPCs are issued for those levels (inspect L1 node RPC logs) and that a single shard_committee_level_out_of_window Notice fires per shard. Valid shards at the current head level should continue to be received and stored normally.

Note that injecting such crafted shard messages is non-trivial in practice: shard messages arrive via the GossipSub P2P layer and there is no existing mechanism to synthesise a raw P2P message with an arbitrary header from outside the node (e.g. from a Tezt sandbox). This is also the reason the guard behaviour is not covered by an automated test; only the in-flight deduplication logic (change 1), which is purely internal, has a unit test.