objectstore_service/backend/
changelog.rs

1//! Change lifecycle tracking and durable write-ahead log.
2//!
3//! When a storage mutation spans both the high-volume (HV) and long-term (LT)
4//! backends, several non-atomic steps must happen in sequence: uploading to LT,
5//! committing a tombstone in HV via compare-and-swap, and cleaning up
6//! unreferenced blobs. A crash at any point can leave orphaned LT blobs.
7//!
8//! This module provides two layers of protection:
9//!
10//! 1. **In-process tracking** — [`ChangeGuard`] is an RAII guard that tracks
11//!    the current [`ChangePhase`] of an operation. When dropped, it spawns a
12//!    background task to clean up whichever blob is unreferenced based on the
13//!    phase reached before the drop. This handles normal errors and early
14//!    returns within a running process.
15//!
16//! 2. **Durable write-ahead log** — The [`ChangeLog`] trait records a
17//!    [`Change`] to durable storage *before* any LT side effects begin. If the
18//!    process crashes, a recovery scan reads outstanding entries and cleans up
19//!    orphaned blobs. Recovery is garbage collection — it never replays CAS
20//!    mutations or finishes incomplete operations.
21
22use std::collections::HashMap;
23use std::fmt;
24use std::sync::{Arc, Mutex};
25use std::time::{Duration, SystemTime};
26
27use objectstore_types::time::Timestamp;
28use sentry::{Hub, SentryFutureExt};
29use tokio_util::task::TaskTracker;
30use tokio_util::task::task_tracker::TaskTrackerToken;
31
32use crate::backend::common::{HighVolumeBackend, MultipartUploadBackend, TieredMetadata};
33use crate::error::Result;
34use crate::id::ObjectId;
35
36/// Initial delay for exponential backoff retries in background cleanup tasks.
37const INITIAL_BACKOFF: Duration = Duration::from_millis(100);
38/// Maximum delay for exponential backoff retries in background cleanup tasks.
39const MAX_BACKOFF: Duration = Duration::from_secs(30);
40
41/// Unique identifier for a change log entry.
42///
43/// Generated per-operation as a UUIDv7. In durable storage, scoped to the
44/// owning service instance (e.g., `~oplog/{instance_id}/{change_id}`).
45#[derive(Debug, Clone, PartialEq, Eq, Hash)]
46pub struct ChangeId(uuid::Uuid);
47
48impl ChangeId {
49    /// Generates a new unique change ID.
50    #[allow(clippy::new_without_default)]
51    pub fn new() -> Self {
52        Self(uuid::Uuid::now_v7())
53    }
54}
55
56impl fmt::Display for ChangeId {
57    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
58        self.0.fmt(f)
59    }
60}
61
62/// Describes the LT blobs involved in a multi-step storage change.
63///
64/// Every mutating flow maps to: "I may have written a `new` LT blob and I
65/// may be replacing an `old` LT blob." Recovery uses these fields to determine
66/// which blobs are orphaned by reading the current HV state.
67#[derive(Debug, Clone)]
68pub struct Change {
69    /// The logical object being mutated.
70    ///
71    /// Used by cleanup to query HV and determine which blob is currently referenced.
72    pub id: ObjectId,
73    /// The new LT blob written by this operation.
74    ///
75    /// Needs cleanup on failure (the CAS did not commit).
76    pub new: Option<ObjectId>,
77    /// The old LT blob being replaced.
78    ///
79    /// Needs cleanup on success (the CAS committed and the old blob is unreferenced).
80    pub old: Option<ObjectId>,
81    /// Earliest time at which this entry becomes eligible for cleanup.
82    ///
83    /// [`ChangeLog::scan`] filters out the entry, unless the deadline has passed.
84    pub cleanup_after: Option<SystemTime>,
85}
86
87/// Manager for multi-step storage changes, including backends and durable log.
88///
89/// Encapsulates the state and logic for recording changes, advancing their phases,
90/// and performing cleanup on drop. The `TieredStorage` backend holds an instance
91/// of this manager to use it for its multi-step operations.
92#[derive(Debug)]
93pub struct ChangeManager {
94    /// The backend for small objects (≤ 1 MiB).
95    pub(crate) high_volume: Box<dyn HighVolumeBackend>,
96    /// The backend for large objects (> 1 MiB).
97    pub(crate) long_term: Box<dyn MultipartUploadBackend>,
98    /// Durable write-ahead log for multi-step changes.
99    pub(crate) changelog: Box<dyn ChangeLog>,
100    /// Tracks outstanding background cleanup operations for graceful shutdown.
101    pub(crate) tracker: TaskTracker,
102}
103
104impl ChangeManager {
105    /// Creates a new `ChangeManager` with the given backends and changelog.
106    pub fn new(
107        high_volume: Box<dyn HighVolumeBackend>,
108        long_term: Box<dyn MultipartUploadBackend>,
109        changelog: Box<dyn ChangeLog>,
110    ) -> Arc<Self> {
111        Arc::new(Self {
112            high_volume,
113            long_term,
114            changelog,
115            tracker: TaskTracker::new(),
116        })
117    }
118
119    /// Records the change to the log and returns a guard.
120    ///
121    /// Generates a unique [`ChangeId`] and writes a durable log entry before
122    /// returning. The caller may proceed with LT side effects immediately after.
123    ///
124    /// When the [`ChangeGuard`] is dropped, a background process is spawned to
125    /// clean up any unreferenced objects in LT storage.
126    #[tracing::instrument(level = "debug", fields(id = ?change.id, new = ?change.new, old = ?change.old), skip_all)]
127    pub async fn record(self: Arc<Self>, change: Change) -> Result<ChangeGuard> {
128        let token = self.tracker.token();
129
130        let id = ChangeId::new();
131        self.changelog.record(&id, &change).await?;
132
133        let state = ChangeState {
134            id,
135            change,
136            phase: ChangePhase::Recorded,
137            manager: self.clone(),
138            _token: token,
139        };
140
141        Ok(ChangeGuard {
142            state: Some(state),
143            hub: Hub::current(),
144        })
145    }
146
147    /// Records the change to the log and returns a guard in the `Assembling` state.
148    ///
149    /// Behaves like [`Self::record`], except that the guard is created in the `Assembling` state.
150    /// Unlike other states, this guard does nothing on drop, leaving the burden of cleaning up to
151    /// the [`ChangeLog`].
152    #[tracing::instrument(level = "debug", fields(id = ?change.id, new = ?change.new, old = ?change.old), skip_all)]
153    pub async fn record_assembling(self: Arc<Self>, change: Change) -> Result<ChangeGuard> {
154        let token = self.tracker.token();
155
156        let id = ChangeId::new();
157        self.changelog.record(&id, &change).await?;
158
159        let state = ChangeState {
160            id,
161            change,
162            phase: ChangePhase::Assembling,
163            manager: self.clone(),
164            _token: token,
165        };
166
167        Ok(ChangeGuard {
168            state: Some(state),
169            hub: Hub::current(),
170        })
171    }
172
173    /// Scans the changelog for outstanding entries and runs cleanup for each.
174    ///
175    /// Spawn this into a background task at startup to recover from any orphaned objects after a
176    /// crash. During normal operation, this should return an empty list and have no effect.
177    #[tracing::instrument(level = "debug", skip_all)]
178    pub async fn recover(self: Arc<Self>) -> Result<()> {
179        // Hold one token for the duration of recovery to prevent premature shutdown.
180        let _token = self.tracker.token();
181
182        let entries =
183            self.changelog.scan().await.inspect_err(|e| {
184                objectstore_log::error!(!!e, "Failed to run changelog recovery")
185            })?;
186
187        // NB: Intentionally clean up sequentially to reduce load on the system.
188        for (id, change) in entries {
189            let state = ChangeState {
190                id,
191                change,
192                phase: ChangePhase::Recovered,
193                manager: self.clone(),
194                _token: self.tracker.token(),
195            };
196
197            state.cleanup().await;
198        }
199
200        Ok(())
201    }
202}
203
204/// Durable write-ahead log for multi-step storage changes.
205///
206/// Records in-progress changes that span both HV and LT backends so that
207/// recovery can identify and clean up orphaned LT blobs after crashes.
208/// The log is stored independently from the data backend (though it may
209/// share infrastructure) and is scoped per service instance.
210///
211/// Recovery is garbage collection — it reads HV state to determine which
212/// blobs are unreferenced and deletes them. It never replays CAS mutations
213/// or finishes incomplete operations.
214///
215/// Implementations handle instance identity, heartbeats, and key prefixing
216/// internally — callers interact only with entries.
217#[async_trait::async_trait]
218pub trait ChangeLog: fmt::Debug + Send + Sync {
219    /// Records a change before any side effects begin (write-ahead).
220    ///
221    /// Must be durable before returning — the caller will proceed with
222    /// LT writes immediately after.
223    async fn record(&self, id: &ChangeId, change: &Change) -> Result<()>;
224
225    /// Removes a completed change from the log.
226    ///
227    /// Called after all cleanup (LT blob deletion) is finished. Removing
228    /// a nonexistent entry is not an error (idempotent).
229    async fn remove(&self, id: &ChangeId) -> Result<()>;
230
231    /// Returns all outstanding changes eligible for recovery.
232    ///
233    /// During normal operation this returns only the calling instance's
234    /// entries. During recovery of a dead instance, the implementation
235    /// may return that instance's entries after the caller has claimed
236    /// ownership (via heartbeat CAS).
237    ///
238    /// The returned entries are unordered.
239    async fn scan(&self) -> Result<Vec<(ChangeId, Change)>>;
240}
241
242/// In-memory [`ChangeLog`] for tests and deployments without durable logging.
243///
244/// Stores entries in a `HashMap`. [`Clone`]-able so tests can hold a handle
245/// for direct inspection while the service owns a boxed copy.
246#[derive(Debug, Clone, Default)]
247pub struct InMemoryChangeLog {
248    entries: Arc<Mutex<HashMap<ChangeId, Change>>>,
249}
250
251#[async_trait::async_trait]
252impl ChangeLog for InMemoryChangeLog {
253    async fn record(&self, id: &ChangeId, change: &Change) -> Result<()> {
254        let mut entries = self.entries.lock().expect("lock poisoned");
255        entries.insert(id.clone(), change.clone());
256        Ok(())
257    }
258
259    async fn remove(&self, id: &ChangeId) -> Result<()> {
260        let mut entries = self.entries.lock().expect("lock poisoned");
261        entries.remove(id);
262        Ok(())
263    }
264
265    async fn scan(&self) -> Result<Vec<(ChangeId, Change)>> {
266        let now = SystemTime::now();
267        let entries = self.entries.lock().expect("lock poisoned");
268        let result = entries
269            .iter()
270            .filter(|(_, change)| match change.cleanup_after {
271                None => true,
272                Some(deadline) => now >= deadline,
273            })
274            .map(|(id, change)| (id.clone(), change.clone()))
275            .collect();
276        Ok(result)
277    }
278}
279
280#[cfg(test)]
281impl InMemoryChangeLog {
282    /// Sets [`Change::cleanup_after`] to the past for all entries, forcing them to be returned by a subsequent [`ChangeLog::scan`].
283    pub fn expire_all(&self) {
284        let mut entries = self.entries.lock().expect("lock poisoned");
285        for change in entries.values_mut() {
286            change.cleanup_after = Some(SystemTime::UNIX_EPOCH);
287        }
288    }
289}
290
291/// [`ChangeLog`] implementation that discards all entries.
292///
293/// Used as the default when no durable log is configured. Provides no
294/// crash-recovery guarantees — orphan cleanup relies entirely on in-process
295/// [`ChangeGuard`] drop logic.
296#[derive(Debug, Default)]
297pub struct NoopChangeLog;
298
299#[async_trait::async_trait]
300impl ChangeLog for NoopChangeLog {
301    async fn record(&self, _id: &ChangeId, _change: &Change) -> Result<()> {
302        Ok(())
303    }
304
305    async fn remove(&self, _id: &ChangeId) -> Result<()> {
306        Ok(())
307    }
308
309    async fn scan(&self) -> Result<Vec<(ChangeId, Change)>> {
310        Ok(Vec::new())
311    }
312}
313
314/// Phase of a multi-step storage change.
315#[derive(Debug, PartialEq, Eq)]
316pub enum ChangePhase {
317    /// The change was recovered from changelog and the phase is unknown.
318    Recovered,
319    /// The change is recorded in the log and LT upload has started.
320    Recorded,
321    /// The LT blob originated from a multipart upload and is being assembled.
322    ///
323    /// Multipart upload completion can fail, and we want the client to be able to retry it
324    /// without the change cleanup process racing to delete the LT blob.
325    /// Therefore, cleanup of changes in this phase is deferred.
326    Assembling,
327    /// LT upload has succeeded and the tombstone is being updated.
328    Written,
329    /// The tombstone update failed due to a conflict.
330    Lost,
331    /// The tombstone update succeeded.
332    Updated,
333    /// Cleanup complete.
334    Completed,
335}
336
337impl ChangePhase {
338    /// Returns the phase corresponding to the outcome of a compare-and-write operation.
339    pub fn compare_and_write(succeeded: bool) -> Self {
340        if succeeded { Self::Updated } else { Self::Lost }
341    }
342}
343
344/// Internal state for a [`ChangeGuard`].
345///
346/// Logs an error if dropped in any phase other than `Completed`.
347#[derive(Debug)]
348struct ChangeState {
349    id: ChangeId,
350    change: Change,
351    phase: ChangePhase,
352    manager: Arc<ChangeManager>,
353    _token: TaskTrackerToken,
354}
355
356impl ChangeState {
357    /// Marks the operation as completed, preventing any cleanup on drop.
358    fn mark_completed(mut self) {
359        self.phase = ChangePhase::Completed;
360    }
361
362    /// Determines tombstone state and runs cleanup for unreferenced objects.
363    #[tracing::instrument(level = "debug", skip_all, fields(phase= ?self.phase, new = ?self.change.new, old = ?self.change.old))]
364    async fn cleanup(self) {
365        let current = match self.phase {
366            // For `Recovered`, we must first check the state of the tombstone.
367            ChangePhase::Recovered => self.read_tombstone().await,
368            ChangePhase::Recorded => self.change.old.clone(),
369            // For `Written`, the CAS outcome is unknown — read HV to determine it.
370            ChangePhase::Written => self.read_tombstone().await,
371            ChangePhase::Lost => self.change.old.clone(),
372            ChangePhase::Updated => self.change.new.clone(),
373            ChangePhase::Assembling | ChangePhase::Completed => return, // unreachable
374        };
375
376        if current != self.change.old
377            && let Some(ref old) = self.change.old
378        {
379            self.cleanup_lt(old).await;
380        }
381
382        if current != self.change.new
383            && let Some(ref new) = self.change.new
384        {
385            self.cleanup_lt(new).await;
386        }
387
388        self.cleanup_log().await;
389        self.mark_completed();
390    }
391
392    /// Reads the tombstone target for `id` from HV, retrying with exponential backoff on error.
393    ///
394    /// Returns `None` if the entry holds an inline object or is absent.
395    async fn read_tombstone(&self) -> Option<ObjectId> {
396        let mut delay = INITIAL_BACKOFF;
397        loop {
398            match self
399                .manager
400                .high_volume
401                .get_tiered_metadata(&self.change.id, Timestamp::now())
402                .await
403            {
404                Ok(TieredMetadata::Tombstone(t)) => return Some(t.target),
405                Ok(TieredMetadata::Object(_)) => return None,
406                Ok(TieredMetadata::NotFound) => return None,
407                Err(_) => {
408                    tokio::time::sleep(delay).await;
409                    delay = (delay.mul_f32(1.5)).min(MAX_BACKOFF);
410                }
411            }
412        }
413    }
414
415    /// Deletes `target` from `lt`, retrying with exponential backoff until success.
416    async fn cleanup_lt(&self, target: &ObjectId) {
417        let mut delay = INITIAL_BACKOFF;
418        while self
419            .manager
420            .long_term
421            .delete_object(target, Timestamp::now())
422            .await
423            .is_err()
424        {
425            tokio::time::sleep(delay).await;
426            delay = (delay.mul_f32(1.5)).min(MAX_BACKOFF);
427        }
428    }
429
430    /// Removes this change's log entry, retrying with exponential backoff until success.
431    async fn cleanup_log(&self) {
432        let mut delay = INITIAL_BACKOFF;
433        while self.manager.changelog.remove(&self.id).await.is_err() {
434            tokio::time::sleep(delay).await;
435            delay = (delay.mul_f32(1.5)).min(MAX_BACKOFF);
436        }
437    }
438}
439
440impl Drop for ChangeState {
441    fn drop(&mut self) {
442        match self.phase {
443            ChangePhase::Completed => {}
444            ChangePhase::Assembling => {
445                objectstore_log::warn!(
446                    change = ?self.change,
447                    "Operation dropped in Assembling state, cleanup deferred to ChangeLog recovery"
448                );
449            }
450            _ => {
451                objectstore_log::error!(
452                    change = ?self.change,
453                    phase = ?self.phase,
454                    "Operation dropped without completing cleanup"
455                );
456            }
457        }
458    }
459}
460
461/// RAII guard that tracks cleanup state for a multi-step storage change.
462///
463/// When dropped in a non-`Completed` phase, determines the LT blob to clean up
464/// and spawns a background task to delete it. If no tokio runtime is available
465/// (e.g., during shutdown), the drop logs an error instead of panicking.
466pub struct ChangeGuard {
467    state: Option<ChangeState>,
468    hub: Arc<Hub>,
469}
470
471impl ChangeGuard {
472    /// Advances the operation to the given phase. Zero-cost, no I/O.
473    pub(crate) fn advance(&mut self, phase: ChangePhase) {
474        if let Some(ref mut state) = self.state {
475            state.phase = phase;
476        }
477    }
478}
479
480impl fmt::Debug for ChangeGuard {
481    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
482        f.debug_struct("ChangeGuard")
483            .field("state", &self.state)
484            // hide `hub` intentionally
485            .finish()
486    }
487}
488
489impl Drop for ChangeGuard {
490    fn drop(&mut self) {
491        if let Some(state) = self.state.take()
492            && state.phase != ChangePhase::Assembling
493            && state.phase != ChangePhase::Completed
494            && let Ok(handle) = tokio::runtime::Handle::try_current()
495        {
496            let hub = Hub::new_from_top(&self.hub);
497            handle.spawn(state.cleanup().bind_hub(hub));
498        }
499
500        // NB: Drop of `ChangeState` logs an error if cleanup is not scheduled.
501    }
502}
503
504#[cfg(test)]
505mod tests {
506    use objectstore_types::scope::{Scope, Scopes};
507
508    use super::*;
509    use crate::id::ObjectContext;
510
511    fn make_id(key: &str) -> ObjectId {
512        ObjectId::new(
513            ObjectContext {
514                usecase: "testing".into(),
515                scopes: Scopes::from_iter([Scope::create("testing", "value").unwrap()]),
516            },
517            key.into(),
518        )
519    }
520
521    #[tokio::test]
522    async fn record_then_scan_returns_entry() {
523        let log = InMemoryChangeLog::default();
524        let id = ChangeId::new();
525        let change = Change {
526            id: make_id("object-key"),
527            new: Some(make_id("object-key/rev1")),
528            old: None,
529            cleanup_after: None,
530        };
531
532        log.record(&id, &change).await.unwrap();
533
534        let entries = log.scan().await.unwrap();
535        assert_eq!(entries.len(), 1);
536        assert_eq!(entries[0].0, id);
537    }
538
539    #[tokio::test]
540    async fn remove_then_scan_does_not_return_entry() {
541        let log = InMemoryChangeLog::default();
542        let id = ChangeId::new();
543        let change = Change {
544            id: make_id("object-key"),
545            new: None,
546            old: Some(make_id("object-key/rev1")),
547            cleanup_after: None,
548        };
549
550        log.record(&id, &change).await.unwrap();
551        log.remove(&id).await.unwrap();
552
553        let entries = log.scan().await.unwrap();
554        assert!(entries.is_empty());
555    }
556
557    #[tokio::test]
558    async fn remove_nonexistent_entry_is_not_an_error() {
559        let log = InMemoryChangeLog::default();
560        let id = ChangeId::new();
561
562        log.remove(&id).await.unwrap();
563    }
564
565    /// When the tokio runtime is dropped while an operation is in flight, the `ChangeGuard`
566    /// drops outside any runtime and cannot schedule cleanup. The log entry must persist
567    /// so that a future recovery pass can identify and clean up orphaned blobs.
568    #[test]
569    fn runtime_drop_while_pending_preserves_log_entry() {
570        use crate::backend::in_memory::InMemoryBackend;
571
572        let log = InMemoryChangeLog::default();
573        let manager = ChangeManager::new(
574            Box::new(InMemoryBackend::new("hv")),
575            Box::new(InMemoryBackend::new("lt")),
576            Box::new(log.clone()),
577        );
578
579        let guard = {
580            let rt = tokio::runtime::Runtime::new().unwrap();
581            // Simulate a mid-flight operation that recorded its change but did not complete.
582            rt.block_on(manager.record(Change {
583                id: make_id("crash-test"),
584                new: Some(make_id("crash-test/rev")),
585                old: None,
586                cleanup_after: None,
587            }))
588            .unwrap()
589            // Runtime drops here while `guard` is still alive outside it.
590        };
591
592        // Guard drops with no runtime active: cleanup cannot be scheduled.
593        drop(guard);
594
595        // Log entry must survive so recovery can clean up the orphaned blob.
596        let rt = tokio::runtime::Runtime::new().unwrap();
597        let entries = rt.block_on(log.scan()).unwrap();
598        assert_eq!(entries.len(), 1, "log entry must persist");
599    }
600
601    #[tokio::test]
602    async fn scan_filters_by_cleanup_after() {
603        let log = InMemoryChangeLog::default();
604
605        let ready_id = ChangeId::new();
606        log.record(
607            &ready_id,
608            &Change {
609                id: make_id("ready"),
610                new: Some(make_id("ready/rev")),
611                old: None,
612                cleanup_after: None,
613            },
614        )
615        .await
616        .unwrap();
617
618        let expired_id = ChangeId::new();
619        log.record(
620            &expired_id,
621            &Change {
622                id: make_id("expired"),
623                new: Some(make_id("expired/rev")),
624                old: None,
625                cleanup_after: Some(SystemTime::now() - Duration::from_secs(1)),
626            },
627        )
628        .await
629        .unwrap();
630
631        let deferred_id = ChangeId::new();
632        log.record(
633            &deferred_id,
634            &Change {
635                id: make_id("deferred"),
636                new: Some(make_id("deferred/rev")),
637                old: None,
638                cleanup_after: Some(SystemTime::now() + Duration::from_hours(24)),
639            },
640        )
641        .await
642        .unwrap();
643
644        let entries = log.scan().await.unwrap();
645        assert_eq!(entries.len(), 2);
646
647        let ids: Vec<_> = entries.iter().map(|(id, _)| id).collect();
648        assert!(ids.contains(&&ready_id));
649        assert!(ids.contains(&&expired_id));
650    }
651}
objectstore_service/backend/changelog.rs

objectstore_service/backend/
changelog.rs
