feat: record confirmed public Session history through durable Journal Jobs

crates/session-store/src/fs_store.rs

@@ -48,17 +48,30 @@ pub struct FsStore {
     /// `write_all` may have extended the file, so rollback is safe only while
     /// no sibling writer can append behind it.
     append_lock: Arc<Mutex<()>>,
+    retention: crate::retention::SessionRetentionStore,
 }
 
 impl FsStore {
     /// Create a new `FsStore` rooted at the given directory.
     /// Creates the directory if it does not exist.
     pub fn new(root: impl Into<PathBuf>) -> Result<Self, StoreError> {
+        let root = root.into();
+        let retention = crate::retention::fs_retention(&root);
+        Self::new_with_retention(root, retention)
+    }
+
+    /// Hosts that relocate Worker originals must keep control state outside the relocated
+    /// aggregate and pass the same store to readers and deletion/transfer operations.
+    pub fn new_with_retention(
+        root: impl Into<PathBuf>,
+        retention: crate::retention::SessionRetentionStore,
+    ) -> Result<Self, StoreError> {
         let root = root.into();
         fs::create_dir_all(&root)?;
         Ok(Self {
             root,
             append_lock: Arc::new(Mutex::new(())),
+            retention,
         })
     }
 
@@ -98,12 +111,46 @@ impl FsStore {
     /// Returns `Ok(true)` when a Session directory was removed and `Ok(false)`
     /// when it was already absent.
     pub fn delete_session(&self, session_id: SessionId) -> Result<bool, StoreError> {
-        let session_dir = self.session_dir(session_id);
-        match fs::remove_dir_all(&session_dir) {
-            Ok(()) => Ok(true),
-            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
-            Err(error) => Err(error.into()),
-        }
+        self.delete_session_with_intent(
+            session_id,
+            crate::retention::SessionDeletionIntent::Ordinary,
+        )
+    }
+
+    pub fn retention(&self) -> crate::retention::SessionRetentionStore {
+        self.retention.clone()
+    }
+
+    pub fn pin_session(
+        &self,
+        session_id: SessionId,
+        pin: crate::retention::SessionRetentionPin,
+    ) -> Result<(), StoreError> {
+        self.retention().pin(session_id, pin, || {
+            Ok(self.session_dir(session_id).is_dir())
+        })
+    }
+
+    pub fn release_session_pin(
+        &self,
+        session_id: SessionId,
+        pin: crate::retention::SessionRetentionPin,
+    ) -> Result<(), StoreError> {
+        self.retention().release(session_id, pin)
+    }
+
+    pub fn delete_session_with_intent(
+        &self,
+        session_id: SessionId,
+        intent: crate::retention::SessionDeletionIntent,
+    ) -> Result<bool, StoreError> {
+        self.retention().delete(session_id, intent, || {
+            match fs::remove_dir_all(self.session_dir(session_id)) {
+                Ok(()) => Ok(true),
+                Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
+                Err(error) => Err(error.into()),
+            }
+        })
     }
 
     fn session_dir(&self, session_id: SessionId) -> PathBuf {
@@ -288,6 +335,7 @@ impl Store for FsStore {
         segment_id: SegmentId,
         entry: &LogEntry,
     ) -> Result<(), StoreError> {
+        let _retention = self.retention().lock_live(session_id)?;
         let line = serde_json::to_string(entry)?;
         self.append_line(&self.log_path(session_id, segment_id), &line)
     }
@@ -374,6 +422,7 @@ impl Store for FsStore {
         segment_id: SegmentId,
         entries: &[LogEntry],
     ) -> Result<(), StoreError> {
+        let _retention = self.retention().lock_live(session_id)?;
         let path = self.log_path(session_id, segment_id);
         if let Some(parent) = path.parent() {
             fs::create_dir_all(parent)?;
@@ -423,6 +472,7 @@ impl Store for FsStore {
         content: &str,
         limits: PasteArtifactLimits,
     ) -> Result<PasteArtifactRef, StoreError> {
+        let _retention = self.retention().lock_live(session_id)?;
         let _guard = self
             .append_lock
             .lock()
@@ -451,6 +501,7 @@ impl Store for FsStore {
         content: &[u8],
         limits: UploadedFileLimits,
     ) -> Result<UploadedFileRef, StoreError> {
+        let _retention = self.retention().lock_live(session_id)?;
         let _guard = self
             .append_lock
             .lock()
@@ -474,6 +525,7 @@ impl Store for FsStore {
         context: &UploadedFileUploadContext,
         limits: UploadedFileLimits,
     ) -> Result<UploadedFileRef, StoreError> {
+        let _retention = self.retention().lock_live(session_id)?;
         let _guard = self
             .append_lock
             .lock()
@@ -510,6 +562,7 @@ impl Store for FsStore {
         reference: &UploadedFileRef,
         source_entry_id: &str,
     ) -> Result<UploadedFileRef, StoreError> {
+        let _retention = self.retention().lock_live(session_id)?;
         let _guard = self
             .append_lock
             .lock()
@@ -537,6 +590,7 @@ impl Store for FsStore {
         reference: &UploadedFileRef,
         owner_id: &str,
     ) -> Result<(), StoreError> {
+        let _retention = self.retention().lock_live(session_id)?;
         let _guard = self
             .append_lock
             .lock()
@@ -550,6 +604,7 @@ impl Store for FsStore {
         artifact_id: &str,
         owner_id: &str,
     ) -> Result<(), StoreError> {
+        let _retention = self.retention().lock_live(session_id)?;
         let _guard = self
             .append_lock
             .lock()
@@ -563,6 +618,7 @@ impl Store for FsStore {
         artifact_id: &str,
         source_entry_id: &str,
     ) -> Result<(), StoreError> {
+        let _retention = self.retention().lock_live(session_id)?;
         let _guard = self
             .append_lock
             .lock()
@@ -579,6 +635,7 @@ impl Store for FsStore {
         session_id: SessionId,
         live_owner_ids: &[String],
     ) -> Result<u64, StoreError> {
+        let _retention = self.retention().lock_live(session_id)?;
         let _guard = self
             .append_lock
             .lock()
@@ -591,6 +648,7 @@ impl Store for FsStore {
         session_id: SessionId,
         artifact_id: &str,
     ) -> Result<bool, StoreError> {
+        let _retention = self.retention().lock_live(session_id)?;
         let _guard = self
             .append_lock
             .lock()
@@ -599,6 +657,7 @@ impl Store for FsStore {
     }
 
     fn delete_uncommitted_uploaded_files(&self, session_id: SessionId) -> Result<u64, StoreError> {
+        let _retention = self.retention().lock_live(session_id)?;
         let _guard = self
             .append_lock
             .lock()
@@ -631,6 +690,13 @@ impl Store for FsStore {
         source_session_id: SessionId,
         target_session_id: SessionId,
     ) -> Result<u64, StoreError> {
+        let mut sessions = vec![source_session_id, target_session_id];
+        sessions.sort();
+        sessions.dedup();
+        let _retention = sessions
+            .into_iter()
+            .map(|id| self.retention().lock_live(id))
+            .collect::<Result<Vec<_>, _>>()?;
         let _guard = self
             .append_lock
             .lock()
@@ -647,6 +713,7 @@ impl Store for FsStore {
         segment_id: SegmentId,
         entry: &TraceEntry,
     ) -> Result<(), StoreError> {
+        let _retention = self.retention().lock_live(session_id)?;
         let line = serde_json::to_string(entry)?;
         self.append_line(&self.trace_path(session_id, segment_id), &line)
     }

crates/session-store/src/lib.rs

@@ -45,7 +45,9 @@ mod legacy_session_log;
 pub mod logged_item;
 mod paste_artifact;
 pub mod public_index;
+pub mod public_material;
 pub mod public_snapshot;
+pub mod retention;
 pub mod segment;
 pub mod segment_log;
 pub mod store;

crates/session-store/src/retention.rs

@@ -0,0 +1,344 @@
+//! Durable, domain-neutral Session retention pins and permanent erasure fences.
+//!
+//! Hosts choose a control root whose lifetime outlasts the physical Session.
+//! Every deletion path must execute its physical removal inside `delete`, and
+//! publishers must hold `lock_live` while creating or appending source data.
+//! A check followed by unlocked removal is not a retention guarantee.
+//!
+//! Pins have explicit owners and no timer/expiry: domain code releases its pin
+//! only after its durable consumer no longer needs originals. A permanent erase
+//! persists its fence before physical removal, invalidates pins, and rejects
+//! late pins/publication even after reopening the control store. Retrying an
+//! interrupted physical erase uses the same intent and is safe.
+
+use std::collections::BTreeSet;
+use std::fs::{self, File, OpenOptions};
+use std::path::{Path, PathBuf};
+
+use fs4::fs_std::FileExt;
+use serde::{Deserialize, Serialize};
+
+use crate::{SessionId, StoreError};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
+#[serde(transparent)]
+pub struct SessionRetentionPin(pub uuid::Uuid);
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum SessionDeletionIntent {
+    Ordinary,
+    ErasePermanently,
+}
+
+#[derive(Debug, Default, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct RetentionState {
+    erased: bool,
+    pins: BTreeSet<SessionRetentionPin>,
+}
+
+#[derive(Debug, Clone)]
+pub struct SessionRetentionStore {
+    root: PathBuf,
+}
+
+/// A process-independent lock; dropping it releases the filesystem lock.
+/// Do not recursively acquire the same Session lock inside a guarded callback.
+pub struct SessionRetentionGuard {
+    _file: File,
+}
+
+impl SessionRetentionStore {
+    pub fn new(control_root: impl Into<PathBuf>) -> Self {
+        Self {
+            root: control_root.into(),
+        }
+    }
+
+    fn state_path(&self, session: SessionId) -> PathBuf {
+        self.root.join(format!("{session}.json"))
+    }
+
+    fn lock(&self, session: SessionId) -> Result<SessionRetentionGuard, StoreError> {
+        fs::create_dir_all(&self.root)?;
+        let mut options = OpenOptions::new();
+        options.read(true).write(true).create(true).truncate(false);
+        #[cfg(any(target_os = "linux", target_os = "android"))]
+        {
+            use std::os::unix::fs::OpenOptionsExt;
+            // Control roots are Host-owned; never follow a replaced lock file.
+            options.custom_flags(libc::O_NOFOLLOW);
+        }
+        let file = options.open(self.root.join(format!("{session}.lock")))?;
+        FileExt::lock_exclusive(&file)?;
+        Ok(SessionRetentionGuard { _file: file })
+    }
+
+    fn read(&self, session: SessionId) -> Result<RetentionState, StoreError> {
+        match fs::read(self.state_path(session)) {
+            Ok(bytes) => Ok(serde_json::from_slice(&bytes)?),
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+                Ok(RetentionState::default())
+            }
+            Err(error) => Err(error.into()),
+        }
+    }
+
+    fn save(&self, session: SessionId, state: &RetentionState) -> Result<(), StoreError> {
+        crate::worker_session_store::atomic_write_bytes(
+            &self.state_path(session),
+            &serde_json::to_vec(state)?,
+        )
+    }
+
+    /// Observational fence check; unlike `lock_live`, never creates storage.
+    /// Hosts still fence result adoption in their own durable consumer: a read
+    /// finishing before erasure is not authority for a delayed domain write.
+    pub fn ensure_live(&self, session: SessionId) -> Result<(), StoreError> {
+        if self.read(session)?.erased {
+            return Err(StoreError::SessionErased(session));
+        }
+        Ok(())
+    }
+
+    pub fn lock_live(&self, session: SessionId) -> Result<SessionRetentionGuard, StoreError> {
+        let guard = self.lock(session)?;
+        if self.read(session)?.erased {
+            return Err(StoreError::SessionErased(session));
+        }
+        Ok(guard)
+    }
+
+    /// Call `source_exists` under the same lock used by deletion. Replaying the
+    /// same pin is idempotent; missing source never becomes a successful lease.
+    pub fn pin(
+        &self,
+        session: SessionId,
+        pin: SessionRetentionPin,
+        source_exists: impl FnOnce() -> Result<bool, StoreError>,
+    ) -> Result<(), StoreError> {
+        let _guard = self.lock_live(session)?;
+        if !source_exists()? {
+            return Err(StoreError::SessionSourceMissing(session));
+        }
+        let mut state = self.read(session)?;
+        if state.pins.insert(pin) {
+            self.save(session, &state)?;
+        }
+        Ok(())
+    }
+
+    pub fn release(&self, session: SessionId, pin: SessionRetentionPin) -> Result<(), StoreError> {
+        let _guard = self.lock(session)?;
+        let mut state = self.read(session)?;
+        if state.pins.remove(&pin) {
+            self.save(session, &state)?;
+        }
+        Ok(())
+    }
+
+    pub fn pins(&self, session: SessionId) -> Result<Vec<SessionRetentionPin>, StoreError> {
+        let _guard = self.lock(session)?;
+        Ok(self.read(session)?.pins.into_iter().collect())
+    }
+
+    /// Execute physical removal while fencing concurrent pins and publishers.
+    /// An erasure fence is retained even if removal fails or the source is
+    /// already absent. Ordinary deletion after erasure is rejected: recovery
+    /// must preserve the caller's explicit erase intent.
+    pub fn delete<T>(
+        &self,
+        session: SessionId,
+        intent: SessionDeletionIntent,
+        remove: impl FnOnce() -> Result<T, StoreError>,
+    ) -> Result<T, StoreError> {
+        let _guard = self.lock(session)?;
+        let mut state = self.read(session)?;
+        match intent {
+            SessionDeletionIntent::Ordinary => {
+                if state.erased {
+                    return Err(StoreError::SessionErased(session));
+                }
+                if !state.pins.is_empty() {
+                    return Err(StoreError::SessionPinned(session));
+                }
+            }
+            SessionDeletionIntent::ErasePermanently => {
+                if !state.erased || !state.pins.is_empty() {
+                    state.erased = true;
+                    state.pins.clear();
+                    self.save(session, &state)?;
+                }
+            }
+        }
+        remove()
+    }
+}
+
+/// FsStore's control state is outside all per-Session source directories.
+pub(crate) fn fs_retention(root: &Path) -> SessionRetentionStore {
+    SessionRetentionStore::new(root.join(".retention"))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn pin_survives_reopen_blocks_delete_and_releases_idempotently() {
+        let temp = tempfile::tempdir().unwrap();
+        let id = crate::new_session_id();
+        let pin = SessionRetentionPin(uuid::Uuid::now_v7());
+        let store = SessionRetentionStore::new(temp.path());
+        store.pin(id, pin, || Ok(true)).unwrap();
+        store.pin(id, pin, || Ok(true)).unwrap();
+        let reopened = SessionRetentionStore::new(temp.path());
+        assert_eq!(reopened.pins(id).unwrap(), [pin]);
+        assert!(matches!(
+            reopened.delete(id, SessionDeletionIntent::Ordinary, || Ok(())),
+            Err(StoreError::SessionPinned(_))
+        ));
+        reopened.release(id, pin).unwrap();
+        reopened.release(id, pin).unwrap();
+        assert_eq!(
+            reopened
+                .delete(id, SessionDeletionIntent::Ordinary, || Ok(7))
+                .unwrap(),
+            7
+        );
+    }
+
+    #[test]
+    fn erase_failure_persists_fence_invalidates_pin_and_rejects_late_publication() {
+        let temp = tempfile::tempdir().unwrap();
+        let id = crate::new_session_id();
+        let pin = SessionRetentionPin(uuid::Uuid::now_v7());
+        let store = SessionRetentionStore::new(temp.path());
+        store.pin(id, pin, || Ok(true)).unwrap();
+        assert!(
+            store
+                .delete(
+                    id,
+                    SessionDeletionIntent::ErasePermanently,
+                    || Err::<(), _>(std::io::Error::other("injected removal failure").into())
+                )
+                .is_err()
+        );
+        let reopened = SessionRetentionStore::new(temp.path());
+        assert!(reopened.pins(id).unwrap().is_empty());
+        assert!(matches!(
+            reopened.pin(id, pin, || Ok(true)),
+            Err(StoreError::SessionErased(_))
+        ));
+        assert!(matches!(
+            reopened.lock_live(id),
+            Err(StoreError::SessionErased(_))
+        ));
+        reopened
+            .delete(id, SessionDeletionIntent::ErasePermanently, || Ok(()))
+            .unwrap();
+        reopened.release(id, pin).unwrap();
+        assert!(matches!(
+            reopened.lock_live(id),
+            Err(StoreError::SessionErased(_))
+        ));
+    }
+
+    #[test]
+    fn fs_normal_delete_obeys_pin_and_permanent_erase_fences_all_source_writes() {
+        use crate::{FsStore, LogEntry, Store};
+        let temp = tempfile::tempdir().unwrap();
+        let store = FsStore::new(temp.path()).unwrap();
+        let session = crate::new_session_id();
+        let segment = crate::new_segment_id();
+        let entry = LogEntry::TurnEnd {
+            ts: 1,
+            turn_count: 1,
+        };
+        store
+            .create_segment(session, segment, &[entry.clone()])
+            .unwrap();
+        let pin = SessionRetentionPin(uuid::Uuid::now_v7());
+        store.pin_session(session, pin).unwrap();
+        assert!(matches!(
+            store.delete_session(session),
+            Err(StoreError::SessionPinned(_))
+        ));
+        assert!(store.exists(session, segment).unwrap());
+        store
+            .delete_session_with_intent(session, SessionDeletionIntent::ErasePermanently)
+            .unwrap();
+        let reopened = FsStore::new(temp.path()).unwrap();
+        assert!(!reopened.exists(session, segment).unwrap());
+        assert!(matches!(
+            reopened.pin_session(session, pin),
+            Err(StoreError::SessionErased(_))
+        ));
+        assert!(matches!(
+            reopened.append(session, segment, &entry),
+            Err(StoreError::SessionErased(_))
+        ));
+        assert!(matches!(
+            reopened.create_segment(session, segment, &[entry]),
+            Err(StoreError::SessionErased(_))
+        ));
+        assert!(
+            !reopened
+                .delete_session_with_intent(session, SessionDeletionIntent::ErasePermanently)
+                .unwrap()
+        );
+        assert!(!reopened.exists(session, segment).unwrap());
+    }
+
+    #[test]
+    fn independent_handles_serialize_delete_with_late_pin() {
+        let temp = tempfile::tempdir().unwrap();
+        let source = temp.path().join("source");
+        fs::create_dir(&source).unwrap();
+        let control = temp.path().join("control");
+        let session = crate::new_session_id();
+        let pin = SessionRetentionPin(uuid::Uuid::now_v7());
+        let (entered_tx, entered_rx) = std::sync::mpsc::channel();
+        let (release_tx, release_rx) = std::sync::mpsc::channel();
+        let control2 = control.clone();
+        let source2 = source.clone();
+        let delete = std::thread::spawn(move || {
+            SessionRetentionStore::new(control2).delete(
+                session,
+                SessionDeletionIntent::Ordinary,
+                || {
+                    entered_tx.send(()).unwrap();
+                    release_rx
+                        .recv_timeout(std::time::Duration::from_secs(10))
+                        .unwrap();
+                    fs::remove_dir(&source2)?;
+                    Ok(())
+                },
+            )
+        });
+        entered_rx
+            .recv_timeout(std::time::Duration::from_secs(10))
+            .unwrap();
+        let late = std::thread::spawn(move || {
+            SessionRetentionStore::new(control).pin(session, pin, || Ok(source.is_dir()))
+        });
+        release_tx.send(()).unwrap();
+        delete.join().unwrap().unwrap();
+        assert!(matches!(
+            late.join().unwrap(),
+            Err(StoreError::SessionSourceMissing(_))
+        ));
+    }
+
+    #[test]
+    fn missing_source_cannot_acquire_pin() {
+        let temp = tempfile::tempdir().unwrap();
+        let id = crate::new_session_id();
+        let store = SessionRetentionStore::new(temp.path());
+        assert!(matches!(
+            store.pin(id, SessionRetentionPin(uuid::Uuid::now_v7()), || Ok(false)),
+            Err(StoreError::SessionSourceMissing(_))
+        ));
+        assert!(store.pins(id).unwrap().is_empty());
+    }
+}

crates/session-store/src/store.rs

@@ -33,6 +33,15 @@ pub enum StoreError {
     #[error("log corrupted at line {line}: {message}")]
     Corrupt { line: usize, message: String },
 
+    #[error("session source is missing: {0}")]
+    SessionSourceMissing(SessionId),
+
+    #[error("session original material is retained by a durable consumer: {0}")]
+    SessionPinned(SessionId),
+
+    #[error("session was permanently erased: {0}")]
+    SessionErased(SessionId),
+
     #[error("retained observation read limit exceeded")]
     ReadLimitExceeded,
 

crates/session-store/src/worker_session_store.rs

@@ -48,6 +48,7 @@ pub struct WorkerSessionStore {
     root: PathBuf,
     session_id: Arc<Mutex<Option<SessionId>>>,
     append_lock: Arc<Mutex<()>>,
+    retention: Option<crate::retention::SessionRetentionStore>,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -164,6 +165,37 @@ impl RetainedSegmentReader {
 }
 
 impl WorkerSessionStore {
+    /// Publishes through the same external control store used by Runtime retention.
+    /// Existing manifest migration is fenced too; future writes keep the guard until their
+    /// physical publication finishes. The control root must outlive this aggregate.
+    pub fn new_with_retention(
+        root: impl Into<PathBuf>,
+        retention: crate::retention::SessionRetentionStore,
+    ) -> Result<Self, StoreError> {
+        let root = root.into();
+        let existing = match fs::read(root.join(SESSION_FILE)) {
+            Ok(bytes) => Some(serde_json::from_slice::<SessionManifest>(&bytes)?.session_id),
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
+            Err(error) => return Err(error.into()),
+        };
+        let _guard = existing
+            .map(|session| retention.lock_live(session))
+            .transpose()?;
+        let mut store = Self::new(root)?;
+        store.retention = Some(retention);
+        Ok(store)
+    }
+
+    fn lock_publication(
+        &self,
+        session: SessionId,
+    ) -> Result<Option<crate::retention::SessionRetentionGuard>, StoreError> {
+        self.retention
+            .as_ref()
+            .map(|store| store.lock_live(session))
+            .transpose()
+    }
+
     /// Open the Session store rooted at `<worker-aggregate>/session`.
     pub fn new(root: impl Into<PathBuf>) -> Result<Self, StoreError> {
         let root = root.into();
@@ -202,6 +234,7 @@ impl WorkerSessionStore {
             root,
             session_id: Arc::new(Mutex::new(session_id)),
             append_lock: Arc::new(Mutex::new(())),
+            retention: None,
         })
     }
 
@@ -224,6 +257,7 @@ impl WorkerSessionStore {
             root,
             session_id: Arc::new(Mutex::new(Some(manifest.session_id))),
             append_lock: Arc::new(Mutex::new(())),
+            retention: None,
         })
     }
 
@@ -688,6 +722,7 @@ impl Store for WorkerSessionStore {
         segment_id: SegmentId,
         entry: &LogEntry,
     ) -> Result<(), StoreError> {
+        let _retention = self.lock_publication(session_id)?;
         self.ensure_session(session_id, true)?;
         self.append_log_entry(&self.log_path(segment_id), entry)
     }
@@ -740,6 +775,7 @@ impl Store for WorkerSessionStore {
         segment_id: SegmentId,
         entries: &[LogEntry],
     ) -> Result<(), StoreError> {
+        let _retention = self.lock_publication(session_id)?;
         self.ensure_session(session_id, true)?;
         let mut content = Vec::new();
         for entry in entries {
@@ -788,6 +824,7 @@ impl Store for WorkerSessionStore {
         content: &str,
         limits: PasteArtifactLimits,
     ) -> Result<PasteArtifactRef, StoreError> {
+        let _retention = self.lock_publication(session_id)?;
         self.ensure_session(session_id, true)?;
         let _guard = self
             .append_lock
@@ -818,6 +855,7 @@ impl Store for WorkerSessionStore {
         content: &[u8],
         limits: UploadedFileLimits,
     ) -> Result<UploadedFileRef, StoreError> {
+        let _retention = self.lock_publication(session_id)?;
         self.ensure_session(session_id, true)?;
         let _guard = self
             .append_lock
@@ -842,6 +880,7 @@ impl Store for WorkerSessionStore {
         context: &UploadedFileUploadContext,
         limits: UploadedFileLimits,
     ) -> Result<UploadedFileRef, StoreError> {
+        let _retention = self.lock_publication(session_id)?;
         self.ensure_session(session_id, true)?;
         let _guard = self
             .append_lock
@@ -881,6 +920,7 @@ impl Store for WorkerSessionStore {
         reference: &UploadedFileRef,
         source_entry_id: &str,
     ) -> Result<UploadedFileRef, StoreError> {
+        let _retention = self.lock_publication(session_id)?;
         self.ensure_session(session_id, false)?;
         let _guard = self
             .append_lock
@@ -909,6 +949,7 @@ impl Store for WorkerSessionStore {
         reference: &UploadedFileRef,
         owner_id: &str,
     ) -> Result<(), StoreError> {
+        let _retention = self.lock_publication(session_id)?;
         self.ensure_session(session_id, false)?;
         let _guard = self
             .append_lock
@@ -923,6 +964,7 @@ impl Store for WorkerSessionStore {
         artifact_id: &str,
         owner_id: &str,
     ) -> Result<(), StoreError> {
+        let _retention = self.lock_publication(session_id)?;
         self.ensure_session(session_id, false)?;
         let _guard = self
             .append_lock
@@ -937,6 +979,7 @@ impl Store for WorkerSessionStore {
         artifact_id: &str,
         source_entry_id: &str,
     ) -> Result<(), StoreError> {
+        let _retention = self.lock_publication(session_id)?;
         self.ensure_session(session_id, false)?;
         let _guard = self
             .append_lock
@@ -954,6 +997,7 @@ impl Store for WorkerSessionStore {
         session_id: SessionId,
         live_owner_ids: &[String],
     ) -> Result<u64, StoreError> {
+        let _retention = self.lock_publication(session_id)?;
         self.ensure_session(session_id, false)?;
         let _guard = self
             .append_lock
@@ -967,6 +1011,7 @@ impl Store for WorkerSessionStore {
         session_id: SessionId,
         artifact_id: &str,
     ) -> Result<bool, StoreError> {
+        let _retention = self.lock_publication(session_id)?;
         self.ensure_session(session_id, false)?;
         let _guard = self
             .append_lock
@@ -976,6 +1021,7 @@ impl Store for WorkerSessionStore {
     }
 
     fn delete_uncommitted_uploaded_files(&self, session_id: SessionId) -> Result<u64, StoreError> {
+        let _retention = self.lock_publication(session_id)?;
         self.ensure_session(session_id, false)?;
         let _guard = self
             .append_lock
@@ -1010,6 +1056,7 @@ impl Store for WorkerSessionStore {
         segment_id: SegmentId,
         entry: &TraceEntry,
     ) -> Result<(), StoreError> {
+        let _retention = self.lock_publication(session_id)?;
         self.ensure_session(session_id, true)?;
         self.append_line(&self.trace_path(segment_id), &serde_json::to_string(entry)?)
     }