blockwise: stream single-record SenML-CBOR vd blobs over Block1 (Step L15)

ThingsBoard wraps every RPC write in SenML-CBOR for a ct="11542 112"
client — including multi-kilobyte blobs, which then ride Block1. The
Block1 lane was opaque-only (structured formats drew 4.13 by design), so
every TB blob write failed on the first block. This teaches the lane the
one SenML shape that can stream: a single-record CBOR pack whose only
value is a definite-length vd byte string, on a resource-level PUT.

parse_cbor_blob_prefix reads the record header off block 0 (all of it
must fit there, or 4.13); the blob bytes then flow into write_chunk
blob-relative, and max_block1_size bounds the blob, not the envelope.

The wire truth, pinned from a live capture against leshan-demo-server:
Leshan (and therefore ThingsBoard) emits the value FIRST and the name
AFTER the blob — [{vd, n}] — so the record's name is generally unknown
until the final block. Fields trailing the blob buffer (<= 512 B, more
is not a name) and parse_cbor_blob_trailing resolves the name at
transfer end; the finale — write_chunk(last) inside the transaction
bracket — runs only after the resolved name matches the request path.
A lying name aborts with 4.00 and nothing commits. Name-leading packs
(this library's own encoder order) validate on block 0 instead.
SenML-JSON, multi-record packs and non-vd values stay 4.13; envelopes
that lie about the blob length draw 4.00 mid-transfer.

Tests: 14 blockwise scenarios (both field orders, end-validation abort
with no transaction opened, trailing spanning blocks with the empty
finale, SZX downshift blob-offset math, cap-excludes-header, trailing
cap, unstreamable/malformed matrices), 7 parser vectors including the
live-captured Leshan prefix bytes, and a Leshan e2e writing a 4 KiB
opaque as SENML_CBOR through real Californium Block1 plus a small-write
dispatch-path leg (verified against a native Leshan before push). The
blockwise fuzz target gained a streaming write_chunk sink and a SenML
Block1 seed (2.3M-run smoke clean).

Version 0.1.6.

Author: suda

Cargo.lock

@@ -8,18 +8,18 @@ members = [
 ]
 
 [workspace.package]
-version = "0.1.5"
+version = "0.1.6"
 edition = "2024"
 rust-version = "1.85"
 license = "Apache-2.0"
 authors = ["Wojtek Siudzinski"]
 
 [workspace.dependencies]
 # Internal
-lekki = { path = "crates/lekki", version = "0.1.5" }
-lekki-mbedtls = { path = "crates/lekki-mbedtls", version = "0.1.5" }
-lekki-testkit = { path = "crates/lekki-testkit", version = "0.1.5" }
-mbedtls-host-sys = { path = "crates/mbedtls-host-sys", version = "0.1.5" }
+lekki = { path = "crates/lekki", version = "0.1.6" }
+lekki-mbedtls = { path = "crates/lekki-mbedtls", version = "0.1.6" }
+lekki-testkit = { path = "crates/lekki-testkit", version = "0.1.6" }
+mbedtls-host-sys = { path = "crates/mbedtls-host-sys", version = "0.1.6" }
 
 coap-lite = "=0.13.3"
 minicbor = "=2.2.2"

Cargo.toml

@@ -16,8 +16,10 @@ Linux hosts and ESP-IDF (ESP32-S3) devices. *Lekki* is Polish for
 ## Scope (what this deliberately is not)
 
 Single server, binding U only. No bootstrap, no Queue Mode, no SMS/TCP, no
-ACLs. Streaming server→client Block1 uploads, client Block2 pull
-downloads (firmware), and client Block1 Register/Update for oversized
+ACLs. Streaming server→client Block1 uploads (raw opaque, and — under
+`lwm2m11` — single-record SenML-CBOR `vd` blobs, the shape ThingsBoard
+wraps every blob write in), client Block2 pull downloads (firmware), and
+client Block1 Register/Update for oversized
 object lists are in scope. Phase A is LwM2M 1.0 (TLV); Phase B adds
 1.1 + SenML-CBOR behind the `lwm2m11` feature; Observe/Notify with
 pmin/pmax (poll-and-diff, all-CON) lives behind the default-off `observe`

README.md

@@ -66,6 +66,11 @@ struct DeviceState {
     timezone: String,
     reboots: u32,
     manifest_len: usize,
+    /// Streaming-write staging for /33099/0/1 (the blob-upload e2e).
+    blob_staged: Vec<u8>,
+    blob_committed: Option<Vec<u8>>,
+    blob_chunks: usize,
+    blob_finishing: bool,
 }
 
 /// Standard-modeled /3 Device facade (Leshan ships the object model, so
@@ -182,6 +187,94 @@ impl Lwm2mObject for ManifestObject {
     }
 }
 
+/// Deliberately **unmodeled** object (no XML anywhere): Leshan encodes
+/// writes to it purely from the REST request's explicit `type`, which is
+/// also how it behaves toward any vendor object it has no model for.
+/// /33099/0/1 is the streaming blob sink (the dongle's profile-upload
+/// shape, miniaturized): offset 0 resets the staging, `last` awaits its
+/// transaction Commit.
+struct BlobSinkObject {
+    state: Arc<Mutex<DeviceState>>,
+}
+
+impl Lwm2mObject for BlobSinkObject {
+    fn oid(&self) -> u16 {
+        33099
+    }
+
+    fn instances(&mut self, out: &mut InstanceLister) {
+        out.instance(0);
+    }
+
+    fn resources(&mut self, _iid: u16, out: &mut ResourceLister) {
+        out.resource(1, ResourceKind::W, Presence::Absent);
+    }
+
+    fn read(&mut self, _: u16, _: u16, _: Option<u16>, _: &mut ResourceWriter<'_>) -> DmResult {
+        Err(CoapStatus::NotFound)
+    }
+
+    fn write(
+        &mut self,
+        iid: u16,
+        rid: u16,
+        _riid: Option<u16>,
+        value: &ResourceReader<'_>,
+    ) -> DmResult {
+        // A single-datagram blob write is the degenerate one-chunk case —
+        // the dongle's /33001/0/1 handler does exactly this.
+        if rid != 1 {
+            return Err(CoapStatus::MethodNotAllowed);
+        }
+        let data = value.opaque()?.to_vec();
+        self.write_chunk(iid, rid, 0, &data, true)
+    }
+
+    fn write_chunk(
+        &mut self,
+        _iid: u16,
+        rid: u16,
+        offset: usize,
+        data: &[u8],
+        last: bool,
+    ) -> DmResult {
+        if rid != 1 {
+            return Err(CoapStatus::MethodNotAllowed);
+        }
+        let mut state = self.state.lock().unwrap();
+        if offset == 0 {
+            state.blob_staged.clear();
+        }
+        if offset != state.blob_staged.len() {
+            return Err(CoapStatus::RequestEntityIncomplete);
+        }
+        state.blob_staged.extend_from_slice(data);
+        state.blob_chunks += 1;
+        if last {
+            state.blob_finishing = true;
+        }
+        Ok(())
+    }
+
+    fn transaction(&mut self, t: lekki::dm::Transaction) -> DmResult {
+        let mut state = self.state.lock().unwrap();
+        match t {
+            lekki::dm::Transaction::Begin => {}
+            lekki::dm::Transaction::Commit if state.blob_finishing => {
+                let staged = core::mem::take(&mut state.blob_staged);
+                state.blob_committed = Some(staged);
+                state.blob_finishing = false;
+            }
+            lekki::dm::Transaction::Commit => {}
+            lekki::dm::Transaction::Rollback => {
+                state.blob_staged.clear();
+                state.blob_finishing = false;
+            }
+        }
+        Ok(())
+    }
+}
+
 // ---- client thread ----------------------------------------------------------
 
 enum Cmd {
@@ -214,6 +307,10 @@ impl Dut {
             timezone: "UTC".into(),
             reboots: 0,
             manifest_len: 4096,
+            blob_staged: Vec::new(),
+            blob_committed: None,
+            blob_chunks: 0,
+            blob_finishing: false,
         }));
         let (cmd_tx, cmd_rx) = mpsc::channel::<Cmd>();
         let (event_tx, event_rx) = mpsc::channel::<ClientEvent>();
@@ -235,6 +332,11 @@ impl Dut {
                 .unwrap();
             client
                 .register_object(Box::new(ManifestObject {
+                    state: Arc::clone(&device_state),
+                }))
+                .unwrap();
+            client
+                .register_object(Box::new(BlobSinkObject {
                     state: device_state,
                 }))
                 .unwrap();
@@ -783,6 +885,66 @@ fn composite_read_against_leshan() {
     dut.shutdown();
 }
 
+/// The ThingsBoard blob-write shape against real Leshan/Californium
+/// (Step L15): a single-resource **opaque write big enough to ride
+/// Block1**, encoded as SENML_CBOR. TB wraps every RPC write in
+/// SenML-CBOR for a `ct="11542 112"` client, so a profile upload arrives
+/// exactly like this — one `[{n, vd}]` record chunked by the server.
+/// The client must parse the record header off block 0 and stream the
+/// blob bytes into `write_chunk` (this is the wire-format pin the unit
+/// tests cannot give: the pack here is built by Leshan's own serializer,
+/// not ours).
+#[cfg(feature = "lwm2m11")]
+#[test]
+#[ignore = "requires the Leshan docker harness (see module docs)"]
+fn blob_write_over_block1_senml_cbor_against_leshan() {
+    let endpoint = "lekki-e2e-blob";
+    provision_psk(endpoint);
+    let dut = Dut::spawn(endpoint);
+    dut.wait_event(|e| *e == ClientEvent::Registered, Duration::from_secs(15));
+
+    // 4 KiB — well past Californium's MAX_MESSAGE_SIZE (1024), so the
+    // request arrives block-wise.
+    let blob: Vec<u8> = (0..4096u32).map(|i| (i % 253) as u8).collect();
+    let body = rest_put(
+        &format!("/api/clients/{endpoint}/33099/0/1?format=SENML_CBOR&timeout=30"),
+        &format!(
+            "{{\"id\":1,\"kind\":\"singleResource\",\"value\":\"{}\",\"type\":\"opaque\"}}",
+            hex(&blob)
+        ),
+    );
+    assert!(body.contains("CHANGED"), "blob write: {body}");
+
+    {
+        let state = dut.state.lock().unwrap();
+        assert_eq!(
+            state.blob_committed.as_deref(),
+            Some(&blob[..]),
+            "committed blob differs from what Leshan sent"
+        );
+        assert!(
+            state.blob_chunks > 1,
+            "expected a chunked (Block1) delivery, got {} chunk(s)",
+            state.blob_chunks
+        );
+    }
+
+    // A small opaque write in the same format lands through the plain
+    // dispatch path (no Block1) — both lanes end in the same staging.
+    let body = rest_put(
+        &format!("/api/clients/{endpoint}/33099/0/1?format=SENML_CBOR&timeout=10"),
+        "{\"id\":1,\"kind\":\"singleResource\",\"value\":\"c0ffee\",\"type\":\"opaque\"}",
+    );
+    assert!(body.contains("CHANGED"), "small blob write: {body}");
+    assert_eq!(
+        dut.state.lock().unwrap().blob_committed.as_deref(),
+        Some(&[0xC0u8, 0xFF, 0xEE][..]),
+        "small SenML opaque write did not land"
+    );
+
+    dut.shutdown();
+}
+
 // ---- Observe e2e (Step L13, feature `observe`) ------------------------------
 
 /// The plan's four-beat observe proof against a real Californium: