diff --git a/js/net/src/group.ts b/js/net/src/group.ts index 759da126163..7196b38418b 100644 --- a/js/net/src/group.ts +++ b/js/net/src/group.ts @@ -770,13 +770,18 @@ class Expired implements GetPromise { } } -// A consumer can finish from its own expiry verdict or from the shared group state. Keep -// one stable public handle while letting peek() observe either source synchronously, before -// signal subscribers run in the next microtask. The source is re-read on every check, since a +// A consumer can finish from its own verdict or from the shared group state. Keep one stable +// public handle while letting peek() observe either source synchronously, before signal +// subscribers run in the next microtask. The source is re-read on every check, since a // recovered group moves to another copy's state, and `final` says whether its value ends the -// consumer: a failure recovery can still continue does not. +// consumer: a failure recovery can still continue does not, so it never reaches a caller. Built +// on first read, which only a caller holding the group makes: a reader attaches its recovery +// (`hooks.recoverGroup`) before handing the group out, and the track's own scans read +// `hooks.groupFailed` instead. A front serving another front is the exception: the inner front +// holds the handle before the outer attaches its recovery, so attaching rebuilds a handle that +// latched a failure, and the inner front alone keeps the failure it saw. class CombinedClosed implements GetPromise { - #preferred: Once; + #preferred: Once; #source: () => Once; #final: (value: Error | null) => boolean; #value = new Once(); @@ -785,11 +790,7 @@ class CombinedClosed implements GetPromise { #unwatch?: Dispose; #dispose: Dispose[] = []; - constructor( - preferred: Once, - source: () => Once, - final: (value: Error | null) => boolean, - ) { + constructor(preferred: Once, source: () => Once, final: (value: Error | null) => boolean) { this.#preferred = preferred; this.#source = source; this.#final = final; @@ -875,9 +876,9 @@ export class Consumer { #attempt?: Attempt; // The frame a replacement copy resumes at: frames below it were read from an earlier copy. #skip = 0; - // Sticky failure, set once, including a budget that gave up on the group. - #terminal?: Error; - #verdict = new Once(); + // This reader's own failure, set once, including a budget that gave up on the group: it + // ends the consumer whatever its copy does. See `#fail`. + #verdict = new Once(); #closed?: CombinedClosed; #expired?: Expired; // Closed by its reader, which ends it whatever is left unread. @@ -935,18 +936,27 @@ export class Consumer { this.#expired ??= new Expired( () => { this.#expire(); - return this.#terminal instanceof StreamError && this.#terminal.code === StreamCode.Old - ? this.#terminal - : undefined; + const verdict = this.#verdict.peek(); + return verdict instanceof StreamError && verdict.code === StreamCode.Old ? verdict : undefined; }, () => this.#expiry?.changed ?? [], ); return this.#expired; } + // The failure that ends this reader, as `closed` would settle with it, without building that + // handle: a track's scans read this, so none latches a failure before a reader attaches the + // recovery that continues past it. + #failed(): Error | undefined { + const verdict = this.#verdict.peek(); + if (verdict) return verdict; + const closed = this.#state.closed.peek(); + return closed instanceof Error && !this.#recovers(closed) ? closed : undefined; + } + // Whether a failure of the copy being read may still be continued from another route. #recovers(failed: Error): boolean { - return !this.#terminal && !this.#torn() && this.#recover?.wants(failed) === true; + return !this.#verdict.peek() && !this.#torn() && this.#recover?.wants(failed) === true; } // Whether a frame handed out to stream was cut short. @@ -984,10 +994,11 @@ export class Consumer { hooks.expireGroup = (group, expiry) => { group.#expiry = expiry; }; + hooks.groupFailed = (group) => group.#failed(); hooks.recoverGroup = (group, recover) => { group.#recover = recover; - // A track's drift scan may have read `closed` already, latching a failure recovery can - // now continue: derive it again with the recovery in view. + // A front serving another front attaches its recovery after the inner one handed the + // group out, whose reads may have latched a failure this recovery continues past. if (group.#closed?.peek() instanceof Error && group.#verdict.peek() === undefined) group.#closed = undefined; }; @@ -1007,9 +1018,10 @@ export class Consumer { // reader operation: its verdict and replacement copy, not a raw producer end, decide it. async #end(until?: number): Promise<{ end: number; complete: boolean }> { for (;;) { - if (this.#terminal) { + const verdict = this.#verdict.peek(); + if (verdict) { this.#abandon(); - throw this.#terminal; + throw verdict; } const closed = this.#state.closed.peek(); const count = this.frameCount; @@ -1046,17 +1058,25 @@ export class Consumer { // A reader that took every frame still fails with Old: the producer never finished // the group, so a clean end would pass the prefix off as the whole group. #expire(): boolean { - if (this.#terminal) return false; + if (this.#verdict.peek()) return false; if (this.#state.closed.peek() instanceof Error) return false; if (!this.#expiry?.expired()) return false; - this.#terminal = new StreamError(StreamCode.Old, { - message: "group exceeded the subscription max delay budget", - }); - if (this.#verdict.peek() === undefined) this.#verdict.set(this.#terminal); + this.#fail( + new StreamError(StreamCode.Old, { + message: "group exceeded the subscription max delay budget", + }), + ); return true; } + // Fail this reader with `error`, unless it already failed: the first verdict stands. + #fail(error: unknown): Error { + if (this.#verdict.peek() === undefined) + this.#verdict.set(error instanceof Error ? error : new Error(String(error))); + return this.#verdict.peek() as Error; + } + #evict(lost: Error): boolean { // A group read on from another copy is that copy's to reclaim. if (this.#state !== this.#mirrored) return false; @@ -1066,10 +1086,7 @@ export class Consumer { // on this group that is the more specific reason, so ask it first; otherwise // the cache reclaimed it before the reader got to it. this.#expire(); - if (!this.#terminal) { - this.#terminal = lost; - if (this.#verdict.peek() === undefined) this.#verdict.set(this.#terminal); - } + this.#fail(lost); } if (frames.length > 0) { this.#state.start += frames.length; @@ -1112,8 +1129,7 @@ export class Consumer { (err: unknown) => { started.settled = true; if (abort.signal.aborted) return; - this.#terminal ??= err instanceof Error ? err : new Error(String(err)); - if (this.#verdict.peek() === undefined) this.#verdict.set(this.#terminal); + this.#fail(err); }, ); attempt = started; @@ -1161,8 +1177,7 @@ export class Consumer { const index = this.#index(); if (next.start > index) { replacement.discard(); - this.#terminal ??= new Error(`group ${this.sequence} cannot resume at frame ${index}`); - if (this.#verdict.peek() === undefined) this.#verdict.set(this.#terminal); + this.#fail(new Error(`group ${this.sequence} cannot resume at frame ${index}`)); return; } replacement.adopt(); @@ -1229,10 +1244,9 @@ export class Consumer { arrived: buffered.arrived, }; } catch (error) { - this.#terminal = error instanceof Error ? error : new Error(String(error)); - if (this.#verdict.peek() === undefined) this.#verdict.set(this.#terminal); + const verdict = this.#fail(error); this.#closed?.refresh(); - throw this.#terminal; + throw verdict; } if (frames.length === 0 && this.#state.closed.peek() !== undefined) this.#state.flate = undefined; const sequence = this.#state.start++; @@ -1244,13 +1258,14 @@ export class Consumer { /** True once no further frames can be read: this handle or the group has closed, and every buffered frame is read. */ get done(): boolean { return ( - this.#terminal !== undefined || (this.#bufferedFrames().length === 0 && (this.#released || this.#settled())) + this.#verdict.peek() !== undefined || + (this.#bufferedFrames().length === 0 && (this.#released || this.#settled())) ); } /** True once this handle or the group has been closed, regardless of whether buffered frames remain unread. Synchronous complement to the {@link closed} promise. */ get isClosed(): boolean { - return this.#terminal !== undefined || this.#released || this.#settled(); + return this.#verdict.peek() !== undefined || this.#released || this.#settled(); } // Whether the copy being read ended, for good: a failure recovery may continue is not an end. @@ -1277,13 +1292,13 @@ export class Consumer { * end-of-group: check {@link done} to tell "no frame buffered yet" from "finished". */ tryReadFrame(): Frame | undefined { - if (this.#terminal) return undefined; + if (this.#verdict.peek()) return undefined; return this.#readBufferedFrame()?.frame; } /** Like {@link tryReadFrame} but also reports the frame's sequence number within the group. */ tryReadFrameSequence(): ({ sequence: number } & Frame) | undefined { - if (this.#terminal) return undefined; + if (this.#verdict.peek()) return undefined; const read = this.#readBufferedFrame(); if (!read) return undefined; return { sequence: read.sequence, ...read.frame }; @@ -1292,7 +1307,7 @@ export class Consumer { /** Resolves once {@link readFrame} would not block. */ async readable(): Promise { for (;;) { - if (this.#terminal) { + if (this.#verdict.peek()) { this.#abandon(); return; } @@ -1343,9 +1358,10 @@ export class Consumer { async #readFrame(from: number, compressed = false, streaming = false): Promise { for (;;) { - if (this.#terminal) { + const verdict = this.#verdict.peek(); + if (verdict) { this.#abandon(); - throw this.#terminal; + throw verdict; } // A whole read streams a frame still being written into its own buffer, rather @@ -1391,7 +1407,8 @@ export class Consumer { let size = 0; try { for (;;) { - if (this.#terminal) throw this.#terminal; + const verdict = this.#verdict.peek(); + if (verdict) throw verdict; const chunk = pollChunk(chunks); if (chunk === undefined) break; if (chunk !== WAIT) { @@ -1412,10 +1429,9 @@ export class Consumer { const payload = parts.length === 0 ? EMPTY : joined(parts, size); return { sequence, frame: { payload, timestamp: chunks.timestamp, arrived: chunks.arrived } }; } catch (error) { - this.#terminal ??= error instanceof Error ? error : new Error(String(error)); - if (this.#verdict.peek() === undefined) this.#verdict.set(this.#terminal); + const verdict = this.#fail(error); this.#closed?.refresh(); - throw this.#terminal; + throw verdict; } } @@ -1453,8 +1469,7 @@ export class Consumer { #release(abort: Error | undefined): void { // A copy failure recovery was continuing past ends with the abort instead. if (abort && this.#recover && this.#state.closed.peek() instanceof Error) { - this.#terminal ??= abort; - if (this.#verdict.peek() === undefined) this.#verdict.set(this.#terminal); + this.#fail(abort); } this.#abandon(); this.#recover?.close?.(); diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index 7e76bf956a7..ae144d4f70f 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -476,7 +476,7 @@ export interface StagedTrack { /** * How a received copy's start bounds the fills its readers ask below it. Mirrors the Rust - * `TrackState::start_resolved` and `fills_from_start`. + * `Start::Resolved` and `Producer::fills_from_start`. * * Below a start the publisher resolved from the budget (`resolvesStart`), a reader fills the * history its own budget reaches: an update can cross the response, so nobody knows which @@ -560,8 +560,9 @@ export const hooks: { */ supersedeStale: (producer: Producer, judged: readonly Run[] | undefined) => Run[]; /** - * Mark a track as fed by a wire subscription, which owes missing groups. `starts` says how - * that wire's start bounds the fills its readers ask below it. Assigned by `track.ts`. + * Mark a track as fed by a wire subscription, which owes missing groups, until + * {@link cacheTrack} ends it. `starts` says how that wire's start bounds the fills its readers + * ask below it. Assigned by `track.ts`. */ receiveTrack: (producer: Producer, starts?: FillStarts) => void; /** Create a received group at its wire offset before publishing it to readers. Assigned by `track.ts`. */ @@ -588,7 +589,8 @@ export const hooks: { * IETF SUBSCRIBE_OK's Largest) for the floor its SUBSCRIBE requested: while the demand keeps * that floor, a pending tail owes nothing below it. Mirrors the Rust `start_at`; assigned by * `track.ts`. A null `start` resolves an empty range; without `start`, the SUBSCRIBE still - * waits on its START, as the Rust `request_start`. + * waits on its START, as the Rust `request_start`. Only after {@link receiveTrack}: the start + * is that subscription's, and ends with it. */ startTrack: (producer: Producer, served: { start?: bigint | null; requested?: bigint }) => void; /** @@ -826,7 +828,12 @@ export const hooks: { arriveGroup: (group: GroupConsumer, at: Instant) => void; /** Keep applying a subscription's drift policy after it hands a group out. */ expireGroup: (group: GroupConsumer, expiry: Expiry) => void; - /** Carry a handed-out group across its logical track's route changes. */ + /** The failure that ends a group's reader, as its `closed` would settle, without building that handle. */ + groupFailed: (group: GroupConsumer) => Error | undefined; + /** + * Carry a group across its logical track's route changes. Attached before the group is handed + * out, so its `closed` never settles on a failure the recovery continues past. + */ recoverGroup: (group: GroupConsumer, recover: GroupRecover) => void; /** Read a payload chunk, releasing its wait when the output stream closes. */ readFrameChunk: (frame: FrameConsumer, closed: Promise) => Promise; @@ -1162,6 +1169,9 @@ export const hooks: { expireGroup: () => { throw new Error("group.ts not loaded"); }, + groupFailed: () => { + throw new Error("group.ts not loaded"); + }, recoverGroup: () => { throw new Error("group.ts not loaded"); }, diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index 88ce011cb7b..4de694fa007 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -1704,6 +1704,7 @@ for (const version of [Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07]) { test(`lite ${version}: descending delivery resolves source start ${source} at ${expected}`, async () => { const sub = await servedSubscription({ version, startGroup: 2n }); try { + hooks.receiveTrack(sub.track); hooks.startTrack(sub.track, { start: source, requested: 2n }); sub.serve(6n); expect(await sub.servedSequence()).toBe(6n); @@ -1727,6 +1728,7 @@ for (const version of [Version.DRAFT_06, Version.DRAFT_07]) { test(`lite ${version}: START waits for the upstream START before resolving`, async () => { const sub = await servedSubscription({ version, startGroup: 2n }); try { + hooks.receiveTrack(sub.track); hooks.startTrack(sub.track, { requested: 2n }); sub.serve(6n); await flush(); @@ -1744,6 +1746,7 @@ for (const version of [Version.DRAFT_06, Version.DRAFT_07]) { test(`lite ${version}: a closed source releases the START it was waiting for`, async () => { const sub = await servedSubscription({ version, startGroup: 2n }); try { + hooks.receiveTrack(sub.track); hooks.startTrack(sub.track, { requested: 2n }); sub.serve(6n); await flush(); @@ -1766,6 +1769,7 @@ for (const skipped of ["failed", "stale"] as const) { maxAge: skipped === "stale" ? Milli(0) : undefined, }); try { + hooks.receiveTrack(sub.track); hooks.startTrack(sub.track, { start: 5n, requested: 5n }); const group = sub.track.createGroup(5n); group.writeFrame(textFrame("old", Timestamp.fromMillis(5))); diff --git a/js/net/src/resume.test.ts b/js/net/src/resume.test.ts index f3b2b5e1b9b..8948611f37b 100644 --- a/js/net/src/resume.test.ts +++ b/js/net/src/resume.test.ts @@ -326,6 +326,8 @@ test("a routed source start follows the serving copy and its subscriber floor", const b = new BroadcastProducer(); const ta = copy(a); const tb = copy(b); + hooks.receiveTrack(ta.producer); + hooks.receiveTrack(tb.producer); hooks.startTrack(ta.producer, { start: 1n }); hooks.startTrack(tb.producer, { start: 5n, requested: 3n }); const front = new Front(); @@ -2895,14 +2897,12 @@ test("an end a replaced copy declared holds across a swap", async () => { expect(hooks.subscriberEnd(sub)).toEqual({ group: 2n, frame: 0 }); expect(await sub.finished()).toEqual({ group: 2n, frame: 0 }); - // A copy declaring another end claims different content. - tb.producer.finishAt({ group: 3n, frame: 0 }); + // A copy declaring another end leaves the first one, and its error with it. + tb.producer.finishAt({ group: 3n, frame: 0 }, new StreamError(StreamCode(70), { message: "other" })); await settle(); - expect(sub.closed.peek()).toBeInstanceOf(Error); + expect(sub.closed.peek()).not.toBeInstanceOf(Error); expect(hooks.subscriberEnd(sub)).toEqual({ group: 2n, frame: 0 }); - // Refused, the reader holds no copy. - expect(ta.producer.demand().wanted.peek()).toBeUndefined(); - expect(tb.producer.demand().wanted.peek()).toBeUndefined(); + expect(hooks.subscriberEndError(sub)).toBeUndefined(); sub.close(); front.consumer.close(); @@ -3154,7 +3154,9 @@ test("a clean end keeps the serving copy while a group read from a replaced one } }); -test("a retained copy declaring another end after a clean end fails the reads", async () => { +// Once the track ended cleanly, a copy kept for a group it may still deliver can claim another +// end without failing the reader: `closed` and the reads agree. +test("a retained copy declaring another end after a clean end fails nothing", async () => { const front = new Front(); const a = new BroadcastProducer(); const b = new BroadcastProducer(); @@ -3178,11 +3180,14 @@ test("a retained copy declaring another end after a clean end fails the reads", await settle(); expect(sub.closed.peek()).toBeNull(); - // The copy still holding group 1 claims another end: its content is refused. + // The copy still holding group 1 claims another end. ta.producer.finishAt({ group: 5n, frame: 0 }); + ta.write(4n); await settle(); expect(hooks.subscriberEnd(sub)).toEqual({ group: 3n, frame: 0 }); - await expect(sub.recvGroup()).rejects.toThrow("changed its properties"); + expect((await sub.recvGroup())?.sequence).toBe(1n); + expect(await sub.recvGroup()).toBeUndefined(); + expect(sub.closed.peek()).toBeNull(); } finally { sub.close(); front.consumer.close(); @@ -3228,7 +3233,9 @@ test("a clean end keeps the serving copy while a group handed out before the swa } }); -test.each(["handed out", "buffered"])("an end below a group another copy %s is refused", async (seen) => { +// A copy contradicting another's end breaks the track's identity, which hurts only this track: +// the first end stands and the reader never fails over it. +test.each(["handed out", "buffered"])("an end below a group another copy %s leaves the first end", async (seen) => { const front = new Front(); const a = new BroadcastProducer(); const b = new BroadcastProducer(); @@ -3238,6 +3245,7 @@ test.each(["handed out", "buffered"])("an end below a group another copy %s is r const sub = front.consumer.track("seq").subscribe(); try { ta.write(5n); + ta.producer.finishAt({ group: 6n, frame: 0 }); if (seen === "handed out") expect(drain(sub)).toEqual([5n]); front.serve(b.consume()); await settle(); @@ -3245,36 +3253,9 @@ test.each(["handed out", "buffered"])("an end below a group another copy %s is r // The new route ends the track before a group the old one has. tb.producer.finishAt({ group: 3n, frame: 0 }); await settle(); - expect(() => sub.tryRecvGroup()).toThrow("changed its properties"); - expect(sub.closed.peek()).toBeInstanceOf(Error); - } finally { - sub.close(); - front.consumer.close(); - a.close(); - b.close(); - } -}); - -test("a group past the declared end on another copy is refused", async () => { - const front = new Front(); - const a = new BroadcastProducer(); - const b = new BroadcastProducer(); - const ta = copy(a); - const tb = copy(b); - front.serve(a.consume()); - const sub = front.consumer.track("seq").subscribe(); - try { - ta.write(0n); - ta.producer.finishAt({ group: 2n, frame: 0 }); - expect(drain(sub)).toEqual([0n]); - front.serve(b.consume()); - await settle(); - - // The new route has a group the end says never exists. - tb.write(3n); - await settle(); - expect(() => sub.tryRecvGroup()).toThrow("changed its properties"); - expect(sub.closed.peek()).toBeInstanceOf(Error); + expect(drain(sub)).toEqual(seen === "handed out" ? [] : [5n]); + expect(await sub.finished()).toEqual({ group: 6n, frame: 0 }); + expect(sub.closed.peek()).not.toBeInstanceOf(Error); } finally { sub.close(); front.consumer.close(); @@ -3670,3 +3651,34 @@ test("routed metadata receives a refusal without polling a group cursor", async installed.close(); } }); + +test("a group an inner front hands out already failed still continues on the outer front's next route", async () => { + const inner = new Front(); + const outer = new Front(); + const a = new BroadcastProducer(); + const b = new BroadcastProducer(); + const ta = copy(a); + const tb = copy(b); + inner.serve(a.consume()); + outer.serve(inner.consumer); + const sub = outer.consumer.track("seq").subscribe(); + const failed = ta.write(0n, 1, false); + // The outer front switches routes before the reader takes the group: the inner front has + // no other route, so only the outer one may continue it. + outer.serve(b.consume()); + failed.abort(new StreamError(StreamCode(70), { message: "cut" })); + await settle(); + const group = sub.tryRecvGroup(); + if (!group) throw new Error("expected group 0"); + expect(dec.decode((await group.readFrame())?.payload)).toBe("0.0"); + const read = group.readFrame(); + tb.write(0n, 2); + expect(dec.decode((await read)?.payload)).toBe("0.1"); + expect(await group.closed).toBeNull(); + + sub.close(); + outer.consumer.close(); + inner.consumer.close(); + a.close(); + b.close(); +}); diff --git a/js/net/src/resume.ts b/js/net/src/resume.ts index 54dc9d9289d..a93a52a81b5 100644 --- a/js/net/src/resume.ts +++ b/js/net/src/resume.ts @@ -39,8 +39,9 @@ * The logical track ends cleanly once its serving copy does and no group the reader lacks may * still land below that end on a copy it replaced. Its `closed` settles and the reader leaves the * route, keeping each copy only while it holds something left to read, as a direct reader keeps - * its buffer. Its end is the first one any copy declared, which every later copy has to match, and - * no copy may hold a group at or past it. + * its buffer. Its end is the first one any copy declared, which a later mid-group end may refine; + * a copy declaring another breaks the track's identity, which hurts only its own broadcast, so + * nothing checks that copies agree. * * @module */ @@ -388,13 +389,16 @@ class Reader implements ResumeSource { #changed = new Signal(0); #dispose: Dispose[] = []; #closed = false; - /** A copy declared another end: reads fail with this. */ - #refused?: Error; - /** The logical track ended cleanly: the reader drains what its copies hold and follows no route. */ - #ended = false; - /** The serving copy's largest when the track ended, kept once that copy goes. */ - #last?: { largest?: track.Position; object?: track.Position; ended: true }; - /** The first end any copy declared: the logical track's, whichever copy serves. */ + /** + * Set once the logical track ended cleanly: the reader drains what its copies hold and follows + * no route. Holds the serving copy's largest then, kept once that copy goes. + */ + #ended?: { largest?: track.Position; object?: track.Position; ended: true }; + /** + * The first end any copy declared: the logical track's, whichever copy serves. A copy declaring + * another breaks the track's identity, which hurts only its own broadcast, so nothing checks that + * copies agree. + */ #declared?: track.Position; // The error a copy's end carried, kept with the boundary so releasing that copy keeps it. #declaredError?: Error; @@ -786,9 +790,8 @@ class Reader implements ResumeSource { tryGroup(ordered: boolean, delivered: Delivered, dispositions: boolean): Recv { this.#delivered = delivered; this.sync(); - // Before anything is handed out, so a copy claiming other content hands out none of it. this.#learnEnd(); - if (this.#closed) return this.#refused ? { kind: "error", error: this.#refused } : { kind: "done" }; + if (this.#closed) return { kind: "done" }; this.#groupsRead = true; this.#backfillGaps(delivered); if (dispositions) { @@ -1071,8 +1074,7 @@ class Reader implements ResumeSource { const serving = this.#serving(); if (serving) this.#boundAtEnd(serving); if (this.#letGo(this.#delivered) || this.#closed) return false; - this.#ended = true; - this.#last = { + this.#ended = { largest: serving && hooks.peekLargest(serving.sub).largest, object: serving && hooks.objectLargest(serving.sub), ended: true, @@ -1246,7 +1248,7 @@ class Reader implements ResumeSource { /** The logical track ended cleanly, so no route continues it. */ get ended(): boolean { - return this.#ended; + return this.#ended !== undefined; } /** A group handed out ended or moved copies: let go of what nothing needs any more. */ @@ -1293,57 +1295,36 @@ class Reader implements ResumeSource { return last?.generation === this.#generation ? last : undefined; } - // The logical track's end, learning it from any copy that declared one. A group - // at or past it, on any copy or already handed out, claims other content: a direct track - // refuses an end below what it received. + // The logical track's end, learning it from any copy that declared one. #learnEnd(): track.Position | undefined { for (const copy of this.#copies) { const end = hooks.subscriberEnd(copy.sub); if (!end) continue; this.#declare(end); - this.#declaredError ??= hooks.subscriberEndError(copy.sub); - } - const declared = this.#declared; - if (declared && !this.#closed) { - const final = finalSequence(declared); - const seen = (sequence: bigint | undefined) => sequence !== undefined && sequence >= final; - if ( - seen(this.#newest) || - this.#copies.some( - (copy) => seen(hooks.latestSequence(copy.sub)) || seen(hooks.peekLargest(copy.sub).largest?.group), - ) - ) - this.#refuse(); + // Only a copy at the end that stands says how the track ended there. + const declared = this.#declared; + const same = + declared !== undefined && + ((declared.group === end.group && declared.frame === end.frame) || + (end.frame === 0 && finalSequence(end) === finalSequence(declared))); + if (same) this.#declaredError ??= hooks.subscriberEndError(copy.sub); } return this.#declared; } - // Record `end` as the logical track's end. One logical track has one end, as it has one set - // of properties: a copy declaring another is refused. A mid-group end refines the group - // boundary it falls just before, as a wire end may. + // Record `end` as the logical track's end, unless a copy declared one first. A mid-group end + // refines the group boundary it falls just before, as a wire end may. #declare(end: track.Position): void { if (this.#closed) return; const declared = this.#declared; if (!declared) { this.#declared = end; this.wake(); - } else if (declared.group === end.group && declared.frame === end.frame) { - return; - } else if (finalSequence(declared) === finalSequence(end) && (declared.frame === 0 || end.frame === 0)) { - if (declared.frame === 0) this.#declared = end; - } else { - this.#refuse(); + } else if (declared.frame === 0 && end.frame !== 0 && finalSequence(declared) === finalSequence(end)) { + this.#declared = end; } } - // Stop reading the copies, which claim other content. `closed` settles with the refusal, - // unless it already settled cleanly. - #refuse(): void { - this.#refused = new Incompatible(this.route.name); - this.#shell.close(this.#refused); - this.close(this.#refused); - } - start(): bigint | "pending" | undefined { this.sync(); const serving = this.#serving(); @@ -1445,20 +1426,20 @@ class Reader implements ResumeSource { objectLargest(): track.Position | undefined { this.sync(); const serving = this.#serving(); - return serving ? hooks.objectLargest(serving.sub) : this.#last?.object; + return serving ? hooks.objectLargest(serving.sub) : this.#ended?.object; } peekLargest(): { largest?: track.Position; ended: boolean } { this.sync(); const serving = this.#serving(); - return serving ? hooks.peekLargest(serving.sub) : (this.#last ?? { ended: false }); + return serving ? hooks.peekLargest(serving.sub) : (this.#ended ?? { ended: false }); } async largest(seen: track.Position | undefined): Promise { for (;;) { this.sync(); const serving = this.#serving(); - const view = serving ? hooks.peekLargest(serving.sub) : this.#last; + const view = serving ? hooks.peekLargest(serving.sub) : this.#ended; const largest = view?.largest; const { start, end } = this.#shell.cursor.peek(); // A replacement copy can start below what an earlier one reported. diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index abc467d4437..f2cda57b339 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -3971,6 +3971,7 @@ async function settled(read: Promise): Promise { test("a received end behind a hole settles only once the hole fills", async () => { const producer = new TrackProducer("test"); const reader = producer.subscribe({ delay: { max: Milli(60_000) } }); + hooks.receiveTrack(producer); hooks.startTrack(producer, { start: 0n }); const late = producer.createGroup(2n); late.finish(); @@ -4020,6 +4021,46 @@ test("a pending tail ends readers at the boundary once the track closes", async expect(await reader.closed).toBeNull(); }); +// A FETCH_OK's end of track on a copy no subscription feeds: nothing is in flight below it, so a +// reader above the groups another reader's floor still lacks ends at the boundary. +test("an end no subscription feeds owes no tail", async () => { + const producer = new TrackProducer("test"); + const history = producer.subscribe({ ranges: [{ start: { absolute: 0n } }], delay: { max: Milli(60_000) } }); + const group = producer.createGroup(2n); + group.finish(); + hooks.endTrack(producer, { group: 3n, frame: 0 }); + const live = producer.subscribe({ ranges: [{ start: { absolute: 2n } }], delay: { max: Milli(60_000) } }); + expect((await live.recvGroup())?.sequence).toBe(2n); + const next = live.recvGroup(); + expect(await settled(next)).toBe(true); + expect(await next).toBeUndefined(); + history.close(); + live.close(); + producer.close(); +}); + +// A group still in flight when the subscription ended may refine the end, but no subscription +// feeds the tail `cacheTrack` cleared anymore. +test("an end refined after the subscription ended owes no tail", async () => { + const producer = new TrackProducer("test"); + hooks.receiveTrack(producer); + const history = producer.subscribe({ ranges: [{ start: { absolute: 0n } }], delay: { max: Milli(60_000) } }); + const group = producer.createGroup(2n); + group.writeFrame({ payload: new Uint8Array([1]) }); + group.finish(); + hooks.endTrack(producer, { group: 3n, frame: 0 }); + hooks.cacheTrack(producer); + hooks.endTrack(producer, { group: 2n, frame: 1 }); + const live = producer.subscribe({ ranges: [{ start: { absolute: 2n } }], delay: { max: Milli(60_000) } }); + expect((await live.recvGroup())?.sequence).toBe(2n); + const next = live.recvGroup(); + expect(await settled(next)).toBe(true); + expect(await next).toBeUndefined(); + history.close(); + live.close(); + producer.close(); +}); + test("without a declared start, the first arrival stands in for it", async () => { const producer = new TrackProducer("test"); const reader = producer.subscribe({ delay: { max: Milli(60_000) } }); @@ -4036,6 +4077,7 @@ test("a source start follows the current requested floor after an update", async const producer = new TrackProducer("test"); const reader = producer.subscribe({ ranges: [{ start: { absolute: 5n } }] }); try { + hooks.receiveTrack(producer); hooks.startTrack(producer, { start: 6n, requested: 5n }); expect(hooks.sourceStart(reader)).toBe(6n); const lower = producer.subscribe({ ranges: [{ start: { absolute: 2n } }] }); @@ -4064,6 +4106,7 @@ test("a source start follows the current requested floor after an update", async test("a pending tail owes from SUBSCRIBE_START while the demand keeps its floor", async () => { const producer = new TrackProducer("test"); const reader = producer.subscribe({ delay: { max: Milli(60_000) }, ranges: [{ start: { absolute: 1n } }] }); + hooks.receiveTrack(producer); hooks.startTrack(producer, { start: 3n, requested: 1n }); hooks.endTrack(producer, { group: 4n, frame: 0 }); const group = producer.createGroup(3n); @@ -4084,6 +4127,7 @@ test("a pending tail owes from SUBSCRIBE_START while the demand keeps its floor" test("an end after the highest group holds the reader at the hole", async () => { const producer = new TrackProducer("test"); const reader = producer.subscribe({ delay: { max: Milli(60_000) } }); + hooks.receiveTrack(producer); hooks.startTrack(producer, { start: 0n }); const high = producer.createGroup(2n); high.finish(); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 17f499ba0c2..fd3bd6a4d8a 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -724,14 +724,11 @@ class TrackState { // One past the highest group received, including evicted content. received = new Signal(0n); /** - * The exclusive end, declared by {@link Producer.finishAt} or stamped by a clean close - * at the head of the sequence after the highest produced. Mirrors - * the Rust `end`. + * The declared end, by {@link Producer.finishAt} or a clean close at the head of the + * sequence after the highest produced, with the error the track ended with there. One + * value, so a wake on it sees both. Mirrors the Rust `End`. */ - end = new Signal(undefined); - // The error the track ended with at `end`, if it did not end cleanly: a reader reaching the - // end gets it instead of the end of stream. Set before `end`, so its wake sees it. - endError?: Error; + end = new Signal(undefined); // A wire subscription's pending tail still owes a group below the end, so reaching the // end does not end readers yet. Mirrors the Rust `awaits_tail`. owed = new Signal(false); @@ -755,6 +752,30 @@ class TrackState { } } +// A track's declared end. Mirrors the Rust `End`. +type End = { + // The exclusive end: no frame at or past it exists. + at: Position; + // The error the track ended with at `at`, if it did not end cleanly: a reader reaching the + // end gets it instead of the end of stream, while everything below stays readable. + error?: Error; + // A received end, declared ahead of the group streams below it that QUIC may deliver in any + // order: readers wait for a hole below it to fill, or for the track to close, rather than end + // at the boundary with groups still to arrive. Cleared once the subscription feeding the copy + // ends (see `hooks.cacheTrack`). Mirrors the Rust `End::tail`. + tail: boolean; +}; + +// The wire subscription feeding a received copy, and what it declared. Ending the subscription +// drops it whole, so nothing it declared outlives it. +type Wire = { + // How its start bounds the fills readers ask below it. + starts: FillStarts; + // The first group its live feed serves (see `hooks.startTrack`), unset until its START + // resolves it, null for a resolved-empty range, and the floor its SUBSCRIBE requested. + served?: { start?: bigint | null; requested?: bigint }; +}; + // Settle a track state's closed Once. Idempotent: Once.set throws on a second settle, and a // Producer closing its sinks races the Subscriber closing itself (the sink is only removed from // #sinks a microtask later, via the closed subscription below). @@ -1025,17 +1046,9 @@ export class Producer { // Only real objects or upstream metadata advance this, never the head of an empty group. #objectLargest?: Position; - // A wire subscription declared the end ahead of the group streams below it, which QUIC - // may deliver in any order: readers wait for a hole below it to fill, or for the track - // to close, rather than end at the boundary with groups still to arrive. - #tailPending = false; - // The first group the wire subscription's live feed serves (see `hooks.startTrack`), unset - // until its START resolves it, null for a resolved-empty range, and the requested floor. - #served?: { start?: bigint | null; requested?: bigint }; - // Only copies fed by a wire subscription owe missing groups. - #subscribed = false; - // How the subscription's start bounds the fills readers ask below it. - #starts: FillStarts = {}; + // The wire subscription feeding this copy, from `hooks.receiveTrack` until `hooks.cacheTrack` + // ends it: only a copy one feeds owes missing groups. Unset on a track's own publisher. + #wire?: Wire; // The misses its readers asked of a {@link Requests} handle. #fills: FillState; // The sequences the live feed delivered, as a group: what a pending tail is @@ -1097,10 +1110,14 @@ export class Producer { hooks.requested = (producer) => producer.#state.update; resubscribe = (producer) => producer.#updateSubscription(); hooks.cacheTrack = (producer) => { - producer.#subscribed = false; - producer.#served = undefined; - producer.#tailPending = false; - for (const state of [producer.#state, ...producer.#sinks]) state.owed.set(false); + producer.#wire = undefined; + // Nothing feeds the tail anymore, so readers end at the boundary. + const end = producer.#state.end.peek(); + const untailed = end?.tail ? { ...end, tail: false } : undefined; + for (const state of [producer.#state, ...producer.#sinks]) { + if (untailed) state.end.set(untailed); + state.owed.set(false); + } producer.#settle(); }; hooks.refuseLive = (producer, edge, error) => { @@ -1113,24 +1130,23 @@ export class Producer { } }; fillEdge = (producer) => - !producer.#subscribed && producer.#state.info.peek()?.ietf !== undefined + !producer.#wire && producer.#state.info.peek()?.ietf !== undefined ? producer.#state.largest.peek()?.group : undefined; hooks.receiveTrack = (producer, starts = {}) => { - producer.#subscribed = true; - producer.#starts = starts; + producer.#wire = { starts, served: producer.#wire?.served }; }; hooks.receiveGroup = (producer, { sequence, start }) => producer.#createGroup(sequence, start); hooks.arrivingGroup = (producer, sequence) => { const closed = producer.#cached.get(sequence)?.group.closed; return closed?.peek() === undefined ? closed : undefined; }; - sourceStart = (producer) => producer.#served && producer.#feedStart(); + sourceStart = (producer) => producer.#wire?.served && producer.#feedStart(); fillFloor = (producer, named) => producer.#fillFloor(named); startsOf = (producer) => producer.#bounds; goneAt = (producer, sequence) => { // Without a declared start, the first arrival stands in for it, as the tail's does. - const floor = producer.#served?.start ?? producer.#arrivals.first() ?? 0n; + const floor = producer.#wire?.served?.start ?? producer.#arrivals.first() ?? 0n; if (sequence < floor) return floor; const dropped = producer.#state.drops.peek().find((run) => run.start <= sequence && sequence < run.end); return dropped?.end ?? producer.#arrivals.runEnd(sequence); @@ -1158,8 +1174,9 @@ export class Producer { for (const state of [producer.#state, ...producer.#sinks]) state.dropsAnchors.set(true); }; hooks.startTrack = (producer, served) => { + if (!producer.#wire) throw new Error("only a wire subscription declares a start"); producer.#settle(); - producer.#served = served; + producer.#wire.served = served; producer.#owe(); for (const state of [producer.#state, ...producer.#sinks]) state.startChanged.update((rev) => rev + 1); }; @@ -1211,9 +1228,9 @@ export class Producer { }; mayLand = (producer, land) => producer.#landable(land) !== undefined; owedRuns = (producer, start, end) => { - if (!producer.#subscribed) return []; + if (!producer.#wire) return []; const feed = producer.#feedStart(); - const floor = feed === "pending" ? producer.#served?.requested : feed; + const floor = feed === "pending" ? producer.#wire.served?.requested : feed; return producer.#landing({ start: max(start, floor ?? 0n), end, @@ -1244,8 +1261,8 @@ export class Producer { if (first === undefined) return; start = max(start, first); } - start = max(start, this.#served?.start ?? 0n); - const declared = this.#state.end.peek(); + start = max(start, this.#wire?.served?.start ?? 0n); + const declared = this.#state.end.peek()?.at; const end = declared === undefined ? land.end : min(land.end, finalSequence(declared)); const at = now(); const limit = Nano.fromMilli(land.grace); @@ -1490,7 +1507,6 @@ export class Producer { sink.judged = this.#state.judged; sink.received.set(this.#received); - sink.endError = this.#state.endError; sink.end.set(this.#state.end.peek()); sink.owed.set(this.#state.owed.peek()); if (closed !== undefined) { @@ -1784,7 +1800,7 @@ export class Producer { // Refuse a write once the track is closed, or at or past its declared end. #writable(sequence: bigint): void { if (this.#state.closed.peek() !== undefined) throw new Error("track is closed"); - const end = this.#state.end.peek(); + const end = this.#state.end.peek()?.at; if (end !== undefined && sequence >= finalSequence(end)) { throw new Error(`sequence ${sequence} is at or past the track's end ${end.group}/${end.frame}`); } @@ -1792,7 +1808,7 @@ export class Producer { // Cap a group the end cuts, so it never holds a frame past the end. #cap(group: GroupProducer): void { - const end = this.#state.end.peek(); + const end = this.#state.end.peek()?.at; if (end !== undefined && end.frame !== 0 && end.group === group.sequence) hooks.limitGroup(group, end.frame); } @@ -1872,7 +1888,7 @@ export class Producer { // Record a sequence this track's own publisher produced, for the next append to skip around. #produce(sequence: bigint): void { - if (this.#subscribed || (this.#appended !== undefined && sequence < this.#appended)) return; + if (this.#wire || (this.#appended !== undefined && sequence < this.#appended)) return; insertRun(this.#produced, sequence, sequence + 1n); } @@ -1965,28 +1981,30 @@ export class Producer { */ finishAt(end: Position, error?: Error): void { if (this.#state.closed.peek() !== undefined) throw new Error("track is closed"); - const declared = this.#state.end.peek(); + const declared = this.#state.end.peek()?.at; if (declared !== undefined) throw new Error(`track already ends at ${declared.group}/${declared.frame}`); - this.#declareEnd(end, error); + this.#declareEnd(end, error, false); } // The wire's end: repeating it is a no-op, and a mid-group end may refine the group // boundary it falls just before. Either must carry the same error code. Mirrors the Rust - // `finish_at` on a `received()` copy. + // `finish_at` on a `received()` copy. Only a wire subscription owes a tail below the end: a + // FETCH's end leaves nothing in flight, as the Rust `set_end` a fill declares. #endPending(end: Position, error?: Error): void { - // Before the end lands, so settling it already counts what the tail still owes. - this.#tailPending = true; const declared = this.#state.end.peek(); const code = (err?: Error) => (err === undefined ? undefined : toStreamCode(err)); + const tail = this.#wire !== undefined; if (declared === undefined) { - this.finishAt(end, error); - } else if (code(this.#state.endError) !== code(error)) { + if (this.#state.closed.peek() !== undefined) throw new Error("track is closed"); + this.#declareEnd(end, error, tail); + } else if (code(declared.error) !== code(error)) { throw new ProtocolViolation(`conflicting track end error: ${error}`); - } else if (declared.group !== end.group || declared.frame !== end.frame) { - if (declared.frame !== 0 || end.frame === 0 || finalSequence(end) !== declared.group) { + } else if (declared.at.group !== end.group || declared.at.frame !== end.frame) { + if (declared.at.frame !== 0 || end.frame === 0 || finalSequence(end) !== declared.at.group) { throw new ProtocolViolation(`conflicting track end: ${end.group}/${end.frame}`); } - this.#declareEnd(end, error); + // A tail `cacheTrack` cleared stays cleared: no subscription feeds it anymore. + this.#declareEnd(end, error, declared.tail || tail); } this.#owe(); } @@ -2011,9 +2029,9 @@ export class Producer { // Tell every reader whether the pending tail still owes a group below the end. #owe(): void { - if (!this.#tailPending) return; const end = this.#state.end.peek(); - const owed = end !== undefined && this.#owes(finalSequence(end)); + if (!end?.tail) return; + const owed = this.#owes(finalSequence(end.at)); for (const state of [this.#state, ...this.#sinks]) state.owed.set(owed); this.#settle(); } @@ -2023,7 +2041,7 @@ export class Producer { // write is still served, but the clean outcome stays final. Mirrors the Rust `latch_settled`. #settle(): void { if (this.#state.outcome.peek() === undefined && this.#settled()) - this.#state.outcome.set(this.#state.endError ?? null); + this.#state.outcome.set(this.#state.end.peek()?.error ?? null); } // Whether the pending tail misses a sequence from where the feed starts up to `final`: one @@ -2037,7 +2055,7 @@ export class Producer { #feedStart(): bigint | "pending" | undefined { const ranges = this.#state.update.peek()?.ranges; const requested = ranges && wireSpan(ranges).start; - const served = this.#served; + const served = this.#wire?.served; if (!served || served.requested !== requested) return requested; if (served.start === null) return requested; if (served.start !== undefined) return served.start; @@ -2051,16 +2069,17 @@ export class Producer { // name; and never on a wire whose fills cannot judge a group by that budget. Mirrors the // Rust `Misses::judge`. Fetches are never bounded by it. #fillFloor(named: bigint | undefined): bigint | undefined { - if (!this.#subscribed) return undefined; + const wire = this.#wire; + if (!wire) return undefined; const feed = this.#feedStart(); - const start = feed === "pending" ? this.#served?.requested : feed; + const start = feed === "pending" ? wire.served?.requested : feed; if (start === undefined || start === 0n) return undefined; - if (this.#starts.fillsFromStart) return start; + if (wire.starts.fillsFromStart) return start; // The START applies only while the demand keeps the floor it answered, as `#feedStart`. const ranges = this.#state.update.peek()?.ranges; - const served = this.#served; + const served = wire.served; const resolved = - this.#starts.resolvesStart === true && + wire.starts.resolvesStart === true && typeof served?.start === "bigint" && served.requested === (ranges && wireSpan(ranges).start); if (resolved) return undefined; @@ -2068,14 +2087,14 @@ export class Producer { } #owes(final: bigint): boolean { - if (!this.#tailPending) return false; + if (!this.#state.end.peek()?.tail) return false; const feed = this.#feedStart(); - const floor = (feed === "pending" ? this.#served?.requested : feed) ?? this.#arrivals.first(); + const floor = (feed === "pending" ? this.#wire?.served?.requested : feed) ?? this.#arrivals.first(); if (floor === undefined) return final > 0n; return !this.#arrivals.covers(min(floor, final), final); } - #declareEnd(end: Position, error: Error | undefined): void { + #declareEnd(end: Position, error: Error | undefined, tail: boolean): void { const { group, frame } = end; if (typeof group !== "bigint" || group < 0n || !Number.isSafeInteger(frame) || frame < 0) { throw new RangeError(`invalid track end: ${group}/${frame}`); @@ -2086,11 +2105,8 @@ export class Producer { } const cut = frame === 0 ? undefined : this.#cached.get(group)?.group; if (cut) hooks.limitGroup(cut, frame); - end = { group, frame }; - for (const state of [this.#state, ...this.#sinks]) { - state.endError = error; - state.end.set(end); - } + const declared: End = { at: { group, frame }, error, tail }; + for (const state of [this.#state, ...this.#sinks]) state.end.set(declared); this.#endLargest(); } @@ -2108,7 +2124,7 @@ export class Producer { #endLargest(): void { this.#settle(); if (this.#state.largestEnded.peek() || this.#largestOpen) return; - const end = this.#state.end.peek(); + const end = this.#state.end.peek()?.at; if (end === undefined || !reaches(this.#received, this.#state.drops.peek(), finalSequence(end))) return; for (const state of [this.#state, ...this.#sinks]) state.largestEnded.set(true); } @@ -2155,13 +2171,14 @@ export class Producer { #close(abort: Error | undefined) { if (this.#state.closed.peek() !== undefined) return; // A gap past the grace was lost before the abort rather than cut off by it. - if (abort && this.#tailPending) this.#arrivals.expire(now(), this.#grace()); + if (abort && this.#state.end.peek()?.tail) this.#arrivals.expire(now(), this.#grace()); if (abort && (this.#state.outcome.peek() === null || this.#settled())) abort = undefined; // Nobody will finish these, so a subscriber that has not taken one yet never sees it. // Not evicted: a reader already holding one keeps its frames and sees the abort. const open = [...this.#cache].filter((entry) => entry.group.closed.peek() === undefined); closeTrackState(this.#state, abort); - if (this.#state.outcome.peek() === undefined) this.#state.outcome.set(abort ?? this.#state.endError ?? null); + if (this.#state.outcome.peek() === undefined) + this.#state.outcome.set(abort ?? this.#state.end.peek()?.error ?? null); // Closing a copy never supplies the missing tail of a group. Already finished groups // ignore the abort and stay readable; unfinished ones retain their truncation cause. const groupError = @@ -2192,7 +2209,7 @@ export class Producer { // abort over a group the tail still owed cut it off, so readers get the error rather than // a clean end without it. Mirrors the Rust `is_settled`. #settled(): boolean { - const end = this.#state.end.peek(); + const end = this.#state.end.peek()?.at; if (end === undefined) return false; const final = finalSequence(end); if (!reaches(this.#received, this.#state.drops.peek(), final) || this.#owes(final)) return false; @@ -2257,8 +2274,9 @@ export class Subscriber { // At most the narrowest budget in #skipped, so a read whose budget did not widen skips the scan. #narrowest = Number.POSITIVE_INFINITY; // This reader closed itself, so lowering its floor offers nothing: a producer's clean close - // alone leaves the cache readable, but the reader leaving must stay terminal. - #closed = false; + // alone leaves the cache readable, but the reader leaving must stay terminal. Leaving also + // settles the shared `#state.closed`, so a guard for either close reads that alone. + #left = false; // What {@link largest} last reported, and whether it ended: moving the bounds afterwards // does not bring it back. #largestSeen?: Position; @@ -2349,7 +2367,7 @@ export class Subscriber { // stamped group's latest frame: walk back from the cap to the first one. for (let i = (end === undefined ? timeline.length : timelineIndex(timeline, end)) - 1; i >= 0; i--) { const group = timeline[i]; - if (group.closed.peek() instanceof Error || !this.#judges(group.sequence)) continue; + if (hooks.groupFailed(group) || !this.#judges(group.sequence)) continue; const timestamp = hooks.groupTimestamp(group); if (timestamp === undefined) continue; presentation = { sequence: group.sequence, timestamp: hooks.groupLatest(group) ?? timestamp }; @@ -2403,7 +2421,7 @@ export class Subscriber { for (let i = timelineIndex(timeline, sequence + 1n); i < timeline.length; i++) { const successor = timeline[i]; if (end !== undefined && successor.sequence >= end) break; - if (successor.closed.peek() instanceof Error || !this.#judges(successor.sequence)) continue; + if (hooks.groupFailed(successor) || !this.#judges(successor.sequence)) continue; if (counts(successor)) return successor; } return undefined; @@ -2682,7 +2700,7 @@ export class Subscriber { } this.#narrowest = narrowest; const producer = this.#state.producer; - if (producer && !this.#closed) reofferSequences(producer, this.#state, sequences); + if (producer && !this.#left) reofferSequences(producer, this.#state, sequences); } // Owe again every run this cursor gave up as stale, reported or not: an update reopens the @@ -2723,7 +2741,7 @@ export class Subscriber { for (const run of kept) this.#pending = insertDrop(this.#pending, run.start, run.end, hole); } const producer = this.#state.producer; - if (sequences.length > 0 && producer && !this.#closed) reofferSequences(producer, this.#state, sequences); + if (sequences.length > 0 && producer && !this.#left) reofferSequences(producer, this.#state, sequences); return taken; } @@ -2853,7 +2871,7 @@ export class Subscriber { // The parts of `[start, end)` this reader still reads: inside its cursor and window, not // handed out, and not passed by the sequence cursor, which never goes back. #reads(start: bigint, end: bigint): Run[] { - if (this.#closed || this.#state.closed.peek() !== undefined) return []; + if (this.#state.closed.peek() !== undefined) return []; return this.#unread(start, end); } @@ -2879,7 +2897,7 @@ export class Subscriber { // Whether this reader will still read group `sequence` once buffered: what it wants from a // fill, judged even once the track closed. #reading(sequence: bigint): boolean { - if (this.#closed) return false; + if (this.#left) return false; const unread = this.#unread(sequence, sequence + 1n); return unread.length > 0 && runGaps(this.#givenUp, sequence, sequence + 1n).length > 0; } @@ -2911,7 +2929,7 @@ export class Subscriber { // the wakeup for when that changes. Mirrors the Rust `Misses::judge`. #askFills(start: bigint, end: bigint, drift?: Drift): bigint | undefined { const producer = this.#state.producer; - if (this.#resume || !producer || this.#closed) return undefined; + if (this.#resume || !producer || this.#left) return undefined; const fills = fillsOf(producer); // A fill already out can be joined even once its handler's handle closed. if (fills.serving || fills.active) { @@ -2956,7 +2974,7 @@ export class Subscriber { const group = timeline[i]; if (group.sequence >= open.end) break; // A failed copy holds nothing to read, so it is missed like an evicted one. - if (group.closed.peek() instanceof Error) continue; + if (hooks.groupFailed(group)) continue; this.#askMiss(next, group.sequence, drift, asker); next = group.sequence + 1n; } @@ -3117,9 +3135,9 @@ export class Subscriber { static { makeSubscriber = (name, state) => new Subscriber(name, state); hooks.subscriberEnd = (subscriber) => - subscriber.#resume ? subscriber.#resume.end() : subscriber.#state.end.peek(); + subscriber.#resume ? subscriber.#resume.end() : subscriber.#state.end.peek()?.at; hooks.subscriberEndError = (subscriber) => - subscriber.#resume ? subscriber.#resume.endError() : subscriber.#state.endError; + subscriber.#resume ? subscriber.#resume.endError() : subscriber.#state.end.peek()?.error; hooks.joinSubscriber = (subscriber) => { if (subscriber.#resume) subscriber.#resume.join(); else if (subscriber.#state.producer) hooks.joinProducer(subscriber.#state.producer); @@ -3201,7 +3219,7 @@ export class Subscriber { group.sequence >= start && (end === undefined || group.sequence < end) && !delivered.has(group.sequence) && - !(group.closed.peek() instanceof Error), + !hooks.groupFailed(group), ); hooks.spent = (subscriber, start, end, delivered) => (subscriber.#state.closed.peek() !== undefined || subscriber.#complete()) && @@ -3360,7 +3378,7 @@ export class Subscriber { async finished(): Promise { if (this.#resume) return this.#resume.finished(); for (;;) { - const end = this.#state.end.peek(); + const end = this.#state.end.peek()?.at; if (end !== undefined) return end; const closed = this.#state.closed.peek(); @@ -3407,7 +3425,7 @@ export class Subscriber { for (;;) { const start = hooks.sourceStart(this); if (start !== "pending") return start; - if (this.#closed || this.#state.closed.peek() !== undefined) return undefined; + if (this.#state.closed.peek() !== undefined) return undefined; await this.#wait(); } } @@ -3524,7 +3542,7 @@ export class Subscriber { this.#bound(); const { start } = this.#cursor.peek(); const producer = this.#state.producer; - if (!this.#resume && !this.#closed && producer && start < before.start) { + if (!this.#resume && !this.#left && producer && start < before.start) { // The ordered cursor let go of what it passed below the old floor; the arrival // cursor keeps it buffered, which this skips. reofferCached( @@ -3540,14 +3558,14 @@ export class Subscriber { /** Close the track (optionally with an error), closing any pending groups. Idempotent. */ close(abort?: Error) { - if (!this.#closed) { - const arrivals = this.#state.timeline.filter((group) => !(group.closed.peek() instanceof Error)); + if (!this.#left) { + const arrivals = this.#state.timeline.filter((group) => !hooks.groupFailed(group)); this.#arrivals = arrivals.map((group) => ({ sequence: group.sequence, arrived: hooks.groupArrived(group), })); } - this.#closed = true; + this.#left = true; this.#state.waitingOrdered = false; // Settle if we're first (the producer may already have); either way drop anything // still buffered. Groups parked at the withRange cap deliberately outlive a clean @@ -3666,8 +3684,8 @@ export class Subscriber { continue; } groups.splice(index, 1); - const failed = group.closed.peek(); - if (dispositions && failed instanceof Error) { + const failed = hooks.groupFailed(group); + if (dispositions && failed) { // A group expired under this subscription's budget is a stale verdict. const hole = toStreamCode(failed) === StreamCode.Old @@ -3815,7 +3833,7 @@ export class Subscriber { // owes below it. Reader completion does not settle the tail: open streams remain // accounted for by the wire layer, and an abort still wins on the next read. #complete(): boolean { - const end = this.#state.end.peek(); + const end = this.#state.end.peek()?.at; return ( end !== undefined && reaches(this.#state.received.peek(), this.#state.drops.peek(), finalSequence(end)) && @@ -4001,7 +4019,7 @@ export class Subscriber { this.#state.waitingOrdered = true; return { kind: "idle" }; } - const end = this.#state.end.peek(); + const end = this.#state.end.peek()?.at; if (this.#exhausted()) return { kind: "done" }; if ( !this.#parked(groups) && diff --git a/quest/m1/explicit-state/README.md b/quest/m1/explicit-state/README.md index 02506d23ad4..3231cec1438 100644 --- a/quest/m1/explicit-state/README.md +++ b/quest/m1/explicit-state/README.md @@ -73,7 +73,6 @@ which collapses that state and waits on [Drop set](/quest/m1/drop-set.md). - [Publish capture phases](/quest/m1/explicit-state/publish-capture.md) - dismissing the screen-share picker turns sharing off - [js/qmux close](/quest/m1/explicit-state/qmux-close.md) - a graceful remote close ends qmux streams cleanly - [FFI handle state](/quest/m1/explicit-state/ffi-state.md) - FFI handles hold one state each: video follow under one lock, requests answered once, drivers Cancelled after shutdown -- [Track and group end-of-life](/quest/m1/explicit-state/track-end.md) - moq-net's track and group outcomes are one enum each, in Rust and JS - [moq-net phase data](/quest/m1/explicit-state/net-phases.md) - moq-net's session and subscription state machines keep each phase's data in its variant - [moq-mux and moq-gst lifecycles](/quest/m1/explicit-state/mux-gst.md) - export sources, TS and fMP4 export tracks, and gst pads are explicit state - [Video's declared-end freeze is not a stall](/quest/m1/explicit-state/video-declared-end.md) - video holding its last picture through a declared end counts no stall, like audio's declared silence diff --git a/quest/m1/explicit-state/track-end.md b/quest/m1/explicit-state/track-end.md deleted file mode 100644 index 092a290e4df..00000000000 --- a/quest/m1/explicit-state/track-end.md +++ /dev/null @@ -1,66 +0,0 @@ -# [XL] Track and group end-of-life - -## Goal - -In moq-net and js/net, each track and group has one outcome value: open, -ended, sealed, aborted, or dropped. Every predicate that currently re-derives -an outcome from separate flags reads that value instead, and in JS one -`closed` handle stays stable for a consumer's whole life. The Rust and JS -models change together, as the cross-package sync rule requires. - -## Plan - -The shared decisions are in the [epic](/quest/m1/explicit-state/README.md). -This is moq-net's hot path, so benchmark it as the epic says. - -Bug to reproduce and fix: - -- **`recoverGroup` replaces the stable handle** (`js/net/src/group.ts`). - - The bug: `hooks.recoverGroup` resets `group.#closed = undefined`. A - caller that read `closed` before the recovery keeps the old handle, - latched to the error, while a new read builds a handle that can settle - `null`. That breaks the `CombinedClosed` doc's promise of one stable - public handle. - - Decide in the PR whether a non-final error that a recovery continues - past may ever reach a caller. If the current behavior turns out to be - intended, keep it and document it in one place. - -Instances to redesign: - -- **Rust `TrackState`** (`rs/moq-net/src/model/track.rs`): - - Eight fields hold its end-of-life: `end`, `end_error`, `sealed`, - `closed`, `abort`, `resume`, `settled`, and `tail`. - - `end` and `abort` can both be set, and which outcome wins depends on - whether `settled` latched first. - - `sealed || abort.is_some()` is repeated at five or more sites. - - The start and anchor flag triples (`start_sequence`, `start_pending`, - `start_resolved`; `received`, `anchored`, `cached_anchors`) belong to - the same redesign. -- **Rust group** (`rs/moq-net/src/model/group.rs`): - - `GroupState` keeps `fin` and `abort` side by side. - - `Alive` copies the abort into three atomics (`cut`, `failed`, - `aborted`), and readers can see the steps in between. One atomic phase - should replace them. -- **JS track** (`js/net/src/track.ts`): - - `TrackState` keeps `end` as a Signal and `endError` as a plain field - that has to be set first. - - The `Producer` wire-mode flags (`#subscribed`, `#served`, - `#tailPending`, `#starts`). - - `Subscriber` has two closed sources, `#closed` and `#state.closed`, and - its guards check them inconsistently. -- **JS group** (`js/net/src/group.ts`): `#terminal` and `#verdict` are synced - at seven sites, and one of them assigns without `??=`. -- **Resume** (`js/net/src/resume.ts` `Reader`, `rs/moq-net/src/model/resume.rs` - `Copy`): - - Reproduce whether a retained copy can refuse after a clean end - (`#learnEnd` checks only `#closed`, not `#ended`). If it can, `closed` - reports clean while reads fail. - -[Group errors are cached by kind](/quest/m1/group-error-cache.md) and -[Drop set](/quest/m1/drop-set.md) edit the same track state. Coordinate with -whichever lands first rather than wait. - -## Related - -- [Group errors are cached by kind](/quest/m1/group-error-cache.md) - changes how group outcomes are recorded -- [Drop set](/quest/m1/drop-set.md) - rewrites the track's dropped-run state diff --git a/quest/m2/finished-handle.md b/quest/m2/finished-handle.md index 83957f56fce..c8bac8fa546 100644 --- a/quest/m2/finished-handle.md +++ b/quest/m2/finished-handle.md @@ -63,5 +63,5 @@ packages; moq-ffi and moq-c keep their signatures and runtime checks. Wire: none ## Related -- [Track and group end-of-life](/quest/m1/explicit-state/track-end.md) - the outcome enums `Finished` would wrap +- [Track and group end-of-life](https://github.com/moq-dev/moq/pull/8243) - the outcome enums `Finished` would wrap (`Outcome` in `track.rs` and `group.rs`) - [A failed finish is final](https://github.com/moq-dev/moq/pull/8210) - the explicit state this builds on diff --git a/rs/moq-net/src/model/frame.rs b/rs/moq-net/src/model/frame.rs index 10287fb8891..103731bb08c 100644 --- a/rs/moq-net/src/model/frame.rs +++ b/rs/moq-net/src/model/frame.rs @@ -1732,7 +1732,7 @@ impl Consumer { // Park on the group's channel; the producer notifies it on each write and // on abort. Re-check the atomic on wake. ready!(poll_state(&self.state, waiter, |state| { - if let Some(err) = &state.abort { + if let Some(err) = state.outcome.error() { return Poll::Ready(Err(err.clone())); } let w = buf.written(Ordering::Acquire); @@ -1775,7 +1775,7 @@ impl Consumer { return Poll::Ready(Err(Error::Old)); } ready!(poll_state(&self.state, waiter, |state| { - if let Some(err) = &state.abort { + if let Some(err) = state.outcome.error() { return Poll::Ready(Err(err.clone())); } if buf.written(Ordering::Acquire) >= size { @@ -1799,7 +1799,7 @@ where { Poll::Ready(match ready!(state.poll(waiter, f)) { Ok(res) => res, - Err(state) => Err(state.abort.clone().unwrap_or(Error::Dropped)), + Err(state) => Err(state.outcome.error().cloned().unwrap_or(Error::Dropped)), }) } diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 8353dddecbc..fe7f68f8e0e 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -36,7 +36,7 @@ use content::{Input, Spec}; use std::collections::VecDeque; use std::mem::MaybeUninit; use std::ops::RangeBounds; -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex, MutexGuard, OnceLock, PoisonError, Weak}; use std::task::{Poll, ready}; @@ -214,7 +214,7 @@ pub(crate) struct Bounds { pub end: Option, } -/// Shared group state. Model-visible so [`frame`] handles can observe the abort flag +/// Shared group state. Model-visible so [`frame`] handles can observe its outcome /// while streaming a partial frame. #[derive(Default)] pub(super) struct GroupState { @@ -266,14 +266,10 @@ pub(super) struct GroupState { // How the track judged the start against its neighbors' when frame 0 landed. origin: Origin, - // Once finalized, the total number of frames the group will ever contain. Recorded - // at finish so the count outlives an abort that clears the cache. - pub(super) fin: Option, - - // The error that caused the group to be aborted, if any. Mirrored into - // `Alive::aborted`, so [`Producer::abort`] stays the only writer: anything else - // setting this would leave track scans reading a group as live. - pub(super) abort: Option, + // How the group ended, if it did. The finished count outlives an abort that frees the + // cache. Mirrored into `Alive::fall` by the one abort path, `Handle::commit_abort`: + // anything else ending it would leave track scans reading a group as live. + pub(super) outcome: Outcome, // The readers that could resume this group on another route. A release that is no // cache policy moves the frames to them rather than dropping them. @@ -287,6 +283,67 @@ pub(super) struct GroupState { pace: Arc, } +/// How far a group got toward its end; each step replaces the last. +#[derive(Clone, Debug, Default)] +pub(super) enum Outcome { + /// Frames may still be written. + #[default] + Open, + /// Finished with this many frames. + Finished(usize), + /// Finished with this many frames, then aborted to free its cache: a reader that took + /// every frame got the whole group, and one short of it gets the error. + Released(usize, Error), + /// Aborted before it finished. + Aborted(Error), +} + +impl Outcome { + /// The total number of frames, once finished. + pub(super) fn frames(&self) -> Option { + match self { + Self::Finished(frames) | Self::Released(frames, _) => Some(*frames), + Self::Open | Self::Aborted(_) => None, + } + } + + /// The error the group was aborted with, before or after it finished. + pub(super) fn error(&self) -> Option<&Error> { + match self { + Self::Released(_, err) | Self::Aborted(err) => Some(err), + Self::Open | Self::Finished(_) => None, + } + } +} + +/// Whether and why a group was aborted, as track scans read it without taking the group +/// lock: one atomic phase in [`Alive`], so a scan never sees part of an abort. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[repr(u8)] +enum Fall { + /// Not aborted. + Live, + /// Freed by cache policy (`Old`, `Timeout`, `Evicted`), or after it finished: it aged + /// out of a cache rather than failed. + Released, + /// Given up deliberately: a reader cancelling, a lagged copy, or the upstream's own + /// cache policy. It failed, but readers never see a cut. + Abandoned, + /// The publisher gave it up, with the error in [`Alive::cut`]. + Cut, +} + +impl Fall { + fn from_u8(phase: u8) -> Self { + match phase { + 0 => Self::Live, + 1 => Self::Released, + 2 => Self::Abandoned, + _ => Self::Cut, + } + } +} + /// The frames a copy of a group holds, from where it starts, which a successor copy of /// the same group on another route adopts below where it starts. pub(super) struct Head { @@ -807,7 +864,7 @@ impl GroupState { false => plain.is_some_and(|plain| plain.shed_past(index)), }; if shed { - return Err(self.abort.clone().unwrap_or(Error::Evicted)); + return Err(self.outcome.error().cloned().unwrap_or(Error::Evicted)); } Ok(()) } @@ -958,10 +1015,10 @@ impl GroupState { /// carrying the debt over as a whole group does. A compressed group's plain copy sheds /// its front the same way. fn shed(&mut self, debt: u64, cap: u64) -> u64 { - if self.abort.is_some() { + if self.outcome.error().is_some() { return 0; } - let last = match self.fin { + let last = match self.outcome.frames() { Some(_) => self.committed(), None => self.committed().saturating_sub(1), }; @@ -983,7 +1040,9 @@ impl GroupState { && freed < cap && freed < debt { - freed += flate.plain.shed(debt - freed, cap - freed, self.fin.is_none()); + freed += flate + .plain + .shed(debt - freed, cap - freed, self.outcome.frames().is_none()); } freed } @@ -1020,7 +1079,7 @@ impl GroupState { /// Whether the first frame opened, or the group ended without one. fn started(&self) -> bool { - !matches!(self.timeline, Timeline::Empty) || self.fin.is_some() || self.abort.is_some() + !matches!(self.timeline, Timeline::Empty) || self.outcome.frames().is_some() || self.outcome.error().is_some() } /// The bytes the in-flight frame declared but has not written yet, which its open @@ -1083,13 +1142,13 @@ impl GroupState { let from = flate.plain.len.load(Ordering::Acquire).min(last + 1); // The slices the plain copy needs next were shed before it inflated them. if from <= last && from < self.held() { - return Err(self.abort.clone().unwrap_or(Error::Evicted)); + return Err(self.outcome.error().cloned().unwrap_or(Error::Evicted)); } Ok(Inflate { release: flate .release .clone() - .ok_or_else(|| self.abort.clone().unwrap_or(Error::Dropped))?, + .ok_or_else(|| self.outcome.error().cloned().unwrap_or(Error::Dropped))?, plain: flate.plain.clone(), from, keep: first, @@ -1160,11 +1219,11 @@ impl GroupState { /// that already consumed every frame is missing nothing, so it gets the clean end of /// group; one that fell short sees the abort rather than a silently truncated stream. fn poll_terminal(&self, index: usize) -> Poll> { - match (self.fin, &self.abort) { - (Some(total), Some(err)) if index < total => Poll::Ready(Err(err.clone())), - (Some(_), _) => Poll::Ready(Ok(())), - (None, Some(err)) => Poll::Ready(Err(err.clone())), - (None, None) => { + match &self.outcome { + Outcome::Released(total, err) if index < *total => Poll::Ready(Err(err.clone())), + Outcome::Finished(_) | Outcome::Released(..) => Poll::Ready(Ok(())), + Outcome::Aborted(err) => Poll::Ready(Err(err.clone())), + Outcome::Open => { // Waiting on frames the writer has yet to commit, wherever the reader skipped // to, so it holds the writer back from none of them. if index >= self.committed() { @@ -1178,10 +1237,10 @@ impl GroupState { /// The group's outcome once final: `Ok` once finished, which a later abort only frees /// the cache of, or the abort error. fn poll_outcome(&self) -> Poll> { - match (self.fin, &self.abort) { - (Some(_), _) => Poll::Ready(Ok(())), - (None, Some(err)) => Poll::Ready(Err(err.clone())), - (None, None) => Poll::Pending, + match &self.outcome { + Outcome::Finished(_) | Outcome::Released(..) => Poll::Ready(Ok(())), + Outcome::Aborted(err) => Poll::Ready(Err(err.clone())), + Outcome::Open => Poll::Pending, } } @@ -1343,13 +1402,13 @@ impl GroupState { let head = handoff.as_ref().and_then(|_| self.head()); let shared = self.content.is_some(); if let Some(flate) = &mut self.flate { - flate.release(self.abort.clone().unwrap_or(Error::Dropped), shared); + flate.release(self.outcome.error().cloned().unwrap_or(Error::Dropped), shared); } self.front = VecDeque::new(); self.evict_log(self.log.end()); self.partial = None; if let Some(content) = self.content.take() { - content.detach(self.abort.clone().unwrap_or(Error::Dropped)); + content.detach(self.outcome.error().cloned().unwrap_or(Error::Dropped)); } self.cache = 0; self.charge.clear(); @@ -1383,7 +1442,9 @@ impl Drop for GroupState { } fn modify(state: &kio::Producer) -> Result> { - state.write().map_err(|r| r.abort.clone().unwrap_or(Error::Dropped)) + state + .write() + .map_err(|r| r.outcome.error().cloned().unwrap_or(Error::Dropped)) } /// Writes frames to a group in order. @@ -1457,33 +1518,41 @@ impl Detached { /// Whether a group failed, without holding it open or pinning its frames. pub(super) struct Fate(Arc); +impl Alive { + /// The abort's phase. Acquire pairs with the abort's release, so a [`Fall::Cut`] finds + /// its error in [`Self::cut`]. + fn fall(&self) -> Fall { + Fall::from_u8(self.fall.load(Ordering::Acquire)) + } +} + impl Fate { /// Whether the group was aborted other than by cache policy (`Old`, `Timeout`, /// `Evicted`): it failed or was given up, rather than aged out of a cache. pub(super) fn failed(&self) -> bool { - self.0.failed.load(Ordering::Acquire) + matches!(self.0.fall(), Fall::Abandoned | Fall::Cut) } } struct Alive { - // Monotone mirror of `GroupState::abort` for track scans that already hold the - // track lock. A stale false only hands out a group that is concurrently aborting; - // true is stored after the abort exists, so it can never hide a live group. + // Monotone mirror of an aborted `GroupState::outcome`, with why, for track scans that + // already hold the track lock. Set once, whole: an unset one only hands out a group + // that is concurrently aborting, and it is set after the outcome, so it can never hide + // a live group. Readable through [`Fate`] once the group is gone. // // Only ever read on its own. A decision that pairs the abort with something else // out of `GroupState` has to read both under one guard, or the two halves can // straddle the abort: see `Handle::live_first_frame`. - aborted: AtomicBool, + // + // A [`Fall`] as one atomic phase rather than a `OnceLock`, whose first set takes a + // slow path on every released group. + fall: AtomicU8, + // The publisher's error once `fall` reads [`Fall::Cut`]: set before that phase is + // published, so a scan reading it finds the error. + cut: OnceLock, // Set once frame 0's rewind check ran, so later writes skip the track lock it takes. A // stale false only takes that lock once more. checked: AtomicBool, - // The abort's error when it was the publisher giving the group up rather than cache - // policy (`Old`, `Timeout`, `Evicted`). Set before `aborted`, under the same rules, so - // a track scan learns why without taking the group lock. - cut: OnceLock, - // Set before `aborted` when the abort was not cache policy: the group failed or was given - // up rather than aged out of a cache. Read through [`Fate`] once the group is gone. - failed: AtomicBool, // Monotone mirror of `GroupState::started` for the first frame or the finish, set after // the group lock is released and before the track is told: a track scan reading a stale // false judges the group frameless until that wake re-judges it. @@ -1504,7 +1573,7 @@ struct Alive { impl Drop for Producer { fn drop(&mut self) { if let Ok(state) = self.handle.state.write() { - if state.fin.is_some() || state.abort.is_some() { + if !matches!(state.outcome, Outcome::Open) { return; } tracing::warn!( @@ -1578,7 +1647,7 @@ impl Producer { #[cfg(feature = "async")] pub(crate) fn set_end_object(&self, fields: super::object::Fields) -> Result<()> { let mut state = modify(&self.handle.state)?; - if state.fin.is_some() || state.relay.as_ref().is_some_and(|relay| relay.end_object.is_some()) { + if state.outcome.frames().is_some() || state.relay.as_ref().is_some_and(|relay| relay.end_object.is_some()) { return Err(Error::MalformedTrack); } let bytes = fields.heap(); @@ -1621,10 +1690,9 @@ impl Producer { (state.log.reader(), state.pace.clone()) }; let alive = Arc::new(Alive { - aborted: AtomicBool::new(false), - checked: AtomicBool::new(false), + fall: AtomicU8::new(Fall::Live as u8), cut: OnceLock::new(), - failed: AtomicBool::new(false), + checked: AtomicBool::new(false), started: AtomicBool::new(false), access, log, @@ -1698,7 +1766,7 @@ impl Producer { let mut state = modify(&self.handle.state)?; // Every write advances `next_index` past `offset`, so this is "nothing written // yet". - if state.fin.is_some() || state.next_index != state.offset || index < state.offset { + if state.outcome.frames().is_some() || state.next_index != state.offset || index < state.offset { return Err(Error::Closed); } state.offset = index; @@ -1742,7 +1810,7 @@ impl Producer { .as_ref() .and_then(|track| track.neighbors(self.handle.info.sequence)); let mut state = modify(&self.handle.state)?; - if state.fin.is_some() { + if state.outcome.frames().is_some() { return Err(Error::Closed); } debug_assert!(state.partial.is_none(), "a frame is already open"); @@ -1828,7 +1896,7 @@ impl Producer { .as_ref() .and_then(|track| track.neighbors(self.handle.info.sequence)); let mut state = modify(&self.handle.state)?; - if state.fin.is_some() { + if state.outcome.frames().is_some() { return Err(Error::Closed); } debug_assert!(state.partial.is_none(), "a frame is already open"); @@ -2000,7 +2068,7 @@ impl Producer { .as_ref() .and_then(|track| track.neighbors(self.handle.info.sequence)); let mut state = modify(&self.handle.state)?; - if state.fin.is_some() { + if state.outcome.frames().is_some() { return Err(Error::Closed); } debug_assert!(state.partial.is_none(), "a frame is already open"); @@ -2196,7 +2264,7 @@ impl Producer { let mut state = modify(&self.handle.state)?; debug_assert!(state.partial.is_none(), "a frame is already open"); let fin = state.next_index; - state.fin = Some(fin); + state.outcome = Outcome::Finished(fin); // No frame follows to compress. let original_fin = state.range_end.is_none().then_some(fin); if let Some(flate) = &mut state.flate { @@ -2334,7 +2402,7 @@ impl Handle { let index = reader.cursor.index; let state = self.state.read(); // Frames adopted below the log sit off this copy's pace scale. - if state.abort.is_some() || (index as u64) < state.log.base() { + if state.outcome.error().is_some() || (index as u64) < state.log.base() { return false; } // A plain read of a compressed copy inflates its frame from every slice since the @@ -2375,7 +2443,7 @@ impl Handle { // Sessions cancel an unfinished FETCH while its writer belongs to its data stream. pub(crate) fn abort_open(&self, err: Error) { if let Ok(state) = self.state.write() - && state.fin.is_none() + && state.outcome.frames().is_none() { self.commit_abort(state, err); } @@ -2389,7 +2457,7 @@ impl Handle { let Ok(mut state) = self.state.write() else { return Ok(None); }; - if state.abort.is_some() { + if state.outcome.error().is_some() { return Ok(None); } if let Some(content) = &state.content { @@ -2511,7 +2579,7 @@ impl Handle { let same_form = head.compressed == state.flate.is_some(); // A copy that shed its front holds no run from `offset` for the head to precede. let shed = state.held() != offset; - if state.abort.is_some() || state.sparse() || !same_form || shed || head.start >= offset || !reaches { + if state.outcome.error().is_some() || state.sparse() || !same_form || shed || head.start >= offset || !reaches { return; } let count = offset - head.start; @@ -2638,18 +2706,19 @@ impl Handle { fn commit_abort(&self, mut guard: kio::Mut<'_, GroupState>, err: Error) { let policy = matches!(err, Error::Old | Error::Timeout | Error::Evicted); - // A finished group's outcome is settled: an abort only releases its cache, so it - // neither failed nor was cut. - let settled = guard.fin.is_some(); // Cache policy frees the frames, as does an oversized group whichever write found // it; a copy that failed or was given up lends them on. let salvage = !policy && !matches!(err, Error::GroupTooLarge); - // A reset carrying a policy code is the upstream's cache policy, a cancel is a reader - // no longer wanting the group, and a fill refused as unfetchable abandons backfill: - // deliberate holes, never a cut readers see. The publisher giving the group up is. - let cut = !settled - && !policy - && !matches!( + let fall = match guard.outcome { + // A finished group's outcome is settled: an abort only releases its cache, so it + // neither failed nor was cut. + Outcome::Finished(_) => Fall::Released, + _ if policy => Fall::Released, + // A reset carrying a policy code is the upstream's cache policy, a cancel is a + // reader no longer wanting the group, and a fill refused as unfetchable abandons + // backfill: deliberate holes, never a cut readers see. The publisher giving the + // group up is. + _ if matches!( err, Error::Cancel | Error::Lagged | Error::Stream( @@ -2658,15 +2727,23 @@ impl Handle { | StreamError::DeliveryTimeout | StreamError::TooFarBehind ) - ); + ) => + { + Fall::Abandoned + } + _ => Fall::Cut, + }; + let cut = fall == Fall::Cut; if cut { let _ = self.alive.cut.set(err.clone()); } - if !settled && !policy { - self.alive.failed.store(true, Ordering::Release); - } - guard.abort = Some(err); - self.alive.aborted.store(true, Ordering::Release); + guard.outcome = match guard.outcome { + Outcome::Finished(frames) => Outcome::Released(frames, err), + _ => Outcome::Aborted(err), + }; + // After the outcome exists, so a scan reading the mirror never hides a live group. + // The group lock admits one abort, so this is the only store. + self.alive.fall.store(fall as u8, Ordering::Release); guard.release(salvage); guard.close(); self.cache.aborted(self.sequence, cut); @@ -2680,7 +2757,7 @@ impl Handle { /// concurrent abort can still read as live for the length of [`Self::abort`], /// which hands out a group whose consumer then surfaces the abort. pub(crate) fn is_aborted(&self) -> bool { - self.alive.aborted.load(Ordering::Acquire) + self.alive.fall() != Fall::Live } /// Why the publisher gave this group up, once it did, rather than cache policy @@ -2688,7 +2765,10 @@ impl Handle { /// /// Reads the mirror, like [`Self::is_aborted`], so a track scan never takes the group lock. pub(super) fn cut_error(&self) -> Option<&Error> { - self.alive.cut.get() + match self.alive.fall() { + Fall::Cut => self.alive.cut.get(), + Fall::Live | Fall::Released | Fall::Abandoned => None, + } } /// This group as [`Detached`], for the track to end once it stops caching it. @@ -2710,7 +2790,7 @@ impl Handle { /// Whether a transport-bounded copy may answer these requested frames. pub(super) fn covers(&self, frames: Bounds) -> bool { let state = self.state.read(); - state.abort.is_none() + state.outcome.error().is_none() && state.offset as u64 <= frames.start && match (state.range_end, frames.end) { (None, _) => true, @@ -2727,14 +2807,14 @@ impl Handle { /// Whether the writer finished this copy, which a transport bound may end short of the /// group's end. pub(super) fn is_settled(&self) -> bool { - self.state.read().fin.is_some() + self.state.read().outcome.frames().is_some() } /// Whether the group was finished: it holds every frame it will ever have. pub(crate) fn is_finished(&self) -> bool { { let state = self.state.read(); - state.fin.is_some() && state.range_end.is_none() + state.outcome.frames().is_some() && state.range_end.is_none() } } @@ -2749,7 +2829,7 @@ impl Handle { /// ([`Self::is_aborted`]) is for scans that ask about the abort alone. pub(super) fn live_first_frame(&self) -> Option { let state = self.state.read(); - state.abort.is_none().then_some(state.offset) + state.outcome.error().is_none().then_some(state.offset) } /// One past the last frame committed to an unfinished group, when that is past its @@ -2765,7 +2845,7 @@ impl Handle { /// tail, and the count outlives the released cache. pub(super) fn resume_frame(&self) -> Option { let state = self.state.read(); - if (state.fin.is_some() && state.range_end.is_none()) || state.sparse() { + if (state.outcome.frames().is_some() && state.range_end.is_none()) || state.sparse() { return None; } (state.committed() > state.offset).then_some(state.committed()) @@ -2889,7 +2969,7 @@ impl Handle { return false; } let Some(first) = state.entry(0) else { - return state.abort.is_none() && state.held() > 0; + return state.outcome.error().is_none() && state.held() > 0; }; let first = first.frame.payload.len() as u64; state.cache.saturating_sub(first) > first.saturating_mul(ratio.into()) @@ -3016,7 +3096,7 @@ impl Handle { /// The recorded abort reason, or [`Error::Dropped`] if the group closed without one. fn abort_reason(&self) -> Error { - self.state.read().abort.clone().unwrap_or(Error::Dropped) + self.state.read().outcome.error().cloned().unwrap_or(Error::Dropped) } } @@ -3094,7 +3174,7 @@ impl Demand { } fn abort_reason(&self) -> Error { - self.state.read().abort.clone().unwrap_or(Error::Dropped) + self.state.read().outcome.error().cloned().unwrap_or(Error::Dropped) } } @@ -3946,7 +4026,7 @@ impl Consumer { /// how many this consumer has read. The final total once the group is finished. pub fn frame_count(&self) -> usize { let state = self.cursor.state.read(); - state.fin.unwrap_or(state.next_index) + state.outcome.frames().unwrap_or(state.next_index) } /// The next frame, as a consumer for chunked reading. @@ -3999,7 +4079,7 @@ impl Consumer { /// nothing more. A range cap ends only this reader, and can be raised, so it keeps it. fn release_if_finished(&self) { if let Some(expiry) = &self.expiry - && self.cursor.state.read().fin.is_some() + && self.cursor.state.read().outcome.frames().is_some() { expiry.ended(); } @@ -4208,7 +4288,7 @@ impl Consumer { fn shed_next(&self) -> bool { let state = self.cursor.state.read(); let index = self.cursor.index; - if state.abort.is_some() || state.sparse() || index < state.offset { + if state.outcome.error().is_some() || state.sparse() || index < state.offset { return false; } index < state.held() || state.plain(self.cursor.raw).is_some_and(|plain| plain.shed_past(index)) @@ -4367,7 +4447,7 @@ impl Cursor { } let state = self.state.read(); - state.abort.is_none() && state.fin.is_none_or(|fin| self.index < fin) + state.outcome.error().is_none() && state.outcome.frames().is_none_or(|fin| self.index < fin) } /// Content this cursor has neither returned nor already counted in a prefetch batch. @@ -4396,7 +4476,7 @@ impl Cursor { Poll::Ready(match ready!(self.state.poll(waiter, f)) { Ok(res) => res, // We try to clone abort just in case the function forgot to check for terminal state. - Err(state) => Err(state.abort.clone().unwrap_or(Error::Dropped)), + Err(state) => Err(state.outcome.error().cloned().unwrap_or(Error::Dropped)), }) } @@ -4580,7 +4660,7 @@ impl Cursor { Ok(Ok(Some(job))) => self.prefetch.inflate(&job, index, stats)?, Ok(Ok(None)) => {} Ok(Err(err)) => return Poll::Ready(Err(err)), - Err(state) => return Poll::Ready(Err(state.abort.clone().unwrap_or(Error::Dropped))), + Err(state) => return Poll::Ready(Err(state.outcome.error().cloned().unwrap_or(Error::Dropped))), } // The refill already updated the eviction rank under the group lock. @@ -5539,7 +5619,7 @@ mod test { } { let state = producer.handle.state.read(); - assert!(state.abort.is_none()); + assert!(state.outcome.error().is_none()); assert_eq!(state.offset, 0); assert!((48 - 32..48 - 30).contains(&state.held()), "held from {}", state.held()); } @@ -5567,7 +5647,7 @@ mod test { let state = producer.handle.state.read(); assert!(state.held() > 0, "the front was shed"); assert!(state.slots() <= MAX_CACHE_BYTES, "{} bytes of slots", state.slots()); - assert!(state.abort.is_none()); + assert!(state.outcome.error().is_none()); } /// Batches of empty frames count their new slots too, so they shed the front as @@ -5593,7 +5673,7 @@ mod test { "{} bytes of slots", state.slots() ); - assert!(state.abort.is_none()); + assert!(state.outcome.error().is_none()); } /// A frame count past the old 8192-frame cap is held whole while it fits the tail. @@ -5609,7 +5689,7 @@ mod test { let state = producer.handle.state.read(); assert_eq!(state.offset, 0); assert_eq!(state.entries(0, usize::MAX).count(), 10_000); - assert!(state.abort.is_none()); + assert!(state.outcome.error().is_none()); } /// Pool debt sheds a group's front, keeping an open group's newest frame for the diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 457df779de0..06942f1fd2e 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -401,7 +401,6 @@ impl Consumer { reported_stale: RangeSet::new(), declared: None, declared_error: None, - refused: false, }) }); // Subscribed to the serving copy at once, so groups it has now are the reader's @@ -1028,12 +1027,13 @@ impl Candidate { /// A [`Subscriber`]'s state. struct Reader { - /// The first exclusive end any copy declared, refined by a matching mid-group end. + /// The first exclusive end any copy declared, refined by a matching mid-group end. A copy + /// declaring another breaks the track's identity, which hurts only its own broadcast, so + /// nothing checks that copies agree. declared: Option, - /// The first error a copy's end carried, kept with the end so a route switch keeps it. + /// The first error a copy's end carried, from a copy at the end [`Self::declared`] holds, kept + /// with the end so a route switch keeps it. declared_error: Option, - /// Copies of one logical track contradicted its end. - refused: bool, this: Weak>, route: Consumer, /// The reader's own preferences, mirrored into every copy's subscription. @@ -1130,9 +1130,6 @@ impl Reader { fn sync(&mut self, waiter: &kio::Waiter) { // Learn before a swap or retention drops the only copy that knows the end. self.learn_end(None); - if self.refused { - return; - } // Snapshot and re-arm together, including when a change was observed: another // update must wake a reader that parks after applying this one. let mut changed = None; @@ -1276,52 +1273,29 @@ impl Reader { let Some(end) = end else { continue }; match self.declared { None => self.declared = Some(end), - Some(declared) if declared == end => {} // FETCH can name only the group boundary just after a mid-group end. Some(declared) - if declared.final_sequence() == end.final_sequence() && (declared.frame == 0 || end.frame == 0) => + if declared.frame == 0 && end.frame != 0 && declared.final_sequence() == end.final_sequence() => { - if declared.frame == 0 { - self.declared = Some(end); - } + self.declared = Some(end); } - Some(_) => self.refused = true, + Some(_) => {} } - // A copy over a version with no end error declares the same end cleanly. - if self.declared_error.is_none() { + // A copy over a version with no end error declares the same end cleanly. Only a copy + // at the end that stands says how the track ended there. + let same = self.declared.is_some_and(|declared| { + declared == end || end.frame == 0 && end.final_sequence() == declared.final_sequence() + }); + if same && self.declared_error.is_none() { self.declared_error = copy.track.declared_error(); } } - // A group or frame at or past it, handed out or on any copy, claims - // other content. - if let Some(declared) = self.declared { - let end = declared.final_sequence(); - let seen = |sequence: Option| sequence.zip(end).is_some_and(|(sequence, end)| sequence >= end); - let cut = |copy: &Copy| { - declared.frame != 0 - && copy - .track - .peek_group(declared.group) - .is_some_and(|group| group.frame_count() as u64 > declared.frame) - }; - if seen(self.newest) - || self - .copies - .iter() - .any(|copy| seen(copy.track.latest_sequence()) || cut(copy)) - { - self.refused = true; - } - } } - /// Why the reader can read nothing more: copies contradicted the track's end, or its - /// ranges hold an [`Anchor::Oldest`] end. That end is refused here, before waiting on a - /// copy to refuse it, since the demand asked upstream leaves it out. + /// Why the reader can read nothing more: its ranges hold an [`Anchor::Oldest`] end. That + /// end is refused here, before waiting on a copy to refuse it, since the demand asked + /// upstream leaves it out. fn refusal(&self) -> Option { - if self.refused { - return Some(Error::ProtocolViolation); - } self.mirrored .ranges .iter() @@ -1329,15 +1303,6 @@ impl Reader { .then_some(Error::InvalidRange) } - /// Whether `at` is past the declared end, which refuses the track: another copy - /// contradicts the copy that declared it. - fn past_end(&mut self, at: Position) -> bool { - if self.declared.is_some_and(|end| at >= end) { - self.refused = true; - } - self.refused - } - /// Drop the replaced copies nothing needs any more; see [`Copy::retains`]. Whether one /// may still deliver a group the reader lacks, which a clean end waits for. fn let_go(&mut self, delivered: &Delivered, waiter: &kio::Waiter) -> bool { @@ -1475,9 +1440,6 @@ impl Reader { /// eviction can abort that group but cannot substitute a later one and advance the /// logical reader past a gap. fn take(&mut self, candidate: Candidate) -> Result { - if self.past_end(Position::group(candidate.sequence())) { - return Err(Error::ProtocolViolation); - } let index = self .copies .iter() @@ -2954,34 +2916,61 @@ mod test { )); } + /// The first end any copy declares stands, refined only by a mid-group end just before + /// it. A copy contradicting it breaks the track's identity, which hurts only this track: + /// the reader never fails or panics over it. #[kio::test] - async fn copies_must_agree_on_the_declared_end() { - for next in [Position::group(5), Position { group: 4, frame: 3 }, Position::group(6)] { + async fn the_first_declared_end_stands() { + for (next, end) in [ + (Position::group(5), Position::group(5)), + (Position { group: 4, frame: 3 }, Position { group: 4, frame: 3 }), + (Position::group(6), Position::group(5)), + (Position::group(3), Position::group(5)), + ] { let routes = Producer::new(); let logical = logical(&routes); let mut a = copy(); routes.serve(a.consume()); let mut sub = subscribe(&logical, HOUR); + write(&a, 4); + assert_eq!(recv(&mut sub).sequence, 4); a.finish_at(Position::group(5), None).unwrap(); let mut b = copy(); b.finish_at(next, None).unwrap(); routes.serve(b.consume()); - let result = sub.poll_finished(&kio::Waiter::noop()); - if next == Position::group(6) { - assert!(matches!(result, Poll::Ready(Err(Error::ProtocolViolation)))); - assert!(matches!( - sub.poll_recv_group(&kio::Waiter::noop()), - Poll::Ready(Err(Error::ProtocolViolation)) - )); - } else { - assert_eq!(result.map(Result::unwrap), Poll::Ready(next)); - } + assert_eq!( + sub.poll_finished(&kio::Waiter::noop()).map(Result::unwrap), + Poll::Ready(end), + "{next:?}" + ); } } - /// A copy that delivers past an end another copy declared contradicts it. + /// Only a copy at the end that stands says how the track ended: another copy's end, with + /// its error, leaves the first end clean. + #[kio::test] + async fn a_contradicting_copys_end_error_leaves_the_end_clean() { + let routes = Producer::new(); + let logical = logical(&routes); + let mut a = copy(); + routes.serve(a.consume()); + let mut sub = subscribe(&logical, HOUR); + write(&a, 4); + assert_eq!(recv(&mut sub).sequence, 4); + a.finish_at(Position::group(5), None).unwrap(); + let mut b = copy(); + b.finish_at(Position::group(6), Some(Error::App(7))).unwrap(); + routes.serve(b.consume()); + assert_eq!( + sub.poll_finished(&kio::Waiter::noop()).map(Result::unwrap), + Poll::Ready(Position::group(5)) + ); + assert!(sub.end_error().is_none()); + } + + /// A copy holding a group past an end another copy declared never fails the reader. #[kio::test] - async fn content_past_the_declared_end_refuses_the_track() { + async fn content_past_the_declared_end_leaves_the_end() { let routes = Producer::new(); let logical = logical(&routes); let mut a = copy(); @@ -2993,14 +2982,12 @@ mod test { write(&b, 4); assert_eq!(recv(&mut sub).sequence, 4); write(&b, 5); - assert!(matches!( - sub.poll_recv_group(&kio::Waiter::noop()), - Poll::Ready(Err(Error::ProtocolViolation)) - )); - assert!(matches!( - sub.poll_finished(&kio::Waiter::noop()), - Poll::Ready(Err(Error::ProtocolViolation)) - )); + write(&b, 6); + let _ = drain(&mut sub, false); + assert_eq!( + sub.poll_finished(&kio::Waiter::noop()).map(Result::unwrap), + Poll::Ready(Position::group(5)) + ); } /// A mid-group end another copy declared caps the group it falls in, even for frames @@ -3026,54 +3013,6 @@ mod test { assert_eq!(read(&mut reading), Some(b"x".to_vec())); assert_eq!(read(&mut reading), Some(b"x".to_vec())); assert_eq!(read(&mut reading), None); - // The copy holding them contradicts the end. - assert!(matches!( - sub.poll_finished(&kio::Waiter::noop()), - Poll::Ready(Err(Error::ProtocolViolation)) - )); - } - - /// An end below a group a copy handed out or holds claims other content. - #[kio::test] - async fn an_end_below_a_seen_group_refuses_the_track() { - for handed_out in [false, true] { - let routes = Producer::new(); - let logical = logical(&routes); - let a = copy(); - routes.serve(a.consume()); - let mut sub = subscribe(&logical, HOUR); - write(&a, 5); - if handed_out { - assert_eq!(drain(&mut sub, false), [5]); - } - let mut b = copy(); - routes.serve(b.consume()); - b.finish_at(Position::group(3), None).unwrap(); - assert!(matches!( - sub.poll_finished(&kio::Waiter::noop()), - Poll::Ready(Err(Error::ProtocolViolation)) - )); - } - } - - /// A mid-group end learned late, below a frame already seen, claims other content. - #[kio::test] - async fn a_late_mid_group_end_below_a_seen_object_refuses_the_track() { - let routes = Producer::new(); - let logical = logical(&routes); - let a = copy(); - routes.serve(a.consume()); - let mut sub = subscribe(&logical, HOUR); - let mut group = a.create_group(group::Info { sequence: 4 }).unwrap(); - for frame in 0..4 { - group.write_frame(ts(frame), b"x".as_ref()).unwrap(); - } - let _held = (group, recv(&mut sub)); - let mut b = copy(); - routes.serve(b.consume()); - b.finish_at(Position { group: 4, frame: 2 }, None).unwrap(); - let res = sub.poll_finished(&kio::Waiter::noop()); - assert!(matches!(res, Poll::Ready(Err(Error::ProtocolViolation))), "{res:?}"); } /// A gap that stays open across route flaps keeps one backfill per distinct route: a diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 6d4536dfaeb..d9139965e18 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -355,6 +355,176 @@ enum Publication { Published, } +/// How far a track got toward its end; each step replaces the last, and only an abort +/// follows a seal or a drop. +#[derive(Debug, Default)] +enum Outcome { + /// Producing, with no end declared. + #[default] + Open, + /// The end is declared and a producer remains: groups below it may still arrive. + Ending(End), + /// The end is declared and the last producer dropped: nothing more arrives. + Sealed(End), + /// The last producer dropped without declaring an end or aborting. + Dropped { + /// Where production stopped when the open groups went; see + /// [`TrackState::resume_position`]. + resume: Option, + }, + /// Aborted, or refused before anyone published it. + Aborted { + err: Error, + /// The end declared before the abort. Readers reaching it get the clean end only if it + /// had settled first ([`End::settled`]); otherwise a group below it was still open, and + /// they get `err`. + end: Option, + /// Where production stopped when the open groups went. + resume: Option, + /// How long the upstream refusing the request asked to wait before asking again, kept + /// so a relay refuses its own readers with the same interval rather than deriving one. + #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] + retry: Option, + }, +} + +/// A track's declared end. +#[derive(Debug)] +struct End { + /// The exclusive end: no frame at or past it exists. + at: Position, + /// The error the track ended with at `at`, if it did not end cleanly: a reader reaching + /// the end gets it instead of the end of stream, while everything below stays readable. + error: Option, + /// On a received copy, when its readers began waiting out the grace for group streams + /// below the end that QUIC may deliver in any order: when the end or a start was + /// declared, or a group last arrived since. Readers wait for that tail (see + /// [`TrackState::tail_deadline`]) rather than end at the boundary with groups still to + /// arrive. `None` on a local track, which owes no tail. + tail: Option, + /// Whether the end settled before a write that could reopen it or an abort, or + /// `closed()` reported it: every group below it was produced and finished then, so the + /// clean end stands for good. Atomic so reporting it latches through a read guard; see + /// [`TrackState::latch_settled`]. + settled: AtomicBool, +} + +impl End { + fn settled(&self) -> bool { + self.settled.load(Ordering::Relaxed) + } +} + +impl Outcome { + /// The declared end, which an abort keeps. + fn end(&self) -> Option<&End> { + match self { + Self::Ending(end) | Self::Sealed(end) | Self::Aborted { end: Some(end), .. } => Some(end), + Self::Open | Self::Dropped { .. } | Self::Aborted { end: None, .. } => None, + } + } + + fn end_mut(&mut self) -> Option<&mut End> { + match self { + Self::Ending(end) | Self::Sealed(end) | Self::Aborted { end: Some(end), .. } => Some(end), + Self::Open | Self::Dropped { .. } | Self::Aborted { end: None, .. } => None, + } + } + + /// The abort's error. + fn aborted(&self) -> Option<&Error> { + match self { + Self::Aborted { err, .. } => Some(err), + _ => None, + } + } + + /// The abort's error, if it cut the track off before its declared end settled. + fn cut_off(&self) -> Option<&Error> { + match self { + Self::Aborted { err, end, .. } if !end.as_ref().is_some_and(End::settled) => Some(err), + _ => None, + } + } + + /// How long the upstream refusing the track asked to wait before asking again. + #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] + fn retry(&self) -> Option { + match self { + Self::Aborted { retry, .. } => *retry, + _ => None, + } + } + + /// Whether nothing more can arrive: the track was sealed or aborted. + fn is_final(&self) -> bool { + matches!(self, Self::Sealed(_) | Self::Aborted { .. }) + } + + /// Whether no producer remains (sealed, dropped, or aborted), so nothing protects the + /// live edge any longer: the pool's idle expiry reclaims it like every other group, and a + /// stale consumer cannot pin the cache. + fn is_released(&self) -> bool { + !matches!(self, Self::Open | Self::Ending(_)) + } +} + +/// Where a track's live feed starts, as its publisher declared it; each declaration +/// replaces the last. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +enum Start { + /// Nothing declared. + #[default] + Unknown, + /// The floor a subscription asked for (`None` at the live edge), still waiting on the + /// serving session to resolve where the live feed begins (a lite-06+ SUBSCRIBE_START). + /// Readers that must know the resolved start (see [`Subscriber::poll_start`]) wait on it; + /// everything else treats the floor as declared. + Pending(Option), + /// The first group the live feed serves. + Declared(u64), + /// The first group the live feed serves, which its publisher resolved from its + /// subscriber's budget (a lite-06+ SUBSCRIBE_START): it may sit above history a reader + /// can still fill. + Resolved(u64), +} + +impl Start { + /// The first group the live feed serves, or the floor asked while that is pending. + fn sequence(self) -> Option { + match self { + Self::Unknown => None, + Self::Pending(floor) => floor, + Self::Declared(start) | Self::Resolved(start) => Some(start), + } + } +} + +/// Whether a track is a copy received from a remote publisher, and what its readers' +/// relative and time anchors resolve against. Each step replaces the last. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +enum Anchors { + /// The track's own publisher's copy: anchors resolve against its own groups. + #[default] + Local, + /// A received copy holding only part of the track, whose declared end waits for groups + /// still in flight below it (see [`End::tail`]): its readers' relative and time anchors are + /// refused (see [`Holdings::drops_anchors`]). + Refused, + /// A received copy whose wire tells it what relative and time anchors resolve against, + /// so they resolve here as at the publisher. + Wire, + /// A received copy that can translate only time anchors its cache has already settled. A + /// relative one waits for the live edge its wire reports, as under [`Self::Wire`]. + Cached, +} + +impl Anchors { + fn received(self) -> bool { + self != Self::Local + } +} + #[derive(Default)] pub(crate) struct TrackState { // The publisher's properties, once known; always Some for Subscriber/Producer. @@ -447,7 +617,7 @@ pub(crate) struct TrackState { evicted: Option<(u64, Timestamp)>, // The sequence of the newest cached group: the live edge, protected from - // eviction by never entering the eviction order until the track is `closed`. + // eviction by never entering the eviction order until no producer remains. // Tracked separately from `max_sequence`, which a reserved sequence can push past // any cached group. latest_group: Option, @@ -455,21 +625,10 @@ pub(crate) struct TrackState { // Incarnation counter for `Slot::stamp`. next_stamp: u32, - // The exclusive end: no frame at or past it exists. Retained independently of the - // cache, so an evicted last group does not lose where the track ended. - end: Option, - // The error the track ended with at `end`, if it did not end cleanly: a reader reaching - // the end gets it instead of the end of stream, while everything below stays readable. - end_error: Option, - - // Nothing more can arrive: the last producer dropped after a declared boundary. - sealed: bool, - - // Set once a received copy's end is declared, ahead of group streams below it that QUIC - // may deliver in any order: when the end or a start was declared, or a group last arrived since. - // Readers wait out the grace for that tail (see [`Self::tail_deadline`]), rather than - // end at the boundary with groups still to arrive. - tail: Option, + // How far the track got toward its end: declared, sealed, dropped, or aborted, with + // what each of those phases knows. Retained independently of the cache, so an evicted + // last group does not lose where the track ended. + outcome: Outcome, // The sequences the live feed delivered: what a pending tail // is judged by. Fetched copies never count, and eviction never touches it. A gap folds @@ -477,37 +636,15 @@ pub(crate) struct TrackState { // does, so this holds only the gaps still worth waiting for. arrivals: Ledger, - // No producer remains (aborted, sealed, or dropped), so nothing protects the live - // edge any longer: the pool's idle expiry reclaims it like every other group, and a - // stale consumer cannot pin the cache. - closed: bool, - - // The first sequence the live feed serves, once the publisher declared one - // (the wire's SUBSCRIBE_START). Lower groups are not waited for, though one - // may still arrive late or be fetched. - start_sequence: Option, - - // Whether `start_sequence` is only the floor a subscription asked for, still - // waiting on the serving session to resolve where the live feed begins (a - // lite-06+ SUBSCRIBE_START). Readers that must know the resolved start (see - // [`Subscriber::poll_start`]) wait on it; everything else treats the floor as usual. - start_pending: bool, - - // Whether the publisher resolved `start_sequence` (a lite-06+ SUBSCRIBE_START), which - // may sit above history a reader can still fill. Cleared by any other declaration. - start_resolved: bool, - - // A copy received from a remote publisher, whose declared end waits for groups still - // in flight below it (see `tail`), and whose readers' relative and time anchors are - // refused unless `anchored` (see `Holdings::drops_anchors`). - received: bool, - // Whether a received copy's wire tells it what relative and time anchors resolve - // against, so they resolve here as at the publisher. - anchored: bool, - // This copy can translate only time anchors its cache has already settled. A relative one - // waits for the live edge its wire reports, as on an `anchored` copy. - cached_anchors: bool, - // Whether readers ask no fills below `start_sequence`: its publisher resolved it from the + // The first sequence the live feed serves, as the publisher declared it (the wire's + // SUBSCRIBE_START). Lower groups are not waited for, though one may still arrive late + // or be fetched. + start: Start, + + // Whether this is a copy received from a remote publisher, and what its readers' + // relative and time anchors resolve against there. + anchors: Anchors, + // Whether readers ask no fills below `start`: its publisher resolved it from the // budget, but neither this copy's timestamps nor a fill can judge an older group against // a reader's budget. A widened budget reaches lower groups through the feed's re-offer. fills_from_start: bool, @@ -559,24 +696,6 @@ pub(crate) struct TrackState { // below this group is unjudgeable for readers with a finite delay budget. live_floor: Option, - // Where production stopped, snapshotted when the open groups are released (an - // abort, or the last producer dropping). Computed live from the cache otherwise; - // see [`Self::resume_position`]. - resume: Option, - - // The error that caused the track to be aborted, if any. - abort: Option, - // How long the upstream refusing the request asked to wait before asking again, kept so - // a relay refuses its own readers with the same interval rather than deriving one. - #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] - retry: Option, - - // Whether the declared end settled before a write that could reopen it or an abort, or - // `closed()` reported it: every group below it was produced and finished then, so the - // clean end stands for good. Atomic so reporting it latches through a read guard. See - // [`Self::latch_settled`]. - settled: AtomicBool, - // Active subscriptions, in their own [`kio::Lock`] so a read-only `Consumer` // registers under that lock instead of writing back into the track state. // Kept here (rather than threaded through every handle) so any holder reaches it. @@ -837,12 +956,12 @@ impl Holdings for TrackState { } fn drops_anchors(&self) -> bool { - self.received && !self.anchored + self.anchors == Anchors::Refused } fn looks_up(&self) -> bool { // A received copy's handler is its session, which fetches by group alone. - !self.received && self.fills.read().has_handlers() + !self.anchors.received() && self.fills.read().has_handlers() } fn containing(&self, pts: Timestamp) -> Containing { @@ -895,7 +1014,7 @@ impl Holdings for TrackState { return dropped; } // Without a declared start, the first arrival stands in for it, as the tail's does. - let floor = self.start_sequence.or(self.arrivals.first()).unwrap_or(0); + let floor = self.start.sequence().or(self.arrivals.first()).unwrap_or(0); if sequence < floor { return Some(floor); } @@ -903,12 +1022,12 @@ impl Holdings for TrackState { } fn complete(&self) -> bool { - self.sealed || self.abort.is_some() + self.outcome.is_final() } fn changes(&self, seen: &mut Seen) -> Option>> { let global = [ - self.start_sequence.map_or(0, |start| start.saturating_add(1)), + self.start.sequence().map_or(0, |start| start.saturating_add(1)), u64::from(self.complete()), self.fetched, // A handler appearing or leaving changes which starts count and what is gone. @@ -1319,7 +1438,7 @@ impl TrackState { fn poll_info(&self) -> Poll> { if let Some(info) = self.info.as_ref().filter(|_| self.publication != Publication::Staged) { Poll::Ready(Ok(info.clone())) - } else if let Some(err) = &self.abort { + } else if let Some(err) = self.outcome.aborted() { // Aborted before anyone served it, so the info can never arrive: fail the // waiting subscribes instead of parking them on a track nobody will fill. Poll::Ready(Err(err.clone())) @@ -1332,7 +1451,7 @@ impl TrackState { /// reached its end, which makes the cache the whole of it. An end declared but not /// reached does not: the groups before it are still to come from the feed. fn readable(&self) -> bool { - self.feed.is_live() || self.sealed || self.abort.is_some() || self.is_complete() + self.feed.is_live() || self.outcome.is_final() || self.is_complete() } /// Where relative starts count back from: the live feed's newest group, produced or @@ -1398,7 +1517,7 @@ impl TrackState { // TODO once we have drop notifications, check if index == final_sequence. if self.is_complete() { Poll::Ready(Ok(None)) - } else if let Some(err) = &self.abort { + } else if let Some(err) = self.outcome.aborted() { Poll::Ready(Err(err.clone())) } else { Poll::Pending @@ -1423,12 +1542,12 @@ impl TrackState { // Nothing more can arrive once the track was sealed (dropped with its end // declared) or aborted. Either closed the channel, so parking is not an option: // the consumer would be handed the abort, or `Dropped`, instead of how it ended. - let closed = self.sealed || self.abort.is_some(); + let closed = self.outcome.is_final(); // Once nothing more is in range: an abort before the end settled cut the track // off. One after it is a clean end (see `is_complete`), as is reaching the end. - let end = || match &self.abort { - Some(err) if !self.settled.load(Ordering::Relaxed) => Err(err.clone()), - _ => Ok(None), + let end = || match self.outcome.cut_off() { + Some(err) => Err(err.clone()), + None => Ok(None), }; // If the exclusive end is already at or below where we'd resume, no @@ -1908,7 +2027,7 @@ impl TrackState { return Poll::Ready(Ok(consumer)); } - if let Some(err) = &self.abort { + if let Some(err) = self.outcome.aborted() { return Poll::Ready(Err(err.clone())); } @@ -2103,16 +2222,13 @@ impl TrackState { /// Whether `sequence` is the live edge, which eviction and expiry never take /// while a producer remains. fn protects(&self, sequence: u64) -> bool { - !self.closed && Some(sequence) == self.latest_group + !self.outcome.is_released() && Some(sequence) == self.latest_group } /// No producer remains: stop protecting the live edge, so the pool's idle expiry /// reclaims every group once readers stop touching it, and a stale consumer can't /// pin the cache (and its frame buffers) forever. fn close_cache(&mut self) { - if std::mem::replace(&mut self.closed, true) { - return; - } self.cache.end_largest(); if let Some(latest) = self.latest_group && let Some(slot) = self.lookup.get(&latest) @@ -2125,12 +2241,12 @@ impl TrackState { /// End the track abruptly with `err`: drop the open groups nobody will finish, keep the /// finished ones for readers still draining, and stop protecting the live edge. /// - /// Returns every unfinished copy, cached or not, to abort once the track lock is released, - /// so a reader that already pulled one ends with `err` too and its writer fails. - fn tear_down(&mut self, err: Error) -> OpenGroups { - // Snapshot the frame boundary before the open group it may sit in goes away: an - // abrupt end is exactly when a replacement route asks where to resume. - self.resume = self.resume_position(); + /// Returns where production stopped, snapshotted before the open group it may sit in + /// goes (an abrupt end is exactly when a replacement route asks where to resume), and + /// every unfinished copy, cached or not, to abort once the track lock is released, so a + /// reader that already pulled one ends with `err` too and its writer fails. + fn tear_down(&mut self, err: Error) -> (Option, OpenGroups) { + let resume = self.resume_position(); let groups = self .lookup .values() @@ -2164,8 +2280,7 @@ impl TrackState { .get_mut() .unwrap_or_else(PoisonError::into_inner) .retain(|sequence, _| lookup.contains_key(sequence)); - self.close_cache(); - OpenGroups { groups, err } + (resume, OpenGroups { groups, err }) } /// Take `sequence` out of the cache, along with its checked start. @@ -2186,16 +2301,14 @@ impl TrackState { /// Record that the live feed delivered `sequence`, then fold the gaps that aged out. fn record_arrival(&mut self, sequence: u64, now: crate::time::Instant) { self.arrivals.arrive(sequence, now); - if !self.received + if !self.anchors.received() && self.appended.is_none_or(|appended| sequence >= appended) && let Some(end) = sequence.checked_add(1) { self.produced.insert(sequence..end); } // A feed still delivering keeps the gap above its last arrival open. - if self.tail.is_some() { - self.tail = Some(now); - } + self.refresh_tail(now); self.fold_arrivals(now); } @@ -2217,6 +2330,22 @@ impl TrackState { ledger::grace(max_delay) } + /// Refuse a request nobody published with `err`, waking the subscribes waiting on it. + /// No producer ever wrote to it, so there is nothing to tear down, but a fill answered + /// before the refusal may have cached the live edge, which nothing protects anymore. + fn reject(&mut self, err: Error, retry: Option) { + let released = self.outcome.is_released(); + self.outcome = Outcome::Aborted { + err, + end: None, + resume: None, + retry, + }; + if !released { + self.close_cache(); + } + } + /// Attach the publisher's immutable metadata without replacing it with local cache policy. fn install(&mut self, info: Info) { // A replaced max age moves every parked read's budget. @@ -2591,7 +2720,7 @@ impl TrackState { { return Poll::Ready(Some(slot.serving().consume())); } - match self.start_sequence.is_some_and(|start| start > sequence) + match self.start.sequence().is_some_and(|start| start > sequence) || self.final_sequence().is_some_and(|fin| fin <= sequence) { true => Poll::Ready(None), @@ -2664,7 +2793,7 @@ impl TrackState { true => groups.start, false => groups.start.max(self.arrivals.first()?), }; - let start = start.max(self.start_sequence.unwrap_or(0)); + let start = start.max(self.start.sequence().unwrap_or(0)); let end = groups.end.min(self.final_sequence().unwrap_or(u64::MAX)); let now = self.cache.pool().now_wall(); for (gap, opened) in self.arrivals.gaps(start..end) { @@ -2684,10 +2813,10 @@ impl TrackState { /// group accounted for by its arrivals (since evicted or aborted), dropped, /// outside the feed, or in a gap its grace stopped waiting for. A local track owes none. fn owed(&self, groups: ops::Range, window: &Window) -> Option { - if !self.received || self.closed || !self.readable() { + if !self.anchors.received() || self.outcome.is_released() || !self.readable() { return None; } - let asked = self.start_sequence.is_some(); + let asked = self.start.sequence().is_some(); self.find_landable(groups, asked, self.arrival_grace(), |open, _| { window.clip(open).map(|run| run.start) }) @@ -2701,18 +2830,21 @@ impl TrackState { /// A declared start asks the feed for every group from it, so a gap in the arrivals at /// or above it, and a pending tail above the last arrival, is waited for a fresh grace: /// the feed may only now be sending it. - fn set_start(&mut self, start_sequence: Option, pending: bool) { + fn set_start(&mut self, start: Start) { self.latch_settled(); - self.start_sequence = start_sequence; - self.start_pending = pending; - self.start_resolved = false; - if let Some(start) = start_sequence { + self.start = start; + if let Some(start) = start.sequence() { self.cache.wakes().arrived(..start); let now = self.cache.pool().now_wall(); self.arrivals.demand(start..u64::MAX, now); - if self.tail.is_some() { - self.tail = Some(now); - } + self.refresh_tail(now); + } + } + + /// Restart a received copy's tail grace from `now`, once it is waiting one out. + fn refresh_tail(&mut self, now: crate::time::Instant) { + if let Some(tail) = self.outcome.end_mut().and_then(|end| end.tail.as_mut()) { + *tail = now; } } @@ -2720,9 +2852,21 @@ impl TrackState { /// already-produced groups, or one inside a group that already holds a frame past it. /// A mid-group end caps that group, so no later write can land past it either. fn set_end(&mut self, end: Position, error: Option) -> Result<()> { - if self.end.is_some() { + if !matches!(self.outcome, Outcome::Open) { return Err(Error::Closed); } + self.declare(end, error, None, false) + } + + /// Declare `end` on an open track, or one whose declared end it refines (the declared end + /// already taken out of `outcome`), carrying over that end's `tail` and settling. + fn declare( + &mut self, + end: Position, + error: Option, + tail: Option, + settled: bool, + ) -> Result<()> { let fin = end.final_sequence().ok_or(coding::BoundsExceeded)?; if let Some(max) = self.max_sequence && fin <= max @@ -2742,16 +2886,25 @@ impl TrackState { } } } - self.end = Some(end); - self.end_error = error; + self.outcome = Outcome::Ending(End { + at: end, + error, + tail, + settled: AtomicBool::new(settled), + }); self.cache.declare_end(); self.cache.wakes().arrived(fin..); Ok(()) } + /// The declared end's position. + fn declared(&self) -> Option { + self.outcome.end().map(|end| end.at) + } + /// What a reader past the end gets: the end of stream, or the error the track ended with. fn ended(&self) -> Result> { - match &self.end_error { + match self.outcome.end().and_then(|end| end.error.as_ref()) { Some(err) => Err(err.clone()), None => Ok(None), } @@ -2762,17 +2915,17 @@ impl TrackState { if !self.is_complete() { return None; } - Some((self.final_sequence()?, self.end_error.clone()?)) + Some((self.final_sequence()?, self.outcome.end()?.error.clone()?)) } /// The first group the end excludes whole: one past the last group that may exist. fn final_sequence(&self) -> Option { - self.end.and_then(Position::final_sequence) + self.declared().and_then(Position::final_sequence) } /// Cap a new group the end cuts, so it never holds a frame past the end. fn cap(&self, group: &group::Handle) -> Result<()> { - match self.end { + match self.declared() { Some(end) if end.frame != 0 && end.group == group.sequence => group.limit(Some(end.frame)), _ => Ok(()), } @@ -2788,19 +2941,23 @@ impl TrackState { /// An abort before the end settled wins over it: a group below the boundary was /// still open, so the track was cut off rather than ended. fn is_complete(&self) -> bool { - // `sealed` is a declared end the last producer dropped short of. An abort still wins - // unless that end had already settled: a group below it was still open. - let reached = self.sealed - || self + match &self.outcome { + Outcome::Ending(end) => end + .at .final_sequence() - .is_some_and(|fin| self.reaches(fin) && !self.awaits_tail(fin)); - reached && (self.abort.is_none() || self.settled.load(Ordering::Relaxed)) + .is_some_and(|fin| self.reaches(fin) && !self.awaits_tail(fin)), + // A declared end the last producer dropped short of. + Outcome::Sealed(_) => true, + // An abort wins unless the end had already settled: a group below it was still open. + Outcome::Aborted { end, .. } => end.as_ref().is_some_and(End::settled), + Outcome::Open | Outcome::Dropped { .. } => false, + } } /// Whether nothing more reaches the cache but what fills answer: the track was aborted, /// reached its end, or its session holds no subscription that feeds it. pub(super) fn nothing_arrives(&self) -> bool { - self.abort.is_some() || self.is_complete() || self.unfed + self.outcome.aborted().is_some() || self.is_complete() || self.unfed } /// Whether a bounded window reads from the cache: whenever readers may, and on a copy no @@ -2815,7 +2972,9 @@ impl TrackState { /// misses up to `fin` is waited for only as long as the grace. fn reaches(&self, fin: u64) -> bool { let edge = self.edge(); - edge >= fin || self.tail.is_some() || self.drops.gaps(&(edge..fin)).next().is_none() + edge >= fin + || self.outcome.end().is_some_and(|end| end.tail.is_some()) + || self.drops.gaps(&(edge..fin)).next().is_none() } /// Whether the largest can no longer advance while a producer is still held: the track @@ -2835,7 +2994,7 @@ impl TrackState { /// Only such a hole holds readers, so a tail with nothing missing ends them at once. fn awaits_tail(&self, fin: u64) -> bool { // Nothing arrives after an abort, so a hole left then is final. - self.abort.is_none() && self.owes(fin) + self.outcome.aborted().is_none() && self.owes(fin) } /// Whether the tail misses a sequence from where the feed starts up to `fin` that is @@ -2857,22 +3016,22 @@ impl TrackState { /// end declared ahead of it is not cut off. Judged by the arrivals, not the cache, so an evicted group /// counts as delivered, and the check costs the same however much is cached. fn tail_deadline(&self, fin: u64) -> Option> { - let active = self.tail?; + let active = self.outcome.end()?.tail?; let grace = self.arrival_grace(); // Fetched backfill never reaches a reader in arrival order, so only the live feed's // arrivals count. Without a declared start, the first one stands in for it. - let floor = match self.start_sequence { - Some(start) => start, + let floor = match self.start { + Start::Declared(start) | Start::Resolved(start) | Start::Pending(Some(start)) => start, // Where the feed starts is still unknown, so everything below the end not yet // delivered or dropped is owed. - None if self.start_pending => { + Start::Pending(None) => { let missing = self .arrivals .gaps(0..fin) .any(|(gap, _)| self.drops.gaps(&gap).next().is_some()); return missing.then(|| active.checked_add(grace)); } - None => self.arrivals.first().unwrap_or(0), + Start::Unknown => self.arrivals.first().unwrap_or(0), }; // Nothing opened the gap above the last arrival, so its grace runs from `active`. self.arrivals @@ -2899,7 +3058,7 @@ impl TrackState { let Some(fin) = self.final_sequence() else { return false; }; - let reached = self.sealed || self.reaches(fin); + let reached = matches!(self.outcome, Outcome::Sealed(_)) || self.reaches(fin); reached && !self.owes(fin) && self.lookup.range(..fin).all(|(_, slot)| slot.is_settled()) } @@ -2908,8 +3067,11 @@ impl TrackState { /// then stands, and a later abort ends readers cleanly. A late group or a wider /// subscription is still served; it just cannot turn the outcome into an error. fn latch_settled(&mut self) { - if self.abort.is_none() && self.is_settled() { - self.settled.store(true, Ordering::Relaxed); + if let Outcome::Ending(end) | Outcome::Sealed(end) = &self.outcome + && !end.settled() + && self.is_settled() + { + end.settled.store(true, Ordering::Relaxed); } } @@ -2921,8 +3083,12 @@ impl TrackState { fn resume_position(&self) -> Option { // A snapshot taken when the open groups were released wins; the group it was // derived from may be gone. - if self.resume.is_some() { - return self.resume; + if let Outcome::Dropped { resume: Some(resume) } + | Outcome::Aborted { + resume: Some(resume), .. + } = self.outcome + { + return Some(resume); } let max = self.latest_group?; @@ -3095,9 +3261,9 @@ impl TrackState { } fn poll_finished(&self) -> Poll> { - if let Some(end) = self.end { + if let Some(end) = self.declared() { Poll::Ready(Ok(end)) - } else if let Some(err) = &self.abort { + } else if let Some(err) = self.outcome.aborted() { Poll::Ready(Err(err.clone())) } else { Poll::Pending @@ -3105,7 +3271,9 @@ impl TrackState { } fn modify(producer: &kio::Producer) -> Result> { - producer.write().map_err(|r| r.abort.clone().unwrap_or(Error::Dropped)) + producer + .write() + .map_err(|r| r.outcome.aborted().cloned().unwrap_or(Error::Dropped)) } /// Observe a shared track's abort, or every producer leaving, without extending its @@ -3114,7 +3282,7 @@ impl TrackState { pub(crate) fn poll_aborted(producer: &kio::Producer, waiter: &kio::Waiter) -> Poll { producer .poll_closed(waiter) - .map(|()| producer.read().abort.clone().unwrap_or(Error::Dropped)) + .map(|()| producer.read().outcome.aborted().cloned().unwrap_or(Error::Dropped)) } /// The track's outcome once final: `Ok` once its declared end is reached with every @@ -3124,7 +3292,7 @@ impl TrackState { /// A clean outcome latches once reported: a received copy's tail grace is judged when /// polled, and a later subscription asking for a longer one must not reopen it. fn poll_outcome(&self, waiter: &kio::Waiter) -> Poll> { - if self.settled.load(Ordering::Relaxed) { + if self.outcome.end().is_some_and(End::settled) { return Poll::Ready(self.ended().map(|_: Option<()>| ())); } if self.is_complete() { @@ -3132,7 +3300,7 @@ impl TrackState { // may still abort, which would cut the track off, so wait for each slot to settle // ([`Slot::is_settled`]). One whose last serving copy aborts leaves the outcome to // the track's own abort or drop. - if self.abort.is_none() + if self.outcome.aborted().is_none() && let Some(fin) = self.final_sequence() { for (_, slot) in self.lookup.range(..fin).filter(|(_, slot)| !slot.is_settled()) { @@ -3146,9 +3314,11 @@ impl TrackState { } } } - self.settled.store(true, Ordering::Relaxed); + if let Some(end) = self.outcome.end() { + end.settled.store(true, Ordering::Relaxed); + } Poll::Ready(self.ended().map(|_: Option<()>| ())) - } else if let Some(err) = &self.abort { + } else if let Some(err) = self.outcome.aborted() { Poll::Ready(Err(err.clone())) } else { self.watch_tail(waiter); @@ -3162,14 +3332,14 @@ impl TrackState { Poll::Ready(match ready!(poll) { Ok(outcome) => outcome, Err(closed) if closed.is_complete() => closed.ended().map(|_: Option<()>| ()), - Err(closed) => Err(closed.abort.clone().unwrap_or(Error::Dropped)), + Err(closed) => Err(closed.outcome.aborted().cloned().unwrap_or(Error::Dropped)), }) } /// Abort shared track state without extending its publisher lifetime. #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] pub(crate) fn abort(producer: &kio::Producer, err: Error) -> Result<()> { - commit_abort(Self::modify(producer)?, err).abort(); + commit_abort(Self::modify(producer)?, err, None).abort(); Ok(()) } @@ -3198,7 +3368,7 @@ impl TrackState { frames: group::Bounds, info: Option, ) -> Result { - if let Some(err) = &self.abort { + if let Some(err) = self.outcome.aborted() { return Err(err.clone()); } if let Some(fin) = self.final_sequence() @@ -3288,7 +3458,7 @@ impl TrackState { frames: group::Bounds, info: Option, ) -> Result { - if let Some(err) = &self.abort { + if let Some(err) = self.outcome.aborted() { return Err(err.clone()); } if self.info.is_none() { @@ -3448,17 +3618,30 @@ impl Deferred { /// Record `err` and close the track, the shared tail of every abort, returning the open /// groups to abort once the guard is gone. -fn commit_abort(mut state: kio::Mut<'_, TrackState>, err: Error) -> OpenGroups { +fn commit_abort(mut state: kio::Mut<'_, TrackState>, err: Error, retry: Option) -> OpenGroups { // Decided before the open groups go: the groups below the end are the evidence. A gap // past the grace was lost before the abort rather than cut off by it. let now = state.cache.pool().now_wall(); state.fold_arrivals(now); state.latch_settled(); - state.abort = Some(err.clone()); // Keep what finished for consumers still draining: they get it, then the abort (or // the clean end, when the end settled). Clearing here would abort finished groups a // slower reader has not pulled yet. The pool's idle expiry bounds how long they stay. - let open = state.tear_down(err); + let released = state.outcome.is_released(); + let (resume, open) = state.tear_down(err.clone()); + let end = match std::mem::take(&mut state.outcome) { + Outcome::Ending(end) | Outcome::Sealed(end) => Some(end), + _ => None, + }; + state.outcome = Outcome::Aborted { + err, + end, + resume, + retry, + }; + if !released { + state.close_cache(); + } state.close(); open } @@ -3522,7 +3705,7 @@ fn poll_largest( /// end that is not reached yet does not count: an idle copy's largest is still only what /// its ended feed left. fn largest_live(state: &TrackState, waiter: &kio::Waiter) -> Poll<()> { - match state.feed.is_live() || state.sealed || state.abort.is_some() || state.largest_settled() { + match state.feed.is_live() || state.outcome.is_final() || state.largest_settled() { true => Poll::Ready(()), false => { state.watch_tail(waiter); @@ -3665,7 +3848,8 @@ impl Producer { fn create_group_inner(&self, info: group::Info, received: Option) -> Result { let mut state = self.modify()?; - if state.sealed || state.final_sequence().is_some_and(|fin| info.sequence >= fin) { + if matches!(state.outcome, Outcome::Sealed(_)) || state.final_sequence().is_some_and(|fin| info.sequence >= fin) + { return Err(Error::Closed); } // A skip says the sequence will never be produced, so it is final. A drop is not: a @@ -3810,26 +3994,33 @@ impl Producer { /// refined end must carry the same error code. pub fn finish_at(&mut self, end: Position, error: Option) -> Result<()> { let mut state = self.modify()?; - let code = |err: &Option| err.as_ref().map(|err| StreamError::from(err).to_code()); - if state.end.is_some() && code(&state.end_error) != code(&error) { - return Err(Error::ProtocolViolation); - } - match state.end { - Some(declared) if declared == end => return Ok(()), - Some(declared) if declared.frame == 0 && end.frame != 0 && end.final_sequence() == Some(declared.group) => { - state.end = None; - if let Err(err) = state.set_end(end, error) { - state.end = Some(declared); - return Err(err); + let code = |err: Option<&Error>| err.map(|err| StreamError::from(err).to_code()); + // A received copy's readers wait out a grace for groups still in flight below its end. + let tail = state.anchors.received().then(|| state.cache.pool().now_wall()); + match std::mem::take(&mut state.outcome) { + Outcome::Open => state.declare(end, error, tail, false), + Outcome::Ending(declared) if code(declared.error.as_ref()) != code(error.as_ref()) => { + state.outcome = Outcome::Ending(declared); + Err(Error::ProtocolViolation) + } + Outcome::Ending(declared) if declared.at == end => { + state.outcome = Outcome::Ending(declared); + Ok(()) + } + Outcome::Ending(declared) + if declared.at.frame == 0 && end.frame != 0 && end.final_sequence() == Some(declared.at.group) => + { + let refined = state.declare(end, error, declared.tail.or(tail), declared.settled()); + if refined.is_err() { + state.outcome = Outcome::Ending(declared); } + refined + } + outcome => { + state.outcome = outcome; + Err(Error::ProtocolViolation) } - Some(_) => return Err(Error::ProtocolViolation), - None => state.set_end(end, error)?, - } - if state.received && state.tail.is_none() { - state.tail = Some(state.cache.pool().now_wall()); } - Ok(()) } /// Declare the first group the live feed serves (the wire's SUBSCRIBE_START, @@ -3845,7 +4036,8 @@ impl Producer { /// /// Resolves a start a subscriber session left pending. pub fn start_at(&mut self, sequence: impl Into>) -> Result<()> { - self.modify()?.set_start(sequence.into(), false); + let start = sequence.into().map_or(Start::Unknown, Start::Declared); + self.modify()?.set_start(start); Ok(()) } @@ -3854,9 +4046,7 @@ impl Producer { /// fillable, and a fill that finds none ends empty. #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] pub(crate) fn start_resolved(&mut self, sequence: u64) -> Result<()> { - let mut state = self.modify()?; - state.set_start(Some(sequence), false); - state.start_resolved = true; + self.modify()?.set_start(Start::Resolved(sequence)); Ok(()) } @@ -3868,8 +4058,10 @@ impl Producer { /// unless its session's wire lets it resolve them, since it holds only part of the track. #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] pub(crate) fn received(&self) { - if let Ok(mut state) = self.modify() { - state.received = true; + if let Ok(mut state) = self.modify() + && state.anchors == Anchors::Local + { + state.anchors = Anchors::Refused; } } @@ -3878,7 +4070,7 @@ impl Producer { #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] pub(crate) fn anchored(&self) { if let Ok(mut state) = self.modify() { - state.anchored = true; + state.anchors = Anchors::Wire; } } @@ -3887,8 +4079,7 @@ impl Producer { #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] pub(crate) fn anchored_cached(&self) { if let Ok(mut state) = self.modify() { - state.anchored = true; - state.cached_anchors = true; + state.anchors = Anchors::Cached; } } @@ -4178,7 +4369,7 @@ impl Producer { /// subscription asks for again. #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] pub(crate) fn request_start(&mut self, sequence: Option) -> Result<()> { - self.modify()?.set_start(sequence, true); + self.modify()?.set_start(Start::Pending(sequence)); Ok(()) } @@ -4187,7 +4378,7 @@ impl Producer { #[cfg(test)] #[cfg(feature = "async")] pub(crate) fn start_sequence(&self) -> Option { - self.state.read().start_sequence + self.state.read().start.sequence() } /// Evict group `sequence` from the cache, as expiry does. @@ -4201,7 +4392,7 @@ impl Producer { /// `None` while the track is still open ended. Both methods reject a second end, so /// callers that may have already declared one check here first. pub fn finished(&self) -> Option { - self.state.read().end + self.state.read().declared() } /// Abort the track with the given error. @@ -4222,7 +4413,7 @@ impl Producer { pub fn abort(self, err: Error) -> Result<()> { // Unlisted first, so the track leaves the listing before its name is free again. self.alive.unlist(false); - commit_abort(self.modify()?, err).abort(); + commit_abort(self.modify()?, err, None).abort(); Ok(()) } @@ -4231,9 +4422,7 @@ impl Producer { #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] pub(crate) fn refuse(self, err: Error, retry: Duration) -> Result<()> { self.alive.unlist(false); - let mut state = self.modify()?; - state.retry = Some(retry); - commit_abort(state, err).abort(); + commit_abort(self.modify()?, err, Some(retry)).abort(); Ok(()) } @@ -4258,7 +4447,7 @@ impl Producer { pub(crate) fn abort_unwanted(self, err: Error) -> std::result::Result<(), Self> { let wanted = match self.state.write_unwanted() { kio::Unwanted::Idle(guard) => { - commit_abort(guard, err).abort(); + commit_abort(guard, err, None).abort(); false } kio::Unwanted::Closed => false, @@ -4422,7 +4611,7 @@ impl Producer { // Surface an abort as the stream ending. `poll_closed` parks on the closed // waiters, so per-group churn on the track state never wakes this poll. if self.state.poll_closed(waiter).is_ready() { - let abort = self.state.read().abort.clone(); + let abort = self.state.read().outcome.aborted().cloned(); return Poll::Ready(Err(abort.unwrap_or(Error::Dropped))); } @@ -4619,7 +4808,7 @@ impl Drop for Alive { fn drop(&mut self) { // Withdrawn before the state closes, so the track leaves the listing before a // later one can take its name. It finished if its producer declared its end. - let finished = self.state.read().end.is_some(); + let finished = self.state.read().outcome.end().is_some(); self.unlist(finished); // A request nobody accepted was never publishing; there's nothing to tear down. if !self.published.load(Ordering::Relaxed) { @@ -4632,27 +4821,28 @@ impl Drop for Alive { // `abort()` closes the channel, so `write()` returns `Err(Ref)`. `finish()` // leaves it open with `end` set, so inspect both outcomes. match self.state.write() { - Ok(mut state) => { - if state.end.is_some() { - // Groups still missing below the boundary can no longer arrive, so a - // reader waiting on one ends cleanly instead of with `Dropped`. - state.sealed = true; + Ok(mut state) => match std::mem::take(&mut state.outcome) { + Outcome::Open => { + tracing::warn!( + track = %self.name, + "track::Producer dropped without finish() or abort()" + ); + let (resume, open) = state.tear_down(Error::Dropped); + state.outcome = Outcome::Dropped { resume }; state.close_cache(); - return; + drop(state); + open.abort(); } - if state.abort.is_some() { - return; + // Groups still missing below the boundary can no longer arrive, so a reader + // waiting on one ends cleanly instead of with `Dropped`. + Outcome::Ending(end) => { + state.outcome = Outcome::Sealed(end); + state.close_cache(); } - tracing::warn!( - track = %self.name, - "track::Producer dropped without finish() or abort()" - ); - let open = state.tear_down(Error::Dropped); - drop(state); - open.abort(); - } + outcome => state.outcome = outcome, + }, Err(state) => { - if state.sealed || state.end.is_some() || state.abort.is_some() { + if !matches!(state.outcome, Outcome::Open) { return; } tracing::warn!( @@ -4763,10 +4953,10 @@ impl TrackWeak { let Ok(mut state) = producer.write() else { return false; }; - if state.publication != Publication::Pending || state.abort.is_some() { + if state.publication != Publication::Pending || state.outcome.aborted().is_some() { return false; } - state.abort = Some(err); + state.reject(err, None); state.close(); true } @@ -4783,12 +4973,12 @@ impl TrackWeak { let Ok(state) = producer.write() else { return; }; - commit_abort(state, err).abort(); + commit_abort(state, err, None).abort(); } /// Whether its producer declared the track's end, which is how its listing ends too. pub(super) fn is_finished(&self) -> bool { - self.state.read().end.is_some() + self.state.read().outcome.end().is_some() } /// Whether anyone is consuming the track right now. A closed track doesn't @@ -4820,7 +5010,7 @@ impl TrackWeak { }; match producer.write_unwanted() { kio::Unwanted::Idle(guard) => { - commit_abort(guard, err).abort(); + commit_abort(guard, err, None).abort(); true } kio::Unwanted::Closed => true, @@ -4989,7 +5179,7 @@ impl Demand { /// The recorded abort reason, or [`Error::Dropped`] if the track closed without one. fn abort_reason(&self) -> Error { - self.state.read().abort.clone().unwrap_or(Error::Dropped) + self.state.read().outcome.aborted().cloned().unwrap_or(Error::Dropped) } } @@ -5096,7 +5286,7 @@ impl ArrivalRead<'_> { } }; let closed = self.state.is_closed(); - let start = state.start_sequence.unwrap_or(0); + let start = state.start.sequence().unwrap_or(0); let fin = state.final_sequence().unwrap_or(u64::MAX); for hole in miss::holes(state, groups) { if closed { @@ -5171,13 +5361,13 @@ impl Consumer { /// An already declared end, without resolving a pending subscription or registering a wake. pub(super) fn declared_end(&self) -> Option { - self.with_state(|state| state?.read().end) + self.with_state(|state| state?.read().declared()) } /// The error the declared end carries, reached or not. #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] pub(super) fn declared_error(&self) -> Option { - self.with_state(|state| state?.read().end_error.clone()) + self.with_state(|state| state?.read().outcome.end()?.error.clone()) } fn new(name: Arc, state: kio::ConsumerWeak) -> Self { @@ -5295,13 +5485,13 @@ impl Consumer { /// whatever it last declared, since nothing will resolve it anymore. #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] fn poll_state_start(state: &kio::Consumer, waiter: &kio::Waiter) -> Poll> { - let res = state.poll(waiter, |state| match state.start_pending && state.abort.is_none() { - true => Poll::Pending, - false => Poll::Ready(state.start_sequence), + let res = state.poll(waiter, |state| match state.start { + Start::Pending(_) if state.outcome.aborted().is_none() => Poll::Pending, + start => Poll::Ready(start.sequence()), }); match ready!(res) { Ok(start) => Poll::Ready(start), - Err(state) => Poll::Ready(state.start_sequence), + Err(state) => Poll::Ready(state.start.sequence()), } } @@ -5521,7 +5711,7 @@ impl Consumer { let track = state.read(); let group = track.covering_group(sequence, frame_start)?; let whole = !track.is_closed() - && track.abort.is_none() + && track.outcome.aborted().is_none() && group.live_first_frame() == Some(0) && group.is_finished(); if !whole { @@ -5565,7 +5755,7 @@ impl Consumer { self.with_state(|state| { state.is_some_and(|state| { let state = state.read(); - state.routes.is_none() && state.minted() && !state.closed && state.feed.is_live() + state.routes.is_none() && state.minted() && !state.outcome.is_released() && state.feed.is_live() }) }) } @@ -5830,7 +6020,7 @@ impl Consumer { Ok(_) => Poll::Ready(Ok(())), // Closed before completing. Read through the returned guard: it holds // the lock, so re-locking the channel here would deadlock. - Err(closed) => Poll::Ready(Err(closed.abort.clone().unwrap_or(Error::Dropped))), + Err(closed) => Poll::Ready(Err(closed.outcome.aborted().cloned().unwrap_or(Error::Dropped))), } }) } @@ -5860,7 +6050,7 @@ impl Subscribing { /// How long the upstream that refused the track asked to wait before asking again. #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] pub(crate) fn retry(&self) -> Option { - self.state.read().retry + self.state.read().outcome.retry() } /// Poll until the peer confirms the subscription, yielding the [`Subscriber`]. @@ -5868,7 +6058,7 @@ impl Subscribing { pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll> { // Wait until the track info is available let info = ready!(self.state.poll(waiter, |state| state.poll_info())) - .map_err(|e| e.abort.clone().unwrap_or(Error::Dropped))??; + .map_err(|e| e.outcome.aborted().cloned().unwrap_or(Error::Dropped))??; refuse_untimed(&info, &self.subscription.read())?; if self.subscription.read().has_time() { Consumer::new(self.name.clone(), self.state.weak()).refuse_subscription(&self.subscription.read())?; @@ -5930,7 +6120,7 @@ impl Resolving { /// How long the upstream that refused the track asked to wait before asking again. #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] pub(crate) fn retry(&self) -> Option { - self.state.read().retry + self.state.read().outcome.retry() } /// A handle on the track instance this resolves, whatever its broadcast serves later. @@ -5948,7 +6138,7 @@ impl Resolving { pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll> { // Wait until the track info is available let info = ready!(self.state.poll(waiter, |state| state.poll_info())) - .map_err(|e| e.abort.clone().unwrap_or(Error::Dropped))??; + .map_err(|e| e.outcome.aborted().cloned().unwrap_or(Error::Dropped))??; Poll::Ready(Ok(info)) } } @@ -6168,7 +6358,7 @@ impl Fetching { let cached = |waiter: &kio::Waiter| match state.poll(waiter, |state| state.poll_fetch_cached(sequence, frames)) { Poll::Ready(Ok(res)) => Poll::Ready(res.map(resolve)), - Poll::Ready(Err(closed)) => Poll::Ready(Err(closed.abort.clone().unwrap_or(Error::Dropped))), + Poll::Ready(Err(closed)) => Poll::Ready(Err(closed.outcome.aborted().cloned().unwrap_or(Error::Dropped))), Poll::Pending => Poll::Pending, }; if let Poll::Ready(res) = cached(waiter) { @@ -7941,7 +8131,7 @@ impl Cursor { Poll::Ready(match ready!(polled) { Ok(res) => res, // We try to clone abort just in case the function forgot to check for terminal state. - Err(state) => Err(state.abort.clone().unwrap_or(Error::Dropped)), + Err(state) => Err(state.outcome.aborted().cloned().unwrap_or(Error::Dropped)), }) } @@ -7956,7 +8146,7 @@ impl Cursor { .get_or_insert_with(|| Box::new(kio::Watch::new(self.state.clone()))); Poll::Ready(match ready!(watch.poll(waiter, |state| state.poll_finished())) { Ok(res) => res, - Err(state) => Err(state.abort.clone().unwrap_or(Error::Dropped)), + Err(state) => Err(state.outcome.aborted().cloned().unwrap_or(Error::Dropped)), }) } @@ -8838,7 +9028,7 @@ impl Control { /// A group-only wire refuses an unresolved time anchor before changing shared demand. fn refuse_cached_anchors(state: &TrackState, subscription: &Subscription) -> Result<()> { - if state.cached_anchors && subscription.has_time() { + if state.anchors == Anchors::Cached && subscription.has_time() { let window = Window::new(&subscription.ranges, state.latest_live(), state)?; if !window.is_settled() { return Err(Error::Unsupported); @@ -9085,7 +9275,7 @@ impl Subscriber { #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] pub(crate) fn end_error(&self) -> Option { match &self.inner { - Inner::Plain(cursor) => cursor.state.read().end_error.clone(), + Inner::Plain(cursor) => cursor.state.read().outcome.end().and_then(|end| end.error.clone()), Inner::Resume(resume, _) => resume.declared_error(), } } @@ -9470,7 +9660,7 @@ impl Subscriber { pub(super) fn declared_end(&mut self) -> Option { match &mut self.inner { Inner::Plain(cursor) => match cursor.cache.end_declared() { - true => cursor.state.read().end, + true => cursor.state.read().declared(), false => None, }, Inner::Resume(..) => match self.poll_finished(&kio::Waiter::noop()) { @@ -10093,7 +10283,7 @@ impl Request { pub(crate) fn reject_unwanted(&self, err: Error) -> bool { match self.state.write_unwanted() { kio::Unwanted::Idle(guard) => { - commit_abort(guard, err).abort(); + commit_abort(guard, err, None).abort(); true } kio::Unwanted::Closed => true, @@ -10125,7 +10315,7 @@ impl Request { Ok(mut state) => { state.stage(info.clone()); if let RequestedStart::Pending(floor) = self.requested_start { - state.set_start(floor, true); + state.set_start(Start::Pending(floor)); } if self.idle { state.set_idle(); @@ -10158,7 +10348,7 @@ impl Request { /// Reject the request, waking all waiting subscribers with `err`. pub fn reject(self, err: Error) { if let Ok(mut state) = self.state.write() { - state.abort = Some(err); + state.reject(err, None); } } @@ -10166,8 +10356,7 @@ impl Request { #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] pub(crate) fn refuse(self, err: Error, retry: Duration) { if let Ok(mut state) = self.state.write() { - state.abort = Some(err); - state.retry = Some(retry); + state.reject(err, Some(retry)); } } @@ -12114,6 +12303,22 @@ mod test { assert_eq!(state.max_age_bound(), None); } + /// A refused request stops protecting the live edge a fill cached before the refusal: it + /// joins the eviction order, so pool pressure and idle expiry can still reclaim it. + #[test] + fn a_refusal_releases_the_live_edge_a_fill_cached() { + let state = TrackState::spawn(Arc::new(broadcast::Info::default())); + let mut guard = state.write().ok().unwrap(); + let _group = guard.insert_group_request(0, group::Bounds::default(), None).unwrap(); + assert!(guard.protects(0)); + guard.reject(Error::NotFound, None); + assert!(!guard.protects(0)); + assert!( + guard.evict.iter().any(|(sequence, _)| *sequence == 0), + "the released live edge is reclaimable" + ); + } + #[test] fn optional_age_keeps_publisher_metadata_and_local_policy_separate() { assert_eq!(Info::default().max_age, Some(Duration::ZERO)); @@ -15086,7 +15291,7 @@ mod test { #[test] fn drop_after_abort_does_not_warn() { - // abort() closes the channel after recording `abort`. Drop must treat the + // abort() closes the channel after recording the abort. Drop must treat the // read-only guard returned by write() as clean or it emits a false WARN. let warns = count_drop_warnings("track::Producer dropped without finish", || { let producer = track_producer("test", Info::timed()); @@ -20489,6 +20694,33 @@ mod test { assert_eq!((fill.start().group, fill.end()), (0, 2)); } + /// A received copy's start the publisher later resolves from a budget, at the same group, + /// turns the history below it from unasked into fillable: a reader asks for it then. + #[test] + #[cfg(feature = "async")] + fn a_start_resolved_in_place_asks_the_history_below_it() { + let mut producer = track_producer("test", Info::timed()); + producer.received(); + let dynamic = producer.requests(); + let mut reader = producer.subscribe(replay()); + producer.start_at(2).unwrap(); + write_group(&producer, 2); + write_group(&producer, 3); + let waiter = kio::Waiter::noop(); + assert_eq!(drain_groups(&mut reader), [2, 3]); + assert!( + dynamic.poll_requested_fill(&waiter).is_pending(), + "below a declared start is no miss" + ); + + producer.start_resolved(2).unwrap(); + assert!(reader.poll_recv_group(&waiter).is_pending()); + let Poll::Ready(Ok(fill)) = dynamic.poll_requested_fill(&waiter) else { + panic!("the reader asks below the resolved start"); + }; + assert_eq!((fill.start().group, fill.end()), (0, 2)); + } + /// A reader arriving after another's fill cached the groups under the start reads them /// from the cache: the arrival walk never reaches a backfill, and no fill is asked again. #[test] diff --git a/rs/moq-net/src/model/track/fill.rs b/rs/moq-net/src/model/track/fill.rs index 56aba61e3ce..dcc8d7b369b 100644 --- a/rs/moq-net/src/model/track/fill.rs +++ b/rs/moq-net/src/model/track/fill.rs @@ -798,12 +798,12 @@ pub(super) fn poll_requested( } // No fill queued: surface a track abort so the handler loop can exit. - match state.poll_ref(waiter, |state| match &state.abort { + match state.poll_ref(waiter, |state| match state.outcome.aborted() { Some(err) => Poll::Ready(err.clone()), None => Poll::Pending, }) { Poll::Ready(Ok(err)) => Poll::Ready(Err(err)), - Poll::Ready(Err(closed)) => Poll::Ready(Err(closed.abort.clone().unwrap_or(Error::Dropped))), + Poll::Ready(Err(closed)) => Poll::Ready(Err(closed.outcome.aborted().cloned().unwrap_or(Error::Dropped))), Poll::Pending => Poll::Pending, } } @@ -1417,7 +1417,7 @@ impl Inner { #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] fn finish_track_at(&self, final_sequence: u64) { if let Ok(mut state) = TrackState::modify(&self.state) - && state.end.is_none() + && state.declared().is_none() { let _ = state.set_end(Position::group(final_sequence), None); } @@ -1457,7 +1457,7 @@ impl TrackState { /// or the live feed may still deliver: a fill refused is no verdict on the feed. fn drop_unfed(&mut self, groups: ops::Range, err: StreamError) { // Until the serving session resolves where the feed starts, it may deliver anything. - if self.start_pending { + if matches!(self.start, super::Start::Pending(_)) { return; } let now = self.cache.pool().now_wall(); diff --git a/rs/moq-net/src/model/track/miss.rs b/rs/moq-net/src/model/track/miss.rs index 42dbabcc797..5039d1e49e1 100644 --- a/rs/moq-net/src/model/track/miss.rs +++ b/rs/moq-net/src/model/track/miss.rs @@ -36,7 +36,7 @@ use crate::{Error, ledger}; /// What a probe judges against: the span read, the feed's newest group, where the feed /// and readers start, and the drops revision. -type JudgedAt = (u64, u64, Option, Option, Option, Option, u64); +type JudgedAt = (u64, u64, Option, Option, Option, super::Start, u64); /// The misses of one cursor, judged as what it reads or the feed moves. #[derive(Default)] @@ -623,7 +623,7 @@ impl Misses { let bound = bound.filter(|_| state.unfed).and_then(|end| end.checked_sub(1)); // The feed's start is unknown until the serving session resolves it, and readers // take nothing from a cache that does not reflect the feed. - if state.start_pending || !state.readable() && bound.is_none() { + if matches!(state.start, super::Start::Pending(_)) || !state.readable() && bound.is_none() { return asks; } // A received stale run this reader cannot trust is still asked for, though nothing above @@ -646,7 +646,9 @@ impl Misses { key.2, Some(newest), state.live_floor, - state.start_sequence, + // Whole, not just its sequence: a start resolving from a budget turns the history + // below it from unasked into fillable. + state.start, state.drops_revision, ); let ended = state.nothing_arrives(); @@ -899,18 +901,23 @@ impl Misses { self.view = view.cloned(); for run in candidates.iter() { self.deferred.remove(run.clone()); - self.judged.insert(run.clone()); } - let start = state.start_sequence.unwrap_or(0); + let start = state.start.sequence().unwrap_or(0); // Below a start its publisher resolved from a budget, which an update may have changed // in flight, every older group is history a fill may still find. A feed that began at // the live edge or the floor the copy asked for leaves older groups as history only a // reader naming them asked for. let unasked = state.fills_from_start - || !state.start_resolved && state.received && named.is_none_or(|named| named >= start); + || !matches!(state.start, super::Start::Resolved(_)) + && state.anchors.received() + && named.is_none_or(|named| named >= start); if unasked && start > 0 { candidates.remove(0..start); } + // History left unasked stays unjudged, so a start that changes judges it again. + for run in candidates.iter() { + self.judged.insert(run.clone()); + } for run in candidates.iter() { // Backfill another reader's fill cached is no hole, yet the arrival walk never // reaches it: this reader takes it as filled, as it would its own fill's answer,