1import assert from 'node:assert/strict';2import test from 'node:test';3import {4 ADAPTIVE_COUNTER_DELTA_MAX,5 AdaptiveFeedbackPublisher,6 formatAdaptiveDecision,7 normalizeAdaptiveDecisionEnvelope,8 normalizeAdaptiveFeedbackSnapshot,9} from './adaptive-feedback.mjs';1011function snapshot(overrides = {}) {12 return {13 frontendQueueDepth: 1,14 frontendQueueCapacity: 4,15 decoderQueueDepth: 1,16 decoderQueueCapacity: 2,17 presenterQueueDepth: 1,18 presenterQueueCapacity: 2,19 transportDroppedTotal: 3,20 frontendDroppedTotal: 4,21 decoderDroppedTotal: 5,22 presenterDroppedTotal: 6,23 transportDeliveryP95Ms: 18.5,24 decodeLatencyP95Ms: 4.5,25 presentationLatencyP95Ms: 24.5,26 resyncActive: false,27 lastSequence: 99,28 ...overrides,29 };30}3132function flushPromises() {33 return new Promise((resolve) => setImmediate(resolve));34}3536test('feedback snapshots reject unsafe values and clamp wire bounds', () => {37 assert.throws(() => normalizeAdaptiveFeedbackSnapshot(snapshot({ decoderQueueDepth: NaN })), /safe integer/);38 assert.throws(() => normalizeAdaptiveFeedbackSnapshot(snapshot({ resyncActive: 1 })), /boolean/);39 const value = normalizeAdaptiveFeedbackSnapshot(snapshot({40 decoderQueueDepth: 99,41 transportDeliveryP95Ms: 1e9,42 }));43 assert.equal(value.decode_queue_depth, 2);44 assert.equal(value.transport_delivery_p95_ms, 60_000);45});4647test('publisher is generation-scoped, one hertz, single-flight, and reports deltas', async () => {48 let now = 0;49 const calls = [];50 const completions = [];51 const publisher = new AdaptiveFeedbackPublisher({52 now: () => now,53 invokeCommand: (command, args) => {54 calls.push({ command, args });55 return new Promise((resolve) => completions.push(resolve));56 },57 });58 publisher.start(7, true);59 assert.equal(publisher.publish(snapshot()), false);60 now = 1000;61 assert.equal(publisher.publish(snapshot()), true);62 assert.equal(publisher.publish(snapshot({ transportDroppedTotal: 10 })), false);63 now = 2000;64 assert.equal(publisher.publish(snapshot({ transportDroppedTotal: 11 })), false);65 assert.equal(calls.length, 1);66 completions.shift()(true);67 await flushPromises();68 assert.equal(publisher.publish(snapshot({ transportDroppedTotal: 12 })), true);69 assert.equal(calls[1].args.generation, 7);70 assert.equal(calls[1].args.report.transport_dropped_delta, 9);71 assert.equal(calls[1].args.report.frontend_dropped_delta, 0);72});7374test('counter deltas saturate and counter resets do not underflow', async () => {75 let now = 0;76 const calls = [];77 const publisher = new AdaptiveFeedbackPublisher({78 now: () => now,79 invokeCommand: async (_command, args) => {80 calls.push(args.report);81 return true;82 },83 });84 publisher.start(1, true);85 publisher.publish(snapshot({86 transportDroppedTotal: 0,87 frontendDroppedTotal: 0,88 decoderDroppedTotal: 0,89 presenterDroppedTotal: 0,90 }));91 now = 1000;92 publisher.publish(snapshot({ transportDroppedTotal: ADAPTIVE_COUNTER_DELTA_MAX + 10 }));93 await flushPromises();94 now = 2000;95 publisher.publish(snapshot({ transportDroppedTotal: 1 }));96 assert.equal(calls[0].transport_dropped_delta, ADAPTIVE_COUNTER_DELTA_MAX);97 assert.equal(calls[1].transport_dropped_delta, 0);98});99100test('stale completion cannot unlock or alter a newer generation', async () => {101 let resolveOld;102 let now = 0;103 const calls = [];104 const publisher = new AdaptiveFeedbackPublisher({105 now: () => now,106 invokeCommand: (_command, args) => {107 calls.push(args.generation);108 if (args.generation === 1) return new Promise((resolve) => { resolveOld = resolve; });109 return Promise.resolve(true);110 },111 });112 publisher.start(1, true);113 publisher.publish(snapshot());114 now = 1000;115 publisher.publish(snapshot());116 publisher.start(2, true);117 publisher.publish(snapshot());118 now = 2000;119 publisher.publish(snapshot());120 resolveOld();121 await flushPromises();122 assert.deepEqual(calls, [1, 2]);123});124125test('false and rejected sends retain cumulative loss until a report is accepted', async () => {126 let now = 0;127 const reports = [];128 const outcomes = [129 () => false,130 () => Promise.reject(new Error('closed')),131 () => true,132 () => true,133 ];134 const originalWarn = console.warn;135 console.warn = () => {};136 try {137 const publisher = new AdaptiveFeedbackPublisher({138 now: () => now,139 invokeCommand: (_command, args) => {140 reports.push(args.report);141 return outcomes.shift()();142 },143 });144 publisher.start(1, true);145 publisher.publish(snapshot({146 transportDroppedTotal: 0,147 frontendDroppedTotal: 0,148 decoderDroppedTotal: 0,149 presenterDroppedTotal: 0,150 }));151 now = 1000;152 publisher.publish(snapshot({ transportDroppedTotal: 3 }));153 await flushPromises();154 now = 2000;155 publisher.publish(snapshot({ transportDroppedTotal: 7 }));156 await flushPromises();157 now = 3000;158 publisher.publish(snapshot({ transportDroppedTotal: 9 }));159 await flushPromises();160 now = 4000;161 publisher.publish(snapshot({ transportDroppedTotal: 10 }));162 assert.deepEqual(163 reports.map((report) => report.transport_dropped_delta),164 [3, 7, 9, 1],165 );166 assert.deepEqual(167 reports.map((report) => report.interval_ms),168 [1000, 2000, 3000, 1000],169 );170 } finally {171 console.warn = originalWarn;172 }173});174175test('a synchronous invoke throw clears single-flight state and permits retry', async () => {176 let attempts = 0;177 let now = 0;178 const originalWarn = console.warn;179 console.warn = () => {};180 try {181 const publisher = new AdaptiveFeedbackPublisher({182 now: () => now,183 invokeCommand: () => {184 attempts++;185 if (attempts === 1) throw new Error('synchronous bridge failure');186 return Promise.resolve(true);187 },188 });189 publisher.start(1, true);190 publisher.publish(snapshot());191 now = 1000;192 assert.equal(publisher.publish(snapshot()), false);193 assert.equal(publisher.inFlight, false);194 assert.equal(publisher.publish(snapshot()), true);195 await flushPromises();196 assert.equal(publisher.inFlight, false);197 assert.equal(attempts, 2);198 } finally {199 console.warn = originalWarn;200 }201});202203test('publisher measures and bounds the actual baseline interval', async () => {204 let now = 0;205 const reports = [];206 const publisher = new AdaptiveFeedbackPublisher({207 now: () => now,208 invokeCommand: async (_command, args) => {209 reports.push(args.report);210 return true;211 },212 });213 publisher.start(1, true);214 publisher.publish(snapshot());215 now = 249;216 assert.equal(publisher.publish(snapshot()), false);217 now = 1_275;218 assert.equal(publisher.publish(snapshot()), true);219 await flushPromises();220 now = 8_000;221 assert.equal(publisher.publish(snapshot({ transportDroppedTotal: 7 })), true);222 assert.deepEqual(reports.map((report) => report.interval_ms), [1_275, 5_000]);223});224225test('decision diagnostics reject stale generations and say advisory not applied', () => {226 assert.equal(normalizeAdaptiveDecisionEnvelope({ generation: 8, decision: {} }, 7), null);227 const decision = normalizeAdaptiveDecisionEnvelope({228 generation: 7,229 decision: {230 decision_id: 4,231 report_id: 3,232 target_kbps: 8000,233 floor_kbps: 4000,234 ceiling_kbps: 20000,235 state: 'hold',236 reasons: ['clean-recovery'],237 applied: false,238 contribution: 'clean',239 recovery_state: 'active',240 scope: 'local-viewer',241 },242 }, 7);243 assert.match(formatAdaptiveDecision(decision, true), /advisory only \(not applied\)/);244 assert.match(formatAdaptiveDecision(decision, true), /local clean ยท recovery active/);245 assert.equal(decision.scope, 'local-viewer');246 assert.equal(formatAdaptiveDecision(null, false), 'unavailable');247});