feat: M1 lumen-core (FEC/crypto/packet/session + C ABI) and workspace scaffold
Ground-up low-latency streaming stack per docs/implementation-plan.md. M1 is
complete and tested; Linux host backends are cfg-gated stubs to be filled in on
real hardware (M0/M2).
lumen-core (built + tested on macOS/aarch64 — 21 tests):
- fec: ErasureCoder over GF(2^8) (reed-solomon-erasure, Moonlight-compatible)
and GF(2^16) Leopard-RS (reed-solomon-simd, the >1 Gbps wall-breaker); proptested
- packet: zero-copy #[repr(C)] framing, multi-block, FEC-aware reassembly
- crypto: AES-128-GCM with per-direction nonce salts + sequence-as-AAD
- session: host submit / client poll hot paths + input; loopback & UDP transports
- abi: opaque handles, versioned LumenConfig, panic guards; cbindgen-generated header
- acceptance: Rust loopback+proptest and a C harness that links the staticlib
Scaffold (compiles green on all platforms): lumen-host (vdisplay/capture/encode/
inject/web/pipeline seams under cfg(linux)), lumen-client-rs, tools/{loss-harness,
latency-probe}, Apple/Android client stubs, Gitea CI, docs.
Hardened against a multi-agent adversarial review (13 verified findings fixed,
regression-tested): reassembler memory-DoS bounds + block-consistency validation,
GCM nonce-reuse direction separation, ABI struct_size guard + range checks, FEC
shard-length guards, shard_payload datagram bound, key zeroization + Debug redaction.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,84 @@
|
||||
//! GF(2¹⁶) Leopard-RS backend (`reed-solomon-simd`). SIMD, O(n log n), up to 65535
|
||||
//! shards/block — this is what removes the GameStream 255-shard / ~1 Gbps wall.
|
||||
//! Shard length must be even.
|
||||
|
||||
use super::{validate_block_shape, validate_encode_shape, ErasureCoder, FecError};
|
||||
use crate::config::FecScheme;
|
||||
|
||||
pub struct Gf16Coder;
|
||||
|
||||
impl ErasureCoder for Gf16Coder {
|
||||
fn scheme(&self) -> FecScheme {
|
||||
FecScheme::Gf16
|
||||
}
|
||||
|
||||
fn encode(&self, data: &[Vec<u8>], recovery_count: usize) -> Result<Vec<Vec<u8>>, FecError> {
|
||||
if recovery_count == 0 {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
validate_encode_shape(data)?;
|
||||
let k = data.len();
|
||||
if data[0].len() % 2 != 0 {
|
||||
return Err(FecError::Config("GF(2^16) shard length must be even"));
|
||||
}
|
||||
reed_solomon_simd::encode(k, recovery_count, data)
|
||||
.map_err(|_| FecError::Backend("gf16 encode"))
|
||||
}
|
||||
|
||||
fn reconstruct(
|
||||
&self,
|
||||
data_count: usize,
|
||||
recovery_count: usize,
|
||||
received: &mut [Option<Vec<u8>>],
|
||||
) -> Result<Vec<Vec<u8>>, FecError> {
|
||||
validate_block_shape(received, data_count, recovery_count)?;
|
||||
let present = received.iter().filter(|s| s.is_some()).count();
|
||||
if present < data_count {
|
||||
return Err(FecError::TooFewShards {
|
||||
have: present,
|
||||
need: data_count,
|
||||
});
|
||||
}
|
||||
// Fast path: all originals already present, or FEC disabled.
|
||||
let originals_complete = received[..data_count].iter().all(|s| s.is_some());
|
||||
if recovery_count == 0 || originals_complete {
|
||||
let mut out = Vec::with_capacity(data_count);
|
||||
for slot in received.iter().take(data_count) {
|
||||
out.push(slot.clone().ok_or(FecError::TooFewShards {
|
||||
have: present,
|
||||
need: data_count,
|
||||
})?);
|
||||
}
|
||||
return Ok(out);
|
||||
}
|
||||
|
||||
// Hand the decoder the surviving originals and recovery shards, indexed.
|
||||
let original_in: Vec<(usize, &[u8])> = received[..data_count]
|
||||
.iter()
|
||||
.enumerate()
|
||||
.filter_map(|(i, s)| s.as_deref().map(|b| (i, b)))
|
||||
.collect();
|
||||
let recovery_in: Vec<(usize, &[u8])> = received[data_count..data_count + recovery_count]
|
||||
.iter()
|
||||
.enumerate()
|
||||
.filter_map(|(j, s)| s.as_deref().map(|b| (j, b)))
|
||||
.collect();
|
||||
|
||||
let restored =
|
||||
reed_solomon_simd::decode(data_count, recovery_count, original_in, recovery_in)
|
||||
.map_err(|_| FecError::Backend("gf16 decode"))?;
|
||||
|
||||
// Merge surviving originals with the recovered ones.
|
||||
let mut out: Vec<Vec<u8>> = Vec::with_capacity(data_count);
|
||||
for (i, slot) in received[..data_count].iter().enumerate() {
|
||||
if let Some(s) = slot {
|
||||
out.push(s.clone());
|
||||
} else if let Some(s) = restored.get(&i) {
|
||||
out.push(s.clone());
|
||||
} else {
|
||||
return Err(FecError::Backend("gf16 decode left an original missing"));
|
||||
}
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
//! GF(2⁸) classic Reed–Solomon backend (`reed-solomon-erasure`), equivalent to the
|
||||
//! `nanors` library Moonlight uses. Hard ceiling: data + recovery ≤ 255 shards/block.
|
||||
|
||||
use super::{validate_block_shape, validate_encode_shape, ErasureCoder, FecError};
|
||||
use crate::config::FecScheme;
|
||||
use reed_solomon_erasure::galois_8::ReedSolomon;
|
||||
|
||||
pub struct Gf8Coder;
|
||||
|
||||
impl ErasureCoder for Gf8Coder {
|
||||
fn scheme(&self) -> FecScheme {
|
||||
FecScheme::Gf8
|
||||
}
|
||||
|
||||
fn encode(&self, data: &[Vec<u8>], recovery_count: usize) -> Result<Vec<Vec<u8>>, FecError> {
|
||||
if recovery_count == 0 {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
validate_encode_shape(data)?;
|
||||
let k = data.len();
|
||||
let shard_len = data[0].len();
|
||||
let rs = ReedSolomon::new(k, recovery_count)
|
||||
.map_err(|_| FecError::Config("invalid GF(2^8) shard counts"))?;
|
||||
// reed-solomon-erasure fills parity in place: shards = data || zeroed parity.
|
||||
let mut shards: Vec<Vec<u8>> = Vec::with_capacity(k + recovery_count);
|
||||
shards.extend_from_slice(data);
|
||||
shards.resize_with(k + recovery_count, || vec![0u8; shard_len]);
|
||||
rs.encode(&mut shards)
|
||||
.map_err(|_| FecError::Backend("gf8 encode"))?;
|
||||
Ok(shards.split_off(k))
|
||||
}
|
||||
|
||||
fn reconstruct(
|
||||
&self,
|
||||
data_count: usize,
|
||||
recovery_count: usize,
|
||||
received: &mut [Option<Vec<u8>>],
|
||||
) -> Result<Vec<Vec<u8>>, FecError> {
|
||||
validate_block_shape(received, data_count, recovery_count)?;
|
||||
let present = received.iter().filter(|s| s.is_some()).count();
|
||||
if present < data_count {
|
||||
return Err(FecError::TooFewShards {
|
||||
have: present,
|
||||
need: data_count,
|
||||
});
|
||||
}
|
||||
if recovery_count == 0 {
|
||||
// No FEC: every original must already be present.
|
||||
return collect_originals(received, data_count);
|
||||
}
|
||||
let rs = ReedSolomon::new(data_count, recovery_count)
|
||||
.map_err(|_| FecError::Config("invalid GF(2^8) shard counts"))?;
|
||||
rs.reconstruct_data(received)
|
||||
.map_err(|_| FecError::Backend("gf8 reconstruct"))?;
|
||||
collect_originals(received, data_count)
|
||||
}
|
||||
}
|
||||
|
||||
fn collect_originals(
|
||||
received: &[Option<Vec<u8>>],
|
||||
data_count: usize,
|
||||
) -> Result<Vec<Vec<u8>>, FecError> {
|
||||
let mut out = Vec::with_capacity(data_count);
|
||||
for slot in received.iter().take(data_count) {
|
||||
out.push(
|
||||
slot.clone()
|
||||
.ok_or(FecError::Backend("reconstruction left an original missing"))?,
|
||||
);
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
@@ -0,0 +1,167 @@
|
||||
//! Erasure coding. Two backends behind one [`ErasureCoder`] trait: GF(2⁸) (classic
|
||||
//! Reed–Solomon, Moonlight-compatible, P1) and GF(2¹⁶) Leopard-RS (the wall-breaker, P2).
|
||||
//!
|
||||
//! The wall this breaks: GameStream's GF(2⁸) RS caps a block at 255 shards, which at
|
||||
//! 5120×1440@240 is hit around 1 Gbps. GF(2¹⁶) raises that ceiling to 65535 shards and
|
||||
//! runs in O(n log n) with SIMD, so the per-frame shard count stops being the limiter.
|
||||
|
||||
mod gf16;
|
||||
mod gf8;
|
||||
|
||||
pub use gf16::Gf16Coder;
|
||||
pub use gf8::Gf8Coder;
|
||||
|
||||
use crate::config::FecScheme;
|
||||
use thiserror::Error;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum FecError {
|
||||
#[error("invalid shard configuration: {0}")]
|
||||
Config(&'static str),
|
||||
#[error("too few shards to reconstruct (have {have}, need {need})")]
|
||||
TooFewShards { have: usize, need: usize },
|
||||
#[error("backend error: {0}")]
|
||||
Backend(&'static str),
|
||||
}
|
||||
|
||||
/// Backend-agnostic erasure coder. All shards in a block are equal length.
|
||||
pub trait ErasureCoder: Send + Sync {
|
||||
fn scheme(&self) -> FecScheme;
|
||||
|
||||
/// Encode `data` (K original shards) into `recovery_count` (M) parity shards.
|
||||
/// Returns the M recovery shards. `recovery_count == 0` returns an empty `Vec`.
|
||||
fn encode(&self, data: &[Vec<u8>], recovery_count: usize) -> Result<Vec<Vec<u8>>, FecError>;
|
||||
|
||||
/// Reconstruct the K original shards. `received` has length K+M: indices `0..K` are
|
||||
/// originals, `K..K+M` are recovery shards; `Some` = present, `None` = lost.
|
||||
/// Returns the K original shards in order.
|
||||
fn reconstruct(
|
||||
&self,
|
||||
data_count: usize,
|
||||
recovery_count: usize,
|
||||
received: &mut [Option<Vec<u8>>],
|
||||
) -> Result<Vec<Vec<u8>>, FecError>;
|
||||
}
|
||||
|
||||
/// Construct the coder for a scheme.
|
||||
pub fn coder_for(scheme: FecScheme) -> Box<dyn ErasureCoder> {
|
||||
match scheme {
|
||||
FecScheme::Gf8 => Box::new(Gf8Coder),
|
||||
FecScheme::Gf16 => Box::new(Gf16Coder),
|
||||
}
|
||||
}
|
||||
|
||||
/// Validate the shape `reconstruct` promises: `received.len() == data + recovery`, and
|
||||
/// every present shard shares one length. Both backends call this first so neither the
|
||||
/// fast path nor a malformed caller can slip mismatched-length or wrong-count shards
|
||||
/// through (the fast paths bypass the backend's own length checks otherwise).
|
||||
pub(crate) fn validate_block_shape(
|
||||
received: &[Option<Vec<u8>>],
|
||||
data_count: usize,
|
||||
recovery_count: usize,
|
||||
) -> Result<(), FecError> {
|
||||
if received.len() != data_count + recovery_count {
|
||||
return Err(FecError::Config(
|
||||
"received length must equal data + recovery",
|
||||
));
|
||||
}
|
||||
let mut len = None;
|
||||
for s in received.iter().flatten() {
|
||||
match len {
|
||||
None => len = Some(s.len()),
|
||||
Some(l) if l != s.len() => {
|
||||
return Err(FecError::Config("shards in a block must be equal length"));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Validate `encode` inputs: at least one data shard, all of equal length.
|
||||
pub(crate) fn validate_encode_shape(data: &[Vec<u8>]) -> Result<(), FecError> {
|
||||
let first = data
|
||||
.first()
|
||||
.ok_or(FecError::Config("no data shards"))?
|
||||
.len();
|
||||
if data.iter().any(|s| s.len() != first) {
|
||||
return Err(FecError::Config("data shards must be equal length"));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// Round-trip a block through a coder, losing exactly `lose` shards (some data,
|
||||
/// some recovery), and assert the originals come back byte-identical.
|
||||
fn roundtrip(coder: &dyn ErasureCoder, k: usize, m: usize, shard_len: usize, lose: &[usize]) {
|
||||
let data: Vec<Vec<u8>> = (0..k)
|
||||
.map(|i| (0..shard_len).map(|b| (i * 31 + b * 7) as u8).collect())
|
||||
.collect();
|
||||
let recovery = coder.encode(&data, m).unwrap();
|
||||
assert_eq!(recovery.len(), m);
|
||||
|
||||
let mut received: Vec<Option<Vec<u8>>> = Vec::with_capacity(k + m);
|
||||
received.extend(data.iter().cloned().map(Some));
|
||||
received.extend(recovery.iter().cloned().map(Some));
|
||||
for &idx in lose {
|
||||
received[idx] = None;
|
||||
}
|
||||
|
||||
let restored = coder.reconstruct(k, m, &mut received).unwrap();
|
||||
assert_eq!(restored, data);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gf8_recovers_within_budget() {
|
||||
// 16 data + 4 recovery; lose 2 data + 2 recovery (== budget).
|
||||
roundtrip(&Gf8Coder, 16, 4, 256, &[0, 7, 16, 19]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gf16_recovers_within_budget() {
|
||||
roundtrip(&Gf16Coder, 16, 4, 256, &[1, 9, 17, 18]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gf8_too_much_loss_errors() {
|
||||
let data: Vec<Vec<u8>> = (0..8).map(|_| vec![0u8; 64]).collect();
|
||||
let recovery = Gf8Coder.encode(&data, 2).unwrap();
|
||||
let mut received: Vec<Option<Vec<u8>>> = data
|
||||
.iter()
|
||||
.cloned()
|
||||
.map(Some)
|
||||
.chain(recovery.into_iter().map(Some))
|
||||
.collect();
|
||||
// Lose 3 with only 2 recovery shards → unrecoverable.
|
||||
received[0] = None;
|
||||
received[1] = None;
|
||||
received[2] = None;
|
||||
assert!(Gf16Coder.scheme() == FecScheme::Gf16);
|
||||
let err = Gf8Coder.reconstruct(8, 2, &mut received);
|
||||
assert!(err.is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reconstruct_rejects_wrong_received_length() {
|
||||
// data=2, recovery=2 expects a 4-element slice; a 3-element one must error, not
|
||||
// panic on the recovery-slice index (both backends).
|
||||
let mut recv: Vec<Option<Vec<u8>>> = vec![Some(vec![0u8; 8]), None, Some(vec![0u8; 8])];
|
||||
assert!(Gf16Coder.reconstruct(2, 2, &mut recv).is_err());
|
||||
let mut recv: Vec<Option<Vec<u8>>> = vec![Some(vec![0u8; 8]), None, Some(vec![0u8; 8])];
|
||||
assert!(Gf8Coder.reconstruct(2, 2, &mut recv).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reconstruct_rejects_mismatched_shard_lengths() {
|
||||
// The GF16 fast path used to clone shards verbatim without a length check.
|
||||
let mut recv: Vec<Option<Vec<u8>>> =
|
||||
vec![Some(vec![0u8; 8]), Some(vec![0u8; 6]), None, None];
|
||||
assert!(Gf16Coder.reconstruct(2, 2, &mut recv).is_err());
|
||||
let mut recv: Vec<Option<Vec<u8>>> =
|
||||
vec![Some(vec![0u8; 8]), Some(vec![0u8; 6]), None, None];
|
||||
assert!(Gf8Coder.reconstruct(2, 2, &mut recv).is_err());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user