diff --git a/cpp/moq/MIGRATE.md b/cpp/moq/MIGRATE.md index b95a4e97690..89169b3ff78 100644 --- a/cpp/moq/MIGRATE.md +++ b/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. diff --git a/dart/moq/MIGRATE.md b/dart/moq/MIGRATE.md index 3e45fc90d91..8154fd68df2 100644 --- a/dart/moq/MIGRATE.md +++ b/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. diff --git a/dart/moq/lib/json.dart b/dart/moq/lib/json.dart index 539fb5f6340..7c477812460 100644 --- a/dart/moq/lib/json.dart +++ b/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; diff --git a/dart/moq/test/moq_test.dart b/dart/moq/test/moq_test.dart index 0cf8ee36cb3..3bcdcfa02fb 100644 --- a/dart/moq/test/moq_test.dart +++ b/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(); diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 2947697cf54..8c807bd0d31 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/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 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 next(); + Future latest(); + Future next(); } final _MoqJsonSnapshotConsumerFinalizer = Finalizer>((ptr) { @@ -8801,7 +8861,20 @@ class MoqJsonSnapshotConsumer implements MoqJsonSnapshotConsumerInterface { }, null); } - Future next() { + Future 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 next() { return uniffiRustCallAsync( () => uniffi_moq_ffi_fn_method_moqjsonsnapshotconsumer_next( uniffiClonePointer(), @@ -8809,7 +8882,7 @@ class MoqJsonSnapshotConsumer implements MoqJsonSnapshotConsumerInterface { 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 next(); + Future next(); } final _MoqJsonStreamConsumerFinalizer = Finalizer>((ptr) { @@ -9017,7 +9090,7 @@ class MoqJsonStreamConsumer implements MoqJsonStreamConsumerInterface { }, null); } - Future next() { + Future next() { return uniffiRustCallAsync( () => uniffi_moq_ffi_fn_method_moqjsonstreamconsumer_next( uniffiClonePointer(), @@ -9025,7 +9098,7 @@ class MoqJsonStreamConsumer implements MoqJsonStreamConsumerInterface { 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 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(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 uniffiStatus, ); +@Native Function(Pointer)>(assetId: _uniffiAssetId) +external Pointer uniffi_moq_ffi_fn_method_moqjsonsnapshotconsumer_latest( + Pointer ptr, +); + @Native Function(Pointer)>(assetId: _uniffiAssetId) external Pointer uniffi_moq_ffi_fn_method_moqjsonsnapshotconsumer_next( Pointer ptr, @@ -17821,6 +17946,9 @@ external int uniffi_moq_ffi_checksum_method_moqflatestreamproducer_finish(); @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqjsonsnapshotconsumer_cancel(); +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqjsonsnapshotconsumer_latest(); + @Native(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) { diff --git a/doc/lib/c/index.md b/doc/lib/c/index.md index b04a67ba4a7..07eb92bc5cd 100644 --- a/doc/lib/c/index.md +++ b/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 `_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. diff --git a/doc/lib/dart/index.md b/doc/lib/dart/index.md index 91326288472..73bdd8968c3 100644 --- a/doc/lib/dart/index.md +++ b/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 diff --git a/doc/lib/go/index.md b/doc/lib/go/index.md index e960e656e10..debf9f3bf99 100644 --- a/doc/lib/go/index.md +++ b/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. diff --git a/doc/lib/index.md b/doc/lib/index.md index d2a08e444a3..291995ff87d 100644 --- a/doc/lib/index.md +++ b/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. diff --git a/doc/lib/kt/index.md b/doc/lib/kt/index.md index 2f147a60fae..4c8b59e9c2c 100644 --- a/doc/lib/kt/index.md +++ b/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()` emits `Timed`; +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 diff --git a/doc/lib/py/index.md b/doc/lib/py/index.md index 5288bf5e289..fd702699d9d 100644 --- a/doc/lib/py/index.md +++ b/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). diff --git a/doc/lib/rs/moq-flate.md b/doc/lib/rs/moq-flate.md index 0a3ab119654..b835c8520b9 100644 --- a/doc/lib/rs/moq-flate.md +++ b/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`: 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). diff --git a/doc/lib/rs/moq-json.md b/doc/lib/rs/moq-json.md index 2338572eae0..0b944f17eb8 100644 --- a/doc/lib/rs/moq-json.md +++ b/doc/lib/rs/moq-json.md @@ -23,6 +23,13 @@ The timestamp must match the track: an untimed track has no timescale, and a timed track requires a timestamp on every write. Editing guards take the same optional capture time when opened, including when they publish on drop. +Consumers return `Timed`: 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-json's [MIGRATE.md](https://github.com/moq-dev/moq/blob/main/rs/moq-json/MIGRATE.md). diff --git a/doc/lib/swift/index.md b/doc/lib/swift/index.md index 37832526ff0..e55f0029f41 100644 --- a/doc/lib/swift/index.md +++ b/doc/lib/swift/index.md @@ -19,6 +19,8 @@ updates catalog properties and sections without keeping the broadcast open. Its writes fail after closing or releasing the broadcast. `Flate.SnapshotProducer` / `StreamProducer` publish opaque bytes with an optional `timestampUs`. A missing data timestamp stays untimed; the library never substitutes zero or now. +`Json` consumers yield `Json.Timed` (`value`, `timestampUs`); 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 `nil` 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 `Media.CatalogProducer.now()` to stamp a locally diff --git a/go/wrapper/MIGRATE.md b/go/wrapper/MIGRATE.md index 5dbc9ca53ea..bd15cc49975 100644 --- a/go/wrapper/MIGRATE.md +++ b/go/wrapper/MIGRATE.md @@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first. ## Unreleased +### JSON consumers return `*json.Timed` (#5677) + +`SnapshotConsumer.Next`, `StreamConsumer.Next`, and `All` return `*Timed{Value, Timestamp}` instead of the raw message; `Timestamp` is a `*time.Duration`, nil when untimed. +A snapshot consumer's `Next` now yields every state in order; call `Latest(ctx)` 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. diff --git a/go/wrapper/json/json.go b/go/wrapper/json/json.go index 57f3e9c3ba2..9a1293db8fd 100644 --- a/go/wrapper/json/json.go +++ b/go/wrapper/json/json.go @@ -152,16 +152,29 @@ func (p *StreamProducer) Finish() error { return p.inner.Finish() } -// rawMessage converts the result of a Next call, keeping (nil, nil) for the end. -func rawMessage(value *string, err error) (*stdjson.RawMessage, error) { - if err != nil || value == nil { - return nil, err +// Timed is a JSON value and the timestamp of the frame that carried it. +type Timed struct { + // Value is the JSON document. + Value stdjson.RawMessage + // Timestamp is on the broadcast clock, or nil on an untimed track. It is rounded down + // to a whole microsecond, so a tick on a track like 90 kHz may not survive a read and + // republish exactly. + Timestamp *time.Duration +} + +func timed(value *ffi.MoqJsonValue) *Timed { + if value == nil { + return nil + } + out := &Timed{Value: stdjson.RawMessage(value.Value)} + if value.TimestampUs != nil { + at := time.Duration(*value.TimestampUs) * time.Microsecond + out.Timestamp = &at } - decoded := stdjson.RawMessage(*value) - return &decoded, nil + return out } -// SnapshotConsumer receives the latest reconstructed JSON value. +// SnapshotConsumer receives each reconstructed JSON state. type SnapshotConsumer struct { inner *ffi.MoqJsonSnapshotConsumer } @@ -180,13 +193,23 @@ func NewSnapshotConsumer(track *moq.TrackConsumer, options ConsumerOptions) (*Sn return &SnapshotConsumer{inner: inner}, nil } -// Next returns the next JSON value, or nil when the track ends. -func (c *SnapshotConsumer) Next(ctx context.Context) (*stdjson.RawMessage, error) { - return rawMessage(bridge.Call(ctx, c.inner.Cancel, c.inner.Next)) +// Next returns the next state in order, or nil when the track ends. Groups older +// than the subscription's MaxDelay are skipped, so a playhead reader sets that +// budget to its distance behind live. +func (c *SnapshotConsumer) Next(ctx context.Context) (*Timed, error) { + value, err := bridge.Call(ctx, c.inner.Cancel, c.inner.Next) + return timed(value), err +} + +// Latest skips to the newest state, or waits for one; nil when the track ends. +// Skipped states are gone, and Next continues after this one. +func (c *SnapshotConsumer) Latest(ctx context.Context) (*Timed, error) { + value, err := bridge.Call(ctx, c.inner.Cancel, c.inner.Latest) + return timed(value), err } -// All ranges over reconstructed JSON values until the track ends. -func (c *SnapshotConsumer) All(ctx context.Context) iter.Seq2[*stdjson.RawMessage, error] { +// All ranges over each JSON state until the track ends. +func (c *SnapshotConsumer) All(ctx context.Context) iter.Seq2[*Timed, error] { return bridge.Seq(ctx, c.Next) } @@ -214,12 +237,13 @@ func NewStreamConsumer(track *moq.TrackConsumer, options ConsumerOptions) (*Stre } // Next returns the next JSON record, or nil when the track ends. -func (c *StreamConsumer) Next(ctx context.Context) (*stdjson.RawMessage, error) { - return rawMessage(bridge.Call(ctx, c.inner.Cancel, c.inner.Next)) +func (c *StreamConsumer) Next(ctx context.Context) (*Timed, error) { + value, err := bridge.Call(ctx, c.inner.Cancel, c.inner.Next) + return timed(value), err } // All ranges over every JSON record until the track ends. -func (c *StreamConsumer) All(ctx context.Context) iter.Seq2[*stdjson.RawMessage, error] { +func (c *StreamConsumer) All(ctx context.Context) iter.Seq2[*Timed, error] { return bridge.Seq(ctx, c.Next) } diff --git a/go/wrapper/json/json_test.go b/go/wrapper/json/json_test.go index 978901bfe30..16f57e22fb9 100644 --- a/go/wrapper/json/json_test.go +++ b/go/wrapper/json/json_test.go @@ -54,29 +54,50 @@ func TestTracks(t *testing.T) { defer cancel() broadcast, consumer := newBroadcast(t) - snapshot, err := moqjson.NewSnapshotProducer(broadcast, publishTrack(t, broadcast, "status"), moqjson.SnapshotOptions{Compression: true}) + scale := uint64(1_000_000) + track, err := broadcast.PublishTrack("status", &moq.TrackInfo{Priority: 127, Timescale: &scale}) if err != nil { t.Fatal(err) } - snapshotConsumer, err := moqjson.NewSnapshotConsumer(subscribeTrack(ctx, t, consumer, "status"), moqjson.ConsumerOptions{Compression: true}) + snapshot, err := moqjson.NewSnapshotProducer(broadcast, track, moqjson.SnapshotOptions{Compression: true}) if err != nil { t.Fatal(err) } - defer snapshotConsumer.Cancel() - if err := snapshot.Update(map[string]any{"viewers": 42}, nil); err != nil { - t.Fatal(err) - } - value, err := snapshotConsumer.Next(ctx) + snapshotConsumer, err := moqjson.NewSnapshotConsumer(subscribeTrack(ctx, t, consumer, "status"), moqjson.ConsumerOptions{Compression: true}) if err != nil { t.Fatal(err) } - var decoded map[string]any - if err := json.Unmarshal(*value, &decoded); err != nil { - t.Fatal(err) + defer snapshotConsumer.Cancel() + for viewers := 1; viewers <= 4; viewers++ { + at := time.Duration(viewers) * time.Millisecond + if err := snapshot.Update(map[string]any{"viewers": viewers}, &at); err != nil { + t.Fatal(err) + } } - if decoded["viewers"] != float64(42) { - t.Fatalf("snapshot = %s", *value) + + // Next yields every state in order; Latest skips the buffered third. + expect := func(value *moqjson.Timed, err error, viewers int) { + t.Helper() + if err != nil { + t.Fatal(err) + } + var decoded map[string]any + if err := json.Unmarshal(value.Value, &decoded); err != nil { + t.Fatal(err) + } + if decoded["viewers"] != float64(viewers) { + t.Fatalf("snapshot = %s, want viewers %d", value.Value, viewers) + } + if want := time.Duration(viewers) * time.Millisecond; value.Timestamp == nil || *value.Timestamp != want { + t.Fatalf("timestamp = %v, want %v", value.Timestamp, want) + } } + value, err := snapshotConsumer.Next(ctx) + expect(value, err, 1) + value, err = snapshotConsumer.Next(ctx) + expect(value, err, 2) + value, err = snapshotConsumer.Latest(ctx) + expect(value, err, 4) stream, err := moqjson.NewStreamProducer(broadcast, publishTrack(t, broadcast, "events"), moqjson.StreamOptions{Compression: true}) if err != nil { @@ -94,8 +115,8 @@ func TestTracks(t *testing.T) { if err != nil { t.Fatal(err) } - if string(*record) != `{"n":1}` { - t.Fatalf("record = %s", *record) + if string(record.Value) != `{"n":1}` || record.Timestamp != nil { + t.Fatalf("record = %s at %v, want untimed", record.Value, record.Timestamp) } if err := snapshot.Finish(); err != nil { diff --git a/kt/moq/MIGRATE.md b/kt/moq/MIGRATE.md index 880c0c67e5b..51b55968ef0 100644 --- a/kt/moq/MIGRATE.md +++ b/kt/moq/MIGRATE.md @@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first. ## Unreleased +### JSON consumers emit timestamps (#5677) + +`values()` emits `json.Value` (`value`, plus `timestamp` as a `Duration?`) instead of a string, and `valuesAs()` emits `Timed(value, timestamp)`. +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. diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/json/Tracks.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/json/Tracks.kt index 52585b0c7eb..cff82a8e7f1 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/json/Tracks.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/json/Tracks.kt @@ -13,6 +13,8 @@ import dev.moq.pull import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.map import kotlinx.serialization.serializer +import kotlin.time.Duration +import kotlin.time.Duration.Companion.microseconds /** Publishes lossy latest-value JSON snapshots on a track it takes over. */ typealias SnapshotProducer = uniffi.moq.MoqJsonSnapshotProducer @@ -26,6 +28,22 @@ typealias StreamConsumer = uniffi.moq.MoqJsonStreamConsumer typealias SnapshotConfig = uniffi.moq.MoqJsonSnapshotConfig /** Configures a lossless JSON stream track. */ typealias StreamConfig = uniffi.moq.MoqJsonStreamConfig +/** A JSON string and the timestamp of the frame that carried it. */ +typealias Value = uniffi.moq.MoqJsonValue + +/** + * The frame's timestamp on the broadcast clock, 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. + */ +val Value.timestamp: Duration? + get() = timestampUs?.toLong()?.microseconds + +/** A decoded JSON value and its frame timestamp on the broadcast clock, or null when untimed. */ +data class Timed(val value: T, val timestamp: Duration?) + +/** Decode [Value.value] as [T], keeping the timestamp. */ +inline fun Value.decode(): Timed = + Timed(MoqJson.decodeFromString(serializer(), value), timestamp) /** * Publish [value] as a JSON snapshot, superseding the last. A no-op when unchanged. @@ -44,27 +62,28 @@ inline fun StreamProducer.append(value: T, timestampUs: ULong? = nul } /** - * Stream of JSON values (as strings) from a snapshot track, yielding the latest reconstructed - * value. A consumer that has fallen behind collapses the backlog to the latest. + * Stream of each JSON state from a snapshot track, in order. + * + * Groups older than the subscription's `maxDelayUs` are skipped, so a playhead reader sets that + * budget to its distance behind live. Call [SnapshotConsumer.latest] to skip to the newest state. */ -fun SnapshotConsumer.values(): Flow = pull({ cancel() }) { next() } +fun SnapshotConsumer.values(): Flow = pull({ cancel() }) { next() } -/** Stream of JSON records (as strings) from a stream track, in order. */ -fun StreamConsumer.values(): Flow = pull({ cancel() }) { next() } +/** Stream of JSON records from a stream track, in order. */ +fun StreamConsumer.values(): Flow = pull({ cancel() }) { next() } /** - * Stream of decoded snapshot values, yielding the latest reconstructed value. + * Stream of decoded snapshot states, in order. * - * A consumer that has fallen behind collapses the backlog to the latest. Use - * [SnapshotConsumer.values] for the undecoded JSON strings. + * Use [SnapshotConsumer.values] for the undecoded JSON strings. */ -inline fun SnapshotConsumer.valuesAs(): Flow = - values().map { MoqJson.decodeFromString(serializer(), it) } +inline fun SnapshotConsumer.valuesAs(): Flow> = + values().map { it.decode() } /** * Stream of decoded stream records, in order. * * Use [StreamConsumer.values] for the undecoded JSON strings. */ -inline fun StreamConsumer.valuesAs(): Flow = - values().map { MoqJson.decodeFromString(serializer(), it) } +inline fun StreamConsumer.valuesAs(): Flow> = + values().map { it.decode() } diff --git a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt index 47d16e5a667..691a8799972 100644 --- a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt +++ b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt @@ -16,6 +16,8 @@ import dev.moq.json.SnapshotConfig import dev.moq.json.SnapshotConsumer import dev.moq.json.SnapshotProducer import dev.moq.json.StreamConfig +import dev.moq.json.Timed +import dev.moq.json.decode import dev.moq.json.update import dev.moq.json.valuesAs import kotlinx.serialization.Serializable @@ -409,12 +411,16 @@ class SmokeTest { @Test fun `typed json snapshot round-trips a serializable value`() = runTest { BroadcastProducer().use { broadcast -> - val config = SnapshotConfig(deltaRatio = 0u, compression = false) - val producer = SnapshotProducer(broadcast, broadcast.publishTrack("status", null), config) - producer.update(Status(state = "live")) - + val config = SnapshotConfig(deltaRatio = 8u, compression = false) + val track = broadcast.publishTrack("status", TrackInfo(timescale = 1_000_000uL)) + val producer = SnapshotProducer(broadcast, track, config) val consumer = SnapshotConsumer(broadcast.consume().subscribeTrack("status", null), config) - assertEquals(Status(state = "live"), consumer.valuesAs().first()) + for (n in 1..4) producer.update(Status(state = "live$n"), timestampUs = n.toULong() * 1_000uL) + + // `next` yields every state in order; `latest` skips the buffered third. + assertEquals(Timed(Status(state = "live1"), 1.milliseconds), consumer.next()?.decode()) + assertEquals(Timed(Status(state = "live2"), 2.milliseconds), consumer.next()?.decode()) + assertEquals(Timed(Status(state = "live4"), 4.milliseconds), consumer.latest()?.decode()) } } @@ -445,7 +451,7 @@ class SmokeTest { val consumer = SnapshotConsumer(broadcast.consume().subscribeTrack("status", null), config) while (demand.wantedChanged() == null) {} // `first` cancels the collection, which must release the subscription without an explicit cancel. - assertEquals(Status(state = "live"), consumer.valuesAs().first()) + assertEquals(Status(state = "live"), consumer.valuesAs().first().value) while (demand.wantedChanged() != null) {} } } @@ -463,7 +469,7 @@ class SmokeTest { producer.update("""{"state":"raw"}""") val consumer = SnapshotConsumer(broadcast.consume().subscribeTrack("status", null), config) - assertEquals(Status(state = "raw"), consumer.valuesAs().first()) + assertEquals(Timed(Status(state = "raw"), null), consumer.valuesAs().first()) } } diff --git a/py/moq-rs/MIGRATE.md b/py/moq-rs/MIGRATE.md index 3a49615cdf0..8dac627d188 100644 --- a/py/moq-rs/MIGRATE.md +++ b/py/moq-rs/MIGRATE.md @@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first. ## Unreleased +### JSON consumers yield `json.Timed` (#5677) + +`json.SnapshotConsumer` and `json.StreamConsumer` yield `Timed(value, timestamp)` instead of the bare value; `timestamp` is a `timedelta`, or `None` when untimed. +Iterating a snapshot consumer now yields every state in order; `await consumer.latest()` returns 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. diff --git a/py/moq-rs/docs/index.md b/py/moq-rs/docs/index.md index 5e32dc7a299..94050e1c24c 100644 --- a/py/moq-rs/docs/index.md +++ b/py/moq-rs/docs/index.md @@ -146,6 +146,7 @@ broadcast, and producers advertise it in the catalog. StreamProducer SnapshotConsumer StreamConsumer + Timed ``` ## Opaque data diff --git a/py/moq-rs/moq/json.py b/py/moq-rs/moq/json.py index eafd01cfa35..434d4bc5ea5 100644 --- a/py/moq-rs/moq/json.py +++ b/py/moq-rs/moq/json.py @@ -9,6 +9,7 @@ from __future__ import annotations import json +from dataclasses import dataclass from datetime import timedelta from typing import Any @@ -19,14 +20,34 @@ MoqJsonStreamConfig, MoqJsonStreamConsumer, MoqJsonStreamProducer, + MoqJsonValue, ) -from ._records import _opt_us +from ._records import _opt_from_us, _opt_us from ._stream import _Stream from .publish import BroadcastProducer, TrackDemand, TrackProducer from .subscribe import TrackConsumer -__all__ = ["SnapshotConsumer", "SnapshotProducer", "StreamConsumer", "StreamProducer"] +__all__ = ["SnapshotConsumer", "SnapshotProducer", "StreamConsumer", "StreamProducer", "Timed"] + + +@dataclass(frozen=True) +class Timed: + """A parsed JSON value and the timestamp of the frame that carried it. + + ``timestamp`` is on the broadcast clock, and ``None`` on an untimed track. It is + rounded down to a whole microsecond, so a tick on a track like 90 kHz may not + survive a read and republish exactly. + """ + + value: Any + timestamp: timedelta | None + + @staticmethod + def _from_ffi(value: MoqJsonValue | None) -> Timed | None: + if value is None: + return None + return Timed(value=json.loads(value.value), timestamp=_opt_from_us(value.timestamp_us)) class SnapshotProducer: @@ -93,12 +114,13 @@ def finish(self) -> None: self._inner.finish() -class SnapshotConsumer(_Stream[Any]): - """Async iterator over a JSON snapshot track, yielding the latest value (lossy). +class SnapshotConsumer(_Stream[Timed]): + """Async iterator over a JSON snapshot track, yielding each state in order (lossy). - Each item is a parsed Python object. A consumer that has fallen behind collapses the backlog - and yields only the latest value. Pass the same ``compression`` the producer used. Usable as - an async context manager that cancels on exit. + Each item is a :class:`Timed`. Groups older than the subscription's ``max_delay`` are skipped, + so a playhead reader sets that budget to its distance behind live; :meth:`latest` skips to the + newest state instead. Pass the same ``compression`` the producer used. Usable as an async + context manager that cancels on exit. """ def __init__(self, track: TrackConsumer, *, compression: bool = False) -> None: @@ -106,35 +128,34 @@ def __init__(self, track: TrackConsumer, *, compression: bool = False) -> None: config = MoqJsonSnapshotConfig(compression=compression) self._inner = MoqJsonSnapshotConsumer(track._inner, config) - # Overrides the base so a JSON null still yields None; only a missing raw value ends the stream. - async def __anext__(self) -> Any: - value = await self._inner.next() - if value is None: - raise StopAsyncIteration - return json.loads(value) + async def _next(self) -> Timed | None: + return Timed._from_ffi(await self._inner.next()) + + async def latest(self) -> Timed | None: + """Skip to the newest state, or wait for one. ``None`` once the track ends. + + Skipped states are gone; iteration continues after this one. + """ + return Timed._from_ffi(await self._inner.latest()) def cancel(self) -> None: - """Cancel all current and future next() calls.""" + """Cancel all current and future reads.""" self._inner.cancel() -class StreamConsumer(_Stream[Any]): +class StreamConsumer(_Stream[Timed]): """Async iterator over a JSON stream track, yielding every record in order (lossless). - Each item is a parsed Python object. Pass the same ``compression`` the producer used. Usable - as an async context manager that cancels on exit. + Each item is a :class:`Timed`. Pass the same ``compression`` the producer used. Usable as an + async context manager that cancels on exit. """ def __init__(self, track: TrackConsumer, *, compression: bool = False) -> None: config = MoqJsonStreamConfig(compression=compression) self._inner = MoqJsonStreamConsumer(track._inner, config) - # Overrides the base so a JSON null still yields None; only a missing raw value ends the stream. - async def __anext__(self) -> Any: - value = await self._inner.next() - if value is None: - raise StopAsyncIteration - return json.loads(value) + async def _next(self) -> Timed | None: + return Timed._from_ffi(await self._inner.next()) def cancel(self) -> None: """Cancel all current and future next() calls.""" diff --git a/py/moq-rs/tests/test_local.py b/py/moq-rs/tests/test_local.py index a71234853bb..38621566fa3 100644 --- a/py/moq-rs/tests/test_local.py +++ b/py/moq-rs/tests/test_local.py @@ -488,18 +488,23 @@ async def test_fetch_group_and_serve_dynamic_miss(): async def test_json_snapshot_roundtrip(): broadcast = moq.BroadcastProducer() - producer = moq.json.SnapshotProducer(broadcast, broadcast.publish_track("status"), compression=True) + track = broadcast.publish_track("status", moq.TrackInfo(timescale=1_000_000)) + producer = moq.json.SnapshotProducer(broadcast, track, compression=True) consumer = moq.json.SnapshotConsumer(await broadcast.consume().subscribe_track("status"), compression=True) - producer.update({"state": "live", "viewers": 1}) - value = await asyncio.wait_for(anext(consumer), timeout=5.0) - assert value == {"state": "live", "viewers": 1} + for viewers in range(1, 5): + producer.update({"viewers": viewers}, timestamp=timedelta(milliseconds=viewers)) - producer.update({"state": "live", "viewers": 2}) + # Iteration yields every state in order; latest() skips the buffered third. + value = await asyncio.wait_for(anext(consumer), timeout=5.0) + assert value == moq.json.Timed({"viewers": 1}, timedelta(milliseconds=1)) value = await asyncio.wait_for(anext(consumer), timeout=5.0) - assert value == {"state": "live", "viewers": 2} + assert value == moq.json.Timed({"viewers": 2}, timedelta(milliseconds=2)) + value = await asyncio.wait_for(consumer.latest(), timeout=5.0) + assert value == moq.json.Timed({"viewers": 4}, timedelta(milliseconds=4)) producer.finish() + assert await asyncio.wait_for(consumer.latest(), timeout=5.0) is None async def test_json_stream_roundtrip(): @@ -510,7 +515,7 @@ async def test_json_stream_roundtrip(): for n in range(3): producer.append({"n": n}) record = await asyncio.wait_for(anext(consumer), timeout=5.0) - assert record == {"n": n} + assert record == moq.json.Timed({"n": n}, None) producer.finish() diff --git a/quest/m1/README.md b/quest/m1/README.md index b644de46934..2e51156a06b 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -69,7 +69,6 @@ lands before the cut. - [More tests under load](/quest/m1/test-flakes-2/README.md) - the second round of load-only failures, one quest per flake, fixed at the cause - [A public model](/quest/m1/public-model/README.md) - every model item a built-in session reaches is public, unnecessary, or marked deliberately private, so a custom publisher, subscriber, or relay builds from the public model alone - [Lazy discovery](/quest/m1/lazy-discovery/README.md) - a `consume` session requests announcements only for prefixes the origin is watching, in JS then Rust; `follow`, `broadcasts`, and the `discovery` option go away -- [Data consumer timestamps](/quest/m1/data-consumer-timestamps.md) - json and binary consumers return each value's timestamp, in Rust and every binding; snapshots add `latest()` beside an in-order `next()` - [Exact timestamps in every binding](/quest/m1/binding-timestamps.md) - bindings carry `{value, scale}` timestamps instead of lossy microseconds - [JSON window timestamps](/quest/m1/json-window-timestamps.md) - Rust and JS window consumers return each event's frame timestamp like the snapshot and stream consumers - [JavaScript FETCH](/quest/m1/js-fetch.md) - browser publishers answer IETF FETCH through the JS ranges request surface diff --git a/quest/m1/binding-timestamps.md b/quest/m1/binding-timestamps.md index fd8058c6a24..49014093c58 100644 --- a/quest/m1/binding-timestamps.md +++ b/quest/m1/binding-timestamps.md @@ -6,8 +6,8 @@ Every binding (moq-ffi, moq-c, cpp and cpp/obs, and the py, swift, kt, go and dart wrappers) carries media timestamps as an exact `{value, scale}` record mirroring `moq_net::Timestamp`, with untimed as absent. It replaces the lossy microsecond fields on frames, datagrams, media and audio frames, producer -arguments, and the data values that -[Data consumer timestamps](/quest/m1/data-consumer-timestamps.md) adds. +arguments, and the JSON data values (`MoqJsonValue`, `moq_json_value`, and each +wrapper's `Timed`). The Rust core and `@moq/net` already carry exact timestamps. Wire: none. Public API: breaking in every binding. @@ -55,10 +55,6 @@ Verification: round-trip a 90 kHz tick, a present zero, and untimed through moq-ffi, moq-c, C++ and every wrapper's tests; refuse scale 0, an out-of-range scale, and an out-of-range value; `just test interop --all`. -## Required - -- [Data consumer timestamps](/quest/m1/data-consumer-timestamps.md) - lands first with the interim microsecond fields this replaces - ## Related - [Controlled time](/quest/m1/time/README.md) - wall and monotonic time, not media timestamps; keep the types distinct diff --git a/quest/m1/compress/migrate.md b/quest/m1/compress/migrate.md index ada8ad631ea..7258fc9012f 100644 --- a/quest/m1/compress/migrate.md +++ b/quest/m1/compress/migrate.md @@ -50,4 +50,3 @@ Public API: removals listed above. Wire: the hang catalog loses - [Core](/quest/m1/compress/core.md) - Rust compresses opted-in tracks - [JavaScript](/quest/m1/compress/js.md) - js/publish and js/watch need the JS side before they drop `.z` -- [Data consumer timestamps](/quest/m1/data-consumer-timestamps.md) - in review on the same flate and binary consumers this deletes; one owner at a time diff --git a/quest/m1/data-consumer-timestamps.md b/quest/m1/data-consumer-timestamps.md deleted file mode 100644 index c5a72eb1629..00000000000 --- a/quest/m1/data-consumer-timestamps.md +++ /dev/null @@ -1,45 +0,0 @@ -# [M] Data consumers return each value's timestamp - -## Goal - -A JSON or binary data consumer returns each value with its frame's timestamp, -so an application can carry data onto another track, or sync it with video, -at the exact time it was published. Covers moq-json, moq-flate, the moq-mux -wrappers, moq-ffi, and every binding wrapper. - -## Plan - -An application that translates MAVLink into its own telemetry must keep -each value's timestamp to stay in sync with video. - -Every `moq_net::Frame` carries an optional timestamp (`None` when untimed, -since #4822), but the consumers decode only `frame.payload` and drop it: `moq_mux::{json,binary}::Consumer::next` and the -moq-json and moq-flate snapshot and stream consumers they wrap. Each snapshot -state carries the timestamp of the frame that produced it. - -Decided: `next()` and `poll_next()` return `Timed`, the same type the producers take. -`at` is the frame's media timestamp on the track's timescale, and `None` for -an untimed frame: absence survives the wire rather than becoming arrival -time (decided 2026-10-01, landed in [#4822](https://github.com/moq-dev/moq/pull/4822)). A republisher passes `at` straight to a moq-mux data producer. - -Decided (2026-10-01): snapshot consumers get two reads, both returning -`Timed`. Today the moq-json snapshot consumer applies every buffered delta -but yields only the newest state, and moq-flate jumps to the newest group, so -a 9s state is lost when 11s is already buffered. A caller syncing to a -playhead needs the newest state at or before it. - -- `next()` (and `poll_next()`) yields every state in order, like a stream - consumer. A reader that falls behind the drift budget (`Lagged`) still - jumps to the newest group, so a slow reader stays bounded. -- `latest()` (and `poll_latest()`) skips to the newest state, today's - behavior, kept as the optimization for callers that only want the current - value. -- Same names in the moq-mux wrappers, moq-ffi, and every binding. - -Existing moq-ffi consumers return the timestamp too, and the py, swift, -kt, go, and dart wrappers and `doc/lib/*` follow. Flate FFI consumers are not -added: [Compressed tracks](/quest/m1/compress/README.md) replaces moq-flate's -track modes with a `compress` flag on any track. `@moq/json` and `@moq/flate` -already return `Timed` with the same `next()` and `latest()`. - -Public API: breaking. Wire: none. diff --git a/quest/m1/json-window-timestamps.md b/quest/m1/json-window-timestamps.md index e878e3687b4..962b76724a9 100644 --- a/quest/m1/json-window-timestamps.md +++ b/quest/m1/json-window-timestamps.md @@ -3,9 +3,8 @@ ## Goal The moq-json and `@moq/json` window consumers return each event with a -timestamp. In JS all three data consumers then return `Timed`; the Rust -snapshot and stream consumers follow in -[Data consumer timestamps](/quest/m1/data-consumer-timestamps.md). +timestamp. The Rust and JS snapshot and stream consumers already return +`Timed` (#5099, #5677); the window consumer is the last one. ## Plan @@ -22,11 +21,7 @@ stream consumers `Timed`. The window consumer (`window::Consumer::next` in - Update in-repo readers (`js/hang` timeline, `js/room` chat) and add a line to `## Unreleased` of each affected package's `MIGRATE.md`. No new doc page. - Land before the next `@moq/json` release, so its break ships together with - #5099's. Not blocked on the Rust snapshot and stream quest: both Rust breaks - use the same `moq_net::Timed` and ship in the next moq-json release together. + #5099's. The Rust break uses the same `moq_net::Timed` as #5677 and ships in + the next moq-json release with it. Public API: breaking (Rust and JS). Wire: none. - -## Related - -- [Data consumer timestamps](/quest/m1/data-consumer-timestamps.md) - the Rust snapshot and stream side of the same change diff --git a/quest/m1/public-model/serve.md b/quest/m1/public-model/serve.md index 7f30db46b6c..e08d6b52774 100644 --- a/quest/m1/public-model/serve.md +++ b/quest/m1/public-model/serve.md @@ -61,6 +61,9 @@ Rust items (`rs/moq-net/src/model/`), as they stood at b4eab0db1: `setGroups({ start: { included: latest.group } })`, whose omitted end removes the caller's cap. The rebinding must keep that cap; land it with a regression test where a capped snapshot reader stays below its cap. + A datagram also advances `largest()` past every group, stranding the + read. The Rust consumers instead drain the groups the reader already + has and keep the newest (#5677), which needs no floor at all. - `rs/moq-hls` export narrows a fetched group with `set_frames(..end)`. - Readers that set a floor once right after `poll_largest` (moq-audio, moq-video, moq-cli play, moq-json window, moq-mux) fit the builder. diff --git a/quest/m1/refresh/json.md b/quest/m1/refresh/json.md index afd9086fc12..c499af0a2ab 100644 --- a/quest/m1/refresh/json.md +++ b/quest/m1/refresh/json.md @@ -91,4 +91,3 @@ method because moq-ffi drives it internally). Wire: none. ## Related - [JSON window timestamps](/quest/m1/json-window-timestamps.md) - touches the same window code -- [Data consumer timestamps](/quest/m1/data-consumer-timestamps.md) - touches the same snapshot consumers and bindings diff --git a/rs/hang/src/catalog/consumer.rs b/rs/hang/src/catalog/consumer.rs index bf10060c0f0..b7d57aef55e 100644 --- a/rs/hang/src/catalog/consumer.rs +++ b/rs/hang/src/catalog/consumer.rs @@ -39,7 +39,9 @@ impl Consumer { /// Poll for the next catalog update. pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>>> { - let value = ready!(self.inner.poll_next(waiter)).map_err(|err| crate::Error::Json(err.to_string()))?; + let value = ready!(self.inner.poll_latest(waiter)) + .map_err(|err| crate::Error::Json(err.to_string()))? + .map(|value| value.value); let catalog = value.map(Catalog::from_value).transpose()?; if let Some(catalog) = catalog.as_ref() { catalog.check_renditions()?; diff --git a/rs/moq-boy/src/input.rs b/rs/moq-boy/src/input.rs index 93ec558ccef..5da2942145e 100644 --- a/rs/moq-boy/src/input.rs +++ b/rs/moq-boy/src/input.rs @@ -129,8 +129,8 @@ async fn handle_viewer_commands( moq_json::snapshot::Consumer::::new(track, moq_json::snapshot::consumer::Config::default()); loop { - let command = match commands.next().await { - Ok(Some(command)) => command, + let command = match commands.latest().await { + Ok(Some(command)) => command.value, Ok(None) => break, // A malformed command is skipped; a track error ends the stream. Err(err @ (moq_json::Error::Json(_) | moq_json::Error::DuplicateKey)) => { diff --git a/rs/moq-c/MIGRATE.md b/rs/moq-c/MIGRATE.md index 6d7a0db5163..dac364224f1 100644 --- a/rs/moq-c/MIGRATE.md +++ b/rs/moq-c/MIGRATE.md @@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first. ## Unreleased +### JSON values carry their timestamp (#5677) + +`moq_json_value` gains `timestamp_us` and `timestamp_present`, which changes its ABI. +A snapshot consumer's callback now fires for each state in order instead of only the newest. +Timestamps are rounded down to a whole microsecond. + ### Ranges filter reads (#5929) A raw track consumer now reads only the groups its `ranges` cover, rather than every group from the lowest start to the last absolute end. diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index fd318e25216..7948ed0c53e 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -456,6 +456,14 @@ pub struct moq_json_value { /// The JSON document as UTF-8, NOT NULL terminated. pub json: *const c_char, pub json_len: usize, + + /// The timestamp of the frame that carried the value in microseconds, or 0 when + /// `timestamp_present` is false. Rounded down to a whole microsecond, so a tick on a track + /// like 90 kHz may not survive a read and republish exactly. + pub timestamp_us: u64, + + /// Whether the frame carries a timestamp. A value read from an untimed track does not. + pub timestamp_present: bool, } /// Information about a frame of media. @@ -3867,8 +3875,8 @@ pub extern "C" fn moq_consume_datagrams_cancel(task: u32) -> i32 { /// Subscribe to a JSON snapshot track (lossy latest-value) by name. /// -/// `on_value` is called with a positive value ID for each new latest value; a consumer that -/// falls behind collapses the backlog and only sees the newest. It is called exactly once more +/// `on_value` is called with a positive value ID for each state, in order; a group superseded by +/// a newer one before it is read is skipped, so a slow reader stays bounded. It is called exactly once more /// with a terminal `0` (track ended / closed) or a negative error, after which `user_data` is /// never touched again, so release it there. Read each value with [moq_consume_json_value] and /// release it with [moq_consume_json_value_free]. Pass the same compression the producer used. @@ -3937,7 +3945,7 @@ pub unsafe extern "C" fn moq_consume_json_stream( /// Read a JSON value delivered via a [moq_consume_json_snapshot] or [moq_consume_json_stream] callback. /// -/// Fills `dst.json` / `dst.json_len`; the pointer is valid until the value is released with +/// Fills `dst.json` / `dst.json_len` and the frame timestamp; the pointer is valid until the value is released with /// [moq_consume_json_value_free]. /// /// Returns a zero on success, or a negative code on failure. diff --git a/rs/moq-c/src/consume.rs b/rs/moq-c/src/consume.rs index 4c2871f0cef..4b248c878c7 100644 --- a/rs/moq-c/src/consume.rs +++ b/rs/moq-c/src/consume.rs @@ -146,7 +146,7 @@ enum RawStep { Stop, } -/// A raw frame or datagram timestamp in microseconds, `None` when untimed. +/// A raw frame, datagram, or JSON value timestamp in microseconds, `None` when untimed. fn timestamp_us(timestamp: Option) -> Result, Error> { timestamp .map(|timestamp| timestamp.as_micros().try_into()) @@ -154,6 +154,21 @@ fn timestamp_us(timestamp: Option) -> Result, Er .map_err(|_| Error::TimestampOverflow(moq_net::TimeOverflow)) } +/// A JSON value pre-serialized to a string, with its frame timestamp. +struct JsonValue { + json: String, + timestamp_us: Option, +} + +impl JsonValue { + fn new(value: moq_net::Timed) -> Result { + Ok(Self { + json: serde_json::to_string(&value.value)?, + timestamp_us: timestamp_us(value.at)?, + }) + } +} + #[derive(Default)] pub struct Consume { /// Active broadcast consumers. @@ -186,8 +201,8 @@ pub struct Consume { /// JSON consumer tasks (snapshot and stream share this slab). json_task: NonZeroSlab, - /// Buffered JSON values ready for consumption, pre-serialized to a string. - json_value: NonZeroSlab, + /// Buffered JSON values ready for consumption. + json_value: NonZeroSlab, } impl Consume { @@ -823,8 +838,8 @@ impl Consume { /// Subscribe to a JSON snapshot track (lossy latest-value) by name. /// - /// `on_value` is called with a value ID for each new latest value; a consumer that falls - /// behind collapses the backlog and only sees the newest. Values must be released with + /// `on_value` is called with a value ID for each state, in order; a group superseded by a + /// newer one before it is read is skipped. Values must be released with /// [`Self::json_value_close`]. Pass the same compression the producer used. pub fn json_snapshot( &mut self, @@ -874,9 +889,10 @@ impl Consume { }, }; - let json = serde_json::to_string(&value)?; - // Hold the lock only to buffer the value; release it before the callback. - let value_id = State::lock().consume.json_value.insert(json)?; + // Serialize before locking, hold the lock only to buffer the value, and release it + // before the callback. + let value = JsonValue::new(value)?; + let value_id = State::lock().consume.json_value.insert(value)?; callback.call(Ok(value_id)); } } @@ -933,8 +949,8 @@ impl Consume { }, }; - let json = serde_json::to_string(&value)?; - let value_id = State::lock().consume.json_value.insert(json)?; + let value = JsonValue::new(value)?; + let value_id = State::lock().consume.json_value.insert(value)?; callback.call(Ok(value_id)); } } @@ -948,11 +964,13 @@ impl Consume { /// Fill `dst` with a JSON value delivered via a consumer callback. The pointer is valid /// until the value is released with [`Self::json_value_close`]. pub fn json_value(&self, value: Id, dst: &mut moq_json_value) -> Result<(), Error> { - let json = self.json_value.get(value).ok_or(Error::FrameNotFound)?; + let value = self.json_value.get(value).ok_or(Error::FrameNotFound)?; *dst = moq_json_value { - json: json.as_str().as_ptr() as *const c_char, - json_len: json.len(), + json: value.json.as_ptr() as *const c_char, + json_len: value.json.len(), + timestamp_us: value.timestamp_us.unwrap_or_default(), + timestamp_present: value.timestamp_us.is_some(), }; Ok(()) diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index 3dec15a63fc..366481f9cde 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -1934,6 +1934,7 @@ fn json_snapshot_publish_consume() { ) }); + let mut last_us = 0; for expected in [r#"{"a":1}"#, r#"{"a":2}"#] { assert_eq!( unsafe { moq_publish_json_snapshot_update(producer, expected.as_ptr() as *const c_char, expected.len()) }, @@ -1943,8 +1944,14 @@ fn json_snapshot_publish_consume() { let mut value = moq_json_value { json: std::ptr::null(), json_len: 0, + timestamp_us: 0, + timestamp_present: false, }; assert_eq!(unsafe { moq_consume_json_value(value_id, &mut value) }, 0); + // The producer stamps each value with the broadcast clock. + assert!(value.timestamp_present); + assert!(value.timestamp_us >= last_us); + last_us = value.timestamp_us; let received = unsafe { std::slice::from_raw_parts(value.json.cast::(), value.json_len) }; assert_eq!( serde_json::from_slice::(received).unwrap(), @@ -1997,6 +2004,7 @@ fn json_stream_publish_consume() { ) }); + let mut last_us = 0; for expected in [r#"{"n":0}"#, r#"{"n":1}"#, r#"{"n":2}"#] { assert_eq!( unsafe { moq_publish_json_stream_append(producer, expected.as_ptr() as *const c_char, expected.len()) }, @@ -2006,8 +2014,14 @@ fn json_stream_publish_consume() { let mut value = moq_json_value { json: std::ptr::null(), json_len: 0, + timestamp_us: 0, + timestamp_present: false, }; assert_eq!(unsafe { moq_consume_json_value(value_id, &mut value) }, 0); + // The producer stamps each value with the broadcast clock. + assert!(value.timestamp_present); + assert!(value.timestamp_us >= last_us); + last_us = value.timestamp_us; let received = unsafe { std::slice::from_raw_parts(value.json.cast::(), value.json_len) }; assert_eq!( serde_json::from_slice::(received).unwrap(), diff --git a/rs/moq-ffi/MIGRATE.md b/rs/moq-ffi/MIGRATE.md index c71da0c5e19..7c30fe86411 100644 --- a/rs/moq-ffi/MIGRATE.md +++ b/rs/moq-ffi/MIGRATE.md @@ -5,6 +5,12 @@ These are the UniFFI names; each language wrapper has its own `MIGRATE.md` in it ## Unreleased +### JSON consumers return `MoqJsonValue` (#5677) + +`MoqJsonSnapshotConsumer::next` and `MoqJsonStreamConsumer::next` return `MoqJsonValue { value, timestamp_us }` instead of a string; `timestamp_us` is null when untimed. +Snapshot `next` now yields every state in order; call the new `MoqJsonSnapshotConsumer::latest` for the old collapse-to-newest read. +Timestamps are rounded down to a whole microsecond. + ### Ranges filter reads (#5929) A subscription now reads only the groups its own `ranges` cover, rather than starting at the lowest one and taking whatever arrives. diff --git a/rs/moq-ffi/src/json.rs b/rs/moq-ffi/src/json.rs index 429c7963361..82353be924f 100644 --- a/rs/moq-ffi/src/json.rs +++ b/rs/moq-ffi/src/json.rs @@ -10,7 +10,7 @@ use std::sync::Arc; use serde_json::Value; -use crate::consumer::MoqTrackConsumer; +use crate::consumer::{MoqTrackConsumer, timestamp_us}; use crate::demand::MoqTrackDemand; use crate::error::MoqError; use crate::ffi::Task; @@ -166,20 +166,32 @@ impl MoqJsonSnapshotProducer { } } -struct SnapshotConsumer { - inner: moq_json::snapshot::Consumer, +/// A JSON value read from a track, with the timestamp of the frame that carried it. +#[derive(Clone, Debug, PartialEq, uniffi::Record)] +pub struct MoqJsonValue { + /// The JSON document. + pub value: String, + /// The frame's timestamp on the broadcast clock 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. + pub timestamp_us: Option, } -impl SnapshotConsumer { - async fn next(&mut self) -> Result, MoqError> { - match self.inner.next().await? { - Some(value) => Ok(Some(serde_json::to_string(&value)?)), - None => Ok(None), - } +impl MoqJsonValue { + fn new(timed: Option>) -> Result, MoqError> { + let Some(timed) = timed else { return Ok(None) }; + Ok(Some(Self { + value: serde_json::to_string(&timed.value)?, + timestamp_us: timed.at.map(timestamp_us).transpose()?, + })) } } -/// Consumes a JSON snapshot track, yielding the latest reconstructed value. +struct SnapshotConsumer { + inner: moq_json::snapshot::Consumer, +} + +/// Consumes a JSON snapshot track, yielding each reconstructed state or skipping to the newest. #[derive(uniffi::Object)] pub struct MoqJsonSnapshotConsumer { task: Task, @@ -201,14 +213,26 @@ impl MoqJsonSnapshotConsumer { })) } - /// Get the next value as a JSON string. Returns `None` once the track ends. + /// Get the next state in order. Returns `None` once the track ends. /// - /// A consumer that has fallen behind collapses the backlog and yields only the latest value. - pub async fn next(&self) -> Result, MoqError> { - self.task.run(|mut state| async move { state.next().await }).await + /// Groups older than the subscription's `max_delay_us` are skipped, so a playhead + /// reader sets that budget to its distance behind live. + pub async fn next(&self) -> Result, MoqError> { + self.task + .run(|mut state| async move { MoqJsonValue::new(state.inner.next().await?) }) + .await } - /// Cancel all current and future `next()` calls. + /// Skip to the newest state, or wait for one. Returns `None` once the track ends. + /// + /// Skipped states are gone; a later `next()` continues after this one. + pub async fn latest(&self) -> Result, MoqError> { + self.task + .run(|mut state| async move { MoqJsonValue::new(state.inner.latest().await?) }) + .await + } + + /// Cancel all current and future `next()` and `latest()` calls. /// /// Terminal: the subscription is released here, not when the handle is. pub fn cancel(&self) { @@ -283,15 +307,6 @@ struct StreamConsumer { inner: moq_json::stream::Consumer, } -impl StreamConsumer { - async fn next(&mut self) -> Result, MoqError> { - match self.inner.next().await? { - Some(value) => Ok(Some(serde_json::to_string(&value)?)), - None => Ok(None), - } - } -} - /// Consumes an ordered log of JSON records, yielding every record in order. #[derive(uniffi::Object)] pub struct MoqJsonStreamConsumer { @@ -314,9 +329,11 @@ impl MoqJsonStreamConsumer { })) } - /// Get the next record as a JSON string. Returns `None` once the track ends. - pub async fn next(&self) -> Result, MoqError> { - self.task.run(|mut state| async move { state.next().await }).await + /// Get the next record. Returns `None` once the track ends. + pub async fn next(&self) -> Result, MoqError> { + self.task + .run(|mut state| async move { MoqJsonValue::new(state.inner.next().await?) }) + .await } /// Cancel all current and future `next()` calls. diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 58ea35e056d..a65983bedc1 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -1094,6 +1094,7 @@ async fn catalog_now_fixes_the_clock_for_later_media() { assert!(matches!(catalog.now(), Err(MoqError::Closed))); } +/// `next` yields every state in order with its timestamp; `latest` skips to the newest. #[tokio::test] async fn json_snapshot_roundtrip() { let broadcast = MoqBroadcastProducer::new().unwrap(); @@ -1101,7 +1102,13 @@ async fn json_snapshot_roundtrip() { delta_ratio: 8, compression: true, }; - let track = broadcast.publish_track("meta".into(), None).unwrap(); + let info = MoqTrackInfo { + priority: 127, + max_age_us: None, + timescale: Some(1_000_000), + refresh: None, + }; + let track = broadcast.publish_track("meta".into(), Some(info)).unwrap(); let producer = MoqJsonSnapshotProducer::new(&broadcast, &track, config.clone()).unwrap(); assert!( matches!(track.demand(), Err(MoqError::Closed)), @@ -1109,34 +1116,54 @@ async fn json_snapshot_roundtrip() { ); let consumer = MoqJsonSnapshotConsumer::new(&*subscribe(&broadcast, "meta").await, config).unwrap(); - producer.update(r#"{"a":1}"#.into(), None).unwrap(); + for n in 1..=4 { + producer.update(format!(r#"{{"a":{n}}}"#), Some(n * 1000)).unwrap(); + } + let value = tokio::time::timeout(TIMEOUT, consumer.next()) .await .expect("timed out waiting for json snapshot") .unwrap() .expect("expected a value"); assert_eq!( - serde_json::from_str::(&value).unwrap(), + serde_json::from_str::(&value.value).unwrap(), serde_json::json!({ "a": 1 }) ); + assert_eq!(value.timestamp_us, Some(1000)); - // A second update supersedes the first; a late reader collapses to the latest. - producer.update(r#"{"a":2}"#.into(), None).unwrap(); let value = tokio::time::timeout(TIMEOUT, consumer.next()) .await .expect("timed out waiting for json snapshot delta") .unwrap() .expect("expected a value"); assert_eq!( - serde_json::from_str::(&value).unwrap(), + serde_json::from_str::(&value.value).unwrap(), serde_json::json!({ "a": 2 }) ); + assert_eq!(value.timestamp_us, Some(2000)); + + // `latest` skips the buffered third state. + let value = tokio::time::timeout(TIMEOUT, consumer.latest()) + .await + .expect("timed out waiting for the latest json snapshot") + .unwrap() + .expect("expected a value"); + assert_eq!( + serde_json::from_str::(&value.value).unwrap(), + serde_json::json!({ "a": 4 }) + ); + assert_eq!(value.timestamp_us, Some(4000)); producer.finish().unwrap(); assert!(matches!( - producer.update(r#"{"a":3}"#.into(), None), + producer.update(r#"{"a":5}"#.into(), Some(5000)), Err(MoqError::Closed) )); + let end = tokio::time::timeout(TIMEOUT, consumer.next()) + .await + .expect("timed out waiting for the end of the json snapshot") + .unwrap(); + assert_eq!(end, None); } /// JSON producers report subscriber demand through a handle, like the media and raw track producers. @@ -1245,9 +1272,10 @@ async fn json_stream_roundtrip() { .unwrap() .expect("expected a record"); assert_eq!( - serde_json::from_str::(&value).unwrap(), + serde_json::from_str::(&value.value).unwrap(), serde_json::json!({ "n": n }) ); + assert_eq!(value.timestamp_us, None, "an untimed track yields no timestamp"); } producer.finish().unwrap(); } @@ -1282,7 +1310,7 @@ async fn json_on_requested_track() { .expect("timed out waiting for json stream record") .unwrap() .expect("expected a record"); - assert_eq!(value, r#"{"n":1}"#); + assert_eq!(value.value, r#"{"n":1}"#); } /// A JSON consumer refuses a track that has already read a group, and closes one it takes over. diff --git a/rs/moq-flate/MIGRATE.md b/rs/moq-flate/MIGRATE.md index a9fdeef918e..97a02c1bc80 100644 --- a/rs/moq-flate/MIGRATE.md +++ b/rs/moq-flate/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Consumers return `Timed` (#5677) + +Snapshot and stream `next` / `poll_next` return `moq_net::Timed`: read `.value` for the payload and `.at` for the frame's exact timestamp, `None` when untimed. +Snapshot `next` now yields every state in order inside the subscription's `max_delay`; call `latest` / `poll_latest` for the old collapse-to-newest read. + ### moq-binary is moq-flate The opaque `snapshot` and `stream` tracks moved beside the codec, so `moq_binary::X` is `moq_flate::X`; the wire is unchanged. diff --git a/rs/moq-flate/src/lib.rs b/rs/moq-flate/src/lib.rs index 53170f4780d..1c17293eef9 100644 --- a/rs/moq-flate/src/lib.rs +++ b/rs/moq-flate/src/lib.rs @@ -2,8 +2,8 @@ //! //! Two track modes carry the payloads: //! -//! - [`snapshot`]: **lossy**. One value updated over time; a consumer only gets the most recent -//! one. Older values are superseded and dropped. +//! - [`snapshot`]: **lossy**. One value updated over time; `next` yields states inside the drift +//! budget, while `latest` skips to the newest state. Superseded groups may be lost. //! - [`stream`]: **lossless**. An ordered append-log of self-contained payloads, delivered in order //! with nothing superseded. Bounded by the group budget: see [`stream`] for what happens once //! it is spent. diff --git a/rs/moq-flate/src/snapshot/consumer.rs b/rs/moq-flate/src/snapshot/consumer.rs index 9ce149d1112..0fea756618d 100644 --- a/rs/moq-flate/src/snapshot/consumer.rs +++ b/rs/moq-flate/src/snapshot/consumer.rs @@ -1,4 +1,4 @@ -//! Consuming an opaque value from a track. +//! Consuming opaque values and their timestamps from a track. use std::task::Poll; @@ -8,10 +8,7 @@ use crate::Result; pub use super::Config; -/// Consumes an opaque value from a track, yielding the newest one. -/// -/// Jumps to the newest group and reads the value out of it, so a late joiner starts at the current -/// value rather than replaying superseded ones. +/// Consumes opaque states in order with `next`, or skips to the newest with `latest`. pub struct Consumer { track: moq_net::track::Ordered, group: Option, @@ -34,84 +31,88 @@ impl Consumer { } } - /// Get the next value, or `None` once the track ends. - pub async fn next(&mut self) -> Result> { + /// Get the next state and its exact frame timestamp, or `None` once the track ends. + pub async fn next(&mut self) -> Result>> { kio::wait(|waiter| self.poll_next(waiter)).await } - /// Poll for the next value, without blocking. + /// Poll for each state in order, retaining the subscription's drift budget. /// - /// Jumps to the newest group and drains everything buffered in it, yielding only the last value: - /// the earlier ones are already superseded, so a consumer that has fallen behind catches up to - /// the head in a single step. A compressed group's frames are still decoded in order, since they - /// share one window; only the yield is skipped. Switching to a newer group discards the older one. + /// Finish the current group before taking the next. Lost groups resynchronize from a + /// later snapshot. A timed playhead reader sets its subscription's `max_delay` to its + /// distance behind live; the default zero budget keeps only groups not yet superseded. + pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>>> { + self.poll_read(waiter, false) + } + + /// Skip to the newest buffered state and its timestamp, or wait for a new one. + pub async fn latest(&mut self) -> Result>> { + kio::wait(|waiter| self.poll_latest(waiter)).await + } + + /// Poll for the newest state, discarding superseded groups and intermediate yields. /// - /// A group the transport can no longer serve is discarded the same way, not reported: on a - /// snapshot track its content is superseded by definition, so the reader waits for the - /// replacement. Only a failure of the track itself ends the stream. - pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>> { - // Drain to the newest group, starting a cold window whenever we switch. - let track_finished = loop { - match self.track.poll_next_group(waiter)? { - Poll::Ready(Some(group)) => { - self.group = Some(group); - self.flate = self.compression.then(crate::Decoder::new); - } - Poll::Ready(None) => break true, - Poll::Pending => break false, + /// Frames inside the newest group are still decoded in order. What this skips is gone; + /// a subsequent `next` resumes after it. An untimed frame keeps `at: None`. + pub fn poll_latest(&mut self, waiter: &kio::Waiter) -> Poll>>> { + self.poll_read(waiter, true) + } + + fn poll_read(&mut self, waiter: &kio::Waiter, skip: bool) -> Poll>>> { + if skip { + // Jump to the newest group the reader already has; a newer one supersedes the rest. + // Taking groups, not the track's largest position, ignores datagrams (which share + // the sequence space) and stays inside any cap the caller set on the reader. + while let Poll::Ready(Some(group)) = self.track.poll_next_group(waiter)? { + self.group = Some(group); + self.flate = self.compression.then(crate::Decoder::new); } - }; + } - // Decode every frame currently buffered in the group, keeping only the last. - // `poll_read_frame` returns an owned `Poll`, so the borrow of `self.group` ends before the - // match arms, leaving `decode` (and clearing the group) free to take `&mut self`. let mut latest = None; - let mut group_pending = false; - while let Some(group) = &mut self.group { + loop { + let Some(group) = &mut self.group else { + match self.track.poll_next_group(waiter)? { + Poll::Ready(Some(group)) => { + self.group = Some(group); + self.flate = self.compression.then(crate::Decoder::new); + continue; + } + Poll::Ready(None) => return Poll::Ready(Ok(latest)), + Poll::Pending => break, + } + }; + match group.poll_read_frame(waiter) { - Poll::Ready(Ok(Some(frame))) => latest = Some(self.decode(&frame.payload)?), - // The current group is exhausted; wait for a newer one. - Poll::Ready(Ok(None)) => { - self.group = None; - break; + Poll::Ready(Ok(Some(frame))) => { + let value = moq_net::Timed { + value: self.decode(&frame.payload)?, + at: frame.timestamp, + }; + if !skip { + return Poll::Ready(Ok(Some(value))); + } + latest = Some(value); } - // The transport can no longer serve the rest of this group: it was superseded and - // reclaimed (`Old`), dropped under memory pressure (`Evicted`), or read past the - // drift budget (`Lagged`). A snapshot reader only ever wants the newest value, so a - // group whose content is gone is never fatal: drop it and wait for its replacement. - // A track- or session-level failure still arrives through `poll_next_group` above. - Poll::Ready(Err(err)) => { - let sequence = group.sequence; + Poll::Ready(Ok(None)) => self.group = None, + Poll::Ready(Err( + err @ (moq_net::Error::Old + | moq_net::Error::Evicted + | moq_net::Error::Lagged + | moq_net::Error::GroupTooLarge), + )) => { + // A later group starts a fresh snapshot and compression window. + tracing::warn!(track = self.track.name(), group = group.sequence, error = ?err, + "snapshot group lost; waiting for a newer one"); self.group = None; - tracing::warn!( - track = self.track.name(), - group = sequence, - error = ?err, - "snapshot group lost; waiting for a newer one" - ); - break; - } - // The group is still open but has nothing buffered yet. - Poll::Pending => { - group_pending = true; - break; } + Poll::Ready(Err(err)) => return Poll::Ready(Err(err.into())), + Poll::Pending => break, } } - - if let Some(payload) = latest { - return Poll::Ready(Ok(Some(payload))); - } - - // An open group may still deliver frames even after the track finishes (it was appended before - // the finish), so wait on it rather than ending the stream. - if group_pending { - return Poll::Pending; - } - - match track_finished { - true => Poll::Ready(Ok(None)), - false => Poll::Pending, + match latest { + Some(value) => Poll::Ready(Ok(Some(value))), + None => Poll::Pending, } } @@ -123,3 +124,148 @@ impl Consumer { }) } } + +#[cfg(test)] +mod timestamps { + use super::*; + + #[test] + fn exact_timestamps_survive_reads() { + for compression in [crate::Compression::None, crate::Compression::Deflate] { + let scale = moq_net::Timescale::new(90_000).unwrap(); + let track = moq_net::broadcast::Info::new() + .produce() + .create_track("data", moq_net::track::Info::default().with_timescale(scale)) + .unwrap(); + let mut subscriber = track.subscribe(None); + let mut subscription = subscriber.subscription(); + subscription.max_delay = std::time::Duration::MAX; + subscriber.update(subscription).unwrap(); + let config = crate::snapshot::Config { compression }; + let mut producer = crate::snapshot::Producer::new(track, config.clone()); + let mut consumer = Consumer::new(subscriber, config); + let mut newest = Consumer::new(producer.consume(), crate::snapshot::Config { compression }); + let times = [0, 810_001, 990_007]; + for (n, ticks) in times.into_iter().enumerate() { + let at = moq_net::Timestamp::new(ticks, scale).unwrap(); + producer + .update(moq_net::Timed { + value: Bytes::from(vec![n as u8; 32]), + at: Some(at), + }) + .unwrap(); + } + producer.finish().unwrap(); + for (n, ticks) in times.into_iter().enumerate() { + let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&kio::Waiter::noop()) else { + panic!("missing state {n}") + }; + assert_eq!(value.value[0] as u64, n as u64); + let at = value.at.unwrap(); + assert_eq!(at.value(), ticks); + assert_eq!(at.scale(), scale); + } + assert!(matches!( + consumer.poll_next(&kio::Waiter::noop()), + Poll::Ready(Ok(None)) + )); + let Poll::Ready(Ok(Some(value))) = newest.poll_latest(&kio::Waiter::noop()) else { + panic!("missing latest state") + }; + assert_eq!(value.value[0] as u64, 2); + assert_eq!(value.at.unwrap().value(), 990_007); + assert_eq!(value.at.unwrap().scale(), scale); + } + } + + #[test] + fn latest_advances_the_same_cursor_as_next() { + let track = moq_net::broadcast::Info::new() + .produce() + .create_track("data", moq_net::track::Info::default()) + .unwrap(); + let subscriber = track.subscribe(None); + let config = crate::snapshot::Config { + compression: crate::Compression::Deflate, + }; + let mut producer = crate::snapshot::Producer::new(track, config.clone()); + let mut consumer = Consumer::new(subscriber, config); + let waiter = kio::Waiter::noop(); + for n in 0..3 { + producer.update(Bytes::from(vec![n as u8; 32])).unwrap(); + if n == 0 { + let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&waiter) else { + panic!("missing first state") + }; + assert_eq!(value.value[0] as u64, 0); + assert_eq!(value.at, None); + } + } + let Poll::Ready(Ok(Some(value))) = consumer.poll_latest(&waiter) else { + panic!("missing latest state") + }; + assert_eq!(value.value[0] as u64, 2); + assert_eq!(value.at, None); + assert!(consumer.poll_next(&waiter).is_pending()); + let n = 3; + producer.update(Bytes::from(vec![n as u8; 32])).unwrap(); + let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&waiter) else { + panic!("missing new state") + }; + assert_eq!(value.value[0] as u64, 3); + producer.finish().unwrap(); + assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(None)))); + } + + #[test] + fn latest_keeps_a_cap_set_before_the_consumer() { + let track = moq_net::broadcast::Info::new() + .produce() + .create_track("data", moq_net::track::Info::default()) + .unwrap(); + let mut subscriber = track.subscribe(None); + subscriber.set_groups(..5); + let config = crate::snapshot::Config::default(); + let mut producer = crate::snapshot::Producer::new(track, config.clone()); + let mut consumer = Consumer::new(subscriber, config); + let waiter = kio::Waiter::noop(); + for n in 0..4u8 { + producer.update(Bytes::from(vec![n; 32])).unwrap(); + } + let Poll::Ready(Ok(Some(value))) = consumer.poll_latest(&waiter) else { + panic!("missing latest state") + }; + assert_eq!(value.value[0], 3); + for n in 4..7u8 { + producer.update(Bytes::from(vec![n; 32])).unwrap(); + } + let Poll::Ready(Ok(Some(value))) = consumer.poll_latest(&waiter) else { + panic!("missing capped state") + }; + assert_eq!(value.value[0], 4, "latest crossed the cap"); + assert!(consumer.poll_next(&waiter).is_pending()); + } + + #[test] + fn latest_ignores_datagrams() { + let track = moq_net::broadcast::Info::new() + .produce() + .create_track("data", moq_net::track::Info::default()) + .unwrap(); + let mut raw = track.clone(); + let subscriber = track.subscribe(None); + let config = crate::snapshot::Config::default(); + let mut producer = crate::snapshot::Producer::new(track, config.clone()); + let mut consumer = Consumer::new(subscriber, config); + let waiter = kio::Waiter::noop(); + for n in 0..2u8 { + producer.update(Bytes::from(vec![n; 32])).unwrap(); + } + // A datagram shares the sequence space, so it outnumbers every group. + raw.append_datagram(None, &b"noise"[..]).unwrap(); + let Poll::Ready(Ok(Some(value))) = consumer.poll_latest(&waiter) else { + panic!("a datagram hid the latest state") + }; + assert_eq!(value.value[0], 1); + } +} diff --git a/rs/moq-flate/src/snapshot/mod.rs b/rs/moq-flate/src/snapshot/mod.rs index 25f93ee587d..91887e02d3c 100644 --- a/rs/moq-flate/src/snapshot/mod.rs +++ b/rs/moq-flate/src/snapshot/mod.rs @@ -1,10 +1,9 @@ //! Lossy latest-value opaque publishing over [`moq-net`](moq_net) tracks. //! -//! One opaque value updated over time, for consumers that only care about the current state (a -//! poster image, a serialized state blob). This mode is **lossy** by design: a consumer yields only -//! the most recent value. A late joiner (or a consumer that falls behind) jumps straight to the -//! newest group, and older groups are dropped entirely. For an ordered log where every payload is -//! preserved, use [`stream`](crate::stream) instead. +//! One value updated over time. `Consumer::next` yields each state with its frame timestamp, +//! in order within the subscription's drift budget; `Consumer::latest` skips to the newest +//! state. Superseded groups may be lost, so use [`stream`](crate::stream) when every record +//! must survive. A playhead reader sets `max_delay` to its distance behind the live edge. //! //! On the wire each value is one group holding one frame, so a group is self-contained and a //! consumer never needs an older one. With [`Config::compression`] set to @@ -111,16 +110,42 @@ mod test { Consumer::new(track, cfg(compression)) } - /// Drain every value currently available without blocking. + /// Drain newest states currently available without blocking. fn drain(mut consumer: Consumer) -> Vec { let waiter = kio::Waiter::noop(); let mut out = Vec::new(); - while let Poll::Ready(Ok(Some(payload))) = consumer.poll_next(&waiter) { - out.push(payload); + while let Poll::Ready(Ok(Some(payload))) = consumer.poll_latest(&waiter) { + out.push(payload.value); } out } + #[test] + fn next_preserves_intermediate_states() { + let (mut producer, mut track) = producer(true); + let mut subscription = track.subscription(); + subscription.max_delay = std::time::Duration::MAX; + track.update(subscription).unwrap(); + for value in [b"one".as_slice(), b"two", b"three"] { + producer.update(value).unwrap(); + } + producer.finish().unwrap(); + let mut consumer = consume(track, true); + let mut values = Vec::new(); + while let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&kio::Waiter::noop()) { + assert_eq!(value.at, None); + values.push(value.value); + } + assert_eq!( + values, + vec![ + Bytes::from_static(b"one"), + Bytes::from_static(b"two"), + Bytes::from_static(b"three") + ] + ); + } + /// A snapshot group the transport can no longer serve -- `Old` when the relay reclaims a /// superseded group, `Evicted` under memory pressure, `Lagged` past the drift budget -- is not /// fatal. A snapshot reader only wants the newest value, so it drops the group and takes the @@ -139,13 +164,13 @@ mod test { group .write_frame(moq_net::Timestamp::now(), Bytes::from_static(b"one")) .unwrap(); - assert!(matches!(consumer.poll_next(&waiter), Poll::Ready(Ok(Some(v))) if v == "one")); - assert!(consumer.poll_next(&waiter).is_pending()); + assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(Some(v))) if v.value == "one")); + assert!(consumer.poll_latest(&waiter).is_pending()); // The relay reclaims the group out from under the reader. group.abort(moq_net::Error::Old).unwrap(); assert!( - consumer.poll_next(&waiter).is_pending(), + consumer.poll_latest(&waiter).is_pending(), "a lost group must not end the reader" ); @@ -154,7 +179,7 @@ mod test { group .write_frame(moq_net::Timestamp::now(), Bytes::from_static(b"two")) .unwrap(); - assert!(matches!(consumer.poll_next(&waiter), Poll::Ready(Ok(Some(v))) if v == "two")); + assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(Some(v))) if v.value == "two")); } /// A stamped payload is written at its capture time, and the returned size is the encoded frame @@ -205,8 +230,8 @@ mod test { for n in 0..3u8 { producer.update(vec![n]).unwrap(); - match consumer.poll_next(&waiter) { - Poll::Ready(Ok(Some(payload))) => assert_eq!(&payload[..], &[n]), + match consumer.poll_latest(&waiter) { + Poll::Ready(Ok(Some(payload))) => assert_eq!(&payload.value[..], &[n]), other => panic!("expected value, got {other:?}"), } } @@ -285,7 +310,7 @@ mod test { let mut consumer = consume(track, false); let waiter = kio::Waiter::noop(); - assert!(matches!(consumer.poll_next(&waiter), Poll::Ready(Ok(Some(_))))); - assert!(matches!(consumer.poll_next(&waiter), Poll::Ready(Ok(None)))); + assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(Some(_))))); + assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(None)))); } } diff --git a/rs/moq-flate/src/snapshot/producer.rs b/rs/moq-flate/src/snapshot/producer.rs index bd59e71727c..b2dcd80f075 100644 --- a/rs/moq-flate/src/snapshot/producer.rs +++ b/rs/moq-flate/src/snapshot/producer.rs @@ -12,7 +12,8 @@ pub use super::Config; /// Publishes an opaque value over a track, one value per group. /// /// Each [`update`](Self::update) rolls a new group holding the whole value, so a consumer only ever -/// needs the newest group and older ones are dropped. For a log where every payload survives, use +/// needs the newest group to reconstruct the value. A reader can retain earlier states within its +/// drift budget. For a log where every payload survives, use /// [`stream`](crate::stream) instead. /// /// Cheaply clonable: clones share one underlying track, like other MoQ producers. diff --git a/rs/moq-flate/src/stream/consumer.rs b/rs/moq-flate/src/stream/consumer.rs index 9ead657026f..e52d48e678e 100644 --- a/rs/moq-flate/src/stream/consumer.rs +++ b/rs/moq-flate/src/stream/consumer.rs @@ -47,13 +47,13 @@ impl Consumer { } } - /// Get the next payload, or `None` once the track ends. - pub async fn next(&mut self) -> Result> { + /// Get the next payload and its exact frame timestamp, or `None` once the track ends. + pub async fn next(&mut self) -> Result>> { kio::wait(|waiter| self.poll_next(waiter)).await } /// Poll for the next payload, without blocking. - pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>> { + pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>>> { loop { if self.rolled { return Poll::Ready(Err(crate::Error::Rolled)); @@ -78,7 +78,12 @@ impl Consumer { }; match group.poll_read_frame(waiter)? { - Poll::Ready(Some(frame)) => return Poll::Ready(self.decode(&frame.payload).map(Some)), + Poll::Ready(Some(frame)) => { + return Poll::Ready(Ok(Some(moq_net::Timed { + value: self.decode(&frame.payload)?, + at: frame.timestamp, + }))); + } Poll::Ready(None) => { // The log's one group is exhausted. Keep polling the track so a clean end still // reports the log as complete, and so a second group is caught as `Rolled`. @@ -105,3 +110,50 @@ impl Consumer { }) } } + +#[cfg(test)] +mod timestamps { + use super::*; + + #[test] + fn exact_timestamps_survive_reads() { + for compression in [crate::Compression::None, crate::Compression::Deflate] { + let scale = moq_net::Timescale::new(90_000).unwrap(); + let track = moq_net::broadcast::Info::new() + .produce() + .create_track("data", moq_net::track::Info::default().with_timescale(scale)) + .unwrap(); + let mut subscriber = track.subscribe(None); + let mut subscription = subscriber.subscription(); + subscription.max_delay = std::time::Duration::MAX; + subscriber.update(subscription).unwrap(); + let config = crate::stream::Config { compression }; + let mut producer = crate::stream::Producer::new(track, config.clone()); + let mut consumer = Consumer::new(subscriber, config); + let times = [0, 810_001, 990_007]; + for (n, ticks) in times.into_iter().enumerate() { + let at = moq_net::Timestamp::new(ticks, scale).unwrap(); + producer + .append(moq_net::Timed { + value: Bytes::from(vec![n as u8; 32]), + at: Some(at), + }) + .unwrap(); + } + producer.finish().unwrap(); + for (n, ticks) in times.into_iter().enumerate() { + let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&kio::Waiter::noop()) else { + panic!("missing state {n}") + }; + assert_eq!(value.value[0] as u64, n as u64); + let at = value.at.unwrap(); + assert_eq!(at.value(), ticks); + assert_eq!(at.scale(), scale); + } + assert!(matches!( + consumer.poll_next(&kio::Waiter::noop()), + Poll::Ready(Ok(None)) + )); + } + } +} diff --git a/rs/moq-flate/src/stream/mod.rs b/rs/moq-flate/src/stream/mod.rs index 1a24503c91b..0d5f190f5c8 100644 --- a/rs/moq-flate/src/stream/mod.rs +++ b/rs/moq-flate/src/stream/mod.rs @@ -140,7 +140,7 @@ mod test { let waiter = kio::Waiter::noop(); let mut out = Vec::new(); while let Poll::Ready(Ok(Some(payload))) = consumer.poll_next(&waiter) { - out.push(payload); + out.push(payload.value); } out } @@ -498,7 +498,7 @@ mod test { assert!(matches!( consumer.poll_next(&waiter), - Poll::Ready(Ok(Some(payload))) if payload == Bytes::from_static(b"first") + Poll::Ready(Ok(Some(payload))) if payload.value == Bytes::from_static(b"first") )); assert!(matches!( consumer.poll_next(&waiter), @@ -540,7 +540,7 @@ mod test { assert!(matches!( consumer.poll_next(&waiter), - Poll::Ready(Ok(Some(payload))) if payload == vec![1u8; 8] + Poll::Ready(Ok(Some(payload))) if payload.value == vec![1u8; 8] )); assert!(matches!( consumer.poll_next(&waiter), diff --git a/rs/moq-json/MIGRATE.md b/rs/moq-json/MIGRATE.md index 132a07630d0..9eeae6499ab 100644 --- a/rs/moq-json/MIGRATE.md +++ b/rs/moq-json/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Consumers return `Timed` (#5677) + +Snapshot and stream `next` / `poll_next` return `moq_net::Timed`: read `.value` for the value and `.at` for the frame's exact timestamp, `None` when untimed. +Snapshot `next` now yields every state in order inside the subscription's `max_delay`; call `latest` / `poll_latest` for the old collapse-to-newest read. + ### Duplicate keys are refused Every decoder refuses a frame in which a JSON object repeats a key with `Error::DuplicateKey`, where stream and window decoders used to keep the last copy and snapshots returned `Error::Json`. diff --git a/rs/moq-json/examples/telemetry.rs b/rs/moq-json/examples/telemetry.rs index 7fdceb8db1c..dbf377d7a33 100644 --- a/rs/moq-json/examples/telemetry.rs +++ b/rs/moq-json/examples/telemetry.rs @@ -131,8 +131,8 @@ fn verify(producer_config: snapshot::Config, ticks: u64) { .unwrap(); // The producer emits exactly one frame per update, so the live consumer yields exactly one // reconstructed value: it must match the input byte-for-byte after decompression + patching. - match consumer.poll_next(&waiter) { - Poll::Ready(Ok(Some(value))) => assert_eq!(value, expected, "tick {tick} reconstruction mismatch"), + match consumer.poll_latest(&waiter) { + Poll::Ready(Ok(Some(value))) => assert_eq!(value.value, expected, "tick {tick} reconstruction mismatch"), other => panic!("tick {tick}: expected a value, got {other:?}"), } } @@ -140,7 +140,7 @@ fn verify(producer_config: snapshot::Config, ticks: u64) { // Drain: nothing left and the stream ends cleanly. assert!( - matches!(consumer.poll_next(&waiter), Poll::Ready(Ok(None))), + matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(None))), "stream did not end cleanly" ); } @@ -174,8 +174,8 @@ fn verify_late_joiner(producer_config: snapshot::Config, ticks: u64) -> usize { let waiter = kio::Waiter::noop(); let mut last = None; let mut yielded = 0; - while let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&waiter) { - last = Some(value); + while let Poll::Ready(Ok(Some(value))) = consumer.poll_latest(&waiter) { + last = Some(value.value); yielded += 1; } assert_eq!( diff --git a/rs/moq-json/examples/teleop.rs b/rs/moq-json/examples/teleop.rs index 59ba7009566..0d9499a0bc0 100644 --- a/rs/moq-json/examples/teleop.rs +++ b/rs/moq-json/examples/teleop.rs @@ -52,18 +52,21 @@ fn main() -> Result<(), Box> { commands.finish()?; let waiter = kio::Waiter::noop(); - let Poll::Ready(Ok(Some(latest))) = state_reader.poll_next(&waiter) else { + // `latest` skips the stale steering values; `next` would replay them in order. + let Poll::Ready(Ok(Some(latest))) = state_reader.poll_latest(&waiter) else { return Err("finished control track did not yield its latest state".into()); }; - assert_eq!(latest, Control { speed: 0, turn: 0 }); - println!("Current control: {latest:?}"); - assert!(matches!(state_reader.poll_next(&waiter), Poll::Ready(Ok(None)))); - for expected in expected { + assert_eq!(latest.value, Control { speed: 0, turn: 0 }); + assert_eq!(latest.at, Some(Timestamp::from_millis(2)?)); + println!("Current control: {:?} at {:?}", latest.value, latest.at); + assert!(matches!(state_reader.poll_latest(&waiter), Poll::Ready(Ok(None)))); + for (tick, expected) in expected.into_iter().enumerate() { let Poll::Ready(Ok(Some(command))) = command_reader.poll_next(&waiter) else { return Err("finished command track lost an ordered command".into()); }; - assert_eq!(command, expected); - println!("Command: {command:?}"); + assert_eq!(command.value, expected); + assert_eq!(command.at, Some(Timestamp::from_millis(tick as u64)?)); + println!("Command: {:?} at {:?}", command.value, command.at); } assert!(matches!(command_reader.poll_next(&waiter), Poll::Ready(Ok(None)))); Ok(()) diff --git a/rs/moq-json/src/lib.rs b/rs/moq-json/src/lib.rs index f64c30715a2..67d8a3ede7b 100644 --- a/rs/moq-json/src/lib.rs +++ b/rs/moq-json/src/lib.rs @@ -1,7 +1,7 @@ //! JSON publishing over [`moq-net`](moq_net) tracks, in three modes: //! -//! - [`snapshot`]: **lossy**. One JSON value updated over time; a consumer only gets the most -//! recent value. Intermediate updates are collapsed and older groups are dropped. +//! - [`snapshot`]: **lossy**. One JSON value updated over time; `next` yields states inside the +//! drift budget, while `latest` skips to the newest state. Superseded groups may be lost. //! - [`stream`]: **lossless**. An ordered append-log of self-contained records; every record is //! preserved and delivered in order, nothing is ever superseded. //! - [`window`]: **bounded**. An ordered run of records appended to the back and dropped from the diff --git a/rs/moq-json/src/snapshot/consumer.rs b/rs/moq-json/src/snapshot/consumer.rs index 7457b8b947b..e630536ce56 100644 --- a/rs/moq-json/src/snapshot/consumer.rs +++ b/rs/moq-json/src/snapshot/consumer.rs @@ -26,7 +26,8 @@ pub struct Config { /// Consumes a JSON value from a track, reconstructing it from snapshots and deltas. /// /// A [`Decoder`] that owns its track: it reads groups, routes each frame by its position, and -/// yields the reconstructed value. When something else already owns the track, use the [`Decoder`] +/// yields each reconstructed state and its frame timestamp, or skips to the newest with `latest`. +/// When something else already owns the track, use the [`Decoder`] /// directly. pub struct Consumer { track: moq_net::track::Ordered, @@ -49,94 +50,98 @@ impl Consumer { } } - /// Get the next reconstructed value, or `None` once the track ends. - pub async fn next(&mut self) -> Result> + /// Get the next state and its exact frame timestamp, or `None` once the track ends. + pub async fn next(&mut self) -> Result>> where T: Unpin, { kio::wait(|waiter| self.poll_next(waiter)).await } - /// Poll for the next reconstructed value, without blocking. + /// Poll for each state in order, retaining the subscription's drift budget. /// - /// Jumps to the newest group, reads its snapshot, and applies deltas in order. All frames already - /// buffered in the group are applied in one poll but only the resulting *latest* value is yielded: - /// the intermediate reconstructions are stale, so a late joiner (or any consumer that has fallen - /// behind) catches up to the head in a single step instead of replaying every superseded state. - /// Frames must still be decoded in order (the DEFLATE window and merge patches are sequential); - /// only the per-frame deserialize and yield are skipped. Switching to a newer group discards the - /// older one. + /// Finish the current group before taking the next. Lost groups resynchronize from a + /// later snapshot. A timed playhead reader sets its subscription's `max_delay` to its + /// distance behind live; the default zero budget keeps only groups not yet superseded. + pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>>> { + self.poll_read(waiter, false) + } + + /// Skip to the newest buffered state and its timestamp, or wait for a new one. + pub async fn latest(&mut self) -> Result>> + where + T: Unpin, + { + kio::wait(|waiter| self.poll_latest(waiter)).await + } + + /// Poll for the newest state, discarding superseded groups and intermediate yields. /// - /// A group the transport can no longer serve is discarded the same way, not reported: on a - /// snapshot track its content is superseded by definition, so the reader waits for the - /// replacement. Only a failure of the track itself ends the stream. - pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>> { - // Drain to the newest group, resetting reconstruction state whenever we switch. - let track_finished = loop { - match self.track.poll_next_group(waiter)? { - Poll::Ready(Some(group)) => { - self.group = Some(group); - // The next frame is the new group's snapshot, which also restarts the decoder's window. - self.frames_read = 0; - } - Poll::Ready(None) => break true, - Poll::Pending => break false, + /// Frames inside the newest group are still decoded in order. What this skips is gone; + /// a subsequent `next` resumes after it. An untimed frame keeps `at: None`. + pub fn poll_latest(&mut self, waiter: &kio::Waiter) -> Poll>>> { + self.poll_read(waiter, true) + } + + fn poll_read(&mut self, waiter: &kio::Waiter, skip: bool) -> Poll>>> { + if skip { + // Jump to the newest group the reader already has; a newer one supersedes the rest. + // Taking groups, not the track's largest position, ignores datagrams (which share + // the sequence space) and stays inside any cap the caller set on the reader. + while let Poll::Ready(Some(group)) = self.track.poll_next_group(waiter)? { + self.group = Some(group); + self.frames_read = 0; } - }; + } + + let mut advanced = None; + let mut finished = false; + loop { + let Some(group) = &mut self.group else { + match self.track.poll_next_group(waiter)? { + Poll::Ready(Some(group)) => { + self.group = Some(group); + self.frames_read = 0; + continue; + } + Poll::Ready(None) => { + finished = true; + break; + } + Poll::Pending => break, + } + }; - // Apply every frame currently buffered in the group, tracking whether any moved us forward and - // whether the group is still open with nothing buffered yet (vs. exhausted). - // `poll_read_frame` returns an owned `Poll`, so the borrow of `self.group` ends before the - // match arms, leaving `apply` (and clearing the group) free to take `&mut self`. - let mut advanced = false; - let mut group_pending = false; - while let Some(group) = &mut self.group { match group.poll_read_frame(waiter) { Poll::Ready(Ok(Some(frame))) => { self.apply(&frame.payload)?; - advanced = true; - } - // The current group is exhausted; wait for a newer one. - Poll::Ready(Ok(None)) => { - self.group = None; - break; + if !skip { + return Poll::Ready(Ok(self.decoder.decode()?.map(|value| moq_net::Timed { + value, + at: frame.timestamp, + }))); + } + advanced = Some(frame.timestamp); } - // The transport can no longer serve the rest of this group: it was superseded and - // reclaimed (`Old`), dropped under memory pressure (`Evicted`), or read past the - // drift budget (`Lagged`). A snapshot reader only ever wants the newest value, so a - // group whose content is gone is never fatal: drop it and wait for its replacement. - // A track- or session-level failure still arrives through `poll_next_group` above. - Poll::Ready(Err(err)) => { - let sequence = group.sequence; + Poll::Ready(Ok(None)) => self.group = None, + Poll::Ready(Err( + err @ (moq_net::Error::Old + | moq_net::Error::Evicted + | moq_net::Error::Lagged + | moq_net::Error::GroupTooLarge), + )) => { + // A later group starts a fresh snapshot and compression window. + tracing::warn!(track = self.track.name(), group = group.sequence, error = ?err, + "snapshot group lost; waiting for a newer one"); self.group = None; - tracing::warn!( - track = self.track.name(), - group = sequence, - error = ?err, - "snapshot group lost; waiting for a newer one" - ); - break; - } - // The group is still open but has nothing buffered yet. - Poll::Pending => { - group_pending = true; - break; } + Poll::Ready(Err(err)) => return Poll::Ready(Err(err.into())), + Poll::Pending => break, } } - - if advanced { - // Deserialize once, from the head of the backlog we just drained. - return Poll::Ready(Ok(self.decoder.decode()?)); - } - - // An open group may still deliver frames even after the track finishes (it was appended before - // the finish), so wait on it rather than ending the stream. - if group_pending { - return Poll::Pending; - } - - if track_finished { + if let Some(at) = advanced { + Poll::Ready(Ok(self.decoder.decode()?.map(|value| moq_net::Timed { value, at }))) + } else if finished { Poll::Ready(Ok(None)) } else { Poll::Pending @@ -153,3 +158,178 @@ impl Consumer { Ok(()) } } + +#[cfg(test)] +mod timestamps { + use super::*; + + #[test] + fn exact_timestamps_survive_reads() { + for compression in [crate::Compression::None, crate::Compression::Deflate] { + let scale = moq_net::Timescale::new(90_000).unwrap(); + let track = moq_net::broadcast::Info::new() + .produce() + .create_track("data", moq_net::track::Info::default().with_timescale(scale)) + .unwrap(); + let mut subscriber = track.subscribe(None); + let mut subscription = subscriber.subscription(); + subscription.max_delay = std::time::Duration::MAX; + subscriber.update(subscription).unwrap(); + let mut config = crate::snapshot::Config::default().with_delta_ratio(100); + config.compression = compression; + let mut producer = crate::snapshot::Producer::new(track, config.clone()); + let mut consumer = Consumer::::new( + subscriber, + crate::snapshot::consumer::Config { + compression, + ..Default::default() + }, + ); + let mut newest = Consumer::::new( + producer.consume(), + crate::snapshot::consumer::Config { + compression, + ..Default::default() + }, + ); + let times = [0, 810_001, 990_007]; + for (n, ticks) in times.into_iter().enumerate() { + let at = moq_net::Timestamp::new(ticks, scale).unwrap(); + producer + .update(moq_net::Timed { + value: &serde_json::json!({"n": n, "padding": "large-repeated-value-for-deltas"}), + at: Some(at), + }) + .unwrap(); + if n == 1 { + producer.cut().unwrap(); + } + } + producer.finish().unwrap(); + for (n, ticks) in times.into_iter().enumerate() { + let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&kio::Waiter::noop()) else { + panic!("missing state {n}") + }; + assert_eq!(value.value["n"].as_u64().unwrap(), n as u64); + let at = value.at.unwrap(); + assert_eq!(at.value(), ticks); + assert_eq!(at.scale(), scale); + } + assert!(matches!( + consumer.poll_next(&kio::Waiter::noop()), + Poll::Ready(Ok(None)) + )); + let Poll::Ready(Ok(Some(value))) = newest.poll_latest(&kio::Waiter::noop()) else { + panic!("missing latest state") + }; + assert_eq!(value.value["n"].as_u64().unwrap(), 2); + assert_eq!(value.at.unwrap().value(), 990_007); + assert_eq!(value.at.unwrap().scale(), scale); + } + } + + #[test] + fn latest_advances_the_same_cursor_as_next() { + let track = moq_net::broadcast::Info::new() + .produce() + .create_track("data", moq_net::track::Info::default()) + .unwrap(); + let subscriber = track.subscribe(None); + let config = crate::snapshot::Config { + compression: crate::Compression::Deflate, + ..crate::snapshot::Config::default().with_delta_ratio(100) + }; + let mut producer = crate::snapshot::Producer::new(track, config.clone()); + let mut consumer = Consumer::::new( + subscriber, + crate::snapshot::consumer::Config { + compression: crate::Compression::Deflate, + ..Default::default() + }, + ); + let waiter = kio::Waiter::noop(); + for n in 0..3 { + producer + .update(&serde_json::json!({"n": n, "padding": "large-repeated-value-for-deltas"})) + .unwrap(); + if n == 0 { + let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&waiter) else { + panic!("missing first state") + }; + assert_eq!(value.value["n"].as_u64().unwrap(), 0); + assert_eq!(value.at, None); + } + } + let Poll::Ready(Ok(Some(value))) = consumer.poll_latest(&waiter) else { + panic!("missing latest state") + }; + assert_eq!(value.value["n"].as_u64().unwrap(), 2); + assert_eq!(value.at, None); + assert!(consumer.poll_next(&waiter).is_pending()); + let n = 3; + producer + .update(&serde_json::json!({"n": n, "padding": "large-repeated-value-for-deltas"})) + .unwrap(); + let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&waiter) else { + panic!("missing new state") + }; + assert_eq!(value.value["n"].as_u64().unwrap(), 3); + producer.finish().unwrap(); + assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(None)))); + } + + #[test] + fn latest_keeps_a_cap_set_before_the_consumer() { + let track = moq_net::broadcast::Info::new() + .produce() + .create_track("data", moq_net::track::Info::default()) + .unwrap(); + let mut subscriber = track.subscribe(None); + subscriber.set_groups(..5); + let mut producer = crate::snapshot::Producer::new(track, crate::snapshot::Config::default()); + let mut consumer = Consumer::::new(subscriber, Default::default()); + let waiter = kio::Waiter::noop(); + let mut write = |n: u64| { + producer.update(&serde_json::json!({ "n": n })).unwrap(); + producer.cut().unwrap(); + }; + for n in 0..4 { + write(n); + } + let Poll::Ready(Ok(Some(value))) = consumer.poll_latest(&waiter) else { + panic!("missing latest state") + }; + assert_eq!(value.value["n"], 3); + for n in 4..7 { + write(n); + } + let Poll::Ready(Ok(Some(value))) = consumer.poll_latest(&waiter) else { + panic!("missing capped state") + }; + assert_eq!(value.value["n"], 4, "latest crossed the cap"); + assert!(consumer.poll_next(&waiter).is_pending()); + } + + #[test] + fn latest_ignores_datagrams() { + let track = moq_net::broadcast::Info::new() + .produce() + .create_track("data", moq_net::track::Info::default()) + .unwrap(); + let mut raw = track.clone(); + let subscriber = track.subscribe(None); + let mut producer = crate::snapshot::Producer::new(track, crate::snapshot::Config::default()); + let mut consumer = Consumer::::new(subscriber, Default::default()); + let waiter = kio::Waiter::noop(); + for n in 0..2 { + producer.update(&serde_json::json!({ "n": n })).unwrap(); + producer.cut().unwrap(); + } + // A datagram shares the sequence space, so it outnumbers every group. + raw.append_datagram(None, &b"noise"[..]).unwrap(); + let Poll::Ready(Ok(Some(value))) = consumer.poll_latest(&waiter) else { + panic!("a datagram hid the latest state") + }; + assert_eq!(value.value["n"], 1); + } +} diff --git a/rs/moq-json/src/snapshot/mod.rs b/rs/moq-json/src/snapshot/mod.rs index 8f51bc3e2de..6bc03283839 100644 --- a/rs/moq-json/src/snapshot/mod.rs +++ b/rs/moq-json/src/snapshot/mod.rs @@ -1,17 +1,15 @@ //! Lossy latest-value JSON publishing over [`moq-net`](moq_net) tracks. //! -//! One JSON value updated over time, for consumers that only care about the current state (a -//! catalog, a status document). This mode is **lossy** by design: a consumer yields only the -//! most recent value. A late joiner (or a consumer that falls behind) jumps straight to the -//! newest group and collapses any buffered backlog into a single yield, and older groups are -//! dropped entirely. Intermediate updates are never replayed. For an ordered log where every -//! record is preserved, use [`stream`](crate::stream) instead. +//! One value updated over time. `Consumer::next` yields each state with its frame timestamp, +//! in order within the subscription's drift budget; `Consumer::latest` skips to the newest +//! state. Superseded groups may be lost, so use [`stream`](crate::stream) when every record +//! must survive. A playhead reader sets `max_delay` to its distance behind the live edge. //! //! On the wire the value is published as a series of groups, where each group is //! self-contained: its first frame is a full snapshot and any following frames are //! [RFC 7396](https://www.rfc-editor.org/rfc/rfc7396.html) JSON Merge Patch deltas applied in -//! order. A consumer jumps to the newest group, reads the snapshot, and applies the deltas, so -//! a late joiner never needs older groups. +//! order. Every group starts independently, so a reader can resynchronize after skipping +//! superseded groups without needing earlier deltas. //! //! Deltas are controlled by [`Config::delta_ratio`]. A ratio of `0` disables them, so every //! change is a fresh snapshot group, matching a plain "one JSON blob per group" track. @@ -196,21 +194,42 @@ mod test { assert_eq!(demand.wanted(), Some(0)); } - /// Drain every value currently available from a plaintext consumer without blocking. + /// Drain newest states currently available from a plaintext consumer without blocking. fn drain(track: moq_net::track::Subscriber) -> Vec { drain_with(Consumer::::new(track, consumer::Config::default())) } - /// Drain every value currently available from an already-built consumer without blocking. + /// Drain newest states currently available from an already-built consumer without blocking. fn drain_with(mut consumer: Consumer) -> Vec { let waiter = kio::Waiter::noop(); let mut out = Vec::new(); - while let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&waiter) { - out.push(value); + while let Poll::Ready(Ok(Some(value))) = consumer.poll_latest(&waiter) { + out.push(value.value); } out } + #[test] + fn next_preserves_intermediate_states() { + let (mut producer, mut track) = producer(cfg(100)); + let mut subscription = track.subscription(); + subscription.max_delay = std::time::Duration::MAX; + track.update(subscription).unwrap(); + for n in [1, 2, 3] { + producer + .update(&json!({ "padding": "same-long-value-so-deltas-stay-in-one-group", "n": n })) + .unwrap(); + } + producer.finish().unwrap(); + let mut consumer = Consumer::::new(track, consumer::Config::default()); + let mut values = Vec::new(); + while let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&kio::Waiter::noop()) { + assert_eq!(value.at, None); + values.push(value.value["n"].as_u64().unwrap()); + } + assert_eq!(values, vec![1, 2, 3]); + } + /// A snapshot group the transport can no longer serve -- `Old` when the relay reclaims a /// superseded group, `Evicted` under memory pressure, `Lagged` past the drift budget -- is not /// fatal. A snapshot reader only wants the newest value, so it drops the group and takes the @@ -230,13 +249,13 @@ mod test { group .write_frame(moq_net::Timestamp::now(), Bytes::from_static(br#"{"a":1}"#)) .unwrap(); - assert!(matches!(consumer.poll_next(&waiter), Poll::Ready(Ok(Some(v))) if v == json!({ "a": 1 }))); - assert!(consumer.poll_next(&waiter).is_pending()); + assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(Some(v))) if v.value == json!({ "a": 1 }))); + assert!(consumer.poll_latest(&waiter).is_pending()); // The relay reclaims the group out from under the reader. group.abort(moq_net::Error::Old).unwrap(); assert!( - consumer.poll_next(&waiter).is_pending(), + consumer.poll_latest(&waiter).is_pending(), "a lost group must not end the reader" ); @@ -245,7 +264,7 @@ mod test { group .write_frame(moq_net::Timestamp::now(), Bytes::from_static(br#"{"a":2}"#)) .unwrap(); - assert!(matches!(consumer.poll_next(&waiter), Poll::Ready(Ok(Some(v))) if v == json!({ "a": 2 }))); + assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(Some(v))) if v.value == json!({ "a": 2 }))); } /// Nothing replaces a lost group once the track is finished, so the reader ends cleanly on the @@ -263,11 +282,11 @@ mod test { group .write_frame(moq_net::Timestamp::now(), Bytes::from_static(br#"{"a":1}"#)) .unwrap(); - assert!(matches!(consumer.poll_next(&waiter), Poll::Ready(Ok(Some(v))) if v == json!({ "a": 1 }))); + assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(Some(v))) if v.value == json!({ "a": 1 }))); group.abort(moq_net::Error::Old).unwrap(); track.finish().unwrap(); - assert!(matches!(consumer.poll_next(&waiter), Poll::Ready(Ok(None)))); + assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(None)))); } #[test] @@ -390,8 +409,8 @@ mod test { for n in 1..=3 { producer.update(&json!({ "a": n })).unwrap(); - match consumer.poll_next(&waiter) { - Poll::Ready(Ok(Some(value))) => assert_eq!(value, json!({ "a": n })), + match consumer.poll_latest(&waiter) { + Poll::Ready(Ok(Some(value))) => assert_eq!(value.value, json!({ "a": n })), other => panic!("expected value, got {other:?}"), } } @@ -642,8 +661,8 @@ mod test { let mut consumer = Consumer::::new(consumer, consumer::Config::default()); let waiter = kio::Waiter::noop(); let mut last = None; - while let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&waiter) { - last = Some(value); + while let Poll::Ready(Ok(Some(value))) = consumer.poll_latest(&waiter) { + last = Some(value.value); } assert_eq!( last.unwrap(), @@ -694,8 +713,8 @@ mod test { let mut consumer = Consumer::::new(consumer, consumer::Config::default()); let waiter = kio::Waiter::noop(); let mut last = None; - while let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&waiter) { - last = Some(value); + while let Poll::Ready(Ok(Some(value))) = consumer.poll_latest(&waiter) { + last = Some(value.value); } assert_eq!( last.unwrap(), @@ -943,8 +962,8 @@ mod test { let waiter = kio::Waiter::noop(); producer.update(&json!({ "a": 1 })).unwrap(); // snapshot, group 0 - match consumer.poll_next(&waiter) { - Poll::Ready(Ok(Some(value))) => assert_eq!(value, json!({ "a": 1 })), + match consumer.poll_latest(&waiter) { + Poll::Ready(Ok(Some(value))) => assert_eq!(value.value, json!({ "a": 1 })), other => panic!("expected first value, got {other:?}"), } @@ -963,8 +982,8 @@ mod test { // The consumer jumps to the newest group and never yields a stale value. let mut last = None; - while let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&waiter) { - last = Some(value); + while let Poll::Ready(Ok(Some(value))) = consumer.poll_latest(&waiter) { + last = Some(value.value); } assert_eq!(last.unwrap(), json!({ "a": 4 })); } @@ -985,7 +1004,7 @@ mod test { let waiter = kio::Waiter::noop(); // Track is finished but the open group is empty: pending, not end-of-stream. - assert!(matches!(consumer.poll_next(&waiter), Poll::Pending)); + assert!(matches!(consumer.poll_latest(&waiter), Poll::Pending)); group .write_frame( @@ -995,8 +1014,8 @@ mod test { .unwrap(); group.finish().unwrap(); - match consumer.poll_next(&waiter) { - Poll::Ready(Ok(Some(value))) => assert_eq!(value, json!({ "a": 1 })), + match consumer.poll_latest(&waiter) { + Poll::Ready(Ok(Some(value))) => assert_eq!(value.value, json!({ "a": 1 })), other => panic!("expected the catalog value, got {other:?}"), } } diff --git a/rs/moq-json/src/stream/consumer.rs b/rs/moq-json/src/stream/consumer.rs index d43f7aa1990..12c0a548924 100644 --- a/rs/moq-json/src/stream/consumer.rs +++ b/rs/moq-json/src/stream/consumer.rs @@ -47,8 +47,8 @@ impl Consumer { } } - /// Get the next record, or `None` once the track ends. - pub async fn next(&mut self) -> Result> + /// Get the next record and its exact frame timestamp, or `None` once the track ends. + pub async fn next(&mut self) -> Result>> where T: Unpin, { @@ -56,7 +56,7 @@ impl Consumer { } /// Poll for the next record, without blocking. - pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>> { + pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>>> { loop { if self.rolled { return Poll::Ready(Err(crate::Error::Rolled)); @@ -81,7 +81,12 @@ impl Consumer { }; match group.poll_read_frame(waiter)? { - Poll::Ready(Some(frame)) => return Poll::Ready(Ok(Some(self.decoder.decode(&frame.payload)?))), + Poll::Ready(Some(frame)) => { + return Poll::Ready(Ok(Some(moq_net::Timed { + value: self.decoder.decode(&frame.payload)?, + at: frame.timestamp, + }))); + } Poll::Ready(None) => { // The log's one group is exhausted. Keep polling the track so a clean end still // reports the log as complete, and so a second group is caught as `Rolled`. @@ -100,3 +105,50 @@ impl Consumer { } } } + +#[cfg(test)] +mod timestamps { + use super::*; + + #[test] + fn exact_timestamps_survive_reads() { + for compression in [crate::Compression::None, crate::Compression::Deflate] { + let scale = moq_net::Timescale::new(90_000).unwrap(); + let track = moq_net::broadcast::Info::new() + .produce() + .create_track("data", moq_net::track::Info::default().with_timescale(scale)) + .unwrap(); + let mut subscriber = track.subscribe(None); + let mut subscription = subscriber.subscription(); + subscription.max_delay = std::time::Duration::MAX; + subscriber.update(subscription).unwrap(); + let config = crate::stream::Config { compression }; + let mut producer = crate::stream::Producer::new(track, config.clone()); + let mut consumer = Consumer::::new(subscriber, config); + let times = [0, 810_001, 990_007]; + for (n, ticks) in times.into_iter().enumerate() { + let at = moq_net::Timestamp::new(ticks, scale).unwrap(); + producer + .append(moq_net::Timed { + value: &serde_json::json!({"n": n, "padding": "large-repeated-value-for-deltas"}), + at: Some(at), + }) + .unwrap(); + } + producer.finish().unwrap(); + for (n, ticks) in times.into_iter().enumerate() { + let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&kio::Waiter::noop()) else { + panic!("missing state {n}") + }; + assert_eq!(value.value["n"].as_u64().unwrap(), n as u64); + let at = value.at.unwrap(); + assert_eq!(at.value(), ticks); + assert_eq!(at.scale(), scale); + } + assert!(matches!( + consumer.poll_next(&kio::Waiter::noop()), + Poll::Ready(Ok(None)) + )); + } + } +} diff --git a/rs/moq-json/src/stream/mod.rs b/rs/moq-json/src/stream/mod.rs index 7dcfb0d8472..7239ca41db5 100644 --- a/rs/moq-json/src/stream/mod.rs +++ b/rs/moq-json/src/stream/mod.rs @@ -159,7 +159,7 @@ mod test { let waiter = kio::Waiter::noop(); let mut out = Vec::new(); while let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&waiter) { - out.push(value); + out.push(value.value); } out } @@ -234,7 +234,7 @@ mod test { for n in 0..3 { producer.append(&json!({ "n": n })).unwrap(); match consumer.poll_next(&waiter) { - Poll::Ready(Ok(Some(value))) => assert_eq!(value, json!({ "n": n })), + Poll::Ready(Ok(Some(value))) => assert_eq!(value.value, json!({ "n": n })), other => panic!("expected record, got {other:?}"), } } @@ -419,7 +419,7 @@ mod test { assert!(matches!( consumer.poll_next(&waiter), - Poll::Ready(Ok(Some(value))) if value == json!({ "n": 0 }) + Poll::Ready(Ok(Some(value))) if value.value == json!({ "n": 0 }) )); assert!(matches!( consumer.poll_next(&waiter), diff --git a/rs/moq-mux/MIGRATE.md b/rs/moq-mux/MIGRATE.md index f3477c78df7..91b85a9b273 100644 --- a/rs/moq-mux/MIGRATE.md +++ b/rs/moq-mux/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### JSON and binary consumers return `Timed` (#5677) + +The `json` and `binary` consumers' `next` / `poll_next` return `moq_net::Timed`; read `.value` and the optional `.at`. +On a snapshot track `next` yields every state; call `latest` / `poll_latest` for the newest. `latest` on a stream track fails with `UnsupportedMode`. + ### `ts::Passthrough` describes the whole track in its `m2ts` section (#5967) The section's `randomAccess` turns false at the first group that does not start at a random access point and stays false, instead of recovering with the next two-second window; its `muxRate` follows the source's measured rate. diff --git a/rs/moq-mux/src/binary.rs b/rs/moq-mux/src/binary.rs index 0de8e72a7d9..236802643aa 100644 --- a/rs/moq-mux/src/binary.rs +++ b/rs/moq-mux/src/binary.rs @@ -348,20 +348,20 @@ impl Consumer { /// The mode the publisher chose. /// - /// [`Mode::Snapshot`] is lossy: intermediate payloads are superseded and only the newest is - /// yielded. [`Mode::Stream`] preserves every payload. Check this when the application can only + /// [`Mode::Snapshot`] is lossy: groups may be skipped once superseded beyond the subscription + /// drift budget. [`Mode::Stream`] preserves every payload. Check this when the application can only /// work with one of them. pub fn mode(&self) -> &Mode { &self.mode } /// Get the next payload, or `None` once the track ends. - pub async fn next(&mut self) -> crate::Result> { + pub async fn next(&mut self) -> crate::Result>> { kio::wait(|waiter| self.poll_next(waiter)).await } /// Poll for the next payload, without blocking. - pub fn poll_next(&mut self, waiter: &kio::Waiter) -> std::task::Poll>> { + pub fn poll_next(&mut self, waiter: &kio::Waiter) -> std::task::Poll>>> { let result = match &mut self.inner { Inner::Snapshot(consumer) => consumer.poll_next(waiter), Inner::Stream(consumer) => consumer.poll_next(waiter), @@ -372,6 +372,19 @@ impl Consumer { std::task::Poll::Pending => std::task::Poll::Pending, } } + + /// Skip to the newest snapshot state; stream mode returns `UnsupportedMode`. + pub async fn latest(&mut self) -> crate::Result>> { + kio::wait(|waiter| self.poll_latest(waiter)).await + } + + /// Poll for the newest snapshot state without discarding records from a lossless stream. + pub fn poll_latest(&mut self, waiter: &kio::Waiter) -> std::task::Poll>>> { + match &mut self.inner { + Inner::Snapshot(consumer) => consumer.poll_latest(waiter).map(|result| result.map_err(Into::into)), + Inner::Stream(_) => std::task::Poll::Ready(Err(crate::Error::UnsupportedMode(self.mode.to_string()))), + } + } } impl crate::catalog::Entry<'_, BinaryConfig> { @@ -443,8 +456,12 @@ mod test { fn drain(mut consumer: Consumer) -> Vec { let waiter = kio::Waiter::noop(); let mut out = Vec::new(); - while let Poll::Ready(Ok(Some(payload))) = consumer.poll_next(&waiter) { - out.push(payload); + while let Poll::Ready(Ok(Some(payload))) = if consumer.mode() == &Mode::Snapshot { + consumer.poll_latest(&waiter) + } else { + consumer.poll_next(&waiter) + } { + out.push(payload.value); } out } @@ -857,8 +874,11 @@ mod test { .await .unwrap(); telemetry.finish().unwrap(); - assert_eq!(reader.next().await.unwrap(), Some(Bytes::from_static(b"heartbeat"))); - assert_eq!(reader.next().await.unwrap(), None); + assert_eq!( + reader.next().await.unwrap().map(|value| value.value), + Some(Bytes::from_static(b"heartbeat")) + ); + assert_eq!(reader.next().await.unwrap().map(|value| value.value), None); } #[test] diff --git a/rs/moq-mux/src/catalog/hang/consumer.rs b/rs/moq-mux/src/catalog/hang/consumer.rs index ab9d173fb25..7f259270e1e 100644 --- a/rs/moq-mux/src/catalog/hang/consumer.rs +++ b/rs/moq-mux/src/catalog/hang/consumer.rs @@ -66,7 +66,7 @@ impl Consumer { /// that from the operator while the missing rendition surfaces much later as a track that /// never fills. pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>>> { - let value = ready!(self.inner.poll_next(waiter))?; + let value = ready!(self.inner.poll_latest(waiter))?.map(|value| value.value); let catalog = value.map(Catalog::from_value).transpose()?; if let Some(catalog) = catalog.as_ref() { catalog.check_renditions()?; diff --git a/rs/moq-mux/src/json.rs b/rs/moq-mux/src/json.rs index 0a1fdb054c6..85b6381e9ab 100644 --- a/rs/moq-mux/src/json.rs +++ b/rs/moq-mux/src/json.rs @@ -403,20 +403,20 @@ impl Consumer { /// The mode the publisher chose. /// - /// [`Mode::Snapshot`] is lossy: intermediate values are superseded and only the newest is - /// yielded. [`Mode::Stream`] preserves every record. Check this when the application can only + /// [`Mode::Snapshot`] is lossy: groups may be skipped once superseded beyond the subscription + /// drift budget. [`Mode::Stream`] preserves every record. Check this when the application can only /// work with one of them. pub fn mode(&self) -> &Mode { &self.mode } /// Get the next value, or `None` once the track ends. - pub async fn next(&mut self) -> crate::Result> { + pub async fn next(&mut self) -> crate::Result>> { kio::wait(|waiter| self.poll_next(waiter)).await } /// Poll for the next value, without blocking. - pub fn poll_next(&mut self, waiter: &kio::Waiter) -> std::task::Poll>> { + pub fn poll_next(&mut self, waiter: &kio::Waiter) -> std::task::Poll>>> { let result = match &mut self.inner { Inner::Snapshot(consumer) => consumer.poll_next(waiter), Inner::Stream(consumer) => consumer.poll_next(waiter), @@ -427,6 +427,19 @@ impl Consumer { std::task::Poll::Pending => std::task::Poll::Pending, } } + + /// Skip to the newest snapshot state; stream mode returns `UnsupportedMode`. + pub async fn latest(&mut self) -> crate::Result>> { + kio::wait(|waiter| self.poll_latest(waiter)).await + } + + /// Poll for the newest snapshot state without discarding records from a lossless stream. + pub fn poll_latest(&mut self, waiter: &kio::Waiter) -> std::task::Poll>>> { + match &mut self.inner { + Inner::Snapshot(consumer) => consumer.poll_latest(waiter).map(|result| result.map_err(Into::into)), + Inner::Stream(_) => std::task::Poll::Ready(Err(crate::Error::UnsupportedMode(self.mode.to_string()))), + } + } } impl crate::catalog::Entry<'_, JsonConfig> { @@ -488,8 +501,12 @@ mod test { fn drain(mut consumer: Consumer) -> Vec { let waiter = kio::Waiter::noop(); let mut out = Vec::new(); - while let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&waiter) { - out.push(value); + while let Poll::Ready(Ok(Some(value))) = if consumer.mode() == &Mode::Snapshot { + consumer.poll_latest(&waiter) + } else { + consumer.poll_next(&waiter) + } { + out.push(value.value); } out } @@ -794,8 +811,15 @@ mod test { let mut consumer = entry.subscribe::(&source).await.unwrap(); chat.finish().unwrap(); - assert_eq!(consumer.next().await.unwrap(), Some(json!({ "text": "hello" }))); - assert_eq!(consumer.next().await.unwrap(), None); + assert!(matches!( + consumer.poll_latest(&kio::Waiter::noop()), + Poll::Ready(Err(crate::Error::UnsupportedMode(_))) + )); + assert_eq!( + consumer.next().await.unwrap().map(|value| value.value), + Some(json!({ "text": "hello" })) + ); + assert_eq!(consumer.next().await.unwrap().map(|value| value.value), None); } /// A value that fails to serialize never reaches the track, but the log is missing it all the diff --git a/rs/moq-stats/src/aggregate.rs b/rs/moq-stats/src/aggregate.rs index a3bb8a6885a..51adae0e929 100644 --- a/rs/moq-stats/src/aggregate.rs +++ b/rs/moq-stats/src/aggregate.rs @@ -925,9 +925,9 @@ fn advance( } Poll::Pending => return changed, }, - Reader::Active(reader) => match reader.poll_next(waiter) { + Reader::Active(reader) => match reader.poll_latest(waiter) { Poll::Ready(Ok(Some(frame))) => { - node.observe(frame); + node.observe(frame.value); changed = true; } // The subscription ended: the serving session died (a failover to @@ -1457,11 +1457,11 @@ mod tests { let mut warm = moq_json::snapshot::Consumer::::new(subscriber, Default::default()); let frame = futures::executor::block_on(kio::wait(|waiter| { driver.poll(now, waiter).unwrap(); - warm.poll_next(waiter) + warm.poll_latest(waiter) })) .unwrap() .unwrap(); - assert_eq!(frame["acme/room"].bytes, 30); + assert_eq!(frame.value["acme/room"].bytes, 30); ready.wait(); release.wait(); (new, front, warm, driver) diff --git a/rs/moq-stats/src/consume.rs b/rs/moq-stats/src/consume.rs index 41a3289177d..fa515e62dbd 100644 --- a/rs/moq-stats/src/consume.rs +++ b/rs/moq-stats/src/consume.rs @@ -118,7 +118,7 @@ pub struct Traffic { impl Traffic { /// The next frame, or `None` once the track ends (the producer went away). pub async fn next(&mut self) -> Result> { - Ok(self.inner.next().await?) + Ok(self.inner.latest().await?.map(|value| value.value)) } } @@ -130,7 +130,7 @@ pub struct Sessions { impl Sessions { /// The next frame, or `None` once the track ends (the producer went away). pub async fn next(&mut self) -> Result> { - Ok(self.inner.next().await?) + Ok(self.inner.latest().await?.map(|value| value.value)) } } @@ -142,7 +142,7 @@ pub struct Totals { impl Totals { /// The next frame, or `None` once the track ends (the producer went away). pub async fn next(&mut self) -> Result> { - Ok(self.inner.next().await?) + Ok(self.inner.latest().await?.map(|value| value.value)) } } @@ -155,7 +155,7 @@ impl Presence { /// The next frame, or `None` once the track ends (the producer went away /// or its group unannounced). pub async fn next(&mut self) -> Result> { - Ok(self.inner.next().await?) + Ok(self.inner.latest().await?.map(|value| value.value)) } } @@ -170,7 +170,7 @@ impl Prefix { /// [`FrameTooLarge`](moq_net::Error::FrameTooLarge) once the frame outgrows /// the cache. pub async fn next(&mut self) -> Result> { - Ok(self.inner.next().await?) + Ok(self.inner.latest().await?.map(|value| value.value)) } } diff --git a/rs/moq-stats/src/produce.rs b/rs/moq-stats/src/produce.rs index d98dd4fce38..2c4fa9996a7 100644 --- a/rs/moq-stats/src/produce.rs +++ b/rs/moq-stats/src/produce.rs @@ -1736,7 +1736,7 @@ mod tests { drop(group); drop(cursor); let mut reader = moq_json::snapshot::Consumer::::new(subscriber, config.clone()); - let initial = pumping(&mut drain, reader.next()).await.unwrap().unwrap(); + let initial = pumping(&mut drain, reader.next()).await.unwrap().unwrap().value; assert_eq!( (initial[""].wire_tx, initial[""].wire_rx), (0, 0), @@ -1745,14 +1745,14 @@ mod tests { store(reading + delta); drain.collect(); drain.publish(Instant::now()); - let next = pumping(&mut drain, reader.next()).await.unwrap().unwrap(); + let next = pumping(&mut drain, reader.next()).await.unwrap().unwrap().value; assert_eq!((next[""].wire_tx, next[""].wire_rx), (delta, delta * 2)); if compressed { // Cross the codec's group budget naturally, leaving a completed // group cached when unsubscribe aborts the still-open last group. for extra in 1..=300 { store(reading + delta + extra); - let next = pumping(&mut drain, reader.next()).await.unwrap().unwrap(); + let next = pumping(&mut drain, reader.next()).await.unwrap().unwrap().value; assert_eq!( (next[""].wire_tx, next[""].wire_rx), (delta + extra, (delta + extra) * 2) @@ -1831,7 +1831,13 @@ mod tests { }; let mut plain = read("t.json", moq_json::Compression::None); let mut compressed = read("t.json.z", moq_json::Compression::Deflate); - let next = |reader: &mut moq_json::snapshot::Consumer| reader.next().now_or_never().unwrap(); + let next = |reader: &mut moq_json::snapshot::Consumer| { + reader + .next() + .now_or_never() + .unwrap() + .map(|value| value.map(|value| value.value)) + }; let small = Value::from([("small".to_string(), 1)]); assert_eq!(next(&mut plain).unwrap(), Some(small.clone())); assert_eq!(next(&mut compressed).unwrap(), Some(small)); diff --git a/swift/MIGRATE.md b/swift/MIGRATE.md index 4a3ab3425d5..1e51e61d5e4 100644 --- a/swift/MIGRATE.md +++ b/swift/MIGRATE.md @@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first. ## Unreleased +### JSON consumers yield `Json.Timed` (#5677) + +`Json.SnapshotConsumer` and `Json.StreamConsumer` yield `Json.Timed` (`value`, `timestampUs`) instead of the bare value; `timestampUs` is `nil` 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. diff --git a/swift/Sources/Moq/Json.swift b/swift/Sources/Moq/Json.swift index e1f36f3acf0..e6c3c6a9d7a 100644 --- a/swift/Sources/Moq/Json.swift +++ b/swift/Sources/Moq/Json.swift @@ -12,6 +12,27 @@ import MoqFFI public enum Json {} extension Json { + /// A decoded JSON value and the timestamp of the frame that carried it. + public struct Timed { + /// The decoded value. + public let value: Value + /// The frame's timestamp on the broadcast clock in microseconds, or `nil` 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. + public let timestampUs: UInt64? + + /// Pair `value` with its frame timestamp. + public init(value: Value, timestampUs: UInt64?) { + self.value = value + self.timestampUs = timestampUs + } + + init?(_ ffi: MoqJsonValue?) throws where Value: Decodable { + guard let ffi else { return nil } + self.init(value: try decodeJson(ffi.value), timestampUs: ffi.timestampUs) + } + } + /// Write side of a JSON snapshot track (lossy latest-value). /// /// Each `update` supersedes the last, so a late joiner only sees the newest value. @@ -55,11 +76,12 @@ extension Json { } } - /// Read side of a JSON snapshot track (lossy latest-value). Iterating yields the latest value, - /// collapsing the backlog for a reader that has fallen behind: `for try await value in json { ... }`. + /// Read side of a JSON snapshot track (lossy latest-value). Iterating yields each state in + /// order with its timestamp: `for try await state in json { ... }`. Groups older than the + /// subscription's `maxDelayUs` are skipped; `latest()` skips to the newest state instead. public final class SnapshotConsumer: AsyncSequence, Sendable { - /// The decoded snapshot value emitted by this sequence. - public typealias Element = Value + /// The decoded snapshot state emitted by this sequence. + public typealias Element = Timed let ffi: MoqJsonSnapshotConsumer @@ -71,10 +93,15 @@ extension Json { ffi = try MoqJsonSnapshotConsumer(track: track.ffi, config: MoqJsonSnapshotConfig(compression: compression)) } - /// The next value, decoded as `Value`, or `nil` once the track ends or is closed. - public func next() async throws -> Value? { - guard let json = try await ffi.next() else { return nil } - return try decodeJson(json) as Value + /// The next state in order, or `nil` once the track ends or is closed. + public func next() async throws -> Timed? { + try Timed(try await ffi.next()) + } + + /// Skip to the newest state, or wait for one; `nil` once the track ends or is closed. + /// Skipped states are gone, and `next()` continues after this one. + public func latest() async throws -> Timed? { + try Timed(try await ffi.latest()) } /// Cancel all current and future reads. @@ -83,7 +110,7 @@ extension Json { } /// Create an iterator that cancels native reads when iteration ends. - public func makeAsyncIterator() -> AsyncThrowingStream.Iterator { + public func makeAsyncIterator() -> AsyncThrowingStream, Swift.Error>.Iterator { moqStream(cancel: { [ffi] in ffi.cancel() }) { [self] in try await next() }.makeAsyncIterator() @@ -128,7 +155,7 @@ extension Json { /// order: `for try await record in json { ... }`. public final class StreamConsumer: AsyncSequence, Sendable { /// The decoded stream record emitted by this sequence. - public typealias Element = Value + public typealias Element = Timed let ffi: MoqJsonStreamConsumer @@ -139,10 +166,9 @@ extension Json { ffi = try MoqJsonStreamConsumer(track: track.ffi, config: MoqJsonStreamConfig(compression: compression)) } - /// The next record, decoded as `Value`, or `nil` once the track ends or is closed. - public func next() async throws -> Value? { - guard let json = try await ffi.next() else { return nil } - return try decodeJson(json) as Value + /// The next record, or `nil` once the track ends or is closed. + public func next() async throws -> Timed? { + try Timed(try await ffi.next()) } /// Cancel all current and future reads. @@ -151,7 +177,7 @@ extension Json { } /// Create an iterator that cancels native reads when iteration ends. - public func makeAsyncIterator() -> AsyncThrowingStream.Iterator { + public func makeAsyncIterator() -> AsyncThrowingStream, Swift.Error>.Iterator { moqStream(cancel: { [ffi] in ffi.cancel() }) { [self] in try await next() }.makeAsyncIterator() @@ -159,6 +185,9 @@ extension Json { } } +extension Json.Timed: Sendable where Value: Sendable {} +extension Json.Timed: Equatable where Value: Equatable {} + private func encodeJson(_ value: some Encodable) throws -> String { String(decoding: try JSONEncoder().encode(value), as: UTF8.self) } diff --git a/swift/Tests/MoqTests/SmokeTests.swift b/swift/Tests/MoqTests/SmokeTests.swift index 3c26a118ccc..d34f06f0f51 100644 --- a/swift/Tests/MoqTests/SmokeTests.swift +++ b/swift/Tests/MoqTests/SmokeTests.swift @@ -278,18 +278,23 @@ final class SmokeTests: XCTestCase { let broadcast = try BroadcastProducer() let producer = try Json.SnapshotProducer( - broadcast: broadcast, track: try broadcast.publishTrack(name: "status"), compression: true) + broadcast: broadcast, + track: try broadcast.publishTrack(name: "status", info: TrackInfo(timescale: 1_000_000)), + compression: true) let consumer = try Json.SnapshotConsumer( track: try await broadcast.consume().subscribeTrack(name: "status"), compression: true) - try producer.update(Status(state: "live", viewers: 42)) - let first = try await consumer.next() - XCTAssertEqual(first, Status(state: "live", viewers: 42)) + for viewers in 1...4 { + try producer.update(Status(state: "live", viewers: viewers), timestampUs: UInt64(viewers) * 1000) + } - // A second update supersedes the first; a late reader collapses to the latest. - try producer.update(Status(state: "live", viewers: 43)) + // `next` yields every state in order; `latest` skips the buffered third. + let first = try await consumer.next() + XCTAssertEqual(first, Json.Timed(value: Status(state: "live", viewers: 1), timestampUs: 1000)) let second = try await consumer.next() - XCTAssertEqual(second, Status(state: "live", viewers: 43)) + XCTAssertEqual(second, Json.Timed(value: Status(state: "live", viewers: 2), timestampUs: 2000)) + let latest = try await consumer.latest() + XCTAssertEqual(latest, Json.Timed(value: Status(state: "live", viewers: 4), timestampUs: 4000)) consumer.cancel() try producer.finish() @@ -310,7 +315,7 @@ final class SmokeTests: XCTestCase { for n in 0..<3 { try producer.append(Event(n: n)) let record = try await consumer.next() - XCTAssertEqual(record, Event(n: n)) + XCTAssertEqual(record, Json.Timed(value: Event(n: n), timestampUs: nil)) } consumer.cancel()