Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 39 additions & 5 deletions src/link/sim.rs
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ use std::collections::VecDeque;
use std::time::Duration;

use super::{encode_wire, Link, WireDecoder, WireFormat};
use crate::app_fs::crc16_ccitt;
use crate::device_conn::fake::FakeConn;
use crate::device_conn::{CmdResult, LinkKind};
use crate::ricrest::frame::{text_frame, ElemCode, Frame, MsgType};
Expand All @@ -30,6 +31,8 @@ struct Upload {
block_count: u32,
stream_id: u8,
timer_acks: u32,
// The CRC16 given in ufStart
crc: Option<u16>,
}

#[derive(Default)]
Expand Down Expand Up @@ -63,6 +66,11 @@ pub(crate) struct SimDevice {
// block after it)
pub(crate) ack_out_of_sequence: bool,
out_of_sequence_acked: bool,
// RaftCore from 2026-10-25: an upload is checked against the CRC16 given in ufStart (a mismatch cancels it with
// failCRC; a match is confirmed in the ufEnd reply). Earlier versions ignore the CRC
pub(crate) check_crc: bool,
// The last upload checked against its CRC (file, CRC), for the ufEnd reply
checked_crc: Option<(String, u16)>,
// Timer acks sent (each one is 1s, or 5s before the second block, that a device would have waited)
pub(crate) timer_acks: u32,
// The link's inbound block size (1200 on serial and WebSocket, 500 on BLE); uploads clamp blocks to 2/3 of it
Expand All @@ -88,6 +96,8 @@ impl SimDevice {
faults: Faults::default(),
ack_out_of_sequence: true,
out_of_sequence_acked: false,
check_crc: true,
checked_crc: None,
timer_acks: 0,
chan_block_max: if kind == LinkKind::Ble { 500 } else { 1200 },
upload: None,
Expand All @@ -113,6 +123,7 @@ impl SimDevice {
block_count: 1,
stream_id: self.next_stream_id,
timer_acks: 0,
crc: None,
});
self.next_stream_id += 1;
}
Expand Down Expand Up @@ -167,6 +178,11 @@ impl SimDevice {
let name = json["fileName"].as_str().unwrap_or("").trim_start_matches('/');
let key = if name.starts_with("local/") || name.starts_with("sd/") { name.to_string() } else { format!("local/{}", name) };
self.out_of_sequence_acked = false;
// As strtoul(..., 0): hex with 0x, otherwise decimal
let crc = json["CRC16"].as_str().and_then(|c| match c.strip_prefix("0x").or_else(|| c.strip_prefix("0X")) {
Some(hex) => u16::from_str_radix(hex, 16).ok(),
None => c.parse::<u16>().ok(),
});
let stream_id = self.next_stream_id;
self.next_stream_id = self.next_stream_id.wrapping_add(1).max(1);
self.upload = Some(Upload {
Expand All @@ -178,6 +194,7 @@ impl SimDevice {
block_count: 0,
stream_id,
timer_acks: 0,
crc,
});
vec![reply(msg_num, serde_json::json!({"rslt": "ok", "batchMsgSize": block, "batchAckSize": batch, "streamID": stream_id}))]
}
Expand All @@ -188,7 +205,14 @@ impl SimDevice {
self.store.add(&upload.key, &[]);
}
}
vec![reply(msg_num, serde_json::json!({"rslt": "ok"}))]
// The CRC the file was checked against, as ProtocolExchange gives it
let name = json["fileName"].as_str().unwrap_or("").trim_start_matches('/').to_string();
match self.checked_crc.take() {
Some((key, crc)) if key == name => {
vec![reply(msg_num, serde_json::json!({"rslt": "ok", "crc16": format!("{:04x}", crc)}))]
}
_ => vec![reply(msg_num, serde_json::json!({"rslt": "ok"}))],
}
}
"ufCancel" => {
// Found by stream ID, or by file name if there isn't one (as ProtocolExchange does)
Expand Down Expand Up @@ -259,16 +283,26 @@ impl SimDevice {
}
let okto = upload.data.len();
let mut out = Vec::new();
if ack {
out.push(reply(0, serde_json::json!({"req": "ufBlock", "rslt": "ok", "okto": okto})));
}
if is_final {
let mut upload = self.upload.take().unwrap();
if self.faults.damage_uploads {
upload.data[0] ^= 0xFF;
}
// Checked against the CRC given once written: a mismatch is cancelled (in place of the final ack) and
// the file isn't kept
if let (true, Some(crc)) = (self.check_crc, upload.crc) {
if crc16_ccitt(&upload.data) != crc {
out.push(reply(0, serde_json::json!({"req": "", "rslt": "ok", "cmdName": "ufCancel", "reason": "failCRC"})));
return out;
}
self.checked_crc = Some((upload.key.clone(), crc));
}
self.store.add(&upload.key, &upload.data);
} else if Some(blocks_in) == self.faults.cancel_after_blocks {
}
if ack {
out.push(reply(0, serde_json::json!({"req": "ufBlock", "rslt": "ok", "okto": okto})));
}
if !is_final && Some(blocks_in) == self.faults.cancel_after_blocks {
self.upload = None;
out.push(reply(0, serde_json::json!({"req": "", "rslt": "ok", "cmdName": "ufCancel", "reason": "failRetries"})));
}
Expand Down
7 changes: 5 additions & 2 deletions src/ricrest/conn.rs
Original file line number Diff line number Diff line change
Expand Up @@ -91,9 +91,12 @@ impl<L: Link> DeviceConn for RicrestConn<L> {
let remote = remote.trim_start_matches('/');
// The file system is always named, so --fs is honoured (the device prepends its default FS otherwise)
let file_name = format!("/{}/{}", fs, remote);
upload(&mut self.session.borrow_mut(), &file_name, &data, upload_sizes(self.kind), request_policy(self.kind), progress)?;
let checked = upload(&mut self.session.borrow_mut(), &file_name, &data, upload_sizes(self.kind), request_policy(self.kind), progress)?;

// Check it arrived intact
// Check it arrived intact, unless the device has (against the CRC given in ufStart)
if checked {
return Ok(());
}
check_uploaded(self, &data, remote, fs)
}

Expand Down
26 changes: 24 additions & 2 deletions src/ricrest/tests.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
// RaftCLI: RICREST tests against the simulated device (src/link/sim.rs)

use crate::app_fs::crc16_ccitt;
use std::path::PathBuf;
use std::sync::atomic::{AtomicUsize, Ordering};

Expand Down Expand Up @@ -61,14 +62,26 @@ fn uploads_use_the_devices_block_size() {
let sizes: Vec<usize> = blocks.iter().map(|b| b.rsplit(' ').next().unwrap().parse().unwrap()).collect();
assert_eq!(sizes.iter().max(), Some(&max_block), "{:?}", kind);
assert_eq!(sizes.iter().sum::<usize>(), 5000, "each block sent once: {:?}", blocks);
// The file system is named in the upload, and the result is checked
// The file system is named in the upload, with the file's CRC, which the device checks (so no filehash)
let start = &received(&conn, r#"{"cmdName":"ufStart""#)[0];
assert!(start.contains(r#""fileName":"/local/web/app.js""#), "{}", start);
assert_eq!(received(&conn, "filehash/local/web~app.js").len(), 1);
assert!(start.contains(&format!(r#""CRC16":"0x{:04x}""#, crc16_ccitt(&bytes))), "{}", start);
assert_eq!(received(&conn, "filehash/").len(), 0);
assert!(!conn.session().link().device.uploading());
}
}

#[test]
fn uploads_to_older_firmware_are_checked_with_filehash() {
// Firmware before 2026-10-25 ignores the CRC in ufStart (no crc16 in the ufEnd reply)
let conn = sim(LinkKind::WebSocket);
conn.session_mut().link_mut().device.check_crc = false;
let bytes = data(5000);
put(&conn, "web/app.js", &bytes).unwrap();
assert_eq!(stored(&conn, "local/web/app.js").unwrap(), bytes);
assert_eq!(received(&conn, "filehash/local/web~app.js").len(), 1);
}

// Upload 200 blocks (WebSocket: 800-byte blocks, batches of 10) with blocks at these positions (in 100s of bytes)
// lost; returns (blocks sent, timer acks the device needed to send)
fn upload_with_losses(ack_out_of_sequence: bool, lose: &[u32]) -> (usize, u32) {
Expand Down Expand Up @@ -140,7 +153,16 @@ fn a_silent_device_is_given_up_on_and_the_upload_cancelled() {

#[test]
fn a_damaged_upload_is_removed() {
// Firmware that checks the CRC cancels the upload (failCRC) and doesn't keep the file
let conn = sim(LinkKind::Serial);
conn.session_mut().link_mut().device.faults.damage_uploads = true;
let err = put(&conn, "x.bin", &data(1000)).unwrap_err().to_string();
assert!(err.contains("didn't match"), "{}", err);
assert!(stored(&conn, "local/x.bin").is_none());

// Older firmware keeps it: filehash finds the damage and the file is deleted
let conn = sim(LinkKind::Serial);
conn.session_mut().link_mut().device.check_crc = false;
conn.session_mut().link_mut().device.faults.damage_uploads = true;
let err = put(&conn, "x.bin", &data(1000)).unwrap_err().to_string();
assert!(err.contains("arrived damaged"), "{}", err);
Expand Down
33 changes: 26 additions & 7 deletions src/ricrest/upload.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,13 +6,16 @@
// arriving); each batch is sent from the latest ack (go-back-N)
// ufEnd -> done (the device has usually closed the session after the last block)
// On an error after ufStart, ufCancel is sent so the next upload doesn't find the transfer still in progress.
// The device doesn't check a CRC, so RicrestConn checks the file with filehash afterwards.
// ufStart gives the file's CRC16. Firmware from 2026-10-25 checks the file against it once written: a mismatch
// cancels the upload (failCRC, the file deleted) and a match is confirmed in the ufEnd reply (crc16). Older firmware
// ignores it, so without that confirmation RicrestConn checks the file with filehash afterwards.
// Rob Dobson 2026

use std::time::{Duration, Instant};

use super::frame::{file_block, ElemCode, Frame, MsgType};
use super::session::{RequestPolicy, Session};
use crate::app_fs::crc16_ccitt;
use crate::device_conn::CmdResult;
use crate::http_client::ProgressFn;
use crate::link::Link;
Expand All @@ -33,24 +36,27 @@ pub struct UploadSizes {
pub max_wire_len: Option<usize>,
}

// Upload a file; returns true if the device checked it against its CRC (so it needn't be checked again)
pub fn upload<L: Link>(
session: &mut Session<L>,
file_name: &str,
data: &[u8],
sizes: UploadSizes,
policy: RequestPolicy,
progress: &mut ProgressFn,
) -> CmdResult<()> {
) -> CmdResult<bool> {
if data.len() >= MAX_FILE_LEN {
return Err(format!("{} is too big to upload this way (16 MB or more)", file_name).into());
}
let name_json = serde_json::to_string(file_name)?;
let crc = crc16_ccitt(data);
let start = format!(
r#"{{"cmdName":"ufStart","reqStr":"fileupload","fileType":"fs","fileName":{},"fileLen":{},"batchMsgSize":{},"batchAckSize":{}}}"#,
r#"{{"cmdName":"ufStart","reqStr":"fileupload","fileType":"fs","fileName":{},"fileLen":{},"batchMsgSize":{},"batchAckSize":{},"CRC16":"0x{:04x}"}}"#,
name_json,
data.len(),
sizes.block,
sizes.batch
sizes.batch,
crc
);
let mut reply = command(session, &start, policy)?;
// An earlier upload of this file was interrupted (Ctrl-C, a lost link) and the device is still waiting for it
Expand Down Expand Up @@ -85,7 +91,20 @@ pub fn upload<L: Link>(
if reply["rslt"].as_str() != Some("ok") {
return Err(format!("the device didn't finish the upload of {}: {}", file_name, reply).into());
}
Ok(())
// The device checked the file against the CRC given (firmware from 2026-10-25)
let checked = reply["crc16"].as_str().and_then(|c| u16::from_str_radix(c, 16).ok()) == Some(crc);
Ok(checked)
}

// The error for a reason the device gave for cancelling or failing an upload
fn upload_failed(reason: &str, cancelled: bool) -> String {
match reason {
"failCRC" | "CRCMismatch" => {
"the file the device wrote didn't match what was sent (CRC check), so it deleted it - try again".to_string()
}
_ if cancelled => format!("the device cancelled the upload ({})", reason),
_ => format!("the device failed to write the file ({})", reason),
}
}

// A JSON command frame and its JSON reply
Expand Down Expand Up @@ -154,10 +173,10 @@ fn send_blocks<L: Link>(
while let Some(frame) = session.wait_for(|f| upload_status(f).is_some(), timeout)? {
let json = upload_status(&frame).unwrap_or_default();
if json["cmdName"].as_str() == Some("ufCancel") {
return Err(format!("the device cancelled the upload ({})", json["reason"].as_str().unwrap_or("no reason")).into());
return Err(upload_failed(json["reason"].as_str().unwrap_or("no reason"), true).into());
}
if json["rslt"].as_str() == Some("fail") {
return Err(format!("the device failed to write the file ({})", json["reason"].as_str().unwrap_or("no reason")).into());
return Err(upload_failed(json["reason"].as_str().unwrap_or("no reason"), false).into());
}
if let Some(pos) = json["okto"].as_u64() {
let pos = (pos as usize).min(data.len());
Expand Down
Loading