diff --git a/src/plugins/nicojk/App.tsx b/src/plugins/nicojk/App.tsx index bc0097f..e7cd5cb 100644 --- a/src/plugins/nicojk/App.tsx +++ b/src/plugins/nicojk/App.tsx @@ -540,6 +540,15 @@ function scopeLiveComment( }; } +function compareComments(left: NiconicoComment, right: NiconicoComment) { + return ( + left.vpos - right.vpos || + left.date - right.date || + left.date_usec - right.date_usec || + left.id - right.id + ); +} + function mergeComments( existing: NiconicoComment[], incoming: NiconicoComment[], @@ -549,13 +558,38 @@ function mergeComments( return existing; } - const merged = [...existing, ...incoming].sort( - (a, b) => - a.vpos - b.vpos || - a.date - b.date || - a.date_usec - b.date_usec || - a.id - b.id, - ); + // Live comments normally arrive one at a time. Keep the existing sorted + // array and insert into it instead of sorting all 1000 retained comments + // for every incoming message. + if (incoming.length === 1) { + const comment = incoming[0]; + if (!comment || existing.some((item) => item.id === comment.id)) { + return existing; + } + + let insertAt = existing.length; + const lastComment = existing.at(-1); + if (lastComment && compareComments(lastComment, comment) > 0) { + let low = 0; + let high = existing.length; + while (low < high) { + const middle = low + Math.floor((high - low) / 2); + const middleComment = existing[middle]; + if (middleComment && compareComments(middleComment, comment) <= 0) { + low = middle + 1; + } else { + high = middle; + } + } + insertAt = low; + } + + const merged = existing.slice(); + merged.splice(insertAt, 0, comment); + return typeof maxCount === "number" ? merged.slice(-maxCount) : merged; + } + + const merged = [...existing, ...incoming].sort(compareComments); const deduped: NiconicoComment[] = []; const seen = new Set(); for (const comment of merged) { @@ -1231,6 +1265,19 @@ export default function App() { setInterruptedSources(data?.interruptedSources || []); }; + const pendingLiveSyncPlayerIds = new Set(); + let liveSyncTimer: number | null = null; + const scheduleLiveTargetSync = (playerID: string) => { + pendingLiveSyncPlayerIds.add(playerID); + if (liveSyncTimer !== null) return; + liveSyncTimer = window.setTimeout(() => { + liveSyncTimer = null; + const playerIDs = [...pendingLiveSyncPlayerIds]; + pendingLiveSyncPlayerIds.clear(); + for (const id of playerIDs) syncTargetState(id); + }, 50); + }; + const startRecordedCommentsFetch = ( playerID: string, priorityTime?: number, @@ -1925,7 +1972,7 @@ export default function App() { MAX_LIVE_COMMENTS, ); if (targetPlayableRef.current?.playerID === pid) { - syncTargetState(pid); + scheduleLiveTargetSync(pid); } } }); @@ -2129,6 +2176,11 @@ export default function App() { handleLiveCommentSourceChange, ); clearInterval(interval); + if (liveSyncTimer !== null) { + window.clearTimeout(liveSyncTimer); + liveSyncTimer = null; + } + pendingLiveSyncPlayerIds.clear(); startRecordedCommentsFetchRef.current = null; pendingRecordedReloadRef.current.clear(); for (const client of clientsRef.current.values()) client.disconnect(); diff --git a/src/plugins/nicojk/components/OverlayPage.tsx b/src/plugins/nicojk/components/OverlayPage.tsx index 0f4820d..4b1b956 100644 --- a/src/plugins/nicojk/components/OverlayPage.tsx +++ b/src/plugins/nicojk/components/OverlayPage.tsx @@ -36,6 +36,10 @@ const RENDER_SEGMENT_SIZE = 3600; const RENDER_SEGMENT_OVERLAP = 60; // App側のライブコメント上限に余裕を加えた件数でrendererを再構築し、内部保持量を固定する const MAX_LIVE_RENDERER_COMMENTS = 1200; +const DEFAULT_LIVE_COMMENT_DURATION_VPOS = 300; +const MAX_LIVE_COMMENT_DURATION_VPOS = 12_000; +// 横書きコメントは vpos の前後にも描画・衝突判定のために保持される。 +const LIVE_COMMENT_DRAW_LEAD_VPOS = 125; function getSegmentComments( comments: NiconicoComment[], @@ -55,6 +59,33 @@ function getSegmentComments( ); } +function getLiveCommentDurationVpos(comment: NiconicoComment) { + let durationVpos = DEFAULT_LIVE_COMMENT_DURATION_VPOS; + for (const mail of comment.mail || []) { + const match = /^[@＠]([0-9.]+)$/.exec(mail); + if (!match) continue; + const duration = Math.floor(Number(match[1]) * 100); + durationVpos = + Number.isFinite(duration) && duration > 0 + ? Math.min(duration, MAX_LIVE_COMMENT_DURATION_VPOS) + : DEFAULT_LIVE_COMMENT_DURATION_VPOS; + } + return durationVpos; +} + +export function getLiveRendererSeedComments( + comments: NiconicoComment[], + nowVpos: number, +) { + return comments.filter( + (comment) => + comment.vpos + + getLiveCommentDurationVpos(comment) + + LIVE_COMMENT_DRAW_LEAD_VPOS >= + nowVpos, + ); +} + function isCommentVisibleForSource( comment: NiconicoComment, jkContext: NicoJKContext | null, @@ -389,7 +420,10 @@ export default function OverlayPage({ : []; const liveComments = isLive ? toFormattedComments( - segmentComments, + getLiveRendererSeedComments( + segmentComments, + Math.floor(Date.now() / 10), + ), visibleSourceKeys, jkContext, currentSettings, diff --git a/src/plugins/nicojk/components/PanelPage.tsx b/src/plugins/nicojk/components/PanelPage.tsx index f64f15a..f11b8fb 100644 --- a/src/plugins/nicojk/components/PanelPage.tsx +++ b/src/plugins/nicojk/components/PanelPage.tsx @@ -1458,63 +1458,21 @@ export default function PanelPage({ )} {source.chapterCorrection && (
- チャプター補正{" "} - {formatCorrectionSeconds( - source.chapterCorrection.offsetSeconds, - )} - {source.chapterCorrection.enabled - ? " 適用中" - : " 無効"} -
- )} - {isInterrupted && interruptedChunks.length > 0 && ( -
- {interruptedChunks.map((chunk) => { - const chunkLabel = `${formatTimeRange(chunk.startAt, chunk.endAt)}`; - return ( - - ); - })} - - {interruptedInfo?.fetchedChunkCount || 0}/ - {interruptedInfo?.totalChunkCount || 0} + + チャプター補正{" "} + {formatCorrectionSeconds( + source.chapterCorrection.offsetSeconds, + )} + {source.chapterCorrection.enabled + ? " 適用中" + : " 無効"} -
- )} - - {(isInterrupted || source.chapterCorrection) && ( -
- {isInterrupted && ( - - )} - {source.chapterCorrection && ( - )} +
+ )} + {isInterrupted && interruptedChunks.length > 0 && ( +
+ {interruptedChunks.map((chunk) => { + const chunkLabel = `${formatTimeRange(chunk.startAt, chunk.endAt)}`; + return ( + + ); + })} + + {interruptedInfo?.fetchedChunkCount || 0}/ + {interruptedInfo?.totalChunkCount || 0} + +
+ )} + + {isInterrupted && ( +
+
)}
; + timeouts?: Partial; + random?: () => number; +} + +const DEFAULT_NDGR_RETRY_POLICY: NdgrRetryPolicy = { + maxNumberOfRetry: 5, + startingIntervalMs: 500, + timeMultiple: 1.5, + randomizationFactor: 0.5, +}; + +const DEFAULT_NDGR_TIMEOUT_POLICY: NdgrTimeoutPolicy = { + viewFetchHeaderMs: 60_000, + viewStreamReadMs: 60_000, + segmentFetchHeaderMs: 1_000, + segmentStreamReadMs: 30_000, +}; + +type NdgrTimeoutPhase = "fetch-header" | "stream-read"; + +class NdgrFetchTimeoutError extends Error { + override readonly name = "NdgrFetchTimeoutError"; + + constructor(requestKind: "view" | "segment", phase: NdgrTimeoutPhase) { + super(`NicoNico ${requestKind} ${phase} timed out`); + } +} + const OPACITY_MAIL_VALUES: Record = { Normal: "nico:opacity:1", Translucent: "nico:opacity:0.5", @@ -31,10 +75,71 @@ export function getNiconicoLiveVpos(receivedAtMs: number, dateUsec: number) { return baseVpos + 200 + jitter; } +export function getNdgrRetryDelayMs( + retryNumber: number, + policy: NdgrRetryPolicy = DEFAULT_NDGR_RETRY_POLICY, + randomValue = Math.random(), +) { + const baseDelay = + policy.startingIntervalMs * + policy.timeMultiple ** Math.max(0, retryNumber - 1); + const jitterMultiplier = + 1 - + policy.randomizationFactor + + randomValue * policy.randomizationFactor * 2; + return Math.max(0, Math.round(baseDelay * jitterMultiplier)); +} + function isAbortError(error: unknown) { return error instanceof Error && error.name === "AbortError"; } +function getAbortError(signal: AbortSignal) { + return signal.reason instanceof Error + ? signal.reason + : new DOMException("The connection was aborted", "AbortError"); +} + +function createNdgrAttempt(parentSignal: AbortSignal) { + const controller = new AbortController(); + const abortFromParent = () => controller.abort(getAbortError(parentSignal)); + if (parentSignal.aborted) abortFromParent(); + else parentSignal.addEventListener("abort", abortFromParent, { once: true }); + + return { + signal: controller.signal, + async runWithTimeout( + operation: () => Promise, + timeoutMs: number, + requestKind: "view" | "segment", + phase: NdgrTimeoutPhase, + ) { + if (parentSignal.aborted) throw getAbortError(parentSignal); + const timeoutError = new NdgrFetchTimeoutError(requestKind, phase); + const timeout = setTimeout( + () => controller.abort(timeoutError), + timeoutMs, + ); + let removeAttemptAbortListener = () => {}; + const attemptAborted = new Promise((_resolve, reject) => { + const onAbort = () => reject(controller.signal.reason ?? timeoutError); + controller.signal.addEventListener("abort", onAbort, { once: true }); + removeAttemptAbortListener = () => + controller.signal.removeEventListener("abort", onAbort); + }); + try { + return await Promise.race([operation(), attemptAborted]); + } finally { + clearTimeout(timeout); + removeAttemptAbortListener(); + } + }, + dispose() { + parentSignal.removeEventListener("abort", abortFromParent); + }, + }; +} + function waitForAbort(signal: AbortSignal): Promise { if (signal.aborted) return Promise.resolve(); return new Promise((resolve) => @@ -58,6 +163,22 @@ function waitForReconnect(signal: AbortSignal): Promise { }); } +function waitForRetry(delayMs: number, signal: AbortSignal): Promise { + if (signal.aborted) return Promise.resolve(false); + return new Promise((resolve) => { + const timeout = setTimeout(() => { + signal.removeEventListener("abort", onAbort); + resolve(true); + }, delayMs); + const onAbort = () => { + clearTimeout(timeout); + signal.removeEventListener("abort", onAbort); + resolve(false); + }; + signal.addEventListener("abort", onAbort, { once: true }); + }); +} + export function modifierToMail(modifier: DecodedChat["modifier"]): string[] { if (!modifier) return []; const commands = [ @@ -72,6 +193,9 @@ export function modifierToMail(modifier: DecodedChat["modifier"]): string[] { } export class NiconicoCommentClient implements LiveCommentClient { + private readonly retryPolicy: NdgrRetryPolicy; + private readonly timeoutPolicy: NdgrTimeoutPolicy; + private readonly random: () => number; private socket: SocketLike | null = null; private bc: BroadcastChannel | null = null; private abortController: AbortController | null = null; @@ -80,6 +204,20 @@ export class NiconicoCommentClient implements LiveCommentClient { private status: ConnectionStatus = "disconnected"; private sourceKey: string | null = null; private generation = 0; + private activeSegmentTasks = new Map< + string, + { generation: number; task: Promise } + >(); + private seenSegmentUris = new Set(); + + constructor(options: NiconicoCommentClientOptions = {}) { + this.retryPolicy = { ...DEFAULT_NDGR_RETRY_POLICY, ...options.retry }; + this.timeoutPolicy = { + ...DEFAULT_NDGR_TIMEOUT_POLICY, + ...options.timeouts, + }; + this.random = options.random ?? Math.random; + } public connect( source: ResolvedCommentSource, @@ -151,6 +289,8 @@ export class NiconicoCommentClient implements LiveCommentClient { this.generation += 1; this.abortController?.abort(); this.abortController = null; + this.activeSegmentTasks.clear(); + this.seenSegmentUris.clear(); this.socket?.close(); this.socket = null; this.bc?.close(); @@ -188,22 +328,22 @@ export class NiconicoCommentClient implements LiveCommentClient { if (!source.nicoliveCommunityId) { throw new Error("NicoNico community ID is unavailable"); } - const page = await resolveNiconicoWatchPage(source.nicoliveCommunityId); - if (!this.isCurrent(generation)) return; - while (this.isCurrent(generation) && !signal.aborted) { + this.seenSegmentUris.clear(); const sessionController = new AbortController(); const abortSession = () => sessionController.abort(); signal.addEventListener("abort", abortSession, { once: true }); - let sessionSocket: SocketLike | null = null; try { - const { viewUri, socket } = await this.openWatchSocket( + // WebSocket URL は再接続ごとに取り直す。古い視聴ページに含まれる + // URL を再利用すると、失効後に回復できなくなる。 + const page = await resolveNiconicoWatchPage(source.nicoliveCommunityId); + if (!this.isCurrent(generation) || signal.aborted) return; + const viewUri = await this.openWatchSocket( page.webSocketUrl, sessionController.signal, generation, abortSession, ); - sessionSocket = socket; if (!this.isCurrent(generation) || signal.aborted) return; this.updateStatus("connected"); await this.consumeView( @@ -224,10 +364,6 @@ export class NiconicoCommentClient implements LiveCommentClient { } finally { signal.removeEventListener("abort", abortSession); sessionController.abort(); - if (sessionSocket && this.socket === sessionSocket) { - sessionSocket.close(); - this.socket = null; - } } if (signal.aborted || !this.isCurrent(generation)) return; @@ -241,11 +377,12 @@ export class NiconicoCommentClient implements LiveCommentClient { signal: AbortSignal, generation: number, onDisconnected: () => void, - ): Promise<{ viewUri: string; socket: SocketLike }> { + ): Promise { return new Promise((resolve, reject) => { const socket = new WebSocket(url); this.socket = socket; let settled = false; + let closedAfterMessageServer = false; const timeout = setTimeout(() => { if (!settled) { settled = true; @@ -270,6 +407,7 @@ export class NiconicoCommentClient implements LiveCommentClient { }; socket.onmessage = (event) => { try { + if (closedAfterMessageServer) return; const message = JSON.parse(String(event.data)); if (message.type === "ping") { socket.send(JSON.stringify({ type: "pong" })); @@ -296,7 +434,12 @@ export class NiconicoCommentClient implements LiveCommentClient { settled = true; clearTimeout(timeout); signal.removeEventListener("abort", abort); - resolve({ viewUri, socket }); + // messageServer は View URI を取得するためだけの接続。NDGRClient + // と同様にここで閉じ、以降の NDGR HTTP ストリームと結び付けない。 + closedAfterMessageServer = true; + socket.close(); + if (this.socket === socket) this.socket = null; + resolve(viewUri); } catch (error) { if (!settled) { settled = true; @@ -306,7 +449,7 @@ export class NiconicoCommentClient implements LiveCommentClient { } }; socket.onerror = () => { - if (settled) onDisconnected(); + if (settled && !closedAfterMessageServer) onDisconnected(); if (!settled) { settled = true; clearTimeout(timeout); @@ -314,7 +457,8 @@ export class NiconicoCommentClient implements LiveCommentClient { } }; socket.onclose = () => { - if (settled && !signal.aborted) onDisconnected(); + if (settled && !signal.aborted && !closedAfterMessageServer) + onDisconnected(); if (!settled && !signal.aborted) { settled = true; clearTimeout(timeout); @@ -324,6 +468,32 @@ export class NiconicoCommentClient implements LiveCommentClient { }); } + private async runNdgrWithRetry( + signal: AbortSignal, + generation: number, + operation: () => Promise, + ): Promise { + let retryNumber = 0; + while (true) { + if (signal.aborted || !this.isCurrent(generation)) { + throw getAbortError(signal); + } + try { + return await operation(); + } catch (error) { + if (signal.aborted || !this.isCurrent(generation)) throw error; + if (retryNumber >= this.retryPolicy.maxNumberOfRetry) throw error; + retryNumber += 1; + const delayMs = getNdgrRetryDelayMs( + retryNumber, + this.retryPolicy, + this.random(), + ); + if (!(await waitForRetry(delayMs, signal))) throw getAbortError(signal); + } + } + } + private async consumeView( viewUri: string, at: number | "now", @@ -353,41 +523,64 @@ export class NiconicoCommentClient implements LiveCommentClient { vposBaseTime: number, signal: AbortSignal, generation: number, + ): Promise { + return this.runNdgrWithRetry(signal, generation, () => + this.consumeViewSegmentAttempt( + viewUri, + at, + vposBaseTime, + signal, + generation, + ), + ); + } + + private async consumeViewSegmentAttempt( + viewUri: string, + at: number | "now", + vposBaseTime: number, + signal: AbortSignal, + generation: number, ): Promise { const url = new URL(viewUri); url.searchParams.set("at", String(at)); - const response = await fetch(url, { credentials: "omit", signal }); - if (!response.ok || !response.body) - throw new Error(`NicoNico view request failed: ${response.status}`); - const reader = response.body.getReader(); - const frames = new LengthDelimitedReader(); - const segmentTasks: Promise[] = []; - let nextAt: number | undefined; + const attemptContext = createNdgrAttempt(signal); + let reader: ReadableStreamDefaultReader | undefined; try { + const response = await attemptContext.runWithTimeout( + () => + fetch(url, { + cache: "no-store", + credentials: "omit", + signal: attemptContext.signal, + }), + this.timeoutPolicy.viewFetchHeaderMs, + "view", + "fetch-header", + ); + if (!response.ok || !response.body) + throw new Error(`NicoNico view request failed: ${response.status}`); + reader = response.body.getReader(); + const frames = new LengthDelimitedReader(); + let nextAt: number | undefined; while (true) { - const result = await reader.read(); + const result = await attemptContext.runWithTimeout( + () => + reader?.read() ?? Promise.reject(new Error("View reader missing")), + this.timeoutPolicy.viewStreamReadMs, + "view", + "stream-read", + ); if (result.done) break; for (const frame of frames.push(result.value)) { const entry = decodeChunkedEntry(frame); if (entry.segment?.uri) { - const segmentTask = this.consumeSegment( + this.startSegment( entry.segment.uri, vposBaseTime, signal, generation, ); - segmentTasks.push( - segmentTask.catch((error) => { - if (!this.isIntentionalFailure(error, signal, generation)) { - // A segment is independent of the view stream. Its failure - // should not make an otherwise active live connection red. - console.error( - "[NicoJK] NicoNico comment segment failed", - error, - ); - } - }), - ); } if (entry.next) nextAt = entry.next.at; } @@ -395,11 +588,45 @@ export class NiconicoCommentClient implements LiveCommentClient { if (frames.hasPendingFrame()) { throw new Error("NicoNico view stream ended with a truncated frame"); } + return nextAt; } finally { - reader.releaseLock(); - await Promise.allSettled(segmentTasks); + try { + reader?.releaseLock(); + } catch { + // Aborting a timed-out read may keep the lock until rejection settles. + } + attemptContext.dispose(); } - return nextAt; + } + + private startSegment( + segmentUri: string, + vposBaseTime: number, + signal: AbortSignal, + generation: number, + ): void { + const active = this.activeSegmentTasks.get(segmentUri); + if (active?.generation === generation) return; + if (active) this.activeSegmentTasks.delete(segmentUri); + if (this.seenSegmentUris.has(segmentUri)) return; + this.seenSegmentUris.add(segmentUri); + + let task: Promise; + task = this.consumeSegment(segmentUri, vposBaseTime, signal, generation) + .catch((error) => { + if (this.activeSegmentTasks.get(segmentUri)?.task === task) { + this.seenSegmentUris.delete(segmentUri); + } + if (!this.isIntentionalFailure(error, signal, generation)) { + console.error("[NicoJK] NicoNico comment segment failed", error); + } + }) + .finally(() => { + if (this.activeSegmentTasks.get(segmentUri)?.task === task) { + this.activeSegmentTasks.delete(segmentUri); + } + }); + this.activeSegmentTasks.set(segmentUri, { generation, task }); } private async consumeSegment( @@ -408,27 +635,63 @@ export class NiconicoCommentClient implements LiveCommentClient { signal: AbortSignal, generation: number, ): Promise { - const response = await fetch(segmentUri, { credentials: "omit", signal }); - if (!response.ok || !response.body) - throw new Error(`NicoNico segment request failed: ${response.status}`); - const reader = response.body.getReader(); - const frames = new LengthDelimitedReader(); + return this.runNdgrWithRetry(signal, generation, () => + this.consumeSegmentAttempt(segmentUri, vposBaseTime, signal, generation), + ); + } + + private async consumeSegmentAttempt( + segmentUri: string, + vposBaseTime: number, + signal: AbortSignal, + generation: number, + ): Promise { + const attemptContext = createNdgrAttempt(signal); + let reader: ReadableStreamDefaultReader | undefined; try { + const response = await attemptContext.runWithTimeout( + () => + fetch(segmentUri, { + cache: "no-store", + credentials: "omit", + signal: attemptContext.signal, + }), + this.timeoutPolicy.segmentFetchHeaderMs, + "segment", + "fetch-header", + ); + if (!response.ok || !response.body) + throw new Error(`NicoNico segment request failed: ${response.status}`); + reader = response.body.getReader(); + const frames = new LengthDelimitedReader(); while (true) { - const result = await reader.read(); + const result = await attemptContext.runWithTimeout( + () => + reader?.read() ?? + Promise.reject(new Error("Segment reader missing")), + this.timeoutPolicy.segmentStreamReadMs, + "segment", + "stream-read", + ); if (result.done) break; for (const frame of frames.push(result.value)) { const message = decodeChunkedMessage(frame).message; const chat = message?.chat ?? message?.overflowedChat; - if (chat && this.isCurrent(generation)) + if (chat && this.isCurrent(generation)) { this.notifyListeners(this.toComment(chat, vposBaseTime)); + } } } if (frames.hasPendingFrame()) { throw new Error("NicoNico segment stream ended with a truncated frame"); } } finally { - reader.releaseLock(); + try { + reader?.releaseLock(); + } catch { + // Aborting a timed-out read may keep the lock until rejection settles. + } + attemptContext.dispose(); } } diff --git a/tests/nicojk-niconico-comment-client.test.ts b/tests/nicojk-niconico-comment-client.test.ts index 7d6e10d..da80567 100644 --- a/tests/nicojk-niconico-comment-client.test.ts +++ b/tests/nicojk-niconico-comment-client.test.ts @@ -4,6 +4,7 @@ import test from "node:test"; import protobuf from "@n-air-app/nicolive-comment-protobuf"; import { LengthDelimitedReader } from "../src/plugins/nicojk/ndgr-protobuf"; import { + getNdgrRetryDelayMs, getNiconicoLiveVpos, modifierToMail, NiconicoCommentClient, @@ -183,6 +184,19 @@ test("anchors live comments to receive time while preserving their timestamp", ( ); }); +test("uses the official NDGR exponential retry delay and jitter", () => { + const policy = { + maxNumberOfRetry: 5, + startingIntervalMs: 500, + timeMultiple: 1.5, + randomizationFactor: 0.5, + }; + assert.equal(getNdgrRetryDelayMs(1, policy, 0), 250); + assert.equal(getNdgrRetryDelayMs(1, policy, 0.5), 500); + assert.equal(getNdgrRetryDelayMs(1, policy, 1), 750); + assert.equal(getNdgrRetryDelayMs(2, policy, 0.5), 750); +}); + test("moves to error without fetching when the community ID is missing", () => { FakeWebSocket.instances = []; const originalFetch = globalThis.fetch; @@ -339,6 +353,88 @@ test("receives anonymous NDGR comments through the shared client boundary", asyn } }); +test("advances the view while an announced segment is still streaming", async () => { + FakeWebSocket.instances = []; + const originalWebSocket = globalThis.WebSocket; + const originalFetch = globalThis.fetch; + const segment = frame( + Uint8Array.from( + ChunkedEntry.encode({ + segment: MessageSegment.create({ + uri: "https://example.test/segment/pending", + }), + }).finish(), + ), + ); + const next = frame( + Uint8Array.from(ChunkedEntry.encode({ next: { at: 123 } }).finish()), + ); + const page = ``; + let segmentSignal: AbortSignal | undefined; + let nextViewSignal: AbortSignal | undefined; + let nextViewStarted = false; + let segmentRequests = 0; + ( + globalThis as typeof globalThis & { WebSocket: typeof FakeWebSocket } + ).WebSocket = FakeWebSocket as never; + globalThis.fetch = async (input, init) => { + const url = String(input); + if (url.startsWith("https://live.nicovideo.jp/watch/")) { + const response = new Response(page, { status: 200 }); + Object.defineProperty(response, "url", { + value: "https://live.nicovideo.jp/watch/lv1", + }); + return response; + } + if (url.includes("/view?at=now")) { + return new Response(streamOf(concat(segment, next)), { status: 200 }); + } + if (url.includes("/view?at=123")) { + if (!init?.signal) throw new Error("next view signal is missing"); + nextViewStarted = true; + nextViewSignal = init.signal; + return new Response(streamUntilAbort([], init.signal), { status: 200 }); + } + if (url.includes("/segment/pending")) { + if (!init?.signal) throw new Error("segment signal is missing"); + segmentRequests += 1; + segmentSignal = init.signal; + return new Response(streamUntilAbort([], init.signal), { status: 200 }); + } + throw new Error(`Unexpected request: ${url}`); + }; + + try { + const client = new NiconicoCommentClient(); + client.connect(source()); + await new Promise((resolve) => setTimeout(resolve, 0)); + const socket = FakeWebSocket.instances[0]; + assert.ok(socket); + socket.open(); + socket.message( + JSON.stringify({ + type: "messageServer", + data: { viewUri: "https://example.test/view" }, + }), + ); + await new Promise((resolve) => setTimeout(resolve, 20)); + assert.equal(nextViewStarted, true); + assert.equal(segmentRequests, 1); + assert.ok(segmentSignal); + assert.ok(nextViewSignal); + + client.disconnect(); + await new Promise((resolve) => setTimeout(resolve, 0)); + assert.equal(segmentSignal.aborted, true); + assert.equal(nextViewSignal.aborted, true); + } finally { + globalThis.fetch = originalFetch; + ( + globalThis as typeof globalThis & { WebSocket: typeof WebSocket } + ).WebSocket = originalWebSocket; + } +}); + test("propagates comments across clients via BroadcastChannel", async () => { FakeWebSocket.instances = []; const originalWebSocket = globalThis.WebSocket; @@ -454,7 +550,7 @@ test("propagates comments across clients via BroadcastChannel", async () => { } }); -test("keeps connected status when an individual segment fails", async () => { +test("keeps the session connected when an individual segment fails", async () => { FakeWebSocket.instances = []; const originalWebSocket = globalThis.WebSocket; const originalFetch = globalThis.fetch; @@ -493,8 +589,10 @@ test("keeps connected status when an individual segment fails", async () => { }); }; + const client = new NiconicoCommentClient({ + retry: { maxNumberOfRetry: 0 }, + }); try { - const client = new NiconicoCommentClient(); const statuses: string[] = []; client.onStatusUpdate((status) => statuses.push(status)); client.connect(source()); @@ -510,6 +608,7 @@ test("keeps connected status when an individual segment fails", async () => { ); await new Promise((resolve) => setTimeout(resolve, 30)); assert.equal(statuses.at(-1), "connected"); + assert.equal(FakeWebSocket.instances.length, 1); assert.ok( errors.some((args) => args.some( @@ -519,8 +618,94 @@ test("keeps connected status when an individual segment fails", async () => { ), ), ); + } finally { client.disconnect(); + console.error = originalConsoleError; + globalThis.fetch = originalFetch; + ( + globalThis as typeof globalThis & { WebSocket: typeof WebSocket } + ).WebSocket = originalWebSocket; + } +}); + +test("fetches a segment again when the view announces it after retries are exhausted", async () => { + FakeWebSocket.instances = []; + const originalWebSocket = globalThis.WebSocket; + const originalFetch = globalThis.fetch; + const originalConsoleError = console.error; + const segmentUri = "https://example.test/segment/reannounced"; + const segmentEntry = frame( + Uint8Array.from( + ChunkedEntry.encode({ + segment: MessageSegment.create({ uri: segmentUri }), + }).finish(), + ), + ); + const nextEntry = frame( + Uint8Array.from(ChunkedEntry.encode({ next: { at: 123 } }).finish()), + ); + const page = ``; + let segmentRequests = 0; + console.error = () => {}; + ( + globalThis as typeof globalThis & { WebSocket: typeof FakeWebSocket } + ).WebSocket = FakeWebSocket as never; + globalThis.fetch = async (input, init) => { + const url = String(input); + if (url.startsWith("https://live.nicovideo.jp/watch/")) { + const response = new Response(page, { status: 200 }); + Object.defineProperty(response, "url", { + value: "https://live.nicovideo.jp/watch/lv1", + }); + return response; + } + if (url.includes("/view?at=now")) { + return new Response(streamOf(concat(segmentEntry, nextEntry)), { + status: 200, + }); + } + if (url.includes("/view?at=123")) { + await new Promise((resolve) => setTimeout(resolve, 10)); + if (!init?.signal) throw new Error("view signal is missing"); + return new Response(streamUntilAbort([segmentEntry], init.signal), { + status: 200, + }); + } + if (!init?.signal) throw new Error("segment signal is missing"); + segmentRequests += 1; + if (segmentRequests === 1) { + return new Response( + streamThatErrorsAfter( + new Uint8Array(), + new Error("first segment attempt failed"), + ), + { status: 200 }, + ); + } + return new Response(streamOf(), { status: 200 }); + }; + + const client = new NiconicoCommentClient({ + retry: { maxNumberOfRetry: 0 }, + }); + try { + client.connect(source()); + await new Promise((resolve) => setTimeout(resolve, 0)); + const socket = FakeWebSocket.instances[0]; + assert.ok(socket); + socket.open(); + socket.message( + JSON.stringify({ + type: "messageServer", + data: { viewUri: "https://example.test/view" }, + }), + ); + + await new Promise((resolve) => setTimeout(resolve, 30)); + assert.equal(segmentRequests, 2); + assert.equal(client.getStatus(), "connected"); } finally { + client.disconnect(); console.error = originalConsoleError; globalThis.fetch = originalFetch; ( @@ -529,7 +714,179 @@ test("keeps connected status when an individual segment fails", async () => { } }); -test("aborts view and segment streams when NicoNico disconnects", async () => { +test("retries a timed-out segment without closing the view", async () => { + FakeWebSocket.instances = []; + const originalWebSocket = globalThis.WebSocket; + const originalFetch = globalThis.fetch; + const view = frame( + Uint8Array.from( + ChunkedEntry.encode({ + segment: MessageSegment.create({ + uri: "https://example.test/segment", + }), + }).finish(), + ), + ); + const page = ``; + let watchPageRequests = 0; + let segmentRequests = 0; + let viewSignal: AbortSignal | undefined; + ( + globalThis as typeof globalThis & { WebSocket: typeof FakeWebSocket } + ).WebSocket = FakeWebSocket as never; + globalThis.fetch = async (input, init) => { + const url = String(input); + if (url.startsWith("https://live.nicovideo.jp/watch/")) { + watchPageRequests += 1; + const response = new Response(page, { status: 200 }); + Object.defineProperty(response, "url", { + value: "https://live.nicovideo.jp/watch/lv1", + }); + return response; + } + if (url.includes("/view?at=now")) { + if (!init?.signal) throw new Error("view signal is missing"); + viewSignal = init.signal; + return new Response(streamUntilAbort([view], init.signal), { + status: 200, + }); + } + if (!init?.signal) throw new Error("segment signal is missing"); + segmentRequests += 1; + if (segmentRequests === 2) { + return new Response(streamOf(), { status: 200 }); + } + return new Promise((_resolve, reject) => { + init.signal?.addEventListener( + "abort", + () => reject(new DOMException("The stream was aborted", "AbortError")), + { once: true }, + ); + }); + }; + + const client = new NiconicoCommentClient({ + retry: { + maxNumberOfRetry: 1, + startingIntervalMs: 1, + timeMultiple: 1, + randomizationFactor: 0, + }, + timeouts: { segmentFetchHeaderMs: 5 }, + }); + try { + client.connect(source()); + await new Promise((resolve) => setTimeout(resolve, 0)); + const socket = FakeWebSocket.instances[0]; + assert.ok(socket); + socket.open(); + socket.message( + JSON.stringify({ + type: "messageServer", + data: { viewUri: "https://example.test/view" }, + }), + ); + + await new Promise((resolve) => setTimeout(resolve, 30)); + assert.equal(viewSignal?.aborted, false); + assert.equal(watchPageRequests, 1); + assert.equal(segmentRequests, 2); + assert.equal(FakeWebSocket.instances.length, 1); + assert.equal(client.getStatus(), "connected"); + } finally { + client.disconnect(); + globalThis.fetch = originalFetch; + ( + globalThis as typeof globalThis & { WebSocket: typeof WebSocket } + ).WebSocket = originalWebSocket; + } +}); + +test("retries a failed view without aborting an active segment", async () => { + FakeWebSocket.instances = []; + const originalWebSocket = globalThis.WebSocket; + const originalFetch = globalThis.fetch; + const viewError = new Error("view stream failed"); + const view = frame( + Uint8Array.from( + ChunkedEntry.encode({ + segment: MessageSegment.create({ + uri: "https://example.test/segment", + }), + }).finish(), + ), + ); + const page = ``; + let viewRequests = 0; + let segmentRequests = 0; + let segmentSignal: AbortSignal | undefined; + ( + globalThis as typeof globalThis & { WebSocket: typeof FakeWebSocket } + ).WebSocket = FakeWebSocket as never; + globalThis.fetch = async (input, init) => { + const url = String(input); + if (url.startsWith("https://live.nicovideo.jp/watch/")) { + const response = new Response(page, { status: 200 }); + Object.defineProperty(response, "url", { + value: "https://live.nicovideo.jp/watch/lv1", + }); + return response; + } + if (url.includes("/view?at=now")) { + if (!init?.signal) throw new Error("view signal is missing"); + viewRequests += 1; + return new Response( + viewRequests === 1 + ? streamThatErrorsAfter(view, viewError) + : streamUntilAbort([], init.signal), + { status: 200 }, + ); + } + if (url.includes("/segment")) { + if (!init?.signal) throw new Error("segment signal is missing"); + segmentRequests += 1; + segmentSignal = init.signal; + return new Response(streamUntilAbort([], init.signal), { status: 200 }); + } + throw new Error(`Unexpected request: ${url}`); + }; + + const client = new NiconicoCommentClient({ + retry: { + maxNumberOfRetry: 1, + startingIntervalMs: 1, + timeMultiple: 1, + randomizationFactor: 0, + }, + }); + try { + client.connect(source()); + await new Promise((resolve) => setTimeout(resolve, 0)); + const socket = FakeWebSocket.instances[0]; + assert.ok(socket); + socket.open(); + socket.message( + JSON.stringify({ + type: "messageServer", + data: { viewUri: "https://example.test/view" }, + }), + ); + await new Promise((resolve) => setTimeout(resolve, 30)); + assert.equal(viewRequests, 2); + assert.equal(segmentRequests, 1); + assert.equal(segmentSignal?.aborted, false); + assert.equal(FakeWebSocket.instances.length, 1); + assert.equal(client.getStatus(), "connected"); + } finally { + client.disconnect(); + globalThis.fetch = originalFetch; + ( + globalThis as typeof globalThis & { WebSocket: typeof WebSocket } + ).WebSocket = originalWebSocket; + } +}); + +test("closes the watch socket after receiving the View URI without aborting NDGR streams", async () => { FakeWebSocket.instances = []; const originalWebSocket = globalThis.WebSocket; const originalFetch = globalThis.fetch; @@ -570,8 +927,10 @@ test("aborts view and segment streams when NicoNico disconnects", async () => { throw new Error(`Unexpected request: ${url}`); }; + const client = new NiconicoCommentClient({ + retry: { maxNumberOfRetry: 0 }, + }); try { - const client = new NiconicoCommentClient(); client.connect(source()); await new Promise((resolve) => setTimeout(resolve, 0)); const socket = FakeWebSocket.instances[0]; @@ -587,18 +946,16 @@ test("aborts view and segment streams when NicoNico disconnects", async () => { assert.ok(viewSignal); assert.ok(segmentSignal); - socket.message( - JSON.stringify({ - type: "disconnect", - data: { reason: "PING_TIMEOUT" }, - }), - ); + assert.equal(socket.readyState, 3); + // A message after close cannot happen in browsers, but FakeWebSocket lets us + // verify that an obsolete watch socket cannot abort the HTTP streams. + socket.message(JSON.stringify({ type: "disconnect", data: {} })); await new Promise((resolve) => setTimeout(resolve, 10)); - assert.equal(viewSignal.aborted, true); - assert.equal(segmentSignal.aborted, true); - assert.equal(client.getStatus(), "connecting"); - client.disconnect(); + assert.equal(viewSignal.aborted, false); + assert.equal(segmentSignal.aborted, false); + assert.equal(client.getStatus(), "connected"); } finally { + client.disconnect(); globalThis.fetch = originalFetch; ( globalThis as typeof globalThis & { WebSocket: typeof WebSocket } @@ -606,7 +963,7 @@ test("aborts view and segment streams when NicoNico disconnects", async () => { } }); -test("reconnects after the view stream fails without an unhandled rejection", async () => { +test("reconnects after view retries are exhausted without an unhandled rejection", async () => { FakeWebSocket.instances = []; const originalWebSocket = globalThis.WebSocket; const originalFetch = globalThis.fetch; @@ -649,8 +1006,10 @@ test("reconnects after the view stream fails without an unhandled rejection", as ); }; + const client = new NiconicoCommentClient({ + retry: { maxNumberOfRetry: 0 }, + }); try { - const client = new NiconicoCommentClient(); const statuses: string[] = []; client.onStatusUpdate((status) => statuses.push(status)); client.connect(source()); @@ -678,8 +1037,8 @@ test("reconnects after the view stream fails without an unhandled rejection", as ); await new Promise((resolve) => setTimeout(resolve, 5_100)); assert.equal(FakeWebSocket.instances.length, 2); - client.disconnect(); } finally { + client.disconnect(); process.removeListener("unhandledRejection", onUnhandledRejection); console.error = originalConsoleError; globalThis.fetch = originalFetch; diff --git a/tests/nicojk-overlay-page.test.ts b/tests/nicojk-overlay-page.test.ts new file mode 100644 index 0000000..cdc460e --- /dev/null +++ b/tests/nicojk-overlay-page.test.ts @@ -0,0 +1,42 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import type { NiconicoComment } from "../src/plugins/nicojk/comment-client"; +import { getLiveRendererSeedComments } from "../src/plugins/nicojk/components/OverlayPage"; + +function comment(vpos: number, mail: string[] = []): NiconicoComment { + return { + id: vpos, + no: vpos, + vpos, + content: "test", + date: 0, + date_usec: 0, + mail, + user_id: "test", + premium: 0, + anonymity: 1, + origin: "ws", + }; +} + +test("live renderer seeds only comments that can still be drawn", () => { + const nowVpos = 10_000; + assert.deepEqual( + getLiveRendererSeedComments( + [comment(nowVpos - 426), comment(nowVpos - 425), comment(nowVpos + 200)], + nowVpos, + ).map((item) => item.id), + [nowVpos - 425, nowVpos + 200], + ); +}); + +test("live renderer preserves a visible long comment across a rebuild", () => { + const nowVpos = 10_000; + assert.deepEqual( + getLiveRendererSeedComments( + [comment(nowVpos - 1_300, ["@12"]), comment(nowVpos - 1_326, ["@12"])], + nowVpos, + ).map((item) => item.id), + [nowVpos - 1_300], + ); +}); diff --git a/tests/nicojk-settings.test.ts b/tests/nicojk-settings.test.ts index f9c89c4..a02318a 100644 --- a/tests/nicojk-settings.test.ts +++ b/tests/nicojk-settings.test.ts @@ -3,11 +3,12 @@ import test from "node:test"; import { normalizeLiveCommentSource } from "../src/plugins/nicojk/ng-settings"; -test("defaults live comments to anonymous niconico", () => { - assert.equal(normalizeLiveCommentSource(undefined), "niconico"); - assert.equal(normalizeLiveCommentSource("unexpected"), "niconico"); +test("defaults live comments to nx-jikkyo", () => { + assert.equal(normalizeLiveCommentSource(undefined), "nx-jikkyo"); + assert.equal(normalizeLiveCommentSource("unexpected"), "nx-jikkyo"); }); -test("allows explicit nx-jikkyo live comments", () => { +test("allows explicit live comment sources", () => { + assert.equal(normalizeLiveCommentSource("niconico"), "niconico"); assert.equal(normalizeLiveCommentSource("nx-jikkyo"), "nx-jikkyo"); });