pub const MAX_FULL_TX_BODY: usize = 1 << 16;
pub const WIRE_VERSION: u8 = crate::protocol::WIRE_VERSION as u8;
pub const PREAMBLE: &[u8; 6] = b"PLS2\x02\x00";
pub const MSG_TX: u8 = 1;
pub const MSG_HEARTBEAT: u8 = 2;
pub const MSG_SHED: u8 = 3;
pub const FLAG_ALT_INCOMPLETE: u8 = 0x01;
pub const TLV_LOADED_WRITABLE: u8 = 1;
pub const TLV_LOADED_READONLY: u8 = 2;
pub const TLV_SERVER_TS_MS: u8 = 3;
pub const TLV_HIGHEST_SEQ: u8 = 4;
#[derive(Debug, PartialEq, Eq)]
pub struct FrameError;
pub fn put_tlv(buf: &mut Vec<u8>, t: u8, value: &[u8]) {
debug_assert!(value.len() <= u16::MAX as usize);
buf.push(t);
buf.extend_from_slice(&(value.len() as u16).to_le_bytes());
buf.extend_from_slice(value);
}
pub fn parse_tlvs(src: &[u8]) -> Result<Vec<(u8, &[u8])>, FrameError> {
let mut out: Vec<(u8, &[u8])> = Vec::new();
let mut off = 0usize;
while off < src.len() {
let t = *src.get(off).ok_or(FrameError)?;
let lo = *src.get(off + 1).ok_or(FrameError)?;
let hi = *src.get(off + 2).ok_or(FrameError)?;
let len = u16::from_le_bytes([lo, hi]) as usize;
let start = off.checked_add(3).ok_or(FrameError)?;
let end = start.checked_add(len).ok_or(FrameError)?;
let value = src.get(start..end).ok_or(FrameError)?;
if out.iter().any(|(seen, _)| *seen == t) {
return Err(FrameError);
}
out.push((t, value));
off = end;
}
Ok(out)
}
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct FullInstruction {
pub program_id_index: u32,
pub accounts: Vec<u8>,
pub data: Vec<u8>,
}
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct FullAtl {
pub account_key: [u8; 32],
pub writable_indexes: Vec<u8>,
pub readonly_indexes: Vec<u8>,
}
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct FullTx {
pub slot: u64,
pub versioned: bool,
pub num_required_signatures: u32,
pub num_readonly_signed_accounts: u32,
pub num_readonly_unsigned_accounts: u32,
pub recent_blockhash: [u8; 32],
pub signatures: Vec<[u8; 64]>,
pub account_keys: Vec<[u8; 32]>,
pub instructions: Vec<FullInstruction>,
pub address_table_lookups: Vec<FullAtl>,
}
pub fn encode_full_tx(ft: &FullTx) -> Vec<u8> {
let mut b = Vec::with_capacity(256);
b.extend_from_slice(&ft.slot.to_le_bytes());
b.push(ft.num_required_signatures as u8);
b.push(ft.num_readonly_signed_accounts as u8);
b.push(ft.num_readonly_unsigned_accounts as u8);
b.push(ft.versioned as u8);
b.extend_from_slice(&ft.recent_blockhash);
put_u16(&mut b, ft.signatures.len());
for s in &ft.signatures {
b.extend_from_slice(s);
}
put_u16(&mut b, ft.account_keys.len());
for k in &ft.account_keys {
b.extend_from_slice(k);
}
put_u16(&mut b, ft.instructions.len());
for ix in &ft.instructions {
b.push(ix.program_id_index as u8);
put_u16(&mut b, ix.accounts.len());
b.extend_from_slice(&ix.accounts);
put_u16(&mut b, ix.data.len());
b.extend_from_slice(&ix.data);
}
put_u16(&mut b, ft.address_table_lookups.len());
for l in &ft.address_table_lookups {
b.extend_from_slice(&l.account_key);
put_u16(&mut b, l.writable_indexes.len());
b.extend_from_slice(&l.writable_indexes);
put_u16(&mut b, l.readonly_indexes.len());
b.extend_from_slice(&l.readonly_indexes);
}
b
}
pub fn decode_full_tx(src: &[u8]) -> Result<FullTx, FrameError> {
let (ft, consumed) = decode_full_tx_prefix(src)?;
if consumed != src.len() {
return Err(FrameError); }
Ok(ft)
}
pub fn decode_full_tx_prefix(src: &[u8]) -> Result<(FullTx, usize), FrameError> {
if src.len() > MAX_FULL_TX_BODY {
return Err(FrameError);
}
let mut d = Decoder { b: src, off: 0 };
let mut ft = FullTx {
slot: d.u64()?,
num_required_signatures: d.u8()? as u32,
num_readonly_signed_accounts: d.u8()? as u32,
num_readonly_unsigned_accounts: d.u8()? as u32,
versioned: d.u8()? != 0,
..Default::default()
};
ft.recent_blockhash.copy_from_slice(d.take(32)?);
let n = d.count(64)?;
ft.signatures.reserve(n);
for _ in 0..n {
let mut s = [0u8; 64];
s.copy_from_slice(d.take(64)?);
ft.signatures.push(s);
}
let n = d.count(32)?;
ft.account_keys.reserve(n);
for _ in 0..n {
let mut k = [0u8; 32];
k.copy_from_slice(d.take(32)?);
ft.account_keys.push(k);
}
let n = d.count(5)?; ft.instructions.reserve(n);
for _ in 0..n {
let program_id_index = d.u8()? as u32;
let alen = d.u16()?;
let accounts = d.take(alen)?.to_vec();
let dlen = d.u16()?;
let data = d.take(dlen)?.to_vec();
ft.instructions.push(FullInstruction {
program_id_index,
accounts,
data,
});
}
let n = d.count(36)?; ft.address_table_lookups.reserve(n);
for _ in 0..n {
let mut account_key = [0u8; 32];
account_key.copy_from_slice(d.take(32)?);
let wlen = d.u16()?;
let writable_indexes = d.take(wlen)?.to_vec();
let rlen = d.u16()?;
let readonly_indexes = d.take(rlen)?.to_vec();
ft.address_table_lookups.push(FullAtl {
account_key,
writable_indexes,
readonly_indexes,
});
}
Ok((ft, d.off))
}
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct FullTxV2 {
pub tx: FullTx,
pub alt_incomplete: bool,
pub loaded_writable: Vec<[u8; 32]>,
pub loaded_readonly: Vec<[u8; 32]>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Frame {
Tx(FullTxV2),
Heartbeat { server_ts_ms: u64, highest_seq: u64 },
Unknown(u8),
}
fn flatten32(addrs: &[[u8; 32]]) -> Vec<u8> {
let mut v = Vec::with_capacity(addrs.len() * 32);
for a in addrs {
v.extend_from_slice(a);
}
v
}
fn unflatten32(src: &[u8]) -> Result<Vec<[u8; 32]>, FrameError> {
if src.len() % 32 != 0 {
return Err(FrameError);
}
let mut out = Vec::with_capacity(src.len() / 32);
for c in src.chunks_exact(32) {
out.push(<[u8; 32]>::try_from(c).map_err(|_| FrameError)?);
}
Ok(out)
}
pub fn encode_frame_tx(
ft: &FullTx,
alt_incomplete: bool,
loaded_writable: &[[u8; 32]],
loaded_readonly: &[[u8; 32]],
) -> Vec<u8> {
let body = encode_full_tx(ft);
let mut b = Vec::with_capacity(body.len() + 2 + 64);
b.push(MSG_TX);
b.push(if alt_incomplete {
FLAG_ALT_INCOMPLETE
} else {
0
});
b.extend_from_slice(&body);
if !loaded_writable.is_empty() {
put_tlv(&mut b, TLV_LOADED_WRITABLE, &flatten32(loaded_writable));
}
if !loaded_readonly.is_empty() {
put_tlv(&mut b, TLV_LOADED_READONLY, &flatten32(loaded_readonly));
}
b
}
pub fn decode_frame(src: &[u8]) -> Result<Frame, FrameError> {
let msg_type = *src.first().ok_or(FrameError)?;
let flags = *src.get(1).ok_or(FrameError)?;
let rest = src.get(2..).ok_or(FrameError)?;
match msg_type {
MSG_TX => {
if flags & !FLAG_ALT_INCOMPLETE != 0 {
return Err(FrameError); }
let (tx, consumed) = decode_full_tx_prefix(rest)?;
let tlvs = parse_tlvs(rest.get(consumed..).ok_or(FrameError)?)?;
let mut v2 = FullTxV2 {
tx,
alt_incomplete: flags & FLAG_ALT_INCOMPLETE != 0,
loaded_writable: Vec::new(),
loaded_readonly: Vec::new(),
};
for (t, value) in tlvs {
match t {
TLV_LOADED_WRITABLE => v2.loaded_writable = unflatten32(value)?,
TLV_LOADED_READONLY => v2.loaded_readonly = unflatten32(value)?,
_ => {} }
}
Ok(Frame::Tx(v2))
}
MSG_HEARTBEAT => {
if flags != 0 {
return Err(FrameError);
}
let tlvs = parse_tlvs(rest)?;
let mut server_ts_ms = 0u64;
let mut highest_seq = 0u64;
for (t, value) in tlvs {
match t {
TLV_SERVER_TS_MS => {
server_ts_ms =
u64::from_le_bytes(<[u8; 8]>::try_from(value).map_err(|_| FrameError)?)
}
TLV_HIGHEST_SEQ => {
highest_seq =
u64::from_le_bytes(<[u8; 8]>::try_from(value).map_err(|_| FrameError)?)
}
_ => {}
}
}
Ok(Frame::Heartbeat {
server_ts_ms,
highest_seq,
})
}
other => Ok(Frame::Unknown(other)),
}
}
pub const DG_SIG_FIRST: u8 = 1;
pub const DG_HEARTBEAT: u8 = 2;
pub const DG_SIG_FIRST_MIN: usize = 1 + 8 + 8 + 64;
pub const DG_HEARTBEAT_MIN: usize = 1 + 8 + 8;
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Datagram {
SigFirst {
slot: u64,
seq: u64,
signature: [u8; 64],
},
Heartbeat {
server_ts_ms: u64,
highest_seq: u64,
},
Unknown(u8),
}
pub fn encode_dg_sig_first(buf: &mut [u8; DG_SIG_FIRST_MIN], slot: u64, seq: u64, sig: &[u8; 64]) {
buf[0] = DG_SIG_FIRST;
buf[1..9].copy_from_slice(&slot.to_le_bytes());
buf[9..17].copy_from_slice(&seq.to_le_bytes());
buf[17..81].copy_from_slice(sig);
}
pub fn encode_dg_heartbeat(buf: &mut [u8; DG_HEARTBEAT_MIN], server_ts_ms: u64, highest_seq: u64) {
buf[0] = DG_HEARTBEAT;
buf[1..9].copy_from_slice(&server_ts_ms.to_le_bytes());
buf[9..17].copy_from_slice(&highest_seq.to_le_bytes());
}
pub fn decode_datagram(src: &[u8]) -> Option<Datagram> {
match *src.first()? {
DG_SIG_FIRST if src.len() >= DG_SIG_FIRST_MIN => Some(Datagram::SigFirst {
slot: u64::from_le_bytes(src.get(1..9)?.try_into().ok()?),
seq: u64::from_le_bytes(src.get(9..17)?.try_into().ok()?),
signature: src.get(17..81)?.try_into().ok()?,
}),
DG_HEARTBEAT if src.len() >= DG_HEARTBEAT_MIN => Some(Datagram::Heartbeat {
server_ts_ms: u64::from_le_bytes(src.get(1..9)?.try_into().ok()?),
highest_seq: u64::from_le_bytes(src.get(9..17)?.try_into().ok()?),
}),
DG_SIG_FIRST | DG_HEARTBEAT => None, other => Some(Datagram::Unknown(other)),
}
}
fn put_u16(b: &mut Vec<u8>, n: usize) {
b.extend_from_slice(&(n as u16).to_le_bytes());
}
struct Decoder<'a> {
b: &'a [u8],
off: usize,
}
impl<'a> Decoder<'a> {
fn take(&mut self, n: usize) -> Result<&'a [u8], FrameError> {
let end = self.off.checked_add(n).ok_or(FrameError)?;
if end > self.b.len() {
return Err(FrameError);
}
let s = &self.b[self.off..end];
self.off = end;
Ok(s)
}
fn u8(&mut self) -> Result<u8, FrameError> {
Ok(self.take(1)?[0])
}
fn u16(&mut self) -> Result<usize, FrameError> {
let s = self.take(2)?;
Ok(u16::from_le_bytes([s[0], s[1]]) as usize)
}
fn u64(&mut self) -> Result<u64, FrameError> {
let s = self.take(8)?;
Ok(u64::from_le_bytes(s.try_into().unwrap()))
}
fn count(&mut self, min_elem: usize) -> Result<usize, FrameError> {
let n = self.u16()?;
if min_elem == 0 || n > (self.b.len() - self.off) / min_elem {
return Err(FrameError);
}
Ok(n)
}
}
#[cfg(test)]
mod tests {
use super::*;
fn sample(versioned: bool) -> FullTx {
FullTx {
slot: 7,
versioned,
num_required_signatures: 1,
recent_blockhash: [0xCC; 32],
signatures: vec![[7u8; 64]],
account_keys: vec![[0xA1; 32]],
instructions: vec![FullInstruction {
program_id_index: 1,
accounts: vec![0],
data: vec![0xDE, 0xAD, 0xBE],
}],
address_table_lookups: if versioned {
vec![FullAtl {
account_key: [0xEE; 32],
writable_indexes: vec![5],
readonly_indexes: vec![7],
}]
} else {
vec![]
},
..Default::default()
}
}
#[test]
fn full_tx_round_trip_legacy_and_v0() {
for versioned in [false, true] {
let ft = sample(versioned);
let body = encode_full_tx(&ft);
let got = decode_full_tx(&body).unwrap();
assert_eq!(got, ft);
}
}
#[test]
fn full_tx_rejects_truncated() {
let body = encode_full_tx(&sample(true));
assert_eq!(decode_full_tx(&body[..body.len() - 3]), Err(FrameError));
}
#[test]
fn preamble_is_six_bytes_and_starts_nonzero() {
assert_eq!(PREAMBLE.len(), 6);
assert_ne!(PREAMBLE[0], 0x00);
assert_eq!(&PREAMBLE[0..4], b"PLS2");
assert_eq!(PREAMBLE[4], WIRE_VERSION);
assert_eq!(PREAMBLE[5], 0);
}
#[test]
fn tlv_round_trips_in_order() {
let mut b = Vec::new();
put_tlv(&mut b, TLV_LOADED_WRITABLE, &[1u8; 64]);
put_tlv(&mut b, TLV_LOADED_READONLY, &[2u8; 32]);
let got = parse_tlvs(&b).expect("valid");
assert_eq!(got.len(), 2);
assert_eq!(got[0].0, TLV_LOADED_WRITABLE);
assert_eq!(got[0].1, &[1u8; 64][..]);
assert_eq!(got[1].0, TLV_LOADED_READONLY);
assert_eq!(got[1].1, &[2u8; 32][..]);
}
#[test]
fn tlv_length_is_u16_so_a_large_value_fits() {
let big = vec![7u8; 3200];
let mut b = Vec::new();
put_tlv(&mut b, TLV_LOADED_WRITABLE, &big);
let got = parse_tlvs(&b).expect("valid");
assert_eq!(got[0].1.len(), 3200);
}
#[test]
fn tlv_unknown_type_is_kept_for_the_caller_to_skip() {
let mut b = Vec::new();
put_tlv(&mut b, 200, &[9u8; 4]);
let got = parse_tlvs(&b).expect("unknown types parse fine");
assert_eq!(got[0].0, 200);
}
#[test]
fn tlv_duplicate_type_is_rejected() {
let mut b = Vec::new();
put_tlv(&mut b, TLV_LOADED_WRITABLE, &[1u8; 32]);
put_tlv(&mut b, TLV_LOADED_WRITABLE, &[2u8; 32]);
assert_eq!(parse_tlvs(&b), Err(FrameError));
}
#[test]
fn tlv_truncated_is_rejected() {
let mut b = Vec::new();
put_tlv(&mut b, TLV_LOADED_WRITABLE, &[1u8; 32]);
for cut in 1..b.len() {
assert_eq!(parse_tlvs(&b[..cut]), Err(FrameError), "cut={cut}");
}
assert_eq!(parse_tlvs(&[]), Ok(Vec::new()));
}
#[test]
fn tlv_length_overrunning_the_buffer_is_rejected() {
let b = vec![1u8, 0xFF, 0xFF];
assert_eq!(parse_tlvs(&b), Err(FrameError));
}
fn sample_fulltx() -> FullTx {
FullTx {
slot: 438_690_000,
versioned: true,
num_required_signatures: 1,
num_readonly_signed_accounts: 0,
num_readonly_unsigned_accounts: 1,
recent_blockhash: [0xCC; 32],
signatures: vec![[7u8; 64]],
account_keys: vec![[0xA1; 32], [0xB2; 32]],
instructions: vec![FullInstruction {
program_id_index: 1,
accounts: vec![0],
data: vec![9, 9],
}],
address_table_lookups: vec![FullAtl {
account_key: [0xEE; 32],
writable_indexes: vec![0],
readonly_indexes: vec![1],
}],
}
}
#[test]
fn v2_tx_frame_round_trips_bare() {
let tx = sample_fulltx();
let enc = encode_frame_tx(&tx, false, &[], &[]);
assert_eq!(enc[0], MSG_TX);
assert_eq!(enc[1], 0, "no flags set");
match decode_frame(&enc).expect("valid") {
Frame::Tx(v2) => {
assert_eq!(v2.tx, tx);
assert!(!v2.alt_incomplete);
assert!(v2.loaded_writable.is_empty());
assert!(v2.loaded_readonly.is_empty());
}
other => panic!("expected Tx, got {other:?}"),
}
}
#[test]
fn v2_tx_frame_round_trips_enriched() {
let tx = sample_fulltx();
let w = [[0x11u8; 32], [0x22u8; 32]];
let r = [[0x33u8; 32]];
let enc = encode_frame_tx(&tx, true, &w, &r);
assert_eq!(enc[1] & FLAG_ALT_INCOMPLETE, FLAG_ALT_INCOMPLETE);
match decode_frame(&enc).expect("valid") {
Frame::Tx(v2) => {
assert!(v2.alt_incomplete);
assert_eq!(v2.loaded_writable, w.to_vec());
assert_eq!(v2.loaded_readonly, r.to_vec());
}
other => panic!("expected Tx, got {other:?}"),
}
}
#[test]
fn v2_body_is_byte_identical_to_v1_encoding() {
let tx = sample_fulltx();
let v1 = encode_full_tx(&tx);
let v2 = encode_frame_tx(&tx, false, &[], &[]);
assert_eq!(&v2[2..2 + v1.len()], &v1[..]);
}
#[test]
fn unknown_msg_type_is_reported_not_rejected() {
let mut enc = encode_frame_tx(&sample_fulltx(), false, &[], &[]);
enc[0] = 99;
match decode_frame(&enc).expect("unknown type must not error") {
Frame::Unknown(99) => {}
other => panic!("expected Unknown(99), got {other:?}"),
}
}
#[test]
fn reserved_flag_bits_are_rejected() {
let mut enc = encode_frame_tx(&sample_fulltx(), false, &[], &[]);
enc[1] = 0x02; assert_eq!(decode_frame(&enc), Err(FrameError));
}
#[test]
fn loaded_address_tlv_with_a_non_multiple_of_32_is_rejected() {
let mut enc = Vec::new();
enc.push(MSG_TX);
enc.push(0);
enc.extend_from_slice(&encode_full_tx(&sample_fulltx()));
put_tlv(&mut enc, TLV_LOADED_WRITABLE, &[0u8; 33]);
assert_eq!(decode_frame(&enc), Err(FrameError));
}
#[test]
fn heartbeat_frame_round_trips() {
let mut enc = Vec::new();
enc.push(MSG_HEARTBEAT);
enc.push(0);
put_tlv(
&mut enc,
TLV_SERVER_TS_MS,
&1_700_000_000_123u64.to_le_bytes(),
);
put_tlv(&mut enc, TLV_HIGHEST_SEQ, &4242u64.to_le_bytes());
match decode_frame(&enc).expect("valid") {
Frame::Heartbeat {
server_ts_ms,
highest_seq,
} => {
assert_eq!(server_ts_ms, 1_700_000_000_123);
assert_eq!(highest_seq, 4242);
}
other => panic!("expected Heartbeat, got {other:?}"),
}
}
#[test]
fn heartbeat_frame_rejects_any_nonzero_flags() {
for flags in [FLAG_ALT_INCOMPLETE, 0x02, 0xFF] {
let mut enc = Vec::new();
enc.push(MSG_HEARTBEAT);
enc.push(flags);
put_tlv(&mut enc, TLV_SERVER_TS_MS, &1u64.to_le_bytes());
assert_eq!(decode_frame(&enc), Err(FrameError), "flags={flags:#x}");
}
}
#[test]
fn frame_too_short_is_rejected() {
assert_eq!(decode_frame(&[]), Err(FrameError));
assert_eq!(decode_frame(&[MSG_TX]), Err(FrameError));
}
#[test]
fn dg_sig_first_round_trips() {
let mut buf = [0u8; DG_SIG_FIRST_MIN];
encode_dg_sig_first(&mut buf, 438_690_000, 12345, &[9u8; 64]);
assert_eq!(buf[0], DG_SIG_FIRST);
match decode_datagram(&buf).expect("valid") {
Datagram::SigFirst {
slot,
seq,
signature,
} => {
assert_eq!(slot, 438_690_000);
assert_eq!(seq, 12345);
assert_eq!(signature, [9u8; 64]);
}
other => panic!("expected SigFirst, got {other:?}"),
}
}
#[test]
fn dg_heartbeat_round_trips() {
let mut buf = [0u8; DG_HEARTBEAT_MIN];
encode_dg_heartbeat(&mut buf, 1_700_000_000_123, 999);
match decode_datagram(&buf).expect("valid") {
Datagram::Heartbeat {
server_ts_ms,
highest_seq,
} => {
assert_eq!(server_ts_ms, 1_700_000_000_123);
assert_eq!(highest_seq, 999);
}
other => panic!("expected Heartbeat, got {other:?}"),
}
}
#[test]
fn dg_minimum_length_not_exact_length() {
let mut buf = [0u8; DG_SIG_FIRST_MIN];
encode_dg_sig_first(&mut buf, 7, 8, &[3u8; 64]);
let mut longer = buf.to_vec();
longer.extend_from_slice(&[0xAB; 16]);
match decode_datagram(&longer).expect("trailing bytes must be ignored") {
Datagram::SigFirst { slot, seq, .. } => {
assert_eq!(slot, 7);
assert_eq!(seq, 8);
}
other => panic!("expected SigFirst, got {other:?}"),
}
}
#[test]
fn dg_below_minimum_is_rejected() {
let mut buf = [0u8; DG_SIG_FIRST_MIN];
encode_dg_sig_first(&mut buf, 1, 2, &[0u8; 64]);
assert!(decode_datagram(&buf[..DG_SIG_FIRST_MIN - 1]).is_none());
let mut hb = [0u8; DG_HEARTBEAT_MIN];
encode_dg_heartbeat(&mut hb, 1, 2);
assert!(decode_datagram(&hb[..DG_HEARTBEAT_MIN - 1]).is_none());
assert!(decode_datagram(&[]).is_none());
}
#[test]
fn dg_unknown_type_is_reported_not_rejected() {
let buf = [200u8, 1, 2, 3];
match decode_datagram(&buf).expect("unknown type must not be an error") {
Datagram::Unknown(200) => {}
other => panic!("expected Unknown(200), got {other:?}"),
}
}
#[test]
fn a_v1_72_byte_datagram_is_rejected_by_the_length_rule() {
let v1 = [1u8; 72];
assert_eq!(decode_datagram(&v1), None);
}
}