Validate complete recording duration and drain FFmpeg source output

tests/contracts/tooling/active-recording-utils.test.mjs

@@ -0,0 +1,50 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import {EventEmitter,once} from 'node:events';
+import {PassThrough} from 'node:stream';
+import {pipeRecordingSource,validateRecordingDurations} from '../../performance/active-recording-utils.mjs';
+
+const row=(stream,duration)=>({stream,probe:{format:{duration}}});
+
+test('full and short tail segments count toward each camera duration',()=>{
+ assert.deepEqual(validateRecordingDurations([row('Load0','60'),row('Load0','6'),row('Load1','64')],2,65),{Load0:66,Load1:64});
+});
+test('a long recording for one camera cannot hide another truncated camera',()=>{
+ assert.throws(()=>validateRecordingDurations([row('Load0',120),row('Load1',3)],2,60),/shorter than expected for Load1/);
+});
+test('missing streams and invalid segment durations fail',()=>{
+ assert.throws(()=>validateRecordingDurations([row('Load0',60)],2,60),/Missing finalized/);
+ for(const value of [undefined,null,'',0,-1,'N/A',Infinity]){
+  assert.throws(()=>validateRecordingDurations([row('Load0',value)],1,10),/invalid recording duration/);
+ }
+ assert.throws(()=>validateRecordingDurations([row('Other',60)],1,60),/Unexpected recording stream/);
+});
+function source(){
+ const child=new EventEmitter();child.stdout=new PassThrough();child.stderr=new PassThrough();
+ child.signals=[];child.kill=signal=>child.signals.push(signal);
+ const res=new PassThrough(),failures=[],chunks=[];
+ res.on('data',chunk=>chunks.push(chunk));res.on('error',()=>{});
+ pipeRecordingSource(child,res,error=>failures.push(error));
+ return {child,res,failures,chunks};
+}
+test('process exit does not truncate the final stdout bytes',async()=>{
+ const {child,res,chunks,failures}=source();
+ child.stdout.write('first');child.emit('exit',0,null);
+ assert.equal(res.writableEnded,false);
+ const ended=once(res,'end');child.stdout.end('tail');await ended;
+ child.emit('close',0,null);
+ assert.equal(Buffer.concat(chunks).toString(),'firsttail');
+ assert.equal(failures.length,1); // A looping source should not end on its own.
+});
+test('intentional client disconnect terminates source without failure',async()=>{
+ const {child,res,failures}=source();
+ const closed=once(res,'close');res.destroy();await closed;
+ child.emit('close',null,'SIGTERM');
+ assert.deepEqual(child.signals,['SIGTERM']);assert.deepEqual(failures,[]);
+});
+test('spawn failure and unexpected nonzero exit are recorded',()=>{
+ const first=source();first.child.emit('error',new Error('spawn ENOENT'));
+ assert.match(first.failures[0],/ENOENT/);
+ const second=source();second.child.stderr.write('decoder failed');second.child.emit('close',1,null);
+ assert.match(second.failures[0],/decoder failed/);
+});

tests/performance/active-recording-utils.mjs

@@ -0,0 +1,30 @@
+/** Validate footage across segment boundaries, including the short final segment. */
+export function validateRecordingDurations(recordings, streamCount, seconds, toleranceSeconds=5) {
+ const totals=new Map(Array.from({length:streamCount},(_,i)=>[`Load${i}`,0]));
+ for(const row of recordings){
+  if(!totals.has(row.stream))throw new Error(`Unexpected recording stream: ${row.stream}`);
+  const duration=Number(row.probe?.format?.duration);
+  if(!Number.isFinite(duration)||duration<=0)throw new Error(`Missing or invalid recording duration for ${row.stream}`);
+  totals.set(row.stream,totals.get(row.stream)+duration);
+ }
+ const minimum=Math.max(0,seconds-toleranceSeconds);
+ for(const [stream,duration] of totals){
+  if(duration===0)throw new Error(`Missing finalized camera footage for ${stream}`);
+  if(duration<minimum)throw new Error(`Recording shorter than expected for ${stream}: ${duration}s < ${minimum}s`);
+ }
+ return Object.fromEntries(totals);
+}
+
+/** Pipe until stdout ends; process exit can precede the final buffered bytes. */
+export function pipeRecordingSource(child,res,onFailure) {
+ let stderr='',clientStopped=false;
+ child.stderr.on('data',chunk=>{stderr=(stderr+chunk).slice(-16384);});
+ child.once('error',error=>{onFailure(String(error));res.destroy(error);});
+ child.once('close',(code,signal)=>{
+  if(!clientStopped)onFailure(`Source exited unexpectedly (code ${code}, signal ${signal}): ${stderr}`);
+ });
+ res.once('close',()=>{
+  if(!res.writableEnded){clientStopped=true;child.kill('SIGTERM');}
+ });
+ child.stdout.pipe(res);
+}

tests/performance/active-recording.md

@@ -15,7 +15,10 @@ FFmpeg processes share the host; their CPU/memory are not included in the native
 process samples. Streams start sequentially before the sampling interval.
 
 After the harness automatically stops recordings at the configured duration, every catalogued file must pass ffprobe and full FFmpeg
-`-xerror` decoding, with footage present for every stream. Ordinary FFmpeg exit
+`-xerror` decoding, with footage present for every stream. Each stream must have at least the requested
+duration minus five seconds across all its segments; a short final segment is valid.
+Missing, zero or non-finite durations fail validation. Source process failures,
+including failures discovered during cleanup, also fail the run. Ordinary FFmpeg exit
 status alone is insufficient: it can report corrupt frames while exiting zero.
 The regression in `mediaLifecycle.api.spec.ts` uses one stream for ten seconds;
 `LIGHTNVR_CONTRACT_BIN` selects the binary as in the other native contract tests.

tests/performance/active-recording.mjs

@@ -2,6 +2,7 @@
 /** Bounded synthetic active-recording measurements on an owned loopback fixture. */
 import {startFixture} from '../contracts/fixture-server.mjs';
 import {trackedDiffSha256} from './git-provenance.mjs';
+import {pipeRecordingSource,validateRecordingDurations} from './active-recording-utils.mjs';
 import {createServer} from 'node:http';
 import {spawn,execFileSync} from 'node:child_process';
 import {mkdir,readFile,readdir,writeFile} from 'node:fs/promises';
@@ -24,6 +25,7 @@ const source=path.join(directory,'source.ts');
 execFileSync('ffmpeg',['-nostdin','-hide_banner','-loglevel','error','-f','lavfi','-i','testsrc2=size=1920x1080:rate=15','-t','5','-an','-c:v','libx264','-threads','2','-preset','ultrafast','-pix_fmt','yuv420p','-g','15','-f','mpegts',source],{timeout:30000});
 report.sourceSha256=hash(await readFile(source));
 report.harnessSha256=hash(await readFile(new URL(import.meta.url)));
+report.validationSha256=hash(await readFile(new URL('./active-recording-utils.mjs',import.meta.url)));
 report.nativeSourceSha256=hash(await readFile('src/video/mp4_segment_recorder.c'));
 const buildDirectory=path.dirname(path.dirname(binary));
 for(const [key,relative] of [['cmakeCache','CMakeCache.txt'],['compileFlags','CMakeFiles/lightnvr.dir/flags.make']]) {
@@ -38,9 +40,7 @@ const http=createServer((req,res)=>{
  res.writeHead(200,{'Content-Type':'video/mp2t'});
  const child=spawn('ffmpeg',['-nostdin','-hide_banner','-loglevel','error','-re','-stream_loop','-1','-i',source,'-c','copy','-f','mpegts','pipe:1'],{stdio:['ignore','pipe','pipe']});
  children.add(child);childExits.push(new Promise(resolve=>child.once('close',()=>{children.delete(child);resolve();})));
- child.stdout.pipe(res);let error='';child.stderr.on('data',chunk=>{error+=chunk;});
- child.once('error',error=>res.destroy(error));child.once('exit',code=>{if(code)report.sourceError=error;res.end();});
- res.on('close',()=>child.kill('SIGTERM'));
+ pipeRecordingSource(child,res,error=>{report.sourceError ||= error;});
 });
 await new Promise(resolve=>http.listen(0,'127.0.0.1',resolve));
 let server;
@@ -78,7 +78,8 @@ try {
   execFileSync('ffmpeg',['-nostdin','-hide_banner','-loglevel','error','-xerror','-i',filename,'-f','null','-'],{timeout:60000});
   report.recordings.push({id:row.id,stream:row.stream,probe});
  }
- if(new Set(report.recordings.map(row=>row.stream)).size!==count)throw new Error('Missing finalized camera footage');
+ report.recordedDurationSeconds=validateRecordingDurations(report.recordings,count,seconds);
+ if(report.sourceError)throw new Error(`Source process failed: ${report.sourceError}`);
  report.status='passed';
 }catch(error){report.status='failed';report.error=String(error);process.exitCode=1;}
 finally{
@@ -87,6 +88,7 @@ finally{
  for(const child of children)child.kill('SIGTERM');
  const killTimer=setTimeout(()=>{for(const child of children)child.kill('SIGKILL');},2000);
  await Promise.all(childExits);clearTimeout(killTimer);
+ if(report.sourceError){report.status='failed';report.error ||= `Source process failed: ${report.sourceError}`;process.exitCode=1;}
  await save();
 }
 console.log(JSON.stringify({status:report.status,output:directory,error:report.error}));