feat!: demand reads as wanted everywhere

dart/moq_ffi/lib/src/moq.dart

@@ -12624,16 +12624,15 @@ abstract class MoqFillInterface {
   MoqGroupProducer accept({required int sequence});
   void dropGroups({required int start, required int end});
   int endGroup();
-  bool isUsed();
   bool isWanted({required int sequence});
   Duration? maxDelay();
-  int? next();
   MoqOrder order();
   int priority();
   int startFrame();
   int startGroup();
-  UniffiRustFuture<void> unused();
   UniffiRustFuture<void> unwanted({required int sequence});
+  int? wanted();
+  UniffiRustFuture<int?> wantedChanged();
 }
 
 final _MoqFillFinalizer = Finalizer<Pointer<Void>>((ptr) {
@@ -12699,17 +12698,6 @@ class MoqFill implements MoqFillInterface {
     );
   }
 
-  bool isUsed() {
-    return rustCallWithLifter(
-      (status) => uniffi_moq_ffi_fn_method_moqfill_is_used(
-        uniffiClonePointer(),
-        status,
-      ),
-      FfiConverterBool.lift,
-      null,
-    );
-  }
-
   bool isWanted({required int sequence}) {
     return rustCallWithLifter(
       (status) => uniffi_moq_ffi_fn_method_moqfill_is_wanted(
@@ -12733,15 +12721,6 @@ class MoqFill implements MoqFillInterface {
     );
   }
 
-  int? next() {
-    return rustCallWithLifter(
-      (status) =>
-          uniffi_moq_ffi_fn_method_moqfill_next(uniffiClonePointer(), status),
-      FfiConverterOptionalUInt64.lift,
-      null,
-    );
-  }
-
   MoqOrder order() {
     return rustCallWithLifter(
       (status) =>
@@ -12784,18 +12763,6 @@ class MoqFill implements MoqFillInterface {
     );
   }
 
-  UniffiRustFuture<void> unused() {
-    return uniffiRustCallAsync(
-      () => uniffi_moq_ffi_fn_method_moqfill_unused(uniffiClonePointer()),
-      ffi_moq_ffi_rust_future_poll_void,
-      ffi_moq_ffi_rust_future_complete_void,
-      ffi_moq_ffi_rust_future_cancel_void,
-      ffi_moq_ffi_rust_future_free_void,
-      (_) {},
-      moqExceptionErrorHandler,
-    );
-  }
-
   UniffiRustFuture<void> unwanted({required int sequence}) {
     return uniffiRustCallAsync(
       () => uniffi_moq_ffi_fn_method_moqfill_unwanted(
@@ -12810,6 +12777,28 @@ class MoqFill implements MoqFillInterface {
       moqExceptionErrorHandler,
     );
   }
+
+  int? wanted() {
+    return rustCallWithLifter(
+      (status) =>
+          uniffi_moq_ffi_fn_method_moqfill_wanted(uniffiClonePointer(), status),
+      FfiConverterOptionalUInt64.lift,
+      null,
+    );
+  }
+
+  UniffiRustFuture<int?> wantedChanged() {
+    return uniffiRustCallAsync(
+      () =>
+          uniffi_moq_ffi_fn_method_moqfill_wanted_changed(uniffiClonePointer()),
+      ffi_moq_ffi_rust_future_poll_rust_buffer,
+      ffi_moq_ffi_rust_future_complete_rust_buffer,
+      ffi_moq_ffi_rust_future_cancel_rust_buffer,
+      ffi_moq_ffi_rust_future_free_rust_buffer,
+      FfiConverterOptionalUInt64.lift,
+      moqExceptionErrorHandler,
+    );
+  }
 }
 
 class FfiConverterMoqFill {
@@ -18874,14 +18863,6 @@ external int uniffi_moq_ffi_fn_method_moqfill_end_group(
   Pointer<RustCallStatus> uniffiStatus,
 );
 
-@Native<Int8 Function(Pointer<Void>, Pointer<RustCallStatus>)>(
-  assetId: _uniffiAssetId,
-)
-external int uniffi_moq_ffi_fn_method_moqfill_is_used(
-  Pointer<Void> ptr,
-  Pointer<RustCallStatus> uniffiStatus,
-);
-
 @Native<Int8 Function(Pointer<Void>, Uint64, Pointer<RustCallStatus>)>(
   assetId: _uniffiAssetId,
 )
@@ -18899,14 +18880,6 @@ external RustBuffer uniffi_moq_ffi_fn_method_moqfill_max_delay(
   Pointer<RustCallStatus> uniffiStatus,
 );
 
-@Native<RustBuffer Function(Pointer<Void>, Pointer<RustCallStatus>)>(
-  assetId: _uniffiAssetId,
-)
-external RustBuffer uniffi_moq_ffi_fn_method_moqfill_next(
-  Pointer<Void> ptr,
-  Pointer<RustCallStatus> uniffiStatus,
-);
-
 @Native<RustBuffer Function(Pointer<Void>, Pointer<RustCallStatus>)>(
   assetId: _uniffiAssetId,
 )
@@ -18939,17 +18912,25 @@ external int uniffi_moq_ffi_fn_method_moqfill_start_group(
   Pointer<RustCallStatus> uniffiStatus,
 );
 
-@Native<Pointer<Void> Function(Pointer<Void>)>(assetId: _uniffiAssetId)
-external Pointer<Void> uniffi_moq_ffi_fn_method_moqfill_unused(
-  Pointer<Void> ptr,
-);
-
 @Native<Pointer<Void> Function(Pointer<Void>, Uint64)>(assetId: _uniffiAssetId)
 external Pointer<Void> uniffi_moq_ffi_fn_method_moqfill_unwanted(
   Pointer<Void> ptr,
   int sequence,
 );
 
+@Native<RustBuffer Function(Pointer<Void>, Pointer<RustCallStatus>)>(
+  assetId: _uniffiAssetId,
+)
+external RustBuffer uniffi_moq_ffi_fn_method_moqfill_wanted(
+  Pointer<Void> ptr,
+  Pointer<RustCallStatus> uniffiStatus,
+);
+
+@Native<Pointer<Void> Function(Pointer<Void>)>(assetId: _uniffiAssetId)
+external Pointer<Void> uniffi_moq_ffi_fn_method_moqfill_wanted_changed(
+  Pointer<Void> ptr,
+);
+
 @Native<Pointer<Void> Function(Pointer<Void>, Pointer<RustCallStatus>)>(
   assetId: _uniffiAssetId,
 )
@@ -20487,18 +20468,12 @@ external int uniffi_moq_ffi_checksum_method_moqfill_drop_groups();
 @Native<Uint16 Function()>(assetId: _uniffiAssetId)
 external int uniffi_moq_ffi_checksum_method_moqfill_end_group();
 
-@Native<Uint16 Function()>(assetId: _uniffiAssetId)
-external int uniffi_moq_ffi_checksum_method_moqfill_is_used();
-
 @Native<Uint16 Function()>(assetId: _uniffiAssetId)
 external int uniffi_moq_ffi_checksum_method_moqfill_is_wanted();
 
 @Native<Uint16 Function()>(assetId: _uniffiAssetId)
 external int uniffi_moq_ffi_checksum_method_moqfill_max_delay();
 
-@Native<Uint16 Function()>(assetId: _uniffiAssetId)
-external int uniffi_moq_ffi_checksum_method_moqfill_next();
-
 @Native<Uint16 Function()>(assetId: _uniffiAssetId)
 external int uniffi_moq_ffi_checksum_method_moqfill_order();
 
@@ -20512,10 +20487,13 @@ external int uniffi_moq_ffi_checksum_method_moqfill_start_frame();
 external int uniffi_moq_ffi_checksum_method_moqfill_start_group();
 
 @Native<Uint16 Function()>(assetId: _uniffiAssetId)
-external int uniffi_moq_ffi_checksum_method_moqfill_unused();
+external int uniffi_moq_ffi_checksum_method_moqfill_unwanted();
 
 @Native<Uint16 Function()>(assetId: _uniffiAssetId)
-external int uniffi_moq_ffi_checksum_method_moqfill_unwanted();
+external int uniffi_moq_ffi_checksum_method_moqfill_wanted();
+
+@Native<Uint16 Function()>(assetId: _uniffiAssetId)
+external int uniffi_moq_ffi_checksum_method_moqfill_wanted_changed();
 
 @Native<Uint16 Function()>(assetId: _uniffiAssetId)
 external int uniffi_moq_ffi_checksum_method_moqgroupproducer_abort();
@@ -21207,18 +21185,12 @@ void _checkApiChecksums() {
   if (uniffi_moq_ffi_checksum_method_moqfill_end_group() != 42504) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
-  if (uniffi_moq_ffi_checksum_method_moqfill_is_used() != 6429) {
-    throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
-  }
   if (uniffi_moq_ffi_checksum_method_moqfill_is_wanted() != 47156) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
   if (uniffi_moq_ffi_checksum_method_moqfill_max_delay() != 1258) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
-  if (uniffi_moq_ffi_checksum_method_moqfill_next() != 7725) {
-    throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
-  }
   if (uniffi_moq_ffi_checksum_method_moqfill_order() != 32232) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
@@ -21231,10 +21203,13 @@ void _checkApiChecksums() {
   if (uniffi_moq_ffi_checksum_method_moqfill_start_group() != 8622) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
-  if (uniffi_moq_ffi_checksum_method_moqfill_unused() != 3173) {
+  if (uniffi_moq_ffi_checksum_method_moqfill_unwanted() != 13637) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
-  if (uniffi_moq_ffi_checksum_method_moqfill_unwanted() != 13637) {
+  if (uniffi_moq_ffi_checksum_method_moqfill_wanted() != 65141) {
+    throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
+  }
+  if (uniffi_moq_ffi_checksum_method_moqfill_wanted_changed() != 42560) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
   if (uniffi_moq_ffi_checksum_method_moqgroupproducer_abort() != 59787) {

doc/lib/c/index.md

@@ -110,7 +110,7 @@ inside. The header is the reference: each function carries a doc comment.
 - **JSON windows.** `moq_publish_json_window` publishes a sliding run of records: `moq_publish_json_window_push` appends at the back, `moq_publish_json_window_pop` drops from the front, and `moq_publish_json_window_cut` closes the open group at a natural boundary. The catalog has no window mode, so a reader names the track. `moq_consume_json_window_event` fills `moq_json_window_event` with a push (the record at `start`, in `json`), a pop of `[start, end)`, or a skip of `[start, end)` that this reader joined too late to receive. Set `checkpoint_records` to bound what each new group restates for a window you never pop.
 - **Decoded reads.** `moq_decode_audio` and `moq_decode_video` deliver read ids in order: a frame, a gap where media was lost, or an end where the publisher stopped its encoder. Check `kind` on what `moq_decode_audio_read` or `moq_decode_video_read` writes. Free each id with `moq_decode_audio_read_free` or `moq_decode_video_read_free` promptly: a held video frame holds a decoder buffer, and the decoder stalls. A decoder reads everything the track still holds unless a live player sets `start` to `MOQ_DECODE_START_LATEST`, which joins at the newest group and skips the cached backlog.
 - **Stats.** `moq_session_stats()` fills `moq_connection_stats` with `rtt_us`, `estimated_send_rate_bps`, and the byte and packet counters (`bytes_sent`, `bytes_received`, `bytes_lost`, `packets_sent`, `packets_received`, `packets_lost`). Each has a `<field>_valid` flag, false when the transport does not report it, which is not the same as zero. `moq_session_snapshot()` adds the negotiated `protocol`. A failed dial or dropped connection reaches the session's `on_state` with its `failure` code, and `moq_error()` gives the reason during that callback. 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.
+- **Demand.** A watcher from `moq_publish_track_demand()` (or the media, video, and audio variants) calls back `MOQ_DEMAND_WANTED` or `MOQ_DEMAND_UNWANTED`, 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.
 - **Disabled renditions.** A rendition with `enabled: false` has no frames coming until its publisher re-enables it. Select an enabled rendition from the latest catalog before opening a decoder. Keep one already playing while every rendition of its kind is disabled: the convenience decoders accept a disabled configuration, and a re-enable, such as an unmute, then plays without a resubscribe. Among the video renditions you can decode, prefer the highest `preference`; `0` is the default and `-1` marks a compatibility transcode.
 
 ## Reference

go/wrapper/README.md

@@ -145,7 +145,7 @@ the session closed, i.e. not a real failure) and `moq.IsAuthError(err)` (HTTP
 
 Every call that can block takes a `context.Context` and aborts cleanly when it is
 cancelled, leaving nothing running in the background. A one-shot call (a subscribe,
-a fetch, `Request`, `Resolve`, `WantedChanged`, `Used`/`Unused`, `Server.Accept`) aborts on
+a fetch, `Request`, `Resolve`, `WantedChanged`, `Server.Accept`) aborts on
 its own and leaves the object usable; a stream read cancels the stream it reads.
 See the package doc for details.
 

go/wrapper/doc.go

@@ -24,8 +24,8 @@
 // owner.
 //
 // What a cancel tears down depends on the call. A one-shot call (a subscribe, a
-// fetch, Request, Resolve, a TrackDemand's WantedChanged, a
-// Fill's Unused, Server.Accept) aborts on its own and leaves the
+// fetch, Request, Resolve, a TrackDemand's or Fill's WantedChanged,
+// Server.Accept) aborts on its own and leaves the
 // object it was made on usable, so the same
 // broadcast, producer, or server takes the next call. A stream read (any Next,
 // RecvGroup, ReadFrame, or the iterators over them) instead cancels the stream

go/wrapper/moq_test.go

@@ -657,8 +657,8 @@ func TestFetchGroupAndServeDynamicMiss(t *testing.T) {
 	if fill.StartGroup() != 7 || fill.EndGroup() != 8 || fill.Priority() != 11 {
 		t.Fatalf("unexpected fill: start=%d end=%d priority=%d", fill.StartGroup(), fill.EndGroup(), fill.Priority())
 	}
-	if next := fill.Next(); next == nil || *next != 7 || !fill.IsUsed() {
-		t.Fatalf("unexpected fill: next=%v used=%v", next, fill.IsUsed())
+	if wanted := fill.Wanted(); wanted == nil || *wanted != 7 {
+		t.Fatalf("unexpected fill: wanted=%v", wanted)
 	}
 	if !fill.IsWanted(7) || fill.IsWanted(8) {
 		t.Fatalf("unexpected wants: 7=%v 8=%v", fill.IsWanted(7), fill.IsWanted(8))

go/wrapper/publish.go

@@ -333,11 +333,11 @@ func (d *TrackRequests) Cancel() {
 
 // Fill is a run of groups readers miss, for a TrackRequests to serve.
 //
-// Accept each group you have and DropGroups the rest; Next names what is still
+// Accept each group you have and DropGroups the rest; Wanted names what is still
 // owed, in the fill's order. Releasing the fill names whatever is left as dropped
 // while a reader still wants it; once nobody does, it is withdrawn, so stop
-// serving it when Unused returns. Stop serving a single group when Unwanted
-// returns for it, and move on to Next.
+// serving it once Wanted is nil. Stop serving a single group when Unwanted
+// returns for it, and move on to Wanted.
 type Fill struct {
 	inner *ffi.MoqFill
 }
@@ -373,9 +373,15 @@ func (f *Fill) Order() Order {
 	return f.inner.Order()
 }
 
-// Next is the next group still owed in the fill's order, or nil once every group is answered.
-func (f *Fill) Next() *uint64 {
-	return f.inner.Next()
+// Wanted is the next group still owed in the fill's order, or nil once every
+// group is answered or no reader waits on the fill.
+func (f *Fill) Wanted() *uint64 {
+	return f.inner.Wanted()
+}
+
+// WantedChanged blocks until Wanted changes and returns it. The first call returns at once.
+func (f *Fill) WantedChanged(ctx context.Context) (*uint64, error) {
+	return f.inner.WantedChanged(ctx)
 }
 
 // Accept inserts group sequence and returns a producer for it. The first group
@@ -393,16 +399,6 @@ func (f *Fill) DropGroups(start, end uint64) {
 	f.inner.DropGroups(start, end)
 }
 
-// IsUsed reports whether a reader still waits on this fill.
-func (f *Fill) IsUsed() bool {
-	return f.inner.IsUsed()
-}
-
-// Unused blocks until no reader waits on this fill.
-func (f *Fill) Unused(ctx context.Context) error {
-	return f.inner.Unused(ctx)
-}
-
 // IsWanted reports whether a reader still waits for group sequence from this fill:
 // false once every reader took it or narrowed past it, or nobody waits on the fill.
 func (f *Fill) IsWanted(sequence uint64) bool {

js/e2ee/src/e2ee.test.ts

@@ -429,7 +429,7 @@ test("grouped failure wakes pending reads and releases demand", async () => {
 	const good = await recvGroup(p.consumer);
 	expect(await readText(good)).toBe("ok");
 	const forged = await recvGroup(p.consumer);
-	expect(badDemand.used.peek()).toBe(true);
+	expect(badDemand.wanted.peek()).toBe(true);
 
 	// Sibling frame, next group, and datagram reads have nothing to return until the failure.
 	const frame = good.readFrame();
@@ -442,8 +442,8 @@ test("grouped failure wakes pending reads and releases demand", async () => {
 	expect(await failure(datagram)).toBe("authentication");
 
 	expect(demand.wanted.peek()).toBeUndefined();
-	expect(badDemand.used.peek()).toBe(false);
-	expect(open.demand().used.peek()).toBe(false);
+	expect(badDemand.wanted.peek()).toBe(false);
+	expect(open.demand().wanted.peek()).toBe(false);
 	expect(p.net.closed.peek()).toBeUndefined();
 });
 

js/hang/src/container/consumer.test.ts

@@ -728,10 +728,10 @@ test("Consumer releases its open groups when an error ends it", async () => {
 	const later = track.createGroup(1);
 	later.writeFrame({ payload: new Uint8Array([0x01]), timestamp: Time.Timestamp.now() });
 	await open;
-	expect(later.demand().used.peek()).toBe(true);
+	expect(later.demand().wanted.peek()).toBe(true);
 
 	await expect(consumer.next()).rejects.toThrow("malformed");
-	expect(later.demand().used.peek()).toBe(false);
+	expect(later.demand().wanted.peek()).toBe(false);
 	consumer.close();
 });
 

js/json/src/stream/stream.test.ts

@@ -91,8 +91,8 @@ test("a second group is reported while the first is still open", async () => {
 
 	// Both mirrors are released. The read that lost the race would otherwise stay registered on the
 	// first group, keeping this consumer's subscription reachable after the caller drops it.
-	expect(first.demand().used.peek()).toBe(false);
-	expect(second.demand().used.peek()).toBe(false);
+	expect(first.demand().wanted.peek()).toBe(false);
+	expect(second.demand().wanted.peek()).toBe(false);
 
 	// Sticky: a later read must not report the rest of the first group as a whole log.
 	first.writeFrame({ payload: encode({ n: 2 }), timestamp: Time.Timestamp.now() });