@@ -2,8 +2,6 @@ mod fork;
2
2
pub mod meta_store;
3
3
pub mod replication_wal;
4
4
5
- use std:: collections:: hash_map:: Entry ;
6
- use std:: collections:: HashMap ;
7
5
use std:: fmt;
8
6
use std:: path:: { Path , PathBuf } ;
9
7
use std:: sync:: atomic:: { AtomicBool , Ordering } ;
@@ -175,8 +173,6 @@ pub trait MakeNamespace: Sync + Send + 'static {
175
173
timestamp : Option < NaiveDateTime > ,
176
174
meta_store : & MetaStore ,
177
175
) -> crate :: Result < Namespace < Self :: Database > > ;
178
-
179
- async fn exists ( & self , namespace : & NamespaceName ) -> bool ;
180
176
}
181
177
182
178
/// Creates new primary `Namespace`
@@ -292,34 +288,6 @@ impl MakeNamespace for PrimaryNamespaceMaker {
292
288
let ns = fork_task. fork ( ) . await ?;
293
289
Ok ( ns)
294
290
}
295
-
296
- async fn exists ( & self , namespace : & NamespaceName ) -> bool {
297
- let ns_path = self . config . base_path . join ( "dbs" ) . join ( namespace. as_str ( ) ) ;
298
- if let Ok ( true ) = ns_path. try_exists ( ) {
299
- return true ;
300
- }
301
-
302
- if let Some ( replication_options) = self . config . bottomless_replication . as_ref ( ) {
303
- tracing:: info!( "Bottomless: {:?}" , replication_options) ;
304
- match bottomless:: replicator:: Replicator :: has_backup_of ( namespace, replication_options)
305
- . await
306
- {
307
- Ok ( true ) => {
308
- tracing:: debug!( "Bottomless: Backup found" ) ;
309
- return true ;
310
- }
311
- Ok ( false ) => {
312
- tracing:: debug!( "Bottomless: No backup found" ) ;
313
- }
314
- Err ( err) => {
315
- tracing:: debug!( "Bottomless: Error checking backup: {}" , err) ;
316
- }
317
- }
318
- } else {
319
- tracing:: debug!( "Bottomless: No backup configured" ) ;
320
- }
321
- false
322
- }
323
291
}
324
292
325
293
/// Creates new replica `Namespace`
@@ -375,11 +343,6 @@ impl MakeNamespace for ReplicaNamespaceMaker {
375
343
) -> crate :: Result < Namespace < Self :: Database > > {
376
344
return Err ( ForkError :: ForkReplica . into ( ) ) ;
377
345
}
378
-
379
- async fn exists ( & self , namespace : & NamespaceName ) -> bool {
380
- let ns_path = self . config . base_path . join ( "dbs" ) . join ( namespace. as_str ( ) ) ;
381
- ns_path. try_exists ( ) . unwrap_or ( false )
382
- }
383
346
}
384
347
385
348
type NamespaceEntry < T > = Arc < RwLock < Option < Namespace < T > > > > ;
@@ -408,13 +371,14 @@ struct NamespaceStoreInner<M: MakeNamespace> {
408
371
}
409
372
410
373
impl < M : MakeNamespace > NamespaceStore < M > {
411
- pub fn new (
374
+ pub async fn new (
412
375
make_namespace : M ,
413
376
allow_lazy_creation : bool ,
414
377
snapshot_at_shutdown : bool ,
415
378
meta_store_path : impl AsRef < Path > ,
416
379
max_active_namespaces : usize ,
417
- ) -> Self {
380
+ ) -> crate :: Result < Self > {
381
+ let metadata = MetaStore :: new ( meta_store_path) . await ?;
418
382
tracing:: trace!( "Max active namespaces: {max_active_namespaces}" ) ;
419
383
let store = Cache :: < NamespaceName , NamespaceEntry < M :: Database > > :: builder ( )
420
384
. async_eviction_listener ( move |name, ns, cause| {
@@ -437,7 +401,7 @@ impl<M: MakeNamespace> NamespaceStore<M> {
437
401
. max_capacity ( max_active_namespaces as u64 )
438
402
. time_to_idle ( Duration :: from_secs ( 86400 ) )
439
403
. build ( ) ;
440
- Self {
404
+ Ok ( Self {
441
405
inner : Arc :: new ( NamespaceStoreInner {
442
406
store,
443
407
metadata,
@@ -446,7 +410,7 @@ impl<M: MakeNamespace> NamespaceStore<M> {
446
410
has_shutdown : AtomicBool :: new ( false ) ,
447
411
snapshot_at_shutdown,
448
412
} ) ,
449
- }
413
+ } )
450
414
}
451
415
452
416
pub async fn destroy ( & self , namespace : NamespaceName ) -> crate :: Result < ( ) > {
@@ -570,7 +534,7 @@ impl<M: MakeNamespace> NamespaceStore<M> {
570
534
}
571
535
572
536
// check that the source namespace exists
573
- if !self . inner . make_namespace . exists ( & from) . await {
537
+ if !self . inner . metadata . exists ( & from) {
574
538
return Err ( crate :: error:: Error :: NamespaceDoesntExist ( from. to_string ( ) ) ) ;
575
539
}
576
540
@@ -637,7 +601,7 @@ impl<M: MakeNamespace> NamespaceStore<M> {
637
601
let namespace = namespace. clone ( ) ;
638
602
async move {
639
603
if namespace != NamespaceName :: default ( )
640
- && !self . inner . make_namespace . exists ( & namespace) . await
604
+ && !self . inner . metadata . exists ( & namespace)
641
605
&& !self . inner . allow_lazy_creation
642
606
{
643
607
return Err ( Error :: NamespaceDoesntExist ( namespace. to_string ( ) ) ) ;
@@ -650,6 +614,7 @@ impl<M: MakeNamespace> NamespaceStore<M> {
650
614
RestoreOption :: Latest ,
651
615
NamespaceBottomlessDbId :: NotProvided ,
652
616
self . make_reset_cb ( ) ,
617
+ & self . inner . metadata ,
653
618
)
654
619
. await ?;
655
620
tracing:: info!( "loaded namespace: `{namespace}`" ) ;
@@ -707,7 +672,7 @@ impl<M: MakeNamespace> NamespaceStore<M> {
707
672
// otherwise it's an error.
708
673
if self . inner . allow_lazy_creation || namespace == NamespaceName :: default ( ) {
709
674
tracing:: trace!( "auto-creating the namespace" ) ;
710
- } else if self . inner . make_namespace . exists ( & namespace) . await {
675
+ } else if self . inner . metadata . exists ( & namespace) {
711
676
return Err ( Error :: NamespaceAlreadyExist ( namespace. to_string ( ) ) ) ;
712
677
}
713
678
@@ -722,6 +687,7 @@ impl<M: MakeNamespace> NamespaceStore<M> {
722
687
restore_option,
723
688
bottomless_db_id_for_init,
724
689
self . make_reset_cb ( ) ,
690
+ & self . inner . metadata ,
725
691
)
726
692
. await ;
727
693
match ns {
0 commit comments