fix(ecstore): accept empty remote version_id in tier recovery paths (#4552)

An object transitioned to an unversioned remote tier legally records an
empty remote version_id (per CLAUDE.md: a tier version of `None`/`""`
means the tier bucket is unversioned, so remote GET/DELETE must omit the
versionId). Two recovery paths wrongly treated that empty sentinel as an
incomplete/unrecoverable record while the persist/encode and worker
paths accept it, producing permanent leaks in the crash-recovery window.

- tier_delete_journal::into_jentry rejected entries with an empty
  version_id as "incomplete", so unversioned-tier WAL entries could never
  be decoded during recovery: the remote object was orphaned and the
  journal file leaked forever. Drop the version_id emptiness check; keep
  the obj_name/tier_name checks. Truncated payloads are still rejected at
  JSON deserialization (all fields are required, non-Option, no default,
  with deny_unknown_fields).

- tier_free_version_recovery::is_recoverable_tier_free_version required a
  non-empty version_id, silently filtering out free versions of
  unversioned-tier objects so a first enqueue failure leaked the remote
  object and local free version permanently. Drop the version_id check;
  keep free_version + name + tier checks.

Both downstream paths already issue a versionless remote delete for an
empty version_id, so no further changes are needed. Adds regression
tests covering the empty-version_id case and the preserved guards.

Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs

@@ -63,7 +63,12 @@ impl PersistedTierDeleteJournalEntry {
         if self.version != TIER_DELETE_JOURNAL_VERSION {
             return Err(Error::other(format!("unsupported tier delete journal version {}", self.version)));
         }
-        if self.obj_name.is_empty() || self.version_id.is_empty() || self.tier_name.is_empty() {
+        // Empty `version_id` is a legal sentinel for objects transitioned to an
+        // unversioned remote tier (see CLAUDE.md: a tier version of `None`/`""`
+        // means the tier bucket is unversioned, so the remote delete is issued
+        // without a versionId). Only reject entries missing the object or tier
+        // name, which are always populated for a TRANSITION_COMPLETE object.
+        if self.obj_name.is_empty() || self.tier_name.is_empty() {
             return Err(Error::other("tier delete journal entry is incomplete"));
         }
         Ok(Jentry {
@@ -320,4 +325,39 @@ mod tests {
 
         assert!(err.to_string().contains("incomplete"));
     }
+
+    #[test]
+    fn tier_delete_journal_recovers_unversioned_tier_entry() {
+        // A remote tier that is unversioned records an empty `version_id`. Such a
+        // WAL entry must decode successfully so recovery can drive a versionless
+        // remote delete, otherwise the remote object is orphaned and the journal
+        // file leaks forever.
+        let payload = br#"{"version":1,"obj_name":"remote/object","version_id":"","tier_name":"WARM"}"#;
+
+        let decoded = decode_tier_delete_journal_entry(payload).expect("unversioned tier entry should decode");
+
+        assert_eq!(decoded.obj_name, "remote/object");
+        assert!(decoded.version_id.is_empty());
+        assert_eq!(decoded.tier_name, "WARM");
+    }
+
+    #[test]
+    fn tier_delete_journal_rejects_missing_tier_name() {
+        let payload = br#"{"version":1,"obj_name":"remote/object","version_id":"v1","tier_name":""}"#;
+
+        let err = decode_tier_delete_journal_entry(payload).expect_err("entry missing tier name should be rejected");
+
+        assert!(err.to_string().contains("incomplete"));
+    }
+
+    #[test]
+    fn tier_delete_journal_rejects_truncated_payload() {
+        // A partially written journal file fails at JSON deserialization, so
+        // relaxing the empty-version_id check does not admit truncated records.
+        let payload = br#"{"version":1,"obj_name":"remote/object","version_id":""#;
+
+        let err = decode_tier_delete_journal_entry(payload).expect_err("truncated journal payload should be rejected");
+
+        assert!(err.to_string().contains("decode tier delete journal failed"));
+    }
 }

crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs

@@ -314,10 +314,13 @@ fn mark_scan_truncated_if_needed(
 }
 
 fn is_recoverable_tier_free_version(oi: &ObjectInfo) -> bool {
-    oi.transitioned_object.free_version
-        && !oi.transitioned_object.name.is_empty()
-        && !oi.transitioned_object.version_id.is_empty()
-        && !oi.transitioned_object.tier.is_empty()
+    // A free version transitioned to an unversioned remote tier legally carries an
+    // empty `version_id` (see CLAUDE.md: a tier version of `None`/`""` means the
+    // tier bucket is unversioned). The worker path handles that by issuing a
+    // versionless remote delete, so the recovery gate must not treat an empty
+    // `version_id` as unrecoverable — the object name and tier are sufficient to
+    // identify a free version that still needs remote cleanup.
+    oi.transitioned_object.free_version && !oi.transitioned_object.name.is_empty() && !oi.transitioned_object.tier.is_empty()
 }
 
 #[cfg(test)]
@@ -420,7 +423,33 @@ mod tests {
 
         assert!(is_recoverable_tier_free_version(&oi));
 
+        // An unversioned remote tier records an empty `version_id`; such a free
+        // version must still be recoverable so it can be re-enqueued for a
+        // versionless remote delete instead of leaking forever.
         oi.transitioned_object.version_id.clear();
+        assert!(is_recoverable_tier_free_version(&oi));
+
+        // The object name and tier are the load-bearing fields; missing either
+        // still marks the entry as unrecoverable.
+        let mut missing_name = oi.clone();
+        missing_name.transitioned_object.name.clear();
+        assert!(!is_recoverable_tier_free_version(&missing_name));
+
+        let mut missing_tier = oi.clone();
+        missing_tier.transitioned_object.tier.clear();
+        assert!(!is_recoverable_tier_free_version(&missing_tier));
+    }
+
+    #[test]
+    fn recoverable_tier_free_version_rejects_non_free_version() {
+        // Relaxing the empty-version_id gate must not admit ordinary (non-free)
+        // versions into the remote-cleanup recovery path.
+        let mut oi = ObjectInfo::default();
+        oi.transitioned_object.free_version = false;
+        oi.transitioned_object.name = "remote/object".to_string();
+        oi.transitioned_object.version_id = "remote-version".to_string();
+        oi.transitioned_object.tier = "WARM".to_string();
+
         assert!(!is_recoverable_tier_free_version(&oi));
     }