diff --git a/js/net/src/auth_session.ts b/js/net/src/auth_session.ts index e2871d95149..12cd2aa9aa3 100644 --- a/js/net/src/auth_session.ts +++ b/js/net/src/auth_session.ts @@ -471,9 +471,11 @@ class IssuedGrant implements Issued { #stream: Stream; #wire: AuthWire; #writes = Promise.resolve(); - /** The latest grant, until its write starts: a newer one replaces it rather than queueing. */ - #queued: Grant | undefined; - #done = false; + /** + * Open, holding the latest grant until its write starts, so a newer one replaces it rather + * than queueing; or done: closed, revoked, or a reply that could not be sent. + */ + #state: { kind: "open"; queued?: Grant } | { kind: "done" } = { kind: "open" }; constructor(stream: Stream, wire: AuthWire) { this.#stream = stream; @@ -495,41 +497,51 @@ class IssuedGrant implements Issued { // The peer already closed the stream: nothing left to tell it. if (err instanceof StreamError) return; // A grant the wire cannot carry is withheld, never widened, and so is any other - // reply that fails to encode. + // reply that fails to encode. An update since then is dropped with it. if (!(err instanceof Unsupported)) console.warn("auth reply not sent", err); - this.#done = true; - this.#queued = undefined; + this.#state = { kind: "done" }; return this.#wire.unsupported(this.#stream, error(err).message).catch(() => void 0); }); } update(grant: Grant): void { - if (this.#done) return; + const state = this.#state; + if (state.kind === "done") return; // A grant still waiting to be written is replaced, so a peer withholding stream credit // holds at most one. - const queued = this.#queued !== undefined; - this.#queued = grant; + const queued = state.queued !== undefined; + state.queued = grant; if (queued) return; this.#write(async () => { - const grant = this.#queued; - this.#queued = undefined; - if (grant === undefined) return; - const expires = grant.expires === undefined ? undefined : grant.expires - wall(); - await this.#wire.grant(this.#stream, { publish: grant.publish, subscribe: grant.subscribe, expires }); + // A close since then sends the grant itself. + const state = this.#state; + if (state.kind === "done" || state.queued === undefined) return; + const grant = state.queued; + state.queued = undefined; + await this.#send(grant); }); } + async #send(grant: Grant): Promise { + const expires = grant.expires === undefined ? undefined : grant.expires - wall(); + await this.#wire.grant(this.#stream, { publish: grant.publish, subscribe: grant.subscribe, expires }); + } + revoke(code: SessionCode, reason: string): void { - if (this.#done) return; + if (this.#state.kind === "done") return; // The peer never saw a grant still queued, so the refusal replaces it. - this.#queued = undefined; + this.#state = { kind: "done" }; this.#write(() => this.#wire.refuse(this.#stream, code, reason)); - this.close(); + this.#writes = this.#writes.then(() => this.#stream.writer.close()); } close(): void { - if (this.#done) return; - this.#done = true; + const state = this.#state; + if (state.kind === "done") return; + this.#state = { kind: "done" }; + // The grant still queued is the last thing the peer hears before the end. + const last = state.queued; + if (last !== undefined) this.#write(() => this.#send(last)); this.#writes = this.#writes.then(() => this.#stream.writer.close()); } } @@ -621,6 +633,8 @@ export class AuthSession { #presented = new Set(); // The handle's tokens as this session presents them, so a request carries only those. #byToken = new Map(); + // How many presented tokens await a reply: at setup, and with bytes. Recounted from + // #presented by #recount after every change to it or to an `isAnswered`. #setupPending = new Signal(0); #pending = new Signal(0); #closed = false; @@ -843,8 +857,6 @@ export class AuthSession { close(err: Error = new Error("session closed")) { if (this.#closed) return; this.#closed = true; - this.#pending.set(0); - this.#setupPending.set(0); for (const presented of this.#presented) { presented.ended = true; presented.isAnswered = true; @@ -859,8 +871,7 @@ export class AuthSession { #present(wire: AuthWire, presented: Presented) { this.#presented.add(presented); - if (presented.setup) this.#setupPending.update((n) => n + 1); - if (presented.token.byteLength > 0) this.#pending.update((n) => n + 1); + this.#recount(); void this.#run(wire, presented); } @@ -967,7 +978,7 @@ export class AuthSession { granted = false; // Unanswered again, as on its first presentation: a request or publication the token // might cover waits for the retry's verdict rather than being judged without it. - this.#unsettle(presented); + presented.isAnswered = false; this.#recompute(); delay = delay === undefined ? RETRY_INITIAL : Milli(Math.min(delay * 2, RETRY_MAX)); presented.retry = new AbortController(); @@ -992,7 +1003,7 @@ export class AuthSession { presented.grant = undefined; // A token that ends unanswered counts as answered for enforcement: its refusal is // the reply. - this.#settle(presented); + presented.isAnswered = true; this.#presented.delete(presented); const owner = presented.owner; if (owner && this.#byToken.get(owner) === presented) this.#byToken.delete(owner); @@ -1020,27 +1031,25 @@ export class AuthSession { /** The peer granted `presented`: it counts as answered, and so does the handle's token. */ #answered(presented: Presented) { - this.#settle(presented); - presented.owner?.answer(); - } - - /** Stop counting `presented` as unanswered, for enforcement and the setup wait. */ - #settle(presented: Presented) { - if (presented.isAnswered) return; presented.isAnswered = true; - if (presented.setup) this.#setupPending.update((n) => n - 1); - if (presented.token.byteLength > 0 && !this.#closed) this.#pending.update((n) => n - 1); + presented.owner?.answer(); } - /** Count `presented` as unanswered again, as a retry presents it anew. */ - #unsettle(presented: Presented) { - if (!presented.isAnswered) return; - presented.isAnswered = false; - if (presented.setup) this.#setupPending.update((n) => n + 1); - if (presented.token.byteLength > 0) this.#pending.update((n) => n + 1); + /** Count the presented tokens still awaiting a reply, for enforcement and the setup wait. */ + #recount() { + let setup = 0; + let pending = 0; + for (const presented of this.#presented) { + if (presented.isAnswered) continue; + if (presented.setup) setup += 1; + if (presented.token.byteLength > 0) pending += 1; + } + this.#setupPending.set(setup); + this.#pending.set(pending); } #recompute() { + this.#recount(); const granted: Grant[] = []; for (const presented of this.#presented) { if (presented.grant) granted.push(presented.grant); diff --git a/js/net/src/connection/reload.ts b/js/net/src/connection/reload.ts index 6ca7327d64c..4beaad32499 100644 --- a/js/net/src/connection/reload.ts +++ b/js/net/src/connection/reload.ts @@ -329,7 +329,7 @@ export class Reload { // outlives the capped delay, or a new sequence starts. #redirects = new Redirects(); - // Whether a connect attempt is in flight, so a retiring predecessor reports the right status. + // Whether a run is dialing: from its start until each connect attempt settles. #dialing = false; // The current wait between attempts, doubling per failure, and when the retry window expires. @@ -524,8 +524,7 @@ export class Reload { this.#givenUpHref = undefined; } - // A drained predecessor still serves while its replacement dials. - if (!this.#draining) this.status.set("connecting"); + this.#setDialing(true); // This run's teardown, handed to connect() so a rerun cancels the attempt in flight. const signal = effect.abort; @@ -536,10 +535,10 @@ export class Reload { effect.cleanup(() => { if (current) { this.#end(current); - if (this.established.peek() === current) this.established.set(undefined); + if (this.established.peek() === current) this.#serve(undefined); current = undefined; } - if (this.established.peek() === undefined) this.status.set("disconnected"); + this.#setDialing(false); }); effect.spawn(async () => { @@ -556,7 +555,7 @@ export class Reload { // spends it too. this.#deadline ??= this.#window(now()); - this.#dialing = true; + this.#setDialing(true); this.#dialedTokens = this.tokens.peek(); let connection: Established; try { @@ -574,8 +573,9 @@ export class Reload { viewers: this.viewers, signal, }); - } finally { - this.#dialing = false; + } catch (err) { + this.#setDialing(false); + throw err; } // Hand the connection to the effect. One that lands after this run is over never @@ -587,8 +587,8 @@ export class Reload { current = connection; // The replacement serves now; a predecessor keeps draining its groups in flight. - this.established.set(connection); - this.status.set("connected"); + this.#serve(connection); + this.#setDialing(false); connected = now(); // A cancelled effect resolves undefined, so the sentinel tells the session @@ -600,7 +600,7 @@ export class Reload { console.warn(`connection closed, reconnecting: ${ended ? reason(ended) : "no error"}`); this.#redirects.reset(); current = undefined; - if (this.established.peek() === connection) this.established.set(undefined); + if (this.established.peek() === connection) this.#serve(undefined); this.#retry(effect, connected, ended ?? undefined); return; } @@ -682,8 +682,7 @@ export class Reload { // would ignore it, so stop here. console.warn("GOAWAY redirect refused:", err); this.#end(connection); - this.established.set(undefined); - this.status.set("disconnected"); + this.#serve(undefined); this.#giveUp(error(err)); return false; } @@ -701,9 +700,7 @@ export class Reload { const draining = new Draining(connection, cap, () => { if (this.#draining === draining) this.#draining = undefined; // If nothing replaced it yet, nothing is serving. - if (this.established.peek() !== connection) return; - this.established.set(undefined); - this.status.set(this.#dialing ? "connecting" : "disconnected"); + if (this.established.peek() === connection) this.#serve(undefined); }); this.#draining = draining; return true; @@ -725,6 +722,23 @@ export class Reload { void ending.then(() => this.#ending.delete(connection)); } + /** Serve `connection`, or nothing. */ + #serve(connection: Established | undefined): void { + this.established.set(connection); + this.#report(); + } + + #setDialing(dialing: boolean): void { + this.#dialing = dialing; + this.#report(); + } + + // The one derivation of status: serving, else dialing, else neither. A drained predecessor + // serves until it retires, so its replacement's dial reads as connected. + #report(): void { + this.status.set(this.established.peek() ? "connected" : this.#dialing ? "connecting" : "disconnected"); + } + /** Stop serving from a drained predecessor, ending it if still open. */ #retire(): void { const draining = this.#draining?.retire(); @@ -749,7 +763,6 @@ export class Reload { * window: the window restarts once the wait is over. */ #restart(effect: Effect): void { - if (this.established.peek() === undefined) this.status.set("disconnected"); const wait = this.#backoff(); const tick = this.#tick.peek() + 1; effect.timer(() => { @@ -774,10 +787,6 @@ export class Reload { * `cause` the error that killed it, if it died with one. */ #retry(effect: Effect, connected: Instant | undefined, cause?: unknown): void { - // Report disconnected during the backoff rather than when the retry reruns the - // effect, unless a drained predecessor still serves until it retires. - if (this.established.peek() === undefined) this.status.set("disconnected"); - // A healthy session's end clears the backoff and starts a fresh retry window: a one-off // drop should reconnect promptly. Anything shorter is a peer that accepts and soon // severs, which has to keep escalating or we hammer it forever at the initial delay. diff --git a/js/net/src/ietf/join.test.ts b/js/net/src/ietf/join.test.ts index 6692127cc15..75bf83a10a9 100644 --- a/js/net/src/ietf/join.test.ts +++ b/js/net/src/ietf/join.test.ts @@ -564,7 +564,7 @@ test("a clean end finishes a head nothing claimed", () => { expect(j.open(() => producer)).toBe(producer); producer.writeFrame({ payload: text("c") }); j.wrote(); - j.finish(); + j.finish(producer); expect(j.outstanding).toBe(true); j.end(); expect(producer.closed.peek()).toBeNull(); @@ -671,3 +671,29 @@ test("an evicted joined group is fetched again as standalone history", async () group.close(); consumer.close(); }); + +test("a second fetch stream after the track ended leaves the first head to its own stream", () => { + const j = new Join(0n, new Tail()); + j.end(); + const first = new track.Producer("catalog").createGroup(0n); + expect(j.open(() => first)).toBe(first); + const second = new track.Producer("catalog").createGroup(0n); + expect(j.open(() => second)).toBeUndefined(); + // The second stream, discarding, ends first: the head is still the first stream's. + j.finish(undefined); + expect(first.closed.peek()).toBeUndefined(); + first.writeFrame({ payload: text("c") }); + j.finish(first); + expect(first.closed.peek()).toBeNull(); +}); + +test("a discarding second fetch stream leaves a head being written alone", () => { + const j = new Join(0n, new Tail()); + const first = new track.Producer("catalog").createGroup(0n); + expect(j.open(() => first)).toBe(first); + expect(j.open(() => new track.Producer("catalog").createGroup(0n))).toBeUndefined(); + j.fail(new Error("reset"), undefined); + expect(first.closed.peek()).toBeUndefined(); + expect(j.pending).toBe(false); + expect(j.outstanding).toBe(true); +}); diff --git a/js/net/src/ietf/join.ts b/js/net/src/ietf/join.ts index 68adac2115a..5b4e254a3d2 100644 --- a/js/net/src/ietf/join.ts +++ b/js/net/src/ietf/join.ts @@ -16,8 +16,11 @@ type State = | { kind: "ready"; producer: netGroup.Producer; next: number } /** No head is coming: refused, empty, failed, claimed, or the subscription went. */ | { kind: "done" } - /** The track ended, so no tail is coming and a head is the whole group. */ - | { kind: "ended" }; + /** + * The track ended, so no tail is coming and a head is the whole group: `producer` is a head + * its fetch stream is still writing. + */ + | { kind: "ended"; producer?: netGroup.Producer }; /** * The join of one live group: the head a draft-18 joining FETCH or a draft-22 fill brings and @@ -47,8 +50,6 @@ export class Join { #tail: Tail; #settled: Promise; #settle!: () => void; - // A head still being written when the track ended, finished with its stream. - #ended?: netGroup.Producer; /** * `counted` says the head's stream is one of the subscription's own, which PUBLISH_DONE's @@ -87,16 +88,19 @@ export class Join { * sequence, the join gave up, or a second stream answers the one request. */ open(create: () => netGroup.Producer): netGroup.Producer | undefined { - const { kind } = this.#state; + const state = this.#state; + if (state.kind === "pending") { + const producer = create(); + this.#state = { kind: "writing", producer, next: 0 }; + return producer; + } // After the track ended the head still lands, as the whole group. - if (kind !== "pending" && kind !== "ended") return undefined; - const producer = create(); - if (kind === "ended") { - this.#ended = producer; + if (state.kind === "ended" && !state.producer) { + const producer = create(); + this.#state = { kind: "ended", producer }; return producer; } - this.#state = { kind: "writing", producer, next: 0 }; - return producer; + return undefined; } /** The head wrote its next object. */ @@ -104,32 +108,38 @@ export class Join { if (this.#state.kind === "writing") this.#state.next += 1; } - /** The fetch stream ended cleanly: the head waits for its tail, or none came at all. */ - finish(): void { - if (this.#ended) { - this.#ended.finish(); - this.#ended = undefined; - return; - } + /** + * A fetch stream ended cleanly, having written `producer` (what `open` gave it): the head + * waits for its tail, or none came at all. A stream `open` turned away changes nothing. + */ + finish(producer: netGroup.Producer | undefined): void { const state = this.#state; - if (state.kind === "writing") { - this.#transition({ kind: "ready", producer: state.producer, next: state.next }); - } else if (state.kind === "pending") { + if (state.kind === "pending") { // An empty fetch stream: there is no head. this.#transition({ kind: "done" }); + } else if (state.kind === "writing" && state.producer === producer) { + this.#transition({ kind: "ready", producer: state.producer, next: state.next }); + } else if (state.kind === "ended" && state.producer && state.producer === producer) { + state.producer.finish(); + this.#state = { kind: "done" }; } } - /** The fetch stream failed: a half-written head is a group with no end. */ - fail(err: Error): void { - if (this.#ended) { - this.#ended.abort(err); - this.#ended = undefined; - return; - } + /** + * A fetch stream failed, having written `producer` (what `open` gave it, if anything): a + * half-written head is a group with no end. A stream `open` turned away changes nothing. + */ + fail(err: Error, producer: netGroup.Producer | undefined): void { const state = this.#state; - if (state.kind === "writing" || state.kind === "ready") state.producer.abort(err); - if (state.kind !== "ended") this.#transition({ kind: "done" }); + if (state.kind === "pending") { + this.#transition({ kind: "done" }); + } else if (state.kind === "writing" && state.producer === producer) { + state.producer.abort(err); + this.#transition({ kind: "done" }); + } else if (state.kind === "ended" && state.producer && state.producer === producer) { + state.producer.abort(err); + this.#state = { kind: "done" }; + } } /** @@ -211,8 +221,8 @@ export class Join { end(): void { const state = this.#state; if (state.kind === "ready") state.producer.finish(); - if (state.kind === "writing") this.#ended = state.producer; - if (state.kind !== "done") this.#transition({ kind: "ended" }); + if (state.kind === "pending" || state.kind === "ready") this.#transition({ kind: "ended" }); + if (state.kind === "writing") this.#transition({ kind: "ended", producer: state.producer }); } /** @@ -221,10 +231,16 @@ export class Join { */ cancel(err: Error): void { const state = this.#state; - if (state.kind === "writing" || state.kind === "ready") state.producer.abort(err); - this.#ended?.abort(err); - this.#ended = undefined; - if (state.kind !== "done" && state.kind !== "ended") this.#transition({ kind: "done" }); + if (state.kind === "writing" || state.kind === "ready") { + state.producer.abort(err); + this.#transition({ kind: "done" }); + } else if (state.kind === "pending") { + this.#transition({ kind: "done" }); + } else if (state.kind === "ended") { + // Already settled: only a head still being written is left to drop. + state.producer?.abort(err); + this.#state = { kind: "done" }; + } } #transition(next: State): void { diff --git a/js/net/src/ietf/request_stream.test.ts b/js/net/src/ietf/request_stream.test.ts index 121adbba81e..56353ca3654 100644 --- a/js/net/src/ietf/request_stream.test.ts +++ b/js/net/src/ietf/request_stream.test.ts @@ -236,3 +236,35 @@ test("a request waiting for stream credit outlasts its deadline, which runs from installed.close(); } }); + +test("a refusal that fails to write still ends the request", async () => { + const failed = new Error("stop sending"); + const stream = new Stream({ + readable: new ReadableStream(), + writable: new WritableStream({ + write: () => { + throw failed; + }, + }), + version: Version.DRAFT_22, + }); + const aborts = spyOn(stream, "abort"); + const session = { version: Version.DRAFT_22, client: true } as unknown as Session; + const request = new RequestStream(stream, session); + await expect(request.refuse("fetch", refusal)).rejects.toBe(failed); + expect(aborts).toHaveBeenCalledTimes(1); + await request.ended; + + const bare = new Stream({ + readable: new ReadableStream(), + writable: new WritableStream({ + write: () => { + throw failed; + }, + }), + version: Version.DRAFT_22, + }); + const bareAborts = spyOn(bare, "abort"); + await expect(refuse(bare, Version.DRAFT_22, "fetch", refusal)).rejects.toBe(failed); + expect(bareAborts).toHaveBeenCalledTimes(1); +}); diff --git a/js/net/src/ietf/request_stream.ts b/js/net/src/ietf/request_stream.ts index 7f04a7e4f2f..56cc794a73b 100644 --- a/js/net/src/ietf/request_stream.ts +++ b/js/net/src/ietf/request_stream.ts @@ -55,7 +55,12 @@ async function writeRefusal(w: Writer, version: IetfVersion, kind: RequestKind, * @internal */ export async function refuse(stream: Stream, version: IetfVersion, kind: RequestKind, refusal: Refusal): Promise { - await writeRefusal(stream.writer, version, kind, refusal); + try { + await writeRefusal(stream.writer, version, kind, refusal); + } catch (err) { + stream.abort(error(err)); + throw err; + } stream.close(); } @@ -226,10 +231,9 @@ export class RequestStream { } } - /** Refuse the request and end it. */ + /** Refuse the request and end it, resetting it if the refusal cannot be written. */ async refuse(kind: RequestKind, refusal: Refusal): Promise { - await writeRefusal(this.writer, this.#session.version, kind, refusal); - this.close(); + await this.serve(writeRefusal(this.writer, this.#session.version, kind, refusal)); } /** diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 0368a398cf7..56944e46031 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -2127,7 +2127,7 @@ export class Subscriber { stream.tryDecode(decode) ?? (await race([stream.decodeMaybe(decode), producer?.closed ?? hooks.trackReleased(track)])); if (object === null || object instanceof Error) { - join.fail(object ?? new Error("the track closed under the head")); + join.fail(object ?? new Error("the track closed under the head"), producer); stream.stop(object ?? new StreamError(StreamCode.Cancel)); return; } @@ -2186,11 +2186,11 @@ export class Subscriber { next += 1; } console.debug(`join head: group=${join.sequence} objects=${next}`); - join.finish(); + join.finish(producer); } catch (err: unknown) { const e = await sessionCause(this.#quic, err); if (e instanceof JoinUnsupported) console.warn(`${e.message}, joining at the next whole group`); - join.fail(e); + join.fail(e, producer); if (e instanceof ProtocolViolation) this.#fail(e.code, e.message); // As for a subgroup object: the track is malformed, not just its head. if (e instanceof StreamError && e.code === StreamCode.MalformedTrack) track.abort(e); diff --git a/js/net/src/listing.test.ts b/js/net/src/listing.test.ts index e3c478f2b78..10438b23079 100644 --- a/js/net/src/listing.test.ts +++ b/js/net/src/listing.test.ts @@ -351,6 +351,21 @@ test("a closed lister withdraws its answer and syncs", async () => { expect(await tracks.drain()).toEqual(["-up", "synced"]); }); +test("a closed lister answers nothing more, even beside its successor", async () => { + const broadcast = new BroadcastProducer(); + const stale = wireOf(broadcast).lister(); + stale.close(); + const lister = wireOf(broadcast).lister(); + const tracks = list(broadcast); + stale.add("stale", { priority: 0 }); + stale.synced(); + stale.abort(new Error("stale")); + expect(await tracks.drain()).toEqual([]); + lister.add("up", { priority: 0 }); + lister.synced(); + expect(await tracks.drain()).toEqual(["+up", "synced"]); +}); + test("an owner lists each broadcast once", async () => { const broadcast = new BroadcastProducer(); const other = new BroadcastProducer(); diff --git a/js/net/src/listing.ts b/js/net/src/listing.ts index 80e45d4a491..ec7ba1152d6 100644 --- a/js/net/src/listing.ts +++ b/js/net/src/listing.ts @@ -420,15 +420,18 @@ export let makeLister: (listing: Listing) => Lister; * * While it exists, readers wait for it to {@link synced} before reporting `synced`. Answer * while {@link wanted}: list what the publisher behind it holds and keep it current, then - * {@link clear} once unwanted. {@link close} withdraws what it listed. + * {@link clear} once unwanted. {@link close} withdraws what it listed; a closed lister answers + * nothing more. */ export class Lister { - #listing: Listing; - #closed = false; + readonly #listing: Listing; + // The listing this lister answers into, until it closes. + #answering?: Listing; private constructor(listing: Listing) { listing.attach(); this.#listing = listing; + this.#answering = listing; } static { @@ -447,45 +450,46 @@ export class Lister { /** List a track. Throws if this answer already lists it. */ add(name: string, info: track.Info): void { - this.#listing.answerAdd(name, info); + this.#answering?.answerAdd(name, info); } /** Stop listing a track that left unfinished, reported as `removed`. Throws if this answer does not list it. */ remove(name: string): void { - this.#listing.answerRemove(name, false); + this.#answering?.answerRemove(name, false); } /** Stop listing a track that finished, reported as `finished`. Throws if this answer does not list it. */ finish(name: string): void { - this.#listing.answerRemove(name, true); + this.#answering?.answerRemove(name, true); } /** The answer holds everything this hop knows now; later changes are live. */ synced(): void { - this.#listing.answerSynced(); + this.#answering?.answerSynced(); } /** Withdraw the answer: every track it listed, and its sync. */ clear(): void { - this.#listing.answerClear(); + this.#answering?.answerClear(); } /** End every current reader as if the broadcast ended, and withdraw the answer. A later reader asks again. */ end(): void { - this.#listing.detach(null); - this.#listing.answerClear(); + const listing = this.#answering; + listing?.detach(null); + listing?.answerClear(); } /** Fail every current reader with `err`, and withdraw the answer. A later reader asks again. */ abort(err: Error): void { - this.#listing.detach(err); - this.#listing.answerClear(); + const listing = this.#answering; + listing?.detach(err); + listing?.answerClear(); } /** Stop answering: the broadcast lists only its producer's tracks from here on. Idempotent. */ close(): void { - if (this.#closed) return; - this.#closed = true; - this.#listing.detachLister(); + this.#answering?.detachLister(); + this.#answering = undefined; } } diff --git a/js/signals/src/index.test.ts b/js/signals/src/index.test.ts index 0453e62e8a4..05b9b8390a4 100644 --- a/js/signals/src/index.test.ts +++ b/js/signals/src/index.test.ts @@ -864,6 +864,41 @@ describe("Effect", () => { expect(log).toEqual(["served", "released"]); }); + test("get() from a stale spawn does not subscribe the next run", async () => { + const tick = new Signal(0); + const other = new Signal(0); + const gate = Promise.withResolvers(); + let runs = 0; + + const effect = new Effect((e) => { + e.get(tick); + if (++runs > 1) return; + + e.spawn(async () => { + await gate.promise; + expect(e.get(other)).toBe(0); + }); + }); + + try { + await settle(); + tick.set(1); + await settle(); + expect(runs).toBe(1); // stale window + + gate.resolve(); + await settle(); + expect(runs).toBe(2); + + // Only the dead run read `other`. + other.set(1); + await settle(); + expect(runs).toBe(2); + } finally { + effect.close(); + } + }); + test("cleanup after close fires immediately", async () => { const effect = new Effect((e) => { e.get(new Signal(0)); diff --git a/js/signals/src/index.ts b/js/signals/src/index.ts index 1eb72ecacf6..21be471d1d7 100644 --- a/js/signals/src/index.ts +++ b/js/signals/src/index.ts @@ -743,12 +743,16 @@ export class Effect { } } - /** Reads a signal and tracks it, rerunning the effect whenever it changes. */ + /** + * Reads a signal and tracks it, rerunning the effect whenever it changes. A {@link spawn} task + * that outlived its run only reads: the next run tracks what it reads itself. + */ get(signal: Getter): T { if (this.#dispose === undefined) { console.warn("Effect.get called when closed, returning current value"); return signal.peek(); } + if (this.#stale) return signal.peek(); const value = signal.peek(); diff --git a/quest/m1/explicit-state/net-phases.md b/quest/m1/explicit-state/net-phases.md index 35d89e9b45c..31300e99327 100644 --- a/quest/m1/explicit-state/net-phases.md +++ b/quest/m1/explicit-state/net-phases.md @@ -18,14 +18,14 @@ redesigning it, and reproduce any inconsistent state you find. publisher's `TrackServe` phase and its reopened-tail fix, the datagram forward, the session drivers' terminal result, IETF `Dispatch`, lite `PeerSetup`, the listing `Queue`, and the js/net connections' close phase. -What remains overlapped open PRs when that one started: +[moq-dev/moq#8322](https://github.com/moq-dev/moq/pull/8322) landed js/net's +`Join`, `RequestStream.refuse`, `Lister`, auth counters and `IssuedGrant`, +`Reload.status`, the signals `Effect` stale read, and Rust's `Following`. + +The lite items overlap [moq-dev/moq#8220](https://github.com/moq-dev/moq/pull/8220) (lite publisher -and subscriber, `fills.rs`, js/net lite), -[moq-dev/moq#8063](https://github.com/moq-dev/moq/pull/8063) (lite announce -and subscriber, `model/origin.rs`, `model/follow.rs`, js/net `origin.ts`), -and [moq-dev/moq#8222](https://github.com/moq-dev/moq/pull/8222) (js/net -`reload.ts` and lite `subscriber.ts`). Start each item once the PR touching -its file has landed or closed. +and subscriber, `fills.rs`, js/net lite). Start each once it has landed or +closed. Rust: @@ -42,32 +42,17 @@ Rust: - `SubStream.served` plus `started` are really `Option>`. - **`auth::Slot`** (`rs/moq-net/src/auth.rs`): `streamed`, `refused`, `withdrawn`, `grant`, `answered`, `retry`, and `ended` are seven fields - coordinated by hand. -- **`Following`** (`rs/moq-net/src/model/follow.rs`): eight fields, of which - `last`, `moved`, and `absorbed` are always reset together. + coordinated by hand. Its tests call `granted` on slots whose stream never + opened, so the phase enum has them open the stream first. - **Smaller ones**: - `model/origin.rs` `TrackIo` keeps `request` and `accepted` side by side. - `fills.rs` `Recovery`. JS (`js/net`): -- **Auth** (`auth_session.ts`): - - `AuthSession`/`Presented` derive counters by hand at six sites. - - `IssuedGrant` sets `#done` asynchronously, so two updates in one tick - both pass its check. -- **`Reload`** (`connection/reload.ts`): `status` is written at seven sites, - each re-deriving it from `established`, `#draining`, and `#dialing`. - **Lite `SubscribeEntry`** (`lite/subscriber.ts`): eleven optional fields. -- **`Lister`** (`listing.ts`): it still writes into the listing after - `close()`. -- **`RequestStream`** (`ietf/request_stream.ts`): `refuse()` leaves the stream - open if writing the refusal throws. -- **Signals `Effect`** (`js/signals/src/index.ts`): `get()` during the stale - window pushes into the next run's unwatch list. - **Origin `Request` and `OriginState`** (`origin.ts`): the closed state is encoded by about 40 `?.` guards. -- **`Join`** (`ietf/join.ts`): a second fetch stream after the track ended - leaks the first producer. Keep the lite `TrackRun` start state out of this child. It belongs to [Delete the old start bookkeeping](/quest/m2/subscribe-ranges/start-bookkeeping.md). diff --git a/rs/moq-net/src/model/follow.rs b/rs/moq-net/src/model/follow.rs index d4468c5bbc0..893d3c4bff3 100644 --- a/rs/moq-net/src/model/follow.rs +++ b/rs/moq-net/src/model/follow.rs @@ -197,24 +197,39 @@ pub struct Following { unseen: bool, /// The resolution in flight, kept across a dropped wait so a retry picks it up. resolving: Option, - /// The broadcast handed out last, until the path goes offline or the broadcast ends. Weak, - /// so it never keeps one the caller let go of from retiring. - last: Option, - /// Once the broadcast handed out last closed, waits for its instance to stop serving the - /// path. - moved: Option, - /// A change resolved to the broadcast handed out last while it was still open, so it counts - /// as delivered only until that broadcast closes. - absorbed: bool, - /// A broadcast was handed out and its end not reported yet. - live: bool, - /// The broadcast handed out last stopped serving the path, and its end is owed. - ended: bool, + /// The broadcast handed out last, until the path goes offline, the broadcast ends, or a + /// resolution is refused. + last: Option, + /// Whether the end of what was handed out is still to be reported. + handed: Handed, /// The last resolution was refused: any change to the serving route, even the same /// instance over another route, is worth asking again. refused: bool, } +/// The broadcast a [`Following`] handed out last, and what it waits on once that closes. +struct Last { + /// Weak, so it never keeps one the caller let go of from retiring. + broadcast: broadcast::WeakConsumer, + /// Once the broadcast closed, waits for its instance to stop serving the path. + moved: Option, + /// A change resolved to this broadcast while it was still open, so it counts as delivered + /// only until the broadcast closes. + absorbed: bool, +} + +/// Whether a [`Following`] owes the end of the broadcast it handed out. +#[derive(Clone, Copy, PartialEq, Eq)] +enum Handed { + /// Nothing handed out, or its end already reported. + Nothing, + /// A broadcast was handed out and its end is not yet due, even if a refused takeover means + /// it no longer serves the path. + Live, + /// The broadcast handed out stopped serving the path, and its end is owed. + Ended, +} + impl Following { pub(super) fn new(consumer: Consumer, path: Path, instance: Option, follow: Follow) -> Self { Self { @@ -225,10 +240,7 @@ impl Following { unseen: false, resolving: None, last: None, - moved: None, - absorbed: false, - live: false, - ended: false, + handed: Handed::Nothing, refused: false, } } @@ -253,9 +265,8 @@ impl Following { self.resolving = None; } // Report an end before whatever serves next, so a reader drops the old broadcast first. - if self.ended { - self.ended = false; - self.live = false; + if self.handed == Handed::Ended { + self.handed = Handed::Nothing; return Poll::Ready(Ok(None)); } // The broadcast handed out ended while the path is still served: its instance's @@ -265,12 +276,13 @@ impl Following { // change it absorbed is owed. A refusal forgets it, so only a route change asks again. if !self.unseen && self.follow.serving().is_some() - && let Some(last) = self.last.as_ref().map(broadcast::WeakConsumer::consume) - && last.poll_closed(waiter).is_ready() + && let Some(last) = &mut self.last + && let broadcast = last.broadcast.consume() + && broadcast.poll_closed(waiter).is_ready() { - let moved = match &last.info().instance { - _ if self.absorbed => true, - Some(instance) => self + let moved = match &broadcast.info().instance { + _ if last.absorbed => true, + Some(instance) => last .moved .get_or_insert_with(|| { self.consumer @@ -281,7 +293,7 @@ impl Following { None => true, }; if moved { - self.moved = None; + last.moved = None; self.unseen = true; } } @@ -301,12 +313,10 @@ impl Following { self.refused = resolved.is_err(); let broadcast = match resolved { Ok(broadcast) => broadcast, - // `live` stays: the end of what was handed out is still owed once the path goes + // `handed` stays: the end of what was handed out is still owed once the path goes // offline. Err(err) => { self.last = None; - self.moved = None; - self.absorbed = false; return Poll::Ready(Err(err)); } }; @@ -315,14 +325,18 @@ impl Following { // the origin holds its restart, and that restart then names the instance handed // out. The caller freeing it before reading the restart is owed it again. let weak = broadcast.weak(); - if self.last.as_ref().is_some_and(|last| last.same_channel(&weak)) { - self.absorbed = true; + if let Some(last) = &mut self.last + && last.broadcast.same_channel(&weak) + { + last.absorbed = true; continue; } - self.last = Some(weak); - self.moved = None; - self.absorbed = false; - self.live = true; + self.last = Some(Last { + broadcast: weak, + moved: None, + absorbed: false, + }); + self.handed = Handed::Live; return Poll::Ready(Ok(Some(broadcast))); } } @@ -339,10 +353,10 @@ impl Following { /// The broadcast handed out last stopped serving the path: owe its end, and hand out /// whatever serves next, even the same instance coming back. fn end(&mut self) { - self.ended |= self.live; + if self.handed == Handed::Live { + self.handed = Handed::Ended; + } self.last = None; - self.moved = None; - self.absorbed = false; } /// Apply one change to the path, returning whether it is one.