feat(data)!: data consumers return each value's timestamp

cpp/moq/MIGRATE.md

@@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first.
 
 ## Unreleased
 
+### JSON consumers return `JsonValue` (#5677)
+
+`JsonSnapshotConsumer::next()` and `JsonStreamConsumer::next()` return `JsonValue` (`value`, `timestamp_us`) instead of a string; `timestamp_us` is empty when untimed.
+A snapshot consumer's `next()` now yields every state in order; call `latest()` for the newest.
+Timestamps are rounded down to a whole microsecond.
+
 ### Ranges filter reads (#5929)
 
 A subscription now reads only the groups its own `ranges` cover.

dart/moq/MIGRATE.md

@@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first.
 
 ## Unreleased
 
+### JSON consumers return `Value` (#5677)
+
+`SnapshotConsumer.next()` and `StreamConsumer.next()` return `Value` (`value`, `timestampUs`) instead of a string; `timestampUs` is null when untimed.
+A snapshot consumer's `next()` now yields every state in order; call `latest()` for the newest.
+Timestamps are rounded down to a whole microsecond.
+
 ### Ranges filter reads (#5929)
 
 A subscription now reads only the groups its own `ranges` cover.

dart/moq/lib/json.dart

@@ -21,7 +21,8 @@ import 'package:moq_ffi/moq_ffi.dart';
 /// Publishes lossy latest-value JSON snapshots on a track it takes over.
 typedef SnapshotProducer = MoqJsonSnapshotProducer;
 
-/// Consumes reconstructed latest-value JSON snapshots from a track it takes over.
+/// Consumes each JSON snapshot state from a track it takes over, or skips to
+/// the newest with `latest()`.
 typedef SnapshotConsumer = MoqJsonSnapshotConsumer;
 
 /// Configures a lossy latest-value JSON track.
@@ -35,3 +36,9 @@ typedef StreamConsumer = MoqJsonStreamConsumer;
 
 /// Configures a lossless JSON stream track.
 typedef StreamConfig = MoqJsonStreamConfig;
+
+/// A JSON string read from a track, with the timestamp of the frame that
+/// carried it in microseconds, or null when untimed. Rounded down to a whole
+/// microsecond, so a tick on a track like 90 kHz may not survive a read and
+/// republish exactly.
+typedef Value = MoqJsonValue;

dart/moq/test/moq_test.dart

@@ -444,7 +444,10 @@ void main() {
     final broadcast = MoqBroadcastProducer();
     final producer = moq_json.SnapshotProducer(
       broadcast: broadcast,
-      track: broadcast.publishTrack(name: 'status', info: null),
+      track: broadcast.publishTrack(
+        name: 'status',
+        info: MoqTrackInfo(timescale: 1000000),
+      ),
       config: moq_json.SnapshotConfig(),
     );
     final track = await broadcast
@@ -456,9 +459,19 @@ void main() {
       config: moq_json.SnapshotConfig(),
     );
 
-    producer.update(value: jsonEncode({'state': 'live'}));
-    final value = await consumer.next().timeout(timeout);
-    expect(jsonDecode(value!), {'state': 'live'});
+    for (var n = 1; n <= 4; n++) {
+      producer.update(value: jsonEncode({'n': n}), timestampUs: n * 1000);
+    }
+
+    // next() yields every state in order; latest() skips the buffered third.
+    moq_json.Value? value = await consumer.next().timeout(timeout);
+    expect(jsonDecode(value!.value), {'n': 1});
+    expect(value.timestampUs, 1000);
+    value = await consumer.next().timeout(timeout);
+    expect(jsonDecode(value!.value), {'n': 2});
+    value = await consumer.latest().timeout(timeout);
+    expect(jsonDecode(value!.value), {'n': 4});
+    expect(value.timestampUs, 4000);
 
     consumer.cancel();
     producer.finish();

dart/moq_ffi/lib/src/moq.dart

@@ -963,6 +963,65 @@ class FfiConverterMoqJsonStreamConfig {
   }
 }
 
+class MoqJsonValue {
+  final String value;
+  final int? timestampUs;
+  MoqJsonValue({required this.value, this.timestampUs});
+}
+
+class FfiConverterMoqJsonValue {
+  static MoqJsonValue lift(RustBuffer buf) {
+    return FfiConverterMoqJsonValue.read(buf.asUint8List()).value;
+  }
+
+  static LiftRetVal<MoqJsonValue> read(Uint8List buf) {
+    int new_offset = buf.offsetInBytes;
+    final value_lifted = FfiConverterString.read(
+      Uint8List.view(buf.buffer, new_offset),
+    );
+    final value = value_lifted.value;
+    new_offset += value_lifted.bytesRead;
+    final timestampUs_lifted = FfiConverterOptionalUInt64.read(
+      Uint8List.view(buf.buffer, new_offset),
+    );
+    final timestampUs = timestampUs_lifted.value;
+    new_offset += timestampUs_lifted.bytesRead;
+    return LiftRetVal(
+      MoqJsonValue(value: value, timestampUs: timestampUs),
+      new_offset - buf.offsetInBytes,
+    );
+  }
+
+  static RustBuffer lower(MoqJsonValue value) {
+    final total_length =
+        FfiConverterString.allocationSize(value.value) +
+        FfiConverterOptionalUInt64.allocationSize(value.timestampUs) +
+        0;
+    final buf = Uint8List(total_length);
+    write(value, buf);
+    return toRustBuffer(buf);
+  }
+
+  static int write(MoqJsonValue value, Uint8List buf) {
+    int new_offset = buf.offsetInBytes;
+    new_offset += FfiConverterString.write(
+      value.value,
+      Uint8List.view(buf.buffer, new_offset),
+    );
+    new_offset += FfiConverterOptionalUInt64.write(
+      value.timestampUs,
+      Uint8List.view(buf.buffer, new_offset),
+    );
+    return new_offset - buf.offsetInBytes;
+  }
+
+  static int allocationSize(MoqJsonValue value) {
+    return FfiConverterString.allocationSize(value.value) +
+        FfiConverterOptionalUInt64.allocationSize(value.timestampUs) +
+        0;
+  }
+}
+
 class MoqAudio {
   final String? label;
   final String? broadcast;
@@ -8741,7 +8800,8 @@ class FfiConverterMoqFlateStreamProducer {
 
 abstract class MoqJsonSnapshotConsumerInterface {
   void cancel();
-  Future<String?> next();
+  Future<MoqJsonValue?> latest();
+  Future<MoqJsonValue?> next();
 }
 
 final _MoqJsonSnapshotConsumerFinalizer = Finalizer<Pointer<Void>>((ptr) {
@@ -8801,15 +8861,28 @@ class MoqJsonSnapshotConsumer implements MoqJsonSnapshotConsumerInterface {
     }, null);
   }
 
-  Future<String?> next() {
+  Future<MoqJsonValue?> latest() {
+    return uniffiRustCallAsync(
+      () => uniffi_moq_ffi_fn_method_moqjsonsnapshotconsumer_latest(
+        uniffiClonePointer(),
+      ),
+      ffi_moq_ffi_rust_future_poll_rust_buffer,
+      ffi_moq_ffi_rust_future_complete_rust_buffer,
+      ffi_moq_ffi_rust_future_free_rust_buffer,
+      FfiConverterOptionalMoqJsonValue.lift,
+      moqExceptionErrorHandler,
+    );
+  }
+
+  Future<MoqJsonValue?> next() {
     return uniffiRustCallAsync(
       () => uniffi_moq_ffi_fn_method_moqjsonsnapshotconsumer_next(
         uniffiClonePointer(),
       ),
       ffi_moq_ffi_rust_future_poll_rust_buffer,
       ffi_moq_ffi_rust_future_complete_rust_buffer,
       ffi_moq_ffi_rust_future_free_rust_buffer,
-      FfiConverterOptionalString.lift,
+      FfiConverterOptionalMoqJsonValue.lift,
       moqExceptionErrorHandler,
     );
   }
@@ -8957,7 +9030,7 @@ class FfiConverterMoqJsonSnapshotProducer {
 
 abstract class MoqJsonStreamConsumerInterface {
   void cancel();
-  Future<String?> next();
+  Future<MoqJsonValue?> next();
 }
 
 final _MoqJsonStreamConsumerFinalizer = Finalizer<Pointer<Void>>((ptr) {
@@ -9017,15 +9090,15 @@ class MoqJsonStreamConsumer implements MoqJsonStreamConsumerInterface {
     }, null);
   }
 
-  Future<String?> next() {
+  Future<MoqJsonValue?> next() {
     return uniffiRustCallAsync(
       () => uniffi_moq_ffi_fn_method_moqjsonstreamconsumer_next(
         uniffiClonePointer(),
       ),
       ffi_moq_ffi_rust_future_poll_rust_buffer,
       ffi_moq_ffi_rust_future_complete_rust_buffer,
       ffi_moq_ffi_rust_future_free_rust_buffer,
-      FfiConverterOptionalString.lift,
+      FfiConverterOptionalMoqJsonValue.lift,
       moqExceptionErrorHandler,
     );
   }
@@ -13599,6 +13672,53 @@ class FfiConverterOptionalMoqGroupConsumer {
   }
 }
 
+class FfiConverterOptionalMoqJsonValue {
+  static MoqJsonValue? lift(RustBuffer buf) {
+    return FfiConverterOptionalMoqJsonValue.read(buf.asUint8List()).value;
+  }
+
+  static LiftRetVal<MoqJsonValue?> read(Uint8List buf) {
+    if (ByteData.view(buf.buffer, buf.offsetInBytes).getInt8(0) == 0) {
+      return LiftRetVal(null, 1);
+    }
+    final result = FfiConverterMoqJsonValue.read(
+      Uint8List.view(buf.buffer, buf.offsetInBytes + 1),
+    );
+    return LiftRetVal<MoqJsonValue?>(result.value, result.bytesRead + 1);
+  }
+
+  static int allocationSize([MoqJsonValue? value]) {
+    if (value == null) {
+      return 1;
+    }
+    return FfiConverterMoqJsonValue.allocationSize(value) + 1;
+  }
+
+  static RustBuffer lower(MoqJsonValue? value) {
+    if (value == null) {
+      return toRustBuffer(Uint8List.fromList([0]));
+    }
+    final buf = Uint8List(
+      FfiConverterOptionalMoqJsonValue.allocationSize(value),
+    );
+    FfiConverterOptionalMoqJsonValue.write(value, buf);
+    return toRustBuffer(buf);
+  }
+
+  static int write(MoqJsonValue? value, Uint8List buf) {
+    if (value == null) {
+      buf[0] = 0;
+      return 1;
+    }
+    buf[0] = 1;
+    return FfiConverterMoqJsonValue.write(
+          value,
+          Uint8List.view(buf.buffer, buf.offsetInBytes + 1),
+        ) +
+        1;
+  }
+}
+
 class FfiConverterOptionalMoqMediaFrame {
   static MoqMediaFrame? lift(RustBuffer buf) {
     return FfiConverterOptionalMoqMediaFrame.read(buf.asUint8List()).value;
@@ -15550,6 +15670,11 @@ external void uniffi_moq_ffi_fn_method_moqjsonsnapshotconsumer_cancel(
   Pointer<RustCallStatus> uniffiStatus,
 );
 
+@Native<Pointer<Void> Function(Pointer<Void>)>(assetId: _uniffiAssetId)
+external Pointer<Void> uniffi_moq_ffi_fn_method_moqjsonsnapshotconsumer_latest(
+  Pointer<Void> ptr,
+);
+
 @Native<Pointer<Void> Function(Pointer<Void>)>(assetId: _uniffiAssetId)
 external Pointer<Void> uniffi_moq_ffi_fn_method_moqjsonsnapshotconsumer_next(
   Pointer<Void> ptr,
@@ -17821,6 +17946,9 @@ external int uniffi_moq_ffi_checksum_method_moqflatestreamproducer_finish();
 @Native<Uint16 Function()>(assetId: _uniffiAssetId)
 external int uniffi_moq_ffi_checksum_method_moqjsonsnapshotconsumer_cancel();
 
+@Native<Uint16 Function()>(assetId: _uniffiAssetId)
+external int uniffi_moq_ffi_checksum_method_moqjsonsnapshotconsumer_latest();
+
 @Native<Uint16 Function()>(assetId: _uniffiAssetId)
 external int uniffi_moq_ffi_checksum_method_moqjsonsnapshotconsumer_next();
 
@@ -18469,10 +18597,14 @@ void _checkApiChecksums() {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
   if (uniffi_moq_ffi_checksum_method_moqjsonsnapshotconsumer_cancel() !=
-      45114) {
+      52062) {
+    throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
+  }
+  if (uniffi_moq_ffi_checksum_method_moqjsonsnapshotconsumer_latest() !=
+      60932) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
-  if (uniffi_moq_ffi_checksum_method_moqjsonsnapshotconsumer_next() != 64727) {
+  if (uniffi_moq_ffi_checksum_method_moqjsonsnapshotconsumer_next() != 65302) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
   if (uniffi_moq_ffi_checksum_method_moqjsonsnapshotproducer_demand() !=
@@ -18490,7 +18622,7 @@ void _checkApiChecksums() {
   if (uniffi_moq_ffi_checksum_method_moqjsonstreamconsumer_cancel() != 29308) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
-  if (uniffi_moq_ffi_checksum_method_moqjsonstreamconsumer_next() != 7523) {
+  if (uniffi_moq_ffi_checksum_method_moqjsonstreamconsumer_next() != 21257) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
   if (uniffi_moq_ffi_checksum_method_moqjsonstreamproducer_append() != 36078) {

doc/lib/c/index.md

@@ -64,6 +64,7 @@ inside. The header is the reference: each function carries a doc comment.
 - **Zeroed config is the default.** A zeroed `moq_client_config` means the default for every field, so new fields don't break callers. A field whose default isn't zero has a `has_*` flag: `backoff_timeout_us = 0` means "retry forever" only with `has_backoff_timeout = true`.
 - **Audio needs cuts.** Video groups at its keyframes, but audio forms a group only where you call `moq_publish_media_cut`: after every frame, or at a segment cadence to align with video.
 - **Live encoder timing.** After writing a frame you encoded yourself, call `moq_publish_media_flush` with the same timestamp so the catalog advertises your jitter. Skip it for file and network imports. On a seek or pause, call `moq_publish_media_discontinuity`, then keep timestamps moving forward and resume video on a keyframe.
+- **JSON values.** `moq_consume_json_value` fills `moq_json_value` with the document and its frame's `timestamp_us`, rounded down to a whole microsecond; `timestamp_present` is false on an untimed track. A snapshot consumer's callback fires for each state in order, but a group superseded before it is read is skipped.
 - **Decoded frames hold decoder buffers.** Free each frame id with `moq_decode_video_frame_free` promptly, or the decoder stalls.
 - **Stats.** `moq_session_stats()` fills `moq_connection_stats` with `rtt_us`, `estimated_send_rate_bps`, and the byte and packet counters (`bytes_sent`, `bytes_received`, `bytes_lost`, `packets_sent`, `packets_received`, `packets_lost`). Each has a `<field>_valid` flag, false when the transport does not report it, which is not the same as zero. `moq_session_snapshot()` adds the negotiated `protocol`, and while a session reconnects `moq_session_failure()` returns why, with the reason in `moq_error()`. Hold the handle returned by `moq_session_recv_bandwidth()` to request receive estimates. `moq_bandwidth_consumer_peek()` reads the estimate and `moq_bandwidth_consumer_changed()` waits for a change. `moq_bandwidth_consumer_close()` releases demand; the last consumer stops PROBE. Stats snapshots do not request it.
 - **Demand.** A watcher from `moq_publish_track_demand()` (or the media, video, and audio variants) calls back `MOQ_DEMAND_USED` or `MOQ_DEMAND_UNUSED`, again whenever the wanted group moves. `moq_publish_demand_wanted()` reads that group: start a new one once it passes your largest. Set `refresh` in `moq_track_info` to have a joining subscriber ask for a new group once the newest has grown past `refresh` times its first frame.

doc/lib/dart/index.md

@@ -19,6 +19,8 @@ and a `Named` or `Requested` target. `CatalogProducer(broadcast: ...)` owns cata
 and sections; it holds the broadcast weakly and fails after the broadcast closes.
 `package:moq/flate.dart` exposes opaque `SnapshotProducer` / `StreamProducer` with an optional `timestampUs`.
 A missing data timestamp stays untimed; the library never substitutes zero or now.
+`package:moq/json.dart` consumers return `Value` (`value`, `timestampUs`) from `next()`; a snapshot consumer
+yields each state in order, and `latest()` skips to the newest. Read timestamps are rounded down to a whole microsecond.
 Raw tracks with a `null` timescale are untimed on both publish and receive.
 Whole-frame media writes require caller time; container imports keep their own timeline handling.
 Timed raw/data tracks require caller-supplied timestamps. Use `CatalogProducer.now()` to stamp a locally

doc/lib/go/index.md

@@ -21,6 +21,8 @@ A missing data timestamp stays untimed; the library never substitutes zero or no
 Raw tracks with a `nil` timescale are untimed on both publish and receive.
 `moq.dev/moq/flate` publishes opaque snapshot and stream payloads; its `Update` / `Append`
 and the JSON producers take a final `*time.Duration` timestamp argument (`nil` for untimed).
+JSON consumers return a `*Timed` holding the value and that timestamp; `SnapshotConsumer.Latest`
+skips to the newest state where `Next` yields each one. Read timestamps are rounded down to a whole microsecond.
 Whole-frame media writes require caller time; container imports keep their own timeline handling.
 Timed raw/data tracks require caller-supplied timestamps. Use `CatalogProducer.Now()` to stamp a locally
 created event on the broadcast clock; reading it fixes that clock for later media producers.

doc/lib/index.md

@@ -40,7 +40,7 @@ how it looks in that language:
 - **Discover** broadcasts by prefix, wait for a specific one, or request one by path. Advertise an exact path (a created broadcast is invisible until announced), or claim a prefix and serve requests beneath it on demand.
 - **Publish and subscribe to media** with the hang catalog filled in from the bitstream, or hand over raw pixels and PCM and let the binding encode and decode (VideoToolbox, Media Foundation, NVENC, openh264, Opus). The built-in video encoder can follow the connection's send estimate. The audio decoder reports the playout delay to buffer ahead of the playhead, fixed or estimated from arrivals, so read each frame as soon as the last is handled rather than pacing reads to the speaker.
 - **Raw tracks** of arbitrary bytes, with per-subscriber priority and max delay, and best-effort datagrams.
-- **JSON tracks** as a latest value with merge-patch deltas, or as an append log.
+- **JSON tracks** as a latest value with merge-patch deltas, or as an append log. Each value reads back with its frame timestamp, and a snapshot reader either replays every state or skips to the newest.
 - **Fetch** a single cached group by sequence.
 - **Catalog extensions**: write your own section next to `video` and `audio`, and read others' back.
 - **Routes**: see which relays a broadcast came through, and advertise a cost as a standby publisher.

doc/lib/kt/index.md

@@ -18,6 +18,8 @@ x86\_64) and desktop JVM (Linux x86\_64/aarch64, macOS arm64, Windows x64).
 `CatalogProducer(broadcast)` holds its broadcast weakly and fails with `Closed` after it closes.
 `dev.moq.flate.SnapshotProducer` / `StreamProducer` publish opaque bytes with an optional `timestampUs`.
 A missing data timestamp stays untimed; the library never substitutes zero or now.
+`dev.moq.json` consumers' `values()` emits `Value` (`value`, `timestamp`) and `valuesAs<T>()` emits `Timed<T>`;
+a snapshot consumer yields each state in order, and `latest()` skips to the newest. Read timestamps are rounded down to a whole microsecond.
 Raw tracks with a `null` timescale are untimed on both publish and receive.
 Whole-frame media writes require caller time; container imports keep their own timeline handling.
 Timed raw/data tracks require caller-supplied timestamps. Use `CatalogProducer.now()` to stamp a locally

doc/lib/py/index.md

@@ -23,6 +23,8 @@ broadcast and does not keep it open. `ContainerProducer` takes chunks split anyw
 so a raw byte stream needs no other importer.
 `MediaFrame.timestamp`, media writes, and `flush` use `datetime.timedelta`.
 A missing data timestamp stays untimed; the library never substitutes zero or now.
+JSON consumers yield `moq.json.Timed(value, timestamp)`; a snapshot consumer yields each state in order,
+and `await consumer.latest()` skips to the newest. Read timestamps are rounded down to a whole microsecond.
 Raw tracks with a `None` timescale are untimed on both publish and receive.
 `moq.flate.SnapshotProducer` / `StreamProducer` publish opaque payloads; their `update` / `append`
 and the JSON producers accept `timestamp=timedelta(...)` (`None` for untimed).

doc/lib/rs/moq-flate.md

@@ -32,6 +32,13 @@ an untimed track has no timescale, and a timed track requires a timestamp on
 every write. A publisher that wants to stamp now samples its own clock.
 Writes return the encoded frame size.
 
+Consumers return `Timed<T>`: read `.value` for the decoded value and `.at` for
+its optional frame timestamp, with the track's exact scale preserved. Snapshot
+`next()` yields states in order inside the subscription's `max_delay` budget;
+set that budget when following a playhead behind live. Use `latest()` to skip
+to the newest buffered state. Skipped states are gone, and a later `next()`
+continues after them.
+
 Upgrading from an older release: see moq-flate's
 [MIGRATE.md](https://github.com/moq-dev/moq/blob/main/rs/moq-flate/MIGRATE.md).