1macro_rules! unstable_internal_modules {
126 ($($name:ident),* $(,)?) => {
127 $(
128 #[cfg(feature = "unstable-internals")]
129 pub mod $name;
130 #[cfg(not(feature = "unstable-internals"))]
131 #[allow(dead_code, unused_imports)]
132 mod $name;
133 )*
134 };
135}
136
137unstable_internal_modules! {
138 adversarial_transport,
139 concurrency_fuzzing,
140 crash_injection,
141 replay_fuzzing,
142 permutation_scheduler,
143 persistence_reality,
144 in_process_transport,
145 query_execution_diagnostics,
146 workload_diagnostics,
147 cardinality_endpoint,
148}
149
150#[cfg(test)]
151mod acceptance_tests;
152mod acquisition;
153pub mod admission;
154#[cfg(test)]
155mod admission_contract_tests;
156pub mod analytics;
157pub mod application;
158pub mod application_runtime;
159pub mod lifecycle_store;
160#[cfg(not(target_arch = "wasm32"))]
161pub mod authority;
162pub mod authority_failover;
163pub mod authorization;
164#[cfg(test)]
165mod authorization_security_tests;
166pub mod capabilities;
167mod capability;
168mod cardinality_diagnostics;
169#[cfg(test)]
170mod causal_backlog_bound;
171pub mod causal_dependency_barrier;
172#[cfg(test)]
173mod causal_dependency_barrier_phase_7_1;
174pub mod consistency_contract;
175mod content_distribution;
176pub mod convergence;
177pub mod crash_atomic_boundary;
178#[cfg(test)]
179mod crash_recovery_tests;
180mod cron;
181#[cfg(test)]
182mod dedup_bound_investigation;
183pub mod distributed_indexing;
184#[cfg(test)]
185mod distributed_tests;
186pub mod distributed_transactions;
187pub mod durability_guarantees;
188pub mod durable_dedup_set;
189#[cfg(test)]
190mod durable_operation_identity;
191pub mod durable_operation_log;
192pub mod durable_sync;
193pub mod equality_index;
194mod execution;
195pub mod indexing;
196#[cfg(test)]
197mod managed_cas_tests;
198mod materialization;
199pub mod metrics;
200pub mod multi_node_convergence;
201pub mod multi_operation_transaction;
202pub mod observability;
203mod operation;
204pub mod operation_algebra;
205pub mod operation_log;
206#[cfg(test)]
207mod p1_application_atomicity;
208#[cfg(test)]
209mod p1_atomicity_acceptance;
210pub mod partition_reconciliation;
211mod peer_registry;
212#[cfg(test)]
213mod phase1b_acceptance;
214#[cfg(test)]
215mod phase1c1_acceptance;
216#[cfg(test)]
217mod phase1c2_acceptance;
218#[cfg(test)]
219mod phase1c3_acceptance;
220#[cfg(test)]
221mod phase1c_atomicity_proof;
222#[cfg(test)]
223mod phase5_integration;
224#[cfg(test)]
225mod phase5_scenarios;
226#[cfg(test)]
227mod phase6_adversarial_scenarios;
228#[cfg(test)]
229mod phase6_convergence_validator;
230#[cfg(test)]
231mod phase6_persistence;
232#[cfg(test)]
233#[cfg(test)]
234mod phase_1c_real_tcp;
235#[cfg(test)]
236mod phase_2a_failures;
237#[cfg(test)]
238mod phase_2b_network;
239#[cfg(test)]
240mod phase_2c_cascading;
241#[cfg(test)]
242mod phase_3_durability;
243#[cfg(test)]
244mod phase_4_baseline;
245#[cfg(test)]
246mod phase_5_soak;
247pub mod policy_evaluation;
248pub mod production_api;
249mod provenance;
250pub mod query_performance;
251#[cfg(test)]
252mod read_basis;
253mod references;
254pub mod replica_acknowledgements;
255pub mod replica_membership;
256pub mod replication_manager;
257pub mod replication_protocol;
258mod routing;
259pub mod semantic_decision;
260pub mod semantic_query;
261pub mod sharding;
262pub mod agent_state;
263pub mod agent_transition_contract;
264pub mod state_conflict_contract;
265pub mod state_contract;
266pub mod state_diff_contract;
267pub mod state_facade;
268pub mod state_hash;
269pub mod state_model;
270pub mod state_transition_store;
271pub mod state_trigger;
272mod storage;
273pub mod submission;
274mod sync;
275pub mod sync_contract;
276#[cfg(not(target_arch = "wasm32"))]
277pub mod tcp_transport;
278pub mod transaction_api;
279#[cfg(test)]
280mod transaction_invariants;
281#[cfg(test)]
282mod transaction_preconditions;
283pub mod transactions;
284mod trigger;
285pub mod trigger_contract;
286pub mod versioned_object_substrate;
287pub mod worker_mesh;
288mod workflow;
289#[cfg(test)]
290mod workflow_acceptance_tests;
291mod workflow_integration;
292mod workflow_runtime;
293pub mod workload;
294
295pub use acquisition::{
296 AcquisitionDeduplicator, AcquisitionPolicy, AcquisitionResult, ContentRequest, ContentResponse,
297 ContentTransferError,
298};
299pub use admission::{ProductionRetryPolicy, RejectionReason, SubmissionOutcome};
300#[cfg(feature = "unstable-internals")]
301pub use adversarial_transport::{
302 AdversarialTransport, ConnectionId, ConnectivityMatrix, FaultSeed, InMemoryTransport,
303 LinkState, MessageId, NetworkEvent, NetworkSchedule, NodeId, ReplicationTransport,
304};
305pub use analytics::{
306 AutoTuner, IndexAnalytics, IndexMetrics, PerformanceReport, QueryTrace, TuningRecommendation,
307};
308pub use capability::{
309 CapabilityAccessModel, CapabilityCheckpoint, CapabilityContext, CapabilityExecutionContext,
310 CapabilityExecutionResult, CapabilityId, CapabilityLifecycleState, CapabilityMetadata,
311 CapabilityOperationResult, CapabilityQuery, CapabilityQueryResult, CapabilityRef,
312 CapabilityRegistry, CapabilityRequirements, CapabilityResolver, CapabilityRole,
313 DistributedCapabilityAdvertisement, ExecutionRef, FlowCapability, StateRequirement,
314};
315pub use causal_dependency_barrier::{
316 ApplyResult, CausalDependencyBarrier, OperationState, PendingOperation, ReceiveResult,
317 StateCounters,
318};
319pub use consistency_contract::{
320 CommutativeVerificationResult, ConsensusEnforcementResult, ConsensusRequirementChecker,
321 ConsistencyContractEnforcer, DeterministicMergeVerificationResult,
322 NoDowngradeVerificationResult, OperationApplication,
323};
324pub use content_distribution::{
325 CapabilityLocationRegistry, ContentAcquisitionManager, ContentAddressableStore,
326 DistributedReferenceResolver, ProjectionDistribution,
327};
328pub use convergence::{
329 CausalHistory, ConflictResolution, ConflictResolutionStrategy, MergeResult, StateVersion,
330 VectorClock,
331};
332pub use crash_atomic_boundary::{
333 CrashAtomicExecutor, CrashBoundaryPoint, CrashRecoveryState, InvariantCheckResult,
334 RecoveryDecision,
335};
336pub use cron::{CronSchedule, CronScheduler};
337pub use distributed_indexing::{
338 ConsistencyLevel, ConsistencyProtocol, DistributedIndexManager, DistributedSyncStatus,
339 IndexOperation, IndexReplicationState, IndexSyncMessage, IndexVectorClock, RemoteIndexMetadata,
340 RemoteIndexState, ReplicationStatus, VersionConflict,
341};
342pub use distributed_transactions::{
343 DistributedTransactionExecutor, EnvelopeId, ReplicaState, ReplicationMessage,
344 TransactionEnvelope,
345};
346pub use equality_index::{EqualityIndex, EqualityIndexStats, IndexKey};
347pub use execution::{Execution, ExecutionQueue, ExecutionStatus, RetryPolicy};
348pub use indexing::{IndexConfig, IndexManager, IndexStats, IndexType};
349pub use materialization::{MaterializationHandler, ReactiveCollectionIntegration};
350pub use multi_node_convergence::{
351 ConvergenceAggregation, ConvergenceResult, MultiNodeConvergenceSimulator, NodeExecutionResult,
352};
353pub use operation::{MutationOrder, Operation, OperationType, DEFAULT_MUTATION_ORDER};
354pub use operation_algebra::{
355 ConflictType, ConvergenceSemantic, OperationAlgebra, OperationAlgebraRegistry,
356};
357pub use operation_log::{MemoryOperationLog, OperationLogEntry};
358pub use partition_reconciliation::{
359 DeterminismCheckResult, PartitionDeterminismChecker, PartitionTrace, PartitionedReplica,
360 ReconciliationResult, ReplicaPartition,
361};
362pub use peer_registry::PeerRegistry;
363#[cfg(feature = "unstable-internals")]
364pub use permutation_scheduler::{OperationSchedule, PermutationScheduler, ScheduleStrategy};
365#[cfg(feature = "unstable-internals")]
366pub use persistence_reality::{
367 CorruptionDetectionResult, DurablePersistenceLayer, FsyncDurabilityResult,
368 PersistenceContractVerifier, PersistenceCrashPoint, RecoveryResult, WALEntry, WriteResult,
369};
370pub use provenance::{
371 AcquisitionMetadata, DerivedState, DerivedStateProvenance, DerivedValue, OperationSource,
372};
373pub use references::{
374 ContentIdentity, ContentRange, FlowRef, PeerAdvertisement, PeerId, ReferenceResolver,
375 ResolutionMetadata, ResolutionResult,
376};
377pub use routing::{
378 CapabilityRouter, ExecutionClaim, ExecutionClaimer, ExecutionResultPublisher, PeerStateVersion,
379 RouteDecision,
380};
381pub use semantic_decision::{
382 authorize_context, configured_decision_runtime, decide_record, decision_cache_key,
383 deterministic_request_key, reuse_record_judgment,
384 execute_decision_batch, execute_decision_transport, execute_inline_decision_transport, jev,
385 run_choice_diagnostics, run_decision_benchmark, validate_decision_definition,
386 validate_decision_result, BinaryDecisionResult, ChoiceDecisionResult, ChoiceDiagnosticsReport,
387 ChoiceExpansionObservation, ChoicePermutationObservation, DecisionAuthorizationConfig,
388 DecisionAuthorizationDecision, DecisionAuthorizationGate, DecisionBatch, DecisionBatchDecision,
389 DecisionBatchEvaluation, DecisionBenchmarkCase, DecisionBenchmarkReport, DecisionContextPacket,
390 DecisionDefinition, DecisionError, DecisionEvaluation, DecisionEvidence, DecisionEvidenceField,
391 DecisionExecutionMethod, DecisionJudgmentRecord, DecisionKind, DecisionOptions,
392 DecisionPrincipal, DecisionProvenance, DecisionRecordRef, DecisionRequest, DecisionResult,
393 DecisionRuntime, DecisionRuntimeCapabilities, DecisionRuntimeMetadata, DecisionStateReference,
394 DecisionStatus, DecisionTransportAuthorization, DecisionTransportExecutionRequest,
395 DecisionTransportExecutionResponse, DecisionTransportRuntime, DeterministicDecisionRuntime,
396 ScoreDecisionResult, WeightedOption, DEFAULT_DECISION_COLLECTION,
397};
398pub use semantic_query::{
399 compose_semantic_page, semantic_clause_hash, semantic_definition_hash,
400 validate_semantic_clause, SemanticAnnotation, SemanticCandidate, SemanticOrderDirection,
401 SemanticPage, SemanticQueryClause, SemanticQueryError, SemanticQueryFilter,
402 SemanticQueryMode, SemanticQueryOrder, SemanticStageSummary, SemanticUnevaluated,
403 SemanticUnevaluatedReason, MAX_SEMANTIC_CANDIDATES,
404};
405pub use sharding::{
406 HotspotAlert, RebalanceOperation, ShardDistributionSummary, ShardId, ShardKey, ShardManager,
407 ShardMetrics, ShardRange, ShardingStrategy,
408};
409pub use state_conflict_contract::{
410 classify_states, StateConflictInput, StateConflictOutput, STATE_CONFLICT_CONTRACT_ID,
411 STATE_CONFLICT_CONTRACT_VERSION,
412};
413pub use state_diff_contract::{
414 diff_states, StateDiffInput, StateDiffOutput, STATE_DIFF_CONTRACT_ID,
415 STATE_DIFF_CONTRACT_VERSION,
416};
417pub use state_facade::FeltDBStateSystem;
418pub use state_hash::{CanonicalState, StateHash};
419pub use state_model::{
420 apply_reconciliation_plan, path_relation, paths_overlap, reconcile, resolve_path, ChangeKind,
421 ConflictClass, ConflictClassification, GarbageCollectionError, GarbageCollectionReport,
422 MaterializationError, MaterializationProjection, PathComponent, PathConflict, PathRelation,
423 ReconciliationPlan, RefError, Relationship, SemanticChange, SemanticDiff, StateId, StateObject,
424 StateReconciliationResult, StateRef, StateRevision, StateStore, StateTopology,
425 STATE_MODEL_VERSION,
426};
427pub use state_transition_store::{MemoryStateTransitionStore, StateTransitionRecord};
428pub use state_trigger::{
429 StateTrigger, StateTriggerChangeKind, StateTriggerError, StateTriggerPredicate,
430 StateTriggerStore, TriggeredWorkload,
431};
432pub use storage::{CheckpointData, FileStorage, MemoryStorage, Storage};
433pub use submission::{SubmissionManager, SubmissionMetrics};
434pub use sync::{ChangeLog, Conflict, ConflictDetector, PeerState, SyncMessage, SyncState};
435pub use transaction_api::{TransactionManager, TxError, TxResult};
436pub use transactions::{
437 CommitId, CommitValidator, ConsistencyContract, OperationCommand, OperationId, StateDependency,
438 StateTransition, TransactionExecutor, TransitionResult,
439};
440pub use trigger::{Trigger, TriggerFilter, TriggerRegistry};
441pub use trigger_contract::{
442 canonical_input_bytes, canonical_output_bytes, evaluate_canonical, evaluate_trigger,
443 ContractChange, ContractChangeKind, ContractErrorBody, ContractErrorCode, ContractPredicate,
444 ContractRuntime, ContractTrigger, MatchOutcome, MatchReason, TriggerMatchInput,
445 TriggerMatchOutput, TRIGGER_MATCH_CONTRACT_ID, TRIGGER_MATCH_CONTRACT_VERSION,
446};
447pub use versioned_object_substrate::{
448 DurableObjectStore, DurableRef, LocalObjectStore, Reachability, S3CompatibleBucket,
449 S3CompatibleObjectStore, SubstrateError, VersionedObjectSubstrate,
450};
451pub use workflow::{
452 BlockedReason, WorkflowGraph, WorkflowInstance, WorkflowNode, WorkflowOperation, WorkflowRef,
453 WorkflowState, WorkflowStepRef, WorkflowStepState,
454};
455pub use workflow_integration::{
456 WorkflowCronTrigger, WorkflowIntegration, WorkflowProvenance, WorkflowTrigger,
457};
458pub use workflow_runtime::{WorkflowOperationEntry, WorkflowRuntime};
459
460use crate::state_contract::AuthorizationContext;
461use serde::de::DeserializeOwned;
462use serde::{Deserialize, Serialize};
463use serde_json::Value;
464use sha2::{Digest, Sha256};
465use std::any::type_name;
466use std::collections::hash_map::DefaultHasher;
467use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque};
468use std::fmt::{Display, Formatter};
469use std::fs::{self, File, OpenOptions};
470use std::hash::{Hash, Hasher};
471use std::io::{BufRead, BufReader, Write};
472use std::ops::Bound::{Excluded, Unbounded};
473use std::path::{Path, PathBuf};
474use std::sync::{Arc, Mutex, OnceLock, Weak};
475use std::time::{SystemTime, UNIX_EPOCH};
476use tokio::sync::broadcast;
477
478pub type Result<T> = std::result::Result<T, FlowError>;
479
480#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
482#[serde(rename_all = "snake_case")]
483pub enum PageAccess {
484 KeyRange,
486 EqualityIndex,
488 Scan,
490}
491
492#[derive(Debug, Clone)]
494pub struct SelectedPage {
495 pub rows: Vec<StoredRow>,
497 pub has_more: bool,
499 pub examined: usize,
501 pub matched_examined: usize,
503 pub access: PageAccess,
504}
505
506impl SelectedPage {
507 fn empty(access: PageAccess) -> Self {
508 SelectedPage { rows: Vec::new(), has_more: false, examined: 0, matched_examined: 0, access }
509 }
510}
511
512#[derive(Debug)]
519#[non_exhaustive]
520pub enum FlowError {
521 Io(std::io::Error),
522 Serde(serde_json::Error),
523 CorruptLogLine(Box<LogCorruption>),
524 CapabilityError(String),
525 PreconditionFailed(Box<PreconditionFailure>),
531 TransactionIdPayloadMismatch {
539 transaction_id: String,
540 },
541 RevisionConflict {
543 expected: u64,
544 actual: u64,
545 },
546 BackupRejected(Box<BackupProblem>),
548 IncompatibleFormat(Box<FormatIncompatibility>),
554 Durability(Box<DurabilityFailure>),
561}
562
563#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
565#[serde(rename_all = "snake_case")]
566#[non_exhaustive]
567pub enum DurabilityFailureKind {
568 StorageFull,
571 WriteFailed,
573 SyncFailed,
576 Indeterminate,
580 RecoveryRequired,
583}
584
585impl DurabilityFailureKind {
586 pub fn code(self) -> &'static str {
588 match self {
589 DurabilityFailureKind::StorageFull => "STORAGE_FULL",
590 DurabilityFailureKind::WriteFailed => "STORAGE_WRITE_FAILED",
591 DurabilityFailureKind::SyncFailed => "STORAGE_SYNC_FAILED",
592 DurabilityFailureKind::Indeterminate => "DURABILITY_FAILURE",
593 DurabilityFailureKind::RecoveryRequired => "RECOVERY_REQUIRED",
594 }
595 }
596}
597
598#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
600pub struct DurabilityFailure {
601 pub kind: DurabilityFailureKind,
602 pub committed: Option<bool>,
605 pub detail: String,
607}
608
609impl DurabilityFailure {
610 pub fn code(&self) -> &'static str {
611 self.kind.code()
612 }
613}
614
615impl Display for DurabilityFailure {
616 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
617 let outcome = match self.committed {
618 Some(false) => "nothing was committed",
619 Some(true) => "the write was committed",
620 None => "whether the write was committed is unknown until the database is reopened",
621 };
622 write!(f, "{}: {}; {outcome}", self.code(), self.detail)
623 }
624}
625
626impl Display for FlowError {
627 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
628 match self {
629 FlowError::Io(e) => write!(f, "io error: {e}"),
630 FlowError::Serde(e) => write!(f, "serde error: {e}"),
631 FlowError::CorruptLogLine(corruption) => write!(f, "CORRUPT_DURABLE_LOG: {corruption}"),
632 FlowError::CapabilityError(msg) => write!(f, "capability error: {msg}"),
633 FlowError::PreconditionFailed(failure) => write!(f, "PRECONDITION_FAILED: {failure}"),
634 FlowError::TransactionIdPayloadMismatch { transaction_id } => write!(
635 f,
636 "IDEMPOTENCY_CONFLICT: transaction {transaction_id} was already committed with a different payload"
637 ),
638 FlowError::RevisionConflict { expected, actual } => write!(
639 f,
640 "REVISION_CONFLICT: expected authority revision {expected}, current revision is {actual}"
641 ),
642 FlowError::IncompatibleFormat(incompatibility) => {
643 write!(f, "INCOMPATIBLE_DURABLE_FORMAT: {incompatibility}")
644 }
645 FlowError::BackupRejected(problem) => write!(f, "BACKUP_REJECTED: {problem}"),
646 FlowError::Durability(failure) => write!(f, "{failure}"),
647 }
648 }
649}
650
651impl std::error::Error for FlowError {}
652
653impl From<std::io::Error> for FlowError {
654 fn from(value: std::io::Error) -> Self {
655 FlowError::Io(value)
656 }
657}
658
659impl From<serde_json::Error> for FlowError {
660 fn from(value: serde_json::Error) -> Self {
661 FlowError::Serde(value)
662 }
663}
664
665impl From<String> for FlowError {
666 fn from(value: String) -> Self {
667 FlowError::CapabilityError(value)
668 }
669}
670
671#[derive(Clone)]
672pub struct FeltDb {
673 inner: Arc<Mutex<Inner>>,
674 event_tx: broadcast::Sender<ChangeEvent>,
675 _ownership: Arc<File>,
676 history: Arc<std::sync::RwLock<VersionHistory>>,
680}
681
682struct OpenDatabase {
683 inner: Weak<Mutex<Inner>>,
684 event_tx: broadcast::Sender<ChangeEvent>,
685 ownership: Weak<File>,
686 history: Weak<std::sync::RwLock<VersionHistory>>,
687}
688
689static OPEN_DATABASES: OnceLock<Mutex<HashMap<PathBuf, OpenDatabase>>> = OnceLock::new();
690
691#[derive(Default)]
692#[allow(dead_code)] struct Inner {
694 history: VersionHistory,
697 path: PathBuf,
698 rows: HashMap<String, BTreeMap<String, StoredRow>>,
699 key_counters: HashMap<String, u64>,
700 query_count_by_type: HashMap<String, usize>,
701 adaptive_indexes: HashSet<String>,
702 instance_id: String,
703 sequence: u64,
704 authority_revision: u64,
707 sync_state: SyncState,
708 change_log: ChangeLog,
709 execution_queue: ExecutionQueue,
710 trigger_registry: TriggerRegistry,
711 cron_scheduler: CronScheduler,
712 capability_registry: CapabilityRegistry,
713 peer_registry: PeerRegistry,
714 content_store: ContentAddressableStore,
715 capability_locations: CapabilityLocationRegistry,
716 bootstrapped: bool,
717 applied_transactions: HashSet<String>,
718 transaction_payload_hashes: HashMap<String, String>,
719 transaction_revisions: HashMap<String, (u64, u64)>,
720 transaction_attributions: VecDeque<TransactionAttribution>,
733 transaction_history: Vec<TransactionAttribution>,
737 collection_cardinality: HashMap<String, u64>,
738 equality_index: EqualityIndex,
743 log_recovery: LogRecovery,
748 durable_format: DurableFormat,
751 durability_mode: DurabilityMode,
753 writes_since_barrier: u32,
755 write_fault: Option<String>,
760 log_bytes_after_rewrite: u64,
765 compaction_policy: CompactionPolicy,
767 pruned_since_rewrite: usize,
769 revision_heads: HashMap<String, (state_model::StateId, u64)>,
776 collection_versions: HashMap<String, u64>,
785 collection_clock: u64,
786}
787
788struct StateGuard<'a> {
803 inner: std::sync::MutexGuard<'a, Inner>,
804 held_since: Option<std::time::Instant>,
805}
806
807impl std::ops::Deref for StateGuard<'_> {
808 type Target = Inner;
809 fn deref(&self) -> &Inner {
810 &self.inner
811 }
812}
813
814impl std::ops::DerefMut for StateGuard<'_> {
815 fn deref_mut(&mut self) -> &mut Inner {
816 &mut self.inner
817 }
818}
819
820impl Drop for StateGuard<'_> {
821 fn drop(&mut self) {
822 if let Some(since) = self.held_since {
823 workload_diagnostics::record(
824 workload_diagnostics::Phase::StateLockHold,
825 since.elapsed().as_nanos().min(u64::MAX as u128) as u64,
826 );
827 }
828 }
829}
830
831impl Inner {
847 fn put_row(&mut self, row: StoredRow) -> bool {
852 let capability = row.capability.clone();
853 let key = row.key.clone();
854 self.bump_collection_version(&capability);
855 let indexed = self.equality_index.indexes_collection(&capability);
856 let mutation = workload_diagnostics::span(workload_diagnostics::Phase::RecordMutation);
857 let bucket = self.rows.entry(capability.clone()).or_default();
858 let previous = bucket.insert(key.clone(), row);
859 drop(mutation);
860 if previous.is_none() {
861 *self
862 .collection_cardinality
863 .entry(capability.clone())
864 .or_default() += 1;
865 }
866 if !indexed {
867 return previous.is_none();
868 }
869 let _index = workload_diagnostics::span(workload_diagnostics::Phase::IndexMaintenance);
873 if let Some(previous) = &previous {
877 self.equality_index
878 .remove_record(&capability, &key, &previous.value);
879 }
880 let Inner {
883 rows,
884 equality_index,
885 ..
886 } = self;
887 if let Some(current) = rows.get(&capability).and_then(|rows| rows.get(&key)) {
888 equality_index.insert_record(&capability, &key, ¤t.value);
889 }
890 previous.is_none()
891 }
892
893 fn remove_row(&mut self, capability: &str, key: &str) -> bool {
895 let mutation = workload_diagnostics::span(workload_diagnostics::Phase::RecordMutation);
896 let Some(removed) = self
897 .rows
898 .get_mut(capability)
899 .and_then(|bucket| bucket.remove(key))
900 else {
901 return false;
902 };
903 drop(mutation);
904 self.bump_collection_version(capability);
905 let count = self
906 .collection_cardinality
907 .entry(capability.to_string())
908 .or_default();
909 *count = count.saturating_sub(1);
910 let _index = workload_diagnostics::span(workload_diagnostics::Phase::IndexMaintenance);
911 self.equality_index
912 .remove_record(capability, key, &removed.value);
913 true
914 }
915
916 fn bump_collection_version(&mut self, capability: &str) {
918 self.collection_clock += 1;
919 self.collection_versions
920 .insert(capability.to_string(), self.collection_clock);
921 }
922
923 fn clear_rows(&mut self) {
925 let cleared: Vec<String> = self.rows.keys().cloned().collect();
926 for capability in cleared {
927 self.bump_collection_version(&capability);
928 }
929 self.rows.clear();
930 self.collection_cardinality.clear();
931 self.equality_index.clear();
932 }
933}
934
935#[derive(Debug, Clone, Serialize, Deserialize)]
936pub struct AtomicMutation {
937 pub capability: String,
938 pub key: String,
939 pub value: Option<Value>,
940}
941#[derive(Debug, Clone, Serialize, Deserialize)]
942pub struct AtomicPrecondition {
943 pub capability: String,
944 pub key: String,
945 pub expected_version: Option<u64>,
946}
947
948#[derive(Debug, Clone, Default, Serialize, Deserialize)]
966pub struct RecordPrecondition {
967 pub capability: String,
968 pub key: String,
969 #[serde(default)]
971 pub require_absent: bool,
972 #[serde(default)]
974 pub expected_version: Option<u64>,
975 #[serde(default)]
977 pub expected_epoch: Option<u64>,
978 #[serde(default)]
980 pub expected_lease_id: Option<String>,
981}
982
983impl RecordPrecondition {
984 pub fn is_empty(&self) -> bool {
988 !self.require_absent
989 && self.expected_version.is_none()
990 && self.expected_epoch.is_none()
991 && self.expected_lease_id.is_none()
992 }
993}
994
995#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
1001pub struct ReadSet {
1002 #[serde(default)]
1003 pub records: Vec<ReadRecord>,
1004 #[serde(default)]
1005 pub predicates: Vec<ReadPredicate>,
1006}
1007
1008#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
1009pub struct ReadRecord {
1010 pub collection: String,
1011 pub id: String,
1012}
1013
1014#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
1015pub struct ReadPredicate {
1016 pub collection: String,
1017 #[serde(rename = "where")]
1018 pub conditions: Vec<ReadCondition>,
1019}
1020
1021#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
1022pub struct ReadCondition {
1023 pub field: String,
1024 pub eq: Value,
1025}
1026
1027#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
1033pub struct ReadBasis {
1034 pub reads: ReadSet,
1035 pub token: String,
1036}
1037
1038pub(crate) const COLLECTION_FENCE_KEY: &str = "collection-version";
1040
1041pub(crate) struct ConsistentRows {
1044 pub(crate) view: HashMap<String, BTreeMap<String, StoredRow>>,
1045 pub(crate) versions: HashMap<String, u64>,
1046}
1047
1048pub const READ_BASIS_PREFIX: &str = "fr1:";
1050pub const READ_SET_LIMIT: usize = 1_000;
1052
1053impl ReadSet {
1054 pub fn normalised(&self) -> Result<ReadSet> {
1057 let valid = |segment: &str| {
1058 !segment.is_empty()
1059 && segment.len() <= 128
1060 && segment
1061 .bytes()
1062 .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'))
1063 };
1064 let mut records: BTreeMap<(String, String), ReadRecord> = BTreeMap::new();
1065 for record in &self.records {
1066 if !valid(&record.collection) || !valid(&record.id) {
1067 return Err(FlowError::CapabilityError(format!(
1068 "INVALID_READ_SET:invalid read {}:{}",
1069 record.collection, record.id
1070 )));
1071 }
1072 records.insert(
1073 (record.collection.clone(), record.id.clone()),
1074 record.clone(),
1075 );
1076 }
1077 let mut predicates: BTreeMap<(String, String), ReadPredicate> = BTreeMap::new();
1078 for predicate in &self.predicates {
1079 if !valid(&predicate.collection) || predicate.conditions.is_empty() {
1080 return Err(FlowError::CapabilityError(format!(
1081 "INVALID_READ_SET:a predicate over {} must name at least one field",
1082 predicate.collection
1083 )));
1084 }
1085 let mut conditions = predicate.conditions.clone();
1086 if conditions
1087 .iter()
1088 .any(|condition| condition.field.is_empty())
1089 {
1090 return Err(FlowError::CapabilityError(
1091 "INVALID_READ_SET:a predicate condition requires a field".into(),
1092 ));
1093 }
1094 conditions.sort_by(|left, right| left.field.cmp(&right.field));
1095 let identity = serde_json::to_string(&conditions).unwrap_or_default();
1096 predicates.insert(
1097 (predicate.collection.clone(), identity),
1098 ReadPredicate {
1099 collection: predicate.collection.clone(),
1100 conditions,
1101 },
1102 );
1103 }
1104 if records.is_empty() && predicates.is_empty() {
1105 return Err(FlowError::CapabilityError(
1106 "INVALID_READ_SET:a read set that names nothing fences nothing".into(),
1107 ));
1108 }
1109 if records.len() + predicates.len() > READ_SET_LIMIT {
1110 return Err(FlowError::CapabilityError(format!(
1111 "INVALID_READ_SET:a read set may name at most {READ_SET_LIMIT} records and predicates"
1112 )));
1113 }
1114 Ok(ReadSet {
1115 records: records.into_values().collect(),
1116 predicates: predicates.into_values().collect(),
1117 })
1118 }
1119}
1120
1121#[derive(Debug, Clone, PartialEq, Eq)]
1126pub enum ReadLayout {
1127 Direct,
1129 Namespace(String),
1131}
1132
1133impl ReadLayout {
1134 fn capability(&self, collection: &str) -> String {
1135 match self {
1136 Self::Direct => collection.to_string(),
1137 Self::Namespace(namespace) => format!("{namespace}:{collection}"),
1138 }
1139 }
1140
1141 fn key(&self, collection: &str, id: &str) -> String {
1142 match self {
1143 Self::Direct => format!("{collection}:{id}"),
1144 Self::Namespace(_) => id.to_string(),
1145 }
1146 }
1147
1148 fn scope_tag(&self) -> String {
1151 match self {
1152 Self::Direct => "d".into(),
1153 Self::Namespace(namespace) => {
1154 let digest = format!(
1155 "{:x}",
1156 Sha256::digest(format!("namespace\0{namespace}").as_bytes())
1157 );
1158 format!("n{}", &digest[..16])
1159 }
1160 }
1161 }
1162
1163 fn token_prefix(&self) -> String {
1164 format!("{READ_BASIS_PREFIX}{}:", self.scope_tag())
1165 }
1166}
1167
1168fn read_basis_token(
1172 rows: &HashMap<String, BTreeMap<String, StoredRow>>,
1173 reads: &ReadSet,
1174 layout: &ReadLayout,
1175) -> String {
1176 let live = |capability: &str, key: &str| {
1177 rows.get(capability)
1178 .and_then(|bucket| bucket.get(key))
1179 .filter(|row| !row.deleted)
1180 .map(|row| row.value.clone())
1181 };
1182 let records: Vec<Value> = reads
1183 .records
1184 .iter()
1185 .map(|record| {
1186 let value = live(
1187 &layout.capability(&record.collection),
1188 &layout.key(&record.collection, &record.id),
1189 );
1190 serde_json::json!([record.collection, record.id, value])
1191 })
1192 .collect();
1193 let predicates: Vec<Value> = reads
1194 .predicates
1195 .iter()
1196 .map(|predicate| {
1197 let matching: Vec<Value> = rows
1198 .get(&layout.capability(&predicate.collection))
1199 .map(|bucket| {
1200 bucket
1201 .iter()
1202 .filter(|(_, row)| !row.deleted)
1203 .filter(|(_, row)| {
1204 predicate.conditions.iter().all(|condition| {
1205 row.value.get(&condition.field) == Some(&condition.eq)
1206 })
1207 })
1208 .map(|(key, row)| serde_json::json!([key, row.value]))
1209 .collect()
1210 })
1211 .unwrap_or_default();
1212 serde_json::json!([predicate.collection, predicate.conditions, matching])
1213 })
1214 .collect();
1215 let observed = serde_json::json!({ "records": records, "predicates": predicates });
1216 format!(
1217 "{}{:x}",
1218 layout.token_prefix(),
1219 Sha256::digest(serde_json::to_vec(&observed).unwrap_or_default())
1220 )
1221}
1222
1223#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
1229#[serde(rename_all = "camelCase", tag = "predicate")]
1230pub enum PreconditionFailure {
1231 Present { collection: String, key: String },
1233 Missing { collection: String, key: String },
1235 Version {
1236 collection: String,
1237 key: String,
1238 expected: u64,
1239 actual: u64,
1240 },
1241 Epoch {
1242 collection: String,
1243 key: String,
1244 expected: u64,
1245 actual: u64,
1246 },
1247 Lease {
1248 collection: String,
1249 key: String,
1250 expected: String,
1251 actual: Option<String>,
1252 },
1253 Fence {
1258 collection: String,
1259 key: String,
1260 #[serde(skip_serializing, default)]
1261 expected: String,
1262 #[serde(skip_serializing, default)]
1263 actual: String,
1264 },
1265 History {
1272 collection: String,
1273 key: String,
1274 expected: Option<String>,
1275 actual: String,
1276 },
1277}
1278
1279impl PreconditionFailure {
1280 pub fn collection(&self) -> &str {
1281 match self {
1282 Self::Present { collection, .. }
1283 | Self::Missing { collection, .. }
1284 | Self::Version { collection, .. }
1285 | Self::Epoch { collection, .. }
1286 | Self::Lease { collection, .. }
1287 | Self::Fence { collection, .. }
1288 | Self::History { collection, .. } => collection,
1289 }
1290 }
1291 pub fn key(&self) -> &str {
1293 match self {
1294 Self::Present { key, .. }
1295 | Self::Missing { key, .. }
1296 | Self::Version { key, .. }
1297 | Self::Epoch { key, .. }
1298 | Self::Lease { key, .. }
1299 | Self::Fence { key, .. }
1300 | Self::History { key, .. } => key,
1301 }
1302 }
1303}
1304
1305impl PreconditionFailure {
1306 pub fn public_json(&self, layout: &ReadLayout) -> Value {
1316 let mut value = serde_json::to_value(self).unwrap_or(Value::Null);
1317 if matches!(self, Self::Fence { .. }) {
1318 return value;
1319 }
1320 let (collection, id) = match layout {
1321 ReadLayout::Direct => {
1322 let collection = self.collection().to_string();
1323 let id = self
1324 .key()
1325 .strip_prefix(&format!("{collection}:"))
1326 .unwrap_or(self.key())
1327 .to_string();
1328 (collection, id)
1329 }
1330 ReadLayout::Namespace(namespace) => (
1331 self.collection()
1332 .strip_prefix(&format!("{namespace}:"))
1333 .unwrap_or(self.collection())
1334 .to_string(),
1335 self.key().to_string(),
1336 ),
1337 };
1338 if let Some(object) = value.as_object_mut() {
1339 object.insert("key".into(), Value::String(format!("{collection}:{id}")));
1340 object.insert("collection".into(), Value::String(collection));
1341 object.insert("id".into(), Value::String(id));
1342 }
1343 value
1344 }
1345}
1346
1347impl Display for PreconditionFailure {
1348 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1349 match self {
1350 Self::Present { collection, key } => {
1351 write!(f, "{collection}/{key} was required to be absent but exists")
1352 }
1353 Self::Missing { collection, key } => write!(f, "{collection}/{key} does not exist"),
1354 Self::Version {
1355 collection,
1356 key,
1357 expected,
1358 actual,
1359 } => write!(
1360 f,
1361 "{collection}/{key} is at version {actual}, expected {expected}"
1362 ),
1363 Self::Epoch {
1364 collection,
1365 key,
1366 expected,
1367 actual,
1368 } => write!(
1369 f,
1370 "{collection}/{key} is at epoch {actual}, expected {expected}"
1371 ),
1372 Self::Lease {
1373 collection,
1374 key,
1375 expected,
1376 actual,
1377 } => write!(
1378 f,
1379 "{collection}/{key} holds lease {actual:?}, expected {expected}"
1380 ),
1381 Self::Fence { collection, .. } => write!(
1382 f,
1383 "state read by this transaction has changed (read basis over {collection})"
1384 ),
1385 Self::History { collection, key, expected: Some(expected), actual } => write!(
1386 f,
1387 "{collection}/{key}: the version was issued in database history {expected}, not this one ({actual}); read the record again"
1388 ),
1389 Self::History { collection, key, expected: None, actual } => write!(
1390 f,
1391 "{collection}/{key}: a numeric version names no database history, and this database was restored (history {actual}); fence with a version token read from it"
1392 ),
1393 }
1394 }
1395}
1396
1397#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
1399#[serde(rename_all = "snake_case")]
1400pub enum HistoryOrigin {
1401 #[default]
1404 Genesis,
1405 Restored,
1409}
1410
1411#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
1421pub struct VersionHistory {
1422 pub id: String,
1424 pub origin: HistoryOrigin,
1425}
1426
1427pub const VERSION_TOKEN_PREFIX: &str = "fv1";
1429
1430impl VersionHistory {
1431 fn new(origin: HistoryOrigin) -> Self {
1432 Self { id: format!("{:032x}", rand::random::<u128>()), origin }
1433 }
1434
1435 pub fn token(&self, version: u64) -> String {
1437 format!("{VERSION_TOKEN_PREFIX}:{}:{version}", self.id)
1438 }
1439}
1440
1441pub fn parse_version_token(token: &str) -> Option<(&str, u64)> {
1444 let mut parts = token.split(':');
1445 let (prefix, history, version) = (parts.next()?, parts.next()?, parts.next()?);
1446 if parts.next().is_some()
1447 || prefix != VERSION_TOKEN_PREFIX
1448 || history.len() != 32
1449 || !history.bytes().all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
1450 || version.is_empty()
1451 || !version.bytes().all(|byte| byte.is_ascii_digit())
1452 {
1453 return None;
1454 }
1455 Some((history, version.parse().ok()?))
1456}
1457#[derive(Debug, Clone, Serialize, Deserialize)]
1458pub struct AtomicCommit {
1459 pub transaction_id: String,
1460 pub base_revision: u64,
1461 pub commit_revision: u64,
1462 pub status: TransactionStatus,
1463 pub state_before: u64,
1464 pub state_after: u64,
1465 pub rows: Vec<StoredRow>,
1466 pub duplicate: bool,
1467 #[serde(default, skip_serializing_if = "Option::is_none")]
1471 pub payload_binding: Option<PayloadBinding>,
1472}
1473
1474#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
1476#[serde(rename_all = "snake_case")]
1477pub enum PayloadBinding {
1478 Verified,
1481 LegacyUnverified,
1484}
1485
1486#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
1487#[serde(rename_all = "snake_case")]
1488pub enum TransactionStatus {
1489 Committed,
1490}
1491
1492#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1493pub enum JsonCasResult {
1494 Updated {
1495 current_version: u64,
1496 current_epoch: u64,
1497 value: Value,
1498 },
1499 VersionConflict {
1500 current_version: u64,
1501 current_epoch: u64,
1502 },
1503 AuthorityConflict {
1504 current_version: u64,
1505 current_epoch: u64,
1506 },
1507 LeaseConflict {
1508 current_version: u64,
1509 current_epoch: u64,
1510 },
1511 ReplicationConflict {
1512 current_version: u64,
1513 current_epoch: u64,
1514 },
1515 NotFound,
1516}
1517const TRANSACTION_ATTRIBUTION_WINDOW: usize = 512;
1523
1524pub const STANDALONE_OPERATION_WINDOW: usize = 4096;
1528
1529const TRANSACTION_ATTRIBUTION_KEYS: usize = 32;
1535
1536#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1543pub struct TransactionAttribution {
1544 pub transaction_id: String,
1545 #[serde(default)]
1548 pub operation_ids: Vec<String>,
1549 pub subject: Option<String>,
1551 pub tenant_id: Option<String>,
1552 pub application_id: Option<String>,
1553 pub application_revision: Option<String>,
1555 pub base_revision: u64,
1556 pub commit_revision: u64,
1557 pub state_before: u64,
1558 pub state_after: u64,
1559 pub keys: Vec<String>,
1562 pub operations: usize,
1565 #[serde(default)]
1566 pub principal_id: Option<String>,
1567 #[serde(default)]
1568 pub agent_id: Option<String>,
1569 #[serde(default)]
1570 pub session_id: Option<String>,
1571 #[serde(default)]
1572 pub delegator_id: Option<String>,
1573 #[serde(default)]
1574 pub work_id: Option<String>,
1575 pub unix_ms: u128,
1577 #[serde(default)]
1581 pub audit: Option<Value>,
1582}
1583
1584impl TransactionAttribution {
1585 fn from_record(
1587 transaction_id: &str,
1588 audit: Option<&Value>,
1589 base_revision: u64,
1590 commit_revision: u64,
1591 state_before: u64,
1592 state_after: u64,
1593 rows: &[StoredRow],
1594 unix_ms: u128,
1595 ) -> Self {
1596 let text = |field: &str| {
1597 audit
1598 .and_then(|value| value.get(field))
1599 .and_then(Value::as_str)
1600 .map(str::to_string)
1601 };
1602 TransactionAttribution {
1603 transaction_id: transaction_id.to_string(),
1604 operation_ids: rows
1605 .iter()
1606 .filter_map(|row| row.operation.as_ref())
1607 .map(|operation| {
1608 format!("operation:{}:{}", operation.instance_id, operation.sequence)
1609 })
1610 .collect(),
1611 subject: text("subject"),
1612 tenant_id: text("tenant"),
1613 application_id: text("application"),
1614 application_revision: text("revision"),
1615 base_revision,
1616 commit_revision,
1617 state_before,
1618 state_after,
1619 keys: rows
1620 .iter()
1621 .take(TRANSACTION_ATTRIBUTION_KEYS)
1622 .map(|row| format!("{}:{}", row.capability, row.key))
1623 .collect(),
1624 operations: rows.len(),
1625 principal_id: text("principal_id"),
1626 agent_id: text("agent_id"),
1627 session_id: text("session_id"),
1628 delegator_id: text("delegator_id"),
1629 work_id: text("work_id").or_else(|| text("workload_id")),
1630 unix_ms,
1631 audit: audit.cloned(),
1632 }
1633 }
1634
1635 pub fn touched(&self, capability: &str, key: &str) -> bool {
1637 let target = format!("{capability}:{key}");
1638 self.keys.iter().any(|held| *held == target)
1639 }
1640}
1641
1642fn normalize_transaction_attribution(
1646 mut attribution: TransactionAttribution,
1647) -> TransactionAttribution {
1648 if attribution.operation_ids.is_empty() {
1649 if attribution.transaction_id.starts_with("operation:") {
1650 let operation_id = attribution.transaction_id.clone();
1651 attribution.transaction_id = operation_id.replacen("operation:", "transaction:", 1);
1652 attribution.operation_ids.push(operation_id);
1653 } else {
1654 attribution.operation_ids = (1..=attribution.operations)
1655 .map(|index| {
1656 format!(
1657 "operation:legacy-transaction:{}:{index}",
1658 attribution.transaction_id
1659 )
1660 })
1661 .collect();
1662 }
1663 }
1664 attribution
1665}
1666
1667fn remember_attribution(inner: &mut Inner, attribution: TransactionAttribution) {
1669 inner.transaction_history.push(attribution.clone());
1670 if inner.transaction_attributions.len() >= TRANSACTION_ATTRIBUTION_WINDOW {
1671 inner.transaction_attributions.pop_front();
1672 }
1673 inner.transaction_attributions.push_back(attribution);
1674}
1675
1676fn remember_operation_attribution(
1677 inner: &mut Inner,
1678 row: &StoredRow,
1679 base_revision: u64,
1680 commit_revision: u64,
1681) {
1682 let Some(operation) = row.operation.as_ref() else {
1683 return;
1684 };
1685 remember_attribution(
1686 inner,
1687 TransactionAttribution {
1688 transaction_id: format!(
1689 "transaction:{}:{}",
1690 operation.instance_id, operation.sequence
1691 ),
1692 operation_ids: vec![format!(
1693 "operation:{}:{}",
1694 operation.instance_id, operation.sequence
1695 )],
1696 subject: None,
1697 tenant_id: None,
1698 application_id: None,
1699 application_revision: operation
1700 .revision
1701 .as_ref()
1702 .map(|revision| revision.id.clone()),
1703 base_revision,
1704 commit_revision,
1705 state_before: operation.sequence.saturating_sub(1),
1706 state_after: operation.sequence,
1707 keys: vec![format!("{}:{}", row.capability, row.key)],
1708 operations: 1,
1709 principal_id: None,
1710 agent_id: None,
1711 session_id: None,
1712 delegator_id: None,
1713 work_id: None,
1714 unix_ms: row.unix_ms,
1715 audit: None,
1716 },
1717 );
1718}
1719
1720#[derive(Debug, Clone, Serialize, Deserialize)]
1721struct TransactionLogRecord {
1722 record_type: String,
1723 transaction_id: String,
1724 #[serde(default)]
1725 payload_hash: Option<String>,
1726 state_before: u64,
1727 state_after: u64,
1728 #[serde(default)]
1729 base_revision: u64,
1730 #[serde(default)]
1731 commit_revision: u64,
1732 rows: Vec<StoredRow>,
1733 #[serde(default)]
1734 audit: Option<Value>,
1735}
1736
1737#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
1741pub struct ChangeEvent {
1742 pub capability: String,
1744 pub key: String,
1746 pub rust_type: String,
1748 pub unix_ms: u128,
1750}
1751
1752#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
1753pub struct StoredRow {
1754 pub capability: String,
1755 pub key: String,
1756 pub rust_type: String,
1757 pub value: Value,
1758 pub unix_ms: u128,
1759 #[serde(default)]
1761 pub content_hash: Option<String>,
1762 #[serde(default)]
1764 pub flow_ref: Option<FlowRef>,
1765 #[serde(default)]
1767 pub deleted: bool,
1768 #[serde(default)]
1770 pub operation: Option<Operation>,
1771 #[serde(default, skip_serializing_if = "Option::is_none")]
1778 pub recorded_version: Option<u64>,
1779}
1780
1781impl StoredRow {
1782 pub fn version(&self) -> Option<u64> {
1791 if self.deleted {
1792 return None;
1793 }
1794 Some(match &self.operation {
1795 Some(operation) => operation.sequence,
1796 None => self.recorded_version.unwrap_or(0),
1797 })
1798 }
1799}
1800
1801fn snapshot_row(row: &StoredRow) -> StoredRow {
1805 let mut snapshot = row.clone();
1806 snapshot.recorded_version = row.version();
1807 snapshot.operation = None;
1808 snapshot
1809}
1810
1811#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1821pub struct CompactionPolicy {
1822 pub rewrite_after_pruned: usize,
1825}
1826
1827impl Default for CompactionPolicy {
1828 fn default() -> Self {
1829 CompactionPolicy {
1833 rewrite_after_pruned: 1000,
1834 }
1835 }
1836}
1837
1838#[derive(Debug, Clone, PartialEq, Eq)]
1840pub enum CompactionOutcome {
1841 Idle,
1843 Deferred {
1846 pruned: usize,
1847 pending_since_rewrite: usize,
1848 },
1849 Rewritten { pruned: usize },
1851}
1852
1853impl CompactionOutcome {
1854 pub fn pruned(&self) -> usize {
1856 match self {
1857 CompactionOutcome::Idle => 0,
1858 CompactionOutcome::Deferred { pruned, .. }
1859 | CompactionOutcome::Rewritten { pruned } => *pruned,
1860 }
1861 }
1862
1863 pub fn rewrote_log(&self) -> bool {
1865 matches!(self, CompactionOutcome::Rewritten { .. })
1866 }
1867}
1868
1869#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1876pub enum DurabilityMode {
1877 Flushed,
1882 Synced,
1889 Grouped { every: u32 },
1894}
1895
1896impl Default for DurabilityMode {
1897 fn default() -> Self {
1898 DurabilityMode::Flushed
1899 }
1900}
1901
1902impl DurabilityMode {
1903 pub fn guarantee(&self) -> &'static str {
1906 match self {
1907 DurabilityMode::Flushed => {
1908 "the record has left the process; it survives a process crash. Stable-storage durability is not established."
1909 }
1910 DurabilityMode::Synced => {
1911 "a stable-storage barrier was issued for this record. Whether it reached the device depends on the storage stack, which this database cannot observe."
1912 }
1913 DurabilityMode::Grouped { .. } => {
1914 "the record has left the process. A stable-storage barrier is issued periodically, so an acknowledged write may not yet have been through one."
1915 }
1916 }
1917 }
1918
1919 pub fn unbarriered_window(&self) -> Option<u32> {
1923 match self {
1924 DurabilityMode::Flushed => None,
1925 DurabilityMode::Synced => Some(0),
1926 DurabilityMode::Grouped { every } => Some(every.saturating_sub(1)),
1927 }
1928 }
1929}
1930
1931#[derive(Debug, Clone, PartialEq, Eq)]
1938pub enum StorageHealth {
1939 Clean,
1941 RecoveredIncompleteWrite {
1948 byte_offset: u64,
1949 discarded_bytes: usize,
1950 },
1951}
1952
1953impl StorageHealth {
1954 pub fn is_clean(&self) -> bool {
1956 matches!(self, StorageHealth::Clean)
1957 }
1958
1959 pub fn label(&self) -> &'static str {
1961 match self {
1962 StorageHealth::Clean => "clean",
1963 StorageHealth::RecoveredIncompleteWrite { .. } => "recovered-incomplete-write",
1964 }
1965 }
1966}
1967
1968impl Display for StorageHealth {
1969 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1970 match self {
1971 StorageHealth::Clean => write!(f, "durable log replayed cleanly"),
1972 StorageHealth::RecoveredIncompleteWrite {
1973 byte_offset,
1974 discarded_bytes,
1975 } => write!(
1976 f,
1977 "durable log ended in an incomplete append: {discarded_bytes} bytes at offset \
1978 {byte_offset} were discarded, so the last write did not land"
1979 ),
1980 }
1981 }
1982}
1983
1984#[derive(Debug, Clone, PartialEq, Eq)]
1991pub struct DatabaseHealth {
1992 pub durable_format: DurableFormat,
1997 pub recovery: LogRecovery,
1999 pub storage: StorageHealth,
2001 pub writes: WriteAvailability,
2003}
2004
2005#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
2007#[serde(tag = "state", rename_all = "kebab-case")]
2008pub enum WriteAvailability {
2009 Available,
2011 RecoveryRequired { reason: String },
2016}
2017
2018impl WriteAvailability {
2019 pub fn label(&self) -> &'static str {
2020 match self {
2021 WriteAvailability::Available => "available",
2022 WriteAvailability::RecoveryRequired { .. } => "recovery-required",
2023 }
2024 }
2025}
2026
2027impl DatabaseHealth {
2028 pub fn is_nominal(&self) -> bool {
2034 self.storage.is_clean() && self.writes == WriteAvailability::Available
2035 }
2036}
2037
2038pub const BACKUP_VERSION: u32 = 1;
2044
2045const BACKUP_RECORD_TYPE: &str = "feltdb.backup.v1";
2046
2047#[derive(Debug, Clone, Serialize, Deserialize)]
2049struct BackupHeader {
2050 record_type: String,
2051 backup_version: u32,
2052 format_version: u32,
2054 created_unix_ms: u64,
2055 row_count: usize,
2056 observed_versions: HashMap<String, u64>,
2057 content_digest: String,
2059 meaning_digest: String,
2061}
2062
2063#[derive(Debug, Clone, PartialEq, Eq)]
2065pub struct BackupProblem {
2066 pub reason: String,
2068 pub line_number: Option<usize>,
2070 pub action: String,
2072}
2073
2074impl Display for BackupProblem {
2075 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
2076 match self.line_number {
2077 Some(line) => write!(f, "record {line}: {}. {}", self.reason, self.action),
2078 None => write!(f, "{}. {}", self.reason, self.action),
2079 }
2080 }
2081}
2082
2083#[derive(Debug, Clone, PartialEq, Eq)]
2100pub struct BackupVerification {
2101 pub backup_version: u32,
2102 pub format_version: u32,
2103 pub created_unix_ms: u64,
2104 pub rows: usize,
2105 pub content_digest: String,
2106 pub meaning_digest: String,
2107}
2108
2109fn backup_problem(reason: &str, line_number: Option<usize>, action: &str) -> FlowError {
2110 FlowError::BackupRejected(Box::new(BackupProblem {
2111 reason: reason.to_string(),
2112 line_number,
2113 action: action.to_string(),
2114 }))
2115}
2116
2117fn backup_content_digest(rows: &[StoredRow]) -> Result<String> {
2119 let mut hasher = Sha256::new();
2120 for row in rows {
2121 hasher.update(serde_json::to_vec(row)?);
2122 hasher.update(b"\n");
2123 }
2124 Ok(format!("{:x}", hasher.finalize()))
2125}
2126
2127fn backup_meaning_digest(rows: &[StoredRow]) -> Result<String> {
2133 let mut meaningful: Vec<(&str, &str, String)> = rows
2134 .iter()
2135 .filter(|row| !row.deleted)
2136 .map(|row| {
2137 (
2138 row.capability.as_str(),
2139 row.key.as_str(),
2140 row.value.to_string(),
2141 )
2142 })
2143 .collect();
2144 meaningful.sort();
2145
2146 let mut hasher = Sha256::new();
2147 hasher.update(b"feltdb.backup.meaning.v1");
2148 for (capability, key, value) in meaningful {
2149 hasher.update(capability.as_bytes());
2150 hasher.update([0u8]);
2151 hasher.update(key.as_bytes());
2152 hasher.update([0u8]);
2153 hasher.update(value.as_bytes());
2154 hasher.update([0u8]);
2155 }
2156 Ok(format!("{:x}", hasher.finalize()))
2157}
2158
2159pub fn verify_backup(path: &Path) -> Result<BackupVerification> {
2165 let text = std::fs::read_to_string(path)?;
2166 let mut lines = text.lines();
2167
2168 let Some(header_line) = lines.next() else {
2169 return Err(backup_problem(
2170 "the artifact is empty",
2171 None,
2172 "Restore from a different backup.",
2173 ));
2174 };
2175 let header: BackupHeader = serde_json::from_str(header_line).map_err(|error| {
2176 backup_problem(
2177 &format!("the header is not a FeltDB backup header: {error}"),
2178 Some(1),
2179 "Check that this file is a FeltDB backup and was not truncated at the start.",
2180 )
2181 })?;
2182 if header.record_type != BACKUP_RECORD_TYPE {
2183 return Err(backup_problem(
2184 &format!("unexpected record type {}", header.record_type),
2185 Some(1),
2186 "Check that this file is a FeltDB backup.",
2187 ));
2188 }
2189 if header.backup_version != BACKUP_VERSION {
2190 return Err(backup_problem(
2191 &format!(
2192 "backup format version {} is not the version this build reads ({BACKUP_VERSION})",
2193 header.backup_version
2194 ),
2195 Some(1),
2196 "Restore with a FeltDB build that reads this backup version.",
2197 ));
2198 }
2199 if header.format_version != DURABLE_FORMAT_VERSION {
2200 return Err(backup_problem(
2201 &format!(
2202 "the rows are in durable format version {} and this build requires version {DURABLE_FORMAT_VERSION}",
2203 header.format_version
2204 ),
2205 Some(1),
2206 "Restore with a FeltDB build matching the format the backup was taken in.",
2207 ));
2208 }
2209
2210 let mut rows = Vec::with_capacity(header.row_count);
2211 for (index, line) in lines.enumerate() {
2212 if line.trim().is_empty() {
2213 continue;
2214 }
2215 let row: StoredRow = serde_json::from_str(line).map_err(|error| {
2216 backup_problem(
2217 &format!("not a durable record: {error}"),
2218 Some(index + 2),
2219 "The backup is damaged. Restore from another copy.",
2220 )
2221 })?;
2222 rows.push(row);
2223 }
2224
2225 if rows.len() != header.row_count {
2226 return Err(backup_problem(
2227 &format!(
2228 "the header declares {} records and the artifact holds {}",
2229 header.row_count,
2230 rows.len()
2231 ),
2232 None,
2233 "The backup is truncated or was appended to. Restore from another copy.",
2234 ));
2235 }
2236
2237 let content_digest = backup_content_digest(&rows)?;
2238 if content_digest != header.content_digest {
2239 return Err(backup_problem(
2240 "the records do not match the digest in the header",
2241 None,
2242 "The backup has been altered or corrupted. Restore from another copy.",
2243 ));
2244 }
2245
2246 let meaning_digest = backup_meaning_digest(&rows)?;
2247 if meaning_digest != header.meaning_digest {
2248 return Err(backup_problem(
2249 "the records do not mean what the header says they mean",
2250 None,
2251 "The backup has been altered. Restore from another copy.",
2252 ));
2253 }
2254
2255 Ok(BackupVerification {
2256 backup_version: header.backup_version,
2257 format_version: header.format_version,
2258 created_unix_ms: header.created_unix_ms,
2259 rows: rows.len(),
2260 content_digest,
2261 meaning_digest,
2262 })
2263}
2264
2265#[derive(Debug, Clone, PartialEq, Eq)]
2270pub struct LogCorruption {
2271 pub line_number: usize,
2273 pub byte_offset: u64,
2275 pub reason: String,
2277 pub excerpt: String,
2279}
2280
2281impl Display for LogCorruption {
2282 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
2283 write!(
2284 f,
2285 "durable log record {} at byte {} is corrupt: {}. Excerpt: {}. \
2286 The database was not opened and was not modified.",
2287 self.line_number, self.byte_offset, self.reason, self.excerpt
2288 )
2289 }
2290}
2291
2292#[derive(Debug, Clone, PartialEq, Eq)]
2299pub enum LogRecovery {
2300 Clean,
2302 RecoveredTornTail {
2313 byte_offset: u64,
2315 discarded_bytes: usize,
2317 },
2318}
2319
2320impl Default for LogRecovery {
2321 fn default() -> Self {
2322 LogRecovery::Clean
2323 }
2324}
2325
2326impl Default for DurableFormat {
2327 fn default() -> Self {
2328 DurableFormat::Versioned(DURABLE_FORMAT_VERSION)
2329 }
2330}
2331
2332impl LogRecovery {
2333 pub fn is_clean(&self) -> bool {
2335 matches!(self, LogRecovery::Clean)
2336 }
2337}
2338
2339impl Display for LogRecovery {
2340 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
2341 match self {
2342 LogRecovery::Clean => write!(f, "clean"),
2343 LogRecovery::RecoveredTornTail {
2344 byte_offset,
2345 discarded_bytes,
2346 } => write!(
2347 f,
2348 "recovered from an incomplete final append: {discarded_bytes} bytes at offset \
2349 {byte_offset} were not a complete record and were discarded"
2350 ),
2351 }
2352 }
2353}
2354
2355pub const DURABLE_FORMAT_VERSION: u32 = 2;
2365
2366const FORMAT_RECORD_TYPE: &str = "feltdb.format.v1";
2368
2369#[derive(Debug, Serialize, Deserialize)]
2370struct FormatRecord {
2371 record_type: String,
2372 format_version: u32,
2373}
2374
2375#[derive(Debug, Clone, PartialEq, Eq)]
2377pub enum DurableFormat {
2378 Versioned(u32),
2380 Unversioned,
2387}
2388
2389impl Display for DurableFormat {
2390 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
2391 match self {
2392 DurableFormat::Versioned(version) => write!(f, "version {version}"),
2393 DurableFormat::Unversioned => write!(f, "unversioned (predates format versioning)"),
2394 }
2395 }
2396}
2397
2398#[derive(Debug, Clone, PartialEq, Eq)]
2400pub struct FormatIncompatibility {
2401 pub found: DurableFormat,
2403 pub required: u32,
2405 pub reason: String,
2407 pub action: String,
2409}
2410
2411impl Display for FormatIncompatibility {
2412 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
2413 write!(
2414 f,
2415 "durable format {} is incompatible with this FeltDB, which requires \
2416 format version {}; {}. No data was read or modified. {}",
2417 self.found, self.required, self.reason, self.action
2418 )
2419 }
2420}
2421
2422#[derive(Debug, Clone, PartialEq, Eq)]
2424pub enum FormatCompatibility {
2425 Compatible(DurableFormat),
2427 Incompatible(FormatIncompatibility),
2429}
2430
2431impl FormatCompatibility {
2432 pub fn is_compatible(&self) -> bool {
2434 matches!(self, FormatCompatibility::Compatible(_))
2435 }
2436}
2437
2438#[derive(Debug, Clone, Serialize, Deserialize)]
2439pub struct DatabaseSnapshot {
2440 pub rows: Vec<StoredRow>,
2441 pub versions: HashMap<String, u64>,
2442 pub content_hash: String,
2443 #[serde(default)]
2455 pub format_version: u32,
2456 #[serde(default, skip_serializing_if = "Option::is_none")]
2461 pub history: Option<VersionHistory>,
2462}
2463
2464impl DatabaseSnapshot {
2465 fn compute_hash(rows: &[StoredRow], versions: &HashMap<String, u64>) -> Result<String> {
2466 let mut rows = rows.to_vec();
2467 rows.sort_by(|left, right| left.key.cmp(&right.key));
2468 let versions: std::collections::BTreeMap<_, _> = versions.iter().collect();
2469 let mut hasher = Sha256::new();
2470 hasher.update(serde_json::to_vec(&(rows, versions))?);
2471 Ok(format!("{:x}", hasher.finalize()))
2472 }
2473
2474 pub fn verify(&self) -> Result<()> {
2475 if self.content_hash != Self::compute_hash(&self.rows, &self.versions)? {
2476 return Err(FlowError::CapabilityError(
2477 "snapshot content hash mismatch".to_string(),
2478 ));
2479 }
2480 Ok(())
2481 }
2482
2483 pub fn validate(&self) -> Result<()> {
2494 if self.format_version != DURABLE_FORMAT_VERSION {
2495 return Err(FlowError::IncompatibleFormat(Box::new(FormatIncompatibility {
2496 found: if self.format_version == 0 {
2497 DurableFormat::Unversioned
2498 } else {
2499 DurableFormat::Versioned(self.format_version)
2500 },
2501 required: DURABLE_FORMAT_VERSION,
2502 reason: "the snapshot was produced in a different durable format".to_string(),
2503 action: "Take a fresh snapshot from a FeltDB build matching this one.".to_string(),
2504 })));
2505 }
2506 self.verify()?;
2507 let invalid = |row: &StoredRow, why: &str| {
2508 FlowError::CapabilityError(format!(
2509 "snapshot contains an invalid state row: {why} ({:?} {:?})",
2510 row.capability, row.key
2511 ))
2512 };
2513 let mut identities = HashSet::with_capacity(self.rows.len());
2514 for row in &self.rows {
2515 if row.deleted {
2516 return Err(invalid(row, "a tombstone"));
2517 }
2518 if row.operation.is_some() {
2519 return Err(invalid(row, "an operation, not state"));
2520 }
2521 if row.capability.is_empty() || row.key.is_empty() {
2522 return Err(invalid(row, "no collection or no key"));
2523 }
2524 if !identities.insert((row.capability.as_str(), row.key.as_str())) {
2525 return Err(invalid(row, "a second row with the same identity"));
2526 }
2527 }
2528 Ok(())
2529 }
2530}
2531
2532#[derive(Debug, Serialize, Deserialize)]
2533struct LogHeader {
2534 record_type: String,
2535 local_sequence: u64,
2536 observed_versions: HashMap<String, u64>,
2537 #[serde(default)]
2540 format_version: u32,
2541 #[serde(default)]
2544 transaction_history: Vec<TransactionAttribution>,
2545 #[serde(default)]
2549 transaction_payload_hashes: HashMap<String, String>,
2550 #[serde(default)]
2554 authority_revision: u64,
2555}
2556
2557#[derive(Serialize)]
2559struct LogHeaderRef<'a> {
2560 record_type: &'a str,
2561 local_sequence: u64,
2562 observed_versions: &'a HashMap<String, u64>,
2563 format_version: u32,
2564 transaction_history: &'a Vec<TransactionAttribution>,
2565 transaction_payload_hashes: &'a HashMap<String, String>,
2566 authority_revision: u64,
2567}
2568
2569#[derive(Debug, Default, Serialize, Deserialize)]
2570struct SyncMetadata {
2571 local_sequence: u64,
2572 observed_versions: HashMap<String, u64>,
2573 peer_acknowledgements: HashMap<String, HashMap<String, u64>>,
2574 #[serde(default)]
2576 history: Option<VersionHistory>,
2577}
2578
2579pub fn open<P: AsRef<Path>>(path: P) -> Result<FeltDb> {
2580 FeltDb::open(path)
2581}
2582
2583const UNVERSIONED_CAPABILITIES: &[&str] = &["state", "_retention"];
2589
2590fn is_versioned(capability: &str) -> bool {
2592 !UNVERSIONED_CAPABILITIES.contains(&capability)
2593}
2594
2595fn rebuild_revision_heads(inner: &mut Inner) {
2601 let mut heads: HashMap<String, (state_model::StateId, u64)> = HashMap::new();
2602 if let Some(bucket) = inner.rows.get("state") {
2603 for row in bucket.values() {
2604 if row.deleted || !row.key.starts_with("state:revision:") {
2605 continue;
2606 }
2607 let Ok(revision) =
2608 serde_json::from_value::<state_model::StateRevision>(row.value.clone())
2609 else {
2610 continue;
2611 };
2612 let entry = heads
2613 .entry(revision.resource.clone())
2614 .or_insert_with(|| (revision.id.clone(), revision.sequence));
2615 if revision.sequence >= entry.1 {
2616 *entry = (revision.id.clone(), revision.sequence);
2617 }
2618 }
2619 }
2620 inner.revision_heads = heads;
2621}
2622
2623fn revision_row(revision: &state_model::StateRevision) -> Result<StoredRow> {
2641 Ok(StoredRow {
2642 recorded_version: None,
2643 capability: "state".to_string(),
2644 key: format!("state:revision:{}", revision.id.as_hex()),
2645 rust_type: "feltdb::state_model::StateRevision".to_string(),
2646 value: serde_json::to_value(revision)?,
2647 unix_ms: now_ms(),
2648 content_hash: None,
2649 flow_ref: None,
2650 deleted: false,
2651 operation: None,
2652 })
2653}
2654
2655fn commit_local_write_locked(
2660 inner: &mut Inner,
2661 row: &StoredRow,
2662 planned: Option<&state_model::StateRevision>,
2663 reserved: (u64, Option<HashMap<String, u64>>),
2664) -> Result<()> {
2665 let revision = revision_row_to_append(inner, planned)?;
2666 let mut lines = vec![row];
2667 lines.extend(revision.as_ref());
2668 if let Err(error) = append_rows_locked(inner, &lines) {
2669 inner.sequence = reserved.0;
2670 inner.sync_state.vector_clock = reserved.1;
2671 return Err(error);
2672 }
2673 if let Some(revision) = revision {
2674 inner.put_row(revision);
2675 }
2676 Ok(())
2677}
2678
2679fn revision_row_to_append(
2682 inner: &Inner,
2683 planned: Option<&state_model::StateRevision>,
2684) -> Result<Option<StoredRow>> {
2685 let Some(revision) = planned else {
2686 return Ok(None);
2687 };
2688 let row = revision_row(revision)?;
2689 let stored = inner
2690 .rows
2691 .get("state")
2692 .and_then(|bucket| bucket.get(&row.key))
2693 .is_some_and(|row| !row.deleted);
2694 Ok((!stored).then_some(row))
2695}
2696
2697fn plan_revision(inner: &Inner, resource: &str, content: String) -> state_model::StateRevision {
2705 let parent = inner.revision_heads.get(resource).cloned();
2706 let (parent_id, sequence) = match &parent {
2707 Some((id, sequence)) => (Some(id.clone()), sequence + 1),
2708 None => (None, 0),
2709 };
2710 state_model::StateRevision::at(
2711 resource.to_string(),
2712 content,
2713 parent_id,
2714 sequence,
2715 inner.instance_id.clone(),
2716 )
2717}
2718
2719fn provenance_of(revision: &state_model::StateRevision) -> operation::RevisionProvenance {
2721 operation::RevisionProvenance {
2722 resource: revision.resource.clone(),
2723 id: revision.id.as_hex().to_string(),
2724 parent_id: revision
2725 .parent_id
2726 .as_ref()
2727 .map(|id| id.as_hex().to_string()),
2728 sequence: revision.sequence,
2729 content_id: revision.content_id.as_hex().to_string(),
2730 authority: revision.authority.clone(),
2731 }
2732}
2733
2734fn persist_revision_locked(inner: &mut Inner, revision: state_model::StateRevision) -> Result<()> {
2741 let key = format!("state:revision:{}", revision.id.as_hex());
2742 let already = inner
2743 .rows
2744 .get("state")
2745 .and_then(|bucket| bucket.get(&key))
2746 .is_some_and(|row| !row.deleted);
2747
2748 let resource = revision.resource.clone();
2749 let id = revision.id.clone();
2750 let sequence = revision.sequence;
2751
2752 if !already {
2753 let row = revision_row(&revision)?;
2754 let _ = append_event_locked(inner, &row);
2760 inner.put_row(row);
2761 }
2762
2763 let advance = inner
2766 .revision_heads
2767 .get(&resource)
2768 .is_none_or(|(_, current)| sequence > *current);
2769 if advance {
2770 inner
2771 .revision_heads
2772 .insert(resource.clone(), (id, sequence));
2773 }
2774
2775 let _ = apply_retention_locked(inner, &resource);
2779 Ok(())
2780}
2781
2782fn reconstruct_revision(
2790 provenance: &operation::RevisionProvenance,
2791 value: &Value,
2792) -> Result<state_model::StateRevision> {
2793 let content = value.to_string();
2794 let revision = state_model::StateRevision::at(
2795 provenance.resource.clone(),
2796 content,
2797 provenance
2798 .parent_id
2799 .as_ref()
2800 .map(|id| state_model::StateId::from_hex(id.clone())),
2801 provenance.sequence,
2802 provenance.authority.clone(),
2803 );
2804 if revision.content_id.as_hex() != provenance.content_id {
2805 return Err(FlowError::CapabilityError(format!(
2806 "replicated revision {} does not match the value it arrived with",
2807 provenance.id
2808 )));
2809 }
2810 if revision.id.as_hex() != provenance.id {
2811 return Err(FlowError::CapabilityError(format!(
2812 "replicated revision identity {} does not follow from its own resource, parent and sequence",
2813 provenance.id
2814 )));
2815 }
2816 Ok(revision)
2817}
2818
2819fn apply_retention_locked(inner: &mut Inner, resource: &str) -> Result<()> {
2825 let retention_key = format!("_retention:{resource}");
2826 let Some(state_row) = inner
2827 .rows
2828 .get("_retention")
2829 .and_then(|bucket| bucket.get(&retention_key))
2830 else {
2831 return Ok(());
2832 };
2833 let Ok(mut state) = serde_json::from_value::<RetentionRecord>(state_row.value.clone()) else {
2834 return Ok(());
2835 };
2836 if state.policy.keep_last.is_none() {
2837 return Ok(());
2838 }
2839
2840 let mut history: Vec<(state_model::StateId, u64)> = Vec::new();
2841 if let Some(bucket) = inner.rows.get("state") {
2842 for row in bucket.values() {
2843 if row.deleted || !row.key.starts_with("state:revision:") {
2844 continue;
2845 }
2846 if let Ok(revision) =
2847 serde_json::from_value::<state_model::StateRevision>(row.value.clone())
2848 {
2849 if revision.resource == resource {
2850 history.push((revision.id, revision.sequence));
2851 }
2852 }
2853 }
2854 }
2855 history.sort_by_key(|(_, sequence)| *sequence);
2856
2857 let (expire, horizon) = state_model::revisions_to_expire(&history, state.policy.keep_last);
2858 if expire.is_empty() {
2859 return Ok(());
2860 }
2861 let tombstones: Vec<StoredRow> = expire
2862 .iter()
2863 .map(|id| StoredRow {
2864 recorded_version: None,
2865 capability: "state".to_string(),
2866 key: format!("state:revision:{}", id.as_hex()),
2867 rust_type: "feltdb::state_model::StateRevision".to_string(),
2868 value: Value::Null,
2869 unix_ms: now_ms(),
2870 content_hash: None,
2871 flow_ref: None,
2872 deleted: true,
2873 operation: None,
2874 })
2875 .collect();
2876 if let Some(horizon) = horizon {
2877 state.horizon = horizon;
2878 }
2879 let retention = StoredRow {
2880 recorded_version: None,
2881 capability: "_retention".to_string(),
2882 key: retention_key,
2883 rust_type: "feltdb::state_model::StateRevision".to_string(),
2884 value: serde_json::to_value(&state)?,
2885 unix_ms: now_ms(),
2886 content_hash: None,
2887 flow_ref: None,
2888 deleted: false,
2889 operation: None,
2890 };
2891 let mut lines: Vec<&StoredRow> = tombstones.iter().collect();
2894 lines.push(&retention);
2895 append_rows_locked(inner, &lines)?;
2896 if let Some(bucket) = inner.rows.get_mut("state") {
2897 for tombstone in &tombstones {
2898 bucket.remove(&tombstone.key);
2899 }
2900 }
2901 inner.put_row(retention);
2902 Ok(())
2903}
2904
2905#[derive(Clone, Debug, Default, Serialize, Deserialize)]
2910struct RetentionRecord {
2911 policy: state_model::RetentionPolicy,
2912 horizon: u64,
2913}
2914
2915impl FeltDb {
2916 pub fn open<P: AsRef<Path>>(path: P) -> Result<Self> {
2917 let requested_path = path.as_ref();
2918 let parent = requested_path
2919 .parent()
2920 .filter(|parent| !parent.as_os_str().is_empty())
2921 .unwrap_or_else(|| Path::new("."));
2922 fs::create_dir_all(parent)?;
2923 let path = if requested_path.exists() {
2924 fs::canonicalize(requested_path)?
2925 } else {
2926 let file_name = requested_path.file_name().ok_or_else(|| {
2927 std::io::Error::new(
2928 std::io::ErrorKind::InvalidInput,
2929 "FeltDB database path has no file name",
2930 )
2931 })?;
2932 fs::canonicalize(parent)?.join(file_name)
2933 };
2934
2935 let databases = OPEN_DATABASES.get_or_init(|| Mutex::new(HashMap::new()));
2936 let mut databases = databases.lock().expect("open database registry poisoned");
2937 if let Some(open) = databases.get(&path) {
2938 if let (Some(inner), Some(ownership), Some(history)) =
2939 (open.inner.upgrade(), open.ownership.upgrade(), open.history.upgrade())
2940 {
2941 return Ok(Self {
2942 inner,
2943 event_tx: open.event_tx.clone(),
2944 _ownership: ownership,
2945 history,
2946 });
2947 }
2948 }
2949 databases.remove(&path);
2950
2951 let mut ownership_path = path.as_os_str().to_os_string();
2952 ownership_path.push(".lock");
2953 let ownership = OpenOptions::new()
2954 .create(true)
2955 .read(true)
2956 .write(true)
2957 .open(PathBuf::from(ownership_path))?;
2958 ownership.try_lock().map_err(|error| match error {
2959 std::fs::TryLockError::WouldBlock => std::io::Error::new(
2960 std::io::ErrorKind::WouldBlock,
2961 format!(
2962 "FeltDB database {} is already owned by another process",
2963 path.display()
2964 ),
2965 ),
2966 std::fs::TryLockError::Error(error) => error,
2967 })?;
2968
2969 for leftover in ["compact.tmp", "snapshot.tmp"] {
2974 match fs::remove_file(sibling(&path, leftover)) {
2975 Ok(()) => {}
2976 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
2977 Err(error) => return Err(error.into()),
2978 }
2979 }
2980
2981 let accepted_format = match inspect_durable_format(&path)? {
2986 FormatCompatibility::Incompatible(incompatibility) => {
2987 return Err(FlowError::IncompatibleFormat(Box::new(incompatibility)));
2988 }
2989 FormatCompatibility::Compatible(format) => format,
2990 };
2991
2992 if !path.exists() {
2993 let mut file = OpenOptions::new().create(true).append(true).open(&path)?;
2994 write_json_line(
2997 &mut file,
2998 &FormatRecord {
2999 record_type: FORMAT_RECORD_TYPE.to_string(),
3000 format_version: DURABLE_FORMAT_VERSION,
3001 },
3002 )?;
3003 file.sync_data()?;
3006 sync_parent_directory(&path)?;
3007 }
3008
3009 let instance_id = format!("feltdb-{:x}", {
3011 let mut hasher = DefaultHasher::new();
3012 path.to_string_lossy().hash(&mut hasher);
3013 hasher.finish()
3014 });
3015
3016 let mut inner = Inner {
3017 path: path.clone(),
3018 instance_id: instance_id.clone(),
3019 durable_format: accepted_format,
3020 ..Inner::default()
3021 };
3022
3023 inner.sync_state = SyncState::new(instance_id);
3025 replay_log(&mut inner)?;
3026 load_sync_metadata(&mut inner)?;
3027 if inner.history.id.is_empty() {
3028 inner.history = VersionHistory::new(HistoryOrigin::Genesis);
3031 persist_sync_metadata(&inner)?;
3032 }
3033 rebuild_revision_heads(&mut inner);
3034
3035 let (event_tx, _) = broadcast::channel(1024);
3036
3037 let history = Arc::new(std::sync::RwLock::new(inner.history.clone()));
3038 let database = Self {
3039 inner: Arc::new(Mutex::new(inner)),
3040 event_tx,
3041 _ownership: Arc::new(ownership),
3042 history,
3043 };
3044 databases.insert(
3045 path,
3046 OpenDatabase {
3047 inner: Arc::downgrade(&database.inner),
3048 event_tx: database.event_tx.clone(),
3049 ownership: Arc::downgrade(&database._ownership),
3050 history: Arc::downgrade(&database.history),
3051 },
3052 );
3053 Ok(database)
3054 }
3055
3056 pub fn write_backup(&self, path: &Path) -> Result<BackupVerification> {
3063 let (rows, observed_versions) = {
3064 let inner = self.state();
3065 let mut rows: Vec<StoredRow> = inner
3066 .rows
3067 .values()
3068 .flat_map(|bucket| bucket.values().cloned())
3069 .filter(|row| !row.deleted)
3070 .collect();
3071 rows.sort_by(|left, right| {
3072 left.capability
3073 .cmp(&right.capability)
3074 .then_with(|| left.key.cmp(&right.key))
3075 });
3076 (rows, inner.change_log.versions())
3077 };
3078
3079 let content_digest = backup_content_digest(&rows)?;
3080 let meaning_digest = backup_meaning_digest(&rows)?;
3081 let header = BackupHeader {
3082 record_type: BACKUP_RECORD_TYPE.to_string(),
3083 backup_version: BACKUP_VERSION,
3084 format_version: DURABLE_FORMAT_VERSION,
3085 created_unix_ms: now_ms() as u64,
3086 row_count: rows.len(),
3087 observed_versions,
3088 content_digest: content_digest.clone(),
3089 meaning_digest: meaning_digest.clone(),
3090 };
3091
3092 if let Some(parent) = path.parent() {
3093 fs::create_dir_all(parent)?;
3094 }
3095 let partial = sibling(path, "partial");
3099 let mut file = OpenOptions::new()
3100 .create(true)
3101 .write(true)
3102 .truncate(true)
3103 .open(&partial)?;
3104 write_json_line(&mut file, &header)?;
3105 for row in &rows {
3106 write_json_line(&mut file, row)?;
3107 }
3108 file.flush()?;
3109 file.sync_all()?;
3110 drop(file);
3111 fs::rename(&partial, path)?;
3112 sync_parent_directory(path)?;
3113
3114 Ok(BackupVerification {
3115 backup_version: BACKUP_VERSION,
3116 format_version: DURABLE_FORMAT_VERSION,
3117 created_unix_ms: header.created_unix_ms,
3118 rows: rows.len(),
3119 content_digest,
3120 meaning_digest,
3121 })
3122 }
3123
3124 pub fn restore_backup(backup: &Path, target: &Path) -> Result<(Self, BackupVerification)> {
3133 let verification = verify_backup(backup)?;
3134 if target.exists() {
3135 return Err(backup_problem(
3136 "the restore target already exists",
3137 None,
3138 "Restore into a new path. FeltDB will not write over an existing database.",
3139 ));
3140 }
3141
3142 let text = std::fs::read_to_string(backup)?;
3143 let mut lines = text.lines();
3144 let header: BackupHeader = serde_json::from_str(lines.next().unwrap_or_default())?;
3145
3146 if let Some(parent) = target.parent() {
3147 fs::create_dir_all(parent)?;
3148 }
3149 let staging = sibling(target, "restoring");
3154 remove_restore_staging(&staging)?;
3155 {
3156 let mut file = OpenOptions::new()
3157 .create_new(true)
3158 .write(true)
3159 .open(&staging)?;
3160 write_json_line(
3161 &mut file,
3162 &FormatRecord {
3163 record_type: FORMAT_RECORD_TYPE.to_string(),
3164 format_version: header.format_version,
3165 },
3166 )?;
3167 for line in lines {
3168 if line.trim().is_empty() {
3169 continue;
3170 }
3171 writeln!(file, "{line}")?;
3172 }
3173 file.flush()?;
3174 file.sync_all()?;
3175 }
3176
3177 let restored_meaning = {
3181 let staged = FeltDb::open(&staging)?;
3182 let inner = staged.state();
3183 let mut rows: Vec<StoredRow> = inner
3184 .rows
3185 .values()
3186 .flat_map(|bucket| bucket.values().cloned())
3187 .filter(|row| !row.deleted)
3188 .collect();
3189 rows.sort_by(|left, right| {
3190 left.capability
3191 .cmp(&right.capability)
3192 .then_with(|| left.key.cmp(&right.key))
3193 });
3194 backup_meaning_digest(&rows)?
3195 };
3196 if restored_meaning != verification.meaning_digest {
3197 remove_restore_staging(&staging)?;
3198 return Err(backup_problem(
3199 "the restored database does not mean what the backup says it means",
3200 None,
3201 "The backup or the restore path is damaged. Do not use this database.",
3202 ));
3203 }
3204 if target.exists() {
3205 remove_restore_staging(&staging)?;
3206 return Err(backup_problem(
3207 "the restore target appeared while the restore was running",
3208 None,
3209 "Restore into a new path. FeltDB will not write over an existing database.",
3210 ));
3211 }
3212 fs::rename(&staging, target)?;
3213 sync_parent_directory(target)?;
3214 remove_restore_staging(&staging)?;
3215
3216 let restored = FeltDb::open(target)?;
3217 Ok((restored, verification))
3218 }
3219
3220 pub fn state_digest(&self) -> Result<String> {
3230 let inner = self.state();
3231 let mut entries: Vec<(String, String, String)> = inner
3232 .rows
3233 .iter()
3234 .filter(|(capability, _)| is_versioned(capability))
3235 .flat_map(|(capability, bucket)| {
3236 bucket
3237 .values()
3238 .filter(|row| !row.deleted)
3239 .map(move |row| (capability.clone(), row.key.clone(), row.value.to_string()))
3240 })
3241 .collect();
3242 entries.sort();
3243
3244 let mut hasher = Sha256::new();
3245 hasher.update(b"feltdb.state.v1");
3246 for (capability, key, value) in entries {
3247 hasher.update(capability.as_bytes());
3248 hasher.update([0u8]);
3249 hasher.update(key.as_bytes());
3250 hasher.update([0u8]);
3251 hasher.update(value.as_bytes());
3252 hasher.update([0u8]);
3253 }
3254 Ok(format!("{:x}", hasher.finalize()))
3255 }
3256
3257 pub fn durability_mode(&self) -> DurabilityMode {
3259 self.state().durability_mode
3260 }
3261
3262 pub fn set_durability_mode(&self, mode: DurabilityMode) {
3267 let mut inner = self.state();
3268 inner.durability_mode = mode;
3269 inner.writes_since_barrier = 0;
3270 }
3271
3272 pub fn health(&self) -> DatabaseHealth {
3283 let recovery = self.log_recovery();
3284 let storage = match &recovery {
3285 LogRecovery::Clean => StorageHealth::Clean,
3286 LogRecovery::RecoveredTornTail {
3287 byte_offset,
3288 discarded_bytes,
3289 } => StorageHealth::RecoveredIncompleteWrite {
3290 byte_offset: *byte_offset,
3291 discarded_bytes: *discarded_bytes,
3292 },
3293 };
3294 let inner = self.state();
3295 DatabaseHealth {
3296 durable_format: inner.durable_format.clone(),
3297 recovery,
3298 storage,
3299 writes: match &inner.write_fault {
3300 None => WriteAvailability::Available,
3301 Some(reason) => WriteAvailability::RecoveryRequired {
3302 reason: reason.clone(),
3303 },
3304 },
3305 }
3306 }
3307
3308 pub fn log_recovery(&self) -> LogRecovery {
3318 self.state().log_recovery.clone()
3319 }
3320
3321 pub fn insert<T: Serialize>(&self, key: &str, value: T) -> Result<()> {
3322 let capability = key
3323 .split_once(':')
3324 .map(|(cap, _)| cap)
3325 .unwrap_or("default")
3326 .to_string();
3327 self.insert_internal(capability, key.to_string(), value, false)?;
3328 Ok(())
3329 }
3330
3331 pub fn insert_if_absent<T: Serialize>(&self, key: &str, value: T) -> Result<()> {
3334 let capability = key
3335 .split_once(':')
3336 .map(|(cap, _)| cap)
3337 .unwrap_or("default")
3338 .to_string();
3339 self.insert_internal(capability, key.to_string(), value, true)
3340 }
3341
3342 pub fn update<T: Serialize>(&self, key: &str, value: T) -> Result<()> {
3344 let capability = key
3345 .split_once(':')
3346 .map(|(cap, _)| cap)
3347 .unwrap_or("default")
3348 .to_string();
3349 self.update_internal(capability, key.to_string(), value)?;
3350 Ok(())
3351 }
3352
3353 pub fn compare_and_set_json(
3357 &self,
3358 key: &str,
3359 expected_version: u64,
3360 expected_epoch: Option<u64>,
3361 expected_lease_id: Option<&str>,
3362 reject_if_diverged: bool,
3363 mut value: Value,
3364 ) -> Result<JsonCasResult> {
3365 let capability = key
3366 .split_once(':')
3367 .map(|(cap, _)| cap)
3368 .unwrap_or("default")
3369 .to_string();
3370 let rust_type = type_name::<Value>().to_string();
3371 let (result, event) = {
3372 let mut inner = self.state();
3373 let Some(current) = inner.rows.get(&capability).and_then(|rows| rows.get(key)) else {
3374 return Ok(JsonCasResult::NotFound);
3375 };
3376 let current_version = current
3377 .value
3378 .get("__version")
3379 .and_then(Value::as_u64)
3380 .unwrap_or(1);
3381 let current_epoch = current
3382 .value
3383 .pointer("/authority/epoch")
3384 .and_then(Value::as_u64)
3385 .unwrap_or(0);
3386 if current_version != expected_version {
3387 return Ok(JsonCasResult::VersionConflict {
3388 current_version,
3389 current_epoch,
3390 });
3391 }
3392 if expected_epoch.is_some_and(|expected| expected != current_epoch) {
3393 return Ok(JsonCasResult::AuthorityConflict {
3394 current_version,
3395 current_epoch,
3396 });
3397 }
3398 let persisted_lease = current.value.get("lease").filter(|lease| !lease.is_null());
3399 let lease_matches = match (persisted_lease, expected_lease_id) {
3400 (None, None) => true,
3401 (None, Some(_)) => false,
3402 (Some(lease), Some(expected_id)) => {
3403 lease.get("leaseId").and_then(Value::as_str) == Some(expected_id)
3404 && lease
3405 .get("expiresAt")
3406 .and_then(Value::as_u64)
3407 .is_some_and(|expires_at| expires_at > now_ms() as u64)
3408 }
3409 (Some(lease), None) => lease
3410 .get("expiresAt")
3411 .and_then(Value::as_u64)
3412 .is_some_and(|expires_at| expires_at <= now_ms() as u64),
3413 };
3414 if !lease_matches {
3415 return Ok(JsonCasResult::LeaseConflict {
3416 current_version,
3417 current_epoch,
3418 });
3419 }
3420 if reject_if_diverged && capability == "_cells" {
3421 let replica_key = key
3422 .strip_prefix("_cells:")
3423 .map(|suffix| format!("_replicas:{suffix}"));
3424 let divergent = replica_key.as_ref().and_then(|replica_key| {
3425 inner
3426 .rows
3427 .get("_replicas")
3428 .and_then(|rows| rows.get(replica_key))
3429 .and_then(|row| row.value.get("divergent"))
3430 .and_then(Value::as_bool)
3431 }) == Some(true);
3432 if divergent {
3433 return Ok(JsonCasResult::ReplicationConflict {
3434 current_version,
3435 current_epoch,
3436 });
3437 }
3438 }
3439
3440 let object = value.as_object_mut().ok_or_else(|| {
3441 FlowError::CapabilityError("CAS value must be a JSON object".to_string())
3442 })?;
3443 object.insert("__version".to_string(), Value::from(expected_version + 1));
3444
3445 let reserved = (inner.sequence, inner.sync_state.vector_clock.clone());
3446 inner.sequence += 1;
3447 inner.sync_state.increment_vector_clock();
3448 let vector_clock = inner.sync_state.get_or_init_vector_clock().clone();
3449 let planned =
3450 is_versioned(&capability).then(|| plan_revision(&inner, key, value.to_string()));
3451 let mut operation = Operation::update(
3452 inner.sequence,
3453 inner.instance_id.clone(),
3454 inner.sequence,
3455 key.to_string(),
3456 value.clone(),
3457 rust_type.clone(),
3458 capability.clone(),
3459 )
3460 .with_vector_clock(vector_clock);
3461 operation.revision = planned.as_ref().map(provenance_of);
3462 let row = StoredRow {
3463 recorded_version: None,
3464 capability: capability.clone(),
3465 key: key.to_string(),
3466 rust_type: rust_type.clone(),
3467 value: value.clone(),
3468 unix_ms: now_ms(),
3469 content_hash: Some(operation.content_hash.clone()),
3470 flow_ref: None,
3471 deleted: false,
3472 operation: Some(operation.clone()),
3473 };
3474 commit_local_write_locked(&mut inner, &row, planned.as_ref(), reserved)?;
3475 inner.authority_revision += 1;
3476 inner.change_log.add_operation(operation);
3477 inner.put_row(row.clone());
3478 if let Some(revision) = planned {
3479 persist_revision_locked(&mut inner, revision)?;
3480 }
3481 let event = ChangeEvent {
3482 capability,
3483 key: key.to_string(),
3484 rust_type,
3485 unix_ms: row.unix_ms,
3486 };
3487 (
3488 JsonCasResult::Updated {
3489 current_version: expected_version + 1,
3490 current_epoch: value
3491 .pointer("/authority/epoch")
3492 .and_then(Value::as_u64)
3493 .unwrap_or(current_epoch),
3494 value,
3495 },
3496 event,
3497 )
3498 };
3499 let _ = self.event_tx.send(event);
3500 Ok(result)
3501 }
3502
3503 pub fn delete(&self, key: &str) -> Result<()> {
3505 let capability = key
3506 .split_once(':')
3507 .map(|(cap, _)| cap)
3508 .unwrap_or("default")
3509 .to_string();
3510 self.delete_internal(capability, key.to_string())?;
3511 Ok(())
3512 }
3513
3514 fn evaluate_authorization(&self, context: &AuthorizationContext) -> Result<()> {
3516 if !context.capabilities.contains(&"state:write".to_string()) {
3518 return Err(FlowError::CapabilityError(
3519 "AUTHORIZATION_DENIED:missing_state_write_capability".to_string(),
3520 ));
3521 }
3522
3523 Ok(())
3527 }
3528
3529 pub fn insert_with_authorization<T: Serialize>(
3531 &self,
3532 key: &str,
3533 value: T,
3534 context: &AuthorizationContext,
3535 ) -> Result<()> {
3536 self.evaluate_authorization(context)?;
3538
3539 let capability = key
3540 .split_once(':')
3541 .map(|(cap, _)| cap)
3542 .unwrap_or("default")
3543 .to_string();
3544
3545 if !context.writable_collections.is_empty()
3547 && !context.writable_collections.contains(&capability)
3548 {
3549 return Err(FlowError::CapabilityError(format!(
3550 "AUTHORIZATION_DENIED:not_writable_collection:{}",
3551 capability
3552 )));
3553 }
3554
3555 self.insert_internal(capability, key.to_string(), value, false)?;
3557 Ok(())
3558 }
3559
3560 pub fn update_with_authorization<T: Serialize>(
3562 &self,
3563 key: &str,
3564 value: T,
3565 context: &AuthorizationContext,
3566 ) -> Result<()> {
3567 self.evaluate_authorization(context)?;
3569
3570 let capability = key
3571 .split_once(':')
3572 .map(|(cap, _)| cap)
3573 .unwrap_or("default")
3574 .to_string();
3575
3576 if !context.writable_collections.is_empty()
3578 && !context.writable_collections.contains(&capability)
3579 {
3580 return Err(FlowError::CapabilityError(format!(
3581 "AUTHORIZATION_DENIED:not_writable_collection:{}",
3582 capability
3583 )));
3584 }
3585
3586 self.update_internal(capability, key.to_string(), value)?;
3589 Ok(())
3590 }
3591
3592 pub fn delete_with_authorization(
3594 &self,
3595 key: &str,
3596 context: &AuthorizationContext,
3597 ) -> Result<()> {
3598 self.evaluate_authorization(context)?;
3600
3601 let capability = key
3602 .split_once(':')
3603 .map(|(cap, _)| cap)
3604 .unwrap_or("default")
3605 .to_string();
3606
3607 if !context.writable_collections.is_empty()
3609 && !context.writable_collections.contains(&capability)
3610 {
3611 return Err(FlowError::CapabilityError(format!(
3612 "AUTHORIZATION_DENIED:not_writable_collection:{}",
3613 capability
3614 )));
3615 }
3616
3617 self.delete_internal(capability, key.to_string())?;
3620 Ok(())
3621 }
3622
3623 pub fn get<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>> {
3624 let capability = key
3625 .split_once(':')
3626 .map(|(cap, _)| cap)
3627 .unwrap_or("default")
3628 .to_string();
3629 let inner = self.state();
3630 let value = inner
3631 .rows
3632 .get(&capability)
3633 .and_then(|bucket| bucket.get(key))
3634 .map(|row| serde_json::from_value::<T>(row.value.clone()))
3635 .transpose()?;
3636 Ok(value)
3637 }
3638
3639 pub fn query<T, F>(&self, predicate: F) -> Result<Vec<T>>
3640 where
3641 T: DeserializeOwned,
3642 F: Fn(&T) -> bool,
3643 {
3644 let rust_type = type_name::<T>().to_string();
3645 let mut inner = self.state();
3646 *inner
3647 .query_count_by_type
3648 .entry(rust_type.clone())
3649 .or_default() += 1;
3650 if inner.query_count_by_type[&rust_type] >= 3 {
3651 inner.adaptive_indexes.insert(rust_type.clone());
3652 }
3653
3654 let mut out = Vec::new();
3655 for bucket in inner.rows.values() {
3656 for row in bucket.values() {
3657 if row.rust_type != rust_type {
3658 continue;
3659 }
3660 let parsed = serde_json::from_value::<T>(row.value.clone())?;
3661 if predicate(&parsed) {
3662 out.push(parsed);
3663 }
3664 }
3665 }
3666 Ok(out)
3667 }
3668
3669 pub fn capability(&self, name: &str) -> Capability {
3670 Capability {
3671 db: self.clone(),
3672 name: name.to_string(),
3673 }
3674 }
3675
3676 pub fn is_auto_indexed<T>(&self) -> bool {
3677 let rust_type = type_name::<T>().to_string();
3678 let inner = self.state();
3679 inner.adaptive_indexes.contains(&rust_type)
3680 }
3681
3682 pub fn instance_id(&self) -> Result<String> {
3684 let inner = self.state();
3685 Ok(inner.instance_id.clone())
3686 }
3687
3688 pub fn sequence(&self) -> Result<u64> {
3690 let inner = self.state();
3691 Ok(inner.sequence)
3692 }
3693
3694 pub fn current_revision(&self) -> Result<u64> {
3698 Ok(self.state().authority_revision)
3699 }
3700
3701 pub(crate) fn authority_state(&self) -> (u64, Vec<StoredRow>) {
3704 let inner = self.state();
3705 let rows = inner
3706 .rows
3707 .values()
3708 .flat_map(|bucket| bucket.values().cloned())
3709 .collect();
3710 (inner.authority_revision, rows)
3711 }
3712
3713 pub fn collection_cardinality(&self, capability: &str) -> Result<u64> {
3716 let inner = self.state();
3717 Ok(inner
3718 .collection_cardinality
3719 .get(capability)
3720 .copied()
3721 .unwrap_or(0))
3722 }
3723
3724 pub fn list_cardinalities(&self) -> Result<Vec<(String, u64)>> {
3727 let inner = self.state();
3728 let mut result: Vec<_> = inner
3729 .collection_cardinality
3730 .iter()
3731 .map(|(k, v)| (k.clone(), *v))
3732 .collect();
3733 result.sort_by(|a, b| b.1.cmp(&a.1)); Ok(result)
3735 }
3736
3737 pub fn diagnostic_row_count(&self, capability_pattern: &str) -> Result<usize> {
3740 let inner = self.state();
3741 Ok(inner
3742 .rows
3743 .iter()
3744 .filter(|(cap, _)| cap.contains(capability_pattern))
3745 .map(|(_, bucket)| bucket.values().filter(|row| !row.deleted).count())
3746 .sum())
3747 }
3748
3749 pub fn diagnostic_capability_keys(&self, capability_pattern: &str) -> Result<Vec<String>> {
3752 let inner = self.state();
3753 let mut keys: Vec<_> = inner
3754 .rows
3755 .keys()
3756 .filter(|k| k.contains(capability_pattern))
3757 .cloned()
3758 .collect();
3759 keys.sort();
3760 Ok(keys)
3761 }
3762
3763 pub(crate) fn revision_head_ids(&self) -> Vec<state_model::StateId> {
3767 self.state()
3768 .revision_heads
3769 .values()
3770 .map(|(id, _)| id.clone())
3771 .collect()
3772 }
3773
3774 pub fn committed_transaction(
3780 &self,
3781 transaction_id: &str,
3782 payload_hash: Option<&str>,
3783 ) -> Result<Option<AtomicCommit>> {
3784 let inner = self.state();
3785 committed_transaction_locked(&inner, transaction_id, payload_hash).transpose()
3786 }
3787
3788 pub fn has_applied_transaction(&self, transaction_id: &str) -> Result<bool> {
3789 Ok(self
3790 .inner
3791 .lock()
3792 .expect("lock poisoned")
3793 .applied_transactions
3794 .contains(transaction_id))
3795 }
3796
3797 pub fn applied_transaction_payload_hash(&self, transaction_id: &str) -> Result<Option<String>> {
3798 Ok(self
3799 .inner
3800 .lock()
3801 .expect("lock poisoned")
3802 .transaction_payload_hashes
3803 .get(transaction_id)
3804 .cloned())
3805 }
3806
3807 pub fn apply_atomic_transaction(
3811 &self,
3812 transaction_id: &str,
3813 expected_parent: Option<u64>,
3814 preconditions: &[AtomicPrecondition],
3815 mutations: &[AtomicMutation],
3816 audit: Option<Value>,
3817 ) -> Result<AtomicCommit> {
3818 self.apply_atomic_transaction_content_addressed(
3819 transaction_id,
3820 None,
3821 expected_parent,
3822 preconditions,
3823 mutations,
3824 audit,
3825 )
3826 }
3827
3828 pub fn apply_atomic_transaction_content_addressed(
3831 &self,
3832 transaction_id: &str,
3833 payload_hash: Option<&str>,
3834 expected_parent: Option<u64>,
3835 preconditions: &[AtomicPrecondition],
3836 mutations: &[AtomicMutation],
3837 audit: Option<Value>,
3838 ) -> Result<AtomicCommit> {
3839 self.apply_atomic_transaction_guarded(
3840 transaction_id,
3841 payload_hash,
3842 expected_parent,
3843 preconditions,
3844 &[],
3845 mutations,
3846 audit,
3847 )
3848 }
3849
3850 #[allow(clippy::too_many_arguments)]
3864 pub fn apply_atomic_transaction_guarded(
3865 &self,
3866 transaction_id: &str,
3867 payload_hash: Option<&str>,
3868 expected_parent: Option<u64>,
3869 preconditions: &[AtomicPrecondition],
3870 record_preconditions: &[RecordPrecondition],
3871 mutations: &[AtomicMutation],
3872 audit: Option<Value>,
3873 ) -> Result<AtomicCommit> {
3874 self.apply_atomic_transaction_fenced(
3875 transaction_id,
3876 payload_hash,
3877 expected_parent,
3878 preconditions,
3879 record_preconditions,
3880 &[],
3881 mutations,
3882 audit,
3883 )
3884 }
3885
3886 pub fn read_basis(&self, reads: &ReadSet) -> Result<ReadBasis> {
3888 self.read_basis_in(reads, &ReadLayout::Direct)
3889 }
3890
3891 pub fn read_basis_in(&self, reads: &ReadSet, layout: &ReadLayout) -> Result<ReadBasis> {
3893 let reads = reads.normalised()?;
3894 let inner = self.state();
3895 let token = read_basis_token(&inner.rows, &reads, layout);
3896 Ok(ReadBasis { reads, token })
3897 }
3898
3899 #[allow(clippy::too_many_arguments)]
3907 pub fn apply_atomic_transaction_fenced(
3908 &self,
3909 transaction_id: &str,
3910 payload_hash: Option<&str>,
3911 expected_parent: Option<u64>,
3912 preconditions: &[AtomicPrecondition],
3913 record_preconditions: &[RecordPrecondition],
3914 fences: &[ReadBasis],
3915 mutations: &[AtomicMutation],
3916 audit: Option<Value>,
3917 ) -> Result<AtomicCommit> {
3918 self.apply_atomic_transaction_fenced_in(
3919 transaction_id,
3920 payload_hash,
3921 expected_parent,
3922 preconditions,
3923 record_preconditions,
3924 &ReadLayout::Direct,
3925 fences,
3926 mutations,
3927 audit,
3928 )
3929 }
3930
3931 #[allow(clippy::too_many_arguments)]
3934 pub fn apply_atomic_transaction_fenced_in(
3935 &self,
3936 transaction_id: &str,
3937 payload_hash: Option<&str>,
3938 expected_parent: Option<u64>,
3939 preconditions: &[AtomicPrecondition],
3940 record_preconditions: &[RecordPrecondition],
3941 layout: &ReadLayout,
3942 fences: &[ReadBasis],
3943 mutations: &[AtomicMutation],
3944 audit: Option<Value>,
3945 ) -> Result<AtomicCommit> {
3946 self.apply_atomic_transaction_collection_fenced_in(
3947 transaction_id,
3948 payload_hash,
3949 expected_parent,
3950 preconditions,
3951 record_preconditions,
3952 layout,
3953 fences,
3954 &[],
3955 mutations,
3956 audit,
3957 )
3958 }
3959
3960 pub(crate) fn apply_atomic_transaction_collection_fenced_in(
3967 &self,
3968 transaction_id: &str,
3969 payload_hash: Option<&str>,
3970 expected_parent: Option<u64>,
3971 preconditions: &[AtomicPrecondition],
3972 record_preconditions: &[RecordPrecondition],
3973 layout: &ReadLayout,
3974 fences: &[ReadBasis],
3975 collection_fences: &[(String, u64)],
3976 mutations: &[AtomicMutation],
3977 audit: Option<Value>,
3978 ) -> Result<AtomicCommit> {
3979 let (commit, events) = {
3980 let mut inner = self.state();
3981 if let Some(duplicate) =
3982 committed_transaction_locked(&inner, transaction_id, payload_hash)
3983 {
3984 return duplicate;
3985 }
3986 for condition in record_preconditions {
3991 if condition.is_empty() {
3992 return Err(FlowError::CapabilityError(format!(
3993 "EMPTY_PRECONDITION:{}:{}",
3994 condition.capability, condition.key
3995 )));
3996 }
3997 let collection = condition.capability.clone();
3998 let key = condition.key.clone();
3999 let current = inner
4000 .rows
4001 .get(&condition.capability)
4002 .and_then(|bucket| bucket.get(&condition.key));
4003
4004 if condition.require_absent {
4005 if current.is_some() {
4006 return Err(FlowError::PreconditionFailed(Box::new(
4007 PreconditionFailure::Present { collection, key },
4008 )));
4009 }
4010 continue;
4011 }
4012
4013 let Some(current) = current else {
4014 return Err(FlowError::PreconditionFailed(Box::new(
4015 PreconditionFailure::Missing { collection, key },
4016 )));
4017 };
4018
4019 if let Some(expected) = condition.expected_version {
4021 let actual = current
4022 .value
4023 .get("__version")
4024 .and_then(Value::as_u64)
4025 .unwrap_or(1);
4026 if actual != expected {
4027 return Err(FlowError::PreconditionFailed(Box::new(
4028 PreconditionFailure::Version {
4029 collection,
4030 key,
4031 expected,
4032 actual,
4033 },
4034 )));
4035 }
4036 }
4037 if let Some(expected) = condition.expected_epoch {
4038 let actual = current
4039 .value
4040 .pointer("/authority/epoch")
4041 .and_then(Value::as_u64)
4042 .unwrap_or(0);
4043 if actual != expected {
4044 return Err(FlowError::PreconditionFailed(Box::new(
4045 PreconditionFailure::Epoch {
4046 collection,
4047 key,
4048 expected,
4049 actual,
4050 },
4051 )));
4052 }
4053 }
4054 if let Some(expected) = &condition.expected_lease_id {
4055 let lease = current.value.get("lease").filter(|lease| !lease.is_null());
4056 let held = lease.and_then(|lease| {
4057 lease
4058 .get("leaseId")
4059 .and_then(Value::as_str)
4060 .map(str::to_string)
4061 });
4062 let unexpired = lease
4065 .and_then(|lease| lease.get("expiresAt").and_then(Value::as_u64))
4066 .is_some_and(|expires_at| expires_at > now_ms() as u64);
4067 if held.as_deref() != Some(expected.as_str()) || !unexpired {
4068 return Err(FlowError::PreconditionFailed(Box::new(
4069 PreconditionFailure::Lease {
4070 collection,
4071 key,
4072 expected: expected.clone(),
4073 actual: held,
4074 },
4075 )));
4076 }
4077 }
4078 }
4079
4080 for fence in fences {
4081 if !fence.token.starts_with(&layout.token_prefix()) {
4082 return Err(FlowError::CapabilityError(
4083 "INVALID_READ_SET:a read basis was not issued by this authority scope"
4084 .into(),
4085 ));
4086 }
4087 let reads = fence.reads.normalised()?;
4088 let actual = read_basis_token(&inner.rows, &reads, layout);
4089 if actual != fence.token {
4090 let collection = reads
4091 .records
4092 .first()
4093 .map(|record| record.collection.clone())
4094 .or_else(|| {
4095 reads
4096 .predicates
4097 .first()
4098 .map(|predicate| predicate.collection.clone())
4099 })
4100 .unwrap_or_default();
4101 return Err(FlowError::PreconditionFailed(Box::new(
4102 PreconditionFailure::Fence {
4103 collection,
4104 key: "read-basis".into(),
4105 expected: fence.token.clone(),
4106 actual,
4107 },
4108 )));
4109 }
4110 }
4111 for (capability, expected) in collection_fences {
4112 let actual = inner.collection_versions.get(capability).copied().unwrap_or(0);
4113 if actual != *expected {
4114 return Err(FlowError::PreconditionFailed(Box::new(
4115 PreconditionFailure::Fence {
4116 collection: capability.clone(),
4117 key: COLLECTION_FENCE_KEY.into(),
4118 expected: format!("collection-version:{expected}"),
4119 actual: format!("collection-version:{actual}"),
4120 },
4121 )));
4122 }
4123 }
4124
4125 let advanced: Vec<AtomicMutation>;
4135 let creates: HashSet<(&str, &str)> = preconditions
4136 .iter()
4137 .filter(|condition| condition.expected_version.is_none())
4138 .map(|condition| (condition.capability.as_str(), condition.key.as_str()))
4139 .chain(
4140 record_preconditions
4141 .iter()
4142 .filter(|condition| condition.require_absent)
4143 .map(|condition| (condition.capability.as_str(), condition.key.as_str())),
4144 )
4145 .collect();
4146 let fenced: HashMap<(&str, &str), u64> = record_preconditions
4147 .iter()
4148 .filter_map(|condition| {
4149 condition.expected_version.map(|version| {
4150 (
4151 (condition.capability.as_str(), condition.key.as_str()),
4152 version,
4153 )
4154 })
4155 })
4156 .collect();
4157 let mutations: &[AtomicMutation] = if fenced.is_empty() && creates.is_empty() {
4158 mutations
4159 } else {
4160 advanced = mutations
4161 .iter()
4162 .map(|mutation| {
4163 let record = (mutation.capability.as_str(), mutation.key.as_str());
4164 let version = if creates.contains(&record) {
4165 Some(1)
4166 } else {
4167 fenced.get(&record).map(|expected| expected + 1)
4168 };
4169 let Some(version) = version else {
4170 return mutation.clone();
4171 };
4172 let Some(Value::Object(fields)) = mutation.value.clone() else {
4175 return mutation.clone();
4176 };
4177 let mut fields = fields;
4178 fields.insert("__version".to_string(), Value::from(version));
4181 AtomicMutation {
4182 capability: mutation.capability.clone(),
4183 key: mutation.key.clone(),
4184 value: Some(Value::Object(fields)),
4185 }
4186 })
4187 .collect();
4188 &advanced
4189 };
4190
4191 let mut bookkeeping_nanos = 0u64;
4198 let prepare_began = workload_diagnostics::enabled().then(std::time::Instant::now);
4199 let base_revision = inner.authority_revision;
4200 if let Some(expected) = expected_parent {
4201 if expected != base_revision {
4202 return Err(FlowError::RevisionConflict {
4203 expected,
4204 actual: base_revision,
4205 });
4206 }
4207 }
4208 for condition in preconditions {
4209 let actual = inner
4210 .rows
4211 .get(&condition.capability)
4212 .and_then(|bucket| bucket.get(&condition.key))
4213 .and_then(StoredRow::version);
4214 if actual != condition.expected_version {
4215 return Err(FlowError::CapabilityError(format!(
4216 "PRECONDITION_FAILED:{}:{}:{:?}:{actual:?}",
4217 condition.capability, condition.key, condition.expected_version
4218 )));
4219 }
4220 }
4221 let state_before = inner.sequence;
4222 let mut next_sequence = state_before;
4223 let mut rows = Vec::with_capacity(mutations.len());
4224 let mut vector_clock = inner.sync_state.get_or_init_vector_clock().clone();
4225 for mutation in mutations {
4226 next_sequence += 1;
4227 *vector_clock.entry(inner.instance_id.clone()).or_default() += 1;
4228 let rust_type = "serde_json::Value".to_string();
4229 let operation = match &mutation.value {
4230 Some(value)
4231 if inner
4232 .rows
4233 .get(&mutation.capability)
4234 .is_some_and(|bucket| bucket.contains_key(&mutation.key)) =>
4235 {
4236 Operation::update(
4237 next_sequence,
4238 inner.instance_id.clone(),
4239 next_sequence,
4240 mutation.key.clone(),
4241 value.clone(),
4242 rust_type.clone(),
4243 mutation.capability.clone(),
4244 )
4245 }
4246 Some(value) => Operation::insert(
4247 next_sequence,
4248 inner.instance_id.clone(),
4249 next_sequence,
4250 mutation.key.clone(),
4251 value.clone(),
4252 rust_type.clone(),
4253 mutation.capability.clone(),
4254 ),
4255 None => Operation::delete(
4256 next_sequence,
4257 inner.instance_id.clone(),
4258 next_sequence,
4259 mutation.key.clone(),
4260 rust_type.clone(),
4261 mutation.capability.clone(),
4262 ),
4263 }
4264 .with_vector_clock(vector_clock.clone());
4265 rows.push(StoredRow {
4266 recorded_version: None,
4267 capability: mutation.capability.clone(),
4268 key: mutation.key.clone(),
4269 rust_type,
4270 value: mutation.value.clone().unwrap_or(Value::Null),
4271 unix_ms: now_ms(),
4272 content_hash: Some(operation.content_hash.clone()),
4273 flow_ref: None,
4274 deleted: mutation.value.is_none(),
4275 operation: Some(operation),
4276 });
4277 }
4278 let mut audit = audit;
4279 if let Some(object) = audit.as_mut().and_then(Value::as_object_mut) {
4280 object.insert("state_before".into(), Value::from(state_before));
4281 object.insert("state_after".into(), Value::from(next_sequence));
4282 }
4283 let record = TransactionLogRecord {
4284 record_type: "feltdb.transaction.v1".into(),
4285 transaction_id: transaction_id.into(),
4286 payload_hash: payload_hash.map(str::to_owned),
4287 state_before,
4288 state_after: next_sequence,
4289 base_revision,
4290 commit_revision: base_revision + 1,
4291 rows: rows.clone(),
4292 audit,
4293 };
4294 if let Some(began) = prepare_began {
4295 bookkeeping_nanos += began.elapsed().as_nanos().min(u64::MAX as u128) as u64;
4296 }
4297 append_transaction_locked(&mut inner, &record)?;
4298 remember_attribution(
4302 &mut inner,
4303 TransactionAttribution::from_record(
4304 transaction_id,
4305 record.audit.as_ref(),
4306 base_revision,
4307 base_revision + 1,
4308 state_before,
4309 next_sequence,
4310 &record.rows,
4311 now_ms(),
4312 ),
4313 );
4314 inner.authority_revision = base_revision + 1;
4315 inner.sequence = next_sequence;
4316 for row in &rows {
4317 if let Some(operation) = row.operation.clone() {
4318 inner.change_log.add_operation(operation);
4319 }
4320 if row.deleted {
4321 inner.remove_row(&row.capability, &row.key);
4322 } else {
4323 inner.put_row(row.clone());
4327 }
4328 }
4329 let tail_began = workload_diagnostics::enabled().then(std::time::Instant::now);
4330 inner.sync_state.merge_vector_clock(&vector_clock);
4331 inner.applied_transactions.insert(transaction_id.into());
4332 let commit_revision = inner.authority_revision;
4333 inner
4334 .transaction_revisions
4335 .insert(transaction_id.into(), (base_revision, commit_revision));
4336 if let Some(hash) = payload_hash {
4337 inner
4338 .transaction_payload_hashes
4339 .insert(transaction_id.into(), hash.into());
4340 }
4341 if let Some(began) = tail_began {
4342 bookkeeping_nanos += began.elapsed().as_nanos().min(u64::MAX as u128) as u64;
4343 }
4344 workload_diagnostics::record(
4345 workload_diagnostics::Phase::TransactionCommit,
4346 bookkeeping_nanos,
4347 );
4348 let events = rows
4349 .iter()
4350 .map(|row| ChangeEvent {
4351 capability: row.capability.clone(),
4352 key: row.key.clone(),
4353 rust_type: row.rust_type.clone(),
4354 unix_ms: row.unix_ms,
4355 })
4356 .collect::<Vec<_>>();
4357 (
4358 AtomicCommit {
4359 transaction_id: transaction_id.into(),
4360 base_revision,
4361 commit_revision: inner.authority_revision,
4362 status: TransactionStatus::Committed,
4363 state_before,
4364 state_after: inner.sequence,
4365 rows,
4366 duplicate: false,
4367 payload_binding: None,
4368 },
4369 events,
4370 )
4371 };
4372 for event in events {
4373 let _ = self.event_tx.send(event);
4374 }
4375 Ok(commit)
4376 }
4377
4378 pub fn apply_atomic_transaction_with_authorization(
4380 &self,
4381 transaction_id: &str,
4382 expected_parent: Option<u64>,
4383 preconditions: &[AtomicPrecondition],
4384 mutations: &[AtomicMutation],
4385 audit: Option<Value>,
4386 context: &AuthorizationContext,
4387 ) -> Result<AtomicCommit> {
4388 self.evaluate_authorization(context)?;
4390
4391 for mutation in mutations {
4393 if !context.writable_collections.is_empty()
4394 && !context.writable_collections.contains(&mutation.capability)
4395 {
4396 return Err(FlowError::CapabilityError(format!(
4397 "AUTHORIZATION_DENIED:not_writable_collection:{}",
4398 mutation.capability
4399 )));
4400 }
4401 }
4402
4403 self.apply_atomic_transaction(
4405 transaction_id,
4406 expected_parent,
4407 preconditions,
4408 mutations,
4409 audit,
4410 )
4411 }
4412
4413 fn insert_internal<T: Serialize>(
4414 &self,
4415 capability: String,
4416 key: String,
4417 value: T,
4418 require_absent: bool,
4419 ) -> Result<()> {
4420 let rust_type = type_name::<T>().to_string();
4421 let value_json = serde_json::to_value(value)?;
4422 let canonical = value_json.to_string();
4425 let mut row = StoredRow {
4426 recorded_version: None,
4427 capability: capability.clone(),
4428 key: key.clone(),
4429 rust_type: rust_type.clone(),
4430 value: value_json.clone(),
4431 unix_ms: now_ms(),
4432 content_hash: None,
4433 flow_ref: None,
4434 deleted: false,
4435 operation: None,
4436 };
4437
4438 {
4439 let mut inner = self.state();
4440
4441 if require_absent
4442 && inner
4443 .rows
4444 .get(&capability)
4445 .and_then(|bucket| bucket.get(&key))
4446 .is_some()
4447 {
4448 return Err(FlowError::PreconditionFailed(Box::new(
4449 PreconditionFailure::Present {
4450 collection: capability,
4451 key,
4452 },
4453 )));
4454 }
4455
4456 let reserved = (inner.sequence, inner.sync_state.vector_clock.clone());
4457 inner.sequence += 1;
4458 inner.sync_state.increment_vector_clock();
4459
4460 let vector_clock = inner.sync_state.get_or_init_vector_clock().clone();
4462
4463 let planned =
4466 is_versioned(&capability).then(|| plan_revision(&inner, &key, canonical.clone()));
4467
4468 let mut op = Operation::insert(
4470 inner.sequence,
4471 inner.instance_id.clone(),
4472 inner.sequence,
4473 key.clone(),
4474 value_json,
4475 rust_type.clone(),
4476 capability.clone(),
4477 )
4478 .with_vector_clock(vector_clock);
4479 op.revision = planned.as_ref().map(provenance_of);
4480
4481 row.operation = Some(op.clone());
4484 commit_local_write_locked(&mut inner, &row, planned.as_ref(), reserved)?;
4485 inner.change_log.add_operation(op);
4486 let base_revision = inner.authority_revision;
4487 inner.authority_revision += 1;
4488 remember_operation_attribution(&mut inner, &row, base_revision, base_revision + 1);
4489 inner.put_row(row);
4490 if let Some(revision) = planned {
4491 persist_revision_locked(&mut inner, revision)?;
4492 }
4493 let suffix = key.rsplit_once(':').map(|(_, right)| right);
4494 if let Some(id) = suffix.and_then(|s| s.parse::<u64>().ok()) {
4495 let entry = inner.key_counters.entry(capability.clone()).or_default();
4496 *entry = (*entry).max(id + 1);
4497 }
4498 }
4499
4500 let event = ChangeEvent {
4501 capability,
4502 key,
4503 rust_type,
4504 unix_ms: now_ms(),
4505 };
4506 let _ = self.event_tx.send(event);
4507
4508 Ok(())
4509 }
4510
4511 fn update_internal<T: Serialize>(
4512 &self,
4513 capability: String,
4514 key: String,
4515 value: T,
4516 ) -> Result<()> {
4517 let rust_type = type_name::<T>().to_string();
4518 let value_json = serde_json::to_value(value)?;
4519 let canonical = value_json.to_string();
4522 let mut row = StoredRow {
4523 recorded_version: None,
4524 capability: capability.clone(),
4525 key: key.clone(),
4526 rust_type: rust_type.clone(),
4527 value: value_json.clone(),
4528 unix_ms: now_ms(),
4529 content_hash: None,
4530 flow_ref: None,
4531 deleted: false,
4532 operation: None,
4533 };
4534
4535 {
4536 let mut inner = self.state();
4537
4538 let reserved = (inner.sequence, inner.sync_state.vector_clock.clone());
4539 inner.sequence += 1;
4540 inner.sync_state.increment_vector_clock();
4541
4542 let vector_clock = inner.sync_state.get_or_init_vector_clock().clone();
4544
4545 let planned =
4546 is_versioned(&capability).then(|| plan_revision(&inner, &key, canonical.clone()));
4547
4548 let mut op = Operation::update(
4550 inner.sequence,
4551 inner.instance_id.clone(),
4552 inner.sequence,
4553 key.clone(),
4554 value_json,
4555 rust_type.clone(),
4556 capability.clone(),
4557 )
4558 .with_vector_clock(vector_clock);
4559 op.revision = planned.as_ref().map(provenance_of);
4560
4561 row.operation = Some(op.clone());
4564 commit_local_write_locked(&mut inner, &row, planned.as_ref(), reserved)?;
4565 inner.change_log.add_operation(op);
4566 let base_revision = inner.authority_revision;
4567 inner.authority_revision += 1;
4568 remember_operation_attribution(&mut inner, &row, base_revision, base_revision + 1);
4569 inner.put_row(row);
4570 if let Some(revision) = planned {
4571 persist_revision_locked(&mut inner, revision)?;
4572 }
4573 }
4574
4575 let event = ChangeEvent {
4576 capability,
4577 key,
4578 rust_type,
4579 unix_ms: now_ms(),
4580 };
4581 let _ = self.event_tx.send(event);
4582
4583 Ok(())
4584 }
4585
4586 fn delete_internal(&self, capability: String, key: String) -> Result<()> {
4587 {
4588 let mut inner = self.state();
4589
4590 let rust_type = inner
4592 .rows
4593 .get(&capability)
4594 .and_then(|bucket| bucket.get(&key))
4595 .map(|row| row.rust_type.clone())
4596 .unwrap_or_else(|| "Unknown".to_string());
4597
4598 let reserved = (inner.sequence, inner.sync_state.vector_clock.clone());
4599 inner.sequence += 1;
4600 inner.sync_state.increment_vector_clock();
4601
4602 let vector_clock = inner.sync_state.get_or_init_vector_clock().clone();
4604
4605 let op = Operation::delete(
4607 inner.sequence,
4608 inner.instance_id.clone(),
4609 inner.sequence,
4610 key.clone(),
4611 rust_type.clone(),
4612 capability.clone(),
4613 )
4614 .with_vector_clock(vector_clock);
4615
4616 let tombstone = StoredRow {
4620 recorded_version: None,
4621 capability: capability.clone(),
4622 key: key.clone(),
4623 rust_type,
4624 value: serde_json::Value::Null,
4625 unix_ms: now_ms(),
4626 content_hash: None,
4627 flow_ref: None,
4628 deleted: true,
4629 operation: Some(op.clone()),
4630 };
4631 commit_local_write_locked(&mut inner, &tombstone, None, reserved)?;
4632 inner.change_log.add_operation(op);
4634 let base_revision = inner.authority_revision;
4635 inner.authority_revision += 1;
4636 remember_operation_attribution(
4637 &mut inner,
4638 &tombstone,
4639 base_revision,
4640 base_revision + 1,
4641 );
4642
4643 inner.remove_row(&capability, &key);
4645 }
4646
4647 let event = ChangeEvent {
4648 capability,
4649 key,
4650 rust_type: "Deleted".to_string(),
4651 unix_ms: now_ms(),
4652 };
4653 let _ = self.event_tx.send(event);
4654
4655 Ok(())
4656 }
4657 pub async fn live_query<T, F>(&self, predicate: F) -> Result<LiveQuerySubscription<T>>
4677 where
4678 T: DeserializeOwned + Send + Sync + 'static,
4679 F: Fn(&T) -> bool + Send + Sync + 'static,
4680 {
4681 let rust_type = type_name::<T>().to_string();
4682 {
4683 let mut inner = self.state();
4684 *inner
4685 .query_count_by_type
4686 .entry(rust_type.clone())
4687 .or_default() += 1;
4688 if inner.query_count_by_type[&rust_type] >= 3 {
4689 inner.adaptive_indexes.insert(rust_type.clone());
4690 }
4691 }
4692 let rx = self.event_tx.subscribe();
4694 let mut matches = LiveMatches::new(rust_type, None);
4695 matches.rescan(self, &predicate);
4696 let subscription = LiveQuerySubscription {
4697 initial_results: matches.results(),
4698 rx,
4699 db: self.clone(),
4700 matches,
4701 predicate: Arc::new(Mutex::new(Some(Box::new(predicate)))),
4702 };
4703
4704 Ok(subscription)
4705 }
4706
4707 pub fn sync_info(&self) -> Result<SyncState> {
4709 let inner = self.state();
4710 Ok(inner.sync_state.clone())
4711 }
4712
4713 pub fn authority_time_ms(&self) -> u64 {
4719 now_ms() as u64
4720 }
4721
4722 pub fn get_value(&self, key: &str) -> Result<Option<Value>> {
4724 self.get::<Value>(key)
4725 }
4726
4727 pub(crate) fn live_row(&self, capability: &str, key: &str) -> Option<StoredRow> {
4733 let inner = self.state();
4734 inner
4735 .rows
4736 .get(capability)
4737 .and_then(|rows| rows.get(key))
4738 .filter(|row| !row.deleted)
4739 .cloned()
4740 }
4741
4742 pub(crate) fn consistent_rows(
4754 &self,
4755 capabilities: &BTreeSet<String>,
4756 records: &BTreeSet<(String, String)>,
4757 ) -> ConsistentRows {
4758 let inner = self.state();
4759 let versions = capabilities
4760 .iter()
4761 .map(|capability| {
4762 (
4763 capability.clone(),
4764 inner.collection_versions.get(capability).copied().unwrap_or(0),
4765 )
4766 })
4767 .collect();
4768 let mut view: HashMap<String, BTreeMap<String, StoredRow>> = HashMap::new();
4769 for capability in capabilities {
4770 if let Some(bucket) = inner.rows.get(capability) {
4771 query_execution_diagnostics::record_full_materialization(bucket.len());
4772 view.insert(capability.clone(), bucket.clone());
4773 }
4774 }
4775 for (capability, key) in records {
4776 if capabilities.contains(capability) {
4777 continue;
4778 }
4779 if let Some(row) = inner
4780 .rows
4781 .get(capability)
4782 .and_then(|bucket| bucket.get(key))
4783 .filter(|row| !row.deleted)
4784 {
4785 view.entry(capability.clone())
4786 .or_default()
4787 .insert(key.clone(), row.clone());
4788 }
4789 }
4790 ConsistentRows { view, versions }
4791 }
4792
4793 pub fn list_collection(&self, capability: &str) -> Result<Vec<StoredRow>> {
4794 let inner = self.state();
4795 let rows: Vec<StoredRow> = inner
4796 .rows
4797 .get(capability)
4798 .map(|rows| rows.values().cloned().collect())
4799 .unwrap_or_default();
4800 query_execution_diagnostics::record_full_materialization(rows.len());
4801 Ok(rows)
4802 }
4803
4804 pub fn query_collection<F>(
4821 &self,
4822 capability: &str,
4823 limit: Option<usize>,
4824 mut predicate: F,
4825 ) -> Result<Vec<StoredRow>>
4826 where
4827 F: FnMut(&StoredRow) -> bool,
4828 {
4829 let mut matched = Vec::new();
4830 let mut visited = 0usize;
4831 {
4832 let inner = self.state();
4833 let _query = workload_diagnostics::span(workload_diagnostics::Phase::ScanQuery);
4834 if let Some(rows) = inner.rows.get(capability) {
4835 for row in rows.values() {
4836 if limit.is_some_and(|bound| matched.len() >= bound) {
4837 break;
4838 }
4839 visited += 1;
4840 let _predicate = workload_diagnostics::span(
4844 workload_diagnostics::Phase::PredicateEvaluation,
4845 );
4846 if predicate(row) {
4847 drop(_predicate);
4848 matched.push(row.clone());
4849 }
4850 }
4851 }
4852 }
4853 query_execution_diagnostics::record_scan(visited, matched.len());
4854 Ok(matched)
4855 }
4856
4857 fn state(&self) -> StateGuard<'_> {
4873 if !workload_diagnostics::enabled() {
4874 return StateGuard {
4875 inner: self.inner.lock().expect("lock poisoned"),
4876 held_since: None,
4877 };
4878 }
4879 let requested = std::time::Instant::now();
4880 let inner = self.inner.lock().expect("lock poisoned");
4881 let acquired = std::time::Instant::now();
4882 workload_diagnostics::record(
4883 workload_diagnostics::Phase::StateLockWait,
4884 acquired
4885 .duration_since(requested)
4886 .as_nanos()
4887 .min(u64::MAX as u128) as u64,
4888 );
4889 StateGuard {
4890 inner,
4891 held_since: Some(acquired),
4892 }
4893 }
4894
4895 pub fn create_equality_index(&self, collection: &str, field: &str) -> Result<bool> {
4907 let mut inner = self.state();
4908 let created = inner
4909 .equality_index
4910 .declare(collection, field)
4911 .map_err(FlowError::CapabilityError)?;
4912 if created {
4913 let Inner {
4914 rows,
4915 equality_index,
4916 ..
4917 } = &mut *inner;
4918 equality_index.rebuild_field(collection, field, rows);
4919 }
4920 Ok(created)
4921 }
4922
4923 pub fn drop_equality_index(&self, collection: &str, field: &str) -> Result<bool> {
4925 let mut inner = self.state();
4926 Ok(inner.equality_index.undeclare(collection, field))
4927 }
4928
4929 pub fn equality_indexes(&self) -> Result<Vec<(String, String)>> {
4931 let inner = self.state();
4932 Ok(inner.equality_index.declarations())
4933 }
4934
4935 pub fn rebuild_equality_indexes(&self) -> Result<EqualityIndexStats> {
4941 let mut inner = self.state();
4942 let Inner {
4943 rows,
4944 equality_index,
4945 ..
4946 } = &mut *inner;
4947 equality_index.rebuild(rows);
4948 Ok(equality_index.stats())
4949 }
4950
4951 pub fn equality_index_stats(&self) -> Result<EqualityIndexStats> {
4953 let inner = self.state();
4954 Ok(inner.equality_index.stats())
4955 }
4956
4957 #[doc(hidden)]
4963 pub fn equality_index_snapshot(
4964 &self,
4965 ) -> Result<BTreeMap<String, BTreeMap<String, BTreeMap<String, Vec<String>>>>> {
4966 let inner = self.state();
4967 Ok(inner.equality_index.snapshot())
4968 }
4969
4970 pub fn verify_equality_index(&self) -> Result<std::result::Result<(), String>> {
4978 let inner = self.state();
4979 let live = inner.equality_index.snapshot();
4980 let mut expected = inner.equality_index.clone();
4981 expected.clear();
4982 expected.rebuild(&inner.rows);
4983 let expected = expected.snapshot();
4984 if live == expected {
4985 return Ok(Ok(()));
4986 }
4987 for (collection, fields) in expected.iter() {
4988 for (field, buckets) in fields {
4989 let observed = live.get(collection).and_then(|fields| fields.get(field));
4990 if observed != Some(buckets) {
4991 return Ok(Err(format!(
4992 "index for {collection}.{field} diverges from authoritative state: \
4993 {} value buckets held, {} derived",
4994 observed.map(BTreeMap::len).unwrap_or(0),
4995 buckets.len(),
4996 )));
4997 }
4998 }
4999 }
5000 Ok(Err(
5001 "the live index holds entries no authoritative record derives".to_string(),
5002 ))
5003 }
5004
5005 pub fn equality_index_candidates(
5014 &self,
5015 collection: &str,
5016 field: &str,
5017 value: &Value,
5018 ) -> Result<Option<Vec<String>>> {
5019 let inner = self.state();
5020 Ok(inner
5021 .equality_index
5022 .candidates(collection, field, value)
5023 .map(|keys| keys.iter().cloned().collect()))
5024 }
5025
5026 pub fn query_collection_by_equality<F>(
5047 &self,
5048 capability: &str,
5049 equalities: &[(&str, &Value)],
5050 mut predicate: F,
5051 ) -> Result<Option<Vec<StoredRow>>>
5052 where
5053 F: FnMut(&StoredRow) -> bool,
5054 {
5055 let mut matched = Vec::new();
5056 let mut examined = 0usize;
5057 let mut evaluated = 0usize;
5058 {
5059 let inner = self.state();
5060 let _query = workload_diagnostics::span(workload_diagnostics::Phase::IndexedQuery);
5061 let lookup = workload_diagnostics::span(workload_diagnostics::Phase::IndexLookup);
5066 let mut applicable: Vec<&BTreeSet<String>> = Vec::new();
5067 for (field, value) in equalities {
5068 if let Some(keys) = inner.equality_index.candidates(capability, field, value) {
5069 applicable.push(keys);
5070 }
5071 }
5072 if applicable.is_empty() {
5073 drop(lookup);
5074 drop(_query);
5075 drop(inner);
5076 query_execution_diagnostics::record_index_miss();
5077 return Ok(None);
5078 }
5079 applicable.sort_by_key(|keys| keys.len());
5083 drop(lookup);
5084 let (smallest, rest) = applicable.split_first().expect("non-empty");
5085 let bucket = inner.rows.get(capability);
5086 for key in smallest.iter() {
5087 if !rest.iter().all(|keys| keys.contains(key)) {
5088 continue;
5089 }
5090 examined += 1;
5091 let retrieval =
5092 workload_diagnostics::span(workload_diagnostics::Phase::CandidateRetrieval);
5093 let row = bucket.and_then(|rows| rows.get(key));
5094 drop(retrieval);
5095 let Some(row) = row else {
5096 continue;
5100 };
5101 evaluated += 1;
5102 let _predicate =
5103 workload_diagnostics::span(workload_diagnostics::Phase::PredicateEvaluation);
5104 if predicate(row) {
5105 drop(_predicate);
5106 matched.push(row.clone());
5107 }
5108 }
5109 }
5110 query_execution_diagnostics::record_indexed_query(examined, evaluated, matched.len());
5111 Ok(Some(matched))
5112 }
5113
5114 #[allow(clippy::too_many_arguments)]
5138 pub fn select_page(
5139 &self,
5140 capability: &str,
5141 equalities: &[(&str, &Value)],
5142 key_order: Option<Option<&str>>,
5143 limit: usize,
5144 mut matches: impl FnMut(&StoredRow) -> bool,
5145 mut after: impl FnMut(&StoredRow) -> bool,
5146 mut compare: impl FnMut(&StoredRow, &StoredRow) -> std::cmp::Ordering,
5147 ) -> Result<SelectedPage> {
5148 let keep = limit.saturating_add(1);
5149 let inner = self.state();
5150 let Some(bucket) = inner.rows.get(capability) else {
5151 let indexed = equalities
5156 .iter()
5157 .any(|(field, value)| inner.equality_index.candidates(capability, field, value).is_some());
5158 let access = match (indexed, key_order) {
5159 (true, _) => PageAccess::EqualityIndex,
5160 (false, Some(_)) => PageAccess::KeyRange,
5161 (false, None) => PageAccess::Scan,
5162 };
5163 return Ok(SelectedPage::empty(access));
5164 };
5165
5166 let mut applicable: Vec<&BTreeSet<String>> = equalities
5168 .iter()
5169 .filter_map(|(field, value)| inner.equality_index.candidates(capability, field, value))
5170 .collect();
5171 applicable.sort_by_key(|keys| keys.len());
5172 if applicable.is_empty() && !equalities.is_empty() {
5173 query_execution_diagnostics::record_index_miss();
5175 }
5176 let access = if applicable.is_empty() {
5177 match key_order {
5178 Some(_) => PageAccess::KeyRange,
5179 None => PageAccess::Scan,
5180 }
5181 } else {
5182 PageAccess::EqualityIndex
5183 };
5184
5185 let mut examined = 0usize;
5186 let mut matched = 0usize;
5187 let mut held: Vec<&StoredRow> = Vec::new();
5188 let mut consider = |row: &StoredRow| -> bool {
5191 examined += 1;
5192 if row.deleted || !matches(row) {
5193 return false;
5194 }
5195 matched += 1;
5196 after(row)
5197 };
5198
5199 match (applicable.split_first(), key_order) {
5200 (None, Some(after_key)) => {
5203 let range: Box<dyn Iterator<Item = &StoredRow>> = match after_key {
5204 Some(key) => Box::new(
5205 bucket
5206 .range::<str, _>((Excluded(key), Unbounded))
5207 .map(|(_, row)| row),
5208 ),
5209 None => Box::new(bucket.values()),
5210 };
5211 for row in range {
5212 if consider(row) {
5213 held.push(row);
5214 }
5215 if held.len() >= keep {
5216 break;
5217 }
5218 }
5219 }
5220 (Some((smallest, rest)), Some(after_key)) => {
5221 let keys: Box<dyn Iterator<Item = &String>> = match after_key {
5222 Some(key) => Box::new(
5223 smallest.range::<str, _>((Excluded(key), Unbounded)),
5224 ),
5225 None => Box::new(smallest.iter()),
5226 };
5227 for key in keys {
5228 if !rest.iter().all(|keys| keys.contains(key)) {
5229 continue;
5230 }
5231 if let Some(row) = bucket.get(key) {
5232 if consider(row) {
5233 held.push(row);
5234 }
5235 if held.len() >= keep {
5236 break;
5237 }
5238 }
5239 }
5240 }
5241 (applicable, None) => {
5243 let mut trim = |held: &mut Vec<&StoredRow>| {
5244 if held.len() >= keep.saturating_mul(2) {
5245 held.select_nth_unstable_by(keep - 1, |left, right| compare(left, right));
5246 held.truncate(keep);
5247 }
5248 };
5249 match applicable {
5250 Some((smallest, rest)) => {
5251 for key in smallest.iter() {
5252 if !rest.iter().all(|keys| keys.contains(key)) {
5253 continue;
5254 }
5255 if let Some(row) = bucket.get(key) {
5256 if consider(row) {
5257 held.push(row);
5258 trim(&mut held);
5259 }
5260 }
5261 }
5262 }
5263 None => {
5264 for row in bucket.values() {
5265 if consider(row) {
5266 held.push(row);
5267 trim(&mut held);
5268 }
5269 }
5270 }
5271 }
5272 }
5273 }
5274
5275 held.sort_by(|left, right| compare(left, right));
5276 let has_more = held.len() > limit;
5277 held.truncate(limit);
5278 let rows: Vec<StoredRow> = held.into_iter().cloned().collect();
5279 drop(inner);
5280 match access {
5281 PageAccess::EqualityIndex => {
5282 query_execution_diagnostics::record_indexed_query(examined, examined, rows.len())
5283 }
5284 _ => query_execution_diagnostics::record_scan(examined, rows.len()),
5285 }
5286 Ok(SelectedPage {
5287 rows,
5288 has_more,
5289 examined,
5290 matched_examined: matched,
5291 access,
5292 })
5293 }
5294
5295 pub fn list_collection_page(
5297 &self,
5298 capability: &str,
5299 after: Option<&str>,
5300 limit: usize,
5301 ) -> Result<Vec<StoredRow>> {
5302 let inner = self.state();
5303 let Some(rows) = inner.rows.get(capability) else {
5304 return Ok(vec![]);
5305 };
5306 let values: Box<dyn Iterator<Item = &StoredRow>> = match after {
5307 Some(key) => Box::new(
5308 rows.range::<str, _>((Excluded(key), Unbounded))
5309 .map(|(_, row)| row),
5310 ),
5311 None => Box::new(rows.values()),
5312 };
5313 Ok(values
5314 .filter(|row| !row.deleted)
5315 .take(limit)
5316 .cloned()
5317 .collect())
5318 }
5319
5320 pub fn collection_write_versions(&self, capabilities: &[String]) -> Vec<u64> {
5326 let inner = self.state();
5327 capabilities
5328 .iter()
5329 .map(|capability| inner.collection_versions.get(capability).copied().unwrap_or(0))
5330 .collect()
5331 }
5332
5333 pub fn get_collection_record(&self, capability: &str, key: &str) -> Result<Option<StoredRow>> {
5334 let inner = self.state();
5335 Ok(inner
5336 .rows
5337 .get(capability)
5338 .and_then(|rows| rows.get(key))
5339 .filter(|row| !row.deleted)
5340 .cloned())
5341 }
5342
5343 pub fn get_collection_records<'a>(
5345 &self,
5346 capability: &str,
5347 keys: impl IntoIterator<Item = &'a str>,
5348 ) -> Result<Vec<StoredRow>> {
5349 let inner = self.state();
5350 let Some(rows) = inner.rows.get(capability) else {
5351 return Ok(Vec::new());
5352 };
5353 Ok(keys
5354 .into_iter()
5355 .filter_map(|key| rows.get(key))
5356 .filter(|row| !row.deleted)
5357 .cloned()
5358 .collect())
5359 }
5360
5361 pub fn subscribe_changes(&self) -> broadcast::Receiver<ChangeEvent> {
5363 self.event_tx.subscribe()
5364 }
5365
5366 pub fn add_sync_peer(&self, peer_id: String) -> Result<()> {
5368 let mut inner = self.state();
5369 inner.sync_state.add_peer(peer_id);
5370 Ok(())
5371 }
5372
5373 pub fn remove_sync_peer(&self, peer_id: &str) -> Result<()> {
5375 let mut inner = self.state();
5376 inner.sync_state.remove_peer(peer_id);
5377 Ok(())
5378 }
5379
5380 pub fn get_pending_for_peer(
5382 &self,
5383 _peer_id: &str,
5384 since_sequence: u64,
5385 ) -> Result<Vec<Operation>> {
5386 let inner = self.state();
5387 let pending: Vec<Operation> = inner
5388 .change_log
5389 .pending_operations()
5390 .into_iter()
5391 .filter(|op| op.sequence > since_sequence)
5392 .collect();
5393 Ok(pending)
5394 }
5395
5396 pub fn operations_since(
5398 &self,
5399 versions: &std::collections::HashMap<String, u64>,
5400 ) -> Result<Vec<Operation>> {
5401 let inner = self.state();
5402 Ok(inner.change_log.operations_since(versions))
5403 }
5404
5405 pub fn operation_versions(&self) -> Result<std::collections::HashMap<String, u64>> {
5407 let inner = self.state();
5408 Ok(inner.change_log.versions())
5409 }
5410
5411 pub fn version_history(&self) -> VersionHistory {
5413 self.history.read().expect("history lock poisoned").clone()
5414 }
5415
5416 pub fn version_token(&self, version: u64) -> String {
5418 self.version_history().token(version)
5419 }
5420
5421 fn publish_history(&self, history: &VersionHistory) {
5422 *self.history.write().expect("history lock poisoned") = history.clone();
5423 }
5424
5425 pub fn begin_restored_history(&self) -> Result<VersionHistory> {
5430 let mut inner = self.state();
5431 refuse_if_faulted(&inner)?;
5432 inner.history = VersionHistory::new(HistoryOrigin::Restored);
5433 persist_sync_metadata(&inner)?;
5434 self.publish_history(&inner.history);
5435 Ok(inner.history.clone())
5436 }
5437
5438 pub fn accept_client_version(
5449 &self,
5450 collection: &str,
5451 key: &str,
5452 version: Option<u64>,
5453 token: Option<&str>,
5454 ) -> Result<Option<u64>> {
5455 let history = self.version_history();
5456 let refused = |expected: Option<String>| {
5457 FlowError::PreconditionFailed(Box::new(PreconditionFailure::History {
5458 collection: collection.to_string(),
5459 key: key.to_string(),
5460 expected,
5461 actual: history.id.clone(),
5462 }))
5463 };
5464 match token {
5465 Some(token) => {
5466 let (issued_in, number) = parse_version_token(token).ok_or_else(|| {
5467 FlowError::CapabilityError(format!(
5468 "INVALID_VERSION_TOKEN: {token:?} is not a version token (fv1:<history>:<version>)"
5469 ))
5470 })?;
5471 if version.is_some_and(|version| version != number) {
5472 return Err(FlowError::CapabilityError(format!(
5473 "INVALID_VERSION_TOKEN: {token:?} names version {number}, but version {} was also given",
5474 version.unwrap_or_default()
5475 )));
5476 }
5477 if issued_in != history.id {
5478 return Err(refused(Some(issued_in.to_string())));
5479 }
5480 Ok(Some(number))
5481 }
5482 None => match (version, history.origin) {
5483 (None, _) => Ok(None),
5484 (Some(version), HistoryOrigin::Genesis) => Ok(Some(version)),
5485 (Some(_), HistoryOrigin::Restored) => Err(refused(None)),
5486 },
5487 }
5488 }
5489
5490 pub fn export_snapshot(&self) -> Result<DatabaseSnapshot> {
5491 let inner = self.state();
5492 let mut rows: Vec<StoredRow> = inner
5493 .rows
5494 .values()
5495 .flat_map(|bucket| bucket.values().map(snapshot_row))
5496 .collect();
5497 rows.sort_by(|left, right| {
5502 left.key.cmp(&right.key).then_with(|| left.capability.cmp(&right.capability))
5503 });
5504 let versions = inner.change_log.versions();
5505 let content_hash = DatabaseSnapshot::compute_hash(&rows, &versions)?;
5506 let snapshot = DatabaseSnapshot {
5507 rows,
5508 versions,
5509 content_hash,
5510 format_version: DURABLE_FORMAT_VERSION,
5511 history: Some(inner.history.clone()),
5512 };
5513 snapshot.validate()?;
5515 Ok(snapshot)
5516 }
5517 pub fn state_rows(&self) -> Result<Vec<StoredRow>> {
5521 let inner = self.state();
5522 Ok(inner
5523 .rows
5524 .values()
5525 .flat_map(|bucket| bucket.values().cloned())
5526 .collect())
5527 }
5528
5529 pub fn state_rows_for_namespace(&self, state_namespace: &str) -> Result<Vec<StoredRow>> {
5535 let _span = workload_diagnostics::span(workload_diagnostics::Phase::NamespaceSnapshot);
5536 let inner = self.state();
5537 let prefix = format!("{state_namespace}:");
5538 Ok(inner
5539 .rows
5540 .iter()
5541 .filter(|(capability, _)| capability.starts_with(&prefix))
5542 .flat_map(|(_, bucket)| bucket.values().cloned())
5543 .collect())
5544 }
5545
5546 pub fn can_install_snapshot(&self) -> Result<bool> {
5547 let inner = self.state();
5548 Ok(!inner.bootstrapped && inner.rows.is_empty() && inner.change_log.pending.is_empty())
5549 }
5550
5551 pub fn install_snapshot(&self, snapshot: DatabaseSnapshot) -> Result<()> {
5554 snapshot.validate()?;
5557 let mut inner = self.state();
5558 refuse_if_faulted(&inner)?;
5559 if !inner.rows.is_empty() || !inner.change_log.pending.is_empty() {
5560 return Err(FlowError::CapabilityError(
5561 "snapshot bootstrap requires a pristine node".to_string(),
5562 ));
5563 }
5564
5565 let temporary = sibling(&inner.path, "snapshot.tmp");
5566 let mut file = OpenOptions::new()
5567 .create(true)
5568 .write(true)
5569 .truncate(true)
5570 .open(&temporary)?;
5571 write_json_line(
5572 &mut file,
5573 &LogHeader {
5574 record_type: "feltdb.snapshot.v1".to_string(),
5575 format_version: DURABLE_FORMAT_VERSION,
5576 local_sequence: inner.sequence,
5577 observed_versions: snapshot.versions.clone(),
5578 transaction_history: Vec::new(),
5579 transaction_payload_hashes: HashMap::new(),
5580 authority_revision: 0,
5582 },
5583 )?;
5584 for row in &snapshot.rows {
5585 write_json_line(&mut file, row)?;
5586 }
5587 file.flush()?;
5588 file.sync_all()?;
5589 fs::rename(temporary, &inner.path)?;
5590 sync_parent_directory(&inner.path)?;
5591
5592 inner.clear_rows();
5593 inner.key_counters.clear();
5594 for row in snapshot.rows {
5595 if let Some(id) = row
5596 .key
5597 .rsplit_once(':')
5598 .and_then(|(_, right)| right.parse::<u64>().ok())
5599 {
5600 let counter = inner
5601 .key_counters
5602 .entry(row.capability.clone())
5603 .or_default();
5604 *counter = (*counter).max(id + 1);
5605 }
5606 inner.put_row(row);
5607 }
5608 inner.change_log.observed_versions = snapshot.versions;
5609 inner.bootstrapped = true;
5610 inner.history = snapshot
5613 .history
5614 .unwrap_or_else(|| VersionHistory::new(HistoryOrigin::Genesis));
5615 self.publish_history(&inner.history);
5616 persist_sync_metadata(&inner)
5617 }
5618
5619 pub fn acknowledge_peer_operations(&self, peer_id: String, sequence: u64) -> Result<()> {
5621 let mut inner = self.state();
5622 inner.change_log.acknowledge_peer(peer_id, sequence);
5623 persist_sync_metadata(&inner)?;
5624 Ok(())
5625 }
5626
5627 pub fn acknowledge_peer_versions(
5629 &self,
5630 peer_id: String,
5631 versions: HashMap<String, u64>,
5632 ) -> Result<()> {
5633 let mut inner = self.state();
5634 inner.change_log.acknowledge_versions(peer_id, versions);
5635 persist_sync_metadata(&inner)
5636 }
5637
5638 pub fn acknowledged_peer_versions(&self, peer_id: &str) -> Result<HashMap<String, u64>> {
5639 let inner = self.state();
5640 Ok(inner.change_log.acknowledged_versions(peer_id))
5641 }
5642
5643 pub fn recent_transactions(&self, limit: usize) -> Result<Vec<TransactionAttribution>> {
5651 let inner = self.state();
5652 Ok(inner
5653 .transaction_attributions
5654 .iter()
5655 .rev()
5656 .take(limit)
5657 .cloned()
5658 .collect())
5659 }
5660
5661 pub fn transaction_attribution(
5664 &self,
5665 transaction_id: &str,
5666 ) -> Result<Option<TransactionAttribution>> {
5667 let inner = self.state();
5668 Ok(inner
5669 .transaction_history
5670 .iter()
5671 .rev()
5672 .find(|attribution| attribution.transaction_id == transaction_id)
5673 .cloned())
5674 }
5675
5676 pub fn committed_transactions_after(
5679 &self,
5680 commit_revision: u64,
5681 limit: usize,
5682 ) -> Result<Vec<TransactionAttribution>> {
5683 let inner = self.state();
5684 Ok(inner
5685 .transaction_history
5686 .iter()
5687 .filter(|transaction| transaction.commit_revision > commit_revision)
5688 .take(limit)
5689 .cloned()
5690 .collect())
5691 }
5692
5693 pub fn transactions_touching(
5699 &self,
5700 capability: &str,
5701 key: &str,
5702 limit: usize,
5703 ) -> Result<Vec<TransactionAttribution>> {
5704 let inner = self.state();
5705 Ok(inner
5706 .transaction_history
5707 .iter()
5708 .rev()
5709 .filter(|attribution| attribution.touched(capability, key))
5710 .take(limit)
5711 .cloned()
5712 .collect())
5713 }
5714
5715 pub fn transaction_attribution_window(&self) -> Result<(usize, usize)> {
5720 let inner = self.state();
5721 Ok((
5722 inner.transaction_attributions.len(),
5723 TRANSACTION_ATTRIBUTION_WINDOW,
5724 ))
5725 }
5726
5727 pub fn retained_operation_floors(&self) -> Result<HashMap<String, u64>> {
5728 let inner = self.state();
5729 Ok(inner.change_log.retained_from())
5730 }
5731
5732 pub fn acknowledged_operation_count(&self, active_peers: &[String]) -> Result<usize> {
5733 let inner = self.state();
5734 Ok(inner.change_log.acknowledged_operation_count(active_peers))
5735 }
5736
5737 pub fn compact_operation_log(&self, active_peers: &[String]) -> Result<usize> {
5740 self.compact_internal(active_peers, true)
5741 .map(|outcome| outcome.pruned())
5742 }
5743
5744 pub fn maybe_compact_operation_log(
5757 &self,
5758 active_peers: &[String],
5759 ) -> Result<CompactionOutcome> {
5760 self.compact_internal(active_peers, false)
5761 }
5762
5763 pub fn compaction_policy(&self) -> CompactionPolicy {
5765 self.state().compaction_policy
5766 }
5767
5768 pub fn set_compaction_policy(&self, policy: CompactionPolicy) {
5770 self.state().compaction_policy = policy;
5771 }
5772
5773 fn compact_internal(&self, active_peers: &[String], force: bool) -> Result<CompactionOutcome> {
5774 let mut inner = self.state();
5775 refuse_if_faulted(&inner)?;
5778 let standalone = active_peers.is_empty() && !inner.change_log.has_peer_history();
5782 let removed = if standalone {
5783 inner.change_log.prune_standalone(STANDALONE_OPERATION_WINDOW)
5784 } else {
5785 inner.change_log.prune_acknowledged(active_peers)
5786 };
5787 if removed == 0 {
5788 return Ok(CompactionOutcome::Idle);
5789 }
5790 persist_sync_metadata(&inner)?;
5791 inner.pruned_since_rewrite += removed;
5792
5793 let log_bytes = fs::metadata(&inner.path).map(|m| m.len()).unwrap_or(0);
5799 let doubled = log_bytes >= inner.log_bytes_after_rewrite.saturating_mul(2);
5800 if !force
5801 && (inner.pruned_since_rewrite < inner.compaction_policy.rewrite_after_pruned
5802 || (standalone && !doubled))
5803 {
5804 return Ok(CompactionOutcome::Deferred {
5805 pruned: removed,
5806 pending_since_rewrite: inner.pruned_since_rewrite,
5807 });
5808 }
5809
5810 rewrite_log_locked(&mut inner)?;
5811 inner.bootstrapped = true;
5812 inner.pruned_since_rewrite = 0;
5813 inner.log_bytes_after_rewrite = fs::metadata(&inner.path).map(|m| m.len()).unwrap_or(0);
5814 Ok(CompactionOutcome::Rewritten { pruned: removed })
5815 }
5816
5817 pub fn register_peer(&self, advertisement: PeerAdvertisement) -> Result<()> {
5819 let inner = self.state();
5820 inner
5821 .peer_registry
5822 .register(advertisement)
5823 .map_err(|e| FlowError::CapabilityError(e))
5824 }
5825
5826 pub fn get_peer(&self, peer_id: &PeerId) -> Result<Option<PeerAdvertisement>> {
5828 let inner = self.state();
5829 inner
5830 .peer_registry
5831 .get(peer_id)
5832 .map_err(|e| FlowError::CapabilityError(e))
5833 }
5834
5835 pub fn peers_for_namespace(&self, namespace: &str) -> Result<Vec<PeerAdvertisement>> {
5837 let inner = self.state();
5838 inner
5839 .peer_registry
5840 .peers_for_namespace(namespace)
5841 .map_err(|e| FlowError::CapabilityError(e))
5842 }
5843
5844 pub fn peers_for_capability(&self, capability: &str) -> Result<Vec<PeerAdvertisement>> {
5846 let inner = self.state();
5847 inner
5848 .peer_registry
5849 .peers_for_capability(capability)
5850 .map_err(|e| FlowError::CapabilityError(e))
5851 }
5852
5853 pub fn all_peers(&self) -> Result<Vec<PeerAdvertisement>> {
5855 let inner = self.state();
5856 inner
5857 .peer_registry
5858 .all_peers()
5859 .map_err(|e| FlowError::CapabilityError(e))
5860 }
5861
5862 pub fn resolve_ref<T: DeserializeOwned>(&self, flow_ref: &FlowRef) -> Result<Option<T>> {
5864 let key = format!("{}:{}", flow_ref.namespace, flow_ref.object_id);
5866 self.get::<T>(&key)
5867 }
5868
5869 pub fn sync_state(&self) -> Result<SyncState> {
5871 let inner = self.state();
5872 Ok(inner.sync_state.clone())
5873 }
5874
5875 pub fn apply_remote_operation(&self, op: Operation) -> Result<()> {
5878 let mut inner = self.state();
5879
5880 if op.content_hash != op.compute_content_hash() {
5881 return Err(FlowError::CapabilityError(
5882 "remote operation content hash mismatch".to_string(),
5883 ));
5884 }
5885 if inner.change_log.has_operation(&op.instance_id, op.sequence) {
5886 return Ok(());
5887 }
5888 let expected = inner
5889 .change_log
5890 .versions()
5891 .get(&op.instance_id)
5892 .copied()
5893 .unwrap_or(0)
5894 + 1;
5895 if op.sequence != expected {
5896 return Err(FlowError::CapabilityError(format!(
5897 "remote operation sequence gap for {}: expected {}, received {}",
5898 op.instance_id, expected, op.sequence
5899 )));
5900 }
5901
5902 if let Some(remote_clock) = &op.vector_clock {
5904 inner.sync_state.merge_vector_clock(remote_clock);
5905 }
5906
5907 let local_op_for_key = inner
5909 .change_log
5910 .pending_operations()
5911 .into_iter()
5912 .find(|local_op| local_op.key == op.key && local_op.capability == op.capability);
5913
5914 let op_to_apply = if let Some(local_op) = local_op_for_key {
5916 let local_vc = local_op.vector_clock.clone().unwrap_or_default();
5919 let remote_vc = op.vector_clock.clone().unwrap_or_default();
5920
5921 let local_history = CausalHistory::new(
5922 StateVersion::from_clocks(local_vc, local_op.timestamp_ms),
5923 local_op.instance_id.clone(),
5924 local_op.sequence,
5925 );
5926
5927 let remote_history = CausalHistory::new(
5928 StateVersion::from_clocks(remote_vc, op.timestamp_ms),
5929 op.instance_id.clone(),
5930 op.sequence,
5931 );
5932
5933 let resolution = ConflictResolution::resolve_lww(&[local_history, remote_history]);
5935
5936 if resolution.winner_index == Some(1) {
5938 op.clone()
5939 } else {
5940 if let Some(mut history_row) = inner
5943 .rows
5944 .get(&op.capability)
5945 .and_then(|rows| rows.get(&op.key))
5946 .cloned()
5947 {
5948 history_row.operation = Some(op.clone());
5949 append_event_locked(&mut inner, &history_row)?;
5950 inner.authority_revision += 1;
5951 }
5952 inner.change_log.add_operation(op);
5953 inner.sync_state.increment_received(1);
5954 return Ok(());
5955 }
5956 } else {
5957 op.clone()
5958 };
5959
5960 let row = StoredRow {
5962 recorded_version: None,
5963 capability: op_to_apply.capability.clone(),
5964 key: op_to_apply.key.clone(),
5965 rust_type: op_to_apply.rust_type.clone(),
5966 value: op_to_apply.value.clone().unwrap_or(serde_json::Value::Null),
5967 unix_ms: op_to_apply.timestamp_ms,
5968 content_hash: Some(op_to_apply.content_hash.clone()),
5969 flow_ref: None,
5970 deleted: op_to_apply.op_type == OperationType::Delete,
5971 operation: Some(op.clone()),
5972 };
5973
5974 match op_to_apply.op_type {
5975 OperationType::Insert | OperationType::Update => {
5976 append_event_locked(&mut inner, &row)?;
5977 inner.authority_revision += 1;
5978 inner.put_row(row);
5979 }
5980 OperationType::Delete => {
5981 append_event_locked(&mut inner, &row)?;
5982 inner.authority_revision += 1;
5983 inner.remove_row(&op_to_apply.capability, &op_to_apply.key);
5984 }
5985 }
5986 if let Some(id) = op_to_apply
5987 .key
5988 .rsplit_once(':')
5989 .and_then(|(_, right)| right.parse::<u64>().ok())
5990 {
5991 let counter = inner
5992 .key_counters
5993 .entry(op_to_apply.capability.clone())
5994 .or_default();
5995 *counter = (*counter).max(id + 1);
5996 }
5997
5998 if let Some(provenance) = op.revision.clone() {
6008 if let Some(value) = op.value.clone() {
6009 let revision = reconstruct_revision(&provenance, &value)?;
6010 persist_revision_locked(&mut inner, revision)?;
6011 }
6012 }
6013
6014 inner.change_log.add_operation(op);
6015 inner.sync_state.increment_received(1);
6016 let event = ChangeEvent {
6017 capability: op_to_apply.capability,
6018 key: op_to_apply.key,
6019 rust_type: op_to_apply.rust_type,
6020 unix_ms: op_to_apply.timestamp_ms,
6021 };
6022 drop(inner);
6023 let _ = self.event_tx.send(event);
6024
6025 Ok(())
6026 }
6027
6028 pub fn get_vector_clock(&self) -> Result<Option<std::collections::HashMap<String, u64>>> {
6030 let inner = self.state();
6031 Ok(inner.sync_state.vector_clock.clone())
6032 }
6033}
6034
6035#[derive(Clone)]
6036pub struct Capability {
6037 db: FeltDb,
6038 name: String,
6039}
6040
6041impl Capability {
6042 pub fn insert<T: Serialize>(&self, value: T) -> Result<String> {
6043 let key = {
6044 let mut inner = self.db.state();
6045 let next = inner.key_counters.entry(self.name.clone()).or_default();
6046 let key = format!("{}:{}", self.name, *next);
6047 *next += 1;
6048 key
6049 };
6050 self.db
6051 .insert_internal(self.name.clone(), key.clone(), value, false)?;
6052 Ok(key)
6053 }
6054
6055 pub fn insert_with_key<T: Serialize>(&self, key: &str, value: T) -> Result<()> {
6056 let full_key = if key.contains(':') {
6057 key.to_string()
6058 } else {
6059 format!("{}:{key}", self.name)
6060 };
6061 self.db
6062 .insert_internal(self.name.clone(), full_key, value, false)
6063 }
6064
6065 pub fn update<T: Serialize>(&self, key: &str, value: T) -> Result<()> {
6066 let full_key = if key.contains(':') {
6067 key.to_string()
6068 } else {
6069 format!("{}:{key}", self.name)
6070 };
6071 self.db.update_internal(self.name.clone(), full_key, value)
6072 }
6073
6074 pub fn delete(&self, key: &str) -> Result<()> {
6075 let full_key = if key.contains(':') {
6076 key.to_string()
6077 } else {
6078 format!("{}:{key}", self.name)
6079 };
6080 self.db.delete_internal(self.name.clone(), full_key)
6081 }
6082
6083 pub fn get<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>> {
6084 let full_key = if key.contains(':') {
6085 key.to_string()
6086 } else {
6087 format!("{}:{key}", self.name)
6088 };
6089 self.db.get(&full_key)
6090 }
6091
6092 pub fn find<T, F>(&self, predicate: F) -> Result<Vec<T>>
6093 where
6094 T: DeserializeOwned,
6095 F: Fn(&T) -> bool,
6096 {
6097 let rust_type = type_name::<T>().to_string();
6098 let mut inner = self.db.state();
6099 *inner
6100 .query_count_by_type
6101 .entry(rust_type.clone())
6102 .or_default() += 1;
6103 if inner.query_count_by_type[&rust_type] >= 3 {
6104 inner.adaptive_indexes.insert(rust_type.clone());
6105 }
6106
6107 let mut out = Vec::new();
6108 if let Some(bucket) = inner.rows.get(&self.name) {
6109 for row in bucket.values() {
6110 if row.rust_type != rust_type {
6111 continue;
6112 }
6113 let parsed = serde_json::from_value::<T>(row.value.clone())?;
6114 if predicate(&parsed) {
6115 out.push(parsed);
6116 }
6117 }
6118 }
6119 Ok(out)
6120 }
6121
6122 pub async fn live_find<T, F>(&self, predicate: F) -> Result<LiveCapabilitySubscription<T>>
6144 where
6145 T: DeserializeOwned + Send + Sync + 'static,
6146 F: Fn(&T) -> bool + Send + Sync + 'static,
6147 {
6148 let rust_type = type_name::<T>().to_string();
6149 {
6150 let mut inner = self.db.state();
6151 *inner
6152 .query_count_by_type
6153 .entry(rust_type.clone())
6154 .or_default() += 1;
6155 if inner.query_count_by_type[&rust_type] >= 3 {
6156 inner.adaptive_indexes.insert(rust_type.clone());
6157 }
6158 }
6159 let rx = self.db.event_tx.subscribe();
6160 let mut matches = LiveMatches::new(rust_type, Some(self.name.clone()));
6161 matches.rescan(&self.db, &predicate);
6162 let subscription = LiveCapabilitySubscription {
6163 initial_results: matches.results(),
6164 rx,
6165 db: self.db.clone(),
6166 matches,
6167 predicate: Arc::new(Mutex::new(Some(Box::new(predicate)))),
6168 };
6169
6170 Ok(subscription)
6171 }
6172}
6173
6174type PredicateBox<T> = Arc<Mutex<Option<Box<dyn Fn(&T) -> bool + Send + Sync>>>>;
6176
6177struct LiveMatches {
6184 rust_type: String,
6185 capability: Option<String>,
6187 matches: BTreeMap<(String, String), Value>,
6188 stale: bool,
6189}
6190
6191impl LiveMatches {
6192 fn new(rust_type: String, capability: Option<String>) -> Self {
6193 LiveMatches { rust_type, capability, matches: BTreeMap::new(), stale: false }
6194 }
6195
6196 fn in_scope(&self, capability: &str) -> bool {
6197 self.capability.as_deref().is_none_or(|scope| scope == capability)
6198 }
6199
6200 fn admits<T: DeserializeOwned>(&self, row: &StoredRow, predicate: &dyn Fn(&T) -> bool) -> bool {
6201 !row.deleted
6202 && row.rust_type == self.rust_type
6203 && serde_json::from_value::<T>(row.value.clone())
6204 .is_ok_and(|parsed| predicate(&parsed))
6205 }
6206
6207 fn rescan<T: DeserializeOwned>(&mut self, db: &FeltDb, predicate: &dyn Fn(&T) -> bool) {
6209 let inner = db.state();
6210 self.matches.clear();
6211 let buckets: Box<dyn Iterator<Item = (&String, &BTreeMap<String, StoredRow>)>> =
6212 match &self.capability {
6213 Some(name) => Box::new(inner.rows.get_key_value(name).into_iter()),
6214 None => Box::new(inner.rows.iter()),
6215 };
6216 for (capability, bucket) in buckets {
6217 for (key, row) in bucket {
6218 if self.admits(row, predicate) {
6219 self.matches.insert((capability.clone(), key.clone()), row.value.clone());
6220 }
6221 }
6222 }
6223 self.stale = false;
6224 }
6225
6226 fn apply<T: DeserializeOwned>(
6228 &mut self,
6229 db: &FeltDb,
6230 event: &ChangeEvent,
6231 predicate: &dyn Fn(&T) -> bool,
6232 ) -> bool {
6233 if !self.in_scope(&event.capability) {
6234 return false;
6235 }
6236 let id = (event.capability.clone(), event.key.clone());
6237 let current = {
6238 let inner = db.state();
6239 inner
6240 .rows
6241 .get(&event.capability)
6242 .and_then(|bucket| bucket.get(&event.key))
6243 .filter(|row| self.admits(row, predicate))
6244 .map(|row| row.value.clone())
6245 };
6246 match current {
6247 Some(value) => self.matches.insert(id, value.clone()).as_ref() != Some(&value),
6248 None => self.matches.remove(&id).is_some(),
6249 }
6250 }
6251
6252 fn results<T: DeserializeOwned>(&self) -> Vec<T> {
6253 self.matches
6254 .values()
6255 .filter_map(|value| serde_json::from_value::<T>(value.clone()).ok())
6256 .collect()
6257 }
6258}
6259
6260async fn next_live_result<T: DeserializeOwned>(
6262 rx: &mut broadcast::Receiver<ChangeEvent>,
6263 db: &FeltDb,
6264 matches: &mut LiveMatches,
6265 predicate: &PredicateBox<T>,
6266) -> Option<Vec<T>> {
6267 loop {
6268 if matches.stale {
6269 let guard = predicate.lock().expect("lock poisoned");
6270 let Some(predicate) = guard.as_ref() else { return None };
6271 matches.rescan(db, predicate.as_ref());
6272 return Some(matches.results());
6273 }
6274 match rx.recv().await {
6275 Ok(event) => {
6276 let guard = predicate.lock().expect("lock poisoned");
6277 let Some(predicate) = guard.as_ref() else { return None };
6278 if matches.apply(db, &event, predicate.as_ref()) {
6279 return Some(matches.results());
6280 }
6281 }
6282 Err(broadcast::error::RecvError::Lagged(_)) => matches.stale = true,
6285 Err(broadcast::error::RecvError::Closed) => return None,
6286 }
6287 }
6288}
6289
6290pub struct LiveQuerySubscription<T> {
6295 pub initial_results: Vec<T>,
6297 rx: broadcast::Receiver<ChangeEvent>,
6298 db: FeltDb,
6299 matches: LiveMatches,
6300 predicate: PredicateBox<T>,
6301}
6302
6303impl<T> LiveQuerySubscription<T>
6304where
6305 T: DeserializeOwned + Send + Sync + 'static,
6306{
6307 pub fn set_predicate<F>(&mut self, predicate: F)
6308 where
6309 F: Fn(&T) -> bool + Send + Sync + 'static,
6310 {
6311 *self.predicate.lock().expect("lock poisoned") = Some(Box::new(predicate));
6312 self.matches.stale = true;
6314 }
6315
6316 pub async fn next(&mut self) -> Option<Vec<T>> {
6326 next_live_result(&mut self.rx, &self.db, &mut self.matches, &self.predicate).await
6327 }
6328
6329 pub fn initial_results(&self) -> &[T] {
6334 &self.initial_results
6335 }
6336}
6337
6338pub struct LiveCapabilitySubscription<T> {
6343 pub initial_results: Vec<T>,
6344 rx: broadcast::Receiver<ChangeEvent>,
6345 db: FeltDb,
6346 matches: LiveMatches,
6347 predicate: PredicateBox<T>,
6348}
6349
6350impl<T> LiveCapabilitySubscription<T>
6351where
6352 T: DeserializeOwned + Send + Sync + 'static,
6353{
6354 pub fn set_predicate<F>(&mut self, predicate: F)
6355 where
6356 F: Fn(&T) -> bool + Send + Sync + 'static,
6357 {
6358 *self.predicate.lock().expect("lock poisoned") = Some(Box::new(predicate));
6359 self.matches.stale = true;
6361 }
6362
6363 pub async fn next(&mut self) -> Option<Vec<T>> {
6368 next_live_result(&mut self.rx, &self.db, &mut self.matches, &self.predicate).await
6369 }
6370
6371 pub fn initial_results(&self) -> &[T] {
6372 &self.initial_results
6373 }
6374}
6375
6376pub fn inspect_durable_format(path: &Path) -> Result<FormatCompatibility> {
6395 if !path.exists() {
6396 return Ok(FormatCompatibility::Compatible(DurableFormat::Versioned(
6399 DURABLE_FORMAT_VERSION,
6400 )));
6401 }
6402
6403 #[derive(Deserialize)]
6407 struct FormatProbe {
6408 #[serde(default)]
6409 record_type: Option<Value>,
6410 #[serde(default)]
6411 format_version: Option<Value>,
6412 #[serde(default)]
6413 key: Option<Value>,
6414 }
6415 let mut declared: Option<u32> = None;
6416 let mut legacy_revision = false;
6417 visit_log(path, |line| {
6418 if line.text.trim().is_empty() {
6419 return Ok(());
6420 }
6421 let Ok(probe) = serde_json::from_str::<FormatProbe>(&line.text) else {
6422 return Ok(());
6423 };
6424 match probe.record_type.as_ref().and_then(Value::as_str) {
6425 Some(FORMAT_RECORD_TYPE) => {
6426 if let Some(version) = probe.format_version.as_ref().and_then(Value::as_u64) {
6427 declared = Some(version as u32);
6428 }
6429 }
6430 Some("feltdb.snapshot.v1" | "flowdb.snapshot.v1") => {
6431 match probe.format_version.as_ref().and_then(Value::as_u64) {
6432 Some(version) if version > 0 => declared = Some(version as u32),
6433 _ => {}
6434 }
6435 }
6436 _ => {
6437 let is_revision = probe
6442 .key
6443 .as_ref()
6444 .and_then(Value::as_str)
6445 .is_some_and(|key| key.starts_with("state:revision:"));
6446 if is_revision
6447 && serde_json::from_str::<Value>(&line.text).is_ok_and(|value| {
6448 value
6449 .get("value")
6450 .and_then(Value::as_object)
6451 .is_some_and(|revision| !revision.contains_key("resource"))
6452 })
6453 {
6454 legacy_revision = true;
6455 }
6456 }
6457 }
6458 Ok(())
6459 })?;
6460
6461 if let Some(version) = declared {
6462 return Ok(match version.cmp(&DURABLE_FORMAT_VERSION) {
6463 std::cmp::Ordering::Equal => {
6464 FormatCompatibility::Compatible(DurableFormat::Versioned(version))
6465 }
6466 std::cmp::Ordering::Greater => {
6467 FormatCompatibility::Incompatible(FormatIncompatibility {
6468 found: DurableFormat::Versioned(version),
6469 required: DURABLE_FORMAT_VERSION,
6470 reason: "the database was written by a newer FeltDB".to_string(),
6471 action: "Run a FeltDB build that understands this format, or restore \
6472 a backup taken in an older format."
6473 .to_string(),
6474 })
6475 }
6476 std::cmp::Ordering::Less => FormatCompatibility::Incompatible(FormatIncompatibility {
6477 found: DurableFormat::Versioned(version),
6478 required: DURABLE_FORMAT_VERSION,
6479 reason: "the database was written in an older durable format".to_string(),
6480 action: "Migrate the database with a FeltDB build that supports both \
6481 formats, or restore a backup."
6482 .to_string(),
6483 }),
6484 });
6485 }
6486
6487 if legacy_revision {
6488 return Ok(FormatCompatibility::Incompatible(FormatIncompatibility {
6489 found: DurableFormat::Unversioned,
6490 required: DURABLE_FORMAT_VERSION,
6491 reason: "it holds revision records written before revisions belonged to a \
6492 resource, and those records cannot be migrated without inventing \
6493 information they never contained: a revision recorded no resource, \
6494 so which history each one belongs to is not recoverable from the data"
6495 .to_string(),
6496 action: "Export the application state you need from a FeltDB build that \
6497 understands the old format, then load it into a new database. The \
6498 existing file has not been modified."
6499 .to_string(),
6500 }));
6501 }
6502
6503 Ok(FormatCompatibility::Compatible(DurableFormat::Unversioned))
6506}
6507
6508struct ScannedLine {
6510 text: String,
6511 line_number: usize,
6512 byte_offset: u64,
6513}
6514
6515#[derive(Clone)]
6517struct TornTail {
6518 byte_offset: u64,
6519 bytes: usize,
6520}
6521
6522fn visit_log(
6537 path: &Path,
6538 mut visit: impl FnMut(&ScannedLine) -> Result<()>,
6539) -> Result<Option<TornTail>> {
6540 let mut file = BufReader::with_capacity(1 << 20, OpenOptions::new().read(true).open(path)?);
6541 let mut buffer = Vec::new();
6542 let mut offset = 0u64;
6543 let mut line_number = 0usize;
6544
6545 loop {
6546 buffer.clear();
6547 let read = file.read_until(b'\n', &mut buffer)?;
6548 if read == 0 {
6549 return Ok(None);
6550 }
6551 line_number += 1;
6552 if !buffer.ends_with(b"\n") {
6553 let blank = buffer.iter().all(u8::is_ascii_whitespace);
6556 return Ok((!blank).then_some(TornTail {
6557 byte_offset: offset,
6558 bytes: read,
6559 }));
6560 }
6561 buffer.pop();
6562 let line = ScannedLine {
6563 text: match String::from_utf8(std::mem::take(&mut buffer)) {
6564 Ok(text) => text,
6565 Err(error) => String::from_utf8_lossy(error.as_bytes()).into_owned(),
6566 },
6567 line_number,
6568 byte_offset: offset,
6569 };
6570 visit(&line)?;
6571 buffer = line.text.into_bytes();
6572 offset += read as u64;
6573 }
6574}
6575
6576#[derive(Deserialize)]
6579struct RecordProbe {
6580 #[serde(default)]
6581 record_type: Option<Value>,
6582}
6583
6584fn unreadable(line: &ScannedLine) -> FlowError {
6587 if serde_json::from_str::<serde::de::IgnoredAny>(&line.text).is_ok() {
6588 corruption(line, "not a durable record this build understands")
6589 } else {
6590 corruption(line, "not valid JSON")
6591 }
6592}
6593
6594fn truncate_torn_tail(path: &Path, tail: &TornTail) -> Result<()> {
6600 let file = OpenOptions::new().write(true).open(path)?;
6601 file.set_len(tail.byte_offset)?;
6602 file.sync_all()?;
6603 Ok(())
6604}
6605
6606fn corruption(line: &ScannedLine, reason: &str) -> FlowError {
6607 let excerpt: String = line.text.chars().take(120).collect();
6608 FlowError::CorruptLogLine(Box::new(LogCorruption {
6609 line_number: line.line_number,
6610 byte_offset: line.byte_offset,
6611 reason: reason.to_string(),
6612 excerpt,
6613 }))
6614}
6615
6616fn replay_log(inner: &mut Inner) -> Result<()> {
6617 let path = inner.path.clone();
6618 let mut snapshot_log = false;
6619 let mut snapshot_authority = false;
6620 let mut snapshot_versions: HashMap<String, u64> = HashMap::new();
6621 let mut recoverable_revisions: HashMap<String, (operation::RevisionProvenance, Value)> =
6622 HashMap::new();
6623 let torn_tail = visit_log(&path, |line| {
6624 if line.text.trim().is_empty() {
6625 return Ok(());
6626 }
6627 let probe: RecordProbe =
6628 serde_json::from_str(&line.text).map_err(|_| unreadable(line))?;
6629 let record_type = probe.record_type.as_ref().and_then(Value::as_str);
6630 if matches!(record_type, Some("feltdb.snapshot.v1" | "flowdb.snapshot.v1")) {
6631 let header: LogHeader = serde_json::from_str(&line.text)
6632 .map_err(|error| corruption(line, &format!("malformed log header: {error}")))?;
6633 inner.sequence = inner.sequence.max(header.local_sequence);
6634 if header.authority_revision > 0 {
6638 inner.authority_revision = inner.authority_revision.max(header.authority_revision);
6639 snapshot_authority = true;
6640 }
6641 snapshot_versions = header.observed_versions;
6642 for attribution in header.transaction_history {
6643 let attribution = normalize_transaction_attribution(attribution);
6644 inner.authority_revision =
6645 inner.authority_revision.max(attribution.commit_revision);
6646 inner.transaction_revisions.insert(
6647 attribution.transaction_id.clone(),
6648 (attribution.base_revision, attribution.commit_revision),
6649 );
6650 inner
6651 .applied_transactions
6652 .insert(attribution.transaction_id.clone());
6653 remember_attribution(inner, attribution);
6654 }
6655 inner
6656 .transaction_payload_hashes
6657 .extend(header.transaction_payload_hashes);
6658 inner.bootstrapped = true;
6659 snapshot_log = true;
6660 return Ok(());
6661 }
6662 if record_type == Some(FORMAT_RECORD_TYPE) {
6663 return Ok(());
6665 }
6666 if record_type == Some("feltdb.transaction.v1") {
6667 let transaction: TransactionLogRecord =
6668 serde_json::from_str(&line.text).map_err(|error| {
6669 corruption(line, &format!("malformed transaction record: {error}"))
6670 })?;
6671 inner.authority_revision =
6674 inner
6675 .authority_revision
6676 .max(if transaction.commit_revision == 0 {
6677 inner.authority_revision + 1
6678 } else {
6679 transaction.commit_revision
6680 });
6681 let base_revision = if transaction.commit_revision == 0 {
6682 inner.authority_revision.saturating_sub(1)
6683 } else {
6684 transaction.base_revision
6685 };
6686 inner.transaction_revisions.insert(
6687 transaction.transaction_id.clone(),
6688 (base_revision, inner.authority_revision),
6689 );
6690 if let Some(hash) = transaction.payload_hash.clone() {
6691 inner
6692 .transaction_payload_hashes
6693 .insert(transaction.transaction_id.clone(), hash);
6694 }
6695 let attribution = TransactionAttribution::from_record(
6699 &transaction.transaction_id,
6700 transaction.audit.as_ref(),
6701 base_revision,
6702 inner.authority_revision,
6703 transaction.state_before,
6704 transaction.state_after,
6705 &transaction.rows,
6706 transaction
6707 .rows
6708 .first()
6709 .map(|row| row.unix_ms)
6710 .unwrap_or_default(),
6711 );
6712 remember_attribution(inner, attribution);
6713 if inner
6714 .applied_transactions
6715 .insert(transaction.transaction_id)
6716 {
6717 for row in transaction.rows {
6718 if let Some(operation) = row.operation.clone() {
6719 inner.sequence = inner.sequence.max(operation.sequence);
6720 if let Some(clock) = &operation.vector_clock {
6721 inner.sync_state.merge_vector_clock(clock);
6722 }
6723 inner.change_log.add_operation(operation);
6724 }
6725 if row.deleted {
6726 inner.remove_row(&row.capability, &row.key);
6727 } else {
6728 inner.put_row(row);
6729 }
6730 }
6731 }
6732 inner.sequence = inner.sequence.max(transaction.state_after);
6733 return Ok(());
6734 }
6735 let row: StoredRow = serde_json::from_str(&line.text)
6736 .map_err(|_| corruption(line, "not a durable record this build understands"))?;
6737 let derived_revision = row.capability == "state" && row.operation.is_none();
6738 if derived_revision {
6739 if let Some(id) = row.key.strip_prefix("state:revision:") {
6740 recoverable_revisions.remove(id);
6741 }
6742 }
6743 let counted_by_header = snapshot_authority
6748 && row.operation.as_ref().is_none_or(|operation| {
6749 snapshot_versions.get(&operation.instance_id).copied().unwrap_or(0)
6750 >= operation.sequence
6751 });
6752 if let Some(operation) = row.operation.clone() {
6753 if let (Some(provenance), Some(value)) = (&operation.revision, &operation.value) {
6760 recoverable_revisions.insert(provenance.id.clone(), (provenance.clone(), value.clone()));
6761 }
6762 inner.sequence = inner.sequence.max(
6763 (operation.instance_id == inner.instance_id)
6764 .then_some(operation.sequence)
6765 .unwrap_or(0),
6766 );
6767 if let Some(clock) = &operation.vector_clock {
6768 inner.sync_state.merge_vector_clock(clock);
6769 }
6770 inner.change_log.add_operation(operation);
6771 } else if !snapshot_log && !derived_revision {
6772 inner.sequence += 1;
6775 }
6776 if !derived_revision && !counted_by_header {
6777 let base_revision = inner.authority_revision;
6778 inner.authority_revision += 1;
6779 if !snapshot_log || snapshot_authority {
6783 remember_operation_attribution(inner, &row, base_revision, base_revision + 1);
6784 }
6785 }
6786 if let Some(id) = row
6787 .key
6788 .rsplit_once(':')
6789 .and_then(|(_, right)| right.parse::<u64>().ok())
6790 {
6791 let counter = inner
6792 .key_counters
6793 .entry(row.capability.clone())
6794 .or_default();
6795 *counter = (*counter).max(id + 1);
6796 }
6797 if row.deleted {
6798 inner.remove_row(&row.capability, &row.key);
6799 } else if row.operation.is_none() && row.recorded_version.is_none() {
6800 let mut row = row;
6804 if let (Some(previous), Some(hash)) = (
6805 inner.rows.get(&row.capability).and_then(|bucket| bucket.get(&row.key)),
6806 &row.content_hash,
6807 ) {
6808 if previous.operation.as_ref().is_some_and(|operation| &operation.content_hash == hash) {
6809 row.operation = previous.operation.clone();
6810 }
6811 }
6812 inner.put_row(row);
6813 } else {
6814 inner.put_row(row);
6815 }
6816 Ok(())
6817 })?;
6818 for (origin, sequence) in snapshot_versions {
6821 inner
6822 .change_log
6823 .observed_versions
6824 .entry(origin)
6825 .and_modify(|current| *current = (*current).max(sequence))
6826 .or_insert(sequence);
6827 }
6828
6829 for (provenance, value) in recoverable_revisions.into_values() {
6842 let key = format!("state:revision:{}", provenance.id);
6843 let present = inner
6844 .rows
6845 .get("state")
6846 .and_then(|bucket| bucket.get(&key))
6847 .is_some_and(|row| !row.deleted);
6848 if present {
6849 continue;
6850 }
6851
6852 let horizon = inner
6857 .rows
6858 .get("_retention")
6859 .and_then(|bucket| bucket.get(&format!("_retention:{}", provenance.resource)))
6860 .and_then(|row| serde_json::from_value::<RetentionRecord>(row.value.clone()).ok())
6861 .map(|state| state.horizon);
6862 if horizon.is_some_and(|horizon| provenance.sequence < horizon) {
6863 continue;
6864 }
6865 let Ok(revision) = reconstruct_revision(&provenance, &value) else {
6866 continue;
6867 };
6868 let Ok(encoded) = serde_json::to_value(&revision) else {
6869 continue;
6870 };
6871 inner.put_row(StoredRow {
6872 recorded_version: None,
6873 capability: "state".to_string(),
6874 key,
6875 rust_type: "feltdb::state_model::StateRevision".to_string(),
6876 value: encoded,
6877 unix_ms: now_ms(),
6878 content_hash: None,
6879 flow_ref: None,
6880 deleted: false,
6881 operation: None,
6882 });
6883 }
6884
6885 for (capability, bucket) in &inner.rows {
6887 let count = bucket.values().filter(|row| !row.deleted).count() as u64;
6888 inner
6889 .collection_cardinality
6890 .insert(capability.clone(), count);
6891 }
6892
6893 if let Some(tail) = torn_tail {
6897 truncate_torn_tail(&inner.path, &tail)?;
6898 inner.log_recovery = LogRecovery::RecoveredTornTail {
6899 byte_offset: tail.byte_offset,
6900 discarded_bytes: tail.bytes,
6901 };
6902 }
6903
6904 Ok(())
6905}
6906
6907fn write_json_line<T: Serialize>(file: &mut std::fs::File, value: &T) -> Result<()> {
6908 serde_json::to_writer(&mut *file, value)?;
6909 file.write_all(b"\n")?;
6910 Ok(())
6911}
6912
6913fn stored_row_for_operation(operation: Operation) -> StoredRow {
6914 StoredRow {
6915 recorded_version: None,
6916 capability: operation.capability.clone(),
6917 key: operation.key.clone(),
6918 rust_type: operation.rust_type.clone(),
6919 value: operation.value.clone().unwrap_or(Value::Null),
6920 unix_ms: operation.timestamp_ms,
6921 content_hash: Some(operation.content_hash.clone()),
6922 flow_ref: None,
6923 deleted: operation.op_type == OperationType::Delete,
6924 operation: Some(operation),
6925 }
6926}
6927
6928fn sync_metadata_path(path: &Path) -> PathBuf {
6929 path.with_extension("sync.json")
6930}
6931
6932fn load_sync_metadata(inner: &mut Inner) -> Result<()> {
6933 let path = sync_metadata_path(&inner.path);
6934 if !path.exists() {
6935 return Ok(());
6936 }
6937 let metadata: SyncMetadata = serde_json::from_slice(&fs::read(path)?)?;
6938 if let Some(history) = metadata.history {
6939 inner.history = history;
6940 }
6941 inner.sequence = inner.sequence.max(metadata.local_sequence);
6942 for (origin, sequence) in metadata.observed_versions {
6943 inner
6944 .change_log
6945 .observed_versions
6946 .entry(origin)
6947 .and_modify(|current| *current = (*current).max(sequence))
6948 .or_insert(sequence);
6949 }
6950 inner.change_log.peer_acknowledgements = metadata.peer_acknowledgements;
6951 Ok(())
6952}
6953
6954fn persist_sync_metadata(inner: &Inner) -> Result<()> {
6955 let path = sync_metadata_path(&inner.path);
6956 let temporary = path.with_extension("sync.tmp");
6957 let metadata = SyncMetadata {
6958 local_sequence: inner.sequence,
6959 observed_versions: inner.change_log.versions(),
6960 peer_acknowledgements: inner.change_log.peer_acknowledgements.clone(),
6961 history: Some(inner.history.clone()),
6962 };
6963 let mut file = OpenOptions::new()
6964 .create(true)
6965 .write(true)
6966 .truncate(true)
6967 .open(&temporary)?;
6968 file.write_all(&serde_json::to_vec_pretty(&metadata)?)?;
6969 file.flush()?;
6970 file.sync_all()?;
6971 fs::rename(temporary, &path)?;
6972 sync_parent_directory(&path)?;
6973 Ok(())
6974}
6975
6976fn append_event_locked(inner: &mut Inner, event: &StoredRow) -> Result<()> {
6978 append_rows_locked(inner, &[event])
6979}
6980
6981fn append_rows_locked(inner: &mut Inner, rows: &[&StoredRow]) -> Result<()> {
6987 let mut bytes = Vec::new();
6988 for row in rows {
6989 serde_json::to_writer(&mut bytes, row)?;
6990 bytes.push(b'\n');
6991 }
6992 let barrier = match inner.durability_mode {
6993 DurabilityMode::Flushed => Barrier::None,
6994 DurabilityMode::Synced => Barrier::Now,
6995 DurabilityMode::Grouped { every } => {
6996 inner.writes_since_barrier += 1;
6997 if inner.writes_since_barrier >= every.max(1) {
6998 Barrier::Group
6999 } else {
7000 Barrier::None
7001 }
7002 }
7003 };
7004 let outcome = append_bytes_locked(inner, &bytes, barrier);
7005 if barrier == Barrier::Group && outcome.is_ok() {
7006 inner.writes_since_barrier = 0;
7007 }
7008 outcome
7009}
7010
7011#[derive(Clone, Copy, PartialEq, Eq)]
7013enum Barrier {
7014 None,
7016 Now,
7018 Group,
7020}
7021
7022fn append_bytes_locked(inner: &mut Inner, bytes: &[u8], barrier: Barrier) -> Result<()> {
7038 let _span = workload_diagnostics::span(workload_diagnostics::Phase::Persistence);
7039 refuse_if_faulted(inner)?;
7040 let failed = |kind, error: &std::io::Error| {
7041 durability_failure(kind, Some(false), error.to_string())
7042 };
7043 let mut file = OpenOptions::new()
7044 .append(true)
7045 .open(&inner.path)
7046 .map_err(|error| failed(classify_write_error(&error), &error))?;
7047 let before = file
7048 .metadata()
7049 .map_err(|error| failed(DurabilityFailureKind::WriteFailed, &error))?
7050 .len();
7051
7052 let written = fault::write_all(&mut file, bytes).and_then(|()| file.flush());
7053 let outcome = match written {
7054 Err(error) => Err((classify_write_error(&error), error)),
7055 Ok(()) if barrier == Barrier::None => Ok(()),
7056 Ok(()) => fault::sync_data(&file).map_err(|error| {
7057 let kind = if barrier == Barrier::Group {
7061 DurabilityFailureKind::Indeterminate
7062 } else {
7063 DurabilityFailureKind::SyncFailed
7064 };
7065 (kind, error)
7066 }),
7067 };
7068 let Err((kind, error)) = outcome else {
7069 return Ok(());
7070 };
7071
7072 let rollback = fault::rollback(&file, before);
7073 let reason = match (rollback, kind) {
7074 (Err(rollback_error), _) => format!(
7075 "{error}; cutting the log back to byte {before} also failed: {rollback_error}"
7076 ),
7077 (Ok(()), DurabilityFailureKind::Indeterminate) => format!(
7078 "{error}; writes acknowledged earlier in this durability group may not be durable"
7079 ),
7080 (Ok(()), kind) => return Err(failed(kind, &error)),
7081 };
7082 inner.write_fault = Some(reason.clone());
7083 Err(durability_failure(
7084 DurabilityFailureKind::Indeterminate,
7085 None,
7086 reason,
7087 ))
7088}
7089
7090fn sibling(path: &Path, suffix: &str) -> PathBuf {
7092 let mut name = path.as_os_str().to_os_string();
7093 name.push(".");
7094 name.push(suffix);
7095 PathBuf::from(name)
7096}
7097
7098fn remove_restore_staging(staging: &Path) -> std::io::Result<()> {
7101 for path in [staging.to_path_buf(), sibling(staging, "lock")] {
7102 match fs::remove_file(&path) {
7103 Ok(()) => {}
7104 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
7105 Err(error) => return Err(error),
7106 }
7107 }
7108 Ok(())
7109}
7110
7111fn refuse_if_faulted(inner: &Inner) -> Result<()> {
7114 match &inner.write_fault {
7115 Some(reason) => Err(durability_failure(
7116 DurabilityFailureKind::RecoveryRequired,
7117 Some(false),
7118 format!("an earlier write left the durable log indeterminate ({reason}); reopen the database"),
7119 )),
7120 None => Ok(()),
7121 }
7122}
7123
7124fn rewrite_log_locked(inner: &mut Inner) -> Result<()> {
7146 let temporary = sibling(&inner.path, "compact.tmp");
7147 let written = (|| -> Result<()> {
7148 fault::stage(fault::Stage::CompactionWrite)?;
7149 let file = OpenOptions::new()
7150 .create(true)
7151 .write(true)
7152 .truncate(true)
7153 .open(&temporary)?;
7154 let mut out = std::io::BufWriter::with_capacity(1 << 20, file);
7155 write_json_line_to(
7159 &mut out,
7160 &LogHeaderRef {
7161 record_type: "feltdb.snapshot.v1",
7162 format_version: DURABLE_FORMAT_VERSION,
7163 local_sequence: inner.sequence,
7164 observed_versions: &inner.change_log.observed_versions,
7165 transaction_history: &inner.transaction_history,
7166 transaction_payload_hashes: &inner.transaction_payload_hashes,
7167 authority_revision: inner.authority_revision,
7168 },
7169 )?;
7170 fault::stage(fault::Stage::CompactionMidWrite)?;
7171 for operation in inner.change_log.pending.iter().cloned() {
7172 write_json_line_to(&mut out, &stored_row_for_operation(operation))?;
7173 }
7174 for bucket in inner.rows.values() {
7175 for row in bucket.values() {
7176 write_json_line_to(&mut out, &snapshot_row(row))?;
7177 }
7178 }
7179 let file = out.into_inner().map_err(|error| error.into_error())?;
7180 fault::stage(fault::Stage::CompactionSync)?;
7181 file.sync_all()?;
7182 fault::stage(fault::Stage::CompactionRename)?;
7183 fs::rename(&temporary, &inner.path)?;
7184 Ok(())
7185 })();
7186 if let Err(error) = written {
7187 let _ = fs::remove_file(&temporary);
7189 return Err(error);
7190 }
7191 if let Err(error) = fault::stage(fault::Stage::CompactionDirectorySync)
7192 .and_then(|()| sync_parent_directory(&inner.path))
7193 {
7194 inner.write_fault = Some(format!(
7195 "the compacted log replaced the old one, but its directory entry could not be made durable: {error}"
7196 ));
7197 return Err(FlowError::Io(error));
7198 }
7199 Ok(())
7200}
7201
7202fn write_json_line_to<W: Write, T: Serialize>(out: &mut W, value: &T) -> Result<()> {
7203 serde_json::to_writer(&mut *out, value)?;
7204 out.write_all(b"\n")?;
7205 Ok(())
7206}
7207
7208fn durability_failure(
7209 kind: DurabilityFailureKind,
7210 committed: Option<bool>,
7211 detail: String,
7212) -> FlowError {
7213 FlowError::Durability(Box::new(DurabilityFailure {
7214 kind,
7215 committed,
7216 detail,
7217 }))
7218}
7219
7220fn classify_write_error(error: &std::io::Error) -> DurabilityFailureKind {
7223 use std::io::ErrorKind;
7224 match error.kind() {
7225 ErrorKind::StorageFull | ErrorKind::FileTooLarge | ErrorKind::QuotaExceeded => {
7226 DurabilityFailureKind::StorageFull
7227 }
7228 _ => DurabilityFailureKind::WriteFailed,
7229 }
7230}
7231
7232fn sync_parent_directory(path: &Path) -> std::io::Result<()> {
7237 #[cfg(unix)]
7238 {
7239 let parent = path
7240 .parent()
7241 .filter(|parent| !parent.as_os_str().is_empty())
7242 .unwrap_or(Path::new("."));
7243 File::open(parent)?.sync_all()
7244 }
7245 #[cfg(not(unix))]
7246 {
7247 let _ = path;
7248 Ok(())
7249 }
7250}
7251
7252mod fault {
7256 use std::fs::File;
7257 use std::io::Write;
7258
7259 #[cfg(test)]
7260 pub(crate) use injection::*;
7261
7262 #[derive(Clone, Copy, PartialEq, Eq, Debug)]
7265 pub(crate) enum Stage {
7266 CompactionWrite,
7267 CompactionMidWrite,
7268 CompactionSync,
7269 CompactionRename,
7270 CompactionDirectorySync,
7271 }
7272
7273 pub(crate) fn stage(stage: Stage) -> std::io::Result<()> {
7274 #[cfg(test)]
7275 if let Some(error) = injection::take(injection::Fault::At(stage)) {
7276 return Err(error);
7277 }
7278 let _ = stage;
7279 Ok(())
7280 }
7281
7282 pub(crate) fn write_all(file: &mut File, bytes: &[u8]) -> std::io::Result<()> {
7283 #[cfg(test)]
7284 if let Some(result) = injection::on_write(file, bytes) {
7285 return result;
7286 }
7287 file.write_all(bytes)
7288 }
7289
7290 pub(crate) fn sync_data(file: &File) -> std::io::Result<()> {
7291 #[cfg(test)]
7292 if let Some(error) = injection::take(injection::Fault::Sync) {
7293 return Err(error);
7294 }
7295 file.sync_data()
7296 }
7297
7298 pub(crate) fn rollback(file: &File, length: u64) -> std::io::Result<()> {
7300 #[cfg(test)]
7301 if let Some(error) = injection::take(injection::Fault::Rollback) {
7302 return Err(error);
7303 }
7304 file.set_len(length)?;
7305 file.sync_all()
7306 }
7307
7308 #[cfg(test)]
7309 mod injection {
7310 use std::cell::RefCell;
7311 use std::fs::File;
7312 use std::io::Write;
7313
7314 #[derive(Clone, Copy, PartialEq, Eq, Debug)]
7316 pub(crate) enum Fault {
7317 ShortWrite { bytes: usize, errno: i32 },
7319 Sync,
7321 Rollback,
7323 At(super::Stage),
7325 }
7326
7327 thread_local! {
7328 static ARMED: RefCell<Vec<Fault>> = const { RefCell::new(Vec::new()) };
7329 }
7330
7331 pub(crate) fn arm(fault: Fault) {
7332 ARMED.with(|armed| armed.borrow_mut().push(fault));
7333 }
7334
7335 pub(crate) fn disarm() {
7336 ARMED.with(|armed| armed.borrow_mut().clear());
7337 }
7338
7339 pub(super) fn take(kind: Fault) -> Option<std::io::Error> {
7340 ARMED.with(|armed| {
7341 let mut armed = armed.borrow_mut();
7342 let index = armed.iter().position(|fault| *fault == kind)?;
7343 armed.remove(index);
7344 Some(std::io::Error::from_raw_os_error(5))
7345 })
7346 }
7347
7348 pub(super) fn on_write(file: &mut File, bytes: &[u8]) -> Option<std::io::Result<()>> {
7349 let fault = ARMED.with(|armed| {
7350 let mut armed = armed.borrow_mut();
7351 let index = armed
7352 .iter()
7353 .position(|fault| matches!(fault, Fault::ShortWrite { .. }))?;
7354 Some(armed.remove(index))
7355 })?;
7356 let Fault::ShortWrite { bytes: partial, errno } = fault else {
7357 unreachable!()
7358 };
7359 let partial = partial.min(bytes.len());
7360 Some(
7361 file.write_all(&bytes[..partial])
7362 .and_then(|()| file.flush())
7363 .and(Err(std::io::Error::from_raw_os_error(errno))),
7364 )
7365 }
7366 }
7367}
7368
7369fn committed_transaction_locked(
7377 inner: &Inner,
7378 transaction_id: &str,
7379 payload_hash: Option<&str>,
7380) -> Option<Result<AtomicCommit>> {
7381 if !inner.applied_transactions.contains(transaction_id) {
7382 return None;
7383 }
7384 if let Some(expected_hash) = payload_hash {
7385 if inner
7386 .transaction_payload_hashes
7387 .get(transaction_id)
7388 .map(String::as_str)
7389 != Some(expected_hash)
7390 {
7391 return Some(Err(FlowError::TransactionIdPayloadMismatch {
7392 transaction_id: transaction_id.to_string(),
7393 }));
7394 }
7395 }
7396 let (base_revision, commit_revision) = inner
7397 .transaction_revisions
7398 .get(transaction_id)
7399 .copied()
7400 .unwrap_or((inner.authority_revision, inner.authority_revision));
7401 let (state_before, state_after) = inner
7402 .transaction_history
7403 .iter()
7404 .find(|attribution| attribution.transaction_id == transaction_id)
7405 .map(|attribution| (attribution.state_before, attribution.state_after))
7406 .unwrap_or((inner.sequence, inner.sequence));
7407 Some(Ok(AtomicCommit {
7408 transaction_id: transaction_id.into(),
7409 base_revision,
7410 commit_revision,
7411 status: TransactionStatus::Committed,
7412 state_before,
7413 state_after,
7414 rows: vec![],
7415 duplicate: true,
7416 payload_binding: payload_hash.map(|_| PayloadBinding::Verified),
7417 }))
7418}
7419
7420fn append_transaction_locked(inner: &mut Inner, transaction: &TransactionLogRecord) -> Result<()> {
7423 let mut bytes = serde_json::to_vec(transaction)?;
7424 bytes.push(b'\n');
7425 append_bytes_locked(inner, &bytes, Barrier::Now)
7426}
7427
7428fn now_ms() -> u128 {
7429 SystemTime::now()
7430 .duration_since(UNIX_EPOCH)
7431 .expect("time moved backwards")
7432 .as_millis()
7433}
7434
7435#[cfg(test)]
7436mod tests {
7437 use super::*;
7438 use std::sync::atomic::{AtomicU64, Ordering};
7439 use std::time::{SystemTime, UNIX_EPOCH};
7440
7441 static TEMP_FILE_SEQUENCE: AtomicU64 = AtomicU64::new(0);
7442
7443 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
7444 struct User {
7445 email: String,
7446 age: u8,
7447 active: bool,
7448 }
7449
7450 fn temp_file() -> PathBuf {
7451 let nanos = SystemTime::now()
7452 .duration_since(UNIX_EPOCH)
7453 .expect("time")
7454 .as_nanos();
7455 let sequence = TEMP_FILE_SEQUENCE.fetch_add(1, Ordering::Relaxed);
7456 std::env::temp_dir().join(format!(
7457 "feltdb-test-{}-{nanos}-{sequence}.data",
7458 std::process::id()
7459 ))
7460 }
7461
7462 #[test]
7465 fn a_committed_transaction_records_who_committed_it() {
7466 let path = temp_file();
7467 let db = open(&path).expect("open");
7468 db.apply_atomic_transaction(
7469 "tx_agent_1",
7470 None,
7471 &[],
7472 &[AtomicMutation {
7473 capability: "billing:Customer".into(),
7474 key: "cus_1".into(),
7475 value: Some(serde_json::json!({"name": "Ada", "status": "active"})),
7476 }],
7477 Some(serde_json::json!({
7478 "subject": "agent_data_cleanup",
7479 "tenant": "acme",
7480 "application": "app_billing",
7481 "revision": "rev_7",
7482 })),
7483 )
7484 .expect("commit");
7485
7486 let recent = db.recent_transactions(10).expect("attributions");
7487 assert_eq!(recent.len(), 1);
7488 assert_eq!(recent[0].transaction_id, "tx_agent_1");
7489 assert_eq!(recent[0].subject.as_deref(), Some("agent_data_cleanup"));
7490 assert_eq!(recent[0].application_id.as_deref(), Some("app_billing"));
7491 assert_eq!(recent[0].operations, 1);
7492 assert_eq!(recent[0].keys, vec!["billing:Customer:cus_1".to_string()]);
7493 assert!(!serde_json::to_string(&recent[0]).unwrap().contains("Ada"));
7495
7496 let touching = db
7497 .transactions_touching("billing:Customer", "cus_1", 10)
7498 .expect("touching");
7499 assert_eq!(touching.len(), 1);
7500 assert_eq!(touching[0].transaction_id, "tx_agent_1");
7501 assert!(db
7502 .transactions_touching("billing:Customer", "cus_2", 10)
7503 .expect("touching")
7504 .is_empty());
7505
7506 assert_eq!(
7507 db.transaction_attribution("tx_agent_1")
7508 .expect("lookup")
7509 .map(|value| value.subject)
7510 .flatten()
7511 .as_deref(),
7512 Some("agent_data_cleanup")
7513 );
7514
7515 drop(db);
7518 let reopened = open(&path).expect("reopen");
7519 let after = reopened.recent_transactions(10).expect("attributions");
7520 assert_eq!(after.len(), 1);
7521 assert_eq!(after[0].subject.as_deref(), Some("agent_data_cleanup"));
7522 assert_eq!(after[0].keys, vec!["billing:Customer:cus_1".to_string()]);
7523 drop(reopened);
7524 let _ = fs::remove_file(path);
7525 }
7526
7527 #[test]
7529 fn transaction_attribution_is_bounded_and_newest_first() {
7530 let path = temp_file();
7531 let db = open(&path).expect("open");
7532 let commits = TRANSACTION_ATTRIBUTION_WINDOW + 5;
7533 for index in 0..commits {
7534 db.apply_atomic_transaction(
7535 &format!("tx_{index}"),
7536 None,
7537 &[],
7538 &[AtomicMutation {
7539 capability: "app:Note".into(),
7540 key: format!("note_{index}"),
7541 value: Some(serde_json::json!({"index": index})),
7542 }],
7543 Some(serde_json::json!({"subject": format!("actor_{index}")})),
7544 )
7545 .expect("commit");
7546 }
7547 let (held, capacity) = db.transaction_attribution_window().expect("window");
7548 assert_eq!(capacity, TRANSACTION_ATTRIBUTION_WINDOW);
7549 assert_eq!(held, TRANSACTION_ATTRIBUTION_WINDOW);
7550 let recent = db.recent_transactions(3).expect("attributions");
7551 assert_eq!(
7552 recent
7553 .iter()
7554 .map(|value| value.transaction_id.clone())
7555 .collect::<Vec<_>>(),
7556 vec![
7557 format!("tx_{}", commits - 1),
7558 format!("tx_{}", commits - 2),
7559 format!("tx_{}", commits - 3),
7560 ]
7561 );
7562 assert!(db
7565 .transaction_attribution("tx_0")
7566 .expect("lookup")
7567 .is_some());
7568 drop(db);
7569 let _ = fs::remove_file(path);
7570 }
7571
7572 #[test]
7573 fn canonical_transaction_history_survives_window_restart_and_compaction() {
7574 let path = temp_file();
7575 let db = open(&path).expect("open");
7576 let commits = TRANSACTION_ATTRIBUTION_WINDOW + 1;
7577 for index in 0..commits {
7578 db.apply_atomic_transaction(
7579 &format!("history_tx_{index}"),
7580 None,
7581 &[],
7582 &[AtomicMutation {
7583 capability: "history:Note".into(),
7584 key: format!("note_{index}"),
7585 value: Some(serde_json::json!({"index": index})),
7586 }],
7587 Some(serde_json::json!({"subject": "agent", "work_id": "work-1"})),
7588 )
7589 .expect("commit");
7590 }
7591 let complete = db.committed_transactions_after(0, usize::MAX).unwrap();
7592 assert_eq!(complete.len(), commits);
7593 assert_eq!(complete[0].transaction_id, "history_tx_0");
7594 assert_eq!(complete[0].operation_ids.len(), 1);
7595 assert_ne!(complete[0].operation_ids[0], complete[0].transaction_id);
7596 assert_eq!(complete[0].work_id.as_deref(), Some("work-1"));
7597 assert_eq!(
7598 db.transactions_touching("history:Note", "note_0", 1)
7599 .unwrap()[0]
7600 .transaction_id,
7601 "history_tx_0"
7602 );
7603 assert_eq!(
7604 db.transaction_attribution("history_tx_0")
7605 .unwrap()
7606 .unwrap()
7607 .audit
7608 .unwrap()["work_id"],
7609 "work-1"
7610 );
7611 let versions = db.operation_versions().unwrap();
7612 db.acknowledge_peer_versions("history-peer".into(), versions)
7613 .unwrap();
7614 db.compact_operation_log(&["history-peer".into()]).unwrap();
7615 drop(db);
7616
7617 let reopened = open(&path).expect("reopen");
7618 assert_eq!(
7619 reopened
7620 .committed_transactions_after(0, usize::MAX)
7621 .unwrap()
7622 .len(),
7623 commits
7624 );
7625 assert!(reopened
7626 .transaction_attribution("history_tx_0")
7627 .unwrap()
7628 .is_some());
7629 assert_eq!(
7630 reopened
7631 .transactions_touching("history:Note", "note_0", 1)
7632 .unwrap()[0]
7633 .transaction_id,
7634 "history_tx_0"
7635 );
7636 assert_eq!(
7637 reopened
7638 .transaction_attribution("history_tx_0")
7639 .unwrap()
7640 .unwrap()
7641 .operation_ids,
7642 complete[0].operation_ids
7643 );
7644 let _ = fs::remove_file(sync_metadata_path(&path));
7645 let _ = fs::remove_file(path);
7646 }
7647
7648 #[test]
7649 fn transaction_payload_binding_survives_restart_and_compaction() {
7650 let path = temp_file();
7651 let db = open(&path).expect("open");
7652 let mutation = AtomicMutation {
7653 capability: "managed:State".into(),
7654 key: "record-1".into(),
7655 value: Some(serde_json::json!({"value": 1})),
7656 };
7657 let first = db
7658 .apply_atomic_transaction_content_addressed(
7659 "managed-idempotency-1",
7660 Some("sha256:request-a"),
7661 None,
7662 &[],
7663 &[mutation.clone()],
7664 None,
7665 )
7666 .expect("initial commit");
7667 let duplicate = db
7668 .apply_atomic_transaction_content_addressed(
7669 "managed-idempotency-1",
7670 Some("sha256:request-a"),
7671 None,
7672 &[],
7673 &[mutation.clone()],
7674 None,
7675 )
7676 .expect("same semantic request");
7677 assert!(duplicate.duplicate);
7678 assert_eq!(duplicate.base_revision, first.base_revision);
7679 assert_eq!(duplicate.commit_revision, first.commit_revision);
7680 assert_eq!(duplicate.state_before, first.state_before);
7681 assert_eq!(duplicate.state_after, first.state_after);
7682
7683 let versions = db.operation_versions().unwrap();
7684 db.acknowledge_peer_versions("managed-peer".into(), versions)
7685 .unwrap();
7686 db.compact_operation_log(&["managed-peer".into()]).unwrap();
7687 drop(db);
7688
7689 let reopened = open(&path).expect("restart");
7690 let replay = reopened
7691 .apply_atomic_transaction_content_addressed(
7692 "managed-idempotency-1",
7693 Some("sha256:request-a"),
7694 None,
7695 &[],
7696 &[mutation.clone()],
7697 None,
7698 )
7699 .expect("retry after restart");
7700 assert!(replay.duplicate);
7701 assert_eq!(replay.commit_revision, first.commit_revision);
7702 let conflict = reopened
7703 .apply_atomic_transaction_content_addressed(
7704 "managed-idempotency-1",
7705 Some("sha256:request-b"),
7706 None,
7707 &[],
7708 &[mutation],
7709 None,
7710 )
7711 .unwrap_err();
7712 assert!(
7713 matches!(
7714 &conflict,
7715 FlowError::TransactionIdPayloadMismatch { transaction_id } if transaction_id == "managed-idempotency-1"
7716 ),
7717 "expected the typed idempotency conflict, got {conflict:?}"
7718 );
7719 let _ = fs::remove_file(sync_metadata_path(&path));
7720 let _ = fs::remove_file(path);
7721 }
7722
7723 #[test]
7724 fn single_record_commits_are_canonical_history_positions() {
7725 let path = temp_file();
7726 let db = open(&path).unwrap();
7727 db.insert("orders:one", serde_json::json!({"status":"paid"}))
7728 .unwrap();
7729 let history = db.committed_transactions_after(0, 10).unwrap();
7730 assert_eq!(history.len(), 1);
7731 assert!(history[0].transaction_id.starts_with("transaction:"));
7732 assert!(history[0].operation_ids[0].starts_with("operation:"));
7733 assert_ne!(history[0].transaction_id, history[0].operation_ids[0]);
7734 assert_eq!(history[0].keys, vec!["orders:orders:one"]);
7735 drop(db);
7736 let reopened = open(&path).unwrap();
7737 assert_eq!(
7738 reopened.committed_transactions_after(0, 10).unwrap()[0].transaction_id,
7739 history[0].transaction_id
7740 );
7741 assert_eq!(
7742 reopened.committed_transactions_after(0, 10).unwrap()[0].operation_ids,
7743 history[0].operation_ids
7744 );
7745 let _ = fs::remove_file(path);
7746 }
7747
7748 #[test]
7749 fn compacted_legacy_attribution_gets_stable_distinct_operation_identity() {
7750 let legacy = TransactionAttribution {
7751 transaction_id: "legacy-transaction".into(),
7752 operation_ids: Vec::new(),
7753 subject: Some("agent-1".into()),
7754 tenant_id: Some("tenant-1".into()),
7755 application_id: Some("app-1".into()),
7756 application_revision: None,
7757 base_revision: 4,
7758 commit_revision: 5,
7759 state_before: 4,
7760 state_after: 5,
7761 keys: vec!["orders:one".into()],
7762 operations: 2,
7763 principal_id: None,
7764 agent_id: Some("agent-1".into()),
7765 session_id: None,
7766 delegator_id: None,
7767 work_id: None,
7768 unix_ms: 1,
7769 audit: None,
7770 };
7771 let first = normalize_transaction_attribution(legacy.clone());
7772 let second = normalize_transaction_attribution(legacy);
7773 assert_eq!(first, second);
7774 assert_eq!(first.transaction_id, "legacy-transaction");
7775 assert_eq!(first.operation_ids.len(), 2);
7776 assert!(first
7777 .operation_ids
7778 .iter()
7779 .all(|operation| operation.starts_with("operation:legacy-transaction:")));
7780
7781 let single = TransactionAttribution {
7782 transaction_id: "operation:authority:9".into(),
7783 operation_ids: Vec::new(),
7784 operations: 1,
7785 ..first
7786 };
7787 let upgraded = normalize_transaction_attribution(single);
7788 assert_eq!(upgraded.transaction_id, "transaction:authority:9");
7789 assert_eq!(upgraded.operation_ids, vec!["operation:authority:9"]);
7790 }
7791
7792 #[test]
7794 fn a_bulk_transaction_truncates_its_key_list_and_keeps_the_count() {
7795 let path = temp_file();
7796 let db = open(&path).expect("open");
7797 let mutations = (0..TRANSACTION_ATTRIBUTION_KEYS + 20)
7798 .map(|index| AtomicMutation {
7799 capability: "app:Customer".into(),
7800 key: format!("cus_{index}"),
7801 value: Some(serde_json::json!({"index": index})),
7802 })
7803 .collect::<Vec<_>>();
7804 db.apply_atomic_transaction(
7805 "tx_bulk",
7806 None,
7807 &[],
7808 &mutations,
7809 Some(serde_json::json!({"subject": "migration_agent"})),
7810 )
7811 .expect("commit");
7812 let attribution = db
7813 .transaction_attribution("tx_bulk")
7814 .expect("lookup")
7815 .expect("held");
7816 assert_eq!(attribution.operations, mutations.len());
7817 assert_eq!(attribution.keys.len(), TRANSACTION_ATTRIBUTION_KEYS);
7818 drop(db);
7819 let _ = fs::remove_file(path);
7820 }
7821
7822 #[test]
7823 fn open_is_zero_init_and_creates_store() {
7824 let path = temp_file();
7825 let db = open(&path).expect("open");
7826 assert!(path.exists());
7827 drop(db);
7828 let _ = fs::remove_file(path);
7829 }
7830
7831 #[test]
7832 fn insert_get_and_reopen_are_durable() {
7833 let path = temp_file();
7834 let db = open(&path).expect("open");
7835 let user = User {
7836 email: "a@feltdb.dev".to_string(),
7837 age: 33,
7838 active: true,
7839 };
7840 db.insert("user:1", &user).expect("insert");
7841 let loaded: User = db.get("user:1").expect("get").expect("some");
7842 assert_eq!(loaded, user);
7843 drop(db);
7844
7845 let reopened = open(&path).expect("reopen");
7846 let loaded_after_restart: User = reopened.get("user:1").expect("get").expect("some");
7847 assert_eq!(loaded_after_restart, user);
7848 let _ = fs::remove_file(path);
7849 }
7850
7851 #[test]
7852 fn delete_and_reopen_preserves_tombstone() {
7853 let path = temp_file();
7854 let db = open(&path).expect("open");
7855 db.insert("users:deleted", serde_json::json!({"name": "gone"}))
7856 .expect("insert");
7857 db.delete("users:deleted").expect("delete");
7858 drop(db);
7859
7860 let reopened = open(&path).expect("reopen");
7861 assert_eq!(reopened.get_value("users:deleted").expect("get"), None);
7862 let _ = fs::remove_file(path);
7863 }
7864
7865 #[test]
7866 fn closure_query_and_capability_find_work() {
7867 let path = temp_file();
7868 let db = open(&path).expect("open");
7869 let users = db.capability("users");
7870 users
7871 .insert(User {
7872 email: "one@feltdb.dev".to_string(),
7873 age: 28,
7874 active: true,
7875 })
7876 .expect("insert");
7877 users
7878 .insert(User {
7879 email: "two@feltdb.dev".to_string(),
7880 age: 41,
7881 active: false,
7882 })
7883 .expect("insert");
7884
7885 let global = db.query::<User, _>(|u| u.age > 30).expect("query");
7886 assert_eq!(global.len(), 1);
7887
7888 let active = users.find::<User, _>(|u| u.active).expect("find");
7889 assert_eq!(active.len(), 1);
7890 assert_eq!(active[0].email, "one@feltdb.dev");
7891 let _ = fs::remove_file(path);
7892 }
7893
7894 #[test]
7895 fn adaptive_index_marks_frequent_query_types() {
7896 let path = temp_file();
7897 let db = open(&path).expect("open");
7898 db.insert(
7899 "user:1",
7900 User {
7901 email: "index@feltdb.dev".to_string(),
7902 age: 21,
7903 active: true,
7904 },
7905 )
7906 .expect("insert");
7907
7908 for _ in 0..3 {
7909 let _ = db.query::<User, _>(|u| u.active).expect("query");
7910 }
7911 assert!(db.is_auto_indexed::<User>());
7912 let _ = fs::remove_file(path);
7913 }
7914
7915 #[tokio::test]
7916 async fn live_query_returns_initial_results() {
7917 let path = temp_file();
7918 let db = open(&path).expect("open");
7919 let users = db.capability("users");
7920 users
7921 .insert(User {
7922 email: "live@feltdb.dev".to_string(),
7923 age: 28,
7924 active: true,
7925 })
7926 .expect("insert");
7927 users
7928 .insert(User {
7929 email: "inactive@feltdb.dev".to_string(),
7930 age: 35,
7931 active: false,
7932 })
7933 .expect("insert");
7934
7935 let subscription = db
7936 .live_query::<User, _>(|u| u.active)
7937 .await
7938 .expect("live_query");
7939 let initial = subscription.initial_results();
7940 assert_eq!(initial.len(), 1);
7941 assert_eq!(initial[0].email, "live@feltdb.dev");
7942 let _ = fs::remove_file(path);
7943 }
7944
7945 #[tokio::test]
7946 async fn live_query_streams_new_changes() {
7947 let path = temp_file();
7948 let db = open(&path).expect("open");
7949 let users = db.capability("users");
7950 users
7951 .insert(User {
7952 email: "initial@feltdb.dev".to_string(),
7953 age: 28,
7954 active: true,
7955 })
7956 .expect("insert");
7957
7958 let mut subscription = db
7959 .live_query::<User, _>(|u| u.active)
7960 .await
7961 .expect("live_query");
7962 let initial = subscription.initial_results();
7963 assert_eq!(initial.len(), 1);
7964
7965 let db_for_insert = db.clone();
7966 tokio::spawn(async move {
7967 tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
7968 let users_inner = db_for_insert.capability("users");
7969 users_inner
7970 .insert(User {
7971 email: "new@feltdb.dev".to_string(),
7972 age: 25,
7973 active: true,
7974 })
7975 .expect("insert");
7976 });
7977
7978 let updated = subscription.next().await;
7979 assert!(updated.is_some());
7980 let results = updated.unwrap();
7981 assert_eq!(results.len(), 2);
7982 let _ = fs::remove_file(path);
7983 }
7984
7985 #[tokio::test]
7986 async fn live_query_filters_by_predicate() {
7987 let path = temp_file();
7988 let db = open(&path).expect("open");
7989 let users = db.capability("users");
7990 users
7991 .insert(User {
7992 email: "active@feltdb.dev".to_string(),
7993 age: 28,
7994 active: true,
7995 })
7996 .expect("insert");
7997 users
7998 .insert(User {
7999 email: "inactive@feltdb.dev".to_string(),
8000 age: 35,
8001 active: false,
8002 })
8003 .expect("insert");
8004
8005 let subscription = db
8006 .live_query::<User, _>(|u| u.active)
8007 .await
8008 .expect("live_query");
8009 let initial = subscription.initial_results();
8010 assert_eq!(initial.len(), 1);
8011 assert!(initial[0].active);
8012 let _ = fs::remove_file(path);
8013 }
8014
8015 #[tokio::test]
8016 async fn change_events_are_emitted_on_insert() {
8017 let path = temp_file();
8018 let db = open(&path).expect("open");
8019
8020 let mut subscription = db
8021 .live_query::<User, _>(|_| true)
8022 .await
8023 .expect("live_query");
8024 let initial = subscription.initial_results();
8025 assert_eq!(initial.len(), 0);
8026
8027 let db_for_insert = db.clone();
8028 tokio::spawn(async move {
8029 tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
8030 let users_inner = db_for_insert.capability("users");
8031 users_inner
8032 .insert(User {
8033 email: "event@feltdb.dev".to_string(),
8034 age: 30,
8035 active: true,
8036 })
8037 .expect("insert");
8038 });
8039
8040 let update = subscription.next().await;
8041 assert!(update.is_some());
8042 let users_result = update.unwrap();
8043 assert_eq!(users_result.len(), 1);
8044 let _ = fs::remove_file(path);
8045 }
8046
8047 #[tokio::test]
8048 async fn capability_live_find_returns_initial_results() {
8049 let path = temp_file();
8050 let db = open(&path).expect("open");
8051 let users = db.capability("users");
8052 users
8053 .insert(User {
8054 email: "cap_live@feltdb.dev".to_string(),
8055 age: 28,
8056 active: true,
8057 })
8058 .expect("insert");
8059 users
8060 .insert(User {
8061 email: "cap_inactive@feltdb.dev".to_string(),
8062 age: 35,
8063 active: false,
8064 })
8065 .expect("insert");
8066
8067 let subscription = users
8068 .live_find::<User, _>(|u| u.active)
8069 .await
8070 .expect("live_find");
8071 let initial = subscription.initial_results();
8072 assert_eq!(initial.len(), 1);
8073 assert_eq!(initial[0].email, "cap_live@feltdb.dev");
8074 let _ = fs::remove_file(path);
8075 }
8076
8077 #[tokio::test]
8078 async fn capability_live_find_streams_scoped_changes() {
8079 let path = temp_file();
8080 let db = open(&path).expect("open");
8081 let users = db.capability("users");
8082 let _orders = db.capability("orders");
8083 users
8084 .insert(User {
8085 email: "user_scoped@feltdb.dev".to_string(),
8086 age: 28,
8087 active: true,
8088 })
8089 .expect("insert");
8090
8091 let mut subscription = users
8092 .live_find::<User, _>(|u| u.active)
8093 .await
8094 .expect("live_find");
8095 let initial = subscription.initial_results();
8096 assert_eq!(initial.len(), 1);
8097
8098 let db_for_insert = db.clone();
8099 tokio::spawn(async move {
8100 tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
8101 let users_inner = db_for_insert.capability("users");
8102 users_inner
8103 .insert(User {
8104 email: "new_user_scoped@feltdb.dev".to_string(),
8105 age: 25,
8106 active: true,
8107 })
8108 .expect("insert");
8109 });
8110
8111 let updated = subscription.next().await;
8112 assert!(updated.is_some());
8113 let results = updated.unwrap();
8114 assert_eq!(results.len(), 2);
8115 let _ = fs::remove_file(path);
8116 }
8117
8118 #[test]
8119 fn vector_clocks_increment_on_mutations() {
8120 let path = temp_file();
8121 let db = open(&path).expect("open");
8122
8123 let initial_clock = db.get_vector_clock().expect("get_vector_clock");
8125 assert!(initial_clock.is_none() || initial_clock.as_ref().unwrap().is_empty());
8126
8127 db.insert(
8129 "users:1",
8130 User {
8131 email: "user1@feltdb.dev".to_string(),
8132 age: 25,
8133 active: true,
8134 },
8135 )
8136 .expect("insert");
8137
8138 let pending = db
8140 .get_pending_for_peer("peer-1", 0)
8141 .expect("get_pending_for_peer");
8142 assert_eq!(pending.len(), 1);
8143 assert!(pending[0].vector_clock.is_some());
8144
8145 db.insert(
8147 "users:2",
8148 User {
8149 email: "user2@feltdb.dev".to_string(),
8150 age: 30,
8151 active: false,
8152 },
8153 )
8154 .expect("insert");
8155
8156 let pending = db
8158 .get_pending_for_peer("peer-1", 0)
8159 .expect("get_pending_for_peer");
8160 assert_eq!(pending.len(), 2);
8161 assert_eq!(pending[0].sequence, 1);
8162 assert_eq!(pending[1].sequence, 2);
8163
8164 let _ = fs::remove_file(path);
8165 }
8166
8167 #[test]
8168 fn update_creates_operation_with_vector_clock() {
8169 let path = temp_file();
8170 let db = open(&path).expect("open");
8171
8172 db.insert(
8174 "users:1",
8175 User {
8176 email: "user1@feltdb.dev".to_string(),
8177 age: 25,
8178 active: true,
8179 },
8180 )
8181 .expect("insert");
8182
8183 db.update(
8185 "users:1",
8186 User {
8187 email: "user1@feltdb.dev".to_string(),
8188 age: 26,
8189 active: true,
8190 },
8191 )
8192 .expect("update");
8193
8194 let pending = db
8196 .get_pending_for_peer("peer-1", 0)
8197 .expect("get_pending_for_peer");
8198 assert_eq!(pending.len(), 2);
8199 assert_eq!(pending[0].op_type, OperationType::Insert);
8200 assert_eq!(pending[1].op_type, OperationType::Update);
8201
8202 assert!(pending[0].vector_clock.is_some());
8204 assert!(pending[1].vector_clock.is_some());
8205
8206 let _ = fs::remove_file(path);
8207 }
8208
8209 #[test]
8210 fn delete_creates_operation_with_vector_clock() {
8211 let path = temp_file();
8212 let db = open(&path).expect("open");
8213
8214 db.insert(
8216 "users:1",
8217 User {
8218 email: "user1@feltdb.dev".to_string(),
8219 age: 25,
8220 active: true,
8221 },
8222 )
8223 .expect("insert");
8224
8225 db.delete("users:1").expect("delete");
8227
8228 let pending = db
8230 .get_pending_for_peer("peer-1", 0)
8231 .expect("get_pending_for_peer");
8232 assert_eq!(pending.len(), 2);
8233 assert_eq!(pending[0].op_type, OperationType::Insert);
8234 assert_eq!(pending[1].op_type, OperationType::Delete);
8235
8236 assert!(pending[0].vector_clock.is_some());
8238 assert!(pending[1].vector_clock.is_some());
8239
8240 let _ = fs::remove_file(path);
8241 }
8242
8243 #[test]
8244 fn apply_remote_operation_merges_vector_clocks() {
8245 let path = temp_file();
8246 let db = open(&path).expect("open");
8247
8248 let remote_op = Operation::insert(
8250 1,
8251 "remote-instance".to_string(),
8252 1,
8253 "users:100".to_string(),
8254 serde_json::json!({"email": "remote@feltdb.dev", "age": 30, "active": true}),
8255 "User".to_string(),
8256 "users".to_string(),
8257 )
8258 .with_vector_clock({
8259 let mut vc = std::collections::HashMap::new();
8260 vc.insert("remote-instance".to_string(), 1);
8261 vc
8262 });
8263
8264 db.apply_remote_operation(remote_op)
8266 .expect("apply_remote_operation");
8267
8268 let clock = db.get_vector_clock().expect("get_vector_clock");
8270 assert!(clock.is_some());
8271 let clock = clock.unwrap();
8272 assert_eq!(clock.get("remote-instance"), Some(&1));
8273
8274 let user: Option<User> = db.get("users:100").expect("get");
8276 assert!(user.is_some());
8277
8278 let _ = fs::remove_file(path);
8279 }
8280
8281 #[test]
8282 fn operation_history_survives_restart_and_relays_transitively() {
8283 let source_path = temp_file();
8284 let relay_path = temp_file();
8285 let destination_path = temp_file();
8286 let source = open(&source_path).expect("open source");
8287 source
8288 .insert("docs:1", serde_json::json!({"title": "one"}))
8289 .unwrap();
8290 source
8291 .insert("docs:2", serde_json::json!({"title": "two"}))
8292 .unwrap();
8293
8294 let relay = open(&relay_path).expect("open relay");
8295 for operation in source.operations_since(&HashMap::new()).unwrap() {
8296 relay.apply_remote_operation(operation).unwrap();
8297 }
8298 drop(relay);
8299
8300 let relay = open(&relay_path).expect("reopen relay");
8301 let relayed = relay.operations_since(&HashMap::new()).unwrap();
8302 assert_eq!(relayed.len(), 2);
8303 let destination = open(&destination_path).expect("open destination");
8304 for operation in relayed {
8305 destination.apply_remote_operation(operation).unwrap();
8306 }
8307 assert_eq!(
8308 destination.get_value("docs:2").unwrap(),
8309 Some(serde_json::json!({"title": "two"}))
8310 );
8311
8312 let _ = fs::remove_file(source_path);
8313 let _ = fs::remove_file(relay_path);
8314 let _ = fs::remove_file(destination_path);
8315 }
8316
8317 #[test]
8318 fn remote_operation_sequence_gaps_are_rejected() {
8319 let path = temp_file();
8320 let db = open(&path).unwrap();
8321 let operation = Operation::insert(
8322 2,
8323 "remote".to_string(),
8324 2,
8325 "docs:2".to_string(),
8326 serde_json::json!({"title": "two"}),
8327 "Value".to_string(),
8328 "docs".to_string(),
8329 );
8330 assert!(db.apply_remote_operation(operation).is_err());
8331 assert!(db.get_value("docs:2").unwrap().is_none());
8332 let _ = fs::remove_file(path);
8333 }
8334
8335 #[test]
8336 fn peer_version_acknowledgements_survive_restart() {
8337 let path = temp_file();
8338 let db = open(&path).unwrap();
8339 db.insert("docs:1", serde_json::json!({"title": "one"}))
8340 .unwrap();
8341 let versions = db.operation_versions().unwrap();
8342 db.acknowledge_peer_versions("peer-a".to_string(), versions.clone())
8343 .unwrap();
8344 drop(db);
8345
8346 let reopened = open(&path).unwrap();
8347 assert_eq!(
8348 reopened.acknowledged_peer_versions("peer-a").unwrap(),
8349 versions
8350 );
8351 assert_eq!(
8352 reopened
8353 .acknowledged_operation_count(&["peer-a".to_string()])
8354 .unwrap(),
8355 1
8356 );
8357 let _ = fs::remove_file(sync_metadata_path(&path));
8358 let _ = fs::remove_file(path);
8359 }
8360
8361 #[test]
8362 fn compacted_history_bootstraps_a_new_peer_without_sequence_gaps() {
8363 let source_path = temp_file();
8364 let destination_path = temp_file();
8365 let source = open(&source_path).unwrap();
8366 source
8367 .insert("docs:1", serde_json::json!({"title": "one"}))
8368 .unwrap();
8369 source
8370 .insert("docs:2", serde_json::json!({"title": "two"}))
8371 .unwrap();
8372 let versions = source.operation_versions().unwrap();
8373 source
8374 .acknowledge_peer_versions("peer-a".to_string(), versions.clone())
8375 .unwrap();
8376 assert_eq!(
8377 source
8378 .compact_operation_log(&["peer-a".to_string()])
8379 .unwrap(),
8380 2
8381 );
8382 assert!(source.operations_since(&HashMap::new()).unwrap().is_empty());
8383 drop(source);
8384
8385 let source = open(&source_path).unwrap();
8386 assert_eq!(source.operation_versions().unwrap(), versions);
8387 let snapshot = source.export_snapshot().unwrap();
8388 let destination = open(&destination_path).unwrap();
8389 destination.install_snapshot(snapshot).unwrap();
8390 assert_eq!(destination.operation_versions().unwrap(), versions);
8391 assert_eq!(
8392 destination.get_value("docs:2").unwrap(),
8393 Some(serde_json::json!({"title": "two"}))
8394 );
8395
8396 source
8397 .insert("docs:3", serde_json::json!({"title": "three"}))
8398 .unwrap();
8399 let next = source.operations_since(&versions).unwrap();
8400 assert_eq!(next.len(), 1);
8401 destination.apply_remote_operation(next[0].clone()).unwrap();
8402 assert!(destination.get_value("docs:3").unwrap().is_some());
8403
8404 let _ = fs::remove_file(sync_metadata_path(&source_path));
8405 let _ = fs::remove_file(sync_metadata_path(&destination_path));
8406 let _ = fs::remove_file(source_path);
8407 let _ = fs::remove_file(destination_path);
8408 }
8409
8410 #[test]
8411 fn maintained_cardinality_unfiltered_count_basic() {
8412 let path = temp_file();
8413 let db = open(&path).unwrap();
8414 let capability = "test_collection";
8415
8416 let mutations = vec![
8417 AtomicMutation {
8418 capability: capability.to_string(),
8419 key: "rec-1".to_string(),
8420 value: Some(serde_json::json!({"id": 1, "name": "Record 1"})),
8421 },
8422 AtomicMutation {
8423 capability: capability.to_string(),
8424 key: "rec-2".to_string(),
8425 value: Some(serde_json::json!({"id": 2, "name": "Record 2"})),
8426 },
8427 ];
8428
8429 db.apply_atomic_transaction("txn-1", None, &[], &mutations, None)
8430 .unwrap();
8431
8432 let cardinality = db.collection_cardinality(capability).unwrap();
8433 assert_eq!(cardinality, 2, "Cardinality should be 2 after 2 inserts");
8434
8435 let _ = fs::remove_file(path);
8436 }
8437
8438 #[test]
8439 fn maintained_cardinality_insert_increments() {
8440 let path = temp_file();
8441 let db = open(&path).unwrap();
8442 let capability = "test_collection";
8443
8444 for i in 1..=5 {
8445 let mutations = vec![AtomicMutation {
8446 capability: capability.to_string(),
8447 key: format!("rec-{}", i),
8448 value: Some(serde_json::json!({"id": i, "name": format!("Record {}", i)})),
8449 }];
8450
8451 db.apply_atomic_transaction(&format!("txn-{}", i), None, &[], &mutations, None)
8452 .unwrap();
8453 let cardinality = db.collection_cardinality(capability).unwrap();
8454 assert_eq!(
8455 cardinality, i as u64,
8456 "Cardinality should be {} after {} inserts",
8457 i, i
8458 );
8459 }
8460
8461 let _ = fs::remove_file(path);
8462 }
8463
8464 #[test]
8465 fn maintained_cardinality_delete_decrements() {
8466 let path = temp_file();
8467 let db = open(&path).unwrap();
8468 let capability = "test_collection";
8469
8470 let insert_mutations = vec![
8471 AtomicMutation {
8472 capability: capability.to_string(),
8473 key: "rec-1".to_string(),
8474 value: Some(serde_json::json!({"id": 1})),
8475 },
8476 AtomicMutation {
8477 capability: capability.to_string(),
8478 key: "rec-2".to_string(),
8479 value: Some(serde_json::json!({"id": 2})),
8480 },
8481 AtomicMutation {
8482 capability: capability.to_string(),
8483 key: "rec-3".to_string(),
8484 value: Some(serde_json::json!({"id": 3})),
8485 },
8486 ];
8487
8488 db.apply_atomic_transaction("txn-insert", None, &[], &insert_mutations, None)
8489 .unwrap();
8490 assert_eq!(db.collection_cardinality(capability).unwrap(), 3);
8491
8492 let delete_mutations = vec![AtomicMutation {
8493 capability: capability.to_string(),
8494 key: "rec-1".to_string(),
8495 value: None,
8496 }];
8497
8498 db.apply_atomic_transaction("txn-delete", None, &[], &delete_mutations, None)
8499 .unwrap();
8500 assert_eq!(
8501 db.collection_cardinality(capability).unwrap(),
8502 2,
8503 "Cardinality should be 2 after deleting 1 record"
8504 );
8505
8506 let _ = fs::remove_file(path);
8507 }
8508
8509 #[test]
8510 fn maintained_cardinality_update_does_not_change_count() {
8511 let path = temp_file();
8512 let db = open(&path).unwrap();
8513 let capability = "test_collection";
8514
8515 let insert_mutations = vec![AtomicMutation {
8516 capability: capability.to_string(),
8517 key: "rec-1".to_string(),
8518 value: Some(serde_json::json!({"id": 1, "status": "active"})),
8519 }];
8520
8521 db.apply_atomic_transaction("txn-insert", None, &[], &insert_mutations, None)
8522 .unwrap();
8523 assert_eq!(db.collection_cardinality(capability).unwrap(), 1);
8524
8525 let update_mutations = vec![AtomicMutation {
8526 capability: capability.to_string(),
8527 key: "rec-1".to_string(),
8528 value: Some(serde_json::json!({"id": 1, "status": "inactive"})),
8529 }];
8530
8531 db.apply_atomic_transaction("txn-update", None, &[], &update_mutations, None)
8532 .unwrap();
8533 assert_eq!(
8534 db.collection_cardinality(capability).unwrap(),
8535 1,
8536 "Cardinality should remain 1 after update"
8537 );
8538
8539 let _ = fs::remove_file(path);
8540 }
8541
8542 #[test]
8543 fn maintained_cardinality_survives_restart() {
8544 let path = temp_file();
8545
8546 {
8547 let db = open(&path).unwrap();
8548 let capability = "test_collection";
8549
8550 let mutations = vec![
8551 AtomicMutation {
8552 capability: capability.to_string(),
8553 key: "rec-1".to_string(),
8554 value: Some(serde_json::json!({"id": 1})),
8555 },
8556 AtomicMutation {
8557 capability: capability.to_string(),
8558 key: "rec-2".to_string(),
8559 value: Some(serde_json::json!({"id": 2})),
8560 },
8561 ];
8562
8563 db.apply_atomic_transaction("txn-1", None, &[], &mutations, None)
8564 .unwrap();
8565 assert_eq!(db.collection_cardinality(capability).unwrap(), 2);
8566 }
8567
8568 let reopened = open(&path).unwrap();
8569 let capability = "test_collection";
8570 let cardinality = reopened.collection_cardinality(capability).unwrap();
8571 assert_eq!(
8572 cardinality, 2,
8573 "Cardinality should survive restart and be recomputed from rows"
8574 );
8575
8576 let _ = fs::remove_file(path);
8577 }
8578
8579 #[test]
8580 fn maintained_cardinality_empty_collection() {
8581 let path = temp_file();
8582 let db = open(&path).unwrap();
8583 let capability = "empty_collection";
8584
8585 let cardinality = db.collection_cardinality(capability).unwrap();
8586 assert_eq!(cardinality, 0, "Empty collection should have cardinality 0");
8587
8588 let _ = fs::remove_file(path);
8589 }
8590
8591 #[test]
8592 fn maintained_cardinality_large_collection() {
8593 let path = temp_file();
8594 let db = open(&path).unwrap();
8595 let capability = "large_collection";
8596
8597 let mut mutations = Vec::new();
8598 for i in 1..=100 {
8599 mutations.push(AtomicMutation {
8600 capability: capability.to_string(),
8601 key: format!("rec-{}", i),
8602 value: Some(serde_json::json!({"id": i, "name": format!("Record {}", i)})),
8603 });
8604 }
8605
8606 db.apply_atomic_transaction("txn-large", None, &[], &mutations, None)
8607 .unwrap();
8608 let cardinality = db.collection_cardinality(capability).unwrap();
8609 assert_eq!(
8610 cardinality, 100,
8611 "Large collection with 100 records should have cardinality 100"
8612 );
8613
8614 let _ = fs::remove_file(path);
8615 }
8616
8617 #[test]
8618 fn maintained_cardinality_multiple_collections_isolated() {
8619 let path = temp_file();
8620 let db = open(&path).unwrap();
8621
8622 let col1 = "collection_1";
8623 let col2 = "collection_2";
8624
8625 let mutations1 = vec![
8626 AtomicMutation {
8627 capability: col1.to_string(),
8628 key: "rec-1".to_string(),
8629 value: Some(serde_json::json!({"id": 1})),
8630 },
8631 AtomicMutation {
8632 capability: col1.to_string(),
8633 key: "rec-2".to_string(),
8634 value: Some(serde_json::json!({"id": 2})),
8635 },
8636 ];
8637
8638 db.apply_atomic_transaction("txn-col1", None, &[], &mutations1, None)
8639 .unwrap();
8640
8641 let mutations2 = vec![
8642 AtomicMutation {
8643 capability: col2.to_string(),
8644 key: "rec-1".to_string(),
8645 value: Some(serde_json::json!({"id": 1})),
8646 },
8647 AtomicMutation {
8648 capability: col2.to_string(),
8649 key: "rec-2".to_string(),
8650 value: Some(serde_json::json!({"id": 2})),
8651 },
8652 AtomicMutation {
8653 capability: col2.to_string(),
8654 key: "rec-3".to_string(),
8655 value: Some(serde_json::json!({"id": 3})),
8656 },
8657 ];
8658
8659 db.apply_atomic_transaction("txn-col2", None, &[], &mutations2, None)
8660 .unwrap();
8661
8662 assert_eq!(db.collection_cardinality(col1).unwrap(), 2);
8663 assert_eq!(db.collection_cardinality(col2).unwrap(), 3);
8664
8665 let _ = fs::remove_file(path);
8666 }
8667}
8668
8669#[cfg(test)]
8677mod durability_tests {
8678 use super::fault::{arm, disarm, Fault};
8679 use super::*;
8680 use serde_json::json;
8681
8682 const ENOSPC: i32 = 28;
8683 const EFBIG: i32 = 27;
8684 const EIO: i32 = 5;
8685
8686 fn database() -> (PathBuf, FeltDb) {
8687 let directory = std::env::temp_dir().join(format!(
8688 "feltdb-durability-{}-{}",
8689 std::process::id(),
8690 SystemTime::now()
8691 .duration_since(UNIX_EPOCH)
8692 .unwrap()
8693 .as_nanos()
8694 ));
8695 fs::create_dir_all(&directory).unwrap();
8696 let path = directory.join("state.log");
8697 let db = FeltDb::open(&path).unwrap();
8698 db.set_durability_mode(DurabilityMode::Synced);
8699 (path, db)
8700 }
8701
8702 fn reopen(path: &Path, db: FeltDb) -> FeltDb {
8703 drop(db);
8704 let db = FeltDb::open(path).expect("the log reopens");
8705 db.set_durability_mode(DurabilityMode::Synced);
8706 db
8707 }
8708
8709 fn failure(result: Result<impl std::fmt::Debug>) -> DurabilityFailure {
8710 disarm();
8711 match result {
8712 Err(FlowError::Durability(failure)) => *failure,
8713 other => panic!("expected a durability failure, got {other:?}"),
8714 }
8715 }
8716
8717 fn insert_tx(db: &FeltDb, id: &str) -> Result<AtomicCommit> {
8718 db.apply_atomic_transaction(
8719 id,
8720 None,
8721 &[],
8722 &[AtomicMutation {
8723 capability: "notes".into(),
8724 key: format!("notes:{id}"),
8725 value: Some(json!({ "id": id })),
8726 }],
8727 None,
8728 )
8729 }
8730
8731 #[test]
8732 fn a_full_disk_fails_the_write_completely_and_the_next_write_lands() {
8733 let (path, db) = database();
8734 db.insert("notes:a", json!({ "id": "a" })).unwrap();
8735 let length = fs::metadata(&path).unwrap().len();
8736 let operations = db.operations_since(&HashMap::new()).unwrap().len();
8737 let sequence = db.sequence().unwrap();
8738
8739 arm(Fault::ShortWrite { bytes: 40, errno: ENOSPC });
8741 let refused = failure(db.insert("notes:b", json!({ "id": "b" })));
8742 assert_eq!(refused.kind, DurabilityFailureKind::StorageFull);
8743 assert_eq!(refused.code(), "STORAGE_FULL");
8744 assert_eq!(refused.committed, Some(false));
8745
8746 assert_eq!(fs::metadata(&path).unwrap().len(), length, "the partial bytes were removed");
8747 assert_eq!(db.get_value("notes:b").unwrap(), None, "not visible");
8748 assert_eq!(db.sequence().unwrap(), sequence, "the reserved sequence was given back");
8749 assert_eq!(
8750 db.operations_since(&HashMap::new()).unwrap().len(),
8751 operations,
8752 "replication was never offered the refused write"
8753 );
8754
8755 db.insert("notes:c", json!({ "id": "c" })).unwrap();
8757 let db = reopen(&path, db);
8758 assert_eq!(db.log_recovery(), LogRecovery::Clean);
8759 assert!(db.get_value("notes:a").unwrap().is_some());
8760 assert_eq!(db.get_value("notes:b").unwrap(), None);
8761 assert!(db.get_value("notes:c").unwrap().is_some());
8762 }
8763
8764 #[test]
8765 fn a_partial_transaction_record_never_splices_onto_the_next_commit() {
8766 let (path, db) = database();
8769 insert_tx(&db, "t1").unwrap();
8770 arm(Fault::ShortWrite { bytes: 64, errno: EFBIG });
8771 let refused = failure(insert_tx(&db, "t2"));
8772 assert_eq!(refused.kind, DurabilityFailureKind::StorageFull);
8773 assert_eq!(refused.committed, Some(false));
8774 insert_tx(&db, "t3").unwrap();
8775
8776 let db = reopen(&path, db);
8777 assert_eq!(db.log_recovery(), LogRecovery::Clean);
8778 assert!(db.get_value("notes:t1").unwrap().is_some());
8779 assert_eq!(db.get_value("notes:t2").unwrap(), None);
8780 assert!(db.get_value("notes:t3").unwrap().is_some());
8781 assert!(!insert_tx(&db, "t2").unwrap().duplicate);
8783 }
8784
8785 #[test]
8786 fn a_failed_barrier_is_rolled_back_so_a_reported_failure_never_resurrects() {
8787 let (path, db) = database();
8790 arm(Fault::Sync);
8791 let refused = failure(insert_tx(&db, "t1"));
8792 assert_eq!(refused.kind, DurabilityFailureKind::SyncFailed);
8793 assert_eq!(refused.code(), "STORAGE_SYNC_FAILED");
8794 assert_eq!(refused.committed, Some(false));
8795
8796 arm(Fault::Sync);
8797 let refused = failure(db.insert("notes:s", json!({ "id": "s" })));
8798 assert_eq!(refused.kind, DurabilityFailureKind::SyncFailed);
8799
8800 let db = reopen(&path, db);
8801 assert_eq!(db.get_value("notes:t1").unwrap(), None);
8802 assert_eq!(db.get_value("notes:s").unwrap(), None);
8803 assert_eq!(db.log_recovery(), LogRecovery::Clean);
8804 }
8805
8806 #[test]
8807 fn a_write_that_cannot_be_rolled_back_stops_writes_until_reopen_decides() {
8808 let (path, db) = database();
8809 db.insert("notes:a", json!({ "id": "a" })).unwrap();
8810
8811 arm(Fault::Sync);
8813 arm(Fault::Rollback);
8814 let unknown = failure(insert_tx(&db, "t1"));
8815 assert_eq!(unknown.kind, DurabilityFailureKind::Indeterminate);
8816 assert_eq!(unknown.code(), "DURABILITY_FAILURE");
8817 assert_eq!(unknown.committed, None, "the outcome is unknown, and said so");
8818
8819 let length = fs::metadata(&path).unwrap().len();
8821 for refused in [
8822 failure(db.insert("notes:b", json!({ "id": "b" }))),
8823 failure(db.delete("notes:a")),
8824 failure(insert_tx(&db, "t2")),
8825 failure(db.compact_operation_log(&[])),
8826 ] {
8827 assert_eq!(refused.kind, DurabilityFailureKind::RecoveryRequired);
8828 assert_eq!(refused.code(), "RECOVERY_REQUIRED");
8829 assert_eq!(refused.committed, Some(false));
8830 }
8831 assert_eq!(fs::metadata(&path).unwrap().len(), length, "nothing more was written");
8832 assert!(db.get_value("notes:a").unwrap().is_some(), "reads keep serving acknowledged writes");
8833 let health = db.health();
8834 assert_eq!(health.writes.label(), "recovery-required");
8835 assert!(!health.is_nominal());
8836
8837 let db = reopen(&path, db);
8840 assert_eq!(db.health().writes, WriteAvailability::Available);
8841 assert!(db.get_value("notes:t1").unwrap().is_some());
8842 assert!(insert_tx(&db, "t1").unwrap().duplicate, "a same-id retry reconciles");
8843 db.insert("notes:b", json!({ "id": "b" })).unwrap();
8844 }
8845
8846 #[test]
8847 fn an_io_error_on_write_is_typed_and_rolled_back() {
8848 let (path, db) = database();
8849 arm(Fault::ShortWrite { bytes: 0, errno: EIO });
8850 let refused = failure(db.update("notes:x", json!({ "id": "x" })));
8851 assert_eq!(refused.kind, DurabilityFailureKind::WriteFailed);
8852 assert_eq!(refused.code(), "STORAGE_WRITE_FAILED");
8853 assert_eq!(refused.committed, Some(false));
8854 let db = reopen(&path, db);
8855 assert_eq!(db.get_value("notes:x").unwrap(), None);
8856 }
8857
8858 #[test]
8859 fn a_single_record_write_and_its_revision_are_one_append() {
8860 let (path, db) = database();
8863 let before = fs::metadata(&path).unwrap().len();
8864 let record_bytes = {
8865 db.insert("notes:a", json!({ "id": "a" })).unwrap();
8866 fs::metadata(&path).unwrap().len() - before
8867 };
8868 let length = fs::metadata(&path).unwrap().len();
8870 arm(Fault::ShortWrite { bytes: (record_bytes as usize) - 20, errno: ENOSPC });
8871 let refused = failure(db.insert("notes:b", json!({ "id": "b" })));
8872 assert_eq!(refused.committed, Some(false));
8873 assert_eq!(fs::metadata(&path).unwrap().len(), length);
8874 assert_eq!(db.get_value("notes:b").unwrap(), None);
8875 let db = reopen(&path, db);
8876 assert_eq!(db.get_value("notes:b").unwrap(), None, "the failure is the truth after restart too");
8877 assert_eq!(db.log_recovery(), LogRecovery::Clean);
8878 }
8879
8880 #[test]
8881 fn a_failed_group_barrier_is_indeterminate_for_the_writes_it_covered() {
8882 let (path, db) = database();
8883 db.set_durability_mode(DurabilityMode::Grouped { every: 2 });
8884 db.insert("notes:a", json!({ "id": "a" })).unwrap(); arm(Fault::Sync);
8886 let unknown = failure(db.insert("notes:b", json!({ "id": "b" })));
8887 assert_eq!(unknown.kind, DurabilityFailureKind::Indeterminate);
8888 assert_eq!(db.health().writes.label(), "recovery-required");
8889 drop(reopen(&path, db));
8890 }
8891
8892 #[test]
8893 fn an_interrupted_restore_leaves_no_database_at_the_target() {
8894 let (path, db) = database();
8895 for id in ["a", "b", "c"] {
8896 db.insert(&format!("notes:{id}"), json!({ "id": id })).unwrap();
8897 }
8898 let directory = path.parent().unwrap().to_path_buf();
8899 let backup = directory.join("state.backup");
8900 db.write_backup(&backup).unwrap();
8901
8902 let target = directory.join("restored.log");
8904 let staging = sibling(&target, "restoring");
8905 fs::write(&staging, b"{\"record_type\":\"feltdb.format\"").unwrap();
8906 assert!(!target.exists());
8907
8908 let (restored, verification) = FeltDb::restore_backup(&backup, &target).unwrap();
8910 assert_eq!(verification.rows, 6, "three records and their three revisions");
8911 assert!(!staging.exists());
8912 assert!(restored.get_value("notes:c").unwrap().is_some());
8913 assert_eq!(
8914 restored.state_digest().unwrap(),
8915 db.state_digest().unwrap()
8916 );
8917 drop(restored);
8919 assert!(matches!(
8920 FeltDb::restore_backup(&backup, &target),
8921 Err(FlowError::BackupRejected(_))
8922 ));
8923 }
8924}
8925
8926
8927#[cfg(test)]
8934mod growth_tests {
8935 use super::fault::{arm, disarm, Fault, Stage};
8936 use super::*;
8937 use serde_json::json;
8938
8939 fn store(name: &str) -> (PathBuf, FeltDb) {
8940 let directory = std::env::temp_dir().join(format!(
8941 "feltdb-growth-{name}-{}-{}",
8942 std::process::id(),
8943 SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos()
8944 ));
8945 fs::create_dir_all(&directory).unwrap();
8946 let path = directory.join("state.log");
8947 let db = FeltDb::open(&path).unwrap();
8948 (path, db)
8949 }
8950
8951 fn reopen(path: &Path, db: FeltDb) -> FeltDb {
8952 drop(db);
8953 FeltDb::open(path).expect("the store reopens")
8954 }
8955
8956 fn tx(db: &FeltDb, id: &str, key: &str, value: Option<Value>) {
8957 db.apply_atomic_transaction(
8958 id,
8959 None,
8960 &[],
8961 &[AtomicMutation { capability: "notes".into(), key: key.into(), value }],
8962 None,
8963 )
8964 .unwrap();
8965 }
8966
8967 fn churn(db: &FeltDb, records: usize, rounds: usize) {
8970 for round in 0..rounds {
8971 for id in 0..records {
8972 let value = json!({ "id": id, "round": round, "body": "x".repeat(200) });
8975 if id % 2 == 0 {
8976 tx(db, &format!("t-{id}-{round}"), &format!("notes:{id}"), Some(value));
8977 } else if round == 0 {
8978 db.insert(&format!("notes:{id}"), value).unwrap();
8979 } else {
8980 db.update(&format!("notes:{id}"), value).unwrap();
8981 }
8982 }
8983 }
8984 for id in (0..records).step_by(4) {
8985 if id % 2 == 0 {
8986 tx(db, &format!("d-{id}"), &format!("notes:{id}"), None);
8987 } else {
8988 db.delete(&format!("notes:{id}")).unwrap();
8989 }
8990 }
8991 }
8992
8993 #[derive(Debug, PartialEq)]
8995 struct Meaning {
8996 state: String,
8997 revisions: u64,
8998 commits: usize,
8999 authority_revision: u64,
9000 sequence: u64,
9001 versions: HashMap<String, u64>,
9002 }
9003
9004 fn meaning(db: &FeltDb) -> Meaning {
9005 Meaning {
9006 state: db.state_digest().unwrap(),
9007 revisions: db
9008 .list_cardinalities()
9009 .unwrap()
9010 .into_iter()
9011 .find(|(name, _)| name == "state")
9012 .map(|(_, count)| count)
9013 .unwrap_or(0),
9014 commits: db.committed_transactions_after(0, usize::MAX).unwrap().len(),
9015 authority_revision: db.current_revision().unwrap(),
9016 sequence: db.sequence().unwrap(),
9017 versions: db.operation_versions().unwrap(),
9018 }
9019 }
9020
9021 fn retained(db: &FeltDb) -> usize {
9022 db.operations_since(&HashMap::new()).unwrap().len()
9023 }
9024
9025 fn cas_version(db: &FeltDb, key: &str) -> u64 {
9028 db.get_value(key).unwrap().and_then(|value| value.get("__version").and_then(Value::as_u64)).unwrap_or(1)
9029 }
9030
9031 #[test]
9032 fn a_history_is_created_once_and_survives_restart_and_compaction() {
9033 let (path, db) = store("history-stable");
9034 let history = db.version_history();
9035 assert_eq!(history.origin, HistoryOrigin::Genesis);
9036 assert_eq!(history.id.len(), 32);
9037 db.insert("notes:1", json!({"n": 1})).unwrap();
9038 db.compact_operation_log(&[]).unwrap();
9039 drop(db);
9040 let reopened = FeltDb::open(&path).unwrap();
9041 assert_eq!(reopened.version_history(), history, "restart and compaction keep the history");
9042 let token = reopened.version_token(7);
9043 assert_eq!(parse_version_token(&token), Some((history.id.as_str(), 7)));
9044 assert_eq!(reopened.accept_client_version("notes", "notes:1", None, Some(&token)).unwrap(), Some(7));
9045 assert_eq!(reopened.accept_client_version("notes", "notes:1", Some(7), None).unwrap(), Some(7), "a bare number is accepted in a genesis history");
9046 }
9047
9048 #[test]
9049 fn a_restore_starts_a_history_no_earlier_token_or_bare_number_can_fence() {
9050 let (_, db) = store("history-restore-source");
9051 db.insert("notes:1", json!({"n": 1})).unwrap();
9052 let old = db.version_token(cas_version(&db, "notes:1"));
9053 let snapshot = db.export_snapshot().unwrap();
9054 let (_, restored) = store("history-restore-target");
9055 restored.install_snapshot(snapshot).unwrap();
9056 assert_eq!(restored.version_history(), db.version_history(), "an install alone (replica bootstrap) shares the source's history");
9057 let fresh = restored.begin_restored_history().unwrap();
9058 assert_eq!(fresh.origin, HistoryOrigin::Restored);
9059 assert_ne!(fresh.id, db.version_history().id);
9060 let version = cas_version(&restored, "notes:1");
9062 let refused = restored.accept_client_version("notes", "notes:1", None, Some(&old)).unwrap_err();
9063 assert!(matches!(&refused, FlowError::PreconditionFailed(failure) if matches!(**failure, PreconditionFailure::History { expected: Some(_), .. })), "{refused}");
9064 let bare = restored.accept_client_version("notes", "notes:1", Some(version), None).unwrap_err();
9065 assert!(matches!(&bare, FlowError::PreconditionFailed(failure) if matches!(**failure, PreconditionFailure::History { expected: None, .. })), "{bare}");
9066 assert_eq!(restored.accept_client_version("notes", "notes:1", None, Some(&restored.version_token(version))).unwrap(), Some(version));
9067 let again = restored.begin_restored_history().unwrap();
9069 assert_ne!(again.id, fresh.id);
9070 assert!(restored.accept_client_version("notes", "notes:1", None, Some(&fresh.token(version))).is_err());
9071 }
9072
9073 #[test]
9074 fn malformed_or_disagreeing_tokens_are_refused_as_invalid_input() {
9075 let (_, db) = store("history-malformed");
9076 let good = db.version_token(3);
9077 for bad in ["", "3", "fv1:abc:3", "fv2:00000000000000000000000000000000:3", &format!("{good}:9"), "fv1:0000000000000000000000000000000G:3", "fv1:00000000000000000000000000000000:-1"] {
9078 let error = db.accept_client_version("c", "c:1", None, Some(bad)).unwrap_err();
9079 assert!(matches!(&error, FlowError::CapabilityError(message) if message.starts_with("INVALID_VERSION_TOKEN")), "{bad:?}: {error}");
9080 }
9081 assert!(db.accept_client_version("c", "c:1", Some(4), Some(&good)).is_err(), "a token and a number naming different versions");
9082 assert_eq!(db.accept_client_version("c", "c:1", Some(3), Some(&good)).unwrap(), Some(3));
9083 }
9084
9085 fn put_at(db: &FeltDb, capability: &str, key: &str, expected: Option<u64>, value: Value) -> Result<()> {
9088 db.apply_atomic_transaction(
9089 &format!("put-{capability}-{key}-{}", rand::random::<u64>()),
9090 None,
9091 &[AtomicPrecondition { capability: capability.into(), key: key.into(), expected_version: expected }],
9092 &[AtomicMutation { capability: capability.into(), key: key.into(), value: Some(value) }],
9093 None,
9094 )
9095 .map(|_| ())
9096 }
9097
9098 fn identities(db: &FeltDb) -> BTreeMap<(String, String), (Value, Option<u64>)> {
9100 db.state_rows()
9101 .unwrap()
9102 .into_iter()
9103 .filter(|row| !row.deleted)
9104 .map(|row| {
9105 let version = row.version();
9106 ((row.capability, row.key), (row.value, version))
9107 })
9108 .collect()
9109 }
9110
9111 fn mixed_store(label: &str) -> (PathBuf, FeltDb) {
9115 let (path, db) = store(label);
9116 db.insert("notes:1", json!({"kind": "key-value"})).unwrap();
9117 put_at(&db, "tenant:app:production:notes", "a-p", None, json!({"body": "A"})).unwrap();
9118 put_at(&db, "tenant:app:staging:notes", "a-p", None, json!({"body": "A staging"})).unwrap();
9119 put_at(&db, "orders", "1", None, json!({"n": 1})).unwrap();
9120 put_at(&db, "invoices", "1", None, json!({"n": 2})).unwrap();
9121 put_at(&db, "system_durable_work", "tenant/app/work_1", None, json!({"state": "READY"})).unwrap();
9122 put_at(&db, "links", "https://x.test/a:b", None, json!({"ok": true})).unwrap();
9123 put_at(&db, "notes", "notes:2", None, json!({"kind": "prefixed through a transaction"})).unwrap();
9124 put_at(&db, "orders", "2", None, json!({"n": 1})).unwrap();
9126 let version = identities(&db)[&("orders".to_string(), "2".to_string())].1;
9127 put_at(&db, "orders", "2", version, json!({"n": 2})).unwrap();
9128 put_at(&db, "orders", "gone", None, json!({})).unwrap();
9129 db.apply_atomic_transaction("delete-gone", None, &[], &[AtomicMutation { capability: "orders".into(), key: "gone".into(), value: None }], None).unwrap();
9130 (path, db)
9131 }
9132
9133 #[test]
9134 fn a_snapshot_preserves_every_record_identity_value_and_version_and_is_deterministic() {
9135 let (_, db) = mixed_store("snapshot-identity");
9136 let before = identities(&db);
9137 assert_eq!(before.len(), 10);
9138 assert!(!before.contains_key(&("orders".to_string(), "gone".to_string())));
9139 let snapshot = db.export_snapshot().unwrap();
9140 snapshot.validate().unwrap();
9141 let again = db.export_snapshot().unwrap();
9143 assert_eq!(serde_json::to_vec(&snapshot).unwrap(), serde_json::to_vec(&again).unwrap());
9144
9145 let (path, restored) = store("snapshot-identity-restored");
9146 restored.install_snapshot(snapshot.clone()).unwrap();
9147 assert_eq!(identities(&restored), before, "identities, values and versions are the source's");
9148 let second = restored.export_snapshot().unwrap();
9150 assert_eq!(second.content_hash, snapshot.content_hash);
9151 drop(restored);
9152 let reopened = FeltDb::open(&path).unwrap();
9153 assert_eq!(identities(&reopened), before, "and durable across a restart");
9154 assert_eq!(reopened.get_value("notes:1").unwrap(), Some(json!({"kind": "key-value"})));
9156 let order = before[&("orders".to_string(), "2".to_string())].1;
9158 assert!(put_at(&reopened, "orders", "2", None, json!({})).is_err(), "an existing record is not absent");
9159 put_at(&reopened, "orders", "2", order, json!({"n": 3})).unwrap();
9160 let (_, twice) = store("snapshot-identity-twice");
9162 twice.install_snapshot(snapshot.clone()).unwrap();
9163 assert_eq!(identities(&twice), before);
9164 assert!(twice.install_snapshot(snapshot).is_err(), "a store with state refuses a snapshot");
9165 }
9166
9167 #[test]
9168 fn snapshot_validation_refuses_ambiguous_or_forged_rows_and_install_agrees() {
9169 let (_, db) = mixed_store("snapshot-refusals");
9170 let good = db.export_snapshot().unwrap();
9171 let resealed = |mut snapshot: DatabaseSnapshot| {
9172 snapshot.content_hash = DatabaseSnapshot::compute_hash(&snapshot.rows, &snapshot.versions).unwrap();
9173 snapshot
9174 };
9175 let mut cases: Vec<(&str, DatabaseSnapshot)> = Vec::new();
9176 let mut duplicate = good.clone();
9177 duplicate.rows.push(duplicate.rows[0].clone());
9178 cases.push(("same identity", resealed(duplicate)));
9179 let mut tombstone = good.clone();
9180 tombstone.rows[0].deleted = true;
9181 cases.push(("tombstone", resealed(tombstone)));
9182 let mut operation = good.clone();
9183 operation.rows[0].operation = db.state_rows().unwrap().into_iter().find_map(|row| row.operation);
9184 cases.push(("operation", resealed(operation)));
9185 let mut empty = good.clone();
9186 empty.rows[0].key = String::new();
9187 cases.push(("no key", resealed(empty)));
9188 let mut tampered = good.clone();
9189 tampered.rows[0].value = json!({"tampered": true});
9190 cases.push(("tampered", tampered));
9191 for (label, snapshot) in cases {
9192 assert!(snapshot.validate().is_err(), "{label}: validation refuses");
9193 let (_, fresh) = store("snapshot-refusal");
9194 assert!(fresh.install_snapshot(snapshot).is_err(), "{label}: install refuses");
9195 assert!(identities(&fresh).is_empty(), "{label}: a refused install applies nothing");
9196 }
9197 }
9198
9199 #[test]
9200 fn a_snapshot_from_an_earlier_release_installs_and_its_records_fence_as_present() {
9201 let (_, db) = mixed_store("snapshot-legacy");
9202 let before = identities(&db);
9203 let mut legacy = db.export_snapshot().unwrap();
9206 for row in legacy.rows.iter_mut().skip(1) {
9207 row.recorded_version = None;
9208 }
9209 legacy.content_hash = DatabaseSnapshot::compute_hash(&legacy.rows, &legacy.versions).unwrap();
9210 let (path, restored) = store("snapshot-legacy-restored");
9211 restored.install_snapshot(legacy).unwrap();
9212 let after = identities(&restored);
9213 assert_eq!(after.keys().collect::<Vec<_>>(), before.keys().collect::<Vec<_>>(), "no identity changes");
9214 for (identity, (value, version)) in &after {
9215 assert_eq!(value, &before[identity].0, "{identity:?}: value");
9216 assert!(version == &before[identity].1 || version == &Some(0), "{identity:?}: its version, or 0 when none was recorded");
9217 }
9218 drop(restored);
9219 let restored = FeltDb::open(&path).unwrap();
9220 let unversioned = after.iter().find(|(_, (_, version))| *version == Some(0)).unwrap().0.clone();
9223 assert!(put_at(&restored, &unversioned.0, &unversioned.1, None, json!({})).is_err());
9224 put_at(&restored, &unversioned.0, &unversioned.1, Some(0), json!({"replaced": true})).unwrap();
9225 assert_ne!(identities(&restored)[&unversioned].1, Some(0));
9226 }
9227
9228 #[test]
9229 fn compaction_keeps_record_versions_across_a_restart_and_a_snapshot() {
9230 let (path, db) = mixed_store("compaction-versions");
9231 for index in 0..STANDALONE_OPERATION_WINDOW + 50 {
9233 put_at(&db, "padding", &index.to_string(), None, json!(index)).unwrap();
9234 }
9235 let before = identities(&db);
9236 db.compact_operation_log(&[]).unwrap();
9237 drop(db);
9238 let db = FeltDb::open(&path).unwrap();
9239 assert_eq!(identities(&db), before, "versions survive compaction and restart");
9240 let order = before[&("orders".to_string(), "2".to_string())].1;
9241 assert!(put_at(&db, "orders", "2", None, json!({})).is_err(), "an existing record is not absent");
9242 put_at(&db, "orders", "2", order, json!({"n": 4})).unwrap();
9243 let current = identities(&db);
9245 let (_, restored) = store("compaction-versions-restored");
9246 restored.install_snapshot(db.export_snapshot().unwrap()).unwrap();
9247 assert_eq!(identities(&restored), current);
9248 }
9249
9250 #[test]
9251 fn a_log_compacted_by_an_earlier_release_reads_its_records_as_present() {
9252 let (path, db) = mixed_store("compaction-legacy");
9253 for index in 0..STANDALONE_OPERATION_WINDOW + 50 {
9254 put_at(&db, "padding", &index.to_string(), None, json!(index)).unwrap();
9255 }
9256 db.compact_operation_log(&[]).unwrap();
9257 let before = identities(&db);
9258 drop(db);
9259 let text = fs::read_to_string(&path).unwrap();
9261 let legacy: Vec<String> = text
9262 .lines()
9263 .map(|line| {
9264 let mut value: Value = serde_json::from_str(line).unwrap();
9265 if let Some(object) = value.as_object_mut() {
9266 object.remove("recorded_version");
9267 }
9268 serde_json::to_string(&value).unwrap()
9269 })
9270 .collect();
9271 fs::write(&path, legacy.join("\n") + "\n").unwrap();
9272 let db = FeltDb::open(&path).unwrap();
9273 let after = identities(&db);
9274 assert_eq!(after.keys().collect::<Vec<_>>(), before.keys().collect::<Vec<_>>());
9275 let order = ("orders".to_string(), "2".to_string());
9276 assert_eq!(after[&order].0, before[&order].0);
9277 assert_eq!(after[&order].1, Some(0));
9280 assert!(put_at(&db, "orders", "2", None, json!({})).is_err(), "an existing record is not absent");
9281 put_at(&db, "orders", "2", Some(0), json!({"n": 5})).unwrap();
9282 let padding = ("padding".to_string(), (STANDALONE_OPERATION_WINDOW + 49).to_string());
9284 assert_eq!(after[&padding].1, before[&padding].1);
9285 }
9286
9287 #[test]
9288 fn a_standalone_store_keeps_a_window_of_operations_and_compaction_preserves_meaning() {
9289 let (path, db) = store("window");
9290 churn(&db, 3000, 3);
9291 let before = meaning(&db);
9292 let log_before = fs::metadata(&path).unwrap().len();
9293 assert!(retained(&db) > STANDALONE_OPERATION_WINDOW);
9294
9295 let outcome = db.maybe_compact_operation_log(&[]).unwrap();
9296 assert!(outcome.rewrote_log(), "{outcome:?}");
9297 assert_eq!(retained(&db), STANDALONE_OPERATION_WINDOW);
9298 assert_eq!(meaning(&db), before, "compaction changes no meaning");
9299 let log_after = fs::metadata(&path).unwrap().len();
9302 assert!(log_after < log_before, "{log_after} vs {log_before}");
9303 let current_rows: u64 = db.list_cardinalities().unwrap().iter().map(|(_, n)| n).sum();
9304 let lines = fs::read_to_string(&path).unwrap().lines().count() as u64;
9305 assert_eq!(
9306 lines,
9307 1 + STANDALONE_OPERATION_WINDOW as u64 + current_rows,
9308 "header + retained window + current rows: no superseded value, no tombstone"
9309 );
9310
9311 let db = reopen(&path, db);
9313 assert_eq!(meaning(&db), before);
9314 assert_eq!(retained(&db), STANDALONE_OPERATION_WINDOW, "retained operations survive reopen");
9315 assert!(db
9317 .apply_atomic_transaction("t-0-0", None, &[], &[AtomicMutation {
9318 capability: "notes".into(), key: "notes:0".into(), value: Some(json!({ "id": 0, "round": 0 })),
9319 }], None)
9320 .unwrap()
9321 .duplicate);
9322 db.insert("notes:new", json!({ "n": 1 })).unwrap();
9324 assert_eq!(db.current_revision().unwrap(), before.authority_revision + 1);
9325 let after_write = meaning(&db);
9326 let db = reopen(&path, db);
9327 assert_eq!(meaning(&db), after_write);
9328 }
9329
9330 #[test]
9331 fn interrupting_compaction_at_any_stage_before_the_switch_changes_nothing() {
9332 for stage in [Stage::CompactionWrite, Stage::CompactionMidWrite, Stage::CompactionSync, Stage::CompactionRename] {
9333 let (path, db) = store("stage");
9334 churn(&db, 2500, 2);
9335 let before = meaning(&db);
9336 let bytes = fs::read(&path).unwrap();
9337 arm(Fault::At(stage));
9338 let error = db.compact_operation_log(&[]);
9339 disarm();
9340 assert!(error.is_err(), "{stage:?}: the injected failure surfaced");
9341 assert_eq!(fs::read(&path).unwrap(), bytes, "{stage:?}: the log is untouched");
9342 assert!(!sibling(&path, "compact.tmp").exists(), "{stage:?}: no partial snapshot left");
9343 assert_eq!(meaning(&db), before, "{stage:?}");
9344 db.insert("notes:after", json!({ "after": true })).unwrap();
9346 db.compact_operation_log(&[]).unwrap();
9347 let expected = meaning(&db);
9348 let db = reopen(&path, db);
9349 assert_eq!(meaning(&db), expected, "{stage:?}");
9350 }
9351 }
9352
9353 #[test]
9354 fn a_switch_whose_directory_entry_is_not_durable_stops_writes_until_reopen() {
9355 let (path, db) = store("dirsync");
9356 churn(&db, 2500, 2);
9357 let before = meaning(&db);
9358 arm(Fault::At(Stage::CompactionDirectorySync));
9359 assert!(db.compact_operation_log(&[]).is_err());
9360 disarm();
9361 assert_eq!(db.health().writes.label(), "recovery-required");
9362 assert!(matches!(
9363 db.insert("notes:x", json!({})),
9364 Err(FlowError::Durability(failure)) if failure.kind == DurabilityFailureKind::RecoveryRequired
9365 ));
9366 assert_eq!(meaning(&db), before, "reads keep serving");
9367 let db = reopen(&path, db);
9368 assert_eq!(meaning(&db), before);
9369 db.insert("notes:x", json!({})).unwrap();
9370 }
9371
9372 #[test]
9373 fn a_process_killed_mid_compaction_leaves_the_old_log_authoritative() {
9374 for complete in [false, true] {
9377 let (path, db) = store("killed");
9378 churn(&db, 2500, 2);
9379 let before = meaning(&db);
9380 drop(db);
9381 let snapshot = sibling(&path, "compact.tmp");
9382 {
9383 let copy = path.with_file_name("copy.log");
9385 fs::copy(&path, ©).unwrap();
9386 let other = FeltDb::open(©).unwrap();
9387 other.compact_operation_log(&[]).unwrap();
9388 drop(other);
9389 let mut bytes = fs::read(©).unwrap();
9390 if !complete {
9391 bytes.truncate(bytes.len() / 2);
9392 }
9393 fs::write(&snapshot, bytes).unwrap();
9394 }
9395 let db = FeltDb::open(&path).unwrap();
9396 assert_eq!(meaning(&db), before, "complete={complete}");
9397 assert!(!snapshot.exists(), "the leftover snapshot was removed at open");
9398 }
9399 }
9400
9401 #[test]
9402 fn writes_during_repeated_compaction_are_all_preserved() {
9403 let (path, db) = store("concurrent");
9404 churn(&db, 2000, 2);
9405 let writer = {
9406 let db = db.clone();
9407 std::thread::spawn(move || {
9408 for i in 0..600 {
9409 db.insert(&format!("live:{i}"), json!({ "i": i })).unwrap();
9410 }
9411 })
9412 };
9413 let mut rewrites = 0;
9414 while !writer.is_finished() {
9415 if db.compact_operation_log(&[]).is_ok() {
9416 rewrites += 1;
9417 }
9418 std::thread::sleep(std::time::Duration::from_millis(20));
9420 }
9421 writer.join().unwrap();
9422 assert!(rewrites > 0);
9423 let expected = meaning(&db);
9424 let db = reopen(&path, db);
9425 assert_eq!(meaning(&db), expected);
9426 for i in 0..600 {
9427 assert!(db.get_value(&format!("live:{i}")).unwrap().is_some(), "live:{i}");
9428 }
9429 }
9430
9431 #[test]
9432 fn repeated_compaction_and_recovery_converge() {
9433 let (path, mut db) = store("repeat");
9434 for cycle in 0..5 {
9435 churn(&db, 1500, 2);
9436 db.compact_operation_log(&[]).unwrap();
9437 let expected = meaning(&db);
9438 db = reopen(&path, db);
9439 assert_eq!(meaning(&db), expected, "cycle {cycle}");
9440 db = reopen(&path, db);
9441 assert_eq!(meaning(&db), expected, "cycle {cycle}, second reopen");
9442 let log = fs::read(&path).unwrap();
9445 db.insert(&format!("cycle:{cycle}"), json!({})).unwrap();
9446 db.delete(&format!("cycle:{cycle}")).unwrap();
9447 db.compact_operation_log(&[]).unwrap();
9448 let again = fs::metadata(&path).unwrap().len() as i64;
9449 assert!((again - log.len() as i64).abs() < 4096, "cycle {cycle}: {again} vs {}", log.len());
9450 }
9451 }
9452
9453 #[test]
9454 fn backups_taken_before_and_after_compaction_restore_the_same_state() {
9455 let (path, db) = store("backup");
9456 churn(&db, 2500, 2);
9457 let directory = path.parent().unwrap().to_path_buf();
9458 db.write_backup(&directory.join("before.backup")).unwrap();
9459 let state = db.state_digest().unwrap();
9460 db.compact_operation_log(&[]).unwrap();
9461 db.write_backup(&directory.join("after.backup")).unwrap();
9462 for name in ["before", "after"] {
9463 let (restored, _) = FeltDb::restore_backup(
9464 &directory.join(format!("{name}.backup")),
9465 &directory.join(format!("{name}.log")),
9466 )
9467 .unwrap();
9468 assert_eq!(restored.state_digest().unwrap(), state, "{name}");
9469 }
9470 }
9471
9472 #[test]
9473 fn a_store_that_has_replicated_is_never_pruned_as_standalone() {
9474 let (path, db) = store("replicated");
9475 churn(&db, 2500, 2);
9476 let held = retained(&db);
9477 db.acknowledge_peer_versions("peer-1".into(), HashMap::new()).unwrap();
9478 assert!(matches!(db.compact_operation_log(&[]), Ok(0)));
9479 assert_eq!(retained(&db), held, "nothing pruned without the peer's acknowledgement");
9480 let db = reopen(&path, db);
9482 assert!(matches!(db.compact_operation_log(&[]), Ok(0)));
9483 }
9484
9485 #[test]
9486 fn a_log_rewritten_by_an_older_build_still_replays_as_before() {
9487 let (path, db) = store("legacy-header");
9490 churn(&db, 2500, 2);
9491 db.compact_operation_log(&[]).unwrap();
9492 drop(db);
9493 let text = fs::read_to_string(&path).unwrap();
9494 let (header, rest) = text.split_once('\n').unwrap();
9495 let mut header: Value = serde_json::from_str(header).unwrap();
9496 header.as_object_mut().unwrap().remove("authority_revision");
9497 fs::write(&path, format!("{header}\n{rest}")).unwrap();
9498 let first = FeltDb::open(&path).unwrap();
9499 let revision = first.current_revision().unwrap();
9500 let state = first.state_digest().unwrap();
9501 let first = reopen(&path, first);
9502 assert_eq!(first.current_revision().unwrap(), revision, "deterministic");
9503 assert_eq!(first.state_digest().unwrap(), state);
9504 }
9505}
9506
9507#[cfg(test)]
9514mod read_path_tests {
9515 use super::*;
9516 use serde_json::json;
9517 use std::cmp::Ordering;
9518
9519 fn store(name: &str) -> FeltDb {
9520 let directory = std::env::temp_dir().join(format!(
9521 "feltdb-read-{name}-{}-{}",
9522 std::process::id(),
9523 SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos()
9524 ));
9525 fs::create_dir_all(&directory).unwrap();
9526 FeltDb::open(directory.join("state.log")).unwrap()
9527 }
9528
9529 fn rank(row: &StoredRow) -> Option<i64> {
9531 row.value.get("rank").and_then(Value::as_i64)
9532 }
9533 fn compare(left: &StoredRow, right: &StoredRow) -> Ordering {
9534 rank(left).cmp(&rank(right)).then_with(|| left.key.cmp(&right.key))
9535 }
9536
9537 fn seed(db: &FeltDb, count: usize) {
9538 for n in 0..count {
9539 let value = if n % 7 == 0 {
9542 json!({ "n": n, "even": n % 2 == 0 })
9543 } else if n % 11 == 0 {
9544 json!({ "n": n, "rank": null, "even": n % 2 == 0 })
9545 } else {
9546 json!({ "n": n, "rank": (n * 37) % 50, "even": n % 2 == 0 })
9547 };
9548 db.insert(&format!("items:{n:05}"), value).unwrap();
9549 }
9550 }
9551
9552 fn walk(db: &FeltDb, ordered: bool, limit: usize, filter: impl Fn(&StoredRow) -> bool) -> Vec<String> {
9554 let mut keys = Vec::new();
9555 let mut after: Option<StoredRow> = None;
9556 loop {
9557 let key_order = (!ordered).then(|| after.as_ref().map(|row| row.key.as_str()));
9558 let page = db
9559 .select_page(
9560 "items",
9561 &[],
9562 key_order,
9563 limit,
9564 &filter,
9565 |row| after.as_ref().is_none_or(|last| {
9566 if ordered { compare(row, last) == Ordering::Greater } else { row.key > last.key }
9567 }),
9568 |left, right| if ordered { compare(left, right) } else { left.key.cmp(&right.key) },
9569 )
9570 .unwrap();
9571 assert!(page.rows.len() <= limit);
9572 keys.extend(page.rows.iter().map(|row| row.key.clone()));
9573 if !page.has_more {
9574 return keys;
9575 }
9576 assert_eq!(page.rows.len(), limit, "a page with more after it is full");
9577 after = page.rows.last().cloned();
9578 }
9579 }
9580
9581 fn reference(db: &FeltDb, ordered: bool, filter: impl Fn(&StoredRow) -> bool) -> Vec<String> {
9582 let mut rows = db.list_collection("items").unwrap();
9583 rows.retain(|row| filter(row));
9584 if ordered { rows.sort_by(compare) } else { rows.sort_by(|a, b| a.key.cmp(&b.key)) }
9585 rows.into_iter().map(|row| row.key).collect()
9586 }
9587
9588 #[test]
9589 fn every_page_size_walks_exactly_the_sorted_result() {
9590 let db = store("walk");
9591 seed(&db, 600);
9592 let even = |row: &StoredRow| row.value.get("even") == Some(&json!(true));
9593 for ordered in [true, false] {
9594 for limit in [1, 7, 50, 599, 600, 1000] {
9595 assert_eq!(walk(&db, ordered, limit, |_| true), reference(&db, ordered, |_| true), "ordered={ordered} limit={limit}");
9596 assert_eq!(walk(&db, ordered, limit, even), reference(&db, ordered, even), "filtered ordered={ordered} limit={limit}");
9597 }
9598 }
9599 }
9600
9601 #[test]
9602 fn a_key_order_page_examines_the_page_not_the_collection() {
9603 let db = store("seek");
9604 seed(&db, 5_000);
9605 let cursor = "items:04000".to_string();
9606 let page = db
9607 .select_page("items", &[], Some(Some(&cursor)), 25, |_| true, |row| row.key > cursor, |a, b| a.key.cmp(&b.key))
9608 .unwrap();
9609 assert_eq!(page.access, PageAccess::KeyRange);
9610 assert_eq!(page.rows.first().unwrap().key, "items:04001");
9611 assert!(page.examined <= 26, "examined {} rows for a page of 25", page.examined);
9612 assert!(page.has_more);
9613 }
9614
9615 #[test]
9616 fn an_ordered_page_holds_a_page_and_reports_the_scan_it_ran() {
9617 let db = store("scan");
9618 seed(&db, 3_000);
9619 let page = db.select_page("items", &[], None, 10, |_| true, |_| true, compare).unwrap();
9620 assert_eq!(page.access, PageAccess::Scan, "no index applies: the plan says scan");
9621 assert_eq!(page.examined, 3_000, "and it examined the collection, as reported");
9622 assert_eq!(page.rows.len(), 10);
9623 assert!(page.has_more);
9624 let expected: Vec<String> = reference(&db, true, |_| true).into_iter().take(10).collect();
9625 assert_eq!(page.rows.iter().map(|row| row.key.clone()).collect::<Vec<_>>(), expected);
9626 }
9627
9628 #[test]
9629 fn an_equality_index_bounds_the_candidates_and_is_reported() {
9630 let db = store("index");
9631 seed(&db, 2_000);
9632 db.create_equality_index("items", "even").unwrap();
9633 let yes = json!(true);
9634 let page = db
9635 .select_page("items", &[("even", &yes)], None, 5, |row| row.value.get("even") == Some(&yes), |_| true, compare)
9636 .unwrap();
9637 assert_eq!(page.access, PageAccess::EqualityIndex);
9638 assert_eq!(page.examined, 1_000, "the bucket, not the collection");
9639 }
9640
9641 #[test]
9642 fn pages_stay_stable_while_records_are_inserted_and_deleted_between_them() {
9643 for ordered in [true, false] {
9644 let db = store("concurrent");
9645 seed(&db, 300);
9646 let survivors: Vec<String> = reference(&db, ordered, |_| true);
9647 let mut seen: Vec<String> = Vec::new();
9648 let mut after: Option<StoredRow> = None;
9649 let mut round = 0;
9650 loop {
9651 let key_order = (!ordered).then(|| after.as_ref().map(|row| row.key.as_str()));
9652 let page = db
9653 .select_page(
9654 "items",
9655 &[],
9656 key_order,
9657 20,
9658 |_| true,
9659 |row| after.as_ref().is_none_or(|last| if ordered { compare(row, last) == Ordering::Greater } else { row.key > last.key }),
9660 |a, b| if ordered { compare(a, b) } else { a.key.cmp(&b.key) },
9661 )
9662 .unwrap();
9663 seen.extend(page.rows.iter().map(|row| row.key.clone()));
9664 if !page.has_more {
9665 break;
9666 }
9667 after = page.rows.last().cloned();
9668 round += 1;
9671 db.insert(&format!("items:9{round:04}"), json!({ "n": 9000 + round, "rank": round % 50 })).unwrap();
9672 db.delete(&seen[0]).unwrap();
9673 }
9674 let mut unique = seen.clone();
9675 unique.sort();
9676 unique.dedup();
9677 assert_eq!(unique.len(), seen.len(), "ordered={ordered}: no record appears twice");
9678 for key in &survivors {
9679 let deleted_before_reached = !seen.contains(key) && db.get_value(key).unwrap().is_none();
9680 assert!(seen.contains(key) || deleted_before_reached, "ordered={ordered}: {key} existed throughout and was skipped");
9681 }
9682 }
9683 }
9684
9685 #[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
9688 struct Item {
9689 n: u64,
9690 #[serde(default)]
9691 active: bool,
9692 }
9693
9694 fn item(n: u64, active: bool) -> Item {
9695 Item { n, active }
9696 }
9697
9698 #[tokio::test]
9699 async fn a_live_query_applies_its_predicate_to_every_change_and_sees_deletes() {
9700 let db = store("live");
9701 db.insert("items:1", item(1, true)).unwrap();
9702 db.insert("items:2", item(2, false)).unwrap();
9703 let mut live = db.live_query::<Item, _>(|item| item.active).await.unwrap();
9704 assert_eq!(live.initial_results().iter().map(|item| item.n).collect::<Vec<_>>(), vec![1]);
9705
9706 db.update("items:2", item(2, true)).unwrap();
9707 let next = live.next().await.unwrap();
9708 assert_eq!(next.iter().map(|item| item.n).collect::<Vec<_>>(), vec![1, 2], "the predicate decides membership on update");
9709
9710 db.update("items:1", item(1, false)).unwrap();
9711 let next = live.next().await.unwrap();
9712 assert_eq!(next.iter().map(|item| item.n).collect::<Vec<_>>(), vec![2], "an update can take a record out");
9713
9714 db.insert("items:3", item(3, false)).unwrap();
9715 db.delete("items:2").unwrap();
9716 let next = live.next().await.unwrap();
9717 assert!(next.is_empty(), "a non-matching insert does not wake it; a delete does: {next:?}");
9718 }
9719
9720 #[tokio::test]
9721 async fn a_write_that_changes_no_result_does_not_wake_a_subscription() {
9722 let db = store("quiet");
9723 db.insert("items:1", item(1, true)).unwrap();
9724 let mut live = db.capability("items").live_find::<Item, _>(|item| item.n < 10).await.unwrap();
9725 db.insert("other:1", item(1, true)).unwrap();
9726 db.insert("items:50", item(50, true)).unwrap();
9727 let woke = tokio::time::timeout(std::time::Duration::from_millis(200), live.next()).await;
9728 assert!(woke.is_err(), "neither another collection nor a non-match reevaluates it");
9729 db.insert("items:2", item(2, true)).unwrap();
9730 let next = tokio::time::timeout(std::time::Duration::from_secs(5), live.next())
9731 .await
9732 .expect("a matching insert wakes it")
9733 .unwrap();
9734 assert_eq!(next.len(), 2);
9735 }
9736
9737 #[tokio::test]
9738 async fn a_lagged_subscription_recovers_instead_of_ending() {
9739 let db = store("lag");
9740 let mut live = db.capability("items").live_find::<Item, _>(|_| true).await.unwrap();
9741 for n in 0..3_000u64 {
9743 db.insert(&format!("items:{n}"), item(n, true)).unwrap();
9744 }
9745 let recovered = tokio::time::timeout(std::time::Duration::from_secs(10), live.next())
9746 .await
9747 .expect("a lagged subscription answers")
9748 .expect("and continues");
9749 assert_eq!(recovered.len(), 3_000, "its result is recomputed in full");
9750 }
9751}