mirror of
https://github.com/telemt/telemt.git
synced 2026-10-30 20:40:51 +03:00
Native macOS status-schema preserved
Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com>
This commit is contained in:
+14
-21
@@ -23,13 +23,7 @@ const pending=[],upPending=[],recoveryPending=[],lanes=new Map(),closedLanes=new
|
|||||||
const canonicalFailures=['timeout','network','upgrade','http','protocol'];
|
const canonicalFailures=['timeout','network','upgrade','http','protocol'];
|
||||||
const failure=(reason,message)=>Object.assign(new Error(message||reason),{telemtReason:reason});
|
const failure=(reason,message)=>Object.assign(new Error(message||reason),{telemtReason:reason});
|
||||||
const failureReason=(error,fallback)=>error&&canonicalFailures.includes(error.telemtReason)?error.telemtReason:fallback;
|
const failureReason=(error,fallback)=>error&&canonicalFailures.includes(error.telemtReason)?error.telemtReason:fallback;
|
||||||
const status=(state,phase,reason,deadlineMs)=>{
|
const status=state=>{if(port&&!closed)port.postMessage({t:'status',state})};
|
||||||
if(!port||closed)return;
|
|
||||||
const currentPhase=phase||(state==='connected'?'committed':state==='reconnecting'?'retrying':state==='failed'?'terminal':createStarted?'negotiating':'starting');
|
|
||||||
let currentDeadline=deadlineMs;
|
|
||||||
if(currentDeadline===undefined)currentDeadline=negotiationStartedAt?Math.max(0,negotiationStartedAt+negotiatedFinalDeadline*1000-Date.now()):0;
|
|
||||||
port.postMessage({t:'status',state,phase:currentPhase,reason:reason||'',deadline_ms:Math.max(0,Math.ceil(currentDeadline))});
|
|
||||||
};
|
|
||||||
const socketURL=()=>relayOrigin.replace(/^https:/,'wss:')+'/api/v1/ws';
|
const socketURL=()=>relayOrigin.replace(/^https:/,'wss:')+'/api/v1/ws';
|
||||||
const requestClient=requestSupport.create({
|
const requestClient=requestSupport.create({
|
||||||
origin:()=>relayOrigin,closed:()=>closed,retryMs:()=>bridgeRetryMs,longPollMs:()=>longPollMs,requestMs:()=>bridgeRequestMs,
|
origin:()=>relayOrigin,closed:()=>closed,retryMs:()=>bridgeRetryMs,longPollMs:()=>longPollMs,requestMs:()=>bridgeRequestMs,
|
||||||
@@ -48,7 +42,7 @@ function settleBatch(lease){if(!buffers.settleBatch(lease))return false;detachLe
|
|||||||
function cancelBatch(lease){if(!lease||lease.settled)return;buffers.cancelBatch(lease);detachLease(lease)}
|
function cancelBatch(lease){if(!lease||lease.settled)return;buffers.cancelBatch(lease);detachLease(lease)}
|
||||||
const attemptHeaders=(attempt,failure)=>negotiationEnabled?Object.assign({'X-Carrier-Capabilities':carrierCapabilities,'X-Carrier-Attempt':String(attempt)},failure?{'X-Carrier-Failure':failure}:{}):{};
|
const attemptHeaders=(attempt,failure)=>negotiationEnabled?Object.assign({'X-Carrier-Capabilities':carrierCapabilities,'X-Carrier-Attempt':String(attempt)},failure?{'X-Carrier-Failure':failure}:{}):{};
|
||||||
function finishOldRecovery(){
|
function finishOldRecovery(){
|
||||||
recoveryReplaced=false;resetScheduler();status('connected','committed','',0);
|
recoveryReplaced=false;resetScheduler();status('connected');
|
||||||
while(recoveryPending.length&&!closed){const data=recoveryPending.shift();release(data.byteLength,1,null);queueCarrier(data)}
|
while(recoveryPending.length&&!closed){const data=recoveryPending.shift();release(data.byteLength,1,null);queueCarrier(data)}
|
||||||
}
|
}
|
||||||
function rejectRecoveryCommit(error){
|
function rejectRecoveryCommit(error){
|
||||||
@@ -110,13 +104,13 @@ function sampleScheduler(){
|
|||||||
}
|
}
|
||||||
function observeResumeTrigger(){
|
function observeResumeTrigger(){
|
||||||
const gap=Math.max(schedulerGapPending,schedulerGap());schedulerGapPending=0;if(!carrierCommitted||closed)return;
|
const gap=Math.max(schedulerGapPending,schedulerGap());schedulerGapPending=0;if(!carrierCommitted||closed)return;
|
||||||
if(gap>=2*longPollMs)status('reconnecting','retrying','',bridgeRecoveryMs);
|
if(gap>=2*longPollMs)status('reconnecting');
|
||||||
if(gap>=reconnectGraceMs&&!recoveryController.active())recoveryController.recover('timeout',null);
|
if(gap>=reconnectGraceMs&&!recoveryController.active())recoveryController.recover('timeout',null);
|
||||||
}
|
}
|
||||||
function fail(reason){
|
function fail(reason){
|
||||||
if(closed)return;reason=reason||'protocol';if(canonicalFailures.includes(reason))terminalFailure=reason;
|
if(closed)return;reason=reason||'protocol';if(canonicalFailures.includes(reason))terminalFailure=reason;
|
||||||
rejectRecoveryCommit(failure(reason));
|
rejectRecoveryCommit(failure(reason));
|
||||||
status('failed','terminal',reason,0);if(port)port.postMessage({t:'close'});close(true);
|
status('failed');if(port)port.postMessage({t:'close'});close(true);
|
||||||
}
|
}
|
||||||
function knownCarrier(value){return value==='https'||value==='https-lanes'||value==='websocket'||value==='websocket-lanes'}
|
function knownCarrier(value){return value==='https'||value==='https-lanes'||value==='websocket'||value==='websocket-lanes'}
|
||||||
function sessionEcho(response,expectedAttempt,states,exactAttempt){
|
function sessionEcho(response,expectedAttempt,states,exactAttempt){
|
||||||
@@ -184,7 +178,7 @@ async function resolveAttempt(reason,epoch,snapshot){
|
|||||||
if(!token||cursor!=='0'||(snapshot.selected&&echo.selected!==snapshot.selected))throw new Error('changed carrier replay');
|
if(!token||cursor!=='0'||(snapshot.selected&&echo.selected!==snapshot.selected))throw new Error('changed carrier replay');
|
||||||
const welcome=response.body;if(closed||epoch!==attemptEpoch)return;
|
const welcome=response.body;if(closed||epoch!==attemptEpoch)return;
|
||||||
cleanupToken=token;
|
cleanupToken=token;
|
||||||
if(!welcomeSent){welcomeSent=true;port.postMessage(welcome,[welcome]);status('connecting','provisional','',Math.max(0,negotiationStartedAt+negotiatedFinalDeadline*1000-Date.now()))}
|
if(!welcomeSent){welcomeSent=true;port.postMessage(welcome,[welcome]);status('connecting')}
|
||||||
if(echo.state!=='provisional'){switching=false;fail('protocol');return}
|
if(echo.state!=='provisional'){switching=false;fail('protocol');return}
|
||||||
advanceConfirmed(reason,epoch);
|
advanceConfirmed(reason,epoch);
|
||||||
}catch(error){if(!closed&&epoch===attemptEpoch){switching=false;fail(failureReason(error,'protocol'))}}
|
}catch(error){if(!closed&&epoch===attemptEpoch){switching=false;fail(failureReason(error,'protocol'))}}
|
||||||
@@ -222,7 +216,7 @@ async function createSession(epoch){
|
|||||||
if(!token||cursor!=='0'){advanceCarrier('protocol',epoch);return}
|
if(!token||cursor!=='0'){advanceCarrier('protocol',epoch);return}
|
||||||
const welcome=response.body;if(closed||epoch!==attemptEpoch)return;
|
const welcome=response.body;if(closed||epoch!==attemptEpoch)return;
|
||||||
carrier=selected;sessionToken=token;cleanupToken=token;downCursor=cursor;
|
carrier=selected;sessionToken=token;cleanupToken=token;downCursor=cursor;
|
||||||
if(!welcomeSent){welcomeSent=true;port.postMessage(welcome,[welcome]);status('connecting','provisional','',Math.max(0,negotiationStartedAt+negotiatedFinalDeadline*1000-Date.now()))}
|
if(!welcomeSent){welcomeSent=true;port.postMessage(welcome,[welcome]);status('connecting')}
|
||||||
if(carrier==='websocket')openCandidateSocket(null,null,epoch);
|
if(carrier==='websocket')openCandidateSocket(null,null,epoch);
|
||||||
maybeStartCandidate();
|
maybeStartCandidate();
|
||||||
}catch(error){if(closed||epoch!==attemptEpoch)return;advanceCarrier(failureReason(error,'network'),epoch)}
|
}catch(error){if(closed||epoch!==attemptEpoch)return;advanceCarrier(failureReason(error,'network'),epoch)}
|
||||||
@@ -245,7 +239,7 @@ function commitCarrier(probe,epoch){
|
|||||||
carrierCommitted=true;candidateRunning=false;if(carrierTimer)clearTimeout(carrierTimer);carrierTimer=null;
|
carrierCommitted=true;candidateRunning=false;if(carrierTimer)clearTimeout(carrierTimer);carrierTimer=null;
|
||||||
resetScheduler();
|
resetScheduler();
|
||||||
attemptController=null;currentAttempt=null;
|
attemptController=null;currentAttempt=null;
|
||||||
status('connected','committed','',0);
|
status('connected');
|
||||||
if(carrier==='https')poll();
|
if(carrier==='https')poll();
|
||||||
else if(carrier==='https-lanes'){const lane=lanes.get(probe.id);if(lane&&!lane.polling)pollLane(lane)}
|
else if(carrier==='https-lanes'){const lane=lanes.get(probe.id);if(lane&&!lane.polling)pollLane(lane)}
|
||||||
while(pending.length&&!closed){const data=pending.shift();release(data.byteLength,1,null);queueCarrier(data)}
|
while(pending.length&&!closed){const data=pending.shift();release(data.byteLength,1,null);queueCarrier(data)}
|
||||||
@@ -395,17 +389,16 @@ 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;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)};
|
const opened=lane.socket,openTimer=setTimeout(()=>{if(!closed&&lanes.get(lane.id)===lane&&lane.socket===opened)finishLane(lane,true)},websocketOpenMs);
|
||||||
openTimer=setTimeout(()=>finishSocket('timeout'),websocketOpenMs);
|
lane.socket.onopen=()=>{if(closed||lanes.get(lane.id)!==lane)return;lane.ready=true;status('connected');runLaneSocketUp(lane)};
|
||||||
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=>{
|
||||||
if(openTimer)clearTimeout(openTimer);openTimer=null;
|
clearTimeout(openTimer);
|
||||||
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=()=>finishSocket(upgraded?'network':'upgrade');
|
lane.socket.onerror=()=>{};lane.socket.onclose=()=>{clearTimeout(openTimer);lane.ready=false;lane.socket=null;if(!closed)finishLane(lane,true)};
|
||||||
}
|
}
|
||||||
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;
|
||||||
@@ -505,13 +498,13 @@ function activatePort(nextPort){
|
|||||||
else if(!carrierCommitted){if(!reserve(data,null)){fail('capacity');return}pending.push(data);maybeStartCandidate()}
|
else if(!carrierCommitted){if(!reserve(data,null)){fail('capacity');return}pending.push(data);maybeStartCandidate()}
|
||||||
else queueCarrier(data);
|
else queueCarrier(data);
|
||||||
}
|
}
|
||||||
}else if(message.data&&message.data.t==='close'){status('failed','terminal','closed',0);close(true)}
|
}else if(message.data&&message.data.t==='close'){status('failed');close(true)}
|
||||||
};
|
};
|
||||||
port.start();status('connecting','starting','',bridgeRequestMs);helloTimer=setTimeout(()=>fail('timeout'),bridgeRequestMs);
|
port.start();status('connecting');helloTimer=setTimeout(()=>fail('timeout'),bridgeRequestMs);
|
||||||
}
|
}
|
||||||
recoveryController=recoverySupport.create({
|
recoveryController=recoverySupport.create({
|
||||||
budgetMs:()=>bridgeRecoveryMs,requestMs:()=>bridgeRequestMs,url:()=>relayOrigin+recoveryPath,token:()=>cleanupToken||sessionToken,
|
budgetMs:()=>bridgeRecoveryMs,requestMs:()=>bridgeRequestMs,url:()=>relayOrigin+recoveryPath,token:()=>cleanupToken||sessionToken,
|
||||||
read:(response,limit,exact,signal)=>responseBody.read(response,limit,exact,signal),cancel:responseBody.cancel,status:remaining=>status('reconnecting','retrying','',remaining),
|
read:(response,limit,exact,signal)=>responseBody.read(response,limit,exact,signal),cancel:responseBody.cancel,status:()=>status('reconnecting'),
|
||||||
restored:finishOldRecovery,replace:replaceCarrier,replaceable:error=>failureReason(error,'network')!=='protocol',
|
restored:finishOldRecovery,replace:replaceCarrier,replaceable:error=>failureReason(error,'network')!=='protocol',
|
||||||
reason:(error,fallback)=>failureReason(error,fallback),terminal:reason=>fail(recoveryController.remaining()<=0?'timeout':reason)
|
reason:(error,fallback)=>failureReason(error,fallback),terminal:reason=>fail(recoveryController.remaining()<=0?'timeout':reason)
|
||||||
});
|
});
|
||||||
|
|||||||
+9
-17
@@ -179,24 +179,16 @@ fn ambiguous_commit_is_resolved_before_carrier_advance() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn committed_websocket_lane_escalates_only_pre_upgrade_failure() {
|
fn rendered_page_preserves_exact_v1_status_control_envelope() {
|
||||||
let page = render_page("HHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHH", 4);
|
let page = render_page("HHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHH", 4);
|
||||||
|
|
||||||
assert!(page.body.contains("let upgraded=false,settled=false"));
|
assert_eq!(
|
||||||
assert!(page.body.contains(
|
page.body
|
||||||
"if(settled||closed||lanes.get(lane.id)!==lane||lane.socket!==opened)return"
|
.matches("port.postMessage({t:'status',state})")
|
||||||
));
|
.count(),
|
||||||
assert!(page.body.contains(
|
1
|
||||||
"if(!upgraded){lane.socket=null;opened.close();recoveryController.recover(reason,null);return}"
|
);
|
||||||
));
|
assert!(!page
|
||||||
assert!(page
|
|
||||||
.body
|
.body
|
||||||
.contains("recoveryController.recover(reason,null);return}finishLane(lane,true)"));
|
.contains("port.postMessage({t:'status',state,"));
|
||||||
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