add new av store message type, send message for early fetches

Author: timwu20

polkadot/node/core/av-store/src/lib.rs

@@ -1232,6 +1232,24 @@ fn process_message(
 				},
 			}
 		},
+		AvailabilityStoreMessage::NoteBackableCandidate{ candidate_hash, tx } => {
+			let res = note_backable_candidate(
+				&subsystem.db,
+				&subsystem.config,
+				candidate_hash,
+				subsystem,
+			);
+
+			match res {
+				Ok(_) => {
+					let _ = tx.send(Ok(()));
+				},
+				Err(e) => {
+					let _ = tx.send(Err(()));
+					return Err(e)
+				},
+			}
+		},
 	}
 
 	Ok(())
@@ -1430,3 +1448,29 @@ fn prune_all(db: &Arc<dyn Database>, config: &Config, now: Duration) -> Result<(
 	db.write(tx)?;
 	Ok(())
 }
+
+fn note_backable_candidate(
+	db: &Arc<dyn Database>,
+	config: &Config,
+	candidate_hash: CandidateHash,
+	subsystem: &AvailabilityStoreSubsystem,
+) -> Result<(), Error> {
+	let mut tx = DBTransaction::new();
+
+	if load_meta(db, config, &candidate_hash)?.is_none() {
+		let now = subsystem.clock.now()?;
+		let meta = CandidateMeta {
+			state: State::Unavailable(now.into()),
+			data_available: false,
+			chunks_stored: bitvec::bitvec![u8, BitOrderLsb0; 0; 12],
+		};
+
+		let prune_at = now + KEEP_UNAVAILABLE_FOR;
+
+		write_pruning_key(&mut tx, config, prune_at, &candidate_hash);
+		write_meta(&mut tx, config, &candidate_hash, &meta);
+	}
+
+	db.write(tx)?;
+	Ok(())
+}

polkadot/node/network/availability-distribution/src/requester/mod.rs

@@ -31,7 +31,11 @@ use futures::{
 
 use polkadot_node_network_protocol::request_response::{v1, v2, IsRequest, ReqProtocolNames};
 use polkadot_node_subsystem::{
+<<<<<<< Updated upstream
 	messages::{ChainApiMessage, RuntimeApiMessage, CandidateBackingMessage},
+=======
+	messages::{ChainApiMessage, RuntimeApiMessage, CandidateBackingMessage, AvailabilityStoreMessage},
+>>>>>>> Stashed changes
 	overseer, ActivatedLeaf, ActiveLeavesUpdate,
 	SubsystemSender,
 };
@@ -307,6 +311,19 @@ impl Requester {
 				?scheduled_cores,
 				"Adding scheduled cores"
 			);
+			let sender = &mut ctx.sender().clone();
+			for (core_index, core_info) in &scheduled_cores {
+				let (tx, rx) = oneshot::channel();
+				sender
+					.send_message(
+						AvailabilityStoreMessage::NoteBackableCandidate{ candidate_hash: core_info.candidate_hash, tx },
+					)
+					.await;
+				if let Err(err) = rx.await {
+					gum::error!(target: LOG_TARGET, "Sending NoteBackableCandidate message failed: {:?}", err);
+				}
+				
+			}
 			self.add_cores(ctx, runtime, leaf, leaf_session_index, scheduled_cores).await?;
 		}
 

polkadot/node/subsystem-types/src/messages.rs

@@ -572,6 +572,15 @@ pub enum AvailabilityStoreMessage {
 		tx: oneshot::Sender<Result<(), ()>>,
 	},
 
+	/// Note that a candidate is backable. Only used by Speculative Availability to signal to Availability Store
+	/// that it should keep track of this candidate and it is highly likely the candidate will be backed.
+	NoteBackableCandidate{
+		/// A hash of the backable candidate.
+		candidate_hash: CandidateHash,
+		/// Sending side of the channel to send result to.
+		tx: oneshot::Sender<Result<(), ()>>,
+	},
+
 	/// Computes and checks the erasure root of `AvailableData` before storing all of its chunks in
 	/// the AV store.
 	///