cultcache-rs: decode_payload grows its stack through serde_stacker

A payload inside the 256 nesting bound no longer depends on the caller's
thread stack: the rmp-serde deserializer runs under serde_stacker with a
256 KiB red zone and 2 MiB segments. Adds tests/decode_stack.rs (child
process per test, sized decode thread) and renames the 2 MiB union_parity
stack test to a 1 MiB pin.

Not yet built or run at this commit.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

packages/cultcache-rs/Cargo.toml

@@ -22,6 +22,7 @@ redb = "4.1.0"
 serde = { version = "1", features = ["derive"] }
 serde_bytes = "0.11"
 serde_json = "1"
+serde_stacker = "0.1.14"
 uuid = { version = "1", features = ["v4"] }
 
 [dev-dependencies]

packages/cultcache-rs/README.md

@@ -456,6 +456,16 @@ That is the Rust version of the original trick:
 The derive/registry work should preserve that shape. Its job is to remove
 manual registration ceremony, not to loosen the closed-world boundary.
 
+Payload nesting is bounded at 256 MessagePack containers, checked before any
+recursion, and typed decode then runs under `serde_stacker`, which moves onto a
+fresh stack segment whenever less than 256 KiB of stack remains. A payload inside
+the bound therefore decodes on a small thread instead of aborting the process;
+past the bound the refusal is a typed `PayloadTooDeep`. The residual is a single
+stack frame larger than the 256 KiB red zone. Stack growth on Windows is not yet
+verified. On wasm32 the shadow stack grows the same way, but the engine's own
+call stack remains the host's limit. Building the crate needs a C toolchain
+(`psm`, `stacker`).
+
 ## License
 
 Private GameCult infrastructure for now. Public packaging can wait until the

packages/cultcache-rs/src/lib.rs

@@ -595,12 +595,30 @@ fn element_header(bytes: &[u8], position: usize) -> Option<(usize, Option<usize>
     })
 }
 
+/// The stack left below which a decode moves to a fresh segment. Measured (map D4): the worst
+/// distance between two stack checks is 128-144 KiB in a debug build and under 32 KiB in release,
+/// so 256 KiB leaves a margin in both. A frame larger than this between two checks still overruns.
+const DECODE_RED_ZONE: usize = 256 * 1024;
+
+/// The size of each segment a decode grows onto. It does not change the cost of a decode (map D6).
+const DECODE_STACK_SEGMENT: usize = 2 * 1024 * 1024;
+
 /// The one decode of a document payload: the nesting bound first, then the deserializer, whose own
 /// depth counter (1024) is only a backstop. A failure keeps its kind and never quotes the bytes.
 /// Readers of stored envelope payloads outside this crate decode through it too.
+///
+/// The deserializer runs under `serde_stacker`, which moves onto a fresh stack segment whenever less
+/// than the red zone remains, at every container level serde's impls descend through, skipped values
+/// included. The calling thread's stack size therefore never decides whether a payload inside the
+/// bound decodes. The one residual is a single frame larger than the red zone. On wasm32 the shadow
+/// stack grows the same way, while the engine's own call stack stays the host's limit.
 pub fn decode_payload<T: serde::de::DeserializeOwned>(payload: &[u8]) -> Result<T> {
     check_payload_depth(payload)?;
-    rmp_serde::from_slice(payload).map_err(opaque_decode_error)
+    let mut reader = rmp_serde::Deserializer::from_read_ref(payload);
+    let mut grown = serde_stacker::Deserializer::new(&mut reader);
+    grown.red_zone = DECODE_RED_ZONE;
+    grown.stack_size = DECODE_STACK_SEGMENT;
+    <T as serde::Deserialize>::deserialize(grown).map_err(opaque_decode_error)
 }
 
 /// The nesting bound for an envelope admitted without a typed decode, from the same walk that

packages/cultcache-rs/tests/decode_stack.rs

@@ -0,0 +1,239 @@
+//! Typed decode never aborts a payload inside the nesting bound, whatever the caller's stack.
+//! Each test runs its decode in a child test process on a thread of a stated size, so an abort
+//! fails one named test. Values are built, encoded, compared and dropped on a large thread; only
+//! the decode runs on the small one.
+
+use cultcache_rs::CultSlots;
+use cultcache_rs::MAX_PAYLOAD_DEPTH;
+use cultcache_rs::PayloadTooDeep;
+use cultcache_rs::decode_payload;
+use serde::de::DeserializeOwned;
+
+const CHILD_MARKER: &str = "CULTCACHE_DECODE_STACK_CHILD";
+const KIB: usize = 1024;
+
+#[derive(Clone, Debug, PartialEq, CultSlots)]
+struct Wide32 {
+    #[cultcache(key = 0)]
+    a0: [u64; 32],
+    #[cultcache(key = 1)]
+    a1: [u64; 32],
+    #[cultcache(key = 2)]
+    a2: [u64; 32],
+    #[cultcache(key = 3)]
+    a3: [u64; 32],
+    #[cultcache(key = 4)]
+    a4: [u64; 32],
+    #[cultcache(key = 5)]
+    a5: [u64; 32],
+    #[cultcache(key = 6)]
+    a6: [u64; 32],
+    #[cultcache(key = 7)]
+    a7: [u64; 32],
+    #[cultcache(key = 8)]
+    a8: [u64; 32],
+    #[cultcache(key = 9)]
+    a9: [u64; 32],
+    #[cultcache(key = 10)]
+    a10: [u64; 32],
+    #[cultcache(key = 11)]
+    a11: [u64; 32],
+    #[cultcache(key = 12)]
+    a12: [u64; 32],
+    #[cultcache(key = 13)]
+    a13: [u64; 32],
+    #[cultcache(key = 14)]
+    a14: [u64; 32],
+    #[cultcache(key = 15)]
+    a15: [u64; 32],
+    #[cultcache(key = 16)]
+    a16: [u64; 32],
+    #[cultcache(key = 17)]
+    a17: [u64; 32],
+    #[cultcache(key = 18)]
+    a18: [u64; 32],
+    #[cultcache(key = 19)]
+    a19: [u64; 32],
+    #[cultcache(key = 20)]
+    a20: [u64; 32],
+    #[cultcache(key = 21)]
+    a21: [u64; 32],
+    #[cultcache(key = 22)]
+    a22: [u64; 32],
+    #[cultcache(key = 23)]
+    a23: [u64; 32],
+    #[cultcache(key = 24)]
+    a24: [u64; 32],
+    #[cultcache(key = 25)]
+    a25: [u64; 32],
+    #[cultcache(key = 26)]
+    a26: [u64; 32],
+    #[cultcache(key = 27)]
+    a27: [u64; 32],
+    #[cultcache(key = 28)]
+    a28: [u64; 32],
+    #[cultcache(key = 29)]
+    a29: [u64; 32],
+    #[cultcache(key = 30)]
+    a30: [u64; 32],
+    #[cultcache(key = 31)]
+    a31: [u64; 32],
+    #[cultcache(key = 32)]
+    next: Option<Box<Wide32>>,
+}
+
+#[derive(Clone, Debug, PartialEq, CultSlots)]
+struct One {
+    #[cultcache(key = 0)]
+    a: u8,
+}
+
+#[derive(Clone, Debug, PartialEq, CultSlots)]
+struct DynamicSlot {
+    #[cultcache(key = 0)]
+    v: serde_json::Value,
+}
+
+/// Runs `body` on a thread with a stack of `bytes`, returning what it returns.
+fn on_stack<R: Send + 'static>(bytes: usize, body: impl FnOnce() -> R + Send + 'static) -> R {
+    std::thread::Builder::new().stack_size(bytes).spawn(body).unwrap().join().unwrap()
+}
+
+/// Runs `body` in a child process that runs only the test `name`; the parent passes when the
+/// child exits cleanly having run exactly that one test. An abort in the child fails the parent.
+fn in_child(name: &str, body: impl FnOnce()) {
+    if std::env::var_os(CHILD_MARKER).is_some() {
+        body();
+        return;
+    }
+    let output = std::process::Command::new(std::env::current_exe().unwrap())
+        .args([name, "--exact", "--nocapture", "--test-threads=1"])
+        .env(CHILD_MARKER, "1")
+        .output()
+        .unwrap();
+    let stdout = String::from_utf8_lossy(&output.stdout);
+    let stderr = String::from_utf8_lossy(&output.stderr);
+    assert!(
+        output.status.success() && stdout.contains("1 passed"),
+        "child {name} failed: {:?}
+{stdout}
+{stderr}",
+        output.status
+    );
+}
+
+/// A chain of `levels` Wide32 records: `levels` containers deep, plus one for its last arrays.
+fn wide_chain(levels: usize) -> Wide32 {
+    let mut chain: Option<Box<Wide32>> = None;
+    for level in (0..levels).rev() {
+        let mut node = Wide32 { next: chain, ..empty_wide() };
+        node.a0[0] = level as u64;
+        chain = Some(Box::new(node));
+    }
+    *chain.unwrap()
+}
+
+fn empty_wide() -> Wide32 {
+    Wide32 {
+        a0: [0; 32],
+        a1: [0; 32],
+        a2: [0; 32],
+        a3: [0; 32],
+        a4: [0; 32],
+        a5: [0; 32],
+        a6: [0; 32],
+        a7: [0; 32],
+        a8: [0; 32],
+        a9: [0; 32],
+        a10: [0; 32],
+        a11: [0; 32],
+        a12: [0; 32],
+        a13: [0; 32],
+        a14: [0; 32],
+        a15: [0; 32],
+        a16: [0; 32],
+        a17: [0; 32],
+        a18: [0; 32],
+        a19: [0; 32],
+        a20: [0; 32],
+        a21: [0; 32],
+        a22: [0; 32],
+        a23: [0; 32],
+        a24: [0; 32],
+        a25: [0; 32],
+        a26: [0; 32],
+        a27: [0; 32],
+        a28: [0; 32],
+        a29: [0; 32],
+        a30: [0; 32],
+        a31: [0; 32],
+        next: None,
+    }
+}
+
+fn decode_on<T: DeserializeOwned + Send + 'static>(stack: usize, bytes: Vec<u8>) -> anyhow::Result<T> {
+    on_stack(stack, move || decode_payload::<T>(&bytes))
+}
+
+#[test]
+fn a_wide_record_at_the_limit_decodes_on_a_1mib_thread() {
+    in_child("a_wide_record_at_the_limit_decodes_on_a_1mib_thread", || {
+        let bytes = on_stack(64 * KIB * KIB, || {
+            rmp_serde::to_vec(&wide_chain(MAX_PAYLOAD_DEPTH - 1)).unwrap()
+        });
+        let decoded = decode_on::<Wide32>(1024 * KIB, bytes).unwrap();
+        on_stack(64 * KIB * KIB, move || {
+            assert!(decoded == wide_chain(MAX_PAYLOAD_DEPTH - 1));
+        });
+    });
+}
+
+#[test]
+fn one_past_the_limit_is_refused_typed_on_a_1mib_thread() {
+    in_child("one_past_the_limit_is_refused_typed_on_a_1mib_thread", || {
+        let bytes = on_stack(64 * KIB * KIB, || {
+            rmp_serde::to_vec(&wide_chain(MAX_PAYLOAD_DEPTH)).unwrap()
+        });
+        let error = decode_on::<Wide32>(1024 * KIB, bytes).unwrap_err();
+        assert_eq!(
+            error.downcast_ref::<PayloadTooDeep>(),
+            Some(&PayloadTooDeep { depth: MAX_PAYLOAD_DEPTH + 1, limit: MAX_PAYLOAD_DEPTH })
+        );
+    });
+}
+
+#[test]
+fn a_foreign_value_at_the_limit_decodes_on_a_64kib_thread() {
+    in_child("a_foreign_value_at_the_limit_decodes_on_a_64kib_thread", || {
+        // The slot's array is depth 1, so 255 nested arrays under it reach depth 256.
+        let value = on_stack(64 * KIB * KIB, || {
+            let mut value = serde_json::Value::Null;
+            for _ in 0..MAX_PAYLOAD_DEPTH - 1 {
+                value = serde_json::Value::Array(vec![value]);
+            }
+            value
+        });
+        let expected = DynamicSlot { v: value };
+        let bytes = on_stack(64 * KIB * KIB, {
+            let expected = expected.clone();
+            move || rmp_serde::to_vec(&expected).unwrap()
+        });
+        let decoded = decode_on::<DynamicSlot>(64 * KIB, bytes).unwrap();
+        on_stack(64 * KIB * KIB, move || {
+            assert!(decoded == expected);
+            drop(expected);
+        });
+    });
+}
+
+#[test]
+fn a_skipped_slot_at_the_limit_decodes_on_a_64kib_thread() {
+    in_child("a_skipped_slot_at_the_limit_decodes_on_a_64kib_thread", || {
+        // [1, [[...]]]: slot 1 is unknown to One and is skipped; the whole payload is 256 deep.
+        let mut bytes = vec![0x92u8, 0x01];
+        bytes.extend(std::iter::repeat(0x91u8).take(MAX_PAYLOAD_DEPTH - 2));
+        bytes.push(0x90);
+        let decoded = decode_on::<One>(64 * KIB, bytes).unwrap();
+        assert_eq!(decoded, One { a: 1 });
+    });
+}

packages/cultcache-rs/tests/union_parity.rs

@@ -781,12 +781,12 @@ fn the_raw_doors_hold_the_bound_for_arms_maps_and_arrays() {
     }
 }
 
-// A debug build on a 2 MiB thread carries 405 containers of this document; the bound keeps every
-// accepted payload below that, and every refusal is an error, not an abort.
+// The bound counts containers, not stack: decode grows its own stack, so the document at the limit
+// decodes on a 1 MiB thread in both profiles, and every refusal is an error, not an abort.
 #[test]
-fn stack_budget_debug_2mib() {
+fn the_union_chain_at_the_limit_decodes_on_a_1mib_thread() {
     std::thread::Builder::new()
-        .stack_size(2 * 1024 * 1024)
+        .stack_size(1024 * 1024)
         .spawn(|| {
             assert_eq!(
                 through_cache(&vector("depth-at-limit.msgpack")).unwrap(),