mirror of
https://github.com/Shik3i/KoalaSync.git
synced 2026-08-30 12:29:27 +00:00
feat(sync): coalesce offline media intent
This commit is contained in:
+25
-2
@@ -50,8 +50,31 @@ relay media event.
|
|||||||
|
|
||||||
Force Sync remains a two-phase ACK protocol. `PREPARE` is temporary choreography;
|
Force Sync remains a two-phase ACK protocol. `PREPARE` is temporary choreography;
|
||||||
the matching `EXECUTE` commits its validated target to canonical state. Per-sender
|
the matching `EXECUTE` commits its validated target to canonical state. Per-sender
|
||||||
`seq`, peer heartbeats, and the existing reconnect event queue remain separate.
|
`seq`, peer heartbeats, and the reconnect queue remain separate mechanisms.
|
||||||
Offline media-command compaction is intentionally deferred.
|
|
||||||
|
## 3.2 Offline Media Intent
|
||||||
|
|
||||||
|
Canonical Media State and Offline Media Intent have different ownership:
|
||||||
|
|
||||||
|
- Canonical Media State is the relay's last accepted shared playback truth.
|
||||||
|
- Offline Media Intent is one room-scoped client representation of local
|
||||||
|
`PLAY`/`PAUSE`/`SEEK` commands that have not reached the relay yet.
|
||||||
|
|
||||||
|
Contiguous offline controls merge into a bounded logical queue entry. Every
|
||||||
|
retained coordination event, including Force Sync and Episode Lobby events, is
|
||||||
|
an ordering barrier. Stale offline `PING`, `PONG`, heartbeat `PEER_STATUS`, and
|
||||||
|
`EVENT_ACK` frames are not persisted because they are no longer meaningful after
|
||||||
|
reconnect. Force Sync ACK remains transactional and is not dropped or merged.
|
||||||
|
|
||||||
|
Media intent waits for the reconnecting room's `ROOM_DATA`. An authorized intent
|
||||||
|
takes precedence over the older canonical snapshot, materializes into the
|
||||||
|
minimum ordered legacy `SEEK` plus `PLAY`/`PAUSE` frames needed by old peers, and
|
||||||
|
thereby advances relay canonical state normally. With no intent, canonical
|
||||||
|
recovery is unchanged. Role loss discards room-driving intent before recovery;
|
||||||
|
intentional Host Control solo mode and an active Episode Lobby remain
|
||||||
|
authoritative. MV3 session restoration migrates the previous raw queue format,
|
||||||
|
preserves barriers, repairs `localSeq` monotonically, and rejects another room's
|
||||||
|
intent.
|
||||||
|
|
||||||
## 4. Episode Auto-Sync
|
## 4. Episode Auto-Sync
|
||||||
Maintains continuous synchronized viewing when watching series:
|
Maintains continuous synchronized viewing when watching series:
|
||||||
|
|||||||
+23
-2
@@ -141,8 +141,29 @@ relaying the established event names, payloads, and order unchanged. This makes
|
|||||||
server-first rollout safe: old clients populate recovery state without needing to
|
server-first rollout safe: old clients populate recovery state without needing to
|
||||||
understand or acknowledge it, and new clients consume it only when the relay
|
understand or acknowledge it, and new clients consume it only when the relay
|
||||||
advertises the capability. No protocol-version or minimum-version bump is
|
advertises the capability. No protocol-version or minimum-version bump is
|
||||||
required. Offline `play`/`pause`/`seek` compaction is planned separately and is
|
required. Offline `play`/`pause`/`seek` compaction remains the separate
|
||||||
not part of Media State v1.
|
client-owned layer described below rather than part of the relay capability.
|
||||||
|
|
||||||
|
### Offline media intent
|
||||||
|
|
||||||
|
Offline media intent is client-side queue state, not a relay protocol feature.
|
||||||
|
An updated extension coalesces contiguous unsent `play`, `pause`, and `seek`
|
||||||
|
commands for one room. Retained non-media events are ordering barriers. On a
|
||||||
|
successful rejoin, the extension first reads `room_data` so current Host Control
|
||||||
|
and Episode Lobby authority can be applied, then replays an authorized intent as
|
||||||
|
the minimum existing legacy media-event sequence. Actual wire frames, rather
|
||||||
|
than logical queue entries, consume the paced reconnect budget.
|
||||||
|
|
||||||
|
Pending authorized local intent takes precedence over an older canonical
|
||||||
|
snapshot because it has not yet been accepted by the relay. Its legacy replay
|
||||||
|
then updates canonical state like any other accepted control. Without pending
|
||||||
|
intent, canonical recovery proceeds normally. Intent made stale by a room switch,
|
||||||
|
role loss, intentional solo mode, or an active Episode Lobby is discarded and
|
||||||
|
cannot suppress server recovery.
|
||||||
|
|
||||||
|
This requires no event, capability, ACK, protocol-version, or minimum-version
|
||||||
|
change. Old relays receive ordinary `play`/`pause`/`seek`; old peers see only the
|
||||||
|
same existing relayed events.
|
||||||
|
|
||||||
## Ephemeral encrypted chat
|
## Ephemeral encrypted chat
|
||||||
|
|
||||||
|
|||||||
+177
-51
@@ -9,6 +9,19 @@ import { buildChatRelayPayload, encodeSocketEvent } from './chat-wire.js';
|
|||||||
import { createChatEchoTracker, createChatSendLimiter, createLatestTaskQueue, normalizeRoomId, shouldShowChatNotification } from './chat-session.js';
|
import { createChatEchoTracker, createChatSendLimiter, createLatestTaskQueue, normalizeRoomId, shouldShowChatNotification } from './chat-session.js';
|
||||||
import { createChatActivityStore } from './chat-activity.js';
|
import { createChatActivityStore } from './chat-activity.js';
|
||||||
import { canonicalMediaStateFromRoomData, createCanonicalMediaStateTracker } from './canonical-media-state.js';
|
import { canonicalMediaStateFromRoomData, createCanonicalMediaStateTracker } from './canonical-media-state.js';
|
||||||
|
import {
|
||||||
|
drainQueuedBatch,
|
||||||
|
enqueueQueuedEvent,
|
||||||
|
isMediaQueueEvent,
|
||||||
|
isQueuedMediaIntent,
|
||||||
|
maxQueuedSequence,
|
||||||
|
mediaIntentNeedsSequenceReservation,
|
||||||
|
normalizePersistedEventQueue,
|
||||||
|
queuedMediaIntentCount,
|
||||||
|
queuedWireCount,
|
||||||
|
reconcileQueuedRoomIntent,
|
||||||
|
reserveLatestMediaIntentSequence
|
||||||
|
} from './offline-media-intent.js';
|
||||||
import { HOST_ACCESS_REQUIRED_STATUS, normalizeTabId, inspectTabHostAccess, isHostAccessError, addTabHostAccessRequest, removeTabHostAccessRequest } from './host-access.js';
|
import { HOST_ACCESS_REQUIRED_STATUS, normalizeTabId, inspectTabHostAccess, isHostAccessError, addTabHostAccessRequest, removeTabHostAccessRequest } from './host-access.js';
|
||||||
import {
|
import {
|
||||||
MEDIA_FRAME_ACCESS_REQUIRED,
|
MEDIA_FRAME_ACCESS_REQUIRED,
|
||||||
@@ -234,20 +247,16 @@ let storageInitialized = false;
|
|||||||
let pendingLogs = [];
|
let pendingLogs = [];
|
||||||
let pendingHistory = [];
|
let pendingHistory = [];
|
||||||
let eventQueue = [];
|
let eventQueue = [];
|
||||||
|
let eventQueueVersion = 0;
|
||||||
let flushTimer = null; // paces draining of eventQueue after (re)connect
|
let flushTimer = null; // paces draining of eventQueue after (re)connect
|
||||||
|
let flushInProgress = false;
|
||||||
let isNamespaceJoined = false;
|
let isNamespaceJoined = false;
|
||||||
|
let awaitingRoomData = false;
|
||||||
|
let pendingRoomDataRoomId = null;
|
||||||
let lastActionState = { action: null, senderId: null, timestamp: 0, acks: [] };
|
let lastActionState = { action: null, senderId: null, timestamp: 0, acks: [] };
|
||||||
let localSeq = 0; // Monotonically increasing command sequence for this peer
|
let localSeq = 0; // Monotonically increasing command sequence for this peer
|
||||||
const lastSeqBySender = {}; // senderId → last received seq (stale command guard)
|
const lastSeqBySender = {}; // senderId → last received seq (stale command guard)
|
||||||
const canonicalMediaStateTracker = createCanonicalMediaStateTracker();
|
const canonicalMediaStateTracker = createCanonicalMediaStateTracker();
|
||||||
const CANONICAL_MEDIA_EVENTS = new Set([
|
|
||||||
EVENTS.PLAY,
|
|
||||||
EVENTS.PAUSE,
|
|
||||||
EVENTS.SEEK,
|
|
||||||
EVENTS.FORCE_SYNC_PREPARE,
|
|
||||||
EVENTS.FORCE_SYNC_EXECUTE
|
|
||||||
]);
|
|
||||||
let flushedMediaControlsBeforeRoomData = false;
|
|
||||||
|
|
||||||
// --- Host Control Mode ---
|
// --- Host Control Mode ---
|
||||||
let controlMode = CONTROL_MODES.EVERYONE; // 'everyone' | 'host-only'
|
let controlMode = CONTROL_MODES.EVERYONE; // 'everyone' | 'host-only'
|
||||||
@@ -296,6 +305,7 @@ function clearChatActivity() {
|
|||||||
async function clearFailedJoinCredentials() {
|
async function clearFailedJoinCredentials() {
|
||||||
webJoinCoordinator.invalidate();
|
webJoinCoordinator.invalidate();
|
||||||
connectIntent = false;
|
connectIntent = false;
|
||||||
|
clearEventQueue();
|
||||||
clearCanonicalMediaRecovery();
|
clearCanonicalMediaRecovery();
|
||||||
chatSecretGuard = '';
|
chatSecretGuard = '';
|
||||||
invalidateChatSession();
|
invalidateChatSession();
|
||||||
@@ -442,7 +452,6 @@ function ensureState() {
|
|||||||
);
|
);
|
||||||
if (data.lastActionState) lastActionState = data.lastActionState;
|
if (data.lastActionState) lastActionState = data.lastActionState;
|
||||||
|
|
||||||
if (data.eventQueue) eventQueue = [...eventQueue, ...data.eventQueue].slice(0, 50);
|
|
||||||
if (data.isForceSyncInitiator !== undefined && isForceSyncInitiator === false) {
|
if (data.isForceSyncInitiator !== undefined && isForceSyncInitiator === false) {
|
||||||
isForceSyncInitiator = data.isForceSyncInitiator;
|
isForceSyncInitiator = data.isForceSyncInitiator;
|
||||||
}
|
}
|
||||||
@@ -482,10 +491,19 @@ function ensureState() {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if (data.localSeq !== undefined && !isNaN(data.localSeq)) localSeq = data.localSeq;
|
if (Number.isSafeInteger(data.localSeq) && data.localSeq >= 0) localSeq = data.localSeq;
|
||||||
|
eventQueue = normalizePersistedEventQueue(
|
||||||
|
[...eventQueue, ...(Array.isArray(data.eventQueue) ? data.eventQueue : [])],
|
||||||
|
currentRoom?.roomId || null
|
||||||
|
);
|
||||||
|
eventQueueVersion++;
|
||||||
|
// An MV3 restart can restore queue and sequence writes in either
|
||||||
|
// order. Never let the sender sequence fall behind persisted work.
|
||||||
|
localSeq = Math.max(localSeq, maxQueuedSequence(eventQueue));
|
||||||
if (data.lastSeqBySender && typeof data.lastSeqBySender === 'object') Object.assign(lastSeqBySender, data.lastSeqBySender);
|
if (data.lastSeqBySender && typeof data.lastSeqBySender === 'object') Object.assign(lastSeqBySender, data.lastSeqBySender);
|
||||||
|
|
||||||
storageInitialized = true;
|
storageInitialized = true;
|
||||||
|
chrome.storage.session.set({ eventQueue, localSeq }).catch(() => {});
|
||||||
|
|
||||||
// Process any early logs/history that weren't captured in the spread
|
// Process any early logs/history that weren't captured in the spread
|
||||||
if (pendingLogs.length > 0) {
|
if (pendingLogs.length > 0) {
|
||||||
@@ -728,6 +746,8 @@ function forceDisconnect() {
|
|||||||
currentServerUrl = null;
|
currentServerUrl = null;
|
||||||
isConnecting = false;
|
isConnecting = false;
|
||||||
isNamespaceJoined = false;
|
isNamespaceJoined = false;
|
||||||
|
awaitingRoomData = false;
|
||||||
|
pendingRoomDataRoomId = null;
|
||||||
invalidateChatSession();
|
invalidateChatSession();
|
||||||
isForceSyncInitiator = false;
|
isForceSyncInitiator = false;
|
||||||
expectedAcksCount = 0;
|
expectedAcksCount = 0;
|
||||||
@@ -735,7 +755,9 @@ function forceDisconnect() {
|
|||||||
lastContentHeartbeatAt = null;
|
lastContentHeartbeatAt = null;
|
||||||
forceSyncAcks.clear();
|
forceSyncAcks.clear();
|
||||||
if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; }
|
if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; }
|
||||||
|
flushInProgress = false;
|
||||||
eventQueue = [];
|
eventQueue = [];
|
||||||
|
eventQueueVersion++;
|
||||||
chrome.storage.session.set({
|
chrome.storage.session.set({
|
||||||
isForceSyncInitiator: false,
|
isForceSyncInitiator: false,
|
||||||
forceSyncAcks: [],
|
forceSyncAcks: [],
|
||||||
@@ -1082,6 +1104,7 @@ async function connect() {
|
|||||||
try {
|
try {
|
||||||
if (!peerId) peerId = await getPeerId();
|
if (!peerId) peerId = await getPeerId();
|
||||||
settings = await getSettings();
|
settings = await getSettings();
|
||||||
|
pendingRoomDataRoomId = settings.roomId || currentRoom?.roomId || null;
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
throw new Error(`[Storage Error] ${e.message}`);
|
throw new Error(`[Storage Error] ${e.message}`);
|
||||||
}
|
}
|
||||||
@@ -1180,6 +1203,8 @@ async function connect() {
|
|||||||
addLog('Joined Namespace /', 'success');
|
addLog('Joined Namespace /', 'success');
|
||||||
const settings = await getSettings();
|
const settings = await getSettings();
|
||||||
if (settings.roomId) {
|
if (settings.roomId) {
|
||||||
|
awaitingRoomData = true;
|
||||||
|
pendingRoomDataRoomId = settings.roomId;
|
||||||
const sharedTitles = getSharedTitleFields(settings);
|
const sharedTitles = getSharedTitleFields(settings);
|
||||||
emit(EVENTS.JOIN_ROOM, {
|
emit(EVENTS.JOIN_ROOM, {
|
||||||
roomId: settings.roomId,
|
roomId: settings.roomId,
|
||||||
@@ -1190,14 +1215,11 @@ async function connect() {
|
|||||||
clientCapabilities: CLIENT_CAPABILITIES,
|
clientCapabilities: CLIENT_CAPABILITIES,
|
||||||
protocolVersion: PROTOCOL_VERSION
|
protocolVersion: PROTOCOL_VERSION
|
||||||
});
|
});
|
||||||
|
} else {
|
||||||
|
awaitingRoomData = false;
|
||||||
|
pendingRoomDataRoomId = null;
|
||||||
|
flushEventQueue();
|
||||||
}
|
}
|
||||||
// Preserve the existing immediate queue-drain behavior. Remember
|
|
||||||
// whether local media intent was flushed before ROOM_DATA so an
|
|
||||||
// older snapshot cannot snap the player backward on reconnect.
|
|
||||||
flushedMediaControlsBeforeRoomData = eventQueue.some(item =>
|
|
||||||
item && CANONICAL_MEDIA_EVENTS.has(item.event)
|
|
||||||
);
|
|
||||||
flushEventQueue();
|
|
||||||
} else if (msg.startsWith('42')) {
|
} else if (msg.startsWith('42')) {
|
||||||
try {
|
try {
|
||||||
const payload = JSON.parse(msg.substring(2));
|
const payload = JSON.parse(msg.substring(2));
|
||||||
@@ -1215,6 +1237,8 @@ async function connect() {
|
|||||||
socket.onclose = () => {
|
socket.onclose = () => {
|
||||||
isConnecting = false;
|
isConnecting = false;
|
||||||
isNamespaceJoined = false;
|
isNamespaceJoined = false;
|
||||||
|
awaitingRoomData = false;
|
||||||
|
pendingRoomDataRoomId = null;
|
||||||
invalidateChatSession();
|
invalidateChatSession();
|
||||||
stopPing();
|
stopPing();
|
||||||
if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; }
|
if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; }
|
||||||
@@ -1490,7 +1514,10 @@ function scheduleReconnect() {
|
|||||||
// Slow reconnect logic is now handled in the keepAlive alarm
|
// Slow reconnect logic is now handled in the keepAlive alarm
|
||||||
|
|
||||||
function emit(event, data) {
|
function emit(event, data) {
|
||||||
if (socket && socket.readyState === WebSocket.OPEN && isNamespaceJoined) {
|
const mustWaitForRoomData = awaitingRoomData
|
||||||
|
&& event !== EVENTS.JOIN_ROOM
|
||||||
|
&& event !== EVENTS.GET_ROOMS;
|
||||||
|
if (socket && socket.readyState === WebSocket.OPEN && isNamespaceJoined && !mustWaitForRoomData) {
|
||||||
try {
|
try {
|
||||||
const msg = encodeSocketEvent(event, data, chatSecretGuard);
|
const msg = encodeSocketEvent(event, data, chatSecretGuard);
|
||||||
socket.send(msg);
|
socket.send(msg);
|
||||||
@@ -1522,41 +1549,120 @@ function emitLive(event, data) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
function queueEvent(event, data) {
|
function queueEvent(event, data) {
|
||||||
eventQueue.push({ event, data });
|
const queueRoomId = currentRoom?.roomId || pendingRoomDataRoomId;
|
||||||
if (eventQueue.length > 50) {
|
const queued = enqueueQueuedEvent(eventQueue, event, data, { roomId: queueRoomId });
|
||||||
eventQueue.shift();
|
eventQueue = queued.queue;
|
||||||
|
eventQueueVersion++;
|
||||||
|
|
||||||
|
if (isMediaQueueEvent(event)) {
|
||||||
|
const latest = eventQueue.at(-1);
|
||||||
|
if (mediaIntentNeedsSequenceReservation(latest)) {
|
||||||
|
const nextSequence = Math.max(localSeq, maxQueuedSequence(eventQueue)) + 1;
|
||||||
|
const reserved = reserveLatestMediaIntentSequence(eventQueue, queueRoomId, nextSequence);
|
||||||
|
eventQueue = reserved.queue;
|
||||||
|
if (reserved.reserved) {
|
||||||
|
localSeq = nextSequence;
|
||||||
|
chrome.storage.session.set({ localSeq }).catch(() => {});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
const sourceCount = isQueuedMediaIntent(eventQueue.at(-1))
|
||||||
|
? eventQueue.at(-1).intent.sourceEventCount
|
||||||
|
: 0;
|
||||||
|
if (sourceCount === 1) {
|
||||||
|
addLog('Offline media intent queued; replay deferred until ROOM_DATA', 'info');
|
||||||
|
} else if ([10, 50, 100, 500, 1000].includes(sourceCount)) {
|
||||||
|
addLog(`Collapsed ${sourceCount} queued media commands into one intent`, 'info');
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (queued.dropped > 0) {
|
||||||
addLog('Event queue cap reached, dropping oldest event', 'warn');
|
addLog('Event queue cap reached, dropping oldest event', 'warn');
|
||||||
}
|
}
|
||||||
chrome.storage.session.set({ eventQueue });
|
chrome.storage.session.set({ eventQueue }).catch(() => {});
|
||||||
|
}
|
||||||
|
|
||||||
|
function clearEventQueue() {
|
||||||
|
if (flushTimer) {
|
||||||
|
clearTimeout(flushTimer);
|
||||||
|
flushTimer = null;
|
||||||
|
}
|
||||||
|
eventQueue = [];
|
||||||
|
eventQueueVersion++;
|
||||||
|
chrome.storage.session.set({ eventQueue: [] }).catch(() => {});
|
||||||
|
}
|
||||||
|
|
||||||
|
function applyQueuedRoomPolicy(roomId, policy, reason) {
|
||||||
|
const result = reconcileQueuedRoomIntent(eventQueue, { roomId, ...policy });
|
||||||
|
eventQueue = result.queue;
|
||||||
|
eventQueueVersion++;
|
||||||
|
const { discarded } = result;
|
||||||
|
if (discarded > 0) {
|
||||||
|
chrome.storage.session.set({ eventQueue }).catch(() => {});
|
||||||
|
addLog(`${reason}: discarded ${discarded} queued room action${discarded === 1 ? '' : 's'}`, 'warn');
|
||||||
|
}
|
||||||
|
return result;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Drain the offline event queue in paced batches. A reconnect after a long
|
* Drain logical queue entries only after ROOM_DATA confirms the current room,
|
||||||
* outage can leave up to 50 queued events; dumping them in one tick would
|
* role and lobby. Pacing counts materialized wire frames, not logical entries.
|
||||||
* exceed the server's per-socket event budget and get us disconnected right
|
* A two-frame media intent is kept whole at a batch boundary and retained in
|
||||||
* after rejoining. We send FLUSH_BATCH_SIZE events, then wait
|
* full if either send fails; replaying its first frame again is idempotent and
|
||||||
* FLUSH_BATCH_INTERVAL_MS before the next batch. Remaining events drain across
|
* preserves the final desired state on the next reconnect.
|
||||||
* subsequent batches; if the connection drops mid-drain, the rest stay queued.
|
|
||||||
*/
|
*/
|
||||||
function flushEventQueue() {
|
async function flushEventQueue() {
|
||||||
if (flushTimer) return; // a drain is already in progress
|
if (flushTimer || flushInProgress || awaitingRoomData) return;
|
||||||
const drainBatch = () => {
|
if (!socket || socket.readyState !== WebSocket.OPEN || !isNamespaceJoined) return;
|
||||||
flushTimer = null;
|
flushInProgress = true;
|
||||||
if (!socket || socket.readyState !== WebSocket.OPEN || !isNamespaceJoined) {
|
try {
|
||||||
return; // lost the connection — leave the rest queued for next connect
|
// Resolve privacy settings before taking queue ownership. If new work
|
||||||
}
|
// arrives during the drain, keep the original logical entries too; a
|
||||||
let sent = 0;
|
// later idempotent replay is safer than losing the concurrent intent.
|
||||||
while (eventQueue.length > 0 && sent < FLUSH_BATCH_SIZE) {
|
const replaySettings = eventQueue.some(isQueuedMediaIntent) ? await getSettings() : null;
|
||||||
const queuedMsg = eventQueue.shift();
|
const drainSource = eventQueue;
|
||||||
emit(queuedMsg.event, queuedMsg.data);
|
const drainVersion = eventQueueVersion;
|
||||||
sent++;
|
const result = await drainQueuedBatch(drainSource, {
|
||||||
|
roomId: currentRoom?.roomId || null,
|
||||||
|
maxWireEvents: FLUSH_BATCH_SIZE,
|
||||||
|
sendFrame: async (frame, entry) => {
|
||||||
|
let payload = frame.data && typeof frame.data === 'object' ? { ...frame.data } : {};
|
||||||
|
if (isQueuedMediaIntent(entry)) {
|
||||||
|
payload = withTitlePrivacy(payload, replaySettings, ['mediaTitle']);
|
||||||
|
payload.peerId = peerId;
|
||||||
|
}
|
||||||
|
return emitLive(frame.event, payload);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
if (eventQueueVersion === drainVersion) {
|
||||||
|
eventQueue = result.queue;
|
||||||
|
eventQueueVersion++;
|
||||||
|
} else {
|
||||||
|
// enqueueQueuedEvent preserves object identity for untouched queue
|
||||||
|
// entries. Remove only entries the drain completed, leaving any
|
||||||
|
// concurrently appended or merged work in place. A partial intent
|
||||||
|
// failure returns the complete entry in result.queue, so it is not
|
||||||
|
// included in consumedEntries and remains retryable as a unit.
|
||||||
|
const consumedCount = drainSource.length - result.queue.length;
|
||||||
|
const consumedEntries = new Set(drainSource.slice(0, consumedCount));
|
||||||
|
eventQueue = eventQueue.filter(entry => !consumedEntries.has(entry));
|
||||||
|
eventQueueVersion++;
|
||||||
|
addLog('Queue changed during replay; reconciled completed and concurrent work', 'info');
|
||||||
}
|
}
|
||||||
chrome.storage.session.set({ eventQueue }).catch(() => {});
|
chrome.storage.session.set({ eventQueue }).catch(() => {});
|
||||||
if (eventQueue.length > 0) {
|
if (result.droppedStaleIntents > 0) {
|
||||||
flushTimer = setTimeout(drainBatch, FLUSH_BATCH_INTERVAL_MS);
|
addLog(`Dropped ${result.droppedStaleIntents} stale media intent${result.droppedStaleIntents === 1 ? '' : 's'} for a previous room`, 'warn');
|
||||||
}
|
}
|
||||||
};
|
if (result.sentWireEvents > 0) {
|
||||||
drainBatch();
|
addLog(`Replayed ${result.sentWireEvents} queued wire event${result.sentWireEvents === 1 ? '' : 's'}`, 'info');
|
||||||
|
}
|
||||||
|
if (eventQueue.length > 0 && socket?.readyState === WebSocket.OPEN && isNamespaceJoined) {
|
||||||
|
flushTimer = setTimeout(() => {
|
||||||
|
flushTimer = null;
|
||||||
|
flushEventQueue().catch(error => addLog(`Queue replay failed: ${error.message}`, 'warn'));
|
||||||
|
}, FLUSH_BATCH_INTERVAL_MS);
|
||||||
|
}
|
||||||
|
} finally {
|
||||||
|
flushInProgress = false;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
function addToHistory(action, senderId) {
|
function addToHistory(action, senderId) {
|
||||||
@@ -1678,7 +1784,7 @@ async function tryApplyPendingCanonicalMediaState() {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async function handleCanonicalRoomData(data, hadQueuedMediaControls) {
|
async function handleCanonicalRoomData(data, hasPendingLocalIntent) {
|
||||||
canonicalMediaStateTracker.adoptRoom(data?.roomId || null);
|
canonicalMediaStateTracker.adoptRoom(data?.roomId || null);
|
||||||
const canonicalSnapshot = canonicalMediaStateFromRoomData(data);
|
const canonicalSnapshot = canonicalMediaStateFromRoomData(data);
|
||||||
if (canonicalSnapshot.status === 'unsupported' || canonicalSnapshot.status === 'empty') {
|
if (canonicalSnapshot.status === 'unsupported' || canonicalSnapshot.status === 'empty') {
|
||||||
@@ -1703,9 +1809,9 @@ async function handleCanonicalRoomData(data, hadQueuedMediaControls) {
|
|||||||
persistCanonicalMediaRecovery();
|
persistCanonicalMediaRecovery();
|
||||||
addLog(`Canonical media state received: r${mediaState.revision} ${mediaState.playbackState} @ ${mediaState.currentTime.toFixed(2)}s`, 'info');
|
addLog(`Canonical media state received: r${mediaState.revision} ${mediaState.playbackState} @ ${mediaState.currentTime.toFixed(2)}s`, 'info');
|
||||||
|
|
||||||
if (hadQueuedMediaControls) {
|
if (hasPendingLocalIntent) {
|
||||||
markCanonicalMediaStateHandled(data.roomId, mediaState.revision);
|
markCanonicalMediaStateHandled(data.roomId, mediaState.revision);
|
||||||
addLog(`Canonical media state r${mediaState.revision} skipped: queued local media controls are authoritative for this reconnect`, 'info');
|
addLog(`Canonical media state r${mediaState.revision} skipped: authorized queued local intent takes precedence`, 'info');
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
const result = await tryApplyPendingCanonicalMediaState();
|
const result = await tryApplyPendingCanonicalMediaState();
|
||||||
@@ -1735,8 +1841,6 @@ async function handleServerEvent(event, data) {
|
|||||||
}
|
}
|
||||||
switch (event) {
|
switch (event) {
|
||||||
case EVENTS.ROOM_DATA: {
|
case EVENTS.ROOM_DATA: {
|
||||||
const hadQueuedMediaControls = flushedMediaControlsBeforeRoomData;
|
|
||||||
flushedMediaControlsBeforeRoomData = false;
|
|
||||||
if (currentRoom?.roomId !== data.roomId) {
|
if (currentRoom?.roomId !== data.roomId) {
|
||||||
invalidateChatSession();
|
invalidateChatSession();
|
||||||
clearChatActivity();
|
clearChatActivity();
|
||||||
@@ -1767,7 +1871,10 @@ async function handleServerEvent(event, data) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Recover server-tracked active Episode Lobby if present
|
// Recover server-tracked active Episode Lobby if present
|
||||||
if (data && data.activeLobby && !episodeLobby) {
|
if (!data?.activeLobby && episodeLobby) {
|
||||||
|
clearEpisodeLobbyState();
|
||||||
|
addLog('Discarded stale local Episode Lobby after ROOM_DATA confirmed it ended', 'info');
|
||||||
|
} else if (data && data.activeLobby && !episodeLobby) {
|
||||||
episodeLobby = {
|
episodeLobby = {
|
||||||
expectedTitle: data.activeLobby.expectedTitle,
|
expectedTitle: data.activeLobby.expectedTitle,
|
||||||
initiatorPeerId: data.activeLobby.initiatorPeerId,
|
initiatorPeerId: data.activeLobby.initiatorPeerId,
|
||||||
@@ -1801,7 +1908,22 @@ async function handleServerEvent(event, data) {
|
|||||||
// Inform Website Bridge & Popup
|
// Inform Website Bridge & Popup
|
||||||
const joinStatusMsg = { type: 'JOIN_STATUS', success: true, message: 'Joined' };
|
const joinStatusMsg = { type: 'JOIN_STATUS', success: true, message: 'Joined' };
|
||||||
await broadcastJoinStatus(joinStatusMsg);
|
await broadcastJoinStatus(joinStatusMsg);
|
||||||
await handleCanonicalRoomData(data, hadQueuedMediaControls);
|
awaitingRoomData = false;
|
||||||
|
pendingRoomDataRoomId = null;
|
||||||
|
|
||||||
|
const lostRoomAuthority = controlMode === CONTROL_MODES.HOST_ONLY
|
||||||
|
&& hostPeerId
|
||||||
|
&& !amController();
|
||||||
|
const queuePolicy = applyQueuedRoomPolicy(data.roomId, {
|
||||||
|
canControl: !lostRoomAuthority,
|
||||||
|
activeLobby: !!episodeLobby,
|
||||||
|
desynced: hcmDesynced
|
||||||
|
}, lostRoomAuthority
|
||||||
|
? 'Host Control role changed while offline'
|
||||||
|
: (episodeLobby ? 'Active Episode Lobby takes precedence' : 'Reconnect queue policy'));
|
||||||
|
|
||||||
|
await handleCanonicalRoomData(data, queuePolicy.hasPendingLocalIntent);
|
||||||
|
await flushEventQueue();
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
case EVENTS.CONTROL_MODE:
|
case EVENTS.CONTROL_MODE:
|
||||||
@@ -4013,6 +4135,9 @@ async function handleAsyncMessage(message, sender, sendResponse) {
|
|||||||
episodeLobby: episodeLobby,
|
episodeLobby: episodeLobby,
|
||||||
reconnectAttempts,
|
reconnectAttempts,
|
||||||
reconnectSlowMode: reconnectFailed,
|
reconnectSlowMode: reconnectFailed,
|
||||||
|
queuedLogicalEvents: eventQueue.length,
|
||||||
|
queuedMediaIntents: queuedMediaIntentCount(eventQueue, currentRoom?.roomId || pendingRoomDataRoomId),
|
||||||
|
queuedWireEvents: queuedWireCount(eventQueue),
|
||||||
roomId: currentRoom ? currentRoom.roomId : null,
|
roomId: currentRoom ? currentRoom.roomId : null,
|
||||||
serverUrl: currentServerUrl,
|
serverUrl: currentServerUrl,
|
||||||
version: chrome.runtime.getManifest().version,
|
version: chrome.runtime.getManifest().version,
|
||||||
@@ -4216,6 +4341,7 @@ async function handleAsyncMessage(message, sender, sendResponse) {
|
|||||||
hcmDesynced = !!message.desynced;
|
hcmDesynced = !!message.desynced;
|
||||||
if (storageInitialized) chrome.storage.session.set({ hcmDesynced });
|
if (storageInitialized) chrome.storage.session.set({ hcmDesynced });
|
||||||
if (hcmDesynced) {
|
if (hcmDesynced) {
|
||||||
|
applyQueuedRoomPolicy(currentRoom?.roomId, { desynced: true }, 'Intentional solo mode');
|
||||||
const pending = canonicalMediaStateTracker.getPending(currentRoom?.roomId);
|
const pending = canonicalMediaStateTracker.getPending(currentRoom?.roomId);
|
||||||
if (pending) {
|
if (pending) {
|
||||||
markCanonicalMediaStateHandled(pending.roomId, pending.mediaState.revision);
|
markCanonicalMediaStateHandled(pending.roomId, pending.mediaState.revision);
|
||||||
|
|||||||
@@ -64,8 +64,10 @@ describe('canonical ROOM_DATA recovery contract', () => {
|
|||||||
const roomData = functionBody(backgroundSource, 'handleCanonicalRoomData', 'handleServerEvent');
|
const roomData = functionBody(backgroundSource, 'handleCanonicalRoomData', 'handleServerEvent');
|
||||||
expect(apply).toContain('if (hcmDesynced)');
|
expect(apply).toContain('if (hcmDesynced)');
|
||||||
expect(apply).toContain('if (episodeLobby)');
|
expect(apply).toContain('if (episodeLobby)');
|
||||||
expect(roomData).toContain('if (hadQueuedMediaControls)');
|
expect(roomData).toContain('if (hasPendingLocalIntent)');
|
||||||
expect(backgroundSource).toContain('flushedMediaControlsBeforeRoomData = eventQueue.some');
|
expect(backgroundSource).toContain('awaitingRoomData = true');
|
||||||
|
expect(backgroundSource).toContain('await handleCanonicalRoomData(data, queuePolicy.hasPendingLocalIntent)');
|
||||||
|
expect(backgroundSource).toContain('await flushEventQueue()');
|
||||||
});
|
});
|
||||||
|
|
||||||
it('reuses existing seek abstractions, suppression and drift tolerance', () => {
|
it('reuses existing seek abstractions, suppression and drift tolerance', () => {
|
||||||
|
|||||||
@@ -0,0 +1,73 @@
|
|||||||
|
import fs from 'node:fs';
|
||||||
|
import path from 'node:path';
|
||||||
|
import { fileURLToPath } from 'node:url';
|
||||||
|
import { describe, expect, it } from 'vitest';
|
||||||
|
|
||||||
|
const extensionDir = path.dirname(fileURLToPath(import.meta.url));
|
||||||
|
const backgroundSource = fs.readFileSync(path.join(extensionDir, 'background.js'), 'utf8');
|
||||||
|
|
||||||
|
function functionBody(name, nextName) {
|
||||||
|
const start = backgroundSource.indexOf(`function ${name}(`);
|
||||||
|
const end = backgroundSource.indexOf(`function ${nextName}(`, start + 1);
|
||||||
|
return backgroundSource.slice(start, end === -1 ? backgroundSource.length : end);
|
||||||
|
}
|
||||||
|
|
||||||
|
describe('offline media intent background integration', () => {
|
||||||
|
it('keeps online sends immediate and defers reconnect work only while ROOM_DATA is pending', () => {
|
||||||
|
const emit = functionBody('emit', 'emitLive');
|
||||||
|
expect(emit).toContain('mustWaitForRoomData');
|
||||||
|
expect(emit).toContain('socket.send(msg)');
|
||||||
|
expect(emit).toContain('queueEvent(event, data)');
|
||||||
|
expect(emit).not.toContain('setTimeout');
|
||||||
|
expect(backgroundSource).toMatch(/awaitingRoomData = true;[\s\S]*emit\(EVENTS\.JOIN_ROOM/);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('restores and migrates the persisted MV3 queue with a monotonic local sequence', () => {
|
||||||
|
expect(backgroundSource).toContain('normalizePersistedEventQueue(');
|
||||||
|
expect(backgroundSource).toContain('localSeq = Math.max(localSeq, maxQueuedSequence(eventQueue))');
|
||||||
|
expect(backgroundSource).toContain('chrome.storage.session.set({ eventQueue, localSeq })');
|
||||||
|
expect(backgroundSource).not.toContain('storage.sync.set({ eventQueue');
|
||||||
|
});
|
||||||
|
|
||||||
|
it('reconciles Host Control, Episode Lobby and solo mode before canonical recovery and replay', () => {
|
||||||
|
const roomDataStart = backgroundSource.indexOf('case EVENTS.ROOM_DATA:');
|
||||||
|
const roomDataEnd = backgroundSource.indexOf('case EVENTS.CONTROL_MODE:', roomDataStart);
|
||||||
|
const roomData = backgroundSource.slice(roomDataStart, roomDataEnd);
|
||||||
|
expect(roomData.indexOf('applyQueuedRoomPolicy(data.roomId'))
|
||||||
|
.toBeLessThan(roomData.indexOf('handleCanonicalRoomData(data, queuePolicy.hasPendingLocalIntent)'));
|
||||||
|
expect(roomData.indexOf('handleCanonicalRoomData(data, queuePolicy.hasPendingLocalIntent)'))
|
||||||
|
.toBeLessThan(roomData.indexOf('flushEventQueue()'));
|
||||||
|
expect(roomData).toContain('activeLobby: !!episodeLobby');
|
||||||
|
expect(roomData).toContain('desynced: hcmDesynced');
|
||||||
|
expect(roomData).toContain('if (!data?.activeLobby && episodeLobby)');
|
||||||
|
});
|
||||||
|
|
||||||
|
it('clears queued room intent on failed join, leave and room switch paths', () => {
|
||||||
|
expect(functionBody('clearFailedJoinCredentials', 'invalidateChatSession')).toContain('clearEventQueue()');
|
||||||
|
expect(functionBody('forceDisconnect', 'persistRoomIdleState')).toContain('eventQueue = []');
|
||||||
|
expect(functionBody('leaveOldRoomIfSwitching', 'resetAudioProcessingInTab')).toContain('forceDisconnect()');
|
||||||
|
const leaveHandler = backgroundSource.slice(
|
||||||
|
backgroundSource.indexOf("message.type === 'LEAVE_ROOM'"),
|
||||||
|
backgroundSource.indexOf("message.type === 'CLEAR_LOGS'")
|
||||||
|
);
|
||||||
|
expect(leaveHandler).toContain('forceDisconnect()');
|
||||||
|
});
|
||||||
|
|
||||||
|
it('paces actual frames through a failure-retaining logical drain', () => {
|
||||||
|
const flush = functionBody('flushEventQueue', 'addToHistory');
|
||||||
|
expect(flush).toContain('drainQueuedBatch(drainSource');
|
||||||
|
expect(flush).toContain('maxWireEvents: FLUSH_BATCH_SIZE');
|
||||||
|
expect(flush).toContain('return emitLive(frame.event, payload)');
|
||||||
|
expect(flush).toContain('if (eventQueueVersion === drainVersion)');
|
||||||
|
expect(flush).toContain('const consumedEntries = new Set(drainSource.slice(0, consumedCount))');
|
||||||
|
expect(flush).toContain('eventQueue = eventQueue.filter(entry => !consumedEntries.has(entry))');
|
||||||
|
expect(flush).not.toMatch(/eventQueue\.shift\(\)[\s\S]*emit\(/);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('exposes bounded queue diagnostics without media-title content', () => {
|
||||||
|
expect(backgroundSource).toContain('queuedLogicalEvents: eventQueue.length');
|
||||||
|
expect(backgroundSource).toContain('queuedMediaIntents: queuedMediaIntentCount');
|
||||||
|
expect(backgroundSource).toContain('queuedWireEvents: queuedWireCount(eventQueue)');
|
||||||
|
expect(backgroundSource).not.toMatch(/Offline media intent[^`'\n]*mediaTitle/);
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -0,0 +1,407 @@
|
|||||||
|
import { EVENTS, MAX_MEDIA_TIME } from './shared/constants.js';
|
||||||
|
|
||||||
|
export const MEDIA_INTENT_KIND = 'media-intent';
|
||||||
|
export const MAX_LOGICAL_QUEUE_SIZE = 50;
|
||||||
|
|
||||||
|
const MEDIA_EVENTS = new Set([EVENTS.PLAY, EVENTS.PAUSE, EVENTS.SEEK]);
|
||||||
|
const STALE_OFFLINE_EVENTS = new Set([EVENTS.PING, EVENTS.PONG, EVENTS.PEER_STATUS, EVENTS.EVENT_ACK]);
|
||||||
|
const FORCE_SYNC_EVENTS = new Set([EVENTS.FORCE_SYNC_PREPARE, EVENTS.FORCE_SYNC_EXECUTE]);
|
||||||
|
const HOST_GATED_EVENTS = new Set([
|
||||||
|
...MEDIA_EVENTS,
|
||||||
|
...FORCE_SYNC_EVENTS,
|
||||||
|
EVENTS.EPISODE_LOBBY,
|
||||||
|
EVENTS.EPISODE_LOBBY_CANCEL
|
||||||
|
]);
|
||||||
|
|
||||||
|
function validSequence(value) {
|
||||||
|
return Number.isSafeInteger(value) && value >= 0 ? value : null;
|
||||||
|
}
|
||||||
|
|
||||||
|
function validTimestamp(value) {
|
||||||
|
return typeof value === 'number' && Number.isFinite(value) && value >= 0 ? value : null;
|
||||||
|
}
|
||||||
|
|
||||||
|
function mediaTime(value) {
|
||||||
|
if (typeof value !== 'number' || !Number.isFinite(value)) return null;
|
||||||
|
return Math.max(0, Math.min(MAX_MEDIA_TIME, value));
|
||||||
|
}
|
||||||
|
|
||||||
|
function playbackStateFor(event, data) {
|
||||||
|
if (event === EVENTS.PLAY) return 'playing';
|
||||||
|
if (event === EVENTS.PAUSE) return 'paused';
|
||||||
|
return data?.playbackState === 'playing' || data?.playbackState === 'paused'
|
||||||
|
? data.playbackState
|
||||||
|
: null;
|
||||||
|
}
|
||||||
|
|
||||||
|
function mediaPositionFor(event, data) {
|
||||||
|
if (event === EVENTS.SEEK) {
|
||||||
|
return mediaTime(data?.targetTime) ?? mediaTime(data?.currentTime);
|
||||||
|
}
|
||||||
|
return mediaTime(data?.currentTime);
|
||||||
|
}
|
||||||
|
|
||||||
|
function sanitizedMediaTitle(value) {
|
||||||
|
return typeof value === 'string' && value ? value.substring(0, 100) : null;
|
||||||
|
}
|
||||||
|
|
||||||
|
function trimQueue(queue, maxEntries) {
|
||||||
|
const trimmed = queue.slice();
|
||||||
|
let dropped = 0;
|
||||||
|
while (trimmed.length > maxEntries) {
|
||||||
|
trimmed.shift();
|
||||||
|
dropped++;
|
||||||
|
}
|
||||||
|
return { queue: trimmed, dropped };
|
||||||
|
}
|
||||||
|
|
||||||
|
function createIntentEntry(event, data, roomId) {
|
||||||
|
const currentTime = mediaPositionFor(event, data);
|
||||||
|
const playbackState = playbackStateFor(event, data);
|
||||||
|
if (event === EVENTS.SEEK && currentTime === null) return null;
|
||||||
|
if (playbackState === null && currentTime === null) return null;
|
||||||
|
return {
|
||||||
|
kind: MEDIA_INTENT_KIND,
|
||||||
|
roomId,
|
||||||
|
intent: {
|
||||||
|
playbackState,
|
||||||
|
currentTime,
|
||||||
|
latestEvent: event,
|
||||||
|
previousSeq: null,
|
||||||
|
latestSeq: validSequence(data?.seq),
|
||||||
|
actionTimestamp: validTimestamp(data?.actionTimestamp),
|
||||||
|
mediaTitle: sanitizedMediaTitle(data?.mediaTitle),
|
||||||
|
sourceEventCount: 1
|
||||||
|
}
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
function mergeIntentEntry(entry, event, data) {
|
||||||
|
const incomingSeq = validSequence(data?.seq);
|
||||||
|
const previousLatestSeq = validSequence(entry.intent.latestSeq);
|
||||||
|
if (incomingSeq !== null && previousLatestSeq !== null && incomingSeq <= previousLatestSeq) {
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
const incomingPosition = mediaPositionFor(event, data);
|
||||||
|
if (event === EVENTS.SEEK && incomingPosition === null) return entry;
|
||||||
|
const incomingState = playbackStateFor(event, data);
|
||||||
|
const merged = {
|
||||||
|
...entry,
|
||||||
|
intent: {
|
||||||
|
...entry.intent,
|
||||||
|
playbackState: incomingState ?? entry.intent.playbackState,
|
||||||
|
currentTime: incomingPosition ?? entry.intent.currentTime,
|
||||||
|
latestEvent: event,
|
||||||
|
actionTimestamp: validTimestamp(data?.actionTimestamp) ?? entry.intent.actionTimestamp,
|
||||||
|
mediaTitle: sanitizedMediaTitle(data?.mediaTitle) ?? entry.intent.mediaTitle,
|
||||||
|
sourceEventCount: entry.intent.sourceEventCount + 1
|
||||||
|
}
|
||||||
|
};
|
||||||
|
if (incomingSeq !== null) {
|
||||||
|
merged.intent.previousSeq = previousLatestSeq ?? validSequence(entry.intent.previousSeq);
|
||||||
|
merged.intent.latestSeq = incomingSeq;
|
||||||
|
}
|
||||||
|
return merged;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function isMediaQueueEvent(event) {
|
||||||
|
return MEDIA_EVENTS.has(event);
|
||||||
|
}
|
||||||
|
|
||||||
|
export function isQueuedMediaIntent(entry) {
|
||||||
|
return entry?.kind === MEDIA_INTENT_KIND
|
||||||
|
&& typeof entry.roomId === 'string'
|
||||||
|
&& entry.roomId
|
||||||
|
&& entry.intent
|
||||||
|
&& typeof entry.intent === 'object';
|
||||||
|
}
|
||||||
|
|
||||||
|
export function enqueueQueuedEvent(queue, event, data, {
|
||||||
|
roomId,
|
||||||
|
maxEntries = MAX_LOGICAL_QUEUE_SIZE
|
||||||
|
} = {}) {
|
||||||
|
const next = Array.isArray(queue) ? queue.slice() : [];
|
||||||
|
let collapsed = 0;
|
||||||
|
|
||||||
|
if (STALE_OFFLINE_EVENTS.has(event)) {
|
||||||
|
return { queue: next, collapsed: 0, dropped: 0, droppedStale: 1 };
|
||||||
|
}
|
||||||
|
if (!isMediaQueueEvent(event)) {
|
||||||
|
next.push({ event, data });
|
||||||
|
} else if (typeof roomId === 'string' && roomId) {
|
||||||
|
const last = next.at(-1);
|
||||||
|
const hasMergeTarget = isQueuedMediaIntent(last) && last.roomId === roomId;
|
||||||
|
const merged = hasMergeTarget
|
||||||
|
? mergeIntentEntry(last, event, data)
|
||||||
|
: null;
|
||||||
|
if (merged) {
|
||||||
|
next[next.length - 1] = merged;
|
||||||
|
collapsed = 1;
|
||||||
|
} else if (!hasMergeTarget) {
|
||||||
|
const entry = createIntentEntry(event, data, roomId);
|
||||||
|
if (entry) next.push(entry);
|
||||||
|
} else {
|
||||||
|
return { queue: next, collapsed: 0, dropped: 0, droppedStale: 1 };
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const trimmed = trimQueue(next, maxEntries);
|
||||||
|
return { queue: trimmed.queue, collapsed, dropped: trimmed.dropped, droppedStale: 0 };
|
||||||
|
}
|
||||||
|
|
||||||
|
function normalizeIntentEntry(entry, roomId) {
|
||||||
|
if (!isQueuedMediaIntent(entry) || entry.roomId !== roomId) return null;
|
||||||
|
const intent = entry.intent;
|
||||||
|
const playbackState = intent.playbackState === 'playing' || intent.playbackState === 'paused'
|
||||||
|
? intent.playbackState
|
||||||
|
: null;
|
||||||
|
const currentTime = mediaTime(intent.currentTime);
|
||||||
|
const latestEvent = isMediaQueueEvent(intent.latestEvent) ? intent.latestEvent : null;
|
||||||
|
if (!latestEvent || (playbackState === null && currentTime === null)) return null;
|
||||||
|
return {
|
||||||
|
kind: MEDIA_INTENT_KIND,
|
||||||
|
roomId,
|
||||||
|
intent: {
|
||||||
|
playbackState,
|
||||||
|
currentTime,
|
||||||
|
latestEvent,
|
||||||
|
previousSeq: validSequence(intent.previousSeq),
|
||||||
|
latestSeq: validSequence(intent.latestSeq),
|
||||||
|
actionTimestamp: validTimestamp(intent.actionTimestamp),
|
||||||
|
mediaTitle: sanitizedMediaTitle(intent.mediaTitle),
|
||||||
|
sourceEventCount: Number.isSafeInteger(intent.sourceEventCount) && intent.sourceEventCount > 0
|
||||||
|
? intent.sourceEventCount
|
||||||
|
: 1
|
||||||
|
}
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
export function normalizePersistedEventQueue(value, roomId, maxEntries = MAX_LOGICAL_QUEUE_SIZE) {
|
||||||
|
if (!Array.isArray(value) || typeof roomId !== 'string' || !roomId) return [];
|
||||||
|
let normalized = [];
|
||||||
|
for (const entry of value) {
|
||||||
|
if (isQueuedMediaIntent(entry)) {
|
||||||
|
const intentEntry = normalizeIntentEntry(entry, roomId);
|
||||||
|
if (intentEntry) normalized.push(intentEntry);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if (!entry || typeof entry !== 'object' || typeof entry.event !== 'string') continue;
|
||||||
|
if (STALE_OFFLINE_EVENTS.has(entry.event)) continue;
|
||||||
|
if (isMediaQueueEvent(entry.event)) {
|
||||||
|
normalized = enqueueQueuedEvent(normalized, entry.event, entry.data, { roomId, maxEntries }).queue;
|
||||||
|
} else {
|
||||||
|
normalized.push({ event: entry.event, data: entry.data });
|
||||||
|
}
|
||||||
|
normalized = trimQueue(normalized, maxEntries).queue;
|
||||||
|
}
|
||||||
|
return normalized;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function mediaIntentNeedsSequenceReservation(entry) {
|
||||||
|
if (!isQueuedMediaIntent(entry)) return false;
|
||||||
|
const { playbackState, currentTime, previousSeq, latestSeq } = entry.intent;
|
||||||
|
return playbackState !== null
|
||||||
|
&& currentTime !== null
|
||||||
|
&& validSequence(latestSeq) !== null
|
||||||
|
&& validSequence(previousSeq) === null;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function reserveLatestMediaIntentSequence(queue, roomId, nextSequence) {
|
||||||
|
const next = Array.isArray(queue) ? queue.slice() : [];
|
||||||
|
const index = next.length - 1;
|
||||||
|
const entry = next[index];
|
||||||
|
if (!isQueuedMediaIntent(entry)
|
||||||
|
|| entry.roomId !== roomId
|
||||||
|
|| !mediaIntentNeedsSequenceReservation(entry)
|
||||||
|
|| validSequence(nextSequence) === null
|
||||||
|
|| nextSequence <= entry.intent.latestSeq) {
|
||||||
|
return { queue: next, reserved: false };
|
||||||
|
}
|
||||||
|
next[index] = {
|
||||||
|
...entry,
|
||||||
|
intent: {
|
||||||
|
...entry.intent,
|
||||||
|
previousSeq: entry.intent.latestSeq,
|
||||||
|
latestSeq: nextSequence
|
||||||
|
}
|
||||||
|
};
|
||||||
|
return { queue: next, reserved: true };
|
||||||
|
}
|
||||||
|
|
||||||
|
function frameData(intent, seq) {
|
||||||
|
const data = {};
|
||||||
|
if (seq !== null) data.seq = seq;
|
||||||
|
if (intent.actionTimestamp !== null) data.actionTimestamp = intent.actionTimestamp;
|
||||||
|
if (intent.mediaTitle !== null) data.mediaTitle = intent.mediaTitle;
|
||||||
|
return data;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function materializeMediaIntent(entry) {
|
||||||
|
if (!isQueuedMediaIntent(entry)) return [];
|
||||||
|
const intent = entry.intent;
|
||||||
|
const currentTime = mediaTime(intent.currentTime);
|
||||||
|
const playbackState = intent.playbackState === 'playing' || intent.playbackState === 'paused'
|
||||||
|
? intent.playbackState
|
||||||
|
: null;
|
||||||
|
const latestSeq = validSequence(intent.latestSeq);
|
||||||
|
const previousSeq = validSequence(intent.previousSeq);
|
||||||
|
const stateEvent = playbackState === 'playing' ? EVENTS.PLAY : EVENTS.PAUSE;
|
||||||
|
|
||||||
|
if (playbackState === null && currentTime !== null) {
|
||||||
|
return [{
|
||||||
|
event: EVENTS.SEEK,
|
||||||
|
data: { ...frameData(intent, latestSeq), currentTime, targetTime: currentTime }
|
||||||
|
}];
|
||||||
|
}
|
||||||
|
if (playbackState !== null && currentTime === null) {
|
||||||
|
return [{ event: stateEvent, data: frameData(intent, latestSeq) }];
|
||||||
|
}
|
||||||
|
if (playbackState === null || currentTime === null) return [];
|
||||||
|
|
||||||
|
// Both materialized frames carry the latest genuine action timestamp: it is
|
||||||
|
// correlation metadata for existing ACK/activity paths, not scheduled wall
|
||||||
|
// time. A previous-format single PLAY/PAUSE has only one reserved sequence. Keep
|
||||||
|
// its original one-frame behavior during migration rather than inventing a
|
||||||
|
// sequence that could overtake a later transactional barrier.
|
||||||
|
if (previousSeq === null || latestSeq === null || previousSeq >= latestSeq) {
|
||||||
|
if (intent.latestEvent === EVENTS.SEEK) {
|
||||||
|
return [{
|
||||||
|
event: EVENTS.SEEK,
|
||||||
|
data: { ...frameData(intent, latestSeq), currentTime, targetTime: currentTime }
|
||||||
|
}];
|
||||||
|
}
|
||||||
|
return [{
|
||||||
|
event: stateEvent,
|
||||||
|
data: { ...frameData(intent, latestSeq), currentTime }
|
||||||
|
}];
|
||||||
|
}
|
||||||
|
|
||||||
|
const seekFrame = {
|
||||||
|
event: EVENTS.SEEK,
|
||||||
|
data: { ...frameData(intent, intent.latestEvent === EVENTS.SEEK ? latestSeq : previousSeq), currentTime, targetTime: currentTime }
|
||||||
|
};
|
||||||
|
const stateFrame = {
|
||||||
|
event: stateEvent,
|
||||||
|
data: { ...frameData(intent, intent.latestEvent === EVENTS.SEEK ? previousSeq : latestSeq), currentTime }
|
||||||
|
};
|
||||||
|
return intent.latestEvent === EVENTS.SEEK
|
||||||
|
? [stateFrame, seekFrame]
|
||||||
|
: [seekFrame, stateFrame];
|
||||||
|
}
|
||||||
|
|
||||||
|
export function queuedEntryWireCount(entry) {
|
||||||
|
return isQueuedMediaIntent(entry) ? materializeMediaIntent(entry).length : 1;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function queuedWireCount(queue) {
|
||||||
|
return Array.isArray(queue)
|
||||||
|
? queue.reduce((total, entry) => total + queuedEntryWireCount(entry), 0)
|
||||||
|
: 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function queuedMediaIntentCount(queue, roomId = null) {
|
||||||
|
return Array.isArray(queue)
|
||||||
|
? queue.filter(entry => isQueuedMediaIntent(entry) && (!roomId || entry.roomId === roomId)).length
|
||||||
|
: 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function hasQueuedMediaIntent(queue, roomId) {
|
||||||
|
return queuedMediaIntentCount(queue, roomId) > 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function discardQueuedMediaIntents(queue, roomId) {
|
||||||
|
return Array.isArray(queue)
|
||||||
|
? queue.filter(entry => !isQueuedMediaIntent(entry) || entry.roomId !== roomId)
|
||||||
|
: [];
|
||||||
|
}
|
||||||
|
|
||||||
|
export function reconcileQueuedRoomIntent(queue, {
|
||||||
|
roomId,
|
||||||
|
canControl = true,
|
||||||
|
activeLobby = false,
|
||||||
|
desynced = false
|
||||||
|
} = {}) {
|
||||||
|
const source = Array.isArray(queue) ? queue : [];
|
||||||
|
const blockedEvents = !canControl
|
||||||
|
? HOST_GATED_EVENTS
|
||||||
|
: (activeLobby || desynced ? FORCE_SYNC_EVENTS : null);
|
||||||
|
const reconciled = source.filter(entry => {
|
||||||
|
if (isQueuedMediaIntent(entry) && entry.roomId === roomId) {
|
||||||
|
return canControl && !activeLobby && !desynced;
|
||||||
|
}
|
||||||
|
return !blockedEvents?.has(entry?.event);
|
||||||
|
});
|
||||||
|
const hasPendingLocalIntent = reconciled.some(entry =>
|
||||||
|
(isQueuedMediaIntent(entry) && entry.roomId === roomId)
|
||||||
|
|| (!isQueuedMediaIntent(entry) && FORCE_SYNC_EVENTS.has(entry?.event))
|
||||||
|
);
|
||||||
|
return {
|
||||||
|
queue: reconciled,
|
||||||
|
discarded: source.length - reconciled.length,
|
||||||
|
hasPendingLocalIntent
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
export function maxQueuedSequence(queue) {
|
||||||
|
let max = 0;
|
||||||
|
for (const entry of Array.isArray(queue) ? queue : []) {
|
||||||
|
if (isQueuedMediaIntent(entry)) {
|
||||||
|
max = Math.max(max, validSequence(entry.intent.previousSeq) ?? 0, validSequence(entry.intent.latestSeq) ?? 0);
|
||||||
|
} else {
|
||||||
|
max = Math.max(max, validSequence(entry?.data?.seq) ?? 0);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return max;
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function drainQueuedBatch(queue, {
|
||||||
|
roomId,
|
||||||
|
maxWireEvents,
|
||||||
|
sendFrame
|
||||||
|
}) {
|
||||||
|
const remaining = Array.isArray(queue) ? queue.slice() : [];
|
||||||
|
let sentWireEvents = 0;
|
||||||
|
let droppedStaleIntents = 0;
|
||||||
|
|
||||||
|
while (remaining.length > 0) {
|
||||||
|
const entry = remaining[0];
|
||||||
|
if (isQueuedMediaIntent(entry) && entry.roomId !== roomId) {
|
||||||
|
remaining.shift();
|
||||||
|
droppedStaleIntents++;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
const frames = isQueuedMediaIntent(entry)
|
||||||
|
? materializeMediaIntent(entry)
|
||||||
|
: [{ event: entry.event, data: entry.data }];
|
||||||
|
if (frames.length === 0) {
|
||||||
|
remaining.shift();
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if (sentWireEvents > 0 && sentWireEvents + frames.length > maxWireEvents) {
|
||||||
|
return { queue: remaining, sentWireEvents, droppedStaleIntents, status: 'batch_full' };
|
||||||
|
}
|
||||||
|
if (frames.length > maxWireEvents) {
|
||||||
|
return { queue: remaining, sentWireEvents, droppedStaleIntents, status: 'entry_exceeds_batch' };
|
||||||
|
}
|
||||||
|
|
||||||
|
let sentEntryFrames = 0;
|
||||||
|
for (const frame of frames) {
|
||||||
|
if (!await sendFrame(frame, entry)) {
|
||||||
|
return {
|
||||||
|
queue: remaining,
|
||||||
|
sentWireEvents: sentWireEvents + sentEntryFrames,
|
||||||
|
droppedStaleIntents,
|
||||||
|
status: 'send_failed'
|
||||||
|
};
|
||||||
|
}
|
||||||
|
sentEntryFrames++;
|
||||||
|
}
|
||||||
|
sentWireEvents += sentEntryFrames;
|
||||||
|
remaining.shift();
|
||||||
|
if (sentWireEvents >= maxWireEvents) {
|
||||||
|
return { queue: remaining, sentWireEvents, droppedStaleIntents, status: 'batch_full' };
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return { queue: remaining, sentWireEvents, droppedStaleIntents, status: 'drained' };
|
||||||
|
}
|
||||||
@@ -0,0 +1,265 @@
|
|||||||
|
import { describe, expect, it } from 'vitest';
|
||||||
|
import { EVENTS, MAX_MEDIA_TIME } from './shared/constants.js';
|
||||||
|
import { canonicalMediaStateFromRoomData } from './canonical-media-state.js';
|
||||||
|
import {
|
||||||
|
discardQueuedMediaIntents,
|
||||||
|
drainQueuedBatch,
|
||||||
|
enqueueQueuedEvent,
|
||||||
|
hasQueuedMediaIntent,
|
||||||
|
materializeMediaIntent,
|
||||||
|
maxQueuedSequence,
|
||||||
|
normalizePersistedEventQueue,
|
||||||
|
queuedEntryWireCount,
|
||||||
|
queuedMediaIntentCount,
|
||||||
|
queuedWireCount,
|
||||||
|
reconcileQueuedRoomIntent,
|
||||||
|
reserveLatestMediaIntentSequence
|
||||||
|
} from './offline-media-intent.js';
|
||||||
|
|
||||||
|
const roomId = 'room-a';
|
||||||
|
const media = (event, data, queue = []) => enqueueQueuedEvent(queue, event, data, { roomId }).queue;
|
||||||
|
|
||||||
|
function reserve(queue, sequence) {
|
||||||
|
return reserveLatestMediaIntentSequence(queue, roomId, sequence).queue;
|
||||||
|
}
|
||||||
|
|
||||||
|
describe('offline media intent coalescing', () => {
|
||||||
|
it('normalizes an empty or roomless persisted queue to empty', () => {
|
||||||
|
expect(normalizePersistedEventQueue([], roomId)).toEqual([]);
|
||||||
|
expect(normalizePersistedEventQueue([{ event: EVENTS.PLAY, data: { seq: 1 } }], null)).toEqual([]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('creates a PLAY intent and reserves ordered legacy SEEK + PLAY frames', () => {
|
||||||
|
let queue = media(EVENTS.PLAY, { currentTime: 10, seq: 5, actionTimestamp: 100 });
|
||||||
|
queue = reserve(queue, 6);
|
||||||
|
expect(queue).toHaveLength(1);
|
||||||
|
expect(materializeMediaIntent(queue[0])).toEqual([
|
||||||
|
{ event: EVENTS.SEEK, data: { seq: 5, actionTimestamp: 100, currentTime: 10, targetTime: 10 } },
|
||||||
|
{ event: EVENTS.PLAY, data: { seq: 6, actionTimestamp: 100, currentTime: 10 } }
|
||||||
|
]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('collapses repeated PLAY and SEEK while preserving final state and monotonic sequences', () => {
|
||||||
|
let queue = reserve(media(EVENTS.PLAY, { currentTime: 10, seq: 1 }), 2);
|
||||||
|
queue = media(EVENTS.PLAY, { currentTime: 20, seq: 3 }, queue);
|
||||||
|
queue = media(EVENTS.SEEK, { targetTime: 30, seq: 4 }, queue);
|
||||||
|
expect(queue).toHaveLength(1);
|
||||||
|
expect(queue[0].intent).toMatchObject({
|
||||||
|
playbackState: 'playing',
|
||||||
|
currentTime: 30,
|
||||||
|
latestEvent: EVENTS.SEEK,
|
||||||
|
previousSeq: 3,
|
||||||
|
latestSeq: 4,
|
||||||
|
sourceEventCount: 3
|
||||||
|
});
|
||||||
|
expect(materializeMediaIntent(queue[0]).map(frame => frame.data.seq)).toEqual([3, 4]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('drops a regressing queued sequence instead of emitting stale ordering', () => {
|
||||||
|
let queue = reserve(media(EVENTS.PLAY, { currentTime: 10, seq: 10 }), 11);
|
||||||
|
const result = enqueueQueuedEvent(queue, EVENTS.PAUSE, { currentTime: 20, seq: 9 }, { roomId });
|
||||||
|
expect(result.droppedStale).toBe(1);
|
||||||
|
expect(result.queue).toEqual(queue);
|
||||||
|
expect(maxQueuedSequence(result.queue)).toBe(11);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('merges PLAY -> SEEK -> PAUSE into paused at the final position', () => {
|
||||||
|
let queue = reserve(media(EVENTS.PLAY, { currentTime: 500, seq: 1 }), 2);
|
||||||
|
queue = media(EVENTS.SEEK, { targetTime: 600, seq: 3 }, queue);
|
||||||
|
queue = media(EVENTS.PAUSE, { currentTime: 605, seq: 4 }, queue);
|
||||||
|
expect(materializeMediaIntent(queue[0])).toEqual([
|
||||||
|
{ event: EVENTS.SEEK, data: { seq: 3, currentTime: 605, targetTime: 605 } },
|
||||||
|
{ event: EVENTS.PAUSE, data: { seq: 4, currentTime: 605 } }
|
||||||
|
]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('merges PAUSE -> SEEK -> PLAY into playing at the final position', () => {
|
||||||
|
let queue = reserve(media(EVENTS.PAUSE, { currentTime: 100, seq: 10 }), 11);
|
||||||
|
queue = media(EVENTS.SEEK, { targetTime: 200, seq: 12 }, queue);
|
||||||
|
queue = media(EVENTS.PLAY, { currentTime: 205, seq: 13 }, queue);
|
||||||
|
expect(materializeMediaIntent(queue[0]).map(frame => frame.event)).toEqual([EVENTS.SEEK, EVENTS.PLAY]);
|
||||||
|
expect(queue[0].intent).toMatchObject({ playbackState: 'playing', currentTime: 205 });
|
||||||
|
});
|
||||||
|
|
||||||
|
it('clamps finite positions and rejects a SEEK with no trustworthy position', () => {
|
||||||
|
expect(media(EVENTS.SEEK, { targetTime: NaN, seq: 1 })).toEqual([]);
|
||||||
|
expect(media(EVENTS.SEEK, { targetTime: -50, seq: 1 })[0].intent.currentTime).toBe(0);
|
||||||
|
expect(media(EVENTS.SEEK, { targetTime: MAX_MEDIA_TIME + 50, seq: 1 })[0].intent.currentTime).toBe(MAX_MEDIA_TIME);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('keeps title metadata bounded and never retains arbitrary payload fields', () => {
|
||||||
|
const queue = media(EVENTS.PAUSE, {
|
||||||
|
currentTime: 10,
|
||||||
|
seq: 1,
|
||||||
|
mediaTitle: 'x'.repeat(200),
|
||||||
|
password: 'secret',
|
||||||
|
chatKey: 'secret'
|
||||||
|
});
|
||||||
|
expect(queue[0].intent.mediaTitle).toHaveLength(100);
|
||||||
|
expect(JSON.stringify(queue)).not.toContain('password');
|
||||||
|
expect(JSON.stringify(queue)).not.toContain('chatKey');
|
||||||
|
});
|
||||||
|
|
||||||
|
it('keeps a thousand-event media burst at one logical entry', () => {
|
||||||
|
let queue = reserve(media(EVENTS.PLAY, { currentTime: 0, seq: 1 }), 2);
|
||||||
|
for (let index = 1; index <= 1000; index++) {
|
||||||
|
queue = media(EVENTS.SEEK, { targetTime: index, seq: index + 2 }, queue);
|
||||||
|
}
|
||||||
|
queue = media(EVENTS.PAUSE, { currentTime: 1000, seq: 1003 }, queue);
|
||||||
|
expect(queue).toHaveLength(1);
|
||||||
|
expect(queue[0].intent).toMatchObject({ playbackState: 'paused', currentTime: 1000, sourceEventCount: 1002 });
|
||||||
|
});
|
||||||
|
|
||||||
|
it('treats retained coordination events as barriers', () => {
|
||||||
|
let queue = reserve(media(EVENTS.PLAY, { currentTime: 10, seq: 1 }), 2);
|
||||||
|
queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_PREPARE, { targetTime: 50, seq: 3 }, { roomId }).queue;
|
||||||
|
queue = media(EVENTS.SEEK, { targetTime: 100, seq: 4 }, queue);
|
||||||
|
queue = media(EVENTS.PAUSE, { currentTime: 120, seq: 5 }, queue);
|
||||||
|
expect(queue).toHaveLength(3);
|
||||||
|
expect(queue[0].kind).toBe('media-intent');
|
||||||
|
expect(queue[1].event).toBe(EVENTS.FORCE_SYNC_PREPARE);
|
||||||
|
expect(queue[2].kind).toBe('media-intent');
|
||||||
|
expect(queue[2].intent).toMatchObject({ playbackState: 'paused', currentTime: 120 });
|
||||||
|
});
|
||||||
|
|
||||||
|
it('does not persist stale liveness and command ACK frames as ordering barriers', () => {
|
||||||
|
let queue = reserve(media(EVENTS.PLAY, { currentTime: 10, seq: 1 }), 2);
|
||||||
|
for (const event of [EVENTS.PING, EVENTS.PONG, EVENTS.PEER_STATUS, EVENTS.EVENT_ACK]) {
|
||||||
|
const result = enqueueQueuedEvent(queue, event, { seq: 99 }, { roomId });
|
||||||
|
expect(result.droppedStale).toBe(1);
|
||||||
|
queue = result.queue;
|
||||||
|
}
|
||||||
|
queue = media(EVENTS.PAUSE, { currentTime: 20, seq: 3 }, queue);
|
||||||
|
expect(queue).toHaveLength(1);
|
||||||
|
expect(queue[0].intent).toMatchObject({ playbackState: 'paused', currentTime: 20 });
|
||||||
|
});
|
||||||
|
|
||||||
|
it('preserves the bounded logical queue cap', () => {
|
||||||
|
let queue = [];
|
||||||
|
for (let index = 0; index < 60; index++) {
|
||||||
|
queue = enqueueQueuedEvent(queue, EVENTS.EPISODE_READY, { index }, { roomId }).queue;
|
||||||
|
}
|
||||||
|
expect(queue).toHaveLength(50);
|
||||||
|
expect(queue[0].data.index).toBe(10);
|
||||||
|
expect(queue.at(-1).data.index).toBe(59);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('migrates old raw media entries without crossing transactional barriers', () => {
|
||||||
|
const restored = normalizePersistedEventQueue([
|
||||||
|
{ event: EVENTS.PLAY, data: { currentTime: 10, seq: 1 } },
|
||||||
|
{ event: EVENTS.SEEK, data: { targetTime: 20, seq: 2 } },
|
||||||
|
{ event: EVENTS.PAUSE, data: { currentTime: 25, seq: 3 } },
|
||||||
|
{ event: EVENTS.EPISODE_LOBBY, data: { expectedTitle: 'S01E02' } },
|
||||||
|
{ event: EVENTS.PLAY, data: { currentTime: 30, seq: 4 } }
|
||||||
|
], roomId);
|
||||||
|
expect(restored).toHaveLength(3);
|
||||||
|
expect(restored[0].intent).toMatchObject({ playbackState: 'paused', currentTime: 25 });
|
||||||
|
expect(restored[1].event).toBe(EVENTS.EPISODE_LOBBY);
|
||||||
|
expect(restored[2].intent).toMatchObject({ playbackState: 'playing', currentTime: 30 });
|
||||||
|
expect(maxQueuedSequence(restored)).toBe(4);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('discards stale-room intent without affecting the new room', () => {
|
||||||
|
const queue = reserve(media(EVENTS.PAUSE, { currentTime: 500, seq: 1 }), 2);
|
||||||
|
expect(hasQueuedMediaIntent(queue, roomId)).toBe(true);
|
||||||
|
expect(discardQueuedMediaIntents(queue, roomId)).toEqual([]);
|
||||||
|
expect(hasQueuedMediaIntent(queue, 'room-b')).toBe(false);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('drops queued shared intent after controller role loss so canonical recovery can proceed', () => {
|
||||||
|
let queue = reserve(media(EVENTS.PAUSE, { currentTime: 1200, seq: 1 }), 2);
|
||||||
|
queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_PREPARE, { targetTime: 1300, seq: 3 }, { roomId }).queue;
|
||||||
|
const result = reconcileQueuedRoomIntent(queue, { roomId, canControl: false });
|
||||||
|
expect(result.queue).toEqual([]);
|
||||||
|
expect(result.discarded).toBe(2);
|
||||||
|
expect(result.hasPendingLocalIntent).toBe(false);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('keeps an active Episode Lobby authoritative over queued media and Force Sync', () => {
|
||||||
|
let queue = reserve(media(EVENTS.PLAY, { currentTime: 100, seq: 1 }), 2);
|
||||||
|
queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_PREPARE, { targetTime: 500, seq: 3 }, { roomId }).queue;
|
||||||
|
queue = enqueueQueuedEvent(queue, EVENTS.EPISODE_READY, { title: 'S01E02' }, { roomId }).queue;
|
||||||
|
const result = reconcileQueuedRoomIntent(queue, { roomId, activeLobby: true });
|
||||||
|
expect(result.queue).toEqual([{ event: EVENTS.EPISODE_READY, data: { title: 'S01E02' } }]);
|
||||||
|
expect(result.hasPendingLocalIntent).toBe(false);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('does not let intentional solo mode retain future room-driving intent', () => {
|
||||||
|
const queue = reserve(media(EVENTS.SEEK, { targetTime: 600, playbackState: 'paused', seq: 1 }), 2);
|
||||||
|
const result = reconcileQueuedRoomIntent(queue, { roomId, desynced: true });
|
||||||
|
expect(result.queue).toEqual([]);
|
||||||
|
expect(result.hasPendingLocalIntent).toBe(false);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('materializes legacy events normally when an old relay has no media-state capability', () => {
|
||||||
|
expect(canonicalMediaStateFromRoomData({
|
||||||
|
roomId,
|
||||||
|
capabilities: ['host-control', 'chat-v1']
|
||||||
|
})).toEqual({ status: 'unsupported', mediaState: null });
|
||||||
|
const queue = reserve(media(EVENTS.PAUSE, { currentTime: 75, seq: 20 }), 21);
|
||||||
|
expect(materializeMediaIntent(queue[0]).map(frame => frame.event)).toEqual([EVENTS.SEEK, EVENTS.PAUSE]);
|
||||||
|
expect(materializeMediaIntent(queue[0]).every(frame =>
|
||||||
|
frame.data.mediaState === undefined && frame.data.revision === undefined
|
||||||
|
)).toBe(true);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
describe('offline media intent drain', () => {
|
||||||
|
it('counts actual wire frames and never splits an intent at a batch boundary', async () => {
|
||||||
|
let queue = enqueueQueuedEvent([], EVENTS.EPISODE_READY, { seq: 1 }, { roomId }).queue;
|
||||||
|
queue = reserve(media(EVENTS.PAUSE, { currentTime: 50, seq: 2 }, queue), 3);
|
||||||
|
const sent = [];
|
||||||
|
const first = await drainQueuedBatch(queue, {
|
||||||
|
roomId,
|
||||||
|
maxWireEvents: 2,
|
||||||
|
sendFrame: async frame => { sent.push(frame); return true; }
|
||||||
|
});
|
||||||
|
expect(first.sentWireEvents).toBe(1);
|
||||||
|
expect(first.queue).toHaveLength(1);
|
||||||
|
expect(sent.map(frame => frame.event)).toEqual([EVENTS.EPISODE_READY]);
|
||||||
|
|
||||||
|
const second = await drainQueuedBatch(first.queue, {
|
||||||
|
roomId,
|
||||||
|
maxWireEvents: 2,
|
||||||
|
sendFrame: async frame => { sent.push(frame); return true; }
|
||||||
|
});
|
||||||
|
expect(second.sentWireEvents).toBe(2);
|
||||||
|
expect(second.queue).toEqual([]);
|
||||||
|
expect(sent.slice(1).map(frame => frame.event)).toEqual([EVENTS.SEEK, EVENTS.PAUSE]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('retains the whole logical intent after a partial send failure', async () => {
|
||||||
|
const queue = reserve(media(EVENTS.PLAY, { currentTime: 90, seq: 5 }), 6);
|
||||||
|
let calls = 0;
|
||||||
|
const result = await drainQueuedBatch(queue, {
|
||||||
|
roomId,
|
||||||
|
maxWireEvents: 10,
|
||||||
|
sendFrame: async () => ++calls === 1
|
||||||
|
});
|
||||||
|
expect(result.status).toBe('send_failed');
|
||||||
|
expect(result.sentWireEvents).toBe(1);
|
||||||
|
expect(result.queue).toEqual(queue);
|
||||||
|
expect(materializeMediaIntent(result.queue[0]).map(frame => frame.data.seq)).toEqual([5, 6]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('drops stale-room intent during drain and preserves unrelated events', async () => {
|
||||||
|
let queue = reserve(media(EVENTS.PAUSE, { currentTime: 40, seq: 1 }), 2);
|
||||||
|
queue = enqueueQueuedEvent(queue, EVENTS.EPISODE_READY, { seq: 3 }, { roomId }).queue;
|
||||||
|
const sent = [];
|
||||||
|
const result = await drainQueuedBatch(queue, {
|
||||||
|
roomId: 'room-b',
|
||||||
|
maxWireEvents: 10,
|
||||||
|
sendFrame: async frame => { sent.push(frame); return true; }
|
||||||
|
});
|
||||||
|
expect(result.droppedStaleIntents).toBe(1);
|
||||||
|
expect(sent).toEqual([{ event: EVENTS.EPISODE_READY, data: { seq: 3 } }]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('reports logical and actual-wire queue sizes separately', () => {
|
||||||
|
let queue = reserve(media(EVENTS.PLAY, { currentTime: 10, seq: 1 }), 2);
|
||||||
|
queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_PREPARE, { seq: 3 }, { roomId }).queue;
|
||||||
|
expect(queuedMediaIntentCount(queue, roomId)).toBe(1);
|
||||||
|
expect(queuedEntryWireCount(queue[0])).toBe(2);
|
||||||
|
expect(queuedWireCount(queue)).toBe(3);
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -5,6 +5,11 @@ import { createRequire } from 'node:module';
|
|||||||
import { fileURLToPath } from 'node:url';
|
import { fileURLToPath } from 'node:url';
|
||||||
import path from 'node:path';
|
import path from 'node:path';
|
||||||
import { connectionCounts, clearRateLimitMaps } from '../server/rate-limiter.js';
|
import { connectionCounts, clearRateLimitMaps } from '../server/rate-limiter.js';
|
||||||
|
import {
|
||||||
|
enqueueQueuedEvent,
|
||||||
|
materializeMediaIntent,
|
||||||
|
reserveLatestMediaIntentSequence
|
||||||
|
} from '../extension/offline-media-intent.js';
|
||||||
|
|
||||||
const __dirname = path.dirname(fileURLToPath(import.meta.url));
|
const __dirname = path.dirname(fileURLToPath(import.meta.url));
|
||||||
const require = createRequire(path.join(__dirname, '..', 'server', 'package.json'));
|
const require = createRequire(path.join(__dirname, '..', 'server', 'package.json'));
|
||||||
@@ -70,6 +75,71 @@ try {
|
|||||||
close();
|
close();
|
||||||
resetConnectionRate();
|
resetConnectionRate();
|
||||||
|
|
||||||
|
// --- Combined reconnect model: compacted new-client intent over legacy wire ---
|
||||||
|
// The queue is entirely client-side. Feed its materialized PLAY/PAUSE/SEEK
|
||||||
|
// frames through the real relay and an old-client-like receiver to prove the
|
||||||
|
// optimization needs no new event, capability or ACK.
|
||||||
|
const coalescedRid = 'coalesced-media-'+Date.now();
|
||||||
|
const legacyReceiver = await c(), coalescingSender = await c();
|
||||||
|
await j(legacyReceiver, coalescedRid, 'legacy-receiver');
|
||||||
|
await j(coalescingSender, coalescedRid, 'coalesce-sender', null, ['chat-v1']);
|
||||||
|
legacyReceiver._m.length = coalescingSender._m.length = 0;
|
||||||
|
|
||||||
|
s(legacyReceiver, 'play', { currentTime: 100, seq: 1, actionTimestamp: 1 });
|
||||||
|
await w(coalescingSender, 'play');
|
||||||
|
const canonicalBeforeReplay = { ...mod.rooms.get(coalescedRid).mediaState };
|
||||||
|
|
||||||
|
let compactedQueue = enqueueQueuedEvent([], 'play', {
|
||||||
|
currentTime: 500,
|
||||||
|
seq: 10,
|
||||||
|
actionTimestamp: 10
|
||||||
|
}, { roomId: coalescedRid }).queue;
|
||||||
|
compactedQueue = reserveLatestMediaIntentSequence(compactedQueue, coalescedRid, 11).queue;
|
||||||
|
for (const [targetTime, seq] of [[540, 12], [600, 13]]) {
|
||||||
|
compactedQueue = enqueueQueuedEvent(compactedQueue, 'seek', {
|
||||||
|
currentTime: targetTime,
|
||||||
|
targetTime,
|
||||||
|
seq,
|
||||||
|
actionTimestamp: seq
|
||||||
|
}, { roomId: coalescedRid }).queue;
|
||||||
|
}
|
||||||
|
compactedQueue = enqueueQueuedEvent(compactedQueue, 'pause', {
|
||||||
|
currentTime: 605,
|
||||||
|
seq: 14,
|
||||||
|
actionTimestamp: 14
|
||||||
|
}, { roomId: coalescedRid }).queue;
|
||||||
|
assert.equal(compactedQueue.length, 1, 'offline playback burst is one logical queue entry');
|
||||||
|
const replayFrames = materializeMediaIntent(compactedQueue[0]);
|
||||||
|
assert.deepEqual(replayFrames.map(frame => frame.event), ['seek', 'pause'],
|
||||||
|
'compacted intent uses only the minimum existing legacy events');
|
||||||
|
|
||||||
|
const legacyReplayPayloads = [];
|
||||||
|
for (const frame of replayFrames) {
|
||||||
|
s(coalescingSender, frame.event, frame.data);
|
||||||
|
legacyReplayPayloads.push([frame.event, await w(legacyReceiver, frame.event)]);
|
||||||
|
}
|
||||||
|
assert.equal(legacyReplayPayloads[0][1].targetTime, 605);
|
||||||
|
assert.equal(legacyReplayPayloads[1][1].currentTime, 605);
|
||||||
|
for (const [event, payload] of legacyReplayPayloads) {
|
||||||
|
assert.ok(event === 'seek' || event === 'pause');
|
||||||
|
assert.equal(payload.mediaState, undefined);
|
||||||
|
assert.equal(payload.revision, undefined);
|
||||||
|
}
|
||||||
|
const canonicalAfterReplay = mod.rooms.get(coalescedRid).mediaState;
|
||||||
|
assert.equal(canonicalAfterReplay.revision, canonicalBeforeReplay.revision + replayFrames.length);
|
||||||
|
assert.equal(canonicalAfterReplay.playbackState, 'paused');
|
||||||
|
assert.equal(canonicalAfterReplay.currentTime, 605);
|
||||||
|
assert.equal(canonicalAfterReplay.updatedBy, 'coalesce-sender');
|
||||||
|
|
||||||
|
const coalescedLateJoiner = await c();
|
||||||
|
const coalescedLateRoom = await j(coalescedLateJoiner, coalescedRid, 'coalesced-late');
|
||||||
|
assert.equal(coalescedLateRoom.mediaState.revision, canonicalAfterReplay.revision);
|
||||||
|
assert.equal(coalescedLateRoom.mediaState.playbackState, 'paused');
|
||||||
|
assert.equal(coalescedLateRoom.mediaState.currentTime, 605);
|
||||||
|
assert.equal(coalescedLateRoom.mediaState.updatedBy, 'coalesce-sender');
|
||||||
|
close();
|
||||||
|
resetConnectionRate();
|
||||||
|
|
||||||
// --- Capabilities: ROOM_DATA advertises server features for client detection ---
|
// --- Capabilities: ROOM_DATA advertises server features for client detection ---
|
||||||
const capClient = await c();
|
const capClient = await c();
|
||||||
s(capClient, 'join_room', { roomId: 'cap-'+Date.now(), peerId: 'capp', protocolVersion: '1.0.0' });
|
s(capClient, 'join_room', { roomId: 'cap-'+Date.now(), peerId: 'capp', protocolVersion: '1.0.0' });
|
||||||
|
|||||||
@@ -1,4 +1,12 @@
|
|||||||
import { test, expect } from './helpers/extension-fixture.mjs';
|
import { test, expect } from './helpers/extension-fixture.mjs';
|
||||||
|
import { createRequire } from 'node:module';
|
||||||
|
import path from 'node:path';
|
||||||
|
import { fileURLToPath } from 'node:url';
|
||||||
|
import { OFFICIAL_SERVER_TOKEN, PROTOCOL_VERSION } from '../../shared/constants.js';
|
||||||
|
|
||||||
|
const testDir = path.dirname(fileURLToPath(import.meta.url));
|
||||||
|
const require = createRequire(path.join(testDir, '..', '..', 'server', 'package.json'));
|
||||||
|
const NodeWebSocket = require('ws');
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Drives the packed extension itself: real background service worker, real
|
* Drives the packed extension itself: real background service worker, real
|
||||||
@@ -63,6 +71,61 @@ async function applyCanonicalMediaState(context, extensionId, tabId, mediaState)
|
|||||||
}, { tabId, mediaState }));
|
}, { tabId, mediaState }));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async function connectLegacyRelayClient(port) {
|
||||||
|
const socket = new NodeWebSocket(
|
||||||
|
`ws://127.0.0.1:${port}/socket.io/?EIO=4&transport=websocket&version=3.1.3&token=${OFFICIAL_SERVER_TOKEN}`
|
||||||
|
);
|
||||||
|
socket.messages = [];
|
||||||
|
socket.on('message', value => socket.messages.push(value.toString()));
|
||||||
|
await new Promise((resolve, reject) => {
|
||||||
|
const timeout = setTimeout(() => reject(new Error('legacy relay connection timed out')), 5000);
|
||||||
|
socket.once('open', () => {
|
||||||
|
clearTimeout(timeout);
|
||||||
|
resolve();
|
||||||
|
});
|
||||||
|
});
|
||||||
|
socket.send('40');
|
||||||
|
await expect.poll(() => socket.messages.filter(message => message.startsWith('0') || message.startsWith('40')).length).toBeGreaterThanOrEqual(2);
|
||||||
|
socket.messages.length = 0;
|
||||||
|
return socket;
|
||||||
|
}
|
||||||
|
|
||||||
|
function sendLegacyRelayEvent(socket, event, data = {}) {
|
||||||
|
socket.send(`42${JSON.stringify([event, data])}`);
|
||||||
|
}
|
||||||
|
|
||||||
|
async function waitForLegacyRelayEvent(socket, event, timeoutMs = 10_000) {
|
||||||
|
await expect.poll(() => socket.messages.some(message => {
|
||||||
|
if (!message.startsWith('42')) return false;
|
||||||
|
try { return JSON.parse(message.substring(2))[0] === event; } catch { return false; }
|
||||||
|
}), { timeout: timeoutMs }).toBe(true);
|
||||||
|
const index = socket.messages.findIndex(message => {
|
||||||
|
if (!message.startsWith('42')) return false;
|
||||||
|
try { return JSON.parse(message.substring(2))[0] === event; } catch { return false; }
|
||||||
|
});
|
||||||
|
return JSON.parse(socket.messages.splice(index, 1)[0].substring(2))[1];
|
||||||
|
}
|
||||||
|
|
||||||
|
async function joinLegacyRelayRoom(socket, roomId, peerId) {
|
||||||
|
sendLegacyRelayEvent(socket, 'join_room', { roomId, peerId, protocolVersion: PROTOCOL_VERSION });
|
||||||
|
return waitForLegacyRelayEvent(socket, 'room_data');
|
||||||
|
}
|
||||||
|
|
||||||
|
async function terminateExtensionServiceWorker(context, extensionId, page) {
|
||||||
|
const session = await context.newCDPSession(page);
|
||||||
|
try {
|
||||||
|
const { targetInfos } = await session.send('Target.getTargets');
|
||||||
|
const target = targetInfos.find(candidate =>
|
||||||
|
candidate.type === 'service_worker'
|
||||||
|
&& candidate.url.startsWith(`chrome-extension://${extensionId}/`)
|
||||||
|
);
|
||||||
|
if (!target) throw new Error('extension service worker target not found');
|
||||||
|
await session.send('Target.closeTarget', { targetId: target.targetId });
|
||||||
|
} finally {
|
||||||
|
await session.detach();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
async function setAudioSettings(context, extensionId, settings) {
|
async function setAudioSettings(context, extensionId, settings) {
|
||||||
return withExtensionPage(context, extensionId, page => page.evaluate(
|
return withExtensionPage(context, extensionId, page => page.evaluate(
|
||||||
value => chrome.storage.local.set({ audioSettings: value }),
|
value => chrome.storage.local.set({ audioSettings: value }),
|
||||||
@@ -1047,6 +1110,127 @@ test('keeps controlling a Drive-style player across an ordinary play and pause',
|
|||||||
await expect.poll(() => playerFrame.locator('video').evaluate(video => video.paused)).toBe(false);
|
await expect.poll(() => playerFrame.locator('video').evaluate(video => video.paused)).toBe(false);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test('coalesces persisted offline media intent before canonical reconnect recovery', async ({ context, extensionId, baseURL }) => {
|
||||||
|
test.setTimeout(60_000);
|
||||||
|
const relay = await import('../../server/index.js');
|
||||||
|
let legacy = null;
|
||||||
|
let lateJoiner = null;
|
||||||
|
try {
|
||||||
|
await relay.startServer(0, '127.0.0.1');
|
||||||
|
const port = relay.httpServer.address().port;
|
||||||
|
const roomId = `e2e-coalesced-${Date.now()}`;
|
||||||
|
legacy = await connectLegacyRelayClient(port);
|
||||||
|
await joinLegacyRelayRoom(legacy, roomId, 'legacy-e2e');
|
||||||
|
|
||||||
|
const url = `${baseURL}/pages/simple-player.html`;
|
||||||
|
const page = await context.newPage();
|
||||||
|
await page.goto(url);
|
||||||
|
await page.waitForFunction(() => window.__fixtureReady === true);
|
||||||
|
const { tabId } = await selectTargetTab(context, extensionId, url);
|
||||||
|
|
||||||
|
await withExtensionPage(context, extensionId, extensionPage => extensionPage.evaluate(async settings => {
|
||||||
|
await chrome.storage.local.set(settings);
|
||||||
|
return chrome.runtime.sendMessage({ type: 'CONNECT' });
|
||||||
|
}, {
|
||||||
|
serverUrl: `ws://127.0.0.1:${port}`,
|
||||||
|
useCustomServer: true,
|
||||||
|
roomId,
|
||||||
|
password: '',
|
||||||
|
username: 'current-e2e'
|
||||||
|
}));
|
||||||
|
await expect.poll(() => getExtensionState(context, extensionId, { type: 'GET_STATUS' }))
|
||||||
|
.toMatchObject({ status: 'connected', roomId, queuedLogicalEvents: 0 });
|
||||||
|
|
||||||
|
// Establish accepted server truth that would be stale for this client
|
||||||
|
// after its later offline actions.
|
||||||
|
sendLegacyRelayEvent(legacy, 'play', { currentTime: 1, seq: 1, actionTimestamp: 1 });
|
||||||
|
sendLegacyRelayEvent(legacy, 'seek', { currentTime: 1, targetTime: 1, seq: 2, actionTimestamp: 2 });
|
||||||
|
await expect.poll(() => relay.rooms.get(roomId)?.mediaState)
|
||||||
|
.toMatchObject({ revision: 2, playbackState: 'playing', currentTime: 1, updatedBy: 'legacy-e2e' });
|
||||||
|
await expect.poll(() => page.locator('#player').evaluate(video => video.currentTime)).toBeGreaterThan(0.8);
|
||||||
|
legacy.messages.length = 0;
|
||||||
|
|
||||||
|
const connectedStatus = await getExtensionState(context, extensionId, { type: 'GET_STATUS' });
|
||||||
|
const extensionSocketId = Array.from(relay.rooms.get(roomId).peerData.entries())
|
||||||
|
.find(([, data]) => data.peerId === connectedStatus.peerId)?.[0];
|
||||||
|
expect(extensionSocketId).toBeTruthy();
|
||||||
|
// Point future reconnect attempts at an unused local port, then sever
|
||||||
|
// only the extension socket. The relay and legacy peer stay live.
|
||||||
|
await withExtensionPage(context, extensionId, extensionPage => extensionPage.evaluate(
|
||||||
|
serverUrl => chrome.storage.local.set({ serverUrl }),
|
||||||
|
'ws://127.0.0.1:1'
|
||||||
|
));
|
||||||
|
relay.io.sockets.sockets.get(extensionSocketId).disconnect(true);
|
||||||
|
await expect.poll(() => getExtensionState(context, extensionId, { type: 'GET_STATUS' }).then(status => status.status))
|
||||||
|
.not.toBe('connected');
|
||||||
|
|
||||||
|
expect(await sendServerCommand(context, extensionId, tabId, 'play', { currentTime: 2 })).toMatchObject({ status: 'ok' });
|
||||||
|
expect(await sendServerCommand(context, extensionId, tabId, 'seek', { targetTime: 4 })).toMatchObject({ status: 'ok' });
|
||||||
|
expect(await sendServerCommand(context, extensionId, tabId, 'seek', { targetTime: 6 })).toMatchObject({ status: 'ok' });
|
||||||
|
expect(await sendServerCommand(context, extensionId, tabId, 'pause', { currentTime: 6 })).toMatchObject({ status: 'ok' });
|
||||||
|
await expect.poll(() => getExtensionState(context, extensionId, { type: 'GET_STATUS' }))
|
||||||
|
.toMatchObject({ queuedLogicalEvents: 1, queuedMediaIntents: 1, queuedWireEvents: 2 });
|
||||||
|
await expect.poll(() => page.locator('#player').evaluate(video => ({ paused: video.paused, currentTime: video.currentTime })))
|
||||||
|
.toMatchObject({ paused: true });
|
||||||
|
|
||||||
|
// Terminate the actual MV3 worker. The next runtime message starts a new
|
||||||
|
// worker, which must migrate/restore the logical queue and local sequence.
|
||||||
|
await terminateExtensionServiceWorker(context, extensionId, page);
|
||||||
|
await expect.poll(() => getExtensionState(context, extensionId, { type: 'GET_STATUS' }).then(status => status.queuedMediaIntents))
|
||||||
|
.toBe(1);
|
||||||
|
const restoredQueue = await getExtensionState(context, extensionId, { type: 'GET_STATUS' });
|
||||||
|
expect(restoredQueue.queuedLogicalEvents).toBeGreaterThanOrEqual(1);
|
||||||
|
expect(restoredQueue.queuedWireEvents).toBeGreaterThanOrEqual(2);
|
||||||
|
|
||||||
|
await withExtensionPage(context, extensionId, extensionPage => extensionPage.evaluate(async serverUrl => {
|
||||||
|
await chrome.storage.local.set({ serverUrl });
|
||||||
|
chrome.alarms.create('keepAlive', { when: Date.now() + 50 });
|
||||||
|
}, `ws://127.0.0.1:${port}`));
|
||||||
|
|
||||||
|
await page.locator('#player').evaluate(video => {
|
||||||
|
window.__koalaReconnectSeeks = [];
|
||||||
|
video.addEventListener('seeked', () => window.__koalaReconnectSeeks.push(video.currentTime));
|
||||||
|
});
|
||||||
|
const replaySeek = await waitForLegacyRelayEvent(legacy, 'seek', 20_000);
|
||||||
|
const replayPause = await waitForLegacyRelayEvent(legacy, 'pause', 20_000);
|
||||||
|
expect(replaySeek).toMatchObject({ currentTime: 6, targetTime: 6 });
|
||||||
|
expect(replayPause).toMatchObject({ currentTime: 6 });
|
||||||
|
expect(replaySeek.seq).toBeLessThan(replayPause.seq);
|
||||||
|
await expect.poll(() => getExtensionState(context, extensionId, { type: 'GET_STATUS' }))
|
||||||
|
.toMatchObject({ status: 'connected', roomId, queuedLogicalEvents: 0, queuedMediaIntents: 0 });
|
||||||
|
await expect.poll(() => relay.rooms.get(roomId)?.mediaState)
|
||||||
|
.toMatchObject({ playbackState: 'paused', currentTime: 6 });
|
||||||
|
expect(relay.rooms.get(roomId).mediaState.revision).toBeGreaterThan(2);
|
||||||
|
|
||||||
|
// The stale r2 snapshot must never have sought the local player back to 1
|
||||||
|
// before its authorized pending intent replayed.
|
||||||
|
const reconnectSeeks = await page.evaluate(() => window.__koalaReconnectSeeks || []);
|
||||||
|
expect(reconnectSeeks.some(value => value < 3)).toBe(false);
|
||||||
|
await expect.poll(() => page.locator('#player').evaluate(video => ({ paused: video.paused, currentTime: video.currentTime })))
|
||||||
|
.toMatchObject({ paused: true });
|
||||||
|
|
||||||
|
lateJoiner = await connectLegacyRelayClient(port);
|
||||||
|
const lateRoom = await joinLegacyRelayRoom(lateJoiner, roomId, 'late-e2e');
|
||||||
|
expect(lateRoom.mediaState).toMatchObject({
|
||||||
|
revision: relay.rooms.get(roomId).mediaState.revision,
|
||||||
|
playbackState: 'paused',
|
||||||
|
currentTime: 6
|
||||||
|
});
|
||||||
|
|
||||||
|
await page.waitForTimeout(700);
|
||||||
|
const leakedMediaEvents = legacy.messages.filter(message => {
|
||||||
|
if (!message.startsWith('42')) return false;
|
||||||
|
try { return ['play', 'pause', 'seek'].includes(JSON.parse(message.substring(2))[0]); } catch { return false; }
|
||||||
|
});
|
||||||
|
expect(leakedMediaEvents).toEqual([]);
|
||||||
|
} finally {
|
||||||
|
await context.setOffline(false).catch(() => {});
|
||||||
|
try { legacy?.close(); } catch { /* already closed */ }
|
||||||
|
try { lateJoiner?.close(); } catch { /* already closed */ }
|
||||||
|
await relay.stopServerForTests();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
function FRAMED_VIDEO_PAUSED() {
|
function FRAMED_VIDEO_PAUSED() {
|
||||||
return document.querySelector('iframe').contentDocument.querySelector('video').paused;
|
return document.querySelector('iframe').contentDocument.querySelector('video').paused;
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user