joule-code/joule-code-rs/crates/joule-moq-relay/src/dvr.rs

@@ -111,7 +111,10 @@ use joule_moq::secure_objects::{self, Refusal};
 use joule_moq::{GroupId, ObjectId, TrackName, TrackNamespace};
 use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 use std::io::Write;
+#[cfg(unix)]
 use std::os::unix::fs::FileExt;
+#[cfg(windows)]
+use windows_at::FileExt;
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{mpsc, Arc, Mutex};
@@ -713,6 +716,51 @@ fn read_at(path: &Path, at: u64, len: usize) -> std::io::Result<Vec<u8>> {
     Ok(b)
 }
 
+/// Unix's `read_exact_at` and `write_all_at`, over Windows' `seek_read` and `seek_write`.
+/// These move the file's cursor, which nothing here reads: every segment write is positional.
+#[cfg(windows)]
+mod windows_at {
+    use std::io::{Error, ErrorKind, Result};
+    use std::os::windows::fs::FileExt as _;
+
+    pub trait FileExt {
+        fn read_exact_at(&self, buf: &mut [u8], at: u64) -> Result<()>;
+        fn write_all_at(&self, buf: &[u8], at: u64) -> Result<()>;
+    }
+
+    impl FileExt for std::fs::File {
+        fn read_exact_at(&self, mut buf: &mut [u8], mut at: u64) -> Result<()> {
+            while !buf.is_empty() {
+                match self.seek_read(buf, at) {
+                    Ok(0) => return Err(Error::from(ErrorKind::UnexpectedEof)),
+                    Ok(n) => {
+                        buf = &mut buf[n..];
+                        at += n as u64;
+                    }
+                    Err(e) if e.kind() == ErrorKind::Interrupted => {}
+                    Err(e) => return Err(e),
+                }
+            }
+            Ok(())
+        }
+
+        fn write_all_at(&self, mut buf: &[u8], mut at: u64) -> Result<()> {
+            while !buf.is_empty() {
+                match self.seek_write(buf, at) {
+                    Ok(0) => return Err(Error::from(ErrorKind::WriteZero)),
+                    Ok(n) => {
+                        buf = &buf[n..];
+                        at += n as u64;
+                    }
+                    Err(e) if e.kind() == ErrorKind::Interrupted => {}
+                    Err(e) => return Err(e),
+                }
+            }
+            Ok(())
+        }
+    }
+}
+
 /// A group's record held in memory to be written again: each object with its properties
 /// and payload.
 type Rows = Vec<(DObject, Bytes, Bytes)>;

joule-code/joule-code-rs/crates/joule-moq-relay/tests/cascade.rs

@@ -324,7 +324,13 @@ async fn chain() {
     for (rec, id) in recs.iter().zip([ida, idb, idc]) {
         rec.verify().expect("the record verifies under its hop's key");
         assert_eq!(rec.hop, id, "each hop signs its own record");
-        assert_eq!((rec.joules, rec.class_tag), (0, 0), "each hop's own figure, the unmetered marker, never a sum across hops");
+        // Each hop's own figure: the unmetered marker on a host with no counter, or its own
+        // counter's advance (OnChipCounter) where the relay reads one, as on macOS.
+        match rec.class_tag {
+            0 => assert_eq!(rec.joules, 0, "the unmetered marker is 0 with class tag 0"),
+            4 => {}
+            t => panic!("class tag {t}: the relay signs the unmetered marker or OnChipCounter"),
+        }
     }
     let objects = |rec: &joule_moq_relay::metering::HopRecord, track: [u8; 32], dir: u8| rec.entries.iter().find(|e| e.kind == 1 && e.track == track && e.dir == dir).map_or(0, |e| e.objects);
     for (track, name) in [(dp, "P"), (dq, "Q")] {

joule-code/joule-code-rs/crates/joule-moq-remote/src/main.rs

@@ -55,8 +55,11 @@ fn read_json<T: serde::de::DeserializeOwned>(p: &Path) -> Result<T, String> {
 fn write(p: &Path, body: &str, private: bool) -> Result<(), String> {
     std::fs::write(p, body).map_err(|e| format!("{}: {e}", p.display()))?;
     if private {
-        use std::os::unix::fs::PermissionsExt;
-        std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o600)).map_err(|e| e.to_string())?;
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::PermissionsExt;
+            std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o600)).map_err(|e| e.to_string())?;
+        }
     }
     Ok(())
 }
@@ -179,6 +182,7 @@ fn make_key(a: &[String]) -> Result<(), String> {
     rand_core_06::RngCore::fill_bytes(&mut rand_core_06::OsRng, &mut seed);
     let mut o = std::fs::OpenOptions::new();
     o.write(true).create_new(true);
+    #[cfg(unix)]
     std::os::unix::fs::OpenOptionsExt::mode(&mut o, 0o600);
     let mut f = o.open(&p).map_err(|e| format!("{}: {e}", p.display()))?;
     std::io::Write::write_all(&mut f, format!("{}\n", hex::encode(seed)).as_bytes()).map_err(|e| e.to_string())?;
@@ -188,10 +192,13 @@ fn make_key(a: &[String]) -> Result<(), String> {
 
 /// A host key file, readable by its owner alone.
 fn read_key(p: &Path) -> Result<ed25519_dalek::SigningKey, String> {
-    use std::os::unix::fs::PermissionsExt;
-    let m = std::fs::metadata(p).map_err(|e| format!("{}: {e}", p.display()))?;
-    if m.permissions().mode() & 0o077 != 0 {
-        return Err(format!("{}: the key must be readable by its owner alone (chmod 600)", p.display()));
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::PermissionsExt;
+        let m = std::fs::metadata(p).map_err(|e| format!("{}: {e}", p.display()))?;
+        if m.permissions().mode() & 0o077 != 0 {
+            return Err(format!("{}: the key must be readable by its owner alone (chmod 600)", p.display()));
+        }
     }
     let s = std::fs::read_to_string(p).map_err(|e| format!("{}: {e}", p.display()))?;
     Ok(ed25519_dalek::SigningKey::from_bytes(&room::unhex32(s.trim())?))
@@ -201,6 +208,7 @@ fn read_key(p: &Path) -> Result<ed25519_dalek::SigningKey, String> {
 fn write_private_new(p: &Path, body: &str) -> Result<(), String> {
     let mut o = std::fs::OpenOptions::new();
     o.write(true).create_new(true);
+    #[cfg(unix)]
     std::os::unix::fs::OpenOptionsExt::mode(&mut o, 0o600);
     let mut f = o.open(p).map_err(|e| format!("{}: {e}", p.display()))?;
     std::io::Write::write_all(&mut f, body.as_bytes()).map_err(|e| e.to_string())