bundle: add support for retrieving bundles from a block engine

plugin/bundle/Cargo.toml

@@ -0,0 +1,33 @@
+[package]
+name = "firedancer-plugin-bundle"
+version = "0.1.0"
+edition = "2021"
+
+[lib]
+crate-type = ["staticlib"]
+
+[dependencies]
+tonic = { version = "0.12.2", features = ["tls-roots", "tls", "tls-webpki-roots"] }
+prost = "0.13.3"
+prost-types = "0.13.3"
+log = "0.4.22"
+tokio = "1.40.0"
+tokio-stream = "0.1"
+futures = "0.3.30"
+chrono = "0.4.38"
+thiserror = "1.0.64"
+bs58 = "0.5.1"
+
+[build-dependencies]
+tonic-build = "0.12.2"
+protobuf-src = "2.1.0"
+prost-types = "0.13.3"
+
+[dev-dependencies]
+env_logger = "0.11.5"
+ed25519-dalek = "2.1.1"
+
+[profile.release-with-debug]
+inherits = "release"
+debug = true
+split-debuginfo = "packed"

plugin/bundle/build.rs

@@ -0,0 +1,38 @@
+use tonic_build::configure;
+
+fn main() -> Result<(), std::io::Error> {
+    const PROTOC_ENVAR: &str = "PROTOC";
+    if std::env::var(PROTOC_ENVAR).is_err() {
+        #[cfg(not(windows))]
+        std::env::set_var(PROTOC_ENVAR, protobuf_src::protoc());
+    }
+
+    let proto_base_path = std::path::PathBuf::from("protos");
+    let proto_files = [
+        "auth.proto",
+        "block_engine.proto",
+        "bundle.proto",
+        "packet.proto",
+        "relayer.proto",
+        "shared.proto",
+    ];
+    let mut protos = Vec::new();
+    for proto_file in &proto_files {
+        let proto = proto_base_path.join(proto_file);
+        println!("cargo:rerun-if-changed={}", proto.display());
+        protos.push(proto);
+    }
+
+    configure()
+        .build_client(true)
+        .build_server(false)
+        .type_attribute(
+            "TransactionErrorType",
+            "#[cfg_attr(test, derive(enum_iterator::Sequence))]",
+        )
+        .type_attribute(
+            "InstructionErrorType",
+            "#[cfg_attr(test, derive(enum_iterator::Sequence))]",
+        )
+        .compile(&protos, &[proto_base_path])
+}

plugin/bundle/protos/auth.proto

@@ -0,0 +1,76 @@
+syntax = "proto3";
+
+package auth;
+
+import "google/protobuf/timestamp.proto";
+
+enum Role {
+  RELAYER = 0;
+  SEARCHER = 1;
+  VALIDATOR = 2;
+  SHREDSTREAM_SUBSCRIBER = 3;
+}
+
+message GenerateAuthChallengeRequest {
+  /// Role the client is attempting to generate tokens for.
+  Role role = 1;
+
+  /// Client's 32 byte pubkey.
+  bytes pubkey = 2;
+}
+
+message GenerateAuthChallengeResponse {
+  string challenge = 1;
+}
+
+message GenerateAuthTokensRequest {
+  /// The pre-signed challenge.
+  string challenge = 1;
+
+  /// The signing keypair's corresponding 32 byte pubkey.
+  bytes client_pubkey = 2;
+
+  /// The 64 byte signature of the challenge signed by the client's private key. The private key must correspond to
+  // the pubkey passed in the [GenerateAuthChallenge] method. The client is expected to sign the challenge token
+  // prepended with their pubkey. For example sign(pubkey, challenge).
+  bytes signed_challenge = 3;
+}
+
+message Token {
+    /// The token.
+    string value = 1;
+
+    /// When the token will expire.
+    google.protobuf.Timestamp expires_at_utc = 2;
+}
+
+message GenerateAuthTokensResponse {
+  /// The token granting access to resources.
+  Token access_token = 1;
+
+  /// The token used to refresh the access_token. This has a longer TTL than the access_token.
+  Token refresh_token = 2;
+}
+
+message RefreshAccessTokenRequest {
+  /// Non-expired refresh token obtained from the [GenerateAuthTokens] method.
+  string refresh_token = 1;
+}
+
+message RefreshAccessTokenResponse {
+  /// Fresh access_token.
+  Token access_token = 1;
+}
+
+/// This service is responsible for issuing auth tokens to clients for API access.
+service AuthService {
+  /// Returns a challenge, client is expected to sign this challenge with an appropriate keypair in order to obtain access tokens.
+  rpc GenerateAuthChallenge(GenerateAuthChallengeRequest) returns (GenerateAuthChallengeResponse) {}
+
+  /// Provides the client with the initial pair of auth tokens for API access.
+  rpc GenerateAuthTokens(GenerateAuthTokensRequest) returns (GenerateAuthTokensResponse) {}
+
+  /// Call this method with a non-expired refresh token to obtain a new access token.
+  rpc RefreshAccessToken(RefreshAccessTokenRequest) returns (RefreshAccessTokenResponse) {}
+}
+

plugin/bundle/protos/block_engine.proto

@@ -0,0 +1,98 @@
+syntax = "proto3";
+
+import "packet.proto";
+import "shared.proto";
+import "bundle.proto";
+
+package block_engine;
+
+message SubscribePacketsRequest {}
+message SubscribePacketsResponse {
+  shared.Header header = 1;
+  packet.PacketBatch batch = 2;
+}
+
+message SubscribeBundlesRequest {}
+message SubscribeBundlesResponse {
+  repeated bundle.BundleUuid bundles = 1;
+}
+
+message BlockBuilderFeeInfoRequest {}
+message BlockBuilderFeeInfoResponse {
+  string pubkey = 1;
+
+  // commission (0-100)
+  uint64 commission = 2;
+}
+
+message AccountsOfInterest {
+  // use * for all accounts
+  repeated string accounts = 1;
+}
+
+message AccountsOfInterestRequest {}
+message AccountsOfInterestUpdate {
+  repeated string accounts = 1;
+}
+
+message ProgramsOfInterestRequest {}
+message ProgramsOfInterestUpdate {
+  repeated string programs = 1;
+}
+
+// A series of packets with an expiration attached to them.
+// The header contains a timestamp for when this packet was generated.
+// The expiry is how long the packet batches have before they expire and are forwarded to the validator.
+// This provides a more censorship resistant method to MEV than block engines receiving packets directly.
+message ExpiringPacketBatch {
+  shared.Header header = 1;
+  packet.PacketBatch batch = 2;
+  uint32 expiry_ms = 3;
+}
+
+// Packets and heartbeats are sent over the same stream.
+// ExpiringPacketBatches have an expiration attached to them so the block engine can track
+// how long it has until the relayer forwards the packets to the validator.
+// Heartbeats contain a timestamp from the system and is used as a simple and naive time-sync mechanism
+// so the block engine has some idea on how far their clocks are apart.
+message PacketBatchUpdate {
+  oneof msg {
+    ExpiringPacketBatch batches = 1;
+    shared.Heartbeat heartbeat = 2;
+  }
+}
+
+message StartExpiringPacketStreamResponse {
+  shared.Heartbeat heartbeat = 1;
+}
+
+/// Validators can connect to Block Engines to receive packets and bundles.
+service BlockEngineValidator {
+  /// Validators can subscribe to the block engine to receive a stream of packets
+  rpc SubscribePackets (SubscribePacketsRequest) returns (stream SubscribePacketsResponse) {}
+
+  /// Validators can subscribe to the block engine to receive a stream of simulated and profitable bundles
+  rpc SubscribeBundles (SubscribeBundlesRequest) returns (stream SubscribeBundlesResponse) {}
+
+  // Block builders can optionally collect fees. This returns fee information if a block builder wants to
+  // collect one.
+  rpc GetBlockBuilderFeeInfo (BlockBuilderFeeInfoRequest) returns (BlockBuilderFeeInfoResponse) {}
+}
+
+/// Relayers can forward packets to Block Engines.
+/// Block Engines provide an AccountsOfInterest field to only send transactions that are of interest.
+service BlockEngineRelayer {
+  /// The block engine feeds accounts of interest (AOI) updates to the relayer periodically.
+  /// For all transactions the relayer receives, it forwards transactions to the block engine which write-lock
+  /// any of the accounts in the AOI.
+  rpc SubscribeAccountsOfInterest (AccountsOfInterestRequest) returns (stream AccountsOfInterestUpdate) {}
+
+  rpc SubscribeProgramsOfInterest (ProgramsOfInterestRequest) returns (stream ProgramsOfInterestUpdate) {}
+
+  // Validators can subscribe to packets from the relayer and receive a multiplexed signal that contains a mixture
+  // of packets and heartbeats.
+  // NOTE: This is a bi-directional stream due to a bug with how Envoy handles half closed client-side streams.
+  // The issue is being tracked here: https://github.com/envoyproxy/envoy/issues/22748. In the meantime, the
+  // server will stream heartbeats to clients at some reasonable cadence.
+  rpc StartExpiringPacketStream (stream PacketBatchUpdate) returns (stream StartExpiringPacketStreamResponse) {}
+}

plugin/bundle/protos/bundle.proto

@@ -0,0 +1,111 @@
+syntax = "proto3";
+
+import "packet.proto";
+import "shared.proto";
+
+package bundle;
+
+message Bundle {
+  shared.Header header = 2;
+  repeated packet.Packet packets = 3;
+}
+
+message BundleUuid {
+  bundle.Bundle bundle = 1;
+  string uuid = 2;
+}
+
+/* Bundle Result Types */
+
+// Indicates the bundle was accepted and forwarded to a validator.
+// NOTE: A single bundle may have multiple events emitted if forwarded to many validators.
+message Accepted {
+  // Slot at which bundle was forwarded.
+  uint64 slot = 1;
+
+  // Validator identity bundle was forwarded to.
+  string validator_identity = 2;
+}
+
+// Indicates the bundle was dropped and therefore not forwarded to any validator.
+message Rejected {
+  oneof reason {
+    StateAuctionBidRejected state_auction_bid_rejected = 1;
+    WinningBatchBidRejected winning_batch_bid_rejected = 2;
+    SimulationFailure simulation_failure = 3;
+    InternalError internal_error = 4;
+    DroppedBundle dropped_bundle = 5;
+  }
+}
+
+// Indicates the bundle's bid was high enough to win its state auction.
+// However, not high enough relative to other state auction winners and therefore excluded from being forwarded.
+message WinningBatchBidRejected {
+  // Auction's unique identifier.
+  string auction_id = 1;
+  // Bundle's simulated bid.
+  uint64 simulated_bid_lamports = 2;
+  optional string msg = 3;
+}
+
+// Indicates the bundle's bid was __not__ high enough to be included in its state auction's set of winners.
+message StateAuctionBidRejected {
+  // Auction's unique identifier.
+  string auction_id = 1;
+  // Bundle's simulated bid.
+  uint64 simulated_bid_lamports = 2;
+  optional string msg = 3;
+}
+
+// Bundle dropped due to simulation failure.
+message SimulationFailure {
+  // Signature of the offending transaction.
+  string tx_signature = 1;
+  optional string msg = 2;
+}
+
+// Bundle dropped due to an internal error.
+message InternalError {
+  string msg = 1;
+}
+
+// Bundle dropped (e.g. because no leader upcoming)
+message DroppedBundle {
+  string msg = 1;
+}
+
+message Finalized {}
+message Processed {
+  string validator_identity = 1;
+  uint64 slot = 2;
+  /// Index within the block.
+  uint64 bundle_index = 3;
+}
+message Dropped {
+  DroppedReason reason = 1;
+}
+enum DroppedReason {
+  BlockhashExpired = 0;
+  // One or more transactions in the bundle landed on-chain, invalidating the bundle.
+  PartiallyProcessed = 1;
+  // This indicates bundle was processed but not finalized. This could occur during forks.
+  NotFinalized = 2;
+}
+
+message BundleResult {
+  // Bundle's Uuid.
+  string bundle_id = 1;
+
+  oneof result {
+    // Indicated accepted by the block-engine and forwarded to a validator capable of packing bundles.
+    Accepted accepted = 2;
+    // Rejected by the block-engine.
+    Rejected rejected = 3;
+    // Reached finalized commitment level.
+    Finalized finalized = 4;
+    // Reached a processed commitment level.
+    Processed processed = 5;
+    // Was accepted and forwarded by the block-engine but never landed on-chain.
+    Dropped dropped = 6;
+  }
+}

plugin/bundle/protos/packet.proto

@@ -0,0 +1,30 @@
+syntax = "proto3";
+
+package packet;
+
+message PacketBatch {
+  repeated Packet packets = 1;
+}
+
+message Packet {
+  bytes data = 1;
+  Meta meta = 2;
+}
+
+message Meta {
+  uint64 size = 1;
+  string addr = 2;
+  uint32 port = 3;
+  PacketFlags flags = 4;
+  uint64 sender_stake = 5;
+}
+
+message PacketFlags {
+  bool discard = 1;
+  bool forwarded = 2;
+  bool repair = 3;
+  bool simple_vote_tx = 4;
+  bool tracer_packet = 5;
+  bool from_staked_node = 6;
+}
+