Orchestration e custo

const assert = require('node:assert/strict');
class BoundedQueue {
  constructor(capacity) {
    this.capacity=capacity; this.items=[]; this.readers=[];
    this.producers=[]; this.closed=false; this.peak=0; this.backpressure=0;
  }
  async put(item) {
    while (!this.closed && !this.readers.length && this.items.length===this.capacity) {
      this.backpressure++;
      await new Promise((resolve,reject)=>this.producers.push({resolve,reject}));
    }
    if(this.closed) throw new Error('queue-closed');
    if(this.readers.length) this.readers.shift()(item);
    else {this.items.push(item); this.peak=Math.max(this.peak,this.items.length);}
  }
  async take() {
    if(this.items.length) {
      const item=this.items.shift(); this.producers.shift()?.resolve(); return item;
    }
    if(this.closed) return null;
    return new Promise(resolve=>this.readers.push(resolve));
  }
  close(discard=false) {
    this.closed=true;
    if(discard) this.items=[];
    for(const r of this.readers.splice(0)) r(null);
    for(const p of this.producers.splice(0)) p.reject(new Error('queue-closed'));
  }
}
function sleep(ms,signal) {
  return new Promise((resolve,reject)=>{
    const abort=()=>{clearTimeout(timer);signal.removeEventListener('abort',abort);
      const error=new Error('cancelled');error.cancelled=true;reject(error);};
    const timer=setTimeout(()=>{signal.removeEventListener('abort',abort);resolve();},ms);
    signal.addEventListener('abort',abort,{once:true});if(signal.aborted)abort();
  });
}
class Budget {
  constructor(limit){this.limit=limit;this.reserved=0;this.settled=0;this.minimum=limit;}
  available(){return this.limit-this.reserved-this.settled;}
  reserve(maximum){
    if(maximum>this.available())return null;
    // Operação síncrona e indivisível somente neste processo JavaScript.
    this.reserved+=maximum;this.minimum=Math.min(this.minimum,this.available());
    let done=false;
    return actual=>{
      assert.ok(!done && actual>=0 && actual<=maximum);done=true;
      this.reserved-=maximum;this.settled+=actual;
      assert.ok(this.available()>=0 && this.reserved>=0);
    };
  }
}
async function run(tasks,{limit,deadlineMs=1000,externalSignal}={}){
  const queue=new BoundedQueue(1),budget=new Budget(limit),controller=new AbortController();
  const signal=controller.signal,events=[],results={},effects={};
  const synthesisReservation=budget.reserve(1);
  let synthesisSettled=false;
  let active=0,peakActive=0;
  const cancel=()=>{controller.abort();queue.close(true);};
  const deadline=setTimeout(cancel,deadlineMs);
  externalSignal?.addEventListener('abort',cancel,{once:true});
  if(externalSignal?.aborted)cancel();
  async function worker(workerId){
    for(let task; (task=await queue.take())!==null;){
      if(signal.aborted)break;
      for(let attempt=1;attempt<=2;attempt++){
        if(signal.aborted){results[task.id]={status:'cancelled'};break;}
        const settle=budget.reserve(task.cost);
        if(!settle){results[task.id]={status:'budget'};break;}
        let actual=1,retry=false;
        active++;peakActive=Math.max(peakActive,active);
        events.push({task:task.id,workerId,attempt,event:'start',active});
        try{
          await sleep(task.ms,signal);
          if(task.mode==='denied')throw Object.assign(new Error('denied'),{permanent:true});
          if(task.mode==='transient' && attempt===1)
            throw Object.assign(new Error('temporary'),{transient:true});
          if(task.effect){
            effects[task.id]=(effects[task.id]||0)+1;
            if(task.mode==='ambiguous') {
              actual=task.cost;
              throw Object.assign(new Error('response-lost-after-effect'),{transient:true});
            }
          }
          actual=task.cost;results[task.id]={status:'ok',attempt};
        }catch(error){
          retry=error.transient && !task.effect && !signal.aborted && attempt<2;
          results[task.id]={status:error.cancelled?'cancelled':retry?'retry':'failed',reason:error.message};
        }finally{
          active--;settle(actual);
          events.push({task:task.id,workerId,attempt,event:'settle',actual,
            reserved:budget.reserved,settled:budget.settled,available:budget.available()});
        }
        if(!retry)break;
        // Espera limitada também respeita o deadline; não ocupa uma reserva financeira.
        try{await sleep(2,signal);}catch{results[task.id]={status:'cancelled'};break;}
      }
    }
  }
  const workers=[worker(1),worker(2)];
  const producer=(async()=>{
    try{for(const task of tasks)await queue.put(task);}
    catch(error){if(!signal.aborted)throw error;}
    finally{queue.close(signal.aborted);}
  })();
  try{
    await Promise.all([producer,...workers]);
    for(const task of tasks)results[task.id]??={status:signal.aborted?'not-started-cancelled':'not-started'};
    // A reserva inicial protege a síntese; cancelar libera-a sem consumo.
    const synthesize=!signal.aborted && Object.values(results).some(r=>r.status==='ok');
    if(synthesisReservation){
      synthesisReservation(synthesize?1:0);synthesisSettled=true;
      if(synthesize)events.push({event:'synthesis',actual:1});
    }
    return {results,effects,events,peakActive,queuePeak:queue.peak,
      backpressure:queue.backpressure,cancelled:signal.aborted,
      budget:{limit,reserved:budget.reserved,settled:budget.settled,
        available:budget.available(),minimum:budget.minimum}};
  }finally{
    clearTimeout(deadline);externalSignal?.removeEventListener('abort',cancel);
    if(synthesisReservation && !synthesisSettled)synthesisReservation(0);
  }
}
(async()=>{
  const tasks=[{id:'A',ms:10,cost:3},{id:'B',ms:12,cost:4,mode:'transient'},
    {id:'C',ms:5,cost:2},{id:'D',ms:5,cost:2,mode:'denied'},
    {id:'E',ms:5,cost:2,effect:true,mode:'ambiguous'}];
  const complete=await run(tasks,{limit:20});
  assert.equal(complete.peakActive,2);assert.ok(complete.queuePeak<=1);
  assert.ok(complete.backpressure>0);
  assert.equal(complete.events.filter(e=>e.task==='A'&&e.event==='start').length,1);
  assert.equal(complete.events.filter(e=>e.task==='B'&&e.event==='start').length,2);
  assert.equal(complete.events.filter(e=>e.task==='D'&&e.event==='start').length,1);
  assert.equal(complete.events.filter(e=>e.task==='E'&&e.event==='start').length,1);
  assert.equal(complete.effects.E,1);assert.equal(complete.budget.settled,14);
  const low=await run(tasks,{limit:7});
  const high=await run(tasks,{limit:13});
  const slow=Array.from({length:5},(_,i)=>({id:'slow-'+i,ms:100,cost:3}));
  const expired=await run(slow,{limit:10,deadlineMs:15});
  assert.equal(expired.cancelled,true);assert.equal(expired.budget.settled,2);
  assert.equal(expired.budget.reserved,0);assert.deepEqual(expired.effects,{});
  assert.ok(Object.values(expired.results).some(r=>r.status==='not-started-cancelled'));
  const manualController=new AbortController();manualController.abort();
  const manual=await run(slow,{limit:10,externalSignal:manualController.signal});
  assert.equal(manual.budget.settled,0);assert.equal(manual.peakActive,0);
  for(const report of [complete,low,high,expired,manual]){
    assert.ok(report.budget.minimum>=0 && report.budget.available>=0);
    assert.equal(report.budget.reserved,0);
  }
  console.log(JSON.stringify({complete,low,high,expired,manual},null,2));
})().catch(error=>{console.error(error);process.exitCode=1;});