feat(test): test that websockets get closed

Lewis: May this revision serve well! <lu5a@proton.me>

.tangled/workflows/publish-image.yml

@@ -1,24 +1,36 @@
 when:
-  - event: []
-    branch: []
+  - event: [ "manual" ]
+  - event: [ "push" ]
+    branch: [ "main" ]
 
 engine: nixery
 
 dependencies:
   nixpkgs:
-    - podman
+    - kaniko
+
+environment:
+  DOCKER_CONFIG: "/kaniko/.docker"
 
 steps:
-  - name: Create podman config
+  - name: Configure Kaniko
     command: |
-      mkdir -p ~/.config/containers
-      echo "unqualified-search-registries = [\"docker.io\"]" >> ~/.config/containers/registries.conf
+      mkdir -p /kaniko/.docker/
+      echo "{
+        \"auths\": {
+          \"https://atcr.io/v1\":{
+            \"auth\": \"$ATCR_CREDENTIALS\"
+          }
+        }
+      }" > /kaniko/.docker/config.json
 
   - name: Build image
     command: |
-      podman build . -t tranquil-pds:latest -t "tranquil-pds:$TANGLED_COMMIT_SHA"
-
-  - name: Publish image
-    command: |
-      podman push --creds "$ATCR_USERNAME:$ATCR_PASSWORD" tranquil-pds:latest "atcr.io/tranquil.farm/tranquil-pds:latest"
-      podman push --creds "$ATCR_USERNAME:$ATCR_PASSWORD" "tranquil-pds:$TANGLED_COMMIT_SHA" "atcr.io/tranquil.farm/tranquil-pds:$TANGLED_COMMIT_SHA"
+      executor \
+        --context=$(pwd) \
+        --ignore-path=$(pwd) \
+        --dockerfile=$(pwd)/Dockerfile \
+        --destination="atcr.io/tranquil.farm/tranquil-pds:latest" \
+        --destination="atcr.io/tranquil.farm/tranquil-pds:$TANGLED_COMMIT_SHA" \
+        --push-retry=3 \
+        --skip-push-permission-check

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"
 

Dockerfile

@@ -35,7 +35,7 @@ COPY crates/tranquil-oauth-server ./crates/tranquil-oauth-server
 COPY crates/tranquil-store ./crates/tranquil-store
 COPY crates/tranquil-signal ./crates/tranquil-signal
 COPY crates/tranquil-server ./crates/tranquil-server
-COPY migrations ./crates/tranquil-pds/migrations
+COPY migrations ./migrations
 RUN --mount=type=cache,target=/usr/local/cargo/registry \
     --mount=type=cache,target=/app/target \
     if [ "$SLIM" = "true" ]; then \
@@ -50,7 +50,6 @@ RUN apk add --no-cache msmtp ca-certificates \
     && ln -sf /usr/bin/msmtp /usr/sbin/sendmail
 COPY --from=builder /tmp/tranquil-pds /usr/local/bin/tranquil-pds
 COPY --from=frontend /app/dist /var/lib/tranquil-pds/frontend
-COPY migrations /app/migrations
 WORKDIR /app
 ENV SERVER_HOST=0.0.0.0
 ENV SERVER_PORT=3000

README.md

@@ -62,7 +62,6 @@ podman-compose -f docker-compose.prod.yaml up -d
 ### Installation Guides
 
 - [Nix](docs/install-nix.md)
-- [Debian](docs/install-debian.md)
 - [Containers](docs/install-containers.md)
 - [Kubernetes](docs/install-kubernetes.md)
 

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,

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],

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,

crates/tranquil-config/src/lib.rs

@@ -83,8 +83,9 @@ pub fn ensure_test_defaults() {
 ///
 /// Precedence (highest to lowest):
 /// 1. Environment variables
-/// 2. TOML config file (if provided)
-/// 3. Built-in defaults
+/// 2. Toml config file passed as `config_path`, if provided
+/// 3. `/etc/tranquil-pds/config.toml` - hardcoded fallback, silently skipped if absent
+/// 4. Built-in defaults
 pub fn load(config_path: Option<&PathBuf>) -> Result<TranquilConfig, confique::Error> {
     let mut builder = TranquilConfig::builder().env();
     if let Some(path) = config_path {
@@ -724,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,

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,

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![],

crates/tranquil-pds/tests/common/mod.rs

@@ -191,16 +191,12 @@ async fn setup_with_external_infra() -> String {
 async fn setup_with_testcontainers() -> String {
     let temp_dir = std::env::temp_dir().join(format!("tranquil-pds-test-{}", uuid::Uuid::new_v4()));
     let blob_path = temp_dir.join("blobs");
-    let backup_path = temp_dir.join("backups");
     std::fs::create_dir_all(&blob_path).expect("Failed to create blob temp directory");
-    std::fs::create_dir_all(&backup_path).expect("Failed to create backup temp directory");
     TEST_TEMP_DIR.set(temp_dir).ok();
     let plc_url = setup_mock_plc_directory().await;
     unsafe {
         std::env::set_var("BLOB_STORAGE_BACKEND", "filesystem");
         std::env::set_var("BLOB_STORAGE_PATH", blob_path.to_str().unwrap());
-        std::env::set_var("BACKUP_STORAGE_BACKEND", "filesystem");
-        std::env::set_var("BACKUP_STORAGE_PATH", backup_path.to_str().unwrap());
         std::env::set_var("MAX_IMPORT_SIZE", "100000000");
         std::env::set_var("SKIP_IMPORT_VERIFICATION", "true");
         std::env::set_var("PLC_DIRECTORY_URL", &plc_url);
@@ -242,8 +238,6 @@ async fn setup_with_testcontainers() -> String {
     let plc_url = setup_mock_plc_directory().await;
     unsafe {
         std::env::set_var("BLOB_STORAGE_BACKEND", "s3");
-        std::env::set_var("BACKUP_STORAGE_BACKEND", "s3");
-        std::env::set_var("BACKUP_S3_BUCKET", "test-backups");
         std::env::set_var("S3_BUCKET", "test-bucket");
         std::env::set_var("AWS_ACCESS_KEY_ID", "minioadmin");
         std::env::set_var("AWS_SECRET_ACCESS_KEY", "minioadmin");
@@ -333,8 +327,6 @@ unsafe fn configure_external_storage_env() {
         if std::env::var("S3_ENDPOINT").is_ok() {
             let s3_endpoint = std::env::var("S3_ENDPOINT").unwrap();
             std::env::set_var("BLOB_STORAGE_BACKEND", "s3");
-            std::env::set_var("BACKUP_STORAGE_BACKEND", "s3");
-            std::env::set_var("BACKUP_S3_BUCKET", "test-backups");
             std::env::set_var(
                 "S3_BUCKET",
                 std::env::var("S3_BUCKET").unwrap_or_else(|_| "test-bucket".to_string()),
@@ -356,14 +348,10 @@ unsafe fn configure_external_storage_env() {
             let process_dir =
                 std::env::temp_dir().join(format!("tranquil-pds-test-{}", std::process::id()));
             let blob_path = process_dir.join("blobs");
-            let backup_path = process_dir.join("backups");
             std::fs::create_dir_all(&blob_path).expect("Failed to create blob directory");
-            std::fs::create_dir_all(&backup_path).expect("Failed to create backup directory");
             TEST_TEMP_DIR.set(process_dir).ok();
             std::env::set_var("BLOB_STORAGE_BACKEND", "filesystem");
             std::env::set_var("BLOB_STORAGE_PATH", blob_path.to_str().unwrap());
-            std::env::set_var("BACKUP_STORAGE_BACKEND", "filesystem");
-            std::env::set_var("BACKUP_STORAGE_PATH", backup_path.to_str().unwrap());
         }
         std::env::set_var("MAX_IMPORT_SIZE", "100000000");
         std::env::set_var("SKIP_IMPORT_VERIFICATION", "true");
@@ -622,18 +610,14 @@ async fn setup_store_backend() -> String {
     let temp_dir =
         std::env::temp_dir().join(format!("tranquil-pds-store-{}", uuid::Uuid::new_v4()));
     let blob_path = temp_dir.join("blobs");
-    let backup_path = temp_dir.join("backups");
     let store_path = temp_dir.join("store");
     std::fs::create_dir_all(&blob_path).expect("failed to create blob temp directory");
-    std::fs::create_dir_all(&backup_path).expect("failed to create backup temp directory");
     std::fs::create_dir_all(&store_path).expect("failed to create store temp directory");
     TEST_TEMP_DIR.set(temp_dir).ok();
     let plc_url = setup_mock_plc_directory().await;
     unsafe {
         std::env::set_var("BLOB_STORAGE_BACKEND", "filesystem");
         std::env::set_var("BLOB_STORAGE_PATH", blob_path.to_str().unwrap());
-        std::env::set_var("BACKUP_STORAGE_BACKEND", "filesystem");
-        std::env::set_var("BACKUP_STORAGE_PATH", backup_path.to_str().unwrap());
         std::env::set_var("MAX_IMPORT_SIZE", "100000000");
         std::env::set_var("SKIP_IMPORT_VERIFICATION", "true");
         std::env::set_var("PLC_DIRECTORY_URL", &plc_url);
@@ -790,18 +774,14 @@ async fn setup_cluster_store_backend() -> Option<sqlx::PgPool> {
         uuid::Uuid::new_v4()
     ));
     let blob_path = temp_dir.join("blobs");
-    let backup_path = temp_dir.join("backups");
     let store_path = temp_dir.join("store");
     std::fs::create_dir_all(&blob_path).expect("failed to create blob temp directory");
-    std::fs::create_dir_all(&backup_path).expect("failed to create backup temp directory");
     std::fs::create_dir_all(&store_path).expect("failed to create store temp directory");
     TEST_TEMP_DIR.set(temp_dir).ok();
     let plc_url = setup_mock_plc_directory().await;
     unsafe {
         std::env::set_var("BLOB_STORAGE_BACKEND", "filesystem");
         std::env::set_var("BLOB_STORAGE_PATH", blob_path.to_str().unwrap());
-        std::env::set_var("BACKUP_STORAGE_BACKEND", "filesystem");
-        std::env::set_var("BACKUP_STORAGE_PATH", backup_path.to_str().unwrap());
         std::env::set_var("MAX_IMPORT_SIZE", "100000000");
         std::env::set_var("SKIP_IMPORT_VERIFICATION", "true");
         std::env::set_var("PLC_DIRECTORY_URL", &plc_url);
@@ -847,16 +827,12 @@ async fn setup_cluster_testcontainers() -> Option<sqlx::PgPool> {
     let temp_dir =
         std::env::temp_dir().join(format!("tranquil-pds-cluster-{}", uuid::Uuid::new_v4()));
     let blob_path = temp_dir.join("blobs");
-    let backup_path = temp_dir.join("backups");
     std::fs::create_dir_all(&blob_path).expect("Failed to create blob temp directory");
-    std::fs::create_dir_all(&backup_path).expect("Failed to create backup temp directory");
     TEST_TEMP_DIR.set(temp_dir).ok();
     let plc_url = setup_mock_plc_directory().await;
     unsafe {
         std::env::set_var("BLOB_STORAGE_BACKEND", "filesystem");
         std::env::set_var("BLOB_STORAGE_PATH", blob_path.to_str().unwrap());
-        std::env::set_var("BACKUP_STORAGE_BACKEND", "filesystem");
-        std::env::set_var("BACKUP_STORAGE_PATH", backup_path.to_str().unwrap());
         std::env::set_var("MAX_IMPORT_SIZE", "100000000");
         std::env::set_var("SKIP_IMPORT_VERIFICATION", "true");
         std::env::set_var("PLC_DIRECTORY_URL", &plc_url);

crates/tranquil-pds/tests/handle_domains.rs

@@ -8,7 +8,6 @@ const HANDLE_DOMAIN: &str = "handles.test";
 
 fn set_handle_domain() {
     unsafe {
-        std::env::set_var("AVAILABLE_USER_DOMAINS", HANDLE_DOMAIN);
         std::env::set_var("PDS_USER_HANDLE_DOMAINS", HANDLE_DOMAIN);
     }
 }

crates/tranquil-pds/tests/lifecycle_record.rs

@@ -456,7 +456,7 @@ async fn test_apply_writes_batch() {
         "writes": [
             { "$type": "com.atproto.repo.applyWrites#create", "collection": "app.bsky.feed.post", "rkey": "batch-post-1", "value": { "$type": "app.bsky.feed.post", "text": "First batch post", "createdAt": now } },
             { "$type": "com.atproto.repo.applyWrites#create", "collection": "app.bsky.feed.post", "rkey": "batch-post-2", "value": { "$type": "app.bsky.feed.post", "text": "Second batch post", "createdAt": now } },
-            { "$type": "com.atproto.repo.applyWrites#create", "collection": "app.bsky.actor.profile", "rkey": "self", "value": { "$type": "app.bsky.actor.profile", "displayName": "Batch User" } }
+            { "$type": "com.atproto.repo.applyWrites#update", "collection": "app.bsky.actor.profile", "rkey": "self", "value": { "$type": "app.bsky.actor.profile", "displayName": "Batch User" } }
         ]
     });
     let apply_res = client

crates/tranquil-pds/tests/mst_firehose_e2e.rs

@@ -0,0 +1,475 @@
+mod common;
+mod firehose;
+
+use std::collections::BTreeMap;
+use std::io::Cursor;
+use std::str::FromStr;
+use std::sync::Arc;
+use std::time::Duration;
+
+use bytes::Bytes;
+use cid::Cid;
+use common::*;
+use firehose::{FirehoseConsumer, ParsedCommitFrame};
+use iroh_car::CarReader;
+use jacquard_common::smol_str::SmolStr;
+use jacquard_repo::commit::Commit;
+use jacquard_repo::mst::{Mst, VerifiedWriteOp};
+use jacquard_repo::storage::{BlockStore, MemoryBlockStore};
+use reqwest::StatusCode;
+use serde_json::{Value, json};
+use tranquil_scopes::RepoAction;
+
+async fn car_to_blocks(car_bytes: &[u8]) -> BTreeMap<Cid, Bytes> {
+    let mut reader = CarReader::new(Cursor::new(car_bytes))
+        .await
+        .expect("parse CAR header");
+    let mut blocks = BTreeMap::new();
+    while let Ok(Some((cid, data))) = reader.next_block().await {
+        blocks.insert(cid, Bytes::from(data));
+    }
+    blocks
+}
+
+fn op_to_verified(op: &firehose::ParsedRepoOp) -> Result<VerifiedWriteOp, String> {
+    let key = SmolStr::new(&op.path);
+    match op.action {
+        RepoAction::Create => {
+            let cid = op.cid.ok_or("create op missing cid")?;
+            Ok(VerifiedWriteOp::Create { key, cid })
+        }
+        RepoAction::Update => {
+            let cid = op.cid.ok_or("update op missing cid")?;
+            let prev = op.prev.ok_or("update op missing prev")?;
+            Ok(VerifiedWriteOp::Update { key, cid, prev })
+        }
+        RepoAction::Delete => {
+            let prev = op.prev.ok_or("delete op missing prev")?;
+            Ok(VerifiedWriteOp::Delete { key, prev })
+        }
+    }
+}
+
+async fn verify_frame_forward(frame: &ParsedCommitFrame) -> Result<(), String> {
+    let prev_data = frame
+        .prev_data
+        .ok_or_else(|| "frame missing prev_data (v1.1 required)".to_string())?;
+
+    let blocks = car_to_blocks(&frame.blocks).await;
+    let storage = Arc::new(MemoryBlockStore::new_from_blocks(blocks));
+
+    let commit_bytes = storage
+        .get(&frame.commit)
+        .await
+        .map_err(|e| format!("get commit: {e:?}"))?
+        .ok_or_else(|| format!("CAR missing commit {}", frame.commit))?;
+    let commit = Commit::from_cbor(&commit_bytes).map_err(|e| format!("parse commit: {e:?}"))?;
+    let expected = *commit.data();
+
+    let mut mst = Mst::load(storage, prev_data, None);
+    for op in &frame.ops {
+        let path = &op.path;
+        match op.action {
+            RepoAction::Create | RepoAction::Update => {
+                let cid = op.cid.ok_or_else(|| format!("{path}: op missing cid"))?;
+                mst = mst
+                    .add(path, cid)
+                    .await
+                    .map_err(|e| format!("forward {path}: {e:?}"))?;
+            }
+            RepoAction::Delete => {
+                mst = mst
+                    .delete(path)
+                    .await
+                    .map_err(|e| format!("forward delete {path}: {e:?}"))?;
+            }
+        }
+    }
+    let computed = mst.persist().await.map_err(|e| format!("persist: {e:?}"))?;
+    if computed != expected {
+        return Err(format!(
+            "root mismatch expected={expected} computed={computed}"
+        ));
+    }
+    Ok(())
+}
+
+async fn verify_frame_inverse(frame: &ParsedCommitFrame) -> Result<(), String> {
+    let prev_data = frame
+        .prev_data
+        .ok_or_else(|| "frame missing prev_data (v1.1 required)".to_string())?;
+
+    let blocks = car_to_blocks(&frame.blocks).await;
+    let storage = Arc::new(MemoryBlockStore::new_from_blocks(blocks));
+
+    let commit_bytes = storage
+        .get(&frame.commit)
+        .await
+        .map_err(|e| format!("get commit: {e:?}"))?
+        .ok_or_else(|| format!("CAR missing commit {}", frame.commit))?;
+    let commit = Commit::from_cbor(&commit_bytes).map_err(|e| format!("parse commit: {e:?}"))?;
+    let new_data = *commit.data();
+
+    let mut mst = Mst::load(storage, new_data, None);
+    for op in &frame.ops {
+        let verified = op_to_verified(op)?;
+        let inverted = mst
+            .invert_op(verified.clone())
+            .await
+            .map_err(|e| format!("invert {verified:?}: {e:?}"))?;
+        if !inverted {
+            return Err(format!("op not invertible: {verified:?}"));
+        }
+    }
+    let computed_prev = mst
+        .get_pointer()
+        .await
+        .map_err(|e| format!("get_pointer: {e:?}"))?;
+    if computed_prev != prev_data {
+        return Err(format!(
+            "inverse root mismatch expected={prev_data} computed={computed_prev}"
+        ));
+    }
+    Ok(())
+}
+
+async fn create_record(client: &reqwest::Client, token: &str, did: &str, rkey: &str, text: &str) {
+    let now = chrono::Utc::now().to_rfc3339();
+    let res = client
+        .post(format!(
+            "{}/xrpc/com.atproto.repo.createRecord",
+            base_url().await
+        ))
+        .bearer_auth(token)
+        .json(&json!({
+            "repo": did,
+            "collection": "app.bsky.feed.post",
+            "rkey": rkey,
+            "record": {
+                "$type": "app.bsky.feed.post",
+                "text": text,
+                "createdAt": now,
+            }
+        }))
+        .send()
+        .await
+        .expect("createRecord");
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+async fn put_record(client: &reqwest::Client, token: &str, did: &str, rkey: &str, text: &str) {
+    let now = chrono::Utc::now().to_rfc3339();
+    let res = client
+        .post(format!(
+            "{}/xrpc/com.atproto.repo.putRecord",
+            base_url().await
+        ))
+        .bearer_auth(token)
+        .json(&json!({
+            "repo": did,
+            "collection": "app.bsky.feed.post",
+            "rkey": rkey,
+            "record": {
+                "$type": "app.bsky.feed.post",
+                "text": text,
+                "createdAt": now,
+            }
+        }))
+        .send()
+        .await
+        .expect("putRecord");
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+async fn delete_record(client: &reqwest::Client, token: &str, did: &str, rkey: &str) {
+    let res = client
+        .post(format!(
+            "{}/xrpc/com.atproto.repo.deleteRecord",
+            base_url().await
+        ))
+        .bearer_auth(token)
+        .json(&json!({
+            "repo": did,
+            "collection": "app.bsky.feed.post",
+            "rkey": rkey,
+        }))
+        .send()
+        .await
+        .expect("deleteRecord");
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+async fn apply_writes_batch(client: &reqwest::Client, token: &str, did: &str, writes: Vec<Value>) {
+    let res = client
+        .post(format!(
+            "{}/xrpc/com.atproto.repo.applyWrites",
+            base_url().await
+        ))
+        .bearer_auth(token)
+        .json(&json!({ "repo": did, "writes": writes }))
+        .send()
+        .await
+        .expect("applyWrites");
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+fn rkey_for(i: usize) -> String {
+    format!("3ke2e{:08}", i)
+}
+
+#[tokio::test]
+async fn websocket_firehose_frames_pass_inductive_forward_and_inverse() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+
+    let repos = get_test_repos().await;
+    let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
+    let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    let now = chrono::Utc::now().to_rfc3339();
+    let seed: Vec<Value> = (0..120)
+        .map(|i| {
+            json!({
+                "$type": "com.atproto.repo.applyWrites#create",
+                "collection": "app.bsky.feed.post",
+                "rkey": rkey_for(i),
+                "value": {
+                    "$type": "app.bsky.feed.post",
+                    "text": format!("e2e {i}"),
+                    "createdAt": now,
+                }
+            })
+        })
+        .collect();
+    for chunk in seed.chunks(40) {
+        apply_writes_batch(&client, &token, &did, chunk.to_vec()).await;
+    }
+    for i in (0..120).step_by(6) {
+        put_record(&client, &token, &did, &rkey_for(i), &format!("upd {i}")).await;
+    }
+    for i in (2..120).step_by(11) {
+        delete_record(&client, &token, &did, &rkey_for(i)).await;
+    }
+    create_record(&client, &token, &did, "3ke2efinal001", "final").await;
+
+    let target_commits = 3 + 20 + 11 + 1;
+    let frames = consumer
+        .wait_for_commits(&did, target_commits, Duration::from_secs(90))
+        .await;
+    assert!(
+        frames.len() >= target_commits,
+        "expected {} commit frames, got {}",
+        target_commits,
+        frames.len()
+    );
+
+    let mut forward_failures = Vec::new();
+    let mut inverse_failures = Vec::new();
+    for frame in &frames {
+        if frame.prev_data.is_none() {
+            continue;
+        }
+        if frame.ops.is_empty() {
+            continue;
+        }
+        if let Err(msg) = verify_frame_forward(frame).await {
+            forward_failures.push(format!("seq={}: {msg}", frame.seq));
+        }
+        if let Err(msg) = verify_frame_inverse(frame).await {
+            inverse_failures.push(format!("seq={}: {msg}", frame.seq));
+        }
+    }
+    assert!(
+        forward_failures.is_empty(),
+        "forward verification failures:\n  - {}",
+        forward_failures.join("\n  - ")
+    );
+    assert!(
+        inverse_failures.is_empty(),
+        "inverse verification failures:\n  - {}",
+        inverse_failures.join("\n  - ")
+    );
+}
+
+#[tokio::test]
+async fn websocket_firehose_car_root_matches_commit_cid() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+
+    let repos = get_test_repos().await;
+    let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
+    let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    for i in 0..4 {
+        create_record(&client, &token, &did, &rkey_for(i), "ck").await;
+    }
+
+    let frames = consumer
+        .wait_for_commits(&did, 4, Duration::from_secs(10))
+        .await;
+
+    for frame in &frames {
+        let mut reader = CarReader::new(Cursor::new(&frame.blocks))
+            .await
+            .expect("CAR header");
+        let roots = reader.header().roots();
+        assert_eq!(roots.len(), 1, "CAR must have exactly one root");
+        assert_eq!(
+            roots[0], frame.commit,
+            "CAR root must equal frame commit CID"
+        );
+        let mut found = false;
+        while let Ok(Some((cid, _))) = reader.next_block().await {
+            if cid == frame.commit {
+                found = true;
+            }
+        }
+        assert!(found, "CAR body must contain commit block");
+    }
+}
+
+#[tokio::test]
+async fn websocket_firehose_resumption_from_cursor_yields_valid_frames() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+    let repos = get_test_repos().await;
+
+    for i in 0..5 {
+        create_record(&client, &token, &did, &rkey_for(i), "pre").await;
+    }
+
+    let resume_cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
+
+    for i in 5..12 {
+        create_record(&client, &token, &did, &rkey_for(i), "post").await;
+    }
+
+    let consumer = FirehoseConsumer::connect_with_cursor(app_port(), resume_cursor).await;
+    let frames = consumer
+        .wait_for_commits(&did, 7, Duration::from_secs(20))
+        .await;
+    assert!(
+        frames.len() >= 7,
+        "expected 7+ frames after cursor resume, got {}",
+        frames.len()
+    );
+
+    for frame in &frames {
+        if frame.prev_data.is_none() || frame.ops.is_empty() {
+            continue;
+        }
+        verify_frame_forward(frame)
+            .await
+            .unwrap_or_else(|e| panic!("resumed frame seq={} invalid: {e}", frame.seq));
+    }
+}
+
+#[tokio::test]
+async fn websocket_firehose_ops_include_prev_field_for_update_delete() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+    let repos = get_test_repos().await;
+    let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
+    let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    create_record(&client, &token, &did, "3ke2eprev01", "v1").await;
+    put_record(&client, &token, &did, "3ke2eprev01", "v2").await;
+    delete_record(&client, &token, &did, "3ke2eprev01").await;
+
+    let frames = consumer
+        .wait_for_commits(&did, 3, Duration::from_secs(10))
+        .await;
+    assert!(frames.len() >= 3);
+
+    for frame in &frames {
+        for op in &frame.ops {
+            match op.action {
+                RepoAction::Create => {
+                    assert!(op.cid.is_some(), "create must have cid");
+                    assert!(op.prev.is_none(), "create must not have prev");
+                }
+                RepoAction::Update => {
+                    assert!(op.cid.is_some(), "update must have cid");
+                    assert!(
+                        op.prev.is_some(),
+                        "v1.1 update must carry prev CID (seq={})",
+                        frame.seq
+                    );
+                }
+                RepoAction::Delete => {
+                    assert!(op.cid.is_none(), "delete must have null cid");
+                    assert!(
+                        op.prev.is_some(),
+                        "v1.1 delete must carry prev CID (seq={})",
+                        frame.seq
+                    );
+                }
+            }
+        }
+    }
+}
+
+#[tokio::test]
+async fn websocket_firehose_rebuild_new_mst_from_car_matches_commit_data() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+    let repos = get_test_repos().await;
+    let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
+    let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    let now = chrono::Utc::now().to_rfc3339();
+    let writes: Vec<Value> = (0..30)
+        .map(|i| {
+            json!({
+                "$type": "com.atproto.repo.applyWrites#create",
+                "collection": "app.bsky.feed.post",
+                "rkey": rkey_for(i),
+                "value": {
+                    "$type": "app.bsky.feed.post",
+                    "text": format!("rb {i}"),
+                    "createdAt": now,
+                }
+            })
+        })
+        .collect();
+    apply_writes_batch(&client, &token, &did, writes).await;
+
+    let frames = consumer
+        .wait_for_commits(&did, 1, Duration::from_secs(10))
+        .await;
+    let last = frames.last().expect("frame");
+
+    let blocks = car_to_blocks(&last.blocks).await;
+    let storage = Arc::new(MemoryBlockStore::new_from_blocks(blocks));
+    let commit_bytes = storage
+        .get(&last.commit)
+        .await
+        .unwrap()
+        .expect("commit block");
+    let commit = Commit::from_cbor(&commit_bytes).unwrap();
+
+    let new_root_cid = *commit.data();
+    let mst = Mst::load(storage, new_root_cid, None);
+    let rehydrated_cid = mst.get_pointer().await.expect("rebuild mst");
+    assert_eq!(
+        rehydrated_cid, new_root_cid,
+        "MST loaded from CAR must yield same root as commit.data()"
+    );
+
+    for op in &last.ops {
+        if op.action == RepoAction::Create {
+            let expected_cid = op.cid.unwrap();
+            let got = mst
+                .get(&op.path)
+                .await
+                .expect("mst.get")
+                .unwrap_or_else(|| panic!("key {} missing from rebuilt tree", op.path));
+            assert_eq!(got, expected_cid, "record CID mismatch for {}", op.path);
+            let _ = Cid::from_str(&expected_cid.to_string()).unwrap();
+        }
+    }
+}

crates/tranquil-pds/tests/mst_inductive_firehose.rs

@@ -124,7 +124,7 @@ async fn verify_inductive_inverse(event: &SequencedEvent) -> Result<(Cid, Cid),
     let new_data_cid = new_commit_data_cid(&storage, &commit_cid).await?;
 
     let mut mst = Mst::load(storage.clone(), new_data_cid, None);
-    for op_value in ops_json(event)? {
+    for op_value in ops_json(event)?.iter().rev() {
         let verified = parse_op_to_verified(op_value)?;
         let inverted = mst
             .invert_op(verified.clone())
@@ -546,6 +546,63 @@ async fn inductive_inverse_verifies_every_commit() {
     report_failures(non_genesis.len(), &failures, "any inverse");
 }
 
+#[tokio::test]
+async fn inductive_inverse_handles_same_rkey_in_batch() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+
+    let now = chrono::Utc::now().to_rfc3339();
+    let rkey = rkey_for("dup", 0);
+    create_record(&client, &token, &did, COLLECTION, &rkey).await;
+
+    let writes = vec![
+        json!({
+            "$type": "com.atproto.repo.applyWrites#update",
+            "collection": COLLECTION,
+            "rkey": rkey,
+            "value": {
+                "$type": COLLECTION,
+                "text": "v1",
+                "createdAt": now,
+            }
+        }),
+        json!({
+            "$type": "com.atproto.repo.applyWrites#update",
+            "collection": COLLECTION,
+            "rkey": rkey,
+            "value": {
+                "$type": COLLECTION,
+                "text": "v2",
+                "createdAt": now,
+            }
+        }),
+    ];
+    apply_writes_batch(&client, &token, &did, writes).await;
+
+    let our = our_commit_events(&did).await;
+    let dup_event = our
+        .iter()
+        .find(|e| {
+            ops_json(e)
+                .map(|arr| {
+                    arr.iter()
+                        .filter(|op| op["action"].as_str() == Some("update"))
+                        .count()
+                        == 2
+                })
+                .unwrap_or(false)
+        })
+        .expect("commit event with two same-rkey updates");
+
+    let (exp, got) = verify_inductive_inverse(dup_event)
+        .await
+        .expect("inverse verify should succeed for same-rkey batch");
+    assert_eq!(
+        exp, got,
+        "inverse root mismatch for same-rkey batch: exp={exp} got={got}"
+    );
+}
+
 #[tokio::test]
 async fn prev_cid_chain_walks_to_genesis() {
     let client = client();

crates/tranquil-pds/tests/mst_integrity.rs

@@ -0,0 +1,334 @@
+mod common;
+mod firehose;
+mod helpers;
+
+use std::collections::BTreeMap;
+use std::io::Cursor;
+use std::sync::Arc;
+use std::time::Duration;
+
+use bytes::Bytes;
+use cid::Cid;
+use common::*;
+use firehose::FirehoseConsumer;
+use helpers::build_car_with_signature;
+use iroh_car::CarReader;
+use jacquard_repo::commit::Commit;
+use jacquard_repo::mst::Mst;
+use jacquard_repo::storage::{BlockStore, MemoryBlockStore};
+use k256::ecdsa::SigningKey;
+use reqwest::StatusCode;
+use serde_json::{Value, json};
+use tranquil_db_traits::{EventBlocks, RepoEventType, SequenceNumber, SequencedEvent};
+use tranquil_scopes::RepoAction;
+use tranquil_types::Did;
+
+async fn car_to_blocks(car_bytes: &[u8]) -> (Vec<Cid>, BTreeMap<Cid, Bytes>) {
+    let mut reader = CarReader::new(Cursor::new(car_bytes))
+        .await
+        .expect("parse CAR");
+    let roots = reader.header().roots().to_vec();
+    let mut blocks = BTreeMap::new();
+    while let Ok(Some((cid, data))) = reader.next_block().await {
+        blocks.insert(cid, Bytes::from(data));
+    }
+    (roots, blocks)
+}
+
+async fn create_post(client: &reqwest::Client, token: &str, did: &str, rkey: &str, text: &str) {
+    let now = chrono::Utc::now().to_rfc3339();
+    let res = client
+        .post(format!(
+            "{}/xrpc/com.atproto.repo.createRecord",
+            base_url().await
+        ))
+        .bearer_auth(token)
+        .json(&json!({
+            "repo": did,
+            "collection": "app.bsky.feed.post",
+            "rkey": rkey,
+            "record": {
+                "$type": "app.bsky.feed.post",
+                "text": text,
+                "createdAt": now,
+            }
+        }))
+        .send()
+        .await
+        .expect("createRecord");
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+#[tokio::test]
+async fn getrepo_car_roundtrips_mst_structure_and_records() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+
+    let expected_records: Vec<(String, String)> = (0..20)
+        .map(|i| {
+            let rkey = format!("3krtp{:08}", i);
+            let text = format!("roundtrip record {i}");
+            (rkey, text)
+        })
+        .collect();
+    for (rkey, text) in &expected_records {
+        create_post(&client, &token, &did, rkey, text).await;
+    }
+
+    let res = client
+        .get(format!(
+            "{}/xrpc/com.atproto.sync.getRepo",
+            base_url().await
+        ))
+        .query(&[("did", did.as_str())])
+        .send()
+        .await
+        .expect("getRepo");
+    assert_eq!(res.status(), StatusCode::OK);
+    let car_bytes = res.bytes().await.unwrap();
+
+    let (roots, block_map) = car_to_blocks(&car_bytes).await;
+    assert_eq!(roots.len(), 1, "CAR must have exactly one root");
+    let commit_cid = roots[0];
+    let storage = Arc::new(MemoryBlockStore::new_from_blocks(block_map));
+
+    let commit_bytes = storage
+        .get(&commit_cid)
+        .await
+        .unwrap()
+        .expect("CAR contains commit block");
+    let commit = Commit::from_cbor(&commit_bytes).expect("parse commit");
+    let data_cid = *commit.data();
+
+    let mst = Mst::load(storage.clone(), data_cid, None);
+    let loaded_root = mst.get_pointer().await.expect("load root");
+    assert_eq!(loaded_root, data_cid, "loaded MST pointer == commit.data()");
+
+    for (rkey, _) in &expected_records {
+        let path = format!("app.bsky.feed.post/{rkey}");
+        let leaf = mst
+            .get(&path)
+            .await
+            .expect("mst.get")
+            .unwrap_or_else(|| panic!("record {path} missing from exported MST"));
+        let leaf_bytes = storage
+            .get(&leaf)
+            .await
+            .unwrap()
+            .unwrap_or_else(|| panic!("record block {leaf} missing from CAR"));
+        assert!(!leaf_bytes.is_empty(), "record bytes empty");
+    }
+}
+
+#[tokio::test]
+async fn concurrent_swap_commit_writes_serialize() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+
+    create_post(&client, &token, &did, "3kswap00000001", "anchor").await;
+
+    let latest_res = client
+        .get(format!(
+            "{}/xrpc/com.atproto.sync.getLatestCommit",
+            base_url().await
+        ))
+        .query(&[("did", did.as_str())])
+        .send()
+        .await
+        .expect("getLatestCommit");
+    assert_eq!(latest_res.status(), StatusCode::OK);
+    let latest: Value = latest_res.json().await.unwrap();
+    let swap_cid = latest["cid"].as_str().unwrap().to_string();
+
+    let now = chrono::Utc::now().to_rfc3339();
+    let payload_a = json!({
+        "repo": did,
+        "collection": "app.bsky.feed.post",
+        "rkey": "3kswap00000002",
+        "record": {
+            "$type": "app.bsky.feed.post",
+            "text": "writer A",
+            "createdAt": now,
+        },
+        "swapCommit": swap_cid,
+    });
+    let payload_b = json!({
+        "repo": did,
+        "collection": "app.bsky.feed.post",
+        "rkey": "3kswap00000003",
+        "record": {
+            "$type": "app.bsky.feed.post",
+            "text": "writer B",
+            "createdAt": now,
+        },
+        "swapCommit": swap_cid,
+    });
+
+    let base = base_url().await;
+    let (res_a, res_b) = tokio::join!(
+        client
+            .post(format!("{base}/xrpc/com.atproto.repo.putRecord"))
+            .bearer_auth(&token)
+            .json(&payload_a)
+            .send(),
+        client
+            .post(format!("{base}/xrpc/com.atproto.repo.putRecord"))
+            .bearer_auth(&token)
+            .json(&payload_b)
+            .send(),
+    );
+    let status_a = res_a.expect("A send").status();
+    let status_b = res_b.expect("B send").status();
+
+    let ok_a = status_a == StatusCode::OK;
+    let ok_b = status_b == StatusCode::OK;
+    assert!(
+        ok_a ^ ok_b,
+        "exactly one swap_commit write must succeed: status_a={status_a}, status_b={status_b}"
+    );
+}
+
+#[tokio::test]
+async fn imported_repo_emits_commit_event_with_valid_car() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+
+    let signing_key = SigningKey::random(&mut rand::thread_rng());
+    let (car_bytes, _car_root_cid) = build_car_with_signature(&did, &signing_key);
+
+    let import_res = client
+        .post(format!(
+            "{}/xrpc/com.atproto.repo.importRepo",
+            base_url().await
+        ))
+        .bearer_auth(&token)
+        .header("Content-Type", "application/vnd.ipld.car")
+        .body(car_bytes)
+        .send()
+        .await
+        .expect("importRepo");
+    assert_eq!(
+        import_res.status(),
+        StatusCode::OK,
+        "import failed: {:?}",
+        import_res.text().await.unwrap_or_default()
+    );
+
+    let repos = get_test_repos().await;
+    let typed_did = Did::new(did.clone()).unwrap();
+    let events = repos
+        .repo
+        .get_events_since_seq(SequenceNumber::ZERO, None)
+        .await
+        .expect("events");
+    let our: Vec<&SequencedEvent> = events
+        .iter()
+        .filter(|e| e.did == typed_did && e.event_type == RepoEventType::Commit)
+        .collect();
+    let last = our.last().expect("at least one commit event after import");
+
+    let inline = match last.blocks.as_ref().expect("blocks present") {
+        EventBlocks::Inline(v) => v,
+        _ => panic!("expected inline blocks"),
+    };
+    assert!(
+        !inline.is_empty(),
+        "import event inline blocks must not be empty"
+    );
+
+    let have_commit = inline.iter().any(|b| {
+        let cid = Cid::read_bytes(b.cid_bytes.as_slice()).unwrap();
+        last.commit_cid
+            .as_ref()
+            .and_then(|c| c.to_cid())
+            .map(|commit_cid| cid == commit_cid)
+            .unwrap_or(false)
+    });
+    assert!(have_commit, "import event CAR must include commit block");
+}
+
+#[tokio::test]
+async fn firehose_commit_block_bytes_roundtrip_to_same_cid() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+
+    let repos = get_test_repos().await;
+    let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
+    let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    create_post(&client, &token, &did, "3krt001", "round-trip me").await;
+    let frames = consumer
+        .wait_for_commits(&did, 1, Duration::from_secs(10))
+        .await;
+    let frame = frames.last().expect("frame");
+
+    let (_, block_map) = car_to_blocks(&frame.blocks).await;
+    use sha2::{Digest, Sha256};
+    for (cid, bytes) in &block_map {
+        let mut hasher = Sha256::new();
+        hasher.update(bytes);
+        let hash = hasher.finalize();
+        let mh = multihash::Multihash::wrap(0x12, hash.as_slice()).expect("wrap");
+        let recomputed = Cid::new_v1(cid.codec(), mh);
+        assert_eq!(
+            recomputed, *cid,
+            "CAR block {cid} bytes do not hash back to same CID"
+        );
+    }
+}
+
+#[tokio::test]
+async fn firehose_commit_car_contains_new_record_bytes_for_every_create() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+
+    let repos = get_test_repos().await;
+    let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
+    let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    let now = chrono::Utc::now().to_rfc3339();
+    let writes: Vec<Value> = (0..8)
+        .map(|i| {
+            json!({
+                "$type": "com.atproto.repo.applyWrites#create",
+                "collection": "app.bsky.feed.post",
+                "rkey": format!("3krec{:08}", i),
+                "value": {
+                    "$type": "app.bsky.feed.post",
+                    "text": format!("rec {i}"),
+                    "createdAt": now,
+                }
+            })
+        })
+        .collect();
+    let res = client
+        .post(format!(
+            "{}/xrpc/com.atproto.repo.applyWrites",
+            base_url().await
+        ))
+        .bearer_auth(&token)
+        .json(&json!({ "repo": did, "writes": writes }))
+        .send()
+        .await
+        .expect("applyWrites");
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let frames = consumer
+        .wait_for_commits(&did, 1, Duration::from_secs(10))
+        .await;
+    let frame = frames.last().expect("frame");
+
+    let (_, block_map) = car_to_blocks(&frame.blocks).await;
+    for op in &frame.ops {
+        if op.action == RepoAction::Create {
+            let cid = op.cid.expect("create cid");
+            assert!(
+                block_map.contains_key(&cid),
+                "record CID {cid} for path {} missing from CAR",
+                op.path
+            );
+        }
+    }
+}

crates/tranquil-pds/tests/mst_property_fuzz.rs

@@ -0,0 +1,348 @@
+mod common;
+mod mst_verify;
+
+use std::collections::HashMap;
+use std::str::FromStr;
+
+use cid::Cid;
+use common::*;
+use jacquard_common::smol_str::SmolStr;
+use jacquard_repo::commit::Commit;
+use jacquard_repo::mst::{Mst, VerifiedWriteOp};
+use jacquard_repo::storage::BlockStore;
+use mst_verify::{extract_event_blocks, inline_to_store};
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
+use reqwest::StatusCode;
+use serde_json::{Value, json};
+use tranquil_db_traits::{RepoEventType, SequenceNumber, SequencedEvent};
+use tranquil_types::Did;
+
+const COLLECTIONS: &[&str] = &[
+    "app.bsky.feed.post",
+    "app.bsky.feed.like",
+    "app.bsky.graph.follow",
+    "app.bsky.feed.repost",
+];
+
+#[derive(Copy, Clone, Debug)]
+enum FuzzOp {
+    Create,
+    Update,
+    Delete,
+}
+
+fn pick_op(rng: &mut StdRng, have_keys: bool) -> FuzzOp {
+    match (have_keys, rng.gen_range(0..10)) {
+        (false, _) => FuzzOp::Create,
+        (_, 0..=5) => FuzzOp::Create,
+        (_, 6..=7) => FuzzOp::Update,
+        _ => FuzzOp::Delete,
+    }
+}
+
+fn random_rkey(rng: &mut StdRng) -> String {
+    let tid_char_pool = b"234567abcdefghijklmnopqrstuvwxyz";
+    let mut out = Vec::with_capacity(13);
+    (0..13).for_each(|_| {
+        let c = tid_char_pool[rng.gen_range(0..tid_char_pool.len())];
+        out.push(c);
+    });
+    String::from_utf8(out).unwrap()
+}
+
+fn random_collection(rng: &mut StdRng) -> &'static str {
+    COLLECTIONS[rng.gen_range(0..COLLECTIONS.len())]
+}
+
+fn record_for_collection(col: &str, text: &str, now: &str) -> Value {
+    match col {
+        "app.bsky.feed.post" | "app.bsky.feed.repost" | "app.bsky.feed.like" => json!({
+            "$type": col,
+            "text": text,
+            "createdAt": now,
+        }),
+        _ => json!({
+            "$type": col,
+            "subject": format!("did:plc:synthetic{text}"),
+            "createdAt": now,
+        }),
+    }
+}
+
+async fn verify_commit_forward_and_inverse(event: &SequencedEvent) -> Result<(), String> {
+    let prev_data = event
+        .prev_data_cid
+        .as_ref()
+        .and_then(|c| c.to_cid())
+        .ok_or("no prev_data_cid")?;
+    let commit_cid = event
+        .commit_cid
+        .as_ref()
+        .and_then(|c| c.to_cid())
+        .ok_or("no commit_cid")?;
+    let inline = extract_event_blocks(event)?;
+    let ops = event
+        .ops
+        .as_ref()
+        .and_then(|v| v.as_array())
+        .ok_or("ops not array")?;
+
+    let storage = inline_to_store(inline);
+    let commit_bytes = storage
+        .get(&commit_cid)
+        .await
+        .map_err(|e| format!("get commit: {e:?}"))?
+        .ok_or("missing commit block")?;
+    let commit = Commit::from_cbor(&commit_bytes).map_err(|e| format!("parse commit: {e:?}"))?;
+    let new_data = *commit.data();
+
+    let mut forward = Mst::load(storage.clone(), prev_data, None);
+    for op in ops {
+        let action = op["action"].as_str().ok_or("op.action")?;
+        let path = op["path"].as_str().ok_or("op.path")?;
+        match action {
+            "create" | "update" => {
+                let cid = Cid::from_str(op["cid"].as_str().ok_or("op.cid")?)
+                    .map_err(|e| format!("{e:?}"))?;
+                forward = forward
+                    .add(path, cid)
+                    .await
+                    .map_err(|e| format!("fwd add {path}: {e:?}"))?;
+            }
+            "delete" => {
+                forward = forward
+                    .delete(path)
+                    .await
+                    .map_err(|e| format!("fwd delete {path}: {e:?}"))?;
+            }
+            other => return Err(format!("unknown action {other}")),
+        }
+    }
+    let got = forward
+        .persist()
+        .await
+        .map_err(|e| format!("persist: {e:?}"))?;
+    if got != new_data {
+        return Err(format!("forward root mismatch exp={new_data} got={got}"));
+    }
+
+    let mut inverse = Mst::load(storage, new_data, None);
+    for op in ops {
+        let action = op["action"].as_str().ok_or("op.action")?;
+        let path = op["path"].as_str().ok_or("op.path")?;
+        let key = SmolStr::new(path);
+        let verified = match action {
+            "create" => {
+                let cid = Cid::from_str(op["cid"].as_str().ok_or("op.cid")?)
+                    .map_err(|e| format!("{e:?}"))?;
+                VerifiedWriteOp::Create { key, cid }
+            }
+            "update" => {
+                let cid = Cid::from_str(op["cid"].as_str().ok_or("op.cid")?)
+                    .map_err(|e| format!("{e:?}"))?;
+                let prev = Cid::from_str(op["prev"].as_str().ok_or("op.prev")?)
+                    .map_err(|e| format!("{e:?}"))?;
+                VerifiedWriteOp::Update { key, cid, prev }
+            }
+            "delete" => {
+                let prev = Cid::from_str(op["prev"].as_str().ok_or("op.prev")?)
+                    .map_err(|e| format!("{e:?}"))?;
+                VerifiedWriteOp::Delete { key, prev }
+            }
+            other => return Err(format!("unknown action {other}")),
+        };
+        let inverted = inverse
+            .invert_op(verified.clone())
+            .await
+            .map_err(|e| format!("invert {verified:?}: {e:?}"))?;
+        if !inverted {
+            return Err(format!("op not invertible: {verified:?}"));
+        }
+    }
+    let got_prev = inverse
+        .get_pointer()
+        .await
+        .map_err(|e| format!("get_pointer: {e:?}"))?;
+    if got_prev != prev_data {
+        return Err(format!(
+            "inverse root mismatch exp={prev_data} got={got_prev}"
+        ));
+    }
+    Ok(())
+}
+
+async fn fuzz_run_with_seed(seed: u64, steps: usize) -> Vec<String> {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+    let mut rng = StdRng::seed_from_u64(seed);
+
+    let mut live_keys: HashMap<String, String> = HashMap::new();
+
+    for step in 0..steps {
+        let now = chrono::Utc::now().to_rfc3339();
+        let op = pick_op(&mut rng, !live_keys.is_empty());
+        match op {
+            FuzzOp::Create => {
+                let col = random_collection(&mut rng);
+                let rkey = random_rkey(&mut rng);
+                let path = format!("{col}/{rkey}");
+                if live_keys.contains_key(&path) {
+                    continue;
+                }
+                let record = record_for_collection(col, &format!("s{seed}-n{step}"), &now);
+                let res = client
+                    .post(format!(
+                        "{}/xrpc/com.atproto.repo.createRecord",
+                        base_url().await
+                    ))
+                    .bearer_auth(&token)
+                    .json(&json!({
+                        "repo": did,
+                        "collection": col,
+                        "rkey": rkey,
+                        "record": record,
+                    }))
+                    .send()
+                    .await
+                    .expect("createRecord");
+                if res.status() == StatusCode::OK {
+                    live_keys.insert(path, col.to_string());
+                }
+            }
+            FuzzOp::Update => {
+                let keys: Vec<&String> = live_keys.keys().collect();
+                if keys.is_empty() {
+                    continue;
+                }
+                let path = keys[rng.gen_range(0..keys.len())].clone();
+                let col = live_keys.get(&path).unwrap().clone();
+                let rkey = path.split('/').nth(1).unwrap().to_string();
+                let record = record_for_collection(&col, &format!("s{seed}-u{step}"), &now);
+                let res = client
+                    .post(format!(
+                        "{}/xrpc/com.atproto.repo.putRecord",
+                        base_url().await
+                    ))
+                    .bearer_auth(&token)
+                    .json(&json!({
+                        "repo": did,
+                        "collection": col,
+                        "rkey": rkey,
+                        "record": record,
+                    }))
+                    .send()
+                    .await
+                    .expect("putRecord");
+                assert_eq!(res.status(), StatusCode::OK, "putRecord failed");
+            }
+            FuzzOp::Delete => {
+                let keys: Vec<String> = live_keys.keys().cloned().collect();
+                if keys.is_empty() {
+                    continue;
+                }
+                let path = keys[rng.gen_range(0..keys.len())].clone();
+                let col = live_keys.get(&path).unwrap().clone();
+                let rkey = path.split('/').nth(1).unwrap().to_string();
+                let res = client
+                    .post(format!(
+                        "{}/xrpc/com.atproto.repo.deleteRecord",
+                        base_url().await
+                    ))
+                    .bearer_auth(&token)
+                    .json(&json!({
+                        "repo": did,
+                        "collection": col,
+                        "rkey": rkey,
+                    }))
+                    .send()
+                    .await
+                    .expect("deleteRecord");
+                if res.status() == StatusCode::OK {
+                    live_keys.remove(&path);
+                }
+            }
+        }
+    }
+
+    let repos = get_test_repos().await;
+    let typed_did = Did::new(did.clone()).unwrap();
+    let events = repos
+        .repo
+        .get_events_since_seq(SequenceNumber::ZERO, None)
+        .await
+        .expect("get_events_since_seq");
+
+    let our: Vec<SequencedEvent> = events
+        .into_iter()
+        .filter(|e| {
+            e.did == typed_did
+                && e.event_type == RepoEventType::Commit
+                && e.prev_data_cid.is_some()
+                && e.ops
+                    .as_ref()
+                    .and_then(|v| v.as_array())
+                    .is_some_and(|a| !a.is_empty())
+        })
+        .collect();
+
+    let mut failures = Vec::new();
+    for event in &our {
+        if let Err(msg) = verify_commit_forward_and_inverse(event).await {
+            failures.push(format!(
+                "seed={seed} seq={} ops={:?}: {msg}",
+                event.seq.as_i64(),
+                event
+                    .ops
+                    .as_ref()
+                    .and_then(|v| v.as_array())
+                    .map(|a| a.len())
+            ));
+        }
+    }
+    failures
+}
+
+#[tokio::test]
+async fn mst_property_fuzz_seed_1() {
+    let failures = fuzz_run_with_seed(1, 150).await;
+    assert!(
+        failures.is_empty(),
+        "fuzz seed=1 found {} invalid commits:\n  - {}",
+        failures.len(),
+        failures.join("\n  - ")
+    );
+}
+
+#[tokio::test]
+async fn mst_property_fuzz_seed_42() {
+    let failures = fuzz_run_with_seed(42, 150).await;
+    assert!(
+        failures.is_empty(),
+        "fuzz seed=42 found {} invalid commits:\n  - {}",
+        failures.len(),
+        failures.join("\n  - ")
+    );
+}
+
+#[tokio::test]
+async fn mst_property_fuzz_seed_9001() {
+    let failures = fuzz_run_with_seed(9001, 150).await;
+    assert!(
+        failures.is_empty(),
+        "fuzz seed=9001 found {} invalid commits:\n  - {}",
+        failures.len(),
+        failures.join("\n  - ")
+    );
+}
+
+#[tokio::test]
+async fn mst_property_fuzz_deep_tree_seed_7() {
+    let failures = fuzz_run_with_seed(7, 400).await;
+    assert!(
+        failures.is_empty(),
+        "fuzz deep seed=7 found {} invalid commits:\n  - {}",
+        failures.len(),
+        failures.join("\n  - ")
+    );
+}

crates/tranquil-sync/Cargo.toml

@@ -22,3 +22,9 @@ serde = { workspace = true }
 serde_ipld_dagcbor = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
+
+[dev-dependencies]
+axum-test = { version = "19.1.1", features = [ "ws" ] }
+sqlx = { workspace = true }
+tokio-util = { workspace = true }
+tracing-subscriber.workspace = true

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;
                     }
                 }
@@ -247,6 +302,8 @@ async fn handle_socket_inner(
                         break;
                     };
 
+                    info!("{msg:?}");
+
                     if let Message::Close(_) = msg {
                         info!("Client closed connection");
                         break;
@@ -257,3 +314,44 @@ async fn handle_socket_inner(
     }
     Ok(())
 }
+
+#[cfg(test)]
+mod test {
+    use std::net::SocketAddr;
+    use std::time::Duration;
+
+    use super::super::sync_routes;
+    use super::*;
+    use axum_test::TestServer;
+    use tokio_util::sync::CancellationToken;
+
+    #[tokio::test]
+    async fn test_websockets_closing() {
+        // tracing_subscriber::fmt().init();
+        tranquil_config::ensure_test_defaults();
+        let state = AppState::new(CancellationToken::new()).await.unwrap();
+        let app = sync_routes()
+            .with_state(state)
+            .into_make_service_with_connect_info::<SocketAddr>();
+        let server = TestServer::builder().http_transport().build(app);
+
+        const CONNECTIONS: usize = 100;
+        let mut open_sockets = Vec::with_capacity(CONNECTIONS);
+
+        for _ in 0..CONNECTIONS {
+            let socket = server
+                .get_websocket("/com.atproto.sync.subscribeRepos")
+                .await
+                .into_websocket()
+                .await;
+            open_sockets.push(socket);
+        }
+        assert_eq!(SUBSCRIBER_COUNT.load(Ordering::SeqCst), CONNECTIONS);
+
+        drop(open_sockets);
+        // disgusting awful hack to give tokio time to poll the server futures enough times to actually drop all the
+        // websockets on the other end as well
+        tokio::time::sleep(Duration::from_millis(8)).await;
+        assert_eq!(SUBSCRIBER_COUNT.load(Ordering::SeqCst), 0);
+    }
+}

deploy/quadlets/tranquil-pds-app.container

@@ -5,11 +5,11 @@ After=tranquil-pds-db.service
 ContainerName=tranquil-pds-app
 Image=localhost/tranquil-pds:latest
 Pod=tranquil-pds.pod
-EnvironmentFile=/srv/tranquil-pds/config/tranquil-pds.env
 Environment=SERVER_HOST=0.0.0.0
 Environment=SERVER_PORT=3000
-Volume=/srv/tranquil-pds/blobs:/var/lib/tranquil/blobs:Z
-Volume=/srv/tranquil-pds/backups:/var/lib/tranquil/backups:Z
+Volume=/srv/tranquil-pds/config/config.toml:/etc/tranquil-pds/config.toml:ro,Z
+Volume=/srv/tranquil-pds/blobs:/var/lib/tranquil-pds/blobs:Z
+Volume=/srv/tranquil-pds/store:/var/lib/tranquil-pds/store:Z
 HealthCmd=wget -q --spider http://localhost:3000/xrpc/_health
 HealthInterval=30s
 HealthTimeout=10s

docker-compose.prod.yaml

@@ -9,7 +9,9 @@ services:
       SERVER_HOST: "0.0.0.0"
     volumes:
       - ./config.toml:/etc/tranquil-pds/config.toml:ro
-      - blob_data:/var/lib/tranquil/blobs
+      # In memory of @olaren.dev's blobs when lewis forgot to update /tranquil to /tranquil-pds :(
+      - blob_data:/var/lib/tranquil-pds/blobs
+      - store_data:/var/lib/tranquil-pds/store
     depends_on:
       db:
         condition: service_healthy
@@ -94,5 +96,6 @@ services:
 volumes:
   postgres_data:
   blob_data:
+  store_data:
   prometheus_data:
   acme_challenge:

docker-compose.yaml

@@ -10,7 +10,8 @@ services:
       DATABASE_URL: postgres://postgres:postgres@db:5432/pds
     volumes:
       - ./config.toml:/etc/tranquil-pds/config.toml:ro
-      - blob_data:/var/lib/tranquil/blobs
+      - blob_data:/var/lib/tranquil-pds/blobs
+      - store_data:/var/lib/tranquil-pds/store
     depends_on:
       - db
 
@@ -51,4 +52,5 @@ services:
 volumes:
   postgres_data:
   blob_data:
+  store_data:
   prometheus_data:

docs/install-containers.md

@@ -47,7 +47,7 @@ For production setups with proper service management, continue to either the Deb
 
 ## Standalone containers (no compose)
 
-If you already have postgres running on the host (eg. from the [Debian install guide](install-debian.md)), you can run just the app containers.
+If you already have postgres running on the host, you can run just the app containers.
 
 Build the images:
 ```sh
@@ -60,7 +60,7 @@ Run the backend with host networking (so it can access postgres on localhost) an
 podman run -d --name tranquil-pds \
   --network=host \
   -v /etc/tranquil-pds/config.toml:/etc/tranquil-pds/config.toml:ro,Z \
-  -v /var/lib/tranquil:/var/lib/tranquil:Z \
+  -v /var/lib/tranquil-pds:/var/lib/tranquil-pds:Z \
   tranquil-pds:latest
 ```
 
@@ -91,7 +91,7 @@ location / {
 }
 ```
 
-See the [Debian install guide](install-debian.md) for the full nginx config with all API routes.
+See the Debian with systemd quadlets section below for the full nginx config with all API routes.
 
 ---
 
@@ -110,7 +110,7 @@ apt install -y podman
 
 ```bash
 mkdir -p /etc/containers/systemd
-mkdir -p /srv/tranquil-pds/{postgres,blobs,certs,acme,config}
+mkdir -p /srv/tranquil-pds/{postgres,blobs,store,certs,acme,config}
 ```
 
 ## Create a configuration file

docs/install-debian.md

@@ -1,370 +0,0 @@
-# Tranquil PDS production installation on debian
-
-This guide covers installing Tranquil PDS on Debian.
-
-It is a "compile the thing on the server itself" -style guide.
-This cop-out is because Tranquil isn't built and released via CI as of yet.
-
-## Prerequisites
-
-- A server :p
-- Disk space enough for blobs (depends on usage; plan for ~1GB per active user as a baseline)
-- A domain name pointing to your server's IP
-- A wildcard TLS certificate for `*.pds.example.com` (user handles are served as subdomains)
-- Root/sudo/doas access
-
-## System setup
-
-```bash
-apt update && apt upgrade -y
-apt install -y curl git build-essential pkg-config libssl-dev
-```
-
-## Install rust
-
-```bash
-curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y
-source ~/.cargo/env
-rustup default stable
-```
-
-This installs the latest stable Rust.
-
-## Install postgres
-
-```bash
-apt install -y postgresql postgresql-contrib
-systemctl enable postgresql
-systemctl start postgresql
-sudo -u postgres psql -c "CREATE USER tranquil_pds WITH PASSWORD 'your-secure-password';"
-sudo -u postgres psql -c "CREATE DATABASE pds OWNER tranquil_pds;"
-sudo -u postgres psql -c "GRANT ALL PRIVILEGES ON DATABASE pds TO tranquil_pds;"
-```
-
-## Create blob storage directories
-
-```bash
-mkdir -p /var/lib/tranquil/blobs
-```
-
-We'll set ownership after creating the service user.
-
-## Install Node.js and pnpm (for frontend build)
-
-```bash
-curl -fsSL https://deb.nodesource.com/setup_24.x | bash -
-apt install -y nodejs
-npm install -g pnpm
-```
-
-## Clone and build Tranquil PDS
-
-```bash
-cd /opt
-git clone https://tangled.org/tranquil.farm/tranquil-pds tranquil-pds
-cd tranquil-pds
-cd frontend
-pnpm install --frozen-lockfile
-pnpm build
-cd ..
-cargo build --release
-```
-
-## Configure Tranquil PDS
-
-```bash
-mkdir -p /etc/tranquil-pds
-cp /opt/tranquil-pds/example.toml /etc/tranquil-pds/config.toml
-chmod 600 /etc/tranquil-pds/config.toml
-```
-
-Edit `/etc/tranquil-pds/config.toml` and fill in your values. Generate secrets with:
-```bash
-openssl rand -base64 48
-```
-
-> **Note:** Every config option can also be set via environment variables
-> (see comments in `example.toml`). Environment variables always take
-> precedence over the config file. You can also pass the config file path
-> via the `TRANQUIL_PDS_CONFIG` env var instead of `--config`.
-
-You can validate your configuration before starting the service:
-```bash
-/usr/local/bin/tranquil-pds --config /etc/tranquil-pds/config.toml validate
-```
-
-## Install frontend files
-
-```bash
-mkdir -p /var/www/tranquil-pds
-cp -r /opt/tranquil-pds/frontend/dist/* /var/www/tranquil-pds/
-chown -R www-data:www-data /var/www/tranquil-pds
-```
-
-## Create systemd service
-
-```bash
-useradd -r -s /sbin/nologin tranquil-pds
-chown -R tranquil-pds:tranquil-pds /var/lib/tranquil
-cp /opt/tranquil-pds/target/release/tranquil-pds /usr/local/bin/
-
-cat > /etc/systemd/system/tranquil-pds.service << 'EOF'
-[Unit]
-Description=Tranquil PDS - AT Protocol PDS
-After=network.target postgresql.service
-[Service]
-Type=simple
-User=tranquil-pds
-Group=tranquil-pds
-ExecStart=/usr/local/bin/tranquil-pds --config /etc/tranquil-pds/config.toml
-Restart=always
-RestartSec=5
-ProtectSystem=strict
-ProtectHome=true
-PrivateTmp=true
-ReadWritePaths=/var/lib/tranquil
-[Install]
-WantedBy=multi-user.target
-EOF
-
-systemctl daemon-reload
-systemctl enable tranquil-pds
-systemctl start tranquil-pds
-```
-
-## Install and configure nginx
-
-```bash
-apt install -y nginx certbot python3-certbot-nginx
-
-cat > /etc/nginx/sites-available/tranquil-pds << 'EOF'
-server {
-    listen 80;
-    listen [::]:80;
-    server_name pds.example.com *.pds.example.com;
-
-    location /.well-known/acme-challenge/ {
-        root /var/www/acme;
-    }
-
-    location / {
-        return 301 https://$host$request_uri;
-    }
-}
-
-server {
-    listen 443 ssl;
-    listen [::]:443 ssl;
-    http2 on;
-    server_name pds.example.com *.pds.example.com;
-
-    ssl_certificate /etc/letsencrypt/live/pds.example.com/fullchain.pem;
-    ssl_certificate_key /etc/letsencrypt/live/pds.example.com/privkey.pem;
-
-    client_max_body_size 10G;
-
-    root /var/www/tranquil-pds;
-
-    location /xrpc/ {
-        proxy_pass http://127.0.0.1:3000;
-        proxy_http_version 1.1;
-        proxy_set_header Upgrade $http_upgrade;
-        proxy_set_header Connection "upgrade";
-        proxy_set_header Host $host;
-        proxy_set_header X-Real-IP $remote_addr;
-        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
-        proxy_set_header X-Forwarded-Proto $scheme;
-        proxy_read_timeout 86400;
-        proxy_send_timeout 86400;
-        proxy_buffering off;
-        proxy_request_buffering off;
-    }
-
-    location = /oauth-client-metadata.json {
-        root /var/www/tranquil-pds;
-        default_type application/json;
-        sub_filter_once off;
-        sub_filter_types application/json;
-        sub_filter '__PDS_HOSTNAME__' $host;
-    }
-
-    location /oauth/ {
-        proxy_pass http://127.0.0.1:3000;
-        proxy_http_version 1.1;
-        proxy_set_header Host $host;
-        proxy_set_header X-Real-IP $remote_addr;
-        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
-        proxy_set_header X-Forwarded-Proto $scheme;
-        proxy_read_timeout 300;
-        proxy_send_timeout 300;
-    }
-
-    location /.well-known/ {
-        proxy_pass http://127.0.0.1:3000;
-        proxy_http_version 1.1;
-        proxy_set_header Host $host;
-        proxy_set_header X-Real-IP $remote_addr;
-        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
-        proxy_set_header X-Forwarded-Proto $scheme;
-    }
-
-    location /webhook/ {
-        proxy_pass http://127.0.0.1:3000;
-        proxy_http_version 1.1;
-        proxy_set_header Host $host;
-        proxy_set_header X-Real-IP $remote_addr;
-        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
-        proxy_set_header X-Forwarded-Proto $scheme;
-    }
-
-    location = /metrics {
-        proxy_pass http://127.0.0.1:3000;
-        proxy_http_version 1.1;
-        proxy_set_header Host $host;
-    }
-
-    location = /health {
-        proxy_pass http://127.0.0.1:3000;
-        proxy_http_version 1.1;
-        proxy_set_header Host $host;
-    }
-
-    location = /robots.txt {
-        proxy_pass http://127.0.0.1:3000;
-        proxy_http_version 1.1;
-        proxy_set_header Host $host;
-    }
-
-    location = /logo {
-        proxy_pass http://127.0.0.1:3000;
-        proxy_http_version 1.1;
-        proxy_set_header Host $host;
-    }
-
-    location ~ ^/u/[^/]+/did\.json$ {
-        proxy_pass http://127.0.0.1:3000;
-        proxy_http_version 1.1;
-        proxy_set_header Host $host;
-        proxy_set_header X-Real-IP $remote_addr;
-        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
-        proxy_set_header X-Forwarded-Proto $scheme;
-    }
-
-    location /assets/ {
-        expires 1y;
-        add_header Cache-Control "public, immutable";
-        try_files $uri =404;
-    }
-
-    location /app/ {
-        try_files $uri $uri/ /index.html;
-    }
-
-    location = / {
-        try_files /homepage.html /index.html;
-    }
-
-    location / {
-        try_files $uri $uri/ /index.html;
-    }
-}
-EOF
-
-ln -sf /etc/nginx/sites-available/tranquil-pds /etc/nginx/sites-enabled/
-rm -f /etc/nginx/sites-enabled/default
-mkdir -p /var/www/acme
-nginx -t
-systemctl reload nginx
-```
-
-## Obtain a wildcard SSL cert
-
-User handles are served as subdomains (eg., `alice.pds.example.com`), so you need a wildcard certificate.
-
-Wildcard certs require DNS-01 validation. If your DNS provider has a certbot plugin:
-```bash
-apt install -y python3-certbot-dns-cloudflare
-certbot certonly --dns-cloudflare \
-  --dns-cloudflare-credentials /etc/cloudflare.ini \
-  -d pds.example.com -d '*.pds.example.com'
-```
-
-For manual DNS validation (works with any provider):
-```bash
-certbot certonly --manual --preferred-challenges dns \
-  -d pds.example.com -d '*.pds.example.com'
-```
-
-Follow the prompts to add TXT records to your DNS. Note: manual mode doesn't auto-renew.
-
-After obtaining the cert, reload nginx:
-```bash
-systemctl reload nginx
-```
-
-## Configure firewall if you're into that sort of thing
-
-```bash
-apt install -y ufw
-ufw allow ssh
-ufw allow 80/tcp
-ufw allow 443/tcp
-ufw enable
-```
-
-## Verify installation
-
-```bash
-systemctl status tranquil-pds
-curl -s https://pds.example.com/xrpc/_health | jq
-curl -s https://pds.example.com/.well-known/atproto-did
-```
-
-## Maintenance
-
-View logs:
-```bash
-journalctl -u tranquil-pds -f
-```
-
-Update Tranquil PDS:
-```bash
-cd /opt/tranquil-pds
-git pull
-cd frontend && pnpm install --frozen-lockfile && pnpm build && cd ..
-cargo build --release
-systemctl stop tranquil-pds
-cp target/release/tranquil-pds /usr/local/bin/
-cp -r frontend/dist/* /var/www/tranquil-pds/
-systemctl start tranquil-pds
-```
-
-Tranquil should auto-migrate if there are any new migrations to be applied to the db, so you don't need to worry.
-
-Backup database:
-```bash
-sudo -u postgres pg_dump pds > /var/backups/pds-$(date +%Y%m%d).sql
-```
-
-## Custom homepage
-
-Drop a `homepage.html` in `/var/www/tranquil-pds/` and it becomes your landing page. Account dashboard is at `/app/` so you won't break anything.
-
-```bash
-cat > /var/www/tranquil-pds/homepage.html << 'EOF'
-<!DOCTYPE html>
-<html>
-<head>
-    <title>Welcome to my PDS</title>
-    <style>
-        body { font-family: system-ui; max-width: 600px; margin: 100px auto; padding: 20px; }
-    </style>
-</head>
-<body>
-    <h1>Welcome to my secret PDS</h1>
-    <p>This is a <a href="https://atproto.com">AT Protocol</a> Personal Data Server.</p>
-    <p><a href="/app/">Sign in</a> or learn more at <a href="https://bsky.social">Bluesky</a>.</p>
-</body>
-</html>
-EOF
-```

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`.

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

scripts/install-debian.sh

@@ -1,512 +0,0 @@
-#!/bin/bash
-set -euo pipefail
-
-RED='\033[0;31m'
-GREEN='\033[0;32m'
-YELLOW='\033[1;33m'
-BLUE='\033[0;34m'
-NC='\033[0m'
-
-log_info() { echo -e "${BLUE}[INFO]${NC} $1"; }
-log_success() { echo -e "${GREEN}[OK]${NC} $1"; }
-log_warn() { echo -e "${YELLOW}[WARN]${NC} $1"; }
-log_error() { echo -e "${RED}[ERROR]${NC} $1"; }
-
-if [[ $EUID -ne 0 ]]; then
-   log_error "This script must be run as root"
-   exit 1
-fi
-
-if ! grep -qi "debian" /etc/os-release 2>/dev/null; then
-    log_warn "This script is designed for Debian. Proceed with caution on other distros."
-fi
-
-nuke_installation() {
-    log_warn "NUKING EXISTING INSTALLATION"
-    log_info "Stopping services..."
-    systemctl stop tranquil-pds 2>/dev/null || true
-    systemctl disable tranquil-pds 2>/dev/null || true
-
-    log_info "Removing Tranquil PDS files..."
-    rm -rf /opt/tranquil-pds
-    rm -rf /var/lib/tranquil-pds
-    rm -f /usr/local/bin/tranquil-pds
-    rm -f /usr/local/bin/tranquil-pds-sendmail
-    rm -f /usr/local/bin/tranquil-pds-mailq
-    rm -rf /var/spool/tranquil-pds-mail
-    rm -f /etc/systemd/system/tranquil-pds.service
-    systemctl daemon-reload
-
-    log_info "Removing Tranquil PDS configuration..."
-    rm -rf /etc/tranquil-pds
-
-    log_info "Dropping postgres database and user..."
-    sudo -u postgres psql -c "DROP DATABASE IF EXISTS pds;" 2>/dev/null || true
-    sudo -u postgres psql -c "DROP USER IF EXISTS tranquil_pds;" 2>/dev/null || true
-
-    log_info "Removing blob storage..."
-    rm -rf /var/lib/tranquil 2>/dev/null || true
-
-    log_info "Removing nginx config..."
-    rm -f /etc/nginx/sites-enabled/tranquil-pds
-    rm -f /etc/nginx/sites-available/tranquil-pds
-    systemctl reload nginx 2>/dev/null || true
-
-    log_success "Previous installation nuked"
-}
-
-if [[ -f /etc/tranquil-pds/tranquil-pds.env ]] || [[ -d /opt/tranquil-pds ]] || [[ -f /usr/local/bin/tranquil-pds ]]; then
-    log_warn "Existing installation detected"
-    echo ""
-    echo "Options:"
-    echo "  1) Nuke everything and start fresh (destroys database!)"
-    echo "  2) Continue with existing installation (idempotent update)"
-    echo "  3) Exit"
-    echo ""
-    read -p "Choose an option [1/2/3]: " INSTALL_CHOICE
-
-    case "$INSTALL_CHOICE" in
-        1)
-            echo ""
-            log_warn "This will DELETE:"
-            echo "  - PostgreSQL database 'pds' and all data"
-            echo "  - All Tranquil PDS configuration and credentials"
-            echo "  - All source code in /opt/tranquil-pds"
-            echo "  - All blobs in /var/lib/tranquil/"
-            echo ""
-            read -p "Type 'NUKE' to confirm: " CONFIRM_NUKE
-            if [[ "$CONFIRM_NUKE" == "NUKE" ]]; then
-                nuke_installation
-            else
-                log_error "Nuke cancelled"
-                exit 1
-            fi
-            ;;
-        2)
-            log_info "Continuing with existing installation..."
-            ;;
-        3)
-            exit 0
-            ;;
-        *)
-            log_error "Invalid option"
-            exit 1
-            ;;
-    esac
-fi
-
-echo ""
-log_info "Tranquil PDS Installation Script for Debian"
-echo ""
-
-get_public_ips() {
-    IPV4=$(curl -4 -s --max-time 5 ifconfig.me 2>/dev/null || curl -4 -s --max-time 5 icanhazip.com 2>/dev/null || echo "Could not detect")
-    IPV6=$(curl -6 -s --max-time 5 ifconfig.me 2>/dev/null || curl -6 -s --max-time 5 icanhazip.com 2>/dev/null || echo "")
-}
-
-log_info "Detecting public IP addresses..."
-get_public_ips
-echo "  IPv4: ${IPV4}"
-[[ -n "$IPV6" ]] && echo "  IPv6: ${IPV6}"
-echo ""
-
-read -p "Enter your PDS domain (eg., pds.example.com): " PDS_DOMAIN
-if [[ -z "$PDS_DOMAIN" ]]; then
-    log_error "Domain cannot be empty"
-    exit 1
-fi
-
-read -p "Enter your email for Let's Encrypt: " CERTBOT_EMAIL
-if [[ -z "$CERTBOT_EMAIL" ]]; then
-    log_error "Email cannot be empty"
-    exit 1
-fi
-
-echo ""
-log_info "DNS records required (create these now if you haven't):"
-echo ""
-echo "  ${PDS_DOMAIN}      A      ${IPV4}"
-[[ -n "$IPV6" ]] && echo "  ${PDS_DOMAIN}      AAAA   ${IPV6}"
-echo "  *.${PDS_DOMAIN}    A      ${IPV4}    (for user handles)"
-[[ -n "$IPV6" ]] && echo "  *.${PDS_DOMAIN}    AAAA   ${IPV6}    (for user handles)"
-echo ""
-read -p "Have you created these DNS records? (y/N): " DNS_CONFIRMED
-if [[ ! "$DNS_CONFIRMED" =~ ^[Yy]$ ]]; then
-    log_warn "Please create the DNS records and run this script again."
-    exit 0
-fi
-
-CREDENTIALS_FILE="/etc/tranquil-pds/.credentials"
-if [[ -f "$CREDENTIALS_FILE" ]]; then
-    log_info "Loading existing credentials..."
-    source "$CREDENTIALS_FILE"
-else
-    log_info "Generating secrets..."
-    JWT_SECRET=$(openssl rand -base64 48)
-    DPOP_SECRET=$(openssl rand -base64 48)
-    MASTER_KEY=$(openssl rand -base64 48)
-    DB_PASSWORD=$(openssl rand -base64 24 | tr -dc 'a-zA-Z0-9' | head -c 32)
-
-    mkdir -p /etc/tranquil-pds
-    cat > "$CREDENTIALS_FILE" << EOF
-JWT_SECRET="$JWT_SECRET"
-DPOP_SECRET="$DPOP_SECRET"
-MASTER_KEY="$MASTER_KEY"
-DB_PASSWORD="$DB_PASSWORD"
-EOF
-    chmod 600 "$CREDENTIALS_FILE"
-    log_success "Secrets generated"
-fi
-
-log_info "Checking swap space..."
-TOTAL_MEM_KB=$(grep MemTotal /proc/meminfo | awk '{print $2}')
-TOTAL_SWAP_KB=$(grep SwapTotal /proc/meminfo | awk '{print $2}')
-
-if [[ $TOTAL_SWAP_KB -lt 2000000 ]]; then
-    if [[ ! -f /swapfile ]]; then
-        log_info "Adding swap space for compilation..."
-        SWAP_SIZE="4G"
-        [[ $TOTAL_MEM_KB -ge 4000000 ]] && SWAP_SIZE="2G"
-        fallocate -l $SWAP_SIZE /swapfile || dd if=/dev/zero of=/swapfile bs=1M count=4096
-        chmod 600 /swapfile
-        mkswap /swapfile
-        swapon /swapfile
-        grep -q '/swapfile' /etc/fstab || echo '/swapfile none swap sw 0 0' >> /etc/fstab
-        log_success "Swap added ($SWAP_SIZE)"
-    else
-        swapon /swapfile 2>/dev/null || true
-    fi
-fi
-
-log_info "Updating system packages..."
-apt update && apt upgrade -y
-
-log_info "Installing build dependencies..."
-apt install -y curl git build-essential pkg-config libssl-dev ca-certificates gnupg lsb-release unzip xxd
-
-log_info "Installing postgres..."
-apt install -y postgresql postgresql-contrib
-systemctl enable postgresql
-systemctl start postgresql
-sudo -u postgres psql -c "CREATE USER tranquil_pds WITH PASSWORD '${DB_PASSWORD}';" 2>/dev/null || \
-    sudo -u postgres psql -c "ALTER USER tranquil_pds WITH PASSWORD '${DB_PASSWORD}';"
-sudo -u postgres psql -c "CREATE DATABASE pds OWNER tranquil_pds;" 2>/dev/null || true
-sudo -u postgres psql -c "GRANT ALL PRIVILEGES ON DATABASE pds TO tranquil_pds;"
-log_success "postgres configured"
-
-log_info "Creating blob storage directories..."
-mkdir -p /var/lib/tranquil/blobs
-log_success "Blob storage directories created"
-
-log_info "Installing rust..."
-if [[ -f "$HOME/.cargo/env" ]]; then
-    source "$HOME/.cargo/env"
-fi
-if ! command -v rustc &>/dev/null; then
-    curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y
-    source "$HOME/.cargo/env"
-fi
-
-log_info "Installing Node.js..."
-if ! command -v node &>/dev/null; then
-    curl -fsSL https://deb.nodesource.com/setup_24.x | bash -
-    apt install -y nodejs
-fi
-
-log_info "Installing pnpm..."
-if ! command -v pnpm &>/dev/null; then
-    npm install -g pnpm
-fi
-
-log_info "Cloning Tranquil PDS..."
-if [[ ! -d /opt/tranquil-pds ]]; then
-    git clone https://tangled.org/tranquil.farm/tranquil-pds /opt/tranquil-pds
-else
-    cd /opt/tranquil-pds && git pull
-fi
-cd /opt/tranquil-pds
-
-log_info "Building frontend..."
-cd frontend && pnpm install --frozen-lockfile && pnpm build && cd ..
-log_success "Frontend built"
-
-log_info "Building Tranquil PDS (this takes a while)..."
-source "$HOME/.cargo/env"
-if [[ $TOTAL_MEM_KB -lt 4000000 ]]; then
-    log_info "Low memory - limiting parallel jobs"
-    CARGO_BUILD_JOBS=1 cargo build --release
-else
-    cargo build --release
-fi
-log_success "Tranquil PDS built"
-
-log_info "Running migrations..."
-cargo install sqlx-cli --no-default-features --features postgres
-export DATABASE_URL="postgres://tranquil_pds:${DB_PASSWORD}@localhost:5432/pds"
-"$HOME/.cargo/bin/sqlx" migrate run
-log_success "Migrations complete"
-
-log_info "Setting up mail trap..."
-mkdir -p /var/spool/tranquil-pds-mail
-chmod 1777 /var/spool/tranquil-pds-mail
-
-cat > /usr/local/bin/tranquil-pds-sendmail << 'SENDMAIL_EOF'
-#!/bin/bash
-MAIL_DIR="/var/spool/tranquil-pds-mail"
-TIMESTAMP=$(date +%Y%m%d-%H%M%S)
-RANDOM_ID=$(head -c 4 /dev/urandom | xxd -p)
-MAIL_FILE="${MAIL_DIR}/${TIMESTAMP}-${RANDOM_ID}.eml"
-mkdir -p "$MAIL_DIR"
-{
-    echo "X-Tranquil-PDS-Received: $(date -Iseconds)"
-    echo "X-Tranquil-PDS-Args: $*"
-    echo ""
-    cat
-} > "$MAIL_FILE"
-chmod 644 "$MAIL_FILE"
-exit 0
-SENDMAIL_EOF
-chmod +x /usr/local/bin/tranquil-pds-sendmail
-
-cat > /usr/local/bin/tranquil-pds-mailq << 'MAILQ_EOF'
-#!/bin/bash
-MAIL_DIR="/var/spool/tranquil-pds-mail"
-case "${1:-list}" in
-    list)
-        ls -lt "$MAIL_DIR"/*.eml 2>/dev/null | head -20 || echo "No emails"
-        ;;
-    latest)
-        f=$(ls -t "$MAIL_DIR"/*.eml 2>/dev/null | head -1)
-        [[ -f "$f" ]] && cat "$f" || echo "No emails"
-        ;;
-    clear)
-        rm -f "$MAIL_DIR"/*.eml
-        echo "Cleared"
-        ;;
-    count)
-        ls -1 "$MAIL_DIR"/*.eml 2>/dev/null | wc -l
-        ;;
-    [0-9]*)
-        f=$(ls -t "$MAIL_DIR"/*.eml 2>/dev/null | sed -n "${1}p")
-        [[ -f "$f" ]] && cat "$f" || echo "Not found"
-        ;;
-    *)
-        [[ -f "$MAIL_DIR/$1" ]] && cat "$MAIL_DIR/$1" || echo "Usage: tranquil-pds-mailq [list|latest|clear|count|N]"
-        ;;
-esac
-MAILQ_EOF
-chmod +x /usr/local/bin/tranquil-pds-mailq
-
-log_info "Creating Tranquil PDS configuration..."
-cat > /etc/tranquil-pds/tranquil-pds.env << EOF
-SERVER_HOST=127.0.0.1
-SERVER_PORT=3000
-PDS_HOSTNAME=${PDS_DOMAIN}
-DATABASE_URL=postgres://tranquil_pds:${DB_PASSWORD}@localhost:5432/pds
-DATABASE_MAX_CONNECTIONS=100
-DATABASE_MIN_CONNECTIONS=10
-BLOB_STORAGE_PATH=/var/lib/tranquil/blobs
-JWT_SECRET=${JWT_SECRET}
-DPOP_SECRET=${DPOP_SECRET}
-MASTER_KEY=${MASTER_KEY}
-PLC_DIRECTORY_URL=https://plc.directory
-CRAWLERS=https://bsky.network
-AVAILABLE_USER_DOMAINS=${PDS_DOMAIN}
-MAIL_FROM_ADDRESS=noreply@${PDS_DOMAIN}
-MAIL_FROM_NAME=Tranquil PDS
-SENDMAIL_PATH=/usr/local/bin/tranquil-pds-sendmail
-EOF
-chmod 600 /etc/tranquil-pds/tranquil-pds.env
-
-log_info "Installing Tranquil PDS..."
-id -u tranquil-pds &>/dev/null || useradd -r -s /sbin/nologin tranquil-pds
-cp /opt/tranquil-pds/target/release/tranquil-server /usr/local/bin/tranquil-pds
-mkdir -p /var/lib/tranquil-pds
-cp -r /opt/tranquil-pds/frontend/dist /var/lib/tranquil-pds/frontend
-chown -R tranquil-pds:tranquil-pds /var/lib/tranquil-pds
-chown -R tranquil-pds:tranquil-pds /var/lib/tranquil
-
-cat > /etc/systemd/system/tranquil-pds.service << 'EOF'
-[Unit]
-Description=Tranquil PDS - AT Protocol PDS
-After=network.target postgresql.service
-
-[Service]
-Type=simple
-User=tranquil-pds
-Group=tranquil-pds
-EnvironmentFile=/etc/tranquil-pds/tranquil-pds.env
-ExecStart=/usr/local/bin/tranquil-pds
-Restart=always
-RestartSec=5
-ProtectSystem=strict
-ProtectHome=true
-PrivateTmp=true
-ReadWritePaths=/var/lib/tranquil
-
-[Install]
-WantedBy=multi-user.target
-EOF
-
-systemctl daemon-reload
-systemctl enable tranquil-pds
-systemctl start tranquil-pds
-log_success "Tranquil PDS service started"
-
-log_info "Installing nginx..."
-apt install -y nginx
-cat > /etc/nginx/sites-available/tranquil-pds << EOF
-server {
-    listen 80;
-    listen [::]:80;
-    server_name ${PDS_DOMAIN} *.${PDS_DOMAIN};
-
-    location /.well-known/acme-challenge/ {
-        root /var/www/html;
-    }
-
-    location / {
-        proxy_pass http://127.0.0.1:3000;
-        proxy_http_version 1.1;
-        proxy_set_header Upgrade \$http_upgrade;
-        proxy_set_header Connection "upgrade";
-        proxy_set_header Host \$host;
-        proxy_set_header X-Real-IP \$remote_addr;
-        proxy_set_header X-Forwarded-For \$proxy_add_x_forwarded_for;
-        proxy_set_header X-Forwarded-Proto \$scheme;
-        proxy_read_timeout 86400;
-        proxy_send_timeout 86400;
-        client_max_body_size 100M;
-    }
-}
-EOF
-
-ln -sf /etc/nginx/sites-available/tranquil-pds /etc/nginx/sites-enabled/
-rm -f /etc/nginx/sites-enabled/default
-nginx -t
-systemctl reload nginx
-log_success "nginx configured"
-
-log_info "Configuring firewall..."
-apt install -y ufw
-ufw --force reset
-ufw default deny incoming
-ufw default allow outgoing
-ufw allow ssh
-ufw allow 80/tcp
-ufw allow 443/tcp
-ufw --force enable
-log_success "Firewall configured"
-
-echo ""
-log_info "Obtaining wildcard SSL certificate..."
-echo ""
-echo "User handles are served as subdomains (eg., alice.${PDS_DOMAIN}),"
-echo "so you need a wildcard certificate. This requires DNS validation."
-echo ""
-echo "You'll need to add a TXT record to your DNS when prompted."
-echo ""
-read -p "Ready to proceed? (y/N): " CERT_READY
-
-if [[ "$CERT_READY" =~ ^[Yy]$ ]]; then
-    apt install -y certbot python3-certbot-nginx
-
-    log_info "Running certbot with DNS challenge..."
-    echo ""
-    echo "When prompted, add the TXT record to your DNS, wait a minute"
-    echo "for propagation, then press Enter to continue."
-    echo ""
-
-    if certbot certonly --manual --preferred-challenges dns \
-        -d "${PDS_DOMAIN}" -d "*.${PDS_DOMAIN}" \
-        --email "${CERTBOT_EMAIL}" --agree-tos; then
-
-        cat > /etc/nginx/sites-available/tranquil-pds << EOF
-server {
-    listen 80;
-    listen [::]:80;
-    server_name ${PDS_DOMAIN} *.${PDS_DOMAIN};
-
-    location /.well-known/acme-challenge/ {
-        root /var/www/html;
-    }
-
-    location / {
-        return 301 https://\$host\$request_uri;
-    }
-}
-
-server {
-    listen 443 ssl http2;
-    listen [::]:443 ssl http2;
-    server_name ${PDS_DOMAIN} *.${PDS_DOMAIN};
-
-    ssl_certificate /etc/letsencrypt/live/${PDS_DOMAIN}/fullchain.pem;
-    ssl_certificate_key /etc/letsencrypt/live/${PDS_DOMAIN}/privkey.pem;
-    ssl_protocols TLSv1.2 TLSv1.3;
-    ssl_ciphers HIGH:!aNULL:!MD5;
-    ssl_prefer_server_ciphers on;
-    ssl_session_cache shared:SSL:10m;
-
-    location / {
-        proxy_pass http://127.0.0.1:3000;
-        proxy_http_version 1.1;
-        proxy_set_header Upgrade \$http_upgrade;
-        proxy_set_header Connection "upgrade";
-        proxy_set_header Host \$host;
-        proxy_set_header X-Real-IP \$remote_addr;
-        proxy_set_header X-Forwarded-For \$proxy_add_x_forwarded_for;
-        proxy_set_header X-Forwarded-Proto \$scheme;
-        proxy_read_timeout 86400;
-        proxy_send_timeout 86400;
-        client_max_body_size 100M;
-    }
-}
-EOF
-        nginx -t && systemctl reload nginx
-        log_success "Wildcard SSL certificate installed"
-
-        echo ""
-        log_warn "Certificate renewal note:"
-        echo "Manual DNS challenges don't auto-renew. Before expiry, run:"
-        echo "  certbot renew --manual"
-        echo ""
-        echo "For auto-renewal, consider using a DNS provider plugin:"
-        echo "  apt install python3-certbot-dns-cloudflare  # or your provider"
-        echo ""
-    else
-        log_warn "Wildcard cert failed. You can retry later with:"
-        echo "  certbot certonly --manual --preferred-challenges dns \\"
-        echo "    -d ${PDS_DOMAIN} -d '*.${PDS_DOMAIN}'"
-    fi
-else
-    log_warn "Skipping SSL. Your PDS is running on HTTP only."
-    echo "To add SSL later, run:"
-    echo "  certbot certonly --manual --preferred-challenges dns \\"
-    echo "    -d ${PDS_DOMAIN} -d '*.${PDS_DOMAIN}'"
-fi
-
-log_info "Verifying installation..."
-sleep 3
-if curl -s "http://localhost:3000/xrpc/_health" | grep -q "version"; then
-    log_success "Tranquil PDS is responding"
-else
-    log_warn "Tranquil PDS may still be starting. Check: journalctl -u tranquil-pds -f"
-fi
-
-echo ""
-log_success "Installation complete"
-echo ""
-echo "PDS: https://${PDS_DOMAIN}"
-echo ""
-echo "Credentials (also in /etc/tranquil-pds/.credentials):"
-echo "  DB password: ${DB_PASSWORD}"
-echo ""
-echo "Data locations:"
-echo "  Blobs:   /var/lib/tranquil/blobs"
-echo ""
-echo "Commands:"
-echo "  journalctl -u tranquil-pds -f    # logs"
-echo "  systemctl restart tranquil-pds   # restart"
-echo "  tranquil-pds-mailq               # view trapped emails"
-echo ""