polyc-query 2026.9.0

//! What a maintained projection is told, and what it is never told (#1565,
//! chunk B6).
//!
//! Both projections in this crate — the dashboard rows and the search index's
//! dirty set — take their input from the durable commit feed, not an
//! in-process observer on the event-log host: a [`FeedRecord`] per commit,
//! carrying the journal records that commit bracketed and the positions they
//! landed at.
//!
//! # Two inputs, because the feed has one blind spot
//!
//! A commit feed reports commits. It reports no *mutation*: a destroy, an
//! excision, a repair, or a migration changes a partition's durable content
//! without producing a feed entry, and nothing in the feed's contract says
//! otherwise. So a projection here takes two kinds of news:
//!
//! - [`commit_events`] turns a chunk of the feed into the `(position, event)`
//!   pairs both projections already fold, which is the steady-state path.
//! - [`PartitionChange`] is the other one, and it arrives from whoever issued
//!   the mutating command, after that command returned its receipt. Transport
//!   success is not authority (INV-22): a command that has not yet earned a
//!   receipt may never commit, and invalidating against it would discard
//!   correct state on behalf of a mutation that never happened.
//!
//! # A missed notification costs latency, not correctness
//!
//! Feed delivery is at-least-once over a network, so a chunk may arrive twice
//! and a subscription may die without saying so. Neither may leave a projection
//! permanently wrong. Every consumer here therefore does two things: it dedups
//! on apply, so a redelivered chunk contributes nothing, and it reconciles
//! against durable state on its own schedule, so a subscription that never
//! delivers again is corrected rather than believed. The reconcile is what
//! makes the feed a latency mechanism rather than a correctness one —
//! [`crate::dashboard::DashboardProjection::rebuild_from_full_fleet`] and the
//! search index's own coverage sweep are those backstops.

use polyc_eventlog::{Event, TrustTag};
use polyc_state::feed::FeedRecord;
use polyc_state::journal::RecordTrust;
use polyc_state::page::Positioned as _;

/// How a partition's durable content changed outside the ordinary append path.
///
/// The feed carries none of these — see the module doc — so each one reaches a
/// projection from the caller that issued the command and read its receipt.
/// Three variants rather than the journal's five command shapes, because a
/// projection only ever asks three questions of a mutation: is this partition
/// gone for good, has it moved somewhere else, or does what is left need
/// recomputing from scratch.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PartitionChange {
    /// The partition was destroyed. Nothing is left to rebuild from and
    /// nothing may ever publish for it again, so a projection records the
    /// erasure rather than merely forgetting the rows.
    Destroyed,
    /// The partition's records moved to another partition and this name now
    /// holds nothing. Deliberately not [`Self::Destroyed`]: the conversation is
    /// alive under a new id, so this side must read as though it were never
    /// projected, not as though it were erased.
    MigratedAway,
    /// The partition's durable content changed in place — an excision, a
    /// repair's quarantine drop, or records copied in from a migration source.
    /// Surviving records renumber from the origin, so anything a projection
    /// derived from the old positions is suspect and only a fresh replay
    /// recomputes it.
    Rewritten,
}

/// Something holding per-partition state that a mutation invalidates.
///
/// Named as a capability rather than reached for as a concrete type, so a
/// Container's mutation-reporting path depends on the one thing it actually
/// needs — "drop what you cached for this partition" — instead of on the whole
/// query surface. [`crate::authority::QueryAuthority`] is the implementor in
/// production.
pub trait PartitionInvalidation: Send + Sync {
    /// Drop whatever is held for `partition`.
    ///
    /// Called after the command that changed the partition returned its
    /// receipt, never before — see [`PartitionChange`].
    fn invalidate_partition(&self, partition: &str);
}

/// Returns the `(position, event)` pairs one chunk of the commit feed carries,
/// in feed order, in the coordinate [`crate::journal::PartitionJournal`] reads.
///
/// Flattens the commits: a [`FeedRecord`] is one commit and holds every journal
/// record that commit bracketed, each already carrying the durable position it
/// landed at. Both projections in this crate fold positioned events, and the
/// position is what makes a redelivered chunk harmless — it is compared against
/// what the consumer already applied rather than trusted to be new.
///
/// # The positions are converted, and have to be
///
/// A [`FeedRecord`] carries State's own numbering, which is one-based; the port
/// every consumer of these pairs reads through is zero-based, and its contract
/// says so. So each position converts through
/// [`crate::journal::port_position`] here, at the one place the feed crosses
/// into this crate.
///
/// Neither consumer would survive the pairs arriving in State's coordinate,
/// because both immediately compare them against positions that came off the
/// port. The search index derives its replay bound from these positions and
/// then looks for the boundary event inside the range that bound produced: one
/// too high and the event is outside its own window, the coverage it publishes
/// names content it never read, and the worker refuses — leaving every
/// conversation unsearchable, which is what #2146 was. The dashboard keeps one
/// high-water mark per row and advances it from these positions and from full
/// replays alike, so two bases in one counter would re-fold records it had
/// already applied.
///
/// The provenance tag is carried across, not invented and not dropped: a
/// quarantined record stays quarantined here, because the folds downstream
/// decide what to trust from it.
#[must_use]
pub fn commit_events(records: &[FeedRecord]) -> Vec<(u64, Event)> {
    records
        .iter()
        .flat_map(FeedRecord::records)
        .map(|record| {
            (
                crate::journal::port_position(record.position()),
                Event::with_trust(
                    record.kind().as_str().to_owned(),
                    record.payload().to_vec(),
                    trust_tag(record.trust()),
                ),
            )
        })
        .collect()
}

/// Returns the event-log provenance tag one contract record declares.
///
/// The inverse of the control plane's own `contract_trust`, and exhaustive on
/// purpose: a new provenance class has to be classified here rather than
/// silently folded into "unspecified".
const fn trust_tag(trust: RecordTrust) -> TrustTag {
    match trust {
        RecordTrust::Unspecified => TrustTag::Unspecified,
        RecordTrust::TrustedUser => TrustTag::TrustedUser,
        RecordTrust::QuarantinedContent => TrustTag::QuarantinedContent,
    }
}

/// Builds the commit one batch of `(event, position)` pairs would appear as on
/// the durable feed.
///
/// `positions` are PORT positions — what a caller reads back through
/// [`crate::journal::PartitionJournal`], and what every test here already has
/// in hand — and the records this builds carry the State positions those name.
/// That conversion is the fixture's whole reason to exist: State is what fills
/// a real feed record, so a chunk numbered in the port's coordinate is a chunk
/// no subscription would ever deliver, and a projection tested against one
/// would agree with itself about a numbering production does not use. #2146 hid
/// here for exactly that long.
///
/// Test-only, and shared across this crate's projection tests on purpose: a
/// test that hand-rolled its own shape could drive a chunk no real
/// subscription would ever deliver.
#[cfg(test)]
fn test_checkpoint(
    partition: &str,
    feed_position: polyc_state::revision::JournalPosition,
    journal_position: polyc_state::revision::JournalPosition,
) -> polyc_state::feed::SourceCheckpoint {
    use polyc_state::{
        feed::{ATTESTATION_SIGNATURE_BYTES, ATTESTATION_SIGNER_BYTES, SourceCheckpoint},
        journal::JournalAttestation,
        revision::{CommitRoot, JournalSource, PartitionIncarnation},
    };

    let evidence_leaf = journal_position.get();
    SourceCheckpoint::try_new(
        JournalSource::new(
            polyc_state::id::PartitionId::new(partition),
            PartitionIncarnation::from_bytes([1; PartitionIncarnation::LEN]),
        ),
        feed_position,
        journal_position,
        evidence_leaf,
        JournalAttestation::new(
            CommitRoot::from_bytes(
                [u8::try_from(journal_position.get() % 256).unwrap_or_default(); CommitRoot::LEN],
            ),
            evidence_leaf.saturating_add(1),
            vec![2; ATTESTATION_SIGNATURE_BYTES],
            vec![3; ATTESTATION_SIGNER_BYTES],
        ),
    )
    .expect("the test checkpoint is structurally complete")
}

#[cfg(test)]
pub(crate) fn test_commit(partition: &str, events: &[Event], positions: &[u64]) -> FeedRecord {
    use polyc_state::digest::ContentDigest;
    use polyc_state::feed::CommitEnvelope;
    use polyc_state::id::CommandId;
    use polyc_state::journal::{JournalRecord, RecordKind};
    use polyc_state::revision::JournalPosition;

    let records: Vec<JournalRecord> = events
        .iter()
        .zip(positions.iter().copied())
        .map(|(event, position)| {
            JournalRecord::new(
                crate::journal::state_position(position),
                RecordKind::new(event.kind.clone()),
                match event.trust {
                    TrustTag::Unspecified => RecordTrust::Unspecified,
                    TrustTag::TrustedUser => RecordTrust::TrustedUser,
                    TrustTag::QuarantinedContent => RecordTrust::QuarantinedContent,
                },
                event.payload.clone(),
            )
        })
        .collect();
    // The heads are already State's, with no conversion of their own: a head is
    // "how many records the partition holds", so the head BEFORE a commit whose
    // first record lands at port `p` is `p`, and the head after one whose last
    // record lands at port `q` is `q + 1`. The same two numbers, read as a count
    // rather than as an index.
    let head_before = JournalPosition::new(positions.first().copied().unwrap_or(0));
    let head_after = JournalPosition::new(
        positions
            .last()
            .copied()
            .map_or(0, |last| last.saturating_add(1)),
    );
    let feed_position = JournalPosition::new(head_before.get().saturating_add(1));
    FeedRecord::new(
        feed_position,
        CommitEnvelope::new(
            test_checkpoint(partition, feed_position, head_after),
            CommandId::new(format!("test-commit-{}", head_before.get())),
            ContentDigest::from_bytes([0u8; ContentDigest::LEN]),
            head_before,
            head_after,
            records.len() as u64,
        ),
        records,
    )
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs, clippy::unwrap_used)]

    use super::*;
    use polyc_state::digest::ContentDigest;
    use polyc_state::feed::CommitEnvelope;
    use polyc_state::id::CommandId;
    use polyc_state::journal::{JournalRecord, RecordKind};
    use polyc_state::revision::JournalPosition;

    /// One commit of `records`, which carry the positions STATE assigned them
    /// — one-based, because a commit assigns `position.next()` from
    /// [`JournalPosition::ORIGIN`] and no record ever occupies the origin.
    fn commit(partition: &str, records: Vec<JournalRecord>) -> FeedRecord {
        let head_before = JournalPosition::ORIGIN;
        let head_after = JournalPosition::new(records.len() as u64);
        FeedRecord::new(
            JournalPosition::new(1),
            CommitEnvelope::new(
                test_checkpoint(partition, JournalPosition::new(1), head_after),
                CommandId::new("cmd-1".to_owned()),
                ContentDigest::from_bytes([7u8; ContentDigest::LEN]),
                head_before,
                head_after,
                records.len() as u64,
            ),
            records,
        )
    }

    #[test]
    fn a_commits_records_arrive_with_their_positions_and_provenance() {
        let chunk = vec![commit(
            "conv-a",
            vec![
                JournalRecord::new(
                    JournalPosition::new(1),
                    RecordKind::new("turn_start".to_owned()),
                    RecordTrust::TrustedUser,
                    b"a".to_vec(),
                ),
                JournalRecord::new(
                    JournalPosition::new(2),
                    RecordKind::new("user_msg".to_owned()),
                    RecordTrust::QuarantinedContent,
                    b"b".to_vec(),
                ),
            ],
        )];

        let events = commit_events(&chunk);
        assert_eq!(events.len(), 2);
        assert_eq!(events[0].0, 0);
        assert_eq!(events[0].1.kind, "turn_start");
        assert_eq!(events[0].1.trust, TrustTag::TrustedUser);
        assert_eq!(events[1].0, 1);
        assert_eq!(events[1].1.trust, TrustTag::QuarantinedContent);
        assert_eq!(events[1].1.payload, b"b".to_vec());
    }

    #[test]
    fn several_commits_flatten_into_one_ordered_run() {
        let chunk = vec![
            commit(
                "conv-a",
                vec![JournalRecord::new(
                    JournalPosition::new(1),
                    RecordKind::new("k0".to_owned()),
                    RecordTrust::Unspecified,
                    Vec::new(),
                )],
            ),
            commit(
                "conv-a",
                vec![JournalRecord::new(
                    JournalPosition::new(2),
                    RecordKind::new("k1".to_owned()),
                    RecordTrust::Unspecified,
                    Vec::new(),
                )],
            ),
        ];

        let positions: Vec<u64> = commit_events(&chunk)
            .into_iter()
            .map(|(position, _)| position)
            .collect();
        assert_eq!(positions, vec![0, 1]);
    }

    /// The skew #2146 shipped: the feed handed out State's numbering while
    /// every consumer of it read the port's, so a position off this chunk was
    /// one greater than the coordinate the same record answers to on a replay.
    ///
    /// Stated as the boundary case that actually broke, because it is the one
    /// an off-by-one survives everywhere else: State's FIRST record is position
    /// one, and the port's first record is position zero.
    #[test]
    fn the_first_record_a_partition_holds_is_port_position_zero() {
        let chunk = vec![commit(
            "conv-a",
            vec![JournalRecord::new(
                JournalPosition::new(1),
                RecordKind::new("turn_start".to_owned()),
                RecordTrust::Unspecified,
                Vec::new(),
            )],
        )];

        assert_eq!(
            commit_events(&chunk)[0].0,
            0,
            "State's first durable position is one and the port's is zero; a feed that reported \
             one here would name a position no replay of a one-record partition returns"
        );
    }

    /// Every position off this chunk names the record the port would hand back
    /// at that same number — the round trip, over a run rather than a single
    /// record, so a conversion applied to only the first would show.
    #[test]
    fn a_chunks_positions_round_trip_through_the_ports_coordinate() {
        let chunk = vec![commit(
            "conv-a",
            vec![
                JournalRecord::new(
                    JournalPosition::new(1),
                    RecordKind::new("k0".to_owned()),
                    RecordTrust::Unspecified,
                    Vec::new(),
                ),
                JournalRecord::new(
                    JournalPosition::new(2),
                    RecordKind::new("k1".to_owned()),
                    RecordTrust::Unspecified,
                    Vec::new(),
                ),
                JournalRecord::new(
                    JournalPosition::new(3),
                    RecordKind::new("k2".to_owned()),
                    RecordTrust::Unspecified,
                    Vec::new(),
                ),
            ],
        )];

        for (position, event) in commit_events(&chunk) {
            assert_eq!(
                crate::journal::state_position(position),
                chunk[0]
                    .records()
                    .iter()
                    .find(|record| record.kind().as_str() == event.kind)
                    .expect("the event came off this chunk")
                    .position(),
                "a port position must name the State record it was derived from"
            );
        }
    }
}