polyc-query 2026.9.0

//! Turning a replayed range into the rows and coverage a segment carries.
//!
//! A pure fold: it touches neither the projection nor the event log, so the
//! part of this index where the correctness risk concentrates is testable
//! directly rather than only behind a spawned host.
//!
//! Every rule here exists because of a specific failure. They are stated at
//! the point they are enforced rather than collected in one list, so a change
//! that removes one has to delete the reason with it.

use polyc_eventlog::Event;
use polyc_state::revision::PartitionIncarnation;

use super::IndexedMessage;
use super::store::Coverage;
use super::terms::TermKey;

/// What one projection pass produced.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct Projected {
    /// The messages this pass indexed, ascending by position.
    pub(crate) messages: Vec<IndexedMessage>,
    /// The coverage the resulting segment publishes.
    pub(crate) coverage: Coverage,
    /// Whether the replayed range held a verified excision marker for this
    /// conversation.
    ///
    /// The caller needs this even though the fold already stripped what the
    /// marker names, because stripping only reaches the range this pass
    /// replayed. An excision names positions BELOW the watermark by
    /// construction — that is what makes it invisible to a forward window — so
    /// the removed text lives in segments an append does not rewrite. A forward
    /// pass that saw one must therefore become a rebuild rather than publish a
    /// scanned frontier past a marker it did not actually apply everywhere.
    pub(crate) excision_in_range: bool,
    /// The `turn_start` of the earliest turn this range showed opening without
    /// closing, when one held the watermark below the requested `end`.
    ///
    /// `None` when the barrier did not bind. Reported because "the watermark did
    /// not move" has two causes a caller cannot otherwise tell apart:
    /// the conversation is caught up, or a turn that never closes has pinned it
    /// and every committed turn behind it is unindexable. The first is the
    /// healthy end of a pass and the second is an outage.
    pub(crate) open_turn_at: Option<u64>,
}

/// Project a replayed range into the rows and coverage describing it.
///
/// The result is a DELTA. It holds only what this range committed, never a
/// merge with what an earlier pass indexed, because the store is append-only
/// segments and [`SearchProjection::append`] takes exactly that delta: the
/// prefix a forward pass leaves out already sits in the segments before it, and
/// a read unions them. So a segment that publishes only a window is the
/// intended shape, not a pass that lost its earlier messages.
///
/// This function must never carry the previous rows forward and republish
/// them: against segments that is not merely redundant but wrong — it
/// would rewrite every message of the conversation on every turn, the quadratic
/// cost the segment layout exists to avoid.
///
/// Rebuild against forward pass is therefore the caller's distinction, not this
/// function's — a rebuild replays from position zero and publishes with
/// `rebuild`, which drops the earlier segments outright, so a rewrite or repair
/// that compacted journal positions cannot leave rows behind naming text the
/// journal no longer holds.
///
/// [`SearchProjection::append`]: super::store::SearchProjection::append
pub(crate) fn project(
    term_key: &TermKey,
    partition: &str,
    source_incarnation: PartitionIncarnation,
    events: &[(u64, Event)],
    end: u64,
) -> Projected {
    let mut range = events.to_vec();

    // Excision first: a verified excision marker is an ORDINARY append, so
    // nothing else in this pipeline would notice it. Stripping before
    // projection is what keeps removed text out of the index rather than
    // relying on a later filter.
    //
    // Match on the LOGICAL partition name, which is what the observer keys by
    // and what the journal listing returns. The partition name carries the
    // `conv-` prefix and the marker carries the bare conversation id, so the
    // prefix is what the comparison has to add — without it, no conversation
    // would ever match and excised text would stay indexed forever.
    //
    // This compared ENCODED names until #1748. It had to, because the old
    // mapping was lossy and the raw id could not be recovered from a partition
    // name. The names are reversible now, so the comparison is over the ids
    // themselves.
    let excisions = polyc_facts::verified_excisions_matching(&range, partition, |excision| {
        format!("conv-{}", excision.conversation_id) == partition
    });
    let excision_in_range = !excisions.is_empty();
    let excised = polyc_facts::excised_positions(&range, &excisions);
    polyc_facts::strip_excised(&mut range, &excised);
    // A turn that paused for a human records a marker retracting the narration
    // it already streamed, and the marker is applied at READ time because the
    // log is append-only — the durable event carries no flag of its own. So a
    // reader that only skips already-flagged messages skips nothing, and the
    // narration the marker exists to retract is exactly what gets indexed.
    // Applied here, on the same events the projection reads, the way
    // `partition_tables` and `history_nav` apply it (`#2702`).
    //
    // No window can split a turn from its own marker: the marker rides the
    // completion batch, one event ahead of `turn_complete`, and the barrier
    // below refuses to advance the watermark past a turn that has not
    // completed. So a turn becomes indexable and becomes markable in the same
    // window, never in two.
    let withheld = polyc_facts::withheld_turn_ids_positioned(&range);
    polyc_facts::withhold_paused_turn_text_positioned(&mut range, &withheld);

    // The watermark may not advance past the start of a turn that is still
    // open, because `committed_message_facts` requires BOTH markers to be in
    // the slice it is given. If the watermark moved past an open turn's
    // `turn_start`, the next forward window would carry that turn's
    // `turn_complete` without its start, the turn would read as uncommitted
    // forever, and its text would be permanently invisible behind a watermark
    // claiming to cover it. Stopping short costs one re-read of a short prefix;
    // not stopping loses committed text silently.
    let barrier = open_turn_barrier(&range);
    let effective_end = barrier.map_or(end, |barrier| end.min(barrier));
    // Only a barrier that actually BOUND is worth reporting: one above `end` did
    // not hold anything back, and naming it would make every ordinary pass look
    // pinned.
    let open_turn_at = barrier.filter(|barrier| *barrier < end);

    // ONE pass produces both halves. An earlier version filtered `range` twice
    // with the same predicate — once for the events fed to the projection, once
    // for the positions its ordinals resolve against. If those two ever
    // drifted, every indexed message would land at the wrong journal position
    // and hits would fetch the wrong text, with nothing to catch it.
    let (indexed, bare): (Vec<u64>, Vec<Event>) = range
        .iter()
        .filter(|(position, _)| *position < effective_end)
        .map(|(position, event)| (*position, event.clone()))
        .unzip();
    let facts = polyc_facts::committed_message_facts(&bare);

    // `ordinal` indexes the slice the facts were projected from, so it maps
    // back through `indexed`, which was unzipped from that same slice — never
    // through `range`, which still holds the events above `effective_end`.
    let mut messages = Vec::with_capacity(facts.len());
    for fact in &facts {
        let Some(&position) = usize::try_from(fact.ordinal)
            .ok()
            .and_then(|index| indexed.get(index))
        else {
            continue;
        };
        messages.push(IndexedMessage {
            position,
            turn_id: fact.turn_id.clone(),
            term_hashes: term_key.hash_text(&fact.text),
        });
    }
    // No dedup here, and none needed: within one range each fact resolves to
    // its own journal position, and this pass never merges anything into what
    // an earlier one produced. Recovery is at-least-once, so a range can be
    // replayed twice — what makes that harmless is that the fold is a pure
    // function of the range, so the second pass yields byte-identical rows.
    messages.sort_by_key(|message| message.position);

    Projected {
        coverage: Coverage {
            indexed_through: effective_end,
            source_incarnation,
            available: true,
            // `end`, not `effective_end`. The excision sweep above ran over the
            // WHOLE replayed range, including the part the open-turn barrier
            // holds the watermark below — a marker sitting above the barrier is
            // just as applied as one below it. Recording the barrier instead
            // would leave a conversation with one long-running turn claiming a
            // frontier that never advances, so every read would report an
            // excision it had in fact already handled.
            excision_scanned_through: end,
        },
        messages,
        excision_in_range,
        open_turn_at,
    }
}

/// The lowest position that must stay INSIDE the next replay window: the
/// `turn_start` of the earliest turn this range shows opening without closing.
///
/// `None` when every turn in the range is complete, in which case the caller's
/// requested boundary stands.
fn open_turn_barrier(range: &[(u64, Event)]) -> Option<u64> {
    let mut starts: std::collections::HashMap<uuid::Uuid, u64> = std::collections::HashMap::new();
    let mut completed: std::collections::HashSet<uuid::Uuid> = std::collections::HashSet::new();

    for (position, event) in range {
        let (base, turn_id) = polyc_proto::kinds::parse(&event.kind);
        let Some(turn_id) = turn_id else { continue };
        if base == polyc_proto::kinds::TURN_START {
            starts.entry(turn_id).or_insert(*position);
        } else if base == polyc_proto::kinds::TURN_COMPLETE {
            completed.insert(turn_id);
        }
    }

    starts
        .into_iter()
        .filter(|(turn_id, _)| !completed.contains(turn_id))
        .map(|(_, position)| position)
        .min()
}

#[cfg(test)]
mod tests;