diff --git a/Cargo.lock b/Cargo.lock
index 55f82ec..9f66aa0 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7455,7 +7455,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-api"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "anyhow",
  "axum",
@@ -7506,7 +7506,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-auth"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "anyhow",
  "base32",
@@ -7529,7 +7529,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-cache"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "async-trait",
  "base64 0.22.1",
@@ -7543,7 +7543,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-comms"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "async-trait",
  "base64 0.22.1",
@@ -7561,7 +7561,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-config"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "confique",
  "serde",
@@ -7569,7 +7569,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-crypto"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "aes-gcm",
  "base64 0.22.1",
@@ -7585,7 +7585,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-db"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "async-trait",
  "chrono",
@@ -7602,7 +7602,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-db-traits"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "async-trait",
  "base64 0.22.1",
@@ -7618,7 +7618,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-infra"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "async-trait",
  "bytes",
@@ -7629,7 +7629,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-lexicon"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "chrono",
  "futures",
@@ -7648,7 +7648,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-oauth"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "anyhow",
  "axum",
@@ -7671,7 +7671,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-oauth-server"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "axum",
  "base64 0.22.1",
@@ -7704,7 +7704,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-pds"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "aes-gcm",
  "anyhow",
@@ -7796,7 +7796,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-repo"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "bytes",
  "cid",
@@ -7808,7 +7808,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-ripple"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "async-trait",
  "backon",
@@ -7833,7 +7833,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-scopes"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "axum",
  "futures",
@@ -7849,7 +7849,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-server"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "axum",
  "clap",
@@ -7870,7 +7870,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-signal"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "async-trait",
  "chrono",
@@ -7893,7 +7893,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-storage"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "async-trait",
  "aws-config",
@@ -7910,7 +7910,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-store"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "async-trait",
  "bytes",
@@ -7959,7 +7959,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-sync"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "anyhow",
  "axum",
@@ -7981,7 +7981,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-types"
-version = "0.5.6"
+version = "0.5.7"
 dependencies = [
  "chrono",
  "cid",
diff --git a/Cargo.toml b/Cargo.toml
index fcb7bcc..d4d4345 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -26,7 +26,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.5.6"
+version = "0.5.7"
 edition = "2024"
 license = "AGPL-3.0-or-later"
 
diff --git a/crates/tranquil-api/src/repo/record/batch.rs b/crates/tranquil-api/src/repo/record/batch.rs
index 299d7e0..e91a090 100644
--- a/crates/tranquil-api/src/repo/record/batch.rs
+++ b/crates/tranquil-api/src/repo/record/batch.rs
@@ -27,7 +27,6 @@ struct WriteAccumulator {
     mst: Mst<TrackingBlockStore>,
     results: Vec<WriteResult>,
     ops: Vec<RecordOp>,
-    modified_keys: Vec<String>,
     all_blob_cids: Vec<String>,
     backlinks_to_add: Vec<Backlink>,
     backlinks_to_remove: Vec<AtUri>,
@@ -44,7 +43,6 @@ async fn process_single_write(
         mst,
         mut results,
         mut ops,
-        mut modified_keys,
         mut all_blob_cids,
         mut backlinks_to_add,
         mut backlinks_to_remove,
@@ -69,8 +67,19 @@ async fn process_single_write(
                     .await?,
                 )
             };
-            all_blob_cids.extend(extract_blob_cids(value));
             let rkey = rkey.clone().unwrap_or_else(Rkey::generate);
+            let key = format!("{}/{}", collection, rkey);
+            if mst
+                .get(&key)
+                .await
+                .map_err(|e| ApiError::InternalError(Some(format!("Failed to read MST: {e}"))))?
+                .is_some()
+            {
+                return Err(ApiError::InvalidRequest(format!(
+                    "Record already exists at {key}"
+                )));
+            }
+            all_blob_cids.extend(extract_blob_cids(value));
             let record_ipld = tranquil_pds::util::json_to_ipld(value);
             let record_bytes = serde_ipld_dagcbor::to_vec(&record_ipld)
                 .map_err(|_| ApiError::InvalidRecord("Failed to serialize record".into()))?;
@@ -78,8 +87,6 @@ async fn process_single_write(
                 .put(&record_bytes)
                 .await
                 .map_err(|_| ApiError::InternalError(Some("Failed to store record".into())))?;
-            let key = format!("{}/{}", collection, rkey);
-            modified_keys.push(key.clone());
             let new_mst = mst
                 .add(&key, record_cid)
                 .await
@@ -100,7 +107,6 @@ async fn process_single_write(
                 mst: new_mst,
                 results,
                 ops,
-                modified_keys,
                 all_blob_cids,
                 backlinks_to_add,
                 backlinks_to_remove,
@@ -124,16 +130,7 @@ async fn process_single_write(
                     .await?,
                 )
             };
-            all_blob_cids.extend(extract_blob_cids(value));
-            let record_ipld = tranquil_pds::util::json_to_ipld(value);
-            let record_bytes = serde_ipld_dagcbor::to_vec(&record_ipld)
-                .map_err(|_| ApiError::InvalidRecord("Failed to serialize record".into()))?;
-            let record_cid = tracking_store
-                .put(&record_bytes)
-                .await
-                .map_err(|_| ApiError::InternalError(Some("Failed to store record".into())))?;
             let key = format!("{}/{}", collection, rkey);
-            modified_keys.push(key.clone());
             let prev_record_cid = mst
                 .get(&key)
                 .await
@@ -143,6 +140,14 @@ async fn process_single_write(
                 .ok_or_else(|| {
                     ApiError::InvalidRequest("Update target record does not exist".into())
                 })?;
+            all_blob_cids.extend(extract_blob_cids(value));
+            let record_ipld = tranquil_pds::util::json_to_ipld(value);
+            let record_bytes = serde_ipld_dagcbor::to_vec(&record_ipld)
+                .map_err(|_| ApiError::InvalidRecord("Failed to serialize record".into()))?;
+            let record_cid = tracking_store
+                .put(&record_bytes)
+                .await
+                .map_err(|_| ApiError::InternalError(Some("Failed to store record".into())))?;
             let new_mst = mst
                 .update(&key, record_cid)
                 .await
@@ -165,7 +170,6 @@ async fn process_single_write(
                 mst: new_mst,
                 results,
                 ops,
-                modified_keys,
                 all_blob_cids,
                 backlinks_to_add,
                 backlinks_to_remove,
@@ -173,7 +177,6 @@ async fn process_single_write(
         }
         WriteOp::Delete { collection, rkey } => {
             let key = format!("{}/{}", collection, rkey);
-            modified_keys.push(key.clone());
             let prev_record_cid = mst
                 .get(&key)
                 .await
@@ -198,7 +201,6 @@ async fn process_single_write(
                 mst: new_mst,
                 results,
                 ops,
-                modified_keys,
                 all_blob_cids,
                 backlinks_to_add,
                 backlinks_to_remove,
@@ -219,7 +221,6 @@ async fn process_writes(
         mst: initial_mst,
         results: Vec::new(),
         ops: Vec::new(),
-        modified_keys: Vec::new(),
         all_blob_cids: Vec::new(),
         backlinks_to_add: Vec::new(),
         backlinks_to_remove: Vec::new(),
@@ -351,7 +352,6 @@ pub async fn apply_writes(
         mst: final_mst,
         results,
         ops,
-        modified_keys,
         all_blob_cids,
         backlinks_to_add,
         backlinks_to_remove,
@@ -407,7 +407,6 @@ pub async fn apply_writes(
             controller_did: controller_did.as_ref(),
             delegation_detail: write_summary,
             ops,
-            modified_keys: &modified_keys,
             blob_cids: &all_blob_cids,
             backlinks_to_add,
             backlinks_to_remove,
diff --git a/crates/tranquil-api/src/repo/record/delete.rs b/crates/tranquil-api/src/repo/record/delete.rs
index 6ffd57e..00b39e4 100644
--- a/crates/tranquil-api/src/repo/record/delete.rs
+++ b/crates/tranquil-api/src/repo/record/delete.rs
@@ -74,7 +74,6 @@ pub async fn delete_record(
         prev: RecordCid::from(prev_record_cid),
     };
 
-    let modified_keys = [key];
     let deleted_uri = AtUri::from_parts(&did, &input.collection, &input.rkey);
 
     let commit_result = finalize_repo_write(
@@ -93,7 +92,6 @@ pub async fn delete_record(
                 })
             }),
             ops: vec![op],
-            modified_keys: &modified_keys,
             blob_cids: &[],
             backlinks_to_add: vec![],
             backlinks_to_remove: vec![deleted_uri],
diff --git a/crates/tranquil-api/src/repo/record/write.rs b/crates/tranquil-api/src/repo/record/write.rs
index 64724fd..2a1624f 100644
--- a/crates/tranquil-api/src/repo/record/write.rs
+++ b/crates/tranquil-api/src/repo/record/write.rs
@@ -179,6 +179,18 @@ pub async fn create_record(
         }
     }
 
+    let key = format!("{}/{}", input.collection, rkey);
+    if mst
+        .get(&key)
+        .await
+        .map_err(|e| ApiError::InternalError(Some(format!("Failed to read MST: {e}"))))?
+        .is_some()
+    {
+        return Err(ApiError::InvalidRequest(format!(
+            "Record already exists at {key}"
+        )));
+    }
+
     let record_ipld = tranquil_pds::util::json_to_ipld(&input.record);
     let record_bytes = serde_ipld_dagcbor::to_vec(&record_ipld)
         .map_err(|_| ApiError::InvalidRecord("Failed to serialize record".into()))?;
@@ -187,8 +199,6 @@ pub async fn create_record(
         .put(&record_bytes)
         .await
         .map_err(|_| ApiError::InternalError(Some("Failed to save record block".into())))?;
-
-    let key = format!("{}/{}", input.collection, rkey);
     mst = mst
         .add(&key, record_cid)
         .await
@@ -200,20 +210,6 @@ pub async fn create_record(
         cid: tranquil_pds::cid_types::RecordCid::from(record_cid),
     });
 
-    let modified_keys: Vec<String> = ops
-        .iter()
-        .map(|op| match op {
-            RecordOp::Create {
-                collection, rkey, ..
-            }
-            | RecordOp::Update {
-                collection, rkey, ..
-            }
-            | RecordOp::Delete {
-                collection, rkey, ..
-            } => format!("{}/{}", collection, rkey),
-        })
-        .collect();
     let blob_cids = extract_blob_cids(&input.record);
 
     let created_uri = AtUri::from_parts(&did, &input.collection, &rkey);
@@ -235,7 +231,6 @@ pub async fn create_record(
                 })
             }),
             ops,
-            modified_keys: &modified_keys,
             blob_cids: &blob_cids,
             backlinks_to_add,
             backlinks_to_remove: conflict_uris_to_cleanup,
@@ -367,7 +362,6 @@ pub async fn put_record(
         }
     };
 
-    let modified_keys = [key];
     let blob_cids = extract_blob_cids(&input.record);
     let backlinks_to_add = extract_backlinks(&record_uri, &input.record);
 
@@ -387,7 +381,6 @@ pub async fn put_record(
                 })
             }),
             ops: vec![op],
-            modified_keys: &modified_keys,
             blob_cids: &blob_cids,
             backlinks_to_add,
             backlinks_to_remove,
diff --git a/crates/tranquil-config/src/lib.rs b/crates/tranquil-config/src/lib.rs
index 983de85..601898a 100644
--- a/crates/tranquil-config/src/lib.rs
+++ b/crates/tranquil-config/src/lib.rs
@@ -725,10 +725,6 @@ pub struct FirehoseConfig {
     #[config(env = "FIREHOSE_BACKFILL_HOURS", default = 72)]
     pub backfill_hours: i64,
 
-    /// Maximum number of lagged events before disconnecting a slow consumer.
-    #[config(env = "FIREHOSE_MAX_LAG", default = 5000)]
-    pub max_lag: u64,
-
     /// Maximum concurrent full-repo exports, eg. getRepo without `since`.
     #[config(env = "MAX_CONCURRENT_REPO_EXPORTS", default = 4)]
     pub max_concurrent_repo_exports: usize,
diff --git a/crates/tranquil-db-traits/src/backlink.rs b/crates/tranquil-db-traits/src/backlink.rs
index 6e0cf36..7dbe5ff 100644
--- a/crates/tranquil-db-traits/src/backlink.rs
+++ b/crates/tranquil-db-traits/src/backlink.rs
@@ -7,7 +7,7 @@ use uuid::Uuid;
 
 use crate::DbError;
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum BacklinkPath {
     Subject,
     SubjectUri,
diff --git a/crates/tranquil-pds/src/repo_ops.rs b/crates/tranquil-pds/src/repo_ops.rs
index 5845788..eec7ad2 100644
--- a/crates/tranquil-pds/src/repo_ops.rs
+++ b/crates/tranquil-pds/src/repo_ops.rs
@@ -14,7 +14,7 @@ use jacquard_repo::mst::{Mst, VerifiedWriteOp};
 use jacquard_repo::storage::BlockStore;
 use k256::ecdsa::SigningKey;
 use serde_json::{Value, json};
-use std::collections::{BTreeMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use std::str::FromStr;
 use std::sync::Arc;
 use tokio::sync::OwnedMutexGuard;
@@ -40,6 +40,7 @@ pub enum CommitError {
     MstOperationFailed(String),
     RecordSerializationFailed(String),
     InvalidCid(String),
+    RecordAlreadyExists(String),
 }
 
 impl std::fmt::Display for CommitError {
@@ -65,6 +66,7 @@ impl std::fmt::Display for CommitError {
                 write!(f, "Failed to serialize record: {}", e)
             }
             Self::InvalidCid(e) => write!(f, "Invalid CID: {}", e),
+            Self::RecordAlreadyExists(key) => write!(f, "Record already exists at {}", key),
         }
     }
 }
@@ -79,6 +81,9 @@ impl From<CommitError> for ApiError {
             }
             CommitError::RepoNotFound => ApiError::RepoNotFound(None),
             CommitError::UserNotFound => ApiError::RepoNotFound(Some("User not found".into())),
+            CommitError::RecordAlreadyExists(key) => {
+                ApiError::InvalidRequest(format!("Record already exists at {key}"))
+            }
             other => {
                 error!("Commit failed: {}", other);
                 ApiError::InternalError(Some("Failed to commit changes".into()))
@@ -162,7 +167,6 @@ pub struct FinalizeParams<'a> {
     pub controller_did: Option<&'a Did>,
     pub delegation_detail: Option<serde_json::Value>,
     pub ops: Vec<RecordOp>,
-    pub modified_keys: &'a [String],
     pub blob_cids: &'a [String],
     pub backlinks_to_add: Vec<Backlink>,
     pub backlinks_to_remove: Vec<AtUri>,
@@ -248,18 +252,8 @@ pub async fn finalize_repo_write(
     let mut inverse_trace = new_settled.clone();
     let mut non_invertible: Vec<String> = Vec::new();
     let mut invert_errors: Vec<String> = Vec::new();
-    for op in params.ops.iter() {
-        let (collection, rkey) = match op {
-            RecordOp::Create {
-                collection, rkey, ..
-            }
-            | RecordOp::Update {
-                collection, rkey, ..
-            }
-            | RecordOp::Delete {
-                collection, rkey, ..
-            } => (collection, rkey),
-        };
+    for op in params.ops.iter().rev() {
+        let (collection, rkey) = op.collection_rkey();
         let key = SmolStr::new(format!("{}/{}", collection, rkey));
         let verified = match op {
             RecordOp::Create { cid, .. } => VerifiedWriteOp::Create {
@@ -427,6 +421,22 @@ pub enum RecordOp {
     },
 }
 
+impl RecordOp {
+    pub fn collection_rkey(&self) -> (&Nsid, &Rkey) {
+        match self {
+            Self::Create {
+                collection, rkey, ..
+            }
+            | Self::Update {
+                collection, rkey, ..
+            }
+            | Self::Delete {
+                collection, rkey, ..
+            } => (collection, rkey),
+        }
+    }
+}
+
 pub struct CommitResult {
     pub commit_cid: Cid,
     pub rev: String,
@@ -457,8 +467,6 @@ pub async fn commit_and_log(
         RecordUpsert, RepoEventType,
     };
 
-    let backlinks_to_add = params.backlinks_to_add;
-    let backlinks_to_remove = params.backlinks_to_remove;
     let CommitParams {
         did,
         user_id,
@@ -471,7 +479,8 @@ pub async fn commit_and_log(
         new_tree_cids,
         blobs,
         obsolete_cids,
-        ..
+        backlinks_to_add,
+        backlinks_to_remove,
     } = params;
     debug_assert_eq!(
         current_root_cid.is_some(),
@@ -517,39 +526,65 @@ pub async fn commit_and_log(
 
     let obsolete_bytes: Vec<Vec<u8>> = obsolete_cids.iter().map(|c| c.to_bytes()).collect();
 
-    let (record_upserts, record_deletes): (Vec<RecordUpsert>, Vec<RecordDelete>) = ops.iter().fold(
-        (Vec::new(), Vec::new()),
-        |(mut upserts, mut deletes), op| {
-            match op {
-                RecordOp::Create {
-                    collection,
-                    rkey,
-                    cid,
-                }
-                | RecordOp::Update {
-                    collection,
-                    rkey,
-                    cid,
-                    ..
-                } => {
-                    upserts.push(RecordUpsert {
-                        collection: collection.clone(),
-                        rkey: rkey.clone(),
-                        cid: crate::types::CidLink::from(cid.as_cid()),
-                    });
-                }
-                RecordOp::Delete {
-                    collection, rkey, ..
-                } => {
-                    deletes.push(RecordDelete {
-                        collection: collection.clone(),
-                        rkey: rkey.clone(),
-                    });
-                }
+    let final_ops: HashMap<(&Nsid, &Rkey), &RecordOp> = ops
+        .iter()
+        .map(|op| (op.collection_rkey(), op))
+        .collect();
+
+    let final_record_uris: HashSet<AtUri> = final_ops
+        .iter()
+        .filter(|(_, op)| !matches!(op, RecordOp::Delete { .. }))
+        .map(|((c, r), _)| AtUri::from_parts(did, c, r))
+        .collect();
+
+    let record_upserts: Vec<RecordUpsert> = final_ops
+        .values()
+        .filter_map(|op| match op {
+            RecordOp::Create {
+                collection,
+                rkey,
+                cid,
             }
-            (upserts, deletes)
-        },
-    );
+            | RecordOp::Update {
+                collection,
+                rkey,
+                cid,
+                ..
+            } => Some(RecordUpsert {
+                collection: collection.clone(),
+                rkey: rkey.clone(),
+                cid: crate::types::CidLink::from(cid.as_cid()),
+            }),
+            RecordOp::Delete { .. } => None,
+        })
+        .collect();
+
+    let record_deletes: Vec<RecordDelete> = final_ops
+        .values()
+        .filter_map(|op| match op {
+            RecordOp::Delete {
+                collection, rkey, ..
+            } => Some(RecordDelete {
+                collection: collection.clone(),
+                rkey: rkey.clone(),
+            }),
+            _ => None,
+        })
+        .collect();
+
+    let backlinks_to_add: Vec<Backlink> = backlinks_to_add
+        .into_iter()
+        .filter(|b| final_record_uris.contains(&b.uri))
+        .map(|b| ((b.uri.clone(), b.path), b))
+        .collect::<HashMap<_, _>>()
+        .into_values()
+        .collect();
+
+    let backlinks_to_remove: Vec<AtUri> = backlinks_to_remove
+        .into_iter()
+        .collect::<HashSet<_>>()
+        .into_iter()
+        .collect();
 
     let ops_json: Vec<serde_json::Value> = ops
         .iter()
@@ -684,6 +719,16 @@ pub async fn create_record_internal(
         .await
         .map_err(to_commit_err)?;
 
+    let key = format!("{}/{}", collection, rkey);
+    if mst
+        .get(&key)
+        .await
+        .map_err(|e| CommitError::MstOperationFailed(e.to_string()))?
+        .is_some()
+    {
+        return Err(CommitError::RecordAlreadyExists(key));
+    }
+
     let record_ipld = crate::util::json_to_ipld(record);
     let mut record_bytes = Vec::new();
     serde_ipld_dagcbor::to_writer(&mut record_bytes, &record_ipld)
@@ -693,8 +738,6 @@ pub async fn create_record_internal(
         .put(&record_bytes)
         .await
         .map_err(|e| CommitError::BlockStoreFailed(e.to_string()))?;
-
-    let key = format!("{}/{}", collection, rkey);
     let new_mst = mst
         .add(&key, record_cid)
         .await
@@ -705,7 +748,6 @@ pub async fn create_record_internal(
         rkey: rkey.clone(),
         cid: RecordCid::from(record_cid),
     };
-    let modified_keys = [key];
     let blob_cids = extract_blob_cids(record);
     let record_uri = AtUri::from_parts(did.as_str(), collection.as_str(), rkey.as_str());
     let backlinks = extract_backlinks(&record_uri, record);
@@ -720,7 +762,6 @@ pub async fn create_record_internal(
             controller_did: None,
             delegation_detail: None,
             ops: vec![op],
-            modified_keys: &modified_keys,
             blob_cids: &blob_cids,
             backlinks_to_add: backlinks,
             backlinks_to_remove: vec![],
diff --git a/crates/tranquil-sync/src/subscribe_repos.rs b/crates/tranquil-sync/src/subscribe_repos.rs
index 2fb4b8e..006762a 100644
--- a/crates/tranquil-sync/src/subscribe_repos.rs
+++ b/crates/tranquil-sync/src/subscribe_repos.rs
@@ -48,6 +48,63 @@ pub fn get_subscriber_count() -> usize {
     SUBSCRIBER_COUNT.load(Ordering::SeqCst)
 }
 
+async fn recover_lagged_events(
+    socket: &mut WebSocket,
+    state: &AppState,
+    last_seen: &mut SequenceNumber,
+) -> Result<(), ()> {
+    if !last_seen.is_valid() {
+        *last_seen = state.repos.repo.get_max_seq().await.map_err(|e| {
+            error!("Lag recovery failed to read head sequence: {:?}", e);
+        })?;
+        return Ok(());
+    }
+    loop {
+        let events = match state
+            .repos
+            .repo
+            .get_events_since_cursor(*last_seen, BACKFILL_BATCH_SIZE)
+            .await
+        {
+            Ok(e) => e,
+            Err(e) => {
+                error!("Lag recovery DB query failed: {:?}", e);
+                return Err(());
+            }
+        };
+        if events.is_empty() {
+            return Ok(());
+        }
+        let batch_len = events.len();
+        let prefetched = match prefetch_blocks_for_events(state, &events).await {
+            Ok(b) => b,
+            Err(e) => {
+                error!("Lag recovery prefetch failed: {:?}", e);
+                return Err(());
+            }
+        };
+        for event in events {
+            *last_seen = event.seq;
+            let bytes =
+                match format_event_with_prefetched_blocks(state, event, &prefetched).await {
+                    Ok(b) => b,
+                    Err(e) => {
+                        warn!("Lag recovery format failed: {}", e);
+                        return Err(());
+                    }
+                };
+            if let Err(e) = socket.send(Message::Binary(bytes.into())).await {
+                warn!("Lag recovery send failed: {}", e);
+                return Err(());
+            }
+            tranquil_pds::metrics::record_firehose_event();
+        }
+        if batch_len < BACKFILL_BATCH_SIZE as usize {
+            return Ok(());
+        }
+    }
+}
+
 async fn handle_socket(mut socket: WebSocket, state: AppState, params: SubscribeReposParams) {
     let count = SUBSCRIBER_COUNT.fetch_add(1, Ordering::SeqCst) + 1;
     tranquil_pds::metrics::set_firehose_subscribers(count);
@@ -208,7 +265,6 @@ async fn handle_socket_inner(
             }
         }
     }
-    let max_lag_before_disconnect: u64 = tranquil_config::get().firehose.max_lag;
     loop {
         tokio::select! {
             result = rx.recv() => match result {
@@ -224,10 +280,9 @@ async fn handle_socket_inner(
                     tranquil_pds::metrics::record_firehose_event();
                 }
                 Err(RecvError::Lagged(skipped)) => {
-                    warn!(skipped = skipped, "Firehose subscriber lagged behind");
-                    if skipped > max_lag_before_disconnect {
-                        warn!(skipped = skipped, max_lag = max_lag_before_disconnect,
-                            "Disconnecting slow firehose consumer");
+                    warn!(skipped, last_seen = last_seen.as_i64(),
+                        "Firehose subscriber lagged, recovering missed events from DB");
+                    if let Err(()) = recover_lagged_events(socket, state, &mut last_seen).await {
                         break;
                     }
                 }
diff --git a/example.toml b/example.toml
index 5353aba..49e9138 100644
--- a/example.toml
+++ b/example.toml
@@ -348,13 +348,6 @@
 # Default value: 72
 #backfill_hours = 72
 
-# Maximum number of lagged events before disconnecting a slow consumer.
-#
-# Can also be specified via environment variable `FIREHOSE_MAX_LAG`.
-#
-# Default value: 5000
-#max_lag = 5000
-
 # Maximum concurrent full-repo exports, eg. getRepo without `since`.
 #
 # Can also be specified via environment variable `MAX_CONCURRENT_REPO_EXPORTS`.
diff --git a/justfile b/justfile
index 1a1ec5e..e06348b 100644
--- a/justfile
+++ b/justfile
@@ -32,19 +32,19 @@ gauntlet-nightly HOURS="6":
     SQLX_OFFLINE=true GAUNTLET_DURATION_HOURS={{HOURS}} cargo nextest run -p tranquil-store --features tranquil-store/test-harness --profile gauntlet-nightly --test gauntlet_smoke --run-ignored all
 
 gauntlet-farm SCENARIO HOURS="6" DUMP="proptest-regressions":
-    SQLX_OFFLINE=true cargo run --release -p tranquil-store --bin tranquil-gauntlet --features tranquil-store/gauntlet-cli -- farm --scenario {{SCENARIO}} --hours {{HOURS}} --dump-regressions {{DUMP}}
+    SQLX_OFFLINE=true cargo run --release --bin tranquil-gauntlet --features tranquil-store/gauntlet-cli -- farm --scenario {{SCENARIO}} --hours {{HOURS}} --dump-regressions {{DUMP}}
 
 gauntlet-repro SEED SCENARIO="smoke-pr":
-    SQLX_OFFLINE=true cargo run --release -p tranquil-store --bin tranquil-gauntlet --features tranquil-store/gauntlet-cli -- repro --scenario {{SCENARIO}} --seed {{SEED}}
+    SQLX_OFFLINE=true cargo run --release --bin tranquil-gauntlet --features tranquil-store/gauntlet-cli -- repro --scenario {{SCENARIO}} --seed {{SEED}}
 
 gauntlet-repro-config CONFIG SEED:
-    SQLX_OFFLINE=true cargo run --release -p tranquil-store --bin tranquil-gauntlet --features tranquil-store/gauntlet-cli -- repro --config {{CONFIG}} --seed {{SEED}}
+    SQLX_OFFLINE=true cargo run --release --bin tranquil-gauntlet --features tranquil-store/gauntlet-cli -- repro --config {{CONFIG}} --seed {{SEED}}
 
 gauntlet-repro-from FILE:
-    SQLX_OFFLINE=true cargo run --release -p tranquil-store --bin tranquil-gauntlet --features tranquil-store/gauntlet-cli -- repro --from {{FILE}}
+    SQLX_OFFLINE=true cargo run --release --bin tranquil-gauntlet --features tranquil-store/gauntlet-cli -- repro --from {{FILE}}
 
 gauntlet-sweep CONFIG SEEDS="8" DUMP="proptest-regressions":
-    SQLX_OFFLINE=true cargo run --release -p tranquil-store --bin tranquil-gauntlet --features tranquil-store/gauntlet-cli -- sweep --config {{CONFIG}} --seeds {{SEEDS}} --dump-regressions {{DUMP}}
+    SQLX_OFFLINE=true cargo run --release --bin tranquil-gauntlet --features tranquil-store/gauntlet-cli -- sweep --config {{CONFIG}} --seeds {{SEEDS}} --dump-regressions {{DUMP}}
 
 gauntlet-soak HOURS="24" OUTPUT="":
     SQLX_OFFLINE=true GAUNTLET_SOAK_HOURS={{HOURS}} GAUNTLET_SOAK_OUTPUT={{OUTPUT}} cargo nextest run -p tranquil-store --features tranquil-store/test-harness --profile gauntlet-soak --test gauntlet_soak --run-ignored all -- soak_long_leak_gate