use std::{
fmt,
io::{self, Write},
path::{Path, PathBuf},
sync::{
atomic::{AtomicBool, Ordering},
Arc, Mutex,
},
time::{Duration, Instant},
};
use axum::{
extract::{Json, State},
http::{HeaderMap, StatusCode},
response::{IntoResponse, Response},
routing::post,
Router,
};
use rhiza_core::{
ConfigChange, ConfigurationState, ExecutionProfile, LogAnchor, LogEntry, LogHash, StoredCommand,
};
use rhiza_log::{FileLogStore, LogStore};
use rhiza_quepaxa::{CertifiedDecisionInspection, DecisionProof, Membership, ThreeNodeConsensus};
use serde::{Deserialize, Serialize};
use crate::{
durability::{
adopt_finalized_successor_prestage, finalize_successor_prestage_for_stop,
inspect_successor_prestage, SuccessorPrestage, SuccessorPrestageState,
SuccessorRestorePreparation,
},
header_text, secrets_equal, valid_auth_token, valid_nonblank_header_value,
validate_profile_entry_shape, Materializer, NodeConfig, NodeRuntime, StopInformation,
RECOVERY_GENERATION_HEADER,
};
pub const CERTIFIED_TAIL_PATH: &str = "/v1/learner/certified-tail";
pub const TAIL_PROTOCOL_VERSION: &str = "1";
pub const TAIL_VERSION_HEADER: &str = "x-rhiza-tail-version";
pub const TAIL_CLUSTER_ID_HEADER: &str = "x-rhiza-tail-cluster-id";
pub const TAIL_EPOCH_HEADER: &str = "x-rhiza-tail-epoch";
pub const TAIL_CONFIG_ID_HEADER: &str = "x-rhiza-tail-config-id";
pub const TAIL_MEMBERSHIP_DIGEST_HEADER: &str = "x-rhiza-tail-membership-digest";
pub const MAX_CERTIFIED_TAIL_ENTRIES: u32 = 1_024;
pub const DEFAULT_CERTIFIED_TAIL_ENTRIES: u32 = 64;
pub const MAX_CERTIFIED_TAIL_ENCODED_BYTES: usize = 8 * 1024 * 1024;
pub const MAX_CERTIFIED_TAIL_ENTRY_PAYLOAD_BYTES: usize = 1024 * 1024;
pub const MAX_CONCURRENT_CERTIFIED_TAIL_READS: usize = 2;
pub const CERTIFIED_TAIL_READ_TIMEOUT: Duration = Duration::from_secs(10);
const TAIL_AUTHORIZATION_SCHEME: &str = "Rhiza-Tail ";
#[derive(Clone, Eq, PartialEq)]
pub struct TailReaderConfig {
cluster_id: String,
epoch: u64,
config_id: u64,
membership: Membership,
recovery_generation: u64,
token: String,
}
impl TailReaderConfig {
pub fn new(
cluster_id: impl Into<String>,
epoch: u64,
config_id: u64,
membership: Membership,
recovery_generation: u64,
token: impl Into<String>,
) -> Result<Self, TailReaderConfigError> {
let cluster_id = cluster_id.into();
if !valid_nonblank_header_value(&cluster_id) {
return Err(TailReaderConfigError::InvalidClusterId);
}
if epoch == 0 {
return Err(TailReaderConfigError::InvalidEpoch);
}
if config_id == 0 {
return Err(TailReaderConfigError::InvalidConfigId);
}
if recovery_generation == 0 {
return Err(TailReaderConfigError::InvalidRecoveryGeneration);
}
let token = token.into();
if !valid_auth_token(&token) {
return Err(TailReaderConfigError::InvalidToken);
}
Ok(Self {
cluster_id,
epoch,
config_id,
membership,
recovery_generation,
token,
})
}
pub fn cluster_id(&self) -> &str {
&self.cluster_id
}
pub const fn epoch(&self) -> u64 {
self.epoch
}
pub const fn config_id(&self) -> u64 {
self.config_id
}
pub const fn membership(&self) -> &Membership {
&self.membership
}
pub const fn membership_digest(&self) -> LogHash {
self.membership.digest()
}
pub const fn recovery_generation(&self) -> u64 {
self.recovery_generation
}
}
impl fmt::Debug for TailReaderConfig {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("TailReaderConfig")
.field("cluster_id", &self.cluster_id)
.field("epoch", &self.epoch)
.field("config_id", &self.config_id)
.field("membership_digest", &self.membership.digest())
.field("recovery_generation", &self.recovery_generation)
.field("token", &"[redacted]")
.finish()
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum TailReaderConfigError {
InvalidClusterId,
InvalidEpoch,
InvalidConfigId,
InvalidRecoveryGeneration,
InvalidToken,
ConsensusContextMismatch,
}
impl fmt::Display for TailReaderConfigError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
let message = match self {
Self::InvalidClusterId => "tail reader cluster_id must be a nonblank HTTP header value",
Self::InvalidEpoch => "tail reader epoch must be positive",
Self::InvalidConfigId => "tail reader config_id must be positive",
Self::InvalidRecoveryGeneration => "tail reader recovery_generation must be positive",
Self::InvalidToken => "tail reader token must be nonblank and contain no whitespace",
Self::ConsensusContextMismatch => {
"tail reader membership and config_id must match consensus"
}
};
formatter.write_str(message)
}
}
impl std::error::Error for TailReaderConfigError {}
#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CertifiedTailRequest {
pub from: LogAnchor,
pub max_entries: u32,
}
impl CertifiedTailRequest {
pub const fn from_anchor(from: LogAnchor) -> Self {
Self {
from,
max_entries: DEFAULT_CERTIFIED_TAIL_ENTRIES,
}
}
}
#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
pub struct CertifiedTailRecord {
pub entry: LogEntry,
pub proof: DecisionProof,
}
#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
pub struct CertifiedTailResponse {
pub records: Vec<CertifiedTailRecord>,
pub observed_tip: LogAnchor,
}
#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
#[serde(tag = "error", rename_all = "snake_case", deny_unknown_fields)]
pub enum CertifiedTailErrorResponse {
RebaseRequired { checkpoint: LogAnchor },
}
#[derive(Clone)]
struct TailReaderState {
consensus: Arc<ThreeNodeConsensus>,
log: TailLogSource,
config: TailReaderConfig,
admission: Arc<tokio::sync::Semaphore>,
}
#[derive(Clone)]
enum TailLogSource {
Shared(Arc<FileLogStore>),
Runtime(Arc<NodeRuntime>),
}
impl TailLogSource {
fn store(&self) -> &FileLogStore {
match self {
Self::Shared(log) => log,
Self::Runtime(runtime) => runtime.log_store(),
}
}
}
pub fn certified_tail_router(
consensus: Arc<ThreeNodeConsensus>,
log: Arc<FileLogStore>,
config: TailReaderConfig,
) -> Result<Router, TailReaderConfigError> {
build_certified_tail_router(consensus, TailLogSource::Shared(log), config)
}
pub fn certified_tail_router_for_runtime(
runtime: Arc<NodeRuntime>,
config: TailReaderConfig,
) -> Result<Router, TailReaderConfigError> {
let consensus = Arc::clone(runtime.consensus());
build_certified_tail_router(consensus, TailLogSource::Runtime(runtime), config)
}
fn build_certified_tail_router(
consensus: Arc<ThreeNodeConsensus>,
log: TailLogSource,
config: TailReaderConfig,
) -> Result<Router, TailReaderConfigError> {
if consensus.config_id() != config.config_id || consensus.membership() != &config.membership {
return Err(TailReaderConfigError::ConsensusContextMismatch);
}
Ok(Router::new()
.route(CERTIFIED_TAIL_PATH, post(handle_certified_tail))
.with_state(TailReaderState {
consensus,
log,
config,
admission: Arc::new(tokio::sync::Semaphore::new(
MAX_CONCURRENT_CERTIFIED_TAIL_READS,
)),
}))
}
async fn handle_certified_tail(
State(state): State<TailReaderState>,
headers: HeaderMap,
request: Result<Json<CertifiedTailRequest>, axum::extract::rejection::JsonRejection>,
) -> Response {
if !tail_reader_authenticated(&headers, &state.config) {
return StatusCode::UNAUTHORIZED.into_response();
}
let Ok(Json(request)) = request else {
return StatusCode::BAD_REQUEST.into_response();
};
if request.max_entries == 0
|| request.max_entries > MAX_CERTIFIED_TAIL_ENTRIES
|| (request.from.index() == 0 && request.from.hash() != LogHash::ZERO)
{
return StatusCode::BAD_REQUEST.into_response();
}
let permit = match state.admission.clone().try_acquire_owned() {
Ok(permit) => permit,
Err(_) => return StatusCode::TOO_MANY_REQUESTS.into_response(),
};
let cancellation = TailReadCancellation::new();
let cancelled = cancellation.signal();
let deadline = Instant::now() + CERTIFIED_TAIL_READ_TIMEOUT;
let permit = Arc::new(permit);
let worker_permit = Arc::clone(&permit);
let task = tokio::task::spawn_blocking(move || {
let _permit = worker_permit;
read_certified_tail(&state, request, &cancelled, deadline)
});
let _permit = permit;
match tokio::time::timeout(CERTIFIED_TAIL_READ_TIMEOUT, task).await {
Ok(Ok(Ok(response))) => Json(response).into_response(),
Ok(Ok(Err(TailReadError::DecisionMismatch))) => StatusCode::CONFLICT.into_response(),
Ok(Ok(Err(TailReadError::ResponseTooLarge))) => {
StatusCode::PAYLOAD_TOO_LARGE.into_response()
}
Ok(Ok(Err(TailReadError::RebaseRequired(checkpoint)))) => (
StatusCode::CONFLICT,
Json(CertifiedTailErrorResponse::RebaseRequired { checkpoint }),
)
.into_response(),
Ok(Ok(Err(TailReadError::Unavailable))) | Err(_) => {
StatusCode::SERVICE_UNAVAILABLE.into_response()
}
Ok(Err(_)) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
}
}
fn read_certified_tail(
state: &TailReaderState,
request: CertifiedTailRequest,
cancelled: &AtomicBool,
deadline: Instant,
) -> Result<CertifiedTailResponse, TailReadError> {
ensure_tail_read_active(cancelled, deadline)?;
let logical = state
.log
.store()
.logical_state()
.map_err(|_| TailReadError::Unavailable)?;
if let Some(checkpoint) = logical.anchor.as_ref().map(|anchor| *anchor.compacted()) {
if request.from.index() < checkpoint.index() {
return Err(TailReadError::RebaseRequired(checkpoint));
}
}
let observed_tip = logical
.tip
.unwrap_or_else(|| LogAnchor::new(0, LogHash::ZERO));
if request.from.index() > observed_tip.index()
|| (request.from.index() == observed_tip.index()
&& request.from.hash() != observed_tip.hash())
{
return Err(TailReadError::DecisionMismatch);
}
let available = observed_tip.index().saturating_sub(request.from.index());
let count = available.min(u64::from(request.max_entries));
let mut records = Vec::with_capacity(count as usize);
let mut encoded_bytes = encoded_json_len(&CertifiedTailResponse {
records: Vec::new(),
observed_tip,
})
.map_err(|_| TailReadError::Unavailable)?;
let mut previous = request.from;
for offset in 1..=count {
ensure_tail_read_active(cancelled, deadline)?;
let index = request
.from
.index()
.checked_add(offset)
.ok_or(TailReadError::Unavailable)?;
let Some(local) = state
.log
.store()
.read(index)
.map_err(|_| TailReadError::Unavailable)?
else {
break;
};
if local.payload.len() > MAX_CERTIFIED_TAIL_ENTRY_PAYLOAD_BYTES {
return Err(TailReadError::ResponseTooLarge);
}
if local.cluster_id != state.config.cluster_id
|| local.epoch != state.config.epoch
|| local.config_id != state.config.config_id
|| local.prev_hash != previous.hash()
{
return Err(TailReadError::DecisionMismatch);
}
let inspection = state
.consensus
.inspect_certified_decision_at(index, local.prev_hash)
.map_err(|_| TailReadError::Unavailable)?;
let CertifiedDecisionInspection::Committed(certified) = inspection else {
break;
};
ensure_tail_read_active(cancelled, deadline)?;
certified
.proof
.validate_for_cluster(
&state.config.cluster_id,
index,
state.config.epoch,
state.config.config_id,
&state.config.membership,
)
.map_err(|_| TailReadError::DecisionMismatch)?;
if certified.entry != local {
return Err(TailReadError::DecisionMismatch);
}
let record = CertifiedTailRecord {
entry: local,
proof: certified.proof,
};
let record_bytes = encoded_json_len(&record).map_err(|_| TailReadError::Unavailable)?;
let next_encoded_bytes = encoded_bytes
.checked_add(record_bytes)
.and_then(|bytes| bytes.checked_add(usize::from(!records.is_empty())))
.ok_or(TailReadError::ResponseTooLarge)?;
if next_encoded_bytes > MAX_CERTIFIED_TAIL_ENCODED_BYTES {
if records.is_empty() {
return Err(TailReadError::ResponseTooLarge);
}
break;
}
encoded_bytes = next_encoded_bytes;
let local = &record.entry;
previous = LogAnchor::new(local.index, local.hash);
records.push(record);
}
Ok(CertifiedTailResponse {
records,
observed_tip,
})
}
fn tail_reader_authenticated(headers: &HeaderMap, config: &TailReaderConfig) -> bool {
let Some(token) = header_text(headers, "authorization")
.and_then(|value| value.strip_prefix(TAIL_AUTHORIZATION_SCHEME))
else {
return false;
};
header_text(headers, TAIL_VERSION_HEADER) == Some(TAIL_PROTOCOL_VERSION)
&& header_text(headers, TAIL_CLUSTER_ID_HEADER) == Some(config.cluster_id.as_str())
&& header_text(headers, TAIL_EPOCH_HEADER) == Some(config.epoch.to_string().as_str())
&& header_text(headers, TAIL_CONFIG_ID_HEADER)
== Some(config.config_id.to_string().as_str())
&& header_text(headers, TAIL_MEMBERSHIP_DIGEST_HEADER)
== Some(config.membership.digest().to_hex().as_str())
&& header_text(headers, RECOVERY_GENERATION_HEADER)
== Some(config.recovery_generation.to_string().as_str())
&& secrets_equal(token.as_bytes(), config.token.as_bytes())
}
enum TailReadError {
DecisionMismatch,
RebaseRequired(LogAnchor),
ResponseTooLarge,
Unavailable,
}
struct TailReadCancellation(Arc<AtomicBool>);
impl TailReadCancellation {
fn new() -> Self {
Self(Arc::new(AtomicBool::new(false)))
}
fn signal(&self) -> Arc<AtomicBool> {
Arc::clone(&self.0)
}
}
impl Drop for TailReadCancellation {
fn drop(&mut self) {
self.0.store(true, Ordering::Release);
}
}
fn ensure_tail_read_active(cancelled: &AtomicBool, deadline: Instant) -> Result<(), TailReadError> {
if cancelled.load(Ordering::Acquire) || Instant::now() >= deadline {
Err(TailReadError::Unavailable)
} else {
Ok(())
}
}
#[derive(Default)]
struct EncodedByteCounter(usize);
impl Write for EncodedByteCounter {
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
self.0 = self.0.saturating_add(bytes.len());
Ok(bytes.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
fn encoded_json_len(value: &impl Serialize) -> Result<usize, serde_json::Error> {
let mut counter = EncodedByteCounter::default();
serde_json::to_writer(&mut counter, value)?;
Ok(counter.0)
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum CertifiedTailValidationError {
InvalidRequest,
TooManyRecords,
EncodedPageTooLarge,
EntryTooLarge,
RegressedObservedTip,
InvalidChain,
ForeignContext,
InvalidEntryHash,
InvalidProof,
ProofEntryMismatch,
}
impl fmt::Display for CertifiedTailValidationError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
let message = match self {
Self::InvalidRequest => "certified tail request is invalid",
Self::TooManyRecords => "certified tail response exceeds the requested bound",
Self::EncodedPageTooLarge => "certified tail response exceeds the encoded byte bound",
Self::EntryTooLarge => "certified tail response contains an oversized entry",
Self::RegressedObservedTip => {
"certified tail observed tip regresses the request anchor"
}
Self::InvalidChain => "certified tail records are not consecutive from the request",
Self::ForeignContext => "certified tail record has a foreign cluster or configuration",
Self::InvalidEntryHash => "certified tail record has an invalid entry hash",
Self::InvalidProof => "certified tail record has an invalid decision proof",
Self::ProofEntryMismatch => {
"certified tail decision proof does not identify the returned entry"
}
};
formatter.write_str(message)
}
}
impl std::error::Error for CertifiedTailValidationError {}
pub fn validate_certified_tail_response(
config: &TailReaderConfig,
request: &CertifiedTailRequest,
response: &CertifiedTailResponse,
) -> Result<(), CertifiedTailValidationError> {
if request.max_entries == 0
|| request.max_entries > MAX_CERTIFIED_TAIL_ENTRIES
|| (request.from.index() == 0 && request.from.hash() != LogHash::ZERO)
{
return Err(CertifiedTailValidationError::InvalidRequest);
}
if response.records.len() > request.max_entries as usize {
return Err(CertifiedTailValidationError::TooManyRecords);
}
if response
.records
.iter()
.any(|record| record.entry.payload.len() > MAX_CERTIFIED_TAIL_ENTRY_PAYLOAD_BYTES)
{
return Err(CertifiedTailValidationError::EntryTooLarge);
}
if encoded_json_len(response).unwrap_or(usize::MAX) > MAX_CERTIFIED_TAIL_ENCODED_BYTES {
return Err(CertifiedTailValidationError::EncodedPageTooLarge);
}
if response.observed_tip.index() < request.from.index()
|| (response.observed_tip.index() == request.from.index()
&& response.observed_tip.hash() != request.from.hash())
{
return Err(CertifiedTailValidationError::RegressedObservedTip);
}
let mut previous = request.from;
for record in &response.records {
let entry = &record.entry;
let Some(expected_index) = previous.index().checked_add(1) else {
return Err(CertifiedTailValidationError::InvalidChain);
};
if entry.index != expected_index
|| entry.prev_hash != previous.hash()
|| entry.index > response.observed_tip.index()
{
return Err(CertifiedTailValidationError::InvalidChain);
}
if entry.cluster_id != config.cluster_id
|| entry.epoch != config.epoch
|| entry.config_id != config.config_id
{
return Err(CertifiedTailValidationError::ForeignContext);
}
if entry.recompute_hash() != entry.hash {
return Err(CertifiedTailValidationError::InvalidEntryHash);
}
record
.proof
.validate_for_cluster(
&config.cluster_id,
entry.index,
config.epoch,
config.config_id,
&config.membership,
)
.map_err(|_| CertifiedTailValidationError::InvalidProof)?;
let Some(value) = record.proof.proposal().value.as_ref() else {
return Err(CertifiedTailValidationError::ProofEntryMismatch);
};
let command_hash = StoredCommand::new(entry.entry_type, entry.payload.clone()).hash();
if value.command_hash != command_hash
|| value.prev_hash != entry.prev_hash
|| value.entry_hash != entry.hash
{
return Err(CertifiedTailValidationError::ProofEntryMismatch);
}
previous = LogAnchor::new(entry.index, entry.hash);
}
if previous.index() == response.observed_tip.index()
&& previous.hash() != response.observed_tip.hash()
{
return Err(CertifiedTailValidationError::InvalidChain);
}
Ok(())
}
pub struct LearnerStore {
prestage: SuccessorPrestage,
config: TailReaderConfig,
log: FileLogStore,
materializer: Materializer,
apply_lock: Mutex<()>,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct LearnerProgress {
pub durable: LogAnchor,
pub applied: LogAnchor,
pub observed_source_tip: LogAnchor,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum LearnerStoreError {
PrestageNotPublished,
ContextMismatch,
MissingMaterializer(PathBuf),
Storage(String),
InvalidPage(CertifiedTailValidationError),
ReanchorRequired(LogAnchor),
AnchorMismatch,
InvalidStop,
Poisoned,
}
impl fmt::Display for LearnerStoreError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::PrestageNotPublished => {
formatter.write_str("learner requires a published successor prestage")
}
Self::ContextMismatch => {
formatter.write_str("learner context does not match successor prestage")
}
Self::MissingMaterializer(path) => write!(
formatter,
"published successor prestage is missing materializer {}",
path.display()
),
Self::Storage(message) => write!(formatter, "learner storage failed: {message}"),
Self::InvalidPage(error) => write!(formatter, "learner page is invalid: {error}"),
Self::ReanchorRequired(anchor) => write!(
formatter,
"learner request must restart from durable anchor {}/{}",
anchor.index(),
anchor.hash().to_hex()
),
Self::AnchorMismatch => {
formatter.write_str("learner durable and applied anchors do not match")
}
Self::InvalidStop => {
formatter.write_str("learner finalization requires the exact bound Stop")
}
Self::Poisoned => formatter.write_str("learner apply lock is poisoned"),
}
}
}
impl std::error::Error for LearnerStoreError {}
impl LearnerStore {
pub fn open(
data_dir: impl Into<PathBuf>,
config: TailReaderConfig,
) -> Result<Self, LearnerStoreError> {
let data_dir = data_dir.into();
let prestage = inspect_successor_prestage(
&data_dir,
ConfigurationState::active(config.config_id, config.membership.digest()),
)
.map_err(|error| LearnerStoreError::Storage(error.to_string()))?;
if prestage.state() != SuccessorPrestageState::Published {
return Err(LearnerStoreError::PrestageNotPublished);
}
let identity = prestage.identity();
let target_config_id = config
.config_id
.checked_add(1)
.ok_or(LearnerStoreError::ContextMismatch)?;
if identity.cluster_id() != config.cluster_id
|| identity.epoch() != config.epoch
|| identity.predecessor_config_id() != config.config_id
|| identity.predecessor_recovery_generation() != config.recovery_generation
|| identity.target_config_id() != target_config_id
{
return Err(LearnerStoreError::ContextMismatch);
}
let materializer_path = detached_materializer_path(&data_dir, identity.execution_profile());
if !materializer_path.is_file() {
return Err(LearnerStoreError::MissingMaterializer(materializer_path));
}
let initial_configuration =
ConfigurationState::active(config.config_id, config.membership.digest());
let log = FileLogStore::open_with_configuration(
data_dir.join("consensus/log"),
&config.cluster_id,
config.epoch,
initial_configuration,
)
.map_err(|error| LearnerStoreError::Storage(error.to_string()))?;
let persisted_configuration = log
.configuration_state()
.map_err(|error| LearnerStoreError::Storage(error.to_string()))?;
let materializer = Materializer::open_detached(
&data_dir,
identity.execution_profile(),
&config.cluster_id,
identity.node_id(),
config.epoch,
&persisted_configuration,
)
.map_err(|error| LearnerStoreError::Storage(error.to_string()))?;
let store = Self {
prestage,
config,
log,
materializer,
apply_lock: Mutex::new(()),
};
store.reconcile()?;
store.validate_seed()?;
Ok(store)
}
pub fn durable_anchor(&self) -> Result<LogAnchor, LearnerStoreError> {
self.log
.logical_state()
.map_err(|error| LearnerStoreError::Storage(error.to_string()))
.map(|state| state.tip.unwrap_or_else(genesis_anchor))
}
pub fn applied_anchor(&self) -> Result<LogAnchor, LearnerStoreError> {
self.materializer
.applied_tip()
.map_err(LearnerStoreError::Storage)
}
pub fn tail_request(
&self,
max_entries: u32,
) -> Result<CertifiedTailRequest, LearnerStoreError> {
if max_entries == 0 || max_entries > MAX_CERTIFIED_TAIL_ENTRIES {
return Err(LearnerStoreError::InvalidPage(
CertifiedTailValidationError::InvalidRequest,
));
}
let _guard = self
.apply_lock
.lock()
.map_err(|_| LearnerStoreError::Poisoned)?;
self.reconcile()?;
let durable = self.durable_anchor()?;
if self.applied_anchor()? != durable {
return Err(LearnerStoreError::AnchorMismatch);
}
Ok(CertifiedTailRequest {
from: durable,
max_entries,
})
}
pub fn apply_page(
&self,
request: &CertifiedTailRequest,
response: &CertifiedTailResponse,
) -> Result<LearnerProgress, LearnerStoreError> {
let _guard = self
.apply_lock
.lock()
.map_err(|_| LearnerStoreError::Poisoned)?;
validate_certified_tail_response(&self.config, request, response)
.map_err(LearnerStoreError::InvalidPage)?;
self.reconcile()?;
let durable = self.durable_anchor()?;
let applied = self.applied_anchor()?;
if durable != applied {
return Err(LearnerStoreError::AnchorMismatch);
}
if request.from != durable {
return Err(LearnerStoreError::ReanchorRequired(durable));
}
let mut configuration = self
.log
.configuration_state()
.map_err(|error| LearnerStoreError::Storage(error.to_string()))?;
let mut entries = Vec::with_capacity(response.records.len());
for record in &response.records {
configuration = validate_detached_entry(
self.prestage.identity().execution_profile(),
&self.config,
&configuration,
&record.entry,
)?;
entries.push(record.entry.clone());
}
if !entries.is_empty() {
self.log
.append_batch_buffered(&entries)
.map_err(|error| LearnerStoreError::Storage(error.to_string()))?;
self.log
.sync()
.map_err(|error| LearnerStoreError::Storage(error.to_string()))?;
}
for entry in &entries {
self.materializer
.apply_entry(entry)
.map_err(LearnerStoreError::Storage)?;
}
let durable = self.durable_anchor()?;
let applied = self.applied_anchor()?;
if durable != applied {
return Err(LearnerStoreError::AnchorMismatch);
}
Ok(LearnerProgress {
durable,
applied,
observed_source_tip: response.observed_tip,
})
}
pub fn finalize(
self,
successor_config: &NodeConfig,
stop: &StopInformation,
) -> Result<SuccessorRestorePreparation, LearnerStoreError> {
self.validate_stop(successor_config, stop)?;
let anchor = LogAnchor::new(stop.entry.index, stop.entry.hash);
if self.durable_anchor()? != anchor || self.applied_anchor()? != anchor {
return Err(LearnerStoreError::AnchorMismatch);
}
let predecessor_membership = self.config.membership.clone();
let Self {
prestage,
log,
materializer,
..
} = self;
materializer
.close_for_handoff()
.map_err(LearnerStoreError::Storage)?;
drop(log);
let finalized =
finalize_successor_prestage_for_stop(prestage, stop, &predecessor_membership)
.map_err(|error| LearnerStoreError::Storage(error.to_string()))?;
adopt_finalized_successor_prestage(
finalized,
successor_config,
stop,
&predecessor_membership,
)
.map_err(|error| LearnerStoreError::Storage(error.to_string()))
}
fn reconcile(&self) -> Result<(), LearnerStoreError> {
let durable = self.durable_anchor()?;
let mut applied = self.applied_anchor()?;
if applied.index() > durable.index()
|| (applied.index() == durable.index() && applied.hash() != durable.hash())
|| !self.log_contains(applied)?
{
return Err(LearnerStoreError::AnchorMismatch);
}
let mut configuration =
ConfigurationState::active(self.config.config_id, self.config.membership.digest());
while applied.index() < durable.index() {
let index = applied
.index()
.checked_add(1)
.ok_or(LearnerStoreError::AnchorMismatch)?;
let entry = self
.log
.read(index)
.map_err(|error| LearnerStoreError::Storage(error.to_string()))?
.ok_or(LearnerStoreError::AnchorMismatch)?;
configuration = validate_detached_entry(
self.prestage.identity().execution_profile(),
&self.config,
&configuration,
&entry,
)?;
self.materializer
.apply_entry(&entry)
.map_err(LearnerStoreError::Storage)?;
applied = LogAnchor::new(entry.index, entry.hash);
}
if self.applied_anchor()? != durable {
return Err(LearnerStoreError::AnchorMismatch);
}
Ok(())
}
fn validate_seed(&self) -> Result<(), LearnerStoreError> {
let seed = self.prestage.identity().seed_anchor();
if self.durable_anchor()?.index() < seed.index()
|| self.applied_anchor()?.index() < seed.index()
|| !self.log_contains(seed)?
{
return Err(LearnerStoreError::AnchorMismatch);
}
Ok(())
}
fn log_contains(&self, anchor: LogAnchor) -> Result<bool, LearnerStoreError> {
if anchor == genesis_anchor() {
return Ok(true);
}
let state = self
.log
.logical_state()
.map_err(|error| LearnerStoreError::Storage(error.to_string()))?;
if state
.anchor
.as_ref()
.is_some_and(|recovery| *recovery.compacted() == anchor)
{
return Ok(true);
}
self.log
.read(anchor.index())
.map_err(|error| LearnerStoreError::Storage(error.to_string()))
.map(|entry| entry.is_some_and(|entry| entry.hash == anchor.hash()))
}
fn validate_stop(
&self,
successor_config: &NodeConfig,
stop: &StopInformation,
) -> Result<(), LearnerStoreError> {
let identity = self.prestage.identity();
if stop.entry.cluster_id != self.config.cluster_id
|| stop.entry.epoch != self.config.epoch
|| stop.entry.config_id != self.config.config_id
|| stop.entry.recompute_hash() != stop.entry.hash
{
return Err(LearnerStoreError::InvalidStop);
}
let change = ConfigChange::recognize_parts(stop.entry.entry_type, &stop.entry.payload)
.map_err(|_| LearnerStoreError::InvalidStop)?;
let ConfigChange::BoundStop { successor } = change else {
return Err(LearnerStoreError::InvalidStop);
};
if successor.cluster_id() != self.config.cluster_id
|| successor.predecessor_config_id() != self.config.config_id
|| successor.predecessor_config_digest() != self.config.membership.digest()
|| successor.config_id() != identity.target_config_id()
|| successor.digest() != identity.target_membership_digest()
|| successor_config.cluster_id() != successor.cluster_id()
|| successor_config.epoch() != self.config.epoch
|| successor_config.config_id().checked_add(1) != Some(successor.config_id())
|| successor_config.membership().digest() != successor.digest()
|| successor_config.data_dir() != self.prestage.path()
{
return Err(LearnerStoreError::InvalidStop);
}
stop.proof
.validate_for_cluster(
&self.config.cluster_id,
stop.entry.index,
self.config.epoch,
self.config.config_id,
&self.config.membership,
)
.map_err(|_| LearnerStoreError::InvalidStop)?;
if !proof_matches_entry(&stop.proof, &stop.entry) {
return Err(LearnerStoreError::InvalidStop);
}
let configuration = self
.log
.configuration_state()
.map_err(|error| LearnerStoreError::Storage(error.to_string()))?;
if configuration.stop() != Some(&LogAnchor::new(stop.entry.index, stop.entry.hash)) {
return Err(LearnerStoreError::InvalidStop);
}
Ok(())
}
}
fn validate_detached_entry(
profile: ExecutionProfile,
config: &TailReaderConfig,
configuration: &ConfigurationState,
entry: &LogEntry,
) -> Result<ConfigurationState, LearnerStoreError> {
if entry.cluster_id != config.cluster_id
|| entry.epoch != config.epoch
|| entry.config_id != config.config_id
|| entry.recompute_hash() != entry.hash
{
return Err(LearnerStoreError::ContextMismatch);
}
validate_profile_entry_shape(profile, entry).map_err(LearnerStoreError::Storage)?;
configuration
.validate_entry(entry)
.map_err(|error| LearnerStoreError::Storage(error.to_string()))
}
fn detached_materializer_path(data_dir: &Path, profile: ExecutionProfile) -> PathBuf {
match profile {
ExecutionProfile::Sqlite => data_dir.join("sqlite/db.sqlite"),
ExecutionProfile::Graph => data_dir.join("ladybug/graph.lbug"),
ExecutionProfile::Kv => data_dir.join("kv/data.redb"),
}
}
const fn genesis_anchor() -> LogAnchor {
LogAnchor::new(0, LogHash::ZERO)
}
fn proof_matches_entry(proof: &DecisionProof, entry: &LogEntry) -> bool {
proof.proposal().value.as_ref().is_some_and(|value| {
value.command_hash == StoredCommand::new(entry.entry_type, entry.payload.clone()).hash()
&& value.prev_hash == entry.prev_hash
&& value.entry_hash == entry.hash
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn tail_read_admission_rejects_work_past_the_fixed_capacity() {
let admission = Arc::new(tokio::sync::Semaphore::new(
MAX_CONCURRENT_CERTIFIED_TAIL_READS,
));
let permits = (0..MAX_CONCURRENT_CERTIFIED_TAIL_READS)
.map(|_| admission.clone().try_acquire_owned().unwrap())
.collect::<Vec<_>>();
assert!(admission.clone().try_acquire_owned().is_err());
drop(permits);
assert!(admission.try_acquire_owned().is_ok());
}
#[test]
fn tail_read_stops_after_deadline_or_request_cancellation() {
let cancellation = TailReadCancellation::new();
let signal = cancellation.signal();
assert!(matches!(
ensure_tail_read_active(&signal, Instant::now()),
Err(TailReadError::Unavailable)
));
drop(cancellation);
assert!(matches!(
ensure_tail_read_active(&signal, Instant::now() + Duration::from_secs(1)),
Err(TailReadError::Unavailable)
));
}
}