Documentazione di CapDAG

Esecuzione

Come viene eseguito un piano: ambiti e unità, pianificazione, mappe, valori assenti, politica dei passi, streaming e arresto di un’esecuzione.

In questa sezione

Come viene eseguito un MachinePlan: che cosa viene eseguito quando, che cosa viene inviato a un passo e che cosa se ne raccoglie.

execute_plan

flowchart TD
    MP["MachinePlan"] --> VAL["1. Convalida il piano<br/>e ciò che l'esecuzione lega"]
    VAL --> SC["2. Ambiti: la macchina,<br/>e il corpo di ogni mappa"]
    SC --> UN["3. Unità di un ambito:<br/>catene di passi, e mappe"]
    UN --> RUN["4. Avvia ogni unità quando<br/>tutto ciò che legge è stabilito"]
    RUN --> SEG["5. Una catena è un segmento,<br/>eseguito dall'EngineRuntime"]
    SEG --> RES["6. PipelineResult:<br/>terminali, rapporti dei passi, verdetto"]

execute_plan(plan, runtime, inputs, input_is_sequence, arguments, callbacks) esegue un piano:

  1. Convalida. Il piano è vincolato ai propri invarianti (MachinePlan::validate), e l’esecuzione al piano: a ogni InputSlot viene dato esattamente un valore, e non viene dato nulla che il piano non prenda.
  2. Ambiti. La macchina è un ambito, e lo è anche il corpo di ogni ForEach (Orchestratore).
  3. Unità. I passi di un ambito sono divisi in catene (Esecuzione in pipeline); ogni mappa direttamente dentro l’ambito è un’unità a sé.
  4. Pianificazione. Un’unità si avvia non appena tutto ciò che legge è stabilito (Pianificazione).
  5. Esecuzione. Una catena viene convertita in un ResolvedGraph ed eseguita come un segmento dall’EngineRuntime: le cartucce vengono individuate e ospitate, la richiesta di ogni passo viene aperta, alimentata e raccolta (La richiesta di un passo). Una mappa esegue il suo corpo una volta per elemento (Mappe).
  6. Risultato. Un PipelineResult: un PlanTerminal per ogni nodo Output, l’esito di ogni esecuzione di ogni mappa, uno StepReport per ogni passo che non si è semplicemente completato, e il verdetto dell’esecuzione.

L’EngineRuntime è l’unica cosa che differisce fra gli host: come vengono ospitate le cartucce, se l’output terminale viene reso persistente, e che cosa fa un’esecuzione fallita di una mappa (Mappe). Pianificazione, mappe e assemblaggio del risultato sono uguali ovunque.

Sorgente: capdag-rs/src/orchestrator/execute_plan.rs, executor.rs.

Individuazione delle cartucce

Risoluzione della cartuccia per ogni cap URN del grafo:

  1. Cartucce di sviluppo (override locale): controlla dev_cartridges (una mappa di percorsi di binari locali) usando la corrispondenza is_dispatchable. Questo permette agli sviluppatori di provare build locali senza pubblicarle.
  2. Registro delle cartucce (remoto): in mancanza, ricorre al registro delle cartucce per le cartucce pubblicate. Il registro delle cartucce è versionato: un binario incorpora una CARTRIDGE_REGISTRY_VERSION (attualmente 2), e l’URL del manifest pubblicato porta un segmento di prefisso di versione (.../v2/manifest), mentre il semplice /manifest indica lo stato v0 congelato. Su disco, le cartucce si trovano in {registry_slug}/v{cartridge_registry_version}/{channel}/{name}/{version}/, e l’individuazione fissa la directory v{version} alla versione incorporata nell’host esattamente come fissa il canale, quindi una cartuccia v2 è invisibile a un host v1. Il registry_slug è una trasformazione, sicura per i percorsi, dell’authority dell’URL del registro (host, più :port): la sottostringa dopo :// fino al primo /, ? o #, convertita in minuscolo ASCII, con ogni carattere al di fuori di [a-z0-9.-] sostituito da - (un URL di registro nullo diventa il letterale dev); non è un hash, e dipende solo dall’authority, quindi tutte le versioni di uno stesso registro condividono uno slug. Un registro delle cartucce viene acquisito solo se il fabricRegistryUrl del suo manifest corrisponde al registro fabric incorporato nel client, così ogni registro usato da un client condivide un unico fabric e mantiene le cap confrontabili. La cache locale del fabric è indicizzata per versione del registro, quindi una ripubblicazione in staging della stessa versione su un canale mutabile può servire byte non aggiornati; resolve --no-cache e capdag cache clear|refresh la aggirano o la rinnovano.
  3. Download e verifica: se il binario non è in cache localmente, lo scarica e ne verifica l’hash SHA256.
  4. Registrazione: aggiunge il binario della cartuccia al CartridgeHostRuntime.

Tutte le cap del manifest vengono registrate perché le invocazioni peer da una cartuccia possono avere come destinazione cap di un’altra cartuccia che non fanno parte del DAG. Per esempio, l’handler di una cartuccia di ML può chiamare modelcartridge per scaricare un modello; questa cap non è nel DAG ma deve essere instradabile.

Sorgente: executor.rs.

Gli input di un passo

Un passo è una richiesta. Ogni input del passo è una propria sequenza STREAM_START/CHUNK/STREAM_END all’interno di quella richiesta, consegnata con l’URN con cui la cartuccia lo conosce: l’in= della cap per l’input principale, l’URN dell’argomento per ogni altro. Una raccolta è un solo stream: i valori dei suoi membri nell’ordine dei binding, come una sola sequenza. I valori che l’esecuzione dà a un argomento (il registro degli argomenti) e le costanti scritte nella macchina vengono inviati nella stessa richiesta.

Sorgente: executor.rs.

Pianificazione

Un’unità è pronta quando ogni nodo che legge è stato prodotto o è noto come assente. Le unità pronte si avviano subito; le unità senza un percorso fra loro vengono eseguite contemporaneamente, nella macchina come nel corpo di una mappa. L’algoritmo di Kahn sui passi dà l’ordine delle dipendenze e rifiuta un ciclo; non li serializza.

  • Ciò che limita la concorrenza è l’ammissione, e solo l’ammissione. Un passo avviato attende l’intera catena di pool della sua cap come qualsiasi dispatch (Gestione degli errori). Due passi le cui catene condividono un pool limitato vengono comunque eseguiti uno dopo l’altro, in ordine di arrivo. Non esiste alcun limite a livello di esecutore da configurare.
  • I risultati non dipendono dai tempi. Una raccolta consegna i suoi membri nell’ordine dei binding e una mappa raccoglie nell’ordine degli elementi, qualunque sia finito per primo.
  • Un fallimento termina i suoi fratelli. Ogni ambito ha un segnale di fallimento. Quando un’unità fallisce definitivamente e nulla tollera il fallimento, viene sollevato il segnale del suo ambito: ogni altra richiesta attiva in quell’ambito, e in ogni ambito al suo interno, viene annullata come collaterale prima che venga eseguito qualsiasi join (vedi Gestione degli errori). Un fallimento dentro un’esecuzione di una mappa solleva prima il segnale di quell’esecuzione; se esso raggiunga l’ambito della mappa lo decide la mappa (Mappe).

Sorgente: executor.rs, execute_plan.rs.

Mappe

execute_plan esegue il corpo di un ForEach una volta per elemento, con le esecuzioni in concorrenza, e raccoglie ogni nodo del corpo che qualcosa all’esterno legge in una sequenza nell’ordine degli elementi.

  • Zip. Una mappa su più sequenze prende l’elemento i di ciascuna per l’esecuzione i. Le sequenze le cui lunghezze sono note e differiscono falliscono prima che inizi qualsiasi esecuzione; le sequenze lette man mano che arrivano falliscono quando una termina mentre un’altra ha ancora elementi. In entrambi i casi l’esecuzione fallisce con ExecutionError::ZipLengthMismatch, che nomina ogni nodo sorgente e quanti elementi ha dato.
  • Valori interi. Un nodo esterno alla mappa, una costante, e una sequenza legata a un argomento sequenza vengono consegnati interi a ogni esecuzione. Questo include la sequenza che la mappa itera.
  • Mappe annidate. Una sequenza prodotta dentro un corpo e legata a un argomento che prende un solo valore è una mappa dentro la mappa: il suo corpo viene eseguito una volta per elemento, per ogni esecuzione del corpo esterno. Letto dall’interno del corpo esterno, ciò che la mappa interna ha raccolto è la sequenza di quell’esecuzione (un passo che prende una sequenza la ripiega lì, una volta per esecuzione esterna). Lette dall’esterno di entrambe, le sequenze interne vengono accodate, nell’ordine degli elementi esterni: un’unica sequenza piatta.
  • Indirizzamento. Un’esecuzione di un corpo è indirizzata dal token di passo della mappa e dal suo indice, e dall’esecuzione di ogni mappa che la racchiude in cui si trova (ForEachBodyCoordinate::enclosing). Il registro degli argomenti annota nel giornale il dispatch di un corpo sotto quell’indirizzo.
  • Un’esecuzione che fallisce. Quando un passo di un corpo fallisce definitivamente e la sua politica non lo tollera, decide la MapFailurePolicy dell’host. Fail: la mappa fallisce, e con essa l’esecuzione, come farebbe il fallimento di qualsiasi passo. DropItem: quell’elemento non contribuisce nulla a ciò che la mappa raccoglie, il suo fallimento viene registrato sull’esito del suo corpo, le altre esecuzioni proseguono, e l’esecuzione termina completata con fallimenti tollerati; una mappa di cui ogni esecuzione è fallita fallisce comunque. La CLI capdag esegue le mappe con Fail.

Valori assenti e passi saltati

Un nodo è assente quando l’output che lo avrebbe prodotto non è scattato: un output optional per cui la cartuccia non ha aperto alcuno stream, un output di un passo saltato, o un output di un passo fallito con on_error=absent.

  • Un passo con un argomento obbligatorio legato a un nodo assente viene saltato: non viene aperta alcuna richiesta, viene riportato come skipped (uno StepReport che nomina i nodi assenti), e ogni output che avrebbe prodotto è assente.
  • A un argomento non obbligatorio legato a un nodo assente viene inviato il suo valore predefinito, o nulla quando non ne ha.
  • Una raccolta invia i membri presenti, in ordine; se nessuno è presente invia una sequenza vuota.
  • In una mappa, un’esecuzione il cui passo terminale è stato saltato, o il cui output preso non è scattato, non aggiunge nulla alla sequenza raccolta di quell’output.
  • Un Output del piano il cui nodo è assente viene restituito come PlanTerminal::NotProduced, che non è né un valore vuoto né un errore.

L’assenza viene decisa quando la richiesta che produce termina (END senza alcuno stream per quell’output), quindi un passaggio che esce da un output facoltativo non va mai in pipeline: il consumatore viene aperto solo quando l’output è scattato.

La richiesta di un passo

Che cosa fa l’esecutore per una singola invocazione di cap:

Raccolta degli input

Ogni nodo che il passo legge è stabilito prima che la sua richiesta venga aperta: il suo valore è stato prodotto da un’unità precedente, dal passo che lo precede nella catena (che gli trasmette in streaming in tempo reale), oppure è stato dato all’esecuzione. Un nodo che non è né prodotto né assente quando il passo viene raggiunto è un difetto della pianificazione e fallisce con ExecutionError::NoIncomingData.

sequenceDiagram
    participant EX as Executor
    participant SW as RelaySwitch
    participant P as Cartridge

    EX->>SW: register_external_request(rid, cap_urn)
    Note over SW: Restituisce (XID, response_rx)

    loop Per ogni input del passo
        EX->>SW: STREAM_START (stream_id, media_urn)
        EX->>SW: CHUNK(s)
        EX->>SW: STREAM_END
    end
    EX->>SW: END

    Note over SW: Instrada alla cartuccia tramite corrispondenza delle cap
    SW->>P: (inoltrato)

    loop ciclo select!
        alt Frame di risposta su rx
            P-->>EX: STREAM_START → quale output è questo stream
            P-->>EX: CHUNK → i dati di quell'output
            P-->>EX: LOG → inoltra l'avanzamento
            P-->>EX: END → fatto
            P-->>EX: ERR → fallimento
        else Timeout della pompa (200ms)
            Note over EX: read_from_masters_timeout<br/>Instrada i frame peer attraverso lo switch
        else Timeout di attività (120s)
            Note over EX: ExecutionError::ActivityTimeout
        end
    end

Invocazione della cap

  1. Chiama RelaySwitch::register_external_request(rid, cap_urn) per predisporre l’instradamento. Restituisce (XID, response_rx): l’XID per l’instradamento dei frame e un canale mpsc per ricevere i frame di risposta.
  2. Invia gli stream di input: per ogni input, invia STREAM_START (con stream_id e media_urn) → CHUNK (con il payload suddiviso ai confini di max_chunk) → STREAM_END.
  3. Invia END per segnalare che tutti gli input sono stati consegnati.

Ciclo di raccolta della risposta

La risposta viene raccolta con un ciclo tokio::select! che fa due cose in modo concorrente:

tokio::select! {
    pump_result = read_from_masters_timeout(200ms) => {
        // Instrada i frame peer attraverso il RelaySwitch
    }
    Some(frame) = rx.recv() => {
        last_activity = Instant::now();  // azzera il timeout
        match frame.frame_type {
            StreamStart => { /* quale output dichiarato è questo stream */ }
            Chunk => { /* i dati di quell'output: memorizzati, inoltrati o scartati */ }
            End   => { /* risposta completa */ }
            Log   => { /* inoltra l'avanzamento alla callback */ }
            Err   => { /* esecuzione fallita */ }
        }
    }
}

La pompa concorrente (read_from_masters_timeout) è necessaria. Senza di essa, una cartuccia che effettua una chiamata peer durante la gestione di una richiesta andrebbe in deadlock: i frame della richiesta peer arriverebbero allo switch ma non avrebbero alcun percorso verso la cartuccia di destinazione, perché il ciclo di risposta dell’esecutore legge solo il proprio canale di risposta, senza pompare i frame attraverso lo switch.

Il timeout di 200ms su read_from_masters_timeout garantisce che il ciclo controlli regolarmente sia il canale di risposta sia la pompa dei frame.

Timeout di attività

DEFAULT_ACTIVITY_TIMEOUT_SECS = 120. Qualsiasi frame ricevuto su rx.recv() azzera last_activity. Se non arriva alcun frame per 120 secondi, l’esecutore interrompe con ExecutionError::ActivityTimeout.

Durante l’inattività, l’esecutore registra un avviso ogni 30 secondi per facilitare il debug.

Le singole cap possono sostituire il timeout predefinito tramite la chiave activity_timeout_secs nei loro metadati. Questa viene controllata prima che venga applicato il valore predefinito.

Sorgente: executor.rs (DEFAULT_ACTIVITY_TIMEOUT_SECS, ACTIVITY_TIMEOUT_METADATA_KEY).

Suddivisione dell’avanzamento

Ogni unità di un ambito ha una quota dell’avanzamento dell’ambito in proporzione ai passi che contiene, e l’avanzamento dell’esecuzione è la somma di ciò che ciascuna ha contribuito:

run_progress = sum over units of  share * unit_progress      (unit_progress in [0, 1])

Un’unità non contribuisce nulla prima di avviarsi, il proprio avanzamento mappato della cartuccia mentre è in esecuzione, e la sua intera quota quando termina: completata, saltata, oppure fallita e tollerata. Più unità in esecuzione contemporaneamente muovono ciascuna il proprio termine, quindi il totale non diminuisce mai e non dipende da quale sia finita per prima. La quota di una mappa è divisa fra le sue esecuzioni allo stesso modo, e la mappa riporta l’aggregato delle sue esecuzioni sotto il proprio token di passo. Vedi Mappatura dell’avanzamento per la formula di mappatura.

Gestione degli errori

pub enum ExecutionError {
    StepFailed { step_token_id, source },            // qualsiasi fallimento, attribuito al suo passo immutabile dello strand
    CartridgeNotFound { cap_urn },                    // nessuna cartuccia fornisce la cap
    CartridgeDownloadFailed(String),
    CartridgeExecutionFailed { cap_urn, code, class, details, arg_urn }, // la cartuccia ha restituito ERR
    EffectContractViolation { cap_urn, effect, runtime_input, expected, actual },
    StreamContractViolation { cap_urn, declared_is_sequence, declared_streaming, emitted_is_sequence, emitted_unbounded },
    NoIncomingData { node },                          // mancano i dati del nodo sorgente
    IoError(std::io::Error),
    HostError(String),
    CartridgeUnavailable { cap_urn, details },
    FabricRegistryError(String),
}

Un passo fallito termina l’esecuzione a meno che la sua politica non dica altrimenti (Notazione delle macchine):

  • retry=N. Un fallimento la cui classe di attribuzione è ritentabile (resource, environment, internal; Gestione degli errori) viene tentato di nuovo, fino a N ulteriori volte, dopo un ritardo che cresce con jitter (net_retry::RetryPolicy). input e user non vengono mai ritentati. Un passo con una politica è una catena a sé: nulla di ciò che emette viene inoltrato a un altro passo finché un tentativo non è terminato con successo, e ciò che un tentativo fallito ha emesso, su un file come in memoria, viene scartato. Ogni tentativo legge i valori degli argomenti letti dal primo; un aggiornamento che arriva fra un tentativo e l’altro viene riportato come already-dispatched. Ogni tentativo viene registrato sul passo con la sua classe, il suo codice e il suo messaggio.
  • on_error=absent. Quando il passo è fallito definitivamente, i suoi output sono assenti (Valori assenti e passi saltati), il passo viene riportato come failed-tolerated con il fallimento che ha avuto, e l’esecuzione prosegue. Un annullamento non viene mai tollerato.
  • Nessuna delle due. Il fallimento è terminale per l’esecuzione, attribuito al suo passo.

Un’esecuzione termina quindi in uno di quattro modi: completata; completata con fallimenti tollerati (ogni output che poteva essere prodotto lo è stato, e almeno un passo è failed-tolerated o un elemento di una mappa è stato scartato); fallita; annullata. I primi due sono il RunVerdict di un PipelineResult; gli altri due sono l’errore che execute_plan restituisce. Un host riporta il secondo come verdetto a sé, mai come successo.

ExecutionError acquisisce ZipLengthMismatch { sources } e OutputContractViolation { cap_urn, detail } (uno stream di risposta che non nomina alcun output dichiarato, un secondo stream per uno stesso output, un output obbligatorio senza stream, o un tipo portato che non supera la sua verifica; detail dice quale, nominando l’output).

Smantellamento prima del join. Quando un qualsiasi task di una catena in pipeline fallisce (l’alimentazione dell’input, un inoltro intermedio (un ERR a monte) o la raccolta terminale), la catena viene prima smantellata e poi sottoposta a join: ogni richiesta ancora attiva della catena viene annullata come collaterale: un Cancel attribuito alla classe del fallimento originario, codice ABORTED_COLLATERAL, con un messaggio che nomina il passo fallito (12.2 §Cancel). I suoi feed live si chiudono, i suoi attendenti di credito vengono rilasciati; solo allora si attendono i task di alimentazione/inoltro, che terminano rapidamente perché le loro richieste non esistono più. La catena restituisce l’errore originario, attribuito al suo passo con la sua classe dichiarata; le conseguenze riportate dai task sottoposti a join vengono registrate nel log, mai restituite. Una sorgente live a monte di una cap fallita non può quindi mai tenere un’esecuzione in RUNNING, e il record dell’esecuzione è il fallimento reale, non “cancelled”. Il guasto viene osservato nel momento in cui avviene: i task di invio/inoltro riportano il loro errore tipizzato su un canale dei guasti della catena con cui la raccolta terminale è in competizione (tokio::select!), quindi il collettore non attende un END che non può arrivare.

Esecuzione in pipeline

L’esecutore di riferimento mette in pipeline le catene: sequenze massimali di passi in cui la risposta di ogni passo fluisce in tempo reale nell’input principale del passo successivo. Un passo prosegue la catena del passo che produce il suo input quando valgono tutte queste condizioni:

  • ha esattamente uno stream di input, e quello è l’output principale del produttore così com’è: non una raccolta, e non un altro output del produttore;
  • nient’altro legge qualcosa di ciò che il produttore produce: nessun altro passo, nessun altro output preso, e nulla al di fuori dei passi in esecuzione (un output della macchina, un passo di un altro ambito);
  • l’output principale del produttore non è optional: di un output che può non scattare si sa se è presente o assente solo quando la sua richiesta termina;
  • nessuno dei due passi ha una politica: un passo che può essere tentato di nuovo, o il cui fallimento è tollerato, viene eseguito da solo;
  • le catene di pool delle due cap non condividono alcun pool limitato, e il salto non attraversa alcun confine di contratto di streaming (§Contratti di streaming più sotto: un produttore che può emettere senza una promessa di lunghezza verso un consumatore che non consuma senza di essa).

Ogni altro passo apre una catena e viene alimentato da valori materializzati. All’interno di una catena, tutte le invocazioni di cap vengono aperte in anticipo e task di inoltro per salto (forward_frames) trasmettono in streaming i frame dalla cap i alla cap i+1 in tempo reale, così la cap a valle riceve il suo primo elemento prima che la cap a monte abbia emesso l’ultimo. Il credito scorre per salto (acquisito dalla finestra della cap a valle, concesso alla cap a monte man mano che viene consumato, L10/L14), e l’alimentazione dell’input del passo di testa viene eseguita in modo concorrente con la raccolta terminale (L15).

Un pool limitato condiviso fra due cap adiacenti è un confine di esecuzione perché un permesso di ammissione rappresenta la proprietà di slot concreti del pool dalla REQ fino alla risposta terminale. L’esecutore esegue la prima cap fino al completamento, ne materializza l’output e solo allora acquisisce l’invocazione dipendente. Non deve preacquisirle entrambe: cap adiacenti le cui catene si incontrano in un pool di capacità uno lascerebbero la prima invocazione senza alimentazione mentre l’acquisizione della seconda attende lo slot che la prima detiene. Due cap della stessa cartuccia in pool limitati DISGIUNTI non condividono alcun permesso e continuano lo streaming. Questo confine preserva l’ammissione FIFO e l’isolamento dalla perdita di processi; non è un nuovo tentativo né un fallback.

Ogni STREAM_START di output viene verificato rispetto ai contratti dichiarati della cap che lo produce al momento della ricezione, in tutti e tre i punti di ricezione: il salto in pipeline (forward_frames), il collettore terminale batch (collect_terminal_output) e il consumatore terminale incrementale (TerminalOutput). La verifica (ResponseAudit::audit_frame) vincola lo stream all’output che nomina (ResponseContract, ogni output dichiarato dalla definizione della cap), in quest’ordine:

  1. quale output è: uno stream che non nomina alcun output è quello dell’output principale; uno stream che ne nomina uno (chiave del frame 21, confrontata come tagged URN) è quello di quell’output dichiarato, e mai quello principale. L’output non deve avere ancora alcuno stream. Uno stream dell’output principale da una cap che non dichiara alcun output, uno stream che nomina un output non dichiarato o quello principale, e un secondo stream per uno stesso output sono ciascuno ExecutionError::OutputContractViolation;
  2. che cosa porta: l’output principale è vincolato al contratto di effetto, il media URN emesso rispetto a CapUrn::is_conformant_runtime_output (vedi Input e output e Struttura del cap URN: il contratto di effetto a runtime), e una violazione è ExecutionError::EffectContractViolation; un altro output deve portare un valore del proprio media URN, o esattamente l’input della cap quando dichiara carries = "input", e una violazione è OutputContractViolation;
  3. il contratto di stream di quell’output (OutputContract, il suo CapOutput.is_sequence + CapOutput.streaming): l’is_sequence del frame deve essere uguale alla cardinalità dichiarata, e un’emissione unbounded è consentita solo da un output dichiarato streaming; una violazione è ExecutionError::StreamContractViolation, che nomina sia la dichiarazione sia l’emissione.

Quando la richiesta termina con successo (ResponseAudit::audit_end), ogni output obbligatorio deve aver avuto uno stream; uno che non lo ha avuto è OutputContractViolation. Un output facoltativo senza stream non è scattato.

Sono tutti fallimenti definitivi attribuiti come internal: la cartuccia ha violato la propria dichiarazione. Poiché is_sequence viene verificato, il ragionamento del piano sulla cardinalità può fare affidamento sulla dichiarazione.

Ciò che ogni punto di ricezione prende da una risposta è lo stream dell’output per cui si trova lì. Lo stream di un output che nulla prende viene verificato come ogni altro e poi consumato: i suoi chunk vengono contati e il suo produttore riceve il credito, così una cap non si blocca mai su un output che nessun passo legge, e non vengono consegnati da nessuna parte.

I frame STREAM_START inoltrati vengono poi rietichettati con l’URN con cui viene consegnato l’input del passo a valle (StepInput.stream_urn) prima di essere inviati, esattamente come send_step_input etichetta gli stream della testa di una catena. Le cartucce demultiplexano gli stream di input per equivalenza di arg URN (vedi 13.2), quindi l’URN di output più ricco del produttore non deve raggiungere il consumatore come etichetta dello stream. La verifica viene eseguita PER PRIMA: la rietichettatura sostituisce la pretesa del produttore con l’arg URN dichiarato dal piano, il che senza la verifica maschererebbe una cap che mente. Il piano ha già dimostrato che l’output del produttore è conforme all’input dell’arco; forward_frames riafferma tale conformità e fallisce in modo definitivo in caso di violazione (un bug di collegamento/tipo deve emergere al salto, non come un errore di argomento mancante all’interno della cartuccia).

Nota sul regime: i dati di un nodo intermedio in pipeline non vengono mai materializzati nella mappa dei risultati dell’esecutore, ed è proprio questo lo scopo. I consumatori che hanno bisogno di un valore intermedio devono farvi riferimento come fan-out (il che spezza il segmento in quel punto e lo materializza). I pozzi di catena che SI materializzano (confini di divisione) vengono inoltre catturati su disco quando l’host fornisce una radice transitoria (vedi §Artefatti transitori dell’esecuzione (cattura dell’host)).

Per il consumo incrementale dell’output di una cap terminale (stream illimitati), usa TerminalOutput { next_item(), finish() }, la controparte in streaming di collect_terminal_output, che ne condivide la semantica dei frame (avanzamento finale trasportato da END, concessione del credito, consegna alla callback).

Stato degli argomenti durante l’esecuzione (RunArgumentLedger)

execute_plan non riceve una mappa di argomenti immutabile: legge gli argomenti di ogni passo attraverso un RunArgumentLedger, lo stato mutabile degli argomenti del piano e il suo giornale dei dispatch sotto un unico mutex. Quel mutex è la regola di ordinamento per la corsa fra aggiornamento e dispatch: il dispatch dei passi di una catena, dei passi di un’esecuzione di una mappa, e un aggiornamento di argomento vengono serializzati su di esso, e l’esito per passo di ogni aggiornamento viene deciso nella stessa sezione critica che lo applica.

  • dispatch_steps registra nel giornale i passi di una catena come inviati-ed-esauriti e restituisce i valori che le loro invocazioni leggono, in un unico passo atomico, così un aggiornamento arriva interamente prima o interamente dopo i passi della catena, mai in mezzo. Un passo tentato di nuovo con retry legge ciò che ha letto il suo primo tentativo.
  • dispatch_steps_in_map registra nel giornale il dispatch dei passi in un’esecuzione di una mappa (quell’esecuzione e ogni esecuzione precedente hanno preso i loro valori) e restituisce i valori che quell’esecuzione legge.
  • exhaust_steps segna come chiusi i passi di una mappa quando le sue sorgenti di elementi terminano; non esiste più alcun dispatch che un aggiornamento possa raggiungere.
  • apply convalida un intero lotto (ogni token deve nominare un nodo cap di QUESTO piano; i token dei passi sono generati per piano, quindi un token estraneo fa rifiutare il lotto; e un aggiornamento indirizzato a un argomento che la macchina lega, a un nodo o a una costante, fa rifiutare il lotto nominando il passo e l’argomento: quel valore appartiene alla macchina), poi riporta per ogni aggiornamento: applied (nessun dispatch ancora), applied-to-remaining-bodies(n) (n esecuzioni della mappa hanno già preso il vecchio valore), oppure already-dispatched (valore lasciato intatto, perché falsificare il registro del lavoro già eseguito è esattamente ciò che il giornale esiste per impedire).

Le chiavi sono gli id dei nodi cap del piano (i token generati dei passi dello strand). Un host espone il ledger per la durata di una chiamata a execute_plan; la CLI di riferimento ne costruisce uno per esecuzione a partire dagli argomenti analizzati.

Questo livello esiste solo nell’implementazione di riferimento. Le altre implementazioni non includono l’esecutore di piani (execute_plan, mappe, consegna degli argomenti per passo) e quindi non hanno un ledger degli argomenti.

Macchine con feed live e terminali illimitati

Una macchina la cui sorgente è un riferimento a un feed live (13.2 §Media di riferimento) non è una modalità di esecuzione speciale. È lo stato naturale di una macchina, di cui l’esecuzione limitata su file è il caso degenere il cui input si trova a terminare immediatamente. Il nucleo di esecuzione delle macchine con feed live risiede nella libreria CapDAG ed è esercitato direttamente dalla CLI capdag (capdag run su una macchina con una sorgente live, incluso il feed integrato media:live;synthetic); floom-engine usa la stessa superficie della libreria e aggiunge solo la mediazione dell’host specifica della piattaforma (cattura in sandbox, attribuzione TCC). La CLI è la superficie di esecuzione di riferimento.

Due conseguenze per l’esecutore:

  • L’alimentazione della testa è invariata dove è il runtime a risolvere il riferimento: il motore passa il record del selettore come valore dell’argomento di input principale della cap di testa (un piccolo valore di riferimento, esattamente come un percorso di file), e il livello di risoluzione del trasporto del runtime della cartuccia lo trasforma nello stream di input illimitato all’interno del processo. La cattura mediata dall’host (lo stesso dispatch capdag::capture, aperto dall’host) copre i due casi in cui la cartuccia non può essere il risolutore: una cartuccia in sandbox che non può aprire il dispositivo, e una sorgente live che alimenta direttamente una mappa: il dispatch per elemento avviene lato motore, quindi è il MOTORE ad aprire il feed e a inviare un’esecuzione del corpo per ogni elemento consegnato mentre la cattura è in corso (corpi in volo limitati, snapshot degli elementi pubblicati in sola aggiunta, il tap registrato così che uno stop lo chiuda e i corpi si svuotino).
  • Gli intermedi illimitati ai confini di divisione obbligatori vengono riversati su disco. Una divisione di catena che l’esecutore non può evitare (cap limitate dello stesso dominio di ammissione, fan-out) materializza il suo pozzo; quando lo stream di quel pozzo è illimitato, viene trasmesso a uno spool su file temporaneo invece che in memoria, e la catena a valle (o il driver della mappa) rilegge gli elementi in streaming dallo spool. La memoria resta limitata; i dati sono completi perché la semantica della divisione richiedeva già che il feed terminasse prima.
  • I terminali illimitati devono essere consumati in modo incrementale. Un terminale il cui STREAM_START dichiara unbounded DEVE essere instradato al consumatore incrementale (TerminalOutput) o a un IncrementalWriter (pozzi persistiti); il collect_terminal_output con buffer rifiuta per contratto gli stream illimitati (12.4 §Stream illimitati). La registrazione su disco è il percorso del pozzo persistito applicato a un’esecuzione live; una registrazione arrestata correttamente è un file completo grazie alla semantica di svuotamento descritta più sotto.

Contratti di streaming

La limitatezza è una proprietà a livello di definizione di una cap, dichiarata accanto alla cardinalità e ortogonale a essa, e mai un asse dell’URN (l’URN è l’identità di dispatch; un feed di frame e un file di frame sono dello stesso tipo) né dedotta da is_sequence (una sequenza può essere limitata; un blob scalare può essere illimitato, come un WAV live):

  • CapArg.streaming: l’argomento viene consumato senza una promessa di lunghezza: la cap elabora gli elementi man mano che arrivano e può essere collegata in tempo reale a un produttore illimitato. Solo l’input principale può dichiararlo (10 RULE14): gli argomenti secondari sono valori che il runtime demultiplexa per intero.
  • CapOutput.streaming: l’output può essere emesso senza una promessa di lunghezza (uno STREAM_START unbounded, 12.4).

Cap::streaming_shape() restituisce (input_streams, output_streams), la controparte in streaming di sequence_shape(), ed entrambi viaggiano nella definizione del fabric (streaming nel TOML della cap e nel JSON pubblicato da fabric.js, assente = false).

La regola del salto. All’interno di una catena lineare, un salto da un produttore il cui output ha streaming vero verso un consumatore il cui input principale ha streaming falso è un confine di catena (split_chains_at_streaming_boundary): il produttore viene eseguito fino al completamento, il suo output viene materializzato (riversato su disco quando risulta illimitato; vedi §Macchine con feed live), e il consumatore viene alimentato con l’intero limitato dopo che il produttore ha inviato END. Ogni altro salto va in pipeline. La decisione viene presa in base alle dichiarazioni, così la stessa macchina si comporta allo stesso modo qualunque siano le cartucce che la servono. La topologia dei pool non ha alcun ruolo nella decisione sul confine di streaming.

Perché RULE14. Un feed è una proprietà dell’unico stream a cui una cap è collegata in tempo reale. Un’opzione dichiarata streaming permetterebbe a uno stream illimitato di raggiungere un collettore che deve rifiutarlo (L16) senza che nulla nel piano lo avesse indicato; mantenere lo streaming solo sull’input principale mantiene la regola del salto monodimensionale e la verifica esatta.

Artefatti transitori dell’esecuzione (cattura dell’host)

I pozzi di catena intermedi (la trascrizione fra due cap di modello, i frame fra un bridge e il suo ForEach) non sono né output (nessun elenco; non sono cose destinate a essere conservate) né messaggi (troppo grandi per la memoria). Quando il runtime fornisce una radice transitoria (EngineRuntime::transient_artifact_root, predefinita None; la CLI e i runtime di riferimento mantengono temp+delete), l’esecutore cattura ogni pozzo di catena intermedio nel momento in cui si materializza, durante l’esecuzione:

  • Un pozzo materializzato in memoria (limitato) viene scritto nella sua forma canonica in byte; il file di spool di un pozzo che usa lo spool (illimitato) viene adottato: spostato, non copiato, e i consumatori a valle leggono il percorso adottato.
  • Ogni cattura è una directory: {root}/{node_id}/data più un file accessorio provenance.json che riporta il media urn, la cardinalità, il numero di byte e, per le sequenze, l’[offset, len] di ogni elemento, così un lettore visualizza in anteprima l’elemento N con una sola lettura limitata. La pubblicazione è atomica: le catture vengono preparate in {node_id}.tmp e rinominate al loro posto, così una directory pubblicata contiene sempre il suo file accessorio; una directory .tmp è una cattura in corso (o interrotta da un crash), invisibile ai lettori, di proprietà del reaper TTL dell’host.
  • EngineRuntime::on_transient_artifact scatta per ogni cattura e restituisce Result; un fallimento della pubblicazione è un errore di esecuzione definitivo, perché una superficie di ispezione che diverge silenziosamente dal disco è uno stato illegale.
  • I pozzi dei corpi ForEach non vengono mai catturati (gli interni dei corpi sono esclusi per progetto; gli elementi e gli output di una mappa hanno le proprie superfici).
  • NON ci sono righe di database: il filesystem è l’indice, una directory assente è lo stato legale di scadenza, e la garbage collection anticipata dell’host (eliminazione dell’esecuzione, pulizia all’avvio, reaper TTL) gestisce ogni durata.

Un nodo, una verità. I dati di un pozzo di catena vengono registrati nella forma in memoria della mappa dei risultati OPPURE come percorso di spool, mai in entrambi. Un pozzo che usa lo spool non compare in node_data (il suo residuo in memoria non è lo stream), e il driver della mappa rifiuta un input presente in entrambe le forme come violazione del contratto dell’esecutore invece di tirare a indovinare. Con i transitori attivi, i percorsi di spool adottati sono esclusi dalla pulizia degli spool a fine piano; il reaper ne è proprietario.

Le esecuzioni si arrestano

Una macchina è in esecuzione oppure no. “Cancel” appartiene al livello dei task e al frame di rete che implementa l’interruzione (un frame Cancel annulla una richiesta); a livello di esecuzione il vocabolario e la semantica sono stop:

  • Un’esecuzione si arresta da sola quando il suo stream di input termina: un file subito dopo il suo contenuto; un feed live quando scatta una condizione di stop (duration_ms, max_items) o il dispositivo si chiude. Stesso END, stesso percorso; l’input limitato è il caso in cui questo avviene immediatamente.
  • Un’esecuzione viene arrestata chiudendo il tap: il runtime chiude i feed live della richiesta, lo stream di input del feed termina e la macchina si svuota: ogni elemento in volo prosegue, i terminali terminano normalmente, i writer finalizzano. Gli output di un’esecuzione arrestata sono completi (una registrazione arrestata è un file valido). In rete questo è un frame CloseStream (12.2 §CloseStream) verso la richiesta che porta il feed: il runtime chiude i feed e lascia che la richiesta termini naturalmente invece di interrompere l’handler. Non è un Cancel di alcun tipo (un Cancel annulla sempre), e un CloseStream per una richiesta che non ha alcun feed aperto viene registrato nel log e ignorato: lo stop è uno stop perché lo dice il frame, mai per ciò che il runtime si trova a trovare.
  • Un’esecuzione viene interrotta dalla cascata di annullamento, un Cancel attribuito a user, la variante forte per un’esecuzione che deve morire subito, con il lavoro in volo scartato (il flag force_kill uccide i processi invece di annullarli in modo cooperativo). L’interruzione è l’escalation, mai lo stop predefinito. Le interruzioni proprie dell’host (reaper dei processi bloccati, pressione sulla memoria) sono Cancel attribuiti al motivo dell’host (ABORTED, classe environment / resource); lo smantellamento, da parte dell’esecutore, di una catena attorno a un passo fallito è un Cancel attribuito al fallimento originario (ABORTED_COLLATERAL, la sua classe). Ogni terminale porta quell’attribuzione, così “cancelled” su una superficie significa sempre che è stato un essere umano ad annullare.

Superfici di controllo. Stop e interruzione sono controlli DIVERSI ovunque: il motore espone TaskService.StopTaskInput (chiude i tap live del task in esecuzione: i feed aperti dall’host direttamente, i feed risolti dalla cartuccia tramite un frame CloseStream verso le loro richieste CHE PORTANO IL FEED, che l’esecutore condiviso registra attraverso il punto di aggancio FlowObserver; poi l’esecuzione si svuota e si completa CON i suoi output; lo stato del task resta intatto) accanto a CancelTask (interruzione). In rete lo stop è RelaySwitch::stop_request_feeds, il FRAME CloseStream inviato alla richiesta che porta il feed SENZA toccare lo stato della richiesta lato host (terminare lo stato dell’host è compito dell’interruzione; farlo durante uno stop interrompe l’instradamento a metà svuotamento e blocca l’esecuzione).

Lo stop è una richiesta verificata. Uno stop restituisce un verdetto, mai un successo predefinito nel codice: stop_feed_requests (bifaci::live_feed) chiude i tap dell’host, invia ogni CloseStream e poi attende la prova: che ogni richiesta arrestata raggiunga il suo terminale presso lo switch, entro STOP_INPUT_DRAIN_TIMEOUT, e risponde con StopInputsOutcome { host_taps_closed, feed_requests_stopped, feeds_ended }, dove il successo significa che l’input è terminato. Se non c’è nulla di aperto da arrestare, il risultato è NothingToStop (il motore lo mappa su FAILED_PRECONDITION); un feed che non termina entro il timeout è FeedDidNotEnd, che nomina la richiesta (DEADLINE_EXCEEDED); un feed che termina con ERR è FeedFailed. TaskService.StopTaskInput restituisce StopTaskInputResponse con gli stessi tre conteggi, ed entrambi gli host desktop mostrano il verdetto del motore o il suo rifiuto.

L’intestazione di esecuzione di loom mostra i controlli di un’esecuzione LIVE in primo piano: “Stop input” (chiusura del tap) e “Abort”, senza modificatori nascosti; un’esecuzione limitata mantiene un solo Stop (annullamento cooperativo, ⌥/Alt forza l’uccisione). Nella CLI capdag il primo Ctrl-C è lo stop (i tap si chiudono, la macchina si svuota, gli output vengono emessi) e stampa il verdetto (“Input stopped: N host tap(s) closed, M feed request(s) stopped, M ended”) o il rifiuto; un secondo Ctrl-C, in qualsiasi momento, interrompe con un Cancel attribuito a user. Arrestare l’input è anche il modo in cui una cartuccia con capacità limitata si libera a metà macchina: il suo consumatore live termina sull’input svuotato e il permesso passa alla cap successiva.

Non esiste alcuna restrizione in fase di pianificazione sugli input illimitati: is_sequence sull’argomento che consuma decide il collegamento (un fold che consuma una sequenza si collega direttamente e rimanda l’emissione fino a END, che lo stop consegna; una cap che consuma elementi riceve un ForEach), e streaming decide il salto (§Contratti di streaming: un consumatore solo limitato a valle di un produttore in streaming viene alimentato con l’intero dopo che il produttore termina). La tempistica dell’emissione è una politica dell’operazione, non una struttura di tipo. Un fold su un feed live è un piano valido con emissione differita: concat su un microfono emette l’intera registrazione quando l’esecuzione si arresta.