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

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 \

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_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-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

@@ -302,6 +302,8 @@ async fn handle_socket_inner(
                         break;
                     };
 
+                    info!("{msg:?}");
+
                     if let Message::Close(_) = msg {
                         info!("Client closed connection");
                         break;
@@ -312,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);
+    }
+}