Forwarder architecture fixes: trust model, data integrity, and decomposition

apps/forwarder-ui/src/lib/api.test.ts

@@ -34,6 +34,8 @@ describe("forwarder api client", () => {
           approval_state: "pending",
           waiting_for_approval: true,
           message: "Waiting for server admin approval",
+          cached: true,
+          checked_unix_ms: 1700000000000,
         },
         readers: [],
       }),

apps/forwarder-ui/src/lib/api.ts

@@ -50,6 +50,8 @@ export interface ServerDeviceStatus {
   approval_state: string | null;
   waiting_for_approval: boolean;
   message: string | null;
+  cached: boolean;
+  checked_unix_ms: number | null;
 }
 
 export interface ForwarderStatus {

apps/forwarder-ui/src/lib/config-load.test.ts

@@ -25,6 +25,8 @@ function makeStatus(restart_needed: boolean): ForwarderStatus {
       approval_state: null,
       waiting_for_approval: false,
       message: null,
+      cached: true,
+      checked_unix_ms: null,
     },
     readers: [],
   };

apps/forwarder-ui/src/lib/reader-status-cache.test.ts

@@ -30,6 +30,8 @@ function makeStatus(readers: ForwarderStatus["readers"]): ForwarderStatus {
       approval_state: null,
       waiting_for_approval: false,
       message: null,
+      cached: true,
+      checked_unix_ms: null,
     },
     readers,
   };

apps/forwarder-ui/src/lib/sse.ts

@@ -1,5 +1,10 @@
 import { createSSE, type SseHandle } from "@rusty-timer/shared-ui/lib/sse";
-import type { ReaderInfo, ReaderStatus, UpdateStatusResponse } from "./api";
+import type {
+  ReaderInfo,
+  ReaderStatus,
+  ServerDeviceStatus,
+  UpdateStatusResponse,
+} from "./api";
 
 export type ForwarderSseCallbacks = {
   onStatusChanged: (data: {
@@ -19,6 +24,7 @@ export type ForwarderSseCallbacks = {
     available: boolean;
     status: any | null;
   }) => void;
+  onServerStatusChanged?: (payload: { server: ServerDeviceStatus }) => void;
 };
 
 let handle: SseHandle | null = null;
@@ -57,6 +63,9 @@ export function initSSE(callbacks: ForwarderSseCallbacks): void {
       }) => {
         callbacks.onUpsStatusChanged?.(data);
       },
+      server_status_changed: (data: { server: ServerDeviceStatus }) => {
+        callbacks.onServerStatusChanged?.(data);
+      },
     },
     (connected) => {
       callbacks.onConnectionChange(connected);

apps/forwarder-ui/src/routes/+page.svelte

@@ -508,6 +508,11 @@
       onUpsStatusChanged: (payload) => {
         upsState = { available: payload.available, status: payload.status };
       },
+      onServerStatusChanged: (payload) => {
+        if (status) {
+          status = { ...status, server: payload.server };
+        }
+      },
       onUpdateStatusChanged: (us) => {
         if (
           (us.status === "available" || us.status === "downloaded") &&

apps/receiver-ui/src/lib/components/ForwarderConfigModal.svelte

@@ -257,11 +257,15 @@
 
             <Card title="P2P / Server">
               <div class="space-y-3">
+                <p class="text-xs text-text-muted">
+                  Managed locally on the forwarder — not editable remotely.
+                </p>
                 <label class="block text-sm font-medium text-text-secondary">
                   <span class="inline-flex items-center gap-2">
                     <input
                       class="accent-accent"
                       type="checkbox"
+                      disabled
                       bind:checked={config.p2p.enabled}
                     />
                     Enable P2P
@@ -270,16 +274,18 @@
                 <label class="block text-sm font-medium text-text-secondary">
                   Server URL
                   <input
-                    class="mt-1 {inputClass}"
+                    class="mt-1 {inputClass} opacity-50"
                     type="text"
+                    disabled
                     bind:value={config.p2p.server_url}
                   />
                 </label>
                 <label class="block text-sm font-medium text-text-secondary">
                   Server token file
                   <input
-                    class="mt-1 {inputClass}"
+                    class="mt-1 {inputClass} opacity-50"
                     type="text"
+                    disabled
                     bind:value={config.p2p.server_token_file}
                   />
                 </label>
@@ -322,11 +328,15 @@
 
             <Card title="Control">
               <div class="space-y-3">
+                <p class="text-xs text-text-muted">
+                  Managed locally on the forwarder — not editable remotely.
+                </p>
                 <label class="block text-sm font-medium text-text-secondary">
                   <span class="inline-flex items-center gap-2">
                     <input
                       class="accent-accent"
                       type="checkbox"
+                      disabled
                       bind:checked={config.control.allow_power_actions}
                     />
                     Allow power actions
@@ -337,6 +347,7 @@
                     <input
                       class="accent-accent"
                       type="checkbox"
+                      disabled
                       bind:checked={config.control.allow_remote_config}
                     />
                     Allow remote config

apps/receiver-ui/src/lib/forwarder-config-modal.test.ts

@@ -132,6 +132,49 @@ describe("ForwarderConfigModal", () => {
     expect(submitted.update?.mode).toBeUndefined();
   });
 
+  it("renders protected sections read-only and round-trips their values verbatim", async () => {
+    render(ForwarderConfigModal, {
+      open: true,
+      endpointId: "endpoint-1",
+      displayName: "Timing Forwarder",
+      onClose: vi.fn(),
+    });
+
+    await screen.findByLabelText("Display name");
+
+    // Protected [p2p] and [control] inputs must be disabled.
+    expect(screen.getByLabelText("Enable P2P")).toBeDisabled();
+    expect(screen.getByLabelText("Server URL")).toBeDisabled();
+    expect(screen.getByLabelText("Server token file")).toBeDisabled();
+    expect(screen.getByLabelText("Allow power actions")).toBeDisabled();
+    expect(screen.getByLabelText("Allow remote config")).toBeDisabled();
+
+    // Each protected card explains why the fields are read-only.
+    expect(
+      screen.getAllByText(
+        "Managed locally on the forwarder — not editable remotely.",
+      ),
+    ).toHaveLength(2);
+
+    // Operational fields stay editable.
+    expect(screen.getByLabelText("Display name")).toBeEnabled();
+    expect(screen.getByLabelText("SQLite path")).toBeEnabled();
+
+    await fireEvent.click(screen.getByTestId("forwarder-config-save"));
+
+    await waitFor(() => {
+      expect(mockApi.setForwarderConfig).toHaveBeenCalled();
+    });
+
+    // Protected sections must round-trip unchanged — the forwarder rejects
+    // any write whose [auth]/[p2p]/[control] differ from the on-disk config.
+    const submitted = JSON.parse(mockApi.setForwarderConfig.mock.calls[0][1]);
+    const fetched = sampleConfig();
+    expect(submitted.p2p).toEqual(fetched.p2p);
+    expect(submitted.control).toEqual(fetched.control);
+    expect(submitted.auth).toEqual(fetched.auth);
+  });
+
   it("shows a restart banner after a save that requires restart and restarts on demand", async () => {
     mockApi.setForwarderConfig.mockResolvedValue({
       ok: true,

crates/rt-p2p-protocol/src/codec.rs

@@ -23,16 +23,23 @@ pub fn encode_frame(message: &impl Message) -> Bytes {
     buf.freeze()
 }
 
+/// Decodes and validates a little-endian frame payload length prefix.
+pub fn decode_frame_len(len_buf: [u8; 4]) -> Result<usize, ProtocolError> {
+    let len = u32::from_le_bytes(len_buf) as usize;
+    if len > MAX_FRAME_BYTES {
+        return Err(ProtocolError::frame_too_large(len));
+    }
+
+    Ok(len)
+}
+
 /// Decodes one complete length-prefixed frame, if enough bytes are buffered.
 pub fn decode_frame(buf: &mut BytesMut) -> Result<Option<Frame>, ProtocolError> {
     if buf.len() < 4 {
         return Ok(None);
     }
 
-    let len = u32::from_le_bytes(buf[..4].try_into().expect("slice has 4 bytes")) as usize;
-    if len > MAX_FRAME_BYTES {
-        return Err(ProtocolError::frame_too_large());
-    }
+    let len = decode_frame_len(buf[..4].try_into().expect("slice has 4 bytes"))?;
 
     if buf.len() < 4 + len {
         return Ok(None);
@@ -42,6 +49,19 @@ pub fn decode_frame(buf: &mut BytesMut) -> Result<Option<Frame>, ProtocolError>
     Ok(Some(buf.split_to(len).freeze()))
 }
 
+/// Decodes a protobuf message from a previously read length prefix and payload.
+pub fn decode_frame_payload<M>(len_buf: [u8; 4], payload: &[u8]) -> Result<M, ProtocolError>
+where
+    M: Message + Default,
+{
+    let len = decode_frame_len(len_buf)?;
+    if payload.len() != len {
+        return Err(ProtocolError::protocol_violation());
+    }
+
+    M::decode(payload).map_err(ProtocolError::decode_error)
+}
+
 /// Decodes one complete length-prefixed protobuf message, if enough bytes are buffered.
 pub fn decode_message_frame<M>(buf: &mut BytesMut) -> Result<Option<M>, ProtocolError>
 where

crates/rt-p2p-protocol/src/error.rs

@@ -103,6 +103,8 @@ pub enum ProtocolError {
         retryable: bool,
         /// Stream the error pertains to, if any.
         stream_id: Option<Vec<u8>>,
+        /// Advertised payload length.
+        length: usize,
     },
     /// A protobuf payload could not be decoded.
     #[error("decode error: {source}")]
@@ -188,10 +190,18 @@ impl ProtocolError {
         }
     }
 
-    pub(crate) fn frame_too_large() -> Self {
+    pub(crate) fn frame_too_large(length: usize) -> Self {
         Self::FrameTooLarge {
             retryable: false,
             stream_id: None,
+            length,
+        }
+    }
+
+    pub(crate) fn protocol_violation() -> Self {
+        Self::ProtocolViolation {
+            retryable: false,
+            stream_id: None,
         }
     }
 

crates/rt-p2p-protocol/src/lib.rs

@@ -45,7 +45,10 @@ pub mod proto {
 }
 
 pub use capabilities::{CAP_CONTROL_EVENTS, CAP_READER_CONTROL, CAP_REMOTE_CONFIG, has_capability};
-pub use codec::{Frame, MAX_FRAME_BYTES, decode_frame, decode_message_frame, encode_frame};
+pub use codec::{
+    Frame, MAX_FRAME_BYTES, decode_frame, decode_frame_len, decode_frame_payload,
+    decode_message_frame, encode_frame,
+};
 pub use error::{ProtocolError, ProtocolErrorCode};
 pub use generated::{
     Ack, CaughtUp, ConfigGetRequest, ConfigGetResponse, ConfigSetRequest, ConfigSetResponse,

crates/rt-p2p-protocol/tests/codec_negotiate.rs

@@ -2,7 +2,7 @@ use bytes::{BufMut, BytesMut};
 use prost::Message;
 use rt_p2p_protocol::{
     CAP_CONTROL_EVENTS, CAP_REMOTE_CONFIG, Hello, Ping, ProtocolError, decode_frame,
-    decode_message_frame, encode_frame, has_capability, negotiate,
+    decode_frame_payload, decode_message_frame, encode_frame, has_capability, negotiate,
 };
 
 #[test]
@@ -26,6 +26,39 @@ fn roundtrip_frame() {
     assert_eq!(decoded, original);
 }
 
+#[test]
+fn decode_frame_payload_roundtrips_encoded_message() {
+    let original = Ping { nonce: 42 };
+    let encoded = encode_frame(&original);
+    let len_buf = encoded[..4].try_into().expect("length prefix");
+
+    let decoded = decode_frame_payload::<Ping>(len_buf, &encoded[4..]).expect("decode payload");
+
+    assert_eq!(decoded, original);
+}
+
+#[test]
+fn decode_frame_payload_rejects_oversize_prefix() {
+    let len_buf = u32::try_from(8 * 1024 * 1024 + 1)
+        .expect("fits u32")
+        .to_le_bytes();
+
+    let err = decode_frame_payload::<Ping>(len_buf, &[]).expect_err("oversize frame");
+
+    assert!(matches!(err, ProtocolError::FrameTooLarge { .. }));
+}
+
+#[test]
+fn decode_frame_payload_rejects_payload_shorter_than_prefix() {
+    let encoded = encode_frame(&Ping { nonce: 42 });
+    let len_buf = encoded[..4].try_into().expect("length prefix");
+
+    let err = decode_frame_payload::<Ping>(len_buf, &encoded[4..encoded.len() - 1])
+        .expect_err("truncated payload");
+
+    assert!(matches!(err, ProtocolError::ProtocolViolation { .. }));
+}
+
 #[test]
 fn partial_read_returns_none() {
     let encoded = encode_frame(&Ping { nonce: 7 });

crates/rt-server-api/src/catalog.rs

@@ -22,3 +22,11 @@ pub struct ForwarderCatalogResponse {
     pub endpoint_id: String,
     pub stream_count: usize,
 }
+
+/// Wire-format response for `GET /forwarder/catalog`: the caller's own stored
+/// stream catalog (per-stream `epoch`/`next_seq` high-water), used by a forwarder
+/// to restore stream identity after local journal loss.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ForwarderOwnCatalogResponse {
+    pub streams: Vec<ForwarderCatalogStream>,
+}

crates/rt-test-utils/src/p2p/mod.rs

@@ -26,7 +26,7 @@ use std::time::Duration;
 
 use prost::Message;
 use rt_iroh::{RecvStream, SendStream};
-use rt_p2p_protocol::{MAX_FRAME_BYTES, encode_frame};
+use rt_p2p_protocol::{decode_frame_len, decode_frame_payload, encode_frame};
 
 pub use mock_forwarder::{ForwarderScript, MockForwarderPeer};
 pub use mock_receiver::{DataSubscription, MockReceiverPeer, ReceiverSession};
@@ -47,14 +47,10 @@ where
 {
     let mut len_buf = [0u8; 4];
     recv.read_exact(&mut len_buf).await?;
-    let len = u32::from_le_bytes(len_buf) as usize;
-    if len > MAX_FRAME_BYTES {
-        return Err(format!("frame length {len} exceeds MAX_FRAME_BYTES {MAX_FRAME_BYTES}").into());
-    }
-
+    let len = decode_frame_len(len_buf)?;
     let mut payload = vec![0u8; len];
     recv.read_exact(&mut payload).await?;
-    Ok(M::decode(payload.as_slice())?)
+    Ok(decode_frame_payload(len_buf, payload.as_slice())?)
 }
 
 /// Test-only connectivity fault shim.