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

objectstore_service/backend/
changelog.rs
