bug: rejected overwrite on flush buffer overflow exposes stale disk value after drain

[package]
name = "hybrid-foyer-stale-repro"
version = "0.1.0"
edition = "2024"

[dependencies]
foyer-storage = { git = "https://github.com/foyer-rs/foyer", rev = "0fb762837b38a50681eff9c37c5f1f8b1893af8b", features = ["test_utils"] }
foyer = { git = "https://github.com/foyer-rs/foyer", rev = "0fb762837b38a50681eff9c37c5f1f8b1893af8b", features = ["test_utils"] }
tokio = { version = "1", features = ["full"] }
serde_json = "1"
prometheus = "0.14"
mixtrics = { version = "0.2", features = ["prometheus"] }
libc = "0.2"

[profile.release]
debug = 1

use foyer::*;
use foyer_storage::test_utils::Switch;
use mixtrics::registry::prometheus::PrometheusMetricsRegistry;
use serde_json::{json,Value};
use std::time::{Duration,Instant};
type C=HybridCache<u128,Vec<u8>>;
const SIZE:usize=64*1024;
fn value(key:u64,version:u64)->Vec<u8>{let mut v=vec![7;SIZE];v[..8].copy_from_slice(&key.to_le_bytes());v[8..16].copy_from_slice(&version.to_le_bytes());v}
fn counts(r:&prometheus::Registry)->(u64,u64,u64){
 let(mut ser,mut buf,mut chan)=(0,0,0);
 for f in r.gather(){for m in f.get_metric(){
  if f.name().contains("entry_serde_duration") && m.get_label().iter().any(|l|l.value()=="serialize"){ser+=m.get_histogram().get_sample_count();}
  if f.name()=="foyer_storage_inner_op_total"{for l in m.get_label(){match l.value(){"buffer_overflow"=>buf+=m.get_counter().value() as u64,"channel_overflow"=>chan+=m.get_counter().value() as u64,_=>{}}}}
 }}(ser,buf,chan)
}
async fn processed(r:&prometheus::Registry,n:u64){let start=Instant::now();loop{let(a,b,c)=counts(r);if a+b+c>=n{return}assert!(start.elapsed()<Duration::from_secs(5),"receiver did not reach {n}: {:?}",counts(r));tokio::time::sleep(Duration::from_millis(1)).await;}}
async fn disk(c:&C)->Value {match c.storage().load(&1u128).await.unwrap(){Load::Entry{value,..}=>json!({"source":"disk","key":u64::from_le_bytes(value[..8].try_into().unwrap()),"version":u64::from_le_bytes(value[8..16].try_into().unwrap())}),Load::Piece{piece,..}=>json!({"source":"keeper","version":u64::from_le_bytes(piece.value()[8..16].try_into().unwrap())}),Load::Miss=>json!({"source":"miss"}),Load::Throttled=>panic!("unexpected throttling")}}
async fn case(root:&str,name:&str,buffer:usize,invalidate:bool){
 let dir=format!("{root}/{name}");std::fs::create_dir_all(&dir).unwrap();let registry=prometheus::Registry::new();let sw=Switch::default();
 let device=FsDeviceBuilder::new(&dir).with_capacity(64<<20).with_direct(true).build().unwrap();
 let c:C=HybridCacheBuilder::new().with_name(name.to_owned()).with_policy(HybridCachePolicy::WriteOnEviction)
  .with_metrics_registry(Box::new(PrometheusMetricsRegistry::new(registry.clone())))
  .memory(SIZE+16).with_shards(1).with_eviction_config(LruConfig{high_priority_pool_ratio:0.0}).with_weighter(|_:&u128,v:&Vec<u8>|v.len()+16)
  .storage().with_io_engine_config(PsyncIoEngineConfig::new())
  .with_engine_config(BlockEngineConfig::new(device).with_block_size(1<<20).with_buffer_pool_size(buffer).with_flushers(1).with_reclaimers(1).with_clean_block_threshold(1).with_submit_queue_size_threshold(16<<20).with_flush_switch(sw.clone()).with_eviction_pickers(vec![Box::new(FifoPicker::new(0.0))]))
  .with_recover_mode(RecoverMode::None).build().await.unwrap();
 let mut fds=0;
 for e in std::fs::read_dir("/proc/self/fd").unwrap().flatten(){if let Ok(p)=std::fs::read_link(e.path()){if p.to_string_lossy().starts_with(&dir){let fd=e.file_name().to_string_lossy().parse::<i32>().unwrap();assert_ne!(unsafe{libc::fcntl(fd,libc::F_GETFL)}&libc::O_DIRECT,0);fds+=1;}}}assert!(fds>0);
 drop(c.insert(1,value(1,0)));drop(c.insert(2,value(2,0)));c.storage().wait().await;
 let initial=disk(&c).await;assert_eq!(initial["version"],0);assert!(c.memory().get(&1u128).is_none());
 let before=counts(&registry);let start=before.0+before.1+before.2;
 sw.on();
 drop(c.insert(3,value(3,0)));processed(&registry,start+1).await; // Evict B and fill the held buffer.
 if invalidate{c.storage().delete(&1u128);}
 drop(c.insert(1,value(1,1)));processed(&registry,start+2).await; // Evict C; A's RAM value is now v1.
 assert_eq!(u64::from_le_bytes(c.memory().get(&1u128).unwrap().value()[8..16].try_into().unwrap()),1);
 drop(c.insert(4,value(4,0)));processed(&registry,start+3).await; // Evict A v1 into a full buffer.
 assert!(c.memory().get(&1u128).is_none());
 // Resume and drain all background work before observing: no pending-write timing ambiguity.
 sw.off();c.storage().wait().await;
 let stored=disk(&c).await;
 let observed=c.get(&1u128).await.unwrap().map(|e|u64::from_le_bytes(e.value()[8..16].try_into().unwrap()));
 let after=counts(&registry);
 if buffer<2*(SIZE+4096){assert_eq!(after.1-before.1,2);assert_eq!(observed,if invalidate{None}else{Some(0)});}else{assert_eq!(after.1-before.1,0);assert_eq!(observed,Some(1));}
 println!("{}",json!({"case":name,"direct_fds_verified":fds,"object_bytes":SIZE,"memory_weight_capacity":SIZE+16,"buffer_bytes":buffer,"initial":initial,"latest_inserted_version":1,"after_drain_storage":stored,"hybrid_get_version":observed,"buffer_overflow_delta":after.1-before.1,"channel_overflow_delta":after.2-before.2,"old_index_explicitly_invalidated":invalidate,"all_assertions_passed":true}));
 c.close().await.unwrap();
}
fn main(){let root=std::env::args().nth(1).unwrap();let rt=tokio::runtime::Builder::new_multi_thread().worker_threads(2).enable_all().build().unwrap();rt.block_on(async{
 for repeat in 0..3 {
  case(&root,&format!("overflow-{repeat}"),SIZE+4096,false).await;
  case(&root,&format!("no-overflow-{repeat}"),512<<10,false).await;
  case(&root,&format!("invalidate-old-{repeat}"),SIZE+4096,true).await;
 }
});}