loonfs_core/
commit_engine.rs

1//! [`NamespaceCommitEngine`]: publishes batches of classified mutation
2//! candidates — one WAL segment, one head compare-and-swap, one result per
3//! candidate.
4
5use crate::checkpoint::MetadataTableCache;
6use crate::commit::CommitFingerprint;
7use crate::context::MutationContext;
8use crate::error::{CoreError, MetadataViewError, Result, WriterFence};
9use crate::metadata::MetadataState;
10use crate::namespace::basis::MetadataBasis;
11use crate::namespace::writer_epoch::acquire_writer_epoch;
12use crate::options::DeleteNamespaceOptions;
13use crate::path::write::{commit_fingerprint, CommitRequest, FilesystemOperation};
14use crate::protocol::{
15    load_publish_metadata_view, PublishTailOptions, PublishTailProjection, PublishTailWeight,
16};
17use crate::storage::content_admission::{ContentAdmission, ContentTokenError, PreparedContent};
18use crate::timing::{MonotonicTimer, StdMonotonicTimer};
19use loonfs_api::v0::CommitResponse as ApiCommitResponse;
20use loonfs_api::wire::control::{AcquiredWriter, HeadState};
21use loonfs_api::{
22    ChangeSeq, CommitId, ContentId, DeleteNamespaceResponse, ManifestId, NamespaceId,
23};
24use loonfs_objectstore::ObjectStore;
25use std::collections::HashSet;
26use std::sync::{Arc, Mutex};
27use thiserror::Error;
28
29/// One namespace mutation together with the result of preparing any content
30/// it references.
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct CommitCandidate {
33    request: CommitRequest,
34    content: ContentPreparation,
35}
36
37/// The result of preparing external content referenced by a mutation.
38#[derive(Debug, Clone, PartialEq, Eq)]
39pub enum ContentPreparation {
40    Ready(Vec<ContentAdmission>),
41    Rejected(ContentPreparationError),
42}
43
44/// A typed failure to prepare content referenced by a mutation candidate.
45#[derive(Debug, Clone, PartialEq, Eq, Error)]
46#[non_exhaustive]
47pub enum ContentPreparationError {
48    /// A supplied wire token was rejected before publication.
49    #[error("content token was rejected: {0}")]
50    ContentToken(#[from] ContentTokenError),
51    /// No prepared proof covers the referenced content.
52    #[error("content object `{content_id}` is not prepared for publication")]
53    ContentNotPrepared { content_id: ContentId },
54}
55
56impl CommitCandidate {
57    /// Wraps a mutation request with no attached content proofs.
58    pub fn new(request: CommitRequest) -> Self {
59        Self {
60            request,
61            content: ContentPreparation::Ready(Vec::new()),
62        }
63    }
64
65    /// Wraps a mutation request with opaque proofs for its prepared content.
66    pub fn prepared(request: CommitRequest, content: Vec<PreparedContent>) -> Self {
67        Self {
68            request,
69            content: ContentPreparation::Ready(
70                content
71                    .into_iter()
72                    .map(PreparedContent::into_admission)
73                    .collect(),
74            ),
75        }
76    }
77
78    /// Wraps a mutation request whose content preparation failed.
79    pub fn rejected(request: CommitRequest, error: ContentPreparationError) -> Self {
80        Self {
81            request,
82            content: ContentPreparation::Rejected(error),
83        }
84    }
85
86    pub(crate) fn request(&self) -> &CommitRequest {
87        &self.request
88    }
89
90    pub(crate) fn content_preparation(&self) -> &ContentPreparation {
91        &self.content
92    }
93
94    /// Returns the idempotency key carried by the mutation request.
95    pub fn commit_id(&self) -> &CommitId {
96        &self.request.commit_id
97    }
98
99    /// Computes semantic identity from the request alone, without applying
100    /// current operational request limits.
101    pub fn semantic_identity(&self, namespace_id: &NamespaceId) -> Result<CommitFingerprint> {
102        commit_fingerprint(namespace_id, &self.request)
103    }
104
105    pub(crate) fn validate_request_limits(&self) -> Result<()> {
106        // The ceilings apply to the request as a whole: a batch occupies the
107        // serialized publisher for as long as all of its operations take.
108        if self.request.operations.len() > crate::limits::MAX_COMMIT_OPERATIONS {
109            return Err(CoreError::InvalidCommitRequest(format!(
110                "mutation has {} operations; maximum is {}",
111                self.request.operations.len(),
112                crate::limits::MAX_COMMIT_OPERATIONS
113            )));
114        }
115        if let Some(message) = &self.request.message {
116            if message.len() > crate::limits::MAX_COMMIT_MESSAGE_BYTES {
117                return Err(CoreError::InvalidCommitRequest(format!(
118                    "mutation message is {} bytes; maximum is {}",
119                    message.len(),
120                    crate::limits::MAX_COMMIT_MESSAGE_BYTES
121                )));
122            }
123        }
124        let prepared_count = match &self.content {
125            ContentPreparation::Ready(content) => content.len(),
126            ContentPreparation::Rejected(_) => 0,
127        };
128        if prepared_count > crate::limits::MAX_COMMIT_CONTENT_TOKENS {
129            return Err(CoreError::InvalidCommitRequest(format!(
130                "mutation has {prepared_count} prepared content proofs; maximum is {}",
131                crate::limits::MAX_COMMIT_CONTENT_TOKENS
132            )));
133        }
134        let distinct_content_refs = self
135            .request
136            .operations
137            .iter()
138            .filter_map(|operation| match operation {
139                FilesystemOperation::PutFile { content_ref, .. } => Some(content_ref),
140                _ => None,
141            })
142            .collect::<HashSet<_>>()
143            .len();
144        if distinct_content_refs > crate::limits::MAX_COMMIT_EXTERNAL_CONTENT_REFS {
145            return Err(CoreError::InvalidCommitRequest(format!(
146                "mutation references {distinct_content_refs} distinct external content refs; maximum is {}",
147                crate::limits::MAX_COMMIT_EXTERNAL_CONTENT_REFS
148            )));
149        }
150        Ok(())
151    }
152}
153
154/// The WAL-tail maintenance policy: one authority for "when do we
155/// checkpoint?" and "when do we stop accepting writes?", so the two
156/// thresholds cannot drift apart.
157///
158/// Reads never gate on tail length; the rejection only asks writers to wait
159/// for the maintenance a deployment failed to run (format spec,
160/// "Maintenance operations").
161#[derive(Debug, Clone, Copy, PartialEq, Eq)]
162pub struct WalTailPolicy {
163    /// Visible WAL-tail length, in segments, at which a maintenance step
164    /// publishes a checkpoint. The step fires at or past this length.
165    pub checkpoint_at_segments: u64,
166    /// Visible WAL-tail length past which (strictly greater than) every
167    /// publish surface rejects with `maintenance_required`.
168    pub reject_writes_at_segments: u64,
169}
170
171impl WalTailPolicy {
172    /// The workspace policy: checkpoint at 32 segments, reject past 128.
173    pub const DEFAULT: Self = Self {
174        checkpoint_at_segments: 32,
175        reject_writes_at_segments: 128,
176    };
177}
178
179impl Default for WalTailPolicy {
180    fn default() -> Self {
181        Self::DEFAULT
182    }
183}
184
185// The ordering invariant `0 < checkpoint < reject` holds by construction:
186// a step must be able to relieve backpressure before writes stop.
187const _: () = assert!(
188    0 < WalTailPolicy::DEFAULT.checkpoint_at_segments
189        && WalTailPolicy::DEFAULT.checkpoint_at_segments
190            < WalTailPolicy::DEFAULT.reject_writes_at_segments,
191);
192
193#[derive(Debug, Clone)]
194pub struct NamespaceCommitEnginePublishResult {
195    pub results: Vec<Result<ApiCommitResponse>>,
196    /// WAL tail length observed by this publish, for opportunistic
197    /// maintenance scheduling. Zero when no projection was loaded.
198    pub wal_tail_segments: u64,
199    /// The read state this publish produced, present when the head CAS
200    /// landed unambiguously: callers can seed read caches with it instead
201    /// of invalidating them and rebuilding from the store.
202    pub resulting_read_state: Option<ResultingReadState>,
203}
204
205/// A read anchor plus the projected WAL tail as of one landed publish.
206#[derive(Debug, Clone)]
207pub struct ResultingReadState {
208    pub head: HeadState,
209    pub head_etag: String,
210    /// Basis the publish replayed over; the landed head still resolves to
211    /// it, so a seeded anchor pins the same pair the next read would.
212    pub basis: MetadataBasis,
213    pub manifest_id: ManifestId,
214    pub manifest_head_seq: ChangeSeq,
215    pub tail_rows: Arc<MetadataState>,
216}
217
218/// Writer-session state for one namespace: the epoch this session acquired
219/// and its terminal fencing record.
220///
221/// This is authoritative session state, not a cache of durable state —
222/// nothing in the store can rebuild "this session was fenced". Runtimes keep
223/// one shared instance per namespace in a registry that outlives every
224/// rebuildable cache (invalidation, LRU eviction, cache-disabled
225/// configurations) and hand it to each engine they build for the namespace.
226/// An engine built without one gets private state, which keeps the
227/// documented one-shot semantics: each one-shot commit is its own
228/// acquisition decision.
229#[derive(Debug, Default)]
230pub struct WriterSessionState {
231    /// Epoch acquired lazily on this session's first publish and reused for
232    /// its lifetime; no per-publish acquisition CAS.
233    acquired_writer: Option<AcquiredWriter>,
234    /// Terminal fencing record. Once another session supersedes our epoch,
235    /// every later publish fails with `writer_fenced` without touching the
236    /// store; the session never reacquires on its own. Reacquisition is an
237    /// explicit caller decision, left to a future takeover API.
238    fenced: Option<WriterFence>,
239}
240
241/// Shared handle to one namespace's [`WriterSessionState`].
242pub type SharedWriterSessionState = Arc<Mutex<WriterSessionState>>;
243
244#[derive(Debug, Clone)]
245pub struct NamespaceCommitEngine {
246    namespace_id: NamespaceId,
247    publish_tail_projection: Option<PublishTailProjection>,
248    /// This session's epoch and fencing for the namespace; see
249    /// [`WriterSessionState`].
250    session: SharedWriterSessionState,
251    /// Local monotonic source for the self-enforced publish budget.
252    timer: Arc<dyn MonotonicTimer>,
253    /// Shared decoded-block cache for publish-view table reads. Blocks are
254    /// content-addressed by segment digest, so cached entries can never
255    /// serve stale state; freshness stays enforced by the head etag check.
256    table_cache: Option<Arc<MetadataTableCache>>,
257}
258
259impl NamespaceCommitEngine {
260    pub fn new(namespace_id: NamespaceId) -> Self {
261        Self {
262            namespace_id,
263            publish_tail_projection: None,
264            session: SharedWriterSessionState::default(),
265            timer: Arc::new(StdMonotonicTimer::default()),
266            table_cache: None,
267        }
268    }
269
270    /// Attaches the runtime's session state for this namespace, so the
271    /// acquired epoch and fencing outlive this engine instance.
272    pub fn writer_session(mut self, session: SharedWriterSessionState) -> Self {
273        self.session = session;
274        self
275    }
276
277    fn lock_session(&self) -> std::sync::MutexGuard<'_, WriterSessionState> {
278        // Poisoning is propagated as a panic: every critical section is a
279        // plain field read or write, so a poisoned lock means another
280        // thread panicked mid-update.
281        self.session
282            .lock()
283            .expect("writer session state lock should not be poisoned")
284    }
285
286    #[cfg(test)]
287    pub(crate) fn monotonic_timer(mut self, timer: Arc<dyn MonotonicTimer>) -> Self {
288        self.timer = timer;
289        self
290    }
291
292    pub fn table_cache(mut self, table_cache: Arc<MetadataTableCache>) -> Self {
293        self.table_cache = Some(table_cache);
294        self
295    }
296
297    pub fn invalidate(&mut self) {
298        // Drops only the tail projection. The acquired epoch and fencing
299        // are session state, not cached state: the epoch's validity is
300        // re-checked against the head on every publish view load, and a
301        // fenced session stays fenced.
302        self.publish_tail_projection = None;
303    }
304
305    /// What the tail projection this engine retains weighs, or `None` when
306    /// it retains none.
307    ///
308    /// A runtime holding one engine per namespace bounds its total retention
309    /// with this; the per-projection ceiling in [`PublishTailOptions`] only
310    /// bounds one.
311    pub fn retained_tail_weight(&self) -> Option<PublishTailWeight> {
312        self.publish_tail_projection
313            .as_ref()
314            .map(PublishTailProjection::weight)
315    }
316
317    /// This session's writer epoch for the namespace, acquired on first use
318    /// and reused afterwards.
319    ///
320    /// Fencing is checked first and answered terminally: a superseded session
321    /// never touches the store again, and never reacquires on its own.
322    async fn session_writer_epoch<S: ObjectStore + ?Sized>(
323        &self,
324        store: &S,
325        context: &MutationContext,
326    ) -> Result<AcquiredWriter> {
327        let already_acquired = {
328            let session = self.lock_session();
329            if let Some(fence) = &session.fenced {
330                return Err(CoreError::WriterFenced(fence.clone()));
331            }
332            session.acquired_writer.clone()
333        };
334        if let Some(acquired_writer) = already_acquired {
335            return Ok(acquired_writer);
336        }
337        let acquired_writer = acquire_writer_epoch(store, &self.namespace_id, context)
338            .await
339            .map_err(CoreError::WriterEpoch)?;
340        let mut session = self.lock_session();
341        if let Some(fence) = session.fenced.clone() {
342            // Another engine sharing this session observed fencing while we
343            // were acquiring; the session stays fenced.
344            return Err(CoreError::WriterFenced(fence));
345        }
346        session.acquired_writer = Some(acquired_writer.clone());
347        Ok(acquired_writer)
348    }
349
350    /// Deletes the namespace through this session (format spec, "Tombstones
351    /// and deletion").
352    ///
353    /// Deletion is a head-advancing write, so it takes the same session gate
354    /// as [`Self::publish_batch`]: a fenced session is refused terminally
355    /// without touching the store, and the epoch acquired for publishing is
356    /// the epoch the tombstone swap is fenced by. A takeover observed by the
357    /// swap fences this session for good.
358    pub async fn delete_namespace<S: ObjectStore + ?Sized>(
359        &mut self,
360        store: &S,
361        options: DeleteNamespaceOptions,
362        context: &MutationContext,
363    ) -> Result<DeleteNamespaceResponse> {
364        let acquired_writer = self.session_writer_epoch(store, context).await?;
365        let deleted = crate::namespace::delete::delete_namespace(
366            store,
367            &self.namespace_id,
368            options,
369            acquired_writer,
370        )
371        .await;
372        if let Err(CoreError::WriterFenced(fence)) = &deleted {
373            let mut session = self.lock_session();
374            session.fenced = Some(fence.clone());
375            session.acquired_writer = None;
376        }
377        deleted
378    }
379
380    pub async fn publish_batch<S: ObjectStore + ?Sized>(
381        &mut self,
382        store: &S,
383        candidates: Vec<CommitCandidate>,
384        context: &MutationContext,
385        tail_options: &PublishTailOptions,
386    ) -> NamespaceCommitEnginePublishResult {
387        if candidates.is_empty() {
388            return NamespaceCommitEnginePublishResult {
389                results: Vec::new(),
390                wal_tail_segments: 0,
391                resulting_read_state: None,
392            };
393        }
394
395        let candidate_count = candidates.len();
396        let acquired_writer = match self.session_writer_epoch(store, context).await {
397            Ok(value) => value,
398            Err(error) => {
399                return NamespaceCommitEnginePublishResult {
400                    results: repeated_error(candidate_count, error),
401                    wal_tail_segments: 0,
402                    resulting_read_state: None,
403                };
404            }
405        };
406
407        let (publish_view, projection) = match load_publish_metadata_view(
408            store,
409            self.table_cache.as_deref(),
410            &self.namespace_id,
411            Some(acquired_writer),
412            self.publish_tail_projection.as_ref(),
413            tail_options,
414        )
415        .await
416        {
417            Ok(value) => value,
418            Err(error) => {
419                self.invalidate();
420                if let CoreError::WriterFenced(fence) = &error {
421                    let mut session = self.lock_session();
422                    session.fenced = Some(fence.clone());
423                    session.acquired_writer = None;
424                }
425                return NamespaceCommitEnginePublishResult {
426                    results: repeated_error(candidate_count, error),
427                    wal_tail_segments: 0,
428                    resulting_read_state: None,
429                };
430            }
431        };
432
433        let reject_writes_at_segments = WalTailPolicy::DEFAULT.reject_writes_at_segments;
434        if projection.wal_tail_segments > reject_writes_at_segments {
435            let wal_tail_segments = projection.wal_tail_segments;
436            self.publish_tail_projection = Some(projection);
437            let error = MetadataViewError::MaintenanceRequired {
438                namespace_id: self.namespace_id.clone(),
439                reason: format!(
440                    "wal tail has {wal_tail_segments} segments; publishes resume once maintenance brings it back under {reject_writes_at_segments}"
441                ),
442            };
443            return NamespaceCommitEnginePublishResult {
444                results: repeated_error(candidate_count, CoreError::from(error)),
445                wal_tail_segments,
446                resulting_read_state: None,
447            };
448        }
449
450        let published = crate::protocol::publish_namespace_commits_batch_against_publish_view(
451            store,
452            &self.namespace_id,
453            &candidates,
454            context,
455            &publish_view,
456            self.timer.as_ref(),
457        )
458        .await;
459        let resulting_head = published.resulting_head.clone();
460        let wal_tail_segments =
461            self.update_publish_tail_projection(projection, &published, tail_options);
462        // Seedable only when the CAS landed unambiguously and the updated
463        // projection survived (it carries the post-publish tail and etag).
464        let resulting_read_state = match (resulting_head, self.publish_tail_projection.as_ref()) {
465            (Some(head), Some(projection)) if projection.head_seq == head.seq => {
466                Some(ResultingReadState {
467                    head,
468                    head_etag: projection.head_etag.clone(),
469                    basis: projection.basis.clone(),
470                    manifest_id: projection.manifest_id,
471                    manifest_head_seq: projection.manifest_head_seq,
472                    tail_rows: Arc::new(projection.tail_state.clone()),
473                })
474            }
475            _ => None,
476        };
477        NamespaceCommitEnginePublishResult {
478            results: published.results,
479            wal_tail_segments,
480            resulting_read_state,
481        }
482    }
483
484    fn update_publish_tail_projection(
485        &mut self,
486        mut projection: PublishTailProjection,
487        published: &crate::protocol::PublishBatchAgainstViewResult,
488        tail_options: &PublishTailOptions,
489    ) -> u64 {
490        if !published.published_records.is_empty() {
491            projection.wal_tail_segments = projection.wal_tail_segments.saturating_add(1);
492        }
493        let wal_tail_segments = projection.wal_tail_segments;
494        let Some(resulting_head) = published.resulting_head.clone() else {
495            if published.can_reuse_loaded_projection {
496                self.publish_tail_projection = Some(projection);
497            } else {
498                self.invalidate();
499            }
500            return wal_tail_segments;
501        };
502        let Some(resulting_head_etag) = published.resulting_head_etag.clone() else {
503            self.invalidate();
504            return wal_tail_segments;
505        };
506        for record in &published.published_records {
507            projection.tail_state.apply_committed_wal_record_mut(record);
508        }
509        projection.head_seq = resulting_head.seq;
510        projection.head_etag = resulting_head_etag;
511        if projection.within_limits(tail_options) {
512            self.publish_tail_projection = Some(projection);
513        } else {
514            self.invalidate();
515        }
516        wal_tail_segments
517    }
518}
519
520fn repeated_error(count: usize, error: CoreError) -> Vec<Result<ApiCommitResponse>> {
521    (0..count).map(|_| Err(error.clone())).collect()
522}
523
524/// Publishes one batch through a fresh, uncached commit engine.
525pub(crate) async fn publish_namespace_commits_batch<S: ObjectStore + ?Sized>(
526    store: &S,
527    namespace_id: &NamespaceId,
528    candidates: Vec<CommitCandidate>,
529    context: &MutationContext,
530) -> Vec<Result<ApiCommitResponse>> {
531    let mut engine = NamespaceCommitEngine::new(namespace_id.clone());
532    engine
533        .publish_batch(store, candidates, context, &PublishTailOptions::default())
534        .await
535        .results
536}
537
538/// Deletes a namespace through a fresh, uncached commit engine: a one-shot
539/// session that acquires its own epoch, exactly like a one-shot publish.
540pub(crate) async fn delete_namespace<S: ObjectStore + ?Sized>(
541    store: &S,
542    namespace_id: &NamespaceId,
543    options: DeleteNamespaceOptions,
544    context: &MutationContext,
545) -> Result<DeleteNamespaceResponse> {
546    NamespaceCommitEngine::new(namespace_id.clone())
547        .delete_namespace(store, options, context)
548        .await
549}
550
551#[cfg(test)]
552mod tests {
553    use super::*;
554    use crate::error::ErrorCode;
555    use crate::limits::WAL_PUBLISH_BUDGET_MS;
556    use crate::namespace::bootstrap::bootstrap_namespace;
557    use crate::namespace::control::read_head_object;
558    use futures::StreamExt;
559    use loonfs_api::{ChangeSeq, ContentRef, ContentStoreId, WriterEpoch};
560    use loonfs_objectstore::keys::wal_segment_prefix;
561    use loonfs_objectstore::local_fs_store::LocalFsStore;
562    use loonfs_objectstore::ObjectStore;
563    use loonfs_test_support::stores::{CountingStore, OperationClass};
564    use std::sync::atomic::{AtomicU64, Ordering};
565    use tempfile::tempdir;
566
567    fn context(writer_id: &str) -> MutationContext {
568        MutationContext {
569            writer_id: writer_id.to_owned(),
570            now_ms: 1_000,
571        }
572    }
573
574    fn create_dir_request(commit_id: &str, name: &str) -> CommitRequest {
575        CommitRequest::single(
576            CommitId::parse(commit_id).expect("valid commit id"),
577            None,
578            FilesystemOperation::CreateDirectory {
579                path: loonfs_api::AbsolutePath::parse(format!("/{name}")).expect("valid path"),
580                parents: false,
581            },
582        )
583    }
584
585    #[test]
586    fn semantic_identity_excludes_content_preparation() {
587        let namespace_id = NamespaceId::parse("demo").expect("valid namespace id");
588        let request = create_dir_request("same-mutation", "docs");
589        let ready = CommitCandidate::new(request.clone());
590        let rejected = CommitCandidate::rejected(
591            request,
592            ContentPreparationError::ContentToken(ContentTokenError::Expired),
593        );
594
595        assert_eq!(
596            ready.semantic_identity(&namespace_id).expect("identity"),
597            rejected.semantic_identity(&namespace_id).expect("identity")
598        );
599    }
600
601    #[test]
602    fn semantic_identity_ignores_current_request_limits() {
603        let namespace_id = NamespaceId::parse("demo").expect("valid namespace id");
604        let oversized_ops = CommitCandidate::new(CommitRequest {
605            commit_id: CommitId::parse("too-many-ops").expect("valid commit id"),
606            message: None,
607            operations: (0..=crate::limits::MAX_COMMIT_OPERATIONS)
608                .map(|index| FilesystemOperation::CreateDirectory {
609                    path: loonfs_api::AbsolutePath::parse(format!("/dir-{index}"))
610                        .expect("valid path"),
611                    parents: false,
612                })
613                .collect(),
614        });
615        oversized_ops
616            .semantic_identity(&namespace_id)
617            .expect("operation limits must not affect identity");
618
619        let content_ref = ContentRef::blob_v1(ContentId::generate(), b"proof");
620        let admission = ContentAdmission::for_durable_content_write(
621            ContentStoreId::parse("cs_00000000000000000000000000000001").expect("content store id"),
622            content_ref,
623        );
624        let prepared = PreparedContent::from_admission(admission);
625        let oversized_proofs = CommitCandidate::prepared(
626            create_dir_request("too-many-proofs", "docs"),
627            vec![prepared; crate::limits::MAX_COMMIT_CONTENT_TOKENS + 1],
628        );
629        oversized_proofs
630            .semantic_identity(&namespace_id)
631            .expect("prepared proof limits must not affect identity");
632
633        let oversized_message = CommitCandidate::new(CommitRequest {
634            commit_id: CommitId::parse("too-long-message").expect("valid commit id"),
635            message: Some("m".repeat(crate::limits::MAX_COMMIT_MESSAGE_BYTES + 1)),
636            operations: vec![FilesystemOperation::CreateDirectory {
637                path: loonfs_api::AbsolutePath::parse("/docs").expect("valid path"),
638                parents: false,
639            }],
640        });
641        oversized_message
642            .semantic_identity(&namespace_id)
643            .expect("message limits must not affect identity");
644    }
645
646    /// A batch past the operation ceiling is refused before it can occupy
647    /// the publisher.
648    #[test]
649    fn a_batch_past_the_operation_ceiling_is_rejected() {
650        let oversized = CommitCandidate::new(CommitRequest {
651            commit_id: CommitId::parse("oversized-batch").expect("valid commit id"),
652            message: None,
653            operations: (0..=crate::limits::MAX_COMMIT_OPERATIONS)
654                .map(|index| FilesystemOperation::CreateDirectory {
655                    path: loonfs_api::AbsolutePath::parse(format!("/dir-{index}"))
656                        .expect("valid path"),
657                    parents: false,
658                })
659                .collect(),
660        });
661
662        let error = oversized
663            .validate_request_limits()
664            .expect_err("the batch is over the operation ceiling");
665        assert_eq!(error.code(), ErrorCode::InvalidRequest);
666
667        let at_ceiling = CommitCandidate::new(CommitRequest {
668            commit_id: CommitId::parse("largest-batch").expect("valid commit id"),
669            message: None,
670            operations: (0..crate::limits::MAX_COMMIT_OPERATIONS)
671                .map(|index| FilesystemOperation::CreateDirectory {
672                    path: loonfs_api::AbsolutePath::parse(format!("/dir-{index}"))
673                        .expect("valid path"),
674                    parents: false,
675                })
676                .collect(),
677        });
678        at_ceiling
679            .validate_request_limits()
680            .expect("a batch at the ceiling is admitted");
681    }
682
683    /// A message past the byte ceiling is refused before it can enter the
684    /// durable record or the fingerprint path.
685    #[test]
686    fn a_message_past_the_byte_ceiling_is_rejected() {
687        let operations = vec![FilesystemOperation::CreateDirectory {
688            path: loonfs_api::AbsolutePath::parse("/docs").expect("valid path"),
689            parents: false,
690        }];
691
692        let oversized = CommitCandidate::new(CommitRequest {
693            commit_id: CommitId::parse("oversized-message").expect("valid commit id"),
694            message: Some("m".repeat(crate::limits::MAX_COMMIT_MESSAGE_BYTES + 1)),
695            operations: operations.clone(),
696        });
697        let error = oversized
698            .validate_request_limits()
699            .expect_err("the message is over the byte ceiling");
700        assert_eq!(error.code(), ErrorCode::InvalidRequest);
701
702        let at_ceiling = CommitCandidate::new(CommitRequest {
703            commit_id: CommitId::parse("largest-message").expect("valid commit id"),
704            message: Some("m".repeat(crate::limits::MAX_COMMIT_MESSAGE_BYTES)),
705            operations,
706        });
707        at_ceiling
708            .validate_request_limits()
709            .expect("a message at the ceiling is admitted");
710    }
711
712    fn create_dir(commit_id: &str, display_name: &str) -> CommitCandidate {
713        CommitCandidate::new(create_dir_request(commit_id, display_name))
714    }
715
716    async fn wal_segment_count(store: &LocalFsStore, namespace_id: &NamespaceId) -> usize {
717        store
718            .list_prefix_stream(&wal_segment_prefix(namespace_id.as_str()))
719            .collect::<Vec<_>>()
720            .await
721            .len()
722    }
723
724    #[tokio::test]
725    async fn commit_engine_is_terminally_fenced_after_takeover() {
726        let temp_dir = tempdir().expect("tempdir");
727        let store = LocalFsStore::new(temp_dir.path()).expect("store");
728        let namespace_id = NamespaceId::parse("demo").expect("valid namespace id");
729        let writer_a = context("writer-a");
730        bootstrap_namespace(&store, &namespace_id, &writer_a, false)
731            .await
732            .expect("bootstrap");
733
734        let mut engine_a = NamespaceCommitEngine::new(namespace_id.clone());
735        let first = engine_a
736            .publish_batch(
737                &store,
738                vec![create_dir("from-a-first", "alpha")],
739                &writer_a,
740                &PublishTailOptions::default(),
741            )
742            .await;
743        first.results[0].as_ref().expect("writer a first commit");
744
745        // Writer B's session acquires the epoch; A's cached epoch is now
746        // superseded.
747        let writer_b = context("writer-b");
748        let mut engine_b = NamespaceCommitEngine::new(namespace_id.clone());
749        let takeover = engine_b
750            .publish_batch(
751                &store,
752                vec![create_dir("from-b-first", "beta")],
753                &writer_b,
754                &PublishTailOptions::default(),
755            )
756            .await;
757        takeover.results[0]
758            .as_ref()
759            .expect("writer b takeover commit");
760        let epoch_after_takeover = read_head_object(&store, &namespace_id)
761            .await
762            .expect("read head")
763            .envelope
764            .state
765            .writer_epoch;
766
767        // A is fenced terminally: both attempts fail with writer_fenced, the
768        // second without ever reaching the store, and the session never
769        // bumps the epoch back.
770        for attempt in 0..2 {
771            let fenced = engine_a
772                .publish_batch(
773                    &store,
774                    vec![create_dir("from-a-second", "gamma")],
775                    &writer_a,
776                    &PublishTailOptions::default(),
777                )
778                .await;
779            let error = fenced.results[0].as_ref().expect_err("fenced publish");
780            assert_eq!(error.code(), ErrorCode::WriterFenced, "attempt {attempt}");
781        }
782        let head = read_head_object(&store, &namespace_id)
783            .await
784            .expect("read head")
785            .envelope
786            .state;
787        assert_eq!(head.writer_epoch, epoch_after_takeover);
788        assert_eq!(head.writer.expect("writer block").writer_id, "writer-b");
789    }
790
791    /// A takeover that lands while the loser is mid-load is still a fence,
792    /// not a head race. The loser must be told so — `stale_head` would send a
793    /// permanently fenced session back to retry.
794    #[tokio::test]
795    async fn fencing_during_publish_view_load_still_reports_writer_fenced() {
796        use loonfs_test_support::stores::{BlockingStore, KeyPredicate};
797        use std::sync::Arc as StdArc;
798
799        let temp_dir = tempdir().expect("tempdir");
800        let namespace_id = NamespaceId::parse("demo").expect("valid namespace id");
801        let writer_a = context("writer-a");
802
803        // Block the WAL tail read: it sits between the head snapshot that the
804        // fence check uses and the closing etag recheck.
805        let store = StdArc::new(BlockingStore::new(
806            LocalFsStore::new(temp_dir.path()).expect("store"),
807            KeyPredicate::prefix(wal_segment_prefix(namespace_id.as_str())),
808            OperationClass::Read,
809        ));
810
811        bootstrap_namespace(store.inner(), &namespace_id, &writer_a, false)
812            .await
813            .expect("bootstrap");
814
815        let mut engine_a = NamespaceCommitEngine::new(namespace_id.clone());
816        engine_a
817            .publish_batch(
818                store.inner(),
819                vec![create_dir("from-a-first", "alpha")],
820                &writer_a,
821                &PublishTailOptions::default(),
822            )
823            .await
824            .results[0]
825            .as_ref()
826            .expect("writer a first commit");
827        // Force a fresh manifest load on the next publish.
828        engine_a.invalidate();
829
830        store.block_next();
831        let blocked_store = StdArc::clone(&store);
832        let publish_a = tokio::spawn(async move {
833            let mut engine = engine_a;
834            let result = engine
835                .publish_batch(
836                    blocked_store.as_ref(),
837                    vec![create_dir("from-a-second", "gamma")],
838                    &writer_a,
839                    &PublishTailOptions::default(),
840                )
841                .await;
842            result.results[0].as_ref().err().map(|error| error.code())
843        });
844
845        // A has snapshotted a head that still names it. Writer B takes the
846        // epoch while A is parked mid-load, so A is fenced by the time it
847        // rechecks the etag.
848        store.wait_until_blocked().await;
849        let writer_b = context("writer-b");
850        let mut engine_b = NamespaceCommitEngine::new(namespace_id.clone());
851        engine_b
852            .publish_batch(
853                store.inner(),
854                vec![create_dir("from-b-first", "beta")],
855                &writer_b,
856                &PublishTailOptions::default(),
857            )
858            .await
859            .results[0]
860            .as_ref()
861            .expect("writer b takeover commit");
862        store.release();
863
864        let code = publish_a.await.expect("join publish a");
865        assert_eq!(code, Some(ErrorCode::WriterFenced));
866    }
867
868    #[tokio::test]
869    async fn shared_session_keeps_fencing_across_engine_rebuilds() {
870        let temp_dir = tempdir().expect("tempdir");
871        let store = LocalFsStore::new(temp_dir.path()).expect("store");
872        let namespace_id = NamespaceId::parse("demo").expect("valid namespace id");
873        let writer_a = context("writer-a");
874        bootstrap_namespace(&store, &namespace_id, &writer_a, false)
875            .await
876            .expect("bootstrap");
877
878        let session = SharedWriterSessionState::default();
879        let mut engine_a1 =
880            NamespaceCommitEngine::new(namespace_id.clone()).writer_session(Arc::clone(&session));
881        engine_a1
882            .publish_batch(
883                &store,
884                vec![create_dir("from-a-first", "alpha")],
885                &writer_a,
886                &PublishTailOptions::default(),
887            )
888            .await
889            .results
890            .remove(0)
891            .expect("writer a first commit");
892
893        let writer_b = context("writer-b");
894        let mut engine_b = NamespaceCommitEngine::new(namespace_id.clone());
895        engine_b
896            .publish_batch(
897                &store,
898                vec![create_dir("from-b-first", "beta")],
899                &writer_b,
900                &PublishTailOptions::default(),
901            )
902            .await
903            .results
904            .remove(0)
905            .expect("writer b takeover commit");
906
907        let fenced = engine_a1
908            .publish_batch(
909                &store,
910                vec![create_dir("from-a-second", "gamma")],
911                &writer_a,
912                &PublishTailOptions::default(),
913            )
914            .await;
915        let error = fenced.results[0].as_ref().expect_err("fenced publish");
916        assert_eq!(error.code(), ErrorCode::WriterFenced);
917        let epoch_after_fencing = read_head_object(&store, &namespace_id)
918            .await
919            .expect("read head")
920            .envelope
921            .state
922            .writer_epoch;
923
924        // A rebuilt engine — cache eviction, cache-disabled mode — shares
925        // the session state, so the session stays terminally fenced and
926        // never touches the head.
927        drop(engine_a1);
928        let mut engine_a2 =
929            NamespaceCommitEngine::new(namespace_id.clone()).writer_session(session);
930        let still_fenced = engine_a2
931            .publish_batch(
932                &store,
933                vec![create_dir("from-a-third", "delta")],
934                &writer_a,
935                &PublishTailOptions::default(),
936            )
937            .await;
938        let error = still_fenced.results[0]
939            .as_ref()
940            .expect_err("rebuilt engine stays fenced");
941        assert_eq!(error.code(), ErrorCode::WriterFenced);
942        let head = read_head_object(&store, &namespace_id)
943            .await
944            .expect("read head")
945            .envelope
946            .state;
947        assert_eq!(head.writer_epoch, epoch_after_fencing);
948        assert_eq!(head.writer.expect("writer block").writer_id, "writer-b");
949    }
950
951    /// Advances an entire publish budget per reading, so every publish
952    /// observes an expired budget between segment PUT and head CAS.
953    #[derive(Debug)]
954    struct ExpiredBudgetTimer(AtomicU64);
955
956    impl MonotonicTimer for ExpiredBudgetTimer {
957        fn monotonic_now_ms(&self) -> u64 {
958            self.0
959                .fetch_add(WAL_PUBLISH_BUDGET_MS + 1_000, Ordering::SeqCst)
960        }
961    }
962
963    #[tokio::test]
964    async fn publish_over_budget_abandons_the_segment_and_a_retry_rebuilds() {
965        let temp_dir = tempdir().expect("tempdir");
966        let store = LocalFsStore::new(temp_dir.path()).expect("store");
967        let namespace_id = NamespaceId::parse("demo").expect("valid namespace id");
968        let writer = context("writer-a");
969        bootstrap_namespace(&store, &namespace_id, &writer, false)
970            .await
971            .expect("bootstrap");
972        let head_before = read_head_object(&store, &namespace_id)
973            .await
974            .expect("read head")
975            .envelope
976            .state;
977
978        let mut over_budget = NamespaceCommitEngine::new(namespace_id.clone())
979            .monotonic_timer(Arc::new(ExpiredBudgetTimer(AtomicU64::new(0))));
980        let abandoned = over_budget
981            .publish_batch(
982                &store,
983                vec![create_dir("budgeted", "alpha")],
984                &writer,
985                &PublishTailOptions::default(),
986            )
987            .await;
988        let error = abandoned.results[0]
989            .as_ref()
990            .expect_err("over-budget publish must abandon");
991        assert!(
992            matches!(
993                error,
994                CoreError::HeadPublish(
995                    crate::commit::CommitHeadPublishError::PublishBudgetExceeded { .. }
996                )
997            ),
998            "unexpected error: {error:?}"
999        );
1000        // Retryable exactly like a stale head, so existing retry loops
1001        // rebuild the commit.
1002        assert_eq!(error.code(), ErrorCode::StaleHead);
1003
1004        // The head did not advance; the written segment is an orphan for GC.
1005        let head_after = read_head_object(&store, &namespace_id)
1006            .await
1007            .expect("read head")
1008            .envelope
1009            .state;
1010        assert_eq!(head_after.seq, head_before.seq);
1011        assert_eq!(head_after.visible_wal_tip, head_before.visible_wal_tip);
1012        assert_eq!(wal_segment_count(&store, &namespace_id).await, 1);
1013
1014        // A retry with a healthy budget republishes the same commit as a
1015        // fresh segment; the orphan stays behind.
1016        let mut healthy = NamespaceCommitEngine::new(namespace_id.clone());
1017        let retried = healthy
1018            .publish_batch(
1019                &store,
1020                vec![create_dir("budgeted", "alpha")],
1021                &writer,
1022                &PublishTailOptions::default(),
1023            )
1024            .await;
1025        let response = retried.results[0].as_ref().expect("rebuilt publish");
1026        assert_eq!(response.committed_seq, ChangeSeq(1));
1027        assert_eq!(wal_segment_count(&store, &namespace_id).await, 2);
1028        let head_final = read_head_object(&store, &namespace_id)
1029            .await
1030            .expect("read head")
1031            .envelope
1032            .state;
1033        assert_eq!(head_final.seq, ChangeSeq(1));
1034        // Two engines are two sessions, and each acquires its own epoch: the
1035        // abandoned attempt took 1, the retry took 2.
1036        assert_eq!(head_final.writer_epoch, WriterEpoch(2));
1037    }
1038
1039    #[tokio::test]
1040    async fn publish_views_reuse_cached_table_blocks_across_publishes() {
1041        use crate::cache::MetadataTableCacheConfig;
1042        let temp_dir = tempdir().expect("tempdir");
1043        let store =
1044            CountingStore::metadata_tables(LocalFsStore::new(temp_dir.path()).expect("store"));
1045        let namespace_id = NamespaceId::parse("demo").expect("valid namespace id");
1046        let writer = context("writer-a");
1047        bootstrap_namespace(&store, &namespace_id, &writer, false)
1048            .await
1049            .expect("bootstrap");
1050        let mut seed = NamespaceCommitEngine::new(namespace_id.clone());
1051        seed.publish_batch(
1052            &store,
1053            vec![create_dir("seed-commit", "docs")],
1054            &writer,
1055            &PublishTailOptions::default(),
1056        )
1057        .await
1058        .results
1059        .remove(0)
1060        .expect("seed publish");
1061        crate::checkpoint::create_checkpoint(
1062            &store,
1063            &namespace_id,
1064            loonfs_api::wire::control::CheckpointOwner::User {
1065                name: "test-pin".to_owned(),
1066            },
1067            None,
1068            &writer,
1069        )
1070        .await
1071        .expect("checkpoint");
1072
1073        // Without a cache, every publish view re-fetches the table blocks
1074        // its validation walks need.
1075        let mut uncached = NamespaceCommitEngine::new(namespace_id.clone());
1076        store.reset();
1077        uncached
1078            .publish_batch(
1079                &store,
1080                vec![create_dir("uncached-a", "alpha")],
1081                &writer,
1082                &PublishTailOptions::default(),
1083            )
1084            .await
1085            .results
1086            .remove(0)
1087            .expect("uncached publish a");
1088        assert!(
1089            store.count(OperationClass::Read) > 0,
1090            "publish validation should read table blocks"
1091        );
1092        store.reset();
1093        uncached
1094            .publish_batch(
1095                &store,
1096                vec![create_dir("uncached-b", "beta")],
1097                &writer,
1098                &PublishTailOptions::default(),
1099            )
1100            .await
1101            .results
1102            .remove(0)
1103            .expect("uncached publish b");
1104        assert!(
1105            store.count(OperationClass::Read) > 0,
1106            "without a cache the next publish re-fetches the same blocks"
1107        );
1108
1109        let cache = Arc::new(MetadataTableCache::new(MetadataTableCacheConfig::default()));
1110        let mut cached = NamespaceCommitEngine::new(namespace_id.clone()).table_cache(cache);
1111        store.reset();
1112        cached
1113            .publish_batch(
1114                &store,
1115                vec![create_dir("cached-a", "gamma")],
1116                &writer,
1117                &PublishTailOptions::default(),
1118            )
1119            .await
1120            .results
1121            .remove(0)
1122            .expect("cached publish a");
1123        assert!(
1124            store.count(OperationClass::Read) > 0,
1125            "the first cached publish fills the cache"
1126        );
1127        store.reset();
1128        cached
1129            .publish_batch(
1130                &store,
1131                vec![create_dir("cached-b", "delta")],
1132                &writer,
1133                &PublishTailOptions::default(),
1134            )
1135            .await
1136            .results
1137            .remove(0)
1138            .expect("cached publish b");
1139        assert_eq!(
1140            store.count(OperationClass::Read),
1141            0,
1142            "a warm cache serves every publish-view table read"
1143        );
1144    }
1145}
loonfs_core/commit_engine.rs

loonfs_core/
commit_engine.rs
