sidecar: hand a state input's earlier rows at the times they arrived

cli/tests/exec/test_exec_wasm.py

@@ -1825,32 +1825,41 @@ def test_a_file_an_encoder_in_the_region_writes_says_the_colour_it_was_made_in(
 _EMIT_CHANGES = _built(_BLITZ, "emit_changes")
 
 
-def test_a_reader_of_a_producers_rows_takes_every_frame_of_a_picture_decoded_beside_it(
-    tmp_path: Path,
-) -> None:
-    """blitz's `compose` reads the programme's packets, decoded in the
-    region, and change rows from `emit_changes`, whose clock is the same
-    input's raw picture. Both cross on one pipe, whose ffmpeg writes them
-    ahead of its clock by the offset the sidecar takes off. A producer that
-    writes no rows still lets every frame through."""
+def _composed_changes(
+    emit: str, html: str, out_path: Path, source: Path = _SOURCE
+) -> Compiled:
+    """blitz's `compose` over `source`'s picture, decoded in the region,
+    reading on `changes` the rows `emit_changes` writes from `emit`, clocked
+    by the same input's raw picture."""
     for module in (_COMPOSE, _EMIT_CHANGES):
         if not module.exists():
             pytest.skip(f"module missing: {module}")
-    out_path = tmp_path / "composed.mkv"
     changes = 'STRUCT(at number, "select" text, change text, text text, html text)'
-    compiled = compile_all(
+    return compile_all(
         f"CREATE FUNCTION emit_changes(v video_stream, emit text) RETURNS {changes}[]\n"
         f"  AS '{_EMIT_CHANGES.as_posix()}', 'emit_changes' LANGUAGE wasm;\n"
         f"CREATE FUNCTION compose(v video_stream, changes {changes}[][] DEFAULT NULL,\n"
         "                        html text DEFAULT '') RETURNS video_stream\n"
         f"  AS '{_COMPOSE.as_posix()}', 'compose' LANGUAGE wasm;\n"
         "COPY (SELECT compose(f.video[1],\n"
         "                     changes => emit_changes(ffmpeg.format(f.video[1], 'yuv420p'),\n"
-        "                                             emit => '[]'),\n"
-        "                     html => '<img src=\"ffrwd:0\">')\n"
-        f"      FROM input('{_SOURCE.as_posix()}') f)\n"
+        f"                                             emit => '{emit}'),\n"
+        f"                     html => '{html}')\n"
+        f"      FROM input('{source.as_posix()}') f)\n"
         f"TO '{out_path.as_posix()}' WITH (video_codec 'ffv1')"
     )
+
+
+def test_a_reader_of_a_producers_rows_takes_every_frame_of_a_picture_decoded_beside_it(
+    tmp_path: Path,
+) -> None:
+    """blitz's `compose` reads the programme's packets, decoded in the
+    region, and change rows from `emit_changes`, whose clock is the same
+    input's raw picture. Both cross on one pipe, whose ffmpeg writes them
+    ahead of its clock by the offset the sidecar takes off. A producer that
+    writes no rows still lets every frame through."""
+    out_path = tmp_path / "composed.mkv"
+    compiled = _composed_changes("[]", '<img src="ffrwd:0">', out_path)
     assert compiled.plan is not None
     result = execute_plan(
         compiled.plan, sidecar_argv=wasm.sidecar_argv, overwrite=True,
@@ -1862,3 +1871,57 @@ def test_a_reader_of_a_producers_rows_takes_every_frame_of_a_picture_decoded_bes
         for member in stage.members
     )
     assert _video_stream(out_path)["nb_read_frames"] == _video_stream(_SOURCE)["nb_read_frames"]
+
+
+def test_a_document_folding_a_producers_rows_is_the_same_at_any_worker_count(
+    tmp_path: Path,
+) -> None:
+    """Rows from `emit_changes`, clocked by the input's raw picture, drive
+    blitz's document over the picture the region decodes from MPEG-TS, a
+    clock on another time base. Spread over the default workers, every
+    frame is the one `--jobs 1` makes: an instance is handed the rows of the
+    ticks it did not render at the times they arrived."""
+    source = tmp_path / "in.ts"
+    subprocess.run(
+        [
+            "ffmpeg", "-v", "error", "-y", "-f", "lavfi", "-i", "testsrc2=size=320x240:rate=30",
+            "-t", "4", "-c:v", "libx264", "-pix_fmt", "yuv420p", str(source),
+        ],
+        check=True,
+        timeout=_SUBPROCESS_TIMEOUT,
+    )
+    emit = json.dumps([
+        [0.3 * k, [{"select": "#n", "text": f"row {k}"},
+                   {"select": "#box", "change": f"~c{k % 3}"}]]
+        for k in range(1, 13)
+    ])
+    html = (
+        "<style>body{margin:0;font-family:sans-serif;color:#fff}"
+        "#v{position:absolute;width:100vw;height:100vh}"
+        "#n{position:absolute;left:8px;top:8px;font-size:40px;background:#123}"
+        "#box{position:absolute;right:8px;bottom:8px;width:60px;height:60px}"
+        "#box.c0{background:#a00}#box.c1{background:#0a0}#box.c2{background:#00a}</style>"
+        '<img id="v" src="ffrwd:0"><div id="n"></div><div id="box"></div>'
+    )
+    frames: dict[int | None, list[str]] = {}
+    for jobs in (1, None):
+        out_path = tmp_path / f"composed-{jobs}.mkv"
+        compiled = _composed_changes(emit, html, out_path, source)
+        assert compiled.plan is not None
+        result = execute_plan(
+            compiled.plan,
+            sidecar_argv=functools.partial(wasm.sidecar_argv, jobs=jobs),
+            overwrite=True,
+            timeout=_SUBPROCESS_TIMEOUT,
+        )
+        assert result.exit_code == 0, "\n".join(
+            f"{member.id} exited {member.exit_code}: {member.stderr_tail}"
+            for stage in result.stages
+            for member in stage.members
+        )
+        frames[jobs] = _stream_md5(out_path, 0)
+    assert len(frames[1]) == int(str(_video_stream(source)["nb_read_frames"]))
+    assert len(set(frames[1])) > 1, "the rows change the document"
+    differ = [n for n, (one, many) in enumerate(zip(frames[1], frames[None], strict=True))
+              if one != many]
+    assert not differ, f"frames {differ} differ from --jobs 1's"

sidecar/NODE-CLI.md

@@ -415,7 +415,8 @@ tick that held them.
 - **Lanes.** A pure node runs on as many workers as `-jobs` allows, its
   results put back in tick order before they leave. A pure node with a
   state input opens every instance before its first tick, and each
-  instance is handed the rows of the ticks it did not process. Any other
+  instance is handed the rows of the ticks it did not process, each at
+  the pts it arrived at in its stream's time base. Any other
   node runs one tick at a time.
 - **Finishing.** A node that says it is finished gets its last call on the
   instance that said so, and its outputs end; a rate clock with inputs

sidecar/ffrwd-wasm/src/tick.rs

@@ -291,8 +291,10 @@ impl DataDrive {
 }
 
 /// The rows a state input's instance is owed: those of every tick it did
-/// not process since its previous call, oldest first. A frame-parallel
-/// node's instances each see every row, whichever ticks they were handed.
+/// not process since its previous call, oldest first, each at the pts it
+/// arrived at in its stream's time base, as the instance that processed the
+/// tick was handed it. A frame-parallel node's instances each see every row
+/// at the same time, whichever ticks they were handed.
 #[derive(Default)]
 pub struct EarlierRows {
     /// Per stream id, every tick's rows, by tick number.
@@ -302,15 +304,14 @@ pub struct EarlierRows {
 }
 
 impl EarlierRows {
-    /// The rows stream `id` brought at tick `number`, at `pts`.
-    pub fn record(&mut self, id: u32, number: u64, pts: i64, rows: Vec<String>) {
-        if rows.is_empty() {
-            return;
-        }
-        self.ticks
-            .entry(id)
-            .or_default()
-            .push((number, TimedRows { pts, rows }));
+    /// The rows stream `id` brought at tick `number`.
+    pub fn record(&mut self, id: u32, number: u64, rows: Vec<TimedRows>) {
+        let ticks = self.ticks.entry(id).or_default();
+        ticks.extend(
+            rows.into_iter()
+                .filter(|r| !r.rows.is_empty())
+                .map(|r| (number, r)),
+        );
     }
 
     /// What `instance`, about to process tick `number`, is owed on stream
@@ -1126,18 +1127,21 @@ impl Assembler {
             if !state {
                 continue;
             }
-            let rows: Vec<String> = stream
-                .frames
+            // Each row at its own pts in the stream's time base, in the
+            // order the node reads them: messages, then the rows on frames.
+            let rows: Vec<TimedRows> = stream
+                .messages
                 .iter()
-                .flat_map(|f| f.rows.iter().cloned())
-                .chain(
-                    stream
-                        .messages
-                        .iter()
-                        .map(|m| String::from_utf8_lossy(&m.data).into_owned()),
-                )
+                .map(|m| TimedRows {
+                    pts: m.pts,
+                    rows: vec![String::from_utf8_lossy(&m.data).into_owned()],
+                })
+                .chain(stream.frames.iter().map(|f| TimedRows {
+                    pts: f.pts,
+                    rows: f.rows.clone(),
+                }))
                 .collect();
-            self.earlier.record(stream.id, number, pts, rows);
+            self.earlier.record(stream.id, number, rows);
         }
         Tick {
             pts,
@@ -2124,7 +2128,11 @@ mod tests {
     fn earlier_rows_are_what_an_instance_missed() {
         let mut earlier = EarlierRows::default();
         for n in 0..4u64 {
-            earlier.record(1, n, n as i64 * 10, vec![format!("r{n}")]);
+            let rows = TimedRows {
+                pts: n as i64 * 10,
+                rows: vec![format!("r{n}")],
+            };
+            earlier.record(1, n, vec![rows]);
         }
         assert!(earlier.owed(0, 1, 0).is_empty());
         earlier.processed(0, 0);

sidecar/ffrwd-wasm/tests/node_shapes.rs

@@ -326,6 +326,58 @@ fn every_instance_of_a_state_node_sees_every_row() {
     assert_eq!(ticks[19]["last"], true);
 }
 
+#[test]
+fn every_instance_of_a_state_node_is_handed_each_earlier_row_at_the_time_it_arrived() {
+    // Cues every 0.25 s at 1/48000 from a producer clocked by the sound,
+    // read by a node clocked by pictures at 1/10: most cues fall between
+    // two of the reader's ticks, and none is on the reader's time base.
+    let dir = scratch("state-times");
+    let audio = Stream::audio("f32", 48000, 1).expect("f32 is carried");
+    let mut items: Vec<(usize, Write)> = Vec::new();
+    for k in 0..40i64 {
+        items.push((1, Write::Frame(k * 4800, vec![0u8; 4800 * 4])));
+        items.push((0, Write::Frame(k, vec![0u8; 16])));
+    }
+    write_nut(
+        &dir.join("in.nut"),
+        &[video(2, 2, TENTHS, (10, 1)), audio],
+        &items,
+    );
+    let input = dir.join("in.nut").display().to_string();
+    let out = at_every_jobs(
+        "state-times",
+        &args(&[
+            "-f",
+            "nut",
+            "-i",
+            &input,
+            "-m",
+            &module("shape_window"),
+            "-m",
+            &module("shape_state"),
+            "-filter_complex",
+            "[a=0:a]shape_window=window=12000[cues=c];[v=0:v][words=c]shape_state[seen=s]",
+            "-map",
+            "[s]",
+            "-f",
+            "ndjson",
+            "{dir}/seen.ndjson",
+        ]),
+        &["seen.ndjson"],
+    );
+    let ticks = lines(&out[0]);
+    assert_eq!(ticks.len(), 40);
+    for (n, tick) in ticks.iter().enumerate() {
+        // Cue k starts at k * 0.25 s and lands on tick floor(2.5 k).
+        let cues = (2 * n + 1) / 5;
+        assert_eq!(
+            (tick["seen"].as_u64(), tick["newest"].as_f64()),
+            (Some(cues as u64 + 1), Some(cues as f64 * 0.25)),
+            "tick {n}"
+        );
+    }
+}
+
 #[test]
 fn an_output_sized_by_the_params_is_redrawn_at_that_size() {
     let dir = scratch("canvas");

sidecar/modules/shape-state/src/lib.rs

@@ -1,7 +1,9 @@
 //! A picture clock and words read by interval and kept as state: every tick
 //! says how many words this instance has seen, its own and those of the
-//! ticks it did not process, so a count that matches the serial run's at
-//! every worker count is the host handing each instance its earlier rows.
+//! ticks it did not process, and when the newest of them arrived, read in
+//! the words' time base. A count and a time that match the serial run's at
+//! every worker count are the host handing each instance its earlier rows,
+//! each at the time it arrived.
 
 // Every shape module carries the same few helpers, and not every one uses
 // each of them.
@@ -143,11 +145,17 @@ const SCHEMA: &str = r#"{"type":"object","properties":{"latency":{"type":"number
 
 struct State {
     words: Option<u32>,
+    /// The words' time base.
+    base: (i32, i32),
     seen: u64,
+    /// When the newest word arrived, in the words' time base.
+    newest: Option<i64>,
 }
 
 thread_local! {
-    static STATE: RefCell<State> = const { RefCell::new(State { words: None, seen: 0 }) };
+    static STATE: RefCell<State> = const {
+        RefCell::new(State { words: None, base: (1, 1), seen: 0, newest: None })
+    };
 }
 
 struct Node;
@@ -184,8 +192,14 @@ impl Guest for Node {
     }
 
     fn init(bound: Vec<BoundStream>, _latched: Vec<String>, _params: String) -> Result<(), String> {
-        let words = bound_ids(&bound, "words").first().copied();
-        STATE.with(|s| *s.borrow_mut() = State { words, seen: 0 });
+        let words = bound.iter().find(|b| b.port == "words");
+        let state = State {
+            words: words.map(|b| b.id),
+            base: words.map_or((1, 1), |b| (b.info.time_base.num, b.info.time_base.den)),
+            seen: 0,
+            newest: None,
+        };
+        STATE.with(|s| *s.borrow_mut() = state);
         Ok(())
     }
 
@@ -196,21 +210,27 @@ impl Guest for Node {
     fn process(tick: &Tick) -> Result<Emitted, String> {
         STATE.with(|s| {
             let mut state = s.borrow_mut();
-            let (earlier, now) = match state.words {
-                Some(id) => (
-                    tick.earlier_rows(id)
-                        .iter()
-                        .map(|r| r.rows.len() as u64)
-                        .sum::<u64>(),
-                    tick.messages(id).len() as u64,
-                ),
-                None => (0, 0),
-            };
-            state.seen += earlier + now;
+            let mut now = 0;
+            if let Some(id) = state.words {
+                for rows in tick.earlier_rows(id) {
+                    state.seen += rows.rows.len() as u64;
+                    state.newest = Some(rows.pts);
+                }
+                for m in tick.messages(id) {
+                    now += 1;
+                    state.seen += 1;
+                    state.newest = Some(m.pts);
+                }
+            }
+            let (num, den) = state.base;
+            let newest = state
+                .newest
+                .map(|pts| pts as f64 * f64::from(num) / f64::from(den));
             let item = message(
                 "seen",
                 tick.pts(),
-                json!({"pts": tick.pts(), "seen": state.seen, "now": now, "last": tick.last()}),
+                json!({"pts": tick.pts(), "seen": state.seen, "now": now, "newest": newest,
+                       "last": tick.last()}),
             );
             Ok(emitted(vec![item], false))
         })

sidecar/wit/av.wit

@@ -660,7 +660,8 @@ interface node-tick {
 
     /// For an input whose `rows` is `state`: the rows of every tick this
     /// instance did not process since its previous call, oldest first,
-    /// each beside the pts it arrived at, all earlier than this call's. An
+    /// each beside the pts it arrived at in its stream's time base (a
+    /// message's, or the frame's it rode), all earlier than this call's. An
     /// instance's first call is handed every one before it. A data input's
     /// rows are its messages. Empty on every other input.
     earlier-rows: func(id: u32) -> list<timed-rows>;