funman300/OpenFUT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

openfut-blaze-host: thin Blaze sidecar, live-parity with Python

Browse Source 
Third migration step, and the one that turns fixture parity into transport
parity. A TCP host that frames a Fire2 stream, keeps one Session per
connection, calls openfut-adapter-fifa17::dispatch(), and writes the returned
frames in order. It owns a socket, a buffer, a session and diagnostics --
that is the complete list. No coins, club, packs, profiles or UTAS logic:
those belong to Core, reached through the adapter later.

NO TLS, and that is evidence-based rather than an omission. The Blaze main
port is plaintext: sending a raw Fire2 Util::ping to the running backend
returns a plaintext PingResponse, blaze_handle uses the raw socket, and only
redir_handle wraps ssl. TLS belongs to the redirector phase.

LIVE A/B AGAINST THE RUNNING PYTHON BACKEND: 101 frames across three
conversations, identical normalized traces. This is the first result in the
migration that is not purely offline. check-live-parity.sh replays the
recorded conversations against both endpoints over real sockets and diffs
volatile-masked traces; session keys and clocks are masked, so anything that
differs is behavioural.

Transport tests cover what fixtures cannot: byte-for-byte replay over a
socket, requests dribbled one byte at a time, several requests in one write,
the four-frame login burst ordered on the wire, session state persisting
across frames and NOT leaking between connections, an absurd payload length
closing the connection instead of allocating, and an undecodable body still
getting a reply. 18 tests here, 116 across the three migration crates.

MUTATION TESTED, including the comparison itself. Dropping a post-login
notification is caught by the probe (frame count) AND the diff; a same-length
content change deep inside a notification body (CTY "US"->"GB", payload 116
both sides) is caught ONLY by the trace digest. So the probe's exit code is
not the test -- the diff is, and the README says so. check-live-parity.sh was
itself verified to exit 1 under mutation.

The listen port is required configuration with no default, so the sidecar
cannot silently collide with the working container. OPENFUT_BIND stays the
advertised-config bind (the adapter derives nucleusConnect from it,
reproducing the oracle) and the listener gets its own setting, so the two are
not conflated.

Gates 1-4 pass and are re-runnable. Gates 5-10 need a FIFA client and are
listed in the README, including the Python -> Rust -> Python -> Rust
back-and-forth that proves the rollback path rather than asserting it.

Python backend untouched and still the live runtime; contract suite 446/446
after this work.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
funman300
2026-08-11 01:42:23 +00:00
parent cf961603fe
commit a9eb54ae9c
11 changed files with 1847 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
openfut-blaze-host/tests/live_transport.rs
+472
View File
@@ -0,0 +1,472 @@
//! Transport parity: the recorded conversations over a real TCP socket.
//!
//! The adapter's own suite calls `dispatch()` directly. That proves what the
//! adapter decides, and nothing about what a socket does. This suite runs the
//! actual host process-internally, connects over real TCP, and replays the same
//! recorded conversations — which is what exercises the things fixtures cannot:
//!
//! * a stream that delivers bytes, not frames
//! * requests split across reads and coalesced into one
//! * many frames per connection, and session state surviving between them
//! * multiple frames written back in order for a single request
//! * connection lifecycle and close reasons
//!
//! Byte-exactness is only achievable because `Hooks` injects the session key
//! and clock; with the real ones, no live run could reproduce a recording, and
//! the guarantee would drop to structural.
use std::io::{Read, Write};
use std::net::TcpStream;
use std::sync::Mutex;
use std::time::Duration;
use openfut_adapter_fifa17::blaze::{AdapterConfig, Endpoints, Identity};
use openfut_blaze_host::{bind, config::HostConfig, Hooks};
use openfut_protocol_blaze::fire2::{Frame, Header, HEADER_LEN};
use serde_json::Value as J;
fn records() -> Vec<J> {
let path = format!(
"{}/../openfut-adapter-fifa17/fixtures/blaze_transactions.jsonl",
env!("CARGO_MANIFEST_DIR")
);
std::fs::read_to_string(&path)
.unwrap_or_else(|e| panic!("cannot read {path}: {e}"))
.lines()
.filter(|l| !l.trim().is_empty())
.map(|l| serde_json::from_str(l).expect("valid JSON"))
.collect()
}
fn unhex(s: &str) -> Vec<u8> {
(0..s.len())
.step_by(2)
.map(|i| u8::from_str_radix(&s[i..i + 2], 16).expect("valid hex"))
.collect()
}
fn hex(b: &[u8]) -> String {
b.iter().map(|x| format!("{x:02x}")).collect()
}
fn st(j: &J, k: &str) -> String {
j[k].as_str().expect("string").to_string()
}
fn num(j: &J, k: &str) -> i64 {
j[k].as_i64().expect("number")
}
struct Harness {
addr: String,
records: Vec<J>,
}
/// Start the real host on an ephemeral port with the recording's config.
///
/// Session keys are handed out in the order the recording declares them, so the
/// Nth connection this test opens gets the Nth recorded key. That is why the
/// connection order below must match the fixture's session order.
fn start() -> Harness {
let records = records();
let cfg_rec = records
.iter()
.find(|r| r["kind"] == "config")
.expect("config record")
.clone();
let id = &cfg_rec["identity"];
let adapter_cfg = AdapterConfig {
identity: Identity {
persona_id: num(id, "persona_id"),
persona_name: st(id, "persona_name"),
user_id: num(id, "user_id"),
ext_id: num(id, "ext_id"),
email: st(id, "email"),
namespace: st(id, "namespace"),
client_platform: num(id, "client_platform"),
persona_status: num(id, "persona_status"),
user_session_type: num(id, "user_session_type"),
account_locale: num(id, "account_locale_int"),
locale: st(id, "locale"),
content_id: st(id, "content_id"),
entitlement_tag: st(id, "entitlement_tag"),
entitlement_group: st(id, "entitlement_group"),
title_id: st(id, "title_id"),
client_id: st(id, "client_id"),
platform: st(id, "platform"),
},
endpoints: Endpoints {
advertise: st(&cfg_rec, "advertise"),
bind: st(&cfg_rec, "bind"),
pow_content_host: st(&cfg_rec, "pow_content_host"),
pow_host: st(&cfg_rec, "pow_host"),
..Endpoints::default()
},
server_version: st(id, "server_version"),
};
let host_cfg = HostConfig {
// Port 0: the OS picks a free one, so this can never collide with the
// running Python backend or with a parallel test.
listen_addr: "127.0.0.1".into(),
listen_port: 0,
idle_timeout_secs: 10,
max_payload_bytes: 4 * 1024 * 1024,
trace_path: None,
adapter: adapter_cfg,
};
let keys: Vec<String> = records
.iter()
.filter(|r| r["kind"] == "session")
.map(|r| st(r, "session_key"))
.collect();
let queue = Mutex::new(keys.into_iter().collect::<std::collections::VecDeque<_>>());
let now = num(&cfg_rec, "now");
let hooks = Hooks {
session_key: Box::new(move || {
queue
.lock()
.expect("key queue")
.pop_front()
.expect("a recorded session key for each connection")
}),
now: Box::new(move || now),
};
let server = bind(host_cfg, hooks).expect("bind ephemeral port");
let addr = server.local_addr.to_string();
std::thread::spawn(move || {
let _ = server.run();
});
Harness { addr, records }
}
fn connect(addr: &str) -> TcpStream {
let s = TcpStream::connect(addr).expect("connect to host");
s.set_read_timeout(Some(Duration::from_secs(10))).unwrap();
s.set_nodelay(true).unwrap();
s
}
fn read_frame(stream: &mut TcpStream, buf: &mut Vec<u8>) -> Option<Frame> {
let mut chunk = [0u8; 65536];
while buf.len() < HEADER_LEN {
match stream.read(&mut chunk) {
Ok(0) | Err(_) => return None,
Ok(got) => buf.extend_from_slice(&chunk[..got]),
}
}
let total = Header::parse(&buf[..HEADER_LEN]).ok()?.frame_len();
while buf.len() < total {
match stream.read(&mut chunk) {
Ok(0) | Err(_) => return None,
Ok(got) => buf.extend_from_slice(&chunk[..got]),
}
}
let (frame, used) = Frame::parse(&buf[..total]).ok()?;
buf.drain(..used);
Some(frame)
}
/// Transactions for one recorded session, in order.
fn transactions<'a>(records: &'a [J], session: &str) -> Vec<&'a J> {
records
.iter()
.filter(|r| r["kind"] == "tx" && r["session"] == session)
.collect()
}
fn session_ids(records: &[J]) -> Vec<String> {
records
.iter()
.filter(|r| r["kind"] == "session")
.map(|r| st(r, "id"))
.collect()
}
/// The headline transport test: every recorded conversation, over TCP,
/// byte-for-byte, one connection per recorded session.
#[test]
fn recorded_conversations_replay_byte_for_byte_over_tcp() {
let h = start();
let mut total_frames = 0usize;
for sid in session_ids(&h.records) {
let mut stream = connect(&h.addr);
let mut buf = Vec::new();
for tx in transactions(&h.records, &sid) {
let name = st(tx, "name");
let request = unhex(&st(tx, "request_hex"));
let expected: Vec<String> = tx["responses"]
.as_array()
.unwrap()
.iter()
.map(|v| v.as_str().unwrap().to_string())
.collect();
stream.write_all(&request).expect("send request");
stream.flush().unwrap();
for (i, want) in expected.iter().enumerate() {
let frame = read_frame(&mut stream, &mut buf)
.unwrap_or_else(|| panic!("{sid}/{name}: no frame {i}, expected one"));
assert_eq!(
hex(&frame.encode()),
*want,
"\n{sid}/{name}: frame {i} differs over the wire"
);
total_frames += 1;
}
}
// A clean close between frames must be recognised as such.
drop(stream);
}
assert!(
total_frames >= 50,
"expected the full recorded conversation, saw {total_frames} frames"
);
}
/// The same conversation with every request written ONE BYTE AT A TIME.
///
/// This is the fragmentation case: a real client's frames arrive split across
/// reads, and a host that assumed one read per frame would pass every fixture
/// test and then fail against FIFA.
#[test]
fn requests_split_across_reads_are_reassembled() {
let h = start();
let sid = session_ids(&h.records).into_iter().next().unwrap();
let mut stream = connect(&h.addr);
let mut buf = Vec::new();
for tx in transactions(&h.records, &sid).into_iter().take(6) {
let name = st(tx, "name");
let request = unhex(&st(tx, "request_hex"));
for byte in &request {
stream.write_all(&[*byte]).expect("dribble a byte");
stream.flush().unwrap();
}
for (i, want) in tx["responses"].as_array().unwrap().iter().enumerate() {
let frame = read_frame(&mut stream, &mut buf)
.unwrap_or_else(|| panic!("{name}: no frame {i} after fragmented send"));
assert_eq!(
hex(&frame.encode()),
want.as_str().unwrap(),
"{name} frame {i}"
);
}
}
}
/// Several requests written as ONE write, so the host sees them coalesced in a
/// single read and must split them itself.
#[test]
fn coalesced_requests_in_one_write_are_split() {
let h = start();
let sid = session_ids(&h.records).into_iter().next().unwrap();
let txs: Vec<&J> = transactions(&h.records, &sid).into_iter().take(4).collect();
let mut stream = connect(&h.addr);
let mut buf = Vec::new();
let mut blob = Vec::new();
for tx in &txs {
blob.extend_from_slice(&unhex(&st(tx, "request_hex")));
}
stream.write_all(&blob).expect("one big write");
stream.flush().unwrap();
for tx in &txs {
let name = st(tx, "name");
for (i, want) in tx["responses"].as_array().unwrap().iter().enumerate() {
let frame = read_frame(&mut stream, &mut buf)
.unwrap_or_else(|| panic!("{name}: no frame {i} after coalesced send"));
assert_eq!(
hex(&frame.encode()),
want.as_str().unwrap(),
"{name} frame {i}"
);
}
}
}
/// Login must deliver four frames in order on a real socket, not just from
/// `dispatch()`. This is the ordering guarantee the client depends on.
#[test]
fn login_burst_arrives_in_order_over_the_wire() {
let h = start();
let sid = session_ids(&h.records).into_iter().next().unwrap();
let mut stream = connect(&h.addr);
let mut buf = Vec::new();
for tx in transactions(&h.records, &sid) {
let request = unhex(&st(tx, "request_hex"));
stream.write_all(&request).unwrap();
stream.flush().unwrap();
let expected = tx["responses"].as_array().unwrap().len();
let mut got = Vec::new();
for _ in 0..expected {
got.push(read_frame(&mut stream, &mut buf).expect("frame"));
}
if st(tx, "name") == "login" {
assert_eq!(got.len(), 4, "reply + three pushes");
let routes: Vec<(u16, u16)> = got
.iter()
.map(|f| (f.header.component, f.header.command))
.collect();
assert_eq!(
routes,
vec![
(0x0001, 0x000A), // Authentication::login REPLY
(0x7802, 0x0008), // UserAuthenticated
(0x7802, 0x0001), // UserSessionExtendedDataUpdate
(0x7802, 0x0002), // UserAdded
]
);
return;
}
}
panic!("login transaction never reached");
}
/// Session state must persist across frames on one connection: the auth code
/// login records has to still be there when getAuthToken asks for it later.
#[test]
fn session_state_persists_across_frames_on_one_connection() {
let h = start();
let sid = session_ids(&h.records).into_iter().next().unwrap();
let mut stream = connect(&h.addr);
let mut buf = Vec::new();
let txs = transactions(&h.records, &sid);
let mut saw_before = false;
let mut saw_after = false;
for tx in txs {
let name = st(tx, "name");
stream.write_all(&unhex(&st(tx, "request_hex"))).unwrap();
stream.flush().unwrap();
for want in tx["responses"].as_array().unwrap() {
let frame = read_frame(&mut stream, &mut buf).expect("frame");
let got = hex(&frame.encode());
assert_eq!(got, want.as_str().unwrap(), "{name}");
if name == "get_auth_token_before_login" {
let body = openfut_protocol_blaze::heat2::decode(&frame.payload).unwrap();
let tok = body.get("AUTH").and_then(|v| v.as_str()).unwrap();
assert!(
tok.starts_with("OPENFUT-"),
"synthesised before login: {tok}"
);
saw_before = true;
}
if name == "get_auth_token_after_login" {
let body = openfut_protocol_blaze::heat2::decode(&frame.payload).unwrap();
let tok = body.get("AUTH").and_then(|v| v.as_str()).unwrap();
assert_eq!(tok, "OPENFUT-TEST-AUTHCODE", "echoes the login's code");
saw_after = true;
}
}
}
assert!(
saw_before && saw_after,
"both getAuthToken phases must be exercised"
);
}
/// A separate connection must get a separate session — no leakage.
#[test]
fn a_second_connection_gets_a_fresh_session() {
let h = start();
let ids = session_ids(&h.records);
// Connection 1: log in.
let mut c1 = connect(&h.addr);
let mut b1 = Vec::new();
for tx in transactions(&h.records, &ids[0]) {
c1.write_all(&unhex(&st(tx, "request_hex"))).unwrap();
c1.flush().unwrap();
for _ in 0..tx["responses"].as_array().unwrap().len() {
read_frame(&mut c1, &mut b1).expect("frame");
}
}
// Connection 2 asks for an auth token without logging in. If session state
// leaked it would echo connection 1's code instead of synthesising one.
let mut c2 = connect(&h.addr);
let mut b2 = Vec::new();
let get_token = transactions(&h.records, &ids[0])
.into_iter()
.find(|t| st(t, "name") == "get_auth_token_before_login")
.expect("fixture present");
c2.write_all(&unhex(&st(get_token, "request_hex"))).unwrap();
c2.flush().unwrap();
let frame = read_frame(&mut c2, &mut b2).expect("frame");
let body = openfut_protocol_blaze::heat2::decode(&frame.payload).unwrap();
let tok = body.get("AUTH").and_then(|v| v.as_str()).unwrap();
assert!(
tok.starts_with("OPENFUT-"),
"a fresh connection must not inherit a login: {tok}"
);
assert_ne!(tok, "OPENFUT-TEST-AUTHCODE");
}
/// A header claiming an absurd payload must drop the connection rather than
/// try to allocate for it.
#[test]
fn an_absurd_payload_length_drops_the_connection() {
let h = start();
let mut stream = connect(&h.addr);
let mut header = [0u8; HEADER_LEN];
header[0..4].copy_from_slice(&0x7FFF_FFFFu32.to_be_bytes()); // ~2 GiB
header[6..8].copy_from_slice(&0x0009u16.to_be_bytes());
header[8..10].copy_from_slice(&0x0002u16.to_be_bytes());
stream.write_all(&header).unwrap();
stream.flush().unwrap();
let mut buf = Vec::new();
assert!(
read_frame(&mut stream, &mut buf).is_none(),
"the host must close, not answer, an absurd frame"
);
}
/// A body that will not decode as TDF must not kill the connection: the oracle
/// logs it and dispatches with no fields.
#[test]
fn an_undecodable_body_still_gets_a_reply() {
let h = start();
let mut stream = connect(&h.addr);
// Util::ping with a garbage payload. ping ignores its body, so the reply
// must arrive exactly as if the body had been empty.
let mut frame = Frame::new(
0x0009,
0x0002,
1,
openfut_protocol_blaze::fire2::MsgType::Message,
vec![0xFF; 8],
);
frame.header.payload_len = 8;
stream.write_all(&frame.encode()).unwrap();
stream.flush().unwrap();
let mut buf = Vec::new();
let reply = read_frame(&mut stream, &mut buf).expect("a reply despite the bad body");
assert_eq!(reply.header.component, 0x0009);
assert_eq!(reply.header.command, 0x0002);
assert!(!reply.payload.is_empty(), "ping still answers with STIM");
}