From c3bacd9e04aa0bf6c70d83e9c77d67bbd55b2149 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 17:33:50 -0700 Subject: [PATCH 1/2] chore(quest): claim quest/m1/release-audit/go-reservation.md Quest-Claim: e44ca1a110b698a3 From 5773d5a3afbd479b10e7981deb1a6fa49e225bd1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 17:43:37 -0700 Subject: [PATCH 2/2] docs(go): define deterministic reservation release with Update(0) Co-Authored-By: GPT-6 --- doc/lib/go/index.md | 1 + go/wrapper/README.md | 7 ++ go/wrapper/bandwidth_test.go | 86 ++++++++++++++++++++++++ go/wrapper/session.go | 11 ++- quest/m1/release-audit/README.md | 1 - quest/m1/release-audit/go-reservation.md | 26 ------- rs/moq-net/src/model/bandwidth.rs | 35 ++++++++++ 7 files changed, 137 insertions(+), 30 deletions(-) create mode 100644 go/wrapper/bandwidth_test.go delete mode 100644 quest/m1/release-audit/go-reservation.md diff --git a/doc/lib/go/index.md b/doc/lib/go/index.md index 50534897fd..f930bb8c89 100644 --- a/doc/lib/go/index.md +++ b/doc/lib/go/index.md @@ -123,6 +123,7 @@ shows which module holds each. - **Context cancellation.** Every call that can block takes a `context.Context` first, and cancelling it tears down the native work, not just your wait. A one-shot call (`RequestBroadcast`, `FetchGroup`, `Server.Accept`) aborts alone; a stream read (`Next`, `ReadFrame`, and the iterators over them) cancels the stream it reads. - **Keep owners reachable.** An origin from `moq.NewOriginProducer` has no `Close`. It ends when the garbage collector reaches its last owner (each producer, published broadcast, and `OriginDynamic`), and its consumers then fail with `moq.ErrClosed`. +- **Bandwidth ownership.** An encoder you manage can reserve a share of the session's bandwidth. Reuse its reservation when the ceiling changes; before abandoning or replacing it, call `Update(0)` to return its share and automatic probe ceiling without waiting for garbage collection. The track can stay subscribed and the reservation can be reused. - **Audio needs cuts.** Video groups at its keyframes, but audio forms a group only where you call `Cut()`: after every frame, or at a segment cadence to align with video. - **Live encoder timing.** After writing a frame you encoded yourself, call `Flush(timestamp)` with the same timestamp so the catalog advertises your jitter. Skip it for file and network imports. On a seek or pause, call `Discontinuity()`, then keep timestamps moving forward and resume video on a keyframe. - **Decoded frames hold decoder buffers.** `Close` each `VideoDecodedFrame` promptly, or the decoder stalls. `Resize` is best effort, so read each frame's `Width()` and `Height()`. diff --git a/go/wrapper/README.md b/go/wrapper/README.md index 049e1af44e..6c8801d419 100644 --- a/go/wrapper/README.md +++ b/go/wrapper/README.md @@ -70,6 +70,13 @@ pinning a generated self-signed certificate. `Client.Session().Stats()` returns a connection stats snapshot. Fields are nil when the transport backend does not report that metric yet. +An encoder you manage can claim a share of `Client.Session().Bandwidth()` with +`Reserve(demand, maxBps)`, using the track's `Demand()`. Keep one reservation and change its ceiling +with `Update(maxBps)`. Call `Update(0)` before abandoning or replacing it to +return its share and automatic probe ceiling immediately, even while the track +stays subscribed. A later update can resume the same claim. While the track +stays open, dropping the Go reference alone leaves release to garbage collection. + Incoming server requests expose the query-free `Path()` before `Accept()`. It is consistent across transports and returns an empty string for the root or missing path. `Query()` returns the encoded query and may contain credentials. diff --git a/go/wrapper/bandwidth_test.go b/go/wrapper/bandwidth_test.go new file mode 100644 index 0000000000..a81b537de4 --- /dev/null +++ b/go/wrapper/bandwidth_test.go @@ -0,0 +1,86 @@ +package moq_test + +import ( + "context" + "math" + "runtime" + "testing" + + "moq.dev/moq" +) + +// Retaining the old handle models a replacement before collection. +// Update(0) must hand back its share even while both handles and subscribers live. +func TestReservationUpdateReleasesSubscribedTrack(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + relay := startRelay(t, ctx, "127.0.0.1:0") + defer relay.server.Close() + client, err := moq.Dial(ctx, "https://"+relay.addr, + moq.WithTLSVerify(false), moq.WithReconnect(false)) + if err != nil { + t.Fatal(err) + } + defer client.Close() + bandwidth := client.Session().Bandwidth() + // Grant assertions use a live QUIC estimate, but need no traffic or probing. + off := uint64(0) + bandwidth.SetTarget(&off) + broadcast, err := moq.NewBroadcastProducer() + if err != nil { + t.Fatal(err) + } + defer broadcast.Close() + + reserve := func(name string, priority int8, ceiling uint64) (*moq.TrackDemand, *moq.Reservation) { + t.Helper() + track, err := broadcast.PublishTrack(name, &moq.TrackInfo{Priority: priority}) + if err != nil { + t.Fatal(err) + } + consumer, err := track.Consume(nil) + if err != nil { + t.Fatal(err) + } + t.Cleanup(consumer.Cancel) + // TrackDemand is weak, so keep the producer for the whole test. + t.Cleanup(func() { runtime.KeepAlive(track) }) + demand, err := track.Demand() + if err != nil { + t.Fatal(err) + } + return demand, bandwidth.Reserve(demand, ceiling) + } + grant := func(reservation *moq.Reservation, want uint64) { + t.Helper() + got := reservation.Grant() + if got == nil { + t.Fatalf("grant unavailable, want %d", want) + } + if *got != want { + t.Fatalf("grant = %d, want %d", *got, want) + } + } + + demand, retired := reserve("encoder", 1, math.MaxUint64) + _, sibling := reserve("sibling", 0, 1) + replacement := bandwidth.Reserve(demand, 1) + // Even a retained predecessor must relinquish the allocation explicitly. + // Priority makes this independent of the live estimate's exact value. + grant(sibling, 0) + retired.Update(0) + grant(retired, 0) + grant(replacement, 1) + grant(sibling, 1) + + // Zero is reversible and repeated release is harmless. + retired.Update(math.MaxUint64) + grant(sibling, 0) + retired.Update(0) + retired.Update(0) + grant(sibling, 1) + if demand.Wanted() == nil { + t.Fatal("release ended the track subscription") + } + runtime.KeepAlive(retired) +} diff --git a/go/wrapper/session.go b/go/wrapper/session.go index bf5f2f5737..cd810498c7 100644 --- a/go/wrapper/session.go +++ b/go/wrapper/session.go @@ -144,8 +144,9 @@ type Bandwidth struct { // Reserve claims up to maxBps for the track demand watches; every track // producer hands out its Demand. maxBps is a ceiling, not a measurement: -// reserve the most the track can ever send. Drop the reservation to hand the -// room back. +// reserve the most the track can ever send. Reuse Reservation.Update to change +// the ceiling, or call Update(0) before abandoning or replacing the reservation +// to hand the room back without waiting for garbage collection. func (b *Bandwidth) Reserve(demand *TrackDemand, maxBps uint64) *Reservation { return &Reservation{inner: b.inner.Reserve(demand.inner, maxBps)} } @@ -162,6 +163,8 @@ func (b *Bandwidth) SetTarget(targetBps *uint64) { // // Grant is a snapshot: nil means no estimate or no demand, so hold the current // rate, and 0 is a real zero grant. +// Update(0) relinquishes the allocation while the track stays subscribed; the +// handle remains usable and a later Update can claim bandwidth again. type Reservation struct { inner *ffi.MoqReservation } @@ -171,7 +174,9 @@ func (r *Reservation) Grant() *uint64 { return r.inner.Grant() } -// Update changes the ceiling, keeping the same claim. +// Update changes the ceiling, keeping the same claim. Zero immediately gives +// its share to siblings and removes its contribution to the automatic probe +// target; it does not close the track or destroy the native handle. func (r *Reservation) Update(maxBps uint64) { r.inner.Update(maxBps) } diff --git a/quest/m1/release-audit/README.md b/quest/m1/release-audit/README.md index 7a5bccd119..ebbbacbb6d 100644 --- a/quest/m1/release-audit/README.md +++ b/quest/m1/release-audit/README.md @@ -111,7 +111,6 @@ child records whether its behavior is new on main or inherited from release. - [Release HLS server state](/quest/m1/release-audit/hls-shutdown.md) - final server drop retires its watchers while publishers remain live - [Retire expiry metadata](/quest/m1/release-audit/expiry-index.md) - empty deadline entries do not accumulate with historical read budgets - [Preserve IETF timestamps](/quest/m1/release-audit/ietf-timestamps.md) - JS encoding uses exact same-scale conversion and rejects unsupported values before sending -- [Settle Go reservation release](/quest/m1/release-audit/go-reservation.md) - investigate prompt allocation release through the public wrapper - [Settle finite HLS import](/quest/m1/release-audit/hls-finite.md) - initially ended playlists have an explicit and tested start position - [Settle HLS timestamp resets](/quest/m1/release-audit/hls-discontinuity.md) - unchanged-map discontinuities preserve timeline identity or are explicitly refused diff --git a/quest/m1/release-audit/go-reservation.md b/quest/m1/release-audit/go-reservation.md deleted file mode 100644 index 6d88ac3a77..0000000000 --- a/quest/m1/release-audit/go-reservation.md +++ /dev/null @@ -1,26 +0,0 @@ -# [S] Settle deterministic Go reservation release - -## Goal - -A Go encoder can relinquish a bandwidth allocation predictably while its -track remains subscribed, with an explicit and tested ownership contract. - -## Plan - -P2 investigation, not a proven leak or measured performance regression. -At main 1d43c096e0, `go/wrapper/session.go` says dropping a Reservation hands -room back, but exposes only Grant/Update and hides the generated handle. -Native allocation release depends on handle Drop. Go collection does not -guarantee prompt finalization. An encoder replacing a reservation on a still -subscribed track may therefore retain its old allocation and probe ceiling -until collection. This wrapper shape is inherited; main expands reservation -consumers and uses their ceilings for probe targets. - -First reproduce replacement through the public Go wrapper and observe sibling -shares and probe targets without relying on forced GC. Closed tracks are -pruned, and Update(0) can relinquish the ceiling: this is not a claim of a -permanent native leak. Decide whether the existing operation supplies the -intended release contract and needs documentation, or explicit handle -disposal is required. Prefer the existing mechanism if it fully solves the -consumer's lifetime. Verify the chosen contract deterministically and report -any Go API change; no wire change is expected. diff --git a/rs/moq-net/src/model/bandwidth.rs b/rs/moq-net/src/model/bandwidth.rs index f8a880a61e..aec786c617 100644 --- a/rs/moq-net/src/model/bandwidth.rs +++ b/rs/moq-net/src/model/bandwidth.rs @@ -1135,6 +1135,41 @@ mod tests { assert_eq!(large_reserved.peek(), Some(bps(5_000_000))); } + /// Collected-language callers can relinquish a ceiling before their retained + /// handle is finalized, without ending the track or waiting for its subscribers. + #[test] + fn zero_ceiling_releases_share_and_probe_target() { + let estimate = Producer::new(); + let allocator = Allocator::new(estimate.consume()); + estimate.set(Some(bps(3_000_000))).unwrap(); + let mut target = allocator.target(); + let waiter = kio::Waiter::noop(); + + let (_encoder_broadcast, encoder) = track(VIDEO); + let _encoder_sub = encoder.subscribe(None); + let retired = allocator.reserve(&encoder.demand(), bps(4_000_000)); + let replacement = allocator.reserve(&encoder.demand(), bps(4_000_000)); + let (_sibling_broadcast, sibling) = track(VIDEO); + let _sibling_sub = sibling.subscribe(None); + let sibling = allocator.reserve(&sibling.demand(), bps(4_000_000)); + assert_eq!(sibling.peek(), Some(bps(1_000_000))); + assert_eq!(target.poll_changed(&waiter), Poll::Ready(bps(12_000_000))); + + retired.update(Rate::ZERO); + assert_eq!(retired.peek(), Some(Rate::ZERO)); + assert_eq!(replacement.peek(), Some(bps(1_500_000))); + assert_eq!(sibling.peek(), Some(bps(1_500_000))); + assert_eq!(target.poll_changed(&waiter), Poll::Ready(bps(8_000_000))); + retired.update(Rate::ZERO); + assert!(target.poll_changed(&waiter).is_pending()); + + // The registration remains usable, so an encoder may resume in place. + retired.update(bps(4_000_000)); + assert_eq!(sibling.peek(), Some(bps(1_000_000))); + assert_eq!(target.poll_changed(&waiter), Poll::Ready(bps(12_000_000))); + assert!(encoder.demand().is_used()); + } + /// A reader parked on `changed` has to be woken by its own reservation moving, not /// just by the estimate or a sibling: the encoder is sitting in a `select!` on it. #[moq_clock::test]