frame_conv/handle/

pubsub.rs

//! The pub/sub pattern (`frame:conv-pubsub@v1`, F-3b R1): typed
//! publications with caller-supplied identity, fanned out by the substrate
//! itself.
//!
//! The substrate IS the fan-out authority (amendment A5): topic =
//! conversation, epoch = the substrate's linearized membership, admission
//! witness = the commit receipt, and the epoch snapshot at admission = the
//! membership linearization at that receipt. This module adds the typed
//! `PublicationId` and client-side presentation bookkeeping and never
//! re-publishes — "no fan-out worker" is true by construction.
//!
//! Identity discipline (amendment A4 as SUPERSEDED 2026-07-23 — the brief's
//! amendment section carries the ruling): the `PublicationId`'s bytes ARE
//! the record-admission attempt token (the deterministic map is the
//! identity function), but the pinned substrate treats admission tokens as
//! per-attempt correlation identity, NOT durable idempotency keys — a
//! record-admission body conflict is unrepresentable on the wire at
//! protocol 0.3.2 (`AttemptTokenBodyConflict` has no `RecordAdmission`
//! arm; characterized live in `tests/pattern_pubsub.rs`). NO reuse refusal
//! is therefore claimed on live bindings: republishing an id admits a NEW
//! record, same bytes or different. Id-reuse discipline is the publisher's
//! documented obligation; stronger admission idempotency is the named
//! upstream ask ASK-6 (`docs/upstream/liminal-asks-2026-07-23.md`).
//! Subscriber-side exactly-once remains the O(1) contiguity claim over
//! admitted RECORDS — an id-keyed dedup memory of any size stays banned.

use std::time::{Duration, Instant};

use serde::Serialize;
use serde::de::DeserializeOwned;

use liminal_protocol::wire::RecordAdmissionAttemptToken;

use crate::envelope::Envelope;
use crate::error::{CallError, PublishError};
use crate::id::{MessageKind, PatternId, PublicationId};
use crate::outcome::{PublicationItem, PublishReceipt};
use crate::store::ResumeStore;

use super::pump::PumpStep;
use super::state::{ConversationHandle, QueuedItem};

impl<S: ResumeStore> ConversationHandle<S> {
    /// Publishes one typed publication under the caller-supplied identity.
    /// Schema-invalid values are refused typed BEFORE the wire. The receipt
    /// is the admission witness (A5): the substrate owes the publication
    /// once to every epoch live in its membership linearization at this
    /// receipt — publisher death after it does not retract the broadcast,
    /// and zero live epochs mean this validation plus this one admission
    /// outcome were the entire cost.
    ///
    /// PUBLISHER OBLIGATION (the honest contract at the pinned substrate,
    /// A4 superseded 2026-07-23): a `PublicationId` identifies ONE
    /// publication — never republish an id, with the same or different
    /// bytes. The substrate refuses neither on a live binding (admission
    /// tokens are per-attempt identity there, not durable idempotency
    /// keys), so a reused id admits a NEW record and every active epoch
    /// observes it again under the same id. Stronger admission idempotency
    /// is upstream ask ASK-6, non-gating.
    ///
    /// # Errors
    ///
    /// Returns typed publish failures.
    pub fn publish<P: Serialize>(
        &mut self,
        id: PublicationId,
        body: &P,
    ) -> Result<PublishReceipt, PublishError> {
        let envelope = Envelope::from_typed(
            PatternId::PubSub,
            id.to_correlation(),
            MessageKind::Event,
            body,
        )
        .map_err(|refusal| PublishError::SchemaInvalid {
            detail: refusal.detail,
        })?;
        let bytes = envelope
            .encode()
            .map_err(|refusal| PublishError::Protocol {
                detail: refusal.detail,
            })?;
        self.admit_with_token(bytes, RecordAdmissionAttemptToken::new(id.to_bytes()))
    }

    /// Waits up to `wait` for the next pub/sub item. `Ok(None)` is a benign
    /// quiet wait — never an error, never a protocol outcome (the
    /// assertion-8 pin). Items carry the guarantee documented on
    /// [`PublicationItem`]; duplicate presentation of an admitted
    /// publication is suppressed by the O(1) last-presented-sequence
    /// comparison, with every suppression counted
    /// (`duplicate_publications`), never silent.
    ///
    /// The effective wait granularity is the substrate's IO quantum
    /// (hardcoded 5 s at the pinned release — upstream ASK-2): a wait
    /// smaller than one quantum may still block for one quantum.
    ///
    /// # Errors
    ///
    /// Returns a typed connection fate.
    pub fn next_publication<P: DeserializeOwned>(
        &mut self,
        wait: Duration,
    ) -> Result<Option<PublicationItem<P>>, CallError> {
        let started = Instant::now();
        loop {
            while let Some(item) = self.publications_inbox.pop_front() {
                if let QueuedItem::Event { seq, .. } = &item {
                    let value = seq.value();
                    if self
                        .last_presented_publication
                        .is_some_and(|last| value <= last)
                    {
                        self.counters.duplicate_publications += 1;
                        continue;
                    }
                    self.last_presented_publication = Some(value);
                }
                return Ok(Some(convert_publication(item)));
            }
            if started.elapsed() >= wait {
                return Ok(None);
            }
            match self.pump_step() {
                Ok(PumpStep::Classified | PumpStep::Quiet) => {}
                Err(detail) => return Err(CallError::ConnectionLost { detail }),
            }
        }
    }
}

/// Converts a queued item into the typed pub/sub surface, decoding
/// publication payloads against the caller's typed message
/// (schema-invalid surfaces typed, never dropped) and recovering the
/// publication identity from the record's correlation slot.
fn convert_publication<P: DeserializeOwned>(item: QueuedItem) -> PublicationItem<P> {
    match item {
        QueuedItem::Event {
            envelope,
            publisher,
            seq,
        } => {
            let id = PublicationId::from_correlation(envelope.correlation);
            match envelope.decode_payload::<P>() {
                Ok(body) => PublicationItem::Publication {
                    id,
                    body,
                    publisher,
                    seq,
                },
                Err(refusal) => PublicationItem::SchemaInvalid {
                    id,
                    publisher,
                    seq,
                    detail: refusal.detail,
                },
            }
        }
        QueuedItem::Joined { peer, seq } => PublicationItem::PeerJoined { peer, seq },
        QueuedItem::Departed { peer, seq, reason } => {
            PublicationItem::PeerDeparted { peer, seq, reason }
        }
        QueuedItem::Failed { peer, seq, failure } => {
            PublicationItem::PeerFailed { peer, seq, failure }
        }
        QueuedItem::Compacted { seq } => PublicationItem::HistoryCompacted { seq },
        QueuedItem::Gap { expected, observed } => PublicationItem::Gap { expected, observed },
    }
}