Streaming

Stream multiplexati, suddivisione in chunk, numerazione di sequenza e riordinamento.

Stream multiplexati

Una singola richiesta (identificata dal suo ID di richiesta nella chiave 2) può trasportare più stream con nome. Ogni stream ha uno stream_id univoco (chiave 11) e un media_urn (chiave 12) che identifica il tipo di dati che vi scorre.

Questo rende possibili invocazioni di capacità con più argomenti. Per esempio, una cap che accetta sia un documento PDF sia una stringa di prompt li riceve come due stream separati all’interno della stessa richiesta: uno con media_urn = "media:document;ext=pdf" e uno con media_urn = "media:enc=utf-8". Sul lato della risposta, anche un handler può emettere più stream di output.

Sorgente: capdag-rs/src/bifaci/frame.rs (StreamStart, StreamEnd); cartridge_runtime.rs (InputStream, OutputStream).

Ciclo di vita di uno stream

Uno stream segue questa sequenza:

sequenceDiagram
    participant S as Mittente
    participant R as Destinatario

    S->>R: STREAM_START (stream_id, media_urn)
    S->>R: CHUNK (chunk_index=0, checksum)
    S->>R: CHUNK (chunk_index=1, checksum)
    Note over S,R: ... altri chunk ...
    S->>R: CHUNK (chunk_index=N, checksum)
    S->>R: STREAM_END (chunk_count=N+1)

• STREAM_START annuncia lo stream. Il media_urn indica al destinatario che tipo di dati aspettarsi.
• I frame CHUNK trasportano i dati del payload. Ognuno ha un chunk_index che parte da 0 e un checksum per l’integrità.
• STREAM_END segna lo stream come completo. Il suo campo chunk_count contiene il numero totale di CHUNK inviati, così che il destinatario possa verificare di averli ricevuti tutti. Gli stream illimitati lo omettono (vedi sotto).

Dopo STREAM_END, qualsiasi CHUNK per lo stesso stream_id è un errore di protocollo.

Completamento della richiesta

Quando tutti gli stream di una richiesta sono terminati (ciascuno con il proprio STREAM_END), il mittente invia END per segnalare che non verranno aperti altri stream e che la richiesta è completa:

sequenceDiagram
    participant S as Mittente
    participant R as Destinatario

    rect rgb(240, 248, 255)
        Note right of S: Stream A
        S->>R: STREAM_START (stream A)
        S->>R: CHUNK(s)
        S->>R: STREAM_END (stream A)
    end

    rect rgb(245, 255, 245)
        Note right of S: Stream B
        S->>R: STREAM_START (stream B)
        S->>R: CHUNK(s)
        S->>R: STREAM_END (stream B)
    end

    S->>R: END

Nessun frame di flusso può seguire un END per lo stesso ID di richiesta (L4). I runtime lo impongono nella loro fase di scrittura: un frame di flusso successivo al terminale è un ritardatario BENIGNO: viene soppresso e conteggiato per tipo di frame nei contatori dei ritardatari (mai come scarto, perché nulla è andato storto) e non viene mai scritto. I metadati terminali (l’avanzamento e il messaggio finali) viaggiano nel frame END stesso (L5), quindi nulla di importante può entrare in competizione con il terminale. Gli intermediari inoltrano il terminale prima di rilasciare lo stato di instradamento; i frame che arrivano per uno stato già rilasciato vengono conteggiati come ritardatari benigni quando il registro garantisce la terminazione, e come scarti no_route altrimenti, mai come errori di protocollo (L6).

ERR è un frame terminale alternativo. Segnala che la richiesta è fallita. Come per END, nessun frame di dati può seguirlo. ERR ed END si escludono a vicenda: una richiesta termina con esattamente uno dei due.

I frame LOG (aggiornamenti di avanzamento, messaggi di log) possono comparire in qualsiasi punto della sequenza, intercalati ai CHUNK e tra uno stream e l’altro. Non influiscono sul flusso dei dati.

Stream illimitati

Uno STREAM_START può portare unbounded = true (chiave 20): lo stream non fa alcuna promessa di lunghezza (L16). Una cattura dal vivo, un video generato, una trascrizione senza fine prestabilita: qualsiasi cosa la cui dimensione non sia nota quando lo stream si apre.

• Il produttore lo apre con OutputStream::start_unbounded(); il suo STREAM_END omette chunk_count.
• I destinatari DEVONO consumare gli stream illimitati in modo incrementale (InputStream::recv, TerminalOutput::next_item). I collettori con buffer (collect_bytes, collect_items, collect_value) rifiutano gli stream illimitati con un errore bloccante, perché mettere in buffer uno stream senza promessa di lunghezza significa memoria illimitata.
• “Illimitato” significa “nessuna lunghezza promessa”, non “non termina mai”: lo stream termina comunque tramite STREAM_END, il terminale della richiesta o Cancel. Cancel è efficace a metà stream (L17): produttore e consumatore osservano entrambi la terminazione tempestivamente (le letture bloccate si sbloccano, chi attende credito viene rilasciato).
• È il controllo di flusso a rendere tutto questo sicuro: il produttore non può mai portarsi avanti rispetto al consumatore più della finestra concessa. Vedi Controllo di flusso.
• Un’emissione illimitata viene dichiarata prima che avvenga: una cap può aprire uno stream di output illimitato solo se l’output della sua definizione porta streaming: true, e può essere collegata dal vivo a un produttore illimitato solo se il suo input principale porta streaming: true (15.2 §Contratti di streaming, 10 RULE14). L’esecutore verifica ogni STREAM_START ricevuto rispetto a quella dichiarazione: un flag unbounded proveniente da un output non streaming, o un is_sequence diverso dalla cardinalità dichiarata, fa fallire l’esecuzione alla ricezione (StreamContractViolation). La dichiarazione è anche ciò che decide se un passaggio procede in pipeline o materializza, così un consumatore con buffer non incontra mai uno stream illimitato di sorpresa.

Feed dal vivo

Un feed dal vivo (microfono, webcam, qualsiasi cattura in tempo reale) è uno stream di sequenza illimitato prodotto dal livello di risoluzione del trasporto del runtime a partire da un argomento di riferimento a un feed dal vivo (vedi Input e output §Media di riferimento). Non è un nuovo tipo di stream; è la combinazione illimitato + sequenza che questa sezione già definisce, con un contratto sui metadati:

• Metadati a livello di stream (STREAM_START): i valori effettivi del formato negoziato della cattura, cioè i parametri con cui il dispositivo si è effettivamente aperto (frequenza di campionamento, canali, risoluzione, formato dei pixel), che possono essere più specifici dei tag dell’URN del contenuto.
• Metadati per elemento (lo slot dei metadati di emit_list_item): ogni elemento porta
  • seq: l’indice dell’elemento nell’ordine di cattura, monotono a partire da 0, contando anche gli elementi scartati (un salto nel seq consegnato è reale),
  • pts_us: timestamp di presentazione, in microsecondi dall’inizio della cattura,
  • capture_ts_us: ora di cattura dell’orologio di sistema (µs Unix),
  • gap: presente solo sul primo elemento dopo un overrun: { "dropped": N, "duration_us": D }, il numero e l’intervallo temporale degli elementi scartati al bordo della cattura dall’elemento consegnato in precedenza. A valle un gap DEVE essere trattato come una vera discontinuità (per esempio un trascrittore azzera il proprio stato di frase), mai concatenato silenziosamente attraverso di esso.

L’ordinamento è stratificato: (1) l’ordine sul wire per flusso è garantito dal trasporto (SeqAssigner + ReorderBuffer); (2) i confini tra elementi all’interno di uno stream sono delimitati da sequenze di chunk, quindi l’ordine degli elementi in uno stream è preservato da un capo all’altro; (3) i seq/pts_us per elemento esistono per l’unico punto in cui l’ordine si rimescola davvero (la ricongiunzione di un ForEach parallelo) e per la ricostruzione temporale che tiene conto dei salti. Il wire non riordina mai; il parallelismo sì.

Gli elementi di sequenza non vengono mai accorpati in scrittura (l’esenzione dall’accorpamento), quindi un elemento dal vivo è sul wire nel momento in cui viene emesso, e la latenza del feed è di un elemento, non di una finestra di accorpamento.

Gli input limitati sono il caso degenere di questo modello: un file è un feed dal vivo il cui stream termina subito dopo il suo contenuto. Il meccanismo è identico; cambia solo il momento in cui arriva END.

Suddivisione in chunk

Suddivisione automatica in chunk

Quando il payload da inviare supera max_chunk (negoziato durante l’handshake, predefinito 256 KB), il writer lo suddivide automaticamente. Se ne occupa il metodo FrameWriter::write_chunked():

1. Il primo chunk (chunk_index = 0) porta content_type, len (numero totale di byte) e offset = 0.
2. I chunk successivi portano offset (la posizione in byte all’interno del payload totale).
3. Il chunk finale ha eof = true.

Un payload vuoto viene inviato come un singolo chunk con eof = true e len = 0.

Dal punto di vista dell’handler, tutto questo è trasparente. Il metodo OutputStream::write() chiama internamente write_chunked quando i dati superano il limite di dimensione del chunk. L’InputStream del destinatario riassembla i chunk in ordine.

Sorgente: capdag-rs/src/bifaci/io.rs (FrameWriter::write_chunked); cartridge_runtime.rs (OutputStream::write).

Framing degli elementi in modalità sequenza

Le due modalità di stream incapsulano i payload dei CHUNK in modo diverso, e il demux del destinatario deve decodificarli di conseguenza:

• Modalità scalare (is_sequence = false, OutputStream::write): ogni payload di CHUNK è un valore CBOR autonomo: i byte grezzi vengono suddivisi ai confini di max_chunk e ogni pezzo viene nuovamente incapsulato come un proprio Value::Bytes. La decodifica per singolo frame è sempre valida. I confini tra chunk non hanno alcun significato proprio (i destinatari concatenano i byte interni), ed è questo che rende legittimo l’accorpamento delle scritture: i runtime raccolgono piccole scritture scalari di byte in un unico CHUNK, svuotato quando il lotto raggiunge 4096 byte, quando il byte più vecchio in buffer ha 20 ms alla scrittura successiva, a ogni emissione non Bytes sullo stesso stream (una barriera di ordinamento), e incondizionatamente da close() PRIMA dello STREAM_END, il cui chunk_count promette quindi il conteggio accorpato. Un emettitore LLM che produce un token alla volta genera così al massimo circa un frame per finestra di età durante la generazione a regime, invece di un frame per token, dividendo il traffico di frame e di credito per il fattore di lotto senza alcun cambiamento visibile al destinatario. Gli stream di sequenza non vengono MAI accorpati.
• Modalità sequenza (is_sequence = true, OutputStream::emit_list_item): ogni elemento viene codificato in CBOR una sola volta, e i byte codificati vengono suddivisi tra frame CHUNK ai confini di max_chunk. Il payload di un frame è un frammento RFC 8742 grezzo, non un valore CBOR autonomo. I metadati per elemento viaggiano solo sul primo frame di frammento dell’elemento. I frame non scavalcano mai il confine di un elemento: ogni elemento inizia su un frame nuovo.

Il destinatario riassembla quindi gli stream di sequenza con granularità di elemento: i frammenti vengono messi in buffer per stream e decodificati in modo incrementale (la codifica a lunghezza definita di CBOR è priva di prefissi ambigui, quindi un elemento incompleto significa senza ambiguità “servono altri byte”, mai una decodifica errata). All’handler viene consegnato un valore decodificato per elemento, con i metadati del primo frammento. Uno STREAM_END che arriva con byte in buffer che non completano un elemento è un errore di decodifica bloccante (elemento troncato), mai uno scarto silenzioso.

Interazione con il credito (L10/L12): la finestra di credito ha granularità di frame, ma l’handler consuma e concede per elemento consegnato. Il demux restituisce al mittente il credito di ogni frammento di continuazione (tutti i frame dopo il primo di un elemento) subito all’arrivo, così un elemento che si estende su più frame della finestra di credito può sempre finire di arrivare; il primo frame dell’elemento resta coperto dalla concessione di consumo dell’handler, preservando la contropressione a livello di elemento.

Sorgente: cartridge_runtime.rs (emit_list_item, SeqReassembly, try_decode_sequence_item, demux_multi_stream, demux_single_stream).

Campi del chunk

Ogni frame CHUNK porta:

I campi chunk_index e checksum vengono validati in fase di decodifica, e un frame CHUNK privo di uno dei due viene rifiutato con un errore di protocollo.

Numerazione di sequenza

Perché il protocollo funzioni correttamente, i frame di una richiesta devono arrivare in ordine. Ai confini dei relay, i frame provenienti da master diversi possono intercalarsi o arrivare fuori ordine. I numeri di sequenza permettono di rilevarlo e di correggerlo.

FlowKey

Un flusso è identificato dalla coppia (request_id, optional routing_id), chiamata FlowKey. Ogni FlowKey distinta ha un proprio contatore di sequenza indipendente.

pub struct FlowKey {
    pub rid: MessageId,       // ID della richiesta (chiave 2)
    pub xid: Option<MessageId>, // ID di instradamento (chiave 13), se presente
}

La presenza o l’assenza di un XID conta. (RID=A, XID=None) e (RID=A, XID=Some(5)) sono due flussi separati con contatori indipendenti. La stessa richiesta può essere vista con e senza XID in punti diversi della catena di relay, prima e dopo che il RelaySwitch gliene assegni uno.

Sorgente: frame.rs (FlowKey).

SeqAssigner

Il SeqAssigner assegna numeri di sequenza monotonamente crescenti per FlowKey. Si trova nella fase di output, nel task del writer e non al momento della creazione del frame. Questo garantisce che i valori di seq riflettano l’ordine effettivo in cui i frame vengono scritti sul wire.

pub struct SeqAssigner {
    counters: HashMap<FlowKey, u64>,
}

Per ogni flusso, il contatore parte da 0 e aumenta di 1 a ogni frame. Non ci sono salti. I frame che non appartengono a un flusso (Hello, Heartbeat, RelayNotify, RelayState) vengono saltati, e l’assegnatore lascia il loro seq a 0.

Dopo che è stato assegnato un frame terminale (END o ERR), il contatore del flusso dovrebbe essere ripulito tramite remove().

Sorgente: frame.rs (SeqAssigner).

Buffer di riordinamento

Il ReorderBuffer si trova ai confini dei relay (precisamente nel RelaySlave, sul percorso socket → locale) e garantisce che i frame vengano consegnati in ordine di sequenza.

Che cosa fa un frame in arrivo (consegnato insieme a quelli che sblocca, trattenuto o rifiutato) è deciso dal modello dimostrato (formal/CapDAG/Bifaci/Flow.lean, Reorder.accept), generato in ogni mirror. Il buffer conserva i frame; il modello conserva il numero atteso successivo e i numeri trattenuti, ed è dimostrato che i frame vengono passati oltre nell’ordine in cui sono stati scritti (delivered_in_order), che nessuno va perso (nothing_is_lost) e che, una volta arrivati tutti i frame, tutti i frame sono stati consegnati (all_arrived_all_delivered).

pub struct ReorderBuffer {
    flows: HashMap<FlowKey, FlowState>,
    max_buffer_per_flow: usize,
}

struct FlowState {
    order: Reorder,              // quello del modello: seq atteso + i seq trattenuti
    buffer: HashMap<u64, Frame>, // i frame trattenuti veri e propri
}

Quando arriva un frame:

flowchart TD
    A["Arriva un frame con seq"] --> B{seq == expected?}
    B -->|Sì| C["Consegna il frame"]
    C --> D["Svuota i frame consecutivi<br/>in buffer"]
    D --> E["Avanza expected_seq"]

    B -->|seq > expected| F["Metti il frame in buffer"]
    F --> G{"Dimensione buffer > max?"}
    G -->|Sì| H["Errore di protocollo"]
    G -->|No| I["Attendi che il salto si colmi"]

    B -->|seq < expected<br/>o duplicato| H

1. In ordine (seq == expected_seq): consegnato immediatamente, insieme a eventuali frame consecutivi già in buffer. expected_seq avanza oltre tutti i frame consegnati.
2. Fuori ordine (seq > expected_seq): messo in buffer. Il frame attende finché il salto non viene colmato.
3. Obsoleto o duplicato (seq < expected_seq, o seq già in buffer): errore di protocollo. Il frame viene rifiutato.

Se il buffer contiene più di max_buffer_per_flow frame (predefinito 64) per un singolo flusso, anche questo è un errore di protocollo, perché il salto è troppo grande e qualcosa è andato storto.

I frame che non appartengono a un flusso aggirano completamente il buffer di riordinamento e vengono restituiti immediatamente.

Dopo che è stato consegnato un frame terminale (END o ERR), cleanup_flow() rimuove lo stato del flusso dal buffer.

Sorgente: frame.rs (ReorderBuffer).