Skip to content

fix(storage): TTL millisecond fidelity — entry expiry stored as ms, fossil base_ts plumbing removed (W3) #398

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pilotspacex-byte merged 1 commit into from
Jul 20, 2026
Merged

fix(storage): TTL millisecond fidelity — entry expiry stored as ms, fossil base_ts plumbing removed (W3) #398

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

## [Unreleased]

### Fixed
- **Millisecond-TTL keys no longer expire up to 999 ms early (W3).**
`CompactEntry` stored expiry as *seconds* (`ms / 1000` floor) even though
the field was a full `u64`: a `PEXPIRE key 1500` died at the 1000 ms
boundary, and `PTTL` reported the truncated value. Expiry is now stored
as absolute Unix **milliseconds** end-to-end (same 32-byte entry layout),
`PTTL` reads back the exact deadline, and RDB/snapshot round-trips
preserve millisecond fidelity (the round-trip test's old 5-second
tolerance is now an exact equality).

### Changed
- **Fossil `base_ts` plumbing removed (W3).** `is_expired_at` /
`expires_at_ms` / `set_expires_at_ms` / `new_string_with_expiry` carried a
`base_ts: u32` parameter that has been ignored since expiry became
absolute; the parameter, ~60 dead call-site arguments, dead
`let base_ts = …` bindings, the never-read
`SnapshotState.base_timestamps` field, the dead `u32` element in the
AOF-rewrite/BGSAVE snapshot tuple (`AofFoldSnapshot.dbs` and the
`save_from_snapshot`/`save_snapshot_to_bytes` signatures), and the
caller-less `merge_shard_snapshots` are gone.

### Performance
- **Sync eviction spill now batches like the async path (W2).** The
tick-driven sync spill paths (`run_eviction`, the memory-pressure cascade's
Expand Down
30 changes: 10 additions & 20 deletions src/command/key.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,15 +184,14 @@ pub fn ttl(db: &mut Database, args: &[Frame]) -> Frame {
Some(k) => k,
None => return err_wrong_args("TTL"),
};
let base_ts = db.base_timestamp();
match db.get(key) {
None => Frame::Integer(-2),
Some(entry) => {
if !entry.has_expiry() {
Frame::Integer(-1)
} else {
let now_ms = current_time_ms();
let exp_ms = entry.expires_at_ms(base_ts);
let exp_ms = entry.expires_at_ms();
if now_ms >= exp_ms {
// Edge case: expired between get and now
Frame::Integer(-2)
Expand All @@ -215,15 +214,14 @@ pub fn pttl(db: &mut Database, args: &[Frame]) -> Frame {
Some(k) => k,
None => return err_wrong_args("PTTL"),
};
let base_ts = db.base_timestamp();
match db.get(key) {
None => Frame::Integer(-2),
Some(entry) => {
if !entry.has_expiry() {
Frame::Integer(-1)
} else {
let now_ms = current_time_ms();
let exp_ms = entry.expires_at_ms(base_ts);
let exp_ms = entry.expires_at_ms();
if now_ms >= exp_ms {
Frame::Integer(-2)
} else {
Expand Down Expand Up @@ -362,14 +360,13 @@ pub fn expiretime(db: &mut Database, args: &[Frame]) -> Frame {
Some(k) => k,
None => return err_wrong_args("EXPIRETIME"),
};
let base_ts = db.base_timestamp();
match db.get(key) {
None => Frame::Integer(-2),
Some(entry) => {
if !entry.has_expiry() {
Frame::Integer(-1)
} else {
Frame::Integer((entry.expires_at_ms(base_ts) / 1000) as i64)
Frame::Integer((entry.expires_at_ms() / 1000) as i64)
}
}
}
Expand All @@ -386,14 +383,13 @@ pub fn pexpiretime(db: &mut Database, args: &[Frame]) -> Frame {
Some(k) => k,
None => return err_wrong_args("PEXPIRETIME"),
};
let base_ts = db.base_timestamp();
match db.get(key) {
None => Frame::Integer(-2),
Some(entry) => {
if !entry.has_expiry() {
Frame::Integer(-1)
} else {
Frame::Integer(entry.expires_at_ms(base_ts) as i64)
Frame::Integer(entry.expires_at_ms() as i64)
}
}
}
Expand Down Expand Up @@ -1211,15 +1207,14 @@ pub fn ttl_readonly(db: &Database, args: &[Frame], now_ms: u64) -> Frame {
Some(k) => k,
None => return err_wrong_args("TTL"),
};
let base_ts = db.base_timestamp();
match db.get_if_alive(key, now_ms) {
None => Frame::Integer(-2),
Some(entry) => {
if !entry.has_expiry() {
Frame::Integer(-1)
} else {
let now = current_time_ms();
let exp_ms = entry.expires_at_ms(base_ts);
let exp_ms = entry.expires_at_ms();
if now >= exp_ms {
Frame::Integer(-2)
} else {
Expand All @@ -1239,15 +1234,14 @@ pub fn pttl_readonly(db: &Database, args: &[Frame], now_ms: u64) -> Frame {
Some(k) => k,
None => return err_wrong_args("PTTL"),
};
let base_ts = db.base_timestamp();
match db.get_if_alive(key, now_ms) {
None => Frame::Integer(-2),
Some(entry) => {
if !entry.has_expiry() {
Frame::Integer(-1)
} else {
let now = current_time_ms();
let exp_ms = entry.expires_at_ms(base_ts);
let exp_ms = entry.expires_at_ms();
if now >= exp_ms {
Frame::Integer(-2)
} else {
Expand Down Expand Up @@ -1381,14 +1375,13 @@ pub fn expiretime_readonly(db: &Database, args: &[Frame], now_ms: u64) -> Frame
Some(k) => k,
None => return err_wrong_args("EXPIRETIME"),
};
let base_ts = db.base_timestamp();
match db.get_if_alive(key, now_ms) {
None => Frame::Integer(-2),
Some(entry) => {
if !entry.has_expiry() {
Frame::Integer(-1)
} else {
Frame::Integer((entry.expires_at_ms(base_ts) / 1000) as i64)
Frame::Integer((entry.expires_at_ms() / 1000) as i64)
}
}
}
Expand All @@ -1405,14 +1398,13 @@ pub fn pexpiretime_readonly(db: &Database, args: &[Frame], now_ms: u64) -> Frame
Some(k) => k,
None => return err_wrong_args("PEXPIRETIME"),
};
let base_ts = db.base_timestamp();
match db.get_if_alive(key, now_ms) {
None => Frame::Integer(-2),
Some(entry) => {
if !entry.has_expiry() {
Frame::Integer(-1)
} else {
Frame::Integer(entry.expires_at_ms(base_ts) as i64)
Frame::Integer(entry.expires_at_ms() as i64)
}
}
}
Expand Down Expand Up @@ -1469,10 +1461,9 @@ mod tests {

fn setup_db_with_expiry(key: &[u8], val: &[u8], expires_at_ms: u64) -> Database {
let mut db = Database::new();
let base_ts = db.base_timestamp();
db.set(
Bytes::copy_from_slice(key),
Entry::new_string_with_expiry(Bytes::copy_from_slice(val), expires_at_ms, base_ts),
Entry::new_string_with_expiry(Bytes::copy_from_slice(val), expires_at_ms),
);
db
}
Expand Down Expand Up @@ -2145,10 +2136,9 @@ mod tests {
Entry::new_string(Bytes::from_static(b"1")),
);
let past_ms = current_time_ms() - 1000;
let base_ts = db.base_timestamp();
db.set(
Bytes::from_static(b"dead"),
Entry::new_string_with_expiry(Bytes::from_static(b"2"), past_ms, base_ts),
Entry::new_string_with_expiry(Bytes::from_static(b"2"), past_ms),
);
let result = keys(&mut db, &[bs(b"*")]);
match result {
Expand Down
23 changes: 8 additions & 15 deletions src/command/persistence.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,24 +76,20 @@ pub fn bgsave_start(db: SharedDatabases, dir: String, dbfilename: String) -> Fra
}

// Clone snapshot: lock each db individually with read lock
// Include base_timestamp for TTL delta resolution during serialization
let snapshot: Vec<(
let snapshot: Vec<
Vec<(
crate::storage::compact_key::CompactKey,
crate::storage::entry::Entry,
)>,
u32,
)> = db
> = db
.iter()
.map(|lock| {
let guard = lock.read();
let base_ts = guard.base_timestamp();
let entries = guard
guard
.data()
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
(entries, base_ts)
.collect()
})
.collect();

Expand Down Expand Up @@ -353,23 +349,20 @@ pub fn handle_save(db: &SharedDatabases, dir: &str, dbfilename: &str) -> Frame {
}

// Clone snapshot: lock each db individually with read lock (same pattern as bgsave_start)
let snapshot: Vec<(
let snapshot: Vec<
Vec<(
crate::storage::compact_key::CompactKey,
crate::storage::entry::Entry,
)>,
u32,
)> = db
> = db
.iter()
.map(|lock| {
let guard = lock.read();
let base_ts = guard.base_timestamp();
let entries = guard
guard
.data()
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
(entries, base_ts)
.collect()
})
.collect();

Expand Down
15 changes: 5 additions & 10 deletions src/command/string/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -447,10 +447,9 @@ mod tests {
fn test_incr_preserves_ttl() {
let mut db = make_db();
let exp_ms = current_time_ms() + 100_000;
let base_ts = db.base_timestamp();
db.set(
Bytes::from_static(b"counter"),
Entry::new_string_with_expiry(Bytes::from_static(b"5"), exp_ms, base_ts),
Entry::new_string_with_expiry(Bytes::from_static(b"5"), exp_ms),
);
let result = incr(&mut db, &[bs(b"counter")]);
assert_eq!(result, Frame::Integer(6));
Expand Down Expand Up @@ -539,10 +538,9 @@ mod tests {
fn test_append_preserves_ttl() {
let mut db = make_db();
let exp_ms = current_time_ms() + 100_000;
let base_ts = db.base_timestamp();
db.set(
Bytes::from_static(b"key"),
Entry::new_string_with_expiry(Bytes::from_static(b"hello"), exp_ms, base_ts),
Entry::new_string_with_expiry(Bytes::from_static(b"hello"), exp_ms),
);
append(&mut db, &[bs(b"key"), bs(b" world")]);
let entry = db.get(b"key").unwrap();
Expand Down Expand Up @@ -659,10 +657,9 @@ mod tests {
fn test_getset_removes_ttl() {
let mut db = make_db();
let exp_ms = current_time_ms() + 100_000;
let base_ts = db.base_timestamp();
db.set(
Bytes::from_static(b"key"),
Entry::new_string_with_expiry(Bytes::from_static(b"old"), exp_ms, base_ts),
Entry::new_string_with_expiry(Bytes::from_static(b"old"), exp_ms),
);
getset(&mut db, &[bs(b"key"), bs(b"new")]);
let entry = db.get(b"key").unwrap();
Expand Down Expand Up @@ -762,10 +759,9 @@ mod tests {
fn test_getex_persist() {
let mut db = make_db();
let exp_ms = current_time_ms() + 100_000;
let base_ts = db.base_timestamp();
db.set(
Bytes::from_static(b"key"),
Entry::new_string_with_expiry(Bytes::from_static(b"val"), exp_ms, base_ts),
Entry::new_string_with_expiry(Bytes::from_static(b"val"), exp_ms),
);
let result = getex(&mut db, &[bs(b"key"), bs(b"PERSIST")]);
assert_eq!(result, Frame::BulkString(Bytes::from_static(b"val")));
Expand Down Expand Up @@ -942,10 +938,9 @@ mod tests {
fn test_setrange_preserves_ttl() {
let mut db = make_db();
let exp_ms = current_time_ms() + 100_000;
let base_ts = db.base_timestamp();
db.set(
Bytes::from_static(b"key"),
Entry::new_string_with_expiry(Bytes::from_static(b"Hello"), exp_ms, base_ts),
Entry::new_string_with_expiry(Bytes::from_static(b"Hello"), exp_ms),
);
setrange(&mut db, &[bs(b"key"), bs(b"0"), bs(b"Jello")]);
let entry = db.get(b"key").unwrap();
Expand Down
10 changes: 4 additions & 6 deletions src/command/string/string_bit.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,10 +118,9 @@ pub fn setbit(db: &mut Database, args: &[Frame]) -> Frame {
let byte_idx = offset / 8;
let bit_idx = 7 - (offset % 8);

let base_ts = db.base_timestamp();
let (existing_data, existing_expiry_ms) = match db.get(&key) {
Some(entry) => {
let expiry = entry.expires_at_ms(base_ts);
let expiry = entry.expires_at_ms();
match entry.value.as_bytes() {
Some(v) => (Some(v.to_vec()), expiry),
None => {
Expand Down Expand Up @@ -152,7 +151,7 @@ pub fn setbit(db: &mut Database, args: &[Frame]) -> Frame {

let new_val = Bytes::from(buf);
let mut entry = if existing_expiry_ms > 0 {
Entry::new_string_with_expiry(new_val, existing_expiry_ms, base_ts)
Entry::new_string_with_expiry(new_val, existing_expiry_ms)
} else {
Entry::new_string(new_val)
};
Expand Down Expand Up @@ -783,10 +782,9 @@ pub fn bitfield(db: &mut Database, args: &[Frame]) -> Frame {
None => return err_wrong_args("BITFIELD"),
};

let base_ts = db.base_timestamp();
let (existing_data, existing_expiry_ms) = match db.get(&key) {
Some(entry) => {
let expiry = entry.expires_at_ms(base_ts);
let expiry = entry.expires_at_ms();
match entry.value.as_bytes() {
Some(v) => (v.to_vec(), expiry),
None => {
Expand Down Expand Up @@ -914,7 +912,7 @@ pub fn bitfield(db: &mut Database, args: &[Frame]) -> Frame {
if modified {
let new_val = Bytes::from(buf);
let mut entry = if existing_expiry_ms > 0 {
Entry::new_string_with_expiry(new_val, existing_expiry_ms, base_ts)
Entry::new_string_with_expiry(new_val, existing_expiry_ms)
} else {
Entry::new_string(new_val)
};
Expand Down
11 changes: 5 additions & 6 deletions src/command/string/string_read.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -238,14 +238,13 @@ pub fn getex(db: &mut Database, args: &[Frame]) -> Frame {

// Parse options using zero-alloc case-insensitive comparison
if args.len() > 1 {
let base_ts = db.base_timestamp();
let opt = match extract_bytes(&args[1]) {
Some(b) => b.as_ref(),
None => return Frame::Error(Bytes::from_static(b"ERR syntax error")),
};
if opt.eq_ignore_ascii_case(b"PERSIST") {
if let Some(entry) = db.get_mut(&key) {
entry.set_expires_at_ms(base_ts, 0);
entry.set_expires_at_ms(0);
}
} else if opt.eq_ignore_ascii_case(b"EX") {
if args.len() < 3 {
Expand All @@ -254,7 +253,7 @@ pub fn getex(db: &mut Database, args: &[Frame]) -> Frame {
match parse_positive_i64(&args[2]) {
Some(secs) => {
if let Some(entry) = db.get_mut(&key) {
entry.set_expires_at_ms(base_ts, current_time_ms() + (secs as u64) * 1000);
entry.set_expires_at_ms(current_time_ms() + (secs as u64) * 1000);
}
}
None => {
Expand All @@ -270,7 +269,7 @@ pub fn getex(db: &mut Database, args: &[Frame]) -> Frame {
match parse_positive_i64(&args[2]) {
Some(ms) => {
if let Some(entry) = db.get_mut(&key) {
entry.set_expires_at_ms(base_ts, current_time_ms() + ms as u64);
entry.set_expires_at_ms(current_time_ms() + ms as u64);
}
}
None => {
Expand All @@ -286,7 +285,7 @@ pub fn getex(db: &mut Database, args: &[Frame]) -> Frame {
match parse_positive_i64(&args[2]) {
Some(ts) => {
if let Some(entry) = db.get_mut(&key) {
entry.set_expires_at_ms(base_ts, (ts as u64) * 1000);
entry.set_expires_at_ms((ts as u64) * 1000);
}
}
None => {
Expand All @@ -302,7 +301,7 @@ pub fn getex(db: &mut Database, args: &[Frame]) -> Frame {
match parse_positive_i64(&args[2]) {
Some(ts_ms) => {
if let Some(entry) = db.get_mut(&key) {
entry.set_expires_at_ms(base_ts, ts_ms as u64);
entry.set_expires_at_ms(ts_ms as u64);
}
}
None => {
Expand Down
Loading
Loading