chore: prepare qconnection 0.5.2 release

.github/workflows/publish-crates.yml

@@ -16,7 +16,7 @@ jobs:
   release:
     runs-on: ubuntu-latest
     permissions:
-      contents: read
+      contents: write
       id-token: write
 
     steps:
@@ -33,7 +33,7 @@ jobs:
         uses: rust-lang/crates-io-auth-action@v1
         id: auth
 
-      - name: Release workspace crates
+      - name: Release publishable workspace crates
         shell: bash
         env:
           CARGO_REGISTRY_TOKEN: ${{ steps.auth.outputs.token }}
@@ -62,16 +62,6 @@ jobs:
             dquic
           )
 
-          if [[ "$mode" == "dry-run" ]]; then
-            publish_args=(cargo publish --dry-run --locked)
-            for package in "${packages[@]}"; do
-              publish_args+=(-p "$package")
-            done
-            echo "dry-run publish packages: ${packages[*]}"
-            "${publish_args[@]}"
-            exit 0
-          fi
-
           cargo metadata --format-version 1 > "$RUNNER_TEMP/workspace-metadata.json"
           package_versions="$(
             PACKAGES="$(printf '%s\n' "${packages[@]}")" RUNNER_TEMP="$RUNNER_TEMP" python3 - <<'PY'
@@ -90,7 +80,7 @@ jobs:
           PY
           )"
 
-          missing_packages=()
+          packages_to_publish=()
           while IFS=$'\t' read -r crate_name crate_version; do
             [ -n "$crate_name" ] || continue
             crate_state="$(
@@ -100,37 +90,100 @@ jobs:
           import urllib.request
 
           name, version = sys.argv[1], sys.argv[2]
-          url = f"https://crates.io/api/v1/crates/{name}/{version}"
-          request = urllib.request.Request(url, headers={"User-Agent": "genmeta dquic publish workflow"})
+          headers = {"User-Agent": "genmeta dquic publish workflow"}
+          version_url = f"https://crates.io/api/v1/crates/{name}/{version}"
+          version_request = urllib.request.Request(version_url, headers=headers)
           try:
-              with urllib.request.urlopen(request, timeout=20) as response:
+              with urllib.request.urlopen(version_request, timeout=20) as response:
                   if response.status == 200:
-                      print("published")
+                      print("published_version")
                   else:
                       raise SystemExit(f"unexpected crates.io status for {name} {version}: {response.status}")
           except urllib.error.HTTPError as error:
               if error.code == 404:
-                  print("missing")
+                  crate_url = f"https://crates.io/api/v1/crates/{name}"
+                  crate_request = urllib.request.Request(crate_url, headers=headers)
+                  try:
+                      with urllib.request.urlopen(crate_request, timeout=20) as response:
+                          if response.status == 200:
+                              print("missing_version")
+                          else:
+                              raise SystemExit(f"unexpected crates.io crate status for {name}: {response.status}")
+                  except urllib.error.HTTPError as crate_error:
+                      if crate_error.code == 404:
+                          print("missing_crate")
+                      else:
+                          raise
               else:
                   raise
           PY
             )"
 
-            if [[ "$crate_state" == "published" ]]; then
+            if [[ "$crate_state" == "published_version" ]]; then
               echo "skip $crate_name $crate_version (already on crates.io)"
-            else
+            elif [[ "$crate_state" == "missing_version" ]]; then
               echo "publish $crate_name $crate_version"
-              missing_packages+=("$crate_name")
+              packages_to_publish+=("$crate_name")
+            else
+              echo "skip $crate_name $crate_version (crate not yet initialized on crates.io)"
             fi
           done <<< "$package_versions"
 
-          if [[ "${#missing_packages[@]}" -eq 0 ]]; then
-            echo "all selected packages are already published"
+          if [[ "${#packages_to_publish[@]}" -eq 0 ]]; then
+            echo "no already-initialized crates need a new crates.io release"
             exit 0
           fi
 
-          publish_args=(cargo publish --locked)
-          for package in "${missing_packages[@]}"; do
+          if [[ "$mode" == "dry-run" ]]; then
+            publish_args=(cargo publish --dry-run --locked)
+            echo "dry-run publish packages: ${packages_to_publish[*]}"
+          else
+            publish_args=(cargo publish --locked)
+            echo "publish packages: ${packages_to_publish[*]}"
+          fi
+
+          for package in "${packages_to_publish[@]}"; do
             publish_args+=(-p "$package")
           done
           "${publish_args[@]}"
+
+      - name: Create GitHub Release
+        if: github.ref_type == 'tag' && startsWith(github.ref_name, 'v')
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          set -euo pipefail
+          tag_ref="refs/tags/$GITHUB_REF_NAME"
+          git fetch --force origin "$tag_ref:$tag_ref"
+          tag_object="$(git rev-parse "$tag_ref^{tag}" 2>/dev/null || true)"
+          target_commit="$(git rev-parse "$tag_ref^{commit}")"
+          notes_file="$RUNNER_TEMP/release-notes.md"
+          git for-each-ref "$tag_ref" --format='%(contents)' > "$notes_file"
+          if [ ! -s "$notes_file" ]; then
+            git log -1 --format=%B "$target_commit" > "$notes_file"
+          fi
+          {
+            printf '\n## Authentication and provenance\n\n'
+            printf -- '- Tag: `%s`\n' "$GITHUB_REF_NAME"
+            if [ -n "$tag_object" ]; then
+              printf -- '- Annotated tag object: `%s`\n' "$tag_object"
+            else
+              printf -- '- Tag object: lightweight tag\n'
+            fi
+            printf -- '- Target commit: `%s`\n' "$target_commit"
+            printf -- '- Workflow run: %s/%s/actions/runs/%s\n' \
+              "$GITHUB_SERVER_URL" "$GITHUB_REPOSITORY" "$GITHUB_RUN_ID"
+            printf -- '- Workflow attempt: `%s`\n' "$GITHUB_RUN_ATTEMPT"
+            printf -- '- Published by: GitHub Actions `%s` workflow\n' "$GITHUB_WORKFLOW"
+          } >> "$notes_file"
+
+          if gh release view "$GITHUB_REF_NAME" --repo "$GITHUB_REPOSITORY" >/dev/null 2>&1; then
+            echo "github release $GITHUB_REF_NAME already exists"
+            exit 0
+          fi
+
+          gh release create "$GITHUB_REF_NAME" \
+            --repo "$GITHUB_REPOSITORY" \
+            --verify-tag \
+            --title "$GITHUB_REF_NAME" \
+            --notes-file "$notes_file"

Cargo.toml

@@ -106,7 +106,7 @@ qresolve = { path = "./qresolve", version = "0.5.1" }
 qrecovery = { path = "./qrecovery", version = "0.5.1" }
 qtraversal = { path = "./qtraversal", version = "0.5.1" }
 qcongestion = { path = "./qcongestion", version = "0.5.1" }
-qconnection = { path = "./qconnection", version = "0.5.1" }
+qconnection = { path = "./qconnection", version = "0.5.2" }
 dquic = { path = "./dquic", version = "0.5.1" }
 h3-shim = { path = "./h3-shim", version = "0.5.1" }
 

dquic/tests/echo.rs

@@ -130,6 +130,30 @@ fn shutdown() -> Result<(), BoxError> {
     })
 }
 
+#[test]
+fn application_close_then_drop_does_not_strand_connection() -> Result<(), BoxError> {
+    run(async {
+        let router = Arc::new(QuicRouter::default());
+        let (listeners, server_task) =
+            launch_echo_server(router.clone(), server_parameters()).await?;
+        let _server_task = AbortOnDropHandle::new(tokio::spawn(server_task));
+
+        let server_addr = get_server_addr(&listeners);
+        let client = launch_test_client(router, client_parameters());
+        let connection = client
+            .connected_to_with_source("localhost", [(Source::System, server_addr.into())])
+            .await?;
+
+        send_and_verify_echo(&connection, TEST_DATA).await?;
+        connection.close("client done", 0)?;
+        drop(connection);
+
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        listeners.shutdown();
+        Ok(())
+    })
+}
+
 #[test]
 fn idle_timeout() -> Result<(), BoxError> {
     run(async {

qconnection/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "qconnection"
-version = "0.5.1"
+version = "0.5.2"
 edition.workspace = true
 description = "Encapsulation of QUIC connections, a part of dquic"
 readme.workspace = true

qconnection/src/builder.rs

@@ -551,7 +551,7 @@ impl PendingConnection {
             interfaces: self.interfaces,
             locations: self.locations,
             rcvd_pkt_q: self.rcvd_pkt_q,
-            conn_state,
+            conn_state: conn_state.clone(),
             idle_config: ArcIdleConfig::new(max_idle_timeout, self.defer_idle_timeout),
             paths: ArcPathContexts::new(self.tx_wakers.clone(), event_broker.clone()),
             send_lock: self.send_lock,
@@ -575,8 +575,10 @@ impl PendingConnection {
         spawn_tls_handshake(&components, self.tx_wakers.clone());
         spawn_deliver_and_parse(&components);
 
-        let connection = Arc::new(Connection {
+        let connection = Arc::new_cyclic(|weak_self| Connection {
             state: Ok(components).into(),
+            conn_state: conn_state.clone(),
+            weak_self: weak_self.clone(),
             qlog_span,
             tracing_span,
         });
@@ -739,17 +741,34 @@ fn spawn_drive_connection(
 ) {
     tokio::spawn(
         async move {
+            let mut retained_connection: Option<Arc<Connection>> = None;
             while let Some(event) = events.recv().await {
-                let Some(connection) = weak_connection.upgrade() else {
+                let Some(connection) = retained_connection
+                    .as_ref()
+                    .cloned()
+                    .or_else(|| weak_connection.upgrade())
+                else {
                     break;
                 };
+
                 match event {
                     Event::Handshaked => {}
-                    Event::Failed(quic_error) => _ = connection.enter_closing(quic_error),
-                    Event::ApplicationClose(_app_error) => {}
-                    Event::Closed(ccf) => _ = connection.enter_draining(ccf),
+                    Event::Failed(quic_error) => {
+                        retained_connection.get_or_insert_with(|| connection.clone());
+                        _ = connection.enter_closing(quic_error);
+                    }
+                    Event::ApplicationClose(_app_error) => {
+                        retained_connection.get_or_insert_with(|| connection.clone());
+                    }
+                    Event::Closed(ccf) => {
+                        retained_connection.get_or_insert_with(|| connection.clone());
+                        _ = connection.enter_draining(ccf);
+                    }
                     Event::StatelessReset => {}
-                    Event::Terminated => {}
+                    Event::Terminated => {
+                        retained_connection.take();
+                        break;
+                    }
                 }
             }
         }

qconnection/src/events.rs

@@ -5,7 +5,6 @@ use qbase::{
     error::{AppError, QuicError},
     frame::ConnectionCloseFrame,
 };
-use qevent::quic::connectivity::BaseConnectionStates;
 use tokio::sync::mpsc;
 
 use crate::state::ArcConnState;
@@ -54,25 +53,12 @@ impl EmitEvent for ArcEventBroker {
                     return;
                 }
             }
-            Event::Failed(error) => {
-                if self.conn_state.enter_closing(error).is_none() {
-                    return;
-                }
-            }
-            Event::ApplicationClose(error) => {
-                if self.conn_state.enter_closing(error).is_none() {
-                    return;
-                }
-            }
-            Event::Closed(ccf) => {
-                if self.conn_state.enter_draining(ccf).is_none() {
+            Event::Failed(_) | Event::ApplicationClose(_) | Event::Closed(_) => {}
+            Event::Terminated => {
+                if self.conn_state.enter_closed().is_none() {
                     return;
                 }
             }
-            Event::Terminated => {
-                let terminated_state = BaseConnectionStates::Closed;
-                self.conn_state.update(terminated_state.into());
-            }
             Event::StatelessReset => todo!("unsupported"),
         };
         tracing::debug!(target: "quic", new_state = ?event, "connection state changed");
@@ -88,14 +74,63 @@ impl EmitEvent for mpsc::UnboundedSender<Event> {
 
 #[cfg(test)]
 mod tests {
+    use qbase::{
+        error::{ErrorFrameType, ErrorKind, QuicError},
+        frame::ConnectionCloseFrame,
+        varint::VarInt,
+    };
     use tokio::sync::mpsc;
 
     use super::*;
+    use crate::state;
 
     #[test]
     fn test_emit_event() {
         let (tx, mut rx) = mpsc::unbounded_channel();
         tx.emit(Event::Handshaked);
         assert_eq!(rx.try_recv().unwrap(), Event::Handshaked);
     }
+
+    #[test]
+    fn failed_event_is_forwarded_without_entering_closing() {
+        let conn_state = ArcConnState::new();
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let broker = ArcEventBroker::new(conn_state.clone(), tx);
+        let error = QuicError::with_default_fty(ErrorKind::NoViablePath, "no path");
+
+        broker.emit(Event::Failed(error.clone()));
+
+        assert_eq!(rx.try_recv().unwrap(), Event::Failed(error));
+        assert_ne!(conn_state.current(), Some(state::CLOSING));
+    }
+
+    #[test]
+    fn closed_event_is_forwarded_without_entering_draining() {
+        let conn_state = ArcConnState::new();
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let broker = ArcEventBroker::new(conn_state.clone(), tx);
+        let ccf = ConnectionCloseFrame::new_quic(
+            ErrorKind::NoViablePath,
+            ErrorFrameType::Ext(VarInt::from_u32(0)),
+            "",
+        );
+
+        broker.emit(Event::Closed(ccf.clone()));
+
+        assert_eq!(rx.try_recv().unwrap(), Event::Closed(ccf));
+        assert_ne!(conn_state.current(), Some(state::DRAINING));
+    }
+
+    #[test]
+    fn application_close_event_is_forwarded_without_entering_closing() {
+        let conn_state = ArcConnState::new();
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let broker = ArcEventBroker::new(conn_state.clone(), tx);
+        let error = qbase::error::AppError::new(VarInt::from_u32(0), "");
+
+        broker.emit(Event::ApplicationClose(error.clone()));
+
+        assert_eq!(rx.try_recv().unwrap(), Event::ApplicationClose(error));
+        assert_ne!(conn_state.current(), Some(state::CLOSING));
+    }
 }