diff --git a/services/bstein-dev-home/av-observer/activity-worklet.js b/services/bstein-dev-home/av-observer/activity-worklet.js index 27b4b8a0..c96032cf 100644 --- a/services/bstein-dev-home/av-observer/activity-worklet.js +++ b/services/bstein-dev-home/av-observer/activity-worklet.js @@ -1,12 +1,25 @@ // A cumulative, one-credit metadata reporter: never copies or sends audio. class Activity extends AudioWorkletProcessor { - constructor() { + constructor(options) { super(); this.frames=0; this.inputFrames=0; this.zeroFrames=0; this.energy=0; this.peak=0; this.credit=true; this.next=0; this.stopped=false; - this.port.onmessage=({data})=>{ if(data==='ack')this.credit=true; if(data==='stop')this.stopped=true; }; + this.watch=options?.processorOptions?.continuityWatch===true; + this.startedFrame=this.watch?currentFrame:0;this.lastAckFrame=this.startedFrame; + this.port.onmessage=({data})=>{ + if(data==='ack'){this.credit=true;if(this.watch)this.lastAckFrame=currentFrame;} + if(data==='stop')this.stopped=true; + }; } process(inputs,outputs) { - if(this.stopped || this.frames >= sampleRate*125) return false; + if(this.stopped)return false; + if(this.watch) { + // Long watches require a live UI owner. Withheld ACKs stop work + // autonomously; they never renew the fixed ten-hour deadline. + const age=currentFrame-this.startedFrame,unacked=currentFrame-this.lastAckFrame; + if(!Number.isSafeInteger(age)||age<0||unacked<0||age>=sampleRate*36025||unacked>sampleRate*10) { + this.stopped=true;this.port.postMessage({watchStopped:true});return false; + } + } else if(this.frames >= sampleRate*125)return false; const block=outputs[0]?.[0]?.length ?? 128; const samples=inputs[0]?.[0]; this.frames+=block; diff --git a/services/bstein-dev-home/av-observer/capture-lifecycle.js b/services/bstein-dev-home/av-observer/capture-lifecycle.js new file mode 100644 index 00000000..d38665b4 --- /dev/null +++ b/services/bstein-dev-home/av-observer/capture-lifecycle.js @@ -0,0 +1,27 @@ +// Independent cleanup attempts: a broken optional recorder or audio node must +// never keep the camera/microphone open or prevent the remaining owners closing. +const attempt=operation=>{try{operation();}catch{/* Continue releasing other owners. */}}; + +export function stopTracks(stream) { + if(!stream)return; + for(const track of stream.getTracks())attempt(()=>{track.onended=null;track.stop();}); +} + +export function releaseCapture(run,preview,finishRecording) { + if(run.released)return; + run.released=true; + attempt(()=>stopTracks(run.stream)); + for(const timer of [run.timer,run.setupTimer,run.signalTimer])clearTimeout(timer); + clearInterval(run.updateTimer); + if(run.callback!=null)attempt(()=>preview.cancelVideoFrameCallback(run.callback)); + if(run.recording)attempt(finishRecording); + if(run.node) { + run.node.onprocessorerror=null; + run.node.port.onmessage=null; + attempt(()=>run.node.port.postMessage('stop')); + attempt(()=>run.node.disconnect()); + } + attempt(()=>run.source?.disconnect()); + // A rejected or unresolved AudioContext.close never postpones track release. + attempt(()=>run.context?.close().catch(()=>{})); +} diff --git a/services/bstein-dev-home/av-observer/continuity-signals.js b/services/bstein-dev-home/av-observer/continuity-signals.js index a381a781..1c662e53 100644 --- a/services/bstein-dev-home/av-observer/continuity-signals.js +++ b/services/bstein-dev-home/av-observer/continuity-signals.js @@ -1,4 +1,6 @@ // Known synthetic signatures only. No PCM/frame retention or content upload. +import {PilotWindow} from './pilot-window.js'; + export class VideoContinuity { constructor(){this.frames=0;this.bad=0;this.duplicates=0;this.missing=0;this.ambiguous=0;this.previous=null;} observe(rgba,timeMs) { @@ -37,17 +39,19 @@ export class AudioPilot { constructor(rate) { if(!Number.isFinite(rate)||rate<16000||rate>192000)throw new RangeError('sample rate'); this.rate=rate;this.size=Math.round(rate*.01);this.coeff=2*Math.cos(2*Math.PI*7000/rate); + this.windowMs=this.size/rate*1000;this.recent=new PilotWindow(this.windowMs); this.q1=0;this.q2=0;this.count=0;this.windows=0;this.missing=0;this.pending=0; this.maxGap=0;this.first=null;this.last=null;this.expected=null;this.discontinuities=0; } process(samples,frame) { if(this.expected!==null&&frame!==this.expected) { this.discontinuities++;this.q1=0;this.q2=0;this.count=0; + this.recent.interrupt(); } this.expected=frame+samples.length; for(let i=0;iNumber.isSafeInteger(value)&&value>=0; +const number=value=>Number.isFinite(value)&&value>=0; + +export class ContinuityWatch { + constructor(now=()=>performance.now(),wall=()=>Date.now()) { + this.now=now;this.wall=wall;this.started=now();this.wallStarted=wall(); + this.previousAt=this.started;this.nextRow=0;this.rows=[];this.findings={}; + this.previous=null;this.lastAudioProgress=0;this.lastVideoProgress=0; + this.latest=null;this.terminal=null;this.maxObserverGap=0; + } + note(kind,elapsed) { + const finding=this.findings[kind]??{firstObservedMs:elapsed,observations:0}; + finding.observations++;this.findings[kind]=finding; + } + observe(sample,audioReceivedAt) { + if(this.terminal)return; + const now=this.now(),elapsed=now-this.started,wallElapsed=this.wall()-this.wallStarted; + if(!number(elapsed)||!number(wallElapsed)||now2500)this.note('observer-coverage-gap',elapsed); + if(Math.abs(wallElapsed-elapsed)>2000)this.note('wall-monotonic-divergence',elapsed); + const audio=sample.audio??{},video=sample.video??{}; + const audioAge=audioReceivedAt===null?elapsed:now-audioReceivedAt; + const row={elapsedMs:elapsed,videoCallbacks:video.callbacks,videoResets:video.clockOrCounterResets, + audioInputFrames:audio.inputFrames,audioProcessedFrames:audio.processedFrames, + audioZeroFrames:audio.exactZeroFrames,audioRate:audio.sampleRate, + audioMessageAgeMs:number(audioAge)?audioAge:null, + videoCallbackAgeMs:number(video.observationAgeMs)?video.observationAgeMs:null, + videoGapMaxMs:number(video.callbackGapMaxMs)?video.callbackGapMaxMs:null}; + for(const key of ['videoCallbacks','videoResets','audioInputFrames','audioProcessedFrames','audioZeroFrames','audioRate']) { + if(!integer(row[key]))row[key]=null; + } + if(row.audioMessageAgeMs===null||row.audioMessageAgeMs>2000)this.note('audio-observation-stale',elapsed); + if(row.videoCallbackAgeMs===null||row.videoCallbackAgeMs>2000) { + if(elapsed>2000)this.note('video-observation-stale',elapsed); + } + if(row.audioInputFrames===null||row.audioProcessedFrames===null||row.audioRate===null) { + if(elapsed>2000)this.note('audio-counters-unavailable',elapsed); + } + const previous=this.previous; + if(previous) { + for(const key of ['videoCallbacks','audioInputFrames','audioProcessedFrames','audioZeroFrames']) { + if(row[key]!==null&&previous[key]!==null&&row[key]previous.videoResets) + this.note('video-clock-or-counter-reset',elapsed); + } + if(row.audioInputFrames!==null&&(!previous||row.audioInputFrames>previous.audioInputFrames))this.lastAudioProgress=elapsed; + if(row.videoCallbacks!==null&&(!previous||row.videoCallbacks>previous.videoCallbacks))this.lastVideoProgress=elapsed; + if(elapsed-this.lastAudioProgress>2000)this.note('audio-input-not-advancing',elapsed); + if(elapsed-this.lastVideoProgress>2000)this.note('video-not-advancing',elapsed); + this.previous=row;this.latest=row; + if(elapsed>=this.nextRow) { + this.append(row);this.nextRow=elapsed+30000; + } + } + append(row) { + if(this.rows.lengththis.findings[key]); + this.terminal=reason==='10 hours observed'&&duration>=WATCH_MS&&!observerIncomplete? + (Object.keys(this.findings).length?'duration-observed-with-findings':'duration-observed'):'incomplete'; + } + summary(includeTimeline=false) { + return {kind:'receiver-continuity-diagnostic',state:this.terminal??'observing', + startedUnixMs:integer(this.wallStarted)?this.wallStarted:null,wallClockCalibrated:false, + requestedDurationMs:WATCH_MS,observedDurationMs:this.latest?.elapsedMs??0, + maxObserverGapMs:this.maxObserverGap,findings:structuredClone(this.findings), + retainedRows:this.rows.length,rowLimit:WATCH_ROWS,latest:this.latest?{...this.latest}:null, + ...(includeTimeline?{timeline:this.rows.map(row=>({...row}))}:{}), + releaseEligible:false,mediaRecorded:false,uploaded:false, + note:'Browser observations only. Silence is not loss; clocks and callback gaps do not prove audible quality or A/V sync.'}; + } +} diff --git a/services/bstein-dev-home/av-observer/index.html b/services/bstein-dev-home/av-observer/index.html index fd2382fd..3943294c 100644 --- a/services/bstein-dev-home/av-observer/index.html +++ b/services/bstein-dev-home/av-observer/index.html @@ -4,11 +4,13 @@ A/V Test

A/V Test

-

Select a camera and microphone. Numeric metrics report automatically; audio and video never upload.

+

Select a camera and microphone. Quick checks report numeric metrics automatically. Watches keep metrics here for Save metrics; audio and video never upload.

-
+
+
+

Stopped

diff --git a/services/bstein-dev-home/av-observer/kustomization.yaml b/services/bstein-dev-home/av-observer/kustomization.yaml index 6432d9e9..6bb69675 100644 --- a/services/bstein-dev-home/av-observer/kustomization.yaml +++ b/services/bstein-dev-home/av-observer/kustomization.yaml @@ -22,6 +22,9 @@ configMapGenerator: - paired-signals.js - test-signals.js - continuity-signals.js + - pilot-window.js + - capture-lifecycle.js + - continuity-watch.js - style.css - name: lesavka-av-observer-receipts files: diff --git a/services/bstein-dev-home/av-observer/metrics.js b/services/bstein-dev-home/av-observer/metrics.js index e42fb1a0..2efa8342 100644 --- a/services/bstein-dev-home/av-observer/metrics.js +++ b/services/bstein-dev-home/av-observer/metrics.js @@ -1,23 +1,43 @@ // Metadata only. No frames or PCM are retained. These are OBSERVER metrics, // never an automatically passing sync/freshness/artifact report. export class VideoMetrics { - constructor() { this.last = null; this.intervals = []; this.callbacks = 0; this.unobserved = 0; this.resets = 0; } + constructor(now=()=>performance.now()) { + this.last = null; this.intervals = []; this.callbacks = 0; + this.unobserved = 0; this.resets = 0; this.overflow = 0; this.maxGap = null; + this.now=now;this.lastObservedAt=null;this.clockInvalid=false; + } observe(timeMs, presentedFrames) { if (!Number.isFinite(timeMs) || !Number.isSafeInteger(presentedFrames) || presentedFrames < 0) return; + const receivedAt=this.now(); + if(!Number.isFinite(receivedAt)|| + (this.lastObservedAt!==null&&receivedAta-b); + // Once truncated, the lifetime percentile is unavailable. Long watches + // must not repeatedly sort an old 15,000-entry history they cannot use. + const sorted = this.overflow ? [] : [...this.intervals].sort((a,b)=>a-b); + const now=this.now(),elapsed=this.lastObservedAt===null?null:now-this.lastObservedAt; + const age=!this.clockInvalid&&Number.isFinite(now)&&Number.isFinite(elapsed)&&elapsed>=0?elapsed:null; + // A stopped callback stream is visible without waiting for its next + // frame. This is observer silence, not proof of a transport freeze. + const gap=age===null?this.maxGap:Math.max(this.maxGap??0,age); return {callbacks:this.callbacks,unobservedPresentedFrames:this.unobserved,clockOrCounterResets:this.resets, - callbackIntervalP95Ms:sorted.length ? sorted[Math.ceil(sorted.length*.95)-1] : null, - callbackGapMaxMs:sorted.at(-1) ?? null, + callbackIntervalP95Ms:!this.overflow&&sorted.length ? sorted[Math.ceil(sorted.length*.95)-1] : null, + callbackGapMaxMs:gap,observationAgeMs:age,intervalSampleOverflow:this.overflow, note:"Browser presentation callbacks; gaps are not proof of transport loss or corrupted video."}; } } diff --git a/services/bstein-dev-home/av-observer/nginx.conf b/services/bstein-dev-home/av-observer/nginx.conf index 1957b287..d2b5434f 100644 --- a/services/bstein-dev-home/av-observer/nginx.conf +++ b/services/bstein-dev-home/av-observer/nginx.conf @@ -42,7 +42,7 @@ http { if ($request_method !~ ^(GET|HEAD)$) { return 405; } try_files /av-test/index.html =404; } - location ~ ^/av-test/(index\.html|observer\.js|local-recording\.js|review-controls\.js|metrics\.js|reporting\.js|activity-worklet\.js|paired-observer\.js|paired-worklet\.js|paired-signals\.js|test-signals\.js|continuity-signals\.js|style\.css)$ { + location ~ ^/av-test/(index\.html|observer\.js|local-recording\.js|review-controls\.js|metrics\.js|reporting\.js|activity-worklet\.js|paired-observer\.js|paired-worklet\.js|paired-signals\.js|test-signals\.js|continuity-signals\.js|pilot-window\.js|capture-lifecycle\.js|continuity-watch\.js|style\.css)$ { if ($request_method !~ ^(GET|HEAD)$) { return 405; } try_files $uri =404; } diff --git a/services/bstein-dev-home/av-observer/observer.js b/services/bstein-dev-home/av-observer/observer.js index 252eb9b7..73cb54e8 100644 --- a/services/bstein-dev-home/av-observer/observer.js +++ b/services/bstein-dev-home/av-observer/observer.js @@ -2,20 +2,38 @@ import {VideoMetrics,receipt} from './metrics.js'; import {RunReporter} from './reporting.js'; import {PairedObserver} from './paired-observer.js'; import {reviewControls} from './review-controls.js'; +import {releaseCapture,stopTracks} from './capture-lifecycle.js'; +import {ContinuityWatch,WATCH_MS} from './continuity-watch.js'; const el=id=>document.getElementById(id); const review=reviewControls(document); let generation=0, current=null, last=null, reportingRun=null; -const stopTracks=stream=>stream?.getTracks().forEach(track=>track.stop()); +function modeControls() { + const watch=el('observation-mode').value==='watch'; + el('paired-signal').disabled=watch||current!==null; + review.lock(watch||current!==null); + el('watch-help').hidden=!watch; + el('start').textContent=watch?'Start 10-hour watch':'Start 2-minute check'; +} +el('observation-mode').onchange=()=>{ + if(current)return; + if(el('observation-mode').value==='watch') { + el('paired-signal').checked=false;el('record-review').checked=false;review.discard(); + } + modeControls(); +}; function release(run) { - if(run.released)return; - run.released=true; - if(run.recording)review.finish(); - clearTimeout(run.timer); clearTimeout(run.setupTimer); clearTimeout(run.signalTimer); - if(run.callback)el('preview').cancelVideoFrameCallback(run.callback); - run.node?.port.postMessage('stop'); run.node?.disconnect(); run.source?.disconnect(); - stopTracks(run.stream); - // Do not wait on a browser AudioContext close before fencing a stopped run. - run.context?.close().catch(()=>{}); + releaseCapture(run,el('preview'),()=>review.finish()); +} +function update(run) { + if(current!==run)return; + const sample=receipt(run.video,run.audio,run.settings,performance.now()-run.started,'observing'); + if(run.paired)sample.paired=run.paired.summary(); + if(run.watch) { + run.watch.observe(sample,run.audioReceivedAt); + sample.watch=run.watch.summary(); + } + el('metrics').textContent=JSON.stringify(sample,null,2); + if(run.observing)run.reporter.observe(sample); } function stop(reason='operator') { generation++; @@ -24,15 +42,19 @@ function stop(reason='operator') { release(run); last=receipt(run.video,run.audio,run.settings,performance.now()-run.started,reason); if(run.paired)last.paired=run.paired.summary(); + if(run.watch) { + run.watch.observe(last,run.audioReceivedAt);run.watch.finish(reason); + last.watch=run.watch.summary(true); + } run.reporter.finish(last); - el('metrics').textContent=JSON.stringify(last,null,2); + const display=run.watch?{...last,watch:run.watch.summary()}:last; + el('metrics').textContent=JSON.stringify(display,null,2); el('export').disabled=false; } el('preview').srcObject=null; el('start').disabled=!el('audio-device').value || !el('video-device').value; el('stop').disabled=true; el('devices').disabled=false; - el('paired-signal').disabled=false; - review.lock(false); + el('observation-mode').disabled=false;modeControls(); el('status').textContent=`Stopped · ${reason}`; } el('devices').onclick=async()=>{ @@ -68,14 +90,20 @@ el('start').onclick=async()=>{ el('status').textContent='This browser lacks the required observation APIs.';return; } const ticket=++generation; - const run={started:performance.now(),video:new VideoMetrics(),audio:null,settings:{},pairedEnabled:el('paired-signal').checked}; + const watchEnabled=el('observation-mode').value==='watch'; + const run={started:performance.now(),video:new VideoMetrics(),audio:null,audioReceivedAt:null, + settings:{},watchEnabled,pairedEnabled:!watchEnabled&&el('paired-signal').checked}; current=run; last=null; reportingRun=run; - run.reporter=new RunReporter(message=>{ + run.reporter=watchEnabled?{ + observe(){},finish(){if(reportingRun===run)el('report-status').textContent='Local watch · Save metrics before closing';}, + }:new RunReporter(message=>{ if(reportingRun===run)el('report-status').textContent=message; }); + if(watchEnabled){review.discard();el('report-status').textContent='Local watch · metrics are not uploaded';} el('start').disabled=true; el('devices').disabled=true; el('stop').disabled=false; el('export').disabled=true; el('paired-signal').disabled=true; + el('observation-mode').disabled=true; review.lock(true); el('status').textContent='Starting…'; run.setupTimer=setTimeout(()=>{if(current===run)stop('setup timeout');},20000); @@ -95,10 +123,13 @@ el('start').onclick=async()=>{ await run.context.audioWorklet.addModule(run.pairedEnabled?'paired-worklet.js':'activity-worklet.js'); if(generation!==ticket)return; if(run.pairedEnabled)run.paired=new PairedObserver(run.context,el('preview')); - run.node=new AudioWorkletNode(run.context,run.pairedEnabled?'av-paired-signals':'av-activity'); + run.node=new AudioWorkletNode(run.context,run.pairedEnabled?'av-paired-signals':'av-activity', + {processorOptions:{continuityWatch:watchEnabled}}); run.node.onprocessorerror=()=>{if(current===run)stop('audio observer error');}; run.node.port.onmessage=({data})=>{ if(current!==run)return; + if(watchEnabled&&data.watchStopped===true){stop('audio observer deadline or lost owner');return;} + run.audioReceivedAt=performance.now(); if(run.paired){run.audio=data.activity;run.paired.audioMessage(data);}else run.audio=data; run.node.port.postMessage('ack'); if(run.observing&&run.paired&&!run.signalStarted) { @@ -110,30 +141,36 @@ el('start').onclick=async()=>{ el('status').textContent='Measuring known A/V signals · 120 seconds'; } } - const sample=receipt(run.video,run.audio,run.settings,performance.now()-run.started,'observing'); - if(run.paired)sample.paired=run.paired.summary(); - el('metrics').textContent=JSON.stringify(sample,null,2); - if(run.observing)run.reporter.observe(sample); + update(run); }; run.source=run.context.createMediaStreamSource(new MediaStream(stream.getAudioTracks())); run.source.connect(run.node);run.node.connect(run.context.destination); await run.context.resume(); if(generation!==ticket)return; + if(run.context.state!=='running')throw new Error('audio context did not start'); el('preview').srcObject=stream; await el('preview').play(); if(generation!==ticket)return; clearTimeout(run.setupTimer); run.started=performance.now(); + if(watchEnabled)run.watch=new ContinuityWatch(); run.observing=true; - run.recording=review.begin(stream); + // Reporting and freshness cannot depend on the audio path still working. + run.updateTimer=setInterval(()=>update(run),1000); + run.recording=watchEnabled?false:review.begin(stream); function frame(now,metadata) { if(current!==run)return; - run.video.observe(metadata.expectedDisplayTime,metadata.presentedFrames); - run.paired?.frame(metadata.expectedDisplayTime); - run.callback=el('preview').requestVideoFrameCallback(frame); + try { + run.video.observe(metadata.expectedDisplayTime,metadata.presentedFrames); + run.paired?.frame(metadata.expectedDisplayTime); + run.callback=el('preview').requestVideoFrameCallback(frame); + } catch {if(current===run)stop('video observer error');} } run.callback=el('preview').requestVideoFrameCallback(frame); - if(run.paired) { + if(watchEnabled) { + run.timer=setTimeout(()=>{if(current===run)stop('10 hours observed');},WATCH_MS); + el('status').textContent='Watching continuity · 10-hour limit · local metrics only'; + } else if(run.paired) { run.signalTimer=setTimeout(()=>{if(current===run)stop('test signal timeout');},60000); el('status').textContent='Waiting for known A/V signals · start the source now'; } else { @@ -147,7 +184,8 @@ el('stop').onclick=()=>stop(); el('export').onclick=()=>{ if(!last)return; const url=URL.createObjectURL(new Blob([JSON.stringify(last,null,2)],{type:'application/json'})); - const link=document.createElement('a');link.href=url;link.download='av-test-metrics.json';link.click(); + const link=document.createElement('a');link.href=url; + link.download=last.watch?'av-continuity-watch.json':'av-test-metrics.json';link.click(); setTimeout(()=>URL.revokeObjectURL(url),1000); }; window.addEventListener('pagehide',()=>{stop('page closed');review.discard();}); diff --git a/services/bstein-dev-home/av-observer/paired-observer.js b/services/bstein-dev-home/av-observer/paired-observer.js index 530fe633..33741d6e 100644 --- a/services/bstein-dev-home/av-observer/paired-observer.js +++ b/services/bstein-dev-home/av-observer/paired-observer.js @@ -7,8 +7,9 @@ import {VideoContinuity} from './continuity-signals.js'; // https://www.w3.org/TR/webaudio/#dom-baseaudiocontext-currenttime // https://wicg.github.io/video-rvfc/ export class PairedObserver { - constructor(context,video) { + constructor(context,video,now=()=>performance.now()) { this.context=context;this.video=video;this.audio=[];this.frames=[]; + this.now=now;this.pilotReceivedAt=null; this.edges=new PulseEdges();this.previous=null;this.overflow=0;this.clockInvalid=false; this.canvas=document.createElement('canvas');this.canvas.width=16;this.canvas.height=16; this.drawing=this.canvas.getContext('2d',{willReadFrequently:true}); @@ -21,7 +22,8 @@ export class PairedObserver { if(!Array.isArray(data.signals)||data.signals.length>32){this.clockInvalid=true;return;} this.overflow=Math.max(this.overflow,data.overflow??0); this.pilot=data.pilot??null; - const before=performance.now(),clock=this.context.currentTime*1000,after=performance.now(); + const before=this.now(),clock=this.context.currentTime*1000,after=this.now(); + this.pilotReceivedAt=after; if(this.context.state!=='running'||after-before>5||!Number.isFinite(clock)){this.clockInvalid=true;return;} const offset=(before+after)/2-clock; if(this.offset!==null&&Math.abs(offset-this.offset)>25)this.clockInvalid=true; @@ -63,7 +65,21 @@ export class PairedObserver { if(this.clockInvalid||this.overflow)result.estimatedMedianSignedMs=null; return {...result,audioSignals:this.audio.length,videoSignals:this.frames.length, overflow:this.overflow,clockInvalid:this.clockInvalid, - continuity:this.continuity.summary(),pilot:this.pilot, + continuity:this.continuity.summary(),pilot:this.pilotSummary(), note:'Coded signal / browser processing-time estimate only; input-clock calibration and source age are unmeasured.'}; } + pilotSummary() { + const recent=this.pilot?.recent; + if(!recent)return this.pilot; + const elapsed=this.now()-this.pilotReceivedAt; + const contextAge=this.context.currentTime*1000-recent.endMs; + const valid=this.pilotReceivedAt!==null&&recent.endMs!==null&& + Number.isFinite(elapsed)&&elapsed>=0&&Number.isFinite(contextAge)&&contextAge>=0&& + Number.isFinite(recent.observationAgeMs)&&recent.observationAgeMs>=0; + // The worklet publishes at 1 Hz. Age the cached observation even if + // messages or the audio clock stop; missing callbacks are not silence. + const age=valid?Math.max(contextAge,recent.observationAgeMs+elapsed):null; + return {...this.pilot,recent:{...recent,observationAgeMs:age, + coverageComplete:recent.coverageComplete===true&&!this.clockInvalid&&age!==null&&age<=2000}}; + } } diff --git a/services/bstein-dev-home/av-observer/paired-worklet.js b/services/bstein-dev-home/av-observer/paired-worklet.js index 28573302..1798058f 100644 --- a/services/bstein-dev-home/av-observer/paired-worklet.js +++ b/services/bstein-dev-home/av-observer/paired-worklet.js @@ -11,7 +11,8 @@ class PairedSignals extends AudioWorkletProcessor { } process(inputs,outputs) { if(this.stopped||this.frames>=sampleRate*185)return false; - this.frames+=outputs[0]?.[0]?.length??128; + const block=outputs[0]?.[0]?.length??128; + this.frames+=block; const samples=inputs[0]?.[0]; if(samples) { this.pilot.process(samples,currentFrame); @@ -23,7 +24,8 @@ class PairedSignals extends AudioWorkletProcessor { if(this.credit&&this.frames>=this.next) { this.port.postMessage({activity:{inputFrames:this.inputFrames,processedFrames:this.frames, exactZeroFrames:this.zeros,rms:this.inputFrames?Math.sqrt(this.energy/this.inputFrames):0, - peak:this.peak,sampleRate},signals:this.events,overflow:this.overflow,pilot:this.pilot.summary()}); + peak:this.peak,sampleRate},signals:this.events,overflow:this.overflow, + pilot:this.pilot.summary((currentFrame+block)/sampleRate*1000)}); this.events=[];this.credit=false;this.next=this.frames+sampleRate; } // Browser initializes every output block to zero. No mic monitoring. diff --git a/services/bstein-dev-home/av-observer/pilot-window.js b/services/bstein-dev-home/av-observer/pilot-window.js new file mode 100644 index 00000000..f0e497ab --- /dev/null +++ b/services/bstein-dev-home/av-observer/pilot-window.js @@ -0,0 +1,63 @@ +// Recent known-pilot observations. Fixed memory; no audio or content retained. +const CAPACITY = 1000; + +export class PilotWindow { + constructor(windowMs) { + if (!Number.isFinite(windowMs) || windowMs <= 0 || windowMs > 100) { + throw new RangeError('pilot observation duration'); + } + this.windowMs = windowMs; + this.history = new Uint8Array(CAPACITY); + this.next = 0; + this.count = 0; + this.lastEndMs = null; + this.interruptions = 0; + } + + interrupt() { + this.next = 0; + this.count = 0; + this.lastEndMs = null; + this.interruptions = Math.min(Number.MAX_SAFE_INTEGER, this.interruptions + 1); + } + + observe(present, endMs) { + if (typeof present !== 'boolean' || !Number.isFinite(endMs) || endMs < this.windowMs) { + this.interrupt(); + return; + } + // The detector emits equal-sized, contiguous sample windows. Missing + // callbacks are unknown coverage, not evidence of missing pilot PCM. + if (this.lastEndMs !== null && + Math.abs(endMs - this.lastEndMs - this.windowMs) > 0.001) { + this.interrupt(); + } + this.history[this.next] = present ? 1 : 0; + this.next = (this.next + 1) % CAPACITY; + this.count = Math.min(CAPACITY, this.count + 1); + this.lastEndMs = endMs; + } + + summary(nowMs) { + let missing = 0, run = 0, longest = 0; + const start = (this.next + CAPACITY - this.count) % CAPACITY; + for (let n = 0; n < this.count; n++) { + if (this.history[(start + n) % CAPACITY]) run = 0; + else { missing++; run++; longest = Math.max(longest, run); } + } + const age = this.lastEndMs !== null && Number.isFinite(nowMs) && nowMs >= this.lastEndMs + ? nowMs - this.lastEndMs : null; + return { + windowLimitMs: CAPACITY * this.windowMs, + observedWindows: this.count, + observedMs: this.count * this.windowMs, + endMs: this.lastEndMs, + observationAgeMs: age, + coverageComplete: this.count === CAPACITY && age !== null && age <= 2 * this.windowMs, + absentFraction: this.count ? missing / this.count : null, + absenceMaxMs: this.count ? longest * this.windowMs : null, + trailingAbsenceMs: this.count ? run * this.windowMs : null, + interruptions: this.interruptions, + }; + } +} diff --git a/services/bstein-dev-home/av-observer/report-service/server.py b/services/bstein-dev-home/av-observer/report-service/server.py index 58d546bf..dbbe4dba 100644 --- a/services/bstein-dev-home/av-observer/report-service/server.py +++ b/services/bstein-dev-home/av-observer/report-service/server.py @@ -4,6 +4,7 @@ from http.server import BaseHTTPRequestHandler, HTTPServer from socketserver import ThreadingMixIn import json import os +import socket import threading from receipts import ReceiptStore, Refused @@ -28,9 +29,6 @@ class Handler(BaseHTTPRequestHandler): self.wfile.write(body) def do_GET(self): - """Answer the internal readiness check without reading receipt state.""" - if self.path == "/healthz": - return self.reply(204) self.reply(405) do_HEAD = do_GET @@ -71,8 +69,9 @@ class Handler(BaseHTTPRequestHandler): class Server(ThreadingMixIn, HTTPServer): daemon_threads = True - def __init__(self, address, origin, store): + def __init__(self, address, origin, store, request_timeout=3): self.origin, self.store = origin, store + self.request_timeout = request_timeout self.workers = threading.BoundedSemaphore(4) super().__init__(address, Handler) @@ -92,9 +91,20 @@ class Server(ThreadingMixIn, HTTPServer): raise def process_request_thread(self, request, client_address): + def expire(): + try: + request.shutdown(socket.SHUT_RDWR) + except OSError: + pass # The request may already have completed. + # Socket timeouts alone reset after every byte. Bound the entire + # header/body read so a trickling peer cannot occupy a worker forever. + deadline = threading.Timer(self.request_timeout, expire) + deadline.daemon = True + deadline.start() try: super().process_request_thread(request, client_address) finally: + deadline.cancel() self.workers.release() def handle_error(self, *_args): diff --git a/services/bstein-dev-home/av-observer/report-service/validation.py b/services/bstein-dev-home/av-observer/report-service/validation.py index 8be3da60..f3152396 100644 --- a/services/bstein-dev-home/av-observer/report-service/validation.py +++ b/services/bstein-dev-home/av-observer/report-service/validation.py @@ -19,7 +19,21 @@ BOUNDS = { "pilot_windows": 18_000, "pilot_gap_max_ms": 180_000, "pilot_missing_fraction": 1, "pilot_trailing_gap_ms": 180_000, "pilot_discontinuities": 100_000_000, + "pilot_recent_windows": 1000, "pilot_recent_observed_ms": 100_000, + "pilot_recent_age_ms": 180_000, "pilot_recent_absent_fraction": 1, + "pilot_recent_absence_max_ms": 100_000, "pilot_recent_trailing_ms": 100_000, + "pilot_recent_interruptions": 100_000_000, "pilot_recent_coverage_complete": 1, } +INTEGER_METRICS = frozenset(( + "video_width", "video_height", "audio_rate", "audio_channels", "audio_context_rate", + "video_callbacks", "video_unobserved", "video_resets", "audio_input_frames", + "audio_processed_frames", "audio_zero_frames", "paired_audio_signals", + "paired_video_signals", "paired_matches", "paired_ambiguous", "paired_overflow", + "marker_frames", "marker_bad", "marker_ambiguous", "pilot_windows", "pilot_discontinuities", + "pilot_recent_windows", "pilot_recent_interruptions", "pilot_recent_coverage_complete", +)) +RECENT_METRICS = frozenset(key for key in BOUNDS if key.startswith("pilot_recent_")) +LEGACY_METRICS = BOUNDS.keys() - RECENT_METRICS def no_duplicate_fields(pairs): @@ -43,13 +57,17 @@ def validate_report(data): if type(state) is not str or state not in STATES: raise ValueError("state") metrics = data["metrics"] - if type(metrics) is not dict or metrics.keys() != BOUNDS.keys(): + if type(metrics) is not dict or metrics.keys() not in (BOUNDS.keys(), LEGACY_METRICS): raise ValueError("metric fields") clean = {} for name, limit in BOUNDS.items(): - value = metrics[name] + # A cached older page or retained receipt has no recent window. Only + # its exact previous schema is accepted; missing evidence is not zero. + value = metrics.get(name) if value is not None and (type(value) not in (int, float) or not 0 <= value <= limit or not math.isfinite(value)): raise ValueError("metric value") + if value is not None and name in INTEGER_METRICS and type(value) is not int: + raise ValueError("integer metric required") clean[name] = value if clean["elapsed_ms"] is None: raise ValueError("elapsed missing") diff --git a/services/bstein-dev-home/av-observer/reporting.js b/services/bstein-dev-home/av-observer/reporting.js index 4693f82e..7658326a 100644 --- a/services/bstein-dev-home/av-observer/reporting.js +++ b/services/bstein-dev-home/av-observer/reporting.js @@ -3,7 +3,7 @@ export function reportMetrics(receipt) { const finite=value=>typeof value==='number'&&Number.isFinite(value)&&value>=0 ? value : null; const v=receipt.video??{},a=receipt.audio??{},s=receipt.settings??{},p=receipt.paired??{}; - const c=p.continuity??{},pilot=p.pilot??{}; + const c=p.continuity??{},pilot=p.pilot??{},recent=pilot.recent??{}; const estimate=typeof p.estimatedMedianSignedMs==='number'&&Number.isFinite(p.estimatedMedianSignedMs)?p.estimatedMedianSignedMs:null; return Object.fromEntries(Object.entries({ elapsed_ms:receipt.elapsedMs, @@ -23,6 +23,11 @@ export function reportMetrics(receipt) { pilot_windows:pilot.windows,pilot_gap_max_ms:pilot.gapMaxMs, pilot_missing_fraction:pilot.missingFraction,pilot_trailing_gap_ms:pilot.trailingGapMs, pilot_discontinuities:pilot.discontinuities, + pilot_recent_windows:recent.observedWindows,pilot_recent_observed_ms:recent.observedMs, + pilot_recent_age_ms:recent.observationAgeMs,pilot_recent_absent_fraction:recent.absentFraction, + pilot_recent_absence_max_ms:recent.absenceMaxMs,pilot_recent_trailing_ms:recent.trailingAbsenceMs, + pilot_recent_interruptions:recent.interruptions, + pilot_recent_coverage_complete:typeof recent.coverageComplete==='boolean'?Number(recent.coverageComplete):null, }).map(([key,value])=>[key,finite(value)])); }