Batcher
About Batcher service and how does it works
Batcher is a standalone service which consumes computation events/request from Kafka queue and batches those events into a struct called BatchData before sending them to processor for processing.
Event Ingestion Workflow
The following diagram shows how onchain computation events are relayed from the CoreMgr contract through the Relayer to Kafka, which the Batcher then consumes:
Input: Privacy-enhancing computation events from CoreMgr contract
Output: Validated events published to Kafka topic PET_EVENTS
Note: Events represent encrypted computation requests from onchain contracts leveraging privacy-enhancing technologies (PET) for confidential state transitions.
Batch Processing Workflow
Once events are in Kafka, the Batcher aggregates them into BatchData and submits to the Processor with exponential backoff retry logic:
Input: EventData stream from Kafka Output: BatchData sent to Processor for encrypted computation
Every computation request which is relayed onchain is represented as a RequestParams struct post decoding.
Struct RequestParams looks like this
#[derive(Debug, Clone, Serialize, Deserialize,)]
pub struct RequestParams {
// implement hashing method on RequestParams
pub id: u64,
pub signer: Pubkey, // caller of the event
#[serde(with = "handle_serde")]
pub lhs: Handle, // lhs handle input for the request computation
#[serde(with = "handle_serde")]
pub rhs: Handle, // rhs handle input for the request computation
#[serde(
default,
serialize_with = "handle_serde::serialize_option",
deserialize_with = "handle_serde::deserialize_option",
skip_serializing_if = "Option::is_none"
)]
pub result: Option<Handle,>, // resultant handle for the computation request
#[serde(
default,
serialize_with = "handle_serde::serialize_option",
deserialize_with = "handle_serde::deserialize_option",
skip_serializing_if = "Option::is_none"
)]
pub middle: Option<Handle,>, // middle handle for the cases for condition scenarios like if/else statements
#[serde(
default,
serialize_with = "handle_serde::serialize_anon_transfer_handles_option",
deserialize_with = "handle_serde::deserialize_anon_transfer_handles_option",
skip_serializing_if = "Option::is_none"
)]
pub anon_transfer_handles: Option<AnonTransferHandles,>, // can be ignored for now
#[serde(
default,
serialize_with = "handle_serde::serialize_access_handles_option",
deserialize_with = "handle_serde::deserialize_access_handles_option",
skip_serializing_if = "Option::is_none"
)]
pub access_handles: Option<AccessHandles,>, // handles for offchain ACL checks
#[serde(
default,
serialize_with = "handle_serde::serialize_balance_conversion_handles_option",
deserialize_with = "handle_serde::deserialize_balance_conversion_handles_option",
skip_serializing_if = "Option::is_none"
)]
pub balance_conversion_handles: Option<BalanceConversionHandles,>, // can be ignored for now
pub scalar_byte: ScalarType, // this flag dictates, whether rhs is handle or a plaintext
pub operation: Operation, // type of operation the request params is representing
}The struct RequestParams contains necessary properties to exhaustively capture all the parameters required for computation request.
The batcher recurringly processes a set of RequestParams together and construct a struct called BatchData which looks like this
#[derive(Debug, Serialize, Deserialize, Clone,)]
pub struct BatchData {
pub groups: Vec<Group,>,
pub batch_hash: U256,
#[serde(skip_serializing_if = "Option::is_none")]
pub version: Option<BatchVersion,>,
#[serde(skip_serializing_if = "Option::is_none")]
pub timestamp: Option<u64,>,
}Within BatchData struct there exists a property called groups which is Vec<Group>.
Group is defined as a struct which can capture a set of RequestParams. Right now for simplicity Group only capture a single RequestParams struct.
Group struct looks like this
#[derive(Debug, Serialize, Deserialize, Clone,)]
pub struct Group {
pub inputs: Vec<GroupInput,>,
pub expression: String,
pub signer: Pubkey, // owner who raised the request
pub output: GroupOutput,
#[serde(
serialize_with = "handle_serde::serialize_vec",
deserialize_with = "handle_serde::deserialize_vec"
)]
pub intermediate_handles: Vec<Handle,>, // intermediate handles would be empty given one group represent only one request (can be ignored)
pub requests: Vec<RequestParams,>, // requests len would always be one given one one request per group (length would always be one for current implementation)
}Right now a group would represent only one Request. GroupInput and GroupOutput also corresponds to a single request.
Group has a property called expression of type string which basically normalizes request as a arithemetic expression for example:
Request for additon of two ciphertext could have an expression like this:
(296174089203748596321987120348571092834 + 296174089203748596321987120348571092354)Further, GroupInput and GroupOutput structs contains necessary params to verify the validity of ciphertext and the resultant handle post request computation
#[derive(Debug, Serialize, Deserialize, Clone,)]
pub struct GroupInput {
#[serde(with = "handle_serde")]
pub handle: Handle,
#[serde(skip_serializing_if = "Option::is_none")]
pub ciphertext_with_cts: Option<CiphertextWithCts,>, // need to define ciphertext type
pub leaf_hash: Bytes32,
pub inclusion_proof: MerkleInclusionProof,
pub signature: Vec<u8,>,
pub root_hash: Bytes32,
pub timestamp: u64,
}
#[derive(Debug, Serialize, Deserialize, Clone,)]
pub struct GroupOutput {
#[serde(with = "handle_serde")]
pub handle: Handle, // result handle post computation
}Once the BatchData is constructed it is sent over the processor route for processing. Post computation Batcher receives an acknowlegdement from processor as a signed for processing complete.