nodedb-physical 0.6.1

// SPDX-License-Identifier: Apache-2.0

//! Control plane meta-operations dispatched to the Data Plane.

use std::collections::BTreeMap;

use nodedb_types::calvin::{PassiveReadKey, VersionedReadEntry};
use nodedb_types::timeseries::continuous_agg::ContinuousAggregateDef;
use nodedb_types::{QualifiedCollection, TenantId, Value};

pub use super::meta_calvin::{PassiveKey, PassiveReadKeyId};

/// Meta / maintenance physical operations.
#[derive(
    Debug,
    Clone,
    PartialEq,
    serde::Serialize,
    serde::Deserialize,
    zerompk::ToMessagePack,
    zerompk::FromMessagePack,
)]
pub enum MetaOp {
    /// WAL append (write path).
    WalAppend { payload: Vec<u8> },

    /// Cancellation signal. Data Plane stops the target at next safe point.
    Cancel {
        target_request_id: nodedb_types::id::RequestId,
    },

    /// A transaction's plans as one batch. An embedded (Lite) engine executes
    /// the sub-plans atomically. An Origin core refuses it: a committed
    /// transaction installs there only through its redo record
    /// (`ApplyTransactionRedo`).
    ///
    /// `txn_id` identifies the committing session transaction whose staging
    /// overlay holds the resolve-time bitemporal stamps an install must reuse.
    /// `None` for callers with no session overlay to consult. Wire-additive:
    /// defaults to `None` on decode of older entries.
    TransactionBatch {
        plans: Vec<super::PhysicalPlan>,
        #[serde(default)]
        txn_id: Option<nodedb_types::id::TxnId>,
    },

    /// Create a snapshot: export all engine state for this core.
    CreateSnapshot,

    /// On-demand compaction.
    Compact,

    /// Checkpoint: flush all engine state to disk, report LSN.
    Checkpoint,

    /// Register a continuous aggregate definition on this core's manager.
    RegisterContinuousAggregate { def: ContinuousAggregateDef },

    /// Remove a continuous aggregate from this core's manager.
    UnregisterContinuousAggregate { name: String },

    /// List the request tenant's continuous aggregates from this core's
    /// manager. Returns MessagePack-encoded `Vec<AggregateInfo>`.
    ListContinuousAggregates,

    /// Boot registered every continuous aggregate the catalog holds on this
    /// core. The core drops every persisted image entry of an aggregate
    /// that is not registered, and writes the images.
    ContinuousAggregatesRegistered,

    /// Wait until the refresh cut `refresh_id` of the named continuous
    /// aggregate is fully folded. `collection` is the aggregate's root
    /// source collection: the wait runs on the core that owns its vShard,
    /// where the cut applied. The core holds the request and answers it
    /// from its tick, so the fold runs under the per-tick budget while the
    /// core serves other requests. The request deadline bounds the wait.
    AwaitContinuousAggregateRefresh {
        collection: QualifiedCollection,
        name: String,
        refresh_id: u64,
    },

    /// Convert a collection's storage mode.
    ///
    /// Scans all documents in the collection and re-encodes them for the
    /// target type. The catalog update happens on the Control Plane after
    /// the Data Plane confirms success.
    ///
    /// `target_type`: "document_schemaless", "document_strict", "kv".
    /// `schema_json`: for "document_strict"/"kv", JSON-serialized column definitions.
    /// `source_storage_mode`: the collection's storage mode BEFORE this
    /// conversion, read from the catalog by the Control Plane dispatcher.
    /// The Data Plane's own `doc_configs` cache still reflects the OLD mode
    /// at dispatch time (the catalog flip and re-register happen after this
    /// op returns), so the handler cannot resolve the source format from
    /// that cache. It must take it from the plan instead.
    ConvertCollection {
        collection: QualifiedCollection,
        target_type: String,
        schema_json: String,
        source_storage_mode: super::document::StorageMode,
    },

    /// Snapshot a tenant's data from the sparse engine.
    /// Returns serialized `(documents, indexes)` as JSON payload.
    CreateTenantSnapshot {
        tenant_id: u64,
        /// `Some(W)` asks the node that receives the plan to take a backup's
        /// consistent cut at watermark `W` before it snapshots: every write
        /// committed below `W` has its final outcome there first. The
        /// receiving Control Plane takes the cut and clears the field. The
        /// Data Plane never reads it.
        #[serde(default)]
        cut_watermark: Option<u64>,
        /// `Some` asks the receiving node to take the cut with barriers that
        /// capture the request's tenants, and to answer with the captures it
        /// parked instead of a snapshot. Requires `cut_watermark`. The Data
        /// Plane never reads it.
        #[serde(default)]
        cut_capture: Option<super::meta_snapshot::CutCaptureRequest>,
        /// Export every array cell version. Only a backup and a Raft group
        /// snapshot read them, so every other snapshot skips the export.
        #[serde(default)]
        arrays: bool,
    },

    /// Restore a tenant's data across all engines from a snapshot.
    /// `snapshot` is a MessagePack-serialized `TenantDataSnapshot`.
    ///
    /// `replace_mode` selects the collision policy for engines whose restore is
    /// otherwise fail-closed (columnar / vector / flushed-timeseries):
    /// - `false` (user RESTORE): refuse to overwrite keys already present.
    /// - `true` (Raft InstallSnapshot apply): OVERWRITE present keys, because a
    ///   Raft install must replace local state, not fail against it.
    RestoreTenantSnapshot {
        tenant_id: u64,
        snapshot: Vec<u8>,
        replace_mode: bool,
        /// Collections this core clears before it installs, pre-resolved by
        /// the Raft snapshot applier from the local catalog. Empty = no clear.
        #[serde(default)]
        collections_to_clear: Vec<super::meta_snapshot::SnapshotClearTarget>,
        /// The data group's vShards, whose records the install replaces: the
        /// array cells with the snapshot's `arrays`, and every graph edge
        /// with an endpoint home among them with the snapshot's edges. With
        /// it set, `collections_to_clear` keeps graph edges: an edge lives
        /// on its endpoints' vShards, not on its collection's home. Empty =
        /// no group install: no array replace, and collection clears take
        /// their edges too.
        #[serde(default)]
        group_vshards: Vec<u32>,
        /// The version each of the group's vShards holds once the install
        /// lands: the log position of the snapshot's cut. The installed rows
        /// carry no per-row versions, so a read older than the cut no longer
        /// validates on this node. Empty = no group install.
        #[serde(default)]
        version_floor: Vec<nodedb_types::ShardVersion>,
    },

    /// Purge ALL data for a tenant across every engine and cache.
    ///
    /// Deletes documents, indexes, vectors, graph edges, timeseries,
    /// KV entries, CRDT state, inverted index terms, and cache entries.
    /// Idempotent: safe to re-run after a crash.
    PurgeTenant { tenant_id: u64 },

    /// Purge a single collection across every engine on this core.
    ///
    /// Runs the per-collection equivalent of `PurgeTenant`: each
    /// engine (columnar, vector, FTS, spatial, document/strict/kv,
    /// graph, CRDT, timeseries) reclaims its L1 segment files,
    /// memtables, in-flight compactions, and snapshot references
    /// for this one collection, then a WAL `CollectionTombstoned`
    /// record is appended so replay after purge does not resurrect
    /// rows. The quiesce-drain primitive (see `bridge::quiesce`) is
    /// invoked first so in-flight scans complete before file unlink.
    ///
    /// `purge_lsn` is the metadata-raft commit LSN at which the
    /// `PurgeCollection` entry committed; used by the WAL tombstone
    /// filter and surfaced in the audit trail.
    ///
    /// Idempotent: re-running after partial completion picks up
    /// where the previous run left off.
    UnregisterCollection {
        tenant_id: u64,
        name: String,
        purge_lsn: u64,
        /// Whether this core must reclaim the collection's shared on-disk L1
        /// checkpoint/partition files. Those paths are keyed by
        /// `(database, tenant, collection)` — not by core — so the Control
        /// Plane sets this `true` for exactly one core (the collection's
        /// homing vshard) in the all-cores fan-out. Every core still evicts
        /// its own per-core in-memory state; only the shared-path unlink /
        /// `remove_dir_all` is single-cored, so concurrent cores cannot race
        /// on the same tree.
        reclaim_l1_files: bool,
    },

    /// Reclaim every local Data Plane resource for a single
    /// materialized view. Mirrors `UnregisterCollection` one level
    /// up: an MV has its own columnar segment files (populated by the
    /// CDC refresh loop) that outlive the MV's catalog row when the
    /// MV is dropped, unless reclaim runs on every follower. Drops
    /// the MV's in-memory refresh state + unlinks its segment files.
    ///
    /// Idempotent: missing in-memory state is a no-op; missing files
    /// are a no-op. Runs on every node.
    UnregisterMaterializedView { tenant_id: u64, name: String },

    /// Estimate the on-core data size (in bytes) for a single
    /// `(tenant_id, collection)` pair. Sums per-engine in-memory
    /// state: KV hash-table bytes, columnar flushed-segment byte
    /// count, vector-index byte count, and sparse-redb document
    /// range. Response payload is a u64 LE byte count.
    ///
    /// Used by `_system.dropped_collections.size_bytes_estimate` to
    /// surface "how much storage will a hard-delete reclaim?"
    /// without waiting for a purge cycle.
    QueryCollectionSize { tenant_id: u64, name: String },

    /// Walk a `HASH_CHAIN` collection's chain in install order from genesis
    /// over its raw stored rows. Read-only. The response payload is a
    /// MessagePack `ChainVerdict` naming the first break, if any.
    VerifyHashChain { collection: QualifiedCollection },

    /// Enforce retention on a timeseries collection: drop segments older than
    /// the cutoff. Called by the retention policy enforcement loop.
    ///
    /// `aggregates` lists every continuous aggregate the catalog roots at
    /// `collection`. Retention folds an expiring partition into those
    /// aggregates before it removes it, so the core refuses the pass while
    /// one of them is not registered on it at that incarnation. The loop
    /// retries on its next pass.
    EnforceTimeseriesRetention {
        collection: QualifiedCollection,
        max_age_ms: i64,
        aggregates: Vec<super::timeseries::AggregateIncarnation>,
    },

    /// Bitemporal audit-retention purge for the graph edge store.
    ///
    /// Deletes versioned edge rows (in `edge_store/temporal` layout) where
    /// `system_from_ms < cutoff_system_ms` AND a newer version of the same
    /// base key exists. Never deletes the single surviving (latest) version
    /// of a base key, even if it is older than the cutoff — "audit retain"
    /// reclaims only *superseded* history, not the current row.
    /// Emits one `RecordType::TemporalPurge` WAL record per purged batch.
    TemporalPurgeEdgeStore {
        tenant_id: u64,
        collection: QualifiedCollection,
        cutoff_system_ms: i64,
    },

    /// Bitemporal audit-retention purge for the DocumentStrict versioned
    /// tables (`documents_versioned` + `indexes_versioned`). Same semantics
    /// as `TemporalPurgeEdgeStore`: drop versions older than `cutoff_system_ms`
    /// when a newer version of the same `(tenant, coll, doc_id)` exists;
    /// never drop the single surviving version. Deletes index-version rows
    /// keyed to the purged document versions in the same transaction.
    TemporalPurgeDocumentStrict {
        tenant_id: u64,
        collection: QualifiedCollection,
        cutoff_system_ms: i64,
    },

    /// Bitemporal audit-retention purge for plain columnar collections.
    /// Reuses the timeseries max-system-ts axis on partition meta: any
    /// partition whose `max_system_ts < cutoff_system_ms` and whose max
    /// column version has been superseded by a later partition is removed.
    /// Non-bitemporal columnar collections are a no-op (collection is
    /// expected to be flagged bitemporal by the caller).
    ///
    /// `aggregates` lists every continuous aggregate the catalog roots at
    /// `collection`, as in `EnforceTimeseriesRetention`. A timeseries
    /// collection's purge removes partitions, so the core refuses it while
    /// one of them is not registered on it at that incarnation. A plain
    /// columnar collection roots no aggregate, and its list is empty.
    TemporalPurgeColumnar {
        tenant_id: u64,
        collection: QualifiedCollection,
        cutoff_system_ms: i64,
        aggregates: Vec<super::timeseries::AggregateIncarnation>,
    },

    /// Bitemporal audit-retention purge for CRDT (Loro-backed) collections.
    /// Drops archived row versions whose `_ts_system < cutoff_system_ms`
    /// from the per-collection bitemporal history sibling. Never removes
    /// the live row, so `AS OF now()` reads remain correct post-purge.
    TemporalPurgeCrdt {
        tenant_id: u64,
        collection: QualifiedCollection,
        cutoff_system_ms: i64,
    },

    /// Bitemporal audit-retention purge for the array engine.
    ///
    /// Drops superseded tile versions (those with `system_from_ms <
    /// cutoff_system_ms` where a newer version of the same tile key
    /// exists) from the array's bitemporal storage. The single surviving
    /// (latest) version of each tile is never removed. Arrays are
    /// globally-scoped — `tenant_id` carries the sentinel value `0`.
    TemporalPurgeArray {
        tenant_id: u64,
        /// Global array id from `ArrayCatalogEntry::name`. Arrays are
        /// not yet tenant-scoped.
        array_id: String,
        cutoff_system_ms: i64,
    },

    /// Alter the bitemporal retention policy of an array.
    ///
    /// All real work (catalog rewrite + registry update) is done on the
    /// Control Plane before this op is emitted. The Data Plane handler
    /// returns an 8-byte LE u64 acknowledgement (new `audit_retain_ms`
    /// if set, or 0). This variant exists so the alter command travels
    /// through the standard plan dispatch path and its permission is
    /// classified as `Admin`.
    ///
    /// Double-`Option` semantics:
    /// - `None`          = field omitted from SET clause; do not change.
    /// - `Some(None)`    = SET to NULL (unregister from retention registry).
    /// - `Some(Some(v))` = SET to v.
    AlterArray {
        /// Global array id (name). Arrays are not yet tenant-scoped.
        array_id: String,
        audit_retain_ms: Option<Option<i64>>,
        minimum_audit_retain_ms: Option<Option<u64>>,
    },

    /// Apply retention to continuous aggregate buckets managed by
    /// the aggregate manager. Drops materialized buckets older than
    /// each aggregate's configured retention_period_ms.
    ApplyContinuousAggRetention,

    /// Query the watermark for a named continuous aggregate.
    /// Returns JSON-serialized `WatermarkState`.
    QueryAggregateWatermark { aggregate_name: String },

    /// Query all entries from a collection's last-value cache.
    /// Returns JSON-serialized `Vec<(u64, i64, f64)>` — (series_id, ts, value).
    QueryLastValues { collection: QualifiedCollection },

    /// Query a single series from a collection's last-value cache.
    /// Returns JSON-serialized `Option<(i64, f64)>` — (ts, value).
    QueryLastValue {
        collection: QualifiedCollection,
        series_id: u64,
    },

    /// Calvin deterministic executor: static-set multi-shard transaction.
    ///
    /// The Calvin scheduler dispatches this variant after lock acquisition for
    /// transactions whose read/write set is fully known at submission time (the
    /// common case). Only the vShard's data-group leader dispatches it. The
    /// Data Plane handler validates the read-set and stages `plans` without
    /// mutating base. Once the global verdict is commit, the scheduler
    /// resolves the staged plans into a redo record (`CalvinResolve`) and
    /// proposes it to the data group. Every replica installs it from the log
    /// (`ApplyTransactionRedo`).
    ///
    /// NOTE: This variant occupies the same msgpack positional tag as the
    /// original `CalvinExecute` variant it replaces, preserving wire
    /// compatibility with any in-flight log entries from before the rename.
    CalvinExecuteStatic {
        /// Sequencer epoch this transaction belongs to.
        epoch: u64,
        /// Zero-based position within the epoch batch.
        position: u32,
        /// Tenant scope for all plans in this batch.
        tenant_id: TenantId,
        /// Physical plans to execute atomically.
        plans: Vec<super::PhysicalPlan>,
        /// Wall-clock ms read once on the sequencer leader at epoch creation.
        /// Used by engine handlers as the deterministic time anchor (bitemporal
        /// sys_from, KV TTL expire_at, timeseries system_ms) instead of reading
        /// the wall clock independently. Wire-additive: defaults to 0 on decode
        /// of older entries.
        epoch_system_ms: i64,
        /// The transaction's LSN-versioned read-set from the replicated `TxClass`.
        /// Each participant checks its own vShard's reads at apply and records
        /// whether they were still current, without gating apply. Wire-additive:
        /// defaults to empty on decode of older entries.
        #[serde(default)]
        versioned_reads: Vec<VersionedReadEntry>,
        /// Indexes into `plans` of the plans a trigger body buffered. Their
        /// rows commit under `Trigger`, so they fire no trigger. Empty for a
        /// transaction no body joined.
        #[serde(default)]
        body_plans: Vec<u32>,
    },

    /// Calvin dependent-read executor: a passive participant reads its keys
    /// and returns their values for broadcast.
    ///
    /// The data-group leader of each passive vShard dispatches it once the
    /// transaction holds its locks there. The Data Plane handler reads each
    /// key from base storage and returns a zerompk-encoded
    /// `Vec<(PassiveReadKeyId, Value)>`: the stored bytes as `Value::Bytes`,
    /// `Value::Null` for an absent row. A successful read votes commit. The
    /// scheduler proposes the payload as a `ReplicatedWrite::CalvinReadResult`
    /// to the data group of every active vShard.
    CalvinExecutePassive {
        /// Sequencer epoch this transaction belongs to.
        epoch: u64,
        /// Zero-based position within the epoch batch.
        position: u32,
        /// Tenant scope.
        tenant_id: TenantId,
        /// Keys to read on this passive vshard.
        keys_to_read: Vec<PassiveReadKey>,
    },

    /// Calvin dependent-read executor: active participant writes with
    /// injected read values.
    ///
    /// Dispatched by the scheduler to active vshards (those holding write
    /// keys) once all passive read results have been received. The
    /// `injected_reads` map is keyed by `PassiveReadKeyId` and carries the
    /// values broadcast by the passive participants.
    ///
    /// OLLP verification: before writing, the handler checks whether the
    /// predicate match in the txn's read set still matches the actual rows
    /// in the engine. If mismatched, returns `OllpRetryRequired` status
    /// and does NOT write. The OLLP orchestrator on the Control Plane
    /// interprets this status and retries via `Inbox::submit`.
    CalvinExecuteActive {
        /// Sequencer epoch this transaction belongs to.
        epoch: u64,
        /// Zero-based position within the epoch batch.
        position: u32,
        /// Tenant scope.
        tenant_id: TenantId,
        /// Physical plans to execute atomically.
        plans: Vec<super::PhysicalPlan>,
        /// Read values injected from passive participants.
        /// `BTreeMap` for deterministic iteration order (determinism contract).
        injected_reads: BTreeMap<PassiveReadKeyId, Value>,
        /// Wall-clock ms read once on the sequencer leader at epoch creation.
        /// Same semantics as `CalvinExecuteStatic::epoch_system_ms`.
        epoch_system_ms: i64,
    },

    /// Rebuild a collection's indexes (HNSW, full-text, graph CSR) on
    /// this core.
    ///
    /// `index_name` narrows the rebuild to one kind: `hnsw`, `fts` or
    /// `csr`. `None` rebuilds every kind the collection has.
    ///
    /// Each rebuild runs off the core while the core keeps serving reads
    /// and writes. Writes made during the build are replayed onto the
    /// rebuilt index at cutover, and the core swaps it in on a later tick.
    /// With `concurrent = true` the core answers once the rebuilds have
    /// started. With `concurrent = false` it answers once they have cut
    /// over, or `DeadlineExceeded` at the request deadline; the rebuilds
    /// still complete.
    ///
    /// Errors when a rebuild of the collection already runs
    /// (`ObjectNotInPrerequisiteState`), when a rebuild cannot start, or,
    /// with `concurrent = false`, when a rebuild is discarded.
    RebuildIndex {
        collection: QualifiedCollection,
        index_name: Option<String>,
        concurrent: bool,
    },

    /// Persist a synonym group to the FTS backend on this core.
    ///
    /// Called after the Control Plane has already written to the catalog and
    /// updated the in-memory registry. The Data Plane handler writes the group
    /// to the `FtsIndex` meta store so query-time expansion can find it.
    PutSynonymGroup {
        tenant_id: u64,
        /// Serialized `SynonymGroupRecord` (JSON).
        record_json: String,
    },

    /// Remove a synonym group from the FTS backend on this core.
    ///
    /// Called after the Control Plane has already removed it from the catalog.
    DeleteSynonymGroup { tenant_id: u64, name: String },

    /// Execute a point write at STATEMENT time by STAGING it into the
    /// per-transaction overlay, instead of buffering it for COMMIT.
    ///
    /// `plan` is a single point-write `DocumentOp` (`PointPut` / `PointInsert`
    /// / `PointDelete` / `PointUpdate`). The Data Plane handler evaluates the
    /// write against BASE ∪ OVERLAY: it raises constraint violations
    /// (unique / primary-key) immediately, computes the real affected-row
    /// count, and records the resulting body (or tombstone) in the overlay so
    /// a subsequent same-transaction read-modify-write observes it. It does
    /// NOT make the write durable — COMMIT resolves the overlay into the
    /// transaction's redo record, and the redo install remains the sole
    /// durable apply. Keyed by the request's `txn_id`.
    StageWrite { plan: Box<super::PhysicalPlan> },

    /// Drop the per-transaction staging overlay for a completed (committed
    /// or rolled-back) transaction on this core.
    ///
    /// The overlay (`crate::data::executor::handlers::transaction::overlay::TxnOverlay`)
    /// holds not-yet-durable writes staged during a transaction; once the
    /// transaction resolves, its overlay entry must be released so it does
    /// not leak across the `CoreLoop`'s lifetime. `HashMap::remove` on an
    /// absent key is a no-op, so this is safe to dispatch even when no
    /// overlay was ever populated for the given `txn_id`.
    DropTxnOverlay { txn_id: nodedb_types::id::TxnId },

    /// Mark a savepoint in the per-transaction staging overlays.
    ///
    /// A core holds one value/TTL, one GRAPH and one ARRAY overlay per
    /// transaction, shared by every vShard it hosts. The core records the
    /// three undo-journal lengths under `savepoint`. The Control Plane sends
    /// the mark through each vShard the transaction has staged to, so a core
    /// can receive it more than once. The core keeps its first record: no
    /// write stages between those sends. `savepoint` is unique for the
    /// transaction and grows with each mark. In-memory only, no WAL.
    MarkSavepoint {
        txn_id: nodedb_types::id::TxnId,
        savepoint: u64,
    },

    /// Roll the per-transaction staging overlays back to a savepoint.
    ///
    /// Replays each overlay's undo journal in reverse down to the length the
    /// core recorded under `savepoint`, restoring each prior slot or removing
    /// it when absent, then truncates the journal there. A core with no
    /// record had no staged vShard at the mark, so its overlays rewind to
    /// empty. The core drops its records of later savepoints. The
    /// transaction stays open. In-memory only.
    RollbackToSavepoint {
        txn_id: nodedb_types::id::TxnId,
        savepoint: u64,
    },

    /// Discard the staged writes of a Calvin transaction.
    ///
    /// Dispatched by the leader's scheduler when the slice ends with no log
    /// entry (an abort verdict, or a COMMIT for a slice with no write), and
    /// by a demoted leader for state it staged. The handler removes the
    /// staged plans keyed by `(epoch, position)` and fires nothing — no base
    /// mutation, no side effects. Absent key is an idempotent no-op.
    CalvinDrop { epoch: u64, position: u32 },

    /// Resolve a committing transaction's staged writes into ONE replayable
    /// redo record, WITHOUT mutating base.
    ///
    /// The Data Plane handler reads the per-transaction staging overlay
    /// (`CoreLoop::txn_overlays`) by shared reference and, for every staged
    /// post-image, emits the engine-native WAL sub-record shape that engine's
    /// autocommit path already produces. The encoded `RedoRecord` bytes are
    /// returned in the response payload; the Control Plane appends them as a
    /// single `RecordType::TransactionRedo` WAL record and a later install phase
    /// replays them. No base engine is touched during resolve — a crash before
    /// install loses only an unacknowledged commit.
    ///
    /// `plans` is the transaction's buffered write set: it is walked to classify
    /// every op and to discover which collections' overlay entries to serialize.
    /// An op class whose resolve serializer is not yet built raises a typed
    /// error rather than being silently omitted (a dropped op class would lose
    /// those rows on install). Wire-additive: appended last so older log entries
    /// decode unchanged.
    ResolveTxn {
        txn_id: nodedb_types::id::TxnId,
        plans: Vec<super::PhysicalPlan>,
    },

    /// Resolve a staged Calvin transaction's write plans into ONE replayable
    /// redo record, WITHOUT mutating base.
    ///
    /// Mirrors `ResolveTxn` exactly, but reads Calvin's own staging state
    /// instead of a session transaction's: the plans buffered in the core's
    /// `commit_pending` under `(epoch, position, vshard)` and the per-core
    /// staging overlay written under the corresponding synthetic `TxnId`
    /// (see `calvin_synthetic_txn_id`). Dispatched by the scheduler once the
    /// global verdict is commit. The scheduler proposes the record this op
    /// returns as a stamped `TransactionRedo` entry, and
    /// `ApplyTransactionRedo` installs it. No base engine is touched during
    /// resolve.
    ///
    /// The response payload is a zerompk-encoded [`super::CalvinResolved`]:
    /// the redo record and the reply the staged plans decided.
    CalvinResolve { epoch: u64, position: u32 },

    /// Apply one committed transaction's resolved redo record on the core that
    /// owns its vShard.
    ///
    /// Every replica runs this from the vShard's data-group Raft log, in log
    /// order, and installs the same post-images. The write funnel appends
    /// `redo` to this node's WAL as one `TransactionRedo` record before the
    /// dispatch, so restart replay reproduces the apply.
    ///
    /// `redo` is the zerompk-encoded redo record. `collections` names every
    /// collection the transaction wrote; each gets a collection-floor write
    /// version at the record's LSN. `sum_targets` is the materialized-sum
    /// resolution the transaction's document writes fold into their targets.
    /// `origin` decides which commit-boundary checks the apply runs.
    ///
    /// `calvin` is set when the record is a committed Calvin slice. The
    /// install then stamps the slice's ordinal on the core clock, consumes
    /// the slice's staged state when this core staged it, and answers the
    /// reply `calvin.reply` describes.
    ApplyTransactionRedo {
        redo: Vec<u8>,
        collections: Vec<String>,
        sum_targets: Vec<super::RedoSumTargets>,
        origin: super::RedoOrigin,
        calvin: Option<super::CalvinInstall>,
    },

    /// Install one batch of a RESTORE as part of a Calvin transaction, on
    /// the vShard the batch names.
    ///
    /// It stages nothing. The transaction's resolve appends the batch's rows
    /// and edge versions to its redo record, each edge version applied at
    /// the transaction's ordinal, and its flush installs the record as a
    /// RESTORE. Its write keys are the rows and edges the batch writes.
    RestoreRedo(Box<super::RestoredRedo>),

    /// Report the current version of each home in `probes`.
    ///
    /// A read-only transaction's commit sends this to the leader of the
    /// vShards its cross-shard graph reads observed, one request per leader.
    /// The leader hands each core the probes whose vShard that core owns. The
    /// core answers each probe with the collection's write floor on the core,
    /// or with the core watermark for a probe with no collection. The payload
    /// is a msgpack array of `HomeVersion`. It reads nothing else and writes
    /// nothing.
    HomeVersions {
        probes: Vec<super::HomeVersionProbe>,
    },
}