rocketmq_runtime/
metadata_target.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::HashMap;
16use std::ffi::OsString;
17use std::fmt;
18use std::path::Path;
19use std::sync::atomic::AtomicU64;
20use std::sync::atomic::Ordering;
21use std::sync::Arc;
22
23use parking_lot::Mutex;
24
25use crate::error::RuntimeError;
26use crate::error::RuntimeResult;
27use crate::metadata_io::MetadataGeneration;
28use crate::RuntimeOperation;
29
30pub(crate) const DEFAULT_MAX_METADATA_TARGETS: usize = 4_096;
31
32/// Process-local identity of one retained target history.
33///
34/// Normal actor replacement preserves this identity. Explicit retirement ends
35/// that history; generations and receipts from different identities must never
36/// be compared as durability evidence. Holding an identity grants no write access.
37#[derive(Clone, Debug)]
38pub struct MetadataTargetIdentity(Arc<()>);
39
40impl PartialEq for MetadataTargetIdentity {
41    fn eq(&self, other: &Self) -> bool {
42        Arc::ptr_eq(&self.0, &other.0)
43    }
44}
45
46impl Eq for MetadataTargetIdentity {}
47
48/// Aggregate registry occupancy without resource names or target paths.
49#[derive(Debug, Clone, Copy, serde::Serialize)]
50pub struct MetadataTargetRegistryStats {
51    /// Maximum retained target histories, including fenced and idle entries.
52    pub capacity: usize,
53    /// Current retained target histories.
54    pub retained_targets: usize,
55    /// Histories with at least one live write registration.
56    pub live_owners: usize,
57    /// Histories without a live writer; this includes idle fenced histories.
58    pub idle_histories: usize,
59    /// Histories requiring format-specific reconciliation, with or without a writer.
60    pub fenced_targets: usize,
61    /// Capacity available for distinct target histories.
62    pub remaining_capacity: usize,
63}
64
65/// Result of explicitly ending a quiescent target's history.
66#[derive(Debug, Clone, Copy, PartialEq, Eq)]
67pub enum MetadataTargetRetirementOutcome {
68    /// History was removed; future actors must acquire a new identity.
69    Retired,
70    /// This actor has no history for the requested resource.
71    NotFound,
72    /// The actor can still accept writes.
73    AdmissionOpen,
74    /// Coordinator work, waiters, or actual write authority remain alive.
75    WorkInProgress,
76    /// An unconfirmed replacement requires domain-specific recovery.
77    ReconciliationRequired,
78    /// The registry's identity or durable history differs from this actor's evidence.
79    HistoryChanged,
80}
81
82/// One process-local registry shared by metadata actors under a `RuntimeOwner`.
83#[derive(Clone)]
84pub(crate) struct MetadataTargetRegistry {
85    inner: Arc<RegistryInner>,
86}
87
88struct RegistryInner {
89    state: Mutex<RegistryState>,
90    next_owner: AtomicU64,
91    next_generation: AtomicU64,
92    max_entries: usize,
93}
94
95#[derive(Default)]
96struct RegistryState {
97    targets: HashMap<NormalizedMetadataTarget, TargetEntry>,
98}
99
100struct TargetEntry {
101    identity: MetadataTargetIdentity,
102    resource: Arc<str>,
103    owner: Option<u64>,
104    durable_generation: Option<MetadataGeneration>,
105    reconciliation_required: bool,
106}
107
108#[derive(Clone, PartialEq, Eq, Hash)]
109struct NormalizedMetadataTarget(OsString);
110
111/// A registration that keeps the target write authority alive until every
112/// clone, including an in-flight blocking closure, has been destroyed.
113#[derive(Clone)]
114pub(crate) struct MetadataTargetRegistration {
115    inner: Arc<RegistrationInner>,
116}
117
118struct RegistrationInner {
119    identity: MetadataTargetIdentity,
120    registry: MetadataTargetRegistry,
121    target: NormalizedMetadataTarget,
122    resource: Arc<str>,
123    owner: u64,
124    durable_generation: Option<MetadataGeneration>,
125}
126
127#[derive(Debug)]
128pub(crate) enum MetadataTargetRegistrationOutcome {
129    Registered(MetadataTargetRegistration),
130    Conflict,
131    ReconciliationRequired,
132}
133
134impl MetadataTargetRegistry {
135    #[cfg(test)]
136    pub(crate) fn new() -> Self {
137        Self::with_max_entries(DEFAULT_MAX_METADATA_TARGETS)
138    }
139
140    pub(crate) fn with_max_entries(max_entries: usize) -> Self {
141        Self {
142            inner: Arc::new(RegistryInner {
143                state: Mutex::new(RegistryState::default()),
144                next_owner: AtomicU64::new(1),
145                next_generation: AtomicU64::new(1),
146                max_entries,
147            }),
148        }
149    }
150
151    /// Reserves the next generation for a snapshot admitted under this owner.
152    ///
153    /// The counter belongs to the owner-scoped registry rather than to one
154    /// actor, so a generation remains a usable change identity after an actor
155    /// is replaced. A per-actor counter would restart at one and let a rebuilt
156    /// actor reuse a generation whose durability is still unconfirmed.
157    ///
158    /// Values wrap from `u64::MAX` back to one. Gaps are allowed when admission
159    /// rejects a request; they do not weaken ordering.
160    pub(crate) fn next_generation(&self) -> u64 {
161        let mut generation = self.inner.next_generation.load(Ordering::Relaxed);
162        loop {
163            let next = if generation == u64::MAX { 1 } else { generation + 1 };
164            match self.inner.next_generation.compare_exchange_weak(
165                generation,
166                next,
167                Ordering::Relaxed,
168                Ordering::Relaxed,
169            ) {
170                Ok(_) => return generation,
171                Err(observed) => generation = observed,
172            }
173        }
174    }
175
176    /// Returns whether the target still requires reconciliation.
177    fn requires_reconciliation(&self, target: &NormalizedMetadataTarget) -> bool {
178        let state = self.inner.state.lock();
179        state
180            .targets
181            .get(target)
182            .is_some_and(|entry| entry.reconciliation_required)
183    }
184
185    #[cfg(test)]
186    pub(crate) fn register(
187        &self,
188        target: &Path,
189        resource: Arc<str>,
190    ) -> RuntimeResult<MetadataTargetRegistrationOutcome> {
191        self.register_with_identity(target, resource, None)
192    }
193
194    pub(crate) fn register_with_identity(
195        &self,
196        target: &Path,
197        resource: Arc<str>,
198        expected: Option<&MetadataTargetIdentity>,
199    ) -> RuntimeResult<MetadataTargetRegistrationOutcome> {
200        let target = normalize_target(target)?;
201        let mut state = self.inner.state.lock();
202        if expected.is_some_and(|identity| {
203            state
204                .targets
205                .get(&target)
206                .is_none_or(|entry| &entry.identity != identity)
207        }) {
208            return Err(RuntimeError::context_unavailable(
209                RuntimeOperation::MetadataResourceTarget,
210            ));
211        }
212
213        if let Some(entry) = state.targets.get(&target) {
214            if entry.reconciliation_required {
215                return Ok(MetadataTargetRegistrationOutcome::ReconciliationRequired);
216            }
217            if entry.resource != resource || entry.owner.is_some() {
218                return Ok(MetadataTargetRegistrationOutcome::Conflict);
219            }
220        } else if state.targets.len() >= self.inner.max_entries {
221            return Err(RuntimeError::capacity(RuntimeOperation::AdmitMetadataOperation));
222        }
223
224        let owner = self.inner.next_owner.fetch_add(1, Ordering::Relaxed);
225        let entry = state.targets.entry(target.clone()).or_insert_with(|| TargetEntry {
226            identity: MetadataTargetIdentity(Arc::new(())),
227            resource: Arc::clone(&resource),
228            owner: None,
229            durable_generation: None,
230            reconciliation_required: false,
231        });
232        entry.owner = Some(owner);
233        let durable_generation = entry.durable_generation;
234
235        Ok(MetadataTargetRegistrationOutcome::Registered(
236            MetadataTargetRegistration {
237                inner: Arc::new(RegistrationInner {
238                    identity: entry.identity.clone(),
239                    registry: self.clone(),
240                    target,
241                    resource,
242                    owner,
243                    durable_generation,
244                }),
245            },
246        ))
247    }
248
249    pub(crate) fn same_target(&self, left: &Path, right: &Path) -> RuntimeResult<bool> {
250        Ok(normalize_target(left)? == normalize_target(right)?)
251    }
252
253    pub(crate) fn capacity(&self) -> usize {
254        self.inner.max_entries
255    }
256
257    pub(crate) fn stats(&self) -> MetadataTargetRegistryStats {
258        let state = self.inner.state.lock();
259        let live_owners = state.targets.values().filter(|entry| entry.owner.is_some()).count();
260        MetadataTargetRegistryStats {
261            capacity: self.inner.max_entries,
262            retained_targets: state.targets.len(),
263            live_owners,
264            idle_histories: state.targets.len() - live_owners,
265            fenced_targets: state
266                .targets
267                .values()
268                .filter(|entry| entry.reconciliation_required)
269                .count(),
270            remaining_capacity: self.inner.max_entries.saturating_sub(state.targets.len()),
271        }
272    }
273
274    pub(crate) fn retire(
275        &self,
276        target: &Path,
277        resource: &str,
278        identity: &MetadataTargetIdentity,
279        durable_generation: Option<MetadataGeneration>,
280    ) -> RuntimeResult<MetadataTargetRetirementOutcome> {
281        let target = normalize_target(target)?;
282        let mut state = self.inner.state.lock();
283        let Some(entry) = state.targets.get(&target) else {
284            return Ok(MetadataTargetRetirementOutcome::HistoryChanged);
285        };
286        if &entry.identity != identity
287            || entry.resource.as_ref() != resource
288            || entry.durable_generation != durable_generation
289        {
290            return Ok(MetadataTargetRetirementOutcome::HistoryChanged);
291        }
292        if entry.reconciliation_required {
293            return Ok(MetadataTargetRetirementOutcome::ReconciliationRequired);
294        }
295        if entry.owner.is_some() {
296            return Ok(MetadataTargetRetirementOutcome::WorkInProgress);
297        }
298        state.targets.remove(&target);
299        Ok(MetadataTargetRetirementOutcome::Retired)
300    }
301
302    fn record_durable(
303        &self,
304        target: &NormalizedMetadataTarget,
305        resource: &Arc<str>,
306        owner: u64,
307        generation: MetadataGeneration,
308    ) -> bool {
309        let mut state = self.inner.state.lock();
310        let Some(entry) = state.targets.get_mut(target) else {
311            return false;
312        };
313        if entry.resource != *resource || entry.owner != Some(owner) {
314            return false;
315        }
316        entry.durable_generation = Some(
317            entry
318                .durable_generation
319                .map_or(generation, |durable| durable.max(generation)),
320        );
321        true
322    }
323
324    fn release(&self, target: &NormalizedMetadataTarget, owner: u64) {
325        let mut state = self.inner.state.lock();
326        let Some(entry) = state.targets.get_mut(target) else {
327            return;
328        };
329        if entry.owner == Some(owner) {
330            entry.owner = None;
331        }
332    }
333
334    fn record_reconciliation_required(
335        &self,
336        target: &NormalizedMetadataTarget,
337        resource: &Arc<str>,
338        owner: u64,
339    ) -> bool {
340        let mut state = self.inner.state.lock();
341        let Some(entry) = state.targets.get_mut(target) else {
342            return false;
343        };
344        if entry.resource != *resource || entry.owner != Some(owner) {
345            return false;
346        }
347        entry.reconciliation_required = true;
348        true
349    }
350}
351
352impl MetadataTargetRegistration {
353    pub(crate) fn identity(&self) -> MetadataTargetIdentity {
354        self.inner.identity.clone()
355    }
356    pub(crate) fn durable_generation(&self) -> Option<MetadataGeneration> {
357        self.inner.durable_generation
358    }
359
360    pub(crate) fn record_durable(&self, generation: MetadataGeneration) -> bool {
361        self.inner
362            .registry
363            .record_durable(&self.inner.target, &self.inner.resource, self.inner.owner, generation)
364    }
365
366    pub(crate) fn record_reconciliation_required(&self) -> bool {
367        self.inner
368            .registry
369            .record_reconciliation_required(&self.inner.target, &self.inner.resource, self.inner.owner)
370    }
371
372    /// Returns whether the bound target still requires reconciliation.
373    ///
374    /// The flag describes the *target*, not this writer: an earlier generation
375    /// may have replaced the file without completing the durability protocol.
376    /// A caller that reuses a cached registration must consult it before
377    /// exercising write authority again, because the registry conflict check
378    /// is skipped on that path.
379    pub(crate) fn reconciliation_required(&self) -> bool {
380        self.inner.registry.requires_reconciliation(&self.inner.target)
381    }
382}
383
384impl Drop for RegistrationInner {
385    fn drop(&mut self) {
386        self.registry.release(&self.target, self.owner);
387    }
388}
389
390impl fmt::Debug for MetadataTargetRegistry {
391    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
392        let entries = self.inner.state.lock().targets.len();
393        formatter
394            .debug_struct("MetadataTargetRegistry")
395            .field("entries", &entries)
396            .field("max_entries", &self.inner.max_entries)
397            .finish()
398    }
399}
400
401impl fmt::Debug for MetadataTargetRegistration {
402    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
403        formatter
404            .debug_struct("MetadataTargetRegistration")
405            .field("resource", &self.inner.resource)
406            .finish_non_exhaustive()
407    }
408}
409
410impl fmt::Debug for NormalizedMetadataTarget {
411    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
412        formatter.write_str("<metadata-target>")
413    }
414}
415
416fn normalize_target(target: &Path) -> RuntimeResult<NormalizedMetadataTarget> {
417    let absolute = std::path::absolute(target)
418        .map_err(|source| RuntimeError::io(RuntimeOperation::MetadataResourceTarget, source))?;
419
420    #[cfg(windows)]
421    {
422        let folded = absolute.as_os_str().to_string_lossy().to_lowercase();
423        Ok(NormalizedMetadataTarget(OsString::from(folded)))
424    }
425
426    #[cfg(not(windows))]
427    {
428        Ok(NormalizedMetadataTarget(absolute.into_os_string()))
429    }
430}
431
432#[cfg(test)]
433mod tests {
434    use super::*;
435
436    #[test]
437    fn target_generations_are_allocated_by_the_owner_scoped_registry() {
438        let registry = MetadataTargetRegistry::new();
439        assert_eq!(registry.next_generation(), 1);
440        assert_eq!(registry.next_generation(), 2);
441        assert_eq!(registry.next_generation(), 3);
442
443        // A clone shares the counter, so two actors derived from one owner
444        // cannot reuse a generation whose durability is still unconfirmed.
445        let shared = registry.clone();
446        assert_eq!(shared.next_generation(), 4);
447        assert_eq!(registry.next_generation(), 5);
448    }
449
450    #[test]
451    fn registration_reports_the_target_reconciliation_fence() {
452        let registry = MetadataTargetRegistry::new();
453        let target = std::env::temp_dir().join("rocketmq-runtime-registration-fence");
454        let outcome = registry.register(&target, Arc::<str>::from("resource")).unwrap();
455        let MetadataTargetRegistrationOutcome::Registered(registration) = outcome else {
456            panic!("a free target should register");
457        };
458
459        assert!(!registration.reconciliation_required());
460        assert!(registration.record_reconciliation_required());
461        assert!(
462            registration.reconciliation_required(),
463            "a cached registration must observe an unconfirmed replacement"
464        );
465    }
466
467    #[test]
468    fn different_resources_cannot_bind_the_same_target() {
469        let registry = MetadataTargetRegistry::new();
470        let target = std::env::temp_dir().join("rocketmq-runtime-target-registry");
471
472        let first = registry.register(&target, Arc::<str>::from("first")).unwrap();
473        assert!(matches!(first, MetadataTargetRegistrationOutcome::Registered(_)));
474        assert!(matches!(
475            registry.register(&target, Arc::<str>::from("second")).unwrap(),
476            MetadataTargetRegistrationOutcome::Conflict
477        ));
478    }
479
480    #[test]
481    fn target_registry_capacity_is_bounded() {
482        let registry = MetadataTargetRegistry::with_max_entries(1);
483        let first = std::env::temp_dir().join("rocketmq-runtime-target-capacity-first");
484        let second = std::env::temp_dir().join("rocketmq-runtime-target-capacity-second");
485
486        assert!(matches!(
487            registry.register(&first, Arc::<str>::from("first")).unwrap(),
488            MetadataTargetRegistrationOutcome::Registered(_)
489        ));
490        let error = registry
491            .register(&second, Arc::<str>::from("second"))
492            .expect_err("target registration capacity must be bounded");
493        assert_eq!(error.condition(), rocketmq_error::CanonicalCondition::ResourceExhausted);
494    }
495
496    #[test]
497    fn registration_release_and_durable_generation_outlive_replacement() {
498        let registry = MetadataTargetRegistry::new();
499        let target = std::env::temp_dir().join("rocketmq-runtime-target-replacement");
500        let resource = Arc::<str>::from("resource");
501
502        let first = match registry.register(&target, resource.clone()).unwrap() {
503            MetadataTargetRegistrationOutcome::Registered(registration) => registration,
504            MetadataTargetRegistrationOutcome::Conflict => panic!("first registration must succeed"),
505            MetadataTargetRegistrationOutcome::ReconciliationRequired => {
506                panic!("first registration requires no reconciliation")
507            }
508        };
509        assert!(first.record_durable(MetadataGeneration::new(7)));
510        assert!(matches!(
511            registry.register(&target, resource.clone()).unwrap(),
512            MetadataTargetRegistrationOutcome::Conflict
513        ));
514
515        drop(first);
516        let replacement = match registry.register(&target, resource).unwrap() {
517            MetadataTargetRegistrationOutcome::Registered(registration) => registration,
518            MetadataTargetRegistrationOutcome::Conflict => panic!("replacement registration must succeed"),
519            MetadataTargetRegistrationOutcome::ReconciliationRequired => {
520                panic!("replacement registration requires no reconciliation")
521            }
522        };
523        assert_eq!(replacement.durable_generation(), Some(MetadataGeneration::new(7)));
524    }
525
526    #[test]
527    fn registration_owner_is_released_only_after_every_clone_is_dropped() {
528        let registry = MetadataTargetRegistry::new();
529        let target = std::env::temp_dir().join("rocketmq-runtime-target-clone");
530        let resource = Arc::<str>::from("resource");
531
532        let first = match registry.register(&target, resource.clone()).unwrap() {
533            MetadataTargetRegistrationOutcome::Registered(registration) => registration,
534            MetadataTargetRegistrationOutcome::Conflict => panic!("first registration must succeed"),
535            MetadataTargetRegistrationOutcome::ReconciliationRequired => {
536                panic!("first registration requires no reconciliation")
537            }
538        };
539        let cloned = first.clone();
540        drop(first);
541        assert!(matches!(
542            registry.register(&target, resource.clone()).unwrap(),
543            MetadataTargetRegistrationOutcome::Conflict
544        ));
545
546        drop(cloned);
547        assert!(matches!(
548            registry.register(&target, resource).unwrap(),
549            MetadataTargetRegistrationOutcome::Registered(_)
550        ));
551    }
552
553    #[test]
554    fn same_resource_can_switch_targets_after_the_writer_is_released() {
555        let registry = MetadataTargetRegistry::new();
556        let first_target = std::env::temp_dir().join("rocketmq-runtime-target-first");
557        let second_target = std::env::temp_dir().join("rocketmq-runtime-target-second");
558        let resource = Arc::<str>::from("resource");
559
560        let first = match registry.register(&first_target, resource.clone()).unwrap() {
561            MetadataTargetRegistrationOutcome::Registered(registration) => registration,
562            MetadataTargetRegistrationOutcome::Conflict => panic!("first registration must succeed"),
563            MetadataTargetRegistrationOutcome::ReconciliationRequired => {
564                panic!("first registration requires no reconciliation")
565            }
566        };
567        drop(first);
568
569        assert!(matches!(
570            registry.register(&second_target, resource).unwrap(),
571            MetadataTargetRegistrationOutcome::Registered(_)
572        ));
573    }
574
575    #[test]
576    fn unknown_commit_outcome_blocks_re_registration() {
577        let registry = MetadataTargetRegistry::new();
578        let target = std::env::temp_dir().join("rocketmq-runtime-target-reconcile");
579        let resource = Arc::<str>::from("resource");
580
581        let registration = match registry.register(&target, resource.clone()).unwrap() {
582            MetadataTargetRegistrationOutcome::Registered(registration) => registration,
583            MetadataTargetRegistrationOutcome::Conflict => panic!("registration must succeed"),
584            MetadataTargetRegistrationOutcome::ReconciliationRequired => {
585                panic!("a fresh target must not require reconciliation")
586            }
587        };
588        assert!(registration.record_reconciliation_required());
589        drop(registration);
590        assert!(matches!(
591            registry.register(&target, resource).unwrap(),
592            MetadataTargetRegistrationOutcome::ReconciliationRequired
593        ));
594    }
595
596    #[test]
597    fn retirement_checks_live_authority_and_history_before_reusing_capacity() {
598        let registry = MetadataTargetRegistry::with_max_entries(1);
599        let path = std::env::temp_dir().join("retirement-history");
600        let resource = Arc::<str>::from("resource");
601        let MetadataTargetRegistrationOutcome::Registered(first) = registry.register(&path, resource.clone()).unwrap()
602        else {
603            panic!("free target");
604        };
605        let identity = first.identity();
606        let target = first.inner.target.clone();
607        let old_owner = first.inner.owner;
608        let closure = first.clone();
609        assert!(first.record_durable(MetadataGeneration::new(1)));
610        drop(first);
611        assert_eq!(
612            registry.retire(&path, &resource, &identity, None).unwrap(),
613            MetadataTargetRetirementOutcome::HistoryChanged
614        );
615        assert_eq!(
616            registry
617                .retire(&path, &resource, &identity, Some(MetadataGeneration::new(1)))
618                .unwrap(),
619            MetadataTargetRetirementOutcome::WorkInProgress
620        );
621        drop(closure);
622        assert_eq!(
623            registry
624                .retire(&path, &resource, &identity, Some(MetadataGeneration::new(1)))
625                .unwrap(),
626            MetadataTargetRetirementOutcome::Retired
627        );
628        let MetadataTargetRegistrationOutcome::Registered(next) = registry.register(&path, resource.clone()).unwrap()
629        else {
630            panic!("retired capacity");
631        };
632        assert_ne!(identity, next.identity());
633        assert!(!registry.record_durable(&target, &resource, old_owner, MetadataGeneration::new(99)));
634        assert!(!registry.record_reconciliation_required(&target, &resource, old_owner));
635        assert_eq!(next.durable_generation(), None);
636        assert!(!next.reconciliation_required());
637        assert_eq!(registry.stats().retained_targets, 1);
638    }
639
640    #[cfg(windows)]
641    #[test]
642    fn windows_target_identity_is_case_insensitive() {
643        let registry = MetadataTargetRegistry::new();
644        let lower = std::env::temp_dir().join("rocketmq-runtime-target-case");
645        let upper = std::env::temp_dir().join("ROCKETMQ-RUNTIME-TARGET-CASE");
646
647        assert!(matches!(
648            registry.register(&lower, Arc::<str>::from("first")).unwrap(),
649            MetadataTargetRegistrationOutcome::Registered(_)
650        ));
651        assert!(matches!(
652            registry.register(&upper, Arc::<str>::from("second")).unwrap(),
653            MetadataTargetRegistrationOutcome::Conflict
654        ));
655    }
656}
rocketmq_runtime/metadata_target.rs

rocketmq_runtime/
metadata_target.rs
