Validate rust ffi module version matches git tag

Signed-off-by: Sreekanth <prsreekanth920@gmail.com>

Author: BulkBeing

.github/workflows/pynumaflow-lite.yml

@@ -169,6 +169,19 @@ jobs:
           rustup toolchain install stable --profile minimal
           rustup default stable
 
+      - name: Verify tag matches package version
+        if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags/pynumaflow-lite-v')
+        run: |
+          set -euo pipefail
+
+          tag_version="${GITHUB_REF_NAME#pynumaflow-lite-v}"
+          package_version="$(cargo metadata --locked --no-deps --format-version 1 | python -c 'import json, sys; print(json.load(sys.stdin)["packages"][0]["version"])')"
+
+          if [ "${package_version}" != "${tag_version}" ]; then
+            echo "Tag version ${tag_version} does not match Cargo.toml version ${package_version}"
+            exit 1
+          fi
+
       - name: Build wheel
         uses: PyO3/maturin-action@v1
         with:

packages/pynumaflow-lite/src/pyrs.rs

@@ -12,7 +12,6 @@ pub(crate) fn run_asyncio(tx: Sender<Arc<Py<PyAny>>>) {
     let event_loop = Arc::new(event_loop);
     let _ = tx.send(event_loop.clone());
     Python::attach(|py| {
-        println!("Starting NumaflowCore: event_loop={:?}", event_loop);
         event_loop.call_method0(py, "run_forever").unwrap();
     });
 }

packages/pynumaflow-lite/src/sink/mod.rs

@@ -517,9 +517,7 @@ impl SinkAsyncServer {
         }
 
         pyo3_async_runtimes::tokio::future_into_py(py, async move {
-            crate::sink::server::start(py_func, sock_file, info_file, rx)
-                .await
-                .expect("server failed to start");
+            crate::sink::server::start(py_func, sock_file, info_file, rx).await?;
             Ok(())
         })
     }

packages/pynumaflow-lite/src/sink/server.rs

@@ -61,7 +61,10 @@ pub(super) async fn start(
     shutdown_rx: tokio::sync::oneshot::Receiver<()>,
 ) -> Result<(), pyo3::PyErr> {
     let (tx, rx) = tokio::sync::oneshot::channel();
-    let py_asyncio_loop_handle = tokio::task::spawn_blocking(move || crate::pyrs::run_asyncio(tx));
+    let py_asyncio_loop_handle = tokio::task::spawn_blocking({
+        println!("Starting Sink UDF. socket={}, server_info={}", &sock_file, &info_file);
+        move || crate::pyrs::run_asyncio(tx)
+    });
     let event_loop = rx.await.unwrap();
 
     let (sig_handle, combined_rx) = crate::pyrs::setup_sig_handler(shutdown_rx);