arete-server 0.39.0

use crate::bus::{BusManager, BusMessage};
use crate::cache::{CacheWrite, EntityCache, LifetimeOrdering, PatchOrigin, ViewVersion};
use crate::mutation_batch::{MutationBatch, SlotContext, SlotIndexDomain};
use crate::shared_entity::{split_version, with_version, SharedEntity};
use crate::view::{Projection, ViewIndex, ViewSpec};
use crate::websocket::frame::{apply_wire_format, Mode, SourceFrame};
use arete_interpreter::vm::{VmContext, WholeEntityRequests};
use arete_interpreter::{AccountPosition, CanonicalLog, WholeEntity};
use bytes::Bytes;
use lru::LruCache;
use serde_json::Value;
use smallvec::SmallVec;
use std::collections::HashSet;
use std::future::Future;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex as StdMutex};
use tokio::sync::mpsc;
use tracing::{debug, debug_span, error, instrument, warn};

#[cfg(feature = "otel")]
use crate::metrics::Metrics;

/// Whole-entity requests the projector keeps outstanding at once. One per
/// refused key; asking again after one is forgotten is always safe. The same
/// bound holds for the positions their resends keep.
const WHOLE_ENTITY_REQUEST_CAPACITY: usize = 4_096;

#[derive(Debug, Clone, PartialEq, Eq)]
struct RememberedChange {
    seq: Option<String>,
    source_domain: Option<SlotIndexDomain>,
    account_position: Option<AccountPosition>,
}

/// `(export, key)` to the recency and account lifetime of its latest change.
type LastChanges = LruCache<(String, String), RememberedChange>;

/// The projector's way back to the VM its mutations come from.
///
/// The entity cache is bounded. A patch for a key it does not hold carries
/// only the fields that changed and cannot rebuild the entity, unless it is
/// the mutation that created the entity. A linked VM marks those
/// ([`WholeEntity::Created`]), so the cache stores a creation and refuses any
/// other patch for a key it lacks, whether it evicted the key or never held
/// it. The projector then asks the VM, which holds the whole entity, to send
/// all of it at the end of its next batch of mutations, whether or not the
/// key changes in it (see [`WholeEntityRequests`]).
///
/// A resend is the entity again, not a change, so it must not take the
/// position (`_seq`) of the batch it arrives in: in a view ordered by recency
/// it would rank ahead of entities that really changed there. The projector
/// remembers the position of each requested key's latest change — the refused
/// patch, or any change it sees for the key after it — and stamps the resend
/// with that, in the cache, in derived views and on the `upsert` it
/// publishes. It learns a key's changes in stream order, which the VM cannot:
/// positions are assigned per batch, after the VM has emitted it. A resend
/// with no remembered position (one past the bound, or from a source without
/// a VM, which sends whole entities as changes) takes its batch's.
/// For an account-owned entity it also remembers the explicit account position
/// at request time, so a whole entity produced for an old lifetime cannot land
/// after deletion and recreation.
///
/// The generated runtime hands its VM over through
/// [`crate::snapshot::register_runtime`], inside [`Self::scope`], before it
/// emits anything. A mutation source that registers no VM marks nothing and
/// cannot be asked: the cache takes a patch for a key it lacks for a new
/// entity unless it remembers evicting the key, and a remembered key stays
/// out until the source sends a mutation marked whole
/// ([`arete_interpreter::Mutation::mark_whole_entity`]) or a source delete.
#[derive(Clone)]
pub struct EntityResync {
    requests: WholeEntityRequests,
    /// For each key with a resend on its way: the `_seq` of its latest change
    /// and the account lifetime for which it was requested. Capped by hand at
    /// [`WHOLE_ENTITY_REQUEST_CAPACITY`], least recently changed out first.
    last_changes: Arc<StdMutex<LastChanges>>,
    linked: Arc<AtomicBool>,
    warned_unlinked: Arc<AtomicBool>,
}

tokio::task_local! {
    static ACTIVE_ENTITY_RESYNC: EntityResync;
}

impl Default for EntityResync {
    fn default() -> Self {
        Self::new()
    }
}

impl EntityResync {
    pub fn new() -> Self {
        Self {
            requests: WholeEntityRequests::new(WHOLE_ENTITY_REQUEST_CAPACITY),
            last_changes: Arc::new(StdMutex::new(LruCache::unbounded())),
            linked: Arc::new(AtomicBool::new(false)),
            warned_unlinked: Arc::new(AtomicBool::new(false)),
        }
    }

    /// Run a mutation source (the generated parser) so that the VM it
    /// registers answers this projector's requests.
    pub async fn scope<F: Future>(&self, future: F) -> F::Output {
        ACTIVE_ENTITY_RESYNC.scope(self.clone(), future).await
    }

    /// Answer requests from `vm`.
    pub fn link(&self, vm: &Arc<StdMutex<VmContext>>) {
        vm.lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .set_whole_entity_requests(self.requests.clone());
        self.linked.store(true, Ordering::Release);
    }

    pub fn is_linked(&self) -> bool {
        self.linked.load(Ordering::Acquire)
    }

    /// What an unmarked patch from the source is: with a VM linked, which
    /// marks every creation, a change to an entity created earlier.
    fn unmarked_origin(&self) -> PatchOrigin {
        if self.is_linked() {
            PatchOrigin::Change
        } else {
            PatchOrigin::Unknown
        }
    }

    /// The request set, for inspection.
    pub fn requests(&self) -> &WholeEntityRequests {
        &self.requests
    }

    fn last_changes(&self) -> std::sync::MutexGuard<'_, LastChanges> {
        self.last_changes
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    /// Note a change to `export` `key` at `seq`. A refused change is about to
    /// be requested whole, so its position is remembered for the resend; any
    /// other change moves a remembered position forward. Only a linked VM
    /// resends.
    fn note_change(
        &self,
        export: &str,
        key: &str,
        seq: &Option<String>,
        source_domain: Option<SlotIndexDomain>,
        account_position: Option<AccountPosition>,
        refused: bool,
    ) {
        if !self.is_linked() {
            return;
        }
        let mut last_changes = self.last_changes();
        if refused {
            last_changes.put(
                (export.to_string(), key.to_string()),
                RememberedChange {
                    seq: seq.clone(),
                    source_domain,
                    account_position,
                },
            );
            while last_changes.len() > WHOLE_ENTITY_REQUEST_CAPACITY {
                last_changes.pop_lru();
            }
        } else if !last_changes.is_empty() {
            if let Some(last) = last_changes.get_mut(&(export.to_string(), key.to_string())) {
                last.seq = seq.clone();
                last.source_domain = source_domain;
                if account_position.is_some() {
                    last.account_position = account_position;
                }
            }
        }
    }

    /// The recency and account-lifetime position a resend of `export` `key`
    /// keeps, if one is remembered. Forgets it.
    fn take_last_change(&self, export: &str, key: &str) -> Option<RememberedChange> {
        let mut last_changes = self.last_changes();
        if last_changes.is_empty() {
            return None;
        }
        last_changes.pop(&(export.to_string(), key.to_string()))
    }

    fn request(&self, export: &str, key: &Value) {
        if self.is_linked() {
            self.requests.request(export, key);
        } else if !self.warned_unlinked.swap(true, Ordering::Relaxed) {
            warn!(
                export,
                "Entity cache refused a patch for an evicted key and no VM is registered to \
                 send the whole entity; the key stays out of snapshots until its source sends \
                 a mutation marked whole or deletes it. Raise max_entities_per_view if the \
                 working set exceeds it."
            );
        }
    }

    fn cancel(&self, export: &str, key: &Value, cache_key: &str) {
        self.requests.cancel(export, key);
        self.take_last_change(export, cache_key);
    }
}

/// Link `vm` to the projector of the server whose parser is running, if any.
/// Called by [`crate::snapshot::register_runtime`].
pub(crate) fn link_active_resync(vm: &Arc<StdMutex<VmContext>>) {
    let _ = ACTIVE_ENTITY_RESYNC.try_with(|resync| resync.link(vm));
}

/// The `_version` stamped into every frame's data: `{epoch}:{counter}`.
///
/// Clients order one key's frames by it. `_seq` cannot do that: every update
/// decoded from one transaction shares it, and within a slot account updates
/// and instructions number themselves differently. The counter follows the
/// order this projector merges changes into the entity cache, which is the
/// order that defines the cached entity. The epoch names this projector, so a
/// client never compares counters from another one (a restart, or a stack
/// loaded again).
///
/// An entity restored from a snapshot keeps the version it was saved with,
/// from an earlier epoch. Clients take a version from another epoch as newer,
/// and every change the restored entity takes is stamped in this one, so an
/// old version never follows a new one for the same key.
struct FrameVersions {
    epoch: String,
    last: AtomicU64,
}

impl FrameVersions {
    fn new() -> Self {
        // Eight hex digits: enough to tell one projector's lifetime from
        // another, and every frame carries them.
        let epoch = hex::encode(&uuid::Uuid::new_v4().as_bytes()[..4]);
        Self {
            epoch,
            last: AtomicU64::new(0),
        }
    }

    /// The next version, for a frame whose data is `data`. Mutations are
    /// objects; anything else has nowhere to carry one and goes out
    /// unversioned.
    fn next(&self, data: &Value) -> Option<Value> {
        if !data.is_object() {
            debug!("mutation is not an object; publishing it without a version");
            return None;
        }
        let counter = self.last.fetch_add(1, Ordering::Relaxed) + 1;
        Some(Value::String(format!("{}:{counter}", self.epoch)))
    }
}

/// One view's frame for a mutation, built before any view's cache write.
struct PendingFrame<'a> {
    spec: &'a ViewSpec,
    /// Which of the mutation's distinct projections the view keeps.
    projection: usize,
    /// The `_version` stamped for this view.
    version: Option<Value>,
    output: FrameOutput,
}

enum FrameOutput {
    /// A change's frame, serialized (and, for a replayable append view,
    /// already on the journal).
    Change(Arc<Bytes>),
    /// A resent whole entity's data, published only if the cache stores it.
    Whole(Value),
}

pub struct Projector {
    view_index: Arc<ViewIndex>,
    bus_manager: BusManager,
    entity_cache: EntityCache,
    mutations_rx: mpsc::Receiver<MutationBatch>,
    snapshot_runtime: Option<crate::snapshot::SnapshotRuntime>,
    journal: Option<Arc<crate::journal::EventJournal>>,
    resync: EntityResync,
    versions: FrameVersions,
    #[cfg(feature = "otel")]
    metrics: Option<Arc<Metrics>>,
}

impl Projector {
    #[cfg(feature = "otel")]
    pub fn new(
        view_index: Arc<ViewIndex>,
        bus_manager: BusManager,
        entity_cache: EntityCache,
        mutations_rx: mpsc::Receiver<MutationBatch>,
        metrics: Option<Arc<Metrics>>,
    ) -> Self {
        Self {
            view_index,
            bus_manager,
            entity_cache,
            mutations_rx,
            snapshot_runtime: None,
            journal: None,
            resync: EntityResync::new(),
            versions: FrameVersions::new(),
            metrics,
        }
    }

    #[cfg(not(feature = "otel"))]
    pub fn new(
        view_index: Arc<ViewIndex>,
        bus_manager: BusManager,
        entity_cache: EntityCache,
        mutations_rx: mpsc::Receiver<MutationBatch>,
    ) -> Self {
        Self {
            view_index,
            bus_manager,
            entity_cache,
            mutations_rx,
            snapshot_runtime: None,
            journal: None,
            resync: EntityResync::new(),
            versions: FrameVersions::new(),
        }
    }

    /// Associate projection progress with one server's snapshot lifecycle.
    pub fn with_snapshot_runtime(
        mut self,
        snapshot_runtime: crate::snapshot::SnapshotRuntime,
    ) -> Self {
        self.snapshot_runtime = Some(snapshot_runtime);
        self
    }

    /// Retain published events for replayable append subscriptions.
    pub fn with_journal(mut self, journal: Arc<crate::journal::EventJournal>) -> Self {
        self.journal = Some(journal);
        self
    }

    /// Ask the VM linked to `resync` for whole entities the cache refused.
    pub fn with_entity_resync(mut self, resync: EntityResync) -> Self {
        self.resync = resync;
        self
    }

    pub async fn run(mut self) {
        debug!("Projector started");

        let mut json_buffer = Vec::with_capacity(4096);

        while let Some(mut batch) = self.mutations_rx.recv().await {
            let mut log = CanonicalLog::new();
            log.set("phase", "projector");

            let batch_size = batch.len();
            let slot_context = batch.slot_context;
            let batch_span = debug_span!(
                parent: &batch.span,
                "projector.batch",
                batch.mutations = batch_size,
                batch.position = tracing::field::Empty,
                frames_published = tracing::field::Empty,
            );
            if let (Some(context), false) = (slot_context, batch_span.is_disabled()) {
                batch_span.record("batch.position", context.to_seq_string());
            }
            let _span_guard = batch_span.enter();
            let mut frames_published = 0u32;
            let mut errors = 0u32;

            if let Some(ctx) = batch.event_context.as_ref() {
                log.set("program", &ctx.program)
                    .set("event_kind", &ctx.event_kind)
                    .set("event_type", &ctx.event_type)
                    .set("account", &ctx.account)
                    .set("accounts_count", ctx.accounts_count);
            }

            // Keys this batch carries whole, resent or created: a refused
            // patch ahead of one needs no request of its own.
            let arriving_whole: HashSet<(String, String)> = batch
                .mutations
                .iter()
                .filter(|mutation| mutation.is_whole_entity())
                .map(|mutation| (mutation.export.clone(), Self::extract_key(&mutation.key)))
                .collect();

            for mutation in std::mem::take(&mut batch.mutations).into_iter() {
                #[cfg(feature = "otel")]
                let export = mutation.export.clone();

                match self
                    .process_mutation(mutation, slot_context, &arriving_whole, &mut json_buffer)
                    .await
                {
                    Ok(count) => frames_published += count,
                    Err(e) => {
                        error!("Failed to process mutation: {}", e);
                        errors += 1;
                    }
                }

                #[cfg(feature = "otel")]
                if let Some(ref metrics) = self.metrics {
                    metrics.record_mutation_processed(&export);
                }
            }

            // The batch is now applied to the caches: advance the snapshot
            // resume watermark, then release the processing guard transferred
            // by the VM producer. An exclusive snapshot cut cannot begin until
            // every earlier guarded batch reaches this point.
            if batch_size > 0 {
                if let Some(snapshot_runtime) = &self.snapshot_runtime {
                    snapshot_runtime.record_applied_batch(slot_context.map(|ctx| ctx.slot));
                }
            }
            drop(batch.snapshot_guard.take());

            // Flush markers remain useful to non-snapshot callers that need
            // to observe a drained projector queue.
            if let Some(ack) = batch.flush_ack.take() {
                let _ = ack.send(());
            }

            log.set("batch_size", batch_size)
                .set("frames_published", frames_published)
                .set("errors", errors);
            batch_span.record("frames_published", frames_published);

            #[cfg(feature = "otel")]
            if let Some(ref metrics) = self.metrics {
                metrics.record_projector_latency(log.duration_ms());
            }

            log.emit();
        }

        debug!("Projector stopped");
    }

    #[instrument(
        name = "projector.mutation",
        level = "debug",
        skip(self, mutation, slot_context, arriving_whole, json_buffer),
        fields(export = %mutation.export)
    )]
    async fn process_mutation(
        &self,
        mut mutation: arete_interpreter::Mutation,
        slot_context: Option<SlotContext>,
        arriving_whole: &HashSet<(String, String)>,
        json_buffer: &mut Vec<u8>,
    ) -> anyhow::Result<u32> {
        let specs = self.view_index.by_export(&mutation.export);

        if specs.is_empty() {
            return Ok(0);
        }

        let account_position = mutation
            .take_account_position()
            .map_err(anyhow::Error::msg)?;

        if mutation.is_delete() {
            let key = Self::extract_key(&mutation.key);
            let seq = slot_context.map(|ctx| ctx.to_seq_string());
            let source_domain = slot_context.map(|ctx| ctx.slot_index_domain);
            self.resync.cancel(&mutation.export, &mutation.key, &key);
            let mut published = 0;
            for spec in specs.iter().filter(|spec| spec.filters.matches(&key)) {
                let deleted = if account_position.is_none() && seq.is_none() {
                    self.entity_cache.delete_current(&spec.id, &key).await
                } else {
                    match source_domain {
                        Some(source_domain) => {
                            self.entity_cache
                                .delete_ordered_in_domain(
                                    &spec.id,
                                    &key,
                                    account_position,
                                    seq.as_deref(),
                                    source_domain,
                                )
                                .await
                        }
                        None => {
                            self.entity_cache
                                .delete_ordered(&spec.id, &key, account_position, seq.as_deref())
                                .await
                        }
                    }
                };
                if !deleted {
                    continue;
                }
                let sorted = self.view_index.sorted_caches();
                let mut caches = sorted.write().await;
                for derived in self.view_index.get_derived_views_for_source(&spec.id) {
                    if let Some(cache) = caches.get_mut(&derived.id) {
                        cache.remove_key(&key);
                    }
                }
                drop(caches);
                // Append tapes are historical events; deletion only clears
                // their current entity cache, not their retained history.
                if spec.mode == Mode::Append {
                    continue;
                }
                let frame = SourceFrame {
                    mode: spec.mode,
                    export: spec.id.clone(),
                    op: "delete",
                    key: key.clone(),
                    data: Value::Null,
                    append: Vec::new(),
                    seq: seq.clone(),
                    offset: None,
                };
                json_buffer.clear();
                serde_json::to_writer(&mut *json_buffer, &frame)?;
                self.publish_frame(
                    spec,
                    Arc::new(BusMessage {
                        key: key.clone(),
                        entity: spec.id.clone(),
                        payload: Arc::new(Bytes::copy_from_slice(json_buffer)),
                    }),
                )
                .await;
                published += 1;
            }
            return Ok(published);
        }
        let mark = mutation.take_whole_entity_mark();
        let whole = mark == Some(WholeEntity::Resent);
        let origin = match mark {
            Some(WholeEntity::Created) => PatchOrigin::Creation,
            _ => self.resync.unmarked_origin(),
        };
        let key = Self::extract_key(&mutation.key);
        let arete_interpreter::Mutation {
            export,
            key: source_key,
            mut patch,
            append,
        } = mutation;

        // The position (`_seq`) recency order sorts by: the batch's for a
        // change, the latest change's for a resend (see `EntityResync`).
        let batch_seq = slot_context.map(|ctx| ctx.to_seq_string());
        let batch_source_domain = slot_context.map(|ctx| ctx.slot_index_domain);
        let remembered = if whole {
            self.resync.take_last_change(&export, &key)
        } else {
            None
        };
        let requested_resend = remembered.is_some();
        let seq = if whole {
            remembered
                .as_ref()
                .map(|change| change.seq.clone())
                .unwrap_or(batch_seq)
        } else {
            batch_seq
        };
        let source_domain = if whole {
            remembered
                .as_ref()
                .and_then(|change| change.source_domain)
                .or(batch_source_domain)
        } else {
            batch_source_domain
        };
        let resend_account_position = remembered.and_then(|change| change.account_position);
        if let (Some(seq), Value::Object(map)) = (&seq, &mut patch) {
            map.insert("_seq".to_string(), Value::String(seq.clone()));
        }

        let matching_specs: SmallVec<[&ViewSpec; 4]> = specs
            .iter()
            .filter(|spec| spec.filters.matches(&key))
            .collect();

        let match_count = matching_specs.len();
        if match_count == 0 {
            return Ok(0);
        }
        let current_account_position = self
            .entity_cache
            .account_position(&matching_specs[0].id, &key)
            .await;
        let change_account_position = account_position.or(current_account_position);

        let mut frames_published = 0u32;
        let mut refused = false;

        // The views this mutation reaches. A whole entity is checked when it
        // is stored; a change must first pass each view's lifetime ordering.
        // Each view's ordering is its own, so checking every view before any
        // of them is written decides exactly what checking each in turn would.
        let mut accepted: SmallVec<[&ViewSpec; 4]> = SmallVec::new();
        for spec in matching_specs {
            if !whole {
                let accepts = match source_domain {
                    Some(source_domain) => {
                        self.entity_cache
                            .accepts_ordered_lifetime_mutation_in_domain(
                                &spec.id,
                                &key,
                                origin == PatchOrigin::Creation,
                                account_position,
                                seq.as_deref(),
                                source_domain,
                            )
                            .await
                    }
                    None => {
                        self.entity_cache
                            .accepts_lifetime_mutation(
                                &spec.id,
                                &key,
                                origin == PatchOrigin::Creation,
                                account_position,
                            )
                            .await
                    }
                };
                if !accepts {
                    continue;
                }
            }
            accepted.push(spec);
        }

        // Views that keep the same fields get the same projected patch, which
        // the cache then stores (or merges) once for all of them. A `_version`
        // the patch carries is dropped: each view's frame stamps its own.
        let mut projections: SmallVec<[(&Projection, Value); 1]> = SmallVec::new();
        for spec in &accepted {
            if !projections
                .iter()
                .any(|(projection, _)| projection.fields == spec.projection.fields)
            {
                projections.push((&spec.projection, Value::Null));
            }
        }
        let distinct = projections.len();
        for (index, (projection, projected)) in projections.iter_mut().enumerate() {
            let source = if index + 1 == distinct {
                std::mem::take(&mut patch)
            } else {
                patch.clone()
            };
            *projected = split_version(projection.apply(source)).0;
        }

        let mut pending: SmallVec<[PendingFrame<'_>; 4]> = SmallVec::new();
        for spec in accepted {
            let projection = projections
                .iter()
                .position(|(projection, _)| projection.fields == spec.projection.fields)
                .unwrap_or_default();
            let fields = &projections[projection].1;
            // Stamped after projection, which a field list could otherwise
            // strip it from, and before the cache write and the frame, so
            // both carry it.
            let version = self.versions.next(fields);
            let mut wire_data = with_version(fields.clone(), version.clone());
            apply_wire_format(&mut wire_data, &spec.wire_format);

            if whole {
                pending.push(PendingFrame {
                    spec,
                    projection,
                    version,
                    output: FrameOutput::Whole(wire_data),
                });
                continue;
            }

            // Replayable append views carry the offset the record is about to
            // take, so a live subscriber can checkpoint the same cursor a
            // replay would hand it. The frame is built inside the journal's
            // lock so the published offset is always the one the record gets.
            let journal = self
                .journal
                .as_ref()
                .filter(|journal| journal.is_enabled() && spec.mode == Mode::Append);

            let mut frame = SourceFrame {
                mode: spec.mode,
                export: spec.id.clone(),
                op: if mark == Some(WholeEntity::Created) && spec.mode != Mode::Append {
                    "upsert"
                } else {
                    "patch"
                },
                key: key.clone(),
                data: wire_data,
                append: append.clone(),
                seq: seq.clone(),
                offset: None,
            };

            let retained = match journal {
                Some(journal) => {
                    journal
                        .append_with(&spec.id, &key, |offset| {
                            frame.offset = Some(offset);
                            json_buffer.clear();
                            serde_json::to_writer(&mut *json_buffer, &frame)?;
                            Ok::<_, anyhow::Error>(Arc::new(Bytes::copy_from_slice(json_buffer)))
                        })
                        .await?
                }
                None => None,
            };
            let payload = match retained {
                Some((_offset, payload)) => payload,
                // No tape, or a sealed one: the event still publishes, it just
                // carries no position to resume from.
                None => {
                    frame.offset = None;
                    json_buffer.clear();
                    serde_json::to_writer(&mut *json_buffer, &frame)?;
                    Arc::new(Bytes::copy_from_slice(json_buffer))
                }
            };
            pending.push(PendingFrame {
                spec,
                projection,
                version,
                output: FrameOutput::Change(payload),
            });
        }

        if whole {
            frames_published += self
                .apply_whole_entity(
                    &key,
                    projections,
                    pending,
                    (
                        RememberedChange {
                            seq: seq.clone(),
                            source_domain,
                            account_position: resend_account_position,
                        },
                        requested_resend,
                    ),
                    json_buffer,
                )
                .await?;
        } else {
            // Every view's cache write happens before any view publishes, so a
            // subscriber reading the cache on a frame finds it written.
            let ordering = LifetimeOrdering {
                account_position,
                source_seq: seq.as_deref(),
                // Account mutations carry their authoritative position. Every
                // other source keeps its own recency cursor; instruction and
                // resolver offsets are never compared with one another.
                source_domain: account_position
                    .is_none()
                    .then_some(source_domain)
                    .flatten(),
            };
            let mut writes: SmallVec<[Option<CacheWrite>; 4]> =
                pending.iter().map(|_| None).collect();
            for (projection, (_, fields)) in projections.into_iter().enumerate() {
                let members: SmallVec<[usize; 4]> = pending
                    .iter()
                    .enumerate()
                    .filter(|(_, frame)| frame.projection == projection)
                    .map(|(index, _)| index)
                    .collect();
                let views: SmallVec<[ViewVersion<'_>; 4]> = members
                    .iter()
                    .map(|&index| ViewVersion {
                        view_id: &pending[index].spec.id,
                        version: pending[index].version.clone(),
                    })
                    .collect();
                let results = self
                    .entity_cache
                    .upsert_views(&key, fields, &views, &append, origin, ordering)
                    .await;
                for (index, write) in members.into_iter().zip(results) {
                    writes[index] = Some(write);
                }
            }

            for (frame, write) in pending.into_iter().zip(writes) {
                let (Some(write), FrameOutput::Change(payload)) = (write, frame.output) else {
                    continue;
                };
                let spec = frame.spec;
                match write {
                    CacheWrite::Refused { patch } => {
                        refused = true;
                        if spec.mode == Mode::List {
                            self.merge_into_held_derived_entities(&spec.id, &key, patch, &append)
                                .await;
                        }
                    }
                    CacheWrite::Merged | CacheWrite::Created => {
                        if spec.mode == Mode::List {
                            self.update_derived_view_caches(&spec.id, &key).await;
                        }
                    }
                }

                let message = Arc::new(BusMessage {
                    key: key.clone(),
                    entity: spec.id.clone(),
                    payload,
                });

                self.publish_frame(spec, message).await;
                frames_published += 1;

                #[cfg(feature = "otel")]
                if let Some(ref metrics) = self.metrics {
                    let mode_str = match spec.mode {
                        Mode::List => "list",
                        Mode::State => "state",
                        Mode::Append => "append",
                    };
                    metrics.record_frame_published(mode_str, &spec.export);
                }
            }
        }

        if !whole {
            self.resync.note_change(
                &export,
                &key,
                &seq,
                source_domain,
                change_account_position,
                refused,
            );
        }
        if refused && !arriving_whole.contains(&(export.clone(), key.clone())) {
            self.resync.request(&export, &source_key);
        }

        Ok(frames_published)
    }

    /// Store a whole entity the VM resent for a key the cache had to refuse,
    /// and tell list and state subscribers. `pending` carries each view's
    /// frame data, and `resend` the position of the entity's latest change,
    /// not the resend's batch.
    ///
    /// It is state, not an event: append views get only their cache entry, and
    /// neither the journal nor their tape subscribers see it. List and state
    /// views publish it as an `upsert`, which the WebSocket layer hands to a
    /// client that holds the key and uses to (re)admit it to windows.
    async fn apply_whole_entity(
        &self,
        key: &str,
        projections: SmallVec<[(&Projection, Value); 1]>,
        pending: SmallVec<[PendingFrame<'_>; 4]>,
        resend: (RememberedChange, bool),
        json_buffer: &mut Vec<u8>,
    ) -> anyhow::Result<u32> {
        let (position, requested_resend) = resend;
        let mut stored: SmallVec<[bool; 4]> = pending.iter().map(|_| false).collect();
        // A delete cancels its outstanding request, so with a linked VM a late
        // whole result with no matching request belongs to an old lifetime
        // and is not stored. Sources without a linked VM may still explicitly
        // vouch that a mutation is whole; there is no request to match.
        if requested_resend || !self.resync.is_linked() {
            let requested_source = requested_resend.then(|| {
                (
                    position.seq.as_deref(),
                    position.source_domain.unwrap_or(SlotIndexDomain::Legacy),
                )
            });
            for (projection, (_, fields)) in projections.into_iter().enumerate() {
                let members: SmallVec<[usize; 4]> = pending
                    .iter()
                    .enumerate()
                    .filter(|(_, frame)| frame.projection == projection)
                    .map(|(index, _)| index)
                    .collect();
                let views: SmallVec<[ViewVersion<'_>; 4]> = members
                    .iter()
                    .map(|&index| ViewVersion {
                        view_id: &pending[index].spec.id,
                        version: pending[index].version.clone(),
                    })
                    .collect();
                let results = self
                    .entity_cache
                    .store_whole_views(
                        key,
                        fields,
                        &views,
                        position.account_position,
                        requested_source,
                    )
                    .await;
                for (index, result) in members.into_iter().zip(results) {
                    stored[index] = result;
                }
            }
        }

        let mut published = 0;
        for (frame, stored) in pending.into_iter().zip(stored) {
            let (true, FrameOutput::Whole(wire_data)) = (stored, frame.output) else {
                continue;
            };
            let spec = frame.spec;
            match spec.mode {
                Mode::Append => continue,
                Mode::List => self.update_derived_view_caches(&spec.id, key).await,
                Mode::State => {}
            }
            let frame = SourceFrame {
                mode: spec.mode,
                export: spec.id.clone(),
                op: "upsert",
                key: key.to_string(),
                data: wire_data,
                append: Vec::new(),
                seq: position.seq.clone(),
                offset: None,
            };
            json_buffer.clear();
            serde_json::to_writer(&mut *json_buffer, &frame)?;
            let message = Arc::new(BusMessage {
                key: key.to_string(),
                entity: spec.id.clone(),
                payload: Arc::new(Bytes::copy_from_slice(json_buffer)),
            });
            self.publish_frame(spec, message).await;
            published += 1;
        }
        Ok(published)
    }

    pub(crate) fn extract_key(key: &serde_json::Value) -> String {
        key.as_str()
            .map(|s| s.to_string())
            .or_else(|| key.as_u64().map(|n| n.to_string()))
            .or_else(|| key.as_i64().map(|n| n.to_string()))
            .or_else(|| {
                key.as_array().and_then(|arr| {
                    let bytes: Vec<u8> = arr
                        .iter()
                        .filter_map(|v| v.as_u64().map(|n| n as u8))
                        .collect();
                    if bytes.len() == arr.len() {
                        Some(hex::encode(&bytes))
                    } else {
                        None
                    }
                })
            })
            .unwrap_or_else(|| key.to_string())
    }

    async fn update_derived_view_caches(&self, source_view_id: &str, entity_key: &str) {
        let derived_views = self.view_index.get_derived_views_for_source(source_view_id);
        if derived_views.is_empty() {
            return;
        }

        // The source view's own copy: derived views share its fields.
        let Some(entity) = self
            .entity_cache
            .get_shared(source_view_id, entity_key)
            .await
        else {
            return;
        };

        // Bound each derived sorted view by the source view's cache size,
        // evicting from the bottom of the sort order (see `SortedViewCache`).
        let max_entries = self.entity_cache.max_entities_per_view();
        let sorted_caches = self.view_index.sorted_caches();
        let mut caches = sorted_caches.write().await;

        // A derived view holds only the entities its filter passes, so one
        // that stops passing leaves it. Of the rest, only a view that would
        // keep the entity holds on to it: once a view is full, most updates
        // sort below its last entry.
        for spec in &derived_views {
            let Some(cache) = caches.get_mut(&spec.id) else {
                continue;
            };
            let passes = spec
                .pipeline
                .as_ref()
                .and_then(|pipeline| pipeline.filter.as_ref())
                .is_none_or(|filter| filter.matches(&entity));
            if !passes {
                cache.remove_key(entity_key);
                continue;
            }
            if cache.upsert_if_kept(entity_key, &entity, max_entries) {
                debug!(
                    "Updated sorted cache for derived view {} with key {}",
                    spec.id, entity_key
                );
            }
        }
    }

    /// Apply a patch the entity cache refused (its key was evicted there) to
    /// the derived views that still hold the whole entity.
    ///
    /// A derived view is bounded by sort position, not recency, so it can keep
    /// a top-ranked entity the entity cache has evicted; that copy is whole,
    /// so the patch merges into it exactly as it would have in the entity
    /// cache. A view that does not hold the key gets nothing: the patch alone
    /// is not an entity, and inserting it would rank a handful of changed
    /// fields as though they were the whole entity. Views that shared the
    /// entity's fields share the merged ones.
    async fn merge_into_held_derived_entities(
        &self,
        source_view_id: &str,
        entity_key: &str,
        patch: Value,
        append_paths: &[String],
    ) {
        let derived_views = self.view_index.get_derived_views_for_source(source_view_id);
        if derived_views.is_empty() {
            return;
        }
        let max_entries = self.entity_cache.max_entities_per_view();
        let sorted_caches = self.view_index.sorted_caches();
        let mut caches = sorted_caches.write().await;
        // The entities merged so far: what each view held, and the result.
        let mut merged: SmallVec<[(SharedEntity, SharedEntity); 2]> = SmallVec::new();
        for spec in &derived_views {
            let Some(cache) = caches.get_mut(&spec.id) else {
                continue;
            };
            let Some(held) = cache.get(entity_key).cloned() else {
                continue;
            };
            // The same patch merged into the same entity gives the same
            // entity: reuse it rather than merge (and copy) again.
            let entity = match merged.iter().find(|(before, _)| {
                before.shares_fields_with(&held) && before.version() == held.version()
            }) {
                Some((_, after)) => after.clone(),
                None => {
                    let mut entity = held.clone();
                    self.entity_cache
                        .merge_patch_shared(&mut entity, patch.clone(), append_paths);
                    merged.push((held, entity.clone()));
                    entity
                }
            };
            let passes = spec
                .pipeline
                .as_ref()
                .and_then(|pipeline| pipeline.filter.as_ref())
                .is_none_or(|filter| filter.matches(&entity));
            if passes {
                // Replace rather than merge: `entity` is already the whole
                // merged value, and the cache's own merge knows nothing of
                // append paths.
                cache.remove_key(entity_key);
                cache.upsert_if_kept(entity_key, &entity, max_entries);
            } else {
                cache.remove_key(entity_key);
            }
        }
    }

    #[instrument(
        name = "projector.publish",
        level = "debug",
        skip(self, spec, message),
        fields(view_id = %spec.id, mode = ?spec.mode)
    )]
    async fn publish_frame(&self, spec: &ViewSpec, message: Arc<BusMessage>) {
        match spec.mode {
            Mode::State => {
                self.bus_manager
                    .publish_state(&spec.id, &message.key, message.payload.clone())
                    .await;
            }
            Mode::List | Mode::Append => {
                self.bus_manager.publish_list(&spec.id, message).await;
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn linked() -> EntityResync {
        let resync = EntityResync::new();
        resync.link(&Arc::new(StdMutex::new(VmContext::new())));
        resync
    }

    fn seq(slot: u64) -> Option<String> {
        Some(SlotContext::new(slot, 0).to_seq_string())
    }

    /// Each refusal moves the position forward, and so does any change seen
    /// before the resend; the resend takes the position once.
    #[test]
    fn a_resend_keeps_the_latest_change_seen_before_it() {
        let resync = linked();
        let lifetime = Some(AccountPosition::new(4, 3));
        resync.note_change(
            "Round",
            "1",
            &seq(5),
            Some(SlotIndexDomain::Legacy),
            lifetime,
            true,
        );
        resync.note_change(
            "Round",
            "1",
            &seq(6),
            Some(SlotIndexDomain::Legacy),
            lifetime,
            true,
        );
        resync.note_change(
            "Round",
            "2",
            &seq(7),
            Some(SlotIndexDomain::Legacy),
            None,
            false,
        );
        assert_eq!(resync.take_last_change("Round", "2"), None, "never refused");
        resync.note_change(
            "Round",
            "1",
            &seq(8),
            Some(SlotIndexDomain::Legacy),
            None,
            false,
        );
        assert_eq!(
            resync.take_last_change("Round", "1"),
            Some(RememberedChange {
                seq: seq(8),
                source_domain: Some(SlotIndexDomain::Legacy),
                account_position: lifetime,
            })
        );
        assert_eq!(resync.take_last_change("Round", "1"), None);
    }

    #[test]
    fn remembered_positions_are_bounded_like_requests() {
        let resync = linked();
        for key in 0..=WHOLE_ENTITY_REQUEST_CAPACITY {
            resync.note_change(
                "Round",
                &key.to_string(),
                &seq(key as u64),
                Some(SlotIndexDomain::Legacy),
                None,
                true,
            );
        }
        assert_eq!(
            resync.take_last_change("Round", "0"),
            None,
            "the oldest goes"
        );
        let newest = WHOLE_ENTITY_REQUEST_CAPACITY;
        assert_eq!(
            resync.take_last_change("Round", &newest.to_string()),
            Some(RememberedChange {
                seq: seq(newest as u64),
                source_domain: Some(SlotIndexDomain::Legacy),
                account_position: None,
            })
        );
    }

    /// A source without a VM gets no resends, so nothing is remembered: a
    /// whole entity it sends is a change and takes its batch's position.
    #[test]
    fn nothing_is_remembered_without_a_vm() {
        let resync = EntityResync::new();
        resync.note_change(
            "Round",
            "1",
            &seq(5),
            Some(SlotIndexDomain::Legacy),
            None,
            true,
        );
        assert_eq!(resync.take_last_change("Round", "1"), None);
    }
}