@@ -5,6 +5,7 @@ use async_lock::RwLock;
55use chrono:: NaiveDateTime ;
66use futures:: TryFutureExt ;
77use moka:: future:: Cache ;
8+ use moka:: ops:: compute:: { CompResult , Op } ;
89use once_cell:: sync:: OnceCell ;
910use tokio:: task:: JoinSet ;
1011use tokio:: time:: { Duration , Instant } ;
@@ -16,7 +17,8 @@ use crate::connection::config::DatabaseConfig;
1617use crate :: database:: DatabaseKind ;
1718use crate :: error:: Error ;
1819use crate :: metrics:: {
19- NAMESPACE_EMPTIED_ENTRY , NAMESPACE_EMPTIED_ENTRY_GAVE_UP , NAMESPACE_LOAD_LATENCY ,
20+ NAMESPACE_EMPTIED_ENTRY , NAMESPACE_EMPTIED_ENTRY_GAVE_UP , NAMESPACE_LEFT_EMPTY_ENTRY_DROPPED ,
21+ NAMESPACE_LOAD_LATENCY ,
2022} ;
2123use crate :: namespace:: { NamespaceBottomlessDbId , NamespaceBottomlessDbIdInit , NamespaceName } ;
2224use crate :: stats:: Stats ;
@@ -380,14 +382,17 @@ impl NamespaceStore {
380382 // waiting request: the counter is the signal, not the log.
381383 NAMESPACE_EMPTIED_ENTRY . increment ( 1 ) ;
382384 tracing:: debug!( %namespace, "namespace cache entry was emptied while in use, looking it up again" ) ;
385+ if self . inner . has_shutdown . load ( Ordering :: Relaxed ) {
386+ return Err ( Error :: NamespaceStoreShutdown ) ;
387+ }
388+ // If the emptied entry is still the cached one, nothing is going
389+ // to refill it (a fork that failed before storing its record
390+ // leaves one behind): drop it, so that the next lookup loads the
391+ // namespace. An evicted or destroyed entry is no longer cached,
392+ // and this does nothing.
393+ self . evict_if_still_empty ( & namespace, & entry) . await ;
383394 }
384395
385- // Every lookup found an emptied entry. That is also where an entry
386- // that stays cached but empty ends up (a fork that failed before
387- // storing its record leaves one behind): the namespace answers
388- // `NamespaceUnavailable` until the entry is evicted or the server
389- // restarts. Dropping such an entry so that the next lookup reloads the
390- // namespace is left to a follow-up.
391396 let error = self . emptied_entry_error ( & namespace) . await ;
392397 if matches ! ( error, Error :: NamespaceUnavailable ( _) ) {
393398 // Shutdown and a namespace that is really gone are not failures
@@ -398,6 +403,45 @@ impl NamespaceStore {
398403 Err ( error)
399404 }
400405
406+ /// Removes `entry` from the cache if it is still the cached entry for
407+ /// `namespace` and still empty, so that the next lookup loads the namespace
408+ /// afresh (with its own restore option). It never fills an entry itself.
409+ ///
410+ /// What the checks guarantee, under moka's per-key compute lock: a loaded
411+ /// entry is never removed, and neither is an entry whose lock is held at
412+ /// that moment. What they do not: `try_read` cannot see a `fork` or `reset`
413+ /// that holds the entry but has not locked it yet. That window is not a
414+ /// few instructions: `get_with` makes the entry visible and may then run
415+ /// the cache's housekeeping, awaiting eviction listeners, before it
416+ /// returns. A `fork` or `reset` whose fresh entry is dropped in that window
417+ /// fills an entry the cache no longer holds, which nothing shuts down.
418+ /// `Op::Remove` also removes whatever the cache holds when it runs, which
419+ /// moka does not tie to the entry the closure saw; that case is detected
420+ /// below. Closing both needs the entry state to live in the cache value
421+ /// rather than behind the lock.
422+ async fn evict_if_still_empty ( & self , namespace : & NamespaceName , entry : & NamespaceEntry ) {
423+ let result = self
424+ . inner
425+ . store
426+ . entry_by_ref ( namespace)
427+ . and_compute_with ( |cached| {
428+ let still_empty = cached. is_some_and ( |cached| {
429+ Arc :: ptr_eq ( cached. value ( ) , entry)
430+ && entry. try_read ( ) . is_some_and ( |lock| lock. is_none ( ) )
431+ } ) ;
432+ std:: future:: ready ( if still_empty { Op :: Remove } else { Op :: Nop } )
433+ } )
434+ . await ;
435+ if let CompResult :: Removed ( removed) = result {
436+ if Arc :: ptr_eq ( removed. value ( ) , entry) {
437+ tracing:: warn!( %namespace, "namespace cache entry was left empty, dropped it" ) ;
438+ NAMESPACE_LEFT_EMPTY_ENTRY_DROPPED . increment ( 1 ) ;
439+ } else {
440+ tracing:: error!( %namespace, "dropped a different cache entry than the empty one" ) ;
441+ }
442+ }
443+ }
444+
401445 /// The error for a namespace whose cache entry was emptied while it was
402446 /// being used. `NamespaceDoesntExist` only when the metastore agrees: a
403447 /// client must be able to trust it, and a namespace that still exists but
@@ -518,6 +562,12 @@ impl NamespaceStore {
518562 tracing:: debug!( "storing db config" ) ;
519563 handle. store ( db_config) . await ?;
520564 tracing:: debug!( "completed storing db config, loading namespace" ) ;
565+ // An empty entry left cached (for example by a fork that failed before
566+ // storing its record) would be returned as is, and the namespace would
567+ // not be loaded with `restore_option` (a dump, say).
568+ if let Some ( entry) = self . inner . store . get ( & namespace) . await {
569+ self . evict_if_still_empty ( & namespace, & entry) . await ;
570+ }
521571 self . load_created ( & namespace, handle, restore_option)
522572 . await ?;
523573
@@ -915,7 +965,7 @@ pub(crate) mod test {
915965 }
916966
917967 #[ tokio:: test]
918- async fn with_gives_up_with_unavailable_when_the_entry_stays_empty ( ) {
968+ async fn with_loads_a_namespace_whose_cached_entry_stays_empty ( ) {
919969 let tmp = tempdir ( ) . unwrap ( ) ;
920970 let store = make_store ( tmp. path ( ) ) . await ;
921971 let namespace = create ( & store, "ns" ) . await ;
@@ -925,18 +975,129 @@ pub(crate) mod test {
925975
926976 // The namespace is taken out but its emptied entry stays cached, as an
927977 // entry left behind by a fork that failed before storing its record:
928- // every lookup finds it again .
978+ // nothing else is going to refill it .
929979 lock. take ( ) . unwrap ( ) . shutdown ( false ) . await . unwrap ( ) ;
930980 drop ( lock) ;
931981
932- // The metastore still has the namespace: giving up is temporary, not
933- // "missing".
934- let error = request. await . unwrap ( ) . unwrap_err ( ) ;
935- assert ! ( matches!( error, Error :: NamespaceUnavailable ( _) ) , "{error:?}" ) ;
936- assert ! ( Arc :: ptr_eq(
937- & store. inner. store. get( & namespace) . await . unwrap( ) ,
938- & entry
939- ) ) ;
982+ assert_eq ! ( request. await . unwrap( ) . unwrap( ) , namespace) ;
983+ // The empty entry was dropped and a fresh one loaded; it was not filled.
984+ let cached = store. inner . store . get ( & namespace) . await . unwrap ( ) ;
985+ assert ! ( !Arc :: ptr_eq( & cached, & entry) ) ;
986+ assert ! ( cached. read( ) . await . is_some( ) ) ;
987+ assert ! ( entry. read( ) . await . is_none( ) ) ;
988+ }
989+
990+ #[ tokio:: test( flavor = "multi_thread" ) ]
991+ async fn requests_waiting_on_one_left_empty_entry_share_one_reloaded_namespace ( ) {
992+ let tmp = tempdir ( ) . unwrap ( ) ;
993+ let store = make_store ( tmp. path ( ) ) . await ;
994+ let namespace = create ( & store, "ns" ) . await ;
995+ let entry = store. inner . store . get ( & namespace) . await . unwrap ( ) ;
996+ let mut lock = entry. write ( ) . await ;
997+ settle ( & store, & entry) . await ;
998+ const REQUESTS : usize = 8 ;
999+ let requests: Vec < _ > = ( 0 ..REQUESTS )
1000+ . map ( |_| {
1001+ tokio:: spawn ( {
1002+ let store = store. clone ( ) ;
1003+ let namespace = namespace. clone ( ) ;
1004+ async move { store. with ( namespace, |ns| ns. stats . clone ( ) ) . await }
1005+ } )
1006+ } )
1007+ . collect ( ) ;
1008+ wait_until ( "every request holds the entry" , || {
1009+ Arc :: strong_count ( & entry) >= 2 + REQUESTS
1010+ } )
1011+ . await ;
1012+
1013+ // The namespace is taken out and its emptied entry stays cached: every
1014+ // request finds it, and every one of them may try to drop it.
1015+ lock. take ( ) . unwrap ( ) . shutdown ( false ) . await . unwrap ( ) ;
1016+ drop ( lock) ;
1017+
1018+ let mut stats = Vec :: with_capacity ( REQUESTS ) ;
1019+ for request in requests {
1020+ stats. push ( request. await . unwrap ( ) . unwrap ( ) ) ;
1021+ }
1022+ // One namespace, loaded once, into one fresh entry.
1023+ let cached = store. inner . store . get ( & namespace) . await . unwrap ( ) ;
1024+ assert ! ( !Arc :: ptr_eq( & cached, & entry) ) ;
1025+ let loaded = cached. read ( ) . await . as_ref ( ) . unwrap ( ) . stats . clone ( ) ;
1026+ assert ! ( stats. iter( ) . all( |stats| Arc :: ptr_eq( stats, & loaded) ) ) ;
1027+ assert ! ( entry. read( ) . await . is_none( ) ) ;
1028+ }
1029+
1030+ #[ tokio:: test]
1031+ async fn evict_if_still_empty_drops_only_the_cached_unlocked_empty_entry ( ) {
1032+ let tmp = tempdir ( ) . unwrap ( ) ;
1033+ let store = make_store ( tmp. path ( ) ) . await ;
1034+ let namespace = create ( & store, "ns" ) . await ;
1035+ let cached = |store : & NamespaceStore | {
1036+ let store = store. clone ( ) ;
1037+ let namespace = namespace. clone ( ) ;
1038+ async move { store. inner . store . get ( & namespace) . await }
1039+ } ;
1040+
1041+ // A loaded cached entry is not empty: it stays.
1042+ let loaded = cached ( & store) . await . unwrap ( ) ;
1043+ store. evict_if_still_empty ( & namespace, & loaded) . await ;
1044+ assert ! ( Arc :: ptr_eq( & cached( & store) . await . unwrap( ) , & loaded) ) ;
1045+
1046+ // An empty entry that is not the cached one leaves the cached empty
1047+ // entry alone.
1048+ let empty: NamespaceEntry = Default :: default ( ) ;
1049+ store
1050+ . inner
1051+ . store
1052+ . insert ( namespace. clone ( ) , empty. clone ( ) )
1053+ . await ;
1054+ let other: NamespaceEntry = Default :: default ( ) ;
1055+ store. evict_if_still_empty ( & namespace, & other) . await ;
1056+ assert ! ( Arc :: ptr_eq( & cached( & store) . await . unwrap( ) , & empty) ) ;
1057+
1058+ // An empty cached entry whose lock is held (it may be being filled)
1059+ // stays. Evicting it would wait for that lock, so bound the call.
1060+ let lock = empty. write ( ) . await ;
1061+ tokio:: time:: timeout (
1062+ Duration :: from_secs ( 10 ) ,
1063+ store. evict_if_still_empty ( & namespace, & empty) ,
1064+ )
1065+ . await
1066+ . expect ( "evicting a locked entry must not wait for its lock" ) ;
1067+ assert ! ( Arc :: ptr_eq( & cached( & store) . await . unwrap( ) , & empty) ) ;
1068+ drop ( lock) ;
1069+
1070+ // Once nobody holds it, the empty cached entry is dropped.
1071+ store. evict_if_still_empty ( & namespace, & empty) . await ;
1072+ assert ! ( !store. inner. store. contains_key( & namespace) ) ;
1073+ }
1074+
1075+ #[ tokio:: test]
1076+ async fn create_loads_a_namespace_over_an_empty_cached_entry ( ) {
1077+ let tmp = tempdir ( ) . unwrap ( ) ;
1078+ let store = make_store ( tmp. path ( ) ) . await ;
1079+ let namespace = NamespaceName :: from_string ( "ns" . to_string ( ) ) . unwrap ( ) ;
1080+ // What a fork that failed before storing its record leaves behind.
1081+ let empty: NamespaceEntry = Default :: default ( ) ;
1082+ store
1083+ . inner
1084+ . store
1085+ . insert ( namespace. clone ( ) , empty. clone ( ) )
1086+ . await ;
1087+
1088+ store
1089+ . create (
1090+ namespace. clone ( ) ,
1091+ RestoreOption :: Latest ,
1092+ DatabaseConfig :: default ( ) ,
1093+ )
1094+ . await
1095+ . unwrap ( ) ;
1096+
1097+ // Loaded by create itself, with its restore option.
1098+ let cached = store. inner . store . get ( & namespace) . await . unwrap ( ) ;
1099+ assert ! ( !Arc :: ptr_eq( & cached, & empty) ) ;
1100+ assert ! ( cached. read( ) . await . is_some( ) ) ;
9401101 }
9411102
9421103 #[ tokio:: test]
@@ -1000,6 +1161,44 @@ pub(crate) mod test {
10001161 assert ! ( !store. inner. metadata. exists( & namespace) . await ) ;
10011162 }
10021163
1164+ // `fork`'s cleanup on failure calls `block_in_place`, which panics on
1165+ // the current-thread runtime and would hide the real error.
1166+ #[ tokio:: test( flavor = "multi_thread" ) ]
1167+ async fn fork_loads_a_source_whose_cached_entry_was_left_empty ( ) {
1168+ let tmp = tempdir ( ) . unwrap ( ) ;
1169+ let store = make_store ( tmp. path ( ) ) . await ;
1170+ let from = create ( & store, "from" ) . await ;
1171+ let entry = store. inner . store . get ( & from) . await . unwrap ( ) ;
1172+ entry
1173+ . write ( )
1174+ . await
1175+ . take ( )
1176+ . unwrap ( )
1177+ . shutdown ( false )
1178+ . await
1179+ . unwrap ( ) ;
1180+ let to = NamespaceName :: from_string ( "to" . to_string ( ) ) . unwrap ( ) ;
1181+
1182+ store
1183+ . fork ( from. clone ( ) , to. clone ( ) , DatabaseConfig :: default ( ) , None )
1184+ . await
1185+ . unwrap ( ) ;
1186+
1187+ assert_eq ! (
1188+ store
1189+ . with( to. clone( ) , |ns| ns. name( ) . clone( ) )
1190+ . await
1191+ . unwrap( ) ,
1192+ to
1193+ ) ;
1194+ // The source was reloaded into a fresh entry; the left-empty one was
1195+ // not filled.
1196+ let cached = store. inner . store . get ( & from) . await . unwrap ( ) ;
1197+ assert ! ( !Arc :: ptr_eq( & cached, & entry) ) ;
1198+ assert ! ( cached. read( ) . await . is_some( ) ) ;
1199+ assert ! ( entry. read( ) . await . is_none( ) ) ;
1200+ }
1201+
10031202 #[ tokio:: test]
10041203 async fn with_reports_shutdown_when_the_store_shuts_down_while_waiting ( ) {
10051204 let tmp = tempdir ( ) . unwrap ( ) ;
0 commit comments