feat(taskbroker): Add Push Mode to Taskbroker

Cargo.toml

@@ -17,6 +17,7 @@ clap = { version = "4.5.20", features = ["derive"] }
 derive_builder = "0.20.2"
 elegant-departure = { version = "0.3.1", features = ["tokio"] }
 figment = { version = "0.10.19", features = ["env", "yaml", "test"] }
+flume = "0.12.0"
 futures = "0.3.31"
 futures-util = "0.3.31"
 hex = "0.4.3"
@@ -26,8 +27,8 @@ http-body-util = "0.1.2"
 libsqlite3-sys = "0.30.1"
 metrics = "0.24.0"
 metrics-exporter-statsd = "0.9.0"
-prost = "0.13"
-prost-types = "0.13.3"
+prost = "0.14"
+prost-types = "0.14"
 rand = "0.8.5"
 rdkafka = { version = "0.37.0", features = ["cmake-build", "ssl"] }
 sentry = { version = "0.41.0", default-features = false, features = [
@@ -41,16 +42,16 @@ sentry = { version = "0.41.0", default-features = false, features = [
     "tracing",
     "logs"
 ] }
-sentry_protos = "0.4.11"
+sentry_protos = "0.8.5"
 serde = "1.0.214"
 serde_yaml = "0.9.34"
 sha2 = "0.10.8"
 sqlx = { version = "0.8.3", features = ["sqlite", "runtime-tokio", "chrono", "postgres"] }
 tokio = { version = "1.43.1", features = ["full"] }
 tokio-stream = { version = "0.1.16", features = ["full"] }
 tokio-util = "0.7.12"
-tonic = "0.13"
-tonic-health = "0.13"
+tonic = "0.14"
+tonic-health = "0.14"
 tower = "0.5.1"
 tracing = "0.1.40"
 tracing-subscriber = { version = "0.3.18", features = [

README.md

@@ -77,7 +77,7 @@ The test suite is composed of unit and integration tests in Rust, and end-to-end
 
 ```bash
 # Run unit/integration tests
-make test
+make unit-test
 
 # Run end-to-end tests
 make integration-test

benches/store_bench.rs

@@ -59,7 +59,7 @@ async fn get_pending_activations(num_activations: u32, num_workers: u32) {
             let mut num_activations_processed = 0;
 
             while store
-                .get_pending_activation(Some("sentry"), Some(&ns))
+                .get_pending_activation(Some("sentry"), Some(std::slice::from_ref(&ns)))
                 .await
                 .unwrap()
                 .is_some()

integration_tests/integration_tests/helpers.py

@@ -18,6 +18,9 @@
 TASKBROKER_ROOT = Path(__file__).parent.parent.parent
 TASKBROKER_BIN = TASKBROKER_ROOT / "target/debug/taskbroker"
 TESTS_OUTPUT_ROOT = Path(__file__).parent.parent / ".tests_output"
+
+TASKBROKER_RESTART_PORT_DELAY_SEC = 1.0
+
 TEST_PRODUCER_CONFIG = {
     "bootstrap.servers": "127.0.0.1:9092",
     "broker.address.family": "v4",
@@ -212,7 +215,7 @@ def get_num_tasks_group_by_status(
 
 
 def get_available_ports(count: int) -> list[int]:
-    MIN = 49152
+    MIN = 50051
     MAX = 65535
     res = []
     for i in range(count):

integration_tests/integration_tests/test_consumer_rebalancing.py

@@ -10,6 +10,7 @@
 
 from integration_tests.helpers import (
     TASKBROKER_BIN,
+    TASKBROKER_RESTART_PORT_DELAY_SEC,
     TESTS_OUTPUT_ROOT,
     TaskbrokerConfig,
     create_topic,
@@ -48,6 +49,13 @@ def manage_taskbroker(
                 assert return_code == 0
             except Exception:
                 process.kill()
+                try:
+                    process.wait(timeout=10)
+                except subprocess.TimeoutExpired:
+                    pass
+
+            if i < iterations - 1:
+                time.sleep(TASKBROKER_RESTART_PORT_DELAY_SEC)
 
 
 def test_tasks_written_once_during_rebalancing() -> None: