mirror of
https://github.com/telemt/telemt.git
synced 2026-10-30 12:30:34 +03:00
Stale Websocket-lanes recovers after restart
Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com>
This commit is contained in:
@@ -395,16 +395,17 @@ function queueLane(value){
|
|||||||
}
|
}
|
||||||
function openLaneSocket(lane){
|
function openLaneSocket(lane){
|
||||||
if(lane.socket||closed)return;lane.socket=new WebSocket(socketURL(),'tproxy-lane-v1.'+sessionToken+'.'+String(lane.id));lane.socket.binaryType='arraybuffer';
|
if(lane.socket||closed)return;lane.socket=new WebSocket(socketURL(),'tproxy-lane-v1.'+sessionToken+'.'+String(lane.id));lane.socket.binaryType='arraybuffer';
|
||||||
const opened=lane.socket,openTimer=setTimeout(()=>{if(!closed&&lanes.get(lane.id)===lane&&lane.socket===opened)finishLane(lane,true)},websocketOpenMs);
|
const opened=lane.socket;let upgraded=false,settled=false,openTimer=null;const finishSocket=reason=>{if(settled||closed||lanes.get(lane.id)!==lane||lane.socket!==opened)return;settled=true;if(openTimer)clearTimeout(openTimer);openTimer=null;lane.ready=false;if(!upgraded){lane.socket=null;opened.close();recoveryController.recover(reason,null);return}finishLane(lane,true)};
|
||||||
lane.socket.onopen=()=>{if(closed||lanes.get(lane.id)!==lane)return;lane.ready=true;status('connected');runLaneSocketUp(lane)};
|
openTimer=setTimeout(()=>finishSocket('timeout'),websocketOpenMs);
|
||||||
|
lane.socket.onopen=()=>{if(closed||lanes.get(lane.id)!==lane||lane.socket!==opened){opened.close();return}upgraded=true;lane.ready=true;status('connected');runLaneSocketUp(lane)};
|
||||||
lane.socket.onmessage=event=>{
|
lane.socket.onmessage=event=>{
|
||||||
clearTimeout(openTimer);
|
if(openTimer)clearTimeout(openTimer);openTimer=null;
|
||||||
if(closed||lanes.get(lane.id)!==lane||!(event.data instanceof ArrayBuffer)){finishLane(lane,true);return}
|
if(closed||lanes.get(lane.id)!==lane||!(event.data instanceof ArrayBuffer)){finishLane(lane,true);return}
|
||||||
let values;try{values=splitFrames(event.data);for(const value of values)if(value.id!==lane.id)throw new Error('cross-lane frame')}catch(error){finishLane(lane,true);return}
|
let values;try{values=splitFrames(event.data);for(const value of values)if(value.id!==lane.id)throw new Error('cross-lane frame')}catch(error){finishLane(lane,true);return}
|
||||||
if(values.some(value=>value.type===3))lane.remoteClosed=true;
|
if(values.some(value=>value.type===3))lane.remoteClosed=true;
|
||||||
observeServerFrames(event.data);port.postMessage({t:'traffic',up:0,down:event.data.byteLength});port.postMessage(event.data,[event.data]);status('connected');
|
observeServerFrames(event.data);port.postMessage({t:'traffic',up:0,down:event.data.byteLength});port.postMessage(event.data,[event.data]);status('connected');
|
||||||
};
|
};
|
||||||
lane.socket.onerror=()=>{};lane.socket.onclose=()=>{clearTimeout(openTimer);lane.ready=false;lane.socket=null;if(!closed)finishLane(lane,true)};
|
lane.socket.onerror=()=>{};lane.socket.onclose=()=>finishSocket(upgraded?'network':'upgrade');
|
||||||
}
|
}
|
||||||
async function runLaneSocketUp(lane){
|
async function runLaneSocketUp(lane){
|
||||||
if(lane.running||!lane.ready)return;lane.running=true;let lease=null;
|
if(lane.running||!lane.ready)return;lane.running=true;let lease=null;
|
||||||
|
|||||||
@@ -177,3 +177,26 @@ fn ambiguous_commit_is_resolved_before_carrier_advance() {
|
|||||||
.contains("addEventListener('pagehide',()=>fail('navigation')")
|
.contains("addEventListener('pagehide',()=>fail('navigation')")
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn committed_websocket_lane_escalates_only_pre_upgrade_failure() {
|
||||||
|
let page = render_page("HHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHH", 4);
|
||||||
|
|
||||||
|
assert!(page.body.contains("let upgraded=false,settled=false"));
|
||||||
|
assert!(page.body.contains(
|
||||||
|
"if(settled||closed||lanes.get(lane.id)!==lane||lane.socket!==opened)return"
|
||||||
|
));
|
||||||
|
assert!(page.body.contains(
|
||||||
|
"if(!upgraded){lane.socket=null;opened.close();recoveryController.recover(reason,null);return}"
|
||||||
|
));
|
||||||
|
assert!(page
|
||||||
|
.body
|
||||||
|
.contains("recoveryController.recover(reason,null);return}finishLane(lane,true)"));
|
||||||
|
assert!(page
|
||||||
|
.body
|
||||||
|
.contains("openTimer=setTimeout(()=>finishSocket('timeout'),websocketOpenMs)"));
|
||||||
|
assert!(page.body.contains("upgraded=true;lane.ready=true"));
|
||||||
|
assert!(page
|
||||||
|
.body
|
||||||
|
.contains("lane.socket.onclose=()=>finishSocket(upgraded?'network':'upgrade')"));
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user