From c2863d9a4f28eacd262fb5b83f1b7fca520abf0d Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 28 Oct 2026 01:49:49 -0700 Subject: [PATCH] vector: owner-prefixed lease tags stop cross-server reaping (#12158) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * vector: owner-prefixed lease tags; foreign leases get a long backstop reap_expired exempted only process-local LIVE_TAGS — server B could reap server A's live query lease at the 1h TTL. Tags now carry a per-process instance id: own leases use the short TTL, foreign leases only the 24h backstop, so a live query elsewhere keeps its pin while dead-process tags still collect eventually. * vector: pick the lease-tag format by segment count, not numeric-ness An owner that is all decimal digits parsed as the legacy timestamp, so its real timestamp was ignored and the lease never expired. The third segment is what distinguishes {instance}-{ts}-{uuid} from {ts}-{uuid}. --- seaweed-vector/crates/core/src/maintenance.rs | 89 +++++++++++++++++-- 1 file changed, 82 insertions(+), 7 deletions(-) diff --git a/seaweed-vector/crates/core/src/maintenance.rs b/seaweed-vector/crates/core/src/maintenance.rs index d3a95567b2..0391fd3fc5 100644 --- a/seaweed-vector/crates/core/src/maintenance.rs +++ b/seaweed-vector/crates/core/src/maintenance.rs @@ -14,9 +14,16 @@ const QUERY_TAG_PREFIX: &str = "weed-vector-query-"; /// A lease outliving this is treated as abandoned by a dead or cancelled /// request and reaped by the next [`pin`]. const LEASE_TTL: Duration = Duration::from_secs(3600); +/// Foreign leases — another server's live queries or a dead process's +/// leftovers — get this much longer backstop; a live query on a different +/// server keeps its pin. +const FOREIGN_TTL: Duration = Duration::from_secs(24 * 3600); /// Expired-lease scans run at most once per interval per dataset, not per /// request. const REAP_INTERVAL: Duration = Duration::from_secs(60); +/// A process-unique owner id embedded in lease tags so a reaper only applies +/// the short TTL to its own leases. +static INSTANCE: LazyLock = LazyLock::new(|| Uuid::new_v4().simple().to_string()); /// Nanos since the Unix epoch when each dataset last had its tags scanned, /// keyed by dataset uri. static LAST_REAP: LazyLock>> = LazyLock::new(Mutex::default); @@ -62,7 +69,11 @@ impl Drop for Lease { pub async fn pin(dataset: &Dataset) -> Result { let now = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos(); reap_expired(dataset, now).await; - let tag = format!("{QUERY_TAG_PREFIX}{now}-{}", Uuid::new_v4().simple()); + let tag = format!( + "{QUERY_TAG_PREFIX}{instance}-{now}-{}", + Uuid::new_v4().simple(), + instance = &*INSTANCE, + ); let version = dataset.version().version; dataset .tags() @@ -141,11 +152,25 @@ async fn reap_expired(dataset: &Dataset, now: u128) { .lock() .map(|live| live.contains(name)) .unwrap_or(false); + // {instance}-{ts}-{uuid} is the owned format; {ts}-{uuid} is a + // legacy foreign tag. Segment count picks the format — an all-digit + // owner parses as a number too. Own leases use the short TTL; + // foreign ones only the long backstop. let expired = name .strip_prefix(QUERY_TAG_PREFIX) - .and_then(|rest| rest.split('-').next()) - .and_then(|stamp| stamp.parse::().ok()) - .is_some_and(|stamp| now.saturating_sub(stamp) > LEASE_TTL.as_nanos()); + .and_then(|rest| { + let mut it = rest.splitn(3, '-'); + let a = it.next()?; + let b = it.next()?; + match it.next() { + Some(_) => Some((a == *INSTANCE, b.parse::().ok()?)), + None => Some((false, a.parse::().ok()?)), + } + }) + .is_some_and(|(own, stamp)| { + now.saturating_sub(stamp) + > if own { LEASE_TTL } else { FOREIGN_TTL }.as_nanos() + }); if expired && !held { let _ = dataset.tags().delete(name).await; } @@ -306,12 +331,12 @@ mod tests { dataset .tags() .create( - &format!("{QUERY_TAG_PREFIX}{stale}-deadbeef"), + &format!("{QUERY_TAG_PREFIX}{}-{stale}-deadbeef", &*INSTANCE), Ref::Version(None, Some(dataset.version().version)), ) .await .unwrap(); - let held = format!("{QUERY_TAG_PREFIX}{stale}-livebeef"); + let held = format!("{QUERY_TAG_PREFIX}{}-{stale}-livebeef", &*INSTANCE); dataset .tags() .create(&held, Ref::Version(None, Some(dataset.version().version))) @@ -328,6 +353,56 @@ mod tests { lease.release().await.unwrap(); } + /// A stale lease from another process is not reaped at the short TTL — + /// only the long foreign backstop applies. + #[tokio::test] + async fn foreign_leases_survive_local_reap() { + let dir = TempDir::new().unwrap(); + let dataset = + testing::write_dataset(&dir, &testing::vectors(4, 4), &["a", "b", "c", "d"]).await; + let stale = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + - LEASE_TTL.as_nanos() + - 1; + let foreign = format!("{QUERY_TAG_PREFIX}other-{stale}-foreign"); + dataset + .tags() + .create(&foreign, Ref::Version(None, Some(dataset.version().version))) + .await + .unwrap(); + let lease = pin(&dataset).await.unwrap(); + let tags = dataset.tags().list().await.unwrap(); + assert!(tags.contains_key(&foreign), "foreign lease was reaped"); + lease.release().await.unwrap(); + } + + /// An all-digit foreign owner still gets the backstop TTL applied to its + /// own timestamp rather than being read as a far-future legacy stamp. + #[tokio::test] + async fn all_digit_foreign_owner_still_expires() { + let dir = TempDir::new().unwrap(); + let dataset = + testing::write_dataset(&dir, &testing::vectors(4, 4), &["a", "b", "c", "d"]).await; + let stale = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + - FOREIGN_TTL.as_nanos() + - 1; + let tag = format!("{QUERY_TAG_PREFIX}12345678901234567890123456789012-{stale}-deadbeef"); + dataset + .tags() + .create(&tag, Ref::Version(None, Some(dataset.version().version))) + .await + .unwrap(); + let lease = pin(&dataset).await.unwrap(); + let tags = dataset.tags().list().await.unwrap(); + assert!(!tags.contains_key(&tag), "all-digit foreign owner was kept"); + lease.release().await.unwrap(); + } + /// Pins within one reap interval do not rescan the dataset's tags. #[tokio::test] async fn pin_throttles_reap_scans() { @@ -341,7 +416,7 @@ mod tests { .as_nanos() - LEASE_TTL.as_nanos() - 1; - let tag = format!("{QUERY_TAG_PREFIX}{stale}-slowbeef"); + let tag = format!("{QUERY_TAG_PREFIX}{}-{stale}-slowbeef", &*INSTANCE); dataset .tags() .create(&tag, Ref::Version(None, Some(dataset.version().version)))